src/client/session_pump.zig
Ref: Size: 104.9 KiB History
//! One terminal-free session transport owner. The grid and semantic state
//! are shared under mu; the caller supplies a wake callback and a mailbox.
//! This is the third attach loop after the web hub and agent. The hub's
//! pumpTile is the first candidate to migrate here.
const std = @import("std");
const client = @import("client.zig");
const app_input = @import("input");
const term = @import("term");
const proto = term.protocol;
const Wire = @import("buffered_wire.zig").Wire;
/// Identity of the displayed terminal state, copied with its grid under mu.
/// Live selection keeps its range through redraws and copies current text.
/// History views retain exact source freshness because their cells are cached.
/// Revision invalidates coordinates across reconnects, resizes and mode frames.
pub const SelectionVersion = struct { seq: u64, source: u64 = 0, history_rows: u32, epoch: u64, revision: u64, history: bool = false };
pub const SelectionRequest = struct {
id: u32,
gesture: u32 = 0,
anchor: proto.SelectionPoint,
active: proto.SelectionPoint,
version: SelectionVersion,
ticket: u64 = 0, // stamped by say; cancellation also invalidates queued work
};
pub const Say = union(enum) {
input: []const u8,
paste: []const u8,
wheel: Wheel,
mouse: Mouse,
resize: proto.Size,
selection: SelectionRequest,
end: struct { request: u64, force: bool = false },
detach,
quit,
};
const paste_chunk_len = 32 * 1024;
pub const Wheel = struct {
notches: i32,
col: u16,
row: u16,
pixel_x: u32,
pixel_y: u32,
mods: app_input.Mods = .{},
};
/// A GUI gesture belongs to the modes and wire admitted at its press.
pub const MouseToken = struct { generation: u64, modes: proto.TermModes };
pub const Mouse = struct {
token: MouseToken,
kind: enum { press, motion, release },
button: u8 = 0,
col: u16,
row: u16,
pixel_x: u32,
pixel_y: u32,
mods: app_input.Mods = .{},
};
fn mouseFormat(modes: proto.TermModes) app_input.MouseFormat {
return if (modes.mouse_sgr_pixels) .sgr_pixels else if (modes.mouse_sgr) .sgr else if (modes.mouse_urxvt) .urxvt else if (modes.mouse_utf8) .utf8 else .x10;
}
const HistoryRequest = struct {
start: u32,
size: proto.Size,
revision: u64,
until: i64,
};
pub const SelectionResult = struct { id: u32, gesture: u32 = 0, status: proto.SelectionStatus, text: []u8, version: SelectionVersion };
pub const FollowPosition = struct {
id: u32,
seq: u64,
source: u64,
history_rows: u32,
status: proto.SelectionStatus,
anchor: proto.SelectionPoint,
active: proto.SelectionPoint,
};
pub const Phase = enum { dialing, attached, reconnecting, exited, refused, taken, failed, dial_failed };
pub const EndPhase = enum { idle, pending, accepted, refused, unknown };
pub const EndState = struct {
request: u64 = 0,
phase: EndPhase = .idle,
others: u8 = 0,
reason: [256]u8 = @splat(0),
reason_len: usize = 0,
pub fn reasonText(self: *const EndState) []const u8 {
return self.reason[0..self.reason_len];
}
};
pub const State = struct {
phase: Phase = .dialing,
exit_code: u8 = 0,
bell: bool = false,
ending: EndState = .{},
reason: [1024]u8 = @splat(0),
reason_len: usize = 0,
pub fn reasonText(self: *const State) []const u8 {
return self.reason[0..self.reason_len];
}
};
pub const Options = struct {
// Borrowed until stop returns, including the slices inside target.
target: client.Target,
session: []const u8 = "0",
cols: u16,
rows: u16,
/// Join only: never create a vanished session, including on reconnect.
existing_only: bool = false,
retry_initial: bool = false,
open_timeout_ms: u32 = 15000,
end_timeout_ms: u32 = 2000,
selection_timeout_ms: u32 = 2000,
wake: ?*const fn (?*anyopaque) void = null,
wake_ctx: ?*anyopaque = null,
};
pub const Pump = struct {
alloc: std.mem.Allocator,
opts: Options,
mu: std.Thread.Mutex = .{},
grid: *term.grid.Grid,
replica: term.replica.Replica,
core: client.core.ClientCore = .{},
last_apply_us: u32 = 0,
snapshot_ready: bool = false, // mu: set only after a complete valid snapshot
status: State = .{},
mailbox_mu: std.Thread.Mutex = .{},
mailbox: std.ArrayList(Say) = .empty,
wake_pipe: [2]std.posix.fd_t,
cancel_pipe: [2]std.posix.fd_t,
closing: std.atomic.Value(bool) = .init(false),
thread: ?std.Thread = null,
admitted: bool = false,
end_until: i64 = 0, // guarded by mu with status.ending
selection_pending: ?SelectionRequest = null,
selection_pending_copy: bool = false,
selection_revision: u64 = 0, // mu: connection, geometry and terminal modes
selection_ticket: u64 = 0, // mu: queued and sent request cancellation
selection_until: i64 = 0,
selection_result: ?SelectionResult = null,
selection_gesture: u32 = 0,
selection_clear: ?u32 = null,
follow_position: ?FollowPosition = null,
follow_source: u64 = 0,
follow_seq: u64 = 0,
history_waiting_metadata: bool = false,
scroll_rows: u32 = 0, // mu: requested distance from live output
history: ?*term.grid.Grid = null,
history_start: u32 = 0,
history_version: SelectionVersion = .{ .seq = 0, .history_rows = 0, .epoch = 0, .revision = 0 },
history_revision: u64 = 0,
history_dirty: bool = false,
// A cancelled request stays here until its reply is drained. The wire has
// no request ID, so replacing it could accept an old same-origin reply.
history_pending: ?HistoryRequest = null,
mouse_generation: u64 = 0, // mu: wire, geometry and mode cancellation
mouse_active: ?Mouse = null, // transport thread only, last transmitted report
clipboard: [2]?[]u8 = .{ null, null }, // mu: latest write per desktop target
pub fn start(alloc: std.mem.Allocator, opts: Options) !*Pump {
if (opts.session.len != 0 and !proto.validSessionName(opts.session)) return error.InvalidSession;
if (opts.cols == 0 or opts.rows == 0 or opts.cols > proto.max_cols) return error.InvalidSize;
const g = try term.grid.Grid.init(alloc, opts.cols, opts.rows);
errdefer g.deinit();
const wake_pipe = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
errdefer closePipe(wake_pipe);
const cancel_pipe = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
errdefer closePipe(cancel_pipe);
const self = try alloc.create(Pump);
errdefer alloc.destroy(self);
self.* = .{ .alloc = alloc, .opts = opts, .grid = g, .replica = .init(alloc, g), .wake_pipe = wake_pipe, .cancel_pipe = cancel_pipe };
self.thread = try std.Thread.spawn(.{}, entry, .{self});
return self;
}
/// Input and paste payloads are copied before returning. Quit and detach
/// never allocate.
pub fn say(self: *Pump, msg: Say) !void {
self.mailbox_mu.lock();
defer self.mailbox_mu.unlock();
if (self.closing.load(.acquire)) return;
switch (msg) {
.selection => |req| {
self.mu.lock();
defer self.mu.unlock();
self.cancelPendingSelectionLocked();
var owned = req;
if (owned.gesture == 0) owned.gesture = owned.id;
owned.ticket = self.selection_ticket;
try self.mailbox.append(self.alloc, .{ .selection = owned });
self.selection_pending = owned;
self.selection_until = std.time.milliTimestamp() + self.opts.selection_timeout_ms;
},
.end => |req| {
self.mu.lock();
defer self.mu.unlock();
if (self.status.ending.phase == .pending) return error.EndPending;
if (self.status.ending.phase == .unknown) return error.EndOutcomeUnknown;
if (self.status.phase != .attached) return error.NotAttached;
try self.mailbox.append(self.alloc, msg);
self.status.ending = .{ .request = req.request, .phase = .pending };
self.end_until = std.time.milliTimestamp() + self.opts.end_timeout_ms;
},
.quit, .detach => {
self.closing.store(true, .release);
ring(self.cancel_pipe[1], client.interrupt.detach_key);
},
else => {
var owned = msg;
switch (msg) {
.input => |bytes| owned = .{ .input = try self.alloc.dupe(u8, bytes) },
.paste => |bytes| owned = .{ .paste = try self.alloc.dupe(u8, bytes) },
else => {},
}
errdefer self.freeOwnedSay(owned);
try self.mailbox.append(self.alloc, owned);
},
}
ring(self.wake_pipe[1], 1);
}
pub fn state(self: *Pump) State {
self.mu.lock();
defer self.mu.unlock();
const result = self.status;
self.status.bell = false;
return result;
}
pub fn stop(self: *Pump) void {
self.say(.quit) catch unreachable;
if (self.thread) |thread| thread.join();
for (self.mailbox.items) |msg| self.freeOwnedSay(msg);
if (self.selection_result) |result| self.alloc.free(result.text);
if (self.history) |g| g.deinit();
for (self.clipboard) |text| if (text) |t| self.alloc.free(t);
self.mailbox.deinit(self.alloc);
closePipe(self.wake_pipe);
closePipe(self.cancel_pipe);
self.grid.deinit();
self.alloc.destroy(self);
}
fn freeOwnedSay(self: *Pump, msg: Say) void {
switch (msg) {
.input => |bytes| self.alloc.free(bytes),
.paste => |bytes| self.alloc.free(bytes),
else => {},
}
}
pub fn mouseToken(self: *Pump) ?MouseToken {
self.mu.lock();
defer self.mu.unlock();
if (!self.admitted or self.status.phase != .attached or self.closing.load(.acquire)) return null;
return .{ .generation = self.mouse_generation, .modes = self.core.terminal_modes };
}
pub fn mouseFresh(self: *Pump, token: MouseToken) bool {
const current = self.mouseToken() orelse return false;
return current.generation == token.generation;
}
/// Cancellation cannot allocate or wait for a GUI-thread transport write.
pub fn cancelMouse(self: *Pump, token: MouseToken) void {
self.mu.lock();
if (self.mouse_generation == token.generation) self.mouse_generation +%= 1;
self.mu.unlock();
ring(self.wake_pipe[1], 1);
}
pub fn takeClipboard(self: *Pump, primary: bool) ?[]u8 {
self.mu.lock();
defer self.mu.unlock();
const index: usize = @intFromBool(primary);
const text = self.clipboard[index];
self.clipboard[index] = null;
return text;
}
fn clearClipboardLocked(self: *Pump) void {
for (&self.clipboard) |*text| {
if (text.*) |t| self.alloc.free(t);
text.* = null;
}
}
/// Caller holds mu while copying both this version and the displayed grid.
pub fn selectionVersionLocked(self: *const Pump) SelectionVersion {
if (self.history != null) return self.history_version;
return .{ .seq = self.replica.last_seq, .source = if (self.follow_seq == self.replica.last_seq) self.follow_source else 0, .history_rows = self.replica.history_rows, .epoch = self.replica.session_epoch, .revision = self.selection_revision };
}
pub fn viewGridLocked(self: *const Pump) *const term.grid.Grid {
return self.history orelse self.grid;
}
pub fn viewOriginLocked(self: *const Pump) u32 {
return if (self.history != null) self.history_start else self.replica.history_rows;
}
pub fn followPositionLocked(self: *const Pump) ?FollowPosition {
const position = self.follow_position orelse return null;
if (position.id != self.selection_gesture or position.status != .ok or position.seq != self.replica.last_seq) return null;
if (self.history != null and (self.history_version.seq != position.seq or
self.history_version.source != position.source)) return null;
return position;
}
fn selectionFreshLocked(self: *const Pump, version: SelectionVersion) bool {
return self.selectionAliveLocked(version) and
version.history_rows == self.replica.history_rows and
version.source != 0 and version.source == self.follow_source and
(!version.history or version.seq == self.replica.last_seq);
}
fn selectionAliveLocked(self: *const Pump, version: SelectionVersion) bool {
return !self.closing.load(.acquire) and self.status.phase == .attached and
self.replica.state_since_attach and version.epoch == self.replica.session_epoch and
version.revision == self.selection_revision;
}
fn returnLiveLocked(self: *Pump) void {
if (self.scroll_rows == 0 and self.history == null and !self.history_dirty) return;
self.scroll_rows = 0;
self.history_dirty = false;
self.history_revision +%= 1;
if (self.history) |g| g.deinit();
self.history = null;
self.history_waiting_metadata = false;
}
fn requestHistory(self: *Pump, wire: *Wire) !void {
self.mu.lock();
const pending = self.history_pending;
if (pending) |p| {
self.mu.unlock();
// Reconnect rather than reusing an uncorrelated stream after timeout.
if (std.time.milliTimestamp() >= p.until) return error.ConnectionTimedOut;
return;
}
if (!self.history_dirty or self.scroll_rows == 0 or !self.admitted) {
self.mu.unlock();
return;
}
const req: HistoryRequest = .{ .start = self.replica.scrollStart(self.scroll_rows), .size = .{ .cols = self.opts.cols, .rows = self.opts.rows }, .revision = self.history_revision, .until = std.time.milliTimestamp() + 2000 };
self.history_pending = req;
self.history_dirty = false;
self.mu.unlock();
const bytes = proto.encodeScrollbackReq(req.start, req.size.rows);
try wire.send(.fetch_scrollback, &bytes);
}
fn historyReplyLocked(self: *Pump, payload: []const u8) !Action {
const req = self.history_pending orelse return .skip;
self.history_pending = null;
if (req.revision != self.history_revision or self.scroll_rows == 0) return .skip;
if (payload.len < 6) return error.BadPayload;
const origin = std.mem.readInt(u32, payload[0..4], .little);
const count = std.mem.readInt(u16, payload[4..6], .little);
if (count > req.size.rows or origin > self.replica.history_rows) return error.BadPayload;
const view = try term.grid.Grid.init(self.alloc, req.size.cols, req.size.rows);
errdefer view.deinit();
var bytes = payload[6..];
for (view.lines[0..count]) |*row| bytes = try term.grid.decodeRow(self.alloc, row, bytes, req.size.cols);
if (bytes.len != 0) return error.BadPayload;
view.cursor = .{ .x = req.size.cols, .y = req.size.rows };
if (self.history) |old| old.deinit();
self.history = view;
self.history_start = origin;
self.history_version = .{ .seq = self.replica.last_seq, .history_rows = self.replica.history_rows, .epoch = self.replica.session_epoch, .revision = self.selection_revision, .history = true };
self.history_waiting_metadata = true;
return .changed;
}
pub fn selectionFresh(self: *Pump, version: SelectionVersion) bool {
self.mu.lock();
defer self.mu.unlock();
return self.selectionFreshLocked(version);
}
pub fn selectionAlive(self: *Pump, gesture: u32, version: SelectionVersion) bool {
self.mu.lock();
defer self.mu.unlock();
if (!self.selectionAliveLocked(version)) return false;
if (self.selection_pending) |pending| {
if ((if (pending.gesture == 0) pending.id else pending.gesture) == gesture) return true;
}
if (self.selection_result) |result| if (result.gesture == gesture) return true;
if (self.follow_position) |position| {
if (self.selection_gesture == gesture and position.id == gesture and position.status == .ok) return true;
}
return false;
}
/// Validate again under the same lock that transfers ownership. A newer
/// frame after reply decoding must not leave an old copy ready for the UI.
pub fn takeSelection(self: *Pump) ?SelectionResult {
self.mu.lock();
defer self.mu.unlock();
const result = self.selection_result orelse return null;
self.selection_result = null;
if (!self.selectionAliveLocked(result.version)) {
self.alloc.free(result.text);
return null;
}
return result;
}
fn invalidateSelectionLocked(self: *Pump) void {
if (self.selection_gesture != 0) self.selection_clear = self.selection_gesture;
self.selection_gesture = 0;
self.follow_position = null;
self.cancelPendingSelectionLocked();
}
fn cancelPendingSelectionLocked(self: *Pump) void {
self.selection_ticket +%= 1;
self.selection_pending = null;
self.selection_pending_copy = false;
self.selection_until = 0;
self.core.pending_selection_id = null;
if (self.selection_result) |result| self.alloc.free(result.text);
self.selection_result = null;
}
pub fn cancelSelection(self: *Pump) void {
self.mu.lock();
self.invalidateSelectionLocked();
self.mu.unlock();
ring(self.wake_pipe[1], 1);
}
fn flushSelectionClear(self: *Pump, wire: *Wire) !void {
self.mu.lock();
const gesture = self.selection_clear;
self.selection_clear = null;
self.mu.unlock();
if (gesture) |id| {
const clear = proto.encodeSelectionReq(.{
.action = .clear,
.id = 0,
.gesture = id,
.epoch = 0,
.source = 0,
.anchor = .{ .row = 0, .col = 0 },
.active = .{ .row = 0, .col = 0 },
});
try wire.send(.selection_req, &clear);
}
}
fn beginSelectionLocked(self: *Pump, req: SelectionRequest) ?[proto.selection_req_len]u8 {
const gesture = if (req.gesture == 0) req.id else req.gesture;
const action: @FieldType(proto.SelectionReq, "action") =
if (self.selection_gesture == gesture) .copy else .start;
if (req.ticket != self.selection_ticket or
(if (action == .copy) !self.selectionAliveLocked(req.version) else !self.selectionFreshLocked(req.version))) return null;
self.selection_gesture = gesture;
if (action == .start) self.follow_position = null;
self.selection_pending = req;
self.selection_pending_copy = action == .copy;
self.selection_until = std.time.milliTimestamp() + self.opts.selection_timeout_ms;
return proto.encodeSelectionReq(.{
.action = action,
.id = req.id,
.gesture = gesture,
.epoch = req.version.epoch,
.source = req.version.source,
.anchor = req.anchor,
.active = req.active,
});
}
fn wake(self: *Pump) void {
if (self.opts.wake) |f| f(self.opts.wake_ctx);
}
fn publish(self: *Pump, phase: Phase, code: u8, reason: []const u8) void {
self.mu.lock();
self.setState(phase, code, reason);
self.mu.unlock();
self.wake();
}
fn setState(self: *Pump, phase: Phase, code: u8, reason: []const u8) void {
if (phase != .attached) {
self.mouse_generation +%= 1;
self.clearClipboardLocked();
self.invalidateSelectionLocked();
self.returnLiveLocked();
}
self.status.phase = phase;
self.status.exit_code = code;
self.status.reason_len = @min(reason.len, self.status.reason.len);
@memcpy(self.status.reason[0..self.status.reason_len], reason[0..self.status.reason_len]);
}
fn entry(self: *Pump) void {
defer self.finishPendingEnd("Connection ended before the End reply; outcome unknown");
self.run() catch |err| {
self.publish(.failed, 1, @errorName(err));
return;
};
self.mu.lock();
const phase = self.status.phase;
self.mu.unlock();
switch (phase) {
.dialing, .attached, .reconnecting => self.publish(if (self.closing.load(.acquire)) .exited else .failed, if (self.closing.load(.acquire)) 0 else 1, if (self.closing.load(.acquire)) "" else "session pump stopped unexpectedly"),
else => {},
}
}
fn run(self: *Pump) !void {
var first = true;
var backoff: u64 = 0;
while (!self.closing.load(.acquire)) {
var dial: client.handoff.Dial = .{};
self.expireEnd();
self.expireSelection();
var tr = client.Transport.openUntil(self.alloc, self.opts.target, null, self.cancel_pipe[0], &dial, std.time.milliTimestamp() + self.opts.open_timeout_ms) catch |err| {
if (self.closing.load(.acquire)) return;
var buf: [1024]u8 = undefined;
const failure = client.openFailure(&buf, self.opts.target, err, dial.reason.slice());
if (terminalOpenFailure(first, self.opts.retry_initial, err, dial.reason.slice())) {
self.publish(.dial_failed, 2, failure.msg);
return;
}
self.publish(.reconnecting, 0, failure.msg);
backoff = client.nextBackoffMs(backoff);
try self.waitRetry(backoff);
continue;
};
first = false;
defer tr.close();
if (self.closing.load(.acquire)) return;
var wire = try Wire.init(self.alloc, &tr);
defer wire.deinit();
const ended = self.connected(&wire) catch |err| try connectionFailure(err);
self.finishPendingEnd("Connection lost before the End reply; outcome unknown");
if (ended or self.closing.load(.acquire)) return;
tr.close();
self.publish(.reconnecting, 0, "connection lost");
backoff = client.nextBackoffMs(backoff);
try self.waitRetry(backoff);
}
}
fn waitRetry(self: *Pump, ms: u64) !void {
var fds = [_]std.posix.pollfd{.{ .fd = self.cancel_pipe[0], .events = std.posix.POLL.IN, .revents = 0 }};
_ = try std.posix.poll(&fds, @intCast(ms));
}
fn attach(self: *Pump, wire: *Wire, fresh: bool) !void {
if (fresh) try self.releaseMouse(wire) else self.mouse_active = null;
self.mu.lock();
self.mouse_generation +%= 1;
self.clearClipboardLocked();
self.invalidateSelectionLocked();
self.selection_revision +%= 1;
self.follow_position = null;
self.follow_source = 0;
self.follow_seq = 0;
self.history_waiting_metadata = false;
self.selection_gesture = 0;
self.selection_clear = null;
self.returnLiveLocked();
if (!fresh) self.history_pending = null;
const args = self.replica.attachArgs();
self.replica.state_since_attach = false;
self.admitted = false;
self.mu.unlock();
var buf: [proto.attach_max_len]u8 = undefined;
try wire.send(.attach, proto.encodeAttachNamed(&buf, if (self.opts.existing_only) 0 else self.opts.cols, if (self.opts.existing_only) 0 else self.opts.rows, if (fresh) 0 else args.have_seq, if (fresh) 0 else args.have_epoch, proto.wireName(self.opts.session)));
}
fn mail(self: *Pump, wire: *Wire, allow_wheel: bool) !void {
drain(self.wake_pipe[0]);
self.mailbox_mu.lock();
var messages = self.mailbox;
self.mailbox = .empty;
self.mailbox_mu.unlock();
var consumed = messages.items.len;
defer {
for (messages.items[0..consumed]) |msg| self.freeOwnedSay(msg);
messages.deinit(self.alloc);
}
try self.flushSelectionClear(wire);
for (messages.items, 0..) |msg, i| {
if (msg == .wheel and !allow_wheel) {
// Keep the wheel and following input ordered, while allowing
// earlier resize/key/end work between bounded receive batches.
self.mailbox_mu.lock();
defer self.mailbox_mu.unlock();
try self.mailbox.insertSlice(self.alloc, 0, messages.items[i..]);
consumed = i;
ring(self.wake_pipe[1], 1);
break;
}
switch (msg) {
.end => |req| {
self.expireEnd();
self.mu.lock();
const pending = self.status.ending.request == req.request and self.status.ending.phase == .pending;
self.mu.unlock();
if (!pending) continue;
self.mu.lock();
const admitted = self.admitted;
self.mu.unlock();
if (!admitted) {
self.finishPendingEnd("Attachment changed before End; retry to check the session");
continue;
}
var buf: [proto.end_req_max_len]u8 = undefined;
try wire.send(.end_req, proto.encodeEndReq(&buf, req.force, self.opts.session));
},
.input => |bytes| {
self.mu.lock();
const scrolled = self.scroll_rows != 0;
self.selection_revision +%= 1;
self.invalidateSelectionLocked();
self.returnLiveLocked();
self.mu.unlock();
if (scrolled) self.wake();
try wire.send(.input, bytes);
},
.paste => |bytes| {
self.mu.lock();
const scrolled = self.scroll_rows != 0;
const bracketed = self.core.terminal_modes.bracketed_paste;
self.selection_revision +%= 1;
self.invalidateSelectionLocked();
self.returnLiveLocked();
self.mu.unlock();
if (scrolled) self.wake();
if (bracketed) try wire.send(.input, app_input.paste_begin);
var offset: usize = 0;
while (offset < bytes.len) {
const end = @min(offset + paste_chunk_len, bytes.len);
try wire.send(.input, bytes[offset..end]);
offset = end;
}
if (bracketed) try wire.send(.input, app_input.paste_end);
},
.wheel => |wheel| try self.routeWheel(wire, wheel),
.mouse => |mouse| try self.routeMouse(wire, mouse),
.selection => |req| {
self.mu.lock();
const payload = self.beginSelectionLocked(req);
self.mu.unlock();
if (payload) |bytes| try wire.send(.selection_req, &bytes);
},
.resize => |size| {
try self.releaseMouse(wire);
self.mu.lock();
self.mouse_generation +%= 1;
self.invalidateSelectionLocked();
self.selection_revision +%= 1;
self.returnLiveLocked();
self.mu.unlock();
self.opts.cols = size.cols;
self.opts.rows = size.rows;
self.mu.lock();
const admitted = self.admitted;
self.mu.unlock();
if (!self.opts.existing_only or admitted) {
const buf = proto.encodeSize(size.cols, size.rows);
try wire.send(.resize, &buf);
}
},
.quit, .detach => unreachable,
}
}
// Input/resize can retire a tracker while this batch is being sent.
// Flush after the batch as well so a quiet connection still receives
// the bounded clear request.
try self.flushSelectionClear(wire);
}
fn writeMouse(wire: *Wire, event: Mouse) !bool {
const modes = event.token.modes;
if (event.kind == .release and modes.mouse_x10 and !modes.mouse_normal and !modes.mouse_button and !modes.mouse_any) return false;
var seq: [app_input.mouse_max_seq_len]u8 = undefined;
const bytes = app_input.encodeMouse(mouseFormat(modes), event.button + @as(u8, if (event.kind == .motion) 32 else 0), event.kind == .release, event.col, event.row, event.pixel_x, event.pixel_y, event.mods, &seq);
if (bytes.len == 0) return false;
try wire.send(.input, bytes);
return true;
}
fn releaseMouse(self: *Pump, wire: *Wire) !void {
var event = self.mouse_active orelse return;
self.mouse_active = null;
event.kind = .release;
_ = try writeMouse(wire, event);
}
fn reconcileMouse(self: *Pump, wire: *Wire) !void {
if (self.mouse_active) |active| {
if (!self.mouseFresh(active.token) or self.closing.load(.acquire)) try self.releaseMouse(wire);
}
}
fn routeMouse(self: *Pump, wire: *Wire, event: Mouse) !void {
try self.reconcileMouse(wire);
if (!self.mouseFresh(event.token) or !event.token.modes.appMouse() or event.button > 3) return;
switch (event.kind) {
.press => {
if (event.button == 3) return;
try self.releaseMouse(wire);
self.mu.lock();
self.returnLiveLocked();
self.invalidateSelectionLocked();
self.selection_revision +%= 1;
self.mu.unlock();
if (try writeMouse(wire, event)) self.mouse_active = event;
self.wake();
},
.motion => {
const modes = event.token.modes;
if (self.mouse_active) |active| {
if (active.button != event.button or (!modes.mouse_any and !modes.mouse_button)) return;
const same = if (modes.mouse_sgr_pixels) active.pixel_x == event.pixel_x and active.pixel_y == event.pixel_y else active.col == event.col and active.row == event.row;
if (same) return;
if (try writeMouse(wire, event)) self.mouse_active = event;
} else if (event.button == 3 and modes.mouse_any) {
_ = try writeMouse(wire, event);
}
},
.release => {
const active = self.mouse_active orelse return;
if (event.button != active.button) return;
// If a legacy release is outside its encoding range, release
// at the last representable point instead of leaving it held.
if (!try writeMouse(wire, event)) try self.releaseMouse(wire);
self.mouse_active = null;
},
}
}
fn routeWheel(self: *Pump, wire: *Wire, wheel: Wheel) !void {
self.mu.lock();
if (!self.admitted or self.status.phase != .attached or wheel.notches == 0) {
self.mu.unlock();
return;
}
const modes = self.core.terminal_modes;
const arrows = modes.alt_screen and self.scroll_rows == 0;
if (!modes.appMouse() and !arrows) {
const rows: u32 = @intCast(@min(@as(u64, @abs(wheel.notches)) * 3, std.math.maxInt(u32)));
const next = if (wheel.notches > 0) @min(self.scroll_rows +| rows, self.replica.history_rows) else self.scroll_rows -| rows;
if (next != self.scroll_rows) {
if (next == 0) self.returnLiveLocked() else {
self.scroll_rows = next;
self.history_revision +%= 1;
self.history_dirty = true;
}
}
self.mu.unlock();
self.wake();
return;
}
self.returnLiveLocked();
self.invalidateSelectionLocked();
self.selection_revision +%= 1;
self.mu.unlock();
self.wake();
var seq: [app_input.mouse_max_seq_len]u8 = undefined;
const bytes = if (modes.appMouse()) app_input.encodeWheel(
mouseFormat(modes),
wheel.notches > 0,
wheel.col,
wheel.row,
wheel.pixel_x,
wheel.pixel_y,
wheel.mods,
&seq,
) else app_input.encodeWheelArrow(wheel.notches > 0, modes.cursor_keys, &seq);
if (bytes.len == 0) return;
// Bound a single mailbox event even if an input adapter supplies an
// extreme delta. Ordinary wheels are one or a few notches per event.
var left: u32 = @as(u32, @intCast(@min(@abs(wheel.notches), 1024))) * @as(u32, if (modes.appMouse()) 1 else 3);
var batch: [1024]u8 = undefined;
while (left != 0) {
const n = @min(left, batch.len / bytes.len);
for (0..n) |i| @memcpy(batch[i * bytes.len ..][0..bytes.len], bytes);
try wire.send(.input, batch[0 .. n * bytes.len]);
left -= @intCast(n);
}
}
/// Drain ready stream bytes before interpreting semantic mouse intent.
/// A partial header is progress, not proof that the next mode frame is absent.
const ReadState = enum { idle, busy, closed, ended };
fn receiveReady(self: *Pump, wire: *Wire) !ReadState {
var changed = false;
defer if (changed) self.wake();
var frames: usize = 0;
for (0..256) |_| {
var fd = [_]std.posix.pollfd{.{ .fd = wire.tr.pollFd(), .events = std.posix.POLL.IN, .revents = 0 }};
_ = try std.posix.poll(&fd, 0);
if (fd[0].revents == 0 and wire.tr.link != .quic) return .idle;
const before = wire.input.items.len;
switch (try wire.read()) {
.closed => return .closed,
.incomplete => if (wire.input.items.len == before) return .idle,
.frame => |frame| {
defer frame.deinit(self.alloc);
const action = try self.onFrame(frame.type, frame.payload);
var admitted_now = false;
self.mu.lock();
if (!self.admitted and (frame.type == .snapshot or frame.type == .delta) and action == .changed) {
self.admitted = true;
admitted_now = true;
}
self.mu.unlock();
if (admitted_now and self.opts.existing_only) {
const size = proto.encodeSize(self.opts.cols, self.opts.rows);
try wire.send(.resize, &size);
}
changed = changed or action != .skip;
switch (action) {
.resync => try self.attach(wire, true),
.end => return .ended,
else => {},
}
frames += 1;
if (frames == 64) return .busy;
},
}
}
return .busy;
}
fn connected(self: *Pump, wire: *Wire) !bool {
try self.attach(wire, false);
while (true) {
self.expireEnd();
self.expireSelection();
wire.tr.service();
var busy = false;
if (!self.closing.load(.acquire)) switch (try self.receiveReady(wire)) {
.closed => return false,
.ended => return true,
.busy => busy = true,
.idle => {},
};
try self.reconcileMouse(wire);
try self.mail(wire, !busy);
try self.requestHistory(wire);
if (self.closing.load(.acquire)) {
try wire.send(.detach, "");
// A responsive peer receives detach; a stalled peer cannot
// prevent the owner from joining this thread.
const end = std.time.milliTimestamp() + 100;
while (wire.pending() and std.time.milliTimestamp() < end) {
try wire.flush();
wire.tr.service();
var fds = [_]std.posix.pollfd{.{ .fd = if (wire.tr.link == .quic) wire.tr.pollFd() else wire.writeFd(), .events = if (wire.tr.link == .quic) std.posix.POLL.IN else std.posix.POLL.OUT, .revents = 0 }};
_ = try std.posix.poll(&fds, 5);
}
return true;
}
var fds = [_]std.posix.pollfd{
.{ .fd = wire.tr.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = self.wake_pipe[0], .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = wire.tr.errFd() orelse -1, .events = std.posix.POLL.IN, .revents = 0 },
.{ .fd = if (wire.tr.link != .quic and wire.pending()) wire.writeFd() else -1, .events = std.posix.POLL.OUT, .revents = 0 },
};
_ = try std.posix.poll(&fds, self.endWaitMs(wire.tr.timeoutMs(if (busy) 0 else 1000)));
wire.tr.service();
if (fds[2].revents != 0) wire.tr.drainErr();
if (fds[3].revents != 0) try wire.flush();
}
}
const Action = enum { skip, changed, resync, end };
fn onFrame(self: *Pump, kind: proto.MsgType, payload: []const u8) !Action {
self.mu.lock();
defer self.mu.unlock();
if (kind != .selection_reply) self.history_waiting_metadata = false;
switch (kind) {
.scrollback_chunk => return self.historyReplyLocked(payload),
.end_reply => {
if (self.status.ending.phase != .pending) return .skip;
if (proto.parseEndReply(payload)) |reply| {
self.setEnd(if (reply.accepted) .accepted else .refused, reply.others, reply.reason);
} else self.setEnd(.unknown, 0, "Invalid End reply; retry to check the session");
return .changed;
},
.snapshot, .delta => {
const begin = std.time.nanoTimestamp();
const old_epoch = self.replica.session_epoch;
const old_cols = self.grid.cols;
const old_rows = self.grid.rows;
const applied = self.replica.apply(kind, payload) catch |err| switch (err) {
error.BadPayload => return .skip,
else => return err,
};
self.last_apply_us = @intCast(@min(std.math.maxInt(u32), @max(0, @divTrunc(std.time.nanoTimestamp() - begin, 1000))));
// Output and scrollback growth preserve a tracked source. Only
// a resync, session epoch, or geometry change invalidates its
// coordinates; the follow-state reply remaps ordinary output.
if (applied == .resync or self.replica.session_epoch != old_epoch or
self.grid.cols != old_cols or self.grid.rows != old_rows)
{
self.selection_revision +%= 1;
self.invalidateSelectionLocked();
}
if (self.replica.session_epoch != old_epoch or self.grid.cols != old_cols or self.grid.rows != old_rows) self.mouse_generation +%= 1;
if (self.replica.session_epoch != old_epoch) self.returnLiveLocked();
if (self.scroll_rows != 0) {
self.scroll_rows = @min(self.scroll_rows, self.replica.history_rows);
self.history_revision +%= 1;
self.history_dirty = self.scroll_rows != 0;
if (self.scroll_rows == 0) self.returnLiveLocked();
}
if (applied == .resync) return .resync;
if (kind == .snapshot) self.snapshot_ready = true;
self.setState(.attached, 0, "");
return .changed;
},
.exit_status => {
const admitted = self.replica.state_since_attach;
self.setState(if (admitted) .exited else .refused, if (admitted and payload.len > 0) payload[0] else 1, if (admitted) "" else if (payload.len > 1) payload[1..] else "session refused");
return .end;
},
.taken_over => {
self.setState(.taken, 0, "session taken over");
return .end;
},
.selection_reply => {
const reply = proto.decodeSelectionReply(payload) catch return .skip;
if (reply.seq == self.replica.last_seq) {
self.follow_seq = reply.seq;
self.follow_source = reply.source;
// The server emits this position metadata immediately after
// a history chunk. Bind that chunk to the source that was
// actually rendered; never borrow a later live source.
if (self.history != null and self.history_waiting_metadata and reply.id == 0) {
self.history_version.source = reply.source;
self.history_version.seq = reply.seq;
self.history_version.history_rows = reply.history_rows;
self.history_waiting_metadata = false;
}
if (reply.gesture != 0) {
const matches = self.selection_gesture == reply.gesture;
if (matches) self.follow_position = .{
.id = reply.gesture,
.seq = reply.seq,
.source = reply.source,
.history_rows = reply.history_rows,
.status = if (reply.status == .too_large) .ok else reply.status,
.anchor = reply.anchor,
.active = reply.active,
};
} else if (self.selection_gesture != 0) {
// A zero-gesture state is the daemon's explicit
// tracker retirement (metadata before a gesture has
// no tracker to retire).
self.follow_position = null;
}
}
const pending = self.selection_pending orelse return .changed;
const gesture = if (pending.gesture == 0) pending.id else pending.gesture;
if (reply.id != pending.id or reply.gesture != gesture) return .changed;
const pending_alive = if (self.selection_pending_copy)
self.selectionAliveLocked(pending.version)
else
self.selectionFreshLocked(pending.version);
if (!pending_alive or reply.seq != self.replica.last_seq or
(!self.selection_pending_copy and pending.version.source != reply.source))
{
self.selection_pending = null;
return .changed;
}
const text = try self.alloc.dupe(u8, reply.text);
self.selection_pending = null;
if (self.selection_result) |old| self.alloc.free(old.text);
self.selection_result = .{ .id = reply.id, .gesture = gesture, .status = reply.status, .text = text, .version = pending.version };
return .changed;
},
else => switch (self.core.receive(kind, payload)) {
.effect => |effect| switch (effect) {
.bell => self.status.bell = true,
.clipboard_set => |clip| {
const decoded = client.core.decodeClipboard(self.alloc, clip.target, clip.base64) catch |err| switch (err) {
error.InvalidClipboard => return .skip,
else => return err,
} orelse return .skip;
if (!self.admitted or self.status.phase != .attached) {
self.alloc.free(decoded.text);
return .skip;
}
const index: usize = @intFromBool(decoded.primary);
if (self.clipboard[index]) |old| self.alloc.free(old);
self.clipboard[index] = decoded.text;
},
},
.state => {
self.mouse_generation +%= 1;
self.invalidateSelectionLocked();
self.selection_revision +%= 1;
if (self.scroll_rows != 0) {
self.history_revision +%= 1;
self.history_dirty = true;
}
return .changed;
},
else => return .skip,
},
}
return .changed;
}
fn setEnd(self: *Pump, phase: EndPhase, others: u8, reason: []const u8) void {
self.status.ending.phase = phase;
self.status.ending.others = others;
self.status.ending.reason_len = @min(reason.len, self.status.ending.reason.len);
@memcpy(self.status.ending.reason[0..self.status.ending.reason_len], reason[0..self.status.ending.reason_len]);
}
fn expireEnd(self: *Pump) void {
self.mu.lock();
const expired = self.status.ending.phase == .pending and std.time.milliTimestamp() >= self.end_until;
if (expired) self.setEnd(.unknown, 0, "End reply timed out; outcome unknown. Retry to check the session");
self.mu.unlock();
if (expired) self.wake();
}
fn expireSelection(self: *Pump) void {
self.mu.lock();
const expired = self.selection_pending != null and std.time.milliTimestamp() >= self.selection_until;
if (expired) {
const pending = self.selection_pending.?;
self.invalidateSelectionLocked();
self.selection_result = .{ .id = pending.id, .gesture = if (pending.gesture == 0) pending.id else pending.gesture, .status = .unavailable, .text = &.{}, .version = pending.version };
}
self.mu.unlock();
if (expired) self.wake();
}
fn endWaitMs(self: *Pump, cap: i32) i32 {
self.mu.lock();
defer self.mu.unlock();
const now = std.time.milliTimestamp();
var wait: i64 = cap;
if (self.status.ending.phase == .pending) wait = @min(wait, @max(0, self.end_until - now));
if (self.selection_pending != null) wait = @min(wait, @max(0, self.selection_until - now));
if (self.history_pending) |req| wait = @min(wait, @max(0, req.until - now));
return @intCast(wait);
}
fn finishPendingEnd(self: *Pump, reason: []const u8) void {
self.mu.lock();
const pending = self.status.ending.phase == .pending;
if (pending) self.setEnd(.unknown, 0, reason);
self.mu.unlock();
if (pending) self.wake();
}
};
fn retryableOpen(err: anyerror, reason: []const u8) bool {
if (client.handoff.classifyReason(reason) == .authentication_refused) return false;
return switch (err) {
error.FileNotFound, error.ConnectionRefused, error.ConnectionTimedOut, error.Timeout, error.NetworkUnreachable, error.HostUnreachable, error.QuicHandshakeFailed, error.UnknownHostName, error.UnterminatedLine => true,
else => false,
};
}
fn terminalOpenFailure(first: bool, retry_initial: bool, err: anyerror, reason: []const u8) bool {
return (first and !(retry_initial and retryableOpen(err, reason))) or
(!first and client.handoff.classifyReason(reason) == .authentication_refused);
}
test "retryableOpen: authentication diagnostics are terminal while unknown announce EOF retries" {
try std.testing.expect(!retryableOpen(error.UnterminatedLine, "Permission denied (publickey)."));
try std.testing.expect(!retryableOpen(error.UnterminatedLine, "Received disconnect from box: 2: Too many authentication failures"));
try std.testing.expect(retryableOpen(error.UnterminatedLine, ""));
try std.testing.expect(retryableOpen(error.UnterminatedLine, "ssh: connect to host box port 22: No route to host"));
// The status code alone does not establish an authentication refusal.
try std.testing.expect(retryableOpen(error.UnterminatedLine, "255"));
}
test "reconnect policy: non-auth transport failures retain retry behavior" {
try std.testing.expect(!terminalOpenFailure(false, false, error.ConnectionResetByPeer, "connection reset"));
try std.testing.expect(!terminalOpenFailure(false, false, error.SystemResources, "resource unavailable"));
try std.testing.expect(terminalOpenFailure(false, false, error.UnterminatedLine, "Permission denied (publickey)."));
try std.testing.expect(terminalOpenFailure(true, false, error.UnterminatedLine, ""));
try std.testing.expect(!terminalOpenFailure(true, true, error.UnterminatedLine, ""));
}
// Only a lost connection earns a redial. Resource exhaustion, poll errors,
// and other local failures must reach entry's failure publication.
fn connectionFailure(err: anyerror) anyerror!bool {
return switch (err) {
error.Closed, error.ConnectionLost, error.BrokenPipe, error.ConnectionResetByPeer, error.ConnectionTimedOut, error.SocketNotConnected => false,
else => err,
};
}
fn closePipe(fds: [2]std.posix.fd_t) void {
for (fds) |fd| std.posix.close(fd);
}
fn ring(fd: std.posix.fd_t, byte: u8) void {
_ = std.posix.write(fd, &.{byte}) catch {};
}
fn drain(fd: std.posix.fd_t) void {
var buf: [128]u8 = undefined;
while ((std.posix.read(fd, &buf) catch return) > 0) {}
}
const TestPeer = struct {
tmp: @import("testtmp").TmpDir,
listener: std.net.Server,
path: []u8,
fn init() !TestPeer {
const alloc = std.testing.allocator;
var tmp = try @import("testtmp").TmpDir.make();
errdefer tmp.cleanup();
const path = try std.fmt.allocPrint(alloc, "{s}/native.sock", .{tmp.path()});
errdefer alloc.free(path);
const addr = try std.net.Address.initUnix(path);
return .{ .tmp = tmp, .path = path, .listener = try addr.listen(.{}) };
}
fn deinit(self: *TestPeer) void {
self.listener.deinit();
std.testing.allocator.free(self.path);
self.tmp.cleanup();
}
fn start(self: *TestPeer) !*Pump {
return Pump.start(std.testing.allocator, .{ .target = .{ .sock = self.path }, .session = "native-test", .cols = 11, .rows = 3 });
}
fn accept(self: *TestPeer) !std.net.Stream {
var fds = [_]std.posix.pollfd{.{ .fd = self.listener.stream.handle, .events = std.posix.POLL.IN, .revents = 0 }};
if (try std.posix.poll(&fds, 2000) == 0) return error.AcceptTimeout;
return (try self.listener.accept()).stream;
}
};
fn testFrame(stream: std.net.Stream, kind: proto.MsgType) !proto.Frame {
var l = client.Link{ .fd = stream.handle };
return (try l.awaitFrame(std.testing.allocator, kind, 2000, .{})) orelse error.FrameTimeout;
}
fn testPhase(pump: *Pump, expected: Phase) !State {
const end = std.time.milliTimestamp() + 2000;
while (std.time.milliTimestamp() < end) {
const s = pump.state();
if (s.phase == expected) return s;
std.Thread.sleep(std.time.ns_per_ms);
}
const s = pump.state();
std.debug.print("session pump expected {s}, got {s}: {s}\n", .{ @tagName(expected), @tagName(s.phase), s.reasonText() });
return error.PhaseTimeout;
}
fn testSnapshot() [34]u8 {
var bytes: [34]u8 = @splat(0);
proto.writeSnapshotPrefix(bytes[0..proto.snapshot_prefix_len], .{ .seq = 37, .history_rows = 0, .cols = 11, .rows = 3, .epoch = 93 });
proto.writeSnapshotCursor(bytes[proto.snapshot_prefix_len..][0..proto.snapshot_cursor_len], 4, 2);
return bytes;
}
test "session pump idle snapshot services copied input resize and detach in order" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
defer pump.stop();
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
defer attach_frame.deinit(std.testing.allocator);
const args = try proto.decodeAttach(attach_frame.payload);
try std.testing.expectEqualStrings("native-test", args.name);
try std.testing.expectEqual(@as(u16, 11), args.cols);
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
_ = try testPhase(pump, .attached);
// The peer is now silent; mailbox processing must not wait for another
// inbound frame. The input source can be reused as soon as say returns.
var input = [_]u8{ 'h', 'i' };
try pump.say(.{ .input = &input });
@memset(&input, 'x');
try pump.say(.{ .resize = .{ .cols = 19, .rows = 7 } });
try pump.say(.detach);
const one = (try proto.readFrame(std.testing.allocator, stream.handle)).?;
defer one.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.MsgType.input, one.type);
try std.testing.expectEqualStrings("hi", one.payload);
const two = (try proto.readFrame(std.testing.allocator, stream.handle)).?;
defer two.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.MsgType.resize, two.type);
try std.testing.expectEqual(proto.Size{ .cols = 19, .rows = 7 }, try proto.decodeSize(two.payload));
const three = (try proto.readFrame(std.testing.allocator, stream.handle)).?;
defer three.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.MsgType.detach, three.type);
_ = try testPhase(pump, .exited);
}
test "session pump stop interrupts a partial socket frame" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
var stopped = false;
defer if (!stopped) pump.stop();
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
attach_frame.deinit(std.testing.allocator);
const hdr = proto.encodeHeader(.snapshot, 999);
try proto.writeAllFd(stream.handle, hdr[0..2]);
std.Thread.sleep(10 * std.time.ns_per_ms);
const start = std.time.milliTimestamp();
pump.stop();
stopped = true;
try std.testing.expect(std.time.milliTimestamp() - start < 500);
}
test "session pump reconnect resumes coordinates but refuses before new replay" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
defer pump.stop();
{
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
attach_frame.deinit(std.testing.allocator);
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
_ = try testPhase(pump, .attached);
}
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
defer attach_frame.deinit(std.testing.allocator);
const args = try proto.decodeAttach(attach_frame.payload);
try std.testing.expectEqual(@as(u64, 37), args.have_seq);
try std.testing.expectEqual(@as(u64, 93), args.have_epoch);
try proto.writeFrame(stream.handle, .exit_status, &.{1});
const state = try testPhase(pump, .refused);
try std.testing.expectEqual(@as(u8, 1), state.exit_code);
}
test "session pump malformed short snapshot skips and destructive snapshot fails" {
for ([_]bool{ false, true }) |destructive| {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
defer pump.stop();
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
attach_frame.deinit(std.testing.allocator);
if (destructive) {
const snapshot = testSnapshot();
try proto.writeFrame(stream.handle, .snapshot, snapshot[0..28]);
const state = try testPhase(pump, .failed);
try std.testing.expectEqualStrings("SnapshotAborted", state.reasonText());
} else {
try proto.writeFrame(stream.handle, .snapshot, "short");
try proto.writeFrame(stream.handle, .exit_status, &.{1});
_ = try testPhase(pump, .refused);
pump.mu.lock();
defer pump.mu.unlock();
try std.testing.expectEqual(@as(u64, 0), pump.replica.last_seq);
}
}
}
test "session pump resync requests fresh snapshot and preserves shell exit code" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
defer pump.stop();
const stream = try peer.accept();
defer stream.close();
const first = try testFrame(stream, .attach);
first.deinit(std.testing.allocator);
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
_ = try testPhase(pump, .attached);
try proto.writeFrame(stream.handle, .delta, "bad");
const fresh = try testFrame(stream, .attach);
defer fresh.deinit(std.testing.allocator);
const args = try proto.decodeAttach(fresh.payload);
try std.testing.expectEqual(@as(u64, 0), args.have_seq);
try std.testing.expectEqual(@as(u64, 0), args.have_epoch);
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
try proto.writeFrame(stream.handle, .exit_status, &.{42});
const ended = try testPhase(pump, .exited);
try std.testing.expectEqual(@as(u8, 42), ended.exit_code);
}
test "session pump initial dial failure reports common diagnostic and code two" {
var peer = try TestPeer.init();
defer peer.deinit();
const missing = try std.fmt.allocPrint(std.testing.allocator, "{s}/missing", .{peer.tmp.path()});
defer std.testing.allocator.free(missing);
const pump = try Pump.start(std.testing.allocator, .{ .target = .{ .sock = missing }, .cols = 11, .rows = 3 });
defer pump.stop();
const state = try testPhase(pump, .dial_failed);
try std.testing.expectEqual(@as(u8, 2), state.exit_code);
try std.testing.expect(std.mem.indexOf(u8, state.reasonText(), missing) != null);
}
test "session pump pipe framing remains interruptible while peer sends no frames" {
const pump = try Pump.start(std.testing.allocator, .{ .target = .{ .via = "cat" }, .cols = 11, .rows = 3 });
// cat echoes the attach and input as unknown inbound frame types. A
// draining blocking read would wait forever after that first frame.
try pump.say(.{ .input = "hello" });
std.Thread.sleep(20 * std.time.ns_per_ms);
const start = std.time.milliTimestamp();
pump.stop();
try std.testing.expect(std.time.milliTimestamp() - start < 500);
}
test "session pump stop cancels a silent handoff dial" {
const pump = try Pump.start(std.testing.allocator, .{
.target = .{ .hand = .{ .host = "isolated-test", .ssh_argv = &.{ "sleep", "2" }, .cache_path = null } },
.cols = 11,
.rows = 3,
});
std.Thread.sleep(30 * std.time.ns_per_ms);
const start = std.time.milliTimestamp();
pump.stop();
try std.testing.expect(std.time.milliTimestamp() - start < 500);
}
test "session pump stop cancels reconnect backoff and oversized frame publishes failure" {
for ([_]bool{ false, true }) |oversized| {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
var stopped = false;
defer if (!stopped) pump.stop();
{
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
attach_frame.deinit(std.testing.allocator);
if (oversized) {
try proto.writeAllFd(stream.handle, &proto.encodeHeader(.snapshot, proto.max_payload + 1));
const state = try testPhase(pump, .failed);
try std.testing.expectEqualStrings("FrameTooLarge", state.reasonText());
}
}
if (!oversized) _ = try testPhase(pump, .reconnecting);
const start = std.time.milliTimestamp();
pump.stop();
stopped = true;
try std.testing.expect(std.time.milliTimestamp() - start < 150);
}
}
test "session pump bell is consumed and takeover wakes outside the grid mutex" {
const Wake = struct {
calls: std.atomic.Value(u32) = .init(0),
fn fire(ctx: ?*anyopaque) void {
const self: *@This() = @ptrCast(@alignCast(ctx.?));
_ = self.calls.fetchAdd(1, .monotonic);
}
};
var peer = try TestPeer.init();
defer peer.deinit();
var callback: Wake = .{};
const pump = try Pump.start(std.testing.allocator, .{ .target = .{ .sock = peer.path }, .cols = 11, .rows = 3, .wake = Wake.fire, .wake_ctx = &callback });
defer pump.stop();
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
attach_frame.deinit(std.testing.allocator);
try proto.writeFrame(stream.handle, .term_event, &.{1});
const end = std.time.milliTimestamp() + 2000;
while (callback.calls.load(.monotonic) == 0 and std.time.milliTimestamp() < end) std.Thread.sleep(std.time.ns_per_ms);
try std.testing.expect(pump.state().bell);
try std.testing.expect(!pump.state().bell);
try proto.writeFrame(stream.handle, .taken_over, "");
_ = try testPhase(pump, .taken);
try std.testing.expect(callback.calls.load(.monotonic) >= 2);
}
test "session pump unexpected errors escape reconnect classification" {
try std.testing.expect(!try connectionFailure(error.ConnectionResetByPeer));
try std.testing.expectError(error.SystemResources, connectionFailure(error.SystemResources));
try std.testing.expectError(error.Unexpected, connectionFailure(error.Unexpected));
try std.testing.expectError(error.NotOpenForReading, connectionFailure(error.NotOpenForReading));
}
test "session pump QUIC wakes and services mailbox between bounded buffered batches" {
const alloc = std.testing.allocator;
// A real QUIC client with an isolated UDP destination. Frames already
// delivered into its stream buffer need no new datagram to be consumed.
var addr = std.net.Address.initIp4(.{ 127, 0, 0, 1 }, 0);
const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.DGRAM | std.posix.SOCK.CLOEXEC, 0);
defer std.posix.close(fd);
try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
var addr_len = addr.getOsSockLen();
try std.posix.getsockname(fd, &addr.any, &addr_len);
const q = try @import("quic").Client.connect(alloc, addr, .{ .bytes = @splat(7) }, 5000);
var tr: client.Transport = .{ .link = .{ .quic = .{ .cl = q, .alloc = alloc } } };
defer tr.close();
for (1..131) |seq| {
var snapshot = testSnapshot();
std.mem.writeInt(u64, snapshot[0..8], seq, .little);
try proto.appendFrame(&q.in, alloc, .snapshot, &snapshot);
}
try proto.appendFrame(&q.in, alloc, .exit_status, &.{42});
const Callback = struct {
pump: *Pump,
seqs: [3]u64 = @splat(0),
cols: [3]u16 = @splat(0),
n: usize = 0,
err: ?anyerror = null,
fn fire(ctx: ?*anyopaque) void {
const self: *@This() = @ptrCast(@alignCast(ctx.?));
// Acquiring mu here also pins that wakes never hold the grid.
self.pump.mu.lock();
if (self.n < 3) {
self.seqs[self.n] = self.pump.replica.last_seq;
self.cols[self.n] = self.pump.opts.cols;
}
self.pump.mu.unlock();
self.n += 1;
if (self.n == 1) self.pump.say(.{ .resize = .{ .cols = 19, .rows = 7 } }) catch |err| {
self.err = err;
};
}
};
const pump = setup: {
const g = try term.grid.Grid.init(alloc, 11, 3);
errdefer g.deinit();
const wp = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
errdefer closePipe(wp);
const cp = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
errdefer closePipe(cp);
const p = try alloc.create(Pump);
p.* = .{ .alloc = alloc, .opts = .{ .target = .{ .sock = "" }, .cols = 11, .rows = 3 }, .grid = g, .replica = .init(alloc, g), .wake_pipe = wp, .cancel_pipe = cp };
break :setup p;
};
var callback: Callback = .{ .pump = pump };
pump.opts.wake = Callback.fire;
pump.opts.wake_ctx = &callback;
defer pump.stop();
var wire = try Wire.init(alloc, &tr);
defer wire.deinit();
const start = std.time.milliTimestamp();
try std.testing.expect(try pump.connected(&wire));
try std.testing.expect(std.time.milliTimestamp() - start < 500);
try std.testing.expectEqual(@as(?anyerror, null), callback.err);
try std.testing.expectEqual(@as(usize, 3), callback.n);
try std.testing.expectEqualSlices(u64, &.{ 64, 128, 130 }, &callback.seqs);
try std.testing.expectEqualSlices(u16, &.{ 11, 19, 19 }, &callback.cols);
try std.testing.expectEqual(@as(u8, 42), pump.state().exit_code);
const sent_attach = (try proto.takeFrame(alloc, &tr.link.quic.qout)).?;
defer sent_attach.deinit(alloc);
try std.testing.expectEqual(proto.MsgType.attach, sent_attach.type);
const sent_resize = (try proto.takeFrame(alloc, &tr.link.quic.qout)).?;
defer sent_resize.deinit(alloc);
try std.testing.expectEqual(proto.MsgType.resize, sent_resize.type);
}
test "session pump pipe snapshot paints then remains responsive while idle" {
const alloc = std.testing.allocator;
var tmp = try @import("testtmp").TmpDir.make();
defer tmp.cleanup();
var bytes: std.ArrayList(u8) = .empty;
defer bytes.deinit(alloc);
try proto.appendFrame(&bytes, alloc, .snapshot, &testSnapshot());
try tmp.dir.writeFile(.{ .sub_path = "snapshot", .data = bytes.items });
const command = try std.fmt.allocPrint(alloc, "cat {s}/snapshot -", .{tmp.path()});
defer alloc.free(command);
const pump = try Pump.start(alloc, .{ .target = .{ .via = command }, .cols = 11, .rows = 3 });
var stopped = false;
defer if (!stopped) pump.stop();
_ = try testPhase(pump, .attached);
try pump.say(.{ .input = "after snapshot" });
try pump.say(.{ .resize = .{ .cols = 19, .rows = 7 } });
const start = std.time.milliTimestamp();
pump.stop();
stopped = true;
try std.testing.expect(std.time.milliTimestamp() - start < 500);
}
test "existing-only pump never sends a create-size attach or pre-admission resize" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try Pump.start(std.testing.allocator, .{ .target = .{ .sock = peer.path }, .session = "native-test", .cols = 11, .rows = 3, .existing_only = true });
defer pump.stop();
{
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
defer attach_frame.deinit(std.testing.allocator);
const args = try proto.decodeAttach(attach_frame.payload);
try std.testing.expectEqual(@as(u16, 0), args.cols);
try std.testing.expectEqual(@as(u16, 0), args.rows);
try pump.say(.{ .resize = .{ .cols = 19, .rows = 7 } });
var fds = [_]std.posix.pollfd{.{ .fd = stream.handle, .events = std.posix.POLL.IN, .revents = 0 }};
try std.testing.expectEqual(@as(usize, 0), try std.posix.poll(&fds, 30));
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
const resize = try testFrame(stream, .resize);
defer resize.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.Size{ .cols = 19, .rows = 7 }, try proto.decodeSize(resize.payload));
_ = try testPhase(pump, .attached);
}
const stream = try peer.accept();
defer stream.close();
const again = try testFrame(stream, .attach);
defer again.deinit(std.testing.allocator);
const args = try proto.decodeAttach(again.payload);
try std.testing.expectEqual(@as(u16, 0), args.cols);
try std.testing.expectEqual(@as(u16, 0), args.rows);
try std.testing.expectEqual(@as(u64, 37), args.have_seq);
try pump.say(.{ .resize = .{ .cols = 23, .rows = 9 } });
// An old daemon's ordinary missing-session refusal stays a refusal;
// queued resize cannot resurrect the name between list and reconnect.
try proto.writeFrame(stream.handle, .exit_status, &.{1});
_ = try testPhase(pump, .refused);
var fds = [_]std.posix.pollfd{.{ .fd = stream.handle, .events = std.posix.POLL.IN, .revents = 0 }};
_ = try std.posix.poll(&fds, 100);
var bytes: [8]u8 = undefined;
try std.testing.expectEqual(@as(usize, 0), try std.posix.read(stream.handle, &bytes));
}
fn testEnd(pump: *Pump, phase: EndPhase) !EndState {
const until = std.time.milliTimestamp() + 2500;
while (std.time.milliTimestamp() < until) {
const result = pump.state().ending;
if (result.phase == phase) return result;
std.Thread.sleep(std.time.ns_per_ms);
}
return error.EndResultTimeout;
}
test "pump End uses attached link refuses shared sessions and forces only a second explicit request" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try peer.start();
defer pump.stop();
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
defer attach_frame.deinit(std.testing.allocator);
try std.testing.expectError(error.NotAttached, pump.say(.{ .end = .{ .request = 1 } }));
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
_ = try testPhase(pump, .attached);
try pump.say(.{ .end = .{ .request = 2 } });
const request = try testFrame(stream, .end_req);
defer request.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u8, 0), request.payload[0]);
try std.testing.expectEqualStrings("native-test", request.payload[1..]);
try std.testing.expectError(error.EndPending, pump.say(.{ .end = .{ .request = 3, .force = true } }));
var buf: [proto.end_reply_max_len]u8 = undefined;
try proto.writeFrame(stream.handle, .end_reply, proto.encodeEndReply(&buf, false, 1, proto.end_reason.others_attached));
const refusal = try testEnd(pump, .refused);
try std.testing.expectEqual(@as(u64, 2), refusal.request);
try std.testing.expectEqual(@as(u8, 1), refusal.others);
try std.testing.expectEqual(Phase.attached, pump.state().phase);
try pump.say(.{ .end = .{ .request = 4, .force = true } });
const force = try testFrame(stream, .end_req);
defer force.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(u8, 1), force.payload[0]);
try proto.writeFrame(stream.handle, .end_reply, proto.encodeEndReply(&buf, true, 1, "ended"));
_ = try testEnd(pump, .accepted);
try proto.writeFrame(stream.handle, .exit_status, &.{0});
_ = try testPhase(pump, .exited);
try std.testing.expectEqual(EndPhase.accepted, pump.state().ending.phase);
}
test "unacknowledged End times out without retry and late replies cannot authorize another End" {
var peer = try TestPeer.init();
defer peer.deinit();
const pump = try Pump.start(std.testing.allocator, .{ .target = .{ .sock = peer.path }, .session = "native-test", .cols = 11, .rows = 3, .end_timeout_ms = 30 });
defer pump.stop();
const stream = try peer.accept();
defer stream.close();
const attach_frame = try testFrame(stream, .attach);
defer attach_frame.deinit(std.testing.allocator);
try proto.writeFrame(stream.handle, .snapshot, &testSnapshot());
_ = try testPhase(pump, .attached);
const started = std.time.milliTimestamp();
try pump.say(.{ .end = .{ .request = 1 } });
const request = try testFrame(stream, .end_req);
defer request.deinit(std.testing.allocator);
_ = try testEnd(pump, .unknown);
try std.testing.expect(std.time.milliTimestamp() - started < 500);
var buf: [proto.end_reply_max_len]u8 = undefined;
try proto.writeFrame(stream.handle, .end_reply, proto.encodeEndReply(&buf, false, 1, "late refusal"));
try pump.say(.{ .input = "still usable" });
const input = try testFrame(stream, .input);
defer input.deinit(std.testing.allocator);
try std.testing.expectError(error.EndOutcomeUnknown, pump.say(.{ .end = .{ .request = 2 } }));
try proto.writeFrame(stream.handle, .exit_status, &.{7});
_ = try testPhase(pump, .exited);
try std.testing.expectEqual(EndPhase.unknown, pump.state().ending.phase);
}
test "restored SSH retries an initial announce EOF and joins when the host answers" {
const a = std.testing.allocator;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const path = try tmp.dir.realpathAlloc(a, ".");
defer a.free(path);
var bytes: std.ArrayList(u8) = .empty;
defer bytes.deinit(a);
try proto.appendFrame(&bytes, a, .snapshot, &testSnapshot());
try tmp.dir.writeFile(.{ .sub_path = "snapshot", .data = bytes.items });
const script = try std.fmt.allocPrint(a, "if test ! -e {s}/once; then touch {s}/once; printf 'connection refused\\n' >&2; exit 1; fi; printf 'endpoint none\\n'; cat {s}/snapshot -", .{ path, path, path });
defer a.free(script);
const pump = try Pump.start(a, .{ .target = .{ .hand = .{ .host = "fixture", .ssh_argv = &.{ "/bin/sh", "-c", script }, .cache_path = null, .deadline_ms = 200 } }, .session = "native-test", .cols = 11, .rows = 3, .existing_only = true, .retry_initial = true });
defer pump.stop();
_ = try testPhase(pump, .attached);
pump.mu.lock();
defer pump.mu.unlock();
try std.testing.expect(pump.snapshot_ready);
try std.testing.expectEqual(@as(u64, 37), pump.replica.last_seq);
}
test "restored SSH authentication refusal is terminal and does not retry" {
const a = std.testing.allocator;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const path = try tmp.dir.realpathAlloc(a, ".");
defer a.free(path);
const marker = try std.fmt.allocPrint(a, "{s}/runs", .{path});
defer a.free(marker);
const script = try std.fmt.allocPrint(a, "printf x >> {s}; printf 'Permission denied (publickey).\\n' >&2; exit 255", .{marker});
defer a.free(script);
const pump = try Pump.start(a, .{ .target = .{ .hand = .{ .host = "fixture", .ssh_argv = &.{ "/bin/sh", "-c", script }, .cache_path = null, .deadline_ms = 200 } }, .session = "native-test", .cols = 11, .rows = 3, .existing_only = true, .retry_initial = true, .open_timeout_ms = 1000 });
defer pump.stop();
const state = try testPhase(pump, .dial_failed);
try std.testing.expectEqualStrings("mux: authentication refused by fixture over ssh: Permission denied (publickey).\n", state.reasonText());
const runs = try tmp.dir.readFileAlloc(a, "runs", 16);
defer a.free(runs);
try std.testing.expectEqual(@as(usize, 1), runs.len);
}
test "reconnecting SSH authentication refusal is terminal after an established pipe" {
const a = std.testing.allocator;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const path = try tmp.dir.realpathAlloc(a, ".");
defer a.free(path);
var bytes: std.ArrayList(u8) = .empty;
defer bytes.deinit(a);
try proto.appendFrame(&bytes, a, .snapshot, &testSnapshot());
try tmp.dir.writeFile(.{ .sub_path = "snapshot", .data = bytes.items });
const marker = try std.fmt.allocPrint(a, "{s}/runs", .{path});
defer a.free(marker);
const once = try std.fmt.allocPrint(a, "{s}/once", .{path});
defer a.free(once);
const release = try std.fmt.allocPrint(a, "{s}/release", .{path});
defer a.free(release);
const script = try std.fmt.allocPrint(a, "printf x >> {s}; if test ! -e {s}; then touch {s}; printf 'endpoint none\\n'; cat {s}/snapshot; i=0; while test ! -e {s} && test $i -lt 200; do i=$((i+1)); sleep 0.01; done; else printf 'Permission denied (publickey).\\n' >&2; exit 255; fi", .{ marker, once, once, path, release });
defer a.free(script);
const pump = try Pump.start(a, .{ .target = .{ .hand = .{ .host = "fixture", .ssh_argv = &.{ "/bin/sh", "-c", script }, .cache_path = null, .deadline_ms = 200 } }, .session = "native-test", .cols = 11, .rows = 3, .existing_only = true, .retry_initial = true, .open_timeout_ms = 1000 });
defer pump.stop();
_ = try testPhase(pump, .attached);
try tmp.dir.writeFile(.{ .sub_path = "release", .data = "done" });
const state = try testPhase(pump, .dial_failed);
try std.testing.expectEqualStrings("mux: authentication refused by fixture over ssh: Permission denied (publickey).\n", state.reasonText());
const runs = try tmp.dir.readFileAlloc(a, "runs", 16);
defer a.free(runs);
try std.testing.expectEqual(@as(usize, 2), runs.len);
}
// Exercise the production mailbox admission and frame decoder synchronously;
// these tests own actual pipes/grid memory but need no transport peer or GUI.
fn selectionTestPump() !*Pump {
const a = std.testing.allocator;
const g = try term.grid.Grid.init(a, 11, 3);
errdefer g.deinit();
const wp = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
errdefer closePipe(wp);
const cp = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
errdefer closePipe(cp);
const p = try a.create(Pump);
p.* = .{ .alloc = a, .opts = .{ .target = .{ .sock = "unused" }, .cols = 11, .rows = 3 }, .grid = g, .replica = .init(a, g), .wake_pipe = wp, .cancel_pipe = cp };
_ = try p.onFrame(.snapshot, &testSnapshot());
// The unified native path requires a nonzero follow source. Tests model
// the metadata frame that the daemon sends after the initial snapshot.
p.follow_seq = p.replica.last_seq;
p.follow_source = 1;
return p;
}
fn selectionTestQueue(p: *Pump, id: u32) !SelectionRequest {
p.mu.lock();
const version = p.selectionVersionLocked();
p.mu.unlock();
try p.say(.{ .selection = .{ .id = id, .anchor = .{ .row = 0, .col = 0 }, .active = .{ .row = 1, .col = 3 }, .version = version } });
return p.mailbox.items[p.mailbox.items.len - 1].selection;
}
fn selectionTestBegin(p: *Pump, req: SelectionRequest) ?[proto.selection_req_len]u8 {
p.mu.lock();
defer p.mu.unlock();
return p.beginSelectionLocked(req);
}
fn selectionTestReply(p: *Pump, id: u32, text: []const u8) !void {
var payload: std.ArrayList(u8) = .empty;
defer payload.deinit(std.testing.allocator);
p.mu.lock();
const history_rows = p.replica.history_rows;
const gesture = if (p.selection_pending) |pending| if (pending.gesture == 0) pending.id else pending.gesture else id;
p.mu.unlock();
try proto.encodeSelectionReply(&payload, std.testing.allocator, .{ .id = id, .gesture = gesture, .seq = p.replica.last_seq, .source = p.follow_source, .history_rows = history_rows, .status = .ok, .text = text });
_ = try p.onFrame(.selection_reply, payload.items);
@memset(payload.items, 0xaa); // The published result must own its text.
}
fn selectionTestFollowMetadata(p: *Pump, source: u64) !void {
var payload: std.ArrayList(u8) = .empty;
defer payload.deinit(std.testing.allocator);
try proto.encodeSelectionReply(&payload, std.testing.allocator, .{
.id = 0,
.gesture = 0,
.seq = p.replica.last_seq,
.source = source,
.history_rows = p.replica.history_rows,
.status = .ok,
.text = "",
});
_ = try p.onFrame(.selection_reply, payload.items);
}
test "follow selection sends a source-bound start then a gesture-bound copy" {
const p = try selectionTestPump();
defer p.stop();
p.admitted = true;
p.follow_seq = p.replica.last_seq;
p.follow_source = 91;
const version = p.selectionVersionLocked();
const first: SelectionRequest = .{
.id = 7,
.gesture = 42,
.anchor = .{ .row = 0, .col = 0 },
.active = .{ .row = 1, .col = 3 },
.version = version,
.ticket = p.selection_ticket,
};
const start = p.beginSelectionLocked(first).?;
const start_req = try proto.decodeSelectionReq(&start);
try std.testing.expectEqual(.start, start_req.action);
try std.testing.expectEqual(@as(u32, 7), start_req.id);
try std.testing.expectEqual(@as(u32, 42), start_req.gesture);
try std.testing.expectEqual(@as(u64, 91), start_req.source);
const second = SelectionRequest{ .id = 8, .gesture = 42, .anchor = first.anchor, .active = first.active, .version = version, .ticket = p.selection_ticket };
const copy = p.beginSelectionLocked(second).?;
const copy_req = try proto.decodeSelectionReq(©);
try std.testing.expectEqual(.copy, copy_req.action);
try std.testing.expectEqual(@as(u32, 8), copy_req.id);
try std.testing.expectEqual(@as(u32, 42), copy_req.gesture);
}
test "selection cancels queued work but retains live requests through redraw" {
const p = try selectionTestPump();
defer p.stop();
const cancelled = try selectionTestQueue(p, 1);
p.cancelSelection();
try std.testing.expect(selectionTestBegin(p, cancelled) == null);
const before_redraw = try selectionTestQueue(p, 2);
var newer = testSnapshot();
std.mem.writeInt(u64, newer[0..8], 38, .little);
_ = try p.onFrame(.snapshot, &newer);
p.follow_seq = p.replica.last_seq;
p.follow_source = 1;
try std.testing.expect(selectionTestBegin(p, before_redraw) != null);
const current = try selectionTestQueue(p, 3);
const bytes = selectionTestBegin(p, current).?;
const wire = try proto.decodeSelectionReq(&bytes);
try std.testing.expectEqual(@as(u32, 3), wire.id);
try std.testing.expectEqualDeep(current.active, wire.active);
}
test "selection results own UTF-8, newest request wins, and redraw preserves a decoded result" {
const p = try selectionTestPump();
defer p.stop();
_ = selectionTestBegin(p, try selectionTestQueue(p, 1)).?;
_ = selectionTestBegin(p, try selectionTestQueue(p, 2)).?;
try selectionTestReply(p, 1, "old");
try std.testing.expect(p.takeSelection() == null);
const text = try std.testing.allocator.alloc(u8, 70 * 1024);
defer std.testing.allocator.free(text);
@memset(text, 'x'); // Native copy does not inherit the OSC 52 base64 cap.
try selectionTestReply(p, 2, text);
const copied = p.takeSelection().?;
defer std.testing.allocator.free(copied.text);
try std.testing.expectEqualStrings(text, copied.text);
_ = selectionTestBegin(p, try selectionTestQueue(p, 3)).?;
try selectionTestReply(p, 3, "café");
var redraw = testSnapshot();
std.mem.writeInt(u64, redraw[0..8], 38, .little);
_ = try p.onFrame(.snapshot, &redraw);
const retained = p.takeSelection().?;
defer std.testing.allocator.free(retained.text);
try std.testing.expectEqualStrings("café", retained.text);
}
test "live selection survives pending redraw; history and geometry remain guarded" {
const p = try selectionTestPump();
defer p.stop();
const req = try selectionTestQueue(p, 1);
_ = selectionTestBegin(p, req).?;
var newer = testSnapshot();
std.mem.writeInt(u64, newer[0..8], 38, .little);
_ = try p.onFrame(.snapshot, &newer);
p.follow_seq = p.replica.last_seq;
p.follow_source = 1;
try std.testing.expect(p.selectionFresh(req.version));
try selectionTestReply(p, 1, "current text");
const result = p.takeSelection().?;
defer std.testing.allocator.free(result.text);
try std.testing.expectEqualStrings("current text", result.text);
try std.testing.expectEqualDeep(req.version, result.version);
const before_history = try selectionTestQueue(p, 2);
proto.writeSnapshotPrefix(newer[0..proto.snapshot_prefix_len], .{ .seq = 39, .history_rows = 1, .cols = 11, .rows = 3, .epoch = 93 });
_ = try p.onFrame(.snapshot, &newer);
try std.testing.expect(!p.selectionFresh(before_history.version));
try std.testing.expect(selectionTestBegin(p, before_history) == null);
const before_resize = try selectionTestQueue(p, 3);
// Empty fixture rows also decode at 12 cols.
proto.writeSnapshotPrefix(newer[0..proto.snapshot_prefix_len], .{ .seq = 40, .history_rows = 1, .cols = 12, .rows = 3, .epoch = 93 });
_ = try p.onFrame(.snapshot, &newer);
try std.testing.expect(!p.selectionFresh(before_resize.version));
try std.testing.expect(selectionTestBegin(p, before_resize) == null);
}
test "selection mode changes, cancellation and timeout discard later replies" {
const p = try selectionTestPump();
defer p.stop();
const req = try selectionTestQueue(p, 1);
_ = selectionTestBegin(p, req).?;
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .alt_screen = true }));
try std.testing.expect(!p.selectionFresh(req.version));
try selectionTestReply(p, 1, "wrong screen");
try std.testing.expect(p.takeSelection() == null);
_ = selectionTestBegin(p, try selectionTestQueue(p, 2)).?;
p.cancelSelection();
try selectionTestReply(p, 2, "cleared");
try std.testing.expect(p.takeSelection() == null);
_ = selectionTestBegin(p, try selectionTestQueue(p, 3)).?;
p.mu.lock();
p.selection_until = 0;
p.mu.unlock();
p.expireSelection();
const expired = p.takeSelection().?;
defer std.testing.allocator.free(expired.text);
try std.testing.expectEqual(proto.SelectionStatus.unavailable, expired.status);
try selectionTestReply(p, 3, "too late");
try std.testing.expect(p.takeSelection() == null);
try std.testing.expect(p.core.pending_selection_id == null);
}
test "wheel history tombstones, refresh, resize and timeout preserve the live replica" {
const p = try selectionTestPump();
defer p.stop();
p.replica.history_rows = 20;
p.admitted = true;
const incoming = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
defer closePipe(incoming);
const outgoing = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
defer closePipe(outgoing);
var tr: client.Transport = .{ .link = .{ .pipe = .{ .child = std.process.Child.init(&.{"unused"}, std.testing.allocator), .r = incoming[0], .w = outgoing[1] } } };
var wire = try Wire.init(std.testing.allocator, &tr);
defer wire.deinit();
const up: Wheel = .{ .notches = 1, .col = 2, .row = 1, .pixel_x = 22, .pixel_y = 18 };
const source_version = p.selectionVersionLocked();
try p.routeWheel(&wire, up);
try std.testing.expect(p.selectionFresh(source_version));
try p.requestHistory(&wire);
const first = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer first.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.MsgType.fetch_scrollback, first.type);
try std.testing.expectEqualDeep(proto.ScrollbackReq{ .start = 17, .count = 3 }, try proto.decodeScrollbackReq(first.payload));
const revision = p.history_pending.?.revision;
try p.say(.{ .input = "live" });
try p.mail(&wire, true);
const typed = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer typed.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("live", typed.payload);
try std.testing.expect(!p.selectionFresh(source_version));
try std.testing.expect(p.history_pending != null); // Keep the cancelled on-wire request.
const after_input = p.selectionVersionLocked();
try p.routeWheel(&wire, up);
try p.requestHistory(&wire);
try std.testing.expectEqual(revision, p.history_pending.?.revision);
const chunk = [_]u8{ 17, 0, 0, 0, 1, 0, 0, 0 }; // one valid blank CellRow
try std.testing.expectEqual(Pump.Action.skip, try p.onFrame(.scrollback_chunk, &chunk));
try std.testing.expect(p.history == null);
try p.requestHistory(&wire);
const fresh = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer fresh.deinit(std.testing.allocator);
try std.testing.expect(p.history_pending.?.revision != revision);
try std.testing.expectEqual(Pump.Action.changed, try p.onFrame(.scrollback_chunk, &chunk));
try selectionTestFollowMetadata(p, 1);
try std.testing.expectEqual(@as(u32, 17), p.viewOriginLocked());
try std.testing.expect(p.viewGridLocked() != p.grid);
try std.testing.expectEqual(@as(u16, 3), p.viewGridLocked().cursor.y);
try std.testing.expect(p.selectionFresh(after_input));
const history_version = p.selectionVersionLocked();
_ = selectionTestBegin(p, try selectionTestQueue(p, 41)).?;
p.mu.lock();
p.returnLiveLocked();
p.mu.unlock();
try selectionTestReply(p, 41, "kept through viewport");
const kept = p.takeSelection().?;
defer std.testing.allocator.free(kept.text);
try std.testing.expectEqualStrings("kept through viewport", kept.text);
try std.testing.expectEqualDeep(history_version, kept.version);
try p.routeWheel(&wire, up);
try p.requestHistory(&wire);
const restored = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer restored.deinit(std.testing.allocator);
_ = try p.onFrame(.scrollback_chunk, &chunk);
try selectionTestFollowMetadata(p, 1);
try std.testing.expect(p.selectionFresh(history_version));
const history_request = try selectionTestQueue(p, 42);
try std.testing.expect(history_request.version.history);
_ = selectionTestBegin(p, history_request).?;
p.mu.lock();
p.returnLiveLocked();
p.mu.unlock();
var live_redraw = testSnapshot();
proto.writeSnapshotPrefix(live_redraw[0..proto.snapshot_prefix_len], .{ .seq = 38, .history_rows = 20, .cols = 11, .rows = 3, .epoch = 93 });
_ = try p.onFrame(.snapshot, &live_redraw);
try std.testing.expect(!p.selectionFresh(history_request.version));
try selectionTestReply(p, 42, "stale history");
try std.testing.expect(p.takeSelection() == null);
try p.routeWheel(&wire, up);
try p.requestHistory(&wire);
const after_live = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer after_live.deinit(std.testing.allocator);
_ = try p.onFrame(.scrollback_chunk, &chunk);
try selectionTestFollowMetadata(p, 1);
var newer = testSnapshot();
proto.writeSnapshotPrefix(newer[0..proto.snapshot_prefix_len], .{ .seq = 39, .history_rows = 20, .cols = 11, .rows = 3, .epoch = 93 });
_ = try p.onFrame(.snapshot, &newer);
try std.testing.expect(p.history_dirty);
try std.testing.expect(!p.selectionFresh(p.selectionVersionLocked()));
try p.requestHistory(&wire);
const refresh = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer refresh.deinit(std.testing.allocator);
_ = try p.onFrame(.scrollback_chunk, &chunk);
try selectionTestFollowMetadata(p, 1);
try std.testing.expect(p.selectionFresh(p.selectionVersionLocked()));
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .cursor_keys = true }));
try std.testing.expect(p.history != null and p.history_dirty);
try p.mail(&wire, true);
const cleared = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer cleared.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.MsgType.selection_req, cleared.type);
try p.requestHistory(&wire);
const before_resize = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer before_resize.deinit(std.testing.allocator);
try p.say(.{ .resize = .{ .cols = 12, .rows = 4 } });
try p.mail(&wire, true);
const resized = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer resized.deinit(std.testing.allocator);
try std.testing.expectEqual(proto.MsgType.resize, resized.type);
try std.testing.expect(p.history_pending != null and p.history == null);
_ = try p.onFrame(.scrollback_chunk, &chunk);
try std.testing.expect(p.history == null);
try std.testing.expectEqual(@as(u64, 39), p.replica.last_seq);
p.scroll_rows = 3;
p.history_pending = .{ .start = 17, .size = .{ .cols = 11, .rows = 3 }, .revision = p.history_revision, .until = 0 };
try std.testing.expectError(error.ConnectionTimedOut, p.requestHistory(&wire));
try std.testing.expectError(error.BadPayload, p.onFrame(.scrollback_chunk, &.{ 17, 0, 0, 0, 4, 0 }));
try std.testing.expectEqual(@as(u64, 39), p.replica.last_seq);
p.history_pending = .{ .start = 17, .size = .{ .cols = 11, .rows = 3 }, .revision = p.history_revision, .until = 0 };
try p.attach(&wire, true);
try std.testing.expect(p.history_pending != null);
try std.testing.expectEqual(Pump.Action.skip, try p.onFrame(.scrollback_chunk, &chunk));
p.history_pending = .{ .start = 17, .size = .{ .cols = 11, .rows = 3 }, .revision = p.history_revision, .until = 0 };
try p.attach(&wire, false);
try std.testing.expect(p.history == null and p.history_pending == null and p.scroll_rows == 0);
}
test "selection tracker survives source movement while history source stays paired" {
const p = try selectionTestPump();
defer p.stop();
p.admitted = true;
const start = try selectionTestQueue(p, 7);
try std.testing.expect(p.selectionAlive(7, start.version));
_ = selectionTestBegin(p, start).?;
var position: std.ArrayList(u8) = .empty;
defer position.deinit(std.testing.allocator);
try proto.encodeSelectionReply(&position, std.testing.allocator, .{
.id = 0,
.gesture = 7,
.seq = p.replica.last_seq,
.source = 1,
.history_rows = p.replica.history_rows,
.status = .ok,
.anchor = .{ .row = 0, .col = 0 },
.active = .{ .row = 0, .col = 1 },
.text = "",
});
_ = try p.onFrame(.selection_reply, position.items);
var redraw = testSnapshot();
std.mem.writeInt(u64, redraw[0..8], p.replica.last_seq + 1, .little);
_ = try p.onFrame(.snapshot, &redraw);
try std.testing.expect(p.followPositionLocked() == null);
try std.testing.expect(p.selectionAlive(7, start.version));
position.clearRetainingCapacity();
try proto.encodeSelectionReply(&position, std.testing.allocator, .{
.id = 0,
.gesture = 7,
.seq = p.replica.last_seq,
.source = 2,
.history_rows = p.replica.history_rows,
.status = .ok,
.anchor = .{ .row = 0, .col = 0 },
.active = .{ .row = 0, .col = 1 },
.text = "",
});
_ = try p.onFrame(.selection_reply, position.items);
try std.testing.expect(p.followPositionLocked() != null);
const copy = SelectionRequest{ .id = 8, .gesture = 7, .anchor = start.anchor, .active = start.active, .version = start.version, .ticket = p.selection_ticket };
try std.testing.expect(selectionTestBegin(p, copy) != null);
try selectionTestReply(p, 8, "moved source");
const result = p.takeSelection().?;
defer std.testing.allocator.free(result.text);
try std.testing.expectEqualStrings("moved source", result.text);
const stale_start = SelectionRequest{ .id = 9, .gesture = 9, .anchor = start.anchor, .active = start.active, .version = start.version, .ticket = p.selection_ticket };
try std.testing.expect(selectionTestBegin(p, stale_start) == null);
const h = try selectionTestPump();
defer h.stop();
h.admitted = true;
h.replica.history_rows = 20;
h.scroll_rows = 1;
h.history_pending = .{ .start = 17, .size = .{ .cols = 11, .rows = 3 }, .revision = h.history_revision, .until = std.math.maxInt(i64) };
const chunk = [_]u8{ 17, 0, 0, 0, 1, 0, 0, 0 };
try std.testing.expectEqual(Pump.Action.changed, try h.onFrame(.scrollback_chunk, &chunk));
try std.testing.expectEqual(@as(u64, 0), h.history_version.source);
try selectionTestFollowMetadata(h, 3);
try std.testing.expectEqual(@as(u64, 3), h.history_version.source);
var history_redraw = testSnapshot();
std.mem.writeInt(u64, history_redraw[0..8], h.replica.last_seq + 1, .little);
_ = try h.onFrame(.snapshot, &history_redraw);
try selectionTestFollowMetadata(h, 4);
try std.testing.expectEqual(@as(u64, 3), h.history_version.source);
}
test "ready terminal mode frames precede queued wheel input through the actual wire" {
const p = try selectionTestPump();
defer p.stop();
p.admitted = true;
const incoming = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
defer closePipe(incoming);
const outgoing = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
defer closePipe(outgoing);
var tr: client.Transport = .{ .link = .{ .pipe = .{ .child = std.process.Child.init(&.{"unused"}, std.testing.allocator), .r = incoming[0], .w = outgoing[1] } } };
var wire = try Wire.init(std.testing.allocator, &tr);
defer wire.deinit();
const up: Wheel = .{ .notches = 1, .col = 4, .row = 3, .pixel_x = 422, .pixel_y = 318 };
try p.say(.{ .wheel = up });
// More complete frames than one drain budget. Callers must defer the
// mailbox until the final mode frame has been admitted in FIFO order.
for (0..150) |_| try proto.writeFrame(incoming[1], .term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false }));
try proto.writeFrame(incoming[1], .term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .alt_screen = true, .cursor_keys = true }));
try std.testing.expectEqual(Pump.ReadState.busy, try p.receiveReady(&wire));
try p.mail(&wire, false);
try std.testing.expectEqual(@as(usize, 1), p.mailbox.items.len);
try std.testing.expectEqual(Pump.ReadState.busy, try p.receiveReady(&wire));
try std.testing.expectEqual(Pump.ReadState.idle, try p.receiveReady(&wire));
try p.mail(&wire, true);
const arrows = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer arrows.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("\x1bOA\x1bOA\x1bOA", arrows.payload);
try proto.writeFrame(incoming[1], .term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .mouse_normal = true, .mouse_sgr_pixels = true }));
try std.testing.expectEqual(Pump.ReadState.idle, try p.receiveReady(&wire));
try p.say(.{ .wheel = up });
try p.mail(&wire, true);
const pixels = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer pixels.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("\x1b[<64;423;319M", pixels.payload);
}
test "paste samples modes at delivery and keeps one chunked envelope in mailbox order" {
const a = std.testing.allocator;
const p = try selectionTestPump();
defer p.stop();
p.admitted = true;
var pair: [2]std.posix.fd_t = undefined;
try std.testing.expectEqual(@as(c_int, 0), std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
defer std.posix.close(pair[0]);
defer std.posix.close(pair[1]);
var tr: client.Transport = .{ .link = .{ .fd = pair[0] } };
var wire = try Wire.init(a, &tr);
defer wire.deinit();
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = true }));
const payload = try a.alloc(u8, paste_chunk_len + 7);
defer a.free(payload);
@memset(payload, 'p');
@memcpy(payload[paste_chunk_len..], "tail\nxy");
try p.say(.{ .input = "before" });
try p.say(.{ .paste = payload });
@memset(payload, 'x');
try p.say(.{ .input = "after" });
try p.mail(&wire, true);
const before = (try proto.readFrame(a, pair[1])).?;
defer before.deinit(a);
try std.testing.expectEqualStrings("before", before.payload);
const begin = (try proto.readFrame(a, pair[1])).?;
defer begin.deinit(a);
try std.testing.expectEqualStrings(app_input.paste_begin, begin.payload);
const first = (try proto.readFrame(a, pair[1])).?;
defer first.deinit(a);
try std.testing.expectEqual(paste_chunk_len, first.payload.len);
for (first.payload) |byte| try std.testing.expectEqual(@as(u8, 'p'), byte);
const tail = (try proto.readFrame(a, pair[1])).?;
defer tail.deinit(a);
try std.testing.expectEqualStrings("tail\nxy", tail.payload);
const end = (try proto.readFrame(a, pair[1])).?;
defer end.deinit(a);
try std.testing.expectEqualStrings(app_input.paste_end, end.payload);
const after = (try proto.readFrame(a, pair[1])).?;
defer after.deinit(a);
try std.testing.expectEqualStrings("after", after.payload);
try p.say(.{ .paste = "mode sampled when delivered" });
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false }));
try p.mail(&wire, true);
const raw = (try proto.readFrame(a, pair[1])).?;
defer raw.deinit(a);
try std.testing.expectEqualStrings("mode sampled when delivered", raw.payload);
}
test "queued paste is released when a pump stops" {
const p = try selectionTestPump();
try p.say(.{ .paste = "owned until shutdown" });
p.stop();
}
test "mouse press motion release uses negotiated SGR coordinates" {
const p = try selectionTestPump();
defer p.stop();
p.admitted = true;
p.status.phase = .attached;
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .mouse_button = true, .mouse_sgr = true }));
const incoming = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
defer closePipe(incoming);
const outgoing = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
defer closePipe(outgoing);
var tr: client.Transport = .{ .link = .{ .pipe = .{ .child = std.process.Child.init(&.{"unused"}, std.testing.allocator), .r = incoming[0], .w = outgoing[1] } } };
var wire = try Wire.init(std.testing.allocator, &tr);
defer wire.deinit();
const token = p.mouseToken().?;
try p.say(.{ .mouse = .{ .token = token, .kind = .press, .button = 0, .col = 2, .row = 3, .pixel_x = 20, .pixel_y = 30 } });
try p.mail(&wire, true);
var frame = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer frame.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("\x1b[<0;3;4M", frame.payload);
try p.say(.{ .mouse = .{ .token = token, .kind = .motion, .button = 0, .col = 4, .row = 5, .pixel_x = 40, .pixel_y = 50 } });
try p.mail(&wire, true);
var motion = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer motion.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("\x1b[<32;5;6M", motion.payload);
try p.say(.{ .mouse = .{ .token = token, .kind = .release, .button = 0, .col = 4, .row = 5, .pixel_x = 40, .pixel_y = 50 } });
try p.mail(&wire, true);
var release = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer release.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("\x1b[<0;5;6m", release.payload);
// Normal tracking ignores held motion, then a format change cancels in
// the format of the transmitted press, exactly once.
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .mouse_normal = true, .mouse_sgr = true }));
var event: Mouse = .{ .token = p.mouseToken().?, .kind = .press, .col = 2, .row = 3, .pixel_x = 20, .pixel_y = 30 };
try p.routeMouse(&wire, event);
const normal = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer normal.deinit(std.testing.allocator);
event.kind = .motion;
event.col = 5;
try p.routeMouse(&wire, event);
var probe: [1]u8 = undefined;
try std.testing.expectError(error.WouldBlock, std.posix.read(outgoing[0], &probe));
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .mouse_any = true }));
try p.reconcileMouse(&wire);
const cancelled = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer cancelled.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("\x1b[<0;3;4m", cancelled.payload);
event.kind = .release;
try p.routeMouse(&wire, event);
try p.reconcileMouse(&wire);
try std.testing.expectError(error.WouldBlock, std.posix.read(outgoing[0], &probe));
// A cancelled queued press never starts a replacement gesture.
event.token = p.mouseToken().?;
event.kind = .press;
try p.say(.{ .mouse = event });
p.cancelMouse(event.token);
try p.mail(&wire, true);
try std.testing.expectError(error.WouldBlock, std.posix.read(outgoing[0], &probe));
event.token = p.mouseToken().?;
event.kind = .motion;
event.button = 3;
try p.routeMouse(&wire, event);
const hover = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer hover.deinit(std.testing.allocator);
try std.testing.expectEqualSlices(u8, &.{ 27, '[', 'M', 67, 38, 36 }, hover.payload);
_ = try p.onFrame(.term_modes, &proto.encodeTermModes(.{ .bracketed_paste = false, .mouse_x10 = true }));
event.token = p.mouseToken().?;
event.kind = .press;
event.button = 0;
try p.routeMouse(&wire, event);
const x10 = (try proto.readFrame(std.testing.allocator, outgoing[0])).?;
defer x10.deinit(std.testing.allocator);
try std.testing.expectEqualSlices(u8, &.{ 27, '[', 'M', 32, 38, 36 }, x10.payload);
event.kind = .release;
try p.routeMouse(&wire, event);
try std.testing.expectError(error.WouldBlock, std.posix.read(outgoing[0], &probe));
}
test "clipboard targets are retained separately and cleared on state change" {
const p = try selectionTestPump();
defer p.stop();
p.admitted = true;
p.status.phase = .attached;
_ = try p.onFrame(.term_event, "\x00cYQ==");
_ = try p.onFrame(.term_event, "\x00pYg==");
_ = try p.onFrame(.term_event, "\x00cYw==");
_ = try p.onFrame(.term_event, "\x00cAA=="); // NUL must preserve c.
const ctext = p.takeClipboard(false).?;
defer std.testing.allocator.free(ctext);
const ptext = p.takeClipboard(true).?;
defer std.testing.allocator.free(ptext);
try std.testing.expectEqualStrings("c", ctext);
try std.testing.expectEqualStrings("b", ptext);
_ = try p.onFrame(.term_event, "\x00cYQ==");
_ = try p.onFrame(.term_event, "\x00sYg==");
p.mu.lock();
p.setState(.reconnecting, 0, "reconnecting");
p.mu.unlock();
try std.testing.expect(p.takeClipboard(false) == null);
try std.testing.expect(p.takeClipboard(true) == null);
}