"""Read-only run discovery and bounded provisional output, never passing evidence.""" from __future__ import annotations from datetime import datetime import os from pathlib import Path import re import time from .common import ( atomic_json, identifier, now, read_json, redact, reject_symlinks, state_root, ) MAX_LIVE_BYTES = 65536 MAX_HISTORY_ENTRIES = 10000 MAX_HISTORY_SCAN = 500 ANSI = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\))") CONTROLS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]") def display_text(value: str) -> str: return redact(CONTROLS.sub("", ANSI.sub("", value))) def _closed_lines(value: bytes) -> bytes: boundary = value.rfind(b"\n") return value[: boundary + 1] if boundary >= 0 else b"" def capture_text(snapshot, *, provisional: bool = False) -> str: """Do not expose partial secret/context fragments at live or retention boundaries.""" data = snapshot.stdout omitted = snapshot.omitted_stdout_bytes if omitted: head = _closed_lines(data[: snapshot.stdout_head_bytes]) tail = data[snapshot.stdout_head_bytes :] # A rolling tail can start inside Authorization or a secret: drop that fragment. tail = tail.partition(b"\n")[2] if provisional: tail = _closed_lines(tail) return ( display_text(head.decode("utf-8", errors="replace")) + f"\n[Output truncated: {omitted} bytes omitted between retained head and tail; cut boundary lines are withheld.]\n" + display_text(tail.decode("utf-8", errors="replace")) ) if provisional: data = _closed_lines(data) return display_text(data.decode("utf-8", errors="replace")) def bounded_display(value: str, maximum: int, *, tail_only: bool = False) -> str: if type(maximum) is not int or maximum < 1: raise ValueError("Display bound must be a positive integer") encoded = value.encode("utf-8") if len(encoded) <= maximum: return value if tail_only: return encoded[-maximum:].decode("utf-8", errors="ignore") marker = b"\n[Display bound: middle omitted; retained beginning and final output follow.]\n" if maximum <= len(marker): return encoded[-maximum:].decode("utf-8", errors="ignore") available = max(0, maximum - len(marker)) head = available // 2 return ( encoded[:head].decode("utf-8", errors="ignore") + marker.decode() + encoded[-(available - head) :].decode("utf-8", errors="ignore") ) def write_live(log_path: Path, stage_id: str, snapshot, started: float) -> None: # Complete lines only, even at final callback: the separately finalized log # may include an unterminated final line after normal redaction. excerpt = bounded_display( capture_text(snapshot, provisional=True), MAX_LIVE_BYTES, tail_only=True ) atomic_json( log_path.with_suffix(".live.json"), { "schema_version": 1, "run_id": log_path.parent.name, "stage_id": stage_id, "provisional": True, "updated_at": now(), "elapsed_seconds": round(time.monotonic() - started, 3), "output_truncated": snapshot.truncated, "omitted_bytes": snapshot.omitted_stdout_bytes, "excerpt": excerpt, }, ) def read_live( workspace_root: Path, state_dir: Path | None, run_id: str, stage_id: str ) -> dict: identifier(run_id) identifier(stage_id) path = ( state_root(workspace_root, state_dir) / "runs" / run_id / (stage_id + ".live.json") ) if not path.exists() and not path.is_symlink(): return {"provisional": True, "excerpt": "", "live_available": False} payload = read_json(path, max_bytes=MAX_LIVE_BYTES * 6 + 8192) if ( not isinstance(payload, dict) or payload.get("schema_version") != 1 or payload.get("run_id") != run_id or payload.get("stage_id") != stage_id or payload.get("provisional") is not True or not isinstance(payload.get("excerpt"), str) or len(payload["excerpt"].encode("utf-8")) > MAX_LIVE_BYTES or not isinstance(payload.get("updated_at"), str) or type(payload.get("elapsed_seconds")) not in {int, float} or payload["elapsed_seconds"] < 0 or type(payload.get("output_truncated")) is not bool ): raise ValueError("Invalid provisional log snapshot") return { "provisional": True, "live_available": True, "excerpt": display_text(payload["excerpt"]), "updated_at": payload.get("updated_at"), "elapsed_seconds": payload.get("elapsed_seconds"), "output_truncated": payload.get("output_truncated"), } def elapsed_seconds(receipt: dict) -> float | None: try: started = datetime.fromisoformat(receipt["generated_at"]) finished = datetime.fromisoformat(receipt.get("finished_at") or now()) return max(0, round((finished - started).total_seconds(), 3)) except (KeyError, TypeError, ValueError): return None def list_runs(args) -> dict: from .runner import read_receipt limit = getattr(args, "limit", 10) before = getattr(args, "before", None) if type(limit) is not int or not 1 <= limit <= 100: raise ValueError("Run history limit must be between 1 and 100") if before: identifier(before) base = state_root(args.workspace_root, args.state_dir) / "runs" reject_symlinks(base) names = [] if base.exists(): with os.scandir(base) as entries: for index, entry in enumerate(entries): if index >= MAX_HISTORY_ENTRIES: raise ValueError( "Run history exceeds its directory bound; archive old evidence explicitly before listing" ) if not entry.name.startswith("."): identifier(entry.name) names.append(entry.name) names = sorted( (name for name in names if before is None or name < before), reverse=True ) rows, examined, cursor = [], 0, None wanted_project = ( str(args.project.resolve()) if getattr(args, "project", None) else None ) for identity in names: examined += 1 cursor = identity try: record = read_receipt(args.workspace_root, args.state_dir, identity) if wanted_project and record.get("project_file") != wanted_project: if examined >= MAX_HISTORY_SCAN: break continue rows.append( { "run_id": identity, "status": record["status"], "phase": record.get("phase"), "profile": record.get("profile"), "generated_at": record.get("generated_at"), "elapsed_seconds": elapsed_seconds(record), "snapshot_verified": record["snapshot_verified"], "passed_stages": sum( item["status"] == "passed" for item in record["stages"] ), "total_stages": len(record["stages"]), } ) except (OSError, ValueError) as exc: # Corrupt/newest evidence is visible, never silently replaced with an older pass. rows.append( { "run_id": display_text(identity), "status": "invalid", "error": display_text(str(exc)), } ) if len(rows) >= limit or examined >= MAX_HISTORY_SCAN: break next_cursor = cursor if examined < len(names) else None lines = [ f"{item['run_id']}: {item['status']}" + ( f" ({item.get('profile')}; {item.get('elapsed_seconds')}s)" if item["status"] != "invalid" else " — " + item["error"] ) for item in rows ] if not lines: lines = ["No matching check runs found; no verification is implied."] if next_cursor: lines.append("More history: use --before " + display_text(next_cursor)) return { "runs": rows, "next_cursor": next_cursor, "examined": examined, "summary": lines, "_exit_code": 1 if any(item["status"] == "invalid" for item in rows) else 0, } def latest_run(args) -> dict: from argparse import Namespace from .runner import read_receipt, summarize selected = list_runs(Namespace(**{**vars(args), "limit": 1})) if not selected["runs"]: return { "status": "not_found", "summary": selected["summary"], "next_cursor": selected["next_cursor"], "_exit_code": 1, } row = selected["runs"][0] if row["status"] == "invalid": raise ValueError( "Latest run is invalid; inspect run history instead of assuming an older pass" ) return summarize(read_receipt(args.workspace_root, args.state_dir, row["run_id"]))