Post-commit reactors and the delivery gap

The transaction gives T3 Code durable intent. It does not give every post-commit consumer durable delivery. The engine commits events, SQL projections, and an accepted receipt, returns from that transaction, replaces its in-memory read model, and only then publishes each event to an in-process PubSub. Reactors turn those hot events into provider, checkpoint, terminal, deletion, and awareness effects.

That separation is useful: an external harness call never holds the SQLite transaction open. It also creates the defining failure window of this chapter: the database can say “accepted” while every side-effect subscriber sees nothing.

The transaction ends before delivery begins

        const committedCommand = yield* sql
          .withTransaction(
            Effect.gen(function* () {
              const committedEvents: OrchestrationEvent[] = [];
              let nextCommandReadModel = commandReadModel;
 
              for (const nextEvent of eventBases) {
                const savedEvent = yield* eventStore.append(nextEvent);
                nextCommandReadModel = yield* projectEvent(nextCommandReadModel, savedEvent);
                yield* projectionPipeline.projectEvent(savedEvent);
                committedEvents.push(savedEvent);
              }
 
              const lastSavedEvent = committedEvents.at(-1) ?? null;
              if (lastSavedEvent === null) {
                return yield* new OrchestrationCommandInvariantError({
                  commandType: envelope.command.type,
                  detail: "Command produced no events.",
                });
              }
 
              yield* commandReceiptRepository.upsert({
                commandId: envelope.command.commandId,
                aggregateKind: lastSavedEvent.aggregateKind,
                aggregateId: lastSavedEvent.aggregateId,
                acceptedAt: lastSavedEvent.occurredAt,
                resultSequence: lastSavedEvent.sequence,
                status: "accepted",
                error: null,
              });
 
              return {
                committedEvents,
                lastSequence: lastSavedEvent.sequence,
                nextCommandReadModel,
              } as const;
            }),
          )
          .pipe(
            Effect.catchTag("SqlError", (sqlError) =>
              Effect.fail(
                toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError),
              ),
            ),
          );
 
        commandReadModel = committedCommand.nextCommandReadModel;
        for (const [index, event] of committedCommand.committedEvents.entries()) {
          yield* PubSub.publish(eventPubSub, event);
          if (index === 0) {
            yield* Metric.update(
              Metric.withAttributes(
                orchestrationCommandAckDuration,
                metricAttributes({
                  ...baseMetricAttributes,
                  ackEventType: event.type,
                }),
              ),
              Duration.millis(Math.max(0, (yield* Clock.currentTimeMillis) - envelope.startedAtMs)),
            );
          }
        }
        return { sequence: committedCommand.lastSequence };

flowchart LR
  C[command queue] --> D[decide]
  D --> TX[(event + projections + receipt)]
  TX -->|COMMIT| M[in-memory model]
  M -. hot publish .-> B{{domain PubSub}}
  B -.-> R1[provider reactor worker]
  B -.-> R2[checkpoint worker]
  B -.-> R3[deletion + awareness workers]
  R1 -. fork .-> P[ProviderService + adapter]
  P -. hot runtime event .-> RB{{runtime PubSub}}
  RB -.-> I[runtime ingestion worker]
  I -->|internal command| C
  TX -. "crash: intent durable, delivery absent" .- B

The engine’s returned sequence is therefore an acknowledgement of command acceptance, not provider completion. A client retry using the same command id finds the accepted receipt and returns its stored sequence. That lookup does not republish the event. Receipt idempotency protects the transaction from duplicate command execution; it does not repair the post-commit side-effect bridge.

The cursor exposes two different uncertainties. Before SQL commit, rollback is clean: no accepted receipt survives. After commit but before reactor observation, intent survives while work disappears. Once an external adapter call begins, the problem changes again: the server may no longer know whether repeating the call is safer than leaving it alone.

One queue serializes one handler—not the whole system

OrchestrationReactor.start starts several consumers in one scope: provider runtime ingestion, provider command handling, checkpoint capture/revert, thread deletion, and agent awareness. They do not share one global queue. Each worker serializes its own handler invocations while the consumers themselves run independently.

  const processTurnStartRequested = Effect.fn("processTurnStartRequested")(function* (
    event: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,
  ) {
    const key = turnStartKeyForEvent(event);
    if (yield* hasHandledTurnStartRecently(key)) {
      return;
    }
 
    const thread = yield* resolveThread(event.payload.threadId);
    if (!thread) {
      return;
    }
 
    const message = thread.messages.find((entry) => entry.id === event.payload.messageId);
    if (!message || message.role !== "user") {
      yield* appendProviderFailureActivity({
        threadId: event.payload.threadId,
        kind: "provider.turn.start.failed",
        summary: "Provider turn start failed",
        detail: `User message '${event.payload.messageId}' was not found for turn start request.`,
        turnId: null,
        createdAt: event.payload.createdAt,
      });
      return;
    }
 
    const isFirstUserMessageTurn =
      thread.messages.filter((entry) => entry.role === "user").length === 1;
    if (isFirstUserMessageTurn) {
      const project = yield* resolveProject(thread.projectId);
      const generationCwd =
        resolveThreadWorkspaceCwd({
          thread,
          projects: project ? [project] : [],
        }) ?? process.cwd();
      const generationInput = {
        messageText: message.text,
        ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
        ...(event.payload.titleSeed !== undefined ? { titleSeed: event.payload.titleSeed } : {}),
      };
 
      yield* maybeGenerateAndRenameWorktreeBranchForFirstTurn({
        threadId: event.payload.threadId,
        branch: thread.branch,
        worktreePath: thread.worktreePath,
        ...generationInput,
      }).pipe(Effect.forkScoped);
 
      if (canReplaceThreadTitle(thread.title, event.payload.titleSeed)) {
        yield* maybeGenerateThreadTitleForFirstTurn({
          threadId: event.payload.threadId,
          cwd: generationCwd,
          ...generationInput,
        }).pipe(Effect.forkScoped);
      }
    }
 
    const handleTurnStartFailure = (cause: Cause.Cause<unknown>) => {
      if (Cause.hasInterruptsOnly(cause)) {
        return Effect.void;
      }
      const detail = formatFailureDetail(cause);
      return setThreadSessionErrorOnTurnStartFailure({
        threadId: event.payload.threadId,
        detail,
        createdAt: event.payload.createdAt,
      }).pipe(
        Effect.flatMap(() =>
          appendProviderFailureActivity({
            threadId: event.payload.threadId,
            kind: "provider.turn.start.failed",
            summary: "Provider turn start failed",
            detail,
            turnId: null,
            createdAt: event.payload.createdAt,
          }),
        ),
        Effect.asVoid,
      );
    };
 
    const recoverTurnStartFailure = (cause: Cause.Cause<unknown>) =>
      handleTurnStartFailure(cause).pipe(
        Effect.catchCause((recoveryCause) =>
          Effect.logWarning("provider command reactor failed to recover turn start failure", {
            eventType: event.type,
            threadId: event.payload.threadId,
            cause: Cause.pretty(recoveryCause),
            originalCause: Cause.pretty(cause),
          }),
        ),
      );
 
    const sendTurnRequest = yield* buildSendTurnRequestForThread({
      threadId: event.payload.threadId,
      messageText: message.text,
      ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
      ...(event.payload.modelSelection !== undefined
        ? { modelSelection: event.payload.modelSelection }
        : {}),
      interactionMode: event.payload.interactionMode,
      createdAt: event.payload.createdAt,
    }).pipe(
      Effect.map(Option.some),
      Effect.catchCause((cause) => handleTurnStartFailure(cause).pipe(Effect.as(Option.none()))),
    );
 
    if (Option.isNone(sendTurnRequest)) {
      return;
    }
 
    yield* providerService
      .sendTurn(sendTurnRequest.value)
      .pipe(Effect.catchCause(recoverTurnStartFailure), Effect.forkScoped);
  });

Calling this “serialized provider execution” would be wrong. The provider reactor’s queue orders the start of its handler bodies. At ProviderCommandReactor.ts:1171–1173, sendTurn is forked, so two provider sends can overlap. First-turn title and branch work are also forked. Other reactors may concurrently process the same domain or provider event through their own subscriptions.

Provider intent uses volatile duplicate suppression

For a turn-start request, the provider reactor derives a turn key and checks an in-memory bounded cache with a 30-minute TTL. The key is recorded before the external send. Within that process, a duplicate can therefore be suppressed even if later work fails. Restart clears the cache—but it also clears the hot event, so the durable original does not automatically receive a second attempt.

Failure handling records selected durable error/session activities by dispatching internal commands. Those compensating commands are themselves best-effort from the reactor’s perspective: the handler catches, logs, and swallows non-interruption causes so that one bad event does not terminate its worker.

Within one process, this is an effectively at-most-once observation path per derived turn key, with explicit compensations. It is not a general exactly-once executor: the cache expires and disappears at restart, while the source event is not replayed. The underlying provider may have accepted a call before ProviderService persists its binding or running state, leaving an ambiguous external/local boundary after a crash.

Runtime results return through another hot bridge

Provider adapters emit canonical runtime events. ProviderService writes its best-effort diagnostic log and publishes the event to a second PubSub. Runtime ingestion subscribes to that provider stream alongside selected domain events; its current domain handler is a no-op, while provider observations translate into internal commands and dispatch through the same engine. Only the internal command’s successful transaction makes the result part of durable domain truth.

Runtime ingestion deliberately buffers high-rate content. Its flush path removes a buffer entry before dispatching the aggregate internal command. If dispatch is interrupted or fails, the removed batch is not restored. Handler errors are logged and swallowed. This limits memory and prevents a poisoned event from killing the loop, but it is not a durable queue.

The worker also subscribes to thread.turn-start-requested, but its selected-domain handler currently returns Effect.void. Non-interruption handler failures are logged and swallowed so the worker can continue; this isolates a bad input without making that input durable or retryable.

Tests cover deduplication between selected semantic completion signals, such as an item-completed event followed by turn-completed. They do not inject the exact same provider event twice across a process crash. The distinction matters: semantic coalescing is narrower than an idempotent ingestion log.

Parked roots narrow startup races, not delivery semantics

Startup creates a dedicated reactor scope and forks the reactor composite behind a shared activation deferred. Each activation-aware root first reports that it is parked. Provider-session reconciliation runs before activation; HTTP and auxiliary roots prepare; startup sends welcome, opens activation, and only then admits queued commands.

Startup provider reconciliation compares projected active sessions with live adapter inventory. An orphaned projection is marked stopped/error and the internal dispatch is retried once. That reconciler does not replay the original turn or eagerly resume every persisted provider binding; the pinned production-code audit found no separate arbitrary-reactor replay pass. Later routed operations can adopt or resume a binding lazily when enough cursor data exists.

Scope close interrupts; production does not globally drain

DrainableWorker owns an unbounded transactional queue, one scoped worker, and an outstanding counter. enqueue increments the counter, handler completion or failure decrements it, and drain waits for zero. Its source describes that drain as a deterministic synchronization utility for tests.

On scope close, the worker queue is shut down and the worker fiber is interrupted. The inspected orchestration composite exposes no step that first stops ingress and drains every worker and child fiber. ProviderService has its own substantial best-effort finalizer—persist active bindings, stop adapters, revoke credentials, mark bindings stopped, flush analytics—but that is not an atomic completion barrier for every reactor effect.

The other reactors have different partial recovery stories

Checkpoint capture creates a hidden Git ref, refreshes state, computes a diff, dispatches checkpoint metadata, inspects test receipts, and then records activity. Revert runs in the opposite world: it mutates the workspace, refreshes, rolls the provider back, deletes newer refs, and only then records durable completion. Chapter 13 examines the cross-store crash windows and the staging-state overclaim.

Thread deletion stops the provider session, closes the terminal, and deletes its history. Failures are logged. The pinned production-code audit found no durable cleanup job or startup thread scan that reconstructs a missed hot deletion event.

Agent awareness is intentionally more reconstructive: it keeps volatile dedupe and confirmation timers, then publishes a delayed snapshot of currently active threads after activation. That can restore current liveness presentation. Because both the published-state map and confirmation deadlines are process-local, this path does not replay a missed historical tombstone or preserve an in-flight timer across restart.

What the tests establish—and what they do not