From e63ff0ee7e434aa4eb6cb7a2d4309d8ee0cc94b8 Mon Sep 17 00:00:00 2001 From: karim Date: Fri, 16 Oct 2026 09:37:27 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(318):=20P1=20=E2=80=94=20routes=20on?= =?UTF-8?q?=20the=20source=20task,=20guards=20derived=20at=20materializati?= =?UTF-8?q?on?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A task may declare routes: when its own output matches (output. , single or combined), the tasks it names run. materializeSkillTasks β€” the form every run and every validity check uses β€” gives each routed task the guard its routes imply and the source as a dependency, so a fork is stored once, on the task that decides it. Routes are validated as stored: a route to a missing task or to itself, a task enabled by two sources or carrying its own when, routes that combine into no single-level guard, and a condition that is not a comparison on the source's output. Paths are checked against the source's output through the existing when-path check on the derived guards. A decision step's routes are checked against its questions through a per-operation registry, so the workflow model holds no decision-specific concept. Co-Authored-By: Claude Opus 5.5 --- ...18-a-fork-is-stored-where-it-is-decided.md | 70 ++++++ .../src/__tests__/routes.test.ts | 201 ++++++++++++++++++ .../src/scheduling/decisionRouteChecks.ts | 80 +++++++ .../src/scheduling/graphValidation.ts | 18 ++ .../src/scheduling/routes.ts | 199 +++++++++++++++++ .../src/skillValidity/skillValidity.ts | 25 ++- .../schemas/src/operations/workflow/task.ts | 22 ++ 7 files changed, 611 insertions(+), 4 deletions(-) create mode 100644 docs/plans/318-a-fork-is-stored-where-it-is-decided.md create mode 100644 packages/cybernetic-runtime/src/__tests__/routes.test.ts create mode 100644 packages/cybernetic-runtime/src/scheduling/decisionRouteChecks.ts create mode 100644 packages/cybernetic-runtime/src/scheduling/routes.ts diff --git a/docs/plans/318-a-fork-is-stored-where-it-is-decided.md b/docs/plans/318-a-fork-is-stored-where-it-is-decided.md new file mode 100644 index 00000000..71768a61 --- /dev/null +++ b/docs/plans/318-a-fork-is-stored-where-it-is-decided.md @@ -0,0 +1,70 @@ +# Plan 318 β€” A fork is stored where it is decided: routes on the source task + +**Status:** 🌱 design Β· **Builds on:** Plan 316 (the decision step), Plan 317 (agents propose, operators approve) + +## 1. Problem + +A workflow branches by putting a `when` on each task that may be skipped. The condition lives on the _target_, so one fork β€” "billing goes here, technical there, anything unclear to an agent" β€” is stored as several unrelated guards scattered over the tasks it enables. Nothing in the stored workflow says they belong together. + +That costs three things. + +- **A fork cannot be edited as a fork.** A decision task is authoring vocabulary: assembly expands it into an operation task and hand-shaped guards on its targets, and the decision is gone. Adding one answer to a live triage skill means writing a correct `allOf` guard on a new task, keeping the confidence clause consistent with its siblings and wiring `dependsOn` by hand. Helmsman did it, live, in about eight attempts, succeeding only because each validator named the next mistake. +- **The pieces drift.** Rename a choice option and the guard that reads it still compiles; remove a routed task and nothing notices the branch that no longer goes anywhere. Each guard is checked alone. +- **Branching is not a first-class thing anywhere else.** An API call that should send `status 409` one way and `200` another, an agent score that should gate a retry β€” each is the same scattered-guard pattern, written by hand. + +## 2. Decisions + +**D1 β€” Routes live on the source task, for any task.** A task may declare where its outcome sends the run: + +``` +routes: [ + { when: "output.answers.team.value == 'billing'", enable: ["billing_responder"] }, + { when: { anyOf: ["output.answers.team.value == 'unclear'", + "output.answers.team.decided == false"] }, enable: ["escalate"] } +] +``` + +`when` is a predicate over the source task's own output, in the dialect `poll.until` already uses (`output. `, single or `anyOf` / `allOf`). `enable` names the tasks the route turns on. Nothing about it is decision-specific: an operation, an agent or a human task can route. _Rejected:_ a `decision` field on the workflow task β€” it would carry a concept that only means something for one operation into every consumer of the workflow format. + +**D2 β€” The target's guard is derived, never stored.** `materializeSkillTasks` β€” the derivation every run and every validity check already goes through, and the only form ever executed β€” gives each routed task the `when` its routes imply and adds the source to its `dependsOn`. The stored workflow keeps the routes and nothing else. A routed task may not also declare a `when` of its own. A task is enabled by routes of **one** source; several routes of that source enabling it combine as `anyOf`, and a combination that has no single-level form is refused (the `when` grammar has no nesting and no negation). _Rejected:_ an `otherwise` branch β€” it needs a negation the grammar does not have, and the decision case covers its need with an explicit route. + +**D3 β€” The decision task lowers to routes.** Compose-skill's `decision` task becomes an `ai.decision.decide` operation task carrying `routes`, built by the same rules Plan 316 uses today (a route on a confident answer adds `decided == true`; a task that is also the fallback drops it). Nothing decision-specific is stored. Checks that only make sense for one operation β€” a route naming an option its question does not offer, a score threshold outside the rubric β€” are contributed by that operation through a small route-check registry keyed by operation id, and run alongside the generic checks. _Rejected:_ storing the authored decision draft beside the lowered form β€” two forms of one definition, which drift. + +**D4 β€” Routes are checked as a whole.** On the materialized workflow, validation adds: every `enable` names a task; no target is enabled by two sources; a routed target declares no `when` of its own; every route's path exists in the source's output (the `when_output_path_unknown` check, read from the source instead of the target); and the derived dependencies keep the graph acyclic. Each failure is a structured diagnostic naming the route. + +**D5 β€” A route is edited as a route.** `workflow.manage.patch` stages typed changes for `/tasks/{task}/routes/-`, `/tasks/{task}/routes/{i}` and `/tasks/{task}/routes/{i}/enable`, so a proposal reads "triage: when `answers.team.value == 'sales'` β†’ enable `sales_responder`". Adding a branch is one patch: add the task, add the route. Removing a task that a route enables is refused with the route that still points at it. + +**D6 β€” Existing guards stay what they are.** An authored `when` on a target remains a legitimate construct β€” a condition the target itself imposes β€” and decision skills assembled before this plan keep their guards and keep running. They are not rewritten; they are simply not route-editable until re-composed. This is two constructs with two meanings, not a compatibility layer. + +**D7 β€” The fork is visible where it is authored.** The skill designer draws routes as labelled edges from the source to each target; a routed target's inspector shows "enabled by `triage` when …" from the derived guard. The run surface already reads the materialized workflow and shows derived guards unchanged. + +## 3. Affected packages and contracts + +`packages/schemas` (`routes` on the workflow task, the draft lowering, typed staged ops for routes), `packages/cybernetic-runtime` (`materializeSkillTasks` derivation, route validation, the route-check registry), `apps/aflow-orchestrator` (assembler, patch-to-proposal conversion, proposal application), `packages/web-product` (designer edges and inspector), `packages/platform-artifacts` (compose-skill's description of routes). + +Stored workflows gain an optional `routes` field. No existing workflow changes shape. + +## 4. Phases + +### P1 β€” Routes in the model + +`routes` on the workflow task; derivation in `materializeSkillTasks`; the D4 checks; the route-check registry with the decision operation's checks moved into it. Exit: a hand-written workflow with routes runs the right branch for each output, and each D4 violation is refused with a diagnostic naming the route. + +### P2 β€” Compose-skill lowers to routes + +The `decision` draft task becomes an operation task with `routes`; compose-skill's draft schema and its one line of guidance describe routes rather than guards. Exit: the live triage request from Plan 316 produces a skill whose stored form has routes on the decision and no guards on the responders, and runs every branch as before. + +### P3 β€” Editing a route + +Typed staged ops for route add, remove and replace; proposal text that reads as a route; the refusal for removing a still-enabled task. Exit: asking Helmsman to "send refund requests to a new refund handler" produces one proposal that adds the task and the route, in at most two attempts. + +### P4 β€” The designer + +Route edges and the routed-target inspector. Exit: the triage skill shows its fork as one source with labelled edges. + +## 5. How it is proven + +- The same skill stored with routes and with hand-written guards runs identical branches for the same inputs. +- A route naming a task that does not exist, a path the source does not emit, or a target enabled by two sources is refused at validation, not skipped at run time. +- Renaming a choice option that a route reads is refused by the decision route check. +- Helmsman's refund edit is one proposal, reviewable as "add task, add route". diff --git a/packages/cybernetic-runtime/src/__tests__/routes.test.ts b/packages/cybernetic-runtime/src/__tests__/routes.test.ts new file mode 100644 index 00000000..4fc39a98 --- /dev/null +++ b/packages/cybernetic-runtime/src/__tests__/routes.test.ts @@ -0,0 +1,201 @@ +import { describe, expect, it } from 'vitest'; +import type { WorkflowTask } from '@aflow/schemas'; +import { + materializeAndValidateSkillConfig, + materializeSkillTasks, +} from '../skillValidity/skillValidity.js'; +import { computeReadyTasksWithWhen } from '../scheduling/graph.js'; + +const agent = (taskId: string, extra: Partial = {}): WorkflowTask => ({ + taskId, + name: taskId, + goal: `Handle ${taskId}.`, + ...extra, +}); + +function triage(routes: WorkflowTask['routes'], questions?: Record): WorkflowTask { + return { + taskId: 'triage', + name: 'Triage', + goal: 'Run ai.decision.decide.', + type: 'operation', + operation: 'ai.decision.decide', + inputTemplate: { + state: 'A ticket', + questions: questions ?? { + team: { + type: 'choice', + options: { billing: null, technical: null, unclear: null }, + minConfidence: 0.7, + }, + severity: { type: 'score', levels: ['minor', 'major', 'critical'] }, + }, + }, + ...(routes ? { routes } : {}), + }; +} + +const ROUTES: WorkflowTask['routes'] = [ + { + when: { + allOf: ["output.answers.team.value == 'billing'", 'output.answers.team.decided == true'], + }, + enable: ['billing'], + }, + { when: { expression: "output.answers.team.value == 'unclear'" }, enable: ['escalate'] }, + { when: { expression: 'output.answers.team.decided == false' }, enable: ['escalate'] }, +]; + +const workflow = (routes = ROUTES, extra: WorkflowTask[] = []) => [ + triage(routes), + agent('billing'), + agent('escalate'), + ...extra, +]; + +function errors(tasks: WorkflowTask[]): string[] { + return materializeAndValidateSkillConfig({ tasks }).validity.diagnostics.map((d) => d.code); +} + +describe('routes are stored on the source and guard their targets when materialized', () => { + it('derives each target’s guard and its dependency on the source', () => { + const byId = new Map(materializeSkillTasks(workflow()).map((t) => [t.taskId, t])); + expect(byId.get('billing')).toMatchObject({ + when: { + allOf: [ + "tasks.triage.output.answers.team.value == 'billing'", + 'tasks.triage.output.answers.team.decided == true', + ], + }, + dependsOn: ['triage'], + }); + expect(byId.get('escalate')?.when).toEqual({ + anyOf: [ + "tasks.triage.output.answers.team.value == 'unclear'", + 'tasks.triage.output.answers.team.decided == false', + ], + onMissingRef: 'skip', + }); + }); + + it('is a valid skill', () => { + expect(errors(workflow())).toEqual([]); + }); + + it('runs the branch the answer names, and only that branch', () => { + const tasks = materializeSkillTasks(workflow()); + const decided = (value: string, isDecided: boolean) => + computeReadyTasksWithWhen(tasks, new Set(['triage']), new Set(), { + statuses: new Map([['triage', 'succeeded']]), + outputs: new Map([ + ['triage', { answers: { team: { type: 'choice', value, decided: isDecided } } }], + ]), + }); + + const confident = decided('billing', true); + expect(confident.ready.map((t) => t.taskId)).toEqual(['billing']); + expect(confident.skipped.map((s) => s.task.taskId)).toEqual(['escalate']); + + const unsure = decided('billing', false); + expect(unsure.ready.map((t) => t.taskId)).toEqual(['escalate']); + }); +}); + +describe('routes are checked as a whole', () => { + it('refuses a route to a task that does not exist', () => { + expect( + errors( + workflow([ + { when: { expression: 'output.answers.team.decided == false' }, enable: ['ghost'] }, + ]), + ), + ).toContain('route_target_unknown'); + }); + + it('refuses a route to its own task', () => { + expect( + errors( + workflow([ + { when: { expression: 'output.answers.team.decided == false' }, enable: ['triage'] }, + ]), + ), + ).toContain('route_target_self'); + }); + + it('refuses a task enabled by two sources', () => { + const other = agent('other', { + routes: [{ when: { expression: "output.status == 'done'" }, enable: ['billing'] }], + }); + expect(errors(workflow(ROUTES, [other]))).toContain('route_target_multiple_sources'); + }); + + it('refuses a routed task that carries its own when', () => { + const tasks = workflow().map((t) => + t.taskId === 'billing' + ? { + ...t, + when: { + expression: "tasks.triage.status == 'succeeded'", + onMissingRef: 'skip' as const, + }, + } + : t, + ); + expect(errors(tasks)).toContain('route_target_has_when'); + }); + + it('refuses routes to one task that combine into no single-level guard', () => { + expect( + errors( + workflow([ + ROUTES[0]!, + { when: { expression: 'output.answers.team.decided == false' }, enable: ['billing'] }, + ]), + ), + ).toContain('route_guard_inexpressible'); + }); + + it('refuses a condition that is not a comparison on the source’s output', () => { + expect( + errors( + workflow([ + { when: { expression: "tasks.triage.status == 'succeeded'" }, enable: ['billing'] }, + ]), + ), + ).toContain('route_unsupported_when'); + }); + + it('refuses a route on a path the source never emits', () => { + expect( + errors( + workflow([ + { when: { expression: "output.answers.team.valeu == 'billing'" }, enable: ['billing'] }, + ]), + ), + ).toContain('when_output_path_unknown'); + }); +}); + +describe('a decision step’s routes are checked against its questions', () => { + const route = (expression: string) => [{ when: { expression }, enable: ['billing'] }]; + + it('refuses a route on a question the step does not ask', () => { + expect(errors(workflow(route("output.answers.department.value == 'billing'")))).toContain( + 'route_decision_unknown_question', + ); + }); + + it('refuses a route on an option the question does not offer, naming the options', () => { + const { validity } = materializeAndValidateSkillConfig({ + tasks: workflow(route("output.answers.team.value == 'sales'")), + }); + const finding = validity.diagnostics.find((d) => d.code === 'route_decision_unknown_option'); + expect(finding?.detail).toContain('billing, technical, unclear'); + }); + + it('refuses a score threshold outside the rubric', () => { + expect(errors(workflow(route('output.answers.severity.value >= 3')))).toContain( + 'route_decision_threshold_out_of_range', + ); + }); +}); diff --git a/packages/cybernetic-runtime/src/scheduling/decisionRouteChecks.ts b/packages/cybernetic-runtime/src/scheduling/decisionRouteChecks.ts new file mode 100644 index 00000000..2f6256cd --- /dev/null +++ b/packages/cybernetic-runtime/src/scheduling/decisionRouteChecks.ts @@ -0,0 +1,80 @@ +/** + * A decision step's routes, checked against the questions it asks. + * + * A route that names an option no question offers, or a score past the top of + * its rubric, compiles and then never fires; checking it here turns a branch + * that silently goes nowhere into a diagnostic naming the options that exist. + */ +import { predicateExpressions, type WorkflowTask } from '@aflow/schemas'; +import { parseOutputComparison } from './graph.js'; +import type { GraphValidationError } from './graphValidation.js'; + +interface QuestionShape { + type?: unknown; + options?: unknown; + levels?: unknown; +} + +function literalQuestions(task: WorkflowTask): Record | undefined { + const source = task.inputTemplate ?? task.inputs; + const questions = source?.['questions']; + return questions !== null && typeof questions === 'object' && !Array.isArray(questions) + ? (questions as Record) + : undefined; +} + +export function checkDecisionRoutes(task: WorkflowTask): GraphValidationError[] { + const questions = literalQuestions(task); + if (!questions) return []; + const errors: GraphValidationError[] = []; + + (task.routes ?? []).forEach((route, ri) => { + const field = `routes[${String(ri)}].when`; + for (const expression of predicateExpressions(route.when)) { + const parsed = parseOutputComparison(expression.trim()); + const [answers, name, property] = parsed?.field.split('.') ?? []; + if (!parsed || answers !== 'answers' || name === undefined) continue; + const question = questions[name]; + if (!question) { + errors.push({ + kind: 'route_decision_unknown_question', + taskIds: [task.taskId], + field, + detail: `Task "${task.taskId}" routes on "${name}", which is not one of its questions: ${Object.keys(questions).join(', ')}.`, + }); + continue; + } + if (property !== 'value') continue; + const rhs = parsed.rhs.value; + if (question.type === 'choice' && question.options && typeof question.options === 'object') { + const options = Object.keys(question.options); + if (typeof rhs !== 'string' || !options.includes(rhs)) { + errors.push({ + kind: 'route_decision_unknown_option', + taskIds: [task.taskId], + field, + detail: `Task "${task.taskId}" routes on "${name}" == ${JSON.stringify(rhs)}, which "${name}" never answers. Its options are: ${options.join(', ')}.`, + }); + } + } else if (question.type === 'yes_no' && typeof rhs !== 'boolean') { + errors.push({ + kind: 'route_decision_unknown_option', + taskIds: [task.taskId], + field, + detail: `Task "${task.taskId}" routes on "${name}", a yes_no, compared with ${JSON.stringify(rhs)}; compare it with true or false.`, + }); + } else if (question.type === 'score' && Array.isArray(question.levels)) { + const top = question.levels.length - 1; + if (typeof rhs !== 'number' || rhs < 0 || rhs > top) { + errors.push({ + kind: 'route_decision_threshold_out_of_range', + taskIds: [task.taskId], + field, + detail: `Task "${task.taskId}" routes on "${name}" against ${JSON.stringify(rhs)}, but "${name}" scores from 0 to ${String(top)}.`, + }); + } + } + } + }); + return errors; +} diff --git a/packages/cybernetic-runtime/src/scheduling/graphValidation.ts b/packages/cybernetic-runtime/src/scheduling/graphValidation.ts index 1c462fa2..56c41f17 100644 --- a/packages/cybernetic-runtime/src/scheduling/graphValidation.ts +++ b/packages/cybernetic-runtime/src/scheduling/graphValidation.ts @@ -39,6 +39,24 @@ export interface GraphValidationError { | 'unsupported_when_expression' /** A `when` reads an output path the referenced task's output schema does not have. */ | 'when_output_path_unknown' + /** A route's condition is not a comparison on its own task's output. */ + | 'route_unsupported_when' + /** A route enables its own task. */ + | 'route_target_self' + /** A route enables a task that does not exist. */ + | 'route_target_unknown' + /** A task is enabled by routes of more than one task. */ + | 'route_target_multiple_sources' + /** A routed task also declares a `when` of its own. */ + | 'route_target_has_when' + /** Routes to one task combine into no single-level guard. */ + | 'route_guard_inexpressible' + /** A decision route names a question the step does not ask. */ + | 'route_decision_unknown_question' + /** A decision route compares an answer with a value the question never gives. */ + | 'route_decision_unknown_option' + /** A decision route's score threshold lies outside the rubric. */ + | 'route_decision_threshold_out_of_range' | 'duplicate_task_id' | 'reserved_task_id' | 'binding_dangling_task_ref' diff --git a/packages/cybernetic-runtime/src/scheduling/routes.ts b/packages/cybernetic-runtime/src/scheduling/routes.ts new file mode 100644 index 00000000..5ea0e244 --- /dev/null +++ b/packages/cybernetic-runtime/src/scheduling/routes.ts @@ -0,0 +1,199 @@ +/** + * Routes: a fork stored on the task that decides it. + * + * A task's `routes` say which tasks its outcome enables. The enabled tasks' + * guards are derived here, at materialization β€” the form every run and every + * validity check uses β€” so the stored workflow holds the fork once, on its + * source, and the guards cannot drift from it. + */ +import { + DECISION_OPERATION_ID, + predicateCombinator, + predicateExpressions, + type WorkflowPollUntil, + type WorkflowTask, + type WorkflowTaskRoute, + type WorkflowWhen, +} from '@aflow/schemas'; +import { parseOutputComparison } from './graph.js'; +import type { GraphValidationError } from './graphValidation.js'; +import { checkDecisionRoutes } from './decisionRouteChecks.js'; + +/** + * Checks only one operation can make about its own routes β€” that a route names + * an answer the operation can give. Keyed by operation id, so the workflow + * model carries no operation-specific concept. + */ +const ROUTE_CHECKS: Readonly GraphValidationError[]>> = { + [DECISION_OPERATION_ID]: checkDecisionRoutes, +}; + +/** `output. …` read from the source, rewritten to how the target reads it. */ +function fromSource(sourceId: string, expression: string): string { + return `tasks.${sourceId}.${expression.trim()}`; +} + +/** + * The guard a task enabled by these routes of one source runs under, or null + * when the combination has no single-level form. Several routes combine as + * `anyOf`; a route that is itself an `allOf` combines with nothing, because the + * `when` grammar has no nesting. + */ +export function routeGuardFor( + sourceId: string, + routes: readonly WorkflowTaskRoute[], +): WorkflowWhen | null { + const first = routes[0]; + if (first === undefined) return null; + if (routes.length === 1) return toWhen(sourceId, first.when); + if ( + routes.some( + (r) => predicateCombinator(r.when) === 'allOf' && predicateExpressions(r.when).length > 1, + ) + ) { + return null; + } + const expressions = [ + ...new Set( + routes.flatMap((r) => predicateExpressions(r.when)).map((e) => fromSource(sourceId, e)), + ), + ]; + return expressions.length === 1 + ? { expression: expressions[0]!, onMissingRef: 'skip' } + : { anyOf: expressions, onMissingRef: 'skip' }; +} + +function toWhen(sourceId: string, predicate: WorkflowPollUntil): WorkflowWhen { + const expressions = predicateExpressions(predicate).map((e) => fromSource(sourceId, e)); + switch (predicateCombinator(predicate)) { + case 'expression': + return { expression: expressions[0]!, onMissingRef: 'skip' }; + case 'anyOf': + return { anyOf: expressions, onMissingRef: 'skip' }; + case 'allOf': + return { allOf: expressions, onMissingRef: 'skip' }; + } +} + +/** Every task some route enables, with its one source and the routes that enable it. */ +function claims( + tasks: readonly WorkflowTask[], +): Map> { + const byTarget = new Map>(); + for (const task of tasks) { + for (const route of task.routes ?? []) { + for (const target of route.enable) { + const list = byTarget.get(target) ?? []; + list.push({ source: task.taskId, route }); + byTarget.set(target, list); + } + } + } + return byTarget; +} + +/** + * Give each routed task the guard its routes imply and its source as a + * dependency. A target the routes cannot guard β€” claimed by two sources, + * carrying its own `when`, or enabled by an inexpressible combination β€” is + * left as stored; validation refuses that workflow before it can run. + */ +export function deriveRouteGuards(tasks: WorkflowTask[]): WorkflowTask[] { + const byTarget = claims(tasks); + if (byTarget.size === 0) return tasks; + return tasks.map((task) => { + const claimed = byTarget.get(task.taskId); + if (!claimed || task.when !== undefined) return task; + const sources = new Set(claimed.map((c) => c.source)); + if (sources.size !== 1) return task; + const [source] = sources; + const guard = routeGuardFor( + source!, + claimed.map((c) => c.route), + ); + if (!guard) return task; + return { + ...task, + when: guard, + dependsOn: [...new Set([...(task.dependsOn ?? []), source!])], + }; + }); +} + +/** The route rules, read on the workflow as stored β€” before any guard is derived. */ +export function validateRoutes(tasks: readonly WorkflowTask[]): GraphValidationError[] { + const errors: GraphValidationError[] = []; + const byId = new Map(tasks.map((t) => [t.taskId, t])); + + for (const task of tasks) { + (task.routes ?? []).forEach((route, ri) => { + for (const expression of predicateExpressions(route.when)) { + if (!parseOutputComparison(expression.trim())) { + errors.push({ + kind: 'route_unsupported_when', + taskIds: [task.taskId], + field: `routes[${String(ri)}].when`, + detail: + `Task "${task.taskId}" routes[${String(ri)}] has the condition "${expression}", which is not a comparison on its own output. ` + + "Write output. , e.g. output.answers.team.value == 'billing'.", + }); + } + } + for (const target of route.enable) { + if (target === task.taskId) { + errors.push({ + kind: 'route_target_self', + taskIds: [task.taskId], + field: `routes[${String(ri)}].enable`, + detail: `Task "${task.taskId}" routes[${String(ri)}] enables itself; a route enables the tasks after it.`, + }); + } else if (!byId.has(target)) { + errors.push({ + kind: 'route_target_unknown', + taskIds: [task.taskId, target], + field: `routes[${String(ri)}].enable`, + detail: `Task "${task.taskId}" routes[${String(ri)}] enables "${target}", which is not a task in this workflow.`, + }); + } + } + }); + errors.push(...(task.operation ? (ROUTE_CHECKS[task.operation]?.(task) ?? []) : [])); + } + + for (const [target, claimed] of claims(tasks)) { + const targetTask = byId.get(target); + if (!targetTask) continue; + const sources = [...new Set(claimed.map((c) => c.source))]; + if (sources.length > 1) { + errors.push({ + kind: 'route_target_multiple_sources', + taskIds: [target, ...sources], + detail: `Task "${target}" is enabled by routes of ${sources.map((s) => `"${s}"`).join(' and ')}. A task is enabled by the routes of one task; give each source its own task.`, + }); + continue; + } + if (targetTask.when !== undefined) { + errors.push({ + kind: 'route_target_has_when', + taskIds: [target, sources[0]!], + detail: `Task "${target}" is enabled by routes of "${sources[0]!}" and also declares a when. Its guard comes from those routes; remove its when.`, + }); + continue; + } + if ( + routeGuardFor( + sources[0]!, + claimed.map((c) => c.route), + ) === null + ) { + errors.push({ + kind: 'route_guard_inexpressible', + taskIds: [target, sources[0]!], + detail: + `Task "${target}" is enabled by several routes of "${sources[0]!}", and one of them is an allOf. ` + + 'Routes to one task combine as anyOf, and a condition cannot nest; give the allOf route its own task, or split it.', + }); + } + } + return errors; +} diff --git a/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts b/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts index d81d8994..2a75f9bf 100644 --- a/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts +++ b/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts @@ -27,6 +27,7 @@ import { import { validateWorkflowGraph, type GraphValidationError } from '../scheduling/graphValidation.js'; import { entryRunInputSlots } from '../parentTaskInputs.js'; import { deriveOpBoundProducerShapes } from '../scheduling/deriveOpBoundShapes.js'; +import { deriveRouteGuards, validateRoutes } from '../scheduling/routes.js'; import { bindingStaticSchema } from '../scheduling/bindingStaticSchema.js'; import { validateAgentOpTaskOnlyTools, @@ -154,9 +155,9 @@ export function materializeAndValidateSkillConfig( // convert it into a blocking diagnostic so the verdict stays uniform. let materializedTasks: WorkflowTask[]; try { - materializedTasks = deriveOpBoundProducerShapes(input.tasks); + materializedTasks = deriveRouteGuards(deriveOpBoundProducerShapes(input.tasks)); } catch (err) { - materializedTasks = input.tasks; + materializedTasks = deriveRouteGuards(input.tasks); diagnostics.push({ code: 'derivation_conflict', dimension: 'op_input', @@ -165,6 +166,22 @@ export function materializeAndValidateSkillConfig( }); } + // Routes are read as stored: once their guards are derived, a target's own + // `when` and a derived one look alike. + const opByStoredTaskId = new Map(input.tasks.map((t) => [t.taskId, t.operation] as const)); + try { + for (const err of validateRoutes(input.tasks)) { + diagnostics.push(graphErrorToDiagnostic(err, opByStoredTaskId)); + } + } catch (err) { + diagnostics.push({ + code: 'validation_error', + dimension: 'parse', + severity: 'error', + detail: `Route validation could not complete: ${err instanceof Error ? err.message : String(err)}`, + }); + } + // 2. Graph + op-input contract validation against the derived tasks. Guarded: // a malformed task object that slips past Β§0 must not crash the gate. const opByTaskId = new Map(materializedTasks.map((t) => [t.taskId, t.operation] as const)); @@ -250,9 +267,9 @@ export function materializeAndValidateSkillConfig( export function materializeSkillTasks(tasks: WorkflowTask[]): WorkflowTask[] { if (!Array.isArray(tasks)) return tasks; try { - return deriveOpBoundProducerShapes(tasks); + return deriveRouteGuards(deriveOpBoundProducerShapes(tasks)); } catch { - return tasks; + return deriveRouteGuards(tasks); } } diff --git a/packages/schemas/src/operations/workflow/task.ts b/packages/schemas/src/operations/workflow/task.ts index d9c2a888..e9e2a6f8 100644 --- a/packages/schemas/src/operations/workflow/task.ts +++ b/packages/schemas/src/operations/workflow/task.ts @@ -161,6 +161,26 @@ export const WorkflowPollUntilSchema = z.union([ ]); export type WorkflowPollUntil = z.infer; +/** + * Where a task's outcome sends the run: when the task's own output matches + * `when` β€” read as `output. `, single or combined β€” the + * tasks in `enable` run. Their guards are derived from the route when the + * workflow is materialized; they declare no `when` of their own. + */ +export const WorkflowTaskRouteSchema = z + .object({ + when: WorkflowPollUntilSchema.describe( + "Condition on this task's own output, e.g. { expression: \"output.answers.team.value == 'billing'\" }.", + ), + enable: z + .array(z.string().min(1).max(64)) + .min(1) + .max(20) + .describe('The taskIds this route turns on. Each is enabled by routes of one task only.'), + }) + .strict(); +export type WorkflowTaskRoute = z.infer; + export const WorkflowTaskPollSchema = z .object({ /** @@ -581,6 +601,8 @@ export const WorkflowTaskSchema = z when: WorkflowWhenSchema.optional(), + routes: z.array(WorkflowTaskRouteSchema).min(1).max(50).optional(), + poll: WorkflowTaskPollSchema.optional(), outputProjection: WorkflowTaskOutputProjectionSchema.optional(), From 7206753f386f2a46141eed649c8f0d6e87f2e777 Mon Sep 17 00:00:00 2001 From: karim Date: Fri, 16 Oct 2026 09:42:41 +0200 Subject: [PATCH 2/4] feat(318): compose-skill lowers a decision to routes on the decision step The decision draft task becomes an ai.decision.decide operation task carrying routes over its own output; its targets only depend on it, and their guards are derived at materialization. Draft agent, operation and human tasks accept routes, and the assembler carries them onto the workflow. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/assembleDecisionTask.test.ts | 34 +++++++++- .../handlers/inlineOps/assembleWorkflow.ts | 3 + .../src/__tests__/composeDecisionTask.test.ts | 60 +++++++++-------- .../src/cybernetic/composeDecisionTask.ts | 64 ++++++++++++------- .../schemas/src/cybernetic/composeSkill.ts | 13 ++++ 5 files changed, 122 insertions(+), 52 deletions(-) diff --git a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/assembleDecisionTask.test.ts b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/assembleDecisionTask.test.ts index 0bee3e4b..a4390593 100644 --- a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/assembleDecisionTask.test.ts +++ b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/assembleDecisionTask.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from 'vitest'; import { WorkflowAssemblyInputSchema } from '@aflow/schemas'; -import { materializeAndValidateSkillConfig } from '@aflow/cybernetic-runtime'; +import { + materializeAndValidateSkillConfig, + materializeSkillTasks, +} from '@aflow/cybernetic-runtime'; import { assembleWorkflow } from '../assembleWorkflow.js'; const agent = (taskId: string, extra: Record = {}) => ({ @@ -89,8 +92,34 @@ describe('assembling a draft with a decision task', () => { }); }); - it('guards each routed task with its route and the fallback with the abstention', () => { + it('stores the fork on the decision, and its targets carry no guard of their own', () => { const byId = new Map(assemble().workflow.tasks.map((t) => [t.taskId, t])); + expect(byId.get('triage')?.routes).toEqual([ + { + when: { + allOf: ["output.answers.team.value == 'billing'", 'output.answers.team.decided == true'], + }, + enable: ['billing'], + }, + { + when: { + allOf: [ + "output.answers.team.value == 'technical'", + 'output.answers.team.decided == true', + ], + }, + enable: ['technical'], + }, + { when: { expression: 'output.answers.team.decided == false' }, enable: ['escalate'] }, + ]); + expect(byId.get('billing')?.when).toBeUndefined(); + expect(byId.get('escalate')?.dependsOn).toEqual(['triage']); + }); + + it('guards each routed task with its route and the fallback with the abstention once materialized', () => { + const byId = new Map( + materializeSkillTasks(assemble().workflow.tasks).map((t) => [t.taskId, t]), + ); expect(byId.get('billing')?.when).toEqual({ allOf: [ "tasks.triage.output.answers.team.value == 'billing'", @@ -102,7 +131,6 @@ describe('assembling a draft with a decision task', () => { expression: 'tasks.triage.output.answers.team.decided == false', onMissingRef: 'skip', }); - expect(byId.get('escalate')?.dependsOn).toEqual(['triage']); }); it('reads a run input directly, declaring it on the workflow', () => { diff --git a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/assembleWorkflow.ts b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/assembleWorkflow.ts index f2dfcf85..59a768c6 100644 --- a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/assembleWorkflow.ts +++ b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/assembleWorkflow.ts @@ -441,6 +441,7 @@ function lowerTask(task: TaskGraphDraftTask, opts: LowerOpts): WorkflowTask { ...(inputContract ? { inputContract } : {}), ...(produces ? { produces } : {}), ...(task.when ? { when: lowerWhen(task.when) } : {}), + ...(task.routes ? { routes: task.routes } : {}), ...(context ? { context } : {}), ...(outputContract ? { outputContract } : {}), }; @@ -466,6 +467,7 @@ function lowerTask(task: TaskGraphDraftTask, opts: LowerOpts): WorkflowTask { ...(task.retryability !== undefined ? { retryability: task.retryability } : {}), ...(task.maxAttempts !== undefined ? { maxAttempts: task.maxAttempts } : {}), ...(task.when ? { when: lowerWhen(task.when) } : {}), + ...(task.routes ? { routes: task.routes } : {}), ...(outputContract ? { outputContract } : {}), }; } @@ -487,6 +489,7 @@ function lowerTask(task: TaskGraphDraftTask, opts: LowerOpts): WorkflowTask { ...(dependsOn.length > 0 ? { dependsOn } : {}), ...(produces ? { produces } : {}), ...(task.when ? { when: lowerWhen(task.when) } : {}), + ...(task.routes ? { routes: task.routes } : {}), ...(outputContract ? { outputContract } : {}), }; } diff --git a/packages/schemas/src/__tests__/composeDecisionTask.test.ts b/packages/schemas/src/__tests__/composeDecisionTask.test.ts index 7af68e9d..03708754 100644 --- a/packages/schemas/src/__tests__/composeDecisionTask.test.ts +++ b/packages/schemas/src/__tests__/composeDecisionTask.test.ts @@ -63,7 +63,7 @@ function messages(tasks: unknown[]): string[] { } describe('a decision task in a compose draft', () => { - it('expands into the operation it runs and the guards its routes imply', () => { + it('expands into the operation it runs, carrying its routes', () => { const result = draft([read, triage(), ...downstream]); expect(result.success).toBe(true); if (!result.success) return; @@ -77,24 +77,34 @@ describe('a decision task in a compose draft', () => { questions: { team: { type: 'choice' }, urgent: { type: 'yes_no' } }, }, consumes: [{ taskId: 'read', outputKey: 'ticket', bindAs: 'ticket' }], + routes: [ + { + when: { + allOf: [ + "output.answers.team.value == 'billing'", + 'output.answers.team.decided == true', + ], + }, + enable: ['billing'], + }, + { + when: { + allOf: [ + "output.answers.team.value == 'technical'", + 'output.answers.team.decided == true', + ], + }, + enable: ['technical'], + }, + // A question with no minConfidence is always decided, so its route is one comparison. + { when: { expression: 'output.answers.urgent.value == true' }, enable: ['page'] }, + { when: { expression: 'output.answers.team.decided == false' }, enable: ['escalate'] }, + ], }); - expect(byId.get('billing')).toMatchObject({ - when: { - allOf: [ - "tasks.triage.output.answers.team.value == 'billing'", - 'tasks.triage.output.answers.team.decided == true', - ], - }, - dependsOn: ['triage'], - }); - // A question with no minConfidence is always decided, so its route is one comparison. - expect(byId.get('page')).toMatchObject({ - when: 'tasks.triage.output.answers.urgent.value == true', - }); - expect(byId.get('escalate')).toMatchObject({ - when: 'tasks.triage.output.answers.team.decided == false', - dependsOn: ['triage'], - }); + for (const target of ['billing', 'technical', 'page', 'escalate']) { + expect(byId.get(target)).toMatchObject({ dependsOn: ['triage'] }); + expect(byId.get(target)).not.toHaveProperty('when'); + } expect(result.data.tasks.some((t) => (t as { type: string }).type === 'decision')).toBe(false); }); @@ -110,8 +120,8 @@ describe('a decision task in a compose draft', () => { ]); expect(result.success).toBe(true); if (!result.success) return; - expect(result.data.tasks.find((t) => t.taskId === 'page')).toMatchObject({ - when: 'tasks.triage.output.answers.severity.value >= 1.5', + expect(result.data.tasks.find((t) => t.taskId === 'triage')).toMatchObject({ + routes: [{ when: { expression: 'output.answers.severity.value >= 1.5' }, enable: ['page'] }], }); }); @@ -170,13 +180,13 @@ describe('a decision task in a compose draft', () => { ]); expect(result.success).toBe(true); if (!result.success) return; - expect(result.data.tasks.find((t) => t.taskId === 'escalate')).toMatchObject({ + const routes = (result.data.tasks.find((t) => t.taskId === 'triage') as { routes?: unknown }) + .routes; + expect(routes).toContainEqual({ when: { - anyOf: [ - "tasks.triage.output.answers.team.value == 'ambiguous'", - 'tasks.triage.output.answers.team.decided == false', - ], + anyOf: ["output.answers.team.value == 'ambiguous'", 'output.answers.team.decided == false'], }, + enable: ['escalate'], }); }); diff --git a/packages/schemas/src/cybernetic/composeDecisionTask.ts b/packages/schemas/src/cybernetic/composeDecisionTask.ts index 07c98b5c..bf56ae44 100644 --- a/packages/schemas/src/cybernetic/composeDecisionTask.ts +++ b/packages/schemas/src/cybernetic/composeDecisionTask.ts @@ -16,6 +16,7 @@ import { } from '../operations/aiDecision.js'; import { DECISION_MODELS, enumWithCustom } from '../operations/enums.js'; import { TEMPLATE_BIND_KEY } from '../operations/workflow/taskTemplate.js'; +import type { WorkflowPollUntil, WorkflowTaskRoute } from '../operations/workflow/task.js'; export const DECISION_OPERATION_ID = 'ai.decision.decide'; @@ -253,20 +254,21 @@ function literal(value: string | number | boolean): string { return value.includes("'") ? `"${value}"` : `'${value}'`; } -function routeComparison(taskId: string, route: DecisionRoute): string { - const value = `tasks.${taskId}.output.answers.${route.question}.value`; +/** A comparison on the decision's own output, as a route reads it. */ +function routeComparison(route: DecisionRoute): string { + const value = `output.answers.${route.question}.value`; if (route.equals !== undefined) return `${value} == ${literal(route.equals)}`; if (route.atLeast !== undefined) return `${value} >= ${String(route.atLeast)}`; return `${value} < ${String(route.below)}`; } -function decided(taskId: string, question: string, outcome: boolean): string { - return `tasks.${taskId}.output.answers.${question}.decided == ${String(outcome)}`; +function decided(question: string, outcome: boolean): string { + return `output.answers.${question}.decided == ${String(outcome)}`; } -function combine(kind: 'anyOf' | 'allOf', expressions: string[]): DraftWhen { +function combine(kind: 'anyOf' | 'allOf', expressions: string[]): WorkflowPollUntil { return expressions.length === 1 - ? expressions[0]! + ? { expression: expressions[0]! } : kind === 'anyOf' ? { anyOf: expressions } : { allOf: expressions }; @@ -301,42 +303,55 @@ function claimsOf(task: DecisionTaskLike): Map { * be decided β€” unless the task is the fallback too, where "matches, or was not * decided" makes the confidence clause redundant and the guard a flat anyOf. */ -function guardFor(task: DecisionTaskLike, claim: TargetClaim): DraftWhen | null { +function guardFor(task: DecisionTaskLike, claim: TargetClaim): WorkflowPollUntil | null { const isGuarded = (question: string) => task.questions[question]?.minConfidence !== undefined; const undecided = claim.undecided ? [...new Set(task.routes.map((r) => r.question).filter(isGuarded))].map((q) => - decided(task.taskId, q, false), + decided(q, false), ) : []; const needsConfidence = claim.routes.filter((r) => isGuarded(r.question) && !claim.undecided); if (needsConfidence.length > 0) { if (claim.routes.length > 1) return null; const route = claim.routes[0]!; - return combine('allOf', [ - routeComparison(task.taskId, route), - decided(task.taskId, route.question, true), - ]); + return combine('allOf', [routeComparison(route), decided(route.question, true)]); } - const expressions = [...claim.routes.map((r) => routeComparison(task.taskId, r)), ...undecided]; + const expressions = [...claim.routes.map((r) => routeComparison(r)), ...undecided]; return expressions.length > 0 ? combine('anyOf', expressions) : null; } /** - * Replace every decision task with the operation task it runs, and give each - * task it routes to the guard that route implies. Idempotent: a draft with no + * The routes a decision stores: one per distinct condition, enabling every + * task that runs under it. A task's condition is the one `guardFor` gives, so + * deriving its guard from these routes reproduces it exactly. + */ +function routesOf(task: DecisionTaskLike): WorkflowTaskRoute[] { + const byCondition = new Map(); + for (const [target, claim] of claimsOf(task)) { + const when = guardFor(task, claim); + if (when === null) continue; + const key = JSON.stringify(when); + const existing = byCondition.get(key); + if (existing) existing.enable.push(target); + else byCondition.set(key, { when, enable: [target] }); + } + return [...byCondition.values()]; +} + +/** + * Replace every decision task with the operation task it runs, carrying its + * routes; each task it enables depends on it, and its guard is derived from + * those routes when the workflow is materialized. Idempotent: a draft with no * decision tasks comes back unchanged. */ export function expandDecisionTasks( draft: Draft, ): Omit & { tasks: Array> } { type Expanded = Exclude; - const guards = new Map(); + const enabledBy = new Map(); for (const task of draft.tasks) { if (!isDecisionTask(task)) continue; - for (const [target, claim] of claimsOf(task)) { - const when = guardFor(task, claim); - if (when !== null) guards.set(target, { when, decisionId: task.taskId }); - } + for (const target of claimsOf(task).keys()) enabledBy.set(target, task.taskId); } const tasks = draft.tasks.map((task) => { @@ -344,6 +359,7 @@ export function expandDecisionTasks( const state = Object.fromEntries( task.consumes.map((c) => [c.bindAs, { [TEMPLATE_BIND_KEY]: c.bindAs }]), ); + const routes = routesOf(task); return { type: 'operation', taskId: task.taskId, @@ -356,17 +372,17 @@ export function expandDecisionTasks( }, dependsOn: task.dependsOn, ...(task.when !== undefined ? { when: task.when } : {}), + ...(routes.length > 0 ? { routes } : {}), produces: [], consumes: task.consumes, retryability: 'safe', } as unknown as Expanded; } - const guard = guards.get(task.taskId); - if (!guard) return task as Expanded; + const decisionId = enabledBy.get(task.taskId); + if (decisionId === undefined) return task as Expanded; return { ...task, - when: guard.when, - dependsOn: [...new Set([...task.dependsOn, guard.decisionId])], + dependsOn: [...new Set([...task.dependsOn, decisionId])], } as Expanded; }); diff --git a/packages/schemas/src/cybernetic/composeSkill.ts b/packages/schemas/src/cybernetic/composeSkill.ts index fb298e32..be994997 100644 --- a/packages/schemas/src/cybernetic/composeSkill.ts +++ b/packages/schemas/src/cybernetic/composeSkill.ts @@ -6,6 +6,7 @@ import { WorkflowTaskOutputProjectionSchema, } from '../operations/workflow.js'; import { analyzeInputTemplate, TEMPLATE_BIND_KEY } from '../operations/workflow/taskTemplate.js'; +import { WorkflowTaskRouteSchema } from '../operations/workflow/task.js'; import { DecisionTaskFieldsSchema, DraftWhenSchema, @@ -455,6 +456,15 @@ export const AgentTaskKindSchema = z.enum([ ]); export type AgentTaskKind = z.infer; +const DraftRoutesSchema = z + .array(WorkflowTaskRouteSchema) + .min(1) + .max(50) + .optional() + .describe( + "Where this task's outcome sends the run: each route's when compares this task's own output (output. , single or anyOf/allOf) and enables the listed tasks when it holds. An enabled task runs only through its routes β€” it declares no when of its own, and is enabled by one task's routes. To fork on a judgement, use a decision task; its routes are written for you.", + ); + const AgentTaskSchema = z.object({ type: z.literal('agent'), taskId: z.string().min(1).max(64), @@ -467,6 +477,7 @@ const AgentTaskSchema = z.object({ goal: z.string().min(1).max(4000), dependsOn: z.array(z.string().max(64)).default([]), when: DraftWhenSchema.optional(), + routes: DraftRoutesSchema, produces: z.array(TaskOutputProductionSchema).default([]), consumes: z.array(TaskInputConsumptionSchema).default([]), context: z @@ -496,6 +507,7 @@ const OperationTaskSchema = z.object({ ), dependsOn: z.array(z.string().max(64)).default([]), when: DraftWhenSchema.optional(), + routes: DraftRoutesSchema, produces: z.array(TaskOutputProductionSchema).default([]), consumes: z.array(TaskInputConsumptionSchema).default([]), /** @@ -549,6 +561,7 @@ const HumanTaskSchema = z.object({ failureMode: z.enum(['isolate', 'cancel_siblings']).default('isolate'), dependsOn: z.array(z.string().max(64)).default([]), when: DraftWhenSchema.optional(), + routes: DraftRoutesSchema, produces: z.array(TaskOutputProductionSchema).default([]), }); From 784641724e5c0cc81ed2c92c93b168ef1d87d85f Mon Sep 17 00:00:00 2001 From: karim Date: Fri, 16 Oct 2026 09:50:54 +0200 Subject: [PATCH 3/4] feat(318): a route is edited as a route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A patch under /tasks/{i}/routes stages update_task_routes with the complete new list, so adding a branch is one task and one route. Ratification applies it and re-validates the graph, so removing a task a route still enables is refused naming that route. The proposal card shows routes added and removed as "when … β†’ enable …". The op is authoring-only: a Coach keeps replace_task. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/workflowPatchRoutes.test.ts | 87 +++++++++++++ .../inlineOps/workflowPatchToStagedOps.ts | 58 ++++++++- .../src/__tests__/routeProposals.test.ts | 115 ++++++++++++++++++ .../cybernetic-runtime/src/proposalOpDiff.ts | 34 +++++- .../src/ratificationHelpers.ts | 1 + .../src/stagedChange/applyOpsToSnapshot.ts | 12 ++ .../stagedChange/capabilityBindingApply.ts | 1 + .../src/stagedChange/preconditions.ts | 7 ++ .../src/__tests__/stagedChangeOps.test.ts | 2 +- .../schemas/src/cybernetic/stagedChange.ts | 20 ++- .../cybernetic/stagedChangePreconditions.ts | 1 + .../src/operations/workflow/registrations.ts | 2 + .../schemas/src/operations/workflow/task.ts | 6 + .../cybernetic/ProposalDiffView.tsx | 22 +++- 14 files changed, 363 insertions(+), 5 deletions(-) create mode 100644 apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/workflowPatchRoutes.test.ts create mode 100644 packages/cybernetic-runtime/src/__tests__/routeProposals.test.ts diff --git a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/workflowPatchRoutes.test.ts b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/workflowPatchRoutes.test.ts new file mode 100644 index 00000000..ff05526b --- /dev/null +++ b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/__tests__/workflowPatchRoutes.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it } from 'vitest'; +import type { Workflow, WorkflowTaskRoute } from '@aflow/schemas'; +import { convertWorkflowPatchToStagedOps } from '../workflowPatchToStagedOps.js'; + +const BILLING: WorkflowTaskRoute = { + when: { expression: "output.answers.team.value == 'billing'" }, + enable: ['billing'], +}; +const REFUND: WorkflowTaskRoute = { + when: { expression: "output.answers.team.value == 'refund'" }, + enable: ['refund'], +}; + +const existing = { + id: '00000000-0000-0000-0000-0000000000aa', + slug: 'triage', + name: 'Triage', + description: '', + mode: 'process', + revision: 1, + status: 'approved', + tasks: [ + { + taskId: 'triage', + name: 'Triage', + goal: 'Run ai.decision.decide.', + type: 'operation', + operation: 'ai.decision.decide', + routes: [BILLING], + }, + { taskId: 'billing', name: 'Billing', goal: 'Handle billing.', type: 'agent' }, + ], + outcomes: [], + iteration: { auto: false, maxConsecutiveRuns: 1, stopOnOutcomesMet: true, cooldownMs: 0 }, + stateVariables: [], + createdAt: '2026-10-16T00:00:00.000Z', + updatedAt: '2026-10-16T00:00:00.000Z', +} as Workflow; + +const refundTask = { taskId: 'refund', name: 'Refund', goal: 'Handle refunds.', type: 'agent' }; + +describe('a patch under a task’s routes is staged as its routes', () => { + it('adds a branch as one task and one route', () => { + const { ops, unsupported } = convertWorkflowPatchToStagedOps( + [ + { op: 'add', path: '/tasks/-', value: refundTask }, + { op: 'add', path: '/tasks/0/routes/-', value: REFUND }, + ], + existing, + ); + expect(unsupported).toEqual([]); + expect(ops).toEqual([ + { op: 'add_task', task: refundTask }, + { op: 'update_task_routes', taskId: 'triage', routes: [BILLING, REFUND] }, + ]); + }); + + it('extends what one route enables', () => { + const { ops } = convertWorkflowPatchToStagedOps( + [{ op: 'add', path: '/tasks/0/routes/0/enable/-', value: 'refund' }], + existing, + ); + expect(ops).toEqual([ + { + op: 'update_task_routes', + taskId: 'triage', + routes: [{ ...BILLING, enable: ['billing', 'refund'] }], + }, + ]); + }); + + it('stages an empty list when the routes are removed', () => { + const { ops } = convertWorkflowPatchToStagedOps( + [{ op: 'remove', path: '/tasks/0/routes' }], + existing, + ); + expect(ops).toEqual([{ op: 'update_task_routes', taskId: 'triage', routes: [] }]); + }); + + it('refuses a route that is not a route, naming the task', () => { + const { unsupported } = convertWorkflowPatchToStagedOps( + [{ op: 'add', path: '/tasks/0/routes/-', value: { when: 'x', enable: [] } }], + existing, + ); + expect(unsupported.join()).toContain('task "triage" with invalid routes'); + }); +}); diff --git a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/workflowPatchToStagedOps.ts b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/workflowPatchToStagedOps.ts index 13aba028..5ef7c3b4 100644 --- a/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/workflowPatchToStagedOps.ts +++ b/apps/aflow-orchestrator/src/services/SessionOrchestrator/handlers/inlineOps/workflowPatchToStagedOps.ts @@ -1,6 +1,7 @@ import type { Workflow, StagedChangeOp, SkillGoal, SkillCampaignContract } from '@aflow/schemas'; import { WorkflowTaskSchema, + WorkflowTaskRouteSchema, TaskContextSpecSchema, WorkflowOutputDeclarationSchema, WorkflowStateVariableSchema, @@ -130,6 +131,57 @@ function convertContextPatch( } } +/** + * Any edit under `/tasks/{i}/routes` β€” the whole list, one route, or one + * route's `enable` β€” is applied to the task's routes and staged as the + * complete new list, so the proposal reads as routes rather than a rewritten + * task. Removing the list, or its last route, stages an empty one. + */ +function convertRoutesPatch( + op: WorkflowPatchInput['operations'][number], + existing: Workflow, + index: number, + taskId: string, + segments: string[], +): StagedChangeOp | { unsupported: string } { + const task = existing.tasks[index]; + if (!task) { + return { + unsupported: `Task path "${op.path}" references task index ${String(index)}, which does not exist on the workflow.`, + }; + } + if (op.op !== 'add' && op.op !== 'replace' && op.op !== 'remove') { + return { + unsupported: `Op shape "${op.op} ${op.path}" is not supported for routes. Use add/replace/remove under /tasks/{i}/routes.`, + }; + } + const routesPath = `/${segments.slice(2).map(escapePointerSegment).join('/')}`; + try { + const patched = applyJsonPatch({ routes: task.routes ?? [] }, [ + { ...op, path: routesPath }, + ] as WorkflowPatchInput['operations']) as { routes?: unknown }; + const parse = WorkflowTaskRouteSchema.array() + .max(50) + .safeParse(patched.routes ?? []); + if (!parse.success) { + return { + unsupported: `${op.op} ${op.path} would leave task "${taskId}" with invalid routes: ${parse.error.message}`, + }; + } + return { op: 'update_task_routes', taskId, routes: parse.data } as StagedChangeOp; + } catch (err) { + const detail = + err instanceof JsonPatchError + ? `${err.message}${err.opPath ? ` (at ${err.opPath})` : ''}` + : err instanceof Error + ? err.message + : String(err); + return { + unsupported: `${op.op} ${op.path} cannot be applied to task "${taskId}" routes: ${detail}`, + }; + } +} + function convertTaskPatch( op: WorkflowPatchInput['operations'][number], existing: Workflow, @@ -332,6 +384,10 @@ function convertOne( return convertContextPatch(op, existing, index, taskId, segments); } + if (segments[2] === 'routes') { + return convertRoutesPatch(op, existing, index, taskId, segments); + } + // replace /tasks/{i}/dependsOn if (op.op === 'replace' && segments.length === 3 && segments[2] === 'dependsOn') { if (!Array.isArray(op.value) || !op.value.every((v): v is string => typeof v === 'string')) { @@ -387,7 +443,7 @@ function convertOne( } return { - unsupported: `Op shape "${op.op} ${op.path}" is not yet supported for agent-initiated workflow patches. Supported shapes: task add/remove/replacements under /tasks, replace /outcomes/{i}/evaluator/target, add/replace/remove under /stateVariables or /output, and replace /iteration. Tasks may be addressed as /tasks/{taskId} or /tasks/{index}. Break a larger change into these shapes; each becomes part of one proposal.`, + unsupported: `Op shape "${op.op} ${op.path}" is not yet supported for agent-initiated workflow patches. Supported shapes: task add/remove/replacements under /tasks, routes under /tasks/{i}/routes, replace /outcomes/{i}/evaluator/target, add/replace/remove under /stateVariables or /output, and replace /iteration. Tasks may be addressed as /tasks/{taskId} or /tasks/{index}. Break a larger change into these shapes; each becomes part of one proposal.`, }; } diff --git a/packages/cybernetic-runtime/src/__tests__/routeProposals.test.ts b/packages/cybernetic-runtime/src/__tests__/routeProposals.test.ts new file mode 100644 index 00000000..7fe245f8 --- /dev/null +++ b/packages/cybernetic-runtime/src/__tests__/routeProposals.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, it } from 'vitest'; +import type { StagedChangeOp, Workflow, WorkflowTaskRoute } from '@aflow/schemas'; +import { applyOpsToSnapshot } from '../stagedChange/applyOpsToSnapshot.js'; +import { buildProposalOpDiffs } from '../proposalOpDiff.js'; + +const BILLING: WorkflowTaskRoute = { + when: { expression: "output.answers.team.value == 'billing'" }, + enable: ['billing'], +}; +const REFUND: WorkflowTaskRoute = { + when: { anyOf: ["output.answers.team.value == 'refund'", 'output.answers.urgent.value == true'] }, + enable: ['refund'], +}; + +function workflow(): Workflow { + return { + id: '00000000-0000-0000-0000-0000000000aa', + slug: 'triage', + name: 'Triage', + description: '', + mode: 'process', + revision: 1, + status: 'approved', + tasks: [ + { + taskId: 'triage', + name: 'Triage', + goal: 'Run ai.decision.decide.', + type: 'operation', + operation: 'ai.decision.decide', + inputTemplate: { + state: 'A ticket', + questions: { + team: { type: 'choice', options: { billing: null, refund: null } }, + urgent: { type: 'yes_no' }, + }, + }, + routes: [BILLING], + }, + { + taskId: 'billing', + name: 'Billing', + goal: 'Handle billing.', + type: 'agent', + dependsOn: ['triage'], + }, + { + taskId: 'refund', + name: 'Refund', + goal: 'Handle refunds.', + type: 'agent', + dependsOn: ['triage'], + }, + ], + outcomes: [ + { id: 'routed', name: 'Routed', evaluator: { type: 'manual', instruction: 'Routed.' } }, + ], + iteration: { maxConsecutiveRuns: 1 } as never, + stateVariables: [], + createdAt: '2026-10-16T00:00:00.000Z', + updatedAt: '2026-10-16T00:00:00.000Z', + } as Workflow; +} + +function apply(routes: WorkflowTaskRoute[]) { + const result = applyOpsToSnapshot({ + workflow: workflow(), + evalSuites: new Map(), + ops: [{ op: 'update_task_routes', taskId: 'triage', routes } as StagedChangeOp], + targetSlug: 'triage', + }); + if (!result.ok) throw new Error(result.failureDetail); + return result.ok ? result.candidateWorkflow?.tasks.find((t) => t.taskId === 'triage') : undefined; +} + +describe('a route change ratified onto a workflow', () => { + it('sets the task’s routes to the proposed list', () => { + expect(apply([BILLING, REFUND])?.routes).toEqual([BILLING, REFUND]); + }); + + it('removes the field when the list is empty', () => { + expect(apply([])).not.toHaveProperty('routes'); + }); +}); + +describe('removing a task a route still enables', () => { + it('is refused, naming the route that points at it', () => { + const result = applyOpsToSnapshot({ + workflow: workflow(), + evalSuites: new Map(), + ops: [{ op: 'remove_task', taskId: 'billing' } as StagedChangeOp], + targetSlug: 'triage', + }); + expect(result.ok).toBe(false); + expect(result.ok ? '' : result.failureDetail).toContain( + 'Task "triage" routes[0] enables "billing", which is not a task in this workflow.', + ); + }); +}); + +describe('a route change on the proposal card', () => { + it('reads as the routes added and removed', () => { + const [diff] = buildProposalOpDiffs(workflow(), [ + { op: 'update_task_routes', taskId: 'triage', routes: [REFUND] } as StagedChangeOp, + ]); + expect(diff?.entries).toEqual([ + { field: 'route', before: "when output.answers.team.value == 'billing' β†’ enable billing" }, + { + field: 'route', + after: + "when output.answers.team.value == 'refund' or output.answers.urgent.value == true β†’ enable refund", + }, + ]); + }); +}); diff --git a/packages/cybernetic-runtime/src/proposalOpDiff.ts b/packages/cybernetic-runtime/src/proposalOpDiff.ts index 80ee43ac..04ca11dd 100644 --- a/packages/cybernetic-runtime/src/proposalOpDiff.ts +++ b/packages/cybernetic-runtime/src/proposalOpDiff.ts @@ -9,7 +9,14 @@ * Pure; consumed by the proposal detail route and rendered by * `ProposalDiffView`. */ -import type { SkillManifest, StagedChange, Workflow, WorkflowTask } from '@aflow/schemas'; +import { + describeRoute, + type SkillManifest, + type StagedChange, + type Workflow, + type WorkflowTask, + type WorkflowTaskRoute, +} from '@aflow/schemas'; export interface ProposalOpDiffEntry { field: string; @@ -42,6 +49,23 @@ function taskSummary(task: WorkflowTask | undefined): string { return clip(`${task.taskId} (${family}${target ? `: ${target}` : ''}) β€” ${task.goal}`); } +/** One entry per route added or removed; a route kept as it was is not a change. */ +function routeEntries( + before: readonly WorkflowTaskRoute[], + after: readonly WorkflowTaskRoute[], +): ProposalOpDiffEntry[] { + const beforeText = before.map(describeRoute); + const afterText = after.map(describeRoute); + return [ + ...beforeText + .filter((t) => !afterText.includes(t)) + .map((t) => ({ field: 'route', before: clip(t) })), + ...afterText + .filter((t) => !beforeText.includes(t)) + .map((t) => ({ field: 'route', after: clip(t) })), + ]; +} + export function buildProposalOpDiffs( workflow: Pick | null, ops: StagedChange['proposal']['ops'], @@ -112,6 +136,14 @@ export function buildProposalOpDiffs( after: clip(o['dependsOn']), }); break; + case 'update_task_routes': + entries.push( + ...routeEntries( + task?.routes ?? [], + (o['routes'] as WorkflowTaskRoute[] | undefined) ?? [], + ), + ); + break; case 'update_activation_hint': entries.push({ field: 'activation hint', diff --git a/packages/cybernetic-runtime/src/ratificationHelpers.ts b/packages/cybernetic-runtime/src/ratificationHelpers.ts index 20ff4fb8..3223a40a 100644 --- a/packages/cybernetic-runtime/src/ratificationHelpers.ts +++ b/packages/cybernetic-runtime/src/ratificationHelpers.ts @@ -14,6 +14,7 @@ const HARD_FLOOR_OPS: ReadonlySet = new Set([ 'remove_task', 'reorder_tasks', 'update_task_dependencies', + 'update_task_routes', 'update_outcome_threshold', ]); diff --git a/packages/cybernetic-runtime/src/stagedChange/applyOpsToSnapshot.ts b/packages/cybernetic-runtime/src/stagedChange/applyOpsToSnapshot.ts index be2eb62b..d851f6bf 100644 --- a/packages/cybernetic-runtime/src/stagedChange/applyOpsToSnapshot.ts +++ b/packages/cybernetic-runtime/src/stagedChange/applyOpsToSnapshot.ts @@ -532,6 +532,17 @@ function applyWorkflowOp(workflow: Workflow, op: StagedChangeOp): void { break; } + case 'update_task_routes': { + const task = workflow.tasks.find((t) => t.taskId === op.taskId); + if (!task) throw new Error(`Task ${op.taskId} not found`); + if (op.routes.length > 0) { + task.routes = op.routes; + } else { + delete task.routes; + } + break; + } + case 'update_outcome_threshold': { const outcome = workflow.outcomes.find((o) => o.id === op.outcomeId); if (!outcome) throw new Error(`Outcome ${op.outcomeId} not found`); @@ -908,6 +919,7 @@ function deriveOpFailureCode(op: StagedChangeOp): string { case 'update_task_goal': case 'update_task_context_spec': case 'update_task_dependencies': + case 'update_task_routes': case 'replace_task': case 'remove_task': case 'reorder_tasks': diff --git a/packages/cybernetic-runtime/src/stagedChange/capabilityBindingApply.ts b/packages/cybernetic-runtime/src/stagedChange/capabilityBindingApply.ts index 50a8c137..208092d3 100644 --- a/packages/cybernetic-runtime/src/stagedChange/capabilityBindingApply.ts +++ b/packages/cybernetic-runtime/src/stagedChange/capabilityBindingApply.ts @@ -373,6 +373,7 @@ export async function applyCapabilityBindingOps( case 'update_task_goal': case 'update_task_context_spec': case 'update_task_dependencies': + case 'update_task_routes': case 'add_task': case 'replace_task': case 'remove_task': diff --git a/packages/cybernetic-runtime/src/stagedChange/preconditions.ts b/packages/cybernetic-runtime/src/stagedChange/preconditions.ts index 11cc5ef6..59f43f4c 100644 --- a/packages/cybernetic-runtime/src/stagedChange/preconditions.ts +++ b/packages/cybernetic-runtime/src/stagedChange/preconditions.ts @@ -66,6 +66,8 @@ export function computeOpTargetDescriptor( return { kind: 'task.context', taskId: op.taskId }; case 'update_task_dependencies': return { kind: 'task.dependencies', taskId: op.taskId }; + case 'update_task_routes': + return { kind: 'task.routes', taskId: op.taskId }; case 'replace_task': return { kind: 'task.whole', taskId: op.taskId }; case 'promote_context_strategy': @@ -199,6 +201,11 @@ export function readTargetHash( if (!task) return { hash: null, present: false }; return { hash: hashCanonical(task.dependsOn ?? []), present: true }; } + case 'task.routes': { + const task = artifacts.workflow?.tasks.find((t) => t.taskId === descriptor.taskId); + if (!task) return { hash: null, present: false }; + return { hash: hashCanonical(task.routes ?? []), present: true }; + } case 'task.whole': { const task = artifacts.workflow?.tasks.find((t) => t.taskId === descriptor.taskId); if (!task) return { hash: null, present: false }; diff --git a/packages/schemas/src/__tests__/stagedChangeOps.test.ts b/packages/schemas/src/__tests__/stagedChangeOps.test.ts index c5105839..eb0bf209 100644 --- a/packages/schemas/src/__tests__/stagedChangeOps.test.ts +++ b/packages/schemas/src/__tests__/stagedChangeOps.test.ts @@ -248,7 +248,7 @@ describe('Plan 113 β€” staged change ops', () => { describe('Plan 301 β€” the Coach union is narrower than the storable one', () => { /** Ops only an authoring skill proposes, never a Coach. */ - const AUTHORING_ONLY_OPS = ['eval_case_draft']; + const AUTHORING_ONLY_OPS = ['eval_case_draft', 'update_task_routes']; function opKinds(schema: unknown): string[] { let node: unknown = schema; diff --git a/packages/schemas/src/cybernetic/stagedChange.ts b/packages/schemas/src/cybernetic/stagedChange.ts index 34314a59..0f697fa5 100644 --- a/packages/schemas/src/cybernetic/stagedChange.ts +++ b/packages/schemas/src/cybernetic/stagedChange.ts @@ -12,6 +12,7 @@ import { WorkflowOutputDeclarationSchema, WorkflowStateVariableSchema, WorkflowTaskSchema, + WorkflowTaskRouteSchema, assertAuthoredTask, type AuthoredTaskLike, } from '../operations/workflow.js'; @@ -251,6 +252,19 @@ const UpdateTaskDependenciesOpSchema = z }) .strict(); +/** + * Set where a task's outcome sends the run. The full new list replaces the + * prior one; an empty list removes the task's routes. Only a patch stages it β€” + * a Coach rewrites a fork with `replace_task`. + */ +const UpdateTaskRoutesOpSchema = z + .object({ + op: z.literal('update_task_routes'), + taskId: z.string(), + routes: z.array(WorkflowTaskRouteSchema).max(50), + }) + .strict(); + /** Update a workflow outcome threshold. */ const UpdateOutcomeThresholdOpSchema = z .object({ @@ -1100,7 +1114,11 @@ export const CoachProposableOpSchema = z * proposes. */ export const StagedChangeOpSchema = z - .discriminatedUnion('op', [...COACH_PROPOSABLE_OP_SCHEMAS, EvalCaseDraftOpSchema]) + .discriminatedUnion('op', [ + ...COACH_PROPOSABLE_OP_SCHEMAS, + EvalCaseDraftOpSchema, + UpdateTaskRoutesOpSchema, + ]) .superRefine(assertAuthoredTaskOp); export type StagedChangeOp = z.infer; diff --git a/packages/schemas/src/cybernetic/stagedChangePreconditions.ts b/packages/schemas/src/cybernetic/stagedChangePreconditions.ts index b657e89d..b9434f63 100644 --- a/packages/schemas/src/cybernetic/stagedChangePreconditions.ts +++ b/packages/schemas/src/cybernetic/stagedChangePreconditions.ts @@ -16,6 +16,7 @@ export const TargetDescriptorSchema = z.discriminatedUnion('kind', [ z.object({ kind: z.literal('task.goal'), taskId: z.string() }).strict(), z.object({ kind: z.literal('task.context'), taskId: z.string() }).strict(), z.object({ kind: z.literal('task.dependencies'), taskId: z.string() }).strict(), + z.object({ kind: z.literal('task.routes'), taskId: z.string() }).strict(), z.object({ kind: z.literal('task.whole'), taskId: z.string() }).strict(), z.object({ kind: z.literal('task.absent'), taskId: z.string() }).strict(), z.object({ kind: z.literal('tasks.order') }).strict(), diff --git a/packages/schemas/src/operations/workflow/registrations.ts b/packages/schemas/src/operations/workflow/registrations.ts index f45f1e01..a5678111 100644 --- a/packages/schemas/src/operations/workflow/registrations.ts +++ b/packages/schemas/src/operations/workflow/registrations.ts @@ -235,6 +235,7 @@ export function createWorkflowOperationRegistrations( 'Bumping budget.maxRuns when the agent hits the cap mid-optimization', 'Renaming or retuning a single task without touching the rest', 'Changing the skill goal (replace /goal) or a campaign field (/campaign/fields/{key})', + 'Adding a branch to a fork: add the task, then add /tasks/{source}/routes/- with { when: "output. == ", enable: [] }', ], whenNotToUse: [ 'Initial creation β€” use workflow.manage.put', @@ -245,6 +246,7 @@ export function createWorkflowOperationRegistrations( 'Use expectedRevision to guard against concurrent writers', 'A single patch may not mix metadata (budget/status) with definition (tasks/goal/campaign)', 'Identity campaign fields are immutable β€” their schema cannot be reshaped, only relabeled', + 'A task a route enables gets its guard and its dependency on the source from that route; give it no when of its own, and remove its route before removing it', 'Operations that run only as workflow steps, such as ai.decision.decide, are found with catalog.tool.search workflowSteps: true; a when that reads tasks..output. is refused when the producer does not emit that path', ], minimalExampleInput: { diff --git a/packages/schemas/src/operations/workflow/task.ts b/packages/schemas/src/operations/workflow/task.ts index e9e2a6f8..c83cab64 100644 --- a/packages/schemas/src/operations/workflow/task.ts +++ b/packages/schemas/src/operations/workflow/task.ts @@ -181,6 +181,12 @@ export const WorkflowTaskRouteSchema = z .strict(); export type WorkflowTaskRoute = z.infer; +/** A route as an operator reads it: `when β†’ enable `. */ +export function describeRoute(route: WorkflowTaskRoute): string { + const joiner = predicateCombinator(route.when) === 'anyOf' ? ' or ' : ' and '; + return `when ${predicateExpressions(route.when).join(joiner)} β†’ enable ${route.enable.join(', ')}`; +} + export const WorkflowTaskPollSchema = z .object({ /** diff --git a/packages/web-product/src/ui/components/cybernetic/ProposalDiffView.tsx b/packages/web-product/src/ui/components/cybernetic/ProposalDiffView.tsx index 1758b11e..d6cb3238 100644 --- a/packages/web-product/src/ui/components/cybernetic/ProposalDiffView.tsx +++ b/packages/web-product/src/ui/components/cybernetic/ProposalDiffView.tsx @@ -10,7 +10,7 @@ * bundle overview (task list, eval summary, activation hints). */ import { Badge, Card, CardBody, Column, Row, Text } from '@aflow/design-system'; -import { ListingRequirementsSchema } from '@aflow/schemas'; +import { ListingRequirementsSchema, WorkflowTaskRouteSchema, describeRoute } from '@aflow/schemas'; import { listingRequirementRows } from '../../lib/listing-requirements.js'; import { SkillComposePreview } from './SkillComposePreview.js'; @@ -171,6 +171,25 @@ function renderOp(op: Op): React.ReactNode { case 'remove_task': return ; + case 'update_task_routes': { + const parsed = WorkflowTaskRouteSchema.array().safeParse(op['routes']); + const routes = parsed.success ? parsed.data : []; + return ( + + + {routes.length === 0 ? ( + No routes + ) : ( + routes.map((route, i) => ( + + {describeRoute(route)} + + )) + )} + + ); + } + case 'reorder_tasks': { const ids = op['taskIds']; return ( @@ -501,6 +520,7 @@ const OP_LABELS: Record = { update_task_context_spec: 'Update context spec', add_task: 'Add task', remove_task: 'Remove task', + update_task_routes: 'Update routes', reorder_tasks: 'Reorder tasks', update_outcome_threshold: 'Update outcome', update_activation_hint: 'Update activation', From 17d9696215a49ff6a565628d78a4d0d04b9b04cd Mon Sep 17 00:00:00 2001 From: karim Date: Fri, 16 Oct 2026 10:01:08 +0200 Subject: [PATCH 4/4] feat(318): the designer draws a fork from its routes The guard derivation moves to @aflow/schemas so materialization and the designer share it. The designer draws edges from derived guards, so each route is a labelled conditional edge from its source, while the node keeps the stored task the inspector edits; the inspector shows a source's routes and a routed task's "Enabled by". Route error kinds move to their own module. Co-Authored-By: Claude Opus 5.5 --- ...18-a-fork-is-stored-where-it-is-decided.md | 10 +- .../src/scheduling/graphValidation.ts | 20 +-- .../src/scheduling/routeErrorKinds.ts | 20 +++ .../src/scheduling/routes.ts | 108 +--------------- .../src/skillValidity/skillValidity.ts | 3 +- .../schemas/src/operations/workflow/index.ts | 1 + .../schemas/src/operations/workflow/routes.ts | 120 ++++++++++++++++++ .../schemas/src/operations/workflow/task.ts | 6 - .../skill-designer/SkillInspector.tsx | 36 ++++++ .../skill-designer/skill-graph.test.ts | 33 +++++ .../components/skill-designer/skill-graph.ts | 8 +- 11 files changed, 230 insertions(+), 135 deletions(-) create mode 100644 packages/cybernetic-runtime/src/scheduling/routeErrorKinds.ts create mode 100644 packages/schemas/src/operations/workflow/routes.ts diff --git a/docs/plans/318-a-fork-is-stored-where-it-is-decided.md b/docs/plans/318-a-fork-is-stored-where-it-is-decided.md index 71768a61..bb50d6bc 100644 --- a/docs/plans/318-a-fork-is-stored-where-it-is-decided.md +++ b/docs/plans/318-a-fork-is-stored-where-it-is-decided.md @@ -1,6 +1,6 @@ # Plan 318 β€” A fork is stored where it is decided: routes on the source task -**Status:** 🌱 design Β· **Builds on:** Plan 316 (the decision step), Plan 317 (agents propose, operators approve) +**Status:** πŸ”¨ P1–P4 built; the live compose and patch runs open Β· **Builds on:** Plan 316 (the decision step), Plan 317 (agents propose, operators approve) ## 1. Problem @@ -48,19 +48,19 @@ Stored workflows gain an optional `routes` field. No existing workflow changes s ### P1 β€” Routes in the model -`routes` on the workflow task; derivation in `materializeSkillTasks`; the D4 checks; the route-check registry with the decision operation's checks moved into it. Exit: a hand-written workflow with routes runs the right branch for each output, and each D4 violation is refused with a diagnostic naming the route. +**Delivered:** `routes` on the workflow task; the derivation (`deriveRouteGuards`, in `@aflow/schemas` so materialization and the designer share it) applied in both materialize paths; the D4 checks in `validateRoutes`; the route-check registry with the decision operation's checks. Exit: a hand-written workflow with routes runs the right branch for each output, and each D4 violation is refused with a diagnostic naming the route. ### P2 β€” Compose-skill lowers to routes -The `decision` draft task becomes an operation task with `routes`; compose-skill's draft schema and its one line of guidance describe routes rather than guards. Exit: the live triage request from Plan 316 produces a skill whose stored form has routes on the decision and no guards on the responders, and runs every branch as before. +**Delivered:** the `decision` draft task lowers to an operation task with `routes` (one per distinct condition), its targets carrying only the dependency; agent, operation and human draft tasks accept `routes`, described on the draft schema; the assembler carries them. **Open:** the live run of the exit below. Exit: the live triage request from Plan 316 produces a skill whose stored form has routes on the decision and no guards on the responders, and runs every branch as before. ### P3 β€” Editing a route -Typed staged ops for route add, remove and replace; proposal text that reads as a route; the refusal for removing a still-enabled task. Exit: asking Helmsman to "send refund requests to a new refund handler" produces one proposal that adds the task and the route, in at most two attempts. +**Delivered:** one typed staged op, `update_task_routes` (the complete new list β€” any patch under `/tasks/{i}/routes` lowers to it), authoring-only so the Coach union stays as it was; the proposal card shows routes added and removed as `when … β†’ enable …`; ratification re-validates, so removing a still-enabled task is refused naming the route; `workflow.manage.patch` usage names the add-a-branch shape. **Open:** the live Helmsman run of the exit below. Exit: asking Helmsman to "send refund requests to a new refund handler" produces one proposal that adds the task and the route, in at most two attempts. ### P4 β€” The designer -Route edges and the routed-target inspector. Exit: the triage skill shows its fork as one source with labelled edges. +**Delivered:** the designer draws edges from the derived guards, so each route is a labelled conditional edge from its source, while the node keeps the stored task; the inspector shows a source's routes and a routed target's "Enabled by". Exit: the triage skill shows its fork as one source with labelled edges. ## 5. How it is proven diff --git a/packages/cybernetic-runtime/src/scheduling/graphValidation.ts b/packages/cybernetic-runtime/src/scheduling/graphValidation.ts index 56c41f17..fcea3d9b 100644 --- a/packages/cybernetic-runtime/src/scheduling/graphValidation.ts +++ b/packages/cybernetic-runtime/src/scheduling/graphValidation.ts @@ -1,3 +1,4 @@ +import type { RouteValidationErrorKind } from './routeErrorKinds.js'; import { analyzeInputTemplate, describePredicate, @@ -39,24 +40,7 @@ export interface GraphValidationError { | 'unsupported_when_expression' /** A `when` reads an output path the referenced task's output schema does not have. */ | 'when_output_path_unknown' - /** A route's condition is not a comparison on its own task's output. */ - | 'route_unsupported_when' - /** A route enables its own task. */ - | 'route_target_self' - /** A route enables a task that does not exist. */ - | 'route_target_unknown' - /** A task is enabled by routes of more than one task. */ - | 'route_target_multiple_sources' - /** A routed task also declares a `when` of its own. */ - | 'route_target_has_when' - /** Routes to one task combine into no single-level guard. */ - | 'route_guard_inexpressible' - /** A decision route names a question the step does not ask. */ - | 'route_decision_unknown_question' - /** A decision route compares an answer with a value the question never gives. */ - | 'route_decision_unknown_option' - /** A decision route's score threshold lies outside the rubric. */ - | 'route_decision_threshold_out_of_range' + | RouteValidationErrorKind | 'duplicate_task_id' | 'reserved_task_id' | 'binding_dangling_task_ref' diff --git a/packages/cybernetic-runtime/src/scheduling/routeErrorKinds.ts b/packages/cybernetic-runtime/src/scheduling/routeErrorKinds.ts new file mode 100644 index 00000000..bb2fb54d --- /dev/null +++ b/packages/cybernetic-runtime/src/scheduling/routeErrorKinds.ts @@ -0,0 +1,20 @@ +/** The ways a task's routes can be refused, as `GraphValidationError` kinds. */ +export type RouteValidationErrorKind = + /** A route's condition is not a comparison on its own task's output. */ + | 'route_unsupported_when' + /** A route enables its own task. */ + | 'route_target_self' + /** A route enables a task that does not exist. */ + | 'route_target_unknown' + /** A task is enabled by routes of more than one task. */ + | 'route_target_multiple_sources' + /** A routed task also declares a `when` of its own. */ + | 'route_target_has_when' + /** Routes to one task combine into no single-level guard. */ + | 'route_guard_inexpressible' + /** A decision route names a question the step does not ask. */ + | 'route_decision_unknown_question' + /** A decision route compares an answer with a value the question never gives. */ + | 'route_decision_unknown_option' + /** A decision route's score threshold lies outside the rubric. */ + | 'route_decision_threshold_out_of_range'; diff --git a/packages/cybernetic-runtime/src/scheduling/routes.ts b/packages/cybernetic-runtime/src/scheduling/routes.ts index 5ea0e244..6da45d4a 100644 --- a/packages/cybernetic-runtime/src/scheduling/routes.ts +++ b/packages/cybernetic-runtime/src/scheduling/routes.ts @@ -1,19 +1,13 @@ /** - * Routes: a fork stored on the task that decides it. - * - * A task's `routes` say which tasks its outcome enables. The enabled tasks' - * guards are derived here, at materialization β€” the form every run and every - * validity check uses β€” so the stored workflow holds the fork once, on its - * source, and the guards cannot drift from it. + * The route rules. The guards routes imply are derived in `@aflow/schemas` + * (`deriveRouteGuards`), where both materialization and the designer read them. */ import { DECISION_OPERATION_ID, - predicateCombinator, predicateExpressions, - type WorkflowPollUntil, + routeClaims, + routeGuardFor, type WorkflowTask, - type WorkflowTaskRoute, - type WorkflowWhen, } from '@aflow/schemas'; import { parseOutputComparison } from './graph.js'; import type { GraphValidationError } from './graphValidation.js'; @@ -28,98 +22,6 @@ const ROUTE_CHECKS: Readonly GraphValidat [DECISION_OPERATION_ID]: checkDecisionRoutes, }; -/** `output. …` read from the source, rewritten to how the target reads it. */ -function fromSource(sourceId: string, expression: string): string { - return `tasks.${sourceId}.${expression.trim()}`; -} - -/** - * The guard a task enabled by these routes of one source runs under, or null - * when the combination has no single-level form. Several routes combine as - * `anyOf`; a route that is itself an `allOf` combines with nothing, because the - * `when` grammar has no nesting. - */ -export function routeGuardFor( - sourceId: string, - routes: readonly WorkflowTaskRoute[], -): WorkflowWhen | null { - const first = routes[0]; - if (first === undefined) return null; - if (routes.length === 1) return toWhen(sourceId, first.when); - if ( - routes.some( - (r) => predicateCombinator(r.when) === 'allOf' && predicateExpressions(r.when).length > 1, - ) - ) { - return null; - } - const expressions = [ - ...new Set( - routes.flatMap((r) => predicateExpressions(r.when)).map((e) => fromSource(sourceId, e)), - ), - ]; - return expressions.length === 1 - ? { expression: expressions[0]!, onMissingRef: 'skip' } - : { anyOf: expressions, onMissingRef: 'skip' }; -} - -function toWhen(sourceId: string, predicate: WorkflowPollUntil): WorkflowWhen { - const expressions = predicateExpressions(predicate).map((e) => fromSource(sourceId, e)); - switch (predicateCombinator(predicate)) { - case 'expression': - return { expression: expressions[0]!, onMissingRef: 'skip' }; - case 'anyOf': - return { anyOf: expressions, onMissingRef: 'skip' }; - case 'allOf': - return { allOf: expressions, onMissingRef: 'skip' }; - } -} - -/** Every task some route enables, with its one source and the routes that enable it. */ -function claims( - tasks: readonly WorkflowTask[], -): Map> { - const byTarget = new Map>(); - for (const task of tasks) { - for (const route of task.routes ?? []) { - for (const target of route.enable) { - const list = byTarget.get(target) ?? []; - list.push({ source: task.taskId, route }); - byTarget.set(target, list); - } - } - } - return byTarget; -} - -/** - * Give each routed task the guard its routes imply and its source as a - * dependency. A target the routes cannot guard β€” claimed by two sources, - * carrying its own `when`, or enabled by an inexpressible combination β€” is - * left as stored; validation refuses that workflow before it can run. - */ -export function deriveRouteGuards(tasks: WorkflowTask[]): WorkflowTask[] { - const byTarget = claims(tasks); - if (byTarget.size === 0) return tasks; - return tasks.map((task) => { - const claimed = byTarget.get(task.taskId); - if (!claimed || task.when !== undefined) return task; - const sources = new Set(claimed.map((c) => c.source)); - if (sources.size !== 1) return task; - const [source] = sources; - const guard = routeGuardFor( - source!, - claimed.map((c) => c.route), - ); - if (!guard) return task; - return { - ...task, - when: guard, - dependsOn: [...new Set([...(task.dependsOn ?? []), source!])], - }; - }); -} - /** The route rules, read on the workflow as stored β€” before any guard is derived. */ export function validateRoutes(tasks: readonly WorkflowTask[]): GraphValidationError[] { const errors: GraphValidationError[] = []; @@ -160,7 +62,7 @@ export function validateRoutes(tasks: readonly WorkflowTask[]): GraphValidationE errors.push(...(task.operation ? (ROUTE_CHECKS[task.operation]?.(task) ?? []) : [])); } - for (const [target, claimed] of claims(tasks)) { + for (const [target, claimed] of routeClaims(tasks)) { const targetTask = byId.get(target); if (!targetTask) continue; const sources = [...new Set(claimed.map((c) => c.source))]; diff --git a/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts b/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts index 2a75f9bf..43164066 100644 --- a/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts +++ b/packages/cybernetic-runtime/src/skillValidity/skillValidity.ts @@ -1,4 +1,5 @@ import { + deriveRouteGuards, CAMPAIGN_GOAL_DIRECTIONS, getOperation, isCampaignRef, @@ -27,7 +28,7 @@ import { import { validateWorkflowGraph, type GraphValidationError } from '../scheduling/graphValidation.js'; import { entryRunInputSlots } from '../parentTaskInputs.js'; import { deriveOpBoundProducerShapes } from '../scheduling/deriveOpBoundShapes.js'; -import { deriveRouteGuards, validateRoutes } from '../scheduling/routes.js'; +import { validateRoutes } from '../scheduling/routes.js'; import { bindingStaticSchema } from '../scheduling/bindingStaticSchema.js'; import { validateAgentOpTaskOnlyTools, diff --git a/packages/schemas/src/operations/workflow/index.ts b/packages/schemas/src/operations/workflow/index.ts index fd594ca7..ba685a1f 100644 --- a/packages/schemas/src/operations/workflow/index.ts +++ b/packages/schemas/src/operations/workflow/index.ts @@ -7,6 +7,7 @@ export * from './taskContract.js'; export * from './taskHuman.js'; export * from './taskTemplate.js'; export * from './task.js'; +export * from './routes.js'; export * from './definition.js'; export * from './runResult.js'; export * from './learning.js'; diff --git a/packages/schemas/src/operations/workflow/routes.ts b/packages/schemas/src/operations/workflow/routes.ts new file mode 100644 index 00000000..63942790 --- /dev/null +++ b/packages/schemas/src/operations/workflow/routes.ts @@ -0,0 +1,120 @@ +/** + * Routes: a fork stored on the task that decides it. + * + * A task's `routes` say which tasks its outcome enables. The enabled tasks' + * guards are derived from them β€” at materialization, the form every run and + * every validity check uses, and in the designer β€” so the stored workflow + * holds the fork once, on its source, and the guards cannot drift from it. + */ +import { + predicateCombinator, + predicateExpressions, + type WorkflowPollUntil, + type WorkflowTask, + type WorkflowTaskRoute, + type WorkflowWhen, +} from './task.js'; + +/** A route as an operator reads it: `when β†’ enable `. */ +export function describeRoute(route: WorkflowTaskRoute): string { + const joiner = predicateCombinator(route.when) === 'anyOf' ? ' or ' : ' and '; + return `when ${predicateExpressions(route.when).join(joiner)} β†’ enable ${route.enable.join(', ')}`; +} + +/** `output. …` read from the source, rewritten to how the target reads it. */ +function fromSource(sourceId: string, expression: string): string { + return `tasks.${sourceId}.${expression.trim()}`; +} + +/** + * The guard a task enabled by these routes of one source runs under, or null + * when the combination has no single-level form. Several routes combine as + * `anyOf`; a route that is itself an `allOf` combines with nothing, because the + * `when` grammar has no nesting. + */ +export function routeGuardFor( + sourceId: string, + routes: readonly WorkflowTaskRoute[], +): WorkflowWhen | null { + const first = routes[0]; + if (first === undefined) return null; + if (routes.length === 1) return toWhen(sourceId, first.when); + if ( + routes.some( + (r) => predicateCombinator(r.when) === 'allOf' && predicateExpressions(r.when).length > 1, + ) + ) { + return null; + } + const expressions = [ + ...new Set( + routes.flatMap((r) => predicateExpressions(r.when)).map((e) => fromSource(sourceId, e)), + ), + ]; + return expressions.length === 1 + ? { expression: expressions[0]!, onMissingRef: 'skip' } + : { anyOf: expressions, onMissingRef: 'skip' }; +} + +function toWhen(sourceId: string, predicate: WorkflowPollUntil): WorkflowWhen { + const expressions = predicateExpressions(predicate).map((e) => fromSource(sourceId, e)); + switch (predicateCombinator(predicate)) { + case 'expression': + return { expression: expressions[0]!, onMissingRef: 'skip' }; + case 'anyOf': + return { anyOf: expressions, onMissingRef: 'skip' }; + case 'allOf': + return { allOf: expressions, onMissingRef: 'skip' }; + } +} + +/** Every task some route enables, with the sources and routes that enable it. */ +export function routeClaims( + tasks: readonly WorkflowTask[], +): Map> { + const byTarget = new Map>(); + for (const task of tasks) { + for (const route of task.routes ?? []) { + for (const target of route.enable) { + const list = byTarget.get(target) ?? []; + list.push({ source: task.taskId, route }); + byTarget.set(target, list); + } + } + } + return byTarget; +} + +/** + * Give each routed task the guard its routes imply and its source as a + * dependency. A target the routes cannot guard β€” claimed by two sources, + * carrying its own `when`, or enabled by an inexpressible combination β€” is + * left as stored; validation refuses that workflow before it can run. + */ +export function deriveRouteGuards(tasks: WorkflowTask[]): WorkflowTask[] { + const byTarget = routeClaims(tasks); + if (byTarget.size === 0) return tasks; + return tasks.map((task) => { + const claimed = byTarget.get(task.taskId); + if (!claimed || task.when !== undefined) return task; + const sources = new Set(claimed.map((c) => c.source)); + if (sources.size !== 1) return task; + const [source] = sources; + const guard = routeGuardFor( + source!, + claimed.map((c) => c.route), + ); + if (!guard) return task; + return { + ...task, + when: guard, + dependsOn: [...new Set([...(task.dependsOn ?? []), source!])], + }; + }); +} + +/** The task whose routes enable this one, when exactly one does. */ +export function routeSourceOf(tasks: readonly WorkflowTask[], taskId: string): string | undefined { + const sources = new Set((routeClaims(tasks).get(taskId) ?? []).map((c) => c.source)); + return sources.size === 1 ? [...sources][0] : undefined; +} diff --git a/packages/schemas/src/operations/workflow/task.ts b/packages/schemas/src/operations/workflow/task.ts index c83cab64..e9e2a6f8 100644 --- a/packages/schemas/src/operations/workflow/task.ts +++ b/packages/schemas/src/operations/workflow/task.ts @@ -181,12 +181,6 @@ export const WorkflowTaskRouteSchema = z .strict(); export type WorkflowTaskRoute = z.infer; -/** A route as an operator reads it: `when β†’ enable `. */ -export function describeRoute(route: WorkflowTaskRoute): string { - const joiner = predicateCombinator(route.when) === 'anyOf' ? ' or ' : ' and '; - return `when ${predicateExpressions(route.when).join(joiner)} β†’ enable ${route.enable.join(', ')}`; -} - export const WorkflowTaskPollSchema = z .object({ /** diff --git a/packages/web-product/src/ui/components/skill-designer/SkillInspector.tsx b/packages/web-product/src/ui/components/skill-designer/SkillInspector.tsx index 0a157737..bca3a058 100644 --- a/packages/web-product/src/ui/components/skill-designer/SkillInspector.tsx +++ b/packages/web-product/src/ui/components/skill-designer/SkillInspector.tsx @@ -16,6 +16,9 @@ import { CampaignFormDialog } from '../workflow/CampaignFormDialog.js'; import { inferTaskType, workflowWhenView, + deriveRouteGuards, + describeRoute, + routeSourceOf, type Workflow, type WorkflowTask, type SkillDiagnostic, @@ -1334,6 +1337,39 @@ function TaskInspector({ )} + {task.routes && ( +
+ + {task.routes.map((route, i) => ( + {describeRoute(route)} + ))} + +
+ )} + + {(() => { + const source = routeSourceOf(workflow.tasks, task.taskId); + const guard = source + ? deriveRouteGuards(workflow.tasks).find((t) => t.taskId === task.taskId)?.when + : undefined; + if (!source || !guard) return null; + const view = workflowWhenView(guard); + return ( +
+ + {view.clauses.map((clause) => ( + {clause} + ))} + + } + /> +
+ ); + })()} + {task.when && (() => { const view = workflowWhenView(task.when); diff --git a/packages/web-product/src/ui/components/skill-designer/skill-graph.test.ts b/packages/web-product/src/ui/components/skill-designer/skill-graph.test.ts index b60a4027..875c6794 100644 --- a/packages/web-product/src/ui/components/skill-designer/skill-graph.test.ts +++ b/packages/web-product/src/ui/components/skill-designer/skill-graph.test.ts @@ -161,3 +161,36 @@ describe('branch fans β€” decision reads at the producer', () => { expect(producer.branchSubjects).toEqual(['decision']); }); }); + +describe('routes β€” one source, labelled edges to each task it enables', () => { + const routed = makeWorkflow([ + { + taskId: 'triage', + operation: 'ai.decision.decide', + routes: [ + { when: { expression: "output.answers.team.value == 'billing'" }, enable: ['billing'] }, + { when: { expression: "output.answers.team.value == 'technical'" }, enable: ['technical'] }, + ], + }, + { taskId: 'billing', agent: 'r' }, + { taskId: 'technical', agent: 'r' }, + ]); + const graph = buildSkillGraph(routed); + + it('draws each route as a conditional edge from the source', () => { + const billing = graph.edges.find((e) => e.id === 'triage->billing'); + expect(billing?.data).toEqual({ conditional: true }); + expect(billing?.label).toContain("'billing'"); + expect(graph.edges.find((e) => e.id === 'triage->technical')?.label).toContain("'technical'"); + }); + + it('does not treat a routed task as an entry', () => { + expect(graph.edges.some((e) => e.target === 'billing' && e.source !== 'triage')).toBe(false); + }); + + it('keeps the stored task on the node, with no guard written into it', () => { + const data = graph.nodes.find((n) => n.id === 'billing')?.data as TaskNodeData; + expect(data.task.when).toBeUndefined(); + expect(data.when?.clauses).toEqual(["triage.output.answers.team.value == 'billing'"]); + }); +}); diff --git a/packages/web-product/src/ui/components/skill-designer/skill-graph.ts b/packages/web-product/src/ui/components/skill-designer/skill-graph.ts index 332d370e..16fd114a 100644 --- a/packages/web-product/src/ui/components/skill-designer/skill-graph.ts +++ b/packages/web-product/src/ui/components/skill-designer/skill-graph.ts @@ -1,5 +1,6 @@ import { MarkerType, type Node, type Edge } from '@xyflow/react'; import { + deriveRouteGuards, inferTaskType, predicateExpressions, workflowWhenView, @@ -493,7 +494,10 @@ function applyBranchFans( export function buildSkillGraph(workflow: Workflow, opts?: BuildSkillGraphOptions): SkillGraph { const diagnostics = indexDiagnostics(opts?.validity); - const tasks = workflow.tasks; + // Edges and guards are drawn as the run sees them β€” a routed task carries the + // guard its source's routes imply β€” while the node keeps the stored task the + // inspector edits. + const tasks = deriveRouteGuards(workflow.tasks); const dependedOn = new Set(); for (const t of tasks) for (const d of t.dependsOn ?? []) dependedOn.add(d); @@ -599,7 +603,7 @@ export function buildSkillGraph(workflow: Workflow, opts?: BuildSkillGraphOption agentStepCount: agentRef ? opts?.agentStepCounts?.get(agentRef) : undefined, isEntry, severity: severityFor(diagnostics.byTask.get(task.taskId)), - task, + task: workflow.tasks[index] ?? task, }, });