Strengthen deterministic agent workflow routing

packages/shared/src/five08/agent/orchestrator.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import re
 from datetime import date, datetime, timedelta, timezone
 from typing import Literal, Protocol, cast
@@ -65,6 +66,7 @@
     "december": 12,
 }
 LiteralPlanner = Literal["deterministic_regex", "live_model"]
+logger = logging.getLogger(__name__)
 
 
 class AgentIntentNormalizer(Protocol):
@@ -120,13 +122,21 @@ def plan(self, message: str, context: AgentIdentityContext) -> AgentResponse:
         if resolved_member_agreement is not None:
             return resolved_member_agreement
 
+        planner: LiteralPlanner = "deterministic_regex"
+        planning_text = text
+        action = self._parse_action(text)
+        if action is not None:
+            return self._response_for_deterministic_action(
+                action=action,
+                context=context,
+                planning_text=planning_text,
+                planner=planner,
+            )
+
         planned_response = self._plan_with_model(text, context)
         if planned_response is not None:
             return planned_response
 
-        planner: LiteralPlanner = "deterministic_regex"
-        planning_text = text
-        action = self._parse_action(text)
         if action is None and not re.search(
             r"\bcreate\s+(?:a\s+)?task\b", text, re.IGNORECASE
         ):
@@ -151,6 +161,23 @@ def plan(self, message: str, context: AgentIdentityContext) -> AgentResponse:
                     "Try asking me to manage a task, GitHub issue, CRM contact, or member account."
                 ),
             )
+        return self._response_for_deterministic_action(
+            action=action,
+            context=context,
+            planning_text=planning_text,
+            planner=planner,
+        )
+
+    def _response_for_deterministic_action(
+        self,
+        *,
+        action: AgentToolAction,
+        context: AgentIdentityContext,
+        planning_text: str,
+        planner: LiteralPlanner,
+    ) -> AgentResponse:
+        """Plan a known workflow before asking a model to infer an intent."""
+
         if action.tool_name == "task_read.search_tasks" and not action.arguments.get(
             "project"
         ):
@@ -184,9 +211,15 @@ def _plan_with_model(
                 runtime_config=self.registry.runtime_config,
                 model_tier=model_tier,
             )
-        except Exception:
+        except Exception as exc:
             # Provider errors fall through to deterministic parsing. Do not expose
             # provider internals in a Discord response.
+            logger.warning(
+                "Structured agent planner failed; using deterministic fallback "
+                "model_tier=%s error_type=%s",
+                model_tier,
+                type(exc).__name__,
+            )
             return None
         if result is None:
             return None

tests/unit/test_agent_gateway.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from datetime import date, datetime, timedelta, timezone
 from types import SimpleNamespace
 from typing import Any
@@ -1585,10 +1586,42 @@ def test_task_creation_is_not_rerouted_to_member_agreement_submission() -> None:
     assert response.plan.actions[0].tool_name == "task_write.create_task"
 
 
+def test_deterministic_supported_workflow_does_not_depend_on_model_routing() -> None:
+    class FailingPlanner:
+        def plan(self, **_kwargs: object) -> AgentPlannerResult:
+            raise AssertionError("planner should not run for a known workflow")
+
+    response = AgentOrchestrator(planner=FailingPlanner()).plan(
+        "Create 508 accounts for Jane Doe with mailbox jane@508.dev",
+        _context(roles=["Admin"]),
+    )
+
+    assert response.status == "requires_confirmation"
+    assert response.plan is not None
+    assert response.plan.planner == "deterministic_regex"
+    assert response.plan.actions[0].tool_name == "account_write.create_user_accounts"
+
+
+def test_structured_planner_failures_are_observable_before_fallback(caplog) -> None:
+    class FailingPlanner:
+        def plan(self, **_kwargs: object) -> AgentPlannerResult:
+            raise ValueError("provider did not return valid JSON")
+
+    with caplog.at_level(logging.WARNING, logger="five08.agent.orchestrator"):
+        response = AgentOrchestrator(planner=FailingPlanner()).plan(
+            "Help me with something unusual",
+            _context(),
+        )
+
+    assert response.status == "needs_clarification"
+    assert "Structured agent planner failed" in caplog.text
+    assert "error_type=ValueError" in caplog.text
+
+
 def test_agent_uses_structured_planner_for_multi_action_confirmation() -> None:
     class FakePlanner:
         def plan(self, **kwargs: object) -> AgentPlannerResult:
-            assert kwargs["model_tier"] == "strong"
+            assert kwargs["model_tier"] == "fast"
             return AgentPlannerResult(
                 draft=PlannerDraft(
                     status="planned",
@@ -1618,7 +1651,10 @@ def plan(self, **kwargs: object) -> AgentPlannerResult:
 
     orchestrator = AgentOrchestrator(planner=FakePlanner())
 
-    response = orchestrator.plan("Invite Sarah to Outline", _context(roles=["Admin"]))
+    response = orchestrator.plan(
+        "Help Sarah get started at 508",
+        _context(roles=["Admin"]),
+    )
 
     assert response.status == "requires_confirmation"
     assert response.plan is not None