diff --git a/deploy/orchestrator/.env.example b/deploy/orchestrator/.env.example index 47a97f2..cfa4f8a 100644 --- a/deploy/orchestrator/.env.example +++ b/deploy/orchestrator/.env.example @@ -9,6 +9,7 @@ MEMIND_ORCHESTRATOR_EXECUTION_ENABLED=0 MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS= MEMIND_ORCHESTRATOR_USER_ALLOWLIST= MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST= -MEMIND_EXECUTOR_GOOSED_URL=http://host.docker.internal:18006 +MEMIND_EXECUTOR_GOOSED_URL=https://host.docker.internal:18006 MEMIND_EXECUTOR_GOOSED_TOKEN= +MEMIND_EXECUTOR_GOOSED_TLS_INSECURE=1 MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON={} diff --git a/deploy/orchestrator/compose.yaml b/deploy/orchestrator/compose.yaml index 3ad9466..58880eb 100644 --- a/deploy/orchestrator/compose.yaml +++ b/deploy/orchestrator/compose.yaml @@ -62,9 +62,10 @@ services: MEMIND_ORCHESTRATOR_SERVICE_TOKEN: ${MEMIND_ORCHESTRATOR_SERVICE_TOKEN:?set MEMIND_ORCHESTRATOR_SERVICE_TOKEN} MEMIND_ORCHESTRATOR_WORKER_TOKEN: ${MEMIND_ORCHESTRATOR_WORKER_TOKEN:-local-worker-disabled} MEMIND_EXECUTOR_WORKER_ID: ${MEMIND_EXECUTOR_GOOSED_WORKER_ID:-colima-goosed-1} - MEMIND_EXECUTOR_GOOSED_URL: ${MEMIND_EXECUTOR_GOOSED_URL:-http://host.docker.internal:18006} + MEMIND_EXECUTOR_GOOSED_URL: ${MEMIND_EXECUTOR_GOOSED_URL:-https://host.docker.internal:18006} MEMIND_EXECUTOR_GOOSED_TOKEN: ${MEMIND_EXECUTOR_GOOSED_TOKEN:-} MEMIND_EXECUTOR_ALLOW_REMOTE_GOOSED: "1" + MEMIND_EXECUTOR_GOOSED_TLS_INSECURE: ${MEMIND_EXECUTOR_GOOSED_TLS_INSECURE:-1} MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON: ${MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON:-{}} MEMIND_EXECUTOR_POLL_MS: ${MEMIND_EXECUTOR_POLL_MS:-1000} MEMIND_EXECUTOR_HEARTBEAT_MS: ${MEMIND_EXECUTOR_HEARTBEAT_MS:-10000} diff --git a/services/orchestrator/executor-adapters.mjs b/services/orchestrator/executor-adapters.mjs index a185510..63479bb 100644 --- a/services/orchestrator/executor-adapters.mjs +++ b/services/orchestrator/executor-adapters.mjs @@ -2,6 +2,7 @@ import { spawn as nodeSpawn } from 'node:child_process'; import fs from 'node:fs/promises'; import path from 'node:path'; import { EventEmitter } from 'node:events'; +import { Agent, fetch as undiciFetch } from 'undici'; const DEFAULT_OUTPUT_LIMIT = 64 * 1024; const DEFAULT_ALLOWED_ENV = Object.freeze(['PATH', 'LANG', 'LC_ALL', 'TMPDIR']); @@ -27,6 +28,13 @@ function isLoopbackHostname(hostname) { || normalized.endsWith('.localhost'); } +function isLocalExecutorHostname(hostname) { + const normalized = String(hostname ?? '').trim().toLowerCase(); + return isLoopbackHostname(normalized) + || normalized === 'host.docker.internal' + || normalized === 'gateway.docker.internal'; +} + function normalizeBaseUrl(value, { allowRemote = false } = {}) { const url = new URL(String(value ?? '').trim()); if (!['http:', 'https:'].includes(url.protocol)) { @@ -94,19 +102,28 @@ function terminalGoosedEvent(event) { export function createGoosedExecutorAdapter({ baseUrl, secret = '', - fetchImpl = globalThis.fetch, + fetchImpl = null, allowRemote = false, + tlsInsecure = false, workspaceResolver = async (reference) => reference?.id ?? null, } = {}) { const target = normalizeBaseUrl(baseUrl, { allowRemote }); - if (typeof fetchImpl !== 'function') throw new Error('Goosed adapter requires fetch'); + const targetUrl = new URL(target); + if (tlsInsecure && !isLocalExecutorHostname(targetUrl.hostname)) { + throw new Error('Insecure executor TLS is restricted to local targets'); + } + const transport = fetchImpl ?? undiciFetch; + if (typeof transport !== 'function') throw new Error('Goosed adapter requires fetch'); + const dispatcher = tlsInsecure + ? new Agent({ connect: { rejectUnauthorized: false } }) + : null; const active = new Map(); function headers(extra = {}) { return { accept: 'application/json', 'content-type': 'application/json', - ...(secret ? { authorization: `Bearer ${secret}` } : {}), + ...(secret ? { 'X-Secret-Key': secret } : {}), ...extra, }; } @@ -129,7 +146,7 @@ export function createGoosedExecutorAdapter({ 'Goosed workspace reference could not be resolved', ); } - const start = await fetchImpl(`${target}/agent/start`, { + const start = await transport(`${target}/agent/start`, { method: 'POST', headers: headers(), body: JSON.stringify({ @@ -137,6 +154,7 @@ export function createGoosedExecutorAdapter({ enable_context_memory: false, }), signal: controller.signal, + ...(dispatcher ? { dispatcher } : {}), }); if (!start.ok) throw await responseError(start, 'GOOSED_START_FAILED', 'Goosed start failed'); const session = await start.json(); @@ -148,12 +166,13 @@ export function createGoosedExecutorAdapter({ sessionRef: { kind: 'goosed-session', id: sessionId }, }); - const eventsResponse = await fetchImpl( + const eventsResponse = await transport( `${target}/sessions/${encodeURIComponent(sessionId)}/events`, { method: 'GET', headers: headers({ accept: 'text/event-stream' }), signal: controller.signal, + ...(dispatcher ? { dispatcher } : {}), }, ); if (!eventsResponse.ok || !eventsResponse.body) { @@ -185,7 +204,7 @@ export function createGoosedExecutorAdapter({ return { eventCount, finalEvent }; })(); - const reply = await fetchImpl( + const reply = await transport( `${target}/sessions/${encodeURIComponent(sessionId)}/reply`, { method: 'POST', @@ -198,6 +217,7 @@ export function createGoosedExecutorAdapter({ }, }), signal: controller.signal, + ...(dispatcher ? { dispatcher } : {}), }, ); if (!reply.ok) throw await responseError(reply, 'GOOSED_REPLY_FAILED', 'Goosed reply failed'); @@ -247,6 +267,23 @@ export function createGoosedExecutorAdapter({ async getState(jobId) { return { active: active.has(jobId) }; }, + async health() { + try { + const response = await transport(`${target}/status`, { + method: 'GET', + headers: headers(), + signal: AbortSignal.timeout(2_000), + ...(dispatcher ? { dispatcher } : {}), + }); + return { ok: response.ok, status: response.status }; + } catch (error) { + return { + ok: false, + status: null, + code: String(error?.code ?? error?.name ?? 'GOOSED_UNREACHABLE').slice(0, 128), + }; + } + }, async *streamEvents() { throw adapterError( 'GOOSED_STREAM_PUSH_ONLY', @@ -431,6 +468,18 @@ export function createProcessExecutorAdapter({ async getState(jobId) { return { active: active.has(jobId) }; }, + async health() { + try { + await fs.access(executable); + await fs.access(allowedRoot); + return { ok: true }; + } catch (error) { + return { + ok: false, + code: String(error?.code ?? 'EXECUTOR_PROCESS_UNAVAILABLE').slice(0, 128), + }; + } + }, async *streamEvents() { throw adapterError( 'EXECUTOR_PROCESS_STREAM_PUSH_ONLY', @@ -465,6 +514,7 @@ export const executorAdapterInternals = { adapterError, appendBounded, isLoopbackHostname, + isLocalExecutorHostname, normalizeBaseUrl, parseSse, resolveWorkspace, diff --git a/services/orchestrator/executor-adapters.test.mjs b/services/orchestrator/executor-adapters.test.mjs index 45a0ccf..6bbc782 100644 --- a/services/orchestrator/executor-adapters.test.mjs +++ b/services/orchestrator/executor-adapters.test.mjs @@ -57,6 +57,7 @@ test('Goosed adapter uses the current session API, streams bounded metadata and ]); } if (url.endsWith('/reply')) return Response.json({ ok: true }); + if (url.endsWith('/status')) return Response.json({ ok: true }); throw new Error(`Unexpected URL: ${url}`); }, }); @@ -67,7 +68,7 @@ test('Goosed adapter uses the current session API, streams bounded metadata and assert.deepEqual(result.artifactRefs, [{ kind: 'goosed-session', id: 'session-1' }]); assert.equal(result.metrics.eventCount, 2); assert.equal(calls[0].url, 'http://127.0.0.1:18006/agent/start'); - assert.equal(calls[0].init.headers.authorization, 'Bearer secret'); + assert.equal(calls[0].init.headers['X-Secret-Key'], 'secret'); assert.equal(JSON.parse(calls[0].init.body).working_dir, '/tmp/canary'); assert.equal( JSON.parse(calls.find((call) => call.url.endsWith('/reply')).init.body) @@ -83,6 +84,7 @@ test('Goosed adapter uses the current session API, streams bounded metadata and ], ); assert.equal(JSON.stringify(emitted).includes('not-projected'), false); + assert.deepEqual(await adapter.health(), { ok: true, status: 200 }); }); test('Goosed adapter rejects remote targets unless explicitly allowed', () => { @@ -96,6 +98,21 @@ test('Goosed adapter rejects remote targets unless explicitly allowed', () => { allowRemote: true, }), ); + assert.throws( + () => createGoosedExecutorAdapter({ + baseUrl: 'https://executor.example.com', + allowRemote: true, + tlsInsecure: true, + }), + /restricted to local targets/, + ); + assert.doesNotThrow( + () => createGoosedExecutorAdapter({ + baseUrl: 'https://host.docker.internal:18006', + allowRemote: true, + tlsInsecure: true, + }), + ); }); test('Aider process adapter resolves a workspace alias and never invokes a shell', async () => { diff --git a/services/orchestrator/executor-worker-protocol.mjs b/services/orchestrator/executor-worker-protocol.mjs index e39c087..3011987 100644 --- a/services/orchestrator/executor-worker-protocol.mjs +++ b/services/orchestrator/executor-worker-protocol.mjs @@ -121,6 +121,7 @@ export function createExecutorWorkerCoordinator({ executors, leaseDurationMs = defaultLeaseDurationMs, draining = false, + adapterHealth = {}, } = {}) { const normalizedWorkerId = requiredIdentifier(workerId, 'worker_id'); const normalizedExecutors = [...new Set( @@ -141,6 +142,17 @@ export function createExecutorWorkerCoordinator({ workerId: normalizedWorkerId, executors: normalizedExecutors, draining: draining === true, + adapterHealth: Object.fromEntries( + normalizedExecutors.map((executor) => [ + executor, + { + ok: adapterHealth?.[executor]?.ok === true, + code: adapterHealth?.[executor]?.code == null + ? null + : String(adapterHealth[executor].code).slice(0, 128), + }, + ]), + ), lastSeenAt: now, }); if (draining === true) return null; @@ -414,7 +426,11 @@ export function createExecutorWorkerCoordinator({ version: WORKER_PROTOCOL_VERSION, ...(await store.getQueueStats({ now })), workers, - healthyWorkers: workers.filter((worker) => !worker.stale && !worker.draining).length, + healthyWorkers: workers.filter((worker) => ( + !worker.stale + && !worker.draining + && Object.values(worker.adapterHealth ?? {}).some((health) => health?.ok === true) + )).length, }; } diff --git a/services/orchestrator/executor-worker-runtime.mjs b/services/orchestrator/executor-worker-runtime.mjs index bfa49d5..e4d38ea 100644 --- a/services/orchestrator/executor-worker-runtime.mjs +++ b/services/orchestrator/executor-worker-runtime.mjs @@ -112,6 +112,22 @@ export function createWorkerAdapterRegistry(initialAdapters = []) { ids() { return [...adapters.keys()]; }, + async health() { + const result = {}; + for (const [id, adapter] of adapters) { + try { + result[id] = typeof adapter.health === 'function' + ? await adapter.health() + : { ok: true }; + } catch (error) { + result[id] = { + ok: false, + code: String(error?.code ?? 'EXECUTOR_ADAPTER_HEALTH_FAILED').slice(0, 128), + }; + } + } + return result; + }, _initialize() { for (const adapter of initialAdapters) this.register(adapter); return this; @@ -132,6 +148,7 @@ export function createExecutorWorkerRuntime({ if (!client?.claim) throw new Error('Executor worker runtime requires client'); let stopping = false; const active = new Map(); + let lastAdapterHealth = {}; async function executeAdapter(adapter, job, leaseToken, controller) { return adapter.submit(job.request, { @@ -233,12 +250,23 @@ export function createExecutorWorkerRuntime({ return { async runOnce() { if (stopping) return { processed: false, reason: 'stopping' }; + lastAdapterHealth = await registry.health(); + const healthyExecutors = registry.ids().filter( + (executor) => lastAdapterHealth[executor]?.ok === true, + ); const claim = await client.claim({ workerId: normalizedWorkerId, - executors: registry.ids(), + executors: healthyExecutors.length ? healthyExecutors : registry.ids(), leaseDurationMs, + draining: healthyExecutors.length === 0, + adapterHealth: lastAdapterHealth, }); - if (!claim) return { processed: false, reason: 'queue_empty' }; + if (!claim) { + return { + processed: false, + reason: healthyExecutors.length ? 'queue_empty' : 'adapters_unhealthy', + }; + } return processClaim(claim); }, async recoverExpired(options = {}) { @@ -251,6 +279,7 @@ export function createExecutorWorkerRuntime({ stopping, activeJobs: [...active.keys()], adapters: registry.list(), + adapterHealth: lastAdapterHealth, heartbeatMs, leaseDurationMs, }; @@ -262,6 +291,7 @@ export function createExecutorWorkerRuntime({ executors: registry.ids(), leaseDurationMs, draining: true, + adapterHealth: lastAdapterHealth, }).catch(() => null); for (const { controller } of active.values()) controller.abort('worker_stopping'); }, diff --git a/services/orchestrator/executor-worker-runtime.test.mjs b/services/orchestrator/executor-worker-runtime.test.mjs index be399af..f462462 100644 --- a/services/orchestrator/executor-worker-runtime.test.mjs +++ b/services/orchestrator/executor-worker-runtime.test.mjs @@ -85,6 +85,7 @@ test('worker runtime claims, executes, emits progress and completes through leas client.calls.map((call) => call[0]), ['claim', 'start', 'progress', 'complete'], ); + assert.deepEqual(client.calls[0][1].adapterHealth, { goosed: { ok: true } }); assert.equal(client.calls.at(-1)[2].leaseToken, 'lease-1'); assert.deepEqual(worker.status().activeJobs, []); }); diff --git a/services/orchestrator/package-lock.json b/services/orchestrator/package-lock.json index fafb0d9..c594b98 100644 --- a/services/orchestrator/package-lock.json +++ b/services/orchestrator/package-lock.json @@ -12,7 +12,8 @@ "@langchain/langgraph": "^1.4.8", "@langchain/langgraph-checkpoint-postgres": "^1.0.4", "express": "^4.21.2", - "pg": "^8.12.0" + "pg": "^8.12.0", + "undici": "^6.27.0" }, "engines": { "node": ">=20" @@ -1238,6 +1239,15 @@ "node": ">= 0.6" } }, + "node_modules/undici": { + "version": "6.28.0", + "resolved": "https://registry.npmjs.org/undici/-/undici-6.28.0.tgz", + "integrity": "sha512-LIY910g9TI13YS95lrMFrs8Rm/u/irgHeTWoKCoteeJ04CUJ92eEfj0rVn+7VKMPBpUPiUoBKfhNyLI23EE/KA==", + "license": "MIT", + "engines": { + "node": ">=18.17" + } + }, "node_modules/unpipe": { "version": "1.0.0", "resolved": "https://registry.npmjs.org/unpipe/-/unpipe-1.0.0.tgz", diff --git a/services/orchestrator/package.json b/services/orchestrator/package.json index 20ac427..0b7e433 100644 --- a/services/orchestrator/package.json +++ b/services/orchestrator/package.json @@ -16,6 +16,7 @@ "@langchain/langgraph": "^1.4.8", "@langchain/langgraph-checkpoint-postgres": "^1.0.4", "express": "^4.21.2", - "pg": "^8.12.0" + "pg": "^8.12.0", + "undici": "^6.27.0" } } diff --git a/services/orchestrator/worker.mjs b/services/orchestrator/worker.mjs index a2d832a..7dd687a 100644 --- a/services/orchestrator/worker.mjs +++ b/services/orchestrator/worker.mjs @@ -32,7 +32,7 @@ function jsonObject(value, fallback = {}) { } } -export function createConfiguredWorker({ env = process.env, fetchImpl = globalThis.fetch } = {}) { +export function createConfiguredWorker({ env = process.env, fetchImpl = null } = {}) { const registry = createWorkerAdapterRegistry(); const workspaceAliases = jsonObject(env.MEMIND_EXECUTOR_WORKSPACE_ALIASES_JSON); const allowedRoot = String(env.MEMIND_EXECUTOR_WORKSPACE_ROOT ?? '').trim(); @@ -41,8 +41,9 @@ export function createConfiguredWorker({ env = process.env, fetchImpl = globalTh registry.register(createGoosedExecutorAdapter({ baseUrl: goosedUrl, secret: env.MEMIND_EXECUTOR_GOOSED_TOKEN, - fetchImpl, + ...(fetchImpl ? { fetchImpl } : {}), allowRemote: envFlag(env.MEMIND_EXECUTOR_ALLOW_REMOTE_GOOSED, false), + tlsInsecure: envFlag(env.MEMIND_EXECUTOR_GOOSED_TLS_INSECURE, false), workspaceResolver: async (reference) => workspaceAliases[reference?.id] ?? null, })); } @@ -67,7 +68,7 @@ export function createConfiguredWorker({ env = process.env, fetchImpl = globalTh baseUrl: env.MEMIND_ORCHESTRATOR_URL ?? 'http://127.0.0.1:8093', serviceToken: env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, workerToken: env.MEMIND_ORCHESTRATOR_WORKER_TOKEN, - fetchImpl, + fetchImpl: fetchImpl ?? globalThis.fetch, }); return createExecutorWorkerRuntime({ client,