Add optional adaptive escalation verification for Plan-Execute

src/agent/plan_execute/escalation.py

@@ -0,0 +1,184 @@
+"""Deterministic escalation signal extraction for plan-execute runs."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Iterable
+
+from .models import Plan, StepResult
+
+DEFAULT_SPECIALIST_SERVERS = frozenset({"fmsr", "tsfm", "vibration", "wo"})
+
+DEFAULT_ESCALATION_TERMS = (
+    "work order",
+    "work-order",
+    "diagnostic",
+    "diagnostics",
+    "diagnosis",
+    "failure",
+    "failed",
+    "fault",
+    "alarm",
+    "anomaly",
+)
+
+
+@dataclass
+class EscalationSignals:
+    """Signals that may later inform adaptive escalation decisions."""
+
+    step_count: int
+    dependency_depth: int
+    uses_specialist_servers: bool
+    specialist_servers_used: list[str] = field(default_factory=list)
+    any_step_failed: bool = False
+    failed_steps: list[int] = field(default_factory=list)
+    servers_used: list[str] = field(default_factory=list)
+    tools_used: list[str] = field(default_factory=list)
+    matched_terms: list[str] = field(default_factory=list)
+
+    @property
+    def has_domain_terms(self) -> bool:
+        return bool(self.matched_terms)
+
+
+@dataclass
+class EscalationDecision:
+    """Deterministic escalation decision for a plan-execute run."""
+
+    should_escalate: bool
+    reasons: list[str] = field(default_factory=list)
+
+
+def extract_escalation_signals(
+    question: str,
+    plan: Plan,
+    trajectory: Iterable[StepResult] | None = None,
+    specialist_servers: Iterable[str] = DEFAULT_SPECIALIST_SERVERS,
+    escalation_terms: Iterable[str] = DEFAULT_ESCALATION_TERMS,
+) -> EscalationSignals:
+    """Extract deterministic escalation signals from a plan and trajectory.
+
+    This function is intentionally side-effect free and makes no LLM calls.
+    """
+    results = list(trajectory or [])
+    specialist_server_set = set(specialist_servers)
+
+    servers_used = _unique_sorted(
+        [step.server for step in plan.steps] + [result.server for result in results]
+    )
+    tools_used = _unique_sorted(
+        [step.tool for step in plan.steps if step.tool]
+        + [result.tool for result in results if result.tool]
+    )
+    specialist_servers_used = [
+        server for server in servers_used if server in specialist_server_set
+    ]
+    failed_steps = [result.step_number for result in results if not result.success]
+
+    return EscalationSignals(
+        step_count=len(plan.steps),
+        dependency_depth=_dependency_depth(plan),
+        uses_specialist_servers=bool(specialist_servers_used),
+        specialist_servers_used=specialist_servers_used,
+        any_step_failed=bool(failed_steps),
+        failed_steps=failed_steps,
+        servers_used=servers_used,
+        tools_used=tools_used,
+        matched_terms=_matched_terms(question, plan, results, escalation_terms),
+    )
+
+
+def should_escalate(signals: EscalationSignals) -> EscalationDecision:
+    """Return a deterministic adaptive-escalation decision for extracted signals."""
+    reasons = []
+    if signals.any_step_failed:
+        reasons.append("failed step")
+    if signals.dependency_depth >= 3:
+        reasons.append("dependency depth >= 3")
+    if signals.uses_specialist_servers:
+        reasons.append("specialist server used")
+    if signals.has_domain_terms:
+        reasons.append("domain escalation term matched")
+
+    return EscalationDecision(should_escalate=bool(reasons), reasons=reasons)
+
+
+def _dependency_depth(plan: Plan) -> int:
+    """Return the longest dependency chain length, counting the step itself."""
+    steps_by_number = {step.step_number: step for step in plan.steps}
+    visiting: set[int] = set()
+    memo: dict[int, int] = {}
+
+    def depth(step_number: int) -> int:
+        if step_number in memo:
+            return memo[step_number]
+        if step_number in visiting:
+            return 1
+
+        step = steps_by_number.get(step_number)
+        if step is None:
+            return 0
+
+        visiting.add(step_number)
+        dep_depth = max((depth(dep) for dep in step.dependencies), default=0)
+        visiting.remove(step_number)
+        memo[step_number] = dep_depth + 1
+        return memo[step_number]
+
+    return max((depth(step.step_number) for step in plan.steps), default=0)
+
+
+def _matched_terms(
+    question: str,
+    plan: Plan,
+    trajectory: list[StepResult],
+    escalation_terms: Iterable[str],
+) -> list[str]:
+    text = "\n".join(
+        [
+            question,
+            plan.raw,
+            *[
+                "\n".join([step.task, step.expected_output, step.server, step.tool])
+                for step in plan.steps
+            ],
+            *[
+                "\n".join(
+                    [
+                        result.task,
+                        result.server,
+                        result.tool,
+                        result.response,
+                        result.error or "",
+                    ]
+                )
+                for result in trajectory
+            ],
+        ]
+    )
+    matched = []
+    seen = set()
+    for term in escalation_terms:
+        key = term.casefold()
+        if key in seen:
+            continue
+        if re.search(rf"(?<!\w){re.escape(term)}(?!\w)", text, re.IGNORECASE):
+            seen.add(key)
+            matched.append(term)
+    return matched
+
+
+def _unique_sorted(values: Iterable[str]) -> list[str]:
+    return sorted({value for value in values if value})
+
+
+__all__ = [
+    "DEFAULT_ESCALATION_TERMS",
+    "DEFAULT_SPECIALIST_SERVERS",
+    "EscalationDecision",
+    "EscalationSignals",
+    "extract_escalation_signals",
+    "should_escalate",
+]

src/agent/plan_execute/runner.py

@@ -19,6 +19,7 @@
 from llm import LLMBackend, LLMResult
 from observability import agent_run_span, persist_trajectory
 
+from .escalation import extract_escalation_signals, should_escalate
 from .executor import Executor
 from .models import OrchestratorResult
 from .planner import Planner
@@ -76,6 +77,40 @@ def model_id(self) -> str:
 above. Do not repeat the individual steps — just give the final answer.
 """
 
+_VERIFY_PROMPT = """\
+You are reviewing the evidence gathered by a plan-execute industrial asset \
+operations agent before it gives a final answer.
+
+Original question: {question}
+
+Escalation reasons:
+{reasons}
+
+Step-by-step execution results:
+{results}
+
+Check for failed steps, missing evidence, conflicting evidence, alternative \
+explanations, and whether any work-order or action recommendation would be \
+premature. Provide concise verification notes only.
+"""
+
+_SUMMARIZE_WITH_VERIFICATION_PROMPT = """\
+You are summarizing the results of a multi-step task execution for an \
+industrial asset operations system.
+
+Original question: {question}
+
+Step-by-step execution results:
+{results}
+
+Verification notes:
+{verification}
+
+Provide a concise, direct answer to the original question based on the results \
+and verification notes above. Do not repeat the individual steps — just give \
+the final answer.
+"""
+
 
 class PlanExecuteRunner(AgentRunner):
     """Entry-point for plan-and-execute workflows using MCP servers as tool providers.
@@ -95,14 +130,19 @@ class PlanExecuteRunner(AgentRunner):
                       names the planner will assign steps to.  Values are
                       either a uv entry-point name (str) or a Path to a
                       script file.  Defaults to all five registered servers.
+        adaptive_escalation: Enable an experimental deterministic escalation
+                             policy that can add a verification pass before
+                             summarisation. Defaults to ``False``.
     """
 
     def __init__(
         self,
         llm: LLMBackend,
         server_paths: dict[str, Path | str] | None = None,
+        adaptive_escalation: bool = False,
     ) -> None:
         super().__init__(llm, server_paths)
+        self._adaptive_escalation = adaptive_escalation
         self._meter = _TokenMeter(llm)
         self._planner = Planner(self._meter)
         self._executor = Executor(self._meter, server_paths)
@@ -143,6 +183,31 @@ async def run(self, question: str) -> OrchestratorResult:
             # 3. Execute
             trajectory = await self._executor.execute_plan(plan, question)
 
+            span.set_attribute("agent.escalation.enabled", self._adaptive_escalation)
+            verification = ""
+            escalation_decision = None
+            if self._adaptive_escalation:
+                signals = extract_escalation_signals(question, plan, trajectory)
+                escalation_decision = should_escalate(signals)
+                span.set_attribute(
+                    "agent.escalation.should_escalate",
+                    escalation_decision.should_escalate,
+                )
+                span.set_attribute(
+                    "agent.escalation.reasons",
+                    escalation_decision.reasons,
+                )
+                span.set_attribute(
+                    "agent.escalation.dependency_depth", signals.dependency_depth
+                )
+                span.set_attribute(
+                    "agent.escalation.any_step_failed", signals.any_step_failed
+                )
+                span.set_attribute(
+                    "agent.escalation.uses_specialist_servers",
+                    signals.uses_specialist_servers,
+                )
+
             # 4. Summarise
             _log.info("Summarising...")
             results_text = "\n\n".join(
@@ -151,9 +216,26 @@ async def run(self, question: str) -> OrchestratorResult:
                 for r in trajectory
             )
             summarization_started = time.perf_counter()
-            answer = self._meter.generate(
-                _SUMMARIZE_PROMPT.format(question=question, results=results_text)
-            )
+            if escalation_decision and escalation_decision.should_escalate:
+                _log.info("Running adaptive escalation verification...")
+                verification = self._meter.generate(
+                    _VERIFY_PROMPT.format(
+                        question=question,
+                        reasons="\n".join(escalation_decision.reasons),
+                        results=results_text,
+                    )
+                )
+                answer = self._meter.generate(
+                    _SUMMARIZE_WITH_VERIFICATION_PROMPT.format(
+                        question=question,
+                        results=results_text,
+                        verification=verification,
+                    )
+                )
+            else:
+                answer = self._meter.generate(
+                    _SUMMARIZE_PROMPT.format(question=question, results=results_text)
+                )
             summarization_ms = (time.perf_counter() - summarization_started) * 1000
             duration_ms = (time.perf_counter() - run_started) * 1000