#!/usr/bin/env python3 """Maintain Blacksite's compact, resumable Codex session state. The state is intentionally Markdown rather than an opaque database so a resumed agent can read it cheaply. This module uses only the Python standard library. """ from __future__ import annotations import argparse import contextlib import copy import hashlib import io import os import subprocess import sys import tempfile import tomllib from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable, Sequence TITLE = "# Codex session state" WORKFLOW_PATH = Path(".codex/workflow.toml") VALID_STATES = ( "Implementing", "Engineering-complete", "Acceptance-in-progress", "Candidate-ready", ) CLASSIFICATIONS = { "active-slice-refinement": "active-slice refinement", "active-slice refinement": "active-slice refinement", "newly-discovered-blocker": "newly discovered blocker", "newly discovered blocker": "newly discovered blocker", "discovered-blocker": "newly discovered blocker", "added-acceptance-criterion": "added acceptance criterion", "added acceptance criterion": "added acceptance criterion", "separate-follow-up": "separate follow-up", "separate follow-up": "separate follow-up", } class StateError(RuntimeError): """The state operation cannot be completed safely.""" def clean_text(value: str | None, default: str = "None.") -> str: """Keep state entries single-line and compact.""" if value is None: return default compact = " ".join(value.split()) return compact or default def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def find_repository(start: Path | None = None) -> Path: """Locate the nearest repository containing the workflow configuration.""" override = os.environ.get("BLACKSITE_WORKSPACE_ROOT") configured = None config_path = (start or Path.cwd()) / ".codex/config.toml" if not override and config_path.is_file(): with config_path.open("rb") as handle: configured = tomllib.load(handle).get("workspace_root") candidates = [Path(override or configured)] if (override or configured) else [start or Path.cwd()] if start is None and not override: candidates.append(Path(__file__).parent) visited: set[Path] = set() for candidate in candidates: candidate = candidate.expanduser().absolute() for current in (candidate, *candidate.parents): canonical = current.resolve(strict=False) if canonical in visited: continue visited.add(canonical) if (current / WORKFLOW_PATH).is_file(): return current searched = ", ".join(str(path) for path in candidates) raise StateError(f"could not locate {WORKFLOW_PATH} from {searched}") def load_workflow(root: Path) -> dict[str, Any]: path = root / WORKFLOW_PATH try: with path.open("rb") as handle: workflow = tomllib.load(handle) except OSError as error: raise StateError(f"cannot read workflow configuration {path}: {error}") from error except tomllib.TOMLDecodeError as error: raise StateError(f"invalid workflow configuration {path}: {error}") from error if workflow.get("version") != 1: raise StateError(f"unsupported or missing workflow version in {path}") return workflow def configured_path(root: Path, raw: str, *, label: str) -> Path: candidate = Path(os.path.expandvars(os.path.expanduser(raw))) if not candidate.is_absolute(): candidate = root / candidate candidate = candidate.absolute() try: candidate.resolve(strict=False).relative_to(root.resolve(strict=False)) except ValueError as error: raise StateError(f"configured {label} must remain inside the repository: {candidate}") from error return candidate def state_path(root: Path, workflow: dict[str, Any]) -> Path: session = workflow.get("session", {}) if not isinstance(session, dict): raise StateError("[session] must be a TOML table") raw = session.get("state_file", ".codex/session/STATE.md") if not isinstance(raw, str) or not raw: raise StateError("session.state_file must be a non-empty path string") return configured_path(root, raw, label="session.state_file") def max_state_lines(workflow: dict[str, Any]) -> int: session = workflow.get("session", {}) value = session.get("max_state_lines", 180) if isinstance(session, dict) else 180 if not isinstance(value, int) or value < 48: raise StateError("session.max_state_lines must be an integer of at least 48") return value def run_git(root: Path, *arguments: str, check: bool = True) -> subprocess.CompletedProcess[str]: try: result = subprocess.run( ["git", "-C", str(root), *arguments], check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) except OSError as error: raise StateError(f"cannot execute git: {error}") from error if check and result.returncode != 0: detail = clean_text(result.stderr or result.stdout, "git command failed") raise StateError(detail) return result def worktree_summary(root: Path) -> str: lines = run_git(root, "status", "--porcelain=v1", "--untracked-files=normal").stdout.splitlines() if not lines: return "clean" staged = modified = deleted = untracked = conflicts = 0 conflict_codes = {"DD", "AU", "UD", "UA", "DU", "AA", "UU"} for line in lines: if len(line) < 2: continue code = line[:2] if code == "??": untracked += 1 continue if code in conflict_codes: conflicts += 1 if code[0] not in {" ", "?"}: staged += 1 if code[1] not in {" ", "?"}: modified += 1 if "D" in code: deleted += 1 return ( "dirty " f"(staged={staged}, modified={modified}, deleted={deleted}, " f"untracked={untracked}, conflicts={conflicts})" ) @dataclass class RepositoryState: root: str real_path: str branch: str head: str worktree: str active_processes: str = "None recorded." def repository_snapshot(root: Path, active_processes: str = "None recorded.") -> RepositoryState: git_root = Path(run_git(root, "rev-parse", "--show-toplevel").stdout.strip()) real_path = git_root.resolve(strict=False) branch_result = run_git(root, "symbolic-ref", "--quiet", "--short", "HEAD", check=False) branch = branch_result.stdout.strip() if branch_result.returncode == 0 else "(detached)" head = run_git(root, "rev-parse", "HEAD").stdout.strip() return RepositoryState( root=str(git_root), real_path=str(real_path), branch=branch, head=head, worktree=worktree_summary(root), active_processes=clean_text(active_processes, "None recorded."), ) @dataclass class AcceptanceCriterion: text: str checked: bool = False @dataclass class ScopeDelta: title: str source: str classification: str requirement: str evidence_valid: str invalidated_gates: str tracker_sync: str @dataclass class IntentionalFile: path: str reason: str @dataclass class GateRecord: gate: str status: str digest: str command: str date: str @dataclass class SessionState: repository: RepositoryState goal: str current_state: str = "Implementing" issues: str = "None." milestone: str = "None." active_slice: str = "None." non_goals: str = "None." acceptance: list[AcceptanceCriterion] = field(default_factory=list) deltas: list[ScopeDelta] = field(default_factory=list) files: list[IntentionalFile] = field(default_factory=list) gates: list[GateRecord] = field(default_factory=list) native_evidence: list[str] = field(default_factory=list) decisions: list[str] = field(default_factory=list) remote_actions: list[str] = field(default_factory=list) next_action: str = "Define the next action." def section_map(text: str) -> dict[str, list[str]]: sections: dict[str, list[str]] = {} current: str | None = None for line in text.splitlines(): if line.startswith("## "): current = line[3:].strip() sections[current] = [] elif current is not None: sections[current].append(line) return sections def field_value(lines: Iterable[str], name: str, default: str = "None.") -> str: prefix = f"- {name}:" for line in lines: if line.startswith(prefix): return clean_text(line[len(prefix) :], default) return default def bullet_values(lines: Iterable[str]) -> list[str]: values: list[str] = [] for line in lines: if line.startswith("- ") and not line.startswith("- ["): value = clean_text(line[2:]) if value != "None.": values.append(value) return values def parse_deltas(lines: list[str]) -> list[ScopeDelta]: chunks: list[tuple[str, list[str]]] = [] title: str | None = None body: list[str] = [] for line in lines: if line.startswith("### Delta "): if title is not None: chunks.append((title, body)) heading = line.partition("—")[2].strip() title = heading or "Untitled scope delta" body = [] elif title is not None: body.append(line) if title is not None: chunks.append((title, body)) return [ ScopeDelta( title=clean_text(title), source=field_value(body, "Source"), classification=field_value(body, "Classification"), requirement=field_value(body, "Added/changed requirement"), evidence_valid=field_value(body, "Prior evidence still valid"), invalidated_gates=field_value(body, "Invalidated gates"), tracker_sync=field_value(body, "Tracker sync required", "no"), ) for title, body in chunks ] def parse_state(text: str) -> SessionState: if not text.startswith(TITLE): raise StateError(f"state file must begin with {TITLE!r}") sections = section_map(text) repository_lines = sections.get("Repository", []) task_lines = sections.get("Active task", []) repository = RepositoryState( root=field_value(repository_lines, "Root"), real_path=field_value(repository_lines, "Real path"), branch=field_value(repository_lines, "Branch"), head=field_value(repository_lines, "HEAD"), worktree=field_value(repository_lines, "Worktree status summary"), active_processes=field_value(repository_lines, "Active processes/windows", "None recorded."), ) acceptance: list[AcceptanceCriterion] = [] for line in sections.get("Acceptance target", []): stripped = line.strip() if stripped.startswith("- [ ] "): acceptance.append(AcceptanceCriterion(clean_text(stripped[6:]))) elif stripped.lower().startswith("- [x] "): acceptance.append(AcceptanceCriterion(clean_text(stripped[6:]), checked=True)) files: list[IntentionalFile] = [] for value in bullet_values(sections.get("Files intentionally changed", [])): path, separator, reason = value.partition(" — ") files.append(IntentionalFile(path, reason if separator else "Reason not recorded.")) gates: list[GateRecord] = [] for value in bullet_values(sections.get("Verification ledger summary", [])): parts = value.split(" — ", 4) if len(parts) == 5: gates.append(GateRecord(*parts)) next_values = bullet_values(sections.get("Exact next action", [])) return SessionState( repository=repository, goal=field_value(task_lines, "Goal"), current_state=field_value(task_lines, "Current state", "Implementing"), issues=field_value(task_lines, "Gitea issue(s)"), milestone=field_value(task_lines, "Gitea milestone"), active_slice=field_value(task_lines, "Active slice"), non_goals=field_value(task_lines, "Non-goals"), acceptance=acceptance, deltas=parse_deltas(sections.get("Scope deltas", [])), files=files, gates=gates, native_evidence=bullet_values(sections.get("Native evidence", [])), decisions=bullet_values(sections.get("Decisions", [])), remote_actions=bullet_values(sections.get("Remote actions already performed", [])), next_action=next_values[0] if next_values else "Define the next action.", ) def render_lines(state: SessionState) -> list[str]: lines = [ TITLE, "", "## Repository", f"- Root: {clean_text(state.repository.root)}", f"- Real path: {clean_text(state.repository.real_path)}", f"- Branch: {clean_text(state.repository.branch)}", f"- HEAD: {clean_text(state.repository.head)}", f"- Worktree status summary: {clean_text(state.repository.worktree)}", f"- Active processes/windows: {clean_text(state.repository.active_processes, 'None recorded.')}", "", "## Active task", f"- Goal: {clean_text(state.goal)}", f"- Current state: {clean_text(state.current_state, 'Implementing')}", f"- Gitea issue(s): {clean_text(state.issues)}", f"- Gitea milestone: {clean_text(state.milestone)}", f"- Active slice: {clean_text(state.active_slice)}", f"- Non-goals: {clean_text(state.non_goals)}", "", "## Acceptance target", ] lines.extend( f"- [{'x' if criterion.checked else ' '}] {clean_text(criterion.text)}" for criterion in state.acceptance ) lines.extend(["", "## Scope deltas"]) for index, delta in enumerate(state.deltas, start=1): lines.extend( [ f"### Delta {index} — {clean_text(delta.title)}", f"- Source: {clean_text(delta.source)}", f"- Classification: {clean_text(delta.classification)}", f"- Added/changed requirement: {clean_text(delta.requirement)}", f"- Prior evidence still valid: {clean_text(delta.evidence_valid)}", f"- Invalidated gates: {clean_text(delta.invalidated_gates)}", f"- Tracker sync required: {clean_text(delta.tracker_sync, 'no')}", "", ] ) lines.extend(["## Files intentionally changed"]) lines.extend(f"- {clean_text(item.path)} — {clean_text(item.reason)}" for item in state.files) lines.extend(["", "## Verification ledger summary"]) lines.extend( f"- {clean_text(item.gate)} — {clean_text(item.status)} — {clean_text(item.digest)} — " f"{clean_text(item.command)} — {clean_text(item.date)}" for item in state.gates ) lines.extend(["", "## Native evidence"]) lines.extend(f"- {clean_text(item)}" for item in state.native_evidence) lines.extend(["", "## Decisions"]) lines.extend(f"- {clean_text(item)}" for item in state.decisions) lines.extend(["", "## Remote actions already performed"]) lines.extend(f"- {clean_text(item)}" for item in state.remote_actions) lines.extend( [ "", "## Exact next action", f"- {clean_text(state.next_action, 'Define the next action.')}", ] ) return lines def compact_state(state: SessionState, limit: int) -> SessionState: """Discard oldest ledger detail until the state fits its configured budget.""" compacted = copy.deepcopy(state) collections: list[tuple[list[Any], int]] = [ (compacted.gates, 1), (compacted.native_evidence, 0), (compacted.remote_actions, 0), (compacted.decisions, 1), (compacted.deltas, 1), (compacted.files, 1), (compacted.acceptance, 1), ] while len(render_lines(compacted)) > limit: for values, keep in collections: if len(values) > keep: values.pop(0) break else: raise StateError( f"required state structure exceeds session.max_state_lines={limit}" ) return compacted def ensure_state_is_ignored(root: Path, path: Path) -> None: relative = path.resolve(strict=False).relative_to(root.resolve(strict=False)).as_posix() result = run_git(root, "check-ignore", "--quiet", "--", relative, check=False) if result.returncode != 0: raise StateError( f"refusing to write tracked session state; add /{relative} or its directory to .gitignore" ) def write_state(root: Path, workflow: dict[str, Any], state: SessionState) -> Path: path = state_path(root, workflow) ensure_state_is_ignored(root, path) state = compact_state(state, max_state_lines(workflow)) text = "\n".join(render_lines(state)) + "\n" path.parent.mkdir(parents=True, exist_ok=True) temporary: Path | None = None try: with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False, ) as handle: handle.write(text) handle.flush() os.fsync(handle.fileno()) temporary = Path(handle.name) os.replace(temporary, path) finally: if temporary is not None and temporary.exists(): temporary.unlink() return path def read_state(root: Path, workflow: dict[str, Any]) -> SessionState: path = state_path(root, workflow) try: return parse_state(path.read_text(encoding="utf-8")) except FileNotFoundError as error: raise StateError(f"no session state exists at {path}; run state.py init") from error except OSError as error: raise StateError(f"cannot read session state {path}: {error}") from error def refresh_repository(root: Path, state: SessionState) -> list[str]: previous = state.repository current = repository_snapshot(root, previous.active_processes) changes: list[str] = [] for label, before, after in ( ("real path", previous.real_path, current.real_path), ("branch", previous.branch, current.branch), ("HEAD", previous.head, current.head), ): if before not in {"None.", after}: changes.append(f"{label}: {before} -> {after}") state.repository = current return changes def normalize_repo_file(root: Path, raw: str) -> str: candidate = Path(raw).expanduser() if not candidate.is_absolute(): candidate = root / candidate candidate = candidate.resolve(strict=False) repository = root.resolve(strict=False) try: return candidate.relative_to(repository).as_posix() except ValueError as error: raise StateError(f"intentional file must be inside the repository: {candidate}") from error def replace_or_append_file(state: SessionState, path: str, reason: str) -> None: state.files = [item for item in state.files if item.path != path] state.files.append(IntentionalFile(path, clean_text(reason, "Reason not recorded."))) def replace_or_append_gate(state: SessionState, record: GateRecord) -> None: state.gates = [item for item in state.gates if item.gate != record.gate] state.gates.append(record) def candidate_commit(root: Path, revision: str) -> str: result = run_git(root, "rev-parse", "--verify", f"{revision}^{{commit}}", check=False) if result.returncode != 0: raise StateError(f"candidate revision is not a commit: {revision}") return result.stdout.strip() def git_bytes(root: Path, *arguments: str) -> bytes: try: result = subprocess.run( ["git", "-C", str(root), *arguments], check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) except OSError as error: raise StateError(f"cannot execute git: {error}") from error if result.returncode != 0: detail = clean_text(result.stderr.decode("utf-8", "replace"), "git command failed") raise StateError(detail) return result.stdout def _dirty_tree_digest_once(root: Path) -> str: status_arguments = ("status", "--porcelain=v1", "-z", "--untracked-files=all") status_before = git_bytes(root, *status_arguments) if not status_before: raise StateError("--dirty-tree requires a dirty worktree") digest = hashlib.sha256() digest.update(b"blacksite-dirty-tree-v1\0") digest.update(git_bytes(root, "rev-parse", "HEAD")) digest.update(b"\0status\0") digest.update(status_before) digest.update(b"\0tracked-diff\0") digest.update( git_bytes( root, "diff", "--binary", "--no-ext-diff", "--submodule=diff", "HEAD", "--", ) ) untracked = git_bytes(root, "ls-files", "--others", "--exclude-standard", "-z") repository = root.resolve(strict=False) for encoded in sorted(path for path in untracked.split(b"\0") if path): relative = os.fsdecode(encoded) relative_path = Path(relative) if relative_path.is_absolute() or ".." in relative_path.parts: raise StateError(f"untracked candidate path escapes the repository: {relative}") path = repository / relative_path digest.update(b"\0untracked\0") digest.update(encoded) if path.is_symlink(): digest.update(b"\0symlink\0") digest.update(os.fsencode(os.readlink(path))) elif path.is_file(): digest.update(b"\0file\0") digest.update(str(path.stat().st_mode & 0o777).encode("ascii")) with path.open("rb") as handle: while chunk := handle.read(1024 * 1024): digest.update(chunk) else: digest.update(b"\0special\0") if git_bytes(root, *status_arguments) != status_before: raise StateError("worktree changed while computing the candidate digest; nominate again") return digest.hexdigest() def dirty_tree_digest(root: Path) -> str: """Hash an exact stable base commit, tracked diff, and untracked source set.""" first = _dirty_tree_digest_once(root) second = _dirty_tree_digest_once(root) if first != second: raise StateError("worktree changed while computing the candidate digest; nominate again") return first def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="operation", required=True) init = subparsers.add_parser("init", help="create a new compact session state") init.add_argument("--goal", required=True) init.add_argument("--accept", action="append", required=True, help="acceptance criterion; repeat as needed") init.add_argument("--slice", dest="active_slice", required=True) init.add_argument("--next", dest="next_action", required=True) init.add_argument("--non-goals", default="None.") init.add_argument("--issues", default="None.") init.add_argument("--milestone", default="None.") init.add_argument("--active-processes", default="None recorded.") init.add_argument("--force", action="store_true", help="replace an existing state") show = subparsers.add_parser("show", help="print the current state") show.add_argument("--path", action="store_true", help="print only the state path") subparsers.add_parser("resume", help="refresh repository identity and show the exact next action") update = subparsers.add_parser("set", help="update active task fields") update.add_argument("--goal") update.add_argument("--state", choices=VALID_STATES) update.add_argument("--issues") update.add_argument("--milestone") update.add_argument("--slice", dest="active_slice") update.add_argument("--non-goals") update.add_argument("--next", dest="next_action") update.add_argument("--active-processes") update.add_argument("--accept", action="append", help="replace acceptance criteria; repeat as needed") update.add_argument("--decision", action="append", help="append a concise decision") delta = subparsers.add_parser("scope-delta", help="record authoritative user steering") delta.add_argument("--title", required=True) delta.add_argument("--source", required=True) delta.add_argument("--classification", required=True) delta.add_argument("--requirement", required=True) delta.add_argument("--evidence-valid", required=True) delta.add_argument("--invalidated-gates", required=True) delta.add_argument("--tracker-sync", choices=("yes", "no"), required=True) delta.add_argument("--next", dest="next_action") record_file = subparsers.add_parser("record-file", help="record an intentionally changed file") record_file.add_argument("path") record_file.add_argument("--reason", required=True) record_gate = subparsers.add_parser("record-gate", help="record one compact verification result") record_gate.add_argument("--gate", required=True) record_gate.add_argument("--status", type=str.upper, choices=("PASS", "FAIL"), required=True) record_gate.add_argument("--digest", required=True) record_gate.add_argument("--command", required=True) record_gate.add_argument("--date", default=None) record_gate.add_argument("--next", dest="next_action") nominate = subparsers.add_parser("nominate", help="nominate an exact candidate commit or tree digest") candidate = nominate.add_mutually_exclusive_group() candidate.add_argument("--commit", default="HEAD") candidate.add_argument( "--dirty-tree", action="store_true", help="explicitly nominate the current dirty tree by a reproducible digest", ) nominate.add_argument("--next", dest="next_action") return parser def perform_operation(root: Path, workflow: dict[str, Any], args: argparse.Namespace) -> int: path = state_path(root, workflow) if args.operation == "init": if path.exists() and not args.force: raise StateError(f"session state already exists at {path}; use init --force to replace it") state = SessionState( repository=repository_snapshot(root, args.active_processes), goal=clean_text(args.goal), issues=clean_text(args.issues), milestone=clean_text(args.milestone), active_slice=clean_text(args.active_slice), non_goals=clean_text(args.non_goals), acceptance=[AcceptanceCriterion(clean_text(value)) for value in args.accept], next_action=clean_text(args.next_action), ) write_state(root, workflow, state) print(f"PASS state-init — {path.relative_to(root.resolve(strict=False))}") print(f"NEXT {state.next_action}") return 0 if args.operation == "show": # Loading the file validates that it remains template-compatible. read_state(root, workflow) if args.path: print(path) else: print(path.read_text(encoding="utf-8"), end="") return 0 state = read_state(root, workflow) repository_changes = refresh_repository(root, state) if args.operation == "resume": write_state(root, workflow, state) print( f"RESUME {state.current_state} — {state.repository.branch}@{state.repository.head[:12]} — " f"{state.goal}" ) for change in repository_changes: print(f"REPOSITORY CHANGED {change}") print(f"NEXT {state.next_action}") return 0 if args.operation == "set": changed = False for argument, attribute in ( ("goal", "goal"), ("state", "current_state"), ("issues", "issues"), ("milestone", "milestone"), ("active_slice", "active_slice"), ("non_goals", "non_goals"), ("next_action", "next_action"), ): value = getattr(args, argument) if value is not None: setattr(state, attribute, clean_text(value)) changed = True if args.active_processes is not None: state.repository.active_processes = clean_text(args.active_processes, "None recorded.") changed = True if args.accept is not None: state.acceptance = [AcceptanceCriterion(clean_text(value)) for value in args.accept] changed = True if args.decision: state.decisions.extend(clean_text(value) for value in args.decision) changed = True if not changed: raise StateError("state.py set requires at least one field to update") elif args.operation == "scope-delta": classification_key = clean_text(args.classification).lower() classification = CLASSIFICATIONS.get(classification_key) if classification is None: allowed = ", ".join(sorted(key for key in CLASSIFICATIONS if "-" in key)) raise StateError(f"unknown scope-delta classification; use one of: {allowed}") state.deltas.append( ScopeDelta( title=clean_text(args.title), source=clean_text(args.source), classification=classification, requirement=clean_text(args.requirement), evidence_valid=clean_text(args.evidence_valid), invalidated_gates=clean_text(args.invalidated_gates), tracker_sync=args.tracker_sync, ) ) if classification == "added acceptance criterion": state.acceptance.append(AcceptanceCriterion(clean_text(args.requirement))) if args.next_action: state.next_action = clean_text(args.next_action) elif args.operation == "record-file": relative = normalize_repo_file(root, args.path) replace_or_append_file(state, relative, args.reason) elif args.operation == "record-gate": replace_or_append_gate( state, GateRecord( gate=clean_text(args.gate), status=args.status, digest=clean_text(args.digest), command=clean_text(args.command), date=clean_text(args.date or utc_now()), ), ) if args.next_action: state.next_action = clean_text(args.next_action) elif args.operation == "nominate": if args.dirty_tree: digest = dirty_tree_digest(root) candidate_label = f"dirty-tree sha256:{digest} based on {state.repository.head}" else: if state.repository.worktree != "clean": raise StateError( "candidate nomination requires a clean worktree; use --dirty-tree to nominate an exact dirty-tree digest" ) commit = candidate_commit(root, args.commit) candidate_label = f"commit {commit}" state.current_state = "Candidate-ready" state.decisions = [ item for item in state.decisions if not item.startswith("Candidate nominated:") ] state.decisions.append(f"Candidate nominated: {candidate_label} at {utc_now()}.") state.next_action = clean_text( args.next_action, f"Run candidate verification for {candidate_label}.", ) else: # pragma: no cover - argparse prevents this raise StateError(f"unknown operation: {args.operation}") write_state(root, workflow, state) relative = path.resolve(strict=False).relative_to(root.resolve(strict=False)) print(f"PASS state-{args.operation} — {relative}") print(f"NEXT {state.next_action}") return 0 def _git_fixture(root: Path) -> None: (root / ".codex").mkdir(parents=True) (root / ".codex/workflow.toml").write_text( """version = 1 [session] state_file = ".codex/session/STATE.md" max_state_lines = 70 """, encoding="utf-8", ) (root / ".gitignore").write_text("/.codex/session/\n", encoding="utf-8") (root / "tracked.txt").write_text("fixture\n", encoding="utf-8") env = dict(os.environ) env.update( { "GIT_AUTHOR_NAME": "Blacksite self-test", "GIT_AUTHOR_EMAIL": "self-test@example.invalid", "GIT_COMMITTER_NAME": "Blacksite self-test", "GIT_COMMITTER_EMAIL": "self-test@example.invalid", } ) for command in ( ["git", "init", "--quiet", str(root)], ["git", "-C", str(root), "add", "."], ["git", "-C", str(root), "commit", "--quiet", "-m", "fixture"], ): subprocess.run(command, check=True, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE) def self_test() -> int: with tempfile.TemporaryDirectory(prefix="blacksite-state-self-test-") as temporary: root = Path(temporary) _git_fixture(root) found = find_repository(root / "nested") if found.resolve() != root.resolve(): raise AssertionError("repository locator did not find the temporary workflow") workflow = load_workflow(found) parser = build_parser() operations = [ [ "init", "--goal", "Exercise resumable state", "--accept", "State stays concise", "--slice", "Workflow helpers", "--next", "Update the fixture state.", ], ["set", "--state", "Engineering-complete", "--next", "Record steering."], [ "scope-delta", "--title", "Self-test steering", "--source", "self-test", "--classification", "active-slice-refinement", "--requirement", "Keep the newest exact action", "--evidence-valid", "Earlier fixture setup remains valid", "--invalidated-gates", "None", "--tracker-sync", "no", "--next", "Record the intentional file.", ], [ "record-file", "scripts/codex/state.py", "--reason", "exercise file recording", ], [ "record-gate", "--gate", "state-fixture", "--status", "PASS", "--digest", "digest-fixture", "--command", "python state.py --self-test", "--next", "Resume the fixture.", ], ["resume"], ["nominate", "--next", "Run the candidate fixture gate."], ] for operation in operations: with contextlib.redirect_stdout(io.StringIO()): result = perform_operation(root, workflow, parser.parse_args(operation)) if result != 0: raise AssertionError(f"state operation failed: {operation[0]}") state = read_state(root, workflow) if state.current_state != "Candidate-ready": raise AssertionError("nominate did not enter Candidate-ready state") for index in range(12): state.gates.append( GateRecord( gate=f"gate-{index}", status="PASS", digest=f"digest-{index}", command=f"fixture {index}", date=utc_now(), ) ) state.next_action = "Nominate the fixture commit." path = write_state(root, workflow, state) lines = path.read_text(encoding="utf-8").splitlines() if len(lines) > max_state_lines(workflow): raise AssertionError("state exceeded configured line budget") resumed = read_state(root, workflow) if resumed.next_action != "Nominate the fixture commit.": raise AssertionError("state round-trip lost the exact next action") if resumed.repository.real_path != str(root.resolve()): raise AssertionError("state did not record the canonical repository path") commit = candidate_commit(root, "HEAD") if len(commit) != 40: raise AssertionError("candidate nomination did not resolve a commit") tracked = root / "tracked.txt" tracked.write_text("changed\n", encoding="utf-8") dirty_digest = dirty_tree_digest(root) if dirty_digest != dirty_tree_digest(root): raise AssertionError("dirty candidate digest is not deterministic") tracked.write_text("changed again\n", encoding="utf-8") if dirty_digest == dirty_tree_digest(root): raise AssertionError("dirty candidate digest ignored changed source bytes") tracked.write_text("fixture\n", encoding="utf-8") if worktree_summary(root) != "clean": raise AssertionError("ignored state unexpectedly dirtied the fixture repository") print("PASS state-self-test — temp repository lifecycle and line budget") return 0 def main(argv: Sequence[str] | None = None) -> int: arguments = list(sys.argv[1:] if argv is None else argv) if arguments in (["--self-test"], ["self-test"]): try: return self_test() except (AssertionError, OSError, StateError, subprocess.SubprocessError) as error: print(f"FAIL state-self-test — {error}", file=sys.stderr) return 1 parser = build_parser() args = parser.parse_args(arguments) try: root = find_repository() workflow = load_workflow(root) return perform_operation(root, workflow, args) except StateError as error: print(f"FAIL state-{args.operation} — {error}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())