Leopoldadmin/leocrm
1
0
Fork 0
Code Issues 363 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

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

Browse Source
This commit is contained in:
Agent Zero
2026-08-23 12:18:02 +02:00
parent a7699d3598
commit 44511a8fd7
5 changed files with 108 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/core/job_registry.py
+9
View File
@@ -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
+6 -47
View File
@@ -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)
app/plugins/builtins/knowledge/jobs.py
+62
View File
@@ -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
+4
View File
@@ -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
+27 -9
View File
@@ -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()