feat(control-plane): add shared orchestrator and in-memory adapters

compute/core/package.json

@@ -15,6 +15,7 @@
     ".": "./src/index.ts",
     "./local-runtime": "./src/local-runtime.ts",
     "./api-contracts": "./src/api-contracts/index.ts",
+    "./control-plane": "./src/control-plane/index.ts",
     "./types": "./src/types/index.ts"
   }
 }

compute/core/src/control-plane/in-memory.ts

@@ -0,0 +1,139 @@
+import { EventEmitter } from 'node:events';
+import type {
+  OperationEvent,
+  OperationEventStream,
+  OperationIndexEntry,
+  OperationQueue,
+  OperationState,
+  OperationStateStore,
+  QueuedOperation,
+} from './types';
+import type { WorkerOperationKind } from '../api-contracts';
+
+function topicFor(opId: string): string {
+  return `op.${opId}`;
+}
+
+function normalizeSinceEventId(value: number | undefined): number {
+  if (!Number.isFinite(value ?? 0)) return 0;
+  return Math.max(0, Math.floor(value ?? 0));
+}
+
+export class InMemoryOperationQueue implements OperationQueue {
+  private readonly byKind = new Map<WorkerOperationKind, QueuedOperation[]>();
+
+  constructor() {
+    this.byKind.set('whisper_align', []);
+    this.byKind.set('pdf_layout', []);
+  }
+
+  async enqueue(job: QueuedOperation): Promise<void> {
+    const list = this.byKind.get(job.kind);
+    if (!list) throw new Error(`Unsupported operation kind: ${job.kind}`);
+    list.push(job);
+  }
+
+  async claimNext(kind: WorkerOperationKind): Promise<QueuedOperation | null> {
+    const list = this.byKind.get(kind);
+    if (!list || list.length === 0) return null;
+    return list.shift() ?? null;
+  }
+
+  size(kind?: WorkerOperationKind): number {
+    if (kind) return this.byKind.get(kind)?.length ?? 0;
+    let total = 0;
+    for (const list of this.byKind.values()) total += list.length;
+    return total;
+  }
+}
+
+export class InMemoryOperationStateStore implements OperationStateStore {
+  private readonly stateByOpId = new Map<string, OperationState>();
+  private readonly opIndexByKey = new Map<string, string>();
+
+  async getOpState(opId: string): Promise<OperationState | null> {
+    return this.stateByOpId.get(opId) ?? null;
+  }
+
+  async putOpState(state: OperationState): Promise<void> {
+    this.stateByOpId.set(state.opId, state);
+  }
+
+  async getOpIndex(opKey: string): Promise<OperationIndexEntry | null> {
+    const opId = this.opIndexByKey.get(opKey);
+    return opId ? { opId } : null;
+  }
+
+  async compareAndSetOpIndex(input: {
+    opKey: string;
+    newOpId: string;
+    expectedOpId: string | null;
+  }): Promise<boolean> {
+    const current = this.opIndexByKey.get(input.opKey) ?? null;
+    if (current !== input.expectedOpId) return false;
+    this.opIndexByKey.set(input.opKey, input.newOpId);
+    return true;
+  }
+}
+
+export class InMemoryOperationEventStream implements OperationEventStream {
+  private readonly emitter = new EventEmitter();
+  private readonly lastIdByOpId = new Map<string, number>();
+  private readonly eventsByOpId = new Map<string, OperationEvent[]>();
+
+  constructor() {
+    this.emitter.setMaxListeners(0);
+  }
+
+  async append(opId: string, snapshot: OperationState): Promise<OperationEvent> {
+    const nextEventId = (this.lastIdByOpId.get(opId) ?? 0) + 1;
+    this.lastIdByOpId.set(opId, nextEventId);
+
+    const event: OperationEvent = {
+      eventId: nextEventId,
+      snapshot,
+    };
+
+    const list = this.eventsByOpId.get(opId) ?? [];
+    list.push(event);
+    this.eventsByOpId.set(opId, list);
+    this.emitter.emit(topicFor(opId), event);
+    return event;
+  }
+
+  async listSince(opId: string, sinceEventId: number, limit?: number): Promise<OperationEvent[]> {
+    const list = this.eventsByOpId.get(opId) ?? [];
+    const normalizedSince = normalizeSinceEventId(sinceEventId);
+    const filtered = list.filter((event) => event.eventId > normalizedSince);
+    if (!Number.isFinite(limit ?? 0) || !limit || limit <= 0) return filtered;
+    return filtered.slice(0, Math.floor(limit));
+  }
+
+  async subscribe(input: {
+    opId: string;
+    sinceEventId?: number;
+    onEvent: (event: OperationEvent) => void | Promise<void>;
+    onError?: (error: unknown) => void;
+  }): Promise<() => void> {
+    const replay = await this.listSince(input.opId, normalizeSinceEventId(input.sinceEventId));
+    for (const event of replay) {
+      try {
+        await input.onEvent(event);
+      } catch (error) {
+        input.onError?.(error);
+      }
+    }
+
+    const listener = (event: OperationEvent): void => {
+      Promise.resolve(input.onEvent(event)).catch((error) => {
+        input.onError?.(error);
+      });
+    };
+
+    const topic = topicFor(input.opId);
+    this.emitter.on(topic, listener);
+    return () => {
+      this.emitter.off(topic, listener);
+    };
+  }
+}

compute/core/src/control-plane/index.ts

@@ -0,0 +1,3 @@
+export * from './types';
+export * from './orchestrator';
+export * from './in-memory';

compute/core/src/control-plane/orchestrator.ts

@@ -0,0 +1,325 @@
+import { randomUUID } from 'node:crypto';
+import type {
+  WorkerJobErrorShape,
+  WorkerJobTiming,
+  WorkerOperationKind,
+  WorkerOperationRequest,
+  WorkerOperationState,
+} from '../api-contracts';
+import type {
+  OperationClock,
+  OperationEventStream,
+  OperationIdFactory,
+  OperationLifecycleConfig,
+  OperationQueue,
+  OperationStateStore,
+  QueuedOperation,
+} from './types';
+
+const RETRY_DELAY_MS = 25;
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function isTerminalStatus(status: WorkerOperationState['status']): boolean {
+  return status === 'succeeded' || status === 'failed';
+}
+
+function isInflightStatus(status: WorkerOperationState['status']): boolean {
+  return status === 'queued' || status === 'running';
+}
+
+function createErrorShape(error: unknown): WorkerJobErrorShape {
+  if (error && typeof error === 'object' && 'message' in error && typeof (error as { message: unknown }).message === 'string') {
+    return { message: (error as { message: string }).message };
+  }
+  return { message: String(error) };
+}
+
+function queuedStateFromRequest(input: {
+  request: WorkerOperationRequest;
+  opId: string;
+  jobId: string;
+  queuedAt: number;
+}): WorkerOperationState {
+  return {
+    opId: input.opId,
+    opKey: input.request.opKey,
+    kind: input.request.kind,
+    jobId: input.jobId,
+    status: 'queued',
+    queuedAt: input.queuedAt,
+    updatedAt: input.queuedAt,
+  };
+}
+
+function mapReplacementReason(input: {
+  current: WorkerOperationState;
+  requestKind: WorkerOperationKind;
+  now: number;
+  opStaleMs: number;
+}): string {
+  if (input.current.kind !== input.requestKind) return 'kind_mismatch';
+  const ageMs = input.now - input.current.updatedAt;
+  if (isInflightStatus(input.current.status) && ageMs > input.opStaleMs) return 'stale_running';
+  if (input.current.status === 'failed') return 'failed_prior';
+  return `status_${input.current.status}`;
+}
+
+export interface OperationOrchestratorDeps {
+  queue: OperationQueue;
+  stateStore: OperationStateStore;
+  eventStream: OperationEventStream;
+  config: OperationLifecycleConfig;
+  clock?: OperationClock;
+  idFactory?: OperationIdFactory;
+}
+
+export class OperationOrchestrator {
+  private readonly queue: OperationQueue;
+  private readonly stateStore: OperationStateStore;
+  private readonly eventStream: OperationEventStream;
+  private readonly opStaleMs: number;
+  private readonly maxCasRetries: number;
+  private readonly clock: OperationClock;
+  private readonly ids: OperationIdFactory;
+
+  constructor(deps: OperationOrchestratorDeps) {
+    this.queue = deps.queue;
+    this.stateStore = deps.stateStore;
+    this.eventStream = deps.eventStream;
+    this.opStaleMs = deps.config.opStaleMs;
+    this.maxCasRetries = Math.max(1, Math.floor(deps.config.maxCasRetries ?? 10));
+    this.clock = deps.clock ?? { now: () => Date.now() };
+    this.ids = deps.idFactory ?? {
+      opId: () => randomUUID(),
+      jobId: () => randomUUID(),
+    };
+  }
+
+  private async persistState(state: WorkerOperationState): Promise<void> {
+    await this.stateStore.putOpState(state);
+    await this.eventStream.append(state.opId, state);
+  }
+
+  private buildQueuedJob(state: WorkerOperationState, request: WorkerOperationRequest): QueuedOperation {
+    return {
+      jobId: state.jobId,
+      opId: state.opId,
+      opKey: state.opKey,
+      kind: state.kind,
+      queuedAt: state.queuedAt,
+      payload: request.payload,
+    };
+  }
+
+  async enqueueOrReuse(request: WorkerOperationRequest): Promise<WorkerOperationState> {
+    const opKey = request.opKey.trim();
+
+    for (let attempt = 0; attempt < this.maxCasRetries; attempt += 1) {
+      const indexEntry = await this.stateStore.getOpIndex(opKey);
+      if (indexEntry?.opId) {
+        const current = await this.stateStore.getOpState(indexEntry.opId);
+        if (!current) {
+          await sleep(RETRY_DELAY_MS);
+          continue;
+        }
+
+        const now = this.clock.now();
+        const ageMs = now - current.updatedAt;
+        if (current.kind === request.kind) {
+          if (current.status === 'succeeded') return current;
+          if (isInflightStatus(current.status) && ageMs <= this.opStaleMs) {
+            return current;
+          }
+        }
+
+        const replacement = queuedStateFromRequest({
+          request,
+          opId: this.ids.opId(),
+          jobId: this.ids.jobId(),
+          queuedAt: now,
+        });
+
+        const replaced = await this.stateStore.compareAndSetOpIndex({
+          opKey,
+          newOpId: replacement.opId,
+          expectedOpId: indexEntry.opId,
+        });
+        if (!replaced) {
+          await sleep(RETRY_DELAY_MS);
+          continue;
+        }
+
+        await this.persistState(replacement);
+
+        try {
+          await this.queue.enqueue(this.buildQueuedJob(replacement, request));
+          return replacement;
+        } catch (error) {
+          const failed: WorkerOperationState = {
+            ...replacement,
+            status: 'failed',
+            updatedAt: this.clock.now(),
+            error: createErrorShape(error),
+          };
+          await this.persistState(failed);
+          return failed;
+        }
+      }
+
+      const now = this.clock.now();
+      const created = queuedStateFromRequest({
+        request,
+        opId: this.ids.opId(),
+        jobId: this.ids.jobId(),
+        queuedAt: now,
+      });
+
+      const createdIndex = await this.stateStore.compareAndSetOpIndex({
+        opKey,
+        newOpId: created.opId,
+        expectedOpId: null,
+      });
+      if (!createdIndex) {
+        await sleep(RETRY_DELAY_MS);
+        continue;
+      }
+
+      await this.persistState(created);
+
+      try {
+        await this.queue.enqueue(this.buildQueuedJob(created, request));
+        return created;
+      } catch (error) {
+        const failed: WorkerOperationState = {
+          ...created,
+          status: 'failed',
+          updatedAt: this.clock.now(),
+          error: createErrorShape(error),
+        };
+        await this.persistState(failed);
+        return failed;
+      }
+    }
+
+    throw new Error('Unable to reserve operation after repeated CAS conflicts');
+  }
+
+  async getState(opId: string): Promise<WorkerOperationState | null> {
+    return this.stateStore.getOpState(opId);
+  }
+
+  async markRunning(input: {
+    opId: string;
+    startedAt?: number;
+    updatedAt?: number;
+    timing?: WorkerJobTiming;
+  }): Promise<WorkerOperationState> {
+    const current = await this.requireState(input.opId);
+    const now = input.updatedAt ?? this.clock.now();
+
+    const next: WorkerOperationState = {
+      ...current,
+      status: 'running',
+      startedAt: input.startedAt ?? current.startedAt ?? now,
+      updatedAt: now,
+      ...(input.timing ? { timing: input.timing } : {}),
+    };
+
+    await this.persistState(next);
+    return next;
+  }
+
+  async markProgress(input: {
+    opId: string;
+    progress: WorkerOperationState['progress'];
+    updatedAt?: number;
+    timing?: WorkerJobTiming;
+  }): Promise<WorkerOperationState> {
+    const current = await this.requireState(input.opId);
+    const now = input.updatedAt ?? this.clock.now();
+
+    const next: WorkerOperationState = {
+      ...current,
+      status: 'running',
+      startedAt: current.startedAt ?? now,
+      updatedAt: now,
+      progress: input.progress,
+      ...(input.timing ? { timing: input.timing } : {}),
+    };
+
+    await this.persistState(next);
+    return next;
+  }
+
+  async markSucceeded(input: {
+    opId: string;
+    result: unknown;
+    updatedAt?: number;
+    timing?: WorkerJobTiming;
+  }): Promise<WorkerOperationState> {
+    const current = await this.requireState(input.opId);
+    const now = input.updatedAt ?? this.clock.now();
+
+    const next: WorkerOperationState = {
+      ...current,
+      status: 'succeeded',
+      startedAt: current.startedAt ?? now,
+      updatedAt: now,
+      result: input.result,
+      ...(input.timing ? { timing: input.timing } : {}),
+    };
+
+    await this.persistState(next);
+    return next;
+  }
+
+  async markFailed(input: {
+    opId: string;
+    error: WorkerJobErrorShape | string;
+    updatedAt?: number;
+    timing?: WorkerJobTiming;
+  }): Promise<WorkerOperationState> {
+    const current = await this.requireState(input.opId);
+    const now = input.updatedAt ?? this.clock.now();
+
+    const shape = typeof input.error === 'string' ? { message: input.error } : input.error;
+
+    const next: WorkerOperationState = {
+      ...current,
+      status: 'failed',
+      startedAt: current.startedAt ?? now,
+      updatedAt: now,
+      error: shape,
+      ...(input.timing ? { timing: input.timing } : {}),
+    };
+
+    await this.persistState(next);
+    return next;
+  }
+
+  async explainReuseDecision(input: {
+    current: WorkerOperationState;
+    requestKind: WorkerOperationKind;
+  }): Promise<string> {
+    return mapReplacementReason({
+      current: input.current,
+      requestKind: input.requestKind,
+      now: this.clock.now(),
+      opStaleMs: this.opStaleMs,
+    });
+  }
+
+  private async requireState(opId: string): Promise<WorkerOperationState> {
+    const current = await this.stateStore.getOpState(opId);
+    if (!current) {
+      throw new Error(`Operation not found: ${opId}`);
+    }
+    if (isTerminalStatus(current.status)) {
+      return current;
+    }
+    return current;
+  }
+}

compute/core/src/control-plane/types.ts

@@ -0,0 +1,67 @@
+import type {
+  WorkerOperationEvent,
+  WorkerOperationKind,
+  WorkerOperationRequest,
+  WorkerOperationState,
+} from '../api-contracts';
+
+export type OperationRequest = WorkerOperationRequest;
+export type OperationState<Result = unknown> = WorkerOperationState<Result>;
+export type OperationEvent<Result = unknown> = WorkerOperationEvent<Result>;
+
+export interface QueuedOperation<TPayload = unknown> {
+  jobId: string;
+  opId: string;
+  opKey: string;
+  kind: WorkerOperationKind;
+  queuedAt: number;
+  payload: TPayload;
+}
+
+export interface OperationQueue<TPayload = unknown> {
+  enqueue(job: QueuedOperation<TPayload>): Promise<void>;
+  claimNext(kind: WorkerOperationKind): Promise<QueuedOperation<TPayload> | null>;
+  size(kind?: WorkerOperationKind): number;
+}
+
+export interface OperationIndexEntry {
+  opId: string;
+}
+
+export interface OperationStateStore<Result = unknown> {
+  getOpState(opId: string): Promise<OperationState<Result> | null>;
+  putOpState(state: OperationState<Result>): Promise<void>;
+  getOpIndex(opKey: string): Promise<OperationIndexEntry | null>;
+  compareAndSetOpIndex(input: {
+    opKey: string;
+    newOpId: string;
+    expectedOpId: string | null;
+  }): Promise<boolean>;
+}
+
+export interface OperationEventStream<Result = unknown> {
+  append(opId: string, snapshot: OperationState<Result>): Promise<OperationEvent<Result>>;
+  listSince(opId: string, sinceEventId: number, limit?: number): Promise<OperationEvent<Result>[]>;
+  subscribe(input: {
+    opId: string;
+    sinceEventId?: number;
+    onEvent: (event: OperationEvent<Result>) => void | Promise<void>;
+    onError?: (error: unknown) => void;
+  }): Promise<() => void>;
+}
+
+export interface OperationClock {
+  now(): number;
+}
+
+export interface OperationIdFactory {
+  opId(): string;
+  jobId(): string;
+}
+
+export interface OperationLifecycleConfig {
+  opStaleMs: number;
+  maxCasRetries?: number;
+}
+
+export type OperationTransitionStatus = 'queued' | 'running' | 'succeeded' | 'failed';

compute/core/src/index.ts

@@ -21,3 +21,4 @@ export { normalizeTextItemsForLayout } from './pdf/normalize-text';
 export { mapWordsToSentenceOffsets, type WhisperWord } from './whisper/alignment-map';
 export { buildGoertzelCoefficients, goertzelPower } from './whisper/spectral';
 export { buildWordsFromTimestampedTokens, extractTokenStartTimestamps } from './whisper/token-timestamps';
+export * from './control-plane';

tests/unit/compute-control-plane.spec.ts

@@ -0,0 +1,153 @@
+import { expect, test } from '@playwright/test';
+import type { WorkerOperationRequest, WorkerOperationState } from '../../compute/core/src/api-contracts';
+import {
+  InMemoryOperationEventStream,
+  InMemoryOperationQueue,
+  InMemoryOperationStateStore,
+  OperationOrchestrator,
+} from '../../compute/core/src/control-plane';
+
+function buildPdfLayoutRequest(opKey: string): WorkerOperationRequest {
+  return {
+    kind: 'pdf_layout',
+    opKey,
+    payload: {
+      documentId: `doc-${opKey}`,
+      documentObjectKey: `s3://bucket/${opKey}.pdf`,
+      namespace: null,
+    },
+  };
+}
+
+test.describe('compute control-plane', () => {
+  test('in-memory queue and state store support enqueue/claim/CAS', async () => {
+    const queue = new InMemoryOperationQueue();
+    const store = new InMemoryOperationStateStore();
+
+    await queue.enqueue({
+      jobId: 'job-1',
+      opId: 'op-1',
+      opKey: 'k-1',
+      kind: 'pdf_layout',
+      queuedAt: 1000,
+      payload: { documentId: 'd1', documentObjectKey: 'obj1', namespace: null },
+    });
+    await queue.enqueue({
+      jobId: 'job-2',
+      opId: 'op-2',
+      opKey: 'k-2',
+      kind: 'whisper_align',
+      queuedAt: 1100,
+      payload: { text: 'hello', audioObjectKey: 'obj2' },
+    });
+
+    expect(queue.size()).toBe(2);
+    expect(queue.size('pdf_layout')).toBe(1);
+    expect(queue.size('whisper_align')).toBe(1);
+
+    const claimedLayout = await queue.claimNext('pdf_layout');
+    expect(claimedLayout?.opId).toBe('op-1');
+    expect(queue.size('pdf_layout')).toBe(0);
+
+    const firstCas = await store.compareAndSetOpIndex({
+      opKey: 'k-1',
+      newOpId: 'op-1',
+      expectedOpId: null,
+    });
+    const secondCas = await store.compareAndSetOpIndex({
+      opKey: 'k-1',
+      newOpId: 'op-2',
+      expectedOpId: null,
+    });
+
+    expect(firstCas).toBeTruthy();
+    expect(secondCas).toBeFalsy();
+    expect(await store.getOpIndex('k-1')).toEqual({ opId: 'op-1' });
+  });
+
+  test('in-memory event stream replays from sinceEventId and streams live events', async () => {
+    const stream = new InMemoryOperationEventStream();
+
+    const queued: WorkerOperationState = {
+      opId: 'op-1',
+      opKey: 'k-1',
+      kind: 'pdf_layout',
+      jobId: 'job-1',
+      status: 'queued',
+      queuedAt: 1000,
+      updatedAt: 1000,
+    };
+    const running: WorkerOperationState = {
+      ...queued,
+      status: 'running',
+      startedAt: 1200,
+      updatedAt: 1200,
+    };
+    const succeeded: WorkerOperationState = {
+      ...running,
+      status: 'succeeded',
+      updatedAt: 1400,
+      result: { ok: true },
+    };
+
+    await stream.append('op-1', queued);
+    await stream.append('op-1', running);
+
+    const receivedEventIds: number[] = [];
+    const unsubscribe = await stream.subscribe({
+      opId: 'op-1',
+      sinceEventId: 1,
+      onEvent: (event) => {
+        receivedEventIds.push(event.eventId);
+      },
+    });
+
+    await stream.append('op-1', succeeded);
+    unsubscribe();
+
+    expect(receivedEventIds).toEqual([2, 3]);
+  });
+
+  test('orchestrator reuses fresh inflight operations and replaces stale ones', async () => {
+    let now = 1_000;
+    let nextId = 1;
+    const queue = new InMemoryOperationQueue();
+    const stateStore = new InMemoryOperationStateStore();
+    const eventStream = new InMemoryOperationEventStream();
+    const orchestrator = new OperationOrchestrator({
+      queue,
+      stateStore,
+      eventStream,
+      config: { opStaleMs: 2_000, maxCasRetries: 5 },
+      clock: { now: () => now },
+      idFactory: {
+        opId: () => `op-${nextId}`,
+        jobId: () => `job-${nextId++}`,
+      },
+    });
+
+    const request = buildPdfLayoutRequest('same-op-key');
+
+    const first = await orchestrator.enqueueOrReuse(request);
+    expect(first.opId).toBe('op-1');
+    expect(queue.size('pdf_layout')).toBe(1);
+
+    now = 2_000;
+    const reused = await orchestrator.enqueueOrReuse(request);
+    expect(reused.opId).toBe('op-1');
+    expect(queue.size('pdf_layout')).toBe(1);
+
+    await orchestrator.markRunning({ opId: first.opId, updatedAt: 2_100 });
+
+    now = 6_000;
+    const replaced = await orchestrator.enqueueOrReuse(request);
+    expect(replaced.opId).toBe('op-2');
+    expect(queue.size('pdf_layout')).toBe(2);
+    expect(await stateStore.getOpIndex('same-op-key')).toEqual({ opId: 'op-2' });
+
+    const op1Events = await eventStream.listSince('op-1', 0);
+    const op2Events = await eventStream.listSince('op-2', 0);
+    expect(op1Events.map((event) => event.eventId)).toEqual([1, 2]);
+    expect(op2Events.map((event) => event.eventId)).toEqual([1]);
+  });
+});