From 9349a5ec3882698760541c682eefcc35d711f580 Mon Sep 17 00:00:00 2001 From: shiva Date: Tue, 25 Aug 2026 21:07:29 +0530 Subject: [PATCH 1/2] fix: persist stdout/stderr log paths on job failure and orphan reconcile run_in_process's Err/timeout branches and reconcile_orphaned_running both called fail/marked the row failed without ever writing stdout_log_path, stderr_log_path, stdout_bytes, or stderr_bytes -- even though the log file was already sitting on disk. Every failed row pointed nowhere, and cleanup() (which deletes finished jobs' log files by reading those same columns) could never find them, leaking log files forever. Adds Queue::record_output to persist those columns independently of state, calls it from both failure paths in worker.rs before record_fail, and has reconcile_orphaned_running populate them directly from the already-known {id}.stdout/{id}.stderr paths under log_dir(). Agentflare-Agent: claude-code Agentflare-Branch: task/189-daemon-restart-orphan-recovery-verify-ru Agentflare-Item: 189 --- crates/agentflare-jobs/src/queue.rs | 133 ++++++++++++++++++++++++++- crates/agentflare-jobs/src/worker.rs | 17 ++++ 2 files changed, 148 insertions(+), 2 deletions(-) diff --git a/crates/agentflare-jobs/src/queue.rs b/crates/agentflare-jobs/src/queue.rs index 91fb6fba..0ba1c8bb 100644 --- a/crates/agentflare-jobs/src/queue.rs +++ b/crates/agentflare-jobs/src/queue.rs @@ -201,6 +201,40 @@ impl Queue { Ok(()) } + /// Persists a job's log paths/byte counts independently of `state` — + /// for `run_in_process`'s failure/timeout paths, which already wrote a + /// real stdout log before the executor errored out or the timeout fired + /// and go on to call `fail` (not `complete`, since `fail` alone carries + /// the retry-vs-terminal logic), so without this the log a job actually + /// produced was silently discarded and `stdout_log_path` stayed NULL + /// forever even though the file exists on disk. + pub fn record_output( + &self, + id: &str, + stdout_path: &Path, + stderr_path: &Path, + stdout_bytes: u64, + stderr_bytes: u64, + ) -> Result<(), Error> { + let conn = self.conn.lock(); + conn.execute( + "UPDATE agent_jobs + SET stdout_log_path = ?1, + stderr_log_path = ?2, + stdout_bytes = ?3, + stderr_bytes = ?4 + WHERE id = ?5", + params![ + stdout_path.to_string_lossy().as_ref(), + stderr_path.to_string_lossy().as_ref(), + stdout_bytes as i64, + stderr_bytes as i64, + id + ], + )?; + Ok(()) + } + /// Returns `true` when this failure was terminal (retries exhausted, or /// `fatal` short-circuited them, row left `state = 'failed'`), `false` /// when it went back to `queued` for another attempt — callers that need @@ -289,9 +323,32 @@ impl Queue { .collect::, _>>()? }; for (id, _) in &rows { + // Both the subprocess (`Supervisor`) and in-process + // (`run_in_process`) paths name their log files `{id}.stdout`/ + // `{id}.stderr` under `log_dir()` and may have already written + // real output before the daemon died mid-job — without this, + // that log survives on disk but the row's `stdout_log_path` + // stays NULL forever, so it's unreachable via `JobInfo`/the + // dashboard even though the file is right there. + let stdout_path = self.log_dir.join(format!("{id}.stdout")); + let stderr_path = self.log_dir.join(format!("{id}.stderr")); + let stdout_bytes = std::fs::metadata(&stdout_path).map(|m| m.len()).unwrap_or(0); + let stderr_bytes = std::fs::metadata(&stderr_path).map(|m| m.len()).unwrap_or(0); conn.execute( - "UPDATE agent_jobs SET state = 'failed', error = ?1, finished_at = ?2 WHERE id = ?3", - params![ORPHAN_ERROR, now, id], + "UPDATE agent_jobs + SET state = 'failed', error = ?1, finished_at = ?2, + stdout_log_path = ?3, stderr_log_path = ?4, + stdout_bytes = ?5, stderr_bytes = ?6 + WHERE id = ?7", + params![ + ORPHAN_ERROR, + now, + stdout_path.to_string_lossy().as_ref(), + stderr_path.to_string_lossy().as_ref(), + stdout_bytes as i64, + stderr_bytes as i64, + id + ], )?; } Ok(rows @@ -484,6 +541,44 @@ mod tests { (queue, dir) } + // A `run_in_process` failure/timeout still writes a real stdout log + // before calling `fail` (which alone doesn't touch the log columns) -- + // `record_output` is how that log's path/size gets persisted so it + // isn't orphaned on disk with a `NULL` `stdout_log_path`, regardless of + // whether the row ends up 'queued' (retried) or 'failed' (terminal). + #[test] + fn record_output_persists_log_paths_independently_of_fail() { + let (queue, _dir) = test_queue(); + let job = AgentJob::new("true").max_retries(0); + let info = queue.enqueue(&job).unwrap(); + queue.dequeue().unwrap(); + + queue + .record_output( + &info.id, + Path::new("/tmp/job.stdout"), + Path::new("/tmp/job.stderr"), + 42, + 7, + ) + .unwrap(); + queue.fail(&info.id, "boom", None, true).unwrap(); + + let conn = queue.conn.lock(); + let (stdout_path, stderr_path, stdout_bytes, stderr_bytes): (String, String, i64, i64) = + conn.query_row( + "SELECT stdout_log_path, stderr_log_path, stdout_bytes, stderr_bytes + FROM agent_jobs WHERE id = ?1", + params![info.id], + |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)), + ) + .unwrap(); + assert_eq!(stdout_path, "/tmp/job.stdout"); + assert_eq!(stderr_path, "/tmp/job.stderr"); + assert_eq!(stdout_bytes, 42); + assert_eq!(stderr_bytes, 7); + } + #[test] fn fail_without_retry_delay_requeues_immediately() { let (queue, _dir) = test_queue(); @@ -549,6 +644,40 @@ mod tests { assert_eq!(stored.error.as_deref(), Some("orphaned by daemon restart")); } + // A job killed mid-run by a daemon restart may already have written real + // progress to its `{id}.stdout` file before the process died -- without + // persisting that path/size here, the log survives on disk but is + // unreachable (no DB row points at it) and `cleanup` can never find it + // to delete it either, so it leaks forever. Reproduced live: 599 failed + // rows with an empty `stdout_log_path` alongside real `.stdout` files on + // disk, all originating from this exact path. + #[test] + fn reconcile_orphaned_running_persists_the_stdout_log_already_written_to_disk() { + let (queue, _dir) = test_queue(); + let job = AgentJob::new("agentflare-work") + .args(["item-1", "claude-code"]) + .in_process(); + let info = queue.enqueue(&job).unwrap(); + queue.dequeue().unwrap(); + + std::fs::create_dir_all(queue.log_dir()).unwrap(); + let stdout_path = queue.log_dir().join(format!("{}.stdout", info.id)); + std::fs::write(&stdout_path, "claimed: item-1\nworktree: ...\n").unwrap(); + + queue.reconcile_orphaned_running().unwrap(); + + let conn = queue.conn.lock(); + let (path, bytes): (String, i64) = conn + .query_row( + "SELECT stdout_log_path, stdout_bytes FROM agent_jobs WHERE id = ?1", + params![info.id], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap(); + assert_eq!(path, stdout_path.to_string_lossy()); + assert_eq!(bytes, 30); + } + #[test] fn reconcile_orphaned_running_leaves_terminal_and_queued_jobs_untouched() { let (queue, _dir) = test_queue(); diff --git a/crates/agentflare-jobs/src/worker.rs b/crates/agentflare-jobs/src/worker.rs index 9d2976dd..859338a4 100644 --- a/crates/agentflare-jobs/src/worker.rs +++ b/crates/agentflare-jobs/src/worker.rs @@ -2,6 +2,7 @@ use crate::executor::{InProcessExecutor, JobFailure}; use crate::queue::Queue; use crate::supervisor::Supervisor; use crate::types::JobOutput; +use std::path::Path; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::JoinHandle; @@ -242,6 +243,7 @@ fn run_in_process( } } Ok(Err(failure)) => { + record_output_best_effort(queue, id, &stdout_path, &stderr_path); record_fail(&failure.message, failure.retry_after_secs, failure.fatal); } Err(_) => { @@ -252,7 +254,22 @@ fn run_in_process( what this timeout measures)", job.timeout_secs ); + record_output_best_effort(queue, id, &stdout_path, &stderr_path); record_fail(&msg, None, false); } } } + +/// The log file at `stdout_path` was already written by the executor +/// before it failed or timed out (see `run_in_process`'s call sites) — this +/// persists that path/size the same way the success branch does, so a +/// failed job's log stays reachable via `stdout_log_path` instead of +/// existing on disk with no DB row pointing at it. Best-effort: a failure +/// here just means the job's own failure/timeout error still gets recorded +/// via `record_fail` right after, unaffected by this. +fn record_output_best_effort(queue: &Queue, id: &str, stdout_path: &Path, stderr_path: &Path) { + let stdout_total_bytes = std::fs::metadata(stdout_path).map(|m| m.len()).unwrap_or(0); + if let Err(e) = queue.record_output(id, stdout_path, stderr_path, stdout_total_bytes, 0) { + eprintln!("agentflare-jobs: failed to record output for {id}: {e}"); + } +} From 558134e9a3ae9352387cf6d09e4912e9180d48d2 Mon Sep 17 00:00:00 2001 From: shiva Date: Wed, 26 Aug 2026 14:06:18 +0530 Subject: [PATCH 2/2] fmt: cargo fmt cleanup in queue.rs Agentflare-Agent: claude-code Agentflare-Branch: task/189-daemon-restart-orphan-recovery-verify-ru Agentflare-Item: 189 Agentflare-Session: c5a4ab79-7ae7-4faf-b526-71ee9f9b5e37 --- crates/agentflare-jobs/src/queue.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/crates/agentflare-jobs/src/queue.rs b/crates/agentflare-jobs/src/queue.rs index 0ba1c8bb..ce421831 100644 --- a/crates/agentflare-jobs/src/queue.rs +++ b/crates/agentflare-jobs/src/queue.rs @@ -332,8 +332,12 @@ impl Queue { // dashboard even though the file is right there. let stdout_path = self.log_dir.join(format!("{id}.stdout")); let stderr_path = self.log_dir.join(format!("{id}.stderr")); - let stdout_bytes = std::fs::metadata(&stdout_path).map(|m| m.len()).unwrap_or(0); - let stderr_bytes = std::fs::metadata(&stderr_path).map(|m| m.len()).unwrap_or(0); + let stdout_bytes = std::fs::metadata(&stdout_path) + .map(|m| m.len()) + .unwrap_or(0); + let stderr_bytes = std::fs::metadata(&stderr_path) + .map(|m| m.len()) + .unwrap_or(0); conn.execute( "UPDATE agent_jobs SET state = 'failed', error = ?1, finished_at = ?2,