From d21c4849a9afe3340759d196132298cce82ed02d Mon Sep 17 00:00:00 2001 From: john Date: Fri, 24 Jul 2026 22:50:37 +0800 Subject: [PATCH] feat: add executor gateway contracts --- .../memind-orchestrator-boundary.md | 15 + ops/src/api/admin.ts | 11 + ops/src/pages/admin/OrchestratorPage.tsx | 30 ++ services/orchestrator/README.md | 17 + services/orchestrator/admin-config.mjs | 3 + services/orchestrator/admin-config.test.mjs | 9 + services/orchestrator/executor-gateway.mjs | 418 ++++++++++++++++++ .../orchestrator/executor-gateway.test.mjs | 144 ++++++ 8 files changed, 647 insertions(+) create mode 100644 services/orchestrator/executor-gateway.mjs create mode 100644 services/orchestrator/executor-gateway.test.mjs diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index 29f36b9..339b283 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -86,6 +86,21 @@ The framework-neutral contracts are: - `orchestrator-event-v1` - `workflow-execution-request-v1` - `workflow-execution-decision-v1` +- `executor-job-request-v1` +- `executor-job-state-v1` +- `executor-dispatch-decision-v1` + +The Phase 3.2 Executor Gateway owns the adapter registry and executor job state +transition boundary. Requests contain workspace/artifact references, +authorization and side-effect policy, timeout, cancellation, fallback, and an +idempotency key. The store interface requires atomic create-if-absent semantics +so retries return the original job and conflicting payloads fail closed. + +Goosed, Aider, and OpenHands currently appear only as disabled +`contract-only` adapters. The reference in-memory store is non-durable and is +not a deployment backend. The compile-time dispatch capability remains false, +so creating a job records `blocked` state without importing an executor SDK, +launching a process, or touching a workspace. The implemented internal API is: diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index ade9106..4fd8d61 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -237,6 +237,16 @@ export type OrchestratorEngineDescriptor = { capabilities: string[]; }; +export type OrchestratorExecutorDescriptor = { + id: string; + label: string; + kind: string; + enabled: boolean; + dispatchImplemented: boolean; + status: string; + capabilities: string[]; +}; + export type OrchestratorConfigState = { config: OrchestratorConfig; configVersion: number; @@ -245,6 +255,7 @@ export type OrchestratorConfigState = { source: string; runtime: OrchestratorRuntime; engines: OrchestratorEngineDescriptor[]; + executors: OrchestratorExecutorDescriptor[]; }; export type OrchestratorServiceHealth = { diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx index 352546a..01e4c36 100644 --- a/ops/src/pages/admin/OrchestratorPage.tsx +++ b/ops/src/pages/admin/OrchestratorPage.tsx @@ -169,6 +169,36 @@ export function OrchestratorPage() {

+
+
+

Executor Gateway

+

+ Phase 3.2 仅注册版本化适配器契约；所有执行器均为 contract-only，不会启动进程或接管任务。 +

+
+
+ {(state.executors ?? []).map((executor) => ( +
+
+ {executor.label} + + {executor.status} + +
+

+ {executor.capabilities.join(' · ') || '无能力声明'} +

+ + dispatch {executor.dispatchImplemented ? 'implemented' : 'disabled'} + +
+ ))} +
+
+ diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md index 3b5447a..0002b9f 100644 --- a/services/orchestrator/README.md +++ b/services/orchestrator/README.md @@ -93,6 +93,23 @@ reasons, task-type distribution, Native terminal coverage, recent decisions, and an explicit alert if any stored decision ever reports `handoffAllowed=true`. +Phase 3.2 establishes the Executor Gateway boundary: + +- `executor-job-request-v1`, `executor-job-state-v1`, and + `executor-dispatch-decision-v1` remain independent of Goosed, Aider, + OpenHands, and LangGraph SDK types; +- jobs use workspace and artifact references rather than host filesystem paths; +- the store contract provides atomic idempotency, lookup, and state update + operations; +- timeout, cancellation, fallback, authorization, network, and side-effect + controls are normalized before adapter selection; +- Goosed, Aider, and OpenHands are registered as `contract-only` adapters; +- the reference memory store is explicitly non-durable and used only for local + contract tests. + +`EXECUTOR_DISPATCH_IMPLEMENTED=false` is a code-level gate. Phase 3.2 can record +blocked, idempotent job state but cannot invoke an adapter or launch a process. + 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. diff --git a/services/orchestrator/admin-config.mjs b/services/orchestrator/admin-config.mjs index 3e13fda..aa00113 100644 --- a/services/orchestrator/admin-config.mjs +++ b/services/orchestrator/admin-config.mjs @@ -7,6 +7,7 @@ import { normalizeWorkflowName, stableRolloutBucket, } from './contracts.mjs'; +import { listPhase3ExecutorAdapters } from './executor-gateway.mjs'; const CONFIG_TABLE = 'h5_orchestrator_admin_config'; const CONFIG_SCOPE = 'global'; @@ -301,6 +302,7 @@ export function createOrchestratorAdminConfigService(pool, { ...state, runtime: runtimeState(state.config, env), engines: engineCatalog(state.config), + executors: listPhase3ExecutorAdapters(), }; }, @@ -333,6 +335,7 @@ export function createOrchestratorAdminConfigService(pool, { ...state, runtime: runtimeState(state.config, env), engines: engineCatalog(state.config), + executors: listPhase3ExecutorAdapters(), ...(probe ? { serviceHealth: await probeServiceHealth(state.config, { fetchImpl }) } : {}), }; }, diff --git a/services/orchestrator/admin-config.test.mjs b/services/orchestrator/admin-config.test.mjs index 610ee2c..6b37c8d 100644 --- a/services/orchestrator/admin-config.test.mjs +++ b/services/orchestrator/admin-config.test.mjs @@ -35,6 +35,15 @@ test('orchestrator config defaults to a disabled native-safe runtime', async () assert.equal(state.runtime.reason, 'mode_off'); assert.equal(state.engines.find((engine) => engine.id === 'native').configured, true); assert.equal(state.engines.find((engine) => engine.id === 'langgraph').configured, false); + assert.deepEqual( + state.executors.map((executor) => executor.id), + ['goosed', 'aider', 'openhands'], + ); + assert.equal(state.executors.every((executor) => executor.enabled === false), true); + assert.equal( + state.executors.every((executor) => executor.dispatchImplemented === false), + true, + ); }); test('orchestrator config normalizes unsafe values and keeps a workflow allowlist', () => { diff --git a/services/orchestrator/executor-gateway.mjs b/services/orchestrator/executor-gateway.mjs new file mode 100644 index 0000000..c5b6245 --- /dev/null +++ b/services/orchestrator/executor-gateway.mjs @@ -0,0 +1,418 @@ +import crypto from 'node:crypto'; +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_DISPATCH_IMPLEMENTED = false; +const MAX_INSTRUCTION_CHARACTERS = 32_000; +const TERMINAL_JOB_STATUSES = new Set([ + 'succeeded', + 'failed', + 'cancelled', + 'timed_out', + 'blocked', +]); + +const PHASE3_EXECUTOR_DESCRIPTORS = Object.freeze([ + Object.freeze({ + id: 'goosed', + label: 'Goosed', + kind: 'remote', + capabilities: Object.freeze(['session-execution', 'streaming', 'tools']), + }), + Object.freeze({ + id: 'aider', + label: 'Aider', + kind: 'worker', + capabilities: Object.freeze(['code-edit', 'multi-file']), + }), + Object.freeze({ + id: 'openhands', + label: 'OpenHands', + kind: 'worker', + capabilities: Object.freeze(['repo-task', 'code-edit', 'command-execution']), + }), +]); + +function clone(value) { + return typeof structuredClone === 'function' + ? structuredClone(value) + : JSON.parse(JSON.stringify(value)); +} + +function clampInteger(value, fallback, min, max) { + const number = Number(value); + if (!Number.isFinite(number)) return fallback; + return Math.min(max, Math.max(min, Math.floor(number))); +} + +function normalizeReference(value, fallbackKind) { + const source = value && typeof value === 'object' && !Array.isArray(value) ? value : {}; + const kind = String(source.kind ?? fallbackKind ?? '').trim().slice(0, 64); + const id = String(source.id ?? '').trim().slice(0, 256); + return kind && id ? { kind, id } : null; +} + +function stableValue(value) { + if (Array.isArray(value)) return value.map(stableValue); + if (!value || typeof value !== 'object') return value; + return Object.fromEntries( + Object.keys(value) + .sort() + .map((key) => [key, stableValue(value[key])]), + ); +} + +function fingerprint(value) { + return crypto + .createHash('sha256') + .update(JSON.stringify(stableValue(value))) + .digest('hex'); +} + +function gatewayError(code, message, status = 409) { + const error = new Error(message); + error.code = code; + error.status = status; + return error; +} + +export function normalizeExecutorJobRequest(input = {}) { + const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {}; + const jobId = String(source.jobId ?? crypto.randomUUID()).trim(); + const idempotencyKey = String(source.idempotencyKey ?? '').trim().slice(0, 200); + const executor = normalizeWorkflowEngineId(source.executor, ''); + const instruction = String(source.task?.instruction ?? source.instruction ?? '') + .trim() + .slice(0, MAX_INSTRUCTION_CHARACTERS); + if (!jobId) throw gatewayError('EXECUTOR_JOB_ID_REQUIRED', 'Executor job requires jobId', 422); + if (!idempotencyKey) { + throw gatewayError( + 'EXECUTOR_IDEMPOTENCY_KEY_REQUIRED', + 'Executor job requires idempotencyKey', + 422, + ); + } + if (!executor) { + throw gatewayError('EXECUTOR_ID_REQUIRED', 'Executor job requires executor', 422); + } + if (!instruction) { + throw gatewayError('EXECUTOR_INSTRUCTION_REQUIRED', 'Executor job requires instruction', 422); + } + return { + version: EXECUTOR_JOB_REQUEST_VERSION, + jobId, + idempotencyKey, + executor, + task: { + type: String(source.task?.type ?? 'code_task').trim().slice(0, 128) || 'code_task', + instruction, + workspaceRef: normalizeReference(source.task?.workspaceRef, 'workspace'), + inputRefs: Array.isArray(source.task?.inputRefs) + ? source.task.inputRefs + .map((item) => normalizeReference(item, 'artifact')) + .filter(Boolean) + .slice(0, 100) + : [], + }, + subject: { + tenantId: String(source.subject?.tenantId ?? '').trim() || null, + userId: String(source.subject?.userId ?? '').trim() || null, + }, + authorization: { + executionAllowed: source.authorization?.executionAllowed === true, + actorId: String(source.authorization?.actorId ?? '').trim() || null, + }, + policy: { + sideEffectsAllowed: source.policy?.sideEffectsAllowed === true, + networkAllowed: source.policy?.networkAllowed === true, + }, + controls: { + timeoutMs: clampInteger(source.controls?.timeoutMs, 15 * 60_000, 500, 60 * 60_000), + cancellationAllowed: source.controls?.cancellationAllowed !== false, + fallbackExecutor: normalizeWorkflowEngineId(source.controls?.fallbackExecutor, ''), + }, + metadata: source.metadata && typeof source.metadata === 'object' + ? clone(source.metadata) + : {}, + }; +} + +function validateExecutorAdapter(adapter) { + if (!adapter || typeof adapter !== 'object') { + throw new Error('Executor adapter must be an object'); + } + const id = normalizeWorkflowEngineId(adapter.id, ''); + if (!id) throw new Error('Executor adapter requires a valid id'); + for (const method of ['submit', 'cancel', 'getState', 'streamEvents']) { + if (typeof adapter[method] !== 'function') { + throw new Error(`Executor adapter ${id} missing method: ${method}`); + } + } + return id; +} + +export function createExecutorAdapterRegistry(initialAdapters = []) { + const adapters = new Map(); + + function register(adapter) { + const id = validateExecutorAdapter(adapter); + if (adapters.has(id)) throw new Error(`Executor adapter already registered: ${id}`); + adapters.set(id, adapter); + return adapter; + } + + for (const adapter of initialAdapters) register(adapter); + + return { + register, + has(id) { + return adapters.has(normalizeWorkflowEngineId(id, '')); + }, + get(id) { + return adapters.get(normalizeWorkflowEngineId(id, '')) ?? null; + }, + list() { + return [...adapters.values()].map((adapter) => ({ + id: adapter.id, + label: adapter.label ?? adapter.id, + kind: adapter.kind ?? 'plugin', + enabled: adapter.enabled === true, + dispatchImplemented: adapter.dispatchImplemented === true, + status: adapter.status ?? 'unknown', + capabilities: Array.isArray(adapter.capabilities) + ? [...adapter.capabilities] + : [], + })); + }, + }; +} + +export function createDisabledExecutorAdapter(descriptor) { + const id = normalizeWorkflowEngineId(descriptor?.id, ''); + if (!id) throw new Error('Disabled executor adapter requires a valid id'); + const reject = () => { + throw gatewayError( + 'EXECUTOR_ADAPTER_DISABLED', + `Executor adapter ${id} is contract-only and cannot dispatch`, + ); + }; + return { + id, + label: descriptor.label ?? id, + kind: descriptor.kind ?? 'plugin', + enabled: false, + dispatchImplemented: false, + status: 'contract-only', + capabilities: Array.isArray(descriptor.capabilities) + ? [...descriptor.capabilities] + : [], + submit: reject, + cancel: reject, + getState: reject, + async *streamEvents() { + reject(); + }, + }; +} + +export function createPhase3ExecutorAdapterRegistry() { + return createExecutorAdapterRegistry( + PHASE3_EXECUTOR_DESCRIPTORS.map(createDisabledExecutorAdapter), + ); +} + +export function listPhase3ExecutorAdapters() { + return createPhase3ExecutorAdapterRegistry().list(); +} + +export function createInMemoryExecutorJobStore() { + const byId = new Map(); + const byIdempotencyKey = new Map(); + return { + durable: false, + async getById(jobId) { + const value = byId.get(String(jobId ?? '').trim()); + return value ? clone(value) : null; + }, + async getByIdempotencyKey(key) { + const jobId = byIdempotencyKey.get(String(key ?? '').trim()); + const value = jobId ? byId.get(jobId) : null; + return value ? clone(value) : null; + }, + async createIfAbsent(record) { + const existingId = byIdempotencyKey.get(record.idempotencyKey); + if (existingId) return { created: false, record: clone(byId.get(existingId)) }; + if (byId.has(record.jobId)) { + throw gatewayError('EXECUTOR_JOB_ID_CONFLICT', `Executor job id already exists: ${record.jobId}`); + } + byId.set(record.jobId, clone(record)); + byIdempotencyKey.set(record.idempotencyKey, record.jobId); + return { created: true, record: clone(record) }; + }, + async update(record) { + if (!byId.has(record.jobId)) return null; + byId.set(record.jobId, clone(record)); + return clone(record); + }, + }; +} + +function validateJobStore(store) { + for (const method of ['getById', 'getByIdempotencyKey', 'createIfAbsent', 'update']) { + if (typeof store?.[method] !== 'function') { + throw new Error(`Executor job store missing method: ${method}`); + } + } +} + +export function createExecutorGateway({ + registry = createPhase3ExecutorAdapterRegistry(), + store = createInMemoryExecutorJobStore(), + nowMs = Date.now, +} = {}) { + if (!registry?.has || !registry?.list) { + throw new Error('Executor Gateway requires an adapter registry'); + } + validateJobStore(store); + + function previewNormalized(request) { + const adapter = registry.get(request.executor); + const fallbackAdapter = request.controls.fallbackExecutor + ? registry.get(request.controls.fallbackExecutor) + : null; + const fallbackRegistered = Boolean(fallbackAdapter); + const fallbackAvailable = fallbackAdapter?.enabled === true + && fallbackAdapter?.dispatchImplemented === true; + const gates = { + implementation: EXECUTOR_DISPATCH_IMPLEMENTED, + adapterRegistered: Boolean(adapter), + adapterEnabled: adapter?.enabled === true, + executionAuthorized: request.authorization.executionAllowed, + sideEffectsAllowed: request.policy.sideEffectsAllowed, + idempotencyKeyPresent: Boolean(request.idempotencyKey), + workspaceReferencePresent: Boolean(request.task.workspaceRef), + }; + return { + version: 'executor-dispatch-decision-v1', + jobId: request.jobId, + executor: request.executor, + fallbackExecutor: request.controls.fallbackExecutor || null, + fallbackRegistered, + fallbackAvailable, + dispatchAllowed: Object.values(gates).every(Boolean), + reason: EXECUTOR_DISPATCH_IMPLEMENTED + ? 'executor_adapter_unavailable' + : 'executor_dispatch_not_implemented', + gates, + }; + } + + async function preview(input) { + return previewNormalized(normalizeExecutorJobRequest(input)); + } + + async function createJob(input) { + const request = normalizeExecutorJobRequest(input); + const { + jobId: _jobId, + idempotencyKey: _idempotencyKey, + ...semanticRequest + } = request; + const requestFingerprint = fingerprint(semanticRequest); + const existing = await store.getByIdempotencyKey(request.idempotencyKey); + if (existing) { + if (existing.requestFingerprint !== requestFingerprint) { + throw gatewayError( + 'EXECUTOR_IDEMPOTENCY_CONFLICT', + `Idempotency key already belongs to executor job ${existing.jobId}`, + ); + } + return { created: false, job: existing }; + } + const decision = previewNormalized(request); + const now = nowMs(); + const record = { + version: EXECUTOR_JOB_STATE_VERSION, + jobId: request.jobId, + idempotencyKey: request.idempotencyKey, + requestFingerprint, + executor: request.executor, + status: 'blocked', + reason: decision.reason, + attempts: 0, + dispatchAllowed: false, + cancellationAllowed: request.controls.cancellationAllowed, + timeoutMs: request.controls.timeoutMs, + fallbackExecutor: decision.fallbackExecutor, + fallbackRegistered: decision.fallbackRegistered, + fallbackAvailable: decision.fallbackAvailable, + decision, + createdAt: now, + updatedAt: now, + startedAt: null, + completedAt: now, + }; + const result = await store.createIfAbsent(record); + if ( + !result.created + && result.record.requestFingerprint !== requestFingerprint + ) { + throw gatewayError( + 'EXECUTOR_IDEMPOTENCY_CONFLICT', + `Idempotency key already belongs to executor job ${result.record.jobId}`, + ); + } + return { created: result.created, job: result.record }; + } + + async function cancel(jobId, { reason = 'cancelled_by_request' } = {}) { + const current = await store.getById(jobId); + if (!current) return null; + 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({ + ...current, + status: 'cancelled', + reason: String(reason).slice(0, 256), + updatedAt: now, + completedAt: now, + }); + } + + return { + preview, + createJob, + cancel, + getJob(jobId) { + return store.getById(jobId); + }, + listAdapters() { + return registry.list(); + }, + status() { + return { + version: 'executor-gateway-status-v1', + dispatchImplemented: EXECUTOR_DISPATCH_IMPLEMENTED, + executionEnabled: false, + store: { + kind: store.durable ? 'durable' : 'memory', + durable: store.durable === true, + }, + adapters: registry.list(), + }; + }, + }; +} + +export const executorGatewayInternals = { + EXECUTOR_DISPATCH_IMPLEMENTED, + EXECUTOR_JOB_REQUEST_VERSION, + EXECUTOR_JOB_STATE_VERSION, + MAX_INSTRUCTION_CHARACTERS, + PHASE3_EXECUTOR_DESCRIPTORS, + TERMINAL_JOB_STATUSES, + fingerprint, + stableValue, +}; diff --git a/services/orchestrator/executor-gateway.test.mjs b/services/orchestrator/executor-gateway.test.mjs new file mode 100644 index 0000000..b452465 --- /dev/null +++ b/services/orchestrator/executor-gateway.test.mjs @@ -0,0 +1,144 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + createExecutorAdapterRegistry, + createExecutorGateway, + createInMemoryExecutorJobStore, + createPhase3ExecutorAdapterRegistry, + normalizeExecutorJobRequest, +} from './executor-gateway.mjs'; + +function request(overrides = {}) { + return { + jobId: 'job-1', + idempotencyKey: 'run-1:node-execute:attempt-1', + executor: 'aider', + task: { + type: 'code_change', + instruction: 'Fix the login form validation', + workspaceRef: { kind: 'mindspace-workspace', id: 'workspace-1' }, + }, + subject: { userId: 'user-1' }, + authorization: { executionAllowed: true, actorId: 'user-1' }, + policy: { sideEffectsAllowed: true, networkAllowed: false }, + controls: { + timeoutMs: 20_000, + cancellationAllowed: true, + fallbackExecutor: 'goosed', + }, + ...overrides, + }; +} + +test('executor job contract uses resource references and bounded controls', () => { + const normalized = normalizeExecutorJobRequest(request({ + task: { + instruction: 'x'.repeat(40_000), + workspaceRef: { kind: 'workspace', id: 'workspace-1' }, + inputRefs: [ + { kind: 'artifact', id: 'artifact-1' }, + { kind: '', id: 'invalid' }, + ], + }, + controls: { timeoutMs: 100, fallbackExecutor: 'openhands' }, + })); + assert.equal(normalized.version, 'executor-job-request-v1'); + assert.equal(normalized.task.instruction.length, 32_000); + assert.deepEqual(normalized.task.workspaceRef, { kind: 'workspace', id: 'workspace-1' }); + assert.deepEqual(normalized.task.inputRefs, [{ kind: 'artifact', id: 'artifact-1' }]); + assert.equal(normalized.controls.timeoutMs, 500); + assert.equal(normalized.controls.fallbackExecutor, 'openhands'); + assert.equal('cwd' in normalized.task, false); +}); + +test('Phase 3 executor catalog exposes disabled Goosed, Aider and OpenHands adapters', () => { + const registry = createPhase3ExecutorAdapterRegistry(); + assert.deepEqual( + registry.list().map((adapter) => adapter.id), + ['goosed', 'aider', 'openhands'], + ); + for (const adapter of registry.list()) { + assert.equal(adapter.enabled, false); + assert.equal(adapter.dispatchImplemented, false); + assert.equal(adapter.status, 'contract-only'); + } +}); + +test('executor adapter registry rejects incomplete and duplicate adapters', () => { + const registry = createExecutorAdapterRegistry(); + assert.throws(() => registry.register({ id: 'broken' }), /missing method/); + const adapter = { + id: 'custom', + submit() {}, + cancel() {}, + getState() {}, + async *streamEvents() {}, + }; + registry.register(adapter); + assert.throws(() => registry.register(adapter), /already registered/); +}); + +test('Executor Gateway records a blocked idempotent job without calling an adapter', async () => { + let now = 1000; + const gateway = createExecutorGateway({ + store: createInMemoryExecutorJobStore(), + nowMs: () => now, + }); + const preview = await gateway.preview(request()); + assert.equal(preview.executor, 'aider'); + assert.equal(preview.fallbackExecutor, 'goosed'); + assert.equal(preview.fallbackRegistered, true); + assert.equal(preview.fallbackAvailable, false); + assert.equal(preview.dispatchAllowed, false); + assert.equal(preview.gates.implementation, false); + assert.equal(preview.gates.adapterEnabled, false); + + const first = await gateway.createJob(request()); + assert.equal(first.created, true); + assert.equal(first.job.status, 'blocked'); + assert.equal(first.job.attempts, 0); + assert.equal(first.job.reason, 'executor_dispatch_not_implemented'); + + now = 2000; + const repeated = await gateway.createJob(request()); + assert.equal(repeated.created, false); + assert.equal(repeated.job.jobId, first.job.jobId); + assert.equal(repeated.job.createdAt, 1000); +}); + +test('Executor Gateway rejects idempotency key reuse with a different payload', async () => { + const gateway = createExecutorGateway(); + await gateway.createJob(request()); + await assert.rejects( + () => gateway.createJob(request({ + jobId: 'job-2', + task: { + type: 'code_change', + instruction: 'A different instruction', + workspaceRef: { kind: 'mindspace-workspace', id: 'workspace-1' }, + }, + })), + (error) => error.code === 'EXECUTOR_IDEMPOTENCY_CONFLICT', + ); +}); + +test('Executor Gateway cancellation is idempotent and never invokes a disabled adapter', async () => { + let now = 1000; + const gateway = createExecutorGateway({ nowMs: () => now }); + await gateway.createJob(request()); + now = 1500; + const cancelled = await gateway.cancel('job-1', { reason: 'user_cancelled' }); + assert.equal(cancelled.status, 'cancelled'); + assert.equal(cancelled.reason, 'user_cancelled'); + assert.equal(cancelled.completedAt, 1500); + assert.deepEqual(await gateway.cancel('job-1'), cancelled); + assert.equal(await gateway.cancel('missing-job'), null); +}); + +test('Executor Gateway status remains hard-disabled with a non-durable reference store', () => { + const status = createExecutorGateway().status(); + assert.equal(status.version, 'executor-gateway-status-v1'); + assert.equal(status.dispatchImplemented, false); + assert.equal(status.executionEnabled, false); + assert.deepEqual(status.store, { kind: 'memory', durable: false }); +});