diff --git a/apps/ai-game-creator-shell/src-tauri/src/agent.rs b/apps/ai-game-creator-shell/src-tauri/src/agent.rs index 6643879e7..d421458ae 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/agent.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/agent.rs @@ -1,4 +1,5 @@ use super::*; +use crate::provider_handoff; use crate::provider_retry::{ self, AgentRuntimeProviderRetryIdentity, AgentRuntimeProviderRetryRecord, }; @@ -706,7 +707,7 @@ fn ensure_waiting_provider_retry_projection_at( )); }; if matches!(task.status.as_str(), "completed" | "failed" | "cancelled") { - provider_retry::remove_at(root, agent_id, run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, agent_id, run_id)?; return Ok(None); } if retry.identity.project_id != game_creator_agent_runtime_context_project_id(root)? @@ -778,6 +779,15 @@ fn ensure_waiting_provider_retry_projection_at( read_game_creator_agent_runtime_at(root, agent_id).map(Some) } +fn remove_game_creator_agent_runtime_provider_recovery_at( + root: &Path, + agent_id: &str, + run_id: &str, +) -> Result<(), String> { + provider_handoff::remove_at(root, agent_id, run_id)?; + provider_retry::remove_at(root, agent_id, run_id) +} + pub(crate) fn resume_game_creator_agent_background_tasks_at( root: &Path, ) -> Result, String> { @@ -994,7 +1004,11 @@ pub(crate) fn resume_game_creator_agent_background_tasks_at( } }; if has_durable_control { - if let Err(error) = provider_retry::remove_at(root, &agent_id, &task.run_id) { + if let Err(error) = remove_game_creator_agent_runtime_provider_recovery_at( + root, + &agent_id, + &task.run_id, + ) { resumed.push(mark_waiting_provider_retry_needs_reconciliation_at( root, &agent_id, @@ -1293,7 +1307,9 @@ fn classify_game_creator_agent_runtime_finalization_goal_snapshot_at( journal: &AgentRuntimeFinalizationJournal, state: &AgentRuntimeState, ) -> Result { - if journal.schema_version != AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION { + if journal.schema_version != AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION + && journal.schema_version != AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION + { return Ok(AgentRuntimeFinalizationGoalSnapshotRelation::Matches); } let Some(journal_goal_id) = journal.goal_id.as_deref() else { @@ -1349,7 +1365,9 @@ fn validate_game_creator_agent_runtime_finalization_current_goal_snapshot_at( root: &Path, journal: &AgentRuntimeFinalizationJournal, ) -> Result<(), String> { - if journal.schema_version != AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION { + if journal.schema_version != AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION + && journal.schema_version != AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION + { return Ok(()); } let Some(goal_id) = journal.goal_id.as_deref() else { @@ -1456,7 +1474,11 @@ fn resume_game_creator_agent_finalization_at( } }; if state_reconstructed_from_task - && journal.schema_version == AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION + && matches!( + journal.schema_version.as_str(), + AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION + | AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION + ) { state.plan_revision = journal.plan_revision; state.plan_explanation = journal.plan_explanation.clone(); @@ -1489,7 +1511,7 @@ fn resume_game_creator_agent_finalization_at( journal_revision, current_revision, } if journal.status == AGENT_RUNTIME_FINALIZATION_STATUS_PREPARED && !assistant_exists => { - remove_game_creator_agent_runtime_finalization_journal( + remove_game_creator_agent_runtime_finalization_recovery_sidecars( root, &journal.agent_id, &journal.run_id, @@ -1608,7 +1630,7 @@ fn resume_game_creator_agent_finalization_at( Some("恢复时已丢弃尚未写入会话的最终回复。"), )?; } - remove_game_creator_agent_runtime_finalization_journal( + remove_game_creator_agent_runtime_finalization_recovery_sidecars( root, &journal.agent_id, &journal.run_id, @@ -1661,7 +1683,7 @@ fn resume_game_creator_agent_finalization_at( )? }; if let Some(blocker) = blocker { - remove_game_creator_agent_runtime_finalization_journal( + remove_game_creator_agent_runtime_finalization_recovery_sidecars( root, &journal.agent_id, &journal.run_id, @@ -3428,7 +3450,7 @@ fn resolve_game_creator_agent_finalization_before_cancel_at( return Err("Agent Runtime finalization 已记录 assistant，但会话消息不存在".to_string()); } - remove_game_creator_agent_runtime_finalization_journal( + remove_game_creator_agent_runtime_finalization_recovery_sidecars( root, &journal.agent_id, &journal.run_id, @@ -4104,7 +4126,7 @@ pub(crate) fn append_game_creator_agent_runtime_queued_cancellation( remove_game_creator_agent_runtime_pending_tool_action(root, agent_id, &task.run_id)?; remove_game_creator_agent_runtime_provider_action_batch(root, agent_id, &task.run_id)?; remove_game_creator_agent_runtime_confirmations(root, agent_id, &task.run_id)?; - provider_retry::remove_at(root, agent_id, &task.run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, agent_id, &task.run_id)?; publish_game_creator_agent_delegate_result(root, &cancelled_task, Some(summary)); emit_game_creator_agent_runtime_update(root, agent_id); Ok(()) @@ -6430,6 +6452,9 @@ pub(crate) async fn run_game_creator_agent_background_task_with_context( ); return AgentBackgroundTaskOutcome::WaitingForProviderRetry; } + AgentBackgroundTaskOutcome::WaitingForProviderHandoff => { + return AgentBackgroundTaskOutcome::WaitingForProviderHandoff; + } outcome => return outcome, } } @@ -7053,7 +7078,26 @@ async fn run_game_creator_agent_background_task_pass_with_context( ); } }; - if resumes_final_reply_provider_retry { + let resumes_final_reply_provider_handoff = + match provider_handoff::read_for_run_at(&root, &agent_id, &runtime.run_id) { + Ok(Some(handoff)) => matches!( + handoff.identity.request_kind.as_str(), + "final-reply" | "final-reply-context-compaction" + ), + Ok(None) => false, + Err(error) => { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &format!("读取 final-reply Provider 成功响应交接记录失败：{error}"), + ); + } + }; + let resumes_final_reply_provider_exchange = + resumes_final_reply_provider_retry || resumes_final_reply_provider_handoff; + if resumes_final_reply_provider_exchange { runtime.loop_iteration = u32::try_from(start_loop_index).unwrap_or(u32::MAX); } @@ -7080,7 +7124,7 @@ async fn run_game_creator_agent_background_task_pass_with_context( } 'agent_loop: for loop_index in start_loop_index.. { - if resumes_final_reply_provider_retry { + if resumes_final_reply_provider_exchange { converged = true; break 'agent_loop; } @@ -7390,6 +7434,9 @@ async fn run_game_creator_agent_background_task_pass_with_context( } return AgentBackgroundTaskOutcome::WaitingForProviderRetry; } + Ok(RequestedAgentRuntimeToolPlanOutcome::HandoffPrepared) => { + return AgentBackgroundTaskOutcome::WaitingForProviderHandoff; + } Ok(RequestedAgentRuntimeToolPlanOutcome::Superseded) => { let observation = AgentRuntimeToolObservation { tool: "runtime.provider_retry".to_string(), @@ -9675,6 +9722,9 @@ async fn run_game_creator_agent_background_task_pass_with_context( } return AgentBackgroundTaskOutcome::WaitingForProviderRetry; } + Ok(RequestedAgentRuntimeFinalReplyOutcome::HandoffPrepared) => { + return AgentBackgroundTaskOutcome::WaitingForProviderHandoff; + } Ok(RequestedAgentRuntimeFinalReplyOutcome::Superseded) => { let observation = AgentRuntimeToolObservation { tool: "runtime.provider_retry".to_string(), @@ -9793,11 +9843,6 @@ async fn run_game_creator_agent_background_task_pass_with_context( Ok(AgentBackgroundFinalizationOutcome::Completed(completed)) => { debug_assert_eq!(completed.run_id, runtime.run_id); debug_assert_eq!(completed.session_id, runtime.session_id); - let _ = mark_game_creator_agent_runtime_response_stream_committed_at( - &root, - &completed, - &final_reply, - ); AgentBackgroundTaskOutcome::Finished } Ok(AgentBackgroundFinalizationOutcome::Cancelled(cancelled)) => { @@ -9806,6 +9851,15 @@ async fn run_game_creator_agent_background_task_pass_with_context( AgentBackgroundTaskOutcome::Finished } Ok(AgentBackgroundFinalizationOutcome::Stale(blocker)) => { + if let Err(error) = provider_handoff::remove_at(&root, &agent_id, &runtime.run_id) { + return fail_game_creator_agent_background_context_at( + &root, + &agent_id, + &session_id, + runtime, + &format!("删除过期最终回复的 Provider 成功响应交接失败：{error}"), + ); + } let continuation = match prepare_game_creator_agent_background_stale_continuation_at( &root, &mut runtime, @@ -9858,6 +9912,8 @@ async fn run_game_creator_agent_background_task_pass_with_context( pub(crate) const AGENT_RUNTIME_BACKGROUND_LOOP_LIMIT: usize = 6; pub(crate) const AGENT_RUNTIME_BACKGROUND_TOOL_ACTION_LIMIT: usize = 3; pub(crate) const AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION: &str = + "game-creator-runtime-finalization.v4"; +const AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION: &str = "game-creator-runtime-finalization.v3"; const AGENT_RUNTIME_FINALIZATION_LEGACY_SCHEMA_VERSION: &str = "game-creator-runtime-finalization.v2"; @@ -9874,6 +9930,8 @@ const AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX: &str = "provider-request-needs-reconciliation"; const AGENT_RUNTIME_PROVIDER_TRANSIENT_ERROR_PREFIX: &str = "agent-runtime-provider-transient-error:"; +#[cfg(test)] +const AGENT_RUNTIME_PROVIDER_HANDOFF_TEST_STOP: &str = "agent-runtime-provider-handoff-test-stop"; const AGENT_RUNTIME_PROVIDER_TRANSIENT_RETRY_LIMIT: u32 = 3; const AGENT_RUNTIME_PROVIDER_TRANSIENT_BACKOFF_MAX_MS: u64 = 30_000; const AGENT_RUNTIME_FINALIZATION_LIFECYCLE_RECORD_TYPE: &str = @@ -9929,6 +9987,7 @@ pub(crate) enum AgentBackgroundTaskOutcome { WaitingForConfirmation, WaitingForUserInput, WaitingForProviderRetry, + WaitingForProviderHandoff, WaitingForIsolatedJoin, WaitingForDelegateReceipts, NeedsReconciliation, @@ -9942,24 +10001,28 @@ pub(crate) enum AgentBackgroundTaskOutcome { enum AgentRuntimePersistedProviderRequestOutcome { Response(Option), Waiting(AgentRuntimeProviderRetryRecord), + HandoffPrepared, Superseded, } enum AgentRuntimeContextCompactionOutcome { Completed(Option), Waiting(AgentRuntimeProviderRetryRecord), + HandoffPrepared, Superseded, } enum RequestedAgentRuntimeToolPlanOutcome { Ready(Option), Waiting(AgentRuntimeProviderRetryRecord), + HandoffPrepared, Superseded, } enum RequestedAgentRuntimeFinalReplyOutcome { Ready(Option), Waiting(AgentRuntimeProviderRetryRecord), + HandoffPrepared, Superseded, } @@ -10007,6 +10070,7 @@ pub(crate) enum AgentRuntimeFinalizationCheckpoint { Prepared, AssistantAppended, RuntimeCompleted, + ResponseStreamCommitted, } #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] @@ -10221,6 +10285,8 @@ pub(crate) struct AgentRuntimeFinalizationJournal { pub(crate) response_fingerprint: String, pub(crate) response_revision: u64, #[serde(default)] + pub(crate) response_request_slot: String, + #[serde(default)] pub(crate) response_steer_cursor: u64, #[serde(default)] pub(crate) plan_revision: u64, @@ -10968,7 +11034,7 @@ pub(crate) fn steer_game_creator_agent_runtime_task_at( } let was_waiting_for_provider_retry = state.phase == "waiting-for-provider-retry"; if was_waiting_for_provider_retry { - provider_retry::remove_at(root, &agent_id, run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, &agent_id, run_id)?; let mut wake_state = state.clone(); wake_state.status = "running".to_string(); wake_state.phase = "planning".to_string(); @@ -11280,7 +11346,11 @@ pub(crate) fn consume_game_creator_agent_runtime_steers( } verify_game_creator_agent_runtime_steer_conversation(root, runtime, entry)?; } - provider_retry::remove_at(root, &runtime.agent_id, &runtime.run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at( + root, + &runtime.agent_id, + &runtime.run_id, + )?; let previous_cursor = runtime.applied_steer_cursor; let mut refs = snapshot .entries @@ -11888,6 +11958,31 @@ where snapshot, request, || {}, + |_, _| Ok(()), + ) + .await +} + +async fn await_game_creator_agent_runtime_provider_request_with_snapshot_and_success_commit< + T, + F, + C, +>( + root: &Path, + snapshot: AgentRuntimeProviderRequestSnapshot, + request: F, + success_commit: C, +) -> Result, String> +where + F: std::future::Future>, + C: FnOnce(&str, &T) -> Result<(), String>, +{ + await_game_creator_agent_runtime_provider_request_with_snapshot_and_control_recheck( + root, + snapshot, + request, + || {}, + success_commit, ) .await } @@ -11926,6 +12021,7 @@ where snapshot, request, before_control_recheck, + |_, _| Ok(()), ) .await } @@ -11950,6 +12046,7 @@ where snapshot, request, before_control_recheck, + |_, _| Ok(()), ) .await } @@ -11958,15 +12055,18 @@ async fn await_game_creator_agent_runtime_provider_request_with_snapshot_and_con T, F, H, + C, >( root: &Path, snapshot: AgentRuntimeProviderRequestSnapshot, request: F, before_control_recheck: H, + success_commit: C, ) -> Result, String> where F: std::future::Future>, H: FnOnce(), + C: FnOnce(&str, &T) -> Result<(), String>, { let base_request_id = game_creator_agent_runtime_provider_request_id(&snapshot); let (key, active) = register_game_creator_agent_runtime_provider_request( @@ -12088,8 +12188,39 @@ where result = &mut request => result.map(Some), } }; - unregister_game_creator_agent_runtime_provider_request(&key, &active); let result = result.map_err(|error| redact_agent_runtime_error(root, &error, 500)); + if let Ok(Some(response)) = result.as_ref() { + if let Err(error) = success_commit(&request_id, response) { + unregister_game_creator_agent_runtime_provider_request(&key, &active); + if let Ok(_control_lock) = + acquire_game_creator_agent_runtime_project_write_lock_with_wait( + root, + "runtime.provider_request.success_handoff_reconciliation", + ) + { + let _ = + mark_game_creator_agent_runtime_provider_request_needs_reconciliation_at_locked( + root, + &snapshot, + &request_id, + ); + } + let error = redact_agent_runtime_error(root, &error, 500); + return Err(format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: requestId={request_id} · successHandoff={error}" + )); + } + #[cfg(test)] + if provider_handoff::supports_request_kind(&snapshot.request_kind) { + let injection = root.join(".agent/runtime/test-stop-after-provider-handoff"); + if injection.exists() { + let _ = fs::remove_file(injection); + unregister_game_creator_agent_runtime_provider_request(&key, &active); + return Err(AGENT_RUNTIME_PROVIDER_HANDOFF_TEST_STOP.to_string()); + } + } + } + unregister_game_creator_agent_runtime_provider_request(&key, &active); let status = match &result { Ok(Some(_)) => "completed", Ok(None) => "interrupted", @@ -12345,7 +12476,78 @@ fn game_creator_agent_runtime_provider_retry_drift_fields( fields } -async fn request_game_creator_agent_runtime_llm_with_persisted_transient_retry_using( +fn game_creator_agent_runtime_provider_handoff_reconciliation_error( + root: &Path, + snapshot: &AgentRuntimeProviderRequestSnapshot, + error: &str, +) -> String { + if let Ok(_control_lock) = acquire_game_creator_agent_runtime_project_write_lock_with_wait( + root, + "runtime.provider_handoff.reconciliation", + ) { + let base_request_id = game_creator_agent_runtime_provider_request_id(snapshot); + let request_id = resolve_game_creator_agent_runtime_provider_request_attempt_at_locked( + root, + &base_request_id, + ) + .map(|value| value.0) + .unwrap_or(base_request_id); + let _ = mark_game_creator_agent_runtime_provider_request_needs_reconciliation_at_locked( + root, + snapshot, + &request_id, + ); + } + format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: Provider 成功响应交接失败：{}", + redact_agent_runtime_error(root, error, 320) + ) +} + +fn repair_game_creator_agent_runtime_provider_handoff_lifecycle_at_locked( + root: &Path, + provider_snapshot: &AgentRuntimeProviderRequestSnapshot, + handoff: &provider_handoff::AgentRuntimeProviderHandoffRecord, +) -> Result<(), String> { + let attempt_snapshot = provider_snapshot.with_request_slot(handoff.request_slot.clone()); + let request_id = handoff.provider_request_id.as_str(); + let base_request_id = game_creator_agent_runtime_provider_request_id(&attempt_snapshot); + if !(0..=64).any(|attempt| { + game_creator_agent_runtime_provider_request_attempt_id(&base_request_id, attempt) + == request_id + }) { + return Err("Provider 成功响应交接 requestId 不属于当前 requestSlot".to_string()); + } + let incomplete = read_agent_db_incomplete_provider_request_ids_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; + if incomplete + .iter() + .any(|incomplete_id| incomplete_id != request_id) + { + return Err("Provider 成功响应交接命中其它未闭合 Provider 请求".to_string()); + } + let transitions = read_agent_db_lifecycle_transitions_at( + root, + AGENT_RUNTIME_PROVIDER_REQUEST_LIFECYCLE_RECORD_TYPE, + "requestId", + request_id, + )?; + if transitions != ["started"] && transitions != ["started", "completed"] { + return Err("Provider 成功响应交接缺少匹配的 started lifecycle".to_string()); + } + append_game_creator_agent_runtime_provider_request_lifecycle( + root, + &attempt_snapshot, + request_id, + "completed", + )?; + Ok(()) +} + +async fn request_game_creator_agent_runtime_llm_with_persisted_transient_retry_using( root: &Path, provider_snapshot: &AgentRuntimeProviderRequestSnapshot, llm: &GameCreatorLlmConfig, @@ -12353,16 +12555,183 @@ async fn request_game_creator_agent_runtime_llm_with_persisted_transient_retry_u operation: &str, request: &LlmRunRequest, execute: F, + canonicalize_handoff_response: H, ) -> Result where F: FnOnce(LlmClient, LlmRunRequest) -> Fut, Fut: std::future::Future>, + H: FnOnce(&platform_llm::LlmRunResponse) -> platform_llm::LlmRunResponse, { let identity = game_creator_agent_runtime_provider_retry_identity(provider_snapshot, llm, request)?; let max_retries = llm .max_retries .min(AGENT_RUNTIME_PROVIDER_TRANSIENT_RETRY_LIMIT); + if provider_handoff::supports_request_kind(&identity.request_kind) { + if let Some(handoff) = provider_handoff::read_for_run_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )? { + if handoff.identity != identity { + if !game_creator_agent_runtime_provider_retry_same_durable_run( + &handoff.identity, + &identity, + ) { + return Err("Provider 成功响应交接记录与当前持久 run 身份冲突".to_string()); + } + let handoff_snapshot = + game_creator_agent_runtime_provider_snapshot_from_retry_identity( + &handoff.identity, + ); + let control_lock = acquire_game_creator_agent_runtime_project_write_lock_with_wait( + root, + "runtime.provider_handoff.superseded_lifecycle", + )?; + if let Err(error) = + repair_game_creator_agent_runtime_provider_handoff_lifecycle_at_locked( + root, + &handoff_snapshot, + &handoff, + ) + { + let attempt_snapshot = + handoff_snapshot.with_request_slot(handoff.request_slot.clone()); + let _ = + mark_game_creator_agent_runtime_provider_request_needs_reconciliation_at_locked( + root, + &attempt_snapshot, + &handoff.provider_request_id, + ); + drop(control_lock); + return Err(format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: Provider 成功响应交接作废前 lifecycle 恢复失败：{}", + redact_agent_runtime_error(root, &error, 320) + )); + } + drop(control_lock); + let drift_fields = game_creator_agent_runtime_provider_retry_drift_fields( + &handoff.identity, + &identity, + ); + let _ = append_agent_db_record( + root, + serde_json::json!({ + "recordType": "agent.runtime.provider_request.handoff_superseded", + "agentId": provider_snapshot.agent_id, + "taskId": provider_snapshot.task_id, + "sessionId": provider_snapshot.session_id, + "runId": provider_snapshot.run_id, + "source": provider_snapshot.source, + "requestKind": handoff.identity.request_kind, + "driftFields": drift_fields, + }), + ); + provider_handoff::remove_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; + provider_retry::remove_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; + return Ok(AgentRuntimePersistedProviderRequestOutcome::Superseded); + } + if let Some(retry) = provider_retry::read_for_run_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )? { + if retry.identity != handoff.identity + || retry.next_attempt != handoff.attempt + || retry.next_request_slot != handoff.request_slot + { + let handoff_snapshot = + provider_snapshot.with_request_slot(handoff.request_slot.clone()); + if let Ok(_control_lock) = + acquire_game_creator_agent_runtime_project_write_lock_with_wait( + root, + "runtime.provider_handoff.retry_conflict_reconciliation", + ) + { + let _ = + mark_game_creator_agent_runtime_provider_request_needs_reconciliation_at_locked( + root, + &handoff_snapshot, + &handoff.provider_request_id, + ); + } + return Err(format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: Provider 成功响应交接与 retry sidecar 冲突" + )); + } + } + let control_lock = acquire_game_creator_agent_runtime_project_write_lock_with_wait( + root, + "runtime.provider_handoff.replay", + )?; + if let Err(error) = + repair_game_creator_agent_runtime_provider_handoff_lifecycle_at_locked( + root, + provider_snapshot, + &handoff, + ) + { + let handoff_snapshot = + provider_snapshot.with_request_slot(handoff.request_slot.clone()); + let _ = + mark_game_creator_agent_runtime_provider_request_needs_reconciliation_at_locked( + root, + &handoff_snapshot, + &handoff.provider_request_id, + ); + drop(control_lock); + return Err(format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: Provider 成功响应交接 lifecycle 恢复失败：{}", + redact_agent_runtime_error(root, &error, 320) + )); + } + let has_durable_control = + game_creator_agent_runtime_provider_snapshot_has_durable_control_at_locked( + root, + provider_snapshot, + )?; + let pausing = has_durable_control + && game_creator_agent_runtime_provider_snapshot_is_pausing_at_locked( + root, + provider_snapshot, + )?; + if has_durable_control && !pausing { + provider_handoff::remove_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; + provider_retry::remove_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; + drop(control_lock); + return Ok(AgentRuntimePersistedProviderRequestOutcome::Response(None)); + } + if pausing { + drop(control_lock); + return Ok(AgentRuntimePersistedProviderRequestOutcome::Response(None)); + } + provider_retry::remove_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; + drop(control_lock); + return Ok(AgentRuntimePersistedProviderRequestOutcome::Response(Some( + handoff.to_llm_response(), + ))); + } + } let existing = provider_retry::read_for_run_at( root, &provider_snapshot.agent_id, @@ -12396,6 +12765,11 @@ where &provider_snapshot.agent_id, &provider_snapshot.run_id, )?; + provider_handoff::remove_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + )?; return Ok(AgentRuntimePersistedProviderRequestOutcome::Superseded); } if provider_retry::remaining_ms(record) > 0 { @@ -12432,10 +12806,27 @@ where ) }) }; - match await_game_creator_agent_runtime_provider_request_with_snapshot( + let handoff_identity = identity.clone(); + let handoff_request_slot = attempt_snapshot.request_slot.clone(); + let persist_handoff = provider_handoff::supports_request_kind(&identity.request_kind); + match await_game_creator_agent_runtime_provider_request_with_snapshot_and_success_commit( root, attempt_snapshot.clone(), provider_request, + |provider_request_id, response| { + if persist_handoff { + let response = canonicalize_handoff_response(response); + provider_handoff::write_at( + root, + &handoff_identity, + &handoff_request_slot, + attempt, + provider_request_id, + &response, + )?; + } + Ok(()) + }, ) .await { @@ -12458,6 +12849,39 @@ where response, )); } + let response = if persist_handoff && response.is_some() { + let handoff = provider_handoff::read_for_run_at( + root, + &provider_snapshot.agent_id, + &provider_snapshot.run_id, + ) + .map_err(|error| { + game_creator_agent_runtime_provider_handoff_reconciliation_error( + root, + &attempt_snapshot, + &error, + ) + })? + .ok_or_else(|| { + game_creator_agent_runtime_provider_handoff_reconciliation_error( + root, + &attempt_snapshot, + "Provider lifecycle completed 前缺少成功响应交接记录", + ) + })?; + if handoff.identity != identity { + return Err( + game_creator_agent_runtime_provider_handoff_reconciliation_error( + root, + &attempt_snapshot, + "Provider lifecycle completed 前成功响应交接身份已变化", + ), + ); + } + Some(handoff.to_llm_response()) + } else { + response + }; if let Err(error) = provider_retry::remove_at( root, &provider_snapshot.agent_id, @@ -12494,6 +12918,10 @@ where )) } Err(error) => { + #[cfg(test)] + if error == AGENT_RUNTIME_PROVIDER_HANDOFF_TEST_STOP { + return Ok(AgentRuntimePersistedProviderRequestOutcome::HandoffPrepared); + } let Some(encoded) = error.strip_prefix(AGENT_RUNTIME_PROVIDER_TRANSIENT_ERROR_PREFIX) else { provider_retry::remove_at( @@ -12590,6 +13018,7 @@ async fn request_game_creator_agent_runtime_llm_with_persisted_transient_retry( operation, request, |client, request| async move { client.run(request).await }, + |response| response.clone(), ) .await } @@ -13420,32 +13849,127 @@ fn visible_game_creator_agent_runtime_response_stream_at( fn mark_game_creator_agent_runtime_response_stream_committed_at( root: &Path, state: &AgentRuntimeState, - response: &str, + journal: &AgentRuntimeFinalizationJournal, ) -> Result<(), String> { - let Some(mut stream) = - read_game_creator_agent_runtime_response_stream_at(root, &state.agent_id, &state.run_id)? - else { - return Ok(()); - }; - if stream.task_id != state.task_id - || stream.session_id != state.session_id - || stream.applied_steer_cursor != state.applied_steer_cursor - || stream.response_revision - != read_game_creator_agent_runtime_project_revision(root)?.revision - || stream.request_slot - != game_creator_agent_runtime_response_stream_request_slot( - state, - stream.response_revision, - ) - || stream.status != AGENT_RUNTIME_RESPONSE_STREAM_STATUS_READY - || stream.accumulated_text != response + if journal.agent_id != state.agent_id + || journal.task_id != state.task_id + || journal.session_id != state.session_id + || journal.run_id != state.run_id + || journal.response_steer_cursor != state.applied_steer_cursor { - return Ok(()); + return Err("Agent Runtime finalization 与回复流提交身份不匹配".to_string()); + } + let request_slot = if journal.response_request_slot.trim().is_empty() { + game_creator_agent_runtime_response_stream_request_slot(state, journal.response_revision) + } else { + journal.response_request_slot.clone() + }; + let stream_matches = |stream: &AgentRuntimeResponseStream| { + stream.agent_id == journal.agent_id + && stream.task_id == journal.task_id + && stream.session_id == journal.session_id + && stream.run_id == journal.run_id + && stream.request_kind == "final-reply" + && stream.request_slot == request_slot + && stream.applied_steer_cursor == journal.response_steer_cursor + && stream.response_revision == journal.response_revision + }; + let existing = read_game_creator_agent_runtime_response_stream_at( + root, + &journal.agent_id, + &journal.run_id, + )?; + if existing + .as_ref() + .is_some_and(|stream| !stream_matches(stream)) + { + return Err("Agent Runtime 回复流与 finalization 固定身份冲突".to_string()); + } + let finish_reason = existing + .as_ref() + .and_then(|stream| stream.finish_reason.clone()); + let needs_ready_rebuild = existing.as_ref().is_none_or(|stream| { + !matches!( + stream.status.as_str(), + AGENT_RUNTIME_RESPONSE_STREAM_STATUS_READY + | AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED + ) + }); + if needs_ready_rebuild { + let snapshot = AgentRuntimeProviderRequestSnapshot { + project_id: journal.project_id.clone(), + agent_id: journal.agent_id.clone(), + task_id: journal.task_id.clone(), + session_id: journal.session_id.clone(), + run_id: journal.run_id.clone(), + source: journal.source.clone(), + goal_id: journal.goal_id.clone(), + goal_revision: journal.goal_revision, + goal_snapshot_fingerprint: journal.goal_snapshot_fingerprint.clone(), + applied_steer_cursor: journal.response_steer_cursor, + request_kind: "final-reply".to_string(), + request_slot: request_slot.clone(), + web_search_enabled: false, + allow_idle_context_compaction: false, + }; + write_game_creator_agent_runtime_response_stream_ready_at( + root, + &snapshot, + journal.response_revision, + &journal.response, + finish_reason.as_deref(), + )?; + } + let mut stream = read_game_creator_agent_runtime_response_stream_at( + root, + &journal.agent_id, + &journal.run_id, + )? + .ok_or_else(|| "Agent Runtime 回复流重建后不存在".to_string())?; + if !stream_matches(&stream) { + return Err("Agent Runtime 回复流重建后身份冲突".to_string()); + } + if stream.status == AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED { + return if stream.accumulated_text == journal.response { + Ok(()) + } else { + Err("Agent Runtime 已提交回复流与 finalization 回复冲突".to_string()) + }; + } + if stream.status != AGENT_RUNTIME_RESPONSE_STREAM_STATUS_READY { + return Err(format!( + "Agent Runtime finalization 需要 ready 回复流，当前状态为 {}", + stream.status + )); + } + if stream.accumulated_text != journal.response { + return Err("Agent Runtime ready 回复流与 finalization 回复冲突".to_string()); } stream.status = AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED.to_string(); stream.sequence = stream.sequence.saturating_add(1); stream.updated_at = unix_timestamp(); - write_game_creator_agent_runtime_response_stream_at(root, &stream) + #[cfg(test)] + { + let injection = root.join(".agent/runtime/test-fail-response-stream-commit"); + if injection.exists() { + let _ = fs::remove_file(injection); + return Err("injected-response-stream-commit-failure".to_string()); + } + } + write_game_creator_agent_runtime_response_stream_at(root, &stream)?; + let committed = read_game_creator_agent_runtime_response_stream_at( + root, + &journal.agent_id, + &journal.run_id, + )? + .ok_or_else(|| "Agent Runtime 回复流提交后不存在".to_string())?; + if !stream_matches(&committed) + || committed.status != AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED + || committed.accumulated_text != journal.response + { + return Err("Agent Runtime 回复流提交后回读校验失败".to_string()); + } + Ok(()) } pub(crate) fn game_creator_agent_runtime_project_revision_path(root: &Path) -> PathBuf { @@ -13718,11 +14242,16 @@ fn game_creator_agent_runtime_finalization_id( run_id: &str, response_fingerprint: &str, response_revision: u64, + response_request_slot: &str, response_steer_cursor: u64, plan_snapshot_fingerprint: &str, goal_snapshot_fingerprint: &str, ) -> String { - let payload = if !goal_snapshot_fingerprint.is_empty() { + let payload = if !response_request_slot.is_empty() { + format!( + "{project_id}\n{agent_id}\n{session_id}\n{run_id}\n{response_fingerprint}\n{response_revision}\n{response_request_slot}\n{response_steer_cursor}\n{plan_snapshot_fingerprint}\n{goal_snapshot_fingerprint}" + ) + } else if !goal_snapshot_fingerprint.is_empty() { format!( "{project_id}\n{agent_id}\n{session_id}\n{run_id}\n{response_fingerprint}\n{response_revision}\n{response_steer_cursor}\n{plan_snapshot_fingerprint}\n{goal_snapshot_fingerprint}" ) @@ -13786,6 +14315,8 @@ fn build_game_creator_agent_runtime_finalization_journal( state.active_plan_step_index, ); let goal_snapshot_fingerprint = agent_goal_snapshot_fingerprint_for_state_at(root, state)?; + let response_request_slot = + game_creator_agent_runtime_response_stream_request_slot(state, response_revision); let finalization_id = game_creator_agent_runtime_finalization_id( &project_id, &state.agent_id, @@ -13793,6 +14324,7 @@ fn build_game_creator_agent_runtime_finalization_journal( &state.run_id, &response_fingerprint, response_revision, + &response_request_slot, state.applied_steer_cursor, &plan_snapshot_fingerprint, &goal_snapshot_fingerprint, @@ -13821,6 +14353,7 @@ fn build_game_creator_agent_runtime_finalization_journal( response: response.to_string(), response_fingerprint, response_revision, + response_request_slot, response_steer_cursor: state.applied_steer_cursor, plan_revision: state.plan_revision, plan_explanation: state.plan_explanation.clone(), @@ -13858,7 +14391,10 @@ fn validate_game_creator_agent_runtime_finalization_journal( ) -> Result<(), String> { let legacy_schema = journal.schema_version == AGENT_RUNTIME_FINALIZATION_LEGACY_SCHEMA_VERSION; let older_schema = journal.schema_version == AGENT_RUNTIME_FINALIZATION_OLDER_SCHEMA_VERSION; + let previous_schema = + journal.schema_version == AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION; if journal.schema_version != AGENT_RUNTIME_FINALIZATION_SCHEMA_VERSION + && !previous_schema && !legacy_schema && !older_schema { @@ -13886,6 +14422,18 @@ fn validate_game_creator_agent_runtime_finalization_journal( if journal.response_fingerprint != response_fingerprint { return Err("Agent Runtime finalization 回复指纹不匹配".to_string()); } + if !previous_schema + && !legacy_schema + && !older_schema + && (journal.response_request_slot.trim().is_empty() + || journal.response_request_slot.chars().count() > 256 + || journal.response_request_slot.chars().any(char::is_control) + || !journal + .response_request_slot + .starts_with("final-reply-loop-")) + { + return Err("Agent Runtime finalization 回复流 requestSlot 无效".to_string()); + } if older_schema { if journal.plan_revision != 0 || !journal.plan_explanation.is_empty() @@ -13958,6 +14506,7 @@ fn validate_game_creator_agent_runtime_finalization_journal( &journal.run_id, &journal.response_fingerprint, journal.response_revision, + &journal.response_request_slot, journal.response_steer_cursor, &journal.plan_snapshot_fingerprint, &journal.goal_snapshot_fingerprint, @@ -14129,6 +14678,15 @@ pub(crate) fn remove_game_creator_agent_runtime_finalization_journal( } } +fn remove_game_creator_agent_runtime_finalization_recovery_sidecars( + root: &Path, + agent_id: &str, + run_id: &str, +) -> Result<(), String> { + provider_handoff::remove_at(root, agent_id, run_id)?; + remove_game_creator_agent_runtime_finalization_journal(root, agent_id, run_id) +} + pub(crate) fn game_creator_agent_runtime_context_bundle_path( root: &Path, agent_id: &str, @@ -19804,7 +20362,22 @@ async fn compact_game_creator_agent_runtime_context_at( observations, trigger, )?; + let request_slot = format!( + "source-{}", + source + .source_fingerprint + .chars() + .take(32) + .collect::() + ); if !source.has_new_source { + provider_handoff::remove_consumed_context_at( + root, + agent_id, + run_id, + request_kind, + &request_slot, + )?; let previous = source .previous .as_ref() @@ -19824,14 +20397,6 @@ async fn compact_game_creator_agent_runtime_context_at( estimated_request_tokens, "上下文压缩请求", )?; - let request_slot = format!( - "source-{}", - source - .source_fingerprint - .chars() - .take(32) - .collect::() - ); let snapshot = if allow_idle_context_compaction { capture_idle_game_creator_agent_runtime_context_compaction_snapshot_at_locked( root, @@ -19854,6 +20419,8 @@ async fn compact_game_creator_agent_runtime_context_at( }; (snapshot, source, llm, config_path, request) }; + let handoff_identity = + game_creator_agent_runtime_provider_retry_identity(&snapshot, &llm, &request)?; let response = if !persist_transient_retry { AgentRuntimePersistedProviderRequestOutcome::Response( request_game_creator_agent_runtime_llm_with_transient_retries( @@ -19885,6 +20452,9 @@ async fn compact_game_creator_agent_runtime_context_at( AgentRuntimePersistedProviderRequestOutcome::Waiting(record) => { return Ok(AgentRuntimeContextCompactionOutcome::Waiting(record)); } + AgentRuntimePersistedProviderRequestOutcome::HandoffPrepared => { + return Ok(AgentRuntimeContextCompactionOutcome::HandoffPrepared); + } AgentRuntimePersistedProviderRequestOutcome::Superseded => { return Ok(AgentRuntimeContextCompactionOutcome::Superseded); } @@ -19949,6 +20519,39 @@ async fn compact_game_creator_agent_runtime_context_at( redact_agent_runtime_error(root, &error, 320) )); } + let persisted_sidecar = read_game_creator_agent_runtime_context_compaction( + root, + &sidecar.agent_id, + &sidecar.session_id, + )? + .ok_or_else(|| { + format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: context compaction 写入后不存在" + ) + })?; + if persisted_sidecar != sidecar { + return Err(format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: context compaction 写入后内容冲突" + )); + } + if let Err(error) = provider_handoff::remove_matching_at(root, &handoff_identity) { + let base_request_id = game_creator_agent_runtime_provider_request_id(&snapshot); + let request_id = resolve_game_creator_agent_runtime_provider_request_attempt_at_locked( + root, + &base_request_id, + ) + .map(|value| value.0) + .unwrap_or(base_request_id); + let _ = mark_game_creator_agent_runtime_provider_request_needs_reconciliation_at_locked( + root, + &snapshot, + &request_id, + ); + return Err(format!( + "{AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX}: {}", + redact_agent_runtime_error(root, &error, 320) + )); + } drop(control_lock); if let Ok(runtime) = @@ -20068,6 +20671,7 @@ pub(crate) async fn compact_game_creator_agent_runtime_session_at( return Err("手动上下文压缩被新的控制指令中断".to_string()); } AgentRuntimeContextCompactionOutcome::Waiting(_) + | AgentRuntimeContextCompactionOutcome::HandoffPrepared | AgentRuntimeContextCompactionOutcome::Superseded => { return Err("手动上下文压缩不应进入后台 Provider 重试等待".to_string()); } @@ -20152,6 +20756,9 @@ async fn request_game_creator_agent_background_tool_plan_at( AgentRuntimeContextCompactionOutcome::Waiting(record) => { return Ok(RequestedAgentRuntimeToolPlanOutcome::Waiting(record)); } + AgentRuntimeContextCompactionOutcome::HandoffPrepared => { + return Ok(RequestedAgentRuntimeToolPlanOutcome::HandoffPrepared); + } AgentRuntimeContextCompactionOutcome::Superseded => { return Ok(RequestedAgentRuntimeToolPlanOutcome::Superseded); } @@ -20263,6 +20870,9 @@ async fn request_game_creator_agent_background_tool_plan_at( AgentRuntimePersistedProviderRequestOutcome::Waiting(record) => { return Ok(RequestedAgentRuntimeToolPlanOutcome::Waiting(record)); } + AgentRuntimePersistedProviderRequestOutcome::HandoffPrepared => { + return Ok(RequestedAgentRuntimeToolPlanOutcome::HandoffPrepared); + } AgentRuntimePersistedProviderRequestOutcome::Superseded => { return Ok(RequestedAgentRuntimeToolPlanOutcome::Superseded); } @@ -20627,6 +21237,11 @@ impl AgentRuntimeResponseStreamPublisher { self.finish_with_status(AGENT_RUNTIME_RESPONSE_STREAM_STATUS_FAILED); } + fn handoff(&mut self) { + self.persist(); + self.terminal = true; + } + fn finish_with_status(&mut self, status: &str) { self.stream.status = status.to_string(); self.stream.sequence = self.stream.sequence.saturating_add(1); @@ -21076,7 +21691,7 @@ mod response_stream_tests { .expect("ready response stream exists before finalization"); assert_eq!(ready.sequence, 1); - let completed = match finish_game_creator_agent_background_runtime_turn_at( + let _completed = match finish_game_creator_agent_background_runtime_turn_at( root, state.clone(), response, @@ -21099,9 +21714,6 @@ mod response_stream_tests { panic!("response stream finalization was cancelled") } }; - mark_game_creator_agent_runtime_response_stream_committed_at(root, &completed, response) - .expect("mark finalized response stream committed"); - let committed = read_game_creator_agent_runtime_response_stream_at( root, &state.agent_id, @@ -21140,6 +21752,504 @@ mod response_stream_tests { .collect::>(); assert_eq!(assistants, vec![response]); } + + fn resume_response_stream_finalization(root: &Path, agent_id: &str) -> AgentRuntimeState { + let runtime_lock = acquire_game_creator_agent_runtime_task_lock_with_wait(root, agent_id) + .expect("acquire finalization recovery lane"); + match resume_game_creator_agent_finalization_at(root, agent_id, runtime_lock) + .expect("resume response stream finalization") + { + AgentRuntimeFinalizationResume::Recovered(result, _runtime_lock) => result.state, + AgentRuntimeFinalizationResume::Blocked(result) => { + panic!( + "response stream finalization was blocked: {:?}", + result.state.error + ) + } + AgentRuntimeFinalizationResume::NotFound(_runtime_lock) => { + panic!("response stream finalization journal was not found") + } + } + } + + #[test] + fn response_stream_finalization_recovers_missing_stream_after_project_revision_drift() { + let (project, state, response_revision, _snapshot) = + response_stream_fixture("response-stream-missing-recovery-run"); + let root = project.path(); + let response = "缺失的回复流由 finalization journal 重建。"; + let outcome = finish_game_creator_agent_background_runtime_turn_with_checkpoint_at( + root, + state.clone(), + response, + response_revision, + &[], + |checkpoint| { + if checkpoint == AgentRuntimeFinalizationCheckpoint::RuntimeCompleted { + Err("injected-before-response-stream-commit".to_string()) + } else { + Ok(()) + } + }, + ) + .expect("inject finalization interruption before stream commit"); + assert!(matches!( + outcome, + AgentBackgroundFinalizationOutcome::Pending(_) + )); + assert!(read_game_creator_agent_runtime_response_stream_at( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read absent response stream before recovery") + .is_none()); + + let mut revision = read_game_creator_agent_runtime_project_revision(root) + .expect("read revision before concurrent drift"); + revision.revision = revision.revision.saturating_add(1); + revision.updated_at = unix_timestamp(); + write_game_creator_agent_runtime_project_revision(root, &revision) + .expect("simulate another Agent advancing project revision"); + + let recovered = resume_response_stream_finalization(root, &state.agent_id); + assert_eq!(recovered.phase, "completed"); + let stream = read_game_creator_agent_runtime_response_stream_at( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read rebuilt response stream") + .expect("rebuilt response stream exists"); + assert_eq!( + stream.status, + AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED + ); + assert_eq!(stream.response_revision, response_revision); + assert_eq!(stream.accumulated_text, response); + assert!(read_game_creator_agent_runtime_finalization_journal( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read cleaned finalization journal") + .is_none()); + } + + #[test] + fn response_stream_finalization_repairs_streaming_after_commit_write_failure() { + let (project, state, response_revision, snapshot) = + response_stream_fixture("response-stream-write-failure-recovery-run"); + let root = project.path(); + let response = "提交写失败后从 journal 恢复唯一正文。"; + let mut publisher = + AgentRuntimeResponseStreamPublisher::start(root, &snapshot, response_revision); + publisher.push(&stream_delta("未完成半句", "未完成半句")); + publisher.handoff(); + let injection = root.join(".agent/runtime/test-fail-response-stream-commit"); + std::fs::write(&injection, b"fail-once").expect("write stream commit failure injection"); + + let outcome = finish_game_creator_agent_background_runtime_turn_at( + root, + state.clone(), + response, + response_revision, + &[], + ) + .expect("finalization commit failure remains recoverable"); + assert!(matches!( + outcome, + AgentBackgroundFinalizationOutcome::Pending(ref error) + if error.contains("injected-response-stream-commit-failure") + )); + let ready = read_game_creator_agent_runtime_response_stream_at( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read repaired ready stream") + .expect("repaired ready stream exists"); + assert_eq!(ready.status, AGENT_RUNTIME_RESPONSE_STREAM_STATUS_READY); + assert_eq!(ready.accumulated_text, response); + let journal = read_game_creator_agent_runtime_finalization_journal( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read retained finalization journal") + .expect("finalization journal retained after stream write failure"); + assert_eq!( + journal.status, + AGENT_RUNTIME_FINALIZATION_STATUS_RUNTIME_COMPLETED + ); + + resume_response_stream_finalization(root, &state.agent_id); + let committed = read_game_creator_agent_runtime_response_stream_at( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read committed stream after repair") + .expect("committed stream exists after repair"); + assert_eq!( + committed.status, + AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED + ); + assert_eq!(committed.accumulated_text, response); + assert!(read_game_creator_agent_runtime_finalization_journal( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read cleaned finalization journal after repair") + .is_none()); + } + + #[test] + fn response_stream_committed_checkpoint_recovers_by_idempotent_cleanup() { + let (project, state, response_revision, snapshot) = + response_stream_fixture("response-stream-committed-checkpoint-run"); + let root = project.path(); + let response = "committed 后只剩幂等清理。"; + write_game_creator_agent_runtime_response_stream_ready_at( + root, + &snapshot, + response_revision, + response, + Some("stop"), + ) + .expect("write ready stream before committed checkpoint"); + let outcome = finish_game_creator_agent_background_runtime_turn_with_checkpoint_at( + root, + state.clone(), + response, + response_revision, + &[], + |checkpoint| { + if checkpoint == AgentRuntimeFinalizationCheckpoint::ResponseStreamCommitted { + Err("injected-after-response-stream-committed".to_string()) + } else { + Ok(()) + } + }, + ) + .expect("inject committed checkpoint interruption"); + assert!(matches!( + outcome, + AgentBackgroundFinalizationOutcome::Pending(_) + )); + let committed_before = read_game_creator_agent_runtime_response_stream_at( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read stream at committed checkpoint") + .expect("committed checkpoint stream exists"); + assert_eq!( + committed_before.status, + AGENT_RUNTIME_RESPONSE_STREAM_STATUS_COMMITTED + ); + assert!(read_game_creator_agent_runtime_finalization_journal( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read retained journal at committed checkpoint") + .is_some()); + + resume_response_stream_finalization(root, &state.agent_id); + let committed_after = read_game_creator_agent_runtime_response_stream_at( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read stream after committed checkpoint recovery") + .expect("stream exists after committed checkpoint recovery"); + assert_eq!(committed_after, committed_before); + assert!(read_game_creator_agent_runtime_finalization_journal( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read cleaned journal after committed checkpoint recovery") + .is_none()); + } + + #[tokio::test] + async fn provider_handoff_identity_drift_closes_lifecycle_without_leaking_response() { + let (project, state, _response_revision, snapshot) = + response_stream_fixture("provider-handoff-identity-drift-run"); + let root = project.path(); + let request = LlmRunRequest::new(vec![LlmMessage::user("验证 handoff 身份漂移")]); + let old_llm = GameCreatorLlmConfig { + api_key: "old-provider-key".to_string(), + base_url: "http://127.0.0.1:1/v1".to_string(), + model: "old-provider-model".to_string(), + api_kind: "openai_responses".to_string(), + reasoning_effort: "medium".to_string(), + stream: false, + web_search_enabled: false, + context_window_tokens: 128_000, + auto_compact_token_limit: 96_000, + tool_output_token_limit: 8_000, + request_timeout_ms: 1_000, + max_retries: 0, + retry_backoff_ms: 1, + }; + let mut new_llm = old_llm.clone(); + new_llm.model = "new-provider-model".to_string(); + let old_identity = + game_creator_agent_runtime_provider_retry_identity(&snapshot, &old_llm, &request) + .expect("build old handoff identity"); + let provider_request_id = game_creator_agent_runtime_provider_request_id(&snapshot); + append_game_creator_agent_runtime_provider_request_lifecycle( + root, + &snapshot, + &provider_request_id, + "started", + ) + .expect("append old Provider started lifecycle"); + let private_response = "OLD_PROVIDER_RESPONSE_MUST_NOT_LEAK"; + provider_handoff::write_at( + root, + &old_identity, + &snapshot.request_slot, + 0, + &provider_request_id, + &platform_llm::LlmRunResponse { + provider: platform_llm::LlmProvider::OpenAiCompatible, + model: old_llm.model.clone(), + text: private_response.to_string(), + finish_reason: Some("stop".to_string()), + response_id: None, + usage: None, + tool_calls: Vec::new(), + }, + ) + .expect("write old Provider handoff"); + + let outcome = request_game_creator_agent_runtime_llm_with_persisted_transient_retry_using( + root, + &snapshot, + &new_llm, + "agentLlm.design-director", + "测试 Provider handoff 身份漂移", + &request, + |_client, _request| async { + Err(platform_llm::LlmError::Transport( + "identity drift must not call Provider".to_string(), + )) + }, + |response| response.clone(), + ) + .await + .expect("supersede drifted Provider handoff"); + assert!(matches!( + outcome, + AgentRuntimePersistedProviderRequestOutcome::Superseded + )); + assert!( + provider_handoff::read_for_run_at(root, &state.agent_id, &state.run_id) + .expect("read removed drifted handoff") + .is_none() + ); + assert_eq!( + read_agent_db_lifecycle_transitions_at( + root, + AGENT_RUNTIME_PROVIDER_REQUEST_LIFECYCLE_RECORD_TYPE, + "requestId", + &provider_request_id, + ) + .expect("read closed drifted Provider lifecycle"), + ["started", "completed"] + ); + let audit = std::fs::read_to_string(root.join(".agent/agent.db")) + .expect("read handoff superseded audit"); + assert!(audit.contains("agent.runtime.provider_request.handoff_superseded")); + assert!(!audit.contains(private_response)); + } + + #[tokio::test] + async fn provider_handoff_retry_conflict_preserves_both_sidecars_for_reconciliation() { + let (project, state, _response_revision, snapshot) = + response_stream_fixture("provider-handoff-retry-conflict-run"); + let root = project.path(); + let request = LlmRunRequest::new(vec![LlmMessage::user("验证 handoff/retry 冲突")]); + let llm = GameCreatorLlmConfig { + api_key: "provider-key".to_string(), + base_url: "http://127.0.0.1:1/v1".to_string(), + model: "provider-model".to_string(), + api_kind: "openai_responses".to_string(), + reasoning_effort: "medium".to_string(), + stream: false, + web_search_enabled: false, + context_window_tokens: 128_000, + auto_compact_token_limit: 96_000, + tool_output_token_limit: 8_000, + request_timeout_ms: 1_000, + max_retries: 1, + retry_backoff_ms: 1, + }; + let identity = + game_creator_agent_runtime_provider_retry_identity(&snapshot, &llm, &request) + .expect("build Provider recovery identity"); + let provider_request_id = game_creator_agent_runtime_provider_request_id(&snapshot); + append_game_creator_agent_runtime_provider_request_lifecycle( + root, + &snapshot, + &provider_request_id, + "started", + ) + .expect("append Provider started lifecycle"); + provider_handoff::write_at( + root, + &identity, + &snapshot.request_slot, + 0, + &provider_request_id, + &platform_llm::LlmRunResponse { + provider: platform_llm::LlmProvider::OpenAiCompatible, + model: llm.model.clone(), + text: "已成功但尚未消费的回复".to_string(), + finish_reason: Some("stop".to_string()), + response_id: None, + usage: None, + tool_calls: Vec::new(), + }, + ) + .expect("write Provider handoff"); + provider_retry::write_next_at( + root, + &identity, + &format!("{}-transient-1", snapshot.request_slot), + 1, + 1, + 1_000, + "transport", + &"a".repeat(64), + ) + .expect("write conflicting Provider retry"); + + let result = request_game_creator_agent_runtime_llm_with_persisted_transient_retry_using( + root, + &snapshot, + &llm, + "agentLlm.design-director", + "测试 Provider recovery 冲突", + &request, + |_client, _request| async { + Err(platform_llm::LlmError::Transport( + "reconciliation must not call Provider".to_string(), + )) + }, + |response| response.clone(), + ) + .await; + let error = match result { + Err(error) => error, + Ok(_) => panic!("conflicting handoff/retry must require reconciliation"), + }; + assert!(error.starts_with(AGENT_RUNTIME_PROVIDER_REQUEST_RECONCILIATION_PREFIX)); + assert!( + provider_handoff::read_for_run_at(root, &state.agent_id, &state.run_id) + .expect("read preserved Provider handoff") + .is_some() + ); + assert!( + provider_retry::read_for_run_at(root, &state.agent_id, &state.run_id) + .expect("read preserved Provider retry") + .is_some() + ); + assert_eq!( + read_agent_db_lifecycle_transitions_at( + root, + AGENT_RUNTIME_PROVIDER_REQUEST_LIFECYCLE_RECORD_TYPE, + "requestId", + &provider_request_id, + ) + .expect("read unclosed conflicting Provider lifecycle"), + ["started"] + ); + let runtime = read_game_creator_agent_runtime_at(root, &state.agent_id) + .expect("read reconciliation Runtime") + .state; + assert_eq!(runtime.phase, "needs-reconciliation"); + } + + #[test] + fn finalization_v3_without_response_request_slot_remains_readable() { + let (project, state, response_revision, _snapshot) = + response_stream_fixture("finalization-v3-compatibility-run"); + let root = project.path(); + let journal = build_game_creator_agent_runtime_finalization_journal( + root, + &state, + "兼容读取 v3 finalization。", + response_revision, + ) + .expect("build current finalization journal"); + let mut legacy = serde_json::to_value(&journal).expect("serialize current finalization"); + legacy["schemaVersion"] = serde_json::Value::String( + AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION.to_string(), + ); + legacy["finalizationId"] = + serde_json::Value::String(game_creator_agent_runtime_finalization_id( + &journal.project_id, + &journal.agent_id, + &journal.session_id, + &journal.run_id, + &journal.response_fingerprint, + journal.response_revision, + "", + journal.response_steer_cursor, + &journal.plan_snapshot_fingerprint, + &journal.goal_snapshot_fingerprint, + )); + legacy + .as_object_mut() + .expect("finalization object") + .remove("responseRequestSlot"); + let path = + game_creator_agent_runtime_finalization_path(root, &state.agent_id, &state.run_id); + std::fs::create_dir_all(path.parent().expect("finalization parent")) + .expect("create finalization parent"); + std::fs::write( + &path, + serde_json::to_vec_pretty(&legacy).expect("serialize v3 finalization"), + ) + .expect("write v3 finalization without responseRequestSlot"); + + let recovered = read_game_creator_agent_runtime_finalization_journal( + root, + &state.agent_id, + &state.run_id, + ) + .expect("read v3 finalization") + .expect("v3 finalization exists"); + assert_eq!( + recovered.schema_version, + AGENT_RUNTIME_FINALIZATION_PREVIOUS_SCHEMA_VERSION + ); + assert!(recovered.response_request_slot.is_empty()); + } + + #[test] + fn finalization_v4_binds_response_request_slot_into_identity() { + let (project, state, response_revision, _snapshot) = + response_stream_fixture("finalization-v4-slot-binding-run"); + let root = project.path(); + let mut journal = build_game_creator_agent_runtime_finalization_journal( + root, + &state, + "v4 finalization 固定回复流身份。", + response_revision, + ) + .expect("build v4 finalization journal"); + journal.response_request_slot = "final-reply-loop-999-revision-999".to_string(); + let error = write_game_creator_agent_runtime_finalization_journal(root, &journal) + .expect_err("tampered v4 responseRequestSlot must break finalization identity"); + assert!(error.contains("幂等身份不匹配")); + } } async fn request_game_creator_agent_background_final_reply_at( @@ -21196,6 +22306,9 @@ async fn request_game_creator_agent_background_final_reply_at( AgentRuntimeContextCompactionOutcome::Waiting(record) => { return Ok(RequestedAgentRuntimeFinalReplyOutcome::Waiting(record)); } + AgentRuntimeContextCompactionOutcome::HandoffPrepared => { + return Ok(RequestedAgentRuntimeFinalReplyOutcome::HandoffPrepared); + } AgentRuntimeContextCompactionOutcome::Superseded => { return Ok(RequestedAgentRuntimeFinalReplyOutcome::Superseded); } @@ -21289,12 +22402,7 @@ async fn request_game_creator_agent_background_final_reply_at( .await { Ok(response) => { - let reply = - redact_agent_runtime_private_process_output_from_response( - &strip_llm_thinking_blocks(response.text.as_str()), - observations, - ); - publisher.ready(&reply, response.finish_reason.as_deref()); + publisher.handoff(); Ok(response) } Err(error) => { @@ -21303,27 +22411,18 @@ async fn request_game_creator_agent_background_final_reply_at( } } } else { - match client.run(attempt_request).await { - Ok(response) => { - let reply = - redact_agent_runtime_private_process_output_from_response( - &strip_llm_thinking_blocks(response.text.as_str()), - observations, - ); - let _ = write_game_creator_agent_runtime_response_stream_ready_at( - root, - &stream_snapshot, - response_revision, - &reply, - response.finish_reason.as_deref(), - ); - Ok(response) - } - Err(error) => Err(error), - } + client.run(attempt_request).await } } }, + |response| { + let mut response = response.clone(); + response.text = redact_agent_runtime_private_process_output_from_response( + &strip_llm_thinking_blocks(&response.text), + observations, + ); + response + }, ) .await; if response_result.is_err() { @@ -21345,6 +22444,9 @@ async fn request_game_creator_agent_background_final_reply_at( AgentRuntimePersistedProviderRequestOutcome::Waiting(record) => { return Ok(RequestedAgentRuntimeFinalReplyOutcome::Waiting(record)); } + AgentRuntimePersistedProviderRequestOutcome::HandoffPrepared => { + return Ok(RequestedAgentRuntimeFinalReplyOutcome::HandoffPrepared); + } AgentRuntimePersistedProviderRequestOutcome::Superseded => { return Ok(RequestedAgentRuntimeFinalReplyOutcome::Superseded); } @@ -21365,6 +22467,13 @@ async fn request_game_creator_agent_background_final_reply_at( } return Err(format!("{config_path} 后台 Agent 最终回复为空")); } + write_game_creator_agent_runtime_response_stream_ready_at( + root, + &stream_snapshot, + response_revision, + &reply, + response.finish_reason.as_deref(), + )?; Ok(RequestedAgentRuntimeFinalReplyOutcome::Ready(Some( RequestedAgentRuntimeFinalReply { reply, @@ -33577,7 +34686,7 @@ pub(crate) fn finish_game_creator_agent_runtime_turn_at( remove_game_creator_agent_runtime_pending_tool_action(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_provider_action_batch(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_confirmations(root, &state.agent_id, &state.run_id)?; - provider_retry::remove_at(root, &state.agent_id, &state.run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, &state.agent_id, &state.run_id)?; publish_game_creator_agent_delegate_result_for_state( root, &state, @@ -34043,7 +35152,9 @@ where "goal-completed", journal.runtime_completed_at.unwrap_or(journal.updated_at), )?; - remove_game_creator_agent_runtime_finalization_journal( + mark_game_creator_agent_runtime_response_stream_committed_at(root, &completed, journal)?; + checkpoint(AgentRuntimeFinalizationCheckpoint::ResponseStreamCommitted)?; + remove_game_creator_agent_runtime_finalization_recovery_sidecars( root, &journal.agent_id, &journal.run_id, @@ -34317,7 +35428,7 @@ pub(crate) fn fail_game_creator_agent_runtime_turn_at( remove_game_creator_agent_runtime_pending_tool_action(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_provider_action_batch(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_confirmations(root, &state.agent_id, &state.run_id)?; - provider_retry::remove_at(root, &state.agent_id, &state.run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, &state.agent_id, &state.run_id)?; publish_game_creator_agent_delegate_result_for_state(root, &state, state.error.as_deref()); Ok(state) } @@ -34368,7 +35479,7 @@ pub(crate) fn fail_game_creator_agent_runtime_budget_at( remove_game_creator_agent_runtime_pending_tool_action(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_provider_action_batch(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_confirmations(root, &state.agent_id, &state.run_id)?; - provider_retry::remove_at(root, &state.agent_id, &state.run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, &state.agent_id, &state.run_id)?; publish_game_creator_agent_delegate_result_for_state(root, &state, state.error.as_deref()); Ok(state) } @@ -35728,7 +36839,7 @@ fn mark_game_creator_agent_runtime_cancelled_at_locked( remove_game_creator_agent_runtime_pending_tool_action(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_provider_action_batch(root, &state.agent_id, &state.run_id)?; remove_game_creator_agent_runtime_confirmations(root, &state.agent_id, &state.run_id)?; - provider_retry::remove_at(root, &state.agent_id, &state.run_id)?; + remove_game_creator_agent_runtime_provider_recovery_at(root, &state.agent_id, &state.run_id)?; if let Some(goal) = mark_game_creator_agent_goal_cleared_for_runtime_at_locked(root, state)? { state.goal_status = Some(goal.status); state.updated_at = unix_timestamp(); diff --git a/apps/ai-game-creator-shell/src-tauri/src/main.rs b/apps/ai-game-creator-shell/src-tauri/src/main.rs index 6250d0070..631c91493 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/main.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/main.rs @@ -66,6 +66,7 @@ mod preview; mod process_session; mod process_session_bridge; mod project; +mod provider_handoff; mod provider_retry; mod repository_context; mod runner; diff --git a/apps/ai-game-creator-shell/src-tauri/src/project.rs b/apps/ai-game-creator-shell/src-tauri/src/project.rs index 901705746..23e09f2a6 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/project.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/project.rs @@ -91,6 +91,7 @@ const AGENT_DB_FINALIZATION_LIFECYCLE_SCHEMA_VERSION: &str = const AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V1: &str = "game-creator-runtime-finalization.v1"; const AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V2: &str = "game-creator-runtime-finalization.v2"; const AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V3: &str = "game-creator-runtime-finalization.v3"; +const AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V4: &str = "game-creator-runtime-finalization.v4"; const AGENT_DB_MAX_ACTION_RECEIPT_SCAN_BYTES: u64 = 256 * 1024 * 1024; const AGENT_DB_MAX_SCAN_RECORDS: usize = 1_000_000; const AGENT_DB_TERMINAL_RESERVE_RECORDS: u64 = 64; @@ -1439,6 +1440,7 @@ fn validate_agent_db_finalization_lifecycle_semantics( AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V1 | AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V2 | AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V3 + | AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V4 ) { return Err("Agent DB finalization lifecycle journal schema 无效".to_string()); } @@ -1506,7 +1508,10 @@ fn validate_agent_db_finalization_lifecycle_semantics( .ok_or_else(|| { "Agent DB finalization lifecycle goalSnapshotFingerprint 无效".to_string() })?; - if journal_schema != AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V3 { + if !matches!( + journal_schema, + AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V3 | AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V4 + ) { if !record.get("goalId").is_some_and(serde_json::Value::is_null) || goal_revision != 0 || !goal_fingerprint.is_empty() @@ -8547,7 +8552,7 @@ mod agent_db_security_tests { finalization_lifecycle_record_with_schema( finalization_id, stage, - AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V3, + AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V4, ) } @@ -9507,11 +9512,12 @@ mod agent_db_security_tests { } #[test] - fn finalization_lifecycle_accepts_supported_v1_through_v3_journals() { + fn finalization_lifecycle_accepts_supported_v1_through_v4_journals() { for (index, (schema, hex)) in [ (AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V1, '6'), (AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V2, '7'), (AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V3, '8'), + (AGENT_DB_FINALIZATION_JOURNAL_SCHEMA_V4, '9'), ] .into_iter() .enumerate() diff --git a/apps/ai-game-creator-shell/src-tauri/src/provider_handoff.rs b/apps/ai-game-creator-shell/src-tauri/src/provider_handoff.rs new file mode 100644 index 000000000..6dc87e65f --- /dev/null +++ b/apps/ai-game-creator-shell/src-tauri/src/provider_handoff.rs @@ -0,0 +1,548 @@ +use std::path::{Path, PathBuf}; + +use platform_llm::{LlmProvider, LlmRunResponse, LlmTokenUsage}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::agent::{ + agent_runtime_json_sidecar_backup_path, read_agent_runtime_json_sidecar_with_max_bytes, + redact_agent_runtime_project_paths, redact_secret_tokens, + remove_agent_runtime_json_sidecar_backup, strip_llm_thinking_blocks, + write_agent_runtime_json_sidecar_with_max_bytes, +}; +use crate::provider_retry::{self, validate_identity, AgentRuntimeProviderRetryIdentity}; +use crate::repository_context::redact_absolute_path_tokens; + +pub(crate) const PROVIDER_HANDOFF_SCHEMA_VERSION: &str = "game-creator-provider-handoff.v1"; + +const PROVIDER_HANDOFF_RELATIVE_DIRECTORY: &str = ".agent/runtime/provider-handoffs"; +const PROVIDER_HANDOFF_SIDECAR_MAX_BYTES: usize = 512 * 1024; +const PROVIDER_HANDOFF_RESPONSE_MAX_CHARS: usize = 256 * 1024; +const PROVIDER_HANDOFF_REQUEST_ID_MAX_CHARS: usize = 256; +const PROVIDER_HANDOFF_LABEL: &str = "Agent Runtime Provider 成功响应交接记录"; + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +struct AgentRuntimeProviderHandoffResponse { + provider: LlmProvider, + model: String, + text: String, + finish_reason: Option, + response_id: Option, + usage: Option, +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields, rename_all = "camelCase")] +pub(crate) struct AgentRuntimeProviderHandoffRecord { + pub(crate) schema_version: String, + pub(crate) identity: AgentRuntimeProviderRetryIdentity, + pub(crate) provider_request_id: String, + pub(crate) request_slot: String, + pub(crate) attempt: u32, + response: AgentRuntimeProviderHandoffResponse, + pub(crate) response_fingerprint: String, + pub(crate) created_at_ms: u64, +} + +impl AgentRuntimeProviderHandoffRecord { + pub(crate) fn to_llm_response(&self) -> LlmRunResponse { + LlmRunResponse { + provider: self.response.provider, + model: self.response.model.clone(), + text: self.response.text.clone(), + finish_reason: self.response.finish_reason.clone(), + response_id: self.response.response_id.clone(), + usage: self.response.usage.clone(), + tool_calls: Vec::new(), + } + } +} + +pub(crate) fn supports_request_kind(request_kind: &str) -> bool { + matches!( + request_kind, + "context-compaction" | "final-reply-context-compaction" | "final-reply" + ) +} + +pub(crate) fn read_for_run_at( + root: &Path, + agent_id: &str, + run_id: &str, +) -> Result, String> { + validate_path_identity(agent_id, run_id)?; + let relative_path = provider_handoff_relative_path(agent_id, run_id); + let Some(record) = read_agent_runtime_json_sidecar_with_max_bytes( + root, + &relative_path, + PROVIDER_HANDOFF_LABEL, + PROVIDER_HANDOFF_SIDECAR_MAX_BYTES, + )? + else { + return Ok(None); + }; + validate_record(&record)?; + if record.identity.agent_id != agent_id || record.identity.run_id != run_id { + return Err("Provider 成功响应交接记录与路径 Agent/run 身份冲突".to_string()); + } + Ok(Some(record)) +} + +pub(crate) fn write_at( + root: &Path, + identity: &AgentRuntimeProviderRetryIdentity, + request_slot: &str, + attempt: u32, + provider_request_id: &str, + response: &LlmRunResponse, +) -> Result { + validate_identity(identity)?; + validate_provider_request_id(provider_request_id)?; + if !supports_request_kind(&identity.request_kind) { + return Err("当前 Provider requestKind 不允许持久交接成功响应".to_string()); + } + if !response.tool_calls.is_empty() { + return Err("Provider 成功响应交接禁止保存 tool calls".to_string()); + } + let response_text = strip_llm_thinking_blocks(&response.text); + let response_text = redact_secret_tokens(&response_text); + let response_text = redact_agent_runtime_project_paths( + root, + &response_text, + PROVIDER_HANDOFF_RESPONSE_MAX_CHARS, + ); + let response = AgentRuntimeProviderHandoffResponse { + provider: response.provider, + model: response.model.clone(), + text: redact_absolute_path_tokens(&response_text), + finish_reason: response.finish_reason.clone(), + response_id: response.response_id.clone(), + usage: response.usage.clone(), + }; + let response_fingerprint = response_fingerprint(&response)?; + if let Some(existing) = read_for_run_at(root, &identity.agent_id, &identity.run_id)? { + if existing.identity == *identity + && existing.provider_request_id == provider_request_id + && existing.request_slot == request_slot + && existing.attempt == attempt + && existing.response == response + && existing.response_fingerprint == response_fingerprint + { + return Ok(existing); + } + return Err("Provider 成功响应交接记录内容冲突".to_string()); + } + let record = AgentRuntimeProviderHandoffRecord { + schema_version: PROVIDER_HANDOFF_SCHEMA_VERSION.to_string(), + identity: identity.clone(), + provider_request_id: provider_request_id.to_string(), + request_slot: request_slot.to_string(), + attempt, + response, + response_fingerprint, + created_at_ms: provider_retry::now_ms(), + }; + validate_record(&record)?; + let relative_path = provider_handoff_relative_path(&identity.agent_id, &identity.run_id); + write_agent_runtime_json_sidecar_with_max_bytes( + root, + &relative_path, + PROVIDER_HANDOFF_LABEL, + &record, + PROVIDER_HANDOFF_SIDECAR_MAX_BYTES, + )?; + let persisted = read_for_run_at(root, &identity.agent_id, &identity.run_id)? + .ok_or_else(|| "Provider 成功响应交接记录写入后不存在".to_string())?; + if persisted != record { + return Err("Provider 成功响应交接记录写入后内容冲突".to_string()); + } + Ok(persisted) +} + +pub(crate) fn remove_matching_at( + root: &Path, + identity: &AgentRuntimeProviderRetryIdentity, +) -> Result<(), String> { + validate_identity(identity)?; + let Some(record) = read_for_run_at(root, &identity.agent_id, &identity.run_id)? else { + return Ok(()); + }; + if record.identity != *identity { + return Err("Provider 成功响应交接记录身份冲突".to_string()); + } + remove_at(root, &identity.agent_id, &identity.run_id) +} + +pub(crate) fn remove_consumed_context_at( + root: &Path, + agent_id: &str, + run_id: &str, + request_kind: &str, + base_request_slot: &str, +) -> Result<(), String> { + let Some(record) = read_for_run_at(root, agent_id, run_id)? else { + return Ok(()); + }; + if record.identity.request_kind != request_kind + || record.identity.base_request_slot != base_request_slot + { + return Err("已消费的上下文压缩与 Provider 成功响应交接身份冲突".to_string()); + } + remove_at(root, agent_id, run_id) +} + +pub(crate) fn remove_at(root: &Path, agent_id: &str, run_id: &str) -> Result<(), String> { + validate_path_identity(agent_id, run_id)?; + let path = provider_handoff_path(root, agent_id, run_id); + let backup_path = agent_runtime_json_sidecar_backup_path(&path); + remove_agent_runtime_json_sidecar_backup(&backup_path, PROVIDER_HANDOFF_LABEL)?; + remove_agent_runtime_json_sidecar_backup(&path, PROVIDER_HANDOFF_LABEL) +} + +fn validate_record(record: &AgentRuntimeProviderHandoffRecord) -> Result<(), String> { + if record.schema_version != PROVIDER_HANDOFF_SCHEMA_VERSION { + return Err(format!( + "不支持的 Provider 成功响应交接版本：{}", + record.schema_version + )); + } + validate_identity(&record.identity)?; + if !supports_request_kind(&record.identity.request_kind) { + return Err("Provider 成功响应交接 requestKind 无效".to_string()); + } + validate_provider_request_id(&record.provider_request_id)?; + let expected_slot = request_slot_for_attempt(&record.identity, record.attempt); + if record.request_slot != expected_slot { + return Err("Provider 成功响应交接 requestSlot/attempt 无效".to_string()); + } + validate_short_text("model", &record.response.model, 256, false)?; + if record.response.text.chars().count() > PROVIDER_HANDOFF_RESPONSE_MAX_CHARS { + return Err(format!( + "Provider 成功响应交接正文超过 {PROVIDER_HANDOFF_RESPONSE_MAX_CHARS} 字符上限" + )); + } + if let Some(finish_reason) = record.response.finish_reason.as_deref() { + validate_short_text("finishReason", finish_reason, 80, true)?; + } + if let Some(response_id) = record.response.response_id.as_deref() { + validate_short_text("responseId", response_id, 256, true)?; + } + if record.response_fingerprint != response_fingerprint(&record.response)? { + return Err("Provider 成功响应交接 responseFingerprint 不匹配".to_string()); + } + if record.created_at_ms == 0 { + return Err("Provider 成功响应交接 createdAtMs 无效".to_string()); + } + Ok(()) +} + +fn request_slot_for_attempt(identity: &AgentRuntimeProviderRetryIdentity, attempt: u32) -> String { + if attempt == 0 { + identity.base_request_slot.clone() + } else { + format!("{}-transient-{attempt}", identity.base_request_slot) + } +} + +fn validate_short_text( + label: &str, + value: &str, + max_chars: usize, + allow_empty: bool, +) -> Result<(), String> { + if (!allow_empty && value.trim().is_empty()) + || value.chars().count() > max_chars + || value.chars().any(char::is_control) + { + return Err(format!("Provider 成功响应交接 {label} 无效")); + } + Ok(()) +} + +fn validate_provider_request_id(value: &str) -> Result<(), String> { + validate_short_text( + "Provider lifecycle requestId", + value, + PROVIDER_HANDOFF_REQUEST_ID_MAX_CHARS, + false, + )?; + let fingerprint = value + .strip_prefix("provider-request-") + .ok_or_else(|| "Provider 成功响应交接 Provider lifecycle requestId 无效".to_string())?; + if fingerprint.len() != 64 || !fingerprint.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err("Provider 成功响应交接 Provider lifecycle requestId 无效".to_string()); + } + Ok(()) +} + +fn response_fingerprint(response: &AgentRuntimeProviderHandoffResponse) -> Result { + let bytes = serde_json::to_vec(response) + .map_err(|error| format!("序列化 Provider 成功响应指纹失败：{error}"))?; + Ok(format!("{:x}", Sha256::digest(bytes))) +} + +fn validate_path_identity(agent_id: &str, run_id: &str) -> Result<(), String> { + if agent_id.trim().is_empty() || run_id.trim().is_empty() { + return Err("Provider 成功响应交接路径的 Agent/run 身份不能为空".to_string()); + } + Ok(()) +} + +fn provider_handoff_relative_path(agent_id: &str, run_id: &str) -> String { + format!( + "{PROVIDER_HANDOFF_RELATIVE_DIRECTORY}/{}/{}.json", + path_key(agent_id), + path_key(run_id) + ) +} + +fn provider_handoff_path(root: &Path, agent_id: &str, run_id: &str) -> PathBuf { + root.join(provider_handoff_relative_path(agent_id, run_id)) +} + +fn path_key(value: &str) -> String { + format!("{:x}", Sha256::digest(value.as_bytes())) +} + +#[cfg(test)] +mod tests { + use std::fs; + + use tempfile::tempdir; + + use super::*; + + fn identity(request_kind: &str) -> AgentRuntimeProviderRetryIdentity { + AgentRuntimeProviderRetryIdentity { + project_id: "project-provider-handoff".to_string(), + agent_id: "design-director".to_string(), + task_id: "design-director".to_string(), + session_id: "agent-session-design-director".to_string(), + run_id: "provider-handoff-run".to_string(), + source: "agent-background-task".to_string(), + goal_id: None, + goal_revision: 0, + goal_snapshot_fingerprint: String::new(), + applied_steer_cursor: 0, + request_kind: request_kind.to_string(), + base_request_slot: "final-reply-loop-1-revision-0".to_string(), + request_fingerprint: "1".repeat(64), + provider_config_fingerprint: "2".repeat(64), + web_search_enabled: false, + allow_idle_context_compaction: false, + } + } + + fn response(text: &str) -> LlmRunResponse { + LlmRunResponse { + provider: LlmProvider::OpenAiCompatible, + model: "handoff-model".to_string(), + text: text.to_string(), + finish_reason: Some("stop".to_string()), + response_id: Some("response-handoff".to_string()), + usage: Some(LlmTokenUsage { + prompt_tokens: 11, + completion_tokens: 7, + total_tokens: 18, + }), + tool_calls: Vec::new(), + } + } + + fn provider_request_id(marker: char) -> String { + format!("provider-request-{}", marker.to_string().repeat(64)) + } + + #[test] + fn provider_handoff_round_trips_idempotently_and_removes_both_copies() { + let project = tempdir().expect("provider handoff project"); + let identity = identity("final-reply"); + let actual_request_id = provider_request_id('a'); + let expected = response("最终回复已持久交接。"); + let first = write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &actual_request_id, + &expected, + ) + .expect("write provider handoff"); + let second = write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &actual_request_id, + &expected, + ) + .expect("rewrite same provider handoff"); + assert_eq!(first, second); + assert_eq!(second.provider_request_id, actual_request_id); + assert_eq!(second.request_slot, identity.base_request_slot); + assert_eq!(second.to_llm_response(), expected); + assert_eq!( + read_for_run_at(project.path(), &identity.agent_id, &identity.run_id) + .expect("read provider handoff"), + Some(second) + ); + + remove_matching_at(project.path(), &identity).expect("remove provider handoff"); + assert!( + read_for_run_at(project.path(), &identity.agent_id, &identity.run_id) + .expect("read removed provider handoff") + .is_none() + ); + } + + #[test] + fn provider_handoff_rejects_tool_calls_and_content_conflicts() { + let project = tempdir().expect("provider handoff project"); + let handoff_identity = identity("final-reply"); + let actual_request_id = provider_request_id('a'); + let expected = response("原始回复"); + write_at( + project.path(), + &handoff_identity, + &handoff_identity.base_request_slot, + 0, + &actual_request_id, + &expected, + ) + .expect("write original provider handoff"); + let error = write_at( + project.path(), + &handoff_identity, + &handoff_identity.base_request_slot, + 0, + &actual_request_id, + &response("冲突回复"), + ) + .expect_err("conflicting provider handoff must fail"); + assert!(error.contains("内容冲突")); + + let mut with_tool_call = response("工具回复"); + with_tool_call.tool_calls.push(platform_llm::LlmToolCall { + id: "call-1".to_string(), + name: "unsafe".to_string(), + arguments: "{}".to_string(), + }); + let other = identity("final-reply-context-compaction"); + let error = write_at( + project.path(), + &other, + &other.base_request_slot, + 0, + &provider_request_id('b'), + &with_tool_call, + ) + .expect_err("tool calls must not enter provider handoff"); + assert!(error.contains("tool calls")); + } + + #[test] + fn provider_handoff_rejects_request_id_conflicts() { + let project = tempdir().expect("provider handoff project"); + let identity = identity("final-reply"); + let expected = response("同一回复"); + write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &provider_request_id('a'), + &expected, + ) + .expect("write original provider handoff"); + + let error = write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &provider_request_id('b'), + &expected, + ) + .expect_err("conflicting requestId must fail"); + assert!(error.contains("内容冲突")); + } + + #[test] + fn provider_handoff_rejects_invalid_request_ids() { + let project = tempdir().expect("provider handoff project"); + let identity = identity("final-reply"); + let invalid_request_ids = [ + String::new(), + "provider-request-invalid\nidentity".to_string(), + "provider-request-not-a-sha256".to_string(), + "x".repeat(PROVIDER_HANDOFF_REQUEST_ID_MAX_CHARS + 1), + ]; + + for request_id in invalid_request_ids { + let error = write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &request_id, + &response("不会落盘"), + ) + .expect_err("invalid requestId must fail"); + assert!(error.contains("requestId")); + } + assert!( + read_for_run_at(project.path(), &identity.agent_id, &identity.run_id) + .expect("read absent provider handoff") + .is_none() + ); + } + + #[test] + fn provider_handoff_strips_thinking_before_persisting_response() { + let project = tempdir().expect("provider handoff project"); + let identity = identity("final-reply"); + let private_thinking = "不得进入 handoff 的内部推理"; + let public_response = "这是公开回复。"; + let expected = response(&format!( + "{private_thinking}\n{public_response}" + )); + + let record = write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &provider_request_id('c'), + &expected, + ) + .expect("write provider handoff without thinking"); + assert_eq!(record.to_llm_response().text, public_response); + + let persisted = fs::read_to_string(provider_handoff_path( + project.path(), + &identity.agent_id, + &identity.run_id, + )) + .expect("read persisted provider handoff"); + assert!(!persisted.contains(private_thinking)); + assert!(!persisted.to_ascii_lowercase().contains("")); + assert!(persisted.contains(public_response)); + } + + #[test] + fn provider_handoff_rejects_unknown_request_kind() { + let project = tempdir().expect("provider handoff project"); + let identity = identity("tool-plan"); + let error = write_at( + project.path(), + &identity, + &identity.base_request_slot, + 0, + &provider_request_id('d'), + &response("tool plan"), + ) + .expect_err("tool plan handoff must fail"); + assert!(error.contains("不允许")); + } +} diff --git a/apps/ai-game-creator-shell/src-tauri/src/provider_retry.rs b/apps/ai-game-creator-shell/src-tauri/src/provider_retry.rs index ae7b5ee4b..132f9e1a5 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/provider_retry.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/provider_retry.rs @@ -352,7 +352,9 @@ fn validate_path_identity(agent_id: &str, run_id: &str) -> Result<(), String> { Ok(()) } -fn validate_identity(identity: &AgentRuntimeProviderRetryIdentity) -> Result<(), String> { +pub(crate) fn validate_identity( + identity: &AgentRuntimeProviderRetryIdentity, +) -> Result<(), String> { validate_path_identity(&identity.agent_id, &identity.run_id)?; for (label, value) in [ ("projectId", identity.project_id.as_str()), diff --git a/apps/ai-game-creator-shell/src-tauri/src/runner.rs b/apps/ai-game-creator-shell/src-tauri/src/runner.rs index 76cc0bd9a..32ed1751c 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/runner.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/runner.rs @@ -2745,6 +2745,7 @@ fn external_agent_runner_root_is_idle(root: &Path) -> Result { for durable_dir in [ root.join(".agent/runtime/pending-actions"), root.join(".agent/runtime/finalizations"), + root.join(".agent/runtime/provider-handoffs"), root.join(".agent/runtime/provider-retries"), ] { if external_agent_runner_directory_has_durable_files(&durable_dir)? { @@ -4570,6 +4571,125 @@ mod tests { assert!(state.draining.load(Ordering::Acquire)); } + #[test] + fn durable_provider_handoff_prevents_shutdown_even_when_corrupt() { + let directory = unique_test_directory(); + let root = directory.0.join("project"); + let identity = crate::provider_retry::AgentRuntimeProviderRetryIdentity { + project_id: "project-provider-handoff-idle".to_string(), + agent_id: "code-prototype".to_string(), + task_id: "provider-handoff-task-idle".to_string(), + session_id: "session-provider-handoff-idle".to_string(), + run_id: "run-provider-handoff-idle".to_string(), + source: "agent-chat".to_string(), + goal_id: None, + goal_revision: 0, + goal_snapshot_fingerprint: String::new(), + applied_steer_cursor: 0, + request_kind: "final-reply".to_string(), + base_request_slot: "final-reply-loop-1-revision-0".to_string(), + request_fingerprint: "d".repeat(64), + provider_config_fingerprint: "e".repeat(64), + web_search_enabled: false, + allow_idle_context_compaction: false, + }; + let response = platform_llm::LlmRunResponse { + provider: platform_llm::LlmProvider::OpenAiCompatible, + model: "provider-handoff-runner-test".to_string(), + text: "durable final reply".to_string(), + finish_reason: Some("stop".to_string()), + response_id: Some("provider-handoff-response".to_string()), + usage: None, + tool_calls: Vec::new(), + }; + let provider_request_id = format!("provider-request-{}", "f".repeat(64)); + crate::provider_handoff::write_at( + &root, + &identity, + &identity.base_request_slot, + 0, + &provider_request_id, + &response, + ) + .expect("write durable Provider handoff"); + + assert!(!external_agent_runner_root_is_idle(&root).expect("scan primary handoff")); + let agent_key = format!("{:x}", Sha256::digest(identity.agent_id.as_bytes())); + let run_key = format!("{:x}", Sha256::digest(identity.run_id.as_bytes())); + let handoff_path = root + .join(".agent/runtime/provider-handoffs") + .join(agent_key) + .join(format!("{run_key}.json")); + let previous_path = crate::agent::agent_runtime_json_sidecar_backup_path(&handoff_path); + fs::rename(&handoff_path, &previous_path).expect("move Provider handoff to previous"); + assert_eq!( + crate::provider_handoff::read_for_run_at(&root, &identity.agent_id, &identity.run_id,) + .expect("recover previous Provider handoff") + .map(|record| record.to_llm_response()), + Some(response) + ); + assert!(!external_agent_runner_root_is_idle(&root).expect("scan previous handoff")); + + fs::write(&previous_path, b"{").expect("corrupt Provider handoff"); + crate::provider_handoff::read_for_run_at(&root, &identity.agent_id, &identity.run_id) + .expect_err("corrupt Provider handoff must enter recovery error handling"); + assert!(!external_agent_runner_root_is_idle(&root).expect("scan corrupt handoff")); + + let token = "provider-handoff-shutdown-token-provider-handoff-shutdown-token"; + let state = ExternalAgentRunnerServerState::new( + directory.0.join(EXTERNAL_AGENT_RUNNER_ENDPOINT_FILE_NAME), + test_endpoint(token, "provider-handoff-shutdown-boot", 32325), + ); + state.remember_root(&root); + let busy_response = handle_external_agent_runner_request( + ExternalAgentRunnerRequest { + protocol_version: EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION, + request_id: "shutdown-provider-handoff-busy-1".to_string(), + token: token.to_string(), + method: "runner.shutdown_if_idle".to_string(), + params: ExternalAgentRunnerRequestParams::default(), + }, + &state, + ); + + assert!(busy_response.ok); + assert_eq!( + busy_response + .result + .as_ref() + .and_then(|value| value["idle"].as_bool()), + Some(false) + ); + assert!(!state.shutdown_requested.load(Ordering::Acquire)); + assert!(!state.draining.load(Ordering::Acquire)); + + crate::provider_handoff::remove_at(&root, &identity.agent_id, &identity.run_id) + .expect("remove corrupt Provider handoff"); + assert!(external_agent_runner_root_is_idle(&root).expect("scan idle root")); + + let idle_response = handle_external_agent_runner_request( + ExternalAgentRunnerRequest { + protocol_version: EXTERNAL_AGENT_RUNNER_PROTOCOL_VERSION, + request_id: "shutdown-provider-handoff-idle-1".to_string(), + token: token.to_string(), + method: "runner.shutdown_if_idle".to_string(), + params: ExternalAgentRunnerRequestParams::default(), + }, + &state, + ); + + assert!(idle_response.ok); + assert_eq!( + idle_response + .result + .as_ref() + .and_then(|value| value["idle"].as_bool()), + Some(true) + ); + assert!(state.shutdown_requested.load(Ordering::Acquire)); + assert!(state.draining.load(Ordering::Acquire)); + } + #[test] fn stale_protocol_endpoint_does_not_override_instance_lock_arbitration() { let directory = unique_test_directory(); diff --git a/apps/ai-game-creator-shell/src-tauri/src/tests.rs b/apps/ai-game-creator-shell/src-tauri/src/tests.rs index 5c330a96e..7994fc8c6 100644 --- a/apps/ai-game-creator-shell/src-tauri/src/tests.rs +++ b/apps/ai-game-creator-shell/src-tauri/src/tests.rs @@ -822,7 +822,7 @@ async fn agent_goal_paused_edit_replans_old_confirmation_in_same_run() { } #[test] -fn agent_goal_finalization_v3_treats_new_revision_as_stale_before_assistant_write() { +fn agent_goal_finalization_v4_treats_new_revision_as_stale_before_assistant_write() { let root = unique_project_path(); init_local_game_project_at( &root, @@ -863,7 +863,7 @@ fn agent_goal_finalization_v3_treats_new_revision_as_stale_before_assistant_writ } }, ) - .expect("prepare Goal finalization v3"); + .expect("prepare Goal finalization v4"); assert!(matches!( outcome, AgentBackgroundFinalizationOutcome::Pending(_) @@ -26023,7 +26023,7 @@ fn finalization_resume_recovers_real_assistant_append_checkpoint_once() { ); assert!(lifecycle.iter().enumerate().all(|(index, record)| { record["auditSchemaVersion"] == "game-creator-finalization-lifecycle.v1" - && record["journalSchemaVersion"] == "game-creator-runtime-finalization.v3" + && record["journalSchemaVersion"] == "game-creator-runtime-finalization.v4" && record["finalizationId"] == journal.finalization_id && record["messageId"] == journal.message_id && record["stageOrdinal"] == u64::try_from(index + 1).unwrap_or_default() @@ -34411,6 +34411,526 @@ async fn provider_retry_waiting_final_reply_compaction_resumes_without_new_tool_ fs::remove_dir_all(root).ok(); } +fn wait_for_provider_handoff_test_stop( + root: &Path, + agent_id: &str, + run_id: &str, +) -> crate::provider_handoff::AgentRuntimeProviderHandoffRecord { + for _ in 0..250 { + let handoff = crate::provider_handoff::read_for_run_at(root, agent_id, run_id) + .expect("read Provider success handoff after test stop"); + if let Some(handoff) = handoff { + if game_creator_agent_runtime_task_lock_is_available(root, agent_id) + .expect("probe Provider handoff test-stop lane") + { + return handoff; + } + } + std::thread::sleep(Duration::from_millis(20)); + } + panic!("Provider success handoff was not committed before the test-stop lane released"); +} + +fn wait_for_provider_handoff_terminal_cleanup(root: &Path, agent_id: &str, run_id: &str) { + for _ in 0..250 { + let handoff = crate::provider_handoff::read_for_run_at(root, agent_id, run_id) + .expect("read Provider handoff during terminal cleanup"); + let retry = crate::provider_retry::read_for_run_at(root, agent_id, run_id) + .expect("read Provider retry during terminal cleanup"); + let finalization = + read_game_creator_agent_runtime_finalization_journal(root, agent_id, run_id) + .expect("read finalization during Provider handoff terminal cleanup"); + let lane_available = game_creator_agent_runtime_task_lock_is_available(root, agent_id) + .expect("probe Provider handoff terminal lane"); + if handoff.is_none() && retry.is_none() && finalization.is_none() && lane_available { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + panic!("Provider handoff/retry/finalization did not reach a clean terminal state"); +} + +#[tokio::test] +async fn provider_handoff_final_reply_restart_replays_success_without_network_request() { + 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 final reply handoff 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 injection = root.join(".agent/runtime/test-stop-after-provider-handoff"); + fs::create_dir_all(injection.parent().expect("handoff injection parent")) + .expect("create handoff injection parent"); + fs::write(&injection, b"stop-after-final-reply-provider-handoff") + .expect("write final reply handoff injection"); + let agent_id = "design-director"; + let run_id = "design-provider-handoff-final-reply-run"; + let planning_fallback = "成功交接恢复不得提交 planning fallback。"; + let final_response = "最终回复从成功交接记录恢复并唯一提交。"; + let (request_notice_sender, request_notice_receiver) = mpsc::channel(); + let (request_capture_sender, request_capture_receiver) = mpsc::channel(); + let (base_url, server_handle) = spawn_mock_llm_tool_plan_then_transient_final_reply( + final_tool_plan_response(planning_fallback), + final_response.to_string(), + request_notice_sender, + request_capture_sender, + ); + replace_test_local_config( + &config_path, + format!( + r#"{{ + "agentLlm": {{ + "design-director": {{ + "apiKey": "handoff-final-reply-key", + "baseUrl": {base_url:?}, + "model": "handoff-final-reply-model", + "apiKind": "openai_responses", + "stream": true, + "maxRetries": 1, + "retryBackoffMs": 1000 + }} + }} +}}"# + ), + ); + let started = start_game_creator_agent_background_task_at( + &root, + agent_id, + "验证 final-reply 成功响应 handoff 后恢复不重发 Provider", + run_id, + ) + .expect("start final reply handoff task"); + + for _ in 0..3 { + request_notice_receiver + .recv_timeout(Duration::from_secs(5)) + .expect("expected Provider request before final reply handoff stop"); + } + let handoff = wait_for_provider_handoff_test_stop(&root, agent_id, run_id); + assert!(!injection.exists()); + assert_eq!(handoff.identity.request_kind, "final-reply"); + assert_eq!(handoff.attempt, 1); + assert!(handoff.request_slot.ends_with("-transient-1")); + assert_eq!( + handoff.to_llm_response().text, + final_response, + "handoff must contain the canonical final reply" + ); + let retry = crate::provider_retry::read_for_run_at(&root, agent_id, run_id) + .expect("read retry retained with final reply handoff") + .expect("retry remains until final reply handoff replay"); + assert_eq!(retry.identity, handoff.identity); + let before_resume = read_game_creator_agent_runtime_at(&root, agent_id) + .expect("read Runtime stopped after final reply handoff"); + assert_eq!(before_resume.state.run_id, run_id); + assert_ne!(before_resume.state.phase, "completed"); + assert!( + read_game_creator_agent_runtime_finalization_journal(&root, agent_id, run_id) + .expect("read absent finalization before handoff replay") + .is_none() + ); + let before_conversation = read_local_conversation_for_session_at( + &root, + Some(agent_id), + Some(&started.state.session_id), + ) + .expect("read conversation before final reply handoff replay"); + assert_eq!( + before_conversation + .messages + .iter() + .filter(|message| message.role == "assistant") + .count(), + 0 + ); + let before_records = read_agent_db_records_for_test(&root); + let before_handoff_lifecycle = before_records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.provider_request.lifecycle" + && record["runId"] == run_id + && record["requestId"].as_str() == Some(handoff.provider_request_id.as_str()) + }) + .collect::>(); + assert_eq!(before_handoff_lifecycle.len(), 1); + assert_eq!(before_handoff_lifecycle[0]["status"], "started"); + assert_eq!( + before_handoff_lifecycle[0]["requestSlot"], + handoff.request_slot.as_str() + ); + + server_handle.join().expect("join final reply handoff mock"); + let captured_requests = (0..3) + .map(|_| { + request_capture_receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured Provider request before handoff replay") + }) + .collect::>(); + assert_eq!(captured_requests.len(), 3); + assert!(request_capture_receiver.try_recv().is_err()); + + let resumed = resume_game_creator_agent_background_tasks_at(&root) + .expect("resume final reply Provider handoff"); + assert_eq!(resumed.len(), 1); + let completed = wait_for_agent_runtime_idle(&root, agent_id); + assert_eq!(completed.phase, "completed"); + assert_eq!(completed.run_id, run_id); + assert_eq!(completed.last_response.as_deref(), Some(final_response)); + assert!(request_capture_receiver.try_recv().is_err()); + wait_for_provider_handoff_terminal_cleanup(&root, agent_id, run_id); + assert!( + crate::provider_handoff::read_for_run_at(&root, agent_id, run_id) + .expect("read cleaned final reply handoff") + .is_none() + ); + assert!( + crate::provider_retry::read_for_run_at(&root, agent_id, run_id) + .expect("read cleaned final reply retry") + .is_none() + ); + assert!( + read_game_creator_agent_runtime_finalization_journal(&root, agent_id, run_id) + .expect("read cleaned finalization after handoff replay") + .is_none() + ); + let completed_runtime = read_game_creator_agent_runtime_at(&root, agent_id) + .expect("read completed final reply handoff Runtime"); + assert_eq!( + completed_runtime + .recent_tasks + .iter() + .filter(|task| task.run_id == run_id && task.status == "completed") + .count(), + 1 + ); + let conversation = read_local_conversation_for_session_at( + &root, + Some(agent_id), + Some(&started.state.session_id), + ) + .expect("read completed final reply handoff conversation"); + assert_eq!( + conversation + .messages + .iter() + .filter(|message| message.role == "assistant") + .map(|message| message.content.as_str()) + .collect::>(), + vec![final_response] + ); + let committed_stream = wait_for_response_stream_status(&root, agent_id, run_id, "committed", 1); + assert_eq!(committed_stream.accumulated_text, final_response); + + let records = read_agent_db_records_for_test(&root); + let lifecycle = records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.provider_request.lifecycle" + && record["runId"] == run_id + }) + .collect::>(); + let tool_plan = lifecycle + .iter() + .copied() + .filter(|record| record["requestKind"] == "tool-plan") + .collect::>(); + assert_eq!(tool_plan.len(), 2); + assert_eq!(tool_plan[0]["status"], "started"); + assert_eq!(tool_plan[1]["status"], "completed"); + assert_eq!(tool_plan[0]["requestId"], tool_plan[1]["requestId"]); + let final_reply = lifecycle + .iter() + .copied() + .filter(|record| record["requestKind"] == "final-reply") + .collect::>(); + assert_eq!(final_reply.len(), 4); + assert_eq!( + final_reply + .iter() + .filter_map(|record| record["requestId"].as_str()) + .collect::>() + .len(), + 2 + ); + let resumed_handoff_lifecycle = final_reply + .iter() + .copied() + .filter(|record| record["requestId"].as_str() == Some(handoff.provider_request_id.as_str())) + .collect::>(); + assert_eq!(resumed_handoff_lifecycle.len(), 2); + assert_eq!(resumed_handoff_lifecycle[0]["status"], "started"); + assert_eq!(resumed_handoff_lifecycle[1]["status"], "completed"); + assert_eq!( + resumed_handoff_lifecycle[0]["requestId"], + resumed_handoff_lifecycle[1]["requestId"] + ); + assert_eq!( + resumed_handoff_lifecycle[0]["requestSlot"], + handoff.request_slot.as_str() + ); + assert_eq!( + resumed_handoff_lifecycle[1]["requestSlot"], + handoff.request_slot.as_str() + ); + assert_response_stream_completion_event_details(&root, agent_id, run_id, final_response); + + fs::remove_dir_all(root).ok(); + drop(config_guard); + fs::remove_dir_all(config_dir).ok(); +} + +#[tokio::test] +async fn provider_handoff_final_reply_compaction_restart_only_requests_final_reply() { + 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 final compaction handoff 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 injection = root.join(".agent/runtime/test-stop-after-provider-handoff"); + fs::create_dir_all(injection.parent().expect("handoff injection parent")) + .expect("create handoff injection parent"); + fs::write( + &injection, + b"stop-after-final-reply-context-compaction-handoff", + ) + .expect("write final reply compaction handoff injection"); + let agent_id = "design-director"; + let session_id = "agent-session-design-director"; + let run_id = "design-provider-handoff-final-compaction-run"; + for index in 0..60 { + append_local_conversation_message_for_session_at( + &root, + Some(agent_id), + Some(session_id), + LocalConversationMessage { + role: if index % 2 == 0 { "user" } else { "assistant" }.to_string(), + content: format!("HANDOFF_FINAL_COMPACTION_{index} {}", "x".repeat(1_200)), + agent_id: (index % 2 == 1).then(|| agent_id.to_string()), + }, + ) + .expect("append final compaction handoff conversation"); + } + let planning_fallback = "压缩 handoff 恢复不得提交 planning fallback。"; + let compaction_summary = "HANDOFF_COMPACTION_PRIVATE_SUMMARY：保留任务与观察。"; + let final_response = "压缩成功交接恢复后仅请求必要最终回复。"; + let (request_notice_sender, request_notice_receiver) = mpsc::channel(); + let (request_capture_sender, request_capture_receiver) = mpsc::channel(); + let (base_url, server_handle) = spawn_mock_llm_tool_plan_then_transient_final_compaction( + final_tool_plan_response(planning_fallback), + compaction_summary.to_string(), + final_response.to_string(), + config_path.clone(), + request_notice_sender, + request_capture_sender, + ); + replace_test_local_config( + &config_path, + format!( + r#"{{ + "agentLlm": {{ + "design-director": {{ + "apiKey": "handoff-final-compaction-key", + "baseUrl": {base_url:?}, + "model": "handoff-final-compaction-model", + "apiKind": "openai_responses", + "stream": true, + "contextWindowTokens": 128000, + "autoCompactTokenLimit": 100000, + "toolOutputTokenLimit": 8000, + "maxRetries": 1, + "retryBackoffMs": 1000 + }} + }} +}}"# + ), + ); + start_game_creator_agent_background_task_for_session_at( + &root, + agent_id, + Some(session_id), + "验证 final-reply 前置压缩 handoff 恢复不重复 planning 或 compaction", + run_id, + ) + .expect("start final compaction handoff task"); + + for _ in 0..3 { + request_notice_receiver + .recv_timeout(Duration::from_secs(5)) + .expect("expected Provider request before compaction handoff stop"); + } + let handoff = wait_for_provider_handoff_test_stop(&root, agent_id, run_id); + assert!(!injection.exists()); + assert_eq!( + handoff.identity.request_kind, + "final-reply-context-compaction" + ); + assert_eq!(handoff.attempt, 1); + assert!(handoff.request_slot.ends_with("-transient-1")); + assert_eq!(handoff.to_llm_response().text, compaction_summary); + let retry = crate::provider_retry::read_for_run_at(&root, agent_id, run_id) + .expect("read retry retained with compaction handoff") + .expect("retry remains until compaction handoff replay"); + assert_eq!(retry.identity, handoff.identity); + assert!( + read_game_creator_agent_runtime_finalization_journal(&root, agent_id, run_id) + .expect("read absent finalization before compaction handoff replay") + .is_none() + ); + let before_records = read_agent_db_records_for_test(&root); + let before_handoff_lifecycle = before_records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.provider_request.lifecycle" + && record["runId"] == run_id + && record["requestId"].as_str() == Some(handoff.provider_request_id.as_str()) + }) + .collect::>(); + assert_eq!(before_handoff_lifecycle.len(), 1); + assert_eq!(before_handoff_lifecycle[0]["status"], "started"); + assert!(request_notice_receiver + .recv_timeout(Duration::from_millis(150)) + .is_err()); + + let resumed = resume_game_creator_agent_background_tasks_at(&root) + .expect("resume final compaction Provider handoff"); + assert_eq!(resumed.len(), 1); + request_notice_receiver + .recv_timeout(Duration::from_secs(5)) + .expect("only necessary final reply request after compaction handoff replay"); + server_handle + .join() + .expect("join final compaction handoff mock"); + let captured_requests = (0..4) + .map(|_| { + request_capture_receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured final compaction handoff Provider request") + }) + .collect::>(); + assert_eq!(captured_requests.len(), 4); + assert!(request_capture_receiver.try_recv().is_err()); + + let completed = wait_for_agent_runtime_idle(&root, agent_id); + assert_eq!(completed.phase, "completed"); + assert_eq!(completed.run_id, run_id); + assert_eq!(completed.last_response.as_deref(), Some(final_response)); + wait_for_provider_handoff_terminal_cleanup(&root, agent_id, run_id); + assert!( + crate::provider_handoff::read_for_run_at(&root, agent_id, run_id) + .expect("read cleaned compaction handoff") + .is_none() + ); + assert!( + crate::provider_retry::read_for_run_at(&root, agent_id, run_id) + .expect("read cleaned compaction retry") + .is_none() + ); + assert!( + read_game_creator_agent_runtime_finalization_journal(&root, agent_id, run_id) + .expect("read cleaned finalization after compaction handoff replay") + .is_none() + ); + let compaction = + read_game_creator_agent_runtime_context_compaction(&root, agent_id, session_id) + .expect("read recovered final reply compaction") + .expect("recovered final reply compaction exists"); + assert!(compaction + .summary + .contains("HANDOFF_COMPACTION_PRIVATE_SUMMARY")); + let completed_runtime = read_game_creator_agent_runtime_at(&root, agent_id) + .expect("read completed compaction handoff Runtime"); + assert_eq!( + completed_runtime + .recent_tasks + .iter() + .filter(|task| task.run_id == run_id && task.status == "completed") + .count(), + 1 + ); + let conversation = + read_local_conversation_for_session_at(&root, Some(agent_id), Some(session_id)) + .expect("read completed compaction handoff conversation"); + assert_eq!( + conversation + .messages + .iter() + .filter(|message| message.role == "assistant" && message.content == final_response) + .count(), + 1 + ); + assert_eq!( + conversation + .messages + .iter() + .filter(|message| message.role == "assistant" && message.content == planning_fallback) + .count(), + 0 + ); + let committed_stream = wait_for_response_stream_status(&root, agent_id, run_id, "committed", 1); + assert_eq!(committed_stream.accumulated_text, final_response); + + let records = read_agent_db_records_for_test(&root); + let lifecycle_for = |request_kind: &str| { + records + .iter() + .filter(|record| { + record["recordType"] == "agent.runtime.provider_request.lifecycle" + && record["runId"] == run_id + && record["requestKind"] == request_kind + }) + .collect::>() + }; + let tool_plan = lifecycle_for("tool-plan"); + assert_eq!(tool_plan.len(), 2); + assert_eq!(tool_plan[0]["status"], "started"); + assert_eq!(tool_plan[1]["status"], "completed"); + assert_eq!(tool_plan[0]["requestId"], tool_plan[1]["requestId"]); + let final_compaction = lifecycle_for("final-reply-context-compaction"); + assert_eq!(final_compaction.len(), 4); + assert_eq!( + final_compaction + .iter() + .filter_map(|record| record["requestId"].as_str()) + .collect::>() + .len(), + 2 + ); + let resumed_handoff_lifecycle = final_compaction + .iter() + .copied() + .filter(|record| record["requestId"].as_str() == Some(handoff.provider_request_id.as_str())) + .collect::>(); + assert_eq!(resumed_handoff_lifecycle.len(), 2); + assert_eq!(resumed_handoff_lifecycle[0]["status"], "started"); + assert_eq!(resumed_handoff_lifecycle[1]["status"], "completed"); + assert_eq!( + resumed_handoff_lifecycle[0]["requestId"], + resumed_handoff_lifecycle[1]["requestId"] + ); + assert_eq!( + resumed_handoff_lifecycle[0]["requestSlot"], + handoff.request_slot.as_str() + ); + assert_eq!( + resumed_handoff_lifecycle[1]["requestSlot"], + handoff.request_slot.as_str() + ); + let final_reply = lifecycle_for("final-reply"); + assert_eq!(final_reply.len(), 2); + assert_eq!(final_reply[0]["status"], "started"); + assert_eq!(final_reply[1]["status"], "completed"); + assert_eq!(final_reply[0]["requestId"], final_reply[1]["requestId"]); + assert_response_stream_completion_event_details(&root, agent_id, run_id, final_response); + + fs::remove_dir_all(root).ok(); + drop(config_guard); + fs::remove_dir_all(config_dir).ok(); +} + #[tokio::test] async fn provider_transient_retry_backoff_is_exponential_and_capped_at_thirty_seconds() { let root = unique_project_path(); diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 20fb9a5dc..67a08b007 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -4910,3 +4910,13 @@ - 恢复：进入 Provider 前先同步 response 阶段计划投影与 context bundle。Runner 重启后即使公共 state 先投影 planning，执行 pass 也先按当前 run sidecar 识别 `requestKind=final-reply / final-reply-context-compaction`，恢复原 `nextLoopIndex` 并跳过新的 tool-plan；后者先恢复同一压缩请求，再继续原 final-reply。重建请求或 Provider/Goal/steer/revision 身份漂移时删除旧 sidecar，只记录漂移字段名并在同一 run 重新规划，不提交旧回复。 - 流式与终态：失败 attempt 的半句只进入 failed/discarded response-stream，恢复 attempt 沿原基础 stream 身份推进；唯一 canonical assistant 仍只由 finalization journal 提交。确定性测试覆盖 sidecar-first 投影窗口、到期前零请求、失败/恢复 HTTP body 逐字节一致、无重复 tool-plan、唯一 assistant/completed/committed stream 和终局零 retry sidecar。 - 证据边界：当前 `provider_retry_` 21/21、`provider_transient_retry_` 5/5、`response_stream_` 16/16，Tauri/Rust 串行全量 `969 passed / 4 ignored`。尚未完成真实外部 Provider 的 final-reply 退避期 Runner 强杀，因此不能复用 V1.39 首次 tool-plan 的真实 PASS；手动压缩和 tool-plan `repair-N` 继续保持进程内重试。Provider 成功返回到压缩 sidecar 或 finalization journal `prepared` 之间仍有崩溃窗口，重启可能重新请求 Provider；finalization journal 清理到 stream committed 之间也不是可恢复事务。本切片只保证失败重试与最终 assistant 幂等，不能声称成功请求 exactly-once 或 stream 终态事务已经完成。 + +## 2026-07-20 AI 游戏创作 Agent Runtime V1.41 Provider 成功交接与回复流终态恢复 + +- 持久所有权：`game-creator-provider-handoff.v1` 是 Provider 成功返回与消费端 durable commit 之间的私有交接记录，每个 Agent/run 最多一条。只允许无 tool call 的 `context-compaction / final-reply-context-compaction / final-reply`，绑定完整 retry identity、真实 request slot/attempt、真实 Provider lifecycle requestId、规范化文本响应及其指纹；不保存 prompt、请求消息、API Key、Provider URL、tool call/arguments 或错误正文。 +- 提交顺序：物理 Provider 成功后先去 thinking、按既有规则脱敏并原子写入 handoff，再回读逐字段完全一致，之后才允许为 handoff 中保存的真实 requestId 写 `completed` lifecycle。handoff 成为 durable owner 后，恢复先补齐同一 requestId 的 `started -> completed`，再零网络回放响应；不得生成新 requestId、把真实成功记到 base requestId，或在 handoff 未回读成功时宣称 lifecycle completed。 +- 所有权转移：上下文压缩必须先把规范 compaction sidecar 原子写入并回读一致，才可删除匹配 handoff；final-reply 先把回复转入 finalization journal `prepared`，随后由 journal 持有 assistant、Runtime/Goal 终态和 response stream 提交责任。终局清理不得早于下一 durable owner 建立；取消、steer、Goal 作废、失败或其它终态也必须按完整身份清理所属 handoff。 +- 冲突与漂移：同一 run 的 handoff 与 retry identity/attempt/slot 完全一致时，以 handoff 为成功事实并清理 retry；任一字段冲突必须在网络调用前进入 `needs-reconciliation`，保留两份 sidecar 和真实 requestId 供核对。相同 durable run 的 Goal/steer/request/config 等身份漂移，先按 handoff 的旧真实 requestId 补齐 lifecycle，再只记录漂移字段名、删除旧 handoff/retry 并回到同 run planning；响应正文不得进入公共审计。跨 run 身份冲突直接失败关闭。 +- 回复流事务：finalization journal 升级为 v4，并固定保存 `responseRequestSlot / responseSteerCursor / responseRevision / response`。assistant 与 Runtime/Goal 已幂等完成后，journal 仍必须保留到匹配 response stream 写成 `committed` 且立即回读身份、状态和正文完全一致；之后才可删除 journal 和剩余 handoff。提交使用 journal 的固定 Agent/task/Session/run/request slot/steer cursor/revision，禁止调用面向 UI 的可见性过滤或用当前全局 project revision 静默跳过既定 run 的 stream。 +- 回复流恢复：stream 缺失或仍为 `streaming` 时，可用 journal 固定身份和正文重建 `ready` 后提交；已 `committed` 且正文一致时按幂等成功继续清理。既有 stream 身份冲突、ready/committed 正文冲突、写入失败或回读失败都必须保留 journal 并保持可恢复 finalization，不能删除证据、覆盖冲突正文或把 finalization 当成已经清理。 +- Runner 与边界：primary、`.previous` 或损坏的 handoff 都使所属 root 保持 busy，并阻止 `runner.shutdown_if_idle`。handoff 原子提交并回读前强杀 Runner，仍可能留下 Provider 已成功但本地只有未闭合 `started` 的未知窗口；Runtime 只能失败关闭，V1.41 不因此承诺端到端 exactly-once。`tool-plan` 和 function arguments 明确不在本协议内，真实外部 Provider 的 final-reply 退避期 Runner 强杀仍需独立 E2E 后才能记 PASS。 diff --git a/docs/project-memory/shared-memory/development-workflow.md b/docs/project-memory/shared-memory/development-workflow.md index fa21c7af8..5e4e2131f 100644 --- a/docs/project-memory/shared-memory/development-workflow.md +++ b/docs/project-memory/shared-memory/development-workflow.md @@ -115,6 +115,26 @@ cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml provi 第一条测试必须停在 final-reply retry sidecar 已提交而 task/state 尚未投影的窗口，恢复扫描补齐 waiting 后在到期前保持零请求；第二条必须让 final-reply 前置自动压缩先失败并进入 `requestKind=final-reply-context-compaction` 等待态，到期后恢复同一压缩请求，再继续原 final-reply。两条链路都不得重新调用 tool-plan；失败和恢复请求只比较 HTTP body 字节、SHA-256 和长度，测试失败不得打印正文。真实 Provider 验收必须另起独立 suite，在 Supervisor 已认领全部专业回执、repair 和宿主验证后注入 final-reply 故障并于退避期强杀 Runner；该 suite 尚未 PASS 前，不能复用 V1.39 首次 tool-plan 的真实证据。Provider 成功返回到 finalization journal `prepared` 之间的崩溃窗口仍须独立补齐和验收，当前门禁不能据此宣称 Provider 调用 exactly-once。 +### AI 游戏创作 Runtime V1.41 成功交接与回复流恢复复验 + +修改 Provider 成功响应、持久重试、finalization、response stream 或 Runner idle 判定后，至少运行： + +```bash +cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml provider_handoff_ -- --nocapture --test-threads=1 +cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml response_stream_ -- --nocapture --test-threads=1 +cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml durable_provider_handoff_prevents_shutdown_even_when_corrupt -- --nocapture --test-threads=1 +``` + +复验和排障按 durable ownership 的顺序取证： + +1. 先按 handoff 中保存的真实 `providerRequestId / requestSlot / attempt` 核对 Provider lifecycle。成功响应必须先原子写入 handoff 并回读完全一致，再为同一真实 requestId 补 `completed`；恢复不得生成替代 requestId。 +2. 在 handoff 已提交、lifecycle 仍只有 `started` 的 checkpoint 停止执行，关闭 mock Provider 后再恢复。final-reply 必须零网络回放并只产生唯一 assistant/completed/committed stream；前置压缩只允许在回放压缩结果后发出后续必要的 final-reply，不能重复 tool-plan 或 compaction。 +3. 同一 run 同时存在 handoff 与 retry 时，完全匹配才允许回放并清理 retry；identity、attempt 或 slot 冲突必须零网络进入 `needs-reconciliation`，保留两份 sidecar 和真实 requestId 证据，不要为了让 Runner 退出而手工择一删除。 +4. finalization 已到 `runtime-completed` 后，故意删除 stream、保留 `streaming` 半句、注入 committed 写失败、在 committed 后 journal 清理前停止，并在期间推进全局 project revision。恢复必须始终使用 journal 固定的 run/request slot/steer cursor/response revision 与正文重建或幂等提交，写入后回读成功才可删除 journal。 +5. 固定身份或正文冲突、stream 写入或回读失败时，断言 journal 保留且恢复扫描继续处理同一 finalization；Runner 对 primary、`.previous` 或损坏 handoff 都应保持 busy，`runner.shutdown_if_idle` 不得返回 idle。终局再核对 retry/handoff/finalization sidecar 全部为零，并扫描公共 task/event/Agent DB/CLI，确保没有响应正文、thinking、凭据、Provider URL 或绝对路径。 + +V1.41 只覆盖无 tool call 的 `context-compaction / final-reply-context-compaction / final-reply`。Runner 在 Provider 成功后、handoff 原子提交并回读前被硬杀时，仍只能把未闭合 `started` 视为结果未知并失败关闭；该窗口不是 exactly-once。`tool-plan` 及其 function arguments 不进入 handoff，真实外部 Provider 的 final-reply 退避期 Runner 强杀仍需独立 E2E，不能用上述确定性测试或 V1.39 PASS 代替。 + suite 只能读正式 AppData，在其同级目录写入 sentinel 管理的 `0600` 私有副本和 overlay；启动 CLI/Runner 时须把 loopback 合并进大小写两套 no-proxy 环境，防止系统 HTTP 代理绕过本地故障门禁；source-dir guard 必须证明本 suite 前缀未进入源目录，源配置和 endpoint 身份保持不变，报告不得保存 Provider URL、headers、正文、凭据或绝对配置路径。sidecar 先于 task/state 投影是合法提交窗口，验收器应等待完整等待态后再强杀；若后续协作或终局失败，partial report 仍应保留已取得的 retry checkpoint，但失败轮不得与后续成功轮拼接。 ### AI 游戏创作自主 Swarm 终端复验 diff --git a/docs/project-memory/shared-memory/pitfalls.md b/docs/project-memory/shared-memory/pitfalls.md index c57bd0c1b..6ad1e8bf0 100644 --- a/docs/project-memory/shared-memory/pitfalls.md +++ b/docs/project-memory/shared-memory/pitfalls.md @@ -3367,3 +3367,14 @@ - 真实验收陷阱：sidecar 按设计早于 task/state 等待投影落盘，验收器看到 sidecar 后必须继续等完整 `running / waiting-for-provider-retry`，不能把合法提交窗口误判为 torn projection。共享 Runner 强杀会同时中断其它 Agent 的 in-flight Provider 请求；要隔离验证单个持久 retry，应在子请求产生前对父 Agent 首次规划注入故障，恢复后再完成同一 run 的并行协作。首批同批双委派若只依赖自然语言提示会受模型波动影响，真实 suite 应使用正式 collaboration policy/preflight 固定两个指定 static Agent，并保留无正文的 batch 数量诊断。长链路还可能发生额外真实瞬态失败，不能用“全局 failed/retry 必须等于 1”把已正确恢复的网络抖动误判为注入失败；应按 request identity 锁定唯一受控链，额外 failure/retry 独立计数并继续执行全部 lifecycle、后继终态和零残留门禁。 - final-reply 恢复陷阱：不能把当前 Runtime `status / phase / currentAction` 投影或整份临时 tool-plan 放进要求跨进程稳定的请求指纹。前者在 `waiting -> planning -> response` 恢复过程中必然变化，后者的 `planUpdate/actions` 不会由 context bundle 原样保存；两者都会让合法 `-transient-N` 被误判为 drift。final-reply 必须在请求前同步 response 状态与 context bundle，并只使用可由 bundle 精确恢复的有界收束摘要；恢复 pass 先识别 `final-reply` 或 `final-reply-context-compaction` sidecar、恢复原 loop 并跳过新 planning。Agent DB lifecycle 的 requestKind 白名单也必须同步扩展，否则压缩请求会在网络调用前失败并被 planning fallback 掩盖。测试必须制造两种 sidecar-first 窗口、调用恢复扫描、按网络接收时间证明 `acceptedAtMs >= retryAtMs`，并比较失败/恢复 HTTP body 的 SHA-256 和字节一致性；失败输出不得打印正文片段。Provider 成功返回到压缩 sidecar 或 finalization journal `prepared` 之间仍不是 durable 提交点，进程退出可能重发 Provider；finalization journal 清理后才标记 stream committed 的窗口也可能留下 assistant 已落盘但 stream 仍为 ready。在新增成功响应 journal 与可恢复 stream commit 前不得宣称成功请求 exactly-once 或 stream 终态事务。 - 关联：`apps/ai-game-creator-shell/src-tauri/src/provider_retry.rs`、`agent.rs`、`runner.rs`、`tests.rs`、`docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md`。 + +## Provider 成功不等于已交接，stream ready 也不等于 finalization 已完成 + +- 现象：Provider 已返回完整 final-reply，Runner 在 finalization `prepared` 前退出后却再次请求；或 assistant/completed 已唯一落盘，response stream 长期停在 `ready / streaming`。更危险的修复是看到 handoff 与 retry 同时存在便择一删除，或因为另一个 Agent 已推进全局 project revision，就把当前固定 run 的 stream 当成不可见缓存并静默跳过提交。 +- 原因：Provider 网络 future、成功 handoff、compaction/finalization journal 和 response stream 是连续但不同的 durable owner。只有内存中的成功响应、`started` lifecycle、流式半句或 `ready` 展示缓存都不能证明下一 owner 已接管；面向 UI 的 stream 可见性还会读取当前全局 revision，不适合作为 finalization 的提交判据。 +- 正确顺序：成功响应先规范化并写入 `game-creator-provider-handoff.v1`，原子落盘并回读一致后，才用 handoff 保存的真实 requestId 补 `completed` lifecycle。恢复先修复该真实 requestId，再零网络回放。压缩结果先持久化并回读 compaction sidecar 后再清 handoff；final-reply 至少先进入 finalization `prepared`，journal 持续负责唯一 assistant、Runtime/Goal completed 和 stream committed，直到 committed 写入后的身份、状态、正文回读全部成功才清理。 +- 冲突处理：handoff/retry 只有完整 identity、attempt 和 slot 一致才可把 retry 当作已被成功结果覆盖；冲突时必须零网络进入 reconciliation，并保留两份 sidecar、`.previous` 和真实 requestId 证据。stream 身份或正文与 journal 冲突时也保留 journal；禁止覆盖冲突 stream、删除 journal、补造 requestId 或靠重复 Provider 调用“刷新”现场。Runner 对存在、备份或损坏 handoff 的 root 都必须报告 busy，不能为 idle shutdown 自动删证据。 +- stream 恢复：finalization 使用 journal v4 固定的 Agent/task/Session/run/request slot/steer cursor/response revision 和正文直接读取提交面。缺失或 `streaming` 可由 journal 重建为规范 `ready` 再提交；已 committed 且正文一致可幂等清理。即使全局 project revision 已被其它 Agent 推进，也不能跳过这个既定 run 的 stream；写入、回读、固定身份或正文任一不一致，都保持 journal 和可恢复 finalization。 +- 排障与验证：先核对 handoff 的 `providerRequestId / requestSlot / attempt` 与 Agent DB lifecycle，再看 retry/handoff/finalization/response-stream sidecar，最后才看 Runtime/UI 投影。用关闭 mock Provider 后恢复证明 handoff 回放零网络；分别覆盖 compaction 与 final-reply 消费窗口、handoff/retry 冲突、stream 缺失/streaming、commit 写失败、committed 后清理前退出、全局 revision 漂移和 Runner busy。日志与断言只公开指纹、字符数、状态和差异字段，不能打印 handoff 正文、请求体、凭据、URL 或绝对路径。 +- 保留边界：Runner 若在 Provider 成功后、handoff 原子提交并回读前被硬杀，本地仍只有结果未知的 `started`，不能安全补发或宣称 exactly-once。handoff 只覆盖无 tool call 的 `context-compaction / final-reply-context-compaction / final-reply`；`tool-plan` 及其 function arguments 不在内，真实外部 Provider 的 final-reply Runner 强杀门禁也需单独完成。 +- 关联：`apps/ai-game-creator-shell/src-tauri/src/provider_handoff.rs`、`provider_retry.rs`、`agent.rs`、`project.rs`、`runner.rs`、`tests.rs`、`docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md`。 diff --git a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md index d433bd6e4..8ea12db4a 100644 --- a/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md +++ b/docs/technical/【技术方案】AI游戏创作Agent Runtime V1.1-2026-07-12.md @@ -1391,16 +1391,64 @@ V1.40 把后台 `final-reply` 请求及其前置自动 context-compaction 分别 ### 验收边界 - 确定性测试必须分别停在“final-reply 或其前置压缩 retry sidecar 已提交、task/state 尚未投影”的窗口，确认恢复前 phase 仍为 response、重启扫描补齐 waiting 状态且网络接收时间不早于 `retryAt`；到期后必须直接恢复原 `-transient-1` final-reply，或先恢复原 `final-reply-context-compaction` 再继续 final-reply，不得多一次 tool-plan。失败与恢复两次 HTTP body 必须逐字节相同，只在失败信息中公开 SHA-256、字节数和首个差异位置，不公开正文。 -- 终态必须只有 1 条 completed task、1 条 conversation assistant 和 1 个 committed response stream；final-reply lifecycle 固定为初始 `started -> failed` 加恢复 `started -> completed`，retry audit/waiting 各 1，provider retry sidecar 为 0。前置压缩恢复还必须形成 `final-reply-context-compaction` 的两组唯一 lifecycle，并且不增加 tool-plan。当前确定性门禁为 `provider_retry_` 21/21、`provider_transient_retry_` 5/5、`response_stream_` 16/16，Tauri/Rust 串行全量 `969 passed / 4 ignored`。 +- 终态必须只有 1 条 completed task、1 条 conversation assistant 和 1 个 committed response stream；final-reply lifecycle 固定为初始 `started -> failed` 加恢复 `started -> completed`，retry audit/waiting 各 1，provider retry sidecar 为 0。前置压缩恢复还必须形成 `final-reply-context-compaction` 的两组唯一 lifecycle，并且不增加 tool-plan。最终实现的当前验证计数统一见 V1.41，不再沿用本切片较早的局部计数。 - V1.40 尚未以真实外部 Provider 完成 final-reply 退避期 Runner 强杀门禁，因此不得把 V1.39 首次 tool-plan 的真实 PASS 外推为 V1.40 PASS。后续独立 suite 必须在 Supervisor 已完整认领专业回执、完成 repair 和宿主验证后，对首次 final-reply 注入唯一瞬态故障；退避期强杀后证明原父 Session/run、request fingerprint、retryAt 和 slot 稳定，且 delivery/claim/receipt/finalization/assistant 无重复、终局零 sidecar/泄漏。手动压缩与 tool-plan `repair-N` 仍不在本切片内。 - Provider 成功返回到消费端 durable commit 之间仍有独立崩溃窗口：`final-reply-context-compaction` 的成功响应在压缩 sidecar 落盘前、final-reply 的成功响应在 finalization journal `prepared` 前都可能因 Runner 退出而丢失并重新请求 Provider。当前 response stream ready 不是 durable 结果提交协议；finalization journal 清理后到 stream 标记 committed 之间退出或写失败，还可能留下已持久化唯一 assistant、但展示 stream 仍为 ready 的终态差异。V1.40 只保证瞬态失败后的请求恢复与最终 assistant 幂等，不能把它扩大为成功请求 exactly-once 或 stream 终态事务；后续须以可重放成功响应 journal 和可恢复 stream commit 或等价提交协议单独解决。 +## V1.41 Provider 成功响应持久交接与回复流终态恢复 + +V1.41 为 V1.40 明确留下的成功响应交接窗口增加 `.agent/runtime/provider-handoffs//.json` 私有 sidecar，schema 固定为 `game-creator-provider-handoff.v1`。每个 Agent/run 最多一条记录，绑定完整 Provider retry identity、**实际已注册的物理 `providerRequestId`**、request slot/attempt、Provider/model、脱敏文本响应、finish reason、response ID、usage、响应指纹和创建时间。schema 拒绝未知字段，记录受字段长度、安全路径和 512 KiB 硬上限约束；写入使用 `0600` 临时文件、`sync_data`、原子替换、父目录同步和跨平台 `.previous` 恢复副本，写后必须回读并与内存记录完全一致。 + +本轮实际写入 handoff 的是 tool-plan 前置自动 `context-compaction`、final-reply 前置自动 `final-reply-context-compaction` 和 `final-reply` 三种无工具调用文本请求。手动压缩仍走非持久重试路径，不会写 handoff；tool-plan 本身的成功响应和 function arguments 也不在本轮。响应落盘前会删除 thinking block，并依次过滤密钥、项目路径和其它绝对路径；sidecar 不保存 prompt、请求消息、API Key、Provider URL、tool call/arguments 或错误正文，也不复制到公共 Runtime、event、Agent DB、CLI 或报告。 + +### Durable handoff 提交与 handoff-first 恢复 + +- 支持 handoff 的物理 Provider 请求返回成功后，Runtime 先用该次 lifecycle 实际解析出的 `requestId` 原子写入并回读 handoff，然后才允许同一 request identity 追加 `completed` 终态。handoff 写入、回读或内容校验失败时，原 lifecycle 保持只有 `started`，Runtime 进入 `needs-reconciliation`，不得自动补发该成功请求。 +- 每次可持久化请求在读 retry sidecar 或发起网络请求之前必须先读 handoff。完整 identity 相同时，Runtime 核对可选 retry sidecar 的 identity/attempt/slot，并把原实际 `providerRequestId` 的 `started` lifecycle 幂等补齐为 `completed`；随后删除匹配 retry sidecar 并回放响应，不注册新 `started`、不生成新 request identity，也不调用 Provider。 +- handoff 与 retry sidecar 同时存在时，后者必须精确指向 handoff 的 identity/attempt/slot；任一不符都进入 `needs-reconciliation`，不允许以任意一方覆盖另一方。同一 durable run 内若 Goal/steer/revision/request/config 等稳定 identity 已漂移，Runtime 先用旧 handoff 的实际 `requestId` 闭合旧 lifecycle，公共审计只记录漂移字段名，再删除旧 handoff/retry 并让同一 run 重新规划；跨 durable run 身份冲突则直接失败关闭。 +- cancel、生效 steer、非 pause Goal 控制、失败、budget-exhausted 和其它终态清理 handoff/retry；Goal pause 保留原交接记录，resume 后仍消费同一响应。`provider-handoffs` 中任何 primary、`.previous` 或损坏普通文件都是 Runner busy 事实，必须阻止 `runner.shutdown_if_idle`，直到恢复消费、明确作废或人工核对后清理。 + +### Compaction 消费与 finalization 固定流身份 + +- 自动 context-compaction 消费 handoff 后，先原子写入规范 `game-creator-runtime-context-compaction.v1` sidecar，再回读确认整条记录与本次结果完全一致，最后才删除匹配 handoff。若上次已完成 compaction 但在清理 handoff 前退出，恢复时以 source fingerprint、request kind 和 base slot 识别“没有新 source”，复用已提交 compaction 并幂等删除 handoff，不重新请求 Provider。compaction 回读或 handoff 清理失败保持 reconciliation。 +- 流式 final-reply 在 Provider 成功时调用 publisher `handoff`：持久化最新可见快照并停止 Drop 把它改成 discarded，但不提前写 ready。只有 handoff 完成且经规范脱敏响应回放后，才将同一 stream 身份写为 ready；中途 streaming 半句不是可提交 assistant。 +- finalization journal 正式升级为 `game-creator-runtime-finalization.v4`，在 `prepared` 时固定绑定 `responseRequestSlot`，并把该 slot 与 Agent/Session/run、response fingerprint/revision、steer cursor、计划和 Goal 快照一起纳入 `finalizationId` 指纹；篡改 slot 必须直接造成幂等身份不匹配。v3 journal 缺少 `responseRequestSlot` 仍可读，按 v1-v3 旧指纹规则校验，并在 stream commit 时由已绑定的 Runtime/finalization revision 与 steer 身份派生 legacy slot，不能反向伪装成 v4。`project.rs` 的 Agent DB `agent.runtime.finalization.lifecycle` 白名单必须兼容 `journalSchemaVersion` v1-v4，保证升级前四阶段 lifecycle 仍可扫描和幂等核对；lifecycle audit schema 本身继续为 v1。final-reply handoff 在 `prepared` 后仍保留，直到整个 finalization 和 stream commit 均完成后才与 journal 一起清理。 +- assistant、Runtime completed 和 Goal completed 幂等提交后，finalization 必须用 journal 固定身份读取回复流。流缺失，或仍为 `streaming / failed / discarded` 等非 ready 状态时，Runtime 以 journal 中的 canonical response 重建同身份 ready stream；已有流身份冲突则失败关闭。ready 正文必须与 journal 完全一致才能推进为 committed；已 committed 且正文一致是幂等成功，正文冲突则停止恢复。 +- committed 写入后必须立即回读，再次核对固定身份、`status=committed` 和完整正文；只有回读通过后才删除 finalization journal 与所属 handoff。stream 写入/回读失败，或在 `ResponseStreamCommitted` checkpoint 后、清理前退出，都保留 journal 和可恢复 `finalizing` 状态；Runner 重放同一 finalization，只补 stream commit 或幂等清理，不重写 assistant、不调用 Provider。 + +### Exactly-once 承诺边界 + +- V1.41 **只保证 handoff 已可靠落盘之后的 exactly-once 消费**：同一响应只能由匹配的 compaction 或 finalization 身份消费，恢复只回放该 handoff，不再请求 Provider，终局只有一条 canonical assistant/completed/committed stream。 +- **外部 Provider 已返回完整响应、但 Runtime 尚未写入 handoff 的硬杀窗口仍然不能承诺 Provider 调用 exactly-once**。该窗口没有可重放成功响应，只能把未闭合 `started` 视为未知并进入 reconciliation；后续人工处理可能需要重新请求 Provider，因此不得把消费幂等扩大为端到端 Provider 物理调用幂等。 +- tool-plan 成功响应/function arguments 的 durable handoff 另行设计；真实外部 Provider 的 final-reply 退避期 Runner 强杀也仍需独立 E2E，不得借用 V1.39 首次 tool-plan PASS 或本轮确定性 mock 门禁宣称已通过。 + +### 确定性测试矩阵 + +| 范围 | 故障/恢复窗口 | 必须断言 | 定向用例 | +| --- | --- | --- | --- | +| sidecar 契约 | 首次写入、同内容重写、`.previous`、损坏/超限/路径或内容冲突 | 严格 schema、`0600` 原子交接、实际 requestId 稳定、tool call 拒绝、thinking/密钥/路径零落盘 | `provider_handoff_*` 单元用例 | +| final-reply handoff-first | Provider 成功交接后、lifecycle 终态前停止 | 恢复零新网络请求，原 requestId `started -> completed`，唯一 assistant/completed/committed stream，终局零 retry/handoff/finalization | `provider_handoff_final_reply_restart_replays_success_without_network_request` | +| compaction consumer | 前置压缩 handoff 已落盘、compaction sidecar 未提交，以及 sidecar 已提交但 handoff 未清理 | 压缩结果回读后清理，不重放 tool-plan/压缩，只发后续 final-reply | `provider_handoff_final_reply_compaction_restart_only_requests_final_reply` | +| identity/retry 冲突 | Goal/steer/revision/request/config 漂移，或 retry identity/attempt/slot 与 handoff 不一致 | 漂移先闭合旧 lifecycle 再作废，审计零响应正文；retry 冲突进入 reconciliation 且零 Provider 请求 | `provider_handoff_identity_drift_closes_lifecycle_without_leaking_response` 及 `provider_handoff_` 冲突门禁 | +| finalization schema/身份 | v4 slot 被篡改，或读取缺少 slot 的 v3 journal 与 v1-v4 lifecycle | v4 `responseRequestSlot` 进入 `finalizationId` 指纹；v3 按旧身份可读；Agent DB lifecycle 白名单接受 v1-v4 journal schema | `finalization_v4_binds_response_request_slot_into_identity`、`finalization_v3_without_response_request_slot_remains_readable`、`finalization_lifecycle_accepts_supported_v1_through_v4_journals` | +| finalization 四 checkpoint | `Prepared / AssistantAppended / RuntimeCompleted / ResponseStreamCommitted` 任一点中断 | 原 messageId/finalizationId 幂等恢复，无 Provider/工具重放，最终 journal/handoff 清理 | `finalization_*` 与 `response_stream_committed_checkpoint_recovers_by_idempotent_cleanup` | +| 回复流重建 | stream 缺失、停在 streaming、commit 写失败、已 committed 后清理中断 | 按 v4 固定 slot 从 journal 重建 ready，committed 写后回读，正文唯一，冲突失败关闭 | `response_stream_finalization_recovers_missing_stream_after_project_revision_drift`、`response_stream_finalization_repairs_streaming_after_commit_write_failure`、`response_stream_committed_checkpoint_recovers_by_idempotent_cleanup` | +| Runner busy | primary、`.previous` 或损坏 handoff 存在时请求 idle shutdown | `idle=false`，不进入 draining；清理后才可 shutdown | `durable_provider_handoff_prevents_shutdown_even_when_corrupt` | + +2026-07-20 当前最终实现的最新验证证据为：`provider_retry_` 21/21、`response_stream_` 23/23、`finalization_resume_` 12/12；Tauri/Rust 串行全量共 989 tests，`985 passed / 4 ignored / 0 failed`。这些数字替代 V1.40 较早快照，后续当前结果统一使用本行口径。 + +上表是 V1.41 的确定性门禁，不代表真实外部 Provider E2E 结论。除表内窗口外，还必须继续扫描 task/event/Agent DB/CLI/report，确认 Provider 响应、compaction summary、API Key、Provider URL 和项目/配置绝对路径公共泄漏均为 0。 + ## 验收命令 - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml structured_plan_ -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml agent_goal_ -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml goal_context_bundle_v4_migrates_v3_and_v2_then_rejects_plan_mismatch -- --nocapture` +- `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml provider_handoff_ -- --nocapture --test-threads=1` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml response_stream_ -- --nocapture` +- `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml finalization_v4_binds_response_request_slot_into_identity -- --nocapture` +- `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml finalization_v3_without_response_request_slot_remains_readable -- --nocapture` +- `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml finalization_lifecycle_accepts_supported_v1_through_v4_journals -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml mcp_ -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml provider_action_batch_ -- --nocapture` - `cargo test --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml provider_retry_ -- --nocapture --test-threads=1` diff --git a/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md b/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md index c0c2a1fe8..846ab24d9 100644 --- a/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md +++ b/docs/technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md @@ -607,3 +607,7 @@ game-project/ - 已有 durable/未观察 claim 与 legacy claimed delivery 继续按原 action/group 身份恢复，不要求先创建新绑定；新 claim 必须先成功解析 effective snapshot 并核对 binding，再进入 V1.35-V1.37 的全锁、预算、完整 observation 和 group 数量门禁。snapshot 绑定后 global policy 的 `matched / drifted / unreadable` 只进入有界 status/诊断，不能改变后续执行；新 policy 只由后续新父 run 采用。2026-07-19 self-test、52/52 collaboration 定向回归和 949 passed/4 ignored Rust 全量已完成，终态快照保留也有独立回归；真实 mixed-swarm 功能样本已闭合但受正式 endpoint 外部重启污染，私有配置源的后续独立运行又连续耗尽 transient Provider retry，不能拼接证据，当前仍**不得声称 V1.38 真实 E2E 已 PASS**。详细报告以 Runtime 技术方案 V1.38 节为准。 - 2026-07-19 起，同一 Runtime 文档的“V1.39 首次规划 Provider 瞬态重试持久等待态”作为后台首次规划重试的恢复事实源。tool-plan `repair-0` 及其自动 context-compaction 在瞬态失败后先写 per-Agent/run retry sidecar，再投影 `waiting-for-provider-retry` 并释放 lane；Runner 重启按到期时间恢复同一 Session/run/loop/attempt，同 Agent 后续任务保持 FIFO，其它 Agent 可并行。final-reply、手动压缩和 tool-plan `repair-N` 暂不扩展为持久重试；工具策略刷新时间不得进入请求指纹。最终代码已完成独立真实 `gpt-5.5 / openai_chat / high` PASS：父 Agent 首次规划在 `30s` 退避期执行一次 pidfd Runner 强杀，重启后 sidecar 身份/attempt/retryAt 稳定，到期前零早发且第二个请求网络接收时间不早于 retryAt；随后同一父 run 完成双专业 Agent 真重叠、2+1 delivery、唯一 repair、唯一最终回复和零重复/残留/正文/Key/路径泄漏，`37/37` lifecycle 闭合为 `36 completed + 1 injected failed + 1 retry`，incidental failure/retry 均为 0，隔离现场完整清理。额外真实瞬态失败按 request identity 单独计数且仍须完整恢复，不能混入受控注入链。该证据不改变 V1.38 真实 E2E 尚未 PASS 的独立结论。 - 2026-07-19 V1.40 在上述 sidecar 协议上补齐后台 `final-reply` 和其前置自动 context-compaction，分别使用 `requestKind=final-reply / final-reply-context-compaction`。final-reply prompt 改用可由 context bundle 精确恢复的有界收束摘要，不包含会随恢复阶段变化的 Runtime 投影或无法持久重建的临时 tool-plan 字段；response 状态先与 context bundle 同步，再允许发出 Provider 请求。Runner 恢复命中任一 final-reply sidecar 时按原 `nextLoopIndex / baseRequestSlot / requestFingerprint` 跳过新 planning，压缩链先完成原压缩再继续 final reply；流式失败半句不提交，成功后仍只经原 finalization journal 写一条 canonical assistant。稳定身份漂移则删除旧 sidecar并在同一 run 重新规划。确定性门禁已覆盖两种 sidecar-first 投影窗口、恢复扫描、网络接收时间不早于 `retryAt`、失败/恢复 HTTP body 逐字节一致、唯一 lifecycle/assistant/stream 和终局零 sidecar；当前 `provider_retry_` 为 21/21。真实外部 Provider 的 final-reply 退避期 Runner 强杀尚未独立 PASS；成功响应到压缩 sidecar/finalization journal 的交接窗口，以及 journal 清理到 stream committed 的窗口也尚未补齐，不能借用 V1.39 首次 tool-plan 证据或宣称成功请求 exactly-once/stream 终态事务。手动压缩与 tool-plan `repair-N` 仍保持进程内重试。 +- 2026-07-20 起，同一 Runtime 文档的“V1.41 Provider 成功响应持久交接与回复流终态恢复”覆盖 V1.40 的两个成功窗口。`.agent/runtime/provider-handoffs//.json` 使用严格 `game-creator-provider-handoff.v1`，绑定完整 retry identity、实际物理 `providerRequestId`、slot/attempt 和经 thinking、密钥、项目路径及绝对路径过滤后的规范文本响应；写入采用 `0600` 原子 sidecar、`.previous` 恢复和写后完整回读。当前仅覆盖 tool-plan 前置自动 `context-compaction`、final-reply 前置自动 `final-reply-context-compaction` 与 `final-reply`；手动压缩和 tool-plan 成功响应/function arguments 不写 handoff。 +- V1.41 对支持范围执行 handoff-first：Provider 成功后先提交并回读 handoff，再闭合同一实际 requestId 的 `completed` lifecycle；恢复先消费 handoff，匹配 retry 时复用原 attempt/slot 且零新网络请求，retry identity/attempt/slot 冲突进入 reconciliation，稳定身份漂移先闭合旧 lifecycle 再作废并同 run 重规划。自动 compaction 必须在规范 sidecar 写入并回读一致后才清理 handoff；final-reply handoff 则保留到 finalization 和 stream commit 全部完成。任一 primary、`.previous` 或损坏 handoff 都让 Runner 保持 busy，阻止 `runner.shutdown_if_idle`。 +- finalization schema 已正式升为 `game-creator-runtime-finalization.v4`：`responseRequestSlot` 与 response revision/fingerprint、steer cursor、计划及 Goal 快照共同进入 `finalizationId` 指纹，篡改 slot 必须失败关闭；`project.rs` 的 Agent DB finalization lifecycle 白名单兼容 `journalSchemaVersion` v1-v4，缺少 slot 的 v3 journal 继续按旧指纹和 legacy slot 派生规则读取。assistant、Runtime 和 Goal 提交后，缺失或仍为 streaming/failed/discarded 的回复流由 journal 固定身份重建为 ready，再推进 committed；committed 写后必须回读身份、状态和正文，成功后才一起删除 finalization journal/handoff，失败则保留 `finalizing` 供 Runner 幂等恢复。确定性测试矩阵以 Runtime V1.41 章节中的 handoff、compaction、四 checkpoint、stream 重建、schema 兼容和 Runner busy 用例为准。2026-07-20 最新验证为 `provider_handoff_` 11/11、`provider_retry_` 21/21、`response_stream_` 23/23、`finalization_resume_` 12/12；Tauri/Rust 串行全量共 989 tests，`985 passed / 4 ignored / 0 failed`，不再沿用 V1.40 的旧局部计数。 +- V1.41 的承诺严格限定为 **handoff 已落盘后的 exactly-once 消费**。外部 Provider 已返回完整响应、但 Runtime 尚未写 handoff 的硬杀窗口仍不能承诺 Provider 调用 exactly-once，后续只能进入 reconciliation 并可能需要人工决定是否重发；tool-plan 成功响应仍不在本轮 handoff。真实外部 Provider 的 final-reply 退避期 Runner 强杀继续是独立 E2E，不能用 V1.39 首次 tool-plan PASS 或确定性 mock 结果替代。