Add sbx ingress for bounded-query broker

containers/agent/bounded-query-wrapper.sh

@@ -3,8 +3,9 @@
 #
 # Agent-facing bounded-query CLI (protocol v2).
 #
-# Forwards a *narrow* request to the trusted bounded-query broker over a
-# dedicated Unix socket. It is analogous to gh-cli-proxy-wrapper.sh, but the
+# Forwards a *narrow* request to the trusted bounded-query broker over either
+# the Compose Unix socket or the authenticated sbx HTTP ingress. It is
+# analogous to gh-cli-proxy-wrapper.sh, but the
 # API is deliberately far narrower: this wrapper cannot express a command, an
 # image, a path, a URL, a ref, a mount, a runtime, a timeout, an environment,
 # or a credential. It accepts exactly:
@@ -29,7 +30,9 @@
 # image).
 
 CANONICAL_ERROR='{"status":"error"}'
-SOCKET="${AWF_BOUNDED_QUERY_SOCKET:-/run/awf-bounded-query/broker.sock}"
+SOCKET="${AWF_BOUNDED_QUERY_SOCKET:-}"
+ENDPOINT="${AWF_BOUNDED_QUERY_ENDPOINT:-}"
+CAPABILITY="${AWF_BOUNDED_QUERY_CAPABILITY:-}"
 PROTOCOL_VERSION=2
 # Keep in sync with MAX_SCHEMA_BYTES in src/bounded-query/protocol.ts and
 # containers/bounded-query/broker/protocol.js.
@@ -87,26 +90,51 @@ SCHEMA_B64=$(printf '%s' "$SCHEMA" | base64 | tr -d '\n' | tr '+/' '-_' | tr -d
 # The script must arrive on stdin; an interactive terminal means no script.
 [ ! -t 0 ] || emit_error
 
-[ -S "$SOCKET" ] || emit_error
-
-# --noproxy '*' keeps HTTP(S)_PROXY from redirecting a Unix-socket request.
-# --max-time bounds the wait comfortably above the largest timing bucket (10
-# minutes); the broker always answers at a fixed bucket boundary, so this
-# only guards a dead socket.
-RESPONSE=$(
-  curl --silent --show-error \
-    --noproxy '*' \
-    --unix-socket "$SOCKET" \
-    --max-time 660 \
-    -X POST \
-    -H "Expect:" \
-    -H "Content-Type: application/octet-stream" \
-    -H "X-AWF-Query-Version: ${PROTOCOL_VERSION}" \
-    -H "X-AWF-Repo: ${REPO}" \
-    -H "X-AWF-Schema-B64: ${SCHEMA_B64}" \
-    --data-binary @- \
-    "http://localhost/query" 2>/dev/null
-) || emit_error
+if [ -n "$SOCKET" ] && [ -z "$ENDPOINT" ] && [ -z "$CAPABILITY" ]; then
+  [ -S "$SOCKET" ] || emit_error
+  RESPONSE=$(
+    curl --silent --show-error \
+      --noproxy '*' \
+      --unix-socket "$SOCKET" \
+      --max-time 660 \
+      -X POST \
+      -H "Expect:" \
+      -H "Content-Type: application/octet-stream" \
+      -H "X-AWF-Query-Version: ${PROTOCOL_VERSION}" \
+      -H "X-AWF-Repo: ${REPO}" \
+      -H "X-AWF-Schema-B64: ${SCHEMA_B64}" \
+      --data-binary @- \
+      "http://localhost/query" 2>/dev/null
+  ) || emit_error
+elif [ -z "$SOCKET" ] && [ -n "$ENDPOINT" ] && [ -n "$CAPABILITY" ]; then
+  case "$ENDPOINT" in
+    http://host.docker.internal:*/query)
+      PORT="${ENDPOINT#http://host.docker.internal:}"
+      PORT="${PORT%/query}"
+      printf '%s' "$PORT" | LC_ALL=C grep -Eq '^[0-9]{1,5}$' || emit_error
+      [ "$PORT" -ge 1 ] 2>/dev/null || emit_error
+      [ "$PORT" -le 65535 ] 2>/dev/null || emit_error
+      ;;
+    *) emit_error ;;
+  esac
+  printf '%s' "$CAPABILITY" | LC_ALL=C grep -Eq '^[0-9a-f]{64}$' || emit_error
+  RESPONSE=$(
+    curl --silent --show-error \
+      --noproxy '*' \
+      --max-time 660 \
+      -X POST \
+      -H "Expect:" \
+      -H "Content-Type: application/octet-stream" \
+      -H "X-AWF-Capability: ${CAPABILITY}" \
+      -H "X-AWF-Query-Version: ${PROTOCOL_VERSION}" \
+      -H "X-AWF-Repo: ${REPO}" \
+      -H "X-AWF-Schema-B64: ${SCHEMA_B64}" \
+      --data-binary @- \
+      "$ENDPOINT" 2>/dev/null
+  ) || emit_error
+else
+  emit_error
+fi
 
 # Pass the broker's canonical response through unmodified, but only if it has
 # one of the two shapes the protocol ever produces. Anything else (a dead or

containers/bounded-query/Dockerfile

@@ -53,6 +53,10 @@ RUN chmod -R a-w /opt/awf \
     && node --check /opt/awf/broker/framing.js \
     && node --check /opt/awf/broker/workspace.js \
     && node --check /opt/awf/broker/query-runner.js \
+    && node --check /opt/awf/broker/query-runner-spec.js \
+    && node --check /opt/awf/broker/docker-client.js \
+    && node --check /opt/awf/broker/docker-query-runner.js \
+    && node --check /opt/awf/broker/gvisor-query-runner.js \
     && node --check /opt/awf/broker/healthcheck.js
 
 # Fixed broker-only mount points.

containers/bounded-query/broker/broker.js

@@ -11,7 +11,6 @@ const {
 const { createLedger } = require('./ledger');
 const { createRealClock, waitForBucket } = require('./scheduler');
 const defaultWorkspace = require('./workspace');
-const defaultRunner = require('./query-runner');
 
 /**
  * The trusted bounded-query broker (protocol v2).
@@ -51,7 +50,10 @@ const defaultRunner = require('./query-runner');
 function createBroker(params) {
   const { config, seedMap, runId, audit } = params;
   const workspace = params.workspace || defaultWorkspace;
-  const runner = params.runner || defaultRunner;
+  if (!params.runner) {
+    throw new Error('createBroker requires a trusted QueryRunner');
+  }
+  const runner = params.runner;
   const clock = params.clock || createRealClock();
   const ledger = params.ledger || createLedger(seedMap);
 
@@ -241,6 +243,11 @@ function createBroker(params) {
       return queued;
     },
 
+    /** Resolves when every admitted invocation has finished broker-side work. */
+    drain() {
+      return tail;
+    },
+
     /** @internal Exposed for tests. */
     get invocationsUsed() {
       return invocationsUsed;

containers/bounded-query/broker/config.js

@@ -23,6 +23,7 @@ const CONTROL_DIR = '/run/awf-bounded-query-control';
 const AUDIT_DIR = '/var/log/awf-bounded-query';
 /** Broker-private readiness marker; the control directory is never agent-mounted. */
 const READY_PATH = path.join(CONTROL_DIR, 'broker.ready');
+const SBX_CAPABILITY_PATH = path.join(CONTROL_DIR, 'sbx-ingress.json');
 const QUERY_SECCOMP_PATH = '/opt/awf/query-seccomp.json';
 
 /** Mount points inside the query container. Fixed, never caller-supplied. */
@@ -48,9 +49,27 @@ function parsePositiveInt(name, fallback) {
   if (!Number.isInteger(parsed) || parsed < 1) {
     throw new Error(`Environment variable ${name} must be a positive integer`);
   }
+
   return parsed;
 }
 
+function loadSbxIngressCapabilities(capabilityPath) {
+  const parsed = JSON.parse(fs.readFileSync(capabilityPath, 'utf8'));
+  const pattern = /^[0-9a-f]{64}$/;
+  if (
+    !parsed
+    || parsed.version !== 1
+    || typeof parsed.query !== 'string'
+    || typeof parsed.probe !== 'string'
+    || !pattern.test(parsed.query)
+    || !pattern.test(parsed.probe)
+    || parsed.query === parsed.probe
+  ) {
+    throw new Error('SBX ingress capability file is malformed');
+  }
+  return { query: parsed.query, probe: parsed.probe };
+}
+
 /**
  * Parses the per-invocation timeout, additionally re-enforcing (defense in
  * depth; AWF's host-side preflight already rejects an out-of-range value
@@ -74,9 +93,15 @@ function loadConfig() {
     throw new Error('AWF_BOUNDED_QUERY_MEMORY must be a Docker memory limit (e.g. "512m")');
   }
 
-  const dockerRuntime = process.env.AWF_BOUNDED_QUERY_RUNTIME || '';
-  if (dockerRuntime && !/^[A-Za-z0-9_.-]+$/.test(dockerRuntime)) {
-    throw new Error('AWF_BOUNDED_QUERY_RUNTIME contains unexpected characters');
+  const queryBackend = requireEnv('AWF_BOUNDED_QUERY_BACKEND');
+  if (queryBackend !== 'docker' && queryBackend !== 'gvisor') {
+    throw new Error(`Unsupported AWF_BOUNDED_QUERY_BACKEND: ${queryBackend}`);
+  }
+
+  const tcpPortRaw = process.env.AWF_BOUNDED_QUERY_TCP_PORT;
+  const tcpPort = tcpPortRaw === undefined ? undefined : parsePositiveInt('AWF_BOUNDED_QUERY_TCP_PORT');
+  if (tcpPort !== undefined && tcpPort > 65535) {
+    throw new Error('AWF_BOUNDED_QUERY_TCP_PORT must be a valid TCP port');
   }
 
   return {
@@ -97,12 +122,16 @@ function loadConfig() {
     // The daemon resolves query bind-mount sources in *its* filesystem view,
     // which is not necessarily the broker's (ARC/DinD split filesystems).
     hostWorkDir: requireEnv('AWF_BOUNDED_QUERY_HOST_WORK_DIR'),
-    dockerRuntime,
+    queryBackend,
     timeoutSeconds: parseTimeoutSeconds(),
     maxInvocations: parsePositiveInt('AWF_BOUNDED_QUERY_MAX_INVOCATIONS', 32),
     memoryLimit,
     socketUid: parsePositiveInt('AWF_BOUNDED_QUERY_SOCKET_UID', 0),
     socketGid: parsePositiveInt('AWF_BOUNDED_QUERY_SOCKET_GID', 0),
+    tcpPort,
+    sbxIngressCapabilities: tcpPort === undefined
+      ? undefined
+      : loadSbxIngressCapabilities(SBX_CAPABILITY_PATH),
   };
 }
 
@@ -145,4 +174,4 @@ function loadSeedMap(seedMapPath) {
   return { runId: parsed.runId, seeds };
 }
 
-module.exports = { READY_PATH, loadConfig, loadSeedMap };
+module.exports = { READY_PATH, SBX_CAPABILITY_PATH, loadConfig, loadSeedMap, loadSbxIngressCapabilities };

containers/bounded-query/broker/docker-client.js

@@ -0,0 +1,31 @@
+'use strict';
+
+const { execFile } = require('child_process');
+
+/**
+ * Executes the Docker CLI with bounded output and no inherited credentials.
+ */
+function runDocker(args, timeoutMs) {
+  return new Promise((resolve) => {
+    execFile(
+      'docker',
+      args,
+      {
+        timeout: timeoutMs,
+        killSignal: 'SIGKILL',
+        maxBuffer: 64 * 1024,
+        env: { PATH: process.env.PATH || '/usr/local/bin:/usr/bin:/bin' },
+      },
+      (error, stdout, stderr) => {
+        resolve({
+          exitCode: error && typeof error.code === 'number' ? error.code : error ? 1 : 0,
+          timedOut: Boolean(error && error.killed),
+          stderr: typeof stderr === 'string' ? stderr.slice(0, 2000) : '',
+          stdout: typeof stdout === 'string' ? stdout.slice(0, 2000) : '',
+        });
+      },
+    );
+  });
+}
+
+module.exports = { runDocker };

containers/bounded-query/broker/docker-query-runner.js

@@ -0,0 +1,111 @@
+'use strict';
+
+const defaultDockerClient = require('./docker-client');
+const {
+  CLI_GRACE_MS,
+  buildRemoveArgs,
+  deriveQueryContainerSpec,
+  normalizeTimeoutMs,
+} = require('./query-runner-spec');
+
+/**
+ * QueryRunner using the Docker daemon's default OCI runtime.
+ *
+ * The optional runtimeName is constructor-controlled so subclasses can select
+ * a fixed trusted runtime without accepting runtime data per invocation.
+ */
+class DockerQueryRunner {
+  constructor(config, deps = {}, runtimeName = undefined) {
+    this.config = config;
+    this.runtimeName = runtimeName;
+    this.docker = deps.docker || defaultDockerClient;
+    this.cleanupTail = Promise.resolve();
+  }
+
+  async assertAvailable() {
+    const image = await this.docker.runDocker(['image', 'inspect', this.config.queryImage], 60_000);
+    if (image.exitCode !== 0) {
+      throw new Error(`Query image is not available locally: ${this.config.queryImage}`);
+    }
+  }
+
+  spec(runId, invocationId) {
+    return deriveQueryContainerSpec({
+      config: this.config,
+      runId,
+      invocationId,
+      runtimeName: this.runtimeName,
+    });
+  }
+
+  async listContainerIds(args) {
+    const listed = await this.docker.runDocker(args, 30_000);
+    if (listed.exitCode !== 0) {
+      throw new Error('Failed to reconcile bounded-query containers');
+    }
+    const ids = listed.stdout.split('\n').map((id) => id.trim()).filter(Boolean);
+    if (ids.some((id) => !/^[0-9a-f]{12,64}$/.test(id))) {
+      throw new Error('Docker returned an invalid bounded-query container id');
+    }
+    return ids;
+  }
+
+  async removeListed(args) {
+    const ids = await this.listContainerIds(args);
+    if (ids.length === 0) return;
+    const removed = await this.docker.runDocker(buildRemoveArgs(ids), 30_000);
+    if (removed.exitCode !== 0) {
+      throw new Error('Failed to remove bounded-query containers');
+    }
+  }
+
+  serializeCleanup(operation) {
+    const queued = this.cleanupTail.then(operation, operation);
+    this.cleanupTail = queued.then(
+      () => undefined,
+      () => undefined,
+    );
+    return queued;
+  }
+
+  async reconcileRun(runId) {
+    const spec = this.spec(runId, 'reconcile');
+    await this.serializeCleanup(() => this.removeListed(spec.runListArgs));
+  }
+
+  async cleanupInvocation(runId, invocationId) {
+    const spec = this.spec(runId, invocationId);
+    await this.serializeCleanup(() => this.removeListed(spec.invocationListArgs));
+  }
+
+  async runQueryContainer(params) {
+    const spec = this.spec(params.runId, params.invocationId);
+    const timeoutMs = normalizeTimeoutMs(
+      (params.timeoutMs ?? this.config.timeoutSeconds * 1000) + CLI_GRACE_MS,
+    );
+
+    let result;
+    let runError;
+    try {
+      result = await this.docker.runDocker(spec.launchArgs, timeoutMs);
+    } catch (error) {
+      runError = error;
+    }
+
+    try {
+      await this.cleanupInvocation(params.runId, params.invocationId);
+    } catch (cleanupError) {
+      // A successful `docker run` means the container has already stopped, so
+      // preserve its result as before this refactor. Timeout/error paths may
+      // still have a live sandbox and therefore fail closed when cleanup fails.
+      if (!result || result.timedOut || result.exitCode !== 0) {
+        throw cleanupError;
+      }
+    }
+
+    if (runError) throw runError;
+    return result;
+  }
+}
+
+module.exports = { DockerQueryRunner };

containers/bounded-query/broker/framing.js

@@ -2,6 +2,9 @@
 
 const { MAX_SCHEMA_BYTES, MAX_SCRIPT_BYTES, strictParseJson } = require('./protocol');
 
+/** A peer that stops sending a request body cannot pin a broker connection. */
+const BODY_READ_TIMEOUT_MS = 5_000;
+
 /**
  * Wire framing for the agent → broker request (protocol v2).
  *
@@ -120,10 +123,16 @@ function readBoundedBody(req) {
     const chunks = [];
     let total = 0;
     let settled = false;
+    const timer = setTimeout(() => {
+      req.pause();
+      finish({ error: 'request body deadline exceeded' });
+    }, BODY_READ_TIMEOUT_MS);
+    timer.unref();
 
     const finish = (value) => {
       if (settled) return;
       settled = true;
+      clearTimeout(timer);
       resolve(value);
     };
 
@@ -160,4 +169,5 @@ module.exports = {
   SCHEMA_HEADER,
   buildRequestFromFrame,
   readBoundedBody,
+  BODY_READ_TIMEOUT_MS,
 };