diff options
Diffstat (limited to '9agents/src/main.zig')
| -rw-r--r-- | 9agents/src/main.zig | 382 |
1 files changed, 382 insertions, 0 deletions
diff --git a/9agents/src/main.zig b/9agents/src/main.zig new file mode 100644 index 0000000..2ac9078 --- /dev/null +++ b/9agents/src/main.zig @@ -0,0 +1,382 @@ +//! 9agents: the coding-agents fs daemon — a long-running 9P2000 server that +//! mirrors every AI-agent harness's state (Claude Code, Codex, omp, +//! hermes, dsh + their skills) as one read-only, fresh-from-disk tree. +//! See docs/DESIGN.md for the tree contract and src/tree.zig for the tree. +//! +//! By default it posts itself under the name `agents` with +//! `serve.Runner.listenPosted`, so it appears at $XDG_RUNTIME_DIR/9p/agents +//! and every interactive fish (self-wrapped in `9ns --mntgen`) sees it at +//! /mnt/9p/agents with zero configuration. `--unix`, `--tcp` and `--fd` +//! are the other listen forms; `--no-post` serves without posting; the +//! five harness roots default to $HOME/.<name> and `--root NAME=PATH` +//! pins any of them elsewhere (the tests use fake homes). +//! +//! v1 has no daemonization: it never forks or detaches, so it is run under +//! something that supervises it — a systemd user service with +//! `Restart=always` (see 9agents.service), or a zmx session: +//! +//! zmx run agents -d 9agents +//! +//! SIGTERM or SIGINT stops it cleanly (a posted name is unposted). +const std = @import("std"); +const builtin = @import("builtin"); +const cloud9 = @import("cloud9"); +const tree = @import("tree.zig"); + +const fs = cloud9.fs; +const serve = cloud9.serve; +const post = cloud9.post; +const transport = cloud9.transport; +const Io = std.Io; +const linux = std.os.linux; + +const usage_text = + \\usage: 9agents [--unix PATH | --tcp IP:PORT | --fd N] [--no-post] + \\ [--name NAME] [--root NAME=PATH]... [--proc DIR] + \\ [--allow-move] [--zmx PATH] + \\ + \\A read-only, fresh-from-disk 9P2000 view of every harness's state: + \\ /README the tree, explained in place + \\ /pid /uptime daemon facts + \\ /claude/{projects,history,skills} + \\ /codex/{sessions,session-index,history} + \\ /omp /hermes /dsh full mirrors, raw + \\ /skills/{claude,codex,omp} the union skills view + \\ /active/<harness>/<pid>/ what is running right now + \\ + \\By default the daemon posts itself under the name `agents`, so it is + \\dialable at $XDG_RUNTIME_DIR/9p/agents and mountable by 9ns --mntgen + \\at /mnt/9p/agents. --no-post skips posting; --unix/--tcp add plain + \\listeners beside the post; --fd N serves one 9P session over the + \\connected stream on descriptor N and posts nothing. + \\--root NAME=PATH pins one harness root (NAME: claude, codex, omp, + \\hermes, dsh) somewhere other than $HOME/.<NAME>; repeatable. + \\--proc DIR reads the process tree somewhere other than /proc (for + \\tests); --proc "" leaves /active out of the tree entirely. + \\--allow-move lets a write to /active/<h>/<pid>/zmx move that agent + \\into a zmx session of that name: the daemon kills it and re-execs + \\the harness under zmx with its session resumed. Off by default, + \\because it is the one place the tree is not read-only, and anything + \\that can mount it can then kill an agent. --zmx PATH names the + \\binary a move runs (default: zmx, found on $PATH). + \\ + \\Run it in a zmx session, the zmx way: `zmx run agents -d 9agents`. + \\ +; + +const max_connections = 16; +const Runner = serve.Runner(tree.Harness, tree.opts, .{ + .msize = tree.msize, + .connections = max_connections, + .listeners = 2, // the posted name plus one of --unix/--tcp +}); + +// Static memory, the 9proc-demo way: the harness and the runner are large +// (the path table, the per-connection buffers) and live in .bss. +var harness_mem: tree.Harness = undefined; +var runner_mem: Runner = undefined; +var stop_requested = std.atomic.Value(bool).init(false); + +fn onSignal(sig: linux.SIG) callconv(.c) void { + _ = sig; + stop_requested.store(true, .release); +} + +fn installSignalHandlers() void { + const act = linux.Sigaction{ + .handler = .{ .handler = onSignal }, + .mask = @splat(0), + .flags = 0, + }; + _ = linux.sigaction(.TERM, &act, null); + _ = linux.sigaction(.INT, &act, null); + const ign = linux.Sigaction{ + .handler = .{ .handler = linux.SIG.IGN }, + .mask = @splat(0), + .flags = 0, + }; + _ = linux.sigaction(.PIPE, &ign, null); +} + +// ---- the serve handler --------------------------------------------------------- + +fn serveReq(ctx: ?*anyopaque, conn: *Runner.Conn, req: fs.Req) void { + const h: *tree.Harness = @ptrCast(@alignCast(ctx.?)); + // The answer's bytes point into the harness's shared buffers, so the + // mutex spans handle and reply: the engine copies them into the + // connection's output before the lock goes. + h.mutex.lockUncancelable(h.io); + const a = tree.handle(h, req); + conn.reply(&a.reply, a.bytes); + h.mutex.unlock(h.io); +} + +/// `name` found on `path_env`, as an absolute path in the arena. +fn onPath(io: Io, arena: std.mem.Allocator, path_env: []const u8, name: []const u8) ![]const u8 { + var it = std.mem.splitScalar(u8, path_env, ':'); + while (it.next()) |dir_path| { + if (dir_path.len == 0) continue; + const candidate = try std.fmt.allocPrint(arena, "{s}/{s}", .{ dir_path, name }); + const st = Io.Dir.statFile(.cwd(), io, candidate, .{ .follow_symlinks = true }) catch continue; + if (st.kind != .file) continue; + return candidate; + } + return error.NotFound; +} + +// ---- the CLI -------------------------------------------------------------------- + +const Mode = enum { posted, unix, tcp, fd }; + +pub fn main(init: std.process.Init) !void { + run(init) catch |e| switch (e) { + // Both are reported with their own line above; a returned error + // would bury it under a Debug-build stack trace. + error.Usage, error.AlreadyPosted => std.process.exit(1), + else => return e, + }; +} + +fn run(init: std.process.Init) !void { + const io = init.io; + const arena = init.arena.allocator(); + const args = try init.minimal.args.toSlice(arena); + const envp: post.Env = init.minimal.environ.block.slice.ptr; + + var mode: Mode = .posted; + var unix_path: []const u8 = ""; + var tcp_addr: []const u8 = ""; + var fd_no: i32 = -1; + var post_name: []const u8 = "agents"; + var no_post = false; + var base_overrides: [5]?[]const u8 = @splat(null); + var proc_root: []const u8 = "/proc"; + var zmx_path: []const u8 = "zmx"; + var allow_move = false; + + var i: usize = 1; + while (i < args.len) : (i += 1) { + const a = args[i]; + if (std.mem.eql(u8, a, "--no-post")) { + no_post = true; + } else if (std.mem.eql(u8, a, "--allow-move")) { + allow_move = true; + } else if (std.mem.eql(u8, a, "--unix") or std.mem.eql(u8, a, "--tcp") or + std.mem.eql(u8, a, "--fd") or std.mem.eql(u8, a, "--name") or + std.mem.eql(u8, a, "--proc") or std.mem.eql(u8, a, "--zmx") or + std.mem.eql(u8, a, "--root")) + { + i += 1; + if (i >= args.len) { + std.debug.print("9agents: {s} needs an argument\n{s}", .{ a, usage_text }); + return error.Usage; + } + const v = args[i]; + if (std.mem.eql(u8, a, "--unix")) { + mode = .unix; + unix_path = v; + } else if (std.mem.eql(u8, a, "--tcp")) { + mode = .tcp; + tcp_addr = v; + } else if (std.mem.eql(u8, a, "--fd")) { + mode = .fd; + fd_no = std.fmt.parseInt(i32, v, 10) catch { + std.debug.print("9agents: --fd: not a number: {s}\n", .{v}); + return error.Usage; + }; + } else if (std.mem.eql(u8, a, "--proc")) { + proc_root = v; + } else if (std.mem.eql(u8, a, "--zmx")) { + zmx_path = v; + } else if (std.mem.eql(u8, a, "--name")) { + post_name = v; + if (!post.legalName(post_name)) { + std.debug.print("9agents: illegal post name: {s}\n", .{v}); + return error.Usage; + } + } else { + const eq = std.mem.indexOfScalar(u8, v, '=') orelse { + std.debug.print("9agents: --root NAME=PATH: {s}\n{s}", .{ v, usage_text }); + return error.Usage; + }; + const name = v[0..eq]; + const root = std.meta.stringToEnum(tree.Root, name) orelse { + std.debug.print("9agents: unknown root {s} (claude, codex, omp, hermes, dsh)\n", .{name}); + return error.Usage; + }; + base_overrides[@intFromEnum(root)] = v[eq + 1 ..]; + } + } else if (std.mem.eql(u8, a, "--help") or std.mem.eql(u8, a, "-h")) { + std.debug.print("{s}", .{usage_text}); + return; + } else { + std.debug.print("9agents: unknown argument {s}\n{s}", .{ a, usage_text }); + return error.Usage; + } + } + + // The five pinned roots: $HOME/.<name> unless overridden. + const home = post.getenv(envp, "HOME"); + var bases: [5][]const u8 = @splat(""); + inline for (0..5) |r| { + if (base_overrides[r]) |path| { + bases[r] = path; + } else if (home) |hm| { + bases[r] = std.fmt.bufPrint(&root_bufs[r], "{s}/.{s}", .{ hm, root_dirs[r] }) catch return error.NameTooLong; + } + } + var any = false; + for (bases) |b| any = any or b.len != 0; + if (!any) { + std.debug.print("9agents: no harness roots (set $HOME or pass --root NAME=PATH)\n", .{}); + return error.Usage; + } + + // A move execs zmx by absolute path: the daemon resolves it once, + // at startup, so nothing about $PATH matters when the write lands. + const zmx_abs = if (std.mem.indexOfScalar(u8, zmx_path, '/') != null) + zmx_path + else + onPath(io, arena, post.getenv(envp, "PATH") orelse "", zmx_path) catch zmx_path; + if (allow_move and std.mem.indexOfScalar(u8, zmx_abs, '/') == null) { + std.debug.print("9agents: --allow-move needs zmx on $PATH (or --zmx PATH)\n", .{}); + return error.Usage; + } + + harness_mem.init(.{ + .io = io, + .pid = @intCast(linux.getpid()), + .bases = bases, + .proc = proc_root, + .home = home orelse "", + .zmx = zmx_abs, + .runtime = post.getenv(envp, "XDG_RUNTIME_DIR") orelse "", + .allow_move = allow_move, + .envp = envp, + }); + if (allow_move) { + std.debug.print("9agents: moves allowed — a write to /active/<h>/<pid>/zmx re-execs that agent under {s}\n", .{zmx_abs}); + } + if (no_post and mode == .posted) { + std.debug.print("9agents: --no-post needs a listen form (--unix, --tcp or --fd)\n{s}", .{usage_text}); + return error.Usage; + } + + installSignalHandlers(); + + switch (mode) { + .fd => { + std.debug.print("9agents: serving one session on fd {d}\n", .{fd_no}); + try serveFd(io, fd_no); + return; + }, + .posted, .unix, .tcp => {}, + } + + runner_mem.init(.{ .io = io, .root = tree.root, .handler = .{ .ctx = &harness_mem, .serve = serveReq } }); + defer runner_mem.stop(); + + var posted_something = false; + if (!no_post) { + runner_mem.listenPosted(envp, post_name, 16) catch |err| { + if (err == error.AlreadyPosted) { + std.debug.print("9agents: the name `{s}` is already posted by a live server\n", .{post_name}); + return err; + } + std.debug.print("9agents: cannot post as {s}: {t}\n", .{ post_name, err }); + return err; + }; + posted_something = true; + var pbuf: [transport.sun_path_len]u8 = undefined; + const path = post.registryPath(envp, post_name, &pbuf) catch ""; + std.debug.print("9agents: posted as {s} at {s}\n", .{ post_name, path }); + } + switch (mode) { + .unix => { + _ = try runner_mem.listen(.{ .unix = try arena.dupeZ(u8, unix_path) }, 16); + std.debug.print("9agents: listening on {s}\n", .{unix_path}); + }, + .tcp => { + const addr = Io.net.IpAddress.parseLiteral(tcp_addr) catch { + std.debug.print("9agents: bad --tcp address: {s}\n", .{tcp_addr}); + return error.Usage; + }; + const bound = try runner_mem.listen(.{ .tcp = addr }, 16); + std.debug.print("9agents: listening on tcp!{f}\n", .{bound}); + }, + else => {}, + } + var pinned: u32 = 0; + for (bases) |b| { + if (b.len != 0) pinned += 1; + } + std.debug.print("9agents: serving ({d} roots pinned, {d} bytes of tables)\n", .{ pinned, @sizeOf(tree.Harness) }); + + // The runner's tasks drive themselves; the main thread only waits for + // a stop signal (SIGTERM/SIGINT) and then unposts via stop(). + while (!stop_requested.load(.acquire)) { + io.sleep(.fromMilliseconds(250), .awake) catch break; + } + std.debug.print("9agents: stopping\n", .{}); +} + +const root_dirs = [5][]const u8{ "claude", "codex", "omp", "hermes", "dsh" }; +var root_bufs: [5][tree.base_capacity]u8 = @splat(@splat(0)); + +// ---- --fd N: one 9P session over a connected stream ----------------------------- + +/// Serves exactly one client on the connected stream of descriptor `n` +/// (socket activation, `9ns --spawn`-style handoffs), driving the engine +/// directly, the freestanding way. Returns when the client hangs up. +fn serveFd(io: Io, n: i32) !void { + const stream: Io.net.Stream = .{ .socket = .{ .handle = n, .address = .{ .ip4 = .loopback(0) } } }; + const Engine = fs.Server(tree.Harness, tree.opts); + var in: [tree.msize]u8 = undefined; + var out: [2 * tree.msize]u8 = undefined; + var rbuf: [tree.msize]u8 = undefined; + var wbuf: [2 * tree.msize]u8 = undefined; + var stage: [tree.msize]u8 = undefined; + var engine = Engine.init(.{ .in = &in, .out = &out, .root = tree.root }); + var reader = stream.reader(io, &rbuf); + var writer = stream.writer(io, &wbuf); + while (true) { + const frame = transport.readFrame(&reader.interface, &stage, tree.msize) catch return; + var off: usize = 0; + while (off < frame.len) { + off += engine.push(frame[off..]); + while (true) { + const req = engine.next() orelse break; + const a = answer(req); + engine.reply(&a.reply, a.bytes); + } + const pending = engine.output(); + writer.interface.writeAll(pending) catch return; + engine.wrote(pending.len); + if (engine.protocol.dead) return; + } + writer.interface.flush() catch return; + } +} + +/// One request for the --fd loop: same locking discipline as the runner's +/// handler (the reply bytes live in the harness's shared buffers). +fn answer(req: fs.Req) tree.Answer { + harness_mem.mutex.lockUncancelable(harness_mem.io); + defer harness_mem.mutex.unlock(harness_mem.io); + return tree.handle(&harness_mem, req); +} + +test { + _ = @import("tree.zig"); // the tree's unit tests (exclusions, ids, ...) +} + +test "main: the root override parser pins each named root" { + // Parsing is inline in run(); the mapping itself is what the tests + // rely on, and it is exercised end to end in test/e2e.sh. + _ = std.meta.stringToEnum(tree.Root, "claude").?; + _ = std.meta.stringToEnum(tree.Root, "codex").?; + _ = std.meta.stringToEnum(tree.Root, "omp").?; + _ = std.meta.stringToEnum(tree.Root, "hermes").?; + _ = std.meta.stringToEnum(tree.Root, "dsh").?; + try std.testing.expect(std.meta.stringToEnum(tree.Root, "zmx") == null); +} |