feat(agent): add AsyncIterable pipe runtime with exit-code short-circuit

Execute parsed Node trees. Pipe stages chain stdout -> stdin as
AsyncIterables so downstream commands start consuming before upstream
finishes. && / || short-circuit on exit code. Redirect drains final
stdout to the VFS. Unknown command -> 127, throwing command -> 1,
aborted signal -> 130.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: snomiao

src/agent/shell/runtime.test.ts

@@ -0,0 +1,151 @@
+import { describe, expect, it } from 'vitest'
+
+import { CommandRegistryImpl, runScript } from './runtime'
+import type { ExecContext } from './runtime'
+import { collect, emptyIter, lines, stringIter } from './types'
+import type { Command } from './types'
+import { MemoryVFS } from './vfs/memory'
+
+function setup(): ExecContext & { registry: CommandRegistryImpl } {
+  const registry = new CommandRegistryImpl()
+  const echo: Command = async (ctx) => ({
+    stdout: stringIter(ctx.argv.slice(1).join(' ') + '\n'),
+    exitCode: 0
+  })
+  const cat: Command = async (ctx) => ({ stdout: ctx.stdin, exitCode: 0 })
+  const grep: Command = async (ctx) => {
+    const re = new RegExp(ctx.argv[1])
+    async function* gen(): AsyncIterable<string> {
+      for await (const l of lines(ctx.stdin)) {
+        if (re.test(l)) yield l + '\n'
+      }
+    }
+    return { stdout: gen(), exitCode: 0 }
+  }
+  const fail: Command = async () => ({ stdout: emptyIter(), exitCode: 2 })
+  const count: Command = async (ctx) => {
+    let n = 0
+    for await (const _l of lines(ctx.stdin)) n++
+    return { stdout: stringIter(String(n) + '\n'), exitCode: 0 }
+  }
+  const boom: Command = async () => {
+    throw new Error('kaboom')
+  }
+  registry.register('echo', echo)
+  registry.register('cat', cat)
+  registry.register('grep', grep)
+  registry.register('fail', fail)
+  registry.register('count', count)
+  registry.register('boom', boom)
+  return {
+    registry,
+    vfs: new MemoryVFS(),
+    env: new Map(),
+    cwd: '/'
+  }
+}
+
+describe('runScript', () => {
+  it('runs simple command', async () => {
+    const ctx = setup()
+    const r = await runScript('echo hi', ctx)
+    expect(r.exitCode).toBe(0)
+    expect(await collect(r.stdout)).toBe('hi\n')
+  })
+
+  it('pipes through stages', async () => {
+    const ctx = setup()
+    const r = await runScript('echo a | cat | cat', ctx)
+    expect(await collect(r.stdout)).toBe('a\n')
+  })
+
+  it('grep filters piped input', async () => {
+    const ctx = setup()
+    const r = await runScript('echo foo | grep oo', ctx)
+    expect(await collect(r.stdout)).toBe('foo\n')
+    const r2 = await runScript('echo bar | grep oo', ctx)
+    expect(await collect(r2.stdout)).toBe('')
+  })
+
+  it('&& short-circuits on failure', async () => {
+    const ctx = setup()
+    const r = await runScript('fail && echo nope', ctx)
+    expect(r.exitCode).toBe(2)
+    expect(await collect(r.stdout)).toBe('')
+  })
+
+  it('&& runs right on success', async () => {
+    const ctx = setup()
+    const r = await runScript('echo a && echo b', ctx)
+    expect(r.exitCode).toBe(0)
+    expect(await collect(r.stdout)).toBe('a\nb\n')
+  })
+
+  it('|| runs right on failure', async () => {
+    const ctx = setup()
+    const r = await runScript('fail || echo recover', ctx)
+    expect(r.exitCode).toBe(0)
+    expect(await collect(r.stdout)).toContain('recover')
+  })
+
+  it('redirect > writes stdout to vfs', async () => {
+    const ctx = setup()
+    const r = await runScript('echo hello > /out.txt', ctx)
+    expect(r.exitCode).toBe(0)
+    expect(await collect(r.stdout)).toBe('')
+    expect(await ctx.vfs.read('/out.txt')).toBe('hello\n')
+  })
+
+  it('redirect >> appends', async () => {
+    const ctx = setup()
+    await runScript('echo a >> /log', ctx)
+    await runScript('echo b >> /log', ctx)
+    expect(await ctx.vfs.read('/log')).toBe('a\nb\n')
+  })
+
+  it('pipe redirect writes final stage output', async () => {
+    const ctx = setup()
+    await runScript('echo foo | cat > /p.txt', ctx)
+    expect(await ctx.vfs.read('/p.txt')).toBe('foo\n')
+  })
+
+  it('unknown command returns 127', async () => {
+    const ctx = setup()
+    const r = await runScript('notreal', ctx)
+    expect(r.exitCode).toBe(127)
+    expect(r.stderr).toContain('not found')
+  })
+
+  it('throwing command returns 1', async () => {
+    const ctx = setup()
+    const r = await runScript('boom', ctx)
+    expect(r.exitCode).toBe(1)
+    expect(r.stderr).toContain('kaboom')
+  })
+
+  it('pre-aborted signal returns 130', async () => {
+    const ctx = setup()
+    const ac = new AbortController()
+    ac.abort()
+    const r = await runScript('echo hi', { ...ctx, signal: ac.signal })
+    expect(r.exitCode).toBe(130)
+  })
+
+  it('seq runs both sides', async () => {
+    const ctx = setup()
+    const r = await runScript('echo a ; echo b', ctx)
+    expect(await collect(r.stdout)).toBe('a\nb\n')
+  })
+
+  it('count consumes piped lines', async () => {
+    const ctx = setup()
+    const r = await runScript('echo a | count', ctx)
+    expect(await collect(r.stdout)).toBe('1\n')
+  })
+
+  it('parse error returns exit 2', async () => {
+    const ctx = setup()
+    const r = await runScript('echo $(ls)', ctx)
+    expect(r.exitCode).toBe(2)
+  })
+})

src/agent/shell/runtime.ts

@@ -0,0 +1,171 @@
+import type {
+  Cmd,
+  CmdContext,
+  CmdResult,
+  Command,
+  CommandRegistry,
+  Node,
+  Redirect,
+  VFS
+} from './types'
+import { collect, emptyIter } from './types'
+import { parseScript } from './parser'
+
+export class CommandRegistryImpl implements CommandRegistry {
+  private map = new Map<string, Command>()
+
+  get(name: string): Command | undefined {
+    return this.map.get(name)
+  }
+
+  register(name: string, cmd: Command): void {
+    this.map.set(name, cmd)
+  }
+
+  list(): string[] {
+    return [...this.map.keys()].sort()
+  }
+}
+
+export interface ExecContext {
+  registry: CommandRegistry
+  vfs: VFS
+  env: Map<string, string>
+  cwd: string
+  signal?: AbortSignal
+  stdin?: AsyncIterable<string>
+}
+
+function makeCtx(
+  ctx: ExecContext,
+  argv: string[],
+  stdin: AsyncIterable<string>
+): CmdContext {
+  return {
+    argv,
+    stdin,
+    env: ctx.env,
+    cwd: ctx.cwd,
+    vfs: ctx.vfs,
+    signal: ctx.signal ?? new AbortController().signal
+  }
+}
+
+async function applyRedirect(
+  res: CmdResult,
+  redirect: Redirect,
+  vfs: VFS
+): Promise<CmdResult> {
+  const data = await collect(res.stdout)
+  if (redirect.op === '>') await vfs.write(redirect.path, data)
+  else await vfs.append(redirect.path, data)
+  return { stdout: emptyIter(), exitCode: res.exitCode, stderr: res.stderr }
+}
+
+async function runSimple(
+  cmd: Cmd,
+  ctx: ExecContext,
+  stdin: AsyncIterable<string>
+): Promise<CmdResult> {
+  if (ctx.signal?.aborted) {
+    return { stdout: emptyIter(), exitCode: 130, stderr: 'aborted' }
+  }
+  const name = cmd.argv[0]
+  const handler = ctx.registry.get(name)
+  if (!handler) {
+    return {
+      stdout: emptyIter(),
+      exitCode: 127,
+      stderr: `${name}: command not found`
+    }
+  }
+  let res: CmdResult
+  try {
+    res = await handler(makeCtx(ctx, cmd.argv, stdin))
+  } catch (err) {
+    return {
+      stdout: emptyIter(),
+      exitCode: 1,
+      stderr: err instanceof Error ? err.message : String(err)
+    }
+  }
+  if (cmd.redirect) res = await applyRedirect(res, cmd.redirect, ctx.vfs)
+  return res
+}
+
+async function runPipe(
+  cmds: Cmd[],
+  ctx: ExecContext,
+  stdin: AsyncIterable<string>,
+  redirect: Redirect | undefined
+): Promise<CmdResult> {
+  let cur = stdin
+  let exit = 0
+  let stderr: string | undefined
+  for (let i = 0; i < cmds.length; i++) {
+    const last = i === cmds.length - 1
+    const cmd = cmds[i]
+    const inner = last ? cmd : { ...cmd, redirect: undefined }
+    const res = await runSimple(inner, ctx, cur)
+    cur = res.stdout
+    exit = res.exitCode
+    if (res.stderr) stderr = res.stderr
+    if (ctx.signal?.aborted) {
+      return { stdout: emptyIter(), exitCode: 130, stderr: 'aborted' }
+    }
+  }
+  let result: CmdResult = { stdout: cur, exitCode: exit, stderr }
+  if (redirect) result = await applyRedirect(result, redirect, ctx.vfs)
+  return result
+}
+
+async function runNode(node: Node, ctx: ExecContext): Promise<CmdResult> {
+  const stdin = ctx.stdin ?? emptyIter()
+  if (ctx.signal?.aborted) {
+    return { stdout: emptyIter(), exitCode: 130, stderr: 'aborted' }
+  }
+  if (node.type === 'simple') return runSimple(node.cmd, ctx, stdin)
+  if (node.type === 'pipe') return runPipe(node.cmds, ctx, stdin, node.redirect)
+
+  const left = await runNode(node.left, ctx)
+  const leftOut = await collect(left.stdout)
+  if (node.type === 'and' && left.exitCode !== 0) {
+    return {
+      stdout: toIter(leftOut),
+      exitCode: left.exitCode,
+      stderr: left.stderr
+    }
+  }
+  if (node.type === 'or' && left.exitCode === 0) {
+    return { stdout: toIter(leftOut), exitCode: 0, stderr: left.stderr }
+  }
+  const right = await runNode(node.right, ctx)
+  const rightOut = await collect(right.stdout)
+  const combined = leftOut + rightOut
+  return {
+    stdout: toIter(combined),
+    exitCode: right.exitCode,
+    stderr: right.stderr ?? left.stderr
+  }
+}
+
+async function* toIter(s: string): AsyncIterable<string> {
+  if (s.length > 0) yield s
+}
+
+export async function runScript(
+  src: string,
+  ctx: ExecContext
+): Promise<CmdResult> {
+  let node: Node
+  try {
+    node = parseScript(src, Object.fromEntries(ctx.env))
+  } catch (err) {
+    return {
+      stdout: emptyIter(),
+      exitCode: 2,
+      stderr: err instanceof Error ? err.message : String(err)
+    }
+  }
+  return runNode(node, ctx)
+}

src/agent/shell/vfs/memory.ts

@@ -12,17 +12,6 @@ function normalize(path: string): string {
   return '/' + stack.join('/')
 }
 
-function parentOf(path: string): string {
-  const idx = path.lastIndexOf('/')
-  if (idx <= 0) return '/'
-  return path.slice(0, idx)
-}
-
-function nameOf(path: string): string {
-  const idx = path.lastIndexOf('/')
-  return path.slice(idx + 1)
-}
-
 export class MemoryVFS implements VFS {
   private files = new Map<string, string>()
 
@@ -95,5 +84,3 @@ export class MemoryVFS implements VFS {
     return false
   }
 }
-
-export { normalize, parentOf, nameOf }