Add atomic queue edits and durable addressed run-input admission

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]

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

crates/tinyagents-harness/src/run_queue/mod.rs

@@ -57,8 +57,7 @@ pub struct RunQueue<T> {
     inner: Mutex<RunQueueInner<T>>,
 }
 
-/// 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<T> {
     /// Immediate steering messages, consumed as soon as the loop reaches a
@@ -82,9 +81,9 @@ impl<T> RunQueue<T> {
         }
     }
 
-    /// 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<T>> {
         self.inner.lock().unwrap_or_else(PoisonError::into_inner)
     }
@@ -260,6 +259,76 @@ impl<T> RunQueue<T> {
         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<T> Default for RunQueue<T> {

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

crates/tinyagents-session/src/run_ledger/command_center/README.md

@@ -38,3 +38,25 @@ 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.
+
+`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
+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.

crates/tinyagents-session/src/run_ledger/drivers/events.rs

@@ -22,13 +22,14 @@ pub(super) fn head_after(highest: u64) -> Option<u64> {
     (highest < i64::MAX as u64).then_some(highest + 1)
 }
 
-struct EventCommit {
-    event: RunEvent,
-    writes: Vec<WriteOp>,
+pub(super) struct EventCommit {
+    pub(super) event: RunEvent,
+    pub(super) writes: Vec<WriteOp>,
 }
 
 impl DriverRunLedger {
-    async fn event_commit(&self, input: &RunEventAppend) -> Result<EventCommit> {
+    /// Prepare an event and head update for the caller's atomic document batch.
+    pub(super) async fn event_commit(&self, input: &RunEventAppend) -> Result<EventCommit> {
         let head = self.read::<EventHead>(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