RFC: Built-in orchestration primitives (parallel, pipeline) and tool approval hooks

[iakuf]

Summary

I'd like to propose two complementary features for Flue's runtime:

1. Orchestration primitives — parallel() and pipeline() functions that coordinate multiple session.task() calls with concurrency control, error handling, and observability.
2. Tool approval middleware — A withApproval() wrapper and createApprovalGate() factory that pause tool execution until an external system (UI, webhook, Slack) approves or rejects the action.

Both are inspired by Claude Code Dynamic Workflows' design philosophy: deterministic orchestration in code, autonomous reasoning in the model.

Motivation

Orchestration (parallel/pipeline)

Currently, running multiple tasks concurrently requires manual Promise.all + error handling + concurrency limiting boilerplate:

// Today: manual orchestration
const results = await Promise.all(
  tasks.map(t => session.task(t.prompt).catch(() => null))
);

This gets verbose when you need concurrency limits, labeled progress tracking, or mixed fail modes. Claude Code Workflows solved this with parallel() and pipeline() primitives — Flue would benefit from the same.

Tool Approval (human-in-the-loop)

Customer support agents, financial workflows, and any safety-critical agent need a way to pause before executing sensitive tools (refunds, deletions, deployments) and wait for human approval. Currently there's no standard pattern for this in Flue.

Proposed API

Orchestration

import { parallel, pipeline, phase } from '@flue/runtime/orchestrate';

// parallel(): barrier semantics, wait for all
phase('Research');
const results = await parallel(session, [
  { prompt: 'Analyze auth patterns', label: 'auth', agent: 'researcher' },
  { prompt: 'Analyze rate limiting', label: 'rate-limit' },
  { prompt: 'Check caching strategy', label: 'cache' },
], { concurrency: 8, failMode: 'lenient' });

// pipeline(): no barrier, each item flows independently through all stages
phase('Verify');
const verified = await pipeline(session, results.filter(Boolean), [
  (item) => ({ prompt: `Verify: ${item.text}` }),
  (item) => ({ prompt: `Format: ${item.text}`, agent: 'writer' }),
], { concurrency: 4 });

Key design decisions:

• parallel() uses barrier semantics (waits for ALL tasks), failed tasks return null in lenient mode
• pipeline() is barrier-free — fast items don't wait for slow ones
• Both respect AbortSignal for cancellation
• Concurrency is bounded by a semaphore (default: 16, matching Claude Code's limit)
• phase() and log() emit observable markers for tracing/UI

Tool Approval

import { createApprovalGate, withApproval } from '@flue/runtime/approval';

// Create a gate (transport-agnostic)
const gate = createApprovalGate({
  timeout: 300_000,
  onRequest: async (request) => {
    // Deliver to frontend via WebSocket, Slack, etc.
    ws.send(JSON.stringify({ type: 'approval_required', ...request }));
  },
});

// Wrap any tool
const protectedRefund = withApproval(refundTool, {
  gate,
  when: ({ amount }) => amount > 100,
  reason: ({ amount, orderId }) => `Refund ¥${amount} for order #${orderId}`,
});

// Resolve from external system
gate.resolve(approvalId, { action: 'approve' });

Key design decisions:

• Non-invasive: withApproval() returns a standard ToolDefinition — drop-in replacement
• Transport-agnostic: you provide onRequest, the gate handles the async waiting
• Timeout with auto-reject (configurable)
• Supports approve / reject / modify (execute with changed args)
• onDecision callback for audit logging

Real-world Use Case

I'm building a customer support agent (Vue frontend + Flue backend + Cube.js data layer) for a Mini PC e-commerce site. The flow:

1. parallel() — Classify intent + extract entities simultaneously
2. pipeline() — Each ticket goes through: analyze → query DB → generate response → QA check
3. withApproval() — Refunds over ¥100 pause for human sign-off via WebSocket push to admin panel

Implementation

I have a working implementation with 30 passing tests (vitest). The code is ~200 lines for orchestration and ~150 lines for approval. Happy to submit a PR if the direction looks good.

Reference implementation: https://github.com/iakuf/flue

Questions for Maintainers

1. Should these live in @flue/runtime (e.g. @flue/runtime/orchestrate) or as a separate package?
2. For orchestration: should phase() emit a FlueEvent that observe() subscribers can capture? Or is console logging sufficient for now?
3. For approval: should this integrate with the WebSocket protocol (new message type tool_approval_required), or stay as a userland pattern?
4. Any concerns about the API surface or naming conventions?

Looking forward to feedback! 🙏

[astrobot-houston]

Possibly related contribution from @iakuf — opened #202 which appears to be addressing this.

Their summary: Adds two new modules to @flue/runtime: orchestration primitives (parallel(), pipeline(), registerWorkflow(), phase(), log()) for deterministic task coordination, and a tool approval middleware (createApprovalGate(), withApproval()) that gates tool execution on external approval decisions with timeout and audit support. Includes 34 tests across both modules.

Implementation: iakuf/flue@feat/orchestration-primitives (diff)

This comment was posted automatically when #202 was redirected. The implementation is preserved on the branch above so it can inform the work here.

[iakuf]

Update: I've refactored the implementation based on a deeper look at Flue's concurrency model.

Key fix — respecting the exclusive-operation lock: the original parallel() ran concurrent session.task() calls on a single shared session, which violates runExclusive (it throws "already running ... start another session for parallel conversation branches"). The reworked primitives now take a FlueHarness and allocate one isolated session per concurrent unit, achieving true parallelism, then clean up the sessions afterward. This also lines up with @stainlu's Run -> Harness -> Session -> Operation -> Turn model noted in #147.

Two primitives:

• parallel(harness, tasks, opts) — barrier semantics, concurrency cap, lenient/strict fail modes, each task on its own session
• pipeline(harness, items, stages, opts) — no barrier; each item flows through stages sequentially on a dedicated session, items independent of each other. Stages can be inline functions or named workflows (registerWorkflow)

On the approval half: I think it belongs on top of the hooks.tool.before/after mechanism proposed in #168 rather than as a standalone wrapper — the hook gives the interception point, and a small withApproval helper + transport-agnostic gate handles the async "wait for human" part (WebSocket/Slack/webhook).

Tests now use a lock-faithful mock session (reproduces the exclusive lock), so they actually verify no two operations ever run on one session, plus session isolation and cleanup. Branch: feat/orchestration-primitives on my fork (linked above).

Happy to align naming/shape with whatever the operation + tool-hook contracts settle on.