Parallel, Quorums & Child Workflows

AGBoss provides first-class HarnessDSL (HDSL) flow primitives for deterministic fan-out/fan-in (.toParallelAgents(), .toParallelWorkflows()), deadline-bounded quorum consensus (.acceptedBy(), .joinQuorum(), .joinQuorumGroups(group(...).priority(1)), @Arbiter), modular child workflows (.toWorkflow(Child)), and depth-bounded recursion (ThisWorkflow).

How parallel fan-out, .join(), and quorum consensus work

When a step fans out across multiple specialists or a dynamic list of items (maxItems, maxConcurrency), every branch receives an isolated copy of the Path Scope. You then merge results in one of three ways:

• Wait for all (.join()): Waits for every branch to complete and binds all outputs via .returns(...).
• Count-based quorum (.acceptedBy(Arbiter).joinQuorum({ min, max, deadline })): Evaluates each arriving branch output with one or more @Arbiter classes (anyOf(A, B) accepts if either passes; chaining .acceptedBy(A).acceptedBy(B) requires both). As soon as max (default min) accepted answers arrive, the join proceeds immediately and aborts slower in-flight branches (BranchVerdict.Surplus). Unless min equals the total number of launched agents, single-agent outputs after .joinQuorum() are typed as optional(Dto) in .returns(...).
• Priority group quorum (.acceptedBy(Arbiter).joinQuorumGroups(group(...).priority(1)..., { deadline })): Defines prioritized combinations of agents (priority(1) is highest), where each group(...) declares its own .returns(...) and continuation path:

src/workflows/parallel-review.workflow.ts

import { QuorumNotReachedError, Workflow, group, start, toAction } from "@agboss/core";
import { seconds } from "@agboss/core/units";

@Workflow({
  name: "parallel-review-workflow",
  version: "1.0.0",
  flow: [
    start(ProposalStart)
      .returns(ProposalDto)
      .toParallelAgents(SecurityAgent, LegalAgent, BackupAuditAgent)
      .acceptedBy(ReviewQuorumArbiter)
      .joinQuorumGroups(
        group(SecurityAgent, LegalAgent)
          .priority(1)
          .returns(SecurityVerdictDto)
          .returns(LegalVerdictDto)
          .toAction(MergePrimaryVerdictsAction)
          .returns(ApprovedProposalDto)
          .toFinish(ProposalApprovedFinish),
        group(SecurityAgent, BackupAuditAgent)
          .priority(2)
          .returns(SecurityVerdictDto)
          .returns(BackupVerdictDto)
          .toAction(MergeFallbackVerdictsAction)
          .returns(ApprovedProposalDto)
          .toFinish(ProposalApprovedFinish),
        { deadline: seconds(10) },
      )
      .onJoinError(
        QuorumNotReachedError,
        toAction(RecordQuorumFailureAction)
          .toFinish(ProposalRejectedFinish),
      ),
  ],
})
export class ParallelReviewWorkflow {}

test/parallel-review.workflow.test.ts

import { expect } from "vitest";
import { ExecutionStatus } from "@agboss/core";
import { TestConcurrency, replyWith, step, testWith } from "@agboss/core/testing";
import {
  ReviewApplication,
  ProposalStart,
  SecurityAgent,
  BackupAuditAgent,
  LegalAgent,
  SecurityVerdictDto,
  BackupVerdictDto,
} from "../src/parallel-review.workflow.js";

const test = testWith(ReviewApplication, {
  concurrency: TestConcurrency.inOrder([
    step(BackupAuditAgent),
    step(SecurityAgent),
    step(LegalAgent),
  ]),
});

test("reaches priority-2 quorum when SecurityAgent and BackupAuditAgent finish first", async ({ app, mockLlm }) => {
  mockLlm(BackupAuditAgent).thenReturn(replyWith(BackupVerdictDto, { passed: true }));
  mockLlm(SecurityAgent).thenReturn(replyWith(SecurityVerdictDto, { passed: true, confidence: 95 }));
  const res = await app.execute(ProposalStart, { title: "SOC2 Policy Update" });
  expect(res.status).toBe(ExecutionStatus.Finished);
});

Recursive ThisWorkflow decomposition, soft stop at maxDepth, and KV cache prefixes

A workflow can invoke a modular child workflow (.toWorkflow(ChildWorkflow)) or recursively invoke itself (ThisWorkflow) either deterministically or as a model-driven tool call:

src/workflows/recursive-research.workflow.ts

import { ChildPlacement, ThisWorkflow, Workflow, caseOf, listOf, start } from "@agboss/core";

@Workflow({
  name: "recursive-research-workflow",
  version: "1.0.0",
  flow: [
    start(ResearchStart)
      .returns(TopicDto)
      .toAgent(DecomposeTopicAgent)
      .returns(oneOf(ResolvedDto, SubTopicsDto))
      .switch(
        caseOf(ResolvedDto)
          .toWorkflow(FactCheckChildWorkflow, {
            placement: ChildPlacement.AnyInstance,
          })
          .returns(VerifiedSummaryDto)
          .toFinish(ResearchFinish),
        caseOf(SubTopicsDto)
          .toAction(ExtractSubTopicsAction)
          .returns(listOf(TopicDto))
          .toParallelWorkflows(ThisWorkflow, {
            maxItems: 5,
            maxConcurrency: 2,
            placement: ChildPlacement.AnyInstance,
          })
          .join()
          .returns(listOf(VerifiedSummaryDto))
          .toAction(MergeFindingsAction)
          .returns(VerifiedSummaryDto)
          .toFinish(ResearchFinish),
      ),
  ],
  maxDepth: 3,
  maxConcurrentExecutions: 6,
})
export class RecursiveResearchWorkflow {}

test/recursive-research.workflow.test.ts

import { expect, it } from "vitest";
import { testAction } from "@agboss/core/testing";
import {
  DecomposeTopicAgent,
  ExtractSubTopicsAction,
  RecursiveResearchWorkflow,
} from "../src/recursive-research.workflow.js";

it("extracts sub-topic items for recursive ThisWorkflow fan-out", async () => {
  expect(RecursiveResearchWorkflow).toBeDefined();
  const runExtract = testAction(ExtractSubTopicsAction);
  const { output } = await runExtract({
    subTopics: {
      value: { items: [{ topic: "Raft" }, { topic: "Paxos" }] },
      producer: DecomposeTopicAgent,
    },
  });
  expect(output).toEqual([{ topic: "Raft" }, { topic: "Paxos" }]);
});

How maxDepth enforces a soft stop (and warms at most two KV cache prefixes)

Every recursive workflow requires maxDepth on @Workflow ([graph.missing-max-depth]). How maxDepth behaves at runtime depends on whether ThisWorkflow is launched by the model or by deterministic code:

• Model-chosen recursion (soft stop at depth === maxDepth): When an @Agent or DecidedBy.Model @Router can launch ThisWorkflow, reaching depth === maxDepth does not crash the execution. Instead, AGBoss simply omits the ThisWorkflow tool from the model’s tool definitions on that step, forcing the model to synthesize a base-case answer instead of recursing further. Because the tool list is identical across all depth < maxDepth calls and differs only at depth === maxDepth, the recursive agent warms at most two KV prefix cache entries across the entire recursion tree.
• Deterministic recursion (.toWorkflow(ThisWorkflow)): If a deterministic flow path unconditionally launches ThisWorkflow past maxDepth, the runtime throws DepthLimitExceededError (catchable via .onError(DepthLimitExceededError, ...)).

Key safeguards for parallel and recursive execution

Safeguard	Option / Error	How it protects production
Fan-out truncation	maxItems + OnLaunchItemsTruncated	Caps dynamic list fan-out and emits a typed observer event when excess items are truncated.
Bounded branch concurrency	maxConcurrency	Limits how many parallel branches execute simultaneously within a fan-out step.
Tree-wide execution throttle	maxConcurrentExecutions	Caps how many executions across the root and all child workflow trees run a step simultaneously (default 10; enforced across instances via RateLimitRepository when ChildPlacement.AnyInstance is used).
Distributed child placement	placement: ChildPlacement.AnyInstance	Enqueues .toWorkflow(), .toParallelWorkflows(), or route(Workflow) children into ChildExecutionQueueRepository (SqliteChildExecutionQueueRepository), yielding the parent’s lease while waiting and recovering each child independently under its own 30s lease ([flow.conflicting-placement] if two paths to the same child disagree).
Quorum cancellation	deadline: seconds(10)	Automatically aborts slower losing branches (BranchVerdict.Surplus) via AbortSignal the instant quorum is decided.
Recursion ceiling	maxDepth	Required on every ThisWorkflow graph ([graph.missing-max-depth]); omits the recursion tool at depth === maxDepth for model launches (soft stop) and throws DepthLimitExceededError if exceeded by deterministic launches.