Task Orchestration

The orchestrator produces the turn once; everyone reads it back

The orchestration layer is 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 server adapter is apps/server/src/services/task-runner.ts (startTaskUiStream), and the real execution core is packages/core/src/services/agent/run-task-turn.ts (runTaskTurn). That split is why the API process, the BullMQ worker, and desktop can all drive the same turn body with different ports. Desktop injects its local subset in apps/desktop/src/main/handlers/agent-stream.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.

Design decisions

The request path is allowed to do only the work that must happen before a job might disappear: authorize the user, check quota, persist the inbound message, and reset the stream buffer. Everything else moves into the run so retries, worker execution, and desktop inline execution share the same semantics.

The stream buffer is also a product decision. Live delivery and resume are not two implementations; they are two readers of one already-produced turn. Disconnect recovery therefore belongs to the task runtime rather than to a browser-side patch.

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. It does only transport-shaped work, five steps:

1. Clear the stale abort flag — clearAbortRequest(taskId), first, before any DB round-trip. This POST is sent before the user can even hit Stop, so clearing at the very top is what lets a same-turn Stop’s publishAbort land after the clear and survive 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. Reset the previous turn’s buffer — streamBuffer.reset(taskId). Task streams reuse one key per taskId across turns, so without a reset this turn’s fresh reader would backfill the prior turn’s still-within-TTL chunks — replaying the last answer into this turn. The producer’s own reset runs too late (in the worker, after the reader has already read), so the reset lives here, before the reader is returned, on the request path.
4. 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.
5. Respond by reading the buffer — streamBuffer.read(taskId), the same buffer a reconnect (resumeStream) reads.

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 / knowledge are all task-bound and immutable across turns: they shape the cached prompt prefix, so they’re locked at create and never read per-message — changing one mid-task would break the replayed history / prompt cache / context-window assumptions) + 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 — resolve tier first, then in parallel agent config (with provider key), subagent defs, sandbox readiness.
3. Memory service + loadIndex.
4. createRuntimeContext(...) — built once, after the full tool loadout is known.
5. Assemble toolServices (memory, knowledge — via createScopedKnowledgeService, scoped to only this task’s attached docs, workbench, 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

The run hands its chunks to the stream, but the story isn’t over — finalize has to wait. Only once the stream has fully drained does onEnd fire finalizeExecution. 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. Server and desktop now both reach this through finalizeTaskTurn.
4. saveExecutionRecords — awaited inline, not enqueued: it’s just a cheap local telemetry write (token usage, per-step task_step_info, subagent execution records); the heavy work and external side-effects are what get left to the jobs below.
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

Finalize persists the turn’s results; cleanup hands back the resources the turn held. 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.

With both steps done, the orchestrator doesn’t wait for the agent to wind down — it immediately pushes a task:event (kind: "aborted") WS message to the user. Finalize will push an equivalent terminal event again later, but that’s a harmless duplicate: aborted raises no toast, so the client at most invalidates its list once more.

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

apps/desktop/src/main/handlers/agent-stream.ts is the desktop host for the same runTaskTurn core. It serves task streaming through the loopback HTTP route POST /api/tasks/:id/messages, so the renderer consumes SSE like Web rather than the old agent:run IPC path.

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 via finalizeTaskTurn	saveTurnAnchor via finalizeTaskTurn

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