fix: stabilize code queue proxy and egress

src/components/microservices/code-queue/Dockerfile

@@ -16,6 +16,7 @@ RUN apt-get update \
     git \
     gzip \
     iproute2 \
+    iptables \
     iputils-ping \
     jq \
     make \

src/components/microservices/code-queue/package.json

@@ -8,5 +8,10 @@
   },
   "dependencies": {
     "postgres": "latest"
+  },
+  "devDependencies": {
+    "@types/bun": "latest",
+    "@types/node": "latest",
+    "typescript": "latest"
   }
 }

src/components/microservices/code-queue/src/dev-containers.ts

@@ -82,10 +82,13 @@ async function startDevContainerPlan(plan: DevContainerPlan, options: {
     const before = await run("main-server", ctx().masterProxyEvidenceScript(plan), 15_000, "master-proxy-evidence-before-ping");
     const ping = await run(plan.providerId, ctx().devContainerPingScript(plan), 20_000, "remote-container-ping-google");
     const after = await run("main-server", ctx().masterProxyEvidenceScript(plan), 15_000, "master-proxy-evidence-after-ping");
-    verification.pingGoogleOk = ping.exitCode === 0 && /0% packet loss|1 packets received|1 received/iu.test(ping.stdout);
+    verification.pingGoogleOk = ping.exitCode === 0 && /0% packet loss|1 packets received|1 received|bytes from/iu.test(ping.stdout);
     verification.directPingEvidence = commands.find((command) => command.name === "remote-tunnel-start")?.stdout.includes("direct_ping=failed_expected") === true
       ? `direct ${plan.providerId} container ping failed before tunnel`
       : "direct ping did not fail before tunnel";
+    verification.sealedEgressEvidence = ping.stdout.includes("sealed_direct_ping=blocked_expected")
+      ? `direct ${plan.providerId} container egress blocked after tunnel seal`
+      : "sealed direct egress block was not observed";
     verification.masterProxyEvidenceBefore = ctx().safePreview(before.stdout, 2000);
     verification.pingGoogleLog = ping.stdout;
     verification.masterProxyEvidenceAfter = ctx().safePreview(after.stdout, 2000);
@@ -156,13 +159,14 @@ export async function startDevContainer(req: Request, providerFromPath: string |
         containerWorkdir: plan.containerWorkdir,
       },
       masterProxy: {
-        mode: "ssh-tun-nat",
+        mode: "ssh-tun-nat-sealed",
         masterHost: plan.masterHost,
         tunId: plan.tunId,
         tunName: plan.tunName,
         serverIp: plan.serverIp,
         clientIp: plan.clientIp,
         natChain: plan.natChain,
+        egressFirewallChain: plan.egressFirewallChain,
       },
       verification: result.verification,
       commands: result.commands,
@@ -175,11 +179,12 @@ export async function startDevContainer(req: Request, providerFromPath: string |
       providerId,
       containerName: plan.containerName,
       masterProxy: {
-        mode: "ssh-tun-nat",
+        mode: "ssh-tun-nat-sealed",
         masterHost: plan.masterHost,
         tunName: plan.tunName,
         clientIp: plan.clientIp,
         natChain: plan.natChain,
+        egressFirewallChain: plan.egressFirewallChain,
       },
     }, 500);
   }
@@ -193,7 +198,7 @@ export async function devContainerStatus(providerFromPath: string | null): Promi
 CONTAINER=${ctx().shellQuote(plan.containerName)}
 docker inspect "$CONTAINER" --format 'container={{.Name}} state={{.State.Status}} image={{.Config.Image}} workdir={{ index .Config.Labels "unidesk.workdir" }} started={{.State.StartedAt}}' 2>/dev/null || true
 if docker inspect "$CONTAINER" >/dev/null 2>&1; then
-  docker exec "$CONTAINER" bash -lc 'export PATH=/tmp/unidesk-tools:$PATH; echo default=$(ip route show default | head -1); echo resolv=$(tr "\\n" " " </etc/resolv.conf); echo pwd=$(pwd); command -v codex || true; ip addr show ${plan.tunName} 2>/dev/null || true' || true
+  docker exec "$CONTAINER" bash -lc 'export PATH=/tmp/unidesk-tools:$PATH; source /tmp/unidesk-dev-egress.env 2>/dev/null || true; echo default=$(ip route show default | head -1); echo route_8_8_8_8=$(ip route get 8.8.8.8 | head -1); echo resolv=$(tr "\\n" " " </etc/resolv.conf); echo pwd=$(pwd); command -v codex || true; ip addr show ${plan.tunName} 2>/dev/null || true; if [ -n "\${EGRESS_CHAIN:-}" ]; then iptables -S "$EGRESS_CHAIN" 2>/dev/null || true; iptables -S OUTPUT 2>/dev/null | grep "$EGRESS_CHAIN" || true; fi' || true
 fi`;
   commands.push(await ctx().runCodeQueueSsh(providerId, statusScript, 15_000, "remote-container-status"));
   commands.push(await ctx().runCodeQueueSsh("main-server", ctx().masterProxyEvidenceScript(plan), 15_000, "master-proxy-status"));
@@ -202,11 +207,12 @@ fi`;
     providerId,
     containerName: plan.containerName,
     masterProxy: {
-      mode: "ssh-tun-nat",
+      mode: "ssh-tun-nat-sealed",
       masterHost: plan.masterHost,
       tunName: plan.tunName,
       clientIp: plan.clientIp,
       natChain: plan.natChain,
+      egressFirewallChain: plan.egressFirewallChain,
     },
     commands,
   });

src/components/microservices/code-queue/src/index.ts

@@ -3777,7 +3777,7 @@ async function route(req: Request): Promise<Response> {
         defaultProviderId: plan.providerId,
         startEndpoint: `/api/dev-containers/${encodeURIComponent(plan.providerId)}/start`,
         statusEndpoint: `/api/dev-containers/${encodeURIComponent(plan.providerId)}/status`,
-        masterProxyMode: "ssh-tun-nat",
+        masterProxyMode: "ssh-tun-nat-sealed",
         defaultPlan: {
           containerName: plan.containerName,
           image: plan.image,
@@ -3787,6 +3787,7 @@ async function route(req: Request): Promise<Response> {
           serverIp: plan.serverIp,
           clientIp: plan.clientIp,
           natChain: plan.natChain,
+          egressFirewallChain: plan.egressFirewallChain,
         },
       });
     }

src/components/microservices/code-queue/src/judge.ts

@@ -118,7 +118,7 @@ function fallbackJudge(result: CodexRunResult, minimaxError?: string | JudgeFail
 }
 
 export const retryInstruction = "这是同一个 Codex thread 的 continuation，不是新任务。请基于上文继续完成原始任务；只做最小必要状态核查，避免从头重新摸索、避免重复已经完成的修改。";
-const codeQueueD601DeployCommand = "bun scripts/cli.ts ssh D601 'cd /home/ubuntu/unidesk-code-queue-deploy && CODE_QUEUE_ENV_FILE=/home/ubuntu/unidesk-code-queue-deploy/.state/code-queue-d601.env docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --build --force-recreate code-queue'";
+const codeQueueD601DeployCommand = "bun scripts/cli.ts ssh D601 'cd ~/cq-deploy && CODE_QUEUE_ENV_FILE=.state/code-queue-d601.env docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --build --force-recreate code-queue'";
 const codeQueueRestartSafetyGuidance = `Code Queue 服务具备 restart-recovery，允许在任何时候重启、重建或替换 D601 的 \`code-queue-backend\`；当当前任务修改 Code Queue 自身时，禁止等待当前 Code Queue task 退出或等待队列归零后再重启，因为这会等待自己退出形成自锁。正确做法是在 D601 直接执行 \`${codeQueueD601DeployCommand}\` 或等价 no-deps force-recreate，并在恢复后用 live health/task 查询验证。`;
 
 export function explicitUserInterrupt(task: QueueTask, result: CodexRunResult): boolean {

src/components/microservices/code-queue/src/oa-events.ts

@@ -28,6 +28,20 @@ export interface OaTraceStats extends JsonRecord {
   source: "oa-event-flow";
 }
 
+export interface OaTraceStepSummary {
+  eventSequence: number;
+  seq: number;
+  at: string;
+  kind: string;
+  title: string;
+  status: string;
+  summaryLines: string[];
+  rawSeqs: number[];
+  scopeId: string;
+  attemptIndex: number | null;
+  source: "oa-event-flow";
+}
+
 const postTimeoutMs = 2500;
 let context: OaEventContext | null = null;
 
@@ -119,6 +133,19 @@ function recordString(record: Record<string, unknown> | null, keys: string[]): s
   return "";
 }
 
+function asJsonRecord(value: unknown): JsonRecord {
+  return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : {};
+}
+
+function stringValue(value: unknown, fallback = ""): string {
+  return typeof value === "string" ? value : fallback;
+}
+
+function positiveInteger(value: unknown): number | null {
+  const parsed = Number(value);
+  return Number.isInteger(parsed) && parsed > 0 ? Math.floor(parsed) : null;
+}
+
 export function outputTraceKind(output: LiveOutput): "read" | "edit" | "run" | "error" | "message" | "system" {
   if (output.channel === "diff") return "edit";
   if (output.channel === "error") return "error";
@@ -338,6 +365,125 @@ export async function readOaTraceStatsForTaskAttempts(taskId: string, attemptInd
   return readOaTraceStatsForScopeIds([taskScopeId(taskId), ...uniqueAttempts.map((index) => taskAttemptScopeId(taskId, index))]);
 }
 
+function eventPayloadRecord(event: unknown): JsonRecord {
+  if (typeof event !== "object" || event === null || Array.isArray(event)) return {};
+  return asJsonRecord((event as Record<string, unknown>).payload);
+}
+
+function stringList(value: unknown, maxItems = 8): string[] {
+  if (!Array.isArray(value)) return [];
+  return value.map((item) => String(item ?? "").trimEnd()).filter((item) => item.length > 0).slice(0, maxItems);
+}
+
+function numberList(value: unknown, fallback: number): number[] {
+  if (!Array.isArray(value)) return [fallback];
+  const values = value.map((item) => Number(item)).filter((item) => Number.isFinite(item)).map((item) => Math.floor(item));
+  return values.length > 0 ? values : [fallback];
+}
+
+function commandLifecycleStatus(payload: JsonRecord, title: string, summaryLines: string[]): string {
+  const source = [title, ...summaryLines].join("\n");
+  const status = /\bstatus=([A-Za-z0-9_-]+)/u.exec(source)?.[1];
+  if (status !== undefined && status.length > 0) return status;
+  const method = stringValue(payload.method);
+  if (method.length > 0) return method;
+  return stringValue(payload.status);
+}
+
+function traceStepFromEvent(event: unknown): OaTraceStepSummary | null {
+  if (typeof event !== "object" || event === null || Array.isArray(event)) return null;
+  const record = event as Record<string, unknown>;
+  const payload = eventPayloadRecord(record);
+  const seq = Number(payload.seq ?? payload.stepSeq ?? payload.outputSeq);
+  if (!Number.isFinite(seq)) return null;
+  const eventSequence = Number(record.sequence ?? 0);
+  const title = stringValue(payload.title, stringValue(payload.method, stringValue(payload.kind))).slice(0, 300);
+  const summaryLines = stringList(payload.summaryLines);
+  return {
+    eventSequence: Number.isFinite(eventSequence) ? Math.floor(eventSequence) : 0,
+    seq: Math.floor(seq),
+    at: stringValue(record.createdAt, stringValue(payload.createdAt, ctx().nowIso())),
+    kind: stringValue(payload.kind, stringValue(payload.stepKind, stringValue(payload.channel, "system"))),
+    title,
+    status: commandLifecycleStatus(payload, title, summaryLines).slice(0, 120),
+    summaryLines,
+    rawSeqs: numberList(payload.rawSeqs, Math.floor(seq)),
+    scopeId: stringValue(payload.scopeId, stringValue(payload.attemptScopeId)),
+    attemptIndex: positiveInteger(payload.attemptIndex),
+    source: "oa-event-flow",
+  };
+}
+
+function eventNextAfterSeq(body: Record<string, unknown>, events: unknown[], fallback: number): number {
+  const bodyNext = Number(body.nextAfterSeq);
+  const eventNext = events.reduce<number>((max, event) => {
+    if (typeof event !== "object" || event === null || Array.isArray(event)) return max;
+    const sequence = Number((event as Record<string, unknown>).sequence);
+    return Number.isFinite(sequence) ? Math.max(max, Math.floor(sequence)) : max;
+  }, fallback);
+  return Number.isFinite(bodyNext) && bodyNext > fallback ? Math.max(eventNext, Math.floor(bodyNext)) : eventNext;
+}
+
+const traceStepReadPageLimit = 500;
+const traceStepReadMaxPages = 40;
+const traceStepReadAttempts = 3;
+const traceStepReadTimeoutMs = 10000;
+
+async function waitMs(ms: number): Promise<void> {
+  await new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function fetchOaTraceStepPage(url: URL): Promise<Record<string, unknown>> {
+  let lastError: unknown = null;
+  for (let attempt = 1; attempt <= traceStepReadAttempts; attempt += 1) {
+    const controller = new AbortController();
+    const timer = setTimeout(() => controller.abort(), traceStepReadTimeoutMs);
+    try {
+      const response = await fetch(url, { signal: controller.signal });
+      if (!response.ok) throw new Error(`status=${response.status}`);
+      return await response.json() as Record<string, unknown>;
+    } catch (error) {
+      lastError = error;
+      if (attempt < traceStepReadAttempts) await waitMs(250 * attempt);
+    } finally {
+      clearTimeout(timer);
+    }
+  }
+  throw lastError instanceof Error ? lastError : new Error(String(lastError ?? "unknown error"));
+}
+
+export async function readOaTraceStepsForTask(taskId: string, attemptIndex: number | null = null): Promise<OaTraceStepSummary[]> {
+  const cleanTaskId = String(taskId || "").trim();
+  if (cleanTaskId.length === 0) return [];
+  const runtime = ctx();
+  const tags = [`task:${cleanTaskId}`, "trace"];
+  if (attemptIndex !== null && Number.isInteger(attemptIndex) && attemptIndex > 0) tags.push(`attempt:${Math.floor(attemptIndex)}`);
+  const bySeq = new Map<number, OaTraceStepSummary>();
+  let afterSeq = 0;
+  for (let page = 0; page < traceStepReadMaxPages; page += 1) {
+    const url = new URL(`${runtime.baseUrl}/api/events`);
+    url.searchParams.set("tags", tags.join(","));
+    url.searchParams.set("type", "trace-step-created");
+    url.searchParams.set("limit", String(traceStepReadPageLimit));
+    url.searchParams.set("afterSeq", String(afterSeq));
+    try {
+      const body = await fetchOaTraceStepPage(url);
+      const events = Array.isArray(body.events) ? body.events : [];
+      for (const event of events) {
+        const step = traceStepFromEvent(event);
+        if (step !== null) bySeq.set(step.seq, { ...(bySeq.get(step.seq) ?? {}), ...step });
+      }
+      const nextAfterSeq = eventNextAfterSeq(body, events, afterSeq);
+      if (events.length < traceStepReadPageLimit || nextAfterSeq <= afterSeq) break;
+      afterSeq = nextAfterSeq;
+    } catch (error) {
+      runtime.logger("warn", "oa_trace_steps_read_failed", { taskId: cleanTaskId, attemptIndex, page, error: error instanceof Error ? error.message : String(error) });
+      throw error;
+    }
+  }
+  return Array.from(bySeq.values()).sort((left, right) => left.seq - right.seq);
+}
+
 function statNumber(stats: OaTraceStats | null | undefined, key: string): number | null {
   const value = Number(stats?.[key]);
   return Number.isFinite(value) && value >= 0 ? Math.floor(value) : null;

src/components/microservices/code-queue/src/provider-runtime.ts

@@ -118,7 +118,7 @@ function buildDevContainerPlan(providerId: string, body: Record<string, unknown>
     ? imageFromBody
     : ctx().config.devContainerImage.length > 0
       ? ctx().config.devContainerImage
-      : "unidesk-code-queue:latest";
+      : `unidesk-code-queue:${safeProvider.toLowerCase()}`;
   const workdir = workdirFromBody.length > 0 ? workdirFromBody : ctx().config.devContainerWorkdir;
   const thirdOctet = providerId === "D601" ? 6 : Math.max(1, Math.min(250, Math.floor(tunId / 64)));
   const baseOctet = providerId === "D601" ? 0 : (tunId % 64) * 4;
@@ -138,6 +138,7 @@ function buildDevContainerPlan(providerId: string, body: Record<string, unknown>
     serverIp: `10.214.${thirdOctet}.${serverLastOctet}`,
     clientIp: `10.214.${thirdOctet}.${clientLastOctet}`,
     natChain: safeDockerName(`UNIDESK-CODEX-DEV-${safeProvider}`).toUpperCase().slice(0, 28),
+    egressFirewallChain: safeDockerName(`UD-CQ-EGRESS-${safeProvider}`).toUpperCase().slice(0, 28),
     keyDir: `/home/ubuntu/.unidesk/codex-dev-proxy/${safeProvider}`,
     masterKeyPath: resolve(ctx().config.defaultWorkdir, ".state/code-queue/dev-proxy", safeProvider, "id_ed25519"),
   };
@@ -151,18 +152,27 @@ function appendLimited(buffer: string, chunk: Buffer | string, maxBytes: number)
   return { value: buffer + text.slice(0, remaining), truncated: true };
 }
 
-function runCodeQueueSsh(providerId: string, script: string, timeoutMs: number, name: string): Promise<DevContainerCommandLog> {
-  const started = Date.now();
+interface CliRunResult {
+  code: number | null;
+  signal: NodeJS.Signals | null;
+  stdout: string;
+  stderr: string;
+  timedOut: boolean;
+  truncated: boolean;
+  spawnError: Error | null;
+}
+
+function runCliCommand(args: string[], timeoutMs: number): Promise<CliRunResult> {
   const maxBuffer = 8 * 1024 * 1024;
-  return new Promise((resolveLog) => {
+  return new Promise((resolveRun) => {
     let stdout = "";
     let stderr = "";
     let truncated = false;
     let timedOut = false;
     let spawnError: Error | null = null;
-    const child = spawn("bun", ["scripts/cli.ts", "ssh", providerId, "bash -s"], {
+    const child = spawn("bun", args, {
       cwd: ctx().config.defaultWorkdir,
-      stdio: ["pipe", "pipe", "pipe"],
+      stdio: ["ignore", "pipe", "pipe"],
     });
     const timer = setTimeout(() => {
       timedOut = true;
@@ -183,28 +193,152 @@ function runCodeQueueSsh(providerId: string, script: string, timeoutMs: number,
     child.on("error", (error) => {
       spawnError = error;
     });
-    child.stdin.on("error", () => undefined);
     child.on("close", (code, signal) => {
       clearTimeout(timer);
-      const errorParts: string[] = [];
-      if (stderr.length > 0) errorParts.push(stderr);
-      if (timedOut) errorParts.push(`Error: command timed out after ${timeoutMs}ms`);
-      if (spawnError !== null) errorParts.push(`${spawnError.name}: ${spawnError.message}`);
-      if (truncated) errorParts.push(`output truncated at ${maxBuffer} bytes`);
-      resolveLog({
-        name,
-        providerId,
-        exitCode: timedOut && code === 0 ? null : code,
+      resolveRun({
+        code: timedOut && code === 0 ? null : code,
         signal: signal as NodeJS.Signals | null,
-        durationMs: Date.now() - started,
         stdout,
-        stderr: errorParts.join("\n").trim(),
+        stderr,
+        timedOut,
+        truncated,
+        spawnError,
       });
     });
-    child.stdin.end(script);
   });
 }
 
+function buildFrontendHostSshArgs(providerId: string, command: string, timeoutMs: number): string[] {
+  const waitMs = Math.max(timeoutMs + 5_000, 20_000);
+  return [
+    "scripts/cli.ts",
+    "--main-server-ip",
+    ctx().config.devContainerMasterHost,
+    "debug",
+    "dispatch",
+    providerId,
+    "host.ssh",
+    "--wait-ms",
+    String(waitMs),
+    "--payload-json",
+    JSON.stringify({ source: "code-queue", mode: "exec", command, timeoutMs }),
+  ];
+}
+
+function parseFrontendHostSshResult(run: CliRunResult): { stdout: string; stderr: string; exitCode: number | null } {
+  let stdout = run.stdout;
+  let stderr = run.stderr;
+  let exitCode = run.code;
+  try {
+    const parsed = JSON.parse(run.stdout) as {
+      ok?: boolean;
+      error?: unknown;
+      data?: {
+        dispatch?: unknown;
+        wait?: {
+          task?: {
+            status?: string;
+            result?: {
+              stdout?: string;
+              stderr?: string;
+              exitCode?: number;
+            };
+          };
+        };
+      };
+    };
+    const task = parsed.data?.wait?.task;
+    const result = task?.result;
+    if (result !== undefined) {
+      stdout = typeof result.stdout === "string" ? result.stdout : "";
+      stderr = [typeof result.stderr === "string" ? result.stderr : "", run.stderr].filter((part) => part.length > 0).join("\n");
+      exitCode = typeof result.exitCode === "number" ? result.exitCode : task?.status === "succeeded" ? 0 : 1;
+    } else if (parsed.ok === false || task?.status === "failed") {
+      stdout = "";
+      stderr = [JSON.stringify(parsed.error ?? task ?? parsed.data?.dispatch ?? parsed), run.stderr].filter((part) => part.length > 0).join("\n");
+      exitCode = 1;
+    }
+  } catch {
+    // Keep raw stdout/stderr when the CLI itself failed before returning dispatch JSON.
+  }
+  return { stdout, stderr, exitCode };
+}
+
+async function runFrontendHostSsh(providerId: string, command: string, timeoutMs: number): Promise<CliRunResult & { parsedStdout: string; parsedStderr: string; parsedExitCode: number | null }> {
+  const run = await runCliCommand(buildFrontendHostSshArgs(providerId, command, timeoutMs), Math.max(timeoutMs + 15_000, 30_000));
+  const parsed = parseFrontendHostSshResult(run);
+  return { ...run, parsedStdout: parsed.stdout, parsedStderr: parsed.stderr, parsedExitCode: parsed.exitCode };
+}
+
+function encodedScriptCommand(encodedScript: string): string {
+  return [
+    "set -euo pipefail",
+    "tmp=$(mktemp /tmp/unidesk-code-queue-ssh.XXXXXX)",
+    "cleanup(){ rm -f \"$tmp\"; }",
+    "trap cleanup EXIT",
+    "base64 -d > \"$tmp\" <<'UNIDESK_CODE_QUEUE_SCRIPT'",
+    encodedScript,
+    "UNIDESK_CODE_QUEUE_SCRIPT",
+    "bash \"$tmp\"",
+  ].join("\n");
+}
+
+function commandLogFromRun(name: string, providerId: string, started: number, run: CliRunResult & { parsedStdout: string; parsedStderr: string; parsedExitCode: number | null }): DevContainerCommandLog {
+  const errorParts: string[] = [];
+  if (run.parsedStderr.length > 0) errorParts.push(run.parsedStderr);
+  if (run.timedOut) errorParts.push(`Error: command timed out after ${Date.now() - started}ms`);
+  if (run.spawnError !== null) errorParts.push(`${run.spawnError.name}: ${run.spawnError.message}`);
+  if (run.truncated) errorParts.push("output truncated at 8388608 bytes");
+  return {
+    name,
+    providerId,
+    exitCode: run.parsedExitCode,
+    signal: run.signal,
+    durationMs: Date.now() - started,
+    stdout: run.parsedStdout,
+    stderr: errorParts.join("\n").trim(),
+  };
+}
+
+async function uploadAndRunLargeScript(providerId: string, encodedScript: string, timeoutMs: number, name: string, started: number): Promise<DevContainerCommandLog> {
+  const token = `${Date.now()}-${Math.random().toString(16).slice(2)}`.replace(/[^A-Za-z0-9.-]/gu, "-");
+  const b64Path = `/tmp/unidesk-code-queue-${token}.b64`;
+  const scriptPath = `/tmp/unidesk-code-queue-${token}.sh`;
+  const init = await runFrontendHostSsh(providerId, `set -euo pipefail\numask 077\n: > ${shellQuote(b64Path)}`, 15_000);
+  if (init.parsedExitCode !== 0) return commandLogFromRun(name, providerId, started, init);
+  const chunkSize = 1800;
+  for (let offset = 0; offset < encodedScript.length; offset += chunkSize) {
+    const chunk = encodedScript.slice(offset, offset + chunkSize);
+    const appendCommand = [
+      "set -euo pipefail",
+      `cat >> ${shellQuote(b64Path)} <<'UNIDESK_CODE_QUEUE_CHUNK'`,
+      chunk,
+      "UNIDESK_CODE_QUEUE_CHUNK",
+    ].join("\n");
+    const append = await runFrontendHostSsh(providerId, appendCommand, 15_000);
+    if (append.parsedExitCode !== 0) return commandLogFromRun(name, providerId, started, append);
+  }
+  const runCommand = [
+    "set -uo pipefail",
+    `base64 -d ${shellQuote(b64Path)} > ${shellQuote(scriptPath)}`,
+    `rm -f ${shellQuote(b64Path)}`,
+    `bash ${shellQuote(scriptPath)}`,
+    "code=$?",
+    `rm -f ${shellQuote(scriptPath)}`,
+    "exit \"$code\"",
+  ].join("\n");
+  const run = await runFrontendHostSsh(providerId, runCommand, timeoutMs);
+  return commandLogFromRun(name, providerId, started, run);
+}
+
+function runCodeQueueSsh(providerId: string, script: string, timeoutMs: number, name: string): Promise<DevContainerCommandLog> {
+  const started = Date.now();
+  const encodedScript = Buffer.from(script, "utf8").toString("base64");
+  const inlineCommand = encodedScriptCommand(encodedScript);
+  if (inlineCommand.length > 3_400) return uploadAndRunLargeScript(providerId, encodedScript, timeoutMs, name, started);
+  return runFrontendHostSsh(providerId, inlineCommand, timeoutMs).then((run) => commandLogFromRun(name, providerId, started, run));
+}
+
 function throwIfCommandFailed(command: DevContainerCommandLog): void {
   if (command.exitCode === 0) return;
   throw new Error(`${command.name} failed on ${command.providerId ?? "local"} exit=${command.exitCode} stderr=${ctx().safePreview(command.stderr, 1000)}`);
@@ -295,8 +429,6 @@ function remoteContainerStartScript(plan: DevContainerPlan, forceRecreate: boole
   return `set -euo pipefail
 CONTAINER=${shellQuote(plan.containerName)}
 REQUESTED_IMAGE=${shellQuote(plan.image)}
-CODEX_IMAGE=unidesk-code-queue:latest
-FALLBACK_IMAGE=${shellQuote(`unidesk_provider-gateway:${plan.providerId.toLowerCase()}`)}
 KEY_DIR=${shellQuote(plan.keyDir)}
 WORKDIR=${shellQuote(plan.workdir)}
 CONTAINER_WORKDIR=${shellQuote(plan.containerWorkdir)}
@@ -306,16 +438,9 @@ SSH_MOUNT_ARGS=()
 if [ -d "$SSH_DIR" ]; then SSH_MOUNT_ARGS=(-v "$SSH_DIR":/root/.ssh:ro); fi
 HOST_PATH_MOUNT_ARGS=()
 if [ -d /mnt ]; then HOST_PATH_MOUNT_ARGS+=(-v /mnt:/mnt); fi
-if ! docker image inspect "$IMAGE" >/dev/null 2>&1 && docker image inspect "$CODEX_IMAGE" >/dev/null 2>&1; then
-  IMAGE="$CODEX_IMAGE"
-fi
 if ! docker image inspect "$IMAGE" >/dev/null 2>&1; then
-  if docker image inspect "$FALLBACK_IMAGE" >/dev/null 2>&1; then
-    IMAGE="$FALLBACK_IMAGE"
-  else
-    echo "missing requested image $REQUESTED_IMAGE, codex image $CODEX_IMAGE and fallback $FALLBACK_IMAGE" >&2
-    exit 1
-  fi
+  echo "missing dev container image $REQUESTED_IMAGE" >&2
+  exit 1
 fi
 test -r "$KEY_DIR/id_ed25519"
 test -r "$KEY_DIR/known_hosts"
@@ -336,12 +461,15 @@ cid=$(docker run -d \\
   --name "$CONTAINER" \\
   --hostname ${shellQuote(`codex-dev-${plan.providerId}`)} \\
   --user root \\
+  --no-healthcheck \\
   --cap-add NET_ADMIN \\
   --device /dev/net/tun \\
   --add-host host.docker.internal:host-gateway \\
   --label unidesk.role=codex-dev \\
   --label unidesk.provider=${shellQuote(plan.providerId)} \\
   --label "unidesk.workdir=$WORKDIR" \\
+  --label com.docker.compose.project=unidesk-code-queue-dev \\
+  --label com.docker.compose.service=codex-dev \\
   --env-file "$KEY_DIR/codex-env" \\
   -e CODEX_HOME=${shellQuote(plan.remoteCodexHome)} \\
   -e CODEX_INTERNAL_ORIGINATOR_OVERRIDE=unidesk_code_queue \\
@@ -362,6 +490,7 @@ cid=$(docker run -d \\
   "$IMAGE" \\
   bash -lc 'mkdir -p /tmp/unidesk-tools; ln -sf /usr/local/bin/busybox /tmp/unidesk-tools/ip; ln -sf /usr/local/bin/busybox /tmp/unidesk-tools/ping; export PATH=/tmp/unidesk-tools:$PATH; echo ready; sleep infinity')
 docker exec "$CONTAINER" bash -lc 'mkdir -p /tmp/unidesk-tools; ln -sf /usr/local/bin/busybox /tmp/unidesk-tools/ip; ln -sf /usr/local/bin/busybox /tmp/unidesk-tools/ping; export PATH=/tmp/unidesk-tools:$PATH; echo container=$(hostname); pwd; ip route show default; ls -ld /dev/net/tun /run/unidesk-dev-proxy/id_ed25519 /var/lib/unidesk/code-queue/codex-home /var/lib/unidesk/code-queue/opencode-xdg; command -v ssh; command -v ip; command -v ping'
+docker exec "$CONTAINER" bash -lc 'command -v iptables; iptables --version'
 echo "remote_container_ready container=$CONTAINER cid=$cid image=$IMAGE workdir=$WORKDIR"`;
 }
 
@@ -393,14 +522,21 @@ mkdir -p /tmp/unidesk-tools
 ln -sf /usr/local/bin/busybox /tmp/unidesk-tools/ip
 ln -sf /usr/local/bin/busybox /tmp/unidesk-tools/ping
 export PATH=/tmp/unidesk-tools:$PATH
-MASTER=${plan.masterHost}
-TUN_ID=${plan.tunId}
-TUN=${plan.tunName}
-CLIENT_IP=${plan.clientIp}
-SERVER_IP=${plan.serverIp}
+MASTER=${shellQuote(plan.masterHost)}
+TUN_ID=${shellQuote(String(plan.tunId))}
+TUN=${shellQuote(plan.tunName)}
+CLIENT_IP=${shellQuote(plan.clientIp)}
+SERVER_IP=${shellQuote(plan.serverIp)}
+EGRESS_CHAIN=${shellQuote(plan.egressFirewallChain)}
 KNOWN_HOSTS=/tmp/unidesk-dev-proxy-known_hosts
 cp /run/unidesk-dev-proxy/known_hosts "$KNOWN_HOSTS"
 chmod 600 "$KNOWN_HOSTS"
+MASTER_IP=$(getent ahostsv4 "$MASTER" 2>/dev/null | awk '{print $1; exit}' || true)
+if [ -z "$MASTER_IP" ]; then MASTER_IP="$MASTER"; fi
+case "$MASTER_IP" in
+  *.*.*.*) ;;
+  *) echo "cannot resolve IPv4 master host: $MASTER" >&2; exit 1 ;;
+esac
 DEFAULT_LINE=$(ip route show default | head -1)
 DEFAULT_GW=$(printf '%s\\n' "$DEFAULT_LINE" | sed -n 's/^default via \\([^ ]*\\) dev \\([^ ]*\\).*/\\1/p')
 DEFAULT_DEV=$(printf '%s\\n' "$DEFAULT_LINE" | sed -n 's/^default via \\([^ ]*\\) dev \\([^ ]*\\).*/\\2/p')
@@ -415,6 +551,10 @@ else
 fi
 if [ -z "$ORIG_GW" ] || [ -z "$ORIG_DEV" ]; then echo "cannot detect docker bridge route: default=$DEFAULT_LINE link=$LINK_ROUTE" >&2; exit 1; fi
 ( ping -c 1 -W 3 google.com >/tmp/direct-ping.log 2>&1 && echo direct_ping=unexpected_ok ) || echo direct_ping=failed_expected
+if ! command -v iptables >/dev/null 2>&1; then
+  echo "iptables is required for sealed dev-container egress" >&2
+  exit 1
+fi
 if command -v pkill >/dev/null 2>&1; then
   pkill -f "ssh .* -w $TUN_ID:$TUN_ID .*$MASTER" >/dev/null 2>&1 || true
 elif command -v busybox >/dev/null 2>&1; then
@@ -425,7 +565,11 @@ elif command -v busybox >/dev/null 2>&1; then
   done
 fi
 ip link delete "$TUN" >/dev/null 2>&1 || true
-ip route replace $MASTER/32 via "$ORIG_GW" dev "$ORIG_DEV"
+while iptables -w -D OUTPUT -j "$EGRESS_CHAIN" >/dev/null 2>&1; do :; done
+if command -v ip6tables >/dev/null 2>&1; then
+  while ip6tables -w -D OUTPUT -j "\${EGRESS_CHAIN}6" >/dev/null 2>&1; do :; done
+fi
+ip route replace "$MASTER_IP/32" via "$ORIG_GW" dev "$ORIG_DEV"
 ip route replace default via "$ORIG_GW" dev "$ORIG_DEV"
 SSH_LOG=/tmp/unidesk-dev-proxy-ssh.log
 rm -f "$SSH_LOG"
@@ -445,7 +589,37 @@ ip addr replace $CLIENT_IP peer $SERVER_IP dev "$TUN"
 ip link set "$TUN" up
 ip route replace default via $SERVER_IP dev "$TUN"
 printf 'nameserver 8.8.8.8\\nnameserver 1.1.1.1\\noptions timeout:2 attempts:2\\n' > /etc/resolv.conf
-echo "container_tunnel_ready orig=$ORIG_GW/$ORIG_DEV route_to_master=$(ip route get $MASTER | head -1) default=$(ip route show default | head -1) dns=$(tr '\\n' ' ' </etc/resolv.conf)"
+iptables -w -N "$EGRESS_CHAIN" 2>/dev/null || true
+iptables -w -F "$EGRESS_CHAIN"
+iptables -w -A "$EGRESS_CHAIN" -o lo -j RETURN
+iptables -w -A "$EGRESS_CHAIN" -m conntrack --ctstate ESTABLISHED,RELATED -j RETURN
+iptables -w -A "$EGRESS_CHAIN" -o "$TUN" -j RETURN
+iptables -w -A "$EGRESS_CHAIN" -d "$MASTER_IP/32" -o "$ORIG_DEV" -p tcp --dport 22 -j RETURN
+iptables -w -A "$EGRESS_CHAIN" -j REJECT --reject-with icmp-port-unreachable
+iptables -w -I OUTPUT 1 -j "$EGRESS_CHAIN"
+if command -v ip6tables >/dev/null 2>&1; then
+  CHAIN6="\${EGRESS_CHAIN}6"
+  ip6tables -w -N "$CHAIN6" 2>/dev/null || true
+  ip6tables -w -F "$CHAIN6"
+  ip6tables -w -A "$CHAIN6" -o lo -j RETURN
+  ip6tables -w -A "$CHAIN6" -m conntrack --ctstate ESTABLISHED,RELATED -j RETURN
+  ip6tables -w -A "$CHAIN6" -d ::1/128 -j RETURN
+  ip6tables -w -A "$CHAIN6" -d fc00::/7 -j RETURN
+  ip6tables -w -A "$CHAIN6" -d fe80::/10 -j RETURN
+  ip6tables -w -A "$CHAIN6" -j REJECT
+  ip6tables -w -I OUTPUT 1 -j "$CHAIN6"
+fi
+cat > /tmp/unidesk-dev-egress.env <<EOF
+EGRESS_CHAIN=$EGRESS_CHAIN
+ORIG_DEV=$ORIG_DEV
+TUN=$TUN
+MASTER=$MASTER
+MASTER_IP=$MASTER_IP
+SERVER_IP=$SERVER_IP
+EOF
+echo "container_tunnel_ready orig=$ORIG_GW/$ORIG_DEV route_to_master=$(ip route get $MASTER_IP | head -1) default=$(ip route show default | head -1) dns=$(tr '\\n' ' ' </etc/resolv.conf)"
+echo "sealed_egress_ready chain=$EGRESS_CHAIN allowed=tun:$TUN,master:$MASTER_IP:22/$ORIG_DEV"
+iptables -S "$EGRESS_CHAIN"
 INNER`;
 }
 
@@ -473,7 +647,7 @@ ip -s link show "$TUN"`;
 function devContainerPingScript(plan: DevContainerPlan): string {
   return `set -euo pipefail
 CONTAINER=${shellQuote(plan.containerName)}
-docker exec "$CONTAINER" bash -lc 'export PATH=/tmp/unidesk-tools:$PATH; echo route=$(ip route show default | head -1); echo resolv=$(tr "\\n" " " </etc/resolv.conf); ping -c 1 -W 5 google.com'`;
+docker exec "$CONTAINER" bash -lc 'export PATH=/tmp/unidesk-tools:$PATH; source /tmp/unidesk-dev-egress.env 2>/dev/null || true; echo route=$(ip route show default | head -1); echo route_8_8_8_8=$(ip route get 8.8.8.8 | head -1); echo resolv=$(tr "\\n" " " </etc/resolv.conf); if [ -n "\${EGRESS_CHAIN:-}" ]; then iptables -S "$EGRESS_CHAIN"; fi; if [ -n "\${ORIG_DEV:-}" ]; then ( ping -I "$ORIG_DEV" -c 1 -W 3 8.8.8.8 >/tmp/sealed-direct-ping.log 2>&1 && echo sealed_direct_ping=unexpected_ok ) || echo sealed_direct_ping=blocked_expected; fi; ping -c 1 -W 5 google.com'`;
 }
 
 function windowsNativeBridgeServerSource(): string {

src/components/microservices/code-queue/src/task-view.ts

@@ -22,6 +22,7 @@ import { codeAgentPortForModel, codeAgentPortInfo, codeExecutionModeInfo, extrac
 import { currentTaskPromptMarker, resolvedReferenceContextTitle, stripCodeQueueEnvironmentHint, userPromptForDisplay } from "./prompts";
 import { outputArchiveSignature, taskFullOutput } from "./task-output";
 import { retryPrompt } from "./judge";
+import { readOaTraceStepsForTask, type OaTraceStepSummary } from "./oa-events";
 
 export interface TaskViewContext {
   config: Pick<RuntimeConfig, "codexHome">;
@@ -1716,6 +1717,37 @@ function traceLineVisibleInTraceView(line: TranscriptLine): boolean {
   return line.kind !== "system" || traceSystemLineIsError(line);
 }
 
+function oaTraceStepKind(kind: string): TranscriptKind {
+  const normalized = String(kind || "").trim().toLowerCase();
+  if (normalized === "read" || normalized === "explored" || normalized === "explore") return "explored";
+  if (normalized === "edit" || normalized === "edited") return "edited";
+  if (normalized === "run" || normalized === "ran" || normalized === "command") return "ran";
+  if (normalized === "error" || normalized === "failed") return "error";
+  if (normalized === "message" || normalized === "assistant" || normalized === "user") return "message";
+  return "system";
+}
+
+function oaTraceStepToTranscriptLine(step: OaTraceStepSummary): TranscriptLine {
+  const kind = oaTraceStepKind(step.kind);
+  const summaryText = step.summaryLines.join("\n").trimEnd();
+  const commandText = summaryText.startsWith("item/") ? summaryText : "";
+  return {
+    seq: step.seq,
+    at: step.at,
+    kind,
+    title: step.title || (kind === "explored" ? "Read" : kind === "edited" ? "Edit" : kind === "ran" ? "Run" : "Trace step"),
+    status: step.status || undefined,
+    commandPreview: commandText.length > 0 ? commandText : undefined,
+    bodyPreview: commandText.length > 0 ? undefined : summaryText || undefined,
+    rawSeqs: step.rawSeqs,
+  };
+}
+
+async function oaTraceTranscriptForTask(taskId: string, attemptIndex: number | null): Promise<TranscriptLine[]> {
+  const steps = await readOaTraceStepsForTask(taskId, attemptIndex);
+  return steps.map(oaTraceStepToTranscriptLine).filter(traceLineVisibleInTraceView);
+}
+
 function mergeTraceWindowLines(left: TranscriptLine[], right: TranscriptLine[]): TranscriptLine[] {
   const seen = new Set<string>();
   const merged: TranscriptLine[] = [];
@@ -2114,14 +2146,11 @@ function taskPromptDetailResponse(task: QueueTask, url: URL): Response {
   });
 }
 
-function taskTraceStepsResponse(task: QueueTask, url: URL): Response {
+async function taskTraceStepsResponse(task: QueueTask, url: URL): Promise<Response> {
   const limit = ctx().parseLimit(url);
   const attemptIndex = ctx().parseSeqParam(url, "attempt", null);
   const agentPort = codeAgentPortForModel(task.model);
-  const previewTranscript = cachedPreviewTranscript(task);
-  const attemptWindow = attemptIndex === null ? null : traceAttemptWindows(task, previewTranscript).find((window) => window.index === attemptIndex) ?? null;
-  const sourceTranscript = attemptWindow === null ? previewTranscript : executionLinesForAttempt(attemptWindow.lines);
-  const transcript = sourceTranscript.filter(traceLineVisibleInTraceView);
+  const transcript = await oaTraceTranscriptForTask(task.id, attemptIndex);
   const page = ctx().pageBySeq(transcript, url, limit);
   return ctx().jsonResponse({
     ok: true,
@@ -2132,6 +2161,7 @@ function taskTraceStepsResponse(task: QueueTask, url: URL): Response {
     agentPort,
     agentPortInfo: codeAgentPortInfo(agentPort),
     attempt: attemptIndex,
+    source: "oa-event-flow",
     steps: page.chunk.map((line) => ({
       seq: line.seq,
       at: line.at,
@@ -2155,11 +2185,11 @@ function taskTraceStepsResponse(task: QueueTask, url: URL): Response {
   });
 }
 
-function taskTraceStepDetailResponse(task: QueueTask, url: URL): Response {
+async function taskTraceStepDetailResponse(task: QueueTask, url: URL): Promise<Response> {
   const seq = Number(url.searchParams.get("seq"));
   if (!Number.isFinite(seq)) return ctx().jsonResponse({ ok: false, error: "seq is required" }, 400);
   const agentPort = codeAgentPortForModel(task.model);
-  const transcript = fullTranscript(task).filter(traceLineVisibleInTraceView);
+  const transcript = await oaTraceTranscriptForTask(task.id, null);
   const line = transcript.find((item) => Number(item.seq) === seq || item.rawSeqs.includes(seq));
   if (line === undefined) return ctx().jsonResponse({ ok: false, error: "trace step not found", seq }, 404);
   return ctx().jsonResponse({

src/components/microservices/code-queue/src/types.ts

@@ -446,6 +446,7 @@ export interface DevContainerPlan {
   serverIp: string;
   clientIp: string;
   natChain: string;
+  egressFirewallChain: string;
   keyDir: string;
   masterKeyPath: string;
 }