diff --git a/app/core/job_registry.py b/app/core/job_registry.py index ebd2e30..b94baf1 100644 --- a/app/core/job_registry.py +++ b/app/core/job_registry.py @@ -46,6 +46,15 @@ def get_job(name: str) -> JobFunc | None: return _registry.get(name) +def unregister_job(name: str) -> None: + """Remove a registered job function (plugin deactivation lifecycle). + + Args: + name: The job name to remove. + """ + _registry.pop(name, None) + + def get_all_jobs() -> list[JobFunc]: """Return all registered job functions (order is insertion order). diff --git a/app/core/worker.py b/app/core/worker.py index 9130d7e..5199a48 100644 --- a/app/core/worker.py +++ b/app/core/worker.py @@ -445,51 +445,9 @@ async def cleanup_trash_job(ctx: dict[str, Any]) -> None: register_job("cleanup_trash", cleanup_trash_job) -# ── Knowledge retention cleanup job ───────────────────────────────────────── - -async def cleanup_knowledge_job(ctx: dict[str, Any]) -> None: - """Delete old knowledge extractions (rejected or auto_created) older than 90 days. - - Runs daily. Keeps approved extractions indefinitely. - Iterates per-tenant for RLS compliance. - """ - from sqlalchemy import text as sa_text, delete as sa_delete - from datetime import datetime, timedelta - - from app.core.db import get_worker_session_factory - from app.plugins.builtins.knowledge.models import KnowledgeExtraction - - factory = get_worker_session_factory() - async with factory() as db: - try: - tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) - tenant_ids = [row[0] for row in tenant_result] - - cutoff = datetime.utcnow() - timedelta(days=90) - total_deleted = 0 - for tenant_id in tenant_ids: - await db.execute( - sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"), - {"tid": str(tenant_id)}, - ) - # Delete rejected and auto_created extractions older than 90 days - result = await db.execute( - sa_delete(KnowledgeExtraction).where( - KnowledgeExtraction.status.in_(["rejected", "auto_created"]), - KnowledgeExtraction.created_at < cutoff, - ) - ) - total_deleted += result.rowcount - await db.commit() - - if total_deleted: - logger.info("Knowledge retention: cleaned up %d old extractions", total_deleted) - except Exception: - logger.error("Knowledge retention cleanup failed", exc_info=True) - await db.rollback() - - -register_job("cleanup_knowledge", cleanup_knowledge_job) +# Note: knowledge retention cleanup ("cleanup_knowledge") lives with the +# knowledge plugin (app/plugins/builtins/knowledge/jobs.py) and is discovered +# via the plugin job-module mechanism — no core→plugin import. class WorkerSettings: @@ -531,9 +489,10 @@ class WorkerSettings: _wrap_cron_with_lock("cleanup_trash", cleanup_trash_job, ttl_seconds=300), hour=4, minute=0, ), - # Knowledge retention cleanup — daily at 05:00 (90 days, keeps approved) + # Knowledge retention cleanup — daily at 05:00 (90 days, keeps approved). + # Function comes from the knowledge plugin via the job registry. cron( - _wrap_cron_with_lock("cleanup_knowledge", cleanup_knowledge_job, ttl_seconds=300), + _wrap_cron_with_lock("cleanup_knowledge", get_job("cleanup_knowledge"), ttl_seconds=300), hour=5, minute=0, ), # Scheduled backup — daily at 02:00 (guarded by distributed lock) diff --git a/app/plugins/builtins/knowledge/jobs.py b/app/plugins/builtins/knowledge/jobs.py new file mode 100644 index 0000000..04a8d06 --- /dev/null +++ b/app/plugins/builtins/knowledge/jobs.py @@ -0,0 +1,62 @@ +"""ARQ background jobs for the knowledge plugin. + +Registered via ``register_job()`` at import time; the worker discovers this +module through ``KnowledgePlugin.get_job_modules()`` — no core imports of +plugin models needed. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta +from typing import Any + +from app.core.job_registry import register_job +from app.plugins.builtins.knowledge.models import KnowledgeExtraction + +logger = logging.getLogger(__name__) + +_KNOWLEDGE_RETENTION_DAYS = 90 + + +async def cleanup_knowledge_job(ctx: dict[str, Any]) -> None: + """Delete old knowledge extractions (rejected or auto_created) older than 90 days. + + Runs daily. Keeps approved extractions indefinitely. + Iterates per-tenant for RLS compliance. + """ + from sqlalchemy import text as sa_text, delete as sa_delete + + from app.core.db import get_worker_session_factory + + factory = get_worker_session_factory() + async with factory() as db: + try: + tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) + tenant_ids = [row[0] for row in tenant_result] + + cutoff = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=_KNOWLEDGE_RETENTION_DAYS) + total_deleted = 0 + for tenant_id in tenant_ids: + await db.execute( + sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"), + {"tid": str(tenant_id)}, + ) + # Delete rejected and auto_created extractions older than retention window + result = await db.execute( + sa_delete(KnowledgeExtraction).where( + KnowledgeExtraction.status.in_(["rejected", "auto_created"]), + KnowledgeExtraction.created_at < cutoff, + ) + ) + total_deleted += result.rowcount + await db.commit() + + if total_deleted: + logger.info("Knowledge retention: cleaned up %d old extractions", total_deleted) + except Exception: + logger.error("Knowledge retention cleanup failed", exc_info=True) + await db.rollback() + + +register_job("cleanup_knowledge", cleanup_knowledge_job) diff --git a/app/plugins/builtins/knowledge/plugin.py b/app/plugins/builtins/knowledge/plugin.py index a5acd07..43fea23 100644 --- a/app/plugins/builtins/knowledge/plugin.py +++ b/app/plugins/builtins/knowledge/plugin.py @@ -143,6 +143,10 @@ class KnowledgePlugin(BasePlugin): ) logger.info("Registered knowledge agent tools: ask_knowledge, search_knowledge") + def get_job_modules(self) -> list[str]: + """ARQ job modules — the retention cleanup job lives with this plugin.""" + return ["app.plugins.builtins.knowledge.jobs"] + async def on_deactivate(self, db, service_container, event_bus) -> None: """Clean up on deactivation.""" from app.core.hooks import unregister_actions_by_owner diff --git a/app/routes/compliance.py b/app/routes/compliance.py index c5d0875..9eef822 100644 --- a/app/routes/compliance.py +++ b/app/routes/compliance.py @@ -19,11 +19,21 @@ from app.core.db import get_db from app.deps import require_permission from app.models.audit import AuditLog from app.models.compliance import ComplianceIncident -from app.plugins.builtins.automation.models import AgentDefinition +from app.plugins.builtins.contracts import get_contract as _get_automation_contract router = APIRouter(prefix="/api/v1/compliance", tags=["compliance"]) +def _get_agent_definition_model(): + """Resolve the AgentDefinition model via the automation contract. + + Returns ``None`` when the automation plugin is not active — callers + respond with 503 instead of failing at import time. + """ + contract = _get_automation_contract("automation") + return getattr(contract, "AgentDefinition", None) if contract else None + + # ─── Schemas ─── @@ -160,13 +170,17 @@ async def list_ai_registry( """List all AI agents with their use-case metadata. Admin only.""" tenant_id = uuid.UUID(current_user["tenant_id"]) + agent_model = _get_agent_definition_model() + if agent_model is None: + raise HTTPException(503, detail={"detail": "Automation plugin not active", "code": "plugin_inactive"}) + q = ( - select(AgentDefinition) + select(agent_model) .where( - AgentDefinition.tenant_id == tenant_id, - AgentDefinition.deleted_at.is_(None), + agent_model.tenant_id == tenant_id, + agent_model.deleted_at.is_(None), ) - .order_by(AgentDefinition.name) + .order_by(agent_model.name) ) result = await db.execute(q) agents = result.scalars().all() @@ -214,10 +228,14 @@ async def get_dpia_template( except ValueError: raise HTTPException(400, detail={"detail": "Invalid agent_id", "code": "invalid_id"}) from None - q = select(AgentDefinition).where( - AgentDefinition.id == aid, - AgentDefinition.tenant_id == tenant_id, - AgentDefinition.deleted_at.is_(None), + agent_model = _get_agent_definition_model() + if agent_model is None: + raise HTTPException(503, detail={"detail": "Automation plugin not active", "code": "plugin_inactive"}) + + q = select(agent_model).where( + agent_model.id == aid, + agent_model.tenant_id == tenant_id, + agent_model.deleted_at.is_(None), ) result = await db.execute(q) agent = result.scalar_one_or_none()