Fix thread archive reconciliation

apps/decodex/src/agent.rs

@@ -12,9 +12,10 @@ pub(crate) use self::{
 	app_server::{
 		ACTIVE_RUN_IDLE_TIMEOUT, AppServerCapabilityPreflightFailure, AppServerDynamicToolFailure,
 		AppServerPhaseGoalFailure, AppServerRunRequest, AppServerRunResult,
-		AppServerThreadArchiveRequest, AppServerTurnFailure, PhaseGoalController, PhaseGoalKind,
-		PhaseGoalSpec, PhaseGoalTransition, TurnContinuationGuard, execute_app_server_run,
-		probe_app_server, protocol_activity_idle_timeout,
+		AppServerThreadArchiveOutcome, AppServerThreadArchiveRequest, AppServerTurnFailure,
+		PhaseGoalController, PhaseGoalKind, 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/agent/app_server.rs

@@ -193,6 +193,12 @@ pub(crate) enum PhaseGoalTransition {
 	CompleteRun,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum AppServerThreadArchiveOutcome {
+	Archived,
+	DiscardedMissingThread,
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum AppServerPhaseGoalFailureKind {
 	Unsupported { method: &'static str },
@@ -657,6 +663,16 @@ impl Display for AppServerDynamicToolFailure {
 
 impl Error for AppServerDynamicToolFailure {}
 
+pub(crate) struct AppServerThreadArchiveRequest<'a> {
+	pub(crate) run_id: &'a str,
+	pub(crate) issue_id: &'a str,
+	pub(crate) attempt_number: i64,
+	pub(crate) listen: &'a str,
+	pub(crate) process_env: &'a AppServerProcessEnv,
+	pub(crate) thread_id: &'a str,
+	pub(crate) sequence_number: i64,
+}
+
 #[derive(Clone)]
 pub(crate) struct AppServerRunRequest<'a> {
 	pub(crate) project_id: String,
@@ -681,16 +697,6 @@ pub(crate) struct AppServerRunRequest<'a> {
 	pub(crate) codex_account_provider: Option<&'a dyn CodexAccountProvider>,
 }
 
-pub(crate) struct AppServerThreadArchiveRequest<'a> {
-	pub(crate) run_id: &'a str,
-	pub(crate) issue_id: &'a str,
-	pub(crate) attempt_number: i64,
-	pub(crate) listen: &'a str,
-	pub(crate) process_env: &'a AppServerProcessEnv,
-	pub(crate) thread_id: &'a str,
-	pub(crate) sequence_number: i64,
-}
-
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(crate) struct CommandExecHealthCheck {
 	pub(crate) command: Vec<String>,
@@ -1340,10 +1346,15 @@ pub(crate) fn execute_app_server_run(
 pub(crate) fn archive_app_server_thread_after_success(
 	request: &AppServerThreadArchiveRequest<'_>,
 	state_store: &StateStore,
-) -> crate::prelude::Result<()> {
-	let result = archive_app_server_thread_after_success_inner(request);
+) -> crate::prelude::Result<AppServerThreadArchiveOutcome> {
+	let result = match archive_app_server_thread_after_success_inner(request) {
+		Ok(()) => Ok(AppServerThreadArchiveOutcome::Archived),
+		Err(error) if thread_archive_error_allows_discard(&error) =>
+			Ok(AppServerThreadArchiveOutcome::DiscardedMissingThread),
+		Err(error) => Err(error),
+	};
 
-	record_thread_archive_result_best_effort(state_store, request, result.as_ref().err());
+	record_thread_archive_result_best_effort(state_store, request, result.as_ref());
 
 	result
 }
@@ -1453,24 +1464,33 @@ fn archive_app_server_thread_after_success_inner(
 fn record_thread_archive_result_best_effort(
 	state_store: &StateStore,
 	request: &AppServerThreadArchiveRequest<'_>,
-	error: Option<&Report>,
+	result: std::result::Result<&AppServerThreadArchiveOutcome, &Report>,
 ) {
-	let (event_type, payload) = match error {
-		Some(error) => (
-			"thread/archive/failed",
+	let (event_type, payload) = match result {
+		Ok(AppServerThreadArchiveOutcome::Archived) => (
+			"thread/archive",
 			serde_json::json!({
 				"threadId": request.thread_id,
 				"issueId": request.issue_id,
 				"attemptNumber": request.attempt_number,
-				"error": error.to_string(),
 			}),
 		),
-		None => (
-			"thread/archive",
+		Ok(AppServerThreadArchiveOutcome::DiscardedMissingThread) => (
+			"thread/archive/discarded",
+			serde_json::json!({
+				"threadId": request.thread_id,
+				"issueId": request.issue_id,
+				"attemptNumber": request.attempt_number,
+				"reason": "missing_thread_or_rollout",
+			}),
+		),
+		Err(error) => (
+			"thread/archive/failed",
 			serde_json::json!({
 				"threadId": request.thread_id,
 				"issueId": request.issue_id,
 				"attemptNumber": request.attempt_number,
+				"error": error.to_string(),
 			}),
 		),
 	};
@@ -1493,6 +1513,12 @@ fn record_thread_archive_result_best_effort(
 	}
 }
 
+fn thread_archive_error_allows_discard(error: &Report) -> bool {
+	let message = error.to_string().to_lowercase();
+
+	thread_missing_error_message_allows_discard(&message) || message.contains("already archived")
+}
+
 fn classify_child_activity_event(
 	event_type: &str,
 	payload: &str,
@@ -5344,6 +5370,10 @@ fn normalized_home_path(path: &Path) -> PathBuf {
 fn thread_resume_error_allows_fallback(error: &Report) -> bool {
 	let message = error.to_string().to_lowercase();
 
+	thread_missing_error_message_allows_discard(&message)
+}
+
+fn thread_missing_error_message_allows_discard(message: &str) -> bool {
 	message.contains("no rollout found for thread id") || message.contains("thread not found")
 }
 

apps/decodex/src/agent/app_server/tests.rs

@@ -16,13 +16,13 @@ use crate::{
 		app_server::{
 			APP_SERVER_SCHEMA_REQUIRED_MARKERS, AppServerCapabilityPreflightFailure,
 			AppServerCapabilityPreflightReport, AppServerDynamicToolFailure,
-			AppServerPhaseGoalFailure, AppServerRunResult, AppServerThreadArchiveRequest,
-			AppServerTurnFailure, CommandExecHealthCheck, CommandExecResponse,
-			EffectiveThreadConfig, InitializeResponse, ModelProviderCapabilitiesReadResponse,
-			PhaseGoalController, PhaseGoalKind, PhaseGoalSpec, PhaseGoalTransition,
-			PluginListResponse, ProbeDynamicToolHandler, REQUEST_TIMEOUT, RequestWaitPhase,
-			RunRecorder, RuntimeConfigSummary, SkillsListResponse, TurnContinuationGuard,
-			UserInput,
+			AppServerPhaseGoalFailure, AppServerRunResult, AppServerThreadArchiveOutcome,
+			AppServerThreadArchiveRequest, AppServerTurnFailure, CommandExecHealthCheck,
+			CommandExecResponse, EffectiveThreadConfig, InitializeResponse,
+			ModelProviderCapabilitiesReadResponse, PhaseGoalController, PhaseGoalKind,
+			PhaseGoalSpec, PhaseGoalTransition, PluginListResponse, ProbeDynamicToolHandler,
+			REQUEST_TIMEOUT, RequestWaitPhase, RunRecorder, RuntimeConfigSummary,
+			SkillsListResponse, TurnContinuationGuard, UserInput,
 		},
 		json_rpc::{
 			AppServerHomePreflightFailure, AppServerOutputTimeout, AppServerProcessEnv,
@@ -1185,7 +1185,7 @@ for line in sys.stdin:
 		.record_run_attempt("run-1", "issue-1", 1, "succeeded")
 		.expect("run attempt should record");
 
-	super::archive_app_server_thread_after_success(
+	let outcome = super::archive_app_server_thread_after_success(
 		&AppServerThreadArchiveRequest {
 			run_id: "run-1",
 			issue_id: "issue-1",
@@ -1198,17 +1198,66 @@ for line in sys.stdin:
 		&state_store,
 	)
 	.expect("thread archive should succeed");
-
 	let invocation_log =
 		fs::read_to_string(&invocation_log_path).expect("invocation log should exist");
 
 	assert!(invocation_log.contains(r#""app-server""#));
 	assert!(invocation_log.contains(r#""--listen""#));
 	assert!(invocation_log.contains(r#""method": "thread/archive""#));
 	assert!(invocation_log.contains(r#""threadId": "thread-1""#));
+	assert_eq!(outcome, AppServerThreadArchiveOutcome::Archived);
+	assert!(
+		state_store
+			.run_has_protocol_event("run-1", "thread/archive")
+			.expect("archive event lookup should succeed")
+	);
 	assert_eq!(state_store.event_count("run-1").expect("event count should load"), 1);
 }
 
+#[test]
+fn missing_thread_archive_errors_record_discarded_terminal_event() {
+	let state_store = StateStore::open_in_memory().expect("state store should open");
+	let process_env = AppServerProcessEnv::default();
+	let request = AppServerThreadArchiveRequest {
+		run_id: "run-1",
+		issue_id: "issue-1",
+		attempt_number: 1,
+		listen: "stdio://",
+		process_env: &process_env,
+		thread_id: "thread-1",
+		sequence_number: 1,
+	};
+
+	state_store
+		.record_run_attempt("run-1", "issue-1", 1, "succeeded")
+		.expect("run attempt should record");
+
+	super::record_thread_archive_result_best_effort(
+		&state_store,
+		&request,
+		Ok(&AppServerThreadArchiveOutcome::DiscardedMissingThread),
+	);
+
+	assert!(super::thread_archive_error_allows_discard(&eyre::eyre!(
+		"no rollout found for thread id thread-1"
+	)));
+	assert!(super::thread_archive_error_allows_discard(&eyre::eyre!("thread not found")));
+	assert!(super::thread_archive_error_allows_discard(&eyre::eyre!("already archived")));
+	assert!(!super::thread_archive_error_allows_discard(&eyre::eyre!(
+		"failed to load rollout from disk"
+	)));
+	assert!(
+		state_store
+			.run_has_protocol_event("run-1", "thread/archive/discarded")
+			.expect("discarded archive event lookup should succeed")
+	);
+	assert!(
+		!state_store
+			.run_has_protocol_event("run-1", "thread/archive/failed")
+			.expect("failed archive event lookup should succeed")
+	);
+}
+
 #[test]
 fn command_exec_health_check_validates_exact_buffered_result() {
 	let health_check = CommandExecHealthCheck {

apps/decodex/src/orchestrator/execution.rs

@@ -1,7 +1,7 @@
 use git_credentials::GitSigningConfig;
 use agent::CodexAccountPool;
 use agent::CodexAccountProvider;
-use agent::AppServerThreadArchiveRequest;
+use agent::{AppServerThreadArchiveOutcome, AppServerThreadArchiveRequest};
 use records::LinearExecutionEventPublicProjection;
 use sha2::Digest;
 use state::DecisionContractRecord;
@@ -1681,7 +1681,7 @@ fn push_thread_archive_candidate(
 	source: ThreadArchiveCandidateSource<'_>,
 ) -> Result<()> {
 	if !seen_thread_ids.insert(source.thread_id.to_owned())
-		|| state_store.run_has_protocol_event(source.run_id, "thread/archive")?
+		|| run_has_terminal_thread_archive_event(state_store, source.run_id)?
 	{
 		return Ok(());
 	}
@@ -1700,6 +1700,16 @@ fn push_thread_archive_candidate(
 	Ok(())
 }
 
+fn run_has_terminal_thread_archive_event(state_store: &StateStore, run_id: &str) -> Result<bool> {
+	for event_type in ["thread/archive", "thread/archive/discarded"] {
+		if state_store.run_has_protocol_event(run_id, event_type)? {
+			return Ok(true);
+		}
+	}
+
+	Ok(false)
+}
+
 fn archive_completed_issue_thread_best_effort(
 	project: &ServiceConfig,
 	state_store: &StateStore,
@@ -1719,20 +1729,24 @@ fn archive_completed_issue_thread_best_effort(
 	#[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(),
-	);
+	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(),
+			)
+			.map(|()| AppServerThreadArchiveOutcome::Archived)
+	};
 
 	match archive_result {
-		Ok(()) => tracing::info!(
+		Ok(AppServerThreadArchiveOutcome::Archived) => tracing::info!(
 			project_id = project.service_id(),
 			issue_id = candidate.issue_id,
 			issue = candidate.issue_identifier,
@@ -1741,6 +1755,15 @@ fn archive_completed_issue_thread_best_effort(
 			thread_id = %candidate.thread_id,
 			"Archived completed issue app-server thread."
 		),
+		Ok(AppServerThreadArchiveOutcome::DiscardedMissingThread) => tracing::info!(
+			project_id = project.service_id(),
+			issue_id = candidate.issue_id,
+			issue = candidate.issue_identifier,
+			run_id = candidate.run_id,
+			attempt = candidate.attempt_number,
+			thread_id = %candidate.thread_id,
+			"Discarded completed issue app-server thread archive because the thread is missing."
+		),
 		Err(error) => tracing::warn!(
 			?error,
 			project_id = project.service_id(),

apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs

@@ -87,6 +87,7 @@ fn terminal_thread_archive_backlog_candidates_scan_project_terminal_runs() {
 		("decodex", "issue-running", "run-running", "running", "thread-running"),
 		("other", "issue-other", "run-other", "succeeded", "thread-other"),
 		("decodex", "issue-archived", "run-archived", "succeeded", "thread-archived"),
+		("decodex", "issue-discarded", "run-discarded", "succeeded", "thread-discarded"),
 	] {
 		state_store
 			.try_acquire_lease(project, issue_id, run_id, "In Progress")
@@ -100,6 +101,9 @@ fn terminal_thread_archive_backlog_candidates_scan_project_terminal_runs() {
 	state_store
 		.append_event("run-archived", 1, "thread/archive", "{}")
 		.expect("archive event should record");
+	state_store
+		.append_event("run-discarded", 1, "thread/archive/discarded", "{}")
+		.expect("discarded archive event should record");
 
 	let candidates = super::terminal_thread_archive_backlog_candidates(&state_store, "decodex")
 		.expect("backlog candidates should load");

docs/spec/app-server.md

@@ -126,7 +126,7 @@ To validate an upstream app-server protocol change:
   - `turn/start`
   - `thread/archive` after successful completion writeback, for every locally
     recorded terminal attempt thread on the issue that has not already recorded a
-    successful archive event
+    terminal archive event (`thread/archive` or `thread/archive/discarded`)
 - Required notifications for the MVP:
   - `thread/started`
   - `thread/status/changed`

docs/spec/runtime.md

@@ -383,7 +383,7 @@ If the label intent is recorded without the required explanatory comment, `decod
 If the resolved terminal path is not explicitly finalized through `issue_terminal_finalize`, the app-server wrapper must fail the turn before `decodex` records the attempt as successful.
 The explanatory public summary for `manual_attention` must describe the exact observed blocker and should include the failed command plus raw error text only when those values are public-safe, instead of speculating about unverified capability limits. It must not reuse runtime-owned retry or continued-repair `error_class` values such as app-server timeout/turn failures, stalled-run detection, or repo-gate validation failure classes; those remain Decodex-owned retry, continuation, or architecture-recovery signals until the runtime itself reaches a human-required terminal boundary.
 Execution-state checkpoints are durable progress overlays only. Their phase, focus, next action, blockers, evidence, or verification fields are never a substitute for the explicit terminal-finalization call.
-After successful completion writeback, Decodex must best-effort archive every locally recorded terminal Codex thread for the issue, including earlier failed retry attempts, so old attempts do not keep the issue visible in the Codex conversation list.
+After successful completion writeback, Decodex must best-effort archive every locally recorded terminal Codex thread for the issue, including earlier failed retry attempts, so old attempts do not keep the issue visible in the Codex conversation list. If Codex reports that a recorded terminal thread is already absent, already archived, or has no rollout to archive, Decodex must record a terminal archive-discard outcome instead of treating the attempt as an indefinitely retryable archive failure.
 
 ### Progress checkpoint writeback