Coverage for core/langgraph/runner.py: 94%
2435 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-04 15:34 +0000
1"""
2runner.py — YAML-manifest-pohjainen pipeline-moottori.
4Arkkitehtuuri:
5 ManifestLoader → löytää ja parsii *.yaml manifestit
6 ManifestRegistry → indeksoi Agent / AgentInstance / Pipeline -resurssit
7 AgentResolver → yhdistää AgentInstance + based_on Agent (merge)
8 GraphBuilder → rakentaa LangGraph StateGraphin Pipeline-manifestista
9 build_workflow() → palauttaa compilatun LangGraph-workflown nimellä
11Käyttö:
12 from langgraph.runner import build_workflow
13 workflow = build_workflow("review-pr")
14 result = workflow.invoke({"input": "..."})
15"""
16from __future__ import annotations
18import hashlib
19import json
20import logging
21import os
22import re
23import threading
24from dataclasses import dataclass
25from concurrent.futures import ThreadPoolExecutor, as_completed
26from pathlib import Path
27from typing import Annotated, Any
29import yaml
31from langgraph.types import Send
33# SkillLoadError-import (ei sirkulaari-riskiä — _llm importtaa vain stdlibin):
34# testit ja discover-wrapperit lataavat _llm:n tasaisena moduulina (sys.path
35# insert core/langgraph/), main.py qualified (langgraph._llm, PYTHONPATH=/app).
36# Kummatkin pinnat käännetään samaan _llm-tiedostoon; try/except varmistaa,
37# että luokka on sama objekti kuin nostavalla puolella molemmissa konteksteissa.
38try:
39 from _llm import SkillLoadError
40except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
41 from langgraph._llm import SkillLoadError
43# Same dual-import pattern as _llm (flat sys.path and qualified PYTHONPATH=/app)
44try:
45 from _step_functions import StepFunctionExecutor # noqa: F401
46except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
47 from langgraph._step_functions import StepFunctionExecutor # noqa: F401
49# Agent-instruction error code emitted by the engine error set — same
50# dual-import pattern as _llm.
51try:
52 from _instructions import AGENT_INSTRUCTION_MISSING
53except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
54 from langgraph._instructions import AGENT_INSTRUCTION_MISSING
56# Structural tool resolution layer — same dual-import pattern as _llm.
57try:
58 from tools import (
59 RESOLVED_TOOLS_KEY,
60 ToolCollisionError,
61 ToolRegistryError,
62 ToolResolver,
63 build_platform_registry,
64 )
65except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
66 from langgraph.tools import (
67 RESOLVED_TOOLS_KEY,
68 ToolCollisionError,
69 ToolRegistryError,
70 ToolResolver,
71 build_platform_registry,
72 )
74# COMMUNICATION-context projection (declared input.channels reads) — same
75# dual-import pattern as _llm.
76try:
77 from _data_structure import (
78 COMMUNICATION_CONTEXT_KEY,
79 resolve_communication_context,
80 )
81except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
82 from langgraph._data_structure import (
83 COMMUNICATION_CONTEXT_KEY,
84 resolve_communication_context,
85 )
87# Core-owned channel-context storage/commit/seeding primitives — same dual-import
88# pattern. The channel path commits through these; it no longer materializes
89# native LangGraph channels.
90try:
91 from _channels import (
92 CORE_KEY,
93 LAST_VALUE,
94 ChannelSpec,
95 ChannelWorkingSet,
96 parse_channel_specs,
97 seed_committed,
98 unwrap,
99 wrap,
100 )
101except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
102 from langgraph._channels import (
103 CORE_KEY,
104 LAST_VALUE,
105 ChannelSpec,
106 ChannelWorkingSet,
107 parse_channel_specs,
108 seed_committed,
109 unwrap,
110 wrap,
111 )
113# Core-owned reducer layer — same dual-import pattern. The map-reduce
114# accumulator reducer folds concurrent ``Send`` worker writes per step id.
115try:
116 from _reducers import accumulate_map_reduce
117except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
118 from langgraph._reducers import accumulate_map_reduce
120# Agent output envelope error code — same dual-import pattern.
121try:
122 from _envelope import (
123 ENVELOPE_ERROR_CODE,
124 ENVELOPE_STATUSES,
125 is_error_envelope,
126 project_aggregate_metadata,
127 project_envelope_metadata,
128 )
129except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
130 from langgraph._envelope import ( # noqa: F401
131 ENVELOPE_ERROR_CODE,
132 ENVELOPE_STATUSES,
133 is_error_envelope,
134 project_aggregate_metadata,
135 project_envelope_metadata,
136 )
138log = logging.getLogger(__name__)
141class RunHardFailure(Exception):
142 """Run-tason kova virhe: ajo pysähtyy heti (phase failed, exit 1).
144 Nostetaan kun step-tulos kantaa kovan virheen: TRUNCATION_ERROR-koodi
145 (kuratoidun kontekstin katkaisu), EXECUTION_ERROR-koodi (builtinien/stepien
146 deterministinen tekninen vika), TOOL_FAILURE-koodi (agentin työkalupinnan
147 tekninen vika), AGENT_ERROR-koodi (status:error-envelope,
148 ENVELOPE_ERROR_CODE), CALL_NOT_ALLOWED/CALL_INVALID_PARAMS/CALL_FAILED
149 (natiivi deterministic-tool-kutsu), BUILTIN_NAME_COLLISION (nimi sekä
150 builtin- että StepFunction-rekisterissä) tai kontekstiylityksen
151 signatuurin kantava VALIDATION_ERROR (assembled_tokens + context_budget,
152 _llm._context_overflow_result).
153 Kaikki muut virheluokat (budget:*, parse_error, timeout, LLM-virheet)
154 säilyvät nykyisessä partial-semantiikassa. Stepin details kulkeutuu
155 attribuuttina loppu-PATCHin error.details-kenttään.
156 """
158 ENGINE_ERROR_CODES: frozenset[str] = frozenset(
159 {
160 "TIMEOUT",
161 "VALIDATION_ERROR",
162 "CHANNEL_BINDING_ERROR",
163 AGENT_INSTRUCTION_MISSING,
164 "UNRESOLVED_OUTPUT",
165 "BUILTIN_NAME_COLLISION",
166 "EXECUTION_ERROR",
167 ENVELOPE_ERROR_CODE,
168 "TOOL_FAILURE",
169 "CALL_NOT_ALLOWED",
170 "CALL_INVALID_PARAMS",
171 "CALL_FAILED",
172 "RESOURCE_ERROR",
173 "EXHAUSTED",
174 "TRUNCATION_ERROR",
175 # BLOCKED is API-owned, not engine-emitted; it stays in the set so
176 # the guard never rejects a contract-valid value.
177 "BLOCKED",
178 }
179 )
181 HARD_ERROR_CODES: frozenset[str] = frozenset(
182 {
183 "TRUNCATION_ERROR",
184 "TOOL_FAILURE",
185 ENVELOPE_ERROR_CODE,
186 "CALL_NOT_ALLOWED",
187 "CALL_INVALID_PARAMS",
188 "CALL_FAILED",
189 "BUILTIN_NAME_COLLISION",
190 }
191 )
193 def __init__(self, code: str, message: str, details: dict | None = None) -> None:
194 if code not in self.ENGINE_ERROR_CODES:
195 raise ValueError(
196 f"RunHardFailure code {code!r} is not a member of the documented "
197 f"engine error-code set {sorted(self.ENGINE_ERROR_CODES)}"
198 )
199 if not isinstance(message, str) or not message:
200 raise ValueError("RunHardFailure message must be a non-empty str")
201 super().__init__(message)
202 self.code = code
203 self.message = message
204 self.details = details
207ENGINE_ERROR_CODES = RunHardFailure.ENGINE_ERROR_CODES
210# ---------------------------------------------------------------------------
211# Polut
212# ---------------------------------------------------------------------------
214PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", "/project"))
215MANIFEST_DIRS = [
216 PROJECT_ROOT / ".agent-platform" / "workflows",
217 PROJECT_ROOT / ".agent-platform" / "agents",
218]
219SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "manifest.schema.json"
222def resolve_manifest_dirs(clone_root: Path) -> list[Path]:
223 """Palauttaa olemassa olevat manifestihakemistot kloonatusta manifest-reposta.
225 Skannaa ainoastaan `.agent-platform/workflows` ja `.agent-platform/agents` —
226 ei koskaan koko kloonia. [] jos kumpaakaan ei ole (kutsuja → fail-fast).
227 """
228 return [
229 d for d in (
230 clone_root / ".agent-platform" / "workflows",
231 clone_root / ".agent-platform" / "agents",
232 )
233 if d.is_dir()
234 ]
236# ---------------------------------------------------------------------------
237# Template-resoluutio
238# ---------------------------------------------------------------------------
240_TMPL_RE = re.compile(r'\{\{\s*(.*?)\s*\}\}')
242# Sentinel for an unresolvable dot-path segment (distinct from a resolved None).
243_UNRESOLVED = object()
246def _split_key_indices(part: str) -> tuple[str, list[int]]:
247 """Split a dot-path segment into a base key and a list of indices.
249 ``'a[0][1]'`` → ``('a', [0, 1])``; ``'a'`` → ``('a', [])``; ``'[0]'`` →
250 ``('', [0])``. Only non-negative indices are supported (list access).
251 """
252 m = re.fullmatch(r'([^\[]*)((\[\d+\])*)', part)
253 base = m.group(1)
254 indices = [int(x) for x in re.findall(r'\d+', m.group(2))]
255 return base, indices
258def _apply_indices(obj: Any, indices: list[int]) -> Any:
259 """Index a (possibly chained) list/tuple; ``_UNRESOLVED`` on out-of-range."""
260 for idx in indices:
261 if not isinstance(obj, (list, tuple)) or not (0 <= idx < len(obj)):
262 return _UNRESOLVED
263 obj = obj[idx]
264 return obj
267def _dot_get_segment(obj: Any, part: str) -> Any:
268 """Resolve one dot-path segment against a dict or an indexed sequence.
270 A segment may be a bare key (``'a'``), a keyed list-index (``'a[0]'``,
271 ``'a[0][1]'``), or a bare index into a sequence. Returns the resolved value
272 or the ``_UNRESOLVED`` sentinel when the segment cannot be resolved (missing
273 key, non-dict/non-sequence mid-path, or out-of-range index).
274 """
275 base, indices = _split_key_indices(part)
276 if isinstance(obj, dict):
277 if base:
278 if base not in obj:
279 return _UNRESOLVED
280 obj = obj[base]
281 return _apply_indices(obj, indices)
282 if isinstance(obj, (list, tuple)):
283 if indices:
284 return _apply_indices(obj, indices)
285 if base and base.isdigit():
286 return _apply_indices(obj, [int(base)])
287 return _UNRESOLVED
288 return _UNRESOLVED
291def _dot_get(obj: Any, path: str) -> Any:
292 """Navigate nested dicts/lists via dot-path: 'a.b.c' → obj['a']['b']['c'].
294 Supports list-index segments (``results[1].field``) and dot-into-aggregate
295 traversal (missing key / non-dict-non-list mid-path / out-of-range index
296 resolve to None — lenient semantics for non-output refs).
297 """
298 for part in path.split('.'):
299 obj = _dot_get_segment(obj, part)
300 if obj is _UNRESOLVED:
301 return None
302 return obj
305class _ExpressionError(ValueError):
306 """A malformed template/condition expression at the evaluation boundary.
308 Raised by the expression tokenizer, parser, and evaluator so
309 ``_resolve_template`` relabels only design-level expression failures as
310 ``VALIDATION_ERROR``. Subclasses ``ValueError`` so callers that already
311 treat expression failures as ``ValueError`` keep working.
312 """
315class _UnresolvedReference(_ExpressionError):
316 """A strict dot-path read hit an absent key or a non-mapping segment.
318 Subclasses ``ValueError`` (via ``_ExpressionError``) so existing lenient
319 callers that treat any strict read failure as "unresolved"
320 (``__call_emit`` optional outputs, ``_builtins._resolve_dot_path``) keep
321 working while the template/condition boundary can still tell an unresolved
322 reference apart from a parse error.
323 """
326def _dot_get_strict(obj: Any, path: str) -> Any:
327 """Dot-path navigation that raises on an unresolved reference.
329 A reference is unresolved when a key on the path is absent or a segment is
330 not a mapping — used by condition evaluation so an undefined or skipped
331 step's field is never silently evaluated to falsy (ADR-0012 spirit). Also
332 used by ``__call_emit`` to distinguish a resolved output source from an
333 unresolved one (required output → hard error, ADR-0012).
334 """
335 for part in path.split('.'):
336 if obj is None:
337 raise _UnresolvedReference(
338 f"Unresolved reference '{path}': '{part}' is not a mapping"
339 )
340 obj = _dot_get_segment(obj, part)
341 if obj is _UNRESOLVED:
342 raise _UnresolvedReference(
343 f"Unresolved reference '{path}': '{part}' is not a mapping"
344 )
345 return obj
348_CHANNEL_ADDRESS_ROOT = "channel"
351def _resolve_output_source(final_state: dict, source: str) -> Any:
352 """Resolve a run-end output source, honoring the ``channel.`` address root.
354 A ``channel.<name>[.<field>][[i]]`` address resolves the named channel's
355 committed value out of the flat committed snapshot in ``final_state``
356 (channel keys sit at the top level, there is no ``channel`` root). Any
357 other source — a bare dot-path or step metadata
358 ``<step_id>.<status|error|iteration>`` — resolves as-is via ``_dot_get``.
359 An address with no channel name resolves to ``None``.
360 """
361 if source == _CHANNEL_ADDRESS_ROOT:
362 return None
363 if not source.startswith(_CHANNEL_ADDRESS_ROOT + "."):
364 return _dot_get(final_state, source)
365 channel_path = source[len(_CHANNEL_ADDRESS_ROOT) + 1:]
366 if not channel_path:
367 return None
368 return _dot_get(final_state, channel_path)
371def materialize_pipeline_output(pipeline_spec: dict, final_state: dict) -> Any:
372 """Resolve a Pipeline's declared run-end output.
374 Reads ``spec.output.source`` (a dot-path, optionally a ``channel.``
375 address) out of ``final_state`` via ``_resolve_output_source`` and returns
376 the surfaced value. Returns ``None`` when no ``spec.output`` is declared,
377 when it has no ``source``, or when the source is unresolved — this is a
378 run-end surfacing (distinct from the emit-time hard-error used for
379 function-interface outputs): an optional run output that cannot be resolved
380 surfaces as None, never a run-end failure.
381 """
382 out_spec = (pipeline_spec or {}).get("output")
383 if not out_spec:
384 return None
385 source = out_spec.get("source")
386 if not source:
387 return None
388 return _resolve_output_source(final_state, source)
391def _merge_side_channel(base: dict | None, additions: dict) -> dict:
392 """Yhdistä rinnakkaisen side-channel-envelopen (step_context_metrics /
393 step_tool_meta) kerrokset: myöhempi kirjoitus voittaa (sama semantiikka kuin
394 state-merge {**state, ...})."""
395 merged = dict(base or {})
396 merged.update(additions)
397 return merged
400_MULTI_OPS = ('<=', '>=', '==', '!=')
401_SINGLE_OPS = '+-<>'
402_WORD_BREAK = set("()[],+<=>!'-")
403_CMP_OPS = frozenset(('==', '!=', '<=', '>=', '<', '>'))
406def _tokenize_expr(expr: str) -> list[tuple[str, Any]]:
407 """Split an expression into (kind, value) tokens.
409 Kinds: 'op' (+ - <= >= == != < >), 'punct' ( ) [ ] ,, 'num', 'str',
410 'name'. A 'name' is a dot-path reference or a bare word (e.g. an unquoted
411 comparison literal, true/false/None, and/or/not).
412 """
413 tokens: list[tuple[str, Any]] = []
414 i, n = 0, len(expr)
415 while i < n:
416 ch = expr[i]
417 if ch.isspace():
418 i += 1
419 continue
420 if ch in '()[],':
421 tokens.append(('punct', ch))
422 i += 1
423 continue
424 if ch in '+-<=>!':
425 two = expr[i:i + 2]
426 if two in _MULTI_OPS:
427 tokens.append(('op', two))
428 i += 2
429 continue
430 if ch in _SINGLE_OPS:
431 tokens.append(('op', ch))
432 i += 1
433 continue
434 raise _ExpressionError(f"Invalid operator in expression: {two!r}")
435 if ch in ('\'', '"'):
436 j = i + 1
437 buf: list[str] = []
438 while j < n and expr[j] != ch:
439 buf.append(expr[j])
440 j += 1
441 if j >= n:
442 raise _ExpressionError(f"Unterminated string literal in expression: {expr!r}")
443 tokens.append(('str', ''.join(buf)))
444 i = j + 1
445 continue
446 if ch.isdigit() or (ch == '.' and i + 1 < n and expr[i + 1].isdigit()):
447 j = i
448 if ch == '.':
449 j += 1
450 while j < n and expr[j].isdigit():
451 j += 1
452 tokens.append(('num', float(expr[i:j])))
453 else:
454 while j < n and expr[j].isdigit():
455 j += 1
456 if j < n and expr[j] == '.' and j + 1 < n and expr[j + 1].isdigit():
457 j += 1
458 while j < n and expr[j].isdigit():
459 j += 1
460 tokens.append(('num', float(expr[i:j])))
461 else:
462 tokens.append(('num', int(expr[i:j])))
463 i = j
464 continue
465 j = i
466 while j < n and not expr[j].isspace() and expr[j] not in _WORD_BREAK:
467 j += 1
468 while j < n and expr[j] == '[':
469 index_group = re.match(r"\[\d+\]", expr[j:])
470 if index_group is None:
471 break
472 j += index_group.end()
473 while j < n and not expr[j].isspace() and expr[j] not in _WORD_BREAK:
474 j += 1
475 if j == i:
476 raise _ExpressionError(f"Unexpected character in expression: {expr[i:]!r}")
477 tokens.append(('name', expr[i:j]))
478 i = j
479 return tokens
482# AST node shapes:
483# ('lit', value) literal (str/num/bool/None)
484# ('ref', name) dot-path reference
485# ('list', [node, ...]) [a, b, c] literal
486# ('neg', node) unary minus
487# ('not', node) not X
488# ('cmp', op, left, right) comparison
489# ('binop', 'and'|'or'|'+', l, r) binary operator
492class _ExprParser:
493 """Recursive-descent parser for template/condition expressions.
495 Precedence (loosest first): or < and < not < comparison < '+' < unary.
496 Grammar:
497 expr := or_expr
498 or_expr := and_expr ('or' and_expr)*
499 and_expr := not_expr ('and' not_expr)*
500 not_expr := 'not' not_expr | cmp_expr
501 cmp_expr := add_expr (cmp_op add_expr)?
502 add_expr := unary_expr ('+' unary_expr)*
503 unary_expr:= '-' unary_expr | primary
504 primary := num | str | true/false/None | list | '(' expr ')' | name
505 `and`/`or` parse into a full AST but EVALUATE with left-to-right
506 short-circuit, so the right operand is only resolved if needed and an
507 unresolved read there never raises.
508 """
510 def __init__(self, tokens: list[tuple[str, Any]]) -> None:
511 self._tokens = tokens
512 self._i = 0
514 def parse(self) -> tuple:
515 node = self._or_expr()
516 if self._i != len(self._tokens):
517 tok = self._tokens[self._i]
518 raise _ExpressionError(f"Unexpected token in expression: {tok[1]!r}")
519 return node
521 def _peek(self) -> tuple[str, Any] | None:
522 return self._tokens[self._i] if self._i < len(self._tokens) else None
524 def _next(self) -> tuple[str, Any]:
525 tok = self._tokens[self._i]
526 self._i += 1
527 return tok
529 def _peek_name(self, kw: str) -> bool:
530 tok = self._peek()
531 return tok is not None and tok[0] == 'name' and tok[1] == kw
533 def _peek_op(self, op: str) -> bool:
534 tok = self._peek()
535 return tok is not None and tok[0] == 'op' and tok[1] == op
537 def _expect_punct(self, value: str) -> None:
538 tok = self._next()
539 if tok != ('punct', value):
540 raise _ExpressionError(f"Expected '{value}' in expression, got {tok[1]!r}")
542 def _or_expr(self) -> tuple:
543 node = self._and_expr()
544 while self._peek_name('or'):
545 self._next()
546 right = self._and_expr()
547 node = ('binop', 'or', node, right)
548 return node
550 def _and_expr(self) -> tuple:
551 node = self._not_expr()
552 while self._peek_name('and'):
553 self._next()
554 right = self._not_expr()
555 node = ('binop', 'and', node, right)
556 return node
558 def _not_expr(self) -> tuple:
559 if self._peek_name('not'):
560 self._next()
561 return ('not', self._not_expr())
562 return self._cmp_expr()
564 def _cmp_expr(self) -> tuple:
565 node = self._add_expr()
566 tok = self._peek()
567 if tok is not None and tok[0] == 'op' and tok[1] in _CMP_OPS:
568 self._next()
569 right = self._add_expr()
570 node = ('cmp', tok[1], node, right)
571 return node
573 def _add_expr(self) -> tuple:
574 node = self._unary_expr()
575 while self._peek_op('+'):
576 self._next()
577 right = self._unary_expr()
578 node = ('binop', '+', node, right)
579 return node
581 def _unary_expr(self) -> tuple:
582 if self._peek_op('-'):
583 self._next()
584 node = self._unary_expr()
585 if node[0] == 'lit' and isinstance(node[1], (int, float)) and not isinstance(node[1], bool):
586 return ('lit', -node[1])
587 return ('neg', node)
588 return self._primary()
590 def _primary(self) -> tuple:
591 tok = self._peek()
592 if tok is None:
593 raise _ExpressionError("Unexpected end of expression")
594 kind, value = tok
595 if kind == 'num':
596 self._next()
597 return ('lit', value)
598 if kind == 'str':
599 self._next()
600 return ('lit', value)
601 if kind == 'punct':
602 if value == '(':
603 self._next()
604 node = self._or_expr()
605 self._expect_punct(')')
606 return node
607 if value == '[':
608 self._next()
609 items: list[tuple] = []
610 if self._peek() != ('punct', ']'):
611 while True:
612 items.append(self._or_expr())
613 if self._peek() == ('punct', ','):
614 self._next()
615 continue
616 break
617 self._expect_punct(']')
618 return ('list', items)
619 raise _ExpressionError(f"Unexpected punctuation in expression: {value!r}")
620 if kind == 'name':
621 self._next()
622 lowered = value.lower()
623 if lowered == 'true':
624 return ('lit', True)
625 if lowered == 'false':
626 return ('lit', False)
627 if lowered in ('none', 'null'):
628 return ('lit', None)
629 if value in ('and', 'or', 'not'):
630 raise _ExpressionError(f"Unexpected keyword in expression: {value!r}")
631 return ('ref', value)
632 raise _ExpressionError(f"Unexpected token in expression: {value!r}")
635def _is_number(value: Any) -> bool:
636 return isinstance(value, (int, float)) and not isinstance(value, bool)
639def _add_values(left: Any, right: Any) -> Any:
640 """Binary `+`. Lists concatenate, numbers add, strings concatenate."""
641 if isinstance(left, list) and isinstance(right, list):
642 return left + right
643 if _is_number(left) and _is_number(right):
644 return left + right
645 if isinstance(left, str) and isinstance(right, str):
646 return left + right
647 raise _ExpressionError(
648 f"Unsupported operands for '+': {type(left).__name__} + {type(right).__name__}"
649 )
652def _eval_comparison_rhs(node: tuple, ctx: dict, strict: bool) -> Any:
653 """Comparison right-hand side. A bare non-dotted reference compares as a
654 string literal (preserving `profile == PR`); everything else evaluates
655 normally (numbers, quoted strings, dotted refs, lists, grouped exprs)."""
656 if node[0] == 'ref' and '.' not in node[1]:
657 return node[1]
658 return _eval_ast(node, ctx, strict)
661def _eval_cmp(op: str, left_node: tuple, right_node: tuple, ctx: dict, strict: bool) -> bool:
662 left = _eval_ast(left_node, ctx, strict)
663 right = _eval_comparison_rhs(right_node, ctx, strict)
664 try:
665 if op == '==':
666 return left == right
667 if op == '!=':
668 return left != right
669 if op == '<=':
670 return left <= right
671 if op == '>=':
672 return left >= right
673 if op == '<':
674 return left < right
675 if op == '>':
676 return left > right
677 except TypeError:
678 return False
679 raise _ExpressionError(f"Unknown comparison operator: {op!r}")
682def _eval_ast(node: tuple, ctx: dict, strict: bool) -> Any:
683 """Evaluate a parsed AST. `strict=True` raises on an unresolved reference
684 read (condition/exit_condition); `and`/`or` short-circuit left-to-right."""
685 kind = node[0]
686 if kind == 'lit':
687 return node[1]
688 if kind == 'ref':
689 return _dot_get_strict(ctx, node[1]) if strict else _dot_get(ctx, node[1])
690 if kind == 'list':
691 return [_eval_ast(item, ctx, strict) for item in node[1]]
692 if kind == 'neg':
693 operand = _eval_ast(node[1], ctx, strict)
694 if not _is_number(operand):
695 raise _ExpressionError(
696 f"Unsupported operand for unary '-': {type(operand).__name__}"
697 )
698 return -operand
699 if kind == 'not':
700 return not bool(_eval_ast(node[1], ctx, strict))
701 if kind == 'cmp':
702 return _eval_cmp(node[1], node[2], node[3], ctx, strict)
703 if kind == 'binop':
704 op, left_node, right_node = node[1], node[2], node[3]
705 if op == 'and':
706 if not bool(_eval_ast(left_node, ctx, strict)):
707 return False
708 return bool(_eval_ast(right_node, ctx, strict))
709 if op == 'or':
710 if bool(_eval_ast(left_node, ctx, strict)):
711 return True
712 return bool(_eval_ast(right_node, ctx, strict))
713 if op == '+':
714 left = _eval_ast(left_node, ctx, strict)
715 right = _eval_ast(right_node, ctx, strict)
716 return _add_values(left, right)
717 raise _ExpressionError(f"Unknown binary operator: {op!r}")
718 raise _ExpressionError(f"Unknown expression node: {kind!r}")
721def _eval_expr(expr: str, ctx: dict, strict: bool = False) -> Any:
722 """Evaluate a template expression against context dict.
724 `strict` is used by condition evaluation: dot-path access to an unresolved
725 reference raises (ValueError) instead of silently resolving to falsy, and
726 and/or short-circuit left-to-right so a value is never read when the left
727 side already determines the result.
728 """
729 expr = expr.strip()
730 tokens = _tokenize_expr(expr)
731 if not tokens:
732 return _dot_get_strict(ctx, expr) if strict else _dot_get(ctx, expr)
733 return _eval_ast(_ExprParser(tokens).parse(), ctx, strict)
736def _resolve_template(value: Any, ctx: dict, strict: bool = False) -> Any:
737 """Recursively resolve {{ expr }} templates in strings / dicts / lists.
739 The template-evaluation boundary: a design-level failure (an unresolved
740 strict reference, or a malformed expression) surfaces as a typed
741 ``RunHardFailure`` here, never as a raw parser ``ValueError`` that a
742 concurrent worker would record as a code-less per-item error.
743 """
744 try:
745 return _resolve_template_inner(value, ctx, strict)
746 except _UnresolvedReference as exc:
747 raise RunHardFailure(
748 "UNRESOLVED_OUTPUT",
749 str(exc),
750 ) from exc
751 except _ExpressionError as exc:
752 raise RunHardFailure(
753 "VALIDATION_ERROR",
754 f"Invalid template/condition expression: {exc}",
755 ) from exc
758def _resolve_template_inner(value: Any, ctx: dict, strict: bool = False) -> Any:
759 if isinstance(value, str):
760 stripped = value.strip()
761 # Entire string is one template → replace with actual type (not string)
762 m = _TMPL_RE.fullmatch(stripped)
763 if m:
764 return _eval_expr(m.group(1), ctx, strict)
765 # Inline substitution → always returns string
766 def _replace(m: re.Match) -> str:
767 result = _eval_expr(m.group(1), ctx, strict)
768 return '' if result is None else str(result)
769 return _TMPL_RE.sub(_replace, value)
770 if isinstance(value, dict):
771 return {k: _resolve_template_inner(v, ctx, strict) for k, v in value.items()}
772 if isinstance(value, list):
773 return [_resolve_template_inner(v, ctx, strict) for v in value]
774 return value
777def _resolve_input(input_spec: Any, state: dict, ctx: dict) -> Any:
778 """
779 Resolve step input spec to a concrete value:
780 "{{ item }}" → template
781 "[host-ai-context-map-path]" → literal placeholder (verbatim pass-through)
782 "gather.doc_files" → dot-path into state
783 {"k": "a.b"} → dict with each value resolved
784 None → None
785 """
786 if input_spec is None:
787 return None
788 if isinstance(input_spec, str):
789 if '{{' in input_spec:
790 return _resolve_template(input_spec, ctx)
791 stripped = input_spec.lstrip()
792 # A bracket-prefixed literal (e.g. a host-ai-context-map-path
793 # placeholder) is NOT a dot-path reference into state — hand it to the
794 # builtin verbatim so it resolves the placeholder itself.
795 if stripped.startswith('['):
796 return input_spec
797 # Dot-path into state (no braces)
798 return _dot_get(state, stripped)
799 if isinstance(input_spec, dict):
800 return {k: _resolve_input(v, state, ctx) for k, v in input_spec.items()}
801 if isinstance(input_spec, list):
802 return [_resolve_input(v, state, ctx) for v in input_spec]
803 return input_spec
806def _parse_json_literal(value: str) -> Any:
807 """A JSON array/object string parsed to its value, else ``value`` verbatim.
809 An ``iteration`` literal may arrive as a JSON string (``'["a", "b"]'``)
810 when the flow form is quoted in the manifest. A non-JSON string is returned
811 unchanged so the caller rejects it as a scalar.
812 """
813 stripped = value.strip()
814 if not stripped.startswith(('[', '{')):
815 return value
816 try:
817 return json.loads(stripped)
818 except ValueError:
819 return value
822def _eval_condition(cond: Any, ctx: dict) -> bool:
823 """Evaluate a step condition. None → True (always run).
825 Strict (short-circuit + hard error on unresolved reads) per R1: an
826 undefined or skipped-step field read without short-circuit protection
827 raises, never silently evaluates to falsy.
828 """
829 if cond is None:
830 return True
831 resolved = _resolve_template(cond, ctx, strict=True)
832 if isinstance(resolved, bool):
833 return resolved
834 if isinstance(resolved, str):
835 return resolved.lower() not in ('false', '0', '', 'none', 'null')
836 return bool(resolved)
839def _resolve_int(value: Any, ctx: dict, default: int) -> int:
840 v = _resolve_template(value, ctx) if isinstance(value, str) else value
841 try:
842 return int(v)
843 except (TypeError, ValueError):
844 return default
847def _resolve_int_strict(
848 value: Any,
849 ctx: dict,
850 field: str,
851 step_id: str,
852 *,
853 minimum: int | None = None,
854) -> int:
855 """Resolve a *present* loop/iteration limit, never falling back silently.
857 ADR-0012: a present manifest value that cannot be resolved to a valid
858 integer in the build context is a design-level hard failure, not a reason to
859 substitute a literal default. ``bool`` is rejected explicitly — it is an
860 ``int`` subclass but never a legitimate iteration count. ``minimum`` (when
861 given) rejects a resolved value below the bound, e.g. a present
862 ``max_concurrent`` must be ``>= 1``.
863 """
864 v = _resolve_template(value, ctx) if isinstance(value, str) else value
865 if isinstance(v, bool):
866 raise RunHardFailure(
867 "VALIDATION_ERROR",
868 f"step '{step_id}' field '{field}' must resolve to an integer, "
869 f"got {v!r}",
870 {"step_id": step_id, "field": field},
871 )
872 try:
873 resolved = int(v)
874 except (TypeError, ValueError):
875 raise RunHardFailure(
876 "VALIDATION_ERROR",
877 f"step '{step_id}' field '{field}' must resolve to an integer at "
878 f"build time, got {v!r}",
879 {"step_id": step_id, "field": field},
880 ) from None
881 if minimum is not None and resolved < minimum:
882 raise RunHardFailure(
883 "VALIDATION_ERROR",
884 f"step '{step_id}' field '{field}' must be >= {minimum}, "
885 f"got {resolved}",
886 {"step_id": step_id, "field": field},
887 )
888 return resolved
891# ---------------------------------------------------------------------------
892# ManifestLoader
893# ---------------------------------------------------------------------------
896def _parse_manifest_file(path: Path) -> list[dict]:
897 """Palauttaa kaikki apiVersion: agent-platform/v1 -dokumentit tiedostosta."""
898 try:
899 content = path.read_text(encoding="utf-8")
900 except OSError as exc:
901 log.warning("Cannot read %s: %s", path, exc)
902 return []
904 results: list[dict] = []
905 for raw in content.split("\n---"):
906 raw_clean = "\n".join(
907 line for line in raw.splitlines()
908 if not line.strip().startswith("#")
909 )
910 try:
911 doc = yaml.safe_load(raw_clean)
912 except yaml.YAMLError as exc:
913 log.warning("YAML parse error in %s: %s", path, exc)
914 continue
915 if isinstance(doc, dict) and doc.get("apiVersion") == "agent-platform/v1":
916 results.append(doc)
917 return results
920class ManifestLoader:
921 """Skannaa hakemistot ja lataa kaikki YAML-manifestit."""
923 def __init__(self, dirs: list[Path] | None = None) -> None:
924 self._dirs = dirs or MANIFEST_DIRS
926 def load_all(self) -> list[dict]:
927 manifests: list[dict] = []
928 for d in self._dirs:
929 if not d.exists():
930 log.debug("Manifest dir not found: %s", d)
931 continue
932 for path in sorted(d.glob("**/*.yaml")):
933 docs = _parse_manifest_file(path)
934 manifests.extend(docs)
935 if docs:
936 log.debug("Loaded %d manifest(s) from %s", len(docs), path)
937 return manifests
940# ---------------------------------------------------------------------------
941# ManifestRegistry
942# ---------------------------------------------------------------------------
944ResourceKey = tuple[str, str] # (kind, name)
947class ManifestRegistry:
948 """Indeksoi ladatut manifestit kind+name avaimella."""
950 def __init__(self, manifests: list[dict]) -> None:
951 self._store: dict[ResourceKey, dict] = {}
952 for doc in manifests:
953 kind = doc.get("kind", "")
954 name = (doc.get("metadata") or {}).get("name", "")
955 if kind and name:
956 key: ResourceKey = (kind, name)
957 if key in self._store:
958 log.warning("Duplicate manifest %s/%s — last one wins", kind, name)
959 self._store[key] = doc
961 def get(self, kind: str, name: str) -> dict | None:
962 return self._store.get((kind, name))
964 def all_of_kind(self, kind: str) -> list[dict]:
965 return [v for (k, _), v in self._store.items() if k == kind]
968# ---------------------------------------------------------------------------
969# AgentResolver
970# ---------------------------------------------------------------------------
972def _deep_merge(base: dict, override: dict) -> dict:
973 """Syvä merge: override voittaa konflikteissa."""
974 result = dict(base)
975 for k, v in override.items():
976 if isinstance(v, dict) and isinstance(result.get(k), dict):
977 result[k] = _deep_merge(result[k], v)
978 else:
979 result[k] = v
980 return result
983def _tool_has_description(tool: Any) -> bool:
984 """Report whether a resolved tool exposes a non-empty model description."""
985 description = (tool.definition().get("function") or {}).get("description")
986 return isinstance(description, str) and bool(description)
989class AgentResolver:
990 """
991 Palauttaa fully-resolved agent spec:n instanssille.
992 Jos instanssilla on based_on, mergee Agent-base + AgentInstance-overridet.
993 """
995 def __init__(self, registry: ManifestRegistry) -> None:
996 self._registry = registry
997 self._cache: dict[str, dict] = {}
999 def resolve(self, instance_name: str) -> dict:
1000 if instance_name in self._cache:
1001 return self._cache[instance_name]
1003 instance = self._registry.get("AgentInstance", instance_name)
1004 if instance is None:
1005 agent = self._registry.get("Agent", instance_name)
1006 if agent is None:
1007 raise RunHardFailure(
1008 "VALIDATION_ERROR",
1009 f"No Agent or AgentInstance named '{instance_name}'",
1010 {"agent": instance_name},
1011 )
1012 resolved = agent
1013 else:
1014 based_on = (instance.get("spec") or {}).get("based_on")
1015 if based_on:
1016 base_agent = self._registry.get("Agent", based_on)
1017 if base_agent is None:
1018 raise RunHardFailure(
1019 "VALIDATION_ERROR",
1020 f"AgentInstance '{instance_name}' references unknown "
1021 f"Agent '{based_on}'",
1022 {"agent": instance_name, "based_on": based_on},
1023 )
1024 base_spec = base_agent.get("spec") or {}
1025 inst_spec = {k: v for k, v in (instance.get("spec") or {}).items()
1026 if k != "based_on"}
1027 merged_spec = _deep_merge(base_spec, inst_spec)
1028 else:
1029 merged_spec = {k: v for k, v in (instance.get("spec") or {}).items()
1030 if k != "based_on"}
1032 resolved = {
1033 "apiVersion": "agent-platform/v1",
1034 "kind": "Agent",
1035 "metadata": {"name": instance_name},
1036 "spec": merged_spec,
1037 }
1039 self._cache[instance_name] = resolved
1040 return resolved
1043# ---------------------------------------------------------------------------
1044# State type
1045# ---------------------------------------------------------------------------
1047PipelineState = dict[str, Any]
1049# Engine-reserved channel key: the existing flat Core-managed JSON state lives
1050# under this key when a pipeline declares spec.channels.
1051_CORE = CORE_KEY
1053# Core-owned per-step map-reduce accumulator. A concurrent top-level `parallel`
1054# step fans out through native `Send`; each worker writes `{index: item_result}`
1055# under its step id and the join node consumes it. The accumulator is a
1056# top-level state channel reduced by `accumulate_map_reduce`.
1057_MAP_REDUCE_ACCUMULATOR_KEY = "__map_reduce_accum__"
1059# Reserved key carrying a worker's source item index through a `Send` payload.
1060_MAP_REDUCE_ITEM_INDEX = "__map_reduce_item_index__"
1062# Flat-state keys the engine owns; channel names may not shadow them.
1063_CORE_STATE_KEYS = frozenset({
1064 "errors", "stats", "run_status", "project_root", "run_id",
1065 "_stats", "step_context_metrics", "step_tool_meta",
1066 _MAP_REDUCE_ACCUMULATOR_KEY,
1067})
1069# Engine-local iteration keys the loop machinery injects into the read state.
1070# They are engine plumbing, not user state, so a colliding channel key must
1071# never shadow them.
1072_ENGINE_LOCAL_ITERATION_KEYS = ("item", "_item", "_foreach_item")
1075def _iteration_discriminator(state: dict) -> str | None:
1076 """The engine-local iteration token for *state*, or None when not iterating.
1078 Concurrent foreach/parallel items share a step id but differ in their
1079 iteration key, so the synthesized call id uses this to stay distinct on the
1080 in-cluster sidecar params file. The item is digested so the id stays a
1081 bounded, filename-safe length.
1082 """
1083 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1084 if key in state:
1085 digest = hashlib.sha1(repr(state[key]).encode("utf-8")).hexdigest()[:12]
1086 return f"{key}-{digest}"
1087 return None
1091def _merge_state(left: dict, right: dict) -> dict:
1092 if not left:
1093 return dict(right)
1094 side_channel_keys = ("step_context_metrics", "step_tool_meta")
1095 merged = dict(left)
1096 for key, value in right.items():
1097 existing = merged.get(key)
1098 if (
1099 key in side_channel_keys
1100 and isinstance(existing, dict)
1101 and isinstance(value, dict)
1102 ):
1103 merged[key] = _merge_side_channel(existing, value)
1104 else:
1105 merged[key] = value
1106 return merged
1109def _build_state_schema() -> Any:
1110 """Build the channel-path TypedDict: the Core namespace plus the accumulator.
1112 Channel context is Core-owned, so there are no per-channel native LangGraph
1113 fields; the flat committed snapshot (core keys + channel keys) travels
1114 node-to-node under ``__core__``. The map-reduce accumulator is a separate
1115 top-level channel so native ``Send`` workers reduce into it; LangGraph stays
1116 the graph executor.
1117 """
1118 from typing import TypedDict
1120 return TypedDict(
1121 "PipelineState",
1122 {
1123 _CORE: Annotated[dict[str, Any], _merge_state],
1124 _MAP_REDUCE_ACCUMULATOR_KEY: Annotated[
1125 dict[str, Any], accumulate_map_reduce
1126 ],
1127 },
1128 )
1131# Sentinel for optional kwargs distinguishing "not provided" from None
1132_UNSET: Any = object()
1134# ---------------------------------------------------------------------------
1135# BuiltinExecutor
1136# ---------------------------------------------------------------------------
1138class BuiltinExecutor:
1139 """
1140 Ajaa builtin-komponentit.
1141 Uusia komponentteja voi rekisteröidä register()-metodilla.
1142 Nimet normalisoidaan: viiva → alaviiva (esim. 'list-dir-entries' == 'list_dir_entries').
1143 """
1145 # Lazy modules are imported on first use (they need httpx / heavier deps).
1146 # Profile-specific builtins live here: the PR-profile github-pr-comment
1147 # and the TICKET-profile ticket-publish (per-run local feature branch; push/PR disabled).
1148 _LAZY_MODULES: tuple[str, ...] = ("_builtins_pr", "_builtins_ticket")
1150 def __init__(self) -> None:
1151 self._handlers: dict[str, Any] = {
1152 "passthrough": self._passthrough,
1153 # Engine-internal call-expansion scaffolding (ADR-0037 D4): these are
1154 # synthesized by the build-time call composer, never authored in a
1155 # manifest. A call step becomes a compiled subgraph; these two op
1156 # steps run inside it, binding params and emitting the call result.
1157 "__call_bind": self._call_bind,
1158 "__call_emit": self._call_emit,
1159 }
1160 # Pipeline's LLM invoker (spec, step, state) → state, bound by
1161 # GraphBuilder. Profile builtins that make LLM decisions
1162 # (ticket-dor / ticket-review) read it via their register_all
1163 # wrapper — the executor must have it set before the builtin
1164 # executes (binding happens at build_workflow time).
1165 self.invoke_agent_fn: Any | None = None
1166 self._lazy_loaded = False
1168 def register(self, name: str, fn: Any) -> None:
1169 self._handlers[name] = fn
1170 # Also register with normalised name (hyphen → underscore)
1171 normalised = name.replace("-", "_")
1172 if normalised != name:
1173 self._handlers[normalised] = fn
1175 def _load_lazy_modules(self) -> None:
1176 if self._lazy_loaded:
1177 return
1178 self._lazy_loaded = True
1179 import importlib
1180 for module_name in self._LAZY_MODULES:
1181 try:
1182 mod = importlib.import_module(f".{module_name}", package="langgraph")
1183 except ImportError:
1184 continue
1185 mod.register_all(self)
1187 def has(self, name: str) -> bool:
1188 """True when *name* resolves in the builtin register (lazy modules included)."""
1189 self._load_lazy_modules()
1190 return name in self._handlers or name.replace("-", "_") in self._handlers
1192 def get(self, name: str) -> Any:
1193 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
1194 if fn is not None:
1195 return fn
1196 self._load_lazy_modules()
1197 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
1198 if fn is None:
1199 raise RunHardFailure(
1200 "VALIDATION_ERROR",
1201 f"Unknown builtin component '{name}'",
1202 {"component": name},
1203 )
1204 return fn
1206 def execute(self, component: str, step: dict, state: PipelineState) -> PipelineState:
1207 fn = self.get(component)
1208 try:
1209 return fn(step, state)
1210 except RunHardFailure as exc:
1211 log.error("[hard-failure] %s: %s", exc.code, exc.message)
1212 raise
1214 # --- default builtins ---
1216 @staticmethod
1217 def _passthrough(step: dict, state: PipelineState) -> PipelineState:
1218 return state
1220 @staticmethod
1221 def _call_bind(step: dict, state: PipelineState) -> PipelineState:
1222 """Call-scope entry (synthesized): bind the function call's params.
1224 The caller's function params are resolved against the current state and
1225 stored at ``state[<call_id>] = {"params": {...}}`` so the (rewritten)
1226 function-body templates ``{{ <call_id>.params.<name> }}`` resolve. The
1227 call's reserved ``__call_id`` key is stripped from the bound params.
1228 """
1229 call_id = (step.get("params") or {}).get("__call_id")
1230 if not call_id:
1231 raise ValueError("__call_bind missing __call_id (not a synthesized step)")
1232 bound = {
1233 k: v for k, v in (step.get("params") or {}).items() if k != "__call_id"
1234 }
1235 return {**state, call_id: {"params": bound}}
1237 @staticmethod
1238 def _call_emit(step: dict, state: PipelineState) -> PipelineState:
1239 """Call-scope exit (synthesized): surface the function's outputs.
1241 Per the call step-result contract (00-process-contract r.70) the caller
1242 sees only ``state[<call_id>] = {outputs: {<name>: <source value>}}`` —
1243 the function body step results are invisible to the caller. Each output
1244 name resolves to the manifest interface ``outputs[].source`` dot-path.
1246 Optionality (ADR-0031): an output flagged ``optional: true`` is skipped
1247 when its source-path is unresolved; an output that is NOT optional must
1248 resolve or a RunHardFailure is raised (no silent None — ADR-0012).
1249 """
1250 params = step.get("params") or {}
1251 call_id = params.get("__call_id")
1252 if not call_id:
1253 raise ValueError("__call_emit missing __call_id (not a synthesized step)")
1254 outputs_spec = params.get("__outputs") or []
1255 outputs: dict[str, Any] = {}
1256 for spec in outputs_spec:
1257 if not isinstance(spec, dict):
1258 continue
1259 name = spec.get("name")
1260 source = spec.get("source")
1261 if name is None:
1262 continue
1263 optional = bool(spec.get("optional", False))
1264 if source is None:
1265 # No declared source-path: treat as unresolved against the
1266 # optionality gate (ADR-0012 — no silent None for required).
1267 if optional:
1268 continue
1269 raise RunHardFailure(
1270 "UNRESOLVED_OUTPUT",
1271 f"call '{call_id}' output '{name}' has no source "
1272 "(required output — no silent None, ADR-0012)",
1273 {"call_id": call_id, "name": name, "source": source},
1274 )
1275 try:
1276 value = _dot_get_strict(state, source)
1277 except ValueError:
1278 # Unresolved source-path: required → hard error; optional → skip.
1279 if optional:
1280 continue
1281 raise RunHardFailure(
1282 "UNRESOLVED_OUTPUT",
1283 f"call '{call_id}' output '{name}' unresolved: source "
1284 f"'{source}' not found in state (required output — "
1285 "no silent None, ADR-0012)",
1286 {"call_id": call_id, "name": name, "source": source},
1287 ) from None
1288 outputs[name] = value
1289 return {**state, call_id: {"outputs": outputs}}
1292_DEFAULT_BUILTINS = BuiltinExecutor()
1295def _register_domain_builtins(executor: BuiltinExecutor) -> None:
1296 """Rekisteröi v1-domain-builtinit (list-dir-entries, ticket-publisher, ...).
1298 Erotettu omaan moduliin (`_builtins.py`) — runner.py pysyy generic-engine,
1299 builtinit ovat generiisiä mekanismeja (ADR-0016/0017).
1300 """
1301 try:
1302 from langgraph import _builtins # type: ignore
1303 except ImportError:
1304 # Kun runner.py ajetaan ilman pakettirakennetta (esim. /app/langgraph)
1305 import importlib.util
1306 spec = importlib.util.spec_from_file_location(
1307 "_builtins", Path(__file__).parent / "_builtins.py"
1308 )
1309 if spec and spec.loader:
1310 _builtins = importlib.util.module_from_spec(spec)
1311 spec.loader.exec_module(_builtins) # type: ignore[union-attr]
1312 else:
1313 log.warning("Cannot load _builtins.py — domain builtins disabled")
1314 return
1315 _builtins.register_all(executor)
1316 _register_function_builtins(executor)
1319def _register_function_builtins(executor: BuiltinExecutor) -> None:
1320 """Rekisteröi function-engine builtinit (question-block-form).
1322 Erotettu omaan moduliin (`_builtins_functions.py`) — runner.py pysyy
1323 generic-engine, function-builtinit pysyvät puhtaina ja erillisinä.
1324 """
1325 try:
1326 from langgraph import _builtins_functions # type: ignore
1327 except ImportError:
1328 import importlib.util
1329 spec = importlib.util.spec_from_file_location(
1330 "_builtins_functions", Path(__file__).parent / "_builtins_functions.py"
1331 )
1332 if spec and spec.loader:
1333 _builtins_functions = importlib.util.module_from_spec(spec)
1334 spec.loader.exec_module(_builtins_functions) # type: ignore[union-attr]
1335 else:
1336 log.warning("Cannot load _builtins_functions.py — function builtins disabled")
1337 return
1338 _builtins_functions.register_all(executor)
1341_register_domain_builtins(_DEFAULT_BUILTINS)
1343# ---------------------------------------------------------------------------
1344# StepRunner
1345# ---------------------------------------------------------------------------
1347class StepRunner:
1348 """
1349 Ajaa yksittäisen step:n sen typin mukaan.
1351 run_context: dict joka lisätään template-kontekstiin. Esim:
1352 {"budget": {"max_parallel": 16, ...}, "config": {...}}
1353 """
1355 def __init__(
1356 self,
1357 resolver: AgentResolver,
1358 builtins: BuiltinExecutor | None = None,
1359 invoke_agent_fn: Any | None = None,
1360 run_context: dict | None = None,
1361 *,
1362 step_functions: StepFunctionExecutor | None = None,
1363 tracer: Any | None = None,
1364 tool_resolver: Any | None = None,
1365 ) -> None:
1366 self._resolver = resolver
1367 self._builtins = builtins or _DEFAULT_BUILTINS
1368 self._invoke_agent = self._wrap_hard_failure_gate(
1369 invoke_agent_fn or self._dummy_invoke
1370 )
1371 self._run_context: dict = run_context or {}
1372 self._step_functions = step_functions
1373 self._tracer = tracer
1374 self._tool_resolver = tool_resolver
1375 self._channel_names: frozenset[str] = frozenset()
1376 self._template_spec: dict | None = None
1378 def set_channel_routing(self, channel_names: frozenset[str]) -> None:
1379 """Attach the build-time routing context nested body steps route with."""
1380 self._channel_names = channel_names
1382 def set_template_spec(self, spec: dict) -> None:
1383 """Attach the pipeline's own ``spec`` as the ``spec.`` template root."""
1384 self._template_spec = spec
1386 @staticmethod
1387 def _should_skip_llm(step: dict, ctx: dict) -> bool:
1388 """Truthiness of a step's `skip_llm_if` expression (ENG-12, Q4).
1390 Lenient (non-strict) evaluation: a missing/undefined field resolves to
1391 falsy, so the step behaves normally (LLM invoked). Only an explicit
1392 truthy expression skips the LLM.
1393 """
1394 raw = step.get("skip_llm_if")
1395 if raw is None:
1396 return False
1397 resolved = _resolve_template(raw, ctx)
1398 if isinstance(resolved, bool):
1399 return resolved
1400 if isinstance(resolved, str):
1401 return resolved.lower() not in ('false', '0', '', 'none', 'null')
1402 return bool(resolved)
1404 @staticmethod
1405 def _skip_llm_passthrough(
1406 step: dict,
1407 state: PipelineState,
1408 ctx: dict,
1409 read_state: PipelineState,
1410 ) -> PipelineState:
1411 """Deterministic passthrough of a step's resolved input (no LLM).
1413 ``state[step_id]`` becomes the resolved ``input`` value itself (the
1414 parallel fan-out envelope ``{results: [...]}`` when the input resolves
1415 to a list), so channel routing can project the same payload a builtin
1416 would. In the channel path the persisted copy is stripped to envelope
1417 metadata after routing (ADR-0023 D9 / ADR-0043 D6). ``read_state`` is
1418 the working-set overlay the input resolves against; the returned state
1419 is still built from the raw ``state``.
1420 """
1421 step_id = step.get("id", "?")
1422 input_spec = step.get("input")
1423 resolved = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
1424 if isinstance(resolved, list):
1425 log.debug("[step] %s skip_llm_if passthrough (fanout, no LLM)", step_id)
1426 return {**state, step_id: {"results": resolved}}
1427 log.debug("[step] %s skip_llm_if passthrough (no LLM)", step_id)
1428 return {**state, step_id: resolved}
1430 @staticmethod
1431 def _declared_item_source(step: dict) -> tuple[str, Any] | None:
1432 """A fan-out step's declared item source as ``(kind, spec)``, else None.
1434 ``iteration`` is the only documented fan-out source (a list/object
1435 literal, or a ``{{ }}`` variable). A step declaring none has no item
1436 source (the documented parallel single-call fallback); a ``foreach``
1437 declaring none is a hard error.
1439 ``iteration`` is detected by key presence (not truthiness) so a declared
1440 ``iteration: null`` is reported as an invalid source, never silently
1441 treated as no source.
1442 """
1443 if "iteration" in step:
1444 return ("iteration", step.get("iteration"))
1445 return None
1447 @staticmethod
1448 def _resolve_iteration_source(spec: Any, ctx: dict) -> Any:
1449 """Resolve a declared ``iteration`` source to its raw value.
1451 Documented forms: a JSON list/object literal, or a ``{{ }}`` variable
1452 resolving to a list. A non-template string is a JSON literal candidate
1453 and otherwise a scalar, returned verbatim for the caller's single-call
1454 fallback.
1455 """
1456 if isinstance(spec, str):
1457 if '{{' in spec:
1458 return _resolve_template(spec, ctx)
1459 return _parse_json_literal(spec)
1460 return spec
1462 @staticmethod
1463 def _iteration_items(resolved: Any) -> list | None:
1464 """A resolved ``iteration`` value's fan-out items, or None for a non-list.
1466 Only a JSON list fans out (0..N items). A JSON object or any other
1467 scalar is not a list: the caller applies the documented non-list
1468 fallback (a ``parallel`` single call; a ``foreach`` one-item list).
1469 A null value is rejected by the caller before this is reached.
1470 """
1471 if isinstance(resolved, list):
1472 return resolved
1473 return None
1475 @staticmethod
1476 def _resolve_item_source(source: tuple[str, Any], ctx: dict) -> Any:
1477 """Resolve a declared ``iteration`` item source to its raw value."""
1478 return StepRunner._resolve_iteration_source(source[1], ctx)
1480 @staticmethod
1481 def _unresolvable_item_source_failure(
1482 step_id: str, source: tuple[str, Any]
1483 ) -> "RunHardFailure":
1484 kind, spec = source
1485 return RunHardFailure(
1486 "VALIDATION_ERROR",
1487 f"step '{step_id}' item source '{kind}: {spec}' is unresolvable "
1488 "(no silent fallback — ADR-0012)",
1489 {"step_id": step_id, "source_kind": kind, "source": spec},
1490 )
1492 @staticmethod
1493 def _parallel_fanout_items(
1494 step: dict, ctx: dict
1495 ) -> list | None:
1496 """The resolved fan-out item list for a parallel step, else None.
1498 A parallel step whose resolved item source is a list invokes the agent
1499 once per item and stores a ``{"results": [...]}`` envelope instead of a
1500 single agent output; an empty list performs 0 calls (a valid fan-out of
1501 N=0). A parallel step declaring no item source keeps the single-agent
1502 fallback; any other declared source resolving to a non-list (scalar or
1503 JSON object) is also the documented single-call fallback (None). A
1504 declared source that is unresolvable is a ``VALIDATION_ERROR``
1505 (ADR-0012).
1506 """
1507 if step.get("type") != "parallel":
1508 return None
1509 source = StepRunner._declared_item_source(step)
1510 if source is None:
1511 return None
1512 step_id = step.get("id", "?")
1513 resolved = StepRunner._resolve_item_source(source, ctx)
1514 if resolved is None:
1515 raise StepRunner._unresolvable_item_source_failure(step_id, source)
1516 return StepRunner._iteration_items(resolved)
1518 @staticmethod
1519 def _parallel_fanout_item_list(
1520 step: dict, ctx: dict
1521 ) -> list | None:
1522 """The fan-out item list with the ``max_items`` cap applied."""
1523 items = StepRunner._parallel_fanout_items(step, ctx)
1524 if items is None:
1525 return None
1526 max_items_raw = step.get("max_items")
1527 if max_items_raw is not None:
1528 items = items[: _resolve_int(max_items_raw, ctx, len(items))]
1529 return items
1531 @staticmethod
1532 def _foreach_item_list(
1533 step: dict, ctx: dict
1534 ) -> list:
1535 """The foreach item list: resolved, wrapped, capped.
1537 A missing source and an unresolvable source are hard
1538 ``VALIDATION_ERROR`` failures (ADR-0012). A declared ``iteration``
1539 source that resolves to a list iterates over it (an empty list runs 0
1540 iterations, a valid N=0 fan-out); a non-list (scalar or JSON object)
1541 runs once with that value as the item. ``max_iterations`` caps the
1542 list.
1543 """
1544 step_id = step.get("id", "?")
1545 source = StepRunner._declared_item_source(step)
1546 if source is None:
1547 raise RunHardFailure(
1548 "VALIDATION_ERROR",
1549 f"foreach step '{step_id}' declares no item source "
1550 "('iteration') — no silent [None] fallback (ADR-0012)",
1551 {"step_id": step_id},
1552 )
1553 resolved = StepRunner._resolve_item_source(source, ctx)
1554 if resolved is None:
1555 raise StepRunner._unresolvable_item_source_failure(step_id, source)
1556 items = StepRunner._iteration_items(resolved)
1557 if items is None:
1558 items = [resolved]
1559 max_iter_raw = step.get("max_iterations")
1560 if max_iter_raw is not None:
1561 max_iter = _resolve_int_strict(
1562 max_iter_raw, ctx, "max_iterations", step_id
1563 )
1564 items = items[:max_iter]
1565 return items
1567 def ran_agent(self, step: dict, state: PipelineState) -> bool:
1568 """Whether a step's run invokes its agent.
1570 False for agent-less steps, a ``skip_llm_if`` passthrough, and a
1571 parallel fan-out (per-item agent calls storing a result envelope).
1572 The router uses a true result to select the agent ``content`` route; a
1573 false result means the agent route is skipped (the passthrough is then
1574 routed through the builtin payload route, and a fan-out envelope is
1575 staged per item elsewhere).
1576 """
1577 if self.resolve_agent(step) is None:
1578 return False
1579 ctx = self._make_ctx(state)
1580 if self._should_skip_llm(step, ctx):
1581 return False
1582 return self._parallel_fanout_items(step, ctx) is None
1584 def _make_ctx(self, state: dict, **extras: Any) -> dict:
1585 """Rakentaa template-kontekstin: state + run_context + extras.
1587 Channel-named keys are not spread into the top level: channel data is
1588 reachable only through the ``channel`` root
1589 (``{{ channel.<name>[.<field>] }}``), so the bare
1590 ``{{ lastvalue.<field> }}`` form cannot resolve.
1591 """
1592 ctx: dict = {
1593 key: value
1594 for key, value in state.items()
1595 if key not in self._channel_names
1596 }
1597 ctx.update(self._run_context)
1598 ctx.update(extras)
1599 if self._template_spec is not None:
1600 ctx["spec"] = self._template_spec
1601 ctx["channel"] = {
1602 name: state[name] for name in self._channel_names if name in state
1603 }
1604 return ctx
1606 @staticmethod
1607 def _read_state(
1608 state: PipelineState, working_set: ChannelWorkingSet
1609 ) -> PipelineState:
1610 """Overlay the invocation working set's read view onto the local state.
1612 The working set carries the committed snapshot plus this invocation's
1613 pending writes; the local state contributes keys the snapshot does not
1614 hold (item context, body step results). The engine-local iteration keys
1615 are re-asserted after the overlay so a declared channel with the same
1616 name cannot shadow them.
1617 """
1618 resolved = {**state, **working_set.read_view()}
1619 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1620 if key in state:
1621 resolved[key] = state[key]
1622 return resolved
1624 def _run_body_step(
1625 self,
1626 step: dict,
1627 state: PipelineState,
1628 working_set: ChannelWorkingSet,
1629 ) -> PipelineState:
1630 """Run one nested body step and stage its declared channel writes."""
1631 result = self.run_step(step, state, working_set)
1632 self._stage_body_channel_writes(step, state, result, working_set)
1633 _project_persisted_step_result(result, step)
1634 return result
1636 def _stage_body_channel_writes(
1637 self,
1638 step: dict,
1639 pre_state: PipelineState,
1640 result_state: PipelineState,
1641 working_set: ChannelWorkingSet,
1642 ) -> None:
1643 """Stage a body step's declared channel writes into the working set.
1645 The router selects the routed value by step kind: an agent step routes
1646 its envelope's ``content``, an agent-less step (builtin, call bind/emit)
1647 routes its result payload itself. This path runs inside a superstep
1648 (concurrent foreach / map-reduce item), so the implicit default channel
1649 is forbidden: ``allow_default`` stays unset.
1650 """
1651 if not self._channel_names:
1652 return
1653 writes = _route_declared_channels(
1654 step,
1655 self._read_state(pre_state, working_set),
1656 result_state,
1657 self,
1658 self._channel_names,
1659 )
1660 for name, value in writes.items():
1661 working_set.stage(name, value)
1663 def _resolve_step_runtime(
1664 self,
1665 step: dict,
1666 state: dict,
1667 ctx: dict,
1668 *,
1669 input_override: Any = _UNSET,
1670 extra_params: dict | None = None,
1671 ) -> dict:
1672 """
1673 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
1674 templaattien osalta state + run_context -kontekstia vasten.
1676 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
1677 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
1678 """
1679 resolved: dict = dict(step)
1681 if input_override is not _UNSET:
1682 resolved["input"] = input_override
1683 else:
1684 input_spec = step.get("input")
1685 if input_spec is not None:
1686 resolved["input"] = _resolve_input(input_spec, state, ctx)
1688 params_spec = step.get("params")
1689 if params_spec or extra_params:
1690 base = _resolve_template(params_spec, ctx) if params_spec else {}
1691 if not isinstance(base, dict):
1692 base = {}
1693 if extra_params:
1694 base = {**base, **extra_params}
1695 resolved["params"] = base
1697 return resolved
1699 def _attach_communication_context(
1700 self, resolved_step: dict, read_state: PipelineState
1701 ) -> dict:
1702 """Attach a step's projected COMMUNICATION context when declared.
1704 The projection is resolved from the invocation read overlay (the
1705 committed snapshot plus this invocation's pending writes) and carries
1706 only the step's declared input channels — never whole state. A step
1707 without a read declaration is returned unchanged.
1708 """
1709 communication = resolve_communication_context(resolved_step, read_state)
1710 if not communication:
1711 return resolved_step
1712 return {**resolved_step, COMMUNICATION_CONTEXT_KEY: communication}
1714 def _attach_resolved_tools(
1715 self, resolved_step: dict, resolved_agent: dict
1716 ) -> dict:
1717 """Attach the agent's declared `tools` as resolved `Tool` instances.
1719 Resolves the agent's declared tool names through `ToolResolver` into
1720 ordered `Tool` objects under `RESOLVED_TOOLS_KEY`. An agent without
1721 tools, or a runner without a resolver, is returned unchanged.
1722 """
1723 tools = (resolved_agent.get("spec") or {}).get("tools") or []
1724 if not tools or self._tool_resolver is None:
1725 return resolved_step
1726 return {
1727 **resolved_step,
1728 RESOLVED_TOOLS_KEY: self._resolve_agent_tools(
1729 resolved_agent, list(tools)
1730 ),
1731 }
1733 def _resolve_agent_tools(
1734 self, resolved_agent: dict, tools: list[str]
1735 ) -> list:
1736 """Resolve an agent's declared tools, enforcing the instruction guard.
1738 The low-level resolver stays permissive; the policy lives here
1739 (ADR-0016). A declared name the resolver cannot resolve, or a
1740 resolved tool that exposes no non-empty description, is translated
1741 into the engine's build-time `AGENT_INSTRUCTION_MISSING` hard failure.
1742 A dual-source name keeps its own `BUILTIN_NAME_COLLISION` failure.
1743 """
1744 return [self._resolve_agent_tool(resolved_agent, name) for name in tools]
1746 def _resolve_agent_tool(self, resolved_agent: dict, name: str):
1747 try:
1748 tool = self._tool_resolver.resolve([name])[0]
1749 except ToolCollisionError as exc:
1750 raise RunHardFailure(
1751 "BUILTIN_NAME_COLLISION",
1752 f"tool {name!r} exists in both the platform registry and the "
1753 "StepFunction register",
1754 {"tool": name},
1755 ) from exc
1756 except ToolRegistryError as exc:
1757 raise self._instruction_missing_failure(
1758 resolved_agent, name, str(exc)
1759 ) from exc
1760 if not _tool_has_description(tool):
1761 raise self._instruction_missing_failure(
1762 resolved_agent, name, "the resolved tool exposes no description"
1763 )
1764 return tool
1766 @staticmethod
1767 def _instruction_missing_failure(
1768 resolved_agent: dict, tool_name: str, detail: str
1769 ) -> RunHardFailure:
1770 agent_name = (resolved_agent.get("metadata") or {}).get("name")
1771 return RunHardFailure(
1772 AGENT_INSTRUCTION_MISSING,
1773 f"Agent {agent_name!r} declares tool {tool_name!r} with no usage "
1774 f"instruction: {detail}",
1775 {"agent": agent_name, "tool": tool_name},
1776 )
1778 @staticmethod
1779 def _dummy_invoke(
1780 resolved_spec: dict, step: dict, state: PipelineState
1781 ) -> PipelineState:
1782 """Placeholder — ei kutsu oikeaa LLM:ää."""
1783 step_id = step.get("id", "?")
1784 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
1785 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
1787 @staticmethod
1788 def _merge_errors(base: list, additions: list) -> list:
1789 """Return a new list containing all items from *base* plus any items from
1790 *additions* that are not already present. Order is preserved and no
1791 in-place mutation occurs — safe to call from any context."""
1792 if not additions:
1793 return base
1794 merged = list(base)
1795 for err in additions:
1796 if err not in merged:
1797 merged.append(err)
1798 return merged
1800 @staticmethod
1801 def _wrap_hard_failure_gate(invoke_fn: Any) -> Any:
1802 """Kietoo invokerin: jokaiseen agent-step-tulokseen sovelletaan
1803 kova-virhe-portti (RunHardFailure details.error_code-perusteella).
1804 Suppein funnel joka näkee KAIKKI agent-tulokset: sequential,
1805 parallel (single + fanout), foreach (body + agent-polku) ja while."""
1806 def invoke_with_gate(
1807 resolved_spec: dict, step: dict, state: PipelineState
1808 ) -> PipelineState:
1809 result = invoke_fn(resolved_spec, step, state)
1810 step_id = step.get("id", "?")
1811 StepRunner._raise_on_hard_error(result.get(step_id), step_id)
1812 return result
1813 return invoke_with_gate
1815 @staticmethod
1816 def _is_hard_fail(details: dict) -> bool:
1817 """Kova virhe vain kun: error_code on HARD_ERROR_CODESissa
1818 (TRUNCATION_ERROR, TOOL_FAILURE, AGENT_ERROR) tai VALIDATION_ERROR
1819 kantaa kontekstiylityksen signatuurin (assembled_tokens +
1820 context_budget — _llm._context_overflow_result).
1821 Generic-VALIDATION_ERROR ei kaada ajoa. status:error-envelope kaatuu
1822 _raise_on_error_envelope-polulla ennen tätä tarkistusta."""
1823 code = details.get("error_code")
1824 if code in RunHardFailure.HARD_ERROR_CODES:
1825 return True
1826 if code == "VALIDATION_ERROR":
1827 return {"assembled_tokens", "context_budget"} <= details.keys()
1828 return False
1830 @staticmethod
1831 def _raise_on_hard_error(step_result: Any, step_id: str) -> None:
1832 """Nosta RunHardFailure jos step-tulos kantaa kovan virheen
1833 (_is_hard_fail) tai on status:error-envelope; stepin details
1834 kulkeutuu poikkeuksen mukana."""
1835 if not isinstance(step_result, dict):
1836 return
1837 if is_error_envelope(step_result):
1838 StepRunner._raise_on_error_envelope(step_result, step_id)
1839 return
1840 details = step_result.get("details")
1841 if not isinstance(details, dict):
1842 return
1843 if not StepRunner._is_hard_fail(details):
1844 return
1845 code = details.get("error_code")
1846 err = step_result.get("error", "")
1847 message = err if isinstance(err, str) and err else f"{code}: step '{step_id}' hard failure"
1848 log.error("[hard-failure] %s: %s", code, message)
1849 raise RunHardFailure(code, message, details)
1851 @staticmethod
1852 def _raise_on_error_envelope(step_result: dict, step_id: str) -> None:
1853 """Nosta RunHardFailure status:error-envelopesta; envelopen error
1854 (dict) kulkeutuu details-attribuuttina, muuten minimi details."""
1855 envelope_error = step_result.get("error")
1856 if isinstance(envelope_error, dict):
1857 details = envelope_error
1858 message = envelope_error.get("message")
1859 else:
1860 details = {"error_code": ENVELOPE_ERROR_CODE, "step_id": step_id}
1861 message = None
1862 if not isinstance(message, str) or not message:
1863 message = f"{ENVELOPE_ERROR_CODE}: step '{step_id}' envelope error"
1864 raise RunHardFailure(ENVELOPE_ERROR_CODE, message, details)
1866 @staticmethod
1867 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
1868 """
1869 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
1870 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
1871 epäonnistuneeksi eikä hiljennä virhettä.
1873 Kaikki virhetyypit propagoidaan:
1874 - budget:max_llm_calls / budget:run_timeout → budjettiloki
1875 - parse_error:*, timeout:*, ja muut → error-loki
1876 """
1877 step_result = result.get(step_id)
1878 if not isinstance(step_result, dict):
1879 return result
1880 err = step_result.get("error", "")
1881 if not isinstance(err, str) or not err:
1882 return result
1883 errors = list(result.get("errors") or [])
1884 message = f"{step_id}: {err}"
1885 if message not in errors:
1886 errors.append(message)
1887 if err.startswith("budget:"):
1888 log.warning("[budget] %s", message)
1889 else:
1890 log.error("[step-error] %s", message)
1891 return {**result, "errors": errors}
1893 def resolve_agent(self, step: dict) -> dict | None:
1894 """Resolve the agent/instance a step runs, or None for agent-less steps."""
1895 agent_name = step.get("agent")
1896 if not agent_name:
1897 return None
1898 return self._resolver.resolve(agent_name)
1900 def run_step(
1901 self,
1902 step: dict,
1903 state: PipelineState,
1904 working_set: ChannelWorkingSet,
1905 ) -> PipelineState:
1906 step_id = step.get("id", "?")
1907 read_state = self._read_state(state, working_set)
1908 ctx = self._make_ctx(read_state)
1909 step_type = step.get("type", "sequential")
1911 # Evaluate condition — if false, skip step. A `while` never reaches
1912 # this point as a raw step: build-time composition rewrites it into a
1913 # native cycle placeholder whose head owns the entry-condition skip.
1914 cond = step.get("condition")
1915 if cond is not None and not _eval_condition(cond, ctx):
1916 log.debug("[step] %s skipped (condition false)", step_id)
1917 return state
1919 if step_type == "builtin":
1920 component = step.get("component", "passthrough")
1921 log.debug("[step] %s builtin:%s", step_id, component)
1922 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
1923 # template-arvot (esim. "{{ budget.max_parallel }}")
1924 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1925 # ADR-0037 D2: resolve the component against the platform builtin
1926 # register FIRST, then the kind: StepFunction registry. A name in
1927 # both is a hard error (fail-fast).
1928 is_builtin = self._builtins.has(component)
1929 is_step_function = (
1930 self._step_functions is not None
1931 and self._step_functions.has(component)
1932 )
1933 if is_builtin and is_step_function:
1934 raise RunHardFailure(
1935 "BUILTIN_NAME_COLLISION",
1936 f"component '{component}' is present in both the builtin "
1937 "register and the StepFunction registry (ADR-0037 — "
1938 "fail-fast)",
1939 {"component": component},
1940 )
1941 if is_step_function:
1942 return self._step_functions.execute(component, resolved_step, state)
1943 builtin_state = self._builtins.execute(component, resolved_step, state)
1944 self._raise_on_hard_error(builtin_state.get(step_id), step_id)
1945 return builtin_state
1947 # skip_llm_if (ENG-12, Q4): when the step's skip_llm_if evaluates truthy,
1948 # do NOT invoke the LLM (no inference budget consumed) and set the step
1949 # result to a deterministic passthrough of the step's resolved input.
1950 if (
1951 step_type in ("sequential", "parallel")
1952 and self._should_skip_llm(step, ctx)
1953 ):
1954 return self._skip_llm_passthrough(step, state, ctx, read_state)
1956 if step_type == "parallel":
1957 # Fan-out: if iteration resolves to a list, call agent once per item
1958 items = self._parallel_fanout_items(step, ctx)
1959 if items is not None:
1960 return self._run_parallel_fanout(
1961 step, state, items, ctx, working_set
1962 )
1963 # Fallback: single agent call
1964 agent_name = step.get("agent")
1965 if not agent_name:
1966 log.warning("[step] %s has no agent — skipping", step_id)
1967 return state
1968 resolved = self._resolver.resolve(agent_name)
1969 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
1970 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1971 resolved_step = self._attach_communication_context(
1972 resolved_step, read_state
1973 )
1974 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
1975 result = self._invoke_agent(resolved, resolved_step, state)
1976 return self._propagate_step_errors(result, step_id)
1978 if step_type == "sequential":
1979 agent_name = step.get("agent")
1980 if not agent_name:
1981 log.warning("[step] %s has no agent — skipping", step_id)
1982 return state
1983 resolved = self._resolver.resolve(agent_name)
1984 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
1985 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1986 resolved_step = self._attach_communication_context(
1987 resolved_step, read_state
1988 )
1989 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
1990 result = self._invoke_agent(resolved, resolved_step, state)
1991 return self._propagate_step_errors(result, step_id)
1993 if step_type == "foreach":
1994 return self._run_foreach(step, state, working_set)
1996 if step_type == "call":
1997 # Build-time composition (ADR-0037 D4) rewrites every call step into
1998 # a native subgraph node, so a call reaching the runtime means the
1999 # builder skipped a composition path it was responsible for.
2000 raise RunHardFailure(
2001 "VALIDATION_ERROR",
2002 f"call step '{step_id}' reached runtime without build-time "
2003 "expansion (ADR-0037 D4) — not yet supported",
2004 {"step_id": step_id},
2005 )
2007 if step_type == _SUBGRAPH_STEP_TYPE:
2008 # Build-time composition (ADR-0037 D4) embeds a compiled function
2009 # subgraph as a native node. At the top level ``_assemble_graph``
2010 # wires it through ``_make_subgraph_node``; when the call sits in an
2011 # imperative foreach/while body the placeholder reaches ``run_step``
2012 # instead, so invoke the embedded graph here. The top-of-method
2013 # condition check already handled a false condition (returned the
2014 # input state unchanged), matching the node wrapper's skip.
2015 compiled = step.get(_SUBGRAPH_GRAPH_KEY)
2016 if compiled is None:
2017 raise RunHardFailure(
2018 "VALIDATION_ERROR",
2019 f"subgraph step '{step_id}' has no compiled graph",
2020 {"step_id": step_id},
2021 )
2022 config = _while_invoke_config(step)
2023 composed_step_ids = (
2024 step.get(_SUBGRAPH_STEP_IDS_KEY)
2025 if not step.get(_SUBGRAPH_WHILE_KEY)
2026 and not step.get(_SUBGRAPH_FOREACH_KEY)
2027 else None
2028 )
2029 # The compiled subgraph shares the parent's Core-namespaced schema.
2030 # Re-wrap on entry, flatten on exit. No channel reducer is
2031 # re-applied here — the subgraph's own commits already produced the
2032 # authoritative values.
2033 sub_result = compiled.invoke({_CORE: read_state}, config)
2034 committed = _flatten_channel_state(sub_result)
2035 return _filter_composed_step_ids(
2036 {**state, **committed}, composed_step_ids
2037 )
2039 raise RunHardFailure(
2040 "VALIDATION_ERROR",
2041 f"Unknown step type '{step_type}' in step '{step_id}'",
2042 {"step_id": step_id, "step_type": step_type},
2043 )
2045 def _run_parallel_fanout(
2046 self,
2047 step: dict,
2048 state: PipelineState,
2049 items: list,
2050 ctx: dict,
2051 working_set: ChannelWorkingSet,
2052 ) -> PipelineState:
2053 """
2054 Fan-out: call agent once per item, concurrently.
2055 Results stored as state[step_id] = {"results": [item_result, ...]}.
2056 Each item is available as state["item"] inside the invoke call.
2057 """
2058 step_id = step.get("id", "?")
2059 agent_name = step.get("agent")
2060 if not agent_name:
2061 log.warning("[step] %s parallel fanout: no agent", step_id)
2062 return {
2063 **state,
2064 step_id: project_aggregate_metadata({"results": []}),
2065 }
2067 resolved = self._resolver.resolve(agent_name)
2069 # Apply max_items cap
2070 max_items_raw = step.get("max_items")
2071 if max_items_raw is not None:
2072 max_items = _resolve_int(max_items_raw, ctx, len(items))
2073 items = items[:max_items]
2075 # Concurrency limit
2076 max_conc_raw = step.get("max_concurrent")
2077 if max_conc_raw is not None:
2078 max_conc = _resolve_int_strict(
2079 max_conc_raw,
2080 ctx,
2081 "max_concurrent",
2082 step_id,
2083 minimum=1,
2084 )
2085 else:
2086 max_conc = len(items) or 1
2087 max_conc = max(1, min(max_conc, len(items) or 1))
2089 log.debug(
2090 "[step] %s parallel fanout: %d items, %d concurrent",
2091 step_id, len(items), max_conc,
2092 )
2094 read_state = self._read_state(state, working_set)
2096 def call_one(item: Any) -> tuple[Any, dict | None, dict | None, list]:
2097 item_state = {**read_state, "_item": item, "item": item}
2098 item_ctx = self._make_ctx(item_state)
2099 return self._invoke_parallel_item(
2100 step, resolved, item, item_state, item_ctx
2101 )
2103 # Pre-allocate results to preserve input order regardless of completion
2104 # order (ENG-07), matching the concurrent foreach body-polku.
2105 results: list = [None] * len(items)
2106 item_metrics: dict = {}
2107 item_tool_meta: dict = {}
2108 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2109 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
2110 for future in as_completed(futures):
2111 idx = futures[future]
2112 try:
2113 item_result, item_metrics_map, item_tool_meta_map, *_ = (
2114 future.result()
2115 )
2116 results[idx] = item_result
2117 if isinstance(item_metrics_map, dict):
2118 item_metrics.update(item_metrics_map)
2119 if isinstance(item_tool_meta_map, dict):
2120 item_tool_meta.update(item_tool_meta_map)
2121 except SkillLoadError:
2122 raise
2123 except RunHardFailure:
2124 pool.shutdown(wait=False, cancel_futures=True)
2125 raise
2126 except Exception as exc: # noqa: BLE001
2127 log.error("[step] %s item failed: %s", step_id, exc)
2128 results[idx] = {"error": str(exc)}
2130 self._stage_fanout_channel_writes(step, results, working_set)
2132 aggregate = project_aggregate_metadata({"results": results})
2133 final: PipelineState = {**state, step_id: aggregate}
2134 if item_metrics:
2135 final = {
2136 **final,
2137 "step_context_metrics": _merge_side_channel(
2138 state.get("step_context_metrics"), item_metrics
2139 ),
2140 }
2141 if item_tool_meta:
2142 final = {
2143 **final,
2144 "step_tool_meta": _merge_side_channel(
2145 state.get("step_tool_meta"), item_tool_meta
2146 ),
2147 }
2148 return final
2150 def _invoke_parallel_item(
2151 self,
2152 step: dict,
2153 resolved: dict,
2154 item: Any,
2155 item_state: PipelineState,
2156 item_ctx: dict,
2157 ) -> tuple[Any, dict | None, dict | None, list]:
2158 """Run one parallel fan-out item's agent call.
2160 Shared by the in-node thread-pool fan-out and the native ``Send``
2161 worker, so both paths resolve input, communication context and tools
2162 identically. Returns ``(step_result, metrics, tool_meta, errors)``.
2163 """
2164 step_id = step.get("id", "?")
2165 resolved_step = self._resolve_step_runtime(
2166 step, item_state, item_ctx,
2167 input_override=item,
2168 )
2169 resolved_step = self._attach_communication_context(
2170 resolved_step, item_state
2171 )
2172 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2173 result_state = self._invoke_agent(resolved, resolved_step, item_state)
2174 return (
2175 result_state.get(step_id),
2176 result_state.get("step_context_metrics"),
2177 result_state.get("step_tool_meta"),
2178 list(result_state.get("errors") or []),
2179 )
2181 def _stage_fanout_channel_writes(
2182 self,
2183 step: dict,
2184 item_outputs: list,
2185 working_set: ChannelWorkingSet,
2186 ) -> None:
2187 """Stage each fan-out item's declared channel writes in item order.
2189 Runs on the main thread after the item pool completes, so the shared
2190 working set is never mutated from a worker thread. The fan-out
2191 envelope itself is never routed; only each item's own agent output is.
2192 This is a superstep fan-in staging path, so the implicit default channel
2193 is forbidden: ``allow_default`` stays unset.
2194 """
2195 if not self._channel_names:
2196 return
2197 for output in item_outputs:
2198 writes = _declared_output_channel_writes(
2199 step, output, self._channel_names
2200 )
2201 for name, value in writes.items():
2202 working_set.stage(name, value)
2204 def _run_foreach(
2205 self,
2206 step: dict,
2207 state: PipelineState,
2208 working_set: ChannelWorkingSet,
2209 ) -> PipelineState:
2210 """
2211 Foreach: iterate over the fan-out item list, run body steps per item.
2212 Item source from step.iteration (list/object literal or ``{{ }}``
2213 variable).
2215 When max_concurrent > 1, body iterations run concurrently via
2216 ThreadPoolExecutor — each item gets its own worker thread. The
2217 max_concurrent value sets the thread pool size (mirrors parallel
2218 fan-out semantics). Results list preserves original item order.
2220 max_output budget in concurrent mode: a stop flag prevents new workers
2221 from starting after the budget is exhausted. Workers already executing
2222 when the flag is set will complete naturally — at most max_concurrent
2223 extra items may run past the limit. This is accepted behaviour; the
2224 sequential early-exit guarantee cannot be reproduced without sacrificing
2225 true concurrency.
2227 The invocation working set is threaded into every body step so reads
2228 resolve over the committed snapshot plus this invocation's pending
2229 writes; the returned state is still built from the raw ``state``.
2230 """
2231 step_id = step.get("id", "?")
2232 read_state = self._read_state(state, working_set)
2233 ctx = self._make_ctx(read_state)
2235 items = self._foreach_item_list(step, ctx)
2237 # max_output cap — stop once N iterations have produced a successful
2238 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
2239 max_output_raw = step.get("max_output")
2240 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
2242 # max_concurrent — when > 1, enables parallel body execution. A present
2243 # value must resolve to a valid integer (ADR-0012), mirroring the
2244 # build-time resolution; it is never silently re-decided at runtime.
2245 max_conc_raw = step.get("max_concurrent")
2246 if max_conc_raw is not None:
2247 max_conc = _resolve_int_strict(
2248 max_conc_raw, ctx, "max_concurrent", step_id, minimum=1
2249 )
2250 else:
2251 max_conc = 1
2253 body = step.get("body") or []
2254 agent_name = step.get("agent")
2256 if max_conc <= 1:
2257 raise RunHardFailure(
2258 "VALIDATION_ERROR",
2259 f"foreach step '{step_id}' field 'max_concurrent' must resolve "
2260 "to a statically resolvable value > 1: a sequential foreach is "
2261 "compiled into a native cycle at build time and never reaches "
2262 "the concurrent runtime path",
2263 {"step_id": step_id, "field": "max_concurrent"},
2264 )
2266 # The declared max_conc decides the concurrent path; the runtime item
2267 # count only throttles the pool size, never the path selection.
2268 pool_size = max(1, min(max_conc, len(items)))
2269 return self._run_foreach_concurrent(
2270 step_id, step, state, items, body, agent_name, max_output, pool_size,
2271 working_set,
2272 )
2274 @staticmethod
2275 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
2276 """body-polku per-iteration result (r.67): the declared body-output
2277 step's result; else the iteration's `publish` result; else None."""
2278 if declared_output:
2279 return iter_state.get(declared_output)
2280 return iter_state.get("publish")
2282 @staticmethod
2283 def _body_item_passed(item: Any) -> bool:
2284 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
2285 `passed` bool; an item lacking such a key is not counted as failed."""
2286 if isinstance(item, dict):
2287 verdict = item.get("verdict")
2288 if verdict == "FAIL":
2289 return False
2290 if verdict == "PASS":
2291 return True
2292 passed = item.get("passed")
2293 if isinstance(passed, bool):
2294 return passed
2295 return True
2297 @staticmethod
2298 def _aggregate_body_results(item_results: list) -> dict:
2299 """body-polku aggregate (r.67): {results, passed, failing, failed}.
2301 Each item is projected to its envelope metadata before the verdict is
2302 derived, so both the persisted ``results`` list and ``failing`` are
2303 content-free while ``passed``/``failed`` semantics are unchanged.
2304 """
2305 projected = [project_envelope_metadata(r) for r in item_results]
2306 failing = [r for r in projected if not StepRunner._body_item_passed(r)]
2307 return {
2308 "results": projected,
2309 "passed": not failing,
2310 "failing": failing,
2311 "failed": bool(failing),
2312 }
2314 @staticmethod
2315 def _merge_item_working_sets(
2316 working_set: ChannelWorkingSet,
2317 item_working_sets: list[ChannelWorkingSet],
2318 ) -> None:
2319 """Stage each concurrent item's writes onto the shared set in item order."""
2320 for item_working_set in item_working_sets:
2321 for name, value in item_working_set.pending_writes():
2322 working_set.stage(name, value)
2324 def _run_foreach_concurrent(
2325 self,
2326 step_id: str,
2327 step: dict,
2328 state: PipelineState,
2329 items: list,
2330 body: list,
2331 agent_name: str | None,
2332 max_output: int,
2333 max_conc: int,
2334 working_set: ChannelWorkingSet,
2335 ) -> PipelineState:
2336 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2337 log.info(
2338 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2339 step_id, len(items), max_conc,
2340 )
2342 # Pre-allocate results list to preserve item order across threads.
2343 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2344 # legitimate None results (body produced no output) are kept.
2345 body_polku = bool(body)
2346 declared_output = step.get("output")
2347 _SKIPPED = object()
2348 results: list = [_SKIPPED] * len(items)
2349 # Each item writes into its own fork; the shared set is advanced only
2350 # after the pool completes, so an item skipped by the stop flag leaves
2351 # an empty fork that contributes no writes.
2352 item_working_sets: list[ChannelWorkingSet] = [
2353 working_set.fork() for _ in items
2354 ]
2355 lock = threading.Lock()
2356 produced_outputs = [0] # mutable counter shared across threads
2357 stop_flag = threading.Event() # set when max_output budget is exhausted
2358 item_errors: list = [] # budget errors collected from worker threads
2359 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2360 accumulated_tool_meta: dict = {} # per-item tool metadata from worker threads
2362 def run_item(idx: int, item: Any) -> None:
2363 nonlocal item_errors
2364 item_working_set = item_working_sets[idx]
2365 # Honour stop flag — skip if budget already exhausted before we start
2366 if max_output and stop_flag.is_set():
2367 return
2369 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2370 publish_before = iter_state.get("publish")
2372 if body_polku:
2373 for sub_step in body:
2374 iter_state = self._run_body_step(
2375 sub_step, iter_state, item_working_set
2376 )
2377 # Collect budget errors from this iteration (thread-safe merge)
2378 new_errs = iter_state.get("errors") or []
2379 if new_errs:
2380 with lock:
2381 item_errors = self._merge_errors(item_errors, new_errs)
2382 item_metrics_map = iter_state.get("step_context_metrics")
2383 if isinstance(item_metrics_map, dict):
2384 with lock:
2385 accumulated_metrics.update(item_metrics_map)
2386 item_tool_meta_map = iter_state.get("step_tool_meta")
2387 if isinstance(item_tool_meta_map, dict):
2388 with lock:
2389 accumulated_tool_meta.update(item_tool_meta_map)
2390 publish_out = iter_state.get("publish")
2391 results[idx] = self._body_item_result(iter_state, declared_output)
2392 if (
2393 isinstance(publish_out, dict)
2394 and publish_out.get("path")
2395 and publish_out != publish_before
2396 ):
2397 with lock:
2398 produced_outputs[0] += 1
2399 if max_output and produced_outputs[0] >= max_output:
2400 stop_flag.set()
2401 elif agent_name:
2402 resolved = self._resolver.resolve(agent_name)
2403 item_read_state = self._read_state(iter_state, item_working_set)
2404 item_ctx = self._make_ctx(item_read_state)
2405 resolved_step = self._resolve_step_runtime(
2406 step, item_read_state, item_ctx
2407 )
2408 resolved_step = self._attach_communication_context(
2409 resolved_step, item_read_state
2410 )
2411 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2412 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2413 results[idx] = iter_state.get(step_id)
2414 item_metrics_map = iter_state.get("step_context_metrics")
2415 if isinstance(item_metrics_map, dict):
2416 with lock:
2417 accumulated_metrics.update(item_metrics_map)
2418 item_tool_meta_map = iter_state.get("step_tool_meta")
2419 if isinstance(item_tool_meta_map, dict):
2420 with lock:
2421 accumulated_tool_meta.update(item_tool_meta_map)
2423 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2424 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2425 for future in as_completed(futures):
2426 try:
2427 future.result()
2428 except SkillLoadError:
2429 raise
2430 except RunHardFailure:
2431 pool.shutdown(wait=False, cancel_futures=True)
2432 raise
2433 except Exception as exc: # noqa: BLE001
2434 idx = futures[future]
2435 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2436 results[idx] = {"error": str(exc)}
2438 self._merge_item_working_sets(working_set, item_working_sets)
2440 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2441 ordered = [r for r in results if r is not _SKIPPED]
2442 step_result: Any = (
2443 self._aggregate_body_results(ordered)
2444 if body_polku
2445 else [project_envelope_metadata(r) for r in ordered]
2446 )
2447 final: PipelineState = {**state, step_id: step_result}
2448 if accumulated_metrics:
2449 final = {
2450 **final,
2451 "step_context_metrics": _merge_side_channel(
2452 state.get("step_context_metrics"), accumulated_metrics
2453 ),
2454 }
2455 if accumulated_tool_meta:
2456 final = {
2457 **final,
2458 "step_tool_meta": _merge_side_channel(
2459 state.get("step_tool_meta"), accumulated_tool_meta
2460 ),
2461 }
2462 if item_errors:
2463 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2464 return final
2467# ---------------------------------------------------------------------------
2468# GraphBuilder
2469# ---------------------------------------------------------------------------
2471def _topological_order(steps: list[dict]) -> list[dict]:
2472 """
2473 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2474 """
2475 id_to_step = {s["id"]: s for s in steps}
2476 visited: set[str] = set()
2477 order: list[dict] = []
2479 def visit(step_id: str) -> None:
2480 if step_id in visited:
2481 return
2482 visited.add(step_id)
2483 for dep in (id_to_step[step_id].get("depends_on") or []):
2484 if dep in id_to_step:
2485 visit(dep)
2486 order.append(id_to_step[step_id])
2488 for s in steps:
2489 visit(s["id"])
2490 return order
2493def _group_dependencies(
2494 groups: list[list[dict]],
2495) -> tuple[list[set[int]], list[set[int]]]:
2496 """Group-index adjacency (predecessors, successors) from each step's depends_on."""
2497 step_to_group = {
2498 step["id"]: index
2499 for index, group in enumerate(groups)
2500 for step in group
2501 }
2502 predecessors: list[set[int]] = [set() for _ in groups]
2503 successors: list[set[int]] = [set() for _ in groups]
2504 for index, group in enumerate(groups):
2505 for step in group:
2506 for dep in step.get("depends_on") or []:
2507 dep_index = step_to_group.get(dep)
2508 if dep_index is None or dep_index == index:
2509 continue
2510 predecessors[index].add(dep_index)
2511 successors[dep_index].add(index)
2512 return predecessors, successors
2515@dataclass(frozen=True)
2516class _RoutePlan:
2517 """Compiled `routes` topology: route tables plus the static-edge suppression sets.
2519 `routes` owns a routing step's outgoing control flow (ADR-0025). A route
2520 target reached via a static `depends_on` edge is therefore suppressed and
2521 replaced by the native conditional edge; the plan records both the targets
2522 (so they are never made entry points) and the suppressed `(source, target)`
2523 pairs.
2524 """
2526 step_to_node: dict[str, str]
2527 routes_by_step: dict[str, list[dict]]
2528 target_node_ids: frozenset[str]
2529 routing_node_ids: frozenset[str]
2530 suppressed_edges: frozenset[tuple[str, str]]
2533def _reject_nested_routes(step: dict) -> None:
2534 """Reject `routes` on a nested `body` step (top-level-only rule, ADR-0012).
2536 The engine wires `routes` only for top-level steps, so a nested declaration
2537 would be silently ignored; nested routing is deferred.
2538 """
2539 body = step.get("body")
2540 if not isinstance(body, list):
2541 return
2542 for nested in _iter_steps(body):
2543 if nested.get("routes"):
2544 raise RunHardFailure(
2545 "VALIDATION_ERROR",
2546 f"Step '{nested.get('id')}' declares `routes` inside a nested "
2547 "body; `routes` is only allowed on top-level steps",
2548 )
2551def _build_route_plan(groups: list[list[dict]], node_ids: list[str]) -> _RoutePlan:
2552 """Resolve every top-level step's `routes` into a node-level routing plan."""
2553 step_to_node = {
2554 step["id"]: node_ids[index]
2555 for index, group in enumerate(groups)
2556 for step in group
2557 }
2558 step_to_deps = {
2559 step["id"]: set(step.get("depends_on") or [])
2560 for group in groups
2561 for step in group
2562 }
2563 routes_by_step: dict[str, list[dict]] = {}
2564 target_node_ids: set[str] = set()
2565 routing_node_ids: set[str] = set()
2566 suppressed_edges: set[tuple[str, str]] = set()
2567 for group in groups:
2568 for step in group:
2569 _reject_nested_routes(step)
2570 routes = step.get("routes")
2571 if not routes:
2572 continue
2573 source = step["id"]
2574 source_node = step_to_node[source]
2575 routing_node_ids.add(source_node)
2576 routes_by_step[source] = routes
2577 for route in routes:
2578 target = route.get("target")
2579 target_node = step_to_node.get(target)
2580 if target_node is None:
2581 raise RunHardFailure(
2582 "VALIDATION_ERROR",
2583 f"Step '{source}' route target '{target}' is not a "
2584 "top-level step in this pipeline",
2585 )
2586 target_node_ids.add(target_node)
2587 if source in step_to_deps.get(target, set()):
2588 suppressed_edges.add((source_node, target_node))
2589 log.warning(
2590 "[routes] suppressing static edge %s -> %s; route "
2591 "targets are wired via `routes`",
2592 source,
2593 target,
2594 )
2595 return _RoutePlan(
2596 step_to_node=step_to_node,
2597 routes_by_step=routes_by_step,
2598 target_node_ids=frozenset(target_node_ids),
2599 routing_node_ids=frozenset(routing_node_ids),
2600 suppressed_edges=frozenset(suppressed_edges),
2601 )
2604def _make_route_router(
2605 routes: list[dict], step_to_node: dict[str, str], end, step_runner: Any
2606) -> Any:
2607 """First-match-wins router over a routing step's declared routes.
2609 Evaluates `when` in declaration order and returns the selected target node;
2610 a route without `when` is the default branch. With no match and no default
2611 it returns the graph terminal.
2612 """
2613 def router(state: PipelineState) -> str:
2614 flat = _flatten_channel_state(state)
2615 ctx = step_runner._make_ctx(flat)
2616 for route in routes:
2617 if _eval_condition(route.get("when"), ctx):
2618 return step_to_node[route["target"]]
2619 return end
2621 return router
2624def _wire_route_edges(
2625 graph, route_plan: _RoutePlan, end, step_runner: Any
2626) -> None:
2627 """Add one native conditional edge per routing step, with an explicit path_map."""
2628 for source, routes in route_plan.routes_by_step.items():
2629 target_nodes = [route_plan.step_to_node[r["target"]] for r in routes]
2630 path_map = {node: node for node in target_nodes}
2631 path_map[end] = end
2632 graph.add_conditional_edges(
2633 route_plan.step_to_node[source],
2634 _make_route_router(routes, route_plan.step_to_node, end, step_runner),
2635 path_map,
2636 )
2639def _wire_dependency_edges(
2640 graph,
2641 groups: list[list[dict]],
2642 node_ids: list[str],
2643 end,
2644 route_plan: _RoutePlan,
2645 entry_ids: list[str] | None = None,
2646 exit_ids: list[str] | None = None,
2647) -> None:
2648 """Wire native branches and fan-in joins from group-level depends_on edges.
2650 Static edges originating from a routing step to one of its route targets are
2651 suppressed — `routes` supplies that control flow. A route target with no
2652 other predecessor is reached by its conditional edge, never an entry point.
2653 A group split into a Send fan-out enters through its dispatch node and exits
2654 through its join node, so ``entry_ids``/``exit_ids`` name those boundaries.
2655 """
2656 entry_ids = entry_ids if entry_ids is not None else node_ids
2657 exit_ids = exit_ids if exit_ids is not None else node_ids
2658 predecessors, successors = _group_dependencies(groups)
2659 for index, node_id in enumerate(node_ids):
2660 entry = entry_ids[index]
2661 pred_nodes = [exit_ids[p] for p in sorted(predecessors[index])]
2662 if not pred_nodes:
2663 if node_id not in route_plan.target_node_ids:
2664 graph.set_entry_point(entry)
2665 else:
2666 live_preds = [
2667 pred
2668 for pred in pred_nodes
2669 if (pred, entry) not in route_plan.suppressed_edges
2670 ]
2671 if len(live_preds) == 1:
2672 graph.add_edge(live_preds[0], entry)
2673 elif live_preds:
2674 graph.add_edge(live_preds, entry)
2675 if not successors[index] and node_id not in route_plan.routing_node_ids:
2676 graph.add_edge(exit_ids[index], end)
2679# ---------------------------------------------------------------------------
2680# Call-step build-time expansion (ADR-0037 D3/D4)
2681# ---------------------------------------------------------------------------
2682#
2683# A `type: call` step is composed at build time: the function's body steps plus
2684# two synthesized op steps (`__call_bind` / `__call_emit`) are assembled into a
2685# compiled subgraph added as a single parent-graph node keyed by the call id. The
2686# runtime executes only native steps; LangGraph never sees a call step. The caller
2687# reads the function outputs via the call result shape
2688# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
2689# The function body's `{{ params.<name> }}` references are rewritten to
2690# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves. A
2691# call nested in a foreach/while body is composed the same way: the loop body is a
2692# native subgraph and the call becomes a compiled subgraph node keyed by the call id.
2693#
2694# Isolation (ADR-0037 D5 / 00-functions § Isolation) and recursion (ADR-0037 D7): a
2695# function body step must not reference a step-id outside its own body (a
2696# one-way dependency check — outward references to caller step-ids are build
2697# errors); a cycle in the function call graph (self or mutual) is a build
2698# error. Bounded while loops are runtime constructs and are not cycle-flagged.
2700# A bare `params` reference root (`params`, `params.x`, `params['x']`). A
2701# preceding `.` or word character means it is not the root (`foo.params`,
2702# `myparams`, `paramsfoo`). Quoted literals are consumed whole, so a `params`
2703# inside them is invisible to the bare-reference branch.
2704_QUOTED_OR_BARE_PARAMS_RE = re.compile(
2705 r"'[^']*'|\"[^\"]*\"|(?<![\w.])params(?![\w])"
2706)
2708# A leading `params` root in an `input` bare dot-path (`params.x`, `params`,
2709# `params[0]`), optionally after whitespace.
2710_LEADING_PARAMS_RE = re.compile(r"^(\s*)params(?=\.|\[|\s*$)")
2713def _scope_template_expression(expr: str, call_id: str) -> str:
2714 """Rewrite bare `params` references inside one `{{ ... }}` token."""
2716 def _replace(match: re.Match) -> str:
2717 token = match.group(0)
2718 return token if token[0] in "'\"" else f"{call_id}.params"
2720 return _QUOTED_OR_BARE_PARAMS_RE.sub(_replace, expr)
2723def _scope_template_string(value: str, call_id: str) -> str:
2724 """Scope the `{{ ... }}` expressions of a template-only field value."""
2725 return _TEMPLATE_REF_RE.sub(
2726 lambda m: _scope_template_expression(m.group(0), call_id),
2727 value,
2728 )
2731def _scope_input_string(value: str, call_id: str) -> str:
2732 """Scope an `input`: its `{{ ... }}` templates or its leading bare dot-path."""
2733 if "{{" in value:
2734 return _scope_template_string(value, call_id)
2735 if value.lstrip().startswith("["):
2736 return value
2737 return _LEADING_PARAMS_RE.sub(lambda m: f"{m.group(1)}{call_id}.params", value)
2740def _scope_params_value(value: Any, call_id: str, field: str) -> Any:
2741 """Rewrite bare `params` references to the call-scoped namespace.
2743 ``field`` selects the resolver semantics: ``input`` is a template or a bare
2744 state dot-path; every other scoped field is template-only. Dict keys are
2745 untouched and the input is never mutated.
2746 """
2747 if isinstance(value, str):
2748 if field == "input":
2749 return _scope_input_string(value, call_id)
2750 return _scope_template_string(value, call_id)
2751 if isinstance(value, dict):
2752 return {k: _scope_params_value(v, call_id, field) for k, v in value.items()}
2753 if isinstance(value, list):
2754 return [_scope_params_value(v, call_id, field) for v in value]
2755 return value
2758def _scope_params(step: dict, call_id: str) -> dict:
2759 """Deep-scope a function body step's params references for a call id."""
2760 scoped: dict = dict(step)
2761 for field in ("input", "iteration", "params", "condition", "exit_condition", "skip_llm_if"):
2762 if field in scoped and isinstance(scoped[field], (str, dict, list)):
2763 scoped[field] = _scope_params_value(scoped[field], call_id, field)
2764 if "body" in scoped and isinstance(scoped["body"], list):
2765 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
2766 return scoped
2769def _collect_step_ids(steps: list[dict]) -> set[str]:
2770 """All step ids declared in *steps*, including nested body step ids."""
2771 ids: set[str] = set()
2772 for step in steps:
2773 step_id = step.get("id")
2774 if isinstance(step_id, str):
2775 ids.add(step_id)
2776 body = step.get("body")
2777 if isinstance(body, list):
2778 ids |= _collect_step_ids(body)
2779 return ids
2782def _iter_steps(steps: list[dict]):
2783 """Yield every step in *steps*, recursing into nested bodies."""
2784 for step in steps:
2785 yield step
2786 body = step.get("body")
2787 if isinstance(body, list):
2788 yield from _iter_steps(body)
2791def _collect_composed_step_ids(steps: list[dict]) -> set[str]:
2792 """All declared step ids, including ids composed into native subgraphs.
2794 A ``call`` step becomes a ``__subgraph__`` placeholder whose inner step ids
2795 live only inside the compiled subgraph. The placeholder records them under
2796 ``_SUBGRAPH_STEP_IDS_KEY`` so name-collision validation sees the full set,
2797 which stays encapsulated inside the compiled subgraph.
2798 """
2799 ids = _collect_step_ids(steps)
2800 for step in _iter_steps(steps):
2801 composed_ids = step.get(_SUBGRAPH_STEP_IDS_KEY)
2802 if isinstance(composed_ids, (set, frozenset)):
2803 ids |= composed_ids
2804 return ids
2807_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
2808_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
2809_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
2811_REFERENCE_FIELDS = (
2812 "input", "iteration", "params", "condition", "exit_condition", "skip_llm_if"
2813)
2815# Engine-reserved template namespaces — never step-id references, even if a
2816# caller happens to declare a step with one of these names.
2817_RESERVED_NAMESPACES = frozenset(
2818 {"params", "item", "_item", "budget", "config", "spec", "channel", "lastvalue"}
2819)
2822def _referenced_names(value: Any) -> set[str]:
2823 """Root identifiers read by templated strings.
2825 Only the leading identifier of each reference (before the first ``.`` or
2826 ``[``) can name a step-id; nested field segments are ignored. Quoted
2827 literals are ignored.
2828 """
2829 names: set[str] = set()
2830 if isinstance(value, str):
2831 for expr in _TEMPLATE_REF_RE.findall(value):
2832 bare = _STRING_LITERAL_RE.sub("", expr)
2833 for match in _IDENTIFIER_RE.finditer(bare):
2834 if match.start() == 0 or bare[match.start() - 1] != ".":
2835 names.add(match.group())
2836 elif isinstance(value, dict):
2837 for v in value.values():
2838 names |= _referenced_names(v)
2839 elif isinstance(value, list):
2840 for v in value:
2841 names |= _referenced_names(v)
2842 return names
2845def _validate_function_purity(
2846 fn_name: str, body: list[dict], visible_ids: set[str]
2847) -> None:
2848 """One-way isolation: a function body must not reference a caller step-id.
2850 A body step may read its own body step-ids and the params namespace; a
2851 reference to a step-id that exists outside the body (``visible_ids`` but
2852 not declared in the body) is an outward dependency and a hard build error.
2853 """
2854 local_ids = _collect_step_ids(body)
2855 for step in _iter_steps(body):
2856 step_id = step.get("id", "?")
2857 for dep in step.get("depends_on") or []:
2858 if dep in visible_ids and dep not in local_ids:
2859 raise RunHardFailure(
2860 "VALIDATION_ERROR",
2861 f"function '{fn_name}' step '{step_id}': depends_on "
2862 f"'{dep}' is outside the function body (isolation — "
2863 "function bodies see only their own steps and params)",
2864 )
2865 for field in _REFERENCE_FIELDS:
2866 for ref in _referenced_names(step.get(field)):
2867 if ref in _RESERVED_NAMESPACES:
2868 continue
2869 if ref in visible_ids and ref not in local_ids:
2870 raise RunHardFailure(
2871 "VALIDATION_ERROR",
2872 f"function '{fn_name}' step '{step_id}': '{field}' "
2873 f"references '{ref}', a step-id outside the function "
2874 "body (isolation — function bodies see only their own "
2875 "steps and params)",
2876 )
2879def _reject_function_body_routes(fn_name: str, body: list[dict]) -> None:
2880 """Reject `routes` anywhere in a function body (top-level-only rule).
2882 A function body is composed as a native subgraph, so a body step declaring
2883 `routes` would escape the nested-`body` guard and be silently rewired;
2884 reject it at build time instead.
2885 """
2886 for step in _iter_steps(body):
2887 if step.get("routes"):
2888 raise RunHardFailure(
2889 "VALIDATION_ERROR",
2890 f"function '{fn_name}' step '{step.get('id')}' declares `routes`; "
2891 "`routes` is only allowed on top-level steps",
2892 )
2895def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
2896 """Expand every ``call`` step through the native subgraph composer.
2898 Test-facing seam over the single call-expansion implementation: the same
2899 ``_CallComposer`` that ``GraphBuilder`` uses composes the native subgraph, so
2900 no second expansion path exists. The embedded subgraph value is the un-compiled
2901 sub-step list; a runnable graph only needs the compiled form, which
2902 production obtains through ``GraphBuilder._compose_calls``.
2903 """
2904 composer = _CallComposer(
2905 registry,
2906 assemble_subgraph=lambda sub_steps: sub_steps,
2907 channels=[],
2908 channel_names=frozenset(),
2909 )
2910 return composer.compose(steps)
2913_SUBGRAPH_STEP_TYPE = "__subgraph__"
2914_SUBGRAPH_GRAPH_KEY = "__subgraph_graph__"
2915_SUBGRAPH_STEP_IDS_KEY = "__subgraph_step_ids__"
2917# A `while` step is composed into a `__subgraph__` placeholder whose embedded
2918# graph is a native LangGraph cycle (head -> body -> cond -> head/exit). The
2919# marker flags the placeholder so the parent uses the while node wrapper (which
2920# keeps the loop-internal state out of the parent result) rather than the plain
2921# call-subgraph node, and carries the build-time recursion bound.
2922_SUBGRAPH_WHILE_KEY = "__subgraph_while__"
2923_WHILE_RECURSION_LIMIT_KEY = "__while_recursion_limit__"
2925# Engine-local cycle state: the active while's `{current, iteration, passed,
2926# skipped, max_iter}` marker. Nested whiles live in isolated cycle
2927# invocations, so a shared key cannot collide across them.
2928_WHILE_STATE_KEY = "__while_state__"
2930# State keys the while wrapper propagates from the isolated cycle back to the
2931# parent (the loop result itself is carried separately under the step id).
2932_WHILE_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
2934# A sequential `foreach` (max_concurrent absent or resolving to <= 1) is
2935# composed into a `__subgraph__` placeholder whose embedded graph is a native
2936# LangGraph cycle (head -> body/agent -> cond -> head/exit). The marker flags
2937# the placeholder so the parent uses the foreach node wrapper, and carries the
2938# build-time recursion bound.
2939_SUBGRAPH_FOREACH_KEY = "__subgraph_foreach__"
2940_FOREACH_RECURSION_LIMIT_KEY = "__foreach_recursion_limit__"
2942# Engine-local cycle state: the active foreach's `{items, index, results,
2943# produced_outputs, max_output}` marker. Nested foreachs live in isolated cycle
2944# invocations, so a shared key cannot collide across them.
2945_FOREACH_STATE_KEY = "__foreach_state__"
2947# State keys the foreach wrapper propagates from the isolated cycle back to the
2948# parent (the aggregate result itself is carried separately under the step id).
2949_FOREACH_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
2951# Build-time recursion bound fallback for a foreach with no declared
2952# `max_iterations`: the item list is uncapped, so reserve a generous bound.
2953_FOREACH_DEFAULT_MAX_ITER = 1000
2956class _CallComposer:
2957 """Compose `call` steps into compiled native subgraph nodes (ADR-0037 D4).
2959 A function body becomes its own compiled StateGraph added to the parent as a
2960 single node keyed by the call id; the ``bind``/``emit`` scaffolding and the
2961 body steps live inside that subgraph and share the parent state. Nested
2962 function calls compose recursively. A `call` nested in a foreach/while body
2963 is composed the same way; the loop body executes the embedded compiled graph
2964 imperatively via ``run_step``'s ``__subgraph__`` branch.
2965 """
2967 def __init__(
2968 self,
2969 registry: ManifestRegistry,
2970 assemble_subgraph: Any,
2971 channels: list[dict],
2972 channel_names: frozenset[str],
2973 *,
2974 assemble_while: Any | None = None,
2975 compose_whiles: bool = False,
2976 assemble_foreach: Any | None = None,
2977 compose_foreachs: bool = False,
2978 build_ctx: dict | None = None,
2979 ) -> None:
2980 self._registry = registry
2981 self._assemble_subgraph = assemble_subgraph
2982 self._channels = channels
2983 self._channel_names = channel_names
2984 self._assemble_while = assemble_while
2985 self._compose_whiles = compose_whiles
2986 self._assemble_foreach = assemble_foreach
2987 self._compose_foreachs = compose_foreachs
2988 self._build_ctx = build_ctx or {}
2989 self._stack: list[str] = []
2990 self._expanded_functions: dict[str, str] = {}
2992 def compose(self, steps: list[dict]) -> list[dict]:
2993 visible_ids = _collect_step_ids(steps)
2994 composed: list[dict] = []
2995 for step in steps:
2996 composed.extend(self._compose_step(step, visible_ids=visible_ids))
2997 return composed
2999 def _compose_step(self, step: dict, *, visible_ids: set[str]) -> list[dict]:
3000 step_type = step.get("type")
3001 if step_type == "call":
3002 return [self._compose_subgraph_call(step, visible_ids)]
3003 if step_type == "while" and self._compose_whiles:
3004 return [self._compose_while(step, visible_ids)]
3005 if step_type == "foreach" and self._should_compose_foreach(step):
3006 return [self._compose_foreach(step, visible_ids)]
3007 body = step.get("body")
3008 if not isinstance(body, list):
3009 return [step]
3010 nested: list[dict] = []
3011 for sub in body:
3012 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3013 return [{**step, "body": nested}]
3015 def _compose_while(self, step: dict, visible_ids: set[str]) -> dict:
3016 """Compose a `while` step into a native LangGraph cycle subgraph.
3018 The embedded graph owns the loop topology: a head that resolves the
3019 entry condition / initial ``current``, the composed body step nodes, a
3020 conditional back-edge that computes the feed + exit condition and routes
3021 continue-vs-exit, and an exit node that writes the documented while
3022 result. ``expand_calls`` leaves the step uncomposed (``compose_whiles``
3023 false) so the call-expansion seam still exposes the loop body.
3024 """
3025 _reject_nested_routes(step)
3026 body = step.get("body") or []
3027 nested: list[dict] = []
3028 for sub in body:
3029 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3030 _validate_step_channel_bindings(nested, self._channel_names)
3031 compiled, recursion_limit = self._assemble_while(step, nested)
3032 placeholder: dict = {
3033 "id": step["id"],
3034 "type": _SUBGRAPH_STEP_TYPE,
3035 _SUBGRAPH_GRAPH_KEY: compiled,
3036 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3037 _SUBGRAPH_WHILE_KEY: True,
3038 _WHILE_RECURSION_LIMIT_KEY: recursion_limit,
3039 "depends_on": list(step.get("depends_on") or []),
3040 }
3041 if step.get("routes"):
3042 placeholder["routes"] = step["routes"]
3043 return placeholder
3045 def _should_compose_foreach(self, step: dict) -> bool:
3046 """Whether a `foreach` is the sequential form composed into a cycle.
3048 Only a `max_concurrent` absent or resolving to ``<= 1`` is composed; a
3049 declared ``max_concurrent > 1`` keeps the native ``Send`` / in-node
3050 concurrent behaviour.
3051 """
3052 if not self._compose_foreachs:
3053 return False
3054 return _resolved_send_concurrency(step, self._build_ctx) <= 1
3056 def _compose_foreach(self, step: dict, visible_ids: set[str]) -> dict:
3057 """Compose a sequential `foreach` into a native LangGraph cycle subgraph.
3059 Mirrors ``_compose_while``: the embedded graph owns the loop topology
3060 (a head that resolves the item list and injects the current item, the
3061 composed body step nodes or the per-item agent node, a conditional
3062 back-edge routing next-item-vs-exit, and an exit node that writes the
3063 documented aggregate). ``expand_calls`` leaves the step uncomposed so
3064 the call-expansion seam still exposes the loop body.
3065 """
3066 _reject_nested_routes(step)
3067 body = step.get("body") or []
3068 nested: list[dict] = []
3069 for sub in body:
3070 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3071 _validate_step_channel_bindings(nested, self._channel_names)
3072 compiled, recursion_limit = self._assemble_foreach(step, nested)
3073 placeholder: dict = {
3074 "id": step["id"],
3075 "type": _SUBGRAPH_STEP_TYPE,
3076 _SUBGRAPH_GRAPH_KEY: compiled,
3077 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3078 _SUBGRAPH_FOREACH_KEY: True,
3079 _FOREACH_RECURSION_LIMIT_KEY: recursion_limit,
3080 "depends_on": list(step.get("depends_on") or []),
3081 }
3082 if step.get("routes"):
3083 placeholder["routes"] = step["routes"]
3084 return placeholder
3086 def _compose_subgraph_call(self, step: dict, visible_ids: set[str]) -> dict:
3087 call_id = step["id"]
3088 function, body = self._check_call_guards(step, visible_ids)
3090 nested_visible = visible_ids | _collect_step_ids(body)
3091 outputs_spec = (
3092 ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
3093 )
3094 bind_id = f"_call__{call_id}__bind"
3095 emit_id = f"_call__{call_id}__emit"
3096 bind_step: dict = {
3097 "id": bind_id,
3098 "type": "builtin",
3099 "component": "__call_bind",
3100 "params": {**(step.get("params") or {}), "__call_id": call_id},
3101 "depends_on": [],
3102 }
3104 self._stack.append(step["function"])
3105 try:
3106 expanded_body = self._compose_body(
3107 body, call_id, bind_id, nested_visible
3108 )
3109 finally:
3110 self._stack.pop()
3112 emit_step: dict = {
3113 "id": emit_id,
3114 "type": "builtin",
3115 "component": "__call_emit",
3116 "params": {"__call_id": call_id, "__outputs": outputs_spec},
3117 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
3118 }
3119 sub_steps = [bind_step, *expanded_body, emit_step]
3120 if self._channels:
3121 _validate_channel_inputs(self._channels, sub_steps)
3122 _validate_step_channel_bindings(sub_steps, self._channel_names)
3123 placeholder: dict = {
3124 "id": call_id,
3125 "type": _SUBGRAPH_STEP_TYPE,
3126 _SUBGRAPH_GRAPH_KEY: self._assemble_subgraph(
3127 _topological_order(sub_steps)
3128 ),
3129 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(sub_steps),
3130 "depends_on": list(step.get("depends_on") or []),
3131 }
3132 call_condition = step.get("condition")
3133 if call_condition is not None:
3134 placeholder["condition"] = call_condition
3135 return placeholder
3137 def _compose_body(
3138 self, body: list[dict], call_id: str, bind_id: str, visible_ids: set[str]
3139 ) -> list[dict]:
3140 expanded: list[dict] = []
3141 for body_step in body:
3142 scoped = _scope_params(body_step, call_id)
3143 for sub in self._compose_step(scoped, visible_ids=visible_ids):
3144 deps = list(sub.get("depends_on") or [])
3145 if bind_id in deps:
3146 expanded.append(sub)
3147 else:
3148 expanded.append({**sub, "depends_on": [bind_id, *deps]})
3149 return expanded
3151 def _check_call_guards(
3152 self, step: dict, visible_ids: set[str]
3153 ) -> tuple[dict, list[dict]]:
3154 """Single owner of every build-time ``call`` guard (ADR-0037 D3/D5/D7).
3156 Rejects routes on the call step, function cycles, duplicate expansion,
3157 unknown functions, function-body isolation (purity) violations, and
3158 body-declared routes; returns the resolved function and its body.
3159 """
3160 fn_name = step["function"]
3161 call_id = step["id"]
3163 if step.get("routes"):
3164 raise RunHardFailure(
3165 "VALIDATION_ERROR",
3166 f"call step '{call_id}' declares `routes`; `routes` is not supported "
3167 "on call steps",
3168 )
3169 if fn_name in self._stack:
3170 chain = " -> ".join(self._stack + [fn_name])
3171 raise RunHardFailure(
3172 "VALIDATION_ERROR",
3173 f"cycle detected in function expansion: {chain} "
3174 "(ADR-0037 D7 — recursion must be a bounded while loop or "
3175 "idempotent re-run)",
3176 )
3177 if fn_name in self._expanded_functions:
3178 raise RunHardFailure(
3179 "VALIDATION_ERROR",
3180 f"duplicate call expansion: function '{fn_name}' called at both "
3181 f"'{self._expanded_functions[fn_name]}' and '{call_id}' in one "
3182 "pipeline (ADR-0037 D7 — repeated calls to the same function are "
3183 "not yet supported)",
3184 )
3185 self._expanded_functions[fn_name] = call_id
3187 function = self._registry.get("Function", fn_name)
3188 if function is None:
3189 raise RunHardFailure(
3190 "VALIDATION_ERROR",
3191 f"call step '{call_id}' references unknown function '{fn_name}' "
3192 "(no kind: Function manifest in the workflows tree)",
3193 )
3194 body = (function.get("spec") or {}).get("steps") or []
3195 _validate_function_purity(fn_name, body, visible_ids)
3196 _reject_function_body_routes(fn_name, body)
3197 _validate_unique_step_ids(body)
3198 _validate_foreach_bodies(body)
3199 _validate_loop_limits(body, self._build_ctx)
3200 _validate_fail_if_exhausted_steps(body)
3201 return function, body
3204_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
3205_BINARY_OPERATOR_VALUES = frozenset(
3206 {
3207 "shallow_merge",
3208 "prune_sensitive",
3209 "list_dedup",
3210 "json_merge_patch",
3211 "json_merge_append",
3212 "json_merge_dedup",
3213 }
3214)
3217def _validate_channel_entry(index: int, channel: Any) -> None:
3218 """Validate one spec.channels entry against the frozen ChannelSpec shape.
3220 The engine build path does not run manifest-schema validation, so a
3221 malformed entry is rejected here — before any ``["name"]`` indexing — with
3222 an error that identifies the offending entry.
3223 """
3224 if not isinstance(channel, dict):
3225 raise RunHardFailure(
3226 "CHANNEL_BINDING_ERROR",
3227 f"Pipeline spec.channels[{index}] must be a mapping; got "
3228 f"{type(channel).__name__}",
3229 )
3230 name = channel.get("name")
3231 if not isinstance(name, str) or not name:
3232 raise RunHardFailure(
3233 "CHANNEL_BINDING_ERROR",
3234 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3235 "channel requires a non-empty string 'name'",
3236 )
3237 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3238 raise RunHardFailure(
3239 "CHANNEL_BINDING_ERROR",
3240 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3241 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}",
3242 )
3243 effective_type = channel.get("type", "LastValue")
3244 if "operator" in channel:
3245 operator = channel["operator"]
3246 if not isinstance(operator, str):
3247 raise RunHardFailure(
3248 "CHANNEL_BINDING_ERROR",
3249 f"Pipeline spec.channels[{index}] ('{name}') has a non-string "
3250 f"'operator' of type '{type(operator).__name__}'; 'operator' "
3251 "must be a string",
3252 )
3253 if effective_type != "BinaryOperatorAggregate":
3254 raise RunHardFailure(
3255 "CHANNEL_BINDING_ERROR",
3256 f"Pipeline spec.channels[{index}] ('{name}') sets 'operator' "
3257 "but 'operator' is only valid on a 'BinaryOperatorAggregate' "
3258 f"channel; effective type is '{effective_type}'",
3259 )
3260 if operator not in _BINARY_OPERATOR_VALUES:
3261 raise RunHardFailure(
3262 "CHANNEL_BINDING_ERROR",
3263 f"Pipeline spec.channels[{index}] ('{name}') has unknown "
3264 f"'operator' '{operator}'; expected one of "
3265 f"{sorted(_BINARY_OPERATOR_VALUES)}",
3266 )
3267 elif effective_type == "BinaryOperatorAggregate":
3268 raise RunHardFailure(
3269 "CHANNEL_BINDING_ERROR",
3270 f"Pipeline spec.channels[{index}] ('{name}') is a "
3271 "'BinaryOperatorAggregate' channel and requires an 'operator'; "
3272 f"expected one of {sorted(_BINARY_OPERATOR_VALUES)}",
3273 )
3276def _validated_channels(raw_channels: Any) -> list[dict]:
3277 """Return spec.channels as a validated list, or [] when absent.
3279 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3280 """
3281 if raw_channels is None:
3282 return []
3283 if not isinstance(raw_channels, list):
3284 raise RunHardFailure(
3285 "CHANNEL_BINDING_ERROR",
3286 f"Pipeline spec.channels must be a list; got "
3287 f"{type(raw_channels).__name__}",
3288 )
3289 for index, channel in enumerate(raw_channels):
3290 _validate_channel_entry(index, channel)
3291 return raw_channels
3294def _validate_step_channel_bindings(
3295 steps: list[dict], channel_names: frozenset[str]
3296) -> None:
3297 """Reject malformed or undeclared step-level channel bindings at build time.
3299 Every step's declared ``output_channel`` must be a non-empty declared
3300 channel name, and every declared ``input_channels`` entry must be a
3301 non-empty declared channel name. This runs unconditionally, so a binding
3302 to a channel absent from ``spec.channels`` fails the build even when no
3303 channels are declared.
3304 """
3305 for step in _iter_steps(steps):
3306 _validate_step_output_channel(step, channel_names)
3307 _validate_step_input_channels(step, channel_names)
3310def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3311 channel = step.get("output_channel")
3312 if channel is None:
3313 return
3314 if not isinstance(channel, str) or not channel:
3315 raise RunHardFailure(
3316 "CHANNEL_BINDING_ERROR",
3317 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3318 f"string; got {channel!r}",
3319 )
3320 if channel not in channel_names:
3321 raise RunHardFailure(
3322 "CHANNEL_BINDING_ERROR",
3323 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3324 "which is not declared in spec.channels",
3325 )
3328def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3329 channels = step.get("input_channels")
3330 if channels is None:
3331 return
3332 if not isinstance(channels, list):
3333 raise RunHardFailure(
3334 "CHANNEL_BINDING_ERROR",
3335 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3336 f"channel names; got {type(channels).__name__}",
3337 )
3338 for channel in channels:
3339 if not isinstance(channel, str) or not channel:
3340 raise RunHardFailure(
3341 "CHANNEL_BINDING_ERROR",
3342 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3343 f"channel name '{channel}'; channel names must be non-empty "
3344 "strings",
3345 )
3346 if channel not in channel_names:
3347 raise RunHardFailure(
3348 "CHANNEL_BINDING_ERROR",
3349 f"Step '{step.get('id', '?')}' reads undeclared channel "
3350 f"'{channel}'; the channel must be declared in spec.channels",
3351 )
3354def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3355 """Reject duplicate and engine/step-reserved channel names."""
3356 forbidden = (
3357 {_CORE}
3358 | set(_RESERVED_NAMESPACES)
3359 | _CORE_STATE_KEYS
3360 | _collect_composed_step_ids(steps)
3361 )
3362 seen: set[str] = set()
3363 for channel in channels:
3364 name = channel.get("name")
3365 if name in seen:
3366 raise RunHardFailure(
3367 "CHANNEL_BINDING_ERROR",
3368 f"Pipeline declares channel '{name}' more than once; channel "
3369 "names must be unique",
3370 )
3371 seen.add(name)
3372 if name in forbidden:
3373 raise RunHardFailure(
3374 "CHANNEL_BINDING_ERROR",
3375 f"Pipeline channel '{name}' collides with an engine-reserved "
3376 "name (a step id, engine state key, reserved namespace, or the "
3377 "core state channel); channel names must be unique and "
3378 "non-reserved",
3379 )
3382def _iter_step_lists(steps: list[dict]):
3383 """Yield *steps* and every nested ``body`` list, recursively."""
3384 yield steps
3385 for step in steps:
3386 body = step.get("body")
3387 if isinstance(body, list):
3388 yield from _iter_step_lists(body)
3391def _validate_unique_step_ids(steps: list[dict]) -> None:
3392 """Reject duplicate step ``id``s within the same step list.
3394 Scoped per list: the top-level ``steps`` array and each nested ``body``
3395 array are checked independently, so an id may repeat across a body and its
3396 parent without being flagged. Without this guard a duplicated id is
3397 silently collapsed (last-wins), masking a manifest bug.
3398 """
3399 for step_list in _iter_step_lists(steps):
3400 seen: set[str] = set()
3401 for step in step_list:
3402 step_id = step.get("id")
3403 if not isinstance(step_id, str):
3404 continue
3405 if step_id in seen:
3406 raise RunHardFailure(
3407 "VALIDATION_ERROR",
3408 f"duplicate step id '{step_id}' in the same step list; "
3409 "step ids must be unique within their list",
3410 {"step_id": step_id},
3411 )
3412 seen.add(step_id)
3415def _validate_foreach_bodies(steps: list[dict]) -> None:
3416 """Reject a ``foreach`` with neither a non-empty body nor an agent.
3418 A body-less, agent-less ``foreach`` has no per-item work: the sequential
3419 composition builds an empty agent node and fails at runtime, while the
3420 concurrent paths silently produce an empty aggregate. Reject it at build
3421 time instead.
3422 """
3423 for step in _iter_steps(steps):
3424 if step.get("type") != "foreach":
3425 continue
3426 if step.get("body") or step.get("agent"):
3427 continue
3428 step_id = step.get("id", "?")
3429 raise RunHardFailure(
3430 "VALIDATION_ERROR",
3431 f"foreach step '{step_id}' declares neither a non-empty body nor "
3432 "an agent; per-item work requires one of them",
3433 {"step_id": step_id},
3434 )
3437def _validate_loop_limits(steps: list[dict], build_ctx: dict) -> None:
3438 """Strictly resolve a present ``max_iterations`` on every ``foreach``/``while``.
3440 A concurrent ``foreach`` is not composed into the sequential cycle, so its
3441 ``max_iterations`` would otherwise never be build-validated and a
3442 present-but-unresolvable value would silently degrade to "no cap". This pass
3443 makes the ADR-0012 contract uniform regardless of the composition path.
3444 """
3445 for step in _iter_steps(steps):
3446 if step.get("type") not in ("foreach", "while"):
3447 continue
3448 max_iter_raw = step.get("max_iterations")
3449 if max_iter_raw is None:
3450 continue
3451 _resolve_int_strict(
3452 max_iter_raw, build_ctx, "max_iterations", step.get("id", "?")
3453 )
3456def _validate_fail_if_exhausted(step_id: str, value: Any) -> None:
3457 """Reject a ``fail_if_exhausted`` value that is neither a bool nor a string.
3459 The field is a runtime expression: a bare boolean or a template/condition
3460 string. A list/object/number would otherwise reach ``_eval_condition``,
3461 whose fallback ``bool(resolved)`` silently treats a non-empty list as true.
3462 """
3463 if value is None or isinstance(value, (bool, str)):
3464 return
3465 raise RunHardFailure(
3466 "VALIDATION_ERROR",
3467 f"step '{step_id}' field 'fail_if_exhausted' must be a boolean or a "
3468 f"string; got {type(value).__name__}",
3469 {"step_id": step_id, "field": "fail_if_exhausted"},
3470 )
3473def _validate_fail_if_exhausted_steps(steps: list[dict]) -> None:
3474 """Reject malformed ``fail_if_exhausted`` values across every step list."""
3475 for step in _iter_steps(steps):
3476 _validate_fail_if_exhausted(step.get("id", "?"), step.get("fail_if_exhausted"))
3479def _input_path_root(value: str) -> str | None:
3480 """Leading state key of a bare dot-path input string, or None.
3482 Template strings, bracket-prefixed verbatim literals, and blank strings
3483 are not state reads and yield None.
3484 """
3485 if "{{" in value:
3486 return None
3487 stripped = value.lstrip()
3488 if not stripped or stripped.startswith("["):
3489 return None
3490 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3493def _iter_input_roots(input_spec: Any):
3494 """Yield every bare dot-path root in an input spec, recursing into containers."""
3495 if isinstance(input_spec, str):
3496 root = _input_path_root(input_spec)
3497 if root:
3498 yield root
3499 elif isinstance(input_spec, dict):
3500 for value in input_spec.values():
3501 yield from _iter_input_roots(value)
3502 elif isinstance(input_spec, list):
3503 for value in input_spec:
3504 yield from _iter_input_roots(value)
3507def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3508 """Reject channel-pipeline input roots that name nothing the state exposes.
3510 The channel state is the declared channels plus the engine state, so a bare
3511 dot-path input root must be a declared channel, a known step id, a reserved
3512 namespace, or an engine state key. Anything else is a hard build error —
3513 there is no silent fallback to a missing key.
3514 """
3515 allowed = (
3516 {ch["name"] for ch in channels}
3517 | _collect_step_ids(steps)
3518 | set(_RESERVED_NAMESPACES)
3519 | _CORE_STATE_KEYS
3520 )
3521 for step in _iter_steps(steps):
3522 for root in _iter_input_roots(step.get("input")):
3523 if root not in allowed:
3524 raise RunHardFailure(
3525 "CHANNEL_BINDING_ERROR",
3526 f"Pipeline channel step '{step.get('id', '?')}' input "
3527 f"references unknown root '{root}'; in a channel pipeline "
3528 "an input root must be a declared channel, a step id, a "
3529 "reserved namespace, or an engine state key",
3530 )
3533def _flatten_channel_state(state: dict) -> dict:
3534 """Read the flat committed snapshot nested under the Core namespace."""
3535 return dict(state.get(_CORE) or {})
3538def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3539 """Drop channel-named keys so a node emits only the core state."""
3540 return {k: v for k, v in flat.items() if k not in channel_names}
3543def _project_persisted_step_result(core: dict, step: dict) -> None:
3544 """Strip *step*'s persisted result of its envelope ``content``, in place.
3546 Channel routing reads the in-flight result before this call, so the routed
3547 value is unaffected; only the materialized copy at ``state[step_id]`` loses
3548 its ``content`` payload (ADR-0023 D9 / ADR-0043 D6). Other engine-read
3549 fields (``error``, ``parse_error``, body-item ``verdict``) are retained.
3550 """
3551 step_id = step.get("id")
3552 if step_id is not None and step_id in core:
3553 core[step_id] = project_envelope_metadata(core[step_id])
3556_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3558_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3561def _envelope_content(result: Any) -> Any:
3562 """Return an envelope's optional ``content`` field (ADR-0023 D3).
3564 ``content`` is opaque and optional: a non-mapping result, an absent field,
3565 and an explicit ``null`` all read as ``None``. Callers apply their own
3566 fallback and shape checks.
3567 """
3568 if isinstance(result, dict):
3569 return result.get("content")
3570 return None
3573def _is_failure_envelope(value: Any) -> bool:
3574 """True when a step output is not a successfully parsed dict.
3576 Missing output, non-dict values, and the ``error``/``parse_error``
3577 envelopes written by the LLM failure paths are not routed. A valid
3578 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3579 it carries an explicit ``"error": null``.
3580 """
3581 if not isinstance(value, dict):
3582 return True
3583 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3584 return False
3585 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3588def _resolve_declared_channel(
3589 step: dict,
3590 channel_names: frozenset[str],
3591 *,
3592 allow_default: bool,
3593) -> str | None:
3594 """Resolve a step's single declared output channel, or ``None`` to route nothing.
3596 A step without an ``output_channel`` resolves to the default ``lastvalue``
3597 channel when one exists **and** ``allow_default`` is set; an
3598 ``output_channel`` naming a channel absent from ``spec.channels`` is a hard
3599 error ``CHANNEL_BINDING_ERROR``.
3601 ``allow_default`` is set only on sequential, single-writer execution. The
3602 default channel is not routable inside a superstep (parallel/foreach
3603 fan-out) because its binding resolves only at the superstep boundary: every
3604 superstep staging path keeps the default ``False``. Both the implicit route
3605 and an explicit ``output_channel: lastvalue`` are therefore unrouted there —
3606 ``None`` is returned and a non-fatal advisory is logged (ADR-0040 D5).
3607 """
3608 channel = step.get("output_channel")
3609 if not channel:
3610 if not allow_default:
3611 log.warning(
3612 "[channels] step %s has no output_channel; superstep output is "
3613 "unrouted (advisory, ADR-0040 D5)",
3614 step.get("id", "?"),
3615 )
3616 return None
3617 if "lastvalue" in channel_names:
3618 return "lastvalue"
3619 return None
3620 if channel not in channel_names:
3621 raise RunHardFailure(
3622 "CHANNEL_BINDING_ERROR",
3623 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3624 "which is not declared in spec.channels",
3625 )
3626 if channel == "lastvalue" and not allow_default:
3627 log.warning(
3628 "[channels] step %s declares output_channel 'lastvalue'; superstep "
3629 "output is unrouted (advisory, ADR-0040 D5)",
3630 step.get("id", "?"),
3631 )
3632 return None
3633 return channel
3636def _declared_output_channel_writes(
3637 step: dict,
3638 output: Any,
3639 channel_names: frozenset[str],
3640 *,
3641 allow_default: bool = False,
3642) -> dict[str, Any]:
3643 """Route one agent step output's ``content`` to its declared output channel.
3645 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3646 the working set. Failure envelopes are not routed.
3647 """
3648 if _is_failure_envelope(output):
3649 return {}
3650 channel = _resolve_declared_channel(
3651 step, channel_names, allow_default=allow_default
3652 )
3653 if channel is None:
3654 return {}
3655 return {channel: _envelope_content(output)}
3658def _builtin_output_channel_writes(
3659 step: dict,
3660 output: Any,
3661 channel_names: frozenset[str],
3662 *,
3663 allow_default: bool = False,
3664) -> dict[str, Any]:
3665 """Route one agent-less step's result payload to its declared output channel.
3667 A platform builtin / ``kind: StepFunction`` step stores an arbitrary payload
3668 at ``state[step_id]`` with no envelope, so the routed value is the payload
3669 itself, not its ``content`` field. Builtin failures are raised during
3670 execution, so they are not classified here.
3671 """
3672 channel = _resolve_declared_channel(
3673 step, channel_names, allow_default=allow_default
3674 )
3675 if channel is None:
3676 return {}
3677 return {channel: output}
3680def _route_declared_channels(
3681 step: dict,
3682 pre_state: dict,
3683 core: dict,
3684 step_runner: Any,
3685 channel_names: frozenset[str],
3686 *,
3687 allow_default: bool = False,
3688) -> dict[str, Any]:
3689 """Route a step's declared output channel write to its channel.
3691 Returns ``{channel_name: value}`` for LangGraph to merge through the
3692 channel's reducer, where ``channel_name`` is the step's single declared
3693 ``output_channel``. An actual agent invocation routes the envelope's
3694 ``content``. A ``skip_llm_if`` passthrough routes its resolved input through
3695 the same builtin route as an agent-less step (the payload itself); a parallel
3696 fan-out envelope stays excluded (its per-item outputs are staged elsewhere).
3697 An agent-less step (platform builtin or ``kind: StepFunction``) routes its
3698 result payload itself. A step without an ``output_channel`` routes to the
3699 default ``lastvalue`` channel when one exists **and** ``allow_default`` is
3700 set (otherwise it routes nothing and its output stays in ``core`` untouched).
3702 ``allow_default`` must stay unset on every superstep staging path: a
3703 ``LastValue`` default is forbidden as an in-superstep output target because
3704 its binding resolves only at the superstep boundary (ADR-0040 D4, ADR-0043
3705 D9/D10).
3706 """
3707 step_id = step.get("id", "?")
3708 if step_id not in core:
3709 return {}
3710 output = core[step_id]
3711 if step_runner.resolve_agent(step) is not None:
3712 if step_runner.ran_agent(step, pre_state):
3713 return _declared_output_channel_writes(
3714 step, output, channel_names, allow_default=allow_default
3715 )
3716 if not step_runner._should_skip_llm(
3717 step, step_runner._make_ctx(pre_state)
3718 ):
3719 return {}
3720 return _builtin_output_channel_writes(
3721 step, output, channel_names, allow_default=allow_default
3722 )
3725def _make_step_node(
3726 step: dict,
3727 step_runner: Any,
3728 channel_names: frozenset[str],
3729 channel_specs: dict[str, ChannelSpec],
3730) -> Any:
3731 def node_fn(state: PipelineState) -> PipelineState:
3732 committed = _flatten_channel_state(state)
3733 working_set = ChannelWorkingSet(committed, channel_specs)
3734 core = _strip_channel_keys(
3735 step_runner.run_step(step, committed, working_set), channel_names
3736 )
3737 writes = _route_declared_channels(
3738 step, committed, core, step_runner, channel_names,
3739 allow_default=True,
3740 )
3741 _project_persisted_step_result(core, step)
3742 for name, value in writes.items():
3743 working_set.stage(name, value)
3744 written_names = {name for name, _ in working_set.pending_writes()}
3745 channels = working_set.commit()
3746 emitted = {name: channels[name] for name in written_names}
3747 return {_CORE: {**core, **emitted}}
3748 return node_fn
3751def _filter_composed_step_ids(result: dict, composed_step_ids: Any) -> dict:
3752 """Drop function-body step ids so a call subgraph's internals stay invisible.
3754 ADR-0037 D3: the caller sees only ``state[<call_id>].outputs`` plus
3755 unrelated keys. ``composed_step_ids`` is the build-time set recorded on the
3756 call placeholder. Channel-named keys are disjoint from it (a collision is a
3757 build-time ``CHANNEL_BINDING_ERROR``), so channel state is never touched.
3758 The input is left unmodified.
3759 """
3760 if not composed_step_ids:
3761 return dict(result)
3762 filtered = {
3763 key: value
3764 for key, value in result.items()
3765 if key not in composed_step_ids
3766 }
3767 core = filtered.get(_CORE)
3768 if isinstance(core, dict):
3769 filtered[_CORE] = {
3770 key: value
3771 for key, value in core.items()
3772 if key not in composed_step_ids
3773 }
3774 return filtered
3777def _make_subgraph_node(
3778 compiled: Any,
3779 condition: Any,
3780 step_runner: Any,
3781 channel_names: frozenset[str],
3782 composed_step_ids: Any = None,
3783) -> Any:
3784 """Wrap a compiled call subgraph with the call step's entry condition.
3786 A false condition returns no state updates, so the subgraph never runs and
3787 ``state[<call_id>]`` is not written — matching the soft-skip semantics of a
3788 normal step condition. The ctx is derived exactly like ``run_step``'s,
3789 including channel-mode flattening. Every successful invocation is filtered
3790 through ``_filter_composed_step_ids`` so the function body stays invisible.
3791 """
3792 if condition is None:
3793 def node_fn(state: PipelineState) -> PipelineState:
3794 return _filter_composed_step_ids(
3795 compiled.invoke(state), composed_step_ids
3796 )
3797 return node_fn
3799 def node_fn(state: PipelineState) -> PipelineState:
3800 committed = _flatten_channel_state(state)
3801 if not _eval_condition(condition, step_runner._make_ctx(committed)):
3802 return {}
3803 return _filter_composed_step_ids(
3804 compiled.invoke(state), composed_step_ids
3805 )
3806 return node_fn
3809# ---------------------------------------------------------------------------
3810# Native `while` cycle nodes (ADR-0025 D1/D2)
3811# ---------------------------------------------------------------------------
3812#
3813# A `while` step is a compiled cycle: `head -> body... -> cond` with a
3814# conditional back-edge from `cond` to `head` (continue) or `exit`. The head
3815# owns entry-condition handling and the initial `current`; the cond node owns
3816# the feed, the iteration increment and the exit-condition evaluation; the exit
3817# node writes the documented result and enforces `fail_if_exhausted`. The
3818# nodes read/write the shared graph state, using the `_CORE` container when the
3819# pipeline declares channels.
3822def _while_marker_key(step_id: str) -> str:
3823 """Per-while engine marker key (nested whiles live in isolated cycles)."""
3824 return f"{_WHILE_STATE_KEY}__{step_id}"
3827def _while_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
3828 return _flatten_channel_state(state)
3831def _while_update(update: dict, channel_names: frozenset[str]) -> dict:
3832 return {_CORE: update}
3835def _while_invoke_config(step: dict) -> dict | None:
3836 """Recursion bound for a while/foreach cycle placeholder, else None."""
3837 if step.get(_SUBGRAPH_WHILE_KEY):
3838 return {"recursion_limit": step.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
3839 if step.get(_SUBGRAPH_FOREACH_KEY):
3840 return {"recursion_limit": step.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
3841 return None
3844def _make_while_body_node(
3845 sub: dict,
3846 step_runner: Any,
3847 channel_names: frozenset[str],
3848 channel_specs: dict[str, ChannelSpec],
3849) -> Any:
3850 """A while body step as a node: a nested subgraph or a normal step node."""
3851 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
3852 if sub.get(_SUBGRAPH_WHILE_KEY):
3853 return _make_while_subgraph_node(sub, step_runner, channel_names)
3854 if sub.get(_SUBGRAPH_FOREACH_KEY):
3855 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
3856 return _make_subgraph_node(
3857 sub[_SUBGRAPH_GRAPH_KEY],
3858 sub.get("condition"),
3859 step_runner,
3860 channel_names,
3861 sub.get(_SUBGRAPH_STEP_IDS_KEY),
3862 )
3863 return _make_step_node(sub, step_runner, channel_names, channel_specs)
3866def _make_while_head_node(
3867 step: dict,
3868 step_runner: Any,
3869 channel_names: frozenset[str],
3870 max_iter: int,
3871) -> Any:
3872 """Entry node: handle the entry condition and seed the loop marker."""
3873 step_id = step.get("id", "?")
3874 marker_key = _while_marker_key(step_id)
3876 def node_fn(state: PipelineState) -> PipelineState:
3877 flat = _while_flat(state, channel_names)
3878 marker = flat.get(marker_key)
3879 if marker is not None:
3880 return _while_update({}, channel_names)
3881 ctx = step_runner._make_ctx(flat)
3882 entry_cond = step.get("condition")
3883 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
3884 skip_marker = {
3885 "current": None,
3886 "iteration": 0,
3887 "skipped": True,
3888 "max_iter": max_iter,
3889 }
3890 return _while_update(
3891 {
3892 step_id: {
3893 "iterations": 0,
3894 "output": _resolved_while_input(step, flat, ctx),
3895 "exhausted": False,
3896 "skipped": True,
3897 },
3898 marker_key: skip_marker,
3899 },
3900 channel_names,
3901 )
3902 marker = {
3903 "current": _resolved_while_input(step, flat, ctx),
3904 "iteration": 0,
3905 "max_iter": max_iter,
3906 }
3907 return _while_update({marker_key: marker}, channel_names)
3909 return node_fn
3912def _resolved_while_input(step: dict, flat: dict, ctx: dict) -> Any:
3913 input_spec = step.get("input")
3914 if input_spec is None:
3915 return None
3916 return _resolve_input(input_spec, flat, ctx)
3919def _make_while_head_router(
3920 step: dict,
3921 channel_names: frozenset[str],
3922 first_id: str,
3923 exit_id: str,
3924 end: Any,
3925) -> Any:
3926 """Route from the head: skipped → end, bound already reached → exit, else body."""
3927 marker_key = _while_marker_key(step.get("id", "?"))
3929 def router(state: PipelineState) -> str:
3930 marker = _while_flat(state, channel_names).get(marker_key) or {}
3931 if marker.get("skipped"):
3932 return end
3933 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
3934 return exit_id
3935 return first_id
3937 return router
3940def _make_while_cond_node(
3941 step: dict,
3942 step_runner: Any,
3943 channel_names: frozenset[str],
3944) -> Any:
3945 """Increment the iteration and evaluate the exit condition."""
3946 step_id = step.get("id", "?")
3947 marker_key = _while_marker_key(step_id)
3948 exit_condition = step.get("exit_condition")
3950 def node_fn(state: PipelineState) -> PipelineState:
3951 flat = _while_flat(state, channel_names)
3952 marker = dict(flat.get(marker_key) or {})
3953 passed = False
3954 if exit_condition:
3955 if _eval_condition(
3956 exit_condition, step_runner._make_ctx(flat)
3957 ):
3958 passed = True
3959 else:
3960 passed = True
3961 marker.update(
3962 {
3963 "iteration": marker.get("iteration", 0) + 1,
3964 "passed": passed,
3965 }
3966 )
3967 return _while_update({marker_key: marker}, channel_names)
3969 return node_fn
3972def _make_while_cond_router(
3973 step: dict,
3974 channel_names: frozenset[str],
3975 head_id: str,
3976 exit_id: str,
3977) -> Any:
3978 """Route from the cond: exit on pass or when the bound is reached, else head."""
3979 marker_key = _while_marker_key(step.get("id", "?"))
3981 def router(state: PipelineState) -> str:
3982 marker = _while_flat(state, channel_names).get(marker_key) or {}
3983 if marker.get("passed"):
3984 return exit_id
3985 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
3986 return exit_id
3987 return head_id
3989 return router
3992def _while_result_output(step: dict, flat: dict, marker: dict) -> Any:
3993 """The loop result's `output`: the declared body step's result, else the input.
3995 `while.output: <body_step_id>` names the body step whose result fills the
3996 loop result's `output` field; it does not feed the next round. Without a
3997 declared output the loop's resolved input (the marker current) is surfaced.
3998 """
3999 declared_output = step.get("output")
4000 if not declared_output:
4001 return marker.get("current")
4002 value = flat.get(declared_output)
4003 if isinstance(value, dict) and "output" in value:
4004 return value["output"]
4005 if isinstance(value, dict) and isinstance(value.get("results"), list):
4006 return value["results"]
4007 return value
4010def _make_while_exit_node(
4011 step: dict,
4012 step_runner: Any,
4013 channel_names: frozenset[str],
4014) -> Any:
4015 """Write the documented while result and enforce `fail_if_exhausted`."""
4016 step_id = step.get("id", "?")
4017 marker_key = _while_marker_key(step_id)
4018 fail_if_exhausted = step.get("fail_if_exhausted")
4019 _validate_fail_if_exhausted(step_id, fail_if_exhausted)
4021 def node_fn(state: PipelineState) -> PipelineState:
4022 flat = _while_flat(state, channel_names)
4023 marker = flat.get(marker_key) or {}
4024 passed = bool(marker.get("passed", False))
4025 exhausted = not passed
4026 if (
4027 fail_if_exhausted is not None
4028 and exhausted
4029 and _eval_condition(
4030 fail_if_exhausted, step_runner._make_ctx(flat)
4031 )
4032 ):
4033 raise RunHardFailure(
4034 "EXHAUSTED",
4035 f"while step '{step_id}' exhausted "
4036 f"max_iterations={marker.get('max_iter')} without "
4037 "exit_condition passing (fail_if_exhausted)",
4038 {"step_id": step_id, "exhausted": True},
4039 )
4040 return _while_update(
4041 {
4042 step_id: {
4043 "iterations": marker.get("iteration", 0),
4044 "output": _while_result_output(step, flat, marker),
4045 "exhausted": exhausted,
4046 "skipped": False,
4047 }
4048 },
4049 channel_names,
4050 )
4052 return node_fn
4055def _make_while_subgraph_node(
4056 placeholder: dict,
4057 step_runner: Any,
4058 channel_names: frozenset[str],
4059) -> Any:
4060 """Parent node for a compiled `while` cycle: isolate and surface its result.
4062 The cycle shares the graph state schema, so invoking it directly would leak
4063 the body step results and loop plumbing into the parent. This wrapper runs
4064 the cycle on a copy and returns only the documented result plus the side
4065 channels and any channel values the body committed, preserving the in-node
4066 loop's state contract exactly.
4067 """
4068 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4069 step_id = placeholder["id"]
4070 config = {"recursion_limit": placeholder.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
4072 def node_fn(state: PipelineState) -> PipelineState:
4073 sub = compiled.invoke(state, config)
4074 flat = _flatten_channel_state(sub)
4075 parent_flat = _flatten_channel_state(state)
4076 result: dict = {}
4077 if step_id in flat:
4078 result[step_id] = flat.get(step_id)
4079 for key in _WHILE_SIDE_CHANNEL_KEYS:
4080 if key in flat and flat.get(key) != parent_flat.get(key):
4081 result[key] = flat.get(key)
4082 for name in channel_names:
4083 if name in flat and (
4084 name not in parent_flat or flat.get(name) != parent_flat.get(name)
4085 ):
4086 result[name] = flat.get(name)
4087 if not result:
4088 return {}
4089 return {_CORE: result}
4091 return node_fn
4094# ---------------------------------------------------------------------------
4095# Native sequential `foreach` cycle nodes
4096# ---------------------------------------------------------------------------
4097#
4098# A sequential `foreach` is a compiled cycle: `head -> (body... | agent) ->
4099# cond` with a conditional back-edge from `cond` to `head` (next item) or
4100# `exit` (`max_output` reached or the item list exhausted). The head resolves
4101# the item list once and injects the current item (`item` / `_item` /
4102# `_foreach_item`); the cond collects the item result; the exit node writes the
4103# documented aggregate (`{"results", "passed", "failing", "failed"}` for the
4104# body path, the plain list for the agent path). Nodes read/write the shared
4105# graph state, using the `_CORE` container when the pipeline declares channels.
4108def _foreach_marker_key(step_id: str) -> str:
4109 """Per-foreach engine marker key (nested foreachs live in isolated cycles)."""
4110 return f"{_FOREACH_STATE_KEY}__{step_id}"
4113def _foreach_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
4114 return _flatten_channel_state(state)
4117def _foreach_update(update: dict, channel_names: frozenset[str]) -> dict:
4118 return {_CORE: update}
4121def _make_foreach_head_node(
4122 step: dict,
4123 step_runner: Any,
4124 channel_names: frozenset[str],
4125) -> Any:
4126 """Entry node: evaluate the entry condition, resolve items, inject the item."""
4127 step_id = step.get("id", "?")
4128 marker_key = _foreach_marker_key(step_id)
4130 def node_fn(state: PipelineState) -> PipelineState:
4131 flat = _foreach_flat(state, channel_names)
4132 marker = flat.get(marker_key)
4133 if marker is None:
4134 ctx = step_runner._make_ctx(flat)
4135 entry_cond = step.get("condition")
4136 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4137 return _foreach_update({}, channel_names)
4138 max_output_raw = step.get("max_output")
4139 marker = {
4140 "items": step_runner._foreach_item_list(step, ctx),
4141 "index": 0,
4142 "results": [],
4143 "produced_outputs": 0,
4144 "max_output": (
4145 _resolve_int(max_output_raw, ctx, 0)
4146 if max_output_raw is not None
4147 else 0
4148 ),
4149 }
4150 items = marker.get("items") or []
4151 index = marker.get("index", 0)
4152 if index < len(items):
4153 marker = {**marker, "publish_before": flat.get("publish")}
4154 update = {
4155 "_item": items[index],
4156 "item": items[index],
4157 "_foreach_item": items[index],
4158 }
4159 update[marker_key] = marker
4160 return _foreach_update(update, channel_names)
4161 return _foreach_update({marker_key: marker}, channel_names)
4163 return node_fn
4166def _make_foreach_head_router(
4167 step: dict,
4168 channel_names: frozenset[str],
4169 first_id: str,
4170 exit_id: str,
4171 end: Any,
4172) -> Any:
4173 """Route from the head: condition-skipped → end, exhausted/budget → exit, else body."""
4174 marker_key = _foreach_marker_key(step.get("id", "?"))
4176 def router(state: PipelineState) -> str:
4177 marker = _foreach_flat(state, channel_names).get(marker_key)
4178 if marker is None:
4179 return end
4180 items = marker.get("items") or []
4181 if marker.get("index", 0) >= len(items):
4182 return exit_id
4183 max_output = marker.get("max_output", 0)
4184 if max_output and marker.get("produced_outputs", 0) >= max_output:
4185 return exit_id
4186 return first_id
4188 return router
4191def _make_foreach_body_node(
4192 sub: dict,
4193 step_runner: Any,
4194 channel_names: frozenset[str],
4195 channel_specs: dict[str, ChannelSpec],
4196) -> Any:
4197 """A foreach body step as a node: a nested loop subgraph or a step node."""
4198 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
4199 if sub.get(_SUBGRAPH_WHILE_KEY):
4200 return _make_while_subgraph_node(sub, step_runner, channel_names)
4201 if sub.get(_SUBGRAPH_FOREACH_KEY):
4202 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
4203 return _make_subgraph_node(
4204 sub[_SUBGRAPH_GRAPH_KEY],
4205 sub.get("condition"),
4206 step_runner,
4207 channel_names,
4208 sub.get(_SUBGRAPH_STEP_IDS_KEY),
4209 )
4210 return _make_step_node(sub, step_runner, channel_names, channel_specs)
4213def _make_foreach_agent_node(
4214 step: dict,
4215 step_runner: Any,
4216 channel_names: frozenset[str],
4217 channel_specs: dict[str, ChannelSpec],
4218) -> Any:
4219 """Agent-path node: invoke the foreach step's agent for the current item."""
4220 step_id = step.get("id", "?")
4222 def node_fn(state: PipelineState) -> PipelineState:
4223 flat = _foreach_flat(state, channel_names)
4224 resolved = step_runner.resolve_agent(step)
4225 ctx = step_runner._make_ctx(flat)
4226 resolved_step = step_runner._resolve_step_runtime(step, flat, ctx)
4227 resolved_step = step_runner._attach_communication_context(resolved_step, flat)
4228 resolved_step = step_runner._attach_resolved_tools(resolved_step, resolved)
4229 result_state = step_runner._invoke_agent(resolved, resolved_step, flat)
4230 core = _strip_channel_keys(result_state, channel_names)
4231 _project_persisted_step_result(core, step)
4232 return {_CORE: core}
4234 return node_fn
4237def _make_foreach_cond_node(
4238 step: dict,
4239 step_runner: Any,
4240 channel_names: frozenset[str],
4241) -> Any:
4242 """Collect the current item's result and advance the item index."""
4243 step_id = step.get("id", "?")
4244 marker_key = _foreach_marker_key(step_id)
4245 declared_output = step.get("output")
4246 runs_body = bool(step.get("body"))
4248 def node_fn(state: PipelineState) -> PipelineState:
4249 flat = _foreach_flat(state, channel_names)
4250 marker = dict(flat.get(marker_key) or {})
4251 results = list(marker.get("results") or [])
4252 produced = marker.get("produced_outputs", 0)
4253 if runs_body:
4254 results.append(StepRunner._body_item_result(flat, declared_output))
4255 publish_out = flat.get("publish")
4256 if (
4257 isinstance(publish_out, dict)
4258 and publish_out.get("path")
4259 and publish_out != marker.get("publish_before")
4260 ):
4261 produced += 1
4262 else:
4263 results.append(flat.get(step_id))
4264 marker.update(
4265 {
4266 "results": results,
4267 "index": marker.get("index", 0) + 1,
4268 "produced_outputs": produced,
4269 }
4270 )
4271 return _foreach_update({marker_key: marker}, channel_names)
4273 return node_fn
4276def _make_foreach_cond_router(
4277 step: dict,
4278 channel_names: frozenset[str],
4279 head_id: str,
4280 exit_id: str,
4281) -> Any:
4282 """Route from the cond: exit on exhaustion or budget, else back to the head."""
4283 marker_key = _foreach_marker_key(step.get("id", "?"))
4285 def router(state: PipelineState) -> str:
4286 marker = _foreach_flat(state, channel_names).get(marker_key) or {}
4287 items = marker.get("items") or []
4288 if marker.get("index", 0) >= len(items):
4289 return exit_id
4290 max_output = marker.get("max_output", 0)
4291 if max_output and marker.get("produced_outputs", 0) >= max_output:
4292 return exit_id
4293 return head_id
4295 return router
4298def _make_foreach_exit_node(
4299 step: dict,
4300 step_runner: Any,
4301 channel_names: frozenset[str],
4302) -> Any:
4303 """Write the documented foreach aggregate result."""
4304 step_id = step.get("id", "?")
4305 marker_key = _foreach_marker_key(step_id)
4306 runs_body = bool(step.get("body"))
4308 def node_fn(state: PipelineState) -> PipelineState:
4309 flat = _foreach_flat(state, channel_names)
4310 marker = flat.get(marker_key) or {}
4311 results = list(marker.get("results") or [])
4312 step_result = (
4313 StepRunner._aggregate_body_results(results)
4314 if runs_body
4315 else [project_envelope_metadata(r) for r in results]
4316 )
4317 return _foreach_update({step_id: step_result}, channel_names)
4319 return node_fn
4322def _make_foreach_subgraph_node(
4323 placeholder: dict,
4324 step_runner: Any,
4325 channel_names: frozenset[str],
4326) -> Any:
4327 """Parent node for a compiled sequential `foreach` cycle: isolate and surface.
4329 The cycle shares the graph state schema, so invoking it directly would leak
4330 body step results and loop plumbing into the parent. This wrapper runs the
4331 cycle on a copy and returns only the documented aggregate plus the side
4332 channels and any channel values the body committed. A condition-skipped
4333 entry writes no step result (matching the in-node soft-skip), so the step id
4334 is only surfaced when the cycle wrote it.
4335 """
4336 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4337 step_id = placeholder["id"]
4338 config = {"recursion_limit": placeholder.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
4340 def node_fn(state: PipelineState) -> PipelineState:
4341 sub = compiled.invoke(state, config)
4342 flat = _flatten_channel_state(sub)
4343 parent_flat = _flatten_channel_state(state)
4344 result: dict = {}
4345 if step_id in flat:
4346 result[step_id] = flat.get(step_id)
4347 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4348 if key in flat and flat.get(key) != parent_flat.get(key):
4349 result[key] = flat.get(key)
4350 for name in channel_names:
4351 if name in flat and (
4352 name not in parent_flat or flat.get(name) != parent_flat.get(name)
4353 ):
4354 result[name] = flat.get(name)
4355 if not result:
4356 return {}
4357 return {_CORE: result}
4359 return node_fn
4362def _collect_parallel_channel_writes(
4363 steps: list[dict],
4364 pre_state: dict,
4365 core: dict,
4366 step_runner: Any,
4367 channel_names: frozenset[str],
4368 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
4369) -> list[tuple[str, Any]]:
4370 """Collect one parallel node's declared writes in branch order.
4372 Every branch write is emitted, so fan-in feeds each channel's reducer every
4373 write in deterministic declaration/branch order: Topic appends, LastValue
4374 keeps the last, and BinaryOperatorAggregate routes each write through the
4375 reducer selected by name from its mandatory manifest ``operator`` — never
4376 inferred from the accumulated value's shape. A branch's isolated body
4377 writes are emitted before its own declared write, mirroring the top-level
4378 node's route-after-run order. This is a superstep fan-in path, so the
4379 implicit default channel is forbidden: ``allow_default`` stays unset on the
4380 shared router.
4381 """
4382 writes: list[tuple[str, Any]] = []
4383 for step in steps:
4384 if branch_working_sets is not None:
4385 branch_working_set = branch_working_sets.get(step["id"])
4386 if branch_working_set is not None:
4387 writes.extend(branch_working_set.pending_writes())
4388 branch = _route_declared_channels(
4389 step, pre_state, core, step_runner, channel_names
4390 )
4391 for channel_name, value in branch.items():
4392 writes.append((channel_name, value))
4393 return writes
4396def _make_parallel_node(
4397 group_steps: list[dict],
4398 step_runner: Any,
4399 channel_names: frozenset[str],
4400 channel_specs: dict[str, ChannelSpec],
4401) -> Any:
4402 def node_fn(state: PipelineState) -> PipelineState:
4403 base = _flatten_channel_state(state)
4404 working_set = ChannelWorkingSet(base, channel_specs)
4405 # Each branch writes into its own fork; the shared set is advanced only
4406 # at fan-in, so branch-internal body writes never race.
4407 branch_working_sets = {
4408 s["id"]: working_set.fork() for s in group_steps
4409 }
4411 def submit_branch(pool: Any, step: dict) -> Any:
4412 return pool.submit(
4413 step_runner.run_step, step, base, branch_working_sets[step["id"]]
4414 )
4416 results: dict[str, Any] = {}
4417 branch_metrics: dict = {}
4418 branch_tool_meta: dict = {}
4419 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
4420 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
4421 for future in as_completed(futures):
4422 sid = futures[future]
4423 try:
4424 out = future.result()
4425 results[sid] = out.get(sid)
4426 branch_metrics.update(out.get("step_context_metrics") or {})
4427 branch_tool_meta.update(out.get("step_tool_meta") or {})
4428 except SkillLoadError:
4429 raise
4430 except RunHardFailure:
4431 pool.shutdown(wait=False, cancel_futures=True)
4432 raise
4433 except Exception as exc: # noqa: BLE001
4434 log.error("Step '%s' failed: %s", sid, exc)
4435 results[sid] = {"error": str(exc)}
4436 merged = _strip_channel_keys({**base, **results}, channel_names)
4437 if branch_metrics:
4438 merged["step_context_metrics"] = _merge_side_channel(
4439 merged.get("step_context_metrics"), branch_metrics
4440 )
4441 if branch_tool_meta:
4442 merged["step_tool_meta"] = _merge_side_channel(
4443 merged.get("step_tool_meta"), branch_tool_meta
4444 )
4445 writes = _collect_parallel_channel_writes(
4446 group_steps, base, merged, step_runner, channel_names,
4447 branch_working_sets,
4448 )
4449 for branch_step in group_steps:
4450 _project_persisted_step_result(merged, branch_step)
4451 written_names = {name for name, _ in writes}
4452 for name, value in writes:
4453 working_set.stage(name, value)
4454 channels = working_set.commit()
4455 emitted = {name: channels[name] for name in written_names}
4456 return {_CORE: {**merged, **emitted}}
4457 return node_fn
4460@dataclass(frozen=True)
4461class _MapReduceParts:
4462 """The nodes and edges of one concurrent step's Send fan-out.
4464 ``dispatch_id`` reuses the step id so the step's incoming dependency edges are
4465 unchanged; ``join_id`` carries the step's outgoing edges.
4466 """
4468 dispatch_id: str
4469 worker_id: str
4470 fallback_id: str
4471 join_id: str
4472 dispatch_fn: Any
4473 route_fn: Any
4474 worker_fn: Any
4475 join_fn: Any
4478@dataclass(frozen=True)
4479class _MapReduceItemOutcome:
4480 """One worker's per-item result plus its errors, side channels and writes.
4482 ``errors`` carries the item's soft errors (and the inherited soft errors it
4483 read), so the join can fold them in item order exactly like the in-node
4484 concurrent path. ``channel_writes`` is empty for the parallel strategy (its
4485 channel writes are derived from the item result at the join) and carries the
4486 foreach body's pending writes so the join can stage them in item order.
4487 """
4489 result: Any
4490 metrics: dict | None
4491 tool_meta: dict | None
4492 channel_writes: tuple[tuple[str, Any], ...]
4493 errors: tuple[str, ...] = ()
4496@dataclass(frozen=True)
4497class _MapReducePlan:
4498 """The per-step behaviour injected into the shared Send machinery.
4500 The parallel strategy runs a single agent call per item; the foreach
4501 strategy runs the body (or agent) per item. ``item_lister`` resolves the
4502 fan-out list, ``item_runner`` executes one item, ``result_assembler``
4503 shapes the step result, and ``channel_write_stager`` stages per-item
4504 channel writes at the join.
4505 """
4507 item_lister: Any
4508 item_runner: Any
4509 result_assembler: Any
4510 channel_write_stager: Any
4513def _map_reduce_worker_payload(flat: dict, item: Any, index: int) -> dict:
4514 """Build a worker's ``Send`` payload from the resolved source state."""
4515 return {**flat, "item": item, "_item": item, _MAP_REDUCE_ITEM_INDEX: index}
4518def _map_reduce_worker_output(
4519 step_id: str,
4520 index: int,
4521 outcome: _MapReduceItemOutcome,
4522) -> PipelineState:
4523 """The accumulator record carrying every per-item fold input.
4525 The slot is ``{index: {"result", "writes", "errors", "metrics",
4526 "tool_meta"}}`` so the join can assemble the step result, stage channel
4527 writes, and fold soft errors plus side channels in item order — the same
4528 observable result as the in-node concurrent path.
4529 """
4530 record = {
4531 "result": outcome.result,
4532 "writes": list(outcome.channel_writes),
4533 "errors": list(outcome.errors),
4534 "metrics": outcome.metrics if isinstance(outcome.metrics, dict) else None,
4535 "tool_meta": (
4536 outcome.tool_meta if isinstance(outcome.tool_meta, dict) else None
4537 ),
4538 }
4539 return {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: {index: record}}}
4542def _make_map_reduce_route_fn(
4543 plan: _MapReducePlan,
4544 step_runner: Any,
4545 channel_names: frozenset[str],
4546 worker_id: str,
4547 fallback_id: str,
4548 join_id: str,
4549) -> Any:
4550 def route(state: PipelineState) -> list:
4551 flat = _flatten_channel_state(state)
4552 ctx = step_runner._make_ctx(flat)
4553 items = plan.item_lister(flat, ctx)
4554 if items is None:
4555 return [Send(fallback_id, state)]
4556 if not items:
4557 return [Send(join_id, state)]
4558 return [
4559 Send(worker_id, _map_reduce_worker_payload(flat, item, index))
4560 for index, item in enumerate(items)
4561 ]
4563 return route
4566def _make_map_reduce_worker_fn(
4567 step: dict,
4568 plan: _MapReducePlan,
4569 step_runner: Any,
4570 gate: threading.Semaphore,
4571) -> Any:
4572 step_id = step.get("id", "?")
4574 def worker(state: PipelineState) -> PipelineState:
4575 index = state[_MAP_REDUCE_ITEM_INDEX]
4576 item_state = {
4577 key: value
4578 for key, value in state.items()
4579 if key != _MAP_REDUCE_ITEM_INDEX
4580 }
4581 item = item_state.get("item")
4582 item_ctx = step_runner._make_ctx(item_state)
4583 with gate:
4584 try:
4585 outcome = plan.item_runner(
4586 state, index, item, item_state, item_ctx
4587 )
4588 except SkillLoadError:
4589 raise
4590 except RunHardFailure:
4591 raise
4592 except Exception as exc: # noqa: BLE001
4593 log.error("[step] %s item failed: %s", step_id, exc)
4594 outcome = _MapReduceItemOutcome(
4595 {"error": str(exc)}, None, None, ()
4596 )
4597 return _map_reduce_worker_output(step_id, index, outcome)
4599 return worker
4602def _fold_map_reduce_errors(records: list, base_errors: list) -> list:
4603 """Fold every item's soft errors into *base_errors* in item order."""
4604 merged = list(base_errors)
4605 for record in records:
4606 if not isinstance(record, dict):
4607 continue
4608 merged = StepRunner._merge_errors(merged, record.get("errors") or [])
4609 return merged
4612def _fold_map_reduce_side_channel(records: list, record_key: str) -> dict:
4613 """Fold every item's side-channel map in item order (later item wins)."""
4614 folded: dict = {}
4615 for record in records:
4616 if not isinstance(record, dict):
4617 continue
4618 value = record.get(record_key)
4619 if isinstance(value, dict):
4620 folded.update(value)
4621 return folded
4624def _fold_map_reduce_extras(flat: dict, records: list) -> dict:
4625 """The join's per-item state extras: soft errors plus folded side channels.
4627 Mirrors the in-node concurrent path: errors are merged in item order with
4628 dedup, and the budget/side-channel counters follow the same item order.
4629 """
4630 extras: dict = {}
4631 base_errors = list(flat.get("errors") or [])
4632 errors = _fold_map_reduce_errors(records, base_errors)
4633 if errors != base_errors:
4634 extras["errors"] = errors
4635 for state_key, record_key in (
4636 ("step_context_metrics", "metrics"),
4637 ("step_tool_meta", "tool_meta"),
4638 ):
4639 folded = _fold_map_reduce_side_channel(records, record_key)
4640 if folded:
4641 extras[state_key] = _merge_side_channel(flat.get(state_key), folded)
4642 return extras
4645def _make_map_reduce_join_fn(
4646 step: dict,
4647 plan: _MapReducePlan,
4648 step_runner: Any,
4649 channel_names: frozenset[str],
4650 channel_specs: dict[str, ChannelSpec],
4651) -> Any:
4652 step_id = step.get("id", "?")
4654 def join(state: PipelineState) -> PipelineState:
4655 flat = _flatten_channel_state(state)
4656 accumulated = state.get(_MAP_REDUCE_ACCUMULATOR_KEY) or {}
4657 entries = accumulated.get(step_id) or {}
4658 clear = {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: None}}
4659 ctx = step_runner._make_ctx(flat)
4660 items = plan.item_lister(flat, ctx)
4661 if items is None:
4662 return clear
4663 records = [entries.get(index) for index in range(len(items))]
4664 results = [
4665 record.get("result") if isinstance(record, dict) else None
4666 for record in records
4667 ]
4668 step_result = plan.result_assembler(results)
4669 extras = _fold_map_reduce_extras(flat, records)
4670 working_set = ChannelWorkingSet(flat, channel_specs)
4671 plan.channel_write_stager(results, records, working_set)
4672 written_names = {name for name, _ in working_set.pending_writes()}
4673 channels = working_set.commit()
4674 emitted = {name: channels[name] for name in written_names}
4675 core = _strip_channel_keys(
4676 {**flat, step_id: step_result, **extras}, channel_names
4677 )
4678 _project_persisted_step_result(core, step)
4679 return {_CORE: {**core, **emitted}, **clear}
4681 return join
4684def _make_parallel_map_reduce_plan(
4685 step: dict, step_runner: Any
4686) -> _MapReducePlan:
4687 """Plan for a concurrent `parallel` step: one agent call per item."""
4688 resolved_agent = step_runner.resolve_agent(step)
4690 def run_item(state, index, item, item_state, item_ctx):
4691 result, metrics, tool_meta, errors = step_runner._invoke_parallel_item(
4692 step, resolved_agent, item, item_state, item_ctx
4693 )
4694 return _MapReduceItemOutcome(
4695 result, metrics, tool_meta, (), tuple(errors)
4696 )
4698 return _MapReducePlan(
4699 item_lister=lambda _flat, ctx: step_runner._parallel_fanout_item_list(
4700 step, ctx
4701 ),
4702 item_runner=run_item,
4703 result_assembler=lambda results: project_aggregate_metadata(
4704 {"results": results}
4705 ),
4706 channel_write_stager=lambda results, records, working_set: (
4707 step_runner._stage_fanout_channel_writes(step, results, working_set)
4708 ),
4709 )
4712def _worker_committed_flat(
4713 state: PipelineState, channel_names: frozenset[str]
4714) -> dict:
4715 """The committed channel snapshot visible to one worker.
4717 A channels-path worker receives the flattened snapshot as its payload, so
4718 its channel keys sit at the top level; the typed ``__core__`` shape (only
4719 seen by the join) is unwrapped.
4720 """
4721 if _CORE in state:
4722 return _flatten_channel_state(state)
4723 return {name: state[name] for name in channel_names if name in state}
4726def _pending_channel_writes(
4727 working_set: ChannelWorkingSet,
4728) -> tuple[tuple[str, Any], ...]:
4729 return tuple(working_set.pending_writes())
4732def _stage_map_reduce_item_channel_writes(
4733 results: list, records: list, working_set: ChannelWorkingSet
4734) -> None:
4735 """Stage each item's carried channel writes in item order."""
4736 for record in records:
4737 if not isinstance(record, dict):
4738 continue
4739 for name, value in record.get("writes") or ():
4740 working_set.stage(name, value)
4743def _make_foreach_map_reduce_plan(
4744 step: dict,
4745 step_runner: Any,
4746 channel_names: frozenset[str],
4747 channel_specs: dict[str, ChannelSpec],
4748) -> _MapReducePlan:
4749 """Plan for a concurrent `foreach` step: the body (or agent) runs per item."""
4750 step_id = step.get("id", "?")
4751 body = step.get("body") or []
4752 declared_output = step.get("output")
4753 runs_body = bool(body)
4755 def result_assembler(results):
4756 if runs_body:
4757 return StepRunner._aggregate_body_results(results)
4758 return [project_envelope_metadata(r) for r in results]
4760 def run_item(state, index, item, item_state, _item_ctx):
4761 item_working_set = ChannelWorkingSet(
4762 _worker_committed_flat(state, channel_names), channel_specs
4763 )
4764 iter_state = {
4765 **item_state, "item": item, "_item": item, "_foreach_item": item
4766 }
4767 if runs_body:
4768 for sub_step in body:
4769 iter_state = step_runner._run_body_step(
4770 sub_step, iter_state, item_working_set
4771 )
4772 return _MapReduceItemOutcome(
4773 step_runner._body_item_result(iter_state, declared_output),
4774 iter_state.get("step_context_metrics"),
4775 iter_state.get("step_tool_meta"),
4776 _pending_channel_writes(item_working_set),
4777 tuple(iter_state.get("errors") or []),
4778 )
4779 resolved = step_runner.resolve_agent(step)
4780 item_read_state = step_runner._read_state(iter_state, item_working_set)
4781 item_ctx = step_runner._make_ctx(item_read_state)
4782 resolved_step = step_runner._resolve_step_runtime(
4783 step, item_read_state, item_ctx
4784 )
4785 resolved_step = step_runner._attach_communication_context(
4786 resolved_step, item_read_state
4787 )
4788 resolved_step = step_runner._attach_resolved_tools(
4789 resolved_step, resolved
4790 )
4791 result_state = step_runner._invoke_agent(
4792 resolved, resolved_step, iter_state
4793 )
4794 return _MapReduceItemOutcome(
4795 result_state.get(step_id),
4796 result_state.get("step_context_metrics"),
4797 result_state.get("step_tool_meta"),
4798 _pending_channel_writes(item_working_set),
4799 tuple(result_state.get("errors") or []),
4800 )
4802 return _MapReducePlan(
4803 item_lister=lambda _flat, ctx: step_runner._foreach_item_list(
4804 step, ctx
4805 ),
4806 item_runner=run_item,
4807 result_assembler=result_assembler,
4808 channel_write_stager=_stage_map_reduce_item_channel_writes,
4809 )
4812def _make_map_reduce_parts(
4813 step: dict,
4814 plan: _MapReducePlan,
4815 step_runner: Any,
4816 channel_names: frozenset[str],
4817 channel_specs: dict[str, ChannelSpec],
4818 max_concurrent: int,
4819) -> _MapReduceParts:
4820 step_id = step.get("id", "?")
4821 worker_id = f"{step_id}__worker"
4822 fallback_id = f"{step_id}__fallback"
4823 join_id = f"{step_id}__join"
4824 gate = threading.Semaphore(max(1, max_concurrent))
4825 return _MapReduceParts(
4826 dispatch_id=step_id,
4827 worker_id=worker_id,
4828 fallback_id=fallback_id,
4829 join_id=join_id,
4830 dispatch_fn=lambda state: {},
4831 route_fn=_make_map_reduce_route_fn(
4832 plan, step_runner, channel_names, worker_id, fallback_id, join_id
4833 ),
4834 worker_fn=_make_map_reduce_worker_fn(step, plan, step_runner, gate),
4835 join_fn=_make_map_reduce_join_fn(
4836 step, plan, step_runner, channel_names, channel_specs
4837 ),
4838 )
4841def _resolved_send_concurrency(step: dict, build_ctx: dict) -> int:
4842 """The build-time resolved ``max_concurrent`` for a Send-eligible step."""
4843 raw = step.get("max_concurrent")
4844 if raw is None:
4845 return 1
4846 return _resolve_int_strict(
4847 raw, build_ctx, "max_concurrent", step.get("id", "?"), minimum=1
4848 )
4851def _is_send_parallel_step(
4852 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4853) -> bool:
4854 """Whether a top-level `parallel` step fans out through native ``Send``.
4856 Restricted to a singleton group's statically concurrent, agent-backed,
4857 item-sourced step with no `routes`, no `condition`/`skip_llm_if`, and an
4858 explicit `max_concurrent > 1`; everything else keeps the in-node path.
4859 """
4860 if step.get("type") != "parallel":
4861 return False
4862 if step.get("id") in route_plan.target_node_ids:
4863 return False
4864 if step.get("routes") or not step.get("agent"):
4865 return False
4866 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
4867 return False
4868 if StepRunner._declared_item_source(step) is None:
4869 return False
4870 if step.get("max_concurrent") is None:
4871 return False
4872 return _resolved_send_concurrency(step, build_ctx) > 1
4875def _is_send_foreach_step(
4876 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4877) -> bool:
4878 """Whether a top-level `foreach` step fans out through native ``Send``.
4880 Restricted to a singleton group's body- or agent-backed, item-sourced step
4881 with no `routes`, no `condition`/`skip_llm_if`, no step-level
4882 `output_channel`, and no `max_output`; an explicit `max_concurrent > 1` is
4883 required. The default (absent or ``1``) keeps the sequential in-node path,
4884 and the excluded edges keep their existing in-node semantics.
4885 """
4886 if step.get("type") != "foreach":
4887 return False
4888 if step.get("id") in route_plan.target_node_ids:
4889 return False
4890 if step.get("routes") or step.get("output_channel"):
4891 return False
4892 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
4893 return False
4894 if step.get("max_output") is not None:
4895 return False
4896 if StepRunner._declared_item_source(step) is None:
4897 return False
4898 if not (step.get("body") or step.get("agent")):
4899 return False
4900 if step.get("max_concurrent") is None:
4901 return False
4902 return _resolved_send_concurrency(step, build_ctx) > 1
4905class _ChannelGraphShim:
4906 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
4908 Preserves the invoke contract used by main.py and tests: a flat dict in,
4909 a flat dict out. The flat input is seeded through ``seed_committed`` and
4910 nested under ``__core__`` for the compiled graph; the committed snapshot is
4911 unwrapped back out. Streaming and state surfaces are not exposed because
4912 they would leak or accept the internal ``__core__`` shape; read-only
4913 introspection delegates to the compiled graph.
4914 """
4916 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
4917 self._compiled = compiled
4918 self._specs = specs
4920 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
4921 seeded = seed_committed(self._specs, flat_state)
4922 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
4923 return unwrap(result)
4925 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
4926 seeded = seed_committed(self._specs, flat_state)
4927 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
4928 return unwrap(result)
4930 def stream(self, *args: Any, **kwargs: Any) -> Any:
4931 self._reject_internal_shape("stream")
4933 def astream(self, *args: Any, **kwargs: Any) -> Any:
4934 self._reject_internal_shape("astream")
4936 def update_state(self, *args: Any, **kwargs: Any) -> Any:
4937 self._reject_internal_shape("update_state")
4939 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
4940 self._reject_internal_shape("aupdate_state")
4942 def get_state(self, *args: Any, **kwargs: Any) -> Any:
4943 self._reject_internal_shape("get_state")
4945 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
4946 self._reject_internal_shape("aget_state")
4948 def _reject_internal_shape(self, surface: str) -> None:
4949 raise NotImplementedError(
4950 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
4951 ".invoke()/.ainvoke() contract is exposed through the channel "
4952 "boundary shim"
4953 )
4955 def __getattr__(self, name: str) -> Any:
4956 return getattr(self._compiled, name)
4959class GraphBuilder:
4960 """
4961 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
4963 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
4964 käytetään vain {{ }} -templatejen evaluointiin.
4965 """
4967 def __init__(
4968 self,
4969 registry: ManifestRegistry,
4970 builtins: BuiltinExecutor | None = None,
4971 invoke_agent_fn: Any | None = None,
4972 run_context: dict | None = None,
4973 tracer: Any | None = None,
4974 ) -> None:
4975 self._registry = registry
4976 self._resolver = AgentResolver(registry)
4977 self._invoke_agent_fn = invoke_agent_fn
4978 self._builtins = builtins or _DEFAULT_BUILTINS
4979 if invoke_agent_fn is not None:
4980 self._builtins.invoke_agent_fn = invoke_agent_fn
4981 self._step_functions = StepFunctionExecutor(registry)
4982 self._tool_registry = build_platform_registry()
4983 self._tool_resolver = ToolResolver(self._tool_registry, self._step_functions)
4984 self._step_runner = StepRunner(
4985 self._resolver, self._builtins, invoke_agent_fn, run_context,
4986 step_functions=self._step_functions,
4987 tracer=tracer,
4988 tool_resolver=self._tool_resolver,
4989 )
4991 def build(self, pipeline: dict): # -> CompiledGraph
4992 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
4993 try:
4994 from langgraph.graph import StateGraph, END
4995 except ImportError as exc:
4996 raise ImportError(
4997 "langgraph is required to build a workflow. "
4998 "Install it with: pip install langgraph"
4999 ) from exc
5001 self._step_runner.set_template_spec(pipeline.get("spec") or {})
5003 raw_steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
5004 _validate_unique_step_ids(raw_steps)
5005 _validate_foreach_bodies(raw_steps)
5006 _validate_loop_limits(raw_steps, self._step_runner._make_ctx({}))
5007 _validate_fail_if_exhausted_steps(raw_steps)
5009 channels: list[dict] = _validated_channels(
5010 (pipeline.get("spec") or {}).get("channels")
5011 )
5012 specs = parse_channel_specs(
5013 [
5014 {
5015 "name": ch["name"],
5016 "type": ch.get("type", LAST_VALUE),
5017 "operator": ch.get("operator"),
5018 }
5019 for ch in channels
5020 ]
5021 )
5022 if not any(spec.name == "lastvalue" for spec in specs):
5023 specs = list(specs) + [ChannelSpec("lastvalue", LAST_VALUE)]
5024 channel_specs = {spec.name: spec for spec in specs}
5025 channel_names = frozenset(channel_specs)
5027 # Build-time call-step composition (ADR-0037 D4): each invoked Function
5028 # body is compiled into its own native LangGraph subgraph node, so the
5029 # runtime graph executes only native steps; a call step never survives.
5030 steps = self._compose_calls(
5031 raw_steps, channels, channel_names, channel_specs
5032 )
5033 if "lastvalue" in _collect_composed_step_ids(steps):
5034 raise RunHardFailure(
5035 "CHANNEL_BINDING_ERROR",
5036 "Pipeline step id 'lastvalue' collides with the reserved "
5037 "default channel key; rename the step",
5038 )
5039 if channels:
5040 _validate_channel_names(channels, steps)
5041 _validate_channel_inputs(channels, steps)
5042 _validate_step_channel_bindings(steps, channel_names)
5043 ordered = _topological_order(steps)
5045 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
5046 envs = (pipeline.get("spec") or {}).get("envs")
5047 if envs is not None and not isinstance(envs, dict):
5048 raise RunHardFailure(
5049 "VALIDATION_ERROR",
5050 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
5051 "spec.envs must be a map of env name → image ref (ADR-0012)",
5052 )
5054 graph = self._assemble_graph(ordered, channel_names, channel_specs)
5055 compiled = graph.compile()
5056 return _ChannelGraphShim(compiled, specs)
5058 def _compose_calls(
5059 self,
5060 steps: list[dict],
5061 channels: list[dict],
5062 channel_names: frozenset[str],
5063 channel_specs: dict[str, ChannelSpec],
5064 ) -> list[dict]:
5065 """Replace `call` steps with compiled native subgraph nodes.
5067 Each invoked Function body is assembled through ``_assemble_graph`` and
5068 compiled, then substituted as a single ``__subgraph__`` placeholder that
5069 keeps the call id and its ``depends_on``. Downstream steps addressing the
5070 call id therefore address the subgraph node unchanged.
5071 """
5073 def assemble_subgraph(sub_steps: list[dict]):
5074 return self._assemble_graph(
5075 sub_steps, channel_names, channel_specs
5076 ).compile()
5078 def assemble_while(while_step: dict, body_steps: list[dict]):
5079 return self._assemble_while_graph(
5080 while_step, body_steps, channel_names, channel_specs
5081 )
5083 def assemble_foreach(foreach_step: dict, body_steps: list[dict]):
5084 return self._assemble_foreach_graph(
5085 foreach_step, body_steps, channel_names, channel_specs
5086 )
5088 composer = _CallComposer(
5089 self._registry,
5090 assemble_subgraph,
5091 channels,
5092 channel_names,
5093 assemble_while=assemble_while,
5094 compose_whiles=True,
5095 assemble_foreach=assemble_foreach,
5096 compose_foreachs=True,
5097 build_ctx=self._step_runner._make_ctx({}),
5098 )
5099 return composer.compose(steps)
5101 def _assemble_while_graph(
5102 self,
5103 step: dict,
5104 body_steps: list[dict],
5105 channel_names: frozenset[str],
5106 channel_specs: dict[str, ChannelSpec],
5107 ) -> tuple[Any, int]:
5108 """Compile a `while` step's body into a native LangGraph cycle.
5110 The cycle is ``head -> body... -> cond`` with a conditional back-edge
5111 from ``cond`` to ``head`` (continue) or ``exit`` (termination): the
5112 entry condition/initial value live in the head, the feed + exit
5113 evaluation in the cond node, and the exit node writes the documented
5114 result. Returns the compiled cycle and the recursion bound the wrapper
5115 must apply (a cycle consumes one super-step per node per iteration).
5116 """
5117 from langgraph.graph import StateGraph, END
5119 step_id = step.get("id", "?")
5120 step_runner = self._step_runner
5121 step_runner.set_channel_routing(channel_names)
5123 state_schema = _build_state_schema()
5124 graph: StateGraph = StateGraph(state_schema)
5126 max_iter_raw = step.get("max_iterations")
5127 max_iter = (
5128 _resolve_int_strict(
5129 max_iter_raw, step_runner._make_ctx({}), "max_iterations", step_id
5130 )
5131 if max_iter_raw is not None
5132 else 2
5133 )
5134 head_id = f"__while_{step_id}__head"
5135 cond_id = f"__while_{step_id}__cond"
5136 exit_id = f"__while_{step_id}__exit"
5137 first_id = body_steps[0]["id"] if body_steps else cond_id
5139 graph.add_node(
5140 head_id,
5141 _make_while_head_node(step, step_runner, channel_names, max_iter),
5142 )
5143 for sub in body_steps:
5144 graph.add_node(
5145 sub["id"],
5146 _make_while_body_node(
5147 sub, step_runner, channel_names, channel_specs
5148 ),
5149 )
5150 graph.add_node(
5151 cond_id, _make_while_cond_node(step, step_runner, channel_names)
5152 )
5153 graph.add_node(
5154 exit_id, _make_while_exit_node(step, step_runner, channel_names)
5155 )
5157 graph.set_entry_point(head_id)
5158 head_paths = {first_id: first_id, exit_id: exit_id, END: END}
5159 graph.add_conditional_edges(
5160 head_id,
5161 _make_while_head_router(step, channel_names, first_id, exit_id, END),
5162 head_paths,
5163 )
5164 for previous, sub in zip(body_steps, body_steps[1:]):
5165 graph.add_edge(previous["id"], sub["id"])
5166 if body_steps:
5167 graph.add_edge(body_steps[-1]["id"], cond_id)
5168 graph.add_conditional_edges(
5169 cond_id,
5170 _make_while_cond_router(step, channel_names, head_id, exit_id),
5171 {head_id: head_id, exit_id: exit_id},
5172 )
5173 graph.add_edge(exit_id, END)
5175 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5176 return graph.compile(), recursion_limit
5178 def _assemble_foreach_graph(
5179 self,
5180 step: dict,
5181 body_steps: list[dict],
5182 channel_names: frozenset[str],
5183 channel_specs: dict[str, ChannelSpec],
5184 ) -> tuple[Any, int]:
5185 """Compile a sequential `foreach` step's body into a native LangGraph cycle.
5187 The cycle is ``head -> (body... | agent) -> cond`` with a conditional
5188 back-edge from ``cond`` to ``head`` (next item) or ``exit`` (exhaustion
5189 or `max_output` reached). The head resolves the item list once and
5190 injects the current item; the cond collects the item result; the exit
5191 node writes the documented aggregate. Returns the compiled cycle and
5192 the recursion bound the wrapper must apply.
5193 """
5194 from langgraph.graph import StateGraph, END
5196 step_id = step.get("id", "?")
5197 step_runner = self._step_runner
5198 step_runner.set_channel_routing(channel_names)
5200 state_schema = _build_state_schema()
5201 graph: StateGraph = StateGraph(state_schema)
5203 max_iter_raw = step.get("max_iterations")
5204 max_iter = (
5205 _resolve_int_strict(
5206 max_iter_raw, step_runner._make_ctx({}), "max_iterations", step_id
5207 )
5208 if max_iter_raw is not None
5209 else _FOREACH_DEFAULT_MAX_ITER
5210 )
5211 head_id = f"__foreach_{step_id}__head"
5212 cond_id = f"__foreach_{step_id}__cond"
5213 exit_id = f"__foreach_{step_id}__exit"
5214 agent_id = f"__foreach_{step_id}__agent"
5216 graph.add_node(
5217 head_id, _make_foreach_head_node(step, step_runner, channel_names)
5218 )
5219 if body_steps:
5220 for sub in body_steps:
5221 graph.add_node(
5222 sub["id"],
5223 _make_foreach_body_node(
5224 sub, step_runner, channel_names, channel_specs
5225 ),
5226 )
5227 first_id = body_steps[0]["id"]
5228 else:
5229 graph.add_node(
5230 agent_id,
5231 _make_foreach_agent_node(
5232 step, step_runner, channel_names, channel_specs
5233 ),
5234 )
5235 first_id = agent_id
5236 graph.add_node(
5237 cond_id, _make_foreach_cond_node(step, step_runner, channel_names)
5238 )
5239 graph.add_node(
5240 exit_id, _make_foreach_exit_node(step, step_runner, channel_names)
5241 )
5243 graph.set_entry_point(head_id)
5244 graph.add_conditional_edges(
5245 head_id,
5246 _make_foreach_head_router(step, channel_names, first_id, exit_id, END),
5247 {first_id: first_id, exit_id: exit_id, END: END},
5248 )
5249 for previous, sub in zip(body_steps, body_steps[1:]):
5250 graph.add_edge(previous["id"], sub["id"])
5251 if body_steps:
5252 graph.add_edge(body_steps[-1]["id"], cond_id)
5253 else:
5254 graph.add_edge(agent_id, cond_id)
5255 graph.add_conditional_edges(
5256 cond_id,
5257 _make_foreach_cond_router(step, channel_names, head_id, exit_id),
5258 {head_id: head_id, exit_id: exit_id},
5259 )
5260 graph.add_edge(exit_id, END)
5262 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5263 return graph.compile(), recursion_limit
5265 def _assemble_graph(
5266 self,
5267 steps: list[dict],
5268 channel_names: frozenset[str],
5269 channel_specs: dict[str, ChannelSpec],
5270 ) -> "StateGraph":
5271 from langgraph.graph import StateGraph, END
5273 # Group steps into LangGraph nodes:
5274 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
5275 nodes: list[list[dict]] = []
5276 for step in steps:
5277 dep_sig = frozenset(step.get("depends_on") or [])
5278 if nodes:
5279 last_group_sig = frozenset(
5280 dep
5281 for s in nodes[-1]
5282 for dep in (s.get("depends_on") or [])
5283 )
5284 last_types = {s.get("type") for s in nodes[-1]}
5285 if (
5286 dep_sig == last_group_sig
5287 and step.get("type") == "parallel"
5288 and last_types == {"parallel"}
5289 and not step.get("routes")
5290 and not any(s.get("routes") for s in nodes[-1])
5291 ):
5292 nodes[-1].append(step)
5293 continue
5294 nodes.append([step])
5296 state_schema = _build_state_schema()
5297 graph: StateGraph = StateGraph(state_schema)
5298 step_runner = self._step_runner
5299 step_runner.set_channel_routing(channel_names)
5301 node_ids: list[str] = []
5302 for group in nodes:
5303 if len(group) == 1:
5304 node_ids.append(group[0]["id"])
5305 else:
5306 node_ids.append("parallel_" + "_".join(s["id"] for s in group))
5308 route_plan = _build_route_plan(nodes, node_ids) if nodes else None
5310 send_parts: dict[int, _MapReduceParts] = {}
5311 if route_plan is not None:
5312 build_ctx = step_runner._make_ctx({})
5313 for index, group in enumerate(nodes):
5314 if len(group) != 1:
5315 continue
5316 send_candidate = group[0]
5317 if _is_send_parallel_step(send_candidate, route_plan, build_ctx):
5318 plan = _make_parallel_map_reduce_plan(
5319 send_candidate, step_runner
5320 )
5321 elif _is_send_foreach_step(send_candidate, route_plan, build_ctx):
5322 plan = _make_foreach_map_reduce_plan(
5323 send_candidate, step_runner, channel_names, channel_specs
5324 )
5325 else:
5326 continue
5327 send_parts[index] = _make_map_reduce_parts(
5328 send_candidate,
5329 plan,
5330 step_runner,
5331 channel_names,
5332 channel_specs,
5333 _resolved_send_concurrency(send_candidate, build_ctx),
5334 )
5336 entry_ids: list[str] = []
5337 exit_ids: list[str] = []
5338 for index, group in enumerate(nodes):
5339 parts = send_parts.get(index)
5340 if parts is None:
5341 captured_group = list(group)
5342 if len(captured_group) == 1:
5343 first = captured_group[0]
5344 if first.get("type") == _SUBGRAPH_STEP_TYPE:
5345 if first.get(_SUBGRAPH_WHILE_KEY):
5346 subgraph_node = _make_while_subgraph_node(
5347 first, step_runner, channel_names
5348 )
5349 elif first.get(_SUBGRAPH_FOREACH_KEY):
5350 subgraph_node = _make_foreach_subgraph_node(
5351 first, step_runner, channel_names
5352 )
5353 else:
5354 subgraph_node = _make_subgraph_node(
5355 first[_SUBGRAPH_GRAPH_KEY],
5356 first.get("condition"),
5357 step_runner,
5358 channel_names,
5359 first.get(_SUBGRAPH_STEP_IDS_KEY),
5360 )
5361 graph.add_node(node_ids[index], subgraph_node)
5362 entry_ids.append(node_ids[index])
5363 exit_ids.append(node_ids[index])
5364 continue
5365 fn = _make_step_node(
5366 first, step_runner, channel_names, channel_specs,
5367 )
5368 else:
5369 fn = _make_parallel_node(
5370 captured_group,
5371 step_runner,
5372 channel_names,
5373 channel_specs,
5374 )
5375 graph.add_node(node_ids[index], fn)
5376 entry_ids.append(node_ids[index])
5377 exit_ids.append(node_ids[index])
5378 continue
5379 fallback_step = group[0]
5380 graph.add_node(parts.dispatch_id, parts.dispatch_fn)
5381 graph.add_node(parts.worker_id, parts.worker_fn)
5382 graph.add_node(
5383 parts.fallback_id,
5384 _make_step_node(
5385 fallback_step, step_runner, channel_names, channel_specs
5386 ),
5387 )
5388 graph.add_node(parts.join_id, parts.join_fn)
5389 graph.add_conditional_edges(
5390 parts.dispatch_id,
5391 parts.route_fn,
5392 [parts.worker_id, parts.fallback_id],
5393 )
5394 graph.add_edge(parts.worker_id, parts.join_id)
5395 graph.add_edge(parts.fallback_id, parts.join_id)
5396 entry_ids.append(parts.dispatch_id)
5397 exit_ids.append(parts.join_id)
5399 if nodes:
5400 _wire_dependency_edges(
5401 graph, nodes, node_ids, END, route_plan, entry_ids, exit_ids
5402 )
5403 _wire_route_edges(graph, route_plan, END, step_runner)
5404 else:
5405 graph.add_node("noop", lambda s: s)
5406 graph.set_entry_point("noop")
5407 graph.add_edge("noop", END)
5409 return graph
5412# ---------------------------------------------------------------------------
5413# Julkinen API
5414# ---------------------------------------------------------------------------
5416_registry_cache: ManifestRegistry | None = None
5419def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
5420 global _registry_cache
5421 if _registry_cache is None or dirs is not None:
5422 loader = ManifestLoader(dirs)
5423 manifests = loader.load_all()
5424 _registry_cache = ManifestRegistry(manifests)
5425 return _registry_cache
5428def build_workflow(
5429 pipeline_name: str,
5430 *,
5431 dirs: list[Path] | None = None,
5432 builtins: BuiltinExecutor | None = None,
5433 invoke_agent_fn: Any | None = None,
5434 run_context: dict | None = None,
5435 tracer: Any | None = None,
5436):
5437 """
5438 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
5439 LangGraph-workflown.
5441 Args:
5442 pipeline_name: Pipeline metadata.name -arvo.
5443 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
5444 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
5445 invoke_agent_fn: fn(resolved_spec, step, state) → state.
5446 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
5448 Returns:
5449 Compilattu LangGraph CompiledGraph.
5450 """
5451 registry = _get_registry(dirs)
5452 pipeline = registry.get("Pipeline", pipeline_name)
5453 if pipeline is None:
5454 available = [
5455 (m.get("metadata") or {}).get("name")
5456 for m in registry.all_of_kind("Pipeline")
5457 ]
5458 raise RunHardFailure(
5459 "VALIDATION_ERROR",
5460 f"Pipeline '{pipeline_name}' not found. Available: {available}",
5461 )
5462 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
5463 return builder.build(pipeline)