| """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", |
| ] |
|
|
