diff --git a/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md b/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md index 6ddc8068..d1904516 100644 --- a/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md +++ b/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md @@ -27,7 +27,7 @@ - Create: `tests/test_issue_62_live_evidence_sidecar.py` **Interfaces:** -- Produces: `SidecarConfig`, `validate_config(config)`, `write_capture_record(output_dir, hop, record)`, and `main(argv=None) -> int`. +- Produces: `SidecarConfig`, `validate_config(config)`, `write_capture_record(output_dir, hop, record, capture_key=..., correlation_token=...)`, and `main(argv=None) -> int`. - Consumes: only Python standard-library modules and a pre-existing HMAC key file. - [x] **Step 1: Write failing activation and artifact tests** @@ -43,7 +43,13 @@ def test_config_rejects_non_loopback(host, base_config): sidecar.validate_config(dataclasses.replace(base_config, listen_host=host)) def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path): - record_path = sidecar.write_capture_record(tmp_path, "pre", SAFE_RECORD) + record_path = sidecar.write_capture_record( + tmp_path, + "pre", + SAFE_RECORD, + capture_key=KEY, + correlation_token=CORRELATION_TOKEN, + ) assert json.loads(record_path.read_text(encoding="utf-8"))["schema"] == ( "codexhub.issue62.live-evidence-lane.v1" ) diff --git a/scripts/capture_issue_62_live_evidence.py b/scripts/capture_issue_62_live_evidence.py index 7e52091f..37055380 100644 --- a/scripts/capture_issue_62_live_evidence.py +++ b/scripts/capture_issue_62_live_evidence.py @@ -35,6 +35,7 @@ _FAILURES = frozenset( { "client_cancelled", + "correlation_binding_invalid", "downstream_cancelled", "forwarding_failed", "invalid_content_length", @@ -80,6 +81,9 @@ ) _CAPTURE_ID = re.compile(r"c[0-9a-f]{32}\Z") _DIGEST = re.compile(r"[0-9a-f]{64}\Z") +_CORRELATION_TOKEN = re.compile(r"[0-9a-f]{32}\Z") +_CORRELATION_BINDING = re.compile(r"([0-9a-f]{32})\.([0-9a-f]{64})\Z") +_CORRELATION_HEADER = "X-CodexHub-Issue62-Capture" class ConfigurationError(ValueError): @@ -391,10 +395,76 @@ def validate_capture_record(record: Mapping[str, Any], hop: str) -> None: raise ArtifactValidationError("record_completion_invalid") -def write_capture_record(output_dir: Path, hop: str, record: Mapping[str, Any]) -> Path: - """Atomically publish one sanitized record and always remove its partial.""" +def _capture_id_for_token(key: bytes, correlation_token: str) -> str: + if not isinstance(key, bytes) or not 32 <= len(key) <= 4096: + raise ValueError("capture_key_invalid") + if ( + not isinstance(correlation_token, str) + or _CORRELATION_TOKEN.fullmatch(correlation_token) is None + ): + raise ValueError("correlation_token_invalid") + digest = hmac.new( + key, + b"issue62-capture\0" + correlation_token.encode("ascii"), + hashlib.sha256, + ).hexdigest() + return "c" + digest[:32] + + +def _new_correlation_token() -> str: + return uuid.uuid4().hex + + +def _correlation_binding(key: bytes, correlation_token: str) -> str: + if ( + not isinstance(correlation_token, str) + or _CORRELATION_TOKEN.fullmatch(correlation_token) is None + ): + raise ValueError("correlation_token_invalid") + signature = hmac.new( + key, + b"issue62-correlation\0" + correlation_token.encode("ascii"), + hashlib.sha256, + ).hexdigest() + return f"{correlation_token}.{signature}" + + +def _correlation_token_from_header(key: bytes, value: str | None) -> str | None: + if not isinstance(value, str): + return None + match = _CORRELATION_BINDING.fullmatch(value) + if match is None: + return None + correlation_token, signature = match.groups() + expected = hmac.new( + key, + b"issue62-correlation\0" + correlation_token.encode("ascii"), + hashlib.sha256, + ).hexdigest() + if not hmac.compare_digest(signature, expected): + return None + return correlation_token + + +def write_capture_record( + output_dir: Path, + hop: str, + record: Mapping[str, Any], + *, + capture_key: bytes | None = None, + correlation_token: str | None = None, +) -> Path: + """Atomically publish one live-bound sanitized record and remove its partial.""" validate_capture_record(record, hop) + if capture_key is None or correlation_token is None: + raise ArtifactValidationError("capture_binding_missing") + try: + expected_capture_id = _capture_id_for_token(capture_key, correlation_token) + except (TypeError, ValueError): + raise ArtifactValidationError("capture_binding_invalid") from None + if not hmac.compare_digest(str(record["capture_id"]), expected_capture_id): + raise ArtifactValidationError("capture_binding_invalid") directory = Path(output_dir) directory.mkdir(parents=True, exist_ok=True) capture_id = str(record["capture_id"]) @@ -431,10 +501,6 @@ def write_capture_record(output_dir: Path, hop: str, record: Mapping[str, Any]) ) -def _capture_id() -> str: - return "c" + uuid.uuid4().hex - - def _content_type_class(value: str | None) -> str: if not value: return "absent" @@ -603,11 +669,18 @@ def wait(self) -> bool: return self._thread_failure is None def _write_control_failure(self, failure: str) -> None: + correlation_token = _new_correlation_token() try: write_capture_record( self._config.output_dir, self._config.hop, - _empty_record(self._config.hop, _capture_id(), failure), + _empty_record( + self._config.hop, + _capture_id_for_token(self._key, correlation_token), + failure, + ), + capture_key=self._key, + correlation_token=correlation_token, ) except Exception: return @@ -665,7 +738,8 @@ def _wait_for_active_connections(self, timeout: float) -> bool: return True def handle_post(self, handler: BaseHTTPRequestHandler) -> None: - capture_id = _capture_id() + correlation_token = _new_correlation_token() + capture_id = _capture_id_for_token(self._key, correlation_token) started_at = time.monotonic() request_fingerprint = BodyFingerprint(self._key, b"request-body") response_fingerprint: BodyFingerprint | None = None @@ -677,6 +751,17 @@ def handle_post(self, handler: BaseHTTPRequestHandler) -> None: request_complete = False response_complete = False try: + if self._config.hop == "post": + bound_token = _correlation_token_from_header( + self._key, + handler.headers.get(_CORRELATION_HEADER), + ) + if bound_token is None: + failure = "correlation_binding_invalid" + self._send_bounded_error(handler, 400) + return + correlation_token = bound_token + capture_id = _capture_id_for_token(self._key, correlation_token) raw_length = handler.headers.get("Content-Length") try: content_length = int(raw_length) if raw_length is not None else -1 @@ -716,8 +801,11 @@ def handle_post(self, handler: BaseHTTPRequestHandler) -> None: headers = { name: value for name, value in handler.headers.items() - if name.lower() not in _HOP_BY_HOP_HEADERS | {"host", "content-length"} + if name.lower() + not in _HOP_BY_HOP_HEADERS | {"host", "content-length", _CORRELATION_HEADER.lower()} } + if self._config.hop == "pre": + headers[_CORRELATION_HEADER] = _correlation_binding(self._key, correlation_token) target_path = _join_target_path(self._target.path, handler.path) upstream.request("POST", target_path, body=request_body, headers=headers) if upstream.sock is not None: @@ -810,7 +898,13 @@ def handle_post(self, handler: BaseHTTPRequestHandler) -> None: ), } try: - write_capture_record(self._config.output_dir, self._config.hop, record) + write_capture_record( + self._config.output_dir, + self._config.hop, + record, + capture_key=self._key, + correlation_token=correlation_token, + ) except Exception: return diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index 7b4a6856..88c9aa28 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -1,10 +1,12 @@ """Run the bounded, sanitized orchestration envelope for Issue #62. This module deliberately separates orchestration from the real upstream -controls. The default command-line mode is ``--fixture`` and therefore cannot -qualify Issue #62. A future live runner can reuse :class:`BoundedPhaseRunner` -with real command/phase callbacks after the exact candidate and route have been -reviewed. +controls. The default command-line mode remains ``--fixture`` and therefore +cannot qualify Issue #62. An explicitly enabled ``--enable-live-control`` +mode accepts a sanitized, operator-supplied plan, binds its candidate/catalog/ +route files, runs real CLI and sidecar commands through +:class:`BoundedPhaseRunner`, and still leaves qualification closed for +independent review. The harness has three non-negotiable properties: @@ -25,17 +27,20 @@ import argparse from dataclasses import dataclass from datetime import datetime, timezone +import hashlib import json import os from pathlib import Path import re import signal +import shutil +import stat import subprocess import sys import tempfile import threading import time -from typing import Callable, Literal, Sequence +from typing import Any, Callable, Literal, Mapping, NoReturn, Sequence SCHEMA_VERSION = "codexhub.issue62.live-control.v1" @@ -56,9 +61,94 @@ "process_start_failed", "stale_artifact", "termination_failed", + # Live-plan validation codes are intentionally fixed and safe to + # expose in the operator result. They never include paths, command + # arguments, or exception text. + "live_control_plan_missing", + "live_control_plan_invalid", + "candidate_binding_mismatch", + "plan_path_invalid", + "plan_path_outside_isolation", + "plan_path_linked", + "environment_invalid", + "cli_executable_binding_mismatch", + "helper_executable_binding_mismatch", + "planner_plan_incomplete", + "planner_disposition_invalid", + "candidate_sha_binding_mismatch", + "cli_package_binding_mismatch", + "catalog_binding_mismatch", + "route_binding_mismatch", + "control_labels_incomplete", + "sidecar_capture_missing", + "sidecar_capture_incomplete", + "identity_replay_incomplete", + "identity_replay_artifact_missing", + "identity_replay_artifact_invalid", + "manifest_reconcile_failed", } ) +LIVE_PLAN_SCHEMA = "codexhub.issue62.live-control-plan.v1" +LIVE_SCOPE = "authorized_live_control" +CONTROL_NAMES = ( + "streaming_text", + "streaming_function_history", + "non_streaming_text", + "choice_auto", + "choice_none", + "terminal_success", + "terminal_error", + "error_json", +) +CONTROL_NAME_SET = frozenset(CONTROL_NAMES) +LIVE_PLAN_FIELDS = frozenset( + { + "schema", + "verification_scope", + "isolation_root", + "candidate_identity", + "binding", + "catalog_model_entry_id", + "cli", + "sidecars", + "controls", + "replays", + "environment", + "planner", + } +) +LIVE_DISPOSITIONS = frozenset({"Preserved", "Unsupported", "Unqualified"}) +LIVE_ENV_KEYS = frozenset( + {"PATH", "SystemRoot", "ComSpec", "TEMP", "TMP", "PATHEXT", "PYTHONPATH"} +) +LOCAL_ENV_PATH_KEYS = frozenset({"PATH", "TEMP", "TMP", "PYTHONPATH"}) +SENSITIVE_ENV_KEYS = frozenset( + { + "CODEX_HOME", + "OPENAI_API_KEY", + "OLLAMA_API_KEY", + "CODEX_AUTH", + "HOME", + "USERPROFILE", + "HOMEDRIVE", + "HOMEPATH", + "APPDATA", + "LOCALAPPDATA", + } +) +PLANNER_FIELDS = frozenset( + {"model_visible_plan", "hosted_only_disposition", "unknown_tag_disposition"} +) + + +class LiveControlValidationError(ValueError): + """A deterministic, path-free live-control plan validation failure.""" + + def __init__(self, code: str) -> None: + self.code = code if code in ALLOWED_ERROR_CODES else "live_control_plan_invalid" + super().__init__(self.code) + class HarnessFailure(RuntimeError): """An internal failure represented by a safe, allow-listed code.""" @@ -151,6 +241,14 @@ class CommandResult: terminated: bool = True +@dataclass(frozen=True) +class BackgroundProcess: + """A bounded child kept alive while a control command executes.""" + + phase: str + process: subprocess.Popen[bytes] + + def _validate_timeout(timeout_seconds: float) -> float: if not MIN_TIMEOUT_SECONDS <= timeout_seconds <= MAX_TIMEOUT_SECONDS: raise ValueError("timeout must be between 30 and 60 seconds") @@ -185,6 +283,7 @@ def _count_extra_artifacts(run_root: Path) -> int: entry.name in ALLOWED_ARTIFACT_NAMES and entry.is_file() and not entry.is_symlink() + and os.stat(entry).st_nlink == 1 ): continue extras += 1 @@ -225,12 +324,14 @@ def _write_sanitized_json(target: Path, payload: dict[str, object]) -> None: def _terminate_process(process: subprocess.Popen[bytes]) -> bool: try: if process.poll() is not None: - return True + # A completed parent can still have a descendant. Treat that + # state as unknown rather than claiming the process tree is gone. + return False if os.name == "nt": # ``terminate`` only signals the direct child on Windows. The # Gateway/CLI can spawn grandchildren, so use taskkill's tree mode # and suppress all command output. - subprocess.run( + kill_result = subprocess.run( ["taskkill", "/PID", str(int(process.pid)), "/T", "/F"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, @@ -238,6 +339,8 @@ def _terminate_process(process: subprocess.Popen[bytes]) -> bool: check=False, timeout=5, ) + if kill_result.returncode != 0: + raise OSError("taskkill failed") else: os.killpg(int(process.pid), signal.SIGTERM) process.wait(timeout=5) @@ -268,15 +371,26 @@ def __init__( *, timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, process_factory: Callable[..., subprocess.Popen[bytes]] | None = None, + environment: Mapping[str, str] | None = None, + working_directory: Path | None = None, ) -> None: _ensure_fresh_run_root(run_root) self.run_root = run_root self.identity = identity self.timeout_seconds = _validate_timeout(timeout_seconds) self._process_factory = process_factory or subprocess.Popen + self._working_directory = str(working_directory) if working_directory is not None else None + self._environment = _safe_environment( + environment or {}, + working_directory=Path(working_directory) + if working_directory is not None + else None, + ) self.journal = SanitizedPhaseJournal(run_root, identity) self._active_process: subprocess.Popen[bytes] | None = None self._active_lock = threading.Lock() + self._background_processes: dict[str, BackgroundProcess] = {} + self._background_lock = threading.Lock() self._cancel_requested = False self._cleanup_called = False self._children_terminated = True @@ -291,6 +405,91 @@ def _set_active(self, process: subprocess.Popen[bytes] | None) -> None: with self._active_lock: self._active_process = process + @staticmethod + def _validate_argv(argv: Sequence[str]) -> bool: + return bool(argv) and all( + isinstance(item, str) and bool(item) and "\x00" not in item for item in argv + ) + + def start_background(self, phase: str, argv: Sequence[str]) -> BackgroundProcess: + """Start one bounded sidecar process without retaining its output.""" + + phase_name = self._phase_name(phase) + self.journal.append( + marker=f"{phase_name}_started", status="started", status_code="phase_started" + ) + if not self._validate_argv(argv): + self.journal.append( + marker=f"{phase_name}_completed", + status="failed", + status_code="process_start_failed", + ) + raise HarnessFailure("process_start_failed") + creationflags = 0 + start_new_session = False + if os.name == "nt": + creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200) + else: + start_new_session = True + try: + process = self._process_factory( + list(argv), + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env=self._environment, + cwd=self._working_directory, + close_fds=True, + creationflags=creationflags, + start_new_session=start_new_session, + shell=False, + ) + except Exception as error: + self.journal.append( + marker=f"{phase_name}_completed", + status="failed", + status_code="process_start_failed", + ) + raise HarnessFailure("process_start_failed") from error + handle = BackgroundProcess(phase_name, process) + with self._background_lock: + if phase_name in self._background_processes: + # A duplicate phase would make cleanup ambiguous. Terminate + # the just-started process before failing closed. + _terminate_process(process) + raise HarnessFailure("phase_exception") + self._background_processes[phase_name] = handle + return handle + + def stop_background(self, handle: BackgroundProcess, *, status_code: str = "cancelled") -> bool: + """Stop one background child and emit its terminal journal marker.""" + + with self._background_lock: + current = self._background_processes.pop(handle.phase, None) + if current is None: + return True + started = time.monotonic() + terminated = _terminate_process(current.process) + self._children_terminated = self._children_terminated and terminated + code = status_code if terminated else "termination_failed" + self.journal.append( + marker=f"{current.phase}_completed", + status="completed" if terminated else "failed", + status_code=code, + duration_ms=int((time.monotonic() - started) * 1000), + exit_code=(int(current.process.poll()) if current.process.poll() is not None else None), + ) + return terminated + + def _stop_background_processes(self) -> int: + with self._background_lock: + handles = list(self._background_processes.values()) + failures = 0 + for handle in handles: + if not self.stop_background(handle): + failures += 1 + return failures + def mark_phase( self, marker: str, @@ -351,7 +550,7 @@ def finish(result: CommandResult) -> CommandResult: return finish(CommandResult("cancelled", "cancelled", terminated=True)) if self._cancel_requested: return finish(CommandResult("cancelled", "cancelled", terminated=True)) - if not argv or any(not isinstance(item, str) or not item for item in argv): + if not self._validate_argv(argv): return finish(CommandResult("failed", "process_start_failed", terminated=True)) creationflags = 0 @@ -366,6 +565,9 @@ def finish(result: CommandResult) -> CommandResult: stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + env=self._environment, + cwd=self._working_directory, + close_fds=True, creationflags=creationflags, start_new_session=start_new_session, shell=False, @@ -450,6 +652,7 @@ def cleanup(self, actions: Sequence[CleanupAction] = ()) -> dict[str, object]: self.journal.append(marker="cleanup_started", status="started", status_code="phase_started") if not self.cancel(): failures += 1 + failures += self._stop_background_processes() for action in reversed(tuple(actions)): try: if action() is False: @@ -531,6 +734,1056 @@ def append( self.records.append(record) +def _live_plan_fail(code: str) -> NoReturn: + raise LiveControlValidationError(code) + + +def _validate_plan_argv(value: Any, code: str = "live_control_plan_invalid") -> list[str]: + if not isinstance(value, list) or not value or any( + not isinstance(item, str) + or not item + or "\x00" in item + or Path(item).is_absolute() + or re.match(r"^(?:[A-Za-z]:|[\\/]{2})", item) is not None + or ".." in Path(item).parts + for item in value + ): + _live_plan_fail(code) + # Return a fresh list so callers cannot mutate the source object while a + # plan is executing. + return list(value) + + +def _validate_plan_path(value: Any) -> str: + """Validate a relative plan path before it is joined to an isolated root.""" + + if not isinstance(value, str) or not value or "\x00" in value: + _live_plan_fail("plan_path_invalid") + candidate = Path(value) + # ``Path.is_absolute`` covers native Windows/POSIX roots; the explicit + # drive/UNC checks cover paths written for the other platform. + if candidate.is_absolute() or re.match(r"^(?:[A-Za-z]:|[\\/]{2})", value): + _live_plan_fail("plan_path_invalid") + if any(part in {"..", "."} for part in candidate.parts): + _live_plan_fail("plan_path_invalid") + return value + + +def _is_reparse(path: Path) -> bool: + try: + stat_result = os.lstat(path) + mode = stat_result.st_mode + if stat.S_ISLNK(mode): + return True + attributes = getattr(stat_result, "st_file_attributes", 0) + return bool(attributes & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) + except OSError: + return False + + +def _resolve_isolated_path( + value: Any, + *, + isolated_root: Path, + must_exist: bool = False, + allow_missing_parents: bool = False, +) -> str: + relative = _validate_plan_path(value) + try: + root = isolated_root.resolve(strict=True) + except OSError: + _live_plan_fail("plan_path_outside_isolation") + if _is_reparse(root): + _live_plan_fail("plan_path_linked") + target = root / relative + try: + resolved = target.resolve(strict=must_exist) + resolved.relative_to(root) + except (OSError, ValueError): + _live_plan_fail("plan_path_outside_isolation") + current = root + parts = Path(relative).parts + for index, part in enumerate(parts): + current = current / part + exists = current.exists() + if not exists: + if (index != len(parts) - 1 and not allow_missing_parents) or must_exist: + _live_plan_fail("plan_path_outside_isolation") + continue + if _is_reparse(current): + _live_plan_fail("plan_path_linked") + try: + mode = os.lstat(current).st_mode + if current.is_file() and getattr(os.stat(current), "st_nlink", 1) > 1: + _live_plan_fail("plan_path_linked") + if stat.S_ISREG(mode) and index != len(parts) - 1: + _live_plan_fail("plan_path_outside_isolation") + except OSError: + _live_plan_fail("plan_path_outside_isolation") + return str(resolved) + + +def _environment_value_is_local( + key: str, value: str, *, working_directory: Path | None = None +) -> bool: + """Reject host path injection in child environment overrides.""" + + if key not in LOCAL_ENV_PATH_KEYS: + return True + if working_directory is None: + return False + root = working_directory.resolve(strict=False) if working_directory is not None else None + for entry in value.split(os.pathsep): + if not entry: + continue + candidate = Path(entry) + if candidate.is_absolute() or re.match(r"^(?:[A-Za-z]:|[\\/]{2})", entry): + if root is None: + return False + try: + candidate.resolve(strict=False).relative_to(root) + except (OSError, ValueError, RuntimeError): + return False + elif ".." in candidate.parts: + return False + return True + + +def _validate_environment( + value: Any, *, isolated_root: Path | None = None +) -> dict[str, str]: + if not isinstance(value, Mapping): + _live_plan_fail("environment_invalid") + normalized: dict[str, str] = {} + for key, item in value.items(): + if ( + not isinstance(key, str) + or key in SENSITIVE_ENV_KEYS + or key not in LIVE_ENV_KEYS + or not isinstance(item, str) + or "\x00" in item + or len(item) > 4096 + or not _environment_value_is_local( + key, item, working_directory=isolated_root + ) + ): + _live_plan_fail("environment_invalid") + normalized[key] = item + return normalized + + +def _safe_environment( + overrides: Mapping[str, str], *, working_directory: Path | None = None +) -> dict[str, str]: + """Build a minimal child environment without host credentials/home state.""" + + result: dict[str, str] = {} + for key in ("SystemRoot", "ComSpec"): + value = os.environ.get(key) + if value: + result[key] = value + for key, value in overrides.items(): + if not _environment_value_is_local( + key, value, working_directory=working_directory + ): + continue + result[key] = value + # Never copy sensitive values from arbitrary direct callers. The public + # runner constructor is also used by fixture tests, so this function must + # enforce the credential/home boundary itself rather than relying on the + # live-plan validator. + for key in SENSITIVE_ENV_KEYS: + result.pop(key, None) + return result + + +def _validate_executable_spec( + value: Any, + *, + isolated_root: Path | None, + error_code: str, + extra_fields: frozenset[str] = frozenset(), +) -> dict[str, Any]: + expected = frozenset( + {"argv", "executable_file", "executable_sha256", "argv_file_digests"} + ) | extra_fields + if not isinstance(value, Mapping) or frozenset(value) != expected: + _live_plan_fail(error_code) + argv = _validate_plan_argv(value.get("argv")) + executable_file = _validate_plan_path(value.get("executable_file")) + digest = value.get("executable_sha256") + if not isinstance(digest, str) or re.fullmatch(r"[0-9a-fA-F]{64}", digest) is None: + _live_plan_fail(error_code) + if argv[0] != executable_file: + _live_plan_fail(error_code) + raw_files = value.get("argv_file_digests") + if not isinstance(raw_files, Mapping): + _live_plan_fail(error_code) + file_digests: dict[str, str] = {} + for path, file_digest in raw_files.items(): + if not isinstance(path, str) or not isinstance(file_digest, str): + _live_plan_fail(error_code) + if re.fullmatch(r"[0-9a-fA-F]{64}", file_digest) is None: + _live_plan_fail(error_code) + if isolated_root is not None: + resolved_file = _resolve_isolated_path( + path, isolated_root=isolated_root, must_exist=True + ) + if _file_sha256(resolved_file) != file_digest.lower(): + _live_plan_fail(error_code) + file_digests[path] = file_digest.lower() + if isolated_root is not None: + resolved = _resolve_isolated_path( + executable_file, isolated_root=isolated_root, must_exist=True + ) + if _file_sha256(resolved) != digest.lower(): + _live_plan_fail(error_code) + return { + "argv": argv, + "executable_file": executable_file, + "executable_sha256": digest.lower(), + "argv_file_digests": file_digests, + **{ + key: value[key] + for key in extra_fields + if key in value + }, + } + + +def _file_sha256(path: str) -> str: + try: + target = Path(path) + if target.is_symlink() or not target.is_file() or os.stat(target).st_nlink != 1: + raise OSError + digest = hashlib.sha256() + with target.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + except OSError: + return "" + + +def _validate_live_binding( + identity: CandidateIdentity, + binding: Mapping[str, Any], + *, + isolated_root: Path | None, +) -> dict[str, str]: + expected_fields = frozenset( + {"candidate_sha_file", "cli_package_file", "catalog_file", "route_file"} + ) + if frozenset(binding) != expected_fields: + _live_plan_fail("candidate_binding_mismatch") + if isolated_root is None: + paths = {key: _validate_plan_path(binding[key]) for key in expected_fields} + else: + paths = { + key: _resolve_isolated_path( + binding[key], isolated_root=isolated_root, must_exist=True + ) + for key in expected_fields + } + try: + candidate_text = Path(paths["candidate_sha_file"]).read_text(encoding="ascii").strip() + except (OSError, UnicodeError): + _live_plan_fail("candidate_sha_binding_mismatch") + if candidate_text.lower() != identity.candidate_sha: + _live_plan_fail("candidate_sha_binding_mismatch") + checks = ( + ("cli_package_file", identity.cli_package_sha256, "cli_package_binding_mismatch"), + ("catalog_file", identity.catalog_digest, "catalog_binding_mismatch"), + ("route_file", identity.route_digest, "route_binding_mismatch"), + ) + for field, expected, code in checks: + if _file_sha256(paths[field]) != expected: + _live_plan_fail(code) + return paths + + +def _validate_sidecar_spec(value: Any, *, isolated_root: Path | None) -> dict[str, Any]: + if not isinstance(value, Mapping) or frozenset(value) != frozenset( + { + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + "output_dir", + } + ): + _live_plan_fail("helper_executable_binding_mismatch") + executable = _validate_executable_spec( + { + key: value[key] + for key in ( + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + ) + }, + isolated_root=isolated_root, + error_code="helper_executable_binding_mismatch", + ) + output_dir = _validate_plan_path(value.get("output_dir")) + if isolated_root is not None: + output_dir = _resolve_isolated_path( + output_dir, isolated_root=isolated_root, allow_missing_parents=True + ) + return {**executable, "output_dir": output_dir} + + +def _validate_sidecars( + value: Any, *, isolated_root: Path | None, run_root: Path | None +) -> dict[str, list[dict[str, Any]]]: + if not isinstance(value, Mapping) or frozenset(value) != frozenset({"pre", "post"}): + _live_plan_fail("live_control_plan_invalid") + result: dict[str, list[dict[str, Any]]] = {} + output_dirs: set[str] = set() + for hop in ("pre", "post"): + raw = value.get(hop) + specs = raw if isinstance(raw, list) else [raw] + if not specs or len(specs) > len(CONTROL_NAMES): + _live_plan_fail("live_control_plan_invalid") + normalized = [ + _validate_sidecar_spec(item, isolated_root=isolated_root) for item in specs + ] + for spec in normalized: + directory_path = Path(spec["output_dir"]) + if run_root is not None: + try: + directory_path.resolve().relative_to(run_root.resolve()) + except ValueError: + _live_plan_fail("plan_path_outside_isolation") + directory = str(directory_path.resolve()) + if directory in output_dirs or any( + Path(directory).is_relative_to(Path(existing)) + or Path(existing).is_relative_to(Path(directory)) + for existing in output_dirs + ): + _live_plan_fail("live_control_plan_invalid") + output_dirs.add(directory) + result[hop] = normalized + return result + + +def load_live_control_plan( + source: Path | str | Mapping[str, Any], + *, + isolated_root: Path | None = None, + run_root: Path | None = None, +) -> dict[str, Any]: + """Load and independently bind an explicitly authorized live plan. + + This function performs no subprocess or network work. It is deliberately + strict: a plan must carry all candidate/catalog/route digest bindings and + exactly the eight semantic control labels before the live runner can start. + """ + + if isinstance(source, (str, Path)): + source_path = Path(source) + if ( + source_path.is_symlink() + or not source_path.exists() + or not source_path.is_file() + or os.stat(source_path).st_nlink != 1 + ): + _live_plan_fail("live_control_plan_missing") + try: + payload = json.loads(source_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + _live_plan_fail("live_control_plan_invalid") + else: + payload = source + if not isinstance(payload, Mapping) or frozenset(payload) != LIVE_PLAN_FIELDS: + _live_plan_fail("live_control_plan_invalid") + if payload.get("schema") != LIVE_PLAN_SCHEMA or payload.get("verification_scope") != LIVE_SCOPE: + _live_plan_fail("live_control_plan_invalid") + if payload.get("isolation_root") != ".": + _live_plan_fail("plan_path_invalid") + if isolated_root is not None: + try: + isolated_root = isolated_root.resolve(strict=True) + except OSError: + _live_plan_fail("plan_path_outside_isolation") + + raw_identity = payload.get("candidate_identity") + if not isinstance(raw_identity, Mapping): + _live_plan_fail("candidate_binding_mismatch") + identity_fields = frozenset( + {"candidate_sha", "cli_version", "cli_package_sha256", "catalog_digest", "route_digest"} + ) + if frozenset(raw_identity) not in (identity_fields, identity_fields | {"cli_source_commit"}): + _live_plan_fail("candidate_binding_mismatch") + try: + identity = CandidateIdentity( + candidate_sha=raw_identity["candidate_sha"], + cli_version=raw_identity["cli_version"], + cli_package_sha256=raw_identity["cli_package_sha256"], + catalog_digest=raw_identity["catalog_digest"], + route_digest=raw_identity["route_digest"], + cli_source_commit=raw_identity.get("cli_source_commit"), + ) + except (KeyError, TypeError, ValueError): + _live_plan_fail("candidate_binding_mismatch") + + binding = payload.get("binding") + if not isinstance(binding, Mapping): + _live_plan_fail("candidate_binding_mismatch") + normalized_binding = _validate_live_binding( + identity, binding, isolated_root=isolated_root + ) + + model = payload.get("catalog_model_entry_id") + if ( + not isinstance(model, str) + or re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", model) is None + or model != "gpt-5.6-sol" + ): + _live_plan_fail("catalog_binding_mismatch") + + cli = payload.get("cli") + if not isinstance(cli, Mapping) or frozenset(cli) != frozenset( + { + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + "cli_version", + } + ): + _live_plan_fail("cli_executable_binding_mismatch") + if cli.get("cli_version") != identity.cli_version: + _live_plan_fail("cli_executable_binding_mismatch") + normalized_cli = _validate_executable_spec( + cli, + isolated_root=isolated_root, + error_code="cli_executable_binding_mismatch", + extra_fields=frozenset({"cli_version"}), + ) + normalized_cli["cli_version"] = cli["cli_version"] + normalized_sidecars = _validate_sidecars( + payload.get("sidecars"), isolated_root=isolated_root, run_root=run_root + ) + normalized_environment = _validate_environment( + payload.get("environment"), isolated_root=isolated_root + ) + planner = payload.get("planner") + if not isinstance(planner, Mapping) or frozenset(planner) != PLANNER_FIELDS: + _live_plan_fail("planner_plan_incomplete") + if planner.get("model_visible_plan") != "complete": + _live_plan_fail("planner_plan_incomplete") + if planner.get("hosted_only_disposition") not in LIVE_DISPOSITIONS or planner.get( + "unknown_tag_disposition" + ) not in LIVE_DISPOSITIONS: + _live_plan_fail("planner_disposition_invalid") + + controls = payload.get("controls") + if not isinstance(controls, list) or len(controls) != len(CONTROL_NAMES): + _live_plan_fail("control_labels_incomplete") + normalized_controls: list[dict[str, Any]] = [] + names: list[str] = [] + record_paths: dict[str, set[str]] = {"pre": set(), "post": set()} + allowed_control_fields = frozenset({"name", "args", "capture", "pre_record", "post_record"}) + for item in controls: + if not isinstance(item, Mapping) or not frozenset(item).issubset(allowed_control_fields): + _live_plan_fail("live_control_plan_invalid") + name = item.get("name") + if not isinstance(name, str) or name not in CONTROL_NAME_SET: + _live_plan_fail("control_labels_incomplete") + names.append(name) + raw_args = item.get("args", []) + normalized: dict[str, Any] = { + "name": name, + "args": _validate_plan_argv(raw_args, "live_control_plan_invalid") + if raw_args != [] + else [], + "capture": item.get("capture", {}), + } + if not isinstance(normalized["capture"], Mapping): + _live_plan_fail("live_control_plan_invalid") + for field in ("pre_record", "post_record"): + if field in item: + if isolated_root is None: + normalized[field] = _validate_plan_path(item[field]) + else: + normalized[field] = _resolve_isolated_path( + item[field], isolated_root=isolated_root, allow_missing_parents=True + ) + normalized_path = str(Path(normalized[field]).resolve()) + if normalized_path in record_paths[field.removesuffix("_record")]: + _live_plan_fail("sidecar_capture_incomplete") + record_paths[field.removesuffix("_record")].add(normalized_path) + normalized_controls.append(normalized) + if set(names) != CONTROL_NAME_SET or len(set(names)) != len(names): + _live_plan_fail("control_labels_incomplete") + + replays = payload.get("replays") + replay_names = frozenset({"identity", "mutation", "deletion", "loss"}) + if not isinstance(replays, Mapping) or frozenset(replays) != replay_names: + _live_plan_fail("identity_replay_incomplete") + normalized_replays: dict[str, dict[str, Any]] = {} + for name in replay_names: + spec = replays.get(name) + if not isinstance(spec, Mapping) or frozenset(spec) != frozenset( + { + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + "artifact_file", + "case", + } + ) or spec.get("case") != name: + _live_plan_fail("identity_replay_incomplete") + replay = _validate_executable_spec( + spec, + isolated_root=isolated_root, + error_code="helper_executable_binding_mismatch", + extra_fields=frozenset({"artifact_file", "case"}), + ) + artifact_file = _validate_plan_path(replay.get("artifact_file")) + if isolated_root is not None: + artifact_file = _resolve_isolated_path( + artifact_file, isolated_root=isolated_root, allow_missing_parents=True + ) + if run_root is not None: + try: + Path(artifact_file).resolve().relative_to(run_root.resolve()) + except ValueError: + _live_plan_fail("plan_path_outside_isolation") + replay["artifact_file"] = artifact_file + normalized_replays[name] = replay + + return { + "schema": LIVE_PLAN_SCHEMA, + "verification_scope": LIVE_SCOPE, + "isolation_root": ".", + "candidate_identity": identity.as_dict(), + "binding": normalized_binding, + "catalog_model_entry_id": model, + "cli": normalized_cli, + "sidecars": normalized_sidecars, + "controls": normalized_controls, + "replays": normalized_replays, + "environment": normalized_environment, + "planner": dict(planner), + } + + +def _load_manifest_builder() -> Any: + # The evidence scripts are intentionally loosely coupled. Importing the + # builder lazily keeps fixture mode usable from a source checkout and does + # not add any production routing dependency. + try: + from build_issue_62_control_manifest import ( # type: ignore[import-not-found] + ManifestValidationError, + build_manifest, + reconcile_manifest, + replay_manifest, + ) + except ImportError as error: + raise HarnessFailure("phase_exception") from error + return ManifestValidationError, build_manifest, reconcile_manifest, replay_manifest + + +def _prepare_capture_dirs(sidecars: Mapping[str, Sequence[Mapping[str, Any]]]) -> None: + for specs in sidecars.values(): + for spec in specs: + target = Path(str(spec["output_dir"])) + try: + if target.exists(): + if target.is_symlink() or not target.is_dir() or any(target.iterdir()): + raise OSError + else: + target.mkdir(parents=True, exist_ok=False) + except OSError: + raise LiveControlValidationError("sidecar_capture_missing") + + +def _record_path_for_control( + control: Mapping[str, Any], + *, + hop: str, + specs: Sequence[Mapping[str, Any]], + index: int, +) -> Path: + def capture_files(root: Path) -> list[Path]: + try: + if root.is_symlink() or not root.is_dir(): + raise OSError + entries = list(root.iterdir()) + except OSError: + raise LiveControlValidationError("sidecar_capture_missing") + if len(entries) > len(CONTROL_NAMES) or any( + entry.is_symlink() + or entry.is_dir() + or entry.suffix.lower() != ".json" + or os.stat(entry).st_nlink != 1 + for entry in entries + ): + raise LiveControlValidationError("sidecar_capture_incomplete") + return sorted(entry for entry in entries if entry.is_file()) + + explicit = control.get(f"{hop}_record") + if explicit is not None: + path = Path(str(explicit)) + for spec in specs: + root = Path(str(spec["output_dir"])) + try: + path.resolve().relative_to(root.resolve()) + except ValueError: + continue + if path.resolve() not in {candidate.resolve() for candidate in capture_files(root)}: + raise LiveControlValidationError("sidecar_capture_missing") + return path + raise LiveControlValidationError("sidecar_capture_missing") + if len(specs) == len(CONTROL_NAMES): + root = Path(str(specs[index]["output_dir"])) + candidates = capture_files(root) + if len(candidates) == 1: + return candidates[0] + raise LiveControlValidationError("sidecar_capture_missing") + if len(specs) == 1: + root = Path(str(specs[0]["output_dir"])) + candidates = capture_files(root) + if len(candidates) == len(CONTROL_NAMES): + return candidates[index] + raise LiveControlValidationError("sidecar_capture_missing") + + +def _read_sidecar_record(path: Path, *, hop: str) -> dict[str, Any]: + try: + if path.is_symlink() or not path.is_file() or os.stat(path).st_nlink != 1: + raise OSError + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + raise LiveControlValidationError("sidecar_capture_incomplete") + try: + _ManifestValidationError, _build_manifest, _reconcile_manifest, _replay_manifest = ( + _load_manifest_builder() + ) + from build_issue_62_control_manifest import sanitize_sidecar_record # type: ignore[import-not-found] + + # Validate against the capture-only schema, but keep the raw + # transport metadata in memory for ``build_manifest``. The builder + # drops capture IDs/hop fields before writing the canonical manifest. + sanitize_sidecar_record(record, expected_hop=hop) + return dict(record) + except Exception as error: + if isinstance(error, LiveControlValidationError): + raise + raise LiveControlValidationError("sidecar_capture_incomplete") from error + + +def _build_live_controls( + plan: Mapping[str, Any], +) -> list[dict[str, Any]]: + sidecars = plan["sidecars"] + controls: list[dict[str, Any]] = [] + for index, control in enumerate(plan["controls"]): + semantic = dict(control.get("capture", {})) + semantic["name"] = control["name"] + semantic["pre"] = _read_sidecar_record( + _record_path_for_control(control, hop="pre", specs=sidecars["pre"], index=index), + hop="pre", + ) + semantic["post"] = _read_sidecar_record( + _record_path_for_control(control, hop="post", specs=sidecars["post"], index=index), + hop="post", + ) + controls.append(semantic) + return controls + + +def _manifest_candidate(plan: Mapping[str, Any]) -> dict[str, Any]: + identity = plan["candidate_identity"] + return { + "codexhub_candidate_sha": identity["candidate_sha"], + "cli_version": identity["cli_version"], + "cli_source_commit": identity.get("cli_source_commit"), + "cli_source_commit_status": ( + "published" if identity.get("cli_source_commit") else "not_published_by_registry" + ), + "cli_package_sha256": identity["cli_package_sha256"], + "catalog_snapshot_sha256": identity["catalog_digest"], + "catalog_model_entry_id": plan["catalog_model_entry_id"], + } + + +def _runtime_argv(spec: Mapping[str, Any], *, isolated_root: Path) -> list[str]: + executable = _resolve_isolated_path( + spec["executable_file"], isolated_root=isolated_root, must_exist=True + ) + for path, expected in spec.get("argv_file_digests", {}).items(): + resolved = _resolve_isolated_path(path, isolated_root=isolated_root, must_exist=True) + if _file_sha256(resolved) != expected: + raise LiveControlValidationError("helper_executable_binding_mismatch") + argv = list(spec["argv"]) + argv[0] = executable + return argv + + +def _remove_capture_dirs( + sidecars: Mapping[str, Sequence[Mapping[str, Any]]], *, run_root: Path +) -> bool: + failures = 0 + root = run_root.resolve() + for specs in sidecars.values(): + for spec in specs: + target = Path(str(spec["output_dir"])).resolve() + try: + target.relative_to(root) + if target == root: + raise OSError + if target.is_symlink(): + target.unlink() + elif target.exists(): + if not _remove_directory(target): + raise OSError + except (OSError, ValueError): + failures += 1 + return failures == 0 + + +def _remove_replay_artifacts( + replays: Mapping[str, Mapping[str, Any]], *, run_root: Path +) -> bool: + failures = 0 + root = run_root.resolve() + for replay in replays.values(): + target = Path(str(replay["artifact_file"])).resolve() + try: + target.relative_to(root) + if target.is_symlink(): + target.unlink() + elif target.exists(): + target.unlink() + except (OSError, ValueError): + failures += 1 + return failures == 0 + + +def _remove_directory(path: Path) -> bool: + try: + if path.is_symlink(): + path.unlink() + elif path.exists(): + if not path.is_dir(): + return False + for index, _nested in enumerate(path.rglob("*"), start=1): + if index >= MAX_NESTED_ARTIFACT_SCAN: + return False + shutil.rmtree(path) + return not path.exists() + except OSError: + return False + + +def _ensure_replay_artifacts_fresh( + replays: Mapping[str, Mapping[str, Any]], *, run_root: Path +) -> None: + """Reject a reused replay output before any child process starts.""" + + root = run_root.resolve() + seen: set[Path] = set() + for replay in replays.values(): + target = Path(str(replay["artifact_file"])).resolve() + try: + target.relative_to(root) + except ValueError: + raise LiveControlValidationError("plan_path_outside_isolation") + if target in seen or target.exists() or target.is_symlink(): + raise LiveControlValidationError("stale_artifact") + seen.add(target) + + +def _validate_replay_artifact( + path: Path, + *, + case: str, + candidate_sha: str, + manifest_sha: str, +) -> str: + # Diagnostic kept path-free; only existence is observed. + # (The caller redacts all path values from journal artifacts.) + try: + if path.is_symlink() or not path.is_file(): + raise OSError + artifact_sha = _file_sha256(str(path)) + if re.fullmatch(r"[0-9a-f]{64}", artifact_sha) is None: + raise OSError + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + raise LiveControlValidationError("identity_replay_artifact_missing") + expected_fields = frozenset( + {"schema", "case", "candidate_sha", "capture_manifest_sha256", "wire_replay", "outcome"} + ) + if not isinstance(value, Mapping) or frozenset(value) != expected_fields: + raise LiveControlValidationError("identity_replay_artifact_invalid") + if ( + value.get("schema") != "codexhub.issue62.identity-replay.v1" + or value.get("case") != case + or value.get("candidate_sha") != candidate_sha + or value.get("capture_manifest_sha256") != manifest_sha + or value.get("wire_replay") is not True + or value.get("outcome") != ("accepted" if case == "identity" else "rejected") + ): + raise LiveControlValidationError("identity_replay_artifact_invalid") + return artifact_sha + + +def run_live_control( + plan: Path | str | Mapping[str, Any], + *, + run_root: Path, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + manifest_out: Path | None = None, + isolated_root: Path | None = None, + cancellation_event: threading.Event | None = None, +) -> dict[str, Any]: + """Execute an explicitly supplied live plan through bounded children. + + The result is intentionally non-qualifying even when every command, + capture, and negative replay check succeeds. An independent maintainer + review is still required before any Issue #62 claim can be made. + """ + + try: + if isolated_root is None: + raise LiveControlValidationError("plan_path_outside_isolation") + raw_isolation = Path(isolated_root) + if _is_reparse(raw_isolation): + raise LiveControlValidationError("plan_path_linked") + isolation = raw_isolation.resolve(strict=True) + if isinstance(plan, (str, Path)): + raw_plan_path = Path(plan) + if ( + raw_plan_path.is_symlink() + or not raw_plan_path.is_file() + or os.stat(raw_plan_path).st_nlink != 1 + ): + raise LiveControlValidationError("plan_path_linked") + plan_path = raw_plan_path.resolve(strict=True) + plan_path.relative_to(isolation) + run_path = Path(run_root) + if not run_path.is_absolute(): + run_path = isolation / run_path + if _is_reparse(run_path): + raise LiveControlValidationError("plan_path_linked") + run_path = run_path.resolve(strict=False) + run_path.relative_to(isolation) + if run_path == isolation or _is_reparse(run_path): + raise LiveControlValidationError("plan_path_outside_isolation") + except (OSError, ValueError, RuntimeError): + raise LiveControlValidationError("plan_path_outside_isolation") + try: + loaded = load_live_control_plan( + plan, isolated_root=isolation, run_root=run_path + ) + identity = CandidateIdentity(**loaded["candidate_identity"]) + except LiveControlValidationError: + raise + except (OSError, ValueError, RuntimeError): + raise LiveControlValidationError("live_control_plan_invalid") + output_manifest: Path | None = None + if manifest_out is not None: + try: + output_manifest = Path(manifest_out) + if not output_manifest.is_absolute(): + output_manifest = isolation / output_manifest + if output_manifest.is_symlink(): + raise LiveControlValidationError("plan_path_linked") + output_manifest = output_manifest.resolve(strict=False) + output_manifest.relative_to(isolation) + try: + output_manifest.relative_to(run_path) + except ValueError: + pass + else: + raise LiveControlValidationError("plan_path_outside_isolation") + if output_manifest.exists(): + raise LiveControlValidationError("stale_artifact") + except LiveControlValidationError: + raise + except (OSError, ValueError, RuntimeError): + raise LiveControlValidationError("plan_path_outside_isolation") + case_temp = run_path / "temp" + child_environment = dict(loaded["environment"]) + child_environment["TEMP"] = str(case_temp) + child_environment["TMP"] = str(case_temp) + try: + runner = BoundedPhaseRunner( + run_path, + identity, + timeout_seconds=timeout_seconds, + environment=child_environment, + working_directory=isolation, + ) + except HarnessFailure: + raise + except (OSError, ValueError, RuntimeError): + raise LiveControlValidationError("live_control_plan_invalid") + handles: list[BackgroundProcess] = [] + result_status = "completed" + result_code = "ok" + manifest: dict[str, Any] | None = None + manifest_reconciled = False + replay_results: dict[str, str] = {} + cleanup_actions: tuple[CleanupAction, ...] = ( + lambda: _remove_capture_dirs(loaded["sidecars"], run_root=run_path), + lambda: _remove_replay_artifacts(loaded["replays"], run_root=run_path), + lambda: _remove_directory(case_temp), + ) + + def is_cancelled() -> bool: + return cancellation_event is not None and cancellation_event.is_set() + + try: + runner.mark_phase("binding_validated") + _ensure_replay_artifacts_fresh(loaded["replays"], run_root=run_path) + if is_cancelled(): + result_status, result_code = "cancelled", "cancelled" + else: + try: + case_temp.mkdir(parents=False, exist_ok=False) + except OSError: + result_status, result_code = "failed", "stale_artifact" + if result_status == "completed": + _prepare_capture_dirs(loaded["sidecars"]) + for hop in ("pre", "post"): + for index, spec in enumerate(loaded["sidecars"][hop]): + try: + handles.append( + runner.start_background( + f"{hop}_sidecar_{index}", + _runtime_argv(spec, isolated_root=isolation), + ) + ) + except HarnessFailure as error: + result_status, result_code = "failed", error.code + break + if result_status != "completed": + break + + if result_status == "completed" and any( + handle.process.poll() is not None for handle in handles + ): + result_status, result_code = "failed", "sidecar_capture_incomplete" + if result_status == "completed": + for control in loaded["controls"]: + if is_cancelled(): + result_status, result_code = "cancelled", "cancelled" + break + result = runner.run_subprocess( + f"control_{control['name']}", + [ + *_runtime_argv(loaded["cli"], isolated_root=isolation), + *control["args"], + ], + cancellation_event=cancellation_event, + ) + if result.status != "completed": + result_status, result_code = result.status, result.status_code + break + + # Stop sidecars before reading their atomic records. Cleanup repeats + # this operation defensively for cancellation/error paths. + for handle in reversed(handles): + if not runner.stop_background(handle) and result_status == "completed": + result_status, result_code = "failed", "termination_failed" + + if result_status == "completed": + try: + controls = _build_live_controls(loaded) + _ManifestValidationError, build_manifest, reconcile_manifest, replay_manifest = ( + _load_manifest_builder() + ) + manifest = build_manifest( + controls, + candidate_identity=_manifest_candidate(loaded), + verification_scope=LIVE_SCOPE, + ) + report = reconcile_manifest(manifest) + if not report["reconciled"]: + result_status, result_code = "failed", "manifest_reconcile_failed" + else: + manifest_reconciled = True + if output_manifest is not None: + _write_sanitized_json(output_manifest, manifest) + except LiveControlValidationError as error: + result_status, result_code = "failed", error.code + except Exception as error: + result_status, result_code = "failed", "manifest_reconcile_failed" + + if result_status == "completed" and manifest is not None: + for case in ("identity", "mutation", "deletion", "loss"): + replay = replay_manifest(manifest, case) + report = reconcile_manifest(replay) + expected_reconciled = case == "identity" + replay_results[case] = "pass" if report["reconciled"] is expected_reconciled else "fail" + if replay_results[case] != "pass": + result_status, result_code = "failed", "identity_replay_incomplete" + break + if result_status == "completed": + for case in ("identity", "mutation", "deletion", "loss"): + if is_cancelled(): + result_status, result_code = "cancelled", "cancelled" + break + replay_result = runner.run_subprocess( + f"replay_{case}", + _runtime_argv(loaded["replays"][case], isolated_root=isolation), + cancellation_event=cancellation_event, + ) + if replay_result.status != "completed": + result_status, result_code = "failed", "identity_replay_incomplete" + break + replay = loaded["replays"][case] + artifact_sha = _validate_replay_artifact( + Path(replay["artifact_file"]), + case=case, + candidate_sha=identity.candidate_sha, + manifest_sha=str(manifest["capture_manifest_sha256"]), + ) + replay_results[case] = f"artifact_bound:{artifact_sha}" + except LiveControlValidationError as error: + result_status, result_code = "failed", error.code + except HarnessFailure as error: + result_status, result_code = "failed", error.code + except Exception: + # Keep unexpected manifest/replay errors fail-closed and path-free. + result_status, result_code = "failed", "phase_exception" + finally: + try: + receipt = runner.cleanup(actions=cleanup_actions) + except HarnessFailure: + receipt = { + "cleanup_attempted": True, + "cleanup_completed": False, + "cleanup_failure_count": 1, + "child_processes_terminated": False, + "resources_released": False, + } + if receipt.get("cleanup_completed") is not True and result_status == "completed": + result_status, result_code = "failed", "cleanup_incomplete" + + output: dict[str, Any] = { + "completed": result_status == "completed", + "ready_for_issue62": False, + "status": result_status, + "status_code": result_code, + "cleanup_completed": receipt.get("cleanup_completed") is True, + "manifest_reconciled": manifest_reconciled, + "replay": replay_results, + "reason": "independent_review_required", + "planner": loaded["planner"], + } + if manifest is not None: + output["capture_manifest_sha256"] = manifest.get("capture_manifest_sha256") + return output + + def _fixture_identity(args: argparse.Namespace) -> CandidateIdentity: return CandidateIdentity( candidate_sha=args.candidate_sha, @@ -545,21 +1798,89 @@ def _fixture_identity(args: argparse.Namespace) -> CandidateIdentity: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run-root", type=Path, required=True) - parser.add_argument( - "--fixture", choices=("success", "timeout", "nonzero", "cancelled", "cleanup-failure"), required=True + parser.add_argument("--isolated-root", type=Path, help="Fresh-root parent for relative plan paths") + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument( + "--fixture", choices=("success", "timeout", "nonzero", "cancelled", "cleanup-failure") + ) + mode.add_argument( + "--enable-live-control", + action="store_true", + help="Run an explicitly supplied authorized live-control plan.", ) + parser.add_argument("--plan", type=Path, help="Sanitized live-control plan JSON") + parser.add_argument("--manifest-out", type=Path, help="Sanitized live manifest output") parser.add_argument("--timeout-seconds", type=float, default=DEFAULT_TIMEOUT_SECONDS) - parser.add_argument("--candidate-sha", required=True) - parser.add_argument("--cli-version", required=True) - parser.add_argument("--cli-package-sha256", required=True) - parser.add_argument("--catalog-digest", required=True) - parser.add_argument("--route-digest", required=True) + parser.add_argument("--candidate-sha") + parser.add_argument("--cli-version") + parser.add_argument("--cli-package-sha256") + parser.add_argument("--catalog-digest") + parser.add_argument("--route-digest") parser.add_argument("--cli-source-commit") return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) + if args.enable_live_control: + if args.plan is None: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": "live_control_plan_missing", + "cleanup_completed": False, + "reason": "live_control_plan_missing", + } + print(json.dumps(result, sort_keys=True)) + return 2 + try: + result = run_live_control( + args.plan, + run_root=args.run_root, + timeout_seconds=args.timeout_seconds, + manifest_out=args.manifest_out, + isolated_root=args.isolated_root, + ) + except LiveControlValidationError as error: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": error.code, + "cleanup_completed": False, + "reason": error.code, + } + except HarnessFailure as error: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": error.code, + "cleanup_completed": False, + "reason": error.code, + } + except Exception: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": "phase_exception", + "cleanup_completed": False, + "reason": "phase_exception", + } + print(json.dumps(result, sort_keys=True)) + return 0 if result.get("completed") is True and result.get("cleanup_completed") is True else 2 + + fixture_fields = ( + "candidate_sha", + "cli_version", + "cli_package_sha256", + "catalog_digest", + "route_digest", + ) + if any(getattr(args, field) is None for field in fixture_fields): + raise SystemExit("fixture mode requires candidate identity arguments") identity = _fixture_identity(args) runner = BoundedPhaseRunner( args.run_root, diff --git a/tests/test_issue62_bounded_phase_runner.py b/tests/test_issue62_bounded_phase_runner.py index c48c6f57..9459aebe 100644 --- a/tests/test_issue62_bounded_phase_runner.py +++ b/tests/test_issue62_bounded_phase_runner.py @@ -58,6 +58,26 @@ def test_success_writes_only_sanitized_phase_status_and_cleanup_receipt(tmp_path assert all("argv" not in record and "command" not in record for record in markers) +def test_direct_runner_drops_sensitive_and_host_path_overrides(tmp_path: Path) -> None: + runner = BoundedPhaseRunner( + tmp_path, + IDENTITY, + environment={ + "HOME": "C:/host/home", + "OPENAI_API_KEY": "must-not-pass", + "PATH": "C:/host/bin", + "PYTHONPATH": "../host-modules", + }, + working_directory=tmp_path, + ) + + assert "HOME" not in runner._environment + assert "OPENAI_API_KEY" not in runner._environment + assert "PATH" not in runner._environment + assert "PYTHONPATH" not in runner._environment + assert runner.cleanup()["cleanup_completed"] is True + + class _FakeProcess: def __init__(self, return_code: int | None = None) -> None: self.return_code = return_code diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py new file mode 100644 index 00000000..3d77a76b --- /dev/null +++ b/tests/test_issue_62_live_control_orchestration.py @@ -0,0 +1,335 @@ +from __future__ import annotations + +import hashlib +import importlib +import json +import os +from pathlib import Path +import shutil +import sys + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + +from run_issue_62_live_control import ( # noqa: E402 + CONTROL_NAMES, + LiveControlValidationError, + load_live_control_plan, + main, + run_live_control, +) + + +def _sha(value: str) -> str: + return hashlib.sha256(value.encode("ascii")).hexdigest() + + +def _binding_files(tmp_path: Path) -> dict[str, str]: + (tmp_path / "inputs").mkdir() + files = { + "candidate_sha_file": tmp_path / "inputs" / "candidate-sha", + "cli_package_file": tmp_path / "inputs" / "cli-package.tgz", + "catalog_file": tmp_path / "inputs" / "catalog.json", + "route_file": tmp_path / "inputs" / "route.json", + } + files["candidate_sha_file"].write_text("a" * 40 + "\n", encoding="ascii") + files["cli_package_file"].write_bytes(b"package") + files["catalog_file"].write_bytes(b"catalog") + files["route_file"].write_bytes(b"route") + return {key: str(value.relative_to(tmp_path)) for key, value in files.items()} + + +def _plan(tmp_path: Path) -> dict[str, object]: + binding = _binding_files(tmp_path) + (tmp_path / "helpers").mkdir() + (tmp_path / "run").mkdir() + tools_dir = tmp_path / "tools" + tools_dir.mkdir() + if os.name == "nt": + # A copied setup-python interpreter is not reliably launchable on a + # fresh Hosted image (its loader can reject the relocated DLL set). + # Use the system command host as the bound executable instead; all + # scripts and the timeout helper remain copied under this root. + executable_path = tools_dir / "cmd.exe" + shutil.copy2(Path(os.environ["SystemRoot"]) / "System32" / "cmd.exe", executable_path) + shutil.copy2(Path(os.environ["SystemRoot"]) / "System32" / "PING.EXE", tools_dir / "PING.EXE") + (tmp_path / "helpers" / "cli.cmd").write_text( + "@echo off\nexit /b 0\n", encoding="ascii" + ) + (tmp_path / "helpers" / "sidecar-empty.cmd").write_text( + "@echo off\nping 127.0.0.1 -n 300 >nul\n", encoding="ascii" + ) + (tmp_path / "helpers" / "replay.cmd").write_text( + "@echo off\n" + "set \"case=%~1\"\n" + "set \"candidate=%~2\"\n" + "set \"manifest=%~3\"\n" + "set \"output=%~4\"\n" + "if /I \"%case%\"==\"identity\" (set \"outcome=accepted\") else (set \"outcome=rejected\")\n" + ">\"%output%\" echo {\"schema\":\"codexhub.issue62.identity-replay.v1\",\"case\":\"%case%\",\"candidate_sha\":\"%candidate%\",\"capture_manifest_sha256\":\"%manifest%\",\"wire_replay\":true,\"outcome\":\"%outcome%\"}\n", + encoding="ascii", + ) + executable_file = str(executable_path.relative_to(tmp_path)) + environment = { + "PATH": str(tools_dir), + "PATHEXT": ".COM;.EXE;.BAT;.CMD", + } + cli_argv = [executable_file, "/d", "/c", "helpers\\cli.cmd"] + sidecar_argv = [executable_file, "/d", "/c", "helpers\\sidecar-empty.cmd"] + replay_argv = lambda case: [ + executable_file, + "/d", + "/c", + "helpers\\replay.cmd", + case, + "a" * 40, + "0" * 64, + f"run/replay-{case}.json", + ] + helper_names = {"cli": "cli.cmd", "sidecar": "sidecar-empty.cmd", "replay": "replay.cmd"} + else: + executable_path = tools_dir / "python" + shutil.copy2(sys.executable, executable_path) + (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") + (tmp_path / "helpers" / "sidecar-empty.py").write_text( + "import time\n" + "time.sleep(300)\n", + encoding="utf-8", + ) + (tmp_path / "helpers" / "replay.py").write_text( + "import json, pathlib, sys\n" + "case, candidate, manifest, output = sys.argv[1:5]\n" + "path = pathlib.Path(output)\n" + "path.parent.mkdir(parents=True, exist_ok=True)\n" + "path.write_text(json.dumps({'schema':'codexhub.issue62.identity-replay.v1', 'case':case, 'candidate_sha':candidate, 'capture_manifest_sha256':manifest, 'wire_replay':True, 'outcome':'accepted' if case == 'identity' else 'rejected'}, sort_keys=True, separators=(',', ':')) + '\\n', encoding='utf-8')\n", + encoding="utf-8", + ) + executable_file = str(executable_path.relative_to(tmp_path)) + environment = {} + cli_argv = [executable_file, "helpers/cli.py"] + sidecar_argv = [executable_file, "helpers/sidecar-empty.py"] + replay_argv = lambda case: [ + executable_file, "helpers/replay.py", case, "a" * 40, "0" * 64, f"run/replay-{case}.json" + ] + helper_names = {"cli": "cli.py", "sidecar": "sidecar-empty.py", "replay": "replay.py"} + executable_digest = hashlib.sha256(executable_path.read_bytes()).hexdigest() + executable = { + "executable_file": executable_file, + "executable_sha256": executable_digest, + } + file_digest = lambda name: hashlib.sha256( + (tmp_path / "helpers" / name).read_bytes() + ).hexdigest() + return { + "schema": "codexhub.issue62.live-control-plan.v1", + "verification_scope": "authorized_live_control", + "isolation_root": ".", + "candidate_identity": { + "candidate_sha": "a" * 40, + "cli_version": "0.146.0", + "cli_package_sha256": _sha("package"), + "catalog_digest": _sha("catalog"), + "route_digest": _sha("route"), + }, + "binding": binding, + "catalog_model_entry_id": "gpt-5.6-sol", + "environment": environment, + "planner": { + "model_visible_plan": "complete", + "hosted_only_disposition": "Unqualified", + "unknown_tag_disposition": "Unqualified", + }, + "cli": { + "argv": cli_argv, + **executable, + "argv_file_digests": {f"helpers/{helper_names['cli']}": file_digest(helper_names["cli"])}, + "cli_version": "0.146.0", + }, + "sidecars": { + "pre": { + "argv": sidecar_argv, + "output_dir": "run/pre", + **executable, + "argv_file_digests": { + f"helpers/{helper_names['sidecar']}": file_digest(helper_names["sidecar"]) + }, + }, + "post": { + "argv": sidecar_argv, + "output_dir": "run/post", + **executable, + "argv_file_digests": { + f"helpers/{helper_names['sidecar']}": file_digest(helper_names["sidecar"]) + }, + }, + }, + "controls": [ + {"name": name, "args": [], "capture": {}} + for name in CONTROL_NAMES + ], + "replays": { + case: { + "argv": replay_argv(case), + **executable, + "argv_file_digests": {f"helpers/{helper_names['replay']}": file_digest(helper_names["replay"])}, + "artifact_file": f"run/replay-{case}.json", + "case": case, + } + for case in ("identity", "mutation", "deletion", "loss") + }, + } + + +def test_missing_live_plan_uses_fixed_fail_closed_code(tmp_path: Path) -> None: + with pytest.raises(LiveControlValidationError, match="live_control_plan_missing"): + load_live_control_plan(tmp_path / "missing.json") + + +def test_cli_live_mode_reports_missing_plan_without_starting_children(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + assert main(["--run-root", str(tmp_path / "run"), "--enable-live-control"]) == 2 + output = json.loads(capsys.readouterr().out) + assert output["status_code"] == "live_control_plan_missing" + assert output["ready_for_issue62"] is False + + +def test_live_plan_binds_candidate_and_route_catalog_files(tmp_path: Path) -> None: + plan = _plan(tmp_path) + loaded = load_live_control_plan(plan, isolated_root=tmp_path) + assert loaded["candidate_identity"]["candidate_sha"] == "a" * 40 + + (tmp_path / "inputs" / "route.json").write_bytes(b"changed") + with pytest.raises(LiveControlValidationError, match="route_binding_mismatch"): + load_live_control_plan(plan, isolated_root=tmp_path) + + +def test_live_plan_requires_exactly_eight_control_labels(tmp_path: Path) -> None: + plan = _plan(tmp_path) + plan["controls"] = list(plan["controls"])[1:] + with pytest.raises(LiveControlValidationError, match="control_labels_incomplete"): + load_live_control_plan(plan, isolated_root=tmp_path) + + +def test_live_execution_requires_complete_pre_and_post_sidecar_records(tmp_path: Path) -> None: + result = run_live_control( + _plan(tmp_path), + run_root=tmp_path / "run", + timeout_seconds=30, + isolated_root=tmp_path, + ) + assert result["ready_for_issue62"] is False + assert result["status_code"] == "sidecar_capture_missing" + receipt = json.loads((tmp_path / "run" / "cleanup-receipt.json").read_text(encoding="utf-8")) + assert receipt["cleanup_attempted"] is True + assert receipt["cleanup_completed"] is True + + +def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_path: Path) -> None: + source = importlib.import_module("test_issue_62_control_manifest") + plan = _plan(tmp_path) + controls = source._controls() + pre_dir = tmp_path / "inputs" / "pre-records" + post_dir = tmp_path / "inputs" / "post-records" + pre_dir.mkdir() + post_dir.mkdir() + payloads: dict[str, Path] = {} + for index, control in enumerate(controls): + pre = tmp_path / f"pre-{index}.json" + post = tmp_path / f"post-{index}.json" + pre.write_text(json.dumps(control["pre"]), encoding="utf-8") + post.write_text(json.dumps(control["post"]), encoding="utf-8") + payloads[f"pre-{index}"] = pre + payloads[f"post-{index}"] = post + semantic = { + key: value + for key, value in control.items() + if key not in {"name", "pre", "post"} + } + plan["controls"][index] = { + "name": control["name"], + "args": [], + "capture": semantic, + "pre_record": f"run/pre-{index}/pre-c{index}.json", + "post_record": f"run/post-{index}/post-c{index}.json", + } + if os.name == "nt": + sidecar_name = "sidecar.cmd" + sidecar_script = tmp_path / "helpers" / sidecar_name + sidecar_script.write_text( + "@echo off\n" + "copy /y \"%~2\" \"%~1\\%~3\" >nul\n" + "ping 127.0.0.1 -n 300 >nul\n", + encoding="ascii", + ) + sidecar_prefix = [plan["cli"]["executable_file"], "/d", "/c", "helpers\\sidecar.cmd"] + else: + sidecar_name = "sidecar.py" + sidecar_script = tmp_path / "helpers" / sidecar_name + sidecar_script.write_text( + "import pathlib, shutil, sys, time\n" + "out = pathlib.Path(sys.argv[1]); src = pathlib.Path(sys.argv[2])\n" + "shutil.copyfile(src, out / pathlib.Path(sys.argv[3]).name)\n" + "time.sleep(300)\n", + encoding="utf-8", + ) + sidecar_prefix = [plan["cli"]["executable_file"], "helpers/sidecar.py"] + for hop in ("pre", "post"): + specs = plan["sidecars"][hop] + source_paths = [payloads[f"{hop}-{index}"] for index in range(8)] + # One sidecar per control gives deterministic pairing without relying + # on opaque capture-id ordering. + plan["sidecars"][hop] = [ + { + "argv": [ + *sidecar_prefix, + f"run/{hop}-{index}", + str(source_paths[index].relative_to(tmp_path)), + f"{hop}-c{index}.json", + ], + "output_dir": f"run/{hop}-{index}", + "executable_file": plan["cli"]["executable_file"], + "executable_sha256": plan["cli"]["executable_sha256"], + "argv_file_digests": { + f"helpers/{sidecar_name}": hashlib.sha256(sidecar_script.read_bytes()).hexdigest() + }, + } + for index in range(8) + ] + for index in range(8): + plan["controls"][index][f"{hop}_record"] = ( + f"run/{hop}-{index}/{hop}-c{index}.json" + ) + # Re-run plan validation after replacing the sidecar specs. + result = run_live_control( + plan, + run_root=tmp_path / "run", + timeout_seconds=30, + isolated_root=tmp_path, + ) + assert result["completed"] is False + assert result["ready_for_issue62"] is False + assert result["status_code"] == "identity_replay_artifact_invalid" + assert result["cleanup_completed"] is True + assert result["manifest_reconciled"] is True + assert not (tmp_path / "run" / "pre-0").exists() + assert not (tmp_path / "run" / "post-0").exists() + assert not (tmp_path / "run" / "replay-identity.json").exists() + + +def test_live_plan_rejects_host_credentials_and_duplicate_capture_paths(tmp_path: Path) -> None: + plan = _plan(tmp_path) + plan["environment"] = {"OPENAI_API_KEY": "must-not-pass"} + with pytest.raises(LiveControlValidationError, match="environment_invalid"): + load_live_control_plan(plan, isolated_root=tmp_path) + + duplicate_root = tmp_path / "duplicate" + duplicate_root.mkdir() + plan = _plan(duplicate_root) + plan["controls"][1]["pre_record"] = "run/shared.json" + plan["controls"][0]["pre_record"] = "run/shared.json" + with pytest.raises(LiveControlValidationError, match="sidecar_capture_incomplete"): + load_live_control_plan(plan, isolated_root=duplicate_root) diff --git a/tests/test_issue_62_live_evidence_sidecar.py b/tests/test_issue_62_live_evidence_sidecar.py index 72456162..954ae825 100644 --- a/tests/test_issue_62_live_evidence_sidecar.py +++ b/tests/test_issue_62_live_evidence_sidecar.py @@ -28,6 +28,16 @@ SPEC.loader.exec_module(sidecar) KEY = b"h" * 32 +CORRELATION_TOKEN = "1" * 32 + + +def _capture_id(correlation_token: str = CORRELATION_TOKEN) -> str: + return sidecar._capture_id_for_token(KEY, correlation_token) + + +def test_capture_id_is_deterministically_bound_to_correlation_token() -> None: + assert _capture_id() == _capture_id(CORRELATION_TOKEN) + assert _capture_id() != _capture_id("2" * 32) class _FakeUpstreamHandler(BaseHTTPRequestHandler): @@ -36,6 +46,9 @@ class _FakeUpstreamHandler(BaseHTTPRequestHandler): def do_POST(self) -> None: length = int(self.headers["Content-Length"]) self.server.observed_request = self.rfile.read(length) # type: ignore[attr-defined] + self.server.observed_headers = { # type: ignore[attr-defined] + name.lower(): value for name, value in self.headers.items() + } body = self.server.response_body # type: ignore[attr-defined] delay = self.server.response_delay # type: ignore[attr-defined] if delay: @@ -69,6 +82,7 @@ def _fake_upstream( server = ThreadingHTTPServer(("127.0.0.1", 0), _FakeUpstreamHandler) server.response_body = response_body # type: ignore[attr-defined] server.observed_request = None # type: ignore[attr-defined] + server.observed_headers = {} # type: ignore[attr-defined] server.response_delay = response_delay # type: ignore[attr-defined] server.response_chunk_size = chunk_size # type: ignore[attr-defined] server.chunk_delay = chunk_delay # type: ignore[attr-defined] @@ -214,7 +228,7 @@ def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path: Path) -> Non record = { "schema": "codexhub.issue62.live-evidence-lane.v1", "verification_scope": "capture_only_not_qualification", - "capture_id": "c" + "1" * 32, + "capture_id": _capture_id(), "hop": "pre", "outcome": "complete", "failure": None, @@ -235,7 +249,13 @@ def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path: Path) -> Non "sse": None, } - record_path = sidecar.write_capture_record(tmp_path, "pre", record) + record_path = sidecar.write_capture_record( + tmp_path, + "pre", + record, + capture_key=KEY, + correlation_token=CORRELATION_TOKEN, + ) assert json.loads(record_path.read_text(encoding="utf-8")) == record assert record_path.name.startswith("pre-") @@ -243,6 +263,70 @@ def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path: Path) -> Non assert not list(tmp_path.glob("*.partial")) +def test_atomic_record_rejects_prebuilt_record_without_live_capture_binding(tmp_path: Path) -> None: + record = { + "schema": "codexhub.issue62.live-evidence-lane.v1", + "verification_scope": "capture_only_not_qualification", + "capture_id": _capture_id("2" * 32), + "hop": "pre", + "outcome": "complete", + "failure": None, + "status": 200, + "content_type_class": "json", + "request": { + "bytes": 2, + "sha256": "a" * 64, + "hmac_sha256": "b" * 64, + "complete": True, + }, + "response": { + "bytes": 3, + "sha256": "c" * 64, + "hmac_sha256": "d" * 64, + "complete": True, + }, + "sse": None, + } + + with pytest.raises(sidecar.ArtifactValidationError, match="capture_binding_missing"): + sidecar.write_capture_record(tmp_path, "pre", record) + + +def test_atomic_record_rejects_capture_id_from_different_live_correlation(tmp_path: Path) -> None: + record = { + "schema": "codexhub.issue62.live-evidence-lane.v1", + "verification_scope": "capture_only_not_qualification", + "capture_id": _capture_id("2" * 32), + "hop": "pre", + "outcome": "complete", + "failure": None, + "status": 200, + "content_type_class": "json", + "request": { + "bytes": 2, + "sha256": "a" * 64, + "hmac_sha256": "b" * 64, + "complete": True, + }, + "response": { + "bytes": 3, + "sha256": "c" * 64, + "hmac_sha256": "d" * 64, + "complete": True, + }, + "sse": None, + } + + with pytest.raises(sidecar.ArtifactValidationError, match="capture_binding_invalid"): + sidecar.write_capture_record( + tmp_path, + "pre", + record, + capture_key=KEY, + correlation_token=CORRELATION_TOKEN, + ) + + def test_atomic_record_rejects_unapproved_fields(tmp_path: Path) -> None: with pytest.raises(sidecar.ArtifactValidationError, match="record_fields_invalid"): sidecar.write_capture_record( @@ -515,6 +599,7 @@ def test_two_hops_capture_matching_complete_request_response_and_sse( headers={ "Authorization": "Bearer forbidden-token", "Content-Type": "application/json", + "X-CodexHub-Issue62-Capture": ("0" * 32) + "." + ("f" * 64), }, ) response = connection.getresponse() @@ -529,10 +614,12 @@ def test_two_hops_capture_matching_complete_request_response_and_sse( pre_record = _read_only_record(pre_output) post_record = _read_only_record(post_output) assert pre_record["request"] == post_record["request"] + assert pre_record["capture_id"] == post_record["capture_id"] assert pre_record["response"] == post_record["response"] assert pre_record["sse"]["sequence_sha256"] == post_record["sse"]["sequence_sha256"] # type: ignore[index] assert pre_record["sse"]["sequence_hmac_sha256"] == post_record["sse"]["sequence_hmac_sha256"] # type: ignore[index] assert pre_record["outcome"] == post_record["outcome"] == "complete" + assert "x-codexhub-issue62-capture" not in upstream.observed_headers # type: ignore[attr-defined] serialized = json.dumps([pre_record, post_record], sort_keys=True) for sensitive in ( "private prompt", @@ -548,6 +635,42 @@ def test_two_hops_capture_matching_complete_request_response_and_sse( assert not list(tmp_path.rglob("*.partial")) +@pytest.mark.parametrize( + "correlation_header", + [None, ("0" * 32) + "." + ("f" * 64)], +) +def test_post_rejects_missing_or_invalid_correlation_before_forwarding( + correlation_header: str | None, + tmp_path: Path, + hmac_key_file: Path, +) -> None: + with _fake_upstream(b'data: {"type":"response.completed"}\n\n') as upstream: + upstream_url = f"http://127.0.0.1:{upstream.server_address[1]}" + config = _sidecar_config( + tmp_path, + hmac_key_file, + upstream_url, + hop="post", + ) + with _running_sidecar(config) as server: + connection = http.client.HTTPConnection(server.listen_host, server.listen_port, timeout=2) + headers = {} + if correlation_header is not None: + headers["X-CodexHub-Issue62-Capture"] = correlation_header + connection.request("POST", "/responses", body=b"{}", headers=headers) + response = connection.getresponse() + assert response.status == 400 + response.read() + connection.close() + + assert upstream.observed_request is None # type: ignore[attr-defined] + record = _read_only_record(config.output_dir) + assert record["outcome"] == "incomplete" + assert record["failure"] == "correlation_binding_invalid" + assert "X-CodexHub-Issue62-Capture" not in json.dumps(record, sort_keys=True) + assert not list(tmp_path.rglob("*.partial")) + + def test_request_overflow_fails_closed_before_forwarding( tmp_path: Path, hmac_key_file: Path,