Add bridge-ready Iroh worker integration

Author: rustyconover

README.md

@@ -133,6 +133,9 @@ The `launch:` location scheme starts the worker once behind a flock-coordinated
 
 ## Example worker
 
+For authenticated `iroh://` and HTTP-semantics `httpi://` deployments, see
+[Iroh workers and clients](docs/iroh.md).
+
 The [`vgi-example-worker`](vgi-example-worker/) module (not published) is a complete worker with 90+ functions — scalar, table, aggregate, table-in/out, buffering, partitioned, multi-branch, transactional — that serves the canonical VGI integration suite. It is the best place to look for working patterns of any feature.
 
 ## Related projects

build.gradle.kts

@@ -8,7 +8,7 @@ plugins {
 
 allprojects {
     group = "farm.query"
-    version = "0.27.0"
+    version = "0.27.1"
 
     repositories {
         mavenCentral()

docs/iroh.md

@@ -0,0 +1,31 @@
+# Iroh workers and clients
+
+Java VGI workers use the language-neutral `vgi-iroh-bridge`. Both integration
+styles are one worker call:
+
+```java
+worker.runIrohTcpUpstream(
+        "127.0.0.1", 9400, 0, new IrohBridgeOptions("production"));
+
+worker.runHttp(
+        "127.0.0.1", 9401, new IrohBridgeOptions("production"));
+```
+
+The ordinary `runFromArgs` entry point accepts the same portable flags as the
+other VGI frameworks:
+
+```console
+worker --iroh-raw-upstream 127.0.0.1:9400 --iroh-issuer production
+worker --http --port 9401 --iroh-issuer production
+```
+
+Repeat `--iroh-trusted-proxy <exact-IP>` to replace the loopback trust list.
+Use `--iroh-observe` to expose verified EndpointId evidence without promoting
+it to the application principal. The worker upstream must remain unreachable
+except through that trusted bridge.
+
+For clients, add the optional `farm.query:vgirpc-iroh` module and use
+`HttpRpcConnection.irohBuilder("httpi://<endpoint-id>", options)` for HTTP
+semantics, or `IrohRpcConnection.connect(...)` for stateful Arrow-mux. Both use
+the official JVM Iroh bindings and accept stable identity, custom/private relay,
+direct-address, cancellation, and timeout configuration.

settings.gradle.kts

@@ -21,10 +21,12 @@ if (vgiRpcJavaDir.isDirectory) {
             // (group=farm.query, see ../Development/vgi-rpc-java/build.gradle.kts).
             substitute(module("farm.query:vgirpc")).using(project(":vgirpc"))
             substitute(module("farm.query:vgirpc-oauth")).using(project(":vgirpc-oauth"))
+            substitute(module("farm.query:vgirpc-iroh")).using(project(":vgirpc-iroh"))
             // Back-compat with callers that still use the legacy
             // farm.query.vgirpc:* coordinate naming.
             substitute(module("farm.query.vgirpc:vgirpc")).using(project(":vgirpc"))
             substitute(module("farm.query.vgirpc:vgirpc-oauth")).using(project(":vgirpc-oauth"))
+            substitute(module("farm.query.vgirpc:vgirpc-iroh")).using(project(":vgirpc-iroh"))
         }
     }
 }

vgi/build.gradle.kts

@@ -5,7 +5,7 @@ plugins {
 
 dependencies {
     // farm.query is the published group (see ../build.gradle.kts allprojects).
-    api("farm.query:vgirpc:0.23.0")
+    api("farm.query:vgirpc:0.24.0")
     implementation("org.slf4j:slf4j-api:2.0.17")
     // Cross-process aggregate state store. DuckDB spawns multiple worker
     // subprocesses for parallel aggregation; SQLite's file locking gives us

vgi/src/main/java/farm/query/vgi/IrohBridgeOptions.java

@@ -0,0 +1,65 @@
+// Copyright 2026 Query Farm LLC - https://query.farm
+
+package farm.query.vgi;
+
+import farm.query.vgirpc.http.HttpServer;
+import farm.query.vgirpc.http.IrohPeerIdentityProviders;
+import farm.query.vgirpc.identity.PeerAuthenticationPolicies;
+import farm.query.vgirpc.transport.TcpServerOptions;
+
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+/** Trust boundary between a loopback VGI worker and {@code vgi-iroh-bridge}. */
+public record IrohBridgeOptions(
+        String issuer,
+        Set<String> trustedProxyAddresses,
+        boolean authenticate) {
+
+    /** Authenticate bridge-verified EndpointIds from a loopback bridge. */
+    public IrohBridgeOptions(String issuer) {
+        this(issuer, Set.of("127.0.0.1"), true);
+    }
+
+    public IrohBridgeOptions {
+        if (issuer == null || issuer.isBlank()) {
+            throw new IllegalArgumentException("Iroh bridge issuer is required");
+        }
+        trustedProxyAddresses = trustedProxyAddresses == null || trustedProxyAddresses.isEmpty()
+                ? Set.of("127.0.0.1")
+                : Set.copyOf(trustedProxyAddresses);
+    }
+
+    /** Apply identity forwarding to an HTTP worker configuration. */
+    public HttpServer.Config.Builder apply(HttpServer.Config.Builder builder) {
+        var provider = IrohPeerIdentityProviders.forwarded(issuer, trustedProxyAddresses);
+        return builder
+                .peerIdentityProviders(List.of(provider))
+                .peerAuthenticationPolicy(authenticate
+                        ? PeerAuthenticationPolicies.primary("iroh")
+                        : PeerAuthenticationPolicies::observe);
+    }
+
+    /** Build the strict PROXY-v2 configuration used by the raw bridge upstream. */
+    TcpServerOptions tcpServerOptions() {
+        return TcpServerOptions.builder()
+                .proxyProtocolV2Required(true)
+                .trustedProxyAddresses(trustedProxyAddresses)
+                .irohProxyIssuer(issuer)
+                .peerAuthenticationPolicy(authenticate
+                        ? PeerAuthenticationPolicies.primary("iroh")
+                        : PeerAuthenticationPolicies::observe)
+                .build();
+    }
+
+    static IrohBridgeOptions fromArgs(
+            String issuer, List<String> trustedProxyAddresses, boolean authenticate) {
+        return new IrohBridgeOptions(
+                issuer,
+                trustedProxyAddresses.isEmpty()
+                        ? Set.of("127.0.0.1")
+                        : new LinkedHashSet<>(trustedProxyAddresses),
+                authenticate);
+    }
+}

vgi/src/main/java/farm/query/vgi/Worker.java

@@ -1158,6 +1158,33 @@ public void runTcp(String host, int port, long idleTimeoutMs) throws IOException
                 });
     }
 
+    /**
+     * Serve the identity-preserving raw upstream consumed by
+     * {@code vgi-iroh-bridge}. The upstream is loopback-only and requires the
+     * bridge's EndpointId-bearing PROXY-v2 preamble on every connection.
+     *
+     * @param host loopback bind host
+     * @param port bind port; {@code 0} selects a free port
+     * @param idleTimeoutMs idle watchdog in milliseconds; {@code <= 0} disables it
+     * @param bridge trusted bridge identity configuration
+     * @throws IOException if the socket cannot be bound or served
+     */
+    public void runIrohTcpUpstream(
+            String host, int port, long idleTimeoutMs, IrohBridgeOptions bridge) throws IOException {
+        requireLoopback(host, "Iroh raw bridge upstream");
+        if (bridge == null) throw new IllegalArgumentException("Iroh bridge options are required");
+        TcpSocketTransport.serveForever(
+                host,
+                port,
+                buildServer(false),
+                idleTimeoutMs,
+                (boundHost, boundPort) -> {
+                    System.out.println("TCP:" + boundHost + ":" + boundPort);
+                    System.out.flush();
+                },
+                bridge.tcpServerOptions());
+    }
+
     /**
      * Parsed {@code [HOST:]PORT} TCP bind spec. Host defaults to loopback.
      *
@@ -1195,13 +1222,32 @@ public void runHttp(String host, int port) throws Exception {
         runHttp(HttpServer.Config.builder().host(host).port(port).build());
     }
 
+    /**
+     * Run the ordinary VGI HTTP server behind {@code vgi-iroh-bridge}, retaining
+     * HTTP limits, continuations, externalized batches, and the authenticated
+     * client EndpointId.
+     *
+     * @param host loopback bind host
+     * @param port bind port; {@code 0} selects a free port
+     * @param bridge trusted bridge identity configuration
+     * @throws Exception if the server fails to start or serve
+     */
+    public void runHttp(String host, int port, IrohBridgeOptions bridge) throws Exception {
+        requireLoopback(host, "Iroh HTTP bridge upstream");
+        if (bridge == null) throw new IllegalArgumentException("Iroh bridge options are required");
+        runHttp(bridge.apply(HttpServer.Config.builder().host(host).port(port)).build());
+    }
+
     /**
      * Canonical CLI dispatcher used by worker {@code main} methods. Parses
-     * the four flags every VGI worker accepts and runs the matching transport:
+     * the transport flags every VGI worker accepts and runs the matching transport:
      * <ul>
      *   <li>{@code --unix <path>}: AF_UNIX socket (launcher protocol)
      *   <li>{@code --tcp [<host>:]<port>}: TCP socket (launcher protocol)
      *   <li>{@code --http} with optional {@code --host}, {@code --port}: HTTP
+     *   <li>{@code --iroh-raw-upstream [<host>:]<port>}: trusted raw bridge upstream
+     *   <li>{@code --iroh-issuer}, repeated {@code --iroh-trusted-proxy}, and
+     *       {@code --iroh-observe}: Iroh bridge trust and authentication mode
      *   <li>{@code --idle-timeout <seconds>}: passed to {@code runUnixSocket} / {@code runTcp}
      *   <li>(default): stdio
      * </ul>
@@ -1231,6 +1277,10 @@ public void runFromArgs(String[] args,
         int port = 0;
         String unixSocket = null;
         String tcpAddr = null;
+        String irohRawUpstream = null;
+        String irohIssuer = null;
+        List<String> irohTrustedProxies = new ArrayList<>();
+        boolean irohObserve = false;
         long idleTimeoutMs = 0;
         for (int i = 0; i < args.length; i++) {
             switch (args[i]) {
@@ -1239,20 +1289,55 @@ public void runFromArgs(String[] args,
                 case "--port" -> port = Integer.parseInt(args[++i]);
                 case "--unix" -> unixSocket = args[++i];
                 case "--tcp" -> tcpAddr = args[++i];
+                case "--iroh-raw-upstream" -> irohRawUpstream = args[++i];
+                case "--iroh-issuer" -> irohIssuer = args[++i];
+                case "--iroh-trusted-proxy" -> irohTrustedProxies.add(args[++i]);
+                case "--iroh-observe" -> irohObserve = true;
                 case "--idle-timeout" -> idleTimeoutMs =
                         (long) (Double.parseDouble(args[++i]) * 1000.0);
                 default -> { System.err.println("unknown arg: " + args[i]); System.exit(2); }
             }
         }
+        int selectedTransports = (http ? 1 : 0)
+                + (unixSocket != null ? 1 : 0)
+                + (tcpAddr != null ? 1 : 0)
+                + (irohRawUpstream != null ? 1 : 0);
+        if (selectedTransports > 1) {
+            throw new IllegalArgumentException(
+                    "--http, --unix, --tcp, and --iroh-raw-upstream are mutually exclusive");
+        }
+        if ((irohIssuer != null || !irohTrustedProxies.isEmpty() || irohObserve)
+                && irohRawUpstream == null && !http) {
+            throw new IllegalArgumentException(
+                    "Iroh bridge options require --http or --iroh-raw-upstream");
+        }
+        if ((irohRawUpstream != null || !irohTrustedProxies.isEmpty() || irohObserve)
+                && irohIssuer == null) {
+            throw new IllegalArgumentException("Iroh bridge options require --iroh-issuer");
+        }
         try {
             if (unixSocket != null) {
                 runUnixSocket(Path.of(unixSocket), idleTimeoutMs);
             } else if (tcpAddr != null) {
                 TcpAddr a = parseTcpAddr(tcpAddr);
                 runTcp(a.host(), a.port(), idleTimeoutMs);
+            } else if (irohRawUpstream != null) {
+                if (irohIssuer == null) {
+                    throw new IllegalArgumentException(
+                            "--iroh-raw-upstream requires --iroh-issuer");
+                }
+                TcpAddr a = parseTcpAddr(irohRawUpstream);
+                runIrohTcpUpstream(a.host(), a.port(), idleTimeoutMs,
+                        IrohBridgeOptions.fromArgs(
+                                irohIssuer, irohTrustedProxies, !irohObserve));
             } else if (http) {
                 HttpServer.Config.Builder b = HttpServer.Config.builder().host(host).port(port);
                 if (httpCustomizer != null) b = httpCustomizer.apply(b);
+                if (irohIssuer != null) {
+                    requireLoopback(host, "Iroh HTTP bridge upstream");
+                    b = IrohBridgeOptions.fromArgs(
+                            irohIssuer, irohTrustedProxies, !irohObserve).apply(b);
+                }
                 runHttp(b.build());
             } else {
                 runStdio();
@@ -1309,4 +1394,10 @@ public void runHttp(HttpServer.Config config) throws Exception {
         }, "vgi-http-shutdown"));
         http.join();
     }
+
+    private static void requireLoopback(String host, String label) {
+        if (!("127.0.0.1".equals(host) || "::1".equals(host) || "localhost".equals(host))) {
+            throw new IllegalArgumentException(label + " must bind loopback, got " + host);
+        }
+    }
 }

vgi/src/test/java/farm/query/vgi/IrohBridgeOptionsTest.java

@@ -0,0 +1,32 @@
+// Copyright 2026 Query Farm LLC - https://query.farm
+
+package farm.query.vgi;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+final class IrohBridgeOptionsTest {
+    @Test
+    void defaultsToAuthenticatedLoopbackBridge() {
+        var options = new IrohBridgeOptions("test-mesh");
+        assertEquals(Set.of("127.0.0.1"), options.trustedProxyAddresses());
+        assertTrue(options.authenticate());
+    }
+
+    @Test
+    void rejectsMissingIssuerAndNonLoopbackWorkerBinds() {
+        assertThrows(IllegalArgumentException.class, () -> new IrohBridgeOptions(" "));
+        var worker = Worker.builder();
+        assertThrows(IllegalArgumentException.class,
+                () -> worker.runIrohTcpUpstream(
+                        "0.0.0.0", 9400, 0, new IrohBridgeOptions("test-mesh")));
+        assertThrows(IllegalArgumentException.class,
+                () -> worker.runHttp(
+                        "0.0.0.0", 9401, new IrohBridgeOptions("test-mesh")));
+    }
+}