src/server/server_test_await.zig
Ref: Size: 43.2 KiB History
const std = @import("std");
const proto = @import("term").protocol;
const server_os = @import("server_os");
const shellint = @import("shellint.zig");
const TmpDir = @import("testtmp").TmpDir;
const h = @import("server_test_harness.zig");
const dial = h.dial;
const srv_mod = @import("server.zig");
const Server = srv_mod.Server;
const attachNamed = h.attachNamed;
const awaitFrame = h.awaitFrame;
const awaitGridText = h.awaitGridText;
const connectedPair = h.connectedPair;
test "Server: OSC 133 marks reach attached clients as cmd_state pushes" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "marks");
defer td.deinit();
// A SCRIPT rather than an interactive shell: an integrated shell emits marks
// continuously, so "which push came from which mark" would be a guess. It
// ends on a blocking read, so `deinit`'s SIGTERM lands on the shell itself.
try td.tmp.dir.writeFile(.{
.sub_path = "marks.sh",
.data =
\\#!/bin/sh
\\read -r start
\\printf '\033]133;C\007'
\\read -r second
\\printf '\033]133;D;0\007'
\\read -r stop
\\
,
.flags = .{ .mode = 0o755 },
});
const script = try std.fmt.allocPrintSentinel(alloc, "{s}/marks.sh", .{td.tmp.path()}, 0);
defer alloc.free(script);
try td.start(.{ .shell = script });
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// Release the script into its C. The keystroke is echoed by the line
// discipline, so this pty chunk carries grid content as well as the
// mark — which is the point: the push has to survive sharing a read
// with ordinary output.
try proto.writeFrame(c.handle, .input, "go\n");
const f1 = (try awaitFrame(alloc, &td.srv, c.handle, .cmd_state, 500)) orelse
return error.NoRunningPush;
defer f1.deinit(alloc);
const running = try proto.decodeCmdState(f1.payload);
try std.testing.expectEqual(proto.CmdPhase.running, running.phase);
// Marks are the only mechanism that ever pushes: a push means a mark was
// read, never that a heuristic guessed.
try std.testing.expectEqual(proto.Mechanism.marks, running.mechanism);
// And into its D, which closes the command with a code.
try proto.writeFrame(c.handle, .input, "go\n");
const f2 = (try awaitFrame(alloc, &td.srv, c.handle, .cmd_state, 500)) orelse
return error.NoReturnedPush;
defer f2.deinit(alloc);
const returned = try proto.decodeCmdState(f2.payload);
try std.testing.expectEqual(proto.CmdPhase.returned, returned.phase);
try std.testing.expectEqual(proto.Mechanism.marks, returned.mechanism);
try std.testing.expectEqual(@as(?u8, 0), returned.exit_code);
// The seq is a last-return WATERMARK, not a stamp on the frame carrying it,
// so the running push of the first command a session runs says 0. A consumer
// reading it as "when this command started" reads about a different event.
try std.testing.expectEqual(@as(u64, 0), running.seq);
// The return moves the watermark to the tracker as sampled AFTER the update
// for its own chunk, so it covers the command's output. Ordering only: a C
// and a D in one read legitimately share a seq, hence `>=`.
try std.testing.expect(returned.seq >= running.seq);
try std.testing.expect(returned.seq <= td.srv.sessions.table[0].?.tracker.seq);
try std.testing.expect(returned.seq > 0);
}
// ---------------------------------------------------------------------------
// Shell integration, end to end. Above this line the daemon can READ marks;
// these two prove a real shell EMITS them. The chain is injection → shell →
// pty → engine → tracker → wire, and no part of it is stubbed.
// ---------------------------------------------------------------------------
/// A session's command phase, as a `pumpUntil` context.
const Phase = struct {
srv: *Server,
si: usize,
want: proto.CmdPhase,
fn reached(self: Phase) bool {
return self.srv.sessions.table[self.si].?.cmd.phase == self.want;
}
};
/// A session by name, for the waits that open with "the daemon has seated
/// this one".
const Named = struct {
srv: *Server,
name: []const u8,
fn exists(self: Named) bool {
return self.srv.sessions.find(self.name) != null;
}
};
/// The absence half of the phantom-mark assertions: run nothing, claim
/// nothing.
fn anyReturnWithin(
alloc: std.mem.Allocator,
srv: *Server,
fd: std.posix.fd_t,
budget_ms: i64,
) !?proto.CmdState {
// The answer comes out of the SINK, because a cmd_state in any other
// phase is not it: `want` would take the first one and call it the
// return. The wall clock stays the budget — the harness's pumping await
// counts rounds, and this one's contract is "inside this many
// milliseconds, nothing came back".
const Returned = struct {
st: proto.CmdState = undefined,
fn on(ctx: ?*anyopaque, frame: proto.Frame) anyerror!void {
if (frame.type != .cmd_state) return;
const self: *@This() = @ptrCast(@alignCast(ctx.?));
const st = try proto.decodeCmdState(frame.payload);
if (st.phase != .returned) return;
self.st = st;
return error.CommandReturned;
}
};
var seen: Returned = .{};
const deadline = std.time.milliTimestamp() + budget_ms;
while (std.time.milliTimestamp() < deadline) {
_ = h.awaitFrameSink(alloc, srv, fd, h.never_from_daemon, 1, .{
.ctx = &seen,
.on = Returned.on,
}) catch |err| switch (err) {
error.CommandReturned => return seen.st,
else => return err,
};
}
return null;
}
/// A session whose HOME is a directory this test wrote, so no verdict depends
/// on whose rc files the box carries. The rc is NOT empty: it adds a
/// PROMPT_COMMAND member, the shape that breaks the bash shim, plus a PS1.
const IntegratedSession = struct {
tmp: TmpDir,
home: [:0]const u8,
sock_path: []const u8,
srv: Server,
conn: std.net.Stream,
const prompt = "MUXPROMPT>";
fn start(alloc: std.mem.Allocator, shell: [:0]const u8, tag: []const u8, rc: []const u8) !*IntegratedSession {
const self = try alloc.create(IntegratedSession);
errdefer alloc.destroy(self);
self.tmp = try TmpDir.make();
errdefer self.tmp.cleanup();
try self.tmp.dir.makePath("home");
var home_dir = try self.tmp.dir.openDir("home", .{});
defer home_dir.close();
try home_dir.writeFile(.{ .sub_path = rcName(shell), .data = rc });
self.home = try std.fmt.allocPrintSentinel(alloc, "{s}/home", .{self.tmp.path()}, 0);
errdefer alloc.free(self.home);
self.sock_path = try std.fmt.allocPrint(alloc, "{s}/{s}.sock", .{ self.tmp.path(), tag });
errdefer alloc.free(self.sock_path);
self.srv = try Server.init(alloc, .{
.sock_path = self.sock_path,
.shell = shell,
// Asked for explicitly: the injection is off by default, and
// every test below this line is about what the injection does.
.shell_integration = true,
.extra_env = &.{
.{ .key = "HOME", .value = self.home },
// Emptied, and not cosmetically: the shim copies the DAEMON's
// ZDOTDIR, and here the daemon is the test runner — so a runner
// with ZDOTDIR set would source THAT .zshrc, past the planted
// HOME. Empty reads as "there was none". `extra_env` wins.
.{ .key = "MUX_ORIG_ZDOTDIR", .value = "" },
},
});
errdefer self.srv.deinit();
// The shim really was written where the daemon says it was. Without
// this, a missing directory would present as "no marks ever arrived"
// and the timeout would never say why.
try std.testing.expect(self.srv.shellint_dir != null);
try std.fs.cwd().access(self.srv.shellint_dir.?, .{});
self.conn = try dial.dialAttach(self.sock_path, 80, 24);
errdefer self.conn.close();
return self;
}
fn rcName(shell: [:0]const u8) []const u8 {
return if (shellint.detect(shell) == .zsh) ".zshrc" else ".bashrc";
}
fn deinit(self: *IntegratedSession, alloc: std.mem.Allocator) void {
self.conn.close();
self.srv.deinit();
alloc.free(self.sock_path);
alloc.free(self.home);
self.tmp.cleanup();
alloc.destroy(self);
}
};
/// Planted explicitly, not read from /etc/bash.bashrc, so the test states its
/// own premise.
const bash_rc_with_prompt_member =
\\PS1='MUXPROMPT>'
\\PROMPT_COMMAND+=(': mux-test-member')
\\
;
test "Server: the injection is off unless the caller asks for it" {
const alloc = std.testing.allocator;
// bash, because shellint HAS scripts for bash: a shell it has none for
// would pass this whichever way the default points, and prove nothing.
std.fs.cwd().access("/bin/bash", .{}) catch return error.SkipZigTest;
var td = try h.TestDaemon.init(alloc, "nomarks", .{ .shell = "/bin/bash" });
defer td.deinit();
// No shim directory at all — the user's ZDOTDIR and DEBUG trap are their
// own. `IntegratedSession` is the other side of this pin: it asks for the
// injection and asserts the directory exists.
try std.testing.expect(td.srv.shellint_dir == null);
}
test "Server: bash emits one mark pair per command, and none at an idle prompt" {
const alloc = std.testing.allocator;
std.fs.cwd().access("/bin/bash", .{}) catch return error.SkipZigTest;
const s = try IntegratedSession.start(alloc, "/bin/bash", "bash", bash_rc_with_prompt_member);
defer s.deinit(alloc);
// Wait for the first prompt, so what follows is an assertion about a
// live shell rather than about a slow one.
try std.testing.expect(try awaitGridText(alloc, &s.srv, IntegratedSession.prompt, 10_000));
// NOTHING has been asked to run, so nothing may claim to have returned. A
// DEBUG trap armed one line before the PROMPT_COMMAND assignment fires ON
// that assignment, and the very first prompt reports `D;0` for a command
// that never existed.
if (try anyReturnWithin(alloc, &s.srv, s.conn.handle, 400)) |ghost| {
std.debug.print("phantom return at the opening prompt: {any}\n", .{ghost});
return error.PhantomReturnBeforeAnyCommand;
}
// Now a real command, with a code no fallback could invent. `false`
// rather than `true`: an exit code of 1 cannot be confused with the 0
// that a phantom D carries.
try proto.writeFrame(s.conn.handle, .input, "false\n");
const st = (try anyReturnWithin(alloc, &s.srv, s.conn.handle, 10_000)) orelse
return error.NoReturnedPush;
try std.testing.expectEqual(proto.Mechanism.marks, st.mechanism);
try std.testing.expectEqual(@as(?u8, 1), st.exit_code);
// A bare Enter runs PROMPT_COMMAND again and no command at all. Its members
// fire the DEBUG trap, so without the membership check an untouched prompt
// reports a successful command on every cycle.
try proto.writeFrame(s.conn.handle, .input, "\n\n\n");
if (try anyReturnWithin(alloc, &s.srv, s.conn.handle, 600)) |ghost| {
std.debug.print("phantom return at an idle prompt: {any}\n", .{ghost});
return error.PhantomReturnAtIdlePrompt;
}
}
/// Returns 3 deliberately: the commands under test exit 0 or 1, so a shim
/// reading the wrong $? would still look right half the time.
const zsh_rc_with_precmd_hook =
\\PS1='MUXPROMPT>'
\\autoload -Uz add-zsh-hook
\\_user_precmd() { return 3 }
\\add-zsh-hook precmd _user_precmd
\\
;
test "Server: zsh emits one mark pair per command, and none at an idle prompt" {
const alloc = std.testing.allocator;
// Both spellings, because the box that has zsh does not always agree
// with the box that had it last.
const shell: [:0]const u8 = blk: {
for ([_][:0]const u8{ "/usr/bin/zsh", "/bin/zsh" }) |p| {
std.fs.cwd().access(p, .{}) catch continue;
break :blk p;
}
return error.SkipZigTest;
};
const s = try IntegratedSession.start(alloc, shell, "zsh", zsh_rc_with_precmd_hook);
defer s.deinit(alloc);
try std.testing.expect(try awaitGridText(alloc, &s.srv, IntegratedSession.prompt, 10_000));
// The guarded D earns its keep here: zsh's precmd runs at the opening
// prompt too, and without the `_mux_ran` guard it would report a return
// before the session had run anything.
if (try anyReturnWithin(alloc, &s.srv, s.conn.handle, 400)) |ghost| {
std.debug.print("phantom return at the opening prompt: {any}\n", .{ghost});
return error.PhantomReturnBeforeAnyCommand;
}
try proto.writeFrame(s.conn.handle, .input, "false\n");
const st = (try anyReturnWithin(alloc, &s.srv, s.conn.handle, 10_000)) orelse
return error.NoReturnedPush;
try std.testing.expectEqual(proto.Mechanism.marks, st.mechanism);
// 1, not 0: mux's precmd reads $? before the user's hook can overwrite
// it, which is the property the shim's hook ordering exists to keep.
try std.testing.expectEqual(@as(?u8, 1), st.exit_code);
try proto.writeFrame(s.conn.handle, .input, "\n\n\n");
if (try anyReturnWithin(alloc, &s.srv, s.conn.handle, 600)) |ghost| {
std.debug.print("phantom return at an idle prompt: {any}\n", .{ghost});
return error.PhantomReturnAtIdlePrompt;
}
}
test "Server: the shim directory is private, and teardown takes it with it" {
const alloc = std.testing.allocator;
std.fs.cwd().access("/bin/bash", .{}) catch return error.SkipZigTest;
var td = try h.TestDaemon.open(alloc, "shim");
defer td.deinit();
// Copied out of the server's arena before deinit frees it: the whole
// point of this test is to ask a question after the server is gone.
var dir_buf: [256]u8 = undefined;
var dir: []const u8 = undefined;
{
try td.start(.{
.shell = "/bin/bash",
.shell_integration = true,
});
defer td.shutdown();
dir = try std.fmt.bufPrint(&dir_buf, "{s}", .{td.srv.shellint_dir.?});
// Beside the socket, not somewhere world-readable: the file the
// session shell is about to source is a file that runs code as this
// user, so 0700 on the directory is part of the contract.
try std.testing.expectEqualStrings(std.fs.path.dirname(td.sock_path).?, std.fs.path.dirname(dir).?);
var d = try std.fs.cwd().openDir(dir, .{ .iterate = true });
defer d.close();
const st = try d.stat();
try std.testing.expectEqual(@as(u32, 0o700), @as(u32, @intCast(st.mode & 0o777)));
}
// A daemon that left its shims behind would litter the runtime directory
// once per session, and nothing else in the system would ever notice.
try std.testing.expectError(error.FileNotFound, std.fs.cwd().access(dir, .{}));
}
test "Server: an entry already at the daemon's pid name costs neither the marks nor itself" {
const alloc = std.testing.allocator;
std.fs.cwd().access("/bin/bash", .{}) catch return error.SkipZigTest;
var td = try h.TestDaemon.open(alloc, "degrade");
defer td.deinit();
// Plant a regular FILE at the pid-derived name, which is what a daemon
// SIGKILLed before teardown leaves behind and what a stranger with write
// access to a shared /tmp would aim at. In a test the daemon IS this
// process, so the pid half of the name is exactly the one init draws.
const planted = try std.fmt.allocPrint(
alloc,
"{s}/mux-shellint-{d}",
.{ td.tmp.path(), server_os.getpid() },
);
defer alloc.free(planted);
try std.fs.cwd().writeFile(.{ .sub_path = planted, .data = "not a directory" });
// Scoped so deinit runs before the survival check below.
{
try td.start(.{
.shell = "/bin/bash",
.shell_integration = true,
});
defer td.shutdown();
// The shim name carries a random suffix past the pid, so the planted
// entry is not in the way of anything: this session gets its marks.
// Before that suffix existed, a collision here cost the session its
// marks for the lifetime of the daemon.
const dir = td.srv.shellint_dir orelse return error.NoShimDirectory;
try std.testing.expect(std.mem.startsWith(u8, dir, planted));
try std.testing.expect(dir.len > planted.len);
try std.fs.cwd().access(dir, .{});
// ...and the session actually works. A daemon that starts and then
// cannot answer would satisfy every assertion above.
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
try proto.writeFrame(c.handle, .status_req, "");
const f = (try awaitFrame(alloc, &td.srv, c.handle, .status_reply, 400)) orelse
return error.NoStatusReply;
defer f.deinit(alloc);
const st = try proto.decodeStatusReply(f.payload);
try std.testing.expectEqual(@as(u16, 80), st.cols);
}
// The planted file is untouched — not adopted, not written through, and
// not deleted by the teardown of a shim it was never part of. In the
// field that entry belongs to whatever else drew the pid.
const kept = try std.fs.cwd().readFileAlloc(alloc, planted, 64);
defer alloc.free(kept);
try std.testing.expectEqualStrings("not a directory", kept);
}
test "Server: an unknown shell is not injected into at all — no directory, no shim" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "plain");
defer td.deinit();
// Integration is ON, and /bin/sh still gets nothing: this is what keeps
// every other test in this file describing the session it always did.
try td.start(.{ .shell = "/bin/sh" });
try std.testing.expectEqual(@as(?[]const u8, null), td.srv.shellint_dir);
// Nothing was written beside the socket either — the absence is on disk,
// not merely in a field the teardown consults.
var d = try std.fs.cwd().openDir(td.tmp.path(), .{ .iterate = true });
defer d.close();
var it = d.iterate();
while (try it.next()) |entry| {
try std.testing.expect(!std.mem.startsWith(u8, entry.name, "mux-shellint-"));
}
}
// ---------------------------------------------------------------------------
// Awaits: a request the daemon holds open until something answers it. Every
// test below asserts the REASON an await ended, never how long it took —
// latency here is the run loop's granularity crossed with the machine's load.
// The iteration budgets are outer bounds, finite so a regression fails by name
// instead of hanging the suite.
// ---------------------------------------------------------------------------
/// Read frames already sitting on `fd`, deliberately WITHOUT pumping, until one
/// of `want` turns up or the socket goes quiet — the "did the daemon answer
/// inside that one pump" observable. An answer needing another pump reads as
/// absent, which is the distinction the immediate-answer test makes.
///
/// Its own poll rather than `Link.awaitFrame` for both halves of that: it
/// must not pump, and it ends on SILENCE, which the primitive does not
/// report — a poll that timed out inside it looks the same as a deadline.
fn readQueued(alloc: std.mem.Allocator, fd: std.posix.fd_t, want: proto.MsgType) !?proto.Frame {
var guard: usize = 0;
while (guard < 16) : (guard += 1) {
var pfd = [_]std.posix.pollfd{
.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 },
};
if ((std.posix.poll(&pfd, 1) catch 0) == 0) return null;
const frame = (try proto.readFrame(alloc, fd)) orelse return null;
if (frame.type == want) return frame;
frame.deinit(alloc);
}
return null;
}
test "Server: an await is held open, answered by a mark, and re-answered immediately after" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "awaitmark");
defer td.deinit();
// Script-gated, so there is a moment at which "no reply yet" is a claim
// worth making. The BURST is the point of its shape: real integration writes
// `D;code` and the next prompt's `A` together, so both fold into the tracker
// inside one pty read and the phase is back at `at_prompt` before any await
// is examined. Stopping at the D passes an implementation that could only
// answer in the sliver between the two.
try td.tmp.dir.writeFile(.{
.sub_path = "await.sh",
.data =
\\#!/bin/sh
\\read -r go
\\printf '\033]133;C\007out\r\n\033]133;D;3\007\033]133;A\007'
\\read -r stop
\\
,
.flags = .{ .mode = 0o755 },
});
const script = try std.fmt.allocPrintSentinel(alloc, "{s}/await.sh", .{td.tmp.path()}, 0);
defer alloc.free(script);
try td.start(.{ .shell = script });
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// since_seq is "what I already know about": only a return NEWER than this
// may answer. Nothing has returned on this session at all.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = td.srv.sessions.table[0].?.tracker.seq,
.settle_ms = 0,
.timeout_ms = 5000,
}));
// Held open: no mark has landed, no settle floor was asked for, the
// timeout is seconds away, and this shell never moves its fg pgid. There
// is nothing that may honestly answer yet, so nothing must.
if (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 40)) |early| {
early.deinit(alloc);
return error.AwaitAnsweredBeforeAnythingHappened;
}
// Release the script into its C-output-D. All three land in one pty read,
// so the await resolves in the same pump that pushed the transitions.
try proto.writeFrame(c.handle, .input, "go\n");
const f = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 500)) orelse
return error.NoAwaitReplyFromMark;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.returned, rep.reason);
try std.testing.expectEqual(proto.CmdPhase.returned, rep.state.phase);
// Marks won the race, and only marks carry a code.
try std.testing.expectEqual(proto.Mechanism.marks, rep.state.mechanism);
try std.testing.expectEqual(@as(?u8, 3), rep.state.exit_code);
// Reconnect idempotency: an agent whose answer died with its connection
// re-asks with the seq it last held, and that return has already happened —
// so the daemon answers INLINE, in the very pump that read the request.
// Reading without pumping again is what makes "inline" the assertion.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = 0,
.settle_ms = 0,
.timeout_ms = 5000,
}));
try td.srv.pumpOnce(5);
const f2 = (try readQueued(alloc, c.handle, .await_reply)) orelse
return error.AwaitNotAnsweredOnTheSamePump;
defer f2.deinit(alloc);
const rep2 = try proto.decodeAwaitReply(f2.payload);
try std.testing.expectEqual(proto.AwaitReason.returned, rep2.reason);
try std.testing.expectEqual(proto.Mechanism.marks, rep2.state.mechanism);
try std.testing.expectEqual(@as(?u8, 3), rep2.state.exit_code);
// A client attaching only now is told the state it could not have witnessed,
// rather than learning nothing until the next transition. The EXIT CODE
// carries the news: an untouched tracker reports null, so 3 means this push
// came from the command that ran.
const late = try dial.dialAttach(td.sock_path, 80, 24);
defer late.close();
const f3 = (try awaitFrame(alloc, &td.srv, late.handle, .cmd_state, 200)) orelse
return error.NoCmdStateOnAttach;
defer f3.deinit(alloc);
const st3 = try proto.decodeCmdState(f3.payload);
try std.testing.expectEqual(proto.CmdPhase.at_prompt, st3.phase);
try std.testing.expectEqual(@as(?u8, 3), st3.exit_code);
}
test "Server: a return is still answerable once the next command is running" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "awaitnext");
defer td.deinit();
// One command that finishes with code 3, then a second that stays running.
// The LIVE tracker is describing the second by the time the await is asked,
// so anything answered out of it answers about the wrong command.
try td.tmp.dir.writeFile(.{
.sub_path = "next.sh",
.data =
\\#!/bin/sh
\\read -r first
\\printf '\033]133;C\007out\r\n\033]133;D;3\007\033]133;A\007'
\\read -r second
\\printf '\033]133;C\007working\r\n'
\\read -r stop
\\
,
.flags = .{ .mode = 0o755 },
});
const script = try std.fmt.allocPrintSentinel(alloc, "{s}/next.sh", .{td.tmp.path()}, 0);
defer alloc.free(script);
try td.start(.{ .shell = script });
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// Run the first command to completion, then start the second and wait
// until the daemon has actually seen it open.
try proto.writeFrame(c.handle, .input, "go\n");
try proto.writeFrame(c.handle, .input, "go\n");
try std.testing.expect(try h.pumpUntil(&td.srv, 3000, Phase{ .srv = &td.srv, .si = 0, .want = .running }, Phase.reached));
// Now ask about everything since the beginning of time. The honest answer
// is the FIRST command's return — the client asked what had returned
// since its seq, not what the session is doing at this instant.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = 0,
.settle_ms = 0,
.timeout_ms = 5000,
}));
const f = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 200)) orelse
return error.NoAwaitReplyWhileNextCommandRuns;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.returned, rep.reason);
try std.testing.expectEqual(proto.Mechanism.marks, rep.state.mechanism);
try std.testing.expectEqual(proto.CmdPhase.returned, rep.state.phase);
// The snapshot's code, not the running command's absent one.
try std.testing.expectEqual(@as(?u8, 3), rep.state.exit_code);
// And the live tracker really has moved on, so the assertions above came
// from the snapshot and could not have come from reading it.
try std.testing.expectEqual(proto.CmdPhase.running, td.srv.sessions.table[0].?.cmd.phase);
try std.testing.expectEqual(@as(?u8, null), td.srv.sessions.table[0].?.cmd.exit_code);
}
test "Server: an await with a settle floor is answered by output going quiet" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "settle");
defer td.deinit();
// /bin/cat echoes once and then says nothing, which is the exact shape
// settle exists for: no shell integration, no job leaving the shell's
// process group, and so no evidence available beyond "it stopped
// talking".
try td.start(.{ .shell = "/bin/cat" });
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = td.srv.sessions.table[0].?.tracker.seq,
.settle_ms = 200,
.timeout_ms = 5000,
}));
try proto.writeFrame(c.handle, .input, "quiet\r\n");
// Budget ~1.2s of wall clock against a 200ms floor and a 5s timeout, so
// the only reason that can legally arrive in the window is the settle.
const f = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 200)) orelse
return error.NoSettleReply;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.settled, rep.reason);
try std.testing.expectEqual(proto.Mechanism.settle, rep.state.mechanism);
// Silence is evidence that something finished, never evidence of how.
try std.testing.expectEqual(@as(?u8, null), rep.state.exit_code);
// Deliberately the live phase, not a manufactured `.returned`: settle
// never learned that a command ran, so it has no standing to claim one
// returned. The reason field carries the verdict; the phase keeps
// reporting what the session is actually known to be doing.
try std.testing.expectEqual(proto.CmdPhase.at_prompt, rep.state.phase);
}
test "Server: an await with nothing to answer it ends at the bound the caller set" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "timeout", .{ .shell = "/bin/cat" });
defer td.deinit();
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// Nothing is typed at this session, no settle floor is asked for and no
// mark will ever come: the timeout is the only thing left that can end
// this wait, which is the point.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = td.srv.sessions.table[0].?.tracker.seq,
.settle_ms = 0,
.timeout_ms = 150,
}));
const f = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 120)) orelse
return error.NoTimeoutReply;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.timeout, rep.reason);
// A timed-out await still reports which regime the session is in, the
// same claim status_reply makes: this one has never spoken marks.
try std.testing.expectEqual(proto.Mechanism.pgid, rep.state.mechanism);
try std.testing.expectEqual(proto.CmdPhase.at_prompt, rep.state.phase);
}
test "Server: a timed-out await carries no exit code, not the last command's" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "timeoutmark");
defer td.deinit();
// The same burst the marks test uses, and for the same reason: `D;5` and
// the next prompt's `A` arrive in one pty read, which is what leaves the
// session at a prompt with a code still standing in the tracker. That
// standing code is the thing this test is about.
try td.tmp.dir.writeFile(.{
.sub_path = "stale.sh",
.data =
\\#!/bin/sh
\\read -r go
\\printf '\033]133;C\007out\r\n\033]133;D;5\007\033]133;A\007'
\\read -r stop
\\
,
.flags = .{ .mode = 0o755 },
});
const script = try std.fmt.allocPrintSentinel(alloc, "{s}/stale.sh", .{td.tmp.path()}, 0);
defer alloc.free(script);
try td.start(.{ .shell = script });
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// First, collect the return honestly, exactly as an agent would.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = td.srv.sessions.table[0].?.tracker.seq,
.settle_ms = 0,
.timeout_ms = 5000,
}));
try proto.writeFrame(c.handle, .input, "go\n");
const f = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 500)) orelse
return error.NoAwaitReplyFromMark;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.returned, rep.reason);
try std.testing.expectEqual(@as(?u8, 5), rep.state.exit_code);
// Wait for the NEXT return with the watermark that reply carried. Nothing
// else will come: the script is blocked in `read`, never leaves the shell's
// process group, and no settle floor was asked for.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = rep.state.seq,
.settle_ms = 0,
.timeout_ms = 150,
}));
const f2 = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 300)) orelse
return error.NoTimeoutReply;
defer f2.deinit(alloc);
const rep2 = try proto.decodeAwaitReply(f2.payload);
try std.testing.expectEqual(proto.AwaitReason.timeout, rep2.reason);
// The regime is still reported — this session has spoken marks, and a
// timeout says so the same way status_reply does.
try std.testing.expectEqual(proto.Mechanism.marks, rep2.state.mechanism);
// But the code is gone. `marks` is the one label the published rule tells
// agents an exit code is real under, so a stale 5 riding out beside it
// would read as a verdict on a command that never returned.
try std.testing.expectEqual(@as(?u8, null), rep2.state.exit_code);
// And the live tracker really does still hold that 5, so the null above
// was cleared by the timeout arm rather than there being nothing to clear.
try std.testing.expectEqual(@as(?u8, 5), td.srv.sessions.table[0].?.cmd.exit_code);
}
test "Server: without shell integration a foreground job's end is caught by the pgid edge" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "pgid");
defer td.deinit();
// A real interactive /bin/sh with NO integration: no mark will ever arrive,
// so the only evidence a command ran is the foreground process group leaving
// the shell and coming back.
try td.start(.{ .shell = "/bin/sh" });
const c = try dial.dialAttach(td.sock_path, 80, 24);
defer c.close();
// sleep 2, not sleep 1: the edge has to SEE the pgid off the shell before
// "back on the shell" can mean anything, so the busy window must be wide
// enough to sample. A job too short to observe is one the settle floor is
// for, not this.
try proto.writeFrame(c.handle, .input, "sleep 2\n");
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = td.srv.sessions.table[0].?.tracker.seq,
.settle_ms = 0,
.timeout_ms = 10_000,
}));
// ~9s of budget for a 2s job: bounded, and slack enough that a loaded box
// does not read as a regression.
const f = (try awaitFrame(alloc, &td.srv, c.handle, .await_reply, 1500)) orelse
return error.NoPgidReply;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.returned, rep.reason);
try std.testing.expectEqual(proto.Mechanism.pgid, rep.state.mechanism);
try std.testing.expectEqual(proto.CmdPhase.returned, rep.state.phase);
// The pgid can say THAT a command ended, never with what code — no mark
// carried one, and inventing a 0 here would be the whole failure mode
// this mechanism has to avoid.
try std.testing.expectEqual(@as(?u8, null), rep.state.exit_code);
}
test "Server: re-attaching to another session drops the await it left behind" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "awaithop", .{ .shell = "/bin/cat" });
defer td.deinit();
const c = try dial.dialAttachNamed(td.sock_path, 80, 24, "a");
defer c.close();
try std.testing.expect(try h.pumpUntil(&td.srv, 2000, Named{ .srv = &td.srv, .name = "a" }, Named.exists));
const si_a = td.srv.sessions.find("a") orelse return error.NoSessionA;
// An await is a question about ONE session's seq series: since_seq is a
// watermark in that session's tracker.
try proto.writeFrame(c.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = 0,
.settle_ms = 0,
.timeout_ms = 60_000,
}));
const Awaiting = struct {
srv: *Server,
si: usize,
fn yes(self: @This()) bool {
return slotAwaiting(self.srv, self.si);
}
};
try std.testing.expect(try h.pumpUntil(&td.srv, 2000, Awaiting{ .srv = &td.srv, .si = si_a }, Awaiting.yes));
// Re-attach the SAME connection to a different session. Carrying the
// watermark across would compare session a's seq against session b's
// last_return — an await that answers instantly or never, arbitrarily.
try attachNamed(c.handle, 80, 24, "b");
try std.testing.expect(try h.pumpUntil(&td.srv, 2000, Named{ .srv = &td.srv, .name = "b" }, Named.exists));
const si_b = td.srv.sessions.find("b") orelse return error.NoSessionB;
try std.testing.expect(si_a != si_b);
var moved = false;
for (td.srv.clients) |slot| {
const cs = slot orelse continue;
if ((cs.session orelse continue) != si_b) continue;
moved = true;
try std.testing.expect(cs.await_state == null);
}
try std.testing.expect(moved);
}
/// The await lives on the CLIENT slot — one client's question — while the
/// watermark it carries belongs to the session, which is why the two must not
/// drift apart.
fn slotAwaiting(srv: *Server, si: usize) bool {
for (srv.clients) |slot| {
const cs = slot orelse continue;
if ((cs.session orelse continue) != si) continue;
if (cs.await_state != null) return true;
}
return false;
}
test "Server: cmd_state pushes stay inside their session" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "markshome");
defer td.deinit();
// The mark-emitting scripted shell the cmd_state push test uses, for
// the same reason: it emits exactly one mark when told to, so any push
// arriving at the OTHER session's client is unambiguously a leak.
try td.tmp.dir.writeFile(.{
.sub_path = "marks.sh",
.data =
\\#!/bin/sh
\\read -r start
\\printf '\033]133;C\007'
\\read -r stop
\\
,
.flags = .{ .mode = 0o755 },
});
const script = try std.fmt.allocPrintSentinel(alloc, "{s}/marks.sh", .{td.tmp.path()}, 0);
defer alloc.free(script);
try td.start(.{ .shell = script });
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
const b = try dial.dialAttachNamed(td.sock_path, 80, 24, "b");
defer b.close();
const fa = (try awaitFrame(alloc, &td.srv, a.handle, .snapshot, 400)) orelse
return error.NoSnapshotA;
fa.deinit(alloc);
const fb = (try awaitFrame(alloc, &td.srv, b.handle, .snapshot, 400)) orelse
return error.NoSnapshotB;
fb.deinit(alloc);
// Drive a's session into its C: a push for a's clients alone.
try proto.writeFrame(a.handle, .input, "go\n");
const f1 = (try awaitFrame(alloc, &td.srv, a.handle, .cmd_state, 500)) orelse
return error.NoRunningPush;
defer f1.deinit(alloc);
const running = try proto.decodeCmdState(f1.payload);
try std.testing.expectEqual(proto.CmdPhase.running, running.phase);
try std.testing.expectEqual(proto.Mechanism.marks, running.mechanism);
// b hears nothing: its own shell emitted no mark, and a's must not
// cross. The budget keeps pumping so a misdirected push, which would
// have been queued in the same pump a's was, has amply arrived.
if (try awaitFrame(alloc, &td.srv, b.handle, .cmd_state, 60)) |leak| {
leak.deinit(alloc);
return error.CmdStateCrossedSessions;
}
}
test "Server: an await resolves against the awaiting client's session" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.open(alloc, "awaithome");
defer td.deinit();
// The await test's burst, one more session in the room: `D;3` and the
// next prompt's `A` in one write, so the return is complete the moment
// it lands.
try td.tmp.dir.writeFile(.{
.sub_path = "await.sh",
.data =
\\#!/bin/sh
\\read -r go
\\printf '\033]133;C\007out\r\n\033]133;D;3\007\033]133;A\007'
\\read -r stop
\\
,
.flags = .{ .mode = 0o755 },
});
const script = try std.fmt.allocPrintSentinel(alloc, "{s}/await.sh", .{td.tmp.path()}, 0);
defer alloc.free(script);
try td.start(.{ .shell = script });
const a = try dial.dialAttach(td.sock_path, 80, 24);
defer a.close();
const b = try dial.dialAttachNamed(td.sock_path, 80, 24, "b");
defer b.close();
const fa = (try awaitFrame(alloc, &td.srv, a.handle, .snapshot, 400)) orelse
return error.NoSnapshotA;
fa.deinit(alloc);
const fb = (try awaitFrame(alloc, &td.srv, b.handle, .snapshot, 400)) orelse
return error.NoSnapshotB;
fb.deinit(alloc);
// a asks about ITS session; nothing there may answer yet.
try proto.writeFrame(a.handle, .await_req, &proto.encodeAwaitReq(.{
.since_seq = td.srv.sessions.table[0].?.tracker.seq,
.settle_ms = 0,
.timeout_ms = 10_000,
}));
// b's shell returns a command, and the daemon has demonstrably seen it.
const si_b = td.srv.sessions.find("b") orelse return error.SessionBMissing;
try proto.writeFrame(b.handle, .input, "go\n");
const Returned = struct {
srv: *Server,
si: usize,
fn yes(self: @This()) bool {
return self.srv.sessions.table[self.si].?.last_return != null;
}
};
try std.testing.expect(try h.pumpUntil(&td.srv, 3000, Returned{ .srv = &td.srv, .si = si_b }, Returned.yes));
// That return is another session's and must not resolve a's await.
if (try awaitFrame(alloc, &td.srv, a.handle, .await_reply, 40)) |early| {
early.deinit(alloc);
return error.AwaitAnsweredByAnotherSession;
}
// a's own shell returning is what answers it, code and all.
try proto.writeFrame(a.handle, .input, "go\n");
const f = (try awaitFrame(alloc, &td.srv, a.handle, .await_reply, 500)) orelse
return error.NoAwaitReplyFromOwnSession;
defer f.deinit(alloc);
const rep = try proto.decodeAwaitReply(f.payload);
try std.testing.expectEqual(proto.AwaitReason.returned, rep.reason);
try std.testing.expectEqual(proto.Mechanism.marks, rep.state.mechanism);
try std.testing.expectEqual(@as(?u8, 3), rep.state.exit_code);
}
test "Server: a promoted-but-unattached slot receives nothing" {
const alloc = std.testing.allocator;
var td = try h.TestDaemon.init(alloc, "limbo", .{ .shell = "/bin/cat" });
defer td.deinit();
// The shape a QUIC connection has between handshake and first attach:
// a live client slot with no session. Connection = session means such
// a slot receives nothing — no deltas, no snapshots, no mode bits —
// until an attach says which session it is asking about.
const c = try connectedPair();
defer std.posix.close(c.peer);
td.srv.clients[0] = .{ .sink = .{ .socket = c.daemon } };
// Make the default session produce broadcasts of every kind a pump
// emits: grid updates from the echo, and the mode poll's first report.
try proto.writeAllFd(td.srv.sessions.table[0].?.pty.master, "say-something\n");
var i: usize = 0;
while (i < 40) : (i += 1) try td.srv.pumpOnce(5);
// Liveness: the grid really moved, so silence below is the filter at
// work and not a session that never spoke.
const plain = try td.srv.sessions.table[0].?.eng.dumpPlain(alloc);
defer alloc.free(plain);
try std.testing.expect(std.mem.indexOf(u8, plain, "say-something") != null);
// The slot was told nothing — nothing queued, nothing on the wire.
try std.testing.expect(td.srv.clients[0] != null);
try std.testing.expectEqual(@as(usize, 0), td.srv.clients[0].?.pending.items.len);
// A bare poll, not a wait: the claim is that nothing is readable at this
// instant, and any budget at all would turn it into a slower claim about
// a window instead.
var pfd = [_]std.posix.pollfd{
.{ .fd = c.peer, .events = std.posix.POLL.IN, .revents = 0 },
};
try std.testing.expectEqual(@as(usize, 0), try std.posix.poll(&pfd, 0));
}