feat(remote): add inbound proxy table to endpoint

- Spawn a local Proxy actor on first message from each new remote actor ID
- Route inbound wire exit to the corresponding proxy
- Clean up proxy table entries on proxy_exit notification
- Teardown sends {"exit", "transport_closed"} to all live proxies

src/remote/endpoint.zig

@@ -1,10 +1,12 @@
 /// Parent-side endpoint actor.
 ///
 /// Wraps a subprocess, accumulates framed CBOR from its stdout, and dispatches
-/// decoded send_named messages to named local actors via the env proc table.
+/// decoded messages to local actors or proxies.
 ///
-/// Local actors send outbound messages to the endpoint with:
+/// Outbound messages (local → endpoint):
-///   {"send", from_id: u64, to_name: text, payload: raw_cbor}
+///   {"send", from_id: u64, to_id: u64, payload}    — via proxy, by remote ID
+///   {"send", from_id: u64, to_name: text, payload} — direct by name
+///   {"proxy_exit", remote_id: u64, reason: text}   — proxy notifying of exit
 const std = @import("std");
 const tp = @import("thespian");
 const cbor = @import("cbor");
@@ -27,6 +29,8 @@ const Endpoint = struct {
     allocator: std.mem.Allocator,
     proc: subprocess,
     accumulator: framing.Accumulator,
+    /// Inbound proxy table: remote actor ID → local proxy pid.
+    proxies: std.AutoHashMap(u64, tp.pid),
     receiver: tp.Receiver(*@This()),
 
     fn start(args: Args) tp.result {
@@ -41,6 +45,7 @@ const Endpoint = struct {
             .allocator = args.allocator,
             .proc = proc,
             .accumulator = .{},
+            .proxies = std.AutoHashMap(u64, tp.pid).init(args.allocator),
             .receiver = .init(receive_fn, deinit, self),
         };
         errdefer self.deinit();
@@ -48,6 +53,13 @@ const Endpoint = struct {
     }
 
     fn deinit(self: *@This()) void {
+        // Exit all live proxies (best-effort; endpoint may already be shutting down).
+        var it = self.proxies.valueIterator();
+        while (it.next()) |p| {
+            p.send(.{ "exit", "transport_closed" }) catch {};
+            p.deinit();
+        }
+        self.proxies.deinit();
         self.proc.deinit();
         self.allocator.destroy(self);
     }
@@ -62,6 +74,8 @@ const Endpoint = struct {
         var to_id: u64 = 0;
         var to_name: []const u8 = "";
         var payload: []const u8 = "";
+        var remote_id: u64 = 0;
+
         if (try m.match(.{ proc_tag, "stdout", tp.extract(&bytes) })) {
             if (self.accumulator.feed(bytes)) |frame| try self.dispatch_inbound(frame);
         } else if (try m.match(.{ proc_tag, "stderr", tp.any })) {
@@ -71,10 +85,9 @@ const Endpoint = struct {
         } else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_name), cbor.extract_cbor(&payload) })) {
             // Outbound send_named — from a local actor addressing a well-known remote actor by name.
             try self.send_wire_named(from_id, to_name, payload);
-        } else if (try m.match(.{ "proxy_exit", tp.any, tp.any })) {
+        } else if (try m.match(.{ "proxy_exit", tp.extract(&remote_id), tp.any })) {
-            // A local proxy has exited; remove it from the proxy table (stub until table is implemented).
+            // A local proxy has exited; remove it from the proxy table.
-        } else if (try m.match(.{ proc_tag, "term", "exited", tp.any })) {
+            if (self.proxies.fetchRemove(remote_id)) |entry| entry.value.deinit();
-            return tp.exit("transport_closed");
         } else if (try m.match(.{ proc_tag, "term", tp.any, tp.any })) {
             return tp.exit("transport_closed");
         } else {
@@ -82,18 +95,42 @@ const Endpoint = struct {
         }
     }
 
-    fn dispatch_inbound(_: *@This(), frame: []const u8) !void {
+    /// Get the local proxy for remote_id, creating one if it doesn't exist yet.
+    fn get_or_create_proxy(self: *@This(), remote_id: u64) !tp.pid_ref {
+        if (self.proxies.getPtr(remote_id)) |p| return p.ref();
+        const p = try tp.spawn(self.allocator, proxy.Args{
+            .allocator = self.allocator,
+            .endpoint = tp.self_pid().clone(),
+            .remote_id = remote_id,
+        }, proxy.start, "proxy");
+        try self.proxies.put(remote_id, p);
+        return self.proxies.getPtr(remote_id).?.ref();
+    }
+
+    fn dispatch_inbound(self: *@This(), frame: []const u8) !void {
         const msg = try protocol.decode(frame);
         switch (msg) {
             .send => |s| {
-                _ = s;
+                // Ensure a local proxy exists for the sender.
-                return tp.exit_error(error.UnexpectedSend, null);
+                _ = try self.get_or_create_proxy(s.from_id);
+                // Deliver payload to the local actor identified by to_id.
+                // Requires the outbound ID table (not yet implemented); error for now.
+                _ = s.to_id;
+                _ = s.payload;
+                return tp.exit_error(error.OutboundTableNotImplemented, null);
             },
             .send_named => |s| {
+                // Ensure a local proxy exists for the sender so it can receive replies.
+                _ = try self.get_or_create_proxy(s.from_id);
                 const actor = tp.env.get().proc(s.to_name);
                 try actor.send_raw(tp.message{ .buf = s.payload });
             },
-            .link, .exit, .proxy_id, .transport_error => return tp.exit_error(error.UnexpectedMessage, null),
+            .exit => |e| {
+                // Remote actor has exited; notify its local proxy.
+                if (self.proxies.get(e.id)) |p|
+                    p.send(.{ "exit", e.reason }) catch {};
+            },
+            .proxy_id, .link, .transport_error => return tp.exit_error(error.UnexpectedMessage, null),
         }
     }