feat(microservices): manage code queue through v3s

src/components/microservices/code-queue/Dockerfile

@@ -1,8 +1,10 @@
-FROM oven/bun:1-debian
+ARG CODE_QUEUE_BASE_IMAGE=oven/bun:1-debian
+FROM ${CODE_QUEUE_BASE_IMAGE}
 
 ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright
 
-RUN apt-get update \
+RUN (command -v codex >/dev/null 2>&1 && command -v opencode >/dev/null 2>&1 && command -v docker >/dev/null 2>&1 && command -v rg >/dev/null 2>&1) \
+  || (apt-get update \
   && apt-get install -y --no-install-recommends \
     bash \
     bubblewrap \
@@ -39,11 +41,11 @@ RUN apt-get update \
   && npm install -g @openai/codex@0.128.0 opencode-ai@1.14.48 playwright@1.59.1 \
   && playwright install --with-deps chromium \
   && apt-get clean \
-  && rm -rf /var/lib/apt/lists/*
+  && rm -rf /var/lib/apt/lists/*)
 
 WORKDIR /app/src/components/microservices/code-queue
 COPY src/components/microservices/code-queue/package.json ./package.json
-RUN bun install --production
+RUN test -d node_modules/postgres || bun install --production
 COPY src/components/shared /app/src/components/shared
 COPY src/components/microservices/code-queue/tsconfig.json ./tsconfig.json
 COPY src/components/microservices/code-queue/src ./src

src/components/microservices/code-queue/docker-compose.d601.yml

@@ -4,6 +4,8 @@ services:
     build:
       context: ../../../..
       dockerfile: src/components/microservices/code-queue/Dockerfile
+      args:
+        CODE_QUEUE_BASE_IMAGE: ${CODE_QUEUE_BASE_IMAGE:-oven/bun:1-debian}
     container_name: code-queue-backend
     restart: unless-stopped
     mem_limit: "${CODE_QUEUE_MEM_LIMIT:-3g}"
@@ -12,11 +14,14 @@ services:
       - path: ${CODE_QUEUE_ENV_FILE:-../../../../.state/code-queue-d601.env}
         required: false
     ports:
-      - "127.0.0.1:${CODE_QUEUE_HOST_PORT:-4222}:4222"
+      - "${CODE_QUEUE_HOST_BIND:-127.0.0.1}:${CODE_QUEUE_HOST_PORT:-4222}:4222"
     environment:
       HOST: "0.0.0.0"
       PORT: "4222"
       CODE_QUEUE_DATA_DIR: "/var/lib/unidesk/code-queue"
+      CODE_QUEUE_INSTANCE_ID: "${CODE_QUEUE_INSTANCE_ID:-D601}"
+      CODE_QUEUE_SCHEDULER_ENABLED: "${CODE_QUEUE_SCHEDULER_ENABLED:-true}"
+      CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED: "${CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED:-false}"
       CODE_QUEUE_WORKDIR: "${CODE_QUEUE_WORKDIR:-/workspace}"
       CODE_QUEUE_CODEX_HOME: "/var/lib/unidesk/code-queue/codex-home"
       CODE_QUEUE_OPENCODE_XDG_DIR: "/var/lib/unidesk/code-queue/opencode-xdg"

src/components/microservices/code-queue/src/code-agent/opencode.ts

@@ -1,6 +1,8 @@
 // 重构前 index.ts 只读参考：commit 6a04144d3f5103014f75b637d7e6bc2f45bf007f，blob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58；可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。
 
 import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
+import { mkdirSync } from "node:fs";
+import { resolve } from "node:path";
 import * as readline from "node:readline";
 import type { AppServerExit, CodexEventSummary, CodexRunResult, JsonValue, QueueTask, RuntimeConfig, TerminalStatus } from "../types";
 import type { ActiveRun, CodeAgentClient } from "./common";
@@ -93,8 +95,17 @@ function openCodeConfigContent(): string {
   });
 }
 
-function openCodeEnv(): NodeJS.ProcessEnv {
-  const xdgEnv = ctx().openCodeXdgEnv();
+function openCodeTaskXdgRoot(task: QueueTask, baseRoot?: string): string {
+  return resolve(baseRoot ?? ctx().openCodeXdgEnv().XDG_DATA_HOME, "..", "tasks", task.id);
+}
+
+function ensureOpenCodeXdgDirs(env: Record<string, string>): void {
+  for (const dir of Object.values(env)) mkdirSync(dir, { recursive: true });
+}
+
+function openCodeEnv(task: QueueTask): NodeJS.ProcessEnv {
+  const xdgEnv = ctx().openCodeXdgEnv(openCodeTaskXdgRoot(task));
+  ensureOpenCodeXdgDirs(xdgEnv);
   return {
     ...process.env,
     ...xdgEnv,
@@ -118,14 +129,16 @@ function openCodeRunArgs(task: QueueTask, prompt: string): string[] {
 
 function remoteOpenCodeRunCommand(task: QueueTask, prompt: string): string {
   const plan = ctx().buildDevContainerPlan(task.providerId, { workdir: ctx().remoteHostWorkdirForTask(task) });
+  const xdgEnv = ctx().openCodeXdgEnv(resolve(plan.remoteOpencodeXdgDir, "tasks", task.id));
   const envExports = [
-    ...Object.entries(ctx().openCodeXdgEnv(plan.remoteOpencodeXdgDir)).map(([key, value]) => `export ${key}=${ctx().shellQuote(value)}`),
+    ...Object.entries(xdgEnv).map(([key, value]) => `export ${key}=${ctx().shellQuote(value)}`),
     `export MINIMAX_API_BASE=${ctx().shellQuote(ctx().config.minimaxApiBase)}`,
     `export MINIMAX_MODEL=${ctx().shellQuote(ctx().config.minimaxModel)}`,
     `export OPENCODE_CONFIG_CONTENT=${ctx().shellQuote(openCodeConfigContent())}`,
   ].join("; ");
   const inner = [
     "set -euo pipefail",
+    `mkdir -p ${Object.values(xdgEnv).map(ctx().shellQuote).join(" ")}`,
     `mkdir -p ${ctx().shellQuote(task.cwd)}`,
     `cd ${ctx().shellQuote(task.cwd)}`,
     envExports,
@@ -187,7 +200,7 @@ class OpenCodeRunClient implements CodeAgentClient {
     this.child = ctx().providerIsMain(task.providerId)
       ? spawn("opencode", openCodeRunArgs(task, prompt), {
         cwd: task.cwd,
-        env: openCodeEnv(),
+        env: openCodeEnv(task),
         stdio: "pipe",
       })
       : spawn("bun", ["scripts/cli.ts", "ssh", task.providerId, remoteOpenCodeRunCommand(task, prompt)], {

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

@@ -94,7 +94,6 @@ import {
   notificationTargetConfigured,
   notificationTargetLabel,
   notifyTaskTerminal,
-  persistClaudeQqNotificationOutbox,
   scheduleClaudeQqNotificationDrain,
 } from "./notifications";
 import {
@@ -321,6 +320,9 @@ function readConfig(): RuntimeConfig {
     host: envString("HOST", "0.0.0.0"),
     port: envNumber("PORT", 4222),
     dataDir,
+    instanceId: envString("CODE_QUEUE_INSTANCE_ID", mainProviderId),
+    schedulerEnabled: envBool("CODE_QUEUE_SCHEDULER_ENABLED", true),
+    startupOaBackfillEnabled: envBool("CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED", false),
     outputArchiveDir: envString("CODE_QUEUE_OUTPUT_ARCHIVE_DIR", resolve(dataDir, "output-archive")),
     logFile: envString("LOG_FILE", "/var/log/unidesk/code-queue.jsonl"),
     defaultWorkdir,
@@ -1411,6 +1413,44 @@ async function warmDatabaseOverviewQueries(): Promise<void> {
   }
 }
 
+async function ensureDatabaseIndexes(): Promise<void> {
+  logger("info", "database_index_maintenance_start", {});
+  const started = performance.now();
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_status_updated ON unidesk_code_queue_tasks(status, updated_at DESC)`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_status_updated ON unidesk_code_queue_tasks(queue_id, status, updated_at DESC)`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_provider_updated ON unidesk_code_queue_tasks(provider_id, updated_at DESC)`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_execution_mode_updated ON unidesk_code_queue_tasks(execution_mode, updated_at DESC)`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_created ON unidesk_code_queue_tasks(created_at DESC)`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_created ON unidesk_code_queue_tasks(queue_id, created_at DESC, id DESC)`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_unread_terminal ON unidesk_code_queue_tasks(queue_id, updated_at DESC) WHERE read_at IS NULL AND status IN ('succeeded', 'failed', 'canceled')`;
+  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_model_updated ON unidesk_code_queue_tasks(model, updated_at DESC)`;
+  logger("info", "database_index_maintenance_complete", { durationMs: Math.round(performance.now() - started) });
+}
+
+function scheduleStartupDatabaseMaintenance(): void {
+  setTimeout(() => {
+    void (async () => {
+      const started = performance.now();
+      logger("info", "database_startup_maintenance_start", {
+        queueCount: state.queues.length,
+      });
+      for (const queue of state.queues) dirtyDatabaseQueueIds.add(queue.id);
+      await flushDirtyTasksToDatabase(true);
+      await loadClaudeQqNotificationOutboxFromDatabase();
+      await ensureDatabaseIndexes();
+      runGarbageCollection();
+      await warmDatabaseOverviewQueries();
+      logger("info", "database_startup_maintenance_complete", {
+        databaseNotificationCount: claudeQqNotificationOutboxItemCount(),
+        durationMs: Math.round(performance.now() - started),
+      });
+    })().catch((error) => {
+      databaseLastError = databaseErrorMessage(error);
+      logger("warn", "database_startup_maintenance_failed", { error: errorToJson(error) });
+    });
+  }, 1000).unref?.();
+}
+
 function rememberHotTask(task: QueueTask): QueueTask {
   const existing = findTask(task.id);
   if (existing !== null) return existing;
@@ -1546,16 +1586,6 @@ async function initDatabasePersistence(): Promise<void> {
       AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T'
   `;
   await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_status_updated ON unidesk_code_queue_tasks(status, updated_at DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_status_updated ON unidesk_code_queue_tasks(queue_id, status, updated_at DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_provider_updated ON unidesk_code_queue_tasks(provider_id, updated_at DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_execution_mode_updated ON unidesk_code_queue_tasks(execution_mode, updated_at DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_created ON unidesk_code_queue_tasks(created_at DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_created ON unidesk_code_queue_tasks(queue_id, created_at DESC, id DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_unread_terminal ON unidesk_code_queue_tasks(queue_id, updated_at DESC) WHERE read_at IS NULL AND status IN ('succeeded', 'failed', 'canceled')`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_tasks_model_updated ON unidesk_code_queue_tasks(model, updated_at DESC)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_notifications_pending ON unidesk_code_queue_notifications(sent_at, next_attempt_at)`;
-  await sql`CREATE INDEX IF NOT EXISTS idx_unidesk_code_queue_notifications_created ON unidesk_code_queue_notifications(created_at DESC)`;
 
   const countRows = await sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM unidesk_code_queue_tasks`;
   const hotTasks = await loadTasksFromDatabase("hot");
@@ -1595,20 +1625,16 @@ async function initDatabasePersistence(): Promise<void> {
     }
   }
   state.queues.splice(0, state.queues.length, ...Array.from(queueMap.values()).sort((left, right) => left.id.localeCompare(right.id)));
-  await loadClaudeQqNotificationOutboxFromDatabase();
   databaseReady = true;
-  for (const queue of state.queues) markQueueDirty(queue.id);
-  await flushDirtyTasksToDatabase(true);
+  scheduleStartupDatabaseMaintenance();
   runGarbageCollection();
-  await persistClaudeQqNotificationOutbox();
-  await warmDatabaseOverviewQueries();
   logger("info", "database_persistence_init_complete", {
     databaseTaskCount: Number(countRows[0]?.count ?? hotTasks.length),
     hotTaskCount: state.tasks.length,
     databaseQueueCount: queueRows.length,
-    databaseNotificationCount: claudeQqNotificationOutboxItemCount(),
     taskCount: state.tasks.length,
     queueCount: state.queues.length,
+    maintenanceMode: "background",
   });
 }
 
@@ -3034,6 +3060,9 @@ function queuedStatusReason(task: QueueTask, tasks: QueueTask[] = state.tasks):
   if (!serviceReady) {
     return queuedReason("service", "SERVICE", "Code Queue is still starting and has not enabled scheduling yet.");
   }
+  if (!config.schedulerEnabled) {
+    return queuedReason("scheduler_disabled", "STANDBY", "This Code Queue instance is a v3s-managed standby and does not start queued work.");
+  }
   if (mergingQueues.has(queueId)) {
     return queuedReason("queue_merge", "MERGING", "Queue merge is rewriting queue membership; scheduling will resume immediately after it finishes.");
   }
@@ -3082,7 +3111,7 @@ function nextRunnableTask(queueId: string): QueueTask | null {
 }
 
 async function processQueue(queueId: string): Promise<void> {
-  if (!serviceReady || processingQueues.has(queueId) || shutdownRequested) return;
+  if (!serviceReady || !config.schedulerEnabled || processingQueues.has(queueId) || shutdownRequested) return;
   processingQueues.add(queueId);
   updateProcessingFlag();
   try {
@@ -3116,7 +3145,7 @@ async function processQueue(queueId: string): Promise<void> {
 }
 
 function scheduleQueue(queueId?: string): void {
-  if (!serviceReady || shutdownRequested) return;
+  if (!serviceReady || !config.schedulerEnabled || shutdownRequested) return;
   const ids = queueId === undefined ? runnableQueueIds() : [queueId];
   for (const id of ids) {
     if (mergingQueues.has(id)) continue;
@@ -3785,14 +3814,35 @@ async function route(req: Request): Promise<Response> {
   const url = new URL(req.url);
   if (req.method === "OPTIONS") return jsonResponse({ ok: true });
   try {
-    if (url.pathname === "/" || url.pathname === "/health") return jsonResponse({
+    if (url.pathname === "/live") return jsonResponse({
       ok: true,
       service: "code-queue",
-      queue: await queueSummaryForHealth(false),
-      egressProxy: await providerGatewayEgressProxyStatus(),
-      oaEventPublisher: oaEventPublisherStatus(),
+      instanceId: config.instanceId,
+      databaseReady,
+      serviceReady,
       startedAt: serviceStartedAt,
     });
+    if (url.pathname === "/" || url.pathname === "/health") {
+      if (!databaseReady) return jsonResponse({
+        ok: false,
+        service: "code-queue",
+        instanceId: config.instanceId,
+        status: "starting",
+        databaseReady,
+        databaseLastError,
+        startedAt: serviceStartedAt,
+      }, 503);
+      return jsonResponse({
+        ok: true,
+        service: "code-queue",
+        instanceId: config.instanceId,
+        schedulerEnabled: config.schedulerEnabled,
+        queue: queueSummary(false, state.tasks),
+        egressProxy: await providerGatewayEgressProxyStatus(),
+        oaEventPublisher: oaEventPublisherStatus(),
+        startedAt: serviceStartedAt,
+      });
+    }
     if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-parseLimit(url)) });
     if (url.pathname === "/api/events" && req.method === "GET") return jsonResponse({ ok: false, error: "Code Queue private SSE was removed; subscribe to oa-event-flow /api/events/stream with service:code-queue tags." }, 410);
     if (url.pathname === "/api/dev-ready" && req.method === "GET") return jsonResponse({ ok: true, devReady: collectDevReady() });
@@ -3992,15 +4042,17 @@ startCodexSqliteLogExporter();
 startMemoryWatchdog();
 await initDatabasePersistenceWithRetry();
 Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route });
-logger("info", "service_started", { port: config.port, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres" });
+logger("info", "service_started", { port: config.port, instanceId: config.instanceId, schedulerEnabled: config.schedulerEnabled, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres" });
 {
   const devReady = collectDevReady() as Record<string, JsonValue>;
   logger(devReady.ok === true ? "info" : "warn", "dev_ready_check", devReady);
 }
-const startupRecovered = queueActiveTasksForRestartRetry("Service restarted while task was active", "startup");
+const startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRetry("Service restarted while task was active", "startup") : 0;
 if (startupRecovered > 0) logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered });
 persistState();
 serviceReady = true;
-setTimeout(() => { void backfillOaTraceStats(new URL("http://code-queue.local/api/oa/backfill?limit=2000")).catch((error) => logger("warn", "oa_trace_stats_startup_backfill_failed", { error: errorToJson(error) })); }, 1000).unref?.();
+if (config.startupOaBackfillEnabled) {
+  setTimeout(() => { void backfillOaTraceStats(new URL("http://code-queue.local/api/oa/backfill?limit=2000")).catch((error) => logger("warn", "oa_trace_stats_startup_backfill_failed", { error: errorToJson(error) })); }, 1000).unref?.();
+}
 scheduleQueue();
 scheduleClaudeQqNotificationDrain(1000);

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

@@ -355,7 +355,7 @@ export const defaultJudgeProbeCases: JudgeProbeCase[] = [
   {
     id: "code_queue_no_build_recreate_live_verified_should_complete",
     prompt: "修复 codex_1778762250800_82e71d 展开执行过程摘要 #1 只有 5 个 STEP 的问题；本地测试 OK 后部署到 D601 Code Queue，并验证 remote/frontdoor 能看到完整 STEP。",
-    finalResponse: "已修复 Code Queue trace-steps 的 OA Event Flow 分页读取问题，并完成上线验证。\n\n部署结果：先执行 `docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --build --force-recreate code-queue`，后续一次 build 受 Docker Hub mirror EOF 影响失败；为完成上线，已把当前修复同步进运行容器、commit 成 `unidesk-code-queue:d601`，并执行 `docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --force-recreate code-queue`，容器 `code-queue-backend` 已 Started，status=running，health=ok。\n\nLive Verification：D601 本机 `/api/tasks/codex_1778762250800_82e71d/trace-steps?afterSeq=0&limit=500&attempt=1` 返回 `source=oa-event-flow,total=153,returned=153,hasMore=false`；main server microservice proxy 返回 httpStatus=200 且 total=153；公网 served frontend direct API `/api/code-queue-direct/.../trace-steps?afterSeq=0&limit=500&attempt=1` 返回 source=oa-event-flow,total=153；`/trace-summary` 返回 stepCount=153, statsSource=oa-event-flow。\n\nDockerfile 已补齐依赖，后续重建可直接跑；建议后续修 Docker daemon registry mirror，避免 rebuild 偶发 EOF。",
+    finalResponse: "已修复 Code Queue trace-steps 的 OA Event Flow 分页读取问题，并完成上线验证。\n\n部署结果：先执行 `docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --build --force-recreate code-queue`，后续一次 build 受 Docker Hub mirror EOF 影响失败；为完成上线，已把当前修复同步进运行容器、commit 成 `unidesk-code-queue:d601`，并执行 `docker compose -f src/components/microservices/code-queue/docker-compose.d601.yml up -d --force-recreate code-queue`，容器 `code-queue-backend` 已 Started，status=running，health=ok。\n\nLive Verification：D601 本机 `/api/tasks/codex_1778762250800_82e71d/trace-steps?afterSeq=0&limit=500&attempt=1` 返回 `source=oa-event-flow,total=153,returned=153,hasMore=false`；main server microservice proxy 返回 httpStatus=200 且 total=153；公网 served frontend user-service API `/api/microservices/code-queue/proxy/.../trace-steps?afterSeq=0&limit=500&attempt=1` 返回 source=oa-event-flow,total=153；`/trace-summary` 返回 stepCount=153, statsSource=oa-event-flow。\n\nDockerfile 已补齐依赖，后续重建可直接跑；建议后续修 Docker daemon registry mirror，避免 rebuild 偶发 EOF。",
     expected: "complete",
     terminalStatus: "completed",
     outputs: [

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

@@ -32,6 +32,9 @@ export interface RuntimeConfig {
   host: string;
   port: number;
   dataDir: string;
+  instanceId: string;
+  schedulerEnabled: boolean;
+  startupOaBackfillEnabled: boolean;
   outputArchiveDir: string;
   logFile: string;
   defaultWorkdir: string;

src/components/microservices/v3sctl-adapter/Dockerfile

@@ -0,0 +1,23 @@
+ARG V3SCTL_ADAPTER_BASE_IMAGE=oven/bun:1-debian
+FROM ${V3SCTL_ADAPTER_BASE_IMAGE}
+
+# Never build the adapter FROM a service image: inherited Docker Desktop labels
+# can silently republish old Code Queue ports and mounts.
+RUN test -z "${CODE_QUEUE_DATA_DIR:-}" && test "${PORT:-}" != "4222"
+ENTRYPOINT []
+
+RUN (command -v curl >/dev/null 2>&1 && command -v ssh >/dev/null 2>&1 && command -v ps >/dev/null 2>&1) \
+  || (apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates curl openssh-client procps \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*)
+
+WORKDIR /app/src/components/microservices/v3sctl-adapter
+COPY src/components/shared /app/src/components/shared
+COPY src/components/microservices/v3sctl-adapter/package.json ./package.json
+COPY src/components/microservices/v3sctl-adapter/tsconfig.json ./tsconfig.json
+COPY src/components/microservices/v3sctl-adapter/src ./src
+COPY src/components/microservices/v3sctl-adapter/v3s ./v3s
+
+EXPOSE 4266
+CMD ["bun", "--smol", "run", "src/index.ts"]

src/components/microservices/v3sctl-adapter/docker-compose.d601.yml

@@ -0,0 +1,46 @@
+services:
+  v3sctl-adapter:
+    image: unidesk-v3sctl-adapter:d601
+    build:
+      context: ../../../..
+      dockerfile: src/components/microservices/v3sctl-adapter/Dockerfile
+      args:
+        V3SCTL_ADAPTER_BASE_IMAGE: ${V3SCTL_ADAPTER_BASE_IMAGE:-oven/bun:1-debian}
+    container_name: v3sctl-adapter
+    restart: unless-stopped
+    env_file:
+      - path: ${V3SCTL_ADAPTER_ENV_FILE:-../../../../.state/v3sctl-adapter-d601.env}
+        required: false
+    ports:
+      - "127.0.0.1:${V3SCTL_ADAPTER_HOST_PORT:-4266}:4266"
+    environment:
+      HOST: "0.0.0.0"
+      PORT: "4266"
+      LOG_FILE: "/var/log/unidesk/v3sctl-adapter.jsonl"
+      V3SCTL_CLUSTER_ID: "${V3SCTL_CLUSTER_ID:-D601}"
+      V3SCTL_NODE_ID: "${V3SCTL_NODE_ID:-D601}"
+      V3SCTL_KUBECTL_ENABLED: "${V3SCTL_KUBECTL_ENABLED:-false}"
+      V3SCTL_KUBE_API_PROXY_ENABLED: "${V3SCTL_KUBE_API_PROXY_ENABLED:-true}"
+      V3SCTL_KUBECONFIG_PATH: "/var/lib/unidesk/v3s/kubeconfig"
+      V3SCTL_KUBE_API_CONNECT_HOST: "${V3SCTL_KUBE_API_CONNECT_HOST:-host.docker.internal}"
+      V3SCTL_MANIFEST_PATHS: "${V3SCTL_MANIFEST_PATHS:-v3s/code-queue.v3s.json}"
+      V3SCTL_SERVICES_JSON: "${V3SCTL_SERVICES_JSON:-[]}"
+      UNIDESK_LOG_RETENTION_BYTES: "${UNIDESK_LOG_RETENTION_BYTES:-512MiB}"
+    volumes:
+      - ${V3SCTL_ADAPTER_LOG_DIR:-../../../../.state/v3sctl-adapter/logs}:/var/log/unidesk
+      - ${V3SCTL_KUBECONFIG_HOST_PATH:-../../../../.state/v3s/kubeconfig}:/var/lib/unidesk/v3s/kubeconfig:ro
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    networks:
+      - default
+      - provider-gateway
+    healthcheck:
+      test: ["CMD-SHELL", "curl -fsS --max-time 2 http://127.0.0.1:4266/health >/dev/null"]
+      interval: 5s
+      timeout: 3s
+      retries: 20
+
+networks:
+  provider-gateway:
+    external: true
+    name: ${V3SCTL_PROVIDER_GATEWAY_NETWORK:-unidesk-provider-d601_default}

src/components/microservices/v3sctl-adapter/package.json

@@ -0,0 +1,9 @@
+{
+  "name": "@unidesk/v3sctl-adapter",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "start": "bun run src/index.ts",
+    "check": "tsc -p tsconfig.json --noEmit"
+  }
+}

src/components/microservices/v3sctl-adapter/src/index.ts

@@ -0,0 +1,709 @@
+import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../../shared/src/rotating-jsonl";
+import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
+import { dirname, isAbsolute, join } from "node:path";
+import { tmpdir } from "node:os";
+import { fileURLToPath } from "node:url";
+
+type JsonValue = string | number | boolean | null | JsonValue[] | { [key: string]: JsonValue };
+type JsonRecord = Record<string, JsonValue>;
+
+type InstanceRole = "primary" | "standby" | "worker";
+
+interface ManagedEndpoint {
+  id: string;
+  nodeId: string;
+  role: InstanceRole;
+  baseUrl: string;
+  healthPath: string;
+}
+
+interface ManagedService {
+  id: string;
+  namespace: string;
+  kind: string;
+  controlPlane: JsonRecord;
+  route: JsonRecord;
+  activeInstanceId: string;
+  singleWriter: boolean;
+  requireAllInstancesHealthy: boolean;
+  expectedNodeIds: string[];
+  endpoints: ManagedEndpoint[];
+}
+
+interface RuntimeConfig {
+  host: string;
+  port: number;
+  logFile: string;
+  manifestPaths: string[];
+  clusterId: string;
+  nodeId: string;
+  kubectlEnabled: boolean;
+  kubectlContext: string;
+  kubeApiProxyEnabled: boolean;
+  kubeconfigPath: string;
+  kubeApiConnectHost: string;
+  requestTimeoutMs: number;
+  healthTimeoutMs: number;
+  services: ManagedService[];
+}
+
+interface KubeApiClient {
+  serverUrl: URL;
+  connectHost: string;
+  caFile: string;
+  certFile: string;
+  keyFile: string;
+}
+
+const recentLogs: JsonRecord[] = [];
+const startedAt = new Date().toISOString();
+const adapterRoot = join(dirname(fileURLToPath(import.meta.url)), "..");
+const config = readConfig();
+const logWriter = config.logFile
+  ? createHourlyJsonlWriter({
+      baseLogFile: config.logFile,
+      service: "v3sctl-adapter",
+      maxBytes: logRetentionBytesForService("v3sctl-adapter"),
+    })
+  : null;
+const kubeClient = loadKubeApiClient();
+logWriter?.prune();
+
+function envString(name: string, fallback: string): string {
+  const value = process.env[name];
+  return value === undefined || value.length === 0 ? fallback : value;
+}
+
+function envNumber(name: string, fallback: number): number {
+  const raw = process.env[name];
+  if (raw === undefined || raw.trim().length === 0) return fallback;
+  const value = Number(raw);
+  return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
+}
+
+function envBool(name: string, fallback: boolean): boolean {
+  const raw = process.env[name];
+  if (raw === undefined || raw.trim().length === 0) return fallback;
+  const normalized = raw.trim().toLowerCase();
+  if (["1", "true", "yes", "on"].includes(normalized)) return true;
+  if (["0", "false", "no", "off"].includes(normalized)) return false;
+  return fallback;
+}
+
+function asRecord(value: unknown, path: string): Record<string, unknown> {
+  if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${path} must be an object`);
+  return value as Record<string, unknown>;
+}
+
+function stringField(value: Record<string, unknown>, key: string, path: string): string {
+  const field = value[key];
+  if (typeof field !== "string" || field.length === 0) throw new Error(`${path}.${key} must be a non-empty string`);
+  return field;
+}
+
+function optionalStringField(value: Record<string, unknown>, key: string, fallback: string): string {
+  const field = value[key];
+  if (field === undefined || field === null || field === "") return fallback;
+  if (typeof field !== "string") throw new Error(`${key} must be a string`);
+  return field;
+}
+
+function optionalBoolField(value: Record<string, unknown>, key: string, fallback: boolean): boolean {
+  const field = value[key];
+  if (field === undefined || field === null) return fallback;
+  if (typeof field !== "boolean") throw new Error(`${key} must be a boolean`);
+  return field;
+}
+
+function isJsonValue(value: unknown): value is JsonValue {
+  if (value === null) return true;
+  if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") return true;
+  if (Array.isArray(value)) return value.every(isJsonValue);
+  if (typeof value === "object") return Object.values(value as Record<string, unknown>).every(isJsonValue);
+  return false;
+}
+
+function manifestJsonRecord(value: unknown, path: string): JsonRecord {
+  if (value === undefined || value === null) return {};
+  if (typeof value !== "object" || Array.isArray(value) || !isJsonValue(value)) throw new Error(`${path} must be a JSON object`);
+  return value as JsonRecord;
+}
+
+function stringArrayField(value: Record<string, unknown>, key: string, fallback: string[]): string[] {
+  const field = value[key];
+  if (field === undefined || field === null) return fallback;
+  if (!Array.isArray(field) || field.some((item) => typeof item !== "string" || item.length === 0)) {
+    throw new Error(`${key} must be an array of non-empty strings`);
+  }
+  return field;
+}
+
+function normalizeRole(value: string): InstanceRole {
+  if (value === "primary" || value === "standby" || value === "worker") return value;
+  return "worker";
+}
+
+function parseEndpoint(value: unknown, index: number, ownerPath = "endpoint"): ManagedEndpoint {
+  const path = `${ownerPath}[${index}]`;
+  const item = asRecord(value, path);
+  const id = stringField(item, "id", path);
+  const nodeId = optionalStringField(item, "nodeId", id);
+  return {
+    id,
+    nodeId,
+    role: normalizeRole(optionalStringField(item, "role", id === "D601" ? "primary" : "standby")),
+    baseUrl: stringField(item, "baseUrl", path).replace(/\/+$/u, ""),
+    healthPath: optionalStringField(item, "healthPath", "/health"),
+  };
+}
+
+function parseManagedKubernetesManifest(value: unknown, index: number, ownerPath = "manifests"): ManagedService {
+  const path = `${ownerPath}[${index}]`;
+  const manifest = asRecord(value, path);
+  const metadata = asRecord(manifest.metadata, `${path}.metadata`);
+  const spec = asRecord(manifest.spec, `${path}.spec`);
+  const kind = optionalStringField(manifest, "kind", "ManagedKubernetesService");
+  const serviceId = stringField(metadata, "name", `${path}.metadata`);
+  if (kind !== "ManagedKubernetesService") throw new Error(`${path}.kind must be ManagedKubernetesService; direct ManagedHttpService manifests are not allowed in pure v3s mode`);
+  const instancesRaw = spec.instances;
+  if (!Array.isArray(instancesRaw) || instancesRaw.length === 0) throw new Error(`${path}.spec.instances must be a non-empty array`);
+  const endpoints = instancesRaw.map((endpoint, endpointIndex) => parseEndpoint(endpoint, endpointIndex, `${path}.spec.instances`));
+  const activeInstanceId = optionalStringField(spec, "activeInstanceId", endpoints[0]?.id ?? serviceId);
+  if (!endpoints.some((endpoint) => endpoint.id === activeInstanceId)) throw new Error(`${path}.spec.activeInstanceId must match one instance id`);
+  const controlPlane = manifestJsonRecord(spec.controlPlane, `${path}.spec.controlPlane`);
+  const route = manifestJsonRecord(spec.route, `${path}.spec.route`);
+  if (String(route.kind ?? "") !== "kubernetes-service") throw new Error(`${path}.spec.route.kind must be kubernetes-service`);
+  return {
+    id: serviceId,
+    namespace: optionalStringField(metadata, "namespace", optionalStringField(spec, "namespace", "unidesk")),
+    kind,
+    controlPlane,
+    route,
+    activeInstanceId,
+    singleWriter: optionalBoolField(spec, "singleWriter", true),
+    requireAllInstancesHealthy: optionalBoolField(spec, "requireAllInstancesHealthy", false),
+    expectedNodeIds: stringArrayField(spec, "expectedNodeIds", endpoints.map((endpoint) => endpoint.nodeId)),
+    endpoints,
+  };
+}
+
+function parseServiceOrManifest(value: unknown, index: number, ownerPath = "services"): ManagedService {
+  const item = asRecord(value, `${ownerPath}[${index}]`);
+  if (typeof item.kind === "string" || item.metadata !== undefined || item.spec !== undefined) {
+    return parseManagedKubernetesManifest(item, index, ownerPath);
+  }
+  throw new Error(`${ownerPath}[${index}] must be a ManagedKubernetesService manifest; static HTTP service declarations are not allowed in pure v3s mode`);
+}
+
+function parseServices(raw: string, ownerPath = "V3SCTL_SERVICES_JSON"): ManagedService[] {
+  const value = raw.trim().length === 0 ? [] : JSON.parse(raw) as unknown;
+  if (!Array.isArray(value)) throw new Error("V3SCTL_SERVICES_JSON must be an array");
+  return value.map((item, index) => parseServiceOrManifest(item, index, ownerPath));
+}
+
+function manifestPaths(raw: string): string[] {
+  return raw.split(",").map((item) => item.trim()).filter((item) => item.length > 0);
+}
+
+function resolveManifestPath(path: string): string {
+  if (isAbsolute(path)) return path;
+  const adapterRelative = join(adapterRoot, path);
+  if (existsSync(adapterRelative)) return adapterRelative;
+  return path;
+}
+
+function readManifestServices(paths: string[]): ManagedService[] {
+  const services: ManagedService[] = [];
+  for (const path of paths) {
+    const resolved = resolveManifestPath(path);
+    const parsed = JSON.parse(readFileSync(resolved, "utf8")) as unknown;
+    const records = Array.isArray(parsed) ? parsed : [parsed];
+    for (const [index, record] of records.entries()) {
+      services.push(parseServiceOrManifest(record, index, `manifest:${path}`));
+    }
+  }
+  return services;
+}
+
+function mergeServices(services: ManagedService[]): ManagedService[] {
+  const seen = new Set<string>();
+  for (const service of services) {
+    const key = `${service.namespace}/${service.id}`;
+    if (seen.has(key)) throw new Error(`duplicate v3s managed service: ${key}`);
+    seen.add(key);
+  }
+  return services;
+}
+
+function readConfig(): RuntimeConfig {
+  const paths = manifestPaths(envString("V3SCTL_MANIFEST_PATHS", "v3s/code-queue.v3s.json"));
+  const inlineServices = parseServices(envString("V3SCTL_SERVICES_JSON", "[]"));
+  const manifestServices = readManifestServices(paths);
+  return {
+    host: envString("HOST", "0.0.0.0"),
+    port: envNumber("PORT", 4266),
+    logFile: envString("LOG_FILE", "/var/log/unidesk/v3sctl-adapter.jsonl"),
+    manifestPaths: paths,
+    clusterId: envString("V3SCTL_CLUSTER_ID", "D601"),
+    nodeId: envString("V3SCTL_NODE_ID", "D601"),
+    kubectlEnabled: envBool("V3SCTL_KUBECTL_ENABLED", false),
+    kubectlContext: envString("V3SCTL_KUBECTL_CONTEXT", ""),
+    kubeApiProxyEnabled: envBool("V3SCTL_KUBE_API_PROXY_ENABLED", true),
+    kubeconfigPath: envString("V3SCTL_KUBECONFIG_PATH", "/var/lib/unidesk/v3s/kubeconfig"),
+    kubeApiConnectHost: envString("V3SCTL_KUBE_API_CONNECT_HOST", "host.docker.internal"),
+    requestTimeoutMs: Math.max(1000, Math.min(120_000, envNumber("V3SCTL_REQUEST_TIMEOUT_MS", 30_000))),
+    healthTimeoutMs: Math.max(500, Math.min(30_000, envNumber("V3SCTL_HEALTH_TIMEOUT_MS", 2500))),
+    services: mergeServices([...manifestServices, ...inlineServices]),
+  };
+}
+
+function log(level: "debug" | "info" | "warn" | "error", event: string, detail: JsonRecord = {}): void {
+  const record: JsonRecord = { at: new Date().toISOString(), service: "v3sctl-adapter", level, event, ...detail };
+  recentLogs.push(record);
+  while (recentLogs.length > 500) recentLogs.shift();
+  try {
+    logWriter?.appendJson(record, new Date(String(record.at)));
+  } catch {
+    // Logging must never break proxying.
+  }
+  const line = JSON.stringify(record);
+  const writer = level === "error" ? console.error : level === "warn" ? console.warn : console.log;
+  writer(line);
+}
+
+function kubeconfigScalar(text: string, key: string): string {
+  const escaped = key.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&");
+  const match = text.match(new RegExp(`^\\s*${escaped}:\\s*([^\\s]+)\\s*$`, "mu"));
+  return match?.[1] ?? "";
+}
+
+function writeKubeSecretFile(dir: string, name: string, base64Value: string): string {
+  const path = join(dir, name);
+  writeFileSync(path, Buffer.from(base64Value, "base64"), { mode: 0o600 });
+  return path;
+}
+
+function loadKubeApiClient(): KubeApiClient | null {
+  if (!config.kubeApiProxyEnabled) return null;
+  if (!existsSync(config.kubeconfigPath)) {
+    log("warn", "kubeconfig_missing", { path: config.kubeconfigPath });
+    return null;
+  }
+  try {
+    const raw = readFileSync(config.kubeconfigPath, "utf8");
+    const server = kubeconfigScalar(raw, "server");
+    const ca = kubeconfigScalar(raw, "certificate-authority-data");
+    const cert = kubeconfigScalar(raw, "client-certificate-data");
+    const key = kubeconfigScalar(raw, "client-key-data");
+    if (server.length === 0 || ca.length === 0 || cert.length === 0 || key.length === 0) throw new Error("kubeconfig must include server, CA, client certificate, and client key data");
+    const dir = join(tmpdir(), `unidesk-v3sctl-kube-${config.clusterId}`);
+    mkdirSync(dir, { recursive: true, mode: 0o700 });
+    const client: KubeApiClient = {
+      serverUrl: new URL(server),
+      connectHost: config.kubeApiConnectHost,
+      caFile: writeKubeSecretFile(dir, "ca.crt", ca),
+      certFile: writeKubeSecretFile(dir, "client.crt", cert),
+      keyFile: writeKubeSecretFile(dir, "client.key", key),
+    };
+    log("info", "kube_api_client_loaded", { kubeconfigPath: config.kubeconfigPath, serverHost: client.serverUrl.hostname, connectHost: client.connectHost });
+    return client;
+  } catch (error) {
+    log("error", "kube_api_client_failed", { kubeconfigPath: config.kubeconfigPath, error: errorToJson(error) });
+    return null;
+  }
+}
+
+function jsonResponse(body: unknown, status = 200, headers: Record<string, string> = {}): Response {
+  return new Response(JSON.stringify(body), {
+    status,
+    headers: { "content-type": "application/json; charset=utf-8", ...headers },
+  });
+}
+
+function errorToJson(error: unknown): JsonRecord {
+  if (error instanceof Error) return { name: error.name, message: error.message, stack: error.stack ?? "" };
+  return { message: String(error) };
+}
+
+function serviceById(id: string): ManagedService | null {
+  return config.services.find((service) => service.id === id) ?? null;
+}
+
+function activeEndpoint(service: ManagedService): ManagedEndpoint {
+  const endpoint = service.endpoints.find((item) => item.id === service.activeInstanceId);
+  if (endpoint === undefined) throw new Error(`active endpoint not found for service ${service.id}: ${service.activeInstanceId}`);
+  return endpoint;
+}
+
+function endpointUrl(endpoint: ManagedEndpoint, targetPath: string, query = ""): string {
+  const base = new URL(endpoint.baseUrl);
+  const upstream = new URL(targetPath, base);
+  upstream.search = query;
+  return upstream.toString();
+}
+
+async function boundedText(response: Response, maxBytes = 1_000_000): Promise<{ text: string; truncated: boolean }> {
+  const reader = response.body?.getReader();
+  if (reader === undefined) return { text: "", truncated: false };
+  const chunks: Uint8Array[] = [];
+  let total = 0;
+  let truncated = false;
+  while (true) {
+    const item = await reader.read();
+    if (item.done) break;
+    total += item.value.byteLength;
+    if (total <= maxBytes) {
+      chunks.push(item.value);
+    } else {
+      truncated = true;
+      const remaining = Math.max(0, maxBytes - (total - item.value.byteLength));
+      if (remaining > 0) chunks.push(item.value.slice(0, remaining));
+      break;
+    }
+  }
+  return { text: Buffer.concat(chunks).toString("utf8"), truncated };
+}
+
+function routeString(service: ManagedService, key: string, fallback: string): string {
+  const value = service.route[key];
+  return typeof value === "string" && value.length > 0 ? value : fallback;
+}
+
+function routeNumber(service: ManagedService, key: string, fallback: number): number {
+  const value = service.route[key];
+  return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
+}
+
+function isKubernetesServiceRoute(service: ManagedService): boolean {
+  return String(service.route.kind ?? "") === "kubernetes-service";
+}
+
+function serviceProxyApiPath(service: ManagedService, targetPath: string): string {
+  const serviceName = routeString(service, "serviceName", service.id);
+  const servicePort = routeNumber(service, "servicePort", 80);
+  const safeTargetPath = targetPath.startsWith("/") ? targetPath : `/${targetPath}`;
+  return `/api/v1/namespaces/${encodeURIComponent(service.namespace)}/services/${encodeURIComponent(`${serviceName}:${servicePort}`)}/proxy${safeTargetPath}`;
+}
+
+function kubeProxyCurlArgs(client: KubeApiClient, method: string, url: URL, headers: Headers, hasBody: boolean, timeoutMs: number): string[] {
+  const args = [
+    "-sS",
+    "--show-error",
+    "--location",
+    "--max-time", String(Math.max(1, Math.ceil(timeoutMs / 1000))),
+    "--request", method,
+    "--cacert", client.caFile,
+    "--cert", client.certFile,
+    "--key", client.keyFile,
+    "--dump-header", "-",
+  ];
+  const port = url.port || (url.protocol === "https:" ? "443" : "80");
+  if ((url.hostname === "127.0.0.1" || url.hostname === "localhost") && client.connectHost.length > 0) {
+    args.push("--connect-to", `${url.hostname}:${port}:${client.connectHost}:${port}`);
+  }
+  for (const [name, value] of headers.entries()) args.push("--header", `${name}: ${value}`);
+  if (hasBody) args.push("--data-binary", "@-");
+  args.push(url.toString());
+  return args;
+}
+
+function parseCurlHeaderBody(output: Buffer): { status: number; contentType: string; bodyText: string } {
+  const text = output.toString("utf8");
+  const separator = text.indexOf("\r\n\r\n") >= 0 ? "\r\n\r\n" : "\n\n";
+  const index = text.indexOf(separator);
+  if (index < 0) return { status: 502, contentType: "text/plain; charset=utf-8", bodyText: text };
+  let headerText = text.slice(0, index);
+  let bodyText = text.slice(index + separator.length);
+  while (/^HTTP\/\d(?:\.\d)?\s+1\d\d\b/mu.test(headerText)) {
+    const nextIndex = bodyText.indexOf(separator);
+    if (nextIndex < 0) break;
+    headerText = bodyText.slice(0, nextIndex);
+    bodyText = bodyText.slice(nextIndex + separator.length);
+  }
+  const status = Number(headerText.match(/^HTTP\/\d(?:\.\d)?\s+(\d+)/mu)?.[1] ?? 502);
+  const contentType = headerText.match(/^content-type:\s*(.+)$/imu)?.[1]?.trim() || "application/octet-stream";
+  return { status: Number.isFinite(status) ? status : 502, contentType, bodyText };
+}
+
+async function kubeApiServiceProxyResponse(
+  service: ManagedService,
+  req: Request,
+  targetPath: string,
+  query: string,
+  timeoutMs: number,
+): Promise<Response> {
+  if (kubeClient === null) {
+    return jsonResponse({ ok: false, error: "kubernetes api proxy is not configured", serviceId: service.id, kubeconfigPath: config.kubeconfigPath, noFallback: true }, 502);
+  }
+  const upstreamUrl = new URL(serviceProxyApiPath(service, targetPath), kubeClient.serverUrl);
+  upstreamUrl.search = query;
+  const headers = forwardHeaders(req);
+  const bodyText = req.method === "GET" || req.method === "HEAD" ? "" : await req.text();
+  const args = kubeProxyCurlArgs(kubeClient, req.method, upstreamUrl, headers, bodyText.length > 0, timeoutMs);
+  const proc = Bun.spawn(["curl", ...args], {
+    stdin: bodyText.length > 0 ? "pipe" : "ignore",
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  if (bodyText.length > 0) {
+    proc.stdin?.write(bodyText);
+    proc.stdin?.end();
+  }
+  const [stdout, stderr, exitCode] = await Promise.all([
+    new Response(proc.stdout).arrayBuffer(),
+    new Response(proc.stderr).text(),
+    proc.exited,
+  ]);
+  if (exitCode !== 0) {
+    log("error", "kube_api_proxy_failed", { serviceId: service.id, targetPath, exitCode, stderr: stderr.slice(0, 2000), noFallback: true });
+    return jsonResponse({ ok: false, error: "kubernetes api service proxy failed", serviceId: service.id, detail: stderr.slice(0, 4000), noFallback: true }, 502);
+  }
+  const parsed = parseCurlHeaderBody(Buffer.from(stdout));
+  return new Response(parsed.bodyText, {
+    status: parsed.status,
+    headers: {
+      "content-type": parsed.contentType,
+      "x-unidesk-proxy-mode": "kubernetes-api-service-proxy",
+      "x-unidesk-v3s-service": service.id,
+      "x-unidesk-response-truncated": "false",
+    },
+  });
+}
+
+async function probeEndpoint(endpoint: ManagedEndpoint): Promise<JsonRecord> {
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), config.healthTimeoutMs);
+  const checkedAt = new Date().toISOString();
+  try {
+    const response = await fetch(endpointUrl(endpoint, endpoint.healthPath), {
+      method: "GET",
+      headers: { accept: "application/json" },
+      signal: controller.signal,
+    });
+    const contentType = response.headers.get("content-type") ?? "application/octet-stream";
+    const bodyText = await response.text();
+    let body: JsonValue = bodyText.slice(0, 2000);
+    try {
+      body = JSON.parse(bodyText) as JsonValue;
+    } catch {
+      // Keep text preview for non-JSON health endpoints.
+    }
+    return {
+      id: endpoint.id,
+      nodeId: endpoint.nodeId,
+      role: endpoint.role,
+      baseUrl: endpoint.baseUrl,
+      healthPath: endpoint.healthPath,
+      healthy: response.ok,
+      status: response.ok ? "healthy" : "unhealthy",
+      upstreamStatus: response.status,
+      contentType,
+      checkedAt,
+      body,
+    };
+  } catch (error) {
+    return {
+      id: endpoint.id,
+      nodeId: endpoint.nodeId,
+      role: endpoint.role,
+      baseUrl: endpoint.baseUrl,
+      healthPath: endpoint.healthPath,
+      healthy: false,
+      status: "unhealthy",
+      upstreamStatus: null,
+      contentType: null,
+      checkedAt,
+      error: error instanceof Error ? error.message : String(error),
+    };
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+async function probeKubernetesServiceActive(service: ManagedService): Promise<JsonRecord> {
+  const endpoint = activeEndpoint(service);
+  const checkedAt = new Date().toISOString();
+  const response = await kubeApiServiceProxyResponse(
+    service,
+    new Request("http://v3sctl-adapter.local/health", { method: "GET", headers: { accept: "application/json" } }),
+    endpoint.healthPath,
+    "",
+    config.healthTimeoutMs,
+  );
+  const contentType = response.headers.get("content-type") ?? "application/octet-stream";
+  const bodyText = await response.text();
+  let body: JsonValue = bodyText.slice(0, 2000);
+  try {
+    body = JSON.parse(bodyText) as JsonValue;
+  } catch {
+    // Health endpoint may return text.
+  }
+  return {
+    id: endpoint.id,
+    nodeId: endpoint.nodeId,
+    role: endpoint.role,
+    baseUrl: endpoint.baseUrl,
+    healthPath: endpoint.healthPath,
+    proxyMode: "kubernetes-api-service-proxy",
+    route: service.route,
+    healthy: response.ok,
+    status: response.ok ? "healthy" : "unhealthy",
+    upstreamStatus: response.status,
+    contentType,
+    checkedAt,
+    body,
+  };
+}
+
+async function serviceStatus(service: ManagedService): Promise<JsonRecord> {
+  const instances = isKubernetesServiceRoute(service)
+    ? [await probeKubernetesServiceActive(service)]
+    : [{
+      id: service.activeInstanceId,
+      nodeId: activeEndpoint(service).nodeId,
+      role: activeEndpoint(service).role,
+      baseUrl: activeEndpoint(service).baseUrl,
+      healthPath: activeEndpoint(service).healthPath,
+      healthy: false,
+      status: "invalid-route",
+      upstreamStatus: null,
+      contentType: null,
+      checkedAt: new Date().toISOString(),
+      error: "v3s managed service route must be kubernetes-service",
+      noFallback: true,
+    }];
+  const active = instances.find((item) => item.id === service.activeInstanceId) ?? null;
+  const activeHealthy = active?.healthy === true;
+  const allInstancesHealthy = instances.every((item) => item.healthy === true);
+  const expectedNodeIds = service.expectedNodeIds;
+  const presentNodeIds = Array.from(new Set(instances.map((item) => String(item.nodeId))));
+  const missingNodeIds = expectedNodeIds.filter((nodeId) => !presentNodeIds.includes(nodeId));
+  const topologyComplete = missingNodeIds.length === 0;
+  const requiredTopologyHealthy = !service.requireAllInstancesHealthy || (topologyComplete && allInstancesHealthy);
+  const healthy = activeHealthy && requiredTopologyHealthy;
+  return {
+    id: service.id,
+    namespace: service.namespace,
+    kind: service.kind,
+    controlPlane: service.controlPlane,
+    route: service.route,
+    activeInstanceId: service.activeInstanceId,
+    singleWriter: service.singleWriter,
+    expectedNodeIds,
+    presentNodeIds,
+    missingNodeIds,
+    topologyComplete,
+    topologyHealthy: topologyComplete && allInstancesHealthy,
+    servingHealthy: activeHealthy,
+    healthy,
+    status: healthy ? (topologyComplete ? "healthy" : "degraded") : "unhealthy",
+    active,
+    instances,
+  };
+}
+
+async function kubectlSnapshot(): Promise<JsonValue> {
+  if (!config.kubectlEnabled) return { enabled: false };
+  const args = ["get", "nodes", "-o", "json"];
+  if (config.kubectlContext.length > 0) args.unshift("--context", config.kubectlContext);
+  const proc = Bun.spawn(["kubectl", ...args], { stdout: "pipe", stderr: "pipe" });
+  const [stdout, stderr, exitCode] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text(), proc.exited]);
+  if (exitCode !== 0) return { enabled: true, ok: false, exitCode, stderr: stderr.slice(0, 4000) };
+  try {
+    const parsed = JSON.parse(stdout) as JsonRecord;
+    const items = Array.isArray(parsed.items) ? parsed.items : [];
+    return { enabled: true, ok: true, nodeCount: items.length, nodes: items.slice(0, 20) as JsonValue };
+  } catch (error) {
+    return { enabled: true, ok: false, error: error instanceof Error ? error.message : String(error), stdout: stdout.slice(0, 4000) };
+  }
+}
+
+async function controlPlaneSnapshot(): Promise<JsonRecord> {
+  const services = await Promise.all(config.services.map(serviceStatus));
+  const managedServicesHealthy = services.every((service) => service.healthy === true);
+  return {
+    ok: true,
+    service: "v3sctl-adapter",
+    clusterId: config.clusterId,
+    nodeId: config.nodeId,
+    startedAt,
+    manifestPaths: config.manifestPaths,
+    managedServicesHealthy,
+    noFallback: true,
+    runtimePath: "frontend -> backend-core -> v3sctl-adapter -> kubernetes api service proxy -> v3s service",
+    kubeApiProxy: {
+      enabled: config.kubeApiProxyEnabled,
+      configured: kubeClient !== null,
+      kubeconfigPath: config.kubeconfigPath,
+      connectHost: config.kubeApiConnectHost,
+      serverHost: kubeClient?.serverUrl.hostname ?? null,
+      mode: "kubernetes-api-service-proxy",
+    },
+    services,
+    kubectl: await kubectlSnapshot(),
+  };
+}
+
+function forwardHeaders(request: Request): Headers {
+  const headers = new Headers();
+  for (const name of ["accept", "content-type", "x-requested-with"]) {
+    const value = request.headers.get(name);
+    if (value !== null) headers.set(name, value);
+  }
+  return headers;
+}
+
+async function proxyToService(service: ManagedService, req: Request, targetPath: string, query: string): Promise<Response> {
+  if (isKubernetesServiceRoute(service)) {
+    return kubeApiServiceProxyResponse(service, req, targetPath, query, config.requestTimeoutMs);
+  }
+  log("error", "v3sctl_route_not_kubernetes_service", { serviceId: service.id, route: service.route, noFallback: true });
+  return jsonResponse({ ok: false, error: "v3s managed service route must be kubernetes-service", serviceId: service.id, route: service.route, noFallback: true }, 500);
+}
+
+async function route(req: Request): Promise<Response> {
+  const url = new URL(req.url);
+  if (req.method === "OPTIONS") return jsonResponse({ ok: true });
+  try {
+    if (url.pathname === "/" || url.pathname === "/health") {
+      const snapshot = await controlPlaneSnapshot();
+      return jsonResponse({
+        ok: true,
+        service: "v3sctl-adapter",
+        clusterId: config.clusterId,
+        nodeId: config.nodeId,
+        startedAt,
+        managedServiceCount: config.services.length,
+        managedServicesHealthy: snapshot.managedServicesHealthy,
+        kubeApiProxyConfigured: kubeClient !== null,
+        noFallback: true,
+      });
+    }
+    if (url.pathname === "/logs" && req.method === "GET") return jsonResponse({ ok: true, logs: recentLogs.slice(-100) });
+    if (url.pathname === "/api/services" && req.method === "GET") {
+      return jsonResponse({ ok: true, clusterId: config.clusterId, services: await Promise.all(config.services.map(serviceStatus)) });
+    }
+    if (url.pathname === "/api/control-plane" && req.method === "GET") return jsonResponse(await controlPlaneSnapshot());
+    const healthMatch = url.pathname.match(/^\/api\/services\/([^/]+)\/health$/u);
+    if (healthMatch !== null && (req.method === "GET" || req.method === "HEAD")) {
+      const service = serviceById(decodeURIComponent(healthMatch[1] ?? ""));
+      if (service === null) return jsonResponse({ ok: false, error: "managed service not found" }, 404);
+      const status = await serviceStatus(service);
+      return req.method === "HEAD" ? new Response(null, { status: status.healthy === true ? 200 : 503 }) : jsonResponse({ ok: status.healthy === true, managedService: status }, status.healthy === true ? 200 : 503);
+    }
+    const proxyMatch = url.pathname.match(/^\/api\/services\/([^/]+)\/proxy(\/.*)$/u);
+    if (proxyMatch !== null) {
+      const service = serviceById(decodeURIComponent(proxyMatch[1] ?? ""));
+      if (service === null) return jsonResponse({ ok: false, error: "managed service not found" }, 404);
+      const targetPath = proxyMatch[2] ?? "/";
+      return await proxyToService(service, req, targetPath, url.search);
+    }
+    return jsonResponse({ ok: false, error: "not found" }, 404);
+  } catch (error) {
+    log("error", "request_failed", { path: url.pathname, error: errorToJson(error) });
+    return jsonResponse({ ok: false, error: error instanceof Error ? error.message : String(error) }, 500);
+  }
+}
+
+Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route });
+log("info", "service_started", { port: config.port, clusterId: config.clusterId, nodeId: config.nodeId, managedServiceCount: config.services.length });

src/components/microservices/v3sctl-adapter/tsconfig.json

@@ -0,0 +1,18 @@
+{
+  "compilerOptions": {
+    "composite": true,
+    "target": "ES2022",
+    "module": "ESNext",
+    "moduleResolution": "Bundler",
+    "types": ["bun", "node"],
+    "strict": true,
+    "noImplicitReturns": true,
+    "noFallthroughCasesInSwitch": true,
+    "declaration": true,
+    "emitDeclarationOnly": true,
+    "outDir": "dist",
+    "skipLibCheck": true
+  },
+  "include": ["src/**/*.ts"],
+  "references": [{ "path": "../../shared" }]
+}

src/components/microservices/v3sctl-adapter/v3s/code-queue.k8s.yaml

@@ -0,0 +1,228 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: unidesk
+  labels:
+    app.kubernetes.io/part-of: unidesk
+    unidesk.ai/v3s-cluster: unidesk-v3s
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: code-queue
+  namespace: unidesk
+  labels:
+    app.kubernetes.io/name: code-queue
+    app.kubernetes.io/part-of: unidesk
+    unidesk.ai/deployment-mode: v3sctl-managed
+    unidesk.ai/instance-id: D601
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: code-queue
+      unidesk.ai/instance-id: D601
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: code-queue
+        app.kubernetes.io/part-of: unidesk
+        unidesk.ai/deployment-mode: v3sctl-managed
+        unidesk.ai/instance-id: D601
+        unidesk.ai/node-id: D601
+    spec:
+      terminationGracePeriodSeconds: 30
+      containers:
+        - name: code-queue
+          image: unidesk-code-queue:d601
+          imagePullPolicy: IfNotPresent
+          ports:
+            - name: http
+              containerPort: 4222
+          envFrom:
+            - secretRef:
+                name: code-queue-env
+                optional: true
+          env:
+            - name: HOST
+              value: "0.0.0.0"
+            - name: PORT
+              value: "4222"
+            - name: CODE_QUEUE_INSTANCE_ID
+              value: "D601"
+            - name: CODE_QUEUE_SCHEDULER_ENABLED
+              value: "true"
+            - name: CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED
+              value: "false"
+            - name: CODE_QUEUE_DATA_DIR
+              value: "/var/lib/unidesk/code-queue"
+            - name: CODE_QUEUE_WORKDIR
+              value: "/workspace"
+            - name: CODE_QUEUE_CODEX_HOME
+              value: "/var/lib/unidesk/code-queue/codex-home"
+            - name: CODE_QUEUE_OPENCODE_XDG_DIR
+              value: "/var/lib/unidesk/code-queue/opencode-xdg"
+            - name: CODE_QUEUE_SOURCE_CODEX_CONFIG
+              value: "/root/.codex/config.toml"
+            - name: CODE_QUEUE_DEFAULT_MODEL
+              value: "gpt-5.5"
+            - name: CODE_QUEUE_MODELS
+              value: "gpt-5.5,gpt-5.4-mini,gpt-5.4,minimax-m2.7"
+            - name: CODE_QUEUE_MODEL_REASONING_EFFORTS
+              value: "gpt-5.5=xhigh"
+            - name: CODE_QUEUE_SANDBOX
+              value: "danger-full-access"
+            - name: CODE_QUEUE_APPROVAL_POLICY
+              value: "never"
+            - name: CODE_QUEUE_MAX_ACTIVE_QUEUES
+              value: "0"
+            - name: CODE_QUEUE_DATABASE_POOL_MAX
+              value: "2"
+            - name: NODE_OPTIONS
+              value: "--max-old-space-size=1024"
+            - name: CODE_QUEUE_IN_MEMORY_OUTPUT_RECORDS
+              value: "10"
+            - name: CODE_QUEUE_IN_MEMORY_EVENT_RECORDS
+              value: "10"
+            - name: CODE_QUEUE_MAIN_PROVIDER_ID
+              value: "D601"
+            - name: CODE_QUEUE_REMOTE_WORKDIR
+              value: "/home/ubuntu"
+            - name: CODE_QUEUE_EXECUTION_PROVIDER_IDS
+              value: "D601"
+            - name: CODE_QUEUE_DEV_CONTAINER_MASTER_HOST
+              value: "74.48.78.17"
+            - name: CODE_QUEUE_DEV_CONTAINER_DEFAULT_PROVIDER_ID
+              value: "D601"
+            - name: CODE_QUEUE_DEV_CONTAINER_WORKDIR
+              value: "/home/ubuntu"
+            - name: CODE_QUEUE_EGRESS_PROXY_ENABLED
+              value: "true"
+            - name: CODE_QUEUE_EGRESS_PROXY_URL
+              value: "http://host.docker.internal:18789"
+            - name: CODE_QUEUE_EGRESS_PROXY_NO_PROXY
+              value: "localhost,127.0.0.1,::1,host.docker.internal,unidesk-provider-gateway-D601,74.48.78.17,backend-core,oa-event-flow,database"
+            - name: HTTP_PROXY
+              value: "http://host.docker.internal:18789"
+            - name: HTTPS_PROXY
+              value: "http://host.docker.internal:18789"
+            - name: ALL_PROXY
+              value: "http://host.docker.internal:18789"
+            - name: http_proxy
+              value: "http://host.docker.internal:18789"
+            - name: https_proxy
+              value: "http://host.docker.internal:18789"
+            - name: all_proxy
+              value: "http://host.docker.internal:18789"
+            - name: NO_PROXY
+              value: "localhost,127.0.0.1,::1,host.docker.internal,unidesk-provider-gateway-D601,74.48.78.17,backend-core,oa-event-flow,database"
+            - name: no_proxy
+              value: "localhost,127.0.0.1,::1,host.docker.internal,unidesk-provider-gateway-D601,74.48.78.17,backend-core,oa-event-flow,database"
+            - name: OA_EVENT_FLOW_BASE_URL
+              value: "http://74.48.78.17:4255"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED
+              value: "true"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL
+              value: "http://host.docker.internal:3290"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_TARGET_TYPE
+              value: "private"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_USER_ID
+              value: "645275593"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_MAX_RESPONSE_CHARS
+              value: "12000"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_TIMEOUT_MS
+              value: "15000"
+            - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_SEND_ATTEMPTS
+              value: "3"
+            - name: LOG_FILE
+              value: "/var/log/unidesk/code-queue.jsonl"
+            - name: UNIDESK_LOG_RETENTION_BYTES
+              value: "1GiB"
+          volumeMounts:
+            - name: docker-sock
+              mountPath: /var/run/docker.sock
+            - name: workspace
+              mountPath: /workspace
+            - name: workspace
+              mountPath: /root/unidesk
+            - name: codex-config
+              mountPath: /root/.codex/config.toml
+              readOnly: true
+            - name: ssh-dir
+              mountPath: /root/.ssh
+              readOnly: true
+            - name: logs
+              mountPath: /var/log/unidesk
+            - name: state
+              mountPath: /var/lib/unidesk/code-queue
+          readinessProbe:
+            httpGet:
+              path: /health
+              port: http
+            periodSeconds: 5
+            timeoutSeconds: 3
+            failureThreshold: 20
+          livenessProbe:
+            httpGet:
+              path: /live
+              port: http
+            periodSeconds: 10
+            timeoutSeconds: 3
+            failureThreshold: 6
+          startupProbe:
+            httpGet:
+              path: /live
+              port: http
+            periodSeconds: 5
+            timeoutSeconds: 3
+            failureThreshold: 60
+          resources:
+            requests:
+              cpu: 250m
+              memory: 512Mi
+            limits:
+              memory: 4Gi
+      volumes:
+        - name: docker-sock
+          hostPath:
+            path: /var/run/docker.sock
+            type: Socket
+        - name: workspace
+          hostPath:
+            path: /home/ubuntu/cq-deploy
+            type: Directory
+        - name: codex-config
+          hostPath:
+            path: /home/ubuntu/.codex/config.toml
+            type: File
+        - name: ssh-dir
+          hostPath:
+            path: /home/ubuntu/.ssh
+            type: Directory
+        - name: logs
+          hostPath:
+            path: /home/ubuntu/cq-deploy/.state/code-queue/logs
+            type: DirectoryOrCreate
+        - name: state
+          hostPath:
+            path: /home/ubuntu/cq-deploy/.state/code-queue
+            type: DirectoryOrCreate
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: code-queue
+  namespace: unidesk
+  labels:
+    app.kubernetes.io/name: code-queue
+    app.kubernetes.io/part-of: unidesk
+    unidesk.ai/deployment-mode: v3sctl-managed
+spec:
+  type: ClusterIP
+  selector:
+    app.kubernetes.io/name: code-queue
+    unidesk.ai/instance-id: D601
+  ports:
+    - name: http
+      port: 4222
+      targetPort: http

src/components/microservices/v3sctl-adapter/v3s/code-queue.v3s.json

@@ -0,0 +1,37 @@
+{
+  "apiVersion": "unidesk.ai/v3s/v1",
+  "kind": "ManagedKubernetesService",
+  "metadata": {
+    "name": "code-queue",
+    "namespace": "unidesk"
+  },
+  "spec": {
+    "adapterServiceId": "v3sctl-adapter",
+    "controlPlane": {
+      "type": "kubernetes",
+      "cluster": "unidesk-v3s",
+      "context": "kind-unidesk-v3s"
+    },
+    "route": {
+      "kind": "kubernetes-service",
+      "serviceName": "code-queue",
+      "servicePort": 4222
+    },
+    "activeInstanceId": "D601",
+    "singleWriter": true,
+    "expectedNodeIds": [
+      "D601",
+      "D518"
+    ],
+    "instances": [
+      {
+        "id": "D601",
+        "nodeId": "D601",
+        "role": "primary",
+        "baseUrl": "kubernetes://unidesk/services/code-queue:4222",
+        "healthPath": "/health"
+      }
+    ],
+    "requireAllInstancesHealthy": false
+  }
+}