feat(worker): Add worker-initiated requests support

[taras]

Motivation

The current @effectionx/worker package only supports host-to-worker communication: the host can send messages to workers and receive responses. However, some use cases require the reverse direction - workers initiating requests to the host.

A concrete example is the sweatpants project's web worker transport for MCP tool sessions. In this architecture:

• Tools run in web workers (the "Principal" that initiates requests)
• The host responds to tool requests (the "Operative")

This is the opposite of the current pattern where the host is always the initiator.

What's New

Progress Streaming Support

This PR now includes bidirectional progress streaming - the host can send progress updates back to the worker during request processing:

Worker side (receive progress):

await workerMain<never, never, string, void, string, string>(
  function* ({ send }) {
    // Simple request/response (unchanged)
    const response = yield* send("hello");
    
    // With progress streaming
    const subscription = yield* send.stream<number>("process");
    let next = yield* subscription.next();
    while (!next.done) {
      console.log("Progress:", next.value);  // 25, 50, 75
      next = yield* subscription.next();
    }
    const response = next.value;  // "done"
  }
);

Host side (send progress):

yield* worker.forEach<string, string, number>(function* (request, ctx) {
  yield* ctx.progress(25);
  yield* ctx.progress(50);
  yield* ctx.progress(75);
  return "done";
});

Backpressure Semantics

The progress() method implements true backpressure:

• ctx.progress() blocks until the worker calls subscription.next()
• The host cannot send progress faster than the worker can receive it
• If the worker does async work between next() calls, the host remains blocked

This ensures the worker is never overwhelmed with progress updates. The ACK is sent inside subscription.next(), so the host waits for the worker to be ready for the next value.

Key Features

• True backpressure: Host blocks until worker is ready for next progress
• Backwards compatible: Existing code without progress continues to work
• Type-safe: Progress type is a separate type parameter

Approach

This PR adds symmetric bidirectional communication by introducing:

New Worker-Side API: send() function

Workers receive a send function in their workerMain options:

await workerMain<never, never, string, void, string, string>(
  function* ({ send }) {
    const response = yield* send("hello");  // Worker initiates request
    return `received: ${response}`;
  }
);

New Host-Side API: worker.forEach() method

Hosts can handle worker requests using forEach:

const worker = yield* useWorker<never, never, string, void>(workerUrl, opts);

const result = yield* worker.forEach<string, string>(function* (request) {
  return `echo: ${request}`;  // Host handles worker's request
});

Channel Primitives

Introduces reusable channel primitives for request-response over MessageChannel:

• useChannelResponse<TResponse, TProgress>(options?) - Requester side

  • Creates MessageChannel, returns port to transfer and operation to await response
  • yield* response - waits for response, ignores progress
  • yield* response.progress - returns Subscription that yields progress then response
  • Races message vs close events (detects responder crash)
  • Optional timeout via @effectionx/timebox
  • Returns SerializedResult<T> for type-safe error handling

• useChannelRequest<TResponse, TProgress>(port) - Responder side

  • Wraps received port with resolve(value), reject(error), and progress(data) operations
  • progress() sends update and waits for ACK (true backpressure)
  • Handles serialization internally (wraps in SerializedResult)
  • Races ACK vs close events (detects requester cancellation)

Wire Protocol

Messages sent over channel (Host → Worker):

| { type: "progress"; data: TProgress }
| { type: "response"; result: SerializedResult<TResponse> }

Acknowledgements (Worker → Host):

| { type: "ack" }           // After response
| { type: "progress_ack" }  // After each progress (sent inside next())

Key Design Decisions

1. MessageChannel per request - Each worker request creates a MessageChannel for correlation (no IDs needed)
2. SerializedResult type - Type-safe { ok: true, value } or { ok: false, error: SerializedError } for cross-boundary communication
3. Error serialization - Errors are serialized as { name, message, stack } and wrapped with cause on the receiving side
4. ACK mechanism - Responses require acknowledgment before cleanup, guaranteeing delivery
5. Close detection - Both sides detect if the other crashes/cancels via MessagePort close events
6. Timeout support - Optional timeout parameter on useChannelResponse using @effectionx/timebox
7. Request queueing - Requests sent before forEach() is called are queued (unbounded)
8. Concurrent forEach guard - Only one forEach() call allowed at a time
9. Message loop aligned with resource lifecycle - Uses a pre-attached subscription, resolves the initial "open" via withResolvers, and ensures outcome settles during teardown if the close message arrives late
10. True backpressure - Progress ACK is sent inside next(), so host waits for worker readiness

Type Parameter Naming

• TSend/TRecv - Host-initiated communication (host sends, worker receives/responds)
• WRequest/WResponse - Worker-initiated communication (worker sends, host receives/responds)
• WProgress - Progress updates from host to worker

Tests

All 54 worker tests pass (4 skipped):

Channel primitive tests (32 tests):

• Basic channel creation and round-trip
• ACK verification
• Close detection (responder crashes/exits)
• Timeout behavior (slow responder, fast responder)
• Cancellation handling (requester cancels, responder exits)
• Progress streaming tests (15 tests):
  • Receives multiple progress then response
  • yield* response ignores progress
  • Error response after progress
  • Port close during progress
  • Backpressure verification (host blocks until worker ready)
  • Order preservation
  • Complex progress data
  • Zero progress updates
  • Requester cancellation during progress
  • Timeout applies to entire exchange

Worker-initiated request tests (15 tests):

• Single request handling
• Multiple sequential requests
• Concurrent requests from worker
• Error propagation with cause chain
• Bidirectional communication
• Request queueing before forEach
• Concurrent forEach guard
• Nested send() inside messages.forEach handler

Policy Compliance

Tests comply with the No-Sleep Test Synchronization Policy:

• Uses withResolvers() for callback synchronization
• Uses sleep(0) only for yielding control
• sleep(ms) only inside spawned tasks to trigger conditions

Related Issues

• Add TimeoutError and context-based timeout configuration to @effectionx/timebox #139 - Add TimeoutError and context-based timeout configuration to @effectionx/timebox

[pkg-pr-new]

Open in StackBlitz

npm i https://pkg.pr.new/thefrontside/effectionx/@effectionx/worker@138

commit: 899bedd