From 2b7651fb37e1b846e696d359e11b818d8c516c3e Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Tue, 8 Sep 2026 19:48:13 +0000 Subject: [PATCH] =?UTF-8?q?Add=20ask=5Fexpert=20and=20eval=20capture=20(A1?= =?UTF-8?q?4/A16)=20=E2=80=94=20step=207c?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- nisse-dev/src/nisse_dev/evals.py | 88 +++++++++++ nisse-dev/src/nisse_dev/executor.py | 35 ++++- nisse-dev/src/nisse_dev/expert.py | 214 ++++++++++++++++++++++++++ nisse-dev/src/nisse_dev/state.py | 43 +++++- nisse-dev/tests/test_canonical.py | 7 +- nisse-dev/tests/test_evals.py | 129 ++++++++++++++++ nisse-dev/tests/test_executor.py | 92 +++++++++++ nisse-dev/tests/test_expert.py | 186 ++++++++++++++++++++++ tests/integration/test_expert_live.py | 60 ++++++++ 9 files changed, 846 insertions(+), 8 deletions(-) create mode 100644 nisse-dev/src/nisse_dev/evals.py create mode 100644 nisse-dev/src/nisse_dev/expert.py create mode 100644 nisse-dev/tests/test_evals.py create mode 100644 nisse-dev/tests/test_expert.py create mode 100644 tests/integration/test_expert_live.py diff --git a/nisse-dev/src/nisse_dev/evals.py b/nisse-dev/src/nisse_dev/evals.py new file mode 100644 index 0000000..75ae58b --- /dev/null +++ b/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 diff --git a/nisse-dev/src/nisse_dev/executor.py b/nisse-dev/src/nisse_dev/executor.py index 77cd769..988b75b 100644 --- a/nisse-dev/src/nisse_dev/executor.py +++ b/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, diff --git a/nisse-dev/src/nisse_dev/expert.py b/nisse-dev/src/nisse_dev/expert.py new file mode 100644 index 0000000..181a9e1 --- /dev/null +++ b/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 "", + ) diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py index 978c0d5..2175835 100644 --- a/nisse-dev/src/nisse_dev/state.py +++ b/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() diff --git a/nisse-dev/tests/test_canonical.py b/nisse-dev/tests/test_canonical.py index 1de85e9..da45109 100644 --- a/nisse-dev/tests/test_canonical.py +++ b/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") diff --git a/nisse-dev/tests/test_evals.py b/nisse-dev/tests/test_evals.py new file mode 100644 index 0000000..3336990 --- /dev/null +++ b/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 diff --git a/nisse-dev/tests/test_executor.py b/nisse-dev/tests/test_executor.py index 5d91924..a19ef20 100644 --- a/nisse-dev/tests/test_executor.py +++ b/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 diff --git a/nisse-dev/tests/test_expert.py b/nisse-dev/tests/test_expert.py new file mode 100644 index 0000000..0b5a23b --- /dev/null +++ b/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 diff --git a/tests/integration/test_expert_live.py b/tests/integration/test_expert_live.py new file mode 100644 index 0000000..8197efd --- /dev/null +++ b/tests/integration/test_expert_live.py @@ -0,0 +1,60 @@ +"""A real consult over the normalized route (A14). + +The point is the prompt-level contract: a live model must produce something +that parses against our local schema, since no API-level enforcement exists. +""" + +import io +import os +from pathlib import Path + +import pytest + +from nisse_dev.envfile import load_env_file +from nisse_dev.evals import collect +from nisse_dev.expert import BudgetExhausted, ExpertDesk +from nisse_dev.llm import LiteLLMClient, RoleMap +from nisse_dev.oplog import OpLog +from nisse_dev.state import StateStore + +load_env_file(Path(__file__).resolve().parents[2] / ".env") + +SENTINEL = "NISSE_LIVE_PLANNER_MODEL" + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not os.environ.get(SENTINEL), + reason=f"no live frontier route configured ({SENTINEL} unset)", + ), +] + + +def desk(tmp_path: Path, budget: int = 2) -> tuple[ExpertDesk, StateStore]: + model = os.environ[SENTINEL] + roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model) + store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())) + return ExpertDesk(LiteLLMClient.from_env(), store, roles, budget=budget), store + + +def test_live_consult_parses_against_the_local_schema(tmp_path: Path) -> None: + expert, store = desk(tmp_path) + + answer = expert.ask( + "For a single-process agent storing a few thousand rows of task state, " + "is SQLite sufficient, or is PostgreSQL warranted? Answer briefly." + ) + + assert answer.confidence in {"high", "medium", "low"} + assert len(answer.answer) > 20 + # The consult is captured for A16 without any extra bookkeeping. + [record] = collect(store) + assert record.role == "consultant" + assert record.exit_status == "answered" + + +def test_live_budget_stops_a_circling_caller(tmp_path: Path) -> None: + expert, _ = desk(tmp_path, budget=1) + expert.ask("Name one advantage of WAL mode in SQLite. Be brief.") + with pytest.raises(BudgetExhausted): + expert.ask("And another?")