From 5ea6f19637409e3685e90058dc7a38ae9f908fa6 Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Tue, 3 Nov 2026 19:01:39 -0500 Subject: [PATCH] Claude Code as a cloud worker program; fix self-hosted prefetch; release 0.6.0 plan.workerProgram chooses codex (default) or claude. ClaudeSessionDriver offers Claude Code's stream-json interface through the same contract the runtime uses for a Codex app-server, so operator commands, stop receipts, traces, workflows and evidence are shared. The bootstrap installs one pinned release by checksum. factory-selfhost deploy could not run on a new model: staged sources lacked the profiles, and the helper run needed a secret only the deployment creates. Fixes #15 --- .github/workflows/public-sdk.yml | 2 +- AGENTS.md | 3 +- PUBLIC_SDK.md | 80 +++- package-lock.json | 4 +- package.json | 2 +- scripts/verify-public-sdk.mjs | 1 + selfhost/README.md | 3 +- selfhost/modal/app.py | 59 +-- src/cloud-lite/board.mjs | 2 +- src/cloud-lite/bootstrap.mjs | 43 +- src/cloud-lite/claude-driver.mjs | 308 ++++++++++++++ src/cloud-lite/evidence-worker.mjs | 2 +- src/cloud-lite/fleet.mjs | 23 +- src/cloud-lite/protocol.mjs | 13 +- src/cloud-lite/runtime.mjs | 56 ++- src/public-sdk.mjs | 2 +- src/selfhost/modal.mjs | 3 + test/cloud-lite-claude.test.mjs | 418 +++++++++++++++++++ test/cloud-lite-provider.test.mjs | 3 +- test/cloud-lite.test.mjs | 3 +- test/fixtures/fake-claude.mjs | 80 ++++ test/generic-swarm-evidence.test.mjs | 3 +- test/model-provider.test.mjs | 3 +- test/selfhost-modal.test.mjs | 18 +- test/workflow-m0-lifecycle.test.mjs | 3 +- test/workflow-m1-portable-artifacts.test.mjs | 3 +- test/workflow-m2-ack-recovery.test.mjs | 3 +- test/workflow-m2-authority-recovery.test.mjs | 3 +- test/workflow-native-wake.test.mjs | 3 +- test/workflows.test.mjs | 3 +- 30 files changed, 1044 insertions(+), 108 deletions(-) create mode 100644 src/cloud-lite/claude-driver.mjs create mode 100644 test/cloud-lite-claude.test.mjs create mode 100644 test/fixtures/fake-claude.mjs diff --git a/.github/workflows/public-sdk.yml b/.github/workflows/public-sdk.yml index 23af432..e9d6740 100644 --- a/.github/workflows/public-sdk.yml +++ b/.github/workflows/public-sdk.yml @@ -24,7 +24,7 @@ jobs: - run: npm ci - run: npm audit --omit=dev --audit-level=high - name: Focused tests, leaving nothing in the temp location - run: node scripts/test-temp-guard.mjs test/cloud-lite.test.mjs test/cloud-lite-provider.test.mjs test/model-provider.test.mjs test/selfhost-modal.test.mjs test/public-sdk.test.mjs test/template-catalog.test.mjs test/factory-team-template.test.mjs test/flujo-workspace.test.mjs test/savia-case-plan.test.mjs test/savia-topology.test.mjs test/managed-fleet.test.mjs test/managed-fleet-real-module.test.mjs test/managed-fleet-provider-closure.test.mjs test/provider-cell-retirement.test.mjs test/presentation.test.mjs test/presentation-modal.test.mjs test/presentation-closure.test.mjs test/observatory-client.test.mjs test/workflows.test.mjs test/workflow-native-wake.test.mjs test/workflow-m0-lifecycle.test.mjs test/workflow-m1-portable-artifacts.test.mjs test/workflow-m2-ack-recovery.test.mjs test/workflow-m2-authority-recovery.test.mjs + run: node scripts/test-temp-guard.mjs test/cloud-lite.test.mjs test/cloud-lite-claude.test.mjs test/cloud-lite-provider.test.mjs test/model-provider.test.mjs test/selfhost-modal.test.mjs test/public-sdk.test.mjs test/template-catalog.test.mjs test/factory-team-template.test.mjs test/flujo-workspace.test.mjs test/savia-case-plan.test.mjs test/savia-topology.test.mjs test/managed-fleet.test.mjs test/managed-fleet-real-module.test.mjs test/managed-fleet-provider-closure.test.mjs test/provider-cell-retirement.test.mjs test/presentation.test.mjs test/presentation-modal.test.mjs test/presentation-closure.test.mjs test/observatory-client.test.mjs test/workflows.test.mjs test/workflow-native-wake.test.mjs test/workflow-m0-lifecycle.test.mjs test/workflow-m1-portable-artifacts.test.mjs test/workflow-m2-ack-recovery.test.mjs test/workflow-m2-authority-recovery.test.mjs - name: Self-hosted model settings (Python, offline) run: python3 -B -m unittest discover -s selfhost/modal -p "test_*.py" - name: Install and exercise the packed public package diff --git a/AGENTS.md b/AGENTS.md index a821232..6d9e7bf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -12,7 +12,8 @@ For FACTORY, match `.github/workflows/public-sdk.yml` on both platforms: `npm ci`, production dependency audit, the focused public SDK tests, a fresh install of `npm pack` output, swarm creation, CLI help, and installed MCP stdio smoke. Run the supported Node 22.17 and Node 24 lines. On this Windows host, use an -exact supported Node binary; for Linux, a disposable local Docker container +exact supported Node binary and put Windows' `tar.exe` ahead of Git's on the +PATH (Git's reads `C:` as a remote host); for Linux, a disposable local Docker container with a tracked-source archive avoids platform-mixed `node_modules` and local runtime files. diff --git a/PUBLIC_SDK.md b/PUBLIC_SDK.md index afc7499..7f309e9 100644 --- a/PUBLIC_SDK.md +++ b/PUBLIC_SDK.md @@ -1,17 +1,17 @@ # Public local Factory SDK, CLI, and MCP -Version 0.5.0 is distributed as a versioned package in -[GitHub Releases](https://github.com/flujo-app/FACTORY/releases/tag/v0.5.0): +Version 0.6.0 is distributed as a versioned package in +[GitHub Releases](https://github.com/flujo-app/FACTORY/releases/tag/v0.6.0): ```sh -npm install https://github.com/flujo-app/FACTORY/releases/download/v0.5.0/flujo-factory-0.5.0.tgz +npm install https://github.com/flujo-app/FACTORY/releases/download/v0.6.0/flujo-factory-0.6.0.tgz ``` FACTORY can serve an open-weight model on your own Modal GPUs and run workers on it. `ModalSelfHostedModel` and `factory-selfhost` deploy, verify, stop and retire a vLLM endpoint with Qwen3.8-27B (H100) or Nemotron 3.5 Super (B300) profiles; see [selfhost/README.md](selfhost/README.md). Its endpoint description is a model provider that -cloud lite workers (Codex) and local lite workers (Codex or Claude Code) use directly. See +cloud and local lite workers (Codex or Claude Code) use directly. See [Self-hosted models and operator-driven jobs](#self-hosted-models-and-operator-driven-jobs) for the provider, read-only deploy keys, prepare hooks, operator-driven jobs and quieter traces added in 0.4.0. All of it is optional; existing plans behave as before, except that @@ -203,11 +203,13 @@ reads the key from the variable the file names in `envKey`. ## FLUJO worker dispatch -### Cloud Codex lite workers (development) +### Cloud lite workers (development) `FactoryCloudLiteFleet` provisions a dedicated Fly fleet running native Codex -app-server harnesses with their built-in tools and only FACTORY's board MCP. -Each machine has one orchestrator and `agentsPerMachine - 1` subordinate Codex +app-server harnesses (or Claude Code, see +[Claude Code as the worker program](#claude-code-as-the-worker-program)) with their +built-in tools and only FACTORY's board MCP. +Each machine has one orchestrator and `agentsPerMachine - 1` subordinate sessions, isolated Git worktrees, persisted native histories, and durable traces. The SDK starts and registers the configured pool; additional native automatic fanout is disabled so every subordinate has its own board identity and policy. @@ -260,6 +262,64 @@ while workers execute; a disconnected host can prevent authentication renewal. The runtime does not switch to an API key. Credentials never belong in an image, public journal, source commit or trace. Store the fleet directory privately. +#### Claude Code as the worker program + +Set `workerProgram: 'claude'` in the plan (the default stays `'codex'`). Everything +else works the same way: the board, operator commands and stop receipts, durable +traces, prepare hooks, evidence collection and verified workflows. + +```js +import { FactoryCloudLiteFleet, ClaudeSubscriptionBroker } from 'flujo-factory'; + +// On a subscription: a token from `claude setup-token`, in a private file on this host. +const fleet = new FactoryCloudLiteFleet({ directory: '/absolute/private/fleet-run', + flujoCloudPath: '/absolute/installed/flujo-cloud', + auth: new ClaudeSubscriptionBroker({ tokenFile: '/absolute/private/claude-token' }) }); +await fleet.deploy({ runId: 'my-claude-run', app: 'my-claude-app', goal: ownerGoal, + workerProgram: 'claude', model: 'opus', subagentModel: 'sonnet', + machines: 1, agentsPerMachine: 4 }, { image: immutableWorkerImage, bundlePath: '/absolute/private/flujo-factory.tgz' }); + +// Or on a self-hosted model: the same description Codex uses. +const provider = model.endpoint(); +await selfHostedFleet.deploy({ ...plan, workerProgram: 'claude', model: provider.model, subagentModel: provider.model, modelProvider: provider }, + { image: immutableWorkerImage, bundlePath: '/absolute/private/flujo-factory.tgz', providerKey: await model.key() }); +``` + +- **Install:** the bootstrap installs one pinned Claude Code release, checked against + its SHA-256, plus `bubblewrap` and `socat` for its sandbox. A custom image must + provide `claude` on the path, and `/bin/bash`. +- **Sessions:** each agent slot keeps one session, with history in + `/data/claude/projects`. Each turn is one Claude Code process, so an idle agent uses + no memory beyond its files. A message during a turn joins that turn. A stop + interrupts it; if the program does not answer within 10 seconds its process is + ended. A stop receipt is issued only after the process has gone and the session + reads back as idle. +- **Plans:** a plan without `modelProvider` needs a broker made for its program, and a + provider must offer the program's API: `anthropic` for Claude Code, `responses` for + Codex. The models default to `opus` and `sonnet`; with a provider, name the served model. +- **Subscription:** the token from `claude setup-token` is long-lived, and this host + holds no refresh token for it. To rotate it, replace the file; when a worker reports a + refused token, `fleet.observe()` passes on the file's current token for the next turn. +- **Keys and agent commands:** the key or token reaches only the Claude Code process. + `CLAUDE_CODE_SHELL_PREFIX` removes it from the environment of agent commands and the + board MCP server; with a provider the GitHub token goes too, as with Codex. Journals + and traces do not contain it. With `danger-full-access`, an agent command runs as the + same user and could still read it from the parent process, so use a narrower sandbox + when agents handle untrusted input. Claude Code's own `CLAUDE_CODE_SUBPROCESS_ENV_SCRUB` + is not used, because it also forces interactive permission checks on these sessions. +- **Sandbox:** `danger-full-access` skips permission checks; the Machine is the + sandbox. `workspace-write` runs commands in Claude Code's bubblewrap sandbox, writing + only to the agent's folder and `writableRoots`, with no network, and its file tools + stay in the same folders. `read-only` also blocks writes to the agent's folder and + removes the file-writing tools. Both narrow modes refuse to run commands without + the sandbox, so the worker image must allow bubblewrap. Network tools, automatic + subagents, schedules, background tasks, skills, auto memory and repository settings + files are off in every mode. +- **Traces:** assistant text, thinking, tool calls and results become `item/completed` + records (`agentMessage`, `reasoning`, `toolCall`, `toolResult`), token use becomes + `thread/tokenUsage/updated`, and model retries become `model/retry`. Streaming deltas + are skipped unless `traceDeltas` is on. + #### Self-hosted models and operator-driven jobs To run the model yourself, `ModalSelfHostedModel` (CLI `factory-selfhost`) deploys a @@ -303,11 +363,11 @@ await fleet.command('m0-a1', { id: 'job-1', action: 'message', message: 'First j `streamMaxRetries`, `streamIdleTimeoutMs` and `requestTimeoutMs` are optional. `codexProviderConfig`, `codexProviderOverrides` and `claudeProviderEnvironment` are exported mappings; `providerPrograms(provider)` says which programs can use - it. Cloud lite workers run Codex. Each Codex gets a `config.toml` with the provider, the model's real context + it. Each Codex gets a `config.toml` with the provider, the model's real context window and its compaction point; web search, multi-agent, apps, plugins, goals, memories and skill instructions are off, and agents' shell commands inherit only the core environment. The key is a staged Fly app secret placed only in the - Codex process environment; agent commands, journals and traces do not get it. + worker program's process environment; agent commands, journals and traces do not get it. Machine *i* uses `baseUrls[i % n]`, so each machine keeps its prompt cache on one server. Codex offers no `apply_patch` tool to an unknown model, so ask agents to edit files with shell commands. @@ -320,7 +380,7 @@ await fleet.command('m0-a1', { id: 'job-1', action: 'message', message: 'First j turn, follow-up or wake; every turn comes from a durable `fleet.command(agent, { id, action: 'message' | 'stop' })`. The default, `'autonomous'`, keeps the existing goal-driven behavior. -- **`sandbox`**: Codex's sandbox for agent commands. The default stays +- **`sandbox`**: the sandbox for agent commands. The default stays `danger-full-access`. `workspace-write` lets commands write only the agent's folder plus the prepare hook's `writableRoots`, with no network; `read-only` is also accepted. diff --git a/package-lock.json b/package-lock.json index 2aa52bc..994859f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "flujo-factory", - "version": "0.5.0", + "version": "0.6.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "flujo-factory", - "version": "0.5.0", + "version": "0.6.0", "license": "MIT", "dependencies": { "@modelcontextprotocol/sdk": "1.32.1" diff --git a/package.json b/package.json index fec264c..7a87709 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "flujo-factory", - "version": "0.5.0", + "version": "0.6.0", "description": "Durable FLUJO swarm engine and Factory SDK, CLI, and MCP server", "license": "MIT", "repository": { diff --git a/scripts/verify-public-sdk.mjs b/scripts/verify-public-sdk.mjs index f1a4108..6d1090c 100644 --- a/scripts/verify-public-sdk.mjs +++ b/scripts/verify-public-sdk.mjs @@ -27,6 +27,7 @@ for (const name of [ 'FactoryArtifactStore', 'FactoryWorkflowSupervisor', 'captureGitSource', + 'ClaudeSubscriptionBroker', 'captureWorkspace', 'submitWorkspaceCandidate', 'ContinuousEvidenceCollector', diff --git a/selfhost/README.md b/selfhost/README.md index f8d9594..5735da3 100644 --- a/selfhost/README.md +++ b/selfhost/README.md @@ -3,7 +3,8 @@ FACTORY can serve an open-weight model on your own Modal GPUs with vLLM, so workers run on it instead of a vendor API: source code and prompts stay in your account. The served endpoint speaks the OpenAI Responses API (Codex), the Anthropic Messages API (Claude Code) and Chat -Completions, and plugs straight into `plan.modelProvider` and the local lite workers. +Completions, and plugs straight into `plan.modelProvider` and the local lite workers. Cloud lite workers can +run either program on it (`plan.workerProgram`). This folder is separate from `modal/`, which holds the earlier pilot definitions. diff --git a/selfhost/modal/app.py b/selfhost/modal/app.py index 2390726..53f99dc 100644 --- a/selfhost/modal/app.py +++ b/selfhost/modal/app.py @@ -83,39 +83,42 @@ def _start_vllm(): 'FACTORY_SELFHOST_INPUT_MODALITIES': ','.join(S.input_modalities)}) -GPU_FUNCTION = dict(image=image, gpu=S.gpu_spec, cpu=S.cpu, memory=S.memory_mib, secrets=[auth], - volumes=VOLUMES, scaledown_window=S.scaledown_minutes * MINUTES, - min_containers=0, max_containers=1, timeout=S.request_timeout_minutes * MINUTES) -# A web_server endpoint holds requests while a cold container starts. Modal's own auth stays -# off; vLLM checks the bearer key (VLLM_API_KEY). -WEB = dict(port=S.port, startup_timeout=S.startup_timeout_minutes * MINUTES) -CONCURRENT = dict(max_inputs=min(1000, S.max_num_seqs * 4)) # Modal's limit is 1000 +# The short-lived helper app has no serving functions: they need the key's secret, which only +# the deployment creates, after the weights are in place. +if not HELPERS: + GPU_FUNCTION = dict(image=image, gpu=S.gpu_spec, cpu=S.cpu, memory=S.memory_mib, secrets=[auth], + volumes=VOLUMES, scaledown_window=S.scaledown_minutes * MINUTES, + min_containers=0, max_containers=1, timeout=S.request_timeout_minutes * MINUTES) + # A web_server endpoint holds requests while a cold container starts. Modal's own auth stays + # off; vLLM checks the bearer key (VLLM_API_KEY). + WEB = dict(port=S.port, startup_timeout=S.startup_timeout_minutes * MINUTES) + CONCURRENT = dict(max_inputs=min(1000, S.max_num_seqs * 4)) # Modal's limit is 1000 -@app.function(**GPU_FUNCTION) -@modal.concurrent(**CONCURRENT) -@modal.web_server(**WEB) -def serve(): - _start_vllm() - - -if S.replicas >= 2: @app.function(**GPU_FUNCTION) @modal.concurrent(**CONCURRENT) @modal.web_server(**WEB) - def serve_b(): + def serve(): _start_vllm() -if S.replicas >= 3: - @app.function(**GPU_FUNCTION) - @modal.concurrent(**CONCURRENT) - @modal.web_server(**WEB) - def serve_c(): - _start_vllm() -if S.replicas >= 4: - @app.function(**GPU_FUNCTION) - @modal.concurrent(**CONCURRENT) - @modal.web_server(**WEB) - def serve_d(): - _start_vllm() + if S.replicas >= 2: + @app.function(**GPU_FUNCTION) + @modal.concurrent(**CONCURRENT) + @modal.web_server(**WEB) + def serve_b(): + _start_vllm() + + if S.replicas >= 3: + @app.function(**GPU_FUNCTION) + @modal.concurrent(**CONCURRENT) + @modal.web_server(**WEB) + def serve_c(): + _start_vllm() + + if S.replicas >= 4: + @app.function(**GPU_FUNCTION) + @modal.concurrent(**CONCURRENT) + @modal.web_server(**WEB) + def serve_d(): + _start_vllm() diff --git a/src/cloud-lite/board.mjs b/src/cloud-lite/board.mjs index bec7ce4..c48d3f8 100644 --- a/src/cloud-lite/board.mjs +++ b/src/cloud-lite/board.mjs @@ -84,7 +84,7 @@ export class CloudLiteBoard { threadStatus: receipt.thread_status, ...(receipt.turn_status ? { turnStatus: receipt.turn_status } : {}), eventSequence: receipt.event_seq, updated: receipt.updated } : null; } registerSession(agent, session) { - if (!/^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/.test(session ?? '')) throw new TypeError('Provide the real native Codex thread identity.'); + if (!/^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/.test(session ?? '')) throw new TypeError('Provide the real native thread identity.'); const previous = this.db.prepare('SELECT session FROM cloud_lite_agents WHERE id=?').get(agent); if (previous && previous.session !== session) { this.db.prepare('DELETE FROM cloud_lite_native_runtime WHERE agent_id=?').run(agent); diff --git a/src/cloud-lite/bootstrap.mjs b/src/cloud-lite/bootstrap.mjs index 41341c9..a9c1bfe 100644 --- a/src/cloud-lite/bootstrap.mjs +++ b/src/cloud-lite/bootstrap.mjs @@ -1,26 +1,45 @@ /** Bootstrap on the pinned upstream Node image when a custom registry is unavailable. */ -export function cloudLiteBootstrap(bundleSha256) { - if (!/^[a-f0-9]{64}$/.test(bundleSha256)) throw new TypeError('Bind the exact SDK bundle SHA256.'); - return `#!/bin/bash -set -euo pipefail -mkdir -p /data/factory /data/codex /opt/factory -exec > >(tee -a /data/factory/bootstrap.log) 2>&1 -echo '${bundleSha256} /tmp/factory-sdk.tgz' | sha256sum -c - -apt-get update -apt-get install -y --no-install-recommends git gh openssh-client ca-certificates curl procps python3 make g++ xz-utils bubblewrap -curl -fsSL https://github.com/openai/codex/releases/download/rust-v0.162.1/codex-x86_64-unknown-linux-musl.tar.gz -o /tmp/codex.tar.gz +const CLAUDE_CODE_VERSION = '2.1.311'; +const CLAUDE_CODE_LINUX_X64_SHA256 = '86dc1fcc69498159eaf15f446613e847010f9036bfbbdffd4c49350be8489ada'; +const programInstall = { + codex: { + folder: '/data/codex', packages: '', home: 'CODEX_HOME=/data/codex', + install: `curl -fsSL https://github.com/openai/codex/releases/download/rust-v0.162.1/codex-x86_64-unknown-linux-musl.tar.gz -o /tmp/codex.tar.gz echo '86f268d81b898f3e144c5ecff2ad2fda2c5802e042d6195b72538fe0fa125057 /tmp/codex.tar.gz' | sha256sum -c - tar -xzf /tmp/codex.tar.gz -C /usr/local/bin mv /usr/local/bin/codex-x86_64-unknown-linux-musl /usr/local/bin/codex curl -fsSL https://github.com/openai/codex/releases/download/rust-v0.162.1/codex-code-mode-host-x86_64-unknown-linux-musl.tar.gz -o /tmp/code-mode.tar.gz echo '458239ea45c1fc508d89695f180ab352f2834709a8a46929a4466f4f04399ada /tmp/code-mode.tar.gz' | sha256sum -c - tar -xzf /tmp/code-mode.tar.gz -C /usr/local/bin -mv /usr/local/bin/codex-code-mode-host-x86_64-unknown-linux-musl /usr/local/bin/codex-code-mode-host +mv /usr/local/bin/codex-code-mode-host-x86_64-unknown-linux-musl /usr/local/bin/codex-code-mode-host`, + }, + // socat carries Claude Code's sandbox network proxy. + claude: { + folder: '/data/claude', packages: ' socat', home: 'CLAUDE_CONFIG_DIR=/data/claude', + install: `curl -fsSL https://downloads.claude.ai/claude-code-releases/${CLAUDE_CODE_VERSION}/linux-x64/claude -o /tmp/claude +echo '${CLAUDE_CODE_LINUX_X64_SHA256} /tmp/claude' | sha256sum -c - +install -m 755 /tmp/claude /usr/local/bin/claude +rm -f /tmp/claude`, + }, +}; + +export function cloudLiteBootstrap(bundleSha256, { workerProgram = 'codex' } = {}) { + if (!/^[a-f0-9]{64}$/.test(bundleSha256)) throw new TypeError('Bind the exact SDK bundle SHA256.'); + const program = programInstall[workerProgram]; + if (!program) throw new TypeError('Unknown worker program.'); + return `#!/bin/bash +set -euo pipefail +mkdir -p /data/factory ${program.folder} /opt/factory +exec > >(tee -a /data/factory/bootstrap.log) 2>&1 +echo '${bundleSha256} /tmp/factory-sdk.tgz' | sha256sum -c - +apt-get update +apt-get install -y --no-install-recommends git gh openssh-client ca-certificates curl procps python3 make g++ xz-utils bubblewrap${program.packages} +${program.install} tar -xzf /tmp/factory-sdk.tgz --strip-components=1 -C /opt/factory cd /opt/factory npm install --omit=dev --ignore-scripts --no-audit --no-fund npm install --prefix /opt/flujo-bash --no-audit --no-fund @mario.andreschak/mcp-bash@3.46.2 -export HOME=/data CODEX_HOME=/data/codex +export HOME=/data ${program.home} node --input-type=module - <<'FACTORY_PIN_SOURCE' import { execFileSync } from 'node:child_process'; import { existsSync } from 'node:fs'; diff --git a/src/cloud-lite/claude-driver.mjs b/src/cloud-lite/claude-driver.mjs new file mode 100644 index 0000000..276a8bf --- /dev/null +++ b/src/cloud-lite/claude-driver.mjs @@ -0,0 +1,308 @@ +import { spawn } from 'node:child_process'; +import { EventEmitter } from 'node:events'; +import { createInterface } from 'node:readline'; +import { randomUUID } from 'node:crypto'; +import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { applyProviderEnvironment, claudeProviderEnvironment } from '../model-provider.mjs'; + +// Built-in tools a worker keeps. Automatic fanout, schedules, web access and the rest stay off, so +// every session has its own board identity and policy, as with Codex's disabled multi-agent feature. +const WRITE_TOOLS = ['Bash', 'Edit', 'Write', 'Read', 'NotebookEdit', 'Glob', 'Grep']; +const READ_TOOLS = ['Bash', 'Read', 'Glob', 'Grep']; +const EFFORTS = new Set(['low', 'medium', 'high', 'xhigh', 'max']); +const invalid = message => Object.assign(new Error(message), { code: -32600 }); + +/** + * The environment of one Claude Code process. A self-hosted provider gets its mapped settings and + * key, a subscription gets its token. `CLAUDE_CODE_SHELL_PREFIX` removes both from agent commands. + */ +export function claudeEnvironment({ plan, index = 0, home, commandWrapper, githubToken, deployKey, providerKey, oauthToken, path = process.env.PATH }) { + const base = { PATH: path, HOME: '/data', CLAUDE_CONFIG_DIR: home, GIT_TERMINAL_PROMPT: '0', + // The worker Machine is the disposable sandbox; Claude Code otherwise refuses full access as root. + IS_SANDBOX: '1', + ...(commandWrapper ? { CLAUDE_CODE_SHELL_PREFIX: commandWrapper } : {}), + ENABLE_TOOL_SEARCH: 'false', CLAUDE_CODE_DISABLE_AUTO_MEMORY: '1', CLAUDE_CODE_DISABLE_BACKGROUND_TASKS: '1', + CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: '1', DISABLE_TELEMETRY: '1', DISABLE_ERROR_REPORTING: '1', DISABLE_AUTOUPDATER: '1', + ...(githubToken && !deployKey ? { GH_TOKEN: githubToken } : {}) }; + if (plan.modelProvider) return applyProviderEnvironment(base, claudeProviderEnvironment(plan.modelProvider, { index, key: providerKey, model: plan.model })); + return { ...base, ...(oauthToken ? { CLAUDE_CODE_OAUTH_TOKEN: oauthToken } : {}) }; +} + +/** + * Claude Code runs agent commands and MCP servers as ` ""`. The wrapper drops + * the model credentials first, and with a provider also the GitHub token, like Codex's core policy. + */ +export function claudeCommandWrapper({ provider = false } = {}) { + const removed = ['ANTHROPIC_AUTH_TOKEN', 'ANTHROPIC_API_KEY', 'CLAUDE_CODE_OAUTH_TOKEN', ...(provider ? ['GH_TOKEN'] : [])]; + return `#!/bin/sh\nexec env ${removed.map(name => `-u ${name}`).join(' ')} /bin/bash -c "$1"\n`; +} + +/** + * Maps a Codex sandbox name to Claude Code. The two narrow modes rely on Claude Code's operating + * system sandbox (bubblewrap) for commands, with no network; file tools stay inside the same folders. + */ +export function claudeSandboxPolicy({ sandbox = 'danger-full-access', directory, writableRoots = [] }) { + if (sandbox === 'danger-full-access') return { tools: WRITE_TOOLS, args: ['--dangerously-skip-permissions'], allowed: [] }; + const base = { enabled: true, autoAllowBashIfSandboxed: true, allowUnsandboxedCommands: false, failIfUnavailable: true, network: { allowedDomains: [] } }; + if (sandbox === 'workspace-write') return { tools: WRITE_TOOLS, allowed: ['Bash'], + args: ['--permission-mode', 'acceptEdits', ...writableRoots.flatMap(root => ['--add-dir', root])], + settings: { sandbox: { ...base, filesystem: { allowWrite: writableRoots } } } }; + if (sandbox === 'read-only') return { tools: READ_TOOLS, allowed: READ_TOOLS, args: ['--permission-mode', 'dontAsk'], + settings: { sandbox: { ...base, filesystem: { denyWrite: [directory] } } } }; + throw new TypeError(`Unsupported sandbox ${sandbox}.`); +} + +/** The command line for one turn of one session. Nothing secret goes into arguments. */ +export function claudeSessionArguments({ threadId, resume, model, effort, instructions, policy, mcpConfigFile, settingsFile, boardTools, partialMessages }) { + const allowed = [...policy.allowed, ...(boardTools ? ['mcp__factory__*'] : [])]; + return ['-p', '--input-format', 'stream-json', '--output-format', 'stream-json', '--verbose', '--replay-user-messages', + '--strict-mcp-config', '--setting-sources', 'user', '--disable-slash-commands', '--permission-prompts', 'none', + '--tools', policy.tools.join(','), ...policy.args, + ...(allowed.length ? ['--allowedTools', allowed.join(',')] : []), + ...(settingsFile ? ['--settings', settingsFile] : []), + ...(mcpConfigFile ? ['--mcp-config', mcpConfigFile] : []), + ...(model ? ['--model', model] : []), + ...(EFFORTS.has(effort) ? ['--effort', effort] : []), + ...(instructions ? ['--append-system-prompt', instructions] : []), + ...(partialMessages ? ['--include-partial-messages'] : []), + ...(resume ? ['--resume', threadId] : ['--session-id', threadId])]; +} + +/** + * One Claude Code process per running turn, behind the method and notification contract the cloud + * runtime already uses for a Codex app-server: thread/start|resume|read and turn/start|steer|interrupt, + * with turn and item notifications. An idle session holds no process, so idle is simple to prove. + */ +export class ClaudeSessionDriver extends EventEmitter { + constructor({ command = 'claude', args = [], environment, configDirectory, stateDirectory, fixedModel, partialMessages = false, onStderr, + interruptGraceMs = 10000, killGraceMs = 3000, exitGraceMs = 5000, steerGraceMs = 30000 } = {}) { + super(); + Object.assign(this, { command, baseArgs: args, environment, configDirectory, stateDirectory, fixedModel, partialMessages, onStderr, interruptGraceMs, killGraceMs, exitGraceMs, steerGraceMs }); + this.threads = new Map(); + mkdirSync(stateDirectory, { recursive: true, mode: 0o700 }); + } + + async initialize() {} + + async request(method, params = {}) { + if (method === 'thread/start') return this.startThread(params); + if (method === 'thread/resume') return this.resumeThread(params); + if (method === 'thread/read') return this.readThread(params); + if (method === 'turn/start') return this.startTurn(params); + if (method === 'turn/steer') return this.steerTurn(params); + if (method === 'turn/interrupt') return this.interruptTurn(params); + throw invalid(`Unsupported method ${method}.`); + } + + notify(method, params) { this.emit('notification', { method, params }); } + stateFile(id) { return join(this.stateDirectory, `${id}.claude.json`); } + + transcriptExists(id) { + const projects = join(this.configDirectory, 'projects'); + if (!existsSync(projects)) return false; + return readdirSync(projects).some(name => existsSync(join(projects, name, `${id}.jsonl`))); + } + + /** Session files carry the board credential, so they are private and outside the arguments. */ + addThread(id, { cwd, config, sandbox, model, developerInstructions, lastTurn = null }) { + const server = config?.mcp_servers?.factory; + const directory = this.stateDirectory; + const policy = claudeSandboxPolicy({ sandbox, directory: cwd, writableRoots: config?.['sandbox_workspace_write.writable_roots'] ?? [] }); + let mcpConfigFile, settingsFile; + if (server) { + mcpConfigFile = join(directory, `${id}.claude-mcp.json`); + writeFileSync(mcpConfigFile, JSON.stringify({ mcpServers: { factory: { command: server.command, args: server.args, env: server.env } } }), { mode: 0o600 }); + } + if (policy.settings) { + settingsFile = join(directory, `${id}.claude-settings.json`); + writeFileSync(settingsFile, JSON.stringify(policy.settings), { mode: 0o600 }); + } + const thread = { id, cwd, policy, mcpConfigFile, settingsFile, boardTools: Boolean(server), model: this.fixedModel ?? model, instructions: developerInstructions, lastTurn, turn: null, child: null, totals: { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0 } }; + this.threads.set(id, thread); + this.save(thread); + return thread; + } + + save(thread) { + const file = this.stateFile(thread.id), next = `${file}.next`; + writeFileSync(next, JSON.stringify({ threadId: thread.id, lastTurn: thread.lastTurn }), { mode: 0o600 }); + renameSync(next, file); + } + + describe(thread, includeTurns) { + return { id: thread.id, status: { type: thread.child ? 'active' : 'idle' }, ...(includeTurns ? { turns: thread.lastTurn ? [thread.lastTurn] : [] } : {}) }; + } + + thread(id) { + const thread = this.threads.get(id); + if (!thread) throw invalid(`Unknown thread ${id}.`); + return thread; + } + + startThread(params) { + const thread = this.addThread(randomUUID(), params); + this.notify('thread/started', { threadId: thread.id, thread: { id: thread.id } }); + return { thread: this.describe(thread) }; + } + + resumeThread(params) { + const id = params.threadId; + const unavailable = () => invalid(`no rollout found for thread id ${id}`); + if (!existsSync(this.stateFile(id))) throw unavailable(); + const { lastTurn } = JSON.parse(readFileSync(this.stateFile(id), 'utf8')); + // A session without a turn has no history to lose; one with a turn must still have its history. + if (lastTurn && !this.transcriptExists(id)) throw unavailable(); + // Any process died with the old Machine, so a turn left in progress did not finish. + const recovered = lastTurn?.status === 'inProgress' ? { ...lastTurn, status: 'interrupted' } : lastTurn; + const thread = this.addThread(id, { ...params, lastTurn: recovered }); + this.notify('thread/resumed', { threadId: id, thread: { id } }); + return { thread: this.describe(thread) }; + } + + readThread({ threadId, includeTurns }) { + return { thread: this.describe(this.thread(threadId), includeTurns) }; + } + + startTurn({ threadId, input, effort }) { + const thread = this.thread(threadId); + if (thread.child) throw invalid('This thread already has a running turn.'); + const turn = { id: randomUUID(), written: 0, replayed: 0 }; + const args = [...this.baseArgs, ...claudeSessionArguments({ threadId, resume: this.transcriptExists(threadId), model: thread.model, effort, instructions: thread.instructions, + policy: thread.policy, mcpConfigFile: thread.mcpConfigFile, settingsFile: thread.settingsFile, boardTools: thread.boardTools, partialMessages: this.partialMessages })]; + thread.lastTurn = { id: turn.id, status: 'inProgress' }; + this.save(thread); + const child = spawn(this.command, args, { cwd: thread.cwd, env: this.environment(), stdio: ['pipe', 'pipe', 'pipe'], detached: process.platform !== 'win32', windowsHide: true }); + thread.child = child; thread.turn = turn; + child.stdin.on('error', () => {}); + child.stderr.on('data', chunk => this.onStderr?.(chunk.toString())); + createInterface({ input: child.stdout }).on('line', line => this.receive(thread, turn, line)); + child.on('error', () => this.finish(thread, turn)); + // A command the agent left running can hold the output pipes open after the program itself ended. + child.on('exit', () => { turn.exitTimer = setTimeout(() => this.finish(thread, turn), 1000); }); + child.on('close', () => this.finish(thread, turn)); + this.notify('thread/status/changed', { threadId, status: { type: 'active' } }); + this.notify('turn/started', { threadId, turn: { id: turn.id, status: 'inProgress' } }); + this.write(thread, turn, input); + return { turn: { id: turn.id, status: 'inProgress' } }; + } + + write(thread, turn, input) { + turn.written++; + const content = (input ?? []).filter(item => item.type === 'text').map(item => ({ type: 'text', text: item.text })); + thread.child.stdin.write(JSON.stringify({ type: 'user', session_id: thread.id, parent_tool_use_id: null, message: { role: 'user', content } }) + '\n'); + } + + /** Claude Code takes a message written during a turn into that turn, or runs it straight after. */ + steerTurn({ threadId, expectedTurnId, input }) { + const thread = this.thread(threadId), turn = thread.turn; + if (!turn || turn.closing || turn.interrupting) throw new Error('no active turn to steer'); + if (expectedTurnId && expectedTurnId !== turn.id) throw invalid('The expected turn is not the running turn.'); + this.write(thread, turn, input); + return { turnId: turn.id }; + } + + interruptTurn({ threadId, turnId }) { + const thread = this.thread(threadId), turn = thread.turn; + if (!turn || turn.id !== turnId) { + if (thread.lastTurn?.id === turnId) return {}; + throw invalid('The expected turn is not the running turn.'); + } + if (turn.closing || turn.interrupting) return {}; + turn.interrupting = true; + clearTimeout(turn.steerTimer); + thread.child.stdin.write(JSON.stringify({ type: 'control_request', request_id: randomUUID(), request: { subtype: 'interrupt' } }) + '\n'); + // A session that does not answer is ended outright: a stopped process is the strongest proof. + turn.interruptTimer = setTimeout(() => this.kill(thread, turn), this.interruptGraceMs); + return {}; + } + + receive(thread, turn, line) { + let value; + try { value = JSON.parse(line); } catch { this.emit('protocolError', { bytes: line.length }); return; } + const threadId = thread.id, turnId = turn.id; + const item = body => this.notify('item/completed', { threadId, turnId, item: body }); + if (value.type === 'system' && value.subtype === 'init') { + // The same promise Codex gets from a required MCP server: a worker never runs without its board. + const boardFailed = thread.boardTools && value.mcp_servers?.some(server => server.name === 'factory' && server.status === 'failed'); + if (value.session_id !== threadId || boardFailed) { + this.onStderr?.(boardFailed ? 'The board MCP server did not connect; ending the turn.' : 'Claude Code opened a different session than requested; ending the turn.'); + turn.outcome = { status: 'failed' }; this.kill(thread, turn); + } + } else if (value.type === 'system' && value.subtype === 'api_retry') { + this.notify('model/retry', { threadId, turnId, attempt: value.attempt, status: value.error_status, error: value.error }); + if (value.error === 'authentication_failed') this.emit('authFailed', { threadId }); + } else if (value.type === 'assistant') { + for (const block of value.message?.content ?? []) { + if (block.type === 'text') item({ type: 'agentMessage', text: block.text }); + else if (block.type === 'thinking') item({ type: 'reasoning', text: block.thinking }); + else if (block.type === 'tool_use') item({ type: 'toolCall', id: block.id, tool: block.name, arguments: block.input }); + } + } else if (value.type === 'user') { + if (value.isReplay) { turn.replayed++; return; } + for (const block of Array.isArray(value.message?.content) ? value.message.content : []) { + if (block.type === 'tool_result') item({ type: 'toolResult', id: block.tool_use_id, isError: block.is_error === true, content: block.content }); + } + } else if (value.type === 'stream_event') { + this.notify('item/stream/delta', { threadId, turnId, event: value.event }); + } else if (value.type === 'result') this.receiveResult(thread, turn, value); + } + + receiveResult(thread, turn, value) { + const usage = value.usage ?? {}; + const last = { inputTokens: (usage.input_tokens ?? 0) + (usage.cache_read_input_tokens ?? 0) + (usage.cache_creation_input_tokens ?? 0), cachedInputTokens: usage.cache_read_input_tokens ?? 0, outputTokens: usage.output_tokens ?? 0 }; + for (const name of Object.keys(last)) thread.totals[name] += last[name]; + this.notify('thread/tokenUsage/updated', { threadId: thread.id, turnId: turn.id, tokenUsage: { last, total: { ...thread.totals } } }); + if (turn.interrupting) { + // Stop wins over a queued message, which would otherwise start running now. + clearTimeout(turn.interruptTimer); + turn.outcome = { status: 'interrupted' }; + this.kill(thread, turn); + return; + } + turn.outcome = value.is_error ? { status: 'failed', ...(Number.isSafeInteger(value.api_error_status) ? { code: value.api_error_status } : {}) } : { status: 'completed' }; + clearTimeout(turn.steerTimer); + // A message that has not come back yet is still queued, so this turn goes on with it. + // Without more input the process ends by itself once anything it did take has run. + if (turn.replayed < turn.written) turn.steerTimer = setTimeout(() => { turn.closing = true; thread.child?.stdin.end(); }, this.steerGraceMs); + else this.closeInput(thread, turn); + } + + closeInput(thread, turn) { + turn.closing = true; + thread.child?.stdin.end(); + turn.exitTimer2 = setTimeout(() => this.kill(thread, turn), this.exitGraceMs); + } + + signal(child, name) { + // Claude Code leads its own process group, so commands it started stop with it. + try { if (child.pid && process.platform !== 'win32') { process.kill(-child.pid, name); return; } } catch { /* Falls back to the process itself. */ } + try { child.kill(name); } catch { /* Already gone. */ } + } + + kill(thread, turn) { + if (thread.turn !== turn || turn.killing) return; + turn.killing = true; + const child = thread.child; + child.stdin.end(); + this.signal(child, 'SIGTERM'); + turn.killTimer = setTimeout(() => this.signal(child, 'SIGKILL'), this.killGraceMs); + } + + finish(thread, turn) { + if (turn.finished) return; + turn.finished = true; + for (const timer of ['steerTimer', 'interruptTimer', 'exitTimer', 'exitTimer2', 'killTimer']) clearTimeout(turn[timer]); + const { status, code } = turn.outcome ?? { status: turn.interrupting ? 'interrupted' : 'failed' }; + thread.child = null; thread.turn = null; thread.lastTurn = { id: turn.id, status }; + this.save(thread); + this.notify('turn/completed', { threadId: thread.id, turn: { id: turn.id, status, ...(status === 'completed' ? {} : { error: code === undefined ? {} : { code } }) } }); + this.notify('thread/status/changed', { threadId: thread.id, status: { type: 'idle' } }); + if (this.closed && ![...this.threads.values()].some(other => other.child)) this.emit('exit', { code: 0, signal: null }); + } + + close() { + this.closed = true; + for (const thread of this.threads.values()) if (thread.child) this.kill(thread, thread.turn); + } +} diff --git a/src/cloud-lite/evidence-worker.mjs b/src/cloud-lite/evidence-worker.mjs index c908238..26fe2ae 100644 --- a/src/cloud-lite/evidence-worker.mjs +++ b/src/cloud-lite/evidence-worker.mjs @@ -14,7 +14,7 @@ process.on('message',async message=>{ sessions=message.sessions; const directory='/data/factory-evidence';mkdirSync(directory,{recursive:true,mode:0o700}); const roots=[{name:'workspaces',path:'/data/worktrees',exclude:['node_modules','.git','.next','.cache','.npm','__pycache__']}, - {name:'sessions',path:'/data/codex/sessions'}, {name:'results',path:'/data/results'}, + {name:'sessions',path:message.sessionsPath??'/data/codex/sessions'}, {name:'results',path:'/data/results'}, {name:'runtime',path:'/data/factory',exclude:['control.sqlite','control.sqlite-wal','control.sqlite-shm']}, {name:'board',path:resolve(directory,'board')}]; for(const root of roots)mkdirSync(root.path,{recursive:true,mode:0o700}); diff --git a/src/cloud-lite/fleet.mjs b/src/cloud-lite/fleet.mjs index cda23cb..99ffc95 100644 --- a/src/cloud-lite/fleet.mjs +++ b/src/cloud-lite/fleet.mjs @@ -111,6 +111,7 @@ function persistObservation(file, value) { /** Only this native host owns managed refresh; remote harnesses receive external access tokens. */ export class CodexSubscriptionBroker { + program = 'codex'; constructor({ authFile, codexCommand = 'codex' } = {}) { if (!isAbsolute(authFile ?? '')) throw new TypeError('Select the existing absolute Codex subscription auth file.'); this.authFile = authFile; this.command = codexCommand; @@ -134,6 +135,25 @@ export class CodexSubscriptionBroker { } } +/** + * A Claude subscription token from `claude setup-token`, kept in a private file on this host. It is + * long-lived and has no refresh token to protect; replacing the file rotates it for running workers. + */ +export class ClaudeSubscriptionBroker { + program = 'claude'; + constructor({ tokenFile } = {}) { + if (!isAbsolute(tokenFile ?? '')) throw new TypeError('Select the absolute private Claude subscription token file.'); + this.tokenFile = tokenFile; + } + read() { + const token = readFileSync(this.tokenFile, 'utf8').trim(); + if (token.length < 20 || /\s/.test(token)) throw new Error('The token file must hold one token from claude setup-token.'); + if (token.startsWith('sk-ant-api')) throw new Error('A Claude subscription token is required; API-key fallback is disabled.'); + return { accessToken: token }; + } + async refresh() { return this.read(); } +} + export class FactoryCloudLiteFleet { constructor({ directory, flujoCloudPath = process.env.FACTORY_FLUJO_CLOUD_PATH, infrastructure, flyCommand = 'flyctl', flyEnv = process.env, proxyEnv = flyEnv, execute, auth, onEvent = () => {} } = {}) { if (!isAbsolute(directory ?? '')) throw new TypeError('An absolute private fleet directory is required.'); @@ -195,6 +215,7 @@ export class FactoryCloudLiteFleet { const providerMode = Boolean(input?.modelProvider); if (input.repository && !githubToken && !deployKey || !providerMode && !this.auth) throw new TypeError('Select a subscription auth broker (or a model provider) and, for Git tasks, a GitHub credential or deploy key.'); if (providerMode && (typeof providerKey !== 'string' || providerKey.length < 8)) throw new TypeError('A model provider needs its API key.'); + if (!providerMode && this.auth.program && this.auth.program !== (input.workerProgram ?? 'codex')) throw new TypeError(`A ${this.auth.program} subscription broker cannot serve ${input.workerProgram ?? 'codex'} workers.`); if (deployKey !== undefined && (!input.repository || typeof deployKey !== 'string' || !/-----BEGIN [A-Z ]*PRIVATE KEY-----/.test(deployKey))) throw new TypeError('A deploy key must be a private SSH key for the plan repository.'); if (imageSource) image = (await (await this.upstream()).resolveWorkerImage({ source: imageSource, image, channel })).image; if (!image) throw new TypeError('Select an immutable worker image or its upstream FLUJO compatibility advertisement.'); @@ -220,7 +241,7 @@ export class FactoryCloudLiteFleet { const bundleDigest = hash(readFileSync(bundlePath)); if (this.state.bundleSha256 && this.state.bundleSha256 !== bundleDigest) throw new Error('Existing fleet bundle differs.'); this.state.bundleSha256 = bundleDigest; this.save(); - bootstrapPath = join(this.directory, 'bootstrap.sh'); writeFileSync(bootstrapPath, cloudLiteBootstrap(bundleDigest), { mode: 0o600 }); + bootstrapPath = join(this.directory, 'bootstrap.sh'); writeFileSync(bootstrapPath, cloudLiteBootstrap(bundleDigest, { workerProgram: plan.workerProgram }), { mode: 0o600 }); } const { credentials, operatorToken } = JSON.parse(readFileSync(join(this.directory, 'credentials.json'))); const initialAuth = providerMode ? {} : this.auth.read(); this.knownSecrets = [githubToken, initialAuth.accessToken, operatorToken, deployKey, providerKey, ...Object.values(credentials),...Object.values(evidenceTokens??{})]; diff --git a/src/cloud-lite/protocol.mjs b/src/cloud-lite/protocol.mjs index f79f705..b222fcc 100644 --- a/src/cloud-lite/protocol.mjs +++ b/src/cloud-lite/protocol.mjs @@ -1,5 +1,5 @@ import { createHash, randomBytes } from 'node:crypto'; -import { validateModelProvider, providerApis } from '../model-provider.mjs'; +import { validateModelProvider, providerApis, WORKER_PROGRAM_APIS } from '../model-provider.mjs'; export { validateModelProvider }; @@ -8,7 +8,8 @@ export const hash = value => createHash('sha256').update(typeof value === 'strin export function validateCloudLitePlan(input) { if (!input || typeof input !== 'object') throw new TypeError('A cloud lite-worker plan is required.'); const plan = { cpuKind: 'shared', cpus: 2, memoryMb: 4096, diskGb: 20, machines: 1, - agentsPerMachine: 10, region: 'iad', model: 'gpt-6.1-sol', subagentModel: 'gpt-6-luna', reasoningEffort: 'medium', ...input }; + agentsPerMachine: 10, region: 'iad', ...(input.workerProgram === 'claude' ? { model: 'opus', subagentModel: 'sonnet' } : { model: 'gpt-6.1-sol', subagentModel: 'gpt-6-luna' }), reasoningEffort: 'medium', ...input }; + if (plan.workerProgram !== undefined && !Object.hasOwn(WORKER_PROGRAM_APIS, plan.workerProgram)) throw new TypeError(`workerProgram must be one of ${Object.keys(WORKER_PROGRAM_APIS).join(', ')}.`); for (const name of ['runId', 'app']) if (!/^[a-z][a-z0-9-]{2,62}$/.test(plan[name] ?? '')) throw new TypeError(`Invalid ${name}.`); if (!['shared', 'performance'].includes(plan.cpuKind)) throw new TypeError('Invalid cpuKind.'); for (const name of ['cpus', 'memoryMb', 'diskGb', 'machines', 'agentsPerMachine']) { @@ -31,8 +32,10 @@ export function validateCloudLitePlan(input) { } if (plan.modelProvider !== undefined) { validateModelProvider(plan.modelProvider); - // Cloud lite workers run Codex, which talks to custom providers through the Responses API. - if (!providerApis(plan.modelProvider).includes('responses')) throw new TypeError('Cloud lite workers need a provider that offers the responses API.'); + // Codex talks to custom providers through the Responses API, Claude Code through the Messages API. + const needed = WORKER_PROGRAM_APIS[plan.workerProgram ?? 'codex']; + if (!providerApis(plan.modelProvider).includes(needed)) throw new TypeError(`Cloud lite workers need a provider that offers the ${needed} API.`); + if (plan.workerProgram === 'claude' && !plan.modelProvider.model && !input.model) throw new TypeError('Select the served model name.'); } if (plan.supervision !== undefined && !['autonomous', 'operator'].includes(plan.supervision)) throw new TypeError('supervision must be autonomous or operator.'); if (plan.sandbox !== undefined && !CODEX_SANDBOXES.includes(plan.sandbox)) throw new TypeError(`sandbox must be one of ${CODEX_SANDBOXES.join(', ')}.`); @@ -72,7 +75,7 @@ export function machineConfig(plan, { image, volumeId, index, environment }) { image, env: environment, guest: { cpu_kind: plan.cpuKind, cpus: plan.cpus, memory_mb: plan.memoryMb }, mounts: [{ volume: volumeId, path: '/data' }], - metadata: { 'factory.run': plan.runId, 'factory.machine': `m${index}`, 'factory.runtime': 'codex-lite' }, + metadata: { 'factory.run': plan.runId, 'factory.machine': `m${index}`, 'factory.runtime': plan.workerProgram === 'claude' ? 'claude-lite' : 'codex-lite' }, restart: { policy: 'no' }, // Same private topology as flujo-cloud: host connects using its owned proxy; // remote Codex sessions use the coordinator's exact Fly private Machine DNS. diff --git a/src/cloud-lite/runtime.mjs b/src/cloud-lite/runtime.mjs index 2b56b75..7c9e0f3 100644 --- a/src/cloud-lite/runtime.mjs +++ b/src/cloud-lite/runtime.mjs @@ -6,6 +6,7 @@ import { execFile, fork } from 'node:child_process'; import { promisify } from 'node:util'; import { freemem, totalmem, loadavg } from 'node:os'; import { CodexRpc } from './rpc.mjs'; +import { ClaudeSessionDriver, claudeCommandWrapper, claudeEnvironment } from './claude-driver.mjs'; import { startCloudLiteCoordinator } from './server.mjs'; import { CloudLiteTrace } from './trace.mjs'; import { NativeOperatorStop } from './operator-stop.mjs'; @@ -44,6 +45,8 @@ export async function runCloudLiteMachine(config) { // Optional: a self-hosted model provider instead of the subscription login, read-only // deploy-key repository access, a prepare hook for agent workspaces, operator-driven turns. const providerMode = Boolean(plan.modelProvider); + const claudeProgram = plan.workerProgram === 'claude'; + const evidenceSessionsPath = claudeProgram ? '/data/claude/projects' : '/data/codex/sessions'; const operatorSupervised = plan.supervision === 'operator'; const extraSecrets = [providerKey, deployKey].filter(value => typeof value === 'string' && value.length >= 8); mkdirSync('/data/factory', { recursive: true, mode: 0o700 }); @@ -57,7 +60,7 @@ export async function runCloudLiteMachine(config) { if(!retainedEvidence.receipt?.checkpoint.final){ const child=fork(fileURLToPath(new URL('./evidence-worker.mjs',import.meta.url)),[],{stdio:['ignore','ignore','ignore','ipc'],windowsHide:true}); child.on('error',()=>{}); - child.send({action:'start',runId:plan.runId,machineId:`m${index}`,url:plan.collection.url,token:evidenceTokens?.[`m${index}`],intervalMs:plan.collection.intervalMs??60000,sessions:sealedCheckpoint.sessions}); + child.send({action:'start',runId:plan.runId,machineId:`m${index}`,url:plan.collection.url,token:evidenceTokens?.[`m${index}`],intervalMs:plan.collection.intervalMs??60000,sessions:sealedCheckpoint.sessions,sessionsPath:evidenceSessionsPath}); child.send({action:'finalize',sessions:sealedCheckpoint.sessions}); } return {sealed:true,stop:async()=>coordinator?.close()}; @@ -89,7 +92,7 @@ export async function runCloudLiteMachine(config) { return{agentId:agent.id,threadId:receipt.threadId,stopId:receipt.commandId}; }); const child=fork(fileURLToPath(new URL('./evidence-worker.mjs',import.meta.url)),[],{stdio:['ignore','ignore','ignore','ipc'],windowsHide:true});child.on('error',()=>{}); - child.send({action:'start',runId:plan.runId,machineId:machine,url:plan.collection.url,token:evidenceTokens?.[machine],intervalMs:plan.collection.intervalMs??60000,sessions}); + child.send({action:'start',runId:plan.runId,machineId:machine,url:plan.collection.url,token:evidenceTokens?.[machine],intervalMs:plan.collection.intervalMs??60000,sessions,sessionsPath:evidenceSessionsPath}); child.send({action:'finalize',sessions});return{sealed:true,stop:async()=>coordinator?.close()}; } if(plan.shutdown && Date.now()>=Date.parse(plan.shutdown.deadlineUtc))throw new Error('Work deadline has passed; preserve existing evidence and reconcile shutdown without restarting agents.'); @@ -133,9 +136,9 @@ export async function runCloudLiteMachine(config) { prepared = await runPrepareHook({ plan, index, repositoryDirectory, env: gitEnv }); writeFileSync(preparedFile, JSON.stringify(prepared), { mode: 0o600 }); } - const home = '/data/codex'; mkdirSync(home, { recursive: true, mode: 0o700 }); + const home = claudeProgram ? '/data/claude' : '/data/codex'; mkdirSync(home, { recursive: true, mode: 0o700 }); // A fresh native harness has only the explicitly configured FACTORY MCP server. - writeFileSync(join(home, 'config.toml'), providerMode ? providerConfigToml(plan, index) + if (!claudeProgram) writeFileSync(join(home, 'config.toml'), providerMode ? providerConfigToml(plan, index) : 'cli_auth_credentials_store="ephemeral"\nforced_login_method="chatgpt"\n', { mode: 0o600 }); const traces = new Map(); const workflowWakes = new Map(); @@ -145,18 +148,27 @@ export async function runCloudLiteMachine(config) { secrets: [githubToken, initialAuth.accessToken, operatorToken, ...extraSecrets, ...Object.values(credentials),...Object.values(evidenceTokens??{})] })); } const rootTrace = traces.get(rootId); const threadAgents = new Map(); const states = new Map(); const turns = new Map(); const terminalTurns = new Map(); const inboxVersions = new Map(); const workflowVersions = new Map(); const consumedVersions = new Map(); const stoppedThreads = new Set(); const stoppingThreads = new Set(); const revokedStopCommands = new Set(); const pendingStopCommands = new Map(); const stopControllers = new Map(); const stopTasks = new Map(); let stopping = false; let stopPromise; let timer; let monitoring; let liveAuth = bootstrap.auth; - const rpc = new CodexRpc({ env: codexEnvironment({ plan, home, githubToken, deployKey, providerKey }), cwd: repositoryDirectory, - onStderr: text => rootTrace.record('codex.stderr', { text }), - onRequest: async (method, params) => { - if (method === 'account/chatgptAuthTokens/refresh' && !providerMode) { - const auth = await request('/auth', { refresh: true, previousTokenHash: hash(liveAuth.accessToken) }); - if (auth.chatgptAccountId !== initialAuth.chatgptAccountId) throw new Error('Subscription identity changed.'); - for (const trace of traces.values()) trace.secrets.push(auth.accessToken); - liveAuth = auth; - return auth; - } - throw new Error(`Unexpected Codex host request: ${method}`); - } }); + const refreshAuth = async () => { + const auth = await request('/auth', { refresh: true, previousTokenHash: hash(liveAuth.accessToken) }); + if (auth.chatgptAccountId !== initialAuth.chatgptAccountId) throw new Error('Subscription identity changed.'); + for (const trace of traces.values()) trace.secrets.push(auth.accessToken); + liveAuth = auth; + return auth; + }; + const commandWrapper = '/data/factory/claude-command-env.sh'; + if (claudeProgram) writeFileSync(commandWrapper, claudeCommandWrapper({ provider: providerMode }), { mode: 0o700 }); + const rpc = claudeProgram + ? new ClaudeSessionDriver({ configDirectory: home, stateDirectory: '/data/factory', fixedModel: plan.modelProvider?.model, partialMessages: plan.traceDeltas === true, + // Read for every turn, so a refreshed subscription token reaches the next process. + environment: () => claudeEnvironment({ plan, index, home, commandWrapper, githubToken, deployKey, providerKey, oauthToken: liveAuth?.accessToken }), + onStderr: text => rootTrace.record('claude.stderr', { text }) }) + : new CodexRpc({ env: codexEnvironment({ plan, home, githubToken, deployKey, providerKey }), cwd: repositoryDirectory, + onStderr: text => rootTrace.record('codex.stderr', { text }), + onRequest: async (method, params) => { + if (method === 'account/chatgptAuthTokens/refresh' && !providerMode) return refreshAuth(); + throw new Error(`Unexpected Codex host request: ${method}`); + } }); + if (claudeProgram && !providerMode) rpc.on('authFailed', () => { void refreshAuth().catch(error => rootTrace.record('auth.refresh_pending', { message: error.message })); }); rpc.on('notification', value => { const threadId = value.params?.threadId ?? value.params?.thread?.id; const agent = threadAgents.get(threadId) ?? rootId; @@ -197,8 +209,8 @@ export async function runCloudLiteMachine(config) { }, 1000); await rpc.initialize(); // A self-hosted provider authenticates with its key; no subscription login is involved. - if (!providerMode) await rpc.request('account/login/start', { type: 'chatgptAuthTokens', ...bootstrap.auth }); - rootTrace.record('machine.ready', { machine, node: process.version, codex: await run('codex', ['--version']), configuration: { cpuKind: plan.cpuKind, cpus: plan.cpus, memoryMb: plan.memoryMb, diskGb: plan.diskGb }, agents: bootstrap.agents.map(x => ({ id: x.id, role: x.role })) }); + if (!providerMode && !claudeProgram) await rpc.request('account/login/start', { type: 'chatgptAuthTokens', ...bootstrap.auth }); + rootTrace.record('machine.ready', { machine, node: process.version, ...(claudeProgram ? { claude: await run('claude', ['--version']) } : { codex: await run('codex', ['--version']) }), configuration: { cpuKind: plan.cpuKind, cpus: plan.cpus, memoryMb: plan.memoryMb, diskGb: plan.diskGb }, agents: bootstrap.agents.map(x => ({ id: x.id, role: x.role })) }); const proxy = fileURLToPath(new URL('../../bin/factory-cloud-mcp.mjs', import.meta.url)); for (const agent of bootstrap.agents) { // With a prepare hook, each agent works in the folder the hook prepared for its slot. @@ -216,14 +228,16 @@ export async function runCloudLiteMachine(config) { if (recovering) { const retained = JSON.parse(readFileSync(sessionFile)); try { - result = await rpc.request('thread/resume', { threadId: retained.threadId, cwd: directory, config: configOverride, approvalPolicy: 'never', sandbox }); + result = await rpc.request('thread/resume', { threadId: retained.threadId, cwd: directory, config: configOverride, approvalPolicy: 'never', sandbox, + // Claude Code starts a new process per turn, so it needs these again after a restart. + ...(claudeProgram ? { model: agent.role === 'orchestrator' ? plan.model : plan.subagentModel, developerInstructions: instructions } : {}) }); } catch (error) { if (error.code !== -32600 || error.message !== `no rollout found for thread id ${retained.threadId}`) throw error; unavailableSession = retained.threadId; recovering = false; const preserved = `${sessionFile}.${hash(readFileSync(sessionFile))}.unavailable`; try { copyFileSync(sessionFile, preserved, constants.COPYFILE_EXCL); } catch (copyError) { if (copyError.code !== 'EEXIST') throw copyError; } traces.get(agent.id).record('agent.session_unavailable', { previousThreadId: unavailableSession, preserved, - reason: 'Upstream Codex confirmed that the prepared thread has no persisted rollout. Original board claims and repository state remain; reconcile before continuing.' }); + reason: 'The worker program confirmed that the prepared thread has no persisted history. Original board claims and repository state remain; reconcile before continuing.' }); } } if (!result) { @@ -267,7 +281,7 @@ export async function runCloudLiteMachine(config) { if(message.type==='evidence.final')evidenceFinal=true; writeFileSync('/data/factory-evidence-status.json',JSON.stringify({at:new Date().toISOString(),...message}),{mode:0o600}); }); - evidenceProcess.send({action:'start',runId:plan.runId,machineId:machine,url:plan.collection.url,token:evidenceTokens?.[machine],intervalMs:plan.collection.intervalMs??60000,sessions:evidenceSessions()}); + evidenceProcess.send({action:'start',runId:plan.runId,machineId:machine,url:plan.collection.url,token:evidenceTokens?.[machine],intervalMs:plan.collection.intervalMs??60000,sessions:evidenceSessions(),sessionsPath:evidenceSessionsPath}); }; launchCollector(); monitoring = setInterval(async () => { diff --git a/src/public-sdk.mjs b/src/public-sdk.mjs index cc052a8..fd0c5e1 100644 --- a/src/public-sdk.mjs +++ b/src/public-sdk.mjs @@ -99,7 +99,7 @@ export { createObservatoryClient } from './observatory-client.mjs'; export { liteWorkerCommand, startLiteWorker } from './lite-worker.mjs'; export { ModalSelfHostedModel, selfHostedProfiles, probeSelfHostedModel, modalResult, readBoundedJson, modalEndpointOrigin, runCommand, SELFHOST_MODAL_DIR, SELFHOSTED_KEY_ENV } from './selfhost/modal.mjs'; export { MODEL_PROVIDER_APIS, WORKER_PROGRAM_APIS, validateModelProvider, providerPrograms, codexProviderConfig, codexProviderOverrides, claudeProviderEnvironment } from './model-provider.mjs'; -export { FactoryCloudLiteFleet, CodexSubscriptionBroker } from './cloud-lite/fleet.mjs'; +export { FactoryCloudLiteFleet, CodexSubscriptionBroker, ClaudeSubscriptionBroker } from './cloud-lite/fleet.mjs'; export { validateCloudLitePlan, agentInventory, machineConfig as cloudLiteMachineConfig } from './cloud-lite/protocol.mjs'; export { cloudLiteBootstrap } from './cloud-lite/bootstrap.mjs'; export * from './workflows/index.mjs'; diff --git a/src/selfhost/modal.mjs b/src/selfhost/modal.mjs index 8b24d6c..bdd369e 100644 --- a/src/selfhost/modal.mjs +++ b/src/selfhost/modal.mjs @@ -209,6 +209,9 @@ export class ModalSelfHostedModel { const source = join(this.directory, 'source'); mkdirSync(source, { recursive: true, mode: 0o700 }); for (const name of SOURCES) copyFileSync(join(SELFHOST_MODAL_DIR, name), join(source, name)); + // The settings read their profile next to themselves. + mkdirSync(join(source, 'profiles'), { recursive: true, mode: 0o700 }); + for (const name of readdirSync(join(SELFHOST_MODAL_DIR, 'profiles'))) copyFileSync(join(SELFHOST_MODAL_DIR, 'profiles', name), join(source, 'profiles', name)); const result = await this.commandRunner(this.python, [...this.pythonArgs, '-B', join(source, 'deploy.py')], { signal, timeoutMs, input: JSON.stringify(request), maxBytes: 1_000_000, env: this.#env(), killTree: true, cwd: source }); const answer = lastJsonLine(result?.stdout); diff --git a/test/cloud-lite-claude.test.mjs b/test/cloud-lite-claude.test.mjs new file mode 100644 index 0000000..345af9d --- /dev/null +++ b/test/cloud-lite-claude.test.mjs @@ -0,0 +1,418 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { existsSync, readFileSync, statSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { validateCloudLitePlan, agentInventory, machineConfig, secret } from '../src/cloud-lite/protocol.mjs'; +import { cloudLiteBootstrap } from '../src/cloud-lite/bootstrap.mjs'; +import { ClaudeSessionDriver, claudeCommandWrapper, claudeEnvironment, claudeSandboxPolicy, claudeSessionArguments } from '../src/cloud-lite/claude-driver.mjs'; +import { ClaudeSubscriptionBroker, CodexSubscriptionBroker, FactoryCloudLiteFleet } from '../src/cloud-lite/fleet.mjs'; +import { NativeOperatorStop } from '../src/cloud-lite/operator-stop.mjs'; +import { NativeWorkflowWake } from '../src/cloud-lite/workflow-wake.mjs'; +import { CloudLiteTrace } from '../src/cloud-lite/trace.mjs'; +import { startCloudLiteCoordinator } from '../src/cloud-lite/server.mjs'; +import { nativeStopTurn } from '../src/cloud-lite/runtime.mjs'; +import { tempDir } from './support/temp-dir.mjs'; + +const fresh = () => tempDir('factory-cloud-lite-claude-'); +const fake = fileURLToPath(new URL('./fixtures/fake-claude.mjs', import.meta.url)); +const provider = (extra = {}) => ({ id: 'selfhosted', envKey: 'SELF_HOSTED_MODEL_KEY', model: 'served-model', apis: ['anthropic'], + baseUrls: ['https://model-a.example.test/v1', 'https://model-b.example.test/v1'], contextWindow: 262144, autoCompactTokenLimit: 80000, ...extra }); +const plan = (extra = {}) => validateCloudLitePlan({ runId: 'claude-test', app: 'factory-claude-test', goal: 'Work on operator jobs', machines: 1, agentsPerMachine: 2, + workerProgram: 'claude', modelProvider: provider(), ...extra }); +const text = value => [{ type: 'text', text: value }]; + +function harness({ root = fresh(), args = [], ...options } = {}) { + const log = join(root, 'invocations.jsonl'); + const driver = new ClaudeSessionDriver({ command: process.execPath, args: [fake, ...args], configDirectory: join(root, 'claude'), stateDirectory: join(root, 'state'), + environment: () => ({ ...process.env, CLAUDE_CONFIG_DIR: join(root, 'claude'), FAKE_CLAUDE_LOG: log }), ...options }); + const notes = []; + driver.on('notification', note => notes.push(note)); + const finished = () => new Promise(resolve => { const check = note => { if (note.method === 'turn/completed') { driver.off('notification', check); resolve(note.params.turn); } }; driver.on('notification', check); }); + const invocations = () => existsSync(log) ? readFileSync(log, 'utf8').trim().split('\n').map(line => JSON.parse(line)) : []; + return { root, driver, notes, finished, invocations }; +} +const started = ({ driver, root }, extra = {}) => driver.request('thread/start', { model: 'served-model', cwd: root, sandbox: 'danger-full-access', developerInstructions: 'Be useful.', ...extra }); +const output = (notes, type) => notes.filter(note => note.method === 'item/completed' && note.params.item.type === type).map(note => note.params.item); +const waitFor = async condition => { for (let i = 0; i < 400; i++) { if (condition()) return; await new Promise(resolve => setTimeout(resolve, 25)); } throw new Error('Condition was not reached.'); }; + +test('plans choose a worker program and refuse a provider without the API it needs', () => { + assert.equal('workerProgram' in validateCloudLitePlan({ runId: 'codex-test', app: 'factory-codex-test', goal: 'x' }), false, 'existing fleet bindings stay unchanged'); + assert.equal(validateCloudLitePlan({ runId: 'codex-test', app: 'factory-codex-test', goal: 'x' }).model, 'gpt-6.1-sol'); + const subscription = plan({ modelProvider: undefined }); + assert.deepEqual([subscription.workerProgram, subscription.model, subscription.subagentModel], ['claude', 'opus', 'sonnet']); + assert.equal(plan().modelProvider.model, 'served-model'); + assert.throws(() => plan({ workerProgram: 'other' }), /workerProgram must be one of codex, claude/); + assert.throws(() => plan({ modelProvider: provider({ apis: ['responses'] }) }), /offers the anthropic API/); + assert.throws(() => plan({ workerProgram: 'codex' }), /offers the responses API/); + assert.throws(() => plan({ modelProvider: provider({ model: undefined }) }), /served model/); + assert.equal(plan({ modelProvider: provider({ model: undefined }), model: 'named' }).model, 'named'); + assert.equal(machineConfig(plan(), { image: 'i', volumeId: 'v', index: 0, environment: {} }).metadata['factory.runtime'], 'claude-lite'); + assert.equal(machineConfig(plan({ workerProgram: 'codex', modelProvider: provider({ apis: ['responses'] }) }), { image: 'i', volumeId: 'v', index: 0, environment: {} }).metadata['factory.runtime'], 'codex-lite'); +}); + +test('the environment holds the key, the arguments never do, and agent commands lose it', () => { + const key = secret(), githubToken = secret(); + const env = claudeEnvironment({ plan: plan(), index: 1, home: '/data/claude', commandWrapper: '/w.sh', providerKey: key, githubToken, path: '/usr/bin' }); + assert.equal(env.ANTHROPIC_AUTH_TOKEN, key); + assert.equal(env.ANTHROPIC_BASE_URL, 'https://model-b.example.test'); + assert.equal(env.ANTHROPIC_MODEL, 'served-model'); + assert.equal(env.CLAUDE_CODE_SHELL_PREFIX, '/w.sh'); + assert.equal(env.CLAUDE_CONFIG_DIR, '/data/claude'); + assert.equal(env.GH_TOKEN, githubToken); + assert.equal('CLAUDE_CODE_OAUTH_TOKEN' in env, false); + assert.equal('GH_TOKEN' in claudeEnvironment({ plan: plan(), home: '/h', providerKey: key, githubToken, deployKey: 'k' }), false); + const oauthToken = secret(); + const subscription = claudeEnvironment({ plan: plan({ modelProvider: undefined }), home: '/h', oauthToken }); + assert.equal(subscription.CLAUDE_CODE_OAUTH_TOKEN, oauthToken); + assert.equal('ANTHROPIC_AUTH_TOKEN' in subscription, false); + const providerWrapper = claudeCommandWrapper({ provider: true }); + for (const name of ['ANTHROPIC_AUTH_TOKEN', 'ANTHROPIC_API_KEY', 'CLAUDE_CODE_OAUTH_TOKEN', 'GH_TOKEN']) assert.match(providerWrapper, new RegExp(`-u ${name}\\b`)); + assert.doesNotMatch(claudeCommandWrapper(), /GH_TOKEN/, 'subscription workers keep GitHub access, as Codex does'); + assert.match(providerWrapper, /bash -c "\$1"/); +}); + +test('sandbox names map to permission modes and the operating system sandbox', () => { + const full = claudeSandboxPolicy({ directory: '/w' }); + assert.deepEqual(full.args, ['--dangerously-skip-permissions']); + assert.equal(full.settings, undefined); + const write = claudeSandboxPolicy({ sandbox: 'workspace-write', directory: '/w', writableRoots: ['/scratch'] }); + assert.deepEqual(write.args, ['--permission-mode', 'acceptEdits', '--add-dir', '/scratch']); + assert.deepEqual(write.settings.sandbox.filesystem, { allowWrite: ['/scratch'] }); + assert.deepEqual(write.settings.sandbox.network, { allowedDomains: [] }); + assert.equal(write.settings.sandbox.allowUnsandboxedCommands, false); + assert.equal(write.settings.sandbox.failIfUnavailable, true); + assert.ok(write.tools.includes('Write')); + const read = claudeSandboxPolicy({ sandbox: 'read-only', directory: '/w' }); + assert.deepEqual(read.args, ['--permission-mode', 'dontAsk']); + assert.deepEqual(read.settings.sandbox.filesystem, { denyWrite: ['/w'] }); + assert.equal(read.tools.includes('Write') || read.tools.includes('Edit'), false); + assert.throws(() => claudeSandboxPolicy({ sandbox: 'none', directory: '/w' }), /Unsupported sandbox/); + const args = claudeSessionArguments({ threadId: 't', resume: false, model: 'm', effort: 'unknown', policy: full, boardTools: false }); + assert.equal(args.includes('--effort'), false, 'an effort the program does not know is left out'); + assert.equal(args.includes('--allowedTools'), false, 'no board tools to allow'); + assert.deepEqual(args.slice(-2), ['--session-id', 't']); + const board = claudeSessionArguments({ threadId: 't', resume: true, effort: 'high', policy: write, boardTools: true, mcpConfigFile: '/m.json', settingsFile: '/s.json' }); + assert.equal(board[board.indexOf('--allowedTools') + 1], 'Bash,mcp__factory__*'); + assert.equal(board[board.indexOf('--effort') + 1], 'high'); + assert.equal(board[board.indexOf('--strict-mcp-config') + 1], '--setting-sources'); + assert.deepEqual(board.slice(-2), ['--resume', 't']); +}); + +test('a session starts, maps its events to trace records, and resumes on the next turn', async () => { + const h = harness(); const { driver, notes } = h; + const board = { command: 'node', args: ['proxy.mjs'], env: { FACTORY_BOARD_TOKEN: 'board-credential' } }; + const { thread } = await started(h, { config: { mcp_servers: { factory: { ...board, required: true } } } }); + assert.deepEqual(thread.status, { type: 'idle' }); + assert.match(thread.id, /^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/); + const first = h.finished(); + const { turn } = await driver.request('turn/start', { threadId: thread.id, input: text('TOOL please'), effort: 'medium' }); + assert.equal((await driver.request('thread/read', { threadId: thread.id })).thread.status.type, 'active'); + await assert.rejects(driver.request('turn/start', { threadId: thread.id, input: text('again') }), error => error.code === -32600); + assert.deepEqual(await first, { id: turn.id, status: 'completed' }); + const methods = notes.map(note => note.method); + assert.deepEqual(methods.slice(0, 3), ['thread/started', 'thread/status/changed', 'turn/started']); + assert.deepEqual(methods.slice(-3), ['thread/tokenUsage/updated', 'turn/completed', 'thread/status/changed']); + assert.equal(notes.at(-1).params.status.type, 'idle'); + assert.equal(output(notes, 'reasoning')[0].text, 'choosing'); + assert.deepEqual(output(notes, 'toolCall')[0], { type: 'toolCall', id: 'toolu_1', tool: 'Bash', arguments: { command: 'true' } }); + assert.deepEqual(output(notes, 'toolResult')[0], { type: 'toolResult', id: 'toolu_1', isError: false, content: 'ok' }); + assert.equal(output(notes, 'agentMessage')[0].text, 'reply to: TOOL please'); + assert.deepEqual(notes.find(note => note.method === 'thread/tokenUsage/updated').params.tokenUsage.last, { inputTokens: 100, cachedInputTokens: 90, outputTokens: 5 }); + const read = await driver.request('thread/read', { threadId: thread.id, includeTurns: true }); + assert.deepEqual(read.thread, { id: thread.id, status: { type: 'idle' }, turns: [{ id: turn.id, status: 'completed' }] }); + + const second = h.finished(); + await driver.request('turn/start', { threadId: thread.id, input: text('next') }); + assert.equal((await second).status, 'completed'); + const [one, two] = h.invocations(); + assert.equal(one.args[one.args.indexOf('--session-id') + 1], thread.id); + assert.equal(two.args[two.args.indexOf('--resume') + 1], thread.id); + assert.equal(two.args.includes('--session-id'), false); + assert.equal(one.args[one.args.indexOf('--append-system-prompt') + 1], 'Be useful.'); + assert.equal(one.args[one.args.indexOf('--model') + 1], 'served-model'); + assert.equal(one.args[one.args.indexOf('--effort') + 1], 'medium'); + assert.equal(one.args.includes('board-credential'), false, 'the board credential stays out of arguments'); + const mcpFile = one.args[one.args.indexOf('--mcp-config') + 1]; + assert.deepEqual(JSON.parse(readFileSync(mcpFile, 'utf8')), { mcpServers: { factory: board } }); + if (process.platform !== 'win32') assert.equal(statSync(mcpFile).mode & 0o777, 0o600); + driver.close(); +}); + +test('a restarted machine resumes a session, and only lost history is reported as unavailable', async () => { + const root = fresh(); const first = harness({ root }); + const { thread } = await started(first); + const done = first.finished(); + await first.driver.request('turn/start', { threadId: thread.id, input: text('work') }); + await done; + const later = harness({ root }); + const { thread: resumed } = await later.driver.request('thread/resume', { threadId: thread.id, cwd: root, sandbox: 'danger-full-access', model: 'served-model', developerInstructions: 'Be useful.' }); + assert.equal(resumed.id, thread.id); + assert.equal(later.notes[0].method, 'thread/resumed'); + const next = later.finished(); + await later.driver.request('turn/start', { threadId: thread.id, input: text('more') }); + assert.equal((await next).status, 'completed'); + const invocation = later.invocations().at(-1); + assert.equal(invocation.args[invocation.args.indexOf('--resume') + 1], thread.id); + assert.equal(invocation.args[invocation.args.indexOf('--append-system-prompt') + 1], 'Be useful.'); + + // The runtime recognizes exactly this error and starts a replacement session. + const unknown = '01a13923-1da9-76a1-0a6e-af0c060ec3a4'; + await assert.rejects(later.driver.request('thread/resume', { threadId: unknown, cwd: root }), + error => error.code === -32600 && error.message === `no rollout found for thread id ${unknown}`); + const idle = await started(later); + const { thread: stillIdle } = await harness({ root }).driver.request('thread/resume', { threadId: idle.thread.id, cwd: root, sandbox: 'danger-full-access' }); + assert.equal(stillIdle.status.type, 'idle', 'a session that never ran a turn has nothing to lose'); + writeFileSync(join(root, 'state', `${unknown}.claude.json`), JSON.stringify({ threadId: unknown, lastTurn: { id: 'gone', status: 'completed' } })); + await assert.rejects(later.driver.request('thread/resume', { threadId: unknown, cwd: root }), /no rollout found/); + const interrupted = '01a13924-1da9-76a1-0a6e-af0c060ec3a4'; + writeFileSync(join(root, 'state', `${interrupted}.claude.json`), JSON.stringify({ threadId: interrupted, lastTurn: { id: 'crashed', status: 'inProgress' } })); + await assert.rejects(later.driver.request('thread/resume', { threadId: interrupted, cwd: root }), /no rollout found/); +}); + +test('a turn left running by a lost machine reads back as interrupted', async () => { + const root = fresh(); const first = harness({ root }); + const { thread } = await started(first); + const done = first.finished(); + await first.driver.request('turn/start', { threadId: thread.id, input: text('work') }); + await done; + const file = join(root, 'state', `${thread.id}.claude.json`); + writeFileSync(file, JSON.stringify({ threadId: thread.id, lastTurn: { id: 'crashed', status: 'inProgress' } })); + const later = harness({ root }); + await later.driver.request('thread/resume', { threadId: thread.id, cwd: root, sandbox: 'danger-full-access' }); + const read = await later.driver.request('thread/read', { threadId: thread.id, includeTurns: true }); + assert.deepEqual(read.thread.turns, [{ id: 'crashed', status: 'interrupted' }]); +}); + +test('an operator message steers a running turn, which ends only once', async () => { + const h = harness(); const { driver, notes } = h; + const { thread } = await started(h); + await assert.rejects(driver.request('turn/steer', { threadId: thread.id, input: text('early') }), /no active turn to steer/); + const done = h.finished(); + const { turn } = await driver.request('turn/start', { threadId: thread.id, input: text('LONG job') }); + await assert.rejects(driver.request('turn/steer', { threadId: thread.id, expectedTurnId: '01a13923-1da9-76a1-0a6e-af0c060ec3a4', input: text('wrong') }), error => error.code === -32600); + assert.deepEqual(await driver.request('turn/steer', { threadId: thread.id, expectedTurnId: turn.id, input: text('change course') }), { turnId: turn.id }); + assert.deepEqual(await done, { id: turn.id, status: 'completed' }); + assert.equal(output(notes, 'agentMessage')[0].text, 'steered by: change course'); + assert.equal(notes.filter(note => note.method === 'turn/completed').length, 1); + await assert.rejects(driver.request('turn/steer', { threadId: thread.id, input: text('late') }), /no active turn to steer/); +}); + +test('a turn stays open while a queued message has not run yet', async () => { + const h = harness(); const { driver, notes } = h; + const { thread } = await started(h); + const done = h.finished(); + const { turn } = await driver.request('turn/start', { threadId: thread.id, input: text('QUEUE BRIEF job') }); + await driver.request('turn/steer', { threadId: thread.id, expectedTurnId: turn.id, input: text('and then this') }); + assert.deepEqual(await done, { id: turn.id, status: 'completed' }); + assert.equal(output(notes, 'agentMessage').at(-1).text, 'reply to: and then this'); + assert.equal(notes.filter(note => note.method === 'thread/tokenUsage/updated').length, 2); + assert.equal(notes.filter(note => note.method === 'turn/completed').length, 1); +}); + +test('a queued message that never comes back cannot hold the turn open', async () => { + const h = harness({ steerGraceMs: 100 }); const { driver, notes } = h; + const { thread } = await started(h); + const done = h.finished(); + const { turn } = await driver.request('turn/start', { threadId: thread.id, input: text('DROP BRIEF job') }); + await driver.request('turn/steer', { threadId: thread.id, expectedTurnId: turn.id, input: text('lost') }); + assert.deepEqual(await done, { id: turn.id, status: 'completed' }); + assert.equal(notes.filter(note => note.method === 'turn/completed').length, 1); +}); + +test('a worker whose board did not connect ends its turn as failed', async () => { + const errors = []; const h = harness({ args: ['--fake-board-failed'], onStderr: value => errors.push(value) }); + const { thread } = await started(h, { config: { mcp_servers: { factory: { command: 'node', args: [], env: {} } } } }); + const done = h.finished(); + await h.driver.request('turn/start', { threadId: thread.id, input: text('LONG job') }); + assert.equal((await done).status, 'failed'); + assert.match(errors.join(' '), /board MCP server did not connect/); +}); + +function stopper(h, threadId, destination) { + const turns = new Map(), terminalTurns = new Map(), records = []; + const stop = new NativeOperatorStop({ rpc: h.driver, threadId, getTurn: () => nativeStopTurn(turns, terminalTurns, threadId), + trace: (kind, body) => { records.push({ kind, body }); destination?.record(kind, body); }, flush: async () => destination?.flush(), timeoutMs: 5000 }); + h.driver.on('notification', ({ method, params }) => { + if (method === 'turn/started') { terminalTurns.delete(threadId); turns.set(threadId, params.turn.id); stop.observeTurnStarted({ threadId, turnId: params.turn.id }); } + if (method === 'turn/completed') { terminalTurns.set(threadId, { id: params.turn.id, status: params.turn.status }); stop.observeTerminal({ threadId, turnId: params.turn.id, status: params.turn.status }); } + }); + return { stop, records }; +} + +test('an operator stop interrupts the running turn and proves the session idle', async () => { + const h = harness(); const { driver } = h; + const { thread } = await started(h); + const { stop, records } = stopper(h, thread.id); + const { turn } = await driver.request('turn/start', { threadId: thread.id, input: text('LONG job') }); + const receipt = await stop.stop('stop-1', { expectedThreadId: thread.id, expectedTurnId: turn.id }); + assert.deepEqual(receipt, { commandId: 'stop-1', expectedThreadId: thread.id, expectedTurnId: turn.id, threadId: thread.id, threadStatus: 'idle', turn: { id: turn.id, status: 'interrupted' } }); + assert.deepEqual(records.map(record => record.kind), ['thread/status/changed', 'native.operator.stop.completed']); + assert.equal((await driver.request('thread/read', { threadId: thread.id })).thread.status.type, 'idle'); + // Stopping an idle session is a readback, and a new turn then continues the same session. + assert.equal((await stop.stop('stop-2', { expectedThreadId: thread.id, expectedTurnId: turn.id })).threadStatus, 'idle'); + const again = h.finished(); + await driver.request('turn/start', { threadId: thread.id, input: text('continue') }); + assert.equal((await again).status, 'completed'); +}); + +test('an unanswered interrupt ends the process, and stop wins over a queued message', async () => { + const deaf = harness({ interruptGraceMs: 100, killGraceMs: 2000 }); + const { thread } = await started(deaf); + const { stop } = stopper(deaf, thread.id); + const { turn } = await deaf.driver.request('turn/start', { threadId: thread.id, input: text('DEAF job') }); + const receipt = await stop.stop('stop-deaf', { expectedThreadId: thread.id, expectedTurnId: turn.id }); + assert.equal(receipt.turn.status, 'interrupted'); + assert.equal(receipt.threadStatus, 'idle'); + + const queued = harness(); + const second = await started(queued); + const done = queued.finished(); + const running = await queued.driver.request('turn/start', { threadId: second.thread.id, input: text('QUEUE LONG job') }); + await queued.driver.request('turn/steer', { threadId: second.thread.id, input: text('queued work') }); + await queued.driver.request('turn/interrupt', { threadId: second.thread.id, turnId: running.turn.id }); + await assert.rejects(queued.driver.request('turn/steer', { threadId: second.thread.id, input: text('too late') }), /no active turn to steer/); + assert.equal((await done).status, 'interrupted'); + assert.deepEqual(output(queued.notes, 'agentMessage'), [], 'the queued message did not run'); + assert.deepEqual(await queued.driver.request('turn/interrupt', { threadId: second.thread.id, turnId: running.turn.id }), {}, 'interrupting a finished turn is harmless'); + await assert.rejects(queued.driver.request('turn/interrupt', { threadId: second.thread.id, turnId: 'other' }), error => error.code === -32600); +}); + +test('a crash, a refused key and a wrong session end the turn as failed without blocking the next one', async () => { + const h = harness(); const { driver } = h; + const { thread } = await started(h); + const crashed = h.finished(); + await driver.request('turn/start', { threadId: thread.id, input: text('CRASH') }); + assert.equal((await crashed).status, 'failed'); + const auth = []; driver.on('authFailed', value => auth.push(value)); + const refused = h.finished(); + await driver.request('turn/start', { threadId: thread.id, input: text('REFUSED') }); + assert.deepEqual((await refused).error, { code: 401 }); + assert.deepEqual(auth, [{ threadId: thread.id }]); + const ok = h.finished(); + await driver.request('turn/start', { threadId: thread.id, input: text('fine') }); + assert.equal((await ok).status, 'completed'); + + const other = harness({ args: ['--fake-other'] }); + const { thread: otherThread } = await started(other); + const ended = other.finished(); + await other.driver.request('turn/start', { threadId: otherThread.id, input: text('LONG') }); + assert.equal((await ended).status, 'failed'); + + const missing = new ClaudeSessionDriver({ command: join(h.root, 'not-installed'), configDirectory: join(h.root, 'c2'), stateDirectory: join(h.root, 's2'), environment: () => process.env }); + const { thread: lonely } = await missing.request('thread/start', { cwd: h.root }); + const gone = new Promise(resolve => missing.on('notification', note => { if (note.method === 'turn/completed') resolve(note.params.turn.status); })); + await missing.request('turn/start', { threadId: lonely.id, input: text('hello') }); + assert.equal(await gone, 'failed'); + assert.equal((await missing.request('thread/read', { threadId: lonely.id })).thread.status.type, 'idle'); +}); + +test('a durable workflow wake delivers a turn and reads it back after an unknown outcome', async () => { + const h = harness(); const { driver } = h; + const { thread } = await started(h); + const done = h.finished(); + const wake = new NativeWorkflowWake({ rpc: driver, threadId: thread.id, file: join(h.root, 'wake.json'), effort: 'medium' }); + const delivered = await wake.wake({ trigger: 'first', input: 'start', initial: true }); + assert.equal(delivered.status, 'delivered'); + const turn = await done; + assert.equal(delivered.turnId, turn.id); + assert.deepEqual(await wake.read(), { status: 'idle', turnId: turn.id }); +}); + +test('a real stop command reaches the board as a verified receipt', async () => { + const p = plan({ agentsPerMachine: 1 }); const agent = agentInventory(p)[0]; + const credentials = { [agent.id]: secret() }; const operatorToken = secret(); const root = fresh(); + const coordinator = await startCloudLiteCoordinator({ database: join(root, 'board.sqlite'), plan: p, credentials, operatorToken, auth: {}, host: '127.0.0.1', port: 0 }); + const url = `http://127.0.0.1:${coordinator.server.address().port}`; + const post = async (path, body, token = operatorToken) => { + const response = await fetch(url + path, { method: 'POST', headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' }, body: JSON.stringify(body) }); + assert.equal(response.status, 200); return response.json(); + }; + const h = harness({ root }); const { driver } = h; + try { + const { thread } = await started(h); + await post('/operator/session', { agent: agent.id, session: thread.id }); + const trace = new CloudLiteTrace({ file: join(root, 'trace.jsonl'), url, token: credentials[agent.id] }); + const { stop } = stopper(h, thread.id, trace); + driver.on('notification', ({ method, params }) => { + if (method === 'turn/started') trace.record('native.turn.started', { threadId: thread.id, turn: params.turn }); + if (method === 'turn/completed') trace.record('native.turn.completed', { threadId: thread.id, turn: params.turn }); + if (method === 'item/completed') trace.record(method, params); + }); + trace.record('native.thread.started', { threadId: thread.id, threadStatus: 'idle' }); + const { turn } = await driver.request('turn/start', { threadId: thread.id, input: text('LONG job') }); + await trace.flush(); + await waitFor(() => h.invocations().length === 1); + await post('/operator/command', { id: 'stop-board', recipient: agent.id, action: 'stop', expectedThreadId: thread.id, expectedTurnId: turn.id }); + const { commands } = await post('/commands', {}, credentials[agent.id]); + const command = commands.find(value => value.id === 'stop-board'); + await stop.stop(command.id, { expectedThreadId: command.expectedThreadId, expectedTurnId: command.expectedTurnId }); + const ack = await post('/commands', { acknowledge: command.id }, credentials[agent.id]); + assert.equal(ack.acknowledged, true); + const native = coordinator.board.snapshot().agents.find(value => value.id === agent.id).nativeRuntime; + assert.equal(native.threadStatus, 'idle'); + assert.equal(native.turnStatus, 'interrupted'); + assert.equal(native.operatorStopReceipt.commandId, 'stop-board'); + assert.equal(native.operatorStopReceipt.threadId, thread.id); + assert.equal(native.operatorStopReceipt.turnId, turn.id); + } finally { driver.close(); await coordinator.close(); } +}); + +test('bootstrap installs only the selected program, pinned by checksum', () => { + const digest = 'a'.repeat(64); + const codex = cloudLiteBootstrap(digest); + assert.equal(codex, cloudLiteBootstrap(digest, { workerProgram: 'codex' })); + assert.match(codex, /rust-v0\.162\.1/); + assert.match(codex, /CODEX_HOME=\/data\/codex/); + assert.doesNotMatch(codex, /claude/i); + const claude = cloudLiteBootstrap(digest, { workerProgram: 'claude' }); + assert.match(claude, /claude-code-releases\/\d+\.\d+\.\d+\/linux-x64\/claude/); + assert.match(claude, /^echo '[a-f0-9]{64} {2}\/tmp\/claude' \| sha256sum -c -$/m); + assert.match(claude, /bubblewrap socat/); + assert.match(claude, /CLAUDE_CONFIG_DIR=\/data\/claude/); + assert.doesNotMatch(claude, /codex/i); + assert.ok(claude.indexOf('sha256sum -c -\ninstall') > 0, 'the checksum is checked before install'); + assert.throws(() => cloudLiteBootstrap(digest, { workerProgram: 'other' }), /Unknown worker program/); +}); + +test('the subscription broker holds one token and fleets refuse the wrong broker', async () => { + assert.throws(() => new ClaudeSubscriptionBroker({ tokenFile: 'relative' }), /absolute/); + const root = fresh(); const tokenFile = join(root, 'token'); const token = 'sk-ant-oat01-' + 'x'.repeat(40); + const broker = new ClaudeSubscriptionBroker({ tokenFile }); + writeFileSync(tokenFile, `${token}\n`); + assert.deepEqual(broker.read(), { accessToken: token }); + writeFileSync(tokenFile, `${token}-rotated`); + assert.deepEqual(await broker.refresh(), { accessToken: `${token}-rotated` }); + writeFileSync(tokenFile, 'sk-ant-api03-' + 'x'.repeat(40)); + assert.throws(() => broker.read(), /API-key fallback is disabled/); + writeFileSync(tokenFile, 'short'); + assert.throws(() => broker.read(), /one token/); + const image = 'ghcr.io/example/worker@sha256:' + 'a'.repeat(64); + const codexBroker = new FactoryCloudLiteFleet({ directory: fresh(), auth: new CodexSubscriptionBroker({ authFile: join(root, 'auth.json') }), infrastructure: {} }); + await assert.rejects(codexBroker.deploy(plan({ modelProvider: undefined }), { image }), /codex subscription broker cannot serve claude workers/); + const claudeBroker = new FactoryCloudLiteFleet({ directory: fresh(), auth: broker, infrastructure: {} }); + await assert.rejects(claudeBroker.deploy({ runId: 'codex-test', app: 'factory-codex-test', goal: 'x' }, { image }), /claude subscription broker cannot serve codex workers/); +}); + +test('a Claude fleet stages the token only as a secret and keeps it out of the journal', async () => { + const root = fresh(); const tokenFile = join(root, 'token'); const token = 'sk-ant-oat01-' + 'y'.repeat(40); + writeFileSync(tokenFile, token); + const calls = []; const machines = []; + const infrastructure = { + run: async (args, options = {}) => { calls.push({ args, input: options.input }); if (args[0] === 'volumes') return JSON.stringify({ id: 'vol_1' }); if (args[0] === 'machine') return JSON.stringify(machines); return ''; }, + buildMachineConfig: () => ({ init: { cmd: ['upstream'] }, services: [] }), + createMachine: async ({ name, config }) => { const machine = { id: `machine-${machines.length}`, name, config }; machines.push(machine); return machine; }, + }; + const bundlePath = join(root, 'bundle.tgz'); writeFileSync(bundlePath, 'bundle'); + const fleet = new FactoryCloudLiteFleet({ directory: join(root, 'fleet'), auth: new ClaudeSubscriptionBroker({ tokenFile }), infrastructure }); + const image = 'ghcr.io/example/worker@sha256:' + 'a'.repeat(64); + const result = await fleet.deploy(plan({ modelProvider: undefined, machines: 1 }), { image, bundlePath }); + assert.equal(result.machines.length, 1); + const staged = calls.find(call => call.args[0] === 'secrets'); + const config = JSON.parse(Buffer.from(staged.input.trim().split('=').slice(1).join('='), 'base64').toString()); + assert.deepEqual(config.initialAuth, { accessToken: token }); + assert.equal(config.plan.workerProgram, 'claude'); + assert.equal(readFileSync(join(root, 'fleet', 'fleet.json'), 'utf8').includes(token), false); + assert.match(readFileSync(join(root, 'fleet', 'bootstrap.sh'), 'utf8'), /claude-code-releases/); + assert.equal(machines[0].config.metadata['factory.runtime'], 'claude-lite'); +}); diff --git a/test/cloud-lite-provider.test.mjs b/test/cloud-lite-provider.test.mjs index 3e38652..30d78a0 100644 --- a/test/cloud-lite-provider.test.mjs +++ b/test/cloud-lite-provider.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, readFileSync, writeFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { readFileSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { validateCloudLitePlan, secret } from '../src/cloud-lite/protocol.mjs'; import { cloudLiteBootstrap } from '../src/cloud-lite/bootstrap.mjs'; diff --git a/test/cloud-lite.test.mjs b/test/cloud-lite.test.mjs index 1e96ed7..ddbdebc 100644 --- a/test/cloud-lite.test.mjs +++ b/test/cloud-lite.test.mjs @@ -1,8 +1,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, readFileSync, writeFileSync, existsSync, mkdirSync, statSync, fsync } from 'node:fs'; +import { readFileSync, writeFileSync, existsSync, mkdirSync, statSync, fsync } from 'node:fs'; import { promisify } from 'node:util'; -import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { CloudLiteBoard } from '../src/cloud-lite/board.mjs'; import { validateCloudLitePlan, agentInventory, machineConfig, redact, secret, hash } from '../src/cloud-lite/protocol.mjs'; diff --git a/test/fixtures/fake-claude.mjs b/test/fixtures/fake-claude.mjs new file mode 100644 index 0000000..029b465 --- /dev/null +++ b/test/fixtures/fake-claude.mjs @@ -0,0 +1,80 @@ +#!/usr/bin/env node +// Plays Claude Code's stream-json side as observed from the pinned release, driven by the words of +// each message: LONG waits to be interrupted, DEAF ignores the interrupt, TOOL uses a tool, CRASH +// exits at once, REFUSED reports a refused key. QUEUE keeps later messages for after the turn, and +// BRIEF ends such a turn by itself shortly after. DROP loses later messages. +import { appendFileSync, existsSync, mkdirSync, writeFileSync } from 'node:fs'; +import { randomUUID } from 'node:crypto'; +import { createInterface } from 'node:readline'; +import { join } from 'node:path'; + +const args = process.argv.slice(2); +const flag = name => { const at = args.indexOf(name); return at < 0 ? undefined : args[at + 1]; }; +const resumed = flag('--resume'); +const session = resumed ?? flag('--session-id'); +const transcript = join(process.env.CLAUDE_CONFIG_DIR, 'projects', '-fake', `${session}.jsonl`); +if (process.env.FAKE_CLAUDE_LOG) appendFileSync(process.env.FAKE_CLAUDE_LOG, JSON.stringify({ args, cwd: process.cwd(), env: process.env }) + '\n'); +const send = value => process.stdout.write(JSON.stringify(value) + '\n'); +const emit = (type, extra = {}) => send({ type, session_id: session, uuid: randomUUID(), ...extra }); + +if (resumed && !existsSync(transcript)) { + emit('result', { subtype: 'error_during_execution', is_error: true, errors: [`No conversation found with session ID: ${session}`] }); + process.exit(1); +} +mkdirSync(join(process.env.CLAUDE_CONFIG_DIR, 'projects', '-fake'), { recursive: true }); +writeFileSync(transcript, '{}\n', { flag: 'a' }); +send({ type: 'system', subtype: 'init', session_id: flag('--session-id') && args.includes('--fake-other') ? randomUUID() : session, tools: [], + mcp_servers: args.includes('--fake-board-failed') ? [{ name: 'factory', status: 'failed' }] : [{ name: 'factory', status: 'connected' }] }); +emit('command_lifecycle'); + +const queue = []; let running = null; let inputEnded = false; +const assistant = content => emit('assistant', { message: { role: 'assistant', content } }); +const usage = { input_tokens: 10, cache_read_input_tokens: 90, output_tokens: 5 }; +function finish(result, delayMs = 0) { + running = null; + emit('result', { usage, ...result }); + if (delayMs) setTimeout(next, delayMs); else next(); +} +function next() { + if (running) return; + const message = queue.shift(); + if (!message) { if (inputEnded) process.exit(0); return; } + const text = message.message.content.map(block => block.text).join(' '); + running = { text, deaf: text.includes('DEAF'), queueing: text.includes('QUEUE'), drop: text.includes('DROP') }; + emit('user', { isReplay: true, message: message.message }); + if (text.includes('CRASH')) process.exit(3); + if (text.includes('BRIEF')) { setTimeout(() => finish({ subtype: 'success', is_error: false, result: 'brief' }), 200); return; } + if (text.includes('LONG') || text.includes('DEAF')) return; + if (text.includes('REFUSED')) { + emit('system', { subtype: 'api_retry', attempt: 1, error_status: 401, error: 'authentication_failed' }); + finish({ subtype: 'success', is_error: true, api_error_status: 401, result: 'Failed to authenticate' }); + return; + } + if (text.includes('TOOL')) { + assistant([{ type: 'thinking', thinking: 'choosing' }, { type: 'tool_use', id: 'toolu_1', name: 'Bash', input: { command: 'true' } }]); + emit('user', { message: { role: 'user', content: [{ type: 'tool_result', tool_use_id: 'toolu_1', content: 'ok' }] } }); + } + assistant([{ type: 'text', text: `reply to: ${text}` }]); + finish({ subtype: 'success', is_error: false, result: `reply to: ${text}` }); +} +createInterface({ input: process.stdin }).on('line', line => { + const value = JSON.parse(line); + if (value.type === 'control_request' && value.request?.subtype === 'interrupt') { + send({ type: 'control_response', response: { subtype: 'success', request_id: value.request_id } }); + if (!running || running.deaf) return; + emit('user', { message: { role: 'user', content: [{ type: 'text', text: '[Request interrupted by user]' }] } }); + finish({ subtype: 'error_during_execution', is_error: true, errors: ['interrupted'] }, 1000); + return; + } + // A message written during a running turn joins it: it comes back at once and ends that turn. + if (value.type === 'user' && running?.drop) return; + if (value.type === 'user' && running && !running.deaf && !running.queueing) { + const text = value.message.content.map(block => block.text).join(' '); + emit('user', { isReplay: true, message: value.message }); + assistant([{ type: 'text', text: `steered by: ${text}` }]); + finish({ subtype: 'success', is_error: false, result: `steered by: ${text}` }); + return; + } + if (value.type === 'user') { queue.push(value); next(); } +}).on('close', () => { inputEnded = true; if (!running && !queue.length) process.exit(0); }); +setInterval(() => {}, 1000); diff --git a/test/generic-swarm-evidence.test.mjs b/test/generic-swarm-evidence.test.mjs index 3ca0876..dc85a42 100644 --- a/test/generic-swarm-evidence.test.mjs +++ b/test/generic-swarm-evidence.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, openSync, writeSync, closeSync, renameSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { mkdirSync, writeFileSync, readFileSync, openSync, writeSync, closeSync, renameSync } from 'node:fs'; import { join } from 'node:path'; import { randomBytes } from 'node:crypto'; import { FactoryControl } from '../src/control.mjs'; diff --git a/test/model-provider.test.mjs b/test/model-provider.test.mjs index 667ccf3..e5809fa 100644 --- a/test/model-provider.test.mjs +++ b/test/model-provider.test.mjs @@ -1,8 +1,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { spawnSync } from 'node:child_process'; -import { mkdtempSync, writeFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { claudeProviderEnvironment, codexProviderConfig, codexProviderOverrides, liteWorkerCommand, providerPrograms, validateModelProvider } from '../src/public-sdk.mjs'; diff --git a/test/selfhost-modal.test.mjs b/test/selfhost-modal.test.mjs index 7ab7b8e..f2447ea 100644 --- a/test/selfhost-modal.test.mjs +++ b/test/selfhost-modal.test.mjs @@ -1,8 +1,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { spawnSync } from 'node:child_process'; -import { mkdtempSync, readFileSync, existsSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { readFileSync, existsSync } from 'node:fs'; import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { ModalSelfHostedModel, selfHostedProfiles, codexProviderConfig, claudeProviderEnvironment, validateCloudLitePlan } from '../src/public-sdk.mjs'; @@ -185,3 +184,18 @@ test('factory-selfhost lists profiles and never prints the key', () => { assert.equal(key.status, 1); assert.match(key.stderr, /never printed/); assert.match(spawnSync(process.execPath, [cli, '--help'], { encoding: 'utf8' }).stdout, /retire/); }); + +test('a deployment stages the profiles next to the Python sources that read them', async () => { + const { model } = fixture(); + await model.deploy({ profile: 'qwen' }); + const staged = join(model.directory, 'source', 'profiles'); + for (const name of ['qwen.json', 'nemotron.json']) assert.equal(existsSync(join(staged, name)), true, name); + assert.equal(JSON.parse(read(join(staged, 'qwen.json'))).served_name, selfHostedProfiles().find(profile => profile.id === 'qwen').model); +}); + +test('helper runs define no serving function, which would need the secret the deployment creates', () => { + const source = read(fileURLToPath(new URL('../selfhost/modal/app.py', import.meta.url))); + const guard = source.indexOf('if not HELPERS:'); + assert.ok(guard > 0); + for (const name of ['GPU_FUNCTION = dict(', 'def serve():', 'def serve_b():', 'def serve_d():']) assert.ok(source.indexOf(name) > guard, name); +}); diff --git a/test/workflow-m0-lifecycle.test.mjs b/test/workflow-m0-lifecycle.test.mjs index 80330d6..f58048a 100644 --- a/test/workflow-m0-lifecycle.test.mjs +++ b/test/workflow-m0-lifecycle.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { mkdirSync, writeFileSync } from 'node:fs'; import { join, resolve } from 'node:path'; import { execFileSync } from 'node:child_process'; import { fileURLToPath } from 'node:url'; diff --git a/test/workflow-m1-portable-artifacts.test.mjs b/test/workflow-m1-portable-artifacts.test.mjs index c7f56e5..7c32511 100644 --- a/test/workflow-m1-portable-artifacts.test.mjs +++ b/test/workflow-m1-portable-artifacts.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, lstatSync, existsSync, realpathSync, symlinkSync, renameSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { mkdirSync, writeFileSync, readFileSync, lstatSync, existsSync, realpathSync, symlinkSync, renameSync } from 'node:fs'; import { join, relative, resolve, toNamespacedPath } from 'node:path'; import { execFileSync, spawnSync } from 'node:child_process'; import { createHash } from 'node:crypto'; diff --git a/test/workflow-m2-ack-recovery.test.mjs b/test/workflow-m2-ack-recovery.test.mjs index 94915da..28de5ad 100644 --- a/test/workflow-m2-ack-recovery.test.mjs +++ b/test/workflow-m2-ack-recovery.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, writeFileSync, mkdirSync, existsSync, readdirSync, readFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { writeFileSync, mkdirSync, existsSync, readdirSync, readFileSync } from 'node:fs'; import { join } from 'node:path'; import { CloudLiteTrace } from '../src/cloud-lite/trace.mjs'; import { startCloudLiteCoordinator } from '../src/cloud-lite/server.mjs'; diff --git a/test/workflow-m2-authority-recovery.test.mjs b/test/workflow-m2-authority-recovery.test.mjs index bd51424..1dae401 100644 --- a/test/workflow-m2-authority-recovery.test.mjs +++ b/test/workflow-m2-authority-recovery.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { mkdirSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { execFileSync } from 'node:child_process'; import { createWorkflowClient, createWorkflowComparison, launchWorkflowComparison, summarizeWorkflowComparison } from '../src/public-sdk.mjs'; diff --git a/test/workflow-native-wake.test.mjs b/test/workflow-native-wake.test.mjs index ef94894..eb181a4 100644 --- a/test/workflow-native-wake.test.mjs +++ b/test/workflow-native-wake.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import {mkdtempSync,mkdirSync} from 'node:fs'; -import {tmpdir} from 'node:os'; +import {mkdirSync} from 'node:fs'; import {join} from 'node:path'; import {NativeWorkflowWake} from '../src/cloud-lite/workflow-wake.mjs'; import { tempDir } from './support/temp-dir.mjs'; diff --git a/test/workflows.test.mjs b/test/workflows.test.mjs index 59c90ed..827d0fc 100644 --- a/test/workflows.test.mjs +++ b/test/workflows.test.mjs @@ -1,7 +1,6 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, mkdirSync, writeFileSync, readFileSync } from 'node:fs'; -import { tmpdir } from 'node:os'; +import { mkdirSync, writeFileSync, readFileSync } from 'node:fs'; import { resolve, join } from 'node:path'; import { execFileSync } from 'node:child_process'; import { Factory, FactoryWorkflow, FactoryWorkflowSupervisor, runVerification, createWorkflowClient, createWorkflowComparison, summarizeWorkflowComparison, captureGitSource } from '../src/public-sdk.mjs';