diff --git a/apps/decodex/src/agent.rs b/apps/decodex/src/agent.rs index f71f8f047..2e744d33e 100644 --- a/apps/decodex/src/agent.rs +++ b/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}, diff --git a/apps/decodex/src/agent/app_server.rs b/apps/decodex/src/agent/app_server.rs index 7a6c2d82c..7c24b7038 100644 --- a/apps/decodex/src/agent/app_server.rs +++ b/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, @@ -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 { + 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") } diff --git a/apps/decodex/src/agent/app_server/tests.rs b/apps/decodex/src/agent/app_server/tests.rs index d45927b3d..64718b44c 100644 --- a/apps/decodex/src/agent/app_server/tests.rs +++ b/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,7 +1198,6 @@ 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"); @@ -1206,9 +1205,59 @@ for line in sys.stdin: 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 { diff --git a/apps/decodex/src/orchestrator/execution.rs b/apps/decodex/src/orchestrator/execution.rs index 3c0c7e317..353890b0e 100644 --- a/apps/decodex/src/orchestrator/execution.rs +++ b/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 { + 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(), diff --git a/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs b/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs index 0dee7abb0..d685b460b 100644 --- a/apps/decodex/src/orchestrator/tests/runtime/thread_archive.rs +++ b/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"); diff --git a/docs/spec/app-server.md b/docs/spec/app-server.md index 4cfa0104f..15ec768c0 100644 --- a/docs/spec/app-server.md +++ b/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` diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index a21f8e66f..1e4e7105f 100644 --- a/docs/spec/runtime.md +++ b/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