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.
{
type: "thread.pull-request.unlink" as const,
commandId: command.commandId,
threadId: command.threadId,
host: currentPullRequest.host,
repository: currentPullRequest.repository,
number: currentPullRequest.number,
},
]
: []),
{
type: "thread.pull-request.link",
commandId: command.commandId,
threadId: command.threadId,
...legacyThreadPullRequestKey(linked, host),
url: linked.url,
source: "manual",
},
],
});
}
if (command.linkedPullRequest === null && currentPullRequest !== null) {
const { linkedPullRequest: _linkedPullRequest, ...metadata } = command;
const hasMetadata = Object.entries(metadata).some(
([key, value]) => !["type", "commandId", "threadId"].includes(key) && value !== undefined,
);
return yield* decideCommandSequence({
readModel,
commands: [
...(hasMetadata ? [metadata] : []),
{
type: "thread.pull-request.unlink",
commandId: command.commandId,
threadId: command.threadId,
host: currentPullRequest.host,
repository: currentPullRequest.repository,
number: currentPullRequest.number,
},
],
});
}
const branch =
command.branch !== undefined &&
command.expectedBranch !== undefined &&
thread.branch !== command.expectedBranch
? thread.branch
: command.branch;
const occurredAt = yield* nowIso;
return {
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt,
commandId: command.commandId,
})),
type: "thread.meta-updated",
payload: {
threadId: command.threadId,
...(command.title !== undefined ? { title: command.title } : {}),
...(command.regenerateTitle === true
? {
regenerateTitle: true as const,
previousTitle: thread.title,
titleRegeneration: {
requestId: command.commandId,
startedAt: occurredAt,
},
}
: {}),
...(command.title !== undefined && thread.titleRegeneration != null
? { titleRegeneration: null }
: {}),
...(command.modelSelection !== undefined
? { modelSelection: command.modelSelection }
: {}),
...(branch !== undefined ? { branch } : {}),
...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}),
...(command.linkedPullRequest !== undefined
? { linkedPullRequest: command.linkedPullRequest }
: {}),
updatedAt: occurredAt,
},
};
}
case "thread.pull-request.link": {
const thread = yield* requireThread({
readModel,
command,
threadId: command.threadId,
});
const key = normalizeThreadPullRequestKey(command);
const existing = findPullRequestLink(thread, key);
// An explicit link on a dismissed stack member un-dismisses it; any
// other duplicate is a no-op the engine would reject as zero-event.
const undismisses =
existing?.source === "stack-dismissed" &&
(command.source === "manual" || command.source === "agent" || command.source === "created");
if (existing !== undefined && !undismisses) {
return yield* new OrchestrationCommandInvariantError({
commandType: command.type,
detail: `pull request ${key.host}/${key.repository}#${key.number} is already linked to thread ${command.threadId}`,
});
}
const occurredAt = yield* nowIso;
return {
...(yield* withEventBase({
aggregateKind: "thread",
aggregateId: command.threadId,
occurredAt,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.
return { _tag: "Superseded" } as const;
}
return { _tag: "Completed", title: generated.title } as const;
});
const dispatchThreadTitleRegenerationCompletion = Effect.fn(
"dispatchThreadTitleRegenerationCompletion",
)(function* (input: {
readonly threadId: ThreadId;
readonly requestId: CommandId;
readonly title?: string;
}) {
yield* orchestrationEngine.dispatch({
type: "thread.title.regeneration.complete",
commandId: yield* serverCommandId("thread-title-regeneration-complete"),
threadId: input.threadId,
requestId: input.requestId,
...(input.title !== undefined ? { title: input.title } : {}),
});
});
const findInterruptedThreadTitleRegenerations = Effect.fn(
"findInterruptedThreadTitleRegenerations",
)(function* () {
const readModel = yield* projectionSnapshotQuery.getCommandReadModel();
return readModel.threads.flatMap((thread) => {
const requestId = thread.titleRegeneration?.requestId;
return requestId === undefined ? [] : [{ threadId: thread.id, requestId }];
});
});
const clearInterruptedThreadTitleRegenerations = Effect.fn(
"clearInterruptedThreadTitleRegenerations",
)(function* (
interrupted: ReadonlyArray<{ readonly threadId: ThreadId; readonly requestId: CommandId }>,
) {
yield* Effect.forEach(
interrupted,
({ threadId, requestId }) => {
return dispatchThreadTitleRegenerationCompletion({
threadId,
requestId,
}).pipe(
Effect.catchCause((cause) => {
if (Cause.hasInterruptsOnly(cause)) {
return Effect.interrupt;
}
return Effect.logWarning(
"provider command reactor failed to clear interrupted title regeneration",
{
threadId,
cause: Cause.pretty(cause),
},
);
}),
);
},
{ discard: true },
);
});
const processThreadTitleRegenerationSafely = Effect.fn("processThreadTitleRegenerationSafely")(
function* (event: Extract<ProviderIntentEvent, { type: "thread.meta-updated" }>) {
if (event.payload.regenerateTitle !== true) {
return;
}
const requestId = event.payload.titleRegeneration?.requestId ?? event.commandId;
if (requestId === null) {
return;
}
const result = yield* regenerateThreadTitle(event, requestId).pipe(
Effect.catchCause((cause) => {
if (Cause.hasInterruptsOnly(cause)) {
return Effect.failCause(cause);
}
return Effect.logWarning("provider command reactor failed to regenerate thread title", {
threadId: event.payload.threadId,
cause: Cause.pretty(cause),
}).pipe(Effect.as({ _tag: "Completed", title: undefined } as const));
}),
);
if (result._tag === "Superseded") {
return;
}
const completion = {
threadId: event.payload.threadId,
requestId,
...(result.title !== undefined ? { title: result.title } : {}),
};
yield* dispatchThreadTitleRegenerationCompletion(completion).pipe(
Effect.catchCause((cause) => {
if (Cause.hasInterruptsOnly(cause)) {
return Effect.failCause(cause);
}
return Effect.logWarning(
"provider command reactor retrying title regeneration completion",
{
threadId: event.payload.threadId,
cause: Cause.pretty(cause),
},
).pipe(Effect.andThen(dispatchThreadTitleRegenerationCompletion(completion)));
}),
);
},
(effect, event) =>
effect.pipe(
Effect.catchCause((cause) => {
if (Cause.hasInterruptsOnly(cause)) {
return Effect.failCause(cause);
}
return Effect.logWarning(
"provider command reactor failed to complete title regeneration",
{
threadId: event.payload.threadId,
cause: Cause.pretty(cause),
},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.
) {
return [];
}
return [compacted];
}
function accessibilityNodeHasBounds(node: SnapShotPromptAccessibilityNode): boolean {
return Boolean(node.bounds || node.children?.some(accessibilityNodeHasBounds));
}
function compactAccessibilityForPrompt(
accessibility: SnapShotAccessibility,
): SnapShotPromptAccessibility {
if (accessibility.format === "flat-text") {
return {
format: "flat-text",
text: accessibility.text,
...(accessibility.truncated ? { truncated: true } : {}),
};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.
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":
case "turn.aborted":
if (conflictsWithActiveTurn || missingTurnForActiveTurn) {
return false;
}
// Only the active turn may close the lifecycle state.
if (activeTurnId !== null && eventTurnId !== undefined) {
return sameId(activeTurnId, eventTurnId);
}
// A named completion can recover a lost turn.started event.
// An abort needs an active turn so a delayed stop cannot replace
// a ready session or clear a newer pending start.
return event.type === "turn.completed" && 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" ||
isTerminalTurn
) {
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.aborted":
return "interrupted";
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)
: isTerminalTurn || 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" || status === "interrupted"
? 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,
});
}
}
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;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 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" && event.payload.responseMode !== "message")
? toTurnId(event.turnId)
: undefined;
if (pauseForUserTurnId) {
const hasProjectedMessage = yield* projectionThreadMessages.hasAssistantMessageForTurn({
threadId: thread.id,
turnId: pauseForUserTurnId,
streamingOnly: true,
});
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,
flushedMessageIds,
});
}
if (proposedPlanDelta && proposedPlanDelta.length > 0) {
const planId = proposedPlanIdFromEvent(event, thread.id);
yield* appendBufferedProposedPlan(planId, proposedPlanDelta, now);
}
const assistantCompletion =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.
type: "thread.pull-request-linked",
payload: {
threadId: command.threadId,
link:
existing !== undefined
? { ...existing, url: command.url, source: command.source }
: {
...key,
url: command.url,
source: command.source,
linkedAt: occurredAt,
snapshot: null,
stack: null,
},
updatedAt: occurredAt,
},
};
}
case "thread.pull-request.unlink": {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.
});
});
// Capture the files left by a completed or interrupted turn.
const captureCheckpointFromTurnCompletion = Effect.fn("captureCheckpointFromTurnCompletion")(
function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" | "turn.aborted" }>) {
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:
event.type === "turn.aborted"
? "ready"
: checkpointStatusFromRuntime(event.payload.state),
assistantMessageId: existingPlaceholder?.assistantMessageId ?? undefined,
createdAt: event.createdAt,
});
},
);
const ensurePreTurnBaselineFromTurnStart = Effect.fn("ensurePreTurnBaselineFromTurnStart")(
function* (event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>) {
const turnId = toTurnId(event.turnId);
if (!turnId) {
return;
}
const thread = yield* resolveThreadDetail(event.threadId);
if (!thread) {
return;
}
const projects = yield* resolveThreadProjects(thread.projectId);
const checkpointCwd = yield* resolveCheckpointCwd({
threadId: thread.id,
thread,
projects,
preferSessionRuntime: false,
});
if (!checkpointCwd) {
return;
}
const currentTurnCount = thread.checkpoints.reduce(
(maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),
0,
);
const baselineCheckpointRef = checkpointRefForThreadTurn(thread.id, currentTurnCount);
const baselineExists = yield* checkpointStore.hasCheckpointRef({
cwd: checkpointCwd,
checkpointRef: baselineCheckpointRef,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.