From 5d8d246f6c9ccccdc27a07798c9fdce082de7111 Mon Sep 17 00:00:00 2001 From: "David J. Kim" Date: Thu, 23 Jul 2026 15:21:18 -0400 Subject: [PATCH 1/2] feat: console score-on-arrival engine (serve --score-production / --rebuild-scores) A supervisor-pattern worker inside hotato serve polls the production evidence db (mode=ro + query_only + write-denying authorizer) for sessions reaching COMPLETE/QUIESCENT and scores each one -- one at a time, keyset queries only -- with the deterministic scan scorer over the session's recorded two-channel audio, writing durable records to a versioned console.sqlite3 sidecar beside the evidence db: per-dimension observations (never blended), ranked candidate moments, one measured failure-reason sentence, and timing derived purely from evidence event timestamps (per-hop latency keeps the reporting event's declared authority; turn spans and end-to-end are labeled derived:event_timestamps). Refusal is first-class: absent/unavailable/mono/unreadable audio is a NOT_SCORABLE record with its reason; a scorer crash on one session becomes an ERROR record and the loop continues; a persist failure surfaces through a minimal ERROR write path plus a counter and is retried -- a score is claimed only after its sidecar write commits. The sidecar is derived data, never authority: --rebuild-scores regenerates it entirely from the evidence db and exits, and the same evidence db always rebuilds to a byte-identical canonical dump (the one wall-clock column, created_at, is excluded from the comparison). Bind, auth, and the server's read-only route surface are unchanged. Claude-Session: https://claude.ai/code/session_015njkxtoAKioqTPT2k14xFr --- CHANGELOG.md | 19 +- docs/WORKSPACE.md | 43 ++ llms-full.txt | 43 ++ src/hotato/cli.py | 38 ++ src/hotato/console_store.py | 421 +++++++++++++++++ src/hotato/console_worker.py | 849 +++++++++++++++++++++++++++++++++++ src/hotato/serve/app.py | 31 +- tests/test_console_score.py | 447 ++++++++++++++++++ 8 files changed, 1889 insertions(+), 2 deletions(-) create mode 100644 src/hotato/console_store.py create mode 100644 src/hotato/console_worker.py create mode 100644 tests/test_console_score.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9552841..24302a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,7 +36,24 @@ Every entry reports millisecond measurement error and a confusion matrix. See `d reported. ### Added -- **`--junit` on `hotato suite run` and `hotato prove`.** Both commands also +- **`hotato serve --score-production`: score-on-arrival for the production + evidence plane.** A background worker in the serve process polls the + evidence database (read-only, `mode=ro`) for sessions reaching + `COMPLETE`/`QUIESCENT` and scores each one -- one at a time -- with the + deterministic scorer over the session's recorded two-channel audio, writing + durable records to a `console.sqlite3` sidecar beside the evidence + database: per-dimension observations (never blended), ranked candidate + moments, one measured failure-reason sentence, and timing derived from + evidence event timestamps (per-hop latency keeps the reporting event's + declared authority; turn spans and end-to-end are labeled + `derived:event_timestamps`). Refusal is a first-class `NOT_SCORABLE` record + with its reason; a scorer crash or persist failure on one session becomes a + visible `ERROR` record -- never a silent skip, never a claimed score before + the sidecar write commits -- and the worker continues to the next session. + The sidecar is derived data with a versioned schema: `--rebuild-scores` + regenerates it entirely from the evidence database and exits, and the same + evidence database always rebuilds to identical content. Bind, auth, and the + server's read-only route surface are unchanged. Both commands also write a JUnit XML report any standard CI test widget ingests, mirroring each command's existing grouping: one `` per dimension (suite run) or per evidence lane (prove), one `` per test or lane. A diff --git a/docs/WORKSPACE.md b/docs/WORKSPACE.md index 79c692f..c7914ef 100644 --- a/docs/WORKSPACE.md +++ b/docs/WORKSPACE.md @@ -37,6 +37,8 @@ redirects to strip the token from the address bar (see [Auth](#auth)). | `--port` | `8321` | listen port | | `--registry` | `~/.hotato/fleet` | registry home directory | | `--production-db` | none | read session manifests and alerts from this separate Hotato production SQLite database in `/health` (mode=ro; see below) | +| `--score-production` | off | with `--production-db`: score completed sessions in the background into a `console.sqlite3` sidecar beside the evidence database (see below) | +| `--rebuild-scores` | off | with `--production-db`: deterministically regenerate the entire `console.sqlite3` sidecar from the evidence database, then exit | | `--token` | none | supply the bearer token yourself | | `--token-file` | none | read the bearer token from a file (first line) | @@ -81,6 +83,47 @@ carry a fleet workspace id, so the UI states `workspace_scope = not_encoded_by_production_schema` instead of silently assigning those sessions to the workspace being served. +### Score-on-arrival (`--score-production`) + +```bash +hotato serve --workspace default \ + --production-db .hotato/production.sqlite3 --score-production +``` + +A background worker in the same process polls the evidence database (same +`mode=ro` read-only discipline) for sessions that reached +`COMPLETE`/`QUIESCENT` and scores each one with the deterministic scorer over +the session's recorded two-channel audio (the path named by the +`media.asset.available` event's `data.path`). Bind and auth are unchanged; +the server gains no new routes or write endpoints from this flag. + +Each session becomes one durable record in `console.sqlite3` beside the +evidence database: + +- **`SCORED`** -- per-dimension observations (candidate counts and worst + measured magnitude per scan kind, never blended), the ranked candidate + moments, and one plain-English failure-reason sentence built only from + measured numbers; +- **`NOT_SCORABLE`** -- the scorer's refusal with its reason (audio lane + unavailable, no recorded path, a one-channel or unreadable recording); +- **`ERROR`** -- a scorer crash or persist failure on that session, with its + reason; the worker records it and continues to the next session. + +Every record carries the scorer version and a config hash, and every timing +figure derives from evidence event timestamps: per-hop latency rows keep the +reporting event's declared `authority`, turn spans and the end-to-end figure +are labeled `derived:event_timestamps`, and reported turn fields +(`yield_latency_ms`, `overlap_ms`, `duration_ms`) stay in a separate +`reported` block. Sessions are scored one at a time and a record is claimed +only after its sidecar write commits. + +The sidecar is derived data -- the evidence database stays the only +authority. `--rebuild-scores` regenerates the whole sidecar from the evidence +database and exits; the same evidence database always rebuilds to identical +content (the one wall-clock column, `created_at`, is excluded from the +canonical comparison). A sidecar written by a different schema version is +refused with that rebuild instruction. + ## Auth Every request is authenticated against one shared **bearer token**: diff --git a/llms-full.txt b/llms-full.txt index 5d1bdcc..f2808f6 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -4628,6 +4628,8 @@ redirects to strip the token from the address bar (see [Auth](#auth)). | `--port` | `8321` | listen port | | `--registry` | `~/.hotato/fleet` | registry home directory | | `--production-db` | none | read session manifests and alerts from this separate Hotato production SQLite database in `/health` (mode=ro; see below) | +| `--score-production` | off | with `--production-db`: score completed sessions in the background into a `console.sqlite3` sidecar beside the evidence database (see below) | +| `--rebuild-scores` | off | with `--production-db`: deterministically regenerate the entire `console.sqlite3` sidecar from the evidence database, then exit | | `--token` | none | supply the bearer token yourself | | `--token-file` | none | read the bearer token from a file (first line) | @@ -4672,6 +4674,47 @@ carry a fleet workspace id, so the UI states `workspace_scope = not_encoded_by_production_schema` instead of silently assigning those sessions to the workspace being served. +### Score-on-arrival (`--score-production`) + +```bash +hotato serve --workspace default \ + --production-db .hotato/production.sqlite3 --score-production +``` + +A background worker in the same process polls the evidence database (same +`mode=ro` read-only discipline) for sessions that reached +`COMPLETE`/`QUIESCENT` and scores each one with the deterministic scorer over +the session's recorded two-channel audio (the path named by the +`media.asset.available` event's `data.path`). Bind and auth are unchanged; +the server gains no new routes or write endpoints from this flag. + +Each session becomes one durable record in `console.sqlite3` beside the +evidence database: + +- **`SCORED`** -- per-dimension observations (candidate counts and worst + measured magnitude per scan kind, never blended), the ranked candidate + moments, and one plain-English failure-reason sentence built only from + measured numbers; +- **`NOT_SCORABLE`** -- the scorer's refusal with its reason (audio lane + unavailable, no recorded path, a one-channel or unreadable recording); +- **`ERROR`** -- a scorer crash or persist failure on that session, with its + reason; the worker records it and continues to the next session. + +Every record carries the scorer version and a config hash, and every timing +figure derives from evidence event timestamps: per-hop latency rows keep the +reporting event's declared `authority`, turn spans and the end-to-end figure +are labeled `derived:event_timestamps`, and reported turn fields +(`yield_latency_ms`, `overlap_ms`, `duration_ms`) stay in a separate +`reported` block. Sessions are scored one at a time and a record is claimed +only after its sidecar write commits. + +The sidecar is derived data -- the evidence database stays the only +authority. `--rebuild-scores` regenerates the whole sidecar from the evidence +database and exits; the same evidence database always rebuilds to identical +content (the one wall-clock column, `created_at`, is excluded from the +canonical comparison). A sidecar written by a different schema version is +refused with that rebuild instruction. + ## Auth Every request is authenticated against one shared **bearer token**: diff --git a/src/hotato/cli.py b/src/hotato/cli.py index ad6a8b8..b31a920 100644 --- a/src/hotato/cli.py +++ b/src/hotato/cli.py @@ -5268,12 +5268,29 @@ def _cmd_start(args) -> int: def _cmd_serve(args) -> int: + if args.rebuild_scores: + # One-shot: deterministically regenerate the console sidecar from the + # evidence database and exit; no server is started. + if not args.production_db: + raise ValueError( + "--rebuild-scores needs --production-db (the evidence database " + "the sidecar is derived from)" + ) + from . import console_worker as _console + + result = _console.run_rebuild(args.production_db) + print("hotato serve --rebuild-scores: regenerated %s" % result["sidecar"]) + print(" scored: %d not scorable: %d errors: %d" + % (result["scored"], result["not_scorable"], result["errors"])) + return 0 + from . import serve as _serve # lazy: the workspace server + its stdlib deps return _serve.run_serve( workspace=args.workspace, host=args.host, port=args.port, registry=args.registry, token=args.token, token_file=args.token_file, open_browser=not args.no_open, production_db=args.production_db, + score_production=args.score_production, ) @@ -10301,6 +10318,27 @@ def _fleet_parser(parent, name, dotted, help_text): "database into /health; opened read-only without payload access" ), ) + srv.add_argument( + "--score-production", + action="store_true", + help=( + "with --production-db: run the score-on-arrival worker alongside " + "the server -- completed sessions are scored one at a time with " + "the deterministic scorer and recorded (SCORED / NOT_SCORABLE " + "with reason / ERROR) in a console.sqlite3 sidecar beside the " + "evidence database" + ), + ) + srv.add_argument( + "--rebuild-scores", + action="store_true", + help=( + "with --production-db: deterministically regenerate the entire " + "console.sqlite3 sidecar from the evidence database, then exit " + "(the sidecar is derived data; the same evidence database always " + "rebuilds to identical content)" + ), + ) srv.add_argument("--token", default=None, metavar="TOKEN", help="bearer token (default: reuse the stored one, else " "generate + store 0600 under the workspace state dir)") diff --git a/src/hotato/console_store.py b/src/hotato/console_store.py new file mode 100644 index 0000000..9e2cc4e --- /dev/null +++ b/src/hotato/console_store.py @@ -0,0 +1,421 @@ +"""The console sidecar store: derived per-session score records, never authority. + +``console.sqlite3`` sits BESIDE the production evidence database and holds only +DERIVED data: per-session score records the console worker computed from the +evidence plane with the existing deterministic scorer. Its invariants: + +* the evidence database stays the single authority; every row here is + rebuildable from it (``hotato serve --production-db DB --rebuild-scores``), + so a schema change ships as a rebuild, never a migration; +* the schema is versioned (``console_schema_version`` metadata row); a store + written by a different schema version is refused with a rebuild instruction + rather than half-migrated in place; +* a write commits durably (WAL + ``synchronous=FULL``) before the caller may + report success -- the worker treats any exception from :meth:`upsert_score` + as a persist failure to surface, never a success; +* every scored value derives from evidence timestamps; the only wall-clock + column is ``created_at``, and :meth:`canonical_dump` excludes it, so the + same evidence database always produces a byte-identical canonical dump. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import threading +import time +from typing import Any, Dict, List, Optional, Tuple + +from .errors import safe_json_dumps +from .production import _prepare_private_sqlite_file + +__all__ = [ + "SCHEMA_VERSION", + "SCORE_STATES", + "ConsoleStoreError", + "ConsoleStore", +] + +SCHEMA_VERSION = "1" +SCORE_STATES = ("SCORED", "NOT_SCORABLE", "ERROR") + +_RECORD_FIELDS = ( + "subject", + "state", + "reason", + "session_state", + "evidence_sha256", + "event_count", + "scorer_version", + "config_sha256", + "config", + "dimensions", + "candidates", + "timing", + "audio", + "hops", +) + + +class ConsoleStoreError(RuntimeError): + """The console sidecar cannot be opened or written safely.""" + + +class ConsoleStore: + """A single-process durable sidecar for derived console score records.""" + + def __init__(self, path: str, *, clock=time.time) -> None: + self.path = os.path.abspath(path) + self.clock = clock + self._lock = threading.RLock() + guard, expected_identity = _prepare_private_sqlite_file(self.path) + try: + self.db = sqlite3.connect( + self.path, check_same_thread=False, timeout=30, isolation_level=None + ) + observed = os.stat(self.path, follow_symlinks=False) + if (observed.st_dev, observed.st_ino) != expected_identity: + self.db.close() + raise ConsoleStoreError( + "console sidecar changed while SQLite was opening it" + ) + finally: + os.close(guard) + self.db.row_factory = sqlite3.Row + self.db.execute("PRAGMA journal_mode=WAL") + self.db.execute("PRAGMA synchronous=FULL") + self.db.execute("PRAGMA foreign_keys=ON") + self.db.execute("PRAGMA busy_timeout=30000") + self._schema() + self._verify_schema_version() + + def _schema(self) -> None: + with self._lock: + self.db.executescript( + """ + BEGIN IMMEDIATE; + CREATE TABLE IF NOT EXISTS metadata( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + INSERT OR IGNORE INTO metadata(key,value) + VALUES('console_schema_version','1'); + CREATE TABLE IF NOT EXISTS scores( + subject TEXT PRIMARY KEY, + state TEXT NOT NULL + CHECK(state IN ('SCORED','NOT_SCORABLE','ERROR')), + reason TEXT, + session_state TEXT NOT NULL, + evidence_sha256 TEXT NOT NULL, + event_count INTEGER NOT NULL, + scorer_version TEXT NOT NULL, + config_sha256 TEXT NOT NULL, + config_json TEXT NOT NULL, + dimensions_json TEXT NOT NULL, + candidates_json TEXT NOT NULL, + timing_json TEXT NOT NULL, + audio_json TEXT, + created_at REAL NOT NULL + ); + CREATE TABLE IF NOT EXISTS latency_hops( + subject TEXT NOT NULL, + seq INTEGER NOT NULL, + kind TEXT NOT NULL, + at TEXT NOT NULL, + latency_ms REAL, + authority TEXT NOT NULL, + source TEXT NOT NULL, + event_id TEXT NOT NULL, + PRIMARY KEY(subject,seq) + ); + CREATE INDEX IF NOT EXISTS latency_hops_kind_at + ON latency_hops(kind,at); + COMMIT; + """ + ) + + def _verify_schema_version(self) -> None: + row = self.db.execute( + "SELECT value FROM metadata WHERE key='console_schema_version'" + ).fetchone() + if row is None or row[0] != SCHEMA_VERSION: + observed = None if row is None else row[0] + self.db.close() + raise ConsoleStoreError( + "console sidecar schema version " + + repr(observed) + + " does not match " + + repr(SCHEMA_VERSION) + + "; the sidecar is derived data -- regenerate it with " + "hotato serve --production-db DB --rebuild-scores" + ) + + def _begin(self) -> None: + self.db.execute("BEGIN IMMEDIATE") + + def _commit(self) -> None: + self.db.execute("COMMIT") + + def _rollback(self) -> None: + if self.db.in_transaction: + self.db.execute("ROLLBACK") + + def upsert_score(self, record: Dict[str, Any]) -> None: + """Durably commit one full score record (row + hop rows) atomically. + + Raises on any storage failure; the caller must treat an exception as a + persist failure to surface, never as a stored score. + """ + + missing = sorted(set(_RECORD_FIELDS) - set(record)) + if missing: + raise ConsoleStoreError( + "console score record missing: " + ", ".join(missing) + ) + if record["state"] not in SCORE_STATES: + raise ConsoleStoreError( + "console score state must be one of " + ", ".join(SCORE_STATES) + ) + subject = str(record["subject"]) + now = float(self.clock()) + with self._lock: + self._begin() + try: + self.db.execute( + "INSERT INTO scores(subject,state,reason,session_state," + "evidence_sha256,event_count,scorer_version,config_sha256," + "config_json,dimensions_json,candidates_json,timing_json," + "audio_json,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) " + "ON CONFLICT(subject) DO UPDATE SET state=excluded.state," + "reason=excluded.reason,session_state=excluded.session_state," + "evidence_sha256=excluded.evidence_sha256," + "event_count=excluded.event_count," + "scorer_version=excluded.scorer_version," + "config_sha256=excluded.config_sha256," + "config_json=excluded.config_json," + "dimensions_json=excluded.dimensions_json," + "candidates_json=excluded.candidates_json," + "timing_json=excluded.timing_json," + "audio_json=excluded.audio_json," + "created_at=excluded.created_at", + ( + subject, + record["state"], + record["reason"], + record["session_state"], + record["evidence_sha256"], + int(record["event_count"]), + record["scorer_version"], + record["config_sha256"], + safe_json_dumps(record["config"], sort_keys=True), + safe_json_dumps(record["dimensions"], sort_keys=True), + safe_json_dumps(record["candidates"], sort_keys=True), + safe_json_dumps(record["timing"], sort_keys=True), + ( + None + if record["audio"] is None + else safe_json_dumps(record["audio"], sort_keys=True) + ), + now, + ), + ) + self.db.execute( + "DELETE FROM latency_hops WHERE subject=?", (subject,) + ) + for seq, hop in enumerate(record["hops"]): + self.db.execute( + "INSERT INTO latency_hops(subject,seq,kind,at,latency_ms," + "authority,source,event_id) VALUES(?,?,?,?,?,?,?,?)", + ( + subject, + seq, + hop["kind"], + hop["at"], + hop["latency_ms"], + hop["authority"], + hop["source"], + hop["event_id"], + ), + ) + self._commit() + except BaseException: + self._rollback() + raise + + def record_error( + self, + subject: str, + *, + reason: str, + session_state: str, + evidence_sha256: str, + event_count: int, + scorer_version: str, + config_sha256: str, + ) -> None: + """The deliberately minimal ERROR write path. + + Used when persisting a full record failed: this single-row write is the + smallest durable way to make that failure visible in the sidecar + instead of silently skipping the session (a full re-score is retried + on the next cycle because ERROR rows are never treated as settled). + """ + + self.upsert_score( + { + "subject": subject, + "state": "ERROR", + "reason": reason, + "session_state": session_state, + "evidence_sha256": evidence_sha256, + "event_count": int(event_count), + "scorer_version": scorer_version, + "config_sha256": config_sha256, + "config": {}, + "dimensions": {}, + "candidates": [], + "timing": {}, + "audio": None, + "hops": [], + } + ) + + def score_identity(self, subject: str) -> Optional[Tuple[str, str]]: + """``(state, evidence_sha256)`` for one subject, or ``None``.""" + + with self._lock: + row = self.db.execute( + "SELECT state,evidence_sha256 FROM scores WHERE subject=?", + (subject,), + ).fetchone() + return None if row is None else (row["state"], row["evidence_sha256"]) + + def get_score(self, subject: str) -> Optional[Dict[str, Any]]: + with self._lock: + row = self.db.execute( + "SELECT * FROM scores WHERE subject=?", (subject,) + ).fetchone() + if row is None: + return None + hops = self.db.execute( + "SELECT seq,kind,at,latency_ms,authority,source,event_id " + "FROM latency_hops WHERE subject=? ORDER BY seq", + (subject,), + ).fetchall() + return self._record_from_rows(row, hops) + + def next_subject(self, after: str) -> Optional[str]: + """The next scored subject strictly after ``after`` in canonical + (subject-ascending) order -- one row at a time, for bounded sweeps.""" + + with self._lock: + row = self.db.execute( + "SELECT subject FROM scores WHERE subject>? ORDER BY subject " + "LIMIT 1", + (after,), + ).fetchone() + return None if row is None else row["subject"] + + def delete_score(self, subject: str) -> None: + with self._lock: + self._begin() + try: + self.db.execute( + "DELETE FROM latency_hops WHERE subject=?", (subject,) + ) + self.db.execute("DELETE FROM scores WHERE subject=?", (subject,)) + self._commit() + except BaseException: + self._rollback() + raise + + def clear(self) -> None: + """Delete every derived row (rebuild always starts from empty).""" + + with self._lock: + self._begin() + try: + self.db.execute("DELETE FROM latency_hops") + self.db.execute("DELETE FROM scores") + self._commit() + except BaseException: + self._rollback() + raise + + def counts(self) -> Dict[str, int]: + with self._lock: + observed = { + row[0]: int(row[1]) + for row in self.db.execute( + "SELECT state,COUNT(*) FROM scores GROUP BY state" + ) + } + return {state: observed.get(state, 0) for state in SCORE_STATES} + + def _record_from_rows( + self, row: sqlite3.Row, hops: List[sqlite3.Row] + ) -> Dict[str, Any]: + return { + "subject": row["subject"], + "state": row["state"], + "reason": row["reason"], + "session_state": row["session_state"], + "evidence_sha256": row["evidence_sha256"], + "event_count": row["event_count"], + "scorer_version": row["scorer_version"], + "config_sha256": row["config_sha256"], + "config": json.loads(row["config_json"]), + "dimensions": json.loads(row["dimensions_json"]), + "candidates": json.loads(row["candidates_json"]), + "timing": json.loads(row["timing_json"]), + "audio": ( + None if row["audio_json"] is None else json.loads(row["audio_json"]) + ), + "hops": [ + { + "kind": hop["kind"], + "at": hop["at"], + "latency_ms": hop["latency_ms"], + "authority": hop["authority"], + "source": hop["source"], + "event_id": hop["event_id"], + } + for hop in hops + ], + } + + def canonical_dump(self) -> str: + """Every derived record in canonical order as canonical JSON. + + Ordering is subject-ascending (rows) and hop sequence (per row); + ``created_at`` -- the one wall-clock column -- is excluded, so the same + evidence database always dumps to identical bytes. This is the + rebuild-determinism comparison surface. + """ + + records: List[Dict[str, Any]] = [] + with self._lock: + rows = self.db.execute( + "SELECT * FROM scores ORDER BY subject" + ).fetchall() + for row in rows: + hops = self.db.execute( + "SELECT seq,kind,at,latency_ms,authority,source,event_id " + "FROM latency_hops WHERE subject=? ORDER BY seq", + (row["subject"],), + ).fetchall() + records.append(self._record_from_rows(row, hops)) + return safe_json_dumps( + { + "schema": "hotato.console-scores-dump.v1", + "schema_version": SCHEMA_VERSION, + "records": records, + }, + sort_keys=True, + separators=(",", ":"), + ) + + def close(self) -> None: + with self._lock: + self.db.close() diff --git a/src/hotato/console_worker.py b/src/hotato/console_worker.py new file mode 100644 index 0000000..3ca1f3b --- /dev/null +++ b/src/hotato/console_worker.py @@ -0,0 +1,849 @@ +"""Score-on-arrival for the production evidence plane (the console engine). + +A supervisor-pattern background worker polls the production evidence database +for sessions that reached ``COMPLETE``/``QUIESCENT``, scores each one with the +existing deterministic scorer (:mod:`hotato.scan` over the session's recorded +audio evidence), and durably writes the derived record to the console sidecar +(:mod:`hotato.console_store`). Invariants: + +* **the evidence database is opened strictly read-only** -- SQLite ``mode=ro`` + plus ``query_only`` plus a write-denying authorizer, the same discipline as + :mod:`hotato.serve.production_bridge`. Unlike that metadata-only bridge, + this reader does select stored event payloads: the (already default-deny + redacted) payload fields ARE the scoring evidence -- audio asset paths, + per-hop latency numbers, turn timing; +* **one session at a time** -- sessions and events are walked with keyset + queries; no all-sessions list is ever materialized; +* **a score is claimed only after a durable sidecar commit** -- a persist + failure becomes a visible ERROR record (through the minimal error write + path) plus a counter, never a success return; +* **refusal is a first-class state** -- absent/unavailable/mono/unreadable + audio is the scorer's NOT SCORABLE refusal, recorded with its reason; a + scorer crash on one session becomes an ERROR record and the loop continues + to the next session; +* **no wall-clock values in scored content** -- every timing figure derives + from evidence event timestamps and carries the reporting event's declared + authority, so a rebuild from the same evidence database reproduces the + sidecar byte-for-byte (canonical dump). +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import stat +import threading +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from ._engine.score import ScoreConfig +from .core import _config_block +from .console_store import ConsoleStore +from .production import _canonical, _parse_rfc3339, _sha +from .scan import DEFAULT_MIN_GAP_SEC, KINDS, candidate_plain_english, scan_recording + +__all__ = [ + "SCORABLE_SESSION_STATES", + "ConsoleWorkerError", + "ConsoleScoreWorker", + "default_console_path", + "rebuild_sidecar", + "run_rebuild", + "score_session", + "scorer_provenance", +] + +# The arrival condition (spec R1): a session has quiesced (``session.ended`` +# observed) or was finalized COMPLETE. DEGRADED/OPEN/EXPIRED/DELETED sessions +# are not scored; a scored session that later leaves this set (a late event +# degraded it, or it was deleted) has its derived row pruned so the sidecar +# always equals a fresh rebuild at steady state. +SCORABLE_SESSION_STATES = ("COMPLETE", "QUIESCENT") + +_EVIDENCE_SCHEMA_VERSION = "1" +_HOP_EVENT_TYPES = ("model.operation", "tool.result") +# Timing-span provenance label: the value was computed from two evidence event +# timestamps, not reported by any vendor field. +_DERIVED_AUTHORITY = "derived:event_timestamps" + + +class ConsoleWorkerError(ValueError): + """The evidence database cannot be scored safely.""" + + +def default_console_path(production_db: str) -> str: + """The sidecar's canonical home: ``console.sqlite3`` beside the evidence db.""" + + resolved = os.path.abspath(os.path.expanduser(production_db)) + return os.path.join(os.path.dirname(resolved) or ".", "console.sqlite3") + + +# --------------------------------------------------------------------------- +# read-only evidence access +# --------------------------------------------------------------------------- + +_WRITE_ACTIONS = frozenset( + action + for action in ( + getattr(sqlite3, name, None) + for name in ( + "SQLITE_INSERT", + "SQLITE_UPDATE", + "SQLITE_DELETE", + "SQLITE_CREATE_INDEX", + "SQLITE_CREATE_TABLE", + "SQLITE_CREATE_TEMP_INDEX", + "SQLITE_CREATE_TEMP_TABLE", + "SQLITE_CREATE_TEMP_TRIGGER", + "SQLITE_CREATE_TEMP_VIEW", + "SQLITE_CREATE_TRIGGER", + "SQLITE_CREATE_VIEW", + "SQLITE_CREATE_VTABLE", + "SQLITE_DROP_INDEX", + "SQLITE_DROP_TABLE", + "SQLITE_DROP_TEMP_INDEX", + "SQLITE_DROP_TEMP_TABLE", + "SQLITE_DROP_TEMP_TRIGGER", + "SQLITE_DROP_TEMP_VIEW", + "SQLITE_DROP_TRIGGER", + "SQLITE_DROP_VIEW", + "SQLITE_DROP_VTABLE", + "SQLITE_ALTER_TABLE", + "SQLITE_REINDEX", + "SQLITE_ANALYZE", + "SQLITE_ATTACH", + "SQLITE_DETACH", + ) + ) + if action is not None +) + + +def _deny_writes_authorizer( + action: int, + argument_1: Any, + argument_2: Any, + database_name: Any, + trigger_name: Any, +) -> int: + del argument_1, argument_2, database_name, trigger_name + if action in _WRITE_ACTIONS: + return sqlite3.SQLITE_DENY + return sqlite3.SQLITE_OK + + +def _open_evidence_ro(path: str) -> sqlite3.Connection: + """Open the evidence database read-only (mode=ro + query_only + authorizer).""" + + if not isinstance(path, str) or not path.strip() or "\x00" in path: + raise ConsoleWorkerError("production database path must be a non-empty path") + resolved = os.path.abspath(os.path.expanduser(path)) + try: + before = os.lstat(resolved) + except OSError as exc: + raise ConsoleWorkerError( + f"production database is not readable: {resolved!r} ({exc})" + ) from exc + if not stat.S_ISREG(before.st_mode): + raise ConsoleWorkerError( + f"production database must be a regular file: {resolved!r}" + ) + uri = Path(resolved).as_uri() + "?mode=ro" + try: + db = sqlite3.connect(uri, uri=True, timeout=5, isolation_level=None) + except sqlite3.Error as exc: + raise ConsoleWorkerError( + f"could not open production database read-only: {exc}" + ) from exc + db.row_factory = sqlite3.Row + try: + after = os.lstat(resolved) + if ( + not stat.S_ISREG(after.st_mode) + or (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino) + ): + raise ConsoleWorkerError( + "production database changed while SQLite was opening it" + ) + db.execute("PRAGMA query_only=ON") + db.execute("PRAGMA busy_timeout=5000") + db.set_authorizer(_deny_writes_authorizer) + _verify_evidence_schema(db) + except BaseException: + db.close() + raise + return db + + +def _verify_evidence_schema(db: sqlite3.Connection) -> None: + try: + row = db.execute( + "SELECT value FROM metadata WHERE key='production_schema_version'" + ).fetchone() + except sqlite3.Error as exc: + raise ConsoleWorkerError( + "selected database is not a hotato production evidence store" + ) from exc + if row is None or row[0] != _EVIDENCE_SCHEMA_VERSION: + observed = None if row is None else row[0] + raise ConsoleWorkerError( + "unsupported production database schema version: " + repr(observed) + ) + required = { + "sessions": {"subject", "state", "evidence_json", "event_count"}, + "events": { + "subject", + "source", + "event_id", + "type", + "source_time", + "received", + "stored_sha256", + "payload_json", + }, + } + for table, expected in required.items(): + observed_columns = { + item[1] for item in db.execute(f"PRAGMA table_info({table})") + } + missing = sorted(expected - observed_columns) + if missing: + raise ConsoleWorkerError( + f"incompatible production database table {table!r}; missing " + + ", ".join(missing) + ) + + +# --------------------------------------------------------------------------- +# deterministic scoring of one session +# --------------------------------------------------------------------------- + + +def scorer_provenance() -> Tuple[str, str, Dict[str, Any]]: + """``(scorer_version, config_sha256, config)`` for every record written. + + The config snapshot is the scorer's full self-describing threshold block + (the same one ``dump-frames`` embeds) plus the scan-walk parameter, hashed + canonically, so a verdict row always says exactly which scorer produced it. + """ + + from . import __version__ + + config = dict(_config_block(ScoreConfig())) + config["min_gap_sec"] = DEFAULT_MIN_GAP_SEC + config["scorer"] = "hotato.scan" + return __version__, _sha(_canonical(config)), config + + +def _evidence_sha256(db: sqlite3.Connection, subject: str) -> Tuple[str, int]: + """The session's stored-event-log digest (the manifest's algorithm).""" + + digests = [ + row[0] + for row in db.execute( + "SELECT stored_sha256 FROM events WHERE subject=? " + "ORDER BY received,source,event_id", + (subject,), + ) + ] + return _sha(_canonical(digests)), len(digests) + + +def _session_events(db: sqlite3.Connection, subject: str) -> List[sqlite3.Row]: + """One session's events in evidence-time order (deterministic: the stored + ``source_time`` string, then source, then event id -- no arrival clock).""" + + return db.execute( + "SELECT type,source,event_id,source_time,sequence,payload_json " + "FROM events WHERE subject=? ORDER BY source_time,source,event_id", + (subject,), + ).fetchall() + + +def _payload(row: sqlite3.Row) -> Dict[str, Any]: + try: + value = json.loads(row["payload_json"]) + except (TypeError, ValueError): + return {} + return value if isinstance(value, dict) else {} + + +def _number(value: Any) -> Optional[float]: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def _elapsed_ms(start: str, end: str) -> Optional[float]: + try: + delta = _parse_rfc3339(end) - _parse_rfc3339(start) + except (TypeError, ValueError): + return None + return round(delta.total_seconds() * 1000.0, 3) + + +def _timing(events: List[sqlite3.Row]) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]: + """Timing spans + per-hop latency, derived purely from evidence. + + Per-hop rows carry the reporting event's declared ``authority`` kind (so a + vendor-reported latency never masquerades as a measurement); turn spans and + the end-to-end figure are computed from event timestamps and labeled + ``derived:event_timestamps``. + """ + + hops: List[Dict[str, Any]] = [] + spans: List[Dict[str, Any]] = [] + open_turns: List[Tuple[str, str]] = [] + session_started: Optional[str] = None + session_ended: Optional[str] = None + started_count = 0 + ended_count = 0 + for row in events: + payload = _payload(row) + data = payload.get("data") if isinstance(payload.get("data"), dict) else {} + authority = payload.get("authority") + authority_kind = ( + authority.get("kind") if isinstance(authority, dict) else None + ) or "submitted" + event_type = row["type"] + if event_type == "session.started": + started_count += 1 + session_started = row["source_time"] if started_count == 1 else None + elif event_type == "session.ended": + ended_count += 1 + session_ended = row["source_time"] if ended_count == 1 else None + elif event_type in _HOP_EVENT_TYPES: + latency = _number(data.get("latency_ms")) + if latency is None: + latency = _number(data.get("duration_ms")) + hops.append( + { + "kind": event_type, + "at": row["source_time"], + "latency_ms": latency, + "authority": authority_kind, + "source": row["source"], + "event_id": row["event_id"], + } + ) + elif event_type == "turn.started": + open_turns.append((row["source_time"], row["event_id"])) + elif event_type == "turn.ended": + start = open_turns.pop() if open_turns else None + span: Dict[str, Any] = { + "started_at": start[0] if start else None, + "ended_at": row["source_time"], + "duration_ms": ( + _elapsed_ms(start[0], row["source_time"]) if start else None + ), + "authority": _DERIVED_AUTHORITY, + "started_event_id": start[1] if start else None, + "ended_event_id": row["event_id"], + } + for key in ("yield_latency_ms", "overlap_ms", "duration_ms"): + reported = _number(data.get(key)) + if reported is not None: + span.setdefault("reported", {})[key] = reported + spans.append(span) + for span in spans: + if span["duration_ms"] is not None: + hops.append( + { + "kind": "turn", + "at": span["ended_at"], + "latency_ms": span["duration_ms"], + "authority": _DERIVED_AUTHORITY, + "source": "console", + "event_id": span["ended_event_id"], + } + ) + end_to_end_ms = ( + _elapsed_ms(session_started, session_ended) + if session_started and session_ended + else None + ) + timing = { + "provenance": ( + "derived from evidence event timestamps; per-hop rows carry the " + "reporting event's declared authority" + ), + "end_to_end_ms": end_to_end_ms, + "turn_count": len(spans), + "turn_spans": spans, + "hop_count": len(hops), + } + return timing, hops + + +_MAGNITUDE_KEYS = ("overlap_sec", "gap_sec", "trailing_silence_sec", "activity_sec") + + +def _candidate_magnitude(candidate: Dict[str, Any]) -> Optional[float]: + durations = candidate.get("durations") or {} + for key in _MAGNITUDE_KEYS: + value = _number(durations.get(key)) + if value is not None: + return value + return None + + +def _dimensions(candidates: List[Dict[str, Any]]) -> Dict[str, Any]: + """Per-dimension observations: one entry per scan kind, counts plus the + worst measured magnitude in seconds. Never blended into one score.""" + + output: Dict[str, Any] = {} + for kind in KINDS: + matching = [c for c in candidates if c.get("kind") == kind] + magnitudes = [ + m for m in (_candidate_magnitude(c) for c in matching) if m is not None + ] + output[kind] = { + "candidate_count": len(matching), + "worst_sec": max(magnitudes) if magnitudes else None, + } + return output + + +def _audio_path(events: List[sqlite3.Row]) -> Optional[str]: + """The first recorded local audio path, in evidence-time order. + + ``media.asset.available`` events may carry a ``data.path`` string naming + the local recording. Default-deny redaction reduces that field to a + descriptor, so a redacted store yields no path here -- which the caller + records as the scorer's NOT SCORABLE refusal, never a guess. + """ + + for row in events: + if row["type"] != "media.asset.available": + continue + payload = _payload(row) + data = payload.get("data") if isinstance(payload.get("data"), dict) else {} + path = data.get("path") + if isinstance(path, str) and path and "\x00" not in path: + return path + return None + + +def score_session( + db: sqlite3.Connection, + subject: str, + *, + session_state: str, + evidence_sha256: str, + event_count: int, + scorer_version: str, + config_sha256: str, + config: Dict[str, Any], +) -> Dict[str, Any]: + """Score one session's evidence into a full sidecar record. + + A scorer refusal (no/unavailable audio lane, no recorded path, missing or + mono or unreadable recording) returns a NOT_SCORABLE record carrying the + refusal reason. Only unexpected exceptions propagate (the worker turns + them into an ERROR record). + """ + + session = db.execute( + "SELECT evidence_json FROM sessions WHERE subject=?", (subject,) + ).fetchone() + if session is None: + raise KeyError(subject) + try: + evidence = json.loads(session["evidence_json"]) + except (TypeError, ValueError): + evidence = {} + events = _session_events(db, subject) + timing, hops = _timing(events) + + record: Dict[str, Any] = { + "subject": subject, + "state": "NOT_SCORABLE", + "reason": None, + "session_state": session_state, + "evidence_sha256": evidence_sha256, + "event_count": event_count, + "scorer_version": scorer_version, + "config_sha256": config_sha256, + "config": config, + "dimensions": {}, + "candidates": [], + "timing": timing, + "audio": None, + "hops": hops, + } + + lane = evidence.get("participant_audio") or {} + availability = lane.get("availability", "missing") + if availability != "available": + record["reason"] = ( + f"participant_audio evidence lane is {availability}; scoring reads " + "a recorded two-channel call" + ) + return record + path = _audio_path(events) + if path is None: + record["reason"] = ( + "participant_audio evidence carries no local audio path; scoring " + "reads the recording named by media.asset.available data.path" + ) + return record + try: + scan = scan_recording(path) + except (ValueError, OSError) as exc: + # The scorer's own refusal (mono/one-channel, truncated, corrupt, or + # unreadable recording): a first-class NOT_SCORABLE with its reason. + record["reason"] = str(exc) + return record + candidates = scan["candidates"] + record["state"] = "SCORED" + record["reason"] = ( + candidate_plain_english(candidates[0]) if candidates else None + ) + record["dimensions"] = _dimensions(candidates) + record["candidates"] = candidates + record["audio"] = { + "path": path, + "source": scan["source"], + "duration_sec": scan["duration_sec"], + "sample_rate": scan["sample_rate"], + "total_candidates": scan["total_candidates"], + } + return record + + +# --------------------------------------------------------------------------- +# the supervisor-pattern worker +# --------------------------------------------------------------------------- + + +class ConsoleScoreWorker: + """Run one serialized score-on-arrival cycle at a bounded fixed interval.""" + + def __init__( + self, + production_db: str, + store: ConsoleStore, + *, + interval_seconds: float = 5.0, + clock=time.time, + autostart: bool = True, + ) -> None: + if ( + isinstance(interval_seconds, bool) + or not isinstance(interval_seconds, (int, float)) + or not 0.1 <= float(interval_seconds) <= 86_400 + ): + raise ValueError("interval_seconds must be in [0.1, 86400]") + self.production_db = os.path.abspath(os.path.expanduser(production_db)) + self.store = store + self.interval_seconds = float(interval_seconds) + self.clock = clock + self._stop = threading.Event() + self._cycle_lock = threading.Lock() + self._state_lock = threading.Lock() + self._last: Dict[str, Any] = { + "schema": "hotato.console-score-status.v1", + "state": "STARTING" if autostart else "IDLE", + "cycles": 0, + "last_started_at": None, + "last_completed_at": None, + "last_error": None, + "last_result": None, + } + self.thread = threading.Thread( + target=self._loop, + name="hotato-console-score", + daemon=True, + ) + if autostart: + self.thread.start() + + def run_once(self) -> Dict[str, Any]: + if not self._cycle_lock.acquire(blocking=False): + raise RuntimeError("a console scoring cycle is already running") + started = float(self.clock()) + with self._state_lock: + self._last["state"] = "RUNNING" + self._last["last_started_at"] = started + try: + db = _open_evidence_ro(self.production_db) + try: + result = self._cycle(db) + finally: + db.close() + result.update( + { + "schema": "hotato.console-score-cycle.v1", + "started_at": started, + "completed_at": float(self.clock()), + } + ) + with self._state_lock: + self._last.update( + { + "state": "IDLE", + "cycles": int(self._last["cycles"]) + 1, + "last_completed_at": result["completed_at"], + "last_error": None, + "last_result": result, + } + ) + return result + except Exception as exc: + with self._state_lock: + self._last.update( + { + "state": "ERROR", + "cycles": int(self._last["cycles"]) + 1, + "last_completed_at": float(self.clock()), + "last_error": { + "type": type(exc).__name__, + "message": str(exc)[:1000], + }, + } + ) + raise + finally: + self._cycle_lock.release() + + def _cycle(self, db: sqlite3.Connection) -> Dict[str, Any]: + version, config_sha, config = scorer_provenance() + counts = { + "scored": 0, + "not_scorable": 0, + "errors": 0, + "skipped": 0, + "persist_failures": 0, + "pruned": 0, + } + placeholders = ",".join("?" for _ in SCORABLE_SESSION_STATES) + after = "" + while not self._stop.is_set(): + row = db.execute( + f"SELECT subject,state FROM sessions WHERE state IN ({placeholders}) " + "AND subject>? ORDER BY subject LIMIT 1", + (*SCORABLE_SESSION_STATES, after), + ).fetchone() + if row is None: + break + after = row["subject"] + self._score_one(db, row["subject"], row["state"], counts, + version, config_sha, config) + counts["pruned"] = self._prune(db) + return counts + + def _score_one( + self, + db: sqlite3.Connection, + subject: str, + session_state: str, + counts: Dict[str, int], + version: str, + config_sha: str, + config: Dict[str, Any], + ) -> None: + sha, event_count = _evidence_sha256(db, subject) + existing = self.store.score_identity(subject) + # An ERROR row is never settled: it is retried every cycle until it + # either scores or keeps its (deterministic) error reason. + if existing is not None and existing[1] == sha and existing[0] != "ERROR": + counts["skipped"] += 1 + return + try: + record = score_session( + db, + subject, + session_state=session_state, + evidence_sha256=sha, + event_count=event_count, + scorer_version=version, + config_sha256=config_sha, + config=config, + ) + except Exception as exc: + # A scorer crash on one session must not kill the loop or block + # other sessions: it becomes its own visible ERROR record. + record = { + "subject": subject, + "state": "ERROR", + "reason": f"{type(exc).__name__}: {exc}", + "session_state": session_state, + "evidence_sha256": sha, + "event_count": event_count, + "scorer_version": version, + "config_sha256": config_sha, + "config": config, + "dimensions": {}, + "candidates": [], + "timing": {}, + "audio": None, + "hops": [], + } + try: + self.store.upsert_score(record) + except Exception as exc: + # I1: a persist failure is never a silent skip and never a claimed + # score. Surface it through the minimal error write path; if even + # that fails, the counter still records it and the next cycle + # retries the session. + counts["persist_failures"] += 1 + try: + self.store.record_error( + subject, + reason=f"persist_failure: {type(exc).__name__}: {exc}", + session_state=session_state, + evidence_sha256=sha, + event_count=event_count, + scorer_version=version, + config_sha256=config_sha, + ) + except Exception: + pass + counts["errors"] += 1 + return + if record["state"] == "SCORED": + counts["scored"] += 1 + elif record["state"] == "NOT_SCORABLE": + counts["not_scorable"] += 1 + else: + counts["errors"] += 1 + + def _prune(self, db: sqlite3.Connection) -> int: + """Drop derived rows whose session left the scorable set (deleted, or + degraded by a late event), so the sidecar equals a fresh rebuild.""" + + placeholders = ",".join("?" for _ in SCORABLE_SESSION_STATES) + pruned = 0 + after = "" + while True: + subject = self.store.next_subject(after) + if subject is None: + break + after = subject + row = db.execute( + f"SELECT 1 FROM sessions WHERE subject=? AND state IN ({placeholders})", + (subject, *SCORABLE_SESSION_STATES), + ).fetchone() + if row is None: + self.store.delete_score(subject) + pruned += 1 + return pruned + + def status(self) -> Dict[str, Any]: + with self._state_lock: + return json.loads(json.dumps(self._last, allow_nan=False)) + + def _loop(self) -> None: + while not self._stop.is_set(): + try: + self.run_once() + except Exception: + # Status retains the bounded error; the next interval retries + # because a transient SQLite/disk failure must not silently + # kill scoring. + pass + self._stop.wait(self.interval_seconds) + + def close(self, timeout: float = 10.0) -> None: + self._stop.set() + if self.thread.is_alive(): + self.thread.join(timeout=timeout) + if self.thread.is_alive(): + raise RuntimeError("console scoring thread did not stop") + with self._state_lock: + if self._last["state"] not in {"ERROR", "RUNNING"}: + self._last["state"] = "STOPPED" + + +# --------------------------------------------------------------------------- +# deterministic rebuild +# --------------------------------------------------------------------------- + + +def rebuild_sidecar(production_db: str, store: ConsoleStore) -> Dict[str, Any]: + """Regenerate the ENTIRE sidecar from the evidence database. + + Starts from empty and scores every scorable session in canonical (subject- + ascending) order, one at a time. The same evidence database always + produces the same canonical dump (no wall-clock value enters any scored + field), which is the migration story: a sidecar schema change ships as + this rebuild. + """ + + version, config_sha, config = scorer_provenance() + db = _open_evidence_ro(production_db) + try: + store.clear() + counts = {"scored": 0, "not_scorable": 0, "errors": 0} + placeholders = ",".join("?" for _ in SCORABLE_SESSION_STATES) + after = "" + while True: + row = db.execute( + f"SELECT subject,state FROM sessions WHERE state IN ({placeholders}) " + "AND subject>? ORDER BY subject LIMIT 1", + (*SCORABLE_SESSION_STATES, after), + ).fetchone() + if row is None: + break + after = row["subject"] + sha, event_count = _evidence_sha256(db, row["subject"]) + try: + record = score_session( + db, + row["subject"], + session_state=row["state"], + evidence_sha256=sha, + event_count=event_count, + scorer_version=version, + config_sha256=config_sha, + config=config, + ) + except Exception as exc: + record = { + "subject": row["subject"], + "state": "ERROR", + "reason": f"{type(exc).__name__}: {exc}", + "session_state": row["state"], + "evidence_sha256": sha, + "event_count": event_count, + "scorer_version": version, + "config_sha256": config_sha, + "config": config, + "dimensions": {}, + "candidates": [], + "timing": {}, + "audio": None, + "hops": [], + } + store.upsert_score(record) + counts[ + {"SCORED": "scored", "NOT_SCORABLE": "not_scorable", "ERROR": "errors"}[ + record["state"] + ] + ] += 1 + finally: + db.close() + return { + "schema": "hotato.console-rebuild.v1", + "sidecar": store.path, + "scored": counts["scored"], + "not_scorable": counts["not_scorable"], + "errors": counts["errors"], + } + + +def run_rebuild( + production_db: str, *, console_db: Optional[str] = None +) -> Dict[str, Any]: + """CLI entry: rebuild the sidecar beside the evidence db and return the + summary. Raises ``ValueError``/``OSError`` (HANDLED -> exit 2) on an + unusable evidence database or sidecar path.""" + + sidecar_path = console_db or default_console_path(production_db) + store = ConsoleStore(sidecar_path) + try: + return rebuild_sidecar(production_db, store) + finally: + store.close() diff --git a/src/hotato/serve/app.py b/src/hotato/serve/app.py index 95f988e..588dffe 100644 --- a/src/hotato/serve/app.py +++ b/src/hotato/serve/app.py @@ -522,7 +522,8 @@ def build_server(context: ServeContext, host: str, port: int) -> _WorkspaceServe def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, registry: Optional[str] = None, token: Optional[str] = None, token_file: Optional[str] = None, open_browser: bool = True, - production_db: Optional[str] = None) -> int: + production_db: Optional[str] = None, + score_production: bool = False) -> int: """Start the workspace server and serve until interrupted. Returns 0 on a clean shutdown. Raises ``ValueError``/``OSError`` (HANDLED -> exit 2) on a bad registry, an unusable token, or an unavailable port. @@ -545,6 +546,12 @@ def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, snapshot = read_production_snapshot(production_db) resolved_production_db = snapshot["source"]["path"] + if score_production and resolved_production_db is None: + raise ValueError( + "--score-production needs --production-db (the evidence database " + "the scoring worker reads)" + ) + state_dir = os.path.join(home, "serve", _safe_dirname(workspace)) os.makedirs(state_dir, mode=0o700, exist_ok=True) @@ -557,6 +564,18 @@ def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, ) server = build_server(ctx, host, port) + # Optional score-on-arrival: a background worker (never a second server; + # the bind and auth above are unchanged) reads the evidence db mode=ro and + # writes derived score records to the console sidecar beside it. + score_worker = None + score_store = None + if score_production: + from ..console_store import ConsoleStore + from ..console_worker import ConsoleScoreWorker, default_console_path + + score_store = ConsoleStore(default_console_path(resolved_production_db)) + score_worker = ConsoleScoreWorker(resolved_production_db, score_store) + display_host = "127.0.0.1" if host in ("", "0.0.0.0", "::") else host # The one line that carries the secret: the tokenised URL a browser can open # directly. It is printed once (below) and never written to the audit log. @@ -564,6 +583,11 @@ def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, _print_banner(ctx, host, port, source=source, generated=generated, state_dir=state_dir, url=url, display_host=display_host) + if score_store is not None: + print(" scoring: %s (derived sidecar, rebuildable from the " + "evidence db with --rebuild-scores; completed sessions are " + "scored one at a time in the background)" + % score_store.path, file=sys.stderr) # The listening socket is already bound; a browser opened now queues onto the # accept backlog and is served the moment `serve_forever` runs. @@ -582,6 +606,11 @@ def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, print("\nhotato serve: shutting down.", file=sys.stderr) finally: server.server_close() + if score_worker is not None: + try: + score_worker.close() + finally: + score_store.close() return 0 diff --git a/tests/test_console_score.py b/tests/test_console_score.py new file mode 100644 index 0000000..e9fc98d --- /dev/null +++ b/tests/test_console_score.py @@ -0,0 +1,447 @@ +"""The console score-on-arrival engine (sidecar + worker + rebuild). + +Pinned here: the sidecar schema is versioned and a mismatched version is +refused with a rebuild instruction; the worker scores a completed session +end-to-end from an evidence db built through production.py's own ingest API +(per-dimension observations, ranked candidates, the measured failure-reason +sentence, per-hop latency with authority provenance); refusal is a first-class +NOT_SCORABLE row with its reason (absent path, missing lane, mono recording); +a scorer crash becomes an ERROR row and the loop continues; a persist failure +surfaces as a visible ERROR, never a claimed score; sessions are walked one at +a time (every sessions-table pick is a LIMIT-1 keyset query); and the same +evidence db always rebuilds to a byte-identical canonical dump. +""" + +from __future__ import annotations + +import shutil +import sqlite3 +import wave +from importlib import resources + +import pytest + +from hotato import __version__, cli +from hotato import console_worker as console_worker_mod +from hotato.console_store import SCHEMA_VERSION, ConsoleStore, ConsoleStoreError +from hotato.console_worker import ( + ConsoleScoreWorker, + default_console_path, + rebuild_sidecar, + run_rebuild, +) +from hotato.production import ProductionStore + + +def _stereo_fixture() -> str: + return str( + resources.files("hotato").joinpath( + "data", "audio", "01-hard-interruption.example.wav" + ) + ) + + +def _mono_wav(path) -> str: + with wave.open(str(path), "wb") as handle: + handle.setnchannels(1) + handle.setsampwidth(2) + handle.setframerate(16000) + handle.writeframes(b"\x00\x00" * 8000) + return str(path) + + +def _event( + event_id, + event_type, + *, + subject, + time_value, + data=None, + authority="adapter_reported", + sequence=None, +): + value = { + "specversion": "1.0", + "id": event_id, + "source": "console-fixture", + "type": event_type, + "subject": subject, + "time": time_value, + "data": {} if data is None else data, + "authority": { + "kind": authority, + "eligible_for_execution_claim": authority + in ("measured", "signed_attestation"), + }, + } + if sequence is not None: + value["sequence"] = sequence + return value + + +def _ingest_call( + store, + subject, + *, + audio_path=None, + with_media_event=True, + redact_payloads=False, +): + """One quiesced call session through the production plane's own ingest API: + lifecycle + audio asset + one model hop + one timed turn.""" + + events = [ + _event(f"{subject}-start", "session.started", subject=subject, + time_value="2026-07-17T12:00:00Z"), + _event(f"{subject}-turn-a", "turn.started", subject=subject, + time_value="2026-07-17T12:00:01Z"), + _event( + f"{subject}-model", + "model.operation", + subject=subject, + time_value="2026-07-17T12:00:02Z", + data={"latency_ms": 210.0, "availability": "available"}, + ), + _event( + f"{subject}-turn-b", + "turn.ended", + subject=subject, + time_value="2026-07-17T12:00:03.500Z", + data={"yield_latency_ms": 480.0}, + ), + _event(f"{subject}-end", "session.ended", subject=subject, + time_value="2026-07-17T12:00:06Z"), + ] + if with_media_event: + data = {"availability": "available", "channels": 2} + if audio_path is not None: + data["path"] = audio_path + events.insert( + 1, + _event( + f"{subject}-audio", + "media.asset.available", + subject=subject, + time_value="2026-07-17T12:00:00.500Z", + data=data, + authority="measured", + ), + ) + for event in events: + store.ingest(event, redact_payloads=redact_payloads) + + +def _evidence_db(tmp_path, name="production.sqlite3"): + return ProductionStore(str(tmp_path / name), clock=lambda: 1000.0) + + +def _worker(tmp_path, evidence_path, name="console.sqlite3"): + sidecar = ConsoleStore(str(tmp_path / name)) + worker = ConsoleScoreWorker(str(evidence_path), sidecar, autostart=False) + return worker, sidecar + + +# --- sidecar schema ----------------------------------------------------------- + +def test_sidecar_schema_version_present_and_mismatch_refused(tmp_path): + path = tmp_path / "console.sqlite3" + store = ConsoleStore(str(path)) + assert store.counts() == {"SCORED": 0, "NOT_SCORABLE": 0, "ERROR": 0} + store.close() + + db = sqlite3.connect(str(path)) + row = db.execute( + "SELECT value FROM metadata WHERE key='console_schema_version'" + ).fetchone() + assert row == (SCHEMA_VERSION,) + db.execute( + "UPDATE metadata SET value='999' WHERE key='console_schema_version'" + ) + db.commit() + db.close() + with pytest.raises(ConsoleStoreError, match="--rebuild-scores"): + ConsoleStore(str(path)) + + +# --- worker scores a completed session end-to-end ----------------------------- + +def test_worker_scores_completed_session_end_to_end(tmp_path): + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-a", audio_path=_stereo_fixture()) + worker, sidecar = _worker(tmp_path, evidence.path) + + result = worker.run_once() + assert result["scored"] == 1 + assert result["errors"] == 0 and result["persist_failures"] == 0 + + record = sidecar.get_score("call-a") + assert record["state"] == "SCORED" + assert record["session_state"] == "QUIESCENT" + # The measured failure-reason sentence is the top candidate's plain-English + # timing sentence -- built only from measured numbers. + assert record["reason"].startswith("the caller took the floor") + assert record["candidates"], "ranked candidate moments recorded" + assert record["candidates"][0]["kind"] == "overlap_while_agent_talking" + dims = record["dimensions"] + assert dims["overlap_while_agent_talking"]["candidate_count"] >= 1 + assert dims["overlap_while_agent_talking"]["worst_sec"] > 0 + assert dims["long_response_gap"]["candidate_count"] == 0 + + # Provenance: scorer version + config hash on the record (I5). + assert record["scorer_version"] == __version__ + assert record["config_sha256"].startswith("sha256:") + assert record["config"]["scorer"] == "hotato.scan" + assert record["evidence_sha256"].startswith("sha256:") + + # Timing derives from evidence timestamps only: the model hop keeps its + # reported latency AND its declared authority; the turn span and the + # end-to-end figure are labeled as derived from event timestamps. + hops = {hop["kind"]: hop for hop in record["hops"]} + assert hops["model.operation"]["latency_ms"] == 210.0 + assert hops["model.operation"]["authority"] == "adapter_reported" + assert hops["turn"]["latency_ms"] == 2500.0 + assert hops["turn"]["authority"] == "derived:event_timestamps" + assert record["timing"]["end_to_end_ms"] == 6000.0 + assert record["timing"]["turn_spans"][0]["reported"]["yield_latency_ms"] == 480.0 + + # Unchanged evidence is skipped, never re-scored. + again = worker.run_once() + assert again["skipped"] == 1 and again["scored"] == 0 + + worker.close() + sidecar.close() + evidence.close() + + +# --- refusal is a first-class state ------------------------------------------ + +def test_not_scorable_rows_carry_their_reason(tmp_path): + evidence = _evidence_db(tmp_path) + # Redacted payloads: the media event exists but its path was reduced to a + # descriptor, so scoring has no recording to read. + _ingest_call(evidence, "call-redacted", audio_path=_stereo_fixture(), + redact_payloads=True) + # No media event at all: the participant_audio lane never became available. + _ingest_call(evidence, "call-no-audio", with_media_event=False) + # A one-channel recording: the scorer's own refusal, verbatim. + _ingest_call(evidence, "call-mono", + audio_path=_mono_wav(tmp_path / "mono.wav")) + worker, sidecar = _worker(tmp_path, evidence.path) + + result = worker.run_once() + assert result["not_scorable"] == 3 and result["scored"] == 0 + + redacted = sidecar.get_score("call-redacted") + assert redacted["state"] == "NOT_SCORABLE" + assert "no local audio path" in redacted["reason"] + + missing = sidecar.get_score("call-no-audio") + assert missing["state"] == "NOT_SCORABLE" + assert "participant_audio evidence lane is missing" in missing["reason"] + + mono = sidecar.get_score("call-mono") + assert mono["state"] == "NOT_SCORABLE" + assert "one channel" in mono["reason"] + # Timing still derives from the events even when audio is refused. + assert mono["timing"]["end_to_end_ms"] == 6000.0 + + worker.close() + sidecar.close() + evidence.close() + + +def test_scorer_crash_is_an_error_row_and_the_loop_continues(tmp_path, monkeypatch): + crashing = str(tmp_path / "crash.wav") + shutil.copy(_stereo_fixture(), crashing) + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-a-crash", audio_path=crashing) + _ingest_call(evidence, "call-b-fine", audio_path=_stereo_fixture()) + worker, sidecar = _worker(tmp_path, evidence.path) + + real_scan = console_worker_mod.scan_recording + + def crash_on_marked(path, **kwargs): + if path == crashing: + raise RuntimeError("boom") + return real_scan(path, **kwargs) + + monkeypatch.setattr(console_worker_mod, "scan_recording", crash_on_marked) + result = worker.run_once() + assert result["errors"] == 1 and result["scored"] == 1 + + crashed = sidecar.get_score("call-a-crash") + assert crashed["state"] == "ERROR" + assert crashed["reason"] == "RuntimeError: boom" + assert sidecar.get_score("call-b-fine")["state"] == "SCORED" + + worker.close() + sidecar.close() + evidence.close() + + +def test_persist_failure_surfaces_error_and_never_claims_a_score( + tmp_path, monkeypatch +): + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-a", audio_path=_stereo_fixture()) + worker, sidecar = _worker(tmp_path, evidence.path) + + real_upsert = ConsoleStore.upsert_score + + def failing_upsert(self, record): + if record["state"] != "ERROR": + raise sqlite3.OperationalError("disk I/O error") + return real_upsert(self, record) + + monkeypatch.setattr(ConsoleStore, "upsert_score", failing_upsert) + result = worker.run_once() + assert result["persist_failures"] == 1 + assert result["errors"] == 1 and result["scored"] == 0 + + surfaced = sidecar.get_score("call-a") + assert surfaced["state"] == "ERROR" + assert surfaced["reason"].startswith("persist_failure:") + + # The failure is never settled: once persistence works again, the next + # cycle re-scores the session. + monkeypatch.setattr(ConsoleStore, "upsert_score", real_upsert) + recovered = worker.run_once() + assert recovered["scored"] == 1 + assert sidecar.get_score("call-a")["state"] == "SCORED" + + worker.close() + sidecar.close() + evidence.close() + + +# --- bounded, one-at-a-time processing --------------------------------------- + +def test_sessions_are_walked_one_at_a_time(tmp_path, monkeypatch): + evidence = _evidence_db(tmp_path) + for index in range(5): + _ingest_call(evidence, f"call-{index}", audio_path=_stereo_fixture()) + worker, sidecar = _worker(tmp_path, evidence.path) + + statements = [] + real_open = console_worker_mod._open_evidence_ro + + def traced_open(path): + db = real_open(path) + db.set_trace_callback(statements.append) + return db + + monkeypatch.setattr(console_worker_mod, "_open_evidence_ro", traced_open) + result = worker.run_once() + assert result["scored"] == 5 + + session_picks = [ + sql for sql in statements + if "FROM sessions WHERE state IN" in sql + ] + assert session_picks, "the worker reads sessions through the keyset query" + assert all("LIMIT 1" in sql for sql in session_picks), ( + "every sessions pick is a single-row keyset query; no all-sessions " + "list is materialized" + ) + + worker.close() + sidecar.close() + evidence.close() + + +# --- rebuild determinism ------------------------------------------------------ + +def test_rebuild_is_deterministic_and_matches_the_worker(tmp_path): + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-scored", audio_path=_stereo_fixture()) + _ingest_call(evidence, "call-no-audio", with_media_event=False) + _ingest_call(evidence, "call-mono", + audio_path=_mono_wav(tmp_path / "mono.wav")) + + first = ConsoleStore(str(tmp_path / "rebuild-a.sqlite3")) + second = ConsoleStore(str(tmp_path / "rebuild-b.sqlite3")) + summary = rebuild_sidecar(evidence.path, first) + rebuild_sidecar(evidence.path, second) + assert summary["scored"] == 1 and summary["not_scorable"] == 2 + + dump_a = first.canonical_dump() + assert dump_a == second.canonical_dump() + + # The worker's incremental sidecar converges to the same canonical bytes. + worker, incremental = _worker(tmp_path, evidence.path, name="worker.sqlite3") + worker.run_once() + assert incremental.canonical_dump() == dump_a + + # A rebuild over an already-populated sidecar regenerates from empty and + # lands on the same bytes again. + rebuild_sidecar(evidence.path, first) + assert first.canonical_dump() == dump_a + + worker.close() + incremental.close() + first.close() + second.close() + evidence.close() + + +def test_worker_prunes_rows_for_sessions_that_left_the_scorable_set(tmp_path): + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-keep", audio_path=_stereo_fixture()) + _ingest_call(evidence, "call-gone", with_media_event=False) + worker, sidecar = _worker(tmp_path, evidence.path) + worker.run_once() + assert sidecar.get_score("call-gone") is not None + + evidence.delete_session("call-gone") + result = worker.run_once() + assert result["pruned"] == 1 + assert sidecar.get_score("call-gone") is None + assert sidecar.get_score("call-keep") is not None + + worker.close() + sidecar.close() + evidence.close() + + +# --- CLI wiring --------------------------------------------------------------- + +def test_rebuild_scores_cli_regenerates_the_sidecar_and_exits(tmp_path, capsys): + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-a", audio_path=_stereo_fixture()) + evidence.close() + db_path = str(tmp_path / "production.sqlite3") + + assert cli.main(["serve", "--rebuild-scores", "--production-db", db_path]) == 0 + out = capsys.readouterr().out + assert "scored: 1" in out + + sidecar_path = default_console_path(db_path) + sidecar = ConsoleStore(sidecar_path) + assert sidecar.counts()["SCORED"] == 1 + sidecar.close() + + # Without the evidence source the rebuild is a usage error, not a server. + assert cli.main(["serve", "--rebuild-scores"]) == 2 + + +def test_run_rebuild_returns_the_summary(tmp_path): + evidence = _evidence_db(tmp_path) + _ingest_call(evidence, "call-a", audio_path=_stereo_fixture()) + evidence.close() + summary = run_rebuild(str(tmp_path / "production.sqlite3")) + assert summary["scored"] == 1 + assert summary["sidecar"] == default_console_path( + str(tmp_path / "production.sqlite3") + ) + + +def test_score_production_requires_the_evidence_db(tmp_path): + from hotato.serve.app import run_serve + + with pytest.raises(ValueError, match="--production-db"): + run_serve( + workspace="default", + registry=str(tmp_path / "registry"), + score_production=True, + open_browser=False, + ) From 1add11e47844e2038fd15715942ca6cfe5614084 Mon Sep 17 00:00:00 2001 From: "David J. Kim" Date: Thu, 23 Jul 2026 15:30:33 -0400 Subject: [PATCH 2/2] docs: restore the --junit CHANGELOG heading consumed by the score-on-arrival entry; ruff import-sort on console_worker Claude-Session: https://claude.ai/code/session_015njkxtoAKioqTPT2k14xFr --- CHANGELOG.md | 3 ++- src/hotato/console_worker.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 24302a4..584843b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,7 +53,8 @@ Every entry reports millisecond measurement error and a confusion matrix. See `d The sidecar is derived data with a versioned schema: `--rebuild-scores` regenerates it entirely from the evidence database and exits, and the same evidence database always rebuilds to identical content. Bind, auth, and the - server's read-only route surface are unchanged. Both commands also + server's read-only route surface are unchanged. +- **`--junit` on `hotato suite run` and `hotato prove`.** Both commands also write a JUnit XML report any standard CI test widget ingests, mirroring each command's existing grouping: one `` per dimension (suite run) or per evidence lane (prove), one `` per test or lane. A diff --git a/src/hotato/console_worker.py b/src/hotato/console_worker.py index 3ca1f3b..b2eb2c9 100644 --- a/src/hotato/console_worker.py +++ b/src/hotato/console_worker.py @@ -39,8 +39,8 @@ from typing import Any, Dict, List, Optional, Tuple from ._engine.score import ScoreConfig -from .core import _config_block from .console_store import ConsoleStore +from .core import _config_block from .production import _canonical, _parse_rfc3339, _sha from .scan import DEFAULT_MIN_GAP_SEC, KINDS, candidate_plain_english, scan_recording