Adopt app-server notification turn ids

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(&notification.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(&notification.params).is_none_or(|turn_id| turn_id == target_turn_id)
 }

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<_>>(),
+		vec!["implement_to_validation_ready", "handoff_evidence"]
+	);
+}
+
 #[test]
 fn open_phase_goal_stops_at_max_turns_without_terminal_signal() {
 	let handler = ContinueTokenCompletionHandler;

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