M4.3 — Investigator agent loop

[zestones]

Note

Milestone: M4 — Sentinel + Investigator
Planning doc: docs/planning/M4-sentinel-investigator/issues.md

Scope. backend/agents/investigator.py:

• async def run_investigator(work_order_id: int) -> None
• Full agentic loop using MCPClient + extended thinking (M4.5) + agent-as-tool handoffs (M4.6)

Note

Depends on M3.1 (#17) for anthropic_client.py, M3.5 (#21) for
answer_kb_question, M4.1 (#23) for ws_manager, and M4.2 (#24)
for Sentinel calling this function.

---

1. Initial context load

async def run_investigator(work_order_id: int) -> None:
    # Load work order context
    wo = await mcp_client.call_tool("get_work_order", {"work_order_id": work_order_id})
    # Load failure history for memory flex (M4.7)
    past = await mcp_client.call_tool("get_failure_history", {
        "cell_id": wo_data["cell_id"], "limit": 5
    })

---

2. Tool registry

tools_schema = (
    await mcp_client.get_tools_schema()
    + INVESTIGATOR_RENDER_TOOLS          # from agents/ui_tools.py
    + [SUBMIT_RCA_TOOL, ASK_KB_BUILDER_TOOL]  # local tools (M4.6)
)

INVESTIGATOR_RENDER_TOOLS is already defined in backend/agents/ui_tools.py (M2.9).

---

3. Handle is_error from tool results

Important

get_signal_anomalies raises ValueError when KB is misconfigured — the MCPClient
wraps this as ToolCallResult(is_error=True). If the Investigator does not check
is_error, the exception propagates and crashes the investigation Task.

Return is_error=True tool results to the LLM so it can self-correct:

for tool_use in response_tool_uses:
    if tool_use.name.startswith("render_"):
        # Generative UI — broadcast and return "rendered"
        await ws_manager.broadcast("ui_render", {
            "agent": "investigator",
            "component": tool_use.name.removeprefix("render_"),
            "props": tool_use.input,
            "turn_id": turn_id,
        })
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": tool_use.id,
            "content": "rendered",
        })
        continue

    if tool_use.name == "ask_kb_builder":
        result = await _handle_ask_kb_builder(tool_use.input, turn_id)
    else:
        result = await mcp_client.call_tool(tool_use.name, tool_use.input)

    await ws_manager.broadcast("tool_call_completed", {
        "agent": "investigator",
        "tool_name": tool_use.name,
        "turn_id": turn_id,
    })

    tool_results.append({
        "type": "tool_result",
        "tool_use_id": tool_use.id,
        "content": result.content,
        "is_error": result.is_error,   # forward is_error so LLM can recover
    })

---

4. SUBMIT_RCA_TOOL — local tool definition

SUBMIT_RCA_TOOL = {
    "name": "submit_rca",
    "description": (
        "Submit a completed root cause analysis. Call this when you have enough "
        "evidence to conclude the investigation. This ends the agent loop."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "root_cause": {"type": "string"},
            "confidence": {"type": "number", "minimum": 0, "maximum": 1},
            "contributing_factors": {"type": "array", "items": {"type": "string"}},
            "similar_past_failure": {"type": "string", "description": "Reference a past failure if pattern matches."},
            "recommended_action": {"type": "string"},
        },
        "required": ["root_cause", "confidence", "contributing_factors", "recommended_action"],
    },
}

When the LLM calls submit_rca:

1. UPDATE work_order with rca_summary, confidence, status="investigated"
2. ws_manager.broadcast("rca_ready", {work_order_id, rca_summary, confidence, turn_id})
3. asyncio.create_task(run_work_order_generator(work_order_id))
4. Break the agent loop

---

5. ASK_KB_BUILDER_TOOL — local tool definition (M4.6)

ASK_KB_BUILDER_TOOL = {
    "name": "ask_kb_builder",
    "description": (
        "Consult the KB Builder for a manufacturer detail absent from the current KB "
        "(e.g. max bolt torque, part reference, installation spec)."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "question": {"type": "string"},
            "cell_id": {"type": "integer"},
        },
        "required": ["question", "cell_id"],
    },
}

Handler (see also M4.6 #28):

async def _handle_ask_kb_builder(args: dict, turn_id: str) -> ToolCallResult:
    from agents.kb_builder import answer_kb_question

    await ws_manager.broadcast("agent_handoff", {
        "from_agent": "investigator",
        "to_agent": "kb_builder",
        "reason": args["question"],
        "turn_id": turn_id,
    })
    new_turn_id = str(uuid.uuid4())
    await ws_manager.broadcast("agent_start", {"agent": "kb_builder", "turn_id": new_turn_id})

    answer = await answer_kb_question(args["cell_id"], args["question"])

    await ws_manager.broadcast("agent_end", {
        "agent": "kb_builder", "turn_id": new_turn_id, "finish_reason": "answered"
    })
    return ToolCallResult(content=json.dumps(answer), is_error=False)

---

6. System prompt

INVESTIGATOR_SYSTEM = """You are an industrial maintenance expert agent.
An anomaly has been detected on equipment. Investigate freely using the available tools.
You decide what to consult and in what order.

When you have enough evidence, call submit_rca with:
- root_cause: single-sentence conclusion
- confidence: 0.0-1.0
- contributing_factors: ordered list (most to least significant)
- similar_past_failure: reference a past failure if the pattern matches (or null)
- recommended_action: what the operator should do next

Past failures context: {past_failures}"""

---

7. Acceptance

• Investigator runs to completion and calls submit_rca — test_happy_path_submit_rca_persists_and_broadcasts (live P-02 run deferred to M4.4 — Lifespan integration #26 lifespan + simulator)
• work_order.rca_summary is populated after investigation — same test, via mocked WorkOrderRepository.update
• rca_ready WS event is broadcast — assertions on payload shape {work_order_id, rca_summary, confidence, turn_id}
• run_work_order_generator is spawned after RCA submission — test_work_order_generator_lazy_import_logs_when_missing (stub path until M5.1 — Work Order Generator agent #30 lands; lazy-import will route automatically then)
• is_error=True tool results are forwarded to the LLM (not raised as exceptions) — test_is_error_tool_result_forwarded_not_raised
• get_signal_anomalies is_error response logs and does not crash the Task — same test above + outer try/except fallback covered by test_crash_fallback_flips_status_and_broadcasts_rca_ready
• tool_call_started / tool_call_completed events broadcast for each tool call — test_tool_call_events_carry_expected_fields (both payloads checked; duration_ms int mandatory)
• (added) Timeout + outer try/except fallbacks — test_timeout_fallback_* and test_crash_fallback_*
• (added) failure_history row inserted on submit_rca (audit) — assertions on KbRepository.create_failure fields
• (added) ask_kb_builder dynamic handoff with agent_handoff + child agent_start/agent_end — test_ask_kb_builder_broadcasts_handoff_and_returns_answer
• (added) get_work_order(id) MCP tool shipped alongside — in aria_mcp/tools/context.py

---

Agent loop

sequenceDiagram
    autonumber
    participant Sent as Sentinel
    participant Inv as run_investigator
    participant Opus as Claude Opus (extended thinking)
    participant MCP as MCPClient
    participant KB as answer_kb_question
    participant WSMgr as WSManager

    Sent->>Inv: asyncio.create_task(run_investigator(wo_id))
    Inv->>MCP: get_work_order, get_failure_history
    Inv->>WSMgr: broadcast agent_start
    loop until submit_rca or max_turns
        Inv->>Opus: messages.create(stream=True, thinking=enabled)
        Opus-->>Inv: thinking_delta chunks
        Inv->>WSMgr: broadcast thinking_delta (each chunk)
        Opus-->>Inv: tool_use blocks
        loop per tool_use
            alt render_* tool
                Inv->>WSMgr: broadcast ui_render
            else ask_kb_builder
                Inv->>WSMgr: broadcast agent_handoff
                Inv->>KB: answer_kb_question(cell_id, question)
                KB-->>Inv: {answer, source, confidence}
            else MCP tool
                Inv->>MCP: call_tool(name, args)
                MCP-->>Inv: ToolCallResult (check is_error)
            end
            Inv->>WSMgr: broadcast tool_call_completed
        end
        alt submit_rca called
            Inv->>MCP: update work_order (rca_summary)
            Inv->>WSMgr: broadcast rca_ready
            Inv->>Inv: spawn run_work_order_generator
        end
    end
    Inv->>WSMgr: broadcast agent_end

Loading

[zestones]

Audit follow-up — implementation context

From docs/audits/M4-M5-issue-context-pass.md §2.3 and M4-M5-sentinel-investigator-workorder-qa-audit.md §3.1 / §3.3 / §3.7 / §4.1 / §4.3.

Important

Critical-path issue with the most drifts — read carefully before starting implementation.

Drifts from main

Drift	Issue body says	Real code on main
Status value	status="investigated"	Not in Status Literal (modules/work_order/schemas.py:12). Use "analyzed".
get_work_order MCP tool	mcp_client.call_tool("get_work_order", ...)	Does not exist. Only plural get_work_orders (aria_mcp/tools/context.py:94).
Model slug	model_for("agent")	Not in Literal (agents/anthropic_client.py:49). Use model_for("reasoning").
tool_call_started	Example only broadcasts tool_call_completed	EventBusMap declares both (ws.types.ts:18-29); started carries args, completed carries duration_ms.
duration_ms	Missing in example	Mandatory per FE type.
ask_kb_builder import	Implicit	from agents.kb_builder import answer_kb_question — re-export in __init__.py (issue #21 closed).
Max turns / timeout	Not mentioned	Demo-breaker without it.
Outer try/except	Not mentioned	Sentinel has it, Investigator needs it too.
render_correlation_matrix in mermaid	Referenced	Dropped from ui_tools.py:18-23 — use render_signal_chart with signal_def_ids: list[int].

Real-code anchors

• Model: from agents.anthropic_client import anthropic, model_for — use model_for("reasoning").
• MCP: from aria_mcp.client import mcp_client — returns ToolCallResult(content: str, is_error: bool) (aria_mcp/client.py:35).
• UI tools bundle: from agents.ui_tools import INVESTIGATOR_RENDER_TOOLS — three tools: SignalChart, DiagnosticCard, PatternMatch (ui_tools.py:374-378).
• KB Builder handoff: from agents.kb_builder import answer_kb_question (re-export; file is agents/kb_builder/qa.py).
• WorkOrder status: "analyzed" (not "investigated").

Replace get_work_order call

The get_work_order (singular) MCP tool does not exist. Pick one:

1. Add a new tool (recommended). Append to backend/aria_mcp/tools/context.py:

   @mcp.tool()
   async def get_work_order(work_order_id: int) -> dict | None:
       """Fetch a single work order by id."""
       async with with_conn() as conn:
           row = await WorkOrderRepository(conn).get(work_order_id)
       return WorkOrderOut.model_validate(dict(row)).model_dump(mode="json") if row else None

2. Reuse the plural tool. get_work_orders(cell_id=..., limit=1) + client-side filter on id. Requires knowing cell_id upfront (Sentinel has it; pass through the spawn).

Tool event broadcasting contract

Both tool_call_started and tool_call_completed must fire with the field sets from frontend/src/lib/ws.types.ts:18-29:

import time

for tool_use in response_tool_uses:
    t0 = time.monotonic()
    await ws_manager.broadcast("tool_call_started", {
        "agent": "investigator",
        "tool_name": tool_use.name,
        "args": tool_use.input,
        "turn_id": turn_id,
    })
    # dispatch (render_* / ask_kb_builder / MCP / submit_rca)
    await ws_manager.broadcast("tool_call_completed", {
        "agent": "investigator",
        "tool_name": tool_use.name,
        "duration_ms": int((time.monotonic() - t0) * 1000),
        "turn_id": turn_id,
    })

Safety nets (non-negotiable for demo)

Warning

Without MAX_TURNS + wall-clock timeout + outer try/except, a hung tool call leaves the WO stuck in status='detected' forever — Scene 3 renders nothing.

MAX_TURNS = 12

async def run_investigator(work_order_id: int) -> None:
    turn_id = str(uuid.uuid4())
    current_turn_id.set(turn_id)
    try:
        await asyncio.wait_for(
            _run_investigator_body(work_order_id, turn_id),
            timeout=120.0,
        )
    except asyncio.TimeoutError:
        await _fallback_rca(work_order_id, "Investigation timed out", turn_id)
    except Exception as exc:
        log.exception("Investigator crashed for WO %d", work_order_id)
        await _fallback_rca(work_order_id, f"Investigation failed: {exc}", turn_id)


async def _fallback_rca(work_order_id: int, reason: str, turn_id: str) -> None:
    # UPDATE work_order directly (or via a new update_work_order MCP tool — see above)
    await ws_manager.broadcast("rca_ready", {
        "work_order_id": work_order_id,
        "rca_summary": reason,
        "confidence": 0.0,
        "turn_id": turn_id,
    })

submit_rca handler side-effects

When the LLM calls submit_rca:

1. UPDATE work_order with rca_summary, confidence, status="analyzed".
2. INSERT failure_history(cell_id, failure_time, failure_mode, root_cause, signal_patterns, work_order_id) — signal_patterns jsonb column added in migration 007 (issue M1.3 — Migration 007 : failure_history.signal_patterns #4 closed).
3. ws_manager.broadcast("rca_ready", {work_order_id, rca_summary, confidence, turn_id}).
4. asyncio.create_task(run_work_order_generator(work_order_id)) — see M5.1 — Work Order Generator agent #30.
5. Break the agent loop; then ws_manager.broadcast("agent_end", {agent: "investigator", turn_id, finish_reason: "submit_rca"}).

Extended thinking + tool use (pointer to #27)

Important

Thinking is enabled per #27. The signed thinking block returned by the API must be preserved verbatim in messages[] on the next request during a tool-use turn. Not preserving it causes a 400 on the continuation messages.create(). See #27 for the exact pattern.

Drop render_correlation_matrix

ui_tools.py:18-23 dropped this tool with a strong rationale (LLM would hallucinate correlation numbers — fatal for a trust-critical predictive-maintenance demo). Update the mermaid in the issue body to reference render_signal_chart with signal_def_ids: list[int] (multi-signal overlay). Same visual message, grounded in real data.

Cross-references

• Upstream: M3.1 — Anthropic client wrapper #17 (M3.1 anthropic_client.py), M2.7 — MCPClient singleton (auto-discovery + call_tool) #14 (M2.7 MCPClient), M2.9 — UI tools generative (render_*) déclarés dans les agents #16 (M2.9 INVESTIGATOR_RENDER_TOOLS), M3.5 — KB Builder en mode "agent appelé" (ask_kb_builder handler) #21 (M3.5 answer_kb_question shipped), M4.1 — WSManager broadcast manager #23 (M4.1 WSManager), M4.2 — Sentinel asyncio loop #24 (M4.2 triggers this)
• Related: M4.5 — Extended thinking sur Investigator (Opus 4.7 wow factor) #27 (M4.5 thinking), M4.6 — Agent-as-tool (handoffs dynamiques) #28 (M4.6 ask_kb_builder), M4.7 — Memory flex scene #29 (M4.7 failure_history)
• Downstream: M5.1 — Work Order Generator agent #30 (spawns Work Order Generator)
• Frontend consumers: M7.5 — Artifact registry + dispatcher #44 (M7.5 registry renders diagnostic_card), M8.1 — Artifact SignalChart #45 (M8.1 SignalChart), M8.4 — Agent Activity Feed #48 (M8.4 activity feed), M8.5 — Agent Inspector + thinking stream #49 (M8.5 thinking stream)