Coverage for core/langgraph/runner.py: 94%
2547 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 03:05 +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 A list condition is the logical AND of its elements; this is how a
858 call-level condition is combined with a body step's own condition.
860 Strict (short-circuit + hard error on unresolved reads) per R1: an
861 undefined or skipped-step field read without short-circuit protection
862 raises, never silently evaluates to falsy.
863 """
864 if cond is None:
865 return True
866 if isinstance(cond, (list, tuple)):
867 return all(_eval_condition(part, ctx) for part in cond)
868 resolved = _resolve_template(cond, ctx, strict=True)
869 if isinstance(resolved, bool):
870 return resolved
871 if isinstance(resolved, str):
872 return resolved.lower() not in ('false', '0', '', 'none', 'null')
873 return bool(resolved)
876def _and_conditions(outer: Any, inner: Any) -> Any:
877 """Combine two step conditions with logical AND. None → the other side."""
878 if outer is None:
879 return inner
880 if inner is None:
881 return outer
882 if isinstance(inner, list):
883 return [outer, *inner]
884 return [outer, inner]
887def _resolve_int(value: Any, ctx: dict, default: int) -> int:
888 v = _resolve_template(value, ctx) if isinstance(value, str) else value
889 try:
890 return int(v)
891 except (TypeError, ValueError):
892 return default
895# ---------------------------------------------------------------------------
896# ManifestLoader
897# ---------------------------------------------------------------------------
900def _parse_manifest_file(path: Path) -> list[dict]:
901 """Palauttaa kaikki apiVersion: agent-platform/v1 -dokumentit tiedostosta."""
902 try:
903 content = path.read_text(encoding="utf-8")
904 except OSError as exc:
905 log.warning("Cannot read %s: %s", path, exc)
906 return []
908 results: list[dict] = []
909 for raw in content.split("\n---"):
910 raw_clean = "\n".join(
911 line for line in raw.splitlines()
912 if not line.strip().startswith("#")
913 )
914 try:
915 doc = yaml.safe_load(raw_clean)
916 except yaml.YAMLError as exc:
917 log.warning("YAML parse error in %s: %s", path, exc)
918 continue
919 if isinstance(doc, dict) and doc.get("apiVersion") == "agent-platform/v1":
920 results.append(doc)
921 return results
924class ManifestLoader:
925 """Skannaa hakemistot ja lataa kaikki YAML-manifestit."""
927 def __init__(self, dirs: list[Path] | None = None) -> None:
928 self._dirs = dirs or MANIFEST_DIRS
930 def load_all(self) -> list[dict]:
931 manifests: list[dict] = []
932 for d in self._dirs:
933 if not d.exists():
934 log.debug("Manifest dir not found: %s", d)
935 continue
936 for path in sorted(d.glob("**/*.yaml")):
937 docs = _parse_manifest_file(path)
938 manifests.extend(docs)
939 if docs:
940 log.debug("Loaded %d manifest(s) from %s", len(docs), path)
941 return manifests
944# ---------------------------------------------------------------------------
945# ManifestRegistry
946# ---------------------------------------------------------------------------
948ResourceKey = tuple[str, str] # (kind, name)
951class ManifestRegistry:
952 """Indeksoi ladatut manifestit kind+name avaimella."""
954 def __init__(self, manifests: list[dict]) -> None:
955 self._store: dict[ResourceKey, dict] = {}
956 for doc in manifests:
957 kind = doc.get("kind", "")
958 name = (doc.get("metadata") or {}).get("name", "")
959 if kind and name:
960 key: ResourceKey = (kind, name)
961 if key in self._store:
962 log.warning("Duplicate manifest %s/%s — last one wins", kind, name)
963 self._store[key] = doc
965 def get(self, kind: str, name: str) -> dict | None:
966 return self._store.get((kind, name))
968 def all_of_kind(self, kind: str) -> list[dict]:
969 return [v for (k, _), v in self._store.items() if k == kind]
972# ---------------------------------------------------------------------------
973# AgentResolver
974# ---------------------------------------------------------------------------
976def _deep_merge(base: dict, override: dict) -> dict:
977 """Syvä merge: override voittaa konflikteissa."""
978 result = dict(base)
979 for k, v in override.items():
980 if isinstance(v, dict) and isinstance(result.get(k), dict):
981 result[k] = _deep_merge(result[k], v)
982 else:
983 result[k] = v
984 return result
987def _tool_has_description(tool: Any) -> bool:
988 """Report whether a resolved tool exposes a non-empty model description."""
989 description = (tool.definition().get("function") or {}).get("description")
990 return isinstance(description, str) and bool(description)
993class AgentResolver:
994 """
995 Palauttaa fully-resolved agent spec:n instanssille.
996 Jos instanssilla on based_on, mergee Agent-base + AgentInstance-overridet.
997 """
999 def __init__(self, registry: ManifestRegistry) -> None:
1000 self._registry = registry
1001 self._cache: dict[str, dict] = {}
1003 def resolve(self, instance_name: str) -> dict:
1004 if instance_name in self._cache:
1005 return self._cache[instance_name]
1007 instance = self._registry.get("AgentInstance", instance_name)
1008 if instance is None:
1009 agent = self._registry.get("Agent", instance_name)
1010 if agent is None:
1011 raise RunHardFailure(
1012 "VALIDATION_ERROR",
1013 f"No Agent or AgentInstance named '{instance_name}'",
1014 {"agent": instance_name},
1015 )
1016 resolved = agent
1017 else:
1018 based_on = (instance.get("spec") or {}).get("based_on")
1019 if based_on:
1020 base_agent = self._registry.get("Agent", based_on)
1021 if base_agent is None:
1022 raise RunHardFailure(
1023 "VALIDATION_ERROR",
1024 f"AgentInstance '{instance_name}' references unknown "
1025 f"Agent '{based_on}'",
1026 {"agent": instance_name, "based_on": based_on},
1027 )
1028 base_spec = base_agent.get("spec") or {}
1029 inst_spec = {k: v for k, v in (instance.get("spec") or {}).items()
1030 if k != "based_on"}
1031 merged_spec = _deep_merge(base_spec, inst_spec)
1032 else:
1033 merged_spec = {k: v for k, v in (instance.get("spec") or {}).items()
1034 if k != "based_on"}
1036 resolved = {
1037 "apiVersion": "agent-platform/v1",
1038 "kind": "Agent",
1039 "metadata": {"name": instance_name},
1040 "spec": merged_spec,
1041 }
1043 self._cache[instance_name] = resolved
1044 return resolved
1047# ---------------------------------------------------------------------------
1048# State type
1049# ---------------------------------------------------------------------------
1051PipelineState = dict[str, Any]
1053# Engine-reserved channel key: the existing flat Core-managed JSON state lives
1054# under this key when a pipeline declares spec.channels.
1055_CORE = CORE_KEY
1057# Core-owned per-step map-reduce accumulator. A concurrent top-level `parallel`
1058# step fans out through native `Send`; each worker writes `{index: item_result}`
1059# under its step id and the join node consumes it. In the channels path the
1060# accumulator is a top-level state channel; in the no-channels path it is a
1061# reserved flat key folded by `_merge_state`.
1062_MAP_REDUCE_ACCUMULATOR_KEY = "__map_reduce_accum__"
1064# Reserved key carrying a worker's source item index through a `Send` payload.
1065_MAP_REDUCE_ITEM_INDEX = "__map_reduce_item_index__"
1067# Flat-state keys the engine owns; channel names may not shadow them.
1068_CORE_STATE_KEYS = frozenset({
1069 "errors", "stats", "run_status", "project_root", "run_id",
1070 "_stats", "step_context_metrics", "step_tool_meta",
1071 _MAP_REDUCE_ACCUMULATOR_KEY,
1072})
1074# Engine-local iteration keys the loop machinery injects into the read state.
1075# They are engine plumbing, not user state, so a colliding channel key must
1076# never shadow them.
1077_ENGINE_LOCAL_ITERATION_KEYS = ("item", "_item", "_foreach_item", "loop")
1080def _iteration_discriminator(state: dict) -> str | None:
1081 """The engine-local iteration token for *state*, or None when not iterating.
1083 Concurrent foreach/parallel items share a step id but differ in their
1084 iteration key, so the synthesized call id uses this to stay distinct on the
1085 in-cluster sidecar params file. The item is digested so the id stays a
1086 bounded, filename-safe length.
1087 """
1088 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1089 if key in state:
1090 digest = hashlib.sha1(repr(state[key]).encode("utf-8")).hexdigest()[:12]
1091 return f"{key}-{digest}"
1092 return None
1096def _merge_state(left: dict, right: dict) -> dict:
1097 if not left:
1098 return dict(right)
1099 side_channel_keys = ("step_context_metrics", "step_tool_meta")
1100 merged = dict(left)
1101 for key, value in right.items():
1102 if key == _MAP_REDUCE_ACCUMULATOR_KEY:
1103 folded = accumulate_map_reduce(merged.get(key), value)
1104 if folded:
1105 merged[key] = folded
1106 else:
1107 merged.pop(key, None)
1108 continue
1109 existing = merged.get(key)
1110 if (
1111 key in side_channel_keys
1112 and isinstance(existing, dict)
1113 and isinstance(value, dict)
1114 ):
1115 merged[key] = _merge_side_channel(existing, value)
1116 else:
1117 merged[key] = value
1118 return merged
1121def _build_state_schema() -> Any:
1122 """Build the channel-path TypedDict: the Core namespace plus the accumulator.
1124 Channel context is Core-owned, so there are no per-channel native LangGraph
1125 fields; the flat committed snapshot (core keys + channel keys) travels
1126 node-to-node under ``__core__``. The map-reduce accumulator is a separate
1127 top-level channel so native ``Send`` workers reduce into it; LangGraph stays
1128 the graph executor.
1129 """
1130 from typing import TypedDict
1132 return TypedDict(
1133 "PipelineState",
1134 {
1135 _CORE: Annotated[dict[str, Any], _merge_state],
1136 _MAP_REDUCE_ACCUMULATOR_KEY: Annotated[
1137 dict[str, Any], accumulate_map_reduce
1138 ],
1139 },
1140 )
1143# Sentinel for optional kwargs distinguishing "not provided" from None
1144_UNSET: Any = object()
1146# ---------------------------------------------------------------------------
1147# BuiltinExecutor
1148# ---------------------------------------------------------------------------
1150class BuiltinExecutor:
1151 """
1152 Ajaa builtin-komponentit.
1153 Uusia komponentteja voi rekisteröidä register()-metodilla.
1154 Nimet normalisoidaan: viiva → alaviiva (esim. 'list-dir-entries' == 'list_dir_entries').
1155 """
1157 # Lazy modules are imported on first use (they need httpx / heavier deps).
1158 # Profile-specific builtins live here: the PR-profile github-pr-comment
1159 # and the TICKET-profile ticket-publish (per-run local feature branch; push/PR disabled).
1160 _LAZY_MODULES: tuple[str, ...] = ("_builtins_pr", "_builtins_ticket")
1162 def __init__(self) -> None:
1163 self._handlers: dict[str, Any] = {
1164 "passthrough": self._passthrough,
1165 # Engine-internal call-expansion scaffolding (ADR-0037 D4): these are
1166 # synthesized by the build-time call composer, never authored in a
1167 # manifest. A call step becomes a compiled subgraph; these two op
1168 # steps run inside it, binding params and emitting the call result.
1169 "__call_bind": self._call_bind,
1170 "__call_emit": self._call_emit,
1171 }
1172 # Pipeline's LLM invoker (spec, step, state) → state, bound by
1173 # GraphBuilder. Profile builtins that make LLM decisions
1174 # (ticket-dor / ticket-review) read it via their register_all
1175 # wrapper — the executor must have it set before the builtin
1176 # executes (binding happens at build_workflow time).
1177 self.invoke_agent_fn: Any | None = None
1178 self._lazy_loaded = False
1180 def register(self, name: str, fn: Any) -> None:
1181 self._handlers[name] = fn
1182 # Also register with normalised name (hyphen → underscore)
1183 normalised = name.replace("-", "_")
1184 if normalised != name:
1185 self._handlers[normalised] = fn
1187 def _load_lazy_modules(self) -> None:
1188 if self._lazy_loaded:
1189 return
1190 self._lazy_loaded = True
1191 import importlib
1192 for module_name in self._LAZY_MODULES:
1193 try:
1194 mod = importlib.import_module(f".{module_name}", package="langgraph")
1195 except ImportError:
1196 continue
1197 mod.register_all(self)
1199 def has(self, name: str) -> bool:
1200 """True when *name* resolves in the builtin register (lazy modules included)."""
1201 self._load_lazy_modules()
1202 return name in self._handlers or name.replace("-", "_") in self._handlers
1204 def get(self, name: str) -> Any:
1205 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
1206 if fn is not None:
1207 return fn
1208 self._load_lazy_modules()
1209 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
1210 if fn is None:
1211 raise RunHardFailure(
1212 "VALIDATION_ERROR",
1213 f"Unknown builtin component '{name}'",
1214 {"component": name},
1215 )
1216 return fn
1218 def execute(self, component: str, step: dict, state: PipelineState) -> PipelineState:
1219 fn = self.get(component)
1220 try:
1221 return fn(step, state)
1222 except RunHardFailure as exc:
1223 log.error("[hard-failure] %s: %s", exc.code, exc.message)
1224 raise
1226 # --- default builtins ---
1228 @staticmethod
1229 def _passthrough(step: dict, state: PipelineState) -> PipelineState:
1230 return state
1232 @staticmethod
1233 def _call_bind(step: dict, state: PipelineState) -> PipelineState:
1234 """Call-scope entry (synthesized): bind the function call's params.
1236 The caller's function params are resolved against the current state and
1237 stored at ``state[<call_id>] = {"params": {...}}`` so the (rewritten)
1238 function-body templates ``{{ <call_id>.params.<name> }}`` resolve. The
1239 call's reserved ``__call_id`` key is stripped from the bound params.
1240 """
1241 call_id = (step.get("params") or {}).get("__call_id")
1242 if not call_id:
1243 raise ValueError("__call_bind missing __call_id (not a synthesized step)")
1244 bound = {
1245 k: v for k, v in (step.get("params") or {}).items() if k != "__call_id"
1246 }
1247 return {**state, call_id: {"params": bound}}
1249 @staticmethod
1250 def _call_emit(step: dict, state: PipelineState) -> PipelineState:
1251 """Call-scope exit (synthesized): surface the function's outputs.
1253 Per the call step-result contract (00-process-contract r.70) the caller
1254 sees only ``state[<call_id>] = {outputs: {<name>: <source value>}}`` —
1255 the function body step results are invisible to the caller. Each output
1256 name resolves to the manifest interface ``outputs[].source`` dot-path.
1258 Optionality (ADR-0031): an output flagged ``optional: true`` is skipped
1259 when its source-path is unresolved; an output that is NOT optional must
1260 resolve or a RunHardFailure is raised (no silent None — ADR-0012).
1261 """
1262 params = step.get("params") or {}
1263 call_id = params.get("__call_id")
1264 if not call_id:
1265 raise ValueError("__call_emit missing __call_id (not a synthesized step)")
1266 outputs_spec = params.get("__outputs") or []
1267 outputs: dict[str, Any] = {}
1268 for spec in outputs_spec:
1269 if not isinstance(spec, dict):
1270 continue
1271 name = spec.get("name")
1272 source = spec.get("source")
1273 if name is None:
1274 continue
1275 optional = bool(spec.get("optional", False))
1276 if source is None:
1277 # No declared source-path: treat as unresolved against the
1278 # optionality gate (ADR-0012 — no silent None for required).
1279 if optional:
1280 continue
1281 raise RunHardFailure(
1282 "UNRESOLVED_OUTPUT",
1283 f"call '{call_id}' output '{name}' has no source "
1284 "(required output — no silent None, ADR-0012)",
1285 {"call_id": call_id, "name": name, "source": source},
1286 )
1287 try:
1288 value = _dot_get_strict(state, source)
1289 except ValueError:
1290 # Unresolved source-path: required → hard error; optional → skip.
1291 if optional:
1292 continue
1293 raise RunHardFailure(
1294 "UNRESOLVED_OUTPUT",
1295 f"call '{call_id}' output '{name}' unresolved: source "
1296 f"'{source}' not found in state (required output — "
1297 "no silent None, ADR-0012)",
1298 {"call_id": call_id, "name": name, "source": source},
1299 ) from None
1300 outputs[name] = value
1301 return {**state, call_id: {"outputs": outputs}}
1304_DEFAULT_BUILTINS = BuiltinExecutor()
1307def _register_domain_builtins(executor: BuiltinExecutor) -> None:
1308 """Rekisteröi v1-domain-builtinit (list-dir-entries, ticket-publisher, ...).
1310 Erotettu omaan moduliin (`_builtins.py`) — runner.py pysyy generic-engine,
1311 builtinit ovat generiisiä mekanismeja (ADR-0016/0017).
1312 """
1313 try:
1314 from langgraph import _builtins # type: ignore
1315 except ImportError:
1316 # Kun runner.py ajetaan ilman pakettirakennetta (esim. /app/langgraph)
1317 import importlib.util
1318 spec = importlib.util.spec_from_file_location(
1319 "_builtins", Path(__file__).parent / "_builtins.py"
1320 )
1321 if spec and spec.loader:
1322 _builtins = importlib.util.module_from_spec(spec)
1323 spec.loader.exec_module(_builtins) # type: ignore[union-attr]
1324 else:
1325 log.warning("Cannot load _builtins.py — domain builtins disabled")
1326 return
1327 _builtins.register_all(executor)
1328 _register_function_builtins(executor)
1331def _register_function_builtins(executor: BuiltinExecutor) -> None:
1332 """Rekisteröi function-engine builtinit (question-block-form).
1334 Erotettu omaan moduliin (`_builtins_functions.py`) — runner.py pysyy
1335 generic-engine, function-builtinit pysyvät puhtaina ja erillisinä.
1336 """
1337 try:
1338 from langgraph import _builtins_functions # type: ignore
1339 except ImportError:
1340 import importlib.util
1341 spec = importlib.util.spec_from_file_location(
1342 "_builtins_functions", Path(__file__).parent / "_builtins_functions.py"
1343 )
1344 if spec and spec.loader:
1345 _builtins_functions = importlib.util.module_from_spec(spec)
1346 spec.loader.exec_module(_builtins_functions) # type: ignore[union-attr]
1347 else:
1348 log.warning("Cannot load _builtins_functions.py — function builtins disabled")
1349 return
1350 _builtins_functions.register_all(executor)
1353_register_domain_builtins(_DEFAULT_BUILTINS)
1355# ---------------------------------------------------------------------------
1356# StepRunner
1357# ---------------------------------------------------------------------------
1359class StepRunner:
1360 """
1361 Ajaa yksittäisen step:n sen typin mukaan.
1363 run_context: dict joka lisätään template-kontekstiin. Esim:
1364 {"budget": {"max_parallel": 16, ...}, "config": {...}}
1365 """
1367 def __init__(
1368 self,
1369 resolver: AgentResolver,
1370 builtins: BuiltinExecutor | None = None,
1371 invoke_agent_fn: Any | None = None,
1372 run_context: dict | None = None,
1373 *,
1374 step_functions: StepFunctionExecutor | None = None,
1375 tracer: Any | None = None,
1376 tool_resolver: Any | None = None,
1377 ) -> None:
1378 self._resolver = resolver
1379 self._builtins = builtins or _DEFAULT_BUILTINS
1380 self._invoke_agent = self._wrap_hard_failure_gate(
1381 invoke_agent_fn or self._dummy_invoke
1382 )
1383 self._run_context: dict = run_context or {}
1384 self._step_functions = step_functions
1385 self._tracer = tracer
1386 self._tool_resolver = tool_resolver
1387 self._channel_names: frozenset[str] = frozenset()
1389 def set_channel_routing(self, channel_names: frozenset[str]) -> None:
1390 """Attach the build-time routing context nested body steps route with."""
1391 self._channel_names = channel_names
1393 @staticmethod
1394 def _should_skip_llm(step: dict, ctx: dict) -> bool:
1395 """Truthiness of a step's `skip_llm_if` expression (ENG-12, Q4).
1397 Lenient (non-strict) evaluation: a missing/undefined field resolves to
1398 falsy, so the step behaves normally (LLM invoked). Only an explicit
1399 truthy expression skips the LLM.
1400 """
1401 raw = step.get("skip_llm_if")
1402 if raw is None:
1403 return False
1404 resolved = _resolve_template(raw, ctx)
1405 if isinstance(resolved, bool):
1406 return resolved
1407 if isinstance(resolved, str):
1408 return resolved.lower() not in ('false', '0', '', 'none', 'null')
1409 return bool(resolved)
1411 @staticmethod
1412 def _skip_llm_passthrough(
1413 step: dict,
1414 state: PipelineState,
1415 ctx: dict,
1416 read_state: PipelineState,
1417 ) -> PipelineState:
1418 """Deterministic passthrough of a step's resolved input (no LLM).
1420 `state[step_id]` becomes the resolved `input` value (the parallel
1421 fan-out envelope `{results: [...]}` when the input resolves to a list).
1422 ``read_state`` is the working-set overlay the input resolves against;
1423 the returned state is still built from the raw ``state``.
1424 """
1425 step_id = step.get("id", "?")
1426 input_spec = step.get("input")
1427 resolved = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
1428 if isinstance(resolved, list):
1429 log.debug("[step] %s skip_llm_if passthrough (fanout, no LLM)", step_id)
1430 return {**state, step_id: {"results": resolved}}
1431 log.debug("[step] %s skip_llm_if passthrough (no LLM)", step_id)
1432 return {**state, step_id: resolved}
1434 @staticmethod
1435 def _declared_item_source(step: dict) -> tuple[str, Any] | None:
1436 """A fan-out step's declared item source as ``(kind, spec)``, else None.
1438 ``input`` takes precedence over ``over``. ``over`` addresses a flat
1439 state key; ``input`` supports templates, bracket literals, and dot-paths
1440 (so a runtime envelope set is ``input: "<step_id>.iteration"``). A step
1441 declaring neither has no item source.
1442 """
1443 if step.get("input") is not None:
1444 return ("input", step.get("input"))
1445 if step.get("over"):
1446 return ("over", step.get("over"))
1447 return None
1449 @staticmethod
1450 def _resolve_item_source(
1451 source: tuple[str, Any], state: PipelineState, ctx: dict
1452 ) -> Any:
1453 """Resolve a declared item source to its raw value (no validation)."""
1454 kind, spec = source
1455 if kind == "input":
1456 return _resolve_input(spec, state, ctx)
1457 return state.get(spec)
1459 @staticmethod
1460 def _unresolvable_item_source_failure(
1461 step_id: str, source: tuple[str, Any]
1462 ) -> "RunHardFailure":
1463 kind, spec = source
1464 return RunHardFailure(
1465 "VALIDATION_ERROR",
1466 f"step '{step_id}' item source '{kind}: {spec}' is unresolvable "
1467 "(no silent fallback — ADR-0012)",
1468 {"step_id": step_id, "source_kind": kind, "source": spec},
1469 )
1471 @staticmethod
1472 def _empty_item_source_failure(
1473 step_id: str, source: tuple[str, Any]
1474 ) -> "RunHardFailure":
1475 kind, spec = source
1476 return RunHardFailure(
1477 "VALIDATION_ERROR",
1478 f"step '{step_id}' item source '{kind}: {spec}' resolved to an "
1479 "empty list (no silent fallback — ADR-0012)",
1480 {"step_id": step_id, "source_kind": kind, "source": spec},
1481 )
1483 @staticmethod
1484 def _parallel_fanout_items(
1485 step: dict, state: PipelineState, ctx: dict
1486 ) -> list | None:
1487 """The resolved fan-out item list for a parallel step, else None.
1489 A parallel step whose resolved item source is a list invokes the agent
1490 once per item and stores a ``{"results": [...]}`` envelope instead of a
1491 single agent output. A parallel step declaring no item source keeps the
1492 single-agent fallback; a declared source that is unresolvable or empty
1493 is a ``VALIDATION_ERROR`` (ADR-0012).
1494 """
1495 if step.get("type") != "parallel":
1496 return None
1497 source = StepRunner._declared_item_source(step)
1498 if source is None:
1499 return None
1500 items = StepRunner._resolve_item_source(source, state, ctx)
1501 if items is None:
1502 raise StepRunner._unresolvable_item_source_failure(
1503 step.get("id", "?"), source
1504 )
1505 if not isinstance(items, list):
1506 return None
1507 if not items:
1508 raise StepRunner._empty_item_source_failure(step.get("id", "?"), source)
1509 return items
1511 @staticmethod
1512 def _parallel_fanout_item_list(
1513 step: dict, state: PipelineState, ctx: dict
1514 ) -> list | None:
1515 """The fan-out item list with the ``max_items`` cap applied."""
1516 items = StepRunner._parallel_fanout_items(step, state, ctx)
1517 if items is None:
1518 return None
1519 max_items_raw = step.get("max_items")
1520 if max_items_raw is not None:
1521 items = items[: _resolve_int(max_items_raw, ctx, len(items))]
1522 return items
1524 @staticmethod
1525 def _foreach_item_list(
1526 step: dict, state: PipelineState, ctx: dict
1527 ) -> list:
1528 """The foreach item list: resolved, wrapped, empty-checked, capped.
1530 A missing source, an unresolvable source, and an empty list are hard
1531 ``VALIDATION_ERROR`` failures (ADR-0012). ``max_iterations`` caps the
1532 list.
1533 """
1534 step_id = step.get("id", "?")
1535 source = StepRunner._declared_item_source(step)
1536 if source is None:
1537 raise RunHardFailure(
1538 "VALIDATION_ERROR",
1539 f"foreach step '{step_id}' declares no item source "
1540 "('input' or 'over') — no silent [None] fallback (ADR-0012)",
1541 {"step_id": step_id},
1542 )
1543 resolved_items = StepRunner._resolve_item_source(source, state, ctx)
1544 if resolved_items is None:
1545 raise StepRunner._unresolvable_item_source_failure(step_id, source)
1546 items: list = (
1547 resolved_items
1548 if isinstance(resolved_items, list)
1549 else [resolved_items]
1550 )
1551 if not items:
1552 raise StepRunner._empty_item_source_failure(step_id, source)
1553 max_iter_raw = step.get("max_iterations")
1554 if max_iter_raw is not None:
1555 items = items[: _resolve_int(max_iter_raw, ctx, len(items))]
1556 return items
1558 def ran_agent(self, step: dict, state: PipelineState) -> bool:
1559 """Whether a step's run invokes its agent.
1561 False for agent-less steps, a ``skip_llm_if`` passthrough, and a
1562 parallel fan-out (per-item agent calls storing a result envelope).
1563 Channel routing is restricted to actual agent outputs, so these paths
1564 are excluded from inverse projection.
1565 """
1566 if self.resolve_agent(step) is None:
1567 return False
1568 ctx = self._make_ctx(state)
1569 if self._should_skip_llm(step, ctx):
1570 return False
1571 return self._parallel_fanout_items(step, state, ctx) is None
1573 def _make_ctx(self, state: dict, **extras: Any) -> dict:
1574 """Rakentaa template-kontekstin: state + run_context + extras."""
1575 ctx: dict = {}
1576 ctx.update(state)
1577 ctx.update(self._run_context)
1578 ctx.update(extras)
1579 return ctx
1581 @staticmethod
1582 def _read_state(
1583 state: PipelineState, working_set: ChannelWorkingSet | None
1584 ) -> PipelineState:
1585 """Overlay the invocation working set's read view onto the local state.
1587 The working set carries the committed snapshot plus this invocation's
1588 pending writes; the local state contributes keys the snapshot does not
1589 hold (item context, body step results). With no working set the raw
1590 state is the read view, preserving legacy behaviour exactly. The
1591 engine-local iteration keys are re-asserted after the overlay so a
1592 declared channel with the same name cannot shadow them.
1593 """
1594 if working_set is None:
1595 return state
1596 resolved = {**state, **working_set.read_view()}
1597 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1598 if key in state:
1599 resolved[key] = state[key]
1600 return resolved
1602 def _run_body_step(
1603 self,
1604 step: dict,
1605 state: PipelineState,
1606 working_set: ChannelWorkingSet | None,
1607 ) -> PipelineState:
1608 """Run one nested body step and stage its declared channel writes.
1610 The two-argument call preserves the legacy recursion signature for the
1611 no-channels path (and for callers that monkey-patch ``run_step``).
1612 """
1613 if working_set is None:
1614 return self.run_step(step, state)
1615 result = self.run_step(step, state, working_set)
1616 self._stage_body_channel_writes(step, state, result, working_set)
1617 return result
1619 def _stage_body_channel_writes(
1620 self,
1621 step: dict,
1622 pre_state: PipelineState,
1623 result_state: PipelineState,
1624 working_set: ChannelWorkingSet,
1625 ) -> None:
1626 """Stage a body step's declared channel writes into the working set.
1628 The router early-outs for agent-less steps (builtins, call bind/emit);
1629 a step carrying an agent routes its declared writes.
1630 """
1631 if not self._channel_names:
1632 return
1633 writes = _route_declared_channels(
1634 step,
1635 self._read_state(pre_state, working_set),
1636 result_state,
1637 self,
1638 self._channel_names,
1639 )
1640 for name, value in writes.items():
1641 working_set.stage(name, value)
1643 def _resolve_step_runtime(
1644 self,
1645 step: dict,
1646 state: dict,
1647 ctx: dict,
1648 *,
1649 input_override: Any = _UNSET,
1650 extra_params: dict | None = None,
1651 ) -> dict:
1652 """
1653 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
1654 templaattien osalta state + run_context -kontekstia vasten.
1656 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
1657 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
1658 """
1659 resolved: dict = dict(step)
1661 if input_override is not _UNSET:
1662 resolved["input"] = input_override
1663 else:
1664 input_spec = step.get("input")
1665 if input_spec is not None:
1666 resolved["input"] = _resolve_input(input_spec, state, ctx)
1668 params_spec = step.get("params")
1669 if params_spec or extra_params:
1670 base = _resolve_template(params_spec, ctx) if params_spec else {}
1671 if not isinstance(base, dict):
1672 base = {}
1673 if extra_params:
1674 base = {**base, **extra_params}
1675 resolved["params"] = base
1677 return resolved
1679 def _attach_communication_context(
1680 self, resolved_step: dict, read_state: PipelineState
1681 ) -> dict:
1682 """Attach a step's projected COMMUNICATION context when declared.
1684 The projection is resolved from the invocation read overlay (the
1685 committed snapshot plus this invocation's pending writes) and carries
1686 only the step's declared input channels — never whole state. A step
1687 without a read declaration is returned unchanged.
1688 """
1689 communication = resolve_communication_context(resolved_step, read_state)
1690 if not communication:
1691 return resolved_step
1692 return {**resolved_step, COMMUNICATION_CONTEXT_KEY: communication}
1694 def _attach_resolved_tools(
1695 self, resolved_step: dict, resolved_agent: dict
1696 ) -> dict:
1697 """Attach the agent's declared `tools` as resolved `Tool` instances.
1699 Resolves the agent's declared tool names through `ToolResolver` into
1700 ordered `Tool` objects under `RESOLVED_TOOLS_KEY`. An agent without
1701 tools, or a runner without a resolver, is returned unchanged.
1702 """
1703 tools = (resolved_agent.get("spec") or {}).get("tools") or []
1704 if not tools or self._tool_resolver is None:
1705 return resolved_step
1706 return {
1707 **resolved_step,
1708 RESOLVED_TOOLS_KEY: self._resolve_agent_tools(
1709 resolved_agent, list(tools)
1710 ),
1711 }
1713 def _resolve_agent_tools(
1714 self, resolved_agent: dict, tools: list[str]
1715 ) -> list:
1716 """Resolve an agent's declared tools, enforcing the instruction guard.
1718 The low-level resolver stays permissive; the policy lives here
1719 (ADR-0016). A declared name the resolver cannot resolve, or a
1720 resolved tool that exposes no non-empty description, is translated
1721 into the engine's build-time `AGENT_INSTRUCTION_MISSING` hard failure.
1722 A dual-source name keeps its own `BUILTIN_NAME_COLLISION` failure.
1723 """
1724 return [self._resolve_agent_tool(resolved_agent, name) for name in tools]
1726 def _resolve_agent_tool(self, resolved_agent: dict, name: str):
1727 try:
1728 tool = self._tool_resolver.resolve([name])[0]
1729 except ToolCollisionError as exc:
1730 raise RunHardFailure(
1731 "BUILTIN_NAME_COLLISION",
1732 f"tool {name!r} exists in both the platform registry and the "
1733 "StepFunction register",
1734 {"tool": name},
1735 ) from exc
1736 except ToolRegistryError as exc:
1737 raise self._instruction_missing_failure(
1738 resolved_agent, name, str(exc)
1739 ) from exc
1740 if not _tool_has_description(tool):
1741 raise self._instruction_missing_failure(
1742 resolved_agent, name, "the resolved tool exposes no description"
1743 )
1744 return tool
1746 @staticmethod
1747 def _instruction_missing_failure(
1748 resolved_agent: dict, tool_name: str, detail: str
1749 ) -> RunHardFailure:
1750 agent_name = (resolved_agent.get("metadata") or {}).get("name")
1751 return RunHardFailure(
1752 AGENT_INSTRUCTION_MISSING,
1753 f"Agent {agent_name!r} declares tool {tool_name!r} with no usage "
1754 f"instruction: {detail}",
1755 {"agent": agent_name, "tool": tool_name},
1756 )
1758 @staticmethod
1759 def _dummy_invoke(
1760 resolved_spec: dict, step: dict, state: PipelineState
1761 ) -> PipelineState:
1762 """Placeholder — ei kutsu oikeaa LLM:ää."""
1763 step_id = step.get("id", "?")
1764 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
1765 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
1767 @staticmethod
1768 def _merge_errors(base: list, additions: list) -> list:
1769 """Return a new list containing all items from *base* plus any items from
1770 *additions* that are not already present. Order is preserved and no
1771 in-place mutation occurs — safe to call from any context."""
1772 if not additions:
1773 return base
1774 merged = list(base)
1775 for err in additions:
1776 if err not in merged:
1777 merged.append(err)
1778 return merged
1780 @staticmethod
1781 def _wrap_hard_failure_gate(invoke_fn: Any) -> Any:
1782 """Kietoo invokerin: jokaiseen agent-step-tulokseen sovelletaan
1783 kova-virhe-portti (RunHardFailure details.error_code-perusteella).
1784 Suppein funnel joka näkee KAIKKI agent-tulokset: sequential,
1785 parallel (single + fanout), foreach (body + agent-polku) ja while."""
1786 def invoke_with_gate(
1787 resolved_spec: dict, step: dict, state: PipelineState
1788 ) -> PipelineState:
1789 result = invoke_fn(resolved_spec, step, state)
1790 step_id = step.get("id", "?")
1791 StepRunner._raise_on_hard_error(result.get(step_id), step_id)
1792 return result
1793 return invoke_with_gate
1795 @staticmethod
1796 def _is_hard_fail(details: dict) -> bool:
1797 """Kova virhe vain kun: error_code on HARD_ERROR_CODESissa
1798 (TRUNCATION_ERROR, TOOL_FAILURE, AGENT_ERROR) tai VALIDATION_ERROR
1799 kantaa kontekstiylityksen signatuurin (assembled_tokens +
1800 context_budget — _llm._context_overflow_result).
1801 Generic-VALIDATION_ERROR ei kaada ajoa. status:error-envelope kaatuu
1802 _raise_on_error_envelope-polulla ennen tätä tarkistusta."""
1803 code = details.get("error_code")
1804 if code in RunHardFailure.HARD_ERROR_CODES:
1805 return True
1806 if code == "VALIDATION_ERROR":
1807 return {"assembled_tokens", "context_budget"} <= details.keys()
1808 return False
1810 @staticmethod
1811 def _raise_on_hard_error(step_result: Any, step_id: str) -> None:
1812 """Nosta RunHardFailure jos step-tulos kantaa kovan virheen
1813 (_is_hard_fail) tai on status:error-envelope; stepin details
1814 kulkeutuu poikkeuksen mukana."""
1815 if not isinstance(step_result, dict):
1816 return
1817 if is_error_envelope(step_result):
1818 StepRunner._raise_on_error_envelope(step_result, step_id)
1819 return
1820 details = step_result.get("details")
1821 if not isinstance(details, dict):
1822 return
1823 if not StepRunner._is_hard_fail(details):
1824 return
1825 code = details.get("error_code")
1826 err = step_result.get("error", "")
1827 message = err if isinstance(err, str) and err else f"{code}: step '{step_id}' hard failure"
1828 log.error("[hard-failure] %s: %s", code, message)
1829 raise RunHardFailure(code, message, details)
1831 @staticmethod
1832 def _raise_on_error_envelope(step_result: dict, step_id: str) -> None:
1833 """Nosta RunHardFailure status:error-envelopesta; envelopen error
1834 (dict) kulkeutuu details-attribuuttina, muuten minimi details."""
1835 envelope_error = step_result.get("error")
1836 if isinstance(envelope_error, dict):
1837 details = envelope_error
1838 message = envelope_error.get("message")
1839 else:
1840 details = {"error_code": ENVELOPE_ERROR_CODE, "step_id": step_id}
1841 message = None
1842 if not isinstance(message, str) or not message:
1843 message = f"{ENVELOPE_ERROR_CODE}: step '{step_id}' envelope error"
1844 raise RunHardFailure(ENVELOPE_ERROR_CODE, message, details)
1846 @staticmethod
1847 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
1848 """
1849 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
1850 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
1851 epäonnistuneeksi eikä hiljennä virhettä.
1853 Kaikki virhetyypit propagoidaan:
1854 - budget:max_llm_calls / budget:run_timeout → budjettiloki
1855 - parse_error:*, timeout:*, ja muut → error-loki
1856 """
1857 step_result = result.get(step_id)
1858 if not isinstance(step_result, dict):
1859 return result
1860 err = step_result.get("error", "")
1861 if not isinstance(err, str) or not err:
1862 return result
1863 errors = list(result.get("errors") or [])
1864 message = f"{step_id}: {err}"
1865 if message not in errors:
1866 errors.append(message)
1867 if err.startswith("budget:"):
1868 log.warning("[budget] %s", message)
1869 else:
1870 log.error("[step-error] %s", message)
1871 return {**result, "errors": errors}
1873 def resolve_agent(self, step: dict) -> dict | None:
1874 """Resolve the agent/instance a step runs, or None for agent-less steps."""
1875 agent_name = step.get("agent")
1876 if not agent_name:
1877 return None
1878 return self._resolver.resolve(agent_name)
1880 def run_step(
1881 self,
1882 step: dict,
1883 state: PipelineState,
1884 working_set: ChannelWorkingSet | None = None,
1885 ) -> PipelineState:
1886 step_id = step.get("id", "?")
1887 read_state = self._read_state(state, working_set)
1888 ctx = self._make_ctx(read_state)
1889 step_type = step.get("type", "sequential")
1891 # Evaluate condition — if false, skip step. A `while` never reaches
1892 # this point as a raw step: build-time composition rewrites it into a
1893 # native cycle placeholder whose head owns the entry-condition skip.
1894 cond = step.get("condition")
1895 if cond is not None and not _eval_condition(cond, ctx):
1896 log.debug("[step] %s skipped (condition false)", step_id)
1897 return state
1899 if step_type == "builtin":
1900 component = step.get("component", "passthrough")
1901 log.debug("[step] %s builtin:%s", step_id, component)
1902 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
1903 # template-arvot (esim. "{{ budget.max_parallel }}")
1904 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1905 # ADR-0037 D2: resolve the component against the platform builtin
1906 # register FIRST, then the kind: StepFunction registry. A name in
1907 # both is a hard error (fail-fast).
1908 is_builtin = self._builtins.has(component)
1909 is_step_function = (
1910 self._step_functions is not None
1911 and self._step_functions.has(component)
1912 )
1913 if is_builtin and is_step_function:
1914 raise RunHardFailure(
1915 "BUILTIN_NAME_COLLISION",
1916 f"component '{component}' is present in both the builtin "
1917 "register and the StepFunction registry (ADR-0037 — "
1918 "fail-fast)",
1919 {"component": component},
1920 )
1921 if is_step_function:
1922 return self._step_functions.execute(component, resolved_step, state)
1923 builtin_state = self._builtins.execute(component, resolved_step, state)
1924 self._raise_on_hard_error(builtin_state.get(step_id), step_id)
1925 return builtin_state
1927 # skip_llm_if (ENG-12, Q4): when the step's skip_llm_if evaluates truthy,
1928 # do NOT invoke the LLM (no inference budget consumed) and set the step
1929 # result to a deterministic passthrough of the step's resolved input.
1930 if (
1931 step_type in ("sequential", "parallel")
1932 and self._should_skip_llm(step, ctx)
1933 ):
1934 return self._skip_llm_passthrough(step, state, ctx, read_state)
1936 if step_type == "parallel":
1937 # Fan-out: if input resolves to a list, call agent once per item
1938 items = self._parallel_fanout_items(step, read_state, ctx)
1939 if items is not None:
1940 return self._run_parallel_fanout(
1941 step, state, items, ctx, working_set
1942 )
1943 # Fallback: single agent call
1944 agent_name = step.get("agent")
1945 if not agent_name:
1946 log.warning("[step] %s has no agent — skipping", step_id)
1947 return state
1948 resolved = self._resolver.resolve(agent_name)
1949 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
1950 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1951 resolved_step = self._attach_communication_context(
1952 resolved_step, read_state
1953 )
1954 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
1955 result = self._invoke_agent(resolved, resolved_step, state)
1956 return self._propagate_step_errors(result, step_id)
1958 if step_type == "sequential":
1959 agent_name = step.get("agent")
1960 if not agent_name:
1961 log.warning("[step] %s has no agent — skipping", step_id)
1962 return state
1963 resolved = self._resolver.resolve(agent_name)
1964 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
1965 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1966 resolved_step = self._attach_communication_context(
1967 resolved_step, read_state
1968 )
1969 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
1970 result = self._invoke_agent(resolved, resolved_step, state)
1971 return self._propagate_step_errors(result, step_id)
1973 if step_type == "foreach":
1974 return self._run_foreach(step, state, working_set)
1976 if step_type == "call":
1977 # Build-time composition (ADR-0037 D4) rewrites every call step into
1978 # a native subgraph node, so a call reaching the runtime means the
1979 # builder skipped a composition path it was responsible for.
1980 raise RunHardFailure(
1981 "VALIDATION_ERROR",
1982 f"call step '{step_id}' reached runtime without build-time "
1983 "expansion (ADR-0037 D4) — not yet supported",
1984 {"step_id": step_id},
1985 )
1987 if step_type == _SUBGRAPH_STEP_TYPE:
1988 # Build-time composition (ADR-0037 D4) embeds a compiled function
1989 # subgraph as a native node. At the top level ``_assemble_graph``
1990 # wires it through ``_make_subgraph_node``; when the call sits in an
1991 # imperative foreach/while body the placeholder reaches ``run_step``
1992 # instead, so invoke the embedded graph here. The top-of-method
1993 # condition check already handled a false condition (returned the
1994 # input state unchanged), matching the node wrapper's skip.
1995 compiled = step.get(_SUBGRAPH_GRAPH_KEY)
1996 if compiled is None:
1997 raise RunHardFailure(
1998 "VALIDATION_ERROR",
1999 f"subgraph step '{step_id}' has no compiled graph",
2000 {"step_id": step_id},
2001 )
2002 config = _while_invoke_config(step)
2003 if working_set is not None:
2004 # Channel path: the compiled subgraph shares the parent's
2005 # Core-namespaced schema, while a loop body carries the flat
2006 # committed snapshot. Re-wrap on entry, flatten on exit. No
2007 # channel reducer is re-applied here — the subgraph's own
2008 # commits already produced the authoritative values.
2009 sub_result = compiled.invoke({_CORE: read_state}, config)
2010 committed = _flatten_channel_state(sub_result)
2011 return {**state, **committed}
2012 return compiled.invoke(state, config)
2014 if step_type == "spawn":
2015 return self._run_spawn(step, state, working_set)
2017 raise RunHardFailure(
2018 "VALIDATION_ERROR",
2019 f"Unknown step type '{step_type}' in step '{step_id}'",
2020 {"step_id": step_id, "step_type": step_type},
2021 )
2023 def _run_parallel_fanout(
2024 self,
2025 step: dict,
2026 state: PipelineState,
2027 items: list,
2028 ctx: dict,
2029 working_set: ChannelWorkingSet | None = None,
2030 ) -> PipelineState:
2031 """
2032 Fan-out: call agent once per item, concurrently.
2033 Results stored as state[step_id] = {"results": [item_result, ...]}.
2034 Each item is available as state["item"] inside the invoke call.
2035 """
2036 step_id = step.get("id", "?")
2037 agent_name = step.get("agent")
2038 if not agent_name:
2039 log.warning("[step] %s parallel fanout: no agent", step_id)
2040 return {**state, step_id: {"results": []}}
2042 resolved = self._resolver.resolve(agent_name)
2044 # Apply max_items cap
2045 max_items_raw = step.get("max_items")
2046 if max_items_raw is not None:
2047 max_items = _resolve_int(max_items_raw, ctx, len(items))
2048 items = items[:max_items]
2050 # Concurrency limit
2051 max_conc_raw = step.get("max_concurrent")
2052 if max_conc_raw is not None:
2053 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2054 else:
2055 max_conc = len(items) or 1
2056 max_conc = max(1, min(max_conc, len(items) or 1))
2058 log.debug(
2059 "[step] %s parallel fanout: %d items, %d concurrent",
2060 step_id, len(items), max_conc,
2061 )
2063 read_state = self._read_state(state, working_set)
2065 def call_one(item: Any) -> tuple[Any, dict | None, dict | None, list]:
2066 item_state = {**read_state, "_item": item, "item": item}
2067 item_ctx = self._make_ctx(item_state)
2068 return self._invoke_parallel_item(
2069 step, resolved, item, item_state, item_ctx
2070 )
2072 # Pre-allocate results to preserve input order regardless of completion
2073 # order (ENG-07), matching the concurrent foreach body-polku.
2074 results: list = [None] * len(items)
2075 item_metrics: dict = {}
2076 item_tool_meta: dict = {}
2077 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2078 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
2079 for future in as_completed(futures):
2080 idx = futures[future]
2081 try:
2082 item_result, item_metrics_map, item_tool_meta_map, *_ = (
2083 future.result()
2084 )
2085 results[idx] = item_result
2086 if isinstance(item_metrics_map, dict):
2087 item_metrics.update(item_metrics_map)
2088 if isinstance(item_tool_meta_map, dict):
2089 item_tool_meta.update(item_tool_meta_map)
2090 except SkillLoadError:
2091 raise
2092 except RunHardFailure:
2093 pool.shutdown(wait=False, cancel_futures=True)
2094 raise
2095 except Exception as exc: # noqa: BLE001
2096 log.error("[step] %s item failed: %s", step_id, exc)
2097 results[idx] = {"error": str(exc)}
2099 self._stage_fanout_channel_writes(step, results, working_set)
2101 final: PipelineState = {**state, step_id: {"results": results}}
2102 if item_metrics:
2103 final = {
2104 **final,
2105 "step_context_metrics": _merge_side_channel(
2106 state.get("step_context_metrics"), item_metrics
2107 ),
2108 }
2109 if item_tool_meta:
2110 final = {
2111 **final,
2112 "step_tool_meta": _merge_side_channel(
2113 state.get("step_tool_meta"), item_tool_meta
2114 ),
2115 }
2116 return final
2118 def _invoke_parallel_item(
2119 self,
2120 step: dict,
2121 resolved: dict,
2122 item: Any,
2123 item_state: PipelineState,
2124 item_ctx: dict,
2125 ) -> tuple[Any, dict | None, dict | None, list]:
2126 """Run one parallel fan-out item's agent call.
2128 Shared by the in-node thread-pool fan-out and the native ``Send``
2129 worker, so both paths resolve input, communication context and tools
2130 identically. Returns ``(step_result, metrics, tool_meta, errors)``.
2131 """
2132 step_id = step.get("id", "?")
2133 resolved_step = self._resolve_step_runtime(
2134 step, item_state, item_ctx,
2135 input_override=item,
2136 )
2137 resolved_step = self._attach_communication_context(
2138 resolved_step, item_state
2139 )
2140 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2141 result_state = self._invoke_agent(resolved, resolved_step, item_state)
2142 return (
2143 result_state.get(step_id),
2144 result_state.get("step_context_metrics"),
2145 result_state.get("step_tool_meta"),
2146 list(result_state.get("errors") or []),
2147 )
2149 def _stage_fanout_channel_writes(
2150 self,
2151 step: dict,
2152 item_outputs: list,
2153 working_set: ChannelWorkingSet | None,
2154 ) -> None:
2155 """Stage each fan-out item's declared channel writes in item order.
2157 Runs on the main thread after the item pool completes, so the shared
2158 working set is never mutated from a worker thread. The fan-out
2159 envelope itself is never routed; only each item's own agent output is.
2160 """
2161 if working_set is None or not self._channel_names:
2162 return
2163 for output in item_outputs:
2164 writes = _declared_output_channel_writes(
2165 step, output, self._channel_names
2166 )
2167 for name, value in writes.items():
2168 working_set.stage(name, value)
2170 def _run_foreach(
2171 self,
2172 step: dict,
2173 state: PipelineState,
2174 working_set: ChannelWorkingSet | None = None,
2175 ) -> PipelineState:
2176 """
2177 Foreach: iterate over input list, run body steps per item.
2178 Input from step.input (dot-path or template) or step.over (state key).
2180 When max_concurrent > 1, body iterations run concurrently via
2181 ThreadPoolExecutor — each item gets its own worker thread. The
2182 max_concurrent value sets the thread pool size (mirrors parallel
2183 fan-out semantics). Results list preserves original item order.
2185 max_output budget in concurrent mode: a stop flag prevents new workers
2186 from starting after the budget is exhausted. Workers already executing
2187 when the flag is set will complete naturally — at most max_concurrent
2188 extra items may run past the limit. This is accepted behaviour; the
2189 sequential early-exit guarantee cannot be reproduced without sacrificing
2190 true concurrency.
2192 The invocation working set is threaded into every body step so reads
2193 resolve over the committed snapshot plus this invocation's pending
2194 writes; the returned state is still built from the raw ``state``.
2195 """
2196 step_id = step.get("id", "?")
2197 read_state = self._read_state(state, working_set)
2198 ctx = self._make_ctx(read_state)
2200 items = self._foreach_item_list(step, read_state, ctx)
2202 # max_output cap — stop once N iterations have produced a successful
2203 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
2204 max_output_raw = step.get("max_output")
2205 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
2207 # max_concurrent — when > 1, enables parallel body execution
2208 max_conc_raw = step.get("max_concurrent")
2209 if max_conc_raw is not None:
2210 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2211 max_conc = max(1, min(max_conc, len(items) or 1))
2212 else:
2213 max_conc = 1
2215 body = step.get("body") or []
2216 agent_name = step.get("agent")
2218 if max_conc <= 1:
2219 raise RunHardFailure(
2220 "VALIDATION_ERROR",
2221 f"foreach step '{step_id}' reached runtime without build-time "
2222 "composition (a sequential foreach is compiled into a native "
2223 "cycle) — internal invariant violated",
2224 {"step_id": step_id},
2225 )
2226 return self._run_foreach_concurrent(
2227 step_id, step, state, items, body, agent_name, max_output, max_conc,
2228 working_set,
2229 )
2231 @staticmethod
2232 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
2233 """body-polku per-iteration result (r.67): the declared body-output
2234 step's result; else the iteration's `publish` result; else None."""
2235 if declared_output:
2236 return iter_state.get(declared_output)
2237 return iter_state.get("publish")
2239 @staticmethod
2240 def _body_item_passed(item: Any) -> bool:
2241 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
2242 `passed` bool; an item lacking such a key is not counted as failed."""
2243 if isinstance(item, dict):
2244 verdict = item.get("verdict")
2245 if verdict == "FAIL":
2246 return False
2247 if verdict == "PASS":
2248 return True
2249 passed = item.get("passed")
2250 if isinstance(passed, bool):
2251 return passed
2252 return True
2254 @staticmethod
2255 def _aggregate_body_results(item_results: list) -> dict:
2256 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
2257 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
2258 return {
2259 "results": item_results,
2260 "passed": not failing,
2261 "failing": failing,
2262 "failed": bool(failing),
2263 }
2265 @staticmethod
2266 def _merge_item_working_sets(
2267 working_set: ChannelWorkingSet | None,
2268 item_working_sets: list[ChannelWorkingSet | None],
2269 ) -> None:
2270 """Stage each concurrent item's writes onto the shared set in item order."""
2271 if working_set is None:
2272 return
2273 for item_working_set in item_working_sets:
2274 if item_working_set is None:
2275 continue
2276 for name, value in item_working_set.pending_writes():
2277 working_set.stage(name, value)
2279 def _run_foreach_concurrent(
2280 self,
2281 step_id: str,
2282 step: dict,
2283 state: PipelineState,
2284 items: list,
2285 body: list,
2286 agent_name: str | None,
2287 max_output: int,
2288 max_conc: int,
2289 working_set: ChannelWorkingSet | None = None,
2290 ) -> PipelineState:
2291 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2292 log.info(
2293 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2294 step_id, len(items), max_conc,
2295 )
2297 # Pre-allocate results list to preserve item order across threads.
2298 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2299 # legitimate None results (body produced no output) are kept.
2300 body_polku = bool(body)
2301 declared_output = step.get("output")
2302 _SKIPPED = object()
2303 results: list = [_SKIPPED] * len(items)
2304 item_working_sets: list[ChannelWorkingSet | None] = [None] * len(items)
2305 lock = threading.Lock()
2306 produced_outputs = [0] # mutable counter shared across threads
2307 stop_flag = threading.Event() # set when max_output budget is exhausted
2308 item_errors: list = [] # budget errors collected from worker threads
2309 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2310 accumulated_tool_meta: dict = {} # per-item tool metadata from worker threads
2312 def run_item(idx: int, item: Any) -> None:
2313 nonlocal item_errors
2314 # Honour stop flag — skip if budget already exhausted before we start
2315 if max_output and stop_flag.is_set():
2316 return
2318 item_working_set = (
2319 working_set.fork() if working_set is not None else None
2320 )
2321 item_working_sets[idx] = item_working_set
2322 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2324 if body_polku:
2325 for sub_step in body:
2326 iter_state = self._run_body_step(
2327 sub_step, iter_state, item_working_set
2328 )
2329 # Collect budget errors from this iteration (thread-safe merge)
2330 new_errs = iter_state.get("errors") or []
2331 if new_errs:
2332 with lock:
2333 item_errors = self._merge_errors(item_errors, new_errs)
2334 item_metrics_map = iter_state.get("step_context_metrics")
2335 if isinstance(item_metrics_map, dict):
2336 with lock:
2337 accumulated_metrics.update(item_metrics_map)
2338 item_tool_meta_map = iter_state.get("step_tool_meta")
2339 if isinstance(item_tool_meta_map, dict):
2340 with lock:
2341 accumulated_tool_meta.update(item_tool_meta_map)
2342 publish_out = iter_state.get("publish")
2343 results[idx] = self._body_item_result(iter_state, declared_output)
2344 if isinstance(publish_out, dict) and publish_out.get("path"):
2345 with lock:
2346 produced_outputs[0] += 1
2347 if max_output and produced_outputs[0] >= max_output:
2348 stop_flag.set()
2349 elif agent_name:
2350 resolved = self._resolver.resolve(agent_name)
2351 item_read_state = self._read_state(iter_state, item_working_set)
2352 item_ctx = self._make_ctx(item_read_state)
2353 resolved_step = self._resolve_step_runtime(
2354 step, item_read_state, item_ctx
2355 )
2356 resolved_step = self._attach_communication_context(
2357 resolved_step, item_read_state
2358 )
2359 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2360 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2361 results[idx] = iter_state.get(step_id)
2362 item_metrics_map = iter_state.get("step_context_metrics")
2363 if isinstance(item_metrics_map, dict):
2364 with lock:
2365 accumulated_metrics.update(item_metrics_map)
2366 item_tool_meta_map = iter_state.get("step_tool_meta")
2367 if isinstance(item_tool_meta_map, dict):
2368 with lock:
2369 accumulated_tool_meta.update(item_tool_meta_map)
2371 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2372 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2373 for future in as_completed(futures):
2374 try:
2375 future.result()
2376 except SkillLoadError:
2377 raise
2378 except RunHardFailure:
2379 pool.shutdown(wait=False, cancel_futures=True)
2380 raise
2381 except Exception as exc: # noqa: BLE001
2382 idx = futures[future]
2383 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2384 results[idx] = {"error": str(exc)}
2386 self._merge_item_working_sets(working_set, item_working_sets)
2388 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2389 ordered = [r for r in results if r is not _SKIPPED]
2390 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
2391 final: PipelineState = {**state, step_id: step_result}
2392 if accumulated_metrics:
2393 final = {
2394 **final,
2395 "step_context_metrics": _merge_side_channel(
2396 state.get("step_context_metrics"), accumulated_metrics
2397 ),
2398 }
2399 if accumulated_tool_meta:
2400 final = {
2401 **final,
2402 "step_tool_meta": _merge_side_channel(
2403 state.get("step_tool_meta"), accumulated_tool_meta
2404 ),
2405 }
2406 if item_errors:
2407 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2408 return final
2410 def _run_spawn(
2411 self,
2412 step: dict,
2413 state: PipelineState,
2414 working_set: ChannelWorkingSet | None = None,
2415 ) -> PipelineState:
2416 """Run a spawn step as a native dynamic ``Send`` fan-out subgraph.
2418 The ``agent`` runs once as director at the tree root and may emit a
2419 child plan at ``envelope.content.children``; a shared child-worker node
2420 re-runs the ``child`` role (default = ``agent``) for each child and
2421 re-``Send``s itself for deeper levels. Depth and fan are bounded by the
2422 spawn_policy, and the whole tree draws from one shared SpawnBudget.
2423 """
2424 step_id = step.get("id", "?")
2425 director = step.get("agent")
2426 if not director:
2427 log.warning("[step] %s spawn has no agent — skipping", step_id)
2428 return state
2429 read_state = self._read_state(state, working_set)
2430 ctx = self._make_ctx(read_state)
2431 policy = step.get("spawn_policy") or {}
2432 max_depth = _resolve_int(
2433 policy.get("max_depth"), ctx, _SPAWN_DEFAULT_MAX_DEPTH
2434 )
2435 max_spawn_per_node = _resolve_int(
2436 policy.get("max_spawn_per_node"), ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
2437 )
2438 max_calls_raw = policy.get("max_agent_calls")
2439 budget = SpawnBudget(
2440 _SPAWN_DEFAULT_MAX_AGENT_CALLS
2441 if max_calls_raw is None
2442 else _resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS)
2443 )
2444 run_max_parallel = int(
2445 (self._run_context.get("budget") or {}).get(
2446 "max_parallel", DEFAULT_MAX_PARALLEL
2447 )
2448 )
2449 max_concurrent_raw = policy.get("max_concurrent_children")
2450 max_concurrent_children = (
2451 run_max_parallel
2452 if max_concurrent_raw is None
2453 else _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
2454 )
2455 child_role = step.get("child") or director
2457 outcome = self._run_native_spawn(
2458 step=step,
2459 step_id=step_id,
2460 director=director,
2461 child_role=child_role,
2462 base_state=read_state,
2463 base_ctx=ctx,
2464 budget=budget,
2465 max_depth=max_depth,
2466 max_spawn_per_node=max_spawn_per_node,
2467 max_concurrent_children=max_concurrent_children,
2468 )
2469 envelope = outcome["envelope"]
2470 final: PipelineState = {**state, step_id: envelope}
2471 if outcome["metrics"]:
2472 final = {
2473 **final,
2474 "step_context_metrics": _merge_side_channel(
2475 state.get("step_context_metrics"), outcome["metrics"]
2476 ),
2477 }
2478 if outcome["tool_meta"]:
2479 final = {
2480 **final,
2481 "step_tool_meta": _merge_side_channel(
2482 state.get("step_tool_meta"), outcome["tool_meta"]
2483 ),
2484 }
2485 self._stage_spawn_channel_writes(step, envelope["result"], working_set)
2486 return final
2488 def _run_native_spawn(
2489 self,
2490 *,
2491 step: dict,
2492 step_id: str,
2493 director: str,
2494 child_role: str,
2495 base_state: PipelineState,
2496 base_ctx: dict,
2497 budget: SpawnBudget,
2498 max_depth: int,
2499 max_spawn_per_node: int | None,
2500 max_concurrent_children: int,
2501 ) -> dict:
2502 """Compile and invoke the native spawn tree, returning its outcome.
2504 The graph is a director node that seeds one ``Send`` per child to a
2505 shared worker, whose worker re-``Send``s itself for deeper levels. An
2506 outstanding-descendants barrier (``outstanding`` counter plus a ``check``
2507 node) makes the ``join`` run exactly once for uneven-depth trees; the
2508 join assembles the result envelope by path key in plan order.
2509 """
2510 from contextlib import nullcontext
2511 from typing import TypedDict
2513 from langgraph.graph import END, START, StateGraph
2514 from langgraph.types import Command, Send
2516 tracer = self._tracer
2517 parent_obs = (
2518 tracer.open_parent_observation(
2519 name=f"spawn:{step_id}",
2520 input={"node_id": step_id, "role": director, "depth": 0},
2521 )
2522 if tracer is not None
2523 else None
2524 )
2525 root_obs_id = getattr(parent_obs, "id", None)
2526 try:
2527 child_gates: dict[str, threading.Semaphore] = {}
2528 child_gates_lock = threading.Lock()
2530 def child_gate(parent_id: str) -> threading.Semaphore:
2531 """The concurrency gate for one spawning node's direct children.
2533 ``max_concurrent_children`` bounds a single spawn node's fan-out:
2534 each spawning node owns its own gate, so a shallow node's cap
2535 never throttles a deeper sibling's subtree.
2536 """
2537 with child_gates_lock:
2538 gate = child_gates.get(parent_id)
2539 if gate is None:
2540 gate = threading.Semaphore(max(1, max_concurrent_children))
2541 child_gates[parent_id] = gate
2542 return gate
2544 def run_node(
2545 *,
2546 role: str,
2547 node_step: dict,
2548 input_override: Any,
2549 node_id: str,
2550 depth: int,
2551 parent_id: str | None = None,
2552 ) -> dict:
2553 resolved_role = self._resolver.resolve(role)
2554 resolved_step = self._resolve_step_runtime(
2555 node_step, base_state, base_ctx, input_override=input_override
2556 )
2557 resolved_step = self._attach_communication_context(
2558 resolved_step, base_state
2559 )
2560 resolved_step = self._attach_resolved_tools(
2561 resolved_step, resolved_role
2562 )
2563 if not budget.consume():
2564 raise RunHardFailure(
2565 "SPAWN_BUDGET_EXHAUSTED",
2566 f"spawn tree budget exhausted at node '{node_id}'",
2567 {"node_id": node_id, "depth": depth},
2568 )
2569 gate = child_gate(parent_id) if parent_id is not None else nullcontext()
2570 with gate:
2571 return self._invoke_agent(
2572 resolved_role, resolved_step, base_state
2573 )
2575 def build_record(
2576 *,
2577 node_id: str,
2578 path: tuple,
2579 parent_path: tuple | None,
2580 depth: int,
2581 this_result: Any,
2582 result_state: dict,
2583 children: list,
2584 ) -> tuple[dict, list]:
2585 exhausted = False
2586 plan: list = []
2587 if children:
2588 if depth + 1 > max_depth:
2589 exhausted = True
2590 else:
2591 plan = [
2592 {
2593 "path": path + (index,),
2594 "depth": depth + 1,
2595 "child": child,
2596 "parent_id": node_id,
2597 "index": index,
2598 }
2599 for index, child in enumerate(children)
2600 ]
2601 record = {
2602 "path": path,
2603 "parent_path": parent_path,
2604 "child_id": node_id,
2605 "result": this_result,
2606 "self_passed": not self._spawn_is_soft_fail(this_result),
2607 "exhausted": exhausted,
2608 "metrics": dict(result_state.get("step_context_metrics") or {}),
2609 "tool_meta": dict(result_state.get("step_tool_meta") or {}),
2610 }
2611 return record, plan
2613 def child_plan(node_id: str, this_result: Any, depth: int) -> list:
2614 children = self._spawn_children_from_result(this_result)
2615 if children and max_spawn_per_node is not None and len(children) > max_spawn_per_node:
2616 raise RunHardFailure(
2617 "SPAWN_OVER_FLIGHT",
2618 f"spawn node '{node_id}' emitted {len(children)} children, "
2619 f"exceeds max_spawn_per_node={max_spawn_per_node}",
2620 {
2621 "node_id": node_id,
2622 "depth": depth,
2623 "children": len(children),
2624 "max_spawn_per_node": max_spawn_per_node,
2625 },
2626 )
2627 return children
2629 def director_node(_state: dict) -> Command:
2630 result_state = run_node(
2631 role=director,
2632 node_step=step,
2633 input_override=_UNSET,
2634 node_id=step_id,
2635 depth=0,
2636 )
2637 this_result = result_state.get(step_id)
2638 children = child_plan(step_id, this_result, 0)
2639 record, plan = build_record(
2640 node_id=step_id,
2641 path=(),
2642 parent_path=None,
2643 depth=0,
2644 this_result=this_result,
2645 result_state=result_state,
2646 children=children,
2647 )
2648 update = {
2649 "accum": {step_id: {(): record}},
2650 "outstanding": len(plan) - 1,
2651 }
2652 if not plan:
2653 return Command(update=update, goto="join")
2654 return Command(
2655 update=update,
2656 goto=[
2657 Send("worker", {**payload, "parent_obs_id": root_obs_id})
2658 for payload in plan
2659 ],
2660 )
2662 def worker_node(payload: dict) -> Command:
2663 path = tuple(payload["path"])
2664 depth = payload["depth"]
2665 parent_id = payload["parent_id"]
2666 index = payload["index"]
2667 child = payload["child"]
2668 child_id = (
2669 child["id"]
2670 if isinstance(child, dict) and child.get("id")
2671 else f"{parent_id}.{index}"
2672 )
2673 node_step = self._child_node_step(
2674 node_id=child_id,
2675 child=child,
2676 parent_id=parent_id,
2677 index=index,
2678 base_state=base_state,
2679 base_ctx=base_ctx,
2680 )
2681 # The worker's own graph context does not carry the spawning
2682 # node's observation (each Send target runs in a fresh context),
2683 # so the parent chain is re-established from the payload before
2684 # opening this node's observation (ADR-0035 D5).
2685 parent_obs_id = payload.get("parent_obs_id")
2686 stack_token = (
2687 _PARENT_STACK.set((parent_obs_id,))
2688 if tracer is not None and parent_obs_id is not None
2689 else None
2690 )
2691 child_obs = (
2692 tracer.open_parent_observation(
2693 name=f"spawn:{child_id}",
2694 input={"node_id": child_id, "role": child_role, "depth": depth},
2695 )
2696 if tracer is not None
2697 else None
2698 )
2699 try:
2700 result_state = run_node(
2701 role=child_role,
2702 node_step=node_step,
2703 input_override=node_step["input"],
2704 node_id=child_id,
2705 depth=depth,
2706 parent_id=parent_id,
2707 )
2708 finally:
2709 if tracer is not None and child_obs is not None:
2710 tracer.close_parent_observation(child_obs)
2711 if stack_token is not None:
2712 _PARENT_STACK.reset(stack_token)
2713 this_result = result_state.get(child_id)
2714 children = child_plan(child_id, this_result, depth)
2715 record, plan = build_record(
2716 node_id=child_id,
2717 path=path,
2718 parent_path=path[:-1],
2719 depth=depth,
2720 this_result=this_result,
2721 result_state=result_state,
2722 children=children,
2723 )
2724 update = {
2725 "accum": {step_id: {path: record}},
2726 "outstanding": len(plan) - 1,
2727 }
2728 if plan:
2729 child_obs_id = getattr(child_obs, "id", None)
2730 return Command(
2731 update=update,
2732 goto=[
2733 Send(
2734 "worker",
2735 {**payload, "parent_obs_id": child_obs_id},
2736 )
2737 for payload in plan
2738 ],
2739 )
2740 return Command(update=update, goto="check")
2742 def check_node(state: dict) -> Command:
2743 if state.get("outstanding") == 0:
2744 return Command(goto="join")
2745 return Command(goto=END)
2747 def join_node(state: dict) -> dict:
2748 records = (state.get("accum") or {}).get(step_id) or {}
2749 envelope, metrics, tool_meta = _assemble_spawn_outcome(records)
2750 return {
2751 "spawn_outcome": {
2752 "envelope": envelope,
2753 "metrics": metrics,
2754 "tool_meta": tool_meta,
2755 },
2756 "accum": {step_id: None},
2757 }
2759 schema = TypedDict(
2760 "_SpawnState",
2761 {
2762 "accum": Annotated[dict, accumulate_map_reduce],
2763 "outstanding": Annotated[int, operator.add],
2764 "spawn_outcome": dict,
2765 },
2766 )
2767 graph = StateGraph(schema)
2768 graph.add_node("director", director_node)
2769 graph.add_node("worker", worker_node)
2770 graph.add_node("check", check_node)
2771 graph.add_node("join", join_node)
2772 graph.add_edge(START, "director")
2773 graph.add_edge("join", END)
2774 app = graph.compile()
2775 out = app.invoke(
2776 {"accum": {}, "outstanding": 1},
2777 config={"recursion_limit": _SPAWN_RECURSION_LIMIT},
2778 )
2779 return out["spawn_outcome"]
2780 finally:
2781 if tracer is not None and parent_obs is not None:
2782 tracer.close_parent_observation(parent_obs)
2784 def _child_node_step(
2785 self,
2786 *,
2787 node_id: str,
2788 child: Any,
2789 parent_id: str,
2790 index: int,
2791 base_state: PipelineState,
2792 base_ctx: dict,
2793 ) -> dict:
2794 """Resolve one child plan item into the worker's runtime step.
2796 A child is a runtime-generated step: it carries only its id and its
2797 resolved input, never the spawn step's declared channel bindings, so a
2798 child's output can never route a channel write.
2799 """
2800 if not self._spawn_has_input(child):
2801 raise RunHardFailure(
2802 "INVALID_CHILD_PLAN",
2803 f"spawn child #{index} of node '{parent_id}' must be a dict "
2804 "with a non-empty 'input'",
2805 {"node_id": parent_id, "child_index": index, "child": repr(child)[:200]},
2806 )
2807 return {"id": node_id, "input": _resolve_input(child["input"], base_state, base_ctx)}
2809 def _stage_spawn_channel_writes(
2810 self,
2811 step: dict,
2812 result: Any,
2813 working_set: ChannelWorkingSet | None,
2814 ) -> None:
2815 """Stage a spawn node's declared writes from its own agent result."""
2816 if working_set is None:
2817 return
2818 writes = _declared_output_channel_writes(step, result, self._channel_names)
2819 for name, value in writes.items():
2820 working_set.stage(name, value)
2822 @staticmethod
2823 def _spawn_children_from_result(result: Any) -> list:
2824 """The node's child plan at ``content.children``; [] for a valid leaf.
2826 The director's parsed output envelope carries the plan at
2827 ``content.children``; a result that is not a dict, or whose ``content``
2828 is not an object, or carries no ``children``, is a valid leaf. A
2829 ``children`` value that is not a list is a hard ``INVALID_CHILD_PLAN``.
2830 """
2831 content = _envelope_content(result)
2832 if not isinstance(content, dict):
2833 return []
2834 raw = content.get("children")
2835 if raw is None:
2836 return []
2837 if not isinstance(raw, list):
2838 raise RunHardFailure(
2839 "INVALID_CHILD_PLAN",
2840 "spawn agent content 'children' must be a list",
2841 {"children": repr(raw)[:200]},
2842 )
2843 return raw
2845 @staticmethod
2846 def _spawn_has_input(child: Any) -> bool:
2847 """A valid child plan item: a dict carrying a non-empty 'input'."""
2848 if not isinstance(child, dict):
2849 return False
2850 input_val = child.get("input")
2851 if input_val is None:
2852 return False
2853 if isinstance(input_val, str) and not input_val.strip():
2854 return False
2855 if isinstance(input_val, (list, dict)) and not input_val:
2856 return False
2857 return True
2859 @staticmethod
2860 def _spawn_is_soft_fail(result: Any) -> bool:
2861 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error.
2863 An envelope-shaped result is never soft: ``status:error`` is a hard
2864 failure (the gate raises before this), and ``ok``/``partial``/
2865 ``no_success`` are valid non-fatal results.
2866 """
2867 if not isinstance(result, dict):
2868 return False
2869 if result.get("status") in ENVELOPE_STATUSES:
2870 return False
2871 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
2872 return True
2873 return any(key.startswith("budget:") for key in result)
2876# ---------------------------------------------------------------------------
2877# GraphBuilder
2878# ---------------------------------------------------------------------------
2880def _topological_order(steps: list[dict]) -> list[dict]:
2881 """
2882 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2883 """
2884 id_to_step = {s["id"]: s for s in steps}
2885 visited: set[str] = set()
2886 order: list[dict] = []
2888 def visit(step_id: str) -> None:
2889 if step_id in visited:
2890 return
2891 visited.add(step_id)
2892 for dep in (id_to_step[step_id].get("depends_on") or []):
2893 if dep in id_to_step:
2894 visit(dep)
2895 order.append(id_to_step[step_id])
2897 for s in steps:
2898 visit(s["id"])
2899 return order
2902def _group_dependencies(
2903 groups: list[list[dict]],
2904) -> tuple[list[set[int]], list[set[int]]]:
2905 """Group-index adjacency (predecessors, successors) from each step's depends_on."""
2906 step_to_group = {
2907 step["id"]: index
2908 for index, group in enumerate(groups)
2909 for step in group
2910 }
2911 predecessors: list[set[int]] = [set() for _ in groups]
2912 successors: list[set[int]] = [set() for _ in groups]
2913 for index, group in enumerate(groups):
2914 for step in group:
2915 for dep in step.get("depends_on") or []:
2916 dep_index = step_to_group.get(dep)
2917 if dep_index is None or dep_index == index:
2918 continue
2919 predecessors[index].add(dep_index)
2920 successors[dep_index].add(index)
2921 return predecessors, successors
2924@dataclass(frozen=True)
2925class _RoutePlan:
2926 """Compiled `routes` topology: route tables plus the static-edge suppression sets.
2928 `routes` owns a routing step's outgoing control flow (ADR-0025). A route
2929 target reached via a static `depends_on` edge is therefore suppressed and
2930 replaced by the native conditional edge; the plan records both the targets
2931 (so they are never made entry points) and the suppressed `(source, target)`
2932 pairs.
2933 """
2935 step_to_node: dict[str, str]
2936 routes_by_step: dict[str, list[dict]]
2937 target_node_ids: frozenset[str]
2938 routing_node_ids: frozenset[str]
2939 suppressed_edges: frozenset[tuple[str, str]]
2942def _reject_nested_routes(step: dict) -> None:
2943 """Reject `routes` on a nested `body` step (top-level-only rule, ADR-0012).
2945 The engine wires `routes` only for top-level steps, so a nested declaration
2946 would be silently ignored; nested routing is deferred.
2947 """
2948 body = step.get("body")
2949 if not isinstance(body, list):
2950 return
2951 for nested in _iter_steps(body):
2952 if nested.get("routes"):
2953 raise RunHardFailure(
2954 "VALIDATION_ERROR",
2955 f"Step '{nested.get('id')}' declares `routes` inside a nested "
2956 "body; `routes` is only allowed on top-level steps",
2957 )
2960def _build_route_plan(groups: list[list[dict]], node_ids: list[str]) -> _RoutePlan:
2961 """Resolve every top-level step's `routes` into a node-level routing plan."""
2962 step_to_node = {
2963 step["id"]: node_ids[index]
2964 for index, group in enumerate(groups)
2965 for step in group
2966 }
2967 step_to_deps = {
2968 step["id"]: set(step.get("depends_on") or [])
2969 for group in groups
2970 for step in group
2971 }
2972 routes_by_step: dict[str, list[dict]] = {}
2973 target_node_ids: set[str] = set()
2974 routing_node_ids: set[str] = set()
2975 suppressed_edges: set[tuple[str, str]] = set()
2976 for group in groups:
2977 for step in group:
2978 _reject_nested_routes(step)
2979 routes = step.get("routes")
2980 if not routes:
2981 continue
2982 source = step["id"]
2983 source_node = step_to_node[source]
2984 routing_node_ids.add(source_node)
2985 routes_by_step[source] = routes
2986 for route in routes:
2987 target = route.get("target")
2988 target_node = step_to_node.get(target)
2989 if target_node is None:
2990 raise RunHardFailure(
2991 "VALIDATION_ERROR",
2992 f"Step '{source}' route target '{target}' is not a "
2993 "top-level step in this pipeline",
2994 )
2995 target_node_ids.add(target_node)
2996 if source in step_to_deps.get(target, set()):
2997 suppressed_edges.add((source_node, target_node))
2998 log.warning(
2999 "[routes] suppressing static edge %s -> %s; route "
3000 "targets are wired via `routes`",
3001 source,
3002 target,
3003 )
3004 return _RoutePlan(
3005 step_to_node=step_to_node,
3006 routes_by_step=routes_by_step,
3007 target_node_ids=frozenset(target_node_ids),
3008 routing_node_ids=frozenset(routing_node_ids),
3009 suppressed_edges=frozenset(suppressed_edges),
3010 )
3013def _make_route_router(routes: list[dict], step_to_node: dict[str, str], end) -> Any:
3014 """First-match-wins router over a routing step's declared routes.
3016 Evaluates `when` in declaration order and returns the selected target node;
3017 a route without `when` is the default branch. With no match and no default
3018 it returns the graph terminal.
3019 """
3020 def router(state: PipelineState) -> str:
3021 ctx = _flatten_channel_state(state) if _CORE in state else state
3022 for route in routes:
3023 if _eval_condition(route.get("when"), ctx):
3024 return step_to_node[route["target"]]
3025 return end
3027 return router
3030def _wire_route_edges(graph, route_plan: _RoutePlan, end) -> None:
3031 """Add one native conditional edge per routing step, with an explicit path_map."""
3032 for source, routes in route_plan.routes_by_step.items():
3033 target_nodes = [route_plan.step_to_node[r["target"]] for r in routes]
3034 path_map = {node: node for node in target_nodes}
3035 path_map[end] = end
3036 graph.add_conditional_edges(
3037 route_plan.step_to_node[source],
3038 _make_route_router(routes, route_plan.step_to_node, end),
3039 path_map,
3040 )
3043def _wire_dependency_edges(
3044 graph,
3045 groups: list[list[dict]],
3046 node_ids: list[str],
3047 end,
3048 route_plan: _RoutePlan,
3049 entry_ids: list[str] | None = None,
3050 exit_ids: list[str] | None = None,
3051) -> None:
3052 """Wire native branches and fan-in joins from group-level depends_on edges.
3054 Static edges originating from a routing step to one of its route targets are
3055 suppressed — `routes` supplies that control flow. A route target with no
3056 other predecessor is reached by its conditional edge, never an entry point.
3057 A group split into a Send fan-out enters through its dispatch node and exits
3058 through its join node, so ``entry_ids``/``exit_ids`` name those boundaries.
3059 """
3060 entry_ids = entry_ids if entry_ids is not None else node_ids
3061 exit_ids = exit_ids if exit_ids is not None else node_ids
3062 predecessors, successors = _group_dependencies(groups)
3063 for index, node_id in enumerate(node_ids):
3064 entry = entry_ids[index]
3065 pred_nodes = [exit_ids[p] for p in sorted(predecessors[index])]
3066 if not pred_nodes:
3067 if node_id not in route_plan.target_node_ids:
3068 graph.set_entry_point(entry)
3069 else:
3070 live_preds = [
3071 pred
3072 for pred in pred_nodes
3073 if (pred, entry) not in route_plan.suppressed_edges
3074 ]
3075 if len(live_preds) == 1:
3076 graph.add_edge(live_preds[0], entry)
3077 elif live_preds:
3078 graph.add_edge(live_preds, entry)
3079 if not successors[index] and node_id not in route_plan.routing_node_ids:
3080 graph.add_edge(exit_ids[index], end)
3083# ---------------------------------------------------------------------------
3084# Call-step build-time expansion (ADR-0037 D3/D4)
3085# ---------------------------------------------------------------------------
3086#
3087# A `type: call` step is composed at build time: the function's body steps plus
3088# two synthesized op steps (`__call_bind` / `__call_emit`) are assembled into a
3089# compiled subgraph added as a single parent-graph node keyed by the call id. The
3090# runtime executes only native steps; LangGraph never sees a call step. The caller
3091# reads the function outputs via the call result shape
3092# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
3093# The function body's `{{ params.<name> }}` references are rewritten to
3094# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves. A
3095# call nested in a foreach/while body is flat-inlined instead, because loop bodies
3096# execute imperatively and cannot host a compiled subgraph node.
3097#
3098# Isolation (ADR-0037 D5 / 00-functions § Isolation) and recursion (ADR-0037 D7): a
3099# function body step must not reference a step-id outside its own body (a
3100# one-way dependency check — outward references to caller step-ids are build
3101# errors); a cycle in the function call graph (self or mutual) is a build
3102# error. Bounded while loops are runtime constructs and are not cycle-flagged.
3104# Call step result namespace keys (internal scaffolding).
3105def _scope_params_value(value: Any, call_id: str) -> Any:
3106 """Rewrite `params.` template references to the call-scoped namespace.
3108 Operates on the function body's string values (templates). Dict keys are
3109 untouched — only templated string VALUES are scoped, so a function body's
3110 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
3111 """
3112 if isinstance(value, str):
3113 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
3114 if isinstance(value, dict):
3115 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
3116 if isinstance(value, list):
3117 return [_scope_params_value(v, call_id) for v in value]
3118 return value
3121def _scope_params(step: dict, call_id: str) -> dict:
3122 """Deep-scope a function body step's params references for a call id."""
3123 scoped: dict = dict(step)
3124 for field in ("input", "params", "condition", "exit_condition", "agent", "function"):
3125 if field in scoped and isinstance(scoped[field], (str, dict, list)):
3126 scoped[field] = _scope_params_value(scoped[field], call_id)
3127 if "body" in scoped and isinstance(scoped["body"], list):
3128 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
3129 return scoped
3132def _collect_step_ids(steps: list[dict]) -> set[str]:
3133 """All step ids declared in *steps*, including nested body step ids."""
3134 ids: set[str] = set()
3135 for step in steps:
3136 step_id = step.get("id")
3137 if isinstance(step_id, str):
3138 ids.add(step_id)
3139 body = step.get("body")
3140 if isinstance(body, list):
3141 ids |= _collect_step_ids(body)
3142 return ids
3145def _iter_steps(steps: list[dict]):
3146 """Yield every step in *steps*, recursing into nested bodies."""
3147 for step in steps:
3148 yield step
3149 body = step.get("body")
3150 if isinstance(body, list):
3151 yield from _iter_steps(body)
3154def _collect_composed_step_ids(steps: list[dict]) -> set[str]:
3155 """All declared step ids, including ids composed into native subgraphs.
3157 A ``call`` step becomes a ``__subgraph__`` placeholder whose inner step ids
3158 live only inside the compiled subgraph. The placeholder records them under
3159 ``_SUBGRAPH_STEP_IDS_KEY`` so name-collision validation sees the full set
3160 even after body steps stop being textually inlined.
3161 """
3162 ids = _collect_step_ids(steps)
3163 for step in _iter_steps(steps):
3164 composed_ids = step.get(_SUBGRAPH_STEP_IDS_KEY)
3165 if isinstance(composed_ids, (set, frozenset)):
3166 ids |= composed_ids
3167 return ids
3170_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
3171_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
3172_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
3174_REFERENCE_FIELDS = ("input", "params", "condition", "exit_condition")
3176# Engine-reserved template namespaces — never step-id references, even if a
3177# caller happens to declare a step with one of these names.
3178_RESERVED_NAMESPACES = frozenset({"params", "loop", "item", "_item", "budget", "config"})
3181def _referenced_names(value: Any) -> set[str]:
3182 """Root identifiers read by templated strings.
3184 Only the leading identifier of each reference (before the first ``.`` or
3185 ``[``) can name a step-id; nested field segments are ignored. Quoted
3186 literals are ignored.
3187 """
3188 names: set[str] = set()
3189 if isinstance(value, str):
3190 for expr in _TEMPLATE_REF_RE.findall(value):
3191 bare = _STRING_LITERAL_RE.sub("", expr)
3192 for match in _IDENTIFIER_RE.finditer(bare):
3193 if match.start() == 0 or bare[match.start() - 1] != ".":
3194 names.add(match.group())
3195 elif isinstance(value, dict):
3196 for v in value.values():
3197 names |= _referenced_names(v)
3198 elif isinstance(value, list):
3199 for v in value:
3200 names |= _referenced_names(v)
3201 return names
3204def _validate_function_purity(
3205 fn_name: str, body: list[dict], visible_ids: set[str]
3206) -> None:
3207 """One-way isolation: a function body must not reference a caller step-id.
3209 A body step may read its own body step-ids and the params namespace; a
3210 reference to a step-id that exists outside the body (``visible_ids`` but
3211 not declared in the body) is an outward dependency and a hard build error.
3212 """
3213 local_ids = _collect_step_ids(body)
3214 for step in _iter_steps(body):
3215 step_id = step.get("id", "?")
3216 for dep in step.get("depends_on") or []:
3217 if dep in visible_ids and dep not in local_ids:
3218 raise RunHardFailure(
3219 "VALIDATION_ERROR",
3220 f"function '{fn_name}' step '{step_id}': depends_on "
3221 f"'{dep}' is outside the function body (isolation — "
3222 "function bodies see only their own steps and params)",
3223 )
3224 for field in _REFERENCE_FIELDS:
3225 for ref in _referenced_names(step.get(field)):
3226 if ref in _RESERVED_NAMESPACES:
3227 continue
3228 if ref in visible_ids and ref not in local_ids:
3229 raise RunHardFailure(
3230 "VALIDATION_ERROR",
3231 f"function '{fn_name}' step '{step_id}': '{field}' "
3232 f"references '{ref}', a step-id outside the function "
3233 "body (isolation — function bodies see only their own "
3234 "steps and params)",
3235 )
3238def _reject_function_body_routes(fn_name: str, body: list[dict]) -> None:
3239 """Reject `routes` anywhere in a function body (top-level-only rule).
3241 A function body is inlined to the top level, so a body step declaring
3242 `routes` would escape the nested-`body` guard and be silently rewired;
3243 reject it at build time instead.
3244 """
3245 for step in _iter_steps(body):
3246 if step.get("routes"):
3247 raise RunHardFailure(
3248 "VALIDATION_ERROR",
3249 f"function '{fn_name}' step '{step.get('id')}' declares `routes`; "
3250 "`routes` is only allowed on top-level steps",
3251 )
3254def _rewrite_call_deps(steps: list[dict], call_map: dict[str, str]) -> None:
3255 """Rewrite `depends_on` references to call ids → the corresponding emit id.
3257 After a call is expanded away, a downstream step that depends_on the call id
3258 must instead wait on the emit step (which writes state[<call_id>]). This
3259 guarantees the caller's `{{ <call_id>.outputs.<name> }}` reads resolve after
3260 the function body ran.
3261 """
3262 for step in steps:
3263 deps = step.get("depends_on") or []
3264 if deps:
3265 step["depends_on"] = [call_map.get(d, d) for d in deps]
3266 if "body" in step and isinstance(step["body"], list):
3267 _rewrite_call_deps(step["body"], call_map)
3270def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
3271 """Expand every ``call`` step through the native subgraph composer.
3273 Test-facing seam over the single call-expansion implementation: the same
3274 ``_CallComposer`` that ``GraphBuilder`` uses builds the steps, so no second
3275 (flat) expansion path exists. The embedded subgraph value is the un-compiled
3276 sub-step list; a runnable graph only needs the compiled form, which
3277 production obtains through ``GraphBuilder._compose_calls``.
3278 """
3279 composer = _CallComposer(
3280 registry,
3281 assemble_subgraph=lambda sub_steps: sub_steps,
3282 channels=[],
3283 channel_names=frozenset(),
3284 )
3285 return composer.compose(steps)
3288_SUBGRAPH_STEP_TYPE = "__subgraph__"
3289_SUBGRAPH_GRAPH_KEY = "__subgraph_graph__"
3290_SUBGRAPH_STEP_IDS_KEY = "__subgraph_step_ids__"
3292# A `while` step is composed into a `__subgraph__` placeholder whose embedded
3293# graph is a native LangGraph cycle (head -> body -> cond -> head/exit). The
3294# marker flags the placeholder so the parent uses the while node wrapper (which
3295# keeps the loop-internal state out of the parent result) rather than the plain
3296# call-subgraph node, and carries the build-time recursion bound.
3297_SUBGRAPH_WHILE_KEY = "__subgraph_while__"
3298_WHILE_RECURSION_LIMIT_KEY = "__while_recursion_limit__"
3300# Engine-local cycle state: the active while's `{current, iteration, passed,
3301# skipped, max_iter}` marker. Nested whiles live in isolated cycle invocations,
3302# so a shared key cannot collide across them.
3303_WHILE_STATE_KEY = "__while_state__"
3305# State keys the while wrapper propagates from the isolated cycle back to the
3306# parent (the loop result itself is carried separately under the step id).
3307_WHILE_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
3309# A sequential `foreach` (max_concurrent absent or resolving to <= 1) is
3310# composed into a `__subgraph__` placeholder whose embedded graph is a native
3311# LangGraph cycle (head -> body/agent -> cond -> head/exit). The marker flags
3312# the placeholder so the parent uses the foreach node wrapper, and carries the
3313# build-time recursion bound.
3314_SUBGRAPH_FOREACH_KEY = "__subgraph_foreach__"
3315_FOREACH_RECURSION_LIMIT_KEY = "__foreach_recursion_limit__"
3317# Engine-local cycle state: the active foreach's `{items, index, results,
3318# produced_outputs, max_output}` marker. Nested foreachs live in isolated cycle
3319# invocations, so a shared key cannot collide across them.
3320_FOREACH_STATE_KEY = "__foreach_state__"
3322# State keys the foreach wrapper propagates from the isolated cycle back to the
3323# parent (the aggregate result itself is carried separately under the step id).
3324_FOREACH_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
3326# Build-time recursion bound fallback for a foreach with no declared
3327# `max_iterations`: the item list is uncapped, so reserve a generous bound.
3328_FOREACH_DEFAULT_MAX_ITER = 1000
3331class _CallComposer:
3332 """Compose `call` steps into compiled native subgraph nodes (ADR-0037 D4).
3334 A function body becomes its own compiled StateGraph added to the parent as a
3335 single node keyed by the call id; the ``bind``/``emit`` scaffolding and the
3336 body steps live inside that subgraph and share the parent state. Nested
3337 function calls compose recursively. A `call` nested in a foreach/while body
3338 is composed the same way; the loop body executes the embedded compiled graph
3339 imperatively via ``run_step``'s ``__subgraph__`` branch.
3340 """
3342 def __init__(
3343 self,
3344 registry: ManifestRegistry,
3345 assemble_subgraph: Any,
3346 channels: list[dict],
3347 channel_names: frozenset[str],
3348 *,
3349 assemble_while: Any | None = None,
3350 compose_whiles: bool = False,
3351 assemble_foreach: Any | None = None,
3352 compose_foreachs: bool = False,
3353 build_ctx: dict | None = None,
3354 ) -> None:
3355 self._registry = registry
3356 self._assemble_subgraph = assemble_subgraph
3357 self._channels = channels
3358 self._channel_names = channel_names
3359 self._assemble_while = assemble_while
3360 self._compose_whiles = compose_whiles
3361 self._assemble_foreach = assemble_foreach
3362 self._compose_foreachs = compose_foreachs
3363 self._build_ctx = build_ctx or {}
3364 self._stack: list[str] = []
3365 self._expanded_functions: dict[str, str] = {}
3366 self._call_map: dict[str, str] = {}
3368 def compose(self, steps: list[dict]) -> list[dict]:
3369 visible_ids = _collect_step_ids(steps)
3370 composed: list[dict] = []
3371 for step in steps:
3372 composed.extend(self._compose_step(step, visible_ids=visible_ids))
3373 return composed
3375 def _compose_step(self, step: dict, *, visible_ids: set[str]) -> list[dict]:
3376 step_type = step.get("type")
3377 if step_type == "call":
3378 return [self._compose_subgraph_call(step, visible_ids)]
3379 if step_type == "while" and self._compose_whiles:
3380 return [self._compose_while(step, visible_ids)]
3381 if step_type == "foreach" and self._should_compose_foreach(step):
3382 return [self._compose_foreach(step, visible_ids)]
3383 body = step.get("body")
3384 if not isinstance(body, list):
3385 return [step]
3386 nested: list[dict] = []
3387 for sub in body:
3388 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3389 _rewrite_call_deps(nested, self._call_map)
3390 return [{**step, "body": nested}]
3392 def _compose_while(self, step: dict, visible_ids: set[str]) -> dict:
3393 """Compose a `while` step into a native LangGraph cycle subgraph.
3395 The embedded graph owns the loop topology: a head that resolves the
3396 entry condition / initial ``current``, the composed body step nodes, a
3397 conditional back-edge that computes the feed + exit condition and routes
3398 continue-vs-exit, and an exit node that writes the documented while
3399 result. ``expand_calls`` leaves the step uncomposed (``compose_whiles``
3400 false) so the call-expansion seam still exposes the loop body.
3401 """
3402 _reject_nested_routes(step)
3403 body = step.get("body") or []
3404 nested: list[dict] = []
3405 for sub in body:
3406 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3407 _validate_step_channel_bindings(nested, self._channel_names)
3408 compiled, recursion_limit = self._assemble_while(step, nested)
3409 placeholder: dict = {
3410 "id": step["id"],
3411 "type": _SUBGRAPH_STEP_TYPE,
3412 _SUBGRAPH_GRAPH_KEY: compiled,
3413 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3414 _SUBGRAPH_WHILE_KEY: True,
3415 _WHILE_RECURSION_LIMIT_KEY: recursion_limit,
3416 "depends_on": list(step.get("depends_on") or []),
3417 }
3418 if step.get("routes"):
3419 placeholder["routes"] = step["routes"]
3420 return placeholder
3422 def _should_compose_foreach(self, step: dict) -> bool:
3423 """Whether a `foreach` is the sequential form composed into a cycle.
3425 Only a `max_concurrent` absent or resolving to ``<= 1`` is composed; a
3426 declared ``max_concurrent > 1`` keeps the native ``Send`` / in-node
3427 concurrent behaviour.
3428 """
3429 if not self._compose_foreachs:
3430 return False
3431 return _resolved_send_concurrency(step, self._build_ctx) <= 1
3433 def _compose_foreach(self, step: dict, visible_ids: set[str]) -> dict:
3434 """Compose a sequential `foreach` into a native LangGraph cycle subgraph.
3436 Mirrors ``_compose_while``: the embedded graph owns the loop topology
3437 (a head that resolves the item list and injects the current item, the
3438 composed body step nodes or the per-item agent node, a conditional
3439 back-edge routing next-item-vs-exit, and an exit node that writes the
3440 documented aggregate). ``expand_calls`` leaves the step uncomposed so
3441 the call-expansion seam still exposes the loop body.
3442 """
3443 _reject_nested_routes(step)
3444 body = step.get("body") or []
3445 nested: list[dict] = []
3446 for sub in body:
3447 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3448 _validate_step_channel_bindings(nested, self._channel_names)
3449 compiled, recursion_limit = self._assemble_foreach(step, nested)
3450 placeholder: dict = {
3451 "id": step["id"],
3452 "type": _SUBGRAPH_STEP_TYPE,
3453 _SUBGRAPH_GRAPH_KEY: compiled,
3454 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3455 _SUBGRAPH_FOREACH_KEY: True,
3456 _FOREACH_RECURSION_LIMIT_KEY: recursion_limit,
3457 "depends_on": list(step.get("depends_on") or []),
3458 }
3459 if step.get("routes"):
3460 placeholder["routes"] = step["routes"]
3461 return placeholder
3463 def _compose_subgraph_call(self, step: dict, visible_ids: set[str]) -> dict:
3464 call_id = step["id"]
3465 function, body = self._check_call_guards(step, visible_ids)
3467 nested_visible = visible_ids | _collect_step_ids(body)
3468 outputs_spec = (
3469 ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
3470 )
3471 bind_id = f"_call__{call_id}__bind"
3472 emit_id = f"_call__{call_id}__emit"
3473 bind_step: dict = {
3474 "id": bind_id,
3475 "type": "builtin",
3476 "component": "__call_bind",
3477 "params": {**(step.get("params") or {}), "__call_id": call_id},
3478 "depends_on": [],
3479 }
3481 self._stack.append(step["function"])
3482 try:
3483 expanded_body = self._compose_body(
3484 body, call_id, bind_id, nested_visible
3485 )
3486 finally:
3487 self._stack.pop()
3489 emit_step: dict = {
3490 "id": emit_id,
3491 "type": "builtin",
3492 "component": "__call_emit",
3493 "params": {"__call_id": call_id, "__outputs": outputs_spec},
3494 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
3495 }
3496 sub_steps = [bind_step, *expanded_body, emit_step]
3497 if self._channels:
3498 _validate_channel_inputs(self._channels, sub_steps)
3499 _validate_step_channel_bindings(sub_steps, self._channel_names)
3500 placeholder: dict = {
3501 "id": call_id,
3502 "type": _SUBGRAPH_STEP_TYPE,
3503 _SUBGRAPH_GRAPH_KEY: self._assemble_subgraph(
3504 _topological_order(sub_steps)
3505 ),
3506 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(sub_steps),
3507 "depends_on": list(step.get("depends_on") or []),
3508 }
3509 call_condition = step.get("condition")
3510 if call_condition is not None:
3511 placeholder["condition"] = call_condition
3512 return placeholder
3514 def _compose_body(
3515 self, body: list[dict], call_id: str, bind_id: str, visible_ids: set[str]
3516 ) -> list[dict]:
3517 expanded: list[dict] = []
3518 for body_step in body:
3519 scoped = _scope_params(body_step, call_id)
3520 for sub in self._compose_step(scoped, visible_ids=visible_ids):
3521 deps = list(sub.get("depends_on") or [])
3522 if bind_id in deps:
3523 expanded.append(sub)
3524 else:
3525 expanded.append({**sub, "depends_on": [bind_id, *deps]})
3526 return expanded
3528 def _check_call_guards(
3529 self, step: dict, visible_ids: set[str]
3530 ) -> tuple[dict, list[dict]]:
3531 """Single owner of every build-time ``call`` guard (ADR-0037 D3/D5/D7).
3533 Rejects routes on the call step, function cycles, duplicate expansion,
3534 unknown functions, function-body isolation (purity) violations, and
3535 body-declared routes; returns the resolved function and its body.
3536 """
3537 fn_name = step["function"]
3538 call_id = step["id"]
3540 if step.get("routes"):
3541 raise RunHardFailure(
3542 "VALIDATION_ERROR",
3543 f"call step '{call_id}' declares `routes`; `routes` is not supported "
3544 "on call steps",
3545 )
3546 if fn_name in self._stack:
3547 chain = " -> ".join(self._stack + [fn_name])
3548 raise RunHardFailure(
3549 "VALIDATION_ERROR",
3550 f"cycle detected in function expansion: {chain} "
3551 "(ADR-0037 D7 — recursion must be a bounded while loop or "
3552 "idempotent re-run)",
3553 )
3554 if fn_name in self._expanded_functions:
3555 raise RunHardFailure(
3556 "VALIDATION_ERROR",
3557 f"duplicate call expansion: function '{fn_name}' called at both "
3558 f"'{self._expanded_functions[fn_name]}' and '{call_id}' in one "
3559 "pipeline (ADR-0037 D7 — repeated calls to the same function are "
3560 "not yet supported)",
3561 )
3562 self._expanded_functions[fn_name] = call_id
3564 function = self._registry.get("Function", fn_name)
3565 if function is None:
3566 raise RunHardFailure(
3567 "VALIDATION_ERROR",
3568 f"call step '{call_id}' references unknown function '{fn_name}' "
3569 "(no kind: Function manifest in the workflows tree)",
3570 )
3571 body = (function.get("spec") or {}).get("steps") or []
3572 _validate_function_purity(fn_name, body, visible_ids)
3573 _reject_function_body_routes(fn_name, body)
3574 return function, body
3577_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
3580def _validate_channel_entry(index: int, channel: Any) -> None:
3581 """Validate one spec.channels entry against the frozen ChannelSpec shape.
3583 The engine build path does not run manifest-schema validation, so a
3584 malformed entry is rejected here — before any ``["name"]`` indexing — with
3585 an error that identifies the offending entry.
3586 """
3587 if not isinstance(channel, dict):
3588 raise RunHardFailure(
3589 "CHANNEL_BINDING_ERROR",
3590 f"Pipeline spec.channels[{index}] must be a mapping; got "
3591 f"{type(channel).__name__}",
3592 )
3593 name = channel.get("name")
3594 if not isinstance(name, str) or not name:
3595 raise RunHardFailure(
3596 "CHANNEL_BINDING_ERROR",
3597 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3598 "channel requires a non-empty string 'name'",
3599 )
3600 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3601 raise RunHardFailure(
3602 "CHANNEL_BINDING_ERROR",
3603 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3604 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}",
3605 )
3608def _validated_channels(raw_channels: Any) -> list[dict]:
3609 """Return spec.channels as a validated list, or [] when absent.
3611 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3612 """
3613 if raw_channels is None:
3614 return []
3615 if not isinstance(raw_channels, list):
3616 raise RunHardFailure(
3617 "CHANNEL_BINDING_ERROR",
3618 f"Pipeline spec.channels must be a list; got "
3619 f"{type(raw_channels).__name__}",
3620 )
3621 for index, channel in enumerate(raw_channels):
3622 _validate_channel_entry(index, channel)
3623 return raw_channels
3626def _validate_step_channel_bindings(
3627 steps: list[dict], channel_names: frozenset[str]
3628) -> None:
3629 """Reject malformed or undeclared step-level channel bindings at build time.
3631 Every step's declared ``output_channel`` must be a non-empty declared
3632 channel name, and every declared ``input_channels`` entry must be a
3633 non-empty declared channel name. This runs unconditionally, so a binding
3634 to a channel absent from ``spec.channels`` fails the build even when no
3635 channels are declared.
3636 """
3637 for step in _iter_steps(steps):
3638 _validate_step_output_channel(step, channel_names)
3639 _validate_step_input_channels(step, channel_names)
3642def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3643 channel = step.get("output_channel")
3644 if channel is None:
3645 return
3646 if not isinstance(channel, str) or not channel:
3647 raise RunHardFailure(
3648 "CHANNEL_BINDING_ERROR",
3649 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3650 f"string; got {channel!r}",
3651 )
3652 if channel not in channel_names:
3653 raise RunHardFailure(
3654 "CHANNEL_BINDING_ERROR",
3655 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3656 "which is not declared in spec.channels",
3657 )
3660def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3661 channels = step.get("input_channels")
3662 if channels is None:
3663 return
3664 if not isinstance(channels, list):
3665 raise RunHardFailure(
3666 "CHANNEL_BINDING_ERROR",
3667 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3668 f"channel names; got {type(channels).__name__}",
3669 )
3670 for channel in channels:
3671 if not isinstance(channel, str) or not channel:
3672 raise RunHardFailure(
3673 "CHANNEL_BINDING_ERROR",
3674 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3675 f"channel name '{channel}'; channel names must be non-empty "
3676 "strings",
3677 )
3678 if channel not in channel_names:
3679 raise RunHardFailure(
3680 "CHANNEL_BINDING_ERROR",
3681 f"Step '{step.get('id', '?')}' reads undeclared channel "
3682 f"'{channel}'; the channel must be declared in spec.channels",
3683 )
3686def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3687 """Reject duplicate and engine/step-reserved channel names."""
3688 forbidden = (
3689 {_CORE}
3690 | set(_RESERVED_NAMESPACES)
3691 | _CORE_STATE_KEYS
3692 | _collect_composed_step_ids(steps)
3693 )
3694 seen: set[str] = set()
3695 for channel in channels:
3696 name = channel.get("name")
3697 if name in seen:
3698 raise RunHardFailure(
3699 "CHANNEL_BINDING_ERROR",
3700 f"Pipeline declares channel '{name}' more than once; channel "
3701 "names must be unique",
3702 )
3703 seen.add(name)
3704 if name in forbidden:
3705 raise RunHardFailure(
3706 "CHANNEL_BINDING_ERROR",
3707 f"Pipeline channel '{name}' collides with an engine-reserved "
3708 "name (a step id, engine state key, reserved namespace, or the "
3709 "core state channel); channel names must be unique and "
3710 "non-reserved",
3711 )
3714def _input_path_root(value: str) -> str | None:
3715 """Leading state key of a bare dot-path input string, or None.
3717 Template strings, bracket-prefixed verbatim literals, and blank strings
3718 are not state reads and yield None.
3719 """
3720 if "{{" in value:
3721 return None
3722 stripped = value.lstrip()
3723 if not stripped or stripped.startswith("["):
3724 return None
3725 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3728def _iter_input_roots(input_spec: Any):
3729 """Yield every bare dot-path root in an input spec, recursing into containers."""
3730 if isinstance(input_spec, str):
3731 root = _input_path_root(input_spec)
3732 if root:
3733 yield root
3734 elif isinstance(input_spec, dict):
3735 for value in input_spec.values():
3736 yield from _iter_input_roots(value)
3737 elif isinstance(input_spec, list):
3738 for value in input_spec:
3739 yield from _iter_input_roots(value)
3742def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3743 """Reject channel-pipeline input roots that name nothing the state exposes.
3745 The channel state is the declared channels plus the engine state, so a bare
3746 dot-path input root must be a declared channel, a known step id, a reserved
3747 namespace, or an engine state key. Anything else is a hard build error —
3748 there is no silent fallback to a missing key.
3749 """
3750 allowed = (
3751 {ch["name"] for ch in channels}
3752 | _collect_step_ids(steps)
3753 | set(_RESERVED_NAMESPACES)
3754 | _CORE_STATE_KEYS
3755 )
3756 for step in _iter_steps(steps):
3757 for root in _iter_input_roots(step.get("input")):
3758 if root not in allowed:
3759 raise RunHardFailure(
3760 "CHANNEL_BINDING_ERROR",
3761 f"Pipeline channel step '{step.get('id', '?')}' input "
3762 f"references unknown root '{root}'; in a channel pipeline "
3763 "an input root must be a declared channel, a step id, a "
3764 "reserved namespace, or an engine state key",
3765 )
3768def _flatten_channel_state(state: dict) -> dict:
3769 """Read the flat committed snapshot nested under the Core namespace."""
3770 return dict(state.get(_CORE) or {})
3773def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3774 """Drop channel-named keys so a node emits only the core state."""
3775 return {k: v for k, v in flat.items() if k not in channel_names}
3778_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3780_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3783def _envelope_content(result: Any) -> Any:
3784 """Return an envelope's optional ``content`` field (ADR-0023 D3).
3786 ``content`` is opaque and optional: a non-mapping result, an absent field,
3787 and an explicit ``null`` all read as ``None``. Callers apply their own
3788 fallback and shape checks (e.g. ``[]`` for a spawn child plan).
3789 """
3790 if isinstance(result, dict):
3791 return result.get("content")
3792 return None
3795def _is_failure_envelope(value: Any) -> bool:
3796 """True when a step output is not a successfully parsed dict.
3798 Missing output, non-dict values, and the ``error``/``parse_error``
3799 envelopes written by the LLM failure paths are not routed. A valid
3800 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3801 it carries an explicit ``"error": null``.
3802 """
3803 if not isinstance(value, dict):
3804 return True
3805 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3806 return False
3807 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3810def _declared_output_channel_writes(
3811 step: dict, output: Any, channel_names: frozenset[str]
3812) -> dict[str, Any]:
3813 """Route one step output's content to its declared output channel.
3815 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3816 the working set. Failure envelopes and steps without an ``output_channel``
3817 are not routed; an undeclared target channel is a hard error.
3818 """
3819 if _is_failure_envelope(output):
3820 return {}
3821 channel = step.get("output_channel")
3822 if not channel:
3823 return {}
3824 if channel not in channel_names:
3825 raise RunHardFailure(
3826 "CHANNEL_BINDING_ERROR",
3827 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3828 "which is not declared in spec.channels",
3829 )
3830 return {channel: _envelope_content(output)}
3833def _route_declared_channels(
3834 step: dict,
3835 pre_state: dict,
3836 core: dict,
3837 step_runner: Any,
3838 channel_names: frozenset[str],
3839) -> dict[str, Any]:
3840 """Route a step's declared output channel write to its channel.
3842 Returns ``{channel_name: value}`` for LangGraph to merge through the
3843 channel's reducer, where ``channel_name`` is the step's single declared
3844 ``output_channel`` and ``value`` is the agent envelope's ``content``. Only
3845 an actual agent invocation is routed: a ``skip_llm_if`` passthrough and a
3846 parallel fan-out envelope are excluded. A step without an ``output_channel``
3847 routes nothing (its output stays in ``core`` untouched), and failure
3848 envelopes are not routed. A spawn step is routed at the spawn seam from its
3849 node result, never here from its envelope.
3850 """
3851 step_id = step.get("id", "?")
3852 if step.get("type") == "spawn":
3853 return {}
3854 if step_id not in core:
3855 return {}
3856 agent_spec = step_runner.resolve_agent(step)
3857 if agent_spec is None:
3858 return {}
3859 if not step_runner.ran_agent(step, pre_state):
3860 return {}
3861 return _declared_output_channel_writes(step, core[step_id], channel_names)
3864def _make_step_node(
3865 step: dict,
3866 step_runner: Any,
3867 channel_names: frozenset[str],
3868 channel_types: dict[str, str],
3869) -> Any:
3870 if not channel_names:
3871 def node_fn(state: PipelineState) -> PipelineState:
3872 return step_runner.run_step(step, state)
3873 return node_fn
3875 def node_fn(state: PipelineState) -> PipelineState:
3876 committed = _flatten_channel_state(state)
3877 working_set = ChannelWorkingSet(committed, channel_types)
3878 core = _strip_channel_keys(
3879 step_runner.run_step(step, committed, working_set), channel_names
3880 )
3881 writes = _route_declared_channels(
3882 step, committed, core, step_runner, channel_names
3883 )
3884 for name, value in writes.items():
3885 working_set.stage(name, value)
3886 written_names = {name for name, _ in working_set.pending_writes()}
3887 channels = working_set.commit()
3888 emitted = {name: channels[name] for name in written_names}
3889 return {_CORE: {**core, **emitted}}
3890 return node_fn
3893def _make_subgraph_node(
3894 compiled: Any,
3895 condition: Any,
3896 step_runner: Any,
3897 channel_names: frozenset[str],
3898) -> Any:
3899 """Wrap a compiled call subgraph with the call step's entry condition.
3901 A false condition returns no state updates, so the subgraph never runs and
3902 ``state[<call_id>]`` is not written — matching the soft-skip semantics of a
3903 normal step condition. The ctx is derived exactly like ``run_step``'s,
3904 including channel-mode flattening. With no condition the compiled subgraph
3905 is the node, preserving the existing wiring.
3906 """
3907 if condition is None:
3908 return compiled
3910 if not channel_names:
3911 def node_fn(state: PipelineState) -> PipelineState:
3912 if not _eval_condition(condition, step_runner._make_ctx(state)):
3913 return {}
3914 return compiled.invoke(state)
3915 return node_fn
3917 def node_fn(state: PipelineState) -> PipelineState:
3918 committed = _flatten_channel_state(state)
3919 if not _eval_condition(condition, step_runner._make_ctx(committed)):
3920 return {}
3921 return compiled.invoke(state)
3922 return node_fn
3925# ---------------------------------------------------------------------------
3926# Native `while` cycle nodes (ADR-0025 D1/D2)
3927# ---------------------------------------------------------------------------
3928#
3929# A `while` step is a compiled cycle: `head -> body... -> cond` with a
3930# conditional back-edge from `cond` to `head` (continue) or `exit`. The head
3931# owns entry-condition handling and the initial `current`; the cond node owns
3932# the feed, the iteration increment and the exit-condition evaluation; the exit
3933# node writes the documented result and enforces `fail_if_exhausted`. The
3934# nodes read/write the shared graph state, using the `_CORE` container when the
3935# pipeline declares channels.
3938def _while_marker_key(step_id: str) -> str:
3939 """Per-while engine marker key (nested whiles live in isolated cycles)."""
3940 return f"{_WHILE_STATE_KEY}__{step_id}"
3943def _while_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
3944 return _flatten_channel_state(state) if channel_names else state
3947def _while_update(update: dict, channel_names: frozenset[str]) -> dict:
3948 return {_CORE: update} if channel_names else update
3951def _while_loop_context(step_id: str, marker: dict) -> dict:
3952 """The body-visible `loop` namespace and the intermediate step result.
3954 Written in this order so a step literally named ``loop`` shadows the
3955 injected namespace — the legacy in-node loop had the same collision.
3956 """
3957 current = marker["current"]
3958 iteration = marker["iteration"]
3959 update = {"loop": {"current": current, "iteration": iteration}}
3960 update[step_id] = {"current": current, "passed": False}
3961 return update
3964def _while_invoke_config(step: dict) -> dict | None:
3965 """Recursion bound for a while/foreach cycle placeholder, else None."""
3966 if step.get(_SUBGRAPH_WHILE_KEY):
3967 return {"recursion_limit": step.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
3968 if step.get(_SUBGRAPH_FOREACH_KEY):
3969 return {"recursion_limit": step.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
3970 return None
3973def _make_while_body_node(
3974 sub: dict,
3975 step_runner: Any,
3976 channel_names: frozenset[str],
3977 channel_types: dict[str, str],
3978) -> Any:
3979 """A while body step as a node: a nested subgraph or a normal step node."""
3980 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
3981 if sub.get(_SUBGRAPH_WHILE_KEY):
3982 return _make_while_subgraph_node(sub, step_runner, channel_names)
3983 if sub.get(_SUBGRAPH_FOREACH_KEY):
3984 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
3985 return _make_subgraph_node(
3986 sub[_SUBGRAPH_GRAPH_KEY], sub.get("condition"), step_runner, channel_names
3987 )
3988 return _make_step_node(sub, step_runner, channel_names, channel_types)
3991def _make_while_head_node(
3992 step: dict,
3993 step_runner: Any,
3994 channel_names: frozenset[str],
3995 max_iter: int,
3996) -> Any:
3997 """Entry node: handle the entry condition and seed the loop marker."""
3998 step_id = step.get("id", "?")
3999 marker_key = _while_marker_key(step_id)
4001 def node_fn(state: PipelineState) -> PipelineState:
4002 flat = _while_flat(state, channel_names)
4003 marker = flat.get(marker_key)
4004 if marker is not None:
4005 return _while_update(
4006 _while_loop_context(step_id, marker), channel_names
4007 )
4008 ctx = step_runner._make_ctx(flat)
4009 entry_cond = step.get("condition")
4010 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4011 skip_input = _resolved_while_input(step, flat, ctx)
4012 skip_marker = {
4013 "current": None,
4014 "iteration": 0,
4015 "passed": True,
4016 "skipped": True,
4017 "max_iter": max_iter,
4018 }
4019 return _while_update(
4020 {
4021 step_id: {
4022 "passed": True,
4023 "iterations": 0,
4024 "output": skip_input,
4025 "exhausted": False,
4026 "skipped": True,
4027 },
4028 marker_key: skip_marker,
4029 },
4030 channel_names,
4031 )
4032 marker = {
4033 "current": _resolved_while_input(step, flat, ctx),
4034 "iteration": 0,
4035 "passed": False,
4036 "skipped": False,
4037 "max_iter": max_iter,
4038 }
4039 update = _while_loop_context(step_id, marker)
4040 update[marker_key] = marker
4041 return _while_update(update, channel_names)
4043 return node_fn
4046def _resolved_while_input(step: dict, flat: dict, ctx: dict) -> Any:
4047 input_spec = step.get("input")
4048 if input_spec is None:
4049 return None
4050 return _resolve_input(input_spec, flat, ctx)
4053def _make_while_head_router(
4054 step: dict,
4055 channel_names: frozenset[str],
4056 first_id: str,
4057 exit_id: str,
4058 end: Any,
4059) -> Any:
4060 """Route from the head: skipped → end, bound already reached → exit, else body."""
4061 marker_key = _while_marker_key(step.get("id", "?"))
4063 def router(state: PipelineState) -> str:
4064 marker = _while_flat(state, channel_names).get(marker_key) or {}
4065 if marker.get("skipped"):
4066 return end
4067 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
4068 return exit_id
4069 return first_id
4071 return router
4074def _make_while_cond_node(
4075 step: dict,
4076 step_runner: Any,
4077 channel_names: frozenset[str],
4078) -> Any:
4079 """Feed the declared output, increment the iteration, evaluate the exit."""
4080 step_id = step.get("id", "?")
4081 marker_key = _while_marker_key(step_id)
4082 declared_output = step.get("output")
4083 exit_condition = step.get("exit_condition")
4085 def node_fn(state: PipelineState) -> PipelineState:
4086 flat = _while_flat(state, channel_names)
4087 marker = dict(flat.get(marker_key) or {})
4088 current = marker.get("current")
4089 if declared_output:
4090 feed_value = flat.get(declared_output)
4091 if isinstance(feed_value, dict) and "output" in feed_value:
4092 current = feed_value["output"]
4093 elif isinstance(feed_value, dict) and isinstance(
4094 feed_value.get("results"), list
4095 ):
4096 current = feed_value["results"]
4097 else:
4098 current = feed_value
4099 passed = False
4100 if exit_condition:
4101 if _eval_condition(
4102 exit_condition, step_runner._make_ctx(flat)
4103 ):
4104 passed = True
4105 else:
4106 passed = True
4107 marker.update(
4108 {
4109 "current": current,
4110 "iteration": marker.get("iteration", 0) + 1,
4111 "passed": passed,
4112 }
4113 )
4114 return _while_update({marker_key: marker}, channel_names)
4116 return node_fn
4119def _make_while_cond_router(
4120 step: dict,
4121 channel_names: frozenset[str],
4122 head_id: str,
4123 exit_id: str,
4124) -> Any:
4125 """Route from the cond: exit on pass or when the bound is reached, else head."""
4126 marker_key = _while_marker_key(step.get("id", "?"))
4128 def router(state: PipelineState) -> str:
4129 marker = _while_flat(state, channel_names).get(marker_key) or {}
4130 if marker.get("passed"):
4131 return exit_id
4132 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
4133 return exit_id
4134 return head_id
4136 return router
4139def _make_while_exit_node(
4140 step: dict,
4141 step_runner: Any,
4142 channel_names: frozenset[str],
4143) -> Any:
4144 """Write the documented while result and enforce `fail_if_exhausted`."""
4145 step_id = step.get("id", "?")
4146 marker_key = _while_marker_key(step_id)
4147 fail_if_exhausted = step.get("fail_if_exhausted")
4149 def node_fn(state: PipelineState) -> PipelineState:
4150 flat = _while_flat(state, channel_names)
4151 marker = flat.get(marker_key) or {}
4152 passed = bool(marker.get("passed", False))
4153 exhausted = not passed
4154 if (
4155 fail_if_exhausted is not None
4156 and exhausted
4157 and _eval_condition(
4158 fail_if_exhausted, step_runner._make_ctx(flat)
4159 )
4160 ):
4161 raise RunHardFailure(
4162 "EXHAUSTED",
4163 f"while step '{step_id}' exhausted "
4164 f"max_iterations={marker.get('max_iter')} without "
4165 "exit_condition passing (fail_if_exhausted)",
4166 {"step_id": step_id, "exhausted": True},
4167 )
4168 return _while_update(
4169 {
4170 step_id: {
4171 "passed": passed,
4172 "iterations": marker.get("iteration", 0),
4173 "output": marker.get("current"),
4174 "exhausted": exhausted,
4175 "skipped": False,
4176 }
4177 },
4178 channel_names,
4179 )
4181 return node_fn
4184def _make_while_subgraph_node(
4185 placeholder: dict,
4186 step_runner: Any,
4187 channel_names: frozenset[str],
4188) -> Any:
4189 """Parent node for a compiled `while` cycle: isolate and surface its result.
4191 The cycle shares the graph state schema, so invoking it directly would leak
4192 the body step results and loop plumbing into the parent. This wrapper runs
4193 the cycle on a copy and returns only the documented result plus the side
4194 channels, preserving the in-node loop's state contract exactly.
4195 """
4196 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4197 step_id = placeholder["id"]
4198 config = {"recursion_limit": placeholder.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
4200 if not channel_names:
4201 def node_fn(state: PipelineState) -> PipelineState:
4202 sub = compiled.invoke(state, config)
4203 result = {step_id: sub.get(step_id)}
4204 for key in _WHILE_SIDE_CHANNEL_KEYS:
4205 if key in sub and sub.get(key) != state.get(key):
4206 result[key] = sub.get(key)
4207 return result
4208 return node_fn
4210 def node_fn(state: PipelineState) -> PipelineState:
4211 sub = compiled.invoke(state, config)
4212 flat = _flatten_channel_state(sub)
4213 parent_flat = _flatten_channel_state(state)
4214 result = {step_id: flat.get(step_id)}
4215 for key in _WHILE_SIDE_CHANNEL_KEYS:
4216 if key in flat and flat.get(key) != parent_flat.get(key):
4217 result[key] = flat.get(key)
4218 return {_CORE: _strip_channel_keys(result, channel_names)}
4220 return node_fn
4223# ---------------------------------------------------------------------------
4224# Native sequential `foreach` cycle nodes
4225# ---------------------------------------------------------------------------
4226#
4227# A sequential `foreach` is a compiled cycle: `head -> (body... | agent) ->
4228# cond` with a conditional back-edge from `cond` to `head` (next item) or
4229# `exit` (`max_output` reached or the item list exhausted). The head resolves
4230# the item list once and injects the current item (`item` / `_item` /
4231# `_foreach_item`); the cond collects the item result; the exit node writes the
4232# documented aggregate (`{"results", "passed", "failing", "failed"}` for the
4233# body path, the plain list for the agent path). Nodes read/write the shared
4234# graph state, using the `_CORE` container when the pipeline declares channels.
4237def _foreach_marker_key(step_id: str) -> str:
4238 """Per-foreach engine marker key (nested foreachs live in isolated cycles)."""
4239 return f"{_FOREACH_STATE_KEY}__{step_id}"
4242def _foreach_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
4243 return _flatten_channel_state(state) if channel_names else state
4246def _foreach_update(update: dict, channel_names: frozenset[str]) -> dict:
4247 return {_CORE: update} if channel_names else update
4250def _make_foreach_head_node(
4251 step: dict,
4252 step_runner: Any,
4253 channel_names: frozenset[str],
4254) -> Any:
4255 """Entry node: evaluate the entry condition, resolve items, inject the item."""
4256 step_id = step.get("id", "?")
4257 marker_key = _foreach_marker_key(step_id)
4259 def node_fn(state: PipelineState) -> PipelineState:
4260 flat = _foreach_flat(state, channel_names)
4261 marker = flat.get(marker_key)
4262 if marker is None:
4263 ctx = step_runner._make_ctx(flat)
4264 entry_cond = step.get("condition")
4265 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4266 return _foreach_update({}, channel_names)
4267 max_output_raw = step.get("max_output")
4268 marker = {
4269 "items": step_runner._foreach_item_list(step, flat, ctx),
4270 "index": 0,
4271 "results": [],
4272 "produced_outputs": 0,
4273 "max_output": (
4274 _resolve_int(max_output_raw, ctx, 0)
4275 if max_output_raw is not None
4276 else 0
4277 ),
4278 }
4279 items = marker.get("items") or []
4280 index = marker.get("index", 0)
4281 if index < len(items):
4282 update = {
4283 "_item": items[index],
4284 "item": items[index],
4285 "_foreach_item": items[index],
4286 }
4287 update[marker_key] = marker
4288 return _foreach_update(update, channel_names)
4289 return _foreach_update({marker_key: marker}, channel_names)
4291 return node_fn
4294def _make_foreach_head_router(
4295 step: dict,
4296 channel_names: frozenset[str],
4297 first_id: str,
4298 exit_id: str,
4299 end: Any,
4300) -> Any:
4301 """Route from the head: condition-skipped → end, exhausted/budget → exit, else body."""
4302 marker_key = _foreach_marker_key(step.get("id", "?"))
4304 def router(state: PipelineState) -> str:
4305 marker = _foreach_flat(state, channel_names).get(marker_key)
4306 if marker is None:
4307 return end
4308 items = marker.get("items") or []
4309 if marker.get("index", 0) >= len(items):
4310 return exit_id
4311 max_output = marker.get("max_output", 0)
4312 if max_output and marker.get("produced_outputs", 0) >= max_output:
4313 return exit_id
4314 return first_id
4316 return router
4319def _make_foreach_body_node(
4320 sub: dict,
4321 step_runner: Any,
4322 channel_names: frozenset[str],
4323 channel_types: dict[str, str],
4324) -> Any:
4325 """A foreach body step as a node: a nested loop subgraph or a step node."""
4326 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
4327 if sub.get(_SUBGRAPH_WHILE_KEY):
4328 return _make_while_subgraph_node(sub, step_runner, channel_names)
4329 if sub.get(_SUBGRAPH_FOREACH_KEY):
4330 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
4331 return _make_subgraph_node(
4332 sub[_SUBGRAPH_GRAPH_KEY], sub.get("condition"), step_runner, channel_names
4333 )
4334 return _make_step_node(sub, step_runner, channel_names, channel_types)
4337def _make_foreach_agent_node(
4338 step: dict,
4339 step_runner: Any,
4340 channel_names: frozenset[str],
4341 channel_types: dict[str, str],
4342) -> Any:
4343 """Agent-path node: invoke the foreach step's agent for the current item."""
4344 step_id = step.get("id", "?")
4346 def node_fn(state: PipelineState) -> PipelineState:
4347 flat = _foreach_flat(state, channel_names)
4348 resolved = step_runner.resolve_agent(step)
4349 ctx = step_runner._make_ctx(flat)
4350 resolved_step = step_runner._resolve_step_runtime(step, flat, ctx)
4351 resolved_step = step_runner._attach_communication_context(resolved_step, flat)
4352 resolved_step = step_runner._attach_resolved_tools(resolved_step, resolved)
4353 result_state = step_runner._invoke_agent(resolved, resolved_step, flat)
4354 if not channel_names:
4355 return result_state
4356 return {_CORE: _strip_channel_keys(result_state, channel_names)}
4358 return node_fn
4361def _make_foreach_cond_node(
4362 step: dict,
4363 step_runner: Any,
4364 channel_names: frozenset[str],
4365) -> Any:
4366 """Collect the current item's result and advance the item index."""
4367 step_id = step.get("id", "?")
4368 marker_key = _foreach_marker_key(step_id)
4369 declared_output = step.get("output")
4370 runs_body = bool(step.get("body"))
4372 def node_fn(state: PipelineState) -> PipelineState:
4373 flat = _foreach_flat(state, channel_names)
4374 marker = dict(flat.get(marker_key) or {})
4375 results = list(marker.get("results") or [])
4376 produced = marker.get("produced_outputs", 0)
4377 if runs_body:
4378 results.append(StepRunner._body_item_result(flat, declared_output))
4379 publish_out = flat.get("publish")
4380 if isinstance(publish_out, dict) and publish_out.get("path"):
4381 produced += 1
4382 else:
4383 results.append(flat.get(step_id))
4384 marker.update(
4385 {
4386 "results": results,
4387 "index": marker.get("index", 0) + 1,
4388 "produced_outputs": produced,
4389 }
4390 )
4391 return _foreach_update({marker_key: marker}, channel_names)
4393 return node_fn
4396def _make_foreach_cond_router(
4397 step: dict,
4398 channel_names: frozenset[str],
4399 head_id: str,
4400 exit_id: str,
4401) -> Any:
4402 """Route from the cond: exit on exhaustion or budget, else back to the head."""
4403 marker_key = _foreach_marker_key(step.get("id", "?"))
4405 def router(state: PipelineState) -> str:
4406 marker = _foreach_flat(state, channel_names).get(marker_key) or {}
4407 items = marker.get("items") or []
4408 if marker.get("index", 0) >= len(items):
4409 return exit_id
4410 max_output = marker.get("max_output", 0)
4411 if max_output and marker.get("produced_outputs", 0) >= max_output:
4412 return exit_id
4413 return head_id
4415 return router
4418def _make_foreach_exit_node(
4419 step: dict,
4420 step_runner: Any,
4421 channel_names: frozenset[str],
4422) -> Any:
4423 """Write the documented foreach aggregate result."""
4424 step_id = step.get("id", "?")
4425 marker_key = _foreach_marker_key(step_id)
4426 runs_body = bool(step.get("body"))
4428 def node_fn(state: PipelineState) -> PipelineState:
4429 flat = _foreach_flat(state, channel_names)
4430 marker = flat.get(marker_key) or {}
4431 results = list(marker.get("results") or [])
4432 step_result = (
4433 StepRunner._aggregate_body_results(results) if runs_body else results
4434 )
4435 return _foreach_update({step_id: step_result}, channel_names)
4437 return node_fn
4440def _make_foreach_subgraph_node(
4441 placeholder: dict,
4442 step_runner: Any,
4443 channel_names: frozenset[str],
4444) -> Any:
4445 """Parent node for a compiled sequential `foreach` cycle: isolate and surface.
4447 The cycle shares the graph state schema, so invoking it directly would leak
4448 body step results and loop plumbing into the parent. This wrapper runs the
4449 cycle on a copy and returns only the documented aggregate plus the side
4450 channels. A condition-skipped entry writes no step result (matching the
4451 in-node soft-skip), so the step id is only surfaced when the cycle wrote it.
4452 """
4453 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4454 step_id = placeholder["id"]
4455 config = {"recursion_limit": placeholder.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
4457 if not channel_names:
4458 def node_fn(state: PipelineState) -> PipelineState:
4459 sub = compiled.invoke(state, config)
4460 if step_id not in sub:
4461 return {}
4462 result = {step_id: sub.get(step_id)}
4463 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4464 if key in sub and sub.get(key) != state.get(key):
4465 result[key] = sub.get(key)
4466 return result
4467 return node_fn
4469 def node_fn(state: PipelineState) -> PipelineState:
4470 sub = compiled.invoke(state, config)
4471 flat = _flatten_channel_state(sub)
4472 parent_flat = _flatten_channel_state(state)
4473 if step_id not in flat:
4474 return {}
4475 result = {step_id: flat.get(step_id)}
4476 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4477 if key in flat and flat.get(key) != parent_flat.get(key):
4478 result[key] = flat.get(key)
4479 return {_CORE: _strip_channel_keys(result, channel_names)}
4481 return node_fn
4484def _collect_parallel_channel_writes(
4485 steps: list[dict],
4486 pre_state: dict,
4487 core: dict,
4488 step_runner: Any,
4489 channel_names: frozenset[str],
4490 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
4491) -> list[tuple[str, Any]]:
4492 """Collect one parallel node's declared writes in branch order.
4494 Every branch write is emitted, so fan-in feeds the channel-type reducer
4495 every write in deterministic declaration/branch order: Topic appends each
4496 write, LastValue keeps the last, and BinaryOperatorAggregate combines them
4497 per its accumulated shape (list concat, collection-map union, dict merge,
4498 scalar last). A branch's isolated body writes are emitted before its own
4499 declared write, mirroring the top-level node's route-after-run order.
4500 """
4501 writes: list[tuple[str, Any]] = []
4502 for step in steps:
4503 if branch_working_sets is not None:
4504 branch_working_set = branch_working_sets.get(step["id"])
4505 if branch_working_set is not None:
4506 writes.extend(branch_working_set.pending_writes())
4507 branch = _route_declared_channels(
4508 step, pre_state, core, step_runner, channel_names
4509 )
4510 for channel_name, value in branch.items():
4511 writes.append((channel_name, value))
4512 return writes
4515def _make_parallel_node(
4516 group_steps: list[dict],
4517 step_runner: Any,
4518 channel_names: frozenset[str],
4519 channel_types: dict[str, str],
4520) -> Any:
4521 def node_fn(state: PipelineState) -> PipelineState:
4522 base = (
4523 _flatten_channel_state(state)
4524 if channel_names else state
4525 )
4526 working_set = (
4527 ChannelWorkingSet(base, channel_types) if channel_names else None
4528 )
4529 # Each branch writes into its own fork; the shared set is advanced only
4530 # at fan-in, so branch-internal body writes never race.
4531 branch_working_sets = (
4532 {s["id"]: working_set.fork() for s in group_steps}
4533 if working_set is not None else None
4534 )
4536 def submit_branch(pool: Any, step: dict) -> Any:
4537 if branch_working_sets is None:
4538 return pool.submit(step_runner.run_step, step, base)
4539 return pool.submit(
4540 step_runner.run_step, step, base, branch_working_sets[step["id"]]
4541 )
4543 results: dict[str, Any] = {}
4544 branch_metrics: dict = {}
4545 branch_tool_meta: dict = {}
4546 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
4547 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
4548 for future in as_completed(futures):
4549 sid = futures[future]
4550 try:
4551 out = future.result()
4552 results[sid] = out.get(sid)
4553 branch_metrics.update(out.get("step_context_metrics") or {})
4554 branch_tool_meta.update(out.get("step_tool_meta") or {})
4555 except SkillLoadError:
4556 raise
4557 except RunHardFailure:
4558 pool.shutdown(wait=False, cancel_futures=True)
4559 raise
4560 except Exception as exc: # noqa: BLE001
4561 log.error("Step '%s' failed: %s", sid, exc)
4562 results[sid] = {"error": str(exc)}
4563 if channel_names:
4564 merged = _strip_channel_keys({**base, **results}, channel_names)
4565 if branch_metrics:
4566 merged["step_context_metrics"] = _merge_side_channel(
4567 merged.get("step_context_metrics"), branch_metrics
4568 )
4569 if branch_tool_meta:
4570 merged["step_tool_meta"] = _merge_side_channel(
4571 merged.get("step_tool_meta"), branch_tool_meta
4572 )
4573 writes = _collect_parallel_channel_writes(
4574 group_steps, base, merged, step_runner, channel_names,
4575 branch_working_sets,
4576 )
4577 written_names = {name for name, _ in writes}
4578 for name, value in writes:
4579 working_set.stage(name, value)
4580 channels = working_set.commit()
4581 emitted = {name: channels[name] for name in written_names}
4582 return {_CORE: {**merged, **emitted}}
4583 final: PipelineState = {**state, **results}
4584 if branch_metrics:
4585 final = {
4586 **final,
4587 "step_context_metrics": _merge_side_channel(
4588 state.get("step_context_metrics"), branch_metrics
4589 ),
4590 }
4591 if branch_tool_meta:
4592 final = {
4593 **final,
4594 "step_tool_meta": _merge_side_channel(
4595 state.get("step_tool_meta"), branch_tool_meta
4596 ),
4597 }
4598 return final
4599 return node_fn
4602@dataclass(frozen=True)
4603class _MapReduceParts:
4604 """The nodes and edges of one concurrent step's Send fan-out.
4606 ``dispatch_id`` reuses the step id so the step's incoming dependency edges are
4607 unchanged; ``join_id`` carries the step's outgoing edges.
4608 """
4610 dispatch_id: str
4611 worker_id: str
4612 fallback_id: str
4613 join_id: str
4614 dispatch_fn: Any
4615 route_fn: Any
4616 worker_fn: Any
4617 join_fn: Any
4620@dataclass(frozen=True)
4621class _MapReduceItemOutcome:
4622 """One worker's per-item result plus its errors, side channels and writes.
4624 ``errors`` carries the item's soft errors (and the inherited soft errors it
4625 read), so the join can fold them in item order exactly like the in-node
4626 concurrent path. ``channel_writes`` is empty for the parallel strategy (its
4627 channel writes are derived from the item result at the join) and carries the
4628 foreach body's pending writes so the join can stage them in item order.
4629 """
4631 result: Any
4632 metrics: dict | None
4633 tool_meta: dict | None
4634 channel_writes: tuple[tuple[str, Any], ...]
4635 errors: tuple[str, ...] = ()
4638@dataclass(frozen=True)
4639class _MapReducePlan:
4640 """The per-step behaviour injected into the shared Send machinery.
4642 The parallel strategy runs a single agent call per item; the foreach
4643 strategy runs the body (or agent) per item. ``item_lister`` resolves the
4644 fan-out list, ``item_runner`` executes one item, ``result_assembler``
4645 shapes the step result, and ``channel_write_stager`` stages per-item
4646 channel writes at the join.
4647 """
4649 item_lister: Any
4650 item_runner: Any
4651 result_assembler: Any
4652 channel_write_stager: Any
4655def _map_reduce_worker_payload(flat: dict, item: Any, index: int) -> dict:
4656 """Build a worker's ``Send`` payload from the resolved source state."""
4657 return {**flat, "item": item, "_item": item, _MAP_REDUCE_ITEM_INDEX: index}
4660def _map_reduce_worker_output(
4661 step_id: str,
4662 index: int,
4663 outcome: _MapReduceItemOutcome,
4664) -> PipelineState:
4665 """The accumulator record carrying every per-item fold input.
4667 The slot is ``{index: {"result", "writes", "errors", "metrics",
4668 "tool_meta"}}`` so the join can assemble the step result, stage channel
4669 writes, and fold soft errors plus side channels in item order — the same
4670 observable result as the in-node concurrent path.
4671 """
4672 record = {
4673 "result": outcome.result,
4674 "writes": list(outcome.channel_writes),
4675 "errors": list(outcome.errors),
4676 "metrics": outcome.metrics if isinstance(outcome.metrics, dict) else None,
4677 "tool_meta": (
4678 outcome.tool_meta if isinstance(outcome.tool_meta, dict) else None
4679 ),
4680 }
4681 return {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: {index: record}}}
4684def _make_map_reduce_route_fn(
4685 plan: _MapReducePlan,
4686 step_runner: Any,
4687 channel_names: frozenset[str],
4688 worker_id: str,
4689 fallback_id: str,
4690 join_id: str,
4691) -> Any:
4692 def route(state: PipelineState) -> list:
4693 flat = _flatten_channel_state(state) if channel_names else state
4694 ctx = step_runner._make_ctx(flat)
4695 items = plan.item_lister(flat, ctx)
4696 if items is None:
4697 return [Send(fallback_id, state)]
4698 if not items:
4699 return [Send(join_id, state)]
4700 return [
4701 Send(worker_id, _map_reduce_worker_payload(flat, item, index))
4702 for index, item in enumerate(items)
4703 ]
4705 return route
4708def _make_map_reduce_worker_fn(
4709 step: dict,
4710 plan: _MapReducePlan,
4711 step_runner: Any,
4712 gate: threading.Semaphore,
4713) -> Any:
4714 step_id = step.get("id", "?")
4716 def worker(state: PipelineState) -> PipelineState:
4717 index = state[_MAP_REDUCE_ITEM_INDEX]
4718 item_state = {
4719 key: value
4720 for key, value in state.items()
4721 if key != _MAP_REDUCE_ITEM_INDEX
4722 }
4723 item = item_state.get("item")
4724 item_ctx = step_runner._make_ctx(item_state)
4725 with gate:
4726 try:
4727 outcome = plan.item_runner(
4728 state, index, item, item_state, item_ctx
4729 )
4730 except SkillLoadError:
4731 raise
4732 except RunHardFailure:
4733 raise
4734 except Exception as exc: # noqa: BLE001
4735 log.error("[step] %s item failed: %s", step_id, exc)
4736 outcome = _MapReduceItemOutcome(
4737 {"error": str(exc)}, None, None, ()
4738 )
4739 return _map_reduce_worker_output(step_id, index, outcome)
4741 return worker
4744def _fold_map_reduce_errors(records: list, base_errors: list) -> list:
4745 """Fold every item's soft errors into *base_errors* in item order."""
4746 merged = list(base_errors)
4747 for record in records:
4748 if not isinstance(record, dict):
4749 continue
4750 merged = StepRunner._merge_errors(merged, record.get("errors") or [])
4751 return merged
4754def _fold_map_reduce_side_channel(records: list, record_key: str) -> dict:
4755 """Fold every item's side-channel map in item order (later item wins)."""
4756 folded: dict = {}
4757 for record in records:
4758 if not isinstance(record, dict):
4759 continue
4760 value = record.get(record_key)
4761 if isinstance(value, dict):
4762 folded.update(value)
4763 return folded
4766def _fold_map_reduce_extras(flat: dict, records: list) -> dict:
4767 """The join's per-item state extras: soft errors plus folded side channels.
4769 Mirrors the in-node concurrent path: errors are merged in item order with
4770 dedup, and the budget/side-channel counters follow the same item order.
4771 """
4772 extras: dict = {}
4773 base_errors = list(flat.get("errors") or [])
4774 errors = _fold_map_reduce_errors(records, base_errors)
4775 if errors != base_errors:
4776 extras["errors"] = errors
4777 for state_key, record_key in (
4778 ("step_context_metrics", "metrics"),
4779 ("step_tool_meta", "tool_meta"),
4780 ):
4781 folded = _fold_map_reduce_side_channel(records, record_key)
4782 if folded:
4783 extras[state_key] = _merge_side_channel(flat.get(state_key), folded)
4784 return extras
4787def _make_map_reduce_join_fn(
4788 step: dict,
4789 plan: _MapReducePlan,
4790 step_runner: Any,
4791 channel_names: frozenset[str],
4792 channel_types: dict[str, str],
4793) -> Any:
4794 step_id = step.get("id", "?")
4796 def join(state: PipelineState) -> PipelineState:
4797 flat = _flatten_channel_state(state) if channel_names else state
4798 accumulated = state.get(_MAP_REDUCE_ACCUMULATOR_KEY) or {}
4799 entries = accumulated.get(step_id) or {}
4800 clear = {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: None}}
4801 ctx = step_runner._make_ctx(flat)
4802 items = plan.item_lister(flat, ctx)
4803 if items is None:
4804 return clear
4805 records = [entries.get(index) for index in range(len(items))]
4806 results = [
4807 record.get("result") if isinstance(record, dict) else None
4808 for record in records
4809 ]
4810 step_result = plan.result_assembler(results)
4811 extras = _fold_map_reduce_extras(flat, records)
4812 if not channel_names:
4813 return {step_id: step_result, **extras, **clear}
4814 working_set = ChannelWorkingSet(flat, channel_types)
4815 plan.channel_write_stager(results, records, working_set)
4816 written_names = {name for name, _ in working_set.pending_writes()}
4817 channels = working_set.commit()
4818 emitted = {name: channels[name] for name in written_names}
4819 core = _strip_channel_keys(
4820 {**flat, step_id: step_result, **extras}, channel_names
4821 )
4822 return {_CORE: {**core, **emitted}, **clear}
4824 return join
4827def _make_parallel_map_reduce_plan(
4828 step: dict, step_runner: Any
4829) -> _MapReducePlan:
4830 """Plan for a concurrent `parallel` step: one agent call per item."""
4831 resolved_agent = step_runner.resolve_agent(step)
4833 def run_item(state, index, item, item_state, item_ctx):
4834 result, metrics, tool_meta, errors = step_runner._invoke_parallel_item(
4835 step, resolved_agent, item, item_state, item_ctx
4836 )
4837 return _MapReduceItemOutcome(
4838 result, metrics, tool_meta, (), tuple(errors)
4839 )
4841 return _MapReducePlan(
4842 item_lister=lambda flat, ctx: step_runner._parallel_fanout_item_list(
4843 step, flat, ctx
4844 ),
4845 item_runner=run_item,
4846 result_assembler=lambda results: {"results": results},
4847 channel_write_stager=lambda results, records, working_set: (
4848 step_runner._stage_fanout_channel_writes(step, results, working_set)
4849 ),
4850 )
4853def _worker_committed_flat(
4854 state: PipelineState, channel_names: frozenset[str]
4855) -> dict:
4856 """The committed channel snapshot visible to one worker.
4858 A channels-path worker receives the flattened snapshot as its payload, so
4859 its channel keys sit at the top level; the typed ``__core__`` shape (only
4860 seen by the join) is unwrapped.
4861 """
4862 if _CORE in state:
4863 return _flatten_channel_state(state)
4864 return {name: state[name] for name in channel_names if name in state}
4867def _pending_channel_writes(
4868 working_set: ChannelWorkingSet | None,
4869) -> tuple[tuple[str, Any], ...]:
4870 if working_set is None:
4871 return ()
4872 return tuple(working_set.pending_writes())
4875def _stage_map_reduce_item_channel_writes(
4876 results: list, records: list, working_set: ChannelWorkingSet
4877) -> None:
4878 """Stage each item's carried channel writes in item order."""
4879 for record in records:
4880 if not isinstance(record, dict):
4881 continue
4882 for name, value in record.get("writes") or ():
4883 working_set.stage(name, value)
4886def _make_foreach_map_reduce_plan(
4887 step: dict,
4888 step_runner: Any,
4889 channel_names: frozenset[str],
4890 channel_types: dict[str, str],
4891) -> _MapReducePlan:
4892 """Plan for a concurrent `foreach` step: the body (or agent) runs per item."""
4893 step_id = step.get("id", "?")
4894 body = step.get("body") or []
4895 declared_output = step.get("output")
4896 runs_body = bool(body)
4898 def result_assembler(results):
4899 if runs_body:
4900 return StepRunner._aggregate_body_results(results)
4901 return results
4903 def run_item(state, index, item, item_state, _item_ctx):
4904 item_working_set = (
4905 ChannelWorkingSet(
4906 _worker_committed_flat(state, channel_names), channel_types
4907 )
4908 if channel_names
4909 else None
4910 )
4911 iter_state = {
4912 **item_state, "item": item, "_item": item, "_foreach_item": item
4913 }
4914 if runs_body:
4915 for sub_step in body:
4916 iter_state = step_runner._run_body_step(
4917 sub_step, iter_state, item_working_set
4918 )
4919 return _MapReduceItemOutcome(
4920 step_runner._body_item_result(iter_state, declared_output),
4921 iter_state.get("step_context_metrics"),
4922 iter_state.get("step_tool_meta"),
4923 _pending_channel_writes(item_working_set),
4924 tuple(iter_state.get("errors") or []),
4925 )
4926 resolved = step_runner.resolve_agent(step)
4927 item_read_state = step_runner._read_state(iter_state, item_working_set)
4928 item_ctx = step_runner._make_ctx(item_read_state)
4929 resolved_step = step_runner._resolve_step_runtime(
4930 step, item_read_state, item_ctx
4931 )
4932 resolved_step = step_runner._attach_communication_context(
4933 resolved_step, item_read_state
4934 )
4935 resolved_step = step_runner._attach_resolved_tools(
4936 resolved_step, resolved
4937 )
4938 result_state = step_runner._invoke_agent(
4939 resolved, resolved_step, iter_state
4940 )
4941 return _MapReduceItemOutcome(
4942 result_state.get(step_id),
4943 result_state.get("step_context_metrics"),
4944 result_state.get("step_tool_meta"),
4945 _pending_channel_writes(item_working_set),
4946 tuple(result_state.get("errors") or []),
4947 )
4949 return _MapReducePlan(
4950 item_lister=lambda flat, ctx: step_runner._foreach_item_list(
4951 step, flat, ctx
4952 ),
4953 item_runner=run_item,
4954 result_assembler=result_assembler,
4955 channel_write_stager=_stage_map_reduce_item_channel_writes,
4956 )
4959def _make_map_reduce_parts(
4960 step: dict,
4961 plan: _MapReducePlan,
4962 step_runner: Any,
4963 channel_names: frozenset[str],
4964 channel_types: dict[str, str],
4965 max_concurrent: int,
4966) -> _MapReduceParts:
4967 step_id = step.get("id", "?")
4968 worker_id = f"{step_id}__worker"
4969 fallback_id = f"{step_id}__fallback"
4970 join_id = f"{step_id}__join"
4971 gate = threading.Semaphore(max(1, max_concurrent))
4972 return _MapReduceParts(
4973 dispatch_id=step_id,
4974 worker_id=worker_id,
4975 fallback_id=fallback_id,
4976 join_id=join_id,
4977 dispatch_fn=lambda state: {},
4978 route_fn=_make_map_reduce_route_fn(
4979 plan, step_runner, channel_names, worker_id, fallback_id, join_id
4980 ),
4981 worker_fn=_make_map_reduce_worker_fn(step, plan, step_runner, gate),
4982 join_fn=_make_map_reduce_join_fn(
4983 step, plan, step_runner, channel_names, channel_types
4984 ),
4985 )
4988def _resolved_send_concurrency(step: dict, build_ctx: dict) -> int:
4989 """The build-time resolved ``max_concurrent`` for a Send-eligible step."""
4990 return _resolve_int(step.get("max_concurrent"), build_ctx, 1)
4993def _is_send_parallel_step(
4994 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4995) -> bool:
4996 """Whether a top-level `parallel` step fans out through native ``Send``.
4998 Restricted to a singleton group's statically concurrent, agent-backed,
4999 item-sourced step with no `routes`, no `condition`/`skip_llm_if`, and an
5000 explicit `max_concurrent > 1`; everything else keeps the in-node path.
5001 """
5002 if step.get("type") != "parallel":
5003 return False
5004 if step.get("id") in route_plan.target_node_ids:
5005 return False
5006 if step.get("routes") or not step.get("agent"):
5007 return False
5008 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
5009 return False
5010 if StepRunner._declared_item_source(step) is None:
5011 return False
5012 if step.get("max_concurrent") is None:
5013 return False
5014 return _resolved_send_concurrency(step, build_ctx) > 1
5017def _is_send_foreach_step(
5018 step: dict, route_plan: "_RoutePlan", build_ctx: dict
5019) -> bool:
5020 """Whether a top-level `foreach` step fans out through native ``Send``.
5022 Restricted to a singleton group's body- or agent-backed, item-sourced step
5023 with no `routes`, no `condition`/`skip_llm_if`, no step-level
5024 `output_channel`, and no `max_output`; an explicit `max_concurrent > 1` is
5025 required. The default (absent or ``1``) keeps the sequential in-node path,
5026 and the excluded edges keep their existing in-node semantics.
5027 """
5028 if step.get("type") != "foreach":
5029 return False
5030 if step.get("id") in route_plan.target_node_ids:
5031 return False
5032 if step.get("routes") or step.get("output_channel"):
5033 return False
5034 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
5035 return False
5036 if step.get("max_output") is not None:
5037 return False
5038 if StepRunner._declared_item_source(step) is None:
5039 return False
5040 if not (step.get("body") or step.get("agent")):
5041 return False
5042 if step.get("max_concurrent") is None:
5043 return False
5044 return _resolved_send_concurrency(step, build_ctx) > 1
5047class _ChannelGraphShim:
5048 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
5050 Preserves the invoke contract used by main.py and tests: a flat dict in,
5051 a flat dict out. The flat input is seeded through ``seed_committed`` and
5052 nested under ``__core__`` for the compiled graph; the committed snapshot is
5053 unwrapped back out. Streaming and state surfaces are not exposed because
5054 they would leak or accept the internal ``__core__`` shape; read-only
5055 introspection delegates to the compiled graph.
5056 """
5058 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
5059 self._compiled = compiled
5060 self._specs = specs
5062 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
5063 seeded = seed_committed(self._specs, flat_state)
5064 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
5065 return unwrap(result)
5067 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
5068 seeded = seed_committed(self._specs, flat_state)
5069 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
5070 return unwrap(result)
5072 def stream(self, *args: Any, **kwargs: Any) -> Any:
5073 self._reject_internal_shape("stream")
5075 def astream(self, *args: Any, **kwargs: Any) -> Any:
5076 self._reject_internal_shape("astream")
5078 def update_state(self, *args: Any, **kwargs: Any) -> Any:
5079 self._reject_internal_shape("update_state")
5081 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
5082 self._reject_internal_shape("aupdate_state")
5084 def get_state(self, *args: Any, **kwargs: Any) -> Any:
5085 self._reject_internal_shape("get_state")
5087 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
5088 self._reject_internal_shape("aget_state")
5090 def _reject_internal_shape(self, surface: str) -> None:
5091 raise NotImplementedError(
5092 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
5093 ".invoke()/.ainvoke() contract is exposed through the channel "
5094 "boundary shim"
5095 )
5097 def __getattr__(self, name: str) -> Any:
5098 return getattr(self._compiled, name)
5101class GraphBuilder:
5102 """
5103 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
5105 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
5106 käytetään vain {{ }} -templatejen evaluointiin.
5107 """
5109 def __init__(
5110 self,
5111 registry: ManifestRegistry,
5112 builtins: BuiltinExecutor | None = None,
5113 invoke_agent_fn: Any | None = None,
5114 run_context: dict | None = None,
5115 tracer: Any | None = None,
5116 ) -> None:
5117 self._registry = registry
5118 self._resolver = AgentResolver(registry)
5119 self._invoke_agent_fn = invoke_agent_fn
5120 self._builtins = builtins or _DEFAULT_BUILTINS
5121 if invoke_agent_fn is not None:
5122 self._builtins.invoke_agent_fn = invoke_agent_fn
5123 self._step_functions = StepFunctionExecutor(registry)
5124 self._tool_registry = build_platform_registry()
5125 self._tool_resolver = ToolResolver(self._tool_registry, self._step_functions)
5126 self._step_runner = StepRunner(
5127 self._resolver, self._builtins, invoke_agent_fn, run_context,
5128 step_functions=self._step_functions,
5129 tracer=tracer,
5130 tool_resolver=self._tool_resolver,
5131 )
5133 def build(self, pipeline: dict): # -> CompiledGraph
5134 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
5135 try:
5136 from langgraph.graph import StateGraph, END
5137 except ImportError as exc:
5138 raise ImportError(
5139 "langgraph is required to build a workflow. "
5140 "Install it with: pip install langgraph"
5141 ) from exc
5143 raw_steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
5145 channels: list[dict] = _validated_channels(
5146 (pipeline.get("spec") or {}).get("channels")
5147 )
5148 specs = parse_channel_specs(
5149 [{"name": ch["name"], "type": ch.get("type", LAST_VALUE)} for ch in channels]
5150 )
5151 channel_types = {spec.name: spec.type for spec in specs}
5152 channel_names = frozenset(channel_types)
5154 # Build-time call-step composition (ADR-0037 D4): each invoked Function
5155 # body is compiled into its own native LangGraph subgraph node, so the
5156 # runtime graph executes only native steps; a call step never survives.
5157 steps = self._compose_calls(
5158 raw_steps, channels, channel_names, channel_types
5159 )
5160 if channels:
5161 _validate_channel_names(channels, steps)
5162 _validate_channel_inputs(channels, steps)
5163 _validate_step_channel_bindings(steps, channel_names)
5164 ordered = _topological_order(steps)
5166 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
5167 envs = (pipeline.get("spec") or {}).get("envs")
5168 if envs is not None and not isinstance(envs, dict):
5169 raise RunHardFailure(
5170 "VALIDATION_ERROR",
5171 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
5172 "spec.envs must be a map of env name → image ref (ADR-0012)",
5173 )
5175 graph = self._assemble_graph(ordered, channel_names, channel_types)
5176 compiled = graph.compile()
5177 if channel_names:
5178 return _ChannelGraphShim(compiled, specs)
5179 return compiled
5181 def _compose_calls(
5182 self,
5183 steps: list[dict],
5184 channels: list[dict],
5185 channel_names: frozenset[str],
5186 channel_types: dict[str, str],
5187 ) -> list[dict]:
5188 """Replace `call` steps with compiled native subgraph nodes.
5190 Each invoked Function body is assembled through ``_assemble_graph`` and
5191 compiled, then substituted as a single ``__subgraph__`` placeholder that
5192 keeps the call id and its ``depends_on``. Downstream steps addressing the
5193 call id therefore address the subgraph node unchanged.
5194 """
5196 def assemble_subgraph(sub_steps: list[dict]):
5197 return self._assemble_graph(
5198 sub_steps, channel_names, channel_types
5199 ).compile()
5201 def assemble_while(while_step: dict, body_steps: list[dict]):
5202 return self._assemble_while_graph(
5203 while_step, body_steps, channel_names, channel_types
5204 )
5206 def assemble_foreach(foreach_step: dict, body_steps: list[dict]):
5207 return self._assemble_foreach_graph(
5208 foreach_step, body_steps, channel_names, channel_types
5209 )
5211 composer = _CallComposer(
5212 self._registry,
5213 assemble_subgraph,
5214 channels,
5215 channel_names,
5216 assemble_while=assemble_while,
5217 compose_whiles=True,
5218 assemble_foreach=assemble_foreach,
5219 compose_foreachs=True,
5220 build_ctx=self._step_runner._make_ctx({}),
5221 )
5222 return composer.compose(steps)
5224 def _assemble_while_graph(
5225 self,
5226 step: dict,
5227 body_steps: list[dict],
5228 channel_names: frozenset[str],
5229 channel_types: dict[str, str],
5230 ) -> tuple[Any, int]:
5231 """Compile a `while` step's body into a native LangGraph cycle.
5233 The cycle is ``head -> body... -> cond`` with a conditional back-edge
5234 from ``cond`` to ``head`` (continue) or ``exit`` (termination): the
5235 entry condition/initial value live in the head, the feed + exit
5236 evaluation in the cond node, and the exit node writes the documented
5237 result. Returns the compiled cycle and the recursion bound the wrapper
5238 must apply (a cycle consumes one super-step per node per iteration).
5239 """
5240 from langgraph.graph import StateGraph, END
5242 step_id = step.get("id", "?")
5243 step_runner = self._step_runner
5244 step_runner.set_channel_routing(channel_names)
5246 state_schema = (
5247 _build_state_schema()
5248 if channel_names
5249 else Annotated[dict[str, Any], _merge_state]
5250 )
5251 graph: StateGraph = StateGraph(state_schema)
5253 max_iter = _resolve_int(
5254 step.get("max_iterations", 2), step_runner._make_ctx({}), 2
5255 )
5256 head_id = f"__while_{step_id}__head"
5257 cond_id = f"__while_{step_id}__cond"
5258 exit_id = f"__while_{step_id}__exit"
5259 first_id = body_steps[0]["id"] if body_steps else cond_id
5261 graph.add_node(
5262 head_id,
5263 _make_while_head_node(step, step_runner, channel_names, max_iter),
5264 )
5265 for sub in body_steps:
5266 graph.add_node(
5267 sub["id"],
5268 _make_while_body_node(
5269 sub, step_runner, channel_names, channel_types
5270 ),
5271 )
5272 graph.add_node(
5273 cond_id, _make_while_cond_node(step, step_runner, channel_names)
5274 )
5275 graph.add_node(
5276 exit_id, _make_while_exit_node(step, step_runner, channel_names)
5277 )
5279 graph.set_entry_point(head_id)
5280 head_paths = {first_id: first_id, exit_id: exit_id, END: END}
5281 graph.add_conditional_edges(
5282 head_id,
5283 _make_while_head_router(step, channel_names, first_id, exit_id, END),
5284 head_paths,
5285 )
5286 for previous, sub in zip(body_steps, body_steps[1:]):
5287 graph.add_edge(previous["id"], sub["id"])
5288 if body_steps:
5289 graph.add_edge(body_steps[-1]["id"], cond_id)
5290 graph.add_conditional_edges(
5291 cond_id,
5292 _make_while_cond_router(step, channel_names, head_id, exit_id),
5293 {head_id: head_id, exit_id: exit_id},
5294 )
5295 graph.add_edge(exit_id, END)
5297 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5298 return graph.compile(), recursion_limit
5300 def _assemble_foreach_graph(
5301 self,
5302 step: dict,
5303 body_steps: list[dict],
5304 channel_names: frozenset[str],
5305 channel_types: dict[str, str],
5306 ) -> tuple[Any, int]:
5307 """Compile a sequential `foreach` step's body into a native LangGraph cycle.
5309 The cycle is ``head -> (body... | agent) -> cond`` with a conditional
5310 back-edge from ``cond`` to ``head`` (next item) or ``exit`` (exhaustion
5311 or `max_output` reached). The head resolves the item list once and
5312 injects the current item; the cond collects the item result; the exit
5313 node writes the documented aggregate. Returns the compiled cycle and
5314 the recursion bound the wrapper must apply.
5315 """
5316 from langgraph.graph import StateGraph, END
5318 step_id = step.get("id", "?")
5319 step_runner = self._step_runner
5320 step_runner.set_channel_routing(channel_names)
5322 state_schema = (
5323 _build_state_schema()
5324 if channel_names
5325 else Annotated[dict[str, Any], _merge_state]
5326 )
5327 graph: StateGraph = StateGraph(state_schema)
5329 max_iter = _resolve_int(
5330 step.get("max_iterations"),
5331 step_runner._make_ctx({}),
5332 _FOREACH_DEFAULT_MAX_ITER,
5333 )
5334 head_id = f"__foreach_{step_id}__head"
5335 cond_id = f"__foreach_{step_id}__cond"
5336 exit_id = f"__foreach_{step_id}__exit"
5337 agent_id = f"__foreach_{step_id}__agent"
5339 graph.add_node(
5340 head_id, _make_foreach_head_node(step, step_runner, channel_names)
5341 )
5342 if body_steps:
5343 for sub in body_steps:
5344 graph.add_node(
5345 sub["id"],
5346 _make_foreach_body_node(
5347 sub, step_runner, channel_names, channel_types
5348 ),
5349 )
5350 first_id = body_steps[0]["id"]
5351 else:
5352 graph.add_node(
5353 agent_id,
5354 _make_foreach_agent_node(
5355 step, step_runner, channel_names, channel_types
5356 ),
5357 )
5358 first_id = agent_id
5359 graph.add_node(
5360 cond_id, _make_foreach_cond_node(step, step_runner, channel_names)
5361 )
5362 graph.add_node(
5363 exit_id, _make_foreach_exit_node(step, step_runner, channel_names)
5364 )
5366 graph.set_entry_point(head_id)
5367 graph.add_conditional_edges(
5368 head_id,
5369 _make_foreach_head_router(step, channel_names, first_id, exit_id, END),
5370 {first_id: first_id, exit_id: exit_id, END: END},
5371 )
5372 for previous, sub in zip(body_steps, body_steps[1:]):
5373 graph.add_edge(previous["id"], sub["id"])
5374 if body_steps:
5375 graph.add_edge(body_steps[-1]["id"], cond_id)
5376 else:
5377 graph.add_edge(agent_id, cond_id)
5378 graph.add_conditional_edges(
5379 cond_id,
5380 _make_foreach_cond_router(step, channel_names, head_id, exit_id),
5381 {head_id: head_id, exit_id: exit_id},
5382 )
5383 graph.add_edge(exit_id, END)
5385 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5386 return graph.compile(), recursion_limit
5388 def _assemble_graph(
5389 self,
5390 steps: list[dict],
5391 channel_names: frozenset[str],
5392 channel_types: dict[str, str],
5393 ) -> "StateGraph":
5394 from langgraph.graph import StateGraph, END
5396 # Group steps into LangGraph nodes:
5397 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
5398 nodes: list[list[dict]] = []
5399 for step in steps:
5400 dep_sig = frozenset(step.get("depends_on") or [])
5401 if nodes:
5402 last_group_sig = frozenset(
5403 dep
5404 for s in nodes[-1]
5405 for dep in (s.get("depends_on") or [])
5406 )
5407 last_types = {s.get("type") for s in nodes[-1]}
5408 if (
5409 dep_sig == last_group_sig
5410 and step.get("type") == "parallel"
5411 and last_types == {"parallel"}
5412 and not step.get("routes")
5413 and not any(s.get("routes") for s in nodes[-1])
5414 ):
5415 nodes[-1].append(step)
5416 continue
5417 nodes.append([step])
5419 state_schema = (
5420 _build_state_schema()
5421 if channel_names
5422 else Annotated[dict[str, Any], _merge_state]
5423 )
5424 graph: StateGraph = StateGraph(state_schema)
5425 step_runner = self._step_runner
5426 step_runner.set_channel_routing(channel_names)
5428 node_ids: list[str] = []
5429 for group in nodes:
5430 if len(group) == 1:
5431 node_ids.append(group[0]["id"])
5432 else:
5433 node_ids.append("parallel_" + "_".join(s["id"] for s in group))
5435 route_plan = _build_route_plan(nodes, node_ids) if nodes else None
5437 send_parts: dict[int, _MapReduceParts] = {}
5438 if route_plan is not None:
5439 build_ctx = step_runner._make_ctx({})
5440 for index, group in enumerate(nodes):
5441 if len(group) != 1:
5442 continue
5443 send_candidate = group[0]
5444 if _is_send_parallel_step(send_candidate, route_plan, build_ctx):
5445 plan = _make_parallel_map_reduce_plan(
5446 send_candidate, step_runner
5447 )
5448 elif _is_send_foreach_step(send_candidate, route_plan, build_ctx):
5449 plan = _make_foreach_map_reduce_plan(
5450 send_candidate, step_runner, channel_names, channel_types
5451 )
5452 else:
5453 continue
5454 send_parts[index] = _make_map_reduce_parts(
5455 send_candidate,
5456 plan,
5457 step_runner,
5458 channel_names,
5459 channel_types,
5460 _resolved_send_concurrency(send_candidate, build_ctx),
5461 )
5463 entry_ids: list[str] = []
5464 exit_ids: list[str] = []
5465 for index, group in enumerate(nodes):
5466 parts = send_parts.get(index)
5467 if parts is None:
5468 captured_group = list(group)
5469 if len(captured_group) == 1:
5470 first = captured_group[0]
5471 if first.get("type") == _SUBGRAPH_STEP_TYPE:
5472 if first.get(_SUBGRAPH_WHILE_KEY):
5473 subgraph_node = _make_while_subgraph_node(
5474 first, step_runner, channel_names
5475 )
5476 elif first.get(_SUBGRAPH_FOREACH_KEY):
5477 subgraph_node = _make_foreach_subgraph_node(
5478 first, step_runner, channel_names
5479 )
5480 else:
5481 subgraph_node = _make_subgraph_node(
5482 first[_SUBGRAPH_GRAPH_KEY],
5483 first.get("condition"),
5484 step_runner,
5485 channel_names,
5486 )
5487 graph.add_node(node_ids[index], subgraph_node)
5488 entry_ids.append(node_ids[index])
5489 exit_ids.append(node_ids[index])
5490 continue
5491 fn = _make_step_node(
5492 first, step_runner, channel_names, channel_types,
5493 )
5494 else:
5495 fn = _make_parallel_node(
5496 captured_group,
5497 step_runner,
5498 channel_names,
5499 channel_types,
5500 )
5501 graph.add_node(node_ids[index], fn)
5502 entry_ids.append(node_ids[index])
5503 exit_ids.append(node_ids[index])
5504 continue
5505 fallback_step = group[0]
5506 graph.add_node(parts.dispatch_id, parts.dispatch_fn)
5507 graph.add_node(parts.worker_id, parts.worker_fn)
5508 graph.add_node(
5509 parts.fallback_id,
5510 _make_step_node(
5511 fallback_step, step_runner, channel_names, channel_types
5512 ),
5513 )
5514 graph.add_node(parts.join_id, parts.join_fn)
5515 graph.add_conditional_edges(
5516 parts.dispatch_id,
5517 parts.route_fn,
5518 [parts.worker_id, parts.fallback_id],
5519 )
5520 graph.add_edge(parts.worker_id, parts.join_id)
5521 graph.add_edge(parts.fallback_id, parts.join_id)
5522 entry_ids.append(parts.dispatch_id)
5523 exit_ids.append(parts.join_id)
5525 if nodes:
5526 _wire_dependency_edges(
5527 graph, nodes, node_ids, END, route_plan, entry_ids, exit_ids
5528 )
5529 _wire_route_edges(graph, route_plan, END)
5530 else:
5531 graph.add_node("noop", lambda s: s)
5532 graph.set_entry_point("noop")
5533 graph.add_edge("noop", END)
5535 return graph
5538# ---------------------------------------------------------------------------
5539# Julkinen API
5540# ---------------------------------------------------------------------------
5542_registry_cache: ManifestRegistry | None = None
5545def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
5546 global _registry_cache
5547 if _registry_cache is None or dirs is not None:
5548 loader = ManifestLoader(dirs)
5549 manifests = loader.load_all()
5550 _registry_cache = ManifestRegistry(manifests)
5551 return _registry_cache
5554def build_workflow(
5555 pipeline_name: str,
5556 *,
5557 dirs: list[Path] | None = None,
5558 builtins: BuiltinExecutor | None = None,
5559 invoke_agent_fn: Any | None = None,
5560 run_context: dict | None = None,
5561 tracer: Any | None = None,
5562):
5563 """
5564 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
5565 LangGraph-workflown.
5567 Args:
5568 pipeline_name: Pipeline metadata.name -arvo.
5569 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
5570 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
5571 invoke_agent_fn: fn(resolved_spec, step, state) → state.
5572 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
5574 Returns:
5575 Compilattu LangGraph CompiledGraph.
5576 """
5577 registry = _get_registry(dirs)
5578 pipeline = registry.get("Pipeline", pipeline_name)
5579 if pipeline is None:
5580 available = [
5581 (m.get("metadata") or {}).get("name")
5582 for m in registry.all_of_kind("Pipeline")
5583 ]
5584 raise RunHardFailure(
5585 "VALIDATION_ERROR",
5586 f"Pipeline '{pipeline_name}' not found. Available: {available}",
5587 )
5588 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
5589 return builder.build(pipeline)