From 7336b3b52f79ac5ce0064a9a871a02050ea65eff Mon Sep 17 00:00:00 2001 From: Hem Vadgama Date: Tue, 1 Sep 2026 17:57:00 -0400 Subject: [PATCH 1/3] feat(plan-execute): add evidence-driven bounded recovery Signed-off-by: Hem Vadgama --- src/agent/plan_execute/escalation.py | 247 ++++++++++ src/agent/plan_execute/executor.py | 345 ++++++++++++-- src/agent/plan_execute/models.py | 32 ++ src/agent/plan_execute/planner.py | 6 + src/agent/plan_execute/runner.py | 169 ++++++- src/agent/tests/test_adaptive_recovery.py | 530 ++++++++++++++++++++++ src/agent/tests/test_escalation.py | 241 ++++++++++ src/agent/tests/test_planner.py | 14 + src/agent/tests/test_runner.py | 148 ++++++ src/servers/iot/main.py | 32 +- src/servers/iot/tests/test_tools.py | 7 + src/servers/wo/main.py | 20 +- src/servers/wo/tests/test_workorders.py | 15 + 13 files changed, 1755 insertions(+), 51 deletions(-) create mode 100644 src/agent/plan_execute/escalation.py create mode 100644 src/agent/tests/test_adaptive_recovery.py create mode 100644 src/agent/tests/test_escalation.py diff --git a/src/agent/plan_execute/escalation.py b/src/agent/plan_execute/escalation.py new file mode 100644 index 000000000..7aac88197 --- /dev/null +++ b/src/agent/plan_execute/escalation.py @@ -0,0 +1,247 @@ +"""Deterministic escalation signal extraction for plan-execute runs.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Iterable + +from .models import Plan, StepFailureKind, 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) + failure_kinds: list[str] = field(default_factory=list) + recovery_attempted_steps: list[int] = field(default_factory=list) + recovered_steps: list[int] = field(default_factory=list) + retry_blocked_steps: list[int] = field(default_factory=list) + retry_exhausted_steps: list[int] = field(default_factory=list) + + @property + def has_domain_terms(self) -> bool: + return bool(self.matched_terms) + + +class EscalationAction(StrEnum): + """Bounded action selected after execution risk assessment.""" + + NONE = "none" + RETRY_STEP = "retry_step" + VERIFY = "verify" + REPORT_FAILURE = "report_failure" + + +@dataclass +class EscalationDecision: + """Deterministic escalation decision for a plan-execute run.""" + + action: EscalationAction = EscalationAction.NONE + reasons: list[str] = field(default_factory=list) + step_numbers: list[int] = field(default_factory=list) + + @property + def should_escalate(self) -> bool: + """Compatibility flag for callers that only need a routing boolean.""" + return self.action is not EscalationAction.NONE + + +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] + recovered_steps = [ + result.step_number for result in results if result.recovery_succeeded + ] + + 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), + failure_kinds=_unique_sorted( + result.failure_kind.value for result in results if result.failure_kind + ), + recovery_attempted_steps=[ + result.step_number for result in results if result.recovery_attempted + ], + recovered_steps=recovered_steps, + retry_blocked_steps=[ + result.step_number for result in results if result.retry_blocked + ], + retry_exhausted_steps=[ + result.step_number for result in results if result.retry_exhausted + ], + ) + + +def should_escalate(signals: EscalationSignals) -> EscalationDecision: + """Choose an action from direct execution evidence. + + Specialist-server use, dependency depth, and domain words remain observable + diagnostics but do not route. They do not establish a correctable failure. + """ + if signals.any_step_failed: + hard_failure_kinds = { + StepFailureKind.FAILED_DEPENDENCY.value, + StepFailureKind.UNSUPPORTED_CAPABILITY.value, + } + hard_failure = bool(hard_failure_kinds.intersection(signals.failure_kinds)) + if hard_failure or signals.retry_exhausted_steps: + return EscalationDecision( + action=EscalationAction.REPORT_FAILURE, + reasons=["required execution evidence unavailable"], + step_numbers=signals.failed_steps, + ) + if signals.retry_blocked_steps: + return EscalationDecision( + action=EscalationAction.VERIFY, + reasons=["automatic retry prohibited by tool safety"], + step_numbers=signals.retry_blocked_steps, + ) + return EscalationDecision( + action=EscalationAction.REPORT_FAILURE, + reasons=["unresolved execution failure"], + step_numbers=signals.failed_steps, + ) + + if signals.recovered_steps: + return EscalationDecision( + action=EscalationAction.RETRY_STEP, + reasons=["bounded safe recovery succeeded"], + step_numbers=signals.recovered_steps, + ) + + return EscalationDecision() + + +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"(? list[str]: + return sorted({value for value in values if value}) + + +__all__ = [ + "DEFAULT_ESCALATION_TERMS", + "DEFAULT_SPECIALIST_SERVERS", + "EscalationAction", + "EscalationDecision", + "EscalationSignals", + "extract_escalation_signals", + "should_escalate", +] diff --git a/src/agent/plan_execute/executor.py b/src/agent/plan_execute/executor.py index 3da5b9c29..49471ff23 100644 --- a/src/agent/plan_execute/executor.py +++ b/src/agent/plan_execute/executor.py @@ -16,11 +16,12 @@ from llm import LLMBackend from ..runner import DEFAULT_SERVER_PATHS -from .models import Plan, PlanStep, StepResult +from .models import Plan, PlanStep, RetrySafety, StepFailureKind, StepResult _log = logging.getLogger(__name__) _REPO_ROOT = Path(__file__).parent.parent.parent.parent +_SOURCE_ROOT = _REPO_ROOT / "src" _PLACEHOLDER_RE = re.compile(r"\{step_(\d+)\}") @@ -33,20 +34,44 @@ Question: {question} Tool: {tool} +Tool description: {tool_description} Tool parameters: {tool_schema} Task: {task} Prior step results: {context} +{repair_feedback} + YOUR RESPONSE MUST BE A SINGLE RAW JSON OBJECT AND NOTHING ELSE. Do not write any explanation, reasoning, or prose — output only the JSON object. Use EXACTLY the parameter names listed in "Tool parameters" above. Use the task description and prior step results to determine the correct argument values. If a value comes from a list, use the first relevant element. +Treat optional parameters as filters: omit them when the question requests all values. +Never invent a placeholder value for an optional filter. Include a filter only when +its exact value is supported by the question, tool description, or prior evidence. +For identifiers, use the exact canonical value from prior results; do not paraphrase it. JSON:""" +_SIMPLE_JSON_TYPES: dict[str, type | tuple[type, ...]] = { + "string": str, + "integer": int, + "number": (int, float), + "boolean": bool, + "array": list, + "object": dict, +} + + +class ArgumentResolutionError(ValueError): + """The argument model did not return a usable JSON object.""" + + +class ToolArgumentValidationError(ValueError): + """Generated arguments do not satisfy the advertised input schema.""" + class Executor: """Executes plan steps by routing tool calls to MCP servers.""" @@ -79,33 +104,39 @@ async def get_server_descriptions(self) -> dict[str, str]: descriptions[name] = f" (unavailable: {exc})" return descriptions - async def execute_plan(self, plan: Plan, question: str) -> list[StepResult]: - """Execute all plan steps in dependency order.""" + async def execute_plan( + self, + plan: Plan, + question: str, + *, + adaptive_recovery: bool = False, + max_recovery_attempts: int = 2, + ) -> list[StepResult]: + """Execute all plan steps in dependency order. + + Adaptive recovery is opt-in. It permits at most one retry per failed + step and also enforces the run-wide ``max_recovery_attempts`` budget. + """ ordered = plan.resolved_order() total = len(ordered) # Pre-fetch tool schemas for all servers referenced in the plan so that # _resolve_args_with_llm can include exact parameter names in its prompt. server_names = {step.server for step in ordered} - tool_schemas: dict[str, dict[str, str]] = {} # server -> {tool_name -> sig} + tool_specs: dict[str, dict[str, dict[str, Any]]] = {} for name in server_names: path = self._server_paths.get(name) if path is None: continue try: tools = await _list_tools(path) - tool_schemas[name] = { - t["name"]: ", ".join( - f"{p['name']}: {p['type']}{'?' if not p['required'] else ''}" - for p in t.get("parameters", []) - ) - for t in tools - } + tool_specs[name] = {t["name"]: t for t in tools} except Exception: # noqa: BLE001 - tool_schemas[name] = {} + tool_specs[name] = {} context: dict[int, StepResult] = {} results: list[StepResult] = [] + recovery_attempts = 0 for step in ordered: _log.info( "Step %d/%d [%s]: %s", @@ -114,11 +145,95 @@ async def execute_plan(self, plan: Plan, question: str) -> list[StepResult]: step.server, step.task, ) - schema = tool_schemas.get(step.server, {}).get(step.tool, "") step_started = time.perf_counter() - result = await self.execute_step( - step, context, question, tool_schema=schema - ) + unavailable_dependencies = [ + dependency + for dependency in step.dependencies + if dependency not in context or not context[dependency].success + ] + tool_spec = tool_specs.get(step.server, {}).get(step.tool) + schema = _format_tool_schema(tool_spec) + description = str((tool_spec or {}).get("description") or "").strip() + + if adaptive_recovery and unavailable_dependencies: + result = StepResult( + step_number=step.step_number, + task=step.task, + server=step.server, + response="", + error=( + "Blocked by unavailable dependencies: " + + ", ".join(str(n) for n in unavailable_dependencies) + ), + tool=step.tool, + tool_args=step.tool_args, + failure_kind=StepFailureKind.FAILED_DEPENDENCY, + attempt_count=0, + ) + elif ( + adaptive_recovery + and step.tool + and step.tool.lower() not in {"none", "null"} + and step.server in self._server_paths + and tool_spec is None + ): + result = StepResult( + step_number=step.step_number, + task=step.task, + server=step.server, + response="", + error=f"Tool '{step.tool}' is not advertised by server '{step.server}'", + tool=step.tool, + tool_args=step.tool_args, + failure_kind=StepFailureKind.UNSUPPORTED_CAPABILITY, + ) + else: + result = await self.execute_step( + step, + context, + question, + tool_schema=schema, + tool_description=description, + tool_spec=tool_spec, + detect_failures=adaptive_recovery, + ) + + if adaptive_recovery and not result.success: + retry_safety = _retry_safety(result, tool_spec) + result.retry_safety = retry_safety + may_retry = retry_safety in { + RetrySafety.SAFE_PRE_CALL, + RetrySafety.READ_ONLY, + } + if may_retry and recovery_attempts < max_recovery_attempts: + recovery_attempts += 1 + retry = await self.execute_step( + step, + context, + question, + tool_schema=schema, + tool_description=description, + tool_spec=tool_spec, + detect_failures=True, + repair_feedback=_repair_feedback(result), + ) + retry.attempt_count = result.attempt_count + 1 + retry.recovery_attempted = True + retry.initial_error = result.error + retry.recovery_succeeded = retry.success + retry.retry_safety = ( + retry_safety + if retry.success + else _retry_safety(retry, tool_spec) + ) + retry.retry_exhausted = not retry.success + result = retry + elif may_retry: + result.retry_blocked = True + result.retry_exhausted = True + elif retry_safety in {RetrySafety.MUTATING, RetrySafety.UNKNOWN}: + result.retry_blocked = True + result.duration_ms = (time.perf_counter() - step_started) * 1000 if result.success: _log.info("Step %d OK.", step.step_number) @@ -134,6 +249,10 @@ async def execute_step( context: dict[int, StepResult], question: str, tool_schema: str = "", + tool_description: str = "", + tool_spec: dict[str, Any] | None = None, + detect_failures: bool = False, + repair_feedback: str = "", ) -> StepResult: """Execute a single plan step. @@ -142,6 +261,16 @@ async def execute_step( 3. Call the LLM to generate tool arguments from the task and prior results. 4. Call the tool and return its result. """ + if not step.tool or step.tool.lower() in ("none", "null"): + return StepResult( + step_number=step.step_number, + task=step.task, + server=step.server, + response=step.expected_output, + tool=step.tool, + tool_args=step.tool_args, + ) + server_path = self._server_paths.get(step.server) if server_path is None: return StepResult( @@ -153,25 +282,67 @@ async def execute_step( f"Unknown server '{step.server}'. " f"Registered servers: {list(self._server_paths)}" ), + tool=step.tool, + tool_args=step.tool_args, + failure_kind=StepFailureKind.UNSUPPORTED_CAPABILITY, ) - if not step.tool or step.tool.lower() in ("none", "null"): + try: + _log.info("Step %d: calling LLM to resolve args.", step.step_number) + resolved_args = await _resolve_args_with_llm( + question, + step.task, + step.tool, + tool_schema, + context, + self._llm, + tool_description=tool_description, + repair_feedback=repair_feedback, + require_valid_json=detect_failures, + ) + # Optional MCP arguments should be omitted rather than sent as + # JSON null, which overrides server defaults and can fail schema + # validation for typed parameters. + resolved_args = { + key: value for key, value in resolved_args.items() if value is not None + } + if detect_failures: + _validate_tool_args(resolved_args, tool_spec) + except Exception as exc: # noqa: BLE001 + failure_kind = ( + StepFailureKind.ARGUMENT_VALIDATION + if isinstance(exc, ToolArgumentValidationError) + else StepFailureKind.ARGUMENT_RESOLUTION + ) return StepResult( step_number=step.step_number, task=step.task, server=step.server, - response=step.expected_output, + response="", + error=str(exc), tool=step.tool, - tool_args=step.tool_args, + tool_args=locals().get("resolved_args", step.tool_args), + failure_kind=failure_kind, + retry_safety=RetrySafety.SAFE_PRE_CALL, ) try: - _log.info("Step %d: calling LLM to resolve args.", step.step_number) - resolved_args = await _resolve_args_with_llm( - question, step.task, step.tool, tool_schema, context, self._llm - ) - response = await _call_tool(server_path, step.tool, resolved_args) + if detect_failures: + response_error = _structured_tool_error(response) + if response_error: + raise RuntimeError(response_error) + if not response.strip(): + return StepResult( + step_number=step.step_number, + task=step.task, + server=step.server, + response="", + error="Tool returned empty output", + tool=step.tool, + tool_args=resolved_args, + failure_kind=StepFailureKind.EMPTY_OUTPUT, + ) return StepResult( step_number=step.step_number, task=step.task, @@ -188,7 +359,8 @@ async def execute_step( response="", error=str(exc), tool=step.tool, - tool_args=step.tool_args, + tool_args=resolved_args, + failure_kind=StepFailureKind.TOOL_ERROR, ) @@ -202,6 +374,10 @@ async def _resolve_args_with_llm( tool_schema: str, context: dict[int, StepResult], llm: LLMBackend, + *, + tool_description: str = "", + repair_feedback: str = "", + require_valid_json: bool = False, ) -> dict: """Generate tool arguments from the task description and prior step results.""" context_text = "\n".join( @@ -211,8 +387,13 @@ async def _resolve_args_with_llm( _ARG_RESOLUTION_PROMPT.replace("{question}", question) .replace("{task}", task) .replace("{tool}", tool) + .replace("{tool_description}", tool_description or "(none)") .replace("{tool_schema}", tool_schema or "(unknown)") .replace("{context}", context_text or "(none)") + .replace( + "{repair_feedback}", + repair_feedback or "No previous failed attempt is being repaired.", + ) ) raw = llm.generate(prompt) resolved = _parse_json(raw) @@ -222,6 +403,10 @@ async def _resolve_args_with_llm( tool, raw[:120], ) + if require_valid_json: + raise ArgumentResolutionError( + "Argument generation returned no parseable JSON object" + ) return {} return resolved @@ -255,6 +440,92 @@ def _parse_json(raw: str) -> dict | None: return None +def _format_tool_schema(tool_spec: dict[str, Any] | None) -> str: + if not tool_spec: + return "" + return ", ".join( + f"{parameter['name']}: {parameter['type']}" + f"{'?' if not parameter['required'] else ''}" + for parameter in tool_spec.get("parameters", []) + ) + + +def _validate_tool_args( + args: dict[str, Any], tool_spec: dict[str, Any] | None +) -> None: + """Validate required fields and simple JSON types before a tool is called.""" + if not tool_spec: + return + parameter_names = { + parameter["name"] for parameter in tool_spec.get("parameters", []) + } + unexpected = sorted(set(args) - parameter_names) + if unexpected: + raise ToolArgumentValidationError( + "Unexpected argument(s): " + ", ".join(unexpected) + ) + for parameter in tool_spec.get("parameters", []): + name = parameter["name"] + if parameter.get("required") and name not in args: + raise ToolArgumentValidationError(f"Missing required argument '{name}'") + if name not in args: + continue + expected = _SIMPLE_JSON_TYPES.get(parameter.get("type", "")) + value = args[name] + wrong_type = expected is not None and not isinstance(value, expected) + if parameter.get("type") in {"integer", "number"} and isinstance(value, bool): + wrong_type = True + if wrong_type: + raise ToolArgumentValidationError( + f"Argument '{name}' must be {parameter['type']}" + ) + + +def _structured_tool_error(response: str) -> str | None: + """Return an application error carried by a transport-successful JSON result.""" + payload = _parse_json(response) + if not payload: + return None + error = payload.get("error") + if error in (None, "", False): + return None + if isinstance(error, str): + return error + return json.dumps(error, sort_keys=True, default=str) + + +def _retry_safety( + result: StepResult, tool_spec: dict[str, Any] | None +) -> RetrySafety: + """Classify one failed attempt conservatively for automatic replay.""" + if result.failure_kind in { + StepFailureKind.ARGUMENT_RESOLUTION, + StepFailureKind.ARGUMENT_VALIDATION, + }: + return RetrySafety.SAFE_PRE_CALL + if result.failure_kind in { + StepFailureKind.FAILED_DEPENDENCY, + StepFailureKind.UNSUPPORTED_CAPABILITY, + }: + return RetrySafety.NOT_APPLICABLE + + annotations = (tool_spec or {}).get("annotations", {}) + if annotations.get("destructive") is True or annotations.get("read_only") is False: + return RetrySafety.MUTATING + if annotations.get("read_only") is True: + return RetrySafety.READ_ONLY + return RetrySafety.UNKNOWN + + +def _repair_feedback(result: StepResult) -> str: + return ( + "The previous attempt failed. Generate corrected arguments; do not repeat " + "the same mistake.\n" + f"Previous arguments: {json.dumps(result.tool_args, sort_keys=True, default=str)}\n" + f"Failure: {result.error or 'unknown'}" + ) + + # ── MCP protocol helpers ────────────────────────────────────────────────────── @@ -272,6 +543,10 @@ def _make_stdio_params(server: Path | str) -> "StdioServerParameters": command="uv", args=["run", server], cwd=str(_REPO_ROOT), + # MCP's stdio transport deliberately starts from a restricted + # environment that omits PYTHONPATH. Keep checkout entry points + # importable even when the editable-install .pth is ignored. + env={"PYTHONPATH": str(_SOURCE_ROOT)}, ) try: rel = server.relative_to(_REPO_ROOT) @@ -313,11 +588,21 @@ async def _list_tools(server_path: Path | str) -> list[dict]: "name": t.name, "description": t.description or "", "parameters": parameters, + "annotations": _tool_annotations(t), } ) return tools +def _tool_annotations(tool: Any) -> dict[str, bool | None]: + annotations = getattr(tool, "annotations", None) + return { + "read_only": getattr(annotations, "readOnlyHint", None), + "destructive": getattr(annotations, "destructiveHint", None), + "idempotent": getattr(annotations, "idempotentHint", None), + } + + async def _call_tool(server_path: Path | str, tool_name: str, args: dict) -> str: """Connect to an MCP server via stdio and call a tool.""" from mcp import ClientSession @@ -328,7 +613,15 @@ async def _call_tool(server_path: Path | str, tool_name: str, args: dict) -> str async with ClientSession(read, write) as session: await session.initialize() result = await session.call_tool(tool_name, args) - return _extract_content(result.content) + return _extract_tool_result(result) + + +def _extract_tool_result(result: Any) -> str: + """Return tool text, raising when MCP marks the result as an error.""" + text = _extract_content(result.content) + if getattr(result, "isError", False): + raise RuntimeError(text or "MCP tool call failed") + return text def _extract_content(content: list[Any]) -> str: diff --git a/src/agent/plan_execute/models.py b/src/agent/plan_execute/models.py index d1b8940e0..5ba780a9d 100644 --- a/src/agent/plan_execute/models.py +++ b/src/agent/plan_execute/models.py @@ -3,9 +3,31 @@ from __future__ import annotations from dataclasses import dataclass, field +from enum import StrEnum from typing import Optional +class StepFailureKind(StrEnum): + """Machine-readable phase in which a plan step failed.""" + + ARGUMENT_RESOLUTION = "argument_resolution" + ARGUMENT_VALIDATION = "argument_validation" + TOOL_ERROR = "tool_error" + EMPTY_OUTPUT = "empty_output" + FAILED_DEPENDENCY = "failed_dependency" + UNSUPPORTED_CAPABILITY = "unsupported_capability" + + +class RetrySafety(StrEnum): + """Why an automatic retry is allowed or prohibited.""" + + SAFE_PRE_CALL = "safe_pre_call" + READ_ONLY = "read_only" + MUTATING = "mutating" + UNKNOWN = "unknown" + NOT_APPLICABLE = "not_applicable" + + @dataclass class PlanStep: """A single step in an execution plan.""" @@ -62,6 +84,14 @@ class StepResult: tool: str = "" tool_args: dict = field(default_factory=dict) duration_ms: float | None = None + failure_kind: StepFailureKind | None = None + attempt_count: int = 1 + recovery_attempted: bool = False + recovery_succeeded: bool = False + initial_error: Optional[str] = None + retry_safety: RetrySafety = RetrySafety.NOT_APPLICABLE + retry_blocked: bool = False + retry_exhausted: bool = False @property def success(self) -> bool: @@ -76,3 +106,5 @@ class OrchestratorResult: answer: str plan: Plan trajectory: list[StepResult] + escalation_action: str | None = None + escalation_reasons: list[str] = field(default_factory=list) diff --git a/src/agent/plan_execute/planner.py b/src/agent/plan_execute/planner.py index 8c683943a..5609a8967 100644 --- a/src/agent/plan_execute/planner.py +++ b/src/agent/plan_execute/planner.py @@ -42,6 +42,12 @@ - Server and tool names must exactly match those listed above. - Dependencies use #S notation (e.g., #S1, #S2). Use "None" if none. - Keep tasks specific and actionable. +- When the question describes an identifier indirectly, first use an available + discovery or list tool to obtain the canonical identifier, then make the + consuming step depend on that evidence. +- Do not plan a capability that no listed tool provides. +- Use a "none" reasoning step only to derive an answer from earlier evidence, + and make its evidence dependencies explicit. Question: {question} diff --git a/src/agent/plan_execute/runner.py b/src/agent/plan_execute/runner.py index 445f47b09..ef99bef00 100644 --- a/src/agent/plan_execute/runner.py +++ b/src/agent/plan_execute/runner.py @@ -12,6 +12,7 @@ from __future__ import annotations +import json import logging import time from pathlib import Path @@ -19,8 +20,13 @@ from llm import LLMBackend, LLMResult from observability import agent_run_span, persist_trajectory +from .escalation import ( + EscalationAction, + extract_escalation_signals, + should_escalate, +) from .executor import Executor -from .models import OrchestratorResult +from .models import OrchestratorResult, StepResult from .planner import Planner from ..runner import AgentRunner @@ -76,6 +82,41 @@ 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. Never claim that an action completed when required \ +evidence is missing or a tool failed. If completion cannot be established, say so \ +explicitly. 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. @@ -94,15 +135,34 @@ class PlanExecuteRunner(AgentRunner): server_paths: Override MCP server specs. Keys must match the server 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. + script file. Defaults to all six registered servers. + adaptive_escalation: Enable an experimental deterministic escalation + policy that can add a verification pass before + summarisation. Defaults to ``False``. + adaptive_recovery: Override whether bounded safe recovery runs. By + default it follows ``adaptive_escalation``. The + separate control exists for paired experiments. + max_recovery_attempts: Run-wide retry budget; each step can retry once. """ def __init__( self, llm: LLMBackend, server_paths: dict[str, Path | str] | None = None, + adaptive_escalation: bool = False, + adaptive_recovery: bool | None = None, + max_recovery_attempts: int = 2, ) -> None: super().__init__(llm, server_paths) + if max_recovery_attempts < 0: + raise ValueError("max_recovery_attempts must be non-negative") + self._adaptive_escalation = adaptive_escalation + self._adaptive_recovery = ( + adaptive_escalation if adaptive_recovery is None else adaptive_recovery + ) + if self._adaptive_recovery and not adaptive_escalation: + raise ValueError("adaptive_recovery requires adaptive_escalation") + self._max_recovery_attempts = max_recovery_attempts self._meter = _TokenMeter(llm) self._planner = Planner(self._meter) self._executor = Executor(self._meter, server_paths) @@ -141,19 +201,95 @@ async def run(self, question: str) -> OrchestratorResult: _log.info("Plan has %d step(s).", len(plan.steps)) # 3. Execute - trajectory = await self._executor.execute_plan(plan, question) + trajectory = await self._executor.execute_plan( + plan, + question, + adaptive_recovery=self._adaptive_recovery, + max_recovery_attempts=self._max_recovery_attempts, + ) + + span.set_attribute("agent.escalation.enabled", self._adaptive_escalation) + 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.action", escalation_decision.action.value + ) + 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, + ) + span.set_attribute( + "agent.escalation.recovery_attempted_steps", + signals.recovery_attempted_steps, + ) + span.set_attribute( + "agent.escalation.recovered_steps", signals.recovered_steps + ) + span.set_attribute( + "agent.escalation.retry_blocked_steps", + signals.retry_blocked_steps, + ) # 4. Summarise _log.info("Summarising...") results_text = "\n\n".join( - f"Step {r.step_number} — {r.task} (server: {r.server}):\n" + f"Step {r.step_number} — {r.task} " + f"(server: {r.server}; tool: {r.tool or 'none'}; " + f"args: {json.dumps(r.tool_args, sort_keys=True, default=str)}):\n" + (r.response if r.success else f"ERROR: {r.error}") 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.action is EscalationAction.VERIFY + ): + _log.info("Running adaptive escalation verification...") + try: + 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, + ) + ) + except Exception: # noqa: BLE001 + _log.exception( + "Adaptive verification failed; reporting execution failure" + ) + answer = _failure_answer(trajectory) + elif ( + escalation_decision + and escalation_decision.action is EscalationAction.REPORT_FAILURE + ): + answer = _failure_answer(trajectory) + 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 @@ -162,6 +298,12 @@ async def run(self, question: str) -> OrchestratorResult: answer=answer, plan=plan, trajectory=trajectory, + escalation_action=( + escalation_decision.action.value if escalation_decision else None + ), + escalation_reasons=( + escalation_decision.reasons if escalation_decision else [] + ), ) span.set_attribute("agent.plan.steps", len(plan.steps)) span.set_attribute("agent.answer.length", len(answer or "")) @@ -183,3 +325,16 @@ async def run(self, question: str) -> OrchestratorResult: trajectory=trajectory, ) return result + + +def _failure_answer(trajectory: list[StepResult]) -> str: + """Return an accurate deterministic answer when required evidence is absent.""" + failures = [result for result in trajectory if not result.success] + details = "; ".join( + f"step {result.step_number} ({result.task}): {result.error or 'failed'}" + for result in failures + ) + return ( + "Unable to complete the request because required execution evidence could " + f"not be obtained. {details}" + ) diff --git a/src/agent/tests/test_adaptive_recovery.py b/src/agent/tests/test_adaptive_recovery.py new file mode 100644 index 000000000..fc7076750 --- /dev/null +++ b/src/agent/tests/test_adaptive_recovery.py @@ -0,0 +1,530 @@ +"""Deterministic adaptive-routing and bounded-recovery fixtures.""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from agent.plan_execute.executor import Executor, _tool_annotations +from agent.plan_execute.models import Plan, PlanStep +from agent.plan_execute.runner import PlanExecuteRunner +from llm import LLMBackend + + +class RecordingLLM(LLMBackend): + def __init__(self, responses): + self.responses = iter(responses) + self.prompts: list[str] = [] + + def generate(self, prompt: str, temperature: float = 0.0) -> str: + self.prompts.append(prompt) + response = next(self.responses) + if isinstance(response, Exception): + raise response + return response + + +def test_mcp_retry_safety_annotations_are_extracted_without_guessing(): + annotated = SimpleNamespace( + annotations=SimpleNamespace( + readOnlyHint=True, destructiveHint=False, idempotentHint=True + ) + ) + + assert _tool_annotations(annotated) == { + "read_only": True, + "destructive": False, + "idempotent": True, + } + assert _tool_annotations(SimpleNamespace()) == { + "read_only": None, + "destructive": None, + "idempotent": None, + } + + +def _step( + *, + tool: str = "read_asset", + server: str = "iot", + dependencies: list[int] | None = None, + task: str = "Read asset", +) -> PlanStep: + return PlanStep( + step_number=1, + task=task, + server=server, + tool=tool, + tool_args={}, + dependencies=dependencies or [], + expected_output="asset evidence", + ) + + +def _tool_spec( + *, + name: str = "read_asset", + read_only: bool | None = True, + destructive: bool | None = False, + required: bool = False, +): + return { + "name": name, + "description": "fixture tool", + "parameters": [ + {"name": "site_name", "type": "string", "required": required} + ], + "annotations": { + "read_only": read_only, + "destructive": destructive, + "idempotent": read_only, + }, + } + + +def _plan_text( + *, + server: str = "iot", + tool: str = "read_asset", + task: str = "Read asset evidence", +) -> str: + return ( + f"#Task1: {task}\n" + f"#Server1: {server}\n" + f"#Tool1: {tool}\n" + "#Dependency1: None\n" + "#ExpectedOutput1: Asset evidence\n" + ) + + +@pytest.mark.anyio +async def test_successful_shallow_execution_does_not_escalate(): + llm = RecordingLLM([_plan_text(), "{}", "asset answer"]) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch( + "agent.plan_execute.executor._call_tool", + new=AsyncMock(return_value='{"asset":"A-1"}'), + ), + ): + result = await PlanExecuteRunner( + llm, server_paths={"iot": Path("/fake.py")}, adaptive_escalation=True + ).run("Read A-1") + + assert result.escalation_action == "none" + assert len(llm.prompts) == 3 + + +@pytest.mark.anyio +async def test_successful_specialist_execution_does_not_escalate(): + llm = RecordingLLM( + [_plan_text(server="vibration", tool="diagnose"), "{}", "healthy"] + ) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec(name="diagnose")]), + ), + patch( + "agent.plan_execute.executor._call_tool", + new=AsyncMock(return_value='{"severity":"low"}'), + ), + ): + result = await PlanExecuteRunner( + llm, + server_paths={"vibration": Path("/fake.py")}, + adaptive_escalation=True, + ).run("Diagnose vibration") + + assert result.escalation_action == "none" + + +@pytest.mark.anyio +async def test_domain_words_alone_do_not_escalate(): + llm = RecordingLLM([_plan_text(task="Read work order failure"), "{}", "done"]) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch( + "agent.plan_execute.executor._call_tool", + new=AsyncMock(return_value='{"status":"complete"}'), + ), + ): + result = await PlanExecuteRunner( + llm, server_paths={"iot": Path("/fake.py")}, adaptive_escalation=True + ).run("Review the work order failure") + + assert result.escalation_action == "none" + + +@pytest.mark.anyio +async def test_failed_read_only_tool_is_retried_once(): + llm = RecordingLLM(["{}", "{}"]) + call = AsyncMock(side_effect=[RuntimeError("timeout"), '{"asset":"A-1"}']) + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.success is True + assert result.attempt_count == 2 + assert result.recovery_succeeded is True + assert call.await_count == 2 + + +@pytest.mark.anyio +async def test_semantically_wrong_arguments_are_repaired_after_explicit_error(): + llm = RecordingLLM( + ['{"site_name":"main site"}', '{"site_name":"MAIN"}'] + ) + call = AsyncMock( + side_effect=['{"error":"unknown site main site"}', '{"asset":"A-1"}'] + ) + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec(required=True)]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.success is True + assert result.tool_args == {"site_name": "MAIN"} + assert call.call_args_list[0].args[2] == {"site_name": "main site"} + assert call.call_args_list[1].args[2] == {"site_name": "MAIN"} + + +@pytest.mark.anyio +async def test_malformed_arguments_are_corrected_before_tool_call(): + llm = RecordingLLM(["not json", '{"site_name":"MAIN"}']) + call = AsyncMock(return_value='{"asset":"A-1"}') + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec(required=True)]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.success is True + assert result.initial_error == "Argument generation returned no parseable JSON object" + assert result.tool_args == {"site_name": "MAIN"} + assert call.await_count == 1 + + +@pytest.mark.anyio +async def test_unadvertised_argument_is_repaired_before_tool_call(): + llm = RecordingLLM(['{"Description":"MAIN"}', '{}']) + call = AsyncMock(return_value='{"sites":["MAIN"]}') + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.success is True + assert result.initial_error == "Unexpected argument(s): Description" + assert result.tool_args == {} + assert call.await_count == 1 + + +@pytest.mark.anyio +async def test_missing_artifact_retries_once_then_reports_failure(): + llm = RecordingLLM([_plan_text(), "{}", "{}"]) + call = AsyncMock(return_value='{"error":"No such file: evidence.json"}') + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = await PlanExecuteRunner( + llm, server_paths={"iot": Path("/fake.py")}, adaptive_escalation=True + ).run("Read evidence.json") + + assert result.escalation_action == "report_failure" + assert result.trajectory[0].retry_exhausted is True + assert call.await_count == 2 + assert result.answer.startswith("Unable to complete the request") + + +@pytest.mark.anyio +async def test_side_effecting_tool_failure_is_not_replayed(): + llm = RecordingLLM(["{}"]) + call = AsyncMock(side_effect=RuntimeError("connection dropped")) + write_spec = _tool_spec( + name="create_workorder", read_only=False, destructive=True + ) + executor = Executor(llm, server_paths={"wo": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[write_spec]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step(server="wo", tool="create_workorder")], raw=""), + "Q", + adaptive_recovery=True, + ))[0] + + assert result.success is False + assert result.retry_blocked is True + assert call.await_count == 1 + assert len(llm.prompts) == 1 + + +@pytest.mark.anyio +async def test_unknown_tool_safety_prohibits_replay(): + llm = RecordingLLM(["{}"]) + call = AsyncMock(side_effect=RuntimeError("outcome unknown")) + unknown_spec = _tool_spec(read_only=None, destructive=None) + executor = Executor(llm, server_paths={"custom": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[unknown_spec]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step(server="custom")], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.retry_safety == "unknown" + assert result.retry_blocked is True + assert call.await_count == 1 + + +@pytest.mark.anyio +async def test_empty_read_only_output_is_retried_once(): + llm = RecordingLLM(["{}", "{}"]) + call = AsyncMock(side_effect=["", '{"asset":"A-1"}']) + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.recovery_succeeded is True + assert result.initial_error == "Tool returned empty output" + assert call.await_count == 2 + + +@pytest.mark.anyio +async def test_changed_recovery_arguments_can_legitimately_return_zero_results(): + llm = RecordingLLM( + ['{"site_name":"main site"}', '{"site_name":"MAIN"}'] + ) + call = AsyncMock( + side_effect=[RuntimeError("temporary read failure"), '{"total":0,"assets":[]}'] + ) + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec(required=True)]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.success is True + assert result.recovery_succeeded is True + + +@pytest.mark.anyio +async def test_reasoning_only_step_is_handled_normally(): + executor = Executor(RecordingLLM([]), server_paths={}) + result = await executor.execute_step( + _step(server="none", tool="none", task="Reason over evidence"), {}, "Q" + ) + + assert result.success is True + assert result.response == "asset evidence" + + +@pytest.mark.anyio +async def test_verifier_failure_falls_back_to_deterministic_failure(): + llm = RecordingLLM( + [_plan_text(server="wo", tool="create_workorder"), "{}", RuntimeError("LLM down")] + ) + write_spec = _tool_spec( + name="create_workorder", read_only=False, destructive=True + ) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[write_spec]), + ), + patch( + "agent.plan_execute.executor._call_tool", + new=AsyncMock(side_effect=RuntimeError("outcome unknown")), + ), + ): + result = await PlanExecuteRunner( + llm, server_paths={"wo": Path("/fake.py")}, adaptive_escalation=True + ).run("Create a work order") + + assert result.escalation_action == "verify" + assert result.answer.startswith("Unable to complete the request") + + +@pytest.mark.anyio +async def test_recovery_model_failure_is_captured_and_bounded(): + llm = RecordingLLM(["{}", RuntimeError("recovery model down")]) + call = AsyncMock(side_effect=RuntimeError("read timeout")) + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = (await executor.execute_plan( + Plan([_step()], raw=""), "Q", adaptive_recovery=True + ))[0] + + assert result.success is False + assert result.recovery_attempted is True + assert result.retry_exhausted is True + assert result.error == "recovery model down" + assert call.await_count == 1 + + +@pytest.mark.anyio +async def test_recovery_evidence_reaches_final_summarization(): + llm = RecordingLLM( + [_plan_text(), "not json", '{"site_name":"MAIN"}', "A-1 recovered"] + ) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec(required=True)]), + ), + patch( + "agent.plan_execute.executor._call_tool", + new=AsyncMock(return_value='{"asset":"A-1","site":"MAIN"}'), + ), + ): + result = await PlanExecuteRunner( + llm, server_paths={"iot": Path("/fake.py")}, adaptive_escalation=True + ).run("Read A-1") + + assert result.answer == "A-1 recovered" + assert result.escalation_action == "retry_step" + assert '"asset":"A-1"' in llm.prompts[-1] + assert 'args: {"site_name": "MAIN"}' in llm.prompts[-1] + + +@pytest.mark.anyio +async def test_failed_recovery_is_bounded_and_accurately_reported(): + llm = RecordingLLM([_plan_text(), "{}", "{}"]) + call = AsyncMock(side_effect=RuntimeError("still unavailable")) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = await PlanExecuteRunner( + llm, server_paths={"iot": Path("/fake.py")}, adaptive_escalation=True + ).run("Read A-1") + + assert call.await_count == 2 + assert result.trajectory[0].attempt_count == 2 + assert result.trajectory[0].retry_exhausted is True + assert "still unavailable" in result.answer + + +@pytest.mark.anyio +async def test_adaptive_disabled_preserves_legacy_bad_json_fallback(): + llm = RecordingLLM([_plan_text(), "not json", "legacy answer"]) + call = AsyncMock(return_value='{"asset":"legacy"}') + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec(required=True)]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + result = await PlanExecuteRunner( + llm, server_paths={"iot": Path("/fake.py")}, adaptive_escalation=False + ).run("Read A-1") + + assert result.answer == "legacy answer" + assert result.escalation_action is None + assert result.trajectory[0].tool_args == {} + assert call.await_count == 1 + + +@pytest.mark.anyio +async def test_failed_dependency_is_not_executed_in_adaptive_mode(): + first = _step() + second = PlanStep( + step_number=2, + task="Use asset evidence", + server="iot", + tool="read_asset", + tool_args={}, + dependencies=[1], + expected_output="derived evidence", + ) + llm = RecordingLLM(["{}", "{}"]) + call = AsyncMock(side_effect=RuntimeError("offline")) + executor = Executor(llm, server_paths={"iot": Path("/fake.py")}) + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[_tool_spec()]), + ), + patch("agent.plan_execute.executor._call_tool", new=call), + ): + results = await executor.execute_plan( + Plan([first, second], raw=""), "Q", adaptive_recovery=True + ) + + assert call.await_count == 2 # initial call plus one bounded retry for step 1 + assert results[1].failure_kind == "failed_dependency" + assert results[1].attempt_count == 0 diff --git a/src/agent/tests/test_escalation.py b/src/agent/tests/test_escalation.py new file mode 100644 index 000000000..16d8d065b --- /dev/null +++ b/src/agent/tests/test_escalation.py @@ -0,0 +1,241 @@ +"""Tests for deterministic escalation signal extraction.""" + +from agent.plan_execute.escalation import ( + DEFAULT_SPECIALIST_SERVERS, + EscalationAction, + EscalationDecision, + extract_escalation_signals, + should_escalate, +) +from agent.plan_execute.models import Plan, PlanStep, StepResult +from agent.runner import DEFAULT_SERVER_PATHS + + +def _step( + n: int, + server: str = "iot", + tool: str = "sites", + deps: list[int] | None = None, + task: str | None = None, + expected_output: str = "output", +) -> PlanStep: + return PlanStep( + step_number=n, + task=task or f"Task {n}", + server=server, + tool=tool, + tool_args={}, + dependencies=deps or [], + expected_output=expected_output, + ) + + +def test_extracts_step_count_and_dependency_depth(): + plan = Plan( + steps=[ + _step(1), + _step(2, deps=[1]), + _step(3, deps=[2]), + _step(4, deps=[1]), + ], + raw="", + ) + + signals = extract_escalation_signals("Q", plan) + + assert signals.step_count == 4 + assert signals.dependency_depth == 3 + + +def test_empty_plan_has_zero_dependency_depth(): + signals = extract_escalation_signals("Q", Plan(steps=[], raw="")) + + assert signals.step_count == 0 + assert signals.dependency_depth == 0 + + +def test_detects_specialist_servers(): + plan = Plan( + steps=[ + _step(1, server="iot"), + _step(2, server="wo", tool="work_orders"), + _step(3, server="vibration", tool="analyze"), + ], + raw="", + ) + + signals = extract_escalation_signals("Q", plan) + + assert signals.uses_specialist_servers is True + assert signals.specialist_servers_used == ["vibration", "wo"] + + +def test_default_specialist_servers_match_registered_server_names(): + assert DEFAULT_SPECIALIST_SERVERS <= set(DEFAULT_SERVER_PATHS) + + +def test_collects_servers_and_tools_from_plan_and_trajectory(): + plan = Plan( + steps=[ + _step(1, server="iot", tool="assets"), + _step(2, server="utilities", tool="current_date_time"), + ], + raw="", + ) + trajectory = [ + StepResult( + step_number=1, + task="Task 1", + server="iot", + response="ok", + tool="assets", + ), + StepResult( + step_number=2, + task="Task 2", + server="fmsr", + response="ok", + tool="diagnose_failure", + ), + ] + + signals = extract_escalation_signals("Q", plan, trajectory) + + assert signals.servers_used == ["fmsr", "iot", "utilities"] + assert signals.tools_used == ["assets", "current_date_time", "diagnose_failure"] + + +def test_detects_failed_steps_from_trajectory(): + plan = Plan(steps=[_step(1), _step(2)], raw="") + trajectory = [ + StepResult(step_number=1, task="Task 1", server="iot", response="ok"), + StepResult( + step_number=2, + task="Task 2", + server="iot", + response="", + error="timeout", + ), + ] + + signals = extract_escalation_signals("Q", plan, trajectory) + + assert signals.any_step_failed is True + assert signals.failed_steps == [2] + + +def test_matches_domain_terms_across_question_plan_and_trajectory(): + plan = Plan( + steps=[ + _step( + 1, + task="Open work order history", + expected_output="Recent maintenance records", + ) + ], + raw="#Task1: Run diagnostics", + ) + trajectory = [ + StepResult( + step_number=1, + task="Task 1", + server="iot", + response="Asset reported a pump failure alarm", + ) + ] + + signals = extract_escalation_signals("Any anomaly on CH-1?", plan, trajectory) + + assert signals.has_domain_terms is True + assert signals.matched_terms == [ + "work order", + "diagnostics", + "failure", + "alarm", + "anomaly", + ] + + +def test_custom_specialist_servers_and_terms_are_supported(): + plan = Plan(steps=[_step(1, server="custom", task="Check severe drift")], raw="") + + signals = extract_escalation_signals( + "Q", + plan, + specialist_servers={"custom"}, + escalation_terms=["severe drift"], + ) + + assert signals.uses_specialist_servers is True + assert signals.specialist_servers_used == ["custom"] + assert signals.matched_terms == ["severe drift"] + + +def test_matched_terms_are_case_insensitive_and_deduplicated(): + plan = Plan(steps=[_step(1, task="Investigate FAILURE alarm")], raw="") + + signals = extract_escalation_signals( + "Failure reported", + plan, + escalation_terms=["failure", "Failure", "alarm"], + ) + + assert signals.matched_terms == ["failure", "alarm"] + + +def test_escalation_decision_dataclass_is_available(): + decision = EscalationDecision( + action=EscalationAction.REPORT_FAILURE, reasons=["failed step"] + ) + + assert decision.should_escalate is True + assert decision.action is EscalationAction.REPORT_FAILURE + assert decision.reasons == ["failed step"] + + +def test_policy_escalates_on_failed_steps(): + plan = Plan(steps=[_step(1)], raw="") + trajectory = [ + StepResult( + step_number=1, + task="Task 1", + server="iot", + response="", + error="timeout", + ) + ] + + decision = should_escalate(extract_escalation_signals("Q", plan, trajectory)) + + assert decision.should_escalate is True + assert decision.action is EscalationAction.REPORT_FAILURE + assert decision.reasons == ["unresolved execution failure"] + + +def test_policy_does_not_escalate_on_specialist_server_usage(): + plan = Plan(steps=[_step(1, server="vibration")], raw="") + + decision = should_escalate(extract_escalation_signals("Q", plan)) + + assert decision.should_escalate is False + assert decision.action is EscalationAction.NONE + assert decision.reasons == [] + + +def test_policy_does_not_escalate_on_domain_terms(): + plan = Plan(steps=[_step(1, task="Review work order history")], raw="") + + decision = should_escalate(extract_escalation_signals("Q", plan)) + + assert decision.should_escalate is False + assert decision.action is EscalationAction.NONE + assert decision.reasons == [] + + +def test_policy_does_not_escalate_simple_low_risk_plan(): + plan = Plan(steps=[_step(1, task="List sites", expected_output="Site list")], raw="") + + decision = should_escalate(extract_escalation_signals("Q", plan)) + + assert decision.should_escalate is False + assert decision.reasons == [] diff --git a/src/agent/tests/test_planner.py b/src/agent/tests/test_planner.py index 6ce6e78de..dd8767860 100644 --- a/src/agent/tests/test_planner.py +++ b/src/agent/tests/test_planner.py @@ -189,3 +189,17 @@ def test_generate_plan_prompt_does_not_mention_args(self, mock_llm): Planner(llm).generate_plan("Q", {"iot": " - sites(): List sites"}) assert "#Args" not in captured[0] + + def test_generate_plan_prompt_requires_evidence_for_indirect_identifiers( + self, mock_llm + ): + captured = [] + llm = mock_llm(_TWO_STEP) + original = llm.generate + llm.generate = lambda p, **kw: (captured.append(p), original(p))[1] + + Planner(llm).generate_plan("Count work at the main site", {"iot": "sites()"}) + + assert "canonical" in captured[0] + assert "discovery or list tool" in captured[0] + assert "Do not plan a capability that no listed tool provides" in captured[0] diff --git a/src/agent/tests/test_runner.py b/src/agent/tests/test_runner.py index 65090a93f..192b0bae4 100644 --- a/src/agent/tests/test_runner.py +++ b/src/agent/tests/test_runner.py @@ -3,12 +3,16 @@ from __future__ import annotations import json +from pathlib import Path +from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from agent.plan_execute.executor import ( Executor, + _extract_tool_result, + _make_stdio_params, _parse_json, _parse_tool_call, _resolve_args, @@ -96,6 +100,18 @@ def generate(self, prompt: str, **_kw) -> str: return self._response +class _RecordingSequentialLLM(LLMBackend): + """Records prompts while returning canned responses in order.""" + + def __init__(self, responses: list[str]) -> None: + self.prompts: list[str] = [] + self._responses = iter(responses) + + def generate(self, prompt: str, temperature: float = 0.0) -> str: + self.prompts.append(prompt) + return next(self._responses, "") + + # ── orchestrator tests ──────────────────────────────────────────────────────── @@ -146,6 +162,70 @@ async def test_orchestrator_unknown_server_recorded_as_error(sequential_llm): assert "ghost" in result.trajectory[0].error +def test_orchestrator_adaptive_escalation_disabled_by_default(mock_llm): + runner = PlanExecuteRunner(mock_llm()) + + assert runner._adaptive_escalation is False + + +@pytest.mark.anyio +async def test_orchestrator_adaptive_escalation_no_extra_call_for_low_risk_plan(): + plan = ( + "#Task1: List sites\n" + "#Server1: iot\n" + "#Tool1: sites\n" + "#Dependency1: None\n" + "#ExpectedOutput1: Site list\n" + ) + llm = _RecordingSequentialLLM([plan, "{}", _FINAL_ANSWER]) + + with _patch_mcp()[0], _patch_mcp()[1]: + result = await PlanExecuteRunner(llm, adaptive_escalation=True).run("Q") + + assert result.answer == _FINAL_ANSWER + assert len(llm.prompts) == 3 + assert not any("Verification notes" in prompt for prompt in llm.prompts) + + +@pytest.mark.anyio +async def test_orchestrator_adaptive_escalation_runs_verification_when_enabled(): + plan = ( + "#Task1: Check vibration trend\n" + "#Server1: vibration\n" + "#Tool1: analyze\n" + "#Dependency1: None\n" + "#ExpectedOutput1: Vibration analysis\n" + ) + verification = "Verification: evidence is limited." + final_answer = "Final answer with verification." + llm = _RecordingSequentialLLM([plan, "{}", verification, final_answer]) + mutating_tool = { + "name": "analyze", + "description": "Analyze and record a vibration assessment", + "parameters": [], + "annotations": {"read_only": False, "destructive": True}, + } + + with ( + patch( + "agent.plan_execute.executor._list_tools", + new=AsyncMock(return_value=[mutating_tool]), + ), + patch( + "agent.plan_execute.executor._call_tool", + new=AsyncMock(side_effect=RuntimeError("write outcome unknown")), + ), + ): + result = await PlanExecuteRunner(llm, adaptive_escalation=True).run("Q") + + assert result.answer == final_answer + assert len(llm.prompts) == 4 + assert "Escalation reasons:" in llm.prompts[2] + assert "automatic retry prohibited by tool safety" in llm.prompts[2] + assert verification in llm.prompts[3] + assert result.escalation_action == "verify" + + class _UsageReportingLLM(LLMBackend): """Sequential LLM that reports per-call token usage via LLMResult.""" @@ -201,6 +281,16 @@ async def test_orchestrator_no_tool_returns_expected_output(sequential_llm): # ── executor unit tests ─────────────────────────────────────────────────────── +def test_entry_point_server_receives_checkout_pythonpath(): + params = _make_stdio_params("utilities-mcp-server") + + repo_root = Path(__file__).resolve().parents[3] + assert params.command == "uv" + assert params.args == ["run", "utilities-mcp-server"] + assert params.cwd == str(repo_root) + assert params.env == {"PYTHONPATH": str(repo_root / "src")} + + @pytest.mark.anyio async def test_executor_unknown_server(mock_llm): executor = Executor(mock_llm(""), server_paths={}) @@ -245,6 +335,19 @@ async def test_executor_no_tool_step_skips_llm(): assert llm.prompts == [] # LLM was never called +@pytest.mark.anyio +async def test_executor_no_tool_step_does_not_require_registered_server(): + llm = _CapturingLLM() + executor = Executor(llm, server_paths={}) + + step = _make_step(1, server="none", tool="none", expected_output="42") + result = await executor.execute_step(step, {}, "Q") + + assert result.response == "42" + assert result.success is True + assert llm.prompts == [] + + @pytest.mark.anyio async def test_executor_step_result_carries_resolved_args(sequential_llm): """StepResult.tool_args must reflect the args the LLM generated, not {}.""" @@ -293,6 +396,33 @@ async def test_executor_tool_call_exception_recorded_as_error(sequential_llm): assert "timeout" in result.error +@pytest.mark.anyio +async def test_executor_omits_null_optional_args(sequential_llm): + llm = sequential_llm( + ['{"site_id": "MAIN", "page_size": null, "page_num": null}'] + ) + executor = Executor(llm, server_paths={"wo": Path("/fake/server.py")}) + step = _make_step(1, server="wo", tool="list_workorders") + call_mock = AsyncMock(return_value="{}") + + with patch("agent.plan_execute.executor._call_tool", new=call_mock): + result = await executor.execute_step(step, {}, "List work orders") + + assert result.success is True + assert result.tool_args == {"site_id": "MAIN"} + assert call_mock.call_args.args[2] == {"site_id": "MAIN"} + + +def test_extract_tool_result_raises_for_mcp_error(): + result = SimpleNamespace( + isError=True, + content=[SimpleNamespace(text="argument validation failed")], + ) + + with pytest.raises(RuntimeError, match="argument validation failed"): + _extract_tool_result(result) + + @pytest.mark.anyio async def test_executor_calls_llm_to_generate_args(sequential_llm): """Each tool step triggers exactly one LLM call for arg generation.""" @@ -519,6 +649,24 @@ async def test_resolve_args_with_llm_schema_in_prompt(): assert "site_name: string" in llm.prompts[0] +@pytest.mark.anyio +async def test_resolve_args_prompt_omits_invented_optional_filters(): + llm = _CapturingLLM('{}') + await _resolve_args_with_llm( # type: ignore[arg-type] + "Count all work orders", + "List work orders", + "list_workorders", + "status: string?", + {}, + llm, + tool_description="status accepts OPEN / APPROVED_PENDING", + ) + + assert "Treat optional parameters as filters" in llm.prompts[0] + assert "Never invent a placeholder value" in llm.prompts[0] + assert "status accepts OPEN / APPROVED_PENDING" in llm.prompts[0] + + @pytest.mark.anyio async def test_resolve_args_with_llm_unknown_schema_shows_sentinel(): """Empty schema renders as '(unknown)' in the prompt.""" diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 9e62f578e..cd010d945 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -6,6 +6,7 @@ import couchdb3 from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP +from mcp.types import ToolAnnotations from servers.iot.models import ( AssetDetail, @@ -91,6 +92,13 @@ PAGE_SIZE = 1000 RESERVED_FIELDS = {"_id", "_rev", "asset_id", "timestamp", "dataset", "type", "doctype"} +_READ_ONLY_ANNOTATIONS = ToolAnnotations( + readOnlyHint=True, + destructiveHint=False, + idempotentHint=True, + openWorldHint=True, +) + _registry_sites_cache: Optional[List[str]] = None _sensor_list_cache: Dict[str, List[str]] = {} @@ -181,7 +189,7 @@ def _installed_sensors(asset_id: str, site_name: Optional[str] = None) -> List[s return [] -@mcp.tool(title="List Sites") +@mcp.tool(title="List Sites", annotations=_READ_ONLY_ANNOTATIONS) def sites() -> SitesResult: """List sorted site identifiers available in the asset registry. @@ -192,7 +200,7 @@ def sites() -> SitesResult: return SitesResult(sites=known_sites()) -@mcp.tool(title="List Asset IDs") +@mcp.tool(title="List Asset IDs", annotations=_READ_ONLY_ANNOTATIONS) def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: """List asset identifiers registered at one site. @@ -227,7 +235,7 @@ def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: return ErrorResult(error=str(e)) -@mcp.tool(title="Get Asset Detail") +@mcp.tool(title="Get Asset Detail", annotations=_READ_ONLY_ANNOTATIONS) def asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: """Return registry details for one asset. @@ -298,7 +306,7 @@ def asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResul return ErrorResult(error=str(e)) -@mcp.tool(title="List Measured Sensors") +@mcp.tool(title="List Measured Sensors", annotations=_READ_ONLY_ANNOTATIONS) def measured_sensors( site_name: str, asset_id: str ) -> Union[SensorsResult, ErrorResult]: @@ -336,7 +344,7 @@ def measured_sensors( ) -@mcp.tool(title="List Installed Sensors") +@mcp.tool(title="List Installed Sensors", annotations=_READ_ONLY_ANNOTATIONS) def installed_sensors( site_name: str, asset_id: str ) -> Union[SensorsResult, ErrorResult]: @@ -380,7 +388,7 @@ def installed_sensors( return ErrorResult(error=str(e)) -@mcp.tool(title="List Assets") +@mcp.tool(title="List Assets", annotations=_READ_ONLY_ANNOTATIONS) def assets( site_name: str, assettype: Optional[str] = None ) -> Union[AssetsWithMetadataResult, ErrorResult]: @@ -438,7 +446,7 @@ def assets( return ErrorResult(error=str(e)) -@mcp.tool(title="Find Assets By Sensors") +@mcp.tool(title="Find Assets By Sensors", annotations=_READ_ONLY_ANNOTATIONS) def find_assets_by_sensors( site_name: str, sensors: List[str], @@ -534,7 +542,7 @@ def _hits(sensor_name: str) -> List[str]: ) -@mcp.tool(title="Stream Extent") +@mcp.tool(title="Stream Extent", annotations=_READ_ONLY_ANNOTATIONS) def stream_extent( site_name: str, asset_id: str, @@ -641,7 +649,7 @@ def stream_extent( return ErrorResult(error="unable to inspect telemetry stream extent") -@mcp.tool(title="Get Sensor History") +@mcp.tool(title="Get Sensor History", annotations=_READ_ONLY_ANNOTATIONS) def history( site_name: str, asset_id: str, @@ -786,7 +794,7 @@ def history( ) -@mcp.tool(title="Latest Reading") +@mcp.tool(title="Latest Reading", annotations=_READ_ONLY_ANNOTATIONS) def latest_reading( site_name: str, asset_id: str, @@ -882,7 +890,7 @@ def latest_reading( ) -@mcp.tool(title="Sensor Coverage") +@mcp.tool(title="Sensor Coverage", annotations=_READ_ONLY_ANNOTATIONS) def sensor_coverage( site_name: str, asset_id: str, @@ -949,7 +957,7 @@ def sensor_coverage( ) -@mcp.tool(title="Sensor Statistics") +@mcp.tool(title="Sensor Statistics", annotations=_READ_ONLY_ANNOTATIONS) def sensor_stats( site_name: str, asset_id: str, diff --git a/src/servers/iot/tests/test_tools.py b/src/servers/iot/tests/test_tools.py index 8896275e7..ef3f2e14f 100644 --- a/src/servers/iot/tests/test_tools.py +++ b/src/servers/iot/tests/test_tools.py @@ -25,6 +25,13 @@ async def test_registry_tools_are_registered(self): "stream_extent", ] + @pytest.mark.anyio + async def test_all_iot_tools_are_advertised_read_only(self): + tools = await mcp.list_tools() + + assert all(tool.annotations.readOnlyHint is True for tool in tools) + assert all(tool.annotations.destructiveHint is False for tool in tools) + @pytest.mark.anyio async def test_stream_extent_description_is_storage_neutral(self): tools = await mcp.list_tools() diff --git a/src/servers/wo/main.py b/src/servers/wo/main.py index 55029f858..c9543f9fb 100644 --- a/src/servers/wo/main.py +++ b/src/servers/wo/main.py @@ -405,13 +405,21 @@ async def cancel_workorder( _TOOLS = ( _READ_TOOLS if os.environ.get("AOB_READONLY") == "1" else _READ_TOOLS + _WRITE_TOOLS ) +_READ_ANNOTATIONS = ToolAnnotations( + readOnlyHint=True, + destructiveHint=False, + idempotentHint=True, + openWorldHint=True, +) +_WRITE_ANNOTATIONS = ToolAnnotations( + readOnlyHint=False, + destructiveHint=True, + idempotentHint=False, + openWorldHint=True, +) _TOOL_ANNOTATIONS = { - get_failure_codes: ToolAnnotations( - readOnlyHint=True, - destructiveHint=False, - idempotentHint=True, - openWorldHint=True, - ) + **{fn: _READ_ANNOTATIONS for fn, _title in _READ_TOOLS}, + **{fn: _WRITE_ANNOTATIONS for fn, _title in _WRITE_TOOLS}, } for _fn, _title in _TOOLS: mcp.tool(title=_title, annotations=_TOOL_ANNOTATIONS.get(_fn))(_fn) diff --git a/src/servers/wo/tests/test_workorders.py b/src/servers/wo/tests/test_workorders.py index 6153785d8..81d699a2f 100644 --- a/src/servers/wo/tests/test_workorders.py +++ b/src/servers/wo/tests/test_workorders.py @@ -297,6 +297,21 @@ async def test_failure_code_mcp_tool_is_registered_read_only(monkeypatch) -> Non assert data["failure_codes"][0]["code"] == "FC002" +@pytest.mark.anyio +async def test_workorder_tool_safety_annotations_cover_reads_and_writes() -> None: + tools = {tool.name: tool for tool in await main.mcp.list_tools()} + + for fn, _title in main._READ_TOOLS: + annotations = tools[fn.__name__].annotations + assert annotations.readOnlyHint is True + assert annotations.destructiveHint is False + + for fn, _title in main._WRITE_TOOLS: + annotations = tools[fn.__name__].annotations + assert annotations.readOnlyHint is False + assert annotations.destructiveHint is True + + @pytest.mark.anyio async def test_failure_code_mcp_boundary_returns_typed_database_error( monkeypatch, From ebe55cfa6535980a0c7e77e9f5edd6c7ce814ff7 Mon Sep 17 00:00:00 2001 From: Hem Vadgama Date: Tue, 1 Sep 2026 17:57:21 -0400 Subject: [PATCH 2/3] test(evaluation): harden escalation experiment and scalar scoring Signed-off-by: Hem Vadgama --- benchmarks/adaptive_escalation_experiment.py | 493 ++++++++++++++++++ .../test_adaptive_escalation_experiment.py | 145 ++++++ src/evaluation/scorers/static_json.py | 36 +- .../tests/test_static_json_scorer.py | 16 + 4 files changed, 660 insertions(+), 30 deletions(-) create mode 100644 benchmarks/adaptive_escalation_experiment.py create mode 100644 src/agent/tests/test_adaptive_escalation_experiment.py diff --git a/benchmarks/adaptive_escalation_experiment.py b/benchmarks/adaptive_escalation_experiment.py new file mode 100644 index 000000000..6f7c54123 --- /dev/null +++ b/benchmarks/adaptive_escalation_experiment.py @@ -0,0 +1,493 @@ +#!/usr/bin/env python +"""Run a controlled Plan-Execute adaptive-escalation experiment. + +The four conditions share the same runner, model, tools, and verifier: + +* ``baseline`` disables verification; +* ``original`` replays the PR #432 critique-only routing policy; +* ``redesigned`` uses direct execution evidence and bounded safe recovery; +* ``always`` enables recovery and forces the expensive verification behavior. + +Example: + + PYTHONPATH=src .venv/bin/python benchmarks/adaptive_escalation_experiment.py \ + --scenario-root src/couchdb/scenarios_data \ + --scenario-ids 1,2,3 \ + --acknowledge-external-llm \ + --output-dir /tmp/adaptive-escalation + +The acknowledgement is deliberately required because scenario questions, plans, +tool arguments, and tool responses are sent to the configured LLM provider. +""" + +from __future__ import annotations + +import argparse +import asyncio +import csv +import json +import os +import statistics +import subprocess +import sys +import time +from contextlib import nullcontext +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any +from unittest.mock import patch + +from dotenv import load_dotenv + +import agent.plan_execute.runner as runner_module +from agent.plan_execute.escalation import ( + EscalationAction, + EscalationDecision, + extract_escalation_signals, + should_escalate, +) +from agent.plan_execute.runner import PlanExecuteRunner +from evaluation.evaluator import Evaluator +from evaluation.metrics import _estimate_cost +from llm import LLMBackend, LLMResult, LiteLLMBackend +from observability import init_tracing, set_run_context + +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_MODEL = "watsonx/meta-llama/llama-4-maverick-17b-128e-instruct-fp8" +CONDITIONS = ("baseline", "original", "redesigned", "always") + + +@dataclass(frozen=True) +class CallMetric: + kind: str + input_tokens: int + output_tokens: int + duration_ms: float + + +class RecordingBackend(LLMBackend): + """Collect usage and latency without changing model behavior.""" + + def __init__(self, model_id: str) -> None: + self._inner = LiteLLMBackend(model_id) + self.calls: list[CallMetric] = [] + + @property + def model_id(self) -> str: + return self._inner.model_id + + def generate(self, prompt: str, temperature: float = 0.0) -> str: + return self.generate_with_usage(prompt, temperature).text + + def generate_with_usage( + self, prompt: str, temperature: float = 0.0 + ) -> LLMResult: + started = time.perf_counter() + result = self._inner.generate_with_usage(prompt, temperature) + self.calls.append( + CallMetric( + kind=classify_prompt(prompt), + input_tokens=result.input_tokens, + output_tokens=result.output_tokens, + duration_ms=(time.perf_counter() - started) * 1000, + ) + ) + return result + + +def classify_prompt(prompt: str) -> str: + if prompt.startswith("You are reviewing the evidence"): + return "verification" + if "You are summarizing" in prompt: + return "summarization" + if "Generate the JSON arguments" in prompt: + return "argument_resolution" + return "planning" + + +def parse_scenario_ids(raw: str) -> list[str]: + values = [value.strip() for value in raw.split(",") if value.strip()] + if not values: + raise ValueError("--scenario-ids must contain at least one id") + if len(set(values)) != len(values): + raise ValueError("--scenario-ids contains duplicates") + return values + + +def parse_conditions(raw: str) -> list[str]: + """Return a validated, ordered subset of the experimental conditions.""" + values = [value.strip() for value in raw.split(",") if value.strip()] + if not values: + raise ValueError("--conditions must contain at least one condition") + if len(set(values)) != len(values): + raise ValueError("--conditions contains duplicates") + unknown = [value for value in values if value not in CONDITIONS] + if unknown: + raise ValueError( + "--conditions contains unknown values: " + ", ".join(unknown) + ) + return values + + +def estimate_cost(model_id: str, input_tokens: int, output_tokens: int) -> float | None: + estimate = _estimate_cost(model_id, input_tokens, output_tokens) + if estimate is not None: + return estimate + if "llama-4-maverick" in model_id.lower(): + return round((input_tokens * 0.27 + output_tokens * 0.85) / 1_000_000, 6) + return None + + +def prepare_scenario(scenario_root: Path, scenario_id: str) -> None: + env = os.environ.copy() + env["SCENARIOS_DATA_DIR"] = str(scenario_root.resolve()) + subprocess.run( + [sys.executable, "src/couchdb/init_data.py", "--reset-only"], + check=True, + cwd=REPO_ROOT, + env=env, + stdout=subprocess.DEVNULL, + ) + subprocess.run( + [sys.executable, "src/couchdb/init_data.py", scenario_id], + check=True, + cwd=REPO_ROOT, + env=env, + stdout=subprocess.DEVNULL, + ) + + +def _routing_context(condition: str): + if condition == "original": + return patch.object(runner_module, "should_escalate", side_effect=_original_policy) + if condition == "always": + return patch.object( + runner_module, + "should_escalate", + return_value=EscalationDecision( + action=EscalationAction.VERIFY, + reasons=["always verify experimental condition"], + ), + ) + return nullcontext() + + +def _original_policy(signals) -> EscalationDecision: + """Reproduce PR #432 routing without enabling redesigned recovery.""" + 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( + action=EscalationAction.VERIFY if reasons else EscalationAction.NONE, + reasons=reasons, + ) + + +def _condition_decision(condition: str, signals) -> EscalationDecision: + """Record the decision made by the selected experimental condition.""" + if condition == "baseline": + return EscalationDecision(action=EscalationAction.NONE, reasons=[]) + if condition == "original": + return _original_policy(signals) + if condition == "always": + return EscalationDecision( + action=EscalationAction.VERIFY, + reasons=["always verify experimental condition"], + ) + return should_escalate(signals) + + +async def run_one( + *, + condition: str, + scenario_id: str, + question: str, + model_id: str, + trajectory_dir: Path, +) -> dict[str, Any]: + backend = RecordingBackend(model_id) + run_id = f"plan-execute-{condition}-{scenario_id}" + os.environ["AGENT_TRAJECTORY_DIR"] = str(trajectory_dir) + set_run_context(run_id=run_id, scenario_id=scenario_id) + + started = time.perf_counter() + with _routing_context(condition): + result = await PlanExecuteRunner( + backend, + adaptive_escalation=condition != "baseline", + adaptive_recovery=condition in {"redesigned", "always"}, + ).run(question) + duration_ms = (time.perf_counter() - started) * 1000 + + signals = extract_escalation_signals(question, result.plan, result.trajectory) + policy_decision = _condition_decision(condition, signals) + calls = [asdict(call) for call in backend.calls] + input_tokens = sum(call.input_tokens for call in backend.calls) + output_tokens = sum(call.output_tokens for call in backend.calls) + + return { + "scenario_id": scenario_id, + "condition": condition, + "run_id": run_id, + "model": model_id, + "question": question, + "answer": result.answer, + "plan_steps": len(result.plan.steps), + "failed_steps": sum(not step.success for step in result.trajectory), + "tool_steps": sum( + bool(step.tool and step.tool.lower() not in {"none", "null"}) + for step in result.trajectory + ), + "policy_action": policy_decision.action.value, + "policy_reasons": policy_decision.reasons, + "runner_action": result.escalation_action, + "routed_to_verification": any(call.kind == "verification" for call in backend.calls), + "routed_to_recovery": any( + step.recovery_attempted for step in result.trajectory + ), + "signals": asdict(signals), + "llm_calls": len(backend.calls), + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": input_tokens + output_tokens, + "estimated_cost_usd": estimate_cost(model_id, input_tokens, output_tokens), + "duration_ms": duration_ms, + "llm_duration_ms": sum(call.duration_ms for call in backend.calls), + "planning_duration_ms": sum( + call.duration_ms for call in backend.calls if call.kind == "planning" + ), + "summarization_duration_ms": sum( + call.duration_ms for call in backend.calls if call.kind == "summarization" + ), + "verification_duration_ms": sum( + call.duration_ms for call in backend.calls if call.kind == "verification" + ), + "execution_duration_ms": sum( + step.duration_ms or 0.0 for step in result.trajectory + ), + "calls": calls, + } + + +def evaluate_runs( + records: list[dict[str, Any]], + *, + scenario_root: Path, + scenario_ids: list[str], + trajectory_root: Path, + conditions: list[str] | tuple[str, ...] = CONDITIONS, +) -> None: + evaluator = Evaluator(default_scorer="static_json") + by_key = {(record["condition"], record["scenario_id"]): record for record in records} + for condition in conditions: + report = evaluator.evaluate( + trajectories_path=trajectory_root / condition, + scenarios_paths=[scenario_root], + scenario_ids=scenario_ids, + ) + for result in report.results: + record = by_key[(condition, result.scenario_id)] + record["passed"] = result.score.passed + record["score"] = result.score.score + record["score_rationale"] = result.score.rationale + record["score_details"] = result.score.details + + +def _condition_summary( + records: list[dict[str, Any]], + conditions: list[str] | tuple[str, ...] = CONDITIONS, +) -> dict[str, dict[str, Any]]: + summary: dict[str, dict[str, Any]] = {} + for condition in conditions: + selected = [record for record in records if record["condition"] == condition] + summary[condition] = { + "scenarios": len(selected), + "passed": sum(record["passed"] for record in selected), + "pass_rate": ( + sum(record["passed"] for record in selected) / len(selected) + if selected + else 0.0 + ), + "mean_score": ( + statistics.fmean(record["score"] for record in selected) + if selected + else 0.0 + ), + "input_tokens": sum(record["input_tokens"] for record in selected), + "output_tokens": sum(record["output_tokens"] for record in selected), + "total_tokens": sum(record["total_tokens"] for record in selected), + "llm_calls": sum(record["llm_calls"] for record in selected), + "estimated_cost_usd": ( + sum(record["estimated_cost_usd"] for record in selected) + if selected + and all(record["estimated_cost_usd"] is not None for record in selected) + else None + ), + "median_duration_ms": ( + statistics.median(record["duration_ms"] for record in selected) + if selected + else None + ), + "verification_rate": ( + sum(record["routed_to_verification"] for record in selected) + / len(selected) + if selected + else 0.0 + ), + "recovery_rate": ( + sum(record["routed_to_recovery"] for record in selected) / len(selected) + if selected + else 0.0 + ), + } + return summary + + +def write_outputs( + records: list[dict[str, Any]], + output_dir: Path, + conditions: list[str] | tuple[str, ...] = CONDITIONS, +) -> dict[str, Any]: + summary = { + "conditions": _condition_summary(records, conditions), + "comparison_status": ( + "complete" + if set(CONDITIONS).issubset(conditions) + else "partial_mechanism_check" + ), + "interpretation_warning": ( + "Conditions use independently planned single runs; score differences " + "are diagnostic associations, not causal effects of escalation." + ), + } + (output_dir / "runs.json").write_text( + json.dumps(records, indent=2), encoding="utf-8" + ) + (output_dir / "summary.json").write_text( + json.dumps(summary, indent=2), encoding="utf-8" + ) + + fields = [ + "scenario_id", + "condition", + "passed", + "score", + "plan_steps", + "failed_steps", + "tool_steps", + "policy_action", + "runner_action", + "routed_to_verification", + "routed_to_recovery", + "llm_calls", + "input_tokens", + "output_tokens", + "total_tokens", + "estimated_cost_usd", + "duration_ms", + "llm_duration_ms", + "planning_duration_ms", + "summarization_duration_ms", + "verification_duration_ms", + "execution_duration_ms", + ] + with (output_dir / "runs.csv").open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter( + handle, fieldnames=fields, extrasaction="ignore", lineterminator="\n" + ) + writer.writeheader() + writer.writerows(records) + return summary + + +async def run_experiment(args: argparse.Namespace) -> int: + if not args.acknowledge_external_llm: + raise ValueError( + "--acknowledge-external-llm is required because benchmark content " + "and tool evidence are sent to the configured model provider" + ) + scenario_ids = parse_scenario_ids(args.scenario_ids) + conditions = parse_conditions(args.conditions) + scenario_root = args.scenario_root.resolve() + output_dir = args.output_dir.resolve() + if output_dir.exists() and any(output_dir.iterdir()): + raise ValueError(f"output directory is not empty: {output_dir}") + output_dir.mkdir(parents=True, exist_ok=True) + trajectory_root = output_dir / "trajectories" + for condition in conditions: + (trajectory_root / condition).mkdir(parents=True) + + os.environ["OTEL_TRACES_FILE"] = str(output_dir / "traces.jsonl") + init_tracing("adaptive-escalation-experiment") + + records: list[dict[str, Any]] = [] + for scenario_id in scenario_ids: + question_path = scenario_root / f"scenario_{scenario_id}" / "question.txt" + if not question_path.exists(): + raise FileNotFoundError(f"missing scenario question: {question_path}") + question = question_path.read_text(encoding="utf-8").strip() + for condition in conditions: + print(f"running scenario={scenario_id} condition={condition}", flush=True) + prepare_scenario(scenario_root, scenario_id) + records.append( + await run_one( + condition=condition, + scenario_id=scenario_id, + question=question, + model_id=args.model_id, + trajectory_dir=trajectory_root / condition, + ) + ) + (output_dir / "runs.partial.json").write_text( + json.dumps(records, indent=2), encoding="utf-8" + ) + + evaluate_runs( + records, + scenario_root=scenario_root, + scenario_ids=scenario_ids, + trajectory_root=trajectory_root, + conditions=conditions, + ) + summary = write_outputs(records, output_dir, conditions) + print(json.dumps(summary, indent=2)) + print(f"results={output_dir}") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--scenario-root", type=Path, required=True) + parser.add_argument("--scenario-ids", default="1") + parser.add_argument( + "--conditions", + default=",".join(CONDITIONS), + help="comma-separated subset for gated runs (default: all four)", + ) + parser.add_argument("--model-id", default=DEFAULT_MODEL) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument( + "--acknowledge-external-llm", + action="store_true", + help="confirm that benchmark prompts and tool evidence may leave this host", + ) + return parser + + +def main() -> int: + load_dotenv(REPO_ROOT / ".env") + try: + return asyncio.run(run_experiment(build_parser().parse_args())) + except (FileNotFoundError, ValueError, subprocess.CalledProcessError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/agent/tests/test_adaptive_escalation_experiment.py b/src/agent/tests/test_adaptive_escalation_experiment.py new file mode 100644 index 000000000..aea817756 --- /dev/null +++ b/src/agent/tests/test_adaptive_escalation_experiment.py @@ -0,0 +1,145 @@ +"""Unit tests for the adaptive-escalation experiment harness.""" + +from __future__ import annotations + +import argparse +import importlib.util +import sys +from pathlib import Path + +import pytest + +from agent.plan_execute.escalation import extract_escalation_signals, should_escalate +from agent.plan_execute.models import Plan, PlanStep + + +def _load_experiment_module(): + path = ( + Path(__file__).resolve().parents[3] + / "benchmarks" + / "adaptive_escalation_experiment.py" + ) + spec = importlib.util.spec_from_file_location("adaptive_escalation_experiment", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +experiment = _load_experiment_module() + + +def test_live_harness_defines_all_four_comparison_conditions(): + assert experiment.CONDITIONS == ("baseline", "original", "redesigned", "always") + + +def test_original_and_redesigned_routing_are_technically_distinct(): + plan = Plan( + steps=[ + PlanStep(1, "Routine read", "wo", "list_workorders", {}, [], "records") + ], + raw="", + ) + signals = extract_escalation_signals("List work orders", plan, []) + + assert experiment._original_policy(signals).action.value == "verify" + assert should_escalate(signals).action.value == "none" + + +@pytest.mark.parametrize( + ("prompt", "expected"), + [ + ("You are reviewing the evidence gathered by an agent", "verification"), + ("You are summarizing the results", "summarization"), + ("Generate the JSON arguments for this tool", "argument_resolution"), + ("Create a plan", "planning"), + ], +) +def test_classify_prompt(prompt, expected): + assert experiment.classify_prompt(prompt) == expected + + +def test_parse_scenario_ids_rejects_duplicates(): + with pytest.raises(ValueError, match="duplicates"): + experiment.parse_scenario_ids("1,2,1") + + +def test_parse_conditions_supports_a_gated_single_condition_run(): + assert experiment.parse_conditions("redesigned") == ["redesigned"] + + +@pytest.mark.parametrize("raw", ["", "redesigned,redesigned", "experimental"]) +def test_parse_conditions_rejects_invalid_selections(raw): + with pytest.raises(ValueError, match="--conditions"): + experiment.parse_conditions(raw) + + +def test_recorded_decision_matches_each_experimental_condition(): + plan = Plan( + steps=[ + PlanStep(1, "Routine read", "wo", "list_workorders", {}, [], "records") + ], + raw="", + ) + signals = extract_escalation_signals("List work orders", plan, []) + + assert experiment._condition_decision("baseline", signals).action.value == "none" + assert experiment._condition_decision("original", signals).action.value == "verify" + assert experiment._condition_decision("redesigned", signals).action.value == "none" + assert experiment._condition_decision("always", signals).action.value == "verify" + + +def test_external_llm_acknowledgement_is_explicit(): + parser = experiment.build_parser() + base = ["--scenario-root", "scenarios", "--output-dir", "results"] + + assert not parser.parse_args(base).acknowledge_external_llm + acknowledged = parser.parse_args(base + ["--acknowledge-external-llm"]) + assert acknowledged.acknowledge_external_llm + + +@pytest.mark.anyio +async def test_live_run_refuses_external_calls_without_acknowledgement(): + with pytest.raises(ValueError, match="--acknowledge-external-llm"): + await experiment.run_experiment( + argparse.Namespace(acknowledge_external_llm=False) + ) + + +def test_output_labels_match_the_metrics_the_harness_measures(tmp_path): + record = { + "scenario_id": "1", + "condition": "redesigned", + "passed": True, + "score": 1.0, + "input_tokens": 10, + "output_tokens": 2, + "total_tokens": 12, + "llm_calls": 1, + "estimated_cost_usd": 0.01, + "duration_ms": 5.0, + "routed_to_verification": False, + "routed_to_recovery": True, + "tool_steps": 2, + "execution_duration_ms": 3.0, + } + + summary = experiment.write_outputs([record], tmp_path, ["redesigned"]) + csv_text = (tmp_path / "runs.csv").read_text(encoding="utf-8") + + assert summary["comparison_status"] == "partial_mechanism_check" + assert "not causal" in summary["interpretation_warning"] + assert "tool_steps" in csv_text + assert "execution_duration_ms" in csv_text + assert "tool_calls" not in csv_text + + +def test_estimate_cost_handles_full_watsonx_llama_model_id(): + estimate = experiment.estimate_cost( + "watsonx/meta-llama/llama-4-maverick-17b-128e-instruct-fp8", + 1_000_000, + 1_000_000, + ) + + assert estimate == 1.12 diff --git a/src/evaluation/scorers/static_json.py b/src/evaluation/scorers/static_json.py index 7afa9d564..20771331d 100644 --- a/src/evaluation/scorers/static_json.py +++ b/src/evaluation/scorers/static_json.py @@ -209,37 +209,17 @@ def _parse_json_or_python(content: str) -> Any: return _PARSE_MISSING -def _extract_count_from_text(content: str) -> int | float | None: - """Extract a count when the answer is count-only or nearly count-only.""" - stripped = content.strip() - - if re.fullmatch(r"-?\d+", stripped): - return int(stripped) - - if re.fullmatch(r"-?\d+\.\d+", stripped): - return float(stripped) - - numbers = re.findall( - r"(? int | float | None: - """Extract a final standalone count from a noisy scalar answer.""" + """Extract only an explicit final count from a scalar answer. + + A unique number elsewhere in prose is not necessarily the answer. For example, + ``cannot complete because Step 1 failed`` must not score as the count ``1``. + """ stripped = content.strip() - count = _extract_count_from_text(stripped) - if count is not None: - return count final_number = re.compile( r"^\s*(?:" - r"(?:final\s+answer|answer|count|result)\s*(?:is|:)?\s*" + r"(?:the\s+)?(?:final\s+answer|answer|count|result)\s*(?:is|:)?\s*" r")?(-?\d+(?:\.\d+)?)\s*\.?\s*$", flags=re.IGNORECASE, ) @@ -339,10 +319,6 @@ def parse_structured_answer(value: Any) -> Any: if count is not None: return count - count = _extract_count_from_text(content) - if count is not None: - return count - return content.strip() diff --git a/src/evaluation/tests/test_static_json_scorer.py b/src/evaluation/tests/test_static_json_scorer.py index d28fa6de6..dff2426c2 100644 --- a/src/evaluation/tests/test_static_json_scorer.py +++ b/src/evaluation/tests/test_static_json_scorer.py @@ -80,6 +80,22 @@ def test_parse_noisy_count_answer_prefers_final_standalone_number(): assert parse_structured_answer(raw) == 0 +def test_count_answer_does_not_treat_failed_step_number_as_the_answer(): + model_answer = "The final count cannot be provided due to the failure in Step 1." + + assert parse_structured_answer(model_answer) == model_answer + score = evaluate_static_json("1", model_answer) + assert score.strict_exact_match_accuracy == 0.0 + + +def test_count_answer_does_not_extract_an_unlabelled_number_from_prose(): + model_answer = "The tool failed after examining 34 records." + + assert parse_structured_answer(model_answer) == model_answer + score = evaluate_static_json("34", model_answer) + assert score.strict_exact_match_accuracy == 0.0 + + def test_count_answer_compares_final_number_not_parenthetical_text(): score = evaluate_static_json( "0", From 4ee5fc44c246a6dd2049b84e4985fd2ffe4832ab Mon Sep 17 00:00:00 2001 From: Hem Vadgama Date: Tue, 1 Sep 2026 17:57:38 -0400 Subject: [PATCH 3/3] docs: report adaptive escalation live evidence Signed-off-by: Hem Vadgama --- docs/adaptive_escalation_live_analysis.md | 259 ++++++++++++++++++++++ 1 file changed, 259 insertions(+) create mode 100644 docs/adaptive_escalation_live_analysis.md diff --git a/docs/adaptive_escalation_live_analysis.md b/docs/adaptive_escalation_live_analysis.md new file mode 100644 index 000000000..690048ece --- /dev/null +++ b/docs/adaptive_escalation_live_analysis.md @@ -0,0 +1,259 @@ +# Adaptive escalation: evidence-driven live validation + +Status: **exploratory engineering validation, not a final benchmark claim** + +Date: 2026-09-01 + +## Executive conclusion + +The current redesign is a useful project improvement, but for narrower reasons than +the original proposal implied. + +It does not yet have a statistically established task-accuracy advantage. In a final +single redesigned run it answered the two locally answerable count scenarios exactly +and failed the unsupported end-to-end scenario, for 2/3 strict passes. A current +baseline/redesigned pair also passed scenario 2, so that accuracy result cannot be +attributed solely to adaptive escalation. + +The demonstrated adaptive improvement is evidence integrity and bounded work: + +- successful specialist-server runs are no longer verified merely because they are + deep or contain domain vocabulary; +- an unadvertised argument was rejected and repaired before any tool call; +- post-call retries occurred only for explicitly read-only tools; +- a failed prerequisite stopped four dependent steps; +- the runner returned an explicit failure instead of fabricated anomaly output; and +- on the matched current scenario-3 check, redesigned execution used 16,536 tokens + and four LLM calls versus baseline's 22,863 tokens and seven calls, a 27.7% token + reduction in that diagnostic pair; the redesigned trajectory stopped unsupported + downstream work. + +This is useful behavior for AssetOpsBench even when the task cannot be completed. It +is safer and more auditable than critique-only verification. It should not be sold as +an accuracy win until a larger preregistered repeated study supports that claim. + +## Questions evaluated + +1. Does the policy avoid escalation on successful work that merely looks specialized? +2. Can recovery correct a concrete execution defect without replaying unsafe tools? +3. Does the runner stop when required evidence becomes unavailable? +4. Does it avoid claiming successful completion from failed tool evidence? +5. Do live evaluator results survive inspection against raw ground truth? +6. Is any observed outcome large and controlled enough to call an improvement? + +## Conditions + +Within each comparison, live runs used the same WatsonX model at temperature 0, a +CouchDB reset before every condition, the same local scenario fixtures, and the +`static_json` evaluator. The diagnostic matrix and final follow-up use different code +checkpoints and are reported separately; their scores are not combined. + +| Condition | Execution behavior | Routing behavior | +| --- | --- | --- | +| Baseline | Legacy execution | No verification | +| Original | Legacy execution | PR #432 broad critique-only policy | +| Redesigned | Strict error detection and bounded safe recovery | Direct execution evidence | +| Always | Redesigned execution | Verification forced on | + +Each recorded condition was run once. Provider calls were not seeded or replayed, so +plans varied. Comparisons are diagnostic associations, not causal effect estimates. + +The experiment harness now requires `--acknowledge-external-llm`, supports a selected +condition subset, records the decision made by the actual condition, and labels +incomplete matrices as mechanism checks. Repository scenario text and tool evidence +were sent to WatsonX only after explicit approval; `.env` values were not included. + +## Design boundary + +The redesign acts on execution evidence rather than guessing risk from vocabulary or +plan shape: + +```text +step succeeds -> continue +arguments fail before a tool call -> regenerate once within the run budget +read-only tool call fails -> regenerate and retry once +mutating or unknown-safety call fails -> do not replay; verify the failure +required evidence remains unavailable -> block dependants and report failure +all steps succeed -> summarize normally +``` + +Adaptive behavior remains opt-in. Recovery is bounded to one retry per step and a +small run-wide budget. A post-call retry requires MCP metadata that explicitly marks +the tool read-only and non-destructive; idempotence alone is not enough. A failed +dependency is never called with placeholder evidence. `retry_step` records that a +bounded recovery already succeeded—it does not initiate another tool call after the +plan finishes. + +## Evaluation integrity correction + +The first live matrix exposed a false-positive scorer result. Scenario 2 has scalar +ground truth `1`, but the answer "The final count cannot be provided due to the failure +in Step 1" was scored as an exact match because the scalar parser extracted the only +number in the prose. + +The scorer now accepts a scalar count only when it is: + +- count-only; +- explicitly labelled as the answer, count, or result; or +- the final standalone numeric line. + +Two regression tests cover the failed-step phrase and unrelated numeric prose. After +offline rescoring, the original-policy scenario-2 result correctly changed from pass +to failure. The pre-correction pass is not used anywhere in this report. + +## Diagnostic four-condition matrix + +This matrix preceded the follow-up planner/argument fixes and was used to find design +failures rather than support a positive claim. + +| Condition | Strict passes | Mean score | Tokens | LLM calls | Verify rate | Recovery rate | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | +| Baseline | 1/3 | 0.381 | 47,001 | 14 | 0% | 0% | +| Original | 0/3 | 0.048 | 50,393 | 16 | 100% | 0% | +| Redesigned, first version | 1/3 | 0.333 | 39,802 | 12 | 0% | 67% | +| Always | 1/3 | 0.381 | 42,708 | 13 | 100% | 33% | + +This result did **not** show a reliability improvement. It did show: + +- the original policy still verified every case; +- expensive verification did not repair wrong evidence; +- one recovery returned operational success with semantically wrong filters and an + empty result; and +- structured tool errors remained invisible in legacy execution. + +The work therefore continued locally instead of publishing the matrix as a success. + +## Evidence-driven follow-up changes + +The diagnostic failures led to small general changes rather than scenario-specific +rules: + +1. The planner prompt instructs the model to resolve indirectly described identifiers + using an available discovery/list tool before the consuming step. +2. The planner prompt forbids planning any capability absent from the advertised tool + set. This remains prompt-level guidance, so runtime capability and dependency checks + remain necessary. +3. Argument resolution receives the tool description separately from its parameter + names. +4. Optional filters are omitted when the question asks for all values and are included + only when an exact value is supported by the question, tool documentation, or prior + evidence. +5. Unadvertised parameter names fail validation before a tool call and are eligible + for one safe repair. +6. Verification and summarization evidence includes the exact tool name and arguments. + +These constraints apply across tools and scenarios. They do not encode expected +answers, scenario IDs, work-order numbers, or ground-truth fields. + +## Final current redesigned run + +| Scenario | Ground truth | Answer/action | Strict | Recovery | Tokens | Calls | +| --- | --- | --- | ---: | ---: | ---: | ---: | +| 1 | `2` | `2`; `none` | 1.0 | No | 13,686 | 4 | +| 2 | `1` | `1`; `retry_step` | 1.0 | Yes | 14,611 | 5 | +| 3 | Structured anomaly result | Explicit `report_failure` | 0.0 | Attempted, exhausted | 16,536 | 4 | + +Aggregate: 2/3 strict passes, mean score 0.667, 44,833 tokens, 13 LLM +calls, estimated cost $0.01299. + +Independent rescoring parsed the final answers as integer `2`, integer `1`, and an +explicit failure string respectively, matching the harness's 1.0, 1.0, and 0.0 strict +scores. + +### Scenario 1: selectivity + +The run used the specialist `wo` server, dependency depth 3, and matched `work order`, +`failure`, and `anomaly`. Those signals would have triggered the original policy. All +steps succeeded, so the redesigned policy selected `none` and returned the exact count. +This is direct selectivity evidence. + +### Scenario 2: bounded pre-call recovery + +The argument model invented `filter` for the no-argument `sites` tool. Strict schema +validation rejected it before execution. One `safe_pre_call` repair produced `{}`; +site discovery returned `MAIN`, the work-order query used documented arguments, and +the final answer matched ground truth. + +This demonstrates the intended recovery mechanism within one trajectory. It does not +prove an aggregate accuracy advantage: a separate current baseline run also answered +scenario 2 correctly after the shared planning improvements. + +### Scenario 3: evidence integrity + +The current baseline and redesigned conditions both failed the task. Their behavior +was materially different: + +| Measure | Current baseline | Current redesigned | +| --- | ---: | ---: | +| Strict score | 0.0 | 0.0 | +| Tool failures recorded | 0 | 1 root failure plus 4 blocked dependants | +| Downstream tool calls after missing prerequisite | 3 | 0 | +| LLM calls | 7 | 4 | +| Tokens | 22,863 | 16,536 | + +Baseline received structured errors from asset lookup, history retrieval, file read, +and anomaly execution but treated them as successful evidence. Its answer explicitly +said it was simulating hypothetical success and emitted invented structured values, +including zero observations and `anomalies_found: false`. + +Redesigned execution detected the first structured lookup error, made one read-only +retry, exhausted that retry, blocked every dependant, and returned a deterministic +failure containing the exact missing evidence. This is an improvement in grounding, +safety, and bounded cost, not task accuracy. + +## What is and is not established + +Established by deterministic tests and inspected live trajectories: + +- broad static vocabulary no longer routes successful work; +- strict schema validation can prevent a malformed call; +- safe recovery is bounded; +- mutating and unknown-safety failures are not replayed; +- structured errors and failed dependencies are visible; +- dependent execution stops after missing evidence; and +- failure answers do not claim completion. + +Not established: + +- a statistically reliable accuracy improvement; +- stability across repeated plans, other models, or the closed scenario corpus; +- that recovery's operational success implies semantic correctness; or +- that scenario 3 can be solved with the currently exposed tool graph. + +## Recommendation + +The adaptive work is useful enough to retain as one coherent PR, organized into +reviewable commits: + +1. Production behavior and focused tests: direct-evidence routing, typed actions, + strict structured-error handling, dependency blocking, bounded recovery, tool + safety annotations, and deterministic failure reporting. +2. Evaluation rigor and tests: the scalar false-positive correction and controlled + four-condition harness. +3. This evidence report. + +Do not publish a claim that adaptive escalation improves benchmark accuracy from these +runs. A defensible claim is that it reduces indiscriminate verification and prevents +failed evidence from being treated as completed work. + +## Reproduction and accounting + +The live harness can be rerun, after starting CouchDB and explicitly authorizing the +external model calls, with: + +```bash +PYTHONPATH=src .venv/bin/python benchmarks/adaptive_escalation_experiment.py \ + --scenario-root src/couchdb/scenarios_data \ + --scenario-ids 1,2,3 \ + --acknowledge-external-llm \ + --output-dir /tmp/adaptive-escalation-results +``` + +The output directory must be empty. The harness resets and loads each scenario before +each condition, checkpoints partial results, writes JSON and CSV summaries, and records +the exact model and per-call token usage. Raw trajectories are intentionally excluded +from this PR; the inspected facts needed to audit the conclusions are stated above. + +Measured WatsonX use, including the initial 85-token quota smoke, was 288,928 tokens. +Estimated total cost was approximately $0.0843. No further provider calls were made +after the final baseline scenario-3 check.