Skip to content

Commit 87be539

Browse files
KyleAMathewsclaude
KyleAMathews
and
claude
authored
fix(agents): pull-wake runner resilience — fetch cache stacking, hung connections, xstate lifecycle (#4542)
## Summary Fixes two bugs that left the desktop agents runtime unable to pick up sessions until a full app restart, then ports the pull-wake runner's lifecycle to an xstate v5 state machine so this class of bug can't recur silently. No public API changes — all 21 pre-existing behavioral tests pass unmodified. ## Root cause Two independent failures combined into "runner looks alive but never picks up work, and restarting the runtime doesn't help": 1. **Hung connection couldn't be aborted.** The heartbeat-failure → stream-reset path (`requestStreamReconnect`) was a no-op unless `streamConnected` was true. If the wake stream hung while *(re)connecting* (e.g. after an agents-server restart), heartbeat failures counted up but nothing could cancel the stuck `streamFactory` call — the runner sat in `connecting` forever. 2. **Runtime restart stacked cache interceptors.** Every `BuiltinAgentsServer.start()` composed a new undici cache interceptor onto the global dispatcher without a guard. Restarting the runtime in-process stacked SQLite-backed cache layers over the same file; only a full app restart got a clean dispatcher. ## Approach - `installDurableStreamsFetchCache` gets a module-level installed guard (warns on repeat calls). - The runner lifecycle moves into an xstate machine (`pull-wake-machine.ts`): `stopped → running.{connecting, streaming, reconnecting} → stopping`. The structural win: **invoked actors are auto-aborted on state exit**, so a `STREAM_RESET` event during `connecting` cancels the in-flight connect via the promise actor's signal — no manual `connectAbort` bookkeeping, which is how the original bug slipped through. Backoff timers (`after`) cancel the same way. - Diagnostics, heartbeat coalescing, and claim processing stay in the runner closure as effects the machine triggers (`PullWakeMachineEffects`). The machine decides *when*; the closure does *what*. ## Key invariants - Every (state × event) pair is pinned by an exhaustive 35-case transition matrix test — adding a state or event forces a deliberate decision. - Repeated heartbeat failures (≥2) reset the stream from **any** running substate. - One shutdown sequence regardless of concurrent `stop()` calls; `stop()` rejects with drain errors. - Claim actors survive stream reconnects and only drain (1s grace) at stop. ## Non-goals - No behavioral changes to claiming, heartbeat cadence, or backoff timing (1s → ×2 → 30s cap preserved). - Heartbeat send/coalescing logic was not moved into the machine — it's effect plumbing, not lifecycle. ## Verification ```bash pnpm --filter @electric-ax/agents-runtime exec vitest run test/pull-wake-runner.test.ts # 61 tests pnpm --filter @electric-ax/agents exec vitest run test/durable-streams-cache.test.ts # 2 tests pnpm --filter @electric-ax/agents-runtime build ``` Full agents-runtime suite: 852 passed; 2 pre-existing environment failures unrelated to this PR (`sandbox-docker` Docker-timing assertion, `tool-providers` unbuilt workspace dep). ## Files changed - `packages/agents-runtime/src/pull-wake-machine.ts` — new: xstate v5 lifecycle machine + effects interface - `packages/agents-runtime/src/pull-wake-runner.ts` — rewritten as a thin adapter: same public API, closure keeps diagnostics/heartbeat/claims, machine owns lifecycle - `packages/agents-runtime/test/pull-wake-runner.test.ts` — +1 hung-connection regression test, +40 machine transition tests; 21 original tests untouched - `packages/agents/src/durable-streams-cache.ts` — idempotency guard + warning - `packages/agents/test/durable-streams-cache.test.ts` — idempotency test - `packages/agents-runtime/package.json` — adds `xstate` (zero runtime deps) - `.changeset/pull-wake-runner-resilience.md` — patch bumps for both packages 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 004bea1 commit 87be539

8 files changed

Lines changed: 645 additions & 186 deletions

File tree

‎.changeset/pull-wake-runner-resilience.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
---
2+
'@electric-ax/agents-runtime': patch
3+
'@electric-ax/agents': patch
4+
---
5+
6+
Fix two resilience bugs that could leave the desktop agents runtime unable to pick up sessions until a full app restart, and port the pull-wake runner lifecycle to an xstate state machine.
7+
8+
- `installDurableStreamsFetchCache` is now idempotent (with a warning on repeat calls), so restarting the built-in agents runtime no longer stacks duplicate HTTP cache interceptors on the global undici dispatcher.
9+
- The pull-wake runner now recovers when the wake stream connection hangs during the connecting phase: repeated heartbeat failures abort the in-flight connection attempt instead of only resetting an already-established stream.
10+
- The runner lifecycle (stopped → connecting → streaming → reconnecting → stopping) is now an xstate machine, so in-flight connections, stream sessions, and backoff timers are cancelled automatically on state transitions, and every state × event pair is pinned by an exhaustive transition test matrix.

‎packages/agents-runtime/package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,7 @@
125125
"pino-pretty": "^13.0.0",
126126
"turndown": "^7.2.2",
127127
"turndown-plugin-gfm": "^1.0.2",
128+
"xstate": "^5.32.0",
128129
"zod": "^4.3.6",
129130
"zod-to-json-schema": "^3.25.2"
130131
},

‎packages/agents-runtime/src/pull-wake-machine.ts‎

Lines changed: 222 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,222 @@
1+
import { assign, fromCallback, fromPromise, setup } from 'xstate'
2+
import type { PullWakeEvent, PullWakeStreamResponse } from './pull-wake-runner'
3+
4+
export const INITIAL_RECONNECT_BACKOFF_MS = 1_000
5+
export const MAX_RECONNECT_BACKOFF_MS = 30_000
6+
7+
/**
8+
* Side effects the lifecycle machine triggers but does not own. Diagnostics,
9+
* heartbeating, and claim processing live in the runner closure; the machine
10+
* decides *when* they happen so every (state, event) pair is explicit.
11+
*/
12+
export interface PullWakeMachineEffects {
13+
/** Open the wake stream. The signal is aborted whenever `connecting` exits. */
14+
connectStream: (signal: AbortSignal) => Promise<PullWakeStreamResponse>
15+
onStreamConnected: () => void
16+
onStreamDisconnected: () => void
17+
onWake: (event: PullWakeEvent) => void
18+
onOffset: (offset: string) => void
19+
/** A connect attempt or stream session ended in error. */
20+
onReconnectError: (error: unknown) => void
21+
notifyHeartbeatChange: () => void
22+
cancelResponse: (response: PullWakeStreamResponse, reason: Error) => void
23+
/** Entered `stopping`: abort in-flight work and stop heartbeats. */
24+
onStopping: () => void
25+
/** Drain sequence: wait for claim actors, abort wakes, drain wakes. */
26+
shutdown: () => Promise<void>
27+
}
28+
29+
export interface PullWakeMachineContext {
30+
backoffMs: number
31+
response: PullWakeStreamResponse | null
32+
streamResetError: Error | null
33+
drainError: unknown
34+
}
35+
36+
export type PullWakeMachineEvent =
37+
| { type: `START` }
38+
| { type: `STOP` }
39+
| { type: `STREAM_RESET`; error: Error }
40+
| { type: `WAKE`; event: PullWakeEvent }
41+
| { type: `OFFSET`; offset: string }
42+
| { type: `STREAM_END`; error?: unknown }
43+
44+
export function createPullWakeMachine(effects: PullWakeMachineEffects) {
45+
return setup({
46+
types: {
47+
context: {} as PullWakeMachineContext,
48+
events: {} as PullWakeMachineEvent,
49+
},
50+
actors: {
51+
connectStream: fromPromise<PullWakeStreamResponse>(({ signal }) =>
52+
effects.connectStream(signal)
53+
),
54+
consumeStream: fromCallback<
55+
PullWakeMachineEvent,
56+
{ response: PullWakeStreamResponse }
57+
>(({ sendBack, input }) => {
58+
const { response } = input
59+
let stopped = false
60+
void (async () => {
61+
try {
62+
for await (const event of response.jsonStream()) {
63+
if (stopped) return
64+
if (event?.type === `wake`) {
65+
sendBack({ type: `WAKE`, event })
66+
}
67+
if (response.offset !== undefined) {
68+
sendBack({ type: `OFFSET`, offset: response.offset })
69+
}
70+
}
71+
await response.closed
72+
if (!stopped) sendBack({ type: `STREAM_END` })
73+
} catch (error) {
74+
if (!stopped) sendBack({ type: `STREAM_END`, error })
75+
}
76+
})()
77+
return () => {
78+
stopped = true
79+
}
80+
}),
81+
shutdown: fromPromise(() => effects.shutdown()),
82+
},
83+
delays: {
84+
reconnectBackoff: ({ context }) => context.backoffMs,
85+
},
86+
}).createMachine({
87+
id: `pullWakeRunner`,
88+
context: {
89+
backoffMs: INITIAL_RECONNECT_BACKOFF_MS,
90+
response: null,
91+
streamResetError: null,
92+
drainError: null,
93+
},
94+
initial: `stopped`,
95+
states: {
96+
stopped: {
97+
on: {
98+
START: {
99+
target: `running`,
100+
actions: assign({
101+
backoffMs: INITIAL_RECONNECT_BACKOFF_MS,
102+
response: null,
103+
streamResetError: null,
104+
drainError: null,
105+
}),
106+
},
107+
},
108+
},
109+
running: {
110+
initial: `connecting`,
111+
on: {
112+
STOP: { target: `stopping` },
113+
},
114+
states: {
115+
connecting: {
116+
entry: [
117+
assign({ streamResetError: null }),
118+
() => effects.notifyHeartbeatChange(),
119+
],
120+
invoke: {
121+
src: `connectStream`,
122+
onDone: {
123+
target: `streaming`,
124+
actions: assign({
125+
response: ({ event }) => event.output,
126+
backoffMs: INITIAL_RECONNECT_BACKOFF_MS,
127+
}),
128+
},
129+
onError: {
130+
target: `reconnecting`,
131+
actions: ({ event }) => effects.onReconnectError(event.error),
132+
},
133+
},
134+
on: {
135+
// Leaving `connecting` aborts the in-flight connect attempt —
136+
// the invoked promise actor's signal is cancelled by xstate.
137+
STREAM_RESET: {
138+
target: `reconnecting`,
139+
actions: ({ event }) => effects.onReconnectError(event.error),
140+
},
141+
},
142+
},
143+
streaming: {
144+
entry: () => effects.onStreamConnected(),
145+
exit: [
146+
({ context }) => {
147+
if (context.response) {
148+
effects.cancelResponse(
149+
context.response,
150+
context.streamResetError ??
151+
new Error(`pull wake runner stopped`)
152+
)
153+
}
154+
},
155+
() => effects.onStreamDisconnected(),
156+
assign({ response: null }),
157+
],
158+
invoke: {
159+
src: `consumeStream`,
160+
input: ({ context }) => ({ response: context.response! }),
161+
},
162+
on: {
163+
WAKE: { actions: ({ event }) => effects.onWake(event.event) },
164+
OFFSET: {
165+
actions: ({ event }) => effects.onOffset(event.offset),
166+
},
167+
STREAM_RESET: {
168+
guard: ({ context }) => !context.streamResetError,
169+
// Cancel the stream; the consume actor then observes the end
170+
// of iteration and emits STREAM_END, which routes to the
171+
// error path below via context.streamResetError.
172+
actions: [
173+
assign({ streamResetError: ({ event }) => event.error }),
174+
({ context, event }) => {
175+
if (context.response) {
176+
effects.cancelResponse(context.response, event.error)
177+
}
178+
},
179+
],
180+
},
181+
STREAM_END: [
182+
{
183+
guard: ({ context, event }) =>
184+
Boolean(event.error ?? context.streamResetError),
185+
target: `reconnecting`,
186+
actions: ({ context, event }) =>
187+
effects.onReconnectError(
188+
event.error ?? context.streamResetError
189+
),
190+
},
191+
{ target: `reconnecting` },
192+
],
193+
},
194+
},
195+
reconnecting: {
196+
entry: () => effects.notifyHeartbeatChange(),
197+
after: {
198+
reconnectBackoff: {
199+
target: `connecting`,
200+
actions: assign({
201+
backoffMs: ({ context }) =>
202+
Math.min(context.backoffMs * 2, MAX_RECONNECT_BACKOFF_MS),
203+
}),
204+
},
205+
},
206+
},
207+
},
208+
},
209+
stopping: {
210+
entry: () => effects.onStopping(),
211+
invoke: {
212+
src: `shutdown`,
213+
onDone: { target: `stopped` },
214+
onError: {
215+
target: `stopped`,
216+
actions: assign({ drainError: ({ event }) => event.error }),
217+
},
218+
},
219+
},
220+
},
221+
})
222+
}

0 commit comments

Comments
 (0)