# Pipeshape Topologies

> Orchestration shapes as workflow script snippets. Use when composing a workflow that fans out, loops on review, routes by kind, verifies adversarially, or repeats until exhausted.

- Skill: `terrasnail/pipeshape-topologies` (Agent Skill)
- Install (CLI): `npx skillmds@latest add terrasnail/pipeshape-topologies`
- Raw SKILL.md: https://api.skillmd.com/api/skills/terrasnail/pipeshape-topologies/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Productivity
- Author: terrasnail (https://skillmd.com/u/terrasnail)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/terrasnail/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.

```javascript
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.

```javascript
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.

```javascript
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.

```javascript
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.

```javascript
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.

```javascript
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.

```javascript
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.

```javascript
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}
```