feat: l7 proxy dispatches mtls upstreams to client_dial

src/api/routes/status_metrics.zig

@@ -530,6 +530,7 @@ test "route rollout status reports steering blocker for VIP cutover readiness" {
         testing.allocator,
         "api",
         "consistent_hash",
+        "off",
         &.{
             .{
                 .route_name = "default",
@@ -626,6 +627,7 @@ test "route rollout status reports vip cutover ready after clean backfill audit
         testing.allocator,
         "api",
         "consistent_hash",
+        "off",
         &.{
             .{
                 .route_name = "default",
@@ -733,6 +735,7 @@ test "route rollout status converges from steering blocked to vip cutover ready"
         testing.allocator,
         "api",
         "consistent_hash",
+        "off",
         &.{
             .{
                 .route_name = "default",

src/manifest/orchestrator/startup_runtime.zig

@@ -162,10 +162,12 @@ pub fn syncServiceDefinitions(
             for (route_inputs.items) |route| if (route.backend_services.len > 0) alloc.free(route.backend_services);
         }
         if (route_alloc_failed) continue;
+        const peer_mode_label = if (svc.tls) |tls| tls.peer.label() else "off";
         const record = store.syncServiceConfig(
             alloc,
             svc.name,
             "consistent_hash",
+            peer_mode_label,
             route_inputs.items,
         ) catch |err| {
             log.warn("orchestrator: failed to sync service definition for {s}: {}", .{ svc.name, err });

src/network/proxy/reverse_proxy.zig

@@ -19,6 +19,8 @@ const router = @import("router.zig");
 const upstream_mod = @import("upstream.zig");
 const upstream_pool = @import("upstream_pool.zig");
 const runtime_wait = @import("../../lib/runtime_wait.zig");
+const client_dial = @import("../../tls/client_dial.zig");
+const store_mod = @import("../../state/store.zig");
 
 const proxy_loop_header = "X-Yoq-Proxy";
 const x_forwarded_for_header = "X-Forwarded-For";
@@ -616,6 +618,10 @@ pub const ReverseProxy = struct {
         const request = try self.buildForwardRequestWithClient(raw_request, plan, client_ip);
         defer self.allocator.free(request);
 
+        if (upstream.peer_mode != .off) {
+            return self.forwardSingleAttemptMtls(request, plan, upstream);
+        }
+
         // prefer a pooled, kept-alive connection; dial a fresh one on a miss.
         var from_pool = true;
         var fd = upstream_pool.checkout(upstream.endpoint_id, upstream.address, upstream.port) orelse blk: {
@@ -653,8 +659,128 @@ pub const ReverseProxy = struct {
         }
         return result.bytes;
     }
+
+    /// mTLS-specific dial + request/response. always opens a fresh
+    /// connection (TLS sessions hold encryption state and can't be pooled
+    /// alongside bare-fd siblings keyed on the same address). on `.warn`
+    /// a missing/invalid cluster CA logs and downgrades to plaintext;
+    /// `.require` returns an error in the same case so the caller can
+    /// surface the failure (PR 5 finale will plug this into proper
+    /// observability).
+    fn forwardSingleAttemptMtls(
+        self: *const ReverseProxy,
+        request: []const u8,
+        plan: *const ForwardPlan,
+        upstream: *const upstream_mod.Upstream,
+    ) ![]u8 {
+        const ca_rec_opt = store_mod.getClusterCa(self.allocator) catch null;
+        const ca_rec = ca_rec_opt orelse {
+            if (upstream.peer_mode == .require) return error.ClusterCaMissing;
+            log.warn("mtls upstream {s}: cluster CA not seeded, downgrading to plain dial", .{upstream.address});
+            return self.forwardPlainAttempt(request, plan, upstream);
+        };
+        defer ca_rec.deinit(self.allocator);
+
+        var outcome = client_dial.dial(std.Options.debug_io, self.allocator, .{
+            .address = upstream.address,
+            .port = upstream.port,
+            .connect_timeout_ms = plan.route.connect_timeout_ms,
+            .ca_cert_pem = ca_rec.cert_pem,
+            .server_name = upstream.service,
+            .now_unix = std.Io.Clock.real.now(std.Options.debug_io).toSeconds(),
+        }) catch |err| return err;
+
+        switch (outcome) {
+            .bare => |fd| {
+                // dial returned plaintext somehow (shouldn't happen when
+                // ca_cert_pem is set, but be defensive).
+                defer linux_platform.posix.close(fd);
+                return error.HandshakeFailed;
+            },
+            .session => |*sess| {
+                defer {
+                    sess.deinit();
+                    linux_platform.posix.close(sess.fd);
+                }
+
+                _ = sess.write(request) catch return error.SendFailed;
+                return try readResponseFromSession(self.allocator, sess, self.max_response_bytes);
+            },
+        }
+    }
+
+    /// fallback used by the mTLS `.warn` path when no cluster CA is
+    /// available — runs the plaintext dial/pool path so the connection
+    /// at least succeeds. equivalent to the legacy non-mTLS leg of
+    /// forwardSingleAttempt minus the request build (caller already has
+    /// the bytes).
+    fn forwardPlainAttempt(
+        self: *const ReverseProxy,
+        request: []const u8,
+        plan: *const ForwardPlan,
+        upstream: *const upstream_mod.Upstream,
+    ) ![]u8 {
+        var from_pool = true;
+        var fd = upstream_pool.checkout(upstream.endpoint_id, upstream.address, upstream.port) orelse blk: {
+            from_pool = false;
+            const dialed = try socket_helpers.connectToUpstream(plan.route.connect_timeout_ms, plan.route.request_timeout_ms, upstream);
+            upstream_pool.noteDialed();
+            break :blk dialed;
+        };
+
+        socket_helpers.writeAll(fd, request) catch {
+            upstream_pool.discard(fd);
+            if (!from_pool) return error.SendFailed;
+            from_pool = false;
+            fd = try socket_helpers.connectToUpstream(plan.route.connect_timeout_ms, plan.route.request_timeout_ms, upstream);
+            upstream_pool.noteDialed();
+            socket_helpers.writeAll(fd, request) catch {
+                upstream_pool.discard(fd);
+                return error.SendFailed;
+            };
+        };
+
+        const result = readResponse(self.allocator, fd, self.max_response_bytes, plan.method == .HEAD) catch |err| {
+            upstream_pool.discard(fd);
+            return err;
+        };
+
+        if (result.reusable) {
+            upstream_pool.release(upstream.endpoint_id, upstream.address, upstream.port, fd);
+        } else {
+            upstream_pool.discard(fd);
+        }
+        return result.bytes;
+    }
 };
 
+/// drain an mTLS session into a single buffer up to `max_bytes`. mTLS
+/// connections aren't pooled, so we don't need the framing-aware "is
+/// this connection still reusable" logic that the bare-fd readResponse
+/// provides — we just read until the peer closes (or `PeerClosed`
+/// surfaces via the session) and hand back the bytes. PeerClosed is
+/// the orderly EOF signal; any other read error propagates.
+fn readResponseFromSession(
+    alloc: std.mem.Allocator,
+    sess: anytype,
+    max_bytes: usize,
+) ![]u8 {
+    var out: std.ArrayList(u8) = .empty;
+    errdefer out.deinit(alloc);
+
+    var chunk: [8192]u8 = undefined;
+    while (true) {
+        const n = sess.read(&chunk) catch |err| {
+            if (err == error.PeerClosed) break;
+            return err;
+        };
+        if (n == 0) break;
+        if (out.items.len + n > max_bytes) return error.ResponseTooLarge;
+        try out.appendSlice(alloc, chunk[0..n]);
+    }
+    return try out.toOwnedSlice(alloc);
+}
+
 fn cloneMirrorTask(
     alloc: std.mem.Allocator,
     active_mirror_requests: *std.atomic.Value(usize),
@@ -4761,3 +4887,52 @@ test "handleConnection rejects looped request after listener restart" {
     linux_platform.posix.close(client_fd);
     server_thread.join();
 }
+
+// --- readResponseFromSession ---
+//
+// the mTLS read loop drains a session-shaped reader until either the peer
+// closes (surfaced as ClientSession.read's `PeerClosed`) or `max_bytes` is
+// exceeded. exercised here via a small fake session that implements the
+// duck-typed `.read([]u8) !usize` shape — keeps the test independent of
+// the live TLS stack while still proving the loop's framing.
+
+const FakeSession = struct {
+    chunks: []const []const u8,
+    index: usize = 0,
+    closed: bool = false,
+
+    fn read(self: *FakeSession, buf: []u8) !usize {
+        if (self.closed) return error.PeerClosed;
+        if (self.index >= self.chunks.len) {
+            self.closed = true;
+            return error.PeerClosed;
+        }
+        const chunk = self.chunks[self.index];
+        self.index += 1;
+        const n = @min(buf.len, chunk.len);
+        @memcpy(buf[0..n], chunk[0..n]);
+        return n;
+    }
+};
+
+test "readResponseFromSession concatenates chunks until peer closes" {
+    var sess = FakeSession{ .chunks = &.{ "HTTP/1.1 200 OK\r\n", "Content-Length: 2\r\n\r\nok" } };
+    const body = try readResponseFromSession(std.testing.allocator, &sess, 64 * 1024);
+    defer std.testing.allocator.free(body);
+    try std.testing.expectEqualStrings("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", body);
+}
+
+test "readResponseFromSession returns ResponseTooLarge when max_bytes is exceeded" {
+    var sess = FakeSession{ .chunks = &.{ "AAAA", "BBBB", "CCCC" } };
+    try std.testing.expectError(
+        error.ResponseTooLarge,
+        readResponseFromSession(std.testing.allocator, &sess, 6),
+    );
+}
+
+test "readResponseFromSession returns empty buffer on immediate close" {
+    var sess = FakeSession{ .chunks = &.{} };
+    const body = try readResponseFromSession(std.testing.allocator, &sess, 64);
+    defer std.testing.allocator.free(body);
+    try std.testing.expectEqual(@as(usize, 0), body.len);
+}

src/network/proxy/runtime.zig

@@ -699,6 +699,7 @@ pub fn resolveUpstreamWithPolicy(alloc: std.mem.Allocator, service_name: []const
 
     const now_ms = nowRealMilliseconds();
     const target_port = service.http_proxy_target_port;
+    const peer_mode = service.peer_mode;
     for (endpoints.items) |endpoint| {
         const port: u16 = target_port orelse if (endpoint.port < 0) 0 else @intCast(endpoint.port);
         try candidates.append(alloc, .{
@@ -707,6 +708,7 @@ pub fn resolveUpstreamWithPolicy(alloc: std.mem.Allocator, service_name: []const
             .address = try alloc.dupe(u8, endpoint.ip_address),
             .port = port,
             .eligible = endpoint.eligible and endpointAllowsRequestLocked(endpoint.endpoint_id, now_ms, cb_policy),
+            .peer_mode = peer_mode,
         });
     }
 
@@ -717,6 +719,7 @@ pub fn resolveUpstreamWithPolicy(alloc: std.mem.Allocator, service_name: []const
         .address = try alloc.dupe(u8, selected.address),
         .port = selected.port,
         .eligible = selected.eligible,
+        .peer_mode = selected.peer_mode,
     };
 }
 

src/network/proxy/upstream.zig

@@ -1,11 +1,18 @@
 const std = @import("std");
+const spec = @import("../../manifest/spec.zig");
 
 pub const Upstream = struct {
     service: []const u8,
     endpoint_id: []const u8,
     address: []const u8,
     port: u16,
     eligible: bool = true,
+    /// service-to-service mTLS posture, copied from the service's manifest.
+    /// when `.off` the dial stays plaintext and uses the existing connection
+    /// pool; when `.warn` or `.require` the dial runs an mTLS handshake and
+    /// the resulting session is **not pooled** (each session carries its own
+    /// encryption state — sharing it across requests is unsafe).
+    peer_mode: spec.TlsConfig.PeerMode = .off,
 
     pub fn deinit(self: Upstream, alloc: std.mem.Allocator) void {
         alloc.free(self.service);

src/network/service_registry.zig

@@ -1,4 +1,5 @@
 const std = @import("std");
+const spec = @import("../manifest/spec.zig");
 
 const Allocator = std.mem.Allocator;
 
@@ -87,6 +88,9 @@ pub const ServiceDefinition = struct {
     http_proxy_circuit_breaker_threshold: ?u8 = null,
     http_proxy_circuit_breaker_timeout_ms: ?u32 = null,
     http_proxy_mirror_service: ?[]const u8 = null,
+    /// service-to-service mTLS posture for traffic *to* this service.
+    /// copied verbatim from the manifest's `service.<name>.tls.peer`.
+    peer_mode: spec.TlsConfig.PeerMode = .off,
 };
 
 pub const HttpRouteDefinition = struct {
@@ -192,6 +196,7 @@ pub const ServiceSnapshot = struct {
     http_proxy_circuit_breaker_threshold: ?u8,
     http_proxy_circuit_breaker_timeout_ms: ?u32,
     http_proxy_mirror_service: ?[]const u8,
+    peer_mode: spec.TlsConfig.PeerMode = .off,
     total_endpoints: usize,
     eligible_endpoints: usize,
     healthy_endpoints: usize,
@@ -293,6 +298,7 @@ const ServiceState = struct {
     http_proxy_circuit_breaker_threshold: ?u8 = null,
     http_proxy_circuit_breaker_timeout_ms: ?u32 = null,
     http_proxy_mirror_service: ?[]const u8 = null,
+    peer_mode: spec.TlsConfig.PeerMode = .off,
     endpoints: std.ArrayList(EndpointState) = .empty,
     last_reconcile_status: ReconcileStatus = .idle,
     last_reconcile_error: ?[]const u8 = null,
@@ -673,6 +679,7 @@ fn cloneServiceSnapshot(alloc: Allocator, service: *const ServiceState) Error!Se
         .http_proxy_circuit_breaker_threshold = service.http_proxy_circuit_breaker_threshold,
         .http_proxy_circuit_breaker_timeout_ms = service.http_proxy_circuit_breaker_timeout_ms,
         .http_proxy_mirror_service = if (service.http_proxy_mirror_service) |mirror_service| try alloc.dupe(u8, mirror_service) else null,
+        .peer_mode = service.peer_mode,
         .total_endpoints = total_endpoints,
         .eligible_endpoints = eligible_endpoints,
         .healthy_endpoints = healthy_endpoints,
@@ -812,6 +819,7 @@ fn assignCompatProxyFields(alloc: Allocator, service: *ServiceState, definition:
     service.http_proxy_circuit_breaker_threshold = definition.http_proxy_circuit_breaker_threshold;
     service.http_proxy_circuit_breaker_timeout_ms = definition.http_proxy_circuit_breaker_timeout_ms;
     try replaceOptionalOwned(alloc, &service.http_proxy_mirror_service, definition.http_proxy_mirror_service);
+    service.peer_mode = definition.peer_mode;
 }
 
 fn isEndpointEligible(endpoint: *const EndpointState) bool {

src/network/service_registry_runtime.zig

@@ -5,6 +5,7 @@ const service_registry_backfill = @import("service_registry_backfill.zig");
 const rollout = @import("service_rollout.zig");
 const service_registry = @import("service_registry.zig");
 const store = @import("../state/store.zig");
+const spec = @import("../manifest/spec.zig");
 
 const Allocator = std.mem.Allocator;
 
@@ -299,6 +300,7 @@ fn serviceDefinitionFromRecord(service: *const store.ServiceRecord, route_defini
         .http_proxy_circuit_breaker_threshold = if (service.http_proxy_circuit_breaker_threshold) |v| @intCast(v) else null,
         .http_proxy_circuit_breaker_timeout_ms = if (service.http_proxy_circuit_breaker_timeout_ms) |v| @intCast(v) else null,
         .http_proxy_mirror_service = service.http_proxy_mirror_service,
+        .peer_mode = if (service.peer_mode) |label| (spec.TlsConfig.PeerMode.parse(label) orelse .off) else .off,
     };
 }
 

src/state/schema/migrations.zig

@@ -49,6 +49,7 @@ fn migrateServices(db: *sqlite.Db) void {
     addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN http_proxy_retry_on_5xx INTEGER;") catch {};
     addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN http_proxy_circuit_breaker_threshold INTEGER;") catch {};
     addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN http_proxy_circuit_breaker_timeout_ms INTEGER;") catch {};
+    addColumnIfMissing(db, "ALTER TABLE services ADD COLUMN peer_mode TEXT DEFAULT 'off';") catch {};
     createTableIfMissing(db,
         \\CREATE TABLE IF NOT EXISTS service_http_routes (
         \\    service_name TEXT NOT NULL,

src/state/schema/tables.zig

@@ -79,6 +79,7 @@ pub fn initCoreTables(db: *sqlite.Db) SchemaError!void {
         \\    http_proxy_http2_idle_timeout_ms INTEGER,
         \\    http_proxy_target_port INTEGER,
         \\    http_proxy_preserve_host INTEGER,
+        \\    peer_mode TEXT DEFAULT 'off',
         \\    created_at INTEGER NOT NULL,
         \\    updated_at INTEGER NOT NULL
         \\);