diff --git a/apps/decodex/src/agent.rs b/apps/decodex/src/agent.rs index 7bdef106c..f71f8f047 100644 --- a/apps/decodex/src/agent.rs +++ b/apps/decodex/src/agent.rs @@ -6,15 +6,15 @@ mod tracker_tool_bridge; #[cfg(test)] pub(crate) use self::app_server::AppServerCapabilityPreflightReport; #[cfg(test)] pub(crate) use self::app_server::MODEL_EXECUTION_IDLE_TIMEOUT; +#[cfg(not(test))] pub(crate) use self::app_server::archive_app_server_thread_after_success; #[cfg(test)] pub(crate) use self::tracker_tool_bridge::DynamicToolHandler; pub(crate) use self::{ app_server::{ ACTIVE_RUN_IDLE_TIMEOUT, AppServerCapabilityPreflightFailure, AppServerDynamicToolFailure, AppServerPhaseGoalFailure, AppServerRunRequest, AppServerRunResult, AppServerThreadArchiveRequest, AppServerTurnFailure, PhaseGoalController, PhaseGoalKind, - PhaseGoalSpec, PhaseGoalTransition, TurnContinuationGuard, - archive_app_server_thread_after_success, execute_app_server_run, probe_app_server, - protocol_activity_idle_timeout, + PhaseGoalSpec, PhaseGoalTransition, TurnContinuationGuard, execute_app_server_run, + probe_app_server, protocol_activity_idle_timeout, }, codex_accounts::{CodexAccountPool, CodexAccountProvider}, decodex_tool_bridge::{DecodexRunContext, DecodexToolBridge}, diff --git a/apps/decodex/src/orchestrator/daemon.rs b/apps/decodex/src/orchestrator/daemon.rs index 1a4a51e76..d746b2468 100644 --- a/apps/decodex/src/orchestrator/daemon.rs +++ b/apps/decodex/src/orchestrator/daemon.rs @@ -158,6 +158,12 @@ where ); } + reconcile_terminal_thread_archive_backlog_best_effort( + context.project, + context.workflow, + state_store, + ); + while context .workflow .frontmatter() diff --git a/apps/decodex/src/orchestrator/execution.rs b/apps/decodex/src/orchestrator/execution.rs index 0eb3295af..0a7c709fb 100644 --- a/apps/decodex/src/orchestrator/execution.rs +++ b/apps/decodex/src/orchestrator/execution.rs @@ -374,13 +374,25 @@ impl PhaseGoalController for RepoGatePhaseGoalController<'_> { } } +#[derive(Clone)] struct ThreadArchiveCandidate { + issue_id: String, + issue_identifier: String, run_id: String, attempt_number: i64, thread_id: String, sequence_number: i64, } +struct ThreadArchiveCandidateSource<'a> { + run_id: &'a str, + issue_id: &'a str, + issue_identifier: &'a str, + attempt_number: i64, + thread_id: &'a str, + sequence_number: Option, +} + struct ZeroEvidenceAppServerStartFailureContext { protocol_event_count: i64, private_event_count: usize, @@ -496,6 +508,12 @@ where project.service_id(), &issue_run.issue.id, )?; + + reconcile_terminal_thread_archive_backlog_best_effort( + project, + workflow, + state_store, + ); } tracing::info!( @@ -1176,27 +1194,81 @@ fn archive_completed_issue_threads_best_effort( transport: &str, run_result: &AppServerRunResult, ) { + let current = ThreadArchiveCandidate { + issue_id: issue_run.issue.id.clone(), + issue_identifier: issue_run.issue.identifier.clone(), + run_id: issue_run.run_id.clone(), + attempt_number: issue_run.attempt_number, + thread_id: run_result.thread_id.clone(), + sequence_number: run_result.event_count.saturating_add(1), + }; + + archive_issue_threads_best_effort( + project, + state_store, + &issue_run.issue.id, + &issue_run.issue.identifier, + process_env, + transport, + Some(current), + ); +} + +fn reconcile_terminal_thread_archive_backlog_best_effort( + project: &ServiceConfig, + workflow: &WorkflowDocument, + state_store: &StateStore, +) { + let process_env = AppServerProcessEnv::default(); + let transport = workflow.frontmatter().agent().transport(); + let candidates = match terminal_thread_archive_backlog_candidates(state_store, project.service_id()) + { + Ok(candidates) => candidates, + Err(error) => { + tracing::warn!( + ?error, + project_id = project.service_id(), + "Failed to list terminal thread archive backlog; skipping this archive reconciliation pass." + ); + + return; + }, + }; + + for candidate in candidates { + archive_completed_issue_thread_best_effort( + project, + state_store, + &process_env, + transport, + &candidate, + ); + } +} + +fn archive_issue_threads_best_effort( + project: &ServiceConfig, + state_store: &StateStore, + issue_id: &str, + issue_identifier: &str, + process_env: &AppServerProcessEnv, + transport: &str, + current: Option, +) { + let fallback_candidate = current.clone(); let candidates = - match completed_issue_thread_archive_candidates(state_store, issue_run, run_result) { + match issue_thread_archive_candidates(state_store, issue_id, issue_identifier, current) { Ok(candidates) => candidates, Err(error) => { tracing::warn!( ?error, project_id = project.service_id(), - issue_id = issue_run.issue.id, - issue = issue_run.issue.identifier, - run_id = issue_run.run_id, - attempt = issue_run.attempt_number, - thread_id = %run_result.thread_id, + issue_id, + issue = issue_identifier, "Failed to list completed issue threads for archive; archiving current thread only." ); - vec![ThreadArchiveCandidate { - run_id: issue_run.run_id.clone(), - attempt_number: issue_run.attempt_number, - thread_id: run_result.thread_id.clone(), - sequence_number: run_result.event_count.saturating_add(1), - }] + fallback_candidate.into_iter().collect() }, }; @@ -1204,7 +1276,6 @@ fn archive_completed_issue_threads_best_effort( archive_completed_issue_thread_best_effort( project, state_store, - issue_run, process_env, transport, &candidate, @@ -1212,27 +1283,86 @@ fn archive_completed_issue_threads_best_effort( } } +#[cfg(test)] fn completed_issue_thread_archive_candidates( state_store: &StateStore, issue_run: &IssueRunPlan, run_result: &AppServerRunResult, +) -> Result> { + issue_thread_archive_candidates( + state_store, + &issue_run.issue.id, + &issue_run.issue.identifier, + Some(ThreadArchiveCandidate { + issue_id: issue_run.issue.id.clone(), + issue_identifier: issue_run.issue.identifier.clone(), + run_id: issue_run.run_id.clone(), + attempt_number: issue_run.attempt_number, + thread_id: run_result.thread_id.clone(), + sequence_number: run_result.event_count.saturating_add(1), + }), + ) +} + +fn issue_thread_archive_candidates( + state_store: &StateStore, + issue_id: &str, + issue_identifier: &str, + current: Option, ) -> Result> { let mut seen_thread_ids = HashSet::new(); let mut candidates = Vec::new(); - push_thread_archive_candidate( - state_store, - &mut candidates, - &mut seen_thread_ids, - &issue_run.run_id, - issue_run.attempt_number, - &run_result.thread_id, - )?; + if let Some(current) = current { + push_thread_archive_candidate( + state_store, + &mut candidates, + &mut seen_thread_ids, + ThreadArchiveCandidateSource { + run_id: ¤t.run_id, + issue_id: ¤t.issue_id, + issue_identifier: ¤t.issue_identifier, + attempt_number: current.attempt_number, + thread_id: ¤t.thread_id, + sequence_number: Some(current.sequence_number), + }, + )?; + } - for attempt in state_store.list_run_attempts_for_issue(&issue_run.issue.id)? { - if attempt.run_id() == issue_run.run_id - || !completed_issue_archive_attempt_status(attempt.status()) - { + for attempt in state_store.list_run_attempts_for_issue(issue_id)? { + if !completed_issue_archive_attempt_status(attempt.status()) { + continue; + } + + if let Some(thread_id) = attempt.thread_id() { + push_thread_archive_candidate( + state_store, + &mut candidates, + &mut seen_thread_ids, + ThreadArchiveCandidateSource { + run_id: attempt.run_id(), + issue_id: attempt.issue_id(), + issue_identifier, + attempt_number: attempt.attempt_number(), + thread_id, + sequence_number: None, + }, + )?; + } + } + + Ok(candidates) +} + +fn terminal_thread_archive_backlog_candidates( + state_store: &StateStore, + project_id: &str, +) -> Result> { + let mut seen_thread_ids = HashSet::new(); + let mut candidates = Vec::new(); + + for attempt in state_store.list_run_attempts_for_project(project_id)? { + if !completed_issue_archive_attempt_status(attempt.status()) { continue; } @@ -1241,9 +1371,14 @@ fn completed_issue_thread_archive_candidates( state_store, &mut candidates, &mut seen_thread_ids, - attempt.run_id(), - attempt.attempt_number(), - thread_id, + ThreadArchiveCandidateSource { + run_id: attempt.run_id(), + issue_id: attempt.issue_id(), + issue_identifier: attempt.issue_id(), + attempt_number: attempt.attempt_number(), + thread_id, + sequence_number: None, + }, )?; } } @@ -1252,28 +1387,33 @@ fn completed_issue_thread_archive_candidates( } fn completed_issue_archive_attempt_status(status: &str) -> bool { - matches!(status, "succeeded" | "failed" | "interrupted" | TERMINAL_GUARDED_RUN_STATUS) + matches!( + status, + "succeeded" | "failed" | "interrupted" | "terminated" | TERMINAL_GUARDED_RUN_STATUS + ) } fn push_thread_archive_candidate( state_store: &StateStore, candidates: &mut Vec, seen_thread_ids: &mut HashSet, - run_id: &str, - attempt_number: i64, - thread_id: &str, + source: ThreadArchiveCandidateSource<'_>, ) -> Result<()> { - if !seen_thread_ids.insert(thread_id.to_owned()) - || state_store.run_has_protocol_event(run_id, "thread/archive")? + if !seen_thread_ids.insert(source.thread_id.to_owned()) + || state_store.run_has_protocol_event(source.run_id, "thread/archive")? { return Ok(()); } candidates.push(ThreadArchiveCandidate { - run_id: run_id.to_owned(), - attempt_number, - thread_id: thread_id.to_owned(), - sequence_number: state_store.event_count(run_id)?.saturating_add(1), + issue_id: source.issue_id.to_owned(), + issue_identifier: source.issue_identifier.to_owned(), + run_id: source.run_id.to_owned(), + attempt_number: source.attempt_number, + thread_id: source.thread_id.to_owned(), + sequence_number: source + .sequence_number + .unwrap_or(state_store.event_count(source.run_id)?.saturating_add(1)), }); Ok(()) @@ -1282,26 +1422,39 @@ fn push_thread_archive_candidate( fn archive_completed_issue_thread_best_effort( project: &ServiceConfig, state_store: &StateStore, - issue_run: &IssueRunPlan, process_env: &AppServerProcessEnv, transport: &str, candidate: &ThreadArchiveCandidate, ) { let archive_request = AppServerThreadArchiveRequest { run_id: &candidate.run_id, - issue_id: &issue_run.issue.id, + issue_id: &candidate.issue_id, attempt_number: candidate.attempt_number, listen: transport, process_env, thread_id: &candidate.thread_id, sequence_number: candidate.sequence_number, }; + #[cfg(not(test))] + let archive_result = agent::archive_app_server_thread_after_success(&archive_request, state_store); + #[cfg(test)] + let archive_result = state_store.append_event( + archive_request.run_id, + archive_request.sequence_number, + "thread/archive", + &serde_json::json!({ + "threadId": archive_request.thread_id, + "issueId": archive_request.issue_id, + "attemptNumber": archive_request.attempt_number, + }) + .to_string(), + ); - match agent::archive_app_server_thread_after_success(&archive_request, state_store) { + match archive_result { Ok(()) => tracing::info!( project_id = project.service_id(), - issue_id = issue_run.issue.id, - issue = issue_run.issue.identifier, + issue_id = candidate.issue_id, + issue = candidate.issue_identifier, run_id = candidate.run_id, attempt = candidate.attempt_number, thread_id = %candidate.thread_id, @@ -1310,8 +1463,8 @@ fn archive_completed_issue_thread_best_effort( Err(error) => tracing::warn!( ?error, project_id = project.service_id(), - issue_id = issue_run.issue.id, - issue = issue_run.issue.identifier, + issue_id = candidate.issue_id, + issue = candidate.issue_identifier, run_id = candidate.run_id, attempt = candidate.attempt_number, thread_id = %candidate.thread_id, diff --git a/apps/decodex/src/orchestrator/run_cycle.rs b/apps/decodex/src/orchestrator/run_cycle.rs index 0138c1222..b1d00a87f 100644 --- a/apps/decodex/src/orchestrator/run_cycle.rs +++ b/apps/decodex/src/orchestrator/run_cycle.rs @@ -154,6 +154,10 @@ where excluded_issue_ids, )? else { + if !dry_run { + reconcile_terminal_thread_archive_backlog_best_effort(project, workflow, state_store); + } + return Ok(None); }; @@ -2463,8 +2467,7 @@ where github_token_env_var: Some(project.github().token_env_var().to_owned()), github_command_path: project.github().command_path().map(Path::to_path_buf), }; - - if let Some(retained_summary) = drain_non_github_review_retained_tail_with_inspector( + let summary = if let Some(retained_summary) = drain_non_github_review_retained_tail_with_inspector( tracker, project, workflow, @@ -2479,8 +2482,12 @@ where source_summary, ), )? { - return Ok(Some(retained_summary)); - } + retained_summary + } else { + summary + }; + + reconcile_terminal_thread_archive_backlog_best_effort(project, workflow, state_store); Ok(Some(summary)) } diff --git a/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs b/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs index 2587f4a8e..0dee7abb0 100644 --- a/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs +++ b/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs @@ -75,3 +75,71 @@ fn completed_issue_thread_archive_candidates_include_prior_terminal_attempts() { assert_eq!(candidates[0].sequence_number, 2); assert_eq!(candidates[1].sequence_number, 2); } + +#[test] +fn terminal_thread_archive_backlog_candidates_scan_project_terminal_runs() { + let state_store = StateStore::open_in_memory().expect("state store should open"); + + for (project, issue_id, run_id, status, thread_id) in [ + ("decodex", "issue-succeeded", "run-succeeded", "succeeded", "thread-succeeded"), + ("decodex", "issue-failed", "run-failed", "failed", "thread-failed"), + ("decodex", "issue-terminated", "run-terminated", "terminated", "thread-terminated"), + ("decodex", "issue-running", "run-running", "running", "thread-running"), + ("other", "issue-other", "run-other", "succeeded", "thread-other"), + ("decodex", "issue-archived", "run-archived", "succeeded", "thread-archived"), + ] { + state_store + .try_acquire_lease(project, issue_id, run_id, "In Progress") + .expect("lease should record project ownership"); + state_store + .record_run_attempt(run_id, issue_id, 1, status) + .expect("attempt should record"); + state_store.update_run_thread(run_id, thread_id).expect("thread should attach"); + } + + state_store + .append_event("run-archived", 1, "thread/archive", "{}") + .expect("archive event should record"); + + let candidates = super::terminal_thread_archive_backlog_candidates(&state_store, "decodex") + .expect("backlog candidates should load"); + let mut candidate_threads = candidates + .iter() + .map(|candidate| candidate.thread_id.as_str()) + .collect::>(); + + candidate_threads.sort_unstable(); + + assert_eq!( + candidate_threads, + vec!["thread-failed", "thread-succeeded", "thread-terminated"] + ); +} + +#[test] +fn terminal_thread_archive_reconciler_records_backlog_archive_events() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let issue_id = "issue-succeeded"; + let run_id = "run-succeeded"; + + state_store + .try_acquire_lease(config.service_id(), issue_id, run_id, "In Progress") + .expect("lease should record project ownership"); + state_store + .record_run_attempt(run_id, issue_id, 1, "succeeded") + .expect("attempt should record"); + state_store.update_run_thread(run_id, "thread-succeeded").expect("thread should attach"); + + super::reconcile_terminal_thread_archive_backlog_best_effort( + &config, + &workflow, + &state_store, + ); + + assert!( + state_store + .run_has_protocol_event(run_id, "thread/archive") + .expect("archive event lookup should succeed") + ); +} diff --git a/apps/decodex/src/state/internal.rs b/apps/decodex/src/state/internal.rs index 30c406032..be1b272bb 100644 --- a/apps/decodex/src/state/internal.rs +++ b/apps/decodex/src/state/internal.rs @@ -1662,6 +1662,32 @@ ON CONFLICT(key) DO UPDATE SET value = excluded.value; rows.collect::>>().map_err(Into::into) } + fn list_run_attempts_for_project(&self, project_id: &str) -> Result> { + let mut statement = self.connection.prepare( + "SELECT run_id, project_id, issue_id, attempt_number, status, thread_id, turn_id, \ + updated_at, updated_at_unix FROM run_attempts \ + WHERE project_id = ?1 \ + ORDER BY updated_at_unix DESC, run_id ASC", + )?; + let rows = statement.query_map(params![project_id], run_attempt_record_from_row)?; + + rows.collect::>>().map_err(Into::into) + } + + fn run_has_protocol_event(&self, run_id: &str, event_type: &str) -> Result { + let exists = self.connection.query_row( + "SELECT EXISTS( + SELECT 1 FROM protocol_events + WHERE run_id = ?1 AND event_type = ?2 + LIMIT 1 + )", + params![run_id, event_type], + |row| row.get::<_, i64>(0), + )?; + + Ok(exists != 0) + } + fn load_protocol_event_summaries(&self, state: &mut StateData) -> Result<()> { self.load_compacted_protocol_event_summaries(state)?; diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index f9d3e5823..d91883f95 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -1383,8 +1383,40 @@ impl StateStore { Ok(attempts) } + /// List all locally recorded run attempts for one registered project. + pub fn list_run_attempts_for_project(&self, project_id: &str) -> Result> { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + let attempts = sqlite + .list_run_attempts_for_project(project_id)? + .into_iter() + .map(|attempt| attempt.as_public()) + .collect(); + + return Ok(attempts); + } + + let state = self.lock()?; + let mut attempts = state + .run_attempts + .values() + .filter(|attempt| attempt.project_id.as_deref() == Some(project_id)) + .map(RunAttemptRecord::as_public) + .collect::>(); + + attempts.sort_by(|left, right| right.run_id().cmp(left.run_id())); + + Ok(attempts) + } + /// Return whether one run already has a matching protocol event. pub fn run_has_protocol_event(&self, run_id: &str, event_type: &str) -> Result { + if let Some(sqlite) = &self.sqlite { + let sqlite = sqlite.lock().map_err(|_| eyre::eyre!("State store lock poisoned."))?; + + return sqlite.run_has_protocol_event(run_id, event_type); + } + let state = self.lock()?; Ok(state.events.get(run_id).is_some_and(|events| { diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index 146b15334..9ac79e575 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -1677,6 +1677,47 @@ fn lists_issue_attempts_and_protocol_event_presence() { ); } +#[test] +fn sqlite_lists_project_attempts_and_protocol_event_presence() { + let temp_dir = TempDir::new().expect("tempdir should create"); + let state_path = temp_dir.path().join("runtime.sqlite3"); + let writer = StateStore::open(&state_path).expect("writer state store should open"); + let observer = StateStore::open(&state_path).expect("observer state store should open"); + + writer + .try_acquire_lease("decodex", "issue-1", "run-1", IN_PROGRESS_STATE) + .expect("lease should record project ownership"); + writer + .record_run_attempt("run-1", "issue-1", 1, "succeeded") + .expect("first run attempt should record"); + writer.update_run_thread("run-1", "thread-1").expect("first thread should attach"); + writer.append_event("run-1", 1, "thread/archive", "{}").expect("archive event should record"); + writer + .try_acquire_lease("other", "issue-2", "run-2", IN_PROGRESS_STATE) + .expect("other lease should record project ownership"); + writer + .record_run_attempt("run-2", "issue-2", 1, "succeeded") + .expect("other run attempt should record"); + + let attempts = observer + .list_run_attempts_for_project("decodex") + .expect("project attempts should load from sqlite"); + + assert_eq!(attempts.len(), 1); + assert_eq!(attempts[0].run_id(), "run-1"); + assert_eq!(attempts[0].thread_id(), Some("thread-1")); + assert!( + observer + .run_has_protocol_event("run-1", "thread/archive") + .expect("sqlite event presence should load") + ); + assert!( + !observer + .run_has_protocol_event("run-2", "thread/archive") + .expect("sqlite missing event presence should load") + ); +} + #[test] fn run_activity_marker_round_trips_marker_surfaces() { assert_run_activity_marker_round_trips_clearable_auxiliary_fields();