996 lines
38 KiB
Python
Executable File
996 lines
38 KiB
Python
Executable File
#!/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())
|