From 270b365eceb6d12831382e901f047beba1a7d4ca Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:33:02 -0500 Subject: [PATCH 1/2] Let cloud lite workers use a self-hosted model and operator-driven jobs Cloud lite workers could only use the subscription login. They can now run on a self-hosted OpenAI-compatible (Responses API) model through plan.modelProvider, with the key passed as deploy({ providerKey }) and kept inside the Codex process. Also added, all optional: - deploy({ deployKey }): read-only SSH clone with GitHub's host key pinned - plan.prepare: per-machine hook that returns each agent's folder - plan.supervision 'operator', plan.sandbox, plan.boardTools - streaming deltas no longer traced (plan.traceDeltas restores them), plan.monitorIntervalMs Version 0.4.0. Fixes #9. --- .github/workflows/public-sdk.yml | 2 +- PUBLIC_SDK.md | 100 +++++++++++++++- package-lock.json | 4 +- package.json | 2 +- scripts/verify-public-sdk.mjs | 7 ++ src/cloud-lite/bootstrap.mjs | 5 +- src/cloud-lite/fleet.mjs | 15 ++- src/cloud-lite/protocol.mjs | 39 +++++++ src/cloud-lite/runtime.mjs | 182 ++++++++++++++++++++++++++---- test/cloud-lite-provider.test.mjs | 164 +++++++++++++++++++++++++++ 10 files changed, 484 insertions(+), 36 deletions(-) create mode 100644 test/cloud-lite-provider.test.mjs diff --git a/.github/workflows/public-sdk.yml b/.github/workflows/public-sdk.yml index 6db988d..79fefad 100644 --- a/.github/workflows/public-sdk.yml +++ b/.github/workflows/public-sdk.yml @@ -23,7 +23,7 @@ jobs: cache: npm - run: npm ci - run: npm audit --omit=dev --audit-level=high - - run: node --test test/cloud-lite.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 --test test/cloud-lite.test.mjs test/cloud-lite-provider.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: Install and exercise the packed public package shell: bash run: | diff --git a/PUBLIC_SDK.md b/PUBLIC_SDK.md index b32ac22..6e81fdb 100644 --- a/PUBLIC_SDK.md +++ b/PUBLIC_SDK.md @@ -1,12 +1,19 @@ # Public local Factory SDK, CLI, and MCP -Version 0.3.0 is distributed as a versioned package in -[GitHub Releases](https://github.com/flujo-app/FACTORY/releases/tag/v0.3.0): +Version 0.4.0 is distributed as a versioned package in +[GitHub Releases](https://github.com/flujo-app/FACTORY/releases/tag/v0.4.0): ```sh -npm install https://github.com/flujo-app/FACTORY/releases/download/v0.3.0/flujo-factory-0.3.0.tgz +npm install https://github.com/flujo-app/FACTORY/releases/download/v0.4.0/flujo-factory-0.4.0.tgz ``` +0.4.0 lets cloud lite workers run on a self-hosted OpenAI-compatible model, +clone with a read-only deploy key, prepare agent folders, take operator-driven +jobs in a narrower sandbox, and record far less trace volume. See +[Self-hosted models and operator-driven jobs](#self-hosted-models-and-operator-driven-jobs). +All new plan settings are optional; existing plans behave as before, except +that streaming deltas are no longer traced unless `traceDeltas` is set. + `FactoryControl` and `executeEffect` are public exports for integrations that retain FACTORY's durable admission while using an existing fleet executor. Repeating an effect key observes its original outcome without dispatching again. @@ -243,6 +250,93 @@ 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. +#### Self-hosted models and operator-driven jobs + +Lite workers can run on a self-hosted, OpenAI-compatible model instead of the +subscription login. Codex reaches it through the Responses API +(`/v1/responses`), which vLLM serves for every model. Add `modelProvider` to the +plan and pass its key as `providerKey`; no `auth` broker is needed. A read-only +GitHub deploy key can replace the GitHub token: + +```js +const fleet = new FactoryCloudLiteFleet({ directory: '/absolute/private/fleet-run', + flujoCloudPath: '/absolute/installed/flujo-cloud' }); +await fleet.deploy({ + runId: 'my-selfhosted-run', app: 'my-selfhosted-app', goal: ownerGoal, + repository: 'owner/project', repositoryRevision: pinnedCommitSha, + machines: 10, agentsPerMachine: 19, + model: 'nemotron-3.5-super-vl', subagentModel: 'nemotron-3.5-super-vl', reasoningEffort: 'medium', + modelProvider: { + id: 'selfhosted', name: 'Self-hosted model', envKey: 'SELF_HOSTED_MODEL_KEY', + baseUrls: ['https://model-a.example/v1', 'https://model-b.example/v1'], + contextWindow: 262144, autoCompactTokenLimit: 80000, + }, + supervision: 'operator', sandbox: 'workspace-write', boardTools: false, monitorIntervalMs: 4000, + agentInstructions: 'Wait for operator jobs and work only in your folder.', + prepare: { argv: ['node', 'tools/prepare-agents.mjs'], spec: { jobsPerMachine: 18 } }, +}, { image: immutableWorkerImage, bundlePath: '/absolute/private/flujo-factory.tgz', + providerKey: privateEndpointKey, deployKey: readOnlyDeployKey }); +await fleet.command('m0-a1', { id: 'job-1', action: 'message', message: 'First job ...' }); +``` + +- **`modelProvider`**: `id`, `envKey` (the variable that carries the key) and + `https` `baseUrls` without credentials are required; `name`, `model`, + `contextWindow`, `autoCompactTokenLimit`, `toolOutputTokenLimit`, + `requestMaxRetries`, `streamMaxRetries` and `streamIdleTimeoutMs` are optional. + 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. + 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. +- **`supervision: 'operator'`**: agents start idle. There is no automatic first + 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 + `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. +- **`boardTools: false`** removes the FACTORY board MCP server from the sessions. + The default instructions refer to board tools, so also set `agentInstructions` + or per-agent `instructions` from the prepare hook. Verified workflows need the + board tools and automatic wakes, so they refuse either setting. +- **`prepare`** runs `argv` once per machine after the pinned clone, in the + repository folder, with `FACTORY_MACHINE_INDEX` and `FACTORY_PREPARE_SPEC` (a + JSON file holding `spec`). It prints one JSON line, + `{ "agents": [{ "cwd", "writableRoots"?, "instructions"? }] }`, with one entry per + agent slot and absolute paths. The result is kept on the volume, so a restarted + machine reuses the prepared folders without cloning again; the hook may remove + files agents must not read. Output goes to `/data/factory/prepare.log`. The + hook must be idempotent and finish within `timeoutMs` (default 30 minutes). +- **`deployKey`** clones over SSH with GitHub's published host key pinned. No `gh` + login happens and agents receive no Git credential. Use a read-only deploy key; + images without the bundle bootstrap need `openssh-client`. +- **Traces**: streaming deltas (`*Delta`, `*/delta`, rate-limit updates), the bulk + of all notifications, are no longer recorded; `item/completed` carries the same + content. `traceDeltas: true` records them again. `monitorIntervalMs` (default + 15000) sets how often each machine reads operator commands. + +Example models (vLLM 0.32 with `--enable-auto-tool-choice`, `--enable-prefix-caching`): + +| Model | GPU | Parsers | `contextWindow` | `autoCompactTokenLimit` | +|---|---|---|---|---| +| Qwen3.8-27B FP8 | 1 × H100 80 GB | `--tool-call-parser qwen3_xml --reasoning-parser qwen3` | 262144 | 80000–100000 | +| Nemotron 3.5 Super VL 120B-A12B NVFP4 | 1 × B300 | `--tool-call-parser qwen3_coder --reasoning-parser nemotron_v3 --trust-remote-code --mamba-cache-mode align` | 262144 | 80000 | + +Measured with operator-driven coding jobs: one H100 with the Qwen model kept +about 8 long conversations (histories averaging 55k tokens) at full speed; beyond that its +cache could not hold every history and prompt-cache hits fell from about 90% to +15%. One B300 with the Nemotron model carried a 190-agent run (10 machines × 19 +sessions): about 170 requests at once at the peak with none queued, GPU cache at +most 37% full, 84% of 464 million input tokens served from the prompt cache, and +12.5 million output tokens in 2 h 18 min for about $20 of GPU time. Its +multi-token prediction (`--speculative-config '{"method":"mtp","num_speculative_tokens":1}'`) +makes each agent faster up to about 128 agents; beyond that run without it. +Compacting at 80000 tokens kept agent contexts between 30k and 90k. + The cloud MCP adds identity/status, work claims and progress, authenticated messages/inbox, checkpoints, and supervisor Windows/Linux validation requests. The operator HTTP interface supports authenticated observations, trace export, diff --git a/package-lock.json b/package-lock.json index 3be7cf5..80ffcb7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "flujo-factory", - "version": "0.3.0", + "version": "0.4.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "flujo-factory", - "version": "0.3.0", + "version": "0.4.0", "license": "MIT", "dependencies": { "@modelcontextprotocol/sdk": "1.32.1" diff --git a/package.json b/package.json index 834ede0..fad3419 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "flujo-factory", - "version": "0.3.0", + "version": "0.4.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 e3a543c..b5e2758 100644 --- a/scripts/verify-public-sdk.mjs +++ b/scripts/verify-public-sdk.mjs @@ -86,3 +86,10 @@ assert.equal(updated.revision, 2); assert.equal(factory.getTemplate('package-smoke').goalContext, 'Deliver a synthetic case status'); assert.equal(factory.deleteTemplate('package-smoke', { expectedRevision: updated.revision }).deleted, true); assert.equal(factory.getTemplate('package-smoke'), null); + +// Cloud lite workers accept a self-hosted model provider and operator-driven jobs. +const providerPlan = sdk.validateCloudLitePlan({ runId: 'package-smoke', app: 'package-smoke-app', goal: 'Operator jobs', + modelProvider: { id: 'selfhosted', envKey: 'SELF_HOSTED_MODEL_KEY', baseUrls: ['https://model.example/v1'], contextWindow: 262144, autoCompactTokenLimit: 80000 }, + supervision: 'operator', sandbox: 'workspace-write', boardTools: false }); +assert.equal(providerPlan.modelProvider.autoCompactTokenLimit, 80000); +assert.throws(() => sdk.validateCloudLitePlan({ ...providerPlan, modelProvider: { ...providerPlan.modelProvider, baseUrls: ['http://model.example/v1'] } }), TypeError); diff --git a/src/cloud-lite/bootstrap.mjs b/src/cloud-lite/bootstrap.mjs index 1e61a86..41341c9 100644 --- a/src/cloud-lite/bootstrap.mjs +++ b/src/cloud-lite/bootstrap.mjs @@ -7,7 +7,7 @@ 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 ca-certificates curl procps python3 make g++ xz-utils bubblewrap +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 echo '86f268d81b898f3e144c5ecff2ad2fda2c5802e042d6195b72538fe0fa125057 /tmp/codex.tar.gz' | sha256sum -c - tar -xzf /tmp/codex.tar.gz -C /usr/local/bin @@ -25,7 +25,8 @@ node --input-type=module - <<'FACTORY_PIN_SOURCE' import { execFileSync } from 'node:child_process'; import { existsSync } from 'node:fs'; const config = JSON.parse(Buffer.from(process.env.FACTORY_CLOUD_BOOTSTRAP_B64, 'base64').toString()); -if (config.plan.repositoryRevision) { +// Deploy-key fleets clone and pin inside the runtime over SSH. +if (config.plan.repositoryRevision && !config.deployKey) { if (!/^[a-f0-9]{40}$/.test(config.plan.repositoryRevision)) throw new Error('Invalid source revision.'); const cwd = '/data/repository', env = { ...process.env, GH_TOKEN: config.githubToken, GIT_TERMINAL_PROMPT: '0' }; const run = (command,args,options={}) => execFileSync(command,args,{env,stdio:['ignore','pipe','pipe'],...options}); diff --git a/src/cloud-lite/fleet.mjs b/src/cloud-lite/fleet.mjs index 0abd49d..cda23cb 100644 --- a/src/cloud-lite/fleet.mjs +++ b/src/cloud-lite/fleet.mjs @@ -189,8 +189,13 @@ export class FactoryCloudLiteFleet { await this.stage('app.create', async () => { await this.fly(['apps', 'create', plan.app, '--org', plan.org ?? 'personal']); return { app: plan.app }; }); return this.snapshot(); } - async deploy(input, { image, imageSource, channel, githubToken, bundlePath, evidenceTokens } = {}) { - if (input.repository && !githubToken || !this.auth) throw new TypeError('Select a subscription auth broker and, for Git tasks, a GitHub credential.'); + async deploy(input, { image, imageSource, channel, githubToken, bundlePath, evidenceTokens, deployKey, providerKey } = {}) { + // A self-hosted model provider replaces the subscription broker; a read-only + // SSH deploy key can replace the GitHub token for repository access. + 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 (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.'); await this.prepare(input); const plan = this.state.plan; @@ -218,9 +223,9 @@ export class FactoryCloudLiteFleet { bootstrapPath = join(this.directory, 'bootstrap.sh'); writeFileSync(bootstrapPath, cloudLiteBootstrap(bundleDigest), { mode: 0o600 }); } const { credentials, operatorToken } = JSON.parse(readFileSync(join(this.directory, 'credentials.json'))); - const initialAuth = this.auth.read(); this.knownSecrets = [githubToken, initialAuth.accessToken, operatorToken, ...Object.values(credentials),...Object.values(evidenceTokens??{})]; + const initialAuth = providerMode ? {} : this.auth.read(); this.knownSecrets = [githubToken, initialAuth.accessToken, operatorToken, deployKey, providerKey, ...Object.values(credentials),...Object.values(evidenceTokens??{})]; await this.stage('app.credentials', async () => { - const config = Buffer.from(JSON.stringify({ plan, credentials, operatorToken, initialAuth, githubToken, evidenceTokens })).toString('base64'); + const config = Buffer.from(JSON.stringify({ plan, credentials, operatorToken, initialAuth, githubToken, evidenceTokens, deployKey, providerKey })).toString('base64'); this.knownSecrets.push(config); await this.fly(['secrets', 'import', '--app', plan.app, '--stage'], { input: `FACTORY_CLOUD_BOOTSTRAP_B64=${config}\n` }); return { staged: true }; @@ -319,7 +324,7 @@ export class FactoryCloudLiteFleet { // observations must never replace that controller's provisioning receipts. const observationPath = join(this.directory, 'observation.json'); const snapshot = await this.request('/operator/status'); - if (snapshot.refreshRequested) { const auth = await this.auth.refresh(); this.knownSecrets?.push(auth.accessToken); await this.request('/operator/auth', auth); } + if (snapshot.refreshRequested && this.auth) { const auth = await this.auth.refresh(); this.knownSecrets?.push(auth.accessToken); await this.request('/operator/auth', auth); } const retained = existsSync(observationPath) ? JSON.parse(readFileSync(observationPath, 'utf8')) : null; if (retained && retained.runId !== this.state.plan.runId) throw new Error('Observation journal belongs to another fleet.'); const observedWatermark = Number.isSafeInteger(snapshot.traceSequence) && snapshot.traceSequence >= 0 ? snapshot.traceSequence : null; diff --git a/src/cloud-lite/protocol.mjs b/src/cloud-lite/protocol.mjs index 3fd9d9d..0c5bc5d 100644 --- a/src/cloud-lite/protocol.mjs +++ b/src/cloud-lite/protocol.mjs @@ -26,8 +26,47 @@ export function validateCloudLitePlan(input) { if(plan.shutdown){ if(!plan.collection || typeof plan.shutdown.deadlineUtc!=='string' || !Number.isFinite(Date.parse(plan.shutdown.deadlineUtc)))throw new TypeError('Automatic deadline shutdown requires evidence collection and an explicit UTC deadline.'); } + if (plan.modelProvider !== undefined) validateModelProvider(plan.modelProvider); + 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(', ')}.`); + for (const name of ['boardTools', 'traceDeltas']) if (plan[name] !== undefined && typeof plan[name] !== 'boolean') throw new TypeError(`${name} must be true or false.`); + if (plan.workflow && (plan.supervision === 'operator' || plan.boardTools === false)) throw new TypeError('Verified workflows need automatic wakes and the board tools.'); + if (plan.monitorIntervalMs !== undefined && (!Number.isSafeInteger(plan.monitorIntervalMs) || plan.monitorIntervalMs < 1000)) throw new TypeError('monitorIntervalMs must be at least 1000.'); + if (plan.agentInstructions !== undefined && (typeof plan.agentInstructions !== 'string' || !plan.agentInstructions.trim())) throw new TypeError('agentInstructions must be non-empty text.'); + if (plan.prepare !== undefined) { + const prepare = plan.prepare; + if (!prepare || typeof prepare !== 'object' || !Array.isArray(prepare.argv) || !prepare.argv.length || prepare.argv.some(arg => typeof arg !== 'string' || !arg)) + throw new TypeError('prepare needs a non-empty argv list of strings.'); + if (prepare.spec !== undefined && (!prepare.spec || typeof prepare.spec !== 'object' || Array.isArray(prepare.spec))) throw new TypeError('prepare.spec must be an object.'); + if (prepare.timeoutMs !== undefined && (!Number.isSafeInteger(prepare.timeoutMs) || prepare.timeoutMs < 1000)) throw new TypeError('prepare.timeoutMs must be at least 1000.'); + } return plan; } +export const CODEX_SANDBOXES = ['danger-full-access', 'workspace-write', 'read-only']; +// Variables the runtime sets for the Codex process itself; a provider key must not replace them. +const RESERVED_ENVIRONMENT = new Set(['PATH', 'HOME', 'CODEX_HOME', 'GH_TOKEN', 'GIT_TERMINAL_PROMPT']); +const credentialFreeHttps = value => { + try { const url = new URL(value); return url.protocol === 'https:' && !url.username && !url.password && !url.search && !url.hash; } + catch { return false; } +}; +/** A self-hosted OpenAI-compatible (Responses API) model provider for the Codex workers. */ +export function validateModelProvider(provider) { + if (!provider || typeof provider !== 'object' || Array.isArray(provider)) throw new TypeError('modelProvider must be an object.'); + if (!/^[a-z][a-z0-9_-]{1,40}$/.test(provider.id ?? '')) throw new TypeError('modelProvider.id must be a short lowercase name.'); + if (!/^[A-Z][A-Z0-9_]{2,60}$/.test(provider.envKey ?? '') || RESERVED_ENVIRONMENT.has(provider.envKey)) throw new TypeError('modelProvider.envKey must name a dedicated environment variable.'); + if (!Array.isArray(provider.baseUrls) || !provider.baseUrls.length || provider.baseUrls.length > 64 || !provider.baseUrls.every(credentialFreeHttps)) + throw new TypeError('modelProvider.baseUrls must list https URLs without credentials, query or fragment.'); + for (const name of ['name', 'model']) if (provider[name] !== undefined && (typeof provider[name] !== 'string' || !provider[name].trim() || provider[name].length > 200)) throw new TypeError(`modelProvider.${name} must be short text.`); + for (const name of ['contextWindow', 'autoCompactTokenLimit', 'toolOutputTokenLimit']) { + if (provider[name] !== undefined && (!Number.isSafeInteger(provider[name]) || provider[name] < 1000)) throw new TypeError(`modelProvider.${name} must be a token count of at least 1000.`); + } + if (provider.contextWindow && provider.autoCompactTokenLimit && provider.autoCompactTokenLimit >= provider.contextWindow) throw new TypeError('modelProvider.autoCompactTokenLimit must be below contextWindow.'); + for (const name of ['requestMaxRetries', 'streamMaxRetries']) { + if (provider[name] !== undefined && (!Number.isSafeInteger(provider[name]) || provider[name] < 0 || provider[name] > 100)) throw new TypeError(`modelProvider.${name} must be between 0 and 100.`); + } + if (provider.streamIdleTimeoutMs !== undefined && (!Number.isSafeInteger(provider.streamIdleTimeoutMs) || provider.streamIdleTimeoutMs < 1000)) throw new TypeError('modelProvider.streamIdleTimeoutMs must be at least 1000.'); + return provider; +} export function agentInventory(plan) { return Array.from({ length: plan.machines }, (_, machine) => Array.from({ length: plan.agentsPerMachine }, (_, agent) => ({ id: `m${machine}-a${agent}`, machine: `m${machine}`, role: agent === 0 ? 'orchestrator' : 'subagent', diff --git a/src/cloud-lite/runtime.mjs b/src/cloud-lite/runtime.mjs index cf868bc..e7ecaea 100644 --- a/src/cloud-lite/runtime.mjs +++ b/src/cloud-lite/runtime.mjs @@ -39,7 +39,12 @@ export async function finalizeNativeStopCommand({ prove, publish, flush, acknowl return true; } export async function runCloudLiteMachine(config) { - const { plan, index, credentials, operatorToken, initialAuth, githubToken, evidenceTokens } = config; + const { plan, index, credentials, operatorToken, initialAuth = {}, githubToken, evidenceTokens, providerKey, deployKey } = 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 operatorSupervised = plan.supervision === 'operator'; + const extraSecrets = [providerKey, deployKey].filter(value => typeof value === 'string' && value.length >= 8); mkdirSync('/data/factory', { recursive: true, mode: 0o700 }); let coordinator, evidenceProcess, evidenceSealing=false, evidenceFinal=false, deadlineRequested=false, sealRequested=false; const stopIds=new Map(); @@ -90,37 +95,59 @@ export async function runCloudLiteMachine(config) { const repositoryDirectory = plan.repository ? '/data/repository' : '/data/workspace'; if(!plan.repository)mkdirSync(repositoryDirectory,{recursive:true,mode:0o700}); const gitEnv = { ...process.env, GH_TOKEN: githubToken, GIT_TERMINAL_PROMPT: '0' }; + delete gitEnv.FACTORY_CLOUD_BOOTSTRAP_B64; + if (deployKey) { + // Read-only repository access: an SSH deploy key and GitHub's published host key. + writeFileSync('/data/factory/deploy_key', deployKey.endsWith('\n') ? deployKey : deployKey + '\n', { mode: 0o600 }); + writeFileSync('/data/factory/known_hosts', GITHUB_KNOWN_HOSTS, { mode: 0o600 }); + gitEnv.GIT_SSH_COMMAND = 'ssh -i /data/factory/deploy_key -o IdentitiesOnly=yes -o UserKnownHostsFile=/data/factory/known_hosts -o StrictHostKeyChecking=yes'; + delete gitEnv.GH_TOKEN; + } + const cloneUrl = repository => deployKey ? `git@github.com:${repository}.git` : `https://github.com/${repository}.git`; const execute = promisify(execFile); const run = async (command, args, cwd) => { try { return (await execute(command, args, { cwd, env: gitEnv, encoding: 'utf8', maxBuffer: 16 * 1024 * 1024 })).stdout; } catch (error) { throw new Error(`${command} failed (${error.code}).`); } }; - if (plan.repository && !existsSync(join(repositoryDirectory, '.git'))) await run('git', ['clone', `https://github.com/${plan.repository}.git`, repositoryDirectory]); - if (plan.repositoryRevision) { + // A completed prepare hook may remove repository metadata on purpose; a restarted + // machine then reuses the prepared workspaces instead of cloning again. + const preparedFile = '/data/factory/prepare-result.json'; + const alreadyPrepared = Boolean(plan.prepare) && existsSync(preparedFile); + if (plan.repository && !alreadyPrepared && !existsSync(join(repositoryDirectory, '.git'))) await run('git', ['clone', cloneUrl(plan.repository), repositoryDirectory]); + if (plan.repositoryRevision && !alreadyPrepared) { await run('git', ['fetch', 'origin', plan.repositoryRevision], repositoryDirectory); const pinned = (await run('git', ['rev-parse', `${plan.repositoryRevision}^{commit}`], repositoryDirectory)).trim(); if (pinned !== plan.repositoryRevision) throw new Error('Repository snapshot identity mismatch.'); } - if(plan.repository){ + if(plan.repository && !alreadyPrepared){ await run('git', ['config', 'user.name', 'FACTORY swarm'], repositoryDirectory); await run('git', ['config', 'user.email', 'flujo.com.co@gmail.com'], repositoryDirectory); - await run('gh', ['auth', 'setup-git'], repositoryDirectory); + if (githubToken && !deployKey) await run('gh', ['auth', 'setup-git'], repositoryDirectory); + // Agents that work in prepared folders (no per-agent worktree) need the pinned source checked out. + if (plan.prepare && plan.repositoryRevision) await run('git', ['checkout', '--detach', '--force', plan.repositoryRevision], repositoryDirectory); + } + // The prepare hook builds each agent's workspace from the pinned source once per machine. + let prepared = alreadyPrepared ? JSON.parse(readFileSync(preparedFile, 'utf8')) : null; + if (plan.prepare && !prepared) { + 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 }); // A fresh native harness has only the explicitly configured FACTORY MCP server. - writeFileSync(join(home, 'config.toml'), 'cli_auth_credentials_store="ephemeral"\nforced_login_method="chatgpt"\n', { mode: 0o600 }); + 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(); for (const agent of bootstrap.agents) { const position = await request('/trace-position', {}, agent.token); traces.set(agent.id, new CloudLiteTrace({ file: `/data/factory/${agent.id}.jsonl`, url: boardUrl, token: agent.token, remoteSequence: position.sequence, - secrets: [githubToken, initialAuth.accessToken, operatorToken, ...Object.values(credentials),...Object.values(evidenceTokens??{})] })); + 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: { PATH: process.env.PATH, HOME: '/data', CODEX_HOME: home, GH_TOKEN: githubToken, GIT_TERMINAL_PROMPT: '0' }, cwd: repositoryDirectory, + 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') { + 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); @@ -132,7 +159,8 @@ export async function runCloudLiteMachine(config) { rpc.on('notification', value => { const threadId = value.params?.threadId ?? value.params?.thread?.id; const agent = threadAgents.get(threadId) ?? rootId; - traces.get(agent).record(value.method, value.params ?? {}); + // Streaming deltas are most notifications and carry no state; item/completed has the content. + if (plan.traceDeltas === true || !isStreamingDelta(value.method)) traces.get(agent).record(value.method, value.params ?? {}); if (value.method === 'thread/started' || value.method === 'thread/resumed' || value.method === 'thread/status/changed' && value.params?.status?.type === 'active') stopControllers.get(threadId)?.invalidateReceipts(); const turn = value.params?.turn ?? {}; @@ -167,25 +195,27 @@ export async function runCloudLiteMachine(config) { } }, 1000); await rpc.initialize(); - await rpc.request('account/login/start', { type: 'chatgptAuthTokens', ...bootstrap.auth }); + // 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 })) }); const proxy = fileURLToPath(new URL('../../bin/factory-cloud-mcp.mjs', import.meta.url)); for (const agent of bootstrap.agents) { - const directory = join('/data/worktrees', agent.id); mkdirSync('/data/worktrees', { recursive: true }); + // With a prepare hook, each agent works in the folder the hook prepared for its slot. + const slot = prepared?.agents?.[Number(agent.id.split('-a')[1])]; + const directory = slot?.cwd ?? join('/data/worktrees', agent.id); mkdirSync('/data/worktrees', { recursive: true }); if (!existsSync(directory)) { - if(plan.repository)await run('git', ['worktree', 'add', '--detach', directory, plan.repositoryRevision ?? 'HEAD'], repositoryDirectory); + if(plan.repository && !slot)await run('git', ['worktree', 'add', '--detach', directory, plan.repositoryRevision ?? 'HEAD'], repositoryDirectory); else mkdirSync(directory,{recursive:true,mode:0o700}); } const sessionFile = `/data/factory/${agent.id}.session.json`; - const configOverride = { mcp_servers: { factory: { command: process.execPath, args: [proxy], env: { FACTORY_BOARD_URL: boardUrl, FACTORY_BOARD_TOKEN: agent.token, FACTORY_WORKTREE: directory }, required: true } }, - 'features.apps': false, 'features.plugins': false, 'features.multi_agent': false, - 'model_reasoning_effort': plan.reasoningEffort }; - const instructions = `You are ${agent.id}, ${agent.role}, on machine ${machine}. Your orchestrator is ${rootId}. Team: ${bootstrap.agents.map(x => x.id).join(', ')}.\nUse FACTORY identity/status/claim/message/inbox/checkpoint tools. Claim a stable task: before work (issue:/pr: are available for repository work). One team owns each item. Orchestrators coordinate across teams; workers communicate within their team.\n${agent.role === 'orchestrator' ? 'Actively decompose and complete the original goal. Delegate useful work, resolve dependencies, and integrate the outputs into one coherent final deliverable. Keep unmet acceptance requirements open.' : 'Produce or independently review your assigned deliverable and report evidence and remaining requirements.'}\nRetain native tools. Respect task-specific instructions and validation/publication policy. Work can be research, analysis, documents, data, operations or software; do not invent a repository or software checks for unrelated tasks. Save deliverables under your workspace or /data/results. Record progress and evidence. Credentials must never enter deliverables, messages or traces.\nThe original user goal follows unchanged.`; + const { sandbox, config: configOverride } = nativeAgentConfig({ plan, slot, directory, + board: { command: process.execPath, args: [proxy], env: { FACTORY_BOARD_URL: boardUrl, FACTORY_BOARD_TOKEN: agent.token } } }); + const instructions = slot?.instructions ?? plan.agentInstructions ?? `You are ${agent.id}, ${agent.role}, on machine ${machine}. Your orchestrator is ${rootId}. Team: ${bootstrap.agents.map(x => x.id).join(', ')}.\nUse FACTORY identity/status/claim/message/inbox/checkpoint tools. Claim a stable task: before work (issue:/pr: are available for repository work). One team owns each item. Orchestrators coordinate across teams; workers communicate within their team.\n${agent.role === 'orchestrator' ? 'Actively decompose and complete the original goal. Delegate useful work, resolve dependencies, and integrate the outputs into one coherent final deliverable. Keep unmet acceptance requirements open.' : 'Produce or independently review your assigned deliverable and report evidence and remaining requirements.'}\nRetain native tools. Respect task-specific instructions and validation/publication policy. Work can be research, analysis, documents, data, operations or software; do not invent a repository or software checks for unrelated tasks. Save deliverables under your workspace or /data/results. Record progress and evidence. Credentials must never enter deliverables, messages or traces.\nThe original user goal follows unchanged.`; let result; let recovering = existsSync(sessionFile); let unavailableSession; 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: 'danger-full-access' }); + result = await rpc.request('thread/resume', { threadId: retained.threadId, cwd: directory, config: configOverride, approvalPolicy: 'never', sandbox }); } catch (error) { if (error.code !== -32600 || error.message !== `no rollout found for thread id ${retained.threadId}`) throw error; unavailableSession = retained.threadId; recovering = false; @@ -196,7 +226,7 @@ export async function runCloudLiteMachine(config) { } } if (!result) { - result = await rpc.request('thread/start', { model: agent.role === 'orchestrator' ? plan.model : plan.subagentModel, cwd: directory, approvalPolicy: 'never', sandbox: 'danger-full-access', config: configOverride, + result = await rpc.request('thread/start', { model: agent.role === 'orchestrator' ? plan.model : plan.subagentModel, cwd: directory, approvalPolicy: 'never', sandbox, config: configOverride, developerInstructions: instructions, serviceName: `factory.${plan.runId}.${agent.id}`, historyMode: 'legacy' }); const fd = await open(sessionFile, 'w', 0o600); try { await fd.writeFile(JSON.stringify({ threadId: result.thread.id, agentId: agent.id })); await fd.sync(); } finally { await fd.close(); } @@ -214,7 +244,8 @@ export async function runCloudLiteMachine(config) { const threadEvent = recovering ? 'thread.resumed' : 'thread.started'; traces.get(agent.id).record('native.' + threadEvent, { threadId, ...(observedThreadStatus ? { threadStatus: nativeThreadStatus(observedThreadStatus) } : {}) }); - if (current?.thread.status?.type !== 'active') { + // Operator-supervised agents start idle; every turn comes from a durable operator message. + if (!operatorSupervised && current?.thread.status?.type !== 'active') { const recoveryNote = unavailableSession ? `The prior prepared session ${unavailableSession} has no persisted rollout. Reconcile your existing FACTORY claims, inbox, checkout and GitHub state before continuing; do not replay previous delivery operations.\n` : ''; const workflowNote = plan.workflow ? `\nYour team has exactly one independent reviewer: ${machine}-a${plan.agentsPerMachine-1}. Other workers produce deliverables. Orchestrators use factory_assign to retain the original goal/acceptance as a parent task and delegate child tasks; the SDK selects your reviewer. Choose task-appropriate executable gates or explicitly explain a review-only policy when objective checks do not apply. The orchestrator integrates accepted child outputs and submits the parent deliverable. Use factory_submit_workspace with selected relative paths, a stable submissionId and a summary; the SDK freezes files, transfers chunks and renews the claim. No Git metadata is required. Reviewers inspect the frozen packet and artifacts, then use factory_review; never edit the candidate or approve their own work. Semantic review, executable checks and delivery permission are separate. Revise rejected candidates. Continue useful unblocked work while checks are unavailable. Read your inbox/next actions and finish idle turns; the SDK wakes you on relevant events.\n` : ''; const input = recoveryNote + workflowNote + (agent.role === 'orchestrator' ? plan.goal : `Work under ${rootId} toward this goal. Check inbox and shared ownership first:\n${plan.goal}`); @@ -318,7 +349,7 @@ export async function runCloudLiteMachine(config) { const newWorkflowEvent = plan.workflow && (workflowVersions.get(threadId) ?? 0) > (state.workflowVersion ?? 0); const pendingWorkflowWake = plan.workflow && ['unknown','rejected'].includes(workflowWakes.get(threadId)?.intent?.status); const leaderFollowup = !plan.workflow && agentId === rootId && Date.now() - state.completed >= 120000; - if (stopping || evidenceSealing || stoppingThreads.has(threadId) || state.status !== 'completed' || !newMail && !leaderFollowup && !newWorkflowEvent && !pendingWorkflowWake) continue; + if (operatorSupervised || stopping || evidenceSealing || stoppingThreads.has(threadId) || state.status !== 'completed' || !newMail && !leaderFollowup && !newWorkflowEvent && !pendingWorkflowWake) continue; try { if(plan.workflow){ const wake=await workflowWakes.get(threadId).wake({trigger:`${inboxVersions.get(threadId)??0}:${workflowVersions.get(threadId)??0}`, @@ -330,7 +361,7 @@ export async function runCloudLiteMachine(config) { } } catch (error) { rootTrace.record('monitor.pending', { message: error.message }); } finally { monitoringBusy = false; } - }, 15000); + }, plan.monitorIntervalMs ?? 15000); function stop() { return stopPromise ??= (async () => { stopping = true; clearInterval(timer); clearInterval(monitoring); for (const threadId of threadAgents.keys()) { try { const turnId = turns.get(threadId); if (turnId) await rpc.request('turn/interrupt', { threadId, turnId }, 10000); } catch { /* Retained native state remains authoritative. */ } } @@ -341,3 +372,110 @@ export async function runCloudLiteMachine(config) { process.once('SIGTERM', () => void stop()); process.once('SIGINT', () => void stop()); return { rpc, traces, stop, threadAgents }; } + +// GitHub's published SSH host key (https://api.github.com/meta, ssh_keys). +const GITHUB_KNOWN_HOSTS = 'github.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl\n'; + +/** Token-by-token notifications; their content also arrives in item/completed. */ +export function isStreamingDelta(method) { + return typeof method === 'string' && (/[dD]elta$/.test(method) || method === 'account/rateLimits/updated'); +} + +/** The Codex app-server environment. A provider key reaches only this process. */ +export function codexEnvironment({ plan, home, githubToken, deployKey, providerKey, path = process.env.PATH }) { + return { PATH: path, HOME: '/data', CODEX_HOME: home, GIT_TERMINAL_PROMPT: '0', + ...(githubToken && !deployKey ? { GH_TOKEN: githubToken } : {}), + ...(plan.modelProvider ? { [plan.modelProvider.envKey]: providerKey } : {}) }; +} + +/** Sandbox and per-thread configuration override for one native agent session. */ +export function nativeAgentConfig({ plan, slot, directory, board }) { + const sandbox = plan.sandbox ?? 'danger-full-access'; + const config = { + ...(plan.boardTools === false ? {} : { mcp_servers: { factory: { ...board, env: { ...board.env, FACTORY_WORKTREE: directory }, required: true } } }), + 'features.apps': false, 'features.plugins': false, 'features.multi_agent': false, + 'model_reasoning_effort': plan.reasoningEffort, + // Agent commands may write only their workspace and the listed roots, with no network. + ...(sandbox === 'workspace-write' ? { 'sandbox_workspace_write.writable_roots': slot?.writableRoots ?? [], 'sandbox_workspace_write.network_access': false } : {}), + }; + return { sandbox, config }; +} + +const tomlString = value => JSON.stringify(String(value)); + +/** Codex configuration for a self-hosted OpenAI-compatible (Responses API) model provider. */ +export function providerConfigToml(plan, index) { + const p = plan.modelProvider; + return [ + `model = ${tomlString(p.model ?? plan.model)}`, + `model_provider = ${tomlString(p.id)}`, + `model_reasoning_effort = ${tomlString(plan.reasoningEffort)}`, + ...(p.contextWindow ? [`model_context_window = ${Number(p.contextWindow)}`] : []), + ...(p.autoCompactTokenLimit ? [`model_auto_compact_token_limit = ${Number(p.autoCompactTokenLimit)}`] : []), + ...(p.toolOutputTokenLimit ? [`tool_output_token_limit = ${Number(p.toolOutputTokenLimit)}`] : []), + 'web_search = "disabled"', + 'check_for_update_on_startup = false', + 'cli_auth_credentials_store = "ephemeral"', + '', + '[features]', + 'multi_agent = false', + 'goals = false', + 'apps = false', + 'plugins = false', + 'memories = false', + '', + '[skills]', + 'include_instructions = false', + '', + // Agents' shell commands get only the basic variables, never the provider key. + '[shell_environment_policy]', + 'inherit = "core"', + '', + `[model_providers.${p.id}]`, + `name = ${tomlString(p.name ?? p.id)}`, + // Machines are spread over the endpoints, so each keeps its prompt cache on one server. + `base_url = ${tomlString(p.baseUrls[index % p.baseUrls.length])}`, + `env_key = ${tomlString(p.envKey)}`, + 'wire_api = "responses"', + `request_max_retries = ${Number(p.requestMaxRetries ?? 8)}`, + `stream_max_retries = ${Number(p.streamMaxRetries ?? 10)}`, + `stream_idle_timeout_ms = ${Number(p.streamIdleTimeoutMs ?? 900000)}`, + '', + ].join('\n'); +} + +/** + * Runs the plan's prepare command once per machine. It must be idempotent: a machine that + * restarts before recording the result runs it again. It receives FACTORY_MACHINE_INDEX and + * FACTORY_PREPARE_SPEC (a JSON file) and prints one JSON line + * { agents: [{ cwd, writableRoots?, instructions? }, ...] } indexed by agent slot (a0, a1, ...). + */ +export async function runPrepareHook({ plan, index, repositoryDirectory, env, stateDirectory = '/data/factory' }) { + const specFile = join(stateDirectory, 'prepare-spec.json'); + const logFile = join(stateDirectory, 'prepare.log'); + writeFileSync(specFile, JSON.stringify(plan.prepare.spec ?? {}), { mode: 0o600 }); + const [command, ...args] = plan.prepare.argv; + const childEnv = { PATH: env.PATH, HOME: '/data', FACTORY_MACHINE_INDEX: String(index), FACTORY_PREPARE_SPEC: specFile, + ...(env.GIT_SSH_COMMAND ? { GIT_SSH_COMMAND: env.GIT_SSH_COMMAND } : {}), + ...(env.SystemRoot ? { SystemRoot: env.SystemRoot } : {}) }; + let stdout = '', stderr = ''; + try { + ({ stdout, stderr } = await promisify(execFile)(command, args, { cwd: repositoryDirectory, env: childEnv, encoding: 'utf8', + maxBuffer: 64 * 1024 * 1024, timeout: plan.prepare.timeoutMs ?? 30 * 60000 })); + } catch (error) { + writeFileSync(logFile, `${error.stdout ?? ''}\n--- stderr ---\n${error.stderr ?? ''}\n--- error ---\n${error.message}\n`, { mode: 0o600 }); + throw new Error(`Prepare hook failed (${error.code ?? 'error'}); see ${logFile}.`); + } + writeFileSync(logFile, `${stdout}\n--- stderr ---\n${stderr}\n`, { mode: 0o600 }); + const line = stdout.trim().split('\n').reverse().find(value => value.trim().startsWith('{')); + let result = null; + try { result = line ? JSON.parse(line) : null; } catch { result = null; } + const absolute = value => typeof value === 'string' && value.startsWith('/'); + if (!Array.isArray(result?.agents) || result.agents.length !== plan.agentsPerMachine + || result.agents.some(agent => !absolute(agent?.cwd) + || agent.writableRoots !== undefined && (!Array.isArray(agent.writableRoots) || !agent.writableRoots.every(absolute)) + || agent.instructions !== undefined && (typeof agent.instructions !== 'string' || !agent.instructions.trim()))) { + throw new Error('Prepare hook must print one JSON line with an absolute cwd for every agent slot.'); + } + return result; +} diff --git a/test/cloud-lite-provider.test.mjs b/test/cloud-lite-provider.test.mjs new file mode 100644 index 0000000..4ca5c96 --- /dev/null +++ b/test/cloud-lite-provider.test.mjs @@ -0,0 +1,164 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, readFileSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { validateCloudLitePlan, secret } from '../src/cloud-lite/protocol.mjs'; +import { cloudLiteBootstrap } from '../src/cloud-lite/bootstrap.mjs'; +import { FactoryCloudLiteFleet } from '../src/cloud-lite/fleet.mjs'; +import { codexEnvironment, isStreamingDelta, nativeAgentConfig, providerConfigToml, runPrepareHook } from '../src/cloud-lite/runtime.mjs'; + +const fresh = () => mkdtempSync(join(tmpdir(), 'factory-cloud-lite-provider-')); +// A self-hosted model with its real context window and an earlier compaction point. +const provider = () => ({ id: 'selfhosted', name: 'Self-hosted model', envKey: 'SELF_HOSTED_MODEL_KEY', model: 'served-model', + baseUrls: ['https://model-a.example.test/v1', 'https://model-b.example.test/v1'], contextWindow: 262144, autoCompactTokenLimit: 80000 }); +const plan = (extra = {}) => validateCloudLitePlan({ runId: 'provider-test', app: 'factory-provider-test', repository: 'example/project', + goal: 'Work on operator jobs', machines: 2, agentsPerMachine: 3, model: 'served-model', subagentModel: 'served-model', modelProvider: provider(), ...extra }); +// Assembled at run time so no key-shaped literal appears in the source. +const fixtureDeployKey = () => ['-----BEGIN', 'OPENSSH PRIVATE KEY-----\nfixture-only\n-----END', 'OPENSSH PRIVATE KEY-----\n'].join(' '); + +test('plan validation accepts a self-hosted provider and operator settings and rejects unsafe values', () => { + const valid = plan({ supervision: 'operator', sandbox: 'workspace-write', boardTools: false, monitorIntervalMs: 4000, traceDeltas: false, + agentInstructions: 'Wait for operator jobs.', prepare: { argv: ['node', 'prepare.mjs'], spec: { team: 1 }, timeoutMs: 60000 } }); + assert.equal(valid.modelProvider.contextWindow, 262144); + assert.equal('supervision' in plan(), false, 'new settings add no defaults to existing fleet bindings'); + const rejects = [ + { modelProvider: { ...provider(), baseUrls: ['http://model.example.test/v1'] } }, + { modelProvider: { ...provider(), baseUrls: ['https://user:pass@model.example.test/v1'] } }, + { modelProvider: { ...provider(), baseUrls: ['https://model.example.test/v1?key=x'] } }, + { modelProvider: { ...provider(), baseUrls: [] } }, + { modelProvider: { ...provider(), envKey: 'PATH' } }, + { modelProvider: { ...provider(), envKey: 'lower_case' } }, + { modelProvider: { ...provider(), id: 'Bad Id' } }, + { modelProvider: { ...provider(), contextWindow: 999 } }, + { modelProvider: { ...provider(), autoCompactTokenLimit: 262144 } }, + { modelProvider: { ...provider(), requestMaxRetries: 101 } }, + { supervision: 'manual' }, { sandbox: 'none' }, { boardTools: 'no' }, { traceDeltas: 1 }, { monitorIntervalMs: 999 }, + { agentInstructions: ' ' }, { prepare: { argv: [] } }, { prepare: { argv: ['node', 7] } }, { prepare: { argv: ['node'], spec: [] } }, + { workflow: 'verified-v1', supervision: 'operator' }, { workflow: 'verified-v1', boardTools: false }, + ]; + for (const extra of rejects) assert.throws(() => plan(extra), TypeError, JSON.stringify(extra)); +}); + +test('provider Codex configuration spreads machines over endpoints and never contains the key', () => { + const p = plan({ reasoningEffort: 'high' }); const key = secret(); + const first = providerConfigToml(p, 0); const second = providerConfigToml(p, 1); const third = providerConfigToml(p, 2); + assert.match(first, /^model = "served-model"$/m); + assert.match(first, /^model_provider = "selfhosted"$/m); + assert.match(first, /^model_reasoning_effort = "high"$/m); + assert.match(first, /^model_context_window = 262144$/m); + assert.match(first, /^model_auto_compact_token_limit = 80000$/m); + assert.match(first, /^\[model_providers\.selfhosted\]$/m); + assert.match(first, /^env_key = "SELF_HOSTED_MODEL_KEY"$/m); + assert.match(first, /^wire_api = "responses"$/m); + assert.match(first, /^\[shell_environment_policy\]\ninherit = "core"$/m); + assert.doesNotMatch(first, /forced_login_method|chatgpt/); + assert.match(first, /base_url = "https:\/\/model-a\.example\.test\/v1"/); + assert.match(second, /base_url = "https:\/\/model-b\.example\.test\/v1"/); + assert.match(third, /base_url = "https:\/\/model-a\.example\.test\/v1"/); + assert.equal(first.includes(key), false); + const env = codexEnvironment({ plan: p, home: '/data/codex', providerKey: key, path: '/usr/bin' }); + assert.deepEqual(env, { PATH: '/usr/bin', HOME: '/data', CODEX_HOME: '/data/codex', GIT_TERMINAL_PROMPT: '0', SELF_HOSTED_MODEL_KEY: key }); + const githubToken = secret(); + assert.equal(codexEnvironment({ plan: plan({ modelProvider: undefined }), home: '/h', githubToken }).GH_TOKEN, githubToken); + assert.equal('GH_TOKEN' in codexEnvironment({ plan: p, home: '/h', githubToken, deployKey: fixtureDeployKey(), providerKey: key }), false); +}); + +test('agent sessions keep board tools and full access by default and narrow them on request', () => { + const board = { command: 'node', args: ['proxy.mjs'], env: { FACTORY_BOARD_URL: 'http://board', FACTORY_BOARD_TOKEN: 'agent-token' } }; + const standard = nativeAgentConfig({ plan: plan({ modelProvider: undefined }), directory: '/data/worktrees/m0-a1', board }); + assert.equal(standard.sandbox, 'danger-full-access'); + assert.deepEqual(standard.config.mcp_servers.factory, { ...board, env: { ...board.env, FACTORY_WORKTREE: '/data/worktrees/m0-a1' }, required: true }); + assert.equal('sandbox_workspace_write.writable_roots' in standard.config, false); + const narrowed = nativeAgentConfig({ plan: plan({ supervision: 'operator', sandbox: 'workspace-write', boardTools: false }), + slot: { cwd: '/data/work/a1', writableRoots: ['/data/scratch'] }, directory: '/data/work/a1', board }); + assert.equal(narrowed.sandbox, 'workspace-write'); + assert.equal('mcp_servers' in narrowed.config, false); + assert.deepEqual(narrowed.config['sandbox_workspace_write.writable_roots'], ['/data/scratch']); + assert.equal(narrowed.config['sandbox_workspace_write.network_access'], false); + assert.equal(narrowed.config['features.multi_agent'], false); +}); + +test('streaming deltas are recognized so traces keep only state-bearing notifications', () => { + for (const method of ['item/agentMessage/delta', 'item/reasoning/textDelta', 'item/commandExecution/outputDelta', 'account/rateLimits/updated']) assert.equal(isStreamingDelta(method), true, method); + for (const method of ['item/completed', 'turn/completed', 'thread/started', 'thread/tokenUsage/updated', undefined]) assert.equal(isStreamingDelta(method), false, String(method)); +}); + +test('prepare hook receives its spec and machine index and must return an absolute workspace per agent', async () => { + const stateDirectory = fresh(); const repositoryDirectory = fresh(); + const env = { PATH: process.env.PATH, SystemRoot: process.env.SystemRoot, FACTORY_CLOUD_BOOTSTRAP_B64: 'must-not-reach-the-hook' }; + const script = `const fs=require('node:fs');const spec=JSON.parse(fs.readFileSync(process.env.FACTORY_PREPARE_SPEC,'utf8')); +console.log('preparing', process.env.FACTORY_BOOTSTRAP_PROBE ?? process.env.FACTORY_CLOUD_BOOTSTRAP_B64 ?? 'clean'); +console.log(JSON.stringify({agents:Array.from({length:spec.slots},(_,i)=>({cwd:'/work/m'+process.env.FACTORY_MACHINE_INDEX+'/a'+i,writableRoots:['/scratch'],instructions:spec.team+' member '+i}))}));`; + const p = plan({ prepare: { argv: [process.execPath, '-e', script], spec: { slots: 3, team: 'blue' } } }); + const result = await runPrepareHook({ plan: p, index: 1, repositoryDirectory, env, stateDirectory }); + assert.deepEqual(result.agents.map(agent => agent.cwd), ['/work/m1/a0', '/work/m1/a1', '/work/m1/a2']); + assert.equal(result.agents[2].instructions, 'blue member 2'); + assert.deepEqual(JSON.parse(readFileSync(join(stateDirectory, 'prepare-spec.json'), 'utf8')), { slots: 3, team: 'blue' }); + assert.match(readFileSync(join(stateDirectory, 'prepare.log'), 'utf8'), /preparing clean/); + const wrongCount = plan({ prepare: { argv: [process.execPath, '-e', script], spec: { slots: 2, team: 'blue' } } }); + await assert.rejects(runPrepareHook({ plan: wrongCount, index: 0, repositoryDirectory, env, stateDirectory }), /absolute cwd for every agent slot/); + const relative = plan({ prepare: { argv: [process.execPath, '-e', 'console.log(JSON.stringify({agents:[{cwd:"a"},{cwd:"b"},{cwd:"c"}]}))'] } }); + await assert.rejects(runPrepareHook({ plan: relative, index: 0, repositoryDirectory, env, stateDirectory }), /absolute cwd/); + const failing = plan({ prepare: { argv: [process.execPath, '-e', 'console.error("broken input"); process.exit(3)'] } }); + await assert.rejects(runPrepareHook({ plan: failing, index: 0, repositoryDirectory, env, stateDirectory }), /Prepare hook failed/); + assert.match(readFileSync(join(stateDirectory, 'prepare.log'), 'utf8'), /broken input/); +}); + +function providerInfrastructure(calls, { failVolumes } = {}) { + const machines = []; + return { + run: async (args, options = {}) => { + calls.push({ args, input: options.input }); + if (args[0] === 'volumes') { if (failVolumes) throw new Error(failVolumes()); return JSON.stringify({ id: `vol_${machines.length}` }); } + if (args[0] === 'machine') return JSON.stringify(machines); + return ''; + }, + buildMachineConfig: () => ({ init: { cmd: ['upstream-runtime'] }, services: [] }), + createMachine: async ({ name, config }) => { const machine = { id: `machine-${machines.length}`, name, config }; machines.push(machine); return machine; }, + }; +} +const image = 'ghcr.io/example/worker@sha256:' + 'a'.repeat(64); + +test('a model provider replaces the subscription broker and a deploy key replaces the GitHub token', async () => { + const calls = []; const providerKey = secret(); const deployKey = fixtureDeployKey(); + const fleet = new FactoryCloudLiteFleet({ directory: fresh(), infrastructure: providerInfrastructure(calls) }); + await assert.rejects(fleet.deploy(plan(), { image, deployKey }), /needs its API key/); + await assert.rejects(new FactoryCloudLiteFleet({ directory: fresh(), infrastructure: providerInfrastructure([]) }) + .deploy(plan({ modelProvider: undefined }), { image, deployKey }), /subscription auth broker/); + await assert.rejects(fleet.deploy(plan(), { image, providerKey, deployKey: 'not a key' }), /private SSH key/); + await assert.rejects(fleet.deploy(plan({ repository: undefined }), { image, providerKey, deployKey }), /private SSH key/); + const result = await fleet.deploy(plan(), { image, providerKey, deployKey }); + assert.equal(result.machines.length, 2); + 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, {}); + assert.equal(config.githubToken, undefined); + assert.equal(config.providerKey, providerKey); + assert.equal(config.deployKey, deployKey); + assert.equal(config.plan.modelProvider.envKey, 'SELF_HOSTED_MODEL_KEY'); + const journal = readFileSync(join(fleet.directory, 'fleet.json'), 'utf8'); + assert.equal(journal.includes(providerKey), false); assert.equal(journal.includes('fixture-only'), false); + const bootstrap = cloudLiteBootstrap('b'.repeat(64)); + assert.match(bootstrap, /openssh-client/); + assert.match(bootstrap, /repositoryRevision && !config\.deployKey/); +}); + +test('provider and deploy keys are redacted from uncertain provider outcomes', async () => { + const providerKey = secret(); const deployKey = fixtureDeployKey(); const directory = fresh(); + const fleet = new FactoryCloudLiteFleet({ directory, infrastructure: providerInfrastructure([], { failVolumes: () => `volume failed near ${providerKey} and ${deployKey}` }) }); + await assert.rejects(fleet.deploy(plan(), { image, providerKey, deployKey })); + const journal = readFileSync(join(directory, 'fleet.json'), 'utf8'); + assert.match(journal, /\[REDACTED\]/); + assert.equal(journal.includes(providerKey), false); assert.equal(journal.includes('fixture-only'), false); +}); + +test('a provider fleet observes without a subscription refresh', async () => { + const directory = fresh(); + const fleet = new FactoryCloudLiteFleet({ directory }); fleet.state = { plan: plan(), machines: [], operations: {} }; fleet.save(); + writeFileSync(join(directory, 'observation.json'), '{broken'); + const order = []; + fleet.request = async path => { order.push(path); if (path === '/operator/status') return { traceSequence: 0, agents: [], refreshRequested: true }; throw new Error(`Unexpected ${path}`); }; + await assert.rejects(fleet.observe(), SyntaxError); + assert.deepEqual(order, ['/operator/status']); +}); From 7a36b4f00107e61acb200664602f22b48b363305 Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:38:07 -0500 Subject: [PATCH 2/2] Refresh transitive dependencies for current security advisories npm audit fix within existing ranges: body-parser 2.4.0, zod 4.6.7, fast-uri 3.1.9, ip-address 10.7.6, qs 6.17.1. The production audit gate (high severity) failed on these new advisories; it now reports none. --- package-lock.json | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/package-lock.json b/package-lock.json index 80ffcb7..9e835d2 100644 --- a/package-lock.json +++ b/package-lock.json @@ -120,18 +120,18 @@ } }, "node_modules/body-parser": { - "version": "2.3.0", - "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-2.3.0.tgz", - "integrity": "sha512-2cGmJupaNgg+QUwVLAucDuWuoMZ6EX9iHDRswZ5lsNYEmwPaRknMPCLZz07yTzVq/83p4o/wzbDZbBrTvGGTIw==", + "version": "2.4.0", + "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-2.4.0.tgz", + "integrity": "sha512-Ifk1KcLrxaGt1occQagQNFNNOqRt1LFuehEnHHzB8Z6cRgIxRU1TBaiIsUiCdA+gy5PL7Y1IX31DNKkRmxtXjQ==", "license": "MIT", "dependencies": { "bytes": "^3.1.2", - "content-type": "^2.0.0", + "content-type": "^2.1.0", "debug": "^4.4.3", "http-errors": "^2.0.1", - "iconv-lite": "^0.7.2", + "iconv-lite": "^0.7.3", "on-finished": "^2.4.1", - "qs": "^6.15.2", + "qs": "^6.16.1", "raw-body": "^3.0.2", "type-is": "^2.1.0" }, @@ -455,9 +455,9 @@ "license": "MIT" }, "node_modules/fast-uri": { - "version": "3.1.8", - "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.8.tgz", - "integrity": "sha512-GZMtZUTNRpOVIECoXwLNZS5xUGE+mVNbTB8h/7Rwh2TFWcBQiPzTgyZi05BF9UMZKkLJv8XBRJTlU7zg8+ZfMg==", + "version": "3.1.9", + "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.9.tgz", + "integrity": "sha512-hS/0Wt3Ya043oKXs0sIxrMpKQITXhMLIrVAKKXi4gozGC9v7YRWVnXIZzjNPTO02M9BdxLCWqK1yRXgNziVyJQ==", "funding": [ { "type": "github", @@ -643,9 +643,9 @@ "license": "ISC" }, "node_modules/ip-address": { - "version": "10.7.3", - "resolved": "https://registry.npmjs.org/ip-address/-/ip-address-10.7.3.tgz", - "integrity": "sha512-A1kdq/tSb5QjvKvAMgIoEvDBIgL7qaqVP/jkvSwYYRZ9iEzvPpopxp2wQfu3SuZRHtpHNxMn8Fs0bS+gf5Xmwg==", + "version": "10.7.6", + "resolved": "https://registry.npmjs.org/ip-address/-/ip-address-10.7.6.tgz", + "integrity": "sha512-00LYkM6byILa9eQIx6qoaXCM40xTLd56AZ4vyaPW3YTyb49VZYPo5xsiAq4tuQ54GH4R7gVdYScJ1B7B77PuTg==", "license": "MIT", "engines": { "node": ">= 12" @@ -884,9 +884,9 @@ } }, "node_modules/qs": { - "version": "6.16.0", - "resolved": "https://registry.npmjs.org/qs/-/qs-6.16.0.tgz", - "integrity": "sha512-h6fhOIaRrID2CbEY2fqs+7t+UXZo+MLAnU5gRIq85uFtdiUPCdsApMlHhXogKVM4HM2DVbIjGNTTYH2OcmP1vA==", + "version": "6.17.1", + "resolved": "https://registry.npmjs.org/qs/-/qs-6.17.1.tgz", + "integrity": "sha512-jd3COkaKVgGUW+ZHyEHCG9bH7RoaaiWuMgzsIBjUMVjLAaWGWpoUAVwljVvBHxRu/2qEHnadLGZrkutYUBPQJA==", "license": "BSD-3-Clause", "dependencies": { "es-define-property": "^1.0.1", @@ -1191,9 +1191,9 @@ "license": "ISC" }, "node_modules/zod": { - "version": "4.6.5", - "resolved": "https://registry.npmjs.org/zod/-/zod-4.6.5.tgz", - "integrity": "sha512-v5l/aFXZQeai4awLbOpSoHecE9UiMrnfx75tEXLjNonXVARxQ5mOeipTjROUchszUNCqnE+hqAMujRsRHsut2Q==", + "version": "4.6.7", + "resolved": "https://registry.npmjs.org/zod/-/zod-4.6.7.tgz", + "integrity": "sha512-g8W2D2kU6a8untc053fX1uVBKzM7yqAxW/3zwk83+6RyF2+SKOuV+c0wBDhUwzV1PaGapmZH+rmcUzPf+3CGwg==", "license": "MIT", "funding": { "url": "https://github.com/sponsors/colinhacks"