perf(loop): write finalized assistant reply via one-shot append path

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<LoopMessageRef, AgentLoopHostError> {
         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<S> ThreadBackedLoopTranscriptPort<S>
 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<Option<ThreadMessageRecord>, 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<Option<ThreadMessageId>, 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.

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;