From 9046812b5fb5e14a637d52e0c2c35eb1a38744d7 Mon Sep 17 00:00:00 2001
From: AlyciaBHZ <50111876+AlyciaBHZ@users.noreply.github.com>
Date: Wed, 17 Jun 2026 18:18:07 +0800
Subject: [PATCH 1/3] Add standalone compute pool service integration

---
 integrations/compute-pool-service/.gitignore  |   3 +
 integrations/compute-pool-service/README.md   | 173 ++++++
 .../compute-pool-service/openapi.yaml         | 201 +++++++
 .../compute-pool-service/package.json         |  14 +
 integrations/compute-pool-service/server.mjs  | 508 ++++++++++++++++++
 integrations/compute-pool-service/worker.mjs  | 169 ++++++
 6 files changed, 1068 insertions(+)
 create mode 100644 integrations/compute-pool-service/.gitignore
 create mode 100644 integrations/compute-pool-service/README.md
 create mode 100644 integrations/compute-pool-service/openapi.yaml
 create mode 100644 integrations/compute-pool-service/package.json
 create mode 100644 integrations/compute-pool-service/server.mjs
 create mode 100644 integrations/compute-pool-service/worker.mjs

diff --git a/integrations/compute-pool-service/.gitignore b/integrations/compute-pool-service/.gitignore
new file mode 100644
index 00000000..198deac1
--- /dev/null
+++ b/integrations/compute-pool-service/.gitignore
@@ -0,0 +1,3 @@
+.compute-pool-store.json
+.env
+node_modules/
diff --git a/integrations/compute-pool-service/README.md b/integrations/compute-pool-service/README.md
new file mode 100644
index 00000000..00097d10
--- /dev/null
+++ b/integrations/compute-pool-service/README.md
@@ -0,0 +1,173 @@
+# Compute Pool Service
+
+This integration is the long-term shape for shared GPU / Mac / lab compute:
+the queue and worker protocol live outside NyxID core. NyxID manages the
+service as a normal user/org service: auth, agent API keys, credential
+injection, node routing, audit metadata, and future load balancing stay in
+NyxID; compute-specific task state stays in this service.
+
+This integration does not require a NyxID org model change. To share it with
+company members, create the NyxID service under the existing org owner and
+use the current org membership/admin checks that already apply to services.
+
+## Architecture
+
+```text
+agent / org user
+  -> NyxID proxy / service governance
+  -> optional NyxID Credential Node
+  -> compute-pool-service
+  -> trusted GPU/Mac/Slurm workers
+  -> local OpenAI-compatible backend
+```
+
+NyxID core does not store compute tasks, worker tokens, local backend URLs,
+or local backend credentials.
+
+## Security Boundary
+
+This shares controlled task execution capacity, not host access.
+
+- NyxID does not SSH into worker hosts.
+- NyxID does not execute shell commands.
+- NyxID does not expose worker filesystems or environment variables.
+- NyxID does not store worker-local model endpoint URLs.
+- NyxID does not store worker-local backend bearer tokens.
+- If routed through a Credential Node, the service API token can stay on the
+  node host and be injected locally.
+
+The standalone service stores task input/output in its own local store. The
+default store is a JSON file intended for smoke tests and small trusted
+deployments, not production durability. A production version should replace
+the store with Postgres, Redis, MongoDB, or another managed queue backend.
+
+## Start The Service
+
+Generate two independent tokens:
+
+```bash
+export COMPUTE_POOL_API_TOKEN="$(openssl rand -hex 32)"
+export COMPUTE_POOL_WORKER_TOKEN="$(openssl rand -hex 32)"
+```
+
+Start the queue service on the private host:
+
+```bash
+cd integrations/compute-pool-service
+node server.mjs
+```
+
+For local throwaway testing only:
+
+```bash
+COMPUTE_POOL_DEV_INSECURE=1 node server.mjs
+```
+
+## Add To NyxID As A Service
+
+Recommended: run a NyxID Credential Node on the host that can reach this
+service, then register the service through that node.
+
+```bash
+nyxid service add --custom \
+  --slug chrono-compute \
+  --label "Chrono Compute Pool" \
+  --endpoint-url "http://127.0.0.1:8787" \
+  --auth-method bearer \
+  --auth-key-name "Authorization" \
+  --via-node <node-id>
+```
+
+Then store the service API token on the node:
+
+```bash
+nyxid node credentials add \
+  --service chrono-compute \
+  --url "http://127.0.0.1:8787" \
+  --header "Authorization" \
+  --secret-format bearer
+```
+
+Agents and org members call it through NyxID like any other service:
+
+```bash
+nyxid proxy request chrono-compute /v1/tasks \
+  -m POST \
+  -d '{"model":"codex-local","input":{"messages":[{"role":"user","content":"ping"}]}}'
+```
+
+The returned `task_id` can be polled:
+
+```bash
+nyxid proxy request chrono-compute /v1/tasks/<task_id>
+```
+
+## Run A Worker
+
+Start a local OpenAI-compatible backend first, bound to localhost. Then run a
+worker on that same trusted host:
+
+```bash
+export COMPUTE_POOL_WORKER_TOKEN="..."
+
+node integrations/compute-pool-service/worker.mjs \
+  --service-url http://127.0.0.1:8787 \
+  --worker home-4060-a \
+  --endpoint-url http://127.0.0.1:8000/v1/chat/completions \
+  --backend vllm \
+  --host-kind linux-nvidia \
+  --gpu-name "RTX 4060" \
+  --model codex-local
+```
+
+Use `--model '*'` only for workers that should accept any submitted model.
+Workers with no advertised model do not claim model-routed work.
+
+If the local backend needs a token, keep it on the worker host:
+
+```bash
+export LOCAL_BACKEND_TOKEN="..."
+
+node integrations/compute-pool-service/worker.mjs \
+  --service-url http://127.0.0.1:8787 \
+  --worker home-4090-a \
+  --endpoint-url http://127.0.0.1:8000/v1/chat/completions \
+  --backend-token-env LOCAL_BACKEND_TOKEN \
+  --model codex-local
+```
+
+`LOCAL_BACKEND_TOKEN` is sent only to the local endpoint. It is not sent to
+NyxID and is not sent to compute-pool-service.
+
+## API Summary
+
+Consumer API, called through NyxID service proxy:
+
+- `POST /v1/tasks`
+- `GET /v1/tasks/{task_id}`
+- `POST /v1/tasks/{task_id}/cancel`
+- `GET /v1/status`
+
+Worker API, called directly by trusted workers:
+
+- `POST /worker/task?worker=<label>`
+- `POST /worker/ack`
+- `POST /worker/result`
+
+The OpenAPI spec for the consumer API is in `openapi.yaml`.
+
+## Current Limitations
+
+- Local JSON store only; not multi-process safe.
+- One task per worker process at a time.
+- No built-in Slurm adapter yet.
+- No NyxID catalog seed yet; add as a custom service for now.
+- No service-level load balancing in NyxID yet; route one service instance
+  through one node, then extend NyxID service routing separately.
+
+## Why This Is Not NyxID Core
+
+NyxID is the control plane: identity, org membership, agent API keys,
+credential brokering, node routing, proxying, audit, and service governance.
+Compute pools are a data-plane service. Keeping the queue outside core keeps
+NyxID from becoming a runtime for every business-specific worker protocol.
diff --git a/integrations/compute-pool-service/openapi.yaml b/integrations/compute-pool-service/openapi.yaml
new file mode 100644
index 00000000..cb27f1ee
--- /dev/null
+++ b/integrations/compute-pool-service/openapi.yaml
@@ -0,0 +1,201 @@
+openapi: 3.1.0
+info:
+  title: NyxID Compute Pool Service
+  version: 0.1.0
+  description: Standalone worker-pull compute queue service managed by NyxID as a custom service.
+servers:
+  - url: http://localhost:8787
+security:
+  - bearerAuth: []
+paths:
+  /health:
+    get:
+      security: []
+      summary: Health check
+      responses:
+        "200":
+          description: Service is healthy.
+  /v1/status:
+    get:
+      summary: Get queue and worker status
+      responses:
+        "200":
+          description: Current queue status.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Status"
+  /v1/tasks:
+    post:
+      summary: Submit a compute task
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/SubmitTask"
+      responses:
+        "202":
+          description: Task accepted.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Task"
+  /v1/tasks/{task_id}:
+    get:
+      summary: Get task state or result
+      parameters:
+        - name: task_id
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: Task state.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Task"
+  /v1/tasks/{task_id}/cancel:
+    post:
+      summary: Cancel a queued or dispatched task
+      parameters:
+        - name: task_id
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: Cancelled task state.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Task"
+components:
+  securitySchemes:
+    bearerAuth:
+      type: http
+      scheme: bearer
+  schemas:
+    SubmitTask:
+      type: object
+      required:
+        - model
+      properties:
+        kind:
+          type: string
+          default: chat_completion
+        model:
+          type: string
+        input:
+          type: object
+          additionalProperties: true
+        priority:
+          type: integer
+          default: 0
+        client_ref:
+          type: string
+          description: Optional idempotency key scoped to this service instance.
+    Task:
+      type: object
+      properties:
+        task_id:
+          type: string
+        kind:
+          type: string
+        model:
+          type: string
+        priority:
+          type: integer
+        status:
+          type: string
+          enum:
+            - queued
+            - dispatched
+            - completed
+            - failed
+            - cancelled
+        phase:
+          type:
+            - string
+            - "null"
+        phase_detail:
+          type:
+            - string
+            - "null"
+        assigned_worker:
+          type:
+            - string
+            - "null"
+        output:
+          type:
+            - object
+            - array
+            - string
+            - number
+            - boolean
+            - "null"
+        failure_reason:
+          type:
+            - string
+            - "null"
+        created_at:
+          type: string
+          format: date-time
+        updated_at:
+          type: string
+          format: date-time
+        completed_at:
+          type:
+            - string
+            - "null"
+          format: date-time
+    Status:
+      type: object
+      properties:
+        queued:
+          type: integer
+        dispatched:
+          type: integer
+        active_workers:
+          type: array
+          items:
+            $ref: "#/components/schemas/Worker"
+    Worker:
+      type: object
+      properties:
+        worker_label:
+          type: string
+        host_kind:
+          type:
+            - string
+            - "null"
+        gpu_name:
+          type:
+            - string
+            - "null"
+        backend:
+          type:
+            - string
+            - "null"
+        models:
+          type: array
+          items:
+            type: string
+        max_concurrency:
+          type: integer
+        current_inflight:
+          type: integer
+        current_task_id:
+          type:
+            - string
+            - "null"
+        worker_version:
+          type:
+            - string
+            - "null"
+        last_seen_at:
+          type: string
+          format: date-time
diff --git a/integrations/compute-pool-service/package.json b/integrations/compute-pool-service/package.json
new file mode 100644
index 00000000..7ce29f07
--- /dev/null
+++ b/integrations/compute-pool-service/package.json
@@ -0,0 +1,14 @@
+{
+  "name": "@nyxid/compute-pool-service",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "description": "Standalone worker-pull compute queue service managed through NyxID as a custom service.",
+  "scripts": {
+    "start": "node server.mjs",
+    "worker": "node worker.mjs"
+  },
+  "engines": {
+    "node": ">=20"
+  }
+}
diff --git a/integrations/compute-pool-service/server.mjs b/integrations/compute-pool-service/server.mjs
new file mode 100644
index 00000000..9bf38801
--- /dev/null
+++ b/integrations/compute-pool-service/server.mjs
@@ -0,0 +1,508 @@
+#!/usr/bin/env node
+import { createHash, randomUUID, timingSafeEqual } from "node:crypto";
+import { createServer } from "node:http";
+import { readFile, rename, writeFile } from "node:fs/promises";
+import { dirname, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+
+const HOST = process.env.HOST ?? "127.0.0.1";
+const PORT = Number.parseInt(process.env.PORT ?? "8787", 10);
+const STORE_PATH = resolve(
+  process.env.COMPUTE_POOL_STORE ?? `${__dirname}/.compute-pool-store.json`,
+);
+const API_TOKEN = process.env.COMPUTE_POOL_API_TOKEN ?? "";
+const WORKER_TOKEN = process.env.COMPUTE_POOL_WORKER_TOKEN ?? "";
+const DEV_INSECURE = process.env.COMPUTE_POOL_DEV_INSECURE === "1";
+const TASK_TIMEOUT_SECS = Number.parseInt(
+  process.env.COMPUTE_POOL_TASK_TIMEOUT_SECS ?? "7200",
+  10,
+);
+const RETENTION_DAYS = Number.parseInt(
+  process.env.COMPUTE_POOL_RETENTION_DAYS ?? "30",
+  10,
+);
+const MAX_BODY_BYTES = Number.parseInt(
+  process.env.COMPUTE_POOL_MAX_BODY_BYTES ?? String(8 * 1024 * 1024),
+  10,
+);
+
+if (!DEV_INSECURE && (!API_TOKEN || !WORKER_TOKEN)) {
+  console.error(
+    "COMPUTE_POOL_API_TOKEN and COMPUTE_POOL_WORKER_TOKEN are required. " +
+      "Set COMPUTE_POOL_DEV_INSECURE=1 only for local throwaway testing.",
+  );
+  process.exit(1);
+}
+
+let store = await loadStore();
+
+const server = createServer(async (req, res) => {
+  try {
+    await route(req, res);
+  } catch (err) {
+    if (err?.status) {
+      sendJson(res, err.status, { error: err.message });
+      return;
+    }
+    console.error("request failed:", err?.message ?? String(err));
+    sendJson(res, 500, { error: "internal_error" });
+  }
+});
+
+server.listen(PORT, HOST, () => {
+  console.error(`compute-pool-service listening on http://${HOST}:${PORT}`);
+});
+
+async function route(req, res) {
+  const url = new URL(req.url ?? "/", `http://${req.headers.host ?? "localhost"}`);
+  if (req.method === "GET" && url.pathname === "/health") {
+    sendJson(res, 200, { status: "ok" });
+    return;
+  }
+
+  if (url.pathname.startsWith("/worker/")) {
+    if (!authorized(req, WORKER_TOKEN)) {
+      sendJson(res, 401, { error: "worker_token_invalid" });
+      return;
+    }
+    await routeWorker(req, res, url);
+    return;
+  }
+
+  if (!authorized(req, API_TOKEN)) {
+    sendJson(res, 401, { error: "unauthorized" });
+    return;
+  }
+  await routeConsumer(req, res, url);
+}
+
+async function routeConsumer(req, res, url) {
+  await expireTerminalTasks();
+
+  if (req.method === "GET" && url.pathname === "/v1/status") {
+    sendJson(res, 200, statusPayload());
+    return;
+  }
+
+  if (req.method === "POST" && url.pathname === "/v1/tasks") {
+    const body = await readJson(req);
+    const task = submitTask(body);
+    await saveStore();
+    sendJson(res, 202, taskResponse(task));
+    return;
+  }
+
+  const taskMatch = url.pathname.match(/^\/v1\/tasks\/([^/]+)$/);
+  if (taskMatch && req.method === "GET") {
+    const task = store.tasks[taskMatch[1]];
+    if (!task) {
+      sendJson(res, 404, { error: "task_not_found" });
+      return;
+    }
+    sendJson(res, 200, taskResponse(task));
+    return;
+  }
+
+  const cancelMatch = url.pathname.match(/^\/v1\/tasks\/([^/]+)\/cancel$/);
+  if (cancelMatch && req.method === "POST") {
+    const task = store.tasks[cancelMatch[1]];
+    if (!task) {
+      sendJson(res, 404, { error: "task_not_found" });
+      return;
+    }
+    if (["completed", "failed"].includes(task.status)) {
+      sendJson(res, 409, { error: "task_terminal", status: task.status });
+      return;
+    }
+    if (task.status !== "cancelled") {
+      task.status = "cancelled";
+      task.completed_at = nowIso();
+      task.expires_at = retentionIso();
+      task.updated_at = nowIso();
+      await saveStore();
+    }
+    sendJson(res, 200, taskResponse(task));
+    return;
+  }
+
+  sendJson(res, 404, { error: "not_found" });
+}
+
+async function routeWorker(req, res, url) {
+  await expireTerminalTasks();
+  requeueExpiredLeases();
+
+  if (req.method === "POST" && url.pathname === "/worker/task") {
+    const worker = url.searchParams.get("worker") ?? "";
+    if (!validWorkerLabel(worker)) {
+      sendJson(res, 400, { error: "invalid_worker_label" });
+      return;
+    }
+    const body = await readJson(req);
+    const capabilities = sanitizeCapabilities(body.capabilities ?? {});
+    const existing = Object.values(store.tasks).find(
+      (task) => task.status === "dispatched" && task.assigned_worker === worker,
+    );
+    if (existing) {
+      refreshLease(existing);
+      upsertWorker(worker, capabilities, existing.id);
+      await saveStore();
+      sendJson(res, 200, { status: "task", task: workerTaskPayload(existing, worker) });
+      return;
+    }
+
+    upsertWorker(worker, capabilities, null);
+    const task = nextTaskFor(capabilities);
+    if (!task) {
+      await saveStore();
+      sendJson(res, 200, { status: "idle" });
+      return;
+    }
+    task.status = "dispatched";
+    task.assigned_worker = worker;
+    task.dispatched_at = nowIso();
+    task.lease_expires_at = leaseIso();
+    task.phase = "dispatched";
+    task.phase_at = nowIso();
+    task.updated_at = nowIso();
+    upsertWorker(worker, capabilities, task.id);
+    await saveStore();
+    sendJson(res, 200, { status: "task", task: workerTaskPayload(task, worker) });
+    return;
+  }
+
+  if (req.method === "POST" && url.pathname === "/worker/ack") {
+    const body = await readJson(req);
+    const task = store.tasks[body.task_id];
+    const worker = String(body.worker ?? "");
+    if (!validWorkerLabel(worker)) {
+      sendJson(res, 400, { error: "invalid_worker_label" });
+      return;
+    }
+    upsertWorker(worker, sanitizeCapabilities(body.capabilities ?? {}), body.task_id);
+    if (!task || task.status !== "dispatched" || task.assigned_worker !== worker) {
+      await saveStore();
+      sendJson(res, 200, { status: "cancelled" });
+      return;
+    }
+    task.lease_expires_at = leaseIso();
+    if (typeof body.phase === "string") {
+      task.phase = truncate(body.phase, 80);
+      task.phase_at = nowIso();
+    }
+    if (typeof body.phase_detail === "string") {
+      task.phase_detail = truncate(body.phase_detail, 500);
+    }
+    task.updated_at = nowIso();
+    await saveStore();
+    sendJson(res, 200, { status: "ok" });
+    return;
+  }
+
+  if (req.method === "POST" && url.pathname === "/worker/result") {
+    const body = await readJson(req);
+    const task = store.tasks[body.task_id];
+    const worker = String(body.worker ?? "");
+    if (!task || task.status !== "dispatched" || task.assigned_worker !== worker) {
+      sendJson(res, 200, { status: "ignored" });
+      return;
+    }
+    if (typeof body.failure_reason === "string" && body.failure_reason.trim()) {
+      task.status = "failed";
+      task.failure_reason = truncate(body.failure_reason, 500);
+    } else {
+      task.status = "completed";
+      task.output = body.output ?? null;
+    }
+    task.completed_at = nowIso();
+    task.expires_at = retentionIso();
+    task.lease_expires_at = null;
+    task.updated_at = nowIso();
+    if (store.workers[worker]) {
+      store.workers[worker].current_task_id = null;
+      store.workers[worker].last_seen_at = nowIso();
+    }
+    await saveStore();
+    sendJson(res, 200, { status: task.status === "failed" ? "saved_failed" : "saved" });
+    return;
+  }
+
+  sendJson(res, 404, { error: "not_found" });
+}
+
+function submitTask(body) {
+  const kind = typeof body.kind === "string" && body.kind.trim() ? body.kind.trim() : "chat_completion";
+  const model = String(body.model ?? "").trim();
+  if (!model) {
+    throw httpError(400, "model_required");
+  }
+  const clientRef = typeof body.client_ref === "string" && body.client_ref.trim()
+    ? body.client_ref.trim()
+    : null;
+  if (clientRef && store.client_refs[clientRef]) {
+    return store.tasks[store.client_refs[clientRef]];
+  }
+  const task = {
+    id: randomUUID(),
+    kind: truncate(kind, 64),
+    model: truncate(model, 160),
+    input: body.input ?? {},
+    priority: Number.isFinite(body.priority) ? Math.trunc(body.priority) : 0,
+    client_ref: clientRef,
+    status: "queued",
+    phase: null,
+    phase_detail: null,
+    phase_at: null,
+    assigned_worker: null,
+    dispatched_at: null,
+    lease_expires_at: null,
+    output: null,
+    failure_reason: null,
+    completed_at: null,
+    expires_at: null,
+    created_at: nowIso(),
+    updated_at: nowIso(),
+  };
+  store.tasks[task.id] = task;
+  if (clientRef) {
+    store.client_refs[clientRef] = task.id;
+  }
+  return task;
+}
+
+function nextTaskFor(capabilities) {
+  const models = normalizeModels(capabilities.models ?? []);
+  const acceptsAny = models.includes("*");
+  const candidates = Object.values(store.tasks).filter((task) => {
+    if (task.status !== "queued") return false;
+    if (acceptsAny) return true;
+    if (models.length === 0) return false;
+    return models.includes(task.model);
+  });
+  candidates.sort((a, b) => {
+    if (b.priority !== a.priority) return b.priority - a.priority;
+    return a.created_at.localeCompare(b.created_at);
+  });
+  return candidates[0] ?? null;
+}
+
+function requeueExpiredLeases() {
+  const now = Date.now();
+  for (const task of Object.values(store.tasks)) {
+    if (
+      task.status === "dispatched" &&
+      task.lease_expires_at &&
+      Date.parse(task.lease_expires_at) < now
+    ) {
+      task.status = "queued";
+      task.assigned_worker = null;
+      task.dispatched_at = null;
+      task.lease_expires_at = null;
+      task.phase = "requeued_after_lease_expiry";
+      task.updated_at = nowIso();
+    }
+  }
+}
+
+function refreshLease(task) {
+  task.lease_expires_at = leaseIso();
+  task.updated_at = nowIso();
+}
+
+function upsertWorker(worker, capabilities, currentTaskId) {
+  const existing = store.workers[worker] ?? { first_seen_at: nowIso() };
+  store.workers[worker] = {
+    ...existing,
+    worker_label: worker,
+    host_kind: stringOrNull(capabilities.host_kind),
+    gpu_name: stringOrNull(capabilities.gpu_name),
+    backend: stringOrNull(capabilities.backend),
+    models: normalizeModels(capabilities.models ?? []),
+    max_concurrency: positiveInt(capabilities.max_concurrency, 1),
+    current_inflight: positiveInt(capabilities.current_inflight, currentTaskId ? 1 : 0),
+    current_task_id: currentTaskId,
+    worker_version: stringOrNull(capabilities.worker_version),
+    last_seen_at: nowIso(),
+  };
+}
+
+function statusPayload() {
+  const queued = Object.values(store.tasks).filter((task) => task.status === "queued").length;
+  const dispatched = Object.values(store.tasks).filter((task) => task.status === "dispatched").length;
+  const recent = Date.now() - 120_000;
+  const workers = Object.values(store.workers).filter(
+    (worker) => Date.parse(worker.last_seen_at) >= recent,
+  );
+  return { queued, dispatched, active_workers: workers };
+}
+
+function taskResponse(task) {
+  return {
+    task_id: task.id,
+    kind: task.kind,
+    model: task.model,
+    priority: task.priority,
+    status: task.status,
+    phase: task.phase,
+    phase_detail: task.phase_detail,
+    assigned_worker: task.assigned_worker,
+    output: task.output,
+    failure_reason: task.failure_reason,
+    created_at: task.created_at,
+    updated_at: task.updated_at,
+    completed_at: task.completed_at,
+  };
+}
+
+function workerTaskPayload(task, worker) {
+  return {
+    task_id: task.id,
+    kind: task.kind,
+    model: task.model,
+    input: task.input,
+    priority: task.priority,
+    assigned_worker: worker,
+    submitted_at: task.created_at,
+  };
+}
+
+async function readJson(req) {
+  const chunks = [];
+  let size = 0;
+  for await (const chunk of req) {
+    size += chunk.length;
+    if (size > MAX_BODY_BYTES) {
+      throw httpError(413, "payload_too_large");
+    }
+    chunks.push(chunk);
+  }
+  if (chunks.length === 0) return {};
+  try {
+    return JSON.parse(Buffer.concat(chunks).toString("utf8"));
+  } catch {
+    throw httpError(400, "invalid_json");
+  }
+}
+
+async function loadStore() {
+  try {
+    const raw = await readFile(STORE_PATH, "utf8");
+    const parsed = JSON.parse(raw);
+    return {
+      tasks: parsed.tasks ?? {},
+      workers: parsed.workers ?? {},
+      client_refs: parsed.client_refs ?? {},
+    };
+  } catch {
+    return { tasks: {}, workers: {}, client_refs: {} };
+  }
+}
+
+async function saveStore() {
+  const tmp = `${STORE_PATH}.tmp`;
+  await writeFile(tmp, `${JSON.stringify(store, null, 2)}\n`, { mode: 0o600 });
+  await rename(tmp, STORE_PATH);
+}
+
+async function expireTerminalTasks() {
+  const now = Date.now();
+  let changed = false;
+  for (const [id, task] of Object.entries(store.tasks)) {
+    if (task.expires_at && Date.parse(task.expires_at) < now) {
+      delete store.tasks[id];
+      if (task.client_ref) delete store.client_refs[task.client_ref];
+      changed = true;
+    }
+  }
+  if (changed) await saveStore();
+}
+
+function authorized(req, token) {
+  if (DEV_INSECURE) return true;
+  const auth = req.headers.authorization ?? "";
+  if (!auth.startsWith("Bearer ")) return false;
+  return safeEqual(auth.slice("Bearer ".length), token);
+}
+
+function safeEqual(a, b) {
+  const left = Buffer.from(hash(a));
+  const right = Buffer.from(hash(b));
+  return timingSafeEqual(left, right);
+}
+
+function hash(value) {
+  return createHash("sha256").update(String(value)).digest("hex");
+}
+
+function sendJson(res, status, payload) {
+  const body = JSON.stringify(payload);
+  res.writeHead(status, {
+    "content-type": "application/json",
+    "content-length": Buffer.byteLength(body),
+  });
+  res.end(body);
+}
+
+function normalizeModels(models) {
+  const out = [];
+  for (const model of Array.isArray(models) ? models.slice(0, 200) : []) {
+    const normalized = String(model).trim().slice(0, 160);
+    if (normalized && !out.includes(normalized)) out.push(normalized);
+  }
+  return out;
+}
+
+function sanitizeCapabilities(input) {
+  return {
+    host_kind: stringOrNull(input.host_kind),
+    gpu_name: stringOrNull(input.gpu_name),
+    backend: stringOrNull(input.backend),
+    models: normalizeModels(input.models ?? []),
+    max_concurrency: positiveInt(input.max_concurrency, 1),
+    current_inflight: positiveInt(input.current_inflight, 0),
+    worker_version: stringOrNull(input.worker_version),
+  };
+}
+
+function stringOrNull(value) {
+  return typeof value === "string" && value.trim() ? value.trim().slice(0, 160) : null;
+}
+
+function positiveInt(value, fallback) {
+  const parsed = Number.parseInt(String(value ?? ""), 10);
+  return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback;
+}
+
+function validWorkerLabel(label) {
+  return /^[A-Za-z0-9_-]{1,64}$/.test(label);
+}
+
+function truncate(value, max) {
+  return String(value).slice(0, max);
+}
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function leaseIso() {
+  return new Date(Date.now() + TASK_TIMEOUT_SECS * 1000).toISOString();
+}
+
+function retentionIso() {
+  return new Date(Date.now() + RETENTION_DAYS * 86_400_000).toISOString();
+}
+
+function httpError(status, code) {
+  const err = new Error(code);
+  err.status = status;
+  return err;
+}
+
+process.on("uncaughtException", (err) => {
+  if (err?.status) return;
+  console.error(err);
+  process.exit(1);
+});
diff --git a/integrations/compute-pool-service/worker.mjs b/integrations/compute-pool-service/worker.mjs
new file mode 100644
index 00000000..4c8769b1
--- /dev/null
+++ b/integrations/compute-pool-service/worker.mjs
@@ -0,0 +1,169 @@
+#!/usr/bin/env node
+import { setTimeout as sleep } from "node:timers/promises";
+
+const args = parseArgs(process.argv.slice(2));
+
+const serviceUrl = required(args["service-url"], "--service-url").replace(/\/$/, "");
+const worker = required(args.worker, "--worker");
+const endpointUrl = required(args["endpoint-url"], "--endpoint-url");
+const tokenEnv = args["token-env"] ?? "COMPUTE_POOL_WORKER_TOKEN";
+const workerToken = process.env[tokenEnv];
+if (!workerToken) {
+  throw new Error(`${tokenEnv} is empty or unset`);
+}
+const backendToken = args["backend-token-env"] ? process.env[args["backend-token-env"]] : null;
+const models = arrayArg(args.model);
+const pollIntervalMs = seconds(args["poll-interval-secs"], 5) * 1000;
+const ackIntervalMs = seconds(args["ack-interval-secs"], 30) * 1000;
+const requestTimeoutMs = seconds(args["request-timeout-secs"], 14_400) * 1000;
+
+console.error(`compute worker ${worker} polling ${serviceUrl}`);
+
+for (;;) {
+  const polled = await postJson(`${serviceUrl}/worker/task?worker=${encodeURIComponent(worker)}`, {
+    capabilities: capabilities(0),
+  });
+  if (polled.status === "idle") {
+    await sleep(pollIntervalMs);
+    continue;
+  }
+  const task = polled.task;
+  console.error(`claimed task ${task.task_id} kind=${task.kind} model=${task.model}`);
+  try {
+    const output = await executeWithHeartbeats(task);
+    if (output.cancelled) {
+      console.error(`task ${task.task_id} cancelled`);
+      continue;
+    }
+    await postJson(`${serviceUrl}/worker/result`, {
+      task_id: task.task_id,
+      worker,
+      output: output.value,
+    });
+    console.error(`completed task ${task.task_id}`);
+  } catch (err) {
+    const reason = err?.message ?? String(err);
+    await postJson(`${serviceUrl}/worker/result`, {
+      task_id: task.task_id,
+      worker,
+      failure_reason: reason,
+    });
+    console.error(`failed task ${task.task_id}: ${reason}`);
+  }
+}
+
+async function executeWithHeartbeats(task) {
+  const controller = new AbortController();
+  const local = executeLocal(task, controller.signal);
+  let done = false;
+  const heartbeats = (async () => {
+    while (!done) {
+      await sleep(ackIntervalMs);
+      if (done) return { cancelled: false };
+      const ack = await postJson(`${serviceUrl}/worker/ack`, {
+        task_id: task.task_id,
+        worker,
+        phase: "running",
+        capabilities: capabilities(1),
+      });
+      if (ack.status === "cancelled") {
+        controller.abort();
+        return { cancelled: true };
+      }
+    }
+    return { cancelled: false };
+  })();
+
+  const winner = await Promise.race([
+    local.then((value) => ({ value })),
+    heartbeats,
+  ]);
+  done = true;
+  if (winner.cancelled) return winner;
+  return winner;
+}
+
+async function executeLocal(task, signal) {
+  const body = typeof task.input === "object" && task.input !== null ? { ...task.input } : {};
+  if (!body.model) body.model = task.model;
+
+  const timeout = AbortSignal.timeout(requestTimeoutMs);
+  const combined = AbortSignal.any([signal, timeout]);
+  const headers = { "content-type": "application/json" };
+  if (backendToken) headers.authorization = `Bearer ${backendToken}`;
+
+  const response = await fetch(endpointUrl, {
+    method: "POST",
+    headers,
+    body: JSON.stringify(body),
+    signal: combined,
+  });
+  const text = await response.text();
+  if (!response.ok) {
+    throw new Error(`local backend HTTP ${response.status}: ${text.slice(0, 500)}`);
+  }
+  return text ? JSON.parse(text) : null;
+}
+
+async function postJson(url, body) {
+  const response = await fetch(url, {
+    method: "POST",
+    headers: {
+      authorization: `Bearer ${workerToken}`,
+      "content-type": "application/json",
+    },
+    body: JSON.stringify(body),
+  });
+  const text = await response.text();
+  if (!response.ok) {
+    throw new Error(`service HTTP ${response.status}: ${text.slice(0, 500)}`);
+  }
+  return text ? JSON.parse(text) : {};
+}
+
+function capabilities(currentInflight) {
+  return {
+    host_kind: args["host-kind"] ?? null,
+    gpu_name: args["gpu-name"] ?? null,
+    backend: args.backend ?? null,
+    models,
+    max_concurrency: Number.parseInt(args["max-concurrency"] ?? "1", 10),
+    current_inflight: currentInflight,
+    worker_version: "compute-pool-service-worker/0.1.0",
+  };
+}
+
+function parseArgs(argv) {
+  const out = {};
+  for (let i = 0; i < argv.length; i += 1) {
+    const item = argv[i];
+    if (!item.startsWith("--")) continue;
+    const key = item.slice(2);
+    const next = argv[i + 1];
+    if (!next || next.startsWith("--")) {
+      out[key] = "true";
+    } else if (out[key]) {
+      out[key] = Array.isArray(out[key]) ? [...out[key], next] : [out[key], next];
+      i += 1;
+    } else {
+      out[key] = next;
+      i += 1;
+    }
+  }
+  return out;
+}
+
+function arrayArg(value) {
+  if (!value) return [];
+  return Array.isArray(value) ? value : [value];
+}
+
+function required(value, name) {
+  if (!value) throw new Error(`${name} is required`);
+  return value;
+}
+
+function seconds(value, fallback) {
+  const parsed = Number.parseInt(value ?? "", 10);
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
+}

From 017f12d5b8ad286308d60d3661cd392d88e7a35b Mon Sep 17 00:00:00 2001
From: AlyciaBHZ <50111876+AlyciaBHZ@users.noreply.github.com>
Date: Wed, 17 Jun 2026 18:32:54 +0800
Subject: [PATCH 2/3] Clarify compute service-pool boundary

---
 integrations/compute-pool-service/README.md | 20 ++++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/integrations/compute-pool-service/README.md b/integrations/compute-pool-service/README.md
index 00097d10..ef533e21 100644
--- a/integrations/compute-pool-service/README.md
+++ b/integrations/compute-pool-service/README.md
@@ -1,11 +1,15 @@
 # Compute Pool Service
 
-This integration is the long-term shape for shared GPU / Mac / lab compute:
-the queue and worker protocol live outside NyxID core. NyxID manages the
-service as a normal user/org service: auth, agent API keys, credential
-injection, node routing, audit metadata, and future load balancing stay in
+This integration is a reference external service for shared GPU / Mac / lab
+compute. The queue and worker protocol live outside NyxID core. NyxID
+manages the service as a normal user/org service: auth, agent API keys,
+credential injection, node routing, proxying, and audit metadata stay in
 NyxID; compute-specific task state stays in this service.
 
+This is not a NyxID service-pool framework. Cross-service counting, quotas,
+metering, and load balancing should be handled by a future generic NyxID
+service-pool design rather than by a compute-specific core API.
+
 This integration does not require a NyxID org model change. To share it with
 company members, create the NyxID service under the existing org owner and
 use the current org membership/admin checks that already apply to services.
@@ -40,6 +44,8 @@ The standalone service stores task input/output in its own local store. The
 default store is a JSON file intended for smoke tests and small trusted
 deployments, not production durability. A production version should replace
 the store with Postgres, Redis, MongoDB, or another managed queue backend.
+NyxID-level metering and quota decisions should count proxied calls to this
+service the same way they would count any other registered service.
 
 ## Start The Service
 
@@ -162,8 +168,10 @@ The OpenAPI spec for the consumer API is in `openapi.yaml`.
 - One task per worker process at a time.
 - No built-in Slurm adapter yet.
 - No NyxID catalog seed yet; add as a custom service for now.
-- No service-level load balancing in NyxID yet; route one service instance
-  through one node, then extend NyxID service routing separately.
+- No NyxID-level service pool yet. This service exposes `/v1/status` as a
+  capacity signal, but generic service-instance load balancing, org/agent
+  quotas, usage counting, and metering belong in NyxID's future service-pool
+  layer.
 
 ## Why This Is Not NyxID Core
 

From 0e7165082ee5683f6f44951cf835c55e19e7d656 Mon Sep 17 00:00:00 2001
From: AlyciaBHZ <50111876+AlyciaBHZ@users.noreply.github.com>
Date: Thu, 18 Jun 2026 15:41:05 +0800
Subject: [PATCH 3/3] Address compute service robustness notes

---
 integrations/compute-pool-service/README.md   | 15 ++++++
 .../compute-pool-service/openapi.yaml         | 15 +++++-
 integrations/compute-pool-service/server.mjs  | 11 +++--
 integrations/compute-pool-service/worker.mjs  | 47 +++++++++++++++----
 4 files changed, 76 insertions(+), 12 deletions(-)

diff --git a/integrations/compute-pool-service/README.md b/integrations/compute-pool-service/README.md
index ef533e21..141a7bee 100644
--- a/integrations/compute-pool-service/README.md
+++ b/integrations/compute-pool-service/README.md
@@ -47,6 +47,12 @@ the store with Postgres, Redis, MongoDB, or another managed queue backend.
 NyxID-level metering and quota decisions should count proxied calls to this
 service the same way they would count any other registered service.
 
+The reference service has one consumer API token. Anyone who can call the
+service can read or cancel a task if they know its task id. Use NyxID service
+ownership, agent API-key scopes, and org policy to control who can call the
+service; add per-consumer task ownership in a production backend if multiple
+tenants share one service token.
+
 ## Start The Service
 
 Generate two independent tokens:
@@ -129,6 +135,13 @@ node integrations/compute-pool-service/worker.mjs \
 Use `--model '*'` only for workers that should accept any submitted model.
 Workers with no advertised model do not claim model-routed work.
 
+The worker heartbeats with `/worker/ack` while the local request is running.
+A transient ack failure is retried; after `--max-ack-failures` consecutive
+failures (default 3), the worker aborts the local request and reports failure.
+Set `COMPUTE_POOL_TASK_TIMEOUT_SECS` high enough to cover normal ack outages
+for your deployment. The default lease is 2 hours and each successful ack
+refreshes it.
+
 If the local backend needs a token, keep it on the worker host:
 
 ```bash
@@ -165,6 +178,8 @@ The OpenAPI spec for the consumer API is in `openapi.yaml`.
 ## Current Limitations
 
 - Local JSON store only; not multi-process safe.
+- The JSON store is serialized inside one process and uses per-write tmp
+  files, but it is still a smoke-test backend rather than durable storage.
 - One task per worker process at a time.
 - No built-in Slurm adapter yet.
 - No NyxID catalog seed yet; add as a custom service for now.
diff --git a/integrations/compute-pool-service/openapi.yaml b/integrations/compute-pool-service/openapi.yaml
index cb27f1ee..96b56323 100644
--- a/integrations/compute-pool-service/openapi.yaml
+++ b/integrations/compute-pool-service/openapi.yaml
@@ -4,7 +4,7 @@ info:
   version: 0.1.0
   description: Standalone worker-pull compute queue service managed by NyxID as a custom service.
 servers:
-  - url: http://localhost:8787
+  - url: http://127.0.0.1:8787
 security:
   - bearerAuth: []
 paths:
@@ -73,6 +73,12 @@ paths:
             application/json:
               schema:
                 $ref: "#/components/schemas/Task"
+        "409":
+          description: Task is already completed or failed.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/Error"
 components:
   securitySchemes:
     bearerAuth:
@@ -199,3 +205,10 @@ components:
         last_seen_at:
           type: string
           format: date-time
+    Error:
+      type: object
+      properties:
+        error:
+          type: string
+        status:
+          type: string
diff --git a/integrations/compute-pool-service/server.mjs b/integrations/compute-pool-service/server.mjs
index 9bf38801..ebb5a6e0 100644
--- a/integrations/compute-pool-service/server.mjs
+++ b/integrations/compute-pool-service/server.mjs
@@ -37,6 +37,8 @@ if (!DEV_INSECURE && (!API_TOKEN || !WORKER_TOKEN)) {
 }
 
 let store = await loadStore();
+let saveChain = Promise.resolve();
+let saveSeq = 0;
 
 const server = createServer(async (req, res) => {
   try {
@@ -401,9 +403,12 @@ async function loadStore() {
 }
 
 async function saveStore() {
-  const tmp = `${STORE_PATH}.tmp`;
-  await writeFile(tmp, `${JSON.stringify(store, null, 2)}\n`, { mode: 0o600 });
-  await rename(tmp, STORE_PATH);
+  saveChain = saveChain.catch(() => {}).then(async () => {
+    const tmp = `${STORE_PATH}.${process.pid}.${saveSeq += 1}.tmp`;
+    await writeFile(tmp, `${JSON.stringify(store, null, 2)}\n`, { mode: 0o600 });
+    await rename(tmp, STORE_PATH);
+  });
+  return saveChain;
 }
 
 async function expireTerminalTasks() {
diff --git a/integrations/compute-pool-service/worker.mjs b/integrations/compute-pool-service/worker.mjs
index 4c8769b1..c35560c8 100644
--- a/integrations/compute-pool-service/worker.mjs
+++ b/integrations/compute-pool-service/worker.mjs
@@ -15,6 +15,7 @@ const backendToken = args["backend-token-env"] ? process.env[args["backend-token
 const models = arrayArg(args.model);
 const pollIntervalMs = seconds(args["poll-interval-secs"], 5) * 1000;
 const ackIntervalMs = seconds(args["ack-interval-secs"], 30) * 1000;
+const maxAckFailures = positiveInt(args["max-ack-failures"], 3);
 const requestTimeoutMs = seconds(args["request-timeout-secs"], 14_400) * 1000;
 
 console.error(`compute worker ${worker} polling ${serviceUrl}`);
@@ -54,18 +55,29 @@ for (;;) {
 
 async function executeWithHeartbeats(task) {
   const controller = new AbortController();
-  const local = executeLocal(task, controller.signal);
+  const local = executeLocal(task, controller.signal).then(
+    (value) => ({ value }),
+    (err) => ({ failed: err }),
+  );
   let done = false;
+  let ackFailures = 0;
   const heartbeats = (async () => {
     while (!done) {
       await sleep(ackIntervalMs);
       if (done) return { cancelled: false };
-      const ack = await postJson(`${serviceUrl}/worker/ack`, {
-        task_id: task.task_id,
-        worker,
-        phase: "running",
-        capabilities: capabilities(1),
-      });
+      const ack = await postAck(task);
+      if (ack.failed) {
+        ackFailures += 1;
+        console.error(
+          `ack failed for task ${task.task_id} (${ackFailures}/${maxAckFailures}): ${ack.error}`,
+        );
+        if (ackFailures >= maxAckFailures) {
+          controller.abort();
+          return { failed: new Error(`ack failed ${ackFailures} times; local request aborted`) };
+        }
+        continue;
+      }
+      ackFailures = 0;
       if (ack.status === "cancelled") {
         controller.abort();
         return { cancelled: true };
@@ -75,14 +87,28 @@ async function executeWithHeartbeats(task) {
   })();
 
   const winner = await Promise.race([
-    local.then((value) => ({ value })),
+    local,
     heartbeats,
   ]);
   done = true;
+  if (winner.failed) throw winner.failed;
   if (winner.cancelled) return winner;
   return winner;
 }
 
+async function postAck(task) {
+  try {
+    return await postJson(`${serviceUrl}/worker/ack`, {
+      task_id: task.task_id,
+      worker,
+      phase: "running",
+      capabilities: capabilities(1),
+    });
+  } catch (err) {
+    return { failed: true, error: err?.message ?? String(err) };
+  }
+}
+
 async function executeLocal(task, signal) {
   const body = typeof task.input === "object" && task.input !== null ? { ...task.input } : {};
   if (!body.model) body.model = task.model;
@@ -167,3 +193,8 @@ function seconds(value, fallback) {
   const parsed = Number.parseInt(value ?? "", 10);
   return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
 }
+
+function positiveInt(value, fallback) {
+  const parsed = Number.parseInt(value ?? "", 10);
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
+}