Enforce constrained delegated-agent writes

.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

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

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)
+);

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,
@@ -212,12 +212,34 @@ async fn append_items(
     State(state): State<AppState>,
     auth: AuthContext,
     Path(id): Path<String>,
-    Json(request): Json<AppendItems>,
+    body: Bytes,
 ) -> ApiResult<Json<AppendResult>> {
-    auth.require(Permission::TranscriptAppend)?;
-    Ok(Json(
-        store::append_items(&state.pool, &auth, &id, request).await?,
-    ))
+    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(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<T: serde::de::DeserializeOwned>(body: &[u8]) -> ApiResult<T> {
+    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(