diff --git a/src/sdk/src/daemon/embedded/types.rs b/src/sdk/src/daemon/embedded/types.rs index 0420ac68d..24dd23598 100644 --- a/src/sdk/src/daemon/embedded/types.rs +++ b/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; diff --git a/src/sdk/src/daemon/entry.rs b/src/sdk/src/daemon/entry.rs index bfc4bb083..292038fdb 100644 --- a/src/sdk/src/daemon/entry.rs +++ b/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; diff --git a/src/sdk/src/daemon/mod.rs b/src/sdk/src/daemon/mod.rs index 185ab4692..fa6cd5343 100644 --- a/src/sdk/src/daemon/mod.rs +++ b/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, }; diff --git a/src/sdk/src/daemon/task_loop/run.rs b/src/sdk/src/daemon/task_loop/run.rs index 9eb18d929..cd4e4e246 100644 --- a/src/sdk/src/daemon/task_loop/run.rs +++ b/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))); } diff --git a/src/sdk/src/daemon/tests/capability_tests.rs b/src/sdk/src/daemon/tests/capability_tests.rs index 1567d7f4a..a5e40e6c8 100644 --- a/src/sdk/src/daemon/tests/capability_tests.rs +++ b/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(); diff --git a/src/sdk/src/daemon/tests/mod.rs b/src/sdk/src/daemon/tests/mod.rs index c49a471e9..0dbe71592 100644 --- a/src/sdk/src/daemon/tests/mod.rs +++ b/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| { diff --git a/src/sdk/src/daemon/types.rs b/src/sdk/src/daemon/types.rs index 314c2dd2e..0fd84fc2c 100644 --- a/src/sdk/src/daemon/types.rs +++ b/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 = diff --git a/src/sdk/src/hub/roster/mod.rs b/src/sdk/src/hub/roster/mod.rs index db6eb20ee..596abcae4 100644 --- a/src/sdk/src/hub/roster/mod.rs +++ b/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 diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 26dbacdb3..fa3578aa4 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/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. diff --git a/src/sdk/src/hub/runner/pump/mod.rs b/src/sdk/src/hub/runner/pump/mod.rs index f5fec3de8..6ac7189b9 100644 --- a/src/sdk/src/hub/runner/pump/mod.rs +++ b/src/sdk/src/hub/runner/pump/mod.rs @@ -154,6 +154,13 @@ pub(super) async fn route_frame( } TaskFrameKind::Status => { if let Some(w) = map.get(&key) { + // Control markers first, and read rather than merely forwarded: + // a held session is the one kind of silence that is not a dead + // worker, so it pauses this dispatch's no-progress window + // instead of counting against it (see [`Waiter::held`]). + if let Some(held) = control_marker(&frame.text) { + w.held.store(held, std::sync::atomic::Ordering::Release); + } if let Some(tx) = &w.status { let _ = tx.send(frame.text); } @@ -164,6 +171,24 @@ pub(super) async fn route_frame( } } +/// Whether a `status` frame announces a control change, and which way. +/// +/// `Some(true)` — an operator has taken the session serving the dispatch; +/// `Some(false)` — they handed it back and the hand-back turn has started; +/// `None` — ordinary progress chatter, which says nothing about control. +/// +/// Matched on the shared prefixes the daemon builds these from, so the two ends +/// cannot drift apart behind a copied literal. +fn control_marker(text: &str) -> Option { + if text.starts_with(crate::daemon::SESSION_HELD_STATUS_PREFIX) { + return Some(true); + } + if text.starts_with(crate::daemon::SESSION_RESUMED_STATUS_PREFIX) { + return Some(false); + } + None +} + /// The correlation key a frame routes under: its `correlationId`, or its /// `taskId` when it carries none. fn key_of(frame: &TaskFrame) -> String { diff --git a/src/sdk/src/hub/runner/types.rs b/src/sdk/src/hub/runner/types.rs index 7cd422cd4..818bbad76 100644 --- a/src/sdk/src/hub/runner/types.rs +++ b/src/sdk/src/hub/runner/types.rs @@ -19,6 +19,18 @@ pub(super) struct Waiter { /// Notified on ANY inbound frame for this dispatch — the "peer is alive" /// signal the runner's ack window waits on. pub(super) activity: Arc, + /// Whether an operator currently holds the session serving this dispatch. + /// + /// Set and cleared by the pump from the worker's control markers + /// ([`crate::daemon::SESSION_HELD_STATUS_PREFIX`]), read by the runner's + /// windows: held time does not accrue against the no-progress watchdog, the + /// same way time on a dead link does not. A person reading their session is + /// not a crashed worker, and a hold outlasts every window this runner has. + /// + /// Shared with the [`super::TaskRunner::run`] call rather than only living + /// in the map, so the window keeps reading it after a terminal frame has + /// removed the waiter. + pub(super) held: Arc, } /// Shared registry of in-flight dispatches, keyed by `correlationId`. pub(super) type Waiters = Arc>>; diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index a7f4f9341..acb5abf99 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/src/sdk/src/hub/socket/task_run.rs @@ -31,11 +31,15 @@ use super::{first_obj, str_field, wire_task_id}; /// later" is the one answer that is certainly wrong. medulla bounds the /// re-dispatch with its own attempt ceiling and exponential backoff, so this /// cannot become a hot loop against a saturated worker. -/// A workspace an operator is sitting in counts too. Nothing was attempted, and -/// the harness will be usable again the moment they hand it back — so the task -/// is deferred, not failed. It carries a `reason` on the wire (see -/// [`result_frame`]) precisely so the orchestrator can route elsewhere instead -/// of waiting on a person. +/// A checkout an operator is sitting in counts too — in the one case that still +/// reaches here. A dispatch no longer *refuses* on meeting a person: a held +/// session is not a dispatch candidate at all, and a dispatch with nothing else +/// to run in queues behind the checkout's writer instead of failing. What +/// survives is that queue running out of budget: nothing was attempted, the tree +/// will be usable again the moment the person is done, and the task is deferred +/// rather than failed. It carries a `reason` on the wire (see [`result_frame`]) +/// precisely so the orchestrator can route elsewhere instead of waiting on +/// somebody who may have left for the day. pub(in crate::hub) fn is_retryable(err: &RunError) -> bool { matches!( err, @@ -96,10 +100,19 @@ pub(in crate::hub) fn result_frame( } frame } - // A held workspace is the one failure the orchestrator can act on + // An occupied checkout is the one failure the orchestrator can act on // *specifically*, so it is the one that names itself. `reason` and // `retryAfterMs` are additive and ignorable: a backend that has never // heard of either still reads this as an ordinary retryable failure. + // + // Deliberately kept, and kept byte-identical, even though the blanket + // refusal that used to produce it is gone. It is now reached by exactly + // one path — a dispatch that queued behind a person for its whole + // budget — and that path needs precisely this frame: the backend already + // treats `harnessHeld` as retryable, so replacing it with a terminal + // error would turn a task that was never attempted into one nobody + // retries. The rule is that a dispatch ends in a real result or a real + // error; this is the error half. Err(err @ RunError::Held(_)) => json!({ "taskId": task_id, "ok": false, diff --git a/src/sdk/src/hub/tests/handoff_advert.rs b/src/sdk/src/hub/tests/handoff_advert.rs index 810bc858f..574a6f20b 100644 --- a/src/sdk/src/hub/tests/handoff_advert.rs +++ b/src/sdk/src/hub/tests/handoff_advert.rs @@ -1,9 +1,24 @@ -//! What the roster advert says about who holds a harness. +//! What the roster advert says about who holds a session: **nothing**. //! -//! The advert is the transport for the whole handoff feature: it is already -//! re-emitted on every roster mutation, so a control change is already an event. -//! These pin the two properties that makes safe — that the common case stays -//! byte-stable, and that a stale invitation is never advertised. +//! It used to say a great deal — `control`, `controlReason`, `controlSince`, and +//! the handback brief — and every one of those keys was right when a worker *was* +//! an agent *was* a machine *was* one implicit session. An agent now runs N +//! sessions, a person takes *one*, and none of those keys has anywhere to put +//! which one. A backend folding held-state onto its ledger by `agentId` would +//! therefore mark every pending task on that agent as held, including the ones +//! running fine in sibling sessions. +//! +//! So these tests pin the *absence*. They are written as absence assertions +//! rather than deleted outright because the keys were once present and correct, +//! and a future reader looking at [`HubWorker::control`] — which still exists, +//! and still drives medulla's own dispatch — will otherwise reasonably conclude +//! that not advertising it is an oversight. It is not: saying it correctly needs +//! session grain on the wire, which arrives with inbound session targeting (§C3). +//! +//! The local behaviour these keys used to describe is unchanged and covered +//! elsewhere: dispatch skips a session an operator holds, an in-flight turn +//! suspends instead of being discarded, and the hand-back turn delivers the +//! task's result through the ordinary result frame. use super::super::roster::{register_payload, HubWorker}; use super::super::{HandoffControl, HarnessHandoff}; @@ -45,81 +60,90 @@ fn metadata(w: HubWorker) -> serde_json::Value { #[test] fn an_orchestrator_held_harness_says_nothing_about_control() { - // Absent means orchestrator-held. Omitting the common case is what keeps - // this advert byte-stable across the re-emissions it gets on every roster - // mutation — a key that flips on each one is a diff nobody can read. let meta = metadata(worker()); assert!(meta.get("control").is_none()); assert!(meta.get("controlReason").is_none()); assert!(meta.get("controlSince").is_none()); - // The pre-existing keys are untouched, so a backend that has never heard of - // handoff reads exactly what it read before. + // The keys that do place the agent are untouched. assert_eq!(meta["address"], "GRVaddr"); assert_eq!(meta["workspace"], "/repos/acme"); } #[test] -fn an_operator_held_harness_advertises_the_hold_with_its_reason() { - let meta = metadata(HubWorker { +fn an_operator_held_agent_advertises_exactly_what_an_unheld_one_does() { + // The grain mismatch, stated as a test. Control is per-agent here and + // per-session in the model, so "held" on this advert cannot say *which* of + // the agent's sessions a person took — and a backend keying held-state by + // `agentId` would apply it to every task the agent is running. + // + // Do not "restore" these keys. Advertising a hold needs the session id the + // hold is about, which needs inbound session targeting (§C3). + let held = metadata(HubWorker { control: HandoffControl::Operator, control_reason: Some("pairing on the auth migration".to_string()), control_since: Some(1_753_420_000_000), ..worker() }); - // "operator", not "user": the orchestrator reasons about operators, and one - // word per concept is worth more than matching the local enum's variant - // name. - assert_eq!(meta["control"], "operator"); - assert_eq!(meta["controlReason"], "pairing on the auth migration"); - assert_eq!(meta["controlSince"], 1_753_420_000_000i64); -} - -#[test] -fn a_hold_with_no_reason_omits_the_key_rather_than_sending_blank() { - let meta = metadata(HubWorker { - control: HandoffControl::Operator, - control_reason: Some(" ".to_string()), - ..worker() - }); - - assert_eq!(meta["control"], "operator"); - assert!( - meta.get("controlReason").is_none(), - "whitespace is not a reason" + assert!(held.get("control").is_none(), "a hold is local state"); + assert!(held.get("controlReason").is_none()); + assert!(held.get("controlSince").is_none()); + assert_eq!( + held, + metadata(worker()), + "taking a session must not change one byte of the agent's advert" ); } #[test] -fn a_handed_back_harness_carries_its_brief() { +fn a_handed_back_harness_does_not_carry_its_brief() { + // The brief is per *session* — it names one (`sessionId`, and a transcript + // from that one pty) — but this slot is per *agent*, so two sessions handed + // back on one agent silently overwrite each other and the reader cannot tell + // that happened. It was also emitted through the same per-agent control gate + // as the keys above, which means whether an operator saw a brief at all + // depended on whether some *unrelated* session of that agent was held. + // + // With control off the wire this would be the last piece of agent-grain + // control state left on it: a brief exists only because a person took a + // session and gave it back. It travels again when a brief can name its + // session on the wire (§C3). let meta = metadata(HubWorker { handoff: Some(brief()), ..worker() }); - assert!(meta.get("control").is_none(), "the orchestrator holds it"); - assert_eq!(meta["handoff"]["id"], "w_3-1"); - assert_eq!(meta["handoff"]["workspacePath"], "/repos/acme"); - assert_eq!(meta["handoff"]["branch"], "feat/login"); - assert_eq!(meta["handoff"]["note"], "stuck on the failing e2e"); - assert_eq!(meta["handoff"]["transcriptTruncated"], false); + assert!(meta.get("handoff").is_none()); + assert_eq!( + meta, + metadata(worker()), + "a handback must not change the agent's advert either" + ); } #[test] -fn a_brief_on_a_re_taken_harness_is_not_advertised() { - // The stale-invitation case. A brief says "continue this work here"; on a - // harness the operator has since taken back, acting on it is refused. Left - // advertised it would cost the orchestrator a planning pass every cycle. - let meta = metadata(HubWorker { +fn the_local_hold_state_survives_even_though_it_is_never_advertised() { + // The other half of the change, and the reason `control` is still a field: + // medulla's own dispatch reads it. Dropping the advert keys must not be + // mistaken for dropping the feature — the roster still knows, it just does + // not tell the backend something the backend cannot key correctly. + let held = HubWorker { control: HandoffControl::Operator, + control_reason: Some("pairing on the auth migration".to_string()), + control_since: Some(1_753_420_000_000), handoff: Some(brief()), ..worker() - }); + }; - assert_eq!(meta["control"], "operator"); - assert!( - meta.get("handoff").is_none(), - "an invitation into a workspace the operator holds is not actionable" + assert!(held.control.is_operator()); + assert_eq!( + held.control_reason.as_deref(), + Some("pairing on the auth migration") + ); + assert_eq!(held.control_since, Some(1_753_420_000_000)); + assert_eq!( + held.handoff.as_ref().map(|b| b.session_id.as_str()), + Some("w_3") ); } diff --git a/src/sdk/src/hub/tests/held.rs b/src/sdk/src/hub/tests/held.rs index 376efccbd..8fb6080d5 100644 --- a/src/sdk/src/hub/tests/held.rs +++ b/src/sdk/src/hub/tests/held.rs @@ -1,4 +1,12 @@ -//! Refusing a task because a person is working in the workspace. +//! Deferring a task because a person is working in the checkout. +//! +//! Narrower than it was. A dispatch no longer refuses on meeting a person — +//! their session is simply not a candidate, and a dispatch with nothing else to +//! run in queues behind the checkout's writer. `RunError::Held` is what that +//! queue reports when it outlives the caller's budget, which is the one path +//! left to it and the reason the shape below is unchanged: the backend already +//! reads `harnessHeld` as retryable, and a task that was never attempted must +//! not come back as one that failed. //! //! Two halves, and they have to agree. The daemon can only say so in the text of //! an `error` frame, so [`settle`](super::super::runner) has to recognise that diff --git a/src/sdk/src/hub/tests/held_watchdog.rs b/src/sdk/src/hub/tests/held_watchdog.rs new file mode 100644 index 000000000..111de296b --- /dev/null +++ b/src/sdk/src/hub/tests/held_watchdog.rs @@ -0,0 +1,250 @@ +//! The third gate on the no-progress watchdog: a session a person is holding. +//! +//! [`liveness`](super::liveness) covers the first two — a window only accrues +//! while the link to that peer is live. This is the same idea one layer up. A +//! worker whose session an operator has taken says so +//! ([`SESSION_HELD_STATUS_PREFIX`](crate::daemon::SESSION_HELD_STATUS_PREFIX)), +//! and from that frame until the hand-back it sends nothing at all — the harness +//! is not running a turn, a human is typing in it. Thirty minutes of that is +//! indistinguishable from a crashed worker to a clock that only counts frames, +//! and the old answer was to reap the dispatch: the task died while the person +//! was still working, and the only notice anyone got was `bridge task timed out`. +//! +//! The three tests here are deliberately a set, because any one alone would pass +//! for the wrong reason: +//! +//! - [`a_held_session_outlasts_the_no_progress_window`] proves the clock pauses. +//! - [`a_silent_worker_that_never_reported_a_hold_still_times_out`] proves it was +//! *gated*, not deleted — the same test would pass if the watchdog had simply +//! been removed. +//! - [`a_worker_that_dies_after_the_hand_back_is_still_given_up_on`] proves the +//! pause ends where it should. A hold that leaked past the hand-back would +//! make a dead worker unreapable for ever, which is exactly the leak the +//! window exists to prevent. + +use std::collections::VecDeque; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use tokio::sync::Mutex; + +use crate::bridge::{BridgeLiveness, InboundMessage}; +use crate::hub::{Relay, RunError, TaskRequest, TaskRunner}; +use crate::protocol::{decode_task_frame, encode_task_frame, EncodeFrameInput, TaskFrameKind}; + +/// Comfortably longer than the runner's 240 s no-progress window, and about as +/// long as a person actually keeps a session: the point is the ratio. +const HOLD: Duration = Duration::from_secs(1_800); + +/// Longer than the window too, so a resumed clock has room to fire. +const AFTER_HAND_BACK: Duration = Duration::from_secs(600); + +/// A worker that acks, then goes quiet in the way a held session does. +struct HoldingPeer { + inbox: Mutex>, + /// The correlation id of the dispatch, learned from the frame we were sent. + correlation: Mutex>, + /// Whether to announce the hold at all. `false` is a worker that simply + /// stopped talking — a crash, which must still be reaped. + announces: bool, +} + +impl HoldingPeer { + fn new(announces: bool) -> Arc { + Arc::new(HoldingPeer { + inbox: Mutex::new(VecDeque::new()), + correlation: Mutex::new(None), + announces, + }) + } + + /// Queue one frame from the worker, under the dispatch's correlation id. + async fn emit(&self, kind: TaskFrameKind, text: &str) { + let correlation = self.correlation.lock().await.clone(); + let body = encode_task_frame(EncodeFrameInput { + kind, + task_id: "t1".to_string(), + text: text.to_string(), + ts: crate::clock::iso_now(), + correlation_id: correlation, + harness: None, + provider: None, + custom_harness: None, + model: None, + tool_mode: None, + workflow: None, + workflow_fingerprint: None, + workflow_inputs: Default::default(), + conversation: None, + fleet_depth: 0, + }); + self.inbox.lock().await.push_back(InboundMessage { + from: "host-a".to_string(), + text: body, + }); + } + + /// The status frame the worker sends when an operator takes the session. + async fn report_held(&self) { + self.emit( + TaskFrameKind::Status, + &format!( + "{} · the codex turn is suspended, not lost", + crate::daemon::SESSION_HELD_STATUS_PREFIX + ), + ) + .await; + } + + /// The status frame that ends the hold and restarts the hand-back turn. + async fn report_resumed(&self) { + self.emit( + TaskFrameKind::Status, + &format!( + "{} · reviewing what changed", + crate::daemon::SESSION_RESUMED_STATUS_PREFIX + ), + ) + .await; + } + + /// The hand-back turn's answer — the task's result. + async fn reply(&self) { + self.emit(TaskFrameKind::Reply, "finished after the hand-back") + .await; + } +} + +#[async_trait] +impl Relay for HoldingPeer { + async fn send(&self, _to: &str, body: &str) -> Result<(), String> { + let Some(frame) = decode_task_frame(body) else { + return Ok(()); + }; + if frame.kind != TaskFrameKind::Task { + return Ok(()); + } + *self.correlation.lock().await = frame.correlation_id.clone(); + // Alive, and working — then a person takes the session and the frames + // stop. + self.emit(TaskFrameKind::Ack, "task accepted").await; + if self.announces { + self.report_held().await; + } + Ok(()) + } + + async fn drain_inbox(&self, limit: i64) -> Vec { + if limit <= 0 { + return Vec::new(); + } + let mut inbox = self.inbox.lock().await; + let count = usize::try_from(limit) + .unwrap_or(usize::MAX) + .min(inbox.len()); + inbox.drain(..count).collect() + } + + async fn request_contact(&self, _peer: &str) -> Result<(), String> { + Ok(()) + } + + async fn contact_accepted(&self, _peer: &str) -> bool { + true + } + + async fn reset_session(&self, _peer: &str) {} + + async fn liveness(&self, _peer: &str) -> BridgeLiveness { + BridgeLiveness::Live + } +} + +/// The dispatch every test here runs. +fn req() -> TaskRequest { + TaskRequest { + task_id: "t1".to_string(), + abort_id: "t1".to_string(), + cycle_id: Some("c1".to_string()), + instruction: "finish the migration".to_string(), + worker_address: "host-a".to_string(), + provider: None, + custom_harness: None, + model: None, + tool_mode: None, + workflow: None, + workflow_fingerprint: None, + workflow_inputs: Default::default(), + conversation: None, + fleet_depth: 0, + } +} + +#[tokio::test(start_paused = true)] +async fn a_held_session_outlasts_the_no_progress_window() { + // Half an hour of a person working in the session, against a 240-second + // window. Ungated, this dispatch is reaped ~28 minutes before its result + // exists — and reaped is not merely "late": the runner sends the worker an + // abort, so the hand-back turn would never even run. + let peer = HoldingPeer::new(true); + let runner = TaskRunner::start(peer.clone() as Arc, Duration::from_millis(10)); + + let operator = { + let peer = peer.clone(); + tokio::spawn(async move { + tokio::time::sleep(HOLD).await; + peer.report_resumed().await; + peer.reply().await; + }) + }; + + let outcome = runner.run(req(), None).await; + operator.await.expect("the operator task completes"); + + let outcome = outcome.expect("a held session must not fail its task"); + assert_eq!(outcome.reply, "finished after the hand-back"); +} + +#[tokio::test(start_paused = true)] +async fn a_silent_worker_that_never_reported_a_hold_still_times_out() { + // The gate is a gate. This peer acks and then dies, saying nothing about + // control — the exact case the window exists for, and the one a deleted + // window would leave pinned for ever. + let peer = HoldingPeer::new(false); + let runner = TaskRunner::start(peer as Arc, Duration::from_millis(10)); + + let outcome = runner.run(req(), None).await; + + assert!( + matches!(outcome, Err(RunError::Timeout)), + "a worker that acked and vanished must still be given up on: {outcome:?}" + ); +} + +#[tokio::test(start_paused = true)] +async fn a_worker_that_dies_after_the_hand_back_is_still_given_up_on() { + // The pause ends with the hold. A worker that reports the hand-back and then + // stops talking is a crashed worker again, and must be reaped on the same + // schedule as any other — otherwise one hold makes a dispatch immortal. + let peer = HoldingPeer::new(true); + let runner = TaskRunner::start(peer.clone() as Arc, Duration::from_millis(10)); + + let operator = { + let peer = peer.clone(); + tokio::spawn(async move { + tokio::time::sleep(HOLD).await; + // Handed back — and then nothing. No reply ever comes. + peer.report_resumed().await; + tokio::time::sleep(AFTER_HAND_BACK).await; + }) + }; + + let outcome = runner.run(req(), None).await; + operator.await.expect("the operator task completes"); + + assert!( + matches!(outcome, Err(RunError::Timeout)), + "the window must resume when the hold ends: {outcome:?}" + ); +} diff --git a/src/sdk/src/hub/tests/mod.rs b/src/sdk/src/hub/tests/mod.rs index acd92a9df..540cb5562 100644 --- a/src/sdk/src/hub/tests/mod.rs +++ b/src/sdk/src/hub/tests/mod.rs @@ -3,13 +3,15 @@ //! its attribution; [`roster`] covers advertising, addressing and dedupe; //! [`dispatch`] the sender-runner's full dispatch/route/settle path against a //! fake worker; [`liveness`] the two-layer timeout gate of host-link protocol -//! §6.3. +//! §6.3; [`held_watchdog`] the third gate on that window — a session an operator +//! is holding. mod activity; mod capabilities; mod dispatch; mod handoff_advert; mod held; +mod held_watchdog; mod liveness; mod roster; mod system_info; diff --git a/src/sdk/src/hub/tests/roster.rs b/src/sdk/src/hub/tests/roster.rs index 8345e096d..13fd03297 100644 --- a/src/sdk/src/hub/tests/roster.rs +++ b/src/sdk/src/hub/tests/roster.rs @@ -933,12 +933,17 @@ fn an_agent_with_no_declared_workspace_still_serializes() { assert_eq!(agent["metadata"]["harness"], "claude"); } -/// The keys the backend's per-agent control ingestion and the manager ledger's -/// control folds read. A regression here is silent — the advert still parses, -/// the folds just stop seeing a hold — so the whole metadata object is pinned -/// rather than spot-checked. -#[test] -fn the_control_and_handoff_keys_keep_exactly_the_shape_the_backend_folds() { +/// The whole metadata object, pinned — including what is **not** in it. +/// +/// Control state (`control`, `controlReason`, `controlSince`) and the handback +/// brief are per-*agent* keys describing a per-*session* fact, so a backend +/// folding them by `agentId` would mark every task on the agent as held when a +/// person took one session. They are therefore not advertised at all; see +/// `hub::tests::handoff_advert` for the full reasoning and the local state that +/// replaces them. The object is pinned rather than spot-checked because a +/// regression either way is silent — the advert still parses. +#[test] +fn the_advert_metadata_carries_placement_and_never_control() { let mut held = placed("this-device", "this-device"); held.workspace = Some(crate::runtime::WorkspaceRef::checkout("/repos/acme")); held.control = super::super::HandoffControl::Operator; @@ -959,14 +964,13 @@ fn the_control_and_handoff_keys_keep_exactly_the_shape_the_backend_folds() { "harness": "claude", "maxSessions": 1, "workspace": "/repos/acme", - "control": "operator", - "controlReason": "pairing on the migration", - "controlSince": 1_753_420_600_000i64, }), - "control/controlReason/controlSince must not move, gain a wrapper, or change spelling" + "address/harness/maxSessions/workspace must not move, gain a wrapper, or \ + change spelling — and control must not appear at any grain" ); - // And the handoff brief, which only rides while the orchestrator holds it. + // Nor does the brief a handback produces: it names one session, this slot is + // one per agent, and it exists only because a person held something. let mut handed_back = worker("w1", "GRVaddr"); handed_back.handoff = Some(super::super::HarnessHandoff { id: "w_3-1".to_string(), @@ -982,8 +986,8 @@ fn the_control_and_handoff_keys_keep_exactly_the_shape_the_backend_folds() { transcript_truncated: false, }); let payload = register_payload(&[handed_back], &no_presence(), &[], &[]); - let handoff = &payload["agents"][0]["metadata"]["handoff"]; - assert_eq!(handoff["id"], "w_3-1"); - assert_eq!(handoff["sessionId"], "w_3"); - assert_eq!(handoff["workspacePath"], "/repos/acme"); + assert!( + payload["agents"][0]["metadata"].get("handoff").is_none(), + "a per-session brief has no honest home on a per-agent advert" + ); } diff --git a/src/tui/src/ui/app/render/agents/mod.rs b/src/tui/src/ui/app/render/agents/mod.rs index 87cd6c567..c330dc5bf 100644 --- a/src/tui/src/ui/app/render/agents/mod.rs +++ b/src/tui/src/ui/app/render/agents/mod.rs @@ -126,6 +126,18 @@ impl App { } self.pane_session = selection.session.clone(); self.rail_session = selection.session.clone(); + // A session row this device is not running: watchable, but not takeable + // (§E7). Recorded here because this is the only place that can tell the + // difference — one row down the cursor, both cases are a `None` session. + self.pane_remote_session = match (&selection.session, selection.rows.get(selection.active)) + { + (None, Some(RailRow::Session(row))) => Some( + row.agent_id + .clone() + .unwrap_or_else(|| "another host".to_string()), + ), + _ => None, + }; selection } diff --git a/src/tui/src/ui/app/render/mod.rs b/src/tui/src/ui/app/render/mod.rs index eb961351b..97ec51268 100644 --- a/src/tui/src/ui/app/render/mod.rs +++ b/src/tui/src/ui/app/render/mod.rs @@ -294,6 +294,9 @@ impl App { // tab was showing several frames ago. `draw_agents_pane` fills it back // in when it resolves a session. self.pane_session = None; + // Its counterpart, for the same reason: a remembered remote row would + // answer the take chord on a tab that is not showing it. + self.pane_remote_session = None; // Same reasoning as above: a stale rect would route the wheel into a // terminal that is no longer on screen. self.hit_session = None; diff --git a/src/tui/src/ui/app/session_control.rs b/src/tui/src/ui/app/session_control.rs index 8a789ee6f..af0d7ffc2 100644 --- a/src/tui/src/ui/app/session_control.rs +++ b/src/tui/src/ui/app/session_control.rs @@ -263,6 +263,24 @@ impl App { /// reason the attach chord does: an operator who pressed a key and saw no /// change cannot tell "wrong row" from "broken feature". fn selected_session(&mut self) -> Option<(crate::ui::harness_pane::LocalSessions, String)> { + // A session on another host is a real session the cursor is really on — + // it is just not one this machine can take (§E7). The hub resolves a + // hold by local workspace path, so there is nothing here to flip, and + // the honest answer names the machine rather than pretending the row is + // empty. Watching it is unaffected: the screen mirror is read-only by + // design either way. + // + // Asked before "is this device hosting", because it is the more specific + // answer and the two are not exclusive: a laptop that hosts nothing can + // still be looking at a remote host's session, and "this device is not + // hosting" would be a true sentence about the wrong machine. + if let Some(agent) = self.pane_remote_session.clone() { + self.set_status(format!( + "{agent} runs on another host — you can watch this session, but \ + taking control is local-only for now" + )); + return None; + } let Some(harnesses) = self.local_sessions.clone() else { self.set_status("This device is not hosting, so it has no sessions"); return None; diff --git a/src/tui/src/ui/app/session_control_tests.rs b/src/tui/src/ui/app/session_control_tests.rs index 5ba446c04..fed1e3567 100644 --- a/src/tui/src/ui/app/session_control_tests.rs +++ b/src/tui/src/ui/app/session_control_tests.rs @@ -1,8 +1,15 @@ -//! Focused tests for harness-picker keyboard classification. +//! Focused tests for the session-control chords: what they classify as text, +//! and what they refuse. + +use std::sync::Arc; use crossterm::event::KeyModifiers; +use medulla::config::LoadedConfig; +use medulla::runtime::mock::MockRuntime; +use medulla::runtime::Runtime; use super::session_control::is_text_input; +use super::types::App; #[test] fn workspace_text_accepts_altgr_but_rejects_control_shortcuts() { @@ -12,3 +19,76 @@ fn workspace_text_accepts_altgr_but_rejects_control_shortcuts() { assert!(!is_text_input(KeyModifiers::CONTROL)); assert!(!is_text_input(KeyModifiers::ALT)); } + +fn app() -> App { + let rt: Arc = Arc::new(MockRuntime::demo()); + let mut loaded = LoadedConfig::defaults("medulla.tui.json".into()); + loaded.config.link = Some(medulla::config::LinkConfig::default()); + App::new(rt, loaded) +} + +#[test] +fn taking_a_session_on_another_host_is_refused_by_name() { + // §E7. The hub resolves a hold by *local workspace path*, so there is + // nothing on this machine to flip for a session running on another one — + // the take would silently do nothing. Remote takeover needs the owner → + // machine control frames wired into the hold path, and is a documented + // follow-up (§G). + // + // The refusal has to name the machine. Both "the cursor is on nothing" and + // "the cursor is on someone else's session" leave `pane_session` empty, and + // an operator told "no session on this row" while plainly looking at one + // reads it as a broken feature rather than as a boundary. + let mut app = app(); + app.pane_session = None; + app.pane_remote_session = Some("mac-studio-claude".to_string()); + + app.take_session_control(); + + let status = app.status().to_string(); + assert!( + status.contains("mac-studio-claude") && status.contains("another host"), + "the refusal must name the agent and say why: {status}" + ); + assert!( + status.contains("watch"), + "and must say what the operator CAN do — a remote session is viewable, \ + which is the whole of the screen mirror: {status}" + ); +} + +/// A hosting device with no sessions running on it. +fn hosting(app: &mut App) { + app.local_sessions = Some(crate::ui::harness_pane::LocalSessions { + sessions: crate::worker::pty::PtyManager::new(), + runtimes: Arc::new(std::sync::Mutex::new(Vec::new())), + hub_address: "this-device".to_string(), + env: std::collections::HashMap::new(), + workspace: "/repos/acme".to_string(), + providers: Vec::new(), + custom_harnesses: Vec::new(), + router: None, + attribution: true, + hooks: medulla::harness_hooks::HooksConfig::default(), + log: None, + }); +} + +#[test] +fn the_take_chord_on_an_empty_row_still_says_so() { + // The other side of the same branch: with no remote row recorded, the + // message must stay the plain one. A remote-session sentence on a host row + // or the composer would be worse than the generic answer. + let mut app = app(); + hosting(&mut app); + app.pane_session = None; + app.pane_remote_session = None; + + app.take_session_control(); + + let status = app.status().to_string(); + assert!( + status.contains("No session on this row"), + "unexpected status: {status}" + ); +} diff --git a/src/tui/src/ui/app/state.rs b/src/tui/src/ui/app/state.rs index 2dc0a21e8..4a38492d2 100644 --- a/src/tui/src/ui/app/state.rs +++ b/src/tui/src/ui/app/state.rs @@ -154,6 +154,7 @@ impl App { local_sessions: None, harness_focus: crate::ui::harness_pane::HarnessFocus::default(), pane_session: None, + pane_remote_session: None, rail_session: None, agent_picker: None, handback_prompt: None, diff --git a/src/tui/src/ui/app/types.rs b/src/tui/src/ui/app/types.rs index 926960208..5e563b1ad 100644 --- a/src/tui/src/ui/app/types.rs +++ b/src/tui/src/ui/app/types.rs @@ -1123,6 +1123,16 @@ pub struct App { // is where the rail cursor is turned into a selection; cleared at the top of // every draw so it can never name a pane that is no longer on screen. pub(super) pane_session: Option, + // The agent behind a selected session row that this device does NOT run, + // recorded alongside `pane_session` on the same draw. + // + // Its only purpose is to tell "the cursor is not on a session" apart from + // "the cursor is on somebody else's session", which `pane_session` cannot: + // both leave it `None`. Taking control resolves through the local workspace + // path, so a remote session can be watched but not taken (§E7), and an + // operator who presses the take chord on one deserves that answer rather + // than "no session on this row". + pub(super) pane_remote_session: Option, // The session selected on the Agents rail, retained while another tab is // visible. Unlike `pane_session`, this is navigation state rather than a // keyboard-routing capability: Changes uses it to keep following diff --git a/src/tui/src/worker/executor/hold.rs b/src/tui/src/worker/executor/hold.rs new file mode 100644 index 000000000..fadffac9e --- /dev/null +++ b/src/tui/src/worker/executor/hold.rs @@ -0,0 +1,282 @@ +//! Operator holds: queueing behind one, suspending a turn to one, and the +//! hand-back turn that turns one back into a task result. +//! +//! Three moments, one fact: a session an operator holds is **not a dispatch +//! candidate** (spec §4.1). Everything here follows from that. +//! +//! 1. **Before a turn** — nothing reusable, and the checkout already has a +//! writer in it: the dispatch *queues* behind them +//! ([`await_checkout_release`](PtySessionExecutor::await_checkout_release)). +//! It used to be refused outright, which made a person at a keyboard a task +//! failure the orchestrator had to route around. +//! +//! Note the grain, because it is easy to lose: **holds are on sessions** +//! ([`SessionControl`]), never on directories. What is per-*checkout* is the +//! serialization rule the strategy imposes +//! ([`checkout_writer`](PtySessionExecutor::checkout_writer)) — a separate rule +//! that today happens to be triggered by the same event. +//! 2. **During a turn** — the operator takes the session mid-flight: the turn +//! *suspends* ([`await_handback`](super::super::PtySessionExecutor::await_handback)) +//! rather than being discarded, keeping everything the harness has produced +//! so far and holding the task open. +//! 3. **After the hold** — control comes back: a fresh **hand-back turn** runs +//! in that same session ([`handback_prompt`]), because after minutes of +//! operator activity the suspended continuation may be stale. The session is +//! the only context that saw both the agent's partial work and the person's, +//! so its answer is the authoritative result — and it is emitted as the +//! pending task's own result, under the same task id. +//! +//! The one thing none of this does is *fail*. A dispatch that meets a person +//! ends in a real result or a real error, never in silence and never in a +//! refusal the orchestrator has to interpret. + +use std::time::Duration; + +use medulla::daemon::mappers::HarnessSemanticEvent; +use medulla::daemon::providers::{Abort, OnEvent}; +use medulla::protocol::{HarnessEvent, HarnessProvider}; + +use super::super::pty::{SessionControl, SessionRow}; +use super::types::PtySessionExecutor; + +/// How often a queued or suspended dispatch re-checks who holds the session. +/// +/// A human timescale, not a queue-depth one: nothing here is waiting on a +/// machine. Cheap enough (one atomic read per live session, plus a path compare +/// for the queue case) that a task parked behind someone's lunch break costs +/// nothing measurable. +const HOLD_POLL: Duration = Duration::from_millis(250); + +/// The queue budget for a caller that stated no idle ceiling of its own. +/// +/// Only a floor against an unbounded wait — `[host].taskTimeoutMs` is always set +/// in practice, so this is never what bounds a real dispatch. +const DEFAULT_QUEUE_BUDGET: Duration = Duration::from_secs(600); + +/// The instruction a hand-back turn is given, wrapping the task's own. +/// +/// Deliberately **not** a resumption of the suspended turn (spec §5.1): after +/// minutes of operator activity that continuation may be stale — the person may +/// have finished the work themselves, advanced it partly, or changed the tree +/// under it. So the turn is fresh, and it is told to *look* before it acts. +/// +/// One line, because it is typed into a composer. A line-oriented harness reads +/// a prompt up to the first newline, so a multi-line brief would arrive as a +/// prompt plus stray input. +pub(super) fn handback_prompt(instruction: &str) -> String { + format!( + "An operator took control of this session and has been working in it; \ + you now have it back. Review this session's history and the current \ + state of the workspace, then finish this task: {instruction} — if the \ + work is already done, do not redo it: report the final result. \ + Otherwise continue from where things now stand and complete it.", + ) +} + +/// The `status` frame text that announces a hold to the requester. +/// +/// Built from the shared prefix so the hub recognises it and pauses that +/// dispatch's no-progress watchdog for as long as the hold lasts — a person +/// reading their session is not a crashed worker. +fn held_status(provider: HarnessProvider) -> String { + format!( + "{} · the {} turn is suspended, not lost", + medulla::daemon::SESSION_HELD_STATUS_PREFIX, + provider.as_str(), + ) +} + +/// The `status` frame text that ends a hold, on the same shared-prefix contract. +fn resumed_status(provider: HarnessProvider) -> String { + format!( + "{} · the {} session is reviewing what changed", + medulla::daemon::SESSION_RESUMED_STATUS_PREFIX, + provider.as_str(), + ) +} + +/// A synthetic `status` event, so a control change reaches the peer through the +/// channel every other bit of progress already uses. +/// +/// Synthesised rather than folded from a transcript because no harness writes a +/// record for "a human took the keyboard" — it is a fact about the *session*, +/// not about the turn. [`crate::worker::executor`] is the only place that knows +/// it. +fn control_event(detail: String, state: &str) -> HarnessSemanticEvent { + HarnessSemanticEvent { + line: 0, + timestamp_ms: medulla::clock::now_millis(), + record_type: "medulla:control".to_string(), + event: HarnessEvent { + kind: "status".to_string(), + payload: serde_json::json!({ "state": state, "detail": detail }), + ..Default::default() + }, + } +} + +/// Tell the peer a person has the session. +pub(super) fn report_held(on_event: &mut Option, provider: HarnessProvider) { + if let Some(callback) = on_event.as_mut() { + callback(&control_event(held_status(provider), "held")); + } +} + +/// Tell the peer the session is back and the hand-back turn is starting. +pub(super) fn report_resumed(on_event: &mut Option, provider: HarnessProvider) { + if let Some(callback) = on_event.as_mut() { + callback(&control_event(resumed_status(provider), "running")); + } +} + +impl PtySessionExecutor { + /// The session a new one would collide with if it started in `cwd` — the + /// **serialization** rule, which is about the strategy, not about control. + /// + /// Two rules govern where a dispatch may run, and E deliberately keeps them + /// apart because they only *coincide* today: + /// + /// 1. **Candidacy** (session grain, spec §4.1) — a user-owned session is not + /// a dispatch candidate. That is decided per session, by + /// [`claim_idle`](crate::worker::pty::PtyManager::claim_idle), and says + /// nothing whatever about the session's neighbours. + /// 2. **Serialization** (strategy grain, spec §2.3) — under + /// `strategy: checkout` every session of an agent shares one working + /// tree, so the tree takes one writer at a time. That is decided per + /// *checkout*, and it is why a fresh session cannot simply start beside + /// an existing writer. + /// + /// Conflating them is what produced the behaviour this replaces: a hold on + /// one session refused the whole directory, so a person opening a session to + /// read it failed dispatches that had nothing to do with them. Under + /// `worktree` (phase G) the two separate visibly — sessions get their own + /// trees, rule 2 stops applying, and a held session will correctly imply + /// nothing at all about its siblings, with no control logic to revisit. + /// + /// **What this enforces today, and what it does not.** It prevents the one + /// pair of concurrent writers that nothing else does: the orchestrator + /// starting a harness in a tree a *person* is working in. It does **not** + /// serialize two orchestrator sessions in one checkout — main opens a second + /// session for a second concurrent dispatch (pinned by + /// `concurrent_tasks_from_one_peer_do_not_share_a_session`), and turning + /// that into a queue is a behaviour change with its own capacity story: + /// F3 (per-agent serial queue, semaphore demoted to a backstop) owns it. + /// The gap is stated rather than silently closed, because closing it here + /// would look like a control fix and would in fact be a scheduling one. + pub(super) fn checkout_writer(&self, cwd: &str) -> Option { + self.sessions + .sessions_in(cwd) + .into_iter() + .find(|row| row.control == SessionControl::User) + } + + /// How long a queued dispatch may wait for the checkout before giving up. + /// + /// The caller's own idle ceiling (`[host].taskTimeoutMs`), because that is + /// the requester's stated patience for this task and waiting for a person is + /// not different in kind from waiting for a harness. `0` means the caller + /// set no ceiling, which is never observed from `[host]` — its default is + /// nonzero — so the fallback exists only so an unset budget cannot become an + /// unbounded wait. + pub(super) fn queue_budget(&self, timeout_ms: u64) -> Duration { + if timeout_ms == 0 { + DEFAULT_QUEUE_BUDGET + } else { + Duration::from_millis(timeout_ms) + } + } + + /// Wait for the checkout at `cwd` to take a writer, or give up at + /// `deadline`. + /// + /// The queue, and the whole of what replaced the blanket refusal: the same + /// exclusivity — never two writers in one working tree — bought by waiting + /// rather than by ending the dispatch. + /// + /// Keyed on the directory rather than on one session id because the wait is + /// on the *tree*: the operator may close the session they are in and open + /// another, and the next writer is as much of a collision as the last one. + /// + /// # Errors + /// + /// - The orchestrator aborted while the task was queued — nothing has been + /// started, so there is nothing to stop. + /// - The budget ran out with the person still working. Reported with the + /// shared held prefix so the hub settles it as + /// [`RunError::Held`](medulla::hub::RunError::Held) — "I did not attempt + /// this, come back later" — rather than as a task the harness tried and + /// failed. This is the one path that still reaches that refusal, and it + /// exists so a dispatch always ends in a real answer: silence for a + /// ten-minute lunch break would be worse than a retryable no. + pub(super) async fn await_checkout_release( + &self, + cwd: &str, + abort: &Abort, + deadline: tokio::time::Instant, + ) -> Result<(), String> { + loop { + let Some(held) = self.checkout_writer(cwd) else { + return Ok(()); + }; + if abort.is_aborted() { + return Err("task aborted while queued behind an operator".to_string()); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!( + "{}: an operator is working in {cwd} ({} session {})", + medulla::daemon::HARNESS_HELD_PREFIX, + held.provider.as_str(), + held.id, + )); + } + tokio::time::sleep(HOLD_POLL).await; + } + } + + /// Suspend the turn running in `id` until the operator hands it back. + /// + /// Unbounded on purpose. A hold is a person's decision and lasts a person's + /// amount of time; the watchdogs that would otherwise reap the task are + /// *paused* for its duration rather than merely lengthened — the worker's own + /// idle ceiling by not accruing here (the caller resets it on resume), and + /// the requester's no-progress window by the held status frame this + /// suspension announces. What still ends the wait is an event, not a clock: + /// the operator hands it back, the orchestrator aborts, or the session dies. + /// + /// # Errors + /// + /// - Aborted while held. The session is left exactly as it is: it belongs to + /// the operator now, and interrupting or closing it would take their work + /// away to settle a task they are no longer part of. + /// - The session ended while the operator had it — there is no longer + /// anything to hand back or to run a hand-back turn in. + pub(super) async fn await_handback( + &self, + id: &str, + provider: HarnessProvider, + abort: &Abort, + ) -> Result<(), String> { + loop { + if self.sessions.control(id) != Some(SessionControl::User) { + return Ok(()); + } + if abort.is_aborted() { + return Err(format!( + "{} task aborted while an operator held the session", + provider.as_str(), + )); + } + if !self + .sessions + .row(id) + .is_some_and(|row| row.state.is_running()) + { + return Err(format!( + "{} session ended while an operator held it", + provider.as_str(), + )); + } + tokio::time::sleep(HOLD_POLL).await; + } + } +} diff --git a/src/tui/src/worker/executor/mod.rs b/src/tui/src/worker/executor/mod.rs index e520fc122..9c50b4100 100644 --- a/src/tui/src/worker/executor/mod.rs +++ b/src/tui/src/worker/executor/mod.rs @@ -1,8 +1,11 @@ //! PTY-backed execution of delegated harness tasks. //! //! [`PtySessionExecutor`] is the public adapter. [`run`] owns its execution -//! behavior, while [`types`] owns the executor and session-planning data. +//! behavior, [`hold`] owns what happens when an operator is in the way — queue, +//! suspend, hand back — and [`types`] owns the executor and session-planning +//! data. +mod hold; mod run; #[cfg(test)] mod tests; diff --git a/src/tui/src/worker/executor/run.rs b/src/tui/src/worker/executor/run.rs index a0e015422..5d6fd30d2 100644 --- a/src/tui/src/worker/executor/run.rs +++ b/src/tui/src/worker/executor/run.rs @@ -35,7 +35,7 @@ use medulla::sessions::{SessionClass, TurnStream}; use medulla::wrapper::tail::SessionTailer; use super::super::pty::{LaunchSpec, PtyManager, SessionControl, SessionOrigin}; -use super::types::{OpenedSession, PtySessionExecutor, SessionPlan, WorkspaceContext}; +use super::types::{OpenedSession, PtySessionExecutor, SessionPlan, TurnSpec, WorkspaceContext}; /// How often the transcript is polled while a turn runs. /// @@ -146,10 +146,26 @@ impl PtySessionExecutor { // Two steps, and the split is load-bearing: `RunTaskOptions` is `Send` // but not `Sync`, so a borrow of it held across an await would make this // future un-spawnable. Deciding what to run is synchronous and gives the - // borrow back; only the owned [`LaunchSpec`] crosses the await. - let opened = match self.session_for(&options, class)? { - SessionPlan::Reuse(opened) => opened, - SessionPlan::Launch(spec) => self.launch(*spec).await?, + // borrow back — hence the `let` before the `match`, which ends the + // borrow at the semicolon rather than at the end of the match — and only + // owned values cross the awaits. + // + // A loop, because the third answer is "wait": a checkout with a person + // in it is planned again once they are done, not refused (see + // [`SessionPlan::Queue`]). The budget is the caller's own idle ceiling, + // so a queued task cannot outlive the deadline its requester set for it. + let queue_deadline = tokio::time::Instant::now() + self.queue_budget(options.timeout_ms); + let queue_abort = options.abort.clone(); + let opened = loop { + let plan = self.session_for(&options, class)?; + match plan { + SessionPlan::Reuse(opened) => break opened, + SessionPlan::Launch(spec) => break self.launch(*spec).await?, + SessionPlan::Queue(cwd) => { + self.await_checkout_release(&cwd, &queue_abort, queue_deadline) + .await?; + } + } }; if let Some(pinned) = &opened.harness_session_id { // A reused session's transcript already exists, so the fresh-session @@ -236,14 +252,23 @@ impl PtySessionExecutor { let abort = options.abort.clone(); let on_event = options.on_event; let timeout_ms = options.timeout_ms; + // Kept for the hand-back turn: if an operator takes this session + // mid-flight, what they hand back has to be finished against the task + // that was asked for, and this is the only copy of it that survives the + // borrow rules above. + let instruction = options.prompt.clone(); let outcome = self .await_turn( &id, - (provider, gh_repo_is_set), + TurnSpec { + provider, + gh_repo_is_set, + timeout_ms, + instruction, + }, tailer, abort, on_event, - timeout_ms, ) .await; self.finish_turn(&id, class, outcome.is_ok()); @@ -314,34 +339,27 @@ impl PtySessionExecutor { ); } - /// Decide which session serves this task: reuse an idle one, or launch. + /// Decide which session serves this task: reuse an idle one, launch, or + /// queue behind the person in the checkout. /// - /// Synchronous, and returns a plan rather than a session, because the launch - /// itself must not happen here — see [`PtySessionExecutor::launch`]. + /// Synchronous, and returns a plan rather than a session, because neither + /// the launch nor the wait may happen here — see + /// [`PtySessionExecutor::launch`]. + /// + /// **Candidacy (spec §4.1).** Only *orchestrator-owned* sessions are ever + /// candidates. A user-owned session — born that way as an unmanaged spawn, + /// or taken at runtime — is not one, so a person working in a session never + /// makes a dispatch fail: it is simply not among the things the dispatch can + /// pick up. That rule lives in + /// [`try_claim`](crate::worker::pty::PtyManager::claim_idle), which is why + /// reuse is consulted *first* here now. It used to come second, behind a + /// workspace-wide refusal that turned a person at a keyboard into a task + /// error — even when the agent had another session sitting idle beside them. fn session_for( &self, options: &RunTaskOptions, class: SessionClass, ) -> Result { - // Exclusivity, and it comes first. A workspace an operator is working in - // is not available to the orchestrator at all — not "reuse nothing in - // it", but "start nothing in it". Checked ahead of the reuse branch - // below on purpose: a folder with a person in it is not shared, however - // idle some other harness sitting in it happens to look. - // - // Refused with the shared prefix rather than silently routed around, so - // the hub can settle it as `RunError::Held` and the orchestrator is told - // *why* it cannot work here — a task that vanishes into a retry with no - // reason is the failure this replaces. - if let Some(held) = self.sessions.operator_hold(&options.cwd) { - return Err(format!( - "{}: an operator is working in {} ({} session {})", - medulla::daemon::HARNESS_HELD_PREFIX, - options.cwd, - held.provider.as_str(), - held.id, - )); - } if class == SessionClass::Unbound { // Reuse this peer's session only when it is *idle*. A harness serves // one turn at a time: a fan-out that pastes three prompts into one @@ -361,6 +379,21 @@ impl PtySessionExecutor { })); } } + // Nothing to reuse, so this dispatch needs a session of its own — and + // that is where the *second*, independent rule applies: under + // `strategy: checkout` the working tree takes one writer at a time + // (see [`checkout_writer`](Self::checkout_writer)), so a fresh harness + // cannot simply start beside the one that is there. The work queues + // instead — the same exclusivity the blanket refusal used to buy, + // without ending the dispatch to get it. + // + // Note what this is *not*: it is not "the workspace is held". Holds are + // on sessions, and rule 1 above has already dealt with those. This is + // the strategy's serialization, and under `worktree` it will not apply + // at all. + if self.checkout_writer(&options.cwd).is_some() { + return Ok(SessionPlan::Queue(options.cwd.clone())); + } let label = if options.conversation.is_empty() { format!("task:{}", options.provider.as_str()) } else { @@ -516,6 +549,62 @@ impl PtySessionExecutor { Ok((env, extra_args)) } + /// Fold whatever the harness has written since the last poll, and answer + /// with the turn's result if that fold completed it. + /// + /// Shared by the polling loop and by the suspend path, and shared + /// deliberately: "read what is already there before doing anything else" has + /// to mean the same thing in both, or a turn that finished microseconds + /// before an operator took the session would have its answer read by one + /// path and dropped by the other. + /// + /// `last_line_at` is advanced per line rather than per call, because it is + /// the idle watchdog's clock and a batch of lines is progress at the time + /// each of them was read, not at the time the batch was drained. + fn fold_available( + &self, + id: &str, + provider: HarnessProvider, + tailer: &mut SessionTailer, + stream: &mut TurnStream, + on_event: &mut Option, + last_line_at: &mut i64, + ) -> Option { + let poll = tailer.poll(); + // Codex cannot be told its id, so it is learned from the rollout the + // first time the tailer locates one. + if let Some(located) = &poll.located { + self.sessions + .record_session_id(id, located.harness_session_id.clone()); + } + for line in poll.lines { + *last_line_at = medulla::clock::now_millis(); + let fold = stream.observe(&line.text); + self.workspace_context + .lock() + .expect("workspace context lock poisoned") + .insert(id.to_string(), stream.workspace_context()); + // The peer watches its task through these. Dropping them would + // leave it with an ack, silence, then a reply — which is what + // this executor used to do. + if let Some(callback) = on_event.as_mut() { + for event in &fold.events { + callback(event); + } + } + if let Some(reply) = fold.reply { + return Some(RunTaskResult { + provider, + reply, + events: stream.events(), + usage: stream.usage(), + session_id: self.sessions.row(id).and_then(|row| row.session_id), + }); + } + } + None + } + /// Poll the transcript until the harness says the turn is over. /// /// `timeout_ms` is the caller's configured idle watchdog (`[host] @@ -530,13 +619,17 @@ impl PtySessionExecutor { async fn await_turn( &self, id: &str, - mapper_context: (HarnessProvider, bool), + spec: TurnSpec, mut tailer: SessionTailer, abort: medulla::daemon::providers::Abort, mut on_event: Option, - timeout_ms: u64, ) -> Result { - let (provider, gh_repo_is_set) = mapper_context; + let TurnSpec { + provider, + gh_repo_is_set, + timeout_ms, + instruction, + } = spec; let mut stream = TurnStream::new_with_gh_repo_override(provider, gh_repo_is_set); if let Some((cwd, branch, pull_request)) = self .workspace_context @@ -552,20 +645,63 @@ impl PtySessionExecutor { callback(&event); } } - let started = tokio::time::Instant::now(); + let mut started = tokio::time::Instant::now(); let mut last_line_at = medulla::clock::now_millis(); loop { // Taking control is an ownership transfer, not merely a display - // preference. Yield before processing aborts or transcript output - // so the executor cannot send Ctrl-C, report a stale completion, or - // later close the PTY underneath the operator. + // preference, so it is answered before aborts or transcript output: + // from here the executor must not send Ctrl-C, report a stale + // completion, or close the PTY underneath the operator. + // + // What it does instead is **suspend** (spec §5). The turn used to + // return an error here, throwing away everything the harness had + // produced and telling the orchestrator its task had failed — for + // the entirely ordinary event of a person opening the session to + // look. Now the fold, its events, its usage and its workspace + // context all stay exactly where they are, the session keeps the + // work, and the task stays open. if self.sessions.control(id) == Some(SessionControl::User) { - return Err(format!( - "{}: operator took control of the {} session", - medulla::daemon::HARNESS_HELD_PREFIX, - provider.as_str() - )); + // Everything already written belongs to *this* turn — the + // takeover cannot retroactively unwrite it. Folded out before + // suspending, so a turn that finished in the instant somebody + // took the session still reports the answer it had reached. + if let Some(result) = self.fold_available( + id, + provider, + &mut tailer, + &mut stream, + &mut on_event, + &mut last_line_at, + ) { + return Ok(result); + } + super::hold::report_held(&mut on_event, provider); + self.await_handback(id, provider, &abort).await?; + // The lines the operator's own work wrote are theirs, not this + // turn's: dropped rather than folded, or the person's last + // exchange would settle the task as its answer. What they did is + // not lost — it is in the session, which is exactly what the + // hand-back turn is told to go and read. + let poll = tailer.poll(); + if let Some(located) = &poll.located { + self.sessions + .record_session_id(id, located.harness_session_id.clone()); + } + super::hold::report_resumed(&mut on_event, provider); + super::super::pty::inject_prompt( + &self.sessions, + id, + &super::hold::handback_prompt(&instruction), + ) + .await?; + // Both budgets restart with the hand-back turn, which is what + // "the watchdog is paused, not lengthened" means on this side: + // held time is excluded rather than counted, so a session held + // over lunch is not a task that timed out at the desk. + started = tokio::time::Instant::now(); + last_line_at = medulla::clock::now_millis(); + continue; } if abort.is_aborted() { if abort.is_terminated() { @@ -589,37 +725,15 @@ impl PtySessionExecutor { )); } - let poll = tailer.poll(); - // Codex cannot be told its id, so it is learned from the rollout the - // first time the tailer locates one. - if let Some(located) = &poll.located { - self.sessions - .record_session_id(id, located.harness_session_id.clone()); - } - for line in poll.lines { - last_line_at = medulla::clock::now_millis(); - let fold = stream.observe(&line.text); - self.workspace_context - .lock() - .expect("workspace context lock poisoned") - .insert(id.to_string(), stream.workspace_context()); - // The peer watches its task through these. Dropping them would - // leave it with an ack, silence, then a reply — which is what - // this executor used to do. - if let Some(callback) = on_event.as_mut() { - for event in &fold.events { - callback(event); - } - } - if let Some(reply) = fold.reply { - return Ok(RunTaskResult { - provider, - reply, - events: stream.events(), - usage: stream.usage(), - session_id: self.sessions.row(id).and_then(|row| row.session_id), - }); - } + if let Some(result) = self.fold_available( + id, + provider, + &mut tailer, + &mut stream, + &mut on_event, + &mut last_line_at, + ) { + return Ok(result); } if !tailer.is_located() && started.elapsed() > LOCATE_BUDGET { diff --git a/src/tui/src/worker/executor/types.rs b/src/tui/src/worker/executor/types.rs index 508f02e37..d0275dc2f 100644 --- a/src/tui/src/worker/executor/types.rs +++ b/src/tui/src/worker/executor/types.rs @@ -39,6 +39,34 @@ pub(super) enum SessionPlan { /// the child's whole environment — and every `Reuse` would otherwise pay /// for a launch it is not doing. Launch(Box), + /// Nothing reusable *and* a person is working in this checkout: wait for + /// them, then plan again. + /// + /// Not a failure and not a fresh harness beside theirs. Under + /// `strategy: checkout` an agent's sessions share one working tree, so + /// "create a session" cannot mean "start a second writer in it" — it means + /// queue behind the writer that is there. The workspace path is carried + /// because the wait is on the *directory*, not on any one session: the + /// operator may close theirs and open another. + Queue(String), +} + +/// Everything one turn needs to know about itself, past the session it runs in. +/// +/// A record rather than five parameters, and it earned that when the hand-back +/// turn added the fifth: the polling loop needs the task's own instruction now, +/// because an operator may take the session mid-flight and what they hand back +/// has to be finished against the work that was asked for. +pub(super) struct TurnSpec { + /// Which harness is running, for the fold's dialect and every message. + pub(super) provider: medulla::protocol::HarnessProvider, + /// Whether the child's environment selects a GitHub repository, which the + /// fold uses to resolve pull-request context. + pub(super) gh_repo_is_set: bool, + /// The caller's idle ceiling in ms (`[host].taskTimeoutMs`); `0` is none. + pub(super) timeout_ms: u64, + /// The task's own instruction, as the peer sent it. + pub(super) instruction: String, } /// Runs delegated tasks inside live harness sessions. diff --git a/src/tui/src/worker/executor_tests/control.rs b/src/tui/src/worker/executor_tests/control.rs new file mode 100644 index 000000000..e48ae0e49 --- /dev/null +++ b/src/tui/src/worker/executor_tests/control.rs @@ -0,0 +1,519 @@ +//! What a dispatch does when it meets a person: candidacy, the queue, and the +//! suspend / hand-back cycle. +//! +//! The whole of phase E's control model, exercised on one machine against the +//! fake harness. Each test pins one clause of it, and the clauses are worth +//! naming because the behaviour they replace was the opposite in every case: +//! +//! | Was | Is | +//! |---|---| +//! | a hold on one session refused the whole workspace | a held session is simply not a candidate; a sibling serves the task | +//! | nothing else available ⇒ the dispatch failed `harnessHeld` | it queues, and runs when the session comes back | +//! | a mid-turn takeover discarded the in-flight turn | the turn suspends and keeps everything it had | +//! | a held task hit the idle watchdog | held time does not accrue | +//! | a held task produced no result, ever | the hand-back turn produces one, under the original task id | +//! +//! The last row is why these are the deliverable rather than the safety net: +//! control state is no longer advertised to the backend at all, so the hand-back +//! turn is now the *only* way a held in-flight task reaches a result. A silent +//! dispatch would be an orchestrator waiting forever on a task it cannot see the +//! state of. +//! +//! These replace `a_dispatch_into_a_workspace_the_operator_holds_is_refused` and +//! `a_dispatch_runs_again_once_the_harness_is_handed_back`, which pinned the +//! refusal — the invariant those really protected (never a rival harness in the +//! operator's tree) is asserted here on the queue path instead. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use medulla::protocol::HarnessProvider; + +use super::super::pty::{LaunchSpec, PtyManager, SessionControl, SessionOrigin}; +use super::{conversational_harness_script, fake_harness_script, harness, options}; + +/// How long to let a suspended or queued dispatch prove it is *not* finishing. +/// +/// Long enough to outlast the executor's own 250 ms hold poll several times +/// over, short enough that four of these do not dominate the suite. +const NOT_YET: Duration = Duration::from_millis(900); + +/// The end-to-end deadline for a dispatch that should settle. +const SETTLES: Duration = Duration::from_secs(30); + +/// Flatten one emulated terminal screen into searchable text. +fn screen_text(sessions: &PtyManager, id: &str) -> String { + sessions + .screen_rows(id) + .map(|snapshot| { + snapshot + .cells + .iter() + .map(|row| { + row.iter() + .map(|cell| cell.text.as_str()) + .collect::() + }) + .collect::>() + .join("\n") + }) + .unwrap_or_default() +} + +/// Spin until `check` passes, or fail. +async fn wait_for(what: &str, mut check: impl FnMut() -> bool) { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while std::time::Instant::now() < deadline { + if check() { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("timed out waiting for: {what}"); +} + +/// Open a session the way an operator does: theirs from birth, and running +/// whatever `script` says. +fn operator_session( + sessions: &PtyManager, + cwd: &str, + env: &HashMap, + script: &str, +) -> String { + sessions + .open(LaunchSpec { + provider: HarnessProvider::Codex, + preset: None, + bin: "/bin/sh".to_string(), + cwd: cwd.to_string(), + env: env.clone(), + extra_args: vec!["-c".to_string(), script.to_string()], + skip_permissions: false, + label: "you:codex".to_string(), + model: None, + session_id: None, + control: SessionControl::User, + origin: SessionOrigin::User, + name: None, + mcp_grant_session: None, + }) + .expect("the operator's session must start") +} + +/// A harness that answers the *hand-back* turn and nothing before it. +/// +/// Modelled on the real sequence: the first prompt starts a turn that never +/// finishes (the operator interrupts it by taking the session), everything the +/// person then types is read and echoed but settles nothing, and the turn is +/// only completed by the prompt the executor injects on hand-back — which is +/// matched by its own wording, so a test cannot pass by accident on the +/// operator's typing. +fn handback_harness_script(rollout: &str, cwd: &str, reply: &str) -> String { + format!( + r#" +printf 'ready\r\n' +read -r first +printf 'started: %s\r\n' "$first" +printf '{{"type":"session_meta","payload":{{"session_id":"sess-handback","cwd":"{cwd}"}}}}\n' >> '{rollout}' +printf '{{"type":"event_msg","payload":{{"type":"task_started","turn_id":"t1"}}}}\n' >> '{rollout}' +printf '{{"type":"event_msg","payload":{{"type":"agent_message","message":"halfway through the migration","phase":"main"}}}}\n' >> '{rollout}' +while read -r line; do + printf 'read: %s\r\n' "$line" + case "$line" in + *"you now have it back"*) + printf '{{"type":"event_msg","payload":{{"type":"task_started","turn_id":"t2"}}}}\n' >> '{rollout}' + printf '{{"type":"event_msg","payload":{{"type":"task_complete","turn_id":"t2","last_agent_message":"{reply}"}}}}\n' >> '{rollout}' + ;; + esac +done +"# + ) +} + +#[tokio::test] +async fn a_held_session_is_skipped_and_a_sibling_orchestrator_session_serves_the_task() { + // Candidacy, at session grain (spec §4.1). The conversation already has a + // session of its own, idle and the orchestrator's. The operator then opens a + // session of their own in the same checkout and keeps it. The next dispatch + // must go to the sibling and leave the person alone — where before, the hold + // on *their* session refused the whole workspace, and this task failed with + // an idle harness of its own sitting right beside it. + let dir = tempfile::tempdir().unwrap(); + let cwd = dir.path().to_string_lossy().into_owned(); + let rollout = dir.path().join("rollout-sibling.jsonl"); + let script = conversational_harness_script(&rollout.to_string_lossy(), &cwd); + let (executor, env) = harness(dir.path(), &cwd); + let sessions = executor.sessions_for_test(); + + // The sibling, made the way one really appears: a turn ran in it and + // released it. + let first = tokio::time::timeout( + SETTLES, + executor + .clone() + .run_for_test(options(&env, "peer-bob", &script, &cwd)), + ) + .await + .expect("the first turn settles") + .expect("the first turn succeeds"); + assert_eq!(first.reply, "answer 1"); + let sibling = sessions + .rows() + .into_iter() + .find(|row| row.state.is_running()) + .expect("the conversation's session stays open") + .id; + + // Now a person opens their own session in the same checkout, and keeps it. + let held = operator_session(&sessions, &cwd, &env, "sleep 30"); + wait_for("the operator's session running", || { + sessions.row(&held).is_some_and(|r| r.state.is_running()) + }) + .await; + + let second = tokio::time::timeout( + SETTLES, + executor + .clone() + .run_for_test(options(&env, "peer-bob", &script, &cwd)), + ) + .await + .expect("the dispatch must settle") + .expect("a person in one session must not fail a dispatch"); + + assert_eq!( + second.reply, "answer 2", + "the task must be served by the orchestrator's own session" + ); + assert_eq!( + sessions.rows().len(), + 2, + "reuse, not a third harness: {:?}", + sessions + .rows() + .iter() + .map(|r| r.label.clone()) + .collect::>() + ); + let held_row = sessions + .row(&held) + .expect("the operator keeps their session"); + assert_eq!( + held_row.control, + SessionControl::User, + "a dispatch must never take a session out from under a person" + ); + assert!( + !held_row.busy, + "a held session must not even be claimed, let alone written into" + ); + assert!( + sessions + .row(&sibling) + .is_some_and(|row| row.state.is_running()), + "the session that served the task is the one that was already there" + ); + + sessions.shutdown(); +} + +#[tokio::test] +async fn with_only_a_held_session_the_work_queues_and_runs_on_hand_back() { + // Serialization, at strategy grain (spec §2.3). Nothing to reuse and a + // person writing in the checkout: under `strategy: checkout` the tree takes + // one writer at a time, so the dispatch cannot start a rival harness beside + // them — and no longer fails instead. It waits, and the moment the session + // comes back it runs. + let dir = tempfile::tempdir().unwrap(); + let cwd = dir.path().to_string_lossy().into_owned(); + let (executor, env) = harness(dir.path(), &cwd); + let sessions = executor.sessions_for_test(); + + let held = operator_session(&sessions, &cwd, &env, "sleep 30"); + wait_for("the operator's session running", || { + sessions.row(&held).is_some_and(|r| r.state.is_running()) + }) + .await; + + let rollout = dir.path().join("rollout-queued.jsonl"); + let script = fake_harness_script( + &rollout.to_string_lossy(), + &cwd, + "ran once the tree was free", + ); + let before = sessions.rows().len(); + let mut run = tokio::spawn({ + let (executor, env, script, cwd) = (executor.clone(), env.clone(), script, cwd.clone()); + // Bounded — an agent-targeted dispatch, which never reuses and therefore + // always needs a session of its own. + async move { + executor + .run_for_test(options(&env, "", &script, &cwd)) + .await + } + }); + + let waiting = tokio::time::timeout(NOT_YET, &mut run).await; + assert!( + waiting.is_err(), + "the dispatch must queue rather than settle: {waiting:?}" + ); + assert_eq!( + sessions.rows().len(), + before, + "queuing must not open a second writer in the operator's checkout" + ); + + // The operator hands their session back; the queue drains. + assert!(sessions.set_control(&held, SessionControl::Orchestrator)); + let result = tokio::time::timeout(SETTLES, run) + .await + .expect("the queued dispatch must settle once the tree is free") + .expect("no panic") + .expect("a queued dispatch must run, not fail"); + assert!( + result.reply.contains("ran once the tree was free"), + "got: {result:?}" + ); + + sessions.shutdown(); +} + +#[tokio::test] +async fn a_queue_that_outlives_its_budget_fails_loudly_rather_than_silently() { + // The other half of the queue, and the reason the shared held prefix still + // exists. Control state is not advertised any more, so an orchestrator + // cannot see that a task is parked behind a person — which makes waiting + // forever indistinguishable from losing the task. The wait is therefore + // bounded by the caller's own idle ceiling and ends in a real, retryable + // error the hub settles as `RunError::Held`. + let dir = tempfile::tempdir().unwrap(); + let cwd = dir.path().to_string_lossy().into_owned(); + let (executor, env) = harness(dir.path(), &cwd); + let sessions = executor.sessions_for_test(); + + let held = operator_session(&sessions, &cwd, &env, "sleep 30"); + wait_for("the operator's session running", || { + sessions.row(&held).is_some_and(|r| r.state.is_running()) + }) + .await; + + let script = fake_harness_script( + &dir.path().join("rollout-never.jsonl").to_string_lossy(), + &cwd, + "never runs", + ); + let mut opts = options(&env, "", &script, &cwd); + opts.timeout_ms = 400; + + let error = tokio::time::timeout(SETTLES, executor.clone().run_for_test(opts)) + .await + .expect("the dispatch must settle") + .expect_err("a queue that never drains must end in an error"); + + assert!( + error.starts_with(medulla::daemon::HARNESS_HELD_PREFIX), + "the refusal must carry the shared prefix so the hub settles it as Held \ + — retryable, not a task the harness attempted and failed: {error}" + ); + assert_eq!( + sessions.rows().len(), + 1, + "giving up must not leave a harness in the operator's tree" + ); + + sessions.shutdown(); +} + +#[tokio::test] +async fn a_mid_turn_takeover_suspends_the_turn_and_hands_back_its_result() { + // The centre of the phase, and one test because it is one story: the + // operator takes a running session, the turn suspends instead of being + // discarded, they work in it, they hand it back, and the *same* dispatch + // answers — from the same session, under the same task id, because it never + // stopped being the same call. + let dir = tempfile::tempdir().unwrap(); + let cwd = dir.path().to_string_lossy().into_owned(); + let rollout = dir.path().join("rollout-suspend.jsonl"); + let script = handback_harness_script( + &rollout.to_string_lossy(), + &cwd, + "finished after the operator handed it back", + ); + + let (executor, env) = harness(dir.path(), &cwd); + let sessions = executor.sessions_for_test(); + let (session_tx, session_rx) = tokio::sync::oneshot::channel(); + // The status details the peer is shown. The control markers among them are + // what pause and resume the requester's no-progress watchdog, so the worker + // emitting them is half of a contract whose other half lives in + // `hub::tests::held_watchdog` — and two halves that only ever see their own + // side of a text prefix are two halves that drift. + let reported: Arc>> = Arc::new(Mutex::new(Vec::new())); + let mut opts = options(&env, "peer-bob", &script, &cwd); + opts.on_event = Some({ + let reported = reported.clone(); + Box::new( + move |event: &medulla::daemon::mappers::HarnessSemanticEvent| { + if let Some(detail) = medulla::daemon::status_detail(&event.event) { + reported.lock().unwrap().push(detail); + } + }, + ) + }); + opts.on_session = Some(Box::new(move |id| { + let _ = session_tx.send(id); + })); + let mut run = tokio::spawn(executor.clone().run_for_test(opts)); + + let id = tokio::time::timeout(Duration::from_secs(10), session_rx) + .await + .expect("the executor reports its session") + .expect("the report channel stays open"); + wait_for("the delegated turn to be under way", || { + screen_text(&sessions, &id).contains("halfway through the migration") + || std::fs::read_to_string(&rollout) + .is_ok_and(|text| text.contains("halfway through the migration")) + }) + .await; + + // The operator takes it, mid-turn. + assert!(sessions.set_control(&id, SessionControl::User)); + + let suspended = tokio::time::timeout(NOT_YET, &mut run).await; + assert!( + suspended.is_err(), + "a takeover must suspend the turn, not end it: {suspended:?}" + ); + let row = sessions + .row(&id) + .expect("the session survives the takeover"); + assert!( + row.state.is_running(), + "the operator's session must not be closed under them" + ); + assert_eq!(row.control, SessionControl::User); + assert!( + row.busy, + "the turn is suspended, not abandoned — the session still owes an answer" + ); + + // …and works in it. Their input is read by the same harness and settles + // nothing: only the hand-back prompt completes the turn. + sessions + .write(&id, b"i fixed the migration myself\r") + .expect("the operator can type into the session they hold"); + wait_for("the operator's input to reach the harness", || { + screen_text(&sessions, &id).contains("read: i fixed the migration myself") + }) + .await; + assert!( + !run.is_finished(), + "the operator working in the session must not settle the task" + ); + + // Hand-back: a fresh turn in that same session produces the task's result. + assert!(sessions.set_control(&id, SessionControl::Orchestrator)); + let result = tokio::time::timeout(SETTLES, run) + .await + .expect("the hand-back turn must settle the task") + .expect("no panic") + .expect("a hand-back must produce a result, not an error"); + + assert!( + result + .reply + .contains("finished after the operator handed it back"), + "the result must come from the hand-back turn: {result:?}" + ); + assert_eq!( + result.session_id.as_deref(), + Some("sess-handback"), + "the hand-back turn runs in the session that saw both the agent's work \ + and the operator's — it is the only context that saw either" + ); + assert_eq!( + sessions.rows().len(), + 1, + "no second session was opened for the hand-back" + ); + assert!( + result.events > 0, + "the suspended turn's fold is retained across the hold, not restarted" + ); + let reported = reported.lock().unwrap().clone(); + assert!( + reported + .iter() + .any(|detail| detail.starts_with(medulla::daemon::SESSION_HELD_STATUS_PREFIX)), + "the hold must be announced to the requester — it is what pauses the \ + no-progress watchdog: {reported:?}" + ); + assert!( + reported + .iter() + .any(|detail| detail.starts_with(medulla::daemon::SESSION_RESUMED_STATUS_PREFIX)), + "and the hand-back must end it, or the watchdog never resumes: {reported:?}" + ); + + sessions.shutdown(); +} + +#[tokio::test] +async fn the_idle_watchdog_does_not_fire_across_a_hold() { + // §5: "watchdog paused while control = user; resumes on hand-back". The + // caller's ceiling here is a fraction of the hold, so an unpaused clock + // would kill the task — and would kill it *while a person was working in + // the session*, which is the worst possible moment to close a pty. + let dir = tempfile::tempdir().unwrap(); + let cwd = dir.path().to_string_lossy().into_owned(); + let rollout = dir.path().join("rollout-watchdog.jsonl"); + let script = handback_harness_script(&rollout.to_string_lossy(), &cwd, "survived the hold"); + + let (executor, env) = harness(dir.path(), &cwd); + let sessions = executor.sessions_for_test(); + let (session_tx, session_rx) = tokio::sync::oneshot::channel(); + let mut opts = options(&env, "peer-bob", &script, &cwd); + // A ceiling far shorter than the hold below. `[host].taskTimeoutMs` is ten + // minutes in the field; the ratio is what is under test, not the number. + opts.timeout_ms = 500; + opts.on_session = Some(Box::new(move |id| { + let _ = session_tx.send(id); + })); + let run = tokio::spawn(executor.clone().run_for_test(opts)); + + let id = tokio::time::timeout(Duration::from_secs(10), session_rx) + .await + .expect("the executor reports its session") + .expect("the report channel stays open"); + wait_for("the delegated turn to be under way", || { + std::fs::read_to_string(&rollout).is_ok_and(|text| text.contains("task_started")) + }) + .await; + assert!(sessions.set_control(&id, SessionControl::User)); + + // Six times the ceiling, held. + tokio::time::sleep(Duration::from_millis(3_000)).await; + assert!( + !run.is_finished(), + "a held task must neither time out nor settle" + ); + assert!( + sessions.row(&id).is_some_and(|row| row.state.is_running()), + "an idle-timeout would have stopped the harness the operator is using" + ); + + assert!(sessions.set_control(&id, SessionControl::Orchestrator)); + let result = tokio::time::timeout(SETTLES, run) + .await + .expect("the hand-back turn must settle") + .expect("no panic") + .expect("held time must not count against the idle ceiling"); + assert!( + result.reply.contains("survived the hold"), + "got: {result:?}" + ); + + sessions.shutdown(); +} diff --git a/src/tui/src/worker/executor_tests/mod.rs b/src/tui/src/worker/executor_tests/mod.rs index e2b0e8301..5e7bf1ec5 100644 --- a/src/tui/src/worker/executor_tests/mod.rs +++ b/src/tui/src/worker/executor_tests/mod.rs @@ -21,6 +21,7 @@ use super::executor::PtySessionExecutor; use super::pty::PtyManager; mod basic; +mod control; mod live; mod plumbing; mod sessions; diff --git a/src/tui/src/worker/executor_tests/sessions.rs b/src/tui/src/worker/executor_tests/sessions.rs index 8543c5063..043180d25 100644 --- a/src/tui/src/worker/executor_tests/sessions.rs +++ b/src/tui/src/worker/executor_tests/sessions.rs @@ -353,129 +353,8 @@ async fn sequential_task_frames_from_one_sender_do_not_share_a_session() { sessions.shutdown(); } -#[tokio::test] -async fn a_dispatch_into_a_workspace_the_operator_holds_is_refused() { - // The bug this closes: taking over the only harness in a workspace did not - // stop the orchestrator working there — `session_for` fell through the reuse - // branch and simply OPENED A SECOND HARNESS in the same folder. Two agents, - // one working tree, no mutual exclusion. So the assertion that matters here - // is not just that the task is refused; it is that nothing new was spawned. - use super::super::pty::{LaunchSpec, SessionControl}; - - let dir = tempfile::tempdir().unwrap(); - let cwd = dir.path().to_string_lossy().into_owned(); - let rollout = dir.path().join("rollout-held.jsonl"); - let script = fake_harness_script(&rollout.to_string_lossy(), &cwd, "unreachable"); - - let (executor, env) = harness(dir.path(), &cwd); - let sessions = executor.sessions_for_test(); - - // The operator starts a harness here and keeps it. - let held = sessions - .open(LaunchSpec { - provider: HarnessProvider::Codex, - preset: None, - bin: "/bin/sh".to_string(), - cwd: cwd.clone(), - env: HashMap::new(), - extra_args: vec!["-c".to_string(), "sleep 30".to_string()], - skip_permissions: false, - label: "you:codex".to_string(), - model: None, - session_id: None, - control: SessionControl::User, - origin: crate::worker::pty::SessionOrigin::User, - name: None, - mcp_grant_session: None, - }) - .expect("the operator's harness must start"); - let before = sessions.rows().len(); - - let error = tokio::time::timeout( - Duration::from_secs(30), - executor - .clone() - .run_for_test(options(&env, "peer-bob", &script, &cwd)), - ) - .await - .expect("must settle") - .expect_err("a workspace an operator holds must refuse the task"); - - assert!( - error.starts_with(medulla::daemon::HARNESS_HELD_PREFIX), - "the refusal must carry the shared prefix so the hub can settle it as \ - Held rather than as an ordinary worker error — got: {error}" - ); - assert_eq!( - sessions.rows().len(), - before, - "refusing the task must not leave a rival harness running in the \ - operator's working tree" - ); - - sessions.close(&held); - sessions.shutdown(); -} - -#[tokio::test] -async fn a_dispatch_runs_again_once_the_harness_is_handed_back() { - // The other half: the hold is a pause, not a wall. Handing back must make - // the workspace usable again without the operator restarting anything. - use super::super::pty::{LaunchSpec, SessionControl}; - - let dir = tempfile::tempdir().unwrap(); - let cwd = dir.path().to_string_lossy().into_owned(); - let rollout = dir.path().join("rollout-handback.jsonl"); - let script = fake_harness_script(&rollout.to_string_lossy(), &cwd, "picked it up"); - - let (executor, env) = harness(dir.path(), &cwd); - let sessions = executor.sessions_for_test(); - - // The operator's harness runs the fake agent, not a bare shell: after the - // handback the executor REUSES this very process, so it has to be something - // that can actually serve the turn. - let held = sessions - .open(LaunchSpec { - provider: HarnessProvider::Codex, - preset: None, - bin: "/bin/sh".to_string(), - cwd: cwd.clone(), - env: env.clone(), - extra_args: vec!["-c".to_string(), script.clone()], - skip_permissions: false, - label: "you:codex".to_string(), - model: None, - session_id: None, - control: SessionControl::User, - origin: crate::worker::pty::SessionOrigin::User, - name: None, - mcp_grant_session: None, - }) - .expect("the operator's harness must start"); - - assert!(tokio::time::timeout( - Duration::from_secs(30), - executor - .clone() - .run_for_test(options(&env, "peer-bob", &script, &cwd)), - ) - .await - .expect("must settle") - .is_err()); - - // The operator hands it back. - assert!(sessions.set_control(&held, SessionControl::Orchestrator)); - - let reply = tokio::time::timeout( - Duration::from_secs(30), - executor - .clone() - .run_for_test(options(&env, "peer-bob", &script, &cwd)), - ) - .await - .expect("must settle") - .expect("a handed-back workspace must accept work again"); - assert!(reply.reply.contains("picked it up"), "got: {reply:?}"); - - sessions.shutdown(); -} +// Dispatch into a checkout an operator is working in is covered by +// [`super::control`], which owns the whole control model: a held session is not +// a dispatch candidate, a dispatch with nothing else available queues rather +// than failing, and a queue that outlives its budget ends in a real error. The +// two tests that used to live here asserted the refusal those replaced. diff --git a/src/tui/src/worker/pty/manager/session.rs b/src/tui/src/worker/pty/manager/session.rs index 3824d91a2..2dfa684ae 100644 --- a/src/tui/src/worker/pty/manager/session.rs +++ b/src/tui/src/worker/pty/manager/session.rs @@ -52,31 +52,32 @@ impl PtyManager { Some(session.row()) } - /// The operator-held session covering `cwd`, if there is one. + /// Every live session working in `cwd`, with who holds each. /// - /// The half of exclusivity [`claim_idle`](Self::claim_idle) cannot express. - /// `claim_idle` answers "may I reuse *this* session", which a caller can walk - /// straight past by opening a second harness in the same folder — and did: - /// taking over the only harness in a workspace made the next task frame spawn - /// a rival process in the same working tree, two agents editing one repo with - /// no mutual exclusion at all. This answers the question that actually - /// governs: "may anything start here right now". + /// A neutral question — "what is running in this directory" — and + /// deliberately not a policy. It replaced `operator_hold(cwd)`, which asked + /// "is this *workspace* held" and answered by scanning for a user-held + /// session in it. That was an artifact of the model where an agent had one + /// implicit session, in which "held session" and "held workspace" were the + /// same sentence. They are not: a **hold is on a session**, never on a + /// directory, and an agent now has as many sessions as its strategy allows. /// - /// A workspace with a person in it is not shared, however idle another - /// harness in it looks, so this is consulted *before* reuse rather than after. + /// What remains true is that sessions sharing one *checkout* share one + /// working tree, so how many of them may write at once is a property of the + /// agent's [`WorkspaceStrategy`](medulla::runtime::WorkspaceStrategy) — a + /// separate rule, applied by the executor, that this only supplies the facts + /// for. Under `worktree` (phase G) sessions in one declared workspace get + /// their own trees and the rule changes without this query changing at all. /// /// Walks a cloned list of handles rather than holding the registry, because /// `same_workspace` canonicalizes both paths and that is a filesystem call — /// exactly the blocking work no caller may hold the registry across. - pub fn operator_hold(&self, cwd: &str) -> Option { + pub fn sessions_in(&self, cwd: &str) -> Vec { self.handles() .into_iter() - .find(|session| { - session.control() == SessionControl::User - && session.is_running() - && same_workspace(session.cwd(), cwd) - }) + .filter(|session| session.is_running() && same_workspace(session.cwd(), cwd)) .map(|session| session.row()) + .collect() } /// Who currently holds `id`, if it is a session we know about. diff --git a/src/tui/src/worker/pty/tests/control.rs b/src/tui/src/worker/pty/tests/control.rs index 2299374f6..4dc35214c 100644 --- a/src/tui/src/worker/pty/tests/control.rs +++ b/src/tui/src/worker/pty/tests/control.rs @@ -164,7 +164,12 @@ fn user_sh_in(script: &str, cwd: &std::path::Path) -> LaunchSpec { } #[test] -fn operator_hold_reports_the_session_holding_a_workspace() { +fn sessions_in_reports_what_is_running_in_a_directory_and_who_holds_it() { + // A neutral question, and the replacement for the old `operator_hold(cwd)`. + // That one asked "is this *workspace* held" — an artifact of the model where + // an agent had one implicit session. A hold is on a session; a directory + // only ever has sessions *in* it, and what that implies is the strategy's + // business, not control's. let dir = tempfile::tempdir().unwrap(); let manager = PtyManager::new(); let id = manager.open(user_sh_in("sleep 30", dir.path())).unwrap(); @@ -172,19 +177,24 @@ fn operator_hold_reports_the_session_holding_a_workspace() { manager.row(&id).is_some_and(|r| r.state.is_running()) }); - let held = manager - .operator_hold(&dir.path().to_string_lossy()) - .expect("a workspace the operator is working in must report its hold"); - assert_eq!(held.id, id); + let running = manager.sessions_in(&dir.path().to_string_lossy()); + assert_eq!(running.len(), 1); + assert_eq!(running[0].id, id); + assert_eq!( + running[0].control, + SessionControl::User, + "the answer carries who holds each session rather than filtering on it" + ); manager.close(&id); } #[test] -fn operator_hold_ignores_a_session_the_orchestrator_holds() { - // The whole point of the check: it must gate on *who holds it*, not on - // "is there a harness here". An orchestrator session in a folder is not a - // reason to refuse the orchestrator work in that folder. +fn sessions_in_lists_an_orchestrator_session_too() { + // The old query dropped these, because it was really asking "may anything + // start here". This one reports them, so the caller applying the checkout's + // one-writer rule can see every writer rather than only the human one — the + // seam F3 needs to serialize orchestrator sessions without a new query. let dir = tempfile::tempdir().unwrap(); let manager = PtyManager::new(); let spec = LaunchSpec { @@ -196,13 +206,15 @@ fn operator_hold_ignores_a_session_the_orchestrator_holds() { manager.row(&id).is_some_and(|r| r.state.is_running()) }); - assert_eq!(manager.operator_hold(&dir.path().to_string_lossy()), None); + let running = manager.sessions_in(&dir.path().to_string_lossy()); + assert_eq!(running.len(), 1); + assert_eq!(running[0].control, SessionControl::Orchestrator); manager.close(&id); } #[test] -fn operator_hold_ignores_a_workspace_nobody_is_in() { +fn sessions_in_does_not_leak_across_directories() { let dir = tempfile::tempdir().unwrap(); let other = tempfile::tempdir().unwrap(); let manager = PtyManager::new(); @@ -211,19 +223,20 @@ fn operator_hold_ignores_a_workspace_nobody_is_in() { manager.row(&id).is_some_and(|r| r.state.is_running()) }); - assert_eq!( - manager.operator_hold(&other.path().to_string_lossy()), - None, - "a hold must not leak across workspaces" + assert!( + manager + .sessions_in(&other.path().to_string_lossy()) + .is_empty(), + "a session must not be reported in a directory it is not running in" ); manager.close(&id); } #[test] -fn operator_hold_releases_when_the_session_exits() { - // A dead harness cannot be handed back, so a hold that outlived its process - // would wedge the workspace shut with no way to reopen it. +fn sessions_in_forgets_a_session_that_exited() { + // A dead harness writes nothing, so counting it as a writer would wedge the + // checkout shut with no way to reopen it. let dir = tempfile::tempdir().unwrap(); let manager = PtyManager::new(); let id = manager.open(user_sh_in("true", dir.path())).unwrap(); @@ -231,15 +244,17 @@ fn operator_hold_releases_when_the_session_exits() { manager.row(&id).is_some_and(|r| !r.state.is_running()) }); - assert_eq!(manager.operator_hold(&dir.path().to_string_lossy()), None); + assert!(manager + .sessions_in(&dir.path().to_string_lossy()) + .is_empty()); manager.close(&id); } #[test] -fn operator_hold_matches_the_same_directory_written_two_ways() { +fn sessions_in_matches_the_same_directory_written_two_ways() { // The two sides arrive by different routes — an operator-spawned harness had - // its path expanded, a task frame's cwd is verbatim — so the hold has to + // its path expanded, a task frame's cwd is verbatim — so the match has to // survive a trailing slash and a symlinked path. Exclusivity that can be // defeated by spelling is not exclusivity. let dir = tempfile::tempdir().unwrap(); @@ -259,113 +274,12 @@ fn operator_hold_matches_the_same_directory_written_two_ways() { format!("{}/", real.to_string_lossy()), link.to_string_lossy().into_owned(), ] { - assert!( - manager.operator_hold(&spelling).is_some(), - "the hold was lost when the path was written as {spelling}" + assert_eq!( + manager.sessions_in(&spelling).len(), + 1, + "the session was lost when the path was written as {spelling}" ); } manager.close(&id); } - -// ------------------------------------------------------------ provenance --- - -#[test] -fn taking_a_dispatched_session_and_handing_it_back_never_changes_its_origin() { - // The distinction this whole field exists for. Control is a question about - // *now* and moves with every takeover; origin is a fact about how the - // session was born and moves never. A rail that read them as one thing would - // lose a dispatched session out of its agent's group the moment an operator - // pressed ctrl-g on it. - let manager = PtyManager::new(); - let id = manager.open(sh("sleep 30")).unwrap(); - let row = manager.row(&id).unwrap(); - assert!(row.origin.is_orchestrator()); - assert_eq!(row.control, SessionControl::Orchestrator); - - assert!(manager.set_control(&id, SessionControl::User)); - let taken = manager.row(&id).unwrap(); - assert_eq!(taken.control, SessionControl::User, "the operator holds it"); - assert!( - taken.origin.is_orchestrator(), - "holding a session is not having started it" - ); - - assert!(manager.set_control(&id, SessionControl::Orchestrator)); - assert!( - manager.row(&id).unwrap().origin.is_orchestrator(), - "handing it back changes control, and only control" - ); - - manager.close(&id); -} - -#[test] -fn handing_an_operator_started_session_to_the_orchestrator_keeps_it_user_originated() { - // The mirror case, and the one that makes the two axes visibly independent: - // the session becomes dispatchable — a control fact — while still being one - // a person started. - let manager = PtyManager::new(); - let id = manager.open(user_sh("sleep 30")).unwrap(); - wait_for("session running", || { - manager.row(&id).is_some_and(|r| r.state.is_running()) - }); - - assert!(manager.set_control(&id, SessionControl::Orchestrator)); - let row = manager.row(&id).unwrap(); - assert!(row.origin.is_user(), "origin is fixed at birth"); - assert_eq!(row.control, SessionControl::Orchestrator); - assert!( - manager - .claim_idle("test", HarnessProvider::Codex) - .is_some_and(|claimed| claimed.id == id), - "a handed-over session is dispatchable however it was started" - ); - - manager.close(&id); -} - -#[test] -fn a_session_name_round_trips_and_a_blank_one_clears_it() { - // The name is the operator's label for a session they spun up; it is display - // identity only, and it never touches provenance or control. - let manager = PtyManager::new(); - let id = manager - .open(LaunchSpec { - name: Some("debug login".to_string()), - ..user_sh("sleep 30") - }) - .unwrap(); - assert_eq!( - manager.row(&id).unwrap().name.as_deref(), - Some("debug login") - ); - - assert!(manager.set_name(&id, Some("chasing the 500".to_string()))); - let renamed = manager.row(&id).unwrap(); - assert_eq!(renamed.name.as_deref(), Some("chasing the 500")); - assert!(renamed.origin.is_user(), "renaming is not re-parenting"); - assert_eq!(renamed.control, SessionControl::User); - - // Blank is not a name: storing it would render as a gap in the rail. - assert!(manager.set_name(&id, Some(" ".to_string()))); - assert_eq!(manager.row(&id).unwrap().name, None); - assert!(manager.set_name(&id, Some("back".to_string()))); - assert!(manager.set_name(&id, None)); - assert_eq!(manager.row(&id).unwrap().name, None); - - assert!(!manager.set_name("w_nope", Some("ghost".to_string()))); - - manager.close(&id); -} - -#[test] -fn a_dispatched_session_is_born_unnamed() { - // Nothing to name it: the UI labels an orchestrator-originated session from - // the task it was created for, which is why this stays `None` rather than - // getting a synthetic string here. - let manager = PtyManager::new(); - let id = manager.open(sh("sleep 30")).unwrap(); - assert_eq!(manager.row(&id).unwrap().name, None); - manager.close(&id); -} diff --git a/src/tui/tests/e2e_session_takeover.rs b/src/tui/tests/e2e_session_takeover.rs index e0e1887f3..b5721cc3a 100644 --- a/src/tui/tests/e2e_session_takeover.rs +++ b/src/tui/tests/e2e_session_takeover.rs @@ -1,10 +1,20 @@ -//! End-to-end coverage for taking a live session back from the orchestrator. +//! End-to-end coverage for taking a live session back from the orchestrator — +//! and giving it back again. //! //! A deterministic shell stands in for Codex, but everything around it is the //! production path: a task opens a real PTY, its prompt is injected, its rollout //! is tailed, and the operator takes control while the turn is still running. //! The same PTY must then accept the operator's input instead of being kept or //! closed by the task executor. +//! +//! What the takeover must *not* do is end the task. It used to: the executor +//! returned `harness held by operator` the moment control flipped, discarding +//! everything the turn had produced and telling the orchestrator its work had +//! failed — for the entirely ordinary event of a person opening the session to +//! look at it. The turn now suspends, the session keeps it, and the hand-back +//! runs a fresh turn *in that same session* whose answer is emitted as the +//! task's own result. That path is the only way a held task ever reaches a +//! result, so it is exercised end to end here rather than only in unit tests. #![cfg(unix)] @@ -55,8 +65,14 @@ fn screen_text(sessions: &PtyManager, id: &str) -> String { async fn taking_back_a_running_codex_session_yields_the_same_pty_to_the_operator() { let temp = tempfile::tempdir().expect("a temporary workspace"); let cwd = temp.path().to_string_lossy().into_owned(); - let rollout = temp.path().join("rollout.jsonl"); + // `rollout-*.jsonl` — codex transcript discovery matches on that prefix, and + // a file named otherwise is invisible to the tailer. + let rollout = temp.path().join("rollout-takeover.jsonl"); let rollout = rollout.to_string_lossy().into_owned(); + // Reads for ever, so the operator's typing and the hand-back prompt arrive + // through the same channel a real session would deliver them on. Only the + // hand-back prompt — matched by its own wording — completes a turn, so + // nothing here can settle the task by accident. let script = format!( r#" printf 'codex ready\r\n' @@ -64,9 +80,15 @@ read -r prompt printf 'task started: %s\r\n' "$prompt" printf '{{"type":"session_meta","payload":{{"session_id":"codex-takeover-e2e","cwd":"{cwd}"}}}}\n' >> '{rollout}' printf '{{"type":"event_msg","payload":{{"type":"task_started","turn_id":"turn-1"}}}}\n' >> '{rollout}' -read -r operator_input -printf 'operator typed: %s\r\n' "$operator_input" -sleep 30 +while read -r line; do + printf 'operator typed: %s\r\n' "$line" + case "$line" in + *"you now have it back"*) + printf '{{"type":"event_msg","payload":{{"type":"task_started","turn_id":"turn-2"}}}}\n' >> '{rollout}' + printf '{{"type":"event_msg","payload":{{"type":"task_complete","turn_id":"turn-2","last_agent_message":"reviewed the session and finished it"}}}}\n' >> '{rollout}' + ;; + esac +done "# ); @@ -119,20 +141,22 @@ sleep 30 .await; assert!(sessions.set_control(&id, SessionControl::User)); - let outcome = tokio::time::timeout(Duration::from_secs(2), run) - .await - .expect("the executor yields promptly when the operator takes control") - .expect("the executor task does not panic"); - let error = outcome.expect_err("the delegated turn is yielded, not reported complete"); + let mut run = run; + let suspended = tokio::time::timeout(Duration::from_secs(2), &mut run).await; assert!( - error.starts_with(medulla::daemon::HARNESS_HELD_PREFIX), - "unexpected takeover error: {error}" + suspended.is_err(), + "the turn suspends when the operator takes control; it is not discarded \ + and reported as a failure: {suspended:?}" ); let row = sessions.row(&id).expect("the taken-over session remains"); assert!(row.state.is_running(), "the taken-over PTY must stay alive"); assert_eq!(row.control, SessionControl::User); - assert!(!row.busy, "the abandoned delegated turn must be released"); + assert!( + row.busy, + "the delegated turn is suspended, not abandoned — the session still owes \ + this task an answer" + ); sessions .write(&id, b"back in\r") @@ -142,6 +166,31 @@ sleep 30 || screen_text(&sessions, &id).contains("operator typed: back in"), ) .await; + assert!( + !run.is_finished(), + "the operator working in the session must not settle the task" + ); + + // Hand it back: the runtime runs a fresh turn in this same session, and its + // answer is this task's result — the session is the only context that saw + // both the agent's partial work and the operator's. + assert!(sessions.set_control(&id, SessionControl::Orchestrator)); + let outcome = tokio::time::timeout(PATIENCE * 3, run) + .await + .expect("the hand-back turn settles the task") + .expect("the executor task does not panic") + .expect("a hand-back must produce a result, not an error"); + assert!( + outcome + .reply + .contains("reviewed the session and finished it"), + "the result must come from the hand-back turn: {outcome:?}" + ); + assert_eq!( + outcome.session_id.as_deref(), + Some("codex-takeover-e2e"), + "the hand-back turn runs in the same session, not a fresh one" + ); sessions.close(&id); }