establishment: the mechanics — reply capabilities, helloExchange, lineage routing

P0 of docs/establishment-planes-plan.md; no behavior changes yet, nothing
sets the new flag or sends a hello capability.

- service.run gains a reply-capability out-slot (replyWithCapability), the
  registry idiom init already uses, lifted into the harness; null stays the
  untouched common path. Subscribers gains claimArrival() so a provider
  handler can keep a turn capability through the same flag the reserved
  subscribe uses.
- Hello wire struct: the padding byte becomes wants_channel — old callers
  wire-compatibly say 0, and the manager nominates a reply capability ONLY
  when asked, because a capability sent to a caller that never reads one is
  a leaked slot in that caller's table.
- driver.helloExchange: one handshake can hand a serving endpoint up and
  receive the device's provider channel down (usb-storage will need both at
  once). No channel in the reply is retryable, never a verdict.
- The manager stores each instance's serving endpoint on its Driver entry,
  routes consumer hellos by lineage (child -> reporter -> endpoint), replaces
  on re-hello, and closes the stale handle on death - the 32-slot table is
  the bound that makes forgetting this boot-fatal.

library/device/driver/driver.zig

@@ -253,6 +253,29 @@ const lookup_pause_ms: u64 = 20;
 /// The device this driver was assigned is the packet's `Header.target` — the manager's
 /// object addressing, so `no_device` here is a driver that serves none.
 pub fn hello(role: Role, device_id: u64) ?ipc.Handle {
+    const exchanged = helloExchange(role, device_id, null, false) orelse return null;
+    return exchanged.manager;
+}
+
+/// What one hello moved, besides the handshake itself: the manager's endpoint
+/// (every hello), and — when asked — the channel to the driver that provides
+/// this device, shared into our handle table by the reply.
+pub const Exchange = struct {
+    manager: ipc.Handle,
+    /// The provider's channel, when `want_channel` asked and the manager's
+    /// lineage had one. Null with `want_channel` set means the provider is not
+    /// there YET (its own hello has not landed, or it is mid-restart) — a
+    /// retryable condition, never a verdict.
+    channel: ?ipc.Handle,
+};
+
+/// The full handshake (communication.md "Establishment: two planes, one
+/// namespace"): a provider hands `serving` — the endpoint its consumers will
+/// be routed to — up with the request; a consumer sets `want_channel` and
+/// receives its device's provider channel with the reply. One call can do
+/// both (usb-storage serves block and consumes usb-transfer). The kernel
+/// shares capabilities as refcounted copies, so `serving` stays ours too.
+pub fn helloExchange(role: Role, device_id: u64, serving: ?ipc.Handle, want_channel: bool) ?Exchange {
     var attempts: u32 = 0;
     const manager = while (attempts < lookup_attempts) : (attempts += 1) {
         if (channel.openEndpoint("device-manager")) |handle| break handle;
@@ -266,24 +289,26 @@ pub fn hello(role: Role, device_id: u64) ?ipc.Handle {
     const framed = device_manager_protocol.Protocol.encodeRequest(
         .hello,
         device_id,
-        .{ .role = @intFromEnum(role) },
+        .{ .role = @intFromEnum(role), .wants_channel = @intFromBool(want_channel) },
         &.{},
         &packet,
     ) orelse return null;
 
     var reply: [device_manager_protocol.message_maximum]u8 = undefined;
-    const length = ipc.call(manager, framed, &reply) catch {
+    const answered = ipc.callCap(manager, framed, &reply, serving) catch {
         std.log.info("hello call failed", .{});
         return null;
     };
-    const status = envelope.statusOf(reply[0..length]) orelse {
+    const status = envelope.statusOf(reply[0..answered.len]) orelse {
+        if (answered.cap) |stray| _ = ipc.close(stray); // never keep what we cannot read
         std.log.info("hello answered nothing readable", .{});
         return null;
     };
     if (status.status != 0) {
+        if (answered.cap) |stray| _ = ipc.close(stray);
         std.log.info("hello refused", .{});
         return null;
     }
     std.log.info("hello acknowledged", .{});
-    return manager;
+    return .{ .manager = manager, .channel = answered.cap };
 }

library/kernel/service.zig

@@ -192,6 +192,14 @@ pub fn Subscribers(comptime Protocol: type, comptime Context: type) type {
             return false;
         }
 
+        /// A provider's own handler kept this turn's capability (stored it
+        /// somewhere with a lifetime beyond the turn) — the same claim the
+        /// reserved `subscribe` makes for its slot table. Composes with it:
+        /// one flag, one `take()`, whoever claims first wins the turn.
+        pub fn claimArrival() void {
+            claimed = true;
+        }
+
         /// Answer one received packet, with the reserved `subscribe` and
         /// `unsubscribe` verbs already wired — a provider that leaves those two
         /// handlers null (every provider should) gets the harness's. The turn's
@@ -296,6 +304,22 @@ pub fn Subscribers(comptime Protocol: type, comptime Context: type) type {
 /// the clean exit the supervisor reads as `ExitReason.exited`.
 /// `maximum_message` sizes the receive and reply buffers (a service passes its
 /// protocol's message maximum).
+/// The capability the current turn's handler nominates to ride out with its
+/// reply — init's registry idiom (`pending_capability`), lifted into the
+/// harness so any service can answer an establishment request with a channel
+/// (communication.md "Establishment: two planes"). Consumed by the loop at the
+/// very next `replyWait`, which is the reply this turn owes; null is the
+/// untouched common path. The kernel shares the endpoint as a refcounted copy,
+/// so the nominating service keeps its own handle.
+var pending_reply_capability: ?ipc.Handle = null;
+
+/// Called from inside an `on_message` handler: send `handle` with this turn's
+/// reply. One capability per turn — the last nomination wins, matching the
+/// transport (a reply carries at most one).
+pub fn replyWithCapability(handle: ipc.Handle) void {
+    pending_reply_capability = handle;
+}
+
 pub fn run(comptime maximum_message: usize, callbacks: Callbacks) void {
     const endpoint = ipc.createIpcEndpoint() orelse return;
     if (callbacks.service) |name| {
@@ -313,7 +337,12 @@ pub fn run(comptime maximum_message: usize, callbacks: Callbacks) void {
     var reply_len: usize = 0;
     var receive: [maximum_message]u8 = undefined;
     while (true) {
-        const got = ipc.replyWait(endpoint, reply_buffer[0..reply_len], &receive, null);
+        // The reply going out is the one the just-run handler wrote, so the
+        // capability it nominated (if any) rides this exact replyWait and is
+        // reset before the next turn can see a stale one.
+        const reply_capability = pending_reply_capability;
+        pending_reply_capability = null;
+        const got = ipc.replyWait(endpoint, reply_buffer[0..reply_len], &receive, reply_capability);
         // Whatever capability came with this turn is the turn's, and the turn
         // closes it unless a callback claims it (`ipc.Arrival`). Structural
         // rather than a close per branch, because the branches are exactly what

library/protocol/device-manager/device-manager-protocol.zig

@@ -79,7 +79,13 @@ pub const Role = enum(u8) {
 pub const Hello = extern struct {
     /// A `Role` value.
     role: u8,
-    _padding: u8 = 0,
+    /// 1 = the caller asks the reply to carry a capability to the channel of
+    /// the driver that provides the caller's device — the target's reporter;
+    /// establishment by lineage (communication.md "Establishment: two
+    /// planes"). 0 = a plain handshake, and the reply carries nothing — which
+    /// keeps every caller that never reads a reply capability from leaking
+    /// one. Was padding, so old callers wire-compatibly say 0.
+    wants_channel: u8 = 0,
     /// The protocol version this driver was built against (`version`).
     version: u16 = version,
 };

system/services/device-manager/device-manager.zig

@@ -140,6 +140,12 @@ const Driver = struct {
     spawn_ns: u64 = 0,
     hello_deadline_ns: u64 = 0,
     restart_due_ns: u64 = 0,
+    // The serving endpoint this instance handed up in its hello — what consumer
+    // hellos for its reported children are answered with (establishment by
+    // lineage, communication.md "Establishment: two planes"). A refcounted
+    // handle in OUR table: onDriverExit must close it, or every restart leaks a
+    // slot of the manager's 32 until no capability can arrive at all.
+    endpoint: ?ipc.Handle = null,
 
     fn name(driver: *const Driver) []const u8 {
         return driver.name_buffer[0..driver.name_len];
@@ -217,6 +223,19 @@ fn childCountOf(reporter: u32) u32 {
     return n;
 }
 
+/// The child a kernel device id belongs to — the lineage lookup: its
+/// `.reporter` names the driver instance that provides it, which is how a
+/// consumer's hello is routed to the right provider (communication.md
+/// "Establishment: two planes"). Null when nothing reported it, or its
+/// reporter died and pruned it — a retryable gap, not a verdict.
+fn childByDevice(device_id: u64) ?*Child {
+    if (device_id == device_manager_protocol.no_device) return null;
+    for (&children) |*child| {
+        if (child.used and child.device_id == device_id) return child;
+    }
+    return null;
+}
+
 /// The driver entry a live process id belongs to. Zero is not a process id here:
 /// it is what `onDriverExit` writes back to retire an id it has already acted on,
 /// so a second notification for the same death matches nothing.
@@ -328,6 +347,13 @@ fn onDriverExit(driver: *Driver) void {
     // Without this the backoff would count one death twice and the crash-loop cap
     // would fire at half the deaths it names.
     driver.process_id = 0;
+    // The dead instance's serving endpoint is stale the moment it died — the
+    // kernel marked the endpoint dead, but our refcounted handle would sit in
+    // the 32-slot table forever. The successor's hello stores a fresh one.
+    if (driver.endpoint) |stale| {
+        _ = ipc.close(stale);
+        driver.endpoint = null;
+    }
     pruneChildrenOf(dead);
     const reason = process.exitReason(dead) orelse .fault;
     if (reason == .exited) {
@@ -489,6 +515,30 @@ fn onHello(_: void, invocation: Invocation(device_manager_protocol.Hello), _: An
     };
     driver.state = .running;
 
+    // A provider's hello carries its serving endpoint — the channel consumer
+    // hellos for its reported children are answered with. Stored on the entry
+    // (claimed from the turn, so the harness's defer leaves it alone); a
+    // re-hello replaces, closing the old handle rather than leaking the slot.
+    if (invocation.capability) |serving| {
+        if (driver.endpoint) |previous| _ = ipc.close(previous);
+        driver.endpoint = serving;
+        Serve.claimArrival();
+    }
+
+    // A consumer's hello asks for its device's provider: the child's reporter
+    // is the routing fact (establishment by lineage). No channel is not a
+    // refusal — the provider may be mid-restart and its re-report on the way —
+    // so the hello still acks and the consumer retries. Nothing is nominated
+    // unless asked: a capability sent to a caller that never reads one is a
+    // leaked slot in ITS table.
+    if (invocation.request.wants_channel != 0) {
+        if (childByDevice(invocation.target)) |child| {
+            if (driverByProcess(child.reporter)) |provider| {
+                if (provider.endpoint) |serving| service.replyWithCapability(serving);
+            }
+        }
+    }
+
     std.log.info("hello from {s} (device {d})", .{ driver.name(), invocation.target });
     // Resilience drill (V6): once, kill the virtio-gpu driver a moment after it hellos, so
     // the normal restart policy respawns it — the compositor must survive and re-attach.