diff --git a/.agents/skills/rpce-contribution-check/scripts/preflight.sh b/.agents/skills/rpce-contribution-check/scripts/preflight.sh index 0d2a4fa08..041b70a53 100755 --- a/.agents/skills/rpce-contribution-check/scripts/preflight.sh +++ b/.agents/skills/rpce-contribution-check/scripts/preflight.sh @@ -98,7 +98,7 @@ run_pr_ready_path_validations() { files="$tmp_root/range-files.z" write_range_files "$files" - local control_plane_paths_pattern='^(Scripts/conductor\.py|Scripts/guardrails\.sh|Scripts/test_conductor_(lifecycle|output)\.py|Scripts/test_contribution_preflight\.py|\.agents/skills/rpce-contribution-check/scripts/preflight\.sh|Makefile)$' + local control_plane_paths_pattern='^(Scripts/conductor\.py|Scripts/failure_diagnostics\.py|Scripts/guardrails\.sh|Scripts/source_layout_guardrails\.sh|Scripts/test_conductor_(lifecycle|output|failure_diagnostics)\.py|Scripts/test_contribution_preflight\.py|\.agents/skills/rpce-contribution-check/scripts/preflight\.sh|Makefile)$' local ci_app_test_runner_paths_pattern='^(Scripts/ci_app_test_runner\.py|Scripts/test_ci_app_test_runner\.py|\.github/workflows/ci\.yml)$' local swift_paths_pattern='\.swift$' local root_test_paths_pattern='^(Sources/RepoPrompt/|Tests/RepoPrompt[^/]*Tests/)' diff --git a/Makefile b/Makefile index 54febb3e0..07bfe6265 100644 --- a/Makefile +++ b/Makefile @@ -117,6 +117,7 @@ conductor-selftest: python3 Scripts/test_contribution_preflight.py python3 Scripts/test_ci_app_test_runner.py python3 Scripts/test_conductor_output.py + python3 Scripts/test_conductor_failure_diagnostics.py python3 Scripts/test_agent_mode_file_tools_benchmark.py python3 Scripts/test_conductor_lifecycle.py python3 Scripts/test_local_production_installer.py diff --git a/Scripts/conductor.py b/Scripts/conductor.py index e599407b7..edcf09af9 100755 --- a/Scripts/conductor.py +++ b/Scripts/conductor.py @@ -35,6 +35,7 @@ from typing import Any, Deque, Dict, List, Optional, Sequence, Tuple from debug_app_process import ProcessIdentityError, matching_processes, terminate_matching_processes +import failure_diagnostics PROTOCOL_VERSION = 10 TERMINAL_STATES = {"completed", "failed", "canceled"} @@ -148,6 +149,7 @@ (without --launch/--packaged-app, requires the CE debug app to already be running and CLI installed) ./conductor diagnostics agent-mode-on [--log-file ] ./conductor diagnostics build-cache [--limit ] + ./conductor diagnostics recent-failures [--limit ] [--operation ] [--failure-class ] [--hours ] [--json] ./conductor release preflight|artifact|package|local-install Foundation validation operation: @@ -1142,6 +1144,8 @@ class Job: diagnostic_paths: List[Path] = dataclasses.field(default_factory=list, repr=False) output_summary: Optional[Dict[str, Any]] = None tail: Deque[str] = dataclasses.field(default_factory=lambda: deque(maxlen=LOG_TAIL_LINES)) + failure_record_pending: bool = False + failure_record_written: bool = False def to_payload(self, include_tail: bool = True, include_summary: bool = True) -> Dict[str, Any]: queue_wait_seconds = None @@ -1526,6 +1530,11 @@ def __init__(self, paths: Paths) -> None: self.active_lanes: Dict[str, str] = {} self.shutdown_requested = False self.server: Optional[socketserver.BaseServer] = None + self.failure_store = failure_diagnostics.FailureRecordStore( + paths.jobs_dir, + max_age_seconds=TERMINAL_RETENTION_SECONDS, + max_records=MAX_TERMINAL_JOBS, + ) def _global_heavy_slot_paths(self, env: Optional[Dict[str, str]] = None) -> List[Path]: return global_heavy_slot_paths(env) @@ -2167,10 +2176,13 @@ def _run_job(self, ticket: str) -> None: self._release_global_heavy_slot(global_heavy_slot) if output_transport is not None: output_transport.close_all() - refresh_after_release = False + # Persist the failure record before any job cleanup (in particular + # before retention pass) so the record owns the lifetime of the job + # summary and local log it references. + if job is not None and job.state in TERMINAL_STATES: + self._refresh_output_summary(job) with self.condition: if job is not None: - refresh_after_release = job.state in TERMINAL_STATES and job.output_summary is None for lane in list(job.lanes): if self.active_lanes.get(lane) == job.ticket: del self.active_lanes[lane] @@ -2178,8 +2190,6 @@ def _run_job(self, ticket: str) -> None: self._retention_pass_locked() self._schedule_locked() self.condition.notify_all() - if job is not None and refresh_after_release: - threading.Thread(target=self._refresh_output_summary, args=(job,), daemon=True).start() @staticmethod def _take_complete_output_lines(pending: bytearray, chunk: bytes) -> List[bytes]: @@ -2558,9 +2568,31 @@ def _refresh_output_summary(self, job: Job) -> None: ) with self.condition: current = self.jobs.get(job.ticket) - if current is job and current.output_summary is None: - current.output_summary = summary + if current is None or current is not job: + return + if current.failure_record_pending or current.failure_record_written: + return + current.failure_record_pending = True + # Snapshot the job state while protected by the condition, then + # persist the record outside the critical section. + record = failure_diagnostics.FailureRecord.from_job( + current, + summary, + jobs_dir=self.paths.jobs_dir, + ) + try: + self.failure_store.write(record, summary) + except Exception as exc: + with self.condition: + current.failure_record_pending = False + self._append_system_line_locked(job, f"failure record write failed: {exc}\n") self.condition.notify_all() + return + with self.condition: + current.output_summary = summary + current.failure_record_written = True + current.failure_record_pending = False + self.condition.notify_all() def _append_tail_locked(self, job: Job, text: str) -> None: lines = text.splitlines(keepends=True) @@ -2896,6 +2928,7 @@ def run_daemon(paths: Paths) -> int: with contextlib.suppress(OSError): os.chmod(paths.running_processes_path, 0o600) state = DaemonState(paths) + state.failure_store.retention_pass() server = ThreadedUnixServer(str(paths.socket_path), RequestHandler, state) with contextlib.suppress(OSError): os.chmod(paths.socket_path, 0o600) @@ -4570,6 +4603,33 @@ def parse_no_args(prog: str, argv: List[str]) -> None: parser.parse_args(argv) +def handle_recent_failures_query(paths: Paths, args: Dict[str, Any], json_mode: bool) -> int: + """Read-only query for recent failure records; does not touch the daemon.""" + store = failure_diagnostics.FailureRecordStore( + paths.jobs_dir, + max_age_seconds=TERMINAL_RETENTION_SECONDS, + max_records=MAX_TERMINAL_JOBS, + ) + hours = args.get("hours") + since = now() - hours * 3600.0 if hours is not None else None + requested_limit = int(args.get("limit") or 10) + failure_class = args.get("failureClass") + # Fetch a generous window so the default "non-success" filter can still + # honor the user's limit. + records = store.query_recent( + limit=MAX_TERMINAL_JOBS, + operation=args.get("operation"), + failure_class=failure_class, + since_timestamp=since, + ) + if failure_class is None: + records = [r for r in records if r.failure_class != "none"] + records = records[:requested_limit] + output = failure_diagnostics.format_recent_failures(records, json_mode=json_mode) + print(output) + return 0 + + def handle_real_operation(paths: Paths, operation: str, argv: List[str]) -> int: global_flags, rest = split_operation_flags(argv) if global_flags.timeout is not None and global_flags.timeout < 0: @@ -4682,6 +4742,12 @@ def handle_real_operation(paths: Paths, operation: str, argv: List[str]) -> int: build_cache = subparsers.add_parser("build-cache") build_cache.add_argument("--limit", type=int, default=BUILD_CACHE_DIAGNOSTIC_MAX_ROWS) + recent_failures = subparsers.add_parser("recent-failures") + recent_failures.add_argument("--limit", type=int, default=10) + recent_failures.add_argument("--operation") + recent_failures.add_argument("--failure-class", choices=failure_diagnostics.FAILURE_CLASSES) + recent_failures.add_argument("--hours", type=float) + ns = parser.parse_args(rest) args["subcommand"] = ns.subcommand if ns.subcommand == "agent-mode-on": @@ -4690,6 +4756,16 @@ def handle_real_operation(paths: Paths, operation: str, argv: List[str]) -> int: if ns.limit <= 0: raise ConductorError("diagnostics build-cache --limit must be greater than zero") args["limit"] = ns.limit + elif ns.subcommand == "recent-failures": + if ns.limit <= 0: + raise ConductorError("diagnostics recent-failures --limit must be greater than zero") + if ns.hours is not None and ns.hours <= 0: + raise ConductorError("diagnostics recent-failures --hours must be greater than zero") + args["limit"] = ns.limit + args["operation"] = ns.operation + args["failureClass"] = ns.failure_class + args["hours"] = ns.hours + return handle_recent_failures_query(paths, args, global_flags.json) elif operation == "release": parser = argparse.ArgumentParser(prog="conductor release") parser.add_argument("subcommand", choices=["preflight", "artifact", "package", "local-install"]) diff --git a/Scripts/failure_diagnostics.py b/Scripts/failure_diagnostics.py new file mode 100644 index 000000000..a244c6e9e --- /dev/null +++ b/Scripts/failure_diagnostics.py @@ -0,0 +1,719 @@ +#!/usr/bin/env python3 +"""Local-first structured build/test failure diagnostics for conductor. + +This module is intentionally independent from the daemon: it defines a small, +versioned failure record, a bounded on-disk store, conservative classification, +and a read-only query surface. It never copies raw log text, source content, +prompts, transcripts, credentials, or environment dumps into aggregate records. +""" + +from __future__ import annotations + +import contextlib +import dataclasses +import json +import os +import tempfile +import threading +import time +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Tuple + + +CURRENT_SCHEMA_VERSION = 1 +SCHEMA_LINEAGE = "repoprompt-ce.failure-record" +LEGACY_UNLINEAGED_SCHEMA_VERSION_CEILING = 0 + +# Bounded by the same floor as terminal job retention in conductor.py. +DEFAULT_FAILURE_RECORD_RETENTION_SECONDS = 24 * 60 * 60 +DEFAULT_MAX_FAILURE_RECORDS = 200 + +# Environment keys treated as toolchain-known metadata. The daemon already +# snapshots a filtered env list, so this is a further narrow selection. +TOOLCHAIN_KNOWN_ENV_KEYS = { + "CC", + "CXX", + "DEVELOPER_DIR", + "SDKROOT", + "SWIFT_EXEC", + "TOOLCHAINS", +} + +FAILURE_CLASSES = { + "none", + "timeout", + "sourceMutatedBuild", + "compilerFailure", + "testFailure", + "processCleanupFailure", + "heavyLaneWait", + "infrastructureOrRPCFailure", + "cancellation", + "unknown", +} + +# Operation args that are unbounded, user-supplied strings or paths. These are +# filtered out of aggregate failure records so the record never carries a raw +# prompt, log contents, or a filesystem path. +REDACTED_RECORD_ARG_KEYS = {"message", "logFile"} + +EXIT_CLASSES = { + "completed", + "timeout", + "canceled", + "killed", + "infrastructure", + "nonZero", + "unknown", +} + +AGGREGATE_JSON_PRIVACY_NOTE = ( + "aggregate-only; excludes raw logs, source content, prompts, transcripts, " + "credentials, environment dumps, and unbounded operation args" +) + + +class FailureDiagnosticsError(Exception): + """Error raised by failure diagnostics internals.""" + + +@dataclasses.dataclass +class FailureRecord: + """Versioned local failure record for a terminal conductor job. + + The record is intentionally small and referential: it points to the local + job log and a persisted output summary, but it does not contain raw log + text or source content. + """ + + schema_version: int = CURRENT_SCHEMA_VERSION + schema_lineage: str = SCHEMA_LINEAGE + recorded_at: Optional[float] = None + ticket: str = "" + request_key: Optional[str] = None + fingerprint: Optional[str] = None + operation: str = "" + operation_label: str = "" + args: Dict[str, Any] = dataclasses.field(default_factory=dict) + lanes: List[str] = dataclasses.field(default_factory=list) + created_at: Optional[float] = None + started_at: Optional[float] = None + finished_at: Optional[float] = None + queued_at: Optional[float] = None + process_started_at: Optional[float] = None + process_finished_at: Optional[float] = None + queue_wait_seconds: Optional[float] = None + execution_seconds: Optional[float] = None + terminal_state: str = "" + exit_code: Optional[int] = None + exit_class: str = "" + failure_class: str = "" + failure_class_reason: str = "" + timed_out: bool = False + cancel_requested: bool = False + measurement_invalid: bool = False + superseded_by_ticket: Optional[str] = None + superseded_by_operation: Optional[str] = None + global_heavy_slot_wait_seconds: Optional[float] = None + global_heavy_slot_path: Optional[str] = None + global_heavy_slot_holder: Optional[str] = None + toolchain_known_metadata: Dict[str, Optional[str]] = dataclasses.field(default_factory=dict) + resource_summary: Dict[str, Any] = dataclasses.field(default_factory=dict) + local_log_path: Optional[str] = None + diagnostic_paths: List[str] = dataclasses.field(default_factory=list) + + @classmethod + def from_job( + cls, + job: Any, + output_summary: Optional[Dict[str, Any]], + jobs_dir: Optional[Path] = None, + ) -> FailureRecord: + """Build a failure record from a conductor Job object and its summary.""" + state = getattr(job, "state", "") + exit_code = getattr(job, "exit_code", None) + timed_out = bool(getattr(job, "timed_out", False)) + measurement_invalid = bool(getattr(job, "measurement_invalid", False)) + error = getattr(job, "error", None) or "" + operation = getattr(job, "operation", "") + process_started_at = getattr(job, "process_started_at", None) + global_heavy_slot_wait_seconds = getattr(job, "global_heavy_slot_wait_seconds", None) + + exit_class = classify_exit( + state=state, + exit_code=exit_code, + timed_out=timed_out, + measurement_invalid=measurement_invalid, + ) + failure_class, failure_class_reason = classify_failure( + state=state, + exit_code=exit_code, + timed_out=timed_out, + measurement_invalid=measurement_invalid, + error=error, + operation=operation, + output_summary=output_summary, + global_heavy_slot_wait_seconds=global_heavy_slot_wait_seconds, + process_started_at=process_started_at, + superseded_by_operation=getattr(job, "superseded_by_operation", None), + ) + + resource_summary = build_resource_summary( + ticket=getattr(job, "ticket", ""), + output_summary=output_summary, + jobs_dir=jobs_dir, + ) + + env = getattr(job, "env", {}) or {} + toolchain_known_metadata = { + key: env.get(key) + for key in TOOLCHAIN_KNOWN_ENV_KEYS + if key in env and env.get(key) is not None + } + + return cls( + schema_version=CURRENT_SCHEMA_VERSION, + schema_lineage=SCHEMA_LINEAGE, + recorded_at=time.time(), + ticket=getattr(job, "ticket", ""), + request_key=getattr(job, "request_key", None), + fingerprint=getattr(job, "fingerprint", None), + operation=operation, + operation_label=operation_display_name(operation, getattr(job, "args", {})), + args=filter_args_for_record(getattr(job, "args", {}) or {}), + lanes=list(getattr(job, "lanes", []) or []), + created_at=getattr(job, "created_at", None), + started_at=getattr(job, "started_at", None), + finished_at=getattr(job, "finished_at", None), + queued_at=getattr(job, "created_at", None), + process_started_at=process_started_at, + process_finished_at=getattr(job, "process_finished_at", None), + queue_wait_seconds=compute_queue_wait_seconds(job), + execution_seconds=compute_execution_seconds(job), + terminal_state=state, + exit_code=exit_code, + exit_class=exit_class, + failure_class=failure_class, + failure_class_reason=failure_class_reason, + timed_out=timed_out, + cancel_requested=bool(getattr(job, "cancel_requested", False)), + measurement_invalid=measurement_invalid, + superseded_by_ticket=getattr(job, "superseded_by_ticket", None), + superseded_by_operation=getattr(job, "superseded_by_operation", None), + global_heavy_slot_wait_seconds=global_heavy_slot_wait_seconds, + global_heavy_slot_path=getattr(job, "global_heavy_slot_path", None), + global_heavy_slot_holder=getattr(job, "global_heavy_slot_holder", None), + toolchain_known_metadata=toolchain_known_metadata, + resource_summary=resource_summary, + local_log_path=str(getattr(job, "log_path", None)) if getattr(job, "log_path", None) else None, + diagnostic_paths=[str(p) for p in getattr(job, "diagnostic_paths", []) or []], + ) + + def to_dict(self) -> Dict[str, Any]: + """Serialize to a JSON-ready dict with camelCase top-level keys.""" + data = dataclasses.asdict(self) + return { + _snake_to_camel(key): value + for key, value in data.items() + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> FailureRecord: + """Deserialize a failure record with schema identity checks. + + Unknown top-level fields are ignored. Records with a future schema + version or a foreign lineage are rejected. + """ + if not isinstance(data, dict): + raise FailureDiagnosticsError("record must be a JSON object") + + schema_version = data.get("schemaVersion") + schema_lineage = data.get("schemaLineage") + + if not isinstance(schema_version, int): + raise FailureDiagnosticsError("missing schemaVersion") + + if schema_version > CURRENT_SCHEMA_VERSION: + raise FailureDiagnosticsError( + f"future schema version {schema_version} > {CURRENT_SCHEMA_VERSION}" + ) + + normalized_lineage = (schema_lineage or "").strip() if isinstance(schema_lineage, str) else "" + if normalized_lineage == SCHEMA_LINEAGE: + pass + elif normalized_lineage: + raise FailureDiagnosticsError( + f"foreign schema lineage {schema_lineage!r}" + ) + else: + if schema_version > LEGACY_UNLINEAGED_SCHEMA_VERSION_CEILING: + raise FailureDiagnosticsError( + f"unlineaged schema version {schema_version} is not recognized" + ) + + kwargs: Dict[str, Any] = {} + for field in dataclasses.fields(cls): + camel = _snake_to_camel(field.name) + value = data.get(camel) + if value is None: + value = _field_default(field) + kwargs[field.name] = value + + kwargs["args"] = filter_args_for_record(kwargs.get("args")) + return cls(**kwargs) + + +def _field_default(field: dataclasses.Field) -> Any: + """Return the declared default for a dataclass field, preferring static defaults over factories.""" + if field.default is not dataclasses.MISSING: + return field.default + if field.default_factory is not None: + return field.default_factory() + return None + + +def filter_args_for_record(args: Optional[Dict[str, Any]]) -> Dict[str, Any]: + """Return a copy of operation args with unbounded/redacted keys removed.""" + if not args: + return {} + return {key: value for key, value in args.items() if key not in REDACTED_RECORD_ARG_KEYS} + + +class FailureRecordStore: + """Bounded on-disk store for versioned failure records and summary refs. + + The store is thread-safe and writes records with an atomic temp-file + + os.replace so readers never see a partially-written file. Temporary files + are cleaned up on failure and are never visible to the retention glob + patterns, so an in-flight write cannot be deleted by a concurrent retention + pass. + """ + + def __init__( + self, + jobs_dir: Path, + max_age_seconds: float = DEFAULT_FAILURE_RECORD_RETENTION_SECONDS, + max_records: int = DEFAULT_MAX_FAILURE_RECORDS, + now: Callable[[], float] = time.time, + ) -> None: + self.jobs_dir = Path(jobs_dir) + self.max_age_seconds = max_age_seconds + self.max_records = max_records + self.now = now + self._lock = threading.RLock() + self._written_tickets: set[str] = set() + + def _record_path(self, ticket: str) -> Path: + return self.jobs_dir / f"{ticket}.failure.json" + + def _summary_path(self, ticket: str) -> Path: + return self.jobs_dir / f"{ticket}.summary.json" + + def _ticket_from_record_path(self, path: Path) -> str: + marker = ".failure.json" + if path.name.endswith(marker): + return path.name[: -len(marker)] + return path.stem + + def _ticket_from_summary_path(self, path: Path) -> str: + marker = ".summary.json" + if path.name.endswith(marker): + return path.name[: -len(marker)] + return path.stem + + def _write_text_atomic(self, path: Path, text: str) -> None: + """Write text to a temp file in the same directory and atomically replace ``path``.""" + fd, tmp_path_str = tempfile.mkstemp( + dir=str(self.jobs_dir), + suffix=".tmp", + prefix=".", + ) + tmp_path = Path(tmp_path_str) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(text) + # mkstemp creates with 0o600 on most systems; keep it explicit. + os.chmod(tmp_path, 0o600) + os.replace(tmp_path, str(path)) + except Exception: + with contextlib.suppress(OSError): + os.close(fd) + with contextlib.suppress(OSError): + tmp_path.unlink() + raise + + def write( + self, + record: FailureRecord, + summary: Optional[Dict[str, Any]] = None, + ) -> Path: + """Write a failure record and, if provided, its summary resource. + + Both the record and its summary are written to temp files and then + atomically renamed so readers never see a partially-written file. + If the record write fails after a summary was already replaced, the + summary is rolled back so the store is never left pointing to a record + that does not exist. + """ + with self._lock: + self.jobs_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + + ticket = record.ticket + if ticket in self._written_tickets: + return self._record_path(ticket) + + summary_path = self._summary_path(ticket) + record.resource_summary = dict(record.resource_summary) + summary_written = False + + if summary is not None: + try: + self._write_text_atomic( + summary_path, + json.dumps(summary, indent=2, sort_keys=True), + ) + record.resource_summary["summaryPath"] = str(summary_path) + summary_written = True + except OSError: + # Do not let a summary write failure point the record at a + # summary file that was not persisted. + record.resource_summary.pop("summaryPath", None) + else: + record.resource_summary.pop("summaryPath", None) + + record_path = self._record_path(ticket) + try: + self._write_text_atomic( + record_path, + json.dumps(record.to_dict(), indent=2, sort_keys=True), + ) + except OSError as exc: + if summary_written: + # Roll back the summary so an orphan summary file does not + # reference a record that was not persisted. + with contextlib.suppress(OSError): + summary_path.unlink() + record.resource_summary.pop("summaryPath", None) + raise FailureDiagnosticsError(f"could not write failure record: {exc}") from exc + + self._written_tickets.add(ticket) + self.retention_pass() + return record_path + + def load(self, path: Path) -> Optional[FailureRecord]: + """Load a single failure record, returning None if it is incompatible.""" + with self._lock: + try: + raw = path.read_text(encoding="utf-8", errors="replace") + data = json.loads(raw) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(data, dict): + return None + try: + return FailureRecord.from_dict(data) + except FailureDiagnosticsError: + return None + + def load_all(self) -> List[FailureRecord]: + """Load all compatible failure records in the store.""" + with self._lock: + records: List[FailureRecord] = [] + for path in self._list_record_paths(): + record = self.load(path) + if record is not None: + records.append(record) + return records + + def query_recent( + self, + limit: int = 50, + operation: Optional[str] = None, + failure_class: Optional[str] = None, + since_timestamp: Optional[float] = None, + max_age_seconds: Optional[float] = None, + ) -> List[FailureRecord]: + """Return recent failure records, newest first, with optional filters.""" + with self._lock: + now = self.now() + if since_timestamp is not None: + cutoff = since_timestamp + else: + cutoff = now - (max_age_seconds if max_age_seconds is not None else self.max_age_seconds) + + records = self.load_all() + records = [r for r in records if (r.recorded_at or r.finished_at or 0) >= cutoff] + + if operation: + records = [r for r in records if r.operation == operation] + if failure_class: + records = [r for r in records if r.failure_class == failure_class] + + records.sort(key=lambda r: (r.finished_at or r.recorded_at or 0), reverse=True) + return records[:limit] + + def retention_pass( + self, + max_age_seconds: Optional[float] = None, + max_records: Optional[int] = None, + ) -> None: + """Delete expired or excess failure records and orphaned summaries. + + Only files matching ``*.failure.json`` and ``*.summary.json`` are + considered, so in-flight ``*.tmp`` files cannot be deleted. + """ + with self._lock: + max_age = max_age_seconds if max_age_seconds is not None else self.max_age_seconds + max_count = max_records if max_records is not None else self.max_records + now = self.now() + cutoff = now - max_age + + record_paths = self._list_record_paths() + records: List[Tuple[Path, FailureRecord, float]] = [] + for path in record_paths: + record = self.load(path) + if record is None: + # Unknown/foreign schema: fall back to file mtime for retention. + try: + mtime = path.stat().st_mtime + except OSError: + continue + if mtime < cutoff: + self._delete_record_and_summary(path) + continue + + timestamp = record.recorded_at or record.finished_at or 0 + if timestamp == 0: + try: + timestamp = path.stat().st_mtime + except OSError: + pass + if timestamp < cutoff: + self._delete_record_and_summary(path) + else: + records.append((path, record, timestamp)) + + records.sort(key=lambda item: item[2], reverse=True) + for path, _record, _timestamp in records[max_count:]: + self._delete_record_and_summary(path) + + # Clean orphaned summary files. + retained_tickets = { + self._ticket_from_record_path(p) for p in self._list_record_paths() + } + for summary_path in self._list_summary_paths(): + if self._ticket_from_summary_path(summary_path) not in retained_tickets: + with contextlib.suppress(FileNotFoundError): + summary_path.unlink() + + def _list_record_paths(self) -> List[Path]: + with contextlib.suppress(FileNotFoundError): + return sorted(self.jobs_dir.glob("*.failure.json")) + return [] + + def _list_summary_paths(self) -> List[Path]: + with contextlib.suppress(FileNotFoundError): + return sorted(self.jobs_dir.glob("*.summary.json")) + return [] + + def _delete_record_and_summary(self, record_path: Path) -> None: + ticket = self._ticket_from_record_path(record_path) + with contextlib.suppress(FileNotFoundError): + record_path.unlink() + summary_path = self._summary_path(ticket) + with contextlib.suppress(FileNotFoundError): + summary_path.unlink() + + +def classify_exit( + state: str, + exit_code: Optional[int], + timed_out: bool, + measurement_invalid: bool, +) -> str: + """Return a bounded exit class for a terminal job.""" + if state == "completed" or (exit_code == 0 and not timed_out and not measurement_invalid): + return "completed" + if timed_out or exit_code == 124: + return "timeout" + if state == "canceled" or exit_code == 130: + return "canceled" + if exit_code == 137: + return "killed" + if measurement_invalid or exit_code == 70: + return "infrastructure" + if exit_code is not None and exit_code != 0: + return "nonZero" + return "unknown" + + +def classify_failure( + state: str, + exit_code: Optional[int], + timed_out: bool, + measurement_invalid: bool, + error: str, + operation: str, + output_summary: Optional[Dict[str, Any]], + global_heavy_slot_wait_seconds: Optional[float], + process_started_at: Optional[float], + superseded_by_operation: Optional[str], +) -> Tuple[str, str]: + """Return a bounded failure class and a short reason. + + Classification is conservative: if evidence is ambiguous, the record is + marked ``unknown`` rather than guessed into a more specific bucket. + """ + error = (error or "").strip() + + if state == "completed" and exit_code == 0: + return "none", "completed successfully" + + if state == "canceled" or exit_code == 130: + if superseded_by_operation: + return "cancellation", f"canceled (superseded by {superseded_by_operation})" + return "cancellation", "canceled" + + if timed_out or exit_code == 124: + return "timeout", error or "timed out" + + lifecycle = (output_summary or {}).get("launchLifecycle") or {} + if lifecycle.get("sourceChangedDuringBuild"): + return "sourceMutatedBuild", "source files changed during the build" + + sections = { + section.get("title") + for section in (output_summary or {}).get("sections", []) + if isinstance(section, dict) + } + if "Swift compiler errors" in sections: + return "compilerFailure", "Swift compiler errors" + if "Test failures" in sections: + return "testFailure", "test failures" + + if error and ( + "did not exit after SIGKILL" in error + or "remained alive after SIGKILL" in error + or "canceled job descendants remained alive" in error + ): + return "processCleanupFailure", error + + if measurement_invalid or exit_code == 70 or "daemon runner error" in error: + return "infrastructureOrRPCFailure", error or "infrastructure failure" + + if ( + global_heavy_slot_wait_seconds is not None + and global_heavy_slot_wait_seconds >= 30.0 + and process_started_at is None + and state != "completed" + ): + return ( + "heavyLaneWait", + f"waited {global_heavy_slot_wait_seconds:.1f}s for a global heavy slot before process start", + ) + + if "Style findings" in sections: + return "unknown", "style findings; no recognized failure class" + + return "unknown", "failure evidence not recognized by current classifier" + + +def build_resource_summary( + ticket: str, + output_summary: Optional[Dict[str, Any]], + jobs_dir: Optional[Path] = None, +) -> Dict[str, Any]: + """Build a bounded resource summary reference from an output summary.""" + summary_path: Optional[str] = None + if jobs_dir is not None and ticket: + summary_path = str(jobs_dir / f"{ticket}.summary.json") + + if not isinstance(output_summary, dict): + return {"summaryPath": summary_path} + + sections = output_summary.get("sections") or [] + section_titles = [ + section["title"] + for section in sections + if isinstance(section, dict) and section.get("title") + ] + + return { + "summaryPath": summary_path, + "headline": output_summary.get("headline"), + "summarySectionTitles": section_titles, + "errorCount": output_summary.get("errorCount"), + "warningCount": output_summary.get("warningCount"), + "logLineCount": output_summary.get("logLineCount"), + "truncated": output_summary.get("truncated"), + "launchLifecycle": output_summary.get("launchLifecycle"), + } + + +def operation_display_name(operation: str, args: Dict[str, Any]) -> str: + """Mirror of conductor.operation_display_name for the record label.""" + if operation == "app" and args.get("subcommand") in {"status", "stop", "launch-existing", "relaunch"}: + return f"app {args['subcommand']}" + return operation + + +def compute_queue_wait_seconds(job: Any) -> Optional[float]: + created_at = getattr(job, "created_at", None) + started_at = getattr(job, "started_at", None) + if created_at is not None and started_at is not None: + return max(0.0, started_at - created_at) + return None + + +def compute_execution_seconds(job: Any) -> Optional[float]: + process_started_at = getattr(job, "process_started_at", None) + process_finished_at = getattr(job, "process_finished_at", None) + if process_started_at is not None and process_finished_at is not None: + return max(0.0, process_finished_at - process_started_at) + return None + + +def _snake_to_camel(name: str) -> str: + parts = name.split("_") + return parts[0] + "".join(part.capitalize() for part in parts[1:]) + + +def format_recent_failures(records: List[FailureRecord], json_mode: bool = False) -> str: + """Render a list of recent failure records for human or JSON output.""" + if json_mode: + return json.dumps( + { + "schemaVersion": CURRENT_SCHEMA_VERSION, + "schemaLineage": SCHEMA_LINEAGE, + "privacy": AGGREGATE_JSON_PRIVACY_NOTE, + "count": len(records), + "records": [r.to_dict() for r in records], + }, + indent=2, + sort_keys=True, + ) + + if not records: + return "No recent failure records." + + lines = ["Recent failure records:", ""] + lines.append( + f"{'Ticket':<36} {'Operation':<20} {'State':<10} {'Exit':<6} {'Failure class':<26} {'Reason'}" + ) + for record in records: + operation = record.operation_label or record.operation + if len(operation) > 20: + operation = operation[:17] + "..." + reason = record.failure_class_reason or "" + if len(reason) > 60: + reason = reason[:57] + "..." + state = record.terminal_state or "" + exit_code = str(record.exit_code) if record.exit_code is not None else "" + lines.append( + f"{record.ticket:<36} {operation:<20} {state:<10} {exit_code:<6} {record.failure_class:<26} {reason}" + ) + if record.local_log_path: + lines.append(f" log: {record.local_log_path}") + if record.resource_summary.get("summaryPath"): + lines.append(f" summary: {record.resource_summary['summaryPath']}") + return "\n".join(lines) diff --git a/Scripts/source_layout_guardrails.sh b/Scripts/source_layout_guardrails.sh index 1caafc936..014e0264a 100755 --- a/Scripts/source_layout_guardrails.sh +++ b/Scripts/source_layout_guardrails.sh @@ -330,6 +330,7 @@ print_matches \ # 8. Agent-authored reports and working notes stay local unless explicitly # promoted into the contributor-facing documentation set. allowed_tracked_docs=( + "docs/architecture/failure-diagnostics.md" "docs/architecture/provider-plugins.md" "docs/architecture/settings-persistence.md" "docs/architecture/source-layout.md" diff --git a/Scripts/test_conductor_failure_diagnostics.py b/Scripts/test_conductor_failure_diagnostics.py new file mode 100644 index 000000000..ace2c5999 --- /dev/null +++ b/Scripts/test_conductor_failure_diagnostics.py @@ -0,0 +1,864 @@ +#!/usr/bin/env python3 +"""Focused tests for the local-first structured failure diagnostics surface.""" + +from __future__ import annotations + +import contextlib +import json +import os +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import patch + +SCRIPT_DIR = Path(__file__).resolve().parent +if str(SCRIPT_DIR) not in __import__("sys").path: + __import__("sys").path.insert(0, str(SCRIPT_DIR)) + +import failure_diagnostics # noqa: E402 +import conductor # noqa: E402 + + +class FakeJob: + """Minimal stand-in for a conductor Job object.""" + + def __init__(self, **kwargs: Any) -> None: + defaults = { + "ticket": "ticket-1", + "request_key": None, + "fingerprint": "fp", + "operation": "build", + "args": {}, + "lanes": ["build", "debugArtifact"], + "timeout": None, + "verbose": False, + "env": {}, + "created_at": 1000.0, + "log_path": Path("/tmp/jobs/ticket-1.log"), + "state": "failed", + "started_at": 1001.0, + "finished_at": 1010.0, + "process_started_at": 1002.0, + "process_finished_at": 1009.0, + "process_pid": None, + "process_pgid": None, + "process_start": None, + "tracked_processes": {}, + "process_group_identity_confirmed": False, + "global_heavy_slot_wait_seconds": None, + "global_heavy_slot_path": None, + "global_heavy_slot_holder": None, + "exit_code": 1, + "error": "", + "result_summary": "", + "cancel_requested": False, + "superseded_by_ticket": None, + "superseded_by_operation": None, + "timed_out": False, + "measurement_invalid": False, + "progress_transport": None, + "xctest_progress_sequence": 0, + "xctest_progress_deadline": None, + "xctest_current_test": None, + "xctest_previous_test": None, + "xctest_last_progress_test": None, + "xctest_last_progress_action": None, + "xctest_last_progress_observed_at": None, + "xctest_watchdog_triggered": False, + "xctest_process_finished": False, + "diagnostics": [], + "diagnostic_paths": [], + "output_summary": None, + "tail": [], + } + defaults.update(kwargs) + for key, value in defaults.items(): + setattr(self, key, value) + + +class FailureClassificationTests(unittest.TestCase): + def _classify(self, **kwargs: Any) -> tuple[str, str]: + defaults = { + "state": "failed", + "exit_code": 1, + "timed_out": False, + "measurement_invalid": False, + "error": "", + "operation": "build", + "output_summary": None, + "global_heavy_slot_wait_seconds": None, + "process_started_at": 1000.0, + "superseded_by_operation": None, + } + defaults.update(kwargs) + return failure_diagnostics.classify_failure(**defaults) + + def test_completed_success(self) -> None: + failure_class, reason = self._classify(state="completed", exit_code=0) + self.assertEqual(failure_class, "none") + self.assertEqual(reason, "completed successfully") + + def test_cancellation(self) -> None: + failure_class, reason = self._classify(state="canceled", exit_code=130) + self.assertEqual(failure_class, "cancellation") + self.assertEqual(reason, "canceled") + + def test_superseded_cancellation(self) -> None: + failure_class, reason = self._classify( + state="canceled", exit_code=130, superseded_by_operation="app relaunch" + ) + self.assertEqual(failure_class, "cancellation") + self.assertIn("superseded", reason) + self.assertIn("app relaunch", reason) + + def test_timeout(self) -> None: + failure_class, reason = self._classify(state="failed", exit_code=124, timed_out=True, error="timed out after 300.0s") + self.assertEqual(failure_class, "timeout") + self.assertIn("timed out", reason) + + def test_source_mutated_build(self) -> None: + summary = {"launchLifecycle": {"sourceChangedDuringBuild": True}} + failure_class, reason = self._classify(output_summary=summary) + self.assertEqual(failure_class, "sourceMutatedBuild") + self.assertIn("source files changed", reason) + + def test_compiler_failure(self) -> None: + summary = {"sections": [{"title": "Swift compiler errors"}]} + failure_class, reason = self._classify(output_summary=summary) + self.assertEqual(failure_class, "compilerFailure") + self.assertIn("Swift compiler errors", reason) + + def test_test_failure(self) -> None: + summary = {"sections": [{"title": "Test failures"}]} + failure_class, reason = self._classify(output_summary=summary) + self.assertEqual(failure_class, "testFailure") + self.assertIn("test failures", reason) + + def test_process_cleanup_failure(self) -> None: + failure_class, reason = self._classify( + error="timed out after 300.0s; root process did not exit after SIGKILL escalation" + ) + self.assertEqual(failure_class, "processCleanupFailure") + self.assertIn("SIGKILL", reason) + + def test_infrastructure_failure(self) -> None: + failure_class, reason = self._classify( + exit_code=70, + measurement_invalid=True, + error="XCTest progress stall watchdog invalidated this measurement", + ) + self.assertEqual(failure_class, "infrastructureOrRPCFailure") + + def test_daemon_runner_error_class(self) -> None: + failure_class, reason = self._classify( + exit_code=1, + error="daemon runner error: something broke", + ) + self.assertEqual(failure_class, "infrastructureOrRPCFailure") + self.assertIn("daemon runner error", reason) + + def test_heavy_lane_wait(self) -> None: + failure_class, reason = self._classify( + state="failed", + exit_code=1, + global_heavy_slot_wait_seconds=45.0, + process_started_at=None, + ) + self.assertEqual(failure_class, "heavyLaneWait") + self.assertIn("45.0s", reason) + + def test_unknown_style_findings(self) -> None: + summary = {"sections": [{"title": "Style findings"}]} + failure_class, reason = self._classify(output_summary=summary) + self.assertEqual(failure_class, "unknown") + self.assertIn("style findings", reason) + + def test_unknown_no_evidence(self) -> None: + failure_class, reason = self._classify(state="failed", exit_code=1, error="") + self.assertEqual(failure_class, "unknown") + self.assertIn("not recognized", reason) + + +class ExitClassTests(unittest.TestCase): + def _classify(self, **kwargs: Any) -> str: + defaults = { + "state": "failed", + "exit_code": 1, + "timed_out": False, + "measurement_invalid": False, + } + defaults.update(kwargs) + return failure_diagnostics.classify_exit(**defaults) + + def test_completed(self) -> None: + self.assertEqual(self._classify(state="completed", exit_code=0), "completed") + + def test_timeout(self) -> None: + self.assertEqual(self._classify(state="failed", exit_code=124, timed_out=True), "timeout") + + def test_canceled(self) -> None: + self.assertEqual(self._classify(state="canceled", exit_code=130), "canceled") + + def test_killed(self) -> None: + self.assertEqual(self._classify(state="failed", exit_code=137), "killed") + + def test_infrastructure(self) -> None: + self.assertEqual(self._classify(state="failed", exit_code=70, measurement_invalid=True), "infrastructure") + + def test_non_zero(self) -> None: + self.assertEqual(self._classify(state="failed", exit_code=1), "nonZero") + + +class FailureRecordTests(unittest.TestCase): + def test_record_from_job_filters_env_dump(self) -> None: + job = FakeJob( + env={ + "PATH": "/usr/bin:/bin", + "HOME": "/root", + "DEVELOPER_DIR": "/Applications/Xcode.app/Contents/Developer", + "TOOLCHAINS": "org.swift.512", + "SWIFT_EXEC": "swiftc", + "SIGN_IDENTITY": "Apple Development: ", + } + ) + record = failure_diagnostics.FailureRecord.from_job(job, output_summary=None) + self.assertEqual(record.toolchain_known_metadata.get("DEVELOPER_DIR"), "/Applications/Xcode.app/Contents/Developer") + self.assertEqual(record.toolchain_known_metadata.get("TOOLCHAINS"), "org.swift.512") + self.assertEqual(record.toolchain_known_metadata.get("SWIFT_EXEC"), "swiftc") + self.assertNotIn("PATH", record.toolchain_known_metadata) + self.assertNotIn("HOME", record.toolchain_known_metadata) + self.assertNotIn("SIGN_IDENTITY", record.toolchain_known_metadata) + + def test_record_summary_reference_no_raw_lines(self) -> None: + summary = { + "headline": "failed with exit code 1", + "errorCount": 5, + "warningCount": 2, + "logLineCount": 1000, + "truncated": True, + "sections": [ + { + "title": "Swift compiler errors", + "lines": ["Sources/Foo.swift:10:5: error: cannot find 'x' in scope"], + "truncated": False, + "omittedLineCount": 0, + } + ], + } + with tempfile.TemporaryDirectory() as tmp: + jobs_dir = Path(tmp) + job = FakeJob(log_path=jobs_dir / "ticket-1.log") + record = failure_diagnostics.FailureRecord.from_job(job, summary, jobs_dir=jobs_dir) + + self.assertEqual(record.resource_summary.get("headline"), "failed with exit code 1") + self.assertEqual(record.resource_summary.get("summarySectionTitles"), ["Swift compiler errors"]) + self.assertEqual(record.resource_summary.get("errorCount"), 5) + self.assertEqual(record.resource_summary.get("logLineCount"), 1000) + self.assertEqual(record.resource_summary.get("summaryPath"), str(jobs_dir / "ticket-1.summary.json")) + self.assertNotIn("lines", record.resource_summary) + # Ensure the raw line from the summary did not leak into the record. + record_json = json.dumps(record.to_dict()) + self.assertNotIn("cannot find 'x' in scope", record_json) + + def test_record_serialization_round_trip(self) -> None: + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="test", + terminal_state="failed", + failure_class="testFailure", + ) + data = record.to_dict() + self.assertEqual(data["schemaVersion"], 1) + self.assertEqual(data["schemaLineage"], "repoprompt-ce.failure-record") + self.assertEqual(data["ticket"], "abc") + self.assertEqual(data["failureClass"], "testFailure") + restored = failure_diagnostics.FailureRecord.from_dict(data) + self.assertEqual(restored.ticket, "abc") + self.assertEqual(restored.failure_class, "testFailure") + self.assertEqual(restored.schema_lineage, "repoprompt-ce.failure-record") + + def test_from_dict_missing_fields_use_defaults(self) -> None: + data = { + "schemaVersion": 1, + "schemaLineage": "repoprompt-ce.failure-record", + "ticket": "abc", + "finishedAt": 123.0, + } + record = failure_diagnostics.FailureRecord.from_dict(data) + self.assertEqual(record.ticket, "abc") + self.assertEqual(record.operation, "") + self.assertEqual(record.terminal_state, "") + self.assertEqual(record.failure_class, "") + self.assertEqual(record.recorded_at, None) + self.assertEqual(record.finished_at, 123.0) + self.assertEqual(record.args, {}) + self.assertEqual(record.lanes, []) + self.assertEqual(record.toolchain_known_metadata, {}) + self.assertEqual(record.resource_summary, {}) + self.assertEqual(record.diagnostic_paths, []) + self.assertFalse(record.timed_out) + + def test_from_dict_null_field_uses_default(self) -> None: + data = { + "schemaVersion": 1, + "schemaLineage": "repoprompt-ce.failure-record", + "ticket": None, + "args": None, + "resourceSummary": None, + "lanes": None, + "diagnosticPaths": None, + } + record = failure_diagnostics.FailureRecord.from_dict(data) + self.assertEqual(record.ticket, "") + self.assertEqual(record.args, {}) + self.assertEqual(record.resource_summary, {}) + self.assertEqual(record.lanes, []) + self.assertEqual(record.diagnostic_paths, []) + + def test_aggregate_json_privacy_note(self) -> None: + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="test", + terminal_state="failed", + failure_class="testFailure", + ) + output = failure_diagnostics.format_recent_failures([record], json_mode=True) + payload = json.loads(output) + self.assertEqual(payload["privacy"], failure_diagnostics.AGGREGATE_JSON_PRIVACY_NOTE) + self.assertEqual(payload["count"], 1) + + def test_record_from_job_filters_unbounded_args(self) -> None: + job = FakeJob( + args={ + "product": "RepoPrompt", + "message": "secret user prompt", + "logFile": "/tmp/secret.log", + "filter": "ExampleTests", + } + ) + record = failure_diagnostics.FailureRecord.from_job(job, output_summary=None) + self.assertEqual(record.args, {"product": "RepoPrompt", "filter": "ExampleTests"}) + self.assertNotIn("message", record.args) + self.assertNotIn("logFile", record.args) + + def test_from_dict_filters_unbounded_args(self) -> None: + data = { + "schemaVersion": 1, + "schemaLineage": "repoprompt-ce.failure-record", + "ticket": "abc", + "terminalState": "failed", + "failureClass": "testFailure", + "args": { + "product": "RepoPrompt", + "message": "secret user prompt", + "logFile": "/tmp/secret.log", + }, + } + record = failure_diagnostics.FailureRecord.from_dict(data) + self.assertEqual(record.args, {"product": "RepoPrompt"}) + self.assertNotIn("message", record.args) + self.assertNotIn("logFile", record.args) + + def test_format_recent_failures_json_omits_redacted_args(self) -> None: + data = { + "schemaVersion": 1, + "schemaLineage": "repoprompt-ce.failure-record", + "ticket": "abc", + "terminalState": "failed", + "failureClass": "testFailure", + "args": { + "product": "RepoPrompt", + "message": "secret user prompt", + "logFile": "/tmp/secret.log", + }, + } + record = failure_diagnostics.FailureRecord.from_dict(data) + output = failure_diagnostics.format_recent_failures([record], json_mode=True) + payload = json.loads(output) + args = payload["records"][0]["args"] + self.assertEqual(args, {"product": "RepoPrompt"}) + self.assertNotIn("message", args) + self.assertNotIn("logFile", args) + + +class FailureRecordStoreTests(unittest.TestCase): + def make_store(self, **kwargs: Any) -> failure_diagnostics.FailureRecordStore: + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + defaults = {"max_age_seconds": 3600, "max_records": 10, "now": time.time} + defaults.update(kwargs) + return failure_diagnostics.FailureRecordStore( + Path(tmp.name), **defaults + ) + + def test_write_and_load_round_trip(self) -> None: + store = self.make_store() + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="build", + terminal_state="failed", + failure_class="compilerFailure", + recorded_at=time.time(), + ) + summary = {"headline": "failed with exit code 1", "sections": []} + store.write(record, summary) + + loaded = store.load_all() + self.assertEqual(len(loaded), 1) + self.assertEqual(loaded[0].ticket, "abc") + self.assertEqual(loaded[0].failure_class, "compilerFailure") + self.assertTrue((store._summary_path("abc")).exists()) + + def test_retention_by_age(self) -> None: + now = 1000000.0 + store = self.make_store(now=lambda: now, max_age_seconds=60, max_records=100) + old = failure_diagnostics.FailureRecord( + ticket="old", + operation="build", + terminal_state="failed", + failure_class="unknown", + recorded_at=now - 120, + ) + new = failure_diagnostics.FailureRecord( + ticket="new", + operation="build", + terminal_state="failed", + failure_class="unknown", + recorded_at=now - 10, + ) + store.write(old, None) + store.write(new, None) + + store.retention_pass() + self.assertEqual({r.ticket for r in store.load_all()}, {"new"}) + + def test_retention_by_count(self) -> None: + now = 1000000.0 + store = self.make_store(now=lambda: now, max_age_seconds=3600, max_records=2) + for index in range(5): + record = failure_diagnostics.FailureRecord( + ticket=f"ticket-{index}", + operation="build", + terminal_state="failed", + failure_class="unknown", + recorded_at=now - index, + finished_at=now - index, + ) + store.write(record, None) + + # Each write calls retention_pass, so only the newest two should remain. + loaded = store.load_all() + self.assertEqual({r.ticket for r in loaded}, {"ticket-0", "ticket-1"}) + + def test_query_filters(self) -> None: + now = 1000000.0 + store = self.make_store(now=lambda: now, max_age_seconds=3600, max_records=10) + for ticket, operation, failure_class in [ + ("a", "test", "testFailure"), + ("b", "test", "compilerFailure"), + ("c", "build", "compilerFailure"), + ]: + record = failure_diagnostics.FailureRecord( + ticket=ticket, + operation=operation, + terminal_state="failed", + failure_class=failure_class, + recorded_at=now, + finished_at=now, + ) + store.write(record, None) + + by_operation = store.query_recent(operation="test") + self.assertEqual({r.ticket for r in by_operation}, {"a", "b"}) + + by_class = store.query_recent(failure_class="compilerFailure") + self.assertEqual({r.ticket for r in by_class}, {"b", "c"}) + + by_both = store.query_recent(operation="test", failure_class="compilerFailure") + self.assertEqual({r.ticket for r in by_both}, {"b"}) + + def test_schema_compatibility_skips_future_version(self) -> None: + store = self.make_store() + future = { + "schemaVersion": 99, + "schemaLineage": "repoprompt-ce.failure-record", + "ticket": "future", + "operation": "build", + "terminalState": "failed", + "failureClass": "unknown", + "recordedAt": time.time(), + } + store._record_path("future").write_text(json.dumps(future), encoding="utf-8") + self.assertEqual(store.load_all(), []) + + def test_schema_compatibility_skips_foreign_lineage(self) -> None: + store = self.make_store() + foreign = { + "schemaVersion": 1, + "schemaLineage": "some-other-project.failure-record", + "ticket": "foreign", + "operation": "build", + "terminalState": "failed", + "failureClass": "unknown", + "recordedAt": time.time(), + } + store._record_path("foreign").write_text(json.dumps(foreign), encoding="utf-8") + self.assertEqual(store.load_all(), []) + + def test_orphaned_summary_cleanup(self) -> None: + store = self.make_store() + summary_path = store._summary_path("orphan") + summary_path.write_text(json.dumps({"headline": "x"}), encoding="utf-8") + store.retention_pass() + self.assertFalse(summary_path.exists()) + + def test_atomic_write_leaves_no_temp_files(self) -> None: + store = self.make_store() + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="build", + terminal_state="failed", + failure_class="compilerFailure", + recorded_at=time.time(), + ) + summary = {"headline": "failed", "sections": []} + store.write(record, summary) + + record_path = store._record_path("abc") + summary_path = store._summary_path("abc") + self.assertTrue(record_path.exists()) + self.assertTrue(summary_path.exists()) + # No leftover temp files from the atomic write. + self.assertEqual(list(store.jobs_dir.glob("*.tmp")), []) + + def test_write_cleanup_temp_on_record_failure(self) -> None: + store = self.make_store() + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="build", + terminal_state="failed", + failure_class="compilerFailure", + recorded_at=time.time(), + ) + # Create the final record path as a directory so os.replace cannot overwrite it. + record_path = store._record_path("abc") + record_path.mkdir() + try: + with self.assertRaises(failure_diagnostics.FailureDiagnosticsError): + store.write(record, None) + # Temp files created during the failed write should be removed. + self.assertEqual(list(store.jobs_dir.glob("*.tmp")), []) + finally: + with contextlib.suppress(OSError): + record_path.rmdir() + + def test_concurrent_writes_are_safe(self) -> None: + now = 1000000.0 + store = self.make_store(now=lambda: now, max_age_seconds=3600, max_records=100) + + def write_record(index: int) -> None: + record = failure_diagnostics.FailureRecord( + ticket=f"ticket-{index}", + operation="build", + terminal_state="failed", + failure_class="unknown", + recorded_at=now, + finished_at=now, + ) + store.write(record, None) + + threads = [threading.Thread(target=write_record, args=(i,)) for i in range(20)] + for t in threads: + t.start() + for t in threads: + t.join() + + loaded = store.load_all() + self.assertEqual(len(loaded), 20) + self.assertEqual({r.ticket for r in loaded}, {f"ticket-{i}" for i in range(20)}) + + def test_retention_does_not_delete_temp_files(self) -> None: + store = self.make_store() + # Create a temp file that looks like an in-flight write. + temp_path = store.jobs_dir / ".inflight.tmp" + temp_path.write_text("partial", encoding="utf-8") + + old = failure_diagnostics.FailureRecord( + ticket="old", + operation="build", + terminal_state="failed", + failure_class="unknown", + recorded_at=store.now() - 3600, + ) + store.write(old, None) + store.retention_pass() + + self.assertTrue(temp_path.exists()) + + def test_write_suppresses_duplicate_per_ticket(self) -> None: + store = self.make_store() + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="test", + terminal_state="failed", + failure_class="testFailure", + recorded_at=time.time(), + ) + summary = {"headline": "first"} + record_path = store.write(record, summary) + record_text = record_path.read_text(encoding="utf-8") + summary_text = store._summary_path("abc").read_text(encoding="utf-8") + + record2 = failure_diagnostics.FailureRecord( + ticket="abc", + operation="build", + terminal_state="failed", + failure_class="compilerFailure", + recorded_at=time.time(), + ) + summary2 = {"headline": "second"} + record_path2 = store.write(record2, summary2) + + self.assertEqual(record_path, record_path2) + self.assertEqual(record_path.read_text(encoding="utf-8"), record_text) + self.assertEqual(store._summary_path("abc").read_text(encoding="utf-8"), summary_text) + loaded = store.load_all() + self.assertEqual(len(loaded), 1) + self.assertEqual(loaded[0].operation, "test") + self.assertEqual(loaded[0].failure_class, "testFailure") + + def test_concurrent_duplicate_writes_are_suppressed(self) -> None: + now = 1000000.0 + store = self.make_store(now=lambda: now, max_age_seconds=3600, max_records=100) + + def write_record() -> None: + record = failure_diagnostics.FailureRecord( + ticket="dup", + operation="test", + terminal_state="failed", + failure_class="testFailure", + recorded_at=now, + finished_at=now, + ) + store.write(record, None) + + threads = [threading.Thread(target=write_record) for _ in range(10)] + for t in threads: + t.start() + for t in threads: + t.join() + + loaded = store.load_all() + self.assertEqual(len(loaded), 1) + self.assertEqual(loaded[0].ticket, "dup") + + def test_write_rolls_back_summary_when_record_write_fails(self) -> None: + store = self.make_store() + record = failure_diagnostics.FailureRecord( + ticket="abc", + operation="test", + terminal_state="failed", + failure_class="testFailure", + recorded_at=time.time(), + ) + summary = {"headline": "first"} + + def fake_write_text_atomic(path: Path, text: str) -> None: + if path.name.endswith(".summary.json"): + path.write_text(text, encoding="utf-8") + else: + raise OSError("disk full") + + with patch.object(store, "_write_text_atomic", side_effect=fake_write_text_atomic): + with self.assertRaises(failure_diagnostics.FailureDiagnosticsError): + store.write(record, summary) + + self.assertFalse(store._summary_path("abc").exists()) + self.assertFalse(store._record_path("abc").exists()) + + +class ConductorIntegrationTests(unittest.TestCase): + def test_refresh_output_summary_writes_failure_record(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + jobs_dir = root / "jobs" + jobs_dir.mkdir() + paths = conductor.Paths( + repo_root=root, + repo_hash="test", + state_dir=root, + socket_path=root / "conductor.sock", + pid_path=root / "conductor.pid", + lock_path=root / "conductor.lock", + jobs_dir=jobs_dir, + daemon_log_path=root / "daemon.log", + daemon_meta_path=root / "daemon.json", + running_processes_path=root / "running.json", + ) + log = jobs_dir / "ticket.log" + log.write_text( + "==> Building\nSources/Foo.swift:10:5: error: cannot find 'x' in scope\n", + encoding="utf-8", + ) + state = conductor.DaemonState(paths) + job = conductor.Job( + ticket="ticket", + request_key=None, + fingerprint="fp", + operation="swift-build", + args={}, + lanes=["build"], + timeout=None, + verbose=False, + env={"DEVELOPER_DIR": "/x"}, + created_at=conductor.now(), + log_path=log, + state="failed", + exit_code=1, + finished_at=conductor.now(), + ) + state.jobs["ticket"] = job + state._refresh_output_summary(job) + + record_path = jobs_dir / "ticket.failure.json" + self.assertTrue(record_path.exists()) + data = json.loads(record_path.read_text(encoding="utf-8")) + self.assertEqual(data["ticket"], "ticket") + self.assertEqual(data["operation"], "swift-build") + self.assertEqual(data["failureClass"], "compilerFailure") + self.assertEqual(data["toolchainKnownMetadata"], {"DEVELOPER_DIR": "/x"}) + + summary_path = jobs_dir / "ticket.summary.json" + self.assertTrue(summary_path.exists()) + + def test_handle_recent_failures_query(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + jobs_dir = root / "jobs" + jobs_dir.mkdir() + paths = conductor.Paths( + repo_root=root, + repo_hash="test", + state_dir=root, + socket_path=root / "conductor.sock", + pid_path=root / "conductor.pid", + lock_path=root / "conductor.lock", + jobs_dir=jobs_dir, + daemon_log_path=root / "daemon.log", + daemon_meta_path=root / "daemon.json", + running_processes_path=root / "running.json", + ) + store = failure_diagnostics.FailureRecordStore(jobs_dir, max_age_seconds=3600, max_records=10) + store.write( + failure_diagnostics.FailureRecord( + ticket="recent", + operation="test", + terminal_state="failed", + failure_class="testFailure", + recorded_at=time.time(), + finished_at=time.time(), + ), + None, + ) + + args = {"limit": 10, "operation": None, "failureClass": None, "hours": None} + with contextlib.redirect_stdout(__import__("io").StringIO()) as output: + code = conductor.handle_recent_failures_query(paths, args, json_mode=True) + + self.assertEqual(code, 0) + payload = json.loads(output.getvalue()) + self.assertEqual(len(payload["records"]), 1) + self.assertEqual(payload["records"][0]["ticket"], "recent") + + def test_recent_failures_command_line_filter(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + jobs_dir = root / "jobs" + jobs_dir.mkdir() + paths = conductor.Paths( + repo_root=root, + repo_hash="test", + state_dir=root, + socket_path=root / "conductor.sock", + pid_path=root / "conductor.pid", + lock_path=root / "conductor.lock", + jobs_dir=jobs_dir, + daemon_log_path=root / "daemon.log", + daemon_meta_path=root / "daemon.json", + running_processes_path=root / "running.json", + ) + store = failure_diagnostics.FailureRecordStore(jobs_dir, max_age_seconds=3600, max_records=10) + for op, cls in [("test", "testFailure"), ("build", "compilerFailure")]: + store.write( + failure_diagnostics.FailureRecord( + ticket=f"{op}-1", + operation=op, + terminal_state="failed", + failure_class=cls, + recorded_at=time.time(), + finished_at=time.time(), + ), + None, + ) + + args = {"limit": 10, "operation": "test", "failureClass": None, "hours": None} + with contextlib.redirect_stdout(__import__("io").StringIO()) as output: + conductor.handle_recent_failures_query(paths, args, json_mode=True) + payload = json.loads(output.getvalue()) + self.assertEqual([r["ticket"] for r in payload["records"]], ["test-1"]) + + def test_handle_recent_failures_query_hours_filter(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + jobs_dir = root / "jobs" + jobs_dir.mkdir() + paths = conductor.Paths( + repo_root=root, + repo_hash="test", + state_dir=root, + socket_path=root / "conductor.sock", + pid_path=root / "conductor.pid", + lock_path=root / "conductor.lock", + jobs_dir=jobs_dir, + daemon_log_path=root / "daemon.log", + daemon_meta_path=root / "daemon.json", + running_processes_path=root / "running.json", + ) + now = 1000000.0 + store = failure_diagnostics.FailureRecordStore( + jobs_dir, max_age_seconds=3600, max_records=10, now=lambda: now + ) + store.write( + failure_diagnostics.FailureRecord( + ticket="old", + operation="test", + terminal_state="failed", + failure_class="testFailure", + recorded_at=now - 100.0, + finished_at=now - 100.0, + ), + None, + ) + + with patch("conductor.now", return_value=now): + # 1 hour window should include the 100-second-old record. + args = {"limit": 10, "operation": None, "failureClass": None, "hours": 1.0} + with contextlib.redirect_stdout(__import__("io").StringIO()) as output: + conductor.handle_recent_failures_query(paths, args, json_mode=True) + payload = json.loads(output.getvalue()) + self.assertEqual([r["ticket"] for r in payload["records"]], ["old"]) + + # 0.01 hour (36 seconds) window should exclude the 100-second-old record. + args = {"limit": 10, "operation": None, "failureClass": None, "hours": 0.01} + with contextlib.redirect_stdout(__import__("io").StringIO()) as output: + conductor.handle_recent_failures_query(paths, args, json_mode=True) + payload = json.loads(output.getvalue()) + self.assertEqual(payload["records"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/Scripts/test_conductor_lifecycle.py b/Scripts/test_conductor_lifecycle.py index 5bde3f164..d451ef3b1 100644 --- a/Scripts/test_conductor_lifecycle.py +++ b/Scripts/test_conductor_lifecycle.py @@ -795,7 +795,10 @@ def wait_for_terminal_job(self, state: conductor.DaemonState, ticket: str, timeo while time.monotonic() < deadline: with state.condition: job = state.jobs[ticket] - if job.state in conductor.TERMINAL_STATES: + # A job becomes terminal before the runner's final persistence + # and lane-release work is complete. Wait for that finalizer so + # TemporaryDirectory cleanup cannot race its record write. + if job.state in conductor.TERMINAL_STATES and ticket not in state.active_lanes.values(): return job time.sleep(0.01) with state.condition: @@ -1015,6 +1018,94 @@ def test_live_app_lock_serializes_across_processes_without_gui_launch(self) -> N self.assertEqual([row.split()[0:2] for row in rows], [["start", "a"], ["end", "a"], ["start", "b"], ["end", "b"]]) + def test_terminal_job_status_persists_failure_record_and_summary(self) -> None: + tmp, state = self.make_state() + self.addCleanup(tmp.cleanup) + log = state.paths.jobs_dir / "ticket.log" + log.write_text( + "==> Building\nSources/Foo.swift:10:5: error: cannot find 'x' in scope\n", + encoding="utf-8", + ) + job = self.make_job( + state, + "ticket", + "swift-build", + {"product": "RepoPrompt", "message": "secret prompt", "logFile": "/tmp/secret.log"}, + ["build"], + job_state="failed", + ) + job.exit_code = 1 + job.error = "process exited with status 1" + job.finished_at = conductor.now() + job.log_path = log + state.jobs["ticket"] = job + + payload = state.job_status("ticket", None) + + self.assertIn("outputSummary", payload) + self.assertTrue(job.failure_record_written) + self.assertFalse(job.failure_record_pending) + record_path = state.paths.jobs_dir / "ticket.failure.json" + summary_path = state.paths.jobs_dir / "ticket.summary.json" + self.assertTrue(record_path.exists()) + self.assertTrue(summary_path.exists()) + data = json.loads(record_path.read_text(encoding="utf-8")) + self.assertEqual(data["ticket"], "ticket") + self.assertEqual(data["failureClass"], "compilerFailure") + # Unbounded user args are redacted from the aggregate record. + self.assertNotIn("message", data["args"]) + self.assertNotIn("logFile", data["args"]) + self.assertEqual(data["args"]["product"], "RepoPrompt") + + def test_job_status_is_idempotent_for_failure_record_write(self) -> None: + tmp, state = self.make_state() + self.addCleanup(tmp.cleanup) + log = state.paths.jobs_dir / "ticket.log" + log.write_text("error: cannot find 'x' in scope\n", encoding="utf-8") + job = self.make_job( + state, + "ticket", + "swift-build", + {"product": "RepoPrompt"}, + ["build"], + job_state="failed", + ) + job.exit_code = 1 + job.finished_at = conductor.now() + job.log_path = log + state.jobs["ticket"] = job + + with mock.patch.object(state.failure_store, "write") as write_mock: + state.job_status("ticket", None) + state.job_status("ticket", None) + + write_mock.assert_called_once() + self.assertTrue(job.failure_record_written) + + def test_failure_record_write_error_updates_job_under_condition_lock(self) -> None: + tmp, state = self.make_state() + self.addCleanup(tmp.cleanup) + job = self.make_job(state, "ticket", "swift-build", {"product": "RepoPrompt"}, ["build"], job_state="failed") + job.exit_code = 1 + job.finished_at = conductor.now() + state.jobs[job.ticket] = job + lock_held: list[bool] = [] + append_system_line = state._append_system_line_locked + + def record_lock_state(current_job: conductor.Job, text: str) -> None: + lock_held.append(state.condition._is_owned()) + append_system_line(current_job, text) + + with mock.patch.object(state.failure_store, "write", side_effect=OSError("fixture write failure")), mock.patch.object( + state, "_append_system_line_locked", side_effect=record_lock_state + ): + state._refresh_output_summary(job) + + self.assertEqual(lock_held, [True]) + self.assertFalse(job.failure_record_pending) + self.assertFalse(job.failure_record_written) + self.assertIn("failure record write failed: fixture write failure", job.log_path.read_text(encoding="utf-8")) + class XCTestStallWatchdogTests(LifecycleTestCase): def make_watchdog_job( @@ -2077,6 +2168,7 @@ def test_guarded_failed_relaunch_does_not_inspect_or_stop_before_packaging_succe run_script = scripts / "run.sh" shutil.copy2(SCRIPT_DIR / "run.sh", run_script) shutil.copy2(SCRIPT_DIR / "conductor.py", scripts / "conductor.py") + shutil.copy2(SCRIPT_DIR / "failure_diagnostics.py", scripts / "failure_diagnostics.py") run_script.chmod(0o755) package_script = scripts / "package_app.sh" package_script.write_text("#!/usr/bin/env bash\necho package failed\nexit 23\n", encoding="utf-8") @@ -2128,6 +2220,7 @@ def test_direct_run_packages_before_waiting_for_live_lock_then_activates(self) - run_script = scripts / "run.sh" shutil.copy2(SCRIPT_DIR / "run.sh", run_script) shutil.copy2(SCRIPT_DIR / "conductor.py", scripts / "conductor.py") + shutil.copy2(SCRIPT_DIR / "failure_diagnostics.py", scripts / "failure_diagnostics.py") run_script.chmod(0o755) event_log = root / "events.log" launched_marker = root / "launched" @@ -2247,6 +2340,7 @@ def test_successful_relaunch_uses_debug_executable_for_stop_and_readiness(self) run_script = scripts / "run.sh" shutil.copy2(SCRIPT_DIR / "run.sh", run_script) shutil.copy2(SCRIPT_DIR / "conductor.py", scripts / "conductor.py") + shutil.copy2(SCRIPT_DIR / "failure_diagnostics.py", scripts / "failure_diagnostics.py") run_script.chmod(0o755) event_log = root / "events.log" launched_marker = root / "launched" diff --git a/Tests/RepoPromptTests/MCP/MCPCodeStructureWorktreeTests.swift b/Tests/RepoPromptTests/MCP/MCPCodeStructureWorktreeTests.swift index bb522a68a..9f0052b2c 100644 --- a/Tests/RepoPromptTests/MCP/MCPCodeStructureWorktreeTests.swift +++ b/Tests/RepoPromptTests/MCP/MCPCodeStructureWorktreeTests.swift @@ -639,7 +639,12 @@ final class MCPCodeStructureWorktreeTests: XCTestCase { let tabID = try XCTUnwrap(workspace.activeComposeTabID) var composeTab = try XCTUnwrap(window.workspaceManager.composeTab(with: tabID)) composeTab.selection = StoredSelection(selectedPaths: files.map(\.standardizedFullPath)) - window.workspaceManager.updateComposeTab(composeTab, markDirty: false) + XCTAssertTrue( + window.workspaceManager.updateComposeTabStoredOnly( + composeTab, + inWorkspaceID: workspace.id + ) + ) try await AsyncTestWait.waitUntil("selected code structure candidates are cataloged", timeout: 5) { let resolution = await store.resolveSelectedCodeStructureFiles( atPaths: files.map(\.standardizedFullPath), diff --git a/docs/architecture/failure-diagnostics.md b/docs/architecture/failure-diagnostics.md new file mode 100644 index 000000000..f934ec95c --- /dev/null +++ b/docs/architecture/failure-diagnostics.md @@ -0,0 +1,109 @@ +# Local-first structured failure diagnostics + +`conductor` now writes a small, versioned failure record for every terminal job +and exposes a read-only `conductor diagnostics recent-failures` query. + +## Goals + +- Keep structured, recent-failure context locally on the worktree. +- Never copy raw log text, source content, prompts/transcripts, credentials, or + environment dumps into aggregate records. +- Provide a bounded taxonomy and retention so the surface stays stable and + privacy-safe. + +## On-disk record + +For each terminal job, `conductor` writes a sibling file next to the job log: + +```text +/.failure.json +/.summary.json +``` + +The aggregate record (`.failure.json`) is intentionally small and +referential. It contains: + +- Schema identity (`schemaVersion`, `schemaLineage`). +- Job identity and timing (`ticket`, `operation`, `lanes`, `createdAt`, + `startedAt`, `finishedAt`, `queueWaitSeconds`, `executionSeconds`, ...). +- Terminal state and exit code. +- `exitClass` and `failureClass` with a short `failureClassReason`. +- A bounded `toolchainKnownMetadata` object (selected build toolchain env keys + only: `CC`, `CXX`, `DEVELOPER_DIR`, `SDKROOT`, `SWIFT_EXEC`, `TOOLCHAINS`). +- A `resourceSummary` reference to the persisted summary file, including + `headline`, `summarySectionTitles`, `errorCount`, `warningCount`, + `logLineCount`, `truncated`, and `launchLifecycle`. +- The local `localLogPath` and `diagnosticPaths`. + +The full output summary is written to `.summary.json` and referenced, +not embedded, so the aggregate record stays small. + +## Schema versioning + +The record format is versioned with a `schemaVersion` and a `schemaLineage` +(`"repoprompt-ce.failure-record"`). + +- Backward compatible: a reader may load older records with the same lineage. +- Forward safe: a reader must skip records whose `schemaVersion` is newer than + it understands, and records whose `schemaLineage` is foreign. + +This mirrors the `GlobalSettingsDocument` schema versioning approach in the +Swift codebase. + +## Classification + +`failureClass` is assigned conservatively from current evidence: + +| Class | Evidence | +|-------|----------| +| `none` | `completed` with exit code 0. | +| `cancellation` | Job state `canceled` or exit code 130, including superseded. | +| `timeout` | `timedOut` or exit code 124. | +| `sourceMutatedBuild` | `launchLifecycle.sourceChangedDuringBuild` is true. | +| `compilerFailure` | Summary contains `Swift compiler errors`. | +| `testFailure` | Summary contains `Test failures`. | +| `processCleanupFailure` | Error message contains SIGKILL cleanup failures. | +| `infrastructureOrRPCFailure` | `measurementInvalid`, exit code 70, or `daemon runner error`. | +| `heavyLaneWait` | Waited >= 30s for a global heavy slot before process start. | +| `unknown` | No recognized evidence. | + +`exitClass` is a separate, bounded field derived from `state`, `exitCode`, and +`timedOut`/`measurementInvalid`. + +## Retention + +`FailureRecordStore` enforces the same retention as terminal jobs: + +- Maximum `200` records. +- Maximum age `24` hours. + +The store deletes expired records and orphaned `.summary.json` files during its +retention pass. `conductor` runs the store's retention pass on daemon startup +and after each record write. + +## Query surface + +The query is read-only and does not require the daemon: + +```bash +./conductor diagnostics recent-failures [--limit ] [--operation ] [--failure-class ] [--hours ] [--json] +``` + +Results are newest-first, filtered by optional operation and failure class, and +limited to the requested window. + +## Implementation + +- `Scripts/failure_diagnostics.py` — the record, store, classification, and + rendering logic. +- `Scripts/conductor.py` — writes records in `_refresh_output_summary` and + serves the `diagnostics recent-failures` command. +- `Scripts/test_conductor_failure_diagnostics.py` — focused unit tests for + classification, schema, retention, query, and redaction. + +## Privacy notes + +- `toolchainKnownMetadata` is filtered to a fixed allow-list of build toolchain + identifiers; no broad `PATH`, `HOME`, `USER`, or signing secrets are captured. +- The aggregate record never contains the raw job log, full source lines, or + prompt/transcript content. Those remain in the local log and summary files.