path: root/9agents/src/main.zig

//! 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
    \\  /active/<harness>/<pid>/chat/<id>-<kind>   its conversation (claude, codex)
    \\
    \\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 (the path table) lives in
// .bss. The runner and the chat pool are larger — every connection's fid
// table, every transcript index — and each gets a mapping of its own
// (`mapped`), backed only as far as it is used.
var harness_mem: tree.Harness = undefined;

/// A `T` in its own anonymous `MAP_NORESERVE` mapping: address space until
/// written, and never written just to initialize it (a Debug build fills
/// an `undefined` global with 0xaa, which for these is hundreds of MB).
fn mapped(comptime T: type) error{OutOfMemory}!*T {
    const rc = linux.mmap(null, @sizeOf(T), .{ .READ = true, .WRITE = true }, .{
        .TYPE = .PRIVATE,
        .ANONYMOUS = true,
        .NORESERVE = true,
    }, -1, 0);
    if (linux.errno(rc) != .SUCCESS) return error.OutOfMemory;
    // Small pages: with transparent huge pages on, touching each
    // connection's few header fields would back 2 MB apiece.
    _ = linux.madvise(@ptrFromInt(rc), @sizeOf(T), linux.MADV.NOHUGEPAGE);
    return @ptrFromInt(rc);
}
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;
    }

    // The chat indexes live in their own lazily backed mapping: ~100 MB of
    // address space, none of it memory until a chat is read.
    const chats = try tree.chat.Pool.create();
    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,
        .chats = chats,
    });
    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 => {},
    }

    const runner = try mapped(Runner);
    runner.init(.{ .io = io, .root = tree.root, .handler = .{ .ctx = &harness_mem, .serve = serveReq } });
    defer runner.stop();

    var posted_something = false;
    if (!no_post) {
        runner.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.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.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;
    // Far too big for the stack at this fid capacity.
    const engine = try mapped(Engine);
    engine.initIn(.{ .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);
}