diff --git a/apps/decodex/src/agent/app_server.rs b/apps/decodex/src/agent/app_server.rs index 8281b98f7..31dcdc560 100644 --- a/apps/decodex/src/agent/app_server.rs +++ b/apps/decodex/src/agent/app_server.rs @@ -4160,6 +4160,13 @@ 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 +4303,41 @@ 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_none_or(|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