feat(dag): harness batch — parallel writers, REJECT checkpoints, union-arg tolerance

packages/core/src/plugin/command/orchestration-policy.md

@@ -267,6 +267,22 @@ of it in the same wake turn with exactly one of:
 4. A reasoned stop — tell the user, finding by finding, why no further wave
    is warranted. Silence is not a stop decision.
 
+Classify the findings first; the class selects the option:
+
+| Finding nature | Disposal |
+| --- | --- |
+| Bounded within the current scope | Option 1, same graph, tier unchanged |
+| Reveals cross-module, contract, persistence, or boundary risk the current shape cannot cover | Option 1 escalated: append full-shaped assurance lanes (broader review axes, extra verification) instead of the lite correction alone |
+| The situation itself was misclassified (a change assumed, an unknown-cause defect found; a repair assumed, a design gap found) | Option 3 only — the single legitimate route switch; start the workflow whose backbone matches the real deliverable and name which prior evidence carries over |
+| Unbounded or foggy — findings that no bounded wave can discharge | Option 4, or escalate to the user with a decision request; do not launder fog into a speculative wave |
+
+Route reselection is never the default: same-objective work stays in one
+workflow, and escalation keeps additive semantics — full-shaped assurance
+lanes are appended waves, not replacement graphs. The tier discriminator is
+risk only (reversibility, module span, public contracts, concurrency,
+persistence, migration, identity, authorization, upstream executables,
+CI/release). Role or block count never selects a tier.
+
 Merely summarizing a non-ACCEPT verdict and ending the turn is an
 orchestration failure. The runtime's `orchestrator_unresponsive` guard only
 fires for workflows that are still live; a checkpoint that terminalizes its

packages/core/src/plugin/command/workflow-blocks.md

@@ -147,9 +147,16 @@ stay quiet. A block immediately after a review gate is conditioned on its
 accepted verdict. Because the condition language handles one verdict reference,
 fan multiple review lanes into one review block before continuing.
 
-All block workers share one workspace. The compiler serializes
-otherwise-unordered `coding` and `prototype` writers, while read-only lanes may
-remain parallel. The resident Orchestration Router owns route selection and
+All block workers share one workspace. Unordered `coding` and `prototype`
+writers run in parallel, so their work packages must be triple-disjoint:
+source files, generated artifacts, and lockfiles must not overlap, and no
+shared build may be triggered. The plan block owns this partition. For an
+implementation review over parallel writers, the compiler injects one
+read-only aggregation node between the writers and the verification gate; it
+fails loudly when the declared write sets overlap and otherwise publishes the
+union with a single implementation fingerprint computed at the convergence
+point. Total-ordered writer chains compile unchanged. Read-only lanes remain
+parallel throughout. The resident Orchestration Router owns route selection and
 phase pruning; this guide owns block fields, contracts, and graph mechanics.
 
 ## When to use low-level nodes

packages/core/src/plugin/command/workflow-routing.md

@@ -53,12 +53,18 @@ suffice, work is reversible, and no high-risk boundary is involved. Use `full`
 when any are true: requirements or design are uncertain; work crosses modules
 or write owners; a public contract, concurrency, persistence, migration,
 identity, authorization, upstream executable dependencies, CI/release, or
-production behavior is in scope. A single matching custom workflow has no tier
-to infer: read and retarget it directly.
-
-If a lite reporting gate returns non-`ACCEPT`, let that graph finish and use
-additive `extend` with new node IDs after reassessing the live library. Do not
-pause or replan a completed workflow; the parent owns this control decision.
+production behavior is in scope. Only these risk dimensions select a tier —
+never role count or block count, which are consequences of risk, not causes.
+A single matching custom workflow has no tier to infer: read and retarget it
+directly.
+
+If a lite reporting gate returns non-`ACCEPT`, let that graph finish and
+dispose of the verdict under the Verdict Disposal Contract. When the findings
+cross any `full` criterion above, the correction wave MUST be full-shaped:
+escalate by appending full-shaped assurance lanes with new node IDs in the
+same workflow — a tier escalation is an additive wave, never a replacement
+workflow. Do not pause or replan a completed workflow; the parent owns this
+control decision.
 
 The primary reference follows the final artifact, not every concern. For code
 or repairs, review, security, and performance are secondary assurance in that

packages/core/test/plugin/command.test.ts

@@ -118,8 +118,10 @@ describe("CommandPlugin.Plugin", () => {
       expect(CommandPlugin.WorkflowContent).toContain("upstream executable dependencies")
       expect(CommandPlugin.WorkflowContent).toContain("single matching custom workflow")
       expect(CommandPlugin.WorkflowContent).toContain("Do not concatenate two complete references")
-      expect(CommandPlugin.WorkflowContent).toContain("additive `extend` with new node IDs")
-      expect(CommandPlugin.WorkflowContent).toContain("Do not\npause or replan a completed workflow")
+      expect(CommandPlugin.WorkflowContent).toContain("never role count or block count, which are consequences of risk")
+      expect(CommandPlugin.WorkflowContent).toContain("dispose of the verdict under the Verdict Disposal Contract")
+      expect(CommandPlugin.WorkflowContent).toContain("the correction wave MUST be full-shaped")
+      expect(CommandPlugin.WorkflowContent).toContain("Do not pause or replan a completed workflow")
     }),
   )
 
@@ -249,6 +251,9 @@ describe("CommandPlugin.Plugin", () => {
       )
       expect(CommandPlugin.OrchestrationPolicyContent).toContain("escapes that guard")
       expect(CommandPlugin.OrchestrationPolicyContent).toContain("Silence is not a stop decision")
+      expect(CommandPlugin.OrchestrationPolicyContent).toContain("Classify the findings first; the class selects the option")
+      expect(CommandPlugin.OrchestrationPolicyContent).toContain("The situation itself was misclassified")
+      expect(CommandPlugin.OrchestrationPolicyContent).toContain("Role or block count never selects a tier")
       expect(CommandPlugin.WorkflowFactsContent).toContain("Verdict Disposal Contract")
     }),
   )

packages/opencode/src/dag/blocks.ts

@@ -120,9 +120,9 @@ export function compileWorkflowBlocks(
   options: WorkflowBlockCompileOptions = {},
 ): NodeConfig[] {
   requireValidBlockGraph(graph, options)
-  const blocks = serializeWorkspaceWriters(graph.blocks)
-  requireValidReviewRoutes(blocks)
-  const nodes = blocks.flatMap((block) => compileBlock(graph.objective, block, blocks))
+  requireValidReviewRoutes(graph.blocks)
+  const { blocks, aggregations, verifyAggregators } = aggregateParallelWriters(graph.blocks)
+  const nodes = blocks.flatMap((block) => compileBlock(graph.objective, block, blocks, aggregations, verifyAggregators))
   const duplicateNodeIDs = uniqueDuplicates(nodes.map((node) => node.id))
   if (duplicateNodeIDs.length > 0) {
     throw new Error(
@@ -132,7 +132,13 @@ export function compileWorkflowBlocks(
   return nodes
 }
 
-function compileBlock(objective: string, block: WorkflowBlock, blocks: WorkflowBlock[]): NodeConfig[] {
+function compileBlock(
+  objective: string,
+  block: WorkflowBlock,
+  blocks: WorkflowBlock[],
+  aggregations: Map<string, WriterAggregation>,
+  verifyAggregators: Map<string, string[]>,
+): NodeConfig[] {
   const dependencies = block.depends_on ?? []
   const required = block.required ?? (block.kind === "plan" || block.kind === "verify" || block.kind === "synthesize")
   const reviewDependency = dependencies.find(
@@ -171,20 +177,22 @@ function compileBlock(objective: string, block: WorkflowBlock, blocks: WorkflowB
   }
 
   if (block.kind === "review") {
-    const standardsID = `${block.id}--standards`
-    const intentID = `${block.id}--intent`
-    const route = implementationReviewRoute(block, blocks)
-    const reviewCondition = route ? `${route.verification.id}.output.verdict == "PASS"` : condition
+    const aggregation = aggregations.get(block.id)
+    const legacyRoute = aggregation ? undefined : implementationReviewRoute(block, blocks)
+    const implementationID = aggregation ? aggregation.aggregatorID : legacyRoute?.implementation.id
+    const verificationID = aggregation ? aggregation.verificationID : legacyRoute?.verification.id
+    const route = implementationID && verificationID ? { implementationID, verificationID } : undefined
+    const reviewCondition = route ? `${route.verificationID}.output.verdict == "PASS"` : condition
     const reviewEvidence = route
       ? {
-          implementation_changed_files: `${route.implementation.id}.output.changed_files`,
-          implementation_fingerprint: `${route.implementation.id}.output.fingerprint`,
-          verification: `${route.verification.id}.output`,
+          implementation_changed_files: `${route.implementationID}.output.changed_files`,
+          implementation_fingerprint: `${route.implementationID}.output.fingerprint`,
+          verification: `${route.verificationID}.output`,
         }
       : undefined
-    return [
+    const lanes = [
       node({
-        id: standardsID,
+        id: `${block.id}--standards`,
         name: `${block.id}: standards review`,
         workerType: block.worker_type ?? "general",
         dependencies,
@@ -197,7 +205,7 @@ function compileBlock(objective: string, block: WorkflowBlock, blocks: WorkflowB
         inputMapping: reviewEvidence,
       }),
       node({
-        id: intentID,
+        id: `${block.id}--intent`,
         name: `${block.id}: intent review`,
         workerType: block.worker_type ?? "general",
         dependencies,
@@ -213,7 +221,7 @@ function compileBlock(objective: string, block: WorkflowBlock, blocks: WorkflowB
         id: block.id,
         name: `${block.id}: review decision`,
         workerType: block.worker_type ?? "general",
-        dependencies: [standardsID, intentID, ...(route ? [route.verification.id] : [])],
+        dependencies: [`${block.id}--standards`, `${block.id}--intent`, ...(route ? [route.verificationID] : [])],
         objective,
         instruction: block.instruction,
         contract: [
@@ -229,22 +237,45 @@ function compileBlock(objective: string, block: WorkflowBlock, blocks: WorkflowB
         inputMapping: route
           ? {
               ...reviewEvidence,
-              standards_review: `${standardsID}.output`,
-              intent_review: `${intentID}.output`,
+              standards_review: `${block.id}--standards.output`,
+              intent_review: `${block.id}--intent.output`,
             }
           : undefined,
         review: route
           ? {
               phase: "diff",
-              implementation_node_id: route.implementation.id,
-              verification_node_id: route.verification.id,
+              implementation_node_id: route.implementationID,
+              verification_node_id: route.verificationID,
             }
           : undefined,
         outputSchema: route ? DIFF_REVIEW_SCHEMA : GENERAL_VERDICT_SCHEMA,
       }),
     ]
+    if (!aggregation) return lanes
+    return [
+      node({
+        id: aggregation.aggregatorID,
+        name: `${block.id}: aggregate parallel implementation evidence`,
+        workerType: "explore",
+        dependencies: aggregation.writerIDs,
+        objective,
+        contract: AGGREGATOR_CONTRACT,
+        required: true,
+        reportToParent: false,
+        inputMapping: Object.fromEntries(
+          aggregation.writerIDs.flatMap((writerID: string) => [
+            [`${writerID.replace(/-/g, "_")}_changed_files`, `${writerID}.output.changed_files`],
+            [`${writerID.replace(/-/g, "_")}_summary`, `${writerID}.output.summary`],
+          ]),
+        ),
+        outputSchema: IMPLEMENTATION_SCHEMA,
+      }),
+      ...lanes,
+    ]
   }
 
+  const verifyAggregatorIDs = verifyAggregators.get(block.id)
+  const verifyAggregator = verifyAggregatorIDs && verifyAggregatorIDs.length > 0 ? verifyAggregatorIDs[0] : undefined
   return [
     node({
       id: block.id,
@@ -257,6 +288,12 @@ function compileBlock(objective: string, block: WorkflowBlock, blocks: WorkflowB
       required,
       reportToParent: block.report_to_parent ?? block.kind === "synthesize",
       condition,
+      inputMapping: verifyAggregator
+        ? {
+            implementation_changed_files: `${verifyAggregator}.output.changed_files`,
+            implementation_fingerprint: `${verifyAggregator}.output.fingerprint`,
+          }
+        : undefined,
       outputSchema: WRITER_KINDS.has(block.kind)
         ? IMPLEMENTATION_SCHEMA
         : block.kind === "verify"
@@ -350,33 +387,75 @@ function requireValidBlockGraph(graph: WorkflowBlockGraph, options: WorkflowBloc
   topologicalBlocks(graph.blocks)
 }
 
-function serializeWorkspaceWriters(blocks: WorkflowBlock[]) {
-  const writers = topologicalBlocks(blocks).filter((block) => WRITER_KINDS.has(block.kind))
-  const previousWriter = new Map(
-    writers.slice(1).map((block, index) => [block.id, writers[index]?.id ?? block.id] as const),
-  )
-  const serialized = blocks.map((block) => {
-    const previous = previousWriter.get(block.id)
-    if (!previous || dependsTransitively(blocks, block.id, previous)) return block
+// Injected between parallel implementation writers and their verification
+// gate: mechanically detects declared write-set overlap (loud node failure)
+// and publishes the union with one fingerprint computed at the convergence
+// point, so diff review binds to a single post-merge state.
+const AGGREGATOR_CONTRACT =
+  "Collect the supplied changed-file lists and summaries from each parallel implementation writer. If any file path appears in more than one list, do not submit; fail the node naming the exact overlapping paths. Otherwise submit the union of all changed files and one stable fingerprint computed at this convergence point (for example a sha256 over the sorted union of current file contents, reporting the exact commands used). Do not modify any file."
+
+interface WriterAggregation {
+  aggregatorID: string
+  writerIDs: string[]
+  verificationID: string
+}
+
+function aggregateParallelWriters(blocks: WorkflowBlock[]) {
+  const aggregations = new Map<string, WriterAggregation>()
+  for (const block of blocks) {
+    if (block.kind !== "review") continue
+    const topology = reviewWriterTopology(block, blocks)
+    if (!topology) continue
+    if (canonicalWriter(topology, blocks)) continue
+    aggregations.set(block.id, {
+      aggregatorID: `${block.id}--aggregate`,
+      writerIDs: topology.implementations.map((writer) => writer.id),
+      verificationID: topology.verification.id,
+    })
+  }
+  if (aggregations.size === 0) {
+    return { blocks, aggregations, verifyAggregators: new Map<string, string[]>() }
+  }
+  const writerToAggregators = new Map<string, string[]>()
+  for (const aggregation of aggregations.values()) {
+    for (const writerID of aggregation.writerIDs) {
+      writerToAggregators.set(writerID, [...(writerToAggregators.get(writerID) ?? []), aggregation.aggregatorID])
+    }
+  }
+  const aggregatorIDs = new Set([...aggregations.values()].map((aggregation) => aggregation.aggregatorID))
+  const verifyAggregators = new Map<string, string[]>()
+  const rewired = blocks.map((block) => {
+    if (block.kind !== "verify") return block
+    const original = block.depends_on ?? []
+    const replaced = [...new Set(original.flatMap((dependency) => writerToAggregators.get(dependency) ?? [dependency]))]
+    if (replaced.length === original.length && replaced.every((dependency, index) => dependency === original[index])) {
+      return block
+    }
+    verifyAggregators.set(block.id, replaced.filter((dependency) => aggregatorIDs.has(dependency)))
     return new WorkflowBlock({
       id: block.id,
       kind: block.kind,
-      depends_on: [...(block.depends_on ?? []), previous],
+      depends_on: replaced,
       instruction: block.instruction,
       worker_type: block.worker_type,
       required: block.required,
       report_to_parent: block.report_to_parent,
     })
   })
-  topologicalBlocks(serialized)
-  return serialized
+  topologicalBlocks(rewired)
+  return { blocks: rewired, aggregations, verifyAggregators }
 }
 
 function requireValidReviewRoutes(blocks: WorkflowBlock[]) {
-  blocks.filter((block) => block.kind === "review").forEach((block) => implementationReviewRoute(block, blocks))
+  blocks.filter((block) => block.kind === "review").forEach((block) => reviewWriterTopology(block, blocks))
 }
 
-function implementationReviewRoute(block: WorkflowBlock, blocks: WorkflowBlock[]) {
+interface ReviewWriterTopology {
+  implementations: WorkflowBlock[]
+  verification: WorkflowBlock
+}
+
+function reviewWriterTopology(block: WorkflowBlock, blocks: WorkflowBlock[]): ReviewWriterTopology | undefined {
   const implementations = blocks.filter(
     (candidate) => WRITER_KINDS.has(candidate.kind) && dependsTransitively(blocks, block.id, candidate.id),
   )
@@ -389,8 +468,7 @@ function implementationReviewRoute(block: WorkflowBlock, blocks: WorkflowBlock[]
       `Implementation review "${block.id}" requires exactly one verification ancestor; found ${verifications.length}`,
     )
   }
-  const verification = verifications[0]
-  if (!verification) throw new Error(`Implementation review "${block.id}" has no verification ancestor`)
+  const verification = verifications[0]!
   const verifiedImplementations = implementations.filter((candidate) =>
     dependsTransitively(blocks, verification.id, candidate.id),
   )
@@ -399,15 +477,27 @@ function implementationReviewRoute(block: WorkflowBlock, blocks: WorkflowBlock[]
       `Implementation review "${block.id}" requires its verification ancestor to depend on every implementation writer`,
     )
   }
-  const implementation = verifiedImplementations.find((candidate) =>
-    verifiedImplementations.every(
+  return { implementations: verifiedImplementations, verification }
+}
+
+function canonicalWriter(topology: ReviewWriterTopology, blocks: WorkflowBlock[]): WorkflowBlock | undefined {
+  return topology.implementations.find((candidate) =>
+    topology.implementations.every(
       (other) => other.id === candidate.id || dependsTransitively(blocks, candidate.id, other.id),
     ),
   )
+}
+
+function implementationReviewRoute(block: WorkflowBlock, blocks: WorkflowBlock[]) {
+  const topology = reviewWriterTopology(block, blocks)
+  if (!topology) return undefined
+  const implementation = canonicalWriter(topology, blocks)
   if (!implementation) {
+    // Unreachable for compiled graphs: aggregateParallelWriters injects an
+    // aggregator whenever no canonical writer exists.
     throw new Error(`Implementation review "${block.id}" has no canonical serialized implementation writer`)
   }
-  return { implementation, verification }
+  return { implementation, verification: topology.verification }
 }
 
 function dependsTransitively(