a73x

efa56bfd

refactor: one tile's thread is wall_pump.zig

a73x   2026-08-28 20:50

Commit message
refactor: one tile's thread is wall_pump.zig

The transport a tile owns in both directions — attach, claims, agent
channels, redials, the paint hooks — is the pump's, and nothing in it
reaches another tile. The keyboard keeps the mailbox and the doorbell,
which both threads use.
docscheck.budget
Old New
@@ -50,3 +50,4 @@ server_agent.zig 0
50 server_sessions.zig 0 50 server_sessions.zig 0
51 wall_host.zig 0 51 wall_host.zig 0
52 wall_picker.zig 0 52 wall_picker.zig 0
53 wall_pump.zig 0
src/tui/wall_pump.zig
Old New
@@ -0,0 +1,1073 @@
1 //! One tile's thread: the transport it owns in BOTH directions, its
2 //! attach, its claims and releases, its agent channels, its redials, and
3 //! the paint hooks the interaction core calls back on. The keyboard hands
4 //! this thread work through the tile's mailbox and doorbell and reads its
5 //! answers back the same way; nothing here touches another tile.
6 const std = @import("std");
7 const proto = @import("protocol");
8 const client = @import("client");
9 const interact = @import("interact");
10 const wv = @import("wallview.zig");
11 const EndReason = wv.EndReason;
12 const Shared = wv.Shared;
13 const State = wv.State;
14 const Tile = wv.Tile;
15
16 /// Under `paint_mu`: a clear spliced into a 64 KiB OSC 52 write eats the
17 /// paint after it.
18 pub fn copySelection(
19 t: *Tile,
20 alloc: std.mem.Allocator,
21 core: *interact.Core,
22 payload: []const u8,
23 ) void {
24 var answer: interact.Copy = .none;
25 {
26 t.shared.paint_mu.lock();
27 defer t.shared.paint_mu.unlock();
28 // The drag is this tile's own Core's now: a drag is per tile, and
29 // the pump that owns the link is the one whose Core holds it.
30 const held = core.drag.range();
31 answer = core.selectionCopy(payload, held);
32 switch (answer) {
33 // `is_tty` and not the tile's claim: a tile that has released
34 // the terminal can still be the one whose finished drag the
35 // answer reaches, and the copy is still the user's. A piped
36 // `mux` asks its terminal for no mouse modes, so it can have no
37 // drag to copy in the first place.
38 .text => |text| if (t.shared.is_tty)
39 interact.writeSelectionCopy(alloc, t.shared.out_fd, text) catch {},
40 .none, .too_large => {},
41 }
42 }
43 // Outside the hold: `tileBanner` takes the same lock, which is not
44 // reentrant. The banner lands in this tile's own rect.
45 if (answer == .too_large) wv.tileBanner(t, "[selection too large to copy]");
46 }
47
48 /// Every live tile paints its own rect, so the sink admits a paint whenever
49 /// this tile has not been forgotten. `paint_mu` is held for the whole paint.
50 pub fn tilePaintBegin(ctx: ?*anyopaque) bool {
51 const t: *Tile = @ptrCast(@alignCast(ctx.?));
52 if (t.gone.load(.acquire)) return false;
53 t.shared.paint_mu.lock();
54 // Read UNDER the lock, not before it: a pump that tested the flag and
55 // then lost the race for `paint_mu` would paint its rect on top of the
56 // box the keyboard had just drawn — and `picker_stamp` suppresses the
57 // identical repaint that would have repaired it, so the damage sticks
58 // until a key changes the frame.
59 if (t.shared.picker_open.load(.acquire)) {
60 t.shared.paint_mu.unlock();
61 return false;
62 }
63 return true;
64 }
65
66 pub fn tilePaintEnd(ctx: ?*anyopaque) void {
67 const t: *Tile = @ptrCast(@alignCast(ctx.?));
68 defer t.shared.paint_mu.unlock();
69 // Before releasing the terminal: the cursor belongs to the FOCUSED tile.
70 // A focused paint records where its cursor landed; an unfocused paint's
71 // last act is to put the cursor back there, so a redraw in tile 2 cannot
72 // steal the eye while the keys go to tile 1. The cursor is hidden for
73 // the move so the show after it does not flash it at the unfocused
74 // tile's final position before the CUP lands — then re-shown so the
75 // focused tile's cursor rests visible until its next paint.
76 if (t.idx == t.shared.sel) {
77 if (t.core) |core| t.shared.cursor = core.screenCursor();
78 } else {
79 var cbuf: [26]u8 = undefined;
80 const cup = std.fmt.bufPrint(&cbuf, "\x1b[?25l\x1b[{d};{d}H\x1b[?25h", .{ t.shared.cursor.y + 1, t.shared.cursor.x + 1 }) catch return;
81 proto.writeAllFd(t.shared.out_fd, cup) catch {};
82 }
83 }
84
85 /// The one place a wall tile puts an attach on the wire. A tile the user
86 /// asked for — the entry tile, a chord-born tile, a local line off the
87 /// saved wall — claims its rect on attach, and that size is what lets the
88 /// daemon create the session. A view tile (wall argv, a remote saved
89 /// line) attaches at 0x0 so it can only JOIN, then takes its rect with the
90 /// resize doorbell one frame later — every tile still claims its
91 /// rectangle, and a redial comes back claiming what the tile claimed.
92 pub fn sendAttach(t: *Tile, tr: *client.Transport, have_seq: u64, have_epoch: u64) !void {
93 // Snapshot under `paint_mu`: the keyboard may relayout (re-cut stripes,
94 // resize) concurrently with the pump's first attach.
95 const snap = blk: {
96 t.shared.paint_mu.lock();
97 defer t.shared.paint_mu.unlock();
98 break :blk .{
99 .cols = t.viewCols(),
100 .view_rows = t.viewRows(),
101 };
102 };
103 const cols: u16 = if (t.creates) snap.cols else 0;
104 const rows: u16 = if (t.creates) snap.view_rows else 0;
105 var buf: [proto.attach_max_len]u8 = undefined;
106 try tr.writeFrame(.attach, proto.encodeAttachNamed(
107 &buf,
108 cols,
109 rows,
110 have_seq,
111 have_epoch,
112 proto.wireName(t.r.session),
113 ));
114 // A view tile made no size claim, so it owes its rect now: the same
115 // doorbell path a relayout takes (adoptSize + .resize), run on this
116 // pump thread which is the transport's only writer.
117 if (!t.creates) {
118 t.shared.paint_mu.lock();
119 t.resize_pending = true;
120 t.shared.paint_mu.unlock();
121 wv.ring(t);
122 }
123 // Re-armed on EVERY attach, not once per process: a redial is a fresh
124 // attach onto a fresh daemon-side slot, which remembers no offer. Empty
125 // payload, and a daemon too old to know the frame skips it.
126 if (t.r.agent) try tr.writeFrame(.agent_offer, "");
127 }
128
129 /// What a pump owes itself after asking for its focus claim.
130 pub const ClaimStep = enum {
131 /// The claim landed, or the Core already held it: finish the pass.
132 done,
133 /// The sink refused and this tile still has the focus: try next pass.
134 rearmed,
135 /// The focus moved on while the popup was up: this arm is stale.
136 dropped,
137 };
138
139 /// May this tile take the terminal for the arm it is holding?
140 fn claimAllowed(t: *Tile) bool {
141 // BEFORE the claim, never after. The keyboard arms `claim_pending` and
142 // the pump reads it a pass later, and the focus can move in between —
143 // a poller's tile arriving, the birth an Enter makes, all of it under
144 // the host picker's popup. An arm that outlives its focus and then
145 // SUCCEEDS puts two tiles' modes on one terminal, rests the cursor on
146 // the loser, and takes the focus notice with it; the outgoing tile's
147 // release was consumed a pass earlier, so nothing undoes any of it.
148 //
149 // Judging the claim's REFUSAL cannot cover this: the arm that matters
150 // is the one retried after the popup closed, which the sink admits.
151 t.shared.paint_mu.lock();
152 defer t.shared.paint_mu.unlock();
153 return t.shared.sel == t.idx;
154 }
155
156 /// The wall's ONLY door to `Core.claimTerminal`: the focus test, the size
157 /// adopt, and the claim, in that order and never apart. A caller that could
158 /// reach the claim around this is a caller that can mint a stale one.
159 pub fn claimFocus(t: *Tile, core: *interact.Core, rect: proto.Size) ClaimStep {
160 if (!claimAllowed(t)) return .dropped;
161 // A tile born before somebody resized the terminal clips its paints to
162 // a screen that is gone; the tile's current RECT, never the whole
163 // terminal, which on a wall of two would let it paint over a neighbour.
164 if (core.size.cols != rect.cols or core.size.rows != rect.rows)
165 core.adoptSize(rect);
166 return afterClaim(t, core.claimTerminal(), core.claim != .none, core.is_tty);
167 }
168
169 /// A refused focus claim, judged.
170 pub fn afterClaim(t: *Tile, claimed: bool, held: bool, is_tty: bool) ClaimStep {
171 // `held` is `Core.claim != .none`. A claim answers false for three
172 // reasons and only ONE of them is worth another pass: not a tty (there
173 // is no terminal to hold), already held (re-arming would re-take the
174 // focus notice every pass, forever), and the SINK refused — which is
175 // the host picker, whose popup admits no paint and a claim writes the
176 // session's modes through one. Read off the Core, never off
177 // `picker_open` a second time: the keyboard can clear that flag between
178 // the two loads, and then the refusal is dropped exactly as before.
179 if (claimed or held or !is_tty) return .done;
180 t.shared.paint_mu.lock();
181 defer t.shared.paint_mu.unlock();
182 // ...and only while this tile is STILL the focus. `claimAllowed` asked
183 // the same question before the claim; the keyboard can answer it
184 // differently in between, and re-arming then would mint the very stale
185 // arm that check exists to stop. Left CLEAR rather than cleared — the
186 // caller's `swap` did that — because a focus that came back inside this
187 // window re-armed it legitimately.
188 if (t.shared.sel != t.idx) return .dropped;
189 t.claim_pending.store(true, .release);
190 return .rearmed;
191 }
192
193 /// FOCUSED pumps only: a tile that never held the terminal has zero
194 /// counters, and publishing them would clobber the tile the user typed at.
195 fn publishStats(shared: *Shared, c: interact.PredictCounters) void {
196 shared.paint_mu.lock();
197 defer shared.paint_mu.unlock();
198 shared.stats = c;
199 }
200
201 /// Validated first: these bytes came out of a peer's `sessions_reply` and
202 /// `SessionName.of` memcpys with no bound of its own (`client.validPick`).
203 /// A name it refuses is a name nobody is moved to.
204 fn postAnswer(t: *Tile, pick: []const u8) void {
205 const name = client.validPick(pick) orelse return;
206 {
207 t.ans_mu.lock();
208 defer t.ans_mu.unlock();
209 t.ans = name;
210 }
211 t.ans_ready.store(true, .release);
212 wv.ringKeyboard(t.shared);
213 }
214
215 /// Written before `alive` clears, so no dead tile is ever seen without a
216 /// reason. Rings NOTHING — the keyboard's test is `!alive`.
217 fn endWith(t: *Tile, reason: EndReason, code: u8) void {
218 t.code.store(code, .release);
219 t.end.store(@intFromEnum(reason), .release);
220 }
221
222 /// Whole-mailbox chunking, so `offerKeystroke` (one-byte chunks only) counts
223 /// keystrokes that arrive between polls as suppressed: a wall predicts a
224 /// little less. Splitting would speculate against a stale replica.
225 pub fn takeKeys(t: *Tile, out: []u8) []u8 {
226 t.in_mu.lock();
227 defer t.in_mu.unlock();
228 const n = @min(out.len, t.in_len);
229 @memcpy(out[0..n], t.in[0..n]);
230 std.mem.copyForwards(u8, t.in[0 .. t.in_len - n], t.in[n..t.in_len]);
231 t.in_len -= n;
232 return out[0..n];
233 }
234
235 fn drainWake(t: *const Tile) void {
236 wv.drainBell(t.wake_r);
237 }
238
239 fn dial(alloc: std.mem.Allocator, t: *Tile, target_in: client.Target) ?client.Transport {
240 var target = target_in;
241 var backoff_ms: u64 = 0;
242 // `gone` as well as `running`: a tile forgotten while it is retrying a
243 // dead host must stop retrying, not keep a thread and a backoff alive
244 // for a tile that is no longer on the wall.
245 while (t.shared.running.load(.acquire) and !t.gone.load(.acquire)) {
246 if (client.Transport.open(alloc, target, null, -1)) |tr| return tr else |_| {}
247 // An ask buys ONE attempt. Every retry below is the wall's own
248 // idea: a `muxd start` per backoff would restart a daemon for as
249 // long as the tile lives, and a fallback line per backoff would
250 // scroll the alternate screen the tiles are painted on.
251 if (target == .hand) target.hand.asked = false;
252 backoff_ms = client.nextBackoffMs(backoff_ms);
253 // Sliced sleep so quit is never behind a full backoff.
254 var slept: u64 = 0;
255 while (slept < backoff_ms and t.shared.running.load(.acquire) and
256 !t.gone.load(.acquire)) : (slept += 50)
257 {
258 std.Thread.sleep(50 * std.time.ns_per_ms);
259 }
260 }
261 return null;
262 }
263
264 /// One forwarded ssh-agent channel, this end of it: the id the daemon
265 /// allocated, and an fd to THIS machine's agent.
266 ///
267 /// Thread-local by construction — the table lives in `pumpTile`'s frame and
268 /// no other thread can see it, which is why nothing here takes a lock and
269 /// why these helpers take the table as a slice rather than reaching for one.
270 pub const AgentLocal = struct { id: u32, fd: std.posix.fd_t };
271
272 /// Fixed at `proto.agent_chans_max`, which is what the daemon opens anyway:
273 /// a full table costs one failed lookup, not the pump's hot path an alloc.
274 pub fn storeLocal(locals: []?AgentLocal, id: u32, fd: std.posix.fd_t) ?usize {
275 for (locals, 0..) |c, s| {
276 if (c != null) continue;
277 locals[s] = .{ .id = id, .fd = fd };
278 return s;
279 }
280 return null;
281 }
282
283 pub fn findLocal(locals: []?AgentLocal, id: u32) ?usize {
284 for (locals, 0..) |c, s| if (c) |ch| {
285 if (ch.id == id) return s;
286 };
287 return null;
288 }
289
290 /// Hang one channel up from this end and say so, because the daemon is
291 /// holding the far socket open waiting for bytes that are not coming. A
292 /// failed write is the transport itself being gone, which the caller's next
293 /// pass turns into a redial.
294 pub fn closeLocal(locals: []?AgentLocal, slot: usize, transport: *client.Transport) void {
295 const ch = locals[slot] orelse return;
296 locals[slot] = null;
297 std.posix.close(ch.fd);
298 transport.writeFrame(.agent_close, &proto.encodeAgentId(ch.id)) catch {};
299 }
300
301 /// Lifted out of the pump so the `offered` gate has a seam a test can watch.
302 pub fn openAgentChan(
303 locals: []?AgentLocal,
304 id: u32,
305 offered: bool,
306 sock: []const u8,
307 ) bool {
308 // The OFFER is the consent, and it is per TILE: a wall where one tile
309 // was typed with `-A` must not hand another tile's host the keys,
310 // whoever asks.
311 if (!offered) return false;
312 // A live id reused. Refusing keeps the channel already on that id
313 // intact, which is the half of the collision that has real bytes moving
314 // through it.
315 if (findLocal(locals, id) != null) return false;
316 const fd = client.connectAgent(sock) orelse return false;
317 if (storeLocal(locals, id, fd) == null) {
318 std.posix.close(fd);
319 return false;
320 }
321 return true;
322 }
323
324 /// Lifted out of the pump so the length cap has a seam a test can reach.
325 pub fn deliverAgentData(
326 locals: []?AgentLocal,
327 payload: []const u8,
328 transport: *client.Transport,
329 ) bool {
330 const id = proto.decodeAgentId(payload) catch return false;
331 const s = findLocal(locals, id) orelse return true;
332 if (proto.agentDataOversize(payload)) {
333 closeLocal(locals, s, transport);
334 return true;
335 }
336 // Bytes onto the fd in order, never parsed and never reassembled: the
337 // daemon reads the far end in `agent_data_max` bites, so one agent
338 // message can arrive as several frames and several messages as one. The
339 // agent protocol delimits itself over a stream, and this end is a pipe.
340 proto.writeAllFd(locals[s].?.fd, payload[proto.agent_id_len..]) catch
341 closeLocal(locals, s, transport);
342 return true;
343 }
344
345 /// Redial only: these belonged to the dead connection, whose `dropClient`
346 /// already reaped the server side.
347 pub fn dropLocals(locals: []?AgentLocal) void {
348 for (locals, 0..) |c, s| if (c) |ch| {
349 locals[s] = null;
350 std.posix.close(ch.fd);
351 };
352 }
353
354 /// The transport died, or the dial has to be redone: rebuild it on the CLI's
355 /// backoff and re-attach quoting what this tile holds. False means the pump
356 /// is finished — the wall quit, or the tile was forgotten while retrying.
357 ///
358 /// One function for what were four copies of five steps, which had begun to
359 /// differ: only some dropped a scroll view a resync was about to invalidate.
360 fn redial(
361 t: *Tile,
362 alloc: std.mem.Allocator,
363 core: *interact.Core,
364 transport: *client.Transport,
365 target: client.Target,
366 state: *State,
367 /// This tile's agent channels, which the dying connection owned. Dropped
368 /// HERE, and here only, for the reason this function exists at all:
369 /// every call site that had to remember would be one more chance to
370 /// strand a channel on a connection that cannot close it.
371 agents: []?AgentLocal,
372 ) bool {
373 // A close that follows our own detach is the daemon saying goodbye back,
374 // not a tear to heal: the pump wrote the .detach frame and set
375 // `detach_ack` before the save's file I/O window let readFrame see the
376 // daemon's side. Redialing here would re-attach a slot the user just
377 // released.
378 if (t.detach_ack.load(.acquire)) return false;
379 // Before the cold-dial refusal below, which returns without reconnecting
380 // — a pump that ends still owes these fds.
381 dropLocals(agents);
382 // The plain client's rule, kept for the tile a `mux TARGET` is: a
383 // transport that died before a single frame of state carried no
384 // session, so there is nothing to resume and retrying a bad host or a
385 // typo'd `--via` only makes an unkillable client. `session_epoch` is
386 // the right signal because it is set from the first snapshot and never
387 // reset. Wall tiles do the opposite deliberately — they retry forever,
388 // because a wall is a thing you leave up while a box reboots.
389 if (!t.retry_cold and core.rep.session_epoch == 0) {
390 endWith(t, .lost, 1);
391 return false;
392 }
393 transport.close();
394 state.* = .reconnecting;
395 wv.paintLabel(t, state.*);
396 // ...and the same news for a one-tile wall, which has no label
397 // bar on screen to read it off. The corner banner is the plain client's
398 // own, said before the dial rather than inside it for its reason: it is
399 // a PAINT on the session's screen, and the Core is what paints. `banner`
400 // is gated on the sink, so a stripe's re-dial writes nothing here.
401 core.banner("[reconnecting]");
402 // The resync's own paint is what will arrive, so a history page held
403 // here would be silently replaced a moment later.
404 core.dropScrollView();
405 transport.* = dial(alloc, t, target) orelse return false;
406 // Clears `state_since_attach` (so the next exit_status is read as a
407 // refusal again) and drops speculation made against a connection that
408 // no longer exists — the Core's own highlight with it.
409 core.reattached();
410 const have = core.rep.attachArgs();
411 sendAttach(t, transport, have.have_seq, have.have_epoch) catch return false;
412 return true;
413 }
414
415 /// One pump pass's geometry, plus the `.resize` that pass owes the daemon.
416 const Pass = struct {
417 top: u16,
418 left: u16,
419 rows: u16,
420 cols: u16,
421 label_rows: u16,
422 term_cols: u16,
423 term_rows: u16,
424 // A relayout re-cut this tile: the pump owes the daemon THIS pass's
425 // content size.
426 resize: bool,
427 };
428
429 pub fn takePass(t: *Tile) Pass {
430 t.shared.paint_mu.lock();
431 defer t.shared.paint_mu.unlock();
432 // The flag comes out of the SAME hold as the rect it describes. Read a
433 // pass apart from its doorbell and a relayout landing between the two
434 // is swallowed: the pump sends the rect it snapshotted first, clears
435 // the flag, and the daemon keeps a grid the tile has already stopped
436 // painting at — nothing re-sends, because the claim path does not.
437 const owed = t.resize_pending;
438 t.resize_pending = false;
439 return .{
440 .top = t.rect.top,
441 .left = t.rect.left,
442 .rows = t.rect.rows,
443 .cols = t.rect.cols,
444 .label_rows = t.shared.label_rows,
445 .term_cols = t.shared.size.cols,
446 .term_rows = t.shared.size.rows,
447 .resize = owed,
448 };
449 }
450
451 /// Absolute rows count from the oldest row the daemon keeps, and a resync
452 /// renames that space: a kept highlight inverts rows nobody selected.
453 /// One tile's life: dial → attach → replay frames into its Core → repaint
454 /// at its rect. Runs on its own thread (see module header). On transport
455 /// death: reconnect on the CLI's backoff schedule, quoting
456 /// have_seq/have_epoch, and the snapshot-vs-delta resolution does the rest.
457 /// Ends when `running` clears, the session exits, or the attach is refused.
458 ///
459 /// This thread is also the tile's WRITER: every frame the keyboard doorbells
460 /// for — a resize, a detach, a focus claim or release — goes out from here,
461 /// because a Transport has exactly one owning thread (module header).
462 pub fn pumpTile(t: *Tile) void {
463 // FIRST defer, so it runs LAST: every `return` below — a refused
464 // attach, an exited session, a dial the quit interrupted, a Core that
465 // would not initialise — is this tile going quiet for good, and the
466 // keyboard needs to know which tiles it has to paint for. Declared
467 // before the allocator's own defer so nothing can end this thread
468 // without it running.
469 // The bell goes with the store and after it, `endWith`'s reason: the
470 // keyboard's test is `!alive`, so a ring that precedes the store is a
471 // wake-up that finds nothing.
472 defer {
473 t.alive.store(false, .release);
474 wv.ringKeyboard(t.shared);
475 // LAST, after the bell above has finished reading `t.shared`: this
476 // is what hands the slot to `birthTile`, and nothing may touch the
477 // tile after it.
478 t.pump_done.store(true, .release);
479 }
480
481 // Per-thread allocator: nothing allocated here crosses threads except
482 // painted bytes, which go out under the paint mutex.
483 var gpa: std.heap.DebugAllocator(.{}) = .init;
484 defer _ = gpa.deinit();
485 const alloc = gpa.allocator();
486
487 // Whether this tile may narrate and may start a daemon travels IN its
488 // target, set once by whoever made the tile: a picker Enter is an ask,
489 // a poll's list is not. This is the pump's own copy — `t.r.target` is
490 // read by the KEYBOARD thread under `paint_mu` for chord births and is
491 // never written from here — and it is spent below, once.
492 var target = t.r.target;
493
494 // ONE Core per tile, from birth. It owns this tile's replica, its
495 // prediction overlay and its drag for the tile's whole life: the tile
496 // paints from that replica at its own offset, and what decides whether
497 // a keystroke may be speculated at all is the pty's line discipline,
498 // which arrives in `.pty_mode` frames long before the tile is focused.
499 //
500 // `in_fd` is the wall's stdin and this Core never reads it — the
501 // keyboard thread does, on the far side of the mailbox. It is passed
502 // because it is the truth about whether there is a terminal here at all
503 // (`is_tty`), which the mouse split and the side channels are gated on.
504 var core = interact.Core.initSized(
505 alloc,
506 std.posix.STDIN_FILENO,
507 t.shared.out_fd,
508 t.shared.size,
509 ) catch return;
510 defer core.deinit();
511 // Whether this tile has EVER held the terminal, which is the only
512 // question the publish below is gated on: a tile that was never focused
513 // has nothing to say about prediction and must not overwrite what the
514 // tile that was does.
515 var ever_focused = false;
516 // Focus, as this pump knows it. Not `core.claim`: a claim needs a
517 // terminal, and `mux` on a pipe has none, yet its one tile is focused
518 // and its predictions still expire and still get counted.
519 var focused = false;
520 // The last word on this tile's prediction, whichever way the pump ends
521 // — an exit_status arrives and RETURNS, so the per-pass publish inside
522 // the loop is always one pass stale by then.
523 defer if (ever_focused) publishStats(t.shared, core.overlay.counters);
524 // Where this Core's paints land: this tile's rect. The sink admits a
525 // paint whenever the tile has not been forgotten; see `tilePaintBegin`.
526 core.sink = .{ .ctx = t, .begin = tilePaintBegin, .end = tilePaintEnd };
527 // The paint-end hook reads the core's screen cursor; only this pump
528 // thread dereferences it, and the core outlives the pump.
529 t.core = &core;
530 // This thread does not own the exit and cannot print the stats line —
531 // see `Shared.stats`.
532 core.owns_stats = false;
533
534 wv.paintLabel(t, .connecting);
535 // The ENTRY tile arrives with its link already up — dialled on the main
536 // thread, where the tty was, so ssh could prompt. `adopt` is what moves
537 // its QUIC out-queue onto this thread's allocator; see there.
538 var transport = if (t.pre) |pre| blk: {
539 var tr = pre;
540 t.pre = null;
541 tr.adopt(alloc);
542 break :blk tr;
543 } else dial(alloc, t, target) orelse return;
544 defer transport.close();
545 // The ask is SPENT, on whichever of the two branches above got the
546 // link: the entry tile's dial happened on the main thread, a picker
547 // birth's just happened here. Every `redial` below is handed this
548 // copy, so a reconnect can neither start a daemon — `muxd stop` typed
549 // on that box would otherwise be undone by the next backoff, the
550 // poll's bug moved onto a tile — nor print the fallback line onto the
551 // alternate screen the tiles are painted on.
552 if (target == .hand) target.hand.asked = false;
553 // The entry tile's attach carries its rect, so the session is sized to
554 // the terminal the tile claims and no second resize follows —
555 // re-asserting a size the daemon just heard costs one more snapshot on
556 // every `mux`, which is exactly the round trip the convergence must not
557 // add. A view tile attaches at 0x0 (join-only) and `sendAttach` doorbells
558 // its rect behind the attach; that path is inside `sendAttach`.
559 sendAttach(t, &transport, 0, 0) catch {
560 endWith(t, .lost, 1);
561 return;
562 };
563 // One question at a time, with the deadline that makes a daemon too old
564 // to have heard it (`sessions_req` is 0x0c) say so instead of swallowing
565 // every chord for the rest of the session. `client.PendingSwitch`
566 // verbatim — the client asked the same question and this is the same
567 // answer, moved to the thread that owns the link.
568 var pending: client.PendingSwitch = .{};
569 // The ssh-agent channels this tile is serving, one open fd each to this
570 // machine's agent. Everything about them is thread-local: this table,
571 // the fds in it, and the frames that move them all live on this pump.
572 var agent_locals: [proto.agent_chans_max]?AgentLocal = @splat(null);
573 // Every `return` below is this tile going quiet with channels possibly
574 // still open, and the daemon's side of them dies with the transport the
575 // defer above closes.
576 defer dropLocals(&agent_locals);
577
578 var state: State = .connecting;
579 // What this tile's paint is worth: while it matches the wall's
580 // generation the terminal still holds what this thread drew.
581 var painted_gen = t.shared.repaint_gen.load(.acquire);
582 // `gone` ends this thread exactly as `running` does — the defers close
583 // the transport, which frees the daemon slot and NOTHING else. The
584 // session goes on running: "remove is detach".
585 outer: while (t.shared.running.load(.acquire) and !t.gone.load(.acquire)) {
586 // The link and the doorbell, then one fd per live agent channel —
587 // joined into the pump's own poll rather than given a thread each,
588 // because a Transport has exactly one owning thread and these bytes
589 // leave through it. `at` remembers which table slot each of those
590 // trailing fds came from, so a readable one can be traced back to
591 // its channel without a second search.
592 var fdbuf: [2 + agent_locals.len]std.posix.pollfd = undefined;
593 fdbuf[0] = .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
594 fdbuf[1] = .{ .fd = t.wake_r, .events = std.posix.POLL.IN, .revents = 0 };
595 var nfds: usize = 2;
596 var at: [agent_locals.len]usize = undefined;
597 for (agent_locals, 0..) |c, s| if (c) |ch| {
598 at[nfds - 2] = s;
599 fdbuf[nfds] = .{ .fd = ch.fd, .events = std.posix.POLL.IN, .revents = 0 };
600 nfds += 1;
601 };
602 _ = std.posix.poll(fdbuf[0..nfds], transport.timeoutMs(100)) catch return;
603 transport.service();
604 if (fdbuf[1].revents != 0) drainWake(t);
605
606 // The paint offset for this pass: a relayout may have re-cut this
607 // tile's rect, and every paint the pass drives through the Core
608 // has to land in the rect the tile currently owns. Snapshot under
609 // `paint_mu` — the keyboard writes `rect` and `label_rows` under
610 // it — and use the snapshot for the whole pass.
611 const snap = takePass(t);
612 core.row_off = snap.top + snap.label_rows;
613 core.col_off = snap.left;
614 core.owns_screen = snap.top == 0 and snap.left == 0 and
615 snap.label_rows == 0 and snap.cols == snap.term_cols and
616 snap.rows == snap.term_rows;
617 const snap_view_rows: u16 = snap.rows -| snap.label_rows;
618 const snap_view_cols: u16 = snap.cols;
619
620 // FOCUS CLAIM. The keyboard moved the focus onto this tile; the
621 // session's mouse modes and side channels go on here, on the thread
622 // that owns the transport and the Core. The resize the claim used
623 // to send is gone — the attach already carried the rect, and a
624 // relayout doorbells `resize_pending` for any later change.
625 if (t.claim_pending.swap(false, .acq_rel)) {
626 // A refused claim is re-armed rather than lost: the host
627 // picker's popup admits no paint, and a claim writes the
628 // session's modes through the paint sink, so a claim dropped
629 // under it leaves this pump focused holding no terminal — no
630 // mouse modes, no side channels, the notice below eaten — until
631 // the user moves the focus away and back. The picker's close
632 // rings this pump, so the retry is a keystroke away.
633 //
634 // The rest of the PASS still runs whatever comes back:
635 // `takePass` has already cleared `resize_pending`, so skipping
636 // out here would drop a relayout this tile owes the daemon.
637 const step = claimFocus(t, &core, .{ .cols = snap_view_cols, .rows = snap_view_rows });
638 // A stale arm — the keyboard moved the focus between arming
639 // this and this pass — takes nothing: no claim, no modes, no
640 // notice, and these predictions are not the focus's to publish
641 // either. The tile that DOES hold the focus was armed by the
642 // same `setFocus` that took it from this one.
643 focused = step != .dropped;
644 ever_focused = ever_focused or focused;
645 if (step == .done) {
646 // A sentence the keyboard left for whoever owns the terminal
647 // next — a refused `Ctrl-\ c`, so far. Painted here because
648 // a banner belongs to a Core and this is the Core that has
649 // just taken the screen; painted AFTER the repaint below
650 // would be wrong, so it is taken now and shown once the grid
651 // is up.
652 var notice_buf: [96]u8 = undefined;
653 const notice = wv.takeNotice(t.shared, &notice_buf);
654 // The replica has been hot the whole time, so a claim paints
655 // from it NOW rather than waiting for the daemon's answering
656 // snapshot. That is the headline: moving the focus costs a
657 // local repaint, never a wire frame.
658 // ...but only when there IS one. A tile focused before its
659 // first snapshot — the entry tile, on every `mux` — would
660 // otherwise paint a blank grid over the terminal before the
661 // session has said anything, which is a screen the plain
662 // client never drew and bytes a capture never held.
663 if (core.rep.session_epoch != 0) core.repaint() catch {};
664 if (notice.len > 0) core.banner(notice);
665 }
666 }
667 // FOCUS RELEASE. The keyboard moved the focus off this tile and
668 // wrote the session's release itself, under `paint_mu`, before
669 // doorbelling — so the handover is ordered and this pump owes only
670 // its own state. `.already_written` is that discipline.
671 if (t.release_pending.swap(false, .acq_rel)) {
672 focused = false;
673 core.releaseTerminal(.already_written);
674 // The speculation described a screen this terminal no longer
675 // shows.
676 core.overlay.flush();
677 }
678
679 // RELAYOUT DOORBELL: this tile's rect changed. The pump is the
680 // transport's only writer, so relayout sets the flag and the pump
681 // sends the `.resize` from here.
682 if (snap.resize) {
683 core.overlay.setResizePending(true);
684 // The Core clips every paint to its size; a resize the daemon
685 // hears but the Core does not leaves the bottom of the new
686 // grid cut off on screen forever.
687 core.adoptSize(.{ .cols = snap_view_cols, .rows = snap_view_rows });
688 transport.writeFrame(
689 .resize,
690 &proto.encodeSize(snap_view_cols, snap_view_rows),
691 ) catch {
692 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
693 continue :outer;
694 };
695 }
696
697 // `Ctrl-\ d`: hand the daemon its slot back before the process
698 // dies, rather than leaving it for the socket's death to be
699 // noticed. Only this thread may write the frame, so the keyboard
700 // asked and is waiting for the ack.
701 if (t.detach_req.swap(false, .acq_rel)) {
702 transport.writeFrame(.detach, "") catch {};
703 t.detach_ack.store(true, .release);
704 wv.ringKeyboard(t.shared);
705 }
706
707 // A focus chord's question, put on the wire from the thread that
708 // owns the link. The ANSWER is read out of the `.sessions_reply`
709 // arm below and handed to the keyboard, which is the only thread
710 // that may move the focus or grow the wall.
711 const asked: client.SwitchIntent = @enumFromInt(t.ask.swap(0, .acq_rel));
712 if (asked != .none) {
713 // The deadline starts HERE, when the question goes on the wire,
714 // not when the key was typed: it exists to catch a daemon too
715 // old to have heard `sessions_req`, and the time a chord spent
716 // in the mailbox is not that daemon's silence.
717 pending.arm(asked, std.time.milliTimestamp());
718 var end_buf: [proto.end_req_max_len]u8 = undefined;
719 const sent = switch (asked) {
720 // The FORCE is the second press, not a second frame: the
721 // daemon refused the first and this says the user meant it.
722 .end, .end_force => transport.writeFrame(.end_req, proto.encodeEndReq(
723 &end_buf,
724 asked == .end_force,
725 proto.wireName(t.r.session),
726 )),
727 else => transport.writeFrame(.sessions_req, ""),
728 };
729 sent catch {
730 pending.clear();
731 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
732 continue :outer;
733 };
734 }
735
736 // A chord that was never answered. Said with the banner rather than
737 // stderr for the client's reason: the terminal is in raw mode on
738 // the alternate screen and a print there lands mid-grid. Reachable
739 // at all because the poll below is capped at 100ms.
740 //
741 // Read before `expired` spends it: the sentence names the verb the
742 // daemon did not know, and "no session list" over an `x` sends the
743 // user looking at the wrong end of the wire.
744 const waiting = pending.intent;
745 if (pending.expired(std.time.milliTimestamp()))
746 core.banner(switch (waiting) {
747 .end, .end_force => "[daemon too old to end a session]",
748 else => "[no session list: upgrade muxd]",
749 });
750
751 // FRAMES BEFORE KEYS, and that order is load-bearing rather than
752 // arbitrary. What the session has already said must be known before
753 // what the user is saying is interpreted, because the interpreting
754 // depends on it: `Core.forward` splits a wheel notch by whether the
755 // session's application asked for the mouse, and that fact arrives
756 // in a `term_modes` frame. A pass that holds both a readable frame
757 // and a mailbox of keystrokes and drains the mailbox first judges
758 // the keystrokes against a session it has not finished listening
759 // to — the plain client's loop read frames first, and an
760 // application holding the mouse lost its first wheel notch to this
761 // client's scrollback the moment the order was reversed. Found by
762 // the suite, not by reading.
763 //
764 // The `.quic` disjunct is the hub's lesson verbatim: QUIC frames
765 // can arrive from the stream layer with the socket never going
766 // readable.
767 if (fdbuf[0].revents != 0 or transport.link == .quic) frames: {
768 while (true) {
769 const incoming = transport.readFrame(alloc) catch return;
770 const frame = switch (incoming) {
771 .frame => |f| f,
772 .incomplete => break :frames,
773 .closed => {
774 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
775 continue :outer;
776 },
777 };
778 defer frame.deinit(alloc);
779 // Everything the frame means to the replica and the screen,
780 // in the one place that switch lives. The Core paints at
781 // this tile's offset through the sink, and writes side
782 // channels only while it holds the claim — so a tile that
783 // is not focused still keeps its replica hot and its grid
784 // on its rect, but its title and mouse modes reach no
785 // terminal until the focus comes to it.
786 //
787 // The one exception is deliberate and is `.pty_mode`: the
788 // Core feeds the overlay's mode whatever the claim, because
789 // the pty's line discipline is the entire gate on
790 // speculation — a password prompt must never be predicted —
791 // and a claim has to start from the truth rather than
792 // from `.never` and a round trip.
793 const routed = core.frame(frame.type, frame.payload) catch break :frames;
794 switch (routed) {
795 .skip, .handled => {},
796 .state => {
797 if (state != .up) {
798 state = .up;
799 t.ever_up.store(true, .release);
800 wv.paintLabel(t, state);
801 }
802 },
803 // The replica is suspect, not the transport: re-attach
804 // quoting (0,0) explicitly — a quoted seq would invite
805 // the delta that cannot fix us.
806 .resync => {
807 core.rep.state_since_attach = false;
808 // A resync renames the absolute row space, so a
809 // held drag now names rows nobody selected — and
810 // on an EPOCH change `sel_range` still matches it,
811 // so an in-flight reply would copy the new session's
812 // text. `redial` drops it for this reason; this path
813 // re-attaches without going through it.
814 core.drag.clear();
815 sendAttach(t, &transport, 0, 0) catch return;
816 },
817 .not_mine => switch (frame.type) {
818 .exit_status => {
819 // Before any replay frame this is the refusal
820 // path; after, the session really ended.
821 const landed = core.rep.state_since_attach;
822 state = if (landed) .exited else .refused;
823 wv.paintLabel(t, state);
824 endWith(
825 t,
826 if (landed) .exited else .refused,
827 if (landed and frame.payload.len >= 1) frame.payload[0] else 1,
828 );
829 return;
830 },
831 .taken_over => {
832 // Unsent by this daemon (wire-compat only), but
833 // the plain client had an answer for it and the
834 // converged one keeps it: somebody else took the
835 // session, so this tile is finished rather than
836 // reconnecting into a fight over the grid.
837 state = .exited;
838 wv.paintLabel(t, state);
839 endWith(t, .taken, 0);
840 return;
841 },
842 .sessions_reply => {
843 // Gated on the intent this tile's own chord
844 // armed: only a question we asked may move the
845 // focus, so an unasked-for reply is ignored. The
846 // intent is SPENT here whichever name it picks
847 // — one question, one answer.
848 var name_buf: [proto.session_name_max]u8 = undefined;
849 // A list is only ever asked for to NAME a new
850 // session: `n`/`p` walk the wall's own tiles and
851 // ask nothing, and the end verbs are answered by
852 // `end_reply`. Anything else here is a reply to
853 // a question this tile did not put.
854 const pick: ?[]const u8 = switch (pending.take()) {
855 .new => client.nextFreeName(&name_buf, frame.payload),
856 else => null,
857 };
858 if (pick) |p| postAnswer(t, p);
859 },
860 .end_reply => {
861 const r = proto.parseEndReply(frame.payload) orelse break :frames;
862 _ = pending.take();
863 wv.onEndReply(t, r, std.time.milliTimestamp());
864 // An ACCEPTED end says nothing: the hangup's
865 // `exit_status` is on its way and the `.exited`
866 // path narrates it, so a banner here would be
867 // the client congratulating itself ahead of the
868 // daemon. Not a `break` either — that exit
869 // status is very likely the next frame in this
870 // same pass, and leaving it for the next poll
871 // is 100ms of a tile that is already gone.
872 if (!r.accepted) {
873 var b: [96]u8 = undefined;
874 core.banner(if (r.others == 0)
875 endRefusal(r.reason)
876 else
877 std.fmt.bufPrint(&b, "[{d} other{s} attached - x again to end]", .{
878 r.others,
879 if (r.others == 1) "" else "s",
880 }) catch "[others attached - x again to end]");
881 }
882 },
883 .selection_reply => copySelection(t, alloc, &core, frame.payload),
884 .agent_open => {
885 const id = proto.decodeAgentId(frame.payload) catch break :frames;
886 // Every refusal is the same answer on the wire —
887 // a channel that closes without a byte, which the
888 // far side's ssh reads as "no agent" and gives up
889 // on rather than hanging. Only one of the four
890 // reasons is ordinary — no agent on this machine;
891 // the rest are a daemon asking for something it
892 // should not.
893 const opened = openAgentChan(
894 &agent_locals,
895 id,
896 t.r.agent,
897 std.posix.getenv(proto.agent_sock_env) orelse "",
898 );
899 if (!opened)
900 transport.writeFrame(
901 .agent_close,
902 &proto.encodeAgentId(id),
903 ) catch {};
904 },
905 .agent_data => if (!deliverAgentData(
906 &agent_locals,
907 frame.payload,
908 &transport,
909 )) break :frames,
910 .agent_close => {
911 const id = proto.decodeAgentId(frame.payload) catch break :frames;
912 // Silent, mirroring the daemon: it has already
913 // retired this id, so an `agent_close` back would
914 // be an echo it has to learn to ignore.
915 if (findLocal(&agent_locals, id)) |s| {
916 const ch = agent_locals[s].?;
917 agent_locals[s] = null;
918 std.posix.close(ch.fd);
919 }
920 },
921 // MsgType is an open enum, so the compiler still
922 // wants an arm for everything `.not_mine` cannot be.
923 else => {},
924 },
925 }
926 // Only the socket link guarantees one readable event is
927 // one frame; QUIC may have buffered more.
928 //
929 // But "one event, one frame" is not "one pass, one frame".
930 // A resync is a BURST — snapshot, pty mode, title, terminal
931 // modes — and stopping after the first leaves the rest to be
932 // read a pass at a time, with the user's keystrokes
933 // interleaved between them. `Core.forward` decides what a
934 // wheel notch MEANS from the session's terminal modes, so a
935 // notch judged before the `term_modes` at the END of that
936 // burst is stolen for this client's scrollback instead of
937 // reaching the application that asked for the mouse. A
938 // single-threaded client hid the hazard by being busy
939 // painting the snapshot when the notch arrived; a keyboard
940 // on its own thread notices it immediately, and the suite
941 // caught it two runs out of three.
942 //
943 // So: drain what has already ARRIVED. A zero timeout waits
944 // for nothing, which is what keeps this a drain and not a
945 // second blocking read.
946 //
947 // It cannot starve the rest of the pass, and two measured
948 // facts are why. The daemon coalesces its deltas, so a
949 // loud session hands over a bounded burst and `poll(0)`
950 // finds the socket empty within it rather than a stream
951 // that refills as fast as it is read. And the keyboard is
952 // a different thread with its own 400ms bound on the one
953 // thing it ever waits for a pump to do (`awaitDetach`), so
954 // even a pump that did stall here could not hold the
955 // terminal hostage — which is the failure this would
956 // otherwise have to be argued safe against.
957 if (transport.link != .quic) {
958 var more = [_]std.posix.pollfd{
959 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
960 };
961 const ready = std.posix.poll(&more, 0) catch 0;
962 if (ready == 0 or more[0].revents == 0) break :frames;
963 }
964 }
965 }
966
967 // The other direction: what this machine's agent answered, back out
968 // as `agent_data`. AFTER the frame drain, deliberately — the table
969 // this walks is then the one the daemon's latest word left behind,
970 // so a channel it has just closed is already gone rather than read
971 // once more on its way out.
972 for (2..nfds) |i| {
973 if (fdbuf[i].revents == 0) continue;
974 const s = at[i - 2];
975 const ch = agent_locals[s] orelse continue;
976 // The id is written into the head of the very buffer the read
977 // fills, so a frame costs no second copy. The read is capped at
978 // the wire's `agent_data_max` because that cap is what keeps one
979 // busy channel from holding the link against the session's own
980 // bytes — the same bound the daemon reads its end with.
981 var buf: [proto.agent_id_len + proto.agent_data_max]u8 = undefined;
982 buf[0..proto.agent_id_len].* = proto.encodeAgentId(ch.id);
983 // DONTWAIT, and it is load-bearing rather than belt-and-braces:
984 // the poll above happened before the frame drain, so the daemon
985 // may since have closed this slot's channel and an `agent_open`
986 // in the same burst may have refilled the slot with a fresh
987 // connection that has said nothing yet. A blocking read there
988 // would stop the whole tile — its keystrokes, its paints — on a
989 // socket that is merely idle. `agent_locals[s]` is re-read above,
990 // so whatever this does return is attributed to the id that
991 // actually owns the fd.
992 //
993 // EOF and a broken connection are one case: the agent is done
994 // with this channel either way, and the far side needs to hear so.
995 const n = std.posix.recv(
996 ch.fd,
997 buf[proto.agent_id_len..],
998 std.posix.MSG.DONTWAIT,
999 ) catch |err| switch (err) {
1000 error.WouldBlock => continue,
1001 else => 0,
1002 };
1003 if (n == 0) {
1004 closeLocal(&agent_locals, s, &transport);
1005 continue;
1006 }
1007 transport.writeFrame(.agent_data, buf[0 .. proto.agent_id_len + n]) catch {
1008 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1009 continue :outer;
1010 };
1011 }
1012
1013 // Whatever the keyboard left, through everything a plain client's
1014 // keystrokes go through: the mouse split, the wheel, alternate
1015 // scroll, the scrollback view, the prediction and the input frame.
1016 // Only the focused tile's mailbox is ever written, but every pump
1017 // drains its own — an empty mailbox is the common, cheap answer.
1018 var keys_buf: [wv.mailbox_max]u8 = undefined;
1019 const keys = takeKeys(t, &keys_buf);
1020 if (keys.len > 0) {
1021 // A paint that would not allocate is not a dead link: the
1022 // replica is untouched and the next frame redraws from it.
1023 const step = core.forward(&transport, keys) catch interact.Step.ok;
1024 if (step == .lost) {
1025 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1026 continue :outer;
1027 }
1028 // Input is moving again, so "input dropped" has stopped being
1029 // news. Repainted rather than merely cleared, because the bar
1030 // is still carrying the old sentence until something draws
1031 // over it — and only when there is a bar, which `paintLabel`
1032 // already decides.
1033 if (t.in_dropped.swap(false, .acq_rel)) wv.paintLabel(t, state);
1034 }
1035
1036 // A prediction the daemon never answered must not sit on the
1037 // screen forever, and only the clock can say so — no frame will.
1038 // Only while focused: an unfocused tile's rect shows no prediction
1039 // to retire.
1040 if (focused) {
1041 core.idle() catch {};
1042 publishStats(t.shared, core.overlay.counters);
1043 }
1044
1045 // A relayout re-cut the stripes (or cleared the screen for an empty
1046 // wall), and a quiet session sends nothing to trigger a repaint.
1047 // The replica is current — the SCREEN is not — so the generation is
1048 // what puts it back. Checked on the poll timeout, so a tile comes
1049 // back within ~100ms of a relayout whether or not its session ever
1050 // speaks again. The drag clears here too: a relayout moved the rect
1051 // a held drag's anchor was resolved against.
1052 const gen = t.shared.repaint_gen.load(.acquire);
1053 if (gen != painted_gen) {
1054 core.drag.clear();
1055 wv.paintLabel(t, state);
1056 // A 0-row rect (fullscreened out) paints nothing: the daemon
1057 // refused the 0×0 resize, so the session keeps its real grid.
1058 if (snap_view_rows > 0) core.repaint() catch {};
1059 painted_gen = gen;
1060 }
1061 }
1062 }
1063
1064 /// A refusal in THIS client's words. `parseEndReply` hands back the frame's
1065 /// tail unfiltered and `paintBanner` writes it verbatim, so a peer's
1066 /// `\x1b]0;..\x07` would run outside the replica. The reasons are constants.
1067 pub fn endRefusal(reason: []const u8) []const u8 {
1068 if (std.mem.eql(u8, reason, proto.end_reason.no_session))
1069 return "[no such session on that daemon]";
1070 if (std.mem.eql(u8, reason, proto.end_reason.bad_frame))
1071 return "[the daemon could not read the end request]";
1072 return "[the daemon refused to end this session]";
1073 }
src/tui/wallview.zig
Old New
@@ -38,8 +38,11 @@ const layout = @import("layout");
38 const TmpDir = @import("testtmp").TmpDir; 38 const TmpDir = @import("testtmp").TmpDir;
39 const wall_host = @import("wall_host.zig"); 39 const wall_host = @import("wall_host.zig");
40 const wall_picker = @import("wall_picker.zig"); 40 const wall_picker = @import("wall_picker.zig");
41 const wall_pump = @import("wall_pump.zig");
41 const AddHost = wall_host.AddHost; 42 const AddHost = wall_host.AddHost;
43 const AgentLocal = wall_pump.AgentLocal;
42 const BirthNames = wall_host.BirthNames; 44 const BirthNames = wall_host.BirthNames;
45 const ClaimStep = wall_pump.ClaimStep;
43 const Host = wall_host.Host; 46 const Host = wall_host.Host;
44 const PickerAuto = wall_picker.PickerAuto; 47 const PickerAuto = wall_picker.PickerAuto;
45 const PickerBody = wall_picker.PickerBody; 48 const PickerBody = wall_picker.PickerBody;
@@ -149,14 +152,14 @@ fn doResize(
149 return moved; 152 return moved;
150 } 153 }
151 154
152 const State = enum { 155 pub const State = enum {
153 connecting, 156 connecting,
154 up, 157 up,
155 reconnecting, 158 reconnecting,
156 exited, 159 exited,
157 refused, 160 refused,
158 161
159 fn word(s: State) []const u8 { 162 pub fn word(s: State) []const u8 {
160 return switch (s) { 163 return switch (s) {
161 .connecting => "connecting", 164 .connecting => "connecting",
162 .up => "up", 165 .up => "up",
@@ -170,7 +173,7 @@ const State = enum {
170 /// The mailbox: how many typed bytes can wait for a pump to put them on 173 /// The mailbox: how many typed bytes can wait for a pump to put them on
171 /// the wire. One stdin read's worth, which is all a keyboard can produce 174 /// the wire. One stdin read's worth, which is all a keyboard can produce
172 /// between two turns of a pump's poll loop. 175 /// between two turns of a pump's poll loop.
173 const mailbox_max = 4096; 176 pub const mailbox_max = 4096;
174 177
175 /// The most tiles one wall can hold. 178 /// The most tiles one wall can hold.
176 /// 179 ///
@@ -468,11 +471,11 @@ pub const Tile = struct {
468 /// doorbelling — the handover is ordered, not eventual. 471 /// doorbelling — the handover is ordered, not eventual.
469 release_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), 472 release_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
470 473
471 fn viewRows(t: *const Tile) u16 { 474 pub fn viewRows(t: *const Tile) u16 {
472 return t.rect.rows -| t.shared.label_rows; 475 return t.rect.rows -| t.shared.label_rows;
473 } 476 }
474 477
475 fn viewCols(t: *const Tile) u16 { 478 pub fn viewCols(t: *const Tile) u16 {
476 return t.rect.cols; 479 return t.rect.cols;
477 } 480 }
478 }; 481 };
@@ -551,7 +554,7 @@ fn intentForEnd(t: *const Tile, now: i64) client.SwitchIntent {
551 } 554 }
552 555
553 /// An accepted end DISARMS: the window must not outlive its session. 556 /// An accepted end DISARMS: the window must not outlive its session.
554 fn onEndReply(t: *Tile, r: proto.EndReply, now: i64) void { 557 pub fn onEndReply(t: *Tile, r: proto.EndReply, now: i64) void {
555 t.end_armed_until.store(if (r.accepted) 0 else now + end_arm_ms, .release); 558 t.end_armed_until.store(if (r.accepted) 0 else now + end_arm_ms, .release);
556 } 559 }
557 560
@@ -584,7 +587,7 @@ fn stepLive(tiles: []const Tile, present: []const bool, sel: usize, forward: boo
584 587
585 /// The state is remembered on the tile: the keyboard repaints this bar 588 /// The state is remembered on the tile: the keyboard repaints this bar
586 /// with no frame, and only when the wall shows more than one tile. 589 /// with no frame, and only when the wall shows more than one tile.
587 fn paintLabel(t: *Tile, state: State) void { 590 pub fn paintLabel(t: *Tile, state: State) void {
588 t.shared.paint_mu.lock(); 591 t.shared.paint_mu.lock();
589 defer t.shared.paint_mu.unlock(); 592 defer t.shared.paint_mu.unlock();
590 t.state = state; 593 t.state = state;
@@ -673,7 +676,7 @@ fn paintDeadBarsLocked(tiles: []Tile) void {
673 /// wider than a left-hand pane must not cross the rail into its neighbour, 676 /// wider than a left-hand pane must not cross the rail into its neighbour,
674 /// and one right-aligned to the SCREEN lands in whichever pane owns that 677 /// and one right-aligned to the SCREEN lands in whichever pane owns that
675 /// corner. The only owner of a per-tile banner, for exactly that reason. 678 /// corner. The only owner of a per-tile banner, for exactly that reason.
676 fn tileBanner(t: *Tile, text: []const u8) void { 679 pub fn tileBanner(t: *Tile, text: []const u8) void {
677 if (!t.shared.is_tty) return; 680 if (!t.shared.is_tty) return;
678 // The tail, so the cursor end of a long spelling is what is on screen. 681 // The tail, so the cursor end of a long spelling is what is on screen.
679 // Bounded by the paint's own cap as well as the rect: a label past it 682 // Bounded by the paint's own cap as well as the rect: a label past it
@@ -689,185 +692,8 @@ fn tileBanner(t: *Tile, text: []const u8) void {
689 paint.paintBanner(t.shared.out_fd, t.rect.cols, shown, t.rect.top + t.shared.label_rows, t.rect.left); 692 paint.paintBanner(t.shared.out_fd, t.rect.cols, shown, t.rect.top + t.shared.label_rows, t.rect.left);
690 } 693 }
691 694
692 /// Under `paint_mu`: a clear spliced into a 64 KiB OSC 52 write eats the
693 /// paint after it.
694 fn copySelection(
695 t: *Tile,
696 alloc: std.mem.Allocator,
697 core: *interact.Core,
698 payload: []const u8,
699 ) void {
700 var answer: interact.Copy = .none;
701 {
702 t.shared.paint_mu.lock();
703 defer t.shared.paint_mu.unlock();
704 // The drag is this tile's own Core's now: a drag is per tile, and
705 // the pump that owns the link is the one whose Core holds it.
706 const held = core.drag.range();
707 answer = core.selectionCopy(payload, held);
708 switch (answer) {
709 // `is_tty` and not the tile's claim: a tile that has released
710 // the terminal can still be the one whose finished drag the
711 // answer reaches, and the copy is still the user's. A piped
712 // `mux` asks its terminal for no mouse modes, so it can have no
713 // drag to copy in the first place.
714 .text => |text| if (t.shared.is_tty)
715 interact.writeSelectionCopy(alloc, t.shared.out_fd, text) catch {},
716 .none, .too_large => {},
717 }
718 }
719 // Outside the hold: `tileBanner` takes the same lock, which is not
720 // reentrant. The banner lands in this tile's own rect.
721 if (answer == .too_large) tileBanner(t, "[selection too large to copy]");
722 }
723
724 /// Every live tile paints its own rect, so the sink admits a paint whenever
725 /// this tile has not been forgotten. `paint_mu` is held for the whole paint.
726 fn tilePaintBegin(ctx: ?*anyopaque) bool {
727 const t: *Tile = @ptrCast(@alignCast(ctx.?));
728 if (t.gone.load(.acquire)) return false;
729 t.shared.paint_mu.lock();
730 // Read UNDER the lock, not before it: a pump that tested the flag and
731 // then lost the race for `paint_mu` would paint its rect on top of the
732 // box the keyboard had just drawn — and `picker_stamp` suppresses the
733 // identical repaint that would have repaired it, so the damage sticks
734 // until a key changes the frame.
735 if (t.shared.picker_open.load(.acquire)) {
736 t.shared.paint_mu.unlock();
737 return false;
738 }
739 return true;
740 }
741
742 fn tilePaintEnd(ctx: ?*anyopaque) void {
743 const t: *Tile = @ptrCast(@alignCast(ctx.?));
744 defer t.shared.paint_mu.unlock();
745 // Before releasing the terminal: the cursor belongs to the FOCUSED tile.
746 // A focused paint records where its cursor landed; an unfocused paint's
747 // last act is to put the cursor back there, so a redraw in tile 2 cannot
748 // steal the eye while the keys go to tile 1. The cursor is hidden for
749 // the move so the show after it does not flash it at the unfocused
750 // tile's final position before the CUP lands — then re-shown so the
751 // focused tile's cursor rests visible until its next paint.
752 if (t.idx == t.shared.sel) {
753 if (t.core) |core| t.shared.cursor = core.screenCursor();
754 } else {
755 var cbuf: [26]u8 = undefined;
756 const cup = std.fmt.bufPrint(&cbuf, "\x1b[?25l\x1b[{d};{d}H\x1b[?25h", .{ t.shared.cursor.y + 1, t.shared.cursor.x + 1 }) catch return;
757 proto.writeAllFd(t.shared.out_fd, cup) catch {};
758 }
759 }
760
761 /// The one place a wall tile puts an attach on the wire. A tile the user
762 /// asked for — the entry tile, a chord-born tile, a local line off the
763 /// saved wall — claims its rect on attach, and that size is what lets the
764 /// daemon create the session. A view tile (wall argv, a remote saved
765 /// line) attaches at 0x0 so it can only JOIN, then takes its rect with the
766 /// resize doorbell one frame later — every tile still claims its
767 /// rectangle, and a redial comes back claiming what the tile claimed.
768 fn sendAttach(t: *Tile, tr: *client.Transport, have_seq: u64, have_epoch: u64) !void {
769 // Snapshot under `paint_mu`: the keyboard may relayout (re-cut stripes,
770 // resize) concurrently with the pump's first attach.
771 const snap = blk: {
772 t.shared.paint_mu.lock();
773 defer t.shared.paint_mu.unlock();
774 break :blk .{
775 .cols = t.viewCols(),
776 .view_rows = t.viewRows(),
777 };
778 };
779 const cols: u16 = if (t.creates) snap.cols else 0;
780 const rows: u16 = if (t.creates) snap.view_rows else 0;
781 var buf: [proto.attach_max_len]u8 = undefined;
782 try tr.writeFrame(.attach, proto.encodeAttachNamed(
783 &buf,
784 cols,
785 rows,
786 have_seq,
787 have_epoch,
788 proto.wireName(t.r.session),
789 ));
790 // A view tile made no size claim, so it owes its rect now: the same
791 // doorbell path a relayout takes (adoptSize + .resize), run on this
792 // pump thread which is the transport's only writer.
793 if (!t.creates) {
794 t.shared.paint_mu.lock();
795 t.resize_pending = true;
796 t.shared.paint_mu.unlock();
797 ring(t);
798 }
799 // Re-armed on EVERY attach, not once per process: a redial is a fresh
800 // attach onto a fresh daemon-side slot, which remembers no offer. Empty
801 // payload, and a daemon too old to know the frame skips it.
802 if (t.r.agent) try tr.writeFrame(.agent_offer, "");
803 }
804
805 /// What a pump owes itself after asking for its focus claim.
806 const ClaimStep = enum {
807 /// The claim landed, or the Core already held it: finish the pass.
808 done,
809 /// The sink refused and this tile still has the focus: try next pass.
810 rearmed,
811 /// The focus moved on while the popup was up: this arm is stale.
812 dropped,
813 };
814
815 /// May this tile take the terminal for the arm it is holding?
816 fn claimAllowed(t: *Tile) bool {
817 // BEFORE the claim, never after. The keyboard arms `claim_pending` and
818 // the pump reads it a pass later, and the focus can move in between —
819 // a poller's tile arriving, the birth an Enter makes, all of it under
820 // the host picker's popup. An arm that outlives its focus and then
821 // SUCCEEDS puts two tiles' modes on one terminal, rests the cursor on
822 // the loser, and takes the focus notice with it; the outgoing tile's
823 // release was consumed a pass earlier, so nothing undoes any of it.
824 //
825 // Judging the claim's REFUSAL cannot cover this: the arm that matters
826 // is the one retried after the popup closed, which the sink admits.
827 t.shared.paint_mu.lock();
828 defer t.shared.paint_mu.unlock();
829 return t.shared.sel == t.idx;
830 }
831
832 /// The wall's ONLY door to `Core.claimTerminal`: the focus test, the size
833 /// adopt, and the claim, in that order and never apart. A caller that could
834 /// reach the claim around this is a caller that can mint a stale one.
835 fn claimFocus(t: *Tile, core: *interact.Core, rect: proto.Size) ClaimStep {
836 if (!claimAllowed(t)) return .dropped;
837 // A tile born before somebody resized the terminal clips its paints to
838 // a screen that is gone; the tile's current RECT, never the whole
839 // terminal, which on a wall of two would let it paint over a neighbour.
840 if (core.size.cols != rect.cols or core.size.rows != rect.rows)
841 core.adoptSize(rect);
842 return afterClaim(t, core.claimTerminal(), core.claim != .none, core.is_tty);
843 }
844
845 /// A refused focus claim, judged.
846 fn afterClaim(t: *Tile, claimed: bool, held: bool, is_tty: bool) ClaimStep {
847 // `held` is `Core.claim != .none`. A claim answers false for three
848 // reasons and only ONE of them is worth another pass: not a tty (there
849 // is no terminal to hold), already held (re-arming would re-take the
850 // focus notice every pass, forever), and the SINK refused — which is
851 // the host picker, whose popup admits no paint and a claim writes the
852 // session's modes through one. Read off the Core, never off
853 // `picker_open` a second time: the keyboard can clear that flag between
854 // the two loads, and then the refusal is dropped exactly as before.
855 if (claimed or held or !is_tty) return .done;
856 t.shared.paint_mu.lock();
857 defer t.shared.paint_mu.unlock();
858 // ...and only while this tile is STILL the focus. `claimAllowed` asked
859 // the same question before the claim; the keyboard can answer it
860 // differently in between, and re-arming then would mint the very stale
861 // arm that check exists to stop. Left CLEAR rather than cleared — the
862 // caller's `swap` did that — because a focus that came back inside this
863 // window re-armed it legitimately.
864 if (t.shared.sel != t.idx) return .dropped;
865 t.claim_pending.store(true, .release);
866 return .rearmed;
867 }
868
869 /// Never blocks and never reports: a full pipe is a bell already ringing. 695 /// Never blocks and never reports: a full pipe is a bell already ringing.
870 fn ring(t: *const Tile) void { 696 pub fn ring(t: *const Tile) void {
871 _ = std.posix.write(t.wake_w, "\x00") catch {}; 697 _ = std.posix.write(t.wake_w, "\x00") catch {};
872 } 698 }
873 699
@@ -896,21 +722,13 @@ pub fn ringKeyboard(shared: *const Shared) void {
896 722
897 /// Both ends of both bells are non-blocking, which is what makes this safe 723 /// Both ends of both bells are non-blocking, which is what makes this safe
898 /// on a pipe nobody has rung. 724 /// on a pipe nobody has rung.
899 fn drainBell(fd: std.posix.fd_t) void { 725 pub fn drainBell(fd: std.posix.fd_t) void {
900 var sink: [64]u8 = undefined; 726 var sink: [64]u8 = undefined;
901 while (std.posix.read(fd, &sink)) |n| { 727 while (std.posix.read(fd, &sink)) |n| {
902 if (n < sink.len) break; 728 if (n < sink.len) break;
903 } else |_| {} 729 } else |_| {}
904 } 730 }
905 731
906 /// FOCUSED pumps only: a tile that never held the terminal has zero
907 /// counters, and publishing them would clobber the tile the user typed at.
908 fn publishStats(shared: *Shared, c: interact.PredictCounters) void {
909 shared.paint_mu.lock();
910 defer shared.paint_mu.unlock();
911 shared.stats = c;
912 }
913
914 /// Leave a sentence for whichever tile owns the terminal next. Keyboard 732 /// Leave a sentence for whichever tile owns the terminal next. Keyboard
915 /// thread only, and only for a focus it is about to move — the pump that 733 /// thread only, and only for a focus it is about to move — the pump that
916 /// lands there paints it as a banner on its claim. 734 /// lands there paints it as a banner on its claim.
@@ -934,7 +752,7 @@ pub fn setNoticeIdle(shared: *Shared, text: []const u8) void {
934 } 752 }
935 753
936 /// Take it, once. 754 /// Take it, once.
937 fn takeNotice(shared: *Shared, out: []u8) []const u8 { 755 pub fn takeNotice(shared: *Shared, out: []u8) []const u8 {
938 // Copied out because the caller paints it after releasing the lock: 756 // Copied out because the caller paints it after releasing the lock:
939 // `Core.banner` takes `paint_mu` itself, through the sink, and the 757 // `Core.banner` takes `paint_mu` itself, through the sink, and the
940 // sink is not reentrant. 758 // sink is not reentrant.
@@ -943,27 +761,6 @@ fn takeNotice(shared: *Shared, out: []u8) []const u8 {
943 return takeNoticeLocked(shared, out); 761 return takeNoticeLocked(shared, out);
944 } 762 }
945 763
946 /// Validated first: these bytes came out of a peer's `sessions_reply` and
947 /// `SessionName.of` memcpys with no bound of its own (`client.validPick`).
948 /// A name it refuses is a name nobody is moved to.
949 fn postAnswer(t: *Tile, pick: []const u8) void {
950 const name = client.validPick(pick) orelse return;
951 {
952 t.ans_mu.lock();
953 defer t.ans_mu.unlock();
954 t.ans = name;
955 }
956 t.ans_ready.store(true, .release);
957 ringKeyboard(t.shared);
958 }
959
960 /// Written before `alive` clears, so no dead tile is ever seen without a
961 /// reason. Rings NOTHING — the keyboard's test is `!alive`.
962 fn endWith(t: *Tile, reason: EndReason, code: u8) void {
963 t.code.store(code, .release);
964 t.end.store(@intFromEnum(reason), .release);
965 }
966
967 /// Called ONLY from the keyboard loop and ONLY while `t` is the focused tile: 764 /// Called ONLY from the keyboard loop and ONLY while `t` is the focused tile:
968 /// that restriction IS the enforcement of "an unfocused tile claims no terminal modes". 765 /// that restriction IS the enforcement of "an unfocused tile claims no terminal modes".
969 /// 766 ///
@@ -990,848 +787,6 @@ fn sendKeys(t: *Tile, keys: []const u8) void {
990 ring(t); 787 ring(t);
991 } 788 }
992 789
993 /// Whole-mailbox chunking, so `offerKeystroke` (one-byte chunks only) counts
994 /// keystrokes that arrive between polls as suppressed: a wall predicts a
995 /// little less. Splitting would speculate against a stale replica.
996 fn takeKeys(t: *Tile, out: []u8) []u8 {
997 t.in_mu.lock();
998 defer t.in_mu.unlock();
999 const n = @min(out.len, t.in_len);
1000 @memcpy(out[0..n], t.in[0..n]);
1001 std.mem.copyForwards(u8, t.in[0 .. t.in_len - n], t.in[n..t.in_len]);
1002 t.in_len -= n;
1003 return out[0..n];
1004 }
1005
1006 fn drainWake(t: *const Tile) void {
1007 drainBell(t.wake_r);
1008 }
1009
1010 fn dial(alloc: std.mem.Allocator, t: *Tile, target_in: client.Target) ?client.Transport {
1011 var target = target_in;
1012 var backoff_ms: u64 = 0;
1013 // `gone` as well as `running`: a tile forgotten while it is retrying a
1014 // dead host must stop retrying, not keep a thread and a backoff alive
1015 // for a tile that is no longer on the wall.
1016 while (t.shared.running.load(.acquire) and !t.gone.load(.acquire)) {
1017 if (client.Transport.open(alloc, target, null, -1)) |tr| return tr else |_| {}
1018 // An ask buys ONE attempt. Every retry below is the wall's own
1019 // idea: a `muxd start` per backoff would restart a daemon for as
1020 // long as the tile lives, and a fallback line per backoff would
1021 // scroll the alternate screen the tiles are painted on.
1022 if (target == .hand) target.hand.asked = false;
1023 backoff_ms = client.nextBackoffMs(backoff_ms);
1024 // Sliced sleep so quit is never behind a full backoff.
1025 var slept: u64 = 0;
1026 while (slept < backoff_ms and t.shared.running.load(.acquire) and
1027 !t.gone.load(.acquire)) : (slept += 50)
1028 {
1029 std.Thread.sleep(50 * std.time.ns_per_ms);
1030 }
1031 }
1032 return null;
1033 }
1034
1035 /// One forwarded ssh-agent channel, this end of it: the id the daemon
1036 /// allocated, and an fd to THIS machine's agent.
1037 ///
1038 /// Thread-local by construction — the table lives in `pumpTile`'s frame and
1039 /// no other thread can see it, which is why nothing here takes a lock and
1040 /// why these helpers take the table as a slice rather than reaching for one.
1041 const AgentLocal = struct { id: u32, fd: std.posix.fd_t };
1042
1043 /// Fixed at `proto.agent_chans_max`, which is what the daemon opens anyway:
1044 /// a full table costs one failed lookup, not the pump's hot path an alloc.
1045 fn storeLocal(locals: []?AgentLocal, id: u32, fd: std.posix.fd_t) ?usize {
1046 for (locals, 0..) |c, s| {
1047 if (c != null) continue;
1048 locals[s] = .{ .id = id, .fd = fd };
1049 return s;
1050 }
1051 return null;
1052 }
1053
1054 fn findLocal(locals: []?AgentLocal, id: u32) ?usize {
1055 for (locals, 0..) |c, s| if (c) |ch| {
1056 if (ch.id == id) return s;
1057 };
1058 return null;
1059 }
1060
1061 /// Hang one channel up from this end and say so, because the daemon is
1062 /// holding the far socket open waiting for bytes that are not coming. A
1063 /// failed write is the transport itself being gone, which the caller's next
1064 /// pass turns into a redial.
1065 fn closeLocal(locals: []?AgentLocal, slot: usize, transport: *client.Transport) void {
1066 const ch = locals[slot] orelse return;
1067 locals[slot] = null;
1068 std.posix.close(ch.fd);
1069 transport.writeFrame(.agent_close, &proto.encodeAgentId(ch.id)) catch {};
1070 }
1071
1072 /// Lifted out of the pump so the `offered` gate has a seam a test can watch.
1073 fn openAgentChan(
1074 locals: []?AgentLocal,
1075 id: u32,
1076 offered: bool,
1077 sock: []const u8,
1078 ) bool {
1079 // The OFFER is the consent, and it is per TILE: a wall where one tile
1080 // was typed with `-A` must not hand another tile's host the keys,
1081 // whoever asks.
1082 if (!offered) return false;
1083 // A live id reused. Refusing keeps the channel already on that id
1084 // intact, which is the half of the collision that has real bytes moving
1085 // through it.
1086 if (findLocal(locals, id) != null) return false;
1087 const fd = client.connectAgent(sock) orelse return false;
1088 if (storeLocal(locals, id, fd) == null) {
1089 std.posix.close(fd);
1090 return false;
1091 }
1092 return true;
1093 }
1094
1095 /// Lifted out of the pump so the length cap has a seam a test can reach.
1096 fn deliverAgentData(
1097 locals: []?AgentLocal,
1098 payload: []const u8,
1099 transport: *client.Transport,
1100 ) bool {
1101 const id = proto.decodeAgentId(payload) catch return false;
1102 const s = findLocal(locals, id) orelse return true;
1103 if (proto.agentDataOversize(payload)) {
1104 closeLocal(locals, s, transport);
1105 return true;
1106 }
1107 // Bytes onto the fd in order, never parsed and never reassembled: the
1108 // daemon reads the far end in `agent_data_max` bites, so one agent
1109 // message can arrive as several frames and several messages as one. The
1110 // agent protocol delimits itself over a stream, and this end is a pipe.
1111 proto.writeAllFd(locals[s].?.fd, payload[proto.agent_id_len..]) catch
1112 closeLocal(locals, s, transport);
1113 return true;
1114 }
1115
1116 /// Redial only: these belonged to the dead connection, whose `dropClient`
1117 /// already reaped the server side.
1118 fn dropLocals(locals: []?AgentLocal) void {
1119 for (locals, 0..) |c, s| if (c) |ch| {
1120 locals[s] = null;
1121 std.posix.close(ch.fd);
1122 };
1123 }
1124
1125 /// The transport died, or the dial has to be redone: rebuild it on the CLI's
1126 /// backoff and re-attach quoting what this tile holds. False means the pump
1127 /// is finished — the wall quit, or the tile was forgotten while retrying.
1128 ///
1129 /// One function for what were four copies of five steps, which had begun to
1130 /// differ: only some dropped a scroll view a resync was about to invalidate.
1131 fn redial(
1132 t: *Tile,
1133 alloc: std.mem.Allocator,
1134 core: *interact.Core,
1135 transport: *client.Transport,
1136 target: client.Target,
1137 state: *State,
1138 /// This tile's agent channels, which the dying connection owned. Dropped
1139 /// HERE, and here only, for the reason this function exists at all:
1140 /// every call site that had to remember would be one more chance to
1141 /// strand a channel on a connection that cannot close it.
1142 agents: []?AgentLocal,
1143 ) bool {
1144 // A close that follows our own detach is the daemon saying goodbye back,
1145 // not a tear to heal: the pump wrote the .detach frame and set
1146 // `detach_ack` before the save's file I/O window let readFrame see the
1147 // daemon's side. Redialing here would re-attach a slot the user just
1148 // released.
1149 if (t.detach_ack.load(.acquire)) return false;
1150 // Before the cold-dial refusal below, which returns without reconnecting
1151 // — a pump that ends still owes these fds.
1152 dropLocals(agents);
1153 // The plain client's rule, kept for the tile a `mux TARGET` is: a
1154 // transport that died before a single frame of state carried no
1155 // session, so there is nothing to resume and retrying a bad host or a
1156 // typo'd `--via` only makes an unkillable client. `session_epoch` is
1157 // the right signal because it is set from the first snapshot and never
1158 // reset. Wall tiles do the opposite deliberately — they retry forever,
1159 // because a wall is a thing you leave up while a box reboots.
1160 if (!t.retry_cold and core.rep.session_epoch == 0) {
1161 endWith(t, .lost, 1);
1162 return false;
1163 }
1164 transport.close();
1165 state.* = .reconnecting;
1166 paintLabel(t, state.*);
1167 // ...and the same news for a one-tile wall, which has no label
1168 // bar on screen to read it off. The corner banner is the plain client's
1169 // own, said before the dial rather than inside it for its reason: it is
1170 // a PAINT on the session's screen, and the Core is what paints. `banner`
1171 // is gated on the sink, so a stripe's re-dial writes nothing here.
1172 core.banner("[reconnecting]");
1173 // The resync's own paint is what will arrive, so a history page held
1174 // here would be silently replaced a moment later.
1175 core.dropScrollView();
1176 transport.* = dial(alloc, t, target) orelse return false;
1177 // Clears `state_since_attach` (so the next exit_status is read as a
1178 // refusal again) and drops speculation made against a connection that
1179 // no longer exists — the Core's own highlight with it.
1180 core.reattached();
1181 const have = core.rep.attachArgs();
1182 sendAttach(t, transport, have.have_seq, have.have_epoch) catch return false;
1183 return true;
1184 }
1185
1186 /// One pump pass's geometry, plus the `.resize` that pass owes the daemon.
1187 const Pass = struct {
1188 top: u16,
1189 left: u16,
1190 rows: u16,
1191 cols: u16,
1192 label_rows: u16,
1193 term_cols: u16,
1194 term_rows: u16,
1195 // A relayout re-cut this tile: the pump owes the daemon THIS pass's
1196 // content size.
1197 resize: bool,
1198 };
1199
1200 fn takePass(t: *Tile) Pass {
1201 t.shared.paint_mu.lock();
1202 defer t.shared.paint_mu.unlock();
1203 // The flag comes out of the SAME hold as the rect it describes. Read a
1204 // pass apart from its doorbell and a relayout landing between the two
1205 // is swallowed: the pump sends the rect it snapshotted first, clears
1206 // the flag, and the daemon keeps a grid the tile has already stopped
1207 // painting at — nothing re-sends, because the claim path does not.
1208 const owed = t.resize_pending;
1209 t.resize_pending = false;
1210 return .{
1211 .top = t.rect.top,
1212 .left = t.rect.left,
1213 .rows = t.rect.rows,
1214 .cols = t.rect.cols,
1215 .label_rows = t.shared.label_rows,
1216 .term_cols = t.shared.size.cols,
1217 .term_rows = t.shared.size.rows,
1218 .resize = owed,
1219 };
1220 }
1221
1222 /// Absolute rows count from the oldest row the daemon keeps, and a resync
1223 /// renames that space: a kept highlight inverts rows nobody selected.
1224 /// One tile's life: dial → attach → replay frames into its Core → repaint
1225 /// at its rect. Runs on its own thread (see module header). On transport
1226 /// death: reconnect on the CLI's backoff schedule, quoting
1227 /// have_seq/have_epoch, and the snapshot-vs-delta resolution does the rest.
1228 /// Ends when `running` clears, the session exits, or the attach is refused.
1229 ///
1230 /// This thread is also the tile's WRITER: every frame the keyboard doorbells
1231 /// for — a resize, a detach, a focus claim or release — goes out from here,
1232 /// because a Transport has exactly one owning thread (module header).
1233 fn pumpTile(t: *Tile) void {
1234 // FIRST defer, so it runs LAST: every `return` below — a refused
1235 // attach, an exited session, a dial the quit interrupted, a Core that
1236 // would not initialise — is this tile going quiet for good, and the
1237 // keyboard needs to know which tiles it has to paint for. Declared
1238 // before the allocator's own defer so nothing can end this thread
1239 // without it running.
1240 // The bell goes with the store and after it, `endWith`'s reason: the
1241 // keyboard's test is `!alive`, so a ring that precedes the store is a
1242 // wake-up that finds nothing.
1243 defer {
1244 t.alive.store(false, .release);
1245 ringKeyboard(t.shared);
1246 // LAST, after the bell above has finished reading `t.shared`: this
1247 // is what hands the slot to `birthTile`, and nothing may touch the
1248 // tile after it.
1249 t.pump_done.store(true, .release);
1250 }
1251
1252 // Per-thread allocator: nothing allocated here crosses threads except
1253 // painted bytes, which go out under the paint mutex.
1254 var gpa: std.heap.DebugAllocator(.{}) = .init;
1255 defer _ = gpa.deinit();
1256 const alloc = gpa.allocator();
1257
1258 // Whether this tile may narrate and may start a daemon travels IN its
1259 // target, set once by whoever made the tile: a picker Enter is an ask,
1260 // a poll's list is not. This is the pump's own copy — `t.r.target` is
1261 // read by the KEYBOARD thread under `paint_mu` for chord births and is
1262 // never written from here — and it is spent below, once.
1263 var target = t.r.target;
1264
1265 // ONE Core per tile, from birth. It owns this tile's replica, its
1266 // prediction overlay and its drag for the tile's whole life: the tile
1267 // paints from that replica at its own offset, and what decides whether
1268 // a keystroke may be speculated at all is the pty's line discipline,
1269 // which arrives in `.pty_mode` frames long before the tile is focused.
1270 //
1271 // `in_fd` is the wall's stdin and this Core never reads it — the
1272 // keyboard thread does, on the far side of the mailbox. It is passed
1273 // because it is the truth about whether there is a terminal here at all
1274 // (`is_tty`), which the mouse split and the side channels are gated on.
1275 var core = interact.Core.initSized(
1276 alloc,
1277 std.posix.STDIN_FILENO,
1278 t.shared.out_fd,
1279 t.shared.size,
1280 ) catch return;
1281 defer core.deinit();
1282 // Whether this tile has EVER held the terminal, which is the only
1283 // question the publish below is gated on: a tile that was never focused
1284 // has nothing to say about prediction and must not overwrite what the
1285 // tile that was does.
1286 var ever_focused = false;
1287 // Focus, as this pump knows it. Not `core.claim`: a claim needs a
1288 // terminal, and `mux` on a pipe has none, yet its one tile is focused
1289 // and its predictions still expire and still get counted.
1290 var focused = false;
1291 // The last word on this tile's prediction, whichever way the pump ends
1292 // — an exit_status arrives and RETURNS, so the per-pass publish inside
1293 // the loop is always one pass stale by then.
1294 defer if (ever_focused) publishStats(t.shared, core.overlay.counters);
1295 // Where this Core's paints land: this tile's rect. The sink admits a
1296 // paint whenever the tile has not been forgotten; see `tilePaintBegin`.
1297 core.sink = .{ .ctx = t, .begin = tilePaintBegin, .end = tilePaintEnd };
1298 // The paint-end hook reads the core's screen cursor; only this pump
1299 // thread dereferences it, and the core outlives the pump.
1300 t.core = &core;
1301 // This thread does not own the exit and cannot print the stats line —
1302 // see `Shared.stats`.
1303 core.owns_stats = false;
1304
1305 paintLabel(t, .connecting);
1306 // The ENTRY tile arrives with its link already up — dialled on the main
1307 // thread, where the tty was, so ssh could prompt. `adopt` is what moves
1308 // its QUIC out-queue onto this thread's allocator; see there.
1309 var transport = if (t.pre) |pre| blk: {
1310 var tr = pre;
1311 t.pre = null;
1312 tr.adopt(alloc);
1313 break :blk tr;
1314 } else dial(alloc, t, target) orelse return;
1315 defer transport.close();
1316 // The ask is SPENT, on whichever of the two branches above got the
1317 // link: the entry tile's dial happened on the main thread, a picker
1318 // birth's just happened here. Every `redial` below is handed this
1319 // copy, so a reconnect can neither start a daemon — `muxd stop` typed
1320 // on that box would otherwise be undone by the next backoff, the
1321 // poll's bug moved onto a tile — nor print the fallback line onto the
1322 // alternate screen the tiles are painted on.
1323 if (target == .hand) target.hand.asked = false;
1324 // The entry tile's attach carries its rect, so the session is sized to
1325 // the terminal the tile claims and no second resize follows —
1326 // re-asserting a size the daemon just heard costs one more snapshot on
1327 // every `mux`, which is exactly the round trip the convergence must not
1328 // add. A view tile attaches at 0x0 (join-only) and `sendAttach` doorbells
1329 // its rect behind the attach; that path is inside `sendAttach`.
1330 sendAttach(t, &transport, 0, 0) catch {
1331 endWith(t, .lost, 1);
1332 return;
1333 };
1334 // One question at a time, with the deadline that makes a daemon too old
1335 // to have heard it (`sessions_req` is 0x0c) say so instead of swallowing
1336 // every chord for the rest of the session. `client.PendingSwitch`
1337 // verbatim — the client asked the same question and this is the same
1338 // answer, moved to the thread that owns the link.
1339 var pending: client.PendingSwitch = .{};
1340 // The ssh-agent channels this tile is serving, one open fd each to this
1341 // machine's agent. Everything about them is thread-local: this table,
1342 // the fds in it, and the frames that move them all live on this pump.
1343 var agent_locals: [proto.agent_chans_max]?AgentLocal = @splat(null);
1344 // Every `return` below is this tile going quiet with channels possibly
1345 // still open, and the daemon's side of them dies with the transport the
1346 // defer above closes.
1347 defer dropLocals(&agent_locals);
1348
1349 var state: State = .connecting;
1350 // What this tile's paint is worth: while it matches the wall's
1351 // generation the terminal still holds what this thread drew.
1352 var painted_gen = t.shared.repaint_gen.load(.acquire);
1353 // `gone` ends this thread exactly as `running` does — the defers close
1354 // the transport, which frees the daemon slot and NOTHING else. The
1355 // session goes on running: "remove is detach".
1356 outer: while (t.shared.running.load(.acquire) and !t.gone.load(.acquire)) {
1357 // The link and the doorbell, then one fd per live agent channel —
1358 // joined into the pump's own poll rather than given a thread each,
1359 // because a Transport has exactly one owning thread and these bytes
1360 // leave through it. `at` remembers which table slot each of those
1361 // trailing fds came from, so a readable one can be traced back to
1362 // its channel without a second search.
1363 var fdbuf: [2 + agent_locals.len]std.posix.pollfd = undefined;
1364 fdbuf[0] = .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
1365 fdbuf[1] = .{ .fd = t.wake_r, .events = std.posix.POLL.IN, .revents = 0 };
1366 var nfds: usize = 2;
1367 var at: [agent_locals.len]usize = undefined;
1368 for (agent_locals, 0..) |c, s| if (c) |ch| {
1369 at[nfds - 2] = s;
1370 fdbuf[nfds] = .{ .fd = ch.fd, .events = std.posix.POLL.IN, .revents = 0 };
1371 nfds += 1;
1372 };
1373 _ = std.posix.poll(fdbuf[0..nfds], transport.timeoutMs(100)) catch return;
1374 transport.service();
1375 if (fdbuf[1].revents != 0) drainWake(t);
1376
1377 // The paint offset for this pass: a relayout may have re-cut this
1378 // tile's rect, and every paint the pass drives through the Core
1379 // has to land in the rect the tile currently owns. Snapshot under
1380 // `paint_mu` — the keyboard writes `rect` and `label_rows` under
1381 // it — and use the snapshot for the whole pass.
1382 const snap = takePass(t);
1383 core.row_off = snap.top + snap.label_rows;
1384 core.col_off = snap.left;
1385 core.owns_screen = snap.top == 0 and snap.left == 0 and
1386 snap.label_rows == 0 and snap.cols == snap.term_cols and
1387 snap.rows == snap.term_rows;
1388 const snap_view_rows: u16 = snap.rows -| snap.label_rows;
1389 const snap_view_cols: u16 = snap.cols;
1390
1391 // FOCUS CLAIM. The keyboard moved the focus onto this tile; the
1392 // session's mouse modes and side channels go on here, on the thread
1393 // that owns the transport and the Core. The resize the claim used
1394 // to send is gone — the attach already carried the rect, and a
1395 // relayout doorbells `resize_pending` for any later change.
1396 if (t.claim_pending.swap(false, .acq_rel)) {
1397 // A refused claim is re-armed rather than lost: the host
1398 // picker's popup admits no paint, and a claim writes the
1399 // session's modes through the paint sink, so a claim dropped
1400 // under it leaves this pump focused holding no terminal — no
1401 // mouse modes, no side channels, the notice below eaten — until
1402 // the user moves the focus away and back. The picker's close
1403 // rings this pump, so the retry is a keystroke away.
1404 //
1405 // The rest of the PASS still runs whatever comes back:
1406 // `takePass` has already cleared `resize_pending`, so skipping
1407 // out here would drop a relayout this tile owes the daemon.
1408 const step = claimFocus(t, &core, .{ .cols = snap_view_cols, .rows = snap_view_rows });
1409 // A stale arm — the keyboard moved the focus between arming
1410 // this and this pass — takes nothing: no claim, no modes, no
1411 // notice, and these predictions are not the focus's to publish
1412 // either. The tile that DOES hold the focus was armed by the
1413 // same `setFocus` that took it from this one.
1414 focused = step != .dropped;
1415 ever_focused = ever_focused or focused;
1416 if (step == .done) {
1417 // A sentence the keyboard left for whoever owns the terminal
1418 // next — a refused `Ctrl-\ c`, so far. Painted here because
1419 // a banner belongs to a Core and this is the Core that has
1420 // just taken the screen; painted AFTER the repaint below
1421 // would be wrong, so it is taken now and shown once the grid
1422 // is up.
1423 var notice_buf: [96]u8 = undefined;
1424 const notice = takeNotice(t.shared, &notice_buf);
1425 // The replica has been hot the whole time, so a claim paints
1426 // from it NOW rather than waiting for the daemon's answering
1427 // snapshot. That is the headline: moving the focus costs a
1428 // local repaint, never a wire frame.
1429 // ...but only when there IS one. A tile focused before its
1430 // first snapshot — the entry tile, on every `mux` — would
1431 // otherwise paint a blank grid over the terminal before the
1432 // session has said anything, which is a screen the plain
1433 // client never drew and bytes a capture never held.
1434 if (core.rep.session_epoch != 0) core.repaint() catch {};
1435 if (notice.len > 0) core.banner(notice);
1436 }
1437 }
1438 // FOCUS RELEASE. The keyboard moved the focus off this tile and
1439 // wrote the session's release itself, under `paint_mu`, before
1440 // doorbelling — so the handover is ordered and this pump owes only
1441 // its own state. `.already_written` is that discipline.
1442 if (t.release_pending.swap(false, .acq_rel)) {
1443 focused = false;
1444 core.releaseTerminal(.already_written);
1445 // The speculation described a screen this terminal no longer
1446 // shows.
1447 core.overlay.flush();
1448 }
1449
1450 // RELAYOUT DOORBELL: this tile's rect changed. The pump is the
1451 // transport's only writer, so relayout sets the flag and the pump
1452 // sends the `.resize` from here.
1453 if (snap.resize) {
1454 core.overlay.setResizePending(true);
1455 // The Core clips every paint to its size; a resize the daemon
1456 // hears but the Core does not leaves the bottom of the new
1457 // grid cut off on screen forever.
1458 core.adoptSize(.{ .cols = snap_view_cols, .rows = snap_view_rows });
1459 transport.writeFrame(
1460 .resize,
1461 &proto.encodeSize(snap_view_cols, snap_view_rows),
1462 ) catch {
1463 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1464 continue :outer;
1465 };
1466 }
1467
1468 // `Ctrl-\ d`: hand the daemon its slot back before the process
1469 // dies, rather than leaving it for the socket's death to be
1470 // noticed. Only this thread may write the frame, so the keyboard
1471 // asked and is waiting for the ack.
1472 if (t.detach_req.swap(false, .acq_rel)) {
1473 transport.writeFrame(.detach, "") catch {};
1474 t.detach_ack.store(true, .release);
1475 ringKeyboard(t.shared);
1476 }
1477
1478 // A focus chord's question, put on the wire from the thread that
1479 // owns the link. The ANSWER is read out of the `.sessions_reply`
1480 // arm below and handed to the keyboard, which is the only thread
1481 // that may move the focus or grow the wall.
1482 const asked: client.SwitchIntent = @enumFromInt(t.ask.swap(0, .acq_rel));
1483 if (asked != .none) {
1484 // The deadline starts HERE, when the question goes on the wire,
1485 // not when the key was typed: it exists to catch a daemon too
1486 // old to have heard `sessions_req`, and the time a chord spent
1487 // in the mailbox is not that daemon's silence.
1488 pending.arm(asked, std.time.milliTimestamp());
1489 var end_buf: [proto.end_req_max_len]u8 = undefined;
1490 const sent = switch (asked) {
1491 // The FORCE is the second press, not a second frame: the
1492 // daemon refused the first and this says the user meant it.
1493 .end, .end_force => transport.writeFrame(.end_req, proto.encodeEndReq(
1494 &end_buf,
1495 asked == .end_force,
1496 proto.wireName(t.r.session),
1497 )),
1498 else => transport.writeFrame(.sessions_req, ""),
1499 };
1500 sent catch {
1501 pending.clear();
1502 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1503 continue :outer;
1504 };
1505 }
1506
1507 // A chord that was never answered. Said with the banner rather than
1508 // stderr for the client's reason: the terminal is in raw mode on
1509 // the alternate screen and a print there lands mid-grid. Reachable
1510 // at all because the poll below is capped at 100ms.
1511 //
1512 // Read before `expired` spends it: the sentence names the verb the
1513 // daemon did not know, and "no session list" over an `x` sends the
1514 // user looking at the wrong end of the wire.
1515 const waiting = pending.intent;
1516 if (pending.expired(std.time.milliTimestamp()))
1517 core.banner(switch (waiting) {
1518 .end, .end_force => "[daemon too old to end a session]",
1519 else => "[no session list: upgrade muxd]",
1520 });
1521
1522 // FRAMES BEFORE KEYS, and that order is load-bearing rather than
1523 // arbitrary. What the session has already said must be known before
1524 // what the user is saying is interpreted, because the interpreting
1525 // depends on it: `Core.forward` splits a wheel notch by whether the
1526 // session's application asked for the mouse, and that fact arrives
1527 // in a `term_modes` frame. A pass that holds both a readable frame
1528 // and a mailbox of keystrokes and drains the mailbox first judges
1529 // the keystrokes against a session it has not finished listening
1530 // to — the plain client's loop read frames first, and an
1531 // application holding the mouse lost its first wheel notch to this
1532 // client's scrollback the moment the order was reversed. Found by
1533 // the suite, not by reading.
1534 //
1535 // The `.quic` disjunct is the hub's lesson verbatim: QUIC frames
1536 // can arrive from the stream layer with the socket never going
1537 // readable.
1538 if (fdbuf[0].revents != 0 or transport.link == .quic) frames: {
1539 while (true) {
1540 const incoming = transport.readFrame(alloc) catch return;
1541 const frame = switch (incoming) {
1542 .frame => |f| f,
1543 .incomplete => break :frames,
1544 .closed => {
1545 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1546 continue :outer;
1547 },
1548 };
1549 defer frame.deinit(alloc);
1550 // Everything the frame means to the replica and the screen,
1551 // in the one place that switch lives. The Core paints at
1552 // this tile's offset through the sink, and writes side
1553 // channels only while it holds the claim — so a tile that
1554 // is not focused still keeps its replica hot and its grid
1555 // on its rect, but its title and mouse modes reach no
1556 // terminal until the focus comes to it.
1557 //
1558 // The one exception is deliberate and is `.pty_mode`: the
1559 // Core feeds the overlay's mode whatever the claim, because
1560 // the pty's line discipline is the entire gate on
1561 // speculation — a password prompt must never be predicted —
1562 // and a claim has to start from the truth rather than
1563 // from `.never` and a round trip.
1564 const routed = core.frame(frame.type, frame.payload) catch break :frames;
1565 switch (routed) {
1566 .skip, .handled => {},
1567 .state => {
1568 if (state != .up) {
1569 state = .up;
1570 t.ever_up.store(true, .release);
1571 paintLabel(t, state);
1572 }
1573 },
1574 // The replica is suspect, not the transport: re-attach
1575 // quoting (0,0) explicitly — a quoted seq would invite
1576 // the delta that cannot fix us.
1577 .resync => {
1578 core.rep.state_since_attach = false;
1579 // A resync renames the absolute row space, so a
1580 // held drag now names rows nobody selected — and
1581 // on an EPOCH change `sel_range` still matches it,
1582 // so an in-flight reply would copy the new session's
1583 // text. `redial` drops it for this reason; this path
1584 // re-attaches without going through it.
1585 core.drag.clear();
1586 sendAttach(t, &transport, 0, 0) catch return;
1587 },
1588 .not_mine => switch (frame.type) {
1589 .exit_status => {
1590 // Before any replay frame this is the refusal
1591 // path; after, the session really ended.
1592 const landed = core.rep.state_since_attach;
1593 state = if (landed) .exited else .refused;
1594 paintLabel(t, state);
1595 endWith(
1596 t,
1597 if (landed) .exited else .refused,
1598 if (landed and frame.payload.len >= 1) frame.payload[0] else 1,
1599 );
1600 return;
1601 },
1602 .taken_over => {
1603 // Unsent by this daemon (wire-compat only), but
1604 // the plain client had an answer for it and the
1605 // converged one keeps it: somebody else took the
1606 // session, so this tile is finished rather than
1607 // reconnecting into a fight over the grid.
1608 state = .exited;
1609 paintLabel(t, state);
1610 endWith(t, .taken, 0);
1611 return;
1612 },
1613 .sessions_reply => {
1614 // Gated on the intent this tile's own chord
1615 // armed: only a question we asked may move the
1616 // focus, so an unasked-for reply is ignored. The
1617 // intent is SPENT here whichever name it picks
1618 // — one question, one answer.
1619 var name_buf: [proto.session_name_max]u8 = undefined;
1620 // A list is only ever asked for to NAME a new
1621 // session: `n`/`p` walk the wall's own tiles and
1622 // ask nothing, and the end verbs are answered by
1623 // `end_reply`. Anything else here is a reply to
1624 // a question this tile did not put.
1625 const pick: ?[]const u8 = switch (pending.take()) {
1626 .new => client.nextFreeName(&name_buf, frame.payload),
1627 else => null,
1628 };
1629 if (pick) |p| postAnswer(t, p);
1630 },
1631 .end_reply => {
1632 const r = proto.parseEndReply(frame.payload) orelse break :frames;
1633 _ = pending.take();
1634 onEndReply(t, r, std.time.milliTimestamp());
1635 // An ACCEPTED end says nothing: the hangup's
1636 // `exit_status` is on its way and the `.exited`
1637 // path narrates it, so a banner here would be
1638 // the client congratulating itself ahead of the
1639 // daemon. Not a `break` either — that exit
1640 // status is very likely the next frame in this
1641 // same pass, and leaving it for the next poll
1642 // is 100ms of a tile that is already gone.
1643 if (!r.accepted) {
1644 var b: [96]u8 = undefined;
1645 core.banner(if (r.others == 0)
1646 endRefusal(r.reason)
1647 else
1648 std.fmt.bufPrint(&b, "[{d} other{s} attached - x again to end]", .{
1649 r.others,
1650 if (r.others == 1) "" else "s",
1651 }) catch "[others attached - x again to end]");
1652 }
1653 },
1654 .selection_reply => copySelection(t, alloc, &core, frame.payload),
1655 .agent_open => {
1656 const id = proto.decodeAgentId(frame.payload) catch break :frames;
1657 // Every refusal is the same answer on the wire —
1658 // a channel that closes without a byte, which the
1659 // far side's ssh reads as "no agent" and gives up
1660 // on rather than hanging. Only one of the four
1661 // reasons is ordinary — no agent on this machine;
1662 // the rest are a daemon asking for something it
1663 // should not.
1664 const opened = openAgentChan(
1665 &agent_locals,
1666 id,
1667 t.r.agent,
1668 std.posix.getenv(proto.agent_sock_env) orelse "",
1669 );
1670 if (!opened)
1671 transport.writeFrame(
1672 .agent_close,
1673 &proto.encodeAgentId(id),
1674 ) catch {};
1675 },
1676 .agent_data => if (!deliverAgentData(
1677 &agent_locals,
1678 frame.payload,
1679 &transport,
1680 )) break :frames,
1681 .agent_close => {
1682 const id = proto.decodeAgentId(frame.payload) catch break :frames;
1683 // Silent, mirroring the daemon: it has already
1684 // retired this id, so an `agent_close` back would
1685 // be an echo it has to learn to ignore.
1686 if (findLocal(&agent_locals, id)) |s| {
1687 const ch = agent_locals[s].?;
1688 agent_locals[s] = null;
1689 std.posix.close(ch.fd);
1690 }
1691 },
1692 // MsgType is an open enum, so the compiler still
1693 // wants an arm for everything `.not_mine` cannot be.
1694 else => {},
1695 },
1696 }
1697 // Only the socket link guarantees one readable event is
1698 // one frame; QUIC may have buffered more.
1699 //
1700 // But "one event, one frame" is not "one pass, one frame".
1701 // A resync is a BURST — snapshot, pty mode, title, terminal
1702 // modes — and stopping after the first leaves the rest to be
1703 // read a pass at a time, with the user's keystrokes
1704 // interleaved between them. `Core.forward` decides what a
1705 // wheel notch MEANS from the session's terminal modes, so a
1706 // notch judged before the `term_modes` at the END of that
1707 // burst is stolen for this client's scrollback instead of
1708 // reaching the application that asked for the mouse. A
1709 // single-threaded client hid the hazard by being busy
1710 // painting the snapshot when the notch arrived; a keyboard
1711 // on its own thread notices it immediately, and the suite
1712 // caught it two runs out of three.
1713 //
1714 // So: drain what has already ARRIVED. A zero timeout waits
1715 // for nothing, which is what keeps this a drain and not a
1716 // second blocking read.
1717 //
1718 // It cannot starve the rest of the pass, and two measured
1719 // facts are why. The daemon coalesces its deltas, so a
1720 // loud session hands over a bounded burst and `poll(0)`
1721 // finds the socket empty within it rather than a stream
1722 // that refills as fast as it is read. And the keyboard is
1723 // a different thread with its own 400ms bound on the one
1724 // thing it ever waits for a pump to do (`awaitDetach`), so
1725 // even a pump that did stall here could not hold the
1726 // terminal hostage — which is the failure this would
1727 // otherwise have to be argued safe against.
1728 if (transport.link != .quic) {
1729 var more = [_]std.posix.pollfd{
1730 .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
1731 };
1732 const ready = std.posix.poll(&more, 0) catch 0;
1733 if (ready == 0 or more[0].revents == 0) break :frames;
1734 }
1735 }
1736 }
1737
1738 // The other direction: what this machine's agent answered, back out
1739 // as `agent_data`. AFTER the frame drain, deliberately — the table
1740 // this walks is then the one the daemon's latest word left behind,
1741 // so a channel it has just closed is already gone rather than read
1742 // once more on its way out.
1743 for (2..nfds) |i| {
1744 if (fdbuf[i].revents == 0) continue;
1745 const s = at[i - 2];
1746 const ch = agent_locals[s] orelse continue;
1747 // The id is written into the head of the very buffer the read
1748 // fills, so a frame costs no second copy. The read is capped at
1749 // the wire's `agent_data_max` because that cap is what keeps one
1750 // busy channel from holding the link against the session's own
1751 // bytes — the same bound the daemon reads its end with.
1752 var buf: [proto.agent_id_len + proto.agent_data_max]u8 = undefined;
1753 buf[0..proto.agent_id_len].* = proto.encodeAgentId(ch.id);
1754 // DONTWAIT, and it is load-bearing rather than belt-and-braces:
1755 // the poll above happened before the frame drain, so the daemon
1756 // may since have closed this slot's channel and an `agent_open`
1757 // in the same burst may have refilled the slot with a fresh
1758 // connection that has said nothing yet. A blocking read there
1759 // would stop the whole tile — its keystrokes, its paints — on a
1760 // socket that is merely idle. `agent_locals[s]` is re-read above,
1761 // so whatever this does return is attributed to the id that
1762 // actually owns the fd.
1763 //
1764 // EOF and a broken connection are one case: the agent is done
1765 // with this channel either way, and the far side needs to hear so.
1766 const n = std.posix.recv(
1767 ch.fd,
1768 buf[proto.agent_id_len..],
1769 std.posix.MSG.DONTWAIT,
1770 ) catch |err| switch (err) {
1771 error.WouldBlock => continue,
1772 else => 0,
1773 };
1774 if (n == 0) {
1775 closeLocal(&agent_locals, s, &transport);
1776 continue;
1777 }
1778 transport.writeFrame(.agent_data, buf[0 .. proto.agent_id_len + n]) catch {
1779 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1780 continue :outer;
1781 };
1782 }
1783
1784 // Whatever the keyboard left, through everything a plain client's
1785 // keystrokes go through: the mouse split, the wheel, alternate
1786 // scroll, the scrollback view, the prediction and the input frame.
1787 // Only the focused tile's mailbox is ever written, but every pump
1788 // drains its own — an empty mailbox is the common, cheap answer.
1789 var keys_buf: [mailbox_max]u8 = undefined;
1790 const keys = takeKeys(t, &keys_buf);
1791 if (keys.len > 0) {
1792 // A paint that would not allocate is not a dead link: the
1793 // replica is untouched and the next frame redraws from it.
1794 const step = core.forward(&transport, keys) catch interact.Step.ok;
1795 if (step == .lost) {
1796 if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
1797 continue :outer;
1798 }
1799 // Input is moving again, so "input dropped" has stopped being
1800 // news. Repainted rather than merely cleared, because the bar
1801 // is still carrying the old sentence until something draws
1802 // over it — and only when there is a bar, which `paintLabel`
1803 // already decides.
1804 if (t.in_dropped.swap(false, .acq_rel)) paintLabel(t, state);
1805 }
1806
1807 // A prediction the daemon never answered must not sit on the
1808 // screen forever, and only the clock can say so — no frame will.
1809 // Only while focused: an unfocused tile's rect shows no prediction
1810 // to retire.
1811 if (focused) {
1812 core.idle() catch {};
1813 publishStats(t.shared, core.overlay.counters);
1814 }
1815
1816 // A relayout re-cut the stripes (or cleared the screen for an empty
1817 // wall), and a quiet session sends nothing to trigger a repaint.
1818 // The replica is current — the SCREEN is not — so the generation is
1819 // what puts it back. Checked on the poll timeout, so a tile comes
1820 // back within ~100ms of a relayout whether or not its session ever
1821 // speaks again. The drag clears here too: a relayout moved the rect
1822 // a held drag's anchor was resolved against.
1823 const gen = t.shared.repaint_gen.load(.acquire);
1824 if (gen != painted_gen) {
1825 core.drag.clear();
1826 paintLabel(t, state);
1827 // A 0-row rect (fullscreened out) paints nothing: the daemon
1828 // refused the 0×0 resize, so the session keeps its real grid.
1829 if (snap_view_rows > 0) core.repaint() catch {};
1830 painted_gen = gen;
1831 }
1832 }
1833 }
1834
1835 /// Move the focus to tile `next`. Client-local: decides which pump owns the 790 /// Move the focus to tile `next`. Client-local: decides which pump owns the
1836 /// terminal's modes, not which session the daemon hears. The outgoing tile's 791 /// terminal's modes, not which session the daemon hears. The outgoing tile's
1837 /// `session_release` is written HERE, under `paint_mu`, so the handover is 792 /// `session_release` is written HERE, under `paint_mu`, so the handover is
@@ -2267,7 +1222,7 @@ fn initTile(t: *Tile, r: Resolved, s: layout.Rect, shared: *Shared, idx: usize,
2267 } 1222 }
2268 1223
2269 pub fn spawnPump(t: *Tile) void { 1224 pub fn spawnPump(t: *Tile) void {
2270 const th = std.Thread.spawn(.{}, pumpTile, .{t}) catch { 1225 const th = std.Thread.spawn(.{}, wall_pump.pumpTile, .{t}) catch {
2271 // A tile with no thread is a tile nothing will ever paint — the 1226 // A tile with no thread is a tile nothing will ever paint — the
2272 // same hole `pumpTile`'s exit closes, reached without the pump 1227 // same hole `pumpTile`'s exit closes, reached without the pump
2273 // having run at all. Marked here so the keyboard paints its bar and 1228 // having run at all. Marked here so the keyboard paints its bar and
@@ -2827,17 +1782,6 @@ pub fn selfSession(target: client.Target, env_sock: ?[]const u8, env_session: ?[
2827 return proto.resolveName(es); 1782 return proto.resolveName(es);
2828 } 1783 }
2829 1784
2830 /// A refusal in THIS client's words. `parseEndReply` hands back the frame's
2831 /// tail unfiltered and `paintBanner` writes it verbatim, so a peer's
2832 /// `\x1b]0;..\x07` would run outside the replica. The reasons are constants.
2833 fn endRefusal(reason: []const u8) []const u8 {
2834 if (std.mem.eql(u8, reason, proto.end_reason.no_session))
2835 return "[no such session on that daemon]";
2836 if (std.mem.eql(u8, reason, proto.end_reason.bad_frame))
2837 return "[the daemon could not read the end request]";
2838 return "[the daemon refused to end this session]";
2839 }
2840
2841 /// For a caller already holding `paint_mu`. 1785 /// For a caller already holding `paint_mu`.
2842 fn takeNoticeLocked(shared: *Shared, out: []u8) []const u8 { 1786 fn takeNoticeLocked(shared: *Shared, out: []u8) []const u8 {
2843 const said = wall_picker.peekNoticeLocked(shared, out); 1787 const said = wall_picker.peekNoticeLocked(shared, out);
@@ -4370,7 +3314,7 @@ test "claimAllowed: a stale arm does not claim even when the sink would now admi
4370 // in this file that puts the two in that order. 3314 // in this file that puts the two in that order.
4371 const rect: proto.Size = .{ .cols = 80, .rows = 24 }; 3315 const rect: proto.Size = .{ .cols = 80, .rows = 24 };
4372 _ = t.claim_pending.swap(false, .acq_rel); 3316 _ = t.claim_pending.swap(false, .acq_rel);
4373 try std.testing.expectEqual(ClaimStep.dropped, claimFocus(&t, &core, rect)); 3317 try std.testing.expectEqual(ClaimStep.dropped, wall_pump.claimFocus(&t, &core, rect));
4374 3318
4375 // Nothing claimed, and nothing on the terminal: `session_claim` is the 3319 // Nothing claimed, and nothing on the terminal: `session_claim` is the
4376 // mouse modes, and a second tile setting them is what leaves one 3320 // mouse modes, and a second tile setting them is what leaves one
@@ -4382,7 +3326,7 @@ test "claimAllowed: a stale arm does not claim even when the sink would now admi
4382 // The control: the same call on the tile that DOES hold the focus takes 3326 // The control: the same call on the tile that DOES hold the focus takes
4383 // it, so the guard is a focus test and not a blanket refusal. 3327 // it, so the guard is a focus test and not a blanket refusal.
4384 shared.sel = 2; 3328 shared.sel = 2;
4385 try std.testing.expectEqual(ClaimStep.done, claimFocus(&t, &core, rect)); 3329 try std.testing.expectEqual(ClaimStep.done, wall_pump.claimFocus(&t, &core, rect));
4386 try std.testing.expectEqual(interact.Claim.session, core.claim); 3330 try std.testing.expectEqual(interact.Claim.session, core.claim);
4387 try std.testing.expect((std.posix.read(pipe[0], &buf) catch 0) > 0); 3331 try std.testing.expect((std.posix.read(pipe[0], &buf) catch 0) > 0);
4388 } 3332 }
@@ -4393,7 +3337,7 @@ test "afterClaim: a claim the picker refused is armed again for this tile" {
4393 // against a hard-coded 0 would pass here and nowhere else. 3337 // against a hard-coded 0 would pass here and nowhere else.
4394 shared.sel = 2; 3338 shared.sel = 2;
4395 var t = claimBench(&shared, 2); 3339 var t = claimBench(&shared, 2);
4396 try std.testing.expectEqual(ClaimStep.rearmed, afterClaim(&t, false, false, true)); 3340 try std.testing.expectEqual(ClaimStep.rearmed, wall_pump.afterClaim(&t, false, false, true));
4397 try std.testing.expect(t.claim_pending.load(.acquire)); 3341 try std.testing.expect(t.claim_pending.load(.acquire));
4398 } 3342 }
4399 3343
@@ -4406,7 +3350,7 @@ test "afterClaim: an arm does not outlive the focus that earned it" {
4406 // resting on the loser, and the release already consumed so nothing 3350 // resting on the loser, and the release already consumed so nothing
4407 // takes either off. 3351 // takes either off.
4408 shared.sel = 1; 3352 shared.sel = 1;
4409 try std.testing.expectEqual(ClaimStep.dropped, afterClaim(&t, false, false, true)); 3353 try std.testing.expectEqual(ClaimStep.dropped, wall_pump.afterClaim(&t, false, false, true));
4410 try std.testing.expect(!t.claim_pending.load(.acquire)); 3354 try std.testing.expect(!t.claim_pending.load(.acquire));
4411 } 3355 }
4412 3356
@@ -4415,12 +3359,12 @@ test "afterClaim: only the SINK's refusal is worth another pass" {
4415 shared.sel = 2; 3359 shared.sel = 2;
4416 var t = claimBench(&shared, 2); 3360 var t = claimBench(&shared, 2);
4417 // A claim that landed owes nothing. 3361 // A claim that landed owes nothing.
4418 try std.testing.expectEqual(ClaimStep.done, afterClaim(&t, true, true, true)); 3362 try std.testing.expectEqual(ClaimStep.done, wall_pump.afterClaim(&t, true, true, true));
4419 // A Core that ALREADY holds it answers false too, and re-arming that 3363 // A Core that ALREADY holds it answers false too, and re-arming that
4420 // would re-take the focus notice on every pass, forever. 3364 // would re-take the focus notice on every pass, forever.
4421 try std.testing.expectEqual(ClaimStep.done, afterClaim(&t, false, true, true)); 3365 try std.testing.expectEqual(ClaimStep.done, wall_pump.afterClaim(&t, false, true, true));
4422 // No terminal, nothing to hold: a piped `mux` is one tile and no claim. 3366 // No terminal, nothing to hold: a piped `mux` is one tile and no claim.
4423 try std.testing.expectEqual(ClaimStep.done, afterClaim(&t, false, false, false)); 3367 try std.testing.expectEqual(ClaimStep.done, wall_pump.afterClaim(&t, false, false, false));
4424 // None of the three armed anything. 3368 // None of the three armed anything.
4425 try std.testing.expect(!t.claim_pending.load(.acquire)); 3369 try std.testing.expect(!t.claim_pending.load(.acquire));
4426 } 3370 }
@@ -4439,10 +3383,10 @@ test "tilePaintBegin: no tile paints while the picker owns the screen" {
4439 // CLAIM fail under it (`Core.claimTerminal` writes through the same 3383 // CLAIM fail under it (`Core.claimTerminal` writes through the same
4440 // sink) — which is why the pump re-arms `claim_pending` instead of 3384 // sink) — which is why the pump re-arms `claim_pending` instead of
4441 // treating a false return as "already claimed". 3385 // treating a false return as "already claimed".
4442 try std.testing.expect(tilePaintBegin(&t)); 3386 try std.testing.expect(wall_pump.tilePaintBegin(&t));
4443 tilePaintEnd(&t); 3387 wall_pump.tilePaintEnd(&t);
4444 shared.picker_open.store(true, .release); 3388 shared.picker_open.store(true, .release);
4445 try std.testing.expect(!tilePaintBegin(&t)); 3389 try std.testing.expect(!wall_pump.tilePaintBegin(&t));
4446 // Refused WITHOUT holding the lock: a begin that returned false and 3390 // Refused WITHOUT holding the lock: a begin that returned false and
4447 // kept `paint_mu` would wedge the keyboard on its next paint. 3391 // kept `paint_mu` would wedge the keyboard on its next paint.
4448 try std.testing.expect(shared.paint_mu.tryLock()); 3392 try std.testing.expect(shared.paint_mu.tryLock());
@@ -5018,23 +3962,23 @@ test "endKey: x on a tile that never came up closes the TILE — there is no ses
5018 test "endRefusal: a refusal is said in this client's words, never in the peer's bytes" { 3962 test "endRefusal: a refusal is said in this client's words, never in the peer's bytes" {
5019 try std.testing.expectEqualStrings( 3963 try std.testing.expectEqualStrings(
5020 "[no such session on that daemon]", 3964 "[no such session on that daemon]",
5021 endRefusal(proto.end_reason.no_session), 3965 wall_pump.endRefusal(proto.end_reason.no_session),
5022 ); 3966 );
5023 try std.testing.expectEqualStrings( 3967 try std.testing.expectEqualStrings(
5024 "[the daemon could not read the end request]", 3968 "[the daemon could not read the end request]",
5025 endRefusal(proto.end_reason.bad_frame), 3969 wall_pump.endRefusal(proto.end_reason.bad_frame),
5026 ); 3970 );
5027 // Anything the daemon does not say: an OSC title set, a CSI, or 260 3971 // Anything the daemon does not say: an OSC title set, a CSI, or 260
5028 // bytes of nothing. All three used to reach the terminal verbatim. 3972 // bytes of nothing. All three used to reach the terminal verbatim.
5029 try std.testing.expectEqualStrings( 3973 try std.testing.expectEqualStrings(
5030 "[the daemon refused to end this session]", 3974 "[the daemon refused to end this session]",
5031 endRefusal("\x1b]0;pwned\x07"), 3975 wall_pump.endRefusal("\x1b]0;pwned\x07"),
5032 ); 3976 );
5033 try std.testing.expectEqualStrings( 3977 try std.testing.expectEqualStrings(
5034 "[the daemon refused to end this session]", 3978 "[the daemon refused to end this session]",
5035 endRefusal("\x1b[2J"), 3979 wall_pump.endRefusal("\x1b[2J"),
5036 ); 3980 );
5037 try std.testing.expectEqualStrings("[the daemon refused to end this session]", endRefusal("")); 3981 try std.testing.expectEqualStrings("[the daemon refused to end this session]", wall_pump.endRefusal(""));
5038 } 3982 }
5039 3983
5040 test "pickerRepaint: a screen cleared under the spelling editor still owes a paint, carrying the half-typed line" { 3984 test "pickerRepaint: a screen cleared under the spelling editor still owes a paint, carrying the half-typed line" {
@@ -5604,7 +4548,7 @@ test "sendKeys: a chunk that does not fit is dropped whole, and says so" {
5604 // the next chunk cannot splice onto a half-delivered one. 4548 // the next chunk cannot splice onto a half-delivered one.
5605 sendKeys(&t, "def"); 4549 sendKeys(&t, "def");
5606 var out: [mailbox_max]u8 = undefined; 4550 var out: [mailbox_max]u8 = undefined;
5607 try std.testing.expectEqualStrings("abcdef", takeKeys(&t, &out)); 4551 try std.testing.expectEqualStrings("abcdef", wall_pump.takeKeys(&t, &out));
5608 4552
5609 // Exactly filling it is not overflow — the bound is `>`, not `>=`. 4553 // Exactly filling it is not overflow — the bound is `>`, not `>=`.
5610 const exact = [_]u8{'q'} ** mailbox_max; 4554 const exact = [_]u8{'q'} ** mailbox_max;
@@ -6002,7 +4946,7 @@ const ResizeWitness = struct {
6002 4946
6003 fn run(w: *ResizeWitness) void { 4947 fn run(w: *ResizeWitness) void {
6004 while (!w.stop.load(.acquire)) { 4948 while (!w.stop.load(.acquire)) {
6005 const p = takePass(w.t); 4949 const p = wall_pump.takePass(w.t);
6006 if (p.resize) w.sent.store(pack(p.cols, p.rows -| p.label_rows), .release); 4950 if (p.resize) w.sent.store(pack(p.cols, p.rows -| p.label_rows), .release);
6007 _ = w.passes.fetchAdd(1, .release); 4951 _ = w.passes.fetchAdd(1, .release);
6008 } 4952 }
@@ -6631,7 +5575,7 @@ test "a copy too big for OSC 52 is said out loud rather than dropped" {
6631 defer alloc.free(rbuf); 5575 defer alloc.free(rbuf);
6632 const payload = replyBytes(rbuf, core.sel_id, .ok, 0, big); 5576 const payload = replyBytes(rbuf, core.sel_id, .ok, 0, big);
6633 5577
6634 copySelection(&t, alloc, &core, payload); 5578 wall_pump.copySelection(&t, alloc, &core, payload);
6635 var buf: [8192]u8 = undefined; 5579 var buf: [8192]u8 = undefined;
6636 const said = readAvail(p[0], &buf); 5580 const said = readAvail(p[0], &buf);
6637 try std.testing.expect(std.mem.indexOf(u8, said, "too large") != null); 5581 try std.testing.expect(std.mem.indexOf(u8, said, "too large") != null);
@@ -6973,15 +5917,15 @@ test "agent channels: slots fill in order, a full table refuses, and a close is
6973 5917
6974 for (0..3) |i| try std.testing.expectEqual( 5918 for (0..3) |i| try std.testing.expectEqual(
6975 @as(?usize, i), 5919 @as(?usize, i),
6976 storeLocal(&locals, @intCast(10 + i), chans[i]), 5920 wall_pump.storeLocal(&locals, @intCast(10 + i), chans[i]),
6977 ); 5921 );
6978 // The full table the pump answers with `agent_close`. The fourth fd stays 5922 // The full table the pump answers with `agent_close`. The fourth fd stays
6979 // this test's to close: a refused store never took ownership of it. 5923 // this test's to close: a refused store never took ownership of it.
6980 try std.testing.expectEqual(@as(?usize, null), storeLocal(&locals, 99, chans[3])); 5924 try std.testing.expectEqual(@as(?usize, null), wall_pump.storeLocal(&locals, 99, chans[3]));
6981 std.posix.close(chans[3]); 5925 std.posix.close(chans[3]);
6982 5926
6983 try std.testing.expectEqual(@as(?usize, 1), findLocal(&locals, 11)); 5927 try std.testing.expectEqual(@as(?usize, 1), wall_pump.findLocal(&locals, 11));
6984 try std.testing.expectEqual(@as(?usize, null), findLocal(&locals, 99)); 5928 try std.testing.expectEqual(@as(?usize, null), wall_pump.findLocal(&locals, 99));
6985 5929
6986 // A pipe for the link, which is what a `.fd` transport writes to. The 5930 // A pipe for the link, which is what a `.fd` transport writes to. The
6987 // frame that comes back out of it is the daemon's only notice that this 5931 // frame that comes back out of it is the daemon's only notice that this
@@ -6991,7 +5935,7 @@ test "agent channels: slots fill in order, a full table refuses, and a close is
6991 defer std.posix.close(link[1]); 5935 defer std.posix.close(link[1]);
6992 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 5936 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd };
6993 5937
6994 closeLocal(&locals, 1, &transport); 5938 wall_pump.closeLocal(&locals, 1, &transport);
6995 try std.testing.expectEqual(@as(?AgentLocal, null), locals[1]); 5939 try std.testing.expectEqual(@as(?AgentLocal, null), locals[1]);
6996 try std.testing.expect(peerClosed(peers[1])); 5940 try std.testing.expect(peerClosed(peers[1]));
6997 5941
@@ -7003,7 +5947,7 @@ test "agent channels: slots fill in order, a full table refuses, and a close is
7003 5947
7004 // The redial's sweep: every channel still open goes, and nothing is said 5948 // The redial's sweep: every channel still open goes, and nothing is said
7005 // — the connection that owned them is the one that just died. 5949 // — the connection that owned them is the one that just died.
7006 dropLocals(&locals); 5950 wall_pump.dropLocals(&locals);
7007 for (locals) |c| try std.testing.expectEqual(@as(?AgentLocal, null), c); 5951 for (locals) |c| try std.testing.expectEqual(@as(?AgentLocal, null), c);
7008 try std.testing.expect(peerClosed(peers[0])); 5952 try std.testing.expect(peerClosed(peers[0]));
7009 try std.testing.expect(peerClosed(peers[2])); 5953 try std.testing.expect(peerClosed(peers[2]));
@@ -7033,7 +5977,7 @@ test "agent channels: an oversize frame hangs the channel up, a full one lands"
7033 var locals: [2]?AgentLocal = @splat(null); 5977 var locals: [2]?AgentLocal = @splat(null);
7034 const chan = try std.posix.pipe2(.{ .NONBLOCK = true }); 5978 const chan = try std.posix.pipe2(.{ .NONBLOCK = true });
7035 defer std.posix.close(chan[0]); 5979 defer std.posix.close(chan[0]);
7036 _ = storeLocal(&locals, 5, chan[1]); 5980 _ = wall_pump.storeLocal(&locals, 5, chan[1]);
7037 5981
7038 const link = try std.posix.pipe2(.{ .NONBLOCK = true }); 5982 const link = try std.posix.pipe2(.{ .NONBLOCK = true });
7039 defer std.posix.close(link[0]); 5983 defer std.posix.close(link[0]);
@@ -7045,7 +5989,7 @@ test "agent channels: an oversize frame hangs the channel up, a full one lands"
7045 var small: [proto.agent_id_len + 2]u8 = undefined; 5989 var small: [proto.agent_id_len + 2]u8 = undefined;
7046 small[0..proto.agent_id_len].* = proto.encodeAgentId(5); 5990 small[0..proto.agent_id_len].* = proto.encodeAgentId(5);
7047 @memcpy(small[proto.agent_id_len..], "hi"); 5991 @memcpy(small[proto.agent_id_len..], "hi");
7048 try std.testing.expect(deliverAgentData(&locals, &small, &transport)); 5992 try std.testing.expect(wall_pump.deliverAgentData(&locals, &small, &transport));
7049 var got: [8]u8 = undefined; 5993 var got: [8]u8 = undefined;
7050 try std.testing.expectEqual(@as(usize, 2), try std.posix.read(chan[0], &got)); 5994 try std.testing.expectEqual(@as(usize, 2), try std.posix.read(chan[0], &got));
7051 try std.testing.expectEqualSlices(u8, "hi", got[0..2]); 5995 try std.testing.expectEqualSlices(u8, "hi", got[0..2]);
@@ -7056,7 +6000,7 @@ test "agent channels: an oversize frame hangs the channel up, a full one lands"
7056 defer std.testing.allocator.free(over); 6000 defer std.testing.allocator.free(over);
7057 @memset(over, 'x'); 6001 @memset(over, 'x');
7058 over[0..proto.agent_id_len].* = proto.encodeAgentId(5); 6002 over[0..proto.agent_id_len].* = proto.encodeAgentId(5);
7059 try std.testing.expect(deliverAgentData(&locals, over, &transport)); 6003 try std.testing.expect(wall_pump.deliverAgentData(&locals, over, &transport));
7060 try std.testing.expectEqual(@as(?AgentLocal, null), locals[0]); 6004 try std.testing.expectEqual(@as(?AgentLocal, null), locals[0]);
7061 try std.testing.expect(peerClosed(chan[0])); 6005 try std.testing.expect(peerClosed(chan[0]));
7062 6006
@@ -7065,7 +6009,7 @@ test "agent channels: an oversize frame hangs the channel up, a full one lands"
7065 // A payload too short to name a channel: false, and the pump abandons the 6009 // A payload too short to name a channel: false, and the pump abandons the
7066 // batch on it rather than skipping one frame, because a stream that has 6010 // batch on it rather than skipping one frame, because a stream that has
7067 // lost frame alignment is not one to keep reading. 6011 // lost frame alignment is not one to keep reading.
7068 try std.testing.expect(!deliverAgentData(&locals, "ab", &transport)); 6012 try std.testing.expect(!wall_pump.deliverAgentData(&locals, "ab", &transport));
7069 6013
7070 const frame = (try proto.readFrame(std.testing.allocator, link[0])) orelse 6014 const frame = (try proto.readFrame(std.testing.allocator, link[0])) orelse
7071 return error.NoAgentCloseOnOversize; 6015 return error.NoAgentCloseOnOversize;
@@ -7088,13 +6032,13 @@ test "agent forwarding is per tile: no -A offers nothing and opens nothing" {
7088 defer listener.deinit(); 6032 defer listener.deinit();
7089 6033
7090 var locals: [2]?AgentLocal = @splat(null); 6034 var locals: [2]?AgentLocal = @splat(null);
7091 defer dropLocals(&locals); 6035 defer wall_pump.dropLocals(&locals);
7092 6036
7093 // The gate, both ways round, against the same reachable agent. 6037 // The gate, both ways round, against the same reachable agent.
7094 try std.testing.expect(!openAgentChan(&locals, 1, false, sock)); 6038 try std.testing.expect(!wall_pump.openAgentChan(&locals, 1, false, sock));
7095 try std.testing.expectEqual(@as(?usize, null), findLocal(&locals, 1)); 6039 try std.testing.expectEqual(@as(?usize, null), wall_pump.findLocal(&locals, 1));
7096 try std.testing.expect(openAgentChan(&locals, 1, true, sock)); 6040 try std.testing.expect(wall_pump.openAgentChan(&locals, 1, true, sock));
7097 try std.testing.expect(findLocal(&locals, 1) != null); 6041 try std.testing.expect(wall_pump.findLocal(&locals, 1) != null);
7098 6042
7099 // And the daemon does not get told about an agent the tile never 6043 // And the daemon does not get told about an agent the tile never
7100 // offered — the offer is what the daemon routes on, so a stray one 6044 // offered — the offer is what the daemon routes on, so a stray one
@@ -7113,7 +6057,7 @@ test "agent forwarding is per tile: no -A offers nothing and opens nothing" {
7113 defer std.posix.close(link[1]); 6057 defer std.posix.close(link[1]);
7114 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 6058 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd };
7115 6059
7116 try sendAttach(&t, &transport, 0, 0); 6060 try wall_pump.sendAttach(&t, &transport, 0, 0);
7117 const first = (try proto.readFrame(std.testing.allocator, link[0])) orelse 6061 const first = (try proto.readFrame(std.testing.allocator, link[0])) orelse
7118 return error.NoAttach; 6062 return error.NoAttach;
7119 defer first.deinit(std.testing.allocator); 6063 defer first.deinit(std.testing.allocator);
@@ -7127,7 +6071,7 @@ test "agent forwarding is per tile: no -A offers nothing and opens nothing" {
7127 // The positive control on the same pipe: with `-A` the offer follows the 6071 // The positive control on the same pipe: with `-A` the offer follows the
7128 // attach, so the silence above is the gate and not an unwritten frame. 6072 // attach, so the silence above is the gate and not an unwritten frame.
7129 t.r.agent = true; 6073 t.r.agent = true;
7130 try sendAttach(&t, &transport, 0, 0); 6074 try wall_pump.sendAttach(&t, &transport, 0, 0);
7131 const attach = (try proto.readFrame(std.testing.allocator, link[0])).?; 6075 const attach = (try proto.readFrame(std.testing.allocator, link[0])).?;
7132 defer attach.deinit(std.testing.allocator); 6076 defer attach.deinit(std.testing.allocator);
7133 const offer = (try proto.readFrame(std.testing.allocator, link[0])) orelse 6077 const offer = (try proto.readFrame(std.testing.allocator, link[0])) orelse
@@ -7161,7 +6105,7 @@ test "a view tile's attach makes no size claim; a tile the user asked for does"
7161 // A view tile: 0x0 on the wire, and the rect owed to the doorbell. 6105 // A view tile: 0x0 on the wire, and the rect owed to the doorbell.
7162 t.creates = false; 6106 t.creates = false;
7163 t.resize_pending = false; 6107 t.resize_pending = false;
7164 try sendAttach(&t, &transport, 0, 0); 6108 try wall_pump.sendAttach(&t, &transport, 0, 0);
7165 const view = (try proto.readFrame(std.testing.allocator, link[0])) orelse 6109 const view = (try proto.readFrame(std.testing.allocator, link[0])) orelse
7166 return error.NoAttach; 6110 return error.NoAttach;
7167 defer view.deinit(std.testing.allocator); 6111 defer view.deinit(std.testing.allocator);
@@ -7175,7 +6119,7 @@ test "a view tile's attach makes no size claim; a tile the user asked for does"
7175 // and no doorbell is owed. 6119 // and no doorbell is owed.
7176 t.creates = true; 6120 t.creates = true;
7177 t.resize_pending = false; 6121 t.resize_pending = false;
7178 try sendAttach(&t, &transport, 0, 0); 6122 try wall_pump.sendAttach(&t, &transport, 0, 0);
7179 const entry = (try proto.readFrame(std.testing.allocator, link[0])) orelse 6123 const entry = (try proto.readFrame(std.testing.allocator, link[0])) orelse
7180 return error.NoAttach; 6124 return error.NoAttach;
7181 defer entry.deinit(std.testing.allocator); 6125 defer entry.deinit(std.testing.allocator);
@@ -7209,7 +6153,7 @@ test "a newborn tile's first claim is its real stripe, not the placeholder" {
7209 defer std.posix.close(link[0]); 6153 defer std.posix.close(link[0]);
7210 defer std.posix.close(link[1]); 6154 defer std.posix.close(link[1]);
7211 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd }; 6155 var transport: client.Transport = .{ .conn = .{ .r = link[0], .w = link[1] }, .link = .fd };
7212 try sendAttach(&t, &transport, 0, 0); 6156 try wall_pump.sendAttach(&t, &transport, 0, 0);
7213 const fr = (try proto.readFrame(std.testing.allocator, link[0])) orelse 6157 const fr = (try proto.readFrame(std.testing.allocator, link[0])) orelse
7214 return error.NoAttach; 6158 return error.NoAttach;
7215 defer fr.deinit(std.testing.allocator); 6159 defer fr.deinit(std.testing.allocator);
@@ -7257,15 +6201,15 @@ test "the cursor sleeps in the focused tile, whoever painted last" {
7257 // cursor: hide, CUP to the focused tile's stored position, then show 6201 // cursor: hide, CUP to the focused tile's stored position, then show
7258 // again so the cursor rests visible in the focused tile until its own 6202 // again so the cursor rests visible in the focused tile until its own
7259 // next paint — without the trailing show the cursor stays hidden. 6203 // next paint — without the trailing show the cursor stays hidden.
7260 _ = tilePaintBegin(&tiles[1]); 6204 _ = wall_pump.tilePaintBegin(&tiles[1]);
7261 tilePaintEnd(&tiles[1]); 6205 wall_pump.tilePaintEnd(&tiles[1]);
7262 try std.testing.expectEqualStrings("\x1b[?25l\x1b[6;4H\x1b[?25h", readAvail(p[0], &buf)); 6206 try std.testing.expectEqualStrings("\x1b[?25l\x1b[6;4H\x1b[?25h", readAvail(p[0], &buf));
7263 6207
7264 // A FOCUSED paint (tile 0, core null) records nothing and restores 6208 // A FOCUSED paint (tile 0, core null) records nothing and restores
7265 // nothing: no core means the shared cursor is left untouched, and the 6209 // nothing: no core means the shared cursor is left untouched, and the
7266 // pipe receives no new bytes. 6210 // pipe receives no new bytes.
7267 _ = tilePaintBegin(&tiles[0]); 6211 _ = wall_pump.tilePaintBegin(&tiles[0]);
7268 tilePaintEnd(&tiles[0]); 6212 wall_pump.tilePaintEnd(&tiles[0]);
7269 try std.testing.expectEqual(@as(usize, 0), readAvail(p[0], &buf).len); 6213 try std.testing.expectEqual(@as(usize, 0), readAvail(p[0], &buf).len);
7270 } 6214 }
7271 6215