Add ask_expert and eval capture (A14/A16) — step 7c

Completes step 7.

- expert.py: one stateless call for a question needing no repo access.
  Structured output is contracted at the prompt level, since vendors
  differ on JSON enforcement, so the reply is parsed and validated
  locally with one retry — tolerating a fenced code block, because
  models wrap JSON habitually, but not missing fields
- The per-task budget is counted from the canonical log rather than
  memory, so it survives a restart and cannot hand a circling executor
  a fresh allowance. Exhaustion raises rather than answering: a caller
  out of consults should get on with the work or escalate
- Consults are recorded as delegation_started/completed with role
  consultant. A16 asks for a "consult/delegation eval set", and
  reusing the composite events gets consult evaluation for free rather
  than inventing a parallel record
- evals.py pairs starts with completions into eval records, with
  verdicts kept outside the canonical log: the log records what
  happened, a verdict is a later judgement about it, and judgements
  get revised. Schema v3 adds the outcomes table, so a v1 store now
  traverses two migrations — a better exercise of the hook than one
- ask_expert is offered to the executor as a tool only when a desk is
  configured, in addition to the observable escalation triggers and
  never instead of them (A12). Consulting is not mutating, so a
  completion claim backed only by consults is still challenged

Found live: passing an empty message list and putting the question in
the system prompt fails outright — the Messages API requires at least
one non-system message. The question now travels as a synthesised user
message that is never appended to the log, which keeps the consult
stateless while remaining a valid request.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

nisse-dev/src/nisse_dev/evals.py

@@ -0,0 +1,88 @@
+"""Delegation and consult evaluation capture (spec A16).
+
+Each ``delegation_started`` / ``delegation_completed`` pair is already a clean
+input-output record; this pairs them up and attaches a verdict, so "best model
+this week" can be decided by re-running a captured set rather than by benchmark
+chatter. Without it, swapping the frontier tier becomes something to avoid,
+because the effect is unmeasurable.
+
+Verdicts live outside the canonical log: the log records what happened, a
+verdict is a later judgement about it, and judgements get revised.
+"""
+
+import json
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from nisse_dev.canonical import DelegationCompleted, DelegationStarted
+from nisse_dev.state import StateStore
+
+UNMARKED = "unmarked"
+
+
+@dataclass(frozen=True)
+class EvalRecord:
+    event_id: int  # of the completion, which is what a verdict attaches to
+    role: str
+    model: str
+    brief: str
+    caps: dict[str, Any]
+    handback: str
+    artifacts: list[str]
+    tokens: int
+    duration_s: float
+    exit_status: str
+    verdict: str = UNMARKED
+    note: str = ""
+
+
+def collect(store: StateStore, after: int = 0) -> list[EvalRecord]:
+    """Pair each start with the completion that follows it.
+
+    A start with no completion is skipped rather than reported half-formed:
+    it means the process died mid-delegation, which A3 handles, and an
+    unfinished record would pollute the eval set.
+    """
+    verdicts = store.outcomes()
+    records: list[EvalRecord] = []
+    pending: tuple[int, DelegationStarted] | None = None
+    for stored in store.read_events(after=after):
+        if isinstance(stored.event, DelegationStarted):
+            pending = (stored.id, stored.event)
+        elif isinstance(stored.event, DelegationCompleted) and pending is not None:
+            _, started = pending
+            verdict, note = verdicts.get(stored.id, (UNMARKED, ""))
+            records.append(
+                EvalRecord(
+                    event_id=stored.id,
+                    role=started.role,
+                    model=started.model,
+                    brief=started.brief,
+                    caps=started.caps,
+                    handback=stored.event.handback,
+                    artifacts=stored.event.artifacts,
+                    tokens=stored.event.tokens,
+                    duration_s=stored.event.duration_s,
+                    exit_status=stored.event.exit_status,
+                    verdict=verdict,
+                    note=note,
+                )
+            )
+            pending = None
+    return records
+
+
+def to_jsonl(records: list[EvalRecord]) -> str:
+    """JSON lines, so a captured set can be diffed and re-run per record."""
+    return "\n".join(json.dumps(asdict(r)) for r in records)
+
+
+def summarise(records: list[EvalRecord]) -> dict[str, Any]:
+    by_model: dict[str, dict[str, int]] = {}
+    for record in records:
+        bucket = by_model.setdefault(
+            record.model, {"total": 0, "good": 0, "bad": 0, UNMARKED: 0}
+        )
+        bucket["total"] += 1
+        bucket[record.verdict if record.verdict in ("good", "bad") else UNMARKED] += 1
+    return by_model

nisse-dev/src/nisse_dev/executor.py

@@ -11,12 +11,13 @@ the human with the failures attached rather than engaging a frontier model.
 """
 
 from dataclasses import dataclass
-from typing import Literal
+from typing import Any, Literal
 
 from nisse_dev.canonical import ToolCall, ToolResult
+from nisse_dev.expert import ASK_EXPERT_TOOL, BudgetExhausted, ExpertDesk
 from nisse_dev.llm import LLMClient, LLMError, RoleMap
 from nisse_dev.state import StateStore
-from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, Workspace
+from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, ToolOutcome, Workspace
 
 EXECUTOR_SYSTEM = """You are nisse-dev's executor, doing mechanical work.
 
@@ -56,6 +57,7 @@ class LLMExecutor:
         *,
         max_turns: int = DEFAULT_MAX_TURNS,
         failure_limit: int = DEFAULT_FAILURE_LIMIT,
+        expert: ExpertDesk | None = None,
     ) -> None:
         self._client = client
         self._store = store
@@ -63,6 +65,28 @@ class LLMExecutor:
         self._workspace = workspace
         self._max_turns = max_turns
         self._failure_limit = failure_limit
+        self._expert = expert
+
+    def _tools(self) -> list[dict[str, Any]]:
+        # A12: ask_expert exists in addition to the observable escalation
+        # triggers, never as a replacement for them.
+        if self._expert is None:
+            return EXECUTOR_TOOLS
+        return [*EXECUTOR_TOOLS, ASK_EXPERT_TOOL]
+
+    def _consult(self, arguments: dict[str, Any]) -> ToolOutcome:
+        assert self._expert is not None
+        question = str(arguments.get("question", "")).strip()
+        if not question:
+            return ToolOutcome("ask_expert needs a question", success=False)
+        try:
+            return ToolOutcome(self._expert.ask(question).render(), success=True)
+        except BudgetExhausted as exc:
+            # Not an error the model should route around: it is the signal
+            # that it is circling and should get on with the work or fail.
+            return ToolOutcome(str(exc), success=False)
+        except LLMError as exc:
+            return ToolOutcome(f"the consultant did not answer: {exc}", success=False)
 
     def run(self, milestone: int) -> ExecutorOutcome:
         model = self._roles.resolve("executor")
@@ -78,7 +102,7 @@ class LLMExecutor:
             try:
                 produced = self._client.complete(
                     self._store.read_events(),
-                    tools=EXECUTOR_TOOLS,
+                    tools=self._tools(),
                     model=model,
                     system=system,
                 )
@@ -110,7 +134,10 @@ class LLMExecutor:
             for call in calls:
                 if call.tool in MUTATING_TOOLS:
                     did_work = True
-                outcome = self._workspace.dispatch(call.tool, call.arguments)
+                if call.tool == "ask_expert":
+                    outcome = self._consult(call.arguments)
+                else:
+                    outcome = self._workspace.dispatch(call.tool, call.arguments)
                 self._store.append_event(
                     ToolResult(
                         call_id=call.call_id,

nisse-dev/src/nisse_dev/expert.py

@@ -0,0 +1,214 @@
+"""`ask_expert` (spec A14).
+
+The one case delegation does not cover: a question needing no repo access —
+"which of these two approaches is better", "is this API deprecated". A single
+stateless call, a text answer, no session state, no diff application.
+
+Two constraints from A14 shape the implementation:
+
+**Structured output is contracted at the prompt level, not the API level.**
+Vendors differ on JSON mode and schema enforcement, so the answer is parsed and
+validated against a local schema, with one retry. This runs over LiteLLM's
+*normalized* route, which A13 reserves for calls where fidelity is cheap.
+
+**A per-task budget**, so a confused executor cannot spend twenty consults
+circling. The count comes from the canonical log rather than a counter in
+memory: it survives a restart, and it is observable from outside the model,
+which is the same reasoning A12 applies to escalation triggers.
+
+Consults are recorded as ``delegation_started`` / ``delegation_completed`` with
+role ``consultant``. That is deliberate — A16 wants a "consult/delegation eval
+set", and reusing the composite events gets consult evaluation for free rather
+than inventing a parallel record.
+"""
+
+import json
+import re
+from dataclasses import dataclass
+
+from nisse_dev.canonical import (
+    DelegationCompleted,
+    DelegationStarted,
+    StoredEvent,
+    UserMessage,
+)
+from nisse_dev.llm import LLMClient, LLMError, RoleMap
+from nisse_dev.state import StateStore
+
+CONSULTANT_ROLE = "consultant"
+DEFAULT_BUDGET = 3
+
+SYSTEM_PROMPT = """You are a consultant answering one self-contained question.
+
+You have no access to the caller's repository, so answer from the question
+alone. If the question cannot be answered without seeing the code, say so
+plainly in `answer` rather than guessing.
+
+Reply with a single JSON object and nothing else:
+
+{"answer": "...", "confidence": "high" | "medium" | "low", "caveats": "..."}
+
+Keep `answer` direct and specific. Put what would change your mind, or what you
+are assuming, in `caveats`."""
+
+_RETRY_NUDGE = (
+    "\n\nYour previous reply was not a single valid JSON object with the keys "
+    "answer, confidence and caveats. Reply with only that object."
+)
+_CONFIDENCES = {"high", "medium", "low"}
+
+
+ASK_EXPERT_TOOL: dict = {
+    "type": "function",
+    "function": {
+        "name": "ask_expert",
+        "description": (
+            "Ask a stronger model one self-contained question that needs no "
+            "access to this repository — a design trade-off, or how a library "
+            "behaves. It cannot see your files. Use sparingly; the budget is "
+            "small."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {"question": {"type": "string"}},
+            "required": ["question"],
+        },
+    },
+}
+
+
+class BudgetExhausted(RuntimeError):
+    """Raised rather than silently answering: a caller that has burned its
+    consult budget is circling, and should escalate or ask the operator."""
+
+
+@dataclass(frozen=True)
+class ExpertAnswer:
+    answer: str
+    confidence: str
+    caveats: str
+
+    def render(self) -> str:
+        body = f"{self.answer}\n\n(confidence: {self.confidence})"
+        return f"{body}\n\nCaveats: {self.caveats}" if self.caveats else body
+
+
+class ExpertDesk:
+    def __init__(
+        self,
+        client: LLMClient,
+        store: StateStore,
+        roles: RoleMap,
+        *,
+        budget: int = DEFAULT_BUDGET,
+    ) -> None:
+        self._client = client
+        self._store = store
+        self._roles = roles
+        self._budget = budget
+
+    def consults_used(self, since_event_id: int = 0) -> int:
+        return sum(
+            1
+            for e in self._store.read_events(after=since_event_id)
+            if isinstance(e.event, DelegationStarted)
+            and e.event.role == CONSULTANT_ROLE
+        )
+
+    def ask(self, question: str, *, since_event_id: int = 0) -> ExpertAnswer:
+        used = self.consults_used(since_event_id)
+        if used >= self._budget:
+            raise BudgetExhausted(
+                f"consult budget of {self._budget} is spent for this task "
+                f"({used} used); escalate or ask the operator instead"
+            )
+        model = self._roles.resolve(CONSULTANT_ROLE)
+        self._store.append_event(
+            DelegationStarted(
+                brief=question,
+                role=CONSULTANT_ROLE,
+                model=model,
+                caps={"consults": self._budget, "used": used},
+            )
+        )
+        answer, note = self._answer(question, model)
+        self._store.append_event(
+            DelegationCompleted(
+                handback=answer.render() if answer else note,
+                artifacts=[],
+                tokens=0,
+                duration_s=0.0,
+                exit_status="answered" if answer else "failed",
+            )
+        )
+        if answer is None:
+            raise LLMError(f"ask_expert produced no usable answer: {note}")
+        return answer
+
+    def _answer(self, question: str, model: str) -> tuple[ExpertAnswer | None, str]:
+        system = SYSTEM_PROMPT
+        last = "no reply"
+        for attempt in (1, 2):
+            try:
+                # Stateless by design (A14): the question travels as a
+                # synthesised message that is never appended to the log, so a
+                # consult neither depends on nor leaks repo context. It cannot
+                # ride in the system prompt alone — the Messages API requires
+                # at least one non-system message.
+                produced = self._client.complete(
+                    [
+                        StoredEvent(
+                            id=0,
+                            ts="",
+                            event=UserMessage(text=question, source="ask_expert"),
+                        )
+                    ],
+                    tools=None,
+                    model=model,
+                    system=system,
+                )
+            except LLMError as exc:
+                return None, str(exc)
+            text = "\n".join(
+                getattr(e, "text", "") for e in produced if hasattr(e, "text")
+            ).strip()
+            parsed = parse_answer(text)
+            if parsed is not None:
+                return parsed, "ok"
+            last = text[:300] or "empty reply"
+            if attempt == 1:
+                system = SYSTEM_PROMPT + _RETRY_NUDGE
+        return None, f"unparseable reply: {last}"
+
+
+def parse_answer(text: str) -> ExpertAnswer | None:
+    """Validated locally because the prompt-level contract is all we have.
+
+    Tolerates a fenced code block, since models wrap JSON in one habitually,
+    but not missing or malformed fields — those are a failed turn.
+    """
+    candidate = text.strip()
+    fenced = re.search(r"```(?:json)?\s*(.*?)```", candidate, re.DOTALL)
+    if fenced:
+        candidate = fenced.group(1).strip()
+    start, end = candidate.find("{"), candidate.rfind("}")
+    if start == -1 or end <= start:
+        return None
+    try:
+        data = json.loads(candidate[start : end + 1])
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(data, dict):
+        return None
+    answer = data.get("answer")
+    confidence = data.get("confidence")
+    if not isinstance(answer, str) or not answer.strip():
+        return None
+    if not isinstance(confidence, str) or confidence.lower() not in _CONFIDENCES:
+        return None
+    caveats = data.get("caveats")
+    return ExpertAnswer(
+        answer=answer.strip(),
+        confidence=confidence.lower(),
+        caveats=caveats.strip() if isinstance(caveats, str) else "",
+    )

nisse-dev/src/nisse_dev/state.py

@@ -27,7 +27,7 @@ from nisse_dev.canonical import (
 )
 from nisse_dev.oplog import OpLog
 
-CURRENT_SCHEMA_VERSION = 2
+CURRENT_SCHEMA_VERSION = 3
 
 Migration = Callable[[sqlite3.Connection], None]
 
@@ -41,12 +41,29 @@ CREATE TABLE canonical_events (
 """
 
 
+_EVAL_OUTCOMES_DDL = """
+CREATE TABLE eval_outcomes (
+    event_id INTEGER PRIMARY KEY,
+    verdict TEXT NOT NULL,
+    note TEXT NOT NULL DEFAULT '',
+    marked_at TEXT NOT NULL
+)
+"""
+
+
 def _migrate_v1_add_canonical_events(conn: sqlite3.Connection) -> None:
     conn.execute(_CANONICAL_EVENTS_DDL)
 
 
+def _migrate_v2_add_eval_outcomes(conn: sqlite3.Connection) -> None:
+    conn.execute(_EVAL_OUTCOMES_DDL)
+
+
 # Maps version N to the migration that takes the schema from N to N+1.
-MIGRATIONS: Mapping[int, Migration] = {1: _migrate_v1_add_canonical_events}
+MIGRATIONS: Mapping[int, Migration] = {
+    1: _migrate_v1_add_canonical_events,
+    2: _migrate_v2_add_eval_outcomes,
+}
 
 
 class StateSchemaError(RuntimeError):
@@ -93,6 +110,7 @@ _SCHEMA = [
     )
     """,
     _CANONICAL_EVENTS_DDL,
+    _EVAL_OUTCOMES_DDL,
 ]
 
 
@@ -294,6 +312,27 @@ class StateStore:
             for row in rows
         ]
 
+    def mark_outcome(self, event_id: int, verdict: str, note: str = "") -> None:
+        """Record whether a delegation or consult was good (A16).
+
+        Kept out of the canonical log on purpose: the log is an append-only
+        record of what happened, and a verdict is a later judgement about it
+        that may be revised.
+        """
+        self._conn.execute(
+            "INSERT INTO eval_outcomes (event_id, verdict, note, marked_at) "
+            "VALUES (?, ?, ?, ?) ON CONFLICT(event_id) DO UPDATE SET "
+            "verdict = excluded.verdict, note = excluded.note, "
+            "marked_at = excluded.marked_at",
+            (event_id, verdict, note, _now()),
+        )
+
+    def outcomes(self) -> dict[int, tuple[str, str]]:
+        rows = self._conn.execute(
+            "SELECT event_id, verdict, note FROM eval_outcomes"
+        ).fetchall()
+        return {r["event_id"]: (r["verdict"], r["note"]) for r in rows}
+
     def close(self) -> None:
         self._conn.close()
 

nisse-dev/tests/test_canonical.py

@@ -16,7 +16,7 @@ from nisse_dev.canonical import (
     from_payload,
     mint_call_id,
 )
-from nisse_dev.state import StateStore
+from nisse_dev.state import CURRENT_SCHEMA_VERSION, StateStore
 
 ALL_EVENTS = [
     UserMessage(text="use demo", source="matrix:$abc"),
@@ -116,9 +116,12 @@ def test_default_migration_upgrades_a_real_v1_store(tmp_path: Path) -> None:
     conn.close()
 
     with StateStore(db) as store:
-        assert store.schema_version == 2
+        assert store.schema_version == CURRENT_SCHEMA_VERSION
         state = store.get_state()
         assert state is not None
         assert state.state == "IDLE"  # v1 data intact
         store.append_event(UserMessage(text="hello", source="m:1"))
         assert len(store.read_events()) == 1
+        # Two migrations were traversed, so both later tables must exist.
+        store.mark_outcome(1, "good", "arrived via v1 -> v3")
+        assert store.outcomes()[1] == ("good", "arrived via v1 -> v3")

nisse-dev/tests/test_evals.py

@@ -0,0 +1,129 @@
+import io
+import json
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    DelegationCompleted,
+    DelegationStarted,
+)
+from nisse_dev.evals import UNMARKED, collect, summarise, to_jsonl
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> StateStore:
+    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+
+
+def record_pair(
+    store: StateStore,
+    *,
+    brief: str = "add greet",
+    model: str = "claude-haiku-4-5",
+    role: str = "delegate",
+    status: str = "handback",
+) -> int:
+    store.append_event(
+        DelegationStarted(brief=brief, role=role, model=model, caps={"turns": 5})
+    )
+    store.append_event(
+        DelegationCompleted(
+            handback="what changed: greet.py",
+            artifacts=["greet.py"],
+            tokens=1200,
+            duration_s=9.5,
+            exit_status=status,
+        )
+    )
+    return store.read_events()[-1].id
+
+
+def test_pairs_a_start_with_its_completion(store: StateStore) -> None:
+    record_pair(store)
+    [record] = collect(store)
+    assert record.brief == "add greet"
+    assert record.model == "claude-haiku-4-5"
+    assert record.exit_status == "handback"
+    assert record.artifacts == ["greet.py"]
+    assert record.verdict == UNMARKED
+
+
+def test_unpaired_start_is_skipped(store: StateStore) -> None:
+    # A crash mid-delegation leaves a start with no completion; a half-formed
+    # record would pollute the set A16 exists to make trustworthy.
+    store.append_event(
+        DelegationStarted(brief="died", role="delegate", model="m", caps={})
+    )
+    assert collect(store) == []
+
+
+def test_unrelated_events_between_the_pair_are_ignored(store: StateStore) -> None:
+    store.append_event(
+        DelegationStarted(brief="b", role="delegate", model="m", caps={})
+    )
+    store.append_event(AssistantText(text="working", model="m"))
+    store.append_event(
+        DelegationCompleted(
+            handback="done",
+            artifacts=[],
+            tokens=1,
+            duration_s=1.0,
+            exit_status="handback",
+        )
+    )
+    assert len(collect(store)) == 1
+
+
+def test_verdicts_attach_to_the_completion(store: StateStore) -> None:
+    event_id = record_pair(store)
+    store.mark_outcome(event_id, "good", "clean handback, tests passed")
+    [record] = collect(store)
+    assert record.verdict == "good"
+    assert "tests passed" in record.note
+
+
+def test_a_verdict_can_be_revised(store: StateStore) -> None:
+    # Judgements get revised; the log they describe does not.
+    event_id = record_pair(store)
+    store.mark_outcome(event_id, "good")
+    store.mark_outcome(event_id, "bad", "regressed later")
+    [record] = collect(store)
+    assert record.verdict == "bad"
+    assert record.note == "regressed later"
+
+
+def test_consults_and_delegations_share_the_set(store: StateStore) -> None:
+    record_pair(store, role="delegate")
+    record_pair(store, role="consultant", brief="which approach?")
+    roles = [r.role for r in collect(store)]
+    assert roles == ["delegate", "consultant"]
+
+
+def test_collect_can_start_after_a_marker(store: StateStore) -> None:
+    record_pair(store, brief="old")
+    marker = store.read_events()[-1].id
+    record_pair(store, brief="new")
+    assert [r.brief for r in collect(store, after=marker)] == ["new"]
+
+
+def test_jsonl_round_trips(store: StateStore) -> None:
+    record_pair(store)
+    line = to_jsonl(collect(store))
+    assert json.loads(line)["model"] == "claude-haiku-4-5"
+
+
+def test_summary_groups_by_model(store: StateStore) -> None:
+    good = record_pair(store, model="haiku")
+    record_pair(store, model="haiku", status="capped")
+    record_pair(store, model="opus")
+    store.mark_outcome(good, "good")
+
+    summary = summarise(collect(store))
+    assert summary["haiku"]["total"] == 2
+    assert summary["haiku"]["good"] == 1
+    assert summary["haiku"][UNMARKED] == 1
+    assert summary["opus"]["total"] == 1

nisse-dev/tests/test_executor.py

@@ -223,3 +223,95 @@ def test_multiple_calls_in_one_turn_all_run(store: StateStore, ws: Workspace) ->
         e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
     ]
     assert {r.call_id for r in results} == {"a", "b"}
+
+
+class StubDesk:
+    """Stands in for ExpertDesk without a model behind it."""
+
+    def __init__(self, answer: str = "use sqlite", error: Exception | None = None):
+        self.asked: list[str] = []
+        self._answer = answer
+        self._error = error
+
+    def ask(self, question: str, **kw: Any) -> Any:
+        self.asked.append(question)
+        if self._error is not None:
+            raise self._error
+
+        class _A:
+            def render(self_inner) -> str:
+                return self._answer
+
+        return _A()
+
+
+def test_ask_expert_is_offered_only_when_a_desk_exists(
+    store: StateStore, ws: Workspace
+) -> None:
+    client = ScriptedClient([])
+    build(client, store, ws).run(milestone=1)
+    assert all(t["function"]["name"] != "ask_expert" for t in client.calls[0]["tools"])
+
+    client2 = ScriptedClient([])
+    build(client2, store, ws, expert=StubDesk()).run(milestone=1)
+    names = [t["function"]["name"] for t in client2.calls[0]["tools"]]
+    assert names[-1] == "ask_expert"
+    # A12: in addition to the observable triggers, not instead of them.
+    assert "run_command" in names
+
+
+def test_ask_expert_answer_comes_back_as_a_tool_result(
+    store: StateStore, ws: Workspace
+) -> None:
+    desk = StubDesk(answer="prefer sqlite here")
+    client = ScriptedClient(
+        [
+            [call("ask_expert", question="sqlite or postgres?")],
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="done", model="m")],
+        ]
+    )
+    build(client, store, ws, expert=desk).run(milestone=1)
+
+    assert desk.asked == ["sqlite or postgres?"]
+    results = [
+        e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
+    ]
+    assert "prefer sqlite here" in results[0].result
+
+
+def test_consulting_is_not_doing_the_work(store: StateStore, ws: Workspace) -> None:
+    # Asking is not changing anything, so a completion claim backed only by
+    # consults is still challenged.
+    client = ScriptedClient(
+        [
+            [call("ask_expert", question="what should I do?")],
+            [AssistantText(text="I think it is done", model="m")],
+            [AssistantText(text="really done", model="m")],
+        ]
+    )
+    outcome = build(client, store, ws, expert=StubDesk()).run(milestone=1)
+    assert outcome.status == "escalate"
+
+
+def test_exhausted_budget_is_reported_to_the_model_not_raised(
+    store: StateStore, ws: Workspace
+) -> None:
+    from nisse_dev.expert import BudgetExhausted
+
+    desk = StubDesk(error=BudgetExhausted("consult budget of 3 is spent"))
+    client = ScriptedClient(
+        [
+            [call("ask_expert", question="again?")],
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="done", model="m")],
+        ]
+    )
+    outcome = build(client, store, ws, expert=desk).run(milestone=1)
+
+    assert outcome.status == "completed"  # it got on with the work
+    results = [
+        e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
+    ]
+    assert not results[0].success
+    assert "budget" in results[0].result

nisse-dev/tests/test_expert.py

@@ -0,0 +1,186 @@
+import io
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    CanonicalEvent,
+    DelegationCompleted,
+    DelegationStarted,
+    StoredEvent,
+)
+from nisse_dev.expert import (
+    DEFAULT_BUDGET,
+    BudgetExhausted,
+    ExpertAnswer,
+    ExpertDesk,
+    parse_answer,
+)
+from nisse_dev.llm import LLMError, RoleMap
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+
+ROLES = RoleMap(
+    executor="llama3.2",
+    planner="haiku",
+    delegate="haiku",
+    consultant="claude-haiku-4-5",
+)
+
+GOOD = '{"answer": "Use SQLite; it is enough here.", "confidence": "high", "caveats": "Assumes one writer."}'
+
+
+class ScriptedClient:
+    def __init__(self, replies: list[str]) -> None:
+        self._replies = replies
+        self.calls: list[dict[str, Any]] = []
+
+    def complete(
+        self,
+        events: list[StoredEvent],
+        tools: list[dict[str, Any]] | None,
+        model: str,
+        system: str | None = None,
+    ) -> list[CanonicalEvent]:
+        self.calls.append(
+            {"events": events, "tools": tools, "model": model, "system": system}
+        )
+        if not self._replies:
+            return [AssistantText(text="nothing more", model=model)]
+        return [AssistantText(text=self._replies.pop(0), model=model)]
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> StateStore:
+    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+
+
+def desk(client: Any, store: StateStore, **kw: Any) -> ExpertDesk:
+    return ExpertDesk(client, store, ROLES, **kw)
+
+
+# --- parsing (the contract is prompt-level, so parsing is the guard) ------
+
+
+def test_parses_a_clean_object() -> None:
+    parsed = parse_answer(GOOD)
+    assert parsed == ExpertAnswer(
+        answer="Use SQLite; it is enough here.",
+        confidence="high",
+        caveats="Assumes one writer.",
+    )
+
+
+def test_parses_a_fenced_block_because_models_wrap_json() -> None:
+    assert parse_answer(f"Here you go:\n```json\n{GOOD}\n```") is not None
+
+
+def test_missing_or_bad_fields_are_a_failed_turn() -> None:
+    assert parse_answer('{"answer": "", "confidence": "high"}') is None
+    assert parse_answer('{"answer": "x", "confidence": "certain"}') is None
+    assert parse_answer('{"confidence": "high"}') is None
+    assert parse_answer("no json at all") is None
+    assert parse_answer('["not", "an", "object"]') is None
+
+
+def test_absent_caveats_is_tolerated() -> None:
+    parsed = parse_answer('{"answer": "yes", "confidence": "low"}')
+    assert parsed is not None
+    assert parsed.caveats == ""
+
+
+# --- the desk ------------------------------------------------------------
+
+
+def test_ask_returns_a_validated_answer(store: StateStore) -> None:
+    client = ScriptedClient([GOOD])
+    answer = desk(client, store).ask("SQLite or Postgres?")
+    assert answer.confidence == "high"
+    assert "SQLite" in answer.render()
+
+
+def test_consult_is_stateless_and_uses_the_consultant_role(
+    store: StateStore,
+) -> None:
+    store.append_event(AssistantText(text="secret repo context", model="m"))
+    client = ScriptedClient([GOOD])
+    desk(client, store).ask("Is this API deprecated?")
+
+    call = client.calls[0]
+    # A14: no repo access, no session state — exactly the question, and none
+    # of the log the caller was working from.
+    assert [e.event.text for e in call["events"]] == ["Is this API deprecated?"]
+    assert all("secret repo context" not in e.event.text for e in call["events"])
+    assert call["tools"] is None
+    assert call["model"] == "claude-haiku-4-5"
+
+
+def test_unparseable_reply_is_retried_once_then_raises(store: StateStore) -> None:
+    client = ScriptedClient(["waffle", "more waffle"])
+    with pytest.raises(LLMError, match="no usable answer"):
+        desk(client, store).ask("q")
+    assert len(client.calls) == 2
+    assert "not a single valid JSON object" in client.calls[1]["system"]
+
+
+def test_retry_succeeds_after_a_bad_first_reply(store: StateStore) -> None:
+    client = ScriptedClient(["waffle", GOOD])
+    assert desk(client, store).ask("q").confidence == "high"
+
+
+def test_consults_are_recorded_as_composite_events(store: StateStore) -> None:
+    # A16 wants a consult/delegation eval set; reusing the pair gets consult
+    # evaluation for free.
+    desk(ScriptedClient([GOOD]), store).ask("q")
+    events = [e.event for e in store.read_events()]
+    started = next(e for e in events if isinstance(e, DelegationStarted))
+    completed = next(e for e in events if isinstance(e, DelegationCompleted))
+    assert started.role == "consultant"
+    assert started.brief == "q"
+    assert completed.exit_status == "answered"
+
+
+def test_a_failed_consult_is_recorded_too(store: StateStore) -> None:
+    with pytest.raises(LLMError):
+        desk(ScriptedClient(["waffle", "waffle"]), store).ask("q")
+    completed = [
+        e.event for e in store.read_events() if isinstance(e.event, DelegationCompleted)
+    ]
+    assert completed[0].exit_status == "failed"
+
+
+def test_budget_is_enforced_across_calls(store: StateStore) -> None:
+    client = ScriptedClient([GOOD] * 10)
+    d = desk(client, store, budget=2)
+    d.ask("one")
+    d.ask("two")
+    with pytest.raises(BudgetExhausted, match="budget of 2"):
+        d.ask("three")
+    assert len(client.calls) == 2  # the third never reached the model
+
+
+def test_budget_survives_a_restart(tmp_path: Path) -> None:
+    # Counted from the canonical log, not from memory, so a crash cannot
+    # hand a circling executor a fresh allowance.
+    db = tmp_path / "state.db"
+    with StateStore(db, oplog=OpLog("d", io.StringIO())) as store:
+        desk(ScriptedClient([GOOD]), store, budget=1).ask("one")
+    with (
+        StateStore(db, oplog=OpLog("d", io.StringIO())) as store,
+        pytest.raises(BudgetExhausted),
+    ):
+        desk(ScriptedClient([GOOD]), store, budget=1).ask("two")
+
+
+def test_budget_counts_only_after_the_task_marker(store: StateStore) -> None:
+    d = desk(ScriptedClient([GOOD] * 5), store, budget=1)
+    d.ask("earlier task")
+    marker = store.read_events()[-1].id
+    # A new task starts its allowance afresh from its own marker.
+    assert d.ask("new task", since_event_id=marker).confidence == "high"
+
+
+def test_default_budget_is_small() -> None:
+    assert DEFAULT_BUDGET <= 5  # A14: do not let this become a second subsystem