Add atomic agent turn finalization

README.md

@@ -166,7 +166,7 @@ curl -sS http://localhost:8090/v1/conversations/conv_.../turns \
   -H 'content-type: application/json' \
   -H 'x-threadmark-tenant: acme' \
   -H 'x-threadmark-principal: user_123' \
-  -d '{"idempotency_key":"request-1","agent_ref":"research-agent/prod"}'
+  -d '{"idempotency_key":"request-1","agent_ref":"research-agent/prod","response_id":"resp_abc"}'
 ```
 
 Append an Open Responses user item:
@@ -238,6 +238,7 @@ GET /v1/continuations/resp_abc?agent_ref=research-agent%2Fprod
 | `POST` | `/v1/conversations/{id}/replay` | Build an Open Responses input array |
 | `POST` | `/v1/conversations/{id}/turns` | Create an idempotent turn |
 | `PATCH` | `/v1/turns/{id}` | Update turn state and outcome |
+| `POST` | `/v1/agent-turns/{id}/finalize` | Atomically store terminal output, response, checkpoint, and turn outcome |
 | `POST` | `/v1/conversations/{id}/continuations` | Record an agent checkpoint |
 | `GET` | `/v1/continuations/{response_id}` | Resolve an agent checkpoint |
 | `POST` | `/v1/files` | Upload a tenant-owned S3-backed file |

migrations/0008_atomic_agent_turn_finalization.sql

@@ -0,0 +1,27 @@
+ALTER TABLE turns ADD COLUMN reserved_response_id text;
+
+-- Preserve response IDs already assigned to active turns before this
+-- migration. Legacy active turns without one can reserve their response ID
+-- during their first transactional finalization.
+UPDATE turns
+SET reserved_response_id = response_id
+WHERE status IN ('pending', 'streaming') AND response_id IS NOT NULL;
+
+CREATE TABLE turn_finalizations (
+    turn_id text PRIMARY KEY REFERENCES turns(id) ON DELETE CASCADE,
+    tenant_id text NOT NULL,
+    owner_ref text NOT NULL,
+    agent_ref text NOT NULL,
+    idempotency_key text NOT NULL,
+    response_id text NOT NULL,
+    request_version smallint NOT NULL CHECK (request_version > 0),
+    request_digest bytea NOT NULL CHECK (octet_length(request_digest) = 32),
+    response jsonb NOT NULL,
+    response_digest bytea NOT NULL CHECK (octet_length(response_digest) = 32),
+    first_seq bigint NOT NULL CHECK (first_seq > 0),
+    last_seq bigint NOT NULL CHECK (last_seq >= first_seq - 1),
+    continuation_id text NOT NULL REFERENCES continuations(id) ON DELETE CASCADE,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    UNIQUE (tenant_id, owner_ref, agent_ref, idempotency_key),
+    UNIQUE (tenant_id, agent_ref, response_id)
+);

src/api.rs

@@ -19,9 +19,10 @@ use crate::{
     model::{
         Actor, AppendItems, AppendResult, Continuation, ContinuationQuery, Conversation,
         CreateContinuation, CreateConversation, CreateDownload, CreateTurn, DownloadDelivery,
-        DownloadGrant, FileResponse, Item, ListConversationsQuery, ListItemsQuery,
-        RegenerateResult, ReplayRequest, ReplayResult, StartTurn, StartTurnResult, StrictJson,
-        TruncateConversation, Turn, UpdateConversation, UpdateTurn, validate_json_number_tokens,
+        DownloadGrant, FileResponse, FinalizeTurn, FinalizeTurnResult, Item,
+        ListConversationsQuery, ListItemsQuery, RegenerateResult, ReplayRequest, ReplayResult,
+        StartTurn, StartTurnResult, StrictJson, TruncateConversation, Turn, UpdateConversation,
+        UpdateTurn, validate_json_number_tokens,
     },
     object_store::ObjectStore,
     store, uploads,
@@ -60,6 +61,7 @@ pub fn router(state: AppState) -> Router {
         )
         .route("/v1/conversations/{id}/active-turn", get(get_active_turn))
         .route("/v1/turns/{id}", get(get_turn).patch(update_turn))
+        .route("/v1/agent-turns/{id}/finalize", post(finalize_turn))
         .route(
             "/v1/conversations/{id}/truncate",
             post(truncate_conversation),
@@ -287,6 +289,20 @@ async fn update_turn(
     ))
 }
 
+async fn finalize_turn(
+    State(state): State<AppState>,
+    auth: AuthContext,
+    Path(id): Path<String>,
+    Json(request): Json<FinalizeTurn>,
+) -> ApiResult<Json<FinalizeTurnResult>> {
+    auth.require(Permission::TranscriptAppend)?;
+    auth.require(Permission::TurnUpdate)?;
+    auth.require(Permission::ContinuationWrite)?;
+    Ok(Json(
+        store::finalize_turn(&state.pool, &auth, &id, request).await?,
+    ))
+}
+
 async fn create_continuation(
     State(state): State<AppState>,
     auth: AuthContext,

src/model.rs

@@ -35,6 +35,7 @@ pub struct StartTurn {
     pub conversation_id: Option<String>,
     pub conversation: Option<CreateConversation>,
     pub agent_ref: String,
+    pub response_id: String,
     pub items: Vec<Value>,
 }
 
@@ -353,6 +354,31 @@ pub struct Turn {
 pub struct CreateTurn {
     pub idempotency_key: String,
     pub agent_ref: String,
+    pub response_id: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct FinalizeTurn {
+    pub idempotency_key: String,
+    pub response_id: String,
+    pub status: String,
+    pub output_items: Vec<Value>,
+    pub response: Value,
+    pub parent_response_id: Option<String>,
+    pub state: Option<Value>,
+    pub error: Option<Value>,
+    pub usage: Option<Value>,
+}
+
+#[derive(Debug, Serialize)]
+pub struct FinalizeTurnResult {
+    pub turn: Turn,
+    pub items: Vec<Item>,
+    pub continuation: Continuation,
+    pub response: Value,
+    pub first_seq: i64,
+    pub last_seq: i64,
+    pub replayed: bool,
 }
 
 #[derive(Debug, Deserialize)]