9web: multiplexer, HTTP view of the tree, live streams, richer page

One upstream 9P connection now serves any number of browser WebSocket
sessions and, with --serve, plain 9P clients over TCP or Unix (a 9pserve-
style frame remux: tags and fids remapped, Tflush forwarded, reconnect on
upstream loss). /fs/<path> maps HTTP onto the tree: GET file or directory
(JSON or HTML), Range, HEAD with 9P headers, PUT (create, truncate, append,
trailing slash makes a directory), DELETE, and ?follow=1 or text/event-stream
turning a blocking read into server-sent events with Tflush on disconnect.
The page gains a lazy tree, stat panel, create, rename, delete, upload and
follow mode. --probe embeds 9proc for self-introspection. New mux-test and
http-fs-test steps; e2e still passes.

Co-Authored-By: Claude Fable 5.1 <[email protected]>

1 files changed, 83 insertions, 0 deletions

diff --git a/web/probe.zig b/web/probe.zig
new file mode 100644
index 0000000..a6f5ceb
--- /dev/null
+++ b/web/probe.zig
@@ -0,0 +1,83 @@
+//! Optional self-introspection: 9web embeds 9proc under `--probe`, off by
+//! default. It exposes the gateway's own live counters (downstream and HTTP
+//! connection counts, the upstream state, reconnect count, negotiated msize)
+//! as files, plus 9proc's /threads view, on a Unix or TCP 9P listener. Small
+//! and read-only; serve it on a private socket.
+const std = @import("std");
+const builtin = @import("builtin");
+const build_options = @import("build_options");
+
+const enabled = build_options.have_probe and builtin.os.tag == .linux;
+
+pub fn Probe(comptime App: type) type {
+    if (!enabled) return struct {
+        pub fn start(_: *App, _: std.mem.Allocator, _: []const u8) !void {
+            return error.Unsupported;
+        }
+        pub fn stop() void {}
+    };
+
+    const proc9 = @import("9proc");
+    const Writer = std.Io.Writer;
+
+    return struct {
+        const Fns = struct {
+            fn app(ctx: *anyopaque) *App {
+                return @ptrCast(@alignCast(ctx));
+            }
+            pub fn connections(ctx: *anyopaque, w: *Writer) anyerror!void {
+                try w.print("{d}", .{app(ctx).connectionCount()});
+            }
+            pub fn downstreams(ctx: *anyopaque, w: *Writer) anyerror!void {
+                try w.print("{d}", .{app(ctx).m.downstreamCount()});
+            }
+            pub fn upstream(ctx: *anyopaque, w: *Writer) anyerror!void {
+                try w.writeAll(app(ctx).m.upstreamState());
+            }
+            pub fn reconnects(ctx: *anyopaque, w: *Writer) anyerror!void {
+                try w.print("{d}", .{app(ctx).m.reconnectCount()});
+            }
+            pub fn negotiated(ctx: *anyopaque, w: *Writer) anyerror!void {
+                try w.print("{d}", .{app(ctx).m.negotiatedMsize()});
+            }
+        };
+
+        const cfg: proc9.Config = .{
+            .name = "9web",
+            .fns = Fns,
+            .msize = 16 * 1024,
+            .max_fids = 64,
+            .max_providers = 8,
+        };
+        const S = proc9.Server(cfg);
+        const ProbeT = proc9.linux.Probe(S);
+        const max_clients = 4;
+
+        var shared: S.Shared = undefined;
+        var storage: ProbeT.Storage(max_clients) = undefined;
+        var probe: ProbeT = undefined;
+        var running: bool = false;
+
+        pub fn start(app: *App, allocator: std.mem.Allocator, addr_text: []const u8) !void {
+            _ = allocator;
+            const listen: proc9.linux.Listen = if (std.mem.startsWith(u8, addr_text, "unix:"))
+                .{ .unix = addr_text[5..] }
+            else if (std.mem.startsWith(u8, addr_text, "tcp:"))
+                .{ .tcp = addr_text[4..] }
+            else
+                return error.BadProbeAddress;
+            shared = .init(app);
+            try probe.init(&shared, &storage, .{ .io = app.io, .listen = listen, .msize = cfg.msize });
+            try probe.start();
+            running = true;
+            std.debug.print("9web: probe on {s}\n", .{addr_text});
+        }
+
+        pub fn stop() void {
+            if (running) {
+                probe.stop();
+                running = false;
+            }
+        }
+    };
+}