Skip to main content
← Back to list
01Issue
FeatureShippedSwamp CLI
Assigneesstack72

Relationships

#949 Worker fleets phase 1 PR2: step_queued events, pending-dispatch model, queue introspection

Opened by stack72 · 7/3/2026· Shipped 7/4/2026

Summary

PR2 of worker-fleets phase 1 (#947). PR1 landed the core queueing behavior (steps queue instead of failing when no workers match). This PR adds visibility into the queue: operators see why a step is waiting, and the queued demand is introspectable as swamp data for autoscaling.

Three workstreams:

  1. step_queued event pipeline — thread a new event through 4 layers so workflow renderers show "waiting for a worker matching …"
  2. Pending-dispatch built-in model — queued demand as versioned swamp data, queryable with data.query
  3. Queue introspection CLI — swamp worker queue command and worker.queue.list serve protocol frame

Depends on PR1 (#1746) and its follow-up (#1747), both shipped.

Design reference

Full design: worker-fleets.md at the repo root (sections "Elastic queueing" and "Pending dispatches are swamp data"). This issue is self-contained for implementation.

Changes

1. Emit step_queued RpcStreamEvent from DispatchService

File: src/serve/dispatch_service.ts

When #acquireWorker enters the wait state (the queue branch at line ~382), emit a new event via request.onEvent before calling #waitForPoolChange. The event should use the existing RpcStreamEvent type with a new kind:

request.onEvent?.({ kind: "queued", requirement: describePlacement(placement) });

This must fire on every entry to the wait state, including re-entries after a worker_busy desync (the continue at line ~223). The describePlacement helper is already imported (PR1).

request.onEvent is currently wired at line ~296 inside #dispatchOnce and forwarded to the gateway's dispatch call. The step_queued event needs to fire from executeRemote (line ~186), which has access to request.onEvent directly. Pass request.onEvent into #acquireWorker or emit from executeRemote's loop before the #acquireWorker call.

Protocol note: RpcStreamEvent is defined in src/domain/remote/protocol.ts. Adding a new kind value does NOT require bumping REMOTE_PROTOCOL_VERSION — this event flows orchestrator→client over the serve protocol, not orchestrator→worker.

2. Bridge queued event to MethodExecutionEvent

File: src/domain/models/method_execution_service.ts

The #executeRemotely method (lines 320-385) bridges dispatch events to the method execution layer. Currently, the bridge at lines 371-383 only forwards kind === "method_event" frames. Add handling for the new queued kind:

if (event.kind === "queued" && "requirement" in event) {
  context.onEvent!({ type: "step_queued", requirement: event.requirement });
}

File: src/domain/models/method_events.ts

Add a new variant to the MethodExecutionEvent union (currently 4 variants at lines 27-48):

| { type: "step_queued"; requirement: string }

Uses the type discriminant, matching the existing convention in this file.

3. Wrap as WorkflowExecutionEvent

File: src/domain/workflows/execution_events.ts

Add a new kind to the WorkflowExecutionEvent union (currently ~20 kinds at lines 44-164):

| {
    kind: "step_queued";
    jobId: string;
    stepId: string;
    requirement: string;
  }

Uses the kind discriminant, matching the existing convention in this file.

File: src/domain/workflows/execution_service.ts

In the onEvent callback (lines 989-1010) that wraps MethodExecutionEvent into WorkflowExecutionEvent, add a case for type === "step_queued":

if (event.type === "step_queued") {
  yield { kind: "step_queued", jobId, stepId, requirement: event.requirement };
}

4. Map to WorkflowRunEvent (libswamp layer)

File: src/libswamp/workflows/run.ts

Add step_queued to the WorkflowRunEvent type (lines 71-189):

| {
    kind: "step_queued";
    jobId: string;
    stepId: string;
    requirement: string;
  }

Add the mapping case in mapWorkflowExecutionEvent (lines 390-459, exhaustive switch).

5. Handle in renderers

File: src/presentation/renderers/workflow_run_tree/state.ts

Two changes:

  1. Add "queued" to StepState.status (lines 42-48, currently: pending | running | waiting_approval | completed | failed | skipped).

  2. Add a case in treeReducerSingle (exhaustive switch at lines 280-642) for kind: "step_queued":

    • Set the step's status to "queued" and store the requirement string
    • The exhaustive check at line 637-639 (const _exhaustive: never = event) will fail to compile until this is added — this is the safety net

File: src/presentation/renderers/workflow_run.ts

Add a handler in LogWorkflowRunRenderer.handlers() (line 66+) for step_queued:

  • Display something like: ⏳ ${stepId}: waiting for a worker matching ${requirement}
  • Match the existing output style (indentation, colors)

File: src/serve/serializer.ts / src/serve/serializer_test.ts

The serializer uses a generic JSON-safe clone (serializeEvent), so no structural change is needed. Add a round-trip test in serializer_test.ts that serializes and deserializes a step_queued event.

6. Create pending-dispatch built-in model

New file: src/domain/models/worker/pending_dispatch_model.ts

Mirror step_lease_model.ts (lines 42-231) — the closest existing shape:

  • Type: PENDING_DISPATCH_MODEL_TYPE = ModelType.create("swamp/pending-dispatch")
  • Instance name: PENDING_DISPATCH_INSTANCE_NAME = "pending" (single instance, like leases use "leases")
  • Named items: pending-${queueId} (one record per queue episode)

States: waiting → dispatched | timed_out | cancelled (+ orphaned, used only by boot reconciliation in PR3)

Schema fields:

  • queueId (string, UUID)
  • state (enum of states above)
  • target (string, optional) — placement target
  • labels (Record<string, string>, optional) — placement labels
  • platform (string, optional) — placement platform
  • workflowName (string, optional)
  • jobName (string, optional)
  • stepName (string, optional)
  • modelType (string)
  • methodName (string)
  • queuedAt (string, ISO datetime)
  • dispatchId (string, optional — set on mark_dispatched)
  • endedAt (string, optional)

Methods: enqueue (kind: "create"), mark_dispatched (kind: "update"), timeout (kind: "update"), cancel (kind: "update"), orphan (kind: "update")

Resource: single resource "dispatch" with lifetime: "infinite", garbageCollection: 10 (matching step-lease)

Registration: import in src/domain/models/models.ts barrel (after line 40, alongside the other worker models). Add "swamp/pending-dispatch" to the built-in internal types list (lines 57-66).

7. Wire pending-dispatch transitions into DispatchService

File: src/serve/dispatch_service.ts

Add a #pendingDispatchTransition method that mirrors #leaseTransition (line 474) — same #transitionTail chaining pattern (sole-writer rule). Or generalize #leaseTransition to accept a model type parameter.

Wire the transitions in executeRemote (line ~186):

  • On wait-state entry: call enqueue with a fresh queueId (UUID), the placement, step/workflow names, model type, method name, and queuedAt: new Date().toISOString()
  • On dispatch: call mark_dispatched with the dispatchId (from #dispatchOnce) and endedAt
  • On timeout: call timeout with endedAt
  • On abort (signal fires while queued): call cancel with endedAt

Record identity: Each wait episode gets a fresh record. A re-queued step (after worker_busy desync) gets a new queueId and a new enqueue call. The previous episode's record was already mark_dispatched (the dispatch happened, it just bounced).

Import PENDING_DISPATCH_MODEL_TYPE and PENDING_DISPATCH_INSTANCE_NAME from the new model file.

8. Queue introspection: serve protocol + libswamp operation

File: src/serve/protocol.ts

Add worker.queue.list request/response types, mirroring worker.list (lines 501, 342, 764, 949):

export type WorkerQueueListPayload = Record<string, never>;
export type WorkerQueueListResponse = { data: Record<string, unknown> };

Add to ServerRequest and ServerMessage unions.

File: src/serve/connection.ts

Add a "worker.queue.list" case in the dispatch switch (line 948), calling a new handleWorkerQueueList handler (admin read). The handler queries pending-dispatch data from the datastore — the same query pattern as handleWorkerList.

New file: src/libswamp/worker/queue_list.ts

Mirror src/libswamp/worker/list.ts (lines 201-248):

  • workerQueueList async generator yielding WorkerQueueListEvent
  • Reads pending-dispatch data from the datastore where state == "waiting"
  • Returns WorkerQueueListItem with: queueId, requirement (target/labels/platform), workflowName, jobName, stepName, modelType, methodName, queuedAt, age (computed from queuedAt)
  • createWorkerQueueListDeps wiring function

File: src/libswamp/mod.ts

Export workerQueueList, WorkerQueueListData, WorkerQueueListDeps, WorkerQueueListEvent, WorkerQueueListItem, createWorkerQueueListDeps from "./worker/queue_list.ts". CLI must NOT import internal paths.

9. swamp worker queue CLI command

New file: src/cli/commands/worker_queue.ts

Add to the existing worker command group (src/cli/commands/worker.ts, line ~38-45). Dual-mode:

  • Local repo: read via libswamp workerQueueList
  • Remote: via requestServerResponse<WorkerQueueListResponse> with --server

File: src/presentation/output/worker_output.ts

Add renderWorkerQueue(data, mode):

  • Log mode: table with columns: REQUIREMENT, STEP, MODEL, QUEUED AT, AGE
  • JSON mode: structured array

Follow the existing renderWorkerList pattern in the same file.

Invariants to preserve

  1. Sole-writer rule. Pending-dispatch transitions must serialize through #transitionTail — same chain as lease transitions.
  2. Event discriminant conventions. MethodExecutionEvent uses type, WorkflowExecutionEvent uses kind. Don't mix them.
  3. Exhaustive switches. treeReducerSingle in state.ts (line 637) and mapWorkflowExecutionEvent in run.ts (line ~459) both have exhaustive never checks. Adding a new event kind without handling it in these switches is a compile error, not a runtime bug — but the implementing agent must handle both.
  4. libswamp import boundary. CLI commands import from src/libswamp/mod.ts only — never from internal paths like src/libswamp/worker/queue_list.ts.
  5. Domain layer boundary. The domain must NOT import CLI-layer modules. Enforced by integration/ddd_layer_rules_test.ts.
  6. Pending-dispatch is the datastore source of truth. swamp worker queue reads from the datastore, not from the dispatch service's in-memory #poolWaiters. The in-memory state is ephemeral; the model data survives for introspection and (eventually) autoscaling.

Tests required

  • MethodExecutionEvent: step_queued variant exists and is distinct from existing variants
  • Serializer round-trip: step_queued event survives serialize→deserialize in serializer_test.ts
  • Pending-dispatch model: state machine transitions (enqueue→mark_dispatched, enqueue→timeout, enqueue→cancel); garbageCollection declared; schema validates all fields
  • DispatchService: step_queued event fires on wait-state entry; pending-dispatch enqueue called on wait, mark_dispatched on dispatch, timeout on deadline, cancel on abort; re-queue after worker_busy creates a fresh record
  • Tree renderer: step_queued event sets step status to "queued" in state
  • Log renderer: step_queued event produces output containing the requirement string
  • Queue introspection: workerQueueList returns waiting entries; swamp worker queue CLI produces output in both log and json modes
  • Integration: extend the existing integration test — submit a step that queues, assert the step_queued event fires, then connect a worker and assert it completes. Verify swamp worker queue (or the equivalent datastore query) shows the entry while it's queued.

Test conventions:

  • Tests live next to source: foo.ts → foo_test.ts
  • Named as Deno.test("functionName: describes behavior", ...)
  • CLI command tests need await initializeLogging({}) and import "../../domain/models/models.ts"
  • Cross-component changes need integration tests under integration/
  • Tests must pass on Linux/macOS/Windows

Acceptance criteria

  1. A placed workflow run against an empty pool shows "waiting for a worker matching …" in the workflow output (both log and tree renderers)
  2. swamp worker queue shows the waiting step with its requirement, step name, and age
  3. swamp data query 'modelType == "swamp/pending-dispatch" && attributes.state == "waiting"' returns the queued entry
  4. When the step dispatches, the pending-dispatch record transitions to dispatched
  5. When the step times out, the record transitions to timed_out
  6. When the step is cancelled (abort signal), the record transitions to cancelled
  7. All existing tests continue to pass
  8. deno check, deno lint, deno fmt, deno run test, deno run compile all pass

Repo conventions

  • New .ts files need the AGPLv3 header — run deno run license-headers
  • TypeScript strict, no any, named exports only
  • CLI commands and renderers import libswamp only via src/libswamp/mod.ts
  • Every new command supports both log and json output modes
  • Use the ddd skill when shaping new domain objects
  • Use the terminal-output skill when writing render code
  • Use the github-pr skill for commits/PRs
  • Keep blast radius small — no adjacent refactors
  • Update worker-fleets.md when implementation diverges from the plan
02Bog Flow
✓OPEN✓TRIAGED✓IN PROGRESS✓SHIPPED+ 1 MOREASSIGNED+ 5 MOREREVIEW+ 4 MOREPR_MERGED+ 1 MORENOTIFICATION_SKIPPED

Shipped

7/4/2026, 1:32:51 AM

Click a lifecycle step above to view its details.

03Sludge Pulse
stack72 assigned stack727/3/2026, 11:40:28 PM

Sign in to post a ripple.