From d4c2936584e84498bc9220b4bb338f5c93234237 Mon Sep 17 00:00:00 2001 From: zengchuang Date: Sun, 15 Nov 2026 16:46:16 +0800 Subject: [PATCH] Connect report Dashboard and trace to fixed attempt records Signed-off-by: zengchuang Agent-assisted-by: Codex (author) --- docs/operations.md | 8 +++ src/inferrsi/apps/views/attempt.py | 19 ++++- src/inferrsi/apps/views/attempt_html.py | 87 +++++++++++++++++++++++ src/inferrsi/apps/views/dashboard.py | 17 ++++- src/inferrsi/apps/views/dashboard_live.py | 29 ++++++-- src/inferrsi/arguments.py | 34 ++++++--- src/inferrsi/cli.py | 76 ++++++++++++++------ tests/test_apps_assess.py | 50 +++++++++++-- 8 files changed, 273 insertions(+), 47 deletions(-) create mode 100644 src/inferrsi/apps/views/attempt_html.py diff --git a/docs/operations.md b/docs/operations.md index 9a3e2420..6cfc475d 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -249,6 +249,14 @@ The same read-only view is available through `inferrsi.apps.api.read_attempt_view`. The legacy `status --run-id` form reads the legacy experiment directory. +Use the same reference files with `trace`, `report` or `dashboard`. +`trace` prints the view plus recorded steps, checks and error codes. An +unknown step time stays null. `report` and offline `dashboard` accept +`--output ` and present the same facts and claims. With +`dashboard --live`, the loopback page and its state endpoint serve the +snapshot pinned at startup. The campaign Dashboard remains available through +`--input`, including its reviewed translation overlay. + ## Campaign summary figure ### Summarize a campaign figure — `tools/campaign_summary_chart.py` diff --git a/src/inferrsi/apps/views/attempt.py b/src/inferrsi/apps/views/attempt.py index 4c7a2467..b415b81b 100644 --- a/src/inferrsi/apps/views/attempt.py +++ b/src/inferrsi/apps/views/attempt.py @@ -15,12 +15,20 @@ "process_outcome", "exit_code", "recorded_at", "prepared_ref", ) _CLAIM_FIELDS = ("claim_id", "validation_scope", "verdict", "reasons", "missing_evidence") +_TRACE_FIELDS = { + "steps": ( + "step_id", "status", "started_at", "ended_at", "elapsed_seconds", "exit_code", "reasons", + ), + "checks": ("check_id", "subject", "phase", "status", "reasons"), + "errors": ("code", "stage", "retryable"), +} def read_attempt_view( record_ref: dict, *, storage_config: dict, assessment_ref: dict | None = None, + include_trace: bool = False, ) -> dict: - """Return fixed execution facts and claim outcomes for one record revision.""" + """Return fixed facts, claim outcomes and optional trace for one record revision.""" try: _check_ref(record_ref, "attempt_id", "record_version") if record_ref["artifact"]["media_type"] != "application/json": @@ -44,6 +52,10 @@ def read_attempt_view( execution["cleanup_status"] = record["cleanup"]["status"] candidate = record["candidate"] execution["candidate_id"] = candidate["candidate_id"] if candidate is not None else None + trace = { + group: [{key: item[key] for key in fields} for item in record[group]] + for group, fields in _TRACE_FIELDS.items() + } if include_trace else None except (KeyError, TypeError, ValueError): raise PublicError( "ArtifactIntegrityError", stage="attempt_view", @@ -82,7 +94,10 @@ def read_attempt_view( message="Invalid archived assessment for this record.", ) from None - return { + view = { "record_ref": deepcopy(record_ref), "execution": execution, "assessment_ref": deepcopy(assessment_ref), "assessment": evaluation, } + if include_trace: + view["trace"] = trace + return view diff --git a/src/inferrsi/apps/views/attempt_html.py b/src/inferrsi/apps/views/attempt_html.py new file mode 100644 index 00000000..c07c9540 --- /dev/null +++ b/src/inferrsi/apps/views/attempt_html.py @@ -0,0 +1,87 @@ +"""Render a fixed attempt for the report and Dashboard.""" + +from html import escape + + +def _text(value: object) -> str: + if value is None: + return "Unknown" + if isinstance(value, list): + return escape(", ".join(str(item) for item in value)) if value else "—" + return escape(str(value)) + + +def _table(rows: list[dict], columns: tuple[tuple[str, str], ...]) -> str: + if not rows: + return "

No entries recorded.

" + head = "".join(f'{label}' for _, label in columns) + body = [ + "" + "".join(f"{_text(row[key])}" for key, _ in columns) + "" + for row in rows + ] + return ( + f'
{head}' + f'{"".join(body)}
' + ) + + +def render_attempt(view: dict, *, title: str) -> str: + """Present the shared read model without exposing raw log or artifact paths.""" + execution, assessment, trace = view["execution"], view["assessment"], view["trace"] + facts = ( + ("Run", execution["run_id"]), + ("Attempt", execution["attempt_id"]), + ("Record revision", execution["record_version"]), + ("Record SHA-256", view["record_ref"]["artifact"]["sha256"]), + ("Recorded at", execution["recorded_at"]), + ("Candidate", execution["candidate_id"]), + ("Phase", execution["phase"]), + ("Process outcome", execution["process_outcome"]), + ("Exit code", execution["exit_code"]), + ("Cleanup", execution["cleanup_status"]), + ) + summary = "".join(f"
{label}
{_text(value)}
" for label, value in facts) + claims = "

No archived assessment selected.

" + if assessment is not None: + digest = _text(view["assessment_ref"]["artifact"]["sha256"]) + claims = f"

Assessment SHA-256 {digest}

" + _table( + assessment["claims"], + (("claim_id", "Claim"), ("validation_scope", "Scope"), ("verdict", "Verdict"), + ("reasons", "Reasons"), ("missing_evidence", "Missing evidence")), + ) + + steps = _table( + trace["steps"], + (("step_id", "Step"), ("status", "Status"), ("started_at", "Started"), + ("ended_at", "Ended"), ("elapsed_seconds", "Seconds"), ("exit_code", "Exit code"), + ("reasons", "Reasons")), + ) + checks = _table( + trace["checks"], + (("check_id", "Check"), ("subject", "Subject"), ("phase", "Phase"), ("status", "Status"), + ("reasons", "Reasons")), + ) + errors = _table( + trace["errors"], (("code", "Code"), ("stage", "Stage"), ("retryable", "Retryable")), + ) + name = escape(title) + return f""" + + + + +InferRSI {name} + +

InferRSI {name}

+

Execution

{summary}
+

Archived assessment

{claims}
+

Trace

Steps

{steps}

Checks

{checks}

Errors

{errors}
+
""" diff --git a/src/inferrsi/apps/views/dashboard.py b/src/inferrsi/apps/views/dashboard.py index dd432d1e..d8744b0a 100644 --- a/src/inferrsi/apps/views/dashboard.py +++ b/src/inferrsi/apps/views/dashboard.py @@ -13,6 +13,7 @@ from inferrsi.harness.records.evidence import inside, sanitized_text from inferrsi.summary_chart import SpecError, build_svg +from .attempt_html import render_attempt from .dashboard_data import _number, load_dashboard from .dashboard_timeline import temporal_chart @@ -334,6 +335,8 @@ def dashboard_snapshot(source: Path, translations: Path | None = None) -> dict: def render_dashboard(snapshot: dict, *, live: bool = False) -> str: """Render a snapshot; only the local live view enables same-origin requests.""" + if "fixed_attempt" in snapshot["data"]: + return render_attempt(snapshot["data"]["fixed_attempt"], title="fixed attempt Dashboard") view = {**snapshot["data"], "live": live} payload = json.dumps(view, ensure_ascii=False, allow_nan=False) # The JSON data block cannot terminate its script element or create markup. @@ -366,6 +369,18 @@ def write_dashboard(source: Path, destination: Path | None = None, source = Path(source).resolve() document = render_dashboard(dashboard_snapshot(source, translations)) output = Path(destination) if destination is not None else source / "dashboard.html" + return _write_view_document(output, document, generator="InferRSI dashboard") + + +def write_attempt_document(view: dict, destination: Path, *, kind: str) -> Path: + """Export the shared attempt view as the selected HTML document.""" + title = "fixed attempt Dashboard" if kind == "dashboard" else "fixed attempt report" + return _write_view_document( + destination, render_attempt(view, title=title), generator=f"InferRSI {title}", + ) + + +def _write_view_document(output: Path, document: str, *, generator: str) -> Path: output.parent.mkdir(parents=True, exist_ok=True) if output.is_symlink(): raise ValueError("dashboard output cannot be a symbolic link") @@ -375,7 +390,7 @@ def write_dashboard(source: Path, destination: Path | None = None, if output.suffix.lower() != ".html": raise ValueError("refusing to overwrite an existing non-dashboard file") with output.open("r", encoding="utf-8", errors="replace") as existing: - if '' not in existing.read(1024): + if f'' not in existing.read(1024): raise ValueError("refusing to overwrite an existing non-dashboard file") temporary = None try: diff --git a/src/inferrsi/apps/views/dashboard_live.py b/src/inferrsi/apps/views/dashboard_live.py index f76f216c..884606ee 100644 --- a/src/inferrsi/apps/views/dashboard_live.py +++ b/src/inferrsi/apps/views/dashboard_live.py @@ -8,6 +8,7 @@ import re import sys from collections import Counter +from copy import deepcopy from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -44,14 +45,25 @@ def _compact_contexts(snapshot: dict) -> dict: class DashboardServer(ThreadingHTTPServer): """Only expose the generated view, never the underlying directory.""" - def __init__(self, source: Path, translations: Path | None, port: int): + def __init__( + self, source: Path | None, translations: Path | None, port: int, + *, fixed_attempt: dict | None = None, + ): if not 0 <= port <= 65535: raise ValueError("dashboard port must be between 0 and 65535") - self.source = Path(source).resolve() + if (source is None) == (fixed_attempt is None): + raise ValueError("select a Dashboard directory or a fixed attempt") + self.source = Path(source).resolve() if source is not None else None self.translations = Path(translations).resolve() if translations else None - dashboard_snapshot(self.source, self.translations) + self._fixed_attempt = deepcopy(fixed_attempt) + self.snapshot() super().__init__(("127.0.0.1", port), DashboardHandler) + def snapshot(self) -> dict: + if self._fixed_attempt is not None: + return {"data": {"fixed_attempt": self._fixed_attempt}, "chart": "", "temporal": ""} + return dashboard_snapshot(self.source, self.translations) + class DashboardHandler(BaseHTTPRequestHandler): server: DashboardServer @@ -107,7 +119,7 @@ def do_GET(self) -> None: self._reply(403, b"Invalid host") return events = re.fullmatch(r"/api/profile-events/([0-9a-f]{64})", self.path) - if events: + if events and self.server.source is not None: from inferrsi.harness.records.profile_timeline import profile_event_table payload = profile_event_table(self.server.source, events.group(1)) @@ -120,7 +132,7 @@ def do_GET(self) -> None: self._reply(404, b"Not found") return try: - snapshot = dashboard_snapshot(self.server.source, self.server.translations) + snapshot = self.server.snapshot() if self.path == "/": html = render_dashboard(_compact_contexts(snapshot), live=True) body, encoding = self._representation(html.encode("utf-8")) @@ -142,9 +154,12 @@ def do_GET(self) -> None: self._reply(503, b"Dashboard snapshot is temporarily unavailable") -def serve_dashboard(source: Path, translations: Path | None = None, port: int = 8765) -> None: +def serve_dashboard( + source: Path | None, translations: Path | None = None, port: int = 8765, + *, fixed_attempt: dict | None = None, +) -> None: """Serve until interrupted; the listening socket never leaves loopback.""" - with DashboardServer(source, translations, port) as server: + with DashboardServer(source, translations, port, fixed_attempt=fixed_attempt) as server: print(json.dumps({"dashboard": f"http://127.0.0.1:{server.server_port}/"}), flush=True) try: server.serve_forever(poll_interval=0.25) diff --git a/src/inferrsi/arguments.py b/src/inferrsi/arguments.py index 22a9be81..e943f2ec 100644 --- a/src/inferrsi/arguments.py +++ b/src/inferrsi/arguments.py @@ -13,6 +13,24 @@ from .runspec import RunSpec +def _fixed_attempt_args( + parser: argparse.ArgumentParser, *, legacy_flag: str | None = None, +) -> None: + source = parser + if legacy_flag is not None: + source = parser.add_mutually_exclusive_group(required=True) + if legacy_flag == "--input": + source.add_argument("--input", type=Path, help="Run or campaign directory") + else: + source.add_argument(legacy_flag) + source.add_argument( + "--record-ref", type=Path, required=legacy_flag is None, + help="JSON file containing a RecordRef", + ) + parser.add_argument("--storage-config", type=Path, help="Trusted storage config JSON") + parser.add_argument("--assessment-ref", type=Path, help="Optional AssessmentRef JSON") + + def _common(parser: argparse.ArgumentParser) -> None: parser.add_argument("--repo", default=".", help="Prepared target Git checkout") parser.add_argument("--repo-type", choices=REPO_TYPES, default="vllm_omni") @@ -211,20 +229,18 @@ def parser() -> argparse.ArgumentParser: name, help="Read a recorded experiment" if name == "status" else "Regenerate its HTML report", ) - if name == "status": - source = child.add_mutually_exclusive_group(required=True) - source.add_argument("--run-id") - source.add_argument("--record-ref", type=Path, help="JSON file containing a RecordRef") - child.add_argument("--storage-config", type=Path, help="Trusted storage config JSON") - child.add_argument("--assessment-ref", type=Path, help="Optional AssessmentRef JSON") - else: - child.add_argument("--run-id", required=True) + _fixed_attempt_args(child, legacy_flag="--run-id") + if name == "report": + child.add_argument("--output", type=Path, help="HTML output for a fixed RecordRef") child.add_argument("--output-dir", default="runs") + trace = commands.add_parser("trace", help="Read recorded steps, checks and errors") + _fixed_attempt_args(trace) + dashboard = commands.add_parser( "dashboard", help="Export an offline dashboard for one run or recorded campaign" ) - dashboard.add_argument("--input", type=Path, required=True, help="Run or campaign directory") + _fixed_attempt_args(dashboard, legacy_flag="--input") dashboard.add_argument( "--output", type=Path, help="HTML output (default: input/dashboard.html)" ) diff --git a/src/inferrsi/cli.py b/src/inferrsi/cli.py index 00288814..f74747df 100644 --- a/src/inferrsi/cli.py +++ b/src/inferrsi/cli.py @@ -23,6 +23,52 @@ def _print(value: object) -> None: print(json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False)) +def _fixed_attempt_command(args) -> int: + from .apps.api import read_attempt_view + from .harness.errors import PublicError + + if args.storage_config is None: + raise ValueError("--record-ref requires --storage-config") + command = args.command + live = command == "dashboard" and args.live + if command == "dashboard": + if args.translations is not None: + raise ValueError("--translations requires a Dashboard --input directory") + if live and args.output is not None: + raise ValueError("--output cannot be used with --live") + if not live and args.port is not None: + raise ValueError("--port requires --live") + if command in ("report", "dashboard") and not live and args.output is None: + raise ValueError(f"fixed {command} requires --output") + + record_ref = json.loads(args.record_ref.read_text(encoding="utf-8")) + config = json.loads(args.storage_config.read_text(encoding="utf-8")) + assessment_ref = None + if args.assessment_ref is not None: + assessment_ref = json.loads(args.assessment_ref.read_text(encoding="utf-8")) + if not isinstance(assessment_ref, dict): + raise ValueError("--assessment-ref requires an AssessmentRef JSON object") + try: + view = read_attempt_view( + record_ref, storage_config=config, assessment_ref=assessment_ref, + include_trace=command != "status", + ) + except PublicError as exc: + raise ValueError(str(exc)) from exc + + if command in ("status", "trace"): + _print(view) + elif live: + from .apps.views.dashboard_live import serve_dashboard + + serve_dashboard(None, port=args.port if args.port is not None else 8765, fixed_attempt=view) + else: + from .apps.views.dashboard import write_attempt_document + + _print({command: str(write_attempt_document(view, args.output, kind=command))}) + return 0 + + def _knowledge_reader(args) -> KnowledgeReader: return knowledge_reader(skill_root=args.skill_root, store_root=args.store) @@ -328,30 +374,14 @@ def main(argv: list[str] | None = None) -> int: else: print(text, end="") return 0 - if args.command in ("status", "report"): - if args.command == "status" and args.record_ref is not None: - if args.storage_config is None: - raise ValueError("--record-ref requires --storage-config") - from .apps.api import read_attempt_view - from .harness.errors import PublicError - - record_ref = json.loads(args.record_ref.read_text(encoding="utf-8")) - config = json.loads(args.storage_config.read_text(encoding="utf-8")) - assessment_ref = ( - json.loads(args.assessment_ref.read_text(encoding="utf-8")) - if args.assessment_ref is not None else None - ) - if args.assessment_ref is not None and not isinstance(assessment_ref, dict): - raise ValueError("--assessment-ref requires an AssessmentRef JSON object") - try: - _print(read_attempt_view( - record_ref, storage_config=config, assessment_ref=assessment_ref, - )) - except PublicError as exc: - raise ValueError(str(exc)) from exc - return 0 - if args.command == "status" and (args.storage_config or args.assessment_ref): + if args.command in ("status", "report", "dashboard", "trace"): + if args.record_ref is not None: + return _fixed_attempt_command(args) + if args.storage_config is not None or args.assessment_ref is not None: raise ValueError("--storage-config and --assessment-ref require --record-ref") + if args.command == "report" and args.output is not None: + raise ValueError("report --output requires --record-ref") + if args.command in ("status", "report"): directory = _run_directory(args.output_dir, args.run_id) if args.command == "status": _print(json.loads((directory / "status.json").read_text(encoding="utf-8"))) diff --git a/tests/test_apps_assess.py b/tests/test_apps_assess.py index 1d529ed8..c910f2c8 100644 --- a/tests/test_apps_assess.py +++ b/tests/test_apps_assess.py @@ -6,6 +6,8 @@ from copy import deepcopy from hashlib import sha256 from pathlib import Path +from threading import Thread +from urllib.request import urlopen import pytest @@ -14,6 +16,7 @@ from inferrsi.apps.api import assess as run_assess from inferrsi.apps.api import read_attempt_view from inferrsi.apps.artifact_reader import ArtifactReader +from inferrsi.apps.views.dashboard_live import DashboardServer from inferrsi.cli import main from inferrsi.harness.errors import PublicError from inferrsi.harness.run_store import RunStore @@ -281,24 +284,60 @@ def test_fixed_attempt_view_rechecks_assessment_semantic_identity(offline): read_attempt_view(request["record_ref"], storage_config=config, assessment_ref=ref) -def test_fixed_status_command_matches_the_python_view(tmp_path, offline, capsys): +def test_fixed_surfaces_read_the_same_record_and_assessment(tmp_path, offline, capsys): request, config, directory = offline record_ref = request["record_ref"] assessment_ref = _fixture("fixed-refs")["assessment_ref"] assessment_ref["artifact"] = _write_export( directory, "assessment.json", encode(_fixture("assessment")), ) - flags = ["status"] + flags = [] for name, value in ( ("record-ref", record_ref), ("storage-config", config), ("assessment-ref", assessment_ref), ): path = tmp_path / f"{name}.json" path.write_bytes(encode(value)) flags.extend((f"--{name}", str(path))) - assert main(flags) == 0 - assert json.loads(capsys.readouterr().out) == read_attempt_view( + assert main(["status", *flags]) == 0 + status = json.loads(capsys.readouterr().out) + assert status == read_attempt_view( record_ref, storage_config=config, assessment_ref=assessment_ref, ) + assert main(["trace", *flags]) == 0 + fixed = json.loads(capsys.readouterr().out) + assert fixed == read_attempt_view( + record_ref, storage_config=config, assessment_ref=assessment_ref, include_trace=True, + ) + assert {key: fixed[key] for key in status} == status + assert fixed["trace"]["steps"][0]["started_at"] is None + assert fixed["trace"]["checks"][0]["check_id"] == "input_identity" + assert fixed["trace"]["errors"][0] == { + "code": "capacity_unavailable", "stage": "admission", "retryable": False, + } + reports = {} + for command in ("report", "dashboard"): + output = tmp_path / f"{command}.html" + assert main([command, *flags, "--output", str(output)]) == 0 + assert json.loads(capsys.readouterr().out) == {command: str(output.resolve())} + document = output.read_text(encoding="utf-8") + assert record_ref["artifact"]["sha256"] in document + assert assessment_ref["artifact"]["sha256"] in document + assert "input_identity" in document + assert "INCONCLUSIVE" in document + assert "documentation fixture; not actual execution" not in document + reports[command] = document + with DashboardServer(None, None, 0, fixed_attempt=fixed) as server: + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + url = f"http://127.0.0.1:{server.server_port}" + try: + with urlopen(f"{url}/api/state", timeout=5) as response: + assert json.load(response)["data"]["fixed_attempt"] == fixed + with urlopen(f"{url}/", timeout=5) as response: + assert response.read().decode("utf-8") == reports["dashboard"] + finally: + server.shutdown() + thread.join(timeout=5) @pytest.mark.parametrize(("args", "message"), [ @@ -354,10 +393,11 @@ def assessor(prepared, evidence, *, artifacts): output_id='reassessed') assert output.result['exit_code'] == 9 view = read_attempt_view(request['record_ref'], storage_config=config, - assessment_ref=output.result['assessment_ref']) + assessment_ref=output.result['assessment_ref'], include_trace=True) assert view['execution']['record_version'] == 1 assert view['execution']['phase'] == 'running' assert view['assessment']['claims'][0]['verdict'] == 'INCONCLUSIVE' +assert view['trace']['checks'][0]['check_id'] == 'input_identity' """], cwd=tmp_path, capture_output=True, text=True, check=True, )