ext-agent

1import type { EngineInterface, Register, StreamHook, TurnStepChunk } from 'claude-code'
2
3/**
4 * ext-agent: adds the Agent tool's `ext-agent:worker` type, whose turns run
5 * pi or opencode instead of a Claude model. The built-in agent types stay
6 * available unless the `blockBuiltin` option is on, in which case they are
7 * hidden from the model and refused with a message saying how to dispatch
8 * instead.
9 *
10 * Each tool call pi / opencode makes becomes one of the worker's own tool
11 * calls (Bash, Read, Write, Edit), answered here with what the CLI already
12 * got, so the subagent view groups them as it does any agent's ("Ran 3
13 * commands"). What the transcript keeps of each result is clipped, and a long
14 * worker's transcript is compacted here without asking any model.
15 */
16
17const TYPE = 'ext-agent:worker'
18const HEADER = /^[ \t]*(model|cwd|worktree)[ \t]*:[ \t]*(\S.*?)[ \t]*$/i
19/** How much of a tool's output the worker's transcript keeps. */
20const KEPT = 1500
21/** Messages of a worker's transcript kept when it is compacted. */
22const KEEP_MESSAGES = 80
23
24type Config = { blockBuiltin: boolean; defaultModel: string; worktreeLayout: 'claude' | 'sibling' }
25let config: Config = { blockBuiltin: false, defaultModel: 'pi', worktreeLayout: 'claude' }
26
27const usage = (why: string) =>
28  `${why}\n\n` +
29  `Dispatch external agents with Agent(subagent_type: "${TYPE}"); the prompt may open with:\n` +
30  `  model: pi[:<provider>/<model>[:off|minimal|low|medium|high|xhigh]]   (pi's own default when no model is named)\n` +
31  `  model: opencode[:<provider/model[#variant]>]                      (\`opencode models\` lists them)\n` +
32  `  cwd: <absolute path>                                              (default: the session's directory)\n` +
33  `  worktree: <name>   (or isolation: "worktree") runs in a git worktree of that name, made or reused\n` +
34  `The default model is "${config.defaultModel}". Give the Agent a \`name\` to continue its session later.` +
35  (config.blockBuiltin ? '\nBuilt-in subagent types are disabled by this plugin\'s blockBuiltin option.' : '')
36
37const isWorker = (type: string) => type === TYPE
38
39const clip = (text: string, max = KEPT) =>
40  text.length <= max ? text : `${text.slice(0, max)}\n… (${text.length - max} more characters kept by the CLI only)`
41
42type Header = { model?: string; cwd?: string; worktree?: string; body: string }
43
44/** Takes the `model:` / `cwd:` / `worktree:` lines off the top of a prompt. */
45function headerOf(text: string): Header {
46  const lines = text.replace(/^\s*\n/, '').split('\n')
47  const out: Header = { body: '' }
48  while (lines.length > 0) {
49    const m = lines[0].match(HEADER)
50    if (!m) break
51    out[m[1].toLowerCase() as 'model' | 'cwd' | 'worktree'] = m[2]
52    lines.shift()
53  }
54  out.body = lines.join('\n').trim()
55  return out
56}
57
58/** `model` is absent when the CLI is left to pick its own default. */
59type Spec = { cli: 'pi' | 'opencode'; model?: string }
60
61let opencodeModels: Promise<Set<string>> | undefined
62
63/** `pi`, `pi:<model>`, `opencode` or `opencode:<model>`, checked. */
64async function specOf($: EngineInterface, model = config.defaultModel): Promise<Spec | { error: string }> {
65  const m = model.trim().match(/^(pi|opencode)(?::(\S+))?$/)
66  if (!m) return { error: `Model "${model}" is not "pi[:<model>]" or "opencode[:<model>]".` }
67  const [, cli, name] = m
68  if (cli === 'pi') {
69    if (name !== undefined && !/^[\w.\-]+\/[\w.\-:/]+$/.test(name) && !/^[\w.\-]+$/.test(name)) return { error: `Unusable pi model "${name}".` }
70    return { cli: 'pi', model: name }
71  }
72  if (name === undefined) return { cli: 'opencode' }
73  opencodeModels ??= $.process
74    .run(['opencode', 'models'])
75    .then(({ stdout }) => new Set(stdout.split('\n').map((l) => l.trim()).filter(Boolean)))
76  const known = await opencodeModels
77  if (known.has(name.split('#')[0])) return { cli: 'opencode', model: name }
78  opencodeModels = undefined
79  return { error: `Unknown opencode model "${name}".` }
80}
81
82/**
83 * argv run from `dir` in a process group of its own, which the shell takes
84 * down whole when the engine ends it (TaskStop, an interrupt), so no tool the
85 * CLI started outlives the worker.
86 */
87const supervised = (dir: string, argv: string[]) => [
88  '/bin/sh', '-c',
89  'cd "$1" && shift; set -m; "$@" & pid=$!; trap "kill -TERM -$pid 2>/dev/null; exit 143" TERM INT HUP; wait $pid',
90  'sh', dir, ...argv,
91]
92
93/**
94 * The session a worker's earlier turn left, by its title. Listed through a
95 * standalone server as the run itself is: the background service can hold a
96 * stale idea of which project a directory is, and would not find it.
97 */
98async function opencodeSession($: EngineInterface, cwd: string, title: string) {
99  const { stdout } = await $.process.run([
100    '/bin/sh', '-c', 'cd "$1" && exec opencode session list --standalone --format json', 'sh', cwd,
101  ])
102  try {
103    const sessions = JSON.parse(stdout) as { id: string; title?: string }[]
104    return sessions.find((s) => s.title === title)?.id
105  } catch {
106    return undefined
107  }
108}
109
110/**
111 * A worktree for a `worktree:` line. Laid out as Claude Code's own are (the
112 * default): `<repo>/.claude/worktrees/<name>` on branch `worktree-<name>`, off
113 * what the `worktree.baseRef` setting says (`fresh`, the default: the remote's
114 * default branch; `head`: the current HEAD). With the `sibling` layout:
115 * `<repo>-wt/<name>` beside the repository on `wip/<name>`, off `main`.
116 * An existing worktree or branch of that name is reused.
117 */
118async function worktreeOf($: EngineInterface, cwd: string, label: string) {
119  const git = (...argv: string[]) => $.process.run(['git', '-C', cwd, ...argv])
120  const top = await git('rev-parse', '--show-toplevel')
121  if (top.exitCode !== 0) return { error: `${cwd} is not in a git repository` }
122  const repo = top.stdout.trim()
123  const sibling = config.worktreeLayout === 'sibling'
124  const slug = label.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-|-$/g, '').slice(0, 40) || 'worker'
125  const path = sibling ? `${repo}-wt/${slug}` : `${repo}/.claude/worktrees/${slug}`
126  const branch = sibling ? `wip/${slug}` : `worktree-${slug}`
127  // Forget worktrees whose directories were deleted, so their paths and
128  // branches can be used again.
129  await git('worktree', 'prune')
130  const listed = await git('worktree', 'list', '--porcelain')
131  if (listed.stdout.split('\n').includes(`worktree ${path}`)) return { path }
132  const exists = await git('rev-parse', '--verify', '-q', `refs/heads/${branch}`)
133  let argv: string[]
134  if (exists.exitCode === 0) argv = [path, branch]
135  else {
136    let base: string | undefined
137    if (sibling) {
138      if ((await git('rev-parse', '--verify', '-q', 'main')).exitCode === 0) base = 'main'
139    } else if (((await $.settings.read()) as { worktree?: { baseRef?: string } }).worktree?.baseRef !== 'head') {
140      const remote = await git('symbolic-ref', '--short', 'refs/remotes/origin/HEAD')
141      if (remote.exitCode === 0) base = remote.stdout.trim()
142    }
143    argv = ['-b', branch, path, ...(base ? [base] : [])]
144  }
145  const added = await git('worktree', 'add', '-q', ...argv)
146  if (added.exitCode !== 0) return { error: added.stderr.trim() || `git worktree add exited ${added.exitCode}` }
147  return { path }
148}
149
150/**
151 * An Agent call's arguments with a `model` the schema would refuse
152 * (`pi…`, `opencode…`) moved into the prompt's `model:` line, so the
153 * engine's validation sees a call it accepts. Unparseable JSON passes as is.
154 */
155function movedModel(json: string) {
156  let input: Record<string, unknown>
157  try { input = JSON.parse(json) } catch { return json }
158  const model = input.model
159  if (typeof model !== 'string' || !/^(pi|opencode)(:|$)/.test(model)) return json
160  const prompt = typeof input.prompt === 'string' ? input.prompt : ''
161  const rest = prompt.replace(/^\s*model[ \t]*:.*(\n|$)/i, '')
162  const { model: _, ...kept } = input
163  return JSON.stringify({ ...kept, prompt: `model: ${model}\n${rest}` })
164}
165
166// ---------------------------------------------------------------------------
167// A run: one CLI process answering one turn of a worker, read piece by piece
168// by whichever dispatch is waiting on it (a step, or a tool call's result).
169
170type Call = { name: 'Bash' | 'Read' | 'Write' | 'Edit'; input: Record<string, unknown> }
171type Event = { kind: 'show'; text: string } | ({ kind: 'tool'; id: string } & Call)
172type Outcome = { output: string; isError: boolean }
173
174type Run = {
175  spec: Spec
176  stream: AsyncGenerator<{ stream: 'stdout' | 'stderr'; text: string }, { code: number | null; signal: string | null }>
177  events: Event[]
178  calls: Map<string, Call>
179  outcomes: Map<string, Outcome>
180  /** The CLI's own id for each call, to ours. */
181  ids: Map<string, string>
182  buffer: string
183  stderr: string
184  error: string
185  answered: string
186  answer: string
187  current: string
188  texts: Map<string, string>
189  lastMessage: string
190  ended?: string
191}
192
193const runs = new Map<string, Run>()
194const ours = new Set<string>()
195/** The header each worker was spawned with, past the 4096 messages a read returns. */
196const spawned = new Map<string, Header>()
197let serial = 0
198
199/** One of the worker's own tool calls standing for a CLI's. */
200function callOf(tool: string, args: Record<string, any> = {}): Call {
201  const path = String(args.path ?? args.filePath ?? args.file_path ?? '')
202  switch (tool.toLowerCase()) {
203    case 'bash':
204    case 'shell':
205      return { name: 'Bash', input: { command: String(args.command ?? '') } }
206    case 'read':
207      return { name: 'Read', input: { file_path: path, ...(args.offset ? { offset: args.offset } : {}), ...(args.limit ? { limit: args.limit } : {}) } }
208    case 'write':
209      return { name: 'Write', input: { file_path: path, content: clip(String(args.content ?? ''), 400) } }
210    case 'edit': {
211      const first = Array.isArray(args.edits) ? args.edits[0] ?? {} : args
212      return {
213        name: 'Edit',
214        input: {
215          file_path: path,
216          old_string: clip(String(first.oldText ?? first.oldString ?? ''), 400),
217          new_string: clip(String(first.newText ?? first.newString ?? ''), 400),
218        },
219      }
220    }
221    default:
222      return { name: 'Bash', input: { command: clip(`${tool} ${JSON.stringify(args)}`, 400) } }
223  }
224}
225
226/** A CLI's tool output as the worker's own tool records a result. */
227function resultOf(call: Call, output: string): unknown {
228  const kept = clip(output)
229  const file_path = String(call.input.file_path ?? '')
230  switch (call.name) {
231    case 'Bash':
232      return { stdout: kept, stderr: '', interrupted: false }
233    case 'Read': {
234      const numLines = kept.split('\n').length
235      return { type: 'text', file: { filePath: file_path, content: kept, numLines, startLine: Number(call.input.offset ?? 1), totalLines: numLines } }
236    }
237    case 'Write':
238      return { type: 'create', filePath: file_path, content: '', structuredPatch: [], originalFile: null }
239    case 'Edit':
240      return {
241        filePath: file_path, oldString: call.input.old_string, newString: call.input.new_string,
242        originalFile: null, structuredPatch: [], userModified: false, replaceAll: false,
243      }
244  }
245}
246
247const textOf = (content: unknown) =>
248  Array.isArray(content) ? content.map((c: any) => (typeof c?.text === 'string' ? c.text : '')).join('') : String(content ?? '')
249
250function called(run: Run, cliId: string, tool: string, args: Record<string, any>) {
251  const id = `toolu_ext_${Date.now().toString(36)}_${(serial++).toString(36)}`
252  const call = callOf(tool, args)
253  ours.add(id)
254  run.ids.set(cliId, id)
255  run.calls.set(id, call)
256  run.events.push({ kind: 'tool', id, ...call })
257  return id
258}
259
260/** Reads the CLI's next piece of output into the run's events and outcomes. */
261async function pull(run: Run) {
262  const piece = await run.stream.next()
263  if (piece.done) {
264    const { code, signal } = piece.value
265    if (run.spec.cli === 'pi' && run.current.trim()) run.answer = run.current
266    if (run.spec.cli === 'opencode') run.answer = run.texts.get(run.lastMessage) ?? ''
267    run.ended = `exit ${code ?? signal}`
268    return
269  }
270  if (piece.value.stream === 'stderr') {
271    run.stderr = (run.stderr + piece.value.text).slice(-4000)
272    return
273  }
274  run.buffer += piece.value.text
275  const lines = run.buffer.split('\n')
276  run.buffer = lines.pop() ?? ''
277  for (const line of lines) {
278    let ev: any
279    try { ev = JSON.parse(line) } catch { continue }
280    const show = (text: string) => text && run.events.push({ kind: 'show', text })
281    if (run.spec.cli === 'pi') {
282      const d = ev.type === 'message_update' ? ev.assistantMessageEvent : undefined
283      if (ev.type === 'message_start' && ev.message?.role === 'assistant') run.current = ''
284      if (d?.type === 'text_delta') { run.current += d.delta ?? ''; show(d.delta ?? '') }
285      if (ev.type === 'tool_execution_start') called(run, ev.toolCallId, ev.toolName, ev.args)
286      if (ev.type === 'tool_execution_end') {
287        const id = run.ids.get(ev.toolCallId)
288        if (id) run.outcomes.set(id, { output: textOf(ev.result?.content), isError: ev.isError === true })
289      }
290      if (ev.type === 'turn_end') {
291        const m = ev.message ?? {}
292        if (m.provider && m.model) run.answered = `${m.provider}/${m.model}`
293        if (m.errorMessage) { run.error = m.errorMessage; show(`\n✗ ${m.errorMessage}\n`) }
294        if (run.current.trim()) run.answer = run.current
295        run.current = ''
296      }
297    } else {
298      const p = ev.part ?? {}
299      if (ev.type === 'tool_use') {
300        const id = called(run, p.callID ?? p.id ?? String(serial), p.tool ?? 'tool', p.state?.input ?? {})
301        const failed = p.state?.status === 'error'
302        run.outcomes.set(id, { output: String((failed ? p.state?.error : p.state?.output) ?? ''), isError: failed })
303      }
304      if (ev.type === 'text' && typeof p.text === 'string') {
305        run.lastMessage = p.messageID
306        run.texts.set(p.messageID, (run.texts.get(p.messageID) ?? '') + p.text)
307        show(`${p.text}\n`)
308      }
309      if (ev.type === 'error') { run.error = JSON.stringify(ev.error ?? ev).slice(0, 500); show(`✗ ${run.error}\n`) }
310    }
311  }
312}
313
314function stop(agentId: string) {
315  const run = runs.get(agentId)
316  runs.delete(agentId)
317  void run?.stream.return(undefined as never).catch(() => {})
318}
319
320/** Why a run ended with no answer, short: the error, else stderr's last lines. */
321function failureOf(run: Run) {
322  const stderr = run.stderr
323    .split('\n')
324    .filter((l) => l.trim() && !/^\[\d+\][+-]?\s+(Done|Exit|Terminated|Killed)/.test(l))
325    .slice(-6)
326    .join('\n')
327  return `ext-agent: ${run.spec.cli} (${run.spec.model ?? 'default model'}) ended without an answer (${run.ended}).` +
328    (run.error ? `\n${run.error}` : '') + (stderr ? `\n${stderr}` : '')
329}
330
331// ---------------------------------------------------------------------------
332
333export const register: Register = (on, options) => {
334  config = {
335    blockBuiltin: options.blockBuiltin === true,
336    defaultModel: String(options.defaultModel || 'pi'),
337    worktreeLayout: options.worktreeLayout === 'sibling' ? 'sibling' : 'claude',
338  }
339
340  if (config.blockBuiltin) on('agent.offer', ($, e, next) => (e.agent === TYPE || e.provider.plugin.split('@')[0] === 'ext-agent' ? next(e) : { isOffered: false }))
341
342  on('tool.describe', { tool: 'Agent' }, async ($, e, next) => {
343    const described = await next(e)
344    return {
345      ...described,
346      description:
347        `${described.description}\n\n` +
348        `ext-agent: subagent_type "${TYPE}" runs the task in pi or opencode instead of a Claude model. Its \`model\` takes ` +
349        `"pi[:<provider>/<model>[:<thinking>]]" or "opencode[:<provider/model[#variant]>]" (\`opencode models\` lists them); ` +
350        `Claude model names do nothing for it. The default is "${config.defaultModel}".` +
351        (config.blockBuiltin ? ' It is the only subagent_type allowed.' : ''),
352    }
353  })
354
355  // The model's Agent calls, rewritten before the engine parses them: the
356  // arguments of each Agent block are held until the block ends, then passed
357  // on as one piece with a pi / opencode `model` moved into the prompt.
358  on('turn.step', async function* ($, e, next) {
359    if (e.agentId !== undefined) return yield* workerStep($, e, next)
360    const held = new Map<number, string>()
361    const stream = next(e)
362    const flush = function* () {
363      for (const [index, json] of held) yield { kind: 'input', index, json: movedModel(json) } as TurnStepChunk
364      held.clear()
365    }
366    for await (const chunk of stream) {
367      if (chunk.kind === 'tool' && chunk.name === 'Agent') held.set(chunk.index, '')
368      else if (chunk.kind === 'input' && held.has(chunk.index)) {
369        held.set(chunk.index, held.get(chunk.index) + chunk.json)
370        continue
371      } else yield* flush()
372      yield chunk
373    }
374    yield* flush()
375    const result = await stream.result
376    return {
377      ...result,
378      toolUses: result.toolUses.map((use) =>
379        use.name === 'Agent' ? { ...use, input: JSON.parse(movedModel(JSON.stringify(use.input))) } : use,
380      ),
381    }
382  })
383
384  on('tool.call', async ($, e, next) => {
385    if (!ours.has(e.tool_use_id)) return next(e)
386    ours.delete(e.tool_use_id)
387    const run = e.agentId === undefined ? undefined : runs.get(e.agentId)
388    if (run === undefined) return { deny: 'ext-agent: the worker that made this call is gone.' }
389    const onAbort = () => stop(e.agentId as string)
390    next.signal.addEventListener('abort', onAbort)
391    try {
392      while (!run.outcomes.has(e.tool_use_id) && run.ended === undefined) await pull(run)
393    } finally {
394      next.signal.removeEventListener('abort', onAbort)
395    }
396    const outcome = run.outcomes.get(e.tool_use_id)
397    const call = run.calls.get(e.tool_use_id)
398    if (outcome === undefined || call === undefined) return { deny: `${run.spec.cli} ended before this call finished.` }
399    if (outcome.isError) return { deny: clip(outcome.output || 'failed') }
400    return { result: resultOf(call, outcome.output) as never }
401  })
402
403  // A worker's transcript mirrors every step of pi / opencode, so a long task
404  // outgrows the context window and the CLI ends the agent with "Prompt is too
405  // long". No model is asked to summarise it: the hook drops the middle itself.
406  on('session.compact', async ($, e, next) => {
407    if (e.agentId === undefined) return next(e)
408    const agent = (await $.agent.list()).find((a) => a.id === e.agentId)
409    if (agent === undefined || !isWorker(agent.type)) return next(e)
410    const m = e.messages
411    // The tail starts on an assistant message, so a tool_use keeps its result.
412    let cut = Math.max(m.length - KEEP_MESSAGES, 1)
413    while (cut < m.length && m[cut].role !== 'assistant') cut++
414    if (cut <= 1 || cut >= m.length) return { skip: 'ext-agent: the worker transcript is already short.' }
415    const isTask = (x: (typeof m)[number]) => x.role === 'user' && x.text !== '' && !x.toolResults?.length
416    const task = m.findLastIndex(isTask)
417    const dropped = m.slice(1, cut).filter((_, i) => i + 1 !== task)
418    const calls = dropped.reduce((n, x) => n + x.toolUses.length, 0)
419    const note = {
420      role: 'assistant' as const,
421      toolUses: [],
422      text: `ext-agent: ${dropped.length} earlier messages (${calls} tool calls) were dropped from this transcript to keep it within the context window. The pi / opencode session still holds the full history.`,
423    }
424    return { messages: [m[0], ...(task > 0 && task < cut ? [m[task]] : []), note, ...m.slice(cut)] }
425  })
426
427  on('tool.call', { tool: 'Agent' }, async ($, e, next) => {
428    const type = e.subagent_type ?? 'general-purpose'
429    if (!isWorker(type)) return config.blockBuiltin ? { deny: usage(`subagent_type "${type}" is not allowed.`) } : next(e)
430    const header = headerOf(e.prompt)
431    const spec = await specOf($, header.model)
432    if ('error' in spec) return { deny: usage(spec.error) }
433    if (e.isolation !== 'worktree') return next(e)
434
435    // The engine's worktree is the subagent's cwd, which pi / opencode never
436    // see; the worker makes its own from a `worktree:` line instead. Only the
437    // prompt changes here, so a call that passes through again is the same.
438    const lines = [
439      `model: ${header.model ?? config.defaultModel}`,
440      ...(header.cwd ? [`cwd: ${header.cwd}`] : []),
441      `worktree: ${header.worktree ?? e.name ?? e.description}`,
442      '',
443      header.body,
444    ]
445    return next({ ...e, isolation: undefined, prompt: lines.join('\n') })
446  })
447}
448
449/** Starts the CLI for a worker's turn, or says why it cannot. */
450async function started($: EngineInterface, agentId: string, name: string): Promise<Run | string> {
451  // The newest user message is the task: the spawn's prompt, or a SendMessage.
452  // Its model, cwd and worktree are the spawn's unless it names its own.
453  const messages = await $.session.messages({ agentId })
454  if ('deny' in messages) return `ext-agent: cannot read this agent's prompt (${messages.deny}).`
455  const headers = messages
456    .filter((m) => m.role === 'user')
457    .map((m) => m.text.replace(/<system-reminder>[\s\S]*?<\/system-reminder>/g, '').trim())
458    .filter(Boolean)
459    .map(headerOf)
460  const first = spawned.get(agentId) ?? headers[0] ?? { body: '' }
461  spawned.set(agentId, first)
462  const header = headers.at(-1) ?? { body: '' }
463  header.model ??= first.model
464  header.cwd ??= first.cwd
465  header.worktree ??= first.worktree
466  const spec = await specOf($, header.model)
467  if ('error' in spec) return `ext-agent: ${usage(spec.error)}`
468  if (!header.body) return 'ext-agent: the prompt is empty.'
469
470  let cwd = header.cwd ?? (await $.session.cwd())
471  if ((await $.process.run(['test', '-d', cwd])).exitCode !== 0) return `ext-agent: cwd "${cwd}" does not exist.`
472  if (header.worktree) {
473    const made = await worktreeOf($, cwd, header.worktree)
474    if ('error' in made) return `ext-agent: worktree "${header.worktree}" failed: ${made.error}`
475    cwd = made.path
476  }
477  const key = `ext-${name.replace(/[^A-Za-z0-9_-]/g, '-')}`
478  let argv: string[]
479  if (spec.cli === 'pi') {
480    argv = ['pi', '-p', '--mode', 'json', ...(spec.model ? ['--model', spec.model] : []), '--session-id', key, header.body]
481  } else {
482    const session = await opencodeSession($, cwd, key)
483    argv = ['opencode', 'run', '--standalone', '--auto', '--format', 'json', ...(spec.model ? ['--model', spec.model] : []), '--title', key,
484      ...(session ? ['--session', session] : []), header.body]
485  }
486  const run: Run = {
487    spec, stream: $.process.spawn({ argv: supervised(cwd, argv) }),
488    events: [{ kind: 'show', text: `${spec.cli} ${spec.model ?? '(default model)'} in ${cwd}\n` }],
489    calls: new Map(), outcomes: new Map(), ids: new Map(),
490    buffer: '', stderr: '', error: '', answered: spec.model ?? spec.cli, answer: '', current: '', texts: new Map(), lastMessage: '',
491  }
492  runs.set(agentId, run)
493  return run
494}
495
496/**
497 * A worker's step: the CLI's output up to its next tool call (the step ends
498 * calling the matching tool) or to its end (the step ends with its answer).
499 */
500const workerStep: StreamHook<'turn.step'> = async function* ($, e, next) {
501  if (e.agentId === undefined) return yield* next(e)
502  const agentId = e.agentId
503  const agent = (await $.agent.list()).find((a) => a.id === agentId)
504  if (agent === undefined || !isWorker(agent.type)) return yield* next(e)
505
506  const finish = async function* (answer: string) {
507    yield { kind: 'text', index: 1, text: answer } as TurnStepChunk
508    yield { kind: 'stop', stopReason: 'end_turn', usage: null } as TurnStepChunk
509    return { turnId: e.turnId, index: e.index, answer, toolUses: [], stopReason: 'end_turn' as const, usage: null }
510  }
511
512  let run = runs.get(agentId)
513  if (e.index === 0 || run === undefined) {
514    stop(agentId)
515    const made = await started($, agentId, agent.name ?? agent.id)
516    if (typeof made === 'string') return yield* finish(made)
517    run = made
518  }
519
520  const onAbort = () => stop(agentId)
521  next.signal.addEventListener('abort', onAbort)
522  try {
523    for (;;) {
524      const ev = run.events.shift()
525      if (ev === undefined) {
526        if (run.ended !== undefined) break
527        await pull(run)
528        continue
529      }
530      if (ev.kind === 'show') {
531        yield { kind: 'thinking', index: 0, text: ev.text } as TurnStepChunk
532        continue
533      }
534      yield { kind: 'tool', index: 1, id: ev.id, name: ev.name } as TurnStepChunk
535      yield { kind: 'input', index: 1, json: JSON.stringify(ev.input) } as TurnStepChunk
536      yield { kind: 'stop', stopReason: 'tool_use', usage: null } as TurnStepChunk
537      return {
538        turnId: e.turnId, index: e.index, answer: '',
539        toolUses: [{ name: ev.name, input: ev.input }], stopReason: 'tool_use' as const, usage: null,
540      }
541    }
542  } finally {
543    next.signal.removeEventListener('abort', onAbort)
544  }
545
546  runs.delete(agentId)
547  if (run.answer.trim()) return yield* finish(`${run.answer.trim()}\n\n— answered by ${run.spec.cli}, ${run.answered}`)
548  return yield* finish(failureOf(run))
549}
550