diff --git a/CHANGELOG.md b/CHANGELOG.md index c7d79b22..eee898db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ Versions follow [SemVer](https://semver.org). ## [Unreleased] +- Kernel events now append to a per-run `trajectory.jsonl` with sequence numbers and a SHA-256 chain. `outerloop trajectory [--verify] [--json]` reads it without changing run state. Operational files and their formats are unchanged; older kernels ignore the new file. Upgrading: runs started before this kernel have no trajectory and read as "no trajectory" until their next recorded event. No backfill is attempted. - Add a printable agent-text inventory and protocol wording lint, with clearer refusal recovery guidance (#487, parts 1–2). - `BENCHMARKS.md` shows `—` instead of "provenance unknown" in the main-commit column of a baseline row that nothing has improved on yet, as on a freshly seeded repository. - The README and the getting-started guide link to the quickstart repository. It runs two CPU benchmarks with the local forge and local compute, and needs no GitHub account or cluster (#563). The docs and the startup warning no longer say an uncontained session runs in a sandbox: it gets a scrubbed environment and its own home, but can read any file the account can. diff --git a/docs/design/trajectory.md b/docs/design/trajectory.md new file mode 100644 index 00000000..6cc43b7d --- /dev/null +++ b/docs/design/trajectory.md @@ -0,0 +1,41 @@ +# Run trajectory + +The kernel alone writes `runs//trajectory.jsonl`, outside the session's +workspace and writable sandbox mounts. Operational state, launch ledgers, +inbox/outbox and session records continue unchanged. The trajectory is a +best-effort timeline, not a replacement for those recovery records. + +Each line is a JSON object with sorted keys: `seq` (starting at zero), `ts` +(Unix seconds), `type`, `actor`, `data`, `refs`, and `prev`. `prev` is SHA-256 +of the previous event's exact bytes excluding its newline; the first event +uses SHA-256 of the UTF-8 run id. Actors use `kernel`, `author:`, +`judge:`, or `human:`; these kernel actions use `kernel`. +Refs contain run-relative `path` and `sha256` fields. Native session logs are +hashed once at session end, and the report is hashed at ending. + +Events cover session start/end, syscall acceptance/refusal, launch submission +and results, gate results, acknowledged inbox delivery, wakes, PR opening +(including reconciliation of an already opened PR), and endings. Inbox events +contain sequence ranges, kinds and origins, never message bodies. Session start +may have an empty id until the backend returns one. Exceptions produce a session +end outcome when the host can catch them; process death can leave an unmatched +start. Measurement credit describes the gate decision, not publication or merge. + +A sibling `trajectory.lock` serializes appenders with exclusive `flock`. A torn +last line is preserved and separated by a newline; the chain continues from +the last parseable event. Verification reports malformed lines even after +recovery. A complete JSON event missing only its newline is retained and its +newline repaired on the next append. No old bytes are rewritten. Writes stop +at 4 MB and warn once per file per kernel process; I/O failures warn without +failing a tick. Events are not transactional with the operational files, so a +crash or write failure can leave missing events or retries. + +`outerloop trajectory --verify` checks sequence gaps and byte hashes; +`--json` emits events (and verification problems when requested). Missing legacy +files read as an empty history and verification says "no trajectory". No history +is reconstructed from older operational files. Older kernels ignore this file. +The chain detects internal edits and missing lines, but without a trusted +external anchor cannot detect a rewritten chain or removal of its tail. +Verification checks the event chain, not the contents of referenced artifacts. + +Tool calls, full measurement provenance and attestation are later steps. diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 6004d310..7a67b3a6 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -280,3 +280,12 @@ outerloop migrate-ledger --target TARGET --main-sha MAIN_SHA [--force] [--dry-ru | `--main-sha` | required | the full current default-branch commit SHA | | `--force` | off | replace an existing branch ledger | | `--dry-run` | off | print the table without writing | + +### `trajectory [--verify] [--json] [--root PATH]` + +Read the kernel's `runs//trajectory.jsonl` timeline. Text output shows +sequence, time, type, actor and short event data. `--verify` checks the hash chain +and sequence gaps; damaged chains exit 1. Missing legacy files show "no trajectory" +and exit 0. JSON output is an event array, or an object with `events` and `problems` +when verification is requested. This command is read-only. +See [Run trajectory](../design/trajectory.md) for the format and limits. diff --git a/docs/storage.md b/docs/storage.md index 904401ad..491586ab 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -185,7 +185,7 @@ The `.outerloop/` directory inside the workspace is the kernel's channel. It hol ## Run state and leases -A run's record is one JSON file, `runs//state.json`, written by atomic rename. A crash mid-write leaves the previous record intact, and the sweep reasons only from these files plus Slurm, never from process memory. The same run directory holds the wake lease, the inbox, the outbox of replies kept before posting, and the redacted transcript of every session. Beside the run directories, a baseline cache holds the measured baseline that a `cached` contract shares among all attempts on one base commit. +A run's record is one JSON file, `runs//state.json`, written by atomic rename. A crash mid-write leaves the previous record intact, and the sweep reasons only from these files plus Slurm, never from process memory. The same run directory holds the wake lease, the inbox, the outbox of replies kept before posting, and the redacted transcript of every session. The kernel also appends `trajectory.jsonl`, a [hash-chained event timeline](design/trajectory.md), under a sibling `trajectory.lock`. Beside the run directories, a baseline cache holds the measured baseline that a `cached` contract shares among all attempts on one base commit. Unknown fields are ignored on read, and new formats for run records are migrated as the kernel reads them, so an upgrade needs no migration step. That compatibility runs forward only: an older kernel cannot interpret newer record formats, so a rollback past a format change needs its runs drained or their records translated by hand. After a run ends, its workspace and run home are removed after a grace period of 24 hours by default; the record, the report, the transcripts and the ledger entries stay. diff --git a/src/outerloop/attempt.py b/src/outerloop/attempt.py index 7efad233..8a963781 100644 --- a/src/outerloop/attempt.py +++ b/src/outerloop/attempt.py @@ -147,6 +147,8 @@ from outerloop.syscall import install_tool as syscall_install_tool from outerloop.syscall import refresh_tool as syscall_refresh_tool from outerloop.syscall import write_siblings as syscall_write_siblings +from outerloop.trajectory import file_refs +from outerloop.trajectory import record as record_event from outerloop.verifier import MAX_CLAIM_CHARS REPLY_MARKER = marker("followup") @@ -1162,6 +1164,11 @@ def persist() -> None: persist() def refuse(index: int, reason: str) -> None: + record_event( + run_dir, + "syscall.decision", + {"verb": "message", "outcome": "refused", "reason": reason}, + ) entry["recipient"] = record.run_id entry["message"] = asdict( Message( @@ -1275,6 +1282,12 @@ def refuse(index: int, reason: str) -> None: ids=[f"message-{e['counter']:020d}" for e in public], references=[str(e.get("in_reply_to", "")) for e in public], ) + for _entry in public: + record_event( + run_dir, + "syscall.decision", + {"verb": "message", "outcome": "accepted", "reason": ""}, + ) for index, entry in enumerate(journal, 1): if entry.get("delivered"): continue @@ -1338,6 +1351,12 @@ def refuse(index: int, reason: str) -> None: append(run_dir, message) else: append(recipient_dir, message) + if message.kind == "agent-message": + record_event( + run_dir, + "syscall.decision", + {"verb": "message", "outcome": "accepted", "reason": ""}, + ) if recipient.run_id != record.run_id: append( run_dir, @@ -3312,6 +3331,18 @@ def _resume_run( # more experiments, or concludes — instead of being decided by policy here. # Falls back to the plain-finish behavior (negative terminal / draft PR) # when the session cannot be resumed. + record_event( + run_dir, + "measure", + { + "baseline": result.baseline, + "candidate": result.candidate, + "metric": bench.metric, + "credited": result.outcome == "improved", + "outcome": result.outcome, + "head_sha": candidate_sha, + }, + ) submitted_park = bool(stage.get("submitted")) if stage.get("panel_skip"): panel_lenses = () @@ -4221,6 +4252,11 @@ def publish( def refuse( text: str, head: str = "", *, moved: bool = False, quoted_text: str = "" ) -> AttemptOutcome: + record_event( + run_dir, + "syscall.decision", + {"verb": "submit", "outcome": "refused", "reason": redact(text, secrets)}, + ) append( run_dir, Message( @@ -4637,6 +4673,7 @@ def refuse( # read: visible, plainly not merge-ready draft=draft, ) + record_event(run_dir, "pr.open", {"url": pr_url, "head_sha": result.candidate_sha}) # Arm auto-merge, best-effort, and ONLY when branch protection # requires a human review — the guard keeps bot-never-merges # enforced in code, not in per-repo config. Never arm a draft, @@ -6266,6 +6303,12 @@ def finish_run( final, stage={**final.stage, "launch_afterany": "afterany:" + ":".join(sorted(jobs))} ) save_record(run_root, final, now) + record_event( + directory, + "ending", + {"state": ending, "reason": note}, + refs=file_refs(directory, report), + ) ref = str(record.stage.get("candidate_ref") or "") if snapshot_attempted and not ref: ws = None diff --git a/src/outerloop/cli.py b/src/outerloop/cli.py index fb7d0baf..6780c8d1 100644 --- a/src/outerloop/cli.py +++ b/src/outerloop/cli.py @@ -1245,6 +1245,11 @@ def main(argv: list[str] | None = None) -> int: p.add_argument("--file", type=Path, help="read the message text from this path") p = sub.add_parser("limits", help="show live operator ceilings and fleet GPU usage") p.add_argument("--root", help="state root (defaults to OUTERLOOP_ROOT or ~/.outerloop)") + p = sub.add_parser("trajectory", help="show a run's kernel event chain (read-only)") + p.add_argument("run_id") + p.add_argument("--root", help="state root (defaults to OUTERLOOP_ROOT or ~/.outerloop)") + p.add_argument("--verify", action="store_true") + p.add_argument("--json", action="store_true") p = sub.add_parser("status", help="show local runs and endpoint outages (read-only)") p.add_argument("--root", help="state root (defaults to OUTERLOOP_ROOT or ~/.outerloop)") p.add_argument("--json", action="store_true", help="print structured JSON") @@ -1312,6 +1317,44 @@ def main(argv: list[str] | None = None) -> int: return 0 if args.command == "pr": return pr(args) + if args.command == "trajectory": + from outerloop.runstate import run_dir + from outerloop.trajectory import read, verify + + values = env_file_values(keys=("OUTERLOOP_ROOT",)) if not args.root else {} + root = Path( + args.root + or os.environ.get("OUTERLOOP_ROOT") + or values.get("OUTERLOOP_ROOT") + or DEFAULT_LOCAL_ROOT + ).expanduser() + if Path(args.run_id).name != args.run_id or args.run_id in (".", ".."): + parser.error("invalid run id") + directory = run_dir(root, args.run_id) + try: + events = read(directory) + except OSError as exc: + print(f"cannot read trajectory: {exc}", file=sys.stderr) + return 1 + problems = verify(directory, args.run_id) if args.verify else [] + if args.json: + print( + json.dumps( + {"events": events, "problems": problems} if args.verify else events, + sort_keys=True, + ) + ) + else: + for event in events: + print( + f"{event['seq']} {event.get('ts')} {event.get('type')} " + f"{event.get('actor')} {json.dumps(event.get('data'), sort_keys=True)[:160]}" + ) + if not events: + print("no trajectory") + if args.verify: + print("; ".join(problems) if problems else "chain intact") + return 1 if problems and problems != ["no trajectory"] else 0 if args.command == "status": from outerloop.status import collect_status, render_text diff --git a/src/outerloop/launchlog.py b/src/outerloop/launchlog.py index 53f1d859..e009c0ec 100644 --- a/src/outerloop/launchlog.py +++ b/src/outerloop/launchlog.py @@ -12,6 +12,7 @@ from outerloop.provenance import producing_author from outerloop.syscall import Launch, LaunchResult, launch_jobs +from outerloop.trajectory import record as record_event LEDGER = "launches.jsonl" # generous: a run is depth_k launches x sleep_k sleeps x the array width, far below this @@ -37,6 +38,29 @@ def _append(run_dir: Path, rows: list[dict[str, Any]]) -> None: fh.write("\n") for row in rows: fh.write(json.dumps(row, sort_keys=True) + "\n") + record_event( + run_dir, + "launch.submit" if row["event"] == "submitted" else "launch.result", + { + key: row[key] + for key in ( + "event", + "sleep", + "name", + "job_ids", + "commit", + "author", + "minutes", + "array", + "concurrency", + "state", + "exit_code", + "why", + ) + if key in row + }, + at=row["at"], + ) def append_submitted( diff --git a/src/outerloop/orchestrator.py b/src/outerloop/orchestrator.py index 4491879c..8d5c3a4b 100644 --- a/src/outerloop/orchestrator.py +++ b/src/outerloop/orchestrator.py @@ -28,6 +28,7 @@ from typing import TYPE_CHECKING, Any, Literal, Protocol from outerloop.image import apptainer_from_env +from outerloop.trajectory import record as record_event if TYPE_CHECKING: from outerloop.measure import Measure @@ -1532,9 +1533,42 @@ def _budgets_line() -> str: gpu_hours=(hour_ceiling - gpu_hours_used if bench.gpus else None), ) + def _decision(request: SyscallRequest, reason: str = "") -> None: + if request.problem: + verbs = ["request"] + else: + verbs = [] + if request.submit: + verbs.append("submit") + elif request.end: + verbs.append("withdraw" if request.withdraw else "stop") + if request.launches: + verbs.append("launch") + elif request.sleep and not verbs: + verbs.append("sleep") + for verb in verbs or ["request"]: + record_event( + inbox_dir, + "syscall.decision", + { + "verb": verb, + "outcome": "refused" if reason else "accepted", + "reason": redact(reason, redact_secrets), + }, + ) + def _ack(messages: list[Message]) -> None: nonlocal inbox_seq if messages: + record_event( + inbox_dir, + "inbox.deliver", + { + "seq_range": [min(m.seq for m in messages), max(m.seq for m in messages)], + "kinds": [m.kind for m in messages], + "origins": [m.origin or m.source for m in messages], + }, + ) inbox_seq = max(m.seq for m in messages) if on_inbox_delivered is not None: on_inbox_delivered(inbox_seq) @@ -1905,6 +1939,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: "Nothing ran and nothing was charged. " f"Authors may change: {allowed}." ) + _decision(request, problem) scope_refused = True tree_rejected = True presealed = "" @@ -1936,6 +1971,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: return failed continue if read_error: + _decision(request, read_error) return AttemptResult( outcome="session-error", baseline=baseline, @@ -1963,6 +1999,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: ) request = dc_replace(request, problem=problem) if request.problem: + _decision(request, request.problem) # told once, and the leg goes on with nothing staged honoured; # a second conflicting request in one pass ends the leg as a # stop (the refused_once bound stops a refuse/re-ask loop) @@ -1992,6 +2029,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: return failed continue if request.end: + _decision(request) if request.report: session = dc_replace(session, final_text=request.report) if on_stop is not None: @@ -2010,6 +2048,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: if not request.sleep: break if stale_submit: + _decision(request, "base moved; checkpointed without compute") # Budget only the effective checkpoint, never the rejected compute. request = SyscallRequest(launches=()) # suite siblings' paired evals are charged as if measured (the @@ -2135,6 +2174,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: # never inheriting a prior park's. tree_rejected = False submitted = request + _decision(dc_replace(request, launches=())) sleeps_used += 1 evals_charge = evals_gpu_hours( request, @@ -2157,6 +2197,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: problem = str(exc) capacity_refused = True else: + _decision(request) gpu_hours_used += launches_gpu_hours(request, gpus=bench.total_gpus) raise RunParked( phase="author-sleep", @@ -2173,6 +2214,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: sleeps_used=sleeps_used + 1, gpu_hours_used=gpu_hours_used, ) + _decision(request, problem) if refused_once or not _can_resume(): if stale_submit: return AttemptResult( @@ -2320,6 +2362,19 @@ def _not_run_note(request: SyscallRequest | None) -> str: claimed_base, claimed, bench.direction, bench.min_delta, bench.min_delta_rel ) ) + record_event( + inbox_dir, + "measure", + { + "self_reported": True, + "baseline": claimed_base, + "candidate": claimed, + "metric": bench.metric, + "credited": passed, + "outcome": "improved" if passed else "no-improvement", + "head_sha": candidate_sha, + }, + ) return AttemptResult( outcome="improved" if passed else "no-improvement", baseline=claimed_base, @@ -2374,6 +2429,23 @@ def _not_run_note(request: SyscallRequest | None) -> str: measurer=measurer, min_relative_improvement=config.min_relative_improvement, ) + verdict_outcome = ( + outcome.outcome if isinstance(outcome, AttemptResult) else "improved" + ) + # a scope refusal returns before measuring; a failed eval was attempted + if verdict_outcome != "scope-violation": + record_event( + inbox_dir, + "measure", + { + "baseline": outcome.baseline, + "candidate": outcome.candidate, + "metric": bench.metric, + "credited": verdict_outcome == "improved", + "outcome": verdict_outcome, + "head_sha": candidate_sha, + }, + ) except MeasurementPending as pending: # PARK 2 (dispatched candidate/suite, after the session): the wake # reads the cached results and decides — or, on a SUBMITTED park, @@ -2389,6 +2461,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: try: launch_afterany = launcher(candidate_sha, submitted) except CapacityError as exc: + _decision(dc_replace(submitted, submit=False), str(exc)) append( inbox_dir, Message( @@ -2411,6 +2484,7 @@ def _not_run_note(request: SyscallRequest | None) -> str: ) submitted = dc_replace(submitted, launches=()) else: + _decision(dc_replace(submitted, submit=False)) launches_used += len(submitted.launches) gpu_hours_used += launches_gpu_hours(submitted, gpus=bench.total_gpus) raise RunParked( diff --git a/src/outerloop/session_evidence.py b/src/outerloop/session_evidence.py index 29dcbfa6..a072100a 100644 --- a/src/outerloop/session_evidence.py +++ b/src/outerloop/session_evidence.py @@ -23,6 +23,8 @@ from pathlib import Path from typing import TYPE_CHECKING, Any +from outerloop.trajectory import session_event + if TYPE_CHECKING: from outerloop.harness import SessionResult @@ -287,6 +289,7 @@ def wrapped( if backend == "claude": # read before the run: an auto-update can move the link mid-session state["cli_version"] = cli_version(str(getattr(self, "binary", "") or "")) + session_event(workspace, "session.start", backend, self.model, resume_session_id or "") token = active.set(state) try: result = run(self, brief_text, workspace, resume_session_id) @@ -317,7 +320,20 @@ def wrapped( ) except Exception: log.warning("could not retain session evidence", exc_info=True) + session_event( + workspace, "session.end", backend, self.model, result.session_id, result=result + ) return result + except Exception as exc: + session_event( + workspace, + "session.end", + backend, + self.model, + resume_session_id or "", + outcome=type(exc).__name__, + ) + raise finally: active.reset(token) diff --git a/src/outerloop/steward.py b/src/outerloop/steward.py index dc918bcf..3c559629 100644 --- a/src/outerloop/steward.py +++ b/src/outerloop/steward.py @@ -518,6 +518,10 @@ def changed_paths() -> list[str]: "resume_session_id": session.session_id or "", } ) + # written before finish_run, which keeps an existing report and + # records its hash in the trajectory's ending event + report_path = run_dir / "report.md" + _best_effort("run report", lambda: report_path.write_text(report), secrets) _best_effort( "final record", lambda: finish_run( @@ -533,8 +537,6 @@ def changed_paths() -> list[str]: ), secrets, ) - report_path = run_dir / "report.md" - _best_effort("run report", lambda: report_path.write_text(report), secrets) if issue_number: _best_effort( "issue report", @@ -696,6 +698,15 @@ def _arm() -> None: } ) report_path = run_dir / "report.md" + # before finish_run, so the ending event hashes the report that stays + wrote = _best_effort( + "error report", + lambda: report_path.write_text( + f"# Steward report — {config.target} / {config.benchmark}\n" + f"Outcome: **{outcome_label}**\nNote: {note}\n" + ), + secrets, + ) _best_effort( "ending record", lambda: finish_run( @@ -711,14 +722,6 @@ def _arm() -> None: ), secrets, ) - wrote = _best_effort( - "error report", - lambda: report_path.write_text( - f"# Steward report — {config.target} / {config.benchmark}\n" - f"Outcome: **{outcome_label}**\nNote: {note}\n" - ), - secrets, - ) if issue_number: if api_outage: release = ( diff --git a/src/outerloop/tick.py b/src/outerloop/tick.py index 65983b93..ac317801 100644 --- a/src/outerloop/tick.py +++ b/src/outerloop/tick.py @@ -89,6 +89,7 @@ save_record, update_lease_holder, ) +from outerloop.trajectory import record as record_event log = logging.getLogger(__name__) @@ -1189,6 +1190,8 @@ def sweep( def wake(record: RunRecord, reason: str, tag: str) -> None: if dry_run or _wake(root, record, reason, dispatcher, now, holder): woken.append((record.run_id, tag)) + if not dry_run: + record_event(run_dir(root, record.run_id), "wake", {"reason": reason, "tag": tag}) ledger_blocked: set[str] = set() if github is not None and not dry_run: diff --git a/src/outerloop/trajectory.py b/src/outerloop/trajectory.py new file mode 100644 index 00000000..a2d3d1c5 --- /dev/null +++ b/src/outerloop/trajectory.py @@ -0,0 +1,185 @@ +"""Kernel-owned, best-effort per-run event chain.""" + +from __future__ import annotations + +import contextlib +import fcntl +import hashlib +import json +import logging +import time +from collections.abc import Iterable +from pathlib import Path +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from outerloop.harness import SessionResult + +log = logging.getLogger(__name__) +LEDGER = "trajectory.jsonl" +MAX_LEDGER_BYTES = 4_000_000 +_cap_warned: set[Path] = set() + + +def _event(line: bytes) -> dict[str, Any] | None: + try: + row = json.loads(line) + except (ValueError, UnicodeError): + return None + if not isinstance(row, dict) or type(row.get("seq")) is not int: + return None + return row + + +def record( + run_dir: Path, + type: str, + data: dict[str, Any], + *, + actor: str = "kernel", + refs: Iterable[dict[str, str]] = (), + at: float | None = None, +) -> None: + """Preserve torn bytes, then continue from the last parseable complete line. + + The directory name is the run id. The lock covers reading and appending; + malformed lines remain visible to verify, including after recovery. + """ + try: + path = run_dir / LEDGER + run_dir.mkdir(parents=True, exist_ok=True) + with (run_dir / "trajectory.lock").open("a") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + with path.open("a+b") as stream: + stream.seek(0, 2) + size = stream.tell() + if size >= MAX_LEDGER_BYTES: + _warn_cap(path) + return + stream.seek(0) + raw = stream.read() + seq, prev = 0, hashlib.sha256(run_dir.name.encode()).hexdigest() + for line in reversed(raw.splitlines(keepends=True)): + if (row := _event(line.removesuffix(b"\n"))) is not None: + seq = row["seq"] + 1 + prev = hashlib.sha256(line.removesuffix(b"\n")).hexdigest() + break + event = dict( + seq=seq, + ts=time.time() if at is None else float(at), + type=type, + actor=actor, + data=data, + refs=list(refs), + prev=prev, + ) + prefix = b"\n" if raw and not raw.endswith(b"\n") else b"" + encoded = prefix + json.dumps(event, sort_keys=True).encode() + b"\n" + if size + len(encoded) > MAX_LEDGER_BYTES: + _warn_cap(path) + return + stream.write(encoded) + except Exception as exc: + log.warning("could not write trajectory in %s: %s", run_dir, exc) + + +def _warn_cap(path: Path) -> None: + if path not in _cap_warned: + _cap_warned.add(path) + log.warning("trajectory size cap reached: %s", path) + + +def file_refs(run_dir: Path, *paths: Path) -> list[dict[str, str]]: + """Hash existing host-owned artifacts once, outside the append lock.""" + refs = [] + for path in paths: + try: + relative = path.resolve().relative_to(run_dir.resolve()) + with path.open("rb") as stream: + digest = hashlib.file_digest(stream, "sha256").hexdigest() + refs.append({"path": str(relative), "sha256": digest}) + except (OSError, ValueError): + continue + return refs + + +def read(run_dir: Path) -> list[dict]: + """Missing legacy histories are empty; damaged lines are skipped. Any other + read failure raises, so an unreadable chain is never shown as absent.""" + try: + with (run_dir / LEDGER).open("rb") as stream: + raw = stream.read(MAX_LEDGER_BYTES) + except FileNotFoundError: + return [] + return [ + row + for line in raw.splitlines(keepends=True) + if line.endswith(b"\n") and (row := _event(line[:-1])) is not None + ] + + +def verify(run_dir: Path, run_id: str) -> list[str]: + """Check exact line bytes and sequence continuity without changing files.""" + try: + with (run_dir / LEDGER).open("rb") as stream: + raw = stream.read(MAX_LEDGER_BYTES + 1) + except FileNotFoundError: + return ["no trajectory"] + except OSError as exc: + return [f"cannot read trajectory: {exc}"] + problems = [] + if len(raw) > MAX_LEDGER_BYTES: + problems.append("trajectory exceeds size cap") + seq, prev = 0, hashlib.sha256(run_id.encode()).hexdigest() + for number, line in enumerate(raw.splitlines(keepends=True), 1): + if not line.endswith(b"\n") or (row := _event(line[:-1])) is None: + problems.append(f"line {number}: torn or malformed line") + continue + if row["seq"] != seq: + problems.append(f"line {number}: sequence gap (expected {seq}, got {row['seq']})") + if row.get("prev") != prev: + problems.append( + f"line {number}: previous hash mismatch (line {number - 1} changed or missing)" + ) + if not all(key in row for key in ("ts", "type", "actor", "data", "refs")): + problems.append(f"line {number}: missing event fields") + seq, prev = row["seq"] + 1, hashlib.sha256(line[:-1]).hexdigest() + return problems + + +def session_event( + workspace: Path, + type: str, + backend: str, + model: str, + session_id: str, + *, + result: SessionResult | None = None, + outcome: str = "", +) -> None: + """Harness invocations outside a kernel run have no run trajectory.""" + try: + directory = next((p for p in workspace.parents if p.parent.name == "runs"), None) + if directory is None: + return + state = json.loads((directory / "state.json").read_text()) + data = { + "backend": backend, + "model": model, + "session_id": session_id, + "stage": state.get("stage", {}).get("phase", "author"), + } + refs = [] + if result is not None: + data.update(outcome=result.stop_reason, is_error=result.is_error) + if result.session_record_path: + with contextlib.suppress(OSError, ValueError, TypeError, AttributeError): + evidence = json.loads(Path(result.session_record_path).read_text()) + native = evidence.get("artifacts", {}).get("native", {}) + if native.get("path"): + refs = file_refs(directory, Path(native["path"])) + elif outcome: + data.update(outcome=outcome, is_error=True) + record(directory, type, data, refs=refs) + except Exception as exc: + log.warning("could not record session trajectory: %s", exc) diff --git a/tests/test_attempt.py b/tests/test_attempt.py index 5fc9a5aa..6f56b0ee 100644 --- a/tests/test_attempt.py +++ b/tests/test_attempt.py @@ -112,6 +112,10 @@ def test_park_run_appends_the_launch_ledger(tmp_path) -> None: {"name": "a", "minutes": 5, "artifacts": [], "why": "probe a"}, {"name": "sw", "minutes": 5, "artifacts": [], "array": 2}, ] + from outerloop.trajectory import read, verify + + assert [event["type"] for event in read(ledger_dir)] == ["launch.submit", "launch.submit"] + assert verify(ledger_dir, record.run_id) == [] # a re-park of the same sleep (a multi-stage gate) adds no second record _park_run( tmp_path, diff --git a/tests/test_attempt_review.py b/tests/test_attempt_review.py index f6280a19..a3b845ba 100644 --- a/tests/test_attempt_review.py +++ b/tests/test_attempt_review.py @@ -938,6 +938,12 @@ def refused_push(self, branch): ) assert load_record(root, record.run_id).stage["hypothesis"] == "Preserve the refused direction." assert outcome.outcome == "publish-refused" + from outerloop.trajectory import read + + decision = read(ws.parent)[-1] + assert decision["type"] == "syscall.decision" + assert decision["data"]["verb"] == "submit" + assert decision["data"]["outcome"] == "refused" assert _git(bare, "rev-parse", PR_BRANCH).strip() == head message = pending(ws.parent, 0)[-1] assert message.kind == ("base-moved" if reason in ("human", "push") else "note") @@ -1535,6 +1541,15 @@ def fail(*args, **kwargs): assert released == ["refs/test/snapshot"] assert load_record(root, "tsp-r1").state == ENDED assert "notebook unavailable" in caplog.text + from outerloop.trajectory import read, verify + + directory = run_dir(root, record.run_id) + events = read(directory) + assert [event["type"] for event in events] == ["ending"] + assert events[0]["data"] == {"state": "merged", "reason": "PR merged"} + assert events[0]["ts"] > NOW # wall-clock time, not the caller's tick time + assert events[0]["refs"][0]["path"] == "report.md" + assert verify(directory, record.run_id) == [] @pytest.mark.parametrize("edit_ledger", [False, True]) diff --git a/tests/test_contract_switches.py b/tests/test_contract_switches.py index 7ee57f37..a7faf10c 100644 --- a/tests/test_contract_switches.py +++ b/tests/test_contract_switches.py @@ -97,6 +97,21 @@ def test_self_report_skips_measurement_panel_and_compute( assert result.provenance == "self_reported" assert result.candidate == candidate and result.baseline == baseline assert result.candidate_sha and result.measured_paths + from outerloop.trajectory import read + from test_orchestrator import run_dir + + measures = [event["data"] for event in read(run_dir(tmp_path)) if event["type"] == "measure"] + assert measures == [ + { + "self_reported": True, + "baseline": baseline, + "candidate": candidate, + "metric": "mean_tour_length", + "credited": True, + "outcome": "improved", + "head_sha": result.candidate_sha, + } + ] assert evaluator.calls == [] and result.panel_rounds == 0 assert meters[-1] == (0, 1, 0.0) assert "self-reported" in result.report(CONFIG) diff --git a/tests/test_launchlog.py b/tests/test_launchlog.py index 4742ec68..ed04f9c2 100644 --- a/tests/test_launchlog.py +++ b/tests/test_launchlog.py @@ -152,6 +152,11 @@ def test_experiments_rows_come_from_the_ledger_including_each_jobs_last_line( ] assert rows[1]["why"] == "" and rows[1]["array"] == 2 and rows[1]["concurrency"] == 1 assert len(history(tmp_path)[0]["jobs"]) == 1 # the duplicate ended row was skipped + from outerloop.trajectory import read + + events = [event["data"] for event in read(tmp_path) if event["type"] == "launch.result"] + assert events[0] == {"event": "ended", "sleep": 1, "name": "wd", "state": "", "exit_code": 0} + assert all("last_line" not in event for event in events) def test_legacy_commit_tolerance_and_retry(tmp_path: Path) -> None: diff --git a/tests/test_messages.py b/tests/test_messages.py index 547fed26..2fc597e3 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -173,6 +173,14 @@ def test_missing_live_recipient_refused(tmp_path, state, target): note = pending(tmp_path / "runs" / sender.run_id, 0)[0] assert note.source == "kernel" and note.kind == "note" assert "Message #1" in note.payload["text"] and "no live run" in note.payload["text"] + from outerloop.trajectory import read + + decisions = [ + event["data"] + for event in read(tmp_path / "runs" / sender.run_id) + if event["type"] == "syscall.decision" + ] + assert [decision["outcome"] for decision in decisions] == ["refused"] assert pending(tmp_path / "runs" / recipient.run_id, 0) == [] assert not wake_pending(tmp_path / "runs" / sender.run_id, sender) diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index ff1a1bf3..567ee16e 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -331,6 +331,12 @@ def test_checkpoint_sleep_parks_with_no_dependency(tmp_path: Path) -> None: run_climb(tmp_path, [], contract=DEEP_CONTRACT, launcher=_fake_launcher([])) assert exc.value.afterany == "" and exc.value.sleeps_used == 1 assert exc.value.launches_used == 0 # a checkpoint burns only the sleep + from outerloop.trajectory import read + + decisions = [ + event["data"] for event in read(run_dir(tmp_path)) if event["type"] == "syscall.decision" + ] + assert decisions == [{"verb": "sleep", "outcome": "accepted", "reason": ""}] @pytest.mark.parametrize("capacity_wait", [False, True]) @@ -504,6 +510,9 @@ def test_identical_resubmit_is_told_so_without_a_second_gate(tmp_path: Path) -> assert harness.resumes == [None, "s1", "s2"] assert len(evaluator.calls) == 2 assert "identical to the candidate the gate already measured" in harness.prompts[2] + from outerloop.trajectory import read + + assert len([event for event in read(run_dir(tmp_path)) if event["type"] == "measure"]) == 1 def test_conceding_after_an_errored_eval_ends_without_another_gate(tmp_path: Path) -> None: diff --git a/tests/test_session_evidence.py b/tests/test_session_evidence.py index 1f4ae2e5..8b999d69 100644 --- a/tests/test_session_evidence.py +++ b/tests/test_session_evidence.py @@ -143,6 +143,9 @@ def run(self, brief, workspace, resume_session_id=None): ], ) def test_evidence_redaction_hash_cap_resume_and_retention(tmp_path, monkeypatch, backend, relative): + tmp_path = tmp_path / "runs" / "synthetic" + tmp_path.mkdir(parents=True) + (tmp_path / "state.json").write_text(json.dumps({"stage": {"phase": "author-sleep"}})) ws = tmp_path / "ws" ws.mkdir() home = tmp_path / "ws-home" @@ -153,6 +156,19 @@ def test_evidence_redaction_hash_cap_resume_and_retention(tmp_path, monkeypatch, first = fake_run(backend, ws, native, "brief synthetic-secret") second = fake_run(backend, ws, native, "wake synthetic-secret", "session-1") assert first.session_record_path != second.session_record_path + from outerloop.trajectory import read, verify + + events = read(tmp_path) + assert [e["type"] for e in events] == ["session.start", "session.end"] * 2 + assert events[0]["data"]["stage"] == "author-sleep" + assert events[1]["data"]["session_id"] == "session-1" + assert len(events[1]["refs"]) == 1 + native_ref = events[1]["refs"][0] + assert ( + hashlib.sha256((tmp_path / native_ref["path"]).read_bytes()).hexdigest() + == native_ref["sha256"] + ) + assert verify(tmp_path, "synthetic") == [] record = json.loads(Path(second.session_record_path).read_text()) assert record["session_id"] == "session-1" assert record["resume_session_id"] == "session-1" diff --git a/tests/test_trajectory.py b/tests/test_trajectory.py new file mode 100644 index 00000000..3fc8fe75 --- /dev/null +++ b/tests/test_trajectory.py @@ -0,0 +1,89 @@ +import json +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from outerloop import trajectory + + +def test_chain_detects_edits_and_gaps(tmp_path: Path) -> None: + for n in range(3): + trajectory.record(tmp_path, "measure", {"value": n}, at=n) + path = tmp_path / trajectory.LEDGER + original = path.read_bytes() + assert trajectory.verify(tmp_path, tmp_path.name) == [] + assert [e["seq"] for e in trajectory.read(tmp_path)] == [0, 1, 2] + lines = original.splitlines(keepends=True) + lines[1] = lines[1].replace(b'"value": 1', b'"value": 9') + path.write_bytes(b"".join(lines)) + assert "line 2 changed" in " ".join(trajectory.verify(tmp_path, tmp_path.name)) + path.write_bytes(b"".join(original.splitlines(keepends=True)[::2])) + assert "sequence gap (expected 1, got 2)" in " ".join( + trajectory.verify(tmp_path, tmp_path.name) + ) + + +@pytest.mark.parametrize("tail", [b'{"seq":', b"\xff"]) +def test_torn_line_remains_visible_after_recovery(tmp_path: Path, tail: bytes) -> None: + trajectory.record(tmp_path, "wake", {}) + path = tmp_path / trajectory.LEDGER + with path.open("ab") as stream: + stream.write(tail) + trajectory.record(tmp_path, "ending", {}) + trajectory.record(tmp_path, "wake", {}) + assert [e["seq"] for e in trajectory.read(tmp_path)] == [0, 1, 2] + assert trajectory.verify(tmp_path, tmp_path.name) == ["line 2: torn or malformed line"] + assert tail in path.read_bytes() + + +def test_complete_json_missing_newline(tmp_path: Path) -> None: + trajectory.record(tmp_path, "wake", {}) + path = tmp_path / trajectory.LEDGER + path.write_bytes(path.read_bytes().rstrip(b"\n")) + assert trajectory.verify(tmp_path, tmp_path.name) == ["line 1: torn or malformed line"] + trajectory.record(tmp_path, "ending", {}) + assert trajectory.verify(tmp_path, tmp_path.name) == [] + + +def test_legacy_run_without_trajectory(tmp_path: Path, capsys) -> None: + from outerloop.cli import main + + directory = tmp_path / "runs" / "legacy" + directory.mkdir(parents=True) + (directory / "state.json").write_text(json.dumps({"run_id": "legacy", "state": "ended"})) + assert trajectory.read(directory) == [] + assert trajectory.verify(directory, "legacy") == ["no trajectory"] + assert main(["trajectory", "legacy", "--root", str(tmp_path), "--verify"]) == 0 + assert "no trajectory" in capsys.readouterr().out + assert not (directory / trajectory.LEDGER).exists() + # an unreadable chain is an error, never "no trajectory" + (directory / trajectory.LEDGER).mkdir() + assert main(["trajectory", "legacy", "--root", str(tmp_path)]) == 1 + assert "cannot read trajectory" in capsys.readouterr().err + + +def test_concurrent_appends_and_best_effort_limits(tmp_path: Path, monkeypatch, caplog) -> None: + with ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(lambda n: trajectory.record(tmp_path, "wake", {"n": n}), range(12))) + assert len(trajectory.read(tmp_path)) == 12 + assert trajectory.verify(tmp_path, tmp_path.name) == [] + monkeypatch.setattr(trajectory, "MAX_LEDGER_BYTES", 1) + for _ in range(2): + trajectory.record(tmp_path, "wake", {}) + assert caplog.text.count("size cap reached") == 1 + blocked = tmp_path / "file" + blocked.write_text("") + trajectory.record(blocked, "wake", {}) + assert "could not write trajectory" in caplog.text + + +def test_non_serializable_data_is_best_effort(tmp_path: Path, caplog) -> None: + circular: dict = {} + circular["self"] = circular + for value in (tmp_path, object(), circular): + trajectory.record(tmp_path, "measure", {"value": value}) + assert caplog.text.count("could not write trajectory") == 3 + assert trajectory.read(tmp_path) == [] + trajectory.record(tmp_path, "wake", {}) + assert trajectory.verify(tmp_path, tmp_path.name) == []