From 7348e35a11dc2d03f72ac2e2bc7cc23aa7a3719b Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 15:20:34 -0700 Subject: [PATCH] docs: capture observable-agent-turns architecture pattern and seed CE session vocabulary ce-compound learning from the live-output/steering work: push-channel- alongside-pull-contract, void-safe detached turns, inactivity watchdog, transient liveActivity + persisted trace, replay suppression. Also adds a "Compound Engineering sessions" cluster to CONCEPTS.md and refreshes the plugin reference doc's Sessions section with the new transport behaviors. Co-Authored-By: Claude Opus 4.8 (1M context) --- CONCEPTS.md | 20 +++ docs/plugins/compound-engineering.md | 11 ++ ...urns-through-blocking-plugin-route-seam.md | 166 ++++++++++++++++++ 3 files changed, 197 insertions(+) create mode 100644 docs/solutions/architecture-patterns/observable-long-running-agent-turns-through-blocking-plugin-route-seam.md diff --git a/CONCEPTS.md b/CONCEPTS.md index 9f015d0e39..c3e67766cf 100644 --- a/CONCEPTS.md +++ b/CONCEPTS.md @@ -27,6 +27,26 @@ A recurring background scan that detects and repairs stuck Task states — stall ### Shared branch group A set of Tasks integrating into a common shared branch instead of each merging straight to the project's default branch. Member integration (task branch → shared branch) is a soft pre-integration step exempt from the global auto-merge gate; promotion (shared branch → default branch) is gated separately. +## Compound Engineering sessions + +### CE Stage +A registered step of the compound-engineering pipeline (e.g. brainstorm, plan, work, compound), each mapped to a bundled skill and a conventional artifact location. Adding a stage is a registry data entry, not new code surface. + +### CE Session +A single interactive run of a CE Stage: an agent drives a question/answer flow with the user and produces the stage's artifact on completion. Sessions are independent pipeline runs — many can exist concurrently, each with its own lifecycle (launching, active, awaiting-input, completed, error, interrupted) and conversation history. A completed work-stage CE Session lands derived Tasks on the board, linked back to the session for provenance. + +### Detached turn +The execution posture for CE Session agent turns: the request that triggers a turn returns as soon as the session reflects it, and the turn runs in the background while clients converge through push events and polling. A detached turn never rejects — every failure persists into session state and emits an observable event, so progress is never silently lost. + +### Live activity +The transient working output of an in-flight agent turn — accumulated thinking, streamed text, and tool execution markers. It is observable while the turn runs but is not session state; when the turn settles or is interrupted, a condensed trace is folded into the conversation history so the transcript keeps the story. + +### Steering +The user's mid-stage feedback channel: free-text guidance attached to an answer, or sent on its own without answering the pending question. Agents treat steering as first-class input — incorporate it, adjust course, and either re-ask or proceed. + +### Rehydration +Re-establishing a live agent handle for a paused CE Session by replaying its recorded conversation against the model. Replay is side-effect-suppressed: it reconstructs the agent's context without re-emitting events, re-streaming Live activity, or re-writing artifacts. + ## Flagged ambiguities - "Merging" a shared-branch-group Task had been used for both member integration and group promotion — these are distinct steps with independent gating and must not be conflated. diff --git a/docs/plugins/compound-engineering.md b/docs/plugins/compound-engineering.md index d372bc6c89..2447daae98 100644 --- a/docs/plugins/compound-engineering.md +++ b/docs/plugins/compound-engineering.md @@ -41,6 +41,17 @@ The orchestrator streams `thinking`/`text` turns, surfaces a structured `interrupted`. Interrupt/error auto-saves progress and emits an observable event; sessions resume/retry back to their current question. +Turn execution is **detached**: start/answer/resume return as soon as the +session row reflects the request, with the agent turn running in the background +(failures persist into session state — never an unhandled rejection). While a +turn runs, the engine streams mid-turn progress (thinking/text deltas + tool +markers) through the seam's `onProgress` option; the orchestrator buffers it +and `GET /sessions/:id` attaches it as transient `liveActivity`. The per-turn +timeout is **inactivity-based** (progress re-arms it), so long actively-working +turns are never killed; on settle/interrupt the working trace is condensed into +the conversation history. Users can also **steer** mid-stage: answers may carry +free-text guidance (`{value, comment}`) or be guidance-only (`{feedback}`). + Updates are **pushed** over the shared `/api/events` SSE stream: the orchestrator emits via `ctx.emitEvent`, the host forwards them as project-scoped `plugin:custom` events, and the view subscribes through the host diff --git a/docs/solutions/architecture-patterns/observable-long-running-agent-turns-through-blocking-plugin-route-seam.md b/docs/solutions/architecture-patterns/observable-long-running-agent-turns-through-blocking-plugin-route-seam.md new file mode 100644 index 0000000000..4390553f27 --- /dev/null +++ b/docs/solutions/architecture-patterns/observable-long-running-agent-turns-through-blocking-plugin-route-seam.md @@ -0,0 +1,166 @@ +--- +title: "Observable long-running agent turns through a blocking plugin-route seam" +date: 2026-06-03 +category: architecture-patterns +module: fusion-plugin-compound-engineering +problem_type: architecture_pattern +component: service_object +severity: high +applies_when: + - "A plugin/HTTP route drives a long-running interactive agent turn behind a blocking request/response seam" + - "Mid-turn agent output (thinking, tool calls, streamed text) is swallowed by a pull-based settle-only event API" + - "A fixed per-turn timeout risks killing legitimately long, tool-heavy turns that are still actively working" + - "Clients need live visibility into agent work without persisting transient activity into durable state" + - "A paused, stateful agent session must be resumable across process restarts without re-emitting prior output" +symptoms: + - "Client blocks on a single POST for minutes with zero visibility into agent progress" + - "All mid-turn thinking, tool calls, and streamed text are swallowed; only question/complete/error surface" + - "A fixed 120s per-turn timeout interrupts legitimately long tool-heavy turns while the agent is still working" +root_cause: async_timing +resolution_type: code_fix +related_components: + - tooling + - frontend_stimulus +tags: + - agent-observability + - sse + - streaming + - detached-execution + - plugin-routes + - interactive-session + - inactivity-timeout + - live-activity + - compound-engineering +--- + +# Observable long-running agent turns through a blocking plugin-route seam + +## Context + +The compound-engineering bundled plugin runs interactive CE-stage agent sessions through plugin routes. The host exposes a deliberately minimal **pull-based** interactive seam (`packages/core/src/plugin-types.ts`): the caller drives one `prompt`/`answer` per turn and awaits `nextEvent()`, which resolves only when the turn *settles* (`question` | `complete` | `error`). That contract is simple to drive deterministically from a route or a scripted test — but it had three structural consequences that surfaced as user-visible failures: + +1. **All mid-turn output was swallowed.** `nextEvent()` does not resolve on intermediate thinking/text/tool activity, so a multi-minute tool-heavy turn produced *nothing* observable until it finished. +2. **Routes blocked blind.** The POST handler ran the whole turn synchronously inside the request, so clients waited minutes with no feedback (and, for the opening turn, no session id to poll). +3. **A fixed 120s turn timeout killed turns that were actively working** — long, legitimately-busy turns hit the wall and died. + +The fix made the agent's work live-streamable, made routes non-blocking (detached turns), made the timeout inactivity-based, and persisted the working trace into the transcript across settle/interrupt and process restarts. + +## Guidance + +### 1. Keep the pull-based settle contract; add a SEPARATE push channel + +Don't convert `nextEvent()` into a stream. Live visibility is a *new, optional, additive* callback (`onProgress`) on the session options — the terminal-only pull semantics are untouched. Scripted test fakes that drive `prompt`/`nextEvent` are completely unaffected, and factories that can't stream simply ignore the option. + +```ts +// packages/core/src/plugin-types.ts +export interface CreateInteractiveAiSessionOptions { + // ... + /** Live progress callback, invoked WHILE a turn runs (the pull-based + * nextEvent() only resolves once the turn settles). Must not throw — + * implementations should swallow callback errors. */ + onProgress?: (event: InteractiveAiSessionProgressEvent) => void; +} +``` + +### 2. Deltas, not snapshots; the consumer accumulates + +Progress events carry incremental *deltas*. The consumer owns accumulation, merge-by-kind, and capping — the protocol stays tiny and the producer holds no buffer state. + +```ts +export type InteractiveAiSessionProgressEvent = + | { type: "thinking"; delta: string } // incremental DELTA, not a snapshot + | { type: "text"; delta: string } + | { type: "tool"; name: string; phase: "start" | "end"; isError?: boolean }; +``` + +The engine adapter (`packages/engine/src/index.ts`) maps the underlying agent hooks (`onText`/`onThinking`/`onToolStart`/`onToolEnd`) into these deltas, and **every callback is wrapped in try/catch so a consumer error can never break the agent turn**. The consumer (`orchestrator.handleProgress`) merges consecutive same-kind deltas into one activity turn, opens/closes discrete tool turns, and caps both per-turn chars and turn count, dropping the oldest (the tail is what the user is watching). + +### 3. Detached turns must NEVER reject + +Routes return immediately after the session row exists; the turn runs as a floating background promise (`void turn`). For that to be void-safe, the background promise can have *no* rejection path: factory-create failure, driver throw, and timeout all resolve into a persisted state transition (`failSession` / `interruptSession` / `applyEvent`) plus an emitted observable event. No unhandled rejections, no silent loss. + +```ts +// orchestrator.start +const turn = this.runOpeningTurn(session.id, stage, opts.openingMessage); +if (opts.detach) { + void turn; // never rejects (failures persist into state) + return { session: this.requireSession(session.id) }; +} +``` + +### 4. Inactivity watchdog, not a fixed turn timeout + +The watchdog rejects only after `turnTimeoutMs` of *no progress*; each progress event stamps `lastProgressAt` and re-arms it, so an actively-working turn survives indefinitely. With a non-streaming factory (no progress ever arrives), it degrades cleanly to the old fixed per-turn timeout. + +```ts +const check = () => { + if (cancelled) return; // cancel() stops the loop when the turn settles + const elapsed = Date.now() - (this.lastProgressAt.get(sessionId) ?? 0); + if (elapsed >= this.turnTimeoutMs) { reject(new CeTurnTimeoutError(this.turnTimeoutMs)); return; } + timer = setTimeout(check, this.turnTimeoutMs - elapsed); // re-arm to the remaining window + timer.unref?.(); +}; +``` + +A watchdog rejection is caught and becomes a preserved-progress `interrupted` state — never silent. + +### 5. Live activity is transient; flush a condensed trace into history on settle/interrupt + +The mid-turn buffer lives only in memory; the GET route reads it from the orchestrator (`getLiveActivity(id)`) and attaches it as a transient `liveActivity` field on the response — never written as session state during the turn. + +On settle (`question`/`complete`/`error`) or interrupt, `flushActivity` writes a condensed copy into conversation history **before** the settling record, so the transcript retains the working trace across restarts. + +### 6. Suppress progress (and all side effects) during rehydration replay + +Resume re-creates a live handle by replaying recorded user turns against the model. That replay re-streams old output — which must not be re-emitted as new work. A `replaying` set gates `handleProgress`, and the replay drains one event per drive but **discards** it (no persist/emit/artifact-write). + +### 7. Throttle push emits and bump the staleness anchor on the same beat + +Progress is high-frequency; per-delta SSE emits would flood clients. Throttle to one emit per interval (500ms here), and on that same beat bump the persisted liveness anchor (`lastActivityAt`) so the stale-session recovery rubric sees an actively-working turn as alive rather than abandoned. The client converges via **push + poll**: an SSE event triggers an immediate refetch (low latency), and a poll interval runs while the turn is mid-flight as a fallback — stopping the moment the session settles. + +## Why This Matters + +- **Observability without protocol churn.** A push side-channel gives live visibility while preserving a settle contract that's trivial to drive deterministically from routes and tests. Converting `nextEvent()` into a stream would have rewritten every consumer and every scripted fake for a purely additive feature. +- **Non-blocking routes need void-safe background work.** Detaching a turn is only safe if the background promise has no rejection path. Routing *every* failure into persisted state + an emitted event is what makes `void turn` correct rather than a latent unhandled-rejection bug. +- **Activity is the liveness signal.** An inactivity watchdog encodes the real intent ("is it still working?") instead of a proxy ("has it taken too long?"), and folding the same signal into the staleness anchor keeps two independent health rubrics coherent. +- **Resilience across restarts.** Persisting a condensed trace on settle/interrupt, plus side-effect-suppressed rehydration, means a paused session resumes in a fresh process with its history intact and without double-streaming. + +## When to Apply + +- Surfacing live agent (or any long-running job) work through a request/response or pull-based seam that only resolves on terminal events +- A route runs a multi-minute operation and clients currently block with no progress and no handle to poll +- A fixed timeout is killing work that is legitimately still active +- Resuming a paused, stateful session across process restarts without re-emitting prior output + +Apply the *push-channel-alongside-pull-contract* and *void-safe-detached-turn* patterns together; they're complementary. Don't reach for this when the operation is short and synchronous — the transient buffer, watchdog, and rehydration machinery are overhead you don't need. + +## Examples + +Before/after, distilled: + +- **Before:** route `await`s the entire turn inside POST; client gets nothing for minutes; mid-turn output is dropped because `nextEvent()` only resolves on settle; a fixed 120s timeout kills busy turns. +- **After:** POST returns `201 {session}` immediately with `detach: true`; `onProgress` deltas accumulate into a transient buffer attached at GET; an inactivity watchdog re-armed by progress lets busy turns run; failures persist into state + emit; resume rehydrates with replay suppressed. + +The regression tests (`plugins/fusion-plugin-compound-engineering/src/__tests__/orchestrator-live-output.test.ts`) lock the load-bearing behaviors: + +- A busy turn pumped with `thinking` deltas at ~45ms intervals for ~3× the 120ms test timeout stays `active`; once quiet, it flips to `interrupted` with the activity trace present in history. +- `answer(detach)` returns immediately (`status: active`, `currentQuestion: null`), then the background turn converges to the next question. +- `start(detach)` with an exploding factory converges to `status: error` with the message preserved and an observable error event emitted — never silent. + +Failure modes this prevents: + +1. Silent mid-turn blackout (pull-only API resolves nothing until terminal) +2. Blocked-blind routes (request held open for the full turn, no handle to poll) +3. Killed-while-working timeouts (fixed timeout vs. activity-based liveness) +4. Unhandled rejection / silent loss from floating detached turns +5. Replay double-streaming during rehydration +6. SSE flooding from per-delta emits +7. Stale-rubric false positives on busy sessions (liveness not bumped with activity) +8. Lost transcript on interrupt/settle (transient buffer never condensed into history) +9. Consumer `onProgress` errors breaking the agent turn (guarded at the adapter) + +## Related + +- [Plugin-bundled skills silently fail to load in interactive sessions](../integration-issues/plugin-bundled-skills-not-loading-in-interactive-sessions.md) — sibling learning on the same `CreateInteractiveAiSessionOptions` seam (it added `requestedSkillNames`/`additionalSkillPaths`; this one adds `onProgress`) +- `docs/plugins/compound-engineering.md` §Sessions — the reference doc for the CE session transport (push + poll) +- Key files: `packages/core/src/plugin-types.ts`, `packages/engine/src/index.ts` (interactive adapter), `plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts`, `src/routes/session-routes.ts`, `src/dashboard/hooks/useCeSession.ts`