Store immutable public responses for owner-scoped recovery

Cargo.toml

@@ -21,7 +21,7 @@ hmac = "0.12"
 jsonwebtoken = { version = "10.3", features = ["rust_crypto"] }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+serde_json = { version = "1", features = ["raw_value"] }
 serde_json_canonicalizer = "0.3"
 sha2 = "0.10"
 sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "chrono", "json", "migrate"] }

README.md

@@ -78,6 +78,7 @@ This is an experiment and its API is not stable. The current slice establishes:
 - Cursor-based item reads.
 - Open Responses replay projection with optional top-level `id` removal.
 - Agent-scoped continuation records and optional private checkpoint state.
+- Immutable terminal public responses with owner-scoped recovery by response ID.
 
 Editing, branching, retention, event delivery, production authentication, and
 fine-grained capabilities are intentionally deferred until the core contract is
@@ -226,6 +227,56 @@ The agent can resolve that state later with:
 GET /v1/continuations/resp_abc?agent_ref=research-agent%2Fprod
 ```
 
+### Store and recover a terminal response
+
+After completing the turn (including setting its `response_id` and terminal
+status), persist the exact public Open Responses object and its continuation in
+one transaction:
+
+```bash
+curl -sS http://localhost:8090/v1/conversations/conv_.../responses \
+  -H 'content-type: application/json' \
+  -H 'x-threadmark-tenant: acme' \
+  -H 'x-threadmark-principal: user_123' \
+  -d '{
+    "agent_ref":"research-agent/prod",
+    "turn_id":"turn_...",
+    "response_created_at":"2026-08-17T10:00:00Z",
+    "terminal_at":"2026-08-17T10:00:03Z",
+    "public_response":{
+      "id":"resp_abc",
+      "object":"response",
+      "status":"completed",
+      "previous_response_id":null,
+      "output":[],
+      "usage":{"input_tokens":10,"output_tokens":4}
+    },
+    "state":{"provider_thread":"thread_xyz"}
+  }'
+```
+
+`schema_marker` defaults to `open-responses/public-response/v1`, and
+`through_seq` defaults to the current transcript boundary. The response must be
+a terminal `object: "response"`, must match the terminal turn's agent, status,
+and response ID, and is limited to 1 MiB of canonical JSON. Duplicate keys,
+non-canonical JSON numbers, and unknown schema markers are rejected. An exact
+retry returns `200`; reusing the scoped response ID for different content or
+linkage returns `409`.
+
+Recover the public object without exposing private continuation state:
+
+```text
+GET /v1/responses/resp_abc?agent_ref=research-agent%2Fprod
+```
+
+The endpoint returns the validated stored JSON text itself, preserving object
+key order, array order, and JSON representation rather than projecting from
+ledger items. A JSONB validation copy, canonical SHA-256 digest, and versioned
+schema marker are verified on every read. Missing, wrong-owner, and wrong-agent
+lookups all return `404`; malformed stored data returns a generic `500` and is
+never served. Callers need `continuation:write` to store and
+`continuation:read` to retrieve responses.
+
 ## API summary
 
 | Method | Path | Purpose |
@@ -240,6 +291,8 @@ GET /v1/continuations/resp_abc?agent_ref=research-agent%2Fprod
 | `PATCH` | `/v1/turns/{id}` | Update turn state and outcome |
 | `POST` | `/v1/conversations/{id}/continuations` | Record an agent checkpoint |
 | `GET` | `/v1/continuations/{response_id}` | Resolve an agent checkpoint |
+| `POST` | `/v1/conversations/{id}/responses` | Atomically store a terminal public response and continuation |
+| `GET` | `/v1/responses/{response_id}` | Recover an owner- and agent-scoped public response |
 | `POST` | `/v1/files` | Upload a tenant-owned S3-backed file |
 | `GET` | `/v1/files/{id}` | Read owned file metadata |
 | `DELETE` | `/v1/files/{id}` | Delete an unreferenced owned file |
@@ -252,10 +305,18 @@ GET /v1/continuations/resp_abc?agent_ref=research-agent%2Fprod
   item to be a JSON object.
 - Sequence numbers are allocated while locking the conversation row. Concurrent
   append requests therefore have deterministic, non-overlapping order.
-- Continuations are namespaced by tenant and `agent_ref`; the same response ID
-  may safely exist for unrelated agents or tenants.
+- Continuations and stored responses are namespaced by tenant, owner, and
+  `agent_ref`; the same response ID may safely exist for unrelated owners,
+  agents, or tenants. Migration `0008` backfills continuation owners from their
+  conversations before replacing the legacy uniqueness constraint.
+- Terminal public responses are immutable at the database layer. They retain
+  the response/previous-response IDs, terminal status, turn and continuation
+  links, transcript boundary, response and terminal timestamps, version marker,
+  canonical size, digest, and the complete public JSON object. Deleting or
+  truncating the owning conversation may remove them as part of normal ledger
+  lifecycle, but they cannot be updated in place.
 - Private continuation `state` is returned only through the continuation API.
-  A future capability system must prevent ordinary UI clients from reading it.
+  The ordinary response API selects only the public response object.
 - The replay endpoint is a convenience projection, not summarization. The
   canonical item ledger remains lossless.
 - Capability signatures bind tenant, owner, file ID, and expiry. Capability

migrations/0008_stored_responses.sql

@@ -0,0 +1,112 @@
+ALTER TABLE continuations ADD COLUMN owner_ref text;
+ALTER TABLE continuations ADD COLUMN turn_id text REFERENCES turns(id) ON DELETE CASCADE;
+
+UPDATE continuations continuation
+SET owner_ref = conversation.owner_ref
+FROM conversations conversation
+WHERE conversation.id = continuation.conversation_id;
+
+ALTER TABLE continuations ALTER COLUMN owner_ref SET NOT NULL;
+ALTER TABLE continuations
+    DROP CONSTRAINT continuations_tenant_id_agent_ref_response_id_key;
+ALTER TABLE continuations
+    ADD CONSTRAINT continuations_tenant_owner_agent_response_key
+    UNIQUE (tenant_id, owner_ref, agent_ref, response_id);
+
+ALTER TABLE conversations
+    ADD CONSTRAINT conversations_id_tenant_owner_key
+    UNIQUE (id, tenant_id, owner_ref);
+ALTER TABLE turns
+    ADD CONSTRAINT turns_id_conversation_agent_key
+    UNIQUE (id, conversation_id, agent_ref);
+ALTER TABLE continuations
+    ADD CONSTRAINT continuations_identity_link_key
+    UNIQUE (id, tenant_id, owner_ref, conversation_id, turn_id, agent_ref,
+            response_id, through_seq);
+ALTER TABLE continuations
+    ADD CONSTRAINT continuations_owned_conversation_fkey
+    FOREIGN KEY (conversation_id, tenant_id, owner_ref)
+    REFERENCES conversations (id, tenant_id, owner_ref) ON DELETE CASCADE;
+ALTER TABLE continuations
+    ADD CONSTRAINT continuations_turn_link_fkey
+    FOREIGN KEY (turn_id, conversation_id, agent_ref)
+    REFERENCES turns (id, conversation_id, agent_ref) ON DELETE CASCADE;
+
+CREATE INDEX continuations_owner_agent_response_idx
+    ON continuations (tenant_id, owner_ref, agent_ref, response_id);
+
+CREATE TABLE stored_responses (
+    id text PRIMARY KEY,
+    tenant_id text NOT NULL,
+    owner_ref text NOT NULL,
+    agent_ref text NOT NULL,
+    conversation_id text NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
+    turn_id text NOT NULL REFERENCES turns(id) ON DELETE CASCADE,
+    continuation_id text NOT NULL UNIQUE REFERENCES continuations(id) ON DELETE CASCADE,
+    response_id text NOT NULL,
+    previous_response_id text,
+    terminal_status text NOT NULL
+        CHECK (terminal_status IN ('completed', 'incomplete', 'failed', 'cancelled')),
+    public_response jsonb NOT NULL CHECK (jsonb_typeof(public_response) = 'object'),
+    public_response_text text NOT NULL
+        CHECK (octet_length(public_response_text) BETWEEN 2 AND 1048576),
+    canonical_digest bytea NOT NULL CHECK (octet_length(canonical_digest) = 32),
+    schema_marker text NOT NULL,
+    canonical_size bigint NOT NULL CHECK (canonical_size BETWEEN 2 AND 1048576),
+    through_seq bigint NOT NULL CHECK (through_seq >= 0),
+    response_created_at timestamptz NOT NULL,
+    terminal_at timestamptz NOT NULL,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    CHECK (terminal_at >= response_created_at),
+    UNIQUE (tenant_id, owner_ref, agent_ref, response_id)
+);
+
+ALTER TABLE stored_responses
+    ADD CONSTRAINT stored_responses_owned_conversation_fkey
+    FOREIGN KEY (conversation_id, tenant_id, owner_ref)
+    REFERENCES conversations (id, tenant_id, owner_ref) ON DELETE CASCADE;
+ALTER TABLE stored_responses
+    ADD CONSTRAINT stored_responses_turn_link_fkey
+    FOREIGN KEY (turn_id, conversation_id, agent_ref)
+    REFERENCES turns (id, conversation_id, agent_ref) ON DELETE CASCADE;
+ALTER TABLE stored_responses
+    ADD CONSTRAINT stored_responses_continuation_link_fkey
+    FOREIGN KEY (continuation_id, tenant_id, owner_ref, conversation_id, turn_id,
+                 agent_ref, response_id, through_seq)
+    REFERENCES continuations
+        (id, tenant_id, owner_ref, conversation_id, turn_id, agent_ref,
+         response_id, through_seq) ON DELETE CASCADE;
+
+CREATE INDEX stored_responses_conversation_turn_idx
+    ON stored_responses (conversation_id, turn_id);
+CREATE INDEX stored_responses_turn_idx ON stored_responses (turn_id);
+
+CREATE FUNCTION reject_stored_response_update() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    RAISE EXCEPTION 'terminal public responses are immutable'
+        USING ERRCODE = '55000';
+END;
+$$;
+
+CREATE TRIGGER stored_responses_immutable
+BEFORE UPDATE ON stored_responses
+FOR EACH ROW EXECUTE FUNCTION reject_stored_response_update();
+
+CREATE FUNCTION reject_stored_response_turn_rewrite() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    IF EXISTS (SELECT 1 FROM stored_responses WHERE turn_id = OLD.id)
+       AND (NEW.status, NEW.response_id, NEW.error, NEW.usage, NEW.completed_at)
+           IS DISTINCT FROM
+           (OLD.status, OLD.response_id, OLD.error, OLD.usage, OLD.completed_at) THEN
+        RAISE EXCEPTION 'a turn with a stored terminal response is immutable'
+            USING ERRCODE = '55000';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE TRIGGER turns_stored_response_immutable
+BEFORE UPDATE ON turns
+FOR EACH ROW EXECUTE FUNCTION reject_stored_response_turn_rewrite();

src/api.rs

@@ -20,8 +20,9 @@ use crate::{
         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,
+        RegenerateResult, ReplayRequest, ReplayResult, StartTurn, StartTurnResult, StoreResponse,
+        StrictJson, TruncateConversation, Turn, UpdateConversation, UpdateTurn,
+        validate_json_number_tokens,
     },
     object_store::ObjectStore,
     store, uploads,
@@ -73,6 +74,11 @@ pub fn router(state: AppState) -> Router {
             post(create_continuation),
         )
         .route("/v1/continuations/{response_id}", get(get_continuation))
+        .route(
+            "/v1/conversations/{id}/responses",
+            post(store_response).layer(DefaultBodyLimit::max(2 * 1024 * 1024)),
+        )
+        .route("/v1/responses/{response_id}", get(get_response))
         .route("/v1/files", post(upload_file))
         .route("/v1/file-uploads", post(initiate_file_upload))
         .route("/v1/file-uploads/{id}/complete", post(complete_file_upload))
@@ -294,6 +300,7 @@ async fn create_continuation(
     Json(request): Json<CreateContinuation>,
 ) -> ApiResult<(StatusCode, Json<Continuation>)> {
     auth.require(Permission::ContinuationWrite)?;
+    auth.require_agent(request.agent_ref.trim())?;
     Ok((
         StatusCode::CREATED,
         Json(store::create_continuation(&state.pool, &auth, &id, request).await?),
@@ -307,11 +314,70 @@ async fn get_continuation(
     Query(query): Query<ContinuationQuery>,
 ) -> ApiResult<Json<Continuation>> {
     auth.require(Permission::ContinuationRead)?;
+    auth.require_agent(query.agent_ref.trim())
+        .map_err(|_| ApiError::NotFound("Continuation not found.".into()))?;
     Ok(Json(
         store::get_continuation(&state.pool, &auth, &response_id, &query.agent_ref).await?,
     ))
 }
 
+async fn store_response(
+    State(state): State<AppState>,
+    auth: AuthContext,
+    Path(id): Path<String>,
+    body: Bytes,
+) -> ApiResult<Response> {
+    auth.require(Permission::ContinuationWrite)?;
+    let request = parse_store_response(&body)?;
+    auth.require_agent(request.agent_ref.trim())?;
+    let (replayed, response) = store::store_response(&state.pool, &auth, &id, request).await?;
+    public_json_response(
+        if replayed {
+            StatusCode::OK
+        } else {
+            StatusCode::CREATED
+        },
+        response,
+    )
+}
+
+async fn get_response(
+    State(state): State<AppState>,
+    auth: AuthContext,
+    Path(response_id): Path<String>,
+    Query(query): Query<ContinuationQuery>,
+) -> ApiResult<Response> {
+    auth.require(Permission::ContinuationRead)?;
+    auth.require_agent(query.agent_ref.trim())
+        .map_err(|_| ApiError::NotFound("Response not found.".into()))?;
+    public_json_response(
+        StatusCode::OK,
+        store::get_stored_response(&state.pool, &auth, &response_id, query.agent_ref.trim())
+            .await?,
+    )
+}
+
+fn parse_store_response(body: &[u8]) -> ApiResult<StoreResponse> {
+    validate_json_number_tokens(body)
+        .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))?;
+    let mut deserializer = serde_json::Deserializer::from_slice(body);
+    let StrictJson(_value) = StrictJson::deserialize(&mut deserializer)
+        .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))?;
+    deserializer
+        .end()
+        .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))?;
+    serde_json::from_slice(body)
+        .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))
+}
+
+fn public_json_response(status: StatusCode, body: String) -> ApiResult<Response> {
+    Response::builder()
+        .status(status)
+        .header(header::CONTENT_TYPE, "application/json")
+        .body(Body::from(body))
+        .map_err(|_| ApiError::CorruptStoredResponse)
+}
+
 async fn truncate_conversation(
     State(state): State<AppState>,
     auth: AuthContext,

src/error.rs

@@ -27,6 +27,8 @@ pub enum ApiError {
     PayloadTooLarge(String),
     #[error("object storage operation failed")]
     ObjectStore(#[source] anyhow::Error),
+    #[error("stored response failed integrity validation")]
+    CorruptStoredResponse,
     #[error("database operation failed")]
     Database(#[from] sqlx::Error),
 }
@@ -47,13 +49,19 @@ impl IntoResponse for ApiError {
                 tracing::error!(?error, "object storage request failed");
                 (StatusCode::BAD_GATEWAY, "object_store_error")
             }
+            Self::CorruptStoredResponse => {
+                tracing::error!("stored response failed integrity validation");
+                (StatusCode::INTERNAL_SERVER_ERROR, "internal_error")
+            }
             Self::Database(error) => {
                 tracing::error!(?error, "database request failed");
                 (StatusCode::INTERNAL_SERVER_ERROR, "internal_error")
             }
         };
         let message = match self {
-            Self::Database(_) | Self::ObjectStore(_) => "Storage operation failed.".to_owned(),
+            Self::Database(_) | Self::ObjectStore(_) | Self::CorruptStoredResponse => {
+                "Storage operation failed.".to_owned()
+            }
             other => other.to_string(),
         };
         let mut response = (

src/model.rs

@@ -1,6 +1,6 @@
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Deserializer, Serialize, de};
-use serde_json::{Map, Number, Value};
+use serde_json::{Map, Number, Value, value::RawValue};
 use sqlx::FromRow;
 use std::str::FromStr;
 
@@ -377,7 +377,9 @@ pub struct RegenerateResult {
 pub struct Continuation {
     pub id: String,
     pub tenant_id: String,
+    pub owner_ref: String,
     pub conversation_id: String,
+    pub turn_id: Option<String>,
     pub agent_ref: String,
     pub response_id: String,
     pub parent_response_id: Option<String>,
@@ -400,6 +402,26 @@ pub struct ContinuationQuery {
     pub agent_ref: String,
 }
 
+pub const STORED_RESPONSE_MAX_BYTES: usize = 1024 * 1024;
+pub const STORED_RESPONSE_SCHEMA: &str = "open-responses/public-response/v1";
+
+#[derive(Debug, Deserialize)]
+pub struct StoreResponse {
+    pub agent_ref: String,
+    pub turn_id: String,
+    pub through_seq: Option<i64>,
+    pub state: Option<Value>,
+    #[serde(default = "stored_response_schema")]
+    pub schema_marker: String,
+    pub response_created_at: DateTime<Utc>,
+    pub terminal_at: DateTime<Utc>,
+    pub public_response: Box<RawValue>,
+}
+
+fn stored_response_schema() -> String {
+    STORED_RESPONSE_SCHEMA.to_owned()
+}
+
 #[derive(Debug, Serialize, FromRow)]
 pub struct FileRecord {
     pub id: String,