Block goals after three empty automatic continuation turns (#44320)

## Why

Automatic goal continuations can repeatedly return empty final answers without
making progress. Stop this loop by marking the goal as `blocked` after three
consecutive empty turns with no other activity.

## What changed

- Observe completed turn items through a new `on_item_completed` lifecycle hook.
- Track empty final answers only for automatically admitted goal turns, resetting
  the streak on activity, user turns, or goal changes.
- Preserve normal turn completion and streamed message deltas when blocking a goal.

## Testing

Add accounting coverage for the three-turn threshold and streak resets, plus
app-server tests for empty continuations and recovery through final-answer text,
commentary, or tool activity.

GitOrigin-RevId: 4b9d2cb2e306b0adc316972429cc35000115b88b

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -107,6 +107,7 @@ mod thread_environments;
 mod thread_fork;
 #[path = "thread_fork_multi_agent_tests.rs"]
 mod thread_fork_multi_agent;
+mod thread_goal_empty_responses;
 mod thread_inject_items;
 mod thread_list;
 mod thread_loaded_list;

codex-rs/app-server/tests/suite/v2/thread_goal_empty_responses.rs

@@ -0,0 +1,110 @@
+//! Exercises the goal extension's empty-response breaker through the public API.
+
+use anyhow::Result;
+use app_test_support::MockResponsesConfig;
+use app_test_support::TestAppServer;
+use app_test_support::create_mock_responses_server_sequence;
+use codex_app_server_protocol::ThreadGoalGetResponse;
+use codex_app_server_protocol::ThreadGoalSetResponse;
+use codex_app_server_protocol::ThreadGoalStatus;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnCompletedNotification;
+use codex_app_server_protocol::TurnStatus;
+use codex_features::Feature;
+use core_test_support::responses;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use tempfile::TempDir;
+use tokio::time::Duration;
+use tokio::time::timeout;
+
+#[test_case::test_case(None; "empty")]
+#[test_case::test_case(Some("final_answer"); "draft")]
+#[test_case::test_case(Some("commentary"); "commentary")]
+#[test_case::test_case(Some("tool"); "tool")]
+#[tokio::test]
+async fn empty_goal_continuations_block_after_three_without_activity(
+    recovery: Option<&str>,
+) -> Result<()> {
+    let turns = if recovery.is_some() { 6 } else { 3 };
+    let mut scripts = Vec::new();
+    for turn in 1..=turns {
+        let mut id = format!("response-{turn}");
+        let mut events = vec![responses::ev_response_created(&id)];
+        if turn == 3
+            && let Some(recovery) = recovery
+        {
+            if recovery == "tool" {
+                events.push(responses::ev_function_call("get-goal", "get_goal", "{}"));
+                events.push(responses::ev_completed(&id));
+                scripts.push(responses::sse(events));
+                id.push_str("-after-tool");
+                events = vec![responses::ev_response_created(&id)];
+            } else {
+                let mut added = responses::ev_assistant_message("progress", "");
+                added["type"] = json!("response.output_item.added");
+                events.push(added);
+                events.push(responses::ev_output_text_delta("Useful progress"));
+                let mut event = responses::ev_assistant_message("progress", "Useful progress");
+                event["item"]["phase"] = json!(recovery);
+                events.push(event);
+            }
+        }
+        let mut empty_final = responses::ev_assistant_message(&format!("empty-final-{turn}"), "");
+        empty_final["item"]["phase"] = json!("final_answer");
+        events.push(empty_final);
+        events.push(responses::ev_completed(&id));
+        scripts.push(responses::sse(events));
+    }
+    let server = create_mock_responses_server_sequence(scripts).await;
+    let codex_home = TempDir::new()?;
+    MockResponsesConfig::new(&server.uri())
+        .with_model("gpt-5.4")
+        .enable_feature(Feature::Goals)
+        .write(codex_home.path())?;
+    let mut mcp = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_managed_config()
+        .build_initialized()
+        .await?;
+    let request = mcp
+        .send_thread_start_request_with_auto_env(ThreadStartParams::default())
+        .await?;
+    let ThreadStartResponse { thread, .. } = mcp.read_response(request).await?;
+    let request = mcp
+        .send_raw_request(
+            "thread/goal/set",
+            Some(json!({"threadId": thread.id, "objective": "Finish the work"})),
+        )
+        .await?;
+    let _: ThreadGoalSetResponse = mcp.read_response(request).await?;
+
+    for turn in 1..=turns {
+        if turn == 3 && matches!(recovery, Some("final_answer" | "commentary")) {
+            let delta: serde_json::Value = timeout(
+                Duration::from_secs(30),
+                mcp.read_notification("item/agentMessage/delta"),
+            )
+            .await??;
+            assert_eq!(json!("Useful progress"), delta["delta"]);
+        }
+        let completed: TurnCompletedNotification = timeout(
+            Duration::from_secs(30),
+            mcp.read_notification("turn/completed"),
+        )
+        .await??;
+        assert_eq!(TurnStatus::Completed, completed.turn.status);
+        assert_eq!(None, completed.turn.error);
+    }
+    let request = mcp
+        .send_raw_request("thread/goal/get", Some(json!({"threadId": thread.id})))
+        .await?;
+    let result: ThreadGoalGetResponse = mcp.read_response(request).await?;
+    assert_eq!(
+        ThreadGoalStatus::Blocked,
+        result.goal.expect("goal exists").status
+    );
+    server.verify().await;
+    Ok(())
+}