修复测试的竞态问题

2026-07-27 05:23:40 +00:00
parent a5c9f68bfe
commit 1580b3daf3
4 changed files with 181 additions and 95 deletions
@@ -380,7 +380,6 @@ async fn supervisor_collaboration_v2_batch_recovers_durable_isolated_spawn_witho
    assert!(batch.collaboration_contract.is_some());
    assert_eq!(batch.actions.len(), 1);

-    let original_batch_id = batch.batch_id.clone();
    let original_agent_id = batch.agent_id.clone();
    let original_session_id = batch.session_id.clone();
    let original_run_id = batch.run_id.clone();
@@ -477,63 +476,30 @@ async fn supervisor_collaboration_v2_batch_recovers_durable_isolated_spawn_witho
    ));

    let recovered_runtime =
-        read_game_creator_agent_runtime_at(&root, GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID)
-            .expect("read recovered supervisor runtime")
+        wait_for_agent_runtime_lane_release_async(&root, GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID)
+            .await
            .state;
    assert_ne!(recovered_runtime.phase, "needs-reconciliation");
    assert_eq!(recovered_runtime.agent_id, original_agent_id);
    assert_eq!(recovered_runtime.session_id, original_session_id);
    assert_eq!(recovered_runtime.run_id, original_run_id);
-    let recovered_pending = read_game_creator_agent_runtime_pending_tool_action(
-        &root,
-        GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID,
-        run_id,
-    )
-    .expect("read recovered isolated pending action");
-    assert_eq!(
-        recovered_pending.status,
-        AGENT_RUNTIME_PENDING_ACTION_STATUS_OBSERVED_APPROVED
-    );
-    assert_eq!(recovered_pending.agent_id, original_agent_id);
-    assert_eq!(recovered_pending.session_id, original_session_id);
-    assert_eq!(recovered_pending.run_id, original_run_id);
-    assert_eq!(recovered_pending.action_id, original_action_id);
-    assert_eq!(
-        recovered_pending.action_fingerprint,
-        original_action_fingerprint
-    );
-    assert_eq!(
-        recovered_pending
-            .observation
-            .as_ref()
-            .map(|observation| observation.status.as_str()),
-        Some("ok")
-    );
-
    assert!(
-        update_game_creator_agent_runtime_provider_batch_member(&root, &recovered_pending)
-            .expect("complete recovered isolated batch cursor")
+        !game_creator_agent_runtime_provider_action_batch_path(
+            &root,
+            GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID,
+            run_id,
+        )
+        .exists(),
+        "completed provider batch must be removed before the Agent lane releases"
    );
-    let completed = read_game_creator_agent_runtime_provider_action_batch(
-        &root,
-        GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID,
-        run_id,
-    )
-    .expect("read completed isolated provider batch");
-    assert_eq!(completed.batch_id, original_batch_id);
-    assert_eq!(completed.agent_id, original_agent_id);
-    assert_eq!(completed.session_id, original_session_id);
-    assert_eq!(completed.run_id, original_run_id);
-    assert_eq!(completed.next_action_index, 1);
-    assert_eq!(completed.status, "completed");
-    assert_eq!(completed.actions[0].action_id, original_action_id);
-    assert_eq!(
-        completed.actions[0].action_fingerprint,
-        original_action_fingerprint
-    );
-    assert_eq!(
-        completed.actions[0].status,
-        AGENT_RUNTIME_PENDING_ACTION_STATUS_OBSERVED_APPROVED
+    assert!(
+        !game_creator_agent_runtime_pending_tool_action_path(
+            &root,
+            GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID,
+            run_id,
+        )
+        .exists(),
+        "observed pending action must be removed before the Agent lane releases"
    );

    let summary =
@@ -556,22 +522,41 @@ async fn supervisor_collaboration_v2_batch_recovers_durable_isolated_spawn_witho
    .expect("re-read isolated group after recovery");
    assert_eq!(stable_group, group);
    let records = read_agent_db_records_for_test(&root);
+    let observed_actions = records
+        .iter()
+        .filter(|record| {
+            record["recordType"] == "agent.runtime.tool_action.observed"
+                && record["runId"] == run_id
+                && record["tool"] == "agent.spawn_isolated"
+        })
+        .collect::<Vec<_>>();
    assert_eq!(
-        records
-            .iter()
-            .filter(|record| {
-                record.get("recordType").and_then(Value::as_str)
-                    == Some("agent.runtime.agent.spawn_isolated")
-                    && record.get("agentId").and_then(Value::as_str)
-                        == Some(GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID)
-                    && record.get("runId").and_then(Value::as_str) == Some(run_id)
-                    && record.get("actionId").and_then(Value::as_str)
-                        == Some(original_action_id.as_str())
-            })
-            .count(),
+        observed_actions.len(),
+        1,
+        "recovery must persist one isolated spawn observation identity"
+    );
+    assert_eq!(observed_actions[0]["actionId"], original_action_id);
+    assert_eq!(
+        observed_actions[0]["actionFingerprint"],
+        original_action_fingerprint
+    );
+    assert_eq!(observed_actions[0]["observationStatus"], "ok");
+    let spawn_audits = records
+        .iter()
+        .filter(|record| {
+            record.get("recordType").and_then(Value::as_str)
+                == Some("agent.runtime.agent.spawn_isolated")
+                && record.get("agentId").and_then(Value::as_str)
+                    == Some(GAME_CREATOR_PROJECT_SUPERVISOR_AGENT_ID)
+                && record.get("runId").and_then(Value::as_str) == Some(run_id)
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(
+        spawn_audits.len(),
        1,
        "recovery must not duplicate the spawn audit",
    );
+    assert_eq!(spawn_audits[0]["actionId"], original_action_id);
    fs::remove_dir_all(root).ok();
 }

@@ -148,6 +148,81 @@ fn wait_for_agent_runtime_terminal_and_lane_release(
    );
 }

+async fn wait_for_agent_runtime_terminal_and_lane_release_async(
+    root: &Path,
+    agent_id: &str,
+    run_id: &str,
+    status: &str,
+    phase: &str,
+) -> AgentRuntimeResult {
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    let mut last_lane_probe_error = None;
+    let mut result = read_game_creator_agent_runtime_at(root, agent_id)
+        .expect("read runtime while asynchronously waiting for terminal lane release");
+    loop {
+        let matches_terminal = result.state.run_id == run_id
+            && result.state.status == status
+            && result.state.phase == phase;
+        let lane_is_available = if matches_terminal {
+            match game_creator_agent_runtime_task_lock_is_available(root, agent_id) {
+                Ok(is_available) => is_available,
+                Err(error) => {
+                    last_lane_probe_error = Some(error);
+                    false
+                }
+            }
+        } else {
+            false
+        };
+        if lane_is_available {
+            let terminal = read_game_creator_agent_runtime_at(root, agent_id)
+                .expect("reread runtime after asynchronous terminal lane release");
+            if terminal.state.run_id == run_id
+                && terminal.state.status == status
+                && terminal.state.phase == phase
+            {
+                return terminal;
+            }
+            result = terminal;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "runtime did not reach {status}/{phase} for run {run_id} before the Agent lane released; last run={} status={} phase={}; last lane probe error={}",
+            result.state.run_id,
+            result.state.status,
+            result.state.phase,
+            last_lane_probe_error.as_deref().unwrap_or("none")
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        result = read_game_creator_agent_runtime_at(root, agent_id)
+            .expect("read runtime while asynchronously waiting for terminal lane release");
+    }
+}
+
+async fn wait_for_agent_runtime_lane_release_async(
+    root: &Path,
+    agent_id: &str,
+) -> AgentRuntimeResult {
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    let mut last_lane_probe_error = None;
+    loop {
+        match game_creator_agent_runtime_task_lock_is_available(root, agent_id) {
+            Ok(true) => {
+                return read_game_creator_agent_runtime_at(root, agent_id)
+                    .expect("read runtime after asynchronous lane release");
+            }
+            Ok(false) => {}
+            Err(error) => last_lane_probe_error = Some(error),
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "Agent lane did not release for {agent_id}; last lane probe error={}",
+            last_lane_probe_error.as_deref().unwrap_or("none")
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
+
 fn wait_for_agent_runtime_phase(root: &Path, agent_id: &str, phase: &str) -> AgentRuntimeState {
    let mut runtime = read_game_creator_agent_runtime_at(root, agent_id)
        .expect("read runtime while waiting for phase")
@@ -780,6 +780,10 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
        "最终回复规范化为空持久重试项目",
    )
    .expect("thinking-only final reply project init");
+    let config_dir = unique_project_path();
+    fs::create_dir_all(&config_dir).expect("create thinking-only final reply config dir");
+    let config_guard = use_test_runtime_config_dir(config_dir.clone());
+    let config_path = config_dir.join(GAME_CREATOR_CONFIG_FILE_NAME);
    let (request_sender, request_receiver) = mpsc::channel();
    let base_url = spawn_mock_llm_server_responses_with_capture(
        vec![
@@ -789,8 +793,10 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
        ],
        Some(request_sender),
    );
-    let _config_guard = write_test_local_config(format!(
-        r#"{{
+    replace_test_local_config(
+        &config_path,
+        format!(
+            r#"{{
  "agentLlm": {{
    "design-director": {{
      "apiKey": "final-reply-empty-normalization-key",
@@ -803,7 +809,8 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
    }}
  }}
 }}"#
-    ));
+        ),
+    );
    let run_id = "provider-retry-final-reply-empty-normalization-run";
    let started = start_game_creator_agent_background_task_at(
        &root,
@@ -813,12 +820,12 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
    )
    .expect("start thinking-only final reply task");

-    request_receiver
-        .recv_timeout(Duration::from_secs(5))
-        .expect("tool-plan Provider request");
-    request_receiver
-        .recv_timeout(Duration::from_secs(5))
-        .expect("thinking-only final-reply Provider request");
+    wait_for_captured_mock_request(&request_receiver, "tool-plan Provider request").await;
+    wait_for_captured_mock_request(
+        &request_receiver,
+        "thinking-only final-reply Provider request",
+    )
+    .await;
    let mut retry = None;
    for _ in 0..250 {
        retry = crate::provider_retry::read_for_run_at(&root, "design-director", run_id)
@@ -829,7 +836,7 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
        {
            break;
        }
-        std::thread::sleep(Duration::from_millis(20));
+        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    let retry = retry.expect("thinking-only final reply must persist retry sidecar");
    assert_eq!(retry.identity.request_kind, "final-reply");
@@ -874,14 +881,20 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
        .expect("force thinking-only final reply retry due");
    resume_game_creator_agent_background_tasks_at(&root)
        .expect("resume thinking-only final reply retry");
-    request_receiver
-        .recv_timeout(Duration::from_secs(5))
-        .expect("recovered final-reply Provider request");
-    assert!(request_receiver
-        .recv_timeout(Duration::from_millis(100))
-        .is_err());
+    wait_for_captured_mock_request(&request_receiver, "recovered final-reply Provider request")
+        .await;
+    tokio::time::sleep(Duration::from_millis(100)).await;
+    assert!(request_receiver.try_recv().is_err());

-    let completed = wait_for_agent_runtime_idle(&root, "design-director");
+    let completed = wait_for_agent_runtime_terminal_and_lane_release_async(
+        &root,
+        "design-director",
+        run_id,
+        "idle",
+        "completed",
+    )
+    .await
+    .state;
    assert_eq!(completed.phase, "completed");
    assert_eq!(completed.last_response.as_deref(), Some(FINAL_RESPONSE));
    assert!(
@@ -894,6 +907,11 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
            .expect("read cleared thinking-only final reply handoff")
            .is_none()
    );
+    assert!(
+        read_game_creator_agent_runtime_finalization_journal(&root, "design-director", run_id)
+            .expect("read cleared thinking-only finalization journal")
+            .is_none()
+    );
    let committed =
        wait_for_response_stream_status(&root, "design-director", run_id, "committed", 1);
    assert_eq!(committed.accumulated_text, FINAL_RESPONSE);
@@ -947,6 +965,8 @@ async fn provider_retry_final_reply_thinking_only_response_retries_before_handof
    assert!(!persisted.contains(PRIVATE_THINKING));

    fs::remove_dir_all(root).ok();
+    drop(config_guard);
+    fs::remove_dir_all(config_dir).ok();
 }

 #[test]
@@ -839,6 +839,10 @@ async fn background_final_reply_failure_keeps_private_conversation_and_hashes_pu
    let root = unique_project_path();
    init_local_game_project_at(&root, "project-1", "最终回复公共失败审计测试")
        .expect("project init");
+    let config_dir = unique_project_path();
+    fs::create_dir_all(&config_dir).expect("create public failure audit config dir");
+    let config_guard = use_test_runtime_config_dir(config_dir.clone());
+    let config_path = config_dir.join(GAME_CREATOR_CONFIG_FILE_NAME);
    let planning_response = serde_json::json!({
        "thinkingSummary": "已有上下文足够，准备生成最终回复",
        "plan": ["回复开发者"],
@@ -847,8 +851,10 @@ async fn background_final_reply_failure_keeps_private_conversation_and_hashes_pu
    })
    .to_string();
    let base_url = spawn_mock_llm_tool_plan_then_invalid_final_reply(planning_response);
-    let _config_guard = write_test_local_config(format!(
-        r#"{{
+    replace_test_local_config(
+        &config_path,
+        format!(
+            r#"{{
  "agentLlm": {{
    "design-director": {{
      "apiKey": "design-key",
@@ -859,7 +865,8 @@ async fn background_final_reply_failure_keeps_private_conversation_and_hashes_pu
    }}
  }}
 }}"#
-    ));
+        ),
+    );
    let run_id = "background-final-reply-public-failure-audit-run";
    start_game_creator_agent_background_task_at(
        &root,
@@ -869,18 +876,15 @@ async fn background_final_reply_failure_keeps_private_conversation_and_hashes_pu
    )
    .expect("start background task");

-    let mut runtime = read_game_creator_agent_runtime_at(&root, "design-director")
-        .expect("read initial runtime")
-        .state;
-    for _ in 0..250 {
-        if runtime.status == "failed" {
-            break;
-        }
-        std::thread::sleep(Duration::from_millis(20));
-        runtime = read_game_creator_agent_runtime_at(&root, "design-director")
-            .expect("read failed runtime")
-            .state;
-    }
+    let runtime = wait_for_agent_runtime_terminal_and_lane_release_async(
+        &root,
+        "design-director",
+        run_id,
+        "failed",
+        "failed",
+    )
+    .await
+    .state;
    assert_eq!(runtime.status, "failed");
    assert_eq!(runtime.phase, "failed");
    let private_error = runtime.error.clone().expect("private runtime error");
@@ -940,6 +944,8 @@ async fn background_final_reply_failure_keeps_private_conversation_and_hashes_pu
    }));

    fs::remove_dir_all(root).ok();
+    drop(config_guard);
+    fs::remove_dir_all(config_dir).ok();
 }

 #[test]