From d87fc4e55c42d48aa17a15b722afe1ab4e86ef36 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Sun, 23 Aug 2026 12:50:53 +0200 Subject: [PATCH] fix(arch-030,arch-047): workflow steps resolve plugins via contracts at runtime --- app/plugins/builtins/automation/contracts.py | 5 ++ app/plugins/builtins/calendar/contracts.py | 5 ++ app/plugins/builtins/dms/contracts.py | 5 ++ app/plugins/builtins/mail/contracts.py | 5 ++ .../builtins/unified_search/contracts.py | 33 ++++++++++ app/workflows/step_handlers.py | 63 +++++++++++-------- 6 files changed, 89 insertions(+), 27 deletions(-) diff --git a/app/plugins/builtins/automation/contracts.py b/app/plugins/builtins/automation/contracts.py index 0d2ae6e..d5517b1 100644 --- a/app/plugins/builtins/automation/contracts.py +++ b/app/plugins/builtins/automation/contracts.py @@ -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 ─── diff --git a/app/plugins/builtins/calendar/contracts.py b/app/plugins/builtins/calendar/contracts.py index 052d7eb..3c9830c 100644 --- a/app/plugins/builtins/calendar/contracts.py +++ b/app/plugins/builtins/calendar/contracts.py @@ -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 ─── diff --git a/app/plugins/builtins/dms/contracts.py b/app/plugins/builtins/dms/contracts.py index c25abe8..68a661a 100644 --- a/app/plugins/builtins/dms/contracts.py +++ b/app/plugins/builtins/dms/contracts.py @@ -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 ─── diff --git a/app/plugins/builtins/mail/contracts.py b/app/plugins/builtins/mail/contracts.py index e998238..55f2862 100644 --- a/app/plugins/builtins/mail/contracts.py +++ b/app/plugins/builtins/mail/contracts.py @@ -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 ─── diff --git a/app/plugins/builtins/unified_search/contracts.py b/app/plugins/builtins/unified_search/contracts.py index e128798..b9fc553 100644 --- a/app/plugins/builtins/unified_search/contracts.py +++ b/app/plugins/builtins/unified_search/contracts.py @@ -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 ─── diff --git a/app/workflows/step_handlers.py b/app/workflows/step_handlers.py index a72232a..827b5f1 100644 --- a/app/workflows/step_handlers.py +++ b/app/workflows/step_handlers.py @@ -218,11 +218,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 +260,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 +279,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 +307,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 +354,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 +396,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 {})