Add coordinator host seams over async, incremental storage

crates/tinyhivemind-hives/README.md

@@ -5,7 +5,11 @@ a desk have one identity. Each registered agent has one continuing session and
 one globally serialized turn stream, even when it belongs to several hives.
 
 Use `Coordinator` with `MemoryStorage`, default-feature `SqliteStorage`, or a
-host implementation of `Storage`. The host supplies `AgentRunner` handles from
+host implementation of the async `Storage` port. A commit writes one bounded
+state row and appends only the new transcript rows; `RetentionPolicy` bounds
+settled episodes and acknowledged deliveries. Mutating APIs are `async` and
+executor-neutral; commits run outside the live lock and reload on a conflict
+with another process. The host supplies `AgentRunner` handles from
 one runtime; this crate never constructs agents or serializes live handles.
 The OpenHuman-specific boundary lives in `tinyhivemind-openhuman`.
 
@@ -15,6 +19,12 @@ claims it. Hosts can create empty hives and join/leave registered agents while
 the scheduler runs; unstarted removed seats are retired before claiming.
 An active turn retains its captured membership.
 
+The host reads everything with `read_transcript`, including replies to
+`send_as_host`, and watches `subscribe()` (the committed revision) and
+`episodes()` for settlement. `SendMessage::starters` chooses who starts an
+episode without hiding the message, and `release_with` hands a parked agent
+a note on its next turn.
+
 Direct sends return durable receipts without awaiting peers. `read_direct`
 exposes only the caller/peer pair, including returned replies, and starts no
 turn. Hive reads enforce membership and private thread audiences.

crates/tinyhivemind-hives/src/coordinator/README.md

@@ -2,17 +2,29 @@
 
 | File | Responsibility |
 | --- | --- |
-| `mod.rs` | Shared handle, dynamic APIs, authorization snapshots, and transactions |
+| `mod.rs` | Shared handle, dynamic APIs, and authorization snapshots |
+| `transaction.rs` | Writer gate, incremental commits outside the live lock, conflict reload |
 | `types.rs` | Host runner port and public payloads |
-| `messaging.rs` | Atomic acceptance, retry IDs, attribution, and private reads |
+| `messaging.rs` | Atomic acceptance, retry IDs, starters, attribution, and private reads |
+| `observe.rs` | Host transcript, committed-revision watch, episode status |
 | `conduct.rs` | Actual `CompletionDriver`/`Conductor` checkpoint lifecycle |
 | `scheduler.rs` | FIFO reservations, concurrent futures, shutdown, cancellation |
 | `test/` | Deterministic contract fixtures and behavior tests |
 
 All clones of `Coordinator` share one scheduler and state. Transactions clone
-state and replace it only after the storage CAS succeeds. Runner futures execute
-outside the state lock. Conductor folding uses copied snapshots and retries if a
-concurrent synchronous API changed the revision; no external tool is repeated.
+state under the live lock, release it, and persist the copy under an async
+writer gate. The storage commit carries the state row plus only the transcript
+rows appended since the base, and the copy is published only after the CAS
+succeeds. Reads never wait on storage. A revision conflict, which means another
+process wrote the store, reloads and recomputes up to four times. Runner
+futures execute outside every lock. A dropped drain applies its interruptions
+to live state immediately, and the next commit persists them. No external tool
+is repeated.
+
+`release_with(agent, note)` stores a note on the agent record. The next claim
+moves it into `TurnRequest::resumption`, once. Host-chosen `starters` must be
+distinct readers of the hive message; they open the episode while every reader
+still sees the message.
 
 Work is ordered by accepted sequence, then agent ID within a round. Shared
 agents retain their queued positions and never run two turns concurrently.

crates/tinyhivemind-hives/src/coordinator/conduct.rs

@@ -253,13 +253,19 @@ pub(super) async fn prepare(state: &mut StoredState, options: &CoordinatorOption
                 }
                 // An empty parked wave must wait for an explicit release.
                 episode.waiting = episode.pending.is_empty() && !conductor.parked().is_empty();
+                checkpoint(&conductor, &mut episode)?;
             }
             Err(error) => {
+                // The conductor still reports itself unfinished after a stall,
+                // so settle after checkpointing; otherwise the episode is
+                // re-prepared, and fails again, on every pass.
+                checkpoint(&conductor, &mut episode)?;
                 episode.finished = true;
                 episode.failure = Some(error.to_string());
+                episode.pending.clear();
+                episode.wave_open = false;
             }
         }
-        checkpoint(&conductor, &mut episode)?;
         state.episodes[index] = episode;
     }
     Ok(())

crates/tinyhivemind-hives/src/coordinator/messaging.rs

@@ -9,25 +9,26 @@ impl Coordinator {
     /// # Errors
     /// Returns unknown sender/destination, missing membership, invalid thread,
     /// conflicting retry identity or persistence errors.
-    pub fn send(&self, request: SendMessage) -> Result<Receipt> {
+    pub async fn send(&self, request: SendMessage) -> Result<Receipt> {
         if request.sender == HOST_ID {
             return Err(Error::InvalidIdentifier("sender"));
         }
-        self.accept(request)
+        self.accept(request).await
     }
     /// Submit through the reserved host identity rather than impersonating an agent.
     /// The request's sender field is overwritten.
     /// # Errors
     /// Returns destination, visibility, retry or storage validation errors.
-    pub fn send_as_host(&self, mut request: SendMessage) -> Result<Receipt> {
+    pub async fn send_as_host(&self, mut request: SendMessage) -> Result<Receipt> {
         request.sender = HOST_ID.into();
-        self.accept(request)
+        self.accept(request).await
     }
-    fn accept(&self, request: SendMessage) -> Result<Receipt> {
+    async fn accept(&self, request: SendMessage) -> Result<Receipt> {
         identifier(&request.message_id, "message id")?;
         if request.message_id.starts_with("hivemind:") {
             return Err(Error::InvalidIdentifier("reserved message id"));
         }
+        let retention = self.inner.options.retention;
         self.update(|state| {
             if let Some(old) = state.accepted.get(&request.message_id) {
                 if old != &request {
@@ -67,6 +68,9 @@ impl Coordinator {
             };
             match &request.destination {
                 Destination::Agent(_) => {
+                    for agent_id in &recipients {
+                        retention.admit_pending(state, agent_id)?;
+                    }
                     for agent_id in recipients {
                         state.deliveries.push(Delivery {
                             sequence,
@@ -94,7 +98,11 @@ impl Coordinator {
                         hive,
                         opened_at: sequence,
                         thread: request.thread,
-                        starters: recipients,
+                        starters: if request.starters.is_empty() {
+                            recipients.clone()
+                        } else {
+                            request.starters.clone()
+                        },
                         conductor: None,
                         pending: Vec::new(),
                         wave_open: false,
@@ -109,10 +117,11 @@ impl Coordinator {
                 .accepted
                 .insert(request.message_id.clone(), request.clone());
             Ok(Receipt {
-                message_id: request.message_id,
+                message_id: request.message_id.clone(),
                 sequence,
             })
         })
+        .await
     }
     /// Read the caller's direct conversation with a registered peer in durable
     /// sequence order, including replies returned by the peer's runner.
@@ -220,7 +229,10 @@ fn recipients(state: &StoredState, request: &SendMessage) -> Result<Vec<String>>
     Ok(match &request.destination {
         Destination::Agent(id) => {
             known_agent(state, id)?;
-            if request.thread.is_some() || !request.only_for.is_empty() {
+            if request.thread.is_some()
+                || !request.only_for.is_empty()
+                || !request.starters.is_empty()
+            {
                 return Err(Error::InvalidIdentifier("direct message attribution"));
             }
             vec![id.clone()]
@@ -268,13 +280,31 @@ fn recipients(state: &StoredState, request: &SendMessage) -> Result<Vec<String>>
             {
                 return Err(Error::InvalidThread(request.thread.unwrap_or_default()));
             }
-            selected
+            let selected: Vec<_> = selected
                 .into_iter()
                 .filter(|recipient| scope.is_empty() || scope.contains(recipient))
-                .collect()
+                .collect();
+            starters_among(&request.starters, &selected, id)?;
+            selected
         }
     })
 }
+/// Host-chosen starters must be distinct readers of the message.
+fn starters_among(starters: &[String], readers: &[String], hive_id: &str) -> Result<()> {
+    let mut unique = BTreeSet::new();
+    for starter in starters {
+        if !readers.contains(starter) {
+            return Err(Error::NotMember {
+                agent_id: starter.clone(),
+                hive_id: hive_id.into(),
+            });
+        }
+        if !unique.insert(starter) {
+            return Err(Error::DuplicateMember(starter.clone()));
+        }
+    }
+    Ok(())
+}
 
 /// Root-private threads can only be read by their original participants.
 pub(super) fn visible_in(state: &StoredState, message: &Message, agent_id: &str) -> bool {