tkmind/memind
1
0
Fork 0
Code Issues Pull Requests 2 Actions 8 Packages Projects Releases Wiki Activity

fix: align orchestrator worker with local goosed

Browse Source
This commit is contained in:
john
2026-07-25 00:04:50 +08:00
parent e3d7b9ee76
commit cf36a9c84a
10 changed files with 145 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/orchestrator/.env.example
+2 -1
View File
@@ -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={}
deploy/orchestrator/compose.yaml
+2 -1
View File
@@ -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}
services/orchestrator/executor-adapters.mjs
+56 -6
View File
@@ -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,
services/orchestrator/executor-adapters.test.mjs
+18 -1
View File
@@ -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 () => {
services/orchestrator/executor-worker-protocol.mjs
+17 -1
View File
@@ -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,
};
}
services/orchestrator/executor-worker-runtime.mjs
+32 -2
View File
@@ -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');
},
services/orchestrator/executor-worker-runtime.test.mjs
+1
View File
@@ -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, []);
});
services/orchestrator/package-lock.json
Generated
+11 -1
View File
@@ -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",
services/orchestrator/package.json
+2 -1
View File
@@ -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"
}
}
services/orchestrator/worker.mjs
+4 -3
View File
@@ -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,