Add bounded snapshot-consistent agent replay

.env.example

@@ -9,6 +9,10 @@ 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
+AGENT_REPLAY_MAX_ITEMS=200
+AGENT_REPLAY_MAX_BYTES=1048576
+AGENT_REPLAY_STRIP_TOP_LEVEL_FIELDS=id
 FILE_MAX_MB=32
 S3_ENDPOINT=http://localhost:9010
 S3_PUBLIC_URL=http://localhost:9010

README.md

@@ -77,6 +77,7 @@ This is an experiment and its API is not stable. The current slice establishes:
 - Idempotent item batches and turn creation.
 - Cursor-based item reads.
 - Open Responses replay projection with optional top-level `id` removal.
+- Snapshot-consistent, size-bounded text replay for delegated agent turns.
 - Agent-scoped continuation records and optional private checkpoint state.
 
 Editing, branching, retention, event delivery, production authentication, and
@@ -140,13 +141,54 @@ 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. It also accepts delegated-agent tokens only for the agent replay
+operation described below. Delegated writes remain disabled.
 
 An agent called by Parley can receive a short-lived token scoped to the same
 tenant, principal, conversation, turn, and agent deployment. That authorization
 layer is deliberately separate from the ledger model.
 
+### Bounded agent replay
+
+`POST /v1/conversations/{conversation_id}/turns/{turn_id}/agent-replay` is the
+initial Bonsai replay integration. It accepts no request body and requires a
+`delegated_agent` JWT with `transcript:read` and exact `tenant`, `principal`,
+`conversation_id`, `turn_id`, and `agent_ref` bounds. Wrong actor or resource
+bounds return a non-enumerating error. The turn must have been created by
+`POST /v1/turn-starts`; its recorded `last_seq` is the immutable replay cursor.
+
+The operation opens a PostgreSQL repeatable-read, read-only transaction before
+resolving ownership, the atomic-start boundary, and ordered items. It verifies
+the complete triggering batch still exists in that snapshot. A concurrent
+truncate is therefore observed wholly before or wholly after its commit; a
+snapshot missing the boundary returns `replay_snapshot_unavailable` and never
+returns a cursor for absent turn-start input.
+
+The first integration supports only these historical message shapes:
+
+- `type=message`, `role=user`, with a nonempty array of `input_text` parts
+  containing only string `text` plus the `type` discriminator;
+- `type=message`, `role=assistant`, with a nonempty array of `output_text` parts
+  containing string `text`, the `type` discriminator, and optional
+  `annotations`.
+
+Other item types, roles, non-text parts, mixed content, and media return
+`unsupported_agent_replay_item`. In particular, file and image parts are never
+forwarded, and any canonical file URI anywhere in an item is rejected, so
+unresolved `threadmark://` resources cannot reach a model provider through this
+endpoint. Accepted top-level item fields are otherwise preserved. Only
+top-level fields named by `AGENT_REPLAY_STRIP_TOP_LEVEL_FIELDS` (comma-separated,
+default `id`) are removed; nested fields are untouched.
+
+`AGENT_REPLAY_MAX_ITEMS` (default `200`) and `AGENT_REPLAY_MAX_BYTES` (default
+`1048576`) are hard inclusive limits. The byte limit is the exact compact JSON
+serialization of the returned `input` array after configured field removal.
+Exceeding either limit returns HTTP `413` with
+`error.code=context_limit_exceeded` before a projection is returned.
+
+The existing owner endpoint, `POST /v1/conversations/{id}/replay`, is unchanged:
+it remains an opaque, multimodal projection with caller-selected file delivery.
+
 ## Example flow
 
 Create a conversation:
@@ -236,6 +278,7 @@ GET /v1/continuations/resp_abc?agent_ref=research-agent%2Fprod
 | `GET` | `/v1/conversations/{id}/items` | Read ordered items after a sequence cursor |
 | `POST` | `/v1/conversations/{id}/items` | Atomically append an idempotent item batch |
 | `POST` | `/v1/conversations/{id}/replay` | Build an Open Responses input array |
+| `POST` | `/v1/conversations/{conversation_id}/turns/{turn_id}/agent-replay` | Build bounded text input for a delegated agent turn |
 | `POST` | `/v1/conversations/{id}/turns` | Create an idempotent turn |
 | `PATCH` | `/v1/turns/{id}` | Update turn state and outcome |
 | `POST` | `/v1/conversations/{id}/continuations` | Record an agent checkpoint |

src/api.rs

@@ -17,11 +17,12 @@ 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, AgentReplayResult, 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,
     },
     object_store::ObjectStore,
     store, uploads,
@@ -54,6 +55,10 @@ pub fn router(state: AppState) -> Router {
             get(list_items).post(append_items),
         )
         .route("/v1/conversations/{id}/replay", post(replay))
+        .route(
+            "/v1/conversations/{conversation_id}/turns/{turn_id}/agent-replay",
+            post(agent_replay),
+        )
         .route(
             "/v1/conversations/{id}/turns",
             get(list_turns).post(create_turn),
@@ -233,6 +238,26 @@ async fn replay(
     Ok(Json(store::replay(&state, &auth, &id, request).await?))
 }
 
+async fn agent_replay(
+    State(state): State<AppState>,
+    auth: AuthContext,
+    Path((conversation_id, turn_id)): Path<(String, String)>,
+    body: Bytes,
+) -> ApiResult<Json<AgentReplayResult>> {
+    auth.require(Permission::AgentReplay)?;
+    if !body.is_empty() {
+        return Err(ApiError::BadRequest(
+            "agent replay does not accept a request body".into(),
+        ));
+    }
+    let agent_ref = auth
+        .require_agent_replay_scope(&conversation_id, &turn_id)?
+        .to_owned();
+    Ok(Json(
+        store::agent_replay(&state, &auth, &conversation_id, &turn_id, &agent_ref).await?,
+    ))
+}
+
 async fn create_turn(
     State(state): State<AppState>,
     auth: AuthContext,

src/auth.rs

@@ -32,6 +32,7 @@ struct JwtVerifier {
     issuer: String,
     audience: String,
     max_owner_seconds: u64,
+    max_delegated_seconds: u64,
     keys: HashMap<String, DecodingKey>,
 }
 
@@ -40,9 +41,18 @@ pub struct AuthContext {
     pub actor: Actor,
     pub client_id: String,
     agent_ref: Option<String>,
+    conversation_id: Option<String>,
+    turn_id: Option<String>,
+    token_kind: TokenKind,
     permissions: HashSet<Permission>,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum TokenKind {
+    OwnerSession,
+    DelegatedAgent,
+}
+
 #[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
 pub enum Permission {
     ConversationList,
@@ -53,6 +63,7 @@ pub enum Permission {
     ConversationTruncate,
     ConversationRegenerate,
     TranscriptRead,
+    AgentReplay,
     TranscriptAppend,
     TurnCreate,
     TurnRead,
@@ -102,6 +113,8 @@ struct Claims {
     principal: String,
     permissions: Vec<String>,
     agent_ref: Option<String>,
+    conversation_id: Option<String>,
+    turn_id: Option<String>,
 }
 
 #[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<Self> {
         ensure!(!jwks.keys.is_empty(), "JWKS contains no keys");
@@ -191,6 +206,7 @@ impl JwtVerifier {
             issuer,
             audience,
             max_owner_seconds,
+            max_delegated_seconds,
             keys,
         })
     }
@@ -219,17 +235,25 @@ impl JwtVerifier {
 impl Claims {
     fn context(self, verifier: &JwtVerifier) -> Option<AuthContext> {
         let now = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs();
+        let token_kind = match self.token_kind.as_str() {
+            "owner_session" => TokenKind::OwnerSession,
+            "delegated_agent" => TokenKind::DelegatedAgent,
+            _ => return None,
+        };
         let valid_audience = match &self.aud {
             Audience::One(value) => value == &verifier.audience,
             Audience::Many(values) => values.len() == 1 && values[0] == verifier.audience,
         };
         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
+            || self.exp.saturating_sub(self.iat)
+                > match token_kind {
+                    TokenKind::OwnerSession => verifier.max_owner_seconds,
+                    TokenKind::DelegatedAgent => verifier.max_delegated_seconds,
+                }
             || !valid_id(&self.sub)
             || !valid_id(&self.client_id)
             || self.client_id == "threadmark:trusted-headers"
@@ -240,21 +264,44 @@ impl Claims {
                 .agent_ref
                 .as_deref()
                 .is_some_and(|value| !valid_id(value))
+            || self
+                .conversation_id
+                .as_deref()
+                .is_some_and(|value| !valid_id(value))
+            || self
+                .turn_id
+                .as_deref()
+                .is_some_and(|value| !valid_id(value))
         {
             return None;
         }
-        let permissions = self
+        let mut permissions = self
             .permissions
             .iter()
             .map(|value| Permission::parse(value))
             .collect::<Option<HashSet<_>>>()?;
+        if token_kind == TokenKind::DelegatedAgent
+            && (self.conversation_id.is_none()
+                || self.turn_id.is_none()
+                || self.agent_ref.is_none()
+                || permissions.len() != 1
+                || !permissions.contains(&Permission::TranscriptRead))
+        {
+            return None;
+        }
+        if token_kind == TokenKind::DelegatedAgent {
+            permissions = [Permission::AgentReplay].into_iter().collect();
+        }
         Some(AuthContext {
             actor: Actor {
                 tenant_id: self.tenant,
                 principal_id: self.principal,
             },
             client_id: self.client_id,
             agent_ref: self.agent_ref,
+            conversation_id: self.conversation_id,
+            turn_id: self.turn_id,
+            token_kind,
             permissions,
         })
     }
@@ -280,6 +327,22 @@ impl AuthContext {
             _ => Ok(()),
         }
     }
+
+    pub fn require_agent_replay_scope(
+        &self,
+        conversation_id: &str,
+        turn_id: &str,
+    ) -> Result<&str, ApiError> {
+        if self.token_kind != TokenKind::DelegatedAgent
+            || self.conversation_id.as_deref() != Some(conversation_id)
+            || self.turn_id.as_deref() != Some(turn_id)
+        {
+            return Err(ApiError::NotFound("Agent replay not found.".into()));
+        }
+        self.agent_ref
+            .as_deref()
+            .ok_or_else(|| ApiError::NotFound("Agent replay not found.".into()))
+    }
 }
 
 impl std::ops::Deref for AuthContext {
@@ -353,6 +416,9 @@ fn trusted_headers(headers: &HeaderMap) -> Result<AuthContext, ApiError> {
         },
         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 +451,7 @@ mod tests {
             "https://issuer.example".into(),
             "threadmark-api".into(),
             300,
+            600,
             jwks,
         )
         .unwrap()
@@ -494,6 +561,57 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn delegated_replay_token_enforces_all_resource_bounds() {
+        let mut claims = claims();
+        claims["token_kind"] = json!("delegated_agent");
+        claims["permissions"] = json!(["transcript:read"]);
+        claims["conversation_id"] = json!("conv_1");
+        claims["turn_id"] = json!("turn_1");
+        claims["agent_ref"] = json!("bonsai/prod");
+        let context = verifier().authenticate(&headers(&claims)).unwrap();
+        assert_eq!(
+            context
+                .require_agent_replay_scope("conv_1", "turn_1")
+                .unwrap(),
+            "bonsai/prod"
+        );
+        assert!(context.require(Permission::AgentReplay).is_ok());
+        assert!(matches!(
+            context.require(Permission::TranscriptRead),
+            Err(ApiError::Forbidden)
+        ));
+        assert!(matches!(
+            context.require_agent_replay_scope("conv_other", "turn_1"),
+            Err(ApiError::NotFound(_))
+        ));
+        assert!(matches!(
+            context.require_agent_replay_scope("conv_1", "turn_other"),
+            Err(ApiError::NotFound(_))
+        ));
+    }
+
+    #[test]
+    fn delegated_tokens_reject_missing_scope_and_owner_permissions() {
+        let mut missing_turn = claims();
+        missing_turn["token_kind"] = json!("delegated_agent");
+        missing_turn["permissions"] = json!(["transcript:read"]);
+        missing_turn["conversation_id"] = json!("conv_1");
+        missing_turn["agent_ref"] = json!("bonsai/prod");
+        assert!(matches!(
+            verifier().authenticate(&headers(&missing_turn)),
+            Err(ApiError::Unauthorized)
+        ));
+
+        let mut broad = missing_turn;
+        broad["turn_id"] = json!("turn_1");
+        broad["permissions"] = json!(["transcript:read", "conversation:read"]);
+        assert!(matches!(
+            verifier().authenticate(&headers(&broad)),
+            Err(ApiError::Unauthorized)
+        ));
+    }
+
     #[test]
     fn rejects_excessive_lifetime_and_multi_audience_tokens() {
         let mut long_lived = claims();

src/capability.rs

@@ -120,6 +120,10 @@ mod tests {
             auth_audience: None,
             auth_jwks_url: None,
             auth_max_owner_token_seconds: 300,
+            auth_max_delegated_token_seconds: 600,
+            agent_replay_max_items: 200,
+            agent_replay_max_bytes: 1024 * 1024,
+            agent_replay_strip_top_level_fields: vec!["id".into()],
         }))
     }