fix(arch-030,arch-047): workflow steps resolve plugins via contracts at runtime

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 ───
 

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 ───
 

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 ───
 

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 ───
 

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 ───
 

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
+        from app.plugins.builtins.contracts import get_contract
-        contract = MailContract
+        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
+        from app.plugins.builtins.contracts import get_contract
-        contract = CalendarContract
+        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
+        from app.plugins.builtins.contracts import get_contract
-        contract = DmsContract
+        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
+        from app.plugins.builtins.contracts import get_contract
-        contract = SearchContract
+        contract = get_contract("unified_search")
-        fn = contract.get_function("unified_search")
+        if contract is None:
-        if fn 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,
+            tenant_id=tenant_id,
-            limit=limit,
+            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
+        from app.plugins.builtins.contracts import get_contract
-        contract = AutomationContract
+        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,
+            str(agent_id),
-            agent_id=uuid.UUID(agent_id),
+            trigger_type="workflow",
-            input_data=user_input,
+            trigger_data={"input": user_input},
-            wait_for_completion=wait,
         )
         instance.context["agent_result"] = result
         return StepResult(output={"agent_result": result} if result else {})