Coverage for core/langgraph/runner.py: 94%
2539 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 05:49 +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 operator
21import os
22import re
23import threading
24from dataclasses import dataclass
25from concurrent.futures import ThreadPoolExecutor, as_completed
26from pathlib import Path
27from typing import Annotated, Any
29import yaml
31from langgraph.types import Send
33# SkillLoadError-import (ei sirkulaari-riskiä — _llm importtaa vain stdlibin):
34# testit ja discover-wrapperit lataavat _llm:n tasaisena moduulina (sys.path
35# insert core/langgraph/), main.py qualified (langgraph._llm, PYTHONPATH=/app).
36# Kummatkin pinnat käännetään samaan _llm-tiedostoon; try/except varmistaa,
37# että luokka on sama objekti kuin nostavalla puolella molemmissa konteksteissa.
38try:
39 from _llm import DEFAULT_MAX_PARALLEL, SkillLoadError, _PARENT_STACK # noqa: F401
40except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
41 from langgraph._llm import ( # noqa: F401
42 DEFAULT_MAX_PARALLEL,
43 SkillLoadError,
44 _PARENT_STACK,
45 )
47# Same dual-import pattern as _llm (flat sys.path and qualified PYTHONPATH=/app)
48try:
49 from _step_functions import StepFunctionExecutor # noqa: F401
50except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
51 from langgraph._step_functions import StepFunctionExecutor # noqa: F401
53# Agent-instruction error code emitted by the engine error set — same
54# dual-import pattern as _llm.
55try:
56 from _instructions import AGENT_INSTRUCTION_MISSING
57except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
58 from langgraph._instructions import AGENT_INSTRUCTION_MISSING
60# Structural tool resolution layer — same dual-import pattern as _llm.
61try:
62 from tools import (
63 RESOLVED_TOOLS_KEY,
64 ToolCollisionError,
65 ToolRegistryError,
66 ToolResolver,
67 build_platform_registry,
68 )
69except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
70 from langgraph.tools import (
71 RESOLVED_TOOLS_KEY,
72 ToolCollisionError,
73 ToolRegistryError,
74 ToolResolver,
75 build_platform_registry,
76 )
78# COMMUNICATION-context projection (declared input.channels reads) — same
79# dual-import pattern as _llm.
80try:
81 from _data_structure import (
82 COMMUNICATION_CONTEXT_KEY,
83 resolve_communication_context,
84 )
85except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
86 from langgraph._data_structure import (
87 COMMUNICATION_CONTEXT_KEY,
88 resolve_communication_context,
89 )
91# Core-owned channel-context storage/commit/seeding primitives — same dual-import
92# pattern. The channel path commits through these; it no longer materializes
93# native LangGraph channels.
94try:
95 from _channels import (
96 CORE_KEY,
97 LAST_VALUE,
98 ChannelSpec,
99 ChannelWorkingSet,
100 parse_channel_specs,
101 seed_committed,
102 unwrap,
103 wrap,
104 )
105except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
106 from langgraph._channels import (
107 CORE_KEY,
108 LAST_VALUE,
109 ChannelSpec,
110 ChannelWorkingSet,
111 parse_channel_specs,
112 seed_committed,
113 unwrap,
114 wrap,
115 )
117# Core-owned reducer layer — same dual-import pattern. The map-reduce
118# accumulator reducer folds concurrent ``Send`` worker writes per step id.
119try:
120 from _reducers import accumulate_map_reduce
121except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
122 from langgraph._reducers import accumulate_map_reduce
124# Agent output envelope error code — same dual-import pattern.
125try:
126 from _envelope import ENVELOPE_ERROR_CODE, ENVELOPE_STATUSES, is_error_envelope
127except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
128 from langgraph._envelope import ( # noqa: F401
129 ENVELOPE_ERROR_CODE,
130 ENVELOPE_STATUSES,
131 is_error_envelope,
132 )
134log = logging.getLogger(__name__)
137class RunHardFailure(Exception):
138 """Run-tason kova virhe: ajo pysähtyy heti (phase failed, exit 1).
140 Nostetaan kun step-tulos kantaa kovan virheen: TRUNCATION_ERROR-koodi
141 (kuratoidun kontekstin katkaisu), EXECUTION_ERROR-koodi (builtinien/stepien
142 deterministinen tekninen vika), TOOL_FAILURE-koodi (agentin työkalupinnan
143 tekninen vika), AGENT_ERROR-koodi (status:error-envelope,
144 ENVELOPE_ERROR_CODE), CALL_NOT_ALLOWED/CALL_INVALID_PARAMS/CALL_FAILED
145 (natiivi deterministic-tool-kutsu), BUILTIN_NAME_COLLISION (nimi sekä
146 builtin- että StepFunction-rekisterissä) tai kontekstiylityksen
147 signatuurin kantava VALIDATION_ERROR (assembled_tokens + context_budget,
148 _llm._context_overflow_result).
149 Kaikki muut virheluokat (budget:*, parse_error, timeout, LLM-virheet)
150 säilyvät nykyisessä partial-semantiikassa. Stepin details kulkeutuu
151 attribuuttina loppu-PATCHin error.details-kenttään.
152 """
154 ENGINE_ERROR_CODES: frozenset[str] = frozenset(
155 {
156 "TIMEOUT",
157 "VALIDATION_ERROR",
158 "CHANNEL_BINDING_ERROR",
159 "INVALID_CHILD_PLAN",
160 AGENT_INSTRUCTION_MISSING,
161 "UNRESOLVED_OUTPUT",
162 "BUILTIN_NAME_COLLISION",
163 "EXECUTION_ERROR",
164 ENVELOPE_ERROR_CODE,
165 "TOOL_FAILURE",
166 "CALL_NOT_ALLOWED",
167 "CALL_INVALID_PARAMS",
168 "CALL_FAILED",
169 "RESOURCE_ERROR",
170 "SPAWN_BUDGET_EXHAUSTED",
171 "EXHAUSTED",
172 "SPAWN_OVER_FLIGHT",
173 "TRUNCATION_ERROR",
174 # BLOCKED is API-owned, not engine-emitted; it stays in the set so
175 # the guard never rejects a contract-valid value.
176 "BLOCKED",
177 }
178 )
180 HARD_ERROR_CODES: frozenset[str] = frozenset(
181 {
182 "TRUNCATION_ERROR",
183 "TOOL_FAILURE",
184 ENVELOPE_ERROR_CODE,
185 "CALL_NOT_ALLOWED",
186 "CALL_INVALID_PARAMS",
187 "CALL_FAILED",
188 "BUILTIN_NAME_COLLISION",
189 }
190 )
192 def __init__(self, code: str, message: str, details: dict | None = None) -> None:
193 if code not in self.ENGINE_ERROR_CODES:
194 raise ValueError(
195 f"RunHardFailure code {code!r} is not a member of the documented "
196 f"engine error-code set {sorted(self.ENGINE_ERROR_CODES)}"
197 )
198 if not isinstance(message, str) or not message:
199 raise ValueError("RunHardFailure message must be a non-empty str")
200 super().__init__(message)
201 self.code = code
202 self.message = message
203 self.details = details
206ENGINE_ERROR_CODES = RunHardFailure.ENGINE_ERROR_CODES
209class SpawnBudget:
210 """Per-tree agent-call budget shared across all spawn nodes.
212 The whole spawn tree draws from a single counter: every agent invocation
213 (the director included) consumes one unit, so the total number of
214 agent invocations the tree may make is bounded by `max_agent_calls`. This is a
215 per-agent-run budget, NOT an LLM-roundtrip budget: a single agent run may
216 make several LLM calls (e.g. an agent-side tool loop over its declared
217 `tools`, resolved into tool entries); those LLM roundtrips are bounded by
218 the run's `max_llm_calls`, not by
219 `max_agent_calls`. There is no engine-level agent retry loop. Thread-safe so
220 a concurrent caller cannot over-commit the shared counter.
221 """
223 def __init__(self, max_agent_calls: int | None) -> None:
224 self._remaining: int | None = (
225 None if max_agent_calls is None else max(0, int(max_agent_calls))
226 )
227 self._lock = threading.Lock()
229 def consume(self) -> bool:
230 """Reserve one agent call. Returns False when the budget is exhausted."""
231 with self._lock:
232 if self._remaining is None:
233 return True
234 if self._remaining <= 0:
235 return False
236 self._remaining -= 1
237 return True
240_SPAWN_SOFT_FAIL_KEYS: frozenset[str] = frozenset({"error", "parse_error", "escalate"})
242_SPAWN_DEFAULT_MAX_AGENT_CALLS: int = 64
243_SPAWN_DEFAULT_MAX_DEPTH: int = 1
244_SPAWN_DEFAULT_MAX_SPAWN_PER_NODE: int = 16
246# The spawn tree's superstep count is bounded by the agent-call budget, not by
247# LangGraph's default recursion cap; lift it so a deep-but-budgeted tree runs.
248_SPAWN_RECURSION_LIMIT: int = 100_000
250# ---------------------------------------------------------------------------
251# Polut
252# ---------------------------------------------------------------------------
254PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", "/project"))
255MANIFEST_DIRS = [
256 PROJECT_ROOT / ".agent-platform" / "workflows",
257 PROJECT_ROOT / ".agent-platform" / "agents",
258]
259SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "manifest.schema.json"
262def resolve_manifest_dirs(clone_root: Path) -> list[Path]:
263 """Palauttaa olemassa olevat manifestihakemistot kloonatusta manifest-reposta.
265 Skannaa ainoastaan `.agent-platform/workflows` ja `.agent-platform/agents` —
266 ei koskaan koko kloonia. [] jos kumpaakaan ei ole (kutsuja → fail-fast).
267 """
268 return [
269 d for d in (
270 clone_root / ".agent-platform" / "workflows",
271 clone_root / ".agent-platform" / "agents",
272 )
273 if d.is_dir()
274 ]
276# ---------------------------------------------------------------------------
277# Template-resoluutio
278# ---------------------------------------------------------------------------
280_TMPL_RE = re.compile(r'\{\{\s*(.*?)\s*\}\}')
282# Sentinel for an unresolvable dot-path segment (distinct from a resolved None).
283_UNRESOLVED = object()
286def _split_key_indices(part: str) -> tuple[str, list[int]]:
287 """Split a dot-path segment into a base key and a list of indices.
289 ``'a[0][1]'`` → ``('a', [0, 1])``; ``'a'`` → ``('a', [])``; ``'[0]'`` →
290 ``('', [0])``. Only non-negative indices are supported (list access).
291 """
292 m = re.fullmatch(r'([^\[]*)((\[\d+\])*)', part)
293 base = m.group(1)
294 indices = [int(x) for x in re.findall(r'\d+', m.group(2))]
295 return base, indices
298def _apply_indices(obj: Any, indices: list[int]) -> Any:
299 """Index a (possibly chained) list/tuple; ``_UNRESOLVED`` on out-of-range."""
300 for idx in indices:
301 if not isinstance(obj, (list, tuple)) or not (0 <= idx < len(obj)):
302 return _UNRESOLVED
303 obj = obj[idx]
304 return obj
307def _dot_get_segment(obj: Any, part: str) -> Any:
308 """Resolve one dot-path segment against a dict or an indexed sequence.
310 A segment may be a bare key (``'a'``), a keyed list-index (``'a[0]'``,
311 ``'a[0][1]'``), or a bare index into a sequence. Returns the resolved value
312 or the ``_UNRESOLVED`` sentinel when the segment cannot be resolved (missing
313 key, non-dict/non-sequence mid-path, or out-of-range index).
314 """
315 base, indices = _split_key_indices(part)
316 if isinstance(obj, dict):
317 if base:
318 if base not in obj:
319 return _UNRESOLVED
320 obj = obj[base]
321 return _apply_indices(obj, indices)
322 if isinstance(obj, (list, tuple)):
323 if indices:
324 return _apply_indices(obj, indices)
325 if base and base.isdigit():
326 return _apply_indices(obj, [int(base)])
327 return _UNRESOLVED
328 return _UNRESOLVED
331def _dot_get(obj: Any, path: str) -> Any:
332 """Navigate nested dicts/lists via dot-path: 'a.b.c' → obj['a']['b']['c'].
334 Supports list-index segments (``results[1].field``) and dot-into-aggregate
335 traversal (missing key / non-dict-non-list mid-path / out-of-range index
336 resolve to None — lenient semantics for non-output refs).
337 """
338 for part in path.split('.'):
339 obj = _dot_get_segment(obj, part)
340 if obj is _UNRESOLVED:
341 return None
342 return obj
345class _ExpressionError(ValueError):
346 """A malformed template/condition expression at the evaluation boundary.
348 Raised by the expression tokenizer, parser, and evaluator so
349 ``_resolve_template`` relabels only design-level expression failures as
350 ``VALIDATION_ERROR``. Subclasses ``ValueError`` so callers that already
351 treat expression failures as ``ValueError`` keep working.
352 """
355class _UnresolvedReference(_ExpressionError):
356 """A strict dot-path read hit an absent key or a non-mapping segment.
358 Subclasses ``ValueError`` (via ``_ExpressionError``) so existing lenient
359 callers that treat any strict read failure as "unresolved"
360 (``__call_emit`` optional outputs, ``_builtins._resolve_dot_path``) keep
361 working while the template/condition boundary can still tell an unresolved
362 reference apart from a parse error.
363 """
366def _dot_get_strict(obj: Any, path: str) -> Any:
367 """Dot-path navigation that raises on an unresolved reference.
369 A reference is unresolved when a key on the path is absent or a segment is
370 not a mapping — used by condition evaluation so an undefined or skipped
371 step's field is never silently evaluated to falsy (ADR-0012 spirit). Also
372 used by ``__call_emit`` to distinguish a resolved output source from an
373 unresolved one (required output → hard error, ADR-0012).
374 """
375 for part in path.split('.'):
376 if obj is None:
377 raise _UnresolvedReference(
378 f"Unresolved reference '{path}': '{part}' is not a mapping"
379 )
380 obj = _dot_get_segment(obj, part)
381 if obj is _UNRESOLVED:
382 raise _UnresolvedReference(
383 f"Unresolved reference '{path}': '{part}' is not a mapping"
384 )
385 return obj
388def materialize_pipeline_output(pipeline_spec: dict, final_state: dict) -> Any:
389 """Resolve a Pipeline's declared run-end output.
391 Reads ``spec.output.source`` (a dot-path) out of ``final_state`` via
392 ``_dot_get`` and returns the surfaced value. Returns ``None`` when no
393 ``spec.output`` is declared, when it has no ``source``, or when the source
394 is unresolved — this is a run-end surfacing (distinct from the emit-time
395 hard-error used for function-interface outputs): an optional run output
396 that cannot be resolved surfaces as None, never a run-end failure.
397 """
398 out_spec = (pipeline_spec or {}).get("output")
399 if not out_spec:
400 return None
401 source = out_spec.get("source")
402 if not source:
403 return None
404 return _dot_get(final_state, source)
407def _merge_side_channel(base: dict | None, additions: dict) -> dict:
408 """Yhdistä rinnakkaisen side-channel-envelopen (step_context_metrics /
409 step_tool_meta) kerrokset: myöhempi kirjoitus voittaa (sama semantiikka kuin
410 state-merge {**state, ...})."""
411 merged = dict(base or {})
412 merged.update(additions)
413 return merged
416def _assemble_spawn_outcome(records: dict) -> tuple[dict, dict, dict]:
417 """Assemble the spawn result envelope and side channels from leaf records.
419 ``records`` maps each node's plan-order path key to its record. Children
420 keys are rebuilt in plan order (paths sort before their descendants), and
421 the director's own envelope is the tree root at ``()``. Side channels fold
422 in plan order (director first, then children).
423 """
424 ordered = sorted(records)
425 envelopes: dict = {}
426 for path in reversed(ordered):
427 record = records[path]
428 children: dict = {}
429 failing: list = []
430 for candidate in ordered:
431 child = records[candidate]
432 if child.get("parent_path") != path:
433 continue
434 child_envelope = envelopes[candidate]
435 children[child["child_id"]] = child_envelope
436 if not child_envelope["passed"]:
437 failing.append(child["child_id"])
438 envelopes[path] = {
439 "result": record["result"],
440 "children": children,
441 "passed": record["self_passed"] and not failing,
442 "failing": failing,
443 "exhausted": record["exhausted"],
444 }
445 metrics: dict = {}
446 tool_meta: dict = {}
447 for path in ordered:
448 metrics = _merge_side_channel(metrics, records[path].get("metrics") or {})
449 tool_meta = _merge_side_channel(
450 tool_meta, records[path].get("tool_meta") or {}
451 )
452 return envelopes[()], metrics, tool_meta
455_MULTI_OPS = ('<=', '>=', '==', '!=')
456_SINGLE_OPS = '+-<>'
457_WORD_BREAK = set("()[],+<=>!'-")
458_CMP_OPS = frozenset(('==', '!=', '<=', '>=', '<', '>'))
461def _tokenize_expr(expr: str) -> list[tuple[str, Any]]:
462 """Split an expression into (kind, value) tokens.
464 Kinds: 'op' (+ - <= >= == != < >), 'punct' ( ) [ ] ,, 'num', 'str',
465 'name'. A 'name' is a dot-path reference or a bare word (e.g. an unquoted
466 comparison literal, true/false/None, and/or/not).
467 """
468 tokens: list[tuple[str, Any]] = []
469 i, n = 0, len(expr)
470 while i < n:
471 ch = expr[i]
472 if ch.isspace():
473 i += 1
474 continue
475 if ch in '()[],':
476 tokens.append(('punct', ch))
477 i += 1
478 continue
479 if ch in '+-<=>!':
480 two = expr[i:i + 2]
481 if two in _MULTI_OPS:
482 tokens.append(('op', two))
483 i += 2
484 continue
485 if ch in _SINGLE_OPS:
486 tokens.append(('op', ch))
487 i += 1
488 continue
489 raise _ExpressionError(f"Invalid operator in expression: {two!r}")
490 if ch in ('\'', '"'):
491 j = i + 1
492 buf: list[str] = []
493 while j < n and expr[j] != ch:
494 buf.append(expr[j])
495 j += 1
496 if j >= n:
497 raise _ExpressionError(f"Unterminated string literal in expression: {expr!r}")
498 tokens.append(('str', ''.join(buf)))
499 i = j + 1
500 continue
501 if ch.isdigit() or (ch == '.' and i + 1 < n and expr[i + 1].isdigit()):
502 j = i
503 if ch == '.':
504 j += 1
505 while j < n and expr[j].isdigit():
506 j += 1
507 tokens.append(('num', float(expr[i:j])))
508 else:
509 while j < n and expr[j].isdigit():
510 j += 1
511 if j < n and expr[j] == '.' and j + 1 < n and expr[j + 1].isdigit():
512 j += 1
513 while j < n and expr[j].isdigit():
514 j += 1
515 tokens.append(('num', float(expr[i:j])))
516 else:
517 tokens.append(('num', int(expr[i:j])))
518 i = j
519 continue
520 j = i
521 while j < n and not expr[j].isspace() and expr[j] not in _WORD_BREAK:
522 j += 1
523 if j == i:
524 raise _ExpressionError(f"Unexpected character in expression: {expr[i:]!r}")
525 tokens.append(('name', expr[i:j]))
526 i = j
527 return tokens
530# AST node shapes:
531# ('lit', value) literal (str/num/bool/None)
532# ('ref', name) dot-path reference
533# ('list', [node, ...]) [a, b, c] literal
534# ('neg', node) unary minus
535# ('not', node) not X
536# ('cmp', op, left, right) comparison
537# ('binop', 'and'|'or'|'+', l, r) binary operator
540class _ExprParser:
541 """Recursive-descent parser for template/condition expressions.
543 Precedence (loosest first): or < and < not < comparison < '+' < unary.
544 Grammar:
545 expr := or_expr
546 or_expr := and_expr ('or' and_expr)*
547 and_expr := not_expr ('and' not_expr)*
548 not_expr := 'not' not_expr | cmp_expr
549 cmp_expr := add_expr (cmp_op add_expr)?
550 add_expr := unary_expr ('+' unary_expr)*
551 unary_expr:= '-' unary_expr | primary
552 primary := num | str | true/false/None | list | '(' expr ')' | name
553 `and`/`or` parse into a full AST but EVALUATE with left-to-right
554 short-circuit, so the right operand is only resolved if needed and an
555 unresolved read there never raises.
556 """
558 def __init__(self, tokens: list[tuple[str, Any]]) -> None:
559 self._tokens = tokens
560 self._i = 0
562 def parse(self) -> tuple:
563 node = self._or_expr()
564 if self._i != len(self._tokens):
565 tok = self._tokens[self._i]
566 raise _ExpressionError(f"Unexpected token in expression: {tok[1]!r}")
567 return node
569 def _peek(self) -> tuple[str, Any] | None:
570 return self._tokens[self._i] if self._i < len(self._tokens) else None
572 def _next(self) -> tuple[str, Any]:
573 tok = self._tokens[self._i]
574 self._i += 1
575 return tok
577 def _peek_name(self, kw: str) -> bool:
578 tok = self._peek()
579 return tok is not None and tok[0] == 'name' and tok[1] == kw
581 def _peek_op(self, op: str) -> bool:
582 tok = self._peek()
583 return tok is not None and tok[0] == 'op' and tok[1] == op
585 def _expect_punct(self, value: str) -> None:
586 tok = self._next()
587 if tok != ('punct', value):
588 raise _ExpressionError(f"Expected '{value}' in expression, got {tok[1]!r}")
590 def _or_expr(self) -> tuple:
591 node = self._and_expr()
592 while self._peek_name('or'):
593 self._next()
594 right = self._and_expr()
595 node = ('binop', 'or', node, right)
596 return node
598 def _and_expr(self) -> tuple:
599 node = self._not_expr()
600 while self._peek_name('and'):
601 self._next()
602 right = self._not_expr()
603 node = ('binop', 'and', node, right)
604 return node
606 def _not_expr(self) -> tuple:
607 if self._peek_name('not'):
608 self._next()
609 return ('not', self._not_expr())
610 return self._cmp_expr()
612 def _cmp_expr(self) -> tuple:
613 node = self._add_expr()
614 tok = self._peek()
615 if tok is not None and tok[0] == 'op' and tok[1] in _CMP_OPS:
616 self._next()
617 right = self._add_expr()
618 node = ('cmp', tok[1], node, right)
619 return node
621 def _add_expr(self) -> tuple:
622 node = self._unary_expr()
623 while self._peek_op('+'):
624 self._next()
625 right = self._unary_expr()
626 node = ('binop', '+', node, right)
627 return node
629 def _unary_expr(self) -> tuple:
630 if self._peek_op('-'):
631 self._next()
632 node = self._unary_expr()
633 if node[0] == 'lit' and isinstance(node[1], (int, float)) and not isinstance(node[1], bool):
634 return ('lit', -node[1])
635 return ('neg', node)
636 return self._primary()
638 def _primary(self) -> tuple:
639 tok = self._peek()
640 if tok is None:
641 raise _ExpressionError("Unexpected end of expression")
642 kind, value = tok
643 if kind == 'num':
644 self._next()
645 return ('lit', value)
646 if kind == 'str':
647 self._next()
648 return ('lit', value)
649 if kind == 'punct':
650 if value == '(':
651 self._next()
652 node = self._or_expr()
653 self._expect_punct(')')
654 return node
655 if value == '[':
656 self._next()
657 items: list[tuple] = []
658 if self._peek() != ('punct', ']'):
659 while True:
660 items.append(self._or_expr())
661 if self._peek() == ('punct', ','):
662 self._next()
663 continue
664 break
665 self._expect_punct(']')
666 return ('list', items)
667 raise _ExpressionError(f"Unexpected punctuation in expression: {value!r}")
668 if kind == 'name':
669 self._next()
670 lowered = value.lower()
671 if lowered == 'true':
672 return ('lit', True)
673 if lowered == 'false':
674 return ('lit', False)
675 if lowered in ('none', 'null'):
676 return ('lit', None)
677 if value in ('and', 'or', 'not'):
678 raise _ExpressionError(f"Unexpected keyword in expression: {value!r}")
679 return ('ref', value)
680 raise _ExpressionError(f"Unexpected token in expression: {value!r}")
683def _is_number(value: Any) -> bool:
684 return isinstance(value, (int, float)) and not isinstance(value, bool)
687def _add_values(left: Any, right: Any) -> Any:
688 """Binary `+`. Lists concatenate, numbers add, strings concatenate."""
689 if isinstance(left, list) and isinstance(right, list):
690 return left + right
691 if _is_number(left) and _is_number(right):
692 return left + right
693 if isinstance(left, str) and isinstance(right, str):
694 return left + right
695 raise _ExpressionError(
696 f"Unsupported operands for '+': {type(left).__name__} + {type(right).__name__}"
697 )
700def _eval_comparison_rhs(node: tuple, ctx: dict, strict: bool) -> Any:
701 """Comparison right-hand side. A bare non-dotted reference compares as a
702 string literal (preserving `profile == PR`); everything else evaluates
703 normally (numbers, quoted strings, dotted refs, lists, grouped exprs)."""
704 if node[0] == 'ref' and '.' not in node[1]:
705 return node[1]
706 return _eval_ast(node, ctx, strict)
709def _eval_cmp(op: str, left_node: tuple, right_node: tuple, ctx: dict, strict: bool) -> bool:
710 left = _eval_ast(left_node, ctx, strict)
711 right = _eval_comparison_rhs(right_node, ctx, strict)
712 try:
713 if op == '==':
714 return left == right
715 if op == '!=':
716 return left != right
717 if op == '<=':
718 return left <= right
719 if op == '>=':
720 return left >= right
721 if op == '<':
722 return left < right
723 if op == '>':
724 return left > right
725 except TypeError:
726 return False
727 raise _ExpressionError(f"Unknown comparison operator: {op!r}")
730def _eval_ast(node: tuple, ctx: dict, strict: bool) -> Any:
731 """Evaluate a parsed AST. `strict=True` raises on an unresolved reference
732 read (condition/exit_condition); `and`/`or` short-circuit left-to-right."""
733 kind = node[0]
734 if kind == 'lit':
735 return node[1]
736 if kind == 'ref':
737 return _dot_get_strict(ctx, node[1]) if strict else _dot_get(ctx, node[1])
738 if kind == 'list':
739 return [_eval_ast(item, ctx, strict) for item in node[1]]
740 if kind == 'neg':
741 operand = _eval_ast(node[1], ctx, strict)
742 if not _is_number(operand):
743 raise _ExpressionError(
744 f"Unsupported operand for unary '-': {type(operand).__name__}"
745 )
746 return -operand
747 if kind == 'not':
748 return not bool(_eval_ast(node[1], ctx, strict))
749 if kind == 'cmp':
750 return _eval_cmp(node[1], node[2], node[3], ctx, strict)
751 if kind == 'binop':
752 op, left_node, right_node = node[1], node[2], node[3]
753 if op == 'and':
754 if not bool(_eval_ast(left_node, ctx, strict)):
755 return False
756 return bool(_eval_ast(right_node, ctx, strict))
757 if op == 'or':
758 if bool(_eval_ast(left_node, ctx, strict)):
759 return True
760 return bool(_eval_ast(right_node, ctx, strict))
761 if op == '+':
762 left = _eval_ast(left_node, ctx, strict)
763 right = _eval_ast(right_node, ctx, strict)
764 return _add_values(left, right)
765 raise _ExpressionError(f"Unknown binary operator: {op!r}")
766 raise _ExpressionError(f"Unknown expression node: {kind!r}")
769def _eval_expr(expr: str, ctx: dict, strict: bool = False) -> Any:
770 """Evaluate a template expression against context dict.
772 `strict` is used by condition evaluation: dot-path access to an unresolved
773 reference raises (ValueError) instead of silently resolving to falsy, and
774 and/or short-circuit left-to-right so a value is never read when the left
775 side already determines the result.
776 """
777 expr = expr.strip()
778 tokens = _tokenize_expr(expr)
779 if not tokens:
780 return _dot_get_strict(ctx, expr) if strict else _dot_get(ctx, expr)
781 return _eval_ast(_ExprParser(tokens).parse(), ctx, strict)
784def _resolve_template(value: Any, ctx: dict, strict: bool = False) -> Any:
785 """Recursively resolve {{ expr }} templates in strings / dicts / lists.
787 The template-evaluation boundary: a design-level failure (an unresolved
788 strict reference, or a malformed expression) surfaces as a typed
789 ``RunHardFailure`` here, never as a raw parser ``ValueError`` that a
790 concurrent worker would record as a code-less per-item error.
791 """
792 try:
793 return _resolve_template_inner(value, ctx, strict)
794 except _UnresolvedReference as exc:
795 raise RunHardFailure(
796 "UNRESOLVED_OUTPUT",
797 str(exc),
798 ) from exc
799 except _ExpressionError as exc:
800 raise RunHardFailure(
801 "VALIDATION_ERROR",
802 f"Invalid template/condition expression: {exc}",
803 ) from exc
806def _resolve_template_inner(value: Any, ctx: dict, strict: bool = False) -> Any:
807 if isinstance(value, str):
808 stripped = value.strip()
809 # Entire string is one template → replace with actual type (not string)
810 m = _TMPL_RE.fullmatch(stripped)
811 if m:
812 return _eval_expr(m.group(1), ctx, strict)
813 # Inline substitution → always returns string
814 def _replace(m: re.Match) -> str:
815 result = _eval_expr(m.group(1), ctx, strict)
816 return '' if result is None else str(result)
817 return _TMPL_RE.sub(_replace, value)
818 if isinstance(value, dict):
819 return {k: _resolve_template_inner(v, ctx, strict) for k, v in value.items()}
820 if isinstance(value, list):
821 return [_resolve_template_inner(v, ctx, strict) for v in value]
822 return value
825def _resolve_input(input_spec: Any, state: dict, ctx: dict) -> Any:
826 """
827 Resolve step input spec to a concrete value:
828 "{{ item }}" → template
829 "[host-ai-context-map-path]" → literal placeholder (verbatim pass-through)
830 "gather.doc_files" → dot-path into state
831 {"k": "a.b"} → dict with each value resolved
832 None → None
833 """
834 if input_spec is None:
835 return None
836 if isinstance(input_spec, str):
837 if '{{' in input_spec:
838 return _resolve_template(input_spec, ctx)
839 stripped = input_spec.lstrip()
840 # A bracket-prefixed literal (e.g. a host-ai-context-map-path
841 # placeholder) is NOT a dot-path reference into state — hand it to the
842 # builtin verbatim so it resolves the placeholder itself.
843 if stripped.startswith('['):
844 return input_spec
845 # Dot-path into state (no braces)
846 return _dot_get(state, stripped)
847 if isinstance(input_spec, dict):
848 return {k: _resolve_input(v, state, ctx) for k, v in input_spec.items()}
849 if isinstance(input_spec, list):
850 return [_resolve_input(v, state, ctx) for v in input_spec]
851 return input_spec
854def _eval_condition(cond: Any, ctx: dict) -> bool:
855 """Evaluate a step condition. None → True (always run).
857 Strict (short-circuit + hard error on unresolved reads) per R1: an
858 undefined or skipped-step field read without short-circuit protection
859 raises, never silently evaluates to falsy.
860 """
861 if cond is None:
862 return True
863 resolved = _resolve_template(cond, ctx, strict=True)
864 if isinstance(resolved, bool):
865 return resolved
866 if isinstance(resolved, str):
867 return resolved.lower() not in ('false', '0', '', 'none', 'null')
868 return bool(resolved)
871def _resolve_int(value: Any, ctx: dict, default: int) -> int:
872 v = _resolve_template(value, ctx) if isinstance(value, str) else value
873 try:
874 return int(v)
875 except (TypeError, ValueError):
876 return default
879# ---------------------------------------------------------------------------
880# ManifestLoader
881# ---------------------------------------------------------------------------
884def _parse_manifest_file(path: Path) -> list[dict]:
885 """Palauttaa kaikki apiVersion: agent-platform/v1 -dokumentit tiedostosta."""
886 try:
887 content = path.read_text(encoding="utf-8")
888 except OSError as exc:
889 log.warning("Cannot read %s: %s", path, exc)
890 return []
892 results: list[dict] = []
893 for raw in content.split("\n---"):
894 raw_clean = "\n".join(
895 line for line in raw.splitlines()
896 if not line.strip().startswith("#")
897 )
898 try:
899 doc = yaml.safe_load(raw_clean)
900 except yaml.YAMLError as exc:
901 log.warning("YAML parse error in %s: %s", path, exc)
902 continue
903 if isinstance(doc, dict) and doc.get("apiVersion") == "agent-platform/v1":
904 results.append(doc)
905 return results
908class ManifestLoader:
909 """Skannaa hakemistot ja lataa kaikki YAML-manifestit."""
911 def __init__(self, dirs: list[Path] | None = None) -> None:
912 self._dirs = dirs or MANIFEST_DIRS
914 def load_all(self) -> list[dict]:
915 manifests: list[dict] = []
916 for d in self._dirs:
917 if not d.exists():
918 log.debug("Manifest dir not found: %s", d)
919 continue
920 for path in sorted(d.glob("**/*.yaml")):
921 docs = _parse_manifest_file(path)
922 manifests.extend(docs)
923 if docs:
924 log.debug("Loaded %d manifest(s) from %s", len(docs), path)
925 return manifests
928# ---------------------------------------------------------------------------
929# ManifestRegistry
930# ---------------------------------------------------------------------------
932ResourceKey = tuple[str, str] # (kind, name)
935class ManifestRegistry:
936 """Indeksoi ladatut manifestit kind+name avaimella."""
938 def __init__(self, manifests: list[dict]) -> None:
939 self._store: dict[ResourceKey, dict] = {}
940 for doc in manifests:
941 kind = doc.get("kind", "")
942 name = (doc.get("metadata") or {}).get("name", "")
943 if kind and name:
944 key: ResourceKey = (kind, name)
945 if key in self._store:
946 log.warning("Duplicate manifest %s/%s — last one wins", kind, name)
947 self._store[key] = doc
949 def get(self, kind: str, name: str) -> dict | None:
950 return self._store.get((kind, name))
952 def all_of_kind(self, kind: str) -> list[dict]:
953 return [v for (k, _), v in self._store.items() if k == kind]
956# ---------------------------------------------------------------------------
957# AgentResolver
958# ---------------------------------------------------------------------------
960def _deep_merge(base: dict, override: dict) -> dict:
961 """Syvä merge: override voittaa konflikteissa."""
962 result = dict(base)
963 for k, v in override.items():
964 if isinstance(v, dict) and isinstance(result.get(k), dict):
965 result[k] = _deep_merge(result[k], v)
966 else:
967 result[k] = v
968 return result
971def _tool_has_description(tool: Any) -> bool:
972 """Report whether a resolved tool exposes a non-empty model description."""
973 description = (tool.definition().get("function") or {}).get("description")
974 return isinstance(description, str) and bool(description)
977class AgentResolver:
978 """
979 Palauttaa fully-resolved agent spec:n instanssille.
980 Jos instanssilla on based_on, mergee Agent-base + AgentInstance-overridet.
981 """
983 def __init__(self, registry: ManifestRegistry) -> None:
984 self._registry = registry
985 self._cache: dict[str, dict] = {}
987 def resolve(self, instance_name: str) -> dict:
988 if instance_name in self._cache:
989 return self._cache[instance_name]
991 instance = self._registry.get("AgentInstance", instance_name)
992 if instance is None:
993 agent = self._registry.get("Agent", instance_name)
994 if agent is None:
995 raise RunHardFailure(
996 "VALIDATION_ERROR",
997 f"No Agent or AgentInstance named '{instance_name}'",
998 {"agent": instance_name},
999 )
1000 resolved = agent
1001 else:
1002 based_on = (instance.get("spec") or {}).get("based_on")
1003 if based_on:
1004 base_agent = self._registry.get("Agent", based_on)
1005 if base_agent is None:
1006 raise RunHardFailure(
1007 "VALIDATION_ERROR",
1008 f"AgentInstance '{instance_name}' references unknown "
1009 f"Agent '{based_on}'",
1010 {"agent": instance_name, "based_on": based_on},
1011 )
1012 base_spec = base_agent.get("spec") or {}
1013 inst_spec = {k: v for k, v in (instance.get("spec") or {}).items()
1014 if k != "based_on"}
1015 merged_spec = _deep_merge(base_spec, inst_spec)
1016 else:
1017 merged_spec = {k: v for k, v in (instance.get("spec") or {}).items()
1018 if k != "based_on"}
1020 resolved = {
1021 "apiVersion": "agent-platform/v1",
1022 "kind": "Agent",
1023 "metadata": {"name": instance_name},
1024 "spec": merged_spec,
1025 }
1027 self._cache[instance_name] = resolved
1028 return resolved
1031# ---------------------------------------------------------------------------
1032# State type
1033# ---------------------------------------------------------------------------
1035PipelineState = dict[str, Any]
1037# Engine-reserved channel key: the existing flat Core-managed JSON state lives
1038# under this key when a pipeline declares spec.channels.
1039_CORE = CORE_KEY
1041# Core-owned per-step map-reduce accumulator. A concurrent top-level `parallel`
1042# step fans out through native `Send`; each worker writes `{index: item_result}`
1043# under its step id and the join node consumes it. In the channels path the
1044# accumulator is a top-level state channel; in the no-channels path it is a
1045# reserved flat key folded by `_merge_state`.
1046_MAP_REDUCE_ACCUMULATOR_KEY = "__map_reduce_accum__"
1048# Reserved key carrying a worker's source item index through a `Send` payload.
1049_MAP_REDUCE_ITEM_INDEX = "__map_reduce_item_index__"
1051# Flat-state keys the engine owns; channel names may not shadow them.
1052_CORE_STATE_KEYS = frozenset({
1053 "errors", "stats", "run_status", "project_root", "run_id",
1054 "_stats", "step_context_metrics", "step_tool_meta",
1055 _MAP_REDUCE_ACCUMULATOR_KEY,
1056})
1058# Engine-local iteration keys the loop machinery injects into the read state.
1059# They are engine plumbing, not user state, so a colliding channel key must
1060# never shadow them.
1061_ENGINE_LOCAL_ITERATION_KEYS = ("item", "_item", "_foreach_item", "loop")
1064def _iteration_discriminator(state: dict) -> str | None:
1065 """The engine-local iteration token for *state*, or None when not iterating.
1067 Concurrent foreach/parallel items share a step id but differ in their
1068 iteration key, so the synthesized call id uses this to stay distinct on the
1069 in-cluster sidecar params file. The item is digested so the id stays a
1070 bounded, filename-safe length.
1071 """
1072 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1073 if key in state:
1074 digest = hashlib.sha1(repr(state[key]).encode("utf-8")).hexdigest()[:12]
1075 return f"{key}-{digest}"
1076 return None
1080def _merge_state(left: dict, right: dict) -> dict:
1081 if not left:
1082 return dict(right)
1083 side_channel_keys = ("step_context_metrics", "step_tool_meta")
1084 merged = dict(left)
1085 for key, value in right.items():
1086 if key == _MAP_REDUCE_ACCUMULATOR_KEY:
1087 folded = accumulate_map_reduce(merged.get(key), value)
1088 if folded:
1089 merged[key] = folded
1090 else:
1091 merged.pop(key, None)
1092 continue
1093 existing = merged.get(key)
1094 if (
1095 key in side_channel_keys
1096 and isinstance(existing, dict)
1097 and isinstance(value, dict)
1098 ):
1099 merged[key] = _merge_side_channel(existing, value)
1100 else:
1101 merged[key] = value
1102 return merged
1105def _build_state_schema() -> Any:
1106 """Build the channel-path TypedDict: the Core namespace plus the accumulator.
1108 Channel context is Core-owned, so there are no per-channel native LangGraph
1109 fields; the flat committed snapshot (core keys + channel keys) travels
1110 node-to-node under ``__core__``. The map-reduce accumulator is a separate
1111 top-level channel so native ``Send`` workers reduce into it; LangGraph stays
1112 the graph executor.
1113 """
1114 from typing import TypedDict
1116 return TypedDict(
1117 "PipelineState",
1118 {
1119 _CORE: Annotated[dict[str, Any], _merge_state],
1120 _MAP_REDUCE_ACCUMULATOR_KEY: Annotated[
1121 dict[str, Any], accumulate_map_reduce
1122 ],
1123 },
1124 )
1127# Sentinel for optional kwargs distinguishing "not provided" from None
1128_UNSET: Any = object()
1130# ---------------------------------------------------------------------------
1131# BuiltinExecutor
1132# ---------------------------------------------------------------------------
1134class BuiltinExecutor:
1135 """
1136 Ajaa builtin-komponentit.
1137 Uusia komponentteja voi rekisteröidä register()-metodilla.
1138 Nimet normalisoidaan: viiva → alaviiva (esim. 'list-dir-entries' == 'list_dir_entries').
1139 """
1141 # Lazy modules are imported on first use (they need httpx / heavier deps).
1142 # Profile-specific builtins live here: the PR-profile github-pr-comment
1143 # and the TICKET-profile ticket-publish (per-run local feature branch; push/PR disabled).
1144 _LAZY_MODULES: tuple[str, ...] = ("_builtins_pr", "_builtins_ticket")
1146 def __init__(self) -> None:
1147 self._handlers: dict[str, Any] = {
1148 "passthrough": self._passthrough,
1149 # Engine-internal call-expansion scaffolding (ADR-0037 D4): these are
1150 # synthesized by the build-time call composer, never authored in a
1151 # manifest. A call step becomes a compiled subgraph; these two op
1152 # steps run inside it, binding params and emitting the call result.
1153 "__call_bind": self._call_bind,
1154 "__call_emit": self._call_emit,
1155 }
1156 # Pipeline's LLM invoker (spec, step, state) → state, bound by
1157 # GraphBuilder. Profile builtins that make LLM decisions
1158 # (ticket-dor / ticket-review) read it via their register_all
1159 # wrapper — the executor must have it set before the builtin
1160 # executes (binding happens at build_workflow time).
1161 self.invoke_agent_fn: Any | None = None
1162 self._lazy_loaded = False
1164 def register(self, name: str, fn: Any) -> None:
1165 self._handlers[name] = fn
1166 # Also register with normalised name (hyphen → underscore)
1167 normalised = name.replace("-", "_")
1168 if normalised != name:
1169 self._handlers[normalised] = fn
1171 def _load_lazy_modules(self) -> None:
1172 if self._lazy_loaded:
1173 return
1174 self._lazy_loaded = True
1175 import importlib
1176 for module_name in self._LAZY_MODULES:
1177 try:
1178 mod = importlib.import_module(f".{module_name}", package="langgraph")
1179 except ImportError:
1180 continue
1181 mod.register_all(self)
1183 def has(self, name: str) -> bool:
1184 """True when *name* resolves in the builtin register (lazy modules included)."""
1185 self._load_lazy_modules()
1186 return name in self._handlers or name.replace("-", "_") in self._handlers
1188 def get(self, name: str) -> Any:
1189 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
1190 if fn is not None:
1191 return fn
1192 self._load_lazy_modules()
1193 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
1194 if fn is None:
1195 raise RunHardFailure(
1196 "VALIDATION_ERROR",
1197 f"Unknown builtin component '{name}'",
1198 {"component": name},
1199 )
1200 return fn
1202 def execute(self, component: str, step: dict, state: PipelineState) -> PipelineState:
1203 fn = self.get(component)
1204 try:
1205 return fn(step, state)
1206 except RunHardFailure as exc:
1207 log.error("[hard-failure] %s: %s", exc.code, exc.message)
1208 raise
1210 # --- default builtins ---
1212 @staticmethod
1213 def _passthrough(step: dict, state: PipelineState) -> PipelineState:
1214 return state
1216 @staticmethod
1217 def _call_bind(step: dict, state: PipelineState) -> PipelineState:
1218 """Call-scope entry (synthesized): bind the function call's params.
1220 The caller's function params are resolved against the current state and
1221 stored at ``state[<call_id>] = {"params": {...}}`` so the (rewritten)
1222 function-body templates ``{{ <call_id>.params.<name> }}`` resolve. The
1223 call's reserved ``__call_id`` key is stripped from the bound params.
1224 """
1225 call_id = (step.get("params") or {}).get("__call_id")
1226 if not call_id:
1227 raise ValueError("__call_bind missing __call_id (not a synthesized step)")
1228 bound = {
1229 k: v for k, v in (step.get("params") or {}).items() if k != "__call_id"
1230 }
1231 return {**state, call_id: {"params": bound}}
1233 @staticmethod
1234 def _call_emit(step: dict, state: PipelineState) -> PipelineState:
1235 """Call-scope exit (synthesized): surface the function's outputs.
1237 Per the call step-result contract (00-process-contract r.70) the caller
1238 sees only ``state[<call_id>] = {outputs: {<name>: <source value>}}`` —
1239 the function body step results are invisible to the caller. Each output
1240 name resolves to the manifest interface ``outputs[].source`` dot-path.
1242 Optionality (ADR-0031): an output flagged ``optional: true`` is skipped
1243 when its source-path is unresolved; an output that is NOT optional must
1244 resolve or a RunHardFailure is raised (no silent None — ADR-0012).
1245 """
1246 params = step.get("params") or {}
1247 call_id = params.get("__call_id")
1248 if not call_id:
1249 raise ValueError("__call_emit missing __call_id (not a synthesized step)")
1250 outputs_spec = params.get("__outputs") or []
1251 outputs: dict[str, Any] = {}
1252 for spec in outputs_spec:
1253 if not isinstance(spec, dict):
1254 continue
1255 name = spec.get("name")
1256 source = spec.get("source")
1257 if name is None:
1258 continue
1259 optional = bool(spec.get("optional", False))
1260 if source is None:
1261 # No declared source-path: treat as unresolved against the
1262 # optionality gate (ADR-0012 — no silent None for required).
1263 if optional:
1264 continue
1265 raise RunHardFailure(
1266 "UNRESOLVED_OUTPUT",
1267 f"call '{call_id}' output '{name}' has no source "
1268 "(required output — no silent None, ADR-0012)",
1269 {"call_id": call_id, "name": name, "source": source},
1270 )
1271 try:
1272 value = _dot_get_strict(state, source)
1273 except ValueError:
1274 # Unresolved source-path: required → hard error; optional → skip.
1275 if optional:
1276 continue
1277 raise RunHardFailure(
1278 "UNRESOLVED_OUTPUT",
1279 f"call '{call_id}' output '{name}' unresolved: source "
1280 f"'{source}' not found in state (required output — "
1281 "no silent None, ADR-0012)",
1282 {"call_id": call_id, "name": name, "source": source},
1283 ) from None
1284 outputs[name] = value
1285 return {**state, call_id: {"outputs": outputs}}
1288_DEFAULT_BUILTINS = BuiltinExecutor()
1291def _register_domain_builtins(executor: BuiltinExecutor) -> None:
1292 """Rekisteröi v1-domain-builtinit (list-dir-entries, ticket-publisher, ...).
1294 Erotettu omaan moduliin (`_builtins.py`) — runner.py pysyy generic-engine,
1295 builtinit ovat generiisiä mekanismeja (ADR-0016/0017).
1296 """
1297 try:
1298 from langgraph import _builtins # type: ignore
1299 except ImportError:
1300 # Kun runner.py ajetaan ilman pakettirakennetta (esim. /app/langgraph)
1301 import importlib.util
1302 spec = importlib.util.spec_from_file_location(
1303 "_builtins", Path(__file__).parent / "_builtins.py"
1304 )
1305 if spec and spec.loader:
1306 _builtins = importlib.util.module_from_spec(spec)
1307 spec.loader.exec_module(_builtins) # type: ignore[union-attr]
1308 else:
1309 log.warning("Cannot load _builtins.py — domain builtins disabled")
1310 return
1311 _builtins.register_all(executor)
1312 _register_function_builtins(executor)
1315def _register_function_builtins(executor: BuiltinExecutor) -> None:
1316 """Rekisteröi function-engine builtinit (question-block-form).
1318 Erotettu omaan moduliin (`_builtins_functions.py`) — runner.py pysyy
1319 generic-engine, function-builtinit pysyvät puhtaina ja erillisinä.
1320 """
1321 try:
1322 from langgraph import _builtins_functions # type: ignore
1323 except ImportError:
1324 import importlib.util
1325 spec = importlib.util.spec_from_file_location(
1326 "_builtins_functions", Path(__file__).parent / "_builtins_functions.py"
1327 )
1328 if spec and spec.loader:
1329 _builtins_functions = importlib.util.module_from_spec(spec)
1330 spec.loader.exec_module(_builtins_functions) # type: ignore[union-attr]
1331 else:
1332 log.warning("Cannot load _builtins_functions.py — function builtins disabled")
1333 return
1334 _builtins_functions.register_all(executor)
1337_register_domain_builtins(_DEFAULT_BUILTINS)
1339# ---------------------------------------------------------------------------
1340# StepRunner
1341# ---------------------------------------------------------------------------
1343class StepRunner:
1344 """
1345 Ajaa yksittäisen step:n sen typin mukaan.
1347 run_context: dict joka lisätään template-kontekstiin. Esim:
1348 {"budget": {"max_parallel": 16, ...}, "config": {...}}
1349 """
1351 def __init__(
1352 self,
1353 resolver: AgentResolver,
1354 builtins: BuiltinExecutor | None = None,
1355 invoke_agent_fn: Any | None = None,
1356 run_context: dict | None = None,
1357 *,
1358 step_functions: StepFunctionExecutor | None = None,
1359 tracer: Any | None = None,
1360 tool_resolver: Any | None = None,
1361 ) -> None:
1362 self._resolver = resolver
1363 self._builtins = builtins or _DEFAULT_BUILTINS
1364 self._invoke_agent = self._wrap_hard_failure_gate(
1365 invoke_agent_fn or self._dummy_invoke
1366 )
1367 self._run_context: dict = run_context or {}
1368 self._step_functions = step_functions
1369 self._tracer = tracer
1370 self._tool_resolver = tool_resolver
1371 self._channel_names: frozenset[str] = frozenset()
1373 def set_channel_routing(self, channel_names: frozenset[str]) -> None:
1374 """Attach the build-time routing context nested body steps route with."""
1375 self._channel_names = channel_names
1377 @staticmethod
1378 def _should_skip_llm(step: dict, ctx: dict) -> bool:
1379 """Truthiness of a step's `skip_llm_if` expression (ENG-12, Q4).
1381 Lenient (non-strict) evaluation: a missing/undefined field resolves to
1382 falsy, so the step behaves normally (LLM invoked). Only an explicit
1383 truthy expression skips the LLM.
1384 """
1385 raw = step.get("skip_llm_if")
1386 if raw is None:
1387 return False
1388 resolved = _resolve_template(raw, ctx)
1389 if isinstance(resolved, bool):
1390 return resolved
1391 if isinstance(resolved, str):
1392 return resolved.lower() not in ('false', '0', '', 'none', 'null')
1393 return bool(resolved)
1395 @staticmethod
1396 def _skip_llm_passthrough(
1397 step: dict,
1398 state: PipelineState,
1399 ctx: dict,
1400 read_state: PipelineState,
1401 ) -> PipelineState:
1402 """Deterministic passthrough of a step's resolved input (no LLM).
1404 `state[step_id]` becomes the resolved `input` value (the parallel
1405 fan-out envelope `{results: [...]}` when the input resolves to a list).
1406 ``read_state`` is the working-set overlay the input resolves against;
1407 the returned state is still built from the raw ``state``.
1408 """
1409 step_id = step.get("id", "?")
1410 input_spec = step.get("input")
1411 resolved = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
1412 if isinstance(resolved, list):
1413 log.debug("[step] %s skip_llm_if passthrough (fanout, no LLM)", step_id)
1414 return {**state, step_id: {"results": resolved}}
1415 log.debug("[step] %s skip_llm_if passthrough (no LLM)", step_id)
1416 return {**state, step_id: resolved}
1418 @staticmethod
1419 def _declared_item_source(step: dict) -> tuple[str, Any] | None:
1420 """A fan-out step's declared item source as ``(kind, spec)``, else None.
1422 ``input`` takes precedence over ``over``. ``over`` addresses a flat
1423 state key; ``input`` supports templates, bracket literals, and dot-paths
1424 (so a runtime envelope set is ``input: "<step_id>.iteration"``). A step
1425 declaring neither has no item source.
1426 """
1427 if step.get("input") is not None:
1428 return ("input", step.get("input"))
1429 if step.get("over"):
1430 return ("over", step.get("over"))
1431 return None
1433 @staticmethod
1434 def _resolve_item_source(
1435 source: tuple[str, Any], state: PipelineState, ctx: dict
1436 ) -> Any:
1437 """Resolve a declared item source to its raw value (no validation)."""
1438 kind, spec = source
1439 if kind == "input":
1440 return _resolve_input(spec, state, ctx)
1441 return state.get(spec)
1443 @staticmethod
1444 def _unresolvable_item_source_failure(
1445 step_id: str, source: tuple[str, Any]
1446 ) -> "RunHardFailure":
1447 kind, spec = source
1448 return RunHardFailure(
1449 "VALIDATION_ERROR",
1450 f"step '{step_id}' item source '{kind}: {spec}' is unresolvable "
1451 "(no silent fallback — ADR-0012)",
1452 {"step_id": step_id, "source_kind": kind, "source": spec},
1453 )
1455 @staticmethod
1456 def _empty_item_source_failure(
1457 step_id: str, source: tuple[str, Any]
1458 ) -> "RunHardFailure":
1459 kind, spec = source
1460 return RunHardFailure(
1461 "VALIDATION_ERROR",
1462 f"step '{step_id}' item source '{kind}: {spec}' resolved to an "
1463 "empty list (no silent fallback — ADR-0012)",
1464 {"step_id": step_id, "source_kind": kind, "source": spec},
1465 )
1467 @staticmethod
1468 def _parallel_fanout_items(
1469 step: dict, state: PipelineState, ctx: dict
1470 ) -> list | None:
1471 """The resolved fan-out item list for a parallel step, else None.
1473 A parallel step whose resolved item source is a list invokes the agent
1474 once per item and stores a ``{"results": [...]}`` envelope instead of a
1475 single agent output. A parallel step declaring no item source keeps the
1476 single-agent fallback; a declared source that is unresolvable or empty
1477 is a ``VALIDATION_ERROR`` (ADR-0012).
1478 """
1479 if step.get("type") != "parallel":
1480 return None
1481 source = StepRunner._declared_item_source(step)
1482 if source is None:
1483 return None
1484 items = StepRunner._resolve_item_source(source, state, ctx)
1485 if items is None:
1486 raise StepRunner._unresolvable_item_source_failure(
1487 step.get("id", "?"), source
1488 )
1489 if not isinstance(items, list):
1490 return None
1491 if not items:
1492 raise StepRunner._empty_item_source_failure(step.get("id", "?"), source)
1493 return items
1495 @staticmethod
1496 def _parallel_fanout_item_list(
1497 step: dict, state: PipelineState, ctx: dict
1498 ) -> list | None:
1499 """The fan-out item list with the ``max_items`` cap applied."""
1500 items = StepRunner._parallel_fanout_items(step, state, ctx)
1501 if items is None:
1502 return None
1503 max_items_raw = step.get("max_items")
1504 if max_items_raw is not None:
1505 items = items[: _resolve_int(max_items_raw, ctx, len(items))]
1506 return items
1508 @staticmethod
1509 def _foreach_item_list(
1510 step: dict, state: PipelineState, ctx: dict
1511 ) -> list:
1512 """The foreach item list: resolved, wrapped, empty-checked, capped.
1514 A missing source, an unresolvable source, and an empty list are hard
1515 ``VALIDATION_ERROR`` failures (ADR-0012). ``max_iterations`` caps the
1516 list.
1517 """
1518 step_id = step.get("id", "?")
1519 source = StepRunner._declared_item_source(step)
1520 if source is None:
1521 raise RunHardFailure(
1522 "VALIDATION_ERROR",
1523 f"foreach step '{step_id}' declares no item source "
1524 "('input' or 'over') — no silent [None] fallback (ADR-0012)",
1525 {"step_id": step_id},
1526 )
1527 resolved_items = StepRunner._resolve_item_source(source, state, ctx)
1528 if resolved_items is None:
1529 raise StepRunner._unresolvable_item_source_failure(step_id, source)
1530 items: list = (
1531 resolved_items
1532 if isinstance(resolved_items, list)
1533 else [resolved_items]
1534 )
1535 if not items:
1536 raise StepRunner._empty_item_source_failure(step_id, source)
1537 max_iter_raw = step.get("max_iterations")
1538 if max_iter_raw is not None:
1539 items = items[: _resolve_int(max_iter_raw, ctx, len(items))]
1540 return items
1542 def ran_agent(self, step: dict, state: PipelineState) -> bool:
1543 """Whether a step's run invokes its agent.
1545 False for agent-less steps, a ``skip_llm_if`` passthrough, and a
1546 parallel fan-out (per-item agent calls storing a result envelope).
1547 Channel routing is restricted to actual agent outputs, so these paths
1548 are excluded from inverse projection.
1549 """
1550 if self.resolve_agent(step) is None:
1551 return False
1552 ctx = self._make_ctx(state)
1553 if self._should_skip_llm(step, ctx):
1554 return False
1555 return self._parallel_fanout_items(step, state, ctx) is None
1557 def _make_ctx(self, state: dict, **extras: Any) -> dict:
1558 """Rakentaa template-kontekstin: state + run_context + extras."""
1559 ctx: dict = {}
1560 ctx.update(state)
1561 ctx.update(self._run_context)
1562 ctx.update(extras)
1563 return ctx
1565 @staticmethod
1566 def _read_state(
1567 state: PipelineState, working_set: ChannelWorkingSet | None
1568 ) -> PipelineState:
1569 """Overlay the invocation working set's read view onto the local state.
1571 The working set carries the committed snapshot plus this invocation's
1572 pending writes; the local state contributes keys the snapshot does not
1573 hold (item context, body step results). With no working set the raw
1574 state is the read view, preserving legacy behaviour exactly. The
1575 engine-local iteration keys are re-asserted after the overlay so a
1576 declared channel with the same name cannot shadow them.
1577 """
1578 if working_set is None:
1579 return state
1580 resolved = {**state, **working_set.read_view()}
1581 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1582 if key in state:
1583 resolved[key] = state[key]
1584 return resolved
1586 def _run_body_step(
1587 self,
1588 step: dict,
1589 state: PipelineState,
1590 working_set: ChannelWorkingSet | None,
1591 ) -> PipelineState:
1592 """Run one nested body step and stage its declared channel writes.
1594 The two-argument call preserves the legacy recursion signature for the
1595 no-channels path (and for callers that monkey-patch ``run_step``).
1596 """
1597 if working_set is None:
1598 return self.run_step(step, state)
1599 result = self.run_step(step, state, working_set)
1600 self._stage_body_channel_writes(step, state, result, working_set)
1601 return result
1603 def _stage_body_channel_writes(
1604 self,
1605 step: dict,
1606 pre_state: PipelineState,
1607 result_state: PipelineState,
1608 working_set: ChannelWorkingSet,
1609 ) -> None:
1610 """Stage a body step's declared channel writes into the working set.
1612 The router early-outs for agent-less steps (builtins, call bind/emit);
1613 a step carrying an agent routes its declared writes.
1614 """
1615 if not self._channel_names:
1616 return
1617 writes = _route_declared_channels(
1618 step,
1619 self._read_state(pre_state, working_set),
1620 result_state,
1621 self,
1622 self._channel_names,
1623 )
1624 for name, value in writes.items():
1625 working_set.stage(name, value)
1627 def _resolve_step_runtime(
1628 self,
1629 step: dict,
1630 state: dict,
1631 ctx: dict,
1632 *,
1633 input_override: Any = _UNSET,
1634 extra_params: dict | None = None,
1635 ) -> dict:
1636 """
1637 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
1638 templaattien osalta state + run_context -kontekstia vasten.
1640 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
1641 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
1642 """
1643 resolved: dict = dict(step)
1645 if input_override is not _UNSET:
1646 resolved["input"] = input_override
1647 else:
1648 input_spec = step.get("input")
1649 if input_spec is not None:
1650 resolved["input"] = _resolve_input(input_spec, state, ctx)
1652 params_spec = step.get("params")
1653 if params_spec or extra_params:
1654 base = _resolve_template(params_spec, ctx) if params_spec else {}
1655 if not isinstance(base, dict):
1656 base = {}
1657 if extra_params:
1658 base = {**base, **extra_params}
1659 resolved["params"] = base
1661 return resolved
1663 def _attach_communication_context(
1664 self, resolved_step: dict, read_state: PipelineState
1665 ) -> dict:
1666 """Attach a step's projected COMMUNICATION context when declared.
1668 The projection is resolved from the invocation read overlay (the
1669 committed snapshot plus this invocation's pending writes) and carries
1670 only the step's declared input channels — never whole state. A step
1671 without a read declaration is returned unchanged.
1672 """
1673 communication = resolve_communication_context(resolved_step, read_state)
1674 if not communication:
1675 return resolved_step
1676 return {**resolved_step, COMMUNICATION_CONTEXT_KEY: communication}
1678 def _attach_resolved_tools(
1679 self, resolved_step: dict, resolved_agent: dict
1680 ) -> dict:
1681 """Attach the agent's declared `tools` as resolved `Tool` instances.
1683 Resolves the agent's declared tool names through `ToolResolver` into
1684 ordered `Tool` objects under `RESOLVED_TOOLS_KEY`. An agent without
1685 tools, or a runner without a resolver, is returned unchanged.
1686 """
1687 tools = (resolved_agent.get("spec") or {}).get("tools") or []
1688 if not tools or self._tool_resolver is None:
1689 return resolved_step
1690 return {
1691 **resolved_step,
1692 RESOLVED_TOOLS_KEY: self._resolve_agent_tools(
1693 resolved_agent, list(tools)
1694 ),
1695 }
1697 def _resolve_agent_tools(
1698 self, resolved_agent: dict, tools: list[str]
1699 ) -> list:
1700 """Resolve an agent's declared tools, enforcing the instruction guard.
1702 The low-level resolver stays permissive; the policy lives here
1703 (ADR-0016). A declared name the resolver cannot resolve, or a
1704 resolved tool that exposes no non-empty description, is translated
1705 into the engine's build-time `AGENT_INSTRUCTION_MISSING` hard failure.
1706 A dual-source name keeps its own `BUILTIN_NAME_COLLISION` failure.
1707 """
1708 return [self._resolve_agent_tool(resolved_agent, name) for name in tools]
1710 def _resolve_agent_tool(self, resolved_agent: dict, name: str):
1711 try:
1712 tool = self._tool_resolver.resolve([name])[0]
1713 except ToolCollisionError as exc:
1714 raise RunHardFailure(
1715 "BUILTIN_NAME_COLLISION",
1716 f"tool {name!r} exists in both the platform registry and the "
1717 "StepFunction register",
1718 {"tool": name},
1719 ) from exc
1720 except ToolRegistryError as exc:
1721 raise self._instruction_missing_failure(
1722 resolved_agent, name, str(exc)
1723 ) from exc
1724 if not _tool_has_description(tool):
1725 raise self._instruction_missing_failure(
1726 resolved_agent, name, "the resolved tool exposes no description"
1727 )
1728 return tool
1730 @staticmethod
1731 def _instruction_missing_failure(
1732 resolved_agent: dict, tool_name: str, detail: str
1733 ) -> RunHardFailure:
1734 agent_name = (resolved_agent.get("metadata") or {}).get("name")
1735 return RunHardFailure(
1736 AGENT_INSTRUCTION_MISSING,
1737 f"Agent {agent_name!r} declares tool {tool_name!r} with no usage "
1738 f"instruction: {detail}",
1739 {"agent": agent_name, "tool": tool_name},
1740 )
1742 @staticmethod
1743 def _dummy_invoke(
1744 resolved_spec: dict, step: dict, state: PipelineState
1745 ) -> PipelineState:
1746 """Placeholder — ei kutsu oikeaa LLM:ää."""
1747 step_id = step.get("id", "?")
1748 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
1749 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
1751 @staticmethod
1752 def _merge_errors(base: list, additions: list) -> list:
1753 """Return a new list containing all items from *base* plus any items from
1754 *additions* that are not already present. Order is preserved and no
1755 in-place mutation occurs — safe to call from any context."""
1756 if not additions:
1757 return base
1758 merged = list(base)
1759 for err in additions:
1760 if err not in merged:
1761 merged.append(err)
1762 return merged
1764 @staticmethod
1765 def _wrap_hard_failure_gate(invoke_fn: Any) -> Any:
1766 """Kietoo invokerin: jokaiseen agent-step-tulokseen sovelletaan
1767 kova-virhe-portti (RunHardFailure details.error_code-perusteella).
1768 Suppein funnel joka näkee KAIKKI agent-tulokset: sequential,
1769 parallel (single + fanout), foreach (body + agent-polku) ja while."""
1770 def invoke_with_gate(
1771 resolved_spec: dict, step: dict, state: PipelineState
1772 ) -> PipelineState:
1773 result = invoke_fn(resolved_spec, step, state)
1774 step_id = step.get("id", "?")
1775 StepRunner._raise_on_hard_error(result.get(step_id), step_id)
1776 return result
1777 return invoke_with_gate
1779 @staticmethod
1780 def _is_hard_fail(details: dict) -> bool:
1781 """Kova virhe vain kun: error_code on HARD_ERROR_CODESissa
1782 (TRUNCATION_ERROR, TOOL_FAILURE, AGENT_ERROR) tai VALIDATION_ERROR
1783 kantaa kontekstiylityksen signatuurin (assembled_tokens +
1784 context_budget — _llm._context_overflow_result).
1785 Generic-VALIDATION_ERROR ei kaada ajoa. status:error-envelope kaatuu
1786 _raise_on_error_envelope-polulla ennen tätä tarkistusta."""
1787 code = details.get("error_code")
1788 if code in RunHardFailure.HARD_ERROR_CODES:
1789 return True
1790 if code == "VALIDATION_ERROR":
1791 return {"assembled_tokens", "context_budget"} <= details.keys()
1792 return False
1794 @staticmethod
1795 def _raise_on_hard_error(step_result: Any, step_id: str) -> None:
1796 """Nosta RunHardFailure jos step-tulos kantaa kovan virheen
1797 (_is_hard_fail) tai on status:error-envelope; stepin details
1798 kulkeutuu poikkeuksen mukana."""
1799 if not isinstance(step_result, dict):
1800 return
1801 if is_error_envelope(step_result):
1802 StepRunner._raise_on_error_envelope(step_result, step_id)
1803 return
1804 details = step_result.get("details")
1805 if not isinstance(details, dict):
1806 return
1807 if not StepRunner._is_hard_fail(details):
1808 return
1809 code = details.get("error_code")
1810 err = step_result.get("error", "")
1811 message = err if isinstance(err, str) and err else f"{code}: step '{step_id}' hard failure"
1812 log.error("[hard-failure] %s: %s", code, message)
1813 raise RunHardFailure(code, message, details)
1815 @staticmethod
1816 def _raise_on_error_envelope(step_result: dict, step_id: str) -> None:
1817 """Nosta RunHardFailure status:error-envelopesta; envelopen error
1818 (dict) kulkeutuu details-attribuuttina, muuten minimi details."""
1819 envelope_error = step_result.get("error")
1820 if isinstance(envelope_error, dict):
1821 details = envelope_error
1822 message = envelope_error.get("message")
1823 else:
1824 details = {"error_code": ENVELOPE_ERROR_CODE, "step_id": step_id}
1825 message = None
1826 if not isinstance(message, str) or not message:
1827 message = f"{ENVELOPE_ERROR_CODE}: step '{step_id}' envelope error"
1828 raise RunHardFailure(ENVELOPE_ERROR_CODE, message, details)
1830 @staticmethod
1831 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
1832 """
1833 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
1834 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
1835 epäonnistuneeksi eikä hiljennä virhettä.
1837 Kaikki virhetyypit propagoidaan:
1838 - budget:max_llm_calls / budget:run_timeout → budjettiloki
1839 - parse_error:*, timeout:*, ja muut → error-loki
1840 """
1841 step_result = result.get(step_id)
1842 if not isinstance(step_result, dict):
1843 return result
1844 err = step_result.get("error", "")
1845 if not isinstance(err, str) or not err:
1846 return result
1847 errors = list(result.get("errors") or [])
1848 message = f"{step_id}: {err}"
1849 if message not in errors:
1850 errors.append(message)
1851 if err.startswith("budget:"):
1852 log.warning("[budget] %s", message)
1853 else:
1854 log.error("[step-error] %s", message)
1855 return {**result, "errors": errors}
1857 def resolve_agent(self, step: dict) -> dict | None:
1858 """Resolve the agent/instance a step runs, or None for agent-less steps."""
1859 agent_name = step.get("agent")
1860 if not agent_name:
1861 return None
1862 return self._resolver.resolve(agent_name)
1864 def run_step(
1865 self,
1866 step: dict,
1867 state: PipelineState,
1868 working_set: ChannelWorkingSet | None = None,
1869 ) -> PipelineState:
1870 step_id = step.get("id", "?")
1871 read_state = self._read_state(state, working_set)
1872 ctx = self._make_ctx(read_state)
1873 step_type = step.get("type", "sequential")
1875 # Evaluate condition — if false, skip step. A `while` never reaches
1876 # this point as a raw step: build-time composition rewrites it into a
1877 # native cycle placeholder whose head owns the entry-condition skip.
1878 cond = step.get("condition")
1879 if cond is not None and not _eval_condition(cond, ctx):
1880 log.debug("[step] %s skipped (condition false)", step_id)
1881 return state
1883 if step_type == "builtin":
1884 component = step.get("component", "passthrough")
1885 log.debug("[step] %s builtin:%s", step_id, component)
1886 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
1887 # template-arvot (esim. "{{ budget.max_parallel }}")
1888 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1889 # ADR-0037 D2: resolve the component against the platform builtin
1890 # register FIRST, then the kind: StepFunction registry. A name in
1891 # both is a hard error (fail-fast).
1892 is_builtin = self._builtins.has(component)
1893 is_step_function = (
1894 self._step_functions is not None
1895 and self._step_functions.has(component)
1896 )
1897 if is_builtin and is_step_function:
1898 raise RunHardFailure(
1899 "BUILTIN_NAME_COLLISION",
1900 f"component '{component}' is present in both the builtin "
1901 "register and the StepFunction registry (ADR-0037 — "
1902 "fail-fast)",
1903 {"component": component},
1904 )
1905 if is_step_function:
1906 return self._step_functions.execute(component, resolved_step, state)
1907 builtin_state = self._builtins.execute(component, resolved_step, state)
1908 self._raise_on_hard_error(builtin_state.get(step_id), step_id)
1909 return builtin_state
1911 # skip_llm_if (ENG-12, Q4): when the step's skip_llm_if evaluates truthy,
1912 # do NOT invoke the LLM (no inference budget consumed) and set the step
1913 # result to a deterministic passthrough of the step's resolved input.
1914 if (
1915 step_type in ("sequential", "parallel")
1916 and self._should_skip_llm(step, ctx)
1917 ):
1918 return self._skip_llm_passthrough(step, state, ctx, read_state)
1920 if step_type == "parallel":
1921 # Fan-out: if input resolves to a list, call agent once per item
1922 items = self._parallel_fanout_items(step, read_state, ctx)
1923 if items is not None:
1924 return self._run_parallel_fanout(
1925 step, state, items, ctx, working_set
1926 )
1927 # Fallback: single agent call
1928 agent_name = step.get("agent")
1929 if not agent_name:
1930 log.warning("[step] %s has no agent — skipping", step_id)
1931 return state
1932 resolved = self._resolver.resolve(agent_name)
1933 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
1934 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1935 resolved_step = self._attach_communication_context(
1936 resolved_step, read_state
1937 )
1938 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
1939 result = self._invoke_agent(resolved, resolved_step, state)
1940 return self._propagate_step_errors(result, step_id)
1942 if step_type == "sequential":
1943 agent_name = step.get("agent")
1944 if not agent_name:
1945 log.warning("[step] %s has no agent — skipping", step_id)
1946 return state
1947 resolved = self._resolver.resolve(agent_name)
1948 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
1949 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1950 resolved_step = self._attach_communication_context(
1951 resolved_step, read_state
1952 )
1953 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
1954 result = self._invoke_agent(resolved, resolved_step, state)
1955 return self._propagate_step_errors(result, step_id)
1957 if step_type == "foreach":
1958 return self._run_foreach(step, state, working_set)
1960 if step_type == "call":
1961 # Build-time composition (ADR-0037 D4) rewrites every call step into
1962 # a native subgraph node, so a call reaching the runtime means the
1963 # builder skipped a composition path it was responsible for.
1964 raise RunHardFailure(
1965 "VALIDATION_ERROR",
1966 f"call step '{step_id}' reached runtime without build-time "
1967 "expansion (ADR-0037 D4) — not yet supported",
1968 {"step_id": step_id},
1969 )
1971 if step_type == _SUBGRAPH_STEP_TYPE:
1972 # Build-time composition (ADR-0037 D4) embeds a compiled function
1973 # subgraph as a native node. At the top level ``_assemble_graph``
1974 # wires it through ``_make_subgraph_node``; when the call sits in an
1975 # imperative foreach/while body the placeholder reaches ``run_step``
1976 # instead, so invoke the embedded graph here. The top-of-method
1977 # condition check already handled a false condition (returned the
1978 # input state unchanged), matching the node wrapper's skip.
1979 compiled = step.get(_SUBGRAPH_GRAPH_KEY)
1980 if compiled is None:
1981 raise RunHardFailure(
1982 "VALIDATION_ERROR",
1983 f"subgraph step '{step_id}' has no compiled graph",
1984 {"step_id": step_id},
1985 )
1986 config = _while_invoke_config(step)
1987 composed_step_ids = (
1988 step.get(_SUBGRAPH_STEP_IDS_KEY)
1989 if not step.get(_SUBGRAPH_WHILE_KEY)
1990 and not step.get(_SUBGRAPH_FOREACH_KEY)
1991 else None
1992 )
1993 if working_set is not None:
1994 # Channel path: the compiled subgraph shares the parent's
1995 # Core-namespaced schema, while a loop body carries the flat
1996 # committed snapshot. Re-wrap on entry, flatten on exit. No
1997 # channel reducer is re-applied here — the subgraph's own
1998 # commits already produced the authoritative values.
1999 sub_result = compiled.invoke({_CORE: read_state}, config)
2000 committed = _flatten_channel_state(sub_result)
2001 return _filter_composed_step_ids(
2002 {**state, **committed}, composed_step_ids
2003 )
2004 return _filter_composed_step_ids(
2005 compiled.invoke(state, config), composed_step_ids
2006 )
2008 if step_type == "spawn":
2009 return self._run_spawn(step, state, working_set)
2011 raise RunHardFailure(
2012 "VALIDATION_ERROR",
2013 f"Unknown step type '{step_type}' in step '{step_id}'",
2014 {"step_id": step_id, "step_type": step_type},
2015 )
2017 def _run_parallel_fanout(
2018 self,
2019 step: dict,
2020 state: PipelineState,
2021 items: list,
2022 ctx: dict,
2023 working_set: ChannelWorkingSet | None = None,
2024 ) -> PipelineState:
2025 """
2026 Fan-out: call agent once per item, concurrently.
2027 Results stored as state[step_id] = {"results": [item_result, ...]}.
2028 Each item is available as state["item"] inside the invoke call.
2029 """
2030 step_id = step.get("id", "?")
2031 agent_name = step.get("agent")
2032 if not agent_name:
2033 log.warning("[step] %s parallel fanout: no agent", step_id)
2034 return {**state, step_id: {"results": []}}
2036 resolved = self._resolver.resolve(agent_name)
2038 # Apply max_items cap
2039 max_items_raw = step.get("max_items")
2040 if max_items_raw is not None:
2041 max_items = _resolve_int(max_items_raw, ctx, len(items))
2042 items = items[:max_items]
2044 # Concurrency limit
2045 max_conc_raw = step.get("max_concurrent")
2046 if max_conc_raw is not None:
2047 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2048 else:
2049 max_conc = len(items) or 1
2050 max_conc = max(1, min(max_conc, len(items) or 1))
2052 log.debug(
2053 "[step] %s parallel fanout: %d items, %d concurrent",
2054 step_id, len(items), max_conc,
2055 )
2057 read_state = self._read_state(state, working_set)
2059 def call_one(item: Any) -> tuple[Any, dict | None, dict | None, list]:
2060 item_state = {**read_state, "_item": item, "item": item}
2061 item_ctx = self._make_ctx(item_state)
2062 return self._invoke_parallel_item(
2063 step, resolved, item, item_state, item_ctx
2064 )
2066 # Pre-allocate results to preserve input order regardless of completion
2067 # order (ENG-07), matching the concurrent foreach body-polku.
2068 results: list = [None] * len(items)
2069 item_metrics: dict = {}
2070 item_tool_meta: dict = {}
2071 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2072 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
2073 for future in as_completed(futures):
2074 idx = futures[future]
2075 try:
2076 item_result, item_metrics_map, item_tool_meta_map, *_ = (
2077 future.result()
2078 )
2079 results[idx] = item_result
2080 if isinstance(item_metrics_map, dict):
2081 item_metrics.update(item_metrics_map)
2082 if isinstance(item_tool_meta_map, dict):
2083 item_tool_meta.update(item_tool_meta_map)
2084 except SkillLoadError:
2085 raise
2086 except RunHardFailure:
2087 pool.shutdown(wait=False, cancel_futures=True)
2088 raise
2089 except Exception as exc: # noqa: BLE001
2090 log.error("[step] %s item failed: %s", step_id, exc)
2091 results[idx] = {"error": str(exc)}
2093 self._stage_fanout_channel_writes(step, results, working_set)
2095 final: PipelineState = {**state, step_id: {"results": results}}
2096 if item_metrics:
2097 final = {
2098 **final,
2099 "step_context_metrics": _merge_side_channel(
2100 state.get("step_context_metrics"), item_metrics
2101 ),
2102 }
2103 if item_tool_meta:
2104 final = {
2105 **final,
2106 "step_tool_meta": _merge_side_channel(
2107 state.get("step_tool_meta"), item_tool_meta
2108 ),
2109 }
2110 return final
2112 def _invoke_parallel_item(
2113 self,
2114 step: dict,
2115 resolved: dict,
2116 item: Any,
2117 item_state: PipelineState,
2118 item_ctx: dict,
2119 ) -> tuple[Any, dict | None, dict | None, list]:
2120 """Run one parallel fan-out item's agent call.
2122 Shared by the in-node thread-pool fan-out and the native ``Send``
2123 worker, so both paths resolve input, communication context and tools
2124 identically. Returns ``(step_result, metrics, tool_meta, errors)``.
2125 """
2126 step_id = step.get("id", "?")
2127 resolved_step = self._resolve_step_runtime(
2128 step, item_state, item_ctx,
2129 input_override=item,
2130 )
2131 resolved_step = self._attach_communication_context(
2132 resolved_step, item_state
2133 )
2134 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2135 result_state = self._invoke_agent(resolved, resolved_step, item_state)
2136 return (
2137 result_state.get(step_id),
2138 result_state.get("step_context_metrics"),
2139 result_state.get("step_tool_meta"),
2140 list(result_state.get("errors") or []),
2141 )
2143 def _stage_fanout_channel_writes(
2144 self,
2145 step: dict,
2146 item_outputs: list,
2147 working_set: ChannelWorkingSet | None,
2148 ) -> None:
2149 """Stage each fan-out item's declared channel writes in item order.
2151 Runs on the main thread after the item pool completes, so the shared
2152 working set is never mutated from a worker thread. The fan-out
2153 envelope itself is never routed; only each item's own agent output is.
2154 """
2155 if working_set is None or not self._channel_names:
2156 return
2157 for output in item_outputs:
2158 writes = _declared_output_channel_writes(
2159 step, output, self._channel_names
2160 )
2161 for name, value in writes.items():
2162 working_set.stage(name, value)
2164 def _run_foreach(
2165 self,
2166 step: dict,
2167 state: PipelineState,
2168 working_set: ChannelWorkingSet | None = None,
2169 ) -> PipelineState:
2170 """
2171 Foreach: iterate over input list, run body steps per item.
2172 Input from step.input (dot-path or template) or step.over (state key).
2174 When max_concurrent > 1, body iterations run concurrently via
2175 ThreadPoolExecutor — each item gets its own worker thread. The
2176 max_concurrent value sets the thread pool size (mirrors parallel
2177 fan-out semantics). Results list preserves original item order.
2179 max_output budget in concurrent mode: a stop flag prevents new workers
2180 from starting after the budget is exhausted. Workers already executing
2181 when the flag is set will complete naturally — at most max_concurrent
2182 extra items may run past the limit. This is accepted behaviour; the
2183 sequential early-exit guarantee cannot be reproduced without sacrificing
2184 true concurrency.
2186 The invocation working set is threaded into every body step so reads
2187 resolve over the committed snapshot plus this invocation's pending
2188 writes; the returned state is still built from the raw ``state``.
2189 """
2190 step_id = step.get("id", "?")
2191 read_state = self._read_state(state, working_set)
2192 ctx = self._make_ctx(read_state)
2194 items = self._foreach_item_list(step, read_state, ctx)
2196 # max_output cap — stop once N iterations have produced a successful
2197 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
2198 max_output_raw = step.get("max_output")
2199 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
2201 # max_concurrent — when > 1, enables parallel body execution
2202 max_conc_raw = step.get("max_concurrent")
2203 if max_conc_raw is not None:
2204 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2205 max_conc = max(1, min(max_conc, len(items) or 1))
2206 else:
2207 max_conc = 1
2209 body = step.get("body") or []
2210 agent_name = step.get("agent")
2212 if max_conc <= 1:
2213 raise RunHardFailure(
2214 "VALIDATION_ERROR",
2215 f"foreach step '{step_id}' reached runtime without build-time "
2216 "composition (a sequential foreach is compiled into a native "
2217 "cycle) — internal invariant violated",
2218 {"step_id": step_id},
2219 )
2220 return self._run_foreach_concurrent(
2221 step_id, step, state, items, body, agent_name, max_output, max_conc,
2222 working_set,
2223 )
2225 @staticmethod
2226 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
2227 """body-polku per-iteration result (r.67): the declared body-output
2228 step's result; else the iteration's `publish` result; else None."""
2229 if declared_output:
2230 return iter_state.get(declared_output)
2231 return iter_state.get("publish")
2233 @staticmethod
2234 def _body_item_passed(item: Any) -> bool:
2235 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
2236 `passed` bool; an item lacking such a key is not counted as failed."""
2237 if isinstance(item, dict):
2238 verdict = item.get("verdict")
2239 if verdict == "FAIL":
2240 return False
2241 if verdict == "PASS":
2242 return True
2243 passed = item.get("passed")
2244 if isinstance(passed, bool):
2245 return passed
2246 return True
2248 @staticmethod
2249 def _aggregate_body_results(item_results: list) -> dict:
2250 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
2251 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
2252 return {
2253 "results": item_results,
2254 "passed": not failing,
2255 "failing": failing,
2256 "failed": bool(failing),
2257 }
2259 @staticmethod
2260 def _merge_item_working_sets(
2261 working_set: ChannelWorkingSet | None,
2262 item_working_sets: list[ChannelWorkingSet | None],
2263 ) -> None:
2264 """Stage each concurrent item's writes onto the shared set in item order."""
2265 if working_set is None:
2266 return
2267 for item_working_set in item_working_sets:
2268 if item_working_set is None:
2269 continue
2270 for name, value in item_working_set.pending_writes():
2271 working_set.stage(name, value)
2273 def _run_foreach_concurrent(
2274 self,
2275 step_id: str,
2276 step: dict,
2277 state: PipelineState,
2278 items: list,
2279 body: list,
2280 agent_name: str | None,
2281 max_output: int,
2282 max_conc: int,
2283 working_set: ChannelWorkingSet | None = None,
2284 ) -> PipelineState:
2285 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2286 log.info(
2287 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2288 step_id, len(items), max_conc,
2289 )
2291 # Pre-allocate results list to preserve item order across threads.
2292 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2293 # legitimate None results (body produced no output) are kept.
2294 body_polku = bool(body)
2295 declared_output = step.get("output")
2296 _SKIPPED = object()
2297 results: list = [_SKIPPED] * len(items)
2298 item_working_sets: list[ChannelWorkingSet | None] = [None] * len(items)
2299 lock = threading.Lock()
2300 produced_outputs = [0] # mutable counter shared across threads
2301 stop_flag = threading.Event() # set when max_output budget is exhausted
2302 item_errors: list = [] # budget errors collected from worker threads
2303 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2304 accumulated_tool_meta: dict = {} # per-item tool metadata from worker threads
2306 def run_item(idx: int, item: Any) -> None:
2307 nonlocal item_errors
2308 # Honour stop flag — skip if budget already exhausted before we start
2309 if max_output and stop_flag.is_set():
2310 return
2312 item_working_set = (
2313 working_set.fork() if working_set is not None else None
2314 )
2315 item_working_sets[idx] = item_working_set
2316 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2318 if body_polku:
2319 for sub_step in body:
2320 iter_state = self._run_body_step(
2321 sub_step, iter_state, item_working_set
2322 )
2323 # Collect budget errors from this iteration (thread-safe merge)
2324 new_errs = iter_state.get("errors") or []
2325 if new_errs:
2326 with lock:
2327 item_errors = self._merge_errors(item_errors, new_errs)
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)
2336 publish_out = iter_state.get("publish")
2337 results[idx] = self._body_item_result(iter_state, declared_output)
2338 if isinstance(publish_out, dict) and publish_out.get("path"):
2339 with lock:
2340 produced_outputs[0] += 1
2341 if max_output and produced_outputs[0] >= max_output:
2342 stop_flag.set()
2343 elif agent_name:
2344 resolved = self._resolver.resolve(agent_name)
2345 item_read_state = self._read_state(iter_state, item_working_set)
2346 item_ctx = self._make_ctx(item_read_state)
2347 resolved_step = self._resolve_step_runtime(
2348 step, item_read_state, item_ctx
2349 )
2350 resolved_step = self._attach_communication_context(
2351 resolved_step, item_read_state
2352 )
2353 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2354 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2355 results[idx] = iter_state.get(step_id)
2356 item_metrics_map = iter_state.get("step_context_metrics")
2357 if isinstance(item_metrics_map, dict):
2358 with lock:
2359 accumulated_metrics.update(item_metrics_map)
2360 item_tool_meta_map = iter_state.get("step_tool_meta")
2361 if isinstance(item_tool_meta_map, dict):
2362 with lock:
2363 accumulated_tool_meta.update(item_tool_meta_map)
2365 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2366 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2367 for future in as_completed(futures):
2368 try:
2369 future.result()
2370 except SkillLoadError:
2371 raise
2372 except RunHardFailure:
2373 pool.shutdown(wait=False, cancel_futures=True)
2374 raise
2375 except Exception as exc: # noqa: BLE001
2376 idx = futures[future]
2377 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2378 results[idx] = {"error": str(exc)}
2380 self._merge_item_working_sets(working_set, item_working_sets)
2382 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2383 ordered = [r for r in results if r is not _SKIPPED]
2384 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
2385 final: PipelineState = {**state, step_id: step_result}
2386 if accumulated_metrics:
2387 final = {
2388 **final,
2389 "step_context_metrics": _merge_side_channel(
2390 state.get("step_context_metrics"), accumulated_metrics
2391 ),
2392 }
2393 if accumulated_tool_meta:
2394 final = {
2395 **final,
2396 "step_tool_meta": _merge_side_channel(
2397 state.get("step_tool_meta"), accumulated_tool_meta
2398 ),
2399 }
2400 if item_errors:
2401 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2402 return final
2404 def _run_spawn(
2405 self,
2406 step: dict,
2407 state: PipelineState,
2408 working_set: ChannelWorkingSet | None = None,
2409 ) -> PipelineState:
2410 """Run a spawn step as a native dynamic ``Send`` fan-out subgraph.
2412 The ``agent`` runs once as director at the tree root and may emit a
2413 child plan at ``envelope.content.children``; a shared child-worker node
2414 re-runs the ``child`` role (default = ``agent``) for each child and
2415 re-``Send``s itself for deeper levels. Depth and fan are bounded by the
2416 spawn_policy, and the whole tree draws from one shared SpawnBudget.
2417 """
2418 step_id = step.get("id", "?")
2419 director = step.get("agent")
2420 if not director:
2421 log.warning("[step] %s spawn has no agent — skipping", step_id)
2422 return state
2423 read_state = self._read_state(state, working_set)
2424 ctx = self._make_ctx(read_state)
2425 policy = step.get("spawn_policy") or {}
2426 max_depth = _resolve_int(
2427 policy.get("max_depth"), ctx, _SPAWN_DEFAULT_MAX_DEPTH
2428 )
2429 max_spawn_per_node = _resolve_int(
2430 policy.get("max_spawn_per_node"), ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
2431 )
2432 max_calls_raw = policy.get("max_agent_calls")
2433 budget = SpawnBudget(
2434 _SPAWN_DEFAULT_MAX_AGENT_CALLS
2435 if max_calls_raw is None
2436 else _resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS)
2437 )
2438 run_max_parallel = int(
2439 (self._run_context.get("budget") or {}).get(
2440 "max_parallel", DEFAULT_MAX_PARALLEL
2441 )
2442 )
2443 max_concurrent_raw = policy.get("max_concurrent_children")
2444 max_concurrent_children = (
2445 run_max_parallel
2446 if max_concurrent_raw is None
2447 else _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
2448 )
2449 child_role = step.get("child") or director
2451 outcome = self._run_native_spawn(
2452 step=step,
2453 step_id=step_id,
2454 director=director,
2455 child_role=child_role,
2456 base_state=read_state,
2457 base_ctx=ctx,
2458 budget=budget,
2459 max_depth=max_depth,
2460 max_spawn_per_node=max_spawn_per_node,
2461 max_concurrent_children=max_concurrent_children,
2462 )
2463 envelope = outcome["envelope"]
2464 final: PipelineState = {**state, step_id: envelope}
2465 if outcome["metrics"]:
2466 final = {
2467 **final,
2468 "step_context_metrics": _merge_side_channel(
2469 state.get("step_context_metrics"), outcome["metrics"]
2470 ),
2471 }
2472 if outcome["tool_meta"]:
2473 final = {
2474 **final,
2475 "step_tool_meta": _merge_side_channel(
2476 state.get("step_tool_meta"), outcome["tool_meta"]
2477 ),
2478 }
2479 self._stage_spawn_channel_writes(step, envelope["result"], working_set)
2480 return final
2482 def _run_native_spawn(
2483 self,
2484 *,
2485 step: dict,
2486 step_id: str,
2487 director: str,
2488 child_role: str,
2489 base_state: PipelineState,
2490 base_ctx: dict,
2491 budget: SpawnBudget,
2492 max_depth: int,
2493 max_spawn_per_node: int | None,
2494 max_concurrent_children: int,
2495 ) -> dict:
2496 """Compile and invoke the native spawn tree, returning its outcome.
2498 The graph is a director node that seeds one ``Send`` per child to a
2499 shared worker, whose worker re-``Send``s itself for deeper levels. An
2500 outstanding-descendants barrier (``outstanding`` counter plus a ``check``
2501 node) makes the ``join`` run exactly once for uneven-depth trees; the
2502 join assembles the result envelope by path key in plan order.
2503 """
2504 from contextlib import nullcontext
2505 from typing import TypedDict
2507 from langgraph.graph import END, START, StateGraph
2508 from langgraph.types import Command, Send
2510 tracer = self._tracer
2511 parent_obs = (
2512 tracer.open_parent_observation(
2513 name=f"spawn:{step_id}",
2514 input={"node_id": step_id, "role": director, "depth": 0},
2515 )
2516 if tracer is not None
2517 else None
2518 )
2519 root_obs_id = getattr(parent_obs, "id", None)
2520 try:
2521 child_gates: dict[str, threading.Semaphore] = {}
2522 child_gates_lock = threading.Lock()
2524 def child_gate(parent_id: str) -> threading.Semaphore:
2525 """The concurrency gate for one spawning node's direct children.
2527 ``max_concurrent_children`` bounds a single spawn node's fan-out:
2528 each spawning node owns its own gate, so a shallow node's cap
2529 never throttles a deeper sibling's subtree.
2530 """
2531 with child_gates_lock:
2532 gate = child_gates.get(parent_id)
2533 if gate is None:
2534 gate = threading.Semaphore(max(1, max_concurrent_children))
2535 child_gates[parent_id] = gate
2536 return gate
2538 def run_node(
2539 *,
2540 role: str,
2541 node_step: dict,
2542 input_override: Any,
2543 node_id: str,
2544 depth: int,
2545 parent_id: str | None = None,
2546 ) -> dict:
2547 resolved_role = self._resolver.resolve(role)
2548 resolved_step = self._resolve_step_runtime(
2549 node_step, base_state, base_ctx, input_override=input_override
2550 )
2551 resolved_step = self._attach_communication_context(
2552 resolved_step, base_state
2553 )
2554 resolved_step = self._attach_resolved_tools(
2555 resolved_step, resolved_role
2556 )
2557 if not budget.consume():
2558 raise RunHardFailure(
2559 "SPAWN_BUDGET_EXHAUSTED",
2560 f"spawn tree budget exhausted at node '{node_id}'",
2561 {"node_id": node_id, "depth": depth},
2562 )
2563 gate = child_gate(parent_id) if parent_id is not None else nullcontext()
2564 with gate:
2565 return self._invoke_agent(
2566 resolved_role, resolved_step, base_state
2567 )
2569 def build_record(
2570 *,
2571 node_id: str,
2572 path: tuple,
2573 parent_path: tuple | None,
2574 depth: int,
2575 this_result: Any,
2576 result_state: dict,
2577 children: list,
2578 ) -> tuple[dict, list]:
2579 exhausted = False
2580 plan: list = []
2581 if children:
2582 if depth + 1 > max_depth:
2583 exhausted = True
2584 else:
2585 plan = [
2586 {
2587 "path": path + (index,),
2588 "depth": depth + 1,
2589 "child": child,
2590 "parent_id": node_id,
2591 "index": index,
2592 }
2593 for index, child in enumerate(children)
2594 ]
2595 record = {
2596 "path": path,
2597 "parent_path": parent_path,
2598 "child_id": node_id,
2599 "result": this_result,
2600 "self_passed": not self._spawn_is_soft_fail(this_result),
2601 "exhausted": exhausted,
2602 "metrics": dict(result_state.get("step_context_metrics") or {}),
2603 "tool_meta": dict(result_state.get("step_tool_meta") or {}),
2604 }
2605 return record, plan
2607 def child_plan(node_id: str, this_result: Any, depth: int) -> list:
2608 children = self._spawn_children_from_result(this_result)
2609 if children and max_spawn_per_node is not None and len(children) > max_spawn_per_node:
2610 raise RunHardFailure(
2611 "SPAWN_OVER_FLIGHT",
2612 f"spawn node '{node_id}' emitted {len(children)} children, "
2613 f"exceeds max_spawn_per_node={max_spawn_per_node}",
2614 {
2615 "node_id": node_id,
2616 "depth": depth,
2617 "children": len(children),
2618 "max_spawn_per_node": max_spawn_per_node,
2619 },
2620 )
2621 return children
2623 def director_node(_state: dict) -> Command:
2624 result_state = run_node(
2625 role=director,
2626 node_step=step,
2627 input_override=_UNSET,
2628 node_id=step_id,
2629 depth=0,
2630 )
2631 this_result = result_state.get(step_id)
2632 children = child_plan(step_id, this_result, 0)
2633 record, plan = build_record(
2634 node_id=step_id,
2635 path=(),
2636 parent_path=None,
2637 depth=0,
2638 this_result=this_result,
2639 result_state=result_state,
2640 children=children,
2641 )
2642 update = {
2643 "accum": {step_id: {(): record}},
2644 "outstanding": len(plan) - 1,
2645 }
2646 if not plan:
2647 return Command(update=update, goto="join")
2648 return Command(
2649 update=update,
2650 goto=[
2651 Send("worker", {**payload, "parent_obs_id": root_obs_id})
2652 for payload in plan
2653 ],
2654 )
2656 def worker_node(payload: dict) -> Command:
2657 path = tuple(payload["path"])
2658 depth = payload["depth"]
2659 parent_id = payload["parent_id"]
2660 index = payload["index"]
2661 child = payload["child"]
2662 child_id = (
2663 child["id"]
2664 if isinstance(child, dict) and child.get("id")
2665 else f"{parent_id}.{index}"
2666 )
2667 node_step = self._child_node_step(
2668 node_id=child_id,
2669 child=child,
2670 parent_id=parent_id,
2671 index=index,
2672 base_state=base_state,
2673 base_ctx=base_ctx,
2674 )
2675 # The worker's own graph context does not carry the spawning
2676 # node's observation (each Send target runs in a fresh context),
2677 # so the parent chain is re-established from the payload before
2678 # opening this node's observation (ADR-0035 D5).
2679 parent_obs_id = payload.get("parent_obs_id")
2680 stack_token = (
2681 _PARENT_STACK.set((parent_obs_id,))
2682 if tracer is not None and parent_obs_id is not None
2683 else None
2684 )
2685 child_obs = (
2686 tracer.open_parent_observation(
2687 name=f"spawn:{child_id}",
2688 input={"node_id": child_id, "role": child_role, "depth": depth},
2689 )
2690 if tracer is not None
2691 else None
2692 )
2693 try:
2694 result_state = run_node(
2695 role=child_role,
2696 node_step=node_step,
2697 input_override=node_step["input"],
2698 node_id=child_id,
2699 depth=depth,
2700 parent_id=parent_id,
2701 )
2702 finally:
2703 if tracer is not None and child_obs is not None:
2704 tracer.close_parent_observation(child_obs)
2705 if stack_token is not None:
2706 _PARENT_STACK.reset(stack_token)
2707 this_result = result_state.get(child_id)
2708 children = child_plan(child_id, this_result, depth)
2709 record, plan = build_record(
2710 node_id=child_id,
2711 path=path,
2712 parent_path=path[:-1],
2713 depth=depth,
2714 this_result=this_result,
2715 result_state=result_state,
2716 children=children,
2717 )
2718 update = {
2719 "accum": {step_id: {path: record}},
2720 "outstanding": len(plan) - 1,
2721 }
2722 if plan:
2723 child_obs_id = getattr(child_obs, "id", None)
2724 return Command(
2725 update=update,
2726 goto=[
2727 Send(
2728 "worker",
2729 {**payload, "parent_obs_id": child_obs_id},
2730 )
2731 for payload in plan
2732 ],
2733 )
2734 return Command(update=update, goto="check")
2736 def check_node(state: dict) -> Command:
2737 if state.get("outstanding") == 0:
2738 return Command(goto="join")
2739 return Command(goto=END)
2741 def join_node(state: dict) -> dict:
2742 records = (state.get("accum") or {}).get(step_id) or {}
2743 envelope, metrics, tool_meta = _assemble_spawn_outcome(records)
2744 return {
2745 "spawn_outcome": {
2746 "envelope": envelope,
2747 "metrics": metrics,
2748 "tool_meta": tool_meta,
2749 },
2750 "accum": {step_id: None},
2751 }
2753 schema = TypedDict(
2754 "_SpawnState",
2755 {
2756 "accum": Annotated[dict, accumulate_map_reduce],
2757 "outstanding": Annotated[int, operator.add],
2758 "spawn_outcome": dict,
2759 },
2760 )
2761 graph = StateGraph(schema)
2762 graph.add_node("director", director_node)
2763 graph.add_node("worker", worker_node)
2764 graph.add_node("check", check_node)
2765 graph.add_node("join", join_node)
2766 graph.add_edge(START, "director")
2767 graph.add_edge("join", END)
2768 app = graph.compile()
2769 out = app.invoke(
2770 {"accum": {}, "outstanding": 1},
2771 config={"recursion_limit": _SPAWN_RECURSION_LIMIT},
2772 )
2773 return out["spawn_outcome"]
2774 finally:
2775 if tracer is not None and parent_obs is not None:
2776 tracer.close_parent_observation(parent_obs)
2778 def _child_node_step(
2779 self,
2780 *,
2781 node_id: str,
2782 child: Any,
2783 parent_id: str,
2784 index: int,
2785 base_state: PipelineState,
2786 base_ctx: dict,
2787 ) -> dict:
2788 """Resolve one child plan item into the worker's runtime step.
2790 A child is a runtime-generated step: it carries only its id and its
2791 resolved input, never the spawn step's declared channel bindings, so a
2792 child's output can never route a channel write.
2793 """
2794 if not self._spawn_has_input(child):
2795 raise RunHardFailure(
2796 "INVALID_CHILD_PLAN",
2797 f"spawn child #{index} of node '{parent_id}' must be a dict "
2798 "with a non-empty 'input'",
2799 {"node_id": parent_id, "child_index": index, "child": repr(child)[:200]},
2800 )
2801 return {"id": node_id, "input": _resolve_input(child["input"], base_state, base_ctx)}
2803 def _stage_spawn_channel_writes(
2804 self,
2805 step: dict,
2806 result: Any,
2807 working_set: ChannelWorkingSet | None,
2808 ) -> None:
2809 """Stage a spawn node's declared writes from its own agent result."""
2810 if working_set is None:
2811 return
2812 writes = _declared_output_channel_writes(step, result, self._channel_names)
2813 for name, value in writes.items():
2814 working_set.stage(name, value)
2816 @staticmethod
2817 def _spawn_children_from_result(result: Any) -> list:
2818 """The node's child plan at ``content.children``; [] for a valid leaf.
2820 The director's parsed output envelope carries the plan at
2821 ``content.children``; a result that is not a dict, or whose ``content``
2822 is not an object, or carries no ``children``, is a valid leaf. A
2823 ``children`` value that is not a list is a hard ``INVALID_CHILD_PLAN``.
2824 """
2825 content = _envelope_content(result)
2826 if not isinstance(content, dict):
2827 return []
2828 raw = content.get("children")
2829 if raw is None:
2830 return []
2831 if not isinstance(raw, list):
2832 raise RunHardFailure(
2833 "INVALID_CHILD_PLAN",
2834 "spawn agent content 'children' must be a list",
2835 {"children": repr(raw)[:200]},
2836 )
2837 return raw
2839 @staticmethod
2840 def _spawn_has_input(child: Any) -> bool:
2841 """A valid child plan item: a dict carrying a non-empty 'input'."""
2842 if not isinstance(child, dict):
2843 return False
2844 input_val = child.get("input")
2845 if input_val is None:
2846 return False
2847 if isinstance(input_val, str) and not input_val.strip():
2848 return False
2849 if isinstance(input_val, (list, dict)) and not input_val:
2850 return False
2851 return True
2853 @staticmethod
2854 def _spawn_is_soft_fail(result: Any) -> bool:
2855 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error.
2857 An envelope-shaped result is never soft: ``status:error`` is a hard
2858 failure (the gate raises before this), and ``ok``/``partial``/
2859 ``no_success`` are valid non-fatal results.
2860 """
2861 if not isinstance(result, dict):
2862 return False
2863 if result.get("status") in ENVELOPE_STATUSES:
2864 return False
2865 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
2866 return True
2867 return any(key.startswith("budget:") for key in result)
2870# ---------------------------------------------------------------------------
2871# GraphBuilder
2872# ---------------------------------------------------------------------------
2874def _topological_order(steps: list[dict]) -> list[dict]:
2875 """
2876 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2877 """
2878 id_to_step = {s["id"]: s for s in steps}
2879 visited: set[str] = set()
2880 order: list[dict] = []
2882 def visit(step_id: str) -> None:
2883 if step_id in visited:
2884 return
2885 visited.add(step_id)
2886 for dep in (id_to_step[step_id].get("depends_on") or []):
2887 if dep in id_to_step:
2888 visit(dep)
2889 order.append(id_to_step[step_id])
2891 for s in steps:
2892 visit(s["id"])
2893 return order
2896def _group_dependencies(
2897 groups: list[list[dict]],
2898) -> tuple[list[set[int]], list[set[int]]]:
2899 """Group-index adjacency (predecessors, successors) from each step's depends_on."""
2900 step_to_group = {
2901 step["id"]: index
2902 for index, group in enumerate(groups)
2903 for step in group
2904 }
2905 predecessors: list[set[int]] = [set() for _ in groups]
2906 successors: list[set[int]] = [set() for _ in groups]
2907 for index, group in enumerate(groups):
2908 for step in group:
2909 for dep in step.get("depends_on") or []:
2910 dep_index = step_to_group.get(dep)
2911 if dep_index is None or dep_index == index:
2912 continue
2913 predecessors[index].add(dep_index)
2914 successors[dep_index].add(index)
2915 return predecessors, successors
2918@dataclass(frozen=True)
2919class _RoutePlan:
2920 """Compiled `routes` topology: route tables plus the static-edge suppression sets.
2922 `routes` owns a routing step's outgoing control flow (ADR-0025). A route
2923 target reached via a static `depends_on` edge is therefore suppressed and
2924 replaced by the native conditional edge; the plan records both the targets
2925 (so they are never made entry points) and the suppressed `(source, target)`
2926 pairs.
2927 """
2929 step_to_node: dict[str, str]
2930 routes_by_step: dict[str, list[dict]]
2931 target_node_ids: frozenset[str]
2932 routing_node_ids: frozenset[str]
2933 suppressed_edges: frozenset[tuple[str, str]]
2936def _reject_nested_routes(step: dict) -> None:
2937 """Reject `routes` on a nested `body` step (top-level-only rule, ADR-0012).
2939 The engine wires `routes` only for top-level steps, so a nested declaration
2940 would be silently ignored; nested routing is deferred.
2941 """
2942 body = step.get("body")
2943 if not isinstance(body, list):
2944 return
2945 for nested in _iter_steps(body):
2946 if nested.get("routes"):
2947 raise RunHardFailure(
2948 "VALIDATION_ERROR",
2949 f"Step '{nested.get('id')}' declares `routes` inside a nested "
2950 "body; `routes` is only allowed on top-level steps",
2951 )
2954def _build_route_plan(groups: list[list[dict]], node_ids: list[str]) -> _RoutePlan:
2955 """Resolve every top-level step's `routes` into a node-level routing plan."""
2956 step_to_node = {
2957 step["id"]: node_ids[index]
2958 for index, group in enumerate(groups)
2959 for step in group
2960 }
2961 step_to_deps = {
2962 step["id"]: set(step.get("depends_on") or [])
2963 for group in groups
2964 for step in group
2965 }
2966 routes_by_step: dict[str, list[dict]] = {}
2967 target_node_ids: set[str] = set()
2968 routing_node_ids: set[str] = set()
2969 suppressed_edges: set[tuple[str, str]] = set()
2970 for group in groups:
2971 for step in group:
2972 _reject_nested_routes(step)
2973 routes = step.get("routes")
2974 if not routes:
2975 continue
2976 source = step["id"]
2977 source_node = step_to_node[source]
2978 routing_node_ids.add(source_node)
2979 routes_by_step[source] = routes
2980 for route in routes:
2981 target = route.get("target")
2982 target_node = step_to_node.get(target)
2983 if target_node is None:
2984 raise RunHardFailure(
2985 "VALIDATION_ERROR",
2986 f"Step '{source}' route target '{target}' is not a "
2987 "top-level step in this pipeline",
2988 )
2989 target_node_ids.add(target_node)
2990 if source in step_to_deps.get(target, set()):
2991 suppressed_edges.add((source_node, target_node))
2992 log.warning(
2993 "[routes] suppressing static edge %s -> %s; route "
2994 "targets are wired via `routes`",
2995 source,
2996 target,
2997 )
2998 return _RoutePlan(
2999 step_to_node=step_to_node,
3000 routes_by_step=routes_by_step,
3001 target_node_ids=frozenset(target_node_ids),
3002 routing_node_ids=frozenset(routing_node_ids),
3003 suppressed_edges=frozenset(suppressed_edges),
3004 )
3007def _make_route_router(routes: list[dict], step_to_node: dict[str, str], end) -> Any:
3008 """First-match-wins router over a routing step's declared routes.
3010 Evaluates `when` in declaration order and returns the selected target node;
3011 a route without `when` is the default branch. With no match and no default
3012 it returns the graph terminal.
3013 """
3014 def router(state: PipelineState) -> str:
3015 ctx = _flatten_channel_state(state) if _CORE in state else state
3016 for route in routes:
3017 if _eval_condition(route.get("when"), ctx):
3018 return step_to_node[route["target"]]
3019 return end
3021 return router
3024def _wire_route_edges(graph, route_plan: _RoutePlan, end) -> None:
3025 """Add one native conditional edge per routing step, with an explicit path_map."""
3026 for source, routes in route_plan.routes_by_step.items():
3027 target_nodes = [route_plan.step_to_node[r["target"]] for r in routes]
3028 path_map = {node: node for node in target_nodes}
3029 path_map[end] = end
3030 graph.add_conditional_edges(
3031 route_plan.step_to_node[source],
3032 _make_route_router(routes, route_plan.step_to_node, end),
3033 path_map,
3034 )
3037def _wire_dependency_edges(
3038 graph,
3039 groups: list[list[dict]],
3040 node_ids: list[str],
3041 end,
3042 route_plan: _RoutePlan,
3043 entry_ids: list[str] | None = None,
3044 exit_ids: list[str] | None = None,
3045) -> None:
3046 """Wire native branches and fan-in joins from group-level depends_on edges.
3048 Static edges originating from a routing step to one of its route targets are
3049 suppressed — `routes` supplies that control flow. A route target with no
3050 other predecessor is reached by its conditional edge, never an entry point.
3051 A group split into a Send fan-out enters through its dispatch node and exits
3052 through its join node, so ``entry_ids``/``exit_ids`` name those boundaries.
3053 """
3054 entry_ids = entry_ids if entry_ids is not None else node_ids
3055 exit_ids = exit_ids if exit_ids is not None else node_ids
3056 predecessors, successors = _group_dependencies(groups)
3057 for index, node_id in enumerate(node_ids):
3058 entry = entry_ids[index]
3059 pred_nodes = [exit_ids[p] for p in sorted(predecessors[index])]
3060 if not pred_nodes:
3061 if node_id not in route_plan.target_node_ids:
3062 graph.set_entry_point(entry)
3063 else:
3064 live_preds = [
3065 pred
3066 for pred in pred_nodes
3067 if (pred, entry) not in route_plan.suppressed_edges
3068 ]
3069 if len(live_preds) == 1:
3070 graph.add_edge(live_preds[0], entry)
3071 elif live_preds:
3072 graph.add_edge(live_preds, entry)
3073 if not successors[index] and node_id not in route_plan.routing_node_ids:
3074 graph.add_edge(exit_ids[index], end)
3077# ---------------------------------------------------------------------------
3078# Call-step build-time expansion (ADR-0037 D3/D4)
3079# ---------------------------------------------------------------------------
3080#
3081# A `type: call` step is composed at build time: the function's body steps plus
3082# two synthesized op steps (`__call_bind` / `__call_emit`) are assembled into a
3083# compiled subgraph added as a single parent-graph node keyed by the call id. The
3084# runtime executes only native steps; LangGraph never sees a call step. The caller
3085# reads the function outputs via the call result shape
3086# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
3087# The function body's `{{ params.<name> }}` references are rewritten to
3088# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves. A
3089# call nested in a foreach/while body is composed the same way: the loop body is a
3090# native subgraph and the call becomes a compiled subgraph node keyed by the call id.
3091#
3092# Isolation (ADR-0037 D5 / 00-functions § Isolation) and recursion (ADR-0037 D7): a
3093# function body step must not reference a step-id outside its own body (a
3094# one-way dependency check — outward references to caller step-ids are build
3095# errors); a cycle in the function call graph (self or mutual) is a build
3096# error. Bounded while loops are runtime constructs and are not cycle-flagged.
3098# Call step result namespace keys (internal scaffolding).
3099def _scope_params_value(value: Any, call_id: str) -> Any:
3100 """Rewrite `params.` template references to the call-scoped namespace.
3102 Operates on the function body's string values (templates). Dict keys are
3103 untouched — only templated string VALUES are scoped, so a function body's
3104 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
3105 """
3106 if isinstance(value, str):
3107 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
3108 if isinstance(value, dict):
3109 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
3110 if isinstance(value, list):
3111 return [_scope_params_value(v, call_id) for v in value]
3112 return value
3115def _scope_params(step: dict, call_id: str) -> dict:
3116 """Deep-scope a function body step's params references for a call id."""
3117 scoped: dict = dict(step)
3118 for field in ("input", "params", "condition", "exit_condition", "agent", "function"):
3119 if field in scoped and isinstance(scoped[field], (str, dict, list)):
3120 scoped[field] = _scope_params_value(scoped[field], call_id)
3121 if "body" in scoped and isinstance(scoped["body"], list):
3122 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
3123 return scoped
3126def _collect_step_ids(steps: list[dict]) -> set[str]:
3127 """All step ids declared in *steps*, including nested body step ids."""
3128 ids: set[str] = set()
3129 for step in steps:
3130 step_id = step.get("id")
3131 if isinstance(step_id, str):
3132 ids.add(step_id)
3133 body = step.get("body")
3134 if isinstance(body, list):
3135 ids |= _collect_step_ids(body)
3136 return ids
3139def _iter_steps(steps: list[dict]):
3140 """Yield every step in *steps*, recursing into nested bodies."""
3141 for step in steps:
3142 yield step
3143 body = step.get("body")
3144 if isinstance(body, list):
3145 yield from _iter_steps(body)
3148def _collect_composed_step_ids(steps: list[dict]) -> set[str]:
3149 """All declared step ids, including ids composed into native subgraphs.
3151 A ``call`` step becomes a ``__subgraph__`` placeholder whose inner step ids
3152 live only inside the compiled subgraph. The placeholder records them under
3153 ``_SUBGRAPH_STEP_IDS_KEY`` so name-collision validation sees the full set,
3154 which stays encapsulated inside the compiled subgraph.
3155 """
3156 ids = _collect_step_ids(steps)
3157 for step in _iter_steps(steps):
3158 composed_ids = step.get(_SUBGRAPH_STEP_IDS_KEY)
3159 if isinstance(composed_ids, (set, frozenset)):
3160 ids |= composed_ids
3161 return ids
3164_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
3165_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
3166_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
3168_REFERENCE_FIELDS = ("input", "params", "condition", "exit_condition")
3170# Engine-reserved template namespaces — never step-id references, even if a
3171# caller happens to declare a step with one of these names.
3172_RESERVED_NAMESPACES = frozenset({"params", "loop", "item", "_item", "budget", "config"})
3175def _referenced_names(value: Any) -> set[str]:
3176 """Root identifiers read by templated strings.
3178 Only the leading identifier of each reference (before the first ``.`` or
3179 ``[``) can name a step-id; nested field segments are ignored. Quoted
3180 literals are ignored.
3181 """
3182 names: set[str] = set()
3183 if isinstance(value, str):
3184 for expr in _TEMPLATE_REF_RE.findall(value):
3185 bare = _STRING_LITERAL_RE.sub("", expr)
3186 for match in _IDENTIFIER_RE.finditer(bare):
3187 if match.start() == 0 or bare[match.start() - 1] != ".":
3188 names.add(match.group())
3189 elif isinstance(value, dict):
3190 for v in value.values():
3191 names |= _referenced_names(v)
3192 elif isinstance(value, list):
3193 for v in value:
3194 names |= _referenced_names(v)
3195 return names
3198def _validate_function_purity(
3199 fn_name: str, body: list[dict], visible_ids: set[str]
3200) -> None:
3201 """One-way isolation: a function body must not reference a caller step-id.
3203 A body step may read its own body step-ids and the params namespace; a
3204 reference to a step-id that exists outside the body (``visible_ids`` but
3205 not declared in the body) is an outward dependency and a hard build error.
3206 """
3207 local_ids = _collect_step_ids(body)
3208 for step in _iter_steps(body):
3209 step_id = step.get("id", "?")
3210 for dep in step.get("depends_on") or []:
3211 if dep in visible_ids and dep not in local_ids:
3212 raise RunHardFailure(
3213 "VALIDATION_ERROR",
3214 f"function '{fn_name}' step '{step_id}': depends_on "
3215 f"'{dep}' is outside the function body (isolation — "
3216 "function bodies see only their own steps and params)",
3217 )
3218 for field in _REFERENCE_FIELDS:
3219 for ref in _referenced_names(step.get(field)):
3220 if ref in _RESERVED_NAMESPACES:
3221 continue
3222 if ref in visible_ids and ref not in local_ids:
3223 raise RunHardFailure(
3224 "VALIDATION_ERROR",
3225 f"function '{fn_name}' step '{step_id}': '{field}' "
3226 f"references '{ref}', a step-id outside the function "
3227 "body (isolation — function bodies see only their own "
3228 "steps and params)",
3229 )
3232def _reject_function_body_routes(fn_name: str, body: list[dict]) -> None:
3233 """Reject `routes` anywhere in a function body (top-level-only rule).
3235 A function body is composed as a native subgraph, so a body step declaring
3236 `routes` would escape the nested-`body` guard and be silently rewired;
3237 reject it at build time instead.
3238 """
3239 for step in _iter_steps(body):
3240 if step.get("routes"):
3241 raise RunHardFailure(
3242 "VALIDATION_ERROR",
3243 f"function '{fn_name}' step '{step.get('id')}' declares `routes`; "
3244 "`routes` is only allowed on top-level steps",
3245 )
3248def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
3249 """Expand every ``call`` step through the native subgraph composer.
3251 Test-facing seam over the single call-expansion implementation: the same
3252 ``_CallComposer`` that ``GraphBuilder`` uses composes the native subgraph, so
3253 no second expansion path exists. The embedded subgraph value is the un-compiled
3254 sub-step list; a runnable graph only needs the compiled form, which
3255 production obtains through ``GraphBuilder._compose_calls``.
3256 """
3257 composer = _CallComposer(
3258 registry,
3259 assemble_subgraph=lambda sub_steps: sub_steps,
3260 channels=[],
3261 channel_names=frozenset(),
3262 )
3263 return composer.compose(steps)
3266_SUBGRAPH_STEP_TYPE = "__subgraph__"
3267_SUBGRAPH_GRAPH_KEY = "__subgraph_graph__"
3268_SUBGRAPH_STEP_IDS_KEY = "__subgraph_step_ids__"
3270# A `while` step is composed into a `__subgraph__` placeholder whose embedded
3271# graph is a native LangGraph cycle (head -> body -> cond -> head/exit). The
3272# marker flags the placeholder so the parent uses the while node wrapper (which
3273# keeps the loop-internal state out of the parent result) rather than the plain
3274# call-subgraph node, and carries the build-time recursion bound.
3275_SUBGRAPH_WHILE_KEY = "__subgraph_while__"
3276_WHILE_RECURSION_LIMIT_KEY = "__while_recursion_limit__"
3278# Engine-local cycle state: the active while's `{current, iteration, passed,
3279# skipped, max_iter}` marker. Nested whiles live in isolated cycle invocations,
3280# so a shared key cannot collide across them.
3281_WHILE_STATE_KEY = "__while_state__"
3283# State keys the while wrapper propagates from the isolated cycle back to the
3284# parent (the loop result itself is carried separately under the step id).
3285_WHILE_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
3287# A sequential `foreach` (max_concurrent absent or resolving to <= 1) is
3288# composed into a `__subgraph__` placeholder whose embedded graph is a native
3289# LangGraph cycle (head -> body/agent -> cond -> head/exit). The marker flags
3290# the placeholder so the parent uses the foreach node wrapper, and carries the
3291# build-time recursion bound.
3292_SUBGRAPH_FOREACH_KEY = "__subgraph_foreach__"
3293_FOREACH_RECURSION_LIMIT_KEY = "__foreach_recursion_limit__"
3295# Engine-local cycle state: the active foreach's `{items, index, results,
3296# produced_outputs, max_output}` marker. Nested foreachs live in isolated cycle
3297# invocations, so a shared key cannot collide across them.
3298_FOREACH_STATE_KEY = "__foreach_state__"
3300# State keys the foreach wrapper propagates from the isolated cycle back to the
3301# parent (the aggregate result itself is carried separately under the step id).
3302_FOREACH_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
3304# Build-time recursion bound fallback for a foreach with no declared
3305# `max_iterations`: the item list is uncapped, so reserve a generous bound.
3306_FOREACH_DEFAULT_MAX_ITER = 1000
3309class _CallComposer:
3310 """Compose `call` steps into compiled native subgraph nodes (ADR-0037 D4).
3312 A function body becomes its own compiled StateGraph added to the parent as a
3313 single node keyed by the call id; the ``bind``/``emit`` scaffolding and the
3314 body steps live inside that subgraph and share the parent state. Nested
3315 function calls compose recursively. A `call` nested in a foreach/while body
3316 is composed the same way; the loop body executes the embedded compiled graph
3317 imperatively via ``run_step``'s ``__subgraph__`` branch.
3318 """
3320 def __init__(
3321 self,
3322 registry: ManifestRegistry,
3323 assemble_subgraph: Any,
3324 channels: list[dict],
3325 channel_names: frozenset[str],
3326 *,
3327 assemble_while: Any | None = None,
3328 compose_whiles: bool = False,
3329 assemble_foreach: Any | None = None,
3330 compose_foreachs: bool = False,
3331 build_ctx: dict | None = None,
3332 ) -> None:
3333 self._registry = registry
3334 self._assemble_subgraph = assemble_subgraph
3335 self._channels = channels
3336 self._channel_names = channel_names
3337 self._assemble_while = assemble_while
3338 self._compose_whiles = compose_whiles
3339 self._assemble_foreach = assemble_foreach
3340 self._compose_foreachs = compose_foreachs
3341 self._build_ctx = build_ctx or {}
3342 self._stack: list[str] = []
3343 self._expanded_functions: dict[str, str] = {}
3345 def compose(self, steps: list[dict]) -> list[dict]:
3346 visible_ids = _collect_step_ids(steps)
3347 composed: list[dict] = []
3348 for step in steps:
3349 composed.extend(self._compose_step(step, visible_ids=visible_ids))
3350 return composed
3352 def _compose_step(self, step: dict, *, visible_ids: set[str]) -> list[dict]:
3353 step_type = step.get("type")
3354 if step_type == "call":
3355 return [self._compose_subgraph_call(step, visible_ids)]
3356 if step_type == "while" and self._compose_whiles:
3357 return [self._compose_while(step, visible_ids)]
3358 if step_type == "foreach" and self._should_compose_foreach(step):
3359 return [self._compose_foreach(step, visible_ids)]
3360 body = step.get("body")
3361 if not isinstance(body, list):
3362 return [step]
3363 nested: list[dict] = []
3364 for sub in body:
3365 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3366 return [{**step, "body": nested}]
3368 def _compose_while(self, step: dict, visible_ids: set[str]) -> dict:
3369 """Compose a `while` step into a native LangGraph cycle subgraph.
3371 The embedded graph owns the loop topology: a head that resolves the
3372 entry condition / initial ``current``, the composed body step nodes, a
3373 conditional back-edge that computes the feed + exit condition and routes
3374 continue-vs-exit, and an exit node that writes the documented while
3375 result. ``expand_calls`` leaves the step uncomposed (``compose_whiles``
3376 false) so the call-expansion seam still exposes the loop body.
3377 """
3378 _reject_nested_routes(step)
3379 body = step.get("body") or []
3380 nested: list[dict] = []
3381 for sub in body:
3382 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3383 _validate_step_channel_bindings(nested, self._channel_names)
3384 compiled, recursion_limit = self._assemble_while(step, nested)
3385 placeholder: dict = {
3386 "id": step["id"],
3387 "type": _SUBGRAPH_STEP_TYPE,
3388 _SUBGRAPH_GRAPH_KEY: compiled,
3389 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3390 _SUBGRAPH_WHILE_KEY: True,
3391 _WHILE_RECURSION_LIMIT_KEY: recursion_limit,
3392 "depends_on": list(step.get("depends_on") or []),
3393 }
3394 if step.get("routes"):
3395 placeholder["routes"] = step["routes"]
3396 return placeholder
3398 def _should_compose_foreach(self, step: dict) -> bool:
3399 """Whether a `foreach` is the sequential form composed into a cycle.
3401 Only a `max_concurrent` absent or resolving to ``<= 1`` is composed; a
3402 declared ``max_concurrent > 1`` keeps the native ``Send`` / in-node
3403 concurrent behaviour.
3404 """
3405 if not self._compose_foreachs:
3406 return False
3407 return _resolved_send_concurrency(step, self._build_ctx) <= 1
3409 def _compose_foreach(self, step: dict, visible_ids: set[str]) -> dict:
3410 """Compose a sequential `foreach` into a native LangGraph cycle subgraph.
3412 Mirrors ``_compose_while``: the embedded graph owns the loop topology
3413 (a head that resolves the item list and injects the current item, the
3414 composed body step nodes or the per-item agent node, a conditional
3415 back-edge routing next-item-vs-exit, and an exit node that writes the
3416 documented aggregate). ``expand_calls`` leaves the step uncomposed so
3417 the call-expansion seam still exposes the loop body.
3418 """
3419 _reject_nested_routes(step)
3420 body = step.get("body") or []
3421 nested: list[dict] = []
3422 for sub in body:
3423 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3424 _validate_step_channel_bindings(nested, self._channel_names)
3425 compiled, recursion_limit = self._assemble_foreach(step, nested)
3426 placeholder: dict = {
3427 "id": step["id"],
3428 "type": _SUBGRAPH_STEP_TYPE,
3429 _SUBGRAPH_GRAPH_KEY: compiled,
3430 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3431 _SUBGRAPH_FOREACH_KEY: True,
3432 _FOREACH_RECURSION_LIMIT_KEY: recursion_limit,
3433 "depends_on": list(step.get("depends_on") or []),
3434 }
3435 if step.get("routes"):
3436 placeholder["routes"] = step["routes"]
3437 return placeholder
3439 def _compose_subgraph_call(self, step: dict, visible_ids: set[str]) -> dict:
3440 call_id = step["id"]
3441 function, body = self._check_call_guards(step, visible_ids)
3443 nested_visible = visible_ids | _collect_step_ids(body)
3444 outputs_spec = (
3445 ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
3446 )
3447 bind_id = f"_call__{call_id}__bind"
3448 emit_id = f"_call__{call_id}__emit"
3449 bind_step: dict = {
3450 "id": bind_id,
3451 "type": "builtin",
3452 "component": "__call_bind",
3453 "params": {**(step.get("params") or {}), "__call_id": call_id},
3454 "depends_on": [],
3455 }
3457 self._stack.append(step["function"])
3458 try:
3459 expanded_body = self._compose_body(
3460 body, call_id, bind_id, nested_visible
3461 )
3462 finally:
3463 self._stack.pop()
3465 emit_step: dict = {
3466 "id": emit_id,
3467 "type": "builtin",
3468 "component": "__call_emit",
3469 "params": {"__call_id": call_id, "__outputs": outputs_spec},
3470 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
3471 }
3472 sub_steps = [bind_step, *expanded_body, emit_step]
3473 if self._channels:
3474 _validate_channel_inputs(self._channels, sub_steps)
3475 _validate_step_channel_bindings(sub_steps, self._channel_names)
3476 placeholder: dict = {
3477 "id": call_id,
3478 "type": _SUBGRAPH_STEP_TYPE,
3479 _SUBGRAPH_GRAPH_KEY: self._assemble_subgraph(
3480 _topological_order(sub_steps)
3481 ),
3482 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(sub_steps),
3483 "depends_on": list(step.get("depends_on") or []),
3484 }
3485 call_condition = step.get("condition")
3486 if call_condition is not None:
3487 placeholder["condition"] = call_condition
3488 return placeholder
3490 def _compose_body(
3491 self, body: list[dict], call_id: str, bind_id: str, visible_ids: set[str]
3492 ) -> list[dict]:
3493 expanded: list[dict] = []
3494 for body_step in body:
3495 scoped = _scope_params(body_step, call_id)
3496 for sub in self._compose_step(scoped, visible_ids=visible_ids):
3497 deps = list(sub.get("depends_on") or [])
3498 if bind_id in deps:
3499 expanded.append(sub)
3500 else:
3501 expanded.append({**sub, "depends_on": [bind_id, *deps]})
3502 return expanded
3504 def _check_call_guards(
3505 self, step: dict, visible_ids: set[str]
3506 ) -> tuple[dict, list[dict]]:
3507 """Single owner of every build-time ``call`` guard (ADR-0037 D3/D5/D7).
3509 Rejects routes on the call step, function cycles, duplicate expansion,
3510 unknown functions, function-body isolation (purity) violations, and
3511 body-declared routes; returns the resolved function and its body.
3512 """
3513 fn_name = step["function"]
3514 call_id = step["id"]
3516 if step.get("routes"):
3517 raise RunHardFailure(
3518 "VALIDATION_ERROR",
3519 f"call step '{call_id}' declares `routes`; `routes` is not supported "
3520 "on call steps",
3521 )
3522 if fn_name in self._stack:
3523 chain = " -> ".join(self._stack + [fn_name])
3524 raise RunHardFailure(
3525 "VALIDATION_ERROR",
3526 f"cycle detected in function expansion: {chain} "
3527 "(ADR-0037 D7 — recursion must be a bounded while loop or "
3528 "idempotent re-run)",
3529 )
3530 if fn_name in self._expanded_functions:
3531 raise RunHardFailure(
3532 "VALIDATION_ERROR",
3533 f"duplicate call expansion: function '{fn_name}' called at both "
3534 f"'{self._expanded_functions[fn_name]}' and '{call_id}' in one "
3535 "pipeline (ADR-0037 D7 — repeated calls to the same function are "
3536 "not yet supported)",
3537 )
3538 self._expanded_functions[fn_name] = call_id
3540 function = self._registry.get("Function", fn_name)
3541 if function is None:
3542 raise RunHardFailure(
3543 "VALIDATION_ERROR",
3544 f"call step '{call_id}' references unknown function '{fn_name}' "
3545 "(no kind: Function manifest in the workflows tree)",
3546 )
3547 body = (function.get("spec") or {}).get("steps") or []
3548 _validate_function_purity(fn_name, body, visible_ids)
3549 _reject_function_body_routes(fn_name, body)
3550 return function, body
3553_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
3556def _validate_channel_entry(index: int, channel: Any) -> None:
3557 """Validate one spec.channels entry against the frozen ChannelSpec shape.
3559 The engine build path does not run manifest-schema validation, so a
3560 malformed entry is rejected here — before any ``["name"]`` indexing — with
3561 an error that identifies the offending entry.
3562 """
3563 if not isinstance(channel, dict):
3564 raise RunHardFailure(
3565 "CHANNEL_BINDING_ERROR",
3566 f"Pipeline spec.channels[{index}] must be a mapping; got "
3567 f"{type(channel).__name__}",
3568 )
3569 name = channel.get("name")
3570 if not isinstance(name, str) or not name:
3571 raise RunHardFailure(
3572 "CHANNEL_BINDING_ERROR",
3573 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3574 "channel requires a non-empty string 'name'",
3575 )
3576 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3577 raise RunHardFailure(
3578 "CHANNEL_BINDING_ERROR",
3579 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3580 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}",
3581 )
3584def _validated_channels(raw_channels: Any) -> list[dict]:
3585 """Return spec.channels as a validated list, or [] when absent.
3587 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3588 """
3589 if raw_channels is None:
3590 return []
3591 if not isinstance(raw_channels, list):
3592 raise RunHardFailure(
3593 "CHANNEL_BINDING_ERROR",
3594 f"Pipeline spec.channels must be a list; got "
3595 f"{type(raw_channels).__name__}",
3596 )
3597 for index, channel in enumerate(raw_channels):
3598 _validate_channel_entry(index, channel)
3599 return raw_channels
3602def _validate_step_channel_bindings(
3603 steps: list[dict], channel_names: frozenset[str]
3604) -> None:
3605 """Reject malformed or undeclared step-level channel bindings at build time.
3607 Every step's declared ``output_channel`` must be a non-empty declared
3608 channel name, and every declared ``input_channels`` entry must be a
3609 non-empty declared channel name. This runs unconditionally, so a binding
3610 to a channel absent from ``spec.channels`` fails the build even when no
3611 channels are declared.
3612 """
3613 for step in _iter_steps(steps):
3614 _validate_step_output_channel(step, channel_names)
3615 _validate_step_input_channels(step, channel_names)
3618def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3619 channel = step.get("output_channel")
3620 if channel is None:
3621 return
3622 if not isinstance(channel, str) or not channel:
3623 raise RunHardFailure(
3624 "CHANNEL_BINDING_ERROR",
3625 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3626 f"string; got {channel!r}",
3627 )
3628 if channel not in channel_names:
3629 raise RunHardFailure(
3630 "CHANNEL_BINDING_ERROR",
3631 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3632 "which is not declared in spec.channels",
3633 )
3636def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3637 channels = step.get("input_channels")
3638 if channels is None:
3639 return
3640 if not isinstance(channels, list):
3641 raise RunHardFailure(
3642 "CHANNEL_BINDING_ERROR",
3643 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3644 f"channel names; got {type(channels).__name__}",
3645 )
3646 for channel in channels:
3647 if not isinstance(channel, str) or not channel:
3648 raise RunHardFailure(
3649 "CHANNEL_BINDING_ERROR",
3650 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3651 f"channel name '{channel}'; channel names must be non-empty "
3652 "strings",
3653 )
3654 if channel not in channel_names:
3655 raise RunHardFailure(
3656 "CHANNEL_BINDING_ERROR",
3657 f"Step '{step.get('id', '?')}' reads undeclared channel "
3658 f"'{channel}'; the channel must be declared in spec.channels",
3659 )
3662def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3663 """Reject duplicate and engine/step-reserved channel names."""
3664 forbidden = (
3665 {_CORE}
3666 | set(_RESERVED_NAMESPACES)
3667 | _CORE_STATE_KEYS
3668 | _collect_composed_step_ids(steps)
3669 )
3670 seen: set[str] = set()
3671 for channel in channels:
3672 name = channel.get("name")
3673 if name in seen:
3674 raise RunHardFailure(
3675 "CHANNEL_BINDING_ERROR",
3676 f"Pipeline declares channel '{name}' more than once; channel "
3677 "names must be unique",
3678 )
3679 seen.add(name)
3680 if name in forbidden:
3681 raise RunHardFailure(
3682 "CHANNEL_BINDING_ERROR",
3683 f"Pipeline channel '{name}' collides with an engine-reserved "
3684 "name (a step id, engine state key, reserved namespace, or the "
3685 "core state channel); channel names must be unique and "
3686 "non-reserved",
3687 )
3690def _input_path_root(value: str) -> str | None:
3691 """Leading state key of a bare dot-path input string, or None.
3693 Template strings, bracket-prefixed verbatim literals, and blank strings
3694 are not state reads and yield None.
3695 """
3696 if "{{" in value:
3697 return None
3698 stripped = value.lstrip()
3699 if not stripped or stripped.startswith("["):
3700 return None
3701 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3704def _iter_input_roots(input_spec: Any):
3705 """Yield every bare dot-path root in an input spec, recursing into containers."""
3706 if isinstance(input_spec, str):
3707 root = _input_path_root(input_spec)
3708 if root:
3709 yield root
3710 elif isinstance(input_spec, dict):
3711 for value in input_spec.values():
3712 yield from _iter_input_roots(value)
3713 elif isinstance(input_spec, list):
3714 for value in input_spec:
3715 yield from _iter_input_roots(value)
3718def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3719 """Reject channel-pipeline input roots that name nothing the state exposes.
3721 The channel state is the declared channels plus the engine state, so a bare
3722 dot-path input root must be a declared channel, a known step id, a reserved
3723 namespace, or an engine state key. Anything else is a hard build error —
3724 there is no silent fallback to a missing key.
3725 """
3726 allowed = (
3727 {ch["name"] for ch in channels}
3728 | _collect_step_ids(steps)
3729 | set(_RESERVED_NAMESPACES)
3730 | _CORE_STATE_KEYS
3731 )
3732 for step in _iter_steps(steps):
3733 for root in _iter_input_roots(step.get("input")):
3734 if root not in allowed:
3735 raise RunHardFailure(
3736 "CHANNEL_BINDING_ERROR",
3737 f"Pipeline channel step '{step.get('id', '?')}' input "
3738 f"references unknown root '{root}'; in a channel pipeline "
3739 "an input root must be a declared channel, a step id, a "
3740 "reserved namespace, or an engine state key",
3741 )
3744def _flatten_channel_state(state: dict) -> dict:
3745 """Read the flat committed snapshot nested under the Core namespace."""
3746 return dict(state.get(_CORE) or {})
3749def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3750 """Drop channel-named keys so a node emits only the core state."""
3751 return {k: v for k, v in flat.items() if k not in channel_names}
3754_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3756_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3759def _envelope_content(result: Any) -> Any:
3760 """Return an envelope's optional ``content`` field (ADR-0023 D3).
3762 ``content`` is opaque and optional: a non-mapping result, an absent field,
3763 and an explicit ``null`` all read as ``None``. Callers apply their own
3764 fallback and shape checks (e.g. ``[]`` for a spawn child plan).
3765 """
3766 if isinstance(result, dict):
3767 return result.get("content")
3768 return None
3771def _is_failure_envelope(value: Any) -> bool:
3772 """True when a step output is not a successfully parsed dict.
3774 Missing output, non-dict values, and the ``error``/``parse_error``
3775 envelopes written by the LLM failure paths are not routed. A valid
3776 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3777 it carries an explicit ``"error": null``.
3778 """
3779 if not isinstance(value, dict):
3780 return True
3781 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3782 return False
3783 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3786def _declared_output_channel_writes(
3787 step: dict, output: Any, channel_names: frozenset[str]
3788) -> dict[str, Any]:
3789 """Route one step output's content to its declared output channel.
3791 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3792 the working set. Failure envelopes and steps without an ``output_channel``
3793 are not routed; an undeclared target channel is a hard error.
3794 """
3795 if _is_failure_envelope(output):
3796 return {}
3797 channel = step.get("output_channel")
3798 if not channel:
3799 return {}
3800 if channel not in channel_names:
3801 raise RunHardFailure(
3802 "CHANNEL_BINDING_ERROR",
3803 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3804 "which is not declared in spec.channels",
3805 )
3806 return {channel: _envelope_content(output)}
3809def _route_declared_channels(
3810 step: dict,
3811 pre_state: dict,
3812 core: dict,
3813 step_runner: Any,
3814 channel_names: frozenset[str],
3815) -> dict[str, Any]:
3816 """Route a step's declared output channel write to its channel.
3818 Returns ``{channel_name: value}`` for LangGraph to merge through the
3819 channel's reducer, where ``channel_name`` is the step's single declared
3820 ``output_channel`` and ``value`` is the agent envelope's ``content``. Only
3821 an actual agent invocation is routed: a ``skip_llm_if`` passthrough and a
3822 parallel fan-out envelope are excluded. A step without an ``output_channel``
3823 routes nothing (its output stays in ``core`` untouched), and failure
3824 envelopes are not routed. A spawn step is routed at the spawn seam from its
3825 node result, never here from its envelope.
3826 """
3827 step_id = step.get("id", "?")
3828 if step.get("type") == "spawn":
3829 return {}
3830 if step_id not in core:
3831 return {}
3832 agent_spec = step_runner.resolve_agent(step)
3833 if agent_spec is None:
3834 return {}
3835 if not step_runner.ran_agent(step, pre_state):
3836 return {}
3837 return _declared_output_channel_writes(step, core[step_id], channel_names)
3840def _make_step_node(
3841 step: dict,
3842 step_runner: Any,
3843 channel_names: frozenset[str],
3844 channel_types: dict[str, str],
3845) -> Any:
3846 if not channel_names:
3847 def node_fn(state: PipelineState) -> PipelineState:
3848 return step_runner.run_step(step, state)
3849 return node_fn
3851 def node_fn(state: PipelineState) -> PipelineState:
3852 committed = _flatten_channel_state(state)
3853 working_set = ChannelWorkingSet(committed, channel_types)
3854 core = _strip_channel_keys(
3855 step_runner.run_step(step, committed, working_set), channel_names
3856 )
3857 writes = _route_declared_channels(
3858 step, committed, core, step_runner, channel_names
3859 )
3860 for name, value in writes.items():
3861 working_set.stage(name, value)
3862 written_names = {name for name, _ in working_set.pending_writes()}
3863 channels = working_set.commit()
3864 emitted = {name: channels[name] for name in written_names}
3865 return {_CORE: {**core, **emitted}}
3866 return node_fn
3869def _filter_composed_step_ids(result: dict, composed_step_ids: Any) -> dict:
3870 """Drop function-body step ids so a call subgraph's internals stay invisible.
3872 ADR-0037 D3: the caller sees only ``state[<call_id>].outputs`` plus
3873 unrelated keys. ``composed_step_ids`` is the build-time set recorded on the
3874 call placeholder. Channel-named keys are disjoint from it (a collision is a
3875 build-time ``CHANNEL_BINDING_ERROR``), so channel state is never touched.
3876 The input is left unmodified.
3877 """
3878 if not composed_step_ids:
3879 return dict(result)
3880 filtered = {
3881 key: value
3882 for key, value in result.items()
3883 if key not in composed_step_ids
3884 }
3885 core = filtered.get(_CORE)
3886 if isinstance(core, dict):
3887 filtered[_CORE] = {
3888 key: value
3889 for key, value in core.items()
3890 if key not in composed_step_ids
3891 }
3892 return filtered
3895def _make_subgraph_node(
3896 compiled: Any,
3897 condition: Any,
3898 step_runner: Any,
3899 channel_names: frozenset[str],
3900 composed_step_ids: Any = None,
3901) -> Any:
3902 """Wrap a compiled call subgraph with the call step's entry condition.
3904 A false condition returns no state updates, so the subgraph never runs and
3905 ``state[<call_id>]`` is not written — matching the soft-skip semantics of a
3906 normal step condition. The ctx is derived exactly like ``run_step``'s,
3907 including channel-mode flattening. Every successful invocation is filtered
3908 through ``_filter_composed_step_ids`` so the function body stays invisible.
3909 """
3910 if condition is None:
3911 def node_fn(state: PipelineState) -> PipelineState:
3912 return _filter_composed_step_ids(
3913 compiled.invoke(state), composed_step_ids
3914 )
3915 return node_fn
3917 if not channel_names:
3918 def node_fn(state: PipelineState) -> PipelineState:
3919 if not _eval_condition(condition, step_runner._make_ctx(state)):
3920 return {}
3921 return _filter_composed_step_ids(
3922 compiled.invoke(state), composed_step_ids
3923 )
3924 return node_fn
3926 def node_fn(state: PipelineState) -> PipelineState:
3927 committed = _flatten_channel_state(state)
3928 if not _eval_condition(condition, step_runner._make_ctx(committed)):
3929 return {}
3930 return _filter_composed_step_ids(
3931 compiled.invoke(state), composed_step_ids
3932 )
3933 return node_fn
3936# ---------------------------------------------------------------------------
3937# Native `while` cycle nodes (ADR-0025 D1/D2)
3938# ---------------------------------------------------------------------------
3939#
3940# A `while` step is a compiled cycle: `head -> body... -> cond` with a
3941# conditional back-edge from `cond` to `head` (continue) or `exit`. The head
3942# owns entry-condition handling and the initial `current`; the cond node owns
3943# the feed, the iteration increment and the exit-condition evaluation; the exit
3944# node writes the documented result and enforces `fail_if_exhausted`. The
3945# nodes read/write the shared graph state, using the `_CORE` container when the
3946# pipeline declares channels.
3949def _while_marker_key(step_id: str) -> str:
3950 """Per-while engine marker key (nested whiles live in isolated cycles)."""
3951 return f"{_WHILE_STATE_KEY}__{step_id}"
3954def _while_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
3955 return _flatten_channel_state(state) if channel_names else state
3958def _while_update(update: dict, channel_names: frozenset[str]) -> dict:
3959 return {_CORE: update} if channel_names else update
3962def _while_loop_context(step_id: str, marker: dict) -> dict:
3963 """The body-visible `loop` namespace and the intermediate step result.
3965 Written in this order so a step literally named ``loop`` shadows the
3966 injected namespace — the legacy in-node loop had the same collision.
3967 """
3968 current = marker["current"]
3969 iteration = marker["iteration"]
3970 update = {"loop": {"current": current, "iteration": iteration}}
3971 update[step_id] = {"current": current, "passed": False}
3972 return update
3975def _while_invoke_config(step: dict) -> dict | None:
3976 """Recursion bound for a while/foreach cycle placeholder, else None."""
3977 if step.get(_SUBGRAPH_WHILE_KEY):
3978 return {"recursion_limit": step.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
3979 if step.get(_SUBGRAPH_FOREACH_KEY):
3980 return {"recursion_limit": step.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
3981 return None
3984def _make_while_body_node(
3985 sub: dict,
3986 step_runner: Any,
3987 channel_names: frozenset[str],
3988 channel_types: dict[str, str],
3989) -> Any:
3990 """A while body step as a node: a nested subgraph or a normal step node."""
3991 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
3992 if sub.get(_SUBGRAPH_WHILE_KEY):
3993 return _make_while_subgraph_node(sub, step_runner, channel_names)
3994 if sub.get(_SUBGRAPH_FOREACH_KEY):
3995 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
3996 return _make_subgraph_node(
3997 sub[_SUBGRAPH_GRAPH_KEY],
3998 sub.get("condition"),
3999 step_runner,
4000 channel_names,
4001 sub.get(_SUBGRAPH_STEP_IDS_KEY),
4002 )
4003 return _make_step_node(sub, step_runner, channel_names, channel_types)
4006def _make_while_head_node(
4007 step: dict,
4008 step_runner: Any,
4009 channel_names: frozenset[str],
4010 max_iter: int,
4011) -> Any:
4012 """Entry node: handle the entry condition and seed the loop marker."""
4013 step_id = step.get("id", "?")
4014 marker_key = _while_marker_key(step_id)
4016 def node_fn(state: PipelineState) -> PipelineState:
4017 flat = _while_flat(state, channel_names)
4018 marker = flat.get(marker_key)
4019 if marker is not None:
4020 return _while_update(
4021 _while_loop_context(step_id, marker), channel_names
4022 )
4023 ctx = step_runner._make_ctx(flat)
4024 entry_cond = step.get("condition")
4025 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4026 skip_input = _resolved_while_input(step, flat, ctx)
4027 skip_marker = {
4028 "current": None,
4029 "iteration": 0,
4030 "passed": True,
4031 "skipped": True,
4032 "max_iter": max_iter,
4033 }
4034 return _while_update(
4035 {
4036 step_id: {
4037 "passed": True,
4038 "iterations": 0,
4039 "output": skip_input,
4040 "exhausted": False,
4041 "skipped": True,
4042 },
4043 marker_key: skip_marker,
4044 },
4045 channel_names,
4046 )
4047 marker = {
4048 "current": _resolved_while_input(step, flat, ctx),
4049 "iteration": 0,
4050 "passed": False,
4051 "skipped": False,
4052 "max_iter": max_iter,
4053 }
4054 update = _while_loop_context(step_id, marker)
4055 update[marker_key] = marker
4056 return _while_update(update, channel_names)
4058 return node_fn
4061def _resolved_while_input(step: dict, flat: dict, ctx: dict) -> Any:
4062 input_spec = step.get("input")
4063 if input_spec is None:
4064 return None
4065 return _resolve_input(input_spec, flat, ctx)
4068def _make_while_head_router(
4069 step: dict,
4070 channel_names: frozenset[str],
4071 first_id: str,
4072 exit_id: str,
4073 end: Any,
4074) -> Any:
4075 """Route from the head: skipped → end, bound already reached → exit, else body."""
4076 marker_key = _while_marker_key(step.get("id", "?"))
4078 def router(state: PipelineState) -> str:
4079 marker = _while_flat(state, channel_names).get(marker_key) or {}
4080 if marker.get("skipped"):
4081 return end
4082 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
4083 return exit_id
4084 return first_id
4086 return router
4089def _make_while_cond_node(
4090 step: dict,
4091 step_runner: Any,
4092 channel_names: frozenset[str],
4093) -> Any:
4094 """Feed the declared output, increment the iteration, evaluate the exit."""
4095 step_id = step.get("id", "?")
4096 marker_key = _while_marker_key(step_id)
4097 declared_output = step.get("output")
4098 exit_condition = step.get("exit_condition")
4100 def node_fn(state: PipelineState) -> PipelineState:
4101 flat = _while_flat(state, channel_names)
4102 marker = dict(flat.get(marker_key) or {})
4103 current = marker.get("current")
4104 if declared_output:
4105 feed_value = flat.get(declared_output)
4106 if isinstance(feed_value, dict) and "output" in feed_value:
4107 current = feed_value["output"]
4108 elif isinstance(feed_value, dict) and isinstance(
4109 feed_value.get("results"), list
4110 ):
4111 current = feed_value["results"]
4112 else:
4113 current = feed_value
4114 passed = False
4115 if exit_condition:
4116 if _eval_condition(
4117 exit_condition, step_runner._make_ctx(flat)
4118 ):
4119 passed = True
4120 else:
4121 passed = True
4122 marker.update(
4123 {
4124 "current": current,
4125 "iteration": marker.get("iteration", 0) + 1,
4126 "passed": passed,
4127 }
4128 )
4129 return _while_update({marker_key: marker}, channel_names)
4131 return node_fn
4134def _make_while_cond_router(
4135 step: dict,
4136 channel_names: frozenset[str],
4137 head_id: str,
4138 exit_id: str,
4139) -> Any:
4140 """Route from the cond: exit on pass or when the bound is reached, else head."""
4141 marker_key = _while_marker_key(step.get("id", "?"))
4143 def router(state: PipelineState) -> str:
4144 marker = _while_flat(state, channel_names).get(marker_key) or {}
4145 if marker.get("passed"):
4146 return exit_id
4147 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
4148 return exit_id
4149 return head_id
4151 return router
4154def _make_while_exit_node(
4155 step: dict,
4156 step_runner: Any,
4157 channel_names: frozenset[str],
4158) -> Any:
4159 """Write the documented while result and enforce `fail_if_exhausted`."""
4160 step_id = step.get("id", "?")
4161 marker_key = _while_marker_key(step_id)
4162 fail_if_exhausted = step.get("fail_if_exhausted")
4164 def node_fn(state: PipelineState) -> PipelineState:
4165 flat = _while_flat(state, channel_names)
4166 marker = flat.get(marker_key) or {}
4167 passed = bool(marker.get("passed", False))
4168 exhausted = not passed
4169 if (
4170 fail_if_exhausted is not None
4171 and exhausted
4172 and _eval_condition(
4173 fail_if_exhausted, step_runner._make_ctx(flat)
4174 )
4175 ):
4176 raise RunHardFailure(
4177 "EXHAUSTED",
4178 f"while step '{step_id}' exhausted "
4179 f"max_iterations={marker.get('max_iter')} without "
4180 "exit_condition passing (fail_if_exhausted)",
4181 {"step_id": step_id, "exhausted": True},
4182 )
4183 return _while_update(
4184 {
4185 step_id: {
4186 "passed": passed,
4187 "iterations": marker.get("iteration", 0),
4188 "output": marker.get("current"),
4189 "exhausted": exhausted,
4190 "skipped": False,
4191 }
4192 },
4193 channel_names,
4194 )
4196 return node_fn
4199def _make_while_subgraph_node(
4200 placeholder: dict,
4201 step_runner: Any,
4202 channel_names: frozenset[str],
4203) -> Any:
4204 """Parent node for a compiled `while` cycle: isolate and surface its result.
4206 The cycle shares the graph state schema, so invoking it directly would leak
4207 the body step results and loop plumbing into the parent. This wrapper runs
4208 the cycle on a copy and returns only the documented result plus the side
4209 channels, preserving the in-node loop's state contract exactly.
4210 """
4211 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4212 step_id = placeholder["id"]
4213 config = {"recursion_limit": placeholder.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
4215 if not channel_names:
4216 def node_fn(state: PipelineState) -> PipelineState:
4217 sub = compiled.invoke(state, config)
4218 result = {step_id: sub.get(step_id)}
4219 for key in _WHILE_SIDE_CHANNEL_KEYS:
4220 if key in sub and sub.get(key) != state.get(key):
4221 result[key] = sub.get(key)
4222 return result
4223 return node_fn
4225 def node_fn(state: PipelineState) -> PipelineState:
4226 sub = compiled.invoke(state, config)
4227 flat = _flatten_channel_state(sub)
4228 parent_flat = _flatten_channel_state(state)
4229 result = {step_id: flat.get(step_id)}
4230 for key in _WHILE_SIDE_CHANNEL_KEYS:
4231 if key in flat and flat.get(key) != parent_flat.get(key):
4232 result[key] = flat.get(key)
4233 return {_CORE: _strip_channel_keys(result, channel_names)}
4235 return node_fn
4238# ---------------------------------------------------------------------------
4239# Native sequential `foreach` cycle nodes
4240# ---------------------------------------------------------------------------
4241#
4242# A sequential `foreach` is a compiled cycle: `head -> (body... | agent) ->
4243# cond` with a conditional back-edge from `cond` to `head` (next item) or
4244# `exit` (`max_output` reached or the item list exhausted). The head resolves
4245# the item list once and injects the current item (`item` / `_item` /
4246# `_foreach_item`); the cond collects the item result; the exit node writes the
4247# documented aggregate (`{"results", "passed", "failing", "failed"}` for the
4248# body path, the plain list for the agent path). Nodes read/write the shared
4249# graph state, using the `_CORE` container when the pipeline declares channels.
4252def _foreach_marker_key(step_id: str) -> str:
4253 """Per-foreach engine marker key (nested foreachs live in isolated cycles)."""
4254 return f"{_FOREACH_STATE_KEY}__{step_id}"
4257def _foreach_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
4258 return _flatten_channel_state(state) if channel_names else state
4261def _foreach_update(update: dict, channel_names: frozenset[str]) -> dict:
4262 return {_CORE: update} if channel_names else update
4265def _make_foreach_head_node(
4266 step: dict,
4267 step_runner: Any,
4268 channel_names: frozenset[str],
4269) -> Any:
4270 """Entry node: evaluate the entry condition, resolve items, inject the item."""
4271 step_id = step.get("id", "?")
4272 marker_key = _foreach_marker_key(step_id)
4274 def node_fn(state: PipelineState) -> PipelineState:
4275 flat = _foreach_flat(state, channel_names)
4276 marker = flat.get(marker_key)
4277 if marker is None:
4278 ctx = step_runner._make_ctx(flat)
4279 entry_cond = step.get("condition")
4280 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4281 return _foreach_update({}, channel_names)
4282 max_output_raw = step.get("max_output")
4283 marker = {
4284 "items": step_runner._foreach_item_list(step, flat, ctx),
4285 "index": 0,
4286 "results": [],
4287 "produced_outputs": 0,
4288 "max_output": (
4289 _resolve_int(max_output_raw, ctx, 0)
4290 if max_output_raw is not None
4291 else 0
4292 ),
4293 }
4294 items = marker.get("items") or []
4295 index = marker.get("index", 0)
4296 if index < len(items):
4297 update = {
4298 "_item": items[index],
4299 "item": items[index],
4300 "_foreach_item": items[index],
4301 }
4302 update[marker_key] = marker
4303 return _foreach_update(update, channel_names)
4304 return _foreach_update({marker_key: marker}, channel_names)
4306 return node_fn
4309def _make_foreach_head_router(
4310 step: dict,
4311 channel_names: frozenset[str],
4312 first_id: str,
4313 exit_id: str,
4314 end: Any,
4315) -> Any:
4316 """Route from the head: condition-skipped → end, exhausted/budget → exit, else body."""
4317 marker_key = _foreach_marker_key(step.get("id", "?"))
4319 def router(state: PipelineState) -> str:
4320 marker = _foreach_flat(state, channel_names).get(marker_key)
4321 if marker is None:
4322 return end
4323 items = marker.get("items") or []
4324 if marker.get("index", 0) >= len(items):
4325 return exit_id
4326 max_output = marker.get("max_output", 0)
4327 if max_output and marker.get("produced_outputs", 0) >= max_output:
4328 return exit_id
4329 return first_id
4331 return router
4334def _make_foreach_body_node(
4335 sub: dict,
4336 step_runner: Any,
4337 channel_names: frozenset[str],
4338 channel_types: dict[str, str],
4339) -> Any:
4340 """A foreach body step as a node: a nested loop subgraph or a step node."""
4341 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
4342 if sub.get(_SUBGRAPH_WHILE_KEY):
4343 return _make_while_subgraph_node(sub, step_runner, channel_names)
4344 if sub.get(_SUBGRAPH_FOREACH_KEY):
4345 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
4346 return _make_subgraph_node(
4347 sub[_SUBGRAPH_GRAPH_KEY],
4348 sub.get("condition"),
4349 step_runner,
4350 channel_names,
4351 sub.get(_SUBGRAPH_STEP_IDS_KEY),
4352 )
4353 return _make_step_node(sub, step_runner, channel_names, channel_types)
4356def _make_foreach_agent_node(
4357 step: dict,
4358 step_runner: Any,
4359 channel_names: frozenset[str],
4360 channel_types: dict[str, str],
4361) -> Any:
4362 """Agent-path node: invoke the foreach step's agent for the current item."""
4363 step_id = step.get("id", "?")
4365 def node_fn(state: PipelineState) -> PipelineState:
4366 flat = _foreach_flat(state, channel_names)
4367 resolved = step_runner.resolve_agent(step)
4368 ctx = step_runner._make_ctx(flat)
4369 resolved_step = step_runner._resolve_step_runtime(step, flat, ctx)
4370 resolved_step = step_runner._attach_communication_context(resolved_step, flat)
4371 resolved_step = step_runner._attach_resolved_tools(resolved_step, resolved)
4372 result_state = step_runner._invoke_agent(resolved, resolved_step, flat)
4373 if not channel_names:
4374 return result_state
4375 return {_CORE: _strip_channel_keys(result_state, channel_names)}
4377 return node_fn
4380def _make_foreach_cond_node(
4381 step: dict,
4382 step_runner: Any,
4383 channel_names: frozenset[str],
4384) -> Any:
4385 """Collect the current item's result and advance the item index."""
4386 step_id = step.get("id", "?")
4387 marker_key = _foreach_marker_key(step_id)
4388 declared_output = step.get("output")
4389 runs_body = bool(step.get("body"))
4391 def node_fn(state: PipelineState) -> PipelineState:
4392 flat = _foreach_flat(state, channel_names)
4393 marker = dict(flat.get(marker_key) or {})
4394 results = list(marker.get("results") or [])
4395 produced = marker.get("produced_outputs", 0)
4396 if runs_body:
4397 results.append(StepRunner._body_item_result(flat, declared_output))
4398 publish_out = flat.get("publish")
4399 if isinstance(publish_out, dict) and publish_out.get("path"):
4400 produced += 1
4401 else:
4402 results.append(flat.get(step_id))
4403 marker.update(
4404 {
4405 "results": results,
4406 "index": marker.get("index", 0) + 1,
4407 "produced_outputs": produced,
4408 }
4409 )
4410 return _foreach_update({marker_key: marker}, channel_names)
4412 return node_fn
4415def _make_foreach_cond_router(
4416 step: dict,
4417 channel_names: frozenset[str],
4418 head_id: str,
4419 exit_id: str,
4420) -> Any:
4421 """Route from the cond: exit on exhaustion or budget, else back to the head."""
4422 marker_key = _foreach_marker_key(step.get("id", "?"))
4424 def router(state: PipelineState) -> str:
4425 marker = _foreach_flat(state, channel_names).get(marker_key) or {}
4426 items = marker.get("items") or []
4427 if marker.get("index", 0) >= len(items):
4428 return exit_id
4429 max_output = marker.get("max_output", 0)
4430 if max_output and marker.get("produced_outputs", 0) >= max_output:
4431 return exit_id
4432 return head_id
4434 return router
4437def _make_foreach_exit_node(
4438 step: dict,
4439 step_runner: Any,
4440 channel_names: frozenset[str],
4441) -> Any:
4442 """Write the documented foreach aggregate result."""
4443 step_id = step.get("id", "?")
4444 marker_key = _foreach_marker_key(step_id)
4445 runs_body = bool(step.get("body"))
4447 def node_fn(state: PipelineState) -> PipelineState:
4448 flat = _foreach_flat(state, channel_names)
4449 marker = flat.get(marker_key) or {}
4450 results = list(marker.get("results") or [])
4451 step_result = (
4452 StepRunner._aggregate_body_results(results) if runs_body else results
4453 )
4454 return _foreach_update({step_id: step_result}, channel_names)
4456 return node_fn
4459def _make_foreach_subgraph_node(
4460 placeholder: dict,
4461 step_runner: Any,
4462 channel_names: frozenset[str],
4463) -> Any:
4464 """Parent node for a compiled sequential `foreach` cycle: isolate and surface.
4466 The cycle shares the graph state schema, so invoking it directly would leak
4467 body step results and loop plumbing into the parent. This wrapper runs the
4468 cycle on a copy and returns only the documented aggregate plus the side
4469 channels. A condition-skipped entry writes no step result (matching the
4470 in-node soft-skip), so the step id is only surfaced when the cycle wrote it.
4471 """
4472 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4473 step_id = placeholder["id"]
4474 config = {"recursion_limit": placeholder.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
4476 if not channel_names:
4477 def node_fn(state: PipelineState) -> PipelineState:
4478 sub = compiled.invoke(state, config)
4479 if step_id not in sub:
4480 return {}
4481 result = {step_id: sub.get(step_id)}
4482 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4483 if key in sub and sub.get(key) != state.get(key):
4484 result[key] = sub.get(key)
4485 return result
4486 return node_fn
4488 def node_fn(state: PipelineState) -> PipelineState:
4489 sub = compiled.invoke(state, config)
4490 flat = _flatten_channel_state(sub)
4491 parent_flat = _flatten_channel_state(state)
4492 if step_id not in flat:
4493 return {}
4494 result = {step_id: flat.get(step_id)}
4495 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4496 if key in flat and flat.get(key) != parent_flat.get(key):
4497 result[key] = flat.get(key)
4498 return {_CORE: _strip_channel_keys(result, channel_names)}
4500 return node_fn
4503def _collect_parallel_channel_writes(
4504 steps: list[dict],
4505 pre_state: dict,
4506 core: dict,
4507 step_runner: Any,
4508 channel_names: frozenset[str],
4509 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
4510) -> list[tuple[str, Any]]:
4511 """Collect one parallel node's declared writes in branch order.
4513 Every branch write is emitted, so fan-in feeds the channel-type reducer
4514 every write in deterministic declaration/branch order: Topic appends each
4515 write, LastValue keeps the last, and BinaryOperatorAggregate combines them
4516 per its accumulated shape (list concat, collection-map union, dict merge,
4517 scalar last). A branch's isolated body writes are emitted before its own
4518 declared write, mirroring the top-level node's route-after-run order.
4519 """
4520 writes: list[tuple[str, Any]] = []
4521 for step in steps:
4522 if branch_working_sets is not None:
4523 branch_working_set = branch_working_sets.get(step["id"])
4524 if branch_working_set is not None:
4525 writes.extend(branch_working_set.pending_writes())
4526 branch = _route_declared_channels(
4527 step, pre_state, core, step_runner, channel_names
4528 )
4529 for channel_name, value in branch.items():
4530 writes.append((channel_name, value))
4531 return writes
4534def _make_parallel_node(
4535 group_steps: list[dict],
4536 step_runner: Any,
4537 channel_names: frozenset[str],
4538 channel_types: dict[str, str],
4539) -> Any:
4540 def node_fn(state: PipelineState) -> PipelineState:
4541 base = (
4542 _flatten_channel_state(state)
4543 if channel_names else state
4544 )
4545 working_set = (
4546 ChannelWorkingSet(base, channel_types) if channel_names else None
4547 )
4548 # Each branch writes into its own fork; the shared set is advanced only
4549 # at fan-in, so branch-internal body writes never race.
4550 branch_working_sets = (
4551 {s["id"]: working_set.fork() for s in group_steps}
4552 if working_set is not None else None
4553 )
4555 def submit_branch(pool: Any, step: dict) -> Any:
4556 if branch_working_sets is None:
4557 return pool.submit(step_runner.run_step, step, base)
4558 return pool.submit(
4559 step_runner.run_step, step, base, branch_working_sets[step["id"]]
4560 )
4562 results: dict[str, Any] = {}
4563 branch_metrics: dict = {}
4564 branch_tool_meta: dict = {}
4565 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
4566 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
4567 for future in as_completed(futures):
4568 sid = futures[future]
4569 try:
4570 out = future.result()
4571 results[sid] = out.get(sid)
4572 branch_metrics.update(out.get("step_context_metrics") or {})
4573 branch_tool_meta.update(out.get("step_tool_meta") or {})
4574 except SkillLoadError:
4575 raise
4576 except RunHardFailure:
4577 pool.shutdown(wait=False, cancel_futures=True)
4578 raise
4579 except Exception as exc: # noqa: BLE001
4580 log.error("Step '%s' failed: %s", sid, exc)
4581 results[sid] = {"error": str(exc)}
4582 if channel_names:
4583 merged = _strip_channel_keys({**base, **results}, channel_names)
4584 if branch_metrics:
4585 merged["step_context_metrics"] = _merge_side_channel(
4586 merged.get("step_context_metrics"), branch_metrics
4587 )
4588 if branch_tool_meta:
4589 merged["step_tool_meta"] = _merge_side_channel(
4590 merged.get("step_tool_meta"), branch_tool_meta
4591 )
4592 writes = _collect_parallel_channel_writes(
4593 group_steps, base, merged, step_runner, channel_names,
4594 branch_working_sets,
4595 )
4596 written_names = {name for name, _ in writes}
4597 for name, value in writes:
4598 working_set.stage(name, value)
4599 channels = working_set.commit()
4600 emitted = {name: channels[name] for name in written_names}
4601 return {_CORE: {**merged, **emitted}}
4602 final: PipelineState = {**state, **results}
4603 if branch_metrics:
4604 final = {
4605 **final,
4606 "step_context_metrics": _merge_side_channel(
4607 state.get("step_context_metrics"), branch_metrics
4608 ),
4609 }
4610 if branch_tool_meta:
4611 final = {
4612 **final,
4613 "step_tool_meta": _merge_side_channel(
4614 state.get("step_tool_meta"), branch_tool_meta
4615 ),
4616 }
4617 return final
4618 return node_fn
4621@dataclass(frozen=True)
4622class _MapReduceParts:
4623 """The nodes and edges of one concurrent step's Send fan-out.
4625 ``dispatch_id`` reuses the step id so the step's incoming dependency edges are
4626 unchanged; ``join_id`` carries the step's outgoing edges.
4627 """
4629 dispatch_id: str
4630 worker_id: str
4631 fallback_id: str
4632 join_id: str
4633 dispatch_fn: Any
4634 route_fn: Any
4635 worker_fn: Any
4636 join_fn: Any
4639@dataclass(frozen=True)
4640class _MapReduceItemOutcome:
4641 """One worker's per-item result plus its errors, side channels and writes.
4643 ``errors`` carries the item's soft errors (and the inherited soft errors it
4644 read), so the join can fold them in item order exactly like the in-node
4645 concurrent path. ``channel_writes`` is empty for the parallel strategy (its
4646 channel writes are derived from the item result at the join) and carries the
4647 foreach body's pending writes so the join can stage them in item order.
4648 """
4650 result: Any
4651 metrics: dict | None
4652 tool_meta: dict | None
4653 channel_writes: tuple[tuple[str, Any], ...]
4654 errors: tuple[str, ...] = ()
4657@dataclass(frozen=True)
4658class _MapReducePlan:
4659 """The per-step behaviour injected into the shared Send machinery.
4661 The parallel strategy runs a single agent call per item; the foreach
4662 strategy runs the body (or agent) per item. ``item_lister`` resolves the
4663 fan-out list, ``item_runner`` executes one item, ``result_assembler``
4664 shapes the step result, and ``channel_write_stager`` stages per-item
4665 channel writes at the join.
4666 """
4668 item_lister: Any
4669 item_runner: Any
4670 result_assembler: Any
4671 channel_write_stager: Any
4674def _map_reduce_worker_payload(flat: dict, item: Any, index: int) -> dict:
4675 """Build a worker's ``Send`` payload from the resolved source state."""
4676 return {**flat, "item": item, "_item": item, _MAP_REDUCE_ITEM_INDEX: index}
4679def _map_reduce_worker_output(
4680 step_id: str,
4681 index: int,
4682 outcome: _MapReduceItemOutcome,
4683) -> PipelineState:
4684 """The accumulator record carrying every per-item fold input.
4686 The slot is ``{index: {"result", "writes", "errors", "metrics",
4687 "tool_meta"}}`` so the join can assemble the step result, stage channel
4688 writes, and fold soft errors plus side channels in item order — the same
4689 observable result as the in-node concurrent path.
4690 """
4691 record = {
4692 "result": outcome.result,
4693 "writes": list(outcome.channel_writes),
4694 "errors": list(outcome.errors),
4695 "metrics": outcome.metrics if isinstance(outcome.metrics, dict) else None,
4696 "tool_meta": (
4697 outcome.tool_meta if isinstance(outcome.tool_meta, dict) else None
4698 ),
4699 }
4700 return {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: {index: record}}}
4703def _make_map_reduce_route_fn(
4704 plan: _MapReducePlan,
4705 step_runner: Any,
4706 channel_names: frozenset[str],
4707 worker_id: str,
4708 fallback_id: str,
4709 join_id: str,
4710) -> Any:
4711 def route(state: PipelineState) -> list:
4712 flat = _flatten_channel_state(state) if channel_names else state
4713 ctx = step_runner._make_ctx(flat)
4714 items = plan.item_lister(flat, ctx)
4715 if items is None:
4716 return [Send(fallback_id, state)]
4717 if not items:
4718 return [Send(join_id, state)]
4719 return [
4720 Send(worker_id, _map_reduce_worker_payload(flat, item, index))
4721 for index, item in enumerate(items)
4722 ]
4724 return route
4727def _make_map_reduce_worker_fn(
4728 step: dict,
4729 plan: _MapReducePlan,
4730 step_runner: Any,
4731 gate: threading.Semaphore,
4732) -> Any:
4733 step_id = step.get("id", "?")
4735 def worker(state: PipelineState) -> PipelineState:
4736 index = state[_MAP_REDUCE_ITEM_INDEX]
4737 item_state = {
4738 key: value
4739 for key, value in state.items()
4740 if key != _MAP_REDUCE_ITEM_INDEX
4741 }
4742 item = item_state.get("item")
4743 item_ctx = step_runner._make_ctx(item_state)
4744 with gate:
4745 try:
4746 outcome = plan.item_runner(
4747 state, index, item, item_state, item_ctx
4748 )
4749 except SkillLoadError:
4750 raise
4751 except RunHardFailure:
4752 raise
4753 except Exception as exc: # noqa: BLE001
4754 log.error("[step] %s item failed: %s", step_id, exc)
4755 outcome = _MapReduceItemOutcome(
4756 {"error": str(exc)}, None, None, ()
4757 )
4758 return _map_reduce_worker_output(step_id, index, outcome)
4760 return worker
4763def _fold_map_reduce_errors(records: list, base_errors: list) -> list:
4764 """Fold every item's soft errors into *base_errors* in item order."""
4765 merged = list(base_errors)
4766 for record in records:
4767 if not isinstance(record, dict):
4768 continue
4769 merged = StepRunner._merge_errors(merged, record.get("errors") or [])
4770 return merged
4773def _fold_map_reduce_side_channel(records: list, record_key: str) -> dict:
4774 """Fold every item's side-channel map in item order (later item wins)."""
4775 folded: dict = {}
4776 for record in records:
4777 if not isinstance(record, dict):
4778 continue
4779 value = record.get(record_key)
4780 if isinstance(value, dict):
4781 folded.update(value)
4782 return folded
4785def _fold_map_reduce_extras(flat: dict, records: list) -> dict:
4786 """The join's per-item state extras: soft errors plus folded side channels.
4788 Mirrors the in-node concurrent path: errors are merged in item order with
4789 dedup, and the budget/side-channel counters follow the same item order.
4790 """
4791 extras: dict = {}
4792 base_errors = list(flat.get("errors") or [])
4793 errors = _fold_map_reduce_errors(records, base_errors)
4794 if errors != base_errors:
4795 extras["errors"] = errors
4796 for state_key, record_key in (
4797 ("step_context_metrics", "metrics"),
4798 ("step_tool_meta", "tool_meta"),
4799 ):
4800 folded = _fold_map_reduce_side_channel(records, record_key)
4801 if folded:
4802 extras[state_key] = _merge_side_channel(flat.get(state_key), folded)
4803 return extras
4806def _make_map_reduce_join_fn(
4807 step: dict,
4808 plan: _MapReducePlan,
4809 step_runner: Any,
4810 channel_names: frozenset[str],
4811 channel_types: dict[str, str],
4812) -> Any:
4813 step_id = step.get("id", "?")
4815 def join(state: PipelineState) -> PipelineState:
4816 flat = _flatten_channel_state(state) if channel_names else state
4817 accumulated = state.get(_MAP_REDUCE_ACCUMULATOR_KEY) or {}
4818 entries = accumulated.get(step_id) or {}
4819 clear = {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: None}}
4820 ctx = step_runner._make_ctx(flat)
4821 items = plan.item_lister(flat, ctx)
4822 if items is None:
4823 return clear
4824 records = [entries.get(index) for index in range(len(items))]
4825 results = [
4826 record.get("result") if isinstance(record, dict) else None
4827 for record in records
4828 ]
4829 step_result = plan.result_assembler(results)
4830 extras = _fold_map_reduce_extras(flat, records)
4831 if not channel_names:
4832 return {step_id: step_result, **extras, **clear}
4833 working_set = ChannelWorkingSet(flat, channel_types)
4834 plan.channel_write_stager(results, records, working_set)
4835 written_names = {name for name, _ in working_set.pending_writes()}
4836 channels = working_set.commit()
4837 emitted = {name: channels[name] for name in written_names}
4838 core = _strip_channel_keys(
4839 {**flat, step_id: step_result, **extras}, channel_names
4840 )
4841 return {_CORE: {**core, **emitted}, **clear}
4843 return join
4846def _make_parallel_map_reduce_plan(
4847 step: dict, step_runner: Any
4848) -> _MapReducePlan:
4849 """Plan for a concurrent `parallel` step: one agent call per item."""
4850 resolved_agent = step_runner.resolve_agent(step)
4852 def run_item(state, index, item, item_state, item_ctx):
4853 result, metrics, tool_meta, errors = step_runner._invoke_parallel_item(
4854 step, resolved_agent, item, item_state, item_ctx
4855 )
4856 return _MapReduceItemOutcome(
4857 result, metrics, tool_meta, (), tuple(errors)
4858 )
4860 return _MapReducePlan(
4861 item_lister=lambda flat, ctx: step_runner._parallel_fanout_item_list(
4862 step, flat, ctx
4863 ),
4864 item_runner=run_item,
4865 result_assembler=lambda results: {"results": results},
4866 channel_write_stager=lambda results, records, working_set: (
4867 step_runner._stage_fanout_channel_writes(step, results, working_set)
4868 ),
4869 )
4872def _worker_committed_flat(
4873 state: PipelineState, channel_names: frozenset[str]
4874) -> dict:
4875 """The committed channel snapshot visible to one worker.
4877 A channels-path worker receives the flattened snapshot as its payload, so
4878 its channel keys sit at the top level; the typed ``__core__`` shape (only
4879 seen by the join) is unwrapped.
4880 """
4881 if _CORE in state:
4882 return _flatten_channel_state(state)
4883 return {name: state[name] for name in channel_names if name in state}
4886def _pending_channel_writes(
4887 working_set: ChannelWorkingSet | None,
4888) -> tuple[tuple[str, Any], ...]:
4889 if working_set is None:
4890 return ()
4891 return tuple(working_set.pending_writes())
4894def _stage_map_reduce_item_channel_writes(
4895 results: list, records: list, working_set: ChannelWorkingSet
4896) -> None:
4897 """Stage each item's carried channel writes in item order."""
4898 for record in records:
4899 if not isinstance(record, dict):
4900 continue
4901 for name, value in record.get("writes") or ():
4902 working_set.stage(name, value)
4905def _make_foreach_map_reduce_plan(
4906 step: dict,
4907 step_runner: Any,
4908 channel_names: frozenset[str],
4909 channel_types: dict[str, str],
4910) -> _MapReducePlan:
4911 """Plan for a concurrent `foreach` step: the body (or agent) runs per item."""
4912 step_id = step.get("id", "?")
4913 body = step.get("body") or []
4914 declared_output = step.get("output")
4915 runs_body = bool(body)
4917 def result_assembler(results):
4918 if runs_body:
4919 return StepRunner._aggregate_body_results(results)
4920 return results
4922 def run_item(state, index, item, item_state, _item_ctx):
4923 item_working_set = (
4924 ChannelWorkingSet(
4925 _worker_committed_flat(state, channel_names), channel_types
4926 )
4927 if channel_names
4928 else None
4929 )
4930 iter_state = {
4931 **item_state, "item": item, "_item": item, "_foreach_item": item
4932 }
4933 if runs_body:
4934 for sub_step in body:
4935 iter_state = step_runner._run_body_step(
4936 sub_step, iter_state, item_working_set
4937 )
4938 return _MapReduceItemOutcome(
4939 step_runner._body_item_result(iter_state, declared_output),
4940 iter_state.get("step_context_metrics"),
4941 iter_state.get("step_tool_meta"),
4942 _pending_channel_writes(item_working_set),
4943 tuple(iter_state.get("errors") or []),
4944 )
4945 resolved = step_runner.resolve_agent(step)
4946 item_read_state = step_runner._read_state(iter_state, item_working_set)
4947 item_ctx = step_runner._make_ctx(item_read_state)
4948 resolved_step = step_runner._resolve_step_runtime(
4949 step, item_read_state, item_ctx
4950 )
4951 resolved_step = step_runner._attach_communication_context(
4952 resolved_step, item_read_state
4953 )
4954 resolved_step = step_runner._attach_resolved_tools(
4955 resolved_step, resolved
4956 )
4957 result_state = step_runner._invoke_agent(
4958 resolved, resolved_step, iter_state
4959 )
4960 return _MapReduceItemOutcome(
4961 result_state.get(step_id),
4962 result_state.get("step_context_metrics"),
4963 result_state.get("step_tool_meta"),
4964 _pending_channel_writes(item_working_set),
4965 tuple(result_state.get("errors") or []),
4966 )
4968 return _MapReducePlan(
4969 item_lister=lambda flat, ctx: step_runner._foreach_item_list(
4970 step, flat, ctx
4971 ),
4972 item_runner=run_item,
4973 result_assembler=result_assembler,
4974 channel_write_stager=_stage_map_reduce_item_channel_writes,
4975 )
4978def _make_map_reduce_parts(
4979 step: dict,
4980 plan: _MapReducePlan,
4981 step_runner: Any,
4982 channel_names: frozenset[str],
4983 channel_types: dict[str, str],
4984 max_concurrent: int,
4985) -> _MapReduceParts:
4986 step_id = step.get("id", "?")
4987 worker_id = f"{step_id}__worker"
4988 fallback_id = f"{step_id}__fallback"
4989 join_id = f"{step_id}__join"
4990 gate = threading.Semaphore(max(1, max_concurrent))
4991 return _MapReduceParts(
4992 dispatch_id=step_id,
4993 worker_id=worker_id,
4994 fallback_id=fallback_id,
4995 join_id=join_id,
4996 dispatch_fn=lambda state: {},
4997 route_fn=_make_map_reduce_route_fn(
4998 plan, step_runner, channel_names, worker_id, fallback_id, join_id
4999 ),
5000 worker_fn=_make_map_reduce_worker_fn(step, plan, step_runner, gate),
5001 join_fn=_make_map_reduce_join_fn(
5002 step, plan, step_runner, channel_names, channel_types
5003 ),
5004 )
5007def _resolved_send_concurrency(step: dict, build_ctx: dict) -> int:
5008 """The build-time resolved ``max_concurrent`` for a Send-eligible step."""
5009 return _resolve_int(step.get("max_concurrent"), build_ctx, 1)
5012def _is_send_parallel_step(
5013 step: dict, route_plan: "_RoutePlan", build_ctx: dict
5014) -> bool:
5015 """Whether a top-level `parallel` step fans out through native ``Send``.
5017 Restricted to a singleton group's statically concurrent, agent-backed,
5018 item-sourced step with no `routes`, no `condition`/`skip_llm_if`, and an
5019 explicit `max_concurrent > 1`; everything else keeps the in-node path.
5020 """
5021 if step.get("type") != "parallel":
5022 return False
5023 if step.get("id") in route_plan.target_node_ids:
5024 return False
5025 if step.get("routes") or not step.get("agent"):
5026 return False
5027 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
5028 return False
5029 if StepRunner._declared_item_source(step) is None:
5030 return False
5031 if step.get("max_concurrent") is None:
5032 return False
5033 return _resolved_send_concurrency(step, build_ctx) > 1
5036def _is_send_foreach_step(
5037 step: dict, route_plan: "_RoutePlan", build_ctx: dict
5038) -> bool:
5039 """Whether a top-level `foreach` step fans out through native ``Send``.
5041 Restricted to a singleton group's body- or agent-backed, item-sourced step
5042 with no `routes`, no `condition`/`skip_llm_if`, no step-level
5043 `output_channel`, and no `max_output`; an explicit `max_concurrent > 1` is
5044 required. The default (absent or ``1``) keeps the sequential in-node path,
5045 and the excluded edges keep their existing in-node semantics.
5046 """
5047 if step.get("type") != "foreach":
5048 return False
5049 if step.get("id") in route_plan.target_node_ids:
5050 return False
5051 if step.get("routes") or step.get("output_channel"):
5052 return False
5053 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
5054 return False
5055 if step.get("max_output") is not None:
5056 return False
5057 if StepRunner._declared_item_source(step) is None:
5058 return False
5059 if not (step.get("body") or step.get("agent")):
5060 return False
5061 if step.get("max_concurrent") is None:
5062 return False
5063 return _resolved_send_concurrency(step, build_ctx) > 1
5066class _ChannelGraphShim:
5067 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
5069 Preserves the invoke contract used by main.py and tests: a flat dict in,
5070 a flat dict out. The flat input is seeded through ``seed_committed`` and
5071 nested under ``__core__`` for the compiled graph; the committed snapshot is
5072 unwrapped back out. Streaming and state surfaces are not exposed because
5073 they would leak or accept the internal ``__core__`` shape; read-only
5074 introspection delegates to the compiled graph.
5075 """
5077 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
5078 self._compiled = compiled
5079 self._specs = specs
5081 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
5082 seeded = seed_committed(self._specs, flat_state)
5083 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
5084 return unwrap(result)
5086 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
5087 seeded = seed_committed(self._specs, flat_state)
5088 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
5089 return unwrap(result)
5091 def stream(self, *args: Any, **kwargs: Any) -> Any:
5092 self._reject_internal_shape("stream")
5094 def astream(self, *args: Any, **kwargs: Any) -> Any:
5095 self._reject_internal_shape("astream")
5097 def update_state(self, *args: Any, **kwargs: Any) -> Any:
5098 self._reject_internal_shape("update_state")
5100 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
5101 self._reject_internal_shape("aupdate_state")
5103 def get_state(self, *args: Any, **kwargs: Any) -> Any:
5104 self._reject_internal_shape("get_state")
5106 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
5107 self._reject_internal_shape("aget_state")
5109 def _reject_internal_shape(self, surface: str) -> None:
5110 raise NotImplementedError(
5111 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
5112 ".invoke()/.ainvoke() contract is exposed through the channel "
5113 "boundary shim"
5114 )
5116 def __getattr__(self, name: str) -> Any:
5117 return getattr(self._compiled, name)
5120class GraphBuilder:
5121 """
5122 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
5124 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
5125 käytetään vain {{ }} -templatejen evaluointiin.
5126 """
5128 def __init__(
5129 self,
5130 registry: ManifestRegistry,
5131 builtins: BuiltinExecutor | None = None,
5132 invoke_agent_fn: Any | None = None,
5133 run_context: dict | None = None,
5134 tracer: Any | None = None,
5135 ) -> None:
5136 self._registry = registry
5137 self._resolver = AgentResolver(registry)
5138 self._invoke_agent_fn = invoke_agent_fn
5139 self._builtins = builtins or _DEFAULT_BUILTINS
5140 if invoke_agent_fn is not None:
5141 self._builtins.invoke_agent_fn = invoke_agent_fn
5142 self._step_functions = StepFunctionExecutor(registry)
5143 self._tool_registry = build_platform_registry()
5144 self._tool_resolver = ToolResolver(self._tool_registry, self._step_functions)
5145 self._step_runner = StepRunner(
5146 self._resolver, self._builtins, invoke_agent_fn, run_context,
5147 step_functions=self._step_functions,
5148 tracer=tracer,
5149 tool_resolver=self._tool_resolver,
5150 )
5152 def build(self, pipeline: dict): # -> CompiledGraph
5153 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
5154 try:
5155 from langgraph.graph import StateGraph, END
5156 except ImportError as exc:
5157 raise ImportError(
5158 "langgraph is required to build a workflow. "
5159 "Install it with: pip install langgraph"
5160 ) from exc
5162 raw_steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
5164 channels: list[dict] = _validated_channels(
5165 (pipeline.get("spec") or {}).get("channels")
5166 )
5167 specs = parse_channel_specs(
5168 [{"name": ch["name"], "type": ch.get("type", LAST_VALUE)} for ch in channels]
5169 )
5170 channel_types = {spec.name: spec.type for spec in specs}
5171 channel_names = frozenset(channel_types)
5173 # Build-time call-step composition (ADR-0037 D4): each invoked Function
5174 # body is compiled into its own native LangGraph subgraph node, so the
5175 # runtime graph executes only native steps; a call step never survives.
5176 steps = self._compose_calls(
5177 raw_steps, channels, channel_names, channel_types
5178 )
5179 if channels:
5180 _validate_channel_names(channels, steps)
5181 _validate_channel_inputs(channels, steps)
5182 _validate_step_channel_bindings(steps, channel_names)
5183 ordered = _topological_order(steps)
5185 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
5186 envs = (pipeline.get("spec") or {}).get("envs")
5187 if envs is not None and not isinstance(envs, dict):
5188 raise RunHardFailure(
5189 "VALIDATION_ERROR",
5190 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
5191 "spec.envs must be a map of env name → image ref (ADR-0012)",
5192 )
5194 graph = self._assemble_graph(ordered, channel_names, channel_types)
5195 compiled = graph.compile()
5196 if channel_names:
5197 return _ChannelGraphShim(compiled, specs)
5198 return compiled
5200 def _compose_calls(
5201 self,
5202 steps: list[dict],
5203 channels: list[dict],
5204 channel_names: frozenset[str],
5205 channel_types: dict[str, str],
5206 ) -> list[dict]:
5207 """Replace `call` steps with compiled native subgraph nodes.
5209 Each invoked Function body is assembled through ``_assemble_graph`` and
5210 compiled, then substituted as a single ``__subgraph__`` placeholder that
5211 keeps the call id and its ``depends_on``. Downstream steps addressing the
5212 call id therefore address the subgraph node unchanged.
5213 """
5215 def assemble_subgraph(sub_steps: list[dict]):
5216 return self._assemble_graph(
5217 sub_steps, channel_names, channel_types
5218 ).compile()
5220 def assemble_while(while_step: dict, body_steps: list[dict]):
5221 return self._assemble_while_graph(
5222 while_step, body_steps, channel_names, channel_types
5223 )
5225 def assemble_foreach(foreach_step: dict, body_steps: list[dict]):
5226 return self._assemble_foreach_graph(
5227 foreach_step, body_steps, channel_names, channel_types
5228 )
5230 composer = _CallComposer(
5231 self._registry,
5232 assemble_subgraph,
5233 channels,
5234 channel_names,
5235 assemble_while=assemble_while,
5236 compose_whiles=True,
5237 assemble_foreach=assemble_foreach,
5238 compose_foreachs=True,
5239 build_ctx=self._step_runner._make_ctx({}),
5240 )
5241 return composer.compose(steps)
5243 def _assemble_while_graph(
5244 self,
5245 step: dict,
5246 body_steps: list[dict],
5247 channel_names: frozenset[str],
5248 channel_types: dict[str, str],
5249 ) -> tuple[Any, int]:
5250 """Compile a `while` step's body into a native LangGraph cycle.
5252 The cycle is ``head -> body... -> cond`` with a conditional back-edge
5253 from ``cond`` to ``head`` (continue) or ``exit`` (termination): the
5254 entry condition/initial value live in the head, the feed + exit
5255 evaluation in the cond node, and the exit node writes the documented
5256 result. Returns the compiled cycle and the recursion bound the wrapper
5257 must apply (a cycle consumes one super-step per node per iteration).
5258 """
5259 from langgraph.graph import StateGraph, END
5261 step_id = step.get("id", "?")
5262 step_runner = self._step_runner
5263 step_runner.set_channel_routing(channel_names)
5265 state_schema = (
5266 _build_state_schema()
5267 if channel_names
5268 else Annotated[dict[str, Any], _merge_state]
5269 )
5270 graph: StateGraph = StateGraph(state_schema)
5272 max_iter = _resolve_int(
5273 step.get("max_iterations", 2), step_runner._make_ctx({}), 2
5274 )
5275 head_id = f"__while_{step_id}__head"
5276 cond_id = f"__while_{step_id}__cond"
5277 exit_id = f"__while_{step_id}__exit"
5278 first_id = body_steps[0]["id"] if body_steps else cond_id
5280 graph.add_node(
5281 head_id,
5282 _make_while_head_node(step, step_runner, channel_names, max_iter),
5283 )
5284 for sub in body_steps:
5285 graph.add_node(
5286 sub["id"],
5287 _make_while_body_node(
5288 sub, step_runner, channel_names, channel_types
5289 ),
5290 )
5291 graph.add_node(
5292 cond_id, _make_while_cond_node(step, step_runner, channel_names)
5293 )
5294 graph.add_node(
5295 exit_id, _make_while_exit_node(step, step_runner, channel_names)
5296 )
5298 graph.set_entry_point(head_id)
5299 head_paths = {first_id: first_id, exit_id: exit_id, END: END}
5300 graph.add_conditional_edges(
5301 head_id,
5302 _make_while_head_router(step, channel_names, first_id, exit_id, END),
5303 head_paths,
5304 )
5305 for previous, sub in zip(body_steps, body_steps[1:]):
5306 graph.add_edge(previous["id"], sub["id"])
5307 if body_steps:
5308 graph.add_edge(body_steps[-1]["id"], cond_id)
5309 graph.add_conditional_edges(
5310 cond_id,
5311 _make_while_cond_router(step, channel_names, head_id, exit_id),
5312 {head_id: head_id, exit_id: exit_id},
5313 )
5314 graph.add_edge(exit_id, END)
5316 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5317 return graph.compile(), recursion_limit
5319 def _assemble_foreach_graph(
5320 self,
5321 step: dict,
5322 body_steps: list[dict],
5323 channel_names: frozenset[str],
5324 channel_types: dict[str, str],
5325 ) -> tuple[Any, int]:
5326 """Compile a sequential `foreach` step's body into a native LangGraph cycle.
5328 The cycle is ``head -> (body... | agent) -> cond`` with a conditional
5329 back-edge from ``cond`` to ``head`` (next item) or ``exit`` (exhaustion
5330 or `max_output` reached). The head resolves the item list once and
5331 injects the current item; the cond collects the item result; the exit
5332 node writes the documented aggregate. Returns the compiled cycle and
5333 the recursion bound the wrapper must apply.
5334 """
5335 from langgraph.graph import StateGraph, END
5337 step_id = step.get("id", "?")
5338 step_runner = self._step_runner
5339 step_runner.set_channel_routing(channel_names)
5341 state_schema = (
5342 _build_state_schema()
5343 if channel_names
5344 else Annotated[dict[str, Any], _merge_state]
5345 )
5346 graph: StateGraph = StateGraph(state_schema)
5348 max_iter = _resolve_int(
5349 step.get("max_iterations"),
5350 step_runner._make_ctx({}),
5351 _FOREACH_DEFAULT_MAX_ITER,
5352 )
5353 head_id = f"__foreach_{step_id}__head"
5354 cond_id = f"__foreach_{step_id}__cond"
5355 exit_id = f"__foreach_{step_id}__exit"
5356 agent_id = f"__foreach_{step_id}__agent"
5358 graph.add_node(
5359 head_id, _make_foreach_head_node(step, step_runner, channel_names)
5360 )
5361 if body_steps:
5362 for sub in body_steps:
5363 graph.add_node(
5364 sub["id"],
5365 _make_foreach_body_node(
5366 sub, step_runner, channel_names, channel_types
5367 ),
5368 )
5369 first_id = body_steps[0]["id"]
5370 else:
5371 graph.add_node(
5372 agent_id,
5373 _make_foreach_agent_node(
5374 step, step_runner, channel_names, channel_types
5375 ),
5376 )
5377 first_id = agent_id
5378 graph.add_node(
5379 cond_id, _make_foreach_cond_node(step, step_runner, channel_names)
5380 )
5381 graph.add_node(
5382 exit_id, _make_foreach_exit_node(step, step_runner, channel_names)
5383 )
5385 graph.set_entry_point(head_id)
5386 graph.add_conditional_edges(
5387 head_id,
5388 _make_foreach_head_router(step, channel_names, first_id, exit_id, END),
5389 {first_id: first_id, exit_id: exit_id, END: END},
5390 )
5391 for previous, sub in zip(body_steps, body_steps[1:]):
5392 graph.add_edge(previous["id"], sub["id"])
5393 if body_steps:
5394 graph.add_edge(body_steps[-1]["id"], cond_id)
5395 else:
5396 graph.add_edge(agent_id, cond_id)
5397 graph.add_conditional_edges(
5398 cond_id,
5399 _make_foreach_cond_router(step, channel_names, head_id, exit_id),
5400 {head_id: head_id, exit_id: exit_id},
5401 )
5402 graph.add_edge(exit_id, END)
5404 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5405 return graph.compile(), recursion_limit
5407 def _assemble_graph(
5408 self,
5409 steps: list[dict],
5410 channel_names: frozenset[str],
5411 channel_types: dict[str, str],
5412 ) -> "StateGraph":
5413 from langgraph.graph import StateGraph, END
5415 # Group steps into LangGraph nodes:
5416 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
5417 nodes: list[list[dict]] = []
5418 for step in steps:
5419 dep_sig = frozenset(step.get("depends_on") or [])
5420 if nodes:
5421 last_group_sig = frozenset(
5422 dep
5423 for s in nodes[-1]
5424 for dep in (s.get("depends_on") or [])
5425 )
5426 last_types = {s.get("type") for s in nodes[-1]}
5427 if (
5428 dep_sig == last_group_sig
5429 and step.get("type") == "parallel"
5430 and last_types == {"parallel"}
5431 and not step.get("routes")
5432 and not any(s.get("routes") for s in nodes[-1])
5433 ):
5434 nodes[-1].append(step)
5435 continue
5436 nodes.append([step])
5438 state_schema = (
5439 _build_state_schema()
5440 if channel_names
5441 else Annotated[dict[str, Any], _merge_state]
5442 )
5443 graph: StateGraph = StateGraph(state_schema)
5444 step_runner = self._step_runner
5445 step_runner.set_channel_routing(channel_names)
5447 node_ids: list[str] = []
5448 for group in nodes:
5449 if len(group) == 1:
5450 node_ids.append(group[0]["id"])
5451 else:
5452 node_ids.append("parallel_" + "_".join(s["id"] for s in group))
5454 route_plan = _build_route_plan(nodes, node_ids) if nodes else None
5456 send_parts: dict[int, _MapReduceParts] = {}
5457 if route_plan is not None:
5458 build_ctx = step_runner._make_ctx({})
5459 for index, group in enumerate(nodes):
5460 if len(group) != 1:
5461 continue
5462 send_candidate = group[0]
5463 if _is_send_parallel_step(send_candidate, route_plan, build_ctx):
5464 plan = _make_parallel_map_reduce_plan(
5465 send_candidate, step_runner
5466 )
5467 elif _is_send_foreach_step(send_candidate, route_plan, build_ctx):
5468 plan = _make_foreach_map_reduce_plan(
5469 send_candidate, step_runner, channel_names, channel_types
5470 )
5471 else:
5472 continue
5473 send_parts[index] = _make_map_reduce_parts(
5474 send_candidate,
5475 plan,
5476 step_runner,
5477 channel_names,
5478 channel_types,
5479 _resolved_send_concurrency(send_candidate, build_ctx),
5480 )
5482 entry_ids: list[str] = []
5483 exit_ids: list[str] = []
5484 for index, group in enumerate(nodes):
5485 parts = send_parts.get(index)
5486 if parts is None:
5487 captured_group = list(group)
5488 if len(captured_group) == 1:
5489 first = captured_group[0]
5490 if first.get("type") == _SUBGRAPH_STEP_TYPE:
5491 if first.get(_SUBGRAPH_WHILE_KEY):
5492 subgraph_node = _make_while_subgraph_node(
5493 first, step_runner, channel_names
5494 )
5495 elif first.get(_SUBGRAPH_FOREACH_KEY):
5496 subgraph_node = _make_foreach_subgraph_node(
5497 first, step_runner, channel_names
5498 )
5499 else:
5500 subgraph_node = _make_subgraph_node(
5501 first[_SUBGRAPH_GRAPH_KEY],
5502 first.get("condition"),
5503 step_runner,
5504 channel_names,
5505 first.get(_SUBGRAPH_STEP_IDS_KEY),
5506 )
5507 graph.add_node(node_ids[index], subgraph_node)
5508 entry_ids.append(node_ids[index])
5509 exit_ids.append(node_ids[index])
5510 continue
5511 fn = _make_step_node(
5512 first, step_runner, channel_names, channel_types,
5513 )
5514 else:
5515 fn = _make_parallel_node(
5516 captured_group,
5517 step_runner,
5518 channel_names,
5519 channel_types,
5520 )
5521 graph.add_node(node_ids[index], fn)
5522 entry_ids.append(node_ids[index])
5523 exit_ids.append(node_ids[index])
5524 continue
5525 fallback_step = group[0]
5526 graph.add_node(parts.dispatch_id, parts.dispatch_fn)
5527 graph.add_node(parts.worker_id, parts.worker_fn)
5528 graph.add_node(
5529 parts.fallback_id,
5530 _make_step_node(
5531 fallback_step, step_runner, channel_names, channel_types
5532 ),
5533 )
5534 graph.add_node(parts.join_id, parts.join_fn)
5535 graph.add_conditional_edges(
5536 parts.dispatch_id,
5537 parts.route_fn,
5538 [parts.worker_id, parts.fallback_id],
5539 )
5540 graph.add_edge(parts.worker_id, parts.join_id)
5541 graph.add_edge(parts.fallback_id, parts.join_id)
5542 entry_ids.append(parts.dispatch_id)
5543 exit_ids.append(parts.join_id)
5545 if nodes:
5546 _wire_dependency_edges(
5547 graph, nodes, node_ids, END, route_plan, entry_ids, exit_ids
5548 )
5549 _wire_route_edges(graph, route_plan, END)
5550 else:
5551 graph.add_node("noop", lambda s: s)
5552 graph.set_entry_point("noop")
5553 graph.add_edge("noop", END)
5555 return graph
5558# ---------------------------------------------------------------------------
5559# Julkinen API
5560# ---------------------------------------------------------------------------
5562_registry_cache: ManifestRegistry | None = None
5565def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
5566 global _registry_cache
5567 if _registry_cache is None or dirs is not None:
5568 loader = ManifestLoader(dirs)
5569 manifests = loader.load_all()
5570 _registry_cache = ManifestRegistry(manifests)
5571 return _registry_cache
5574def build_workflow(
5575 pipeline_name: str,
5576 *,
5577 dirs: list[Path] | None = None,
5578 builtins: BuiltinExecutor | None = None,
5579 invoke_agent_fn: Any | None = None,
5580 run_context: dict | None = None,
5581 tracer: Any | None = None,
5582):
5583 """
5584 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
5585 LangGraph-workflown.
5587 Args:
5588 pipeline_name: Pipeline metadata.name -arvo.
5589 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
5590 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
5591 invoke_agent_fn: fn(resolved_spec, step, state) → state.
5592 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
5594 Returns:
5595 Compilattu LangGraph CompiledGraph.
5596 """
5597 registry = _get_registry(dirs)
5598 pipeline = registry.get("Pipeline", pipeline_name)
5599 if pipeline is None:
5600 available = [
5601 (m.get("metadata") or {}).get("name")
5602 for m in registry.all_of_kind("Pipeline")
5603 ]
5604 raise RunHardFailure(
5605 "VALIDATION_ERROR",
5606 f"Pipeline '{pipeline_name}' not found. Available: {available}",
5607 )
5608 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
5609 return builder.build(pipeline)