Coverage for core/langgraph/runner.py: 94%
2448 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-10-05 00:44 +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 if body_polku:
2443 step_result: Any = self._aggregate_body_results(ordered)
2444 else:
2445 # Agent path: the per-item envelopes still carry their ``content``
2446 # here, so route each item's declared output channel write in item
2447 # order before metadata projection strips the content. A body-path
2448 # foreach staged its per-item writes through ``item_working_sets``
2449 # already; routing the body path here would double-write.
2450 self._stage_fanout_channel_writes(step, ordered, working_set)
2451 step_result = [project_envelope_metadata(r) for r in ordered]
2452 final: PipelineState = {**state, step_id: step_result}
2453 if accumulated_metrics:
2454 final = {
2455 **final,
2456 "step_context_metrics": _merge_side_channel(
2457 state.get("step_context_metrics"), accumulated_metrics
2458 ),
2459 }
2460 if accumulated_tool_meta:
2461 final = {
2462 **final,
2463 "step_tool_meta": _merge_side_channel(
2464 state.get("step_tool_meta"), accumulated_tool_meta
2465 ),
2466 }
2467 if item_errors:
2468 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2469 return final
2472# ---------------------------------------------------------------------------
2473# GraphBuilder
2474# ---------------------------------------------------------------------------
2476def _topological_order(steps: list[dict]) -> list[dict]:
2477 """
2478 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2479 """
2480 id_to_step = {s["id"]: s for s in steps}
2481 visited: set[str] = set()
2482 order: list[dict] = []
2484 def visit(step_id: str) -> None:
2485 if step_id in visited:
2486 return
2487 visited.add(step_id)
2488 for dep in (id_to_step[step_id].get("depends_on") or []):
2489 if dep in id_to_step:
2490 visit(dep)
2491 order.append(id_to_step[step_id])
2493 for s in steps:
2494 visit(s["id"])
2495 return order
2498def _group_dependencies(
2499 groups: list[list[dict]],
2500) -> tuple[list[set[int]], list[set[int]]]:
2501 """Group-index adjacency (predecessors, successors) from each step's depends_on."""
2502 step_to_group = {
2503 step["id"]: index
2504 for index, group in enumerate(groups)
2505 for step in group
2506 }
2507 predecessors: list[set[int]] = [set() for _ in groups]
2508 successors: list[set[int]] = [set() for _ in groups]
2509 for index, group in enumerate(groups):
2510 for step in group:
2511 for dep in step.get("depends_on") or []:
2512 dep_index = step_to_group.get(dep)
2513 if dep_index is None or dep_index == index:
2514 continue
2515 predecessors[index].add(dep_index)
2516 successors[dep_index].add(index)
2517 return predecessors, successors
2520@dataclass(frozen=True)
2521class _RoutePlan:
2522 """Compiled `routes` topology: route tables plus the static-edge suppression sets.
2524 `routes` owns a routing step's outgoing control flow (ADR-0025). A route
2525 target reached via a static `depends_on` edge is therefore suppressed and
2526 replaced by the native conditional edge; the plan records both the targets
2527 (so they are never made entry points) and the suppressed `(source, target)`
2528 pairs.
2529 """
2531 step_to_node: dict[str, str]
2532 routes_by_step: dict[str, list[dict]]
2533 target_node_ids: frozenset[str]
2534 routing_node_ids: frozenset[str]
2535 suppressed_edges: frozenset[tuple[str, str]]
2538def _reject_nested_routes(step: dict) -> None:
2539 """Reject `routes` on a nested `body` step (top-level-only rule, ADR-0012).
2541 The engine wires `routes` only for top-level steps, so a nested declaration
2542 would be silently ignored; nested routing is deferred.
2543 """
2544 body = step.get("body")
2545 if not isinstance(body, list):
2546 return
2547 for nested in _iter_steps(body):
2548 if nested.get("routes"):
2549 raise RunHardFailure(
2550 "VALIDATION_ERROR",
2551 f"Step '{nested.get('id')}' declares `routes` inside a nested "
2552 "body; `routes` is only allowed on top-level steps",
2553 )
2556def _build_route_plan(groups: list[list[dict]], node_ids: list[str]) -> _RoutePlan:
2557 """Resolve every top-level step's `routes` into a node-level routing plan."""
2558 step_to_node = {
2559 step["id"]: node_ids[index]
2560 for index, group in enumerate(groups)
2561 for step in group
2562 }
2563 step_to_deps = {
2564 step["id"]: set(step.get("depends_on") or [])
2565 for group in groups
2566 for step in group
2567 }
2568 routes_by_step: dict[str, list[dict]] = {}
2569 target_node_ids: set[str] = set()
2570 routing_node_ids: set[str] = set()
2571 suppressed_edges: set[tuple[str, str]] = set()
2572 for group in groups:
2573 for step in group:
2574 _reject_nested_routes(step)
2575 routes = step.get("routes")
2576 if not routes:
2577 continue
2578 source = step["id"]
2579 source_node = step_to_node[source]
2580 routing_node_ids.add(source_node)
2581 routes_by_step[source] = routes
2582 for route in routes:
2583 target = route.get("target")
2584 target_node = step_to_node.get(target)
2585 if target_node is None:
2586 raise RunHardFailure(
2587 "VALIDATION_ERROR",
2588 f"Step '{source}' route target '{target}' is not a "
2589 "top-level step in this pipeline",
2590 )
2591 target_node_ids.add(target_node)
2592 if source in step_to_deps.get(target, set()):
2593 suppressed_edges.add((source_node, target_node))
2594 log.warning(
2595 "[routes] suppressing static edge %s -> %s; route "
2596 "targets are wired via `routes`",
2597 source,
2598 target,
2599 )
2600 return _RoutePlan(
2601 step_to_node=step_to_node,
2602 routes_by_step=routes_by_step,
2603 target_node_ids=frozenset(target_node_ids),
2604 routing_node_ids=frozenset(routing_node_ids),
2605 suppressed_edges=frozenset(suppressed_edges),
2606 )
2609def _make_route_router(
2610 routes: list[dict], step_to_node: dict[str, str], end, step_runner: Any
2611) -> Any:
2612 """First-match-wins router over a routing step's declared routes.
2614 Evaluates `when` in declaration order and returns the selected target node;
2615 a route without `when` is the default branch. With no match and no default
2616 it returns the graph terminal.
2617 """
2618 def router(state: PipelineState) -> str:
2619 flat = _flatten_channel_state(state)
2620 ctx = step_runner._make_ctx(flat)
2621 for route in routes:
2622 if _eval_condition(route.get("when"), ctx):
2623 return step_to_node[route["target"]]
2624 return end
2626 return router
2629def _wire_route_edges(
2630 graph, route_plan: _RoutePlan, end, step_runner: Any
2631) -> None:
2632 """Add one native conditional edge per routing step, with an explicit path_map."""
2633 for source, routes in route_plan.routes_by_step.items():
2634 target_nodes = [route_plan.step_to_node[r["target"]] for r in routes]
2635 path_map = {node: node for node in target_nodes}
2636 path_map[end] = end
2637 graph.add_conditional_edges(
2638 route_plan.step_to_node[source],
2639 _make_route_router(routes, route_plan.step_to_node, end, step_runner),
2640 path_map,
2641 )
2644def _wire_dependency_edges(
2645 graph,
2646 groups: list[list[dict]],
2647 node_ids: list[str],
2648 end,
2649 route_plan: _RoutePlan,
2650 entry_ids: list[str] | None = None,
2651 exit_ids: list[str] | None = None,
2652) -> None:
2653 """Wire native branches and fan-in joins from group-level depends_on edges.
2655 Static edges originating from a routing step to one of its route targets are
2656 suppressed — `routes` supplies that control flow. A route target with no
2657 other predecessor is reached by its conditional edge, never an entry point.
2658 A group split into a Send fan-out enters through its dispatch node and exits
2659 through its join node, so ``entry_ids``/``exit_ids`` name those boundaries.
2660 """
2661 entry_ids = entry_ids if entry_ids is not None else node_ids
2662 exit_ids = exit_ids if exit_ids is not None else node_ids
2663 predecessors, successors = _group_dependencies(groups)
2664 for index, node_id in enumerate(node_ids):
2665 entry = entry_ids[index]
2666 pred_nodes = [exit_ids[p] for p in sorted(predecessors[index])]
2667 if not pred_nodes:
2668 if node_id not in route_plan.target_node_ids:
2669 graph.set_entry_point(entry)
2670 else:
2671 live_preds = [
2672 pred
2673 for pred in pred_nodes
2674 if (pred, entry) not in route_plan.suppressed_edges
2675 ]
2676 if len(live_preds) == 1:
2677 graph.add_edge(live_preds[0], entry)
2678 elif live_preds:
2679 graph.add_edge(live_preds, entry)
2680 if not successors[index] and node_id not in route_plan.routing_node_ids:
2681 graph.add_edge(exit_ids[index], end)
2684# ---------------------------------------------------------------------------
2685# Call-step build-time expansion (ADR-0037 D3/D4)
2686# ---------------------------------------------------------------------------
2687#
2688# A `type: call` step is composed at build time: the function's body steps plus
2689# two synthesized op steps (`__call_bind` / `__call_emit`) are assembled into a
2690# compiled subgraph added as a single parent-graph node keyed by the call id. The
2691# runtime executes only native steps; LangGraph never sees a call step. The caller
2692# reads the function outputs via the call result shape
2693# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
2694# The function body's `{{ params.<name> }}` references are rewritten to
2695# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves. A
2696# call nested in a foreach/while body is composed the same way: the loop body is a
2697# native subgraph and the call becomes a compiled subgraph node keyed by the call id.
2698#
2699# Isolation (ADR-0037 D5 / 00-functions § Isolation) and recursion (ADR-0037 D7): a
2700# function body step must not reference a step-id outside its own body (a
2701# one-way dependency check — outward references to caller step-ids are build
2702# errors); a cycle in the function call graph (self or mutual) is a build
2703# error. Bounded while loops are runtime constructs and are not cycle-flagged.
2705# A bare `params` reference root (`params`, `params.x`, `params['x']`). A
2706# preceding `.` or word character means it is not the root (`foo.params`,
2707# `myparams`, `paramsfoo`). Quoted literals are consumed whole, so a `params`
2708# inside them is invisible to the bare-reference branch.
2709_QUOTED_OR_BARE_PARAMS_RE = re.compile(
2710 r"'[^']*'|\"[^\"]*\"|(?<![\w.])params(?![\w])"
2711)
2713# A leading `params` root in an `input` bare dot-path (`params.x`, `params`,
2714# `params[0]`), optionally after whitespace.
2715_LEADING_PARAMS_RE = re.compile(r"^(\s*)params(?=\.|\[|\s*$)")
2718def _scope_template_expression(expr: str, call_id: str) -> str:
2719 """Rewrite bare `params` references inside one `{{ ... }}` token."""
2721 def _replace(match: re.Match) -> str:
2722 token = match.group(0)
2723 return token if token[0] in "'\"" else f"{call_id}.params"
2725 return _QUOTED_OR_BARE_PARAMS_RE.sub(_replace, expr)
2728def _scope_template_string(value: str, call_id: str) -> str:
2729 """Scope the `{{ ... }}` expressions of a template-only field value."""
2730 return _TEMPLATE_REF_RE.sub(
2731 lambda m: _scope_template_expression(m.group(0), call_id),
2732 value,
2733 )
2736def _scope_input_string(value: str, call_id: str) -> str:
2737 """Scope an `input`: its `{{ ... }}` templates or its leading bare dot-path."""
2738 if "{{" in value:
2739 return _scope_template_string(value, call_id)
2740 if value.lstrip().startswith("["):
2741 return value
2742 return _LEADING_PARAMS_RE.sub(lambda m: f"{m.group(1)}{call_id}.params", value)
2745def _scope_params_value(value: Any, call_id: str, field: str) -> Any:
2746 """Rewrite bare `params` references to the call-scoped namespace.
2748 ``field`` selects the resolver semantics: ``input`` is a template or a bare
2749 state dot-path; every other scoped field is template-only. Dict keys are
2750 untouched and the input is never mutated.
2751 """
2752 if isinstance(value, str):
2753 if field == "input":
2754 return _scope_input_string(value, call_id)
2755 return _scope_template_string(value, call_id)
2756 if isinstance(value, dict):
2757 return {k: _scope_params_value(v, call_id, field) for k, v in value.items()}
2758 if isinstance(value, list):
2759 return [_scope_params_value(v, call_id, field) for v in value]
2760 return value
2763def _scope_params(step: dict, call_id: str) -> dict:
2764 """Deep-scope a function body step's params references for a call id."""
2765 scoped: dict = dict(step)
2766 for field in ("input", "iteration", "params", "condition", "exit_condition", "skip_llm_if"):
2767 if field in scoped and isinstance(scoped[field], (str, dict, list)):
2768 scoped[field] = _scope_params_value(scoped[field], call_id, field)
2769 if "body" in scoped and isinstance(scoped["body"], list):
2770 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
2771 return scoped
2774def _collect_step_ids(steps: list[dict]) -> set[str]:
2775 """All step ids declared in *steps*, including nested body step ids."""
2776 ids: set[str] = set()
2777 for step in steps:
2778 step_id = step.get("id")
2779 if isinstance(step_id, str):
2780 ids.add(step_id)
2781 body = step.get("body")
2782 if isinstance(body, list):
2783 ids |= _collect_step_ids(body)
2784 return ids
2787def _iter_steps(steps: list[dict]):
2788 """Yield every step in *steps*, recursing into nested bodies."""
2789 for step in steps:
2790 yield step
2791 body = step.get("body")
2792 if isinstance(body, list):
2793 yield from _iter_steps(body)
2796def _collect_composed_step_ids(steps: list[dict]) -> set[str]:
2797 """All declared step ids, including ids composed into native subgraphs.
2799 A ``call`` step becomes a ``__subgraph__`` placeholder whose inner step ids
2800 live only inside the compiled subgraph. The placeholder records them under
2801 ``_SUBGRAPH_STEP_IDS_KEY`` so name-collision validation sees the full set,
2802 which stays encapsulated inside the compiled subgraph.
2803 """
2804 ids = _collect_step_ids(steps)
2805 for step in _iter_steps(steps):
2806 composed_ids = step.get(_SUBGRAPH_STEP_IDS_KEY)
2807 if isinstance(composed_ids, (set, frozenset)):
2808 ids |= composed_ids
2809 return ids
2812_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
2813_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
2814_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
2816_REFERENCE_FIELDS = (
2817 "input", "iteration", "params", "condition", "exit_condition", "skip_llm_if"
2818)
2820# Engine-reserved template namespaces — never step-id references, even if a
2821# caller happens to declare a step with one of these names.
2822_RESERVED_NAMESPACES = frozenset(
2823 {"params", "item", "_item", "budget", "config", "spec", "channel", "lastvalue"}
2824)
2827def _referenced_names(value: Any) -> set[str]:
2828 """Root identifiers read by templated strings.
2830 Only the leading identifier of each reference (before the first ``.`` or
2831 ``[``) can name a step-id; nested field segments are ignored. Quoted
2832 literals are ignored.
2833 """
2834 names: set[str] = set()
2835 if isinstance(value, str):
2836 for expr in _TEMPLATE_REF_RE.findall(value):
2837 bare = _STRING_LITERAL_RE.sub("", expr)
2838 for match in _IDENTIFIER_RE.finditer(bare):
2839 if match.start() == 0 or bare[match.start() - 1] != ".":
2840 names.add(match.group())
2841 elif isinstance(value, dict):
2842 for v in value.values():
2843 names |= _referenced_names(v)
2844 elif isinstance(value, list):
2845 for v in value:
2846 names |= _referenced_names(v)
2847 return names
2850def _validate_function_purity(
2851 fn_name: str, body: list[dict], visible_ids: set[str]
2852) -> None:
2853 """One-way isolation: a function body must not reference a caller step-id.
2855 A body step may read its own body step-ids and the params namespace; a
2856 reference to a step-id that exists outside the body (``visible_ids`` but
2857 not declared in the body) is an outward dependency and a hard build error.
2858 """
2859 local_ids = _collect_step_ids(body)
2860 for step in _iter_steps(body):
2861 step_id = step.get("id", "?")
2862 for dep in step.get("depends_on") or []:
2863 if dep in visible_ids and dep not in local_ids:
2864 raise RunHardFailure(
2865 "VALIDATION_ERROR",
2866 f"function '{fn_name}' step '{step_id}': depends_on "
2867 f"'{dep}' is outside the function body (isolation — "
2868 "function bodies see only their own steps and params)",
2869 )
2870 for field in _REFERENCE_FIELDS:
2871 for ref in _referenced_names(step.get(field)):
2872 if ref in _RESERVED_NAMESPACES:
2873 continue
2874 if ref in visible_ids and ref not in local_ids:
2875 raise RunHardFailure(
2876 "VALIDATION_ERROR",
2877 f"function '{fn_name}' step '{step_id}': '{field}' "
2878 f"references '{ref}', a step-id outside the function "
2879 "body (isolation — function bodies see only their own "
2880 "steps and params)",
2881 )
2884def _reject_function_body_routes(fn_name: str, body: list[dict]) -> None:
2885 """Reject `routes` anywhere in a function body (top-level-only rule).
2887 A function body is composed as a native subgraph, so a body step declaring
2888 `routes` would escape the nested-`body` guard and be silently rewired;
2889 reject it at build time instead.
2890 """
2891 for step in _iter_steps(body):
2892 if step.get("routes"):
2893 raise RunHardFailure(
2894 "VALIDATION_ERROR",
2895 f"function '{fn_name}' step '{step.get('id')}' declares `routes`; "
2896 "`routes` is only allowed on top-level steps",
2897 )
2900def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
2901 """Expand every ``call`` step through the native subgraph composer.
2903 Test-facing seam over the single call-expansion implementation: the same
2904 ``_CallComposer`` that ``GraphBuilder`` uses composes the native subgraph, so
2905 no second expansion path exists. The embedded subgraph value is the un-compiled
2906 sub-step list; a runnable graph only needs the compiled form, which
2907 production obtains through ``GraphBuilder._compose_calls``.
2908 """
2909 composer = _CallComposer(
2910 registry,
2911 assemble_subgraph=lambda sub_steps: sub_steps,
2912 channels=[],
2913 channel_names=frozenset(),
2914 )
2915 return composer.compose(steps)
2918_SUBGRAPH_STEP_TYPE = "__subgraph__"
2919_SUBGRAPH_GRAPH_KEY = "__subgraph_graph__"
2920_SUBGRAPH_STEP_IDS_KEY = "__subgraph_step_ids__"
2922# A `while` step is composed into a `__subgraph__` placeholder whose embedded
2923# graph is a native LangGraph cycle (head -> body -> cond -> head/exit). The
2924# marker flags the placeholder so the parent uses the while node wrapper (which
2925# keeps the loop-internal state out of the parent result) rather than the plain
2926# call-subgraph node, and carries the build-time recursion bound.
2927_SUBGRAPH_WHILE_KEY = "__subgraph_while__"
2928_WHILE_RECURSION_LIMIT_KEY = "__while_recursion_limit__"
2930# Engine-local cycle state: the active while's `{current, iteration, passed,
2931# skipped, max_iter}` marker. Nested whiles live in isolated cycle
2932# invocations, so a shared key cannot collide across them.
2933_WHILE_STATE_KEY = "__while_state__"
2935# State keys the while wrapper propagates from the isolated cycle back to the
2936# parent (the loop result itself is carried separately under the step id).
2937_WHILE_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
2939# A sequential `foreach` (max_concurrent absent or resolving to <= 1) is
2940# composed into a `__subgraph__` placeholder whose embedded graph is a native
2941# LangGraph cycle (head -> body/agent -> cond -> head/exit). The marker flags
2942# the placeholder so the parent uses the foreach node wrapper, and carries the
2943# build-time recursion bound.
2944_SUBGRAPH_FOREACH_KEY = "__subgraph_foreach__"
2945_FOREACH_RECURSION_LIMIT_KEY = "__foreach_recursion_limit__"
2947# Engine-local cycle state: the active foreach's `{items, index, results,
2948# produced_outputs, max_output}` marker. Nested foreachs live in isolated cycle
2949# invocations, so a shared key cannot collide across them.
2950_FOREACH_STATE_KEY = "__foreach_state__"
2952# State keys the foreach wrapper propagates from the isolated cycle back to the
2953# parent (the aggregate result itself is carried separately under the step id).
2954_FOREACH_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
2956# Build-time recursion bound fallback for a foreach with no declared
2957# `max_iterations`: the item list is uncapped, so reserve a generous bound.
2958_FOREACH_DEFAULT_MAX_ITER = 1000
2961class _CallComposer:
2962 """Compose `call` steps into compiled native subgraph nodes (ADR-0037 D4).
2964 A function body becomes its own compiled StateGraph added to the parent as a
2965 single node keyed by the call id; the ``bind``/``emit`` scaffolding and the
2966 body steps live inside that subgraph and share the parent state. Nested
2967 function calls compose recursively. A `call` nested in a foreach/while body
2968 is composed the same way; the loop body executes the embedded compiled graph
2969 imperatively via ``run_step``'s ``__subgraph__`` branch.
2970 """
2972 def __init__(
2973 self,
2974 registry: ManifestRegistry,
2975 assemble_subgraph: Any,
2976 channels: list[dict],
2977 channel_names: frozenset[str],
2978 *,
2979 assemble_while: Any | None = None,
2980 compose_whiles: bool = False,
2981 assemble_foreach: Any | None = None,
2982 compose_foreachs: bool = False,
2983 build_ctx: dict | None = None,
2984 ) -> None:
2985 self._registry = registry
2986 self._assemble_subgraph = assemble_subgraph
2987 self._channels = channels
2988 self._channel_names = channel_names
2989 self._assemble_while = assemble_while
2990 self._compose_whiles = compose_whiles
2991 self._assemble_foreach = assemble_foreach
2992 self._compose_foreachs = compose_foreachs
2993 self._build_ctx = build_ctx or {}
2994 self._stack: list[str] = []
2995 self._expanded_functions: dict[str, str] = {}
2997 def compose(self, steps: list[dict]) -> list[dict]:
2998 visible_ids = _collect_step_ids(steps)
2999 composed: list[dict] = []
3000 for step in steps:
3001 composed.extend(self._compose_step(step, visible_ids=visible_ids))
3002 return composed
3004 def _compose_step(self, step: dict, *, visible_ids: set[str]) -> list[dict]:
3005 step_type = step.get("type")
3006 if step_type == "call":
3007 return [self._compose_subgraph_call(step, visible_ids)]
3008 if step_type == "while" and self._compose_whiles:
3009 return [self._compose_while(step, visible_ids)]
3010 if step_type == "foreach" and self._should_compose_foreach(step):
3011 return [self._compose_foreach(step, visible_ids)]
3012 body = step.get("body")
3013 if not isinstance(body, list):
3014 return [step]
3015 nested: list[dict] = []
3016 for sub in body:
3017 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3018 return [{**step, "body": nested}]
3020 def _compose_while(self, step: dict, visible_ids: set[str]) -> dict:
3021 """Compose a `while` step into a native LangGraph cycle subgraph.
3023 The embedded graph owns the loop topology: a head that resolves the
3024 entry condition / initial ``current``, the composed body step nodes, a
3025 conditional back-edge that computes the feed + exit condition and routes
3026 continue-vs-exit, and an exit node that writes the documented while
3027 result. ``expand_calls`` leaves the step uncomposed (``compose_whiles``
3028 false) so the call-expansion seam still exposes the loop body.
3029 """
3030 _reject_nested_routes(step)
3031 body = step.get("body") or []
3032 nested: list[dict] = []
3033 for sub in body:
3034 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3035 _validate_step_channel_bindings(nested, self._channel_names)
3036 compiled, recursion_limit = self._assemble_while(step, nested)
3037 placeholder: dict = {
3038 "id": step["id"],
3039 "type": _SUBGRAPH_STEP_TYPE,
3040 _SUBGRAPH_GRAPH_KEY: compiled,
3041 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3042 _SUBGRAPH_WHILE_KEY: True,
3043 _WHILE_RECURSION_LIMIT_KEY: recursion_limit,
3044 "depends_on": list(step.get("depends_on") or []),
3045 }
3046 if step.get("routes"):
3047 placeholder["routes"] = step["routes"]
3048 return placeholder
3050 def _should_compose_foreach(self, step: dict) -> bool:
3051 """Whether a `foreach` is the sequential form composed into a cycle.
3053 Only a `max_concurrent` absent or resolving to ``<= 1`` is composed; a
3054 declared ``max_concurrent > 1`` keeps the native ``Send`` / in-node
3055 concurrent behaviour.
3056 """
3057 if not self._compose_foreachs:
3058 return False
3059 return _resolved_send_concurrency(step, self._build_ctx) <= 1
3061 def _compose_foreach(self, step: dict, visible_ids: set[str]) -> dict:
3062 """Compose a sequential `foreach` into a native LangGraph cycle subgraph.
3064 Mirrors ``_compose_while``: the embedded graph owns the loop topology
3065 (a head that resolves the item list and injects the current item, the
3066 composed body step nodes or the per-item agent node, a conditional
3067 back-edge routing next-item-vs-exit, and an exit node that writes the
3068 documented aggregate). ``expand_calls`` leaves the step uncomposed so
3069 the call-expansion seam still exposes the loop body.
3070 """
3071 _reject_nested_routes(step)
3072 body = step.get("body") or []
3073 nested: list[dict] = []
3074 for sub in body:
3075 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3076 _validate_step_channel_bindings(nested, self._channel_names)
3077 compiled, recursion_limit = self._assemble_foreach(step, nested)
3078 placeholder: dict = {
3079 "id": step["id"],
3080 "type": _SUBGRAPH_STEP_TYPE,
3081 _SUBGRAPH_GRAPH_KEY: compiled,
3082 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3083 _SUBGRAPH_FOREACH_KEY: True,
3084 _FOREACH_RECURSION_LIMIT_KEY: recursion_limit,
3085 "depends_on": list(step.get("depends_on") or []),
3086 }
3087 if step.get("routes"):
3088 placeholder["routes"] = step["routes"]
3089 return placeholder
3091 def _compose_subgraph_call(self, step: dict, visible_ids: set[str]) -> dict:
3092 call_id = step["id"]
3093 function, body = self._check_call_guards(step, visible_ids)
3095 nested_visible = visible_ids | _collect_step_ids(body)
3096 outputs_spec = (
3097 ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
3098 )
3099 bind_id = f"_call__{call_id}__bind"
3100 emit_id = f"_call__{call_id}__emit"
3101 bind_step: dict = {
3102 "id": bind_id,
3103 "type": "builtin",
3104 "component": "__call_bind",
3105 "params": {**(step.get("params") or {}), "__call_id": call_id},
3106 "depends_on": [],
3107 }
3109 self._stack.append(step["function"])
3110 try:
3111 expanded_body = self._compose_body(
3112 body, call_id, bind_id, nested_visible
3113 )
3114 finally:
3115 self._stack.pop()
3117 emit_step: dict = {
3118 "id": emit_id,
3119 "type": "builtin",
3120 "component": "__call_emit",
3121 "params": {"__call_id": call_id, "__outputs": outputs_spec},
3122 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
3123 }
3124 sub_steps = [bind_step, *expanded_body, emit_step]
3125 if self._channels:
3126 _validate_channel_inputs(self._channels, sub_steps)
3127 _validate_step_channel_bindings(sub_steps, self._channel_names)
3128 placeholder: dict = {
3129 "id": call_id,
3130 "type": _SUBGRAPH_STEP_TYPE,
3131 _SUBGRAPH_GRAPH_KEY: self._assemble_subgraph(
3132 _topological_order(sub_steps)
3133 ),
3134 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(sub_steps),
3135 "depends_on": list(step.get("depends_on") or []),
3136 }
3137 call_condition = step.get("condition")
3138 if call_condition is not None:
3139 placeholder["condition"] = call_condition
3140 return placeholder
3142 def _compose_body(
3143 self, body: list[dict], call_id: str, bind_id: str, visible_ids: set[str]
3144 ) -> list[dict]:
3145 expanded: list[dict] = []
3146 for body_step in body:
3147 scoped = _scope_params(body_step, call_id)
3148 for sub in self._compose_step(scoped, visible_ids=visible_ids):
3149 deps = list(sub.get("depends_on") or [])
3150 if bind_id in deps:
3151 expanded.append(sub)
3152 else:
3153 expanded.append({**sub, "depends_on": [bind_id, *deps]})
3154 return expanded
3156 def _check_call_guards(
3157 self, step: dict, visible_ids: set[str]
3158 ) -> tuple[dict, list[dict]]:
3159 """Single owner of every build-time ``call`` guard (ADR-0037 D3/D5/D7).
3161 Rejects routes on the call step, function cycles, duplicate expansion,
3162 unknown functions, function-body isolation (purity) violations, and
3163 body-declared routes; returns the resolved function and its body.
3164 """
3165 fn_name = step["function"]
3166 call_id = step["id"]
3168 if step.get("routes"):
3169 raise RunHardFailure(
3170 "VALIDATION_ERROR",
3171 f"call step '{call_id}' declares `routes`; `routes` is not supported "
3172 "on call steps",
3173 )
3174 if fn_name in self._stack:
3175 chain = " -> ".join(self._stack + [fn_name])
3176 raise RunHardFailure(
3177 "VALIDATION_ERROR",
3178 f"cycle detected in function expansion: {chain} "
3179 "(ADR-0037 D7 — recursion must be a bounded while loop or "
3180 "idempotent re-run)",
3181 )
3182 if fn_name in self._expanded_functions:
3183 raise RunHardFailure(
3184 "VALIDATION_ERROR",
3185 f"duplicate call expansion: function '{fn_name}' called at both "
3186 f"'{self._expanded_functions[fn_name]}' and '{call_id}' in one "
3187 "pipeline (ADR-0037 D7 — repeated calls to the same function are "
3188 "not yet supported)",
3189 )
3190 self._expanded_functions[fn_name] = call_id
3192 function = self._registry.get("Function", fn_name)
3193 if function is None:
3194 raise RunHardFailure(
3195 "VALIDATION_ERROR",
3196 f"call step '{call_id}' references unknown function '{fn_name}' "
3197 "(no kind: Function manifest in the workflows tree)",
3198 )
3199 body = (function.get("spec") or {}).get("steps") or []
3200 _validate_function_purity(fn_name, body, visible_ids)
3201 _reject_function_body_routes(fn_name, body)
3202 _validate_unique_step_ids(body)
3203 _validate_foreach_bodies(body)
3204 _validate_loop_limits(body, self._build_ctx)
3205 _validate_fail_if_exhausted_steps(body)
3206 return function, body
3209_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
3210_BINARY_OPERATOR_VALUES = frozenset(
3211 {
3212 "shallow_merge",
3213 "prune_sensitive",
3214 "list_dedup",
3215 "json_merge_patch",
3216 "json_merge_append",
3217 "json_merge_dedup",
3218 }
3219)
3222def _validate_channel_entry(index: int, channel: Any) -> None:
3223 """Validate one spec.channels entry against the frozen ChannelSpec shape.
3225 The engine build path does not run manifest-schema validation, so a
3226 malformed entry is rejected here — before any ``["name"]`` indexing — with
3227 an error that identifies the offending entry.
3228 """
3229 if not isinstance(channel, dict):
3230 raise RunHardFailure(
3231 "CHANNEL_BINDING_ERROR",
3232 f"Pipeline spec.channels[{index}] must be a mapping; got "
3233 f"{type(channel).__name__}",
3234 )
3235 name = channel.get("name")
3236 if not isinstance(name, str) or not name:
3237 raise RunHardFailure(
3238 "CHANNEL_BINDING_ERROR",
3239 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3240 "channel requires a non-empty string 'name'",
3241 )
3242 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3243 raise RunHardFailure(
3244 "CHANNEL_BINDING_ERROR",
3245 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3246 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}",
3247 )
3248 effective_type = channel.get("type", "LastValue")
3249 if "operator" in channel:
3250 operator = channel["operator"]
3251 if not isinstance(operator, str):
3252 raise RunHardFailure(
3253 "CHANNEL_BINDING_ERROR",
3254 f"Pipeline spec.channels[{index}] ('{name}') has a non-string "
3255 f"'operator' of type '{type(operator).__name__}'; 'operator' "
3256 "must be a string",
3257 )
3258 if effective_type != "BinaryOperatorAggregate":
3259 raise RunHardFailure(
3260 "CHANNEL_BINDING_ERROR",
3261 f"Pipeline spec.channels[{index}] ('{name}') sets 'operator' "
3262 "but 'operator' is only valid on a 'BinaryOperatorAggregate' "
3263 f"channel; effective type is '{effective_type}'",
3264 )
3265 if operator not in _BINARY_OPERATOR_VALUES:
3266 raise RunHardFailure(
3267 "CHANNEL_BINDING_ERROR",
3268 f"Pipeline spec.channels[{index}] ('{name}') has unknown "
3269 f"'operator' '{operator}'; expected one of "
3270 f"{sorted(_BINARY_OPERATOR_VALUES)}",
3271 )
3272 elif effective_type == "BinaryOperatorAggregate":
3273 raise RunHardFailure(
3274 "CHANNEL_BINDING_ERROR",
3275 f"Pipeline spec.channels[{index}] ('{name}') is a "
3276 "'BinaryOperatorAggregate' channel and requires an 'operator'; "
3277 f"expected one of {sorted(_BINARY_OPERATOR_VALUES)}",
3278 )
3281def _validated_channels(raw_channels: Any) -> list[dict]:
3282 """Return spec.channels as a validated list, or [] when absent.
3284 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3285 """
3286 if raw_channels is None:
3287 return []
3288 if not isinstance(raw_channels, list):
3289 raise RunHardFailure(
3290 "CHANNEL_BINDING_ERROR",
3291 f"Pipeline spec.channels must be a list; got "
3292 f"{type(raw_channels).__name__}",
3293 )
3294 for index, channel in enumerate(raw_channels):
3295 _validate_channel_entry(index, channel)
3296 return raw_channels
3299def _validate_step_channel_bindings(
3300 steps: list[dict], channel_names: frozenset[str]
3301) -> None:
3302 """Reject malformed or undeclared step-level channel bindings at build time.
3304 Every step's declared ``output_channel`` must be a non-empty declared
3305 channel name, and every declared ``input_channels`` entry must be a
3306 non-empty declared channel name. This runs unconditionally, so a binding
3307 to a channel absent from ``spec.channels`` fails the build even when no
3308 channels are declared.
3309 """
3310 for step in _iter_steps(steps):
3311 _validate_step_output_channel(step, channel_names)
3312 _validate_step_input_channels(step, channel_names)
3315def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3316 channel = step.get("output_channel")
3317 if channel is None:
3318 return
3319 if not isinstance(channel, str) or not channel:
3320 raise RunHardFailure(
3321 "CHANNEL_BINDING_ERROR",
3322 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3323 f"string; got {channel!r}",
3324 )
3325 if channel not in channel_names:
3326 raise RunHardFailure(
3327 "CHANNEL_BINDING_ERROR",
3328 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3329 "which is not declared in spec.channels",
3330 )
3333def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3334 channels = step.get("input_channels")
3335 if channels is None:
3336 return
3337 if not isinstance(channels, list):
3338 raise RunHardFailure(
3339 "CHANNEL_BINDING_ERROR",
3340 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3341 f"channel names; got {type(channels).__name__}",
3342 )
3343 for channel in channels:
3344 if not isinstance(channel, str) or not channel:
3345 raise RunHardFailure(
3346 "CHANNEL_BINDING_ERROR",
3347 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3348 f"channel name '{channel}'; channel names must be non-empty "
3349 "strings",
3350 )
3351 if channel not in channel_names:
3352 raise RunHardFailure(
3353 "CHANNEL_BINDING_ERROR",
3354 f"Step '{step.get('id', '?')}' reads undeclared channel "
3355 f"'{channel}'; the channel must be declared in spec.channels",
3356 )
3359def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3360 """Reject duplicate and engine/step-reserved channel names."""
3361 forbidden = (
3362 {_CORE}
3363 | set(_RESERVED_NAMESPACES)
3364 | _CORE_STATE_KEYS
3365 | _collect_composed_step_ids(steps)
3366 )
3367 seen: set[str] = set()
3368 for channel in channels:
3369 name = channel.get("name")
3370 if name in seen:
3371 raise RunHardFailure(
3372 "CHANNEL_BINDING_ERROR",
3373 f"Pipeline declares channel '{name}' more than once; channel "
3374 "names must be unique",
3375 )
3376 seen.add(name)
3377 if name in forbidden:
3378 raise RunHardFailure(
3379 "CHANNEL_BINDING_ERROR",
3380 f"Pipeline channel '{name}' collides with an engine-reserved "
3381 "name (a step id, engine state key, reserved namespace, or the "
3382 "core state channel); channel names must be unique and "
3383 "non-reserved",
3384 )
3387def _iter_step_lists(steps: list[dict]):
3388 """Yield *steps* and every nested ``body`` list, recursively."""
3389 yield steps
3390 for step in steps:
3391 body = step.get("body")
3392 if isinstance(body, list):
3393 yield from _iter_step_lists(body)
3396def _validate_unique_step_ids(steps: list[dict]) -> None:
3397 """Reject duplicate step ``id``s within the same step list.
3399 Scoped per list: the top-level ``steps`` array and each nested ``body``
3400 array are checked independently, so an id may repeat across a body and its
3401 parent without being flagged. Without this guard a duplicated id is
3402 silently collapsed (last-wins), masking a manifest bug.
3403 """
3404 for step_list in _iter_step_lists(steps):
3405 seen: set[str] = set()
3406 for step in step_list:
3407 step_id = step.get("id")
3408 if not isinstance(step_id, str):
3409 continue
3410 if step_id in seen:
3411 raise RunHardFailure(
3412 "VALIDATION_ERROR",
3413 f"duplicate step id '{step_id}' in the same step list; "
3414 "step ids must be unique within their list",
3415 {"step_id": step_id},
3416 )
3417 seen.add(step_id)
3420def _validate_foreach_bodies(steps: list[dict]) -> None:
3421 """Reject a ``foreach`` with neither a non-empty body nor an agent.
3423 A body-less, agent-less ``foreach`` has no per-item work: the sequential
3424 composition builds an empty agent node and fails at runtime, while the
3425 concurrent paths silently produce an empty aggregate. Reject it at build
3426 time instead.
3427 """
3428 for step in _iter_steps(steps):
3429 if step.get("type") != "foreach":
3430 continue
3431 if step.get("body") or step.get("agent"):
3432 continue
3433 step_id = step.get("id", "?")
3434 raise RunHardFailure(
3435 "VALIDATION_ERROR",
3436 f"foreach step '{step_id}' declares neither a non-empty body nor "
3437 "an agent; per-item work requires one of them",
3438 {"step_id": step_id},
3439 )
3442def _validate_loop_limits(steps: list[dict], build_ctx: dict) -> None:
3443 """Strictly resolve a present ``max_iterations`` on every ``foreach``/``while``.
3445 A concurrent ``foreach`` is not composed into the sequential cycle, so its
3446 ``max_iterations`` would otherwise never be build-validated and a
3447 present-but-unresolvable value would silently degrade to "no cap". This pass
3448 makes the ADR-0012 contract uniform regardless of the composition path.
3449 """
3450 for step in _iter_steps(steps):
3451 if step.get("type") not in ("foreach", "while"):
3452 continue
3453 max_iter_raw = step.get("max_iterations")
3454 if max_iter_raw is None:
3455 continue
3456 _resolve_int_strict(
3457 max_iter_raw, build_ctx, "max_iterations", step.get("id", "?")
3458 )
3461def _validate_fail_if_exhausted(step_id: str, value: Any) -> None:
3462 """Reject a ``fail_if_exhausted`` value that is neither a bool nor a string.
3464 The field is a runtime expression: a bare boolean or a template/condition
3465 string. A list/object/number would otherwise reach ``_eval_condition``,
3466 whose fallback ``bool(resolved)`` silently treats a non-empty list as true.
3467 """
3468 if value is None or isinstance(value, (bool, str)):
3469 return
3470 raise RunHardFailure(
3471 "VALIDATION_ERROR",
3472 f"step '{step_id}' field 'fail_if_exhausted' must be a boolean or a "
3473 f"string; got {type(value).__name__}",
3474 {"step_id": step_id, "field": "fail_if_exhausted"},
3475 )
3478def _validate_fail_if_exhausted_steps(steps: list[dict]) -> None:
3479 """Reject malformed ``fail_if_exhausted`` values across every step list."""
3480 for step in _iter_steps(steps):
3481 _validate_fail_if_exhausted(step.get("id", "?"), step.get("fail_if_exhausted"))
3484def _input_path_root(value: str) -> str | None:
3485 """Leading state key of a bare dot-path input string, or None.
3487 Template strings, bracket-prefixed verbatim literals, and blank strings
3488 are not state reads and yield None.
3489 """
3490 if "{{" in value:
3491 return None
3492 stripped = value.lstrip()
3493 if not stripped or stripped.startswith("["):
3494 return None
3495 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3498def _iter_input_roots(input_spec: Any):
3499 """Yield every bare dot-path root in an input spec, recursing into containers."""
3500 if isinstance(input_spec, str):
3501 root = _input_path_root(input_spec)
3502 if root:
3503 yield root
3504 elif isinstance(input_spec, dict):
3505 for value in input_spec.values():
3506 yield from _iter_input_roots(value)
3507 elif isinstance(input_spec, list):
3508 for value in input_spec:
3509 yield from _iter_input_roots(value)
3512def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3513 """Reject channel-pipeline input roots that name nothing the state exposes.
3515 The channel state is the declared channels plus the engine state, so a bare
3516 dot-path input root must be a declared channel, a known step id, a reserved
3517 namespace, or an engine state key. Anything else is a hard build error —
3518 there is no silent fallback to a missing key.
3519 """
3520 allowed = (
3521 {ch["name"] for ch in channels}
3522 | _collect_step_ids(steps)
3523 | set(_RESERVED_NAMESPACES)
3524 | _CORE_STATE_KEYS
3525 )
3526 for step in _iter_steps(steps):
3527 for root in _iter_input_roots(step.get("input")):
3528 if root not in allowed:
3529 raise RunHardFailure(
3530 "CHANNEL_BINDING_ERROR",
3531 f"Pipeline channel step '{step.get('id', '?')}' input "
3532 f"references unknown root '{root}'; in a channel pipeline "
3533 "an input root must be a declared channel, a step id, a "
3534 "reserved namespace, or an engine state key",
3535 )
3538def _flatten_channel_state(state: dict) -> dict:
3539 """Read the flat committed snapshot nested under the Core namespace."""
3540 return dict(state.get(_CORE) or {})
3543def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3544 """Drop channel-named keys so a node emits only the core state."""
3545 return {k: v for k, v in flat.items() if k not in channel_names}
3548def _project_persisted_step_result(core: dict, step: dict) -> None:
3549 """Strip *step*'s persisted result of its envelope ``content``, in place.
3551 Channel routing reads the in-flight result before this call, so the routed
3552 value is unaffected; only the materialized copy at ``state[step_id]`` loses
3553 its ``content`` payload (ADR-0023 D9 / ADR-0043 D6). Other engine-read
3554 fields (``error``, ``parse_error``, body-item ``verdict``) are retained.
3555 """
3556 step_id = step.get("id")
3557 if step_id is not None and step_id in core:
3558 core[step_id] = project_envelope_metadata(core[step_id])
3561_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3563_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3566def _envelope_content(result: Any) -> Any:
3567 """Return an envelope's optional ``content`` field (ADR-0023 D3).
3569 ``content`` is opaque and optional: a non-mapping result, an absent field,
3570 and an explicit ``null`` all read as ``None``. Callers apply their own
3571 fallback and shape checks.
3572 """
3573 if isinstance(result, dict):
3574 return result.get("content")
3575 return None
3578def _is_failure_envelope(value: Any) -> bool:
3579 """True when a step output is not a successfully parsed dict.
3581 Missing output, non-dict values, and the ``error``/``parse_error``
3582 envelopes written by the LLM failure paths are not routed. A valid
3583 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3584 it carries an explicit ``"error": null``.
3585 """
3586 if not isinstance(value, dict):
3587 return True
3588 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3589 return False
3590 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3593def _resolve_declared_channel(
3594 step: dict,
3595 channel_names: frozenset[str],
3596 *,
3597 allow_default: bool,
3598) -> str | None:
3599 """Resolve a step's single declared output channel, or ``None`` to route nothing.
3601 A step without an ``output_channel`` resolves to the default ``lastvalue``
3602 channel when one exists **and** ``allow_default`` is set; an
3603 ``output_channel`` naming a channel absent from ``spec.channels`` is a hard
3604 error ``CHANNEL_BINDING_ERROR``.
3606 ``allow_default`` is set only on sequential, single-writer execution. The
3607 default channel is not routable inside a superstep (parallel/foreach
3608 fan-out) because its binding resolves only at the superstep boundary: every
3609 superstep staging path keeps the default ``False``. Both the implicit route
3610 and an explicit ``output_channel: lastvalue`` are therefore unrouted there —
3611 ``None`` is returned and a non-fatal advisory is logged (ADR-0040 D5).
3612 """
3613 channel = step.get("output_channel")
3614 if not channel:
3615 if not allow_default:
3616 log.warning(
3617 "[channels] step %s has no output_channel; superstep output is "
3618 "unrouted (advisory, ADR-0040 D5)",
3619 step.get("id", "?"),
3620 )
3621 return None
3622 if "lastvalue" in channel_names:
3623 return "lastvalue"
3624 return None
3625 if channel not in channel_names:
3626 raise RunHardFailure(
3627 "CHANNEL_BINDING_ERROR",
3628 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3629 "which is not declared in spec.channels",
3630 )
3631 if channel == "lastvalue" and not allow_default:
3632 log.warning(
3633 "[channels] step %s declares output_channel 'lastvalue'; superstep "
3634 "output is unrouted (advisory, ADR-0040 D5)",
3635 step.get("id", "?"),
3636 )
3637 return None
3638 return channel
3641def _declared_output_channel_writes(
3642 step: dict,
3643 output: Any,
3644 channel_names: frozenset[str],
3645 *,
3646 allow_default: bool = False,
3647) -> dict[str, Any]:
3648 """Route one agent step output's ``content`` to its declared output channel.
3650 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3651 the working set. Failure envelopes are not routed.
3652 """
3653 if _is_failure_envelope(output):
3654 return {}
3655 channel = _resolve_declared_channel(
3656 step, channel_names, allow_default=allow_default
3657 )
3658 if channel is None:
3659 return {}
3660 return {channel: _envelope_content(output)}
3663def _builtin_output_channel_writes(
3664 step: dict,
3665 output: Any,
3666 channel_names: frozenset[str],
3667 *,
3668 allow_default: bool = False,
3669) -> dict[str, Any]:
3670 """Route one agent-less step's result payload to its declared output channel.
3672 A platform builtin / ``kind: StepFunction`` step stores an arbitrary payload
3673 at ``state[step_id]`` with no envelope, so the routed value is the payload
3674 itself, not its ``content`` field. Builtin failures are raised during
3675 execution, so they are not classified here.
3676 """
3677 channel = _resolve_declared_channel(
3678 step, channel_names, allow_default=allow_default
3679 )
3680 if channel is None:
3681 return {}
3682 return {channel: output}
3685def _route_declared_channels(
3686 step: dict,
3687 pre_state: dict,
3688 core: dict,
3689 step_runner: Any,
3690 channel_names: frozenset[str],
3691 *,
3692 allow_default: bool = False,
3693) -> dict[str, Any]:
3694 """Route a step's declared output channel write to its channel.
3696 Returns ``{channel_name: value}`` for LangGraph to merge through the
3697 channel's reducer, where ``channel_name`` is the step's single declared
3698 ``output_channel``. An actual agent invocation routes the envelope's
3699 ``content``. A ``skip_llm_if`` passthrough routes its resolved input through
3700 the same builtin route as an agent-less step (the payload itself); a parallel
3701 fan-out envelope stays excluded (its per-item outputs are staged elsewhere).
3702 An agent-less step (platform builtin or ``kind: StepFunction``) routes its
3703 result payload itself. A step without an ``output_channel`` routes to the
3704 default ``lastvalue`` channel when one exists **and** ``allow_default`` is
3705 set (otherwise it routes nothing and its output stays in ``core`` untouched).
3707 ``allow_default`` must stay unset on every superstep staging path: a
3708 ``LastValue`` default is forbidden as an in-superstep output target because
3709 its binding resolves only at the superstep boundary (ADR-0040 D4, ADR-0043
3710 D9/D10).
3711 """
3712 step_id = step.get("id", "?")
3713 if step_id not in core:
3714 return {}
3715 output = core[step_id]
3716 if step_runner.resolve_agent(step) is not None:
3717 if step_runner.ran_agent(step, pre_state):
3718 return _declared_output_channel_writes(
3719 step, output, channel_names, allow_default=allow_default
3720 )
3721 if not step_runner._should_skip_llm(
3722 step, step_runner._make_ctx(pre_state)
3723 ):
3724 return {}
3725 return _builtin_output_channel_writes(
3726 step, output, channel_names, allow_default=allow_default
3727 )
3730def _make_step_node(
3731 step: dict,
3732 step_runner: Any,
3733 channel_names: frozenset[str],
3734 channel_specs: dict[str, ChannelSpec],
3735) -> Any:
3736 def node_fn(state: PipelineState) -> PipelineState:
3737 committed = _flatten_channel_state(state)
3738 working_set = ChannelWorkingSet(committed, channel_specs)
3739 core = _strip_channel_keys(
3740 step_runner.run_step(step, committed, working_set), channel_names
3741 )
3742 writes = _route_declared_channels(
3743 step, committed, core, step_runner, channel_names,
3744 allow_default=True,
3745 )
3746 _project_persisted_step_result(core, step)
3747 for name, value in writes.items():
3748 working_set.stage(name, value)
3749 written_names = {name for name, _ in working_set.pending_writes()}
3750 channels = working_set.commit()
3751 emitted = {name: channels[name] for name in written_names}
3752 return {_CORE: {**core, **emitted}}
3753 return node_fn
3756def _filter_composed_step_ids(result: dict, composed_step_ids: Any) -> dict:
3757 """Drop function-body step ids so a call subgraph's internals stay invisible.
3759 ADR-0037 D3: the caller sees only ``state[<call_id>].outputs`` plus
3760 unrelated keys. ``composed_step_ids`` is the build-time set recorded on the
3761 call placeholder. Channel-named keys are disjoint from it (a collision is a
3762 build-time ``CHANNEL_BINDING_ERROR``), so channel state is never touched.
3763 The input is left unmodified.
3764 """
3765 if not composed_step_ids:
3766 return dict(result)
3767 filtered = {
3768 key: value
3769 for key, value in result.items()
3770 if key not in composed_step_ids
3771 }
3772 core = filtered.get(_CORE)
3773 if isinstance(core, dict):
3774 filtered[_CORE] = {
3775 key: value
3776 for key, value in core.items()
3777 if key not in composed_step_ids
3778 }
3779 return filtered
3782def _make_subgraph_node(
3783 compiled: Any,
3784 condition: Any,
3785 step_runner: Any,
3786 channel_names: frozenset[str],
3787 composed_step_ids: Any = None,
3788) -> Any:
3789 """Wrap a compiled call subgraph with the call step's entry condition.
3791 A false condition returns no state updates, so the subgraph never runs and
3792 ``state[<call_id>]`` is not written — matching the soft-skip semantics of a
3793 normal step condition. The ctx is derived exactly like ``run_step``'s,
3794 including channel-mode flattening. Every successful invocation is filtered
3795 through ``_filter_composed_step_ids`` so the function body stays invisible.
3796 """
3797 if condition is None:
3798 def node_fn(state: PipelineState) -> PipelineState:
3799 return _filter_composed_step_ids(
3800 compiled.invoke(state), composed_step_ids
3801 )
3802 return node_fn
3804 def node_fn(state: PipelineState) -> PipelineState:
3805 committed = _flatten_channel_state(state)
3806 if not _eval_condition(condition, step_runner._make_ctx(committed)):
3807 return {}
3808 return _filter_composed_step_ids(
3809 compiled.invoke(state), composed_step_ids
3810 )
3811 return node_fn
3814# ---------------------------------------------------------------------------
3815# Native `while` cycle nodes (ADR-0025 D1/D2)
3816# ---------------------------------------------------------------------------
3817#
3818# A `while` step is a compiled cycle: `head -> body... -> cond` with a
3819# conditional back-edge from `cond` to `head` (continue) or `exit`. The head
3820# owns entry-condition handling and the initial `current`; the cond node owns
3821# the feed, the iteration increment and the exit-condition evaluation; the exit
3822# node writes the documented result and enforces `fail_if_exhausted`. The
3823# nodes read/write the shared graph state, using the `_CORE` container when the
3824# pipeline declares channels.
3827def _while_marker_key(step_id: str) -> str:
3828 """Per-while engine marker key (nested whiles live in isolated cycles)."""
3829 return f"{_WHILE_STATE_KEY}__{step_id}"
3832def _while_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
3833 return _flatten_channel_state(state)
3836def _while_update(update: dict, channel_names: frozenset[str]) -> dict:
3837 return {_CORE: update}
3840def _while_invoke_config(step: dict) -> dict | None:
3841 """Recursion bound for a while/foreach cycle placeholder, else None."""
3842 if step.get(_SUBGRAPH_WHILE_KEY):
3843 return {"recursion_limit": step.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
3844 if step.get(_SUBGRAPH_FOREACH_KEY):
3845 return {"recursion_limit": step.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
3846 return None
3849def _make_while_body_node(
3850 sub: dict,
3851 step_runner: Any,
3852 channel_names: frozenset[str],
3853 channel_specs: dict[str, ChannelSpec],
3854) -> Any:
3855 """A while body step as a node: a nested subgraph or a normal step node."""
3856 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
3857 if sub.get(_SUBGRAPH_WHILE_KEY):
3858 return _make_while_subgraph_node(sub, step_runner, channel_names)
3859 if sub.get(_SUBGRAPH_FOREACH_KEY):
3860 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
3861 return _make_subgraph_node(
3862 sub[_SUBGRAPH_GRAPH_KEY],
3863 sub.get("condition"),
3864 step_runner,
3865 channel_names,
3866 sub.get(_SUBGRAPH_STEP_IDS_KEY),
3867 )
3868 return _make_step_node(sub, step_runner, channel_names, channel_specs)
3871def _make_while_head_node(
3872 step: dict,
3873 step_runner: Any,
3874 channel_names: frozenset[str],
3875 max_iter: int,
3876) -> Any:
3877 """Entry node: handle the entry condition and seed the loop marker."""
3878 step_id = step.get("id", "?")
3879 marker_key = _while_marker_key(step_id)
3881 def node_fn(state: PipelineState) -> PipelineState:
3882 flat = _while_flat(state, channel_names)
3883 marker = flat.get(marker_key)
3884 if marker is not None:
3885 return _while_update({}, channel_names)
3886 ctx = step_runner._make_ctx(flat)
3887 entry_cond = step.get("condition")
3888 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
3889 skip_marker = {
3890 "current": None,
3891 "iteration": 0,
3892 "skipped": True,
3893 "max_iter": max_iter,
3894 }
3895 return _while_update(
3896 {
3897 step_id: {
3898 "iterations": 0,
3899 "output": _resolved_while_input(step, flat, ctx),
3900 "exhausted": False,
3901 "skipped": True,
3902 },
3903 marker_key: skip_marker,
3904 },
3905 channel_names,
3906 )
3907 marker = {
3908 "current": _resolved_while_input(step, flat, ctx),
3909 "iteration": 0,
3910 "max_iter": max_iter,
3911 }
3912 return _while_update({marker_key: marker}, channel_names)
3914 return node_fn
3917def _resolved_while_input(step: dict, flat: dict, ctx: dict) -> Any:
3918 input_spec = step.get("input")
3919 if input_spec is None:
3920 return None
3921 return _resolve_input(input_spec, flat, ctx)
3924def _make_while_head_router(
3925 step: dict,
3926 channel_names: frozenset[str],
3927 first_id: str,
3928 exit_id: str,
3929 end: Any,
3930) -> Any:
3931 """Route from the head: skipped → end, bound already reached → exit, else body."""
3932 marker_key = _while_marker_key(step.get("id", "?"))
3934 def router(state: PipelineState) -> str:
3935 marker = _while_flat(state, channel_names).get(marker_key) or {}
3936 if marker.get("skipped"):
3937 return end
3938 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
3939 return exit_id
3940 return first_id
3942 return router
3945def _make_while_cond_node(
3946 step: dict,
3947 step_runner: Any,
3948 channel_names: frozenset[str],
3949) -> Any:
3950 """Increment the iteration and evaluate the exit condition."""
3951 step_id = step.get("id", "?")
3952 marker_key = _while_marker_key(step_id)
3953 exit_condition = step.get("exit_condition")
3955 def node_fn(state: PipelineState) -> PipelineState:
3956 flat = _while_flat(state, channel_names)
3957 marker = dict(flat.get(marker_key) or {})
3958 passed = False
3959 if exit_condition:
3960 if _eval_condition(
3961 exit_condition, step_runner._make_ctx(flat)
3962 ):
3963 passed = True
3964 else:
3965 passed = True
3966 marker.update(
3967 {
3968 "iteration": marker.get("iteration", 0) + 1,
3969 "passed": passed,
3970 }
3971 )
3972 return _while_update({marker_key: marker}, channel_names)
3974 return node_fn
3977def _make_while_cond_router(
3978 step: dict,
3979 channel_names: frozenset[str],
3980 head_id: str,
3981 exit_id: str,
3982) -> Any:
3983 """Route from the cond: exit on pass or when the bound is reached, else head."""
3984 marker_key = _while_marker_key(step.get("id", "?"))
3986 def router(state: PipelineState) -> str:
3987 marker = _while_flat(state, channel_names).get(marker_key) or {}
3988 if marker.get("passed"):
3989 return exit_id
3990 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
3991 return exit_id
3992 return head_id
3994 return router
3997def _while_result_output(step: dict, flat: dict, marker: dict) -> Any:
3998 """The loop result's `output`: the declared body step's result, else the input.
4000 `while.output: <body_step_id>` names the body step whose result fills the
4001 loop result's `output` field; it does not feed the next round. Without a
4002 declared output the loop's resolved input (the marker current) is surfaced.
4003 """
4004 declared_output = step.get("output")
4005 if not declared_output:
4006 return marker.get("current")
4007 value = flat.get(declared_output)
4008 if isinstance(value, dict) and "output" in value:
4009 return value["output"]
4010 if isinstance(value, dict) and isinstance(value.get("results"), list):
4011 return value["results"]
4012 return value
4015def _make_while_exit_node(
4016 step: dict,
4017 step_runner: Any,
4018 channel_names: frozenset[str],
4019) -> Any:
4020 """Write the documented while result and enforce `fail_if_exhausted`."""
4021 step_id = step.get("id", "?")
4022 marker_key = _while_marker_key(step_id)
4023 fail_if_exhausted = step.get("fail_if_exhausted")
4024 _validate_fail_if_exhausted(step_id, fail_if_exhausted)
4026 def node_fn(state: PipelineState) -> PipelineState:
4027 flat = _while_flat(state, channel_names)
4028 marker = flat.get(marker_key) or {}
4029 passed = bool(marker.get("passed", False))
4030 exhausted = not passed
4031 if (
4032 fail_if_exhausted is not None
4033 and exhausted
4034 and _eval_condition(
4035 fail_if_exhausted, step_runner._make_ctx(flat)
4036 )
4037 ):
4038 raise RunHardFailure(
4039 "EXHAUSTED",
4040 f"while step '{step_id}' exhausted "
4041 f"max_iterations={marker.get('max_iter')} without "
4042 "exit_condition passing (fail_if_exhausted)",
4043 {"step_id": step_id, "exhausted": True},
4044 )
4045 return _while_update(
4046 {
4047 step_id: {
4048 "iterations": marker.get("iteration", 0),
4049 "output": _while_result_output(step, flat, marker),
4050 "exhausted": exhausted,
4051 "skipped": False,
4052 }
4053 },
4054 channel_names,
4055 )
4057 return node_fn
4060def _make_while_subgraph_node(
4061 placeholder: dict,
4062 step_runner: Any,
4063 channel_names: frozenset[str],
4064) -> Any:
4065 """Parent node for a compiled `while` cycle: isolate and surface its result.
4067 The cycle shares the graph state schema, so invoking it directly would leak
4068 the body step results and loop plumbing into the parent. This wrapper runs
4069 the cycle on a copy and returns only the documented result plus the side
4070 channels and any channel values the body committed, preserving the in-node
4071 loop's state contract exactly.
4072 """
4073 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4074 step_id = placeholder["id"]
4075 config = {"recursion_limit": placeholder.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
4077 def node_fn(state: PipelineState) -> PipelineState:
4078 sub = compiled.invoke(state, config)
4079 flat = _flatten_channel_state(sub)
4080 parent_flat = _flatten_channel_state(state)
4081 result: dict = {}
4082 if step_id in flat:
4083 result[step_id] = flat.get(step_id)
4084 for key in _WHILE_SIDE_CHANNEL_KEYS:
4085 if key in flat and flat.get(key) != parent_flat.get(key):
4086 result[key] = flat.get(key)
4087 for name in channel_names:
4088 if name in flat and (
4089 name not in parent_flat or flat.get(name) != parent_flat.get(name)
4090 ):
4091 result[name] = flat.get(name)
4092 if not result:
4093 return {}
4094 return {_CORE: result}
4096 return node_fn
4099# ---------------------------------------------------------------------------
4100# Native sequential `foreach` cycle nodes
4101# ---------------------------------------------------------------------------
4102#
4103# A sequential `foreach` is a compiled cycle: `head -> (body... | agent) ->
4104# cond` with a conditional back-edge from `cond` to `head` (next item) or
4105# `exit` (`max_output` reached or the item list exhausted). The head resolves
4106# the item list once and injects the current item (`item` / `_item` /
4107# `_foreach_item`); the cond collects the item result; the exit node writes the
4108# documented aggregate (`{"results", "passed", "failing", "failed"}` for the
4109# body path, the plain list for the agent path). Nodes read/write the shared
4110# graph state, using the `_CORE` container when the pipeline declares channels.
4113def _foreach_marker_key(step_id: str) -> str:
4114 """Per-foreach engine marker key (nested foreachs live in isolated cycles)."""
4115 return f"{_FOREACH_STATE_KEY}__{step_id}"
4118def _foreach_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
4119 return _flatten_channel_state(state)
4122def _foreach_update(update: dict, channel_names: frozenset[str]) -> dict:
4123 return {_CORE: update}
4126def _make_foreach_head_node(
4127 step: dict,
4128 step_runner: Any,
4129 channel_names: frozenset[str],
4130) -> Any:
4131 """Entry node: evaluate the entry condition, resolve items, inject the item."""
4132 step_id = step.get("id", "?")
4133 marker_key = _foreach_marker_key(step_id)
4135 def node_fn(state: PipelineState) -> PipelineState:
4136 flat = _foreach_flat(state, channel_names)
4137 marker = flat.get(marker_key)
4138 if marker is None:
4139 ctx = step_runner._make_ctx(flat)
4140 entry_cond = step.get("condition")
4141 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4142 return _foreach_update({}, channel_names)
4143 max_output_raw = step.get("max_output")
4144 marker = {
4145 "items": step_runner._foreach_item_list(step, ctx),
4146 "index": 0,
4147 "results": [],
4148 "produced_outputs": 0,
4149 "max_output": (
4150 _resolve_int(max_output_raw, ctx, 0)
4151 if max_output_raw is not None
4152 else 0
4153 ),
4154 }
4155 items = marker.get("items") or []
4156 index = marker.get("index", 0)
4157 if index < len(items):
4158 marker = {**marker, "publish_before": flat.get("publish")}
4159 update = {
4160 "_item": items[index],
4161 "item": items[index],
4162 "_foreach_item": items[index],
4163 }
4164 update[marker_key] = marker
4165 return _foreach_update(update, channel_names)
4166 return _foreach_update({marker_key: marker}, channel_names)
4168 return node_fn
4171def _make_foreach_head_router(
4172 step: dict,
4173 channel_names: frozenset[str],
4174 first_id: str,
4175 exit_id: str,
4176 end: Any,
4177) -> Any:
4178 """Route from the head: condition-skipped → end, exhausted/budget → exit, else body."""
4179 marker_key = _foreach_marker_key(step.get("id", "?"))
4181 def router(state: PipelineState) -> str:
4182 marker = _foreach_flat(state, channel_names).get(marker_key)
4183 if marker is None:
4184 return end
4185 items = marker.get("items") or []
4186 if marker.get("index", 0) >= len(items):
4187 return exit_id
4188 max_output = marker.get("max_output", 0)
4189 if max_output and marker.get("produced_outputs", 0) >= max_output:
4190 return exit_id
4191 return first_id
4193 return router
4196def _make_foreach_body_node(
4197 sub: dict,
4198 step_runner: Any,
4199 channel_names: frozenset[str],
4200 channel_specs: dict[str, ChannelSpec],
4201) -> Any:
4202 """A foreach body step as a node: a nested loop subgraph or a step node."""
4203 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
4204 if sub.get(_SUBGRAPH_WHILE_KEY):
4205 return _make_while_subgraph_node(sub, step_runner, channel_names)
4206 if sub.get(_SUBGRAPH_FOREACH_KEY):
4207 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
4208 return _make_subgraph_node(
4209 sub[_SUBGRAPH_GRAPH_KEY],
4210 sub.get("condition"),
4211 step_runner,
4212 channel_names,
4213 sub.get(_SUBGRAPH_STEP_IDS_KEY),
4214 )
4215 return _make_step_node(sub, step_runner, channel_names, channel_specs)
4218def _foreach_item_channel_updates(
4219 writes: dict[str, Any],
4220 committed: dict,
4221 channel_specs: dict[str, ChannelSpec],
4222) -> dict[str, Any]:
4223 """Fold one foreach item's declared channel writes over the snapshot.
4225 Returns the full accumulated channel values to merge into the core state,
4226 so a ``Topic`` grows one item at a time in item order — mirroring
4227 ``_stage_fanout_channel_writes``. The committed snapshot is never mutated.
4228 """
4229 if not writes:
4230 return {}
4231 working_set = ChannelWorkingSet(committed, channel_specs)
4232 for name, value in writes.items():
4233 working_set.stage(name, value)
4234 accumulated = working_set.commit()
4235 return {name: accumulated[name] for name in writes}
4238def _make_foreach_agent_node(
4239 step: dict,
4240 step_runner: Any,
4241 channel_names: frozenset[str],
4242 channel_specs: dict[str, ChannelSpec],
4243) -> Any:
4244 """Agent-path node: invoke the foreach step's agent for the current item.
4246 Each item's envelope ``content`` is written to the step's declared output
4247 channel as its own write, in item order. This is a superstep fan-out
4248 staging path, so the implicit default channel is forbidden: ``allow_default``
4249 stays unset.
4250 """
4251 step_id = step.get("id", "?")
4253 def node_fn(state: PipelineState) -> PipelineState:
4254 flat = _foreach_flat(state, channel_names)
4255 resolved = step_runner.resolve_agent(step)
4256 ctx = step_runner._make_ctx(flat)
4257 resolved_step = step_runner._resolve_step_runtime(step, flat, ctx)
4258 resolved_step = step_runner._attach_communication_context(resolved_step, flat)
4259 resolved_step = step_runner._attach_resolved_tools(resolved_step, resolved)
4260 result_state = step_runner._invoke_agent(resolved, resolved_step, flat)
4261 core = _strip_channel_keys(result_state, channel_names)
4262 writes = _declared_output_channel_writes(
4263 step, core.get(step_id), channel_names
4264 )
4265 updates = _foreach_item_channel_updates(writes, flat, channel_specs)
4266 _project_persisted_step_result(core, step)
4267 return {_CORE: {**core, **updates}}
4269 return node_fn
4272def _make_foreach_cond_node(
4273 step: dict,
4274 step_runner: Any,
4275 channel_names: frozenset[str],
4276) -> Any:
4277 """Collect the current item's result and advance the item index."""
4278 step_id = step.get("id", "?")
4279 marker_key = _foreach_marker_key(step_id)
4280 declared_output = step.get("output")
4281 runs_body = bool(step.get("body"))
4283 def node_fn(state: PipelineState) -> PipelineState:
4284 flat = _foreach_flat(state, channel_names)
4285 marker = dict(flat.get(marker_key) or {})
4286 results = list(marker.get("results") or [])
4287 produced = marker.get("produced_outputs", 0)
4288 if runs_body:
4289 results.append(StepRunner._body_item_result(flat, declared_output))
4290 publish_out = flat.get("publish")
4291 if (
4292 isinstance(publish_out, dict)
4293 and publish_out.get("path")
4294 and publish_out != marker.get("publish_before")
4295 ):
4296 produced += 1
4297 else:
4298 results.append(flat.get(step_id))
4299 marker.update(
4300 {
4301 "results": results,
4302 "index": marker.get("index", 0) + 1,
4303 "produced_outputs": produced,
4304 }
4305 )
4306 return _foreach_update({marker_key: marker}, channel_names)
4308 return node_fn
4311def _make_foreach_cond_router(
4312 step: dict,
4313 channel_names: frozenset[str],
4314 head_id: str,
4315 exit_id: str,
4316) -> Any:
4317 """Route from the cond: exit on exhaustion or budget, else back to the head."""
4318 marker_key = _foreach_marker_key(step.get("id", "?"))
4320 def router(state: PipelineState) -> str:
4321 marker = _foreach_flat(state, channel_names).get(marker_key) or {}
4322 items = marker.get("items") or []
4323 if marker.get("index", 0) >= len(items):
4324 return exit_id
4325 max_output = marker.get("max_output", 0)
4326 if max_output and marker.get("produced_outputs", 0) >= max_output:
4327 return exit_id
4328 return head_id
4330 return router
4333def _make_foreach_exit_node(
4334 step: dict,
4335 step_runner: Any,
4336 channel_names: frozenset[str],
4337) -> Any:
4338 """Write the documented foreach aggregate result."""
4339 step_id = step.get("id", "?")
4340 marker_key = _foreach_marker_key(step_id)
4341 runs_body = bool(step.get("body"))
4343 def node_fn(state: PipelineState) -> PipelineState:
4344 flat = _foreach_flat(state, channel_names)
4345 marker = flat.get(marker_key) or {}
4346 results = list(marker.get("results") or [])
4347 step_result = (
4348 StepRunner._aggregate_body_results(results)
4349 if runs_body
4350 else [project_envelope_metadata(r) for r in results]
4351 )
4352 return _foreach_update({step_id: step_result}, channel_names)
4354 return node_fn
4357def _make_foreach_subgraph_node(
4358 placeholder: dict,
4359 step_runner: Any,
4360 channel_names: frozenset[str],
4361) -> Any:
4362 """Parent node for a compiled sequential `foreach` cycle: isolate and surface.
4364 The cycle shares the graph state schema, so invoking it directly would leak
4365 body step results and loop plumbing into the parent. This wrapper runs the
4366 cycle on a copy and returns only the documented aggregate plus the side
4367 channels and any channel values the body committed. A condition-skipped
4368 entry writes no step result (matching the in-node soft-skip), so the step id
4369 is only surfaced when the cycle wrote it.
4370 """
4371 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4372 step_id = placeholder["id"]
4373 config = {"recursion_limit": placeholder.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
4375 def node_fn(state: PipelineState) -> PipelineState:
4376 sub = compiled.invoke(state, config)
4377 flat = _flatten_channel_state(sub)
4378 parent_flat = _flatten_channel_state(state)
4379 result: dict = {}
4380 if step_id in flat:
4381 result[step_id] = flat.get(step_id)
4382 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4383 if key in flat and flat.get(key) != parent_flat.get(key):
4384 result[key] = flat.get(key)
4385 for name in channel_names:
4386 if name in flat and (
4387 name not in parent_flat or flat.get(name) != parent_flat.get(name)
4388 ):
4389 result[name] = flat.get(name)
4390 if not result:
4391 return {}
4392 return {_CORE: result}
4394 return node_fn
4397def _collect_parallel_channel_writes(
4398 steps: list[dict],
4399 pre_state: dict,
4400 core: dict,
4401 step_runner: Any,
4402 channel_names: frozenset[str],
4403 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
4404) -> list[tuple[str, Any]]:
4405 """Collect one parallel node's declared writes in branch order.
4407 Every branch write is emitted, so fan-in feeds each channel's reducer every
4408 write in deterministic declaration/branch order: Topic appends, LastValue
4409 keeps the last, and BinaryOperatorAggregate routes each write through the
4410 reducer selected by name from its mandatory manifest ``operator`` — never
4411 inferred from the accumulated value's shape. A branch's isolated body
4412 writes are emitted before its own declared write, mirroring the top-level
4413 node's route-after-run order. This is a superstep fan-in path, so the
4414 implicit default channel is forbidden: ``allow_default`` stays unset on the
4415 shared router.
4416 """
4417 writes: list[tuple[str, Any]] = []
4418 for step in steps:
4419 if branch_working_sets is not None:
4420 branch_working_set = branch_working_sets.get(step["id"])
4421 if branch_working_set is not None:
4422 writes.extend(branch_working_set.pending_writes())
4423 branch = _route_declared_channels(
4424 step, pre_state, core, step_runner, channel_names
4425 )
4426 for channel_name, value in branch.items():
4427 writes.append((channel_name, value))
4428 return writes
4431def _make_parallel_node(
4432 group_steps: list[dict],
4433 step_runner: Any,
4434 channel_names: frozenset[str],
4435 channel_specs: dict[str, ChannelSpec],
4436) -> Any:
4437 def node_fn(state: PipelineState) -> PipelineState:
4438 base = _flatten_channel_state(state)
4439 working_set = ChannelWorkingSet(base, channel_specs)
4440 # Each branch writes into its own fork; the shared set is advanced only
4441 # at fan-in, so branch-internal body writes never race.
4442 branch_working_sets = {
4443 s["id"]: working_set.fork() for s in group_steps
4444 }
4446 def submit_branch(pool: Any, step: dict) -> Any:
4447 return pool.submit(
4448 step_runner.run_step, step, base, branch_working_sets[step["id"]]
4449 )
4451 results: dict[str, Any] = {}
4452 branch_metrics: dict = {}
4453 branch_tool_meta: dict = {}
4454 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
4455 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
4456 for future in as_completed(futures):
4457 sid = futures[future]
4458 try:
4459 out = future.result()
4460 results[sid] = out.get(sid)
4461 branch_metrics.update(out.get("step_context_metrics") or {})
4462 branch_tool_meta.update(out.get("step_tool_meta") or {})
4463 except SkillLoadError:
4464 raise
4465 except RunHardFailure:
4466 pool.shutdown(wait=False, cancel_futures=True)
4467 raise
4468 except Exception as exc: # noqa: BLE001
4469 log.error("Step '%s' failed: %s", sid, exc)
4470 results[sid] = {"error": str(exc)}
4471 merged = _strip_channel_keys({**base, **results}, channel_names)
4472 if branch_metrics:
4473 merged["step_context_metrics"] = _merge_side_channel(
4474 merged.get("step_context_metrics"), branch_metrics
4475 )
4476 if branch_tool_meta:
4477 merged["step_tool_meta"] = _merge_side_channel(
4478 merged.get("step_tool_meta"), branch_tool_meta
4479 )
4480 writes = _collect_parallel_channel_writes(
4481 group_steps, base, merged, step_runner, channel_names,
4482 branch_working_sets,
4483 )
4484 for branch_step in group_steps:
4485 _project_persisted_step_result(merged, branch_step)
4486 written_names = {name for name, _ in writes}
4487 for name, value in writes:
4488 working_set.stage(name, value)
4489 channels = working_set.commit()
4490 emitted = {name: channels[name] for name in written_names}
4491 return {_CORE: {**merged, **emitted}}
4492 return node_fn
4495@dataclass(frozen=True)
4496class _MapReduceParts:
4497 """The nodes and edges of one concurrent step's Send fan-out.
4499 ``dispatch_id`` reuses the step id so the step's incoming dependency edges are
4500 unchanged; ``join_id`` carries the step's outgoing edges.
4501 """
4503 dispatch_id: str
4504 worker_id: str
4505 fallback_id: str
4506 join_id: str
4507 dispatch_fn: Any
4508 route_fn: Any
4509 worker_fn: Any
4510 join_fn: Any
4513@dataclass(frozen=True)
4514class _MapReduceItemOutcome:
4515 """One worker's per-item result plus its errors, side channels and writes.
4517 ``errors`` carries the item's soft errors (and the inherited soft errors it
4518 read), so the join can fold them in item order exactly like the in-node
4519 concurrent path. ``channel_writes`` is empty for the parallel strategy (its
4520 channel writes are derived from the item result at the join) and carries the
4521 foreach body's pending writes so the join can stage them in item order.
4522 """
4524 result: Any
4525 metrics: dict | None
4526 tool_meta: dict | None
4527 channel_writes: tuple[tuple[str, Any], ...]
4528 errors: tuple[str, ...] = ()
4531@dataclass(frozen=True)
4532class _MapReducePlan:
4533 """The per-step behaviour injected into the shared Send machinery.
4535 The parallel strategy runs a single agent call per item; the foreach
4536 strategy runs the body (or agent) per item. ``item_lister`` resolves the
4537 fan-out list, ``item_runner`` executes one item, ``result_assembler``
4538 shapes the step result, and ``channel_write_stager`` stages per-item
4539 channel writes at the join.
4540 """
4542 item_lister: Any
4543 item_runner: Any
4544 result_assembler: Any
4545 channel_write_stager: Any
4548def _map_reduce_worker_payload(flat: dict, item: Any, index: int) -> dict:
4549 """Build a worker's ``Send`` payload from the resolved source state."""
4550 return {**flat, "item": item, "_item": item, _MAP_REDUCE_ITEM_INDEX: index}
4553def _map_reduce_worker_output(
4554 step_id: str,
4555 index: int,
4556 outcome: _MapReduceItemOutcome,
4557) -> PipelineState:
4558 """The accumulator record carrying every per-item fold input.
4560 The slot is ``{index: {"result", "writes", "errors", "metrics",
4561 "tool_meta"}}`` so the join can assemble the step result, stage channel
4562 writes, and fold soft errors plus side channels in item order — the same
4563 observable result as the in-node concurrent path.
4564 """
4565 record = {
4566 "result": outcome.result,
4567 "writes": list(outcome.channel_writes),
4568 "errors": list(outcome.errors),
4569 "metrics": outcome.metrics if isinstance(outcome.metrics, dict) else None,
4570 "tool_meta": (
4571 outcome.tool_meta if isinstance(outcome.tool_meta, dict) else None
4572 ),
4573 }
4574 return {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: {index: record}}}
4577def _make_map_reduce_route_fn(
4578 plan: _MapReducePlan,
4579 step_runner: Any,
4580 channel_names: frozenset[str],
4581 worker_id: str,
4582 fallback_id: str,
4583 join_id: str,
4584) -> Any:
4585 def route(state: PipelineState) -> list:
4586 flat = _flatten_channel_state(state)
4587 ctx = step_runner._make_ctx(flat)
4588 items = plan.item_lister(flat, ctx)
4589 if items is None:
4590 return [Send(fallback_id, state)]
4591 if not items:
4592 return [Send(join_id, state)]
4593 return [
4594 Send(worker_id, _map_reduce_worker_payload(flat, item, index))
4595 for index, item in enumerate(items)
4596 ]
4598 return route
4601def _make_map_reduce_worker_fn(
4602 step: dict,
4603 plan: _MapReducePlan,
4604 step_runner: Any,
4605 gate: threading.Semaphore,
4606) -> Any:
4607 step_id = step.get("id", "?")
4609 def worker(state: PipelineState) -> PipelineState:
4610 index = state[_MAP_REDUCE_ITEM_INDEX]
4611 item_state = {
4612 key: value
4613 for key, value in state.items()
4614 if key != _MAP_REDUCE_ITEM_INDEX
4615 }
4616 item = item_state.get("item")
4617 item_ctx = step_runner._make_ctx(item_state)
4618 with gate:
4619 try:
4620 outcome = plan.item_runner(
4621 state, index, item, item_state, item_ctx
4622 )
4623 except SkillLoadError:
4624 raise
4625 except RunHardFailure:
4626 raise
4627 except Exception as exc: # noqa: BLE001
4628 log.error("[step] %s item failed: %s", step_id, exc)
4629 outcome = _MapReduceItemOutcome(
4630 {"error": str(exc)}, None, None, ()
4631 )
4632 return _map_reduce_worker_output(step_id, index, outcome)
4634 return worker
4637def _fold_map_reduce_errors(records: list, base_errors: list) -> list:
4638 """Fold every item's soft errors into *base_errors* in item order."""
4639 merged = list(base_errors)
4640 for record in records:
4641 if not isinstance(record, dict):
4642 continue
4643 merged = StepRunner._merge_errors(merged, record.get("errors") or [])
4644 return merged
4647def _fold_map_reduce_side_channel(records: list, record_key: str) -> dict:
4648 """Fold every item's side-channel map in item order (later item wins)."""
4649 folded: dict = {}
4650 for record in records:
4651 if not isinstance(record, dict):
4652 continue
4653 value = record.get(record_key)
4654 if isinstance(value, dict):
4655 folded.update(value)
4656 return folded
4659def _fold_map_reduce_extras(flat: dict, records: list) -> dict:
4660 """The join's per-item state extras: soft errors plus folded side channels.
4662 Mirrors the in-node concurrent path: errors are merged in item order with
4663 dedup, and the budget/side-channel counters follow the same item order.
4664 """
4665 extras: dict = {}
4666 base_errors = list(flat.get("errors") or [])
4667 errors = _fold_map_reduce_errors(records, base_errors)
4668 if errors != base_errors:
4669 extras["errors"] = errors
4670 for state_key, record_key in (
4671 ("step_context_metrics", "metrics"),
4672 ("step_tool_meta", "tool_meta"),
4673 ):
4674 folded = _fold_map_reduce_side_channel(records, record_key)
4675 if folded:
4676 extras[state_key] = _merge_side_channel(flat.get(state_key), folded)
4677 return extras
4680def _make_map_reduce_join_fn(
4681 step: dict,
4682 plan: _MapReducePlan,
4683 step_runner: Any,
4684 channel_names: frozenset[str],
4685 channel_specs: dict[str, ChannelSpec],
4686) -> Any:
4687 step_id = step.get("id", "?")
4689 def join(state: PipelineState) -> PipelineState:
4690 flat = _flatten_channel_state(state)
4691 accumulated = state.get(_MAP_REDUCE_ACCUMULATOR_KEY) or {}
4692 entries = accumulated.get(step_id) or {}
4693 clear = {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: None}}
4694 ctx = step_runner._make_ctx(flat)
4695 items = plan.item_lister(flat, ctx)
4696 if items is None:
4697 return clear
4698 records = [entries.get(index) for index in range(len(items))]
4699 results = [
4700 record.get("result") if isinstance(record, dict) else None
4701 for record in records
4702 ]
4703 step_result = plan.result_assembler(results)
4704 extras = _fold_map_reduce_extras(flat, records)
4705 working_set = ChannelWorkingSet(flat, channel_specs)
4706 plan.channel_write_stager(results, records, working_set)
4707 written_names = {name for name, _ in working_set.pending_writes()}
4708 channels = working_set.commit()
4709 emitted = {name: channels[name] for name in written_names}
4710 core = _strip_channel_keys(
4711 {**flat, step_id: step_result, **extras}, channel_names
4712 )
4713 _project_persisted_step_result(core, step)
4714 return {_CORE: {**core, **emitted}, **clear}
4716 return join
4719def _make_parallel_map_reduce_plan(
4720 step: dict, step_runner: Any
4721) -> _MapReducePlan:
4722 """Plan for a concurrent `parallel` step: one agent call per item."""
4723 resolved_agent = step_runner.resolve_agent(step)
4725 def run_item(state, index, item, item_state, item_ctx):
4726 result, metrics, tool_meta, errors = step_runner._invoke_parallel_item(
4727 step, resolved_agent, item, item_state, item_ctx
4728 )
4729 return _MapReduceItemOutcome(
4730 result, metrics, tool_meta, (), tuple(errors)
4731 )
4733 return _MapReducePlan(
4734 item_lister=lambda _flat, ctx: step_runner._parallel_fanout_item_list(
4735 step, ctx
4736 ),
4737 item_runner=run_item,
4738 result_assembler=lambda results: project_aggregate_metadata(
4739 {"results": results}
4740 ),
4741 channel_write_stager=lambda results, records, working_set: (
4742 step_runner._stage_fanout_channel_writes(step, results, working_set)
4743 ),
4744 )
4747def _worker_committed_flat(
4748 state: PipelineState, channel_names: frozenset[str]
4749) -> dict:
4750 """The committed channel snapshot visible to one worker.
4752 A channels-path worker receives the flattened snapshot as its payload, so
4753 its channel keys sit at the top level; the typed ``__core__`` shape (only
4754 seen by the join) is unwrapped.
4755 """
4756 if _CORE in state:
4757 return _flatten_channel_state(state)
4758 return {name: state[name] for name in channel_names if name in state}
4761def _pending_channel_writes(
4762 working_set: ChannelWorkingSet,
4763) -> tuple[tuple[str, Any], ...]:
4764 return tuple(working_set.pending_writes())
4767def _stage_map_reduce_item_channel_writes(
4768 results: list, records: list, working_set: ChannelWorkingSet
4769) -> None:
4770 """Stage each item's carried channel writes in item order."""
4771 for record in records:
4772 if not isinstance(record, dict):
4773 continue
4774 for name, value in record.get("writes") or ():
4775 working_set.stage(name, value)
4778def _make_foreach_map_reduce_plan(
4779 step: dict,
4780 step_runner: Any,
4781 channel_names: frozenset[str],
4782 channel_specs: dict[str, ChannelSpec],
4783) -> _MapReducePlan:
4784 """Plan for a concurrent `foreach` step: the body (or agent) runs per item."""
4785 step_id = step.get("id", "?")
4786 body = step.get("body") or []
4787 declared_output = step.get("output")
4788 runs_body = bool(body)
4790 def result_assembler(results):
4791 if runs_body:
4792 return StepRunner._aggregate_body_results(results)
4793 return [project_envelope_metadata(r) for r in results]
4795 def run_item(state, index, item, item_state, _item_ctx):
4796 item_working_set = ChannelWorkingSet(
4797 _worker_committed_flat(state, channel_names), channel_specs
4798 )
4799 iter_state = {
4800 **item_state, "item": item, "_item": item, "_foreach_item": item
4801 }
4802 if runs_body:
4803 for sub_step in body:
4804 iter_state = step_runner._run_body_step(
4805 sub_step, iter_state, item_working_set
4806 )
4807 return _MapReduceItemOutcome(
4808 step_runner._body_item_result(iter_state, declared_output),
4809 iter_state.get("step_context_metrics"),
4810 iter_state.get("step_tool_meta"),
4811 _pending_channel_writes(item_working_set),
4812 tuple(iter_state.get("errors") or []),
4813 )
4814 resolved = step_runner.resolve_agent(step)
4815 item_read_state = step_runner._read_state(iter_state, item_working_set)
4816 item_ctx = step_runner._make_ctx(item_read_state)
4817 resolved_step = step_runner._resolve_step_runtime(
4818 step, item_read_state, item_ctx
4819 )
4820 resolved_step = step_runner._attach_communication_context(
4821 resolved_step, item_read_state
4822 )
4823 resolved_step = step_runner._attach_resolved_tools(
4824 resolved_step, resolved
4825 )
4826 result_state = step_runner._invoke_agent(
4827 resolved, resolved_step, iter_state
4828 )
4829 return _MapReduceItemOutcome(
4830 result_state.get(step_id),
4831 result_state.get("step_context_metrics"),
4832 result_state.get("step_tool_meta"),
4833 _pending_channel_writes(item_working_set),
4834 tuple(result_state.get("errors") or []),
4835 )
4837 return _MapReducePlan(
4838 item_lister=lambda _flat, ctx: step_runner._foreach_item_list(
4839 step, ctx
4840 ),
4841 item_runner=run_item,
4842 result_assembler=result_assembler,
4843 channel_write_stager=_stage_map_reduce_item_channel_writes,
4844 )
4847def _make_map_reduce_parts(
4848 step: dict,
4849 plan: _MapReducePlan,
4850 step_runner: Any,
4851 channel_names: frozenset[str],
4852 channel_specs: dict[str, ChannelSpec],
4853 max_concurrent: int,
4854) -> _MapReduceParts:
4855 step_id = step.get("id", "?")
4856 worker_id = f"{step_id}__worker"
4857 fallback_id = f"{step_id}__fallback"
4858 join_id = f"{step_id}__join"
4859 gate = threading.Semaphore(max(1, max_concurrent))
4860 return _MapReduceParts(
4861 dispatch_id=step_id,
4862 worker_id=worker_id,
4863 fallback_id=fallback_id,
4864 join_id=join_id,
4865 dispatch_fn=lambda state: {},
4866 route_fn=_make_map_reduce_route_fn(
4867 plan, step_runner, channel_names, worker_id, fallback_id, join_id
4868 ),
4869 worker_fn=_make_map_reduce_worker_fn(step, plan, step_runner, gate),
4870 join_fn=_make_map_reduce_join_fn(
4871 step, plan, step_runner, channel_names, channel_specs
4872 ),
4873 )
4876def _resolved_send_concurrency(step: dict, build_ctx: dict) -> int:
4877 """The build-time resolved ``max_concurrent`` for a Send-eligible step."""
4878 raw = step.get("max_concurrent")
4879 if raw is None:
4880 return 1
4881 return _resolve_int_strict(
4882 raw, build_ctx, "max_concurrent", step.get("id", "?"), minimum=1
4883 )
4886def _is_send_parallel_step(
4887 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4888) -> bool:
4889 """Whether a top-level `parallel` step fans out through native ``Send``.
4891 Restricted to a singleton group's statically concurrent, agent-backed,
4892 item-sourced step with no `routes`, no `condition`/`skip_llm_if`, and an
4893 explicit `max_concurrent > 1`; everything else keeps the in-node path.
4894 """
4895 if step.get("type") != "parallel":
4896 return False
4897 if step.get("id") in route_plan.target_node_ids:
4898 return False
4899 if step.get("routes") or not step.get("agent"):
4900 return False
4901 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
4902 return False
4903 if StepRunner._declared_item_source(step) is None:
4904 return False
4905 if step.get("max_concurrent") is None:
4906 return False
4907 return _resolved_send_concurrency(step, build_ctx) > 1
4910def _is_send_foreach_step(
4911 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4912) -> bool:
4913 """Whether a top-level `foreach` step fans out through native ``Send``.
4915 Restricted to a singleton group's body- or agent-backed, item-sourced step
4916 with no `routes`, no `condition`/`skip_llm_if`, no step-level
4917 `output_channel`, and no `max_output`; an explicit `max_concurrent > 1` is
4918 required. The default (absent or ``1``) keeps the sequential in-node path,
4919 and the excluded edges keep their existing in-node semantics.
4920 """
4921 if step.get("type") != "foreach":
4922 return False
4923 if step.get("id") in route_plan.target_node_ids:
4924 return False
4925 if step.get("routes") or step.get("output_channel"):
4926 return False
4927 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
4928 return False
4929 if step.get("max_output") is not None:
4930 return False
4931 if StepRunner._declared_item_source(step) is None:
4932 return False
4933 if not (step.get("body") or step.get("agent")):
4934 return False
4935 if step.get("max_concurrent") is None:
4936 return False
4937 return _resolved_send_concurrency(step, build_ctx) > 1
4940class _ChannelGraphShim:
4941 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
4943 Preserves the invoke contract used by main.py and tests: a flat dict in,
4944 a flat dict out. The flat input is seeded through ``seed_committed`` and
4945 nested under ``__core__`` for the compiled graph; the committed snapshot is
4946 unwrapped back out. Streaming and state surfaces are not exposed because
4947 they would leak or accept the internal ``__core__`` shape; read-only
4948 introspection delegates to the compiled graph.
4949 """
4951 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
4952 self._compiled = compiled
4953 self._specs = specs
4955 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
4956 seeded = seed_committed(self._specs, flat_state)
4957 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
4958 return unwrap(result)
4960 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
4961 seeded = seed_committed(self._specs, flat_state)
4962 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
4963 return unwrap(result)
4965 def stream(self, *args: Any, **kwargs: Any) -> Any:
4966 self._reject_internal_shape("stream")
4968 def astream(self, *args: Any, **kwargs: Any) -> Any:
4969 self._reject_internal_shape("astream")
4971 def update_state(self, *args: Any, **kwargs: Any) -> Any:
4972 self._reject_internal_shape("update_state")
4974 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
4975 self._reject_internal_shape("aupdate_state")
4977 def get_state(self, *args: Any, **kwargs: Any) -> Any:
4978 self._reject_internal_shape("get_state")
4980 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
4981 self._reject_internal_shape("aget_state")
4983 def _reject_internal_shape(self, surface: str) -> None:
4984 raise NotImplementedError(
4985 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
4986 ".invoke()/.ainvoke() contract is exposed through the channel "
4987 "boundary shim"
4988 )
4990 def __getattr__(self, name: str) -> Any:
4991 return getattr(self._compiled, name)
4994class GraphBuilder:
4995 """
4996 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
4998 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
4999 käytetään vain {{ }} -templatejen evaluointiin.
5000 """
5002 def __init__(
5003 self,
5004 registry: ManifestRegistry,
5005 builtins: BuiltinExecutor | None = None,
5006 invoke_agent_fn: Any | None = None,
5007 run_context: dict | None = None,
5008 tracer: Any | None = None,
5009 ) -> None:
5010 self._registry = registry
5011 self._resolver = AgentResolver(registry)
5012 self._invoke_agent_fn = invoke_agent_fn
5013 self._builtins = builtins or _DEFAULT_BUILTINS
5014 if invoke_agent_fn is not None:
5015 self._builtins.invoke_agent_fn = invoke_agent_fn
5016 self._step_functions = StepFunctionExecutor(registry)
5017 self._tool_registry = build_platform_registry()
5018 self._tool_resolver = ToolResolver(self._tool_registry, self._step_functions)
5019 self._step_runner = StepRunner(
5020 self._resolver, self._builtins, invoke_agent_fn, run_context,
5021 step_functions=self._step_functions,
5022 tracer=tracer,
5023 tool_resolver=self._tool_resolver,
5024 )
5026 def build(self, pipeline: dict): # -> CompiledGraph
5027 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
5028 try:
5029 from langgraph.graph import StateGraph, END
5030 except ImportError as exc:
5031 raise ImportError(
5032 "langgraph is required to build a workflow. "
5033 "Install it with: pip install langgraph"
5034 ) from exc
5036 self._step_runner.set_template_spec(pipeline.get("spec") or {})
5038 raw_steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
5039 _validate_unique_step_ids(raw_steps)
5040 _validate_foreach_bodies(raw_steps)
5041 _validate_loop_limits(raw_steps, self._step_runner._make_ctx({}))
5042 _validate_fail_if_exhausted_steps(raw_steps)
5044 channels: list[dict] = _validated_channels(
5045 (pipeline.get("spec") or {}).get("channels")
5046 )
5047 specs = parse_channel_specs(
5048 [
5049 {
5050 "name": ch["name"],
5051 "type": ch.get("type", LAST_VALUE),
5052 "operator": ch.get("operator"),
5053 }
5054 for ch in channels
5055 ]
5056 )
5057 if not any(spec.name == "lastvalue" for spec in specs):
5058 specs = list(specs) + [ChannelSpec("lastvalue", LAST_VALUE)]
5059 channel_specs = {spec.name: spec for spec in specs}
5060 channel_names = frozenset(channel_specs)
5062 # Build-time call-step composition (ADR-0037 D4): each invoked Function
5063 # body is compiled into its own native LangGraph subgraph node, so the
5064 # runtime graph executes only native steps; a call step never survives.
5065 steps = self._compose_calls(
5066 raw_steps, channels, channel_names, channel_specs
5067 )
5068 if "lastvalue" in _collect_composed_step_ids(steps):
5069 raise RunHardFailure(
5070 "CHANNEL_BINDING_ERROR",
5071 "Pipeline step id 'lastvalue' collides with the reserved "
5072 "default channel key; rename the step",
5073 )
5074 if channels:
5075 _validate_channel_names(channels, steps)
5076 _validate_channel_inputs(channels, steps)
5077 _validate_step_channel_bindings(steps, channel_names)
5078 ordered = _topological_order(steps)
5080 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
5081 envs = (pipeline.get("spec") or {}).get("envs")
5082 if envs is not None and not isinstance(envs, dict):
5083 raise RunHardFailure(
5084 "VALIDATION_ERROR",
5085 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
5086 "spec.envs must be a map of env name → image ref (ADR-0012)",
5087 )
5089 graph = self._assemble_graph(ordered, channel_names, channel_specs)
5090 compiled = graph.compile()
5091 return _ChannelGraphShim(compiled, specs)
5093 def _compose_calls(
5094 self,
5095 steps: list[dict],
5096 channels: list[dict],
5097 channel_names: frozenset[str],
5098 channel_specs: dict[str, ChannelSpec],
5099 ) -> list[dict]:
5100 """Replace `call` steps with compiled native subgraph nodes.
5102 Each invoked Function body is assembled through ``_assemble_graph`` and
5103 compiled, then substituted as a single ``__subgraph__`` placeholder that
5104 keeps the call id and its ``depends_on``. Downstream steps addressing the
5105 call id therefore address the subgraph node unchanged.
5106 """
5108 def assemble_subgraph(sub_steps: list[dict]):
5109 return self._assemble_graph(
5110 sub_steps, channel_names, channel_specs
5111 ).compile()
5113 def assemble_while(while_step: dict, body_steps: list[dict]):
5114 return self._assemble_while_graph(
5115 while_step, body_steps, channel_names, channel_specs
5116 )
5118 def assemble_foreach(foreach_step: dict, body_steps: list[dict]):
5119 return self._assemble_foreach_graph(
5120 foreach_step, body_steps, channel_names, channel_specs
5121 )
5123 composer = _CallComposer(
5124 self._registry,
5125 assemble_subgraph,
5126 channels,
5127 channel_names,
5128 assemble_while=assemble_while,
5129 compose_whiles=True,
5130 assemble_foreach=assemble_foreach,
5131 compose_foreachs=True,
5132 build_ctx=self._step_runner._make_ctx({}),
5133 )
5134 return composer.compose(steps)
5136 def _assemble_while_graph(
5137 self,
5138 step: dict,
5139 body_steps: list[dict],
5140 channel_names: frozenset[str],
5141 channel_specs: dict[str, ChannelSpec],
5142 ) -> tuple[Any, int]:
5143 """Compile a `while` step's body into a native LangGraph cycle.
5145 The cycle is ``head -> body... -> cond`` with a conditional back-edge
5146 from ``cond`` to ``head`` (continue) or ``exit`` (termination): the
5147 entry condition/initial value live in the head, the feed + exit
5148 evaluation in the cond node, and the exit node writes the documented
5149 result. Returns the compiled cycle and the recursion bound the wrapper
5150 must apply (a cycle consumes one super-step per node per iteration).
5151 """
5152 from langgraph.graph import StateGraph, END
5154 step_id = step.get("id", "?")
5155 step_runner = self._step_runner
5156 step_runner.set_channel_routing(channel_names)
5158 state_schema = _build_state_schema()
5159 graph: StateGraph = StateGraph(state_schema)
5161 max_iter_raw = step.get("max_iterations")
5162 max_iter = (
5163 _resolve_int_strict(
5164 max_iter_raw, step_runner._make_ctx({}), "max_iterations", step_id
5165 )
5166 if max_iter_raw is not None
5167 else 2
5168 )
5169 head_id = f"__while_{step_id}__head"
5170 cond_id = f"__while_{step_id}__cond"
5171 exit_id = f"__while_{step_id}__exit"
5172 first_id = body_steps[0]["id"] if body_steps else cond_id
5174 graph.add_node(
5175 head_id,
5176 _make_while_head_node(step, step_runner, channel_names, max_iter),
5177 )
5178 for sub in body_steps:
5179 graph.add_node(
5180 sub["id"],
5181 _make_while_body_node(
5182 sub, step_runner, channel_names, channel_specs
5183 ),
5184 )
5185 graph.add_node(
5186 cond_id, _make_while_cond_node(step, step_runner, channel_names)
5187 )
5188 graph.add_node(
5189 exit_id, _make_while_exit_node(step, step_runner, channel_names)
5190 )
5192 graph.set_entry_point(head_id)
5193 head_paths = {first_id: first_id, exit_id: exit_id, END: END}
5194 graph.add_conditional_edges(
5195 head_id,
5196 _make_while_head_router(step, channel_names, first_id, exit_id, END),
5197 head_paths,
5198 )
5199 for previous, sub in zip(body_steps, body_steps[1:]):
5200 graph.add_edge(previous["id"], sub["id"])
5201 if body_steps:
5202 graph.add_edge(body_steps[-1]["id"], cond_id)
5203 graph.add_conditional_edges(
5204 cond_id,
5205 _make_while_cond_router(step, channel_names, head_id, exit_id),
5206 {head_id: head_id, exit_id: exit_id},
5207 )
5208 graph.add_edge(exit_id, END)
5210 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5211 return graph.compile(), recursion_limit
5213 def _assemble_foreach_graph(
5214 self,
5215 step: dict,
5216 body_steps: list[dict],
5217 channel_names: frozenset[str],
5218 channel_specs: dict[str, ChannelSpec],
5219 ) -> tuple[Any, int]:
5220 """Compile a sequential `foreach` step's body into a native LangGraph cycle.
5222 The cycle is ``head -> (body... | agent) -> cond`` with a conditional
5223 back-edge from ``cond`` to ``head`` (next item) or ``exit`` (exhaustion
5224 or `max_output` reached). The head resolves the item list once and
5225 injects the current item; the cond collects the item result; the exit
5226 node writes the documented aggregate. Returns the compiled cycle and
5227 the recursion bound the wrapper must apply.
5228 """
5229 from langgraph.graph import StateGraph, END
5231 step_id = step.get("id", "?")
5232 step_runner = self._step_runner
5233 step_runner.set_channel_routing(channel_names)
5235 state_schema = _build_state_schema()
5236 graph: StateGraph = StateGraph(state_schema)
5238 max_iter_raw = step.get("max_iterations")
5239 max_iter = (
5240 _resolve_int_strict(
5241 max_iter_raw, step_runner._make_ctx({}), "max_iterations", step_id
5242 )
5243 if max_iter_raw is not None
5244 else _FOREACH_DEFAULT_MAX_ITER
5245 )
5246 head_id = f"__foreach_{step_id}__head"
5247 cond_id = f"__foreach_{step_id}__cond"
5248 exit_id = f"__foreach_{step_id}__exit"
5249 agent_id = f"__foreach_{step_id}__agent"
5251 graph.add_node(
5252 head_id, _make_foreach_head_node(step, step_runner, channel_names)
5253 )
5254 if body_steps:
5255 for sub in body_steps:
5256 graph.add_node(
5257 sub["id"],
5258 _make_foreach_body_node(
5259 sub, step_runner, channel_names, channel_specs
5260 ),
5261 )
5262 first_id = body_steps[0]["id"]
5263 else:
5264 graph.add_node(
5265 agent_id,
5266 _make_foreach_agent_node(
5267 step, step_runner, channel_names, channel_specs
5268 ),
5269 )
5270 first_id = agent_id
5271 graph.add_node(
5272 cond_id, _make_foreach_cond_node(step, step_runner, channel_names)
5273 )
5274 graph.add_node(
5275 exit_id, _make_foreach_exit_node(step, step_runner, channel_names)
5276 )
5278 graph.set_entry_point(head_id)
5279 graph.add_conditional_edges(
5280 head_id,
5281 _make_foreach_head_router(step, channel_names, first_id, exit_id, END),
5282 {first_id: first_id, exit_id: exit_id, END: END},
5283 )
5284 for previous, sub in zip(body_steps, body_steps[1:]):
5285 graph.add_edge(previous["id"], sub["id"])
5286 if body_steps:
5287 graph.add_edge(body_steps[-1]["id"], cond_id)
5288 else:
5289 graph.add_edge(agent_id, cond_id)
5290 graph.add_conditional_edges(
5291 cond_id,
5292 _make_foreach_cond_router(step, channel_names, head_id, exit_id),
5293 {head_id: head_id, exit_id: exit_id},
5294 )
5295 graph.add_edge(exit_id, END)
5297 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5298 return graph.compile(), recursion_limit
5300 def _assemble_graph(
5301 self,
5302 steps: list[dict],
5303 channel_names: frozenset[str],
5304 channel_specs: dict[str, ChannelSpec],
5305 ) -> "StateGraph":
5306 from langgraph.graph import StateGraph, END
5308 # Group steps into LangGraph nodes:
5309 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
5310 nodes: list[list[dict]] = []
5311 for step in steps:
5312 dep_sig = frozenset(step.get("depends_on") or [])
5313 if nodes:
5314 last_group_sig = frozenset(
5315 dep
5316 for s in nodes[-1]
5317 for dep in (s.get("depends_on") or [])
5318 )
5319 last_types = {s.get("type") for s in nodes[-1]}
5320 if (
5321 dep_sig == last_group_sig
5322 and step.get("type") == "parallel"
5323 and last_types == {"parallel"}
5324 and not step.get("routes")
5325 and not any(s.get("routes") for s in nodes[-1])
5326 ):
5327 nodes[-1].append(step)
5328 continue
5329 nodes.append([step])
5331 state_schema = _build_state_schema()
5332 graph: StateGraph = StateGraph(state_schema)
5333 step_runner = self._step_runner
5334 step_runner.set_channel_routing(channel_names)
5336 node_ids: list[str] = []
5337 for group in nodes:
5338 if len(group) == 1:
5339 node_ids.append(group[0]["id"])
5340 else:
5341 node_ids.append("parallel_" + "_".join(s["id"] for s in group))
5343 route_plan = _build_route_plan(nodes, node_ids) if nodes else None
5345 send_parts: dict[int, _MapReduceParts] = {}
5346 if route_plan is not None:
5347 build_ctx = step_runner._make_ctx({})
5348 for index, group in enumerate(nodes):
5349 if len(group) != 1:
5350 continue
5351 send_candidate = group[0]
5352 if _is_send_parallel_step(send_candidate, route_plan, build_ctx):
5353 plan = _make_parallel_map_reduce_plan(
5354 send_candidate, step_runner
5355 )
5356 elif _is_send_foreach_step(send_candidate, route_plan, build_ctx):
5357 plan = _make_foreach_map_reduce_plan(
5358 send_candidate, step_runner, channel_names, channel_specs
5359 )
5360 else:
5361 continue
5362 send_parts[index] = _make_map_reduce_parts(
5363 send_candidate,
5364 plan,
5365 step_runner,
5366 channel_names,
5367 channel_specs,
5368 _resolved_send_concurrency(send_candidate, build_ctx),
5369 )
5371 entry_ids: list[str] = []
5372 exit_ids: list[str] = []
5373 for index, group in enumerate(nodes):
5374 parts = send_parts.get(index)
5375 if parts is None:
5376 captured_group = list(group)
5377 if len(captured_group) == 1:
5378 first = captured_group[0]
5379 if first.get("type") == _SUBGRAPH_STEP_TYPE:
5380 if first.get(_SUBGRAPH_WHILE_KEY):
5381 subgraph_node = _make_while_subgraph_node(
5382 first, step_runner, channel_names
5383 )
5384 elif first.get(_SUBGRAPH_FOREACH_KEY):
5385 subgraph_node = _make_foreach_subgraph_node(
5386 first, step_runner, channel_names
5387 )
5388 else:
5389 subgraph_node = _make_subgraph_node(
5390 first[_SUBGRAPH_GRAPH_KEY],
5391 first.get("condition"),
5392 step_runner,
5393 channel_names,
5394 first.get(_SUBGRAPH_STEP_IDS_KEY),
5395 )
5396 graph.add_node(node_ids[index], subgraph_node)
5397 entry_ids.append(node_ids[index])
5398 exit_ids.append(node_ids[index])
5399 continue
5400 fn = _make_step_node(
5401 first, step_runner, channel_names, channel_specs,
5402 )
5403 else:
5404 fn = _make_parallel_node(
5405 captured_group,
5406 step_runner,
5407 channel_names,
5408 channel_specs,
5409 )
5410 graph.add_node(node_ids[index], fn)
5411 entry_ids.append(node_ids[index])
5412 exit_ids.append(node_ids[index])
5413 continue
5414 fallback_step = group[0]
5415 graph.add_node(parts.dispatch_id, parts.dispatch_fn)
5416 graph.add_node(parts.worker_id, parts.worker_fn)
5417 graph.add_node(
5418 parts.fallback_id,
5419 _make_step_node(
5420 fallback_step, step_runner, channel_names, channel_specs
5421 ),
5422 )
5423 graph.add_node(parts.join_id, parts.join_fn)
5424 graph.add_conditional_edges(
5425 parts.dispatch_id,
5426 parts.route_fn,
5427 [parts.worker_id, parts.fallback_id],
5428 )
5429 graph.add_edge(parts.worker_id, parts.join_id)
5430 graph.add_edge(parts.fallback_id, parts.join_id)
5431 entry_ids.append(parts.dispatch_id)
5432 exit_ids.append(parts.join_id)
5434 if nodes:
5435 _wire_dependency_edges(
5436 graph, nodes, node_ids, END, route_plan, entry_ids, exit_ids
5437 )
5438 _wire_route_edges(graph, route_plan, END, step_runner)
5439 else:
5440 graph.add_node("noop", lambda s: s)
5441 graph.set_entry_point("noop")
5442 graph.add_edge("noop", END)
5444 return graph
5447# ---------------------------------------------------------------------------
5448# Julkinen API
5449# ---------------------------------------------------------------------------
5451_registry_cache: ManifestRegistry | None = None
5454def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
5455 global _registry_cache
5456 if _registry_cache is None or dirs is not None:
5457 loader = ManifestLoader(dirs)
5458 manifests = loader.load_all()
5459 _registry_cache = ManifestRegistry(manifests)
5460 return _registry_cache
5463def build_workflow(
5464 pipeline_name: str,
5465 *,
5466 dirs: list[Path] | None = None,
5467 builtins: BuiltinExecutor | None = None,
5468 invoke_agent_fn: Any | None = None,
5469 run_context: dict | None = None,
5470 tracer: Any | None = None,
5471):
5472 """
5473 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
5474 LangGraph-workflown.
5476 Args:
5477 pipeline_name: Pipeline metadata.name -arvo.
5478 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
5479 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
5480 invoke_agent_fn: fn(resolved_spec, step, state) → state.
5481 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
5483 Returns:
5484 Compilattu LangGraph CompiledGraph.
5485 """
5486 registry = _get_registry(dirs)
5487 pipeline = registry.get("Pipeline", pipeline_name)
5488 if pipeline is None:
5489 available = [
5490 (m.get("metadata") or {}).get("name")
5491 for m in registry.all_of_kind("Pipeline")
5492 ]
5493 raise RunHardFailure(
5494 "VALIDATION_ERROR",
5495 f"Pipeline '{pipeline_name}' not found. Available: {available}",
5496 )
5497 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
5498 return builder.build(pipeline)