Phase 3.5: Automation & Agents Plugin
Neues automation Plugin (app/plugins/builtins/automation/): - 7 DB-Modelle: AgentDefinition, AgentVersion, AutomationDefinition, AutomationVersion, AutomationCronJob, AgentRun, AutomationRun - Migration 0001_initial.sql mit allen Tabellen + RLS - PluginManifest erweitert: agent_definitions, automation_templates, cron_jobs, heartbeat_configs, miniapps Contribution-Felder - 21 API-Endpoints: /api/v1/automation (CRUD, execute, dry-run, runs, versions, restore, settings, miniapps) + /api/v1/agents (CRUD, execute, test-run, runs, versions, restore, tools, send-message) Backend Features: - Cron-Scheduler (scheduler.py): ARQ-basiert, liest CronJob-Tabelle, enqueued run_agent/run_automation, croniter fuer next_run_at - Workflow-Timeout-Worker (workflow_timeout.py): prueft abgelaufene WorkflowInstances, setzt cancelled, sendet Notification - Agent Runner (agent_runner.py): LiteLLM + ToolRegistry, proactive/ reactive mode, Rate-Limiting, Budget-Limit, Infinite-Loop-Detection - Automation Execution Engine (execution_engine.py): Condition evaluation (eq/ne/gt/lt/contains/exists), Actions (api_call/ notification/workflow_start), Dry-Run mode - Agent-to-Agent Communication (agent_comm.py): send_agent_message tool, kommunikation plugin integration - Plugin-Beitraege: register/unregister on activate/deactivate, Konfliktloesung mit Plugin-Name als Prefix - Heartbeat-Migration: ai_proactive heartbeat als Cron-Job - Versionshistorie: Auto-Versioning bei Updates, Restore-Endpoint - Settings: GET/PATCH /api/v1/automation/settings - ARQ Worker: 11 functions, 2 cron_jobs (scheduler_tick 30s, check_workflow_timeouts 5min) Frontend: - AutomationDashboard.tsx: Automation Builder UI mit Trigger, Conditions, Actions, Execute, Dry-Run, Run-History, Versions - AgentDashboard.tsx: Agent Builder UI mit Model, Tools, Prompt, Heartbeat, Rate-Limits, Execute, Test-Run, Agent-Chat - AutomationSettings.tsx: Settings + MiniApp-Builder - automation.ts: 24 React Query Hooks - automation.ts types: TypeScript Interfaces - routes/index.tsx: /automation, /agents, /settings/automation Tests: - 22 Frontend-Tests (AutomationDashboard, AgentDashboard, API) — alle bestanden - Backend-Tests: test_automation.py (CRUD, versions, conditions, dry-run, rate-limiting) - TSC: keine neuen Errors (nur pre-existing Dms.tsx) - croniter dependency installiert
This commit is contained in:
Agent Zero
@@ -0,0 +1,737 @@
|
||||
"""Service classes for the Automation & Agents plugin.
|
||||
|
||||
AgentService, AutomationService, CronJobService, RunLogService.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select, text, update
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.plugins.builtins.automation.models import (
|
||||
AgentDefinition,
|
||||
AgentRun,
|
||||
AgentVersion,
|
||||
AutomationCronJob,
|
||||
AutomationDefinition,
|
||||
AutomationRun,
|
||||
AutomationVersion,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ─── Agent Service ───
|
||||
|
||||
|
||||
class AgentService:
|
||||
"""CRUD and version management for AI agent definitions."""
|
||||
|
||||
@staticmethod
|
||||
async def list(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
is_active: bool | None = None,
|
||||
mode: str | None = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[AgentDefinition], int]:
|
||||
"""List agent definitions with optional filters."""
|
||||
query = select(AgentDefinition).where(AgentDefinition.tenant_id == tenant_id)
|
||||
count_query = select(func.count()).select_from(AgentDefinition).where(
|
||||
AgentDefinition.tenant_id == tenant_id
|
||||
)
|
||||
|
||||
if is_active is not None:
|
||||
query = query.where(AgentDefinition.is_active == is_active)
|
||||
count_query = count_query.where(AgentDefinition.is_active == is_active)
|
||||
if mode is not None:
|
||||
query = query.where(AgentDefinition.mode == mode)
|
||||
count_query = count_query.where(AgentDefinition.mode == mode)
|
||||
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AgentDefinition.created_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
items = list(result.scalars().all())
|
||||
return items, total
|
||||
|
||||
@staticmethod
|
||||
async def get_by_id(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID
|
||||
) -> AgentDefinition | None:
|
||||
"""Get a single agent definition by ID."""
|
||||
result = await db.execute(
|
||||
select(AgentDefinition)
|
||||
.where(AgentDefinition.id == agent_id)
|
||||
.where(AgentDefinition.tenant_id == tenant_id)
|
||||
.limit(1)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
@staticmethod
|
||||
async def get_by_name(
|
||||
db: AsyncSession, name: str, tenant_id: uuid.UUID | None = None
|
||||
) -> AgentDefinition | None:
|
||||
"""Get a single agent definition by name (optionally scoped to tenant)."""
|
||||
query = select(AgentDefinition).where(AgentDefinition.name == name)
|
||||
if tenant_id is not None:
|
||||
query = query.where(AgentDefinition.tenant_id == tenant_id)
|
||||
result = await db.execute(query.limit(1))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
@staticmethod
|
||||
async def create(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, data: dict[str, Any], user_id: uuid.UUID | None = None
|
||||
) -> AgentDefinition:
|
||||
"""Create a new agent definition and its initial version snapshot."""
|
||||
agent = AgentDefinition(
|
||||
tenant_id=tenant_id,
|
||||
name=data["name"],
|
||||
description=data.get("description", ""),
|
||||
llm_model=data.get("llm_model", "ollama/deepseek-v4-flash"),
|
||||
system_prompt=data.get("system_prompt", ""),
|
||||
tool_ids=data.get("tool_ids", []),
|
||||
heartbeat_interval_seconds=data.get("heartbeat_interval_seconds", 0),
|
||||
mode=data.get("mode", "reactive"),
|
||||
is_active=data.get("is_active", True),
|
||||
max_executions_per_hour=data.get("max_executions_per_hour", 10),
|
||||
max_duration_seconds=data.get("max_duration_seconds", 300),
|
||||
budget_limit_usd=data.get("budget_limit_usd", 1.0),
|
||||
created_by=user_id,
|
||||
)
|
||||
db.add(agent)
|
||||
await db.flush()
|
||||
|
||||
# Create initial version snapshot
|
||||
version = AgentVersion(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
version_number=1,
|
||||
snapshot={
|
||||
"name": agent.name,
|
||||
"description": agent.description,
|
||||
"llm_model": agent.llm_model,
|
||||
"system_prompt": agent.system_prompt,
|
||||
"tool_ids": agent.tool_ids,
|
||||
"heartbeat_interval_seconds": agent.heartbeat_interval_seconds,
|
||||
"mode": agent.mode,
|
||||
"is_active": agent.is_active,
|
||||
"max_executions_per_hour": agent.max_executions_per_hour,
|
||||
"max_duration_seconds": agent.max_duration_seconds,
|
||||
"budget_limit_usd": agent.budget_limit_usd,
|
||||
},
|
||||
changed_by=user_id,
|
||||
)
|
||||
db.add(version)
|
||||
await db.flush()
|
||||
|
||||
return agent
|
||||
|
||||
@staticmethod
|
||||
async def update(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
agent_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
user_id: uuid.UUID | None = None,
|
||||
) -> AgentDefinition | None:
|
||||
"""Update an agent definition and create a new version snapshot if changed."""
|
||||
agent = await AgentService.get_by_id(db, tenant_id, agent_id)
|
||||
if agent is None:
|
||||
return None
|
||||
|
||||
changed = False
|
||||
for key, value in data.items():
|
||||
if hasattr(agent, key) and value is not None:
|
||||
setattr(agent, key, value)
|
||||
changed = True
|
||||
|
||||
if changed:
|
||||
# Get latest version number
|
||||
result = await db.execute(
|
||||
select(func.coalesce(func.max(AgentVersion.version_number), 0))
|
||||
.where(AgentVersion.agent_id == agent_id)
|
||||
.where(AgentVersion.tenant_id == tenant_id)
|
||||
)
|
||||
max_version = result.scalar() or 0
|
||||
|
||||
version = AgentVersion(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
version_number=max_version + 1,
|
||||
snapshot={
|
||||
"name": agent.name,
|
||||
"description": agent.description,
|
||||
"llm_model": agent.llm_model,
|
||||
"system_prompt": agent.system_prompt,
|
||||
"tool_ids": agent.tool_ids,
|
||||
"heartbeat_interval_seconds": agent.heartbeat_interval_seconds,
|
||||
"mode": agent.mode,
|
||||
"is_active": agent.is_active,
|
||||
"max_executions_per_hour": agent.max_executions_per_hour,
|
||||
"max_duration_seconds": agent.max_duration_seconds,
|
||||
"budget_limit_usd": agent.budget_limit_usd,
|
||||
},
|
||||
changed_by=user_id,
|
||||
)
|
||||
db.add(version)
|
||||
await db.flush()
|
||||
|
||||
return agent
|
||||
|
||||
@staticmethod
|
||||
async def delete(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID
|
||||
) -> bool:
|
||||
"""Delete an agent definition (cascades to versions and runs)."""
|
||||
agent = await AgentService.get_by_id(db, tenant_id, agent_id)
|
||||
if agent is None:
|
||||
return False
|
||||
await db.delete(agent)
|
||||
await db.flush()
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
async def get_versions(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, agent_id: uuid.UUID, limit: int = 50, offset: int = 0
|
||||
) -> tuple[list[AgentVersion], int]:
|
||||
"""Get version history for an agent definition."""
|
||||
query = (
|
||||
select(AgentVersion)
|
||||
.where(AgentVersion.agent_id == agent_id)
|
||||
.where(AgentVersion.tenant_id == tenant_id)
|
||||
)
|
||||
count_query = (
|
||||
select(func.count())
|
||||
.select_from(AgentVersion)
|
||||
.where(AgentVersion.agent_id == agent_id)
|
||||
.where(AgentVersion.tenant_id == tenant_id)
|
||||
)
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AgentVersion.version_number.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
|
||||
@staticmethod
|
||||
async def restore_version(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
agent_id: uuid.UUID,
|
||||
version_id: uuid.UUID,
|
||||
user_id: uuid.UUID | None = None,
|
||||
) -> AgentDefinition | None:
|
||||
"""Restore an agent definition from a specific version snapshot."""
|
||||
agent = await AgentService.get_by_id(db, tenant_id, agent_id)
|
||||
if agent is None:
|
||||
return None
|
||||
|
||||
result = await db.execute(
|
||||
select(AgentVersion)
|
||||
.where(AgentVersion.id == version_id)
|
||||
.where(AgentVersion.agent_id == agent_id)
|
||||
.where(AgentVersion.tenant_id == tenant_id)
|
||||
.limit(1)
|
||||
)
|
||||
version = result.scalar_one_or_none()
|
||||
if version is None:
|
||||
return None
|
||||
|
||||
snapshot = version.snapshot
|
||||
for key, value in snapshot.items():
|
||||
if hasattr(agent, key):
|
||||
setattr(agent, key, value)
|
||||
|
||||
# Create new version reflecting the restore
|
||||
max_ver_result = await db.execute(
|
||||
select(func.coalesce(func.max(AgentVersion.version_number), 0))
|
||||
.where(AgentVersion.agent_id == agent_id)
|
||||
.where(AgentVersion.tenant_id == tenant_id)
|
||||
)
|
||||
max_version = max_ver_result.scalar() or 0
|
||||
|
||||
new_version = AgentVersion(
|
||||
tenant_id=tenant_id,
|
||||
agent_id=agent.id,
|
||||
version_number=max_version + 1,
|
||||
snapshot=snapshot,
|
||||
changed_by=user_id,
|
||||
)
|
||||
db.add(new_version)
|
||||
await db.flush()
|
||||
|
||||
return agent
|
||||
|
||||
|
||||
# ─── Automation Service ───
|
||||
|
||||
|
||||
class AutomationService:
|
||||
"""CRUD and version management for automation definitions."""
|
||||
|
||||
@staticmethod
|
||||
async def list(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
trigger_type: str | None = None,
|
||||
is_active: bool | None = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[AutomationDefinition], int]:
|
||||
"""List automation definitions with optional filters."""
|
||||
query = select(AutomationDefinition).where(
|
||||
AutomationDefinition.tenant_id == tenant_id
|
||||
)
|
||||
count_query = (
|
||||
select(func.count())
|
||||
.select_from(AutomationDefinition)
|
||||
.where(AutomationDefinition.tenant_id == tenant_id)
|
||||
)
|
||||
|
||||
if trigger_type is not None:
|
||||
query = query.where(AutomationDefinition.trigger_type == trigger_type)
|
||||
count_query = count_query.where(
|
||||
AutomationDefinition.trigger_type == trigger_type
|
||||
)
|
||||
if is_active is not None:
|
||||
query = query.where(AutomationDefinition.is_active == is_active)
|
||||
count_query = count_query.where(
|
||||
AutomationDefinition.is_active == is_active
|
||||
)
|
||||
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AutomationDefinition.created_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
|
||||
@staticmethod
|
||||
async def get_by_id(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, automation_id: uuid.UUID
|
||||
) -> AutomationDefinition | None:
|
||||
"""Get a single automation definition by ID."""
|
||||
result = await db.execute(
|
||||
select(AutomationDefinition)
|
||||
.where(AutomationDefinition.id == automation_id)
|
||||
.where(AutomationDefinition.tenant_id == tenant_id)
|
||||
.limit(1)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
@staticmethod
|
||||
async def get_by_name(
|
||||
db: AsyncSession, name: str, tenant_id: uuid.UUID | None = None
|
||||
) -> AutomationDefinition | None:
|
||||
"""Get a single automation definition by name (optionally scoped to tenant)."""
|
||||
query = select(AutomationDefinition).where(AutomationDefinition.name == name)
|
||||
if tenant_id is not None:
|
||||
query = query.where(AutomationDefinition.tenant_id == tenant_id)
|
||||
result = await db.execute(query.limit(1))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
@staticmethod
|
||||
async def create(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
user_id: uuid.UUID | None = None,
|
||||
) -> AutomationDefinition:
|
||||
"""Create a new automation definition and its initial version snapshot."""
|
||||
automation = AutomationDefinition(
|
||||
tenant_id=tenant_id,
|
||||
name=data["name"],
|
||||
description=data.get("description", ""),
|
||||
trigger_type=data.get("trigger_type", "manual"),
|
||||
trigger_config=data.get("trigger_config", {}),
|
||||
conditions=data.get("conditions", []),
|
||||
actions=data.get("actions", []),
|
||||
is_active=data.get("is_active", True),
|
||||
dry_run=data.get("dry_run", False),
|
||||
created_by=user_id,
|
||||
)
|
||||
db.add(automation)
|
||||
await db.flush()
|
||||
|
||||
# Create initial version snapshot
|
||||
version = AutomationVersion(
|
||||
tenant_id=tenant_id,
|
||||
automation_id=automation.id,
|
||||
version_number=1,
|
||||
snapshot={
|
||||
"name": automation.name,
|
||||
"description": automation.description,
|
||||
"trigger_type": automation.trigger_type,
|
||||
"trigger_config": automation.trigger_config,
|
||||
"conditions": automation.conditions,
|
||||
"actions": automation.actions,
|
||||
"is_active": automation.is_active,
|
||||
"dry_run": automation.dry_run,
|
||||
},
|
||||
changed_by=user_id,
|
||||
)
|
||||
db.add(version)
|
||||
await db.flush()
|
||||
|
||||
return automation
|
||||
|
||||
@staticmethod
|
||||
async def update(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
automation_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
user_id: uuid.UUID | None = None,
|
||||
) -> AutomationDefinition | None:
|
||||
"""Update an automation definition and create a new version snapshot if changed."""
|
||||
automation = await AutomationService.get_by_id(db, tenant_id, automation_id)
|
||||
if automation is None:
|
||||
return None
|
||||
|
||||
changed = False
|
||||
for key, value in data.items():
|
||||
if hasattr(automation, key) and value is not None:
|
||||
setattr(automation, key, value)
|
||||
changed = True
|
||||
|
||||
if changed:
|
||||
result = await db.execute(
|
||||
select(func.coalesce(func.max(AutomationVersion.version_number), 0))
|
||||
.where(AutomationVersion.automation_id == automation_id)
|
||||
.where(AutomationVersion.tenant_id == tenant_id)
|
||||
)
|
||||
max_version = result.scalar() or 0
|
||||
|
||||
version = AutomationVersion(
|
||||
tenant_id=tenant_id,
|
||||
automation_id=automation.id,
|
||||
version_number=max_version + 1,
|
||||
snapshot={
|
||||
"name": automation.name,
|
||||
"description": automation.description,
|
||||
"trigger_type": automation.trigger_type,
|
||||
"trigger_config": automation.trigger_config,
|
||||
"conditions": automation.conditions,
|
||||
"actions": automation.actions,
|
||||
"is_active": automation.is_active,
|
||||
"dry_run": automation.dry_run,
|
||||
},
|
||||
changed_by=user_id,
|
||||
)
|
||||
db.add(version)
|
||||
await db.flush()
|
||||
|
||||
return automation
|
||||
|
||||
@staticmethod
|
||||
async def delete(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, automation_id: uuid.UUID
|
||||
) -> bool:
|
||||
"""Delete an automation definition (cascades to versions and runs)."""
|
||||
automation = await AutomationService.get_by_id(db, tenant_id, automation_id)
|
||||
if automation is None:
|
||||
return False
|
||||
await db.delete(automation)
|
||||
await db.flush()
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
async def get_versions(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
automation_id: uuid.UUID,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[AutomationVersion], int]:
|
||||
"""Get version history for an automation definition."""
|
||||
query = (
|
||||
select(AutomationVersion)
|
||||
.where(AutomationVersion.automation_id == automation_id)
|
||||
.where(AutomationVersion.tenant_id == tenant_id)
|
||||
)
|
||||
count_query = (
|
||||
select(func.count())
|
||||
.select_from(AutomationVersion)
|
||||
.where(AutomationVersion.automation_id == automation_id)
|
||||
.where(AutomationVersion.tenant_id == tenant_id)
|
||||
)
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AutomationVersion.version_number.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
|
||||
@staticmethod
|
||||
async def restore_version(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
automation_id: uuid.UUID,
|
||||
version_id: uuid.UUID,
|
||||
user_id: uuid.UUID | None = None,
|
||||
) -> AutomationDefinition | None:
|
||||
"""Restore an automation definition from a specific version snapshot."""
|
||||
automation = await AutomationService.get_by_id(db, tenant_id, automation_id)
|
||||
if automation is None:
|
||||
return None
|
||||
|
||||
result = await db.execute(
|
||||
select(AutomationVersion)
|
||||
.where(AutomationVersion.id == version_id)
|
||||
.where(AutomationVersion.automation_id == automation_id)
|
||||
.where(AutomationVersion.tenant_id == tenant_id)
|
||||
.limit(1)
|
||||
)
|
||||
version = result.scalar_one_or_none()
|
||||
if version is None:
|
||||
return None
|
||||
|
||||
snapshot = version.snapshot
|
||||
for key, value in snapshot.items():
|
||||
if hasattr(automation, key):
|
||||
setattr(automation, key, value)
|
||||
|
||||
max_ver_result = await db.execute(
|
||||
select(func.coalesce(func.max(AutomationVersion.version_number), 0))
|
||||
.where(AutomationVersion.automation_id == automation_id)
|
||||
.where(AutomationVersion.tenant_id == tenant_id)
|
||||
)
|
||||
max_version = max_ver_result.scalar() or 0
|
||||
|
||||
new_version = AutomationVersion(
|
||||
tenant_id=tenant_id,
|
||||
automation_id=automation.id,
|
||||
version_number=max_version + 1,
|
||||
snapshot=snapshot,
|
||||
changed_by=user_id,
|
||||
)
|
||||
db.add(new_version)
|
||||
await db.flush()
|
||||
|
||||
return automation
|
||||
|
||||
|
||||
# ─── Cron Job Service ───
|
||||
|
||||
|
||||
class CronJobService:
|
||||
"""CRUD and next_run calculation for cron job schedules."""
|
||||
|
||||
@staticmethod
|
||||
async def list(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
job_type: str | None = None,
|
||||
is_active: bool | None = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[AutomationCronJob], int]:
|
||||
"""List cron jobs with optional filters."""
|
||||
query = select(AutomationCronJob).where(
|
||||
AutomationCronJob.tenant_id == tenant_id
|
||||
)
|
||||
count_query = (
|
||||
select(func.count())
|
||||
.select_from(AutomationCronJob)
|
||||
.where(AutomationCronJob.tenant_id == tenant_id)
|
||||
)
|
||||
|
||||
if job_type is not None:
|
||||
query = query.where(AutomationCronJob.job_type == job_type)
|
||||
count_query = count_query.where(AutomationCronJob.job_type == job_type)
|
||||
if is_active is not None:
|
||||
query = query.where(AutomationCronJob.is_active == is_active)
|
||||
count_query = count_query.where(AutomationCronJob.is_active == is_active)
|
||||
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AutomationCronJob.created_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
|
||||
@staticmethod
|
||||
async def get_by_id(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, job_id: uuid.UUID
|
||||
) -> AutomationCronJob | None:
|
||||
"""Get a single cron job by ID."""
|
||||
result = await db.execute(
|
||||
select(AutomationCronJob)
|
||||
.where(AutomationCronJob.id == job_id)
|
||||
.where(AutomationCronJob.tenant_id == tenant_id)
|
||||
.limit(1)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
@staticmethod
|
||||
async def get_by_name(
|
||||
db: AsyncSession, name: str, tenant_id: uuid.UUID | None = None
|
||||
) -> AutomationCronJob | None:
|
||||
"""Get a single cron job by name (optionally scoped to tenant)."""
|
||||
query = select(AutomationCronJob).where(AutomationCronJob.name == name)
|
||||
if tenant_id is not None:
|
||||
query = query.where(AutomationCronJob.tenant_id == tenant_id)
|
||||
result = await db.execute(query.limit(1))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
@staticmethod
|
||||
async def create(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, data: dict[str, Any]
|
||||
) -> AutomationCronJob:
|
||||
"""Create a new cron job entry."""
|
||||
job = AutomationCronJob(
|
||||
tenant_id=tenant_id,
|
||||
name=data["name"],
|
||||
cron_expression=data["cron_expression"],
|
||||
job_type=data.get("job_type", "automation_trigger"),
|
||||
target_id=data.get("target_id"),
|
||||
plugin_name=data.get("plugin_name", ""),
|
||||
is_active=data.get("is_active", True),
|
||||
)
|
||||
db.add(job)
|
||||
await db.flush()
|
||||
return job
|
||||
|
||||
@staticmethod
|
||||
async def update(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
job_id: uuid.UUID,
|
||||
data: dict[str, Any],
|
||||
) -> AutomationCronJob | None:
|
||||
"""Update a cron job entry."""
|
||||
job = await CronJobService.get_by_id(db, tenant_id, job_id)
|
||||
if job is None:
|
||||
return None
|
||||
|
||||
for key, value in data.items():
|
||||
if hasattr(job, key) and value is not None:
|
||||
setattr(job, key, value)
|
||||
await db.flush()
|
||||
return job
|
||||
|
||||
@staticmethod
|
||||
async def delete(
|
||||
db: AsyncSession, tenant_id: uuid.UUID, job_id: uuid.UUID
|
||||
) -> bool:
|
||||
"""Delete a cron job entry."""
|
||||
job = await CronJobService.get_by_id(db, tenant_id, job_id)
|
||||
if job is None:
|
||||
return False
|
||||
await db.delete(job)
|
||||
await db.flush()
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
async def get_due_jobs(
|
||||
db: AsyncSession, limit: int = 100
|
||||
) -> list[AutomationCronJob]:
|
||||
"""Get cron jobs that are due for execution (next_run_at <= now)."""
|
||||
now = datetime.now(UTC)
|
||||
result = await db.execute(
|
||||
select(AutomationCronJob)
|
||||
.where(AutomationCronJob.is_active == True)
|
||||
.where(AutomationCronJob.next_run_at <= now)
|
||||
.limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
# ─── Run Log Service ───
|
||||
|
||||
|
||||
class RunLogService:
|
||||
"""Query execution logs for agents and automations."""
|
||||
|
||||
@staticmethod
|
||||
async def list_agent_runs(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
agent_id: uuid.UUID | None = None,
|
||||
status: str | None = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[AgentRun], int]:
|
||||
"""List agent run logs with optional filters."""
|
||||
query = select(AgentRun).where(AgentRun.tenant_id == tenant_id)
|
||||
count_query = (
|
||||
select(func.count())
|
||||
.select_from(AgentRun)
|
||||
.where(AgentRun.tenant_id == tenant_id)
|
||||
)
|
||||
|
||||
if agent_id is not None:
|
||||
query = query.where(AgentRun.agent_id == agent_id)
|
||||
count_query = count_query.where(AgentRun.agent_id == agent_id)
|
||||
if status is not None:
|
||||
query = query.where(AgentRun.status == status)
|
||||
count_query = count_query.where(AgentRun.status == status)
|
||||
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AgentRun.started_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
|
||||
@staticmethod
|
||||
async def list_automation_runs(
|
||||
db: AsyncSession,
|
||||
tenant_id: uuid.UUID,
|
||||
automation_id: uuid.UUID | None = None,
|
||||
status: str | None = None,
|
||||
limit: int = 50,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[AutomationRun], int]:
|
||||
"""List automation run logs with optional filters."""
|
||||
query = select(AutomationRun).where(AutomationRun.tenant_id == tenant_id)
|
||||
count_query = (
|
||||
select(func.count())
|
||||
.select_from(AutomationRun)
|
||||
.where(AutomationRun.tenant_id == tenant_id)
|
||||
)
|
||||
|
||||
if automation_id is not None:
|
||||
query = query.where(AutomationRun.automation_id == automation_id)
|
||||
count_query = count_query.where(
|
||||
AutomationRun.automation_id == automation_id
|
||||
)
|
||||
if status is not None:
|
||||
query = query.where(AutomationRun.status == status)
|
||||
count_query = count_query.where(AutomationRun.status == status)
|
||||
|
||||
count_result = await db.execute(count_query)
|
||||
total = count_result.scalar() or 0
|
||||
|
||||
result = await db.execute(
|
||||
query.order_by(AutomationRun.started_at.desc())
|
||||
.limit(limit)
|
||||
.offset(offset)
|
||||
)
|
||||
return list(result.scalars().all()), total
|
||||
Reference in New Issue
Block a user