src/tui/wall_pump.zig
Ref: Size: 39.8 KiB History
//! One tile's thread: its transport in BOTH directions, its attach, claims,
//! releases, agent channels, redials, and the paint hooks the interaction
//! core calls back on. The keyboard hands it work through the tile's mailbox
//! and doorbell. Nothing here touches another tile.
const std = @import("std");
const proto = @import("term").protocol;
const client = @import("client");
const handoff = @import("client").handoff;
const interact = @import("interact.zig");
const wv = @import("wallview.zig");
const EndReason = wv.EndReason;
const Shared = wv.Shared;
const State = wv.State;
const Tile = wv.Tile;
/// Under `paint_mu`: a clear spliced into a 64 KiB OSC 52 write eats the
/// paint after it.
pub fn copySelection(
t: *Tile,
alloc: std.mem.Allocator,
core: *interact.Core,
payload: []const u8,
) void {
var answer: interact.Copy = .none;
{
t.shared.paint_mu.lock();
defer t.shared.paint_mu.unlock();
const held = core.drag.range();
answer = core.selectionCopy(payload, held);
switch (answer) {
// `is_tty`, not the claim: a tile that released the terminal can
// still own the finished drag, and the copy is still the user's.
.text => |text| if (t.shared.is_tty)
interact.writeSelectionCopy(alloc, t.shared.out_fd, text) catch {},
.none, .too_large => {},
}
}
// Outside the hold: `tileBanner` takes the same non-reentrant lock.
if (answer == .too_large) wv.tileBanner(t, "[selection too large to copy]");
}
/// Every live tile paints its own rect, so the sink admits a paint whenever
/// this tile has not been forgotten. `paint_mu` is held for the whole paint.
pub fn tilePaintBegin(ctx: ?*anyopaque) bool {
const t: *Tile = @ptrCast(@alignCast(ctx.?));
if (t.removed.load(.acquire)) return false;
t.shared.paint_mu.lock();
// Under the lock, not before: a pump that loses the race paints over the
// popup, and `picker_frame` suppresses the repaint that would repair it.
if (wv.popupOpen(t.shared)) {
t.shared.paint_mu.unlock();
return false;
}
// A pass a relayout superseded paints at a rect the screen no longer
// has, across a neighbour's bar that nothing repaints after.
if (t.shared.repaint_gen.load(.acquire) != t.pass_gen) {
t.shared.paint_mu.unlock();
return false;
}
return true;
}
pub fn tilePaintEnd(ctx: ?*anyopaque) void {
const t: *Tile = @ptrCast(@alignCast(ctx.?));
defer t.shared.paint_mu.unlock();
// The cursor belongs to the FOCUSED tile: an unfocused paint's last act
// puts it back, hidden for the move so the show does not flash it there.
if (t.idx == t.shared.sel) {
if (t.core) |core| t.shared.cursor = core.screenCursor();
} else {
var cbuf: [26]u8 = undefined;
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;
proto.writeAllFd(t.shared.out_fd, cup) catch {};
}
}
/// The one place a wall tile puts an attach on the wire. A tile the user
/// asked for claims its rect, and that size is what lets the daemon create
/// the session; a view tile attaches at 0x0 so it can only JOIN, then takes
/// its rect with the resize doorbell one frame later.
pub fn sendAttach(t: *Tile, tr: *client.Transport, have_seq: u64, have_epoch: u64) !void {
// Snapshot under `paint_mu`: the keyboard may relayout (re-cut stripes,
// resize) concurrently with the pump's first attach.
const snap = blk: {
t.shared.paint_mu.lock();
defer t.shared.paint_mu.unlock();
break :blk .{
.cols = t.viewCols(),
.view_rows = t.viewRows(),
};
};
const cols: u16 = if (t.creates) snap.cols else 0;
const rows: u16 = if (t.creates) snap.view_rows else 0;
var buf: [proto.attach_max_len]u8 = undefined;
try tr.writeFrame(.attach, proto.encodeAttachNamed(
&buf,
cols,
rows,
have_seq,
have_epoch,
proto.wireName(t.r.session),
));
// A view tile made no size claim, so it owes its rect now, on the same
// doorbell path a relayout takes.
if (!t.creates) {
t.shared.paint_mu.lock();
t.resize_pending = true;
t.shared.paint_mu.unlock();
wv.ring(t);
}
// Re-armed on EVERY attach: a redial lands on a fresh daemon-side slot,
// which remembers no offer.
if (t.r.agent) try tr.writeFrame(.agent_offer, "");
}
/// What a pump owes itself after asking for its focus claim.
pub const ClaimStep = enum {
/// The claim landed, or the Core already held it: finish the pass.
done,
/// The sink refused and this tile still has the focus: try next pass.
rearmed,
/// The focus moved on while the popup was up: this arm is stale.
dropped,
};
/// May this tile take the terminal for the arm it is holding?
fn claimAllowed(t: *Tile) bool {
// BEFORE the claim, never after: an arm that outlives its focus and then
// succeeds puts two tiles' modes on one terminal, and nothing undoes it.
t.shared.paint_mu.lock();
defer t.shared.paint_mu.unlock();
return t.shared.sel == t.idx;
}
/// The wall's ONLY door to `Core.claimTerminal`: focus test, size adopt,
/// claim, in that order and never apart. Around it is a stale claim.
pub fn claimFocus(t: *Tile, core: *interact.Core, rect: proto.Size) ClaimStep {
if (!claimAllowed(t)) return .dropped;
// The tile's current RECT, never the whole terminal: a wall of two would
// otherwise let this tile paint over its neighbour.
if (core.size.cols != rect.cols or core.size.rows != rect.rows)
core.adoptSize(rect);
return afterClaim(t, core.claimTerminal(), core.claim != .none, core.is_tty);
}
/// A refused focus claim, judged.
pub fn afterClaim(t: *Tile, claimed: bool, held: bool, is_tty: bool) ClaimStep {
// A claim answers false for three reasons and only the SINK's refusal is
// worth another pass. Read off the Core: `picker_open` can clear between.
if (claimed or held or !is_tty) return .done;
t.shared.paint_mu.lock();
defer t.shared.paint_mu.unlock();
// ...and only while this tile is STILL the focus, or re-arming mints the
// stale arm `claimAllowed` exists to stop.
if (t.shared.sel != t.idx) return .dropped;
t.claim_pending.store(true, .release);
return .rearmed;
}
/// FOCUSED pumps only: a tile that never held the terminal has zero
/// counters, and publishing them would clobber the tile the user typed at.
fn publishStats(shared: *Shared, c: interact.PredictCounters) void {
shared.paint_mu.lock();
defer shared.paint_mu.unlock();
shared.stats = c;
}
/// Validated first: these bytes came out of a peer's `sessions_reply` and
/// `SessionName.of` memcpys with no bound of its own (`client.validPick`).
fn postAnswer(t: *Tile, pick: []const u8) void {
const name = client.validPick(pick) orelse return;
{
t.ans_mu.lock();
defer t.ans_mu.unlock();
t.ans = name;
}
t.ans_ready.store(true, .release);
wv.ringKeyboard(t.shared);
}
/// Written before `alive` clears, so no dead tile is ever seen without a
/// reason. Rings NOTHING — the keyboard's test is `!alive`.
fn endWith(t: *Tile, reason: EndReason, code: u8) void {
t.code.store(code, .release);
t.end.store(@intFromEnum(reason), .release);
}
/// Whole-mailbox chunking, so `offerKeystroke` counts keys arriving between
/// polls as suppressed. Splitting would speculate against a stale replica.
pub fn takeKeys(t: *Tile, out: []u8) []u8 {
t.in_mu.lock();
defer t.in_mu.unlock();
const n = @min(out.len, t.in_len);
@memcpy(out[0..n], t.in[0..n]);
std.mem.copyForwards(u8, t.in[0 .. t.in_len - n], t.in[n..t.in_len]);
t.in_len -= n;
return out[0..n];
}
fn drainWake(t: *const Tile) void {
wv.drainBell(t.wake_r);
}
/// Every wall dial's ssh sends its prompts to this client's popup.
pub fn askOn(target: client.Target, shared: *const Shared) client.Target {
// ONE place, so the two exclusions hold by construction: the entry attach
// opens before any listener exists, and a poller spells `BatchMode`.
var out = target;
if (out != .hand) return out;
const l = shared.prompts orelse return out;
out.hand.ask_sock = l.path;
out.hand.ask_exe = shared.prompt_exe;
return out;
}
fn dial(alloc: std.mem.Allocator, t: *Tile, target_in: client.Target) ?client.Transport {
var target = askOn(target_in, t.shared);
var backoff_ms: u64 = 0;
// `removed` as well as `running`: a tile forgotten while retrying a dead
// host must not keep a thread and a backoff alive.
while (t.shared.running.load(.acquire) and !t.removed.load(.acquire)) {
// The reason is not kept — the picker row polls the same host and
// carries it. The PID is: only it separates a refusal from a dead box.
var d: handoff.Dial = .{};
if (client.Transport.open(alloc, target, null, -1, &d)) |tr| return tr else |_| {}
// A refused prompt is an answer, and retrying is arguing with it:
// without this, Esc is answered with the same prompt every two seconds.
if (t.shared.prompts) |l| {
if (l.declined(d.ssh_pid)) {
// The bar says it before the pump goes: an unfocused tile that
// ends without exiting narrates itself (`wallview.endedTile`).
wv.paintLabel(t, .declined);
endWith(t, .declined, 1);
return null;
}
}
// An ask buys ONE attempt: the asking word per backoff would restart a
// daemon for as long as the tile lives.
if (target == .hand) target.hand.asked = false;
backoff_ms = client.nextBackoffMs(backoff_ms);
// Sliced sleep so quit is never behind a full backoff.
var slept: u64 = 0;
while (slept < backoff_ms and t.shared.running.load(.acquire) and
!t.removed.load(.acquire)) : (slept += 50)
{
std.Thread.sleep(50 * std.time.ns_per_ms);
}
}
return null;
}
/// One forwarded ssh-agent channel, this end of it: the id the daemon
/// allocated, and an fd to THIS machine's agent. Thread-local by
/// construction — the table lives in `pumpTile`'s frame, so nothing locks.
pub const AgentLocal = struct { id: u32, fd: std.posix.fd_t };
/// Fixed at `proto.agent_chans_max`, which is what the daemon opens anyway:
/// a full table costs one failed lookup, not the pump's hot path an alloc.
pub fn storeLocal(locals: []?AgentLocal, id: u32, fd: std.posix.fd_t) ?usize {
for (locals, 0..) |c, s| {
if (c != null) continue;
locals[s] = .{ .id = id, .fd = fd };
return s;
}
return null;
}
pub fn findLocal(locals: []?AgentLocal, id: u32) ?usize {
for (locals, 0..) |c, s| if (c) |ch| {
if (ch.id == id) return s;
};
return null;
}
/// Hang one channel up from this end and say so: the daemon is holding the
/// far socket open for bytes that are not coming.
pub fn closeLocal(locals: []?AgentLocal, slot: usize, transport: *client.Transport) void {
const ch = locals[slot] orelse return;
locals[slot] = null;
std.posix.close(ch.fd);
transport.writeFrame(.agent_close, &proto.encodeAgentId(ch.id)) catch {};
}
/// Lifted out of the pump so the `offered` gate has a seam a test can watch.
pub fn openAgentChan(
locals: []?AgentLocal,
id: u32,
offered: bool,
sock: []const u8,
) bool {
// The OFFER is the consent, and it is per TILE: one tile typed with `-A`
// must not hand another tile's host the keys.
if (!offered) return false;
// A live id reused. Refusing keeps the channel already on that id, which
// is the half of the collision with real bytes moving through it.
if (findLocal(locals, id) != null) return false;
const fd = client.connectAgent(sock) orelse return false;
if (storeLocal(locals, id, fd) == null) {
std.posix.close(fd);
return false;
}
return true;
}
/// Lifted out of the pump so the length cap has a seam a test can reach.
pub fn deliverAgentData(
locals: []?AgentLocal,
payload: []const u8,
transport: *client.Transport,
) bool {
const id = proto.decodeAgentId(payload) catch return false;
const s = findLocal(locals, id) orelse return true;
if (proto.agentDataOversize(payload)) {
closeLocal(locals, s, transport);
return true;
}
// Bytes onto the fd in order, never parsed: one agent message can arrive
// as several frames and several messages as one; the agent self-delimits.
proto.writeAllFd(locals[s].?.fd, payload[proto.agent_id_len..]) catch
closeLocal(locals, s, transport);
return true;
}
/// Redial only: these belonged to the dead connection, whose `dropClient`
/// already reaped the server side.
pub fn dropLocals(locals: []?AgentLocal) void {
for (locals, 0..) |c, s| if (c) |ch| {
locals[s] = null;
std.posix.close(ch.fd);
};
}
/// The transport died, or the dial has to be redone: rebuild it on the CLI's
/// backoff and re-attach quoting what this tile holds. False means the pump
/// is finished — the wall quit, or the tile was forgotten while retrying.
fn redial(
t: *Tile,
alloc: std.mem.Allocator,
core: *interact.Core,
transport: *client.Transport,
target: client.Target,
state: *State,
/// This tile's agent channels, which the dying connection owned. Dropped
/// HERE and only here, so no call site can strand one.
agents: []?AgentLocal,
) bool {
// A close after our own detach is the daemon saying goodbye, not a tear:
// redialing would re-attach the slot the user just released.
if (t.detach_ack.load(.acquire)) return false;
// Before the cold-dial refusal below, which returns without reconnecting
// — a pump that ends still owes these fds.
dropLocals(agents);
// A transport that died before any state carried no session, so there is
// nothing to resume; wall tiles retry forever because a box reboots.
if (!t.retry_cold and core.rep.session_epoch == 0) {
endWith(t, .lost, 1);
return false;
}
transport.close();
state.* = .reconnecting;
wv.paintLabel(t, state.*);
// The bar above already says `reconnecting`; the banner repeats it in
// the tile's corner, where the eye is. `banner` is gated on the sink,
// so a redial a relayout superseded writes nothing here.
core.banner("[reconnecting]");
// The resync's own paint is what will arrive, so a history page held
// here would be silently replaced a moment later.
core.dropScrollView();
transport.* = dial(alloc, t, target) orelse return false;
// Clears `state_since_attach` and drops speculation made against a
// connection that no longer exists.
core.reattached();
const have = core.rep.attachArgs();
sendAttach(t, transport, have.have_seq, have.have_epoch) catch return false;
return true;
}
/// One pump pass's geometry, plus the `.resize` that pass owes the daemon.
const Pass = struct {
top: u16,
left: u16,
rows: u16,
cols: u16,
label_rows: u16,
term_cols: u16,
term_rows: u16,
// A relayout re-cut this tile: the pump owes the daemon THIS pass's
// content size.
resize: bool,
// The repaint generation this pass was taken under: a later read would
// let a relayout be consumed by a paint at this pass's stale rect.
gen: u64,
};
pub fn takePass(t: *Tile) Pass {
t.shared.paint_mu.lock();
defer t.shared.paint_mu.unlock();
// The flag comes out of the SAME hold as the rect it describes: a
// relayout landing between two reads is swallowed, and nothing resends.
const owed = t.resize_pending;
t.resize_pending = false;
const gen = t.shared.repaint_gen.load(.acquire);
t.pass_gen = gen;
return .{
.top = t.rect.top,
.left = t.rect.left,
.rows = t.rect.rows,
.cols = t.rect.cols,
.label_rows = t.shared.labelRows(),
.term_cols = t.shared.size.cols,
.term_rows = t.shared.size.rows,
.resize = owed,
.gen = gen,
};
}
/// One tile's life: dial → attach → replay frames into its Core → repaint at
/// its rect, on its own thread. On transport death it reconnects on the CLI's
/// backoff. This thread is also the tile's only WRITER: every frame the
/// keyboard doorbells for goes out from here.
pub fn pumpTile(t: *Tile) void {
// FIRST defer, so it runs LAST: every `return` below is this tile going
// quiet, and the keyboard's test is `!alive` — so the bell follows the store.
defer {
t.alive.store(false, .release);
wv.ringKeyboard(t.shared);
// LAST, after the bell has finished reading `t.shared`: this hands the
// slot to `birthTile`, and nothing may touch the tile after it.
t.pump_done.store(true, .release);
}
// Per-thread allocator: nothing allocated here crosses threads except
// painted bytes, which go out under the paint mutex.
var gpa: std.heap.DebugAllocator(.{}) = .init;
defer _ = gpa.deinit();
const alloc = gpa.allocator();
// Whether this tile may narrate and may start a daemon travels IN its
// target. The pump's own copy: `t.r.target` belongs to the keyboard.
var target = t.r.target;
// ONE Core per tile, from birth: it owns this tile's replica, prediction
// overlay and drag for the tile's whole life. `in_fd` is the wall's stdin
// and this Core never reads it — it is passed for `is_tty`.
var core = interact.Core.initSized(
alloc,
std.posix.STDIN_FILENO,
t.shared.out_fd,
t.shared.size,
) catch return;
defer core.deinit();
// Whether this tile has EVER held the terminal: a tile that never was
// focused must not overwrite the stats of the tile that was.
var ever_focused = false;
// Focus as this pump knows it, not `core.claim`: `mux` on a pipe has no
// terminal to claim, yet its one tile is focused.
var focused = false;
// The last word on this tile's prediction: an `exit_status` RETURNS, so
// the per-pass publish is always one pass stale by then.
defer if (ever_focused) publishStats(t.shared, core.overlay.counters);
// Where this Core's paints land: this tile's rect. The sink admits a
// paint whenever the tile has not been forgotten; see `tilePaintBegin`.
core.sink = .{ .ctx = t, .begin = tilePaintBegin, .end = tilePaintEnd };
// The paint-end hook reads the core's screen cursor; only this pump
// thread dereferences it, and the core outlives the pump.
t.core = &core;
// This thread does not own the exit and cannot print the stats line —
// see `Shared.stats`.
core.owns_stats = false;
wv.paintLabel(t, .connecting);
// The ENTRY tile arrives with its link already up, dialled where the tty
// was so ssh could prompt; `adopt` moves its QUIC out-queue onto us.
var transport = if (t.pre) |pre| blk: {
var tr = pre;
t.pre = null;
tr.adopt(alloc);
break :blk tr;
} else dial(alloc, t, target) orelse return;
defer transport.close();
// The ask is SPENT on whichever branch got the link, so no reconnect can
// start a daemon or print the fallback line onto the alternate screen.
if (target == .hand) target.hand.asked = false;
// The entry tile's attach carries its rect, so no second resize follows:
// re-asserting a size the daemon just heard costs a snapshot per `mux`.
sendAttach(t, &transport, 0, 0) catch {
endWith(t, .lost, 1);
return;
};
// One question at a time, with the deadline that makes a daemon too old
// for `sessions_req` say so instead of swallowing every chord.
var pending: client.PendingSwitch = .{};
// The ssh-agent channels this tile serves, one open fd each. Table, fds
// and frames are all thread-local to this pump.
var agent_locals: [proto.agent_chans_max]?AgentLocal = @splat(null);
// Every `return` below is this tile going quiet with channels possibly
// still open; the daemon's side dies with the transport.
defer dropLocals(&agent_locals);
var state: State = .connecting;
// What this tile's paint is worth: while it matches the wall's
// generation the terminal still holds what this thread drew.
var painted_gen = t.shared.repaint_gen.load(.acquire);
// `removed` ends this thread exactly as `running` does: the defers free the
// daemon slot and nothing else. The session goes on running.
outer: while (t.shared.running.load(.acquire) and !t.removed.load(.acquire)) {
// The link, the doorbell, then one fd per live agent channel — one
// poll, because the transport has exactly one owning thread.
var fdbuf: [3 + agent_locals.len]std.posix.pollfd = undefined;
fdbuf[0] = .{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
fdbuf[1] = .{ .fd = t.wake_r, .events = std.posix.POLL.IN, .revents = 0 };
var nfds: usize = 2;
// Unread, ssh's stderr fills at 64k and ssh stops talking to the far
// end — this tile going silent for a reason no frame can explain.
const err_slot: ?usize = if (transport.errFd()) |efd| blk: {
fdbuf[nfds] = .{ .fd = efd, .events = std.posix.POLL.IN, .revents = 0 };
nfds += 1;
break :blk nfds - 1;
} else null;
// Where the agent fds start, since the stderr slot may or may not
// be there: `at` maps a readable trailing fd back to its channel.
const agent_base = nfds;
var at: [agent_locals.len]usize = undefined;
for (agent_locals, 0..) |c, s| if (c) |ch| {
at[nfds - agent_base] = s;
fdbuf[nfds] = .{ .fd = ch.fd, .events = std.posix.POLL.IN, .revents = 0 };
nfds += 1;
};
_ = std.posix.poll(fdbuf[0..nfds], transport.timeoutMs(100)) catch return;
transport.service();
if (fdbuf[1].revents != 0) drainWake(t);
if (err_slot) |s| if (fdbuf[s].revents != 0) transport.drainErr();
// The paint offset for this pass. Snapshot under `paint_mu`, which the
// keyboard writes `rect` and `label_rows` under, and use it throughout.
const snap = takePass(t);
core.row_off = snap.top + snap.label_rows;
core.col_off = snap.left;
core.owns_screen = snap.top == 0 and snap.left == 0 and
snap.label_rows == 0 and snap.cols == snap.term_cols and
snap.rows == snap.term_rows;
const snap_view_rows: u16 = snap.rows -| snap.label_rows;
const snap_view_cols: u16 = snap.cols;
// FOCUS CLAIM: the session's mouse modes and side channels go on here,
// on the thread that owns the transport and the Core.
if (t.claim_pending.swap(false, .acq_rel)) {
// A refused claim is re-armed rather than lost, or the tile stays
// focused holding no terminal. The rest of the PASS still runs:
// `takePass` already cleared the resize this tile owes.
const step = claimFocus(t, &core, .{ .cols = snap_view_cols, .rows = snap_view_rows });
// A stale arm takes nothing: no claim, no modes, no notice, and
// these predictions are not the focus's to publish.
focused = step != .dropped;
ever_focused = ever_focused or focused;
if (step == .done) {
// A sentence the keyboard left for whoever owns the terminal
// next. Taken now, shown once the grid below is up.
var notice_buf: [96]u8 = undefined;
const notice = wv.takeNotice(t.shared, ¬ice_buf);
// The replica has been hot, so a claim paints from it NOW:
// moving the focus costs a local repaint, never a wire frame.
// Only when there IS one — a blank grid is a screen mux never drew.
if (core.rep.session_epoch != 0) core.repaint() catch {};
if (notice.len > 0) core.banner(notice);
}
}
// FOCUS RELEASE: the keyboard wrote the session's release itself under
// `paint_mu` before doorbelling, so this pump owes only its own state.
if (t.release_pending.swap(false, .acq_rel)) {
focused = false;
core.releaseTerminal(.already_written);
// The speculation described a screen this terminal no longer
// shows.
core.overlay.flush();
}
// RELAYOUT DOORBELL: the pump is the transport's only writer, so
// relayout sets the flag and the `.resize` goes out from here.
if (snap.resize) {
core.overlay.setResizePending(true);
// The Core clips every paint to its size: a resize the daemon hears
// but the Core does not cuts the new grid's bottom off forever.
core.adoptSize(.{ .cols = snap_view_cols, .rows = snap_view_rows });
transport.writeFrame(
.resize,
&proto.encodeSize(snap_view_cols, snap_view_rows),
) catch {
if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
continue :outer;
};
}
// `Ctrl-\ d`: hand the daemon its slot back before the process dies.
// Only this thread may write the frame, so the keyboard is waiting.
if (t.detach_req.swap(false, .acq_rel)) {
transport.writeFrame(.detach, "") catch {};
t.detach_ack.store(true, .release);
wv.ringKeyboard(t.shared);
}
// A focus chord's question, from the thread that owns the link. The
// ANSWER goes to the keyboard, the only thread that may move focus.
const asked: client.SwitchIntent = @enumFromInt(t.ask.swap(0, .acq_rel));
if (asked != .none) {
// The deadline starts when the question goes on the wire: it
// catches a daemon too old to have heard `sessions_req`.
pending.arm(asked, std.time.milliTimestamp());
var end_buf: [proto.end_req_max_len]u8 = undefined;
const sent = switch (asked) {
// The FORCE is the second press, not a second frame: the
// daemon refused the first and this says the user meant it.
.end, .end_force => transport.writeFrame(.end_req, proto.encodeEndReq(
&end_buf,
asked == .end_force,
proto.wireName(t.r.session),
)),
else => transport.writeFrame(.sessions_req, ""),
};
sent catch {
pending.clear();
if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
continue :outer;
};
}
// Said with a banner, not stderr: the terminal is raw on the alternate
// screen. Read before `expired` spends it, so the verb is named right.
const waiting = pending.intent;
if (pending.expired(std.time.milliTimestamp()))
core.banner(switch (waiting) {
.end, .end_force => "[daemon too old to end a session]",
else => "[no session list: upgrade that daemon]",
});
// FRAMES BEFORE KEYS: `Core.forward` splits a wheel notch by the mouse
// mode a `term_modes` frame carries, so keys judged first read it stale.
// QUIC frames can arrive with the socket never going readable.
if (fdbuf[0].revents != 0 or transport.link == .quic) frames: {
while (true) {
const incoming = transport.readFrame(alloc) catch return;
const frame = switch (incoming) {
.frame => |f| f,
.incomplete => break :frames,
.closed => {
if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
continue :outer;
},
};
defer frame.deinit(alloc);
// The Core paints at this tile's offset and writes side
// channels only while it holds the claim — except `.pty_mode`,
// which gates speculation and must be true before any claim.
const routed = core.frame(frame.type, frame.payload) catch break :frames;
switch (routed) {
.skip, .handled => {},
.state => {
if (state != .up) {
state = .up;
t.ever_up.store(true, .release);
wv.paintLabel(t, state);
}
},
// The replica is suspect, not the transport: re-attach at
// (0,0), since a quoted seq invites an unfixable delta.
.resync => {
core.rep.state_since_attach = false;
// A resync renames the absolute row space, so a held
// drag would copy rows nobody selected.
core.drag.clear();
sendAttach(t, &transport, 0, 0) catch return;
},
.not_mine => switch (frame.type) {
.exit_status => {
// Before any replay frame this is the refusal
// path; after, the session really ended.
const landed = core.rep.state_since_attach;
state = if (landed) .exited else .refused;
wv.paintLabel(t, state);
endWith(
t,
if (landed) .exited else .refused,
if (landed and frame.payload.len >= 1) frame.payload[0] else 1,
);
return;
},
.taken_over => {
// Unsent by this daemon (wire-compat): somebody took
// the session, so this tile is finished, not redialing.
state = .exited;
wv.paintLabel(t, state);
endWith(t, .taken, 0);
return;
},
.sessions_reply => {
// Gated on the intent this tile's chord armed, and
// SPENT here: one question, one answer.
var name_buf: [proto.session_name_max]u8 = undefined;
// A list is only asked for to NAME a new session;
// `n`/`p` walk the wall's own tiles and ask nothing.
const pick: ?[]const u8 = switch (pending.take()) {
.new => client.nextFreeName(&name_buf, frame.payload),
else => null,
};
if (pick) |p| postAnswer(t, p);
},
.end_reply => {
const r = proto.parseEndReply(frame.payload) orelse break :frames;
_ = pending.take();
wv.onEndReply(t, r, std.time.milliTimestamp());
// An ACCEPTED end says nothing: the hangup's
// `exit_status` is next, very likely this same pass.
if (!r.accepted) {
var b: [96]u8 = undefined;
core.banner(if (r.others == 0)
endRefusal(r.reason)
else
// `X`, the key that asked: this banner is
// the tile chord's, and the picker's row
// says `x` for the key the popup uses.
// Naming the wrong one sends the user to
// `Ctrl-\ x`, which removes the pane and
// leaves the session they meant to end.
std.fmt.bufPrint(&b, "[{d} other{s} attached - X again to end]", .{
r.others,
if (r.others == 1) "" else "s",
}) catch "[others attached - X again to end]");
}
},
.selection_reply => copySelection(t, alloc, &core, frame.payload),
.agent_open => {
const id = proto.decodeAgentId(frame.payload) catch break :frames;
// Every refusal is the same answer on the wire: a
// channel that closes unspoken, read as "no agent".
const opened = openAgentChan(
&agent_locals,
id,
t.r.agent,
std.posix.getenv(proto.agent_sock_env) orelse "",
);
if (!opened)
transport.writeFrame(
.agent_close,
&proto.encodeAgentId(id),
) catch {};
},
.agent_data => if (!deliverAgentData(
&agent_locals,
frame.payload,
&transport,
)) break :frames,
.agent_close => {
const id = proto.decodeAgentId(frame.payload) catch break :frames;
// Silent, mirroring the daemon: it has retired this
// id, so a close back would be an echo.
if (findLocal(&agent_locals, id)) |s| {
const ch = agent_locals[s].?;
agent_locals[s] = null;
std.posix.close(ch.fd);
}
},
// MsgType is an open enum, so the compiler still
// wants an arm for everything `.not_mine` cannot be.
else => {},
},
}
// One event is one frame only on the socket link, and one PASS
// is not one frame: a resync is a burst, and a wheel notch
// judged before its trailing `term_modes` is misread. A zero
// timeout drains what has ARRIVED and never blocks.
if (transport.link != .quic) {
var more = [_]std.posix.pollfd{
.{ .fd = transport.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
};
const ready = std.posix.poll(&more, 0) catch 0;
if (ready == 0 or more[0].revents == 0) break :frames;
}
}
}
// What this machine's agent answered, back out as `agent_data`. AFTER
// the frame drain, so a channel just closed is already gone.
for (agent_base..nfds) |i| {
if (fdbuf[i].revents == 0) continue;
const s = at[i - agent_base];
const ch = agent_locals[s] orelse continue;
// The id goes into the head of the read's own buffer, so a frame
// costs no second copy. Capped at the wire's `agent_data_max`.
var buf: [proto.agent_id_len + proto.agent_data_max]u8 = undefined;
buf[0..proto.agent_id_len].* = proto.encodeAgentId(ch.id);
// DONTWAIT: the poll ran before the frame drain, so this slot may
// since have been refilled and be merely idle. EOF and a broken
// connection are one case — the far side needs to hear either.
const n = std.posix.recv(
ch.fd,
buf[proto.agent_id_len..],
std.posix.MSG.DONTWAIT,
) catch |err| switch (err) {
error.WouldBlock => continue,
else => 0,
};
if (n == 0) {
closeLocal(&agent_locals, s, &transport);
continue;
}
transport.writeFrame(.agent_data, buf[0 .. proto.agent_id_len + n]) catch {
if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
continue :outer;
};
}
// Whatever the keyboard left, through everything a plain client's
// keystrokes go through. Every pump drains its own mailbox.
var keys_buf: [wv.mailbox_max]u8 = undefined;
const keys = takeKeys(t, &keys_buf);
if (keys.len > 0) {
// A paint that would not allocate is not a dead link: the
// replica is untouched and the next frame redraws from it.
const step = core.forward(&transport, keys) catch interact.Step.ok;
if (step == .lost) {
if (!redial(t, alloc, &core, &transport, target, &state, &agent_locals)) return;
continue :outer;
}
// Input is moving again, so "input dropped" has stopped being news.
// Repainted, because the bar still carries the old sentence.
if (t.in_dropped.swap(false, .acq_rel)) wv.paintLabel(t, state);
}
// A prediction the daemon never answered must not sit on screen
// forever, and only the clock can say so. Focused tiles only.
if (focused) {
core.idle() catch {};
publishStats(t.shared, core.overlay.counters);
}
// A relayout re-cut the stripes and a quiet session sends nothing to
// trigger a repaint: the generation is what puts the SCREEN back. The
// drag clears too — the rect its anchor was resolved against moved.
if (snap.gen != painted_gen) {
core.drag.clear();
wv.paintLabel(t, state);
// A 0-row rect (fullscreened out) paints nothing: the daemon
// refused the 0×0 resize, so the session keeps its real grid.
if (snap_view_rows > 0) core.repaint() catch {};
painted_gen = snap.gen;
}
}
// The goodbye a removed pane is owed. `Ctrl-\ x` sets `detach_req` and
// then `removed`, and a busy pump usually leaves the loop on `removed`
// before it reaches the detach block inside it — so the frame is
// written HERE, on the way out, while `transport` is still open and
// this thread still owns it. Without it the daemon frees the slot on
// the close instead of on a goodbye, which costs a `mux` typed straight
// after the session it just gave back.
if (t.detach_req.swap(false, .acq_rel)) transport.writeFrame(.detach, "") catch {};
}
/// The same words with no brackets, for a caller that puts the session's
/// name in front of them: the picker's notice is `[NAME: WORDS]`, one pair
/// of brackets and not two. Off `endRefusal` rather than a second table, so
/// the tile's banner and the popup's footer cannot drift apart.
pub fn endRefusalWord(reason: []const u8) []const u8 {
const said = endRefusal(reason);
return said[1 .. said.len - 1];
}
/// A refusal in THIS client's words: `parseEndReply` hands back the frame's
/// tail unfiltered, so a peer's escape would run outside the replica.
pub fn endRefusal(reason: []const u8) []const u8 {
if (std.mem.eql(u8, reason, proto.end_reason.no_session))
return "[no such session on that daemon]";
if (std.mem.eql(u8, reason, proto.end_reason.bad_frame))
return "[the daemon could not read the end request]";
return "[the daemon refused to end this session]";
}