diff --git a/Cargo.toml b/Cargo.toml index 1ec05a5..ed13063 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/README.md b/README.md index 5c4ef44..bd5fcd4 100644 --- a/README.md +++ b/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 diff --git a/migrations/0008_stored_responses.sql b/migrations/0008_stored_responses.sql new file mode 100644 index 0000000..48a0e89 --- /dev/null +++ b/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(); diff --git a/src/api.rs b/src/api.rs index 595fe50..e5d61a1 100644 --- a/src/api.rs +++ b/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, ) -> ApiResult<(StatusCode, Json)> { 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, ) -> ApiResult> { 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, + auth: AuthContext, + Path(id): Path, + body: Bytes, +) -> ApiResult { + 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, + auth: AuthContext, + Path(response_id): Path, + Query(query): Query, +) -> ApiResult { + 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 { + 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::builder() + .status(status) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .map_err(|_| ApiError::CorruptStoredResponse) +} + async fn truncate_conversation( State(state): State, auth: AuthContext, diff --git a/src/error.rs b/src/error.rs index 37aa137..5a5c5e2 100644 --- a/src/error.rs +++ b/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 = ( diff --git a/src/model.rs b/src/model.rs index b689b3a..12a0032 100644 --- a/src/model.rs +++ b/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, pub agent_ref: String, pub response_id: String, pub parent_response_id: Option, @@ -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, + pub state: Option, + #[serde(default = "stored_response_schema")] + pub schema_marker: String, + pub response_created_at: DateTime, + pub terminal_at: DateTime, + pub public_response: Box, +} + +fn stored_response_schema() -> String { + STORED_RESPONSE_SCHEMA.to_owned() +} + #[derive(Debug, Serialize, FromRow)] pub struct FileRecord { pub id: String, diff --git a/src/store.rs b/src/store.rs index 2173340..d6088be 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,8 +1,9 @@ use base64::{Engine, engine::general_purpose::STANDARD}; -use serde::Serialize; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; use serde_json::Value; use sha2::{Digest, Sha256}; -use sqlx::{PgPool, Postgres, Transaction}; +use sqlx::{FromRow, PgPool, Postgres, Transaction}; use crate::{ api::AppState, @@ -12,8 +13,10 @@ use crate::{ ids::new_id, model::{ Actor, AppendItems, AppendResult, Continuation, Conversation, CreateContinuation, - CreateConversation, CreateTurn, FileDelivery, Item, ReplayRequest, ReplayResult, StartTurn, - StartTurnResult, Turn, UpdateConversation, UpdateTurn, + CreateConversation, CreateTurn, FileDelivery, Item, ReplayRequest, ReplayResult, + STORED_RESPONSE_MAX_BYTES, STORED_RESPONSE_SCHEMA, StartTurn, StartTurnResult, + StoreResponse, StrictJson, Turn, UpdateConversation, UpdateTurn, + validate_json_number_tokens, }, }; @@ -624,6 +627,20 @@ fn turn_start_lock_key(tenant: &str, owner: &str, client: &str, key: &str) -> i6 ) } +fn stored_response_lock_key(tenant: &str, owner: &str, agent: &str, response_id: &str) -> i64 { + let mut digest = Sha256::new(); + digest.update(b"threadmark:stored-response-lock:v1\0"); + for value in [tenant, owner, agent, response_id] { + digest.update((value.len() as u32).to_be_bytes()); + digest.update(value.as_bytes()); + } + i64::from_be_bytes( + digest.finalize()[..8] + .try_into() + .expect("eight digest bytes"), + ) +} + #[cfg(test)] mod turn_start_tests { use super::turn_start_lock_key; @@ -1205,12 +1222,14 @@ pub async fn create_continuation( } let result = sqlx::query_as::<_, Continuation>( "INSERT INTO continuations - (id, tenant_id, conversation_id, agent_ref, response_id, parent_response_id, through_seq, state) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - ON CONFLICT (tenant_id, agent_ref, response_id) DO NOTHING RETURNING *", + (id, tenant_id, owner_ref, conversation_id, agent_ref, response_id, + parent_response_id, through_seq, state) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (tenant_id, owner_ref, agent_ref, response_id) DO NOTHING RETURNING *", ) .bind(new_id("cont")) .bind(&actor.tenant_id) + .bind(&actor.principal_id) .bind(conversation_id) .bind(request.agent_ref) .bind(request.response_id) @@ -1229,10 +1248,9 @@ pub async fn get_continuation( agent_ref: &str, ) -> ApiResult { sqlx::query_as::<_, Continuation>( - "SELECT c.* FROM continuations c - JOIN conversations v ON v.id = c.conversation_id - WHERE c.response_id = $1 AND c.agent_ref = $2 - AND c.tenant_id = $3 AND v.owner_ref = $4", + "SELECT * FROM continuations + WHERE response_id = $1 AND agent_ref = $2 + AND tenant_id = $3 AND owner_ref = $4", ) .bind(response_id) .bind(agent_ref) @@ -1243,6 +1261,379 @@ pub async fn get_continuation( .ok_or_else(|| ApiError::NotFound("Continuation not found.".into())) } +#[derive(Debug, FromRow)] +struct StoredResponseRow { + tenant_id: String, + owner_ref: String, + agent_ref: String, + conversation_id: String, + turn_id: String, + response_id: String, + previous_response_id: Option, + continuation_parent_response_id: Option, + terminal_status: String, + public_response: Value, + public_response_text: String, + canonical_digest: Vec, + schema_marker: String, + canonical_size: i64, + through_seq: i64, + response_created_at: DateTime, + terminal_at: DateTime, + state: Option, +} + +struct ValidatedResponse { + value: Value, + public_response_text: String, + response_id: String, + previous_response_id: Option, + status: String, + digest: Vec, + size: i64, +} + +fn validate_public_response( + raw: &serde_json::value::RawValue, + schema_marker: &str, +) -> ApiResult { + if schema_marker != STORED_RESPONSE_SCHEMA { + return Err(ApiError::BadRequest(format!( + "schema_marker must be {STORED_RESPONSE_SCHEMA}" + ))); + } + if raw.get().len() > STORED_RESPONSE_MAX_BYTES { + return Err(ApiError::PayloadTooLarge(format!( + "public_response exceeds {STORED_RESPONSE_MAX_BYTES} bytes" + ))); + } + validate_json_number_tokens(raw.get().as_bytes()) + .map_err(|error| ApiError::BadRequest(format!("invalid public_response JSON: {error}")))?; + let mut deserializer = serde_json::Deserializer::from_str(raw.get()); + let StrictJson(value) = StrictJson::deserialize(&mut deserializer) + .map_err(|error| ApiError::BadRequest(format!("invalid public_response JSON: {error}")))?; + deserializer + .end() + .map_err(|error| ApiError::BadRequest(format!("invalid public_response JSON: {error}")))?; + let object = value + .as_object() + .ok_or_else(|| ApiError::BadRequest("public_response must be a JSON object".into()))?; + if object.get("object").and_then(Value::as_str) != Some("response") { + return Err(ApiError::BadRequest( + "public_response.object must be response".into(), + )); + } + let response_id = bounded_response_id(object.get("id"), "public_response.id")?; + let status = object + .get("status") + .and_then(Value::as_str) + .filter(|status| matches!(*status, "completed" | "incomplete" | "failed" | "cancelled")) + .ok_or_else(|| ApiError::BadRequest("public_response.status must be terminal".into()))? + .to_owned(); + let previous_response_id = match object.get("previous_response_id") { + None | Some(Value::Null) => None, + Some(value) => Some(bounded_response_id( + Some(value), + "public_response.previous_response_id", + )?), + }; + let canonical = serde_json_canonicalizer::to_vec(&value) + .map_err(|error| ApiError::BadRequest(format!("invalid public_response JSON: {error}")))?; + if canonical.len() > STORED_RESPONSE_MAX_BYTES { + return Err(ApiError::PayloadTooLarge(format!( + "public_response exceeds {STORED_RESPONSE_MAX_BYTES} bytes" + ))); + } + Ok(ValidatedResponse { + value, + public_response_text: raw.get().to_owned(), + response_id, + previous_response_id, + status, + digest: Sha256::digest(&canonical).to_vec(), + size: canonical.len() as i64, + }) +} + +fn bounded_response_id(value: Option<&Value>, field: &str) -> ApiResult { + value + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty() && value.chars().count() <= 200) + .map(str::to_owned) + .ok_or_else(|| ApiError::BadRequest(format!("{field} must contain 1 to 200 characters"))) +} + +pub async fn store_response( + pool: &PgPool, + actor: &Actor, + conversation_id: &str, + mut request: StoreResponse, +) -> ApiResult<(bool, String)> { + request.agent_ref = request.agent_ref.trim().to_owned(); + request.turn_id = request.turn_id.trim().to_owned(); + if request.agent_ref.is_empty() || request.agent_ref.chars().count() > 200 { + return Err(ApiError::BadRequest( + "agent_ref must contain 1 to 200 characters".into(), + )); + } + if request.turn_id.is_empty() || request.turn_id.chars().count() > 200 { + return Err(ApiError::BadRequest( + "turn_id must contain 1 to 200 characters".into(), + )); + } + if request.terminal_at < request.response_created_at { + return Err(ApiError::BadRequest( + "terminal_at must not precede response_created_at".into(), + )); + } + let validated = validate_public_response(&request.public_response, &request.schema_marker)?; + let mut tx = pool.begin().await?; + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(stored_response_lock_key( + &actor.tenant_id, + &actor.principal_id, + &request.agent_ref, + &validated.response_id, + )) + .execute(&mut *tx) + .await?; + let conversation = lock_conversation(&mut tx, actor, conversation_id).await?; + let through_seq = request.through_seq.unwrap_or(conversation.next_seq - 1); + if through_seq < 0 || through_seq >= conversation.next_seq { + return Err(ApiError::BadRequest( + "through_seq is outside the conversation transcript".into(), + )); + } + + let turn = sqlx::query_as::<_, (String, String, Option, Option>)>( + "SELECT agent_ref, status, response_id, completed_at FROM turns + WHERE id = $1 AND conversation_id = $2 FOR UPDATE", + ) + .bind(&request.turn_id) + .bind(conversation_id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| ApiError::NotFound("Turn not found.".into()))?; + if turn.0 != request.agent_ref + || turn.1 != validated.status + || turn.2.as_deref() != Some(validated.response_id.as_str()) + || turn.3.is_none() + { + return Err(ApiError::BadRequest( + "public_response does not match the terminal turn".into(), + )); + } + + if let Some(parent) = &validated.previous_response_id { + let parent_exists = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM continuations + WHERE tenant_id = $1 AND owner_ref = $2 AND conversation_id = $3 + AND agent_ref = $4 AND response_id = $5)", + ) + .bind(&actor.tenant_id) + .bind(&actor.principal_id) + .bind(conversation_id) + .bind(&request.agent_ref) + .bind(parent) + .fetch_one(&mut *tx) + .await?; + if !parent_exists { + return Err(ApiError::NotFound("Previous response not found.".into())); + } + } + + let existing = sqlx::query_as::<_, StoredResponseRow>( + "SELECT response.tenant_id, response.owner_ref, response.agent_ref, + response.conversation_id, response.turn_id, response.response_id, + response.previous_response_id, + continuation.parent_response_id AS continuation_parent_response_id, + response.terminal_status, + response.public_response, response.public_response_text, + response.canonical_digest, + response.schema_marker, response.canonical_size, response.through_seq, + response.response_created_at, response.terminal_at, continuation.state + FROM stored_responses response + JOIN continuations continuation ON continuation.id = response.continuation_id + WHERE response.tenant_id = $1 AND response.owner_ref = $2 + AND response.agent_ref = $3 AND response.response_id = $4", + ) + .bind(&actor.tenant_id) + .bind(&actor.principal_id) + .bind(&request.agent_ref) + .bind(&validated.response_id) + .fetch_optional(&mut *tx) + .await?; + if let Some(existing) = existing { + validate_stored_response_row(&existing, actor, &request.agent_ref)?; + let retry_through_seq = request.through_seq.unwrap_or(existing.through_seq); + let identical = existing.conversation_id == conversation_id + && existing.turn_id == request.turn_id + && existing.previous_response_id == validated.previous_response_id + && existing.terminal_status == validated.status + && existing.canonical_digest == validated.digest + && existing.schema_marker == request.schema_marker + && existing.canonical_size == validated.size + && existing.through_seq == retry_through_seq + && existing.response_created_at == request.response_created_at + && existing.terminal_at == request.terminal_at + && existing.state == request.state; + if !identical { + return Err(ApiError::Conflict( + "Response ID already stores a different terminal response.".into(), + )); + } + tx.commit().await?; + return Ok((true, existing.public_response_text)); + } + + let inserted_continuation_id = sqlx::query_scalar::<_, String>( + "INSERT INTO continuations + (id, tenant_id, owner_ref, conversation_id, turn_id, agent_ref, + response_id, parent_response_id, through_seq, state) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + ON CONFLICT (tenant_id, owner_ref, agent_ref, response_id) DO NOTHING + RETURNING id", + ) + .bind(new_id("cont")) + .bind(&actor.tenant_id) + .bind(&actor.principal_id) + .bind(conversation_id) + .bind(&request.turn_id) + .bind(&request.agent_ref) + .bind(&validated.response_id) + .bind(&validated.previous_response_id) + .bind(through_seq) + .bind(&request.state) + .fetch_optional(&mut *tx) + .await?; + let continuation_id = if let Some(id) = inserted_continuation_id { + id + } else { + let existing = sqlx::query_as::< + _, + ( + String, + String, + Option, + Option, + i64, + Option, + ), + >( + "SELECT id, conversation_id, turn_id, parent_response_id, through_seq, state + FROM continuations + WHERE tenant_id = $1 AND owner_ref = $2 AND agent_ref = $3 + AND response_id = $4", + ) + .bind(&actor.tenant_id) + .bind(&actor.principal_id) + .bind(&request.agent_ref) + .bind(&validated.response_id) + .fetch_one(&mut *tx) + .await?; + if existing.1 != conversation_id + || existing.2.as_deref() != Some(request.turn_id.as_str()) + || existing.3 != validated.previous_response_id + || existing.4 != through_seq + || existing.5 != request.state + { + return Err(ApiError::Conflict( + "Response ID already stores a different continuation.".into(), + )); + } + existing.0 + }; + sqlx::query( + "INSERT INTO stored_responses + (id, tenant_id, owner_ref, agent_ref, conversation_id, turn_id, + continuation_id, response_id, previous_response_id, terminal_status, + public_response, public_response_text, canonical_digest, schema_marker, canonical_size, + through_seq, response_created_at, terminal_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)", + ) + .bind(new_id("sresp")) + .bind(&actor.tenant_id) + .bind(&actor.principal_id) + .bind(&request.agent_ref) + .bind(conversation_id) + .bind(&request.turn_id) + .bind(continuation_id) + .bind(&validated.response_id) + .bind(&validated.previous_response_id) + .bind(&validated.status) + .bind(&validated.value) + .bind(&validated.public_response_text) + .bind(&validated.digest) + .bind(&request.schema_marker) + .bind(validated.size) + .bind(through_seq) + .bind(request.response_created_at) + .bind(request.terminal_at) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok((false, validated.public_response_text)) +} + +pub async fn get_stored_response( + pool: &PgPool, + actor: &Actor, + response_id: &str, + agent_ref: &str, +) -> ApiResult { + let row = sqlx::query_as::<_, StoredResponseRow>( + "SELECT response.tenant_id, response.owner_ref, response.agent_ref, + response.conversation_id, response.turn_id, response.response_id, + response.previous_response_id, + continuation.parent_response_id AS continuation_parent_response_id, + response.terminal_status, + response.public_response, response.public_response_text, + response.canonical_digest, + response.schema_marker, response.canonical_size, response.through_seq, + response.response_created_at, response.terminal_at, continuation.state + FROM stored_responses response + JOIN continuations continuation ON continuation.id = response.continuation_id + WHERE response.response_id = $1 AND response.agent_ref = $2 + AND response.tenant_id = $3 AND response.owner_ref = $4", + ) + .bind(response_id) + .bind(agent_ref) + .bind(&actor.tenant_id) + .bind(&actor.principal_id) + .fetch_optional(pool) + .await? + .ok_or_else(|| ApiError::NotFound("Response not found.".into()))?; + + validate_stored_response_row(&row, actor, agent_ref)?; + Ok(row.public_response_text) +} + +fn validate_stored_response_row( + row: &StoredResponseRow, + actor: &Actor, + agent_ref: &str, +) -> ApiResult<()> { + let raw = serde_json::value::RawValue::from_string(row.public_response_text.clone()) + .map_err(|_| ApiError::CorruptStoredResponse)?; + let validated = validate_public_response(&raw, &row.schema_marker) + .map_err(|_| ApiError::CorruptStoredResponse)?; + if row.tenant_id != actor.tenant_id + || row.owner_ref != actor.principal_id + || row.agent_ref != agent_ref + || row.response_id != validated.response_id + || row.previous_response_id != validated.previous_response_id + || row.continuation_parent_response_id != validated.previous_response_id + || row.terminal_status != validated.status + || row.canonical_digest != validated.digest + || row.canonical_size != validated.size + || row.public_response != validated.value + || row.response_created_at > row.terminal_at + { + return Err(ApiError::CorruptStoredResponse); + } + Ok(()) +} + #[cfg(test)] mod tests { use serde_json::json; @@ -1354,4 +1745,78 @@ mod tests { vec!["file_document".to_owned(), "file_image".to_owned()] ); } + + #[test] + fn validates_terminal_public_response_and_digest() { + let response = serde_json::value::RawValue::from_string( + r#"{"usage":{"output_tokens":4,"input_tokens":3},"output":[{"type":"message","id":"msg_1"}],"previous_response_id":null,"status":"completed","object":"response","id":"resp_123"}"#.into(), + ) + .unwrap(); + let validated = validate_public_response(&response, STORED_RESPONSE_SCHEMA).unwrap(); + assert_eq!(validated.response_id, "resp_123"); + assert_eq!(validated.status, "completed"); + assert_eq!(validated.digest.len(), 32); + assert_eq!(validated.public_response_text, response.get()); + } + + #[test] + fn rejects_nonterminal_and_oversized_public_responses() { + let nonterminal = serde_json::value::to_raw_value(&json!({ + "id": "resp_123", + "object": "response", + "status": "in_progress" + })) + .unwrap(); + assert!(validate_public_response(&nonterminal, STORED_RESPONSE_SCHEMA).is_err()); + + let oversized = serde_json::value::to_raw_value(&json!({ + "id": "resp_123", + "object": "response", + "status": "completed", + "output": "x".repeat(STORED_RESPONSE_MAX_BYTES) + })) + .unwrap(); + assert!(matches!( + validate_public_response(&oversized, STORED_RESPONSE_SCHEMA), + Err(ApiError::PayloadTooLarge(_)) + )); + } + + #[test] + fn malformed_stored_response_is_an_internal_integrity_error() { + let now = Utc::now(); + let row = StoredResponseRow { + tenant_id: "tenant-a".into(), + owner_ref: "owner-a".into(), + agent_ref: "agent-a".into(), + conversation_id: "conv-a".into(), + turn_id: "turn-a".into(), + response_id: "resp-a".into(), + previous_response_id: None, + continuation_parent_response_id: None, + terminal_status: "completed".into(), + public_response: json!({ + "id": "resp-a", + "object": "response", + "status": "completed" + }), + public_response_text: r#"{"id":"resp-a","object":"response","status":"completed"}"# + .into(), + canonical_digest: vec![0; 32], + schema_marker: STORED_RESPONSE_SCHEMA.into(), + canonical_size: 1, + through_seq: 0, + response_created_at: now, + terminal_at: now, + state: None, + }; + let actor = Actor { + tenant_id: "tenant-a".into(), + principal_id: "owner-a".into(), + }; + assert!(matches!( + validate_stored_response_row(&row, &actor, "agent-a"), + Err(ApiError::CorruptStoredResponse) + )); + } }