Start, stream, steer, interrupt, settle
A T3 turn begins as a durably accepted command and later crosses hot reactor and provider-runtime boundaries; session state, output, checkpoints, and client liveness each have narrower guarantees.
What this chapter resolves
- Follow a thread turn from a client command through decision, commitment, reactors, and provider events.
- Separate durable thread and turn projections from provider-process liveness and hot delivery.
- Explain buffering, user-interaction pauses, interruption, failures, completion, checkpoints, and settlement without treating any one signal as universal.
- Use correlation and turn guards to reason about delayed, missing, or superseded provider events.
A T3 thread is the durable conversation and workspace record. A turn is one requested user-to-agent cycle within it. They overlap, but they do not share one clock or one source of truth.
The command path can durably record that a turn was requested before a provider process has accepted anything. Later, canonical provider events can make session and message projections durable. Between those crossings, a live process, a hot stream, and a client subscription can all disappear independently.
1. Request: a command records intent, not generated work
The thread.turn.start command schema carries a user message, optional model/title
choices, runtime and interaction mode fields, and optional bootstrap metadata. On
the current web path for an already-created server thread, selected model and modes
are first persisted with separate thread-setting commands. The turn decider then
validates the thread (and optional source plan) and writes that target thread’s
already-stored runtime and interaction modes into a user thread.message-sent plus
causally linked thread.turn-start-requested. Those command fields therefore should
not be read as a separate per-turn override on this path. Activity may also unsettle
or unsnooze the thread.
case "thread.turn.start": {
const targetThread = yield* requireThread({
readModel,
command,
threadId: command.threadId,
});
const sourceProposedPlan = command.sourceProposedPlan;
const sourceThread = sourceProposedPlan
? yield* requireThread({
readModel,
command,
threadId: sourceProposedPlan.threadId,
})
: null;
const sourcePlan =
sourceProposedPlan && sourceThread
? sourceThread.proposedPlans.find((entry) => entry.id === sourceProposedPlan.planId)
: null;
if (sourceProposedPlan && !sourcePlan) {
return yield* new OrchestrationCommandInvariantError({
commandType: command.type,
detail: `Proposed plan '${sourceProposedPlan.planId}' does not exist on thread '${sourceProposedPlan.threadId}'.`,
});
}
if (sourceThread && sourceThread.projectId !== targetThread.projectId) {
return yield* new OrchestrationCommandInvariantError({
commandType: command.type,
detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`,
});
}
const userMessageEvent: Omit<OrchestrationEvent, "sequence"> = {
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt: command.createdAt,
commandId: command.commandId,
})),
type: "thread.message-sent",
payload: {
threadId: command.threadId,
messageId: command.message.messageId,
role: "user",
text: command.message.text,
attachments: command.message.attachments,
turnId: null,
streaming: false,
createdAt: command.createdAt,
updatedAt: command.createdAt,
},
};
const turnStartRequestedEvent: Omit<OrchestrationEvent, "sequence"> = {
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt: command.createdAt,
commandId: command.commandId,
})),
causationEventId: userMessageEvent.eventId,
type: "thread.turn-start-requested",
payload: {
threadId: command.threadId,
messageId: command.message.messageId,
...(command.modelSelection !== undefined
? { modelSelection: command.modelSelection }
: {}),
...(command.titleSeed !== undefined ? { titleSeed: command.titleSeed } : {}),
runtimeMode: targetThread.runtimeMode,
interactionMode: targetThread.interactionMode,
...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}),
createdAt: command.createdAt,
},
};
// Real activity resets ANY override: it wakes an explicitly settled
// thread, and it clears a keep-active pin back to neutral so the
// thread can auto-settle again after this burst of work goes stale.
// A snooze clears the same way — sending a message to a snoozed
// thread is the user re-engaging, so the return ticket is spent.
const lifecycleResetEvents: Array<Omit<OrchestrationEvent, "sequence">> = [];
if (targetThread.settledOverride !== null) {
lifecycleResetEvents.push({
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt: command.createdAt,
commandId: command.commandId,
})),
type: "thread.unsettled",
payload: {
threadId: command.threadId,
reason: "activity",
updatedAt: command.createdAt,
},
});
}
if (targetThread.snoozedUntil != null) {
lifecycleResetEvents.push({
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt: command.createdAt,
commandId: command.commandId,
})),
type: "thread.unsnoozed",
payload: {
threadId: command.threadId,
reason: "activity",
updatedAt: command.createdAt,
},
});
}
return [...lifecycleResetEvents, userMessageEvent, turnStartRequestedEvent];The engine appends that planned batch, applies synchronous projections, and writes the accepted command receipt in one SQL transaction. Only after commit does it publish each event to its in-memory domain stream. A same-id retry can return the receipt rather than create a second command batch; it does not replay the post-commit publication or provider request.
Bootstrap is a surrounding saga, not a turn-state transition
Bootstrap metadata can ask the command entry path to create a thread, prepare a
worktree, and optionally launch setup before the final turn command. Those steps
span separate commands and filesystem effects. The final thread.turn.start is
still the durable intent boundary described above; a worktree or setup outcome
does not make a provider turn running by itself.
This chapter uses “start” for the final turn command. It does not imply that every first-message bootstrap step is atomic with that command.
2. React: committed intent reaches a hot provider boundary
ProviderCommandReactor subscribes to committed domain events and serializes
its work. For a start request it finds the recorded user message, optionally
launches first-turn naming work, ensures or resumes a provider session, marks the
projected session starting while a start is pending, and calls
ProviderService.sendTurn. The call is forked: its eventual failure is handled
asynchronously.
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);
});If session setup or sendTurn fails, the reactor writes a session with error
status and appends a provider.turn.start.failed activity. That is a later,
separate internal command sequence; it does not rewrite the original accepted
receipt. Retrying is therefore a new intent with a new command id, not an
automatic re-execution promised by the original receipt.
3. Run: correlate provider facts before changing the thread
Each adapter emits canonical ProviderRuntimeEvent values. The provider service
checks that the emitting adapter’s configured instance and driver agree with the
event, fills that instance identity into the event when needed, may log it, and
then publishes it to its own hot stream. threadId is the product anchor;
turnId, item ids, request ids, native references, and provider-instance id are
the correlation material around it.
const correlateRuntimeEventWithInstance = (
source: {
readonly instanceId: ProviderInstanceId;
readonly provider: ProviderDriverKind;
},
event: ProviderRuntimeEvent,
): ProviderRuntimeEvent => {
if (event.provider !== source.provider) {
throw new Error(
`ProviderService.streamEvents: provider instance '${source.instanceId}' is backed by driver '${source.provider}' but emitted driver '${event.provider}'.`,
);
}
if (event.providerInstanceId !== undefined && event.providerInstanceId !== source.instanceId) {
throw new Error(
`ProviderService.streamEvents: provider instance '${source.instanceId}' emitted event for instance '${event.providerInstanceId}'.`,
);
}
return { ...event, providerInstanceId: source.instanceId };
};Runtime ingestion guards lifecycle-changing facts. In particular, a completion
for another active turn—or an untargeted completion while no active turn is
known—is rejected from the lifecycle update. A turn.started that names a new
expected turn may supersede the former turn; this supports provider-specific
steering behavior without allowing arbitrary stale events to overwrite the
thread.
const now = event.createdAt;
const eventTurnId = toTurnId(event.turnId);
const activeTurnId = thread.session?.activeTurnId ?? null;
const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({
threadId: thread.id,
});
const hasPendingTurnStart =
Option.isSome(pendingTurnStart) && thread.session?.status === "starting";
const conflictsWithActiveTurn =
activeTurnId !== null && eventTurnId !== undefined && !sameId(activeTurnId, eventTurnId);
const missingTurnForActiveTurn = activeTurnId !== null && eventTurnId === undefined;
// A turn.started that conflicts with the active turn is legitimate when
// the server itself has a turn start pending for this thread AND the
// provider session already tracks the event's turn as its active turn:
// steering a running turn makes some providers (e.g. opencode) open a
// new turn without ever completing the superseded one. A stale
// turn.started for some other turn id still gets rejected.
const conflictingTurnStartIsPendingTurnStart =
event.type === "turn.started" && conflictsWithActiveTurn
? sameId(yield* getExpectedProviderTurnIdForThread(thread.id), eventTurnId) &&
Option.isSome(pendingTurnStart)
: false;
const shouldApplyThreadLifecycle = (() => {
if (!STRICT_PROVIDER_LIFECYCLE_GUARD) {
return true;
}
switch (event.type) {
case "session.exited":
return true;
case "session.started":
case "thread.started":
return true;
case "turn.started":
return !conflictsWithActiveTurn || conflictingTurnStartIsPendingTurnStart;
case "turn.completed":
if (conflictsWithActiveTurn || missingTurnForActiveTurn) {
return false;
}
// Only the active turn may close the lifecycle state.
if (activeTurnId !== null && eventTurnId !== undefined) {
return sameId(activeTurnId, eventTurnId);
}
// No active turn tracked: accept only completions that name their
// turn (covers a real completion whose turn.started was lost). An
// untargeted completion cannot prove it belongs to any turn this
// thread ran — the known emitter was the Claude resume handshake
// (system/init + result(num_turns: 0)), which is not a turn at
// all — and applying it here stomps the "starting" lifecycle
// state while a turn start is pending.
return eventTurnId !== undefined;
default:
return true;
}
})();
const acceptedTurnStartedSourcePlan =
event.type === "turn.started" && shouldApplyThreadLifecycle
? yield* getSourceProposedPlanReferenceForAcceptedTurnStart(thread.id, eventTurnId)
: null;
if (
event.type === "session.started" ||
event.type === "session.state.changed" ||
event.type === "session.exited" ||
event.type === "thread.started" ||
event.type === "turn.started" ||
event.type === "turn.completed"
) {
const status = (() => {
switch (event.type) {
case "session.state.changed": {
const runtimeStatus = orchestrationSessionStatusFromRuntimeState(event.payload.state);
return hasPendingTurnStart && runtimeStatus === "ready" ? "starting" : runtimeStatus;
}
case "turn.started":
return "running";
case "session.exited":
return "stopped";
case "turn.completed":
return normalizeRuntimeTurnState(event.payload.state) === "failed"
? "error"
: "ready";
case "session.started":
case "thread.started":
// Provider thread/session start notifications can arrive during an
// active or pending turn; preserve that lifecycle state.
return activeTurnId !== null ? "running" : hasPendingTurnStart ? "starting" : "ready";
}
})();
const nextActiveTurnId =
event.type === "turn.started"
? (eventTurnId ?? null)
: event.type === "turn.completed" || event.type === "session.exited"
? null
: event.type === "session.state.changed" &&
!sessionStatusAllowsActiveTurn(
orchestrationSessionStatusFromRuntimeState(event.payload.state),
)
? null
: activeTurnId;
const lastError =
event.type === "session.state.changed" && event.payload.state === "error"
? (event.payload.reason ?? thread.session?.lastError ?? "Provider session error")
: event.type === "turn.completed" &&
normalizeRuntimeTurnState(event.payload.state) === "failed"
? (event.payload.errorMessage ?? thread.session?.lastError ?? "Turn failed")
: status === "ready"
? null
: (thread.session?.lastError ?? null);
if (shouldApplyThreadLifecycle) {
if (event.type === "turn.started" && acceptedTurnStartedSourcePlan !== null) {
yield* markSourceProposedPlanImplemented(
acceptedTurnStartedSourcePlan.sourceThreadId,
acceptedTurnStartedSourcePlan.sourcePlanId,
thread.id,
now,
).pipe(
Effect.catchCause((cause) =>
Effect.logWarning(
"provider runtime ingestion failed to mark source proposed plan",
{
eventId: event.eventId,
eventType: event.type,
cause: Cause.pretty(cause),
},
),
),
);
}
yield* orchestrationEngine.dispatch({
type: "thread.session.set",
commandId: yield* providerCommandId(event, "thread-session-set"),
threadId: thread.id,
session: {
threadId: thread.id,
status,
providerName: event.provider,
...(event.providerInstanceId !== undefined
? { providerInstanceId: event.providerInstanceId }
: {}),
runtimeMode: thread.session?.runtimeMode ?? "full-access",
activeTurnId: nextActiveTurnId,
lastError,
updatedAt: now,
},
createdAt: now,
});
}The resulting durable session projection has practical states such as
starting, running, ready, error, and stopped. A turn row separately
uses running, completed, interrupted, or error. Do not read this as a
universal provider state machine: the runtime union has turn.started,
turn.completed, and turn.aborted, while the projection chooses how to fold
selected events into T3’s user-facing state.
The shared client reducer makes the same distinction: a completed assistant
message does not settle its turn while that turn remains the session’s active
running turn; leaving running is the turn-end signal for the client model.
Zoom with the controls, +/−, or Ctrl/⌘ + trackpad scroll. Enable Pan to drag, use two-finger scrolling, or use the arrow keys. 0 fits the diagram; Esc leaves Pan or expanded view.
Text equivalent
A client starts a turn. The engine durably stores a user message, turn-start request, projections, and accepted receipt. A hot provider reactor starts or resumes a session and sends the turn. The provider emits runtime events. ProviderService correlates them to an instance and hot-publishes them. Runtime ingestion emits later durable commands for session state, assistant messages, approvals, and turn records. A checkpoint reactor independently observes relevant events and records git checkpoints when possible. Provider completion can make the session ready or error; the initial receipt never represents provider completion or checkpoint capture.
apps/server/src/orchestration/decider.ts:926–1036 ↗apps/server/src/orchestration/Layers/OrchestrationEngine.ts:197–259 ↗apps/server/src/orchestration/Layers/ProviderCommandReactor.ts:1060–1174 ↗apps/server/src/provider/Layers/ProviderService.ts:195–213 ↗apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts:1505–1656 ↗apps/server/src/orchestration/Layers/CheckpointReactor.ts:820–937 ↗4. Stream: buffering changes cadence, not ownership
For content.delta with assistant_text, ingestion gets or creates an assistant
message associated with the incoming turn. In the default buffered delivery mode,
it collects text and dispatches only spill chunks; the legacy setting dispatches
each delta directly. On a provider request for the user, ingestion flushes the
buffer and finalizes the active assistant segment. On turn.completed, it
finalizes remembered assistant messages and buffered proposed-plan material.
const assistantDelta =
event.type === "content.delta" && event.payload.streamKind === "assistant_text"
? event.payload.delta
: undefined;
const proposedPlanDelta =
event.type === "turn.proposed.delta" ? event.payload.delta : undefined;
if (assistantDelta && assistantDelta.length > 0) {
const turnId = toTurnId(event.turnId);
const assistantMessageId = yield* getOrCreateAssistantMessageId({
threadId: thread.id,
event,
...(turnId ? { turnId } : {}),
});
if (turnId) {
yield* rememberAssistantMessageId(thread.id, turnId, assistantMessageId);
}
const assistantDeliveryMode: AssistantDeliveryMode = yield* Effect.map(
serverSettingsService.getSettings,
(settings) => (settings.enableLegacyTokenStreaming ? "streaming" : "buffered"),
);
if (assistantDeliveryMode === "buffered") {
const spillChunk = yield* appendBufferedAssistantText(assistantMessageId, assistantDelta);
if (spillChunk.length > 0) {
yield* orchestrationEngine.dispatch({
type: "thread.message.assistant.delta",
commandId: yield* providerCommandId(event, "assistant-delta-buffer-spill"),
threadId: thread.id,
messageId: assistantMessageId,
delta: spillChunk,
...(turnId ? { turnId } : {}),
createdAt: now,
});
}
} else {
yield* orchestrationEngine.dispatch({
type: "thread.message.assistant.delta",
commandId: yield* providerCommandId(event, "assistant-delta"),
threadId: thread.id,
messageId: assistantMessageId,
delta: assistantDelta,
...(turnId ? { turnId } : {}),
createdAt: now,
});
}
}
const pauseForUserTurnId =
event.type === "request.opened" || event.type === "user-input.requested"
? toTurnId(event.turnId)
: undefined;
if (pauseForUserTurnId) {
const detailedThread = yield* getLoadedThreadDetail();
const assistantDeliveryMode: AssistantDeliveryMode = yield* Effect.map(
serverSettingsService.getSettings,
(settings) => (settings.enableLegacyTokenStreaming ? "streaming" : "buffered"),
);
const flushedMessageIds =
assistantDeliveryMode === "buffered"
? yield* flushBufferedAssistantMessagesForTurn({
event,
threadId: thread.id,
turnId: pauseForUserTurnId,
createdAt: now,
commandTag:
event.type === "request.opened"
? "assistant-delta-flush-on-request-opened"
: "assistant-delta-flush-on-user-input-requested",
})
: new Set<MessageId>();
yield* finalizeActiveAssistantSegmentForTurn({
event,
threadId: thread.id,
turnId: pauseForUserTurnId,
createdAt: now,
commandTag:
event.type === "request.opened"
? "assistant-complete-on-request-opened"
: "assistant-complete-on-user-input-requested",
finalDeltaCommandTag:
event.type === "request.opened"
? "assistant-delta-finalize-on-request-opened"
: "assistant-delta-finalize-on-user-input-requested",
hasProjectedMessage:
detailedThread !== null &&
hasAssistantMessageForTurn(detailedThread.messages, pauseForUserTurnId, {
streamingOnly: true,
}),
flushedMessageIds,
});
}Buffering is not a second transcript and it is not a guarantee that every native token becomes a durable event. It is a server-side delivery policy before the next internal commands commit. The durable history remains the projected messages that successfully pass through the orchestration engine.
Waiting is an in-flight turn that needs a person
request.opened and user-input.requested are distinct runtime facts. They
create pending interaction state and close the current assistant segment so a
person can respond. The generic command boundary sends an approval decision or
structured answers back to the active provider session. The turn can remain
running while it waits; the UI deliberately keeps a Stop action available in
this state.
This is not the same as a settled thread. Settlement is a separate thread lifecycle classification and refuses to hide a starting/running session or an open approval/user-input request.
5. Control: interrupt and steer have intentionally different meanings
thread.turn.interrupt durably records thread.turn-interrupt-requested first.
The reactor then interrupts by session, because an orchestration turn id is
not a provider turn id. An interrupt request can immediately mark a matching
turn projection interrupted, but native completion/abortion and session state
still arrive through the provider-runtime path.
case "thread.turn.interrupt": {
yield* requireThread({
readModel,
command,
threadId: command.threadId,
});
return {
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt: command.createdAt,
commandId: command.commandId,
})),
type: "thread.turn-interrupt-requested",
payload: {
threadId: command.threadId,
...(command.turnId !== undefined ? { turnId: command.turnId } : {}),
createdAt: command.createdAt,
},
};There is no generic steer method in ProviderAdapter. Sending another start
request while a provider turn is live may mean “steer” for a concrete adapter;
the ingestion guard explicitly has a controlled superseding-start case. Treat it
as provider-specific behavior, not a cross-provider promise.
6. Finish: completion, error, compaction, and checkpoint are separate signals
turn.completed carries a runtime state and may contain stop reason, usage,
cost, or error text. Ingestion maps a failed completion to session error and
other completion to ready, clears its active turn id, and finalizes buffered
content. A runtime.error can also record session error. turn.aborted clears
plan-progress liveness, but its exact durable session fold is not equivalent to
the turn.completed path in the inspected code.
context_compaction is a canonical item type, and ingestion maps a canonical
thread.state.changed value of compacted into a durable
context-compaction activity. It is still not a documented transition in this
chapter’s session-status switch. The activity records provider-originated
provenance; it must not be presented as a new T3 turn, completion, checkpoint,
or a T3-owned summary artifact.
Checkpointing is a parallel best-effort lifecycle. Its earliest baseline path
observes the durable thread.turn-start-requested or qualifying user-message event
and tries to capture the pre-turn Git state; an observed provider turn.started
provides an additional baseline path. On completion, the reactor can capture a
post-turn ref from a correlated runtime event. Because the shared runtime stream may
miss that completion, ingestion can first create a missing checkpoint from
turn.diff.updated, and the reactor listens to the resulting durable domain event
to replace the placeholder with a real capture. A non-Git workspace, missing id,
checkpoint failure, or stream gap can prevent a checkpoint without invalidating the
turn receipt or provider completion.
function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>) {
const turnId = toTurnId(event.turnId);
if (!turnId) {
return;
}
const thread = yield* resolveThreadDetail(event.threadId);
if (!thread) {
return;
}
// When a primary turn is active, only that turn may produce completion checkpoints.
if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) {
return;
}
// Only skip if a real (non-placeholder) checkpoint already exists for this turn.
// ProviderRuntimeIngestion may insert placeholder entries with status "missing"
// before this reactor runs; those must not prevent real git capture.
if (
thread.checkpoints.some(
(checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",
)
) {
return;
}
const projects = yield* resolveThreadProjects(thread.projectId);
const checkpointCwd = yield* resolveCheckpointCwd({
threadId: thread.id,
thread,
projects,
preferSessionRuntime: true,
});
if (!checkpointCwd) {
return;
}
// If a placeholder checkpoint exists for this turn, reuse its turn count
// instead of incrementing past it.
const existingPlaceholder = thread.checkpoints.find(
(checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",
);
const currentTurnCount = thread.checkpoints.reduce(
(maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),
0,
);
const nextTurnCount = existingPlaceholder
? existingPlaceholder.checkpointTurnCount
: currentTurnCount + 1;
yield* captureAndDispatchCheckpoint({
threadId: thread.id,
turnId,
thread,
cwd: checkpointCwd,
turnCount: nextTurnCount,
status: checkpointStatusFromRuntime(event.payload.state),
assistantMessageId: undefined,
createdAt: event.createdAt,
});
},
);
// Captures a real git checkpoint when a placeholder checkpoint (status "missing")
// is detected via a domain event. This replaces the placeholder with a real
// git-ref-based checkpoint.
//
// ProviderRuntimeIngestion creates placeholder checkpoints on turn.diff.updated
// events from the Codex runtime. This handler fires when the corresponding
// domain event arrives, allowing the reactor to capture the actual filesystem
// state into a git ref and dispatch a replacement checkpoint.
const captureCheckpointFromPlaceholder = Effect.fn("captureCheckpointFromPlaceholder")(function* (
event: Extract<OrchestrationEvent, { type: "thread.turn-diff-completed" }>,
) {
const { threadId, turnId, checkpointTurnCount, status } = event.payload;
// Only replace placeholders; skip events from our own real captures.
if (status !== "missing") {
return;
}
const thread = yield* resolveThreadDetail(threadId);
if (!thread) {
yield* Effect.logWarning("checkpoint capture from placeholder skipped: thread not found", {
threadId,
});
return;
}
// If a real checkpoint already exists for this turn, skip.
if (
thread.checkpoints.some(
(checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",
)
) {
yield* Effect.logDebug(
"checkpoint capture from placeholder skipped: real checkpoint already exists",
{ threadId, turnId },
);
return;
}Exercise illustrative paths
The lab deliberately presents a small set of useful teaching paths; each click moves once and stops. These controls are not a complete server-enforced legality matrix. The turn-start decider validates the target thread and optional source plan, but it has no “must be resting/ready” guard. UI affordances and concrete provider behavior constrain concurrent sends elsewhere. The lab distinguishes the durable record from hot provider liveness and includes a static table for no-script and print use.
One durable request, several live outcomes
Choose an illustrative control for the current teaching state. A step flashes once, then rests; this is neither a live provider simulator nor the server's complete legality matrix.
Current state
Resting
No active provider turn is projected. A new user message can create durable start intent.
- Restingno active turn
- Startingintent awaits runtime
- Runningcorrelated live turn
- Approvalwaiting on a decision
- Inputwaiting on answers
- Interruptedstop intent recorded
- Readyturn ended
- Errornew intent required
Durable record
Nothing new is being committed for this simulated turn.
Hot provider liveness
No provider liveness is claimed.
Teaching paths
State is Resting.
Static transition table and boundary notes
| State | Illustrative controls | Durable meaning | Hot meaning |
|---|---|---|---|
| Resting / ready / error | Start a new intent | new user message + start request + receipt can commit | no provider start is proven yet |
| Starting | Provider starts; interrupt | pending start/session projection can be present | reactor may be preparing or sending |
| Running | Stream; request approval; request input; steer*; interrupt; complete; fail | correlated runtime facts can update session/messages | provider liveness remains hot |
| Approval / input | Respond; interrupt | pending request is visible; the turn can remain running | provider waits for a response |
| Interrupted | Start a new intent | a matching projected turn can be marked interrupted | native abortion or session-state facts may still arrive later |
| Completed / ready | Start a new intent | completion may settle session; checkpoint is separate | native session may still exist or later exit |
* “Steer” is provider-specific. The generic adapter contract has no steer method; this lab models its guarded superseding-turn case. The table is intentionally selective: the start decider itself does not enforce a “must be ready” state guard.
What survives which boundary?
| Observation | What it establishes | What it does not establish |
|---|---|---|
| Accepted start receipt | initial message and start intent committed with projections | provider saw, started, or completed the work |
starting / running session projection |
accepted runtime lifecycle folding has been persisted | the native process is currently reachable |
| Buffered or completed assistant message | a selected content result committed | every native token or tool event was durably preserved |
turn.completed folded to ready or error |
a correlated completion event was accepted by ingestion | checkpoint capture, client delivery, or billing settlement |
| Checkpoint row/ref | checkpoint reactor recorded a filesystem-derived milestone | a provider turn receipt or universal rollback semantics |
| Client “live” / “synchronizing” state | subscription/cache synchronization condition | provider execution state or a durable liveness lease |
Ambiguities worth keeping visible
- The canonical union admits more event variants than every adapter emits. This chapter does not claim that every provider reports compaction, usage, pause, or a matching turn id.
- A pending request makes a thread await a person, but an exact provider-native pause/resume protocol remains adapter-owned.
- The start reactor’s hot subscription and short-lived dedupe cache do not by themselves specify crash recovery for a missed initial start. That would need a durable outbox/reconciler design; none is established by these paths.
- A checkpoint is conditional on workspace and Git conditions. The sources do not establish it as a prerequisite for calling a turn complete.