feat(workflow): tasked execution loop replaces role_pipeline

role_pipeline.toml is now analyst → architect → decomposer → loop(claim →
implement → review) → done. The decomposer (require_task_decompose) breaks
the design into a task graph; the implementer stage claims the next ready
task on entry (claim_task = true, kernel-driven) with writes auto-scoped to
the task's affected_paths and propose_scope as the operator-approved widen
valve. The loop is gated by the tasks_ready predicate.

Loop precedence rides the resolver's by-id transition ordering (review-1/2/3)
to avoid composite all_of/not conditions in the flat TOML schema: changes →
implementer, approved+tasks_ready → implementer, else → done.

TomlWorkflowLoader gains a claim_task stage field (→ claimTask metadata) and
extracts the metadata build into StageSection.toMetadata. RolePipelineWorkflowTest
rewritten for the tasked graph.
This commit is contained in:
2026-06-29 11:37:51 +04:00
parent a00bd4aade
commit e0f623a10c
3 changed files with 118 additions and 115 deletions
+56 -74
View File
@@ -1,13 +1,20 @@
# Role pipeline: analyst → architect → brief_echo → planner → implementer ⇄ reviewer # Role pipeline: analyst → architect → decomposer → ⟲(claim → implement review)⟲ → done
# #
# Each stage produces a typed artifact that the next stage `needs`, so work flows forward # The architect's design is broken into a task graph by the decomposer, then a single loop
# without a human relaying notes. The decision journal (pinned into every stage's context) # drains those tasks one at a time. Each implementer entry deterministically CLAIMS the next
# carries steering/approvals/verdicts across the whole run, so the reviewer sees the same # ready task (claim_task = true) — the kernel picks it, not the LLM — and injects that task's
# ground truth as the planner. # acceptance criteria + blockers as L0 context. While a task is claimed, file writes are
# auto-scoped to its affected_paths; the implementer widens scope only via propose_scope
# (operator-approved). The loop is gated by the tasks_ready predicate, so it exits when the
# board is drained.
# #
# The implementer⇄reviewer loop is gated by review_report.verdict: # Loop control depends on transition ORDERING. The resolver evaluates a stage's outgoing
# approved → done # transitions sorted by transition id and takes the first whose condition holds, so the
# changes_requested → back to implementer (capped by implementer.max_retries, then escalates) # reviewer edges are named review-1/2/3 to force this precedence:
# 1. review-1-changes (verdict ≠ approved) → implementer (fix the current task)
# 2. review-2-more (tasks_ready) → implementer (approved: claim next)
# 3. review-3-done (always_true, fallthrough) → done (approved + board empty)
# This avoids needing composite (all_of/not) conditions in the flat TOML schema.
# #
# Requires these artifact kinds in ~/.config/correx/config.toml (schemas under docs/schemas/): # Requires these artifact kinds in ~/.config/correx/config.toml (schemas under docs/schemas/):
# [[artifacts]] # [[artifacts]]
@@ -15,11 +22,7 @@
# [[artifacts]] # [[artifacts]]
# id = "design"; schema_path = "schemas/design.json"; llm_emitted = true # id = "design"; schema_path = "schemas/design.json"; llm_emitted = true
# [[artifacts]] # [[artifacts]]
# id = "impl_plan"; schema_path = "schemas/impl_plan.json"; llm_emitted = true
# [[artifacts]]
# id = "review_report"; schema_path = "schemas/review_report.json"; llm_emitted = true # id = "review_report"; schema_path = "schemas/review_report.json"; llm_emitted = true
# [[artifacts]]
# id = "brief_echo"; schema_path = "schemas/brief_echo.json"; llm_emitted = true
# #
# Prompt files (prompts/*.md, relative to this workflow) must exist for a real run. # Prompt files (prompts/*.md, relative to this workflow) must exist for a real run.
@@ -47,61 +50,42 @@ produces = [{ name = "design", kind = "design" }]
token_budget = 16384 token_budget = 16384
max_retries = 2 max_retries = 2
# 3a. Before planning: echo the analyst brief back as a structured artifact. # 3. Break the design into a task graph via task_decompose. require_task_decompose is a
# The brief_echo gate diffs the restatement against the original analysis; # deterministic kernel gate: stage_complete is blocked until task_decompose (or task_create
# on divergence (dropped requirements or hallucinated files) it fails the # for a trivial single-task request) returns. Each child carries acceptance_criteria and
# stage (retryable) so the pipeline cannot reach plan generation on a misread brief. # affected_paths — those drive the per-task L0 context and the per-task write scope below.
[[stages]] [[stages]]
id = "brief_echo" id = "decomposer"
prompt = "prompts/brief_echo.md" prompt = "prompts/task_planner.md"
needs = ["analysis"]
produces = [{ name = "brief_echo", kind = "brief_echo" }]
brief_echo = true
token_budget = 8192
max_retries = 2
# 3b. Break the design into ordered, verifiable steps.
[[stages]]
id = "planner"
prompt = "prompts/planner.md"
needs = ["design"] needs = ["design"]
produces = [{ name = "impl_plan", kind = "impl_plan" }] allowed_tools = ["file_read", "ShellTool", "task_search", "task_context", "task_decompose", "task_create"]
require_task_decompose = true
token_budget = 16384 token_budget = 16384
max_retries = 2 max_retries = 2
# 4. Implement the plan. Writes files (jailed to the workspace). The loop target — # 4. Implement one task. claim_task = true: on entry the kernel claims the next ready task
# max_retries here caps how many review→implement refinement rounds are allowed. # (or re-surfaces the one already claimed on a review bounce) and injects its acceptance
# Optional: a `writes` manifest (workspace-relative globs) hard-bounds where this # criteria as L0. Writes are auto-jailed to the claimed task's affected_paths; propose_scope
# stage may write — a FILE_WRITE outside it is blocked as scope creep. Left open # is the operator-approved valve to widen that scope mid-task.
# here because the targets are task-specific; a task-scoped workflow would set e.g.
# writes = ["core/sessions/**", "testing/sessions/**"]
# Optional: `static_analysis` runs deterministic tools (compiler/detekt/formatters)
# against the patch BEFORE the reviewer (role-reliability §5). A non-clean command
# fails this stage retryably with its output fed back verbatim, so only static-clean
# code reaches the reviewer — the reviewer never spends inference on what tools catch
# for free. Commands are workspace-specific, so left commented; a Kotlin/Gradle repo
# would set e.g.
# static_analysis = ["./gradlew compileKotlin -q", "./gradlew detekt -q"]
[[stages]] [[stages]]
id = "implementer" id = "implementer"
prompt = "prompts/implementer.md" prompt = "prompts/implementer.md"
needs = ["impl_plan"]
produces = [{ name = "patch", kind = "file_written" }] produces = [{ name = "patch", kind = "file_written" }]
claim_task = true
allowed_tools = [ allowed_tools = [
"file_read", "file_write", "file_edit", "ShellTool", "file_read", "file_write", "file_edit", "ShellTool", "propose_scope",
"task_create", "task_update", "task_context", "task_search", "task_create", "task_update", "task_context", "task_search",
] ]
# static_analysis = ["./gradlew compileKotlin -q", "./gradlew detekt -q"]
token_budget = 32768 token_budget = 32768
max_retries = 3 max_retries = 3
# 5. Review the patch against the plan AND the analyst's acceptance criteria. The reviewer # 5. Review the patch against the claimed task's acceptance criteria and the analyst's brief.
# needs `analysis` so it judges the diff against concrete, pre-stated criteria (§5 narrow # On approval the reviewer marks the task done via task_update — that drops it from the
# question) rather than whole files against taste. # active claim so the next implementer entry claims a fresh task and the loop advances.
[[stages]] [[stages]]
id = "reviewer" id = "reviewer"
prompt = "prompts/reviewer.md" prompt = "prompts/reviewer.md"
needs = ["patch", "impl_plan", "analysis"] needs = ["patch", "analysis"]
produces = [{ name = "review_report", kind = "review_report" }] produces = [{ name = "review_report", kind = "review_report" }]
allowed_tools = ["task_context", "task_update"] allowed_tools = ["task_context", "task_update"]
token_budget = 32768 token_budget = 32768
@@ -117,25 +101,18 @@ condition_type = "artifact_validated"
condition_artifact_id = "analysis" condition_artifact_id = "analysis"
[[transitions]] [[transitions]]
id = "architect-to-brief-echo" id = "architect-to-decomposer"
from = "architect" from = "architect"
to = "brief_echo" to = "decomposer"
condition_type = "artifact_validated" condition_type = "artifact_validated"
condition_artifact_id = "design" condition_artifact_id = "design"
# Only proceed to implementation once the decomposer actually produced ready tasks.
[[transitions]] [[transitions]]
id = "brief-echo-to-planner" id = "decomposer-to-implementer"
from = "brief_echo" from = "decomposer"
to = "planner"
condition_type = "artifact_validated"
condition_artifact_id = "brief_echo"
[[transitions]]
id = "planner-to-implementer"
from = "planner"
to = "implementer" to = "implementer"
condition_type = "artifact_validated" condition_type = "tasks_ready"
condition_artifact_id = "impl_plan"
[[transitions]] [[transitions]]
id = "implementer-to-reviewer" id = "implementer-to-reviewer"
@@ -144,20 +121,11 @@ to = "reviewer"
condition_type = "artifact_validated" condition_type = "artifact_validated"
condition_artifact_id = "patch" condition_artifact_id = "patch"
# --- verdict-gated loop exit / re-entry --- # --- verdict + task-board gated loop (id order = evaluation precedence, see header) ---
# 1. Changes requested → keep refining the current (still-claimed) task.
[[transitions]] [[transitions]]
id = "review-approved" id = "review-1-changes"
from = "reviewer"
to = "done"
condition_type = "artifact_field_equals"
condition_artifact_id = "review_report"
condition_field = "verdict"
condition_value = "approved"
condition_operator = "eq"
[[transitions]]
id = "review-changes-requested"
from = "reviewer" from = "reviewer"
to = "implementer" to = "implementer"
condition_type = "artifact_field_equals" condition_type = "artifact_field_equals"
@@ -165,3 +133,17 @@ condition_artifact_id = "review_report"
condition_field = "verdict" condition_field = "verdict"
condition_value = "approved" condition_value = "approved"
condition_operator = "neq" condition_operator = "neq"
# 2. Approved and tasks remain → claim the next one.
[[transitions]]
id = "review-2-more"
from = "reviewer"
to = "implementer"
condition_type = "tasks_ready"
# 3. Approved and the board is drained → done (fallthrough).
[[transitions]]
id = "review-3-done"
from = "reviewer"
to = "done"
condition_type = "always_true"
@@ -50,6 +50,7 @@ private data class StageSection(
@param:JsonProperty("ground_references") val groundReferences: Boolean = false, @param:JsonProperty("ground_references") val groundReferences: Boolean = false,
@param:JsonProperty("brief_echo") val briefEcho: Boolean = false, @param:JsonProperty("brief_echo") val briefEcho: Boolean = false,
@param:JsonProperty("require_task_decompose") val requireTaskDecompose: Boolean = false, @param:JsonProperty("require_task_decompose") val requireTaskDecompose: Boolean = false,
@param:JsonProperty("claim_task") val claimTask: Boolean = false,
) )
// condition fields flattened into the transition row // condition fields flattened into the transition row
@@ -119,15 +120,7 @@ class TomlWorkflowLoader(
maxTokens = s.tokenBudget, maxTokens = s.tokenBudget,
), ),
maxRetries = s.maxRetries, maxRetries = s.maxRetries,
metadata = buildMap { metadata = s.toMetadata(workflowDir),
s.prompt?.let { put("prompt", resolvePromptPath(it, workflowDir)) }
s.systemPrompt?.let { put("systemPrompt", resolvePromptPath(it, workflowDir)) }
if (s.requiresApproval) put("requiresApproval", "true")
if (s.injectArtifactKinds) put("injectArtifactKinds", "true")
if (s.groundReferences) put("groundReferences", "true")
if (s.briefEcho) put("briefEcho", "true")
if (s.requireTaskDecompose) put("requireTaskDecompose", "true")
},
) )
} }
@@ -164,6 +157,17 @@ class TomlWorkflowLoader(
return WorkflowGraph(id = id, stages = stageMap, transitions = transitionSet, start = startId) return WorkflowGraph(id = id, stages = stageMap, transitions = transitionSet, start = startId)
} }
private fun StageSection.toMetadata(workflowDir: Path?): Map<String, String> = buildMap {
prompt?.let { put("prompt", resolvePromptPath(it, workflowDir)) }
systemPrompt?.let { put("systemPrompt", resolvePromptPath(it, workflowDir)) }
if (requiresApproval) put("requiresApproval", "true")
if (injectArtifactKinds) put("injectArtifactKinds", "true")
if (groundReferences) put("groundReferences", "true")
if (briefEcho) put("briefEcho", "true")
if (requireTaskDecompose) put("requireTaskDecompose", "true")
if (claimTask) put("claimTask", "true")
}
@Suppress("ThrowsCount") @Suppress("ThrowsCount")
private fun validate( private fun validate(
start: String, start: String,
@@ -7,6 +7,7 @@ import com.correx.core.artifacts.kind.JsonSchemaProperty
import com.correx.core.events.types.StageId import com.correx.core.events.types.StageId
import com.correx.core.transitions.conditions.ArtifactFieldEquals import com.correx.core.transitions.conditions.ArtifactFieldEquals
import com.correx.core.transitions.conditions.FieldOperator import com.correx.core.transitions.conditions.FieldOperator
import com.correx.core.transitions.conditions.TasksReady
import org.junit.jupiter.api.Assumptions.assumeTrue import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.Test import org.junit.jupiter.api.Test
import java.nio.file.Path import java.nio.file.Path
@@ -29,8 +30,6 @@ class RolePipelineWorkflowTest {
private val registry = DefaultArtifactKindRegistry().apply { private val registry = DefaultArtifactKindRegistry().apply {
register(llmKind("analysis")) register(llmKind("analysis"))
register(llmKind("design")) register(llmKind("design"))
register(llmKind("brief_echo"))
register(llmKind("impl_plan"))
register(llmKind("review_report")) register(llmKind("review_report"))
} }
@@ -46,49 +45,67 @@ class RolePipelineWorkflowTest {
return null return null
} }
@Test private fun loadGraph() =
fun `shipped role_pipeline toml parses with the full role graph`() { repoFile("examples/workflows/role_pipeline.toml")?.let { TomlWorkflowLoader(registry).load(it) }
val path = repoFile("examples/workflows/role_pipeline.toml")
assumeTrue(path != null, "role_pipeline.toml not found from ${System.getProperty("user.dir")}")
val graph = TomlWorkflowLoader(registry).load(path!!) @Test
fun `shipped role_pipeline toml parses with the tasked execution graph`() {
val graph = loadGraph()
assumeTrue(graph != null, "role_pipeline.toml not found from ${System.getProperty("user.dir")}")
graph!!
assertEquals( assertEquals(
setOf("analyst", "architect", "brief_echo", "planner", "implementer", "reviewer"), setOf("analyst", "architect", "decomposer", "implementer", "reviewer"),
graph.stages.keys.map { it.value }.toSet(), graph.stages.keys.map { it.value }.toSet(),
) )
assertEquals(7, graph.transitions.size) assertEquals(7, graph.transitions.size)
// Forward chaining via needs. // The decomposer is the mandatory task-graph gate.
assertTrue(graph.stages[StageId("reviewer")]!!.needs.map { it.value }.containsAll(listOf("patch", "impl_plan"))) assertEquals("true", graph.stages[StageId("decomposer")]!!.metadata["requireTaskDecompose"])
// The implementer claims a task on entry.
assertEquals("true", graph.stages[StageId("implementer")]!!.metadata["claimTask"])
}
// Verdict-gated loop: approved → done, changes → implementer. @Test
val approved = graph.transitions.first { it.to == StageId("done") }.condition as ArtifactFieldEquals fun `loop is gated by verdict and the task board`() {
assertEquals(FieldOperator.EQ, approved.operator) val graph = loadGraph()
assertEquals("verdict", approved.field) assumeTrue(graph != null, "role_pipeline.toml not found")
val loop = graph.transitions graph!!
.first { it.from == StageId("reviewer") && it.to == StageId("implementer") }
.condition as ArtifactFieldEquals // Decomposer only advances once tasks exist.
assertEquals(FieldOperator.NEQ, loop.operator) val toImpl = graph.transitions
.first { it.from == StageId("decomposer") && it.to == StageId("implementer") }
assertEquals(TasksReady, toImpl.condition)
// Reviewer edges, in id (= evaluation) order: changes → impl, tasks_ready → impl, else → done.
val reviewerEdges = graph.transitions
.filter { it.from == StageId("reviewer") }
.sortedBy { it.id.value }
assertEquals(
listOf("review-1-changes", "review-2-more", "review-3-done"),
reviewerEdges.map { it.id.value },
)
// 1. changes requested (verdict != approved) loops back to the implementer.
val changes = reviewerEdges[0]
assertEquals(StageId("implementer"), changes.to)
assertEquals(FieldOperator.NEQ, (changes.condition as ArtifactFieldEquals).operator)
// 2. approved + tasks remain → claim the next.
assertEquals(StageId("implementer"), reviewerEdges[1].to)
assertEquals(TasksReady, reviewerEdges[1].condition)
// 3. approved + board drained → done (fallthrough).
assertEquals(StageId("done"), reviewerEdges[2].to)
} }
@Test @Test
fun `task tools are granted to the right stages`() { fun `task tools are granted to the right stages`() {
val path = repoFile("examples/workflows/role_pipeline.toml") val graph = loadGraph()
assumeTrue(path != null, "role_pipeline.toml not found from ${System.getProperty("user.dir")}") assumeTrue(graph != null, "role_pipeline.toml not found")
val graph = TomlWorkflowLoader(registry).load(path!!) graph!!
// analyst opens + searches (read-only on files); creation is approval-gated (T2). // implementer owns the lifecycle: claim/submit + the full file set + the scope valve.
assertEquals( val implTools = graph.stages[StageId("implementer")]!!.allowedTools
setOf("file_read", "ShellTool", "task_search", "task_context", "task_create"), assertTrue("propose_scope" in implTools, "implementer must be able to propose scope widening")
graph.stages[StageId("analyst")]!!.allowedTools, assertTrue(implTools.containsAll(setOf("file_write", "file_edit", "task_update")))
)
// implementer owns the lifecycle: claim/submit + the full file set.
assertEquals(
setOf("file_read", "file_write", "file_edit", "ShellTool") +
setOf("task_create", "task_update", "task_context", "task_search"),
graph.stages[StageId("implementer")]!!.allowedTools,
)
// reviewer reads the task and completes it on an approved verdict. // reviewer reads the task and completes it on an approved verdict.
assertEquals( assertEquals(
setOf("task_context", "task_update"), setOf("task_context", "task_update"),