diff --git a/CHANGELOG.md b/CHANGELOG.md index 584843b..d5c6816 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,34 @@ Every entry reports millisecond measurement error and a confusion matrix. See `d reported. ### Added +- **`hotato console` + `/calls`: the daily surface over scored production + calls.** `hotato console --production-db DB` is serve + the production + evidence database + the score-on-arrival worker in one command, one process, + one browser tab, opening on the new `/calls` view (every serve flag and + behavior -- loopback bind, bearer-token auth, append-only audit -- is + identical, and every `hotato serve` flag keeps working). `/calls` is a + browsable feed of the derived score sidecar joined read-only with the + session metadata: evidence time and arrival stamp (each labeled with its + clock), evidence-derived duration, session state, score state, the worst + dimension with the measured failure-reason sentence, per-call hop-latency + p50/p95 with the declared authorities behind it, and evidence-lane + completeness. `SCORED`, `NOT_SCORABLE` (with its reason), and `ERROR` are + first-class rows -- a refusal or crash is shown, never hidden and never + rendered as OK. Filters (session state, scorability, since/until window) + are query params; pagination is keyset from the first page (no OFFSET); a + trends strip reports volume, the score-state split, the candidate-moment + share, and per-kind hop-latency p50/p95 over the filtered window (kinds are + never pooled into one number). `/calls/` renders one call: per-dimension + observations (never blended), the ranked candidate moments with their + measured magnitudes and plain-English timing sentences, the timing + waterfall with derived-vs-event-reported values kept apart, every hop row + with its declared authority, scorer version + config hash, the session's + evidence lanes, and the local audio path exactly as the evidence recorded + it. Both views have `?format=json` mirrors built from the same model; the + feed's mirror carries an ETag, answers a matching `If-None-Match` with a + body-less 304, and the page's small inline script polls it every few + seconds to show "updated Ns ago" and re-render rows only on change -- no + external code, and the page is complete as served with JavaScript off. - **`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 diff --git a/docs/WORKSPACE.md b/docs/WORKSPACE.md index c7914ef..2e2caab 100644 --- a/docs/WORKSPACE.md +++ b/docs/WORKSPACE.md @@ -1,8 +1,8 @@ -# The team workspace: `hotato serve` +# The team workspace: `hotato serve` and `hotato console` A self-hosted, local web app for a team to read a voice agent's -conversation-QA state: release readiness, the scenario matrix, a -conversation inspector, failure clusters, and production health. +conversation-QA state: the live call feed, release readiness, the scenario +matrix, a conversation inspector, failure clusters, and production health. Stdlib-only (`http.server` + `sqlite3`) -- no framework, no build step, nothing that phones home. It serves the same fleet registry and evidence store the CLI writes, reading that data directly instead of passing a @@ -45,13 +45,30 @@ redirects to strip the token from the address bar (see [Auth](#auth)). Exit codes: `0` clean shutdown (Ctrl-C); `2` usage error (unusable registry or token, or the port was unavailable). -## The five views +## `hotato console`: the call console in one command + +``` +hotato console --production-db .hotato/production.sqlite3 +``` + +One command, one process, one browser tab: `serve` with the production +evidence database wired (read-only, `mode=ro`), the score-on-arrival worker +on, and the printed URL landing on the live call feed at `/calls`. Every +serve behavior is identical -- loopback bind by default, bearer-token auth on +every request, the append-only audit log, a `?format=json` mirror on every +view -- and `--workspace`, `--host`, `--port`, `--registry`, `--token`, +`--token-file`, and `--no-open` work exactly as they do on `serve`. The same +exit codes apply. + +## The views Every view has a machine mirror at `?format=json` (same auth, same data) so agents and scripts can drive the workspace without scraping HTML. | View | URL | Shows | |---|---|---| +| **Calls** | `/calls` | The console feed over scored production calls (see [The call feed](#the-call-feed-calls)). | +| **One call** | `/calls/` | One call's derived score record: per-dimension observations, ranked candidate moments, the timing waterfall, evidence lanes, and the audio path as recorded. | | **Release readiness** | `/` | Pre-ship home screen: per-release rollup of suites/runs/evaluations -- required-suite completion, scenario/run counts, **failures by dimension** (outcome / policy / conversation / speech / reliability), inconclusive count, real-vs-simulated split, and **new-vs-fixed since the previous release**. Small samples flagged (`low sample, N=3`), never smoothed. | | **Scenario matrix** | `/scenarios` | Rows are scenarios, columns are the current and previous release, with a per-dimension status and **reliability** (`pass^k` where a scenario has repetitions). Filterable by `agent`, `release`, `suite`, `status`. | | **Conversation inspector** | `/conversation/` | One conversation: evidence manifest, transcript, trace spans, per-dimension evaluations with rationale and citations (deterministic checks and model-judged/advisory results in **separate lanes**), reviewer decisions. Every digest links to the raw evidence (`/evidence/`); redacted transcript segments and trace spans render `[redacted]`, in both HTML and JSON. | @@ -124,6 +141,42 @@ 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. +### The call feed (`/calls`) + +`/calls` renders the score sidecar joined read-only with the evidence +database's session metadata, newest arrival first. Each row shows the +evidence-clock timestamp beside the arrival stamp (each labeled with its +clock), the evidence-derived duration, the session state, the score state, +the worst dimension with the measured failure-reason sentence, per-call +hop-latency p50/p95 with the declared authorities behind it, and +evidence-lane completeness (missing required lanes named). `SCORED`, +`NOT_SCORABLE` (with its reason), and `ERROR` are all first-class rows: a +refused or crashed session is shown, never hidden and never rendered as OK. + +Filters are query params -- `state` (session state), `scorability` +(`SCORED`/`NOT_SCORABLE`/`ERROR`), and a `since`/`until` window (epoch +seconds or RFC3339); a malformed filter or cursor is a 400, never a silently +dropped filter. Pagination is keyset (`cursor`, `limit`) -- every page is one +bounded query, and the trends strip above the table reports call volume, the +score-state split, the candidate-moment share, and per-kind hop-latency +p50/p95 (nearest-rank) over the whole filtered window; hop kinds keep their +own percentiles and state their authorities. + +The feed's JSON mirror carries an `ETag` and answers a matching +`If-None-Match` with a body-less 304. The page's small inline script polls +that mirror every few seconds, shows an "updated Ns ago" indicator, and +re-renders the rows only when the ETag changes -- no external code, and with +JavaScript off the page is complete as served (reload for the latest). + +`/calls/` is one call: per-dimension observations (candidate counts and +worst measured magnitude per scan kind, each on its own), the ranked +candidate moments with their measured magnitudes and plain-English timing +sentences, the timing waterfall (turn spans with derived durations kept apart +from event-reported values; every hop row with its timestamp, latency, and +declared authority), scorer version + config hash, the session's evidence +lanes, a link to the production evidence plane, and the local audio path +exactly as the evidence recorded it -- the recording stays on this machine. + ## Auth Every request is authenticated against one shared **bearer token**: diff --git a/llms-full.txt b/llms-full.txt index f2808f6..7ecfeef 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -4589,11 +4589,11 @@ all per-scenario status changes: FILE: docs/WORKSPACE.md ================================================================================ -# The team workspace: `hotato serve` +# The team workspace: `hotato serve` and `hotato console` A self-hosted, local web app for a team to read a voice agent's -conversation-QA state: release readiness, the scenario matrix, a -conversation inspector, failure clusters, and production health. +conversation-QA state: the live call feed, release readiness, the scenario +matrix, a conversation inspector, failure clusters, and production health. Stdlib-only (`http.server` + `sqlite3`) -- no framework, no build step, nothing that phones home. It serves the same fleet registry and evidence store the CLI writes, reading that data directly instead of passing a @@ -4636,13 +4636,30 @@ redirects to strip the token from the address bar (see [Auth](#auth)). Exit codes: `0` clean shutdown (Ctrl-C); `2` usage error (unusable registry or token, or the port was unavailable). -## The five views +## `hotato console`: the call console in one command + +``` +hotato console --production-db .hotato/production.sqlite3 +``` + +One command, one process, one browser tab: `serve` with the production +evidence database wired (read-only, `mode=ro`), the score-on-arrival worker +on, and the printed URL landing on the live call feed at `/calls`. Every +serve behavior is identical -- loopback bind by default, bearer-token auth on +every request, the append-only audit log, a `?format=json` mirror on every +view -- and `--workspace`, `--host`, `--port`, `--registry`, `--token`, +`--token-file`, and `--no-open` work exactly as they do on `serve`. The same +exit codes apply. + +## The views Every view has a machine mirror at `?format=json` (same auth, same data) so agents and scripts can drive the workspace without scraping HTML. | View | URL | Shows | |---|---|---| +| **Calls** | `/calls` | The console feed over scored production calls (see [The call feed](#the-call-feed-calls)). | +| **One call** | `/calls/` | One call's derived score record: per-dimension observations, ranked candidate moments, the timing waterfall, evidence lanes, and the audio path as recorded. | | **Release readiness** | `/` | Pre-ship home screen: per-release rollup of suites/runs/evaluations -- required-suite completion, scenario/run counts, **failures by dimension** (outcome / policy / conversation / speech / reliability), inconclusive count, real-vs-simulated split, and **new-vs-fixed since the previous release**. Small samples flagged (`low sample, N=3`), never smoothed. | | **Scenario matrix** | `/scenarios` | Rows are scenarios, columns are the current and previous release, with a per-dimension status and **reliability** (`pass^k` where a scenario has repetitions). Filterable by `agent`, `release`, `suite`, `status`. | | **Conversation inspector** | `/conversation/` | One conversation: evidence manifest, transcript, trace spans, per-dimension evaluations with rationale and citations (deterministic checks and model-judged/advisory results in **separate lanes**), reviewer decisions. Every digest links to the raw evidence (`/evidence/`); redacted transcript segments and trace spans render `[redacted]`, in both HTML and JSON. | @@ -4715,6 +4732,42 @@ 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. +### The call feed (`/calls`) + +`/calls` renders the score sidecar joined read-only with the evidence +database's session metadata, newest arrival first. Each row shows the +evidence-clock timestamp beside the arrival stamp (each labeled with its +clock), the evidence-derived duration, the session state, the score state, +the worst dimension with the measured failure-reason sentence, per-call +hop-latency p50/p95 with the declared authorities behind it, and +evidence-lane completeness (missing required lanes named). `SCORED`, +`NOT_SCORABLE` (with its reason), and `ERROR` are all first-class rows: a +refused or crashed session is shown, never hidden and never rendered as OK. + +Filters are query params -- `state` (session state), `scorability` +(`SCORED`/`NOT_SCORABLE`/`ERROR`), and a `since`/`until` window (epoch +seconds or RFC3339); a malformed filter or cursor is a 400, never a silently +dropped filter. Pagination is keyset (`cursor`, `limit`) -- every page is one +bounded query, and the trends strip above the table reports call volume, the +score-state split, the candidate-moment share, and per-kind hop-latency +p50/p95 (nearest-rank) over the whole filtered window; hop kinds keep their +own percentiles and state their authorities. + +The feed's JSON mirror carries an `ETag` and answers a matching +`If-None-Match` with a body-less 304. The page's small inline script polls +that mirror every few seconds, shows an "updated Ns ago" indicator, and +re-renders the rows only when the ETag changes -- no external code, and with +JavaScript off the page is complete as served (reload for the latest). + +`/calls/` is one call: per-dimension observations (candidate counts and +worst measured magnitude per scan kind, each on its own), the ranked +candidate moments with their measured magnitudes and plain-English timing +sentences, the timing waterfall (turn spans with derived durations kept apart +from event-reported values; every hop row with its timestamp, latency, and +declared authority), scorer version + config hash, the session's evidence +lanes, a link to the production evidence plane, and the local audio path +exactly as the evidence recorded it -- the recording stays on this machine. + ## Auth Every request is authenticated against one shared **bearer token**: diff --git a/src/hotato/cli.py b/src/hotato/cli.py index b31a920..8817cf4 100644 --- a/src/hotato/cli.py +++ b/src/hotato/cli.py @@ -155,6 +155,11 @@ def _atomic_write_json(path: str, obj) -> None: (2, "usage error: an unusable registry/token, or the port was " "unavailable"), ), + "console": ( + (0, "the console served and shut down cleanly (Ctrl-C)"), + (2, "usage error: an unusable production database, registry, or " + "token, or the port was unavailable"), + ), "report": ( (0, "report written, every scorable event passed"), (1, "a scorable event failed"), @@ -5294,6 +5299,21 @@ def _cmd_serve(args) -> int: ) +def _cmd_console(args) -> int: + # console = serve with the production evidence db wired and the score-on- + # arrival worker on, landing on the live call feed. One command, one + # process, one browser tab; every serve behavior (bind, auth, audit, JSON + # mirrors) is identical. + from . import serve as _serve + + 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=True, landing="/calls", + ) + + # The canonical core loop: one linear path, first touch to a CI gate. This is the # single source of truth rendered identically everywhere a newcomer meets hotato: # the GET STARTED block atop `hotato --help`, `hotato describe`'s core_loop, the @@ -10350,6 +10370,55 @@ def _fleet_parser(parent, name, dotted, help_text): "$HOTATO_NO_BROWSER is set)") srv.set_defaults(func=_cmd_serve) + con = sub.add_parser( + "console", + help="the call console: serve + score-on-arrival over the production " + "evidence database, landing on the live call feed", + description=( + "One command, one process, one browser tab: run the workspace " + "server with the production evidence database wired (read-only, " + "mode=ro) and the score-on-arrival worker on, opening on the live " + "call feed at /calls. Completed sessions are scored one at a time " + "with the deterministic scorer into a console.sqlite3 sidecar " + "beside the evidence database; SCORED, NOT_SCORABLE (with its " + "reason), and ERROR are all first-class visible states. Every " + "serve behavior is identical: loopback bind by default, " + "bearer-token auth on every request, append-only audit, and a " + "?format=json mirror on every view."), + epilog=_exit_codes_epilog("console"), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + con.add_argument( + "--production-db", + required=True, + metavar="PATH", + help=( + "the production evidence SQLite database to read (mode=ro) and " + "score; the console.sqlite3 sidecar of derived score records " + "lives beside it" + ), + ) + con.add_argument("--workspace", "-w", default="default", metavar="ID", + help="workspace id to serve (default 'default')") + con.add_argument("--host", default="127.0.0.1", metavar="HOST", + help="bind address (default 127.0.0.1; a non-loopback " + "host exposes the workspace to your network and " + "prints a warning)") + con.add_argument("--port", type=int, default=8321, metavar="PORT", + help="listen port (default 8321)") + con.add_argument("--registry", default=None, metavar="PATH", + help="registry home directory (default ~/.hotato/fleet)") + con.add_argument("--token", default=None, metavar="TOKEN", + help="bearer token (default: reuse the stored one, else " + "generate + store 0600 under the workspace state dir)") + con.add_argument("--token-file", default=None, metavar="PATH", + help="read the bearer token from PATH instead of --token") + con.add_argument("--no-open", action="store_true", + help="do not open a browser on start (auto-open is also " + "skipped when stdout is not a TTY, in CI, or when " + "$HOTATO_NO_BROWSER is set)") + con.set_defaults(func=_cmd_console) + # Operational families live in a standalone registrar so this integration # remains a one-line semantic merge as the main CLI evolves. from . import ops_cli as _ops_cli diff --git a/src/hotato/serve/app.py b/src/hotato/serve/app.py index 588dffe..e68c0fb 100644 --- a/src/hotato/serve/app.py +++ b/src/hotato/serve/app.py @@ -27,6 +27,7 @@ """ from __future__ import annotations +import hashlib import http.cookies import json import os @@ -292,6 +293,11 @@ def _route(self, ctx, path, query, fmt): return self._view(ctx, "clusters", query, fmt) if path in ("/health", "/production"): return self._view(ctx, "health", query, fmt) + if path == "/calls": + return self._calls_feed(ctx, query, fmt) + if path.startswith("/calls/"): + subject = unquote(path.split("/", 2)[2]) if path.count("/") >= 2 else "" + return self._call_detail(ctx, subject, fmt) if path == "/records": return self._records_list(ctx, fmt) if path.startswith("/records/"): @@ -358,6 +364,63 @@ def _inspector(self, ctx, conversation_id, fmt): workspace=ctx.workspace) return 200, doc.encode("utf-8"), "text/html; charset=utf-8", {} + def _calls_feed(self, ctx, query, fmt): + """The console call feed (R5) with its trends strip (R8) and the cheap + content hash the polling refresh revalidates against (R7): the model's + canonical JSON bytes are hashed into an ETag; a request presenting the + same ``If-None-Match`` gets a body-less 304, so the every-few-seconds + poll costs one bounded read and no transfer while nothing changed. A + malformed filter/cursor/limit is a 400, never a silently dropped + filter.""" + try: + model = _data.build_calls_feed( + ctx.workspace, ctx.production_db, + state=_q(query, "state"), scorability=_q(query, "scorability"), + since=_q(query, "since"), until=_q(query, "until"), + cursor=_q(query, "cursor"), limit=_q(query, "limit")) + except ValueError as exc: + return self._bad_request(ctx, fmt, str(exc)) + body = _json_bytes(model) + etag = '"%s"' % hashlib.sha256(body).hexdigest()[:32] + if self.headers.get("If-None-Match") == etag: + return 304, b"", "text/plain; charset=utf-8", {"ETag": etag} + if fmt == "json": + return 200, body, "application/json; charset=utf-8", {"ETag": etag} + doc = _render.page("Calls", "/calls", + _render.render_calls_feed(model, etag=etag), + workspace=ctx.workspace) + return 200, doc.encode("utf-8"), "text/html; charset=utf-8", {"ETag": etag} + + def _call_detail(self, ctx, subject, fmt): + """One call's derived record (R6). The JSON mirror returns the same + model the HTML renders; an unknown subject (or a console that is not + wired) is a 404, and an unusable sidecar (schema mismatch) is a 400 + carrying the rebuild instruction -- fail closed, never render wrong + derived data.""" + try: + model = _data.build_call_detail(ctx.workspace, ctx.production_db, + subject) + except ValueError as exc: + return self._bad_request(ctx, fmt, str(exc)) + if model is None: + return self._not_found( + ctx, fmt, "No scored call %r in the console sidecar." % subject, + extra_json={"subject": subject}) + if fmt == "json": + return 200, _json_bytes(model), "application/json; charset=utf-8", {} + doc = _render.page("Call", "/calls", + _render.render_call_detail(model), + workspace=ctx.workspace) + return 200, doc.encode("utf-8"), "text/html; charset=utf-8", {} + + def _bad_request(self, ctx, fmt, message): + if fmt == "json": + return 400, _json_bytes({"error": "bad request", "message": message}), \ + "application/json; charset=utf-8", {} + doc = _render.page("Bad request", "/calls", + _render.render_400(message), workspace=ctx.workspace) + return 400, doc.encode("utf-8"), "text/html; charset=utf-8", {} + def _records_list(self, ctx, fmt): """The read-only Failure Record index. Reads ``/records`` fresh (never writes it); an absent/empty directory renders an explicit empty @@ -523,7 +586,8 @@ 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, - score_production: bool = False) -> int: + score_production: bool = False, + landing: str = "/") -> 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. @@ -531,7 +595,8 @@ def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, On start, unless ``open_browser`` is false (or the environment says not to; see :func:`_maybe_open_browser`), the default browser is pointed at the tokenised URL so the first thing a new user sees is a working workspace, not - an auth wall.""" + an auth wall. ``landing`` is the path that tokenised URL opens (every view + stays reachable either way): ``hotato console`` lands on ``/calls``.""" home = os.path.abspath(os.path.expanduser(registry)) if registry else DEFAULT_HOME # Validate the registry opens (schema-version gate); read-only thereafter. @@ -576,10 +641,14 @@ def run_serve(*, workspace: str, host: str = "127.0.0.1", port: int = 8321, score_store = ConsoleStore(default_console_path(resolved_production_db)) score_worker = ConsoleScoreWorker(resolved_production_db, score_store) + if not landing.startswith("/"): + raise ValueError("landing must be an absolute path like '/calls'") + 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. - url = "http://%s:%d/?token=%s" % (display_host, port, quote(tok, safe="")) + url = "http://%s:%d%s?token=%s" % (display_host, port, landing, + quote(tok, safe="")) _print_banner(ctx, host, port, source=source, generated=generated, state_dir=state_dir, url=url, display_host=display_host) diff --git a/src/hotato/serve/data.py b/src/hotato/serve/data.py index fce5dc0..d9963cf 100644 --- a/src/hotato/serve/data.py +++ b/src/hotato/serve/data.py @@ -29,9 +29,13 @@ import json import os import re +import sqlite3 +import stat from collections import OrderedDict -from typing import Any, Dict, List, Optional +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple +from .._stats import nearest_rank as _nearest_rank from ..errors import open_regular as _open_regular from ..failure_record import KIND as _FR_KIND from ..failure_record import LANES as _FR_LANES @@ -52,6 +56,8 @@ "build_production_health", "build_records_list", "build_record_detail", + "build_calls_feed", + "build_call_detail", ] # The five conversation-QA dimensions (GPT Pro ยง2; never blended into one score). @@ -962,6 +968,558 @@ def build_records_list(home: str, ws: str) -> Dict[str, Any]: "records": records, "record_count": len(records)} +# ========================================================================= +# View 7 -- Calls (the console feed over the derived score sidecar) +# ========================================================================= + +# The feed page is keyset-paginated from day one (I3): every page is one +# LIMIT-bounded query ordered by (last_event DESC, subject ASC); there is no +# OFFSET anywhere on this path. +_CALLS_DEFAULT_LIMIT = 50 +_CALLS_MAX_LIMIT = 200 +# The two evidence-clock provenance labels every feed timestamp carries (I5): +# a session's arrival stamp is the receiver's clock, never an evidence +# measurement; the per-hop ``at`` values come from evidence event timestamps. +_ARRIVAL_AUTHORITY = "received:arrival_clock" +_EVENT_AUTHORITY = "derived:event_timestamps" + + +def _console_paths(production_db: str) -> Tuple[str, str]: + """``(production_path, console_path)`` for a selected evidence database.""" + from ..console_worker import default_console_path + + resolved = os.path.abspath(os.path.expanduser(production_db)) + return resolved, default_console_path(resolved) + + +def _open_console_ro(console_path: str, production_path: str) -> sqlite3.Connection: + """Open the console sidecar READ-ONLY with the production evidence database + ATTACHed read-only, so the calls feed is one SQL join across the two local + sources with zero write capability. + + Discipline mirrors :mod:`hotato.serve.production_bridge`: both files must be + regular files, both are opened ``mode=ro``, ``query_only`` is set, and the + bridge's metadata-only authorizer is installed AFTER the attach -- it denies + every write action (including any further ``ATTACH``) and refuses to read + the production ``events.payload_json`` column, so this view can never touch + payloads even by a future query mistake. A sidecar or evidence database + with a different schema version is refused (``ValueError``) with the + rebuild instruction rather than rendered wrong. + """ + from .production_bridge import _metadata_only_authorizer + + for label, path in (("console sidecar", console_path), + ("production database", production_path)): + try: + mode = os.lstat(path).st_mode + except OSError as exc: + raise ValueError(f"{label} is not readable: {path!r} ({exc})") from exc + if not stat.S_ISREG(mode): + raise ValueError(f"{label} must be a regular file: {path!r}") + uri = Path(console_path).as_uri() + "?mode=ro" + try: + db = sqlite3.connect(uri, uri=True, timeout=5, isolation_level=None) + except sqlite3.Error as exc: + raise ValueError(f"could not open console sidecar read-only: {exc}") from exc + db.row_factory = sqlite3.Row + try: + db.execute( + "ATTACH DATABASE ? AS evidence", + (Path(production_path).as_uri() + "?mode=ro",), + ) + db.execute("PRAGMA query_only=ON") + db.execute("PRAGMA busy_timeout=5000") + db.set_authorizer(_metadata_only_authorizer) + _verify_console_schema(db) + except BaseException: + db.close() + raise + return db + + +def _verify_console_schema(db: sqlite3.Connection) -> None: + from ..console_store import SCHEMA_VERSION as _CONSOLE_SCHEMA_VERSION + + try: + row = db.execute( + "SELECT value FROM metadata WHERE key='console_schema_version'" + ).fetchone() + except sqlite3.Error as exc: + raise ValueError("selected file is not a hotato console sidecar") from exc + if row is None or row[0] != _CONSOLE_SCHEMA_VERSION: + observed = None if row is None else row[0] + raise ValueError( + "console sidecar schema version " + repr(observed) + " does not " + "match " + repr(_CONSOLE_SCHEMA_VERSION) + "; the sidecar is " + "derived data -- regenerate it with hotato serve --production-db " + "DB --rebuild-scores" + ) + try: + row = db.execute( + "SELECT value FROM evidence.metadata " + "WHERE key='production_schema_version'" + ).fetchone() + except sqlite3.Error as exc: + raise ValueError( + "selected database is not a hotato production evidence store" + ) from exc + if row is None or row[0] != "1": + observed = None if row is None else row[0] + raise ValueError( + "unsupported production database schema version: " + repr(observed) + ) + + +def _window_epoch(value: Optional[str], label: str) -> Optional[float]: + """A time-window bound as epoch seconds. Accepts epoch seconds or an + RFC3339 timestamp; anything else is a refused usage error (400), never a + silently ignored filter.""" + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + pass + from ..production import _parse_rfc3339 + + try: + return _parse_rfc3339(value).timestamp() + except (TypeError, ValueError): + raise ValueError( + f"{label} must be epoch seconds or an RFC3339 timestamp, " + f"got {value!r}" + ) from None + + +def _decode_calls_cursor(cursor: Optional[str]) -> Optional[Tuple[float, str]]: + if cursor is None: + return None + head, sep, subject = cursor.partition("~") + if not sep: + raise ValueError("invalid cursor: expected '~'") + try: + return float(head), subject + except ValueError: + raise ValueError("invalid cursor: expected '~'") from None + + +def _encode_calls_cursor(last_event: float, subject: str) -> str: + return f"{last_event!r}~{subject}" + + +def _worst_dimension(dimensions: Any) -> Optional[Dict[str, Any]]: + """The dimension with the largest measured magnitude -- a per-call worst + case for the feed headline, NOT a blended score (each dimension keeps its + own observation; nothing is averaged or weighted).""" + best = None + for kind in sorted(dimensions or {}): + item = dimensions[kind] + if not isinstance(item, dict): + continue + worst = item.get("worst_sec") + if isinstance(worst, bool) or not isinstance(worst, (int, float)): + continue + if best is None or worst > best["worst_sec"]: + best = {"dimension": kind, "worst_sec": worst} + return best + + +def _candidate_total(dimensions: Any) -> int: + total = 0 + for item in (dimensions or {}).values(): + if isinstance(item, dict) and isinstance(item.get("candidate_count"), int): + total += item["candidate_count"] + return total + + +def _round3(value: Optional[float]) -> Optional[float]: + return None if value is None else round(value, 3) + + +def _call_latency(db: sqlite3.Connection, subject: str) -> Dict[str, Any]: + """This call's per-hop latency summary from the sidecar's hop rows (PK- + indexed by subject): nearest-rank p50/p95 over the hops that carry a + latency, the evidence-clock timestamp of the first hop, and the set of + declared authorities so a vendor-reported figure never hides behind the + percentile (I5).""" + rows = db.execute( + "SELECT at,latency_ms,authority FROM latency_hops WHERE subject=? " + "ORDER BY seq", + (subject,), + ).fetchall() + values = [ + float(r["latency_ms"]) for r in rows if r["latency_ms"] is not None + ] + return { + "hop_count": len(values), + "p50_ms": _round3(_nearest_rank(values, 0.50)), + "p95_ms": _round3(_nearest_rank(values, 0.95)), + "authorities": sorted({r["authority"] for r in rows}), + "first_event_at": rows[0]["at"] if rows else None, + "method": "nearest-rank", + } + + +def _lane_completeness(evidence_json: Any, required_json: Any) -> Dict[str, Any]: + """Evidence-lane completeness for one session, through the bridge's own + strict lane parser. A malformed manifest is a visible ``invalid`` marker on + that row -- refused loudly, never smoothed into a completeness number.""" + from .production_bridge import ProductionBridgeError, _evidence, _required_lanes + + try: + lanes = _evidence(evidence_json) + required = _required_lanes(required_json) + except ProductionBridgeError as exc: + return {"status": "invalid", "detail": str(exc)} + missing = [ + lane for lane in required if lanes[lane]["availability"] != "available" + ] + return { + "status": "complete" if not missing else "incomplete", + "required": len(required), + "available_required": len(required) - len(missing), + "missing": missing, + } + + +def _calls_feed_shell(ws: str, filters: Dict[str, Any]) -> Dict[str, Any]: + return { + "view": "calls_feed", + "workspace": ws, + "configured": False, + "source": None, + "sidecar": {"path": None, "present": False}, + "filters": filters, + "trends": None, + "rows": [], + "row_count": 0, + "page": {"limit": filters["limit"], "cursor": filters["cursor"], + "next_cursor": None}, + } + + +def build_calls_feed(ws: str, production_db: Optional[str], *, + state: Optional[str] = None, + scorability: Optional[str] = None, + since: Optional[str] = None, + until: Optional[str] = None, + cursor: Optional[str] = None, + limit: Optional[str] = None) -> Dict[str, Any]: + """The console call feed: the derived score sidecar (read-only) joined with + the production evidence plane's session metadata, newest arrival first. + + SCORED, NOT_SCORABLE, and ERROR are all first-class rows -- a refused or + errored session is never hidden and never shown as OK (I2). Filters (session + ``state``, ``scorability``, ``since``/``until`` window) and the keyset + ``cursor`` come from query params; a malformed filter or cursor raises + ``ValueError`` (a 400, never a silently ignored filter). The trends block is + computed over the whole filtered window with the same bounded queries. + """ + try: + page_limit = int(limit) if limit is not None else _CALLS_DEFAULT_LIMIT + except (TypeError, ValueError): + raise ValueError(f"limit must be an integer, got {limit!r}") from None + if not 1 <= page_limit <= _CALLS_MAX_LIMIT: + raise ValueError(f"limit must be 1..{_CALLS_MAX_LIMIT}") + state = state.upper() if state else None + scorability = scorability.upper() if scorability else None + since_epoch = _window_epoch(since, "since") + until_epoch = _window_epoch(until, "until") + after = _decode_calls_cursor(cursor) + filters = {"state": state, "scorability": scorability, "since": since, + "until": until, "cursor": cursor, "limit": page_limit} + + model = _calls_feed_shell(ws, filters) + if not production_db: + return model + production_path, console_path = _console_paths(production_db) + model["configured"] = True + model["source"] = { + "production_db": production_path, + "console_db": console_path, + "access": "sqlite-mode-ro", + } + model["sidecar"] = {"path": console_path, + "present": os.path.isfile(console_path)} + if not model["sidecar"]["present"]: + return model + + where = ["e.state != 'DELETED'"] + params: List[Any] = [] + if state: + where.append("e.state = ?") + params.append(state) + if scorability: + where.append("s.state = ?") + params.append(scorability) + if since_epoch is not None: + where.append("e.last_event >= ?") + params.append(since_epoch) + if until_epoch is not None: + where.append("e.last_event <= ?") + params.append(until_epoch) + window_where, window_params = list(where), list(params) + if after is not None: + where.append( + "(e.last_event < ? OR (e.last_event = ? AND s.subject > ?))" + ) + params.extend([after[0], after[0], after[1]]) + + db = _open_console_ro(console_path, production_path) + try: + picked = db.execute( + "SELECT s.subject AS subject, s.state AS score_state, s.reason AS " + "reason, s.session_state AS scored_session_state, s.event_count AS " + "event_count, s.scorer_version AS scorer_version, s.config_sha256 " + "AS config_sha256, s.dimensions_json AS dimensions_json, " + "s.timing_json AS timing_json, e.state AS live_state, e.started AS " + "started, e.ended AS ended, e.last_event AS last_event, " + "e.evidence_json AS evidence_json, e.required_evidence_json AS " + "required_evidence_json FROM scores s JOIN evidence.sessions e ON " + "e.subject = s.subject WHERE " + " AND ".join(where) + + " ORDER BY e.last_event DESC, s.subject ASC LIMIT ?", + (*params, page_limit + 1), + ).fetchall() + has_more = len(picked) > page_limit + picked = picked[:page_limit] + model["rows"] = [_calls_feed_row(db, row) for row in picked] + model["row_count"] = len(model["rows"]) + if has_more and picked: + tail = picked[-1] + model["page"]["next_cursor"] = _encode_calls_cursor( + float(tail["last_event"]), tail["subject"]) + model["trends"] = _calls_trends(db, window_where, window_params) + finally: + db.close() + return model + + +def _calls_feed_row(db: sqlite3.Connection, row: sqlite3.Row) -> Dict[str, Any]: + dimensions = _parse_json_field(row["dimensions_json"]) + timing = _parse_json_field(row["timing_json"]) + duration_ms = None + if isinstance(timing, dict): + end_to_end = timing.get("end_to_end_ms") + if isinstance(end_to_end, (int, float)) and not isinstance(end_to_end, bool): + duration_ms = float(end_to_end) + latency = _call_latency(db, row["subject"]) + return { + "subject": row["subject"], + "score_state": row["score_state"], + "reason": row["reason"], + "session_state": row["live_state"], + "scored_session_state": row["scored_session_state"], + "event_count": row["event_count"], + "scorer_version": row["scorer_version"], + "config_sha256": row["config_sha256"], + # evidence-clock anchor (first hop timestamp) beside the arrival stamp, + # each labeled with its clock so neither masquerades as the other (I5) + "evidence_time_at": latency["first_event_at"], + "evidence_time_authority": _EVENT_AUTHORITY, + "received_at": row["last_event"], + "received_authority": _ARRIVAL_AUTHORITY, + "duration_ms": duration_ms, + "duration_authority": _EVENT_AUTHORITY if duration_ms is not None else None, + "worst": _worst_dimension(dimensions), + "candidate_total": _candidate_total(dimensions), + "latency": latency, + "evidence": _lane_completeness(row["evidence_json"], + row["required_evidence_json"]), + } + + +def _calls_trends(db: sqlite3.Connection, window_where: List[str], + window_params: List[Any]) -> Dict[str, Any]: + """R8: volume, score-state split, candidate-moment share, and per-kind hop + latency p50/p95 over the FILTERED window. Kinds are never pooled into one + number (a model-hop latency and a turn duration are different measurements) + and each kind states the declared authorities behind its values (I5).""" + from ..console_store import SCORE_STATES + + joined = " AND ".join(window_where) + states = {s: 0 for s in SCORE_STATES} + for state_row in db.execute( + "SELECT s.state, COUNT(*) FROM scores s JOIN evidence.sessions e ON " + "e.subject = s.subject WHERE " + joined + " GROUP BY s.state", + window_params, + ): + states[state_row[0]] = int(state_row[1]) + volume = sum(states.values()) + with_candidates = int(db.execute( + "SELECT COUNT(*) FROM scores s JOIN evidence.sessions e ON " + "e.subject = s.subject WHERE " + joined + + " AND s.state = 'SCORED' AND s.reason IS NOT NULL", + window_params, + ).fetchone()[0]) + by_kind: Dict[str, Dict[str, Any]] = {} + for hop in db.execute( + "SELECT h.kind, h.latency_ms, h.authority FROM latency_hops h " + "JOIN scores s ON s.subject = h.subject " + "JOIN evidence.sessions e ON e.subject = h.subject " + "WHERE " + joined + " AND h.latency_ms IS NOT NULL", + window_params, + ): + slot = by_kind.setdefault(hop[0], {"values": [], "authorities": set()}) + slot["values"].append(float(hop[1])) + slot["authorities"].add(hop[2]) + hop_latency = { + kind: { + "count": len(slot["values"]), + "p50_ms": _round3(_nearest_rank(slot["values"], 0.50)), + "p95_ms": _round3(_nearest_rank(slot["values"], 0.95)), + "authorities": sorted(slot["authorities"]), + } + for kind, slot in sorted(by_kind.items()) + } + scored = states.get("SCORED", 0) + return { + "volume": volume, + "states": states, + "with_candidate_moments": with_candidates, + "candidate_share": (with_candidates / scored) if scored else None, + "hop_latency_ms": hop_latency, + "method": "nearest-rank", + "provenance": ( + "computed over the filtered window from the sidecar's evidence-" + "derived rows; each hop kind keeps its own percentile and states " + "its declared authorities" + ), + } + + +def build_call_detail(ws: str, production_db: Optional[str], + subject: str) -> Optional[Dict[str, Any]]: + """One call's full derived record: per-dimension observations (never + blended), the ranked candidate moments with their measured magnitudes, the + timing waterfall (turn spans + per-hop rows, each carrying its declared + authority -- derived vs reported is rendered, never flattened), scorer + version + config hash, the session's evidence-lane completeness, and the + local audio reference when the evidence names one. Returns ``None`` (a 404) + when the console is not configured, the sidecar is absent, or the subject + has no derived record.""" + if not production_db or not subject: + return None + production_path, console_path = _console_paths(production_db) + if not os.path.isfile(console_path): + return None + db = _open_console_ro(console_path, production_path) + try: + row = db.execute( + "SELECT * FROM scores WHERE subject=?", (subject,) + ).fetchone() + if row is None: + return None + hops = db.execute( + "SELECT seq,kind,at,latency_ms,authority,source,event_id " + "FROM latency_hops WHERE subject=? ORDER BY seq", + (subject,), + ).fetchall() + sess = db.execute( + "SELECT state,started,ended,last_event,finalized_at," + "finalization_reason,event_count,evidence_json," + "required_evidence_json FROM evidence.sessions WHERE subject=?", + (subject,), + ).fetchone() + latency = _call_latency(db, subject) + finally: + db.close() + + dimensions = _parse_json_field(row["dimensions_json"]) + candidates = _parse_json_field(row["candidates_json"]) + score = { + "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": _parse_json_field(row["config_json"]), + "dimensions": dimensions, + "candidates": _ranked_candidates(candidates), + "timing": _parse_json_field(row["timing_json"]), + "audio": _parse_json_field(row["audio_json"]), + "hops": [ + { + "seq": hop["seq"], + "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 + ], + } + + session = None + if sess is not None: + session = { + "state": sess["state"], + "started": sess["started"], + "ended": sess["ended"], + "last_event": sess["last_event"], + "received_authority": _ARRIVAL_AUTHORITY, + "finalized_at": sess["finalized_at"], + "finalization_reason": sess["finalization_reason"], + "event_count": sess["event_count"], + "completeness": _lane_completeness( + sess["evidence_json"], sess["required_evidence_json"]), + } + from .production_bridge import ProductionBridgeError, _evidence, _required_lanes + + try: + session["evidence"] = _evidence(sess["evidence_json"]) + session["required_evidence_lanes"] = _required_lanes( + sess["required_evidence_json"]) + except ProductionBridgeError: + session["evidence"] = None + session["required_evidence_lanes"] = None + + return { + "view": "call_detail", + "workspace": ws, + "subject": subject, + "worst": _worst_dimension(dimensions), + "score": score, + "session": session, + "latency": latency, + "source": { + "production_db": production_path, + "console_db": console_path, + "access": "sqlite-mode-ro", + }, + } + + +def _ranked_candidates(candidates: Any) -> List[Dict[str, Any]]: + """The stored candidate list (already salience-ranked by the scorer) with a + plain-English timing sentence added per candidate -- built only from the + measured numbers the candidate already carries.""" + if not isinstance(candidates, list): + return [] + from ..scan import candidate_plain_english + + out = [] + for item in candidates: + if not isinstance(item, dict): + continue + enriched = dict(item) + try: + enriched["plain_english"] = candidate_plain_english(item) + except Exception: + enriched["plain_english"] = None + out.append(enriched) + return out + + +# ========================================================================= +# View 6 (continued) -- one Failure Record by id +# ========================================================================= + + def build_record_detail(home: str, record_id: str) -> Optional[Dict[str, Any]]: """Resolve ONE Failure Record by its URL-safe ``record_id`` and return the validated canonical record dict, or ``None`` when the id is unsafe, escapes diff --git a/src/hotato/serve/render.py b/src/hotato/serve/render.py index c0d32ef..bc8de79 100644 --- a/src/hotato/serve/render.py +++ b/src/hotato/serve/render.py @@ -19,7 +19,10 @@ import html as _html import json +import time as _time from typing import Any, List, Optional +from urllib.parse import quote as _quote +from urllib.parse import urlencode as _urlencode # --- house style, imported from the report with a same-look fallback --------- try: @@ -154,10 +157,15 @@ def _esc(x) -> str: _STATUS_COLOR = {"PASS": "green", "FAIL": "red", "INCONCLUSIVE": "ember", - "ERROR": "red", "NOT_RUN": "muted", "UNAVAILABLE": "muted"} + "ERROR": "red", "NOT_RUN": "muted", "UNAVAILABLE": "muted", + # console score states: SCORED means the scorer ran, ember for + # a first-class refusal -- NOT_SCORABLE never reads as OK + "SCORED": "green", "NOT_SCORABLE": "ember"} -# The top tabs. The conversation inspector + a single record are drill-ins. +# The top tabs. The conversation inspector, a single call, and a single record +# are drill-ins. _TABS = ( + ("/calls", "Calls"), ("/", "Release readiness"), ("/scenarios", "Scenario matrix"), ("/clusters", "Failure clusters"), @@ -1229,6 +1237,480 @@ def render_record_detail(record: dict) -> str: return "".join(parts) +# ========================================================================= +# View 7 -- Calls (the console feed + one call) +# ========================================================================= + +def _utc_stamp(epoch: Any) -> str: + """An arrival-clock epoch as a compact UTC stamp; a dash when absent.""" + if isinstance(epoch, bool) or not isinstance(epoch, (int, float)): + return "-" + return _time.strftime("%Y-%m-%d %H:%M:%SZ", _time.gmtime(epoch)) + + +def _fmt_ms(value: Any) -> str: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return "-" + return f"{value:.0f} ms" + + +def _fmt_sec(value: Any) -> str: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return "-" + return f"{value:.3f}s" + + +def _call_link(subject: str) -> str: + href = "/calls/" + _quote(str(subject), safe="") + return f'{_esc(subject)}' + + +def _latency_cell(latency: Optional[dict]) -> str: + if not latency or not latency.get("hop_count"): + return '-' + auth = ", ".join(latency.get("authorities") or []) + return (f'p50 {_fmt_ms(latency.get("p50_ms"))} · ' + f'p95 {_fmt_ms(latency.get("p95_ms"))} ' + f'({latency["hop_count"]} hops)' + f'
authority: {_esc(auth)}
') + + +def _evidence_cell(evidence: Optional[dict]) -> str: + if not evidence: + return '-' + if evidence.get("status") == "invalid": + return 'invalid manifest' + have = evidence.get("available_required", 0) + need = evidence.get("required", 0) + missing = evidence.get("missing") or [] + cell = f'{have}/{need} lanes' + if missing: + cell += ('
missing: ' + f'{_esc(", ".join(missing))}
') + return cell + + +def _worst_cell(row: dict) -> str: + parts = [] + worst = row.get("worst") + if worst: + parts.append(f'{_esc(worst["dimension"])} ' + f'{_fmt_sec(worst.get("worst_sec"))}') + if row.get("reason"): + parts.append(f'
{_esc(row["reason"])}
') + return "".join(parts) or '-' + + +def _calls_filters_form(f: dict) -> str: + state_opts = "".join(_opt(v, f.get("state")) for v in + ("", "COMPLETE", "QUIESCENT")) + scor_opts = "".join(_opt(v, f.get("scorability")) for v in + ("", "SCORED", "NOT_SCORABLE", "ERROR")) + return ( + '
' + f'' + f'' + f'' + f'' + '
' + ) + + +def _calls_trends_html(trends: Optional[dict]) -> str: + if not trends: + return "" + states = trends.get("states") or {} + share = trends.get("candidate_share") + share_txt = "-" if share is None else f"{share * 100:.0f}%" + parts = ['
window trends
', + '
', + _kv("calls", str(trends.get("volume", 0))), + _kv("scored", str(states.get("SCORED", 0))), + _kv("not scorable", str(states.get("NOT_SCORABLE", 0))), + _kv("errors", str(states.get("ERROR", 0))), + _kv("with candidate moments", + f'{trends.get("with_candidate_moments", 0)} ({share_txt} of scored)'), + '
'] + hop_latency = trends.get("hop_latency_ms") or {} + if hop_latency: + parts.append('
hop latency by ' + 'kind (nearest-rank, never pooled across kinds)
' + '
') + for kind, item in hop_latency.items(): + auth = ", ".join(item.get("authorities") or []) + parts.append( + f'{_esc(kind)} ' + f'n={item.get("count", 0)} · ' + f'p50 {_fmt_ms(item.get("p50_ms"))} · ' + f'p95 {_fmt_ms(item.get("p95_ms"))}' + f'
authority: {_esc(auth)}' + '
') + parts.append('
') + parts.append(f'
{_esc(trends.get("provenance") or "")}
') + parts.append('
') + return "".join(parts) + + +def _calls_next_link(m: dict) -> str: + page = m.get("page") or {} + nxt = page.get("next_cursor") + if not nxt: + return "" + f = m.get("filters") or {} + pairs = [(k, f[k]) for k in ("state", "scorability", "since", "until") + if f.get(k)] + if page.get("limit") is not None: # keep the page size stable across pages + pairs.append(("limit", page["limit"])) + pairs.append(("cursor", nxt)) + href = "/calls?" + _urlencode(pairs) + return (f'') + + +# Auto-refresh (R7): poll the view's JSON mirror with If-None-Match every few +# seconds; a 304 costs nothing, a changed ETag re-fetches the server-rendered +# page and swaps only the feed region. No external code, no websocket; with +# JavaScript off the page is complete as served and a manual reload refreshes. +_CALLS_POLL_JS = """ + +""" + + +def render_calls_feed(m: dict, *, etag: Optional[str] = None) -> str: + parts = ['

Calls ' + '

', + '

Production call sessions with their derived score ' + 'records: per-dimension observations, the measured failure-reason ' + 'sentence, per-hop latency, and evidence completeness. SCORED, ' + 'NOT_SCORABLE, and ERROR are all shown -- a refusal is a state, ' + 'not a gap.

'] + + if not m.get("configured"): + parts.append('
Point the workspace at a production ' + 'evidence database to see calls here: ' + 'hotato console --production-db DB (or ' + 'hotato serve --production-db DB ' + '--score-production).
') + return "".join(parts) + + src = m.get("source") or {} + parts.append('
evidence ' + f'{_esc(src.get("production_db"))} ' + '· sidecar ' + f'{_esc(src.get("console_db"))} ' + '· sqlite-mode-ro
') + + if not (m.get("sidecar") or {}).get("present"): + parts.append('
No score sidecar beside this evidence ' + 'database yet. Start scoring with hotato ' + 'console --production-db DB, or build it once with ' + 'hotato serve --production-db DB ' + '--rebuild-scores.
') + return "".join(parts) + + parts.append(_calls_filters_form(m.get("filters") or {})) + + data_etag = f' data-etag="{_esc(etag)}"' if etag else "" + parts.append(f'
') + parts.append(_calls_trends_html(m.get("trends"))) + + rows = m.get("rows") or [] + parts.append(f'
' + f'{m.get("row_count", len(rows))} ' + 'calls on this page (newest arrival first).
') + if not rows: + parts.append('
No scored calls in this window. ' + 'Clear the filters, or wait for the next scoring cycle ' + 'to pick up completed sessions.
') + else: + parts.append('
' + '' + '' + '' + '') + for row in rows: + when = row.get("evidence_time_at") + if when: + time_cell = (f'{_esc(when)}' + '
evidence clock
') + else: + time_cell = (f'{_utc_stamp(row.get("received_at"))}' + '
arrival clock
') + duration = row.get("duration_ms") + if duration is None: + dur_cell = '-' + else: + dur_cell = (f'{_fmt_ms(duration)}' + '
derived
') + parts.append( + f'' + f'' + f'' + f'' + f'' + f'' + f'' + f'') + parts.append('
timecalldurationsessionscoreworst dimension · measured reasonlatencyevidence
{time_cell}{_call_link(row["subject"])}{dur_cell}{_production_state_chip(str(row.get("session_state")))}{_status_chip(row.get("score_state"))}{_worst_cell(row)}{_latency_cell(row.get("latency"))}{_evidence_cell(row.get("evidence"))}
') + parts.append(_calls_next_link(m)) + parts.append('
') # /calls-live + parts.append(_CALLS_POLL_JS) + return "".join(parts) + + +def render_call_detail(m: dict) -> str: + score = m.get("score") or {} + session = m.get("session") + parts = [f'

Call {_esc(m.get("subject"))}

', + '

One call\'s derived score record: per-dimension ' + 'observations, ranked candidate moments, the timing waterfall, and ' + 'the evidence behind each number. Every figure states where it ' + 'came from.

'] + + # summary: score state, reason, session state, provenance (I5) + parts.append('
') + parts.append('
' + f'{_status_chip(score.get("state"))} ') + if session is not None: + parts.append(_production_state_chip(str(session.get("state")))) + parts.append('%s events
' + % _esc(score.get("event_count"))) + if score.get("reason"): + parts.append(f'
{_esc(score["reason"])}
') + parts.append('
') + parts.append(_kv("scorer version", str(score.get("scorer_version")), mono=True)) + timing = score.get("timing") or {} + if isinstance(timing.get("end_to_end_ms"), (int, float)): + parts.append(_kv("end to end", _fmt_ms(timing["end_to_end_ms"]) + + " (derived)", mono=True)) + if session is not None: + parts.append(_kv("received", _utc_stamp(session.get("last_event")) + + " (arrival clock)", mono=True)) + parts.append('
') + parts.append('
config ' + f'{_esc(score.get("config_sha256"))}
') + parts.append('
evidence log ' + f'{_esc(score.get("evidence_sha256"))}
') + parts.append('
raw session: ' + 'production evidence plane' + '
') + parts.append('
') + + # per-dimension observations -- each stands on its own, nothing combined + dims = score.get("dimensions") or {} + parts.append('
Per-dimension ' + 'observations (each on its own, never combined)
') + if not dims: + parts.append('
No dimension observations on this ' + 'record.
') + else: + parts.append('
' + '' + '') + for kind in sorted(dims): + item = dims.get(kind) or {} + parts.append( + f'' + f'' + f'') + parts.append('
dimensioncandidate momentsworst measured magnitude
{_esc(kind)}{_esc(item.get("candidate_count", 0))}{_fmt_sec(item.get("worst_sec"))}
') + parts.append('
') + + # ranked candidate moments with measured magnitudes + candidates = score.get("candidates") or [] + parts.append('
Ranked candidate ' + 'moments
') + if not candidates: + parts.append('
No candidate moments on this record.' + '
') + else: + parts.append('
' + '' + '') + for rank, cand in enumerate(candidates, start=1): + durations = cand.get("durations") or {} + measured = " · ".join( + f'{_esc(k)} {_fmt_sec(v)}' + for k, v in durations.items()) or '-' + sentence = cand.get("plain_english") + sentence_html = (_esc(sentence) if sentence + else '-') + parts.append( + f'' + f'' + f'' + f'' + f'') + parts.append('
#kindatmeasuredwhat was measured
{rank}{_esc(cand.get("kind"))}{_fmt_sec(cand.get("t_sec"))}{measured}{sentence_html}
') + parts.append('
') + + # timing waterfall: turn spans (derived vs reported kept apart) + hops + parts.append(_call_timing_waterfall(timing, score.get("hops") or [], + m.get("latency"))) + + # evidence lanes for this session + parts.append(_call_evidence_section(session)) + + # local audio reference -- a path in the evidence, shown as recorded + audio = score.get("audio") + parts.append('
Audio
') + if not audio: + parts.append('
This record carries no audio ' + 'reference.
') + else: + parts.append(f'
{_esc(audio.get("path"))}
') + parts.append('
' + f'source {_esc(audio.get("source"))} ' + f'· duration {_fmt_sec(audio.get("duration_sec"))} ' + f'· sample rate {_esc(audio.get("sample_rate"))} ' + '· the recording stays on this machine; the path is ' + 'shown exactly as the evidence recorded it.
') + parts.append('
') + + # scorer config snapshot (the hashed block, expandable) + config = score.get("config") + if config: + parts.append('
scorer config ' + f'({_esc(score.get("config_sha256"))})' + '
' + f'{_esc(json.dumps(config, sort_keys=True, indent=2))}' + '
') + return "".join(parts) + + +def _call_timing_waterfall(timing: dict, hops: List[dict], + latency: Optional[dict]) -> str: + spans = timing.get("turn_spans") if isinstance(timing, dict) else None + spans = spans if isinstance(spans, list) else [] + parts = ['
Timing waterfall
'] + if isinstance(timing, dict) and timing.get("provenance"): + parts.append(f'
{_esc(timing["provenance"])}
') + if spans: + durations = [s.get("duration_ms") for s in spans + if isinstance(s.get("duration_ms"), (int, float))] + scale = max(durations) if durations else 0 + parts.append('
' + '' + '' + '') + for index, span in enumerate(spans, start=1): + duration = span.get("duration_ms") + width = 0 + if isinstance(duration, (int, float)) and scale: + width = int(round(200 * duration / scale)) + reported = span.get("reported") or {} + reported_html = " · ".join( + f'{_esc(k)} {_fmt_ms(v)}' + for k, v in reported.items()) or '-' + parts.append( + f'' + f'' + f'' + f'' + f'' + f'') + parts.append('
turnstartedendedduration (derived)event-reported
{index}{_esc(span.get("started_at") or "-")}{_esc(span.get("ended_at") or "-")}{_fmt_ms(duration)}{reported_html}
') + parts.append('
derived durations come from evidence ' + 'event timestamps; event-reported values are what the ' + 'reporting event submitted and are kept apart.
') + else: + parts.append('
No turn spans on this record.
') + + parts.append('
Per-hop latency' + '
') + if latency and latency.get("hop_count"): + parts.append(f'
{_latency_cell(latency)}
') + if hops: + parts.append('
' + '' + '' + '') + for hop in hops: + parts.append( + f'' + f'' + f'' + f'' + f'' + f'' + f'') + parts.append('
#kindatlatencyauthoritysourceevent
{_esc(hop.get("seq"))}{_esc(hop.get("kind"))}{_esc(hop.get("at"))}{_fmt_ms(hop.get("latency_ms"))}{_esc(hop.get("authority"))}{_esc(hop.get("source"))}{_esc(hop.get("event_id"))}
') + else: + parts.append('
No hop rows on this record.
') + parts.append('
') + return "".join(parts) + + +def _call_evidence_section(session: Optional[dict]) -> str: + parts = ['
Evidence lanes
'] + if session is None: + parts.append('
This session is no longer present in ' + 'the evidence database; the derived record above is what ' + 'was scored.
') + return "".join(parts) + completeness = session.get("completeness") or {} + if completeness.get("status") == "invalid": + parts.append('
The session\'s evidence manifest did ' + 'not validate: ' + f'{_esc(completeness.get("detail"))}
') + return "".join(parts) + lanes = session.get("evidence") or {} + required = set(session.get("required_evidence_lanes") or []) + missing = completeness.get("missing") or [] + if missing: + parts.append('
missing required lanes: ' + f'{_esc(", ".join(missing))}
') + parts.append('
') + for lane, item in lanes.items(): + parts.append(_production_lane(lane, item, required=lane in required)) + parts.append('
') + return "".join(parts) + + # ========================================================================= # error pages # ========================================================================= @@ -1237,6 +1719,10 @@ def render_404(what: str) -> str: return (f'

Not found

{_esc(what)}
') +def render_400(what: str) -> str: + return (f'

Bad request

{_esc(what)}
') + + _GATE_CSS = (""" .gate{max-width:560px;margin:0 auto} .gate .brand{display:flex;align-items:center;gap:12px;margin:8vh 0 22px} diff --git a/tests/test_describe_cli.py b/tests/test_describe_cli.py index 4deed25..83900c6 100644 --- a/tests/test_describe_cli.py +++ b/tests/test_describe_cli.py @@ -60,7 +60,7 @@ "fleet ingest", "fleet discover", "fleet review", "fleet label", "fleet status", "fleet benchmark", "fleet experiment", "fleet experiment create", "fleet experiment run", "fleet experiment propose", "fleet experiment approve", "fleet run", "fleet contract", "fleet contract create", "fleet retention", "fleet delete", "fleet redact", "fleet canary", "fleet canary start", "fleet canary rollback", "fleet export", "fleet trend", - "serve", + "serve", "console", "telephony", "telephony capabilities", "telephony create", "telephony status", "telephony cancel", "telephony export", "caller", "caller run", "caller verify", diff --git a/tests/test_serve_calls.py b/tests/test_serve_calls.py new file mode 100644 index 0000000..5b32277 --- /dev/null +++ b/tests/test_serve_calls.py @@ -0,0 +1,608 @@ +"""``/calls`` -- the console's daily surface over the derived score sidecar. + +Starts the real threaded server over an evidence database built through +production.py's own ingest API plus a sidecar built by the deterministic +rebuild, then proves: + +* the feed renders SCORED, NOT_SCORABLE, and ERROR rows distinctly -- a + refused or errored session is never hidden and never reads as OK (I2); +* filters (session state, scorability, time window) are query-param driven and + a malformed filter/cursor is a 400, never a silently dropped filter; +* pagination is keyset from the first page (no OFFSET anywhere on the path) + and pages are disjoint and complete (I3); +* the per-call view renders per-dimension observations (never blended), the + ranked candidate moments with measured magnitudes, the timing waterfall with + derived-vs-reported authority rendered rather than flattened (I5), scorer + version + config hash, evidence-lane completeness, and the local audio path + as recorded -- with no invented file-serving route; +* the JSON mirror carries an ETag and a matching ``If-None-Match`` gets a + body-less 304 (R7), while changed evidence changes the ETag; +* ``hotato console`` is serve + production-db + score-on-arrival in one + command, landing on ``/calls``; +* every ``/calls`` path stays token-gated, and no view grows a blended score. +""" +from __future__ import annotations + +import json +import os +import shutil +import threading +import urllib.error +import urllib.request +from importlib import resources +from urllib.parse import quote + +import pytest + +from hotato import __version__, cli +from hotato import console_worker as console_worker_mod +from hotato.console_store import ConsoleStore +from hotato.console_worker import default_console_path, rebuild_sidecar +from hotato.fleet.registry import Registry +from hotato.production import ProductionStore +from hotato.serve import build_server +from hotato.serve.app import ServeContext +from hotato.serve.security import AuditLog, SessionStore + +_TOKEN = "tok_calls_TEST_0123456789_xyz" + + +# ========================================================================= +# fixture: an evidence db + sidecar with one row per score state +# ========================================================================= + +def _stereo_fixture() -> str: + return str( + resources.files("hotato").joinpath( + "data", "audio", "01-hard-interruption.example.wav" + ) + ) + + +def _event(event_id, event_type, *, subject, time_value, data=None, + authority="adapter_reported"): + return { + "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"), + }, + } + + +def _ingest_call(store, subject, *, audio_path=None, with_media_event=True): + """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=False) + + +def _seed_evidence(tmp_path, monkeypatch) -> str: + """Three sessions at three distinct arrival stamps -- one lands SCORED, + one NOT_SCORABLE (no audio lane), one ERROR (a scorer crash pinned to one + recording during the rebuild) -- then the sidecar is built by the + deterministic rebuild. Returns the evidence db path.""" + db_path = str(tmp_path / "production.sqlite3") + crashing = str(tmp_path / "crash.wav") + shutil.copy(_stereo_fixture(), crashing) + + clock = [1000.0] + evidence = ProductionStore(db_path, clock=lambda: clock[0]) + _ingest_call(evidence, "call-scored", audio_path=_stereo_fixture()) + clock[0] = 2000.0 + _ingest_call(evidence, "call-noaudio", with_media_event=False) + clock[0] = 3000.0 + _ingest_call(evidence, "call-crash", audio_path=crashing) + evidence.close() + + 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) + store = ConsoleStore(default_console_path(db_path)) + try: + summary = rebuild_sidecar(db_path, store) + finally: + store.close() + monkeypatch.setattr(console_worker_mod, "scan_recording", real_scan) + assert summary["scored"] == 1 + assert summary["not_scorable"] == 1 + assert summary["errors"] == 1 + return db_path + + +class _Live: + def __init__(self, base, token, home, production_db, server, thread): + self.base = base + self.token = token + self.home = home + self.production_db = production_db + self._server = server + self._thread = thread + + def stop(self): + self._server.shutdown() + self._thread.join(timeout=5) + self._server.server_close() + + +def _start_server(home: str, production_db): + os.makedirs(home, exist_ok=True) + Registry(home=home).close() + state_dir = os.path.join(home, "serve", "default") + os.makedirs(state_dir, exist_ok=True) + ctx = ServeContext( + home=home, workspace="default", + store_root=os.path.join(home, "artifacts"), token=_TOKEN, + state_dir=state_dir, + audit=AuditLog(os.path.join(state_dir, "audit.jsonl")), + sessions=SessionStore(), bind_host="127.0.0.1", + production_db=production_db) + server = build_server(ctx, "127.0.0.1", 0) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return _Live("http://127.0.0.1:%d" % port, _TOKEN, home, production_db, + server, thread) + + +@pytest.fixture() +def live(tmp_path, monkeypatch): + production_db = _seed_evidence(tmp_path, monkeypatch) + server = _start_server(str(tmp_path / "fleet"), production_db) + try: + yield server + finally: + server.stop() + + +def _req(base, path, *, token=None, headers=None): + req = urllib.request.Request(base + path) + if token is not None: + req.add_header("Authorization", "Bearer " + token) + for key, value in (headers or {}).items(): + req.add_header(key, value) + try: + resp = urllib.request.urlopen(req, timeout=5) + return resp.getcode(), resp.read().decode("utf-8"), dict(resp.headers) + except urllib.error.HTTPError as exc: + return exc.code, exc.read().decode("utf-8"), dict(exc.headers) + + +def _rows(base, path, token): + code, body, _h = _req(base, path, token=token) + assert code == 200, path + return json.loads(body) + + +# ========================================================================= +# auth +# ========================================================================= + +def test_calls_routes_require_token(live): + for path in ["/calls", "/calls?format=json", "/calls/call-scored", + "/calls/call-scored?format=json"]: + code, body, headers = _req(live.base, path) + assert code == 401, path + assert "bearer" in headers.get("WWW-Authenticate", "").lower(), path + assert "call-scored" not in body or path.startswith("/calls/call-"), path + assert "NOT_SCORABLE" not in body, path + + +# ========================================================================= +# the feed: three first-class states, never hidden, never OK +# ========================================================================= + +def test_feed_renders_all_three_score_states_distinctly(live): + code, body, _h = _req(live.base, "/calls", token=live.token) + assert code == 200 + for subject in ("call-scored", "call-noaudio", "call-crash"): + assert subject in body + assert "SCORED" in body + assert "NOT_SCORABLE" in body + assert "ERROR" in body + # the refusal and the crash each carry their reason on the row + assert "participant_audio evidence lane is missing" in body + assert "RuntimeError: boom" in body + # the measured failure-reason sentence for the scored call + assert "the caller took the floor" in body + + m = _rows(live.base, "/calls?format=json", live.token) + states = {r["subject"]: r["score_state"] for r in m["rows"]} + assert states == {"call-scored": "SCORED", "call-noaudio": "NOT_SCORABLE", + "call-crash": "ERROR"} + # a NOT_SCORABLE row keeps its reason in the mirror too + by_subject = {r["subject"]: r for r in m["rows"]} + assert "participant_audio" in by_subject["call-noaudio"]["reason"] + # newest arrival first + assert [r["subject"] for r in m["rows"]] == [ + "call-crash", "call-noaudio", "call-scored"] + + +def test_feed_rows_carry_provenance_labeled_timing_and_evidence(live): + m = _rows(live.base, "/calls?format=json", live.token) + scored = next(r for r in m["rows"] if r["subject"] == "call-scored") + # evidence-clock anchor and arrival stamp each carry their clock label (I5) + assert scored["evidence_time_authority"] == "derived:event_timestamps" + assert scored["received_authority"] == "received:arrival_clock" + assert scored["received_at"] == 1000.0 + # duration derives from evidence event timestamps + assert scored["duration_ms"] == 6000.0 + assert scored["duration_authority"] == "derived:event_timestamps" + # per-call hop percentiles state the declared authorities behind them + assert scored["latency"]["hop_count"] == 2 + assert scored["latency"]["p50_ms"] == 210.0 + assert scored["latency"]["p95_ms"] == 2500.0 + assert set(scored["latency"]["authorities"]) == { + "adapter_reported", "derived:event_timestamps"} + # worst dimension is a per-call worst case, not a blend + assert scored["worst"]["dimension"] == "overlap_while_agent_talking" + assert scored["worst"]["worst_sec"] > 0 + # evidence completeness counts required lanes and names the missing ones + assert scored["evidence"]["status"] == "incomplete" + assert "transcript" in scored["evidence"]["missing"] + + +# ========================================================================= +# filters + keyset pagination (I3: no OFFSET walks) +# ========================================================================= + +def test_feed_filters_by_scorability_state_and_window(live): + m = _rows(live.base, "/calls?scorability=NOT_SCORABLE&format=json", + live.token) + assert [r["subject"] for r in m["rows"]] == ["call-noaudio"] + assert m["trends"]["volume"] == 1 + + m = _rows(live.base, "/calls?state=QUIESCENT&format=json", live.token) + assert m["row_count"] == 3 # every fixture session quiesced + + m = _rows(live.base, "/calls?since=2500&format=json", live.token) + assert [r["subject"] for r in m["rows"]] == ["call-crash"] + m = _rows(live.base, "/calls?until=1500&format=json", live.token) + assert [r["subject"] for r in m["rows"]] == ["call-scored"] + + +def test_feed_bad_filter_or_cursor_is_400_never_ignored(live): + for path in ["/calls?since=yesterday", "/calls?cursor=zzz", + "/calls?limit=0", "/calls?limit=snack"]: + code, _body, _h = _req(live.base, path, token=live.token) + assert code == 400, path + code, body, _h = _req(live.base, path + "&format=json", + token=live.token) + assert code == 400, path + assert json.loads(body)["error"] == "bad request" + + +def test_feed_keyset_pagination_is_disjoint_and_complete(live): + seen = [] + path = "/calls?limit=1&format=json" + for _ in range(4): + m = _rows(live.base, path, live.token) + seen.extend(r["subject"] for r in m["rows"]) + cursor = m["page"]["next_cursor"] + if cursor is None: + break + path = "/calls?limit=1&cursor=%s&format=json" % quote(cursor, safe="") + assert seen == ["call-crash", "call-noaudio", "call-scored"] + assert m["page"]["next_cursor"] is None + + +def test_feed_sql_uses_no_offset(live, monkeypatch): + # I3 at the statement level: walking every page issues no OFFSET; each + # page is one LIMIT-bounded keyset query. + from hotato.serve import data as data_mod + + statements = [] + real_open = data_mod._open_console_ro + + def traced_open(console_path, production_path): + db = real_open(console_path, production_path) + db.set_trace_callback(statements.append) + return db + + monkeypatch.setattr(data_mod, "_open_console_ro", traced_open) + path = "/calls?limit=1&format=json" + while True: + m = _rows(live.base, path, live.token) + cursor = m["page"]["next_cursor"] + if cursor is None: + break + path = "/calls?limit=1&cursor=%s&format=json" % quote(cursor, safe="") + assert statements, "the feed reads through the traced connection" + assert all("OFFSET" not in sql.upper() for sql in statements) + assert any("LIMIT" in sql.upper() for sql in statements) + + +# ========================================================================= +# trends strip (R8) +# ========================================================================= + +def test_trends_strip_over_the_filtered_window(live): + m = _rows(live.base, "/calls?format=json", live.token) + trends = m["trends"] + assert trends["volume"] == 3 + assert trends["states"] == {"SCORED": 1, "NOT_SCORABLE": 1, "ERROR": 1} + assert trends["with_candidate_moments"] == 1 + assert trends["candidate_share"] == 1.0 + # per-kind hop percentiles, never pooled across kinds, each stating the + # declared authorities behind its values (I5) + hop = trends["hop_latency_ms"] + assert hop["model.operation"]["p50_ms"] == 210.0 + assert hop["model.operation"]["authorities"] == ["adapter_reported"] + assert hop["turn"]["p50_ms"] == 2500.0 + assert hop["turn"]["authorities"] == ["derived:event_timestamps"] + # a window filter narrows the trends too + m = _rows(live.base, "/calls?since=2500&format=json", live.token) + assert m["trends"]["volume"] == 1 + assert m["trends"]["states"]["ERROR"] == 1 + + +# ========================================================================= +# per-call view (R6) +# ========================================================================= + +def test_call_detail_renders_observations_moments_and_provenance(live): + code, body, _h = _req(live.base, "/calls/call-scored", token=live.token) + assert code == 200 + # per-dimension observations, each on its own + assert "Per-dimension observations" in body + assert "overlap_while_agent_talking" in body + assert "long_response_gap" in body + # ranked candidate moments with measured magnitudes + assert "Ranked candidate moments" in body + assert "overlap_sec" in body + # the timing waterfall keeps derived and reported apart (I5) + assert "Timing waterfall" in body + assert "derived:event_timestamps" in body + assert "adapter_reported" in body + assert "yield_latency_ms" in body + # scorer provenance is shown + assert __version__ in body + assert "sha256:" in body + # evidence lanes + the local audio path, shown as recorded (no new + # file-serving route: the page carries no audio element) + assert "participant_audio" in body + assert "01-hard-interruption.example.wav" in body + assert "= 1 + assert score["candidates"][0]["plain_english"].startswith( + "the caller took the floor") + hops = {hop["kind"]: hop for hop in score["hops"]} + assert hops["model.operation"]["authority"] == "adapter_reported" + assert hops["turn"]["authority"] == "derived:event_timestamps" + assert score["timing"]["turn_spans"][0]["reported"]["yield_latency_ms"] == 480.0 + assert score["audio"]["path"].endswith(".wav") + assert m["session"]["state"] == "QUIESCENT" + assert m["session"]["completeness"]["status"] == "incomplete" + + +def test_call_detail_error_row_shows_its_reason(live): + code, body, _h = _req(live.base, "/calls/call-crash", token=live.token) + assert code == 200 + assert "ERROR" in body + assert "RuntimeError: boom" in body + + +def test_unknown_call_is_404(live): + for path in ["/calls/nope", "/calls/nope?format=json"]: + code, _body, _h = _req(live.base, path, token=live.token) + assert code == 404, path + + +# ========================================================================= +# ETag + 304 (R7) +# ========================================================================= + +def test_feed_etag_304_and_change_detection(live, tmp_path): + code, body, headers = _req(live.base, "/calls?format=json", + token=live.token) + assert code == 200 + etag = headers.get("ETag") + assert etag + + # a matching If-None-Match is a body-less 304 carrying the same ETag + code, body, headers = _req(live.base, "/calls?format=json", + token=live.token, + headers={"If-None-Match": etag}) + assert code == 304 + assert body == "" + assert headers.get("ETag") == etag + + # new evidence -> new sidecar content -> a fresh ETag and a 200 + evidence = ProductionStore(live.production_db, clock=lambda: 4000.0) + _ingest_call(evidence, "call-later", audio_path=_stereo_fixture()) + evidence.close() + store = ConsoleStore(default_console_path(live.production_db)) + try: + rebuild_sidecar(live.production_db, store) + finally: + store.close() + code, body, headers = _req(live.base, "/calls?format=json", + token=live.token, + headers={"If-None-Match": etag}) + assert code == 200 + assert headers.get("ETag") != etag + assert "call-later" in body + + +def test_feed_page_degrades_without_javascript(live): + # the HTML page is complete as served: every row and the trends strip are + # server-rendered; the polling script only revalidates and swaps + code, body, _h = _req(live.base, "/calls", token=live.token) + assert code == 200 + assert 'id="calls-live"' in body + assert 'id="calls-updated"' in body + assert "If-None-Match" in body # the poll revalidates, not re-reads + for subject in ("call-scored", "call-noaudio", "call-crash"): + assert subject in body + + +# ========================================================================= +# honesty invariants +# ========================================================================= + +def test_no_blended_or_overall_score_on_calls_surfaces(live): + for path in ["/calls", "/calls/call-scored"]: + _c, body, _h = _req(live.base, path, token=live.token) + # the pages echo the operator-selected db paths; this test's own tmp + # directory is named after the test, so mask those before scanning + low = body.replace(os.path.dirname(live.production_db), "").lower() + assert "overall_score" not in low, path + assert "overall score" not in low, path + assert "blended" not in low, path + assert "composite score" not in low, path + for path in ["/calls?format=json", "/calls/call-scored?format=json"]: + _c, body, _h = _req(live.base, path, token=live.token) + assert "overall_score" not in body, path + + +def test_feed_states_absent_sources_explicitly(tmp_path): + # no --production-db: the feed says how to wire one, and shares no rows + server = _start_server(str(tmp_path / "fleet"), None) + try: + code, body, _h = _req(server.base, "/calls", token=server.token) + assert code == 200 + assert "hotato console" in body + m = _rows(server.base, "/calls?format=json", server.token) + assert m["configured"] is False + assert m["rows"] == [] + # the per-call view has nothing to resolve -> 404, never a fabrication + code, _body, _h = _req(server.base, "/calls/anything", + token=server.token) + assert code == 404 + finally: + server.stop() + + +def test_feed_states_missing_sidecar_explicitly(tmp_path): + # an evidence db with no sidecar beside it yet: explicit state, zero rows + db_path = str(tmp_path / "production.sqlite3") + evidence = ProductionStore(db_path, clock=lambda: 1000.0) + _ingest_call(evidence, "call-a", audio_path=_stereo_fixture()) + evidence.close() + server = _start_server(str(tmp_path / "fleet"), db_path) + try: + m = _rows(server.base, "/calls?format=json", server.token) + assert m["configured"] is True + assert m["sidecar"]["present"] is False + assert m["rows"] == [] + code, body, _h = _req(server.base, "/calls", token=server.token) + assert "--rebuild-scores" in body + finally: + server.stop() + + +# ========================================================================= +# the `hotato console` command (R4) +# ========================================================================= + +def test_console_command_defaults_mirror_serve(): + args = cli.build_parser().parse_args( + ["console", "--production-db", "prod.sqlite3"]) + assert args.production_db == "prod.sqlite3" + assert args.host == "127.0.0.1" + assert args.port == 8321 + assert args.workspace == "default" + assert args.no_open is False + + +def test_console_command_requires_the_production_db(capsys): + with pytest.raises(SystemExit) as excinfo: + cli.build_parser().parse_args(["console"]) + assert excinfo.value.code == 2 + + +def test_console_command_is_serve_plus_worker_landing_on_calls(monkeypatch): + import hotato.serve as serve_mod + + calls = {} + + def fake_run_serve(**kwargs): + calls.update(kwargs) + return 0 + + monkeypatch.setattr(serve_mod, "run_serve", fake_run_serve) + assert cli.main(["console", "--production-db", "prod.sqlite3", + "--no-open", "--port", "9999"]) == 0 + assert calls["production_db"] == "prod.sqlite3" + assert calls["score_production"] is True + assert calls["landing"] == "/calls" + assert calls["port"] == 9999 + assert calls["open_browser"] is False + + +def test_console_landing_url_points_at_calls(tmp_path, monkeypatch, capsys): + # the printed tokenised URL opens on the live call feed + db_path = str(tmp_path / "production.sqlite3") + evidence = ProductionStore(db_path, clock=lambda: 1000.0) + _ingest_call(evidence, "call-a", audio_path=_stereo_fixture()) + evidence.close() + + import hotato.serve.app as app_mod + + class _StopServe(Exception): + pass + + def bail(self): + raise KeyboardInterrupt + + monkeypatch.setattr(app_mod._WorkspaceServer, "serve_forever", bail) + monkeypatch.setenv("HOTATO_NO_BROWSER", "1") + code = app_mod.run_serve( + workspace="default", port=0, registry=str(tmp_path / "fleet"), + token="tok-test", open_browser=False, production_db=db_path, + score_production=True, landing="/calls") + assert code == 0 + err = capsys.readouterr().err + assert "/calls?token=" in err + + +def test_calls_tab_is_in_the_nav(live): + code, body, _h = _req(live.base, "/", token=live.token) + assert code == 200 + assert 'href="/calls"' in body + assert ">Calls<" in body