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