Skip to content

Add structured local failure history for conductor jobs #525

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
morluto wants to merge 6 commits into
base: main
Choose a base branch
from
+1,878 −9
Open

Add structured local failure history for conductor jobs #525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
2288af9
Add local-first structured failure diagnostics for conductor jobs
morluto Jul 12, 2026
a38664d
Make conductor failure persistence lifecycle-safe
morluto Jul 12, 2026
a281a23
Merge origin/main into agent/423-local-diagnostics
morluto Jul 12, 2026
3e4fce6
Wait for conductor finalizer in lifecycle tests
morluto Jul 12, 2026
134d52b
Synchronize conductor failure record error logging
morluto Jul 12, 2026
e0a5d17
test: isolate selected code structure scope fixture
morluto Jul 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .agents/skills/rpce-contribution-check/scripts/preflight.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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/)'
Expand Down
1 change: 1 addition & 0 deletions Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
88 changes: 82 additions & 6 deletions Scripts/conductor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}
Expand Down Expand Up @@ -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 <path>]
./conductor diagnostics build-cache [--limit <n>]
./conductor diagnostics recent-failures [--limit <n>] [--operation <op>] [--failure-class <class>] [--hours <n>] [--json]
./conductor release preflight|artifact|package|local-install

Foundation validation operation:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -2167,19 +2176,20 @@ 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]
self._write_running_processes_locked()
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]:
Expand Down Expand Up @@ -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
Comment on lines +2586 to +2590

@sentry sentry Bot Jul 12, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: The method _append_system_line_locked is called without holding the required self.condition lock in an exception handler, creating a race condition when modifying job.tail.
Severity: MEDIUM

Suggested Fix

The call to _append_system_line_locked within the except block of _refresh_output_summary should be moved inside the with self.condition: block that is already present for setting current.failure_record_pending = False. This will ensure the lock is held during the modification of the shared job.tail deque.

Prompt for AI Agent 
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: Scripts/conductor.py#L2586-L2589

Potential issue: In the `_refresh_output_summary` method, if writing a failure record to
the `failure_store` throws an exception, the code enters an `except` block. Inside this
block, it calls `_append_system_line_locked` to log the error. However, this call is
made without acquiring the `self.condition` lock. The `_append_system_line_locked`
method modifies the shared `job.tail` deque, and doing so without synchronization in a
multi-threaded context can lead to race conditions and data corruption. The method's
name explicitly suggests a lock is required.

Did we get this right? 👍 / 👎 to inform future reviews.

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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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":
Expand All @@ -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"])
Expand Down
Loading
Loading