diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index 2e4c022..7923c43 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -114,7 +114,28 @@ The `build_plan` node creates the deterministic preview job `:executor-preview` with idempotency key `:build_plan:v1`. The stored result is terminal `blocked` state with `executor_dispatch_not_implemented`; it is a durable audit and recovery -boundary, not an execution claim. +boundary, not an execution claim. If the run identifier is unsafe or would +overflow the 128-character Job id boundary, the graph substitutes a stable +SHA-256-derived opaque prefix. + +Phase 3.4 adds the append-only `executor-job-event-v1` contract. Job snapshots +remain in `executor_jobs`, while cursor-addressable events live in +`executor_job_events`. A transaction commits the initial snapshot and event +together; later state transitions lock the job row and commit the updated +snapshot and next event together. This prevents a visible state transition +without its corresponding audit event. + +The read boundary is deliberately narrower than an execution control API: + +```text +GET /v1/executor-jobs/:jobId +GET /v1/executor-jobs/:jobId/events?after=&limit= +``` + +There is no HTTP submit, retry, claim, or executor cancel endpoint in this +phase. memindadm resolves the job reference stored in LangGraph plan state and +shows the job snapshot and events beside Native and graph state. The payload +contains no task instruction, secret, absolute path, or executor SDK object. The implemented internal API is: @@ -124,6 +145,8 @@ GET /v1/runs/:id POST /v1/runs/:id/resume POST /v1/runs/:id/cancel GET /v1/runs/:id/events?after= +GET /v1/executor-jobs/:jobId +GET /v1/executor-jobs/:jobId/events?after=&limit= ``` External state changes must use an idempotency key derived from run, graph @@ -164,8 +187,9 @@ read LangGraph checkpoint state and node events. Canary readiness is also a Memind control-plane projection. It excludes synthetic smoke runs and evaluates operational health, durable checkpoint state, -sample volume, session coverage, success rate, latency coverage, P95 latency, -Native terminal coverage, and sample freshness. The result is advisory: +durable Executor Job storage, sample volume, session coverage, success rate, +latency coverage, P95 latency, Native terminal coverage, and sample freshness. +The result is advisory: `manual_canary_review` never mutates routing configuration or transfers execution ownership. diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index c3bbd7e..8c54efa 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -410,6 +410,8 @@ export type OrchestratorCanaryReadiness = { latencyMs: number | null; checkpointKind: string | null; checkpointDurable: boolean | null; + executorJobStoreKind: string | null; + executorJobStoreDurable: boolean | null; execution: string | null; }; checks: OrchestratorCanaryReadinessCheck[]; @@ -452,6 +454,29 @@ export type OrchestratorShadowRunDetail = { data?: Record | null; }>; error: { code: string; message: string } | null; + executorJob: { + available: boolean; + state: { + version?: string; + jobId?: string; + executor?: string; + status?: string; + reason?: string; + attempts?: number; + createdAt?: number; + updatedAt?: number; + } | null; + events: Array<{ + version?: string; + eventId?: string; + jobId?: string; + sequence?: number; + type?: string; + timestamp?: number; + data?: Record | null; + }>; + error: { code: string; message: string } | null; + }; }; }; diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx index 35a8f2c..03bbad5 100644 --- a/ops/src/pages/admin/OrchestratorPage.tsx +++ b/ops/src/pages/admin/OrchestratorPage.tsx @@ -173,7 +173,7 @@ export function OrchestratorPage() {

Executor Gateway

- Phase 3.3 已持久化 blocked 预览任务；所有执行器仍为 contract-only，不会启动进程或接管任务。 + Phase 3.4 已开放只读任务状态与事件游标；所有执行器仍为 contract-only，不会启动进程或接管任务。

diff --git a/ops/src/pages/admin/OrchestratorShadowPanel.tsx b/ops/src/pages/admin/OrchestratorShadowPanel.tsx index 1ba4b30..4a9c5ea 100644 --- a/ops/src/pages/admin/OrchestratorShadowPanel.tsx +++ b/ops/src/pages/admin/OrchestratorShadowPanel.tsx @@ -29,6 +29,7 @@ const readinessCheckLabels: Record = { shadow_mode: '当前保持 Shadow', service_healthy: '服务健康', durable_checkpoint: '持久化 checkpoint', + durable_executor_job_store: '持久化 Executor Job Store', observe_only: '仅观察不执行', sample_volume: '有效样本量', shadow_success_rate: 'Shadow 成功率', @@ -301,6 +302,13 @@ export function OrchestratorShadowPanel() { {selected.remote.error.code}: {selected.remote.error.message}

) : null} + {selected.remote.executorJob.error ? ( +

+ Executor Job：{selected.remote.executorJob.error.code} + {' '} + {selected.remote.executorJob.error.message} +

+ ) : null}
LangGraph checkpoint @@ -315,6 +323,31 @@ export function OrchestratorShadowPanel() {
+ {selected.remote.executorJob.available ? ( +
+
+ Executor Job 状态 + 
+                  {JSON.stringify(selected.remote.executorJob.state, null, 2)}
+
+
+
+ Executor Job 事件 + 
+                  {JSON.stringify(selected.remote.executorJob.events, null, 2)}
+
+
+
+ ) : null}
) : null} diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md index 7d1b4af..baf60ff 100644 --- a/services/orchestrator/README.md +++ b/services/orchestrator/README.md @@ -126,6 +126,27 @@ The persisted job is still a preview terminal state: `status=blocked`, `reason=executor_dispatch_not_implemented`, and `executionEnabled=false`. Phase 3.3 does not call Goosed, Aider, or OpenHands. +Phase 3.4 adds read-only Executor Job observability: + +- `executor-job-event-v1` is stored in the separate PostgreSQL + `executor_job_events` table with a per-job sequence cursor; +- initial state and its first event are committed atomically, as are later + state transitions and their events; +- existing Phase 3.3 snapshots receive a deterministic + `executor_job_snapshot_imported` migration event; +- the versioned API exposes job state and cursor-paged events without exposing + task instructions, credentials, workspace paths, or executor SDK types; +- unsafe or oversized run identifiers are represented by a deterministic + SHA-256-derived opaque Executor Job id so PostgreSQL and URL identifiers stay + bounded; +- Shadow Run detail in memindadm joins Native state, LangGraph checkpoints, + Executor Job state, and Executor Job events; +- Canary readiness now requires both checkpoint and Executor Job storage to be + durable. + +These endpoints are observational only. They do not add submit, retry, claim, +or adapter-dispatch operations. + The default memindadm mode remains `off`. Canary and Active routing are not wired to the LangGraph executor in Phase 3. Native Agent Run remains the sole executor in every mode. @@ -173,6 +194,14 @@ GET /admin-api/orchestrator/canary-readiness GET /admin-api/orchestrator/execution-plans ``` +Selecting a Shadow Run in memindadm also reads the linked Executor Job through +the Orchestrator API: + +```text +GET /v1/executor-jobs/:jobId +GET /v1/executor-jobs/:jobId/events?after=&limit= +``` + ## Colima deployment Colima is the recommended first container host on macOS because this service and diff --git a/services/orchestrator/app.mjs b/services/orchestrator/app.mjs index 966bd32..6e17c9b 100644 --- a/services/orchestrator/app.mjs +++ b/services/orchestrator/app.mjs @@ -95,6 +95,33 @@ export function createOrchestratorApp({ } }); + app.get('/v1/executor-jobs/:jobId', async (request, response, next) => { + try { + const job = await runtime.getExecutorJob(request.params.jobId); + if (!job) { + return response.status(404).json({ error: { code: 'EXECUTOR_JOB_NOT_FOUND' } }); + } + return response.json(job); + } catch (error) { + return next(error); + } + }); + + app.get('/v1/executor-jobs/:jobId/events', async (request, response, next) => { + try { + const result = await runtime.listExecutorJobEvents(request.params.jobId, { + after: request.query.after, + limit: request.query.limit, + }); + if (!result) { + return response.status(404).json({ error: { code: 'EXECUTOR_JOB_NOT_FOUND' } }); + } + return response.json(result); + } catch (error) { + return next(error); + } + }); + app.use((error, _request, response, _next) => { response.status(errorStatus(error)).json({ error: { diff --git a/services/orchestrator/app.test.mjs b/services/orchestrator/app.test.mjs index 3f5b320..cdc25b1 100644 --- a/services/orchestrator/app.test.mjs +++ b/services/orchestrator/app.test.mjs @@ -57,6 +57,10 @@ test('orchestrator HTTP API exposes health and authenticated run endpoints', asy body: JSON.stringify(spec()), }); assert.equal(unauthorized.status, 401); + const unauthorizedExecutorJob = await fetch( + `${server.baseUrl}/v1/executor-jobs/http-shadow-1%3Aexecutor-preview`, + ); + assert.equal(unauthorizedExecutorJob.status, 401); const created = await fetch(`${server.baseUrl}/v1/runs`, { method: 'POST', @@ -75,6 +79,26 @@ test('orchestrator HTTP API exposes health and authenticated run endpoints', asy const eventBody = await events.json(); assert.equal(events.status, 200); assert.deepEqual(eventBody.events.map((event) => event.sequence), [2, 3]); + + const jobId = 'http-shadow-1:executor-preview'; + const executorJob = await fetch( + `${server.baseUrl}/v1/executor-jobs/${encodeURIComponent(jobId)}`, + { headers: { authorization: 'Bearer test-service-token' } }, + ); + assert.equal(executorJob.status, 200); + assert.equal((await executorJob.json()).status, 'blocked'); + + const executorEvents = await fetch( + `${server.baseUrl}/v1/executor-jobs/${encodeURIComponent(jobId)}/events?after=0`, + { headers: { authorization: 'Bearer test-service-token' } }, + ); + const executorEventBody = await executorEvents.json(); + assert.equal(executorEvents.status, 200); + assert.deepEqual( + executorEventBody.events.map((event) => event.type), + ['executor_job_blocked'], + ); + assert.equal(executorEventBody.nextCursor, 1); }); test('orchestrator HTTP API reports missing runs and observe-only violations', async (t) => { @@ -86,6 +110,9 @@ test('orchestrator HTTP API reports missing runs and observe-only violations', a const missing = await fetch(`${server.baseUrl}/v1/runs/missing`); assert.equal(missing.status, 404); + const missingExecutorJob = await fetch(`${server.baseUrl}/v1/executor-jobs/missing`); + assert.equal(missingExecutorJob.status, 404); + assert.equal((await missingExecutorJob.json()).error.code, 'EXECUTOR_JOB_NOT_FOUND'); const activeSpec = spec(); activeSpec.runId = 'http-active-1'; diff --git a/services/orchestrator/code-run-shadow-graph.mjs b/services/orchestrator/code-run-shadow-graph.mjs index 8ed2ae7..3b0dae8 100644 --- a/services/orchestrator/code-run-shadow-graph.mjs +++ b/services/orchestrator/code-run-shadow-graph.mjs @@ -1,3 +1,4 @@ +import crypto from 'node:crypto'; import { Annotation, END, @@ -35,6 +36,22 @@ const ShadowState = Annotation.Root({ }), }); +function executorIdentityForRun(runId) { + const normalizedRunId = String(runId ?? '').trim(); + const jobId = `${normalizedRunId}:executor-preview`; + if (/^[a-zA-Z0-9:_-]+$/.test(normalizedRunId) && jobId.length <= 128) { + return { + jobId, + idempotencyKey: `${normalizedRunId}:build_plan:v1`, + }; + } + const digest = crypto.createHash('sha256').update(normalizedRunId).digest('hex'); + return { + jobId: `run_${digest}:executor-preview`, + idempotencyKey: `run_${digest}:build_plan:v1`, + }; +} + function eventFor(state, type, data = null) { return buildRunEvent({ runId: state.spec.runId, @@ -70,9 +87,10 @@ function validateNode(state) { async function planNode(state, executorGateway) { const taskType = String(state.spec.input?.taskType ?? '').trim() || 'code-change'; const instruction = String(state.spec.input?.instruction ?? ''); + const executorIdentity = executorIdentityForRun(state.spec.runId); const executorJob = await executorGateway.createJob({ - jobId: `${state.spec.runId}:executor-preview`, - idempotencyKey: `${state.spec.runId}:build_plan:v1`, + jobId: executorIdentity.jobId, + idempotencyKey: executorIdentity.idempotencyKey, executor: 'goosed', task: { type: taskType, @@ -162,6 +180,7 @@ export function createCodeRunShadowGraph({ checkpointer, executorGateway } = {}) export const codeRunShadowGraphInternals = { ShadowState, + executorIdentityForRun, eventFor, validateNode, planNode, diff --git a/services/orchestrator/contracts.test.mjs b/services/orchestrator/contracts.test.mjs index 2279079..d53b0cf 100644 --- a/services/orchestrator/contracts.test.mjs +++ b/services/orchestrator/contracts.test.mjs @@ -60,7 +60,9 @@ test('remote workflow engine speaks only the versioned orchestrator HTTP contrac return { ok: true, async json() { - return url.endsWith('/events') ? { events: [{ type: 'run.succeeded' }] } : { ok: true }; + return url.includes('/events') + ? { events: [{ type: 'run.succeeded' }] } + : { ok: true }; }, }; }, @@ -71,6 +73,13 @@ test('remote workflow engine speaks only the versioned orchestrator HTTP contrac await engine.getState('run-1'); const events = []; for await (const event of engine.streamEvents('run-1')) events.push(event); + await engine.getExecutorJob('run-1:executor-preview'); + const executorEvents = []; + for await ( + const event of engine.streamExecutorJobEvents('run-1:executor-preview', 2) + ) { + executorEvents.push(event); + } assert.deepEqual(calls.map((call) => [call.init?.method ?? 'GET', call.url]), [ ['POST', 'http://orchestrator.internal/v1/runs'], @@ -78,6 +87,12 @@ test('remote workflow engine speaks only the versioned orchestrator HTTP contrac ['POST', 'http://orchestrator.internal/v1/runs/run-1/cancel'], ['GET', 'http://orchestrator.internal/v1/runs/run-1'], ['GET', 'http://orchestrator.internal/v1/runs/run-1/events'], + ['GET', 'http://orchestrator.internal/v1/executor-jobs/run-1%3Aexecutor-preview'], + [ + 'GET', + 'http://orchestrator.internal/v1/executor-jobs/run-1%3Aexecutor-preview/events?after=2', + ], ]); assert.deepEqual(events, [{ type: 'run.succeeded' }]); + assert.deepEqual(executorEvents, [{ type: 'run.succeeded' }]); }); diff --git a/services/orchestrator/engine-registry.mjs b/services/orchestrator/engine-registry.mjs index 9553f74..fad6c31 100644 --- a/services/orchestrator/engine-registry.mjs +++ b/services/orchestrator/engine-registry.mjs @@ -105,7 +105,7 @@ export function createRemoteWorkflowEngine({ id: normalizedId, label, kind: 'remote', - capabilities: ['durable-execution', 'interrupt', 'streaming'], + capabilities: ['durable-execution', 'interrupt', 'streaming', 'executor-observability'], async start(spec) { return request('/v1/runs', { method: 'POST', body: JSON.stringify(spec) }); }, @@ -129,5 +129,15 @@ export function createRemoteWorkflowEngine({ const result = await request(`/v1/runs/${encodeURIComponent(runId)}/events${query}`); for (const event of Array.isArray(result?.events) ? result.events : []) yield event; }, + async getExecutorJob(jobId) { + return request(`/v1/executor-jobs/${encodeURIComponent(jobId)}`); + }, + async *streamExecutorJobEvents(jobId, cursor = null) { + const query = cursor == null ? '' : `?after=${encodeURIComponent(cursor)}`; + const result = await request( + `/v1/executor-jobs/${encodeURIComponent(jobId)}/events${query}`, + ); + for (const event of Array.isArray(result?.events) ? result.events : []) yield event; + }, }; } diff --git a/services/orchestrator/executor-gateway.mjs b/services/orchestrator/executor-gateway.mjs index 7d50835..534ca70 100644 --- a/services/orchestrator/executor-gateway.mjs +++ b/services/orchestrator/executor-gateway.mjs @@ -3,8 +3,10 @@ import { normalizeWorkflowEngineId } from './contracts.mjs'; const EXECUTOR_JOB_REQUEST_VERSION = 'executor-job-request-v1'; const EXECUTOR_JOB_STATE_VERSION = 'executor-job-state-v1'; +const EXECUTOR_JOB_EVENT_VERSION = 'executor-job-event-v1'; const EXECUTOR_DISPATCH_IMPLEMENTED = false; const MAX_INSTRUCTION_CHARACTERS = 32_000; +const MAX_EVENT_PAGE_SIZE = 500; const TERMINAL_JOB_STATUSES = new Set([ 'succeeded', 'failed', @@ -77,6 +79,33 @@ function gatewayError(code, message, status = 409) { return error; } +export function buildExecutorJobEvent({ + eventId = crypto.randomUUID(), + jobId, + sequence = 0, + type, + timestamp = Date.now(), + data = null, +} = {}) { + const normalizedJobId = String(jobId ?? '').trim(); + const normalizedType = String(type ?? '').trim(); + if (!normalizedJobId) { + throw gatewayError('EXECUTOR_EVENT_JOB_ID_REQUIRED', 'Executor event requires jobId', 422); + } + if (!normalizedType) { + throw gatewayError('EXECUTOR_EVENT_TYPE_REQUIRED', 'Executor event requires type', 422); + } + return { + version: EXECUTOR_JOB_EVENT_VERSION, + eventId: String(eventId), + jobId: normalizedJobId, + sequence: Math.max(0, Number(sequence) || 0), + type: normalizedType, + timestamp: Number(timestamp) || Date.now(), + data: data == null ? null : clone(data), + }; +} + export function normalizeExecutorJobRequest(input = {}) { const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {}; const jobId = String(source.jobId ?? crypto.randomUUID()).trim(); @@ -86,6 +115,13 @@ export function normalizeExecutorJobRequest(input = {}) { .trim() .slice(0, MAX_INSTRUCTION_CHARACTERS); if (!jobId) throw gatewayError('EXECUTOR_JOB_ID_REQUIRED', 'Executor job requires jobId', 422); + if (!/^[a-zA-Z0-9:_-]{1,128}$/.test(jobId)) { + throw gatewayError( + 'EXECUTOR_JOB_ID_INVALID', + 'Executor jobId must be a safe identifier no longer than 128 characters', + 422, + ); + } if (!idempotencyKey) { throw gatewayError( 'EXECUTOR_IDEMPOTENCY_KEY_REQUIRED', @@ -229,6 +265,19 @@ export function listPhase3ExecutorAdapters() { export function createInMemoryExecutorJobStore() { const byId = new Map(); const byIdempotencyKey = new Map(); + const eventsByJobId = new Map(); + + function appendEvent(jobId, event) { + if (!event) return; + const events = eventsByJobId.get(jobId) ?? []; + events.push({ + ...clone(event), + jobId, + sequence: events.length + 1, + }); + eventsByJobId.set(jobId, events); + } + return { kind: 'memory', durable: false, @@ -241,7 +290,7 @@ export function createInMemoryExecutorJobStore() { const value = jobId ? byId.get(jobId) : null; return value ? clone(value) : null; }, - async createIfAbsent(record) { + async createIfAbsent(record, { initialEvent = null } = {}) { const existingId = byIdempotencyKey.get(record.idempotencyKey); if (existingId) return { created: false, record: clone(byId.get(existingId)) }; if (byId.has(record.jobId)) { @@ -249,18 +298,34 @@ export function createInMemoryExecutorJobStore() { } byId.set(record.jobId, clone(record)); byIdempotencyKey.set(record.idempotencyKey, record.jobId); + appendEvent(record.jobId, initialEvent); return { created: true, record: clone(record) }; }, - async update(record) { + async update(record, { event = null } = {}) { if (!byId.has(record.jobId)) return null; byId.set(record.jobId, clone(record)); + appendEvent(record.jobId, event); return clone(record); }, + async listEvents(jobId, { after = 0, limit = MAX_EVENT_PAGE_SIZE } = {}) { + const cursor = Math.max(0, Number(after) || 0); + const pageSize = clampInteger(limit, MAX_EVENT_PAGE_SIZE, 1, MAX_EVENT_PAGE_SIZE); + return (eventsByJobId.get(String(jobId ?? '').trim()) ?? []) + .filter((event) => event.sequence > cursor) + .slice(0, pageSize) + .map(clone); + }, }; } function validateJobStore(store) { - for (const method of ['getById', 'getByIdempotencyKey', 'createIfAbsent', 'update']) { + for (const method of [ + 'getById', + 'getByIdempotencyKey', + 'createIfAbsent', + 'update', + 'listEvents', + ]) { if (typeof store?.[method] !== 'function') { throw new Error(`Executor job store missing method: ${method}`); } @@ -354,7 +419,20 @@ export function createExecutorGateway({ startedAt: null, completedAt: now, }; - const result = await store.createIfAbsent(record); + const result = await store.createIfAbsent(record, { + initialEvent: buildExecutorJobEvent({ + jobId: request.jobId, + sequence: 1, + type: 'executor_job_blocked', + timestamp: now, + data: { + executor: request.executor, + status: record.status, + reason: record.reason, + dispatchAllowed: false, + }, + }), + }); if ( !result.created && result.record.requestFingerprint !== requestFingerprint @@ -373,12 +451,23 @@ export function createExecutorGateway({ if (current.status === 'cancelled' || !current.cancellationAllowed) return current; if (TERMINAL_JOB_STATUSES.has(current.status) && current.status !== 'blocked') return current; const now = nowMs(); - return store.update({ + const updated = { ...current, status: 'cancelled', reason: String(reason).slice(0, 256), updatedAt: now, completedAt: now, + }; + return store.update(updated, { + event: buildExecutorJobEvent({ + jobId: current.jobId, + type: 'executor_job_cancelled', + timestamp: now, + data: { + status: updated.status, + reason: updated.reason, + }, + }), }); } @@ -389,6 +478,22 @@ export function createExecutorGateway({ getJob(jobId) { return store.getById(jobId); }, + async listEvents(jobId, { after = 0, limit = MAX_EVENT_PAGE_SIZE } = {}) { + const normalizedJobId = String(jobId ?? '').trim(); + if (!normalizedJobId) return null; + const job = await store.getById(normalizedJobId); + if (!job) return null; + const cursor = Math.max(0, Number(after) || 0); + const events = await store.listEvents(normalizedJobId, { + after: cursor, + limit: clampInteger(limit, MAX_EVENT_PAGE_SIZE, 1, MAX_EVENT_PAGE_SIZE), + }); + return { + jobId: normalizedJobId, + events, + nextCursor: events.at(-1)?.sequence ?? cursor, + }; + }, listAdapters() { return registry.list(); }, @@ -409,8 +514,10 @@ export function createExecutorGateway({ export const executorGatewayInternals = { EXECUTOR_DISPATCH_IMPLEMENTED, + EXECUTOR_JOB_EVENT_VERSION, EXECUTOR_JOB_REQUEST_VERSION, EXECUTOR_JOB_STATE_VERSION, + MAX_EVENT_PAGE_SIZE, MAX_INSTRUCTION_CHARACTERS, PHASE3_EXECUTOR_DESCRIPTORS, TERMINAL_JOB_STATUSES, diff --git a/services/orchestrator/executor-gateway.test.mjs b/services/orchestrator/executor-gateway.test.mjs index b452465..7cb490b 100644 --- a/services/orchestrator/executor-gateway.test.mjs +++ b/services/orchestrator/executor-gateway.test.mjs @@ -1,6 +1,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; import { + buildExecutorJobEvent, createExecutorAdapterRegistry, createExecutorGateway, createInMemoryExecutorJobStore, @@ -49,6 +50,22 @@ test('executor job contract uses resource references and bounded controls', () = assert.equal(normalized.controls.timeoutMs, 500); assert.equal(normalized.controls.fallbackExecutor, 'openhands'); assert.equal('cwd' in normalized.task, false); + assert.throws( + () => normalizeExecutorJobRequest(request({ jobId: '../unsafe' })), + (error) => error.code === 'EXECUTOR_JOB_ID_INVALID', + ); + + const event = buildExecutorJobEvent({ + eventId: 'event-1', + jobId: 'job-1', + sequence: 1, + type: 'executor_job_blocked', + timestamp: 1234, + data: { status: 'blocked' }, + }); + assert.equal(event.version, 'executor-job-event-v1'); + assert.equal(event.sequence, 1); + assert.deepEqual(event.data, { status: 'blocked' }); }); test('Phase 3 executor catalog exposes disabled Goosed, Aider and OpenHands adapters', () => { @@ -104,6 +121,9 @@ test('Executor Gateway records a blocked idempotent job without calling an adapt assert.equal(repeated.created, false); assert.equal(repeated.job.jobId, first.job.jobId); assert.equal(repeated.job.createdAt, 1000); + const eventPage = await gateway.listEvents('job-1'); + assert.deepEqual(eventPage.events.map((event) => event.type), ['executor_job_blocked']); + assert.equal(eventPage.nextCursor, 1); }); test('Executor Gateway rejects idempotency key reuse with a different payload', async () => { @@ -133,6 +153,11 @@ test('Executor Gateway cancellation is idempotent and never invokes a disabled a assert.equal(cancelled.completedAt, 1500); assert.deepEqual(await gateway.cancel('job-1'), cancelled); assert.equal(await gateway.cancel('missing-job'), null); + const eventPage = await gateway.listEvents('job-1', { after: 1 }); + assert.deepEqual(eventPage.events.map((event) => event.type), ['executor_job_cancelled']); + assert.equal(eventPage.events[0].sequence, 2); + assert.equal(eventPage.nextCursor, 2); + assert.equal(await gateway.listEvents('missing-job'), null); }); test('Executor Gateway status remains hard-disabled with a non-durable reference store', () => { diff --git a/services/orchestrator/executor-job-store.mjs b/services/orchestrator/executor-job-store.mjs index e4cbdef..8800cf5 100644 --- a/services/orchestrator/executor-job-store.mjs +++ b/services/orchestrator/executor-job-store.mjs @@ -4,6 +4,7 @@ import { createInMemoryExecutorJobStore } from './executor-gateway.mjs'; const { Pool } = pg; const DEFAULT_EXECUTOR_SCHEMA = 'memind_orchestrator'; const EXECUTOR_JOB_TABLE = 'executor_jobs'; +const EXECUTOR_JOB_EVENT_TABLE = 'executor_job_events'; function normalizeMode(value) { const mode = String(value ?? 'postgres').trim().toLowerCase(); @@ -33,6 +34,40 @@ function projectRow(row) { return parseState(row?.state_json); } +function projectEventRow(row) { + return parseState(row?.event_json); +} + +function eventWithSequence(event, jobId, sequence) { + return { + ...structuredClone(event), + jobId, + sequence, + }; +} + +async function withTransaction(pool, operation) { + if (typeof pool?.connect !== 'function') { + throw new Error('PostgreSQL Executor Job Store requires transactional pool.connect'); + } + const client = await pool.connect(); + try { + await client.query('BEGIN'); + const result = await operation(client); + await client.query('COMMIT'); + return result; + } catch (error) { + try { + await client.query('ROLLBACK'); + } catch { + // Preserve the transactional operation failure. + } + throw error; + } finally { + client.release?.(); + } +} + export function createPostgresExecutorJobStore({ pool, schema = DEFAULT_EXECUTOR_SCHEMA, @@ -40,6 +75,7 @@ export function createPostgresExecutorJobStore({ if (!pool?.query) throw new Error('PostgreSQL Executor Job Store requires a pool'); const normalizedSchema = normalizeSchema(schema); const table = `"${normalizedSchema}"."${EXECUTOR_JOB_TABLE}"`; + const eventTable = `"${normalizedSchema}"."${EXECUTOR_JOB_EVENT_TABLE}"`; return { kind: 'postgres', @@ -61,6 +97,45 @@ export function createPostgresExecutorJobStore({ CREATE INDEX IF NOT EXISTS executor_jobs_updated_at_idx ON ${table} (updated_at DESC) `); + await pool.query(` + CREATE TABLE IF NOT EXISTS ${eventTable} ( + event_id VARCHAR(128) PRIMARY KEY, + job_id VARCHAR(128) NOT NULL REFERENCES ${table} (job_id) ON DELETE CASCADE, + sequence INTEGER NOT NULL, + event_json JSONB NOT NULL, + created_at BIGINT NOT NULL, + UNIQUE (job_id, sequence) + ) + `); + await pool.query(` + CREATE INDEX IF NOT EXISTS executor_job_events_job_sequence_idx + ON ${eventTable} (job_id, sequence) + `); + await pool.query(` + INSERT INTO ${eventTable} (event_id, job_id, sequence, event_json, created_at) + SELECT + 'evt_' || md5(job_id || ':snapshot-import'), + job_id, + 1, + jsonb_build_object( + 'version', 'executor-job-event-v1', + 'eventId', 'evt_' || md5(job_id || ':snapshot-import'), + 'jobId', job_id, + 'sequence', 1, + 'type', 'executor_job_snapshot_imported', + 'timestamp', created_at, + 'data', jsonb_build_object( + 'status', state_json->>'status', + 'reason', state_json->>'reason' + ) + ), + created_at + FROM ${table} jobs + WHERE NOT EXISTS ( + SELECT 1 FROM ${eventTable} events WHERE events.job_id = jobs.job_id + ) + ON CONFLICT DO NOTHING + `); }, async getById(jobId) { @@ -79,23 +154,40 @@ export function createPostgresExecutorJobStore({ return projectRow(result.rows?.[0]); }, - async createIfAbsent(record) { - const inserted = await pool.query( - `INSERT INTO ${table} - (job_id, idempotency_key, request_fingerprint, state_json, created_at, updated_at) - VALUES ($1, $2, $3, $4::jsonb, $5, $6) - ON CONFLICT DO NOTHING - RETURNING state_json`, - [ - record.jobId, - record.idempotencyKey, - record.requestFingerprint, - JSON.stringify(record), - record.createdAt, - record.updatedAt, - ], - ); - const created = projectRow(inserted.rows?.[0]); + async createIfAbsent(record, { initialEvent = null } = {}) { + const created = await withTransaction(pool, async (client) => { + const inserted = await client.query( + `INSERT INTO ${table} + (job_id, idempotency_key, request_fingerprint, state_json, created_at, updated_at) + VALUES ($1, $2, $3, $4::jsonb, $5, $6) + ON CONFLICT DO NOTHING + RETURNING state_json`, + [ + record.jobId, + record.idempotencyKey, + record.requestFingerprint, + JSON.stringify(record), + record.createdAt, + record.updatedAt, + ], + ); + const insertedRecord = projectRow(inserted.rows?.[0]); + if (!insertedRecord || !initialEvent) return insertedRecord; + const event = eventWithSequence(initialEvent, record.jobId, 1); + await client.query( + `INSERT INTO ${eventTable} + (event_id, job_id, sequence, event_json, created_at) + VALUES ($1, $2, $3, $4::jsonb, $5)`, + [ + event.eventId, + record.jobId, + event.sequence, + JSON.stringify(event), + event.timestamp, + ], + ); + return insertedRecord; + }); if (created) return { created: true, record: created }; const existingResult = await pool.query( @@ -111,16 +203,59 @@ export function createPostgresExecutorJobStore({ throw error; }, - async update(record) { + async update(record, { event = null } = {}) { + return withTransaction(pool, async (client) => { + const locked = await client.query( + `SELECT job_id FROM ${table} WHERE job_id = $1 FOR UPDATE`, + [record.jobId], + ); + if (!locked.rows?.[0]) return null; + const result = await client.query( + `UPDATE ${table} + SET state_json = $2::jsonb, + updated_at = $3 + WHERE job_id = $1 + RETURNING state_json`, + [record.jobId, JSON.stringify(record), record.updatedAt], + ); + if (event) { + const sequenceResult = await client.query( + `SELECT COALESCE(MAX(sequence), 0) + 1 AS next_sequence + FROM ${eventTable} + WHERE job_id = $1`, + [record.jobId], + ); + const sequence = Number(sequenceResult.rows?.[0]?.next_sequence ?? 1); + const persistedEvent = eventWithSequence(event, record.jobId, sequence); + await client.query( + `INSERT INTO ${eventTable} + (event_id, job_id, sequence, event_json, created_at) + VALUES ($1, $2, $3, $4::jsonb, $5)`, + [ + persistedEvent.eventId, + record.jobId, + sequence, + JSON.stringify(persistedEvent), + persistedEvent.timestamp, + ], + ); + } + return projectRow(result.rows?.[0]); + }); + }, + + async listEvents(jobId, { after = 0, limit = 500 } = {}) { + const cursor = Math.max(0, Number(after) || 0); + const pageSize = Math.min(500, Math.max(1, Number(limit) || 500)); const result = await pool.query( - `UPDATE ${table} - SET state_json = $2::jsonb, - updated_at = $3 - WHERE job_id = $1 - RETURNING state_json`, - [record.jobId, JSON.stringify(record), record.updatedAt], + `SELECT event_json + FROM ${eventTable} + WHERE job_id = $1 AND sequence > $2 + ORDER BY sequence ASC + LIMIT $3`, + [String(jobId ?? '').trim(), cursor, pageSize], ); - return projectRow(result.rows?.[0]); + return (result.rows ?? []).map(projectEventRow).filter(Boolean); }, }; } @@ -184,8 +319,12 @@ export async function createExecutorJobPersistence({ export const executorJobStoreInternals = { DEFAULT_EXECUTOR_SCHEMA, EXECUTOR_JOB_TABLE, + EXECUTOR_JOB_EVENT_TABLE, + eventWithSequence, normalizeMode, normalizeSchema, parseState, + projectEventRow, projectRow, + withTransaction, }; diff --git a/services/orchestrator/executor-job-store.test.mjs b/services/orchestrator/executor-job-store.test.mjs index 0bf0bd1..0a99bf7 100644 --- a/services/orchestrator/executor-job-store.test.mjs +++ b/services/orchestrator/executor-job-store.test.mjs @@ -19,6 +19,18 @@ function record(overrides = {}) { }; } +function transactionalPool(query) { + return { + query, + async connect() { + return { + query, + release() {}, + }; + }, + }; +} + test('PostgreSQL Executor Job Store creates an isolated schema and job table', async () => { const queries = []; const store = createPostgresExecutorJobStore({ @@ -37,53 +49,85 @@ test('PostgreSQL Executor Job Store creates an isolated schema and job table', a assert.match(queries[1].sql, /"memind_orchestrator_test"\."executor_jobs"/); assert.match(queries[1].sql, /idempotency_key VARCHAR\(200\) NOT NULL UNIQUE/); assert.match(queries[2].sql, /executor_jobs_updated_at_idx/); + assert.match(queries[3].sql, /"memind_orchestrator_test"\."executor_job_events"/); + assert.match(queries[3].sql, /UNIQUE \(job_id, sequence\)/); + assert.match(queries[5].sql, /executor_job_snapshot_imported/); }); test('PostgreSQL Executor Job Store atomically creates and reloads idempotent state', async () => { const stored = record(); + const blockedEvent = { + version: 'executor-job-event-v1', + eventId: 'event-1', + jobId: 'job-1', + sequence: 1, + type: 'executor_job_blocked', + timestamp: 1000, + data: { status: 'blocked' }, + }; const calls = []; const store = createPostgresExecutorJobStore({ - pool: { - async query(sql, params = []) { - calls.push({ sql, params }); - if (sql.includes('INSERT INTO')) { - return { rows: [{ state_json: stored }] }; - } - if (sql.includes('UPDATE')) { - return { rows: [{ state_json: { ...stored, status: 'cancelled' } }] }; - } - if (sql.includes('WHERE job_id =')) { - return { rows: [{ state_json: JSON.stringify(stored) }] }; - } - if (sql.includes('WHERE idempotency_key =')) { - return { rows: [{ state_json: stored }] }; - } + pool: transactionalPool(async (sql, params = []) => { + calls.push({ sql, params }); + if (sql.includes('INSERT INTO') && sql.includes('executor_jobs')) { + return { rows: [{ state_json: stored }] }; + } + if (sql.includes('INSERT INTO') && sql.includes('executor_job_events')) { return { rows: [] }; - }, - }, + } + if (sql.includes('SELECT job_id') && sql.includes('FOR UPDATE')) { + return { rows: [{ job_id: 'job-1' }] }; + } + if (sql.includes('UPDATE')) { + return { rows: [{ state_json: { ...stored, status: 'cancelled' } }] }; + } + if (sql.includes('MAX(sequence)')) { + return { rows: [{ next_sequence: 2 }] }; + } + if (sql.includes('SELECT event_json')) { + return { rows: [{ event_json: blockedEvent }] }; + } + if (sql.includes('WHERE job_id =')) { + return { rows: [{ state_json: JSON.stringify(stored) }] }; + } + if (sql.includes('WHERE idempotency_key =')) { + return { rows: [{ state_json: stored }] }; + } + return { rows: [] }; + }), }); - const created = await store.createIfAbsent(stored); + const created = await store.createIfAbsent(stored, { initialEvent: blockedEvent }); assert.equal(created.created, true); assert.deepEqual(created.record, stored); assert.deepEqual(await store.getById('job-1'), stored); assert.deepEqual(await store.getByIdempotencyKey('idem-1'), stored); - assert.equal((await store.update({ ...stored, status: 'cancelled' })).status, 'cancelled'); - assert.equal(JSON.parse(calls[0].params[3]).jobId, 'job-1'); + assert.equal( + (await store.update( + { ...stored, status: 'cancelled' }, + { event: { ...blockedEvent, eventId: 'event-2', type: 'executor_job_cancelled' } }, + )).status, + 'cancelled', + ); + assert.deepEqual(await store.listEvents('job-1'), [blockedEvent]); + const jobInsert = calls.find((call) => ( + call.sql.includes('INSERT INTO') && call.sql.includes('executor_jobs') + )); + assert.equal(JSON.parse(jobInsert.params[3]).jobId, 'job-1'); + assert.equal(calls.filter((call) => call.sql.trim() === 'BEGIN').length, 2); + assert.equal(calls.filter((call) => call.sql.trim() === 'COMMIT').length, 2); }); test('PostgreSQL Executor Job Store returns the existing idempotent record after conflict', async () => { const stored = record(); const store = createPostgresExecutorJobStore({ - pool: { - async query(sql) { - if (sql.includes('INSERT INTO')) return { rows: [] }; - if (sql.includes('WHERE idempotency_key =')) { - return { rows: [{ state_json: stored }] }; - } - return { rows: [] }; - }, - }, + pool: transactionalPool(async (sql) => { + if (sql.includes('INSERT INTO')) return { rows: [] }; + if (sql.includes('WHERE idempotency_key =')) { + return { rows: [{ state_json: stored }] }; + } + return { rows: [] }; + }), }); assert.deepEqual( await store.createIfAbsent(stored), @@ -91,6 +135,37 @@ test('PostgreSQL Executor Job Store returns the existing idempotent record after ); }); +test('PostgreSQL Executor Job Store rolls back job creation when its initial event fails', async () => { + const calls = []; + const eventError = new Error('event insert unavailable'); + const store = createPostgresExecutorJobStore({ + pool: transactionalPool(async (sql) => { + calls.push(sql.trim()); + if (sql.includes('INSERT INTO') && sql.includes('executor_jobs')) { + return { rows: [{ state_json: record() }] }; + } + if (sql.includes('INSERT INTO') && sql.includes('executor_job_events')) { + throw eventError; + } + return { rows: [] }; + }), + }); + + await assert.rejects( + store.createIfAbsent(record(), { + initialEvent: { + eventId: 'event-failing', + jobId: 'job-1', + timestamp: 1000, + }, + }), + eventError, + ); + assert.ok(calls.includes('BEGIN')); + assert.ok(calls.includes('ROLLBACK')); + assert.ok(!calls.includes('COMMIT')); +}); + test('Executor Job persistence supports explicit non-durable memory mode', async () => { const persistence = await createExecutorJobPersistence({ mode: 'memory' }); assert.equal(persistence.kind, 'memory'); diff --git a/services/orchestrator/observability.mjs b/services/orchestrator/observability.mjs index 3eb126e..5ea5477 100644 --- a/services/orchestrator/observability.mjs +++ b/services/orchestrator/observability.mjs @@ -48,6 +48,11 @@ function normalizeRunId(value) { return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : ''; } +function normalizeExecutorJobId(value) { + const jobId = String(value ?? '').trim(); + return /^[a-zA-Z0-9:_-]{1,128}$/.test(jobId) ? jobId : ''; +} + function projectShadowEventRow(row) { const data = parseJsonColumn(row.data_json, {}) ?? {}; const succeeded = row.event_type === 'workflow_shadow_completed'; @@ -159,6 +164,7 @@ function buildCanaryReadiness(loaded, runtimeState, now) { const latencyP95Ms = percentile(latencies, 0.95); const serviceHealth = runtimeState?.serviceHealth ?? null; const checkpoint = serviceHealth?.details?.checkpoint ?? null; + const executorStore = serviceHealth?.details?.executorGateway?.store ?? null; const checks = [ { @@ -179,6 +185,12 @@ function buildCanaryReadiness(loaded, runtimeState, now) { actual: checkpoint?.durable ?? null, target: true, }, + { + id: 'durable_executor_job_store', + passed: executorStore?.durable === true, + actual: executorStore?.durable ?? null, + target: true, + }, { id: 'observe_only', passed: serviceHealth?.details?.execution === 'observe-only', @@ -274,6 +286,8 @@ function buildCanaryReadiness(loaded, runtimeState, now) { latencyMs: serviceHealth?.latencyMs ?? null, checkpointKind: checkpoint?.kind ?? null, checkpointDurable: checkpoint?.durable ?? null, + executorJobStoreKind: executorStore?.kind ?? null, + executorJobStoreDurable: executorStore?.durable ?? null, execution: serviceHealth?.details?.execution ?? null, }, checks, @@ -513,6 +527,12 @@ export function createOrchestratorObservabilityService({ state: null, events: [], error: null, + executorJob: { + available: false, + state: null, + events: [], + error: null, + }, }; try { const runtimeState = await configService.getRuntimeState(); @@ -534,11 +554,42 @@ export function createOrchestratorObservabilityService({ return collected; })(), ]); + const executorJobId = normalizeExecutorJobId(state?.plan?.executorJob?.id); + let executorJob = { + available: false, + state: null, + events: [], + error: null, + }; + if (executorJobId) { + try { + const [jobState, jobEvents] = await Promise.all([ + engine.getExecutorJob(executorJobId), + (async () => { + const collected = []; + for await (const event of engine.streamExecutorJobEvents(executorJobId)) { + collected.push(event); + if (collected.length >= 500) break; + } + return collected; + })(), + ]); + executorJob = { + available: true, + state: jobState, + events: jobEvents, + error: null, + }; + } catch (error) { + executorJob.error = safeRemoteError(error); + } + } remote = { available: true, state, events, error: null, + executorJob, }; } } catch (error) { @@ -572,6 +623,7 @@ export const orchestratorObservabilityInternals = { MAX_METRIC_EVENTS, SHADOW_EVENT_TYPES, buildCanaryReadiness, + normalizeExecutorJobId, normalizeRunId, parseJsonColumn, percentile, diff --git a/services/orchestrator/observability.test.mjs b/services/orchestrator/observability.test.mjs index 1499275..b201ffc 100644 --- a/services/orchestrator/observability.test.mjs +++ b/services/orchestrator/observability.test.mjs @@ -227,6 +227,18 @@ test('observability detail joins Native state with remote LangGraph checkpoint e serviceToken: 'detail-token', fetchImpl: async (url, init) => { assert.equal(init.headers.authorization, 'Bearer detail-token'); + if (String(url).includes('/executor-jobs/') && String(url).endsWith('/events')) { + return response({ + events: [{ sequence: 1, type: 'executor_job_blocked' }], + }); + } + if (String(url).includes('/executor-jobs/')) { + return response({ + jobId: 'run-detail:executor-preview', + status: 'blocked', + reason: 'executor_dispatch_not_implemented', + }); + } if (String(url).endsWith('/events')) { return response({ events: [{ sequence: 1, type: 'workflow_validated' }], @@ -235,7 +247,10 @@ test('observability detail joins Native state with remote LangGraph checkpoint e return response({ runId: 'run-detail', status: 'succeeded', - plan: { executorAdapter: 'native-agent-run' }, + plan: { + executorAdapter: 'native-agent-run', + executorJob: { id: 'run-detail:executor-preview' }, + }, }); }, }); @@ -246,6 +261,9 @@ test('observability detail joins Native state with remote LangGraph checkpoint e assert.equal(detail.remote.available, true); assert.equal(detail.remote.state.plan.executorAdapter, 'native-agent-run'); assert.equal(detail.remote.events[0].type, 'workflow_validated'); + assert.equal(detail.remote.executorJob.available, true); + assert.equal(detail.remote.executorJob.state.status, 'blocked'); + assert.equal(detail.remote.executorJob.events[0].type, 'executor_job_blocked'); assert.equal(queries.length, 2); }); @@ -327,6 +345,9 @@ test('canary readiness excludes smoke runs and reports explicit blockers', async latencyMs: 5, details: { checkpoint: { kind: 'postgres', durable: true }, + executorGateway: { + store: { kind: 'postgres', durable: true }, + }, execution: 'observe-only', }, }, @@ -384,6 +405,9 @@ test('canary readiness passes only when operational and sample gates all pass', latencyMs: 4, details: { checkpoint: { kind: 'postgres', durable: true }, + executorGateway: { + store: { kind: 'postgres', durable: true }, + }, execution: 'observe-only', }, }, @@ -402,6 +426,7 @@ test('canary readiness passes only when operational and sample gates all pass', assert.equal(readiness.samples.latencyCoverageRate, 1); assert.equal(readiness.samples.nativeSettledRate, 1); assert.equal(readiness.samples.distinctSessions, 5); + assert.equal(readiness.service.executorJobStoreDurable, true); assert.equal(readiness.blockers.length, 0); assert.deepEqual(readiness.failureCodes, [{ code: 'TRANSIENT', count: 1 }]); }); diff --git a/services/orchestrator/runtime.mjs b/services/orchestrator/runtime.mjs index b798863..48145bf 100644 --- a/services/orchestrator/runtime.mjs +++ b/services/orchestrator/runtime.mjs @@ -129,6 +129,14 @@ export function createLangGraphOrchestratorRuntime({ throw error; }, + getExecutorJob(jobId) { + return executorGateway.getJob(jobId); + }, + + listExecutorJobEvents(jobId, options = {}) { + return executorGateway.listEvents(jobId, options); + }, + health, async ready() { diff --git a/services/orchestrator/runtime.test.mjs b/services/orchestrator/runtime.test.mjs index 154ee13..50d7e18 100644 --- a/services/orchestrator/runtime.test.mjs +++ b/services/orchestrator/runtime.test.mjs @@ -44,6 +44,18 @@ test('LangGraph runtime checkpoints a deterministic observe-only code workflow', reason: 'executor_dispatch_not_implemented', durable: false, }); + assert.equal( + (await runtime.getExecutorJob('run-shadow-1:executor-preview')).status, + 'blocked', + ); + const executorEvents = await runtime.listExecutorJobEvents( + 'run-shadow-1:executor-preview', + ); + assert.deepEqual( + executorEvents.events.map((event) => event.type), + ['executor_job_blocked'], + ); + assert.equal(executorEvents.nextCursor, 1); const restored = await runtime.getState('run-shadow-1'); assert.deepEqual(restored, state); @@ -128,3 +140,16 @@ test('LangGraph runtime rejects execution-enabled and unsupported workflows', as (error) => error.code === 'WORKFLOW_NOT_SUPPORTED' && error.status === 422, ); }); + +test('LangGraph runtime derives a bounded opaque Executor Job id for long run ids', async () => { + const runtime = createLangGraphOrchestratorRuntime({ + checkpointer: new MemorySaver(), + }); + const state = await runtime.start(runSpec({ + runId: 'r'.repeat(128), + requestId: 'request-long-run-id', + })); + assert.match(state.plan.executorJob.id, /^run_[a-f0-9]{64}:executor-preview$/); + assert.ok(state.plan.executorJob.id.length <= 128); + assert.equal((await runtime.getExecutorJob(state.plan.executorJob.id)).status, 'blocked'); +});