a73x

db1ce9eb

feat: a daemon that starts from another daemon's memory

a73x   2026-08-26 15:17

Commit message
feat: a daemon that starts from another daemon's memory

`run --resume-fd N` reads the manifest off the inherited descriptor and
rebuilds the daemon around what it names: the listener fd is served
without a claim (the fd IS the claim), each session gets a fresh engine
replayed from its VT dump, its title fed back as an OSC 0 so the engine
owns it again, its pty adopted by fd and pid, and a fresh epoch so every
returning client takes the snapshot path.

`--check` is the old daemon's dry run: parse to the end, adopt nothing,
exit 0 — the candidate proves it can read the manifest while the daemon
that wrote it is still serving.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
build.zig
Old New
@@ -272,7 +272,7 @@ const mod_table = [_]ModSpec{
272 // the sun_path bound, checked before any verb acts on the path; and the 272 // the sun_path bound, checked before any verb acts on the path; and the
273 // keygen round-trip test needs a directory to generate into, which the 273 // keygen round-trip test needs a directory to generate into, which the
274 // daemon itself never touches. 274 // daemon itself never touches.
275 .{ .name = "exe", .path = "src/main.zig", .layer = 3, .link_libc = true, .imports = &.{ "server", "protocol", "proxy", "quic", "quic_server", "xdg", "spawn", "handoff", "sockpath" }, .test_imports = &.{"testtmp"}, .quic_tests = true }, 275 .{ .name = "exe", .path = "src/main.zig", .layer = 3, .link_libc = true, .imports = &.{ "server", "protocol", "proxy", "quic", "quic_server", "xdg", "spawn", "handoff", "sockpath", "upgrade" }, .test_imports = &.{"testtmp"}, .quic_tests = true },
276 // ---- layer 4 ---- 276 // ---- layer 4 ----
277 // The muxweb hub's HTTP/WebSocket decisions: Origin gate, route 277 // The muxweb hub's HTTP/WebSocket decisions: Origin gate, route
278 // table, WS endpoint naming. Assets are injected (the exe root 278 // table, WS endpoint naming. Assets are injected (the exe root
src/main.zig
Old New
@@ -12,6 +12,7 @@ const xdg = @import("xdg");
12 const spawn = @import("spawn"); 12 const spawn = @import("spawn");
13 const handoff = @import("handoff"); 13 const handoff = @import("handoff");
14 const sockpath = @import("sockpath"); 14 const sockpath = @import("sockpath");
15 const upgrade = @import("upgrade");
15 16
16 const usage = 17 const usage =
17 \\usage: 18 \\usage:
@@ -146,6 +147,16 @@ const Opts = struct {
146 /// — no tail at all — so a caller that never passes `--session` builds 147 /// — no tail at all — so a caller that never passes `--session` builds
147 /// byte-identical payloads to before this flag existed. 148 /// byte-identical payloads to before this flag existed.
148 session: []const u8 = "", 149 session: []const u8 = "",
150 /// The inherited manifest descriptor an upgrade exec'd us with. Not a
151 /// user flag: the old daemon writes it into our argv. Its presence is
152 /// what makes `run` an ADOPTION rather than a start, and it is also
153 /// what excuses this process from resolving a socket path — the
154 /// manifest names the socket the inherited listener is already bound to.
155 resume_fd: ?std.posix.fd_t = null,
156 /// The old daemon's dry run: parse the manifest to the end and exit 0
157 /// having adopted nothing. A candidate that cannot read the manifest
158 /// must fail HERE, in a child, while the old daemon is still serving.
159 check: bool = false,
149 }; 160 };
150 161
151 /// A refusal, carrying whatever `main` needs to print one line about it. 162 /// A refusal, carrying whatever `main` needs to print one line about it.
@@ -186,6 +197,10 @@ fn parseArgs(args: []const [:0]const u8) ParseResult {
186 o.vt = true; 197 o.vt = true;
187 continue; 198 continue;
188 } 199 }
200 if (std.mem.eql(u8, a, "--check")) {
201 o.check = true;
202 continue;
203 }
189 // Every remaining flag takes a value, so the missing-value case is 204 // Every remaining flag takes a value, so the missing-value case is
190 // answered once here rather than at each arm — a flag with nothing 205 // answered once here rather than at each arm — a flag with nothing
191 // after it used to fall through to "unknown argument", which named 206 // after it used to fall through to "unknown argument", which named
@@ -197,7 +212,8 @@ fn parseArgs(args: []const [:0]const u8) ParseResult {
197 std.mem.eql(u8, a, "--quic") or 212 std.mem.eql(u8, a, "--quic") or
198 std.mem.eql(u8, a, "--key") or 213 std.mem.eql(u8, a, "--key") or
199 std.mem.eql(u8, a, "--quic-idle-ms") or 214 std.mem.eql(u8, a, "--quic-idle-ms") or
200 std.mem.eql(u8, a, "--session"); 215 std.mem.eql(u8, a, "--session") or
216 std.mem.eql(u8, a, "--resume-fd");
201 if (!takes_value) return .{ .err = .{ .unknown_arg = a } }; 217 if (!takes_value) return .{ .err = .{ .unknown_arg = a } };
202 if (i + 1 >= args.len) return .{ .err = .{ .missing_value = a } }; 218 if (i + 1 >= args.len) return .{ .err = .{ .missing_value = a } };
203 i += 1; 219 i += 1;
@@ -216,6 +232,9 @@ fn parseArgs(args: []const [:0]const u8) ParseResult {
216 // becomes a frame" the socket-path length guard follows. 232 // becomes a frame" the socket-path length guard follows.
217 if (!proto.validSessionName(v)) return .{ .err = .{ .bad_session_name = v } }; 233 if (!proto.validSessionName(v)) return .{ .err = .{ .bad_session_name = v } };
218 o.session = v; 234 o.session = v;
235 } else if (std.mem.eql(u8, a, "--resume-fd")) {
236 o.resume_fd = std.fmt.parseInt(std.posix.fd_t, v, 10) catch
237 return .{ .err = .{ .bad_number = a } };
219 } else if (std.mem.eql(u8, a, "--cols")) { 238 } else if (std.mem.eql(u8, a, "--cols")) {
220 o.cols = std.fmt.parseInt(u16, v, 10) catch return .{ .err = .{ .bad_number = a } }; 239 o.cols = std.fmt.parseInt(u16, v, 10) catch return .{ .err = .{ .bad_number = a } };
221 } else if (std.mem.eql(u8, a, "--rows")) { 240 } else if (std.mem.eql(u8, a, "--rows")) {
@@ -308,7 +327,12 @@ pub fn main() !u8 {
308 // refuse (sockpath.defaultSockPath), and a version string must never 327 // refuse (sockpath.defaultSockPath), and a version string must never
309 // fail on the environment. So a verb that touches no socket gets no 328 // fail on the environment. So a verb that touches no socket gets no
310 // path rather than a path it must first survive resolving. 329 // path rather than a path it must first survive resolving.
311 const uses_socket = specForCmd(o.cmd).uses_socket; 330 // A resuming `run` is exempt from the socket path entirely: the exec
331 // that started it passed only `--resume-fd`, and the path it must serve
332 // is in the manifest, under a listener that is already bound to it.
333 // Resolving the default here would refuse an upgrade on any daemon
334 // started with `--sock` outside XDG_RUNTIME_DIR.
335 const uses_socket = specForCmd(o.cmd).uses_socket and o.resume_fd == null;
312 const sock_path = if (o.sock) |s| 336 const sock_path = if (o.sock) |s|
313 try alloc.dupe(u8, s) 337 try alloc.dupe(u8, s)
314 else if (!uses_socket) 338 else if (!uses_socket)
@@ -356,7 +380,7 @@ pub fn main() !u8 {
356 }, 380 },
357 .keygen => return keygen(alloc), 381 .keygen => return keygen(alloc),
358 .start => return startCmd(alloc, sock_path, args[2..]), 382 .start => return startCmd(alloc, sock_path, args[2..]),
359 .run => return run(alloc, o, sock_path), 383 .run => return if (o.resume_fd) |fd| resumeRun(alloc, o, fd) else run(alloc, o, sock_path),
360 .dump => return dump(alloc, sock_path, o.vt, o.session), 384 .dump => return dump(alloc, sock_path, o.vt, o.session),
361 .stats => return stats(alloc, sock_path), 385 .stats => return stats(alloc, sock_path),
362 .stop => return stopCmd(alloc, sock_path), 386 .stop => return stopCmd(alloc, sock_path),
@@ -383,6 +407,50 @@ fn shellIntegrationEnabled(env: ?[]const u8) bool {
383 return std.mem.eql(u8, env orelse "", "1"); 407 return std.mem.eql(u8, env orelse "", "1");
384 } 408 }
385 409
410 /// Bigger than any manifest a full session table can produce (one replayed
411 /// viewport each) and small enough that a descriptor that is not a manifest
412 /// cannot make this process eat the machine.
413 const manifest_read_max = 64 * 1024 * 1024;
414
415 /// `muxd run --resume-fd N`: the argv an upgrading daemon exec'd this binary
416 /// with. Same pid, same children, same descriptors — the manifest names
417 /// which ones. It is read from the descriptor and never from a path: the
418 /// memfd is anonymous memory, and the QUIC key inside it must not touch disk.
419 fn resumeRun(alloc: std.mem.Allocator, o: Opts, resume_fd: std.posix.fd_t) !u8 {
420 // The writer left the offset at the end of what it wrote, and a child
421 // shares the file description with it, so the rewind is ours to do.
422 var file = std.fs.File{ .handle = resume_fd };
423 file.seekTo(0) catch |err| {
424 std.debug.print("muxd run: --resume-fd {d} does not seek ({t})\n", .{ resume_fd, err });
425 return 1;
426 };
427 const bytes = file.readToEndAlloc(alloc, manifest_read_max) catch |err| {
428 std.debug.print("muxd run: cannot read the manifest on fd {d} ({t})\n", .{ resume_fd, err });
429 return 1;
430 };
431 defer alloc.free(bytes);
432
433 var parsed = upgrade.parseManifest(alloc, bytes) catch |err| {
434 std.debug.print("muxd run: manifest on fd {d} is not one ({t})\n", .{ resume_fd, err });
435 return 1;
436 };
437 defer parsed.deinit();
438
439 // `--check` is the old daemon's dry run and this is the whole of it: the
440 // candidate proves it can read the manifest to the end, exits 0, and
441 // adopts nothing — the daemon that wrote it is still serving.
442 if (o.check) return 0;
443
444 var srv = Server.initFromManifest(alloc, &parsed, build_options.version) catch |err| {
445 std.debug.print("muxd run: cannot adopt the manifest ({t})\n", .{err});
446 return 1;
447 };
448 defer srv.deinit();
449
450 @import("server").installSignalHandlers();
451 return try srv.run();
452 }
453
386 fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 { 454 fn run(alloc: std.mem.Allocator, o: Opts, sock_path: []const u8) !u8 {
387 // Address and key are settled before anything binds: a mistyped address 455 // Address and key are settled before anything binds: a mistyped address
388 // or an unreadable key must not first leave a session socket and a live 456 // or an unreadable key must not first leave a session socket and a live
@@ -1043,7 +1111,7 @@ test "parseArgs: --quic-idle-ms defaults, parses, and refuses nonsense" {
1043 test "parseArgs: a value-taking flag at the end of argv names itself" { 1111 test "parseArgs: a value-taking flag at the end of argv names itself" {
1044 // This used to report "unknown argument: --quic", which blames the flag 1112 // This used to report "unknown argument: --quic", which blames the flag
1045 // rather than the missing value. 1113 // rather than the missing value.
1046 inline for (.{ "--sock", "--shell", "--cols", "--rows", "--quic", "--key", "--quic-idle-ms", "--session" }) |flag| { 1114 inline for (.{ "--sock", "--shell", "--cols", "--rows", "--quic", "--key", "--quic-idle-ms", "--session", "--resume-fd" }) |flag| {
1047 const r = parse(&.{ "muxd", "run", flag }); 1115 const r = parse(&.{ "muxd", "run", flag });
1048 try std.testing.expect(r.err == .missing_value); 1116 try std.testing.expect(r.err == .missing_value);
1049 try std.testing.expectEqualStrings(flag, r.err.missing_value); 1117 try std.testing.expectEqualStrings(flag, r.err.missing_value);
@@ -1200,6 +1268,25 @@ test "parseArgs: endpoint is a command and takes --sock" {
1200 try std.testing.expectEqualStrings("--sock", missing.err.missing_value); 1268 try std.testing.expectEqualStrings("--sock", missing.err.missing_value);
1201 } 1269 }
1202 1270
1271 test "parseArgs: run --resume-fd N --check is the old daemon's dry run" {
1272 const r = parse(&.{ "muxd", "run", "--resume-fd", "7", "--check" });
1273 try std.testing.expect(r == .ok);
1274 try std.testing.expect(r.ok.cmd == .run);
1275 try std.testing.expectEqual(@as(std.posix.fd_t, 7), r.ok.resume_fd.?);
1276 try std.testing.expect(r.ok.check);
1277
1278 // A number, like --cols: an fd that is not one would be read as a
1279 // descriptor the daemon never passed.
1280 try std.testing.expect(parse(&.{ "muxd", "run", "--resume-fd", "x" }).err == .bad_number);
1281
1282 // Neither flag is the ordinary start, and both must stay off there —
1283 // a `run` that thought it was resuming would adopt nothing and serve
1284 // nothing.
1285 const plain = parse(&.{ "muxd", "run" });
1286 try std.testing.expect(plain.ok.resume_fd == null);
1287 try std.testing.expect(!plain.ok.check);
1288 }
1289
1203 test "announceKeyFrom: MUX_KEY_FILE wins, and the default it skipped is not created" { 1290 test "announceKeyFrom: MUX_KEY_FILE wins, and the default it skipped is not created" {
1204 const testtmp = @import("testtmp"); 1291 const testtmp = @import("testtmp");
1205 var tmp = try testtmp.TmpDir.make(); 1292 var tmp = try testtmp.TmpDir.make();
src/server.zig
Old New
@@ -689,6 +689,195 @@ pub const Server = struct {
689 return srv; 689 return srv;
690 } 690 }
691 691
692 /// Adopt the manifest an exec-ing daemon left in a memfd.
693 pub fn initFromManifest(
694 alloc: std.mem.Allocator,
695 parsed: *const upgrade.Parsed,
696 version: []const u8,
697 ) !Server {
698 // Same pid, same children, same descriptors. No `sockpath.claim`
699 // here, deliberately: the inherited listener fd
700 // IS the claim, and claim's probe would find our own socket
701 // answering and refuse the daemon that is already serving it.
702 //
703 // `version` is THIS binary's, never the manifest's — the writer's
704 // version is the rollback story, and the next upgrade_req's skew
705 // check has to measure against what is running now.
706 const d = parsed.daemon;
707
708 // Everything the Server keeps a slice of is copied out of the
709 // manifest arena, which dies with the caller's `Parsed` while the
710 // Server outlives it.
711 var shellint_arena = std.heap.ArenaAllocator.init(alloc);
712 errdefer shellint_arena.deinit();
713 const a = shellint_arena.allocator();
714 const sock_path = try a.dupe(u8, d.sock_path);
715 const shell_z = try a.dupeZ(u8, d.shell);
716 const extra_env = try a.alloc(Pty.EnvPair, d.extra_env.len);
717 for (d.extra_env, extra_env) |src, *dst| dst.* = .{
718 .key = try a.dupeZ(u8, src.key),
719 .value = if (src.value) |v| try a.dupeZ(u8, v) else null,
720 };
721
722 // The shim directory crossed as a path because live shells hold it
723 // in ZDOTDIR (or --init-file) and it is pid-named: this process
724 // could never mint that name again, and a fresh one would leave the
725 // old directory with no owner to delete it. `shellint.prepare`
726 // creates its directory exclusively — adopting one is a symlink
727 // attack — so the path is taken back out and rewritten with THIS
728 // binary's scripts under the name the shells already name.
729 const injection: shellint.Injection = if (d.shellint_dir) |src| blk: {
730 const dir = try a.dupe(u8, src);
731 std.fs.cwd().deleteTree(dir) catch {};
732 break :blk shellint.prepare(a, dir, shell_z) catch |err| {
733 std.debug.print(
734 "muxd: resumed without shell integration ({s}: {t}); " ++
735 "sessions spawned from here run without command marks\n",
736 .{ dir, err },
737 );
738 break :blk shellint.no_injection;
739 };
740 } else shellint.no_injection;
741
742 const opts: Options = .{
743 .sock_path = sock_path,
744 .shell = shell_z,
745 .shell_integration = d.shell_integration,
746 .extra_env = extra_env,
747 .version = version,
748 };
749 const plan = try planFrom(a, opts, injection);
750
751 const agent_dir: ?[]const u8 = if (d.agent_dir) |dir| try alloc.dupe(u8, dir) else null;
752 errdefer if (agent_dir) |dir| alloc.free(dir);
753
754 var srv: Server = .{
755 .alloc = alloc,
756 .spawn_plan = plan,
757 .spawn_shell = shell_z,
758 .spawn_shell_integration = d.shell_integration,
759 .spawn_extra_env = extra_env,
760 .version = version,
761 // The fd is the listener: no bind, no listen, no claim. The
762 // address is rebuilt from the path only because std.net.Server
763 // carries one; nothing reads it back.
764 .listener = .{
765 .listen_address = try std.net.Address.initUnix(sock_path),
766 .stream = .{ .handle = d.listener_fd },
767 },
768 .sock_path = sock_path,
769 .path_id = try sockpath.PathId.of(sock_path),
770 .shellint_arena = shellint_arena,
771 .shellint_dir = injection.dir,
772 .agent_dir = agent_dir,
773 };
774
775 // Cumulative, so an upgrade is not mistaken for a restart by
776 // anything sampling `muxd stats`. Saturating rather than @intCast
777 // below: a manifest is bytes, and a corrupt counter must not panic a
778 // daemon that is otherwise able to serve.
779 srv.stats = .{
780 .snapshots = d.counters.snapshots,
781 .snapshot_bytes = d.counters.snapshot_bytes,
782 .deltas = d.counters.deltas,
783 .delta_bytes = d.counters.delta_bytes,
784 .snapshot_equiv_bytes = d.counters.snapshot_equiv_bytes,
785 .attaches = d.counters.attaches,
786 };
787 srv.agent_refused_no_offer =
788 std.math.cast(u32, d.counters.agent_refused_no_offer) orelse std.math.maxInt(u32);
789 srv.agent_refused_full =
790 std.math.cast(u32, d.counters.agent_refused_full) orelse std.math.maxInt(u32);
791
792 // Frees what this constructor allocated and NOTHING the manifest
793 // handed over: no descriptor is closed and no child is signalled on
794 // a failed adoption, because the rollback exec is about to hand all
795 // of them to the old binary.
796 errdefer for (&srv.sessions) |*slot| {
797 if (slot.*) |*s| {
798 s.eng.deinit();
799 if (s.agent_path) |p| alloc.free(p);
800 slot.* = null;
801 }
802 };
803
804 for (parsed.sessions, 0..) |rec, i| {
805 // A manifest from a binary with a bigger table would otherwise
806 // index past ours; the sessions that fit are still served.
807 if (i >= max_sessions) break;
808 const eng = try Engine.init(alloc, .{
809 .cols = rec.cols,
810 .rows = rec.rows,
811 .clipboard_max = proto.clipboard_base64_max,
812 });
813 // Into the slot before anything else can fail, so the errdefer
814 // above owns the engine from here on.
815 srv.sessions[i] = .{
816 .eng = eng,
817 .pty = Pty.adopt(rec.pty_fd, rec.child_pid),
818 .epoch = freshEpoch(),
819 };
820 const s = &srv.sessions[i].?;
821 eng.feed(rec.vt);
822 // The title goes back in as an OSC 0 so the ENGINE owns it
823 // again: `sampleTermTitle` then announces it to the first client
824 // that attaches, with no resume-shaped exception anywhere
825 // downstream. title_sent stays null — nobody has been told.
826 if (rec.title) |t| {
827 const osc = try std.fmt.allocPrint(alloc, "\x1b]0;{s}\x07", .{t});
828 defer alloc.free(osc);
829 eng.feed(osc);
830 }
831 s.cmd = .{
832 .phase = std.meta.intToEnum(proto.CmdPhase, rec.cmd.phase) catch .at_prompt,
833 .marks_seen = rec.cmd.marks_seen,
834 .start_row = rec.cmd.start_row,
835 .end_row = rec.cmd.end_row,
836 .exit_code = rec.cmd.exit_code,
837 };
838 // The verdict crosses; its watermark cannot. `CmdState.seq` in
839 // a status reply is the RETURN watermark an await compares
840 // `since_seq` against, and the delta tracker is rebuilt from
841 // zero here — a seq from the old space is a watermark from the
842 // future that no later return can exceed, and it swallowed the
843 // first await after every upgrade.
844 s.last_return = if (rec.last_return) |lr| blk: {
845 var restamped = lr;
846 restamped.seq = s.tracker.seq;
847 break :blk restamped;
848 } else null;
849 if (rec.agent_path) |p| {
850 s.agent_path = try alloc.dupeZ(u8, p);
851 s.agent_listener = rec.agent_fd;
852 }
853 const n = @min(rec.name.len, proto.session_name_max);
854 @memcpy(s.name_buf[0..n], rec.name[0..n]);
855 s.name_len = @intCast(n);
856 }
857
858 // The UDP socket crossed bound and the key crossed as bytes, so TLS
859 // stands back up on the same port with the same PSK. Both arms
860 // become `.owned`: whatever held the reference in the previous image
861 // went with it, and deinit is the only thing left that could free
862 // this one. The handler's ctx is bound in `run`, which is the first
863 // place the Server has its final address.
864 switch (d.quic.arm) {
865 .none => {},
866 .borrowed, .owned => {
867 const l = try quic_server.Listener.initFromFd(
868 alloc,
869 d.quic.fd,
870 .{ .bytes = d.quic.key },
871 undefined,
872 d.quic.idle_ms,
873 );
874 srv.quic = .{ .owned = l };
875 },
876 }
877
878 return srv;
879 }
880
692 /// Null if it could not be made. Created exclusively at 0700 under a name 881 /// Null if it could not be made. Created exclusively at 0700 under a name
693 /// with a random half: that parent is a shared `/tmp` whenever there is no 882 /// with a random half: that parent is a shared `/tmp` whenever there is no
694 /// `$XDG_RUNTIME_DIR`, a pid alone is guessable, and an entry pre-created 883 /// `$XDG_RUNTIME_DIR`, a pid alone is guessable, and an entry pre-created
@@ -759,6 +948,20 @@ pub const Server = struct {
759 return .{ .fd = listener.stream.handle, .path = path }; 948 return .{ .fd = listener.stream.handle, .path = path };
760 } 949 }
761 950
951 /// A session instance's identity in every snapshot it sends.
952 fn freshEpoch() u64 {
953 // Random rather than a counter or a timestamp: nothing on disk
954 // survives a daemon, and two daemons started in the same
955 // millisecond (tests do exactly this) must still differ. Never 0 —
956 // that value is a client saying it holds nothing. An adopted
957 // session mints one too: the grid it replays is not the one any
958 // client was being deltaed against, so every returning client has
959 // to take the snapshot path.
960 var epoch: u64 = 0;
961 while (epoch == 0) epoch = std.crypto.random.int(u64);
962 return epoch;
963 }
964
762 /// Takes ownership of `agent`: it lands on the Session or is released 965 /// Takes ownership of `agent`: it lands on the Session or is released
763 /// here; no teardown path can reach it otherwise. 966 /// here; no teardown path can reach it otherwise.
764 fn createSession( 967 fn createSession(
@@ -818,13 +1021,7 @@ pub const Server = struct {
818 }); 1021 });
819 errdefer pty.deinit(); 1022 errdefer pty.deinit();
820 1023
821 // Random rather than a counter or a timestamp: nothing on disk 1024 var s = Session{ .eng = eng, .pty = pty, .epoch = freshEpoch() };
822 // survives a daemon, and two daemons started in the same
823 // millisecond (tests do exactly this) must still differ.
824 var epoch: u64 = 0;
825 while (epoch == 0) epoch = std.crypto.random.int(u64);
826
827 var s = Session{ .eng = eng, .pty = pty, .epoch = epoch };
828 if (agent) |a| { 1025 if (agent) |a| {
829 s.agent_listener = a.fd; 1026 s.agent_listener = a.fd;
830 s.agent_path = a.path; 1027 s.agent_path = a.path;
@@ -861,7 +1058,19 @@ pub const Server = struct {
861 shellint.install(a, std.fs.path.dirname(opts.sock_path) orelse ".", opts.shell) 1058 shellint.install(a, std.fs.path.dirname(opts.sock_path) orelse ".", opts.shell)
862 else 1059 else
863 shellint.no_injection; 1060 shellint.no_injection;
864 1061 return planFrom(a, opts, injection);
1062 }
1063
1064 /// The plan for an injection somebody else decided on.
1065 fn planFrom(
1066 a: std.mem.Allocator,
1067 opts: Options,
1068 injection: shellint.Injection,
1069 ) !SpawnPlan {
1070 // Split from `prepareSpawn` for adoption: a resumed daemon must NOT
1071 // mint a second shim directory — `initFromManifest` re-prepares the
1072 // one the live shells already name — and everything below this line
1073 // is identical either way.
865 // shellint speaks its own EnvPair so it can stay a leaf, and so can 1074 // shellint speaks its own EnvPair so it can stay a leaf, and so can
866 // pty; the daemon is the one place that knows about both, so the 1075 // pty; the daemon is the one place that knows about both, so the
867 // mapping lives here. 1076 // mapping lives here.
@@ -1324,6 +1533,11 @@ pub const Server = struct {
1324 } 1533 }
1325 1534
1326 pub fn run(self: *Server) !u8 { 1535 pub fn run(self: *Server) !u8 {
1536 // The pump cannot start on a listener whose handler points anywhere
1537 // but here. An adopted listener (initFromManifest) is built before
1538 // the Server has its final address, and re-binding a listener the
1539 // caller already wired is the same assignment twice.
1540 if (self.quicListener()) |l| l.setHandler(self.quicHandler());
1327 while (true) { 1541 while (true) {
1328 if (shutdown_flag.load(.acquire)) return 130; 1542 if (shutdown_flag.load(.acquire)) return 130;
1329 // An upgrade was accepted: the reply has drained (pumpOnce 1543 // An upgrade was accepted: the reply has drained (pumpOnce
@@ -11528,6 +11742,141 @@ test "writeManifestTo: what crosses is what a session cannot rebuild" {
11528 try std.testing.expectEqual(s.pty.master, parsed.sessions[0].pty_fd); 11742 try std.testing.expectEqual(s.pty.master, parsed.sessions[0].pty_fd);
11529 } 11743 }
11530 11744
11745 test "initFromManifest: an adopted session answers a status_req without having been created" {
11746 const alloc = std.testing.allocator;
11747
11748 var tmp = try TmpDir.make();
11749 defer tmp.cleanup();
11750 const dir_path = tmp.path();
11751 const sock_path = try std.fmt.allocPrint(alloc, "{s}/adopt.sock", .{dir_path});
11752 defer alloc.free(sock_path);
11753
11754 var srv = try Server.init(alloc, .{
11755 .sock_path = sock_path,
11756 .shell = "/bin/sh",
11757 .cols = 100,
11758 .rows = 30,
11759 });
11760 {
11761 const s = &srv.sessions[0].?;
11762 s.eng.feed("\x1b]0;adopted\x07");
11763 // Set rather than earned: the mark machinery has its own tests, and
11764 // the claim here is that the FACT crosses — a lost marks_seen
11765 // demotes every later await from marks to pgid, silently.
11766 s.cmd.marks_seen = true;
11767 }
11768 const old_epoch = srv.sessions[0].?.epoch;
11769 const child = srv.sessions[0].?.pty.child;
11770
11771 const memfd = try std.posix.memfd_create("mux-adopt-test", 0);
11772 defer std.posix.close(memfd);
11773 try srv.writeManifestTo(memfd, "0.0.1-99");
11774
11775 // Release the first Server's MEMORY by hand instead of calling deinit:
11776 // deinit is the demolition list — it unlinks the socket, SIGKILLs the
11777 // shell and deleteTrees the dirs — and every one of those is what the
11778 // adopting Server is about to inherit. The descriptors and the child
11779 // are left alone here and torn down once, by srv2.
11780 {
11781 const s = &srv.sessions[0].?;
11782 s.tracker.deinit(alloc);
11783 s.freePending(alloc);
11784 if (s.title_sent) |t| alloc.free(t);
11785 s.eng.deinit();
11786 if (s.agent_path) |p| alloc.free(p);
11787 }
11788 if (srv.agent_dir) |d| alloc.free(d);
11789 srv.shellint_arena.deinit();
11790
11791 var file = std.fs.File{ .handle = memfd };
11792 try file.seekTo(0);
11793 const buf = try file.readToEndAlloc(alloc, 4 * 1024 * 1024);
11794 defer alloc.free(buf);
11795 var parsed = try upgrade.parseManifest(alloc, buf);
11796 defer parsed.deinit();
11797
11798 var srv2 = try Server.initFromManifest(alloc, &parsed, "0.0.1-100");
11799 defer srv2.deinit();
11800
11801 // The socket the OLD daemon bound still answers, over the inherited
11802 // listener fd: no second bind, no claim, no unlink in between.
11803 const obs = try std.net.connectUnixSocket(sock_path);
11804 defer obs.close();
11805 try proto.writeFrame(obs.handle, .status_req, "");
11806 const reply = (try awaitFrame(alloc, &srv2, obs.handle, .status_reply, 400)) orelse
11807 return error.NoStatusReply;
11808 defer reply.deinit(alloc);
11809 const st = try proto.decodeStatusReply(reply.payload);
11810 try std.testing.expectEqual(@as(u16, 100), st.cols);
11811 try std.testing.expectEqual(@as(u16, 30), st.rows);
11812 try std.testing.expectEqual(proto.Mechanism.marks, st.cmd.mechanism);
11813
11814 const s2 = &srv2.sessions[0].?;
11815 try std.testing.expectEqualStrings(proto.default_session, s2.name());
11816 // Same pid, so checkExited's waitpid still answers for this shell.
11817 try std.testing.expectEqual(child, s2.pty.child);
11818 // The title is the engine's own again, fed back as an OSC 0.
11819 try std.testing.expectEqualStrings("adopted", s2.eng.title());
11820 // A fresh epoch is what puts every returning client on the snapshot
11821 // path: one quoting the old epoch must never be served deltas over a
11822 // grid this process replayed.
11823 try std.testing.expect(s2.epoch != old_epoch);
11824 }
11825
11826 test "initFromManifest: the return watermark is re-stamped, never carried across seq spaces" {
11827 const alloc = std.testing.allocator;
11828
11829 var tmp = try TmpDir.make();
11830 defer tmp.cleanup();
11831 const sock_path = try std.fmt.allocPrint(alloc, "{s}/watermark.sock", .{tmp.path()});
11832 defer alloc.free(sock_path);
11833
11834 var srv = try Server.init(alloc, .{ .sock_path = sock_path, .shell = "/bin/sh" });
11835 srv.sessions[0].?.last_return = .{
11836 .phase = .returned,
11837 .mechanism = .marks,
11838 .exit_code = 7,
11839 .start_row = 1,
11840 .end_row = 2,
11841 .seq = 999,
11842 };
11843
11844 const memfd = try std.posix.memfd_create("mux-watermark-test", 0);
11845 defer std.posix.close(memfd);
11846 try srv.writeManifestTo(memfd, "0.0.1-99");
11847
11848 // Memory only; the descriptors and the child are srv2's to tear down.
11849 {
11850 const s = &srv.sessions[0].?;
11851 s.tracker.deinit(alloc);
11852 s.freePending(alloc);
11853 if (s.title_sent) |t| alloc.free(t);
11854 s.eng.deinit();
11855 if (s.agent_path) |p| alloc.free(p);
11856 }
11857 if (srv.agent_dir) |d| alloc.free(d);
11858 srv.shellint_arena.deinit();
11859
11860 var file = std.fs.File{ .handle = memfd };
11861 try file.seekTo(0);
11862 const buf = try file.readToEndAlloc(alloc, 4 * 1024 * 1024);
11863 defer alloc.free(buf);
11864 var parsed = try upgrade.parseManifest(alloc, buf);
11865 defer parsed.deinit();
11866
11867 var srv2 = try Server.initFromManifest(alloc, &parsed, "0.0.1-100");
11868 defer srv2.deinit();
11869
11870 const s2 = &srv2.sessions[0].?;
11871 // The verdict crosses: it is what an await about a PAST command answers.
11872 try std.testing.expectEqual(@as(?u8, 7), s2.last_return.?.exit_code);
11873 // The watermark cannot. The delta tracker is rebuilt from zero, so a
11874 // seq from the old space is a watermark from the FUTURE, and no return
11875 // after the upgrade can ever exceed it: measured on a live daemon, the
11876 // first `muxa run` after an upgrade timed out and the second did not.
11877 try std.testing.expectEqual(s2.tracker.seq, s2.last_return.?.seq);
11878 }
11879
11531 test "validateUpgrade: same version without the flag names both versions" { 11880 test "validateUpgrade: same version without the flag names both versions" {
11532 const alloc = std.testing.allocator; 11881 const alloc = std.testing.allocator;
11533 11882