15 Commits

Author SHA1 Message Date
Agent Zero b04cda774b fix(arch-014,arch-020): no contract lazy-resurrect after unregister; event bus dedupes handlers
Check Cross-Plugin Imports / check (push) Has been cancelled
Also fixes ARCH-029/041: none-check before attribute access in trigger dispatcher.
2026-08-23 15:30:12 +02:00
Agent Zero 982b4c9353 fix(arch-003): active-manifests available to every authenticated user 2026-08-23 15:09:39 +02:00
Agent Zero 1d6152fb82 fix(arch-001,arch-002): permissions before on_activate; activate once per process 2026-08-23 15:04:51 +02:00
Agent Zero 337d78ef53 merge: Block H - Agent platform kernel (tools/steps/blocks/tabs plugin-contributable) 2026-08-23 14:47:52 +02:00
Agent Zero 801743ba11 test(block-h): gate H proof - plugin contributes all extension points without core changes 2026-08-23 14:17:02 +02:00
Agent Zero 59fdb614e0 refactor(block-h): ai sidebar tabs resolve via plugin-contributable registry 2026-08-23 13:53:23 +02:00
Agent Zero b7ad5294a5 refactor(block-h): message block types resolve via plugin-contributable registry 2026-08-23 13:45:12 +02:00
Agent Zero 49949066d3 feat(block-h): plugin-contributable intents for fallback action mapper 2026-08-23 13:42:30 +02:00
Agent Zero d87fc4e55c fix(arch-030,arch-047): workflow steps resolve plugins via contracts at runtime 2026-08-23 12:50:53 +02:00
Agent Zero 44511a8fd7 refactor(block-h): knowledge retention job lives with plugin; compliance via contract 2026-08-23 12:18:02 +02:00
Agent Zero a7699d3598 refactor(block-h): resolve CommConversation hardcoding via kommunikation contract 2026-08-23 12:02:29 +02:00
Agent Zero 1f4a621910 refactor(block-h): own integration agent tools in their plugins 2026-08-23 11:44:14 +02:00
Agent Zero 637cfa7940 refactor(block-h): move AI tool registry into core AI layer 2026-08-23 11:44:14 +02:00
Agent Zero 35e2cc8ff2 fix(arch-010): checker scans full app tree by default and handles relative paths 2026-08-23 11:32:12 +02:00
Agent Zero 80775959db fix(arch-043): repair indentation of cron job registration guard 2026-08-23 11:32:12 +02:00
37 changed files with 1010 additions and 477 deletions
+44
View File
@@ -6,9 +6,12 @@ Supports keyword-based intent detection for common CRM operations.
from __future__ import annotations
import logging
import re
from typing import Any
logger = logging.getLogger(__name__)
# Precompiled patterns for intent detection
_PATTERNS = {
"create_contact": re.compile(
@@ -27,6 +30,37 @@ _PATTERNS = {
"help": re.compile(r"\b(help|what can you do|assist)\b", re.IGNORECASE),
}
# Plugin-contributed intents: pattern -> callable(query, context) -> list[dict] | None.
# Registered via ``register_intent_pattern`` so plugins can extend the fallback
# mapper without touching core code (Block H / HC-A).
_CONTRIBUTED_INTENTS: list[tuple[re.Pattern[str], Any]] = []
def register_intent_pattern(
pattern: str | re.Pattern[str],
handler: Any,
*,
owner: str = "",
) -> None:
"""Register a plugin-contributed intent for the fallback action mapper.
Args:
pattern: Regex (compiled or raw string) matching the user query.
handler: Callable ``(query, context) -> list[dict] | None`` producing
proposed actions when the pattern matches.
owner: Optional plugin name, used by ``unregister_intent_patterns``.
"""
compiled = re.compile(pattern) if isinstance(pattern, str) else pattern
_CONTRIBUTED_INTENTS.append((compiled, handler))
def unregister_intent_patterns(owner: str) -> None:
"""Remove all intents contributed by ``owner`` (plugin deactivation)."""
global _CONTRIBUTED_INTENTS
_CONTRIBUTED_INTENTS = [
entry for entry in _CONTRIBUTED_INTENTS if getattr(entry[1], "owner_tag", None) != owner
]
# Name extraction patterns - using single-quoted strings to avoid escaping issues
_NAME_PATTERNS = [
re.compile(r"\b(?:named|called|for)\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE),
@@ -147,6 +181,16 @@ def map_query_to_actions(query: str, context: dict[str, Any] | None = None) -> l
}
)
# --- Plugin-contributed intents (Block H / HC-A) ---
for pattern, handler in _CONTRIBUTED_INTENTS:
try:
if pattern.search(q):
contributed = handler(query, context)
if contributed:
actions.extend(contributed)
except Exception:
logger.warning("Contributed intent handler failed", exc_info=True)
# --- Generic fallback ---
if not actions:
if _PATTERNS["help"].search(q):
+8 -14
View File
@@ -50,7 +50,7 @@ if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
from app.plugins.builtins.ai_assistant.tool_registry import ToolRegistry
from app.ai.tool_registry import ToolRegistry
logger = logging.getLogger(__name__)
@@ -394,27 +394,21 @@ async def run_react_loop(
if agent_run_id:
try:
from app.plugins.builtins.contracts import get_contract_registry
from app.plugins.builtins.kommunikation.models import CommConversation
from sqlalchemy import select as sa_select
komm = get_contract_registry().get("kommunikation")
if komm:
agent_id = getattr(agent_definition, "id", uuid.uuid4())
room_title = f"Agent: {getattr(agent_definition, 'name', 'Agent')}"
existing = await db.execute(
sa_select(CommConversation).where(
CommConversation.tenant_id == tenant_id,
CommConversation.title == room_title,
CommConversation.is_locked.is_(True),
CommConversation.locked_by == "automation",
CommConversation.deleted_at.is_(None),
)
conv_id = await komm.find_locked_room_id(
db=db,
tenant_id=tenant_id,
plugin_name="automation",
title=room_title,
)
conv = existing.scalar_one_or_none()
if conv:
if conv_id:
await komm.send_message(
db=db,
tenant_id=tenant_id,
conversation_id=conv.id,
conversation_id=conv_id,
sender_id=agent_id,
sender_type="agent",
content=f"Approval required for tool '{tool_name}'",
+1 -1
View File
@@ -61,7 +61,7 @@ def _resolve_effective_tool_ids(
orchestrate tools but never grant additional permissions.
"""
from app.ai.skill_registry import get_skill_registry
from app.plugins.builtins.ai_assistant.contracts import get_tool_registry
from app.ai.tool_registry import get_tool_registry
agent_tool_ids: list[str] = list(getattr(agent_definition, "tool_ids", None) or [])
agent_skill_ids: list[str] = list(getattr(agent_definition, "skill_ids", None) or [])
+1 -1
View File
@@ -221,7 +221,7 @@ async def build_agent_context(
tool_descriptions: list[dict[str, str]] = []
tool_ids = list(getattr(agent_definition, "tool_ids", None) or [])
try:
from app.plugins.builtins.ai_assistant.contracts import get_tool_registry
from app.ai.tool_registry import get_tool_registry
registry = get_tool_registry()
if tool_ids:
-170
View File
@@ -1,170 +0,0 @@
"""Integration tools — Agent → Workflow + Agent → Knowledge.
Registers AI tools that allow agents to start workflows and ask knowledge questions.
Builds on: workflow_service, knowledge services, tool_registry.
"""
from __future__ import annotations
import logging
import uuid
from typing import Any
logger = logging.getLogger(__name__)
def register_integration_tools() -> None:
"""Register workflow + knowledge tools in the AI tool registry."""
from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry
registry = get_tool_registry()
# ── I-AW: Agent → Workflow ──
async def _start_workflow_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Start a workflow by ID."""
from app.services.workflow_service import create_instance
from app.core.db import get_worker_session_factory
workflow_id = arguments.get("workflow_id", "")
tenant_id = context.get("tenant_id")
user_id = context.get("user_id")
if not workflow_id or not tenant_id:
return {"error": "workflow_id and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
instance = await create_instance(
db=db,
tenant_id=uuid.UUID(str(tenant_id)),
workflow_id=uuid.UUID(workflow_id),
initiated_by=uuid.UUID(str(user_id)) if user_id else None,
)
await db.commit()
return {"instance_id": str(instance.get("id", "")), "status": instance.get("status", "created")}
registry.register(
name="start_workflow",
description="Start a workflow by its ID. Returns the instance ID and status.",
parameters={
"type": "object",
"properties": {
"workflow_id": {"type": "string", "description": "UUID of the workflow to start"},
},
"required": ["workflow_id"],
},
handler=_start_workflow_handler,
plugin_name="integration",
required_permission="workflows:read",
category="workflow",
)
async def _check_workflow_status_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Check the status of a workflow instance."""
from sqlalchemy import select
from app.models.workflow import WorkflowInstance
from app.core.db import get_worker_session_factory
instance_id = arguments.get("instance_id", "")
tenant_id = context.get("tenant_id")
if not instance_id or not tenant_id:
return {"error": "instance_id and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
result = await db.execute(
select(WorkflowInstance).where(
WorkflowInstance.id == uuid.UUID(instance_id),
WorkflowInstance.tenant_id == uuid.UUID(str(tenant_id)),
)
)
inst = result.scalar_one_or_none()
if not inst:
return {"error": "Instance not found"}
return {
"instance_id": str(inst.id),
"status": inst.status,
"current_step": inst.current_step_index,
"completed_at": inst.completed_at.isoformat() if inst.completed_at else None,
}
registry.register(
name="check_workflow_status",
description="Check the status of a workflow instance by its ID.",
parameters={
"type": "object",
"properties": {
"instance_id": {"type": "string", "description": "UUID of the workflow instance"},
},
"required": ["instance_id"],
},
handler=_check_workflow_status_handler,
plugin_name="integration",
required_permission="workflows:read",
category="workflow",
)
# ── I-AK: Agent → Knowledge ──
async def _ask_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Ask a knowledge question."""
from app.plugins.builtins.knowledge.services import ask_knowledge
from app.core.db import get_worker_session_factory
question = arguments.get("question", "")
tenant_id = context.get("tenant_id")
if not question or not tenant_id:
return {"error": "question and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
result = await ask_knowledge(db=db, tenant_id=uuid.UUID(str(tenant_id)), question=question)
return {"answer": result.get("answer", ""), "evidence_count": len(result.get("evidence", []))}
registry.register(
name="ask_knowledge",
description="Ask a knowledge question. Searches wiki articles and graph relationships for context.",
parameters={
"type": "object",
"properties": {
"question": {"type": "string", "description": "The question to ask"},
},
"required": ["question"],
},
handler=_ask_knowledge_handler,
plugin_name="integration",
required_permission="wiki:read",
category="knowledge",
)
async def _search_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Search wiki articles via unified search."""
from app.plugins.builtins.unified_search.provider_registry import get_search_registry
from app.core.db import get_worker_session_factory
query = arguments.get("query", "")
tenant_id = context.get("tenant_id")
if not query or not tenant_id:
return {"error": "query and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
registry = get_search_registry()
wiki_provider = registry.get("wiki_article")
if not wiki_provider:
return {"error": "Wiki search provider not available"}
results = await wiki_provider._search_fts_filtered(
db=db, tsquery=query, tenant_id=uuid.UUID(str(tenant_id)), limit=5, visible_ids=None
)
return {
"results": [
{"title": r.get("title", ""), "summary": (r.get("summary") or r.get("content", "")[:200] or "")}
for r in results
],
"count": len(results),
}
registry.register(
name="search_knowledge",
description="Search wiki articles by keyword. Returns matching articles with title and summary.",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
},
"required": ["query"],
},
handler=_search_knowledge_handler,
plugin_name="integration",
required_permission="wiki:read",
category="knowledge",
)
logger.info("Registered integration tools: start_workflow, check_workflow_status, ask_knowledge, search_knowledge")
+126
View File
@@ -0,0 +1,126 @@
"""Global tool registry for AI agent tools (core platform service).
Plugins register tools here so AI agents can call them during chat sessions.
Each tool declares a name, description, JSON schema for parameters,
and an async handler. Tools can optionally require specific RBAC permissions.
This registry lives in the core AI layer (not inside a plugin) so that the
agent runtime keeps working regardless of which optional plugins are active.
Plugins contribute tools via ``register()`` / ``unregister_plugin()`` during
their activate/deactivate lifecycle.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any, Protocol
logger = logging.getLogger(__name__)
class ToolHandler(Protocol):
async def __call__(
self,
arguments: dict[str, Any],
context: dict[str, Any],
) -> str: ...
@dataclass
class AITool:
"""Represents a tool that an AI agent can call."""
name: str
description: str
parameters: dict[str, Any] # JSON Schema for parameters
handler: ToolHandler
plugin_name: str = ""
required_permission: str | None = None # e.g. "mail:send"
category: str = "general"
def to_openai_schema(self) -> dict[str, Any]:
"""Convert to OpenAI function-calling tool schema."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
class ToolRegistry:
"""Singleton registry for AI tools."""
_instance: ToolRegistry | None = None
def __new__(cls) -> ToolRegistry:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._tools: dict[str, AITool] = {}
return cls._instance
def register(
self,
name: str,
description: str,
parameters: dict[str, Any],
handler: ToolHandler,
plugin_name: str = "",
required_permission: str | None = None,
category: str = "general",
) -> None:
"""Register a tool."""
tool = AITool(
name=name,
description=description,
parameters=parameters,
handler=handler,
plugin_name=plugin_name,
required_permission=required_permission,
category=category,
)
self._tools[name] = tool
logger.info("AI tool registered: %s (plugin=%s)", name, plugin_name)
def unregister(self, name: str) -> None:
"""Unregister a tool by name."""
self._tools.pop(name, None)
def unregister_plugin(self, plugin_name: str) -> None:
"""Unregister all tools from a plugin."""
to_remove = [
name for name, tool in self._tools.items() if tool.plugin_name == plugin_name
]
for name in to_remove:
self._tools.pop(name, None)
def get(self, name: str) -> AITool | None:
return self._tools.get(name)
def get_all(self) -> list[AITool]:
return list(self._tools.values())
def get_by_names(self, names: list[str]) -> list[AITool]:
return [self._tools[name] for name in names if name in self._tools]
def list_for_api(self) -> list[dict[str, Any]]:
"""Return tool list for API response."""
return [
{
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters,
"plugin_name": tool.plugin_name,
"required_permission": tool.required_permission,
"category": tool.category,
}
for tool in self._tools.values()
]
def get_tool_registry() -> ToolRegistry:
"""Get the global tool registry singleton."""
return ToolRegistry()
+7 -2
View File
@@ -36,8 +36,13 @@ class EventBus:
self._handlers: dict[str, list[EventHandler]] = defaultdict(list)
def subscribe(self, event_name: str, handler: EventHandler) -> None:
"""Subscribe a handler to an event."""
self._handlers[event_name].append(handler)
"""Subscribe a handler to an event.
Idempotent: subscribing the same handler twice is a no-op
(ARCH-020) so double activation cannot fire handlers twice.
"""
if handler not in self._handlers[event_name]:
self._handlers[event_name].append(handler)
def unsubscribe(self, event_name: str, handler: EventHandler) -> None:
"""Unsubscribe a handler from an event."""
+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).
+4 -1
View File
@@ -121,11 +121,14 @@ class TriggerDispatcher:
"""Query DB for active automations matching *event_name* and dispatch."""
from app.core.db import get_session_factory
from app.plugins.builtins.contracts import get_contract
# None-check FIRST — accessing attributes on the contract before the
# check crashed with AttributeError when automation was inactive
# (ARCH-029/041).
automation_contract = get_contract("automation")
AutomationDefinition = automation_contract.Automation # noqa: N806
if automation_contract is None:
logger.debug("Automation plugin not available — trigger skipped")
return
AutomationDefinition = automation_contract.Automation # noqa: N806
factory = get_session_factory()
tenant_id = payload.get("tenant_id")
+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)
+9 -8
View File
@@ -284,22 +284,23 @@ async def lifespan(app: FastAPI):
logger.info(f"Plugin {name} is inactive — skipping activation")
continue
# Activate plugin with a FRESH session per plugin to avoid RLS state leakage
# RLS fail-closed requires app.current_tenant_id for tenant-table writes.
# Plugin activation may fail on duplicate cron job inserts — this is harmless
# since cron jobs already exist from previous startups.
# Activate plugin ONCE per process (ARCH-002 fix): a fresh session with
# the first tenant's RLS context satisfies fail-closed RLS for any
# tenant-table writes during activation. Plugins that need per-tenant
# data must seed it themselves (e.g. via the default-tenant mechanism).
# Calling on_activate once prevents duplicate event listeners, cron
# jobs, mini-apps and other contributions at multi-tenant startups.
plugin_activated = False
for tenant_id in all_tenant_ids:
if all_tenant_ids:
try:
async with async_session() as plugin_db:
await set_tenant_context(plugin_db, tenant_id)
await set_tenant_context(plugin_db, all_tenant_ids[0])
await plugin.on_activate(plugin_db, container, event_bus)
await plugin_db.flush()
await plugin_db.commit()
plugin_activated = True
except Exception as exc:
logger.warning(f"[STARTUP] Plugin {name} activation issue for tenant {tenant_id}: {exc}")
break
logger.warning(f"[STARTUP] Plugin {name} activation issue: {exc}")
if plugin_activated:
plugin_record.status = "active"
@@ -1,121 +1,19 @@
"""Global tool registry for AI Assistant plugin tools.
"""Compatibility shim — the tool registry moved to the core AI layer.
Plugins can register tools that AI agents can call during chat sessions.
Each tool declares a name, description, JSON schema for parameters,
and an async handler. Tools can optionally require specific RBAC permissions.
The agent runtime must not depend on this optional plugin being active,
so ``ToolRegistry`` / ``AITool`` / ``get_tool_registry`` now live in
``app.ai.tool_registry``. Import from here still works for existing
plugin code; new code should import from ``app.ai.tool_registry``
directly (or via the ai_assistant contract).
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Any, Protocol
from app.ai.tool_registry import ( # noqa: F401
AITool,
ToolHandler,
ToolRegistry,
get_tool_registry,
)
logger = logging.getLogger(__name__)
class ToolHandler(Protocol):
async def __call__(
self,
arguments: dict[str, Any],
context: dict[str, Any],
) -> str: ...
@dataclass
class AITool:
"""Represents a tool that an AI agent can call."""
name: str
description: str
parameters: dict[str, Any] # JSON Schema for parameters
handler: ToolHandler
plugin_name: str = ""
required_permission: str | None = None # e.g. "mail:send"
category: str = "general"
def to_openai_schema(self) -> dict[str, Any]:
"""Convert to OpenAI function-calling tool schema."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
class ToolRegistry:
"""Singleton registry for AI tools."""
_instance: ToolRegistry | None = None
def __new__(cls) -> ToolRegistry:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._tools: dict[str, AITool] = {}
return cls._instance
def register(
self,
name: str,
description: str,
parameters: dict[str, Any],
handler: ToolHandler,
plugin_name: str = "",
required_permission: str | None = None,
category: str = "general",
) -> None:
"""Register a tool."""
tool = AITool(
name=name,
description=description,
parameters=parameters,
handler=handler,
plugin_name=plugin_name,
required_permission=required_permission,
category=category,
)
self._tools[name] = tool
logger.info("AI tool registered: %s (plugin=%s)", name, plugin_name)
def unregister(self, name: str) -> None:
"""Unregister a tool by name."""
self._tools.pop(name, None)
def unregister_plugin(self, plugin_name: str) -> None:
"""Unregister all tools from a plugin."""
to_remove = [
name for name, tool in self._tools.items() if tool.plugin_name == plugin_name
]
for name in to_remove:
self._tools.pop(name, None)
def get(self, name: str) -> AITool | None:
return self._tools.get(name)
def get_all(self) -> list[AITool]:
return list(self._tools.values())
def get_by_names(self, names: list[str]) -> list[AITool]:
return [self._tools[name] for name in names if name in self._tools]
def list_for_api(self) -> list[dict[str, Any]]:
"""Return tool list for API response."""
return [
{
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters,
"plugin_name": tool.plugin_name,
"required_permission": tool.required_permission,
"category": tool.category,
}
for tool in self._tools.values()
]
def get_tool_registry() -> ToolRegistry:
"""Get the global tool registry singleton."""
return ToolRegistry()
__all__ = ["AITool", "ToolHandler", "ToolRegistry", "get_tool_registry"]
@@ -63,6 +63,11 @@ class AutomationContract:
# ─── agent_comm ───
send_agent_message = staticmethod(send_agent_message)
@classmethod
def get_function(cls, name: str):
"""Return a callable exposed by this contract, or None if absent."""
return getattr(cls, name, None)
# ─── self-registration ───
+121 -26
View File
@@ -210,12 +210,11 @@ class AutomationPlugin(BasePlugin):
register_agent_coordinator_tools()
except Exception:
logger.exception("Failed to register agent coordinator tools")
# Register integration tools (I-AW: Agent→Workflow, I-AK: Agent→Knowledge)
# Register workflow agent tools (I-AW: Agent→Workflow)
try:
from app.ai.integration_tools import register_integration_tools
register_integration_tools()
self._register_workflow_agent_tools()
except Exception:
logger.exception("Failed to register integration tools")
logger.exception("Failed to register workflow agent tools")
# Register MiniApps from manifest
try:
from app.plugins.builtins.kommunikation.contracts import get_miniapp_registry
@@ -248,7 +247,7 @@ class AutomationPlugin(BasePlugin):
from sqlalchemy import select as sa_select
# Get first tenant + admin user for seeding
from app.models.user import User, UserTenant
from app.models.user import User
from app.models.tenant import Tenant
tenant_result = await db.execute(sa_select(Tenant).limit(1))
tenant = tenant_result.scalar_one_or_none()
@@ -286,6 +285,94 @@ class AutomationPlugin(BasePlugin):
logger.info("Automation plugin activated")
def _register_workflow_agent_tools(self) -> None:
"""Register I-AW agent tools for starting and inspecting workflows."""
import uuid
from typing import Any
from app.ai.tool_registry import get_tool_registry
registry = get_tool_registry()
async def _start_workflow_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Start a workflow by ID."""
from app.services.workflow_service import create_instance
from app.core.db import get_worker_session_factory
workflow_id = arguments.get("workflow_id", "")
tenant_id = context.get("tenant_id")
user_id = context.get("user_id")
if not workflow_id or not tenant_id:
return {"error": "workflow_id and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
instance = await create_instance(
db=db,
tenant_id=uuid.UUID(str(tenant_id)),
workflow_id=uuid.UUID(workflow_id),
initiated_by=uuid.UUID(str(user_id)) if user_id else None,
)
await db.commit()
return {"instance_id": str(instance.get("id", "")), "status": instance.get("status", "created")}
registry.register(
name="start_workflow",
description="Start a workflow by its ID. Returns the instance ID and status.",
parameters={
"type": "object",
"properties": {
"workflow_id": {"type": "string", "description": "UUID of the workflow to start"},
},
"required": ["workflow_id"],
},
handler=_start_workflow_handler,
plugin_name=self.manifest.name,
required_permission="workflows:read",
category="workflow",
)
async def _check_workflow_status_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Check the status of a workflow instance."""
from sqlalchemy import select
from app.models.workflow import WorkflowInstance
from app.core.db import get_worker_session_factory
instance_id = arguments.get("instance_id", "")
tenant_id = context.get("tenant_id")
if not instance_id or not tenant_id:
return {"error": "instance_id and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
result = await db.execute(
select(WorkflowInstance).where(
WorkflowInstance.id == uuid.UUID(instance_id),
WorkflowInstance.tenant_id == uuid.UUID(str(tenant_id)),
)
)
inst = result.scalar_one_or_none()
if not inst:
return {"error": "Instance not found"}
return {
"instance_id": str(inst.id),
"status": inst.status,
"current_step": inst.current_step_index,
"completed_at": inst.completed_at.isoformat() if inst.completed_at else None,
}
registry.register(
name="check_workflow_status",
description="Check the status of a workflow instance by its ID.",
parameters={
"type": "object",
"properties": {
"instance_id": {"type": "string", "description": "UUID of the workflow instance"},
},
"required": ["instance_id"],
},
handler=_check_workflow_status_handler,
plugin_name=self.manifest.name,
required_permission="workflows:read",
category="workflow",
)
logger.info("Registered workflow agent tools: start_workflow, check_workflow_status")
async def on_deactivate(self, db, service_container, event_bus) -> None:
"""Clean up on deactivation."""
# Contract abmelden
@@ -307,6 +394,13 @@ class AutomationPlugin(BasePlugin):
unregister_agent_coordinator_tools()
except Exception:
logger.exception("Failed to unregister agent coordinator tools")
# Unregister workflow agent tools from the core AI tool registry
try:
from app.ai.tool_registry import get_tool_registry
get_tool_registry().unregister_plugin(self.manifest.name)
logger.info("Unregistered AI agent tools for plugin '%s'", self.manifest.name)
except Exception:
logger.exception("Failed to unregister AI agent tools")
# Unregister MiniApps
try:
from app.plugins.builtins.kommunikation.contracts import get_miniapp_registry
@@ -388,29 +482,30 @@ class AutomationPlugin(BasePlugin):
logger.exception("Failed to register contributed automation '%s' from plugin '%s'", prefixed_name, plugin_name)
self._contributed_automations[plugin_name] = automation_names
# Register cron jobs
# Register cron jobs (skip if no tenant exists yet)
cron_job_names: list[str] = []
for cron_def in manifest.cron_jobs:
prefixed_name = f"{plugin_name}.{cron_def.name}"
cron_job_names.append(prefixed_name)
try:
# Check if cron job already exists
result = await db.execute(
select(AutomationCronJob).where(AutomationCronJob.name == prefixed_name).limit(1)
)
existing = result.scalar_one_or_none()
if existing is None:
await CronJobService.create(db, default_tenant_id, {
"name": prefixed_name,
"cron_expression": cron_def.cron_expression,
"job_type": cron_def.job_type,
"target_name": cron_def.target_name,
"plugin_name": plugin_name,
"is_active": True,
})
if default_tenant_id is not None:
for cron_def in manifest.cron_jobs:
prefixed_name = f"{plugin_name}.{cron_def.name}"
cron_job_names.append(prefixed_name)
try:
# Check if cron job already exists
result = await db.execute(
select(AutomationCronJob).where(AutomationCronJob.name == prefixed_name).limit(1)
)
existing = result.scalar_one_or_none()
if existing is None:
await CronJobService.create(db, default_tenant_id, {
"name": prefixed_name,
"cron_expression": cron_def.cron_expression,
"job_type": cron_def.job_type,
"target_name": cron_def.target_name,
"plugin_name": plugin_name,
"is_active": True,
})
logger.info("Registered contributed cron job '%s' from plugin '%s'", prefixed_name, plugin_name)
except Exception:
logger.exception("Failed to register contributed cron job '%s' from plugin '%s'", prefixed_name, plugin_name)
except Exception:
logger.exception("Failed to register contributed cron job '%s' from plugin '%s'", prefixed_name, plugin_name)
self._contributed_cron_jobs[plugin_name] = cron_job_names
# Register heartbeat configs
@@ -15,6 +15,11 @@ class CalendarContract:
CalendarEntry = CalendarEntry
CalendarEntryLink = CalendarEntryLink
@classmethod
def get_function(cls, name: str):
"""Return a callable exposed by this contract, or None if absent."""
return getattr(cls, name, None)
# ─── self-registration ───
+18 -2
View File
@@ -59,20 +59,31 @@ class ContractRegistry:
cls._instance = super().__new__(cls)
cls._instance._contracts: dict[str, Any] = {}
cls._instance._loaded: set[str] = set()
cls._instance._unregistered: set[str] = set()
return cls._instance
# ─── registration ───
def register(self, plugin_name: str, contract: Any) -> None:
"""Register or replace a contract for a plugin."""
"""Register or replace a contract for a plugin.
Clears the unregistered marker so a later deactivation can be
distinguished from a fresh lazy-load again (ARCH-014).
"""
self._unregistered.discard(plugin_name)
self._contracts[plugin_name] = contract
self._loaded.add(plugin_name)
logger.debug("Contract registered for plugin '%s'", plugin_name)
def unregister(self, plugin_name: str) -> None:
"""Remove a contract (e.g. when the plugin is deactivated)."""
"""Remove a contract (e.g. when the plugin is deactivated).
Marks the plugin as explicitly unregistered so later ``get_contract``
calls cannot resurrect the contract via lazy-loading (ARCH-014).
"""
self._contracts.pop(plugin_name, None)
self._loaded.discard(plugin_name)
self._unregistered.add(plugin_name)
# ─── lookup ───
@@ -85,6 +96,11 @@ class ContractRegistry:
if plugin_name in self._contracts:
return self._contracts[plugin_name]
# Explicitly unregistered (deactivated): never resurrect via
# lazy-loading (ARCH-014) — the deactivated contract must stay gone.
if plugin_name in self._unregistered:
return None
if plugin_name not in self._loaded:
self._try_lazy_load(plugin_name)
+5
View File
@@ -15,6 +15,11 @@ class DmsContract:
DmsFile = DmsFile
Folder = Folder
@classmethod
def get_function(cls, name: str):
"""Return a callable exposed by this contract, or None if absent."""
return getattr(cls, name, None)
# ─── self-registration ───
+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)
+94
View File
@@ -1,6 +1,8 @@
"""Knowledge plugin — LLM-based entity/relationship extraction, ask-knowledge, review queue."""
from __future__ import annotations
import logging
import uuid
from typing import Any
from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest, PluginRouteDef
@@ -59,9 +61,101 @@ class KnowledgePlugin(BasePlugin):
logger.info("Registered knowledge extraction hooks")
except Exception:
logger.exception("Failed to register knowledge hooks")
# Register knowledge agent tools (I-AK: Agent→Knowledge)
try:
self._register_knowledge_agent_tools()
except Exception:
logger.exception("Failed to register knowledge agent tools")
def _register_knowledge_agent_tools(self) -> None:
"""Register I-AK agent tools for asking and searching knowledge."""
from app.ai.tool_registry import get_tool_registry
registry = get_tool_registry()
async def _ask_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Ask a knowledge question."""
from app.plugins.builtins.knowledge.services import ask_knowledge
from app.core.db import get_worker_session_factory
question = arguments.get("question", "")
tenant_id = context.get("tenant_id")
if not question or not tenant_id:
return {"error": "question and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
result = await ask_knowledge(db=db, tenant_id=uuid.UUID(str(tenant_id)), question=question)
return {"answer": result.get("answer", ""), "evidence_count": len(result.get("evidence", []))}
registry.register(
name="ask_knowledge",
description="Ask a knowledge question. Searches wiki articles and graph relationships for context.",
parameters={
"type": "object",
"properties": {
"question": {"type": "string", "description": "The question to ask"},
},
"required": ["question"],
},
handler=_ask_knowledge_handler,
plugin_name=self.manifest.name,
required_permission="wiki:read",
category="knowledge",
)
async def _search_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
"""Search wiki articles via unified search."""
from app.plugins.builtins.unified_search.provider_registry import get_search_registry
from app.core.db import get_worker_session_factory
query = arguments.get("query", "")
tenant_id = context.get("tenant_id")
if not query or not tenant_id:
return {"error": "query and tenant_id required"}
factory = get_worker_session_factory()
async with factory() as db:
search_registry = get_search_registry()
wiki_provider = search_registry.get("wiki_article")
if not wiki_provider:
return {"error": "Wiki search provider not available"}
results = await wiki_provider._search_fts_filtered(
db=db, tsquery=query, tenant_id=uuid.UUID(str(tenant_id)), limit=5, visible_ids=None
)
return {
"results": [
{"title": r.get("title", ""), "summary": (r.get("summary") or r.get("content", "")[:200] or "")}
for r in results
],
"count": len(results),
}
registry.register(
name="search_knowledge",
description="Search wiki articles by keyword. Returns matching articles with title and summary.",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"},
},
"required": ["query"],
},
handler=_search_knowledge_handler,
plugin_name=self.manifest.name,
required_permission="wiki:read",
category="knowledge",
)
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
unregister_actions_by_owner("knowledge")
# Unregister knowledge agent tools from the core AI tool registry
try:
from app.ai.tool_registry import get_tool_registry
get_tool_registry().unregister_plugin(self.manifest.name)
logger.info("Unregistered AI agent tools for plugin '%s'", self.manifest.name)
except Exception:
logger.exception("Failed to unregister AI agent tools")
await super().on_deactivate(db, service_container, event_bus)
@@ -31,6 +31,7 @@ from app.plugins.builtins.kommunikation.participant_registry import (
)
from app.plugins.builtins.kommunikation.services import (
create_plugin_room,
find_locked_room_id,
get_conversation,
get_messages,
parse_mentions,
@@ -55,6 +56,7 @@ class KommunikationContract:
get_messages = staticmethod(get_messages)
send_message = staticmethod(send_message)
create_plugin_room = staticmethod(create_plugin_room)
find_locked_room_id = staticmethod(find_locked_room_id)
# ─── participant registry ───
get_participant_registry = staticmethod(get_participant_registry)
@@ -92,6 +94,7 @@ __all__ = [
"get_messages",
"send_message",
"create_plugin_room",
"find_locked_room_id",
"CommConversation",
"CommMessage",
"CommParticipant",
@@ -1047,6 +1047,30 @@ async def _get_unread_count(
# ─── Plugin Room Creation ───
async def find_locked_room_id(
db: AsyncSession,
tenant_id: uuid.UUID,
plugin_name: str,
title: str,
) -> uuid.UUID | None:
"""Find the conversation ID of a locked plugin room by tenant and title.
Matches the same room semantics as ``create_plugin_room``: locked rooms
are owned by the plugin (``locked_by == plugin_name``) and soft-deleted
conversations are excluded. Returns ``None`` when no room exists.
"""
result = await db.execute(
select(CommConversation.id).where(
CommConversation.tenant_id == tenant_id,
CommConversation.title == title,
CommConversation.is_locked.is_(True),
CommConversation.locked_by == plugin_name,
CommConversation.deleted_at.is_(None),
)
)
return result.scalar_one_or_none()
async def create_plugin_room(
db: AsyncSession,
tenant_id: uuid.UUID,
+5
View File
@@ -32,6 +32,11 @@ class MailContract:
# ─── models ───
Mail = Mail
@classmethod
def get_function(cls, name: str):
"""Return a callable exposed by this contract, or None if absent."""
return getattr(cls, name, None)
# ─── self-registration ───
@@ -10,6 +10,33 @@ from app.plugins.builtins.unified_search.query_understanding import llm_analyze_
from app.plugins.builtins.unified_search.search_engine import find_similar_all_types, hybrid_search
async def simple_search(
db: Any,
query: str,
tenant_id: Any,
entity_types: list[str] | None = None,
limit: int = 20,
user_id: Any | None = None,
is_system_admin: bool = False,
) -> list[dict[str, Any]]:
"""Convenience search entry point: analyze a raw query string and run
the hybrid search over all registered providers.
Falls back to a plain normalized-query analysis when the LLM is
unavailable.
"""
analysis = await llm_analyze_query(query, db=db, tenant_id=tenant_id)
return await hybrid_search(
db=db,
query_analysis=analysis,
tenant_id=tenant_id,
entity_types=entity_types,
limit=limit,
user_id=user_id,
is_system_admin=is_system_admin,
)
class UnifiedSearchContract:
"""Public contract for the unified_search plugin."""
@@ -20,8 +47,14 @@ class UnifiedSearchContract:
find_similar_all_types = staticmethod(find_similar_all_types)
get_search_registry = staticmethod(get_search_registry)
llm_analyze_query = staticmethod(llm_analyze_query)
simple_search = staticmethod(simple_search)
BaseSearchProvider = BaseSearchProvider
@classmethod
def get_function(cls, name: str):
"""Return a callable exposed by this contract, or None if absent."""
return getattr(cls, name, None)
# ─── self-registration ───
+16
View File
@@ -608,6 +608,22 @@ class PluginRegistry:
for warning in perm_warnings:
logger.warning(warning)
# Register permissions and entity models BEFORE on_activate (ARCH-001 fix):
# the activation hook may already rely on its own permissions/entities being
# resolvable (e.g. tools declaring required_permission).
from app.core.permission_registry import (
get_permission_registry,
register_plugin_permissions,
)
if plugin.manifest.permissions:
register_plugin_permissions(name, plugin.manifest.permissions)
get_permission_registry()._active_plugins.add(name)
for entity_type, model_class in plugin.get_entity_models().items():
from app.services.entity_permission_service import register_entity_model
register_entity_model(entity_type, model_class)
# Call on_activate hook (registers event listeners)
await plugin.on_activate(db, self._container, self._event_bus)
+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()
+5 -1
View File
@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import require_admin, require_permission
from app.deps import get_current_user
from app.plugins.migration_runner import MigrationValidationError
from app.services.plugin_service import get_plugin_service
@@ -94,7 +95,10 @@ async def check_plugin_updates(
@router.get("/active-manifests")
async def get_active_manifests(
db: AsyncSession = Depends(get_db),
current_user: dict = Depends(require_permission("plugins:read")),
# ARCH-003 fix: every authenticated user needs the UI manifests for the
# dynamic sidebar/routes — the data is pure UI metadata; actual data
# access stays protected by each endpoint's own permission.
current_user: dict = Depends(get_current_user),
):
"""Get UI manifests for all active plugins.
+6 -14
View File
@@ -92,22 +92,16 @@ class WorkflowEngine:
# Post workflow completion to Communication (G-WORK)
try:
from app.plugins.builtins.contracts import get_contract_registry
from app.plugins.builtins.kommunikation.models import CommConversation
from sqlalchemy import select as sa_select
komm = get_contract_registry().get("kommunikation")
if komm and instance.initiated_by:
room_title = f"Workflow: {workflow.name if hasattr(workflow, 'name') else str(instance.workflow_id)}"
existing = await self.db.execute(
sa_select(CommConversation).where(
CommConversation.tenant_id == self.tenant_id,
CommConversation.title == room_title,
CommConversation.is_locked.is_(True),
CommConversation.locked_by == "workflow",
CommConversation.deleted_at.is_(None),
)
conv_id = await komm.find_locked_room_id(
db=self.db,
tenant_id=self.tenant_id,
plugin_name="workflow",
title=room_title,
)
conv = existing.scalar_one_or_none()
if not conv:
if not conv_id:
room = await komm.create_plugin_room(
db=self.db,
tenant_id=self.tenant_id,
@@ -117,8 +111,6 @@ class WorkflowEngine:
participant_type="workflow",
)
conv_id = uuid.UUID(room["conversation_id"])
else:
conv_id = conv.id
await komm.send_message(
db=self.db,
tenant_id=self.tenant_id,
+41 -27
View File
@@ -61,6 +61,11 @@ def register_step_type(step_type: str):
return decorator
def unregister_step_type(step_type: str) -> None:
"""Remove a step handler (plugin deactivation lifecycle symmetry)."""
_HANDLERS.pop(step_type, None)
def get_step_handler(step_type: str) -> StepHandler | None:
return _HANDLERS.get(step_type)
@@ -218,11 +223,13 @@ async def _handle_mail(
return StepResult(error="mail step requires to and subject", abort=True)
try:
from app.plugins.builtins.mail.contracts import MailContract
contract = MailContract
from app.plugins.builtins.contracts import get_contract
contract = get_contract("mail")
if contract is None:
return StepResult(error="mail plugin not available", abort=True)
send_fn = contract.get_function("send_email")
if send_fn is None:
return StepResult(error="mail plugin not available", abort=True)
return StepResult(error="mail send_email not exposed via contract yet", abort=True)
result = await send_fn(
db=db,
@@ -258,12 +265,14 @@ async def _handle_calendar(
action = config.get("action", "create")
try:
from app.plugins.builtins.calendar.contracts import CalendarContract
contract = CalendarContract
from app.plugins.builtins.contracts import get_contract
contract = get_contract("calendar")
if contract is None:
return StepResult(error="calendar plugin not available", abort=True)
if action == "create":
fn = contract.get_function("create_event")
if fn is None:
return StepResult(error="calendar plugin not available", abort=True)
return StepResult(error="calendar create_event not exposed via contract yet", abort=True)
result = await fn(
db=db,
tenant_id=tenant_id,
@@ -275,7 +284,7 @@ async def _handle_calendar(
elif action == "delete":
fn = contract.get_function("delete_event")
if fn is None:
return StepResult(error="calendar plugin not available", abort=True)
return StepResult(error="calendar delete_event not exposed via contract yet", abort=True)
await fn(db=db, tenant_id=tenant_id, event_id=config.get("event_id", ""))
return StepResult()
else:
@@ -303,19 +312,21 @@ async def _handle_dms(
action = config.get("action", "search")
try:
from app.plugins.builtins.dms.contracts import DmsContract
contract = DmsContract
from app.plugins.builtins.contracts import get_contract
contract = get_contract("dms")
if contract is None:
return StepResult(error="dms plugin not available", abort=True)
if action == "search":
fn = contract.get_function("search_files")
if fn is None:
return StepResult(error="dms plugin not available", abort=True)
return StepResult(error="dms search_files not exposed via contract yet", abort=True)
results = await fn(db=db, tenant_id=tenant_id, query=config.get("query", ""))
return StepResult(output={"files": results} if results else {})
elif action == "metadata":
fn = contract.get_function("get_file_metadata")
if fn is None:
return StepResult(error="dms plugin not available", abort=True)
return StepResult(error="dms get_file_metadata not exposed via contract yet", abort=True)
metadata = await fn(db=db, tenant_id=tenant_id, file_id=config.get("file_id", ""))
return StepResult(output={"metadata": metadata} if metadata else {})
else:
@@ -348,17 +359,16 @@ async def _handle_search(
return StepResult(error="search step requires query", abort=True)
try:
from app.plugins.builtins.unified_search.contracts import SearchContract
contract = SearchContract
fn = contract.get_function("unified_search")
if fn is None:
from app.plugins.builtins.contracts import get_contract
contract = get_contract("unified_search")
if contract is None:
return StepResult(error="search plugin not available", abort=True)
results = await fn(
results = await contract.simple_search(
db=db,
tenant_id=tenant_id,
query=query,
entity_type=entity_type,
limit=limit,
tenant_id=tenant_id,
entity_types=[entity_type] if entity_type else None,
limit=int(limit),
)
# Store results in context for later steps
instance.context["search_results"] = results
@@ -391,17 +401,21 @@ async def _handle_agent(
return StepResult(error="agent step requires agent_id", abort=True)
try:
from app.plugins.builtins.automation.contracts import AutomationContract
contract = AutomationContract
from app.plugins.builtins.contracts import get_contract
contract = get_contract("automation")
if contract is None:
return StepResult(error="automation plugin not available", abort=True)
fn = contract.get_function("run_agent")
if fn is None:
return StepResult(error="automation plugin not available", abort=True)
return StepResult(error="automation run_agent not exposed via contract yet", abort=True)
# run_agent is an ARQ job function: it opens its own DB session and runs
# the agent loop to completion, so ``wait`` always ends up true here.
logger.debug("agent step wait_for_completion=%s (step always waits)", wait)
result = await fn(
db=db,
tenant_id=tenant_id,
agent_id=uuid.UUID(agent_id),
input_data=user_input,
wait_for_completion=wait,
{},
str(agent_id),
trigger_type="workflow",
trigger_data={"input": user_input},
)
instance.context["agent_result"] = result
return StepResult(output={"agent_result": result} if result else {})
@@ -6,14 +6,8 @@ import ImageBlock from './ImageBlock';
import AudioBlock from './AudioBlock';
import VideoBlock from './VideoBlock';
import FileBlock from './FileBlock';
import ActionCardBlock from './ActionCardBlock';
import ContactCardBlock from './ContactCardBlock';
import MiniAppBlock from './MiniAppBlock';
import AgentResultBlock from './AgentResultBlock';
import ApprovalRequestBlock from './ApprovalRequestBlock';
import TaskCardBlock from './TaskCardBlock';
import WorkflowCardBlock from './WorkflowCardBlock';
import KnowledgeCardBlock from './KnowledgeCardBlock';
import { getBlockComponent } from './registry';
import './registrations'; // plugin-contributed block registrations
interface BlockRendererProps {
blocks: MessageBlock[];
@@ -51,29 +45,20 @@ const BlockRenderer: React.FC<BlockRendererProps> = ({ blocks }) => {
return <VideoBlock block={block} />;
case 'file':
return <FileBlock block={block} />;
case 'action_card':
return <ActionCardBlock block={block} />;
case 'contact_card':
return <ContactCardBlock block={block} />;
case 'miniapp':
return <MiniAppBlock block={block} />;
case 'agent_result':
return <AgentResultBlock block={block} />;
case 'approval_request':
return <ApprovalRequestBlock block={block} />;
case 'task_card':
return <TaskCardBlock block={block} />;
case 'workflow_card':
return <WorkflowCardBlock block={block} />;
case 'knowledge_card':
return <KnowledgeCardBlock block={block} />;
default:
default: {
// Plugin-contributed block types (Block H / HC-F)
const Contributed = getBlockComponent(block.block_type);
if (Contributed) {
const ContributedBlock = Contributed as React.FC<{ block: MessageBlock }>;
return <ContributedBlock block={block} />;
}
// Fallback for unknown block types
return (
<div className="text-xs text-secondary-400 italic p-2 rounded bg-secondary-50">
Unbekannter Block-Typ: {block.block_type}
</div>
);
}
}
};
@@ -0,0 +1,26 @@
/**
* Plugin block registrations.
*
* Currently a static import list; the dynamic plugin loader (ARCH-019 fix)
* will replace this file with manifest-driven registration. Plugins register
* their block renderers via ``registerBlockType`` with an owner tag so that
* ``unregisterBlockTypes(owner)`` can clean up on deactivation.
*/
import { registerBlockType } from './registry';
import ActionCardBlock from './ActionCardBlock';
import ContactCardBlock from './ContactCardBlock';
import MiniAppBlock from './MiniAppBlock';
import AgentResultBlock from './AgentResultBlock';
import ApprovalRequestBlock from './ApprovalRequestBlock';
import TaskCardBlock from './TaskCardBlock';
import WorkflowCardBlock from './WorkflowCardBlock';
import KnowledgeCardBlock from './KnowledgeCardBlock';
registerBlockType('action_card', ActionCardBlock, 'core-comm');
registerBlockType('contact_card', ContactCardBlock, 'contacts');
registerBlockType('miniapp', MiniAppBlock, 'kommunikation');
registerBlockType('agent_result', AgentResultBlock, 'automation');
registerBlockType('approval_request', ApprovalRequestBlock, 'core-approvals');
registerBlockType('task_card', TaskCardBlock, 'tasks');
registerBlockType('workflow_card', WorkflowCardBlock, 'workflows');
registerBlockType('knowledge_card', KnowledgeCardBlock, 'knowledge');
@@ -0,0 +1,36 @@
/**
* Message block type registry.
*
* Core owns the base block types; plugins can contribute additional block
* renderers via ``registerBlockType`` / ``unregisterBlockTypes`` during their
* frontend activation lifecycle (Block H / HC-F).
*/
import type { ComponentType } from 'react';
import type { MessageBlock } from '@/store/commStore';
export type BlockComponent = ComponentType<{ block: MessageBlock }>;
const _registry = new Map<string, BlockComponent>();
export function registerBlockType(blockType: string, component: BlockComponent, owner = ''): void {
if (_registry.has(blockType)) {
// eslint-disable-next-line no-console
console.warn(`[blockRegistry] block type '${blockType}' re-registered by '${owner || 'unknown'}' — overwriting`);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(component as any).__block_owner__ = owner;
_registry.set(blockType, component);
}
export function unregisterBlockTypes(owner: string): void {
for (const [key, component] of _registry.entries()) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
if ((component as any).__block_owner__ === owner) {
_registry.delete(key);
}
}
}
export function getBlockComponent(blockType: string): BlockComponent | undefined {
return _registry.get(blockType);
}
+15 -1
View File
@@ -5,6 +5,7 @@ import { SuggestionList } from '@/components/ai/SuggestionSidebar';
import { ImprovementPanel } from '@/components/ai/ImprovementPanel';
import { createSession, fetchSessions } from '@/api/ai';
import { useUIStore } from '@/store/uiStore';
import { getContributedTabs } from './sidebarTabs';
import { useTranslation } from 'react-i18next';
import { useUsers, useGroups } from '@/api/hooks';
import { Avatar } from '@/components/ui/Avatar';
@@ -35,7 +36,7 @@ const chevronRightIcon = (
);
interface TabDef {
key: 'proactive' | 'chat' | 'notifications' | 'team' | 'chatroom';
key: string;
label: string;
icon: (cls: string) => React.ReactNode;
testId: string;
@@ -139,6 +140,13 @@ export function AISidebar() {
{ key: 'notifications', label: t('topbar.notifications'), icon: bellIcon, testId: 'ai-sidebar-tab-notifications' },
{ key: 'team', label: 'Team', icon: teamIcon, testId: 'ai-sidebar-tab-team' },
{ key: 'chatroom', label: 'Chat', icon: chatBubbleIcon, testId: 'ai-sidebar-tab-chatroom' },
// Plugin-contributed tabs (Block H / HC-G)
...getContributedTabs().map((ct) => ({
key: ct.key,
label: ct.label,
icon: ct.icon,
testId: ct.testId,
})),
];
const activeTab = tabs.find((tab) => tab.key === aiSidebarTab) || tabs[0];
@@ -209,6 +217,12 @@ export function AISidebar() {
if (aiSidebarTab === 'chatroom') {
return <ChatPanel />;
}
// Plugin-contributed tabs (Block H / HC-G)
const contributed = getContributedTabs().find((ct) => ct.key === aiSidebarTab);
if (contributed) {
const ContributedComponent = contributed.component;
return <ContributedComponent />;
}
// chat tab
if (loading) {
return (
@@ -0,0 +1,42 @@
/**
* AI sidebar tab registry.
*
* Core owns the base tabs (chat/proactive/notifications/team/chatroom);
* plugins contribute additional tabs via ``registerSidebarTab`` during their
* frontend activation lifecycle (Block H / HC-G). Contributed tabs render a
* generic container component provided by the plugin.
*/
import type { ComponentType } from 'react';
type IconFn = (className: string) => JSX.Element;
export interface ContributedTab {
key: string;
label: string;
icon: IconFn;
testId: string;
component: ComponentType;
}
const _tabs = new Map<string, ContributedTab>();
export function registerSidebarTab(tab: ContributedTab): void {
if (_tabs.has(tab.key)) {
// eslint-disable-next-line no-console
console.warn(`[sidebarTabs] tab '${tab.key}' re-registered — overwriting`);
}
_tabs.set(tab.key, tab);
}
export function unregisterSidebarTabs(owner: string): void {
for (const [key, tab] of _tabs.entries()) {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
if ((tab.component as any).__sidebar_owner__ === owner) {
_tabs.delete(key);
}
}
}
export function getContributedTabs(): ContributedTab[] {
return Array.from(_tabs.values());
}
+3 -1
View File
@@ -2,7 +2,9 @@ import { create } from 'zustand';
export type Theme = 'light' | 'dark' | 'system';
export type Locale = 'de' | 'en';
export type AISidebarTab = 'proactive' | 'chat' | 'notifications' | 'team' | 'chatroom';
// Plugin-contributed tabs add arbitrary keys (Block H / HC-G) — the union
// keeps autocomplete for the base tabs while allowing any string key.
export type AISidebarTab = 'proactive' | 'chat' | 'notifications' | 'team' | 'chatroom' | (string & {});
export interface Toast {
id: string;
+8 -9
View File
@@ -200,15 +200,10 @@ def find_python_files(search_path: Path | None = None) -> list[Path]:
files.append(Path(root) / fname)
return sorted(files)
# Default: scan both plugin and core directories
# Default: scan the whole app/ tree so new subpackages are covered automatically
files: list[Path] = []
scan_dirs = [
BUILTINS_DIR,
PROJECT_ROOT / "app" / "core",
PROJECT_ROOT / "app" / "services",
PROJECT_ROOT / "app" / "routes",
PROJECT_ROOT / "app" / "commands",
PROJECT_ROOT / "app" / "ai",
PROJECT_ROOT / "app",
]
for scan_dir in scan_dirs:
if not scan_dir.exists():
@@ -229,8 +224,8 @@ def main() -> int:
parser.add_argument(
"--path",
type=Path,
default=BUILTINS_DIR,
help="Path to check (default: app/plugins/builtins)",
default=None,
help="Path to check (default: full scan of builtins + core dirs)",
)
parser.add_argument(
"--verbose",
@@ -239,6 +234,10 @@ def main() -> int:
)
args = parser.parse_args()
# Resolve user-supplied paths so downstream relative_to(PROJECT_ROOT) works
if args.path is not None:
args.path = args.path.resolve()
files = find_python_files(args.path)
all_violations: list[str] = []
checked = 0
+169
View File
@@ -0,0 +1,169 @@
"""Gate H — prove that a plugin contributes ALL extension points without
changing a single core file.
Covers:
1. Agent tool -> app.ai.tool_registry (core layer)
2. Workflow step type -> app.workflows.step_handlers
3. Fallback intent -> app.ai.action_mapper
4. Deactivate symmetry -> everything removed again
5. Re-activate -> present again, NO duplicates
The plugin under test is defined INLINE in this file — if this test passes,
no core file had to be touched to extend the platform.
"""
from __future__ import annotations
from typing import Any
import pytest
# ─── The Gate-H test plugin (inline — zero core changes) ─────────────────────
def _build_gate_h_plugin():
from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest
class GateHTestPlugin(BasePlugin):
"""Minimal plugin exercising every Block-H contribution point."""
manifest = PluginManifest(
name="gate_h_test",
version="1.0.0",
display_name="Gate H Test",
description="Proves plugin contributions without core changes.",
dependencies=[],
routes=[],
events=[],
migrations=[],
permissions=["gate_h_test:read"],
)
async def on_activate(self, db, service_container, event_bus) -> None:
await super().on_activate(db, service_container, event_bus)
# 1) Agent tool via CORE registry
from app.ai.tool_registry import get_tool_registry
async def _echo_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str:
return f"gate-h says: {arguments.get('text', '')}"
get_tool_registry().register(
name="gate_h_echo",
description="Echo text back (Gate H proof).",
parameters={
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
handler=_echo_handler,
plugin_name=self.manifest.name,
required_permission=None,
category="test",
)
# 2) Workflow step type
from app.workflows.step_handlers import StepResult, register_step_type
@register_step_type("gate_h_noop")
async def _handle_gate_h_noop(db, tenant_id, instance, step): # noqa: ANN001
return StepResult(output={"gate_h": True})
# 3) Fallback intent pattern
from app.ai.action_mapper import register_intent_pattern
def _invoice_intent(query: str, context: dict[str, Any] | None):
return [
{
"method": "POST",
"path": "/api/v1/gate-h/invoices",
"body": {},
"description": "Create invoice (Gate H proof)",
"confidence": 0.9,
}
]
_invoice_intent.owner_tag = self.manifest.name
register_intent_pattern(r"\b(create|new)\b.*\binvoice\b", _invoice_intent, owner=self.manifest.name)
async def on_deactivate(self, db, service_container, event_bus) -> None:
# Symmetry: remove everything registered above
from app.ai.tool_registry import get_tool_registry
from app.workflows.step_handlers import unregister_step_type
from app.ai.action_mapper import unregister_intent_patterns
get_tool_registry().unregister_plugin(self.manifest.name)
unregister_step_type("gate_h_noop")
unregister_intent_patterns(self.manifest.name)
await super().on_deactivate(db, service_container, event_bus)
return GateHTestPlugin
# ─── The proof ────────────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_gate_h_full_contribution_lifecycle():
"""Activate -> contributed everywhere; deactivate -> gone; reactivate -> back, no dupes."""
from app.ai.tool_registry import get_tool_registry
from app.workflows.step_handlers import get_step_handler
from app.ai.action_mapper import map_query_to_actions
plugin_cls = _build_gate_h_plugin()
plugin = plugin_cls()
# ── Before activation: nothing registered ──
assert get_tool_registry().get("gate_h_echo") is None
assert get_step_handler("gate_h_noop") is None
assert map_query_to_actions("create invoice for Acme") == []
# ── Activate ──
await plugin.on_activate(db=None, service_container=None, event_bus=None)
# 1) Agent tool present in CORE registry
tool = get_tool_registry().get("gate_h_echo")
assert tool is not None
assert tool.plugin_name == "gate_h_test"
result = await tool.handler({"text": "hello"}, {})
assert result == "gate-h says: hello"
# 2) Workflow step type resolvable
handler = get_step_handler("gate_h_noop")
assert handler is not None
# 3) Intent contributes actions for a FACHMODUL term unknown to core
actions = map_query_to_actions("create invoice for Acme")
assert len(actions) == 1
assert actions[0]["path"] == "/api/v1/gate-h/invoices"
# ── Deactivate: full symmetry ──
await plugin.on_deactivate(db=None, service_container=None, event_bus=None)
assert get_tool_registry().get("gate_h_echo") is None
assert get_step_handler("gate_h_noop") is None
assert map_query_to_actions("create invoice for Acme") == []
# ── Re-activate: back AND no duplicates ──
await plugin.on_activate(db=None, service_container=None, event_bus=None)
tool = get_tool_registry().get("gate_h_echo")
assert tool is not None # single entry (dict keyed by name — overwrite, not dupe)
assert get_step_handler("gate_h_noop") is not None
actions = map_query_to_actions("create invoice for Acme")
assert len(actions) == 1 # exactly one contributed intent, not two
# Cleanup so other tests are unaffected
await plugin.on_deactivate(db=None, service_container=None, event_bus=None)
@pytest.mark.asyncio
async def test_gate_h_agent_runtime_survives_ai_assistant_absence():
"""The core tool registry works even though ai_assistant plugin module is never imported."""
from app.ai.tool_registry import ToolRegistry, get_tool_registry
reg = get_tool_registry()
assert isinstance(reg, ToolRegistry)
# And it must be THE same singleton the shim exposes
from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry as shim_get
assert shim_get() is reg
+3 -3
View File
@@ -530,7 +530,7 @@ class TestAgentPermissions:
) as mock_resolve:
mock_resolve.return_value = _permissions(permissions=["mail:read"])
with patch(
"app.plugins.builtins.ai_assistant.contracts.get_tool_registry",
"app.ai.tool_registry.get_tool_registry",
return_value=tool_registry,
):
ctx = await resolve_agent_permissions(
@@ -575,7 +575,7 @@ class TestAgentPermissions:
) as mock_resolve:
mock_resolve.return_value = _permissions(permissions=["mail:read"])
with patch(
"app.plugins.builtins.ai_assistant.contracts.get_tool_registry",
"app.ai.tool_registry.get_tool_registry",
return_value=tool_registry,
):
ctx = await resolve_agent_permissions(
@@ -1015,7 +1015,7 @@ class TestContextBuilder:
])
with patch(
"app.plugins.builtins.ai_assistant.contracts.get_tool_registry",
"app.ai.tool_registry.get_tool_registry",
return_value=tool_registry,
):
messages = await build_agent_context(