feat(control) 5/5: a person in a session no longer fails a dispatch

src/sdk/src/daemon/embedded/types.rs

@@ -14,8 +14,20 @@ use super::super::types::{DaemonRuntime, LogFn};
 /// local bus is an in-memory queue: there is no relay to be polite to, and this
 /// interval is the floor on how quickly a locally-dispatched task starts.
 pub const DEFAULT_LOCAL_POLL: Duration = Duration::from_millis(150);
-/// Default concurrent task executions for an embedded host.
-pub const DEFAULT_CONCURRENCY: usize = 2;
+/// Default concurrent task executions for an embedded host — effectively
+/// unlimited.
+///
+/// A host-wide cap predates declared agents: it made sense when a machine was
+/// one worker with one implicit session. It is the wrong grain now, and it
+/// queued invisibly — a third concurrent task waited for a slot even when it
+/// targeted a different agent in a different workspace, where nothing could
+/// collide. The limits that own the real hazard live where the hazard is:
+/// per-agent `max_sessions` from the workspace strategy, and the checkout
+/// serialization that keeps a second writer out of a tree someone is in.
+///
+/// The semaphore stays as the accounting mechanism behind `active_count`, and
+/// an operator can still set `concurrency` to impose a real cap.
+pub const DEFAULT_CONCURRENCY: usize = 1024;
 /// Default per-task execution timeout, in ms.
 pub const DEFAULT_TASK_TIMEOUT_MS: u64 = 600_000;
 

src/sdk/src/daemon/entry.rs

@@ -24,7 +24,24 @@ use super::types::{
     DaemonConfig, DaemonRuntime, SendFn, DEFAULT_MAX_PENDING, DEFAULT_STATUS_THROTTLE_MS,
 };
 
-const DEFAULT_CONCURRENCY: usize = 2;
+/// Host-wide task slots, effectively unlimited by default.
+///
+/// This cap predates declared agents: when a machine was one worker with one
+/// implicit session, a small number was the only thing standing between a
+/// fan-out and a thrashed laptop. It is the wrong instrument now, and it queued
+/// invisibly — a third concurrent task waited on a slot even when it targeted a
+/// different agent in a different workspace, where nothing could collide.
+///
+/// What actually bounds concurrency now sits at the grain that owns the hazard:
+/// per-agent `max_sessions` derived from the agent's workspace strategy, and the
+/// checkout serialization that keeps a second writer out of a tree someone is
+/// already working in. A host-wide count knows about neither, so it can only
+/// delay work that was already safe.
+///
+/// The semaphore itself stays — `active_count` derives the running-task figure
+/// from its permits — and an operator can still set `concurrency` in config to
+/// impose a real cap on a small machine.
+const DEFAULT_CONCURRENCY: usize = 1024;
 const DEFAULT_TASK_TIMEOUT_MS: u64 = 600_000;
 const DEFAULT_POLL_MS: u64 = 2_000;
 

src/sdk/src/daemon/mod.rs

@@ -44,5 +44,5 @@ pub(crate) use status::TOOL_CALL_ID_SEPARATOR;
 pub use status::{status_detail, work_detail, THINKING_PREFIX, TOOL_PREFIX};
 pub use types::{
     DaemonConfig, DaemonRuntime, LogFn, NowFn, SendFn, CAPACITY_REJECTION_PREFIX,
-    HARNESS_HELD_PREFIX,
+    HARNESS_HELD_PREFIX, SESSION_HELD_STATUS_PREFIX, SESSION_RESUMED_STATUS_PREFIX,
 };

src/sdk/src/daemon/task_loop/run.rs

@@ -12,11 +12,19 @@ use super::super::providers::{Abort, RunTaskOptions};
 use super::super::status::{status_detail, work_detail};
 use super::super::types::{
     DaemonRuntime, FrameAttachments, RunningTask, CAPACITY_REJECTION_PREFIX,
+    SESSION_HELD_STATUS_PREFIX, SESSION_RESUMED_STATUS_PREFIX,
 };
 
 /// What a task waiting for a harness slot reports while it waits.
 const QUEUED_STATUS: &str = "queued for a harness slot";
 
+/// Whether a status detail is one of the two control markers the requester's
+/// watchdog reads (see [`SESSION_HELD_STATUS_PREFIX`]).
+fn is_control_marker(detail: &str) -> bool {
+    detail.starts_with(SESSION_HELD_STATUS_PREFIX)
+        || detail.starts_with(SESSION_RESUMED_STATUS_PREFIX)
+}
+
 /// What a running task reports through a stretch with no harness events —
 /// a single long tool call, typically.
 const HEARTBEAT_STATUS: &str = "still working";
@@ -279,7 +287,17 @@ impl DaemonRuntime {
                     semantic.event.decoded(),
                     HarnessEventKind::ToolCall(_) | HarnessEventKind::ToolResult(_)
                 );
-                if !changes_tool && current.saturating_sub(last_status_at) < throttle {
+                // Control changes are state transitions too, and the only ones
+                // the *requester* acts on: the hold marker is what pauses its
+                // no-progress watchdog, and the hand-back marker is what
+                // resumes it. Throttled away, a hold that began a second after
+                // the last tool call would never be announced, and the hub
+                // would reap a task a person is sitting in.
+                let changes_control = is_control_marker(&detail);
+                if !changes_tool
+                    && !changes_control
+                    && current.saturating_sub(last_status_at) < throttle
+                {
                     if is_thinking {
                         *pending_thinking.lock().unwrap() = Some((detail, Some(snapshot)));
                     }

src/sdk/src/daemon/tests/capability_tests.rs

@@ -11,8 +11,8 @@ use crate::protocol::{AgentCapabilities, HarnessEvent, TaskFrameKind};
 
 use super::{
     base_config, capabilities_frame, chatter_status_runner, counting_capability_runner,
-    decoded_frames, quick_thinking_runner, quick_tool_runner, recording_send, status_runner,
-    task_frame, tool_call_event,
+    decoded_frames, held_then_resumed_runner, quick_thinking_runner, quick_tool_runner,
+    recording_send, status_runner, task_frame, tool_call_event,
 };
 
 #[tokio::test]
@@ -52,6 +52,53 @@ async fn throttles_status_frames() {
         .any(|f| f.kind == TaskFrameKind::Reply && f.text == "ok"));
 }
 
+#[tokio::test]
+async fn control_markers_are_never_throttled_away() {
+    // The throttle is a rate cap on *chatter*, and a hold is not chatter: it is
+    // the frame that pauses the requester's no-progress watchdog, and the
+    // hand-back is the one that resumes it. Dropped by the throttle — which is
+    // exactly what happens when a person takes a session a second after the last
+    // status — the hub would go on counting the silence of a session somebody is
+    // sitting in, and reap a healthy task while they worked.
+    let (send, recorded) = recording_send();
+    let runtime = DaemonRuntime::new(base_config(), held_then_resumed_runner(), send);
+    // All three events inside one 4s window: ordinarily only the first survives
+    // (see `throttles_status_frames`, which is the same clock).
+    let seq = Arc::new(vec![10_000i64, 11_000, 12_000]);
+    let index = Arc::new(AtomicUsize::new(0));
+    let now: NowFn = Arc::new(move || {
+        let position = index.fetch_add(1, Ordering::SeqCst);
+        *seq.get(position).unwrap_or(seq.last().unwrap())
+    });
+    let runtime = runtime.with_now(now);
+
+    runtime.handle_message(
+        "peer".into(),
+        String::new(),
+        Some(task_frame("t1", "work", None)),
+    );
+    runtime.idle().await;
+
+    let frames = decoded_frames(&recorded);
+    let statuses: Vec<&str> = frames
+        .iter()
+        .filter(|f| f.kind == TaskFrameKind::Status)
+        .map(|f| f.text.as_str())
+        .collect();
+    assert!(
+        statuses
+            .iter()
+            .any(|text| text.starts_with(crate::daemon::SESSION_HELD_STATUS_PREFIX)),
+        "the hold must reach the requester: {statuses:?}"
+    );
+    assert!(
+        statuses
+            .iter()
+            .any(|text| text.starts_with(crate::daemon::SESSION_RESUMED_STATUS_PREFIX)),
+        "and so must the hand-back, or the watchdog never resumes: {statuses:?}"
+    );
+}
+
 #[tokio::test]
 async fn flushes_final_thinking_snapshot_after_throttling() {
     let (send, recorded) = recording_send();

src/sdk/src/daemon/tests/mod.rs

@@ -278,6 +278,51 @@ pub(super) fn chatter_status_runner(count: usize) -> RunTaskFn {
     })
 }
 
+/// A runner whose second and third statuses are the control markers a held
+/// session emits, all three inside one throttle window.
+///
+/// The ordinary status first, so the throttle's clock is already primed when the
+/// hold is announced — which is the case that matters: a person taking a session
+/// a second after the last tool call must not have the hold silently dropped.
+pub(super) fn held_then_resumed_runner() -> RunTaskFn {
+    Arc::new(move |mut opts: RunTaskOptions| {
+        Box::pin(async move {
+            if let Some(mut on_event) = opts.on_event.take() {
+                for (state, detail) in [
+                    ("working", "reading the migration".to_string()),
+                    (
+                        "held",
+                        crate::daemon::SESSION_HELD_STATUS_PREFIX.to_string(),
+                    ),
+                    (
+                        "running",
+                        crate::daemon::SESSION_RESUMED_STATUS_PREFIX.to_string(),
+                    ),
+                ] {
+                    on_event(&HarnessSemanticEvent {
+                        line: 0,
+                        timestamp_ms: 0,
+                        record_type: "medulla:control".to_string(),
+                        event: HarnessEvent {
+                            kind: "status".to_string(),
+                            role: "system".to_string(),
+                            payload: json!({ "state": state, "detail": detail }),
+                            ..Default::default()
+                        },
+                    });
+                }
+            }
+            Ok(RunTaskResult {
+                session_id: None,
+                usage: None,
+                provider: opts.provider,
+                reply: "ok".to_string(),
+                events: 3,
+            })
+        })
+    })
+}
+
 /// A runner that emits two cumulative thinking snapshots inside one throttle window.
 pub(super) fn quick_thinking_runner() -> RunTaskFn {
     Arc::new(move |mut opts: RunTaskOptions| {

src/sdk/src/daemon/types.rs

@@ -63,6 +63,27 @@ pub const CAPACITY_REJECTION_PREFIX: &str = "daemon at capacity";
 /// person is finished with it.
 pub const HARNESS_HELD_PREFIX: &str = "harness held by operator";
 
+/// The leading text of the `status` frame a worker sends when an operator takes
+/// a session that is *already running a task*.
+///
+/// A wire format in practice, like [`CAPACITY_REJECTION_PREFIX`]: the requesting
+/// hub matches on it to pause its no-progress watchdog for the duration of the
+/// hold (`crate::hub::runner`). A person reading their session is not a crashed
+/// worker, and a 30-minute hold must not be reaped as one — but the window must
+/// *pause* rather than be switched off, so a worker that dies while holding is
+/// still given up on once the session is handed back.
+///
+/// Distinct from [`HARNESS_HELD_PREFIX`], which is a *terminal* refusal ("I did
+/// not attempt this"). This one says the opposite: the task is alive, retained,
+/// and waiting on a human.
+pub const SESSION_HELD_STATUS_PREFIX: &str = "session held by operator";
+
+/// The leading text of the `status` frame that ends a hold.
+///
+/// Sent when control returns to the orchestrator and the hand-back turn starts,
+/// so the watchdog resumes on exactly the frame that says work has restarted.
+pub const SESSION_RESUMED_STATUS_PREFIX: &str = "session handed back";
+
 /// A lock-serialized encrypted send: `(to, body) -> ()`. Errors are handled by
 /// the transport (logged), so the runtime never observes a send failure.
 pub type SendFn =

src/sdk/src/hub/roster/mod.rs

@@ -63,33 +63,29 @@ fn to_agent(w: &HubWorker, catalog: &[crate::runtime::AgentTemplate]) -> Value {
     if let Some(workspace) = w.workspace_path() {
         metadata["workspace"] = json!(workspace);
     }
-    // Who holds the harness, and only when that is a person. Absent means the
-    // orchestrator has it, which is both the common case and the one worth
-    // keeping byte-stable: this advert is re-emitted on every roster mutation,
-    // and a key that flips on each one is a diff nobody can read.
-    if w.control.is_operator() {
-        metadata["control"] = json!(w.control.as_str());
-        if let Some(reason) = w
-            .control_reason
-            .as_deref()
-            .map(str::trim)
-            .filter(|r| !r.is_empty())
-        {
-            metadata["controlReason"] = json!(reason);
-        }
-        if let Some(since) = w.control_since {
-            metadata["controlSince"] = json!(since);
-        }
-    }
-    // The brief from the last handback. Carried only while the orchestrator
-    // actually holds the harness: an invitation to continue work in a workspace
-    // the operator has since re-taken is one the orchestrator cannot act on, and
-    // planning against it wastes a pass.
-    if let (false, Some(handoff)) = (w.control.is_operator(), w.handoff.as_ref()) {
-        if let Ok(value) = serde_json::to_value(handoff) {
-            metadata["handoff"] = value;
-        }
-    }
+    // Control state is deliberately NOT advertised — not the hold, not its
+    // reason, not since when, and not the handback brief that only exists
+    // because of one.
+    //
+    // It used to be, and it was right when a worker *was* an agent *was* a
+    // machine *was* one implicit session: "this worker is held" and "this
+    // session is held" were the same sentence. They are not any more. An agent
+    // now runs N sessions ([`HubWorker::max_sessions`]), a person takes *one* of
+    // them, and this flag has no room to say which — so a backend that folds
+    // held-state onto its ledger by `agentId` would mark every pending task on
+    // the agent as held, including the ones running perfectly well in other
+    // sessions. Emitting something wrong is worse than emitting nothing: the
+    // wrong thing is acted on.
+    //
+    // Saying it correctly needs session identity on the wire, which is only
+    // actionable once a dispatch can name a session (spec §C3, deferred). Until
+    // then this stays local: [`HubWorker::control`] and the whole take /
+    // hand-back path are unchanged and still decide medulla's *own* dispatch
+    // (`session_for` skips a session an operator holds; a held in-flight task
+    // suspends and is delivered by the hand-back turn). None of it crosses the
+    // wire, and the backend learns what happened the only way that cannot be
+    // mis-keyed: through the task's own result.
+    //
     // The host this agent runs on, when this hub knows which one. The backend
     // prefers a supplied id and only synthesizes `host:${socketId}` as a last
     // resort, so saying it here is what stops five machines behind one hub

src/sdk/src/hub/runner/mod.rs

@@ -11,7 +11,7 @@
 //! bounds, and orchestrator-driven abort.
 
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -98,10 +98,18 @@ const LIVENESS_TICK: Duration = Duration::from_millis(100);
 ///
 /// A bridge with no notion of reachability (the in-memory bus, every test fake)
 /// answers `Live` by default, so this behaves exactly like `sleep` for them.
-async fn live_sleep(relay: &dyn Relay, peer: &str, window: Duration) {
+///
+/// `held` is the second gate, and it is the same idea one layer up: the worker
+/// reports (`crate::daemon::SESSION_HELD_STATUS_PREFIX`) that an operator has
+/// taken the session serving this dispatch, and a person reading their own
+/// session is no more "a crashed worker" than an unreachable one is. Held time
+/// therefore does not accrue either, and the window **resumes rather than
+/// resets** when the session is handed back — a worker that dies mid-hold is
+/// still given up on, it just is not given up on *while* a human has it.
+async fn live_sleep(relay: &dyn Relay, peer: &str, window: Duration, held: &AtomicBool) {
     let mut remaining = window;
     while !remaining.is_zero() {
-        if relay.liveness(peer).await == BridgeLiveness::Live {
+        if relay.liveness(peer).await == BridgeLiveness::Live && !held.load(Ordering::Acquire) {
             let step = LIVENESS_TICK.min(remaining);
             tokio::time::sleep(step).await;
             remaining -= step;
@@ -409,6 +417,9 @@ impl TaskRunner {
             );
             let (tx, mut rx) = oneshot::channel();
             let activity = Arc::new(Notify::new());
+            // Held for the whole attempt, not only for as long as the waiter is
+            // registered: the windows below read it, and the pump writes it.
+            let held = Arc::new(AtomicBool::new(false));
             self.waiters.lock().await.insert(
                 cid.clone(),
                 Waiter {
@@ -421,6 +432,7 @@ impl TaskRunner {
                     reply: tx,
                     status: status.clone(),
                     activity: activity.clone(),
+                    held: held.clone(),
                 },
             );
 
@@ -502,7 +514,7 @@ impl TaskRunner {
                             // A frame: the peer is working. Reset the idle clock.
                             _ = activity.notified() => continue,
                             _ = live_sleep(
-                                self.relay.as_ref(), &req.worker_address, self.idle_window,
+                                self.relay.as_ref(), &req.worker_address, self.idle_window, &held,
                             ) => {
                                 self.waiters.lock().await.remove(&cid);
                                 send_abort(
@@ -514,7 +526,7 @@ impl TaskRunner {
                     }
                 }
                 _ = live_sleep(
-                    self.relay.as_ref(), &req.worker_address, self.ack_window,
+                    self.relay.as_ref(), &req.worker_address, self.ack_window, &held,
                 ) => {
                     // Silence while the link was live — so the peer itself is not
                     // answering, not the network. Reset and resend, or give up.