feat(relay): route forwarding through ForwardTcp

Author: pimlock

.agents/skills/debug-openshell-cluster/SKILL.md

@@ -128,6 +128,10 @@ helm -n openshell get values openshell | grep -E 'repository|tag|supervisorImage
 
 The gateway image and `server.supervisorImage` should use the same build tag in branch and E2E deploys. A stale supervisor image can make sandbox behavior lag behind gateway policy or proto changes.
 
+For local/external pull mode (the default local path via `mise run cluster`), local images are tagged to the configured local registry base, pushed to that registry, and pulled by k3s via the `registries.yaml` mirror endpoint. The `cluster` task rebuilds the local gateway image before tagging and pushing it, so a fresh bootstrap should not reuse stale `openshell/gateway:dev` bits from a previous source revision.
+
+Gateway image builds stage a partial Rust workspace from `deploy/docker/Dockerfile.images`. If cargo fails with a missing manifest under `/build/crates/...`, or an imported symbol exists locally but is missing in the image build, verify that every current gateway dependency crate is copied into the staged workspace there.
+
 For plaintext local evaluation, confirm the chart has:
 
 ```bash

architecture/gateway.md

@@ -9,11 +9,12 @@ workloads.
 
 - Authenticate clients and sandbox callbacks.
 - Serve gRPC APIs for sandbox lifecycle, provider management, policy updates,
-  settings, inference configuration, logs, and watch streams.
-- Serve HTTP endpoints for health, SSH tunnel upgrades, and edge-auth flows.
+  settings, inference configuration, logs, watch streams, and relay forwarding.
+- Serve HTTP endpoints for health, WebSocket tunnels, and edge-auth flows.
 - Persist domain objects in SQLite or Postgres.
 - Resolve provider credentials and inference bundles for sandbox supervisors.
-- Coordinate supervisor relay sessions for connect, exec, and file sync.
+- Coordinate supervisor relay sessions for connect, exec, file sync, and
+  service forwarding.
 
 The gateway does not enforce agent network policy at request time. That happens
 inside each sandbox, where the supervisor and proxy can observe local process
@@ -44,7 +45,7 @@ The gateway API is organized around platform objects and operational streams:
 
 | Area | Examples |
 |---|---|
-| Sandbox lifecycle | Create, list, delete, watch, exec, SSH session bootstrap. |
+| Sandbox lifecycle | Create, list, delete, watch, exec, SSH session bootstrap, ForwardTcp service forwarding. |
 | Providers | Store provider records, discover credentials, resolve runtime environment. |
 | Policy and settings | Get effective sandbox config, update sandbox policy, manage global settings. |
 | Inference | Set gateway-level model/provider config and resolve sandbox route bundles. |
@@ -115,22 +116,35 @@ sequenceDiagram
     participant CLI
     participant GW as Gateway
     participant SUP as Sandbox supervisor
-    participant SSH as Sandbox SSH socket
+    participant Target as Sandbox target
 
     SUP->>GW: ConnectSupervisor stream
-    CLI->>GW: connect / exec / sync request
-    GW->>SUP: RelayOpen(channel)
+    CLI->>GW: ForwardTcp / exec / sync request
+    GW->>SUP: RelayOpen(channel, target)
+    SUP->>Target: Dial SSH socket or loopback service
     SUP->>GW: RelayStream(channel)
-    SUP->>SSH: Bridge bytes to Unix socket
     CLI->>GW: Client bytes
     GW-->>CLI: Client bytes
     GW->>SUP: Relay bytes
     SUP-->>GW: Relay bytes
 ```
 
-The same relay pattern backs interactive SSH, command execution, and file sync.
-The gateway tracks live sessions in memory and persists session records so
-tokens can expire or be revoked.
+The same relay pattern backs interactive SSH, command execution, file sync, and
+local service forwarding. The gateway tracks live sessions in memory and
+persists session records so tokens can expire or be revoked.
+
+`ForwardTcp` is the client-facing byte stream for SSH and service forwarding.
+The first frame is a `TcpForwardInit` that carries the sandbox ID, an
+authorization token from `CreateSshSession`, and an explicit target:
+`target.ssh` for the sandbox SSH socket or `target.tcp` for a loopback service
+inside the sandbox. The gateway validates the token and sandbox readiness,
+sends a targeted `RelayOpen` to the supervisor, then bridges
+`TcpForwardFrame::Data` to `RelayFrame::Data` until either side closes.
+
+For `target.tcp`, the gateway only accepts loopback destinations such as
+`localhost`, `127.0.0.0/8`, or `::1`. The gateway never needs to know or dial a
+sandbox pod IP; supervisors connect outbound and bridge only the explicit target
+requested for that relay.
 
 ## PKI Bootstrap
 
@@ -143,13 +157,13 @@ created. Both deployment paths use it:
 | Filesystem | `--output-dir <DIR>` | `<dir>/{ca.crt, ca.key, server/tls.{crt,key}, client/tls.{crt,key}}`. Also copies client materials to `$XDG_CONFIG_HOME/openshell/gateways/openshell/mtls/` for CLI auto-discovery. |
 
 On Kubernetes, the Helm chart runs the command via a pre-install/pre-upgrade
-hook Job using the gateway image itself — no separate cert-generation image,
+hook Job using the gateway image itself -- no separate cert-generation image,
 no extra mirror burden in air-gapped environments. On the RPM gateway, the
 same command runs from the systemd unit's `ExecStartPre` to bootstrap PKI
 into the user's state directory on first start.
 
-Both modes share the same idempotency contract: all targets present → skip;
-partial state → fail with a recovery hint; nothing present → generate and
+Both modes share the same idempotency contract: all targets present -> skip;
+partial state -> fail with a recovery hint; nothing present -> generate and
 write. This guards mTLS continuity across restarts and upgrades while still
 recovering cleanly if an operator deletes everything and starts over.
 

crates/openshell-cli/Cargo.toml

@@ -68,6 +68,7 @@ tokio-tungstenite = { workspace = true }
 
 # Streams
 futures = { workspace = true }
+tokio-stream = { workspace = true }
 nix = { workspace = true }
 
 # URL parsing

crates/openshell-cli/src/main.rs

@@ -8,6 +8,7 @@ use clap_complete::engine::ArgValueCompleter;
 use clap_complete::env::CompleteEnv;
 use miette::Result;
 use owo_colors::OwoColorize;
+use std::collections::HashMap;
 use std::io::Write;
 use std::path::PathBuf;
 
@@ -198,6 +199,7 @@ const HELP_TEMPLATE: &str = "\
 \x1b[1mSANDBOX COMMANDS\x1b[0m
   sandbox:     Manage sandboxes
   forward:     Manage port forwarding to a sandbox
+  service:     Forward sandbox services over gRPC
   logs:        View sandbox logs
   policy:      Manage sandbox policy
   settings:    Manage sandbox and global settings
@@ -270,6 +272,12 @@ const FORWARD_EXAMPLES: &str = "\x1b[1mALIAS\x1b[0m
   $ openshell forward list
 ";
 
+const SERVICE_EXAMPLES: &str = "\x1b[1mEXAMPLES\x1b[0m
+  $ openshell service forward my-sandbox --target-port 8080
+  $ openshell service forward my-sandbox --target-port 5432 --local 15432
+  $ openshell service forward my-sandbox --target-port 3000 --local 127.0.0.1:0
+";
+
 const LOGS_EXAMPLES: &str = "\x1b[1mALIAS\x1b[0m
   lg
 
@@ -407,6 +415,13 @@ enum Commands {
         command: Option<ForwardCommands>,
     },
 
+    /// Forward sandbox services over gRPC.
+    #[command(after_help = SERVICE_EXAMPLES, help_template = SUBCOMMAND_HELP_TEMPLATE)]
+    Service {
+        #[command(subcommand)]
+        command: Option<ServiceCommands>,
+    },
+
     /// View sandbox logs.
     #[command(alias = "lg", after_help = LOGS_EXAMPLES, help_template = LEAF_HELP_TEMPLATE, next_help_heading = "FLAGS")]
     Logs {
@@ -1613,6 +1628,29 @@ enum ForwardCommands {
     List,
 }
 
+#[derive(Subcommand, Debug)]
+enum ServiceCommands {
+    /// Forward a local TCP port to a loopback service inside a sandbox over gRPC.
+    #[command(help_template = LEAF_HELP_TEMPLATE, next_help_heading = "FLAGS")]
+    Forward {
+        /// Sandbox name (defaults to last-used sandbox).
+        #[arg(add = ArgValueCompleter::new(completers::complete_sandbox_names))]
+        name: Option<String>,
+
+        /// Target service port inside the sandbox.
+        #[arg(long)]
+        target_port: u16,
+
+        /// Target service host inside the sandbox. Phase 1 accepts loopback only.
+        #[arg(long, default_value = "127.0.0.1")]
+        target_host: String,
+
+        /// Local bind address and port: [bind_address:]port. Use port 0 for dynamic assignment.
+        #[arg(long)]
+        local: Option<String>,
+    },
+}
+
 #[tokio::main]
 #[allow(clippy::large_stack_frames)] // CLI dispatch holds many futures; OK at top level.
 async fn main() -> Result<()> {
@@ -1777,6 +1815,38 @@ async fn main() -> Result<()> {
             }
         }
 
+        Some(Commands::Service {
+            command:
+                Some(ServiceCommands::Forward {
+                    name,
+                    target_port,
+                    target_host,
+                    local,
+                }),
+        }) => {
+            let ctx = resolve_gateway(&cli.gateway, &cli.gateway_endpoint)?;
+            let mut tls = tls.with_gateway_name(&ctx.name);
+            apply_edge_auth(&mut tls, &ctx.name);
+            let name = resolve_sandbox_name(name, &ctx.name)?;
+            run::service_forward_tcp(
+                &ctx.endpoint,
+                &name,
+                local.as_deref(),
+                &target_host,
+                target_port,
+                &tls,
+            )
+            .await?;
+        }
+
+        Some(Commands::Service { command: None }) => {
+            Cli::command()
+                .find_subcommand_mut("service")
+                .expect("service subcommand exists")
+                .print_help()
+                .expect("Failed to print help");
+        }
+
         // -----------------------------------------------------------
         // Top-level forward (was `sandbox forward`)
         // -----------------------------------------------------------
@@ -2236,7 +2306,7 @@ async fn main() -> Result<()> {
                     };
 
                     // Parse --label flags into a HashMap<String, String>.
-                    let mut labels_map = std::collections::HashMap::new();
+                    let mut labels_map = HashMap::new();
                     for label_str in &labels {
                         let parts: Vec<&str> = label_str.splitn(2, '=').collect();
                         if parts.len() != 2 {