Pipeshape Topologies

pipeshape-topologies

Pick the snippet whose condition matches, then fill in prompts, schemas, and labels. Compose shapes freely inside one script. Each snippet notes the v1 pipeline.yml construct it replaces so an existing pipeline can be ported by eye.

Runtime constraints

• Date.now(), Math.random(), and a no-argument new Date() throw. Pass timestamps through args.
• import() fails before the run starts. Put library work inside an agent.
• The script has no filesystem or shell access. Agents read, write, and run commands; the script coordinates them.
• Caps: 16 concurrent agents, 4,096 items per parallel() or pipeline() call, 1,000 agents per run.
• Every agent() can resolve to null. Filter with .filter(Boolean) before reading results.
• Default to pipeline(). Use parallel() only when the next stage needs every prior result at once.

Sequential

Condition: each step depends on the previous result.

const spec = await agent(`Analyze: ${args.requirement}`, { schema: SPEC })
const impl = await agent(`Implement this spec: ${JSON.stringify(spec)}`)
// v1: workflow: [analyst, implement]

Fan-out and fan-in

Condition: viewpoints or slices are independent and one consumer needs all of them together.

const parts = (await parallel(SLICES.map(s => () =>
  agent(`Audit ${s}`, { label: s, schema: FINDINGS })))).filter(Boolean)
const merged = await agent(`Synthesize: ${JSON.stringify(parts)}`)
// v1: parallel: [audit-api, audit-batch, audit-web] then synthesize (join: all)

Streaming stages

Condition: every item goes through the same stages and items do not depend on each other.

const results = await pipeline(files,
  f => agent(`Migrate ${f}`, { label: f, phase: 'Migrate' }),
  (r, f) => agent(`Verify the migration of ${f}: ${r}`, { label: `verify:${f}`, phase: 'Verify' }))
// v1: not expressible; node count was fixed at authoring time

Producer and reviewer loop

Condition: rework until a judge passes, with a hard ceiling.

let verdict = { pass: false, findings: [] }
for (let round = 0; round < 2 && !verdict.pass; round++) {
  const feedback = round ? `Fix only these findings: ${JSON.stringify(verdict.findings)}` : ''
  await agent(`Implement the spec. ${feedback}`, { label: `implement:${round}` })
  verdict = (await agent('Review the change against the spec.', { schema: VERDICT })) ?? verdict
}
if (!verdict.pass) log(`Escalating after 2 rounds: ${verdict.findings.length} findings remain`)
// v1: review: {if: FAILED, goto: implement, max: 2, exhausted: escalate}

Expert routing

Condition: the input kind decides which chain runs.

const { kind } = await agent(`Classify this request: ${args.request}`, {
  schema: { type: 'object', required: ['kind'], properties: { kind: { type: 'string', enum: ['bug', 'feature', 'docs'] } } },
})
const chains = {
  bug: () => agent('Reproduce, then fix.'),
  feature: () => agent('Design, then implement.'),
  docs: () => agent('Update the documentation.'),
}
const out = await chains[kind]()
// v1: branch: {on: kind, cases: {...}}

Adversarial verification

Condition: findings must survive an attempt to refute them before they are reported.

const confirmed = (await pipeline(findings,
  f => parallel(['correctness', 'reproduces'].map(lens => () =>
    agent(`Try to refute through the ${lens} lens. Default to refuted=true when unsure: ${f.summary}`,
      { label: `refute:${lens}`, phase: 'Verify', schema: REFUTE }))),
  (votes, f) => votes.filter(Boolean).every(v => !v.refuted) ? f : null)).filter(Boolean)
// v1: parallel: [[draft-a, refute-a], [draft-b, refute-b]] then synthesize

Judge panel

Condition: the solution space is wide and one iterated attempt would anchor early.

const drafts = (await parallel(['mvp-first', 'risk-first', 'user-first'].map(angle => () =>
  agent(`Draft a plan from the ${angle} angle.`, { label: angle, schema: PLAN })))).filter(Boolean)
const scored = await agent(`Score these plans and pick a winner: ${JSON.stringify(drafts)}`, { schema: SCORE })
// v1: parallel: [ideate-a, ideate-b, ideate-c] then one filter node

Repeat until exhausted

Condition: the amount of work is unknown; stop when two rounds add nothing.

const seen = new Set()
let dry = 0
while (dry < 2) {
  const r = await agent('Find one more batch of issues not already listed.', { schema: BATCH })
  const fresh = (r?.items ?? []).filter(i => !seen.has(i.key))
  if (!fresh.length) { dry++; continue }
  dry = 0
  fresh.forEach(i => seen.add(i.key))
}
// v1: sweep: {if: remaining == yes, goto: sweep, max: 20}

Gate

Condition: a human must approve before later stages may run. The runtime cannot pause for input, so the gate becomes a workflow boundary. Follow pipeshape-gate.

// v1: spec-gate: {gate: true}