From 4f06fa032c2004e7322e50748c75877b3ccec6ac Mon Sep 17 00:00:00 2001 From: "dispatch-developer[bot]" <306909890+dispatch-developer[bot]@users.noreply.github.com> Date: Mon, 17 Aug 2026 11:33:23 +0000 Subject: [PATCH 1/3] ledger: constrain delegated agent writes --- .env.example | 1 + README.md | 38 +- migrations/0009_delegated_append_batches.sql | 24 + src/api.rs | 30 +- src/auth.rs | 126 ++++- src/capability.rs | 1 + src/config.rs | 8 + src/model.rs | 2 + src/store.rs | 541 ++++++++++++++++++- 9 files changed, 761 insertions(+), 10 deletions(-) create mode 100644 migrations/0009_delegated_append_batches.sql diff --git a/.env.example b/.env.example index d5310fa..9049e78 100644 --- a/.env.example +++ b/.env.example @@ -9,6 +9,7 @@ AUTH_MODE=trusted_headers # AUTH_AUDIENCE=threadmark-api # AUTH_JWKS_URL=https://auth.example.com/.well-known/jwks.json AUTH_MAX_OWNER_TOKEN_SECONDS=300 +AUTH_MAX_DELEGATED_TOKEN_SECONDS=600 FILE_MAX_MB=32 S3_ENDPOINT=http://localhost:9010 S3_PUBLIC_URL=http://localhost:9010 diff --git a/README.md b/README.md index 5c4ef44..e10f4e3 100644 --- a/README.md +++ b/README.md @@ -140,8 +140,14 @@ mode. Production uses `AUTH_MODE=jwt` with `AUTH_ISSUER`, `AUTH_AUDIENCE`, and an HTTPS `AUTH_JWKS_URL`. JWT mode accepts Ed25519 `at+jwt` owner-session tokens and derives tenant, principal, and endpoint permissions exclusively from verified -claims. Delegated-agent tokens remain rejected until their resource-bound write -invariants are implemented. +claims. Owner tokens may live for at most `AUTH_MAX_OWNER_TOKEN_SECONDS`. + +Delegated writes use an Ed25519 `at+jwt` with `token_kind=delegated_agent`, the +`transcript:append_agent` permission, and required `conversation_id`, `turn_id`, +and `agent_ref` claims. The signed tenant and principal remain the owner bounds. +Delegated tokens may live for at most `AUTH_MAX_DELEGATED_TOKEN_SECONDS` (600 by +default). Other recognized delegated permissions are reserved until their +resource-bound routes are enabled. An agent called by Parley can receive a short-lived token scoped to the same tenant, principal, conversation, turn, and agent deployment. That authorization @@ -184,6 +190,29 @@ curl -sS http://localhost:8090/v1/conversations/conv_.../items \ }' ``` +A delegated agent writes to the same endpoint with its bearer token. Its body +must use `source: "agent"` and the token's exact turn. New writes are accepted +only while that turn is `pending` or `streaming`. The initial output allowlist +is deliberately narrow: + +- assistant `message` items whose content consists only of string + `output_text` and/or `refusal` parts; +- `reasoning` items with `summary_text` summary parts, optional + `reasoning_text` content parts, and optional string `encrypted_content`; +- `function_call` items with string `call_id`, `name`, and `arguments`. + +User/system message roles, input parts (including `function_call_output`), +unknown item types, roles on non-message items, malformed fields, duplicate JSON +keys, non-canonical numbers, and new `threadmark://files/...` references are +rejected. Supporting another protocol output type requires adding it to this +versioned allowlist. + +Delegated idempotency binds the ordered payloads and count to source, turn, +conversation, owner, tenant, and agent. An exact retry returns the original item +IDs plus explicit `first_seq` and `last_seq`; any changed retry returns +`409` with `idempotency_key_reused`. A retry remains valid after the turn closes, +but a new append to a terminal turn returns `409` with `turn_not_active`. + Build protocol-ready replay input: ```bash @@ -249,7 +278,10 @@ GET /v1/continuations/resp_abc?agent_ref=research-agent%2Fprod ## Design notes - `payload` is JSONB and remains protocol-owned. Threadmark only requires each - item to be a JSON object. + item to be a JSON object for owner-authorized generic appends. This behavior + is unchanged; only delegated appends use the strict output contract above. + Append responses now add `first_seq` and `last_seq`; existing `items` and + `replayed` fields retain their behavior. - 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 diff --git a/migrations/0009_delegated_append_batches.sql b/migrations/0009_delegated_append_batches.sql new file mode 100644 index 0000000..dcba6fe --- /dev/null +++ b/migrations/0009_delegated_append_batches.sql @@ -0,0 +1,24 @@ +ALTER TABLE append_batches + ADD COLUMN request_version smallint, + ADD COLUMN request_digest bytea, + ADD COLUMN source text, + ADD COLUMN turn_id text, + ADD COLUMN tenant_id text, + ADD COLUMN owner_ref text, + ADD COLUMN agent_ref text, + ADD COLUMN item_count integer, + ADD COLUMN item_ids text[]; + +ALTER TABLE append_batches ADD CONSTRAINT append_batches_delegated_request_check CHECK ( + (request_version IS NULL AND request_digest IS NULL AND source IS NULL AND turn_id IS NULL + AND tenant_id IS NULL AND owner_ref IS NULL AND agent_ref IS NULL AND item_count IS NULL + AND item_ids IS NULL) + OR + (request_version IS NOT NULL AND request_digest IS NOT NULL AND source IS NOT NULL + AND turn_id IS NOT NULL AND tenant_id IS NOT NULL AND owner_ref IS NOT NULL + AND agent_ref IS NOT NULL AND item_count IS NOT NULL AND item_ids IS NOT NULL + AND request_version = 1 AND octet_length(request_digest) = 32 AND source = 'agent' + AND item_count BETWEEN 1 AND 100 AND cardinality(item_ids) = item_count + AND first_seq > 0 AND last_seq >= first_seq + AND item_count::bigint = last_seq - first_seq + 1) +); diff --git a/src/api.rs b/src/api.rs index 595fe50..deef725 100644 --- a/src/api.rs +++ b/src/api.rs @@ -212,14 +212,38 @@ async fn append_items( State(state): State, auth: AuthContext, Path(id): Path, - Json(request): Json, + body: Bytes, ) -> ApiResult> { - auth.require(Permission::TranscriptAppend)?; + let request = if auth.is_delegated() { + auth.require(Permission::TranscriptAppendAgent)?; + parse_strict_json(&body)? + } else { + auth.require(Permission::TranscriptAppend)?; + serde_json::from_slice(&body) + .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))? + }; Ok(Json( - store::append_items(&state.pool, &auth, &id, request).await?, + if auth.is_delegated() { + store::append_delegated_items(&state.pool, &auth, &id, request).await? + } else { + store::append_items(&state.pool, &auth, &id, request).await? + }, )) } +fn parse_strict_json(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_value(value) + .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}"))) +} + async fn replay( State(state): State, auth: AuthContext, diff --git a/src/auth.rs b/src/auth.rs index 48a24b8..05ead42 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -32,6 +32,7 @@ struct JwtVerifier { issuer: String, audience: String, max_owner_seconds: u64, + max_delegated_seconds: u64, keys: HashMap, } @@ -40,9 +41,18 @@ pub struct AuthContext { pub actor: Actor, pub client_id: String, agent_ref: Option, + conversation_id: Option, + turn_id: Option, + token_kind: TokenKind, permissions: HashSet, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TokenKind { + OwnerSession, + DelegatedAgent, +} + #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)] pub enum Permission { ConversationList, @@ -54,6 +64,7 @@ pub enum Permission { ConversationRegenerate, TranscriptRead, TranscriptAppend, + TranscriptAppendAgent, TurnCreate, TurnRead, TurnUpdate, @@ -102,6 +113,8 @@ struct Claims { principal: String, permissions: Vec, agent_ref: Option, + conversation_id: Option, + turn_id: Option, } #[derive(Debug, Deserialize)] @@ -146,6 +159,7 @@ impl Authenticator { issuer, audience, config.auth_max_owner_token_seconds, + config.auth_max_delegated_token_seconds, jwks, )?) } @@ -167,6 +181,7 @@ impl JwtVerifier { issuer: String, audience: String, max_owner_seconds: u64, + max_delegated_seconds: u64, jwks: JwkSet, ) -> anyhow::Result { ensure!(!jwks.keys.is_empty(), "JWKS contains no keys"); @@ -191,6 +206,7 @@ impl JwtVerifier { issuer, audience, max_owner_seconds, + max_delegated_seconds, keys, }) } @@ -225,11 +241,9 @@ impl Claims { }; if self.iss != verifier.issuer || !valid_audience - || self.token_kind != "owner_session" || self.exp <= self.iat || self.exp <= self.nbf || self.iat > now.saturating_add(30) - || self.exp.saturating_sub(self.iat) > verifier.max_owner_seconds || !valid_id(&self.sub) || !valid_id(&self.client_id) || self.client_id == "threadmark:trusted-headers" @@ -248,6 +262,22 @@ impl Claims { .iter() .map(|value| Permission::parse(value)) .collect::>>()?; + let token_kind = match self.token_kind.as_str() { + "owner_session" + if self.exp.saturating_sub(self.iat) <= verifier.max_owner_seconds + && self.conversation_id.is_none() + && self.turn_id.is_none() => TokenKind::OwnerSession, + "delegated_agent" + if self.exp.saturating_sub(self.iat) <= verifier.max_delegated_seconds + && self.conversation_id.as_deref().is_some_and(valid_id) + && self.turn_id.as_deref().is_some_and(valid_id) + && self.agent_ref.as_deref().is_some_and(valid_id) + && permissions.iter().all(Permission::allowed_for_delegated) => + { + TokenKind::DelegatedAgent + } + _ => return None, + }; Some(AuthContext { actor: Actor { tenant_id: self.tenant, @@ -255,6 +285,9 @@ impl Claims { }, client_id: self.client_id, agent_ref: self.agent_ref, + conversation_id: self.conversation_id, + turn_id: self.turn_id, + token_kind, permissions, }) } @@ -262,6 +295,13 @@ impl Claims { impl AuthContext { pub fn require(&self, permission: Permission) -> Result<(), ApiError> { + // Delegated operations are exposed only after their resource-bound store + // contract is implemented. The constrained append path is the first. + if self.token_kind == TokenKind::DelegatedAgent + && permission != Permission::TranscriptAppendAgent + { + return Err(ApiError::Forbidden); + } self.permissions .contains(&permission) .then_some(()) @@ -280,6 +320,42 @@ impl AuthContext { _ => Ok(()), } } + + pub fn is_delegated(&self) -> bool { + self.token_kind == TokenKind::DelegatedAgent + } + + pub fn delegated_bounds(&self) -> Option<(&str, &str, &str)> { + self.is_delegated().then(|| { + ( + self.conversation_id.as_deref().expect("validated claim"), + self.turn_id.as_deref().expect("validated claim"), + self.agent_ref.as_deref().expect("validated claim"), + ) + }) + } + + #[cfg(test)] + pub(crate) fn delegated_for_test( + tenant_id: &str, + principal_id: &str, + conversation_id: &str, + turn_id: &str, + agent_ref: &str, + ) -> Self { + Self { + actor: Actor { + tenant_id: tenant_id.into(), + principal_id: principal_id.into(), + }, + client_id: "test".into(), + agent_ref: Some(agent_ref.into()), + conversation_id: Some(conversation_id.into()), + turn_id: Some(turn_id.into()), + token_kind: TokenKind::DelegatedAgent, + permissions: [Permission::TranscriptAppendAgent].into_iter().collect(), + } + } } impl std::ops::Deref for AuthContext { @@ -313,6 +389,7 @@ impl Permission { "conversation:regenerate" => Self::ConversationRegenerate, "transcript:read" => Self::TranscriptRead, "transcript:append" => Self::TranscriptAppend, + "transcript:append_agent" => Self::TranscriptAppendAgent, "turn:create" => Self::TurnCreate, "turn:read" => Self::TurnRead, "turn:update" => Self::TurnUpdate, @@ -325,6 +402,19 @@ impl Permission { _ => return None, }) } + + fn allowed_for_delegated(&self) -> bool { + matches!( + self, + Self::TranscriptRead + | Self::TranscriptAppendAgent + | Self::TurnRead + | Self::TurnUpdate + | Self::ContinuationRead + | Self::ContinuationWrite + | Self::FileRead + ) + } } fn bearer(headers: &HeaderMap) -> Option<&str> { @@ -353,6 +443,9 @@ fn trusted_headers(headers: &HeaderMap) -> Result { }, client_id: "threadmark:trusted-headers".into(), agent_ref: None, + conversation_id: None, + turn_id: None, + token_kind: TokenKind::OwnerSession, permissions: OWNER_PERMISSIONS.into_iter().collect(), }) } @@ -385,6 +478,7 @@ mod tests { "https://issuer.example".into(), "threadmark-api".into(), 300, + 600, jwks, ) .unwrap() @@ -485,9 +579,35 @@ mod tests { } #[test] - fn rejects_delegated_tokens_until_resource_policy_is_implemented() { + fn accepts_only_fully_bound_delegated_tokens() { let mut claims = claims(); claims["token_kind"] = json!("delegated_agent"); + claims["conversation_id"] = json!("conv_1"); + claims["turn_id"] = json!("turn_1"); + claims["agent_ref"] = json!("agent/prod"); + claims["permissions"] = json!(["transcript:append_agent"]); + let context = verifier().authenticate(&headers(&claims)).unwrap(); + assert_eq!( + context.delegated_bounds(), + Some(("conv_1", "turn_1", "agent/prod")) + ); + assert!(context.require(Permission::TranscriptAppendAgent).is_ok()); + + claims.as_object_mut().unwrap().remove("turn_id"); + assert!(matches!( + verifier().authenticate(&headers(&claims)), + Err(ApiError::Unauthorized) + )); + } + + #[test] + fn rejects_owner_permissions_on_delegated_tokens() { + let mut claims = claims(); + claims["token_kind"] = json!("delegated_agent"); + claims["conversation_id"] = json!("conv_1"); + claims["turn_id"] = json!("turn_1"); + claims["agent_ref"] = json!("agent/prod"); + claims["permissions"] = json!(["transcript:append"]); assert!(matches!( verifier().authenticate(&headers(&claims)), Err(ApiError::Unauthorized) diff --git a/src/capability.rs b/src/capability.rs index 2a2a209..18f48c9 100644 --- a/src/capability.rs +++ b/src/capability.rs @@ -120,6 +120,7 @@ mod tests { auth_audience: None, auth_jwks_url: None, auth_max_owner_token_seconds: 300, + auth_max_delegated_token_seconds: 600, })) } diff --git a/src/config.rs b/src/config.rs index 535bd69..04c5adf 100644 --- a/src/config.rs +++ b/src/config.rs @@ -27,6 +27,7 @@ pub struct Inner { pub auth_audience: Option, pub auth_jwks_url: Option, pub auth_max_owner_token_seconds: u64, + pub auth_max_delegated_token_seconds: u64, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -100,6 +101,12 @@ impl Config { auth_max_owner_token_seconds > 0, "AUTH_MAX_OWNER_TOKEN_SECONDS must be greater than zero" ); + let auth_max_delegated_token_seconds = + parse("AUTH_MAX_DELEGATED_TOKEN_SECONDS", "600")?; + ensure!( + auth_max_delegated_token_seconds > 0, + "AUTH_MAX_DELEGATED_TOKEN_SECONDS must be greater than zero" + ); let direct_upload_enabled = std::env::var("DIRECT_UPLOAD_ENABLED") .unwrap_or_else(|_| "false".into()) .parse() @@ -154,6 +161,7 @@ impl Config { auth_audience, auth_jwks_url, auth_max_owner_token_seconds, + auth_max_delegated_token_seconds, }))) } } diff --git a/src/model.rs b/src/model.rs index b689b3a..8d6cbff 100644 --- a/src/model.rs +++ b/src/model.rs @@ -294,6 +294,8 @@ pub struct AppendItems { #[derive(Debug, Serialize)] pub struct AppendResult { pub items: Vec, + pub first_seq: i64, + pub last_seq: i64, pub replayed: bool, } diff --git a/src/store.rs b/src/store.rs index 2173340..006cf09 100644 --- a/src/store.rs +++ b/src/store.rs @@ -6,6 +6,7 @@ use sqlx::{PgPool, Postgres, Transaction}; use crate::{ api::AppState, + auth::AuthContext, capability, error::{ApiError, ApiResult}, files, @@ -17,6 +18,19 @@ use crate::{ }, }; +#[derive(Serialize)] +struct DelegatedAppendDigest<'a> { + operation: &'static str, + version: i16, + source: &'a str, + tenant_id: &'a str, + owner_ref: &'a str, + conversation_id: &'a str, + turn_id: Option<&'a str>, + agent_ref: &'a str, + items: &'a [Value], +} + #[derive(Serialize)] #[serde(tag = "mode", rename_all = "snake_case")] enum TurnStartDigest<'a> { @@ -81,6 +95,29 @@ fn turn_start_digest_v1(request: &StartTurn) -> ApiResult> { .to_vec()) } +fn delegated_append_digest_v1( + auth: &AuthContext, + conversation_id: &str, + agent_ref: &str, + request: &AppendItems, +) -> ApiResult> { + Ok(Sha256::digest( + serde_json_canonicalizer::to_vec(&DelegatedAppendDigest { + operation: "delegated_append", + version: 1, + source: &request.source, + tenant_id: &auth.tenant_id, + owner_ref: &auth.principal_id, + conversation_id, + turn_id: request.turn_id.as_deref(), + agent_ref, + items: &request.items, + }) + .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))?, + ) + .to_vec()) +} + pub async fn create_conversation( pool: &PgPool, actor: &Actor, @@ -694,6 +731,8 @@ pub async fn append_items( tx.commit().await?; return Ok(AppendResult { items, + first_seq, + last_seq, replayed: true, }); } @@ -786,10 +825,381 @@ pub async fn append_items( tx.commit().await?; Ok(AppendResult { items: inserted, + first_seq, + last_seq, replayed: false, }) } +pub async fn append_delegated_items( + pool: &PgPool, + auth: &AuthContext, + conversation_id: &str, + request: AppendItems, +) -> ApiResult { + let (bound_conversation, bound_turn, bound_agent) = auth + .delegated_bounds() + .ok_or(ApiError::Forbidden)?; + if conversation_id != bound_conversation { + return Err(ApiError::NotFound("Conversation not found.".into())); + } + if request.idempotency_key.trim().is_empty() || request.idempotency_key.len() > 200 { + return Err(ApiError::BadRequest( + "idempotency_key must contain 1 to 200 characters".into(), + )); + } + + // Compute before protocol validation so every changed retry, including a + // changed source, turn, count, order, or payload, receives the same typed + // idempotency conflict instead of disclosing the original result. + let request_digest = + delegated_append_digest_v1(auth, conversation_id, bound_agent, &request)?; + let mut tx = pool.begin().await?; + let conversation = lock_conversation(&mut tx, auth, conversation_id).await?; + let (turn_agent, turn_status) = sqlx::query_as::<_, (String, String)>( + "SELECT agent_ref, status FROM turns + WHERE id = $1 AND conversation_id = $2 FOR UPDATE", + ) + .bind(bound_turn) + .bind(conversation_id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| ApiError::NotFound("Turn not found.".into()))?; + if turn_agent != bound_agent { + return Err(ApiError::NotFound("Turn not found.".into())); + } + + if let Some( + ( + version, + digest, + source, + turn_id, + tenant_id, + owner_ref, + agent_ref, + item_count, + item_ids, + first_seq, + last_seq, + ), + ) = sqlx::query_as::< + _, + ( + Option, + Option>, + Option, + Option, + Option, + Option, + Option, + Option, + Option>, + i64, + i64, + ), + >( + "SELECT request_version, request_digest, source, turn_id, tenant_id, owner_ref, + agent_ref, item_count, item_ids, first_seq, last_seq + FROM append_batches WHERE conversation_id = $1 AND idempotency_key = $2", + ) + .bind(conversation_id) + .bind(&request.idempotency_key) + .fetch_optional(&mut *tx) + .await? + { + let exact = version == Some(1) + && digest.as_deref() == Some(request_digest.as_slice()) + && source.as_deref() == Some("agent") + && turn_id.as_deref() == Some(bound_turn) + && tenant_id.as_deref() == Some(auth.tenant_id.as_str()) + && owner_ref.as_deref() == Some(auth.principal_id.as_str()) + && agent_ref.as_deref() == Some(bound_agent) + && item_count == i32::try_from(request.items.len()).ok() + && item_ids.as_ref().is_some_and(|ids| ids.len() == request.items.len()); + if !exact { + return Err(coded_conflict( + "idempotency_key_reused", + "idempotency_key was already used for a different delegated append", + )); + } + let item_ids = item_ids.expect("exact delegated batch has item IDs"); + let items = sqlx::query_as::<_, Item>( + "SELECT * FROM conversation_items + WHERE conversation_id = $1 AND turn_id = $2 AND source = 'agent' + AND id = ANY($3) AND seq BETWEEN $4 AND $5 ORDER BY seq ASC", + ) + .bind(conversation_id) + .bind(bound_turn) + .bind(&item_ids) + .bind(first_seq) + .bind(last_seq) + .fetch_all(&mut *tx) + .await?; + if items.iter().map(|item| &item.id).ne(item_ids.iter()) { + return Err(coded_conflict( + "idempotency_result_deleted", + "the original delegated append result is no longer available", + )); + } + tx.commit().await?; + return Ok(AppendResult { + items, + first_seq, + last_seq, + replayed: true, + }); + } + + if request.source != "agent" { + return Err(ApiError::BadRequest( + "delegated writes require source agent".into(), + )); + } + if request.turn_id.as_deref() != Some(bound_turn) { + return Err(ApiError::NotFound("Turn not found.".into())); + } + if request.items.is_empty() || request.items.len() > 100 { + return Err(ApiError::BadRequest( + "items must contain between 1 and 100 entries".into(), + )); + } + for item in &request.items { + validate_delegated_output_item(item)?; + if !referenced_file_ids(item).is_empty() { + return Err(ApiError::BadRequest( + "delegated output cannot introduce threadmark file references".into(), + )); + } + } + if !matches!(turn_status.as_str(), "pending" | "streaming") { + return Err(coded_conflict( + "turn_not_active", + "delegated output can only be appended while the turn is active", + )); + } + + let first_seq = conversation.next_seq; + let last_seq = first_seq + .checked_add(request.items.len() as i64 - 1) + .ok_or_else(|| ApiError::Conflict("Conversation sequence is exhausted.".into()))?; + let next_seq = last_seq + .checked_add(1) + .ok_or_else(|| ApiError::Conflict("Conversation sequence is exhausted.".into()))?; + let item_count = i32::try_from(request.items.len()).expect("batch limit fits i32"); + let mut inserted = Vec::with_capacity(request.items.len()); + for (offset, payload) in request.items.into_iter().enumerate() { + inserted.push( + sqlx::query_as::<_, Item>( + "INSERT INTO conversation_items + (id, conversation_id, turn_id, seq, source, payload) + VALUES ($1, $2, $3, $4, 'agent', $5) RETURNING *", + ) + .bind(new_id("item")) + .bind(conversation_id) + .bind(bound_turn) + .bind(first_seq + offset as i64) + .bind(payload) + .fetch_one(&mut *tx) + .await?, + ); + } + let item_ids = inserted + .iter() + .map(|item| item.id.clone()) + .collect::>(); + sqlx::query( + "INSERT INTO append_batches + (conversation_id, idempotency_key, first_seq, last_seq, request_version, + request_digest, source, turn_id, tenant_id, owner_ref, agent_ref, item_count, item_ids) + VALUES ($1, $2, $3, $4, 1, $5, 'agent', $6, $7, $8, $9, $10, $11)", + ) + .bind(conversation_id) + .bind(request.idempotency_key) + .bind(first_seq) + .bind(last_seq) + .bind(request_digest) + .bind(bound_turn) + .bind(&auth.tenant_id) + .bind(&auth.principal_id) + .bind(bound_agent) + .bind(item_count) + .bind(item_ids) + .execute(&mut *tx) + .await?; + sqlx::query("UPDATE conversations SET next_seq = $2, updated_at = now() WHERE id = $1") + .bind(conversation_id) + .bind(next_seq) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(AppendResult { + items: inserted, + first_seq, + last_seq, + replayed: false, + }) +} + +fn validate_delegated_output_item(value: &Value) -> ApiResult<()> { + let object = value.as_object().ok_or_else(|| { + ApiError::BadRequest("each delegated output item must be a JSON object".into()) + })?; + let item_type = object.get("type").and_then(Value::as_str).ok_or_else(|| { + ApiError::BadRequest("each delegated output item requires a string type".into()) + })?; + if object + .iter() + .any(|(key, value)| key != "role" && contains_role_field(value)) + { + return Err(ApiError::BadRequest( + "delegated output cannot contain nested roles".into(), + )); + } + if item_type != "message" && object.contains_key("role") { + return Err(ApiError::BadRequest( + "roles are only allowed on assistant message output".into(), + )); + } + if object.get("id").is_some_and(|value| !value.is_string()) { + return Err(ApiError::BadRequest( + "delegated output id must be a string".into(), + )); + } + if let Some(status) = object.get("status") + && !matches!(status.as_str(), Some("in_progress" | "completed" | "incomplete")) + { + return Err(ApiError::BadRequest( + "delegated output status is unsupported".into(), + )); + } + match item_type { + "message" => { + validate_known_fields(object, &["type", "id", "status", "role", "content"])?; + if object.get("role").and_then(Value::as_str) != Some("assistant") { + return Err(ApiError::BadRequest( + "delegated message output requires role assistant".into(), + )); + } + let content = object.get("content").and_then(Value::as_array).ok_or_else(|| { + ApiError::BadRequest("delegated message output requires content array".into()) + })?; + if content.is_empty() { + return Err(ApiError::BadRequest( + "delegated message content cannot be empty".into(), + )); + } + for part in content { + let part = part.as_object().ok_or_else(|| { + ApiError::BadRequest("delegated message content must contain objects".into()) + })?; + match part.get("type").and_then(Value::as_str) { + Some("output_text") if part.get("text").is_some_and(Value::is_string) => { + validate_known_fields( + part, + &["type", "text", "annotations", "logprobs"], + )?; + for field in ["annotations", "logprobs"] { + if part.get(field).is_some_and(|value| !value.is_array()) { + return Err(ApiError::BadRequest(format!( + "output_text {field} must be an array" + ))); + } + } + } + Some("refusal") if part.get("refusal").is_some_and(Value::is_string) => { + validate_known_fields(part, &["type", "refusal"])?; + } + _ => { + return Err(ApiError::BadRequest( + "delegated message content supports output_text and refusal only".into(), + )); + } + } + } + } + "reasoning" => { + validate_known_fields( + object, + &["type", "id", "status", "summary", "content", "encrypted_content"], + )?; + validate_text_parts(object.get("summary"), "summary_text", "summary")?; + if let Some(content) = object.get("content") { + validate_text_parts(Some(content), "reasoning_text", "content")?; + } + if object + .get("encrypted_content") + .is_some_and(|value| !value.is_string()) + { + return Err(ApiError::BadRequest( + "reasoning encrypted_content must be a string".into(), + )); + } + } + "function_call" => { + validate_known_fields( + object, + &["type", "id", "status", "call_id", "name", "arguments"], + )?; + for field in ["call_id", "name", "arguments"] { + if !object.get(field).is_some_and(Value::is_string) { + return Err(ApiError::BadRequest(format!( + "delegated function_call requires string {field}" + ))); + } + } + } + _ => { + return Err(ApiError::BadRequest(format!( + "unsupported delegated output item type: {item_type}" + ))); + } + } + Ok(()) +} + +fn validate_known_fields( + object: &serde_json::Map, + allowed: &[&str], +) -> ApiResult<()> { + if let Some(field) = object.keys().find(|field| !allowed.contains(&field.as_str())) { + return Err(ApiError::BadRequest(format!( + "unsupported delegated output field: {field}" + ))); + } + Ok(()) +} + +fn contains_role_field(value: &Value) -> bool { + match value { + Value::Array(values) => values.iter().any(contains_role_field), + Value::Object(values) => { + values.contains_key("role") || values.values().any(contains_role_field) + } + _ => false, + } +} + +fn validate_text_parts(value: Option<&Value>, part_type: &str, field: &str) -> ApiResult<()> { + let parts = value.and_then(Value::as_array).ok_or_else(|| { + ApiError::BadRequest(format!("reasoning {field} must be an array")) + })?; + for part in parts { + let part = part.as_object().ok_or_else(|| { + ApiError::BadRequest(format!("reasoning {field} must contain objects")) + })?; + validate_known_fields(part, &["type", "text"])?; + if part.get("type").and_then(Value::as_str) != Some(part_type) + || !part.get("text").is_some_and(Value::is_string) + { + return Err(ApiError::BadRequest(format!( + "reasoning {field} supports {part_type} text parts only" + ))); + } + } + Ok(()) +} + fn referenced_file_ids(value: &Value) -> Vec { fn visit(value: &Value, ids: &mut Vec) { match value { @@ -1099,7 +1509,13 @@ pub async fn truncate_conversation( .bind(seq) .execute(&mut *tx) .await?; - sqlx::query("DELETE FROM append_batches WHERE conversation_id = $1 AND last_seq >= $2") + // Keep delegated records as tombstones so their keys can never allocate a + // different result after truncation. Exact retries report that the original + // result was deleted. Legacy owner batches retain their historical behavior. + sqlx::query( + "DELETE FROM append_batches + WHERE conversation_id = $1 AND last_seq >= $2 AND request_version IS NULL", + ) .bind(conversation_id) .bind(seq) .execute(&mut *tx) @@ -1354,4 +1770,127 @@ mod tests { vec!["file_document".to_owned(), "file_image".to_owned()] ); } + + #[test] + fn accepts_allowlisted_delegated_output_shapes() { + for item in [ + json!({ + "type": "message", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "done"}, + {"type": "refusal", "refusal": "cannot comply"} + ] + }), + json!({ + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "summary"}], + "content": [{"type": "reasoning_text", "text": "reasoning"}], + "encrypted_content": "opaque" + }), + json!({ + "type": "function_call", + "call_id": "call_1", + "name": "lookup", + "arguments": "{\"city\":\"Lisbon\"}" + }), + ] { + validate_delegated_output_item(&item).unwrap(); + } + } + + #[test] + fn rejects_roles_input_parts_and_unknown_delegated_shapes() { + for item in [ + json!({"type": "message", "role": "user", "content": [ + {"type": "output_text", "text": "x"} + ]}), + json!({"type": "message", "role": "system", "content": [ + {"type": "output_text", "text": "x"} + ]}), + json!({"type": "message", "role": "assistant", "content": [ + {"type": "input_text", "text": "x"} + ]}), + json!({"type": "function_call_output", "call_id": "call_1", "output": "x"}), + json!({"type": "future_output", "role": "assistant"}), + json!({ + "type": "function_call", "role": "assistant", "call_id": "call_1", + "name": "x", "arguments": "{}" + }), + json!({"type": "message", "role": "assistant", "content": [ + {"type": "output_text", "text": "x", "role": "user"} + ]}), + json!({"type": "message", "role": "assistant", "status": "queued", "content": [ + {"type": "output_text", "text": "x"} + ]}), + json!({"type": "function_call", "call_id": "call_1", "name": "x", "arguments": "{}", "output": "injected"}), + ] { + assert!( + validate_delegated_output_item(&item).is_err(), + "accepted {item}" + ); + } + } + + #[test] + fn delegated_digest_binds_source_turn_order_payload_and_authorization() { + let auth = AuthContext::delegated_for_test( + "tenant-a", + "owner-a", + "conv-a", + "turn-a", + "agent-a", + ); + let request = AppendItems { + idempotency_key: "retry-1".into(), + turn_id: Some("turn-a".into()), + source: "agent".into(), + items: vec![ + json!({"type": "function_call", "call_id": "1", "name": "a", "arguments": "{}"}), + json!({"type": "function_call", "call_id": "2", "name": "b", "arguments": "{}"}), + ], + }; + let original = delegated_append_digest_v1(&auth, "conv-a", "agent-a", &request).unwrap(); + + let mut changed = AppendItems { + items: request.items.iter().rev().cloned().collect(), + ..request + }; + assert_ne!( + original, + delegated_append_digest_v1(&auth, "conv-a", "agent-a", &changed).unwrap() + ); + changed.items.pop(); + assert_ne!( + original, + delegated_append_digest_v1(&auth, "conv-a", "agent-a", &changed).unwrap() + ); + changed.items[0]["name"] = json!("changed"); + assert_ne!( + original, + delegated_append_digest_v1(&auth, "conv-a", "agent-a", &changed).unwrap() + ); + changed.source = "system".into(); + assert_ne!( + original, + delegated_append_digest_v1(&auth, "conv-a", "agent-a", &changed).unwrap() + ); + changed.source = "agent".into(); + changed.turn_id = Some("turn-b".into()); + assert_ne!( + original, + delegated_append_digest_v1(&auth, "conv-a", "agent-a", &changed).unwrap() + ); + assert_ne!( + original, + delegated_append_digest_v1(&auth, "conv-b", "agent-a", &changed).unwrap() + ); + let other_auth = AuthContext::delegated_for_test( + "tenant-b", "owner-b", "conv-a", "turn-a", "agent-b", + ); + assert_ne!( + original, + delegated_append_digest_v1(&other_auth, "conv-a", "agent-b", &changed).unwrap() + ); + } } From 1d78ec6c5f90c39331c6a6120238e1ae1f7a6a0d Mon Sep 17 00:00:00 2001 From: "dispatch-developer[bot]" <306909890+dispatch-developer[bot]@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:31:37 +0000 Subject: [PATCH 2/3] style: format Rust sources --- src/api.rs | 12 +++--- src/auth.rs | 5 ++- src/config.rs | 3 +- src/store.rs | 108 ++++++++++++++++++++++++++------------------------ 4 files changed, 67 insertions(+), 61 deletions(-) diff --git a/src/api.rs b/src/api.rs index deef725..6839632 100644 --- a/src/api.rs +++ b/src/api.rs @@ -222,13 +222,11 @@ async fn append_items( serde_json::from_slice(&body) .map_err(|error| ApiError::BadRequest(format!("invalid request JSON: {error}")))? }; - Ok(Json( - if auth.is_delegated() { - store::append_delegated_items(&state.pool, &auth, &id, request).await? - } else { - store::append_items(&state.pool, &auth, &id, request).await? - }, - )) + Ok(Json(if auth.is_delegated() { + store::append_delegated_items(&state.pool, &auth, &id, request).await? + } else { + store::append_items(&state.pool, &auth, &id, request).await? + })) } fn parse_strict_json(body: &[u8]) -> ApiResult { diff --git a/src/auth.rs b/src/auth.rs index 05ead42..1973312 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -266,7 +266,10 @@ impl Claims { "owner_session" if self.exp.saturating_sub(self.iat) <= verifier.max_owner_seconds && self.conversation_id.is_none() - && self.turn_id.is_none() => TokenKind::OwnerSession, + && self.turn_id.is_none() => + { + TokenKind::OwnerSession + } "delegated_agent" if self.exp.saturating_sub(self.iat) <= verifier.max_delegated_seconds && self.conversation_id.as_deref().is_some_and(valid_id) diff --git a/src/config.rs b/src/config.rs index 04c5adf..4823794 100644 --- a/src/config.rs +++ b/src/config.rs @@ -101,8 +101,7 @@ impl Config { auth_max_owner_token_seconds > 0, "AUTH_MAX_OWNER_TOKEN_SECONDS must be greater than zero" ); - let auth_max_delegated_token_seconds = - parse("AUTH_MAX_DELEGATED_TOKEN_SECONDS", "600")?; + let auth_max_delegated_token_seconds = parse("AUTH_MAX_DELEGATED_TOKEN_SECONDS", "600")?; ensure!( auth_max_delegated_token_seconds > 0, "AUTH_MAX_DELEGATED_TOKEN_SECONDS must be greater than zero" diff --git a/src/store.rs b/src/store.rs index 006cf09..78eb15e 100644 --- a/src/store.rs +++ b/src/store.rs @@ -837,9 +837,8 @@ pub async fn append_delegated_items( conversation_id: &str, request: AppendItems, ) -> ApiResult { - let (bound_conversation, bound_turn, bound_agent) = auth - .delegated_bounds() - .ok_or(ApiError::Forbidden)?; + let (bound_conversation, bound_turn, bound_agent) = + auth.delegated_bounds().ok_or(ApiError::Forbidden)?; if conversation_id != bound_conversation { return Err(ApiError::NotFound("Conversation not found.".into())); } @@ -852,8 +851,7 @@ pub async fn append_delegated_items( // Compute before protocol validation so every changed retry, including a // changed source, turn, count, order, or payload, receives the same typed // idempotency conflict instead of disclosing the original result. - let request_digest = - delegated_append_digest_v1(auth, conversation_id, bound_agent, &request)?; + let request_digest = delegated_append_digest_v1(auth, conversation_id, bound_agent, &request)?; let mut tx = pool.begin().await?; let conversation = lock_conversation(&mut tx, auth, conversation_id).await?; let (turn_agent, turn_status) = sqlx::query_as::<_, (String, String)>( @@ -869,21 +867,19 @@ pub async fn append_delegated_items( return Err(ApiError::NotFound("Turn not found.".into())); } - if let Some( - ( - version, - digest, - source, - turn_id, - tenant_id, - owner_ref, - agent_ref, - item_count, - item_ids, - first_seq, - last_seq, - ), - ) = sqlx::query_as::< + if let Some(( + version, + digest, + source, + turn_id, + tenant_id, + owner_ref, + agent_ref, + item_count, + item_ids, + first_seq, + last_seq, + )) = sqlx::query_as::< _, ( Option, @@ -899,10 +895,10 @@ pub async fn append_delegated_items( i64, ), >( - "SELECT request_version, request_digest, source, turn_id, tenant_id, owner_ref, + "SELECT request_version, request_digest, source, turn_id, tenant_id, owner_ref, agent_ref, item_count, item_ids, first_seq, last_seq FROM append_batches WHERE conversation_id = $1 AND idempotency_key = $2", - ) + ) .bind(conversation_id) .bind(&request.idempotency_key) .fetch_optional(&mut *tx) @@ -916,7 +912,9 @@ pub async fn append_delegated_items( && owner_ref.as_deref() == Some(auth.principal_id.as_str()) && agent_ref.as_deref() == Some(bound_agent) && item_count == i32::try_from(request.items.len()).ok() - && item_ids.as_ref().is_some_and(|ids| ids.len() == request.items.len()); + && item_ids + .as_ref() + .is_some_and(|ids| ids.len() == request.items.len()); if !exact { return Err(coded_conflict( "idempotency_key_reused", @@ -1067,7 +1065,10 @@ fn validate_delegated_output_item(value: &Value) -> ApiResult<()> { )); } if let Some(status) = object.get("status") - && !matches!(status.as_str(), Some("in_progress" | "completed" | "incomplete")) + && !matches!( + status.as_str(), + Some("in_progress" | "completed" | "incomplete") + ) { return Err(ApiError::BadRequest( "delegated output status is unsupported".into(), @@ -1081,9 +1082,12 @@ fn validate_delegated_output_item(value: &Value) -> ApiResult<()> { "delegated message output requires role assistant".into(), )); } - let content = object.get("content").and_then(Value::as_array).ok_or_else(|| { - ApiError::BadRequest("delegated message output requires content array".into()) - })?; + let content = object + .get("content") + .and_then(Value::as_array) + .ok_or_else(|| { + ApiError::BadRequest("delegated message output requires content array".into()) + })?; if content.is_empty() { return Err(ApiError::BadRequest( "delegated message content cannot be empty".into(), @@ -1095,10 +1099,7 @@ fn validate_delegated_output_item(value: &Value) -> ApiResult<()> { })?; match part.get("type").and_then(Value::as_str) { Some("output_text") if part.get("text").is_some_and(Value::is_string) => { - validate_known_fields( - part, - &["type", "text", "annotations", "logprobs"], - )?; + validate_known_fields(part, &["type", "text", "annotations", "logprobs"])?; for field in ["annotations", "logprobs"] { if part.get(field).is_some_and(|value| !value.is_array()) { return Err(ApiError::BadRequest(format!( @@ -1112,7 +1113,8 @@ fn validate_delegated_output_item(value: &Value) -> ApiResult<()> { } _ => { return Err(ApiError::BadRequest( - "delegated message content supports output_text and refusal only".into(), + "delegated message content supports output_text and refusal only" + .into(), )); } } @@ -1121,7 +1123,14 @@ fn validate_delegated_output_item(value: &Value) -> ApiResult<()> { "reasoning" => { validate_known_fields( object, - &["type", "id", "status", "summary", "content", "encrypted_content"], + &[ + "type", + "id", + "status", + "summary", + "content", + "encrypted_content", + ], )?; validate_text_parts(object.get("summary"), "summary_text", "summary")?; if let Some(content) = object.get("content") { @@ -1162,7 +1171,10 @@ fn validate_known_fields( object: &serde_json::Map, allowed: &[&str], ) -> ApiResult<()> { - if let Some(field) = object.keys().find(|field| !allowed.contains(&field.as_str())) { + if let Some(field) = object + .keys() + .find(|field| !allowed.contains(&field.as_str())) + { return Err(ApiError::BadRequest(format!( "unsupported delegated output field: {field}" ))); @@ -1181,9 +1193,9 @@ fn contains_role_field(value: &Value) -> bool { } fn validate_text_parts(value: Option<&Value>, part_type: &str, field: &str) -> ApiResult<()> { - let parts = value.and_then(Value::as_array).ok_or_else(|| { - ApiError::BadRequest(format!("reasoning {field} must be an array")) - })?; + let parts = value + .and_then(Value::as_array) + .ok_or_else(|| ApiError::BadRequest(format!("reasoning {field} must be an array")))?; for part in parts { let part = part.as_object().ok_or_else(|| { ApiError::BadRequest(format!("reasoning {field} must contain objects")) @@ -1516,10 +1528,10 @@ pub async fn truncate_conversation( "DELETE FROM append_batches WHERE conversation_id = $1 AND last_seq >= $2 AND request_version IS NULL", ) - .bind(conversation_id) - .bind(seq) - .execute(&mut *tx) - .await?; + .bind(conversation_id) + .bind(seq) + .execute(&mut *tx) + .await?; sqlx::query("DELETE FROM continuations WHERE conversation_id = $1 AND through_seq >= $2") .bind(conversation_id) .bind(seq) @@ -1834,13 +1846,8 @@ mod tests { #[test] fn delegated_digest_binds_source_turn_order_payload_and_authorization() { - let auth = AuthContext::delegated_for_test( - "tenant-a", - "owner-a", - "conv-a", - "turn-a", - "agent-a", - ); + let auth = + AuthContext::delegated_for_test("tenant-a", "owner-a", "conv-a", "turn-a", "agent-a"); let request = AppendItems { idempotency_key: "retry-1".into(), turn_id: Some("turn-a".into()), @@ -1885,9 +1892,8 @@ mod tests { original, delegated_append_digest_v1(&auth, "conv-b", "agent-a", &changed).unwrap() ); - let other_auth = AuthContext::delegated_for_test( - "tenant-b", "owner-b", "conv-a", "turn-a", "agent-b", - ); + let other_auth = + AuthContext::delegated_for_test("tenant-b", "owner-b", "conv-a", "turn-a", "agent-b"); assert_ne!( original, delegated_append_digest_v1(&other_auth, "conv-a", "agent-b", &changed).unwrap() From 188bb1c2f2aea4ae321cece23a665a72912bb142 Mon Sep 17 00:00:00 2001 From: "dispatch-developer[bot]" <306909890+dispatch-developer[bot]@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:56:32 +0000 Subject: [PATCH 3/3] fix: remove unused append import --- src/api.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/api.rs b/src/api.rs index 6839632..8a5d81d 100644 --- a/src/api.rs +++ b/src/api.rs @@ -17,11 +17,11 @@ use crate::{ error::{ApiError, ApiResult}, files, 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, + Actor, 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, }, object_store::ObjectStore, store, uploads,