Skip to content

feat(engine): evaluate conditional edges independently #55

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Zhoujie628 merged 3 commits into from
Sep 9, 2026
Merged

feat(engine): evaluate conditional edges independently #55

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ef65401
feat(engine): evaluate conditional edges independently
Zhoujie628 Sep 9, 2026
5571208
docs: describe the per-edge conditional routing contract
Zhoujie628 Sep 9, 2026
167dc78
style: use explicit imports in the routing changeset
Zhoujie628 Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,18 @@

## [Unreleased]

### Breaking: per-edge conditional routing

- `onRoute` now evaluates one conditional edge at a time: `RouteRequest` describes a single edge (step, condition,
upstream window) and the callback returns `RouteDecision.allow()` /
`deny()` for that edge. Unconditional edges always run and bypass the callback; allowed conditional edges activate in
parallel with them. This replaces the previous N-choose-1 candidate selection (`RouteDecision.nextStep` and
`RouteRequest.candidates` are removed)
and fixes mixed-edge steps, where all unconditional edges were not previously guaranteed to activate.
- Workflow validation now rejects duplicate outgoing targets on one step, null edges and blank targets.
- After all route decisions succeed, `route_decision` is emitted once per outgoing edge, including unconditional edges,
instead of once per step. A callback failure emits the workflow error and no partial per-edge decisions.

## [0.0.4] — 2026-09-08

- Make the Spring Boot A2A server auto-configuration opt-in: server beans are registered only
Expand Down
75 changes: 45 additions & 30 deletions docs/en/API_REFERENCE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,8 +274,9 @@ interface ControlPoint {
```

onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content.
onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop.
Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent.
onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation
returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No
echo-success, implicit route approval or automatic consent.

`DefaultControlPoint` preserves those fail-fast defaults and can delegate `onNegotiation` to an injected
`NegotiationStrategy`. Implement `NegotiationStrategy.resolve(NegotiationRequest)` when negotiation policy is the only
Expand All @@ -295,30 +296,30 @@ Override to receive real-time execution events. Event types are defined in `Even

### EventType

| Constant | Description |
|--------------------------|--------------------------------------------------------------|
| `STEP_START` | A workflow step began |
| `STEP_COMPLETE` | A workflow step completed |
| `TASK_REQUEST` | A task was dispatched to an agent |
| `TASK_RESPONSE` | A task response was received |
| `TASK_STATUS_CHANGED` | A task's status changed (pending → running → success/failed) |
| `AGENT_REQUEST` | A message was sent to an agent |
| `AGENT_RESPONSE` | A response was received from an agent |
| `AGENT_STATUS_UPDATE` | Agent SSE status update (SUBMITTED, WORKING, etc.) |
| `AGENT_ARTIFACT_UPDATE` | Agent SSE artifact update |
| `AGENT_MESSAGE_EVENT` | Agent SSE message event |
| `NEGOTIATION_REQUEST` | Agent requested negotiation (INPUT_REQUIRED) |
| `NEGOTIATION_RESOLVED` | Clarification was sent to agent |
| `NEGOTIATION_FAILED` | Negotiation could not be resolved |
| `AUTHORIZATION_REQUEST` | Reserved; not currently emitted (authorization results arrive via the `ExtensionSender` future) |
| `AUTHORIZATION_RESOLVED` | Reserved; not currently emitted |
| Constant | Description |
|--------------------------|----------------------------------------------------------------------------------------------------------|
| `STEP_START` | A workflow step began |
| `STEP_COMPLETE` | A workflow step completed |
| `TASK_REQUEST` | A task was dispatched to an agent |
| `TASK_RESPONSE` | A task response was received |
| `TASK_STATUS_CHANGED` | A task's status changed (pending → running → success/failed) |
| `AGENT_REQUEST` | A message was sent to an agent |
| `AGENT_RESPONSE` | A response was received from an agent |
| `AGENT_STATUS_UPDATE` | Agent SSE status update (SUBMITTED, WORKING, etc.) |
| `AGENT_ARTIFACT_UPDATE` | Agent SSE artifact update |
| `AGENT_MESSAGE_EVENT` | Agent SSE message event |
| `NEGOTIATION_REQUEST` | Agent requested negotiation (INPUT_REQUIRED) |
| `NEGOTIATION_RESOLVED` | Clarification was sent to agent |
| `NEGOTIATION_FAILED` | Negotiation could not be resolved |
| `AUTHORIZATION_REQUEST` | Reserved; not currently emitted (authorization results arrive via the `ExtensionSender` future) |
| `AUTHORIZATION_RESOLVED` | Reserved; not currently emitted |
| `NOTIFICATION` | Reserved; not currently emitted (subscription events arrive via the `NotificationSubscription` callback) |
| `ROUTE_DECISION` | Route decision was made |
| `WORKFLOW_COMPLETE` | DAG scheduling ended; check success, not all nodes necessarily ran |
| `START` | Workflow execution started |
| `COMPLETE` | Workflow execution completed successfully |
| `ERROR` | Workflow execution failed |
| `CLOSE` | Engine client closed |
| `ROUTE_DECISION` | Successful routing result for one outgoing edge |
| `WORKFLOW_COMPLETE` | DAG scheduling ended; check success, not all nodes necessarily ran |
| `START` | Workflow execution started |
| `COMPLETE` | Workflow execution completed successfully |
| `ERROR` | Workflow execution failed |
| `CLOSE` | Engine client closed |

---

Expand Down Expand Up @@ -542,12 +543,26 @@ flattened metadata are transport diagnostics, not substitutes for the complete b
| `stepOutputs` | `Map<String, Map>` | Outputs keyed by step name |
| `error` | `String` | Error message (if failed) |

### RouteDecision
### RouteRequest / RouteDecision

| Field | Type | Description |
|------------|----------|----------------------|
| `nextStep` | `String` | Next step to execute |
| `reason` | `String` | Decision reason |
`RouteRequest` describes one conditional edge, not the complete outgoing-edge list:

| Field | Type | Description |
|------------------|-----------------------------|------------------------------------------|
| `executionId` | `String` | Current workflow execution |
| `stepName` | `String` | Source step |
| `nextStep` | `String` | Target of this edge |
| `condition` | `String` | Nonblank business condition to evaluate |
| `workflowInput` | `WorkflowInput` | Selected protocol-neutral context window |
| `currentResults` | `List<TaskExecutionResult>` | Results produced by the source step |

| `RouteDecision` field | Type | Description |
|-----------------------|-----------|-------------------------------------|
| `allowed` | `boolean` | Whether this edge must be activated |
| `reason` | `String` | Optional business decision reason |

Use `RouteDecision.allow(...)` or `RouteDecision.deny(...)`. Every blank-condition edge is activated without invoking
the callback; every allowed conditional edge is activated as well, so one source step can fan out to multiple targets.

### WorkflowSearchResult

Expand Down
22 changes: 14 additions & 8 deletions docs/en/BUSINESS_CALLBACKS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,9 @@ interface ControlPoint {
```

onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content.
onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop.
Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent.
onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation
returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No
echo-success, implicit route approval or automatic consent.
See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples.

```java
Expand All @@ -27,8 +28,8 @@ ControlPoint callbacks = ControlPoint.builder()
.onSelfTask(request -> CompletableFuture.completedFuture(
TaskResult.success(List.of(Map.of(
"sourceResults", request.getWorkflowInput().upstreamResults())))))
.onRoute(request -> CompletableFuture.failedFuture(
new IllegalStateException("Supply a routing policy for " + request.stepName())))
.onRoute(request -> CompletableFuture.completedFuture(
RouteDecision.deny("Replace with host condition evaluation")))
.onNegotiation(request -> CompletableFuture.completedFuture(
new NegotiationReply.Stop("manual.required", "Manual confirmation required")))
.build();
Expand Down Expand Up @@ -145,11 +146,16 @@ SDK-specific content exception classes.

## 6. Routing, concurrency and failures

RouteRequest(executionId, stepName, workflowInput, currentResults, candidates). Candidates are RouteOption(nextStep,
condition); return RouteDecision.builder().nextStep(allowed).build(). Unconditional edges fan out without calling
onRoute; conditional selection must be an allowed destination.
RouteRequest (executionId, stepName, nextStep, condition, workflowInput, currentResults) describes one conditional edge.
Return RouteDecision.allow (reason) to activate it or RouteDecision.deny (reason) to skip it. Null, empty and
whitespace-only conditions are unconditional: they bypass onRoute and always activate. Each nonblank edge is evaluated
independently, so the result can activate zero, one or many conditional targets together with all unconditional targets.
The engine waits for every route decision before dispatching any successor; a missing, null, exceptional or timed-out
decision fails the workflow and cannot cause partial successor dispatch. Route failure messages identify the source and
target edge while preserving the original cause.

Callbacks may run concurrently; do not hold shared mutable current-task state. Each workflow task activation
Callbacks for different conditional edges may run concurrently; do not hold shared mutable current-task state. Each
workflow task activation
(preparation and dispatch combined) is bounded by the client timeout (default sendTimeoutSeconds=600). Routing has a
callback timeout; dispatch/negotiation additionally has its own total wait deadline. Cancellation/timeout prevents late
sends; it does not automatically stop external business/LLM work. Hosts own cancellation of their resources. Synchronous
Expand Down
21 changes: 13 additions & 8 deletions docs/en/DESIGN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,8 +248,9 @@ interface ControlPoint {
```

onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content.
onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop.
Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent.
onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation
returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No
echo-success, implicit route approval or automatic consent.
See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples.

## 5. A2A-T Extension Model
Expand Down Expand Up @@ -282,12 +283,16 @@ open until the host closes them.
A step's `next` list holds `JumpCondition(step, condition)` entries. The routing rule is:

- **No `next`** - terminal; the step completes the branch.
- **All conditions empty** - unconditional fan-out: dispatch every non-terminal next step in parallel.
- **Has conditions** - conditional: call `ControlPoint.onRoute`, which returns a single `RouteDecision.nextStep`. The
engine enforces that the returned step is among the declared conditions; an invalid return fails the workflow with an
error.

This makes conditional branches an N-choose-1 selection and keeps unconditional fan-out as automatic parallel dispatch.
- **Blank condition** - unconditional edge: activate it without calling `ControlPoint.onRoute`. Null, empty and
whitespace-only values are all blank.
- **Nonblank condition** - conditional edge: invoke `ControlPoint.onRoute` once for that edge and activate it only when
the returned `RouteDecision.allowed` is true.

All conditional edges are evaluated independently and may run concurrently. The engine waits for every decision before
atomically activating all unconditional and allowed conditional targets. Multiple true decisions therefore create a
parallel fan-out; all false decisions are valid and end the branch when no unconditional target exists. Callback
failure, timeout or null fails routing before any successor is dispatched. Route failures identify the source and target
edge and preserve the original cause. Duplicate targets from the same source are rejected during graph validation.

---

Expand Down
Loading