feat: stream a worker's live harness screen to the orchestrator (medulla.screen.v1) - #83
Conversation
📝 WalkthroughWalkthroughThe change adds push-aware relay inbox handling, task-session tracking, a ChangesLive screen streaming
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant TUI
participant HubHandle
participant WorkerScreenRouter
participant SessionStream
participant ScreenStore
TUI->>HubHandle: watch(worker, task_id)
HubHandle->>WorkerScreenRouter: Subscribe(task_id, resync)
WorkerScreenRouter->>SessionStream: start subscription
SessionStream->>ScreenStore: send full or delta frame
ScreenStore-->>TUI: expose updated watched screen
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
Lets the orchestrator watch a delegated task's actual terminal — the claude/codex session running it — instead of inferring progress from a status string every four seconds. Select a task in the Agents tab and its live screen replaces the transcript. The motivating case is the one the status plane structurally cannot show: a harness stopped on "Do you want to allow this command?" writes no transcript records, so it emits no status frames and reads as thinking until the task times out. On screen it is unmistakable in one frame. Modelled on mosh's State Synchronization Protocol. What crosses the wire is the screen — a grid of styled cells plus a cursor — carried as a diff from the state the viewer is known to hold, not a pty byte stream. Cost therefore tracks the screen rather than the output: sampling is timer-driven, an unchanged screen sends nothing, and a one-line paint on a 120x30 screen crosses as a one-row delta. Loss is self-healing — a frame that cannot be applied is never retried; the viewer resynchronises and the next full frame supersedes what was missed, which matters because mailbox ordering is not guaranteed. Streams are addressed by task id, not session id. A session serves one task at a time, so while a task runs the two identify the same thing — and the task id is the one both ends already hold, so nothing has to be published. It also makes authorization structural: the running-task record is keyed by (authenticated sender, task id), so resolving a subscription at all proves the caller dispatched it. The hub is a passive observer. There is no input and no resize message, so there is nothing to negotiate and no path from a subscriber into a session. Geometry belongs to the sender; a resize forces a full frame. Inbound envelopes now arrive over the relay's WebSocket rather than a GET poll — measured 1500ms to ~500ms. Acks stay on HTTPS (the stream is a fan-out, not a consumption), delivery is deduplicated by message id (a reconnect replays the mailbox snapshot, and a screen delta applied twice fails its base_seq check), and the fetch still runs when the socket is down, when the queue overflowed, when a frame will not parse, and on a 30s reconciliation. MEDULLA_INBOX_WS=0 restores pure polling. Ported onto main rather than replayed: main had since split transport, worker_loop, the hub runner/handle and the Agents tab into directory modules, replaced the `Relay` trait with the shared `Bridge` contract, and added workflow/system-info frame kinds. The 15 original commits could not apply through that, so this is one coherent change against the current structure, conforming to the newer module and test-layout rules. Verified on the ported tree: medulla 1460, medulla-tui 486, every integration suite green except the five pre-existing e2e_tinyplace failures that reproduce on an untouched main. Live against staging, two identities and a real pty: full frame at seq 1, delta applied at seq 2.
The four selection tests from the original branch, re-pointed at main's structure: the Agents rail is `RailRow`s now rather than a flat `AgentRow` list, and "Chat" is no longer a tab. Each is strict rather than skipping — the fixture is asserted to have both a task sublane and a lane row, so none can quietly pass by finding nothing to select. They cover the three rules worth having: selecting a task subscribes, reselecting the same one does not (every subscribe carries `resync: true`, so re-issuing per keystroke would make the worker resend a full frame), and leaving the tab releases the stream.
e6854f9 to
71a80be
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/sdk/src/daemon/mod.rs (1)
29-46: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winDocument the newly public listener types.
Making
listenerpublic exposesPushInbox,SeenIds, andListenerGuard, but their declarations lack type-level rustdoc. Add///docs at their definitions before exporting this API. As per coding guidelines, every public Rust item—including types—must have a///doc comment.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/sdk/src/daemon/mod.rs` around lines 29 - 46, Add type-level /// rustdoc comments to the definitions of PushInbox, SeenIds, and ListenerGuard in the listener module before their public re-exports. Describe each type’s purpose, while leaving the existing exports and behavior unchanged.Source: Coding guidelines
🧹 Nitpick comments (5)
src/sdk/src/tinyplace/screen/tests.rs (1)
177-198: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider covering the duplicate/reordered-frame paths the module doc promises.
apply.rsdocuments that "a dropped, duplicated, or reordered delta all fail the same way", but only the dropped case is asserted here. A replayed delta (base_seqolder than the view) and a stale full frame arriving after a newer one both take distinct branches — notably a stale full frame is applied unconditionally and rewindsview.seq.♻️ Suggested additional test
#[test] fn a_replayed_delta_is_refused_without_touching_the_view() { let base = grid(10, vec![row("a")]); let FrameDecision::Send(full) = build_frame(None, &base, "w_1", 1, 0) else { panic!("expected a frame"); }; let mut view = None; apply_frame(&mut view, &full); let next = grid(10, vec![row("b")]); let FrameDecision::Send(delta) = build_frame(Some(&base), &next, "w_1", 2, 1) else { panic!("expected a frame"); }; assert_eq!(apply_frame(&mut view, &delta), ApplyOutcome::Applied); // The same delta arriving twice must not re-apply against seq 2. assert_eq!(apply_frame(&mut view, &delta), ApplyOutcome::NeedsResync); assert_eq!(view.expect("view survives").grid, next); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/sdk/src/tinyplace/screen/tests.rs` around lines 177 - 198, Add tests alongside a_sequence_gap_asks_to_resync_without_touching_the_view covering both replayed deltas and stale full frames. Verify a duplicate delta returns NeedsResync without changing the already-applied view, and verify a full frame with an older sequence is rejected without rewinding the current view or sequence.src/tui/tests/live_screen_stream.rs (2)
350-355: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueHelper name promises more than it does.
worker_client_acceptonly re-requests the contact; the doc comment says "accept any pending contact request". Either rename to something likesettle_contact_edgeor inline the singlerequest_contactcall at Line 177.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/tui/tests/live_screen_stream.rs` around lines 350 - 355, Rename worker_client_accept to settle_contact_edge to accurately reflect that it re-requests the contact rather than explicitly accepting pending requests, and update all call sites accordingly; alternatively, remove the helper and inline the single request_contact call at the indicated usage.
51-68: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSilent
Noneon a real identity-load failure.
load_or_create_identity(...).ok()?makes a corrupt or unreadableconfig.jsonindistinguishable from "never onboarded" — the test just returns and passes. Since this is a manually-run diagnostic, print the error before skipping.♻️ Proposed change
- let (signer, config) = load_or_create_identity(&config_file, &env).ok()?; + let (signer, config) = match load_or_create_identity(&config_file, &env) { + Ok(loaded) => loaded, + Err(error) => { + eprintln!("skipping: {label} identity at {} is unusable: {error}", config_file.display()); + return None; + } + };🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/tui/tests/live_screen_stream.rs` around lines 51 - 68, Update identity so load_or_create_identity failures are handled explicitly instead of being discarded by .ok()?; print the returned error with context before returning None, while preserving the existing skip message for a missing config.json.src/sdk/src/hub/screens/mod.rs (1)
66-91: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueAvoid cloning the held grid on every applied frame.
applyruns per inbound frame (up tomax_fpsper watched session) and clones the full grid before folding, then reinserts. Taking the entry out and putting it back avoids one full-grid copy per frame.♻️ Sketch
- let mut view = screens.get(&key).map(|held| ScreenView { - grid: held.grid.clone(), - seq: held.seq, - }); + let mut view = screens.remove(&key).map(|held| ScreenView { + grid: held.grid, + seq: held.seq, + }); let outcome = apply_frame(&mut view, frame); - if outcome == ApplyOutcome::Applied { - if let Some(view) = view { + match (outcome, view) { + (ApplyOutcome::Applied, Some(view)) => { screens.insert( key, WatchedScreen { worker: worker.to_string(), task_id: frame.task_id.clone(), grid: view.grid, seq: view.seq, updated_at: now, }, ); evict_stalest(&mut screens); } + // A rejected frame must leave the previous screen in place. + (_, Some(view)) => { /* reinsert `view` under `key` unchanged */ } + (_, None) => {} }Note the rejected-frame branch must restore the prior entry (including its original
updated_at), so keep the timestamp alongside the view if you take this route.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/sdk/src/hub/screens/mod.rs` around lines 66 - 91, Update ScreenStore::apply to remove the existing entry from screens and fold the frame directly into its grid instead of cloning the held grid. Preserve the prior WatchedScreen, including updated_at, and restore it unchanged when apply_frame rejects the frame; for applied frames, update the entry’s grid, sequence, and updated_at before reinserting and evicting stale screens.src/tui/src/ui/screen/mod.rs (1)
13-90: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMove rendering implementation out of
mod.rs.Move these helpers into a sibling implementation module (for example,
render.rs) and keepmod.rsto docs, wiring, re-exports, and test declaration.As per coding guidelines, "
mod.rs[must be] focused on docs, module wiring, re-exports, and minimal glue."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/tui/src/ui/screen/mod.rs` around lines 13 - 90, Move the rendering helpers wire_color, run_style, row_line, and grid_lines from mod.rs into a sibling implementation module such as render.rs, preserving their behavior and visibility. Keep screen_title in its current location unless it is part of the rendering implementation, then update mod.rs to contain only module documentation, module declarations, re-exports, and test wiring.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/sdk/src/daemon/listener/mod.rs`:
- Around line 212-229: Update the spawned listener future associated with
ListenerGuard so cancellation performs cleanup before the task exits: add an
RAII drop guard that clears the shared listening state and calls nudge(). Ensure
this guard runs for both normal completion and JoinHandle::abort(), preventing
the drain path from treating an aborted listener as healthy.
In `@src/sdk/src/daemon/transport/mod.rs`:
- Around line 392-394: Update the duplicate-delivery handling around SeenIds and
the acknowledgement path so an ID is not permanently skipped when its
acknowledgement fails. Ensure failed acknowledgements remove the ID from the
seen set or otherwise allow the next drain to retry acknowledgement, while
preserving deduplication after successful acknowledgement.
- Around line 371-386: Update drain_inbox to return immediately for non-positive
limit values, then replace PushInbox::take with a take_up_to(limit) operation so
pushed envelopes are capped at the requested limit. Fetch from the client only
with the remaining capacity after pushed envelopes, preserving the existing
error handling and total result bound.
In `@src/sdk/src/hub/runner/pump.rs`:
- Around line 212-227: The screen pump must authorize frames against the active
worker/task watch registry before applying them or requesting resynchronization.
In src/sdk/src/hub/runner/pump.rs lines 212-227, reject frames whose worker and
task are not currently registered. In src/sdk/src/hub/handle/mod.rs lines
87-111, maintain that registry for dispatched worker/task pairs: register before
sending a watch and roll back on send failure, then remove the entry before
unwatch so late frames are ignored.
In `@src/sdk/tests/live_inbox_push.rs`:
- Around line 107-113: Update drain_mailbox to acknowledge only messages whose
IDs start with the live-push- probe prefix, preserving unrelated mailbox
messages. Keep the existing listing and acknowledgement flow, and retain the
snapshot assertion behavior independent of mailbox emptiness.
In `@src/tui/src/event_loop/cmd_dispatch/mod.rs`:
- Around line 181-198: Serialize Cmd::WatchTask transitions so rapid selection
changes cannot interleave stale starts and stops; update the WatchTask dispatch
flow and its surrounding coordinator to track the latest request generation or
process transitions through a single ordered worker, ensuring obsolete starts
are skipped and the final selection is authoritative. Do not discard stop
failures: surface them or reconcile the watch state so failed stops cannot leave
unselected tasks streaming.
In `@src/tui/src/ui/app/input.rs`:
- Around line 354-364: Ensure every navigation path that changes the active
selection—Tab/BackTab, tab-bar clicks, rail-wheel movement, task clicks, and
arrow navigation—routes through retarget_watch so the previous worker is stopped
and the new target is started alongside any tab-entry command; update
src/tui/src/ui/app/input.rs lines 354-364 around retarget_watch and the relevant
transition handlers. In src/tui/src/ui/app/tests.rs lines 377-390, navigate away
via on_key or handle_click instead of assigning tab_index directly, and assert
that the unwatch command is emitted.
In `@src/tui/src/ui/app/render/agents/transcript.rs`:
- Around line 258-267: Update selected_screen to filter worker_screens by both
held.task_id and the selected watch target’s worker, matching the ScreenStore
(worker, task_id) key. Preserve the existing task-based lookup and return
behavior while ensuring screens from other workers are excluded.
In `@src/tui/src/ui/screen/mod.rs`:
- Around line 72-74: Update grid_lines to account for ScreenGrid.cursor and
hide_cursor when constructing each row, overlaying or styling the cursor cell so
transmitted cursors render in the appropriate position while remaining hidden
when hide_cursor is enabled. Preserve the existing row_line rendering for all
non-cursor cells.
- Around line 40-56: Update run_style so styles with ATTR_INVERSE and both wire
colors set to Default also receive Modifier::REVERSED, preserving inverse video
when color swapping has no visible effect. Add a regression test covering this
default-colour inverse case while keeping existing color inversion behavior
unchanged.
In `@src/tui/src/worker_loop/mod.rs`:
- Around line 246-265: Update the worker shutdown path to call screens.clear()
before dropping the ScreenRouter, ensuring its sampler tasks are aborted through
StreamRegistry::clear() rather than detached via dropped JoinHandles.
In `@src/tui/src/worker/stream/sampler.rs`:
- Around line 150-183: Scope stream registry entries by the authenticated sender
and task instead of task ID alone. Update StreamRegistry and its
subscribe/unsubscribe lookups in src/tui/src/worker/stream/sampler.rs#L150-183
to use a consistent sender-plus-task key; update the unsubscribe call in
src/tui/src/worker/stream/router.rs#L88-94 and the contains/subscribe calls in
src/tui/src/worker/stream/router.rs#L123-136 to construct and use that same
scoped key.
- Around line 173-182: Update the resync flow around SessionStream creation and
the stream folding logic to preserve a monotonic per-stream sequence or epoch
across restarts. Ensure newly spawned streams inherit the prior stream’s
ordering state rather than restarting at sequence 1, and reject delayed stale
full frames during folding while accepting the restarted stream’s current full
frame.
In `@src/tui/tests/e2e_screen_stream.rs`:
- Around line 289-302: Replace the fixed sleeps in the idle-stream assertion
with bounded polling that waits for the outbox to receive its initial frame and
then become stable across two consecutive intervals. Use the existing PATIENCE
bound, preserve the settled-count and unchanged-screen assertions, and avoid
counting a legitimate late first frame in the idle window.
---
Outside diff comments:
In `@src/sdk/src/daemon/mod.rs`:
- Around line 29-46: Add type-level /// rustdoc comments to the definitions of
PushInbox, SeenIds, and ListenerGuard in the listener module before their public
re-exports. Describe each type’s purpose, while leaving the existing exports and
behavior unchanged.
---
Nitpick comments:
In `@src/sdk/src/hub/screens/mod.rs`:
- Around line 66-91: Update ScreenStore::apply to remove the existing entry from
screens and fold the frame directly into its grid instead of cloning the held
grid. Preserve the prior WatchedScreen, including updated_at, and restore it
unchanged when apply_frame rejects the frame; for applied frames, update the
entry’s grid, sequence, and updated_at before reinserting and evicting stale
screens.
In `@src/sdk/src/tinyplace/screen/tests.rs`:
- Around line 177-198: Add tests alongside
a_sequence_gap_asks_to_resync_without_touching_the_view covering both replayed
deltas and stale full frames. Verify a duplicate delta returns NeedsResync
without changing the already-applied view, and verify a full frame with an older
sequence is rejected without rewinding the current view or sequence.
In `@src/tui/src/ui/screen/mod.rs`:
- Around line 13-90: Move the rendering helpers wire_color, run_style, row_line,
and grid_lines from mod.rs into a sibling implementation module such as
render.rs, preserving their behavior and visibility. Keep screen_title in its
current location unless it is part of the rendering implementation, then update
mod.rs to contain only module documentation, module declarations, re-exports,
and test wiring.
In `@src/tui/tests/live_screen_stream.rs`:
- Around line 350-355: Rename worker_client_accept to settle_contact_edge to
accurately reflect that it re-requests the contact rather than explicitly
accepting pending requests, and update all call sites accordingly;
alternatively, remove the helper and inline the single request_contact call at
the indicated usage.
- Around line 51-68: Update identity so load_or_create_identity failures are
handled explicitly instead of being discarded by .ok()?; print the returned
error with context before returning None, while preserving the existing skip
message for a missing config.json.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ae2f47c6-4124-4c84-ac48-1f7d72fd0829
📒 Files selected for processing (60)
src/sdk/src/bridge/routing.rssrc/sdk/src/bridge/tinyplace.rssrc/sdk/src/bridge/types.rssrc/sdk/src/daemon/capabilities/mod.rssrc/sdk/src/daemon/entry.rssrc/sdk/src/daemon/listener/mod.rssrc/sdk/src/daemon/listener/tests.rssrc/sdk/src/daemon/mod.rssrc/sdk/src/daemon/providers/types.rssrc/sdk/src/daemon/runtime.rssrc/sdk/src/daemon/task_loop/run.rssrc/sdk/src/daemon/task_loop/workflow.rssrc/sdk/src/daemon/transport/mod.rssrc/sdk/src/daemon/transport/types.rssrc/sdk/src/daemon/types.rssrc/sdk/src/hub/handle/mod.rssrc/sdk/src/hub/mod.rssrc/sdk/src/hub/runner/mod.rssrc/sdk/src/hub/runner/pump.rssrc/sdk/src/hub/runner/types.rssrc/sdk/src/hub/screens/mod.rssrc/sdk/src/hub/screens/tests.rssrc/sdk/src/runtime/backend/runtime.rssrc/sdk/src/runtime/mod.rssrc/sdk/src/sessions/manager/turns.rssrc/sdk/src/tinyplace/mod.rssrc/sdk/src/tinyplace/screen/apply.rssrc/sdk/src/tinyplace/screen/codec.rssrc/sdk/src/tinyplace/screen/diff.rssrc/sdk/src/tinyplace/screen/mod.rssrc/sdk/src/tinyplace/screen/tests.rssrc/sdk/src/tinyplace/screen/types.rssrc/sdk/tests/e2e_daemon_providers.rssrc/sdk/tests/e2e_daemon_router.rssrc/sdk/tests/live_inbox_push.rssrc/tui/src/event_loop/cmd_dispatch/mod.rssrc/tui/src/ui/app/input.rssrc/tui/src/ui/app/keys/agents.rssrc/tui/src/ui/app/keys/mod.rssrc/tui/src/ui/app/render/agents/transcript.rssrc/tui/src/ui/app/state.rssrc/tui/src/ui/app/tests.rssrc/tui/src/ui/app/types.rssrc/tui/src/ui/mod.rssrc/tui/src/ui/screen/mod.rssrc/tui/src/ui/screen/tests.rssrc/tui/src/worker/executor/mod.rssrc/tui/src/worker/executor_tests/live.rssrc/tui/src/worker/executor_tests/mod.rssrc/tui/src/worker/mod.rssrc/tui/src/worker/stream/convert.rssrc/tui/src/worker/stream/mod.rssrc/tui/src/worker/stream/router.rssrc/tui/src/worker/stream/sampler.rssrc/tui/src/worker/stream/tests.rssrc/tui/src/worker_loop/commands.rssrc/tui/src/worker_loop/mod.rssrc/tui/tests/e2e_screen_stream.rssrc/tui/tests/live_screen_stream.rsvendor/tinyplace
| /// A running listener, stopped when this guard drops. | ||
| /// | ||
| /// A bare [`JoinHandle`] would be wrong here: dropping one *detaches* the task | ||
| /// rather than aborting it, so a listener tied to a session's lifetime would | ||
| /// outlive it and hold a socket open for a drain loop that no longer exists. | ||
| pub struct ListenerGuard(JoinHandle<()>); | ||
|
|
||
| impl ListenerGuard { | ||
| /// Stop the listener now rather than at drop. | ||
| pub fn abort(&self) { | ||
| self.0.abort(); | ||
| } | ||
| } | ||
|
|
||
| impl Drop for ListenerGuard { | ||
| fn drop(&mut self) { | ||
| self.0.abort(); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Clear push state when an aborted listener task is dropped.
JoinHandle::abort() can cancel the task while listening is still true, bypassing Lines 283-306. The drain can then treat a dead socket as healthy and defer fallback fetching. Add cancellation-safe cleanup inside the spawned future (for example, an RAII guard whose Drop clears listening and calls nudge()).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sdk/src/daemon/listener/mod.rs` around lines 212 - 229, Update the
spawned listener future associated with ListenerGuard so cancellation performs
cleanup before the task exits: add an RAII drop guard that clears the shared
listening state and calls nudge(). Ensure this guard runs for both normal
completion and JoinHandle::abort(), preventing the drain path from treating an
aborted listener as healthy.
| pub async fn drain_inbox(&self, limit: i64) -> Vec<InboundMessage> { | ||
| let _guard = self.lock.lock().await; | ||
| let response = match self | ||
| .client | ||
| .messages | ||
| .list(&self.our_agent_id, Some(limit)) | ||
| .await | ||
| { | ||
| Ok(response) => response, | ||
| Err(_) => return Vec::new(), | ||
| }; | ||
|
|
||
| let mut envelopes = self.push.take(); | ||
| if self.should_fetch() { | ||
| match self | ||
| .client | ||
| .messages | ||
| .list(&self.our_agent_id, Some(limit)) | ||
| .await | ||
| { | ||
| Ok(response) => envelopes.extend(response.messages), | ||
| // A failed fetch is not fatal when the socket is up: it has | ||
| // delivered what it delivered, and the next drain retries. | ||
| Err(_) if !envelopes.is_empty() => {} | ||
| Err(_) => return Vec::new(), |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Honor limit for pushed envelopes.
Line 374 drains the entire push queue, then Line 379 can fetch another limit messages. A drain_inbox(50) call can therefore decrypt, dispatch, and acknowledge far more than 50 messages. Add a bounded PushInbox::take_up_to(limit) API, return early for non-positive limits, and fetch only the remaining capacity.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sdk/src/daemon/transport/mod.rs` around lines 371 - 386, Update
drain_inbox to return immediately for non-positive limit values, then replace
PushInbox::take with a take_up_to(limit) operation so pushed envelopes are
capped at the requested limit. Fetch from the client only with the remaining
capacity after pushed envelopes, preserving the existing error handling and
total result bound.
| if !self.seen.insert(&message.id) { | ||
| continue; // already handled; the ack for it is already in flight | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Retry acknowledgement for duplicate deliveries.
If the acknowledgement at Lines 429-433 fails, this ID remains in the relay mailbox but is permanently skipped here by SeenIds; later drains neither process nor acknowledge it. A duplicate should retry acknowledgement (or the ID must be removed from the seen set when acknowledgement fails).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sdk/src/daemon/transport/mod.rs` around lines 392 - 394, Update the
duplicate-delivery handling around SeenIds and the acknowledgement path so an ID
is not permanently skipped when its acknowledgement fails. Ensure failed
acknowledgements remove the ID from the seen set or otherwise allow the next
drain to retry acknowledgement, while preserving deduplication after successful
acknowledgement.
| if screens.apply(from, &frame, crate::clock::now_millis()) | ||
| == crate::tinyplace::ApplyOutcome::NeedsResync | ||
| { | ||
| if let Some(log) = log { | ||
| log(&format!( | ||
| "hub ← screen {task_id} from {from}: out of step at seq {} — resyncing", | ||
| frame.seq | ||
| )); | ||
| } | ||
| let body = | ||
| crate::tinyplace::encode_screen_message(&crate::tinyplace::ScreenMessage::Subscribe { | ||
| task_id, | ||
| max_fps: DEFAULT_SCREEN_FPS, | ||
| resync: true, | ||
| }); | ||
| let _ = relay.send(from, &body).await; |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift
Authorize frames against the active watch set.
The pump accepts every valid screen frame without checking that its (worker, task_id) is currently watched and dispatch-authorized. An authenticated peer can fill the bounded cache with unsolicited frames, evicting real screens; late frames can also recreate a screen after unwatch.
src/sdk/src/hub/runner/pump.rs#L212-L227: reject unregistered or worker/task-mismatched frames before applying or requesting resync.src/sdk/src/hub/handle/mod.rs#L87-L111: maintain an active watch registry tied to the dispatched worker/task relationship; register before sending watch (roll back on failure) and remove before unwatch so late frames are ignored.
📍 Affects 2 files
src/sdk/src/hub/runner/pump.rs#L212-L227(this comment)src/sdk/src/hub/handle/mod.rs#L87-L111
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sdk/src/hub/runner/pump.rs` around lines 212 - 227, The screen pump must
authorize frames against the active worker/task watch registry before applying
them or requesting resynchronization. In src/sdk/src/hub/runner/pump.rs lines
212-227, reject frames whose worker and task are not currently registered. In
src/sdk/src/hub/handle/mod.rs lines 87-111, maintain that registry for
dispatched worker/task pairs: register before sending a watch and roll back on
send failure, then remove the entry before unwatch so late frames are ignored.
| async fn drain_mailbox(client: &TinyPlaceClient, agent_id: &str) { | ||
| if let Ok(response) = client.messages.list(agent_id, Some(50)).await { | ||
| for message in response.messages { | ||
| let _ = client.messages.acknowledge(&message.id, agent_id).await; | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
drain_mailbox deletes the developer's real mail, not just probe messages.
It acknowledges (deletes) every envelope in the configured identity's live mailbox, including anything unrelated that happened to be pending when the test ran. Probe envelopes are already identifiable by their live-push- id prefix — filter on that so an explicit --ignored run can't discard real mail.
🛡️ Proposed change
-/// Delete everything currently in the mailbox, so a probe leaves nothing behind.
+/// Delete only this probe's own messages, so a run leaves the mailbox as it
+/// found it — this is a live identity that may hold real mail.
async fn drain_mailbox(client: &TinyPlaceClient, agent_id: &str) {
if let Ok(response) = client.messages.list(agent_id, Some(50)).await {
- for message in response.messages {
+ for message in response
+ .messages
+ .into_iter()
+ .filter(|message| message.id.starts_with("live-push-"))
+ {
let _ = client.messages.acknowledge(&message.id, agent_id).await;
}
}
}Step 2's snapshot assertion only checks that a frame carries a string type, so it does not depend on an empty mailbox.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async fn drain_mailbox(client: &TinyPlaceClient, agent_id: &str) { | |
| if let Ok(response) = client.messages.list(agent_id, Some(50)).await { | |
| for message in response.messages { | |
| let _ = client.messages.acknowledge(&message.id, agent_id).await; | |
| } | |
| } | |
| } | |
| /// Delete only this probe's own messages, so a run leaves the mailbox as it | |
| /// found it — this is a live identity that may hold real mail. | |
| async fn drain_mailbox(client: &TinyPlaceClient, agent_id: &str) { | |
| if let Ok(response) = client.messages.list(agent_id, Some(50)).await { | |
| for message in response | |
| .messages | |
| .into_iter() | |
| .filter(|message| message.id.starts_with("live-push-")) | |
| { | |
| let _ = client.messages.acknowledge(&message.id, agent_id).await; | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sdk/tests/live_inbox_push.rs` around lines 107 - 113, Update
drain_mailbox to acknowledge only messages whose IDs start with the live-push-
probe prefix, preserving unrelated mailbox messages. Keep the existing listing
and acknowledgement flow, and retain the snapshot assertion behavior independent
of mailbox emptiness.
| pub fn grid_lines(grid: &ScreenGrid) -> Vec<Line<'static>> { | ||
| grid.lines.iter().map(|runs| row_line(runs)).collect() | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Render the transmitted cursor state.
grid_lines ignores ScreenGrid.cursor and hide_cursor; the downstream pane renders only these lines, so remote cursors never appear. Overlay or style the cursor cell while honoring hide_cursor.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/tui/src/ui/screen/mod.rs` around lines 72 - 74, Update grid_lines to
account for ScreenGrid.cursor and hide_cursor when constructing each row,
overlaying or styling the cursor cell so transmitted cursors render in the
appropriate position while remaining hidden when hide_cursor is enabled.
Preserve the existing row_line rendering for all non-cursor cells.
| tokio::spawn(async move { | ||
| // Held here so the push channel dies with the loop it feeds. A detached | ||
| // listener would keep a socket open for a drain that no longer exists — | ||
| // and a `JoinHandle` dropped on its own detaches rather than aborting. | ||
| let _push = push; | ||
| loop { | ||
| for message in transport.drain_inbox(50).await { | ||
| if let Some(screen) = medulla::tinyplace::parse_screen_message(&message.text) { | ||
| screens.handle(&message.from, screen); | ||
| continue; | ||
| } | ||
| let frame = decode_task_frame(&message.text); | ||
| runtime.handle_message(message.from, message.text, frame); | ||
| } | ||
| tokio::time::sleep(INBOX_POLL).await; | ||
| // Returns early when the relay's push channel delivers, so a | ||
| // subscribe is acted on at about a round trip rather than up to a | ||
| // poll interval later. The interval stays the correctness floor. | ||
| transport.wait_for_inbox(INBOX_POLL).await; | ||
| } | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
ast-grep outline src/tui/src/worker/stream --items all
rg -n -C3 'impl Drop for (StreamRegistry|ScreenRouter)|fn clear|\.clear\(\)' src/tui/src/worker src/tui/src/worker_loop
rg -n -C3 'inbox.*abort|JoinHandle|WorkerCmd::Quit' src/tui/src/worker_loopRepository: tinyhumansai/medulla
Length of output: 8543
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,240p' src/tui/src/worker/stream/router.rs
printf '\n--- sampler ---\n'
sed -n '1,280p' src/tui/src/worker/stream/sampler.rsRepository: tinyhumansai/medulla
Length of output: 14464
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,260p' src/tui/src/worker/stream/sampler.rs | nl -ba | sed -n '1,260p'
printf '\n--- router ---\n'
sed -n '1,220p' src/tui/src/worker/stream/router.rs | nl -ba | sed -n '1,220p'Repository: tinyhumansai/medulla
Length of output: 198
Abort the samplers during worker shutdown
Call screens.clear() on the shutdown path. Dropping ScreenRouter only drops the JoinHandles, which detaches the sampler tasks; StreamRegistry::clear() is the path that actually aborts them.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/tui/src/worker_loop/mod.rs` around lines 246 - 265, Update the worker
shutdown path to call screens.clear() before dropping the ScreenRouter, ensuring
its sampler tasks are aborted through StreamRegistry::clear() rather than
detached via dropped JoinHandles.
| pub struct StreamRegistry { | ||
| streams: std::collections::HashMap<String, tokio::task::JoinHandle<()>>, | ||
| } | ||
|
|
||
| impl StreamRegistry { | ||
| /// An empty registry. | ||
| pub fn new() -> Self { | ||
| StreamRegistry::default() | ||
| } | ||
|
|
||
| /// Start (or restart) the stream for `session_id`. | ||
| /// | ||
| /// Restarting is how a resync is served: the replacement stream begins | ||
| /// owing a full frame, which is exactly what the viewer asked for. | ||
| pub fn subscribe( | ||
| &mut self, | ||
| sessions: &PtyManager, | ||
| task_id: &str, | ||
| session_id: &str, | ||
| subscriber: &str, | ||
| max_fps: u8, | ||
| send: SendFn, | ||
| ) { | ||
| self.unsubscribe(task_id); | ||
| let handle = spawn_session_stream( | ||
| sessions.clone(), | ||
| task_id.to_string(), | ||
| session_id.to_string(), | ||
| subscriber.to_string(), | ||
| max_fps, | ||
| send, | ||
| ); | ||
| self.streams.insert(task_id.to_string(), handle); | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Scope stream registry entries by authenticated sender and task.
Task resolution is keyed by (from, task_id), but the registry is keyed only by task_id. Two peers with the same task ID can suppress, replace, or unsubscribe each other’s streams.
src/tui/src/worker/stream/sampler.rs#L150-L183: key streams by a sender-plus-task identifier rather thantask_idalone.src/tui/src/worker/stream/router.rs#L88-L94: unsubscribe using the same sender-scoped key.src/tui/src/worker/stream/router.rs#L123-L136: performcontainsandsubscribewith the sender-scoped key.
📍 Affects 2 files
src/tui/src/worker/stream/sampler.rs#L150-L183(this comment)src/tui/src/worker/stream/router.rs#L88-L94src/tui/src/worker/stream/router.rs#L123-L136
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/tui/src/worker/stream/sampler.rs` around lines 150 - 183, Scope stream
registry entries by the authenticated sender and task instead of task ID alone.
Update StreamRegistry and its subscribe/unsubscribe lookups in
src/tui/src/worker/stream/sampler.rs#L150-183 to use a consistent
sender-plus-task key; update the unsubscribe call in
src/tui/src/worker/stream/router.rs#L88-94 and the contains/subscribe calls in
src/tui/src/worker/stream/router.rs#L123-136 to construct and use that same
scoped key.
| self.unsubscribe(task_id); | ||
| let handle = spawn_session_stream( | ||
| sessions.clone(), | ||
| task_id.to_string(), | ||
| session_id.to_string(), | ||
| subscriber.to_string(), | ||
| max_fps, | ||
| send, | ||
| ); | ||
| self.streams.insert(task_id.to_string(), handle); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Preserve frame ordering across resync restarts.
A resync recreates SessionStream, resetting its next frame to sequence 1. Since full frames are applied independently of base_seq, a delayed old full frame can overwrite a newer view, and the receiver cannot distinguish it from the restarted stream’s new full frame. Preserve a monotonic per-stream sequence/epoch across restarts and reject stale full frames when folding.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/tui/src/worker/stream/sampler.rs` around lines 173 - 182, Update the
resync flow around SessionStream creation and the stream folding logic to
preserve a monotonic per-stream sequence or epoch across restarts. Ensure newly
spawned streams inherit the prior stream’s ordering state rather than restarting
at sequence 1, and reject delayed stale full frames during folding while
accepting the restarted stream’s current full frame.
| // Let it settle: the child paints, the sampler sends, then nothing changes. | ||
| tokio::time::sleep(Duration::from_millis(1_500)).await; | ||
| let settled = outbox.lock().unwrap().len(); | ||
| assert!(settled > 0, "the first frame should have gone out"); | ||
|
|
||
| // A full second at ten samples a second. A still screen must add nothing. | ||
| tokio::time::sleep(Duration::from_millis(1_000)).await; | ||
| let after = outbox.lock().unwrap().len(); | ||
| assert_eq!( | ||
| after, | ||
| settled, | ||
| "an unchanged screen sent {} further frame(s) across ~10 samples", | ||
| after - settled | ||
| ); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fixed 1.5s settle window makes the idle-stream assertion load-sensitive.
If the child's paint or the first sample lands after the 1.5s mark, either settled > 0 fails or a legitimate first/second frame is counted in the "after" window and the strict equality fails. Waiting for the outbox to go quiet (e.g. poll until the count is stable across two consecutive intervals, bounded by PATIENCE) keeps the property under test while removing the wall-clock dependence.
♻️ Sketch
- // Let it settle: the child paints, the sampler paints, then nothing changes.
- tokio::time::sleep(Duration::from_millis(1_500)).await;
- let settled = outbox.lock().unwrap().len();
- assert!(settled > 0, "the first frame should have gone out");
+ // Let it settle: poll until the count stops moving, so a busy box only
+ // makes this slower rather than red.
+ let deadline = Instant::now() + PATIENCE;
+ let mut settled = 0usize;
+ loop {
+ tokio::time::sleep(Duration::from_millis(250)).await;
+ let now = outbox.lock().unwrap().len();
+ if now > 0 && now == settled {
+ break;
+ }
+ settled = now;
+ assert!(Instant::now() < deadline, "the first frame never went out");
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // Let it settle: the child paints, the sampler sends, then nothing changes. | |
| tokio::time::sleep(Duration::from_millis(1_500)).await; | |
| let settled = outbox.lock().unwrap().len(); | |
| assert!(settled > 0, "the first frame should have gone out"); | |
| // A full second at ten samples a second. A still screen must add nothing. | |
| tokio::time::sleep(Duration::from_millis(1_000)).await; | |
| let after = outbox.lock().unwrap().len(); | |
| assert_eq!( | |
| after, | |
| settled, | |
| "an unchanged screen sent {} further frame(s) across ~10 samples", | |
| after - settled | |
| ); | |
| // Let it settle: poll until the count stops moving, so a busy box only | |
| // makes this slower rather than red. | |
| let deadline = Instant::now() + PATIENCE; | |
| let mut settled = 0usize; | |
| loop { | |
| tokio::time::sleep(Duration::from_millis(250)).await; | |
| let now = outbox.lock().unwrap().len(); | |
| if now > 0 && now == settled { | |
| break; | |
| } | |
| settled = now; | |
| assert!(Instant::now() < deadline, "the first frame never went out"); | |
| } | |
| // A full second at ten samples a second. A still screen must add nothing. | |
| tokio::time::sleep(Duration::from_millis(1_000)).await; | |
| let after = outbox.lock().unwrap().len(); | |
| assert_eq!( | |
| after, | |
| settled, | |
| "an unchanged screen sent {} further frame(s) across ~10 samples", | |
| after - settled | |
| ); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/tui/tests/e2e_screen_stream.rs` around lines 289 - 302, Replace the fixed
sleeps in the idle-stream assertion with bounded polling that waits for the
outbox to receive its initial frame and then become stable across two
consecutive intervals. Use the existing PATIENCE bound, preserve the
settled-count and unchanged-screen assertions, and avoid counting a legitimate
late first frame in the idle window.
|
| Filename | Overview |
|---|---|
| src/tui/src/ui/app/keys/mod.rs | Arrow-key navigation correctly calls retarget_watch(), but Tab/BackTab tab-switching goes through tab_enter_cmd() which never calls it — so switching tabs does not release the screen subscription, contradicting the documented intent in watch_target(). |
| src/tui/src/ui/app/render/agents/transcript.rs | Screen supersedes transcript when held — correct priority. selected_screen searches all worker screens by task_id alone rather than the full (worker, task_id) key the store holds. |
| src/sdk/src/hub/screens/mod.rs | LRU-by-timestamp store is correct and bounded (CAPACITY=8). The evict_stalest None arm is dead code — while len > CAPACITY guarantees non-empty so min_by_key always returns Some. |
| src/sdk/src/daemon/listener/mod.rs | WebSocket inbox listener is well-designed: bounded queue, failure cap, nudge() on any unrecognised frame, and reconnect delay. The fetch-on-unknown-frame-type (including keepalives) is documented and intentional. |
| src/sdk/src/daemon/transport/mod.rs | drain_inbox correctly merges push-delivered and HTTPS-fetched envelopes, deduplicates by message ID via SeenIds, and skips HTTPS only when the socket is live, owes nothing, and was reconciled recently. |
| src/tui/src/worker/stream/sampler.rs | Timer-driven sampler: correctly clamps FPS, diffs against last-sent grid, and terminates cleanly when the session disappears. The pure SessionStream::tick design makes it well-tested without a pty. |
| src/tui/src/worker/stream/router.rs | Authorization via structural key lookup is sound; silent rejection of unknown tasks is intentional. Resync correctly replaces the running stream handle, which resets owes_full. |
| src/sdk/src/tinyplace/screen/diff.rs | Row-level diff is correct: guards same-size check before indexing, handles the cursor-only-changed case, and the Unchanged return makes idle sessions free on the wire. |
| src/sdk/src/hub/runner/pump.rs | route_screen correctly ignores non-Frame messages, sends a resync Subscribe on NeedsResync, and silently discards the relay send error (fallback is the viewer pane freezing, which is visible). |
| src/tui/tests/e2e_screen_stream.rs | End-to-end test covers the real pty → emulator → wire → store chain, asserts both full-frame and delta paths, and verifies the unchanged-screen-costs-nothing property. |
Sequence Diagram
sequenceDiagram
participant TUI as TUI (Agents tab)
participant Hub as Hub pump
participant Relay as tiny.place relay
participant Worker as Worker inbox drain
participant Sampler as SessionStream sampler
participant PTY as pty / vt100
TUI->>Hub: retarget_watch() → Cmd::WatchTask
Hub->>Relay: "PUT /messages (Subscribe{task_id, max_fps:1, resync:true})"
Relay-->>Worker: WebSocket push (Subscribe envelope)
Worker->>Worker: parse_screen_message → ScreenRouter::handle
Worker->>Sampler: StreamRegistry::subscribe → spawn_session_stream
loop Every sample_interval(max_fps)
Sampler->>PTY: sessions.screen_rows(session_id)
PTY-->>Sampler: ScreenSnapshot
Sampler->>Sampler: SessionStream::tick → build_frame (diff or full)
alt screen changed
Sampler->>Relay: PUT /messages (Frame envelope)
Relay-->>Hub: WebSocket push (Frame envelope)
Hub->>Hub: parse_screen_message → route_screen → ScreenStore::apply
alt NeedsResync
Hub->>Relay: "PUT /messages (Subscribe{resync:true})"
end
else unchanged
Sampler->>Sampler: FrameDecision::Unchanged — nothing sent
end
end
TUI->>Hub: Tab key → tab_enter_cmd (no retarget_watch called)
Note over Sampler: subscription keeps running — worker streams to nobody
Comments Outside Diff (1)
-
src/tui/src/ui/app/keys/mod.rs, line 233-241 (link)Tab-switch does not release the screen subscription
TabandBackTabadvancetab_indexand calltab_enter_cmd(), which never callsretarget_watch(). So when the user switches away from the Agents tab with the tab key,self.watchingstaysSome(…)and the worker keeps streaming at 1 fps indefinitely — burning a Signal ratchet advance and an HTTPS send every second — until the user returns to Agents and presses an arrow key or clicks.watch_target()explicitly documents the intent: "leaving it releases the subscription." But that only happens whenretarget_watch()is actually called, which isn't wired to tab switching. Adding aretarget_watch()call inside theTab/BackTabarms (before or after updatingtab_index) would close the gap.
Reviews (1): Last reviewed commit: "test(agents): cover the watch trigger ag..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| self.runtime | ||
| .worker_screens() | ||
| .into_iter() | ||
| .find(|held| held.task_id == task.task_id) | ||
| } | ||
|
|
||
| /// Draw a watched worker's screen into `area`. | ||
| /// | ||
| /// No cursor is drawn and no scrollback is offered, matching the worker's | ||
| /// own pane: this is a window, not a keyboard, and the protocol never | ||
| /// synchronised history in the first place. | ||
| fn draw_worker_screen( |
There was a problem hiding this comment.
selected_screen searches by task_id only, not (worker, task_id)
worker_screens() returns screens from all workers, and the iterator searches with .find(|held| held.task_id == task.task_id). The ScreenStore itself is keyed by (worker, task_id), and both self.watching and the watch_target() result carry the worker address alongside the task ID.
If two workers were ever serving tasks with the same string ID, this would return the wrong screen. In practice, task IDs are orchestrator-generated and effectively unique — but using self.watching directly (which already holds the (worker_address, task_id) pair) and calling ScreenStore::get(worker, task_id) would make the lookup structurally correct and remove the dependency on that uniqueness assumption.
| match stalest { | ||
| Some(key) => { | ||
| screens.remove(&key); | ||
| } | ||
| None => return, | ||
| } |
There was a problem hiding this comment.
None arm in evict_stalest is unreachable
The while screens.len() > CAPACITY guard guarantees the map is non-empty (since CAPACITY = 8 > 0), so min_by_key always returns Some. The None => return is dead code that adds noise without providing safety.
| match stalest { | |
| Some(key) => { | |
| screens.remove(&key); | |
| } | |
| None => return, | |
| } | |
| if let Some(key) = stalest { | |
| screens.remove(&key); | |
| } |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
2 participants
Lets the orchestrator watch a delegated task's actual terminal — the claude/codex session running it — instead of inferring progress from a status string every four seconds. Select a task in the Agents tab and its live screen replaces the transcript.
The motivating case is the one the status plane structurally cannot show: a harness stopped on
Do you want to allow this command?writes no transcript records, so it emits no status frames and reads as "thinking" until the task times out. On screen it is unmistakable in one frame. That failure is whyworker/pty/dialog.rsexists; this makes it visible rather than guessed at.The model
Modelled on mosh's State Synchronization Protocol. What crosses the wire is the screen — a grid of styled cells plus a cursor — carried as a diff from the state the viewer is known to hold, not a pty byte stream. Three properties follow, and they're why this is affordable over an encrypted mailbox at all:
Measured end to end on a real pty: a one-line paint on a 120×30 screen crosses as a 1-row delta.
Decisions worth a reviewer's attention
Streams are addressed by task id, not session id. A session serves one task at a time (
claim_idlerefuses a busy session and marks the one it returns, under one lock), so while a task runs the two identify the same thing — and the task id is the one both ends already hold. This needs no protocol change: the worker resolves task → pty internally and the session id never leaves the machine. It also makes authorization structural rather than a check: the running-task record is keyed by(authenticated sender, task id), so resolving a subscription at all proves the caller dispatched it. A peer cannot name another's task, because the key includes an identity it doesn't have.The hub is a passive observer. No
inputand noresizemessage exist, so there is nothing to negotiate and no path from a subscriber into a session. Geometry belongs to the sender; a resize forces a full frame, since row indices don't survive one. Both are additive later behind the version tag.Inverse video is a flag on the wire, resolved at paint time on both sides. Terminals compose
REVERSEDinconsistently with an explicit background, so both renderers swap fg/bg instead. If the two ends disagreed, a harness status bar would be invisible on one view and fine on the other.Inbound envelopes now arrive over the relay's WebSocket rather than a
GET /messagespoll — latency from ~1500 ms to ~500 ms measured. Acks stay on HTTPS (the stream is a fan-out, not a consumption), delivery is deduplicated by message id (a reconnect replays the mailbox snapshot; a screen delta applied twice fails itsbase_seqcheck and forces a needless resync), and the fetch never goes away — it still runs when the socket is down, when the queue overflowed, when a frame won't parse, and on a 30 s reconciliation regardless.MEDULLA_INBOX_WS=0restores pure polling.Verification
Offline, deterministic:
e2e_screen_streamruns the whole chain with a real child on a real pty:/bin/shpainting a pseudo-terminal →vt100→ convert → diff → JSON → the daemon's task record resolving the subscription → the hub's fold, back out as the text the child printed. Both frame shapes, plus the assertion that ~10 samples of an unchanged screen send nothing further.Live, against
staging-api.tiny.place(#[ignore]d, socargo teststill skips them and the offline-and-deterministic rule holds):Two identities, the real relay, a real harness. The run is shorter than the reconciliation window, so everything after the first drain provably crossed the socket.
Two things to know before reviewing
This was ported onto
main, not replayed onto it. The work was written against a base 51 commits back, andmainhad since splittransport,worker_loop, the hubrunner/handleand the whole Agents tab into directory modules, replaced theRelaytrait with the sharedBridgecontract, and added workflow/system-info frame kinds. The original 15 commits could not apply through that — every intermediate state failed to build — so this is one coherent change against the current structure, conforming to the newer module and test-layout rules (#[path]test decls are now directory modules). The original history is preserved on a localbackup/mosh-pre-rebasetag if anyone wants it.The submodule pointer moved —
vendor/tinyplace→d254505. That picks up two upstream fixes this depends on:connect()previously handed a barehttp::Requesttoconnect_async, which passes it through verbatim, so the upgrade went out with noSec-WebSocket-Keyand could never handshake against any server (tinyhumansai/tiny.place#269); and a dropped stream now re-dials under a policy, re-signing each attempt (#270). Both merged upstream. CI needs the submodule initialised or the workspace won't build.Validation
e2e_tinyplacehas 5 pre-existing failures in my sandbox (service_*, panicking atmock_tinyplace.rs:190). I verified rather than assumed: this branch never touches that code, and they reproduce identically on an untouchedmainin a scratch worktree. If they're green in CI, the sandbox is the problem.Known gaps
ScreenCell.textcarries no width, so a column-indexed diff will mis-place CJK and emoji. Cosmetic, not structural — mosh handles this explicitly and we don't.MessageEnvelope.body; a ~10 KB full frame becomes ~14 KB base64. Worth probing before relying on single-message full frames.max_fpsabove ~1 is not yet worth much. The sampler clamps at 10 but nothing consults whether the push channel is up; tying effective rate to that is a natural follow-up.