"""Evidence-bound reproduction, triage, disclosure, and patch workflows.""" from __future__ import annotations import hashlib import hmac import json import os import platform import re import secrets import sys import time from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Callable, Mapping, Sequence from .artifacts import ArtifactStore from .contracts import ( ToolCall, ToolExecutionContext, ToolExecutionResult, ToolParameter, ToolSpec, ) from .repository import atomic_patch_text, atomic_write_text from .sandbox import control_root, ensure_control_root from .security import SecretRedactor from .transactions import TransactionStore RunCommand = Callable[ [str, str, float], tuple[bool, str, str, int | None] ] _ID_RE = re.compile(r"(?:rep|tri|dis|patch|evb)_[a-f0-9]{32}") _SHA256_RE = re.compile(r"[a-f0-9]{64}") _SEVERITIES = ("informational", "low", "moderate", "high", "critical") _AUDIENCES = ("maintainer", "operator", "coordinated", "public") ENGINEERING_TOOL_SPECS: tuple[ToolSpec, ...] = ( ToolSpec( "ReproductionRun", "engineering", "Run one contained reproduction attempt and persist exact environment, state, output, and exit evidence.", "ReproductionRun(command='python -m pytest tests/test_api.py', snapshot_paths=['src', 'tests'])", ( ToolParameter("command", "string", "Exact contained command to execute."), ToolParameter( "working_directory", "string", "Optional workspace-relative working directory.", required=False, ), ToolParameter( "snapshot_paths", "array", "Optional workspace paths whose content state is captured before and after execution.", required=False, ), ), risk="workspace_write", parallel_safe=False, idempotent=False, task_support="optional", ), ToolSpec( "ReproductionCompare", "engineering", "Compare any number of session reproduction receipts for stable outcomes and state drift.", "ReproductionCompare(reproduction_ids=['rep_...', 'rep_...'])", ( ToolParameter( "reproduction_ids", "array", "Reproduction receipt identifiers." ), ), ), ToolSpec( "TriageCreate", "engineering", "Create evidence-linked triage that keeps observed facts separate from hypotheses and next actions.", "TriageCreate(reproduction_ids=['rep_...'], observed_facts=['exit changed'], hypotheses=['configuration drift'], severity='moderate', confidence=0.7, next_actions=['inspect configuration'])", ( ToolParameter( "reproduction_ids", "array", "Cited reproduction receipt identifiers." ), ToolParameter("observed_facts", "array", "Evidence-supported facts."), ToolParameter("hypotheses", "array", "Unconfirmed explanations."), ToolParameter( "severity", "string", "Current impact classification.", enum=_SEVERITIES, ), ToolParameter("confidence", "number", "Confidence from zero to one."), ToolParameter("next_actions", "array", "Model-selected investigation actions."), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "TriageStatus", "engineering", "Read one session-scoped triage receipt with its bound reproduction evidence and drift state.", "TriageStatus(triage_id='tri_...')", (ToolParameter("triage_id", "string", "Triage receipt identifier."),), ), ToolSpec( "TriageUpdate", "engineering", "Append new evidence, facts, hypotheses, and next actions to an existing triage receipt.", "TriageUpdate(triage_id='tri_...', reproduction_ids=['rep_...'], observed_facts=['new fact'])", ( ToolParameter("triage_id", "string", "Triage receipt identifier."), ToolParameter( "reproduction_ids", "array", "Additional reproduction receipt identifiers.", required=False, ), ToolParameter( "observed_facts", "array", "Additional evidence-supported facts.", required=False, ), ToolParameter( "hypotheses", "array", "Additional unconfirmed explanations.", required=False, ), ToolParameter( "severity", "string", "Updated impact classification.", required=False, enum=_SEVERITIES, ), ToolParameter( "confidence", "number", "Updated confidence from zero to one.", required=False, ), ToolParameter( "next_actions", "array", "Additional model-selected investigation actions.", required=False, ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "DisclosureCreate", "engineering", "Create a sanitized disclosure artifact from cited triage and optional verified patch evidence.", "DisclosureCreate(triage_id='tri_...', title='Issue title', summary='Summary', impact='Impact', remediation='Resolution')", ( ToolParameter("triage_id", "string", "Cited triage receipt identifier."), ToolParameter("title", "string", "Disclosure title."), ToolParameter("summary", "string", "Evidence-grounded summary."), ToolParameter("impact", "string", "Observed or bounded impact."), ToolParameter("remediation", "string", "Repair and verification guidance."), ToolParameter( "audience", "string", "Intended disclosure audience.", required=False, enum=_AUDIENCES, ), ToolParameter( "patch_id", "string", "Optional patch receipt to include by digest and verification state.", required=False, ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "DisclosureStatus", "engineering", "Read one session-scoped disclosure receipt and its sanitized artifact reference.", "DisclosureStatus(disclosure_id='dis_...')", (ToolParameter("disclosure_id", "string", "Disclosure receipt identifier."),), ), ToolSpec( "EvidenceBundleCreate", "engineering", "Create a sanitized handoff artifact from cited engineering receipts without raw private output.", "EvidenceBundleCreate(title='Repair handoff', triage_ids=['tri_...'], patch_ids=['patch_...'])", ( ToolParameter( "title", "string", "Bundle title shown in the sanitized handoff artifact.", ), ToolParameter( "reproduction_ids", "array", "Optional reproduction receipt identifiers.", required=False, ), ToolParameter( "triage_ids", "array", "Optional triage receipt identifiers.", required=False, ), ToolParameter( "disclosure_ids", "array", "Optional disclosure receipt identifiers.", required=False, ), ToolParameter( "patch_ids", "array", "Optional patch receipt identifiers.", required=False, ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "EvidenceBundleStatus", "engineering", "Read one session-scoped sanitized evidence bundle receipt.", "EvidenceBundleStatus(bundle_id='evb_...')", (ToolParameter("bundle_id", "string", "Evidence bundle receipt identifier."),), ), ToolSpec( "PatchBegin", "engineering", "Begin an evidence-linked multi-file patch transaction with immutable original snapshots.", "PatchBegin(triage_id='tri_...', paths=['src/app.py', 'tests/test_app.py'])", ( ToolParameter("triage_id", "string", "Cited triage receipt identifier."), ToolParameter("paths", "array", "Exact workspace files covered by the patch."), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "PatchApply", "engineering", "Apply an atomic compare-and-swap patch set and restore every covered file if any change fails.", "PatchApply(patch_id='patch_...', changes=[{'operation':'replace','path':'src/app.py','expected_sha256':'...','old_text':'before','new_text':'after'}])", ( ToolParameter("patch_id", "string", "Patch receipt identifier."), ToolParameter( "changes", "array", "Changes using replace or write operations with exact current digests.", ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "PatchVerify", "engineering", "Run contained model-selected verification and bind its reproduction receipt to the patch.", "PatchVerify(patch_id='patch_...', command='python -m pytest tests/test_app.py')", ( ToolParameter("patch_id", "string", "Patch receipt identifier."), ToolParameter("command", "string", "Exact verification command."), ToolParameter( "working_directory", "string", "Optional workspace-relative working directory.", required=False, ), ), risk="workspace_write", parallel_safe=False, idempotent=False, task_support="optional", ), ToolSpec( "PatchCommit", "engineering", "Commit only a successfully verified patch whose covered files still match the verified state.", "PatchCommit(patch_id='patch_...')", (ToolParameter("patch_id", "string", "Patch receipt identifier."),), risk="workspace_write", parallel_safe=False, ), ToolSpec( "PatchRollback", "engineering", "Restore the original patch snapshots only when every covered file still matches current receipts, then bind a restoration digest proof.", "PatchRollback(patch_id='patch_...')", (ToolParameter("patch_id", "string", "Patch receipt identifier."),), risk="destructive", parallel_safe=False, idempotent=True, ), ToolSpec( "PatchStatus", "engineering", "Read patch mutations, verification receipts, transaction state, and rollback availability.", "PatchStatus(patch_id='patch_...')", (ToolParameter("patch_id", "string", "Patch receipt identifier."),), ), ) ENGINEERING_TOOL_NAMES = frozenset(spec.name for spec in ENGINEERING_TOOL_SPECS) _ENGINEERING_TOOL_BY_NAME = {spec.name: spec for spec in ENGINEERING_TOOL_SPECS} @dataclass(frozen=True) class FileState: path: str sha256: str bytes: int @dataclass(frozen=True) class ReproductionRecord: reproduction_id: str session_sha256: str command_sha256: str working_directory: str environment_sha256: str before_state_sha256: str after_state_sha256: str before_files: tuple[FileState, ...] after_files: tuple[FileState, ...] stdout_artifact_id: str stderr_artifact_id: str stdout_sha256: str stderr_sha256: str exit_code: int | None command_ok: bool status: str created_unix_ms: int finished_unix_ms: int def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(frozen=True) class TriageRecord: triage_id: str session_sha256: str reproduction_ids: tuple[str, ...] observed_facts: tuple[str, ...] hypotheses: tuple[str, ...] severity: str confidence: float next_actions: tuple[str, ...] evidence_sha256: str created_unix_ms: int def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(frozen=True) class DisclosureRecord: disclosure_id: str session_sha256: str triage_id: str patch_id: str audience: str evidence_sha256: str artifact_id: str artifact_sha256: str created_unix_ms: int def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(frozen=True) class EvidenceBundleRecord: bundle_id: str session_sha256: str title: str reproduction_ids: tuple[str, ...] triage_ids: tuple[str, ...] disclosure_ids: tuple[str, ...] patch_ids: tuple[str, ...] evidence_sha256: str artifact_id: str artifact_sha256: str created_unix_ms: int def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(frozen=True) class PatchMutation: operation: str path: str before_sha256: str after_sha256: str bytes: int created: bool @dataclass(frozen=True) class PatchRecord: patch_id: str session_sha256: str triage_id: str transaction_id: str paths: tuple[str, ...] status: str mutations: tuple[PatchMutation, ...] verification_ids: tuple[str, ...] verified_state_sha256: str last_error: str created_unix_ms: int updated_unix_ms: int restoration_sha256: str = "" def to_dict(self) -> dict[str, Any]: return asdict(self) def _atomic_json(path: Path, payload: Mapping[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path.parent, 0o700) temporary = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") temporary.write_text( json.dumps(payload, sort_keys=True, separators=(",", ":")), encoding="utf-8", ) temporary.chmod(0o600) with temporary.open("rb") as handle: os.fsync(handle.fileno()) os.replace(temporary, path) directory_fd = os.open(path.parent, os.O_RDONLY) try: os.fsync(directory_fd) finally: os.close(directory_fd) def _session_sha256(session_id: str) -> str: return hashlib.sha256((session_id or "direct").encode("utf-8")).hexdigest() def _require_session(session_sha256: str, session_id: str) -> None: if not hmac.compare_digest(session_sha256, _session_sha256(session_id)): raise PermissionError("engineering receipt does not belong to this session") def _record_path(root: Path, identifier: str, prefix: str) -> Path: if not _ID_RE.fullmatch(identifier) or not identifier.startswith(prefix + "_"): raise ValueError(f"{prefix} receipt id is invalid") return root / f"{identifier}.json" def _append_unique(existing: tuple[str, ...], additions: tuple[str, ...]) -> tuple[str, ...]: rows = list(existing) seen = set(rows) for item in additions: if item not in seen: rows.append(item) seen.add(item) return tuple(rows) def _load_payload(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise RuntimeError("engineering receipt is invalid") return payload def _string_tuple(value: object, name: str, *, allow_empty: bool = True) -> tuple[str, ...]: if not isinstance(value, list) or any(not isinstance(item, str) for item in value): raise ValueError(f"{name} must be an array of strings") rendered = tuple(item.strip() for item in value) if any(not item for item in rendered) or (not allow_empty and not rendered): raise ValueError(f"{name} must contain non-empty strings") return rendered def _contained_path( workspace: Path, raw_path: str, *, require_directory: bool = False ) -> Path: raw = raw_path.strip() or "." candidate = (workspace / raw).resolve() try: relative = candidate.relative_to(workspace) except ValueError as exc: raise ValueError("engineering path leaves the workspace") from exc if relative.parts and relative.parts[0] == ".nexum": raise ValueError("workspace control paths require dedicated runtime tools") if require_directory and not candidate.is_dir(): raise ValueError("working directory does not exist") return candidate def _sha256_file(path: Path) -> tuple[str, int]: digest = hashlib.sha256() size = 0 with path.open("rb") as handle: while chunk := handle.read(1024 * 1024): digest.update(chunk) size += len(chunk) return digest.hexdigest(), size def _snapshot(workspace: Path, raw_paths: tuple[str, ...]) -> tuple[FileState, ...]: states: list[FileState] = [] seen: set[str] = set() for raw_path in raw_paths: target = _contained_path(workspace, raw_path) if not target.exists(): relative = target.relative_to(workspace).as_posix() if relative not in seen: states.append(FileState(relative, "missing", 0)) seen.add(relative) continue candidates = (target,) if target.is_file() else tuple(sorted(target.rglob("*"))) for candidate in candidates: if candidate.is_symlink(): raise ValueError("snapshot paths cannot contain symbolic links") if not candidate.is_file(): continue relative = candidate.resolve().relative_to(workspace).as_posix() if relative.startswith(".nexum/") or relative.startswith(".git/"): continue if "/.git/" in f"/{relative}/" or relative in seen: continue digest, size = _sha256_file(candidate) states.append(FileState(relative, digest, size)) seen.add(relative) return tuple(sorted(states, key=lambda state: state.path)) def _state_sha256(states: tuple[FileState, ...]) -> str: encoded = json.dumps( [asdict(state) for state in states], sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _sanitize(text: str, workspace: Path) -> str: return SecretRedactor().redact(text).replace(str(workspace), "[WORKSPACE]") class ReproductionStore: def __init__(self, workspace: str | Path) -> None: self.workspace = ensure_control_root(workspace) self.root = control_root(self.workspace) / "engineering" / "reproductions" self.root.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(self.root, 0o700) self.artifacts = ArtifactStore(self.workspace) def run( self, *, command: str, working_directory: str, snapshot_paths: tuple[str, ...], timeout_s: float, session_id: str, run_command: RunCommand, ) -> tuple[ReproductionRecord, str, str]: if not command.strip(): raise ValueError("reproduction command is required") workdir = _contained_path( self.workspace, working_directory, require_directory=True ) before = _snapshot(self.workspace, snapshot_paths) relative_workdir = workdir.relative_to(self.workspace).as_posix() or "." environment = { "architecture": platform.machine(), "os_name": platform.system(), "os_release": platform.release(), "python": ".".join(str(part) for part in sys.version_info[:3]), "working_directory": relative_workdir, } environment_sha256 = hashlib.sha256( json.dumps(environment, sort_keys=True, separators=(",", ":")).encode( "utf-8" ) ).hexdigest() started = int(time.time() * 1000) command_ok, stdout, stderr, exit_code = run_command( command, str(workdir), timeout_s ) finished = int(time.time() * 1000) safe_stdout = _sanitize(stdout, self.workspace) safe_stderr = _sanitize(stderr, self.workspace) after = _snapshot(self.workspace, snapshot_paths) stdout_artifact = self.artifacts.put_text( safe_stdout, source="reproduction_stdout", session_id=session_id or "direct", ) stderr_artifact = self.artifacts.put_text( safe_stderr, source="reproduction_stderr", session_id=session_id or "direct", ) record = ReproductionRecord( reproduction_id="rep_" + secrets.token_hex(16), session_sha256=_session_sha256(session_id), command_sha256=hashlib.sha256(command.encode("utf-8")).hexdigest(), working_directory=relative_workdir, environment_sha256=environment_sha256, before_state_sha256=_state_sha256(before), after_state_sha256=_state_sha256(after), before_files=before, after_files=after, stdout_artifact_id=stdout_artifact.artifact_id, stderr_artifact_id=stderr_artifact.artifact_id, stdout_sha256=hashlib.sha256(safe_stdout.encode("utf-8")).hexdigest(), stderr_sha256=hashlib.sha256(safe_stderr.encode("utf-8")).hexdigest(), exit_code=exit_code, command_ok=command_ok, status="exited" if exit_code is not None else "failed_to_start", created_unix_ms=started, finished_unix_ms=finished, ) _atomic_json( _record_path(self.root, record.reproduction_id, "rep"), record.to_dict(), ) return record, safe_stdout, safe_stderr def get(self, reproduction_id: str, *, session_id: str) -> ReproductionRecord: payload = _load_payload(_record_path(self.root, reproduction_id, "rep")) before = tuple(FileState(**row) for row in payload.pop("before_files", [])) after = tuple(FileState(**row) for row in payload.pop("after_files", [])) record = ReproductionRecord( **payload, before_files=before, after_files=after, ) _require_session(record.session_sha256, session_id) return record def compare( self, reproduction_ids: tuple[str, ...], *, session_id: str ) -> dict[str, Any]: if not reproduction_ids or len(reproduction_ids) != len(set(reproduction_ids)): raise ValueError("reproduction ids must be non-empty and unique") records = tuple( self.get(identifier, session_id=session_id) for identifier in reproduction_ids ) def same(name: str) -> bool: return len({getattr(record, name) for record in records}) == 1 return { "reproduction_ids": list(reproduction_ids), "same_command": same("command_sha256"), "same_environment": same("environment_sha256"), "same_exit_code": same("exit_code"), "same_stdout": same("stdout_sha256"), "same_stderr": same("stderr_sha256"), "same_before_state": same("before_state_sha256"), "same_after_state": same("after_state_sha256"), "stable_outcome": all( same(name) for name in ("exit_code", "stdout_sha256", "stderr_sha256") ), "state_drift_observed": not same("after_state_sha256"), } class TriageStore: def __init__(self, workspace: str | Path) -> None: self.workspace = ensure_control_root(workspace) self.root = control_root(self.workspace) / "engineering" / "triage" self.root.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(self.root, 0o700) self.reproductions = ReproductionStore(self.workspace) def create( self, *, reproduction_ids: tuple[str, ...], observed_facts: tuple[str, ...], hypotheses: tuple[str, ...], severity: str, confidence: float, next_actions: tuple[str, ...], session_id: str, ) -> TriageRecord: if not reproduction_ids or len(reproduction_ids) != len(set(reproduction_ids)): raise ValueError("triage requires unique reproduction evidence") if not observed_facts: raise ValueError("triage requires at least one observed fact") if severity not in _SEVERITIES: raise ValueError("triage severity is invalid") if confidence < 0.0 or confidence > 1.0: raise ValueError("triage confidence must be between zero and one") records = tuple( self.reproductions.get(identifier, session_id=session_id) for identifier in reproduction_ids ) evidence = [ { "id": record.reproduction_id, "command": record.command_sha256, "environment": record.environment_sha256, "exit_code": record.exit_code, "stdout": record.stdout_sha256, "stderr": record.stderr_sha256, "before": record.before_state_sha256, "after": record.after_state_sha256, } for record in records ] evidence_sha256 = hashlib.sha256( json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode( "utf-8" ) ).hexdigest() def sanitize(value: str) -> str: return _sanitize(value, self.workspace) record = TriageRecord( triage_id="tri_" + secrets.token_hex(16), session_sha256=_session_sha256(session_id), reproduction_ids=reproduction_ids, observed_facts=tuple(sanitize(item) for item in observed_facts), hypotheses=tuple(sanitize(item) for item in hypotheses), severity=severity, confidence=confidence, next_actions=tuple(sanitize(item) for item in next_actions), evidence_sha256=evidence_sha256, created_unix_ms=int(time.time() * 1000), ) _atomic_json(_record_path(self.root, record.triage_id, "tri"), record.to_dict()) return record def get(self, triage_id: str, *, session_id: str) -> TriageRecord: payload = _load_payload(_record_path(self.root, triage_id, "tri")) record = TriageRecord( **{ **payload, "reproduction_ids": tuple(payload.get("reproduction_ids", [])), "observed_facts": tuple(payload.get("observed_facts", [])), "hypotheses": tuple(payload.get("hypotheses", [])), "next_actions": tuple(payload.get("next_actions", [])), } ) _require_session(record.session_sha256, session_id) return record def status(self, triage_id: str, *, session_id: str) -> dict[str, Any]: record = self.get(triage_id, session_id=session_id) evidence: list[dict[str, Any]] = [] for identifier in record.reproduction_ids: reproduction = self.reproductions.get( identifier, session_id=session_id ) evidence.append( { "id": reproduction.reproduction_id, "exit_code": reproduction.exit_code, "command_ok": reproduction.command_ok, "status": reproduction.status, "state_drift_observed": reproduction.before_state_sha256 != reproduction.after_state_sha256, "stdout": reproduction.stdout_sha256, "stderr": reproduction.stderr_sha256, } ) return { **record.to_dict(), "evidence": evidence, "state_drift_observed": any( item["state_drift_observed"] for item in evidence ), } def update( self, triage_id: str, *, reproduction_ids: tuple[str, ...], observed_facts: tuple[str, ...], hypotheses: tuple[str, ...], severity: str, confidence: float, next_actions: tuple[str, ...], session_id: str, ) -> TriageRecord: record = self.get(triage_id, session_id=session_id) if severity not in _SEVERITIES: raise ValueError("triage severity is invalid") if confidence < 0.0 or confidence > 1.0: raise ValueError("triage confidence must be between zero and one") merged_reproductions = _append_unique(record.reproduction_ids, reproduction_ids) if not merged_reproductions: raise ValueError("triage requires reproduction evidence") records = tuple( self.reproductions.get(identifier, session_id=session_id) for identifier in merged_reproductions ) evidence = [ { "id": row.reproduction_id, "command": row.command_sha256, "environment": row.environment_sha256, "exit_code": row.exit_code, "stdout": row.stdout_sha256, "stderr": row.stderr_sha256, "before": row.before_state_sha256, "after": row.after_state_sha256, } for row in records ] updated = TriageRecord( triage_id=record.triage_id, session_sha256=record.session_sha256, reproduction_ids=merged_reproductions, observed_facts=_append_unique( record.observed_facts, tuple(_sanitize(item, self.workspace) for item in observed_facts), ), hypotheses=_append_unique( record.hypotheses, tuple(_sanitize(item, self.workspace) for item in hypotheses), ), severity=severity, confidence=confidence, next_actions=_append_unique( record.next_actions, tuple(_sanitize(item, self.workspace) for item in next_actions), ), evidence_sha256=hashlib.sha256( json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode( "utf-8" ) ).hexdigest(), created_unix_ms=record.created_unix_ms, ) _atomic_json(_record_path(self.root, updated.triage_id, "tri"), updated.to_dict()) return updated class PatchStore: def __init__(self, workspace: str | Path) -> None: self.workspace = ensure_control_root(workspace) self.root = control_root(self.workspace) / "engineering" / "patches" self.root.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(self.root, 0o700) self.triage = TriageStore(self.workspace) self.reproductions = ReproductionStore(self.workspace) self.transactions = TransactionStore(self.workspace) def _write(self, record: PatchRecord) -> None: _atomic_json(_record_path(self.root, record.patch_id, "patch"), record.to_dict()) def get(self, patch_id: str, *, session_id: str) -> PatchRecord: payload = _load_payload(_record_path(self.root, patch_id, "patch")) mutations = tuple(PatchMutation(**row) for row in payload.pop("mutations", [])) record = PatchRecord( **{ **payload, "paths": tuple(payload.get("paths", [])), "verification_ids": tuple(payload.get("verification_ids", [])), "mutations": mutations, } ) _require_session(record.session_sha256, session_id) return record def begin( self, *, triage_id: str, paths: tuple[str, ...], session_id: str ) -> PatchRecord: self.triage.get(triage_id, session_id=session_id) if not paths or len(paths) != len(set(paths)): raise ValueError("patch paths must be non-empty and unique") transaction = self.transactions.begin(paths, session_id=session_id) now = int(time.time() * 1000) record = PatchRecord( patch_id="patch_" + secrets.token_hex(16), session_sha256=_session_sha256(session_id), triage_id=triage_id, transaction_id=transaction.transaction_id, paths=tuple(entry.path for entry in transaction.entries), status="open", mutations=(), verification_ids=(), verified_state_sha256="", last_error="", created_unix_ms=now, updated_unix_ms=now, ) self._write(record) return record def _current_digests(self, paths: tuple[str, ...]) -> dict[str, str]: current: dict[str, str] = {} for raw_path in paths: target = _contained_path(self.workspace, raw_path) current[raw_path] = ( _sha256_file(target)[0] if target.is_file() else "missing" ) return current def _covered_state(self, paths: tuple[str, ...]) -> str: states = tuple( FileState(path, digest, 0) for path, digest in sorted(self._current_digests(paths).items()) ) return _state_sha256(states) def apply( self, patch_id: str, changes: Sequence[Mapping[str, object]], *, session_id: str, ) -> tuple[PatchRecord, str]: record = self.get(patch_id, session_id=session_id) if record.status not in {"open", "applied", "verification_failed"}: raise RuntimeError(f"patch is already {record.status}") if not changes: raise ValueError("patch changes must not be empty") changed_paths: set[str] = set() mutations: list[PatchMutation] = list(record.mutations) error = "" try: for change in changes: if not isinstance(change, Mapping): raise ValueError("each patch change must be an object") operation = str(change.get("operation") or "") path = str(change.get("path") or "") if path not in record.paths: raise ValueError("patch change path is not covered by the transaction") if path in changed_paths: raise ValueError("each path may appear once per patch application") changed_paths.add(path) expected = str(change.get("expected_sha256") or "") if expected and not _SHA256_RE.fullmatch(expected): raise ValueError("expected_sha256 is invalid") if operation == "replace": if not expected: raise ValueError("replace changes require expected_sha256") result = atomic_patch_text( self.workspace, path, expected_sha256=expected, old_text=str(change.get("old_text") or ""), new_text=str(change.get("new_text") or ""), ) elif operation == "write": result = atomic_write_text( self.workspace, path, str(change.get("content") or ""), expected_sha256=expected or None, ) else: raise ValueError("patch operation must be replace or write") mutations.append( PatchMutation( operation=operation, path=result.path, before_sha256=result.before_sha256 or "missing", after_sha256=result.after_sha256, bytes=result.bytes, created=result.created, ) ) except (OSError, RuntimeError, UnicodeError, ValueError) as exc: error = _sanitize(f"{type(exc).__name__}: {exc}", self.workspace) status = "rolled_back" try: self.transactions.rollback( record.transaction_id, self._current_digests(record.paths), session_id=session_id, ) except (OSError, RuntimeError, ValueError) as rollback_exc: status = "rollback_failed" error += "; rollback: " + _sanitize( f"{type(rollback_exc).__name__}: {rollback_exc}", self.workspace, ) failed = PatchRecord( **{ **record.to_dict(), "status": status, "mutations": tuple(mutations), "last_error": error, "updated_unix_ms": int(time.time() * 1000), } ) self._write(failed) return failed, error updated = PatchRecord( **{ **record.to_dict(), "status": "applied", "mutations": tuple(mutations), "verified_state_sha256": "", "last_error": "", "updated_unix_ms": int(time.time() * 1000), } ) self._write(updated) return updated, "" def verify( self, patch_id: str, *, command: str, working_directory: str, timeout_s: float, session_id: str, run_command: RunCommand, ) -> tuple[PatchRecord, ReproductionRecord, str, str]: record = self.get(patch_id, session_id=session_id) if record.status not in {"applied", "verification_failed"}: raise RuntimeError("patch must be applied before verification") reproduction, stdout, stderr = self.reproductions.run( command=command, working_directory=working_directory, snapshot_paths=record.paths, timeout_s=timeout_s, session_id=session_id, run_command=run_command, ) passed = reproduction.command_ok and reproduction.exit_code == 0 updated = PatchRecord( **{ **record.to_dict(), "status": "verified" if passed else "verification_failed", "verification_ids": (*record.verification_ids, reproduction.reproduction_id), "verified_state_sha256": self._covered_state(record.paths) if passed else "", "last_error": "" if passed else "verification command did not succeed", "updated_unix_ms": int(time.time() * 1000), } ) self._write(updated) return updated, reproduction, stdout, stderr def commit(self, patch_id: str, *, session_id: str) -> PatchRecord: record = self.get(patch_id, session_id=session_id) if record.status != "verified": raise RuntimeError("patch must have successful current verification") current_state = self._covered_state(record.paths) if not hmac.compare_digest(current_state, record.verified_state_sha256): raise RuntimeError("covered files changed after patch verification") self.transactions.commit(record.transaction_id, session_id=session_id) updated = PatchRecord( **{ **record.to_dict(), "status": "committed", "updated_unix_ms": int(time.time() * 1000), } ) self._write(updated) return updated def require_current_verification( self, patch_id: str, *, session_id: str ) -> PatchRecord: record = self.get(patch_id, session_id=session_id) if record.status not in {"verified", "committed"}: raise ValueError("public disclosure requires a verified or committed patch") current_state = self._covered_state(record.paths) if not record.verified_state_sha256 or not hmac.compare_digest( current_state, record.verified_state_sha256 ): raise RuntimeError("covered files changed after patch verification") return record def rollback(self, patch_id: str, *, session_id: str) -> PatchRecord: record = self.get(patch_id, session_id=session_id) if record.status == "rolled_back": return record if record.status == "rollback_failed": raise RuntimeError("automatic rollback failed; inspect current file state") self.transactions.rollback( record.transaction_id, self._current_digests(record.paths), session_id=session_id, ) transaction = self.transactions.get( record.transaction_id, session_id=session_id ) restored = self._current_digests(record.paths) for entry in transaction.entries: if not hmac.compare_digest(restored[entry.path], entry.sha256): raise RuntimeError("rollback restoration proof failed") restoration_sha256 = _state_sha256( tuple( FileState(path, digest, 0) for path, digest in sorted(restored.items()) ) ) updated = PatchRecord( **{ **record.to_dict(), "status": "rolled_back", "verified_state_sha256": "", "restoration_sha256": restoration_sha256, "updated_unix_ms": int(time.time() * 1000), } ) self._write(updated) return updated class DisclosureStore: def __init__(self, workspace: str | Path) -> None: self.workspace = ensure_control_root(workspace) self.root = control_root(self.workspace) / "engineering" / "disclosures" self.root.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(self.root, 0o700) self.triage = TriageStore(self.workspace) self.patches = PatchStore(self.workspace) self.artifacts = ArtifactStore(self.workspace) def create( self, *, triage_id: str, title: str, summary: str, impact: str, remediation: str, audience: str, patch_id: str, session_id: str, ) -> DisclosureRecord: triage = self.triage.get(triage_id, session_id=session_id) if audience not in _AUDIENCES: raise ValueError("disclosure audience is invalid") fields = tuple( _sanitize(value, self.workspace) for value in (title, summary, impact, remediation) ) if any(not value.strip() for value in fields): raise ValueError("disclosure fields must not be empty") patch = self.patches.get(patch_id, session_id=session_id) if patch_id else None if audience == "public": if patch is None: raise ValueError( "public disclosure requires a verified or committed patch" ) patch = self.patches.require_current_verification( patch.patch_id, session_id=session_id ) evidence = { "triage": triage.evidence_sha256, "reproductions": list(triage.reproduction_ids), "patch": { "id": patch.patch_id, "status": patch.status, "verification_ids": list(patch.verification_ids), "mutations": [ { "before": mutation.before_sha256, "after": mutation.after_sha256, } for mutation in patch.mutations ], } if patch is not None else None, } evidence_sha256 = hashlib.sha256( json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode( "utf-8" ) ).hexdigest() fact_lines = "\n".join(f"- {fact}" for fact in triage.observed_facts) hypothesis_lines = "\n".join( f"- {hypothesis}" for hypothesis in triage.hypotheses ) or "- None recorded" patch_section = "" if patch is not None: patch_section = ( "\n## Repair Evidence\n" f"- Status: {patch.status}\n" f"- Changed files: {len({mutation.path for mutation in patch.mutations})}\n" f"- Verification receipts: {', '.join(patch.verification_ids) or 'none'}\n" ) document = ( f"# {fields[0]}\n\n" f"Audience: {audience}\n\n" "## Summary\n" f"{fields[1]}\n\n" "## Impact\n" f"{fields[2]}\n\n" "## Observed Facts\n" f"{fact_lines}\n\n" "## Hypotheses\n" f"{hypothesis_lines}\n\n" "## Evidence\n" f"- Evidence digest: {evidence_sha256}\n" f"- Reproduction receipts: {', '.join(triage.reproduction_ids)}\n" f"- Severity: {triage.severity}\n" f"- Confidence: {triage.confidence:.3f}\n" f"{patch_section}\n" "## Remediation\n" f"{fields[3]}\n\n" "Raw command output, credentials, absolute workspace paths, private session identifiers, and internal control state are omitted.\n" ) artifact = self.artifacts.put_text( document, media_type="text/markdown; charset=utf-8", source="sanitized_disclosure", session_id=session_id or "direct", ) record = DisclosureRecord( disclosure_id="dis_" + secrets.token_hex(16), session_sha256=_session_sha256(session_id), triage_id=triage_id, patch_id=patch_id, audience=audience, evidence_sha256=evidence_sha256, artifact_id=artifact.artifact_id, artifact_sha256=artifact.sha256, created_unix_ms=int(time.time() * 1000), ) _atomic_json( _record_path(self.root, record.disclosure_id, "dis"), record.to_dict() ) return record def get(self, disclosure_id: str, *, session_id: str) -> DisclosureRecord: payload = _load_payload(_record_path(self.root, disclosure_id, "dis")) record = DisclosureRecord(**payload) _require_session(record.session_sha256, session_id) return record class EvidenceBundleStore: def __init__(self, workspace: str | Path) -> None: self.workspace = ensure_control_root(workspace) self.root = control_root(self.workspace) / "engineering" / "bundles" self.root.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(self.root, 0o700) self.reproductions = ReproductionStore(self.workspace) self.triage = TriageStore(self.workspace) self.disclosures = DisclosureStore(self.workspace) self.patches = PatchStore(self.workspace) self.artifacts = ArtifactStore(self.workspace) def create( self, *, title: str, reproduction_ids: tuple[str, ...], triage_ids: tuple[str, ...], disclosure_ids: tuple[str, ...], patch_ids: tuple[str, ...], session_id: str, ) -> EvidenceBundleRecord: safe_title = _sanitize(title, self.workspace).strip() if not safe_title: raise ValueError("evidence bundle title is required") if not any((reproduction_ids, triage_ids, disclosure_ids, patch_ids)): raise ValueError("evidence bundle requires at least one receipt id") if any( len(rows) != len(set(rows)) for rows in (reproduction_ids, triage_ids, disclosure_ids, patch_ids) ): raise ValueError("evidence bundle receipt ids must be unique per type") reproductions = tuple( self.reproductions.get(identifier, session_id=session_id) for identifier in reproduction_ids ) triages = tuple( self.triage.status(identifier, session_id=session_id) for identifier in triage_ids ) disclosures = tuple( self.disclosures.get(identifier, session_id=session_id) for identifier in disclosure_ids ) patches = tuple( self.patches.get(identifier, session_id=session_id) for identifier in patch_ids ) evidence = { "reproductions": [ { "id": record.reproduction_id, "command": record.command_sha256, "environment": record.environment_sha256, "exit_code": record.exit_code, "status": record.status, "stdout": record.stdout_sha256, "stderr": record.stderr_sha256, "before": record.before_state_sha256, "after": record.after_state_sha256, } for record in reproductions ], "triage": [ { "id": str(row["triage_id"]), "evidence": str(row["evidence_sha256"]), "severity": str(row["severity"]), "confidence": float(row["confidence"]), "state_drift_observed": bool(row["state_drift_observed"]), } for row in triages ], "disclosures": [ { "id": record.disclosure_id, "triage": record.triage_id, "patch": record.patch_id, "audience": record.audience, "artifact": record.artifact_id, "artifact_sha256": record.artifact_sha256, } for record in disclosures ], "patches": [ { "id": record.patch_id, "triage": record.triage_id, "status": record.status, "paths": len(record.paths), "mutations": len(record.mutations), "verification_ids": list(record.verification_ids), "verified_state": record.verified_state_sha256, "restoration": record.restoration_sha256, } for record in patches ], } evidence_sha256 = hashlib.sha256( json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode( "utf-8" ) ).hexdigest() document = ( f"# {safe_title}\n\n" f"Evidence digest: {evidence_sha256}\n\n" "## Reproductions\n" + "\n".join( ( f"- {record.reproduction_id}: status={record.status}, " f"exit={record.exit_code}, stdout={record.stdout_sha256}, " f"stderr={record.stderr_sha256}" ) for record in reproductions ) + ("\n" if reproductions else "- None\n") + "\n## Triage\n" + "\n".join( ( f"- {row['triage_id']}: severity={row['severity']}, " f"confidence={float(row['confidence']):.3f}, " f"evidence={row['evidence_sha256']}" ) for row in triages ) + ("\n" if triages else "- None\n") + "\n## Disclosures\n" + "\n".join( ( f"- {record.disclosure_id}: audience={record.audience}, " f"artifact={record.artifact_id}, sha256={record.artifact_sha256}" ) for record in disclosures ) + ("\n" if disclosures else "- None\n") + "\n## Patches\n" + "\n".join( ( f"- {record.patch_id}: status={record.status}, " f"mutations={len(record.mutations)}, " f"verification={','.join(record.verification_ids) or 'none'}" ) for record in patches ) + ("\n" if patches else "- None\n") + "\nRaw command output, credentials, absolute workspace paths, private session identifiers, and internal control state are omitted.\n" ) artifact = self.artifacts.put_text( document, media_type="text/markdown; charset=utf-8", source="sanitized_evidence_bundle", session_id=session_id or "direct", ) record = EvidenceBundleRecord( bundle_id="evb_" + secrets.token_hex(16), session_sha256=_session_sha256(session_id), title=safe_title, reproduction_ids=reproduction_ids, triage_ids=triage_ids, disclosure_ids=disclosure_ids, patch_ids=patch_ids, evidence_sha256=evidence_sha256, artifact_id=artifact.artifact_id, artifact_sha256=artifact.sha256, created_unix_ms=int(time.time() * 1000), ) _atomic_json(_record_path(self.root, record.bundle_id, "evb"), record.to_dict()) return record def get(self, bundle_id: str, *, session_id: str) -> EvidenceBundleRecord: payload = _load_payload(_record_path(self.root, bundle_id, "evb")) record = EvidenceBundleRecord( **{ **payload, "reproduction_ids": tuple(payload.get("reproduction_ids", [])), "triage_ids": tuple(payload.get("triage_ids", [])), "disclosure_ids": tuple(payload.get("disclosure_ids", [])), "patch_ids": tuple(payload.get("patch_ids", [])), } ) _require_session(record.session_sha256, session_id) return record def _tool_result( call: ToolCall, *, started: float, ok: bool, output: str = "", error: str = "", exit_code: int | None = None, ) -> ToolExecutionResult: rendered = output or error return ToolExecutionResult( name=call.name, args=call.args, ok=ok, tool_call_id=call.call_id, output=output, error=error, exit_code=exit_code, executed=True, elapsed_s=round(time.perf_counter() - started, 4), source_trust="trusted_execution", output_sha256=hashlib.sha256(rendered.encode("utf-8")).hexdigest(), ) def execute_engineering_tool( call: ToolCall, context: ToolExecutionContext, *, run_command: RunCommand, ) -> ToolExecutionResult: """Execute one evidence-bound engineering tool at the runtime boundary.""" started = time.perf_counter() spec = _ENGINEERING_TOOL_BY_NAME.get(call.name) if spec is None: return _tool_result( call, started=started, ok=False, error=f"unsupported engineering tool: {call.name}", ) try: spec.validate_arguments(call.args) workspace = context.workspace session_id = context.session_id if call.name == "ReproductionRun": reproduction_record, stdout, stderr = ReproductionStore(workspace).run( command=str(call.args["command"]), working_directory=str(call.args.get("working_directory") or "."), snapshot_paths=_string_tuple( call.args.get("snapshot_paths", []), "snapshot_paths" ), timeout_s=context.timeout_s, session_id=session_id, run_command=run_command, ) output = json.dumps( { "record": reproduction_record.to_dict(), "stdout": stdout, "stderr": stderr, }, sort_keys=True, ) return _tool_result( call, started=started, ok=reproduction_record.exit_code is not None, output=output, exit_code=reproduction_record.exit_code, ) if call.name == "ReproductionCompare": comparison = ReproductionStore(workspace).compare( _string_tuple( call.args["reproduction_ids"], "reproduction_ids", allow_empty=False, ), session_id=session_id, ) return _tool_result( call, started=started, ok=True, output=json.dumps(comparison, sort_keys=True), ) if call.name == "TriageCreate": triage_record = TriageStore(workspace).create( reproduction_ids=_string_tuple( call.args["reproduction_ids"], "reproduction_ids", allow_empty=False, ), observed_facts=_string_tuple( call.args["observed_facts"], "observed_facts", allow_empty=False, ), hypotheses=_string_tuple(call.args["hypotheses"], "hypotheses"), severity=str(call.args["severity"]), confidence=float(call.args["confidence"]), next_actions=_string_tuple( call.args["next_actions"], "next_actions" ), session_id=session_id, ) return _tool_result( call, started=started, ok=True, output=json.dumps(triage_record.to_dict(), sort_keys=True), ) if call.name == "TriageStatus": triage_status = TriageStore(workspace).status( str(call.args["triage_id"]), session_id=session_id ) return _tool_result( call, started=started, ok=True, output=json.dumps(triage_status, sort_keys=True), ) if call.name == "TriageUpdate": raw_confidence = call.args.get("confidence") current = TriageStore(workspace).get( str(call.args["triage_id"]), session_id=session_id ) triage_record = TriageStore(workspace).update( str(call.args["triage_id"]), reproduction_ids=_string_tuple( call.args.get("reproduction_ids", []), "reproduction_ids", ), observed_facts=_string_tuple( call.args.get("observed_facts", []), "observed_facts" ), hypotheses=_string_tuple(call.args.get("hypotheses", []), "hypotheses"), severity=str(call.args.get("severity") or current.severity), confidence=float(raw_confidence) if raw_confidence is not None else current.confidence, next_actions=_string_tuple( call.args.get("next_actions", []), "next_actions" ), session_id=session_id, ) return _tool_result( call, started=started, ok=True, output=json.dumps(triage_record.to_dict(), sort_keys=True), ) if call.name == "DisclosureCreate": disclosure_record = DisclosureStore(workspace).create( triage_id=str(call.args["triage_id"]), title=str(call.args["title"]), summary=str(call.args["summary"]), impact=str(call.args["impact"]), remediation=str(call.args["remediation"]), audience=str(call.args.get("audience") or "maintainer"), patch_id=str(call.args.get("patch_id") or ""), session_id=session_id, ) return _tool_result( call, started=started, ok=True, output=json.dumps(disclosure_record.to_dict(), sort_keys=True), ) if call.name == "DisclosureStatus": disclosure_record = DisclosureStore(workspace).get( str(call.args["disclosure_id"]), session_id=session_id ) return _tool_result( call, started=started, ok=True, output=json.dumps(disclosure_record.to_dict(), sort_keys=True), ) if call.name == "EvidenceBundleCreate": bundle_record = EvidenceBundleStore(workspace).create( title=str(call.args["title"]), reproduction_ids=_string_tuple( call.args.get("reproduction_ids", []), "reproduction_ids" ), triage_ids=_string_tuple(call.args.get("triage_ids", []), "triage_ids"), disclosure_ids=_string_tuple( call.args.get("disclosure_ids", []), "disclosure_ids" ), patch_ids=_string_tuple(call.args.get("patch_ids", []), "patch_ids"), session_id=session_id, ) return _tool_result( call, started=started, ok=True, output=json.dumps(bundle_record.to_dict(), sort_keys=True), ) if call.name == "EvidenceBundleStatus": bundle_record = EvidenceBundleStore(workspace).get( str(call.args["bundle_id"]), session_id=session_id ) return _tool_result( call, started=started, ok=True, output=json.dumps(bundle_record.to_dict(), sort_keys=True), ) patches = PatchStore(workspace) if call.name == "PatchBegin": patch_record = patches.begin( triage_id=str(call.args["triage_id"]), paths=_string_tuple( call.args["paths"], "paths", allow_empty=False ), session_id=session_id, ) elif call.name == "PatchApply": raw_changes = call.args["changes"] if not isinstance(raw_changes, list): raise ValueError("changes must be an array") patch_record, error = patches.apply( str(call.args["patch_id"]), tuple(raw_changes), session_id=session_id, ) return _tool_result( call, started=started, ok=not error, output=json.dumps(patch_record.to_dict(), sort_keys=True), error=error, ) elif call.name == "PatchVerify": patch_record, verification_record, stdout, stderr = patches.verify( str(call.args["patch_id"]), command=str(call.args["command"]), working_directory=str(call.args.get("working_directory") or "."), timeout_s=context.timeout_s, session_id=session_id, run_command=run_command, ) output = json.dumps( { "patch": patch_record.to_dict(), "reproduction": verification_record.to_dict(), "stdout": stdout, "stderr": stderr, }, sort_keys=True, ) return _tool_result( call, started=started, ok=patch_record.status == "verified", output=output, error=patch_record.last_error, exit_code=verification_record.exit_code, ) elif call.name == "PatchCommit": patch_record = patches.commit( str(call.args["patch_id"]), session_id=session_id ) elif call.name == "PatchRollback": patch_record = patches.rollback( str(call.args["patch_id"]), session_id=session_id ) elif call.name == "PatchStatus": patch_record = patches.get( str(call.args["patch_id"]), session_id=session_id ) else: raise RuntimeError("engineering tool dispatch is incomplete") return _tool_result( call, started=started, ok=True, output=json.dumps(patch_record.to_dict(), sort_keys=True), ) except (OSError, PermissionError, RuntimeError, TypeError, ValueError) as exc: return _tool_result( call, started=started, ok=False, error=f"{type(exc).__name__}: {exc}", ) __all__ = [ "DisclosureRecord", "DisclosureStore", "ENGINEERING_TOOL_NAMES", "ENGINEERING_TOOL_SPECS", "EvidenceBundleRecord", "EvidenceBundleStore", "FileState", "PatchMutation", "PatchRecord", "PatchStore", "ReproductionRecord", "ReproductionStore", "TriageRecord", "TriageStore", "execute_engineering_tool", ]