Coverage for core/langgraph/runner.py: 94%
2558 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 05:04 +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 composed_step_ids = (
2004 step.get(_SUBGRAPH_STEP_IDS_KEY)
2005 if not step.get(_SUBGRAPH_WHILE_KEY)
2006 and not step.get(_SUBGRAPH_FOREACH_KEY)
2007 else None
2008 )
2009 if working_set is not None:
2010 # Channel path: the compiled subgraph shares the parent's
2011 # Core-namespaced schema, while a loop body carries the flat
2012 # committed snapshot. Re-wrap on entry, flatten on exit. No
2013 # channel reducer is re-applied here — the subgraph's own
2014 # commits already produced the authoritative values.
2015 sub_result = compiled.invoke({_CORE: read_state}, config)
2016 committed = _flatten_channel_state(sub_result)
2017 return _filter_composed_step_ids(
2018 {**state, **committed}, composed_step_ids
2019 )
2020 return _filter_composed_step_ids(
2021 compiled.invoke(state, config), composed_step_ids
2022 )
2024 if step_type == "spawn":
2025 return self._run_spawn(step, state, working_set)
2027 raise RunHardFailure(
2028 "VALIDATION_ERROR",
2029 f"Unknown step type '{step_type}' in step '{step_id}'",
2030 {"step_id": step_id, "step_type": step_type},
2031 )
2033 def _run_parallel_fanout(
2034 self,
2035 step: dict,
2036 state: PipelineState,
2037 items: list,
2038 ctx: dict,
2039 working_set: ChannelWorkingSet | None = None,
2040 ) -> PipelineState:
2041 """
2042 Fan-out: call agent once per item, concurrently.
2043 Results stored as state[step_id] = {"results": [item_result, ...]}.
2044 Each item is available as state["item"] inside the invoke call.
2045 """
2046 step_id = step.get("id", "?")
2047 agent_name = step.get("agent")
2048 if not agent_name:
2049 log.warning("[step] %s parallel fanout: no agent", step_id)
2050 return {**state, step_id: {"results": []}}
2052 resolved = self._resolver.resolve(agent_name)
2054 # Apply max_items cap
2055 max_items_raw = step.get("max_items")
2056 if max_items_raw is not None:
2057 max_items = _resolve_int(max_items_raw, ctx, len(items))
2058 items = items[:max_items]
2060 # Concurrency limit
2061 max_conc_raw = step.get("max_concurrent")
2062 if max_conc_raw is not None:
2063 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2064 else:
2065 max_conc = len(items) or 1
2066 max_conc = max(1, min(max_conc, len(items) or 1))
2068 log.debug(
2069 "[step] %s parallel fanout: %d items, %d concurrent",
2070 step_id, len(items), max_conc,
2071 )
2073 read_state = self._read_state(state, working_set)
2075 def call_one(item: Any) -> tuple[Any, dict | None, dict | None, list]:
2076 item_state = {**read_state, "_item": item, "item": item}
2077 item_ctx = self._make_ctx(item_state)
2078 return self._invoke_parallel_item(
2079 step, resolved, item, item_state, item_ctx
2080 )
2082 # Pre-allocate results to preserve input order regardless of completion
2083 # order (ENG-07), matching the concurrent foreach body-polku.
2084 results: list = [None] * len(items)
2085 item_metrics: dict = {}
2086 item_tool_meta: dict = {}
2087 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2088 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
2089 for future in as_completed(futures):
2090 idx = futures[future]
2091 try:
2092 item_result, item_metrics_map, item_tool_meta_map, *_ = (
2093 future.result()
2094 )
2095 results[idx] = item_result
2096 if isinstance(item_metrics_map, dict):
2097 item_metrics.update(item_metrics_map)
2098 if isinstance(item_tool_meta_map, dict):
2099 item_tool_meta.update(item_tool_meta_map)
2100 except SkillLoadError:
2101 raise
2102 except RunHardFailure:
2103 pool.shutdown(wait=False, cancel_futures=True)
2104 raise
2105 except Exception as exc: # noqa: BLE001
2106 log.error("[step] %s item failed: %s", step_id, exc)
2107 results[idx] = {"error": str(exc)}
2109 self._stage_fanout_channel_writes(step, results, working_set)
2111 final: PipelineState = {**state, step_id: {"results": results}}
2112 if item_metrics:
2113 final = {
2114 **final,
2115 "step_context_metrics": _merge_side_channel(
2116 state.get("step_context_metrics"), item_metrics
2117 ),
2118 }
2119 if item_tool_meta:
2120 final = {
2121 **final,
2122 "step_tool_meta": _merge_side_channel(
2123 state.get("step_tool_meta"), item_tool_meta
2124 ),
2125 }
2126 return final
2128 def _invoke_parallel_item(
2129 self,
2130 step: dict,
2131 resolved: dict,
2132 item: Any,
2133 item_state: PipelineState,
2134 item_ctx: dict,
2135 ) -> tuple[Any, dict | None, dict | None, list]:
2136 """Run one parallel fan-out item's agent call.
2138 Shared by the in-node thread-pool fan-out and the native ``Send``
2139 worker, so both paths resolve input, communication context and tools
2140 identically. Returns ``(step_result, metrics, tool_meta, errors)``.
2141 """
2142 step_id = step.get("id", "?")
2143 resolved_step = self._resolve_step_runtime(
2144 step, item_state, item_ctx,
2145 input_override=item,
2146 )
2147 resolved_step = self._attach_communication_context(
2148 resolved_step, item_state
2149 )
2150 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2151 result_state = self._invoke_agent(resolved, resolved_step, item_state)
2152 return (
2153 result_state.get(step_id),
2154 result_state.get("step_context_metrics"),
2155 result_state.get("step_tool_meta"),
2156 list(result_state.get("errors") or []),
2157 )
2159 def _stage_fanout_channel_writes(
2160 self,
2161 step: dict,
2162 item_outputs: list,
2163 working_set: ChannelWorkingSet | None,
2164 ) -> None:
2165 """Stage each fan-out item's declared channel writes in item order.
2167 Runs on the main thread after the item pool completes, so the shared
2168 working set is never mutated from a worker thread. The fan-out
2169 envelope itself is never routed; only each item's own agent output is.
2170 """
2171 if working_set is None or not self._channel_names:
2172 return
2173 for output in item_outputs:
2174 writes = _declared_output_channel_writes(
2175 step, output, self._channel_names
2176 )
2177 for name, value in writes.items():
2178 working_set.stage(name, value)
2180 def _run_foreach(
2181 self,
2182 step: dict,
2183 state: PipelineState,
2184 working_set: ChannelWorkingSet | None = None,
2185 ) -> PipelineState:
2186 """
2187 Foreach: iterate over input list, run body steps per item.
2188 Input from step.input (dot-path or template) or step.over (state key).
2190 When max_concurrent > 1, body iterations run concurrently via
2191 ThreadPoolExecutor — each item gets its own worker thread. The
2192 max_concurrent value sets the thread pool size (mirrors parallel
2193 fan-out semantics). Results list preserves original item order.
2195 max_output budget in concurrent mode: a stop flag prevents new workers
2196 from starting after the budget is exhausted. Workers already executing
2197 when the flag is set will complete naturally — at most max_concurrent
2198 extra items may run past the limit. This is accepted behaviour; the
2199 sequential early-exit guarantee cannot be reproduced without sacrificing
2200 true concurrency.
2202 The invocation working set is threaded into every body step so reads
2203 resolve over the committed snapshot plus this invocation's pending
2204 writes; the returned state is still built from the raw ``state``.
2205 """
2206 step_id = step.get("id", "?")
2207 read_state = self._read_state(state, working_set)
2208 ctx = self._make_ctx(read_state)
2210 items = self._foreach_item_list(step, read_state, ctx)
2212 # max_output cap — stop once N iterations have produced a successful
2213 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
2214 max_output_raw = step.get("max_output")
2215 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
2217 # max_concurrent — when > 1, enables parallel body execution
2218 max_conc_raw = step.get("max_concurrent")
2219 if max_conc_raw is not None:
2220 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2221 max_conc = max(1, min(max_conc, len(items) or 1))
2222 else:
2223 max_conc = 1
2225 body = step.get("body") or []
2226 agent_name = step.get("agent")
2228 if max_conc <= 1:
2229 raise RunHardFailure(
2230 "VALIDATION_ERROR",
2231 f"foreach step '{step_id}' reached runtime without build-time "
2232 "composition (a sequential foreach is compiled into a native "
2233 "cycle) — internal invariant violated",
2234 {"step_id": step_id},
2235 )
2236 return self._run_foreach_concurrent(
2237 step_id, step, state, items, body, agent_name, max_output, max_conc,
2238 working_set,
2239 )
2241 @staticmethod
2242 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
2243 """body-polku per-iteration result (r.67): the declared body-output
2244 step's result; else the iteration's `publish` result; else None."""
2245 if declared_output:
2246 return iter_state.get(declared_output)
2247 return iter_state.get("publish")
2249 @staticmethod
2250 def _body_item_passed(item: Any) -> bool:
2251 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
2252 `passed` bool; an item lacking such a key is not counted as failed."""
2253 if isinstance(item, dict):
2254 verdict = item.get("verdict")
2255 if verdict == "FAIL":
2256 return False
2257 if verdict == "PASS":
2258 return True
2259 passed = item.get("passed")
2260 if isinstance(passed, bool):
2261 return passed
2262 return True
2264 @staticmethod
2265 def _aggregate_body_results(item_results: list) -> dict:
2266 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
2267 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
2268 return {
2269 "results": item_results,
2270 "passed": not failing,
2271 "failing": failing,
2272 "failed": bool(failing),
2273 }
2275 @staticmethod
2276 def _merge_item_working_sets(
2277 working_set: ChannelWorkingSet | None,
2278 item_working_sets: list[ChannelWorkingSet | None],
2279 ) -> None:
2280 """Stage each concurrent item's writes onto the shared set in item order."""
2281 if working_set is None:
2282 return
2283 for item_working_set in item_working_sets:
2284 if item_working_set is None:
2285 continue
2286 for name, value in item_working_set.pending_writes():
2287 working_set.stage(name, value)
2289 def _run_foreach_concurrent(
2290 self,
2291 step_id: str,
2292 step: dict,
2293 state: PipelineState,
2294 items: list,
2295 body: list,
2296 agent_name: str | None,
2297 max_output: int,
2298 max_conc: int,
2299 working_set: ChannelWorkingSet | None = None,
2300 ) -> PipelineState:
2301 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2302 log.info(
2303 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2304 step_id, len(items), max_conc,
2305 )
2307 # Pre-allocate results list to preserve item order across threads.
2308 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2309 # legitimate None results (body produced no output) are kept.
2310 body_polku = bool(body)
2311 declared_output = step.get("output")
2312 _SKIPPED = object()
2313 results: list = [_SKIPPED] * len(items)
2314 item_working_sets: list[ChannelWorkingSet | None] = [None] * len(items)
2315 lock = threading.Lock()
2316 produced_outputs = [0] # mutable counter shared across threads
2317 stop_flag = threading.Event() # set when max_output budget is exhausted
2318 item_errors: list = [] # budget errors collected from worker threads
2319 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2320 accumulated_tool_meta: dict = {} # per-item tool metadata from worker threads
2322 def run_item(idx: int, item: Any) -> None:
2323 nonlocal item_errors
2324 # Honour stop flag — skip if budget already exhausted before we start
2325 if max_output and stop_flag.is_set():
2326 return
2328 item_working_set = (
2329 working_set.fork() if working_set is not None else None
2330 )
2331 item_working_sets[idx] = item_working_set
2332 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2334 if body_polku:
2335 for sub_step in body:
2336 iter_state = self._run_body_step(
2337 sub_step, iter_state, item_working_set
2338 )
2339 # Collect budget errors from this iteration (thread-safe merge)
2340 new_errs = iter_state.get("errors") or []
2341 if new_errs:
2342 with lock:
2343 item_errors = self._merge_errors(item_errors, new_errs)
2344 item_metrics_map = iter_state.get("step_context_metrics")
2345 if isinstance(item_metrics_map, dict):
2346 with lock:
2347 accumulated_metrics.update(item_metrics_map)
2348 item_tool_meta_map = iter_state.get("step_tool_meta")
2349 if isinstance(item_tool_meta_map, dict):
2350 with lock:
2351 accumulated_tool_meta.update(item_tool_meta_map)
2352 publish_out = iter_state.get("publish")
2353 results[idx] = self._body_item_result(iter_state, declared_output)
2354 if isinstance(publish_out, dict) and publish_out.get("path"):
2355 with lock:
2356 produced_outputs[0] += 1
2357 if max_output and produced_outputs[0] >= max_output:
2358 stop_flag.set()
2359 elif agent_name:
2360 resolved = self._resolver.resolve(agent_name)
2361 item_read_state = self._read_state(iter_state, item_working_set)
2362 item_ctx = self._make_ctx(item_read_state)
2363 resolved_step = self._resolve_step_runtime(
2364 step, item_read_state, item_ctx
2365 )
2366 resolved_step = self._attach_communication_context(
2367 resolved_step, item_read_state
2368 )
2369 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2370 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2371 results[idx] = iter_state.get(step_id)
2372 item_metrics_map = iter_state.get("step_context_metrics")
2373 if isinstance(item_metrics_map, dict):
2374 with lock:
2375 accumulated_metrics.update(item_metrics_map)
2376 item_tool_meta_map = iter_state.get("step_tool_meta")
2377 if isinstance(item_tool_meta_map, dict):
2378 with lock:
2379 accumulated_tool_meta.update(item_tool_meta_map)
2381 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2382 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2383 for future in as_completed(futures):
2384 try:
2385 future.result()
2386 except SkillLoadError:
2387 raise
2388 except RunHardFailure:
2389 pool.shutdown(wait=False, cancel_futures=True)
2390 raise
2391 except Exception as exc: # noqa: BLE001
2392 idx = futures[future]
2393 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2394 results[idx] = {"error": str(exc)}
2396 self._merge_item_working_sets(working_set, item_working_sets)
2398 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2399 ordered = [r for r in results if r is not _SKIPPED]
2400 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
2401 final: PipelineState = {**state, step_id: step_result}
2402 if accumulated_metrics:
2403 final = {
2404 **final,
2405 "step_context_metrics": _merge_side_channel(
2406 state.get("step_context_metrics"), accumulated_metrics
2407 ),
2408 }
2409 if accumulated_tool_meta:
2410 final = {
2411 **final,
2412 "step_tool_meta": _merge_side_channel(
2413 state.get("step_tool_meta"), accumulated_tool_meta
2414 ),
2415 }
2416 if item_errors:
2417 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2418 return final
2420 def _run_spawn(
2421 self,
2422 step: dict,
2423 state: PipelineState,
2424 working_set: ChannelWorkingSet | None = None,
2425 ) -> PipelineState:
2426 """Run a spawn step as a native dynamic ``Send`` fan-out subgraph.
2428 The ``agent`` runs once as director at the tree root and may emit a
2429 child plan at ``envelope.content.children``; a shared child-worker node
2430 re-runs the ``child`` role (default = ``agent``) for each child and
2431 re-``Send``s itself for deeper levels. Depth and fan are bounded by the
2432 spawn_policy, and the whole tree draws from one shared SpawnBudget.
2433 """
2434 step_id = step.get("id", "?")
2435 director = step.get("agent")
2436 if not director:
2437 log.warning("[step] %s spawn has no agent — skipping", step_id)
2438 return state
2439 read_state = self._read_state(state, working_set)
2440 ctx = self._make_ctx(read_state)
2441 policy = step.get("spawn_policy") or {}
2442 max_depth = _resolve_int(
2443 policy.get("max_depth"), ctx, _SPAWN_DEFAULT_MAX_DEPTH
2444 )
2445 max_spawn_per_node = _resolve_int(
2446 policy.get("max_spawn_per_node"), ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
2447 )
2448 max_calls_raw = policy.get("max_agent_calls")
2449 budget = SpawnBudget(
2450 _SPAWN_DEFAULT_MAX_AGENT_CALLS
2451 if max_calls_raw is None
2452 else _resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS)
2453 )
2454 run_max_parallel = int(
2455 (self._run_context.get("budget") or {}).get(
2456 "max_parallel", DEFAULT_MAX_PARALLEL
2457 )
2458 )
2459 max_concurrent_raw = policy.get("max_concurrent_children")
2460 max_concurrent_children = (
2461 run_max_parallel
2462 if max_concurrent_raw is None
2463 else _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
2464 )
2465 child_role = step.get("child") or director
2467 outcome = self._run_native_spawn(
2468 step=step,
2469 step_id=step_id,
2470 director=director,
2471 child_role=child_role,
2472 base_state=read_state,
2473 base_ctx=ctx,
2474 budget=budget,
2475 max_depth=max_depth,
2476 max_spawn_per_node=max_spawn_per_node,
2477 max_concurrent_children=max_concurrent_children,
2478 )
2479 envelope = outcome["envelope"]
2480 final: PipelineState = {**state, step_id: envelope}
2481 if outcome["metrics"]:
2482 final = {
2483 **final,
2484 "step_context_metrics": _merge_side_channel(
2485 state.get("step_context_metrics"), outcome["metrics"]
2486 ),
2487 }
2488 if outcome["tool_meta"]:
2489 final = {
2490 **final,
2491 "step_tool_meta": _merge_side_channel(
2492 state.get("step_tool_meta"), outcome["tool_meta"]
2493 ),
2494 }
2495 self._stage_spawn_channel_writes(step, envelope["result"], working_set)
2496 return final
2498 def _run_native_spawn(
2499 self,
2500 *,
2501 step: dict,
2502 step_id: str,
2503 director: str,
2504 child_role: str,
2505 base_state: PipelineState,
2506 base_ctx: dict,
2507 budget: SpawnBudget,
2508 max_depth: int,
2509 max_spawn_per_node: int | None,
2510 max_concurrent_children: int,
2511 ) -> dict:
2512 """Compile and invoke the native spawn tree, returning its outcome.
2514 The graph is a director node that seeds one ``Send`` per child to a
2515 shared worker, whose worker re-``Send``s itself for deeper levels. An
2516 outstanding-descendants barrier (``outstanding`` counter plus a ``check``
2517 node) makes the ``join`` run exactly once for uneven-depth trees; the
2518 join assembles the result envelope by path key in plan order.
2519 """
2520 from contextlib import nullcontext
2521 from typing import TypedDict
2523 from langgraph.graph import END, START, StateGraph
2524 from langgraph.types import Command, Send
2526 tracer = self._tracer
2527 parent_obs = (
2528 tracer.open_parent_observation(
2529 name=f"spawn:{step_id}",
2530 input={"node_id": step_id, "role": director, "depth": 0},
2531 )
2532 if tracer is not None
2533 else None
2534 )
2535 root_obs_id = getattr(parent_obs, "id", None)
2536 try:
2537 child_gates: dict[str, threading.Semaphore] = {}
2538 child_gates_lock = threading.Lock()
2540 def child_gate(parent_id: str) -> threading.Semaphore:
2541 """The concurrency gate for one spawning node's direct children.
2543 ``max_concurrent_children`` bounds a single spawn node's fan-out:
2544 each spawning node owns its own gate, so a shallow node's cap
2545 never throttles a deeper sibling's subtree.
2546 """
2547 with child_gates_lock:
2548 gate = child_gates.get(parent_id)
2549 if gate is None:
2550 gate = threading.Semaphore(max(1, max_concurrent_children))
2551 child_gates[parent_id] = gate
2552 return gate
2554 def run_node(
2555 *,
2556 role: str,
2557 node_step: dict,
2558 input_override: Any,
2559 node_id: str,
2560 depth: int,
2561 parent_id: str | None = None,
2562 ) -> dict:
2563 resolved_role = self._resolver.resolve(role)
2564 resolved_step = self._resolve_step_runtime(
2565 node_step, base_state, base_ctx, input_override=input_override
2566 )
2567 resolved_step = self._attach_communication_context(
2568 resolved_step, base_state
2569 )
2570 resolved_step = self._attach_resolved_tools(
2571 resolved_step, resolved_role
2572 )
2573 if not budget.consume():
2574 raise RunHardFailure(
2575 "SPAWN_BUDGET_EXHAUSTED",
2576 f"spawn tree budget exhausted at node '{node_id}'",
2577 {"node_id": node_id, "depth": depth},
2578 )
2579 gate = child_gate(parent_id) if parent_id is not None else nullcontext()
2580 with gate:
2581 return self._invoke_agent(
2582 resolved_role, resolved_step, base_state
2583 )
2585 def build_record(
2586 *,
2587 node_id: str,
2588 path: tuple,
2589 parent_path: tuple | None,
2590 depth: int,
2591 this_result: Any,
2592 result_state: dict,
2593 children: list,
2594 ) -> tuple[dict, list]:
2595 exhausted = False
2596 plan: list = []
2597 if children:
2598 if depth + 1 > max_depth:
2599 exhausted = True
2600 else:
2601 plan = [
2602 {
2603 "path": path + (index,),
2604 "depth": depth + 1,
2605 "child": child,
2606 "parent_id": node_id,
2607 "index": index,
2608 }
2609 for index, child in enumerate(children)
2610 ]
2611 record = {
2612 "path": path,
2613 "parent_path": parent_path,
2614 "child_id": node_id,
2615 "result": this_result,
2616 "self_passed": not self._spawn_is_soft_fail(this_result),
2617 "exhausted": exhausted,
2618 "metrics": dict(result_state.get("step_context_metrics") or {}),
2619 "tool_meta": dict(result_state.get("step_tool_meta") or {}),
2620 }
2621 return record, plan
2623 def child_plan(node_id: str, this_result: Any, depth: int) -> list:
2624 children = self._spawn_children_from_result(this_result)
2625 if children and max_spawn_per_node is not None and len(children) > max_spawn_per_node:
2626 raise RunHardFailure(
2627 "SPAWN_OVER_FLIGHT",
2628 f"spawn node '{node_id}' emitted {len(children)} children, "
2629 f"exceeds max_spawn_per_node={max_spawn_per_node}",
2630 {
2631 "node_id": node_id,
2632 "depth": depth,
2633 "children": len(children),
2634 "max_spawn_per_node": max_spawn_per_node,
2635 },
2636 )
2637 return children
2639 def director_node(_state: dict) -> Command:
2640 result_state = run_node(
2641 role=director,
2642 node_step=step,
2643 input_override=_UNSET,
2644 node_id=step_id,
2645 depth=0,
2646 )
2647 this_result = result_state.get(step_id)
2648 children = child_plan(step_id, this_result, 0)
2649 record, plan = build_record(
2650 node_id=step_id,
2651 path=(),
2652 parent_path=None,
2653 depth=0,
2654 this_result=this_result,
2655 result_state=result_state,
2656 children=children,
2657 )
2658 update = {
2659 "accum": {step_id: {(): record}},
2660 "outstanding": len(plan) - 1,
2661 }
2662 if not plan:
2663 return Command(update=update, goto="join")
2664 return Command(
2665 update=update,
2666 goto=[
2667 Send("worker", {**payload, "parent_obs_id": root_obs_id})
2668 for payload in plan
2669 ],
2670 )
2672 def worker_node(payload: dict) -> Command:
2673 path = tuple(payload["path"])
2674 depth = payload["depth"]
2675 parent_id = payload["parent_id"]
2676 index = payload["index"]
2677 child = payload["child"]
2678 child_id = (
2679 child["id"]
2680 if isinstance(child, dict) and child.get("id")
2681 else f"{parent_id}.{index}"
2682 )
2683 node_step = self._child_node_step(
2684 node_id=child_id,
2685 child=child,
2686 parent_id=parent_id,
2687 index=index,
2688 base_state=base_state,
2689 base_ctx=base_ctx,
2690 )
2691 # The worker's own graph context does not carry the spawning
2692 # node's observation (each Send target runs in a fresh context),
2693 # so the parent chain is re-established from the payload before
2694 # opening this node's observation (ADR-0035 D5).
2695 parent_obs_id = payload.get("parent_obs_id")
2696 stack_token = (
2697 _PARENT_STACK.set((parent_obs_id,))
2698 if tracer is not None and parent_obs_id is not None
2699 else None
2700 )
2701 child_obs = (
2702 tracer.open_parent_observation(
2703 name=f"spawn:{child_id}",
2704 input={"node_id": child_id, "role": child_role, "depth": depth},
2705 )
2706 if tracer is not None
2707 else None
2708 )
2709 try:
2710 result_state = run_node(
2711 role=child_role,
2712 node_step=node_step,
2713 input_override=node_step["input"],
2714 node_id=child_id,
2715 depth=depth,
2716 parent_id=parent_id,
2717 )
2718 finally:
2719 if tracer is not None and child_obs is not None:
2720 tracer.close_parent_observation(child_obs)
2721 if stack_token is not None:
2722 _PARENT_STACK.reset(stack_token)
2723 this_result = result_state.get(child_id)
2724 children = child_plan(child_id, this_result, depth)
2725 record, plan = build_record(
2726 node_id=child_id,
2727 path=path,
2728 parent_path=path[:-1],
2729 depth=depth,
2730 this_result=this_result,
2731 result_state=result_state,
2732 children=children,
2733 )
2734 update = {
2735 "accum": {step_id: {path: record}},
2736 "outstanding": len(plan) - 1,
2737 }
2738 if plan:
2739 child_obs_id = getattr(child_obs, "id", None)
2740 return Command(
2741 update=update,
2742 goto=[
2743 Send(
2744 "worker",
2745 {**payload, "parent_obs_id": child_obs_id},
2746 )
2747 for payload in plan
2748 ],
2749 )
2750 return Command(update=update, goto="check")
2752 def check_node(state: dict) -> Command:
2753 if state.get("outstanding") == 0:
2754 return Command(goto="join")
2755 return Command(goto=END)
2757 def join_node(state: dict) -> dict:
2758 records = (state.get("accum") or {}).get(step_id) or {}
2759 envelope, metrics, tool_meta = _assemble_spawn_outcome(records)
2760 return {
2761 "spawn_outcome": {
2762 "envelope": envelope,
2763 "metrics": metrics,
2764 "tool_meta": tool_meta,
2765 },
2766 "accum": {step_id: None},
2767 }
2769 schema = TypedDict(
2770 "_SpawnState",
2771 {
2772 "accum": Annotated[dict, accumulate_map_reduce],
2773 "outstanding": Annotated[int, operator.add],
2774 "spawn_outcome": dict,
2775 },
2776 )
2777 graph = StateGraph(schema)
2778 graph.add_node("director", director_node)
2779 graph.add_node("worker", worker_node)
2780 graph.add_node("check", check_node)
2781 graph.add_node("join", join_node)
2782 graph.add_edge(START, "director")
2783 graph.add_edge("join", END)
2784 app = graph.compile()
2785 out = app.invoke(
2786 {"accum": {}, "outstanding": 1},
2787 config={"recursion_limit": _SPAWN_RECURSION_LIMIT},
2788 )
2789 return out["spawn_outcome"]
2790 finally:
2791 if tracer is not None and parent_obs is not None:
2792 tracer.close_parent_observation(parent_obs)
2794 def _child_node_step(
2795 self,
2796 *,
2797 node_id: str,
2798 child: Any,
2799 parent_id: str,
2800 index: int,
2801 base_state: PipelineState,
2802 base_ctx: dict,
2803 ) -> dict:
2804 """Resolve one child plan item into the worker's runtime step.
2806 A child is a runtime-generated step: it carries only its id and its
2807 resolved input, never the spawn step's declared channel bindings, so a
2808 child's output can never route a channel write.
2809 """
2810 if not self._spawn_has_input(child):
2811 raise RunHardFailure(
2812 "INVALID_CHILD_PLAN",
2813 f"spawn child #{index} of node '{parent_id}' must be a dict "
2814 "with a non-empty 'input'",
2815 {"node_id": parent_id, "child_index": index, "child": repr(child)[:200]},
2816 )
2817 return {"id": node_id, "input": _resolve_input(child["input"], base_state, base_ctx)}
2819 def _stage_spawn_channel_writes(
2820 self,
2821 step: dict,
2822 result: Any,
2823 working_set: ChannelWorkingSet | None,
2824 ) -> None:
2825 """Stage a spawn node's declared writes from its own agent result."""
2826 if working_set is None:
2827 return
2828 writes = _declared_output_channel_writes(step, result, self._channel_names)
2829 for name, value in writes.items():
2830 working_set.stage(name, value)
2832 @staticmethod
2833 def _spawn_children_from_result(result: Any) -> list:
2834 """The node's child plan at ``content.children``; [] for a valid leaf.
2836 The director's parsed output envelope carries the plan at
2837 ``content.children``; a result that is not a dict, or whose ``content``
2838 is not an object, or carries no ``children``, is a valid leaf. A
2839 ``children`` value that is not a list is a hard ``INVALID_CHILD_PLAN``.
2840 """
2841 content = _envelope_content(result)
2842 if not isinstance(content, dict):
2843 return []
2844 raw = content.get("children")
2845 if raw is None:
2846 return []
2847 if not isinstance(raw, list):
2848 raise RunHardFailure(
2849 "INVALID_CHILD_PLAN",
2850 "spawn agent content 'children' must be a list",
2851 {"children": repr(raw)[:200]},
2852 )
2853 return raw
2855 @staticmethod
2856 def _spawn_has_input(child: Any) -> bool:
2857 """A valid child plan item: a dict carrying a non-empty 'input'."""
2858 if not isinstance(child, dict):
2859 return False
2860 input_val = child.get("input")
2861 if input_val is None:
2862 return False
2863 if isinstance(input_val, str) and not input_val.strip():
2864 return False
2865 if isinstance(input_val, (list, dict)) and not input_val:
2866 return False
2867 return True
2869 @staticmethod
2870 def _spawn_is_soft_fail(result: Any) -> bool:
2871 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error.
2873 An envelope-shaped result is never soft: ``status:error`` is a hard
2874 failure (the gate raises before this), and ``ok``/``partial``/
2875 ``no_success`` are valid non-fatal results.
2876 """
2877 if not isinstance(result, dict):
2878 return False
2879 if result.get("status") in ENVELOPE_STATUSES:
2880 return False
2881 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
2882 return True
2883 return any(key.startswith("budget:") for key in result)
2886# ---------------------------------------------------------------------------
2887# GraphBuilder
2888# ---------------------------------------------------------------------------
2890def _topological_order(steps: list[dict]) -> list[dict]:
2891 """
2892 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2893 """
2894 id_to_step = {s["id"]: s for s in steps}
2895 visited: set[str] = set()
2896 order: list[dict] = []
2898 def visit(step_id: str) -> None:
2899 if step_id in visited:
2900 return
2901 visited.add(step_id)
2902 for dep in (id_to_step[step_id].get("depends_on") or []):
2903 if dep in id_to_step:
2904 visit(dep)
2905 order.append(id_to_step[step_id])
2907 for s in steps:
2908 visit(s["id"])
2909 return order
2912def _group_dependencies(
2913 groups: list[list[dict]],
2914) -> tuple[list[set[int]], list[set[int]]]:
2915 """Group-index adjacency (predecessors, successors) from each step's depends_on."""
2916 step_to_group = {
2917 step["id"]: index
2918 for index, group in enumerate(groups)
2919 for step in group
2920 }
2921 predecessors: list[set[int]] = [set() for _ in groups]
2922 successors: list[set[int]] = [set() for _ in groups]
2923 for index, group in enumerate(groups):
2924 for step in group:
2925 for dep in step.get("depends_on") or []:
2926 dep_index = step_to_group.get(dep)
2927 if dep_index is None or dep_index == index:
2928 continue
2929 predecessors[index].add(dep_index)
2930 successors[dep_index].add(index)
2931 return predecessors, successors
2934@dataclass(frozen=True)
2935class _RoutePlan:
2936 """Compiled `routes` topology: route tables plus the static-edge suppression sets.
2938 `routes` owns a routing step's outgoing control flow (ADR-0025). A route
2939 target reached via a static `depends_on` edge is therefore suppressed and
2940 replaced by the native conditional edge; the plan records both the targets
2941 (so they are never made entry points) and the suppressed `(source, target)`
2942 pairs.
2943 """
2945 step_to_node: dict[str, str]
2946 routes_by_step: dict[str, list[dict]]
2947 target_node_ids: frozenset[str]
2948 routing_node_ids: frozenset[str]
2949 suppressed_edges: frozenset[tuple[str, str]]
2952def _reject_nested_routes(step: dict) -> None:
2953 """Reject `routes` on a nested `body` step (top-level-only rule, ADR-0012).
2955 The engine wires `routes` only for top-level steps, so a nested declaration
2956 would be silently ignored; nested routing is deferred.
2957 """
2958 body = step.get("body")
2959 if not isinstance(body, list):
2960 return
2961 for nested in _iter_steps(body):
2962 if nested.get("routes"):
2963 raise RunHardFailure(
2964 "VALIDATION_ERROR",
2965 f"Step '{nested.get('id')}' declares `routes` inside a nested "
2966 "body; `routes` is only allowed on top-level steps",
2967 )
2970def _build_route_plan(groups: list[list[dict]], node_ids: list[str]) -> _RoutePlan:
2971 """Resolve every top-level step's `routes` into a node-level routing plan."""
2972 step_to_node = {
2973 step["id"]: node_ids[index]
2974 for index, group in enumerate(groups)
2975 for step in group
2976 }
2977 step_to_deps = {
2978 step["id"]: set(step.get("depends_on") or [])
2979 for group in groups
2980 for step in group
2981 }
2982 routes_by_step: dict[str, list[dict]] = {}
2983 target_node_ids: set[str] = set()
2984 routing_node_ids: set[str] = set()
2985 suppressed_edges: set[tuple[str, str]] = set()
2986 for group in groups:
2987 for step in group:
2988 _reject_nested_routes(step)
2989 routes = step.get("routes")
2990 if not routes:
2991 continue
2992 source = step["id"]
2993 source_node = step_to_node[source]
2994 routing_node_ids.add(source_node)
2995 routes_by_step[source] = routes
2996 for route in routes:
2997 target = route.get("target")
2998 target_node = step_to_node.get(target)
2999 if target_node is None:
3000 raise RunHardFailure(
3001 "VALIDATION_ERROR",
3002 f"Step '{source}' route target '{target}' is not a "
3003 "top-level step in this pipeline",
3004 )
3005 target_node_ids.add(target_node)
3006 if source in step_to_deps.get(target, set()):
3007 suppressed_edges.add((source_node, target_node))
3008 log.warning(
3009 "[routes] suppressing static edge %s -> %s; route "
3010 "targets are wired via `routes`",
3011 source,
3012 target,
3013 )
3014 return _RoutePlan(
3015 step_to_node=step_to_node,
3016 routes_by_step=routes_by_step,
3017 target_node_ids=frozenset(target_node_ids),
3018 routing_node_ids=frozenset(routing_node_ids),
3019 suppressed_edges=frozenset(suppressed_edges),
3020 )
3023def _make_route_router(routes: list[dict], step_to_node: dict[str, str], end) -> Any:
3024 """First-match-wins router over a routing step's declared routes.
3026 Evaluates `when` in declaration order and returns the selected target node;
3027 a route without `when` is the default branch. With no match and no default
3028 it returns the graph terminal.
3029 """
3030 def router(state: PipelineState) -> str:
3031 ctx = _flatten_channel_state(state) if _CORE in state else state
3032 for route in routes:
3033 if _eval_condition(route.get("when"), ctx):
3034 return step_to_node[route["target"]]
3035 return end
3037 return router
3040def _wire_route_edges(graph, route_plan: _RoutePlan, end) -> None:
3041 """Add one native conditional edge per routing step, with an explicit path_map."""
3042 for source, routes in route_plan.routes_by_step.items():
3043 target_nodes = [route_plan.step_to_node[r["target"]] for r in routes]
3044 path_map = {node: node for node in target_nodes}
3045 path_map[end] = end
3046 graph.add_conditional_edges(
3047 route_plan.step_to_node[source],
3048 _make_route_router(routes, route_plan.step_to_node, end),
3049 path_map,
3050 )
3053def _wire_dependency_edges(
3054 graph,
3055 groups: list[list[dict]],
3056 node_ids: list[str],
3057 end,
3058 route_plan: _RoutePlan,
3059 entry_ids: list[str] | None = None,
3060 exit_ids: list[str] | None = None,
3061) -> None:
3062 """Wire native branches and fan-in joins from group-level depends_on edges.
3064 Static edges originating from a routing step to one of its route targets are
3065 suppressed — `routes` supplies that control flow. A route target with no
3066 other predecessor is reached by its conditional edge, never an entry point.
3067 A group split into a Send fan-out enters through its dispatch node and exits
3068 through its join node, so ``entry_ids``/``exit_ids`` name those boundaries.
3069 """
3070 entry_ids = entry_ids if entry_ids is not None else node_ids
3071 exit_ids = exit_ids if exit_ids is not None else node_ids
3072 predecessors, successors = _group_dependencies(groups)
3073 for index, node_id in enumerate(node_ids):
3074 entry = entry_ids[index]
3075 pred_nodes = [exit_ids[p] for p in sorted(predecessors[index])]
3076 if not pred_nodes:
3077 if node_id not in route_plan.target_node_ids:
3078 graph.set_entry_point(entry)
3079 else:
3080 live_preds = [
3081 pred
3082 for pred in pred_nodes
3083 if (pred, entry) not in route_plan.suppressed_edges
3084 ]
3085 if len(live_preds) == 1:
3086 graph.add_edge(live_preds[0], entry)
3087 elif live_preds:
3088 graph.add_edge(live_preds, entry)
3089 if not successors[index] and node_id not in route_plan.routing_node_ids:
3090 graph.add_edge(exit_ids[index], end)
3093# ---------------------------------------------------------------------------
3094# Call-step build-time expansion (ADR-0037 D3/D4)
3095# ---------------------------------------------------------------------------
3096#
3097# A `type: call` step is composed at build time: the function's body steps plus
3098# two synthesized op steps (`__call_bind` / `__call_emit`) are assembled into a
3099# compiled subgraph added as a single parent-graph node keyed by the call id. The
3100# runtime executes only native steps; LangGraph never sees a call step. The caller
3101# reads the function outputs via the call result shape
3102# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
3103# The function body's `{{ params.<name> }}` references are rewritten to
3104# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves. A
3105# call nested in a foreach/while body is flat-inlined instead, because loop bodies
3106# execute imperatively and cannot host a compiled subgraph node.
3107#
3108# Isolation (ADR-0037 D5 / 00-functions § Isolation) and recursion (ADR-0037 D7): a
3109# function body step must not reference a step-id outside its own body (a
3110# one-way dependency check — outward references to caller step-ids are build
3111# errors); a cycle in the function call graph (self or mutual) is a build
3112# error. Bounded while loops are runtime constructs and are not cycle-flagged.
3114# Call step result namespace keys (internal scaffolding).
3115def _scope_params_value(value: Any, call_id: str) -> Any:
3116 """Rewrite `params.` template references to the call-scoped namespace.
3118 Operates on the function body's string values (templates). Dict keys are
3119 untouched — only templated string VALUES are scoped, so a function body's
3120 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
3121 """
3122 if isinstance(value, str):
3123 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
3124 if isinstance(value, dict):
3125 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
3126 if isinstance(value, list):
3127 return [_scope_params_value(v, call_id) for v in value]
3128 return value
3131def _scope_params(step: dict, call_id: str) -> dict:
3132 """Deep-scope a function body step's params references for a call id."""
3133 scoped: dict = dict(step)
3134 for field in ("input", "params", "condition", "exit_condition", "agent", "function"):
3135 if field in scoped and isinstance(scoped[field], (str, dict, list)):
3136 scoped[field] = _scope_params_value(scoped[field], call_id)
3137 if "body" in scoped and isinstance(scoped["body"], list):
3138 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
3139 return scoped
3142def _collect_step_ids(steps: list[dict]) -> set[str]:
3143 """All step ids declared in *steps*, including nested body step ids."""
3144 ids: set[str] = set()
3145 for step in steps:
3146 step_id = step.get("id")
3147 if isinstance(step_id, str):
3148 ids.add(step_id)
3149 body = step.get("body")
3150 if isinstance(body, list):
3151 ids |= _collect_step_ids(body)
3152 return ids
3155def _iter_steps(steps: list[dict]):
3156 """Yield every step in *steps*, recursing into nested bodies."""
3157 for step in steps:
3158 yield step
3159 body = step.get("body")
3160 if isinstance(body, list):
3161 yield from _iter_steps(body)
3164def _collect_composed_step_ids(steps: list[dict]) -> set[str]:
3165 """All declared step ids, including ids composed into native subgraphs.
3167 A ``call`` step becomes a ``__subgraph__`` placeholder whose inner step ids
3168 live only inside the compiled subgraph. The placeholder records them under
3169 ``_SUBGRAPH_STEP_IDS_KEY`` so name-collision validation sees the full set
3170 even after body steps stop being textually inlined.
3171 """
3172 ids = _collect_step_ids(steps)
3173 for step in _iter_steps(steps):
3174 composed_ids = step.get(_SUBGRAPH_STEP_IDS_KEY)
3175 if isinstance(composed_ids, (set, frozenset)):
3176 ids |= composed_ids
3177 return ids
3180_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
3181_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
3182_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
3184_REFERENCE_FIELDS = ("input", "params", "condition", "exit_condition")
3186# Engine-reserved template namespaces — never step-id references, even if a
3187# caller happens to declare a step with one of these names.
3188_RESERVED_NAMESPACES = frozenset({"params", "loop", "item", "_item", "budget", "config"})
3191def _referenced_names(value: Any) -> set[str]:
3192 """Root identifiers read by templated strings.
3194 Only the leading identifier of each reference (before the first ``.`` or
3195 ``[``) can name a step-id; nested field segments are ignored. Quoted
3196 literals are ignored.
3197 """
3198 names: set[str] = set()
3199 if isinstance(value, str):
3200 for expr in _TEMPLATE_REF_RE.findall(value):
3201 bare = _STRING_LITERAL_RE.sub("", expr)
3202 for match in _IDENTIFIER_RE.finditer(bare):
3203 if match.start() == 0 or bare[match.start() - 1] != ".":
3204 names.add(match.group())
3205 elif isinstance(value, dict):
3206 for v in value.values():
3207 names |= _referenced_names(v)
3208 elif isinstance(value, list):
3209 for v in value:
3210 names |= _referenced_names(v)
3211 return names
3214def _validate_function_purity(
3215 fn_name: str, body: list[dict], visible_ids: set[str]
3216) -> None:
3217 """One-way isolation: a function body must not reference a caller step-id.
3219 A body step may read its own body step-ids and the params namespace; a
3220 reference to a step-id that exists outside the body (``visible_ids`` but
3221 not declared in the body) is an outward dependency and a hard build error.
3222 """
3223 local_ids = _collect_step_ids(body)
3224 for step in _iter_steps(body):
3225 step_id = step.get("id", "?")
3226 for dep in step.get("depends_on") or []:
3227 if dep in visible_ids and dep not in local_ids:
3228 raise RunHardFailure(
3229 "VALIDATION_ERROR",
3230 f"function '{fn_name}' step '{step_id}': depends_on "
3231 f"'{dep}' is outside the function body (isolation — "
3232 "function bodies see only their own steps and params)",
3233 )
3234 for field in _REFERENCE_FIELDS:
3235 for ref in _referenced_names(step.get(field)):
3236 if ref in _RESERVED_NAMESPACES:
3237 continue
3238 if ref in visible_ids and ref not in local_ids:
3239 raise RunHardFailure(
3240 "VALIDATION_ERROR",
3241 f"function '{fn_name}' step '{step_id}': '{field}' "
3242 f"references '{ref}', a step-id outside the function "
3243 "body (isolation — function bodies see only their own "
3244 "steps and params)",
3245 )
3248def _reject_function_body_routes(fn_name: str, body: list[dict]) -> None:
3249 """Reject `routes` anywhere in a function body (top-level-only rule).
3251 A function body is inlined to the top level, so a body step declaring
3252 `routes` would escape the nested-`body` guard and be silently rewired;
3253 reject it at build time instead.
3254 """
3255 for step in _iter_steps(body):
3256 if step.get("routes"):
3257 raise RunHardFailure(
3258 "VALIDATION_ERROR",
3259 f"function '{fn_name}' step '{step.get('id')}' declares `routes`; "
3260 "`routes` is only allowed on top-level steps",
3261 )
3264def _rewrite_call_deps(steps: list[dict], call_map: dict[str, str]) -> None:
3265 """Rewrite `depends_on` references to call ids → the corresponding emit id.
3267 After a call is expanded away, a downstream step that depends_on the call id
3268 must instead wait on the emit step (which writes state[<call_id>]). This
3269 guarantees the caller's `{{ <call_id>.outputs.<name> }}` reads resolve after
3270 the function body ran.
3271 """
3272 for step in steps:
3273 deps = step.get("depends_on") or []
3274 if deps:
3275 step["depends_on"] = [call_map.get(d, d) for d in deps]
3276 if "body" in step and isinstance(step["body"], list):
3277 _rewrite_call_deps(step["body"], call_map)
3280def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
3281 """Expand every ``call`` step through the native subgraph composer.
3283 Test-facing seam over the single call-expansion implementation: the same
3284 ``_CallComposer`` that ``GraphBuilder`` uses builds the steps, so no second
3285 (flat) expansion path exists. The embedded subgraph value is the un-compiled
3286 sub-step list; a runnable graph only needs the compiled form, which
3287 production obtains through ``GraphBuilder._compose_calls``.
3288 """
3289 composer = _CallComposer(
3290 registry,
3291 assemble_subgraph=lambda sub_steps: sub_steps,
3292 channels=[],
3293 channel_names=frozenset(),
3294 )
3295 return composer.compose(steps)
3298_SUBGRAPH_STEP_TYPE = "__subgraph__"
3299_SUBGRAPH_GRAPH_KEY = "__subgraph_graph__"
3300_SUBGRAPH_STEP_IDS_KEY = "__subgraph_step_ids__"
3302# A `while` step is composed into a `__subgraph__` placeholder whose embedded
3303# graph is a native LangGraph cycle (head -> body -> cond -> head/exit). The
3304# marker flags the placeholder so the parent uses the while node wrapper (which
3305# keeps the loop-internal state out of the parent result) rather than the plain
3306# call-subgraph node, and carries the build-time recursion bound.
3307_SUBGRAPH_WHILE_KEY = "__subgraph_while__"
3308_WHILE_RECURSION_LIMIT_KEY = "__while_recursion_limit__"
3310# Engine-local cycle state: the active while's `{current, iteration, passed,
3311# skipped, max_iter}` marker. Nested whiles live in isolated cycle invocations,
3312# so a shared key cannot collide across them.
3313_WHILE_STATE_KEY = "__while_state__"
3315# State keys the while wrapper propagates from the isolated cycle back to the
3316# parent (the loop result itself is carried separately under the step id).
3317_WHILE_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
3319# A sequential `foreach` (max_concurrent absent or resolving to <= 1) is
3320# composed into a `__subgraph__` placeholder whose embedded graph is a native
3321# LangGraph cycle (head -> body/agent -> cond -> head/exit). The marker flags
3322# the placeholder so the parent uses the foreach node wrapper, and carries the
3323# build-time recursion bound.
3324_SUBGRAPH_FOREACH_KEY = "__subgraph_foreach__"
3325_FOREACH_RECURSION_LIMIT_KEY = "__foreach_recursion_limit__"
3327# Engine-local cycle state: the active foreach's `{items, index, results,
3328# produced_outputs, max_output}` marker. Nested foreachs live in isolated cycle
3329# invocations, so a shared key cannot collide across them.
3330_FOREACH_STATE_KEY = "__foreach_state__"
3332# State keys the foreach wrapper propagates from the isolated cycle back to the
3333# parent (the aggregate result itself is carried separately under the step id).
3334_FOREACH_SIDE_CHANNEL_KEYS = ("errors", "step_context_metrics", "step_tool_meta")
3336# Build-time recursion bound fallback for a foreach with no declared
3337# `max_iterations`: the item list is uncapped, so reserve a generous bound.
3338_FOREACH_DEFAULT_MAX_ITER = 1000
3341class _CallComposer:
3342 """Compose `call` steps into compiled native subgraph nodes (ADR-0037 D4).
3344 A function body becomes its own compiled StateGraph added to the parent as a
3345 single node keyed by the call id; the ``bind``/``emit`` scaffolding and the
3346 body steps live inside that subgraph and share the parent state. Nested
3347 function calls compose recursively. A `call` nested in a foreach/while body
3348 is composed the same way; the loop body executes the embedded compiled graph
3349 imperatively via ``run_step``'s ``__subgraph__`` branch.
3350 """
3352 def __init__(
3353 self,
3354 registry: ManifestRegistry,
3355 assemble_subgraph: Any,
3356 channels: list[dict],
3357 channel_names: frozenset[str],
3358 *,
3359 assemble_while: Any | None = None,
3360 compose_whiles: bool = False,
3361 assemble_foreach: Any | None = None,
3362 compose_foreachs: bool = False,
3363 build_ctx: dict | None = None,
3364 ) -> None:
3365 self._registry = registry
3366 self._assemble_subgraph = assemble_subgraph
3367 self._channels = channels
3368 self._channel_names = channel_names
3369 self._assemble_while = assemble_while
3370 self._compose_whiles = compose_whiles
3371 self._assemble_foreach = assemble_foreach
3372 self._compose_foreachs = compose_foreachs
3373 self._build_ctx = build_ctx or {}
3374 self._stack: list[str] = []
3375 self._expanded_functions: dict[str, str] = {}
3376 self._call_map: dict[str, str] = {}
3378 def compose(self, steps: list[dict]) -> list[dict]:
3379 visible_ids = _collect_step_ids(steps)
3380 composed: list[dict] = []
3381 for step in steps:
3382 composed.extend(self._compose_step(step, visible_ids=visible_ids))
3383 return composed
3385 def _compose_step(self, step: dict, *, visible_ids: set[str]) -> list[dict]:
3386 step_type = step.get("type")
3387 if step_type == "call":
3388 return [self._compose_subgraph_call(step, visible_ids)]
3389 if step_type == "while" and self._compose_whiles:
3390 return [self._compose_while(step, visible_ids)]
3391 if step_type == "foreach" and self._should_compose_foreach(step):
3392 return [self._compose_foreach(step, visible_ids)]
3393 body = step.get("body")
3394 if not isinstance(body, list):
3395 return [step]
3396 nested: list[dict] = []
3397 for sub in body:
3398 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3399 _rewrite_call_deps(nested, self._call_map)
3400 return [{**step, "body": nested}]
3402 def _compose_while(self, step: dict, visible_ids: set[str]) -> dict:
3403 """Compose a `while` step into a native LangGraph cycle subgraph.
3405 The embedded graph owns the loop topology: a head that resolves the
3406 entry condition / initial ``current``, the composed body step nodes, a
3407 conditional back-edge that computes the feed + exit condition and routes
3408 continue-vs-exit, and an exit node that writes the documented while
3409 result. ``expand_calls`` leaves the step uncomposed (``compose_whiles``
3410 false) so the call-expansion seam still exposes the loop body.
3411 """
3412 _reject_nested_routes(step)
3413 body = step.get("body") or []
3414 nested: list[dict] = []
3415 for sub in body:
3416 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3417 _validate_step_channel_bindings(nested, self._channel_names)
3418 compiled, recursion_limit = self._assemble_while(step, nested)
3419 placeholder: dict = {
3420 "id": step["id"],
3421 "type": _SUBGRAPH_STEP_TYPE,
3422 _SUBGRAPH_GRAPH_KEY: compiled,
3423 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3424 _SUBGRAPH_WHILE_KEY: True,
3425 _WHILE_RECURSION_LIMIT_KEY: recursion_limit,
3426 "depends_on": list(step.get("depends_on") or []),
3427 }
3428 if step.get("routes"):
3429 placeholder["routes"] = step["routes"]
3430 return placeholder
3432 def _should_compose_foreach(self, step: dict) -> bool:
3433 """Whether a `foreach` is the sequential form composed into a cycle.
3435 Only a `max_concurrent` absent or resolving to ``<= 1`` is composed; a
3436 declared ``max_concurrent > 1`` keeps the native ``Send`` / in-node
3437 concurrent behaviour.
3438 """
3439 if not self._compose_foreachs:
3440 return False
3441 return _resolved_send_concurrency(step, self._build_ctx) <= 1
3443 def _compose_foreach(self, step: dict, visible_ids: set[str]) -> dict:
3444 """Compose a sequential `foreach` into a native LangGraph cycle subgraph.
3446 Mirrors ``_compose_while``: the embedded graph owns the loop topology
3447 (a head that resolves the item list and injects the current item, the
3448 composed body step nodes or the per-item agent node, a conditional
3449 back-edge routing next-item-vs-exit, and an exit node that writes the
3450 documented aggregate). ``expand_calls`` leaves the step uncomposed so
3451 the call-expansion seam still exposes the loop body.
3452 """
3453 _reject_nested_routes(step)
3454 body = step.get("body") or []
3455 nested: list[dict] = []
3456 for sub in body:
3457 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3458 _validate_step_channel_bindings(nested, self._channel_names)
3459 compiled, recursion_limit = self._assemble_foreach(step, nested)
3460 placeholder: dict = {
3461 "id": step["id"],
3462 "type": _SUBGRAPH_STEP_TYPE,
3463 _SUBGRAPH_GRAPH_KEY: compiled,
3464 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(nested),
3465 _SUBGRAPH_FOREACH_KEY: True,
3466 _FOREACH_RECURSION_LIMIT_KEY: recursion_limit,
3467 "depends_on": list(step.get("depends_on") or []),
3468 }
3469 if step.get("routes"):
3470 placeholder["routes"] = step["routes"]
3471 return placeholder
3473 def _compose_subgraph_call(self, step: dict, visible_ids: set[str]) -> dict:
3474 call_id = step["id"]
3475 function, body = self._check_call_guards(step, visible_ids)
3477 nested_visible = visible_ids | _collect_step_ids(body)
3478 outputs_spec = (
3479 ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
3480 )
3481 bind_id = f"_call__{call_id}__bind"
3482 emit_id = f"_call__{call_id}__emit"
3483 bind_step: dict = {
3484 "id": bind_id,
3485 "type": "builtin",
3486 "component": "__call_bind",
3487 "params": {**(step.get("params") or {}), "__call_id": call_id},
3488 "depends_on": [],
3489 }
3491 self._stack.append(step["function"])
3492 try:
3493 expanded_body = self._compose_body(
3494 body, call_id, bind_id, nested_visible
3495 )
3496 finally:
3497 self._stack.pop()
3499 emit_step: dict = {
3500 "id": emit_id,
3501 "type": "builtin",
3502 "component": "__call_emit",
3503 "params": {"__call_id": call_id, "__outputs": outputs_spec},
3504 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
3505 }
3506 sub_steps = [bind_step, *expanded_body, emit_step]
3507 if self._channels:
3508 _validate_channel_inputs(self._channels, sub_steps)
3509 _validate_step_channel_bindings(sub_steps, self._channel_names)
3510 placeholder: dict = {
3511 "id": call_id,
3512 "type": _SUBGRAPH_STEP_TYPE,
3513 _SUBGRAPH_GRAPH_KEY: self._assemble_subgraph(
3514 _topological_order(sub_steps)
3515 ),
3516 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(sub_steps),
3517 "depends_on": list(step.get("depends_on") or []),
3518 }
3519 call_condition = step.get("condition")
3520 if call_condition is not None:
3521 placeholder["condition"] = call_condition
3522 return placeholder
3524 def _compose_body(
3525 self, body: list[dict], call_id: str, bind_id: str, visible_ids: set[str]
3526 ) -> list[dict]:
3527 expanded: list[dict] = []
3528 for body_step in body:
3529 scoped = _scope_params(body_step, call_id)
3530 for sub in self._compose_step(scoped, visible_ids=visible_ids):
3531 deps = list(sub.get("depends_on") or [])
3532 if bind_id in deps:
3533 expanded.append(sub)
3534 else:
3535 expanded.append({**sub, "depends_on": [bind_id, *deps]})
3536 return expanded
3538 def _check_call_guards(
3539 self, step: dict, visible_ids: set[str]
3540 ) -> tuple[dict, list[dict]]:
3541 """Single owner of every build-time ``call`` guard (ADR-0037 D3/D5/D7).
3543 Rejects routes on the call step, function cycles, duplicate expansion,
3544 unknown functions, function-body isolation (purity) violations, and
3545 body-declared routes; returns the resolved function and its body.
3546 """
3547 fn_name = step["function"]
3548 call_id = step["id"]
3550 if step.get("routes"):
3551 raise RunHardFailure(
3552 "VALIDATION_ERROR",
3553 f"call step '{call_id}' declares `routes`; `routes` is not supported "
3554 "on call steps",
3555 )
3556 if fn_name in self._stack:
3557 chain = " -> ".join(self._stack + [fn_name])
3558 raise RunHardFailure(
3559 "VALIDATION_ERROR",
3560 f"cycle detected in function expansion: {chain} "
3561 "(ADR-0037 D7 — recursion must be a bounded while loop or "
3562 "idempotent re-run)",
3563 )
3564 if fn_name in self._expanded_functions:
3565 raise RunHardFailure(
3566 "VALIDATION_ERROR",
3567 f"duplicate call expansion: function '{fn_name}' called at both "
3568 f"'{self._expanded_functions[fn_name]}' and '{call_id}' in one "
3569 "pipeline (ADR-0037 D7 — repeated calls to the same function are "
3570 "not yet supported)",
3571 )
3572 self._expanded_functions[fn_name] = call_id
3574 function = self._registry.get("Function", fn_name)
3575 if function is None:
3576 raise RunHardFailure(
3577 "VALIDATION_ERROR",
3578 f"call step '{call_id}' references unknown function '{fn_name}' "
3579 "(no kind: Function manifest in the workflows tree)",
3580 )
3581 body = (function.get("spec") or {}).get("steps") or []
3582 _validate_function_purity(fn_name, body, visible_ids)
3583 _reject_function_body_routes(fn_name, body)
3584 return function, body
3587_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
3590def _validate_channel_entry(index: int, channel: Any) -> None:
3591 """Validate one spec.channels entry against the frozen ChannelSpec shape.
3593 The engine build path does not run manifest-schema validation, so a
3594 malformed entry is rejected here — before any ``["name"]`` indexing — with
3595 an error that identifies the offending entry.
3596 """
3597 if not isinstance(channel, dict):
3598 raise RunHardFailure(
3599 "CHANNEL_BINDING_ERROR",
3600 f"Pipeline spec.channels[{index}] must be a mapping; got "
3601 f"{type(channel).__name__}",
3602 )
3603 name = channel.get("name")
3604 if not isinstance(name, str) or not name:
3605 raise RunHardFailure(
3606 "CHANNEL_BINDING_ERROR",
3607 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3608 "channel requires a non-empty string 'name'",
3609 )
3610 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3611 raise RunHardFailure(
3612 "CHANNEL_BINDING_ERROR",
3613 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3614 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}",
3615 )
3618def _validated_channels(raw_channels: Any) -> list[dict]:
3619 """Return spec.channels as a validated list, or [] when absent.
3621 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3622 """
3623 if raw_channels is None:
3624 return []
3625 if not isinstance(raw_channels, list):
3626 raise RunHardFailure(
3627 "CHANNEL_BINDING_ERROR",
3628 f"Pipeline spec.channels must be a list; got "
3629 f"{type(raw_channels).__name__}",
3630 )
3631 for index, channel in enumerate(raw_channels):
3632 _validate_channel_entry(index, channel)
3633 return raw_channels
3636def _validate_step_channel_bindings(
3637 steps: list[dict], channel_names: frozenset[str]
3638) -> None:
3639 """Reject malformed or undeclared step-level channel bindings at build time.
3641 Every step's declared ``output_channel`` must be a non-empty declared
3642 channel name, and every declared ``input_channels`` entry must be a
3643 non-empty declared channel name. This runs unconditionally, so a binding
3644 to a channel absent from ``spec.channels`` fails the build even when no
3645 channels are declared.
3646 """
3647 for step in _iter_steps(steps):
3648 _validate_step_output_channel(step, channel_names)
3649 _validate_step_input_channels(step, channel_names)
3652def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3653 channel = step.get("output_channel")
3654 if channel is None:
3655 return
3656 if not isinstance(channel, str) or not channel:
3657 raise RunHardFailure(
3658 "CHANNEL_BINDING_ERROR",
3659 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3660 f"string; got {channel!r}",
3661 )
3662 if channel not in channel_names:
3663 raise RunHardFailure(
3664 "CHANNEL_BINDING_ERROR",
3665 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3666 "which is not declared in spec.channels",
3667 )
3670def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3671 channels = step.get("input_channels")
3672 if channels is None:
3673 return
3674 if not isinstance(channels, list):
3675 raise RunHardFailure(
3676 "CHANNEL_BINDING_ERROR",
3677 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3678 f"channel names; got {type(channels).__name__}",
3679 )
3680 for channel in channels:
3681 if not isinstance(channel, str) or not channel:
3682 raise RunHardFailure(
3683 "CHANNEL_BINDING_ERROR",
3684 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3685 f"channel name '{channel}'; channel names must be non-empty "
3686 "strings",
3687 )
3688 if channel not in channel_names:
3689 raise RunHardFailure(
3690 "CHANNEL_BINDING_ERROR",
3691 f"Step '{step.get('id', '?')}' reads undeclared channel "
3692 f"'{channel}'; the channel must be declared in spec.channels",
3693 )
3696def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3697 """Reject duplicate and engine/step-reserved channel names."""
3698 forbidden = (
3699 {_CORE}
3700 | set(_RESERVED_NAMESPACES)
3701 | _CORE_STATE_KEYS
3702 | _collect_composed_step_ids(steps)
3703 )
3704 seen: set[str] = set()
3705 for channel in channels:
3706 name = channel.get("name")
3707 if name in seen:
3708 raise RunHardFailure(
3709 "CHANNEL_BINDING_ERROR",
3710 f"Pipeline declares channel '{name}' more than once; channel "
3711 "names must be unique",
3712 )
3713 seen.add(name)
3714 if name in forbidden:
3715 raise RunHardFailure(
3716 "CHANNEL_BINDING_ERROR",
3717 f"Pipeline channel '{name}' collides with an engine-reserved "
3718 "name (a step id, engine state key, reserved namespace, or the "
3719 "core state channel); channel names must be unique and "
3720 "non-reserved",
3721 )
3724def _input_path_root(value: str) -> str | None:
3725 """Leading state key of a bare dot-path input string, or None.
3727 Template strings, bracket-prefixed verbatim literals, and blank strings
3728 are not state reads and yield None.
3729 """
3730 if "{{" in value:
3731 return None
3732 stripped = value.lstrip()
3733 if not stripped or stripped.startswith("["):
3734 return None
3735 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3738def _iter_input_roots(input_spec: Any):
3739 """Yield every bare dot-path root in an input spec, recursing into containers."""
3740 if isinstance(input_spec, str):
3741 root = _input_path_root(input_spec)
3742 if root:
3743 yield root
3744 elif isinstance(input_spec, dict):
3745 for value in input_spec.values():
3746 yield from _iter_input_roots(value)
3747 elif isinstance(input_spec, list):
3748 for value in input_spec:
3749 yield from _iter_input_roots(value)
3752def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3753 """Reject channel-pipeline input roots that name nothing the state exposes.
3755 The channel state is the declared channels plus the engine state, so a bare
3756 dot-path input root must be a declared channel, a known step id, a reserved
3757 namespace, or an engine state key. Anything else is a hard build error —
3758 there is no silent fallback to a missing key.
3759 """
3760 allowed = (
3761 {ch["name"] for ch in channels}
3762 | _collect_step_ids(steps)
3763 | set(_RESERVED_NAMESPACES)
3764 | _CORE_STATE_KEYS
3765 )
3766 for step in _iter_steps(steps):
3767 for root in _iter_input_roots(step.get("input")):
3768 if root not in allowed:
3769 raise RunHardFailure(
3770 "CHANNEL_BINDING_ERROR",
3771 f"Pipeline channel step '{step.get('id', '?')}' input "
3772 f"references unknown root '{root}'; in a channel pipeline "
3773 "an input root must be a declared channel, a step id, a "
3774 "reserved namespace, or an engine state key",
3775 )
3778def _flatten_channel_state(state: dict) -> dict:
3779 """Read the flat committed snapshot nested under the Core namespace."""
3780 return dict(state.get(_CORE) or {})
3783def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3784 """Drop channel-named keys so a node emits only the core state."""
3785 return {k: v for k, v in flat.items() if k not in channel_names}
3788_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3790_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3793def _envelope_content(result: Any) -> Any:
3794 """Return an envelope's optional ``content`` field (ADR-0023 D3).
3796 ``content`` is opaque and optional: a non-mapping result, an absent field,
3797 and an explicit ``null`` all read as ``None``. Callers apply their own
3798 fallback and shape checks (e.g. ``[]`` for a spawn child plan).
3799 """
3800 if isinstance(result, dict):
3801 return result.get("content")
3802 return None
3805def _is_failure_envelope(value: Any) -> bool:
3806 """True when a step output is not a successfully parsed dict.
3808 Missing output, non-dict values, and the ``error``/``parse_error``
3809 envelopes written by the LLM failure paths are not routed. A valid
3810 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3811 it carries an explicit ``"error": null``.
3812 """
3813 if not isinstance(value, dict):
3814 return True
3815 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3816 return False
3817 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3820def _declared_output_channel_writes(
3821 step: dict, output: Any, channel_names: frozenset[str]
3822) -> dict[str, Any]:
3823 """Route one step output's content to its declared output channel.
3825 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3826 the working set. Failure envelopes and steps without an ``output_channel``
3827 are not routed; an undeclared target channel is a hard error.
3828 """
3829 if _is_failure_envelope(output):
3830 return {}
3831 channel = step.get("output_channel")
3832 if not channel:
3833 return {}
3834 if channel not in channel_names:
3835 raise RunHardFailure(
3836 "CHANNEL_BINDING_ERROR",
3837 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3838 "which is not declared in spec.channels",
3839 )
3840 return {channel: _envelope_content(output)}
3843def _route_declared_channels(
3844 step: dict,
3845 pre_state: dict,
3846 core: dict,
3847 step_runner: Any,
3848 channel_names: frozenset[str],
3849) -> dict[str, Any]:
3850 """Route a step's declared output channel write to its channel.
3852 Returns ``{channel_name: value}`` for LangGraph to merge through the
3853 channel's reducer, where ``channel_name`` is the step's single declared
3854 ``output_channel`` and ``value`` is the agent envelope's ``content``. Only
3855 an actual agent invocation is routed: a ``skip_llm_if`` passthrough and a
3856 parallel fan-out envelope are excluded. A step without an ``output_channel``
3857 routes nothing (its output stays in ``core`` untouched), and failure
3858 envelopes are not routed. A spawn step is routed at the spawn seam from its
3859 node result, never here from its envelope.
3860 """
3861 step_id = step.get("id", "?")
3862 if step.get("type") == "spawn":
3863 return {}
3864 if step_id not in core:
3865 return {}
3866 agent_spec = step_runner.resolve_agent(step)
3867 if agent_spec is None:
3868 return {}
3869 if not step_runner.ran_agent(step, pre_state):
3870 return {}
3871 return _declared_output_channel_writes(step, core[step_id], channel_names)
3874def _make_step_node(
3875 step: dict,
3876 step_runner: Any,
3877 channel_names: frozenset[str],
3878 channel_types: dict[str, str],
3879) -> Any:
3880 if not channel_names:
3881 def node_fn(state: PipelineState) -> PipelineState:
3882 return step_runner.run_step(step, state)
3883 return node_fn
3885 def node_fn(state: PipelineState) -> PipelineState:
3886 committed = _flatten_channel_state(state)
3887 working_set = ChannelWorkingSet(committed, channel_types)
3888 core = _strip_channel_keys(
3889 step_runner.run_step(step, committed, working_set), channel_names
3890 )
3891 writes = _route_declared_channels(
3892 step, committed, core, step_runner, channel_names
3893 )
3894 for name, value in writes.items():
3895 working_set.stage(name, value)
3896 written_names = {name for name, _ in working_set.pending_writes()}
3897 channels = working_set.commit()
3898 emitted = {name: channels[name] for name in written_names}
3899 return {_CORE: {**core, **emitted}}
3900 return node_fn
3903def _filter_composed_step_ids(result: dict, composed_step_ids: Any) -> dict:
3904 """Drop function-body step ids so a call subgraph's internals stay invisible.
3906 ADR-0037 D3: the caller sees only ``state[<call_id>].outputs`` plus
3907 unrelated keys. ``composed_step_ids`` is the build-time set recorded on the
3908 call placeholder. Channel-named keys are disjoint from it (a collision is a
3909 build-time ``CHANNEL_BINDING_ERROR``), so channel state is never touched.
3910 The input is left unmodified.
3911 """
3912 if not composed_step_ids:
3913 return dict(result)
3914 filtered = {
3915 key: value
3916 for key, value in result.items()
3917 if key not in composed_step_ids
3918 }
3919 core = filtered.get(_CORE)
3920 if isinstance(core, dict):
3921 filtered[_CORE] = {
3922 key: value
3923 for key, value in core.items()
3924 if key not in composed_step_ids
3925 }
3926 return filtered
3929def _make_subgraph_node(
3930 compiled: Any,
3931 condition: Any,
3932 step_runner: Any,
3933 channel_names: frozenset[str],
3934 composed_step_ids: Any = None,
3935) -> Any:
3936 """Wrap a compiled call subgraph with the call step's entry condition.
3938 A false condition returns no state updates, so the subgraph never runs and
3939 ``state[<call_id>]`` is not written — matching the soft-skip semantics of a
3940 normal step condition. The ctx is derived exactly like ``run_step``'s,
3941 including channel-mode flattening. Every successful invocation is filtered
3942 through ``_filter_composed_step_ids`` so the function body stays invisible.
3943 """
3944 if condition is None:
3945 def node_fn(state: PipelineState) -> PipelineState:
3946 return _filter_composed_step_ids(
3947 compiled.invoke(state), composed_step_ids
3948 )
3949 return node_fn
3951 if not channel_names:
3952 def node_fn(state: PipelineState) -> PipelineState:
3953 if not _eval_condition(condition, step_runner._make_ctx(state)):
3954 return {}
3955 return _filter_composed_step_ids(
3956 compiled.invoke(state), composed_step_ids
3957 )
3958 return node_fn
3960 def node_fn(state: PipelineState) -> PipelineState:
3961 committed = _flatten_channel_state(state)
3962 if not _eval_condition(condition, step_runner._make_ctx(committed)):
3963 return {}
3964 return _filter_composed_step_ids(
3965 compiled.invoke(state), composed_step_ids
3966 )
3967 return node_fn
3970# ---------------------------------------------------------------------------
3971# Native `while` cycle nodes (ADR-0025 D1/D2)
3972# ---------------------------------------------------------------------------
3973#
3974# A `while` step is a compiled cycle: `head -> body... -> cond` with a
3975# conditional back-edge from `cond` to `head` (continue) or `exit`. The head
3976# owns entry-condition handling and the initial `current`; the cond node owns
3977# the feed, the iteration increment and the exit-condition evaluation; the exit
3978# node writes the documented result and enforces `fail_if_exhausted`. The
3979# nodes read/write the shared graph state, using the `_CORE` container when the
3980# pipeline declares channels.
3983def _while_marker_key(step_id: str) -> str:
3984 """Per-while engine marker key (nested whiles live in isolated cycles)."""
3985 return f"{_WHILE_STATE_KEY}__{step_id}"
3988def _while_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
3989 return _flatten_channel_state(state) if channel_names else state
3992def _while_update(update: dict, channel_names: frozenset[str]) -> dict:
3993 return {_CORE: update} if channel_names else update
3996def _while_loop_context(step_id: str, marker: dict) -> dict:
3997 """The body-visible `loop` namespace and the intermediate step result.
3999 Written in this order so a step literally named ``loop`` shadows the
4000 injected namespace — the legacy in-node loop had the same collision.
4001 """
4002 current = marker["current"]
4003 iteration = marker["iteration"]
4004 update = {"loop": {"current": current, "iteration": iteration}}
4005 update[step_id] = {"current": current, "passed": False}
4006 return update
4009def _while_invoke_config(step: dict) -> dict | None:
4010 """Recursion bound for a while/foreach cycle placeholder, else None."""
4011 if step.get(_SUBGRAPH_WHILE_KEY):
4012 return {"recursion_limit": step.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
4013 if step.get(_SUBGRAPH_FOREACH_KEY):
4014 return {"recursion_limit": step.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
4015 return None
4018def _make_while_body_node(
4019 sub: dict,
4020 step_runner: Any,
4021 channel_names: frozenset[str],
4022 channel_types: dict[str, str],
4023) -> Any:
4024 """A while body step as a node: a nested subgraph or a normal step node."""
4025 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
4026 if sub.get(_SUBGRAPH_WHILE_KEY):
4027 return _make_while_subgraph_node(sub, step_runner, channel_names)
4028 if sub.get(_SUBGRAPH_FOREACH_KEY):
4029 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
4030 return _make_subgraph_node(
4031 sub[_SUBGRAPH_GRAPH_KEY],
4032 sub.get("condition"),
4033 step_runner,
4034 channel_names,
4035 sub.get(_SUBGRAPH_STEP_IDS_KEY),
4036 )
4037 return _make_step_node(sub, step_runner, channel_names, channel_types)
4040def _make_while_head_node(
4041 step: dict,
4042 step_runner: Any,
4043 channel_names: frozenset[str],
4044 max_iter: int,
4045) -> Any:
4046 """Entry node: handle the entry condition and seed the loop marker."""
4047 step_id = step.get("id", "?")
4048 marker_key = _while_marker_key(step_id)
4050 def node_fn(state: PipelineState) -> PipelineState:
4051 flat = _while_flat(state, channel_names)
4052 marker = flat.get(marker_key)
4053 if marker is not None:
4054 return _while_update(
4055 _while_loop_context(step_id, marker), channel_names
4056 )
4057 ctx = step_runner._make_ctx(flat)
4058 entry_cond = step.get("condition")
4059 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4060 skip_input = _resolved_while_input(step, flat, ctx)
4061 skip_marker = {
4062 "current": None,
4063 "iteration": 0,
4064 "passed": True,
4065 "skipped": True,
4066 "max_iter": max_iter,
4067 }
4068 return _while_update(
4069 {
4070 step_id: {
4071 "passed": True,
4072 "iterations": 0,
4073 "output": skip_input,
4074 "exhausted": False,
4075 "skipped": True,
4076 },
4077 marker_key: skip_marker,
4078 },
4079 channel_names,
4080 )
4081 marker = {
4082 "current": _resolved_while_input(step, flat, ctx),
4083 "iteration": 0,
4084 "passed": False,
4085 "skipped": False,
4086 "max_iter": max_iter,
4087 }
4088 update = _while_loop_context(step_id, marker)
4089 update[marker_key] = marker
4090 return _while_update(update, channel_names)
4092 return node_fn
4095def _resolved_while_input(step: dict, flat: dict, ctx: dict) -> Any:
4096 input_spec = step.get("input")
4097 if input_spec is None:
4098 return None
4099 return _resolve_input(input_spec, flat, ctx)
4102def _make_while_head_router(
4103 step: dict,
4104 channel_names: frozenset[str],
4105 first_id: str,
4106 exit_id: str,
4107 end: Any,
4108) -> Any:
4109 """Route from the head: skipped → end, bound already reached → exit, else body."""
4110 marker_key = _while_marker_key(step.get("id", "?"))
4112 def router(state: PipelineState) -> str:
4113 marker = _while_flat(state, channel_names).get(marker_key) or {}
4114 if marker.get("skipped"):
4115 return end
4116 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
4117 return exit_id
4118 return first_id
4120 return router
4123def _make_while_cond_node(
4124 step: dict,
4125 step_runner: Any,
4126 channel_names: frozenset[str],
4127) -> Any:
4128 """Feed the declared output, increment the iteration, evaluate the exit."""
4129 step_id = step.get("id", "?")
4130 marker_key = _while_marker_key(step_id)
4131 declared_output = step.get("output")
4132 exit_condition = step.get("exit_condition")
4134 def node_fn(state: PipelineState) -> PipelineState:
4135 flat = _while_flat(state, channel_names)
4136 marker = dict(flat.get(marker_key) or {})
4137 current = marker.get("current")
4138 if declared_output:
4139 feed_value = flat.get(declared_output)
4140 if isinstance(feed_value, dict) and "output" in feed_value:
4141 current = feed_value["output"]
4142 elif isinstance(feed_value, dict) and isinstance(
4143 feed_value.get("results"), list
4144 ):
4145 current = feed_value["results"]
4146 else:
4147 current = feed_value
4148 passed = False
4149 if exit_condition:
4150 if _eval_condition(
4151 exit_condition, step_runner._make_ctx(flat)
4152 ):
4153 passed = True
4154 else:
4155 passed = True
4156 marker.update(
4157 {
4158 "current": current,
4159 "iteration": marker.get("iteration", 0) + 1,
4160 "passed": passed,
4161 }
4162 )
4163 return _while_update({marker_key: marker}, channel_names)
4165 return node_fn
4168def _make_while_cond_router(
4169 step: dict,
4170 channel_names: frozenset[str],
4171 head_id: str,
4172 exit_id: str,
4173) -> Any:
4174 """Route from the cond: exit on pass or when the bound is reached, else head."""
4175 marker_key = _while_marker_key(step.get("id", "?"))
4177 def router(state: PipelineState) -> str:
4178 marker = _while_flat(state, channel_names).get(marker_key) or {}
4179 if marker.get("passed"):
4180 return exit_id
4181 if marker.get("iteration", 0) >= marker.get("max_iter", 0):
4182 return exit_id
4183 return head_id
4185 return router
4188def _make_while_exit_node(
4189 step: dict,
4190 step_runner: Any,
4191 channel_names: frozenset[str],
4192) -> Any:
4193 """Write the documented while result and enforce `fail_if_exhausted`."""
4194 step_id = step.get("id", "?")
4195 marker_key = _while_marker_key(step_id)
4196 fail_if_exhausted = step.get("fail_if_exhausted")
4198 def node_fn(state: PipelineState) -> PipelineState:
4199 flat = _while_flat(state, channel_names)
4200 marker = flat.get(marker_key) or {}
4201 passed = bool(marker.get("passed", False))
4202 exhausted = not passed
4203 if (
4204 fail_if_exhausted is not None
4205 and exhausted
4206 and _eval_condition(
4207 fail_if_exhausted, step_runner._make_ctx(flat)
4208 )
4209 ):
4210 raise RunHardFailure(
4211 "EXHAUSTED",
4212 f"while step '{step_id}' exhausted "
4213 f"max_iterations={marker.get('max_iter')} without "
4214 "exit_condition passing (fail_if_exhausted)",
4215 {"step_id": step_id, "exhausted": True},
4216 )
4217 return _while_update(
4218 {
4219 step_id: {
4220 "passed": passed,
4221 "iterations": marker.get("iteration", 0),
4222 "output": marker.get("current"),
4223 "exhausted": exhausted,
4224 "skipped": False,
4225 }
4226 },
4227 channel_names,
4228 )
4230 return node_fn
4233def _make_while_subgraph_node(
4234 placeholder: dict,
4235 step_runner: Any,
4236 channel_names: frozenset[str],
4237) -> Any:
4238 """Parent node for a compiled `while` cycle: isolate and surface its result.
4240 The cycle shares the graph state schema, so invoking it directly would leak
4241 the body step results and loop plumbing into the parent. This wrapper runs
4242 the cycle on a copy and returns only the documented result plus the side
4243 channels, preserving the in-node loop's state contract exactly.
4244 """
4245 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4246 step_id = placeholder["id"]
4247 config = {"recursion_limit": placeholder.get(_WHILE_RECURSION_LIMIT_KEY, 25)}
4249 if not channel_names:
4250 def node_fn(state: PipelineState) -> PipelineState:
4251 sub = compiled.invoke(state, config)
4252 result = {step_id: sub.get(step_id)}
4253 for key in _WHILE_SIDE_CHANNEL_KEYS:
4254 if key in sub and sub.get(key) != state.get(key):
4255 result[key] = sub.get(key)
4256 return result
4257 return node_fn
4259 def node_fn(state: PipelineState) -> PipelineState:
4260 sub = compiled.invoke(state, config)
4261 flat = _flatten_channel_state(sub)
4262 parent_flat = _flatten_channel_state(state)
4263 result = {step_id: flat.get(step_id)}
4264 for key in _WHILE_SIDE_CHANNEL_KEYS:
4265 if key in flat and flat.get(key) != parent_flat.get(key):
4266 result[key] = flat.get(key)
4267 return {_CORE: _strip_channel_keys(result, channel_names)}
4269 return node_fn
4272# ---------------------------------------------------------------------------
4273# Native sequential `foreach` cycle nodes
4274# ---------------------------------------------------------------------------
4275#
4276# A sequential `foreach` is a compiled cycle: `head -> (body... | agent) ->
4277# cond` with a conditional back-edge from `cond` to `head` (next item) or
4278# `exit` (`max_output` reached or the item list exhausted). The head resolves
4279# the item list once and injects the current item (`item` / `_item` /
4280# `_foreach_item`); the cond collects the item result; the exit node writes the
4281# documented aggregate (`{"results", "passed", "failing", "failed"}` for the
4282# body path, the plain list for the agent path). Nodes read/write the shared
4283# graph state, using the `_CORE` container when the pipeline declares channels.
4286def _foreach_marker_key(step_id: str) -> str:
4287 """Per-foreach engine marker key (nested foreachs live in isolated cycles)."""
4288 return f"{_FOREACH_STATE_KEY}__{step_id}"
4291def _foreach_flat(state: PipelineState, channel_names: frozenset[str]) -> dict:
4292 return _flatten_channel_state(state) if channel_names else state
4295def _foreach_update(update: dict, channel_names: frozenset[str]) -> dict:
4296 return {_CORE: update} if channel_names else update
4299def _make_foreach_head_node(
4300 step: dict,
4301 step_runner: Any,
4302 channel_names: frozenset[str],
4303) -> Any:
4304 """Entry node: evaluate the entry condition, resolve items, inject the item."""
4305 step_id = step.get("id", "?")
4306 marker_key = _foreach_marker_key(step_id)
4308 def node_fn(state: PipelineState) -> PipelineState:
4309 flat = _foreach_flat(state, channel_names)
4310 marker = flat.get(marker_key)
4311 if marker is None:
4312 ctx = step_runner._make_ctx(flat)
4313 entry_cond = step.get("condition")
4314 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
4315 return _foreach_update({}, channel_names)
4316 max_output_raw = step.get("max_output")
4317 marker = {
4318 "items": step_runner._foreach_item_list(step, flat, ctx),
4319 "index": 0,
4320 "results": [],
4321 "produced_outputs": 0,
4322 "max_output": (
4323 _resolve_int(max_output_raw, ctx, 0)
4324 if max_output_raw is not None
4325 else 0
4326 ),
4327 }
4328 items = marker.get("items") or []
4329 index = marker.get("index", 0)
4330 if index < len(items):
4331 update = {
4332 "_item": items[index],
4333 "item": items[index],
4334 "_foreach_item": items[index],
4335 }
4336 update[marker_key] = marker
4337 return _foreach_update(update, channel_names)
4338 return _foreach_update({marker_key: marker}, channel_names)
4340 return node_fn
4343def _make_foreach_head_router(
4344 step: dict,
4345 channel_names: frozenset[str],
4346 first_id: str,
4347 exit_id: str,
4348 end: Any,
4349) -> Any:
4350 """Route from the head: condition-skipped → end, exhausted/budget → exit, else body."""
4351 marker_key = _foreach_marker_key(step.get("id", "?"))
4353 def router(state: PipelineState) -> str:
4354 marker = _foreach_flat(state, channel_names).get(marker_key)
4355 if marker is None:
4356 return end
4357 items = marker.get("items") or []
4358 if marker.get("index", 0) >= len(items):
4359 return exit_id
4360 max_output = marker.get("max_output", 0)
4361 if max_output and marker.get("produced_outputs", 0) >= max_output:
4362 return exit_id
4363 return first_id
4365 return router
4368def _make_foreach_body_node(
4369 sub: dict,
4370 step_runner: Any,
4371 channel_names: frozenset[str],
4372 channel_types: dict[str, str],
4373) -> Any:
4374 """A foreach body step as a node: a nested loop subgraph or a step node."""
4375 if sub.get("type") == _SUBGRAPH_STEP_TYPE:
4376 if sub.get(_SUBGRAPH_WHILE_KEY):
4377 return _make_while_subgraph_node(sub, step_runner, channel_names)
4378 if sub.get(_SUBGRAPH_FOREACH_KEY):
4379 return _make_foreach_subgraph_node(sub, step_runner, channel_names)
4380 return _make_subgraph_node(
4381 sub[_SUBGRAPH_GRAPH_KEY],
4382 sub.get("condition"),
4383 step_runner,
4384 channel_names,
4385 sub.get(_SUBGRAPH_STEP_IDS_KEY),
4386 )
4387 return _make_step_node(sub, step_runner, channel_names, channel_types)
4390def _make_foreach_agent_node(
4391 step: dict,
4392 step_runner: Any,
4393 channel_names: frozenset[str],
4394 channel_types: dict[str, str],
4395) -> Any:
4396 """Agent-path node: invoke the foreach step's agent for the current item."""
4397 step_id = step.get("id", "?")
4399 def node_fn(state: PipelineState) -> PipelineState:
4400 flat = _foreach_flat(state, channel_names)
4401 resolved = step_runner.resolve_agent(step)
4402 ctx = step_runner._make_ctx(flat)
4403 resolved_step = step_runner._resolve_step_runtime(step, flat, ctx)
4404 resolved_step = step_runner._attach_communication_context(resolved_step, flat)
4405 resolved_step = step_runner._attach_resolved_tools(resolved_step, resolved)
4406 result_state = step_runner._invoke_agent(resolved, resolved_step, flat)
4407 if not channel_names:
4408 return result_state
4409 return {_CORE: _strip_channel_keys(result_state, channel_names)}
4411 return node_fn
4414def _make_foreach_cond_node(
4415 step: dict,
4416 step_runner: Any,
4417 channel_names: frozenset[str],
4418) -> Any:
4419 """Collect the current item's result and advance the item index."""
4420 step_id = step.get("id", "?")
4421 marker_key = _foreach_marker_key(step_id)
4422 declared_output = step.get("output")
4423 runs_body = bool(step.get("body"))
4425 def node_fn(state: PipelineState) -> PipelineState:
4426 flat = _foreach_flat(state, channel_names)
4427 marker = dict(flat.get(marker_key) or {})
4428 results = list(marker.get("results") or [])
4429 produced = marker.get("produced_outputs", 0)
4430 if runs_body:
4431 results.append(StepRunner._body_item_result(flat, declared_output))
4432 publish_out = flat.get("publish")
4433 if isinstance(publish_out, dict) and publish_out.get("path"):
4434 produced += 1
4435 else:
4436 results.append(flat.get(step_id))
4437 marker.update(
4438 {
4439 "results": results,
4440 "index": marker.get("index", 0) + 1,
4441 "produced_outputs": produced,
4442 }
4443 )
4444 return _foreach_update({marker_key: marker}, channel_names)
4446 return node_fn
4449def _make_foreach_cond_router(
4450 step: dict,
4451 channel_names: frozenset[str],
4452 head_id: str,
4453 exit_id: str,
4454) -> Any:
4455 """Route from the cond: exit on exhaustion or budget, else back to the head."""
4456 marker_key = _foreach_marker_key(step.get("id", "?"))
4458 def router(state: PipelineState) -> str:
4459 marker = _foreach_flat(state, channel_names).get(marker_key) or {}
4460 items = marker.get("items") or []
4461 if marker.get("index", 0) >= len(items):
4462 return exit_id
4463 max_output = marker.get("max_output", 0)
4464 if max_output and marker.get("produced_outputs", 0) >= max_output:
4465 return exit_id
4466 return head_id
4468 return router
4471def _make_foreach_exit_node(
4472 step: dict,
4473 step_runner: Any,
4474 channel_names: frozenset[str],
4475) -> Any:
4476 """Write the documented foreach aggregate result."""
4477 step_id = step.get("id", "?")
4478 marker_key = _foreach_marker_key(step_id)
4479 runs_body = bool(step.get("body"))
4481 def node_fn(state: PipelineState) -> PipelineState:
4482 flat = _foreach_flat(state, channel_names)
4483 marker = flat.get(marker_key) or {}
4484 results = list(marker.get("results") or [])
4485 step_result = (
4486 StepRunner._aggregate_body_results(results) if runs_body else results
4487 )
4488 return _foreach_update({step_id: step_result}, channel_names)
4490 return node_fn
4493def _make_foreach_subgraph_node(
4494 placeholder: dict,
4495 step_runner: Any,
4496 channel_names: frozenset[str],
4497) -> Any:
4498 """Parent node for a compiled sequential `foreach` cycle: isolate and surface.
4500 The cycle shares the graph state schema, so invoking it directly would leak
4501 body step results and loop plumbing into the parent. This wrapper runs the
4502 cycle on a copy and returns only the documented aggregate plus the side
4503 channels. A condition-skipped entry writes no step result (matching the
4504 in-node soft-skip), so the step id is only surfaced when the cycle wrote it.
4505 """
4506 compiled = placeholder[_SUBGRAPH_GRAPH_KEY]
4507 step_id = placeholder["id"]
4508 config = {"recursion_limit": placeholder.get(_FOREACH_RECURSION_LIMIT_KEY, 25)}
4510 if not channel_names:
4511 def node_fn(state: PipelineState) -> PipelineState:
4512 sub = compiled.invoke(state, config)
4513 if step_id not in sub:
4514 return {}
4515 result = {step_id: sub.get(step_id)}
4516 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4517 if key in sub and sub.get(key) != state.get(key):
4518 result[key] = sub.get(key)
4519 return result
4520 return node_fn
4522 def node_fn(state: PipelineState) -> PipelineState:
4523 sub = compiled.invoke(state, config)
4524 flat = _flatten_channel_state(sub)
4525 parent_flat = _flatten_channel_state(state)
4526 if step_id not in flat:
4527 return {}
4528 result = {step_id: flat.get(step_id)}
4529 for key in _FOREACH_SIDE_CHANNEL_KEYS:
4530 if key in flat and flat.get(key) != parent_flat.get(key):
4531 result[key] = flat.get(key)
4532 return {_CORE: _strip_channel_keys(result, channel_names)}
4534 return node_fn
4537def _collect_parallel_channel_writes(
4538 steps: list[dict],
4539 pre_state: dict,
4540 core: dict,
4541 step_runner: Any,
4542 channel_names: frozenset[str],
4543 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
4544) -> list[tuple[str, Any]]:
4545 """Collect one parallel node's declared writes in branch order.
4547 Every branch write is emitted, so fan-in feeds the channel-type reducer
4548 every write in deterministic declaration/branch order: Topic appends each
4549 write, LastValue keeps the last, and BinaryOperatorAggregate combines them
4550 per its accumulated shape (list concat, collection-map union, dict merge,
4551 scalar last). A branch's isolated body writes are emitted before its own
4552 declared write, mirroring the top-level node's route-after-run order.
4553 """
4554 writes: list[tuple[str, Any]] = []
4555 for step in steps:
4556 if branch_working_sets is not None:
4557 branch_working_set = branch_working_sets.get(step["id"])
4558 if branch_working_set is not None:
4559 writes.extend(branch_working_set.pending_writes())
4560 branch = _route_declared_channels(
4561 step, pre_state, core, step_runner, channel_names
4562 )
4563 for channel_name, value in branch.items():
4564 writes.append((channel_name, value))
4565 return writes
4568def _make_parallel_node(
4569 group_steps: list[dict],
4570 step_runner: Any,
4571 channel_names: frozenset[str],
4572 channel_types: dict[str, str],
4573) -> Any:
4574 def node_fn(state: PipelineState) -> PipelineState:
4575 base = (
4576 _flatten_channel_state(state)
4577 if channel_names else state
4578 )
4579 working_set = (
4580 ChannelWorkingSet(base, channel_types) if channel_names else None
4581 )
4582 # Each branch writes into its own fork; the shared set is advanced only
4583 # at fan-in, so branch-internal body writes never race.
4584 branch_working_sets = (
4585 {s["id"]: working_set.fork() for s in group_steps}
4586 if working_set is not None else None
4587 )
4589 def submit_branch(pool: Any, step: dict) -> Any:
4590 if branch_working_sets is None:
4591 return pool.submit(step_runner.run_step, step, base)
4592 return pool.submit(
4593 step_runner.run_step, step, base, branch_working_sets[step["id"]]
4594 )
4596 results: dict[str, Any] = {}
4597 branch_metrics: dict = {}
4598 branch_tool_meta: dict = {}
4599 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
4600 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
4601 for future in as_completed(futures):
4602 sid = futures[future]
4603 try:
4604 out = future.result()
4605 results[sid] = out.get(sid)
4606 branch_metrics.update(out.get("step_context_metrics") or {})
4607 branch_tool_meta.update(out.get("step_tool_meta") or {})
4608 except SkillLoadError:
4609 raise
4610 except RunHardFailure:
4611 pool.shutdown(wait=False, cancel_futures=True)
4612 raise
4613 except Exception as exc: # noqa: BLE001
4614 log.error("Step '%s' failed: %s", sid, exc)
4615 results[sid] = {"error": str(exc)}
4616 if channel_names:
4617 merged = _strip_channel_keys({**base, **results}, channel_names)
4618 if branch_metrics:
4619 merged["step_context_metrics"] = _merge_side_channel(
4620 merged.get("step_context_metrics"), branch_metrics
4621 )
4622 if branch_tool_meta:
4623 merged["step_tool_meta"] = _merge_side_channel(
4624 merged.get("step_tool_meta"), branch_tool_meta
4625 )
4626 writes = _collect_parallel_channel_writes(
4627 group_steps, base, merged, step_runner, channel_names,
4628 branch_working_sets,
4629 )
4630 written_names = {name for name, _ in writes}
4631 for name, value in writes:
4632 working_set.stage(name, value)
4633 channels = working_set.commit()
4634 emitted = {name: channels[name] for name in written_names}
4635 return {_CORE: {**merged, **emitted}}
4636 final: PipelineState = {**state, **results}
4637 if branch_metrics:
4638 final = {
4639 **final,
4640 "step_context_metrics": _merge_side_channel(
4641 state.get("step_context_metrics"), branch_metrics
4642 ),
4643 }
4644 if branch_tool_meta:
4645 final = {
4646 **final,
4647 "step_tool_meta": _merge_side_channel(
4648 state.get("step_tool_meta"), branch_tool_meta
4649 ),
4650 }
4651 return final
4652 return node_fn
4655@dataclass(frozen=True)
4656class _MapReduceParts:
4657 """The nodes and edges of one concurrent step's Send fan-out.
4659 ``dispatch_id`` reuses the step id so the step's incoming dependency edges are
4660 unchanged; ``join_id`` carries the step's outgoing edges.
4661 """
4663 dispatch_id: str
4664 worker_id: str
4665 fallback_id: str
4666 join_id: str
4667 dispatch_fn: Any
4668 route_fn: Any
4669 worker_fn: Any
4670 join_fn: Any
4673@dataclass(frozen=True)
4674class _MapReduceItemOutcome:
4675 """One worker's per-item result plus its errors, side channels and writes.
4677 ``errors`` carries the item's soft errors (and the inherited soft errors it
4678 read), so the join can fold them in item order exactly like the in-node
4679 concurrent path. ``channel_writes`` is empty for the parallel strategy (its
4680 channel writes are derived from the item result at the join) and carries the
4681 foreach body's pending writes so the join can stage them in item order.
4682 """
4684 result: Any
4685 metrics: dict | None
4686 tool_meta: dict | None
4687 channel_writes: tuple[tuple[str, Any], ...]
4688 errors: tuple[str, ...] = ()
4691@dataclass(frozen=True)
4692class _MapReducePlan:
4693 """The per-step behaviour injected into the shared Send machinery.
4695 The parallel strategy runs a single agent call per item; the foreach
4696 strategy runs the body (or agent) per item. ``item_lister`` resolves the
4697 fan-out list, ``item_runner`` executes one item, ``result_assembler``
4698 shapes the step result, and ``channel_write_stager`` stages per-item
4699 channel writes at the join.
4700 """
4702 item_lister: Any
4703 item_runner: Any
4704 result_assembler: Any
4705 channel_write_stager: Any
4708def _map_reduce_worker_payload(flat: dict, item: Any, index: int) -> dict:
4709 """Build a worker's ``Send`` payload from the resolved source state."""
4710 return {**flat, "item": item, "_item": item, _MAP_REDUCE_ITEM_INDEX: index}
4713def _map_reduce_worker_output(
4714 step_id: str,
4715 index: int,
4716 outcome: _MapReduceItemOutcome,
4717) -> PipelineState:
4718 """The accumulator record carrying every per-item fold input.
4720 The slot is ``{index: {"result", "writes", "errors", "metrics",
4721 "tool_meta"}}`` so the join can assemble the step result, stage channel
4722 writes, and fold soft errors plus side channels in item order — the same
4723 observable result as the in-node concurrent path.
4724 """
4725 record = {
4726 "result": outcome.result,
4727 "writes": list(outcome.channel_writes),
4728 "errors": list(outcome.errors),
4729 "metrics": outcome.metrics if isinstance(outcome.metrics, dict) else None,
4730 "tool_meta": (
4731 outcome.tool_meta if isinstance(outcome.tool_meta, dict) else None
4732 ),
4733 }
4734 return {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: {index: record}}}
4737def _make_map_reduce_route_fn(
4738 plan: _MapReducePlan,
4739 step_runner: Any,
4740 channel_names: frozenset[str],
4741 worker_id: str,
4742 fallback_id: str,
4743 join_id: str,
4744) -> Any:
4745 def route(state: PipelineState) -> list:
4746 flat = _flatten_channel_state(state) if channel_names else state
4747 ctx = step_runner._make_ctx(flat)
4748 items = plan.item_lister(flat, ctx)
4749 if items is None:
4750 return [Send(fallback_id, state)]
4751 if not items:
4752 return [Send(join_id, state)]
4753 return [
4754 Send(worker_id, _map_reduce_worker_payload(flat, item, index))
4755 for index, item in enumerate(items)
4756 ]
4758 return route
4761def _make_map_reduce_worker_fn(
4762 step: dict,
4763 plan: _MapReducePlan,
4764 step_runner: Any,
4765 gate: threading.Semaphore,
4766) -> Any:
4767 step_id = step.get("id", "?")
4769 def worker(state: PipelineState) -> PipelineState:
4770 index = state[_MAP_REDUCE_ITEM_INDEX]
4771 item_state = {
4772 key: value
4773 for key, value in state.items()
4774 if key != _MAP_REDUCE_ITEM_INDEX
4775 }
4776 item = item_state.get("item")
4777 item_ctx = step_runner._make_ctx(item_state)
4778 with gate:
4779 try:
4780 outcome = plan.item_runner(
4781 state, index, item, item_state, item_ctx
4782 )
4783 except SkillLoadError:
4784 raise
4785 except RunHardFailure:
4786 raise
4787 except Exception as exc: # noqa: BLE001
4788 log.error("[step] %s item failed: %s", step_id, exc)
4789 outcome = _MapReduceItemOutcome(
4790 {"error": str(exc)}, None, None, ()
4791 )
4792 return _map_reduce_worker_output(step_id, index, outcome)
4794 return worker
4797def _fold_map_reduce_errors(records: list, base_errors: list) -> list:
4798 """Fold every item's soft errors into *base_errors* in item order."""
4799 merged = list(base_errors)
4800 for record in records:
4801 if not isinstance(record, dict):
4802 continue
4803 merged = StepRunner._merge_errors(merged, record.get("errors") or [])
4804 return merged
4807def _fold_map_reduce_side_channel(records: list, record_key: str) -> dict:
4808 """Fold every item's side-channel map in item order (later item wins)."""
4809 folded: dict = {}
4810 for record in records:
4811 if not isinstance(record, dict):
4812 continue
4813 value = record.get(record_key)
4814 if isinstance(value, dict):
4815 folded.update(value)
4816 return folded
4819def _fold_map_reduce_extras(flat: dict, records: list) -> dict:
4820 """The join's per-item state extras: soft errors plus folded side channels.
4822 Mirrors the in-node concurrent path: errors are merged in item order with
4823 dedup, and the budget/side-channel counters follow the same item order.
4824 """
4825 extras: dict = {}
4826 base_errors = list(flat.get("errors") or [])
4827 errors = _fold_map_reduce_errors(records, base_errors)
4828 if errors != base_errors:
4829 extras["errors"] = errors
4830 for state_key, record_key in (
4831 ("step_context_metrics", "metrics"),
4832 ("step_tool_meta", "tool_meta"),
4833 ):
4834 folded = _fold_map_reduce_side_channel(records, record_key)
4835 if folded:
4836 extras[state_key] = _merge_side_channel(flat.get(state_key), folded)
4837 return extras
4840def _make_map_reduce_join_fn(
4841 step: dict,
4842 plan: _MapReducePlan,
4843 step_runner: Any,
4844 channel_names: frozenset[str],
4845 channel_types: dict[str, str],
4846) -> Any:
4847 step_id = step.get("id", "?")
4849 def join(state: PipelineState) -> PipelineState:
4850 flat = _flatten_channel_state(state) if channel_names else state
4851 accumulated = state.get(_MAP_REDUCE_ACCUMULATOR_KEY) or {}
4852 entries = accumulated.get(step_id) or {}
4853 clear = {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: None}}
4854 ctx = step_runner._make_ctx(flat)
4855 items = plan.item_lister(flat, ctx)
4856 if items is None:
4857 return clear
4858 records = [entries.get(index) for index in range(len(items))]
4859 results = [
4860 record.get("result") if isinstance(record, dict) else None
4861 for record in records
4862 ]
4863 step_result = plan.result_assembler(results)
4864 extras = _fold_map_reduce_extras(flat, records)
4865 if not channel_names:
4866 return {step_id: step_result, **extras, **clear}
4867 working_set = ChannelWorkingSet(flat, channel_types)
4868 plan.channel_write_stager(results, records, working_set)
4869 written_names = {name for name, _ in working_set.pending_writes()}
4870 channels = working_set.commit()
4871 emitted = {name: channels[name] for name in written_names}
4872 core = _strip_channel_keys(
4873 {**flat, step_id: step_result, **extras}, channel_names
4874 )
4875 return {_CORE: {**core, **emitted}, **clear}
4877 return join
4880def _make_parallel_map_reduce_plan(
4881 step: dict, step_runner: Any
4882) -> _MapReducePlan:
4883 """Plan for a concurrent `parallel` step: one agent call per item."""
4884 resolved_agent = step_runner.resolve_agent(step)
4886 def run_item(state, index, item, item_state, item_ctx):
4887 result, metrics, tool_meta, errors = step_runner._invoke_parallel_item(
4888 step, resolved_agent, item, item_state, item_ctx
4889 )
4890 return _MapReduceItemOutcome(
4891 result, metrics, tool_meta, (), tuple(errors)
4892 )
4894 return _MapReducePlan(
4895 item_lister=lambda flat, ctx: step_runner._parallel_fanout_item_list(
4896 step, flat, ctx
4897 ),
4898 item_runner=run_item,
4899 result_assembler=lambda results: {"results": results},
4900 channel_write_stager=lambda results, records, working_set: (
4901 step_runner._stage_fanout_channel_writes(step, results, working_set)
4902 ),
4903 )
4906def _worker_committed_flat(
4907 state: PipelineState, channel_names: frozenset[str]
4908) -> dict:
4909 """The committed channel snapshot visible to one worker.
4911 A channels-path worker receives the flattened snapshot as its payload, so
4912 its channel keys sit at the top level; the typed ``__core__`` shape (only
4913 seen by the join) is unwrapped.
4914 """
4915 if _CORE in state:
4916 return _flatten_channel_state(state)
4917 return {name: state[name] for name in channel_names if name in state}
4920def _pending_channel_writes(
4921 working_set: ChannelWorkingSet | None,
4922) -> tuple[tuple[str, Any], ...]:
4923 if working_set is None:
4924 return ()
4925 return tuple(working_set.pending_writes())
4928def _stage_map_reduce_item_channel_writes(
4929 results: list, records: list, working_set: ChannelWorkingSet
4930) -> None:
4931 """Stage each item's carried channel writes in item order."""
4932 for record in records:
4933 if not isinstance(record, dict):
4934 continue
4935 for name, value in record.get("writes") or ():
4936 working_set.stage(name, value)
4939def _make_foreach_map_reduce_plan(
4940 step: dict,
4941 step_runner: Any,
4942 channel_names: frozenset[str],
4943 channel_types: dict[str, str],
4944) -> _MapReducePlan:
4945 """Plan for a concurrent `foreach` step: the body (or agent) runs per item."""
4946 step_id = step.get("id", "?")
4947 body = step.get("body") or []
4948 declared_output = step.get("output")
4949 runs_body = bool(body)
4951 def result_assembler(results):
4952 if runs_body:
4953 return StepRunner._aggregate_body_results(results)
4954 return results
4956 def run_item(state, index, item, item_state, _item_ctx):
4957 item_working_set = (
4958 ChannelWorkingSet(
4959 _worker_committed_flat(state, channel_names), channel_types
4960 )
4961 if channel_names
4962 else None
4963 )
4964 iter_state = {
4965 **item_state, "item": item, "_item": item, "_foreach_item": item
4966 }
4967 if runs_body:
4968 for sub_step in body:
4969 iter_state = step_runner._run_body_step(
4970 sub_step, iter_state, item_working_set
4971 )
4972 return _MapReduceItemOutcome(
4973 step_runner._body_item_result(iter_state, declared_output),
4974 iter_state.get("step_context_metrics"),
4975 iter_state.get("step_tool_meta"),
4976 _pending_channel_writes(item_working_set),
4977 tuple(iter_state.get("errors") or []),
4978 )
4979 resolved = step_runner.resolve_agent(step)
4980 item_read_state = step_runner._read_state(iter_state, item_working_set)
4981 item_ctx = step_runner._make_ctx(item_read_state)
4982 resolved_step = step_runner._resolve_step_runtime(
4983 step, item_read_state, item_ctx
4984 )
4985 resolved_step = step_runner._attach_communication_context(
4986 resolved_step, item_read_state
4987 )
4988 resolved_step = step_runner._attach_resolved_tools(
4989 resolved_step, resolved
4990 )
4991 result_state = step_runner._invoke_agent(
4992 resolved, resolved_step, iter_state
4993 )
4994 return _MapReduceItemOutcome(
4995 result_state.get(step_id),
4996 result_state.get("step_context_metrics"),
4997 result_state.get("step_tool_meta"),
4998 _pending_channel_writes(item_working_set),
4999 tuple(result_state.get("errors") or []),
5000 )
5002 return _MapReducePlan(
5003 item_lister=lambda flat, ctx: step_runner._foreach_item_list(
5004 step, flat, ctx
5005 ),
5006 item_runner=run_item,
5007 result_assembler=result_assembler,
5008 channel_write_stager=_stage_map_reduce_item_channel_writes,
5009 )
5012def _make_map_reduce_parts(
5013 step: dict,
5014 plan: _MapReducePlan,
5015 step_runner: Any,
5016 channel_names: frozenset[str],
5017 channel_types: dict[str, str],
5018 max_concurrent: int,
5019) -> _MapReduceParts:
5020 step_id = step.get("id", "?")
5021 worker_id = f"{step_id}__worker"
5022 fallback_id = f"{step_id}__fallback"
5023 join_id = f"{step_id}__join"
5024 gate = threading.Semaphore(max(1, max_concurrent))
5025 return _MapReduceParts(
5026 dispatch_id=step_id,
5027 worker_id=worker_id,
5028 fallback_id=fallback_id,
5029 join_id=join_id,
5030 dispatch_fn=lambda state: {},
5031 route_fn=_make_map_reduce_route_fn(
5032 plan, step_runner, channel_names, worker_id, fallback_id, join_id
5033 ),
5034 worker_fn=_make_map_reduce_worker_fn(step, plan, step_runner, gate),
5035 join_fn=_make_map_reduce_join_fn(
5036 step, plan, step_runner, channel_names, channel_types
5037 ),
5038 )
5041def _resolved_send_concurrency(step: dict, build_ctx: dict) -> int:
5042 """The build-time resolved ``max_concurrent`` for a Send-eligible step."""
5043 return _resolve_int(step.get("max_concurrent"), build_ctx, 1)
5046def _is_send_parallel_step(
5047 step: dict, route_plan: "_RoutePlan", build_ctx: dict
5048) -> bool:
5049 """Whether a top-level `parallel` step fans out through native ``Send``.
5051 Restricted to a singleton group's statically concurrent, agent-backed,
5052 item-sourced step with no `routes`, no `condition`/`skip_llm_if`, and an
5053 explicit `max_concurrent > 1`; everything else keeps the in-node path.
5054 """
5055 if step.get("type") != "parallel":
5056 return False
5057 if step.get("id") in route_plan.target_node_ids:
5058 return False
5059 if step.get("routes") or not step.get("agent"):
5060 return False
5061 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
5062 return False
5063 if StepRunner._declared_item_source(step) is None:
5064 return False
5065 if step.get("max_concurrent") is None:
5066 return False
5067 return _resolved_send_concurrency(step, build_ctx) > 1
5070def _is_send_foreach_step(
5071 step: dict, route_plan: "_RoutePlan", build_ctx: dict
5072) -> bool:
5073 """Whether a top-level `foreach` step fans out through native ``Send``.
5075 Restricted to a singleton group's body- or agent-backed, item-sourced step
5076 with no `routes`, no `condition`/`skip_llm_if`, no step-level
5077 `output_channel`, and no `max_output`; an explicit `max_concurrent > 1` is
5078 required. The default (absent or ``1``) keeps the sequential in-node path,
5079 and the excluded edges keep their existing in-node semantics.
5080 """
5081 if step.get("type") != "foreach":
5082 return False
5083 if step.get("id") in route_plan.target_node_ids:
5084 return False
5085 if step.get("routes") or step.get("output_channel"):
5086 return False
5087 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
5088 return False
5089 if step.get("max_output") is not None:
5090 return False
5091 if StepRunner._declared_item_source(step) is None:
5092 return False
5093 if not (step.get("body") or step.get("agent")):
5094 return False
5095 if step.get("max_concurrent") is None:
5096 return False
5097 return _resolved_send_concurrency(step, build_ctx) > 1
5100class _ChannelGraphShim:
5101 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
5103 Preserves the invoke contract used by main.py and tests: a flat dict in,
5104 a flat dict out. The flat input is seeded through ``seed_committed`` and
5105 nested under ``__core__`` for the compiled graph; the committed snapshot is
5106 unwrapped back out. Streaming and state surfaces are not exposed because
5107 they would leak or accept the internal ``__core__`` shape; read-only
5108 introspection delegates to the compiled graph.
5109 """
5111 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
5112 self._compiled = compiled
5113 self._specs = specs
5115 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
5116 seeded = seed_committed(self._specs, flat_state)
5117 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
5118 return unwrap(result)
5120 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
5121 seeded = seed_committed(self._specs, flat_state)
5122 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
5123 return unwrap(result)
5125 def stream(self, *args: Any, **kwargs: Any) -> Any:
5126 self._reject_internal_shape("stream")
5128 def astream(self, *args: Any, **kwargs: Any) -> Any:
5129 self._reject_internal_shape("astream")
5131 def update_state(self, *args: Any, **kwargs: Any) -> Any:
5132 self._reject_internal_shape("update_state")
5134 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
5135 self._reject_internal_shape("aupdate_state")
5137 def get_state(self, *args: Any, **kwargs: Any) -> Any:
5138 self._reject_internal_shape("get_state")
5140 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
5141 self._reject_internal_shape("aget_state")
5143 def _reject_internal_shape(self, surface: str) -> None:
5144 raise NotImplementedError(
5145 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
5146 ".invoke()/.ainvoke() contract is exposed through the channel "
5147 "boundary shim"
5148 )
5150 def __getattr__(self, name: str) -> Any:
5151 return getattr(self._compiled, name)
5154class GraphBuilder:
5155 """
5156 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
5158 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
5159 käytetään vain {{ }} -templatejen evaluointiin.
5160 """
5162 def __init__(
5163 self,
5164 registry: ManifestRegistry,
5165 builtins: BuiltinExecutor | None = None,
5166 invoke_agent_fn: Any | None = None,
5167 run_context: dict | None = None,
5168 tracer: Any | None = None,
5169 ) -> None:
5170 self._registry = registry
5171 self._resolver = AgentResolver(registry)
5172 self._invoke_agent_fn = invoke_agent_fn
5173 self._builtins = builtins or _DEFAULT_BUILTINS
5174 if invoke_agent_fn is not None:
5175 self._builtins.invoke_agent_fn = invoke_agent_fn
5176 self._step_functions = StepFunctionExecutor(registry)
5177 self._tool_registry = build_platform_registry()
5178 self._tool_resolver = ToolResolver(self._tool_registry, self._step_functions)
5179 self._step_runner = StepRunner(
5180 self._resolver, self._builtins, invoke_agent_fn, run_context,
5181 step_functions=self._step_functions,
5182 tracer=tracer,
5183 tool_resolver=self._tool_resolver,
5184 )
5186 def build(self, pipeline: dict): # -> CompiledGraph
5187 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
5188 try:
5189 from langgraph.graph import StateGraph, END
5190 except ImportError as exc:
5191 raise ImportError(
5192 "langgraph is required to build a workflow. "
5193 "Install it with: pip install langgraph"
5194 ) from exc
5196 raw_steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
5198 channels: list[dict] = _validated_channels(
5199 (pipeline.get("spec") or {}).get("channels")
5200 )
5201 specs = parse_channel_specs(
5202 [{"name": ch["name"], "type": ch.get("type", LAST_VALUE)} for ch in channels]
5203 )
5204 channel_types = {spec.name: spec.type for spec in specs}
5205 channel_names = frozenset(channel_types)
5207 # Build-time call-step composition (ADR-0037 D4): each invoked Function
5208 # body is compiled into its own native LangGraph subgraph node, so the
5209 # runtime graph executes only native steps; a call step never survives.
5210 steps = self._compose_calls(
5211 raw_steps, channels, channel_names, channel_types
5212 )
5213 if channels:
5214 _validate_channel_names(channels, steps)
5215 _validate_channel_inputs(channels, steps)
5216 _validate_step_channel_bindings(steps, channel_names)
5217 ordered = _topological_order(steps)
5219 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
5220 envs = (pipeline.get("spec") or {}).get("envs")
5221 if envs is not None and not isinstance(envs, dict):
5222 raise RunHardFailure(
5223 "VALIDATION_ERROR",
5224 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
5225 "spec.envs must be a map of env name → image ref (ADR-0012)",
5226 )
5228 graph = self._assemble_graph(ordered, channel_names, channel_types)
5229 compiled = graph.compile()
5230 if channel_names:
5231 return _ChannelGraphShim(compiled, specs)
5232 return compiled
5234 def _compose_calls(
5235 self,
5236 steps: list[dict],
5237 channels: list[dict],
5238 channel_names: frozenset[str],
5239 channel_types: dict[str, str],
5240 ) -> list[dict]:
5241 """Replace `call` steps with compiled native subgraph nodes.
5243 Each invoked Function body is assembled through ``_assemble_graph`` and
5244 compiled, then substituted as a single ``__subgraph__`` placeholder that
5245 keeps the call id and its ``depends_on``. Downstream steps addressing the
5246 call id therefore address the subgraph node unchanged.
5247 """
5249 def assemble_subgraph(sub_steps: list[dict]):
5250 return self._assemble_graph(
5251 sub_steps, channel_names, channel_types
5252 ).compile()
5254 def assemble_while(while_step: dict, body_steps: list[dict]):
5255 return self._assemble_while_graph(
5256 while_step, body_steps, channel_names, channel_types
5257 )
5259 def assemble_foreach(foreach_step: dict, body_steps: list[dict]):
5260 return self._assemble_foreach_graph(
5261 foreach_step, body_steps, channel_names, channel_types
5262 )
5264 composer = _CallComposer(
5265 self._registry,
5266 assemble_subgraph,
5267 channels,
5268 channel_names,
5269 assemble_while=assemble_while,
5270 compose_whiles=True,
5271 assemble_foreach=assemble_foreach,
5272 compose_foreachs=True,
5273 build_ctx=self._step_runner._make_ctx({}),
5274 )
5275 return composer.compose(steps)
5277 def _assemble_while_graph(
5278 self,
5279 step: dict,
5280 body_steps: list[dict],
5281 channel_names: frozenset[str],
5282 channel_types: dict[str, str],
5283 ) -> tuple[Any, int]:
5284 """Compile a `while` step's body into a native LangGraph cycle.
5286 The cycle is ``head -> body... -> cond`` with a conditional back-edge
5287 from ``cond`` to ``head`` (continue) or ``exit`` (termination): the
5288 entry condition/initial value live in the head, the feed + exit
5289 evaluation in the cond node, and the exit node writes the documented
5290 result. Returns the compiled cycle and the recursion bound the wrapper
5291 must apply (a cycle consumes one super-step per node per iteration).
5292 """
5293 from langgraph.graph import StateGraph, END
5295 step_id = step.get("id", "?")
5296 step_runner = self._step_runner
5297 step_runner.set_channel_routing(channel_names)
5299 state_schema = (
5300 _build_state_schema()
5301 if channel_names
5302 else Annotated[dict[str, Any], _merge_state]
5303 )
5304 graph: StateGraph = StateGraph(state_schema)
5306 max_iter = _resolve_int(
5307 step.get("max_iterations", 2), step_runner._make_ctx({}), 2
5308 )
5309 head_id = f"__while_{step_id}__head"
5310 cond_id = f"__while_{step_id}__cond"
5311 exit_id = f"__while_{step_id}__exit"
5312 first_id = body_steps[0]["id"] if body_steps else cond_id
5314 graph.add_node(
5315 head_id,
5316 _make_while_head_node(step, step_runner, channel_names, max_iter),
5317 )
5318 for sub in body_steps:
5319 graph.add_node(
5320 sub["id"],
5321 _make_while_body_node(
5322 sub, step_runner, channel_names, channel_types
5323 ),
5324 )
5325 graph.add_node(
5326 cond_id, _make_while_cond_node(step, step_runner, channel_names)
5327 )
5328 graph.add_node(
5329 exit_id, _make_while_exit_node(step, step_runner, channel_names)
5330 )
5332 graph.set_entry_point(head_id)
5333 head_paths = {first_id: first_id, exit_id: exit_id, END: END}
5334 graph.add_conditional_edges(
5335 head_id,
5336 _make_while_head_router(step, channel_names, first_id, exit_id, END),
5337 head_paths,
5338 )
5339 for previous, sub in zip(body_steps, body_steps[1:]):
5340 graph.add_edge(previous["id"], sub["id"])
5341 if body_steps:
5342 graph.add_edge(body_steps[-1]["id"], cond_id)
5343 graph.add_conditional_edges(
5344 cond_id,
5345 _make_while_cond_router(step, channel_names, head_id, exit_id),
5346 {head_id: head_id, exit_id: exit_id},
5347 )
5348 graph.add_edge(exit_id, END)
5350 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5351 return graph.compile(), recursion_limit
5353 def _assemble_foreach_graph(
5354 self,
5355 step: dict,
5356 body_steps: list[dict],
5357 channel_names: frozenset[str],
5358 channel_types: dict[str, str],
5359 ) -> tuple[Any, int]:
5360 """Compile a sequential `foreach` step's body into a native LangGraph cycle.
5362 The cycle is ``head -> (body... | agent) -> cond`` with a conditional
5363 back-edge from ``cond`` to ``head`` (next item) or ``exit`` (exhaustion
5364 or `max_output` reached). The head resolves the item list once and
5365 injects the current item; the cond collects the item result; the exit
5366 node writes the documented aggregate. Returns the compiled cycle and
5367 the recursion bound the wrapper must apply.
5368 """
5369 from langgraph.graph import StateGraph, END
5371 step_id = step.get("id", "?")
5372 step_runner = self._step_runner
5373 step_runner.set_channel_routing(channel_names)
5375 state_schema = (
5376 _build_state_schema()
5377 if channel_names
5378 else Annotated[dict[str, Any], _merge_state]
5379 )
5380 graph: StateGraph = StateGraph(state_schema)
5382 max_iter = _resolve_int(
5383 step.get("max_iterations"),
5384 step_runner._make_ctx({}),
5385 _FOREACH_DEFAULT_MAX_ITER,
5386 )
5387 head_id = f"__foreach_{step_id}__head"
5388 cond_id = f"__foreach_{step_id}__cond"
5389 exit_id = f"__foreach_{step_id}__exit"
5390 agent_id = f"__foreach_{step_id}__agent"
5392 graph.add_node(
5393 head_id, _make_foreach_head_node(step, step_runner, channel_names)
5394 )
5395 if body_steps:
5396 for sub in body_steps:
5397 graph.add_node(
5398 sub["id"],
5399 _make_foreach_body_node(
5400 sub, step_runner, channel_names, channel_types
5401 ),
5402 )
5403 first_id = body_steps[0]["id"]
5404 else:
5405 graph.add_node(
5406 agent_id,
5407 _make_foreach_agent_node(
5408 step, step_runner, channel_names, channel_types
5409 ),
5410 )
5411 first_id = agent_id
5412 graph.add_node(
5413 cond_id, _make_foreach_cond_node(step, step_runner, channel_names)
5414 )
5415 graph.add_node(
5416 exit_id, _make_foreach_exit_node(step, step_runner, channel_names)
5417 )
5419 graph.set_entry_point(head_id)
5420 graph.add_conditional_edges(
5421 head_id,
5422 _make_foreach_head_router(step, channel_names, first_id, exit_id, END),
5423 {first_id: first_id, exit_id: exit_id, END: END},
5424 )
5425 for previous, sub in zip(body_steps, body_steps[1:]):
5426 graph.add_edge(previous["id"], sub["id"])
5427 if body_steps:
5428 graph.add_edge(body_steps[-1]["id"], cond_id)
5429 else:
5430 graph.add_edge(agent_id, cond_id)
5431 graph.add_conditional_edges(
5432 cond_id,
5433 _make_foreach_cond_router(step, channel_names, head_id, exit_id),
5434 {head_id: head_id, exit_id: exit_id},
5435 )
5436 graph.add_edge(exit_id, END)
5438 recursion_limit = max(25, (len(body_steps) + 3) * (max_iter + 1) + 10)
5439 return graph.compile(), recursion_limit
5441 def _assemble_graph(
5442 self,
5443 steps: list[dict],
5444 channel_names: frozenset[str],
5445 channel_types: dict[str, str],
5446 ) -> "StateGraph":
5447 from langgraph.graph import StateGraph, END
5449 # Group steps into LangGraph nodes:
5450 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
5451 nodes: list[list[dict]] = []
5452 for step in steps:
5453 dep_sig = frozenset(step.get("depends_on") or [])
5454 if nodes:
5455 last_group_sig = frozenset(
5456 dep
5457 for s in nodes[-1]
5458 for dep in (s.get("depends_on") or [])
5459 )
5460 last_types = {s.get("type") for s in nodes[-1]}
5461 if (
5462 dep_sig == last_group_sig
5463 and step.get("type") == "parallel"
5464 and last_types == {"parallel"}
5465 and not step.get("routes")
5466 and not any(s.get("routes") for s in nodes[-1])
5467 ):
5468 nodes[-1].append(step)
5469 continue
5470 nodes.append([step])
5472 state_schema = (
5473 _build_state_schema()
5474 if channel_names
5475 else Annotated[dict[str, Any], _merge_state]
5476 )
5477 graph: StateGraph = StateGraph(state_schema)
5478 step_runner = self._step_runner
5479 step_runner.set_channel_routing(channel_names)
5481 node_ids: list[str] = []
5482 for group in nodes:
5483 if len(group) == 1:
5484 node_ids.append(group[0]["id"])
5485 else:
5486 node_ids.append("parallel_" + "_".join(s["id"] for s in group))
5488 route_plan = _build_route_plan(nodes, node_ids) if nodes else None
5490 send_parts: dict[int, _MapReduceParts] = {}
5491 if route_plan is not None:
5492 build_ctx = step_runner._make_ctx({})
5493 for index, group in enumerate(nodes):
5494 if len(group) != 1:
5495 continue
5496 send_candidate = group[0]
5497 if _is_send_parallel_step(send_candidate, route_plan, build_ctx):
5498 plan = _make_parallel_map_reduce_plan(
5499 send_candidate, step_runner
5500 )
5501 elif _is_send_foreach_step(send_candidate, route_plan, build_ctx):
5502 plan = _make_foreach_map_reduce_plan(
5503 send_candidate, step_runner, channel_names, channel_types
5504 )
5505 else:
5506 continue
5507 send_parts[index] = _make_map_reduce_parts(
5508 send_candidate,
5509 plan,
5510 step_runner,
5511 channel_names,
5512 channel_types,
5513 _resolved_send_concurrency(send_candidate, build_ctx),
5514 )
5516 entry_ids: list[str] = []
5517 exit_ids: list[str] = []
5518 for index, group in enumerate(nodes):
5519 parts = send_parts.get(index)
5520 if parts is None:
5521 captured_group = list(group)
5522 if len(captured_group) == 1:
5523 first = captured_group[0]
5524 if first.get("type") == _SUBGRAPH_STEP_TYPE:
5525 if first.get(_SUBGRAPH_WHILE_KEY):
5526 subgraph_node = _make_while_subgraph_node(
5527 first, step_runner, channel_names
5528 )
5529 elif first.get(_SUBGRAPH_FOREACH_KEY):
5530 subgraph_node = _make_foreach_subgraph_node(
5531 first, step_runner, channel_names
5532 )
5533 else:
5534 subgraph_node = _make_subgraph_node(
5535 first[_SUBGRAPH_GRAPH_KEY],
5536 first.get("condition"),
5537 step_runner,
5538 channel_names,
5539 first.get(_SUBGRAPH_STEP_IDS_KEY),
5540 )
5541 graph.add_node(node_ids[index], subgraph_node)
5542 entry_ids.append(node_ids[index])
5543 exit_ids.append(node_ids[index])
5544 continue
5545 fn = _make_step_node(
5546 first, step_runner, channel_names, channel_types,
5547 )
5548 else:
5549 fn = _make_parallel_node(
5550 captured_group,
5551 step_runner,
5552 channel_names,
5553 channel_types,
5554 )
5555 graph.add_node(node_ids[index], fn)
5556 entry_ids.append(node_ids[index])
5557 exit_ids.append(node_ids[index])
5558 continue
5559 fallback_step = group[0]
5560 graph.add_node(parts.dispatch_id, parts.dispatch_fn)
5561 graph.add_node(parts.worker_id, parts.worker_fn)
5562 graph.add_node(
5563 parts.fallback_id,
5564 _make_step_node(
5565 fallback_step, step_runner, channel_names, channel_types
5566 ),
5567 )
5568 graph.add_node(parts.join_id, parts.join_fn)
5569 graph.add_conditional_edges(
5570 parts.dispatch_id,
5571 parts.route_fn,
5572 [parts.worker_id, parts.fallback_id],
5573 )
5574 graph.add_edge(parts.worker_id, parts.join_id)
5575 graph.add_edge(parts.fallback_id, parts.join_id)
5576 entry_ids.append(parts.dispatch_id)
5577 exit_ids.append(parts.join_id)
5579 if nodes:
5580 _wire_dependency_edges(
5581 graph, nodes, node_ids, END, route_plan, entry_ids, exit_ids
5582 )
5583 _wire_route_edges(graph, route_plan, END)
5584 else:
5585 graph.add_node("noop", lambda s: s)
5586 graph.set_entry_point("noop")
5587 graph.add_edge("noop", END)
5589 return graph
5592# ---------------------------------------------------------------------------
5593# Julkinen API
5594# ---------------------------------------------------------------------------
5596_registry_cache: ManifestRegistry | None = None
5599def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
5600 global _registry_cache
5601 if _registry_cache is None or dirs is not None:
5602 loader = ManifestLoader(dirs)
5603 manifests = loader.load_all()
5604 _registry_cache = ManifestRegistry(manifests)
5605 return _registry_cache
5608def build_workflow(
5609 pipeline_name: str,
5610 *,
5611 dirs: list[Path] | None = None,
5612 builtins: BuiltinExecutor | None = None,
5613 invoke_agent_fn: Any | None = None,
5614 run_context: dict | None = None,
5615 tracer: Any | None = None,
5616):
5617 """
5618 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
5619 LangGraph-workflown.
5621 Args:
5622 pipeline_name: Pipeline metadata.name -arvo.
5623 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
5624 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
5625 invoke_agent_fn: fn(resolved_spec, step, state) → state.
5626 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
5628 Returns:
5629 Compilattu LangGraph CompiledGraph.
5630 """
5631 registry = _get_registry(dirs)
5632 pipeline = registry.get("Pipeline", pipeline_name)
5633 if pipeline is None:
5634 available = [
5635 (m.get("metadata") or {}).get("name")
5636 for m in registry.all_of_kind("Pipeline")
5637 ]
5638 raise RunHardFailure(
5639 "VALIDATION_ERROR",
5640 f"Pipeline '{pipeline_name}' not found. Available: {available}",
5641 )
5642 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
5643 return builder.build(pipeline)