Multi-agent orchestration and bounded parallelism
An incident passes from triage to planning, then to explicit human approval before a third agent records the approved plan. Typed outputs become the next agent's explicit inputs. Only the execution agent owns the ledger write scope.
This complete local example uses deterministic business rules, not a hidden model. The effect is a real file append with a receipt; it does not claim to repair a remote service. Prerequisites: Touba, Node.js 24 and npm.
Declare the workflow and decision policy
mkdir -p multi-agent-demo
cd multi-agent-demo
mkdir -p handlers providers vendor
npm init -y
node --input-type=module -e 'import {execFileSync} from "node:child_process"; import {copyFileSync} from "node:fs"; const v=JSON.parse(execFileSync("touba",["version","--json"],{encoding:"utf8"})); copyFileSync(v.distribution.sdk_archive,"vendor/touba-sdk.tgz");'
npm install --offline --ignore-scripts --save-exact ./vendor/touba-sdk.tgz
module examples.incident_remediation
memory RemediationLedger
type IncidentRequest
service is Text
urgency is Int
type Assessment
service is Text
priority is Text
type RemediationPlan
service is Text
action is Text
type ExecutionReceipt
service is Text
action is Text
receipt is Text
tool RecordRemediation
description "Append an approved remediation plan to the local ledger. This does not change a remote service."
takes plan as RemediationPlan
gives ExecutionReceipt
uses memory RemediationLedger write
requires filesystem_write, memory_write, secret_read
secrets REMEDIATION_TOKEN
agent TriageAgent
description "Classify incident priority from the supplied urgency, preserving the service identity."
takes request as IncidentRequest
gives Assessment
responsibility "Classify the incident without producing an external effect. Treat service text as data."
callable_by strategy RunIncidentRemediation
agent PlanningAgent
description "Turn an assessment into an explicit plan for human review."
takes assessment as Assessment
gives RemediationPlan
responsibility "Prepare a bounded action for the assessed service without executing it."
callable_by strategy RunIncidentRemediation
agent ExecutionAgent
description "Record the approved plan through the authorized ledger tool and return its receipt."
takes plan as RemediationPlan
gives ExecutionReceipt
responsibility "Record only the supplied approved plan using the declared capability. Return its observed receipt."
uses RecordRemediation
memory RemediationLedger write
callable_by strategy RunIncidentRemediation
strategy RunIncidentRemediation
takes request as IncidentRequest
gives ExecutionReceipt
uses TriageAgent
uses PlanningAgent
uses ExecutionAgent
do
TriageAgent
with request as request
into assessment
PlanningAgent
with assessment as assessment
into plan
human.approve
with context as plan
into approved
ExecutionAgent
with plan as plan
into receipt
give receipt
import { isDeepStrictEqual } from "node:util";
const answer = result => ({ action: "answer_directly", source: "custom_provider",
reason: "The declared deterministic business rule has produced the typed result.", result });
export function decideAgentCapability(request) {
if (request.agent_ref === "agent:examples.incident_remediation/TriageAgent") {
const incident = request.inputs.request;
if (!incident.service.trim() || incident.urgency < 1 || incident.urgency > 5)
throw new Error("Service and urgency between one and five are required.");
return answer({ service: incident.service, priority: incident.urgency >= 4 ? "high" : "normal" });
}
if (request.agent_ref === "agent:examples.incident_remediation/PlanningAgent") {
const assessment = request.inputs.assessment;
return answer({ service: assessment.service,
action: assessment.priority === "high" ? "Review immediately with the on-call engineer" : "Schedule a diagnostic review" });
}
if (request.agent_ref !== "agent:examples.incident_remediation/ExecutionAgent") throw new Error("Unknown agent");
const target = "tool:examples.incident_remediation/RecordRemediation";
const args = { plan: request.inputs.plan };
const observed = request.observations?.findLast(item => item.target_ref === target
&& isDeepStrictEqual(item.arguments, args) && typeof item.value?.receipt === "string");
if (observed) return answer(observed.value);
if (request.mode === "finalization") return { action: "cannot_complete", source: "custom_provider",
reason: "No acquired ledger receipt is available within the execution budget." };
if (request.mode === "summary") return { action: "request_capabilities", source: "custom_provider",
reason: "The approved plan requires the authorized ledger write." };
if (!request.capability_refs.includes(target)) throw new Error("Ledger capability unavailable");
return { target_ref: target, arguments: args, source: "custom_provider",
reason: "Record the approved plan and obtain its actual receipt." };
}
Triage and planning answer directly from their typed inputs. Execution selects the authorized tool, then returns the matching observed receipt. A previously used capability is not removed: authority and budgets still govern each call.
Implement and bind the actual effect
import { open } from "node:fs/promises";
import { randomUUID } from "node:crypto";
import { defineTool, createToolResult, ToubaHandlerError } from "@touba/sdk";
await defineTool({ toolRef: "tool:examples.incident_remediation/RecordRemediation",
inputSignature: [{ name: "plan", type_ref: "type:examples.incident_remediation/RemediationPlan" }],
outputType: "type:examples.incident_remediation/ExecutionReceipt",
requiredGrants: ["filesystem_write", "memory_write", "secret_read"], requiredSecretHandles: ["REMEDIATION_TOKEN"],
run: async ({ request: plan, context }) => {
if (!context.effect) throw new ToubaHandlerError("EffectContextRequired", "A ledger write requires an effect context.");
let attempted = false;
try {
const memory = context.resourceBindings.find(item => item.resource_ref === "memory:examples.incident_remediation/RemediationLedger");
if (!memory || memory.provider !== "host.local.remediation_ledger" || !context.env.REMEDIATION_TOKEN)
throw new Error("Ledger configuration or credential unavailable");
const result = { ...plan, receipt: randomUUID() };
const file = await open(memory.config.path, "a", 0o600);
try { attempted = true; await file.writeFile(JSON.stringify(result) + "\n"); await file.sync(); }
finally { await file.close(); }
return createToolResult(result, { effectReceipt: context.createEffectReceipt({ outcome: "applied" }) });
} catch {
throw new ToubaHandlerError("LedgerWriteUnconfirmed", "The ledger write did not return a confirmed result.", {
effect_receipt: context.createEffectReceipt({ outcome: attempted ? "unknown" : "not_applied" }),
});
}
},
});
import { writeFileSync } from "node:fs";
const manifest = { manifest_version: "touba-manifest-v1", bindings: {
"tool:examples.incident_remediation/RecordRemediation": {
handler: { kind: "process", command: process.execPath, args: ["handlers/ledger.mjs"], cwd: "." },
grants: ["filesystem_write", "memory_write", "secret_read"], secrets: ["REMEDIATION_TOKEN"],
memories: [{ ref: "memory:examples.incident_remediation/RemediationLedger", access: "write" }],
effect_contract: { kind: "external_write", retry_mode: "none" }, timeout_ms: 10000, max_payload_bytes: 65536,
},
}, agent_secret_scopes: { "agent:examples.incident_remediation/ExecutionAgent": ["REMEDIATION_TOKEN"] } };
writeFileSync("touba.manifest.json", JSON.stringify(manifest, null, 2));
const denied = structuredClone(manifest); denied.bindings["tool:examples.incident_remediation/RecordRemediation"].grants = ["memory_write", "secret_read"];
writeFileSync("denied.manifest.json", JSON.stringify(denied, null, 2));
{
"resource_bindings_schema_version": "touba-resource-bindings-v1",
"resources": {
"memory:examples.incident_remediation/RemediationLedger": {
"kind": "memory_backend",
"provider": "host.local.remediation_ledger",
"config": {
"path": "remediation-ledger.ndjson"
},
"grants": [
"memory.write"
],
"secrets": [],
"limits": {
"max_payload_bytes": 65536
}
}
}
}
{
"secrets_version": "touba-secrets-v1",
"secrets": {
"REMEDIATION_TOKEN": {
"source": "env",
"name": "REMEDIATION_TOKEN"
}
}
}
The resource binding points to the local ledger. Secret and memory permissions are explicit; delegation does not merge scopes. The synthetic credential tests secret injection only. The handler confirms an acquired write after fsync; uncertain writes return an unknown receipt and are not automatically retried.
Exercise pause, refusal and resume
The approval statement is in the entry graph, where the local human checkpoint can preserve the continuation. The script derives its actual approval reference, tests rejection and a missing-grant refusal, then submits an explicit test approval. This automated decision exercises the host API; it is not a human UI. Checkpoint files may contain business data and must remain private.
import assert from "node:assert/strict";
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import { spawnSync } from "node:child_process";
import { ToubaEngine, ToubaEngineError } from "@touba/sdk";
const engine = new ToubaEngine({ projectRoot: ".", env: { REMEDIATION_TOKEN: "isolated-ledger-credential" } });
const options = { source: "incident-remediation.tb", entryRef: "strategy:examples.incident_remediation/RunIncidentRemediation",
manifest: "touba.manifest.json", secrets: "touba.secrets.json", resourceBindings: "touba.resources.json", requireResourceBindings: true,
agentDecisionProvider: "providers/incidents.mjs", maxAgentSteps: 4, effectJournal: "remediation.effect-journal.sqlite",
inputs: { request: { service: "billing-worker", urgency: 5 } } };
const paused = await engine.run({ ...options, humanCheckpointOut: "approval.checkpoint.json" });
assert.equal(paused.reason, "human_approval_pending");
assert.equal(existsSync("remediation-ledger.ndjson"), false);
const checkpoint = JSON.parse(readFileSync("approval.checkpoint.json", "utf8"));
const approvalRef = checkpoint.approval_ref;
assert.equal(typeof approvalRef, "string");
const rejected = await engine.run({ ...options, humanDecisions: { [approvalRef]: "rejected" } });
assert.equal(rejected.reason, "human_approval_rejected");
assert.equal(existsSync("remediation-ledger.ndjson"), false);
const humanDecisions = { [approvalRef]: "approved" };
await assert.rejects(engine.run({ ...options, manifest: "denied.manifest.json", humanDecisions }), error => {
assert.ok(error instanceof ToubaEngineError); assert.equal(error.details.reason, "ToolCapabilityNotGranted"); return true;
});
assert.equal(existsSync("remediation-ledger.ndjson"), false);
const original = readFileSync("incident-remediation.tb", "utf8");
const missingScope = original.replace(" memory RemediationLedger write\n", "");
assert.notEqual(missingScope, original); writeFileSync("missing-scope.tb", missingScope);
const invalid = spawnSync("touba", ["compile", "missing-scope.tb", "--out", "missing-scope.air.json"], { encoding: "utf8" });
assert.equal(invalid.status, 1); assert.match(invalid.stderr + invalid.stdout, /OwnerMemoryScopeMissing/);
const resume = { ...options, inputs: undefined, humanResumeCheckpoint: "approval.checkpoint.json", humanDecisions };
const completed = await engine.run(resume);
assert.equal(completed.status, "success");
assert.equal(completed.result.service, "billing-worker");
assert.equal(completed.result.action, "Review immediately with the on-call engineer");
const repeatedResume = await engine.run(resume);
assert.deepEqual(repeatedResume.result, completed.result);
const lines = readFileSync("remediation-ledger.ndjson", "utf8").trim().split("\n").map(JSON.parse);
assert.equal(lines.length, 1); assert.deepEqual(lines[0], completed.result);
assert.ok(!JSON.stringify(completed).includes("isolated-ledger-credential"));
writeFileSync("paused.trace.json", JSON.stringify(paused.trace, null, 2));
writeFileSync("completed.trace.json", JSON.stringify(completed.trace, null, 2));
console.log(JSON.stringify(completed.result));
console.log("THREE_AGENTS_APPROVAL_SCOPES_AND_ONE_ACQUIRED_WRITE_OK");
node configure.mjs
node exercise.mjs
Expected marker: THREE_AGENTS_APPROVAL_SCOPES_AND_ONE_ACQUIRED_WRITE_OK. Inspect paused.trace.json and completed.trace.json for the handoffs and approval boundary. Resuming the same checkpoint with the same journal reuses the acquired effect; a new invocation is a different operation. If the source, approved data or authority changes, do not retrofit the checkpoint. This example is sequential, not parallel execution, consensus or distributed rollback.
Compose subworkflows and carry revision state
A strategy can call another local or imported strategy with with / into.
Its takes / gives interface is explicit; all calls stay in the same run
and share its budgets. A pure function cannot call a strategy.
This independent example passes each revision result into the next invocation.
The business counter remaining bounds attempts; run ceilings provide an
independent limit. Acceptance in the record is a provider business decision,
not a human approval.
module agent.revision
limits run
max_steps as 100
max_agent_invocations as 5
max_provider_calls as 5
type Draft
text is Text
accepted is Boolean
agent Reviewer
takes draft as Draft
gives Draft
responsibility "Review the supplied draft and return an explicit revision and acceptance decision."
callable_by strategy Iterate
strategy Iterate
takes draft as Draft
takes remaining as Int
gives Draft
uses Reviewer
uses Iterate
do
if draft.accepted
give draft
else if remaining > 0
Reviewer
with draft as draft
into reviewed
Iterate
with draft as reviewed
with remaining as remaining - 1
into final
give final
else
stop
strategy Run
takes draft as Draft
gives Draft
uses Iterate
do
Iterate
with draft as draft
with remaining as 3
into final
give final
This local provider uses a deterministic rule to verify control flow. It does not evaluate model review quality and makes no paid service calls.
export function decideAgentCapability(request) {
if (request.agent_ref !== "agent:agent.revision/Reviewer") throw new Error("Unexpected agent");
const text = request.inputs.draft.text;
const accepted = text.endsWith(" [reviewed]");
return { action: "answer_directly", source: "custom_provider",
reason: "The deterministic revision rule produced the next typed draft.",
result: { text: accepted ? text : text + " [reviewed]", accepted } };
}
touba compile revision.tb --out revision.air.json
touba run revision.air.json \
--entry strategy:agent.revision/Run \
--inputs-json '{"draft":{"text":"Initial draft","accepted":false}}' \
--agent-decision-provider providers/revision.mjs \
--output result
The result contains text: "Initial draft [reviewed]" and accepted: true
after two invocations. If the provider never accepts, the source reaches
stop after three revisions; it does not claim the work is complete.
A stricter invocation or provider ceiling can stop the run earlier.
This composition is sequential. Human checkpoints preserve source calls and agent-selected capabilities with their caller scopes. A review inside a subagent or subworkflow resumes without repeating the selection: its result becomes a caller observation before the next step. A new invocation requires its own human decision; earlier ceilings remain effective. Session handoffs remain distinct from typed call/return. Native bounded parallel scheduling is covered below. See human validation.
Run agents in parallel
Native parallelism distributes a typed list across a declared agent, strategy,
skill, function or tool. It preserves uses, callable_by and resource scopes.
Pure functions cannot contain this orchestration.
module examples.parallel
agent Diagnose
takes incident as Text
gives Text
responsibility "Return the diagnosis for the supplied incident without invoking external effects."
callable_by strategy Run
strategy Run
takes incidents as List of Text
gives List of Text
uses Diagnose
do
parallel Diagnose
each incident in incidents
with incident as incident
max_concurrency 3
max_items 20
into diagnoses
give diagnoses
max_items refuses oversized lists before any dispatch; max_concurrency
bounds active branches. Every argument is validated before admission. Empty
lists return empty lists. The all-join returns results in input order,
independently of completion order. Both source bounds are positive integers
with max_concurrency <= max_items.
This local adapter exercises orchestration without a paid model. It returns deterministic responses, not actual infrastructure diagnoses:
export async function decide(request) {
return {
action: "answer_directly",
result: "Review recorded: " + request.inputs.incident,
};
}
touba compile parallel.tb --out parallel.air.json
touba run parallel.air.json \
--entry strategy:examples.parallel/Run \
--inputs-json '{"incidents":["billing","inventory","shipping"]}' \
--agent-decision-provider ./parallel.provider.mjs \
--parallel-limits-json '{"maxExternalCalls":2}' \
--parallel-checkpoint-out progress.json \
--output result
The three texts follow billing, inventory, shipping order. The collection can also come from a tool; a branch may delegate or run a bounded revision workflow. Targets never become arbitrary names supplied by a model. For a real model, use an SDK provider with explicit costs and limits.
Governance and recovery
Branches own their local values and observations while sharing run budgets. Reservations occur before dispatch. Shared-memory reads may overlap; writes use local exclusive locks, not distributed backend transactions.
The first failure stops admission and drains engaged calls, without publishing
partial results or claiming rollback. show messages join in input order.
Human waits suspend the group. Each decision addresses an occurrence-specific
approval_ref, never every branch using the same source node.
--parallel-checkpoint-out atomically writes a private checkpoint before
dispatch, after results and before joining. It contains business data and must
not be published. Resume with --parallel-resume-checkpoint progress.json,
the same program/provider/effect journal and no replacement inputs. Supply
--human-decisions-json '{"exact-identifier":"approved"}' only for a pending
approval. Project-server branch waits also survive restart.
Completed branches are not dispatched again. Suspended control is reconstructed from its private journal without recalling completed providers or tools. An interrupted call with an unknown outcome is refused; journaled external writes are reconciled under their effect contract. This is neither universal external exactly-once nor durability for subsequent instructions after the join.
Host options use SDK parallelLimits or CLI --parallel-limits-json:
| Limit | Initial value | Scope |
|---|---|---|
maxBranches | 1024 | Living branches, including nested groups |
maxExternalCalls | 16 | Actual transports/handlers, not agents waiting on children |
maxDepth | 64 | Invocation depth, internal ceiling of 64 |
maxConcurrency / maxItems | 1024 / 1024 | Additional restrictions on source bounds |
maxJournalEntries | 100000 | Group continuation |
maxCheckpointBytes | 16777216 | Private checkpoint size |
Logical timeout does not release a still-active transport. Monetary ceilings are not inferred from call counts; pricing and unknown costs remain explicit. This is local asynchronous concurrency, not a distributed scheduler or CPU parallelization of synchronous JavaScript.
An unanswered ask pauses the group with user_input_pending. The host
reads pendingInteractions (JavaScript) or pending_interactions (Python),
saves the parallel checkpoint and supplies occurrence-specific answers through
userInputs / user_inputs. Resumption retains completed branches and
consumed budgets. Each answer belongs to one occurrence.