9. The same flow as explicit code

The failure policy, as a table

A missing manual citation deliberately does not fail the flow: it degrades the answer and says so. Deciding which failures are fatal is the design work, and in the agent version that decision is made implicitly by a planner, once per run, differently each time.

Symptom too vague to discriminate escalated

top two fault codes within 0.02 ([('F-110', 0.0), ('F-109', 0.0)]); refusing to guess

Trade-off, stated plainly

The choice between them is a question about the distribution of incoming requests, not a question about technology. If ninety per cent of requests are this shape, the flow below serves them at a rate the agent cannot reach, and the agent is the fallback for the rest.

The source, read at render time

pipeline/explicit_flow.py, via inspect.getsource. This is the code that produced the three runs above.

"""The same flow, written out as a state machine.

This module is displayed verbatim on screen 9 and executed by it. Whatever you
read there is what ran.

The point being made: everything the agent loop does implicitly -- deciding what
to do next, remembering what it learned, handling a step that fails -- becomes
three visible things here. A declared state object, a transition table, and an
explicit failure policy per step. Nothing is hidden in a prompt.

What you lose: the flow cannot handle a request it was not written for. What you
gain is on screen 6. Both are real, and the choice between them is a question
about the distribution of requests, not a question about technology.
"""

from __future__ import annotations

import re
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable

from . import deterministic
from .mcp_client import MCPClient, MCPError


class Stage(str, Enum):
    PARSE = "parse"
    HISTORY = "history"
    DIAGNOSE = "diagnose"
    PART = "part"
    STOCK = "stock"
    COMPOSE = "compose"
    DONE = "done"
    FAILED = "failed"
    ESCALATED = "escalated"


class Failure(str, Enum):
    """What to do when a step does not produce what the next one needs."""

    RETRY_ONCE = "retry_once"        # transient: a timeout, a 429
    ESCALATE = "escalate"            # a human has to decide
    ABORT = "abort"                  # the request cannot be served


@dataclass
class State:
    """Everything the flow knows. Declared up front, not accumulated in a prompt."""

    request: str
    model_variant: str | None = None
    symptom: str | None = None
    tickets: list[dict] = field(default_factory=list)
    fault_code: str | None = None
    diagnosis_margin: float = 0.0
    part_no: str | None = None
    availability: dict[str, Any] = field(default_factory=dict)
    sourcing: str | None = None
    citation: dict[str, Any] = field(default_factory=dict)
    answer: str = ""
    stage: Stage = Stage.PARSE
    reason: str = ""
    attempts: dict[str, int] = field(default_factory=dict)
    log: list[dict[str, Any]] = field(default_factory=list)

    def record(self, stage: Stage, ok: bool, detail: str, ms: float) -> None:
        self.log.append({"stage": stage.value, "ok": ok, "detail": detail,
                         "ms": round(ms, 1)})


VARIANT_RE = re.compile(r"\b([A-Z]{2}-\d{2,3}-[ABS])\b")
AMBIGUITY_FLOOR = 0.02


class ExplicitFlow:
    """Six steps, a transition table, and a failure policy for each."""

    POLICY: dict[Stage, Failure] = {
        Stage.PARSE: Failure.ABORT,        # no variant, no answer worth giving
        Stage.HISTORY: Failure.RETRY_ONCE,  # a tool call can be transient
        Stage.DIAGNOSE: Failure.ESCALATE,  # ambiguous symptom is a human's call
        Stage.PART: Failure.ESCALATE,      # no fitting part is a catalogue gap
        Stage.STOCK: Failure.RETRY_ONCE,
        Stage.COMPOSE: Failure.ABORT,
    }

    def __init__(self, client: MCPClient | None = None,
                 endpoints: dict[str, str] | None = None):
        self.client = client or MCPClient()
        self.ep = endpoints or {"tickets": "tickets", "parts": "parts",
                                "manuals": "manuals"}

    # -- the transition table ----------------------------------------------
    def run(self, request: str) -> State:
        s = State(request=request)
        steps: list[tuple[Stage, Callable[[State], bool], Stage]] = [
            (Stage.PARSE, self.parse, Stage.HISTORY),
            (Stage.HISTORY, self.history, Stage.DIAGNOSE),
            (Stage.DIAGNOSE, self.diagnose, Stage.PART),
            (Stage.PART, self.part, Stage.STOCK),
            (Stage.STOCK, self.stock, Stage.COMPOSE),
            (Stage.COMPOSE, self.compose, Stage.DONE),
        ]
        for stage, fn, nxt in steps:
            s.stage = stage
            t0 = time.perf_counter()
            try:
                ok = fn(s)
                err = ""
            except MCPError as exc:
                ok, err = False, str(exc)
            ms = (time.perf_counter() - t0) * 1000
            s.record(stage, ok, err or s.reason, ms)
            if ok:
                continue
            policy = self.POLICY[stage]
            if policy is Failure.RETRY_ONCE and s.attempts.get(stage.value, 0) == 0:
                s.attempts[stage.value] = 1
                t1 = time.perf_counter()
                ok2 = False
                try:
                    ok2 = fn(s)
                except MCPError as exc:
                    s.reason = str(exc)
                s.record(stage, ok2, "retry: " + (s.reason or "ok"),
                         (time.perf_counter() - t1) * 1000)
                if ok2:
                    continue
                policy = Failure.ESCALATE
            s.stage = Stage.ESCALATED if policy is Failure.ESCALATE else Stage.FAILED
            return s
        s.stage = Stage.DONE
        return s

    # -- the six steps ------------------------------------------------------
    def parse(self, s: State) -> bool:
        m = VARIANT_RE.search(s.request)
        if not m:
            s.reason = ("no model variant of the form AC-250-S in the request; "
                        "the suffix is not inferable from a serial number")
            return False
        s.model_variant = m.group(1)
        s.symptom = s.request
        return True

    def history(self, s: State) -> bool:
        r = self.client.call_tool(self.ep["tickets"], "search_tickets", {
            "model_variant": s.model_variant, "symptom": s.symptom, "limit": 8})
        if r.get("isError"):
            s.reason = "ticket search returned a tool execution error"
            return False
        s.tickets = (r.get("structuredContent") or {}).get("tickets", [])
        if not s.tickets:
            s.reason = f"no ticket history for {s.model_variant}"
            return False
        return True

    def diagnose(self, s: State) -> bool:
        dx = deterministic.diagnose(s.symptom or "")
        s.fault_code = dx["fault_code"]
        s.diagnosis_margin = dx["margin"]
        if dx["ambiguous"]:
            s.reason = (f"top two fault codes within {AMBIGUITY_FLOOR} "
                        f"({dx['top']}); refusing to guess")
            return False
        return True

    def part(self, s: State) -> bool:
        s.part_no = deterministic.select_part(s.fault_code, s.model_variant)
        if not s.part_no:
            s.reason = (f"no part fits {s.model_variant} for {s.fault_code}; "
                        "do not substitute across variants")
            return False
        return True

    def stock(self, s: State) -> bool:
        r = self.client.call_tool(self.ep["parts"], "check_availability",
                                  {"part_no": s.part_no})
        if r.get("isError"):
            s.reason = "availability check failed"
            return False
        s.availability = r.get("structuredContent") or {}
        s.sourcing = deterministic.sourcing(s.part_no)
        return True

    def compose(self, s: State) -> bool:
        fam = (s.model_variant or "").rsplit("-", 1)[0]
        try:
            r = self.client.call_tool(self.ep["manuals"], "search_manual", {
                "query": f"{s.fault_code} replacement procedure",
                "product_family": fam, "k": 1})
            ps = ((r.get("structuredContent") or {}).get("passages") or [])
            s.citation = ps[0] if ps else {}
        except MCPError:
            s.citation = {}   # a missing citation degrades the answer, not the flow
        s.answer = (
            f"{s.model_variant}: {s.fault_code}. Fit {s.part_no}. "
            f"Sourcing: {s.sourcing}. "
            f"Stock {s.availability.get('stock_qty')}, "
            f"lead time {s.availability.get('lead_time_days')} days."
            + (f" Reference: {s.citation.get('doc_title')} p.{s.citation.get('page')}."
               if s.citation else " No manual citation available.")
        )
        return True