# Capability Spec Delta: queue-distribution-worker-thread (enterprise-queue-distribution) **Package**: `@suaveplan/queue-distribution-worker-thread` **Status**: PROPOSED > Every requirement below carries a `**Verified by:**` line at authoring time. (See `openspec/AGENTS.md` Section 13.) ## ADDED Requirements ### Requirement: WorkerFactory-driven communicator covering Web-Worker and `worker_threads` The package SHALL expose `WorkerThreadCommunicator` implementing `WorkerCommunicator` from `@suaveplan/queue-distribution-http` (re-exported `WorkerCommunicator` interface). The constructor SHALL accept an injectable `WorkerFactory` of shape `() => { postMessage(msg, transfer?); addEventListener(kind, handler); terminate(); }` so the same class works against `globalThis.Worker` (browser) and a thin `worker_threads` wrapper (Node/Bun) without importing platform APIs at module scope. The implementation SHALL maintain a `Map` of outstanding requests. Each outgoing message SHALL include an `id` (generated via injected `IdGenerator`), a `kind: "task" | "status" | "health" | "ping"`, and a `payload` validated by Zod. Reply correlation SHALL match `id`; unknown `id` replies SHALL be logged and dropped. Reply payloads SHALL be Zod-validated against per-kind schemas; failures SHALL reject the in-flight request with `QueueDistributionWorkerThreadError(WT.MESSAGE_MALFORMED)`. Each request SHALL be subject to a per-call timeout (default `defaultTimeoutMs`) honoured by the injected `Clock`; expiry SHALL reject with `WT.REQUEST_TIMEOUT` and remove the entry from the map. #### Scenario: round-trip correlation - **WHEN** `sendTask`/`checkStatus`/`checkHealth`/`ping` are invoked concurrently and the underlying worker replies in arbitrary order - **THEN** each promise resolves with its own reply payload (matched by `id`) #### Scenario: malformed reply throws WT.MESSAGE_MALFORMED - **WHEN** the underlying worker posts back `{ id: , kind: "task", payload: { something: "bad" } }` not conforming to the task-reply Zod schema - **THEN** the in-flight `sendTask` promise rejects with `QueueDistributionWorkerThreadError(WT.MESSAGE_MALFORMED)` #### Scenario: request timeout - **WHEN** a `FakeClock` is injected and the worker never replies within `defaultTimeoutMs` - **THEN** after the clock advances past the deadline, the call rejects with `QueueDistributionWorkerThreadError(WT.REQUEST_TIMEOUT)` AND the outstanding-request map no longer contains the entry **Verified by:** `src/communicator/worker-thread/worker-thread-communicator.test.ts::"correlates concurrent replies by id"` AND `src/communicator/worker-thread/worker-thread-communicator.test.ts::"rejects with WT.MESSAGE_MALFORMED when reply fails Zod validation"` AND `src/communicator/worker-thread/worker-thread-communicator.test.ts::"rejects with WT.REQUEST_TIMEOUT and clears the outstanding-request map"` --- ### Requirement: WorkerThreadWorker — `AbstractWorker` subclass dispatching to a backing worker The package SHALL ship `WorkerThreadWorker` extending `AbstractWorker` from `@suaveplan/queue-distribution`. Constructor SHALL accept an injectable `WorkerFactory` and instantiate a fresh `WorkerThreadCommunicator` per worker, AND a single backing worker thread. `doStart()` SHALL register error / exit listeners on the backing worker — `worker.exit` events SHALL `markUnhealthy("Worker exited")` and flip `status` to `OFFLINE`; uncaught `error` events SHALL emit the `WT.WORKER_CRASHED` OTel counter. `doStop(force)` SHALL `terminate()` the backing worker, fail every outstanding request with `WT.WORKER_TERMINATED`, and clear the outstanding-request map. `doExecuteTask` SHALL delegate to `communicator.sendTask(, task, input, signal)`. `getCapabilities()` SHALL return `{ isRemote: false, isThreaded: true }`. `getType()` SHALL return `"worker-thread"`. #### Scenario: backing worker crash promotes to UNHEALTHY - **WHEN** the backing worker emits an `error` event - **THEN** the OTel `queue_dist_wt.worker.crashed_total` counter increments and the worker is marked UNHEALTHY #### Scenario: terminate fails outstanding requests with WT.WORKER_TERMINATED - **WHEN** two `sendTask` calls are in flight and `worker.stop(true)` is invoked - **THEN** both pending promises reject with `QueueDistributionWorkerThreadError(WT.WORKER_TERMINATED)` AND `worker.status === OFFLINE` **Verified by:** `src/worker/worker-thread/worker-thread-worker.test.ts::"backing worker crash flips status to UNHEALTHY and increments WT.WORKER_CRASHED counter"` AND `src/worker/worker-thread/worker-thread-worker.test.ts::"terminate fails outstanding requests with WT.WORKER_TERMINATED"` --- ### Requirement: Stable error registry for the worker-thread plugin The package SHALL register every error code through `defineCodes("WT", [...])` from `@suaveplan/error` and expose `QueueDistributionWorkerThreadError extends SuaveplanError`. The exhaustive set is: `WORKER_CRASHED`, `MESSAGE_MALFORMED`, `REQUEST_TIMEOUT`, `WORKER_TERMINATED`. Every code SHALL be thrown by ≥1 production call path. #### Scenario: every registered code is thrown - **WHEN** the test suite iterates over every code in the registry - **THEN** a corresponding production throw site exists **Verified by:** `src/errors/queue-distribution-worker-thread-error.test.ts::"every registered WT.* code is thrown by a production call path"` --- ### Requirement: OpenTelemetry instrumentation The package SHALL emit, via `@suaveplan/opentelemetry` under meter name `"@suaveplan/queue-distribution-worker-thread"`: - histogram: `queue_dist_wt.request.duration_ms` (attributes `kind`, `outcome`) - counter: `queue_dist_wt.request_total` (attributes `kind`, `outcome`) - counter: `queue_dist_wt.worker.crashed_total` - counter: `queue_dist_wt.worker.terminated_total` - span: `queue_dist_wt.sendTask` #### Scenario: histogram emission on sendTask round-trip - **WHEN** `sendTask` resolves successfully under an in-memory `MeterProvider` - **THEN** one observation is recorded on `queue_dist_wt.request.duration_ms` with `kind === "task"` AND `outcome === "success"` **Verified by:** `src/communicator/worker-thread/worker-thread-communicator.spec-conformance.test.ts::"emits queue_dist_wt.request.duration_ms histogram with kind and outcome attributes"`