Task Orchestration

The orchestrator produces the turn once; everyone reads it back

The orchestration layer is two files, split by concern. The transport shell apps/server/src/services/task-orchestrator.ts is thin — it only decides how a turn’s stream is delivered and, if dropped, recovered. The real execution core is apps/server/src/services/task-runner.ts (startTaskUiStream), decoupled from both transport and host process — which is exactly why the API and a BullMQ worker drive the same code. (Desktop mirrors the runner in apps/desktop/src/main/handlers/agent-handler.ts.)

The load-bearing decision is that there is one read core. On the SSE path the run does not stream straight to the client — it produces its UIMessageChunks into a Redis stream buffer keyed by taskId, and execute() responds by reading that buffer back. A client that drops and calls resumeStream(taskId) reads the same buffer the same way. So “live” and “resume” are not two code paths; they are two readers of one produced stream. The run itself never touches the transport — it returns a transport-neutral ReadableStream<UIMessageChunk> and lets the caller frame it.

The orchestrator never touches LLM mechanics; everything inside the agent stream is documented in Agent Engine and its subsystem pages.

The request path: preflight, then enqueue

execute(taskId, user, body) must return an HTTP Response fast, but without running the turn itself — the real work is left to the enqueued run. So it does only transport-shaped work, four steps:

1. Clear the stale abort flag — clearAbortRequest(taskId), first, before any DB round-trip, so a same-turn Stop pressed during the client’s “submitting” phase lands after the clear and survives to be honored at run start.
2. Preflight on the request path — preflightTask(deps, …): getTaskForExecution (ownership / not-found → HTTP 404), creditService.checkQuota (→ HTTP 402), and — crucially — saveReceivedMessage persists the inbound message here. A job that is enqueued but never runs must not silently lose the user’s turn, so the durable write lives on the request path, not inside the run.
3. Enqueue the run — deps.jobQueue.enqueue("task.run", …, () => runTaskToStore(deps, …)). On BullMQ this hands off to a worker; with an inline queue it runs in-process. Either way the run produces into the buffer.
4. Respond by reading the buffer — streamBuffer.read(taskId), the same read a reconnect uses.

The no-Redis fallback (dev / desktop) skips the buffer entirely: startTaskUiStream runs inline and streams straight back. executeWs(taskId, userId, body, send) is the WebSocket wrapper — it drives the same startTaskUiStream and sends each chunk as a task:stream:frame; no Redis buffering, and dropped WS clients refetch DB history on reconnect.

The run: startTaskUiStream

This is where a turn actually runs, returned as a transport-neutral stream. Its rhythm is eager on call, lazy on drain — the setup below runs the moment the function is called; the agent loop and finalize wait until the returned stream is drained (by the buffer producer, the WS reader, or the inline SSE response).

Eager (before the stream):

1. Acquire the lock — acquireTaskLock(taskId) (Redis, 409 if already held) + startLockHeartbeat to renew the TTL so a dead process’s lock expires rather than wedging the task.
2. Bind the task row — getTaskForExecution again (model / approval / planning are task-bound and immutable across turns, because they shape the cached prompt prefix) + creditService.checkQuota (the run is self-contained so a worker re-checks).
3. Load task data — loadTaskData(taskId) → raw uiMessages, a fresh assistantMessageId, and oldAssistantMessage (on resume). Turn start needs only the raw stream; the engine’s buildTurnInput rebuilds the compacted prefix from it every turn (see Compaction).
4. Sandbox + uploads — createSandbox({ id: taskId }), then seedUploadsIntoSandbox materializes any files the user attached this turn so read_file / shell can reach them.
5. Session + abort wiring — create the ExecutionSession; abortManager.create(taskId) for in-process Stop, plus subscribeAbort (Redis abort-bus) so a Stop reaches the run even when it lives in a worker, plus an isAbortRequested re-check to cover the pre-subscribe window.

On drain (createUIMessageStream’s execute callback): setup streams progress states as it goes (agent_building → context_building → mcp_connecting → agent_running; see the state machine):

1. Kick off connectMcpTools concurrently (network-bound, overlaps the rest of setup).
2. buildAgentSetup — parallel resolve of tier, provider keys, agent config, subagent defs, sandbox readiness.
3. Memory service + loadIndex.
4. createRuntimeContext(...) — built once, after the full tool loadout is known.
5. Assemble toolServices (memory, browser bridge, kanban, delivery, wait_and_resume, subagents); strip the team tool (task’s request-response stream has no idle-wake lifecycle for teams).
6. Compaction repos (toolCompaction / snapshot / taskBudget) + buildCompactionDeps — kept on the session so finalize can write the cross-turn anchor.
7. Await MCP tools + tryAttachToolDiscovery; injectUserSkills for any /skill-name activations.
8. runAgentLoop(...) — returns a plain AgentLoopResult { agentStream, stepUsages }. A Stop that lands during setup (before the model emits) is caught and swallowed so the stream closes cleanly as an abort, not an error.
9. writer.merge(toAgentUIMessageStream(...)) — chunks start flowing.

Finalize — onEnd → finalizeExecution

Only once the stream has fully drained does finalize run. Five steps:

1. Stamp the terminal verdict — abort wins over a stray error (a late Stop can surface a spurious onError); a server-shutdown abort is labelled interrupted rather than aborted to keep restart churn out of the user-abort metric. Orphaned in-flight tool parts are terminated so the client doesn’t shimmer forever.
2. taskService.finalizeTurn(...) — persists the assistant message, accumulates usage, updates task metadata; returns { roundUsage, lifecycle }. A task:event WS message tells the client the terminal kind.
3. saveTurnAnchor({ ctx, deps, repos }) — writes the cross-turn real-usage anchor (meta:anchor) into task_compaction_snapshot for the next turn’s preflight. It never writes task_message and enqueues no job; failure is logged, never fatal. (This is the server name; desktop’s agent-handler.ts calls the same step finalizeTurnCompaction.)
4. saveExecutionRecords — written inline, not as a job, because the admin UI needs them immediately.
5. Enqueue three background jobs — credit.consume, resource.index, memory.extraction (fire-and-forget via JobQueue; BullMQ on a worker, in-process on the inline fallback).

Cleanup — the onEnd finally

Unconditional, on both success and failure (and mirrored in the setup-threw catch): unsubscribeAbort → abortManager.remove(taskId) → mcpClientManager.disconnect(taskId) → stopLockHeartbeat() → releaseTaskLock(taskId).

The entities that cross callbacks

createUIMessageStream is callback-driven with no return-value passing between execute / onEnd / onError, so a single mutable ExecutionSession is the only channel between them.

Entity	What it carries	Lifetime
ExecutionSession	abortController, streamStartedAt, stepUsages, and (set in execute) resolvedModelId, context, compactionDeps, compactionRepos, memoryService; error set by onError	created eager, read in onEnd
RuntimeContext	the platform-agnostic agent environment — writer, sandbox, memorySandbox, subagentDefs, todos, reminders, and write() / writeTransient() helpers	built once in execute, used through finalize
CompactionDeps	the shared compaction bundle (createModel, contextWindow, compactionModel) — kept on the session for the memory.extraction job; the engine builds its own internally via buildCompactionDeps	built in execute, consumed by a background job

The session is kept deliberately narrow: only cross-callback data lives on it (sandbox stays a local in execute, never read in onEnd). This is one of three closure-isolation patterns that keep the per-request object graph GC-eligible the moment onEnd returns — see below.

Abort — a real Stop, across processes

A real Stop has to cross process boundaries — the run may be in the API process or in a worker. taskOrchestrator.abort(userId, taskId) (from POST /:id/abort and task:stream:abort) runs in a deliberate order:

1. taskService.abort — ownership check + isActive = false. It throws on unauthorized access, so an attacker can’t enumerate taskIds to cancel other users’ runs. Running it first means an unauthorized caller never reaches step 2.
2. abortManager.abort(taskId) (in-process) + publishAbort(taskId) (Redis abort-bus). The run may be in this process or in a worker, so both fire; aborting is idempotent. The run’s subscribeAbort (installed in startTaskUiStream) receives the bus message; the isAbortRequested re-check covers a Stop that raced the subscription. The signal is threaded into runAgentLoop, which interrupts at its next await point.

Closure isolation and prompt GC

A run’s object graph is heavy — message history, the step runner, the StreamTextResult all hang off it. Three patterns keep it from being pinned, so it drops the moment it should:

Pattern	Effect
Keep ExecutionSession fields minimal	Anything not read in onEnd stays a local in execute (e.g. sandbox)
Return a plain result, not a closure over the run	runAgentLoop returns { agentStream, stepUsages }; the orchestrator holds no accessor into the run, so dropping the locals releases the StreamTextResult + step-runner chain
Build long-lived closures via module-level factories	e.g. the title .then closure captures only taskId + writer, never the enclosing scope

Under BullMQ the run’s closures are discarded immediately (the payload is serialized to Redis and the worker rebuilds its deps); under the inline fallback the discipline is what lets the runResult + step runner + message history drop together the moment onEnd returns.

Desktop parallel

agent-handler.ts mirrors the runner’s shape but drops HTTP-centric concerns:

Concern	Server	Desktop
Distributed lock	Redis SET NX + heartbeat	not needed (single process)
Credit quota	checked in preflight + run	not needed
Produce/read-back	Redis stream buffer	not needed (IPC stays alive)
Cross-process abort	Redis abort-bus	in-process abortManager only
Job queue	BullMQ on Redis	inline fire-and-forget
Turn-end anchor	saveTurnAnchor	finalizeTurnCompaction

The phase shape and the closure-isolation discipline are identical — the GC cost is most visible on desktop, where jobs run inline.

Related docs

• Agent Engine — the loop the runner hosts
• Context Compaction — the part-addressed checkpoint and the cross-turn anchor saveTurnAnchor writes
• Streaming Architecture — the resumable SSE buffer, the produce/consume split, and WS transport
• Background Job Queue — BullMQ vs inline, and the three post-turn jobs
• Memory System — the memory sandbox and the memory.extraction job