From dd060566b252ad67f70fa1d30065a986e606394d Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sat, 24 Oct 2026 06:28:53 +0000 Subject: [PATCH] The call record: every model call, for the operator's study (M6.6a) nisse-core records each model call of every agent into a file beside its state store (.calls.db), for the status page (L7, X6): - the request body exactly as sent (never headers), the raw response with its reasoning, the model asked and the one that answered, time, tokens and llama-server's own prompt reuse (timings.cache_n); - its work item, role, run and purpose: Driver.run and nisse-dev's delegation loop open runs, nested; compaction, expert consults, the classifier and the go-ahead reader label their calls; - provenance, worked out in BudgetedClient by comparing the events at each stage: each event sent with its log id and content as sent, which a compaction replaced, which fitting shortened (by how much) and dropped. Content is stored once by hash, compressed. It is diagnostic only: the agent never reads it, a failed write is logged once and recording stops while the call goes on, and the file is deleted when the agent starts, so its layout never migrates; CallReader refuses a layout it doesn't know. While recording, the system prompt ends with one sentence saying so. `call_record: false` in either agent's YAML turns it off. The benchmark records each run beside that run's own state store, and `--keep-runs DIR` keeps both for the page. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk --- nisse-core/src/nisse_core/loop/driver.py | 15 +- nisse-core/src/nisse_core/loop/expert.py | 26 +- nisse-core/src/nisse_core/models/calls.py | 493 ++++++++++++++++++++ nisse-core/src/nisse_core/models/compact.py | 8 +- nisse-core/src/nisse_core/models/context.py | 167 ++++++- nisse-core/src/nisse_core/models/llm.py | 22 + nisse-core/tests/test_calls.py | 376 +++++++++++++++ nisse-dev/src/nisse_dev/bench/runner.py | 38 +- nisse-dev/src/nisse_dev/config.py | 7 + nisse-dev/src/nisse_dev/delegation.py | 16 + nisse-dev/src/nisse_dev/loop/build.py | 5 + nisse-dev/src/nisse_dev/loop/classifier.py | 28 +- nisse-dev/src/nisse_dev/loop/reader.py | 28 +- nisse-dev/tests/test_bench.py | 23 + nisse-dev/tests/test_config.py | 25 + nisse-dev/tests/test_delegation.py | 22 + nisse-dev/tests/test_loop_classifier.py | 24 + nisse-dev/tests/test_loop_labels.py | 16 + nisse-review/src/nisse_review/build.py | 2 + nisse-review/src/nisse_review/config.py | 7 + nisse-review/tests/test_review.py | 40 ++ scripts/benchmark.py | 6 + 22 files changed, 1346 insertions(+), 48 deletions(-) create mode 100644 nisse-core/src/nisse_core/models/calls.py create mode 100644 nisse-core/tests/test_calls.py diff --git a/nisse-core/src/nisse_core/loop/driver.py b/nisse-core/src/nisse_core/loop/driver.py index 374b9c2..9610474 100644 --- a/nisse-core/src/nisse_core/loop/driver.py +++ b/nisse-core/src/nisse_core/loop/driver.py @@ -15,6 +15,7 @@ from typing import Any, Literal, Protocol from nisse_core.log.canonical import AssistantText, Fact, ToolCall, ToolResult from nisse_core.loop.tools import names +from nisse_core.models import calls from nisse_core.models.compact import Compactor from nisse_core.models.llm import LLMClient, LLMError from nisse_core.state.cancel import Cancel, skip_calls @@ -93,7 +94,19 @@ class Driver: self._failure_limit = failure_limit self._cancel = cancel or Cancel() - def run( + def run(self, transcript: Transcript, **options: Any) -> RunOutcome: + """One run, recorded as one (L7): its calls are its turns. See + ``_run`` for the options.""" + with calls.run( + calls.recorder_of(self._client), + work_item=transcript.work_item, + role=transcript.actor, + ) as recorded: + outcome = self._run(transcript, **options) + recorded.end(outcome.kind, outcome.note) + return outcome + + def _run( self, transcript: Transcript, *, diff --git a/nisse-core/src/nisse_core/loop/expert.py b/nisse-core/src/nisse_core/loop/expert.py index 73cac3a..ff85b8b 100644 --- a/nisse-core/src/nisse_core/loop/expert.py +++ b/nisse-core/src/nisse_core/loop/expert.py @@ -32,6 +32,7 @@ from nisse_core.log.canonical import ( StoredEvent, UserMessage, ) +from nisse_core.models import calls from nisse_core.models.llm import LLMClient, LLMError, ModelRefused, RoleMap from nisse_core.state.store import Transcript @@ -156,18 +157,19 @@ class ExpertDesk: # 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, - ) + with calls.purpose("consult", role="expert"): + produced = self._client.complete( + [ + StoredEvent( + id=0, + ts="", + event=UserMessage(text=question, source="ask_expert"), + ) + ], + tools=None, + model=model, + system=system, + ) except ModelRefused as exc: # Not retried: the same question is declined again. return None, f"{exc}. Ask it differently, or work from the repository" diff --git a/nisse-core/src/nisse_core/models/calls.py b/nisse-core/src/nisse_core/models/calls.py new file mode 100644 index 0000000..5086355 --- /dev/null +++ b/nisse-core/src/nisse_core/models/calls.py @@ -0,0 +1,493 @@ +"""The call record (architecture L7, X6): every model call an agent makes, +kept for the operator to study, never for the agent to read. + +Each call is recorded as it was sent and answered: the request body exactly +(never its headers, so no credential), the raw response with its reasoning, +the model asked and the one that answered, time, tokens and the server's own +count of prompt reuse. Beside that, where the request came from: its work +item, role and run, and for each event sent, its log id and its content as +sent, with what compaction replaced, what fitting shortened or dropped. The +status page reads a request against the state store it was made from. + +It is diagnostic only. Nothing in the agent reads it, a write that fails is +logged once and recording stops while the call goes ahead, and the file is +deleted when the agent starts, so its layout never migrates: it carries a +version, and a reader that doesn't know it says so. + +Content is stored once by hash and compressed, so a conversation that grows +by a turn a call costs about one copy of itself. + +Runs and purposes reach the recorder through a context variable: the driver +opens a run around its turns (``run``), and a call made for something else +(a compaction, a consult) says so (``purpose``). +""" + +import contextlib +import hashlib +import json +import sqlite3 +import threading +import uuid +import zlib +from collections.abc import Iterator, Mapping, Sequence +from contextvars import ContextVar +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from nisse_core.log.canonical import StoredEvent, event_type, to_payload +from nisse_core.oplog import OpLog + +LAYOUT_VERSION = 1 + +# Said to the model, at the end of its system prompt, while its calls are +# recorded (L7): it is the operator's record, and the agent should know. +NOTICE = "Your model calls are recorded for the operator's study." + +_SCHEMA = ( + "CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)", + "CREATE TABLE blobs (hash TEXT PRIMARY KEY, data BLOB NOT NULL)", + """ + CREATE TABLE runs ( + id TEXT PRIMARY KEY, + parent TEXT, + work_item TEXT, + role TEXT NOT NULL, + kind TEXT NOT NULL, + started_at TEXT NOT NULL, + ended_at TEXT, + outcome TEXT, + note TEXT + ) + """, + """ + CREATE TABLE calls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + seconds REAL NOT NULL, + run TEXT, + work_item TEXT, + role TEXT, + purpose TEXT, + model TEXT NOT NULL, + answered_by TEXT, + outcome TEXT NOT NULL, + error TEXT, + tokens_in INTEGER, + tokens_out INTEGER, + tokens_cached INTEGER, + cache_n INTEGER, + prompt_n INTEGER, + estimate INTEGER, + factor REAL, + route TEXT, + status INTEGER, + request TEXT, + response TEXT, + sent TEXT NOT NULL, + compacted TEXT NOT NULL, + shortened TEXT NOT NULL, + dropped TEXT NOT NULL, + trim TEXT + ) + """, + "CREATE INDEX calls_by_run ON calls (run, id)", + "CREATE INDEX calls_by_item ON calls (work_item, id)", +) + + +def utc_now() -> str: + return datetime.now(UTC).isoformat() + + +def path_for(state_db: Path) -> Path: + """Where an agent's call record lives: beside its state store, named + after it, so two agents sharing a directory keep two records.""" + return state_db.with_name(f"{state_db.stem}.calls.db") + + +def recorder_for( + state_db: Path, enabled: bool, oplog: OpLog | None = None +) -> "CallRecorder | None": + """An agent's recorder, as its configuration has it (L7): made, and so + the old record deleted, when the agent starts.""" + return CallRecorder(path_for(state_db), oplog=oplog) if enabled else None + + +def with_notice(system: str | None) -> str: + return f"{system}\n\n{NOTICE}" if system else NOTICE + + +# Where a call is made from, for its record. + + +@dataclass(frozen=True) +class Scope: + run: str | None = None + parent: str | None = None + work_item: str | None = None + role: str | None = None + purpose: str | None = None + + +_OUTSIDE = Scope() # no run, no purpose: frozen, so safe to share +_scope: ContextVar[Scope] = ContextVar("nisse_call_scope", default=_OUTSIDE) + + +def current() -> Scope: + return _scope.get() + + +class RunHandle: + """Says how a run ended, for its record.""" + + def __init__(self) -> None: + self.outcome = "" + self.note = "" + + def end(self, outcome: str, note: str = "") -> None: + self.outcome, self.note = outcome, note + + +def recorder_of(client: Any) -> "CallRecorder | None": + """The recorder a client records into, through any wrappers (they pass + attributes on); None if it records nothing.""" + found = getattr(client, "call_recorder", None) + return found if isinstance(found, CallRecorder) else None + + +@contextlib.contextmanager +def run( + recorder: "CallRecorder | None", + *, + work_item: str | None, + role: str, + kind: str = "run", +) -> Iterator[RunHandle]: + """A run: the calls made inside it are its turns. Nested in any run + already open (a delegation inside a driver run).""" + outer = _scope.get() + run_id = uuid.uuid4().hex[:16] + handle = RunHandle() + if recorder is not None: + recorder.run_started(run_id, outer.run, work_item, role, kind) + token = _scope.set( + Scope(run=run_id, parent=outer.run, work_item=work_item, role=role) + ) + try: + yield handle + except BaseException as exc: + handle.end("error", f"{type(exc).__name__}: {exc}"[:500]) + raise + finally: + _scope.reset(token) + if recorder is not None: + recorder.run_ended(run_id, handle.outcome, handle.note) + + +@contextlib.contextmanager +def purpose(name: str, *, role: str | None = None) -> Iterator[None]: + """Calls made inside are for ``name`` (a compaction, a consult), by + ``role`` when it isn't the run's.""" + outer = _scope.get() + token = _scope.set(replace(outer, purpose=name, role=role or outer.role)) + try: + yield + finally: + _scope.reset(token) + + +# What a call did to the events it was given (L6), for its record. + + +@dataclass(frozen=True) +class Provenance: + sent: Sequence[StoredEvent] # what went out, in order, as sent + compacted: Mapping[int, int | None] # event id -> the compaction that replaced it + shortened: Mapping[int, int] # event id -> characters fitting cut + dropped: Sequence[int] # event ids fitting left out + + +def _encode(value: Any) -> tuple[str, bytes]: + text = json.dumps(value, ensure_ascii=False, sort_keys=False) + digest = hashlib.sha256(text.encode()).hexdigest()[:32] + return digest, zlib.compress(text.encode(), 6) + + +class CallRecorder: + """Writes the call record. Never raises into the call it records: the + first failure is logged and recording stops.""" + + def __init__(self, path: str | Path, *, oplog: OpLog | None = None) -> None: + self.path = Path(path) + self._oplog = oplog + self._lock = threading.Lock() + self._conn: sqlite3.Connection | None = None + self.active = False + try: + # Deleted at start (L7): records live as long as the process, so + # the layout never needs migrating. + for suffix in ("", "-wal", "-shm"): + Path(f"{self.path}{suffix}").unlink(missing_ok=True) + conn = sqlite3.connect( + self.path, isolation_level=None, check_same_thread=False + ) + # WAL so the status page can read while the agent writes; no + # fsync per write, since a record lost in a power cut costs nothing. + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.execute("BEGIN") + for statement in _SCHEMA: + conn.execute(statement) + conn.execute( + "INSERT INTO meta (key, value) VALUES ('layout_version', ?)", + (str(LAYOUT_VERSION),), + ) + conn.execute("COMMIT") + self._conn = conn + self.active = True + except (OSError, sqlite3.Error) as exc: + self._failed(exc) + + def _failed(self, exc: BaseException) -> None: + self.active = False + if self._oplog is not None: + self._oplog.emit( + "calls.record_failed", + cause="recording stopped; the calls go on", + path=str(self.path), + error=f"{type(exc).__name__}: {exc}"[:300], + ) + + def _write(self, write: Any) -> None: + if not self.active or self._conn is None: + return + with self._lock: + try: + self._conn.execute("BEGIN") + write(self._conn) + self._conn.execute("COMMIT") + except Exception as exc: # noqa: BLE001 -- never into the call + with contextlib.suppress(sqlite3.Error): + self._conn.execute("ROLLBACK") + self._failed(exc) + + def _blob(self, conn: sqlite3.Connection, value: Any) -> str: + digest, data = _encode(value) + conn.execute( + "INSERT OR IGNORE INTO blobs (hash, data) VALUES (?, ?)", (digest, data) + ) + return digest + + def run_started( + self, + run_id: str, + parent: str | None, + work_item: str | None, + role: str, + kind: str, + ) -> None: + self._write( + lambda conn: conn.execute( + "INSERT INTO runs (id, parent, work_item, role, kind, started_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + (run_id, parent, work_item, role, kind, utc_now()), + ) + ) + + def run_ended(self, run_id: str, outcome: str, note: str) -> None: + self._write( + lambda conn: conn.execute( + "UPDATE runs SET ended_at = ?, outcome = ?, note = ? WHERE id = ?", + (utc_now(), outcome, note, run_id), + ) + ) + + def record( + self, + *, + at: str, + seconds: float, + model: str, + answered_by: str | None, + error: str | None, + usage: Mapping[str, int], + exchange: Mapping[str, Any] | None, + provenance: Provenance, + estimate: int | None, + factor: float, + trim: Mapping[str, Any] | None, + ) -> None: + scope = _scope.get() + + def write(conn: sqlite3.Connection) -> None: + request = response = route = status = None + timings: Mapping[str, Any] = {} + if exchange is not None: + route = exchange.get("route") + status = exchange.get("status") + body = exchange.get("request") + if isinstance(body, Mapping): + request = json.dumps(_split(conn, self, body)) + if "response" in exchange: + raw = exchange["response"] + response = self._blob(conn, raw) + if isinstance(raw, Mapping) and isinstance( + raw.get("timings"), Mapping + ): + timings = raw["timings"] + sent = [ + [ + s.id, + s.actor, + event_type(s.event), + self._blob(conn, to_payload(s.event)), + ] + for s in provenance.sent + ] + conn.execute( + """ + INSERT INTO calls ( + at, seconds, run, work_item, role, purpose, model, answered_by, + outcome, error, tokens_in, tokens_out, tokens_cached, cache_n, + prompt_n, estimate, factor, route, status, request, + response, sent, compacted, shortened, dropped, trim + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, + ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + at, + round(seconds, 3), + scope.run, + scope.work_item, + scope.role, + scope.purpose, + model, + answered_by, + "failed" if error is not None else "answered", + error, + usage.get("input"), + usage.get("output"), + usage.get("cached"), + _int(timings.get("cache_n")), + _int(timings.get("prompt_n")), + estimate, + factor, + route, + status, + request, + response, + json.dumps(sent), + json.dumps({str(k): v for k, v in provenance.compacted.items()}), + json.dumps({str(k): v for k, v in provenance.shortened.items()}), + json.dumps(list(provenance.dropped)), + json.dumps(dict(trim)) if trim is not None else None, + ), + ) + + self._write(write) + + def close(self) -> None: + with self._lock: + self.active = False + if self._conn is not None: + with contextlib.suppress(sqlite3.Error): + self._conn.close() + self._conn = None + + +def _int(value: Any) -> int | None: + return ( + int(value) + if isinstance(value, int | float) and not isinstance(value, bool) + else None + ) + + +def _split( + conn: sqlite3.Connection, recorder: CallRecorder, body: Mapping[str, Any] +) -> dict[str, Any]: + """A request body as hashes: each top-level field once, and the messages + one by one, since each call repeats the conversation before it.""" + fields: list[list[Any]] = [] + for key, value in body.items(): + if key == "messages" and isinstance(value, list): + fields.append([key, [recorder._blob(conn, m) for m in value]]) + else: + fields.append([key, recorder._blob(conn, value)]) + return {"fields": fields} + + +# Reading it back: the status page's side, and the tests'. + + +class UnknownLayout(RuntimeError): + pass + + +class CallReader: + """Opens a call record read-only (``mode=ro``) and never writes to it.""" + + def __init__(self, path: str | Path) -> None: + self._conn = sqlite3.connect(f"file:{Path(path)}?mode=ro", uri=True) + self._conn.row_factory = sqlite3.Row + row = self._conn.execute( + "SELECT value FROM meta WHERE key = 'layout_version'" + ).fetchone() + self.layout_version = int(row["value"]) if row else 0 + if self.layout_version != LAYOUT_VERSION: + self._conn.close() + raise UnknownLayout( + f"call record layout {self.layout_version}; this reader knows " + f"{LAYOUT_VERSION}" + ) + + def blob(self, digest: str) -> Any: + row = self._conn.execute( + "SELECT data FROM blobs WHERE hash = ?", (digest,) + ).fetchone() + if row is None: + raise KeyError(digest) + return json.loads(zlib.decompress(row["data"]).decode()) + + def runs(self) -> list[dict[str, Any]]: + return [ + dict(r) + for r in self._conn.execute("SELECT * FROM runs ORDER BY started_at") + ] + + def calls(self, *, run: str | None = None) -> list[dict[str, Any]]: + if run is None: + rows = self._conn.execute("SELECT * FROM calls ORDER BY id") + else: + rows = self._conn.execute( + "SELECT * FROM calls WHERE run = ? ORDER BY id", (run,) + ) + out = [] + for row in rows: + call = dict(row) + for key in ("sent", "compacted", "shortened", "dropped"): + call[key] = json.loads(call[key]) + call["trim"] = json.loads(call["trim"]) if call["trim"] else None + out.append(call) + return out + + def request(self, call: Mapping[str, Any]) -> dict[str, Any] | None: + """The request body exactly as sent, rebuilt.""" + if not call.get("request"): + return None + body: dict[str, Any] = {} + for key, value in json.loads(call["request"])["fields"]: + body[key] = ( + [self.blob(h) for h in value] + if isinstance(value, list) + else self.blob(value) + ) + return body + + def response(self, call: Mapping[str, Any]) -> Any: + return self.blob(call["response"]) if call.get("response") else None + + def close(self) -> None: + self._conn.close() diff --git a/nisse-core/src/nisse_core/models/compact.py b/nisse-core/src/nisse_core/models/compact.py index c623d2f..6db28f3 100644 --- a/nisse-core/src/nisse_core/models/compact.py +++ b/nisse-core/src/nisse_core/models/compact.py @@ -39,6 +39,7 @@ from nisse_core.log.canonical import ( ToolCall, UserMessage, ) +from nisse_core.models import calls from nisse_core.models.context import _MODEL_SIDE, _turn_starts from nisse_core.models.llm import LLMError from nisse_core.models.render import effective_events @@ -207,9 +208,10 @@ class Compactor: ask = StoredEvent( id=0, ts="", event=UserMessage(text=NOTES_REQUEST, source="compaction") ) - produced = self._client.complete( - [*events, ask], tools, model, system=system, require_tool_call=False - ) + with calls.purpose("compaction"): + produced = self._client.complete( + [*events, ask], tools, model, system=system, require_tool_call=False + ) return "\n".join( e.text.strip() for e in produced if isinstance(e, AssistantText) ).strip() diff --git a/nisse-core/src/nisse_core/models/context.py b/nisse-core/src/nisse_core/models/context.py index 78d29a7..3f11143 100644 --- a/nisse-core/src/nisse_core/models/context.py +++ b/nisse-core/src/nisse_core/models/context.py @@ -39,6 +39,7 @@ from typing import Any from nisse_core.log.canonical import ( AssistantText, CanonicalEvent, + Compaction, DelegationStarted, PlanSubmitted, StoredEvent, @@ -46,6 +47,7 @@ from nisse_core.log.canonical import ( ToolResult, UserMessage, ) +from nisse_core.models import calls from nisse_core.models.llm import DEFAULT_MAX_OUTPUT_TOKENS, LLMError from nisse_core.models.render import effective_events from nisse_core.oplog import OpLog @@ -264,8 +266,12 @@ class BudgetedClient: safety_margin: float = 0.1, enforces_tool_choice: Callable[[str], bool] = lambda model: False, oplog: OpLog | None = None, + call_recorder: calls.CallRecorder | None = None, ) -> None: self._inner = inner + # Every call is recorded here for the operator's study (L7), while it + # is active; the benchmark swaps it per run. + self.call_recorder = call_recorder self._oplog = oplog self._limits_for = limits_for # Whether a model's server holds it to the offered tools when asked @@ -359,10 +365,36 @@ class BudgetedClient: """``require_tool_call``: None asks for a tool call whenever tools are offered (and the model's server enforces it); False lets the model answer in prose, as compaction's notes do.""" + recorder = self.call_recorder + if recorder is not None and not recorder.active: + recorder = None + if recorder is not None: + # The agent is told its calls are recorded, while they are (L7). + system = calls.with_notice(system) budget = self.budget_for(model) factor = self._factors.get(model, 1.0) - events = effective_events(list(events)) - fitted, trim = fit_report(events, system, tools, budget, factor) + given = list(events) + events = effective_events(given) + at = calls.utc_now() + started = time.monotonic() + try: + fitted, trim = fit_report(events, system, tools, budget, factor) + except ContextOverflow as exc: + if recorder is not None: + self._record( + recorder, + at, + started, + model, + given, + events, + events, + None, + factor, + exchange=None, + error=f"ContextOverflow: {exc}", + ) + raise self.last_trim = trim if trim is not None: # Counted so the benchmark can report how often a model's window @@ -386,6 +418,7 @@ class BudgetedClient: ) if require_tool_call is None: require_tool_call = bool(tools) and self._enforces_tool_choice(model) + before = getattr(self._inner, "last_exchange", None) started = time.monotonic() try: produced = self._inner.complete( @@ -399,13 +432,139 @@ class BudgetedClient: # ask_human, a delegation with hand_back. require_tool_call=require_tool_call, ) - except Exception as e: - self._log_call(model, started, error=type(e).__name__) + except BaseException as e: + if isinstance(e, Exception): + self._log_call(model, started, error=type(e).__name__) + if recorder is not None: + self._record( + recorder, + at, + started, + model, + given, + events, + fitted, + trim, + factor, + exchange=self._exchange(before), + error=f"{type(e).__name__}: {e}"[:1000], + tools=tools, + system=system, + ) raise self._log_call(model, started) + if recorder is not None: + self._record( + recorder, + at, + started, + model, + given, + events, + fitted, + trim, + factor, + exchange=self._exchange(before), + error=None, + tools=tools, + system=system, + ) reported = int(self.last_usage.get("input", 0)) if reported > 0: estimated = estimate_tokens(fitted, system, tools) observed = reported / max(estimated, 1) self._factors[model] = max(factor, observed) return produced + + def _exchange(self, before: Any) -> dict[str, Any] | None: + """This call's request and response, if the client set them anew.""" + exchange = getattr(self._inner, "last_exchange", None) + return ( + exchange if isinstance(exchange, dict) and exchange is not before else None + ) + + def _record( + self, + recorder: calls.CallRecorder, + at: str, + started: float, + model: str, + given: list[StoredEvent], + effective: list[StoredEvent], + fitted: list[StoredEvent], + trim: Trim | None, + factor: float, + *, + exchange: dict[str, Any] | None, + error: str | None, + tools: list[dict[str, Any]] | None = None, + system: str | None = None, + ) -> None: + answered = None if error is not None else self.last_model + recorder.record( + at=at, + seconds=time.monotonic() - started, + model=model, + answered_by=answered if answered and answered != model else None, + error=error, + usage={} if error is not None else dict(self.last_usage), + exchange=exchange, + provenance=provenance(given, effective, fitted), + estimate=( + estimate_tokens(fitted, system, tools, factor) + if error is None or exchange is not None + else None + ), + factor=factor, + trim=None + if trim is None + else { + "step": trim.step, + "tokens_before": trim.tokens_before, + "tokens_after": trim.tokens_after, + "limit": trim.limit, + "results_cut": trim.results_cut, + "events_dropped": trim.events_dropped, + }, + ) + + +def provenance( + given: Sequence[StoredEvent], + effective: Sequence[StoredEvent], + fitted: Sequence[StoredEvent], +) -> calls.Provenance: + """What compaction and fitting did to the events a call was given (L6), + for its record: which a compaction replaced, which fitting shortened and + by how much, and which it dropped. Events the harness made up for one + call (id 0) are in what was sent, and in nothing else.""" + compactions = [e for e in effective if isinstance(e.event, Compaction)] + kept = {e.id for e in effective} + compacted: dict[int, int | None] = {} + for stored in given: + if stored.id == 0 or stored.id in kept or isinstance(stored.event, Compaction): + continue + compacted[stored.id] = next( + ( + c.id + for c in compactions + if c.event.replaces[0] <= stored.id <= c.event.replaces[1] + ), + None, + ) + by_id = {e.id: e for e in effective if e.id != 0} + sent_ids = {e.id for e in fitted} + shortened: dict[int, int] = {} + for stored in fitted: + original = by_id.get(stored.id) + if ( + original is not None + and isinstance(stored.event, ToolResult) + and isinstance(original.event, ToolResult) + and stored.event.result != original.event.result + ): + shortened[stored.id] = len(original.event.result) - len(stored.event.result) + dropped = [i for i in by_id if i not in sent_ids] + return calls.Provenance( + sent=list(fitted), compacted=compacted, shortened=shortened, dropped=dropped + ) diff --git a/nisse-core/src/nisse_core/models/llm.py b/nisse-core/src/nisse_core/models/llm.py index 1b7e5f5..9dcaf88 100644 --- a/nisse-core/src/nisse_core/models/llm.py +++ b/nisse-core/src/nisse_core/models/llm.py @@ -188,6 +188,10 @@ class LiteLLMClient: # Set even when the call then fails: time spent waiting out a rate # limit is the proxy's, not the model's (the benchmark subtracts it). self.last_rate_limit_wait_s: float = 0.0 + # The last call's request body and what came back, for the call + # record (L7): a new dict each call, set before it is sent. Never the + # headers, which carry the token. + self.last_exchange: dict[str, Any] | None = None def complete( self, @@ -208,10 +212,12 @@ class LiteLLMClient: # nothing to check against. A caller that knows the model's limits # (the context budget) passes a tighter one. body["max_tokens"] = max_tokens or self._max_tokens + self.last_exchange = exchange = {"route": "chat", "request": body} self.last_rate_limit_wait_s = 0.0 # not the last call's, if this one raises response, self.last_rate_limit_wait_s = _post_with_rate_limit_retry( self._http, "/chat/completions", body, self._sleep ) + _keep_response(exchange, response) if response.status_code != 200: _raise_for_upstream(response) raise LLMError( @@ -321,6 +327,10 @@ class AnthropicPassthroughClient: # Set even when the call then fails: time spent waiting out a rate # limit is the proxy's, not the model's (the benchmark subtracts it). self.last_rate_limit_wait_s: float = 0.0 + # The last call's request body and what came back, for the call + # record (L7): a new dict each call, set before it is sent. Never the + # headers, which carry the token. + self.last_exchange: dict[str, Any] | None = None def complete( self, @@ -341,10 +351,12 @@ class AnthropicPassthroughClient: ) if require_tool_call and tools: body["tool_choice"] = {"type": "any"} # Anthropic's name for it + self.last_exchange = exchange = {"route": "messages", "request": body} self.last_rate_limit_wait_s = 0.0 # not the last call's, if this one raises response, self.last_rate_limit_wait_s = _post_with_rate_limit_retry( self._http, "/v1/messages", body, self._sleep ) + _keep_response(exchange, response) if response.status_code != 200: _raise_for_upstream(response) raise LLMError( @@ -444,6 +456,16 @@ def _post_with_rate_limit_retry( # Model families we have a native renderer adapter for. L9: delegations take # the vendor's passthrough route because a normalising middlebox is lossy # exactly where a long-running delegation lives. +def _keep_response(exchange: dict[str, Any], response: httpx.Response) -> None: + """What came back, for the call record: the JSON as the server sent it + (reasoning included), or its text when it isn't JSON.""" + exchange["status"] = response.status_code + try: + exchange["response"] = response.json() + except ValueError: + exchange["response"] = {"text": response.text[:20000]} + + def _response_cost(response: httpx.Response) -> float: try: return float(response.headers.get("x-litellm-response-cost") or 0.0) diff --git a/nisse-core/tests/test_calls.py b/nisse-core/tests/test_calls.py new file mode 100644 index 0000000..fa1256a --- /dev/null +++ b/nisse-core/tests/test_calls.py @@ -0,0 +1,376 @@ +"""The call record (architecture L7): every model call, as sent and answered, +with where its request came from, for the operator's study.""" + +import io +import json +import sqlite3 +import zlib +from pathlib import Path +from typing import Any + +import httpx +import pytest + +from nisse_core.log.canonical import ( + AssistantText, + CanonicalEvent, + Compaction, + StoredEvent, + ToolCall, + ToolResult, + UserMessage, +) +from nisse_core.loop.driver import Driver +from nisse_core.loop.tools import tool +from nisse_core.models import calls +from nisse_core.models.context import BudgetedClient, ContextOverflow +from nisse_core.models.llm import LiteLLMClient, LLMError +from nisse_core.oplog import OpLog +from nisse_core.state.store import StateStore + +TOKEN = "sk-secret-litellm-token" +FINISH = tool("finish", "finish the work", summary="what was done") + + +def stored(events: list[CanonicalEvent], start: int = 1) -> list[StoredEvent]: + return [ + StoredEvent(id=start + i, ts="t", event=e, work_item="w", actor="driver") + for i, e in enumerate(events) + ] + + +def reply( + content: str = "", + *, + tool_call: str | None = None, + reasoning: str = "", + timings: dict[str, int] | None = None, +) -> dict[str, Any]: + message: dict[str, Any] = {"role": "assistant", "content": content} + if reasoning: + message["reasoning_content"] = reasoning + if tool_call: + message["tool_calls"] = [ + { + "id": "x", + "type": "function", + "function": {"name": tool_call, "arguments": '{"summary": "done"}'}, + } + ] + data: dict[str, Any] = { + "model": "local", + "choices": [{"message": message, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 120, "completion_tokens": 7}, + } + if timings is not None: + data["timings"] = timings + return data + + +class Server: + """A fake LiteLLM: answers each request in turn and keeps what it got.""" + + def __init__(self, *answers: tuple[int, dict[str, Any]]) -> None: + self.answers = list(answers) + self.bodies: list[dict[str, Any]] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.bodies.append(json.loads(request.content)) + status, data = self.answers.pop(0) if self.answers else (200, reply("ok")) + return httpx.Response(status, json=data) + + +def budgeted( + server: Server, recorder: calls.CallRecorder | None, window: int = 32768 +) -> BudgetedClient: + inner = LiteLLMClient( + "https://litellm.test", TOKEN, transport=httpx.MockTransport(server) + ) + return BudgetedClient(inner, lambda m: (window, 1024), call_recorder=recorder) + + +@pytest.fixture +def recorder(tmp_path: Path) -> calls.CallRecorder: + return calls.CallRecorder(tmp_path / "state.calls.db") + + +def read(recorder: calls.CallRecorder) -> calls.CallReader: + return calls.CallReader(recorder.path) + + +# --- what each call was -------------------------------------------------------- + + +def test_a_call_is_recorded_exactly_as_sent_with_its_reasoning( + recorder: calls.CallRecorder, +) -> None: + server = Server( + ( + 200, + reply( + "Finishing.", + tool_call="finish", + reasoning="The tests pass, so the work is done.", + timings={"cache_n": 90, "prompt_n": 30}, + ), + ) + ) + client = budgeted(server, recorder) + client.complete( + stored([UserMessage(text="go", source="t")]), [FINISH], "local", system="sys" + ) + + reader = read(recorder) + [call] = reader.calls() + # The request rebuilt from the record is the body the server got. + assert reader.request(call) == server.bodies[0] + response = reader.response(call) + assert ( + response["choices"][0]["message"]["reasoning_content"] + == "The tests pass, so the work is done." + ) + assert call["outcome"] == "answered" + assert (call["tokens_in"], call["tokens_out"]) == (120, 7) + assert (call["cache_n"], call["prompt_n"]) == (90, 30) + assert call["route"] == "chat" and call["status"] == 200 + + +def test_the_token_is_nowhere_in_the_record(recorder: calls.CallRecorder) -> None: + client = budgeted(Server(), recorder) + client.complete(stored([UserMessage(text="go", source="t")]), None, "local") + recorder.close() + + raw = recorder.path.read_bytes() + conn = sqlite3.connect(recorder.path) + blobs = b"".join( + zlib.decompress(d) for (d,) in conn.execute("SELECT data FROM blobs") + ) + rows = json.dumps(conn.execute("SELECT * FROM calls").fetchall()).encode() + conn.close() + assert TOKEN.encode() not in raw + blobs + rows + + +def test_a_failed_call_is_recorded_with_its_error(recorder: calls.CallRecorder) -> None: + server = Server((400, {"error": {"message": "bad request: no such model"}})) + client = budgeted(server, recorder) + with pytest.raises(LLMError): + client.complete(stored([UserMessage(text="go", source="t")]), None, "local") + + reader = read(recorder) + [call] = reader.calls() + assert call["outcome"] == "failed" + assert "LLMError" in call["error"] + assert call["status"] == 400 + assert reader.request(call) == server.bodies[0] + assert reader.response(call) == {"error": {"message": "bad request: no such model"}} + + +def test_a_request_too_big_to_send_is_recorded_without_one( + recorder: calls.CallRecorder, +) -> None: + server = Server() + client = budgeted(server, recorder, window=4096) + with pytest.raises(ContextOverflow): + client.complete( + stored([UserMessage(text="go", source="t")]), + None, + "local", + system="s" * 20_000, + ) + + [call] = read(recorder).calls() + assert server.bodies == [] + assert call["outcome"] == "failed" and "ContextOverflow" in call["error"] + assert call["request"] is None + + +def test_a_growing_conversation_is_stored_once(recorder: calls.CallRecorder) -> None: + client = budgeted(Server(), recorder) + events: list[CanonicalEvent] = [ + UserMessage(text="message 0 " + "x" * 2000, source="t") + ] + for n in range(1, 6): + client.complete(stored(events), None, "local") + events.append(AssistantText(text=f"reply {n} " + "y" * 2000, model="local")) + events.append(UserMessage(text=f"message {n} " + "z" * 2000, source="t")) + + conn = sqlite3.connect(recorder.path) + stored_chars = sum( + len(zlib.decompress(d)) for (d,) in conn.execute("SELECT data FROM blobs") + ) + conn.close() + # Five calls resend 1 + 3 + 5 + 7 + 9 messages of about 2 kB; stored + # once each, that's 9 messages (as sent, and as events), not 25. + assert stored_chars < 2 * 9 * 2100 + 20_000 + + +# --- where each request came from --------------------------------------------- + + +def test_compaction_shortening_and_dropping_are_recorded( + recorder: calls.CallRecorder, +) -> None: + long = "line\n" * 2000 + events = stored( + [ + UserMessage(text="start", source="t"), # 1 + AssistantText(text="old thinking", model="m"), # 2, compacted + AssistantText(text="more old thinking", model="m"), # 3, compacted + Compaction(summary="notes so far", model="m", replaces=[2, 3]), # 4 + ] + ) + for n in range(8): # turns 5.. : a command and its long output each + events += stored( + [ + ToolCall( + call_id=f"c{n}", tool="run_command", arguments={"command": "make"} + ), + ToolResult(call_id=f"c{n}", result=long, success=True), + ], + start=len(events) + 1, + ) + client = BudgetedClient( + LiteLLMClient( + "https://litellm.test", TOKEN, transport=httpx.MockTransport(Server()) + ), + lambda m: (12_000, 1024), + keep_recent_turns=2, + call_recorder=recorder, + ) + client.complete(events, None, "local") + + [call] = read(recorder).calls() + assert call["compacted"] == {"2": 4, "3": 4} + assert call["shortened"], "older results were cut to fit" + assert all(cut > 0 for cut in call["shortened"].values()) + sent_ids = [entry[0] for entry in call["sent"]] + assert 4 in sent_ids and 2 not in sent_ids + assert set(map(int, call["shortened"])) <= set(sent_ids) + assert not set(call["dropped"]) & set(sent_ids) + + +def test_runs_nest_and_label_their_calls(recorder: calls.CallRecorder) -> None: + client = budgeted(Server(), recorder) + events = stored([UserMessage(text="go", source="t")]) + with calls.run(recorder, work_item="w", role="driver") as outer: + client.complete(events, None, "local") + with calls.purpose("consult", role="expert"): + client.complete(events, None, "local") + with calls.run( + recorder, work_item="w", role="helper", kind="delegation" + ) as inner: + client.complete(events, None, "local") + inner.end("handback", "done") + outer.end("wait", "") + + reader = read(recorder) + runs = {r["kind"]: r for r in reader.runs()} + assert runs["delegation"]["parent"] == runs["run"]["id"] + assert runs["run"]["outcome"] == "wait" and runs["run"]["ended_at"] + assert runs["delegation"]["outcome"] == "handback" + turn, consult, helped = reader.calls() + assert (turn["run"], turn["role"], turn["purpose"]) == ( + runs["run"]["id"], + "driver", + None, + ) + assert (consult["role"], consult["purpose"]) == ("expert", "consult") + assert (helped["run"], helped["role"]) == (runs["delegation"]["id"], "helper") + assert {c["work_item"] for c in (turn, consult, helped)} == {"w"} + + +def test_a_driver_run_is_recorded_with_how_it_ended( + tmp_path: Path, recorder: calls.CallRecorder +) -> None: + store = StateStore(tmp_path / "state.db", oplog=OpLog("t", io.StringIO())) + store.open_work_item("w", title="t", position="RUNNING", cause="test") + store.transcript("w", "operator").append(UserMessage(text="go", source="t")) + client = budgeted(Server((200, reply(tool_call="finish"))), recorder) + + outcome = Driver(client, "local").run( + store.transcript("w", "driver"), + system="sys", + tools=[FINISH], + waiting=("finish",), + workspace=None, + ) + + reader = read(recorder) + [run] = reader.runs() + assert outcome.kind == "wait" + assert (run["work_item"], run["role"], run["outcome"]) == ("w", "driver", "wait") + [call] = reader.calls(run=run["id"]) + # Each event sent is the log's, by id, so the page can read it against + # the state store. + assert [entry[:3] for entry in call["sent"]] == [[1, "operator", "user_message"]] + + +# --- the agent is told, and nothing depends on the record ---------------------- + + +def test_the_agent_is_told_exactly_while_it_is_recorded(tmp_path: Path) -> None: + on, off = Server(), Server() + budgeted(on, calls.CallRecorder(tmp_path / "a.calls.db")).complete( + stored([UserMessage(text="go", source="t")]), None, "local", system="sys" + ) + budgeted(off, None).complete( + stored([UserMessage(text="go", source="t")]), None, "local", system="sys" + ) + assert on.bodies[0]["messages"][0]["content"] == f"sys\n\n{calls.NOTICE}" + assert off.bodies[0]["messages"][0]["content"] == "sys" + + +def test_a_recorder_that_fails_never_fails_the_call(tmp_path: Path) -> None: + out = io.StringIO() + recorder = calls.CallRecorder(tmp_path / "state.calls.db", oplog=OpLog("t", out)) + recorder._conn.close() # the disk went away, say + client = budgeted(Server(), recorder) + + first = client.complete(stored([UserMessage(text="go", source="t")]), None, "local") + second = client.complete( + stored([UserMessage(text="go", source="t")]), None, "local" + ) + + assert [e.text for e in first + second] == ["ok", "ok"] # type: ignore[union-attr] + assert not recorder.active + logged = [json.loads(line) for line in out.getvalue().splitlines()] + assert [e["event"] for e in logged].count("calls.record_failed") == 1 + + +def test_the_record_is_new_each_start(tmp_path: Path) -> None: + path = tmp_path / "state.calls.db" + path.write_bytes(b"an older layout, or not a database at all") + Path(f"{path}-wal").write_bytes(b"stale") + recorder = calls.CallRecorder(path) + budgeted(Server(), recorder).complete( + stored([UserMessage(text="go", source="t")]), None, "local" + ) + assert recorder.active + assert len(read(recorder).calls()) == 1 + + again = calls.CallRecorder(path) + assert again.active and read(again).calls() == [] + + +def test_off_means_no_file(tmp_path: Path) -> None: + assert calls.recorder_for(tmp_path / "state.db", False) is None + assert not list(tmp_path.iterdir()) + recorder = calls.recorder_for(tmp_path / "state.db", True) + assert recorder is not None and recorder.path == tmp_path / "state.calls.db" + + +def test_a_reader_refuses_a_layout_it_does_not_know( + recorder: calls.CallRecorder, +) -> None: + recorder._conn.execute("UPDATE meta SET value = '99' WHERE key = 'layout_version'") + with pytest.raises(calls.UnknownLayout, match="layout 99"): + read(recorder) + + +def test_the_reader_never_writes(recorder: calls.CallRecorder) -> None: + budgeted(Server(), recorder).complete( + stored([UserMessage(text="go", source="t")]), None, "local" + ) + reader = read(recorder) + with pytest.raises(sqlite3.OperationalError, match="readonly"): + reader._conn.execute("DELETE FROM calls") diff --git a/nisse-dev/src/nisse_dev/bench/runner.py b/nisse-dev/src/nisse_dev/bench/runner.py index 89f7144..f76572a 100644 --- a/nisse-dev/src/nisse_dev/bench/runner.py +++ b/nisse-dev/src/nisse_dev/bench/runner.py @@ -1,11 +1,13 @@ """Running one task once with one driver model, and recording it.""" +import contextlib import io import shlex import subprocess import tempfile import time import uuid +from collections.abc import Iterator from dataclasses import asdict, dataclass, field from datetime import UTC, datetime from pathlib import Path @@ -21,6 +23,7 @@ from nisse_core.log.canonical import ( UserMessage, ) from nisse_core.loop.driver import Driver, RunOutcome +from nisse_core.models import calls from nisse_core.models.catalog import ModelFacts from nisse_core.models.compact import Compactor from nisse_core.oplog import OpLog @@ -184,6 +187,37 @@ def _count(events: list[StoredEvent]) -> tuple[int, int]: return failed, invented +@contextlib.contextmanager +def _run_dir(keep: Path | None, workdir: Path | None, name: str) -> Iterator[Path]: + """Where a run's state store and call record go: kept under ``keep`` + for the status page (X6), or a temporary directory.""" + if keep is not None: + root = keep / f"{name}-{uuid.uuid4().hex[:8]}" + root.mkdir(parents=True) + yield root + return + with tempfile.TemporaryDirectory(dir=workdir) as tmp: + yield Path(tmp) + + +@contextlib.contextmanager +def _recording(client: Any, root: Path) -> Iterator[None]: + """The run's calls recorded beside its own state store (L7), as the + agent records its own, so the run measures what the agent would do.""" + recorder = calls.CallRecorder(calls.path_for(root / "state.db")) + had = hasattr(client, "call_recorder") + previous = getattr(client, "call_recorder", None) + client.call_recorder = recorder + try: + yield + finally: + if had: + client.call_recorder = previous + else: + del client.call_recorder + recorder.close() + + def run_task( task: Task, facts: ModelFacts, @@ -194,6 +228,7 @@ def run_task( note: str = "", commit: str = "", workdir: Path | None = None, + keep: Path | None = None, ) -> RunRecord: """Run one task once: seed a workspace, hand the driver an agreed item whose restatement is the task's plan, then judge the result with the @@ -202,8 +237,7 @@ def run_task( It is the loop's own driver, system prompt and tools, so what is measured is what the agent would do, without the harness around it: the milestone is started but no branch is cut, and finishing it ends the run.""" - with tempfile.TemporaryDirectory(dir=workdir) as tmp: - root = Path(tmp) + with _run_dir(keep, workdir, task.name) as root, _recording(client, root): # The workspace is on the build VM, like the agent's own (L13). workspace = Workspace( host, diff --git a/nisse-dev/src/nisse_dev/config.py b/nisse-dev/src/nisse_dev/config.py index d89af65..4d67003 100644 --- a/nisse-dev/src/nisse_dev/config.py +++ b/nisse-dev/src/nisse_dev/config.py @@ -63,6 +63,8 @@ OPTIONAL_KEYS = ( "roles.classifier", "llm.max_output_tokens", "llm.request_timeout_s", + # Record every model call for the operator's study (L7); on unless false. + "call_record", ) # Settings the old loop had (until M2.8), and what to do instead. RETIRED_KEYS = { @@ -444,6 +446,7 @@ class AgentConfig: profiles: Mapping[str, Profile] = field(default_factory=dict) max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS request_timeout_s: float = DEFAULT_REQUEST_TIMEOUT_S + call_record: bool = True # every model call recorded for the operator (L7) tuning: Tuning = field(default_factory=Tuning) vault: VaultConfig | None = None config_file: Path | None = None @@ -521,6 +524,9 @@ class AgentConfig: errors.append("unset or empty — " + ", ".join(missing)) max_output, request_timeout = llm_numbers(values, errors) + record = values.get("call_record", True) + if not isinstance(record, bool): + errors.append("call_record: true or false") if errors: raise ConfigError("cannot start: " + "; ".join(errors)) @@ -548,6 +554,7 @@ class AgentConfig: profiles=profiles, max_output_tokens=max_output, request_timeout_s=request_timeout, + call_record=record, tuning=tuning, vault=vault, config_file=config_file, diff --git a/nisse-dev/src/nisse_dev/delegation.py b/nisse-dev/src/nisse_dev/delegation.py index a579f8c..a2414a3 100644 --- a/nisse-dev/src/nisse_dev/delegation.py +++ b/nisse-dev/src/nisse_dev/delegation.py @@ -38,6 +38,7 @@ from nisse_core.log.canonical import ( ToolCall, ToolResult, ) +from nisse_core.models import calls from nisse_core.models.llm import LLMClient, LLMError, UpstreamUnavailable from nisse_core.state.cancel import Cancel, skip_calls from nisse_core.state.reconcile import Reconciler @@ -333,6 +334,21 @@ class Delegation: def _loop( self, transcript: Transcript, brief: str, branch: str, started_at: float + ) -> DelegationResult: + """The delegation's turns, recorded as a run of its own, inside the + driver's (L7).""" + with calls.run( + calls.recorder_of(self._client), + work_item=transcript.work_item, + role=transcript.actor, + kind="delegation", + ) as recorded: + result = self._turns(transcript, brief, branch, started_at) + recorded.end(result.status, result.note) + return result + + def _turns( + self, transcript: Transcript, brief: str, branch: str, started_at: float ) -> DelegationResult: system = ( f"{SYSTEM_PROMPT.format(reader=self._reader)}\n\n" diff --git a/nisse-dev/src/nisse_dev/loop/build.py b/nisse-dev/src/nisse_dev/loop/build.py index d9b1dd0..dbab37d 100644 --- a/nisse-dev/src/nisse_dev/loop/build.py +++ b/nisse-dev/src/nisse_dev/loop/build.py @@ -14,6 +14,7 @@ from nisse_core.gitea import GiteaClient from nisse_core.knowledge.memory import Memory from nisse_core.knowledge.vault import Vault from nisse_core.loop.driver import Driver +from nisse_core.models import calls from nisse_core.models.catalog import ModelCatalog, select_client from nisse_core.models.compact import Compactor from nisse_core.models.context import BudgetedClient @@ -126,6 +127,9 @@ def make_llm_clients( assert facts.max_input_tokens is not None # require_limits checked return facts.max_input_tokens, facts.max_output_tokens + # Both routes record into one file (L7). + recorder = calls.recorder_for(config.state_db, config.call_record, oplog) + def budgeted(client: object) -> BudgetedClient: # Every request, on either route, is fitted to its model's window. return BudgetedClient( @@ -138,6 +142,7 @@ def make_llm_clients( catalog.facts(m).supports_tool_choice is True ), oplog=oplog, # a cut that fitting has to make is logged + call_recorder=recorder, ) normalized = budgeted( diff --git a/nisse-dev/src/nisse_dev/loop/classifier.py b/nisse-dev/src/nisse_dev/loop/classifier.py index 58601c9..953c822 100644 --- a/nisse-dev/src/nisse_dev/loop/classifier.py +++ b/nisse-dev/src/nisse_dev/loop/classifier.py @@ -18,6 +18,7 @@ from typing import Any from nisse_core.authority.actions import CONTAINED, DANGEROUS, Action, Verdict from nisse_core.log.canonical import StoredEvent, ToolCall, UserMessage +from nisse_core.models.calls import purpose from nisse_core.models.llm import LLMClient, LLMError SAFE = "safe" @@ -120,19 +121,20 @@ class ActionClassifier: shown = f"{model} ({resolved})" if resolved and resolved != model else model prompt = self.input_for(action, scope, workspace) try: - produced = self._client.complete( - [ - StoredEvent( - id=0, - ts="", - event=UserMessage(text=prompt, source="action classifier"), - actor="orchestrator", - ) - ], - tools=[VERDICT], - model=model, - system=SYSTEM, - ) + with purpose("classification", role="classifier"): + produced = self._client.complete( + [ + StoredEvent( + id=0, + ts="", + event=UserMessage(text=prompt, source="action classifier"), + actor="orchestrator", + ) + ], + tools=[VERDICT], + model=model, + system=SYSTEM, + ) except LLMError as exc: return _closed(f"the classifier's call failed: {exc}"), shown except Exception as exc: # noqa: BLE001 -- fails closed, whatever it was diff --git a/nisse-dev/src/nisse_dev/loop/reader.py b/nisse-dev/src/nisse_dev/loop/reader.py index 6298ef2..db705f3 100644 --- a/nisse-dev/src/nisse_dev/loop/reader.py +++ b/nisse-dev/src/nisse_dev/loop/reader.py @@ -11,6 +11,7 @@ The driver's model reads for now; the classifier role takes over in M2.6. from typing import Any from nisse_core.log.canonical import StoredEvent, ToolCall, UserMessage +from nisse_core.models.calls import purpose from nisse_core.models.llm import LLMClient, LLMError @@ -65,19 +66,20 @@ class GoAheadReader: return False prompt = f"The proposal:\n\n{proposal}\n\nThe operator's reply:\n\n{reply}" try: - produced = self._client.complete( - [ - StoredEvent( - id=0, - ts="", - event=UserMessage(text=prompt, source="go-ahead reader"), - actor="orchestrator", - ) - ], - tools=[VERDICT], - model=model or self._model, - system=SYSTEM, - ) + with purpose("go-ahead reading", role="classifier"): + produced = self._client.complete( + [ + StoredEvent( + id=0, + ts="", + event=UserMessage(text=prompt, source="go-ahead reader"), + actor="orchestrator", + ) + ], + tools=[VERDICT], + model=model or self._model, + system=SYSTEM, + ) except LLMError: return False # unread: the driver reads it as an answer instead return any(_yes(event) for event in produced) diff --git a/nisse-dev/tests/test_bench.py b/nisse-dev/tests/test_bench.py index c4a518e..4ef608b 100644 --- a/nisse-dev/tests/test_bench.py +++ b/nisse-dev/tests/test_bench.py @@ -429,3 +429,26 @@ def test_runs_from_before_the_driver_read_as_its_defaults() -> None: old["tuning"]["executor"] = old["tuning"].pop("driver") [row] = summarise([old], current) assert row.tuning == "" + + +def test_a_kept_run_leaves_its_state_and_call_record(tmp_path: Path) -> None: + # As the agent records itself (L7), so the status page can open a + # benchmark run (X6); each run gets its own record, and the client is + # left as it was. + from nisse_core.models.calls import CallReader + + [task] = load_tasks(["write-function"]) + client = Scripted(list(SOLVE_SLUGIFY)) + record = run_task(task, FACTS, client, Tuning(), host=HOST, keep=tmp_path / "runs") + + [kept] = list((tmp_path / "runs").iterdir()) + assert kept.name.startswith("write-function-") + assert (kept / "state.db").exists() + [run] = CallReader(kept / "state.calls.db").runs() + assert (run["work_item"], run["role"], run["outcome"]) == ( + "bench", + "driver", + "wait", + ) + assert record.success + assert not hasattr(client, "call_recorder") diff --git a/nisse-dev/tests/test_config.py b/nisse-dev/tests/test_config.py index 90e9f88..1dd8ab7 100644 --- a/nisse-dev/tests/test_config.py +++ b/nisse-dev/tests/test_config.py @@ -573,3 +573,28 @@ def test_the_vault_is_read() -> None: def test_a_bad_vault_is_refused(vault: dict[str, Any], said: str) -> None: with pytest.raises(ConfigError, match=re.escape(said)): build(vault=vault) + + +def test_the_call_record_is_on_unless_turned_off() -> None: + assert build().call_record is True + assert build(call_record=False).call_record is False + with pytest.raises(ConfigError, match="call_record"): + build(call_record="no") + + +def test_both_routes_record_into_one_file_beside_the_state_store( + tmp_path: Path, +) -> None: + from nisse_core.models.catalog import ModelCatalog + from nisse_dev.loop.build import make_llm_clients + + normalized, passthrough = make_llm_clients( + build(state_db=str(tmp_path / "state.db")), ModelCatalog({}) + ) + assert normalized.call_recorder is passthrough.call_recorder + assert normalized.call_recorder.path == tmp_path / "state.calls.db" + off, _ = make_llm_clients( + build(state_db=str(tmp_path / "other.db"), call_record=False), ModelCatalog({}) + ) + assert off.call_recorder is None + assert not (tmp_path / "other.calls.db").exists() diff --git a/nisse-dev/tests/test_delegation.py b/nisse-dev/tests/test_delegation.py index 6f5a5ce..d41fe0f 100644 --- a/nisse-dev/tests/test_delegation.py +++ b/nisse-dev/tests/test_delegation.py @@ -702,3 +702,25 @@ def test_a_bad_request_is_not_given_the_benefit_of_the_doubt( result = build(BadRequestClient([]), store, ws, repo).run(brief="b", branch="d") assert result.status == "failed" assert not result.interrupted + + +def test_a_delegation_is_recorded_as_a_run_of_its_own( + store: StateStore, ws: Workspace, repo: GitRepo +) -> None: + # Its calls are the helper's turns (L7), inside the run that delegated. + from nisse_core.models import calls + + seen: list[calls.Scope] = [] + + class Watching(ScriptedClient): + def complete(self, *args: Any, **kwargs: Any) -> list[CanonicalEvent]: + seen.append(calls.current()) + return super().complete(*args, **kwargs) + + client = Watching([[hand_back(succeeded=False, changed="nothing")]]) + with calls.run(None, work_item=ITEM, role="driver"): + outer = calls.current().run + build(client, store, ws, repo).run(brief="look into it", branch="d1") + [scope] = seen + assert (scope.role, scope.work_item, scope.parent) == ("helper", ITEM, outer) + assert scope.run not in (None, outer) diff --git a/nisse-dev/tests/test_loop_classifier.py b/nisse-dev/tests/test_loop_classifier.py index 2873b12..ff94a92 100644 --- a/nisse-dev/tests/test_loop_classifier.py +++ b/nisse-dev/tests/test_loop_classifier.py @@ -346,3 +346,27 @@ def test_the_export_script_reads_a_store(tmp_path: Path) -> None: lines = [json.loads(line) for line in run.stdout.splitlines()] assert [line["verdict"] for line in lines] == ["needs_discussion", "safe"] assert lines[0]["input"]["action"]["arguments"] == {"command": "x"} + + +def test_the_classifiers_calls_are_recorded_as_its_own() -> None: + # In the call record (L7), a verdict is the classifier's call, not a + # turn of the run that asked for it. + from nisse_core.models import calls + + seen: list[calls.Scope] = [] + + class Watching(Judge): + def complete(self, events, tools, model, system=None) -> list[Any]: + seen.append(calls.current()) + return super().complete(events, tools, model, system) + + classifier = ActionClassifier(Watching(verdict("contained")), "local") # type: ignore[arg-type] + with calls.run(None, work_item="w", role="driver"): + classifier.judge(Action("run_command", {"command": "make"}), SCOPE, "/w") + [scope] = seen + assert (scope.role, scope.purpose, scope.work_item) == ( + "classifier", + "classification", + "w", + ) + assert scope.run is not None diff --git a/nisse-dev/tests/test_loop_labels.py b/nisse-dev/tests/test_loop_labels.py index eb03476..fc4f984 100644 --- a/nisse-dev/tests/test_loop_labels.py +++ b/nisse-dev/tests/test_loop_labels.py @@ -316,3 +316,19 @@ def test_a_restatement_that_asks_is_sent_back(tmp_path: Path) -> None: if hasattr(e.event, "result") ] assert any("a restatement asks no questions" in r for r in results) + + +def test_the_go_ahead_readers_calls_are_recorded_as_the_classifiers() -> None: + from nisse_core.models import calls + + seen: list[calls.Scope] = [] + + class Watching(_Model): + def complete(self, *args: Any, **kwargs: Any) -> Any: + seen.append(calls.current()) + return super().complete(*args, **kwargs) + + yes = [ToolCall(call_id="v", tool="verdict", arguments={"answer": "yes"})] + GoAheadReader(Watching(yes), "m").agrees("add count_words", "ok") # type: ignore[arg-type] + [scope] = seen + assert (scope.role, scope.purpose) == ("classifier", "go-ahead reading") diff --git a/nisse-review/src/nisse_review/build.py b/nisse-review/src/nisse_review/build.py index d158f8f..a372c8a 100644 --- a/nisse-review/src/nisse_review/build.py +++ b/nisse-review/src/nisse_review/build.py @@ -9,6 +9,7 @@ from nisse_core.knowledge.memory import Memory from nisse_core.knowledge.vault import Vault from nisse_core.loop.driver import Driver from nisse_core.loop.expert import ExpertDesk +from nisse_core.models import calls from nisse_core.models.catalog import ModelCatalog from nisse_core.models.compact import Compactor from nisse_core.models.context import BudgetedClient @@ -94,6 +95,7 @@ def llm_client(config: ReviewConfig, oplog: OpLog | None = None) -> BudgetedClie safety_margin=tuning.safety_margin, enforces_tool_choice=lambda m: catalog.facts(m).supports_tool_choice is True, oplog=oplog, # a cut that fitting has to make is logged + call_recorder=calls.recorder_for(config.state_db, config.call_record, oplog), ) diff --git a/nisse-review/src/nisse_review/config.py b/nisse-review/src/nisse_review/config.py index b081676..4553d98 100644 --- a/nisse-review/src/nisse_review/config.py +++ b/nisse-review/src/nisse_review/config.py @@ -47,6 +47,8 @@ OPTIONAL_KEYS = ( # Patterns of the repositories it reviews ("dev/nisse", "dev/*", "*/*"), # public ones included; unset, every one it was given (R1, ``access``). "repositories", + # Record every model call for the operator's study (L7); on unless false. + "call_record", ) @@ -108,6 +110,7 @@ class ReviewConfig: gitea_operators: frozenset[str] # whose comments are the operator's (R4) max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS request_timeout_s: float = DEFAULT_REQUEST_TIMEOUT_S + call_record: bool = True # every model call recorded for the operator (L7) tuning: Tuning = field(default_factory=Tuning) vault: VaultConfig | None = None repositories: frozenset[str] | None = None # None: all it can read @@ -158,6 +161,9 @@ class ReviewConfig: owner, slash, name = pattern.partition("/") if not (owner and slash and name) or "/" in name: errors.append(f"repositories: {pattern!r} is not owner/name") + record = values.get("call_record", True) + if not isinstance(record, bool): + errors.append("call_record: true or false") if errors: raise ConfigError("cannot start: " + "; ".join(errors)) reviewer = str(values["roles.reviewer"]) @@ -173,6 +179,7 @@ class ReviewConfig: gitea_operators=people(values["gitea.operators"]), max_output_tokens=max_output, request_timeout_s=timeout, + call_record=record, tuning=tuning, vault=vault, repositories=patterns, diff --git a/nisse-review/tests/test_review.py b/nisse-review/tests/test_review.py index 48dfa66..a9dc331 100644 --- a/nisse-review/tests/test_review.py +++ b/nisse-review/tests/test_review.py @@ -645,3 +645,43 @@ def test_code_git_cannot_give_fails_the_run(tmp_path: Path) -> None: w["scheduler"].round() assert (item(w).position, pr_of(item(w)).attempts) == (READY, 1) assert w["gitea"].posted == [] + + +def test_the_review_agent_records_its_calls_unless_told_not_to( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from nisse_core.config import ConfigError + from nisse_core.models.catalog import ModelCatalog + from nisse_review import build as review_build + from nisse_review.config import ReviewConfig + + tree = { + "state_db": str(tmp_path / "s.db"), + "llm": {"base_url": "http://llm", "token": "k"}, + "roles": {"reviewer": "local", "expert": "frontier"}, + "gitea": {"url": "https://g", "token": "t", "operators": ["don"]}, + } + limits = {"max_input_tokens": 32768, "max_output_tokens": 4096} + catalog = ModelCatalog.from_payload( + { + "data": [ + { + "model_name": m, + "litellm_params": {"model": f"x/{m}"}, + "model_info": limits, + } + for m in ("local", "frontier") + ] + } + ) + monkeypatch.setattr(review_build.ModelCatalog, "discover", lambda *a, **k: catalog) + + on = review_build.llm_client(ReviewConfig.from_tree(tree, {})) + assert on.call_recorder is not None + assert on.call_recorder.path == tmp_path / "s.calls.db" + off = review_build.llm_client( + ReviewConfig.from_tree({**tree, "call_record": False}, {}) + ) + assert off.call_recorder is None + with pytest.raises(ConfigError, match="call_record"): + ReviewConfig.from_tree({**tree, "call_record": "yes"}, {}) diff --git a/scripts/benchmark.py b/scripts/benchmark.py index 0e3db67..0433cb0 100644 --- a/scripts/benchmark.py +++ b/scripts/benchmark.py @@ -66,6 +66,7 @@ def run(args: argparse.Namespace) -> int: host=host, note=args.note, commit=commit, + keep=args.keep_runs, ) append(out, record) verdict = "pass" if record.success else f"FAIL ({record.outcome})" @@ -113,6 +114,11 @@ def main() -> int: "--note", default="", help='how the model was served, e.g. "vLLM 0.29, AWQ"' ) run_p.add_argument("--config", help="agent config (default: NISSE_CONFIG)") + run_p.add_argument( + "--keep-runs", + type=Path, + help="keep each run's state store and call record here, for the status page", + ) run_p.set_defaults(handler=run) report_p = sub.add_parser("report", help="compare stored results")