Archive terminal Codex threads from backlog

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},

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()

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<i64>,
+}
+
 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,63 +1194,175 @@ 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<ThreadArchiveCandidate>,
+) {
+	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()
 			},
 		};
 
 	for candidate in candidates {
 		archive_completed_issue_thread_best_effort(
 			project,
 			state_store,
-			issue_run,
 			process_env,
 			transport,
 			&candidate,
 		);
 	}
 }
 
+#[cfg(test)]
 fn completed_issue_thread_archive_candidates(
 	state_store: &StateStore,
 	issue_run: &IssueRunPlan,
 	run_result: &AppServerRunResult,
+) -> Result<Vec<ThreadArchiveCandidate>> {
+	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<ThreadArchiveCandidate>,
 ) -> Result<Vec<ThreadArchiveCandidate>> {
 	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: &current.run_id,
+				issue_id: &current.issue_id,
+				issue_identifier: &current.issue_identifier,
+				attempt_number: current.attempt_number,
+				thread_id: &current.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<Vec<ThreadArchiveCandidate>> {
+	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<ThreadArchiveCandidate>,
 	seen_thread_ids: &mut HashSet<String>,
-	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,