refactor(block-h): knowledge retention job lives with plugin; compliance via contract

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).
 

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 ─────────────────────────────────────────
+# Note: knowledge retention cleanup ("cleanup_knowledge") lives with the
-
+# knowledge plugin (app/plugins/builtins/knowledge/jobs.py) and is discovered
-async def cleanup_knowledge_job(ctx: dict[str, Any]) -> None:
+# via the plugin job-module mechanism — no core→plugin import.
-    """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)
 
 
 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)

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)

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

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,
+            agent_model.tenant_id == tenant_id,
-            AgentDefinition.deleted_at.is_(None),
+            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(
+    agent_model = _get_agent_definition_model()
-        AgentDefinition.id == aid,
+    if agent_model is None:
-        AgentDefinition.tenant_id == tenant_id,
+        raise HTTPException(503, detail={"detail": "Automation plugin not active", "code": "plugin_inactive"})
-        AgentDefinition.deleted_at.is_(None),
+
+    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()