From 12b7f1e33bfc92658fa6dadb088f79f63e5f4e14 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 16 Jun 2026 23:09:50 +0800 Subject: [PATCH 1/3] {"schema":"decodex/commit/1","summary":"Adopt app-server notification turn ids","authority":"manual"} --- apps/decodex/src/agent/app_server.rs | 40 +++++++++++++++ apps/decodex/src/agent/app_server/tests.rs | 57 ++++++++++++++++++++-- docs/spec/app-server.md | 8 +++ 3 files changed, 101 insertions(+), 4 deletions(-) diff --git a/apps/decodex/src/agent/app_server.rs b/apps/decodex/src/agent/app_server.rs index 8281b98f7..acb22450a 100644 --- a/apps/decodex/src/agent/app_server.rs +++ b/apps/decodex/src/agent/app_server.rs @@ -4160,6 +4160,12 @@ fn wait_for_turn_completion( match &wire_message.message { JsonRpcMessage::Notification(notification) => { + adopt_thread_bound_notification_turn_id( + recorder, + notification, + target_thread_id, + &mut target_turn_id, + )?; if let Some(outcome) = handle_turn_execution_notification( notification, target_thread_id, @@ -4296,6 +4302,40 @@ fn handle_turn_execution_notification( Ok(None) } +fn adopt_thread_bound_notification_turn_id( + recorder: &mut RunRecorder<'_>, + notification: &JsonRpcNotification, + target_thread_id: &str, + target_turn_id: &mut String, +) -> crate::prelude::Result<()> { + let Some(observed_turn_id) = turn_id_from_value(¬ification.params) else { + return Ok(()); + }; + + if observed_turn_id == target_turn_id { + return Ok(()); + } + if !thread_id_from_notification(notification) + .is_some_and(|thread_id| thread_id == target_thread_id) + { + return Ok(()); + } + + tracing::warn!( + target_thread_id, + previous_turn_id = target_turn_id.as_str(), + observed_turn_id, + method = notification.method.as_str(), + "App-server notification turn id differed from the turn/start response; adopting thread-bound notification turn id." + ); + + recorder.state_store.update_run_turn(recorder.run_id, observed_turn_id)?; + recorder.set_turn_id(observed_turn_id)?; + *target_turn_id = observed_turn_id.to_owned(); + + Ok(()) +} + fn notification_targets_turn(notification: &JsonRpcNotification, target_turn_id: &str) -> bool { turn_id_from_value(¬ification.params).is_none_or(|turn_id| turn_id == target_turn_id) } diff --git a/apps/decodex/src/agent/app_server/tests.rs b/apps/decodex/src/agent/app_server/tests.rs index f925e2edc..14f562843 100644 --- a/apps/decodex/src/agent/app_server/tests.rs +++ b/apps/decodex/src/agent/app_server/tests.rs @@ -50,6 +50,7 @@ import sys TURN_OUTPUTS = __TURN_OUTPUTS__ GOAL_STATUSES = __GOAL_STATUSES__ UNSUPPORTED_GOAL_METHODS = __UNSUPPORTED_GOAL_METHODS__ +MISMATCH_NOTIFICATION_TURN_IDS = __MISMATCH_NOTIFICATION_TURN_IDS__ goal = { "objective": "", @@ -171,6 +172,7 @@ for line in sys.stdin: elif method == "turn/start": turn_count += 1 turn_id = "turn-" + str(turn_count) + notification_turn_id = "notification-" + turn_id if MISMATCH_NOTIFICATION_TURN_IDS else turn_id if TURN_OUTPUTS: output = TURN_OUTPUTS[min(turn_count - 1, len(TURN_OUTPUTS) - 1)] else: @@ -182,22 +184,22 @@ for line in sys.stdin: }}) send({"method": "turn/started", "params": { "threadId": "thread-1", - "turn": {"id": turn_id, "status": "running", "error": None}, + "turn": {"id": notification_turn_id, "status": "running", "error": None}, }}) if not unsupported_goal_method("thread/goal/updated"): send({"method": "thread/goal/updated", "params": { "threadId": "thread-1", - "turnId": turn_id, + "turnId": notification_turn_id, "goal": goal_payload("active"), }}) send({"method": "item/completed", "params": { "threadId": "thread-1", - "turnId": turn_id, + "turnId": notification_turn_id, "item": {"type": "agentMessage", "text": output}, }}) send({"method": "turn/completed", "params": { "threadId": "thread-1", - "turn": {"id": turn_id, "status": "completed", "error": None}, + "turn": {"id": notification_turn_id, "status": "completed", "error": None}, }}) else: method_not_found(message_id, method) @@ -909,17 +911,33 @@ fn phase_goal_fake_codex_script( turn_outputs: &[&str], goal_statuses: &[&str], unsupported_goal_methods: &[&str], +) -> String { + phase_goal_fake_codex_script_with_notification_turn_mismatch( + turn_outputs, + goal_statuses, + unsupported_goal_methods, + false, + ) +} + +fn phase_goal_fake_codex_script_with_notification_turn_mismatch( + turn_outputs: &[&str], + goal_statuses: &[&str], + unsupported_goal_methods: &[&str], + mismatch_notification_turn_ids: bool, ) -> String { let outputs_json = serde_json::to_string(turn_outputs).expect("turn outputs should serialize"); let statuses_json = serde_json::to_string(goal_statuses).expect("goal statuses should serialize"); let unsupported_goal = serde_json::to_string(unsupported_goal_methods).expect("methods should serialize"); + let mismatch_turn_ids = if mismatch_notification_turn_ids { "True" } else { "False" }; PHASE_GOAL_FAKE_CODEX_SCRIPT_TEMPLATE .replace("__TURN_OUTPUTS__", &outputs_json) .replace("__GOAL_STATUSES__", &statuses_json) .replace("__UNSUPPORTED_GOAL_METHODS__", &unsupported_goal) + .replace("__MISMATCH_NOTIFICATION_TURN_IDS__", mismatch_turn_ids) } fn execute_phase_goal_fake_app_server<'a, F>( @@ -1026,6 +1044,37 @@ fn phase_goal_complete_runs_validation_transition_before_handoff_goal() { assert_eq!(goal_set_events[1]["phase"], "handoff_evidence"); } +#[test] +fn phase_goal_completion_accepts_thread_bound_notification_turn_alias() { + let handler = TerminalTokenCompletionHandler::default(); + let controller = TestPhaseGoalController::new(PhaseGoalKind::ImplementToValidationReady); + let script = phase_goal_fake_codex_script_with_notification_turn_mismatch( + &["DONE", "TERMINAL"], + &["complete", "complete"], + &[], + true, + ); + let (result, state_store) = execute_phase_goal_fake_app_server(script, |request| { + request.max_turns = 3; + request.dynamic_tool_handler = Some(&handler); + request.phase_goal_controller = Some(&controller); + }); + let result = result.expect("thread-bound turn alias should still complete phase goals"); + let completed_events = private_phase_goal_events(&state_store, "phase_goal_completed"); + let run_attempt = state_store + .run_attempt("phase-goal-run") + .expect("run attempt should load") + .expect("run attempt should exist"); + + assert_eq!(result.turn_count, 2); + assert_eq!(result.turn_id, "notification-turn-2"); + assert_eq!(run_attempt.turn_id(), Some("notification-turn-2")); + assert_eq!( + completed_events.iter().filter_map(|event| event["phase"].as_str()).collect::>(), + vec!["implement_to_validation_ready", "handoff_evidence"] + ); +} + #[test] fn open_phase_goal_stops_at_max_turns_without_terminal_signal() { let handler = ContinueTokenCompletionHandler; diff --git a/docs/spec/app-server.md b/docs/spec/app-server.md index ba5b0d16d..f29c7eaed 100644 --- a/docs/spec/app-server.md +++ b/docs/spec/app-server.md @@ -176,6 +176,9 @@ app-server connection for active turns. 6. Send `thread/goal/set` with the controller-owned phase goal for the run. 7. Send `turn/start`. 8. Consume notifications until that turn reaches a terminal outcome. + If the `turn/start` response id and same-thread notification turn id differ, + Decodex treats the notification turn id as the active turn id for subsequent + item, goal, completion, and lane-control readbacks. 9. Send `thread/goal/get` after the turn completes. If the goal status is `complete`, Decodex runs the next owned phase transition such as repository validation, validation repair, review repair, or handoff evidence. If the goal remains active @@ -496,11 +499,16 @@ Rationale: ### `turn/started` - Record the turn identifier and transition the local run into `running`. +- When this id differs from the id returned by `turn/start` but the thread id matches, + adopt the notification id as the current active turn id before filtering later + turn-scoped events. ### `turn/completed` - Record the completed turn payload. - Classify the turn as success, retryable failure, or terminal failure. +- A same-thread completion for the adopted notification turn id completes the active + Decodex turn even when the original `turn/start` response id was different. ## Error handling From afab28f237404f021d9ebcb00e377a5eda0a9727 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 16 Jun 2026 23:14:10 +0800 Subject: [PATCH 2/3] {"schema":"decodex/commit/1","summary":"Fix Rust style spacing","authority":"manual"} --- apps/decodex/src/agent/app_server.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/decodex/src/agent/app_server.rs b/apps/decodex/src/agent/app_server.rs index acb22450a..ff7849d56 100644 --- a/apps/decodex/src/agent/app_server.rs +++ b/apps/decodex/src/agent/app_server.rs @@ -4166,6 +4166,7 @@ fn wait_for_turn_completion( target_thread_id, &mut target_turn_id, )?; + if let Some(outcome) = handle_turn_execution_notification( notification, target_thread_id, @@ -4331,6 +4332,7 @@ fn adopt_thread_bound_notification_turn_id( recorder.state_store.update_run_turn(recorder.run_id, observed_turn_id)?; recorder.set_turn_id(observed_turn_id)?; + *target_turn_id = observed_turn_id.to_owned(); Ok(()) From 72ded7875521a03f17c07e4ff17318b23d0f7f93 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 16 Jun 2026 23:18:07 +0800 Subject: [PATCH 3/3] {"schema":"decodex/commit/1","summary":"Satisfy Rust clippy for turn id adoption","authority":"manual"} --- apps/decodex/src/agent/app_server.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/decodex/src/agent/app_server.rs b/apps/decodex/src/agent/app_server.rs index ff7849d56..31dcdc560 100644 --- a/apps/decodex/src/agent/app_server.rs +++ b/apps/decodex/src/agent/app_server.rs @@ -4316,8 +4316,8 @@ fn adopt_thread_bound_notification_turn_id( if observed_turn_id == target_turn_id { return Ok(()); } - if !thread_id_from_notification(notification) - .is_some_and(|thread_id| thread_id == target_thread_id) + if thread_id_from_notification(notification) + .is_none_or(|thread_id| thread_id != target_thread_id) { return Ok(()); }