From 356b192f0d644c9774c040fa207a5d8d48d8f052 Mon Sep 17 00:00:00 2001 From: Farnam Date: Fri, 23 Oct 2026 13:49:38 +0330 Subject: [PATCH 1/4] Add atomic edits and lane moves for pending run messages --- .../src/run_queue/README.md | 10 ++- .../tinyagents-harness/src/run_queue/mod.rs | 79 +++++++++++++++++-- .../src/run_queue/mod_tests.rs | 57 +++++++++++++ 3 files changed, 140 insertions(+), 6 deletions(-) diff --git a/crates/tinyagents-harness/src/run_queue/README.md b/crates/tinyagents-harness/src/run_queue/README.md index c2c6ae0fc..0f23dd849 100644 --- a/crates/tinyagents-harness/src/run_queue/README.md +++ b/crates/tinyagents-harness/src/run_queue/README.md @@ -33,6 +33,11 @@ stops, pauses, and deferrals leave the queue untouched. Details: `QueueMode`-aware form; `status()` snapshots per-lane depth; `clear()` empties every lane and returns how many items were dropped. + `edit_first_where(predicate, edit)` changes one pending payload in place; + `move_first_where(predicate, lane, index)` relocates one pending item to + a destination lane and position. Both return `false` if delivery already + consumed the item. Indices address the destination after removal and clamp + to its length. Other items keep their relative order. - [`QueueLane`] — which lane an item belongs to: `Steer` (inject at the next safe boundary as an instruction), `Followup` (dispatch as a fresh turn once the active run completes), `Collect` (handed back to the host at run end as @@ -54,11 +59,14 @@ stops, pauses, and deferrals leave the queue untouched. Details: ## Operational constraints - Every method serializes through a single `std::sync::Mutex` guarding all - three lanes together. It is held only for one vector operation and never + three lanes together. It is held only for in-memory operations and never across an `.await`, so the `async` methods never block a runtime thread for long and the synchronous `push_now`/`drain_now` are safe anywhere. The queue is meant to hold a handful of pending items, not a high-throughput channel. +- Edit callbacks run under that lock. Validate replacement data before calling, + keep callbacks brief, and never call back into the queue. A move to `Steer` + still waits for a safe runtime boundary; it does not cancel an active tool. - `SteeringForwarderGuard` drains and delivers under its shutdown gate and requeues residual (delivered-but-unapplied) steers synchronously in `Drop`, so once the guard has dropped, every steer the run did not apply is back in diff --git a/crates/tinyagents-harness/src/run_queue/mod.rs b/crates/tinyagents-harness/src/run_queue/mod.rs index 9d52d304e..97adcc8b6 100644 --- a/crates/tinyagents-harness/src/run_queue/mod.rs +++ b/crates/tinyagents-harness/src/run_queue/mod.rs @@ -57,8 +57,7 @@ pub struct RunQueue { inner: Mutex>, } -/// The three lanes behind [`RunQueue`]'s lock, each an append-only FIFO until -/// drained. +/// The three FIFO lanes behind [`RunQueue`]'s lock. #[derive(Debug)] struct RunQueueInner { /// Immediate steering messages, consumed as soon as the loop reaches a @@ -82,9 +81,9 @@ impl RunQueue { } } - /// Locks the lanes. A panic while the lock was held cannot leave a lane - /// half-updated (every operation is a single vector call), so a poisoned - /// lock is recovered rather than propagated. + /// Locks the lanes, recovering a poisoned lock. Vector invariants remain + /// valid after unwinding; an editing callback must keep its payload valid + /// even when it panics because callback changes are not rolled back. fn lock(&self) -> MutexGuard<'_, RunQueueInner> { self.inner.lock().unwrap_or_else(PoisonError::into_inner) } @@ -260,6 +259,76 @@ impl RunQueue { let after = inner.steers.len() + inner.followups.len() + inner.collects.len(); before - after } + + /// Edits the first matching pending item in delivery order, atomically + /// with respect to delivery, removal, and other queue mutations. + /// + /// Returns `false` if the item has already been consumed. `edit` runs + /// under the queue lock: it must be brief and must not call this queue. + /// Hosts retain responsibility for validating the replacement payload. + pub async fn edit_first_where( + &self, + pred: impl Fn(&T) -> bool, + edit: impl FnOnce(&mut T), + ) -> bool { + let mut inner = self.lock(); + let RunQueueInner { + steers, + followups, + collects, + } = &mut *inner; + let item = steers + .iter_mut() + .chain(followups.iter_mut()) + .chain(collects.iter_mut()) + .find(|item| pred(item)); + match item { + Some(item) => { + edit(item); + true + } + None => false, + } + } + + /// Moves the first matching pending item to `lane` at `index`, atomically + /// with respect to delivery. `index` addresses the destination after the + /// item is removed and is clamped to that lane's length. + /// + /// Moving a follow-up into the steer lane requests delivery at the next + /// safe boundary; it never interrupts a tool batch. Returns `false` when + /// the item has already been consumed. All other items retain their order. + pub async fn move_first_where( + &self, + pred: impl Fn(&T) -> bool, + lane: QueueLane, + index: usize, + ) -> bool { + let mut inner = self.lock(); + let item = { + let RunQueueInner { + steers, + followups, + collects, + } = &mut *inner; + [steers, followups, collects].into_iter().find_map(|items| { + items + .iter() + .position(&pred) + .map(|position| items.remove(position)) + }) + }; + let Some(item) = item else { + return false; + }; + let target = match lane { + QueueLane::Steer => &mut inner.steers, + QueueLane::Followup => &mut inner.followups, + QueueLane::Collect => &mut inner.collects, + }; + target.insert(index.min(target.len()), item); + true + } } impl Default for RunQueue { diff --git a/crates/tinyagents-harness/src/run_queue/mod_tests.rs b/crates/tinyagents-harness/src/run_queue/mod_tests.rs index d27735230..b1abf543e 100644 --- a/crates/tinyagents-harness/src/run_queue/mod_tests.rs +++ b/crates/tinyagents-harness/src/run_queue/mod_tests.rs @@ -164,3 +164,60 @@ async fn remove_where_with_no_match_removes_nothing() { assert_eq!(removed, 0); assert_eq!(queue.status().await.total, 2); } + +#[tokio::test] +async fn editing_changes_only_the_first_pending_match_in_delivery_order() { + let queue = RunQueue::new(); + queue.push(QueueLane::Steer, 1).await; + queue.push(QueueLane::Followup, 1).await; + assert!( + queue + .edit_first_where(|item| *item == 1, |item| *item = 2) + .await + ); + assert_eq!( + queue.snapshot().await, + vec![(QueueLane::Steer, 2), (QueueLane::Followup, 1)] + ); + queue.drain(QueueLane::Steer).await; + assert!( + !queue + .edit_first_where( + |item| *item == 2, + |_| panic!("consumed item must not be edited") + ) + .await + ); +} + +#[tokio::test] +async fn moving_can_reorder_and_promote_a_followup_without_dropping_other_items() { + let queue = RunQueue::new(); + for item in [1, 2, 3] { + queue.push(QueueLane::Followup, item).await; + } + assert!( + queue + .move_first_where(|item| *item == 3, QueueLane::Followup, 0) + .await + ); + assert!( + queue + .move_first_where(|item| *item == 2, QueueLane::Steer, usize::MAX) + .await + ); + assert_eq!( + queue.snapshot().await, + vec![ + (QueueLane::Steer, 2), + (QueueLane::Followup, 3), + (QueueLane::Followup, 1) + ] + ); + assert!( + !queue + .move_first_where(|item| *item == 99, QueueLane::Collect, 0) + .await + ); + assert_eq!(queue.status().await.total, 3); +} From 029657413446b80145a64d4ee82ed27c9a86ff53 Mon Sep 17 00:00:00 2001 From: Farnam Date: Fri, 23 Oct 2026 13:49:56 +0330 Subject: [PATCH 2/4] Persist addressed run inputs with atomic deduplication --- .../src/run_ledger/command_center/README.md | 16 ++ .../src/run_ledger/drivers/events.rs | 8 +- .../src/run_ledger/drivers/inbox.rs | 193 ++++++++++++++++++ .../src/run_ledger/drivers/inbox_tests.rs | 120 +++++++++++ .../src/run_ledger/drivers/mod.rs | 1 + .../src/run_ledger/drivers/schema.rs | 11 +- .../src/run_ledger/drivers/schema_tests.rs | 1 + .../src/run_ledger/drivers/storage.rs | 4 + .../src/run_ledger/inbox.rs | 63 ++++++ .../tinyagents-session/src/run_ledger/mod.rs | 1 + 10 files changed, 412 insertions(+), 6 deletions(-) create mode 100644 crates/tinyagents-session/src/run_ledger/drivers/inbox.rs create mode 100644 crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs create mode 100644 crates/tinyagents-session/src/run_ledger/inbox.rs diff --git a/crates/tinyagents-session/src/run_ledger/command_center/README.md b/crates/tinyagents-session/src/run_ledger/command_center/README.md index b8a01ecf5..5efff064d 100644 --- a/crates/tinyagents-session/src/run_ledger/command_center/README.md +++ b/crates/tinyagents-session/src/run_ledger/command_center/README.md @@ -38,3 +38,19 @@ the host passes an agent-id to display-name closure and owns the RPC surface. returns `InvalidTransition` when the verb no longer applies. - `stop` stamps `completed_at` and stores the optional reason as `error`; `retry` and `continue` clear both; `follow_up` leaves the status unchanged. +# Durable addressed input draft + +`run_ledger::inbox::RunInboxStore` is a separate optional capability implemented +by `DriverRunLedger`. `enqueue_input` validates the exact persisted recipient +lineage and commits the queued record, deduplication identity and `input_queued` +event in one checked batch. Reusing the root/submission id for different content +returns a conflict. Receipts survive process restart and list in event-sequence +order. A backend without checked atomic batches returns its native error. + +This is an admission draft, not a complete delivery implementation. No operation +marks input delivered, changes a run to running, or launches a worker. Recipient +integration must atomically commit admitted input into the durable session and +record consumption before exposing delivery/resumption controls. Live job +request-id deduplication alone is not sufficient for this contract. Entries have +no implicit expiry; retention must preserve receipt identities while retries are +possible. Core host consumption awaits upstream availability. diff --git a/crates/tinyagents-session/src/run_ledger/drivers/events.rs b/crates/tinyagents-session/src/run_ledger/drivers/events.rs index 46ff39def..eeac540bb 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/events.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/events.rs @@ -22,13 +22,13 @@ pub(super) fn head_after(highest: u64) -> Option { (highest < i64::MAX as u64).then_some(highest + 1) } -struct EventCommit { - event: RunEvent, - writes: Vec, +pub(super) struct EventCommit { + pub(super) event: RunEvent, + pub(super) writes: Vec, } impl DriverRunLedger { - async fn event_commit(&self, input: &RunEventAppend) -> Result { + pub(super) async fn event_commit(&self, input: &RunEventAppend) -> Result { let head = self.read::(EVENT_HEADS, &input.run_id).await?; let sequence = match &head { // A run migrated from v1 has events but no head (the v1 head's id diff --git a/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs b/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs new file mode 100644 index 000000000..b81a7952c --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs @@ -0,0 +1,193 @@ +//! Atomic durable input admission with persisted parent-lineage fencing. + +use super::dto::{from_ms, stored_as_is}; +use super::storage::{AGENTS, CAS_ATTEMPTS, INPUTS, key, now, put_op}; +use super::*; +use crate::run_ledger::inbox::{AddressedInput, QueuedInput, RunInboxStore}; +use serde::{Deserialize, Serialize}; +use std::collections::HashSet; +use tinyagents_storage_kit::{ErrorKind, ExpectedDocument, Precondition, ReadCheck}; + +#[derive(Clone, Serialize, Deserialize)] +struct StoredInput { + #[serde(flatten)] + input: AddressedInput, + sequence: u64, + created_ms: i64, +} +stored_as_is!(StoredInput); + +impl StoredInput { + fn receipt(self) -> Result { + Ok(QueuedInput { + input: self.input, + sequence: self.sequence, + created_at: from_ms(self.created_ms)?, + }) + } +} + +fn validate(input: &AddressedInput) -> Result<()> { + for id in [ + &input.submission_id, + &input.root_thread_id, + &input.recipient_run_id, + ] { + if id.trim().is_empty() || id.len() > 512 || id.contains('\0') { + return Err(StorageError::serialization("invalid addressed input identity").into()); + } + } + if input.text.trim().is_empty() || input.text.len() > 1024 * 1024 { + return Err(StorageError::serialization( + "addressed input must contain 1..=1048576 bytes of nonblank text", + ) + .into()); + } + Ok(()) +} + +impl DriverRunLedger { + // Walk exact persisted run links, never infer ownership from a display name + // or an in-memory worker. Every observed link is fenced in the admission. + async fn input_lineage(&self, root: &str, recipient: &str) -> Result> { + let mut current = recipient.to_owned(); + let mut visited = HashSet::new(); + let mut checks = Vec::new(); + for _ in 0..128 { + if !visited.insert(current.clone()) { + return Err(StorageError::serialization("cyclic input recipient lineage").into()); + } + let row = self + .read::(AGENTS, ¤t) + .await? + .ok_or_else(|| StorageError::serialization("input recipient lineage is missing"))?; + checks.push(ReadCheck { + collection: AGENTS.into(), + id: key(&[¤t])?, + expected: ExpectedDocument::Version(row.version), + }); + match row.doc.parent_run_id { + Some(parent) => current = parent, + None if row.doc.parent_thread_id.as_deref() == Some(root) => return Ok(checks), + None => { + return Err(StorageError::serialization( + "input recipient belongs to another root", + ) + .into()); + } + } + } + Err(StorageError::serialization("input recipient lineage exceeds 128 runs").into()) + } +} + +#[async_trait::async_trait] +impl RunInboxStore for DriverRunLedger { + async fn enqueue_input(&self, input: AddressedInput) -> Result { + validate(&input)?; + self.declare().await?; + let id = key(&[&input.root_thread_id, &input.submission_id])?; + for _ in 0..CAS_ATTEMPTS { + let mut checks = self + .input_lineage(&input.root_thread_id, &input.recipient_run_id) + .await?; + let previous = self.scoped.documents().get(INPUTS, &id).await?; + if let Some(previous) = previous { + let stored: StoredInput = super::storage::decode(previous.doc)?; + if stored.input != input { + return Err(StorageError::conflict( + "submission id already addresses different input", + ) + .into()); + } + checks.push(ReadCheck { + collection: INPUTS.into(), + id: id.clone(), + expected: ExpectedDocument::Version(previous.version), + }); + match self + .scoped + .documents() + .atomic_batch_checked(checks, vec![]) + .await + { + Ok(_) => return stored.receipt(), + Err(error) if error.kind() == ErrorKind::Conflict => continue, + Err(error) => return Err(error.into()), + } + } + let commit = self.event_commit(&RunEventAppend { + run_id: input.recipient_run_id.clone(), + event_type: "input_queued".into(), + payload: serde_json::json!({"submission_id": input.submission_id, "mode": input.mode}), + }).await?; + let stored = StoredInput { + input: input.clone(), + sequence: commit.event.sequence, + created_ms: now().timestamp_millis(), + }; + let mut writes = commit.writes; + writes.push(put_op(INPUTS, id.clone(), &stored, Precondition::Absent)?); + match self + .scoped + .documents() + .atomic_batch_checked(checks, writes) + .await + { + Ok(_) => return stored.receipt(), + Err(error) if error.kind() == ErrorKind::Conflict => continue, + Err(error) => return Err(error.into()), + } + } + Err(StorageError::conflict("input admission contention budget exhausted").into()) + } + + async fn pending_inputs( + &self, + root: &str, + recipient: &str, + limit: u32, + ) -> Result> { + if root.trim().is_empty() || recipient.trim().is_empty() { + return Err( + StorageError::serialization("input root and recipient are required").into(), + ); + } + for _ in 0..CAS_ATTEMPTS { + let checks = self.input_lineage(root, recipient).await?; + let page = self + .scoped + .documents() + .query( + INPUTS, + &Query::filter( + Filter::eq("root_thread_id", root) + .and(Filter::eq("recipient_run_id", recipient)), + ) + .sort(Sort::asc("sequence")) + .limit(limit.clamp(1, 500) as usize), + ) + .await?; + let inputs = page + .items + .into_iter() + .map(|row| super::storage::decode::(row.doc)?.receipt()) + .collect::>>()?; + match self + .scoped + .documents() + .atomic_batch_checked(checks, vec![]) + .await + { + Ok(_) => return Ok(inputs), + Err(error) if error.kind() == ErrorKind::Conflict => continue, + Err(error) => return Err(error.into()), + } + } + Err(StorageError::conflict("input read contention budget exhausted").into()) + } +} + +#[cfg(test)] +#[path = "inbox_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs b/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs new file mode 100644 index 000000000..5cdfdba41 --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs @@ -0,0 +1,120 @@ +use super::*; +use crate::run_ledger::inbox::InputMode; +use crate::testkit::MemoryStorage; + +fn ledger() -> DriverRunLedger { + DriverRunLedger::new( + MemoryStorage::new() + .for_scope(&Scope::new("inbox-tests").unwrap()) + .unwrap(), + ) +} + +fn run(id: &str, parent: Option<&str>, thread: Option<&str>) -> AgentRunUpsert { + AgentRunUpsert { + id: id.into(), + kind: AgentRunKind::Subagent, + parent_run_id: parent.map(str::to_owned), + parent_thread_id: thread.map(str::to_owned), + agent_id: None, + status: AgentRunStatus::AwaitingUser, + prompt_ref: None, + worker_thread_id: None, + checkpoint_path: None, + checkpoint: None, + summary: None, + error: None, + metadata: serde_json::json!({}), + started_at: None, + completed_at: None, + } +} + +fn input(id: &str) -> AddressedInput { + AddressedInput { + submission_id: id.into(), + root_thread_id: "root".into(), + recipient_run_id: "child".into(), + mode: InputMode::Answer, + text: "Keep this exact answer\nwith its line break.".into(), + } +} + +async fn prepare(ledger: &DriverRunLedger) { + ledger + .upsert_agent_run(run("parent", None, Some("root"))) + .await + .unwrap(); + ledger + .upsert_agent_run(run("child", Some("parent"), Some("worker-thread"))) + .await + .unwrap(); +} + +#[tokio::test] +async fn duplicate_admission_survives_handle_recreation_and_records_one_event() { + let ledger = ledger(); + prepare(&ledger).await; + let message = input("submission-1"); + let first = ledger.enqueue_input(message.clone()).await.unwrap(); + let reopened = DriverRunLedger::new(ledger.scoped.clone()); + assert_eq!( + reopened.enqueue_input(message.clone()).await.unwrap(), + first + ); + assert_eq!( + reopened.pending_inputs("root", "child", 20).await.unwrap(), + vec![first] + ); + let events = reopened + .list_recent_run_events(&RunEventListRequest { + run_id: "child".into(), + after_sequence: None, + limit: None, + }) + .await + .unwrap(); + assert_eq!(events.events.len(), 1); + assert_eq!(events.events[0].event_type, "input_queued"); + let mut changed = message; + changed.text = "Different content".into(); + assert!(reopened.enqueue_input(changed).await.is_err()); +} + +#[tokio::test] +async fn concurrent_duplicate_admissions_share_one_receipt_and_preserve_order() { + let ledger = ledger(); + prepare(&ledger).await; + let (left, right) = tokio::join!( + ledger.enqueue_input(input("same")), + ledger.enqueue_input(input("same")) + ); + assert_eq!(left.unwrap(), right.unwrap()); + ledger.enqueue_input(input("second")).await.unwrap(); + let queued = ledger.pending_inputs("root", "child", 20).await.unwrap(); + assert_eq!(queued.len(), 2); + assert!(queued[0].sequence < queued[1].sequence); + assert_eq!(queued[0].input.submission_id, "same"); + assert_eq!(queued[1].input.submission_id, "second"); +} + +#[tokio::test] +async fn cross_root_cyclic_missing_and_blank_inputs_are_rejected() { + let ledger = ledger(); + prepare(&ledger).await; + let mut wrong_root = input("wrong"); + wrong_root.root_thread_id = "other".into(); + assert!(ledger.enqueue_input(wrong_root).await.is_err()); + assert!(ledger.pending_inputs("other", "child", 20).await.is_err()); + let mut missing = input("missing"); + missing.recipient_run_id = "absent".into(); + assert!(ledger.enqueue_input(missing).await.is_err()); + let mut blank = input("blank"); + blank.text = " \n ".into(); + assert!(ledger.enqueue_input(blank).await.is_err()); + ledger + .upsert_agent_run(run("parent", Some("child"), Some("root"))) + .await + .unwrap(); + assert!(ledger.enqueue_input(input("cycle")).await.is_err()); +} diff --git a/crates/tinyagents-session/src/run_ledger/drivers/mod.rs b/crates/tinyagents-session/src/run_ledger/drivers/mod.rs index e48d6d0fa..c762653e3 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/mod.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/mod.rs @@ -12,6 +12,7 @@ mod events; mod history; mod history_retention; mod imports; +mod inbox; mod legacy_state; mod retention; mod schema; diff --git a/crates/tinyagents-session/src/run_ledger/drivers/schema.rs b/crates/tinyagents-session/src/run_ledger/drivers/schema.rs index 8da5b0e9e..4c68d8657 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/schema.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/schema.rs @@ -22,7 +22,7 @@ //! and the next sequence of a migrated run is recomputed from its events. use super::storage::{ - EVENT_STREAM, HISTORY_OWNER, LEDGER_OWNER, SCHEMA_VERSION, history_specs, ledger_specs, + EVENT_STREAM, HISTORY_OWNER, INPUTS, LEDGER_OWNER, SCHEMA_VERSION, history_specs, ledger_specs, }; use tinyagents_storage_kit::{Owner, Schema, ScopeClass}; @@ -33,7 +33,14 @@ pub fn ledger_storage_schema() -> Schema { let owner = Owner::new(LEDGER_OWNER).expect("`ledger` is a valid owner"); let mut schema = Schema::new(owner); for spec in ledger_specs() { - schema = schema.collection_with(SCHEMA_VERSION, spec, ScopeClass::Agent); + // Inputs are new, with no legacy v1 envelope to migrate. Their initial + // schema is version one; declaring v2 would demand a nonexistent step. + let version = if spec.name == INPUTS { + 1 + } else { + SCHEMA_VERSION + }; + schema = schema.collection_with(version, spec, ScopeClass::Agent); } schema = schema.stream(EVENT_STREAM, ScopeClass::Agent); for step in super::steps::steps() { diff --git a/crates/tinyagents-session/src/run_ledger/drivers/schema_tests.rs b/crates/tinyagents-session/src/run_ledger/drivers/schema_tests.rs index b9463e5ee..6f5b4ae10 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/schema_tests.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/schema_tests.rs @@ -100,6 +100,7 @@ fn the_schemas_name_what_each_owner_declares() { "ledger.event_heads", "ledger.event_outbox", "ledger.events", + "ledger.inputs", "ledger.members", "ledger.tasks", "ledger.teams", diff --git a/crates/tinyagents-session/src/run_ledger/drivers/storage.rs b/crates/tinyagents-session/src/run_ledger/drivers/storage.rs index 2a9d803b6..3da13c187 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/storage.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/storage.rs @@ -33,6 +33,7 @@ pub(super) const EFFECTS: &str = "ledger.tool_effects"; pub(super) const EVENTS: &str = "ledger.event_outbox"; pub(super) const EVENT_HEADS: &str = "ledger.event_heads"; pub(super) const TASKS: &str = "ledger.tasks"; +pub(super) const INPUTS: &str = "ledger.inputs"; /// The per-run event stream family: `ledger.events/`. pub(super) const EVENT_STREAM: &str = "ledger.events"; pub(super) const HIST_SESSIONS: &str = "history.sessions"; @@ -81,6 +82,9 @@ pub(super) fn eq_nonblank(filter: Filter, field: &str, value: Option<&str>) -> F /// metadata, so none declares a TTL. pub(super) fn ledger_specs() -> Vec { vec![ + CollectionSpec::new(INPUTS) + .index(IndexSpec::new("recipient_sequence", ["recipient_run_id", "sequence"]).unique()) + .index(IndexSpec::new("root", ["root_thread_id"])), CollectionSpec::new(EVENT_HEADS), CollectionSpec::new(EVENTS) .index(IndexSpec::new("run_sequence", ["run_id", "sequence"]).unique()), diff --git a/crates/tinyagents-session/src/run_ledger/inbox.rs b/crates/tinyagents-session/src/run_ledger/inbox.rs new file mode 100644 index 000000000..7a582b01f --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/inbox.rs @@ -0,0 +1,63 @@ +//! Durable addressed input admission, independent of a live worker registry. +//! +//! A receipt means the input is durably queued, not that a model has read it. +//! Recipients must commit input to their own durable session before acknowledging +//! delivery. This admission API deliberately exposes no ledger-only "delivered" +//! mutation. A host must not substitute live steering for durable admission. + +use serde::{Deserialize, Serialize}; + +/// Intended delivery boundary in the recipient's existing session. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InputMode { + /// Admit at the next safe boundary of running work. + Steer, + /// Queue another turn after the current work. + FollowUp, + /// Answer the recipient's pending question. + Answer, +} + +/// Operator input addressed to one exact durable run under a root conversation. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AddressedInput { + /// Unique within the root conversation; retry the same payload with this id. + pub submission_id: String, + /// Root conversation whose persisted lineage authorizes the target. + pub root_thread_id: String, + /// Exact run-ledger identity, never an agent definition or display name. + pub recipient_run_id: String, + /// Required recipient execution boundary. + pub mode: InputMode, + /// Original user text; not trimmed or rewritten after admission. + pub text: String, +} + +/// Durable queued input. Repeated admission returns the original receipt. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct QueuedInput { + /// Original addressed input. + pub input: AddressedInput, + /// Recipient event sequence allocated atomically with admission. + pub sequence: u64, + /// Admission time at storage millisecond precision. + pub created_at: chrono::DateTime, +} + +/// Optional durable inbox capability of an already scoped ledger handle. +#[async_trait::async_trait] +pub trait RunInboxStore: Send + Sync { + /// Validate persisted root lineage and atomically insert input plus receipt. + /// Reusing an id for different content is an error, never another delivery. + async fn enqueue_input(&self, input: AddressedInput) -> super::LedgerResult; + + /// Read queued input in admission order after validating root ownership. + /// Limit is clamped to 1..=500; this never consumes or acknowledges input. + async fn pending_inputs( + &self, + root_thread_id: &str, + recipient_run_id: &str, + limit: u32, + ) -> super::LedgerResult>; +} diff --git a/crates/tinyagents-session/src/run_ledger/mod.rs b/crates/tinyagents-session/src/run_ledger/mod.rs index b00e20132..8113e5df0 100644 --- a/crates/tinyagents-session/src/run_ledger/mod.rs +++ b/crates/tinyagents-session/src/run_ledger/mod.rs @@ -14,6 +14,7 @@ pub mod drivers; pub mod error; pub mod history; pub mod import; +pub mod inbox; pub mod port; pub mod recovery; mod status_map; From f4e620d663ad0ceb96a47d12e9a5a8daa7115036 Mon Sep 17 00:00:00 2001 From: Farnam Date: Fri, 23 Oct 2026 14:16:59 +0330 Subject: [PATCH 3/4] Gate the Unix-only SDK test import on Unix --- .../src/providers/claude_agent_sdk/mod_tests.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/tinyagents-harness/src/providers/claude_agent_sdk/mod_tests.rs b/crates/tinyagents-harness/src/providers/claude_agent_sdk/mod_tests.rs index f1baca185..c29dd3447 100644 --- a/crates/tinyagents-harness/src/providers/claude_agent_sdk/mod_tests.rs +++ b/crates/tinyagents-harness/src/providers/claude_agent_sdk/mod_tests.rs @@ -3,6 +3,7 @@ //! assembly (including the timeout and error-propagation paths). use super::*; +#[cfg(unix)] use tinyinference_llm::tool::ToolCall; #[test] From e575352d1d9241ca3971eab8d0ff8725d2eedba8 Mon Sep 17 00:00:00 2001 From: Farnam Date: Fri, 23 Oct 2026 15:35:00 +0330 Subject: [PATCH 4/4] Page durable run input by exclusive sequence cursor --- .../src/run_ledger/command_center/README.md | 6 ++ .../src/run_ledger/drivers/events.rs | 1 + .../src/run_ledger/drivers/inbox.rs | 23 +++--- .../src/run_ledger/drivers/inbox_tests.rs | 71 ++++++++++++++++++- .../src/run_ledger/inbox.rs | 2 + 5 files changed, 92 insertions(+), 11 deletions(-) diff --git a/crates/tinyagents-session/src/run_ledger/command_center/README.md b/crates/tinyagents-session/src/run_ledger/command_center/README.md index 5efff064d..de762c8cd 100644 --- a/crates/tinyagents-session/src/run_ledger/command_center/README.md +++ b/crates/tinyagents-session/src/run_ledger/command_center/README.md @@ -47,6 +47,12 @@ event in one checked batch. Reusing the root/submission id for different content returns a conflict. Receipts survive process restart and list in event-sequence order. A backend without checked atomic batches returns its native error. +`pending_inputs(root, recipient, after_sequence, limit)` returns at most 500 +records in ascending sequence order. Start with `None`, then pass the last +receipt's sequence as `Some(sequence)` to read the next page. The cursor is +exclusive; paging never consumes or acknowledges input. Each page validates +the recipient's persisted root lineage, including empty pages. + This is an admission draft, not a complete delivery implementation. No operation marks input delivered, changes a run to running, or launches a worker. Recipient integration must atomically commit admitted input into the durable session and diff --git a/crates/tinyagents-session/src/run_ledger/drivers/events.rs b/crates/tinyagents-session/src/run_ledger/drivers/events.rs index eeac540bb..bd6899d31 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/events.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/events.rs @@ -28,6 +28,7 @@ pub(super) struct EventCommit { } impl DriverRunLedger { + /// Prepare an event and head update for the caller's atomic document batch. pub(super) async fn event_commit(&self, input: &RunEventAppend) -> Result { let head = self.read::(EVENT_HEADS, &input.run_id).await?; let sequence = match &head { diff --git a/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs b/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs index b81a7952c..0f8eca004 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/inbox.rs @@ -18,6 +18,7 @@ struct StoredInput { stored_as_is!(StoredInput); impl StoredInput { + /// Convert the persisted admission into its stable public receipt. fn receipt(self) -> Result { Ok(QueuedInput { input: self.input, @@ -27,6 +28,7 @@ impl StoredInput { } } +/// Reject invalid identities and empty or oversized input before storage access. fn validate(input: &AddressedInput) -> Result<()> { for id in [ &input.submission_id, @@ -47,8 +49,8 @@ fn validate(input: &AddressedInput) -> Result<()> { } impl DriverRunLedger { - // Walk exact persisted run links, never infer ownership from a display name - // or an in-memory worker. Every observed link is fenced in the admission. + /// Walk exact persisted run links and fence every observed lineage version. + /// Display names and the live worker registry cannot establish ownership. async fn input_lineage(&self, root: &str, recipient: &str) -> Result> { let mut current = recipient.to_owned(); let mut visited = HashSet::new(); @@ -83,6 +85,7 @@ impl DriverRunLedger { #[async_trait::async_trait] impl RunInboxStore for DriverRunLedger { + /// Atomically admit an input and event, or return its original dedup receipt. async fn enqueue_input(&self, input: AddressedInput) -> Result { validate(&input)?; self.declare().await?; @@ -142,10 +145,12 @@ impl RunInboxStore for DriverRunLedger { Err(StorageError::conflict("input admission contention budget exhausted").into()) } + /// Read an exclusive sequence page while rechecking persisted root ownership. async fn pending_inputs( &self, root: &str, recipient: &str, + after_sequence: Option, limit: u32, ) -> Result> { if root.trim().is_empty() || recipient.trim().is_empty() { @@ -155,17 +160,19 @@ impl RunInboxStore for DriverRunLedger { } for _ in 0..CAS_ATTEMPTS { let checks = self.input_lineage(root, recipient).await?; + let mut filter = + Filter::eq("root_thread_id", root).and(Filter::eq("recipient_run_id", recipient)); + if let Some(cursor) = after_sequence { + filter = filter.and(Filter::gt("sequence", cursor)); + } let page = self .scoped .documents() .query( INPUTS, - &Query::filter( - Filter::eq("root_thread_id", root) - .and(Filter::eq("recipient_run_id", recipient)), - ) - .sort(Sort::asc("sequence")) - .limit(limit.clamp(1, 500) as usize), + &Query::filter(filter) + .sort(Sort::asc("sequence")) + .limit(limit.clamp(1, 500) as usize), ) .await?; let inputs = page diff --git a/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs b/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs index 5cdfdba41..f3d348075 100644 --- a/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs +++ b/crates/tinyagents-session/src/run_ledger/drivers/inbox_tests.rs @@ -2,6 +2,54 @@ use super::*; use crate::run_ledger::inbox::InputMode; use crate::testkit::MemoryStorage; +/// Read every admitted input across the maximum page boundary without consuming it. +#[tokio::test] +async fn pending_inputs_pages_past_five_hundred_with_exclusive_cursor() { + let ledger = ledger(); + prepare(&ledger).await; + let mut receipts = Vec::new(); + for index in 0..503 { + receipts.push( + ledger + .enqueue_input(input(&format!("page-{index}"))) + .await + .unwrap(), + ); + } + let first = ledger + .pending_inputs("root", "child", None, u32::MAX) + .await + .unwrap(); + assert_eq!(first, receipts[..500]); + let cursor = first.last().unwrap().sequence; + let rest = ledger + .pending_inputs("root", "child", Some(cursor), 500) + .await + .unwrap(); + assert_eq!(rest, receipts[500..]); + assert_eq!( + ledger + .pending_inputs("root", "child", Some(cursor), 0) + .await + .unwrap(), + receipts[500..501] + ); + assert!( + ledger + .pending_inputs("root", "child", Some(u64::MAX), 500) + .await + .unwrap() + .is_empty() + ); + assert!( + ledger + .pending_inputs("other", "child", Some(u64::MAX), 500) + .await + .is_err() + ); +} + +/// Create an isolated in-memory ledger for admission tests. fn ledger() -> DriverRunLedger { DriverRunLedger::new( MemoryStorage::new() @@ -10,6 +58,7 @@ fn ledger() -> DriverRunLedger { ) } +/// Construct an awaiting run with explicit persisted ancestry. fn run(id: &str, parent: Option<&str>, thread: Option<&str>) -> AgentRunUpsert { AgentRunUpsert { id: id.into(), @@ -30,6 +79,7 @@ fn run(id: &str, parent: Option<&str>, thread: Option<&str>) -> AgentRunUpsert { } } +/// Address an exact text payload to the prepared child run. fn input(id: &str) -> AddressedInput { AddressedInput { submission_id: id.into(), @@ -40,6 +90,7 @@ fn input(id: &str) -> AddressedInput { } } +/// Persist the root-to-parent-to-child lineage used by admission tests. async fn prepare(ledger: &DriverRunLedger) { ledger .upsert_agent_run(run("parent", None, Some("root"))) @@ -52,6 +103,7 @@ async fn prepare(ledger: &DriverRunLedger) { } #[tokio::test] +/// Reopening a ledger preserves the original receipt and rejects id reuse. async fn duplicate_admission_survives_handle_recreation_and_records_one_event() { let ledger = ledger(); prepare(&ledger).await; @@ -63,7 +115,10 @@ async fn duplicate_admission_survives_handle_recreation_and_records_one_event() first ); assert_eq!( - reopened.pending_inputs("root", "child", 20).await.unwrap(), + reopened + .pending_inputs("root", "child", None, 20) + .await + .unwrap(), vec![first] ); let events = reopened @@ -82,6 +137,7 @@ async fn duplicate_admission_survives_handle_recreation_and_records_one_event() } #[tokio::test] +/// Concurrent retries commit once and later admissions remain ordered. async fn concurrent_duplicate_admissions_share_one_receipt_and_preserve_order() { let ledger = ledger(); prepare(&ledger).await; @@ -91,7 +147,10 @@ async fn concurrent_duplicate_admissions_share_one_receipt_and_preserve_order() ); assert_eq!(left.unwrap(), right.unwrap()); ledger.enqueue_input(input("second")).await.unwrap(); - let queued = ledger.pending_inputs("root", "child", 20).await.unwrap(); + let queued = ledger + .pending_inputs("root", "child", None, 20) + .await + .unwrap(); assert_eq!(queued.len(), 2); assert!(queued[0].sequence < queued[1].sequence); assert_eq!(queued[0].input.submission_id, "same"); @@ -99,13 +158,19 @@ async fn concurrent_duplicate_admissions_share_one_receipt_and_preserve_order() } #[tokio::test] +/// Malformed payloads and invalid ancestry cannot enter the durable inbox. async fn cross_root_cyclic_missing_and_blank_inputs_are_rejected() { let ledger = ledger(); prepare(&ledger).await; let mut wrong_root = input("wrong"); wrong_root.root_thread_id = "other".into(); assert!(ledger.enqueue_input(wrong_root).await.is_err()); - assert!(ledger.pending_inputs("other", "child", 20).await.is_err()); + assert!( + ledger + .pending_inputs("other", "child", None, 20) + .await + .is_err() + ); let mut missing = input("missing"); missing.recipient_run_id = "absent".into(); assert!(ledger.enqueue_input(missing).await.is_err()); diff --git a/crates/tinyagents-session/src/run_ledger/inbox.rs b/crates/tinyagents-session/src/run_ledger/inbox.rs index 7a582b01f..5fee65fec 100644 --- a/crates/tinyagents-session/src/run_ledger/inbox.rs +++ b/crates/tinyagents-session/src/run_ledger/inbox.rs @@ -53,11 +53,13 @@ pub trait RunInboxStore: Send + Sync { async fn enqueue_input(&self, input: AddressedInput) -> super::LedgerResult; /// Read queued input in admission order after validating root ownership. + /// A cursor excludes sequences at or below its value, allowing bounded paging. /// Limit is clamped to 1..=500; this never consumes or acknowledges input. async fn pending_inputs( &self, root_thread_id: &str, recipient_run_id: &str, + after_sequence: Option, limit: u32, ) -> super::LedgerResult>; }