merge: codex deploy fallback

# Conflicts:
#	AGENTS.md
#	TEST.md
#	config.json
#	deploy.json
#	docs/reference/cli.md
#	docs/reference/microservices.md
#	docs/reference/observability.md
#	scripts/cli.ts
#	scripts/src/microservices.ts
#	src/components/backend-core/src/microservice-proxy.ts
#	src/components/microservices/code-queue/src/index.ts
#	src/components/microservices/code-queue/src/queue-api.ts

src/components/backend-core/src/egress-tcp.ts

@@ -23,6 +23,9 @@ function isValidEgressPort(port: number): boolean {
   return Number.isInteger(port) && port > 0 && port <= 65_535;
 }
 
+const egressTcpConnectTimeoutMs = 15_000;
+const egressTcpIdleTimeoutMs = 600_000;
+
 function sendEgressClose(provider: ProviderSocket, connectionId: string, error?: string): void {
   const message: CoreEgressTcpCloseMessage = error === undefined
     ? { type: "egress_tcp_close", connectionId }
@@ -30,13 +33,31 @@ function sendEgressClose(provider: ProviderSocket, connectionId: string, error?:
   wsSendJson(provider, message);
 }
 
+function clearConnectionTimers(connection: EgressTcpConnection): void {
+  if (connection.connectTimer !== null) clearTimeout(connection.connectTimer);
+  if (connection.idleTimer !== null) clearTimeout(connection.idleTimer);
+  connection.connectTimer = null;
+  connection.idleTimer = null;
+}
+
 function closeEgressTcpConnection(providerId: string, connectionId: string, error?: string): void {
   const key = egressTcpKey(providerId, connectionId);
   const connection = ctx.activeEgressTcpConnections.get(key);
   if (connection === undefined) return;
   ctx.activeEgressTcpConnections.delete(key);
+  clearConnectionTimers(connection);
   connection.socket.destroy();
-  if (error !== undefined) sendEgressClose(connection.provider, connectionId, error);
+  if (error !== undefined) {
+    logger("warn", "egress_tcp_connection_closed", { providerId, connectionId, error });
+    sendEgressClose(connection.provider, connectionId, error);
+  }
+}
+
+function refreshConnectionIdle(connection: EgressTcpConnection): void {
+  if (connection.idleTimer !== null) clearTimeout(connection.idleTimer);
+  connection.idleTimer = setTimeout(() => {
+    closeEgressTcpConnection(connection.providerId, connection.connectionId, "egress tcp idle timeout");
+  }, egressTcpIdleTimeoutMs);
 }
 
 export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressTcpOpenMessage): void {
@@ -49,13 +70,32 @@ export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressT
   const key = egressTcpKey(message.providerId, message.connectionId);
   closeEgressTcpConnection(message.providerId, message.connectionId);
   const socket = connectTcp({ host, port });
-  const connection: EgressTcpConnection = { providerId: message.providerId, connectionId: message.connectionId, socket, provider: ws };
+  const connection: EgressTcpConnection = {
+    providerId: message.providerId,
+    connectionId: message.connectionId,
+    socket,
+    provider: ws,
+    connectTimer: null,
+    idleTimer: null,
+  };
   ctx.activeEgressTcpConnections.set(key, connection);
+  connection.connectTimer = setTimeout(() => {
+    closeEgressTcpConnection(message.providerId, message.connectionId, "egress tcp connect timeout");
+  }, egressTcpConnectTimeoutMs);
+  refreshConnectionIdle(connection);
   socket.on("connect", () => {
+    if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
+    if (connection.connectTimer !== null) {
+      clearTimeout(connection.connectTimer);
+      connection.connectTimer = null;
+    }
+    refreshConnectionIdle(connection);
     const opened: CoreEgressTcpOpenedMessage = { type: "egress_tcp_opened", connectionId: message.connectionId };
     wsSendJson(ws, opened);
   });
   socket.on("data", (chunk) => {
+    if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
+    refreshConnectionIdle(connection);
     const data: CoreEgressTcpDataMessage = {
       type: "egress_tcp_data",
       connectionId: message.connectionId,
@@ -67,11 +107,13 @@ export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressT
   socket.on("close", () => {
     if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
     ctx.activeEgressTcpConnections.delete(key);
+    clearConnectionTimers(connection);
     sendEgressClose(ws, message.connectionId);
   });
   socket.on("error", (error) => {
     if (ctx.activeEgressTcpConnections.get(key) !== connection) return;
     ctx.activeEgressTcpConnections.delete(key);
+    clearConnectionTimers(connection);
     sendEgressClose(ws, message.connectionId, error.message);
   });
 }
@@ -79,6 +121,7 @@ export function handleEgressTcpOpen(ws: ProviderSocket, message: ProviderEgressT
 export function handleEgressTcpData(message: ProviderEgressTcpDataMessage): void {
   const connection = ctx.activeEgressTcpConnections.get(egressTcpKey(message.providerId, message.connectionId));
   if (connection === undefined) return;
+  refreshConnectionIdle(connection);
   connection.socket.write(Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8"));
 }
 
@@ -90,6 +133,16 @@ export function closeEgressTcpConnectionsForProvider(providerId: string): void {
   for (const [key, connection] of ctx.activeEgressTcpConnections) {
     if (connection.providerId !== providerId) continue;
     ctx.activeEgressTcpConnections.delete(key);
+    clearConnectionTimers(connection);
+    connection.socket.destroy();
+  }
+}
+
+export function closeEgressTcpConnectionsForSocket(provider: ProviderSocket): void {
+  for (const [key, connection] of ctx.activeEgressTcpConnections) {
+    if (connection.provider !== provider) continue;
+    ctx.activeEgressTcpConnections.delete(key);
+    clearConnectionTimers(connection);
     connection.socket.destroy();
   }
 }

src/components/backend-core/src/microservice-proxy.ts

@@ -381,6 +381,7 @@ function isK3sctlManagedMicroservice(service: MicroserviceConfig): boolean {
 function codeQueueK3sServiceIdForRequest(method: string, targetPath: string): string {
   const normalizedMethod = method.toUpperCase();
   if (targetPath === "/" || targetPath === "/health" || targetPath === "/live" || targetPath === "/api/dev-ready") return "code-queue-scheduler";
+  if (targetPath === "/api/queues" || targetPath === "/api/tasks/overview") return "code-queue-scheduler";
   if (targetPath === "/api/oa/backfill" || targetPath === "/api/notifications/claudeqq/drain" || targetPath === "/api/notifications/claudeqq/backfill") return "code-queue-scheduler";
   if (targetPath === "/api/judge/probe" || targetPath === "/api/judge/self-test" || targetPath === "/api/queue-order/self-test" || targetPath === "/api/reference-injection/self-test" || targetPath === "/api/trace-port/self-test") return "code-queue-scheduler";
   if (/^\/api\/tasks\/[^/]+\/(?:steer|interrupt)$/u.test(targetPath)) return "code-queue-scheduler";
@@ -778,7 +779,8 @@ async function microserviceTunnelSelfTestResponse(service: MicroserviceConfig):
   const bodyRecord = isPlainRecord(body) ? body : {};
   const hasRequestId = typeof bodyRecord.requestId === "string" && bodyRecord.requestId.length > 0;
   const hasStage = typeof bodyRecord.stage === "string" && bodyRecord.stage.length > 0;
-  const ok = response.status === 502 && hasRequestId && hasStage && headers.requestId === bodyRecord.requestId;
+  const expectedStatus = response.status === 502 || response.status === 504;
+  const ok = expectedStatus && hasRequestId && hasStage && headers.requestId === bodyRecord.requestId;
   return jsonResponse({
     ok,
     serviceId: service.id,
@@ -787,7 +789,7 @@ async function microserviceTunnelSelfTestResponse(service: MicroserviceConfig):
     expectedFailure: true,
     status: response.status,
     checks: {
-      expectedStatus: response.status === 502,
+      expectedStatus,
       bodyHasRequestId: hasRequestId,
       bodyHasStage: hasStage,
       headerHasRequestId: typeof headers.requestId === "string" && headers.requestId.length > 0,
@@ -956,7 +958,7 @@ async function providerHttpTunnelMicroserviceResponse(
     const durationMs = Date.now() - startedAt;
     if (message === null) {
       attempts.push({ attempt, requestId, ok: false, reason: reason ?? "timeout", durationMs, timeoutMs });
-      if (retryable && tunnelFailureRetryable(reason) && attempt < maxAttempts) {
+      if (retryable && reason !== "timeout" && tunnelFailureRetryable(reason) && attempt < maxAttempts) {
         logger("warn", "http_tunnel_retry", {
           providerId: service.providerId,
           serviceId: service.id,

src/components/backend-core/src/types.ts

@@ -166,6 +166,8 @@ export interface EgressTcpConnection {
   connectionId: string;
   socket: Socket;
   provider: ProviderSocket;
+  connectTimer: ReturnType<typeof setTimeout> | null;
+  idleTimer: ReturnType<typeof setTimeout> | null;
 }
 
 export type HttpTunnelFailureReason =

src/components/microservices/k3sctl-adapter/k3s/ci/unidesk-ci.pipeline.yaml

@@ -270,6 +270,37 @@ spec:
           shift
           curl -fsS --cacert "$kube_ca" -H "Authorization: Bearer $kube_token" -X "$method" "$@"
         }
+        delete_if_exists() {
+          local path="$1"
+          local code
+          code="$(curl -sS -o /tmp/unidesk-ci-delete-response -w "%{http_code}" --cacert "$kube_ca" -H "Authorization: Bearer $kube_token" -X DELETE "$kube_api/$path")"
+          if [ "$code" = "200" ] || [ "$code" = "202" ] || [ "$code" = "404" ]; then
+            return 0
+          fi
+          cat /tmp/unidesk-ci-delete-response >&2
+          return 1
+        }
+        wait_deleted() {
+          local path="$1"
+          local deadline=$((SECONDS + 120))
+          local code
+          while [ "$SECONDS" -lt "$deadline" ]; do
+            code="$(curl -sS -o /tmp/unidesk-ci-get-response -w "%{http_code}" --cacert "$kube_ca" -H "Authorization: Bearer $kube_token" "$kube_api/$path")"
+            if [ "$code" = "404" ]; then
+              return 0
+            fi
+            if [ "$code" != "200" ]; then
+              cat /tmp/unidesk-ci-get-response >&2
+              return 1
+            fi
+            sleep 2
+          done
+          echo "timeout waiting for $path deletion" >&2
+          return 1
+        }
+        delete_if_exists "apis/apps/v1/namespaces/$kube_namespace/deployments/code-queue-ci-read"
+        delete_if_exists "api/v1/namespaces/$kube_namespace/services/code-queue-ci-read"
+        wait_deleted "apis/apps/v1/namespaces/$kube_namespace/deployments/code-queue-ci-read"
         cat >/tmp/code-queue-ci-read-deployment.yaml <<YAML
         apiVersion: apps/v1
         kind: Deployment
@@ -287,6 +318,8 @@ spec:
             matchLabels:
               app.kubernetes.io/name: code-queue
               app.kubernetes.io/component: ci-read
+          strategy:
+            type: Recreate
           template:
             metadata:
               labels:
@@ -382,6 +415,13 @@ spec:
                     periodSeconds: 5
                     timeoutSeconds: 3
                     failureThreshold: 20
+                  startupProbe:
+                    httpGet:
+                      path: /live
+                      port: http
+                    periodSeconds: 5
+                    timeoutSeconds: 3
+                    failureThreshold: 84
                   livenessProbe:
                     httpGet:
                       path: /live
@@ -396,11 +436,17 @@ spec:
                     limits:
                       memory: 1Gi
                   volumeMounts:
+                    - name: source
+                      mountPath: /app
                     - name: state
                       mountPath: /var/lib/unidesk/code-queue-ci
                     - name: logs
                       mountPath: /var/log/unidesk
               volumes:
+                - name: source
+                  hostPath:
+                    path: "$(workspaces.source.path)/repo"
+                    type: Directory
                 - name: state
                   emptyDir: {}
                 - name: logs
@@ -434,7 +480,7 @@ spec:
           -H "Content-Type: application/apply-patch+yaml" \
           --data-binary @/tmp/code-queue-ci-read-01 \
           "$kube_api/api/v1/namespaces/$kube_namespace/services/code-queue-ci-read?fieldManager=unidesk-ci&force=true" >/dev/null
-        deadline=$((SECONDS + 180))
+        deadline=$((SECONDS + 420))
         while [ "$SECONDS" -lt "$deadline" ]; do
           status="$(kube GET "$kube_api/apis/apps/v1/namespaces/$kube_namespace/deployments/code-queue-ci-read")"
           replicas="$(printf '%s' "$status" | jq -r '.spec.replicas // 1')"

src/components/provider-gateway/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@unidesk/provider-gateway",
-  "version": "0.2.21",
+  "version": "0.2.22",
   "private": true,
   "type": "module",
   "scripts": {

src/components/provider-gateway/src/egress-proxy.ts

@@ -29,7 +29,12 @@ interface Tunnel {
   method: string;
   opened: boolean;
   pending: Buffer[];
+  pendingBytes: number;
   closed: boolean;
+  createdAt: number;
+  lastActivityAt: number;
+  openTimer: ReturnType<typeof setTimeout> | null;
+  idleTimer: ReturnType<typeof setTimeout> | null;
 }
 
 export interface ProviderEgressProxyHandle {
@@ -111,6 +116,10 @@ function proxyUrlFor(host: string, port: number): string {
   return `http://${host}:${port}`;
 }
 
+const tunnelOpenTimeoutMs = 15_000;
+const tunnelIdleTimeoutMs = 600_000;
+const maxPendingBytes = 4 * 1024 * 1024;
+
 export function startProviderEgressProxy(options: ProviderEgressProxyOptions): ProviderEgressProxyHandle {
   const proxyUrl = proxyUrlFor(options.listenHost, options.listenPort);
   const tunnels = new Map<string, Tunnel>();
@@ -118,25 +127,73 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
 
   const send = (message: EgressToCoreMessage): boolean => options.sendToCore(message);
 
-  const status = (): Record<string, JsonValue> => ({
-    enabled: true,
-    providerId: options.providerId,
-    connected: options.isCoreConnected(),
-    proxyUrl,
-    listenHost: options.listenHost,
-    listenPort: options.listenPort,
-    activeTunnels: tunnels.size,
-    channel: "provider-gateway",
-  });
+  const status = (): Record<string, JsonValue> => {
+    const now = Date.now();
+    const tunnelList = Array.from(tunnels.values());
+    const ages = tunnelList.map((tunnel) => now - tunnel.createdAt);
+    return {
+      enabled: true,
+      providerId: options.providerId,
+      connected: options.isCoreConnected(),
+      proxyUrl,
+      listenHost: options.listenHost,
+      listenPort: options.listenPort,
+      activeTunnels: tunnels.size,
+      pendingTunnels: tunnelList.filter((tunnel) => !tunnel.opened).length,
+      oldestTunnelAgeMs: ages.length > 0 ? Math.max(...ages) : 0,
+      openTimeoutMs: tunnelOpenTimeoutMs,
+      idleTimeoutMs: tunnelIdleTimeoutMs,
+      maxPendingBytes,
+      channel: "provider-gateway",
+    };
+  };
+
+  const clearTunnelTimers = (tunnel: Tunnel): void => {
+    if (tunnel.openTimer !== null) clearTimeout(tunnel.openTimer);
+    if (tunnel.idleTimer !== null) clearTimeout(tunnel.idleTimer);
+    tunnel.openTimer = null;
+    tunnel.idleTimer = null;
+  };
+
+  const destroyTunnel = (tunnel: Tunnel, notifyCore: boolean, error?: string): void => {
+    tunnels.delete(tunnel.id);
+    tunnel.closed = true;
+    clearTunnelTimers(tunnel);
+    tunnel.pending.splice(0);
+    tunnel.pendingBytes = 0;
+    if (!tunnel.client.destroyed) tunnel.client.destroy();
+    if (notifyCore) send({ type: "egress_tcp_close", providerId: options.providerId, connectionId: tunnel.id, at: nowIso() });
+    if (error !== undefined) {
+      options.logger("warn", "egress_proxy_tunnel_closed", {
+        connectionId: tunnel.id,
+        opened: tunnel.opened,
+        ageMs: Date.now() - tunnel.createdAt,
+        error,
+      });
+    }
+  };
 
   const closeTunnel = (id: string, error?: string): void => {
     const tunnel = tunnels.get(id);
     if (tunnel === undefined) return;
-    tunnels.delete(id);
-    tunnel.closed = true;
-    if (!tunnel.client.destroyed) tunnel.client.destroy();
-    send({ type: "egress_tcp_close", providerId: options.providerId, connectionId: id, at: nowIso() });
-    if (error !== undefined) options.logger("warn", "egress_proxy_tunnel_closed", { connectionId: id, error });
+    destroyTunnel(tunnel, true, error);
+  };
+
+  const refreshTunnelIdle = (tunnel: Tunnel): void => {
+    if (tunnel.closed) return;
+    tunnel.lastActivityAt = Date.now();
+    if (tunnel.idleTimer !== null) clearTimeout(tunnel.idleTimer);
+    tunnel.idleTimer = setTimeout(() => closeTunnel(tunnel.id, "egress proxy idle timeout"), tunnelIdleTimeoutMs);
+  };
+
+  const queuePendingChunk = (tunnel: Tunnel, chunk: Buffer): boolean => {
+    if (tunnel.closed) return false;
+    tunnel.pending.push(chunk);
+    tunnel.pendingBytes += chunk.byteLength;
+    refreshTunnelIdle(tunnel);
+    if (tunnel.pendingBytes <= maxPendingBytes) return true;
+    closeTunnel(tunnel.id, "egress proxy pending buffer exceeded");
+    return false;
   };
 
   const handleCoreMessage = (message: EgressFromCoreMessage): boolean => {
@@ -144,24 +201,29 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
     if (message.type === "egress_tcp_opened") {
       if (tunnel === undefined || tunnel.closed) return true;
       tunnel.opened = true;
+      if (tunnel.openTimer !== null) {
+        clearTimeout(tunnel.openTimer);
+        tunnel.openTimer = null;
+      }
+      refreshTunnelIdle(tunnel);
       if (tunnel.method === "CONNECT") {
         tunnel.client.write("HTTP/1.1 200 Connection Established\r\nProxy-Agent: UniDesk-ProviderGateway\r\n\r\n");
       }
       for (const chunk of tunnel.pending.splice(0)) {
         send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: tunnel.id, data: chunk.toString("base64"), encoding: "base64", at: nowIso() });
       }
+      tunnel.pendingBytes = 0;
       return true;
     }
     if (message.type === "egress_tcp_data") {
       if (tunnel === undefined || tunnel.client.destroyed) return true;
+      refreshTunnelIdle(tunnel);
       tunnel.client.write(Buffer.from(message.data, message.encoding === "base64" ? "base64" : "utf8"));
       return true;
     }
     if (message.type === "egress_tcp_close") {
       if (tunnel !== undefined) {
-        tunnels.delete(message.connectionId);
-        tunnel.closed = true;
-        if (!tunnel.client.destroyed) tunnel.client.destroy();
+        destroyTunnel(tunnel, false, message.error);
       }
       if (message.error !== undefined && message.error.length > 0) {
         options.logger("warn", "egress_proxy_remote_close", { connectionId: message.connectionId, error: message.error });
@@ -222,14 +284,28 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
         fail("400 Bad Request", "unsupported proxy target\n");
         return;
       }
-      const tunnel: Tunnel = { id, client, method: parsed.method, opened: false, pending: [], closed: false };
+      const createdAt = Date.now();
+      const tunnel: Tunnel = {
+        id,
+        client,
+        method: parsed.method,
+        opened: false,
+        pending: [],
+        pendingBytes: 0,
+        closed: false,
+        createdAt,
+        lastActivityAt: createdAt,
+        openTimer: null,
+        idleTimer: null,
+      };
       tunnels.set(id, tunnel);
       client.on("data", (nextChunk) => {
         const nextBuffer = Buffer.isBuffer(nextChunk) ? nextChunk : Buffer.from(nextChunk);
         if (!tunnel.opened) {
-          tunnel.pending.push(nextBuffer);
+          queuePendingChunk(tunnel, nextBuffer);
           return;
         }
+        refreshTunnelIdle(tunnel);
         send({ type: "egress_tcp_data", providerId: options.providerId, connectionId: id, data: nextBuffer.toString("base64"), encoding: "base64", at: nowIso() });
       });
       client.on("close", () => closeTunnel(id));
@@ -238,10 +314,13 @@ export function startProviderEgressProxy(options: ProviderEgressProxyOptions): P
       if (!opened) {
         tunnels.delete(id);
         tunnel.closed = true;
+        clearTunnelTimers(tunnel);
         fail("503 Service Unavailable", "provider-gateway core channel is not connected\n");
         return;
       }
-      if (firstPayload !== null) tunnel.pending.push(firstPayload);
+      tunnel.openTimer = setTimeout(() => closeTunnel(id, "egress proxy open timeout"), tunnelOpenTimeoutMs);
+      refreshTunnelIdle(tunnel);
+      if (firstPayload !== null) queuePendingChunk(tunnel, firstPayload);
     });
     client.on("error", () => undefined);
   });