From ba0b949766ed186a823c68636646b25632023508 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 16 May 2026 16:47:00 +0000 Subject: [PATCH] feat: split code queue k3s services --- TEST.md | 4 + docs/reference/arch.md | 1 + docs/reference/deployment.md | 4 +- docs/reference/microservices.md | 12 +- docs/reference/provider-gateway.md | 4 +- scripts/src/deploy.ts | 2 +- .../backend-core/src/microservice-proxy.ts | 17 +- .../microservices/code-queue/src/index.ts | 171 +++++- .../microservices/code-queue/src/queue-api.ts | 22 +- .../microservices/code-queue/src/types.ts | 4 + .../k3sctl-adapter/k3s/code-queue.k3s.json | 199 +++++-- .../k3sctl-adapter/k3s/code-queue.k8s.yaml | 549 ++++++++++++++++++ 12 files changed, 907 insertions(+), 82 deletions(-) diff --git a/TEST.md b/TEST.md index 427f7c0f..ce7ee521 100644 --- a/TEST.md +++ b/TEST.md @@ -80,6 +80,10 @@ 阅读 `AGENTS.md`（本项目 `AGENTS.md` 同时承担 `SKILL.md` 对 `scripts/cli.ts` 的解释职责），然后用 cli 手动测试以下内容：运行 `bun scripts/cli.ts e2e run`，确认 `provider:gateway-version-label`、`frontend:gateway-version-records-visible`、`frontend:gateway-duration-subsecond-visible` 和 `frontend:provider-operation-availability-visible` passed；再登录公网 frontend，进入 `资源节点 / 网关版本`，确认每个 Provider 行都显示 provider-gateway 版本号、升级策略、SSH 透传可用性、远程更新可用性、能力摘要、最近远程更新记录，并在下方以表格记录 `provider.upgrade` 的状态、模式、任务 id、来源、耗时、策略、指定 Provider 的 gateway 版本号、结果摘要和更新时间。远程更新记录的真实亚秒级耗时必须显示小数秒，不得显示成 `0s`；远程更新记录默认必须是结构化控件，不得展示裸 JSON；完整 task/result 只能通过 `查看原始JSON` 按钮查看。 +## T18A Provider Gateway HTTP Tunnel + +阅读 `AGENTS.md`（本项目 `AGENTS.md` 同时承担 `SKILL.md` 对 `scripts/cli.ts` 的解释职责），然后用 cli 手动测试以下内容：如果本次变更修改了 `src/components/provider-gateway`、backend-core 用户服务代理或 shared provider 协议，运行 `bun scripts/cli.ts debug health` 或公网 frontend 节点原始 JSON，确认目标 Provider labels 的 `unideskCapabilities` 同时包含 `microservice.http` 与 `microservice.http.tunnel`，且 `providerGatewayVersion` 已上报目标版本。随后对 D601 这类直管用户服务运行 `bun scripts/cli.ts microservice health findjob` 或任一已登记 `unidesk-direct` 用户服务的 health/proxy 命令，确认响应成功且代理诊断头或任务结果显示走 provider WebSocket HTTP tunnel；若目标 Provider 尚未升级到 tunnel capability，响应可以临时走旧 `microservice.http` dispatch，但验收记录必须标明该 Provider 仍需升级。重复执行 10 次同一只读 proxy 请求，不得出现偶发 `upstream proxy failed`、provider task missing 或 WebSocket socket closed 的 502。 + ## T19 前端单服务重建 阅读 `AGENTS.md`（本项目 `AGENTS.md` 同时承担 `SKILL.md` 对 `scripts/cli.ts` 的解释职责），然后用 cli 手动测试以下内容：运行 `bun scripts/cli.ts server rebuild frontend`，确认命令立即返回 `server_rebuild` job id；随后运行 `bun scripts/cli.ts job status latest` 直到状态为 `succeeded`，stdout 中必须能看到先 build、再按 `frontend` 服务容器 label 移除、最后 `--no-deps frontend` 启动的过程。重建后运行 `bun scripts/cli.ts server status` 和 `bun scripts/cli.ts e2e run`，确认公网 frontend 恢复健康、Playwright 登录通过、database 命名卷未被删除；正式验收不得要求人工执行 `docker rm` 作为兜底。 diff --git a/docs/reference/arch.md b/docs/reference/arch.md index 9eaee7fb..c235b919 100644 --- a/docs/reference/arch.md +++ b/docs/reference/arch.md @@ -74,6 +74,7 @@ - All state changes are reported to the main server in real time by Provider Gateway - The main server writes state updates to PostgreSQL, completing the unified closed loop - Pipeline workflow control follows the OA event-flow model: OA is the only control bus, factual node events remain policy-neutral, and runner/monitor/frontend/CLI actions are represented as OA events; detailed constraints live in `docs/reference/pipeline-oa-event-flow.md` + - Queue-oriented user services must separate read, write, and scheduler responsibilities once they move behind Kubernetes Service routing. Read services may scale and roll independently from PostgreSQL-backed state, write services validate and persist commands, and scheduler services own only task claiming, active run control, and worker/runner lifecycle. A scheduler/runner bridge must not be hidden inside the read path, and a bridge service such as `k3sctl-adapter` must stay outside the k3s-managed workload graph. - Single Authoritative Path Discipline - Every durable capability must declare one authoritative data, control, event, or network path in reference docs, API contracts, health output, and E2E checks; frontend, CLI, worker, and service code must consume that path directly. - Hidden fallback is forbidden. A service must not silently answer from local JSON, transcripts, in-memory state, stale task caches, direct host URLs, old service IDs, alternate proxies, local direct internet, or legacy event streams when the documented path is unavailable. diff --git a/docs/reference/deployment.md b/docs/reference/deployment.md index 2bdc64fb..02cf488b 100644 --- a/docs/reference/deployment.md +++ b/docs/reference/deployment.md @@ -10,7 +10,7 @@ - `provider-gateway` 是当前主 server 的本机计算节点代理，通过 WebSocket 主动连到 provider ingress，挂载 `/var/run/docker.sock` 作为自动任务执行主路径，使用 `pid: "host"` 读取节点级进程资源，并周期性上报系统资源指标、进程占用与 Docker daemon 状态；计算节点 provider-gateway 还必须把 egress proxy 仅发布到宿主 loopback `127.0.0.1:18789` 供本节点执行环境出网，维护用 Host SSH / WSL SSH 私钥目录只读挂载到 `/run/host-ssh`，不得作为自动任务调度主路径。 - `todo-note` 是主 server 承载的 Todo Note 纯后端用户服务，容器名 `todo-note-backend`，只在 Compose 内网暴露 `4211/tcp`，使用主 PostgreSQL 存储迁移后的 Todo Note 数据。 - `k3sctl-adapter` 是 D601 上由 UniDesk 直管的 k3s 控制面适配微服务，容器名 `k3sctl-adapter`，只绑定 `127.0.0.1:4266`，由 UniDesk frontend/backend-core 通过用户服务代理访问并提供 `/api/control-plane` 可见性。 -- `code-queue` 是由 D601 k3s/k8s 控制面代管的 Codex/OpenCode 队列用户服务，active/standby 实例通过 `src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json` 声明，运行对象通过 `code-queue.k8s.yaml` 创建 Kubernetes Deployment/ClusterIP Service；任务、queue、未读状态、控制状态和通知 outbox 一律写入主 PostgreSQL，不保留本地状态文件 fallback，浏览器只能通过 UniDesk frontend -> backend-core -> k3sctl-adapter -> Kubernetes API service proxy -> Code Queue Service 查看运行输出、追加 prompt、打断和重试。 +- `code-queue` 是由 D601 k3s/k8s 控制面代管的 Codex/OpenCode 队列用户服务，active/standby 和 read/write/scheduler 服务通过 `src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json` 声明，运行对象通过 `code-queue.k8s.yaml` 创建 Kubernetes Deployment/ClusterIP Service；任务、queue、未读状态、控制状态和通知 outbox 一律写入主 PostgreSQL，不保留本地状态文件 fallback。浏览器只使用稳定的 `code-queue` 用户服务 ID，backend-core 在内部把只读请求分到 `code-queue-read`，把命令写入分到 `code-queue-write`，把执行端 `/health`、dev-container 和 running task control 分到 `code-queue-scheduler`。 - `project-manager` 是主 server 承载的项目管理用户服务，容器名 `project-manager-backend`，仅在 Compose 内网暴露 `4233/tcp`，项目清单写入主 PostgreSQL，浏览器只能通过 UniDesk frontend 同源代理执行增删改查、Excel 导入和 Excel 导出。 - `baidu-netdisk` 是主 server 承载的百度网盘存储用户服务，容器名 `baidu-netdisk-backend`，仅在 Compose 内网暴露 `4244/tcp`，OAuth/token/transfer 状态写入主 PostgreSQL，浏览器只能通过 UniDesk frontend 同源代理执行设备码登录、文件浏览和 staging 传输任务控制。 @@ -52,7 +52,7 @@ frontend 的 Docker 上线顺序为：先运行必要的本地校验，例如 `b ## Health Criteria -服务跑通的最低标准是：backend-core 内网 `/health` 返回 ok，frontend 公网 `/health` 返回 ok，provider ingress 公网 `/health` 返回 ok，database 在容器内 `pg_isready` 可用，Todo Note 后端 `/api/health` 返回 `storage=postgres`，`k3sctl-adapter` `/api/control-plane` 可见 `unidesk-k3s` Kubernetes API service proxy 状态、D601 active serving healthy、D518 standby pod ready、`presentNodeIds=[D601,D518]`、`missingNodeIds=[]` 和 no-fallback 路径，Code Queue `/health` 经 k3s active Service 返回轻量 readiness、默认模型、`queue.storage` 和 `egressProxy.connected=true`，`/api/tasks/overview` 返回 PostgreSQL 队列总览，Project Manager `/health` 返回 `storage.primary=postgres` 和项目数量，backend-core `/api/performance` 返回性能指标，`/api/nodes` 中出现 `main-server`、`D601` 和 `D518` provider 且状态为 `online`，`/api/nodes/system-status` 中出现对应 CPU/内存/硬盘采样，`/api/nodes/docker-status` 中能看到主 server、D601 与 D518 Docker 快照。D518 必须通过原生 k3s agent 加入原生 k3s 控制面并运行 `code-queue-d518` standby Pod；不得用 Docker 化 k3s、D601->D518 直连、NodePort 或 provider-gateway business HTTP 绕过 Kubernetes service route。交付前还必须运行 `bun scripts/cli.ts e2e run`，并以 `docs/reference/e2e.md` 的门禁作为最终判定。 +服务跑通的最低标准是：backend-core 内网 `/health` 返回 ok，frontend 公网 `/health` 返回 ok，provider ingress 公网 `/health` 返回 ok，database 在容器内 `pg_isready` 可用，Todo Note 后端 `/api/health` 返回 `storage=postgres`，`k3sctl-adapter` `/api/control-plane` 可见 `unidesk-k3s` Kubernetes API service proxy 状态、D601 scheduler serving healthy、D601 read/write Service healthy、D518 standby pod ready、`presentNodeIds=[D601,D518]`、`missingNodeIds=[]` 和 no-fallback 路径，Code Queue `/health` 经 k3s scheduler Service 返回轻量 readiness、默认模型、`queue.storage` 和 `egressProxy.connected=true`，`/api/tasks/overview` 经 `code-queue-read` 返回 PostgreSQL 队列总览，写入类入口经 `code-queue-write` 入库后由 `code-queue-scheduler` 轮询并执行，Project Manager `/health` 返回 `storage.primary=postgres` 和项目数量，backend-core `/api/performance` 返回性能指标，`/api/nodes` 中出现 `main-server`、`D601` 和 `D518` provider 且状态为 `online`，`/api/nodes/system-status` 中出现对应 CPU/内存/硬盘采样，`/api/nodes/docker-status` 中能看到主 server、D601 与 D518 Docker 快照。D518 必须通过原生 k3s agent 加入原生 k3s 控制面并运行 `code-queue-d518` standby Pod；不得用 Docker 化 k3s、D601->D518 直连、NodePort 或 provider-gateway business HTTP 绕过 Kubernetes service route。交付前还必须运行 `bun scripts/cli.ts e2e run`，并以 `docs/reference/e2e.md` 的门禁作为最终判定。 ## Code Queue D601 Resource Budget diff --git a/docs/reference/microservices.md b/docs/reference/microservices.md index 7bf61175..4e14700a 100644 --- a/docs/reference/microservices.md +++ b/docs/reference/microservices.md @@ -145,9 +145,11 @@ Baidu Netdisk 在 UniDesk 语境中按纯后端服务管理：不得暴露百度 当前 Code Queue 作为 `id=code-queue` 的 `k3sctl-managed` 用户服务登记在 `config.json`，业务实例由 D601 k3s 控制面代管，并接入统一 `oa-event-flow` 发布 Trace/STEP 事实事件与读取统计中心： -- Orchestrator：`deployment.mode=k3sctl-managed`，`deployment.adapterServiceId=k3sctl-adapter`，`deployment.k3sServiceId=code-queue`，`backend.proxyMode=k3sctl-adapter-http`，`backend.nodeBaseUrl=k3s://code-queue`；backend-core 对 Code Queue 的正式链路只能是 `frontend -> backend-core -> k3sctl-adapter -> Kubernetes API service proxy -> Kubernetes Service code-queue:4222`。 +- Orchestrator：`deployment.mode=k3sctl-managed`，`deployment.adapterServiceId=k3sctl-adapter`，`deployment.k3sServiceId=code-queue`，`backend.proxyMode=k3sctl-adapter-http`，`backend.nodeBaseUrl=k3s://code-queue`；backend-core 对 Code Queue 的正式链路只能是 `frontend -> backend-core -> k3sctl-adapter -> Kubernetes API service proxy -> Kubernetes Service ...:4222`。对外登记的 `code-queue` ID 保持稳定，但内部必须拆成 `code-queue-read`、`code-queue-write` 和 `code-queue-scheduler` 三个 Kubernetes Service，并由 backend-core 按 method/path 分流。 - Direct path ban：`code-queue` 不得再登记 `http://code-queue:4222`、`http://host.docker.internal:4222`、NodePort 或 provider-gateway `microservice.http` 作为业务代理目标；frontend 也不得使用旧 `/api/code-queue-direct` 兼容别名作为 Code Queue 页面数据源。provider-gateway 只允许用于维护 D601/D518、部署 adapter、部署 k3s/k8s 节点或诊断节点本机容器。 -- 实例语义：D601 是默认 active/single-writer 实例，`CODE_QUEUE_INSTANCE_ID=D601` 且 `CODE_QUEUE_SCHEDULER_ENABLED=true`；D518 是 standby 实例，必须设置 `CODE_QUEUE_INSTANCE_ID=D518`、`CODE_QUEUE_SCHEDULER_ENABLED=false` 和 `CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED=false`，避免两个实例同时消费同一 PostgreSQL 队列或重复回放 OA 统计。D601 active 也默认关闭 `CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED`；历史 OA Trace/STEP 回填必须通过显式 `/api/oa/backfill` 运维动作触发，不能在每次 Pod 重启时自动批量发布旧事件。 +- 服务拆分语义：`code-queue-read` 只承载 GET/HEAD 查询、overview、任务详情、Trace/output/transcript、统计和只读健康，可多副本滚动更新；它必须设置 `CODE_QUEUE_SERVICE_ROLE=read` 与 `CODE_QUEUE_SCHEDULER_ENABLED=false`，且不得接受入队、queue 变更、已读、重试、移动、追加 prompt 或打断这类 mutation。`code-queue-write` 承载入队、queue 创建/合并/更新、已读、手动重试、移动等命令写入，初期保持单副本和 `CODE_QUEUE_SERVICE_ROLE=write`，只把命令和任务状态写入 PostgreSQL，不启动 agent 子进程。`code-queue-scheduler` 是唯一拥有 scheduler 和 active run 的执行服务，设置 `CODE_QUEUE_SERVICE_ROLE=scheduler` 与 `CODE_QUEUE_SCHEDULER_ENABLED=true`，负责从 PostgreSQL 热任务集轮询新写入任务、推进队列、启动 Codex/OpenCode、处理 running task 的 steer/interrupt、发送终态通知和暴露执行端 `/health`。普通 Service 负载均衡不得把 mutation 打到 read，也不得把 running task 控制打到 write。 +- 实例语义：D601 是默认 active/single-writer 执行节点，当前 `code-queue-scheduler` 仍以一个 scheduler Pod 承载长生命周期 Codex/OpenCode 子进程；D518 是 standby/read-only 节点，必须设置 `CODE_QUEUE_SERVICE_ROLE=read`、`CODE_QUEUE_SCHEDULER_ENABLED=false` 和 `CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED=false`，避免两个实例同时消费同一 PostgreSQL 队列或重复回放 OA 统计。D601 scheduler 也默认关闭 `CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED`；历史 OA Trace/STEP 回填必须通过显式 `/api/oa/backfill` 运维动作触发，不能在每次 Pod 重启时自动批量发布旧事件。 +- 滚动更新边界：read/write/scheduler 三服务拆分可以保证滚动更新期间 Code Queue 的只读 API 与大部分控制面入口可用，但当前 scheduler Pod 内仍直接承载正在运行的 agent 子进程，scheduler Pod 被替换时 active task 仍会进入 restart-recovery/retry 语义，不能宣称 running task 零中断。真正的长期目标是继续把调度器和执行器拆开：scheduler 只负责 claim task 并创建 Kubernetes Job/Pod 或独立 worker，runner 把输出、状态、attempt、事件和通知写回 PostgreSQL/OA Event Flow/归档；只有这样 controller/scheduler 滚动更新才不会影响正在执行的任务。 - 部署引用：Code Queue 镜像仍复用 `src/components/microservices/code-queue/Dockerfile`，Kubernetes 运行清单为 `src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml`，`config.json` 对外记录 k3s manifest `src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json`；主 server 根目录 `docker-compose.yml` 不包含 `code-queue` service，旧 D601 direct Compose 文件只作为迁移/本地诊断参考，不是正式运行入口。 - 主服务依赖映射：Code Queue 仍以主 PostgreSQL 为权威数据库，但 D601 k3s Pod 不能依赖公网直连 `74.48.78.17:15432/4255`。Pod 内 `DATABASE_URL` 和 `OA_EVENT_FLOW_BASE_URL` 必须指向集群内 `d601-tcp-egress-gateway` Service，再由该 gateway 通过 D601 provider-gateway egress proxy 的 HTTP CONNECT 转发到主 PostgreSQL 和 OA Event Flow；新增 TCP 依赖时扩展 `TCP_EGRESS_ROUTES`，不得在业务容器里新增一次性公网直连或 ad hoc 隧道。D601 active 实例的 `CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL` 直接使用本机 ClaudeQQ 映射 `http://host.docker.internal:3290`。这些端口映射只服务受控节点运行时，必须用防火墙或等价策略限制来源，不得成为浏览器或任意公网客户端入口。 - K8s 探针与启动维护：Kubernetes liveness/startup probe 必须使用轻量 `/live`，readiness 和用户服务健康使用 `/health`；`/health` 不得执行全量任务聚合、历史回填或长事务索引维护，历史任务总览应由 `/api/tasks/overview` 读取 PostgreSQL。启动时允许后台执行队列元数据 flush、通知 outbox 读取、任务表索引维护和 overview warmup，但这些维护不得阻塞 Bun server、readiness endpoint 或 frontend overview；通知表索引和大批量 OA backfill 不得作为默认启动副作用。 @@ -297,8 +299,8 @@ ClaudeQQ 在 UniDesk 语境中按消息网关后端服务管理：不得直接 - `bun scripts/cli.ts microservice health claudeqq`、`bun scripts/cli.ts microservice proxy claudeqq /api/napcat/login`、`bun scripts/cli.ts microservice proxy claudeqq /api/events/recent` 和 `bun scripts/cli.ts microservice proxy claudeqq /api/events/subscriptions`：验证 ClaudeQQ 后端、NapCat 容器登录、事件订阅和私有代理链路；消息推送使用 `POST /api/push/text`，不得开放 D601 `3290/3000/3001/6099` 公网端口。 - `bun scripts/cli.ts microservice health todo-note` 与 `bun scripts/cli.ts microservice proxy todo-note /api/instances`：验证主 server Todo Note 后端、PostgreSQL 存储和本机 provider-gateway 私有代理链路。 - `bun scripts/cli.ts microservice health oa-event-flow`、`bun scripts/cli.ts microservice proxy oa-event-flow /api/diagnostics --raw` 与 `bun scripts/cli.ts microservice proxy oa-event-flow '/api/events?tags=service:code-queue&limit=20' --raw`：验证统一 OA 事件流、事件表、tag 查询和统计中心。 -- `bun scripts/cli.ts microservice health k3sctl-adapter` 与 `bun scripts/cli.ts microservice proxy k3sctl-adapter /api/control-plane --raw`：验证 D601 `unidesk-k3s` 控制面 adapter、manifest、D601 active/D518 standby 实例状态、`presentNodeIds=[D601,D518]`、`missingNodeIds=[]` 和 no-fallback 运行路径。 -- `bun scripts/cli.ts microservice health code-queue` 与 `bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview`：验证 Code Queue 经过 backend-core -> k3sctl-adapter -> k3s active service 的单一路径；输出不得出现 `serviceId=code-queue` 的 provider-gateway `microservice.http` 业务代理任务，写入、追加 prompt、打断和 readAt/未读状态都必须由 backend 写入 PostgreSQL，frontend 不得用本地存储伪造成功状态。 +- `bun scripts/cli.ts microservice health k3sctl-adapter` 与 `bun scripts/cli.ts microservice proxy k3sctl-adapter /api/control-plane --raw`：验证 D601 `unidesk-k3s` 控制面 adapter、manifest、D601 scheduler、D601 read/write、D518 standby 实例状态、`presentNodeIds=[D601,D518]`、`missingNodeIds=[]` 和 no-fallback 运行路径。 +- `bun scripts/cli.ts microservice health code-queue` 与 `bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview`：验证 Code Queue 经过 backend-core -> k3sctl-adapter -> k3s Service proxy 的单一路径，其中 `/health` 指向 `code-queue-scheduler`，overview/详情只读请求指向 `code-queue-read`，写入类请求指向 `code-queue-write`；输出不得出现 `serviceId=code-queue` 的 provider-gateway `microservice.http` 业务代理任务，写入、追加 prompt、打断和 readAt/未读状态都必须由 backend 写入 PostgreSQL，frontend 不得用本地存储伪造成功状态。 - `bun scripts/cli.ts microservice health filebrowser`、`bun scripts/cli.ts microservice health filebrowser-d601` 与 `bun scripts/cli.ts microservice proxy filebrowser / --max-body-bytes 2000`：验证 D518 主 File Browser 和 D601 备用 File Browser 私有代理链路；浏览器 WebUI 必须通过 `/api/microservices/filebrowser/proxy/` 或 `/api/microservices/filebrowser-d601/proxy/` 访问，不得直接开放 `4251` 公网端口。 - `bun scripts/cli.ts --main-server-ip 74.48.78.17 microservice health findjob`：在计算节点或其他非主 server 主机上通过公网 frontend remote CLI 进行同一验证，不需要主 server SSH key。 @@ -325,7 +327,7 @@ ClaudeQQ 在 UniDesk 语境中按消息网关后端服务管理：不得直接 - 运行 `bun scripts/cli.ts microservice health met-nonlinear`、`bun scripts/cli.ts microservice proxy met-nonlinear /api/queue`、`bun scripts/cli.ts microservice proxy met-nonlinear '/api/projects?root=projects&limit=20'` 和 `bun scripts/cli.ts microservice proxy met-nonlinear /api/images`，确认真实链路经过 backend-core、WebSocket、D601 provider-gateway 和 D601 本机 MET Nonlinear TS 后端。 - 运行 `bun scripts/cli.ts microservice health claudeqq`、`bun scripts/cli.ts microservice proxy claudeqq /api/napcat/login`、`bun scripts/cli.ts microservice proxy claudeqq /api/events/recent` 和 `bun scripts/cli.ts microservice proxy claudeqq /api/events/subscriptions`，确认真实链路经过 backend-core、WebSocket、D601 provider-gateway 和 D601 本机 ClaudeQQ 后端；在 D601 上 `curl http://127.0.0.1:3290/health` 应显示 `service=claudeqq`、`pureBackend=true`、`napcat.containerized=true`、NapCat HTTP/WS 状态、二维码状态和订阅计数。 - 运行 `bun scripts/cli.ts microservice health todo-note` 与 `bun scripts/cli.ts microservice proxy todo-note /api/instances`，确认真实链路经过 backend-core、WebSocket、main-server provider-gateway 和主 server `todo-note-backend` 后端；输出中必须包含五个迁移清单和 PostgreSQL 存储健康状态。 -- 运行 `bun scripts/cli.ts microservice health k3sctl-adapter`、`bun scripts/cli.ts microservice proxy k3sctl-adapter /api/control-plane --raw`、`bun scripts/cli.ts microservice health code-queue` 与 `bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview`，确认真实链路经过 backend-core -> k3sctl-adapter -> k3s active service；adapter 验收还必须证明其作为 UniDesk 直管服务运行在 k3s 外部，Docker 形态下挂载宿主 `/etc/rancher/k3s/k3s.yaml` 与 `/run/host-ssh/id_ed25519`，通过容器内 SSH local tunnel 连接 WSL 原生 k3s API，且没有 active `rancher/k3s` 控制面容器。Code Queue `/health` 必须仍返回业务后端自己的 `queue.storage.primary=postgres`、`queue.storage.postgresReady=true`、`queue.notifications.claudeqq.outbox.storage=postgres` 和 `egressProxy.connected=true`，不得被 adapter 聚合健康 JSON 替代。还必须在 active Code Queue Pod 内验证主 PostgreSQL 端口映射、主 OA Event Flow 端口映射、本机 ClaudeQQ `http://host.docker.internal:3290` 和 `d601-provider-egress-proxy` 均可访问，并确认 `/workspace` 与 `/home/ubuntu` 指向同一 WSL home hostPath，`/workspace/cq-deploy` 这类绝对 symlink 可以进入真实目录。再在 adapter 控制页确认 D601 active serving healthy、D518 standby pod ready、`missingNodeIds=[]` 且整体不退化为 hidden fallback。再通过公网 frontend 提交一个 `gpt-5.5` 小任务，确认队列串行推进、输出实时更新、结束后有 judge 判定，且运行中可追加 prompt 或打断。Code Queue 的重启恢复必须作为验收项：运行中任务存在时重启或重建 active 实例后，任务必须从 PostgreSQL 恢复到可继续执行状态，不能丢失 active task、`promptHistory`、后续 queued 任务、readAt/未读状态或已入 outbox 的 ClaudeQQ 通知。Code Queue 服务名、表名前缀或持久化目录发生迁移后，还必须运行 `bun scripts/cli.ts e2e run --only microservice:catalog-code-queue,microservice:code-queue-status,microservice:code-queue-health,microservice:code-queue-tasks`，证明 backend-core catalog、k3s adapter 私有代理、PostgreSQL 队列和任务列表都指向 `code-queue`。批量验收必须通过公网 frontend 设置 `入队份数=5` 或使用多段 prompt 分隔，一次性入队 5 条任务，并确认 5 条任务按顺序进入 running/judging/succeeded，而不是只运行第一条。 +- 运行 `bun scripts/cli.ts microservice health k3sctl-adapter`、`bun scripts/cli.ts microservice proxy k3sctl-adapter /api/control-plane --raw`、`bun scripts/cli.ts microservice health code-queue` 与 `bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview`，确认真实链路经过 backend-core -> k3sctl-adapter -> k3s Service proxy，且 read/write/scheduler 三个内部 Service 都有 ready endpoint；adapter 验收还必须证明其作为 UniDesk 直管服务运行在 k3s 外部，Docker 形态下挂载宿主 `/etc/rancher/k3s/k3s.yaml` 与 `/run/host-ssh/id_ed25519`，通过容器内 SSH local tunnel 连接 WSL 原生 k3s API，且没有 active `rancher/k3s` 控制面容器。Code Queue `/health` 必须仍返回业务后端自己的 `role=scheduler`、`queue.storage.primary=postgres`、`queue.storage.postgresReady=true`、`queue.notifications.claudeqq.outbox.storage=postgres` 和 `egressProxy.connected=true`，不得被 adapter 聚合健康 JSON 替代。还必须在 active Code Queue Pod 内验证主 PostgreSQL 端口映射、主 OA Event Flow 端口映射、本机 ClaudeQQ `http://host.docker.internal:3290` 和 `d601-provider-egress-proxy` 均可访问，并确认 `/workspace` 与 `/home/ubuntu` 指向同一 WSL home hostPath，`/workspace/cq-deploy` 这类绝对 symlink 可以进入真实目录。再在 adapter 控制页确认 D601 scheduler serving healthy、D601 read/write Service healthy、D518 standby pod ready、`missingNodeIds=[]` 且整体不退化为 hidden fallback。再通过公网 frontend 提交一个 `gpt-5.5` 小任务，确认任务由 write 入库、scheduler 轮询执行、read 返回输出实时更新、结束后有 judge 判定，且运行中可追加 prompt 或打断。Code Queue 的重启恢复必须作为验收项：运行中任务存在时重启或重建 scheduler 实例后，任务必须从 PostgreSQL 恢复到可继续执行状态，不能丢失 active task、`promptHistory`、后续 queued 任务、readAt/未读状态或已入 outbox 的 ClaudeQQ 通知。Code Queue 服务名、表名前缀或持久化目录发生迁移后，还必须运行 `bun scripts/cli.ts e2e run --only microservice:catalog-code-queue,microservice:code-queue-status,microservice:code-queue-health,microservice:code-queue-tasks`，证明 backend-core catalog、k3s adapter 私有代理、PostgreSQL 队列和任务列表都指向 `code-queue`。批量验收必须通过公网 frontend 设置 `入队份数=5` 或使用多段 prompt 分隔，一次性入队 5 条任务，并确认 5 条任务按顺序进入 running/judging/succeeded，而不是只运行第一条。 - Code Queue 内存防回归验收：凡是改动 Code Queue 的持久化、scheduler、输出/Trace、health、列表/详情查询、日志导出或容器运行参数，交付前必须在 D601 用 `kubectl -n unidesk get deploy,pod,svc,endpoints -o wide`、`kubectl -n unidesk describe deploy/code-queue` 或等价 Docker inspect 确认 memory/swap 硬上限符合预算，运行 `kubectl -n unidesk top pod` 或 Docker stats 确认常驻内存、`OOMKilled=false` 和 `RestartCount` 未异常增长，再运行 `bun scripts/cli.ts microservice health code-queue` 确认 `/health` 是轻量 readiness 且暴露 PostgreSQL/notification/outbox 状态。验收还必须覆盖有历史任务存在时的 `/api/tasks/overview`、单任务详情和 output/transcript 查询，证明热状态裁剪不会丢历史输出、也不会重新把全部历史 `task_json` 缓存在进程内；涉及 TypeScript/frontend 验证的任务应能在 D601 Code Queue memory/swap 预算中完成 `bun run --cwd src/components/frontend check` 这类短时高内存命令，而不是被 memory watchdog 反复 SIGTERM。 - Code Queue 延迟防回归验收：凡是改动 Code Queue 列表、overview、readAt、Trace/summary 懒加载、实时 output/SSE 事件发布、frontend 请求策略、backend-core 用户服务代理或 frontend Code Queue 请求路径，交付前必须在有历史任务数据且有 active output 流动的 live 环境验证 `GET /api/tasks/overview`、`POST /api/tasks//read`、选定 task 的 `trace-step` 和前端 `/app/code-queue/` 首屏均低于 1s 目标；可运行 `bun scripts/src/code-queue-perf.ts --json --target-ms 1000` 采集公网 frontend 下的首屏耗时、最慢 API 和 DOM 完成指标，并用 `bun scripts/cli.ts microservice proxy code-queue /api/tasks/overview --raw`、D601 Pod `/health` 与 `/api/tasks/overview` curl、性能面板 `/api/performance` 与 `/api/frontend-performance` 失败/慢操作记录、`kubectl -n unidesk top pod` 或 Docker stats 补充后端耗时、代理 502 和内存/CPU 证据。验收结论必须同时说明是否使用了短 TTL cache、cache 如何被 mutation 或 archive append 失效、数据库索引/聚合是否命中、输出热路径是否只读增量指标，以及分页加载是否跳过 selected/active/stats；不能只展示 cache 命中后的单次快照。 - 运行 `bun scripts/cli.ts microservice health filebrowser`、`bun scripts/cli.ts microservice health filebrowser-d601` 和 `bun scripts/cli.ts microservice proxy filebrowser / --max-body-bytes 2000`，确认 File Browser health 返回 `status=OK`，WebUI HTML 包含 `File Browser`，D518/D601 通过 provider-gateway 访问节点本机 `4251`；随后在公网 frontend 的 `用户服务 / File Browser` 中确认 D518 为默认目标、可导出截图、iframe 紧凑布局不再有巨大 `folder` 标记遮挡文件名，并可浏览 `/mnt/c`。 diff --git a/docs/reference/provider-gateway.md b/docs/reference/provider-gateway.md index ba24bb1f..b04b89f1 100644 --- a/docs/reference/provider-gateway.md +++ b/docs/reference/provider-gateway.md @@ -72,7 +72,9 @@ WSL 节点还应补充一次真实调度验证：向该 `PROVIDER_ID` 下发 `do SSH 透传自测是 provider-gateway 部署验收的一部分。目标 Provider 在线后，先确认 frontend 节点清单或 `debug health` 中该节点 labels 显示 `hostSshConfigured=true`、`hostSshKeyPresent=true` 且能力包含 `host.ssh`；再运行 `bun scripts/cli.ts debug dispatch host.ssh --wait-ms 15000`，任务必须 `succeeded`，result 中 `probeLine` 必须包含 `UNIDESK_SSH_TEST` 且 `exitCode=0`；最后运行 `bun scripts/cli.ts ssh hostname`，输出必须是目标宿主或 WSL 的 hostname，进程退出码必须为 0。任何 provider 在线但不声明 `host.ssh` 的状态都只能算未完成部署。 -如果该节点承载用户服务，还必须声明 `microservice.http` capability，并通过 `bun scripts/cli.ts microservice health ` 或 remote CLI 等价命令验证 backend-core 能经 provider-gateway 访问节点本机后端。用户服务后端端口不得映射到公网；provider-gateway 只允许代理节点本地 HTTP 地址或主 server 显式 Compose 服务名，业务 API 路径和 HTTP 方法还要受 backend-core `allowedPathPrefixes` 与 `allowedMethods` 限制。 +如果该节点承载用户服务，还必须声明 `microservice.http` capability，并通过 `bun scripts/cli.ts microservice health ` 或 remote CLI 等价命令验证 backend-core 能经 provider-gateway 访问节点本机后端。新版 provider-gateway 还应声明 `microservice.http.tunnel`，表示 backend-core 可以在同一 provider WebSocket 上发起短生命周期 HTTP tunnel request 并等待 provider 直接回传 upstream 响应；这一路径避免为每个 HTTP 代理请求创建、轮询并读取任务结果，降低高频只读接口出现 “upstream socket closed” 或 pending task 清理竞态导致 502 的概率。旧 gateway 未声明 tunnel 时，backend-core 可以继续使用原 `microservice.http` dispatch 兼容路径，但运行健康验收应以 tunnel capability 为目标状态。 + +用户服务后端端口不得映射到公网；provider-gateway 只允许代理节点本地 HTTP 地址或主 server 显式 Compose 服务名，业务 API 路径和 HTTP 方法还要受 backend-core `allowedPathPrefixes` 与 `allowedMethods` 限制。无论走 tunnel 还是旧 dispatch，响应必须保留 upstream status、content-type、截断标记和代理模式诊断头；provider WebSocket 断开时 backend-core 必须显式返回 504/502，不得静默 fallback 到公网直连、旧缓存或 provider-gateway 以外的业务 HTTP 入口。 自动化验证必须使用 Playwright 访问公网 frontend，而不是在容器内直接调 core API 代替浏览器验收。标准命令是 `bun scripts/cli.ts e2e run`；该命令会让 Playwright 打开公网 `http://74.48.78.17:18081/`、登录、抓取页面中的 Provider 信息和 `查看原始JSON` 内容，并检查 Provider 自接入、资源指标、Docker 状态和 `provider.upgrade` 预检。外部新增节点的人工验收应复用同一套前端路径：先确认 Provider 信息出现在节点清单，再确认资源监控和 Docker 状态页面有该节点的数据，最后通过任务调度向该 Provider 下发 `echo`、`docker.ps` 或维护专用 `host.ssh` probe，并在任务历史中查看耗时、状态、stdout/stderr 摘要和失败原因。 diff --git a/scripts/src/deploy.ts b/scripts/src/deploy.ts index a539e2e9..d9687987 100644 --- a/scripts/src/deploy.ts +++ b/scripts/src/deploy.ts @@ -881,7 +881,7 @@ function importK3sImageScript(service: UniDeskMicroserviceConfig): string { } function k8sDeploymentsForService(service: UniDeskMicroserviceConfig): string[] { - if (service.id === "code-queue") return ["d601-provider-egress-proxy", "d601-tcp-egress-gateway", "code-queue"]; + if (service.id === "code-queue") return ["d601-provider-egress-proxy", "d601-tcp-egress-gateway", "code-queue", "code-queue-read", "code-queue-write"]; return [service.repository.composeService]; } diff --git a/src/components/backend-core/src/microservice-proxy.ts b/src/components/backend-core/src/microservice-proxy.ts index c607fe03..ee4951e4 100644 --- a/src/components/backend-core/src/microservice-proxy.ts +++ b/src/components/backend-core/src/microservice-proxy.ts @@ -377,6 +377,19 @@ function isK3sctlManagedMicroservice(service: MicroserviceConfig): boolean { return service.deployment.mode === "k3sctl-managed" || service.backend.proxyMode === "k3sctl-adapter-http"; } +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/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"; + if (/^\/api\/tasks\/[^/]+$/u.test(targetPath) && normalizedMethod === "DELETE") return "code-queue-scheduler"; + if (targetPath === "/api/dev-containers" || /^\/api\/dev-containers(?:\/[^/]+)?\/start$/u.test(targetPath)) return "code-queue-scheduler"; + if (/^\/api\/dev-containers(?:\/[^/]+)?\/status$/u.test(targetPath)) return "code-queue-scheduler"; + if (normalizedMethod === "GET" || normalizedMethod === "HEAD") return "code-queue-read"; + return "code-queue-write"; +} + // --------------------------------------------------------------------------- // Cache helpers // --------------------------------------------------------------------------- @@ -606,7 +619,9 @@ async function k3sctlAdapterMicroserviceResponse( if (adapter.id === service.id || isK3sctlManagedMicroservice(adapter)) { return jsonResponse({ ok: false, error: "k3sctl adapter must be a UniDesk-direct microservice", serviceId: service.id, adapterServiceId }, 500); } - const k3sServiceId = service.deployment.k3sServiceId ?? service.id; + const k3sServiceId = service.id === "code-queue" + ? codeQueueK3sServiceIdForRequest(method, targetPath) + : service.deployment.k3sServiceId ?? service.id; const adapterTargetPath = `/api/services/${encodeURIComponent(k3sServiceId)}/proxy${targetPath}`; return fetchMicroserviceUpstreamResponse(adapter, method, adapterTargetPath, proxyOptions, requestHeaders, bodyText, abortSignal); } diff --git a/src/components/microservices/code-queue/src/index.ts b/src/components/microservices/code-queue/src/index.ts index d7047d18..9e1cdcb3 100644 --- a/src/components/microservices/code-queue/src/index.ts +++ b/src/components/microservices/code-queue/src/index.ts @@ -23,6 +23,7 @@ import type { QueueTaskRequest, RunMode, RuntimeConfig, + CodeQueueServiceRole, TaskStatus, WorkdirRecord, } from "./types"; @@ -266,6 +267,20 @@ function envBool(name: string, fallback: boolean): boolean { return fallback; } +function serviceRoleValue(raw: string): CodeQueueServiceRole { + const normalized = raw.trim().toLowerCase(); + if (normalized === "read" || normalized === "write" || normalized === "scheduler" || normalized === "combined") return normalized; + return "combined"; +} + +function serviceRoleAllowsWrite(role: CodeQueueServiceRole): boolean { + return role === "combined" || role === "write"; +} + +function serviceRoleAllowsScheduler(role: CodeQueueServiceRole): boolean { + return role === "combined" || role === "scheduler"; +} + function envList(name: string, fallback: string[]): string[] { const raw = process.env[name]; const source = raw === undefined || raw.length === 0 ? fallback.join(",") : raw; @@ -317,6 +332,7 @@ function readConfig(): RuntimeConfig { const defaultWorkdir = envString("CODE_QUEUE_WORKDIR", "/workspace"); const devContainerMasterHost = envString("CODE_QUEUE_DEV_CONTAINER_MASTER_HOST", "74.48.78.17"); const defaultProviderGatewayProxyHost = `unidesk-provider-gateway-${mainProviderId}`; + const serviceRole = serviceRoleValue(envString("CODE_QUEUE_SERVICE_ROLE", "combined")); const executionProviderIds = Array.from(new Set([ mainProviderId, ...envList("CODE_QUEUE_EXECUTION_PROVIDER_IDS", [devContainerDefaultProviderId]), @@ -328,7 +344,9 @@ function readConfig(): RuntimeConfig { instanceId: envString("CODE_QUEUE_INSTANCE_ID", mainProviderId), deployCommit: envString("CODE_QUEUE_DEPLOY_COMMIT", "unknown"), deployRequestedCommit: envString("CODE_QUEUE_DEPLOY_REQUESTED_COMMIT", ""), - schedulerEnabled: envBool("CODE_QUEUE_SCHEDULER_ENABLED", true), + serviceRole, + schedulerEnabled: envBool("CODE_QUEUE_SCHEDULER_ENABLED", serviceRoleAllowsScheduler(serviceRole)), + schedulerPollIntervalMs: Math.max(500, Math.min(30_000, envNumber("CODE_QUEUE_SCHEDULER_POLL_INTERVAL_MS", 2000))), 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"), @@ -1298,6 +1316,13 @@ interface DatabaseTaskRow { task_json: unknown; } +interface DatabaseQueueRow { + id: string; + name: string; + created_at: Date | string; + updated_at: Date | string; +} + interface DatabaseTaskIdRow { id: string; } @@ -1383,6 +1408,31 @@ async function loadTasksFromDatabase(where: "all" | "hot" = "all"): Promise(); + for (const row of rows) { + const id = safeQueueId(row.id); + queueMap.set(id, { + id, + name: safeQueueName(row.name, id), + createdAt: taskTimestamp(String(row.created_at)) ?? nowIso(), + updatedAt: taskTimestamp(String(row.updated_at)) ?? nowIso(), + }); + } + if (!queueMap.has(defaultQueueId)) queueMap.set(defaultQueueId, { id: defaultQueueId, name: defaultQueueId, createdAt: state.updatedAt, updatedAt: state.updatedAt }); + return Array.from(queueMap.values()).sort((left, right) => left.id.localeCompare(right.id)); +} + +async function loadQueuesFromDatabase(): Promise { + if (!databaseReady) return state.queues; + const queueRows = await sql` + SELECT id, name, created_at, updated_at + FROM unidesk_code_queue_queues + ORDER BY id ASC + `; + return databaseQueueRowsToRecords(queueRows); +} + async function loadTasksFromDatabaseByIds(taskIds: string[]): Promise { const ids = Array.from(new Set(taskIds.map((id) => id.trim()).filter(Boolean))); if (ids.length === 0) return []; @@ -1730,22 +1780,13 @@ async function initDatabasePersistence(): Promise { inMemoryOutputRecords: config.maxInMemoryOutputRecords, inMemoryEventRecords: config.maxInMemoryEventRecords, }); - const queueRows = await sql>` + const queueRows = await sql` SELECT id, name, created_at, updated_at FROM unidesk_code_queue_queues ORDER BY id ASC `; runGarbageCollection(); - const queueMap = new Map(); - for (const row of queueRows) { - const id = safeQueueId(row.id); - queueMap.set(id, { - id, - name: safeQueueName(row.name, id), - createdAt: taskTimestamp(String(row.created_at)) ?? nowIso(), - updatedAt: taskTimestamp(String(row.updated_at)) ?? nowIso(), - }); - } + const queueMap = new Map(databaseQueueRowsToRecords(queueRows).map((queue) => [queue.id, queue])); if (!queueMap.has(defaultQueueId)) queueMap.set(defaultQueueId, { id: defaultQueueId, name: defaultQueueId, createdAt: state.updatedAt, updatedAt: state.updatedAt }); for (const task of state.tasks) { const id = queueIdOf(task); @@ -1773,7 +1814,7 @@ async function initDatabasePersistence(): Promise { ensureDefaultWorkdirRecords(); await upsertWorkdirsToDatabase(sortedWorkdirRecords()); databaseReady = true; - scheduleStartupDatabaseMaintenance(); + if (config.serviceRole === "combined" || config.serviceRole === "scheduler") scheduleStartupDatabaseMaintenance(); runGarbageCollection(); logger("info", "database_persistence_init_complete", { databaseTaskCount: Number(countRows[0]?.count ?? hotTasks.length), @@ -2320,6 +2361,7 @@ configureQueueApi({ dirtyDatabaseTaskCount: () => dirtyDatabaseTaskIds.size, jsonResponse, judgeFailRetryLimit, + loadQueuesFromDatabase, loadTaskFromDatabase, loadTasksFromDatabase, loadTasksFromDatabaseByIds, @@ -3223,7 +3265,7 @@ 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) { + if (!config.schedulerEnabled && config.serviceRole !== "read") { return queuedReason("scheduler_disabled", "STANDBY", "This Code Queue instance is a k3s-managed standby and does not start queued work."); } if (mergingQueues.has(queueId)) { @@ -3335,6 +3377,57 @@ function hasRunnableTask(): boolean { return state.tasks.some((task) => task.status === "queued" || task.status === "retry_wait"); } +function shouldPollSchedulerDatabase(): boolean { + return config.schedulerEnabled && config.serviceRole === "scheduler"; +} + +function mergeSchedulerDatabaseTasks(tasks: QueueTask[]): number { + let changed = 0; + for (const task of tasks) { + if (task.status !== "queued" && task.status !== "retry_wait" && task.status !== "running" && task.status !== "judging") continue; + const existing = findTask(task.id); + if (existing === null) { + state.tasks.push(task); + changed += 1; + continue; + } + const existingUpdatedAt = timestampMs(existing.updatedAt) ?? 0; + const taskUpdatedAt = timestampMs(task.updatedAt) ?? 0; + if (taskUpdatedAt > existingUpdatedAt && !activeRunForTask(existing)) { + Object.assign(existing, task); + changed += 1; + } + } + if (changed > 0) { + state.tasks.sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id)); + updateNextSeqFromTasks(); + } + return changed; +} + +async function refreshSchedulerTasksFromDatabase(reason: string): Promise { + if (!databaseReady || !config.schedulerEnabled) return 0; + const tasks = await loadTasksFromDatabase("hot"); + const changed = mergeSchedulerDatabaseTasks(tasks); + if (changed > 0) { + logger("info", "scheduler_database_hot_tasks_refreshed", { reason, changed, loaded: tasks.length }); + scheduleQueue(); + } + return changed; +} + +function startSchedulerDatabasePoller(): void { + if (!shouldPollSchedulerDatabase()) return; + const interval = setInterval(() => { + if (!serviceReady || shutdownRequested) return; + void refreshSchedulerTasksFromDatabase("poll").catch((error) => { + databaseLastError = databaseErrorMessage(error); + logger("warn", "scheduler_database_poll_failed", { error: errorToJson(error) }); + }); + }, config.schedulerPollIntervalMs); + interval.unref?.(); +} + function activeRunForTask(task: QueueTask): ActiveRun | null { const queueRun = activeRuns.get(queueIdOf(task)); if (queueRun?.taskId === task.id) return queueRun; @@ -3440,6 +3533,27 @@ function requestErrorResponse(error: unknown): Response | null { return null; } +function readOnlyRejectResponse(method: string, targetPath: string): Response { + return jsonResponse({ + ok: false, + error: "Code Queue read service is read-only", + serviceRole: config.serviceRole, + method, + path: targetPath, + }, 405); +} + +function schedulerOnlyRejectResponse(method: string, targetPath: string): Response { + return jsonResponse({ + ok: false, + error: "Code Queue write service does not own active scheduler control for this endpoint", + serviceRole: config.serviceRole, + method, + path: targetPath, + hint: "Route active-run control to the code-queue-scheduler service.", + }, 409); +} + function findTask(id: string): QueueTask | null { return state.tasks.find((task) => task.id === id) ?? null; } @@ -3520,6 +3634,7 @@ async function deleteWorkdir(providerIdValue: string, executionModeValue: string } async function createTasks(req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, "/api/tasks"); const body = await readJson(req); const batchRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const batchQueueId = typeof batchRecord.queueId === "string" && batchRecord.queueId.trim().length > 0 ? normalizeQueueId(batchRecord.queueId) : undefined; @@ -3545,6 +3660,7 @@ async function createTasks(req: Request): Promise { } async function steerTask(task: QueueTask, req: Request): Promise { + if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(req.method, `/api/tasks/${task.id}/steer`); const body = await readJson(req); const prompt = typeof (body as Record).prompt === "string" ? String((body as Record).prompt) : ""; if (prompt.trim().length === 0) return jsonResponse({ ok: false, error: "prompt is required" }, 400); @@ -3560,6 +3676,7 @@ async function steerTask(task: QueueTask, req: Request): Promise { } async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/edit`); if (!queuedTaskPromptEditable(task)) { return jsonResponse({ ok: false, @@ -3601,7 +3718,8 @@ async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise { +async function interruptTask(task: QueueTask, method = "POST"): Promise { + if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(method, `/api/tasks/${task.id}/interrupt`); if (task.status === "succeeded" || task.status === "failed" || task.status === "canceled") { return jsonResponse({ ok: false, error: `task is already terminal: ${task.status}`, task: taskForResponse(task) }, 409); } @@ -3631,6 +3749,7 @@ async function interruptTask(task: QueueTask): Promise { } async function manualRetry(task: QueueTask, req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/retry`); if (task.status !== "failed" && task.status !== "canceled" && task.status !== "succeeded") { return jsonResponse({ ok: false, error: `task is not terminal: ${task.status}`, task: taskForResponse(task) }, 409); } @@ -3657,6 +3776,7 @@ async function manualRetry(task: QueueTask, req: Request): Promise { } async function markTaskRead(task: QueueTask): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", `/api/tasks/${task.id}/read`); if (!terminalTask(task)) { return jsonResponse({ ok: false, error: `task is not terminal: ${task.status}`, task: taskForResponse(task) }, 409); } @@ -3678,6 +3798,7 @@ function timestampToIso(value: Date | string | null): string | null { } async function markTaskReadById(taskId: string): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", `/api/tasks/${taskId}/read`); if (!databaseReady) { const task = await findTaskForMutation(taskId); return task === null ? jsonResponse({ ok: false, error: "task not found" }, 404) : markTaskRead(task); @@ -3727,6 +3848,7 @@ async function markTaskReadById(taskId: string): Promise { } async function markTerminalTasksRead(url: URL): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", "/api/tasks/read-all"); const queueFilter = url.searchParams.get("queueId"); const queueId = queueFilter === null || queueFilter.length === 0 ? null : safeQueueId(queueFilter); const readAt = nowIso(); @@ -3780,6 +3902,7 @@ async function markTerminalTasksRead(url: URL): Promise { } async function createQueue(req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, "/api/queues"); const body = await readJson(req); const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const queueId = normalizeQueueId(record.queueId ?? record.id); @@ -3796,6 +3919,7 @@ async function createQueue(req: Request): Promise { } async function updateQueue(queueIdValue: string, req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/queues/${queueIdValue}`); const queueId = normalizeQueueId(queueIdValue); const queue = state.queues.find((item) => item.id === queueId); if (queue === undefined) return jsonResponse({ ok: false, error: "queue not found" }, 404); @@ -3911,6 +4035,7 @@ function deleteQueuesFromState(queueIds: string[]): QueueRecord[] { } async function mergeQueues(targetQueueIdValue: string | null, req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, targetQueueIdValue === null ? "/api/queues/merge" : `/api/queues/${targetQueueIdValue}/merge`); const body = await readJson(req); const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record : {}; const targetQueueId = normalizeQueueId(targetQueueIdValue ?? record.targetQueueId ?? record.intoQueueId ?? record.into); @@ -3994,6 +4119,7 @@ async function mergeQueues(targetQueueIdValue: string | null, req: Request): Pro } async function moveTaskToQueue(task: QueueTask, req: Request): Promise { + if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/move`); if (task.status === "running" || task.status === "judging") { return jsonResponse({ ok: false, error: `cannot move active task ${task.id} while status=${task.status}`, task: taskForResponse(task) }, 409); } @@ -4055,6 +4181,7 @@ async function route(req: Request): Promise { ok: true, service: "code-queue", instanceId: config.instanceId, + role: config.serviceRole, deploy: { commit: config.deployCommit, requestedCommit: config.deployRequestedCommit, @@ -4068,6 +4195,7 @@ async function route(req: Request): Promise { ok: false, service: "code-queue", instanceId: config.instanceId, + role: config.serviceRole, deploy: { commit: config.deployCommit, requestedCommit: config.deployRequestedCommit, @@ -4081,11 +4209,13 @@ async function route(req: Request): Promise { ok: true, service: "code-queue", instanceId: config.instanceId, + role: config.serviceRole, deploy: { commit: config.deployCommit, requestedCommit: config.deployRequestedCommit, }, schedulerEnabled: config.schedulerEnabled, + schedulerPollIntervalMs: config.schedulerPollIntervalMs, queue: queueSummary(false, state.tasks), egressProxy: await providerGatewayEgressProxyStatus(), oaEventPublisher: oaEventPublisherStatus(), @@ -4145,7 +4275,8 @@ async function route(req: Request): Promise { } if (url.pathname === "/api/queues" && req.method === "GET") { const tasks = await loadAllTasksForRead(); - return jsonResponse({ ok: true, queues: perQueueSummaries(tasks), queue: queueSummary(false, tasks) }); + const queueRecords = await loadQueuesFromDatabase(); + return jsonResponse({ ok: true, queues: perQueueSummaries(tasks, queueRecords), queue: await queueSummaryForResponse(false, tasks, queueRecords) }); } if (url.pathname === "/api/workdirs" && req.method === "GET") return await listWorkdirs(url); if (url.pathname === "/api/workdirs" && req.method === "POST") return await createWorkdir(req); @@ -4269,7 +4400,7 @@ async function route(req: Request): Promise { if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404); if (action === "retry" && req.method === "POST") return await manualRetry(task, req); if (action === "steer" && req.method === "POST") return await steerTask(task, req); - if (action === "interrupt" && req.method === "POST") return await interruptTask(task); + if (action === "interrupt" && req.method === "POST") return await interruptTask(task, req.method); if (action === "move" && req.method === "POST") return await moveTaskToQueue(task, req); if (action === "edit" && (req.method === "POST" || req.method === "PATCH")) return await editQueuedTaskPrompt(task, req); if (action !== undefined) return jsonResponse({ ok: false, error: "not found" }, 404); @@ -4279,7 +4410,7 @@ async function route(req: Request): Promise { const includeRaw = url.searchParams.get("raw") === "1" || url.searchParams.get("full") === "1"; return jsonResponse({ ok: true, task: applyOaTraceStatsToTaskJson(taskForResponse(task, true, includeRaw), traceStats) }); } - if (req.method === "DELETE") return await interruptTask(task); + if (req.method === "DELETE") return await interruptTask(task, req.method); return jsonResponse({ ok: false, error: "method not allowed" }, 405); } return jsonResponse({ ok: false, error: "not found", path: url.pathname }, 404); @@ -4301,7 +4432,7 @@ startCodexSqliteLogExporter(); startMemoryWatchdog(); await initDatabasePersistenceWithRetry(); Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route }); -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" }); +logger("info", "service_started", { port: config.port, instanceId: config.instanceId, role: config.serviceRole, schedulerEnabled: config.schedulerEnabled, schedulerPollIntervalMs: config.schedulerPollIntervalMs, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres" }); { const devReady = collectDevReady() as Record; logger(devReady.ok === true ? "info" : "warn", "dev_ready_check", devReady); @@ -4310,6 +4441,8 @@ const startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRet if (startupRecovered > 0) logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered }); persistState(); serviceReady = true; +await refreshSchedulerTasksFromDatabase("startup"); +startSchedulerDatabasePoller(); 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?.(); } diff --git a/src/components/microservices/code-queue/src/queue-api.ts b/src/components/microservices/code-queue/src/queue-api.ts index 38f3afd5..82a99661 100644 --- a/src/components/microservices/code-queue/src/queue-api.ts +++ b/src/components/microservices/code-queue/src/queue-api.ts @@ -25,6 +25,7 @@ export interface QueueApiContext { dirtyDatabaseTaskCount: () => number; jsonResponse: (body: unknown, status?: number) => Response; judgeFailRetryLimit: number; + loadQueuesFromDatabase: () => Promise; loadTaskFromDatabase: (taskId: string) => Promise; loadTasksFromDatabase: (where?: "all" | "hot") => Promise; loadTasksFromDatabaseByIds: (taskIds: string[]) => Promise; @@ -275,7 +276,7 @@ function taskForListResponse(task: QueueTask, lite = false, queueTasks?: QueueTa } as unknown as JsonValue; } -function perQueueSummaries(tasks: QueueTask[] = ctx().tasks()): JsonValue[] { +function perQueueSummaries(tasks: QueueTask[] = ctx().tasks(), queueRecords: QueueRecord[] = ctx().queues()): JsonValue[] { const summaries = new Map(); - for (const queue of ctx().queues()) { - queue.name = ctx().safeQueueName(queue.name, queue.id); + for (const queue of queueRecords) { + const name = ctx().safeQueueName(queue.name, queue.id); summaries.set(queue.id, { - name: queue.name, + name, total: 0, counts: {}, unreadTerminal: 0, @@ -342,7 +343,7 @@ function perQueueSummaries(tasks: QueueTask[] = ctx().tasks()): JsonValue[] { return rows; } -function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks()): JsonValue { +function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks(), queueRecords: QueueRecord[] = ctx().queues()): JsonValue { const counts = tasks.reduce>((memo, task) => { memo[task.status] = (memo[task.status] ?? 0) + 1; return memo; @@ -355,7 +356,7 @@ function queueSummary(includeDevReady = true, tasks: QueueTask[] = ctx().tasks() } const activeTaskIds = Array.from(activeTaskIdSet).sort(); const activeTaskId = activeTaskIds[0] ?? tasks.find((task) => task.status === "running" || task.status === "judging")?.id ?? null; - const queues = perQueueSummaries(tasks); + const queues = perQueueSummaries(tasks, queueRecords); const summary: Record = { total: tasks.length, defaultQueueId: ctx().defaultQueueId, @@ -449,13 +450,14 @@ async function loadAllTasksForRead(): Promise { return Array.from(byId.values()).sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id)); } -async function queueSummaryForResponse(includeDevReady = true, tasks?: QueueTask[]): Promise { - if (tasks !== undefined) return queueSummary(includeDevReady, tasks); +async function queueSummaryForResponse(includeDevReady = true, tasks?: QueueTask[], queueRecords?: QueueRecord[]): Promise { + if (tasks !== undefined || queueRecords !== undefined) return queueSummary(includeDevReady, tasks ?? ctx().tasks(), queueRecords ?? ctx().queues()); return queueSummaryForHealth(includeDevReady); } async function queueSummaryForHealth(includeDevReady = true): Promise { - const summary = queueSummary(includeDevReady, ctx().tasks()) as Record; + const queueRecords = ctx().databaseReady() ? await ctx().loadQueuesFromDatabase() : ctx().queues(); + const summary = queueSummary(includeDevReady, ctx().tasks(), queueRecords) as Record; if (!ctx().databaseReady()) return summary; const aggregateRows = await ctx().sql` SELECT @@ -501,7 +503,7 @@ async function queueSummaryForHealth(includeDevReady = true): Promise createdAt: string | null; updatedAt: string | null; }>(); - for (const queue of ctx().queues()) { + for (const queue of queueRecords) { summaries.set(queue.id, { name: ctx().safeQueueName(queue.name, queue.id), total: 0, diff --git a/src/components/microservices/code-queue/src/types.ts b/src/components/microservices/code-queue/src/types.ts index 6a1b6410..d2ed5653 100644 --- a/src/components/microservices/code-queue/src/types.ts +++ b/src/components/microservices/code-queue/src/types.ts @@ -18,6 +18,8 @@ export interface QueuedStatusReason { export type RunMode = "initial" | "retry"; +export type CodeQueueServiceRole = "combined" | "read" | "write" | "scheduler"; + export type JudgeDecision = "complete" | "retry" | "fail"; export type CodeExecutionMode = "default" | "windows-native"; @@ -35,7 +37,9 @@ export interface RuntimeConfig { instanceId: string; deployCommit: string; deployRequestedCommit: string; + serviceRole: CodeQueueServiceRole; schedulerEnabled: boolean; + schedulerPollIntervalMs: number; startupOaBackfillEnabled: boolean; outputArchiveDir: string; logFile: string; diff --git a/src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json b/src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json index 43717629..cefacbce 100644 --- a/src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json +++ b/src/components/microservices/k3sctl-adapter/k3s/code-queue.k3s.json @@ -1,46 +1,159 @@ -{ - "apiVersion": "unidesk.ai/k3s/v1", - "kind": "ManagedKubernetesService", - "metadata": { - "name": "code-queue", - "namespace": "unidesk" - }, - "spec": { - "adapterServiceId": "k3sctl-adapter", - "controlPlane": { - "type": "kubernetes", - "cluster": "unidesk-k3s", - "context": "unidesk-k3s" +[ + { + "apiVersion": "unidesk.ai/k3s/v1", + "kind": "ManagedKubernetesService", + "metadata": { + "name": "code-queue", + "namespace": "unidesk" }, - "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", - "healthMode": "service-proxy" + "spec": { + "adapterServiceId": "k3sctl-adapter", + "controlPlane": { + "type": "kubernetes", + "cluster": "unidesk-k3s", + "context": "unidesk-k3s" }, - { - "id": "D518", - "nodeId": "D518", - "role": "standby", - "baseUrl": "kubernetes://unidesk/services/code-queue-d518:4222", - "healthPath": "/health", - "healthMode": "pod-ready" - } - ], - "requireAllInstancesHealthy": false + "route": { + "kind": "kubernetes-service", + "serviceName": "code-queue-scheduler", + "servicePort": 4222 + }, + "activeInstanceId": "D601", + "singleWriter": true, + "expectedNodeIds": [ + "D601", + "D518" + ], + "instances": [ + { + "id": "D601", + "nodeId": "D601", + "role": "primary", + "baseUrl": "kubernetes://unidesk/services/code-queue-scheduler:4222", + "healthPath": "/health", + "healthMode": "service-proxy" + }, + { + "id": "D518", + "nodeId": "D518", + "role": "standby", + "baseUrl": "kubernetes://unidesk/services/code-queue-d518:4222", + "healthPath": "/health", + "healthMode": "pod-ready" + } + ], + "requireAllInstancesHealthy": false + } + }, + { + "apiVersion": "unidesk.ai/k3s/v1", + "kind": "ManagedKubernetesService", + "metadata": { + "name": "code-queue-read", + "namespace": "unidesk" + }, + "spec": { + "adapterServiceId": "k3sctl-adapter", + "controlPlane": { + "type": "kubernetes", + "cluster": "unidesk-k3s", + "context": "unidesk-k3s" + }, + "route": { + "kind": "kubernetes-service", + "serviceName": "code-queue-read", + "servicePort": 4222 + }, + "activeInstanceId": "D601-read", + "singleWriter": false, + "expectedNodeIds": [ + "D601" + ], + "instances": [ + { + "id": "D601-read", + "nodeId": "D601", + "role": "standby", + "baseUrl": "kubernetes://unidesk/services/code-queue-read:4222", + "healthPath": "/live", + "healthMode": "service-proxy" + } + ], + "requireAllInstancesHealthy": false + } + }, + { + "apiVersion": "unidesk.ai/k3s/v1", + "kind": "ManagedKubernetesService", + "metadata": { + "name": "code-queue-write", + "namespace": "unidesk" + }, + "spec": { + "adapterServiceId": "k3sctl-adapter", + "controlPlane": { + "type": "kubernetes", + "cluster": "unidesk-k3s", + "context": "unidesk-k3s" + }, + "route": { + "kind": "kubernetes-service", + "serviceName": "code-queue-write", + "servicePort": 4222 + }, + "activeInstanceId": "D601-write", + "singleWriter": true, + "expectedNodeIds": [ + "D601" + ], + "instances": [ + { + "id": "D601-write", + "nodeId": "D601", + "role": "primary", + "baseUrl": "kubernetes://unidesk/services/code-queue-write:4222", + "healthPath": "/health", + "healthMode": "service-proxy" + } + ], + "requireAllInstancesHealthy": false + } + }, + { + "apiVersion": "unidesk.ai/k3s/v1", + "kind": "ManagedKubernetesService", + "metadata": { + "name": "code-queue-scheduler", + "namespace": "unidesk" + }, + "spec": { + "adapterServiceId": "k3sctl-adapter", + "controlPlane": { + "type": "kubernetes", + "cluster": "unidesk-k3s", + "context": "unidesk-k3s" + }, + "route": { + "kind": "kubernetes-service", + "serviceName": "code-queue-scheduler", + "servicePort": 4222 + }, + "activeInstanceId": "D601-scheduler", + "singleWriter": true, + "expectedNodeIds": [ + "D601" + ], + "instances": [ + { + "id": "D601-scheduler", + "nodeId": "D601", + "role": "primary", + "baseUrl": "kubernetes://unidesk/services/code-queue-scheduler:4222", + "healthPath": "/health", + "healthMode": "service-proxy" + } + ], + "requireAllInstancesHealthy": false + } } -} +] diff --git a/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml b/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml index d12bd9cf..8e8e538a 100644 --- a/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml +++ b/src/components/microservices/k3sctl-adapter/k3s/code-queue.k8s.yaml @@ -6,6 +6,544 @@ metadata: app.kubernetes.io/part-of: unidesk unidesk.ai/k3s-cluster: unidesk-k3s --- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: code-queue-read + namespace: unidesk + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: read + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed + unidesk.ai/instance-id: D601-read +spec: + replicas: 2 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: read + unidesk.ai/instance-id: D601-read + template: + metadata: + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: read + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed + unidesk.ai/instance-id: D601-read + unidesk.ai/node-id: D601 + spec: + nodeSelector: + unidesk.ai/node-id: D601 + 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: DATABASE_URL + value: "postgres://unidesk:unidesk_dev_password@d601-tcp-egress-gateway.unidesk.svc.cluster.local:15432/unidesk" + - name: CODE_QUEUE_INSTANCE_ID + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: CODE_QUEUE_SERVICE_ROLE + value: "read" + - name: CODE_QUEUE_SCHEDULER_ENABLED + value: "false" + - 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=512" + - name: GIT_CONFIG_COUNT + value: "1" + - name: GIT_CONFIG_KEY_0 + value: "safe.directory" + - name: GIT_CONFIG_VALUE_0 + value: "*" + - 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: "false" + - name: CODE_QUEUE_EGRESS_PROXY_URL + value: "" + - name: CODE_QUEUE_EGRESS_PROXY_NO_PROXY + value: "localhost,127.0.0.1,::1,host.docker.internal,d601-tcp-egress-gateway,d601-tcp-egress-gateway.unidesk,d601-tcp-egress-gateway.unidesk.svc,d601-tcp-egress-gateway.unidesk.svc.cluster.local,backend-core,oa-event-flow,database" + - name: HTTP_PROXY + value: "" + - name: HTTPS_PROXY + value: "" + - name: ALL_PROXY + value: "" + - name: http_proxy + value: "" + - name: https_proxy + value: "" + - name: all_proxy + value: "" + - name: NO_PROXY + value: "localhost,127.0.0.1,::1,host.docker.internal,d601-tcp-egress-gateway,d601-tcp-egress-gateway.unidesk,d601-tcp-egress-gateway.unidesk.svc,d601-tcp-egress-gateway.unidesk.svc.cluster.local,backend-core,oa-event-flow,database" + - name: no_proxy + value: "localhost,127.0.0.1,::1,host.docker.internal,d601-tcp-egress-gateway,d601-tcp-egress-gateway.unidesk,d601-tcp-egress-gateway.unidesk.svc,d601-tcp-egress-gateway.unidesk.svc.cluster.local,backend-core,oa-event-flow,database" + - name: OA_EVENT_FLOW_BASE_URL + value: "http://d601-tcp-egress-gateway.unidesk.svc.cluster.local:4255" + - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED + value: "false" + - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL + value: "" + - 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-read.jsonl" + - name: UNIDESK_LOG_RETENTION_BYTES + value: "1GiB" + volumeMounts: + - name: docker-sock + mountPath: /var/run/docker.sock + - name: workspace + mountPath: /workspace + - name: workspace + mountPath: /home/ubuntu + - name: repo + mountPath: /root/unidesk + - name: repo + mountPath: /app + - name: codex-config + mountPath: /root/.codex/config.toml + readOnly: true + - name: codex-auth + mountPath: /root/.codex/auth.json + 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: /live + 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: 100m + memory: 256Mi + limits: + memory: 1Gi + volumes: + - name: docker-sock + hostPath: + path: /var/run/docker.sock + type: Socket + - name: workspace + hostPath: + path: /home/ubuntu + type: Directory + - name: repo + hostPath: + path: /home/ubuntu/cq-deploy + type: Directory + - name: codex-config + hostPath: + path: /home/ubuntu/.codex/config.toml + type: File + - name: codex-auth + hostPath: + path: /home/ubuntu/.codex/auth.json + 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: apps/v1 +kind: Deployment +metadata: + name: code-queue-write + namespace: unidesk + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: write + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed + unidesk.ai/instance-id: D601-write +spec: + replicas: 1 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 + selector: + matchLabels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: write + unidesk.ai/instance-id: D601-write + template: + metadata: + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: write + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed + unidesk.ai/instance-id: D601-write + unidesk.ai/node-id: D601 + spec: + nodeSelector: + unidesk.ai/node-id: D601 + 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: DATABASE_URL + value: "postgres://unidesk:unidesk_dev_password@d601-tcp-egress-gateway.unidesk.svc.cluster.local:15432/unidesk" + - name: CODE_QUEUE_INSTANCE_ID + value: "D601-write" + - name: CODE_QUEUE_SERVICE_ROLE + value: "write" + - name: CODE_QUEUE_SCHEDULER_ENABLED + value: "false" + - 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=768" + - name: GIT_CONFIG_COUNT + value: "1" + - name: GIT_CONFIG_KEY_0 + value: "safe.directory" + - name: GIT_CONFIG_VALUE_0 + value: "*" + - 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: "false" + - name: CODE_QUEUE_EGRESS_PROXY_URL + value: "" + - name: CODE_QUEUE_EGRESS_PROXY_NO_PROXY + value: "localhost,127.0.0.1,::1,host.docker.internal,d601-tcp-egress-gateway,d601-tcp-egress-gateway.unidesk,d601-tcp-egress-gateway.unidesk.svc,d601-tcp-egress-gateway.unidesk.svc.cluster.local,backend-core,oa-event-flow,database" + - name: HTTP_PROXY + value: "" + - name: HTTPS_PROXY + value: "" + - name: ALL_PROXY + value: "" + - name: http_proxy + value: "" + - name: https_proxy + value: "" + - name: all_proxy + value: "" + - name: NO_PROXY + value: "localhost,127.0.0.1,::1,host.docker.internal,d601-tcp-egress-gateway,d601-tcp-egress-gateway.unidesk,d601-tcp-egress-gateway.unidesk.svc,d601-tcp-egress-gateway.unidesk.svc.cluster.local,backend-core,oa-event-flow,database" + - name: no_proxy + value: "localhost,127.0.0.1,::1,host.docker.internal,d601-tcp-egress-gateway,d601-tcp-egress-gateway.unidesk,d601-tcp-egress-gateway.unidesk.svc,d601-tcp-egress-gateway.unidesk.svc.cluster.local,backend-core,oa-event-flow,database" + - name: OA_EVENT_FLOW_BASE_URL + value: "http://d601-tcp-egress-gateway.unidesk.svc.cluster.local:4255" + - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED + value: "false" + - name: CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL + value: "" + - 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-write.jsonl" + - name: UNIDESK_LOG_RETENTION_BYTES + value: "1GiB" + volumeMounts: + - name: docker-sock + mountPath: /var/run/docker.sock + - name: workspace + mountPath: /workspace + - name: workspace + mountPath: /home/ubuntu + - name: repo + mountPath: /root/unidesk + - name: repo + mountPath: /app + - name: codex-config + mountPath: /root/.codex/config.toml + readOnly: true + - name: codex-auth + mountPath: /root/.codex/auth.json + 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: /live + 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: 100m + memory: 256Mi + limits: + memory: 1Gi + volumes: + - name: docker-sock + hostPath: + path: /var/run/docker.sock + type: Socket + - name: workspace + hostPath: + path: /home/ubuntu + type: Directory + - name: repo + hostPath: + path: /home/ubuntu/cq-deploy + type: Directory + - name: codex-config + hostPath: + path: /home/ubuntu/.codex/config.toml + type: File + - name: codex-auth + hostPath: + path: /home/ubuntu/.codex/auth.json + 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-scheduler + namespace: unidesk + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: scheduler + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: code-queue + unidesk.ai/instance-id: D601 + ports: + - name: http + port: 4222 + targetPort: http +--- +apiVersion: v1 +kind: Service +metadata: + name: code-queue-read + namespace: unidesk + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: read + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: read + unidesk.ai/instance-id: D601-read + ports: + - name: http + port: 4222 + targetPort: http +--- +apiVersion: v1 +kind: Service +metadata: + name: code-queue-write + namespace: unidesk + labels: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: write + app.kubernetes.io/part-of: unidesk + unidesk.ai/deployment-mode: k3sctl-managed +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: code-queue + app.kubernetes.io/component: write + unidesk.ai/instance-id: D601-write + ports: + - name: http + port: 4222 + targetPort: http +--- apiVersion: v1 kind: Service metadata: @@ -348,6 +886,11 @@ metadata: unidesk.ai/instance-id: D601 spec: replicas: 1 + strategy: + type: RollingUpdate + rollingUpdate: + maxUnavailable: 0 + maxSurge: 1 selector: matchLabels: app.kubernetes.io/name: code-queue @@ -384,8 +927,12 @@ spec: value: "postgres://unidesk:unidesk_dev_password@d601-tcp-egress-gateway.unidesk.svc.cluster.local:15432/unidesk" - name: CODE_QUEUE_INSTANCE_ID value: "D601" + - name: CODE_QUEUE_SERVICE_ROLE + value: "scheduler" - name: CODE_QUEUE_SCHEDULER_ENABLED value: "true" + - name: CODE_QUEUE_SCHEDULER_POLL_INTERVAL_MS + value: "2000" - name: CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED value: "false" - name: CODE_QUEUE_DATA_DIR @@ -631,6 +1178,8 @@ spec: value: "postgres://unidesk:unidesk_dev_password@d601-tcp-egress-gateway.unidesk.svc.cluster.local:15432/unidesk" - name: CODE_QUEUE_INSTANCE_ID value: "D518" + - name: CODE_QUEUE_SERVICE_ROLE + value: "read" - name: CODE_QUEUE_SCHEDULER_ENABLED value: "false" - name: CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED