diff --git a/crates/ironclaw_loop_support/src/lib.rs b/crates/ironclaw_loop_support/src/lib.rs index 165272fe7ba..905b1804bef 100644 --- a/crates/ironclaw_loop_support/src/lib.rs +++ b/crates/ironclaw_loop_support/src/lib.rs @@ -125,10 +125,11 @@ use tokio::sync::{Mutex, OnceCell}; use async_trait::async_trait; use ironclaw_threads::{ - AppendAssistantDraftRequest, AppendToolResultReferenceRequest, ContextMessage, + AppendAssistantDraftRequest, AppendFinalizedAssistantMessageRequest, + AppendToolResultReferenceRequest, ContextMessage, FinalizedAssistantMessageByRunRequest, LoadContextMessagesRequest, LoadContextWindowRequest, MessageContent, MessageKind, - MessageStatus, ProviderToolCallReferenceEnvelope, SessionThreadError, SessionThreadService, - SummaryArtifact, ThreadHistoryRequest, ThreadMessageId, ThreadMessageRecord, ThreadScope, + ProviderToolCallReferenceEnvelope, SessionThreadError, SessionThreadService, SummaryArtifact, + ThreadHistoryRequest, ThreadMessageId, ThreadMessageRecord, ThreadScope, ToolResultReferenceEnvelope, ToolResultSafeSummary, UpdateAssistantDraftRequest, }; use ironclaw_turns::{ @@ -608,57 +609,54 @@ where ) -> Result { validate_thread_scope_for_run(&self.thread_scope, &self.run_context)?; let reply_content = request.reply.content; - let draft = self + // One-shot finalize keyed on `turn_run_id`. This collapses the former + // append-draft-then-CAS-finalize two-write sequence into a single + // call: on the filesystem backend with no prior draft it takes the + // append-only fast path (one log append + sequence index, no + // per-message file rewrite); if the loop streamed a draft first + // (`begin_assistant_draft`), it resolves that draft by run and + // finalizes it in place. It is idempotent for a resumed/retried turn — + // a second call returns the already-finalized message. + let finalized = match self .thread_service - .append_assistant_draft(AppendAssistantDraftRequest { + .append_finalized_assistant_message(AppendFinalizedAssistantMessageRequest { scope: self.thread_scope.clone(), thread_id: self.run_context.thread_id.clone(), turn_run_id: self.run_context.run_id.to_string(), content: MessageContent::text(reply_content.clone()), }) .await - .map_err(transcript_write_error)?; - if draft.status == MessageStatus::Finalized { - if draft.content.as_deref() == Some(reply_content.as_str()) { - let message_ref = message_ref(draft.message_id)?; - self.emit_assistant_reply_finalized(message_ref.clone()) - .await?; - return Ok(message_ref); + { + Ok(message) => message, + Err(append_error) => { + // Concurrency convergence: a racing finalize for the same run + // can win the CAS first, leaving this call's finalize to fail. + // If a finalized reply for this run already exists with matching + // content, converge on it rather than failing the turn — this + // preserves the idempotent-under-concurrent-duplicate contract + // the former append-draft-then-finalize path provided. + match self + .finalized_reply_for_run_matching(&reply_content) + .await? + { + Some(existing) => existing, + None => return Err(transcript_write_error(append_error)), + } } + }; + // Preserve the prior contract: if a finalized reply already exists for + // this run with different content (a divergent re-run), surface a + // transcript write failure rather than silently keeping stale content. + if finalized.content.as_deref() != Some(reply_content.as_str()) { return Err(AgentLoopHostError::new( AgentLoopHostErrorKind::TranscriptWriteFailed, "assistant transcript write failed", )); } - let finalized = self - .thread_service - .finalize_assistant_message( - &self.thread_scope, - &self.run_context.thread_id, - draft.message_id, - MessageContent::text(reply_content.clone()), - ) - .await; - match finalized { - Ok(message) => { - let message_ref = message_ref(message.message_id)?; - self.emit_assistant_reply_finalized(message_ref.clone()) - .await?; - Ok(message_ref) - } - Err(error) => { - if let Some(message_id) = self - .already_finalized_matching_reply(draft.message_id, &reply_content) - .await? - { - let message_ref = message_ref(message_id)?; - self.emit_assistant_reply_finalized(message_ref.clone()) - .await?; - return Ok(message_ref); - } - Err(transcript_write_error(error)) - } - } + let message_ref = message_ref(finalized.message_id)?; + self.emit_assistant_reply_finalized(message_ref.clone()) + .await?; + Ok(message_ref) } async fn append_capability_result_ref( @@ -723,6 +721,25 @@ impl ThreadBackedLoopTranscriptPort where S: SessionThreadService + ?Sized + Send + Sync, { + /// Look up the finalized assistant reply for this run, returning it only + /// when its content matches `reply_content`. Used to converge a finalize + /// that lost a concurrent race onto the winner's message. + async fn finalized_reply_for_run_matching( + &self, + reply_content: &str, + ) -> Result, AgentLoopHostError> { + let existing = self + .thread_service + .finalized_assistant_message_by_run(FinalizedAssistantMessageByRunRequest { + scope: self.thread_scope.clone(), + thread_id: self.run_context.thread_id.clone(), + turn_run_id: self.run_context.run_id.to_string(), + }) + .await + .map_err(transcript_write_error)?; + Ok(existing.filter(|message| message.content.as_deref() == Some(reply_content))) + } + async fn emit_assistant_reply_finalized( &self, message_ref: LoopMessageRef, @@ -779,32 +796,6 @@ where } Ok(message) } - - async fn already_finalized_matching_reply( - &self, - message_id: ThreadMessageId, - reply_content: &str, - ) -> Result, AgentLoopHostError> { - let history = self - .thread_service - .list_thread_history(ThreadHistoryRequest { - scope: self.thread_scope.clone(), - thread_id: self.run_context.thread_id.clone(), - }) - .await - .map_err(transcript_write_error)?; - let expected_run_id = self.run_context.run_id.to_string(); - Ok(history.messages.into_iter().find_map(|message| { - let belongs_to_run = message.turn_run_id.as_deref() == Some(expected_run_id.as_str()); - let matches_reply = message.status == MessageStatus::Finalized - && message.content.as_deref() == Some(reply_content); - if message.message_id == message_id && belongs_to_run && matches_reply { - Some(message.message_id) - } else { - None - } - })) - } } /// Empty capability surface for the text-only loop-support MVP. diff --git a/crates/ironclaw_loop_support/tests/thread_loop_support_contract.rs b/crates/ironclaw_loop_support/tests/thread_loop_support_contract.rs index 45993f2d0f0..1567aa4e004 100644 --- a/crates/ironclaw_loop_support/tests/thread_loop_support_contract.rs +++ b/crates/ironclaw_loop_support/tests/thread_loop_support_contract.rs @@ -2252,6 +2252,126 @@ async fn transcript_port_finalizes_assistant_reply_into_durable_thread_history() ); } +/// Finalizing twice for the same run (e.g. a resumed/retried turn) must be +/// idempotent: one finalized assistant row, same message ref both times. The +/// one-shot `append_finalized_assistant_message` path the port now uses keys on +/// `turn_run_id`, so the second call resolves the existing finalized message +/// rather than appending a duplicate. +#[tokio::test] +async fn transcript_port_finalize_assistant_reply_is_idempotent_for_one_run() { + let fixture = ThreadFixture::new().await; + let adapter = ThreadBackedLoopTranscriptPort::new( + Arc::clone(&fixture.thread_service), + fixture.thread_scope.clone(), + fixture.run_context.clone(), + ); + + let first = adapter + .finalize_assistant_message(FinalizeAssistantMessage { + reply: AssistantReply { + content: "final answer".to_string(), + }, + }) + .await + .unwrap(); + let second = adapter + .finalize_assistant_message(FinalizeAssistantMessage { + reply: AssistantReply { + content: "final answer".to_string(), + }, + }) + .await + .unwrap(); + + assert_eq!(first.as_str(), second.as_str()); + let history = fixture + .thread_service + .list_thread_history(ThreadHistoryRequest { + scope: fixture.thread_scope.clone(), + thread_id: fixture.thread_id.clone(), + }) + .await + .unwrap(); + let assistant_rows: Vec<_> = history + .messages + .iter() + .filter(|message| message.kind == MessageKind::Assistant) + .collect(); + assert_eq!( + assistant_rows.len(), + 1, + "double finalize must not materialize a second assistant row" + ); + assert_eq!(assistant_rows[0].status, MessageStatus::Finalized); + assert_eq!(assistant_rows[0].content.as_deref(), Some("final answer")); +} + +/// When the loop streamed a draft first (`begin_assistant_draft` + +/// `update_assistant_draft`), finalizing must finalize that existing draft IN +/// PLACE — same message id, final content — not append a second message. The +/// one-shot path resolves the streamed draft by `turn_run_id` and finalizes it. +#[tokio::test] +async fn transcript_port_finalize_finalizes_existing_streamed_draft_in_place() { + let fixture = ThreadFixture::new().await; + let adapter = ThreadBackedLoopTranscriptPort::new( + Arc::clone(&fixture.thread_service), + fixture.thread_scope.clone(), + fixture.run_context.clone(), + ); + + let draft_ref = adapter + .begin_assistant_draft(BeginAssistantDraft { + reply: AssistantReply { + content: "partial".to_string(), + }, + }) + .await + .unwrap(); + adapter + .update_assistant_draft(UpdateAssistantDraft { + message_ref: draft_ref.clone(), + reply: AssistantReply { + content: "partial answer".to_string(), + }, + }) + .await + .unwrap(); + + let finalized_ref = adapter + .finalize_assistant_message(FinalizeAssistantMessage { + reply: AssistantReply { + content: "partial answer complete".to_string(), + }, + }) + .await + .unwrap(); + + assert_eq!( + finalized_ref.as_str(), + draft_ref.as_str(), + "finalize must finalize the streamed draft in place, not create a new message" + ); + let history = fixture + .thread_service + .list_thread_history(ThreadHistoryRequest { + scope: fixture.thread_scope.clone(), + thread_id: fixture.thread_id.clone(), + }) + .await + .unwrap(); + let assistant_rows: Vec<_> = history + .messages + .iter() + .filter(|message| message.kind == MessageKind::Assistant) + .collect(); + assert_eq!(assistant_rows.len(), 1); + assert_eq!(assistant_rows[0].status, MessageStatus::Finalized); + assert_eq!( + assistant_rows[0].content.as_deref(), + Some("partial answer complete") + ); +} + #[tokio::test] async fn transcript_port_appends_tool_result_reference_envelope_idempotently() { let fixture = ThreadFixture::new().await;