feat: add executor gateway contracts
This commit is contained in:
john
@@ -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,
|
||||
};
|
||||
Reference in New Issue
Block a user