Coverage for core/langgraph/runner.py: 94%
2284 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 00:49 +0000
1"""
2runner.py — YAML-manifest-pohjainen pipeline-moottori.
4Arkkitehtuuri:
5 ManifestLoader → löytää ja parsii *.yaml manifestit
6 ManifestRegistry → indeksoi Agent / AgentInstance / Pipeline -resurssit
7 AgentResolver → yhdistää AgentInstance + based_on Agent (merge)
8 GraphBuilder → rakentaa LangGraph StateGraphin Pipeline-manifestista
9 build_workflow() → palauttaa compilatun LangGraph-workflown nimellä
11Käyttö:
12 from langgraph.runner import build_workflow
13 workflow = build_workflow("review-pr")
14 result = workflow.invoke({"input": "..."})
15"""
16from __future__ import annotations
18import hashlib
19import logging
20import operator
21import os
22import re
23import threading
24from dataclasses import dataclass
25from concurrent.futures import ThreadPoolExecutor, as_completed
26from pathlib import Path
27from typing import Annotated, Any
29import yaml
31from langgraph.types import Send
33# SkillLoadError-import (ei sirkulaari-riskiä — _llm importtaa vain stdlibin):
34# testit ja discover-wrapperit lataavat _llm:n tasaisena moduulina (sys.path
35# insert core/langgraph/), main.py qualified (langgraph._llm, PYTHONPATH=/app).
36# Kummatkin pinnat käännetään samaan _llm-tiedostoon; try/except varmistaa,
37# että luokka on sama objekti kuin nostavalla puolella molemmissa konteksteissa.
38try:
39 from _llm import DEFAULT_MAX_PARALLEL, SkillLoadError, _PARENT_STACK # noqa: F401
40except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
41 from langgraph._llm import ( # noqa: F401
42 DEFAULT_MAX_PARALLEL,
43 SkillLoadError,
44 _PARENT_STACK,
45 )
47# Same dual-import pattern as _llm (flat sys.path and qualified PYTHONPATH=/app)
48try:
49 from _step_functions import StepFunctionExecutor # noqa: F401
50except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
51 from langgraph._step_functions import StepFunctionExecutor # noqa: F401
53# Agent-instruction error code emitted by the engine error set — same
54# dual-import pattern as _llm.
55try:
56 from _instructions import AGENT_INSTRUCTION_MISSING
57except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
58 from langgraph._instructions import AGENT_INSTRUCTION_MISSING
60# Structural tool resolution layer — same dual-import pattern as _llm.
61try:
62 from tools import (
63 RESOLVED_TOOLS_KEY,
64 ToolCollisionError,
65 ToolRegistryError,
66 ToolResolver,
67 build_platform_registry,
68 )
69except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
70 from langgraph.tools import (
71 RESOLVED_TOOLS_KEY,
72 ToolCollisionError,
73 ToolRegistryError,
74 ToolResolver,
75 build_platform_registry,
76 )
78# COMMUNICATION-context projection (declared input.channels reads) — same
79# dual-import pattern as _llm.
80try:
81 from _data_structure import (
82 COMMUNICATION_CONTEXT_KEY,
83 resolve_communication_context,
84 )
85except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
86 from langgraph._data_structure import (
87 COMMUNICATION_CONTEXT_KEY,
88 resolve_communication_context,
89 )
91# Core-owned channel-context storage/commit/seeding primitives — same dual-import
92# pattern. The channel path commits through these; it no longer materializes
93# native LangGraph channels.
94try:
95 from _channels import (
96 CORE_KEY,
97 LAST_VALUE,
98 ChannelSpec,
99 ChannelWorkingSet,
100 parse_channel_specs,
101 seed_committed,
102 unwrap,
103 wrap,
104 )
105except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
106 from langgraph._channels import (
107 CORE_KEY,
108 LAST_VALUE,
109 ChannelSpec,
110 ChannelWorkingSet,
111 parse_channel_specs,
112 seed_committed,
113 unwrap,
114 wrap,
115 )
117# Core-owned reducer layer — same dual-import pattern. The map-reduce
118# accumulator reducer folds concurrent ``Send`` worker writes per step id.
119try:
120 from _reducers import accumulate_map_reduce
121except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
122 from langgraph._reducers import accumulate_map_reduce
124# Agent output envelope error code — same dual-import pattern.
125try:
126 from _envelope import ENVELOPE_ERROR_CODE, ENVELOPE_STATUSES, is_error_envelope
127except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
128 from langgraph._envelope import ( # noqa: F401
129 ENVELOPE_ERROR_CODE,
130 ENVELOPE_STATUSES,
131 is_error_envelope,
132 )
134log = logging.getLogger(__name__)
137class RunHardFailure(Exception):
138 """Run-tason kova virhe: ajo pysähtyy heti (phase failed, exit 1).
140 Nostetaan kun step-tulos kantaa kovan virheen: TRUNCATION_ERROR-koodi
141 (kuratoidun kontekstin katkaisu), EXECUTION_ERROR-koodi (builtinien/stepien
142 deterministinen tekninen vika), TOOL_FAILURE-koodi (agentin työkalupinnan
143 tekninen vika), AGENT_ERROR-koodi (status:error-envelope,
144 ENVELOPE_ERROR_CODE), CALL_NOT_ALLOWED/CALL_INVALID_PARAMS/CALL_FAILED
145 (natiivi deterministic-tool-kutsu), BUILTIN_NAME_COLLISION (nimi sekä
146 builtin- että StepFunction-rekisterissä) tai kontekstiylityksen
147 signatuurin kantava VALIDATION_ERROR (assembled_tokens + context_budget,
148 _llm._context_overflow_result).
149 Kaikki muut virheluokat (budget:*, parse_error, timeout, LLM-virheet)
150 säilyvät nykyisessä partial-semantiikassa. Stepin details kulkeutuu
151 attribuuttina loppu-PATCHin error.details-kenttään.
152 """
154 ENGINE_ERROR_CODES: frozenset[str] = frozenset(
155 {
156 "TIMEOUT",
157 "VALIDATION_ERROR",
158 "CHANNEL_BINDING_ERROR",
159 "INVALID_CHILD_PLAN",
160 AGENT_INSTRUCTION_MISSING,
161 "UNRESOLVED_OUTPUT",
162 "BUILTIN_NAME_COLLISION",
163 "EXECUTION_ERROR",
164 ENVELOPE_ERROR_CODE,
165 "TOOL_FAILURE",
166 "CALL_NOT_ALLOWED",
167 "CALL_INVALID_PARAMS",
168 "CALL_FAILED",
169 "RESOURCE_ERROR",
170 "SPAWN_BUDGET_EXHAUSTED",
171 "EXHAUSTED",
172 "SPAWN_OVER_FLIGHT",
173 "TRUNCATION_ERROR",
174 # BLOCKED is API-owned, not engine-emitted; it stays in the set so
175 # the guard never rejects a contract-valid value.
176 "BLOCKED",
177 }
178 )
180 HARD_ERROR_CODES: frozenset[str] = frozenset(
181 {
182 "TRUNCATION_ERROR",
183 "TOOL_FAILURE",
184 ENVELOPE_ERROR_CODE,
185 "CALL_NOT_ALLOWED",
186 "CALL_INVALID_PARAMS",
187 "CALL_FAILED",
188 "BUILTIN_NAME_COLLISION",
189 }
190 )
192 def __init__(self, code: str, message: str, details: dict | None = None) -> None:
193 if code not in self.ENGINE_ERROR_CODES:
194 raise ValueError(
195 f"RunHardFailure code {code!r} is not a member of the documented "
196 f"engine error-code set {sorted(self.ENGINE_ERROR_CODES)}"
197 )
198 if not isinstance(message, str) or not message:
199 raise ValueError("RunHardFailure message must be a non-empty str")
200 super().__init__(message)
201 self.code = code
202 self.message = message
203 self.details = details
206ENGINE_ERROR_CODES = RunHardFailure.ENGINE_ERROR_CODES
209class SpawnBudget:
210 """Per-tree agent-call budget shared across all spawn nodes.
212 The whole spawn tree draws from a single counter: every agent invocation
213 (the director included) consumes one unit, so the total number of
214 agent invocations the tree may make is bounded by `max_agent_calls`. This is a
215 per-agent-run budget, NOT an LLM-roundtrip budget: a single agent run may
216 make several LLM calls (e.g. an agent-side tool loop over its declared
217 `tools`, resolved into tool entries); those LLM roundtrips are bounded by
218 the run's `max_llm_calls`, not by
219 `max_agent_calls`. There is no engine-level agent retry loop. Thread-safe so
220 a concurrent caller cannot over-commit the shared counter.
221 """
223 def __init__(self, max_agent_calls: int | None) -> None:
224 self._remaining: int | None = (
225 None if max_agent_calls is None else max(0, int(max_agent_calls))
226 )
227 self._lock = threading.Lock()
229 def consume(self) -> bool:
230 """Reserve one agent call. Returns False when the budget is exhausted."""
231 with self._lock:
232 if self._remaining is None:
233 return True
234 if self._remaining <= 0:
235 return False
236 self._remaining -= 1
237 return True
240_SPAWN_SOFT_FAIL_KEYS: frozenset[str] = frozenset({"error", "parse_error", "escalate"})
242_SPAWN_DEFAULT_MAX_AGENT_CALLS: int = 64
243_SPAWN_DEFAULT_MAX_DEPTH: int = 1
244_SPAWN_DEFAULT_MAX_SPAWN_PER_NODE: int = 16
246# The spawn tree's superstep count is bounded by the agent-call budget, not by
247# LangGraph's default recursion cap; lift it so a deep-but-budgeted tree runs.
248_SPAWN_RECURSION_LIMIT: int = 100_000
250# ---------------------------------------------------------------------------
251# Polut
252# ---------------------------------------------------------------------------
254PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", "/project"))
255MANIFEST_DIRS = [
256 PROJECT_ROOT / ".agent-platform" / "workflows",
257 PROJECT_ROOT / ".agent-platform" / "agents",
258]
259SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "manifest.schema.json"
262def resolve_manifest_dirs(clone_root: Path) -> list[Path]:
263 """Palauttaa olemassa olevat manifestihakemistot kloonatusta manifest-reposta.
265 Skannaa ainoastaan `.agent-platform/workflows` ja `.agent-platform/agents` —
266 ei koskaan koko kloonia. [] jos kumpaakaan ei ole (kutsuja → fail-fast).
267 """
268 return [
269 d for d in (
270 clone_root / ".agent-platform" / "workflows",
271 clone_root / ".agent-platform" / "agents",
272 )
273 if d.is_dir()
274 ]
276# ---------------------------------------------------------------------------
277# Template-resoluutio
278# ---------------------------------------------------------------------------
280_TMPL_RE = re.compile(r'\{\{\s*(.*?)\s*\}\}')
282# Sentinel for an unresolvable dot-path segment (distinct from a resolved None).
283_UNRESOLVED = object()
286def _split_key_indices(part: str) -> tuple[str, list[int]]:
287 """Split a dot-path segment into a base key and a list of indices.
289 ``'a[0][1]'`` → ``('a', [0, 1])``; ``'a'`` → ``('a', [])``; ``'[0]'`` →
290 ``('', [0])``. Only non-negative indices are supported (list access).
291 """
292 m = re.fullmatch(r'([^\[]*)((\[\d+\])*)', part)
293 base = m.group(1)
294 indices = [int(x) for x in re.findall(r'\d+', m.group(2))]
295 return base, indices
298def _apply_indices(obj: Any, indices: list[int]) -> Any:
299 """Index a (possibly chained) list/tuple; ``_UNRESOLVED`` on out-of-range."""
300 for idx in indices:
301 if not isinstance(obj, (list, tuple)) or not (0 <= idx < len(obj)):
302 return _UNRESOLVED
303 obj = obj[idx]
304 return obj
307def _dot_get_segment(obj: Any, part: str) -> Any:
308 """Resolve one dot-path segment against a dict or an indexed sequence.
310 A segment may be a bare key (``'a'``), a keyed list-index (``'a[0]'``,
311 ``'a[0][1]'``), or a bare index into a sequence. Returns the resolved value
312 or the ``_UNRESOLVED`` sentinel when the segment cannot be resolved (missing
313 key, non-dict/non-sequence mid-path, or out-of-range index).
314 """
315 base, indices = _split_key_indices(part)
316 if isinstance(obj, dict):
317 if base:
318 if base not in obj:
319 return _UNRESOLVED
320 obj = obj[base]
321 return _apply_indices(obj, indices)
322 if isinstance(obj, (list, tuple)):
323 if indices:
324 return _apply_indices(obj, indices)
325 if base and base.isdigit():
326 return _apply_indices(obj, [int(base)])
327 return _UNRESOLVED
328 return _UNRESOLVED
331def _dot_get(obj: Any, path: str) -> Any:
332 """Navigate nested dicts/lists via dot-path: 'a.b.c' → obj['a']['b']['c'].
334 Supports list-index segments (``results[1].field``) and dot-into-aggregate
335 traversal (missing key / non-dict-non-list mid-path / out-of-range index
336 resolve to None — lenient semantics for non-output refs).
337 """
338 for part in path.split('.'):
339 obj = _dot_get_segment(obj, part)
340 if obj is _UNRESOLVED:
341 return None
342 return obj
345class _ExpressionError(ValueError):
346 """A malformed template/condition expression at the evaluation boundary.
348 Raised by the expression tokenizer, parser, and evaluator so
349 ``_resolve_template`` relabels only design-level expression failures as
350 ``VALIDATION_ERROR``. Subclasses ``ValueError`` so callers that already
351 treat expression failures as ``ValueError`` keep working.
352 """
355class _UnresolvedReference(_ExpressionError):
356 """A strict dot-path read hit an absent key or a non-mapping segment.
358 Subclasses ``ValueError`` (via ``_ExpressionError``) so existing lenient
359 callers that treat any strict read failure as "unresolved"
360 (``__call_emit`` optional outputs, ``_builtins._resolve_dot_path``) keep
361 working while the template/condition boundary can still tell an unresolved
362 reference apart from a parse error.
363 """
366def _dot_get_strict(obj: Any, path: str) -> Any:
367 """Dot-path navigation that raises on an unresolved reference.
369 A reference is unresolved when a key on the path is absent or a segment is
370 not a mapping — used by condition evaluation so an undefined or skipped
371 step's field is never silently evaluated to falsy (ADR-0012 spirit). Also
372 used by ``__call_emit`` to distinguish a resolved output source from an
373 unresolved one (required output → hard error, ADR-0012).
374 """
375 for part in path.split('.'):
376 if obj is None:
377 raise _UnresolvedReference(
378 f"Unresolved reference '{path}': '{part}' is not a mapping"
379 )
380 obj = _dot_get_segment(obj, part)
381 if obj is _UNRESOLVED:
382 raise _UnresolvedReference(
383 f"Unresolved reference '{path}': '{part}' is not a mapping"
384 )
385 return obj
388def materialize_pipeline_output(pipeline_spec: dict, final_state: dict) -> Any:
389 """Resolve a Pipeline's declared run-end output.
391 Reads ``spec.output.source`` (a dot-path) out of ``final_state`` via
392 ``_dot_get`` and returns the surfaced value. Returns ``None`` when no
393 ``spec.output`` is declared, when it has no ``source``, or when the source
394 is unresolved — this is a run-end surfacing (distinct from the emit-time
395 hard-error used for function-interface outputs): an optional run output
396 that cannot be resolved surfaces as None, never a run-end failure.
397 """
398 out_spec = (pipeline_spec or {}).get("output")
399 if not out_spec:
400 return None
401 source = out_spec.get("source")
402 if not source:
403 return None
404 return _dot_get(final_state, source)
407def _merge_side_channel(base: dict | None, additions: dict) -> dict:
408 """Yhdistä rinnakkaisen side-channel-envelopen (step_context_metrics /
409 step_tool_meta) kerrokset: myöhempi kirjoitus voittaa (sama semantiikka kuin
410 state-merge {**state, ...})."""
411 merged = dict(base or {})
412 merged.update(additions)
413 return merged
416def _assemble_spawn_outcome(records: dict) -> tuple[dict, dict, dict]:
417 """Assemble the spawn result envelope and side channels from leaf records.
419 ``records`` maps each node's plan-order path key to its record. Children
420 keys are rebuilt in plan order (paths sort before their descendants), and
421 the director's own envelope is the tree root at ``()``. Side channels fold
422 in plan order (director first, then children).
423 """
424 ordered = sorted(records)
425 envelopes: dict = {}
426 for path in reversed(ordered):
427 record = records[path]
428 children: dict = {}
429 failing: list = []
430 for candidate in ordered:
431 child = records[candidate]
432 if child.get("parent_path") != path:
433 continue
434 child_envelope = envelopes[candidate]
435 children[child["child_id"]] = child_envelope
436 if not child_envelope["passed"]:
437 failing.append(child["child_id"])
438 envelopes[path] = {
439 "result": record["result"],
440 "children": children,
441 "passed": record["self_passed"] and not failing,
442 "failing": failing,
443 "exhausted": record["exhausted"],
444 }
445 metrics: dict = {}
446 tool_meta: dict = {}
447 for path in ordered:
448 metrics = _merge_side_channel(metrics, records[path].get("metrics") or {})
449 tool_meta = _merge_side_channel(
450 tool_meta, records[path].get("tool_meta") or {}
451 )
452 return envelopes[()], metrics, tool_meta
455_MULTI_OPS = ('<=', '>=', '==', '!=')
456_SINGLE_OPS = '+-<>'
457_WORD_BREAK = set("()[],+<=>!'-")
458_CMP_OPS = frozenset(('==', '!=', '<=', '>=', '<', '>'))
461def _tokenize_expr(expr: str) -> list[tuple[str, Any]]:
462 """Split an expression into (kind, value) tokens.
464 Kinds: 'op' (+ - <= >= == != < >), 'punct' ( ) [ ] ,, 'num', 'str',
465 'name'. A 'name' is a dot-path reference or a bare word (e.g. an unquoted
466 comparison literal, true/false/None, and/or/not).
467 """
468 tokens: list[tuple[str, Any]] = []
469 i, n = 0, len(expr)
470 while i < n:
471 ch = expr[i]
472 if ch.isspace():
473 i += 1
474 continue
475 if ch in '()[],':
476 tokens.append(('punct', ch))
477 i += 1
478 continue
479 if ch in '+-<=>!':
480 two = expr[i:i + 2]
481 if two in _MULTI_OPS:
482 tokens.append(('op', two))
483 i += 2
484 continue
485 if ch in _SINGLE_OPS:
486 tokens.append(('op', ch))
487 i += 1
488 continue
489 raise _ExpressionError(f"Invalid operator in expression: {two!r}")
490 if ch in ('\'', '"'):
491 j = i + 1
492 buf: list[str] = []
493 while j < n and expr[j] != ch:
494 buf.append(expr[j])
495 j += 1
496 if j >= n:
497 raise _ExpressionError(f"Unterminated string literal in expression: {expr!r}")
498 tokens.append(('str', ''.join(buf)))
499 i = j + 1
500 continue
501 if ch.isdigit() or (ch == '.' and i + 1 < n and expr[i + 1].isdigit()):
502 j = i
503 if ch == '.':
504 j += 1
505 while j < n and expr[j].isdigit():
506 j += 1
507 tokens.append(('num', float(expr[i:j])))
508 else:
509 while j < n and expr[j].isdigit():
510 j += 1
511 if j < n and expr[j] == '.' and j + 1 < n and expr[j + 1].isdigit():
512 j += 1
513 while j < n and expr[j].isdigit():
514 j += 1
515 tokens.append(('num', float(expr[i:j])))
516 else:
517 tokens.append(('num', int(expr[i:j])))
518 i = j
519 continue
520 j = i
521 while j < n and not expr[j].isspace() and expr[j] not in _WORD_BREAK:
522 j += 1
523 if j == i:
524 raise _ExpressionError(f"Unexpected character in expression: {expr[i:]!r}")
525 tokens.append(('name', expr[i:j]))
526 i = j
527 return tokens
530# AST node shapes:
531# ('lit', value) literal (str/num/bool/None)
532# ('ref', name) dot-path reference
533# ('list', [node, ...]) [a, b, c] literal
534# ('neg', node) unary minus
535# ('not', node) not X
536# ('cmp', op, left, right) comparison
537# ('binop', 'and'|'or'|'+', l, r) binary operator
540class _ExprParser:
541 """Recursive-descent parser for template/condition expressions.
543 Precedence (loosest first): or < and < not < comparison < '+' < unary.
544 Grammar:
545 expr := or_expr
546 or_expr := and_expr ('or' and_expr)*
547 and_expr := not_expr ('and' not_expr)*
548 not_expr := 'not' not_expr | cmp_expr
549 cmp_expr := add_expr (cmp_op add_expr)?
550 add_expr := unary_expr ('+' unary_expr)*
551 unary_expr:= '-' unary_expr | primary
552 primary := num | str | true/false/None | list | '(' expr ')' | name
553 `and`/`or` parse into a full AST but EVALUATE with left-to-right
554 short-circuit, so the right operand is only resolved if needed and an
555 unresolved read there never raises.
556 """
558 def __init__(self, tokens: list[tuple[str, Any]]) -> None:
559 self._tokens = tokens
560 self._i = 0
562 def parse(self) -> tuple:
563 node = self._or_expr()
564 if self._i != len(self._tokens):
565 tok = self._tokens[self._i]
566 raise _ExpressionError(f"Unexpected token in expression: {tok[1]!r}")
567 return node
569 def _peek(self) -> tuple[str, Any] | None:
570 return self._tokens[self._i] if self._i < len(self._tokens) else None
572 def _next(self) -> tuple[str, Any]:
573 tok = self._tokens[self._i]
574 self._i += 1
575 return tok
577 def _peek_name(self, kw: str) -> bool:
578 tok = self._peek()
579 return tok is not None and tok[0] == 'name' and tok[1] == kw
581 def _peek_op(self, op: str) -> bool:
582 tok = self._peek()
583 return tok is not None and tok[0] == 'op' and tok[1] == op
585 def _expect_punct(self, value: str) -> None:
586 tok = self._next()
587 if tok != ('punct', value):
588 raise _ExpressionError(f"Expected '{value}' in expression, got {tok[1]!r}")
590 def _or_expr(self) -> tuple:
591 node = self._and_expr()
592 while self._peek_name('or'):
593 self._next()
594 right = self._and_expr()
595 node = ('binop', 'or', node, right)
596 return node
598 def _and_expr(self) -> tuple:
599 node = self._not_expr()
600 while self._peek_name('and'):
601 self._next()
602 right = self._not_expr()
603 node = ('binop', 'and', node, right)
604 return node
606 def _not_expr(self) -> tuple:
607 if self._peek_name('not'):
608 self._next()
609 return ('not', self._not_expr())
610 return self._cmp_expr()
612 def _cmp_expr(self) -> tuple:
613 node = self._add_expr()
614 tok = self._peek()
615 if tok is not None and tok[0] == 'op' and tok[1] in _CMP_OPS:
616 self._next()
617 right = self._add_expr()
618 node = ('cmp', tok[1], node, right)
619 return node
621 def _add_expr(self) -> tuple:
622 node = self._unary_expr()
623 while self._peek_op('+'):
624 self._next()
625 right = self._unary_expr()
626 node = ('binop', '+', node, right)
627 return node
629 def _unary_expr(self) -> tuple:
630 if self._peek_op('-'):
631 self._next()
632 node = self._unary_expr()
633 if node[0] == 'lit' and isinstance(node[1], (int, float)) and not isinstance(node[1], bool):
634 return ('lit', -node[1])
635 return ('neg', node)
636 return self._primary()
638 def _primary(self) -> tuple:
639 tok = self._peek()
640 if tok is None:
641 raise _ExpressionError("Unexpected end of expression")
642 kind, value = tok
643 if kind == 'num':
644 self._next()
645 return ('lit', value)
646 if kind == 'str':
647 self._next()
648 return ('lit', value)
649 if kind == 'punct':
650 if value == '(':
651 self._next()
652 node = self._or_expr()
653 self._expect_punct(')')
654 return node
655 if value == '[':
656 self._next()
657 items: list[tuple] = []
658 if self._peek() != ('punct', ']'):
659 while True:
660 items.append(self._or_expr())
661 if self._peek() == ('punct', ','):
662 self._next()
663 continue
664 break
665 self._expect_punct(']')
666 return ('list', items)
667 raise _ExpressionError(f"Unexpected punctuation in expression: {value!r}")
668 if kind == 'name':
669 self._next()
670 lowered = value.lower()
671 if lowered == 'true':
672 return ('lit', True)
673 if lowered == 'false':
674 return ('lit', False)
675 if lowered in ('none', 'null'):
676 return ('lit', None)
677 if value in ('and', 'or', 'not'):
678 raise _ExpressionError(f"Unexpected keyword in expression: {value!r}")
679 return ('ref', value)
680 raise _ExpressionError(f"Unexpected token in expression: {value!r}")
683def _is_number(value: Any) -> bool:
684 return isinstance(value, (int, float)) and not isinstance(value, bool)
687def _add_values(left: Any, right: Any) -> Any:
688 """Binary `+`. Lists concatenate, numbers add, strings concatenate."""
689 if isinstance(left, list) and isinstance(right, list):
690 return left + right
691 if _is_number(left) and _is_number(right):
692 return left + right
693 if isinstance(left, str) and isinstance(right, str):
694 return left + right
695 raise _ExpressionError(
696 f"Unsupported operands for '+': {type(left).__name__} + {type(right).__name__}"
697 )
700def _eval_comparison_rhs(node: tuple, ctx: dict, strict: bool) -> Any:
701 """Comparison right-hand side. A bare non-dotted reference compares as a
702 string literal (preserving `profile == PR`); everything else evaluates
703 normally (numbers, quoted strings, dotted refs, lists, grouped exprs)."""
704 if node[0] == 'ref' and '.' not in node[1]:
705 return node[1]
706 return _eval_ast(node, ctx, strict)
709def _eval_cmp(op: str, left_node: tuple, right_node: tuple, ctx: dict, strict: bool) -> bool:
710 left = _eval_ast(left_node, ctx, strict)
711 right = _eval_comparison_rhs(right_node, ctx, strict)
712 try:
713 if op == '==':
714 return left == right
715 if op == '!=':
716 return left != right
717 if op == '<=':
718 return left <= right
719 if op == '>=':
720 return left >= right
721 if op == '<':
722 return left < right
723 if op == '>':
724 return left > right
725 except TypeError:
726 return False
727 raise _ExpressionError(f"Unknown comparison operator: {op!r}")
730def _eval_ast(node: tuple, ctx: dict, strict: bool) -> Any:
731 """Evaluate a parsed AST. `strict=True` raises on an unresolved reference
732 read (condition/exit_condition); `and`/`or` short-circuit left-to-right."""
733 kind = node[0]
734 if kind == 'lit':
735 return node[1]
736 if kind == 'ref':
737 return _dot_get_strict(ctx, node[1]) if strict else _dot_get(ctx, node[1])
738 if kind == 'list':
739 return [_eval_ast(item, ctx, strict) for item in node[1]]
740 if kind == 'neg':
741 operand = _eval_ast(node[1], ctx, strict)
742 if not _is_number(operand):
743 raise _ExpressionError(
744 f"Unsupported operand for unary '-': {type(operand).__name__}"
745 )
746 return -operand
747 if kind == 'not':
748 return not bool(_eval_ast(node[1], ctx, strict))
749 if kind == 'cmp':
750 return _eval_cmp(node[1], node[2], node[3], ctx, strict)
751 if kind == 'binop':
752 op, left_node, right_node = node[1], node[2], node[3]
753 if op == 'and':
754 if not bool(_eval_ast(left_node, ctx, strict)):
755 return False
756 return bool(_eval_ast(right_node, ctx, strict))
757 if op == 'or':
758 if bool(_eval_ast(left_node, ctx, strict)):
759 return True
760 return bool(_eval_ast(right_node, ctx, strict))
761 if op == '+':
762 left = _eval_ast(left_node, ctx, strict)
763 right = _eval_ast(right_node, ctx, strict)
764 return _add_values(left, right)
765 raise _ExpressionError(f"Unknown binary operator: {op!r}")
766 raise _ExpressionError(f"Unknown expression node: {kind!r}")
769def _eval_expr(expr: str, ctx: dict, strict: bool = False) -> Any:
770 """Evaluate a template expression against context dict.
772 `strict` is used by condition evaluation: dot-path access to an unresolved
773 reference raises (ValueError) instead of silently resolving to falsy, and
774 and/or short-circuit left-to-right so a value is never read when the left
775 side already determines the result.
776 """
777 expr = expr.strip()
778 tokens = _tokenize_expr(expr)
779 if not tokens:
780 return _dot_get_strict(ctx, expr) if strict else _dot_get(ctx, expr)
781 return _eval_ast(_ExprParser(tokens).parse(), ctx, strict)
784def _resolve_template(value: Any, ctx: dict, strict: bool = False) -> Any:
785 """Recursively resolve {{ expr }} templates in strings / dicts / lists.
787 The template-evaluation boundary: a design-level failure (an unresolved
788 strict reference, or a malformed expression) surfaces as a typed
789 ``RunHardFailure`` here, never as a raw parser ``ValueError`` that a
790 concurrent worker would record as a code-less per-item error.
791 """
792 try:
793 return _resolve_template_inner(value, ctx, strict)
794 except _UnresolvedReference as exc:
795 raise RunHardFailure(
796 "UNRESOLVED_OUTPUT",
797 str(exc),
798 ) from exc
799 except _ExpressionError as exc:
800 raise RunHardFailure(
801 "VALIDATION_ERROR",
802 f"Invalid template/condition expression: {exc}",
803 ) from exc
806def _resolve_template_inner(value: Any, ctx: dict, strict: bool = False) -> Any:
807 if isinstance(value, str):
808 stripped = value.strip()
809 # Entire string is one template → replace with actual type (not string)
810 m = _TMPL_RE.fullmatch(stripped)
811 if m:
812 return _eval_expr(m.group(1), ctx, strict)
813 # Inline substitution → always returns string
814 def _replace(m: re.Match) -> str:
815 result = _eval_expr(m.group(1), ctx, strict)
816 return '' if result is None else str(result)
817 return _TMPL_RE.sub(_replace, value)
818 if isinstance(value, dict):
819 return {k: _resolve_template_inner(v, ctx, strict) for k, v in value.items()}
820 if isinstance(value, list):
821 return [_resolve_template_inner(v, ctx, strict) for v in value]
822 return value
825def _resolve_input(input_spec: Any, state: dict, ctx: dict) -> Any:
826 """
827 Resolve step input spec to a concrete value:
828 "{{ item }}" → template
829 "[host-ai-context-map-path]" → literal placeholder (verbatim pass-through)
830 "gather.doc_files" → dot-path into state
831 {"k": "a.b"} → dict with each value resolved
832 None → None
833 """
834 if input_spec is None:
835 return None
836 if isinstance(input_spec, str):
837 if '{{' in input_spec:
838 return _resolve_template(input_spec, ctx)
839 stripped = input_spec.lstrip()
840 # A bracket-prefixed literal (e.g. a host-ai-context-map-path
841 # placeholder) is NOT a dot-path reference into state — hand it to the
842 # builtin verbatim so it resolves the placeholder itself.
843 if stripped.startswith('['):
844 return input_spec
845 # Dot-path into state (no braces)
846 return _dot_get(state, stripped)
847 if isinstance(input_spec, dict):
848 return {k: _resolve_input(v, state, ctx) for k, v in input_spec.items()}
849 if isinstance(input_spec, list):
850 return [_resolve_input(v, state, ctx) for v in input_spec]
851 return input_spec
854def _eval_condition(cond: Any, ctx: dict) -> bool:
855 """Evaluate a step condition. None → True (always run).
857 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` whose entry
1892 # condition is false writes a first-class skip result (r.68) instead of
1893 # returning state unchanged, so `_run_while` owns its skip path.
1894 cond = step.get("condition")
1895 if cond is not None and not _eval_condition(cond, ctx) and step_type != "while":
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 == "while":
1977 return self._run_while(step, state, working_set)
1979 if step_type == "call":
1980 # Build-time composition (ADR-0037 D4) rewrites every call step into
1981 # a native subgraph node, so a call reaching the runtime means the
1982 # builder skipped a composition path it was responsible for.
1983 raise RunHardFailure(
1984 "VALIDATION_ERROR",
1985 f"call step '{step_id}' reached runtime without build-time "
1986 "expansion (ADR-0037 D4) — not yet supported",
1987 {"step_id": step_id},
1988 )
1990 if step_type == _SUBGRAPH_STEP_TYPE:
1991 # Build-time composition (ADR-0037 D4) embeds a compiled function
1992 # subgraph as a native node. At the top level ``_assemble_graph``
1993 # wires it through ``_make_subgraph_node``; when the call sits in an
1994 # imperative foreach/while body the placeholder reaches ``run_step``
1995 # instead, so invoke the embedded graph here. The top-of-method
1996 # condition check already handled a false condition (returned the
1997 # input state unchanged), matching the node wrapper's skip.
1998 compiled = step.get(_SUBGRAPH_GRAPH_KEY)
1999 if compiled is None:
2000 raise RunHardFailure(
2001 "VALIDATION_ERROR",
2002 f"subgraph step '{step_id}' has no compiled graph",
2003 {"step_id": step_id},
2004 )
2005 if working_set is not None:
2006 # Channel path: the compiled subgraph shares the parent's
2007 # Core-namespaced schema, while a loop body carries the flat
2008 # committed snapshot. Re-wrap on entry, flatten on exit. No
2009 # channel reducer is re-applied here — the subgraph's own
2010 # commits already produced the authoritative values.
2011 sub_result = compiled.invoke({_CORE: read_state})
2012 committed = _flatten_channel_state(sub_result)
2013 return {**state, **committed}
2014 return compiled.invoke(state)
2016 if step_type == "spawn":
2017 return self._run_spawn(step, state, working_set)
2019 raise RunHardFailure(
2020 "VALIDATION_ERROR",
2021 f"Unknown step type '{step_type}' in step '{step_id}'",
2022 {"step_id": step_id, "step_type": step_type},
2023 )
2025 def _run_parallel_fanout(
2026 self,
2027 step: dict,
2028 state: PipelineState,
2029 items: list,
2030 ctx: dict,
2031 working_set: ChannelWorkingSet | None = None,
2032 ) -> PipelineState:
2033 """
2034 Fan-out: call agent once per item, concurrently.
2035 Results stored as state[step_id] = {"results": [item_result, ...]}.
2036 Each item is available as state["item"] inside the invoke call.
2037 """
2038 step_id = step.get("id", "?")
2039 agent_name = step.get("agent")
2040 if not agent_name:
2041 log.warning("[step] %s parallel fanout: no agent", step_id)
2042 return {**state, step_id: {"results": []}}
2044 resolved = self._resolver.resolve(agent_name)
2046 # Apply max_items cap
2047 max_items_raw = step.get("max_items")
2048 if max_items_raw is not None:
2049 max_items = _resolve_int(max_items_raw, ctx, len(items))
2050 items = items[:max_items]
2052 # Concurrency limit
2053 max_conc_raw = step.get("max_concurrent")
2054 if max_conc_raw is not None:
2055 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2056 else:
2057 max_conc = len(items) or 1
2058 max_conc = max(1, min(max_conc, len(items) or 1))
2060 log.debug(
2061 "[step] %s parallel fanout: %d items, %d concurrent",
2062 step_id, len(items), max_conc,
2063 )
2065 read_state = self._read_state(state, working_set)
2067 def call_one(item: Any) -> tuple[Any, dict | None, dict | None, list]:
2068 item_state = {**read_state, "_item": item, "item": item}
2069 item_ctx = self._make_ctx(item_state)
2070 return self._invoke_parallel_item(
2071 step, resolved, item, item_state, item_ctx
2072 )
2074 # Pre-allocate results to preserve input order regardless of completion
2075 # order (ENG-07), matching the concurrent foreach body-polku.
2076 results: list = [None] * len(items)
2077 item_metrics: dict = {}
2078 item_tool_meta: dict = {}
2079 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2080 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
2081 for future in as_completed(futures):
2082 idx = futures[future]
2083 try:
2084 item_result, item_metrics_map, item_tool_meta_map, *_ = (
2085 future.result()
2086 )
2087 results[idx] = item_result
2088 if isinstance(item_metrics_map, dict):
2089 item_metrics.update(item_metrics_map)
2090 if isinstance(item_tool_meta_map, dict):
2091 item_tool_meta.update(item_tool_meta_map)
2092 except SkillLoadError:
2093 raise
2094 except RunHardFailure:
2095 pool.shutdown(wait=False, cancel_futures=True)
2096 raise
2097 except Exception as exc: # noqa: BLE001
2098 log.error("[step] %s item failed: %s", step_id, exc)
2099 results[idx] = {"error": str(exc)}
2101 self._stage_fanout_channel_writes(step, results, working_set)
2103 final: PipelineState = {**state, step_id: {"results": results}}
2104 if item_metrics:
2105 final = {
2106 **final,
2107 "step_context_metrics": _merge_side_channel(
2108 state.get("step_context_metrics"), item_metrics
2109 ),
2110 }
2111 if item_tool_meta:
2112 final = {
2113 **final,
2114 "step_tool_meta": _merge_side_channel(
2115 state.get("step_tool_meta"), item_tool_meta
2116 ),
2117 }
2118 return final
2120 def _invoke_parallel_item(
2121 self,
2122 step: dict,
2123 resolved: dict,
2124 item: Any,
2125 item_state: PipelineState,
2126 item_ctx: dict,
2127 ) -> tuple[Any, dict | None, dict | None, list]:
2128 """Run one parallel fan-out item's agent call.
2130 Shared by the in-node thread-pool fan-out and the native ``Send``
2131 worker, so both paths resolve input, communication context and tools
2132 identically. Returns ``(step_result, metrics, tool_meta, errors)``.
2133 """
2134 step_id = step.get("id", "?")
2135 resolved_step = self._resolve_step_runtime(
2136 step, item_state, item_ctx,
2137 input_override=item,
2138 )
2139 resolved_step = self._attach_communication_context(
2140 resolved_step, item_state
2141 )
2142 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2143 result_state = self._invoke_agent(resolved, resolved_step, item_state)
2144 return (
2145 result_state.get(step_id),
2146 result_state.get("step_context_metrics"),
2147 result_state.get("step_tool_meta"),
2148 list(result_state.get("errors") or []),
2149 )
2151 def _stage_fanout_channel_writes(
2152 self,
2153 step: dict,
2154 item_outputs: list,
2155 working_set: ChannelWorkingSet | None,
2156 ) -> None:
2157 """Stage each fan-out item's declared channel writes in item order.
2159 Runs on the main thread after the item pool completes, so the shared
2160 working set is never mutated from a worker thread. The fan-out
2161 envelope itself is never routed; only each item's own agent output is.
2162 """
2163 if working_set is None or not self._channel_names:
2164 return
2165 for output in item_outputs:
2166 writes = _declared_output_channel_writes(
2167 step, output, self._channel_names
2168 )
2169 for name, value in writes.items():
2170 working_set.stage(name, value)
2172 def _run_foreach(
2173 self,
2174 step: dict,
2175 state: PipelineState,
2176 working_set: ChannelWorkingSet | None = None,
2177 ) -> PipelineState:
2178 """
2179 Foreach: iterate over input list, run body steps per item.
2180 Input from step.input (dot-path or template) or step.over (state key).
2182 When max_concurrent > 1, body iterations run concurrently via
2183 ThreadPoolExecutor — each item gets its own worker thread. The
2184 max_concurrent value sets the thread pool size (mirrors parallel
2185 fan-out semantics). Results list preserves original item order.
2187 max_output budget in concurrent mode: a stop flag prevents new workers
2188 from starting after the budget is exhausted. Workers already executing
2189 when the flag is set will complete naturally — at most max_concurrent
2190 extra items may run past the limit. This is accepted behaviour; the
2191 sequential early-exit guarantee cannot be reproduced without sacrificing
2192 true concurrency.
2194 The invocation working set is threaded into every body step so reads
2195 resolve over the committed snapshot plus this invocation's pending
2196 writes; the returned state is still built from the raw ``state``.
2197 """
2198 step_id = step.get("id", "?")
2199 read_state = self._read_state(state, working_set)
2200 ctx = self._make_ctx(read_state)
2202 items = self._foreach_item_list(step, read_state, ctx)
2204 # max_output cap — stop once N iterations have produced a successful
2205 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
2206 max_output_raw = step.get("max_output")
2207 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
2209 # max_concurrent — when > 1, enables parallel body execution
2210 max_conc_raw = step.get("max_concurrent")
2211 if max_conc_raw is not None:
2212 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
2213 max_conc = max(1, min(max_conc, len(items) or 1))
2214 else:
2215 max_conc = 1
2217 body = step.get("body") or []
2218 agent_name = step.get("agent")
2220 if max_conc <= 1:
2221 return self._run_foreach_sequential(
2222 step_id, step, state, items, body, agent_name, max_output,
2223 working_set,
2224 )
2225 return self._run_foreach_concurrent(
2226 step_id, step, state, items, body, agent_name, max_output, max_conc,
2227 working_set,
2228 )
2230 @staticmethod
2231 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
2232 """body-polku per-iteration result (r.67): the declared body-output
2233 step's result; else the iteration's `publish` result; else None."""
2234 if declared_output:
2235 return iter_state.get(declared_output)
2236 return iter_state.get("publish")
2238 @staticmethod
2239 def _body_item_passed(item: Any) -> bool:
2240 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
2241 `passed` bool; an item lacking such a key is not counted as failed."""
2242 if isinstance(item, dict):
2243 verdict = item.get("verdict")
2244 if verdict == "FAIL":
2245 return False
2246 if verdict == "PASS":
2247 return True
2248 passed = item.get("passed")
2249 if isinstance(passed, bool):
2250 return passed
2251 return True
2253 @staticmethod
2254 def _aggregate_body_results(item_results: list) -> dict:
2255 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
2256 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
2257 return {
2258 "results": item_results,
2259 "passed": not failing,
2260 "failing": failing,
2261 "failed": bool(failing),
2262 }
2264 def _run_foreach_sequential(
2265 self,
2266 step_id: str,
2267 step: dict,
2268 state: PipelineState,
2269 items: list,
2270 body: list,
2271 agent_name: str | None,
2272 max_output: int,
2273 working_set: ChannelWorkingSet | None = None,
2274 ) -> PipelineState:
2275 """Sequential foreach — default path when max_concurrent is absent or 1."""
2276 body_polku = bool(body)
2277 declared_output = step.get("output")
2278 results: list = []
2279 produced_outputs = 0
2280 accumulated_errors: list = list(state.get("errors") or [])
2281 accumulated_metrics: dict = {}
2282 accumulated_tool_meta: dict = {}
2284 for item in items:
2285 if max_output and produced_outputs >= max_output:
2286 log.info("[foreach] %s max_output=%d reached — stopping", step_id, max_output)
2287 break
2288 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2290 if body_polku:
2291 for sub_step in body:
2292 iter_state = self._run_body_step(sub_step, iter_state, working_set)
2293 # Propagate any budget errors from this iteration to accumulated list
2294 accumulated_errors = self._merge_errors(
2295 accumulated_errors, iter_state.get("errors") or []
2296 )
2297 item_metrics_map = iter_state.get("step_context_metrics")
2298 if isinstance(item_metrics_map, dict):
2299 accumulated_metrics.update(item_metrics_map)
2300 item_tool_meta_map = iter_state.get("step_tool_meta")
2301 if isinstance(item_tool_meta_map, dict):
2302 accumulated_tool_meta.update(item_tool_meta_map)
2303 publish_out = iter_state.get("publish")
2304 results.append(self._body_item_result(iter_state, declared_output))
2305 if isinstance(publish_out, dict) and publish_out.get("path"):
2306 produced_outputs += 1
2307 elif agent_name:
2308 resolved = self._resolver.resolve(agent_name)
2309 iter_read_state = self._read_state(iter_state, working_set)
2310 iter_ctx = self._make_ctx(iter_read_state)
2311 resolved_step = self._resolve_step_runtime(
2312 step, iter_read_state, iter_ctx
2313 )
2314 resolved_step = self._attach_communication_context(
2315 resolved_step, iter_read_state
2316 )
2317 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2318 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2319 results.append(iter_state.get(step_id))
2320 item_metrics_map = iter_state.get("step_context_metrics")
2321 if isinstance(item_metrics_map, dict):
2322 accumulated_metrics.update(item_metrics_map)
2323 item_tool_meta_map = iter_state.get("step_tool_meta")
2324 if isinstance(item_tool_meta_map, dict):
2325 accumulated_tool_meta.update(item_tool_meta_map)
2327 step_result: Any = self._aggregate_body_results(results) if body_polku else results
2328 final: PipelineState = {**state, step_id: step_result}
2329 if accumulated_metrics:
2330 final = {
2331 **final,
2332 "step_context_metrics": _merge_side_channel(
2333 state.get("step_context_metrics"), accumulated_metrics
2334 ),
2335 }
2336 if accumulated_tool_meta:
2337 final = {
2338 **final,
2339 "step_tool_meta": _merge_side_channel(
2340 state.get("step_tool_meta"), accumulated_tool_meta
2341 ),
2342 }
2343 if accumulated_errors != list(state.get("errors") or []):
2344 final = {**final, "errors": accumulated_errors}
2345 return final
2347 @staticmethod
2348 def _merge_item_working_sets(
2349 working_set: ChannelWorkingSet | None,
2350 item_working_sets: list[ChannelWorkingSet | None],
2351 ) -> None:
2352 """Stage each concurrent item's writes onto the shared set in item order."""
2353 if working_set is None:
2354 return
2355 for item_working_set in item_working_sets:
2356 if item_working_set is None:
2357 continue
2358 for name, value in item_working_set.pending_writes():
2359 working_set.stage(name, value)
2361 def _run_foreach_concurrent(
2362 self,
2363 step_id: str,
2364 step: dict,
2365 state: PipelineState,
2366 items: list,
2367 body: list,
2368 agent_name: str | None,
2369 max_output: int,
2370 max_conc: int,
2371 working_set: ChannelWorkingSet | None = None,
2372 ) -> PipelineState:
2373 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2374 log.info(
2375 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2376 step_id, len(items), max_conc,
2377 )
2379 # Pre-allocate results list to preserve item order across threads.
2380 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2381 # legitimate None results (body produced no output) are kept.
2382 body_polku = bool(body)
2383 declared_output = step.get("output")
2384 _SKIPPED = object()
2385 results: list = [_SKIPPED] * len(items)
2386 item_working_sets: list[ChannelWorkingSet | None] = [None] * len(items)
2387 lock = threading.Lock()
2388 produced_outputs = [0] # mutable counter shared across threads
2389 stop_flag = threading.Event() # set when max_output budget is exhausted
2390 item_errors: list = [] # budget errors collected from worker threads
2391 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2392 accumulated_tool_meta: dict = {} # per-item tool metadata from worker threads
2394 def run_item(idx: int, item: Any) -> None:
2395 nonlocal item_errors
2396 # Honour stop flag — skip if budget already exhausted before we start
2397 if max_output and stop_flag.is_set():
2398 return
2400 item_working_set = (
2401 working_set.fork() if working_set is not None else None
2402 )
2403 item_working_sets[idx] = item_working_set
2404 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2406 if body_polku:
2407 for sub_step in body:
2408 iter_state = self._run_body_step(
2409 sub_step, iter_state, item_working_set
2410 )
2411 # Collect budget errors from this iteration (thread-safe merge)
2412 new_errs = iter_state.get("errors") or []
2413 if new_errs:
2414 with lock:
2415 item_errors = self._merge_errors(item_errors, new_errs)
2416 item_metrics_map = iter_state.get("step_context_metrics")
2417 if isinstance(item_metrics_map, dict):
2418 with lock:
2419 accumulated_metrics.update(item_metrics_map)
2420 item_tool_meta_map = iter_state.get("step_tool_meta")
2421 if isinstance(item_tool_meta_map, dict):
2422 with lock:
2423 accumulated_tool_meta.update(item_tool_meta_map)
2424 publish_out = iter_state.get("publish")
2425 results[idx] = self._body_item_result(iter_state, declared_output)
2426 if isinstance(publish_out, dict) and publish_out.get("path"):
2427 with lock:
2428 produced_outputs[0] += 1
2429 if max_output and produced_outputs[0] >= max_output:
2430 stop_flag.set()
2431 elif agent_name:
2432 resolved = self._resolver.resolve(agent_name)
2433 item_read_state = self._read_state(iter_state, item_working_set)
2434 item_ctx = self._make_ctx(item_read_state)
2435 resolved_step = self._resolve_step_runtime(
2436 step, item_read_state, item_ctx
2437 )
2438 resolved_step = self._attach_communication_context(
2439 resolved_step, item_read_state
2440 )
2441 resolved_step = self._attach_resolved_tools(resolved_step, resolved)
2442 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2443 results[idx] = iter_state.get(step_id)
2444 item_metrics_map = iter_state.get("step_context_metrics")
2445 if isinstance(item_metrics_map, dict):
2446 with lock:
2447 accumulated_metrics.update(item_metrics_map)
2448 item_tool_meta_map = iter_state.get("step_tool_meta")
2449 if isinstance(item_tool_meta_map, dict):
2450 with lock:
2451 accumulated_tool_meta.update(item_tool_meta_map)
2453 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2454 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2455 for future in as_completed(futures):
2456 try:
2457 future.result()
2458 except SkillLoadError:
2459 raise
2460 except RunHardFailure:
2461 pool.shutdown(wait=False, cancel_futures=True)
2462 raise
2463 except Exception as exc: # noqa: BLE001
2464 idx = futures[future]
2465 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2466 results[idx] = {"error": str(exc)}
2468 self._merge_item_working_sets(working_set, item_working_sets)
2470 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2471 ordered = [r for r in results if r is not _SKIPPED]
2472 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
2473 final: PipelineState = {**state, step_id: step_result}
2474 if accumulated_metrics:
2475 final = {
2476 **final,
2477 "step_context_metrics": _merge_side_channel(
2478 state.get("step_context_metrics"), accumulated_metrics
2479 ),
2480 }
2481 if accumulated_tool_meta:
2482 final = {
2483 **final,
2484 "step_tool_meta": _merge_side_channel(
2485 state.get("step_tool_meta"), accumulated_tool_meta
2486 ),
2487 }
2488 if item_errors:
2489 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2490 return final
2492 def _run_while(
2493 self,
2494 step: dict,
2495 state: PipelineState,
2496 working_set: ChannelWorkingSet | None = None,
2497 ) -> PipelineState:
2498 """
2499 While loop with exit_condition and optional entry condition.
2501 Fields:
2502 condition: "{{ triage.approved }}" — if false, skip entirely
2503 input: initial value for loop.current
2504 exit_condition: "{{ review.pass }}" — break when true
2505 max_iterations: int or template
2506 output: body step id whose result feeds the next round's
2507 `loop.current` (envelope-unwrapped when dict with `output`)
2508 body: list of sub-steps
2510 Result shape (r.68): {passed, iterations, output, exhausted, skipped}.
2511 The feed comes ONLY from the declared `output` body step — there is no
2512 hardcoded `refine` feed; a body step named `refine` is not special.
2514 The invocation working set is threaded into every body step and into
2515 condition evaluation so reads resolve over the committed snapshot plus
2516 this invocation's pending writes; the returned state is still built
2517 from the raw ``state``.
2518 """
2519 step_id = step.get("id", "?")
2520 read_state = self._read_state(state, working_set)
2521 ctx = self._make_ctx(read_state)
2523 # Entry condition — skip is a first-class state (r.68): a skipped
2524 # while writes a result with output = the resolved input so downstream
2525 # `.output` reads are non-null on the skipped path.
2526 entry_cond = step.get("condition")
2527 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
2528 input_spec = step.get("input")
2529 skip_input = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
2530 return {**state, step_id: {
2531 "passed": True,
2532 "iterations": 0,
2533 "output": skip_input,
2534 "exhausted": False,
2535 "skipped": True,
2536 }}
2538 # Initial current value from input
2539 input_spec = step.get("input")
2540 if input_spec is not None:
2541 current = _resolve_input(input_spec, read_state, ctx)
2542 else:
2543 current = None
2545 exit_condition = step.get("exit_condition")
2546 max_iter = _resolve_int(step.get("max_iterations", 2), ctx, 2)
2547 body = step.get("body") or []
2548 declared_output = step.get("output")
2550 passed = False
2551 iteration = 0
2552 loop_state: PipelineState = dict(state)
2553 accumulated_metrics: dict = {}
2554 accumulated_tool_meta: dict = {}
2556 while iteration < max_iter:
2557 # Inject loop context
2558 loop_state = {
2559 **state,
2560 "loop": {"current": current, "iteration": iteration},
2561 step_id: {"current": current, "passed": False},
2562 }
2563 # Add current item context (from outer foreach if any)
2564 if "item" in state:
2565 loop_state["item"] = state["item"]
2567 # Run body steps
2568 for sub_step in body:
2569 loop_state = self._run_body_step(sub_step, loop_state, working_set)
2571 iter_metrics_map = loop_state.get("step_context_metrics")
2572 if isinstance(iter_metrics_map, dict):
2573 accumulated_metrics.update(iter_metrics_map)
2574 iter_tool_meta_map = loop_state.get("step_tool_meta")
2575 if isinstance(iter_tool_meta_map, dict):
2576 accumulated_tool_meta.update(iter_tool_meta_map)
2578 iteration += 1
2580 # Feed: next round's loop.current comes from the declared body-output
2581 # step (envelope-unwrapped); otherwise current is unchanged.
2582 # Envelopes: a `{output: ...}` (r.68) and a `{results: [...]}`
2583 # aggregate (parallel/foreach body-step) are unwrapped to their
2584 # inner value so the next round's consumer receives the list, not
2585 # the envelope. A raw value (plain list/dict) feeds as-is.
2586 if declared_output:
2587 feed_value = loop_state.get(declared_output)
2588 if isinstance(feed_value, dict) and "output" in feed_value:
2589 current = feed_value["output"]
2590 elif isinstance(feed_value, dict) and isinstance(feed_value.get("results"), list):
2591 current = feed_value["results"]
2592 else:
2593 current = feed_value
2595 # Check exit condition
2596 if exit_condition:
2597 exit_ctx = self._make_ctx(
2598 self._read_state(loop_state, working_set)
2599 )
2600 if _eval_condition(exit_condition, exit_ctx):
2601 passed = True
2602 break
2603 else:
2604 passed = True
2605 break
2607 skipped = False
2608 exhausted = not passed
2609 final: PipelineState = {**state, step_id: {
2610 "passed": passed,
2611 "iterations": iteration,
2612 "output": current,
2613 "exhausted": exhausted,
2614 "skipped": skipped,
2615 }}
2616 # ENG-11: a while that ends exhausted (ran max_iterations without
2617 # exit_condition passing, not skipped) and whose `fail_if_exhausted`
2618 # evaluates true is a hard run failure — nothing downstream runs. A
2619 # skipped or passed while must never raise.
2620 fail_if_exhausted = step.get("fail_if_exhausted")
2621 if (
2622 fail_if_exhausted is not None
2623 and exhausted
2624 and _eval_condition(fail_if_exhausted, self._make_ctx(read_state))
2625 ):
2626 raise RunHardFailure(
2627 "EXHAUSTED",
2628 f"while step '{step_id}' exhausted max_iterations={max_iter} "
2629 "without exit_condition passing (fail_if_exhausted)",
2630 {"step_id": step_id, "exhausted": True},
2631 )
2632 # Propagate any errors accumulated in the loop body back to the outer state
2633 loop_errors = loop_state.get("errors") or []
2634 if loop_errors:
2635 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), loop_errors)}
2636 if accumulated_metrics:
2637 final = {
2638 **final,
2639 "step_context_metrics": _merge_side_channel(
2640 state.get("step_context_metrics"), accumulated_metrics
2641 ),
2642 }
2643 if accumulated_tool_meta:
2644 final = {
2645 **final,
2646 "step_tool_meta": _merge_side_channel(
2647 state.get("step_tool_meta"), accumulated_tool_meta
2648 ),
2649 }
2650 return final
2652 def _run_spawn(
2653 self,
2654 step: dict,
2655 state: PipelineState,
2656 working_set: ChannelWorkingSet | None = None,
2657 ) -> PipelineState:
2658 """Run a spawn step as a native dynamic ``Send`` fan-out subgraph.
2660 The ``agent`` runs once as director at the tree root and may emit a
2661 child plan at ``envelope.content.children``; a shared child-worker node
2662 re-runs the ``child`` role (default = ``agent``) for each child and
2663 re-``Send``s itself for deeper levels. Depth and fan are bounded by the
2664 spawn_policy, and the whole tree draws from one shared SpawnBudget.
2665 """
2666 step_id = step.get("id", "?")
2667 director = step.get("agent")
2668 if not director:
2669 log.warning("[step] %s spawn has no agent — skipping", step_id)
2670 return state
2671 read_state = self._read_state(state, working_set)
2672 ctx = self._make_ctx(read_state)
2673 policy = step.get("spawn_policy") or {}
2674 max_depth = _resolve_int(
2675 policy.get("max_depth"), ctx, _SPAWN_DEFAULT_MAX_DEPTH
2676 )
2677 max_spawn_per_node = _resolve_int(
2678 policy.get("max_spawn_per_node"), ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
2679 )
2680 max_calls_raw = policy.get("max_agent_calls")
2681 budget = SpawnBudget(
2682 _SPAWN_DEFAULT_MAX_AGENT_CALLS
2683 if max_calls_raw is None
2684 else _resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS)
2685 )
2686 run_max_parallel = int(
2687 (self._run_context.get("budget") or {}).get(
2688 "max_parallel", DEFAULT_MAX_PARALLEL
2689 )
2690 )
2691 max_concurrent_raw = policy.get("max_concurrent_children")
2692 max_concurrent_children = (
2693 run_max_parallel
2694 if max_concurrent_raw is None
2695 else _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
2696 )
2697 child_role = step.get("child") or director
2699 outcome = self._run_native_spawn(
2700 step=step,
2701 step_id=step_id,
2702 director=director,
2703 child_role=child_role,
2704 base_state=read_state,
2705 base_ctx=ctx,
2706 budget=budget,
2707 max_depth=max_depth,
2708 max_spawn_per_node=max_spawn_per_node,
2709 max_concurrent_children=max_concurrent_children,
2710 )
2711 envelope = outcome["envelope"]
2712 final: PipelineState = {**state, step_id: envelope}
2713 if outcome["metrics"]:
2714 final = {
2715 **final,
2716 "step_context_metrics": _merge_side_channel(
2717 state.get("step_context_metrics"), outcome["metrics"]
2718 ),
2719 }
2720 if outcome["tool_meta"]:
2721 final = {
2722 **final,
2723 "step_tool_meta": _merge_side_channel(
2724 state.get("step_tool_meta"), outcome["tool_meta"]
2725 ),
2726 }
2727 self._stage_spawn_channel_writes(step, envelope["result"], working_set)
2728 return final
2730 def _run_native_spawn(
2731 self,
2732 *,
2733 step: dict,
2734 step_id: str,
2735 director: str,
2736 child_role: str,
2737 base_state: PipelineState,
2738 base_ctx: dict,
2739 budget: SpawnBudget,
2740 max_depth: int,
2741 max_spawn_per_node: int | None,
2742 max_concurrent_children: int,
2743 ) -> dict:
2744 """Compile and invoke the native spawn tree, returning its outcome.
2746 The graph is a director node that seeds one ``Send`` per child to a
2747 shared worker, whose worker re-``Send``s itself for deeper levels. An
2748 outstanding-descendants barrier (``outstanding`` counter plus a ``check``
2749 node) makes the ``join`` run exactly once for uneven-depth trees; the
2750 join assembles the result envelope by path key in plan order.
2751 """
2752 from contextlib import nullcontext
2753 from typing import TypedDict
2755 from langgraph.graph import END, START, StateGraph
2756 from langgraph.types import Command, Send
2758 tracer = self._tracer
2759 parent_obs = (
2760 tracer.open_parent_observation(
2761 name=f"spawn:{step_id}",
2762 input={"node_id": step_id, "role": director, "depth": 0},
2763 )
2764 if tracer is not None
2765 else None
2766 )
2767 root_obs_id = getattr(parent_obs, "id", None)
2768 try:
2769 child_gates: dict[str, threading.Semaphore] = {}
2770 child_gates_lock = threading.Lock()
2772 def child_gate(parent_id: str) -> threading.Semaphore:
2773 """The concurrency gate for one spawning node's direct children.
2775 ``max_concurrent_children`` bounds a single spawn node's fan-out:
2776 each spawning node owns its own gate, so a shallow node's cap
2777 never throttles a deeper sibling's subtree.
2778 """
2779 with child_gates_lock:
2780 gate = child_gates.get(parent_id)
2781 if gate is None:
2782 gate = threading.Semaphore(max(1, max_concurrent_children))
2783 child_gates[parent_id] = gate
2784 return gate
2786 def run_node(
2787 *,
2788 role: str,
2789 node_step: dict,
2790 input_override: Any,
2791 node_id: str,
2792 depth: int,
2793 parent_id: str | None = None,
2794 ) -> dict:
2795 resolved_role = self._resolver.resolve(role)
2796 resolved_step = self._resolve_step_runtime(
2797 node_step, base_state, base_ctx, input_override=input_override
2798 )
2799 resolved_step = self._attach_communication_context(
2800 resolved_step, base_state
2801 )
2802 resolved_step = self._attach_resolved_tools(
2803 resolved_step, resolved_role
2804 )
2805 if not budget.consume():
2806 raise RunHardFailure(
2807 "SPAWN_BUDGET_EXHAUSTED",
2808 f"spawn tree budget exhausted at node '{node_id}'",
2809 {"node_id": node_id, "depth": depth},
2810 )
2811 gate = child_gate(parent_id) if parent_id is not None else nullcontext()
2812 with gate:
2813 return self._invoke_agent(
2814 resolved_role, resolved_step, base_state
2815 )
2817 def build_record(
2818 *,
2819 node_id: str,
2820 path: tuple,
2821 parent_path: tuple | None,
2822 depth: int,
2823 this_result: Any,
2824 result_state: dict,
2825 children: list,
2826 ) -> tuple[dict, list]:
2827 exhausted = False
2828 plan: list = []
2829 if children:
2830 if depth + 1 > max_depth:
2831 exhausted = True
2832 else:
2833 plan = [
2834 {
2835 "path": path + (index,),
2836 "depth": depth + 1,
2837 "child": child,
2838 "parent_id": node_id,
2839 "index": index,
2840 }
2841 for index, child in enumerate(children)
2842 ]
2843 record = {
2844 "path": path,
2845 "parent_path": parent_path,
2846 "child_id": node_id,
2847 "result": this_result,
2848 "self_passed": not self._spawn_is_soft_fail(this_result),
2849 "exhausted": exhausted,
2850 "metrics": dict(result_state.get("step_context_metrics") or {}),
2851 "tool_meta": dict(result_state.get("step_tool_meta") or {}),
2852 }
2853 return record, plan
2855 def child_plan(node_id: str, this_result: Any, depth: int) -> list:
2856 children = self._spawn_children_from_result(this_result)
2857 if children and max_spawn_per_node is not None and len(children) > max_spawn_per_node:
2858 raise RunHardFailure(
2859 "SPAWN_OVER_FLIGHT",
2860 f"spawn node '{node_id}' emitted {len(children)} children, "
2861 f"exceeds max_spawn_per_node={max_spawn_per_node}",
2862 {
2863 "node_id": node_id,
2864 "depth": depth,
2865 "children": len(children),
2866 "max_spawn_per_node": max_spawn_per_node,
2867 },
2868 )
2869 return children
2871 def director_node(_state: dict) -> Command:
2872 result_state = run_node(
2873 role=director,
2874 node_step=step,
2875 input_override=_UNSET,
2876 node_id=step_id,
2877 depth=0,
2878 )
2879 this_result = result_state.get(step_id)
2880 children = child_plan(step_id, this_result, 0)
2881 record, plan = build_record(
2882 node_id=step_id,
2883 path=(),
2884 parent_path=None,
2885 depth=0,
2886 this_result=this_result,
2887 result_state=result_state,
2888 children=children,
2889 )
2890 update = {
2891 "accum": {step_id: {(): record}},
2892 "outstanding": len(plan) - 1,
2893 }
2894 if not plan:
2895 return Command(update=update, goto="join")
2896 return Command(
2897 update=update,
2898 goto=[
2899 Send("worker", {**payload, "parent_obs_id": root_obs_id})
2900 for payload in plan
2901 ],
2902 )
2904 def worker_node(payload: dict) -> Command:
2905 path = tuple(payload["path"])
2906 depth = payload["depth"]
2907 parent_id = payload["parent_id"]
2908 index = payload["index"]
2909 child = payload["child"]
2910 child_id = (
2911 child["id"]
2912 if isinstance(child, dict) and child.get("id")
2913 else f"{parent_id}.{index}"
2914 )
2915 node_step = self._child_node_step(
2916 node_id=child_id,
2917 child=child,
2918 parent_id=parent_id,
2919 index=index,
2920 base_state=base_state,
2921 base_ctx=base_ctx,
2922 )
2923 # The worker's own graph context does not carry the spawning
2924 # node's observation (each Send target runs in a fresh context),
2925 # so the parent chain is re-established from the payload before
2926 # opening this node's observation (ADR-0035 D5).
2927 parent_obs_id = payload.get("parent_obs_id")
2928 stack_token = (
2929 _PARENT_STACK.set((parent_obs_id,))
2930 if tracer is not None and parent_obs_id is not None
2931 else None
2932 )
2933 child_obs = (
2934 tracer.open_parent_observation(
2935 name=f"spawn:{child_id}",
2936 input={"node_id": child_id, "role": child_role, "depth": depth},
2937 )
2938 if tracer is not None
2939 else None
2940 )
2941 try:
2942 result_state = run_node(
2943 role=child_role,
2944 node_step=node_step,
2945 input_override=node_step["input"],
2946 node_id=child_id,
2947 depth=depth,
2948 parent_id=parent_id,
2949 )
2950 finally:
2951 if tracer is not None and child_obs is not None:
2952 tracer.close_parent_observation(child_obs)
2953 if stack_token is not None:
2954 _PARENT_STACK.reset(stack_token)
2955 this_result = result_state.get(child_id)
2956 children = child_plan(child_id, this_result, depth)
2957 record, plan = build_record(
2958 node_id=child_id,
2959 path=path,
2960 parent_path=path[:-1],
2961 depth=depth,
2962 this_result=this_result,
2963 result_state=result_state,
2964 children=children,
2965 )
2966 update = {
2967 "accum": {step_id: {path: record}},
2968 "outstanding": len(plan) - 1,
2969 }
2970 if plan:
2971 child_obs_id = getattr(child_obs, "id", None)
2972 return Command(
2973 update=update,
2974 goto=[
2975 Send(
2976 "worker",
2977 {**payload, "parent_obs_id": child_obs_id},
2978 )
2979 for payload in plan
2980 ],
2981 )
2982 return Command(update=update, goto="check")
2984 def check_node(state: dict) -> Command:
2985 if state.get("outstanding") == 0:
2986 return Command(goto="join")
2987 return Command(goto=END)
2989 def join_node(state: dict) -> dict:
2990 records = (state.get("accum") or {}).get(step_id) or {}
2991 envelope, metrics, tool_meta = _assemble_spawn_outcome(records)
2992 return {
2993 "spawn_outcome": {
2994 "envelope": envelope,
2995 "metrics": metrics,
2996 "tool_meta": tool_meta,
2997 },
2998 "accum": {step_id: None},
2999 }
3001 schema = TypedDict(
3002 "_SpawnState",
3003 {
3004 "accum": Annotated[dict, accumulate_map_reduce],
3005 "outstanding": Annotated[int, operator.add],
3006 "spawn_outcome": dict,
3007 },
3008 )
3009 graph = StateGraph(schema)
3010 graph.add_node("director", director_node)
3011 graph.add_node("worker", worker_node)
3012 graph.add_node("check", check_node)
3013 graph.add_node("join", join_node)
3014 graph.add_edge(START, "director")
3015 graph.add_edge("join", END)
3016 app = graph.compile()
3017 out = app.invoke(
3018 {"accum": {}, "outstanding": 1},
3019 config={"recursion_limit": _SPAWN_RECURSION_LIMIT},
3020 )
3021 return out["spawn_outcome"]
3022 finally:
3023 if tracer is not None and parent_obs is not None:
3024 tracer.close_parent_observation(parent_obs)
3026 def _child_node_step(
3027 self,
3028 *,
3029 node_id: str,
3030 child: Any,
3031 parent_id: str,
3032 index: int,
3033 base_state: PipelineState,
3034 base_ctx: dict,
3035 ) -> dict:
3036 """Resolve one child plan item into the worker's runtime step.
3038 A child is a runtime-generated step: it carries only its id and its
3039 resolved input, never the spawn step's declared channel bindings, so a
3040 child's output can never route a channel write.
3041 """
3042 if not self._spawn_has_input(child):
3043 raise RunHardFailure(
3044 "INVALID_CHILD_PLAN",
3045 f"spawn child #{index} of node '{parent_id}' must be a dict "
3046 "with a non-empty 'input'",
3047 {"node_id": parent_id, "child_index": index, "child": repr(child)[:200]},
3048 )
3049 return {"id": node_id, "input": _resolve_input(child["input"], base_state, base_ctx)}
3051 def _stage_spawn_channel_writes(
3052 self,
3053 step: dict,
3054 result: Any,
3055 working_set: ChannelWorkingSet | None,
3056 ) -> None:
3057 """Stage a spawn node's declared writes from its own agent result."""
3058 if working_set is None:
3059 return
3060 writes = _declared_output_channel_writes(step, result, self._channel_names)
3061 for name, value in writes.items():
3062 working_set.stage(name, value)
3064 @staticmethod
3065 def _spawn_children_from_result(result: Any) -> list:
3066 """The node's child plan at ``content.children``; [] for a valid leaf.
3068 The director's parsed output envelope carries the plan at
3069 ``content.children``; a result that is not a dict, or whose ``content``
3070 is not an object, or carries no ``children``, is a valid leaf. A
3071 ``children`` value that is not a list is a hard ``INVALID_CHILD_PLAN``.
3072 """
3073 content = _envelope_content(result)
3074 if not isinstance(content, dict):
3075 return []
3076 raw = content.get("children")
3077 if raw is None:
3078 return []
3079 if not isinstance(raw, list):
3080 raise RunHardFailure(
3081 "INVALID_CHILD_PLAN",
3082 "spawn agent content 'children' must be a list",
3083 {"children": repr(raw)[:200]},
3084 )
3085 return raw
3087 @staticmethod
3088 def _spawn_has_input(child: Any) -> bool:
3089 """A valid child plan item: a dict carrying a non-empty 'input'."""
3090 if not isinstance(child, dict):
3091 return False
3092 input_val = child.get("input")
3093 if input_val is None:
3094 return False
3095 if isinstance(input_val, str) and not input_val.strip():
3096 return False
3097 if isinstance(input_val, (list, dict)) and not input_val:
3098 return False
3099 return True
3101 @staticmethod
3102 def _spawn_is_soft_fail(result: Any) -> bool:
3103 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error.
3105 An envelope-shaped result is never soft: ``status:error`` is a hard
3106 failure (the gate raises before this), and ``ok``/``partial``/
3107 ``no_success`` are valid non-fatal results.
3108 """
3109 if not isinstance(result, dict):
3110 return False
3111 if result.get("status") in ENVELOPE_STATUSES:
3112 return False
3113 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
3114 return True
3115 return any(key.startswith("budget:") for key in result)
3118# ---------------------------------------------------------------------------
3119# GraphBuilder
3120# ---------------------------------------------------------------------------
3122def _topological_order(steps: list[dict]) -> list[dict]:
3123 """
3124 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
3125 """
3126 id_to_step = {s["id"]: s for s in steps}
3127 visited: set[str] = set()
3128 order: list[dict] = []
3130 def visit(step_id: str) -> None:
3131 if step_id in visited:
3132 return
3133 visited.add(step_id)
3134 for dep in (id_to_step[step_id].get("depends_on") or []):
3135 if dep in id_to_step:
3136 visit(dep)
3137 order.append(id_to_step[step_id])
3139 for s in steps:
3140 visit(s["id"])
3141 return order
3144def _group_dependencies(
3145 groups: list[list[dict]],
3146) -> tuple[list[set[int]], list[set[int]]]:
3147 """Group-index adjacency (predecessors, successors) from each step's depends_on."""
3148 step_to_group = {
3149 step["id"]: index
3150 for index, group in enumerate(groups)
3151 for step in group
3152 }
3153 predecessors: list[set[int]] = [set() for _ in groups]
3154 successors: list[set[int]] = [set() for _ in groups]
3155 for index, group in enumerate(groups):
3156 for step in group:
3157 for dep in step.get("depends_on") or []:
3158 dep_index = step_to_group.get(dep)
3159 if dep_index is None or dep_index == index:
3160 continue
3161 predecessors[index].add(dep_index)
3162 successors[dep_index].add(index)
3163 return predecessors, successors
3166@dataclass(frozen=True)
3167class _RoutePlan:
3168 """Compiled `routes` topology: route tables plus the static-edge suppression sets.
3170 `routes` owns a routing step's outgoing control flow (ADR-0025). A route
3171 target reached via a static `depends_on` edge is therefore suppressed and
3172 replaced by the native conditional edge; the plan records both the targets
3173 (so they are never made entry points) and the suppressed `(source, target)`
3174 pairs.
3175 """
3177 step_to_node: dict[str, str]
3178 routes_by_step: dict[str, list[dict]]
3179 target_node_ids: frozenset[str]
3180 routing_node_ids: frozenset[str]
3181 suppressed_edges: frozenset[tuple[str, str]]
3184def _reject_nested_routes(step: dict) -> None:
3185 """Reject `routes` on a nested `body` step (top-level-only rule, ADR-0012).
3187 The engine wires `routes` only for top-level steps, so a nested declaration
3188 would be silently ignored; nested routing is deferred.
3189 """
3190 body = step.get("body")
3191 if not isinstance(body, list):
3192 return
3193 for nested in _iter_steps(body):
3194 if nested.get("routes"):
3195 raise RunHardFailure(
3196 "VALIDATION_ERROR",
3197 f"Step '{nested.get('id')}' declares `routes` inside a nested "
3198 "body; `routes` is only allowed on top-level steps",
3199 )
3202def _build_route_plan(groups: list[list[dict]], node_ids: list[str]) -> _RoutePlan:
3203 """Resolve every top-level step's `routes` into a node-level routing plan."""
3204 step_to_node = {
3205 step["id"]: node_ids[index]
3206 for index, group in enumerate(groups)
3207 for step in group
3208 }
3209 step_to_deps = {
3210 step["id"]: set(step.get("depends_on") or [])
3211 for group in groups
3212 for step in group
3213 }
3214 routes_by_step: dict[str, list[dict]] = {}
3215 target_node_ids: set[str] = set()
3216 routing_node_ids: set[str] = set()
3217 suppressed_edges: set[tuple[str, str]] = set()
3218 for group in groups:
3219 for step in group:
3220 _reject_nested_routes(step)
3221 routes = step.get("routes")
3222 if not routes:
3223 continue
3224 source = step["id"]
3225 source_node = step_to_node[source]
3226 routing_node_ids.add(source_node)
3227 routes_by_step[source] = routes
3228 for route in routes:
3229 target = route.get("target")
3230 target_node = step_to_node.get(target)
3231 if target_node is None:
3232 raise RunHardFailure(
3233 "VALIDATION_ERROR",
3234 f"Step '{source}' route target '{target}' is not a "
3235 "top-level step in this pipeline",
3236 )
3237 target_node_ids.add(target_node)
3238 if source in step_to_deps.get(target, set()):
3239 suppressed_edges.add((source_node, target_node))
3240 log.warning(
3241 "[routes] suppressing static edge %s -> %s; route "
3242 "targets are wired via `routes`",
3243 source,
3244 target,
3245 )
3246 return _RoutePlan(
3247 step_to_node=step_to_node,
3248 routes_by_step=routes_by_step,
3249 target_node_ids=frozenset(target_node_ids),
3250 routing_node_ids=frozenset(routing_node_ids),
3251 suppressed_edges=frozenset(suppressed_edges),
3252 )
3255def _make_route_router(routes: list[dict], step_to_node: dict[str, str], end) -> Any:
3256 """First-match-wins router over a routing step's declared routes.
3258 Evaluates `when` in declaration order and returns the selected target node;
3259 a route without `when` is the default branch. With no match and no default
3260 it returns the graph terminal.
3261 """
3262 def router(state: PipelineState) -> str:
3263 ctx = _flatten_channel_state(state) if _CORE in state else state
3264 for route in routes:
3265 if _eval_condition(route.get("when"), ctx):
3266 return step_to_node[route["target"]]
3267 return end
3269 return router
3272def _wire_route_edges(graph, route_plan: _RoutePlan, end) -> None:
3273 """Add one native conditional edge per routing step, with an explicit path_map."""
3274 for source, routes in route_plan.routes_by_step.items():
3275 target_nodes = [route_plan.step_to_node[r["target"]] for r in routes]
3276 path_map = {node: node for node in target_nodes}
3277 path_map[end] = end
3278 graph.add_conditional_edges(
3279 route_plan.step_to_node[source],
3280 _make_route_router(routes, route_plan.step_to_node, end),
3281 path_map,
3282 )
3285def _wire_dependency_edges(
3286 graph,
3287 groups: list[list[dict]],
3288 node_ids: list[str],
3289 end,
3290 route_plan: _RoutePlan,
3291 entry_ids: list[str] | None = None,
3292 exit_ids: list[str] | None = None,
3293) -> None:
3294 """Wire native branches and fan-in joins from group-level depends_on edges.
3296 Static edges originating from a routing step to one of its route targets are
3297 suppressed — `routes` supplies that control flow. A route target with no
3298 other predecessor is reached by its conditional edge, never an entry point.
3299 A group split into a Send fan-out enters through its dispatch node and exits
3300 through its join node, so ``entry_ids``/``exit_ids`` name those boundaries.
3301 """
3302 entry_ids = entry_ids if entry_ids is not None else node_ids
3303 exit_ids = exit_ids if exit_ids is not None else node_ids
3304 predecessors, successors = _group_dependencies(groups)
3305 for index, node_id in enumerate(node_ids):
3306 entry = entry_ids[index]
3307 pred_nodes = [exit_ids[p] for p in sorted(predecessors[index])]
3308 if not pred_nodes:
3309 if node_id not in route_plan.target_node_ids:
3310 graph.set_entry_point(entry)
3311 else:
3312 live_preds = [
3313 pred
3314 for pred in pred_nodes
3315 if (pred, entry) not in route_plan.suppressed_edges
3316 ]
3317 if len(live_preds) == 1:
3318 graph.add_edge(live_preds[0], entry)
3319 elif live_preds:
3320 graph.add_edge(live_preds, entry)
3321 if not successors[index] and node_id not in route_plan.routing_node_ids:
3322 graph.add_edge(exit_ids[index], end)
3325# ---------------------------------------------------------------------------
3326# Call-step build-time expansion (ADR-0037 D3/D4)
3327# ---------------------------------------------------------------------------
3328#
3329# A `type: call` step is composed at build time: the function's body steps plus
3330# two synthesized op steps (`__call_bind` / `__call_emit`) are assembled into a
3331# compiled subgraph added as a single parent-graph node keyed by the call id. The
3332# runtime executes only native steps; LangGraph never sees a call step. The caller
3333# reads the function outputs via the call result shape
3334# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
3335# The function body's `{{ params.<name> }}` references are rewritten to
3336# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves. A
3337# call nested in a foreach/while body is flat-inlined instead, because loop bodies
3338# execute imperatively and cannot host a compiled subgraph node.
3339#
3340# Isolation (ADR-0037 D5 / 00-functions § Isolation) and recursion (ADR-0037 D7): a
3341# function body step must not reference a step-id outside its own body (a
3342# one-way dependency check — outward references to caller step-ids are build
3343# errors); a cycle in the function call graph (self or mutual) is a build
3344# error. Bounded while loops are runtime constructs and are not cycle-flagged.
3346# Call step result namespace keys (internal scaffolding).
3347def _scope_params_value(value: Any, call_id: str) -> Any:
3348 """Rewrite `params.` template references to the call-scoped namespace.
3350 Operates on the function body's string values (templates). Dict keys are
3351 untouched — only templated string VALUES are scoped, so a function body's
3352 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
3353 """
3354 if isinstance(value, str):
3355 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
3356 if isinstance(value, dict):
3357 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
3358 if isinstance(value, list):
3359 return [_scope_params_value(v, call_id) for v in value]
3360 return value
3363def _scope_params(step: dict, call_id: str) -> dict:
3364 """Deep-scope a function body step's params references for a call id."""
3365 scoped: dict = dict(step)
3366 for field in ("input", "params", "condition", "exit_condition", "agent", "function"):
3367 if field in scoped and isinstance(scoped[field], (str, dict, list)):
3368 scoped[field] = _scope_params_value(scoped[field], call_id)
3369 if "body" in scoped and isinstance(scoped["body"], list):
3370 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
3371 return scoped
3374def _collect_step_ids(steps: list[dict]) -> set[str]:
3375 """All step ids declared in *steps*, including nested body step ids."""
3376 ids: set[str] = set()
3377 for step in steps:
3378 step_id = step.get("id")
3379 if isinstance(step_id, str):
3380 ids.add(step_id)
3381 body = step.get("body")
3382 if isinstance(body, list):
3383 ids |= _collect_step_ids(body)
3384 return ids
3387def _iter_steps(steps: list[dict]):
3388 """Yield every step in *steps*, recursing into nested bodies."""
3389 for step in steps:
3390 yield step
3391 body = step.get("body")
3392 if isinstance(body, list):
3393 yield from _iter_steps(body)
3396def _collect_composed_step_ids(steps: list[dict]) -> set[str]:
3397 """All declared step ids, including ids composed into native subgraphs.
3399 A ``call`` step becomes a ``__subgraph__`` placeholder whose inner step ids
3400 live only inside the compiled subgraph. The placeholder records them under
3401 ``_SUBGRAPH_STEP_IDS_KEY`` so name-collision validation sees the full set
3402 even after body steps stop being textually inlined.
3403 """
3404 ids = _collect_step_ids(steps)
3405 for step in _iter_steps(steps):
3406 composed_ids = step.get(_SUBGRAPH_STEP_IDS_KEY)
3407 if isinstance(composed_ids, (set, frozenset)):
3408 ids |= composed_ids
3409 return ids
3412_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
3413_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
3414_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
3416_REFERENCE_FIELDS = ("input", "params", "condition", "exit_condition")
3418# Engine-reserved template namespaces — never step-id references, even if a
3419# caller happens to declare a step with one of these names.
3420_RESERVED_NAMESPACES = frozenset({"params", "loop", "item", "_item", "budget", "config"})
3423def _referenced_names(value: Any) -> set[str]:
3424 """Root identifiers read by templated strings.
3426 Only the leading identifier of each reference (before the first ``.`` or
3427 ``[``) can name a step-id; nested field segments are ignored. Quoted
3428 literals are ignored.
3429 """
3430 names: set[str] = set()
3431 if isinstance(value, str):
3432 for expr in _TEMPLATE_REF_RE.findall(value):
3433 bare = _STRING_LITERAL_RE.sub("", expr)
3434 for match in _IDENTIFIER_RE.finditer(bare):
3435 if match.start() == 0 or bare[match.start() - 1] != ".":
3436 names.add(match.group())
3437 elif isinstance(value, dict):
3438 for v in value.values():
3439 names |= _referenced_names(v)
3440 elif isinstance(value, list):
3441 for v in value:
3442 names |= _referenced_names(v)
3443 return names
3446def _validate_function_purity(
3447 fn_name: str, body: list[dict], visible_ids: set[str]
3448) -> None:
3449 """One-way isolation: a function body must not reference a caller step-id.
3451 A body step may read its own body step-ids and the params namespace; a
3452 reference to a step-id that exists outside the body (``visible_ids`` but
3453 not declared in the body) is an outward dependency and a hard build error.
3454 """
3455 local_ids = _collect_step_ids(body)
3456 for step in _iter_steps(body):
3457 step_id = step.get("id", "?")
3458 for dep in step.get("depends_on") or []:
3459 if dep in visible_ids and dep not in local_ids:
3460 raise RunHardFailure(
3461 "VALIDATION_ERROR",
3462 f"function '{fn_name}' step '{step_id}': depends_on "
3463 f"'{dep}' is outside the function body (isolation — "
3464 "function bodies see only their own steps and params)",
3465 )
3466 for field in _REFERENCE_FIELDS:
3467 for ref in _referenced_names(step.get(field)):
3468 if ref in _RESERVED_NAMESPACES:
3469 continue
3470 if ref in visible_ids and ref not in local_ids:
3471 raise RunHardFailure(
3472 "VALIDATION_ERROR",
3473 f"function '{fn_name}' step '{step_id}': '{field}' "
3474 f"references '{ref}', a step-id outside the function "
3475 "body (isolation — function bodies see only their own "
3476 "steps and params)",
3477 )
3480def _reject_function_body_routes(fn_name: str, body: list[dict]) -> None:
3481 """Reject `routes` anywhere in a function body (top-level-only rule).
3483 A function body is inlined to the top level, so a body step declaring
3484 `routes` would escape the nested-`body` guard and be silently rewired;
3485 reject it at build time instead.
3486 """
3487 for step in _iter_steps(body):
3488 if step.get("routes"):
3489 raise RunHardFailure(
3490 "VALIDATION_ERROR",
3491 f"function '{fn_name}' step '{step.get('id')}' declares `routes`; "
3492 "`routes` is only allowed on top-level steps",
3493 )
3496def _rewrite_call_deps(steps: list[dict], call_map: dict[str, str]) -> None:
3497 """Rewrite `depends_on` references to call ids → the corresponding emit id.
3499 After a call is expanded away, a downstream step that depends_on the call id
3500 must instead wait on the emit step (which writes state[<call_id>]). This
3501 guarantees the caller's `{{ <call_id>.outputs.<name> }}` reads resolve after
3502 the function body ran.
3503 """
3504 for step in steps:
3505 deps = step.get("depends_on") or []
3506 if deps:
3507 step["depends_on"] = [call_map.get(d, d) for d in deps]
3508 if "body" in step and isinstance(step["body"], list):
3509 _rewrite_call_deps(step["body"], call_map)
3512def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
3513 """Expand every ``call`` step through the native subgraph composer.
3515 Test-facing seam over the single call-expansion implementation: the same
3516 ``_CallComposer`` that ``GraphBuilder`` uses builds the steps, so no second
3517 (flat) expansion path exists. The embedded subgraph value is the un-compiled
3518 sub-step list; a runnable graph only needs the compiled form, which
3519 production obtains through ``GraphBuilder._compose_calls``.
3520 """
3521 composer = _CallComposer(
3522 registry,
3523 assemble_subgraph=lambda sub_steps: sub_steps,
3524 channels=[],
3525 channel_names=frozenset(),
3526 )
3527 return composer.compose(steps)
3530_SUBGRAPH_STEP_TYPE = "__subgraph__"
3531_SUBGRAPH_GRAPH_KEY = "__subgraph_graph__"
3532_SUBGRAPH_STEP_IDS_KEY = "__subgraph_step_ids__"
3535class _CallComposer:
3536 """Compose `call` steps into compiled native subgraph nodes (ADR-0037 D4).
3538 A function body becomes its own compiled StateGraph added to the parent as a
3539 single node keyed by the call id; the ``bind``/``emit`` scaffolding and the
3540 body steps live inside that subgraph and share the parent state. Nested
3541 function calls compose recursively. A `call` nested in a foreach/while body
3542 is composed the same way; the loop body executes the embedded compiled graph
3543 imperatively via ``run_step``'s ``__subgraph__`` branch.
3544 """
3546 def __init__(
3547 self,
3548 registry: ManifestRegistry,
3549 assemble_subgraph: Any,
3550 channels: list[dict],
3551 channel_names: frozenset[str],
3552 ) -> None:
3553 self._registry = registry
3554 self._assemble_subgraph = assemble_subgraph
3555 self._channels = channels
3556 self._channel_names = channel_names
3557 self._stack: list[str] = []
3558 self._expanded_functions: dict[str, str] = {}
3559 self._call_map: dict[str, str] = {}
3561 def compose(self, steps: list[dict]) -> list[dict]:
3562 visible_ids = _collect_step_ids(steps)
3563 composed: list[dict] = []
3564 for step in steps:
3565 composed.extend(self._compose_step(step, visible_ids=visible_ids))
3566 return composed
3568 def _compose_step(self, step: dict, *, visible_ids: set[str]) -> list[dict]:
3569 if step.get("type") == "call":
3570 return [self._compose_subgraph_call(step, visible_ids)]
3571 body = step.get("body")
3572 if not isinstance(body, list):
3573 return [step]
3574 nested: list[dict] = []
3575 for sub in body:
3576 nested.extend(self._compose_step(sub, visible_ids=visible_ids))
3577 _rewrite_call_deps(nested, self._call_map)
3578 return [{**step, "body": nested}]
3580 def _compose_subgraph_call(self, step: dict, visible_ids: set[str]) -> dict:
3581 call_id = step["id"]
3582 function, body = self._check_call_guards(step, visible_ids)
3584 nested_visible = visible_ids | _collect_step_ids(body)
3585 outputs_spec = (
3586 ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
3587 )
3588 bind_id = f"_call__{call_id}__bind"
3589 emit_id = f"_call__{call_id}__emit"
3590 bind_step: dict = {
3591 "id": bind_id,
3592 "type": "builtin",
3593 "component": "__call_bind",
3594 "params": {**(step.get("params") or {}), "__call_id": call_id},
3595 "depends_on": [],
3596 }
3598 self._stack.append(step["function"])
3599 try:
3600 expanded_body = self._compose_body(
3601 body, call_id, bind_id, nested_visible
3602 )
3603 finally:
3604 self._stack.pop()
3606 emit_step: dict = {
3607 "id": emit_id,
3608 "type": "builtin",
3609 "component": "__call_emit",
3610 "params": {"__call_id": call_id, "__outputs": outputs_spec},
3611 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
3612 }
3613 sub_steps = [bind_step, *expanded_body, emit_step]
3614 if self._channels:
3615 _validate_channel_inputs(self._channels, sub_steps)
3616 _validate_step_channel_bindings(sub_steps, self._channel_names)
3617 placeholder: dict = {
3618 "id": call_id,
3619 "type": _SUBGRAPH_STEP_TYPE,
3620 _SUBGRAPH_GRAPH_KEY: self._assemble_subgraph(
3621 _topological_order(sub_steps)
3622 ),
3623 _SUBGRAPH_STEP_IDS_KEY: _collect_composed_step_ids(sub_steps),
3624 "depends_on": list(step.get("depends_on") or []),
3625 }
3626 call_condition = step.get("condition")
3627 if call_condition is not None:
3628 placeholder["condition"] = call_condition
3629 return placeholder
3631 def _compose_body(
3632 self, body: list[dict], call_id: str, bind_id: str, visible_ids: set[str]
3633 ) -> list[dict]:
3634 expanded: list[dict] = []
3635 for body_step in body:
3636 scoped = _scope_params(body_step, call_id)
3637 for sub in self._compose_step(scoped, visible_ids=visible_ids):
3638 deps = list(sub.get("depends_on") or [])
3639 if bind_id in deps:
3640 expanded.append(sub)
3641 else:
3642 expanded.append({**sub, "depends_on": [bind_id, *deps]})
3643 return expanded
3645 def _check_call_guards(
3646 self, step: dict, visible_ids: set[str]
3647 ) -> tuple[dict, list[dict]]:
3648 """Single owner of every build-time ``call`` guard (ADR-0037 D3/D5/D7).
3650 Rejects routes on the call step, function cycles, duplicate expansion,
3651 unknown functions, function-body isolation (purity) violations, and
3652 body-declared routes; returns the resolved function and its body.
3653 """
3654 fn_name = step["function"]
3655 call_id = step["id"]
3657 if step.get("routes"):
3658 raise RunHardFailure(
3659 "VALIDATION_ERROR",
3660 f"call step '{call_id}' declares `routes`; `routes` is not supported "
3661 "on call steps",
3662 )
3663 if fn_name in self._stack:
3664 chain = " -> ".join(self._stack + [fn_name])
3665 raise RunHardFailure(
3666 "VALIDATION_ERROR",
3667 f"cycle detected in function expansion: {chain} "
3668 "(ADR-0037 D7 — recursion must be a bounded while loop or "
3669 "idempotent re-run)",
3670 )
3671 if fn_name in self._expanded_functions:
3672 raise RunHardFailure(
3673 "VALIDATION_ERROR",
3674 f"duplicate call expansion: function '{fn_name}' called at both "
3675 f"'{self._expanded_functions[fn_name]}' and '{call_id}' in one "
3676 "pipeline (ADR-0037 D7 — repeated calls to the same function are "
3677 "not yet supported)",
3678 )
3679 self._expanded_functions[fn_name] = call_id
3681 function = self._registry.get("Function", fn_name)
3682 if function is None:
3683 raise RunHardFailure(
3684 "VALIDATION_ERROR",
3685 f"call step '{call_id}' references unknown function '{fn_name}' "
3686 "(no kind: Function manifest in the workflows tree)",
3687 )
3688 body = (function.get("spec") or {}).get("steps") or []
3689 _validate_function_purity(fn_name, body, visible_ids)
3690 _reject_function_body_routes(fn_name, body)
3691 return function, body
3694_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
3697def _validate_channel_entry(index: int, channel: Any) -> None:
3698 """Validate one spec.channels entry against the frozen ChannelSpec shape.
3700 The engine build path does not run manifest-schema validation, so a
3701 malformed entry is rejected here — before any ``["name"]`` indexing — with
3702 an error that identifies the offending entry.
3703 """
3704 if not isinstance(channel, dict):
3705 raise RunHardFailure(
3706 "CHANNEL_BINDING_ERROR",
3707 f"Pipeline spec.channels[{index}] must be a mapping; got "
3708 f"{type(channel).__name__}",
3709 )
3710 name = channel.get("name")
3711 if not isinstance(name, str) or not name:
3712 raise RunHardFailure(
3713 "CHANNEL_BINDING_ERROR",
3714 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3715 "channel requires a non-empty string 'name'",
3716 )
3717 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3718 raise RunHardFailure(
3719 "CHANNEL_BINDING_ERROR",
3720 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3721 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}",
3722 )
3725def _validated_channels(raw_channels: Any) -> list[dict]:
3726 """Return spec.channels as a validated list, or [] when absent.
3728 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3729 """
3730 if raw_channels is None:
3731 return []
3732 if not isinstance(raw_channels, list):
3733 raise RunHardFailure(
3734 "CHANNEL_BINDING_ERROR",
3735 f"Pipeline spec.channels must be a list; got "
3736 f"{type(raw_channels).__name__}",
3737 )
3738 for index, channel in enumerate(raw_channels):
3739 _validate_channel_entry(index, channel)
3740 return raw_channels
3743def _validate_step_channel_bindings(
3744 steps: list[dict], channel_names: frozenset[str]
3745) -> None:
3746 """Reject malformed or undeclared step-level channel bindings at build time.
3748 Every step's declared ``output_channel`` must be a non-empty declared
3749 channel name, and every declared ``input_channels`` entry must be a
3750 non-empty declared channel name. This runs unconditionally, so a binding
3751 to a channel absent from ``spec.channels`` fails the build even when no
3752 channels are declared.
3753 """
3754 for step in _iter_steps(steps):
3755 _validate_step_output_channel(step, channel_names)
3756 _validate_step_input_channels(step, channel_names)
3759def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3760 channel = step.get("output_channel")
3761 if channel is None:
3762 return
3763 if not isinstance(channel, str) or not channel:
3764 raise RunHardFailure(
3765 "CHANNEL_BINDING_ERROR",
3766 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3767 f"string; got {channel!r}",
3768 )
3769 if channel not in channel_names:
3770 raise RunHardFailure(
3771 "CHANNEL_BINDING_ERROR",
3772 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3773 "which is not declared in spec.channels",
3774 )
3777def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3778 channels = step.get("input_channels")
3779 if channels is None:
3780 return
3781 if not isinstance(channels, list):
3782 raise RunHardFailure(
3783 "CHANNEL_BINDING_ERROR",
3784 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3785 f"channel names; got {type(channels).__name__}",
3786 )
3787 for channel in channels:
3788 if not isinstance(channel, str) or not channel:
3789 raise RunHardFailure(
3790 "CHANNEL_BINDING_ERROR",
3791 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3792 f"channel name '{channel}'; channel names must be non-empty "
3793 "strings",
3794 )
3795 if channel not in channel_names:
3796 raise RunHardFailure(
3797 "CHANNEL_BINDING_ERROR",
3798 f"Step '{step.get('id', '?')}' reads undeclared channel "
3799 f"'{channel}'; the channel must be declared in spec.channels",
3800 )
3803def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3804 """Reject duplicate and engine/step-reserved channel names."""
3805 forbidden = (
3806 {_CORE}
3807 | set(_RESERVED_NAMESPACES)
3808 | _CORE_STATE_KEYS
3809 | _collect_composed_step_ids(steps)
3810 )
3811 seen: set[str] = set()
3812 for channel in channels:
3813 name = channel.get("name")
3814 if name in seen:
3815 raise RunHardFailure(
3816 "CHANNEL_BINDING_ERROR",
3817 f"Pipeline declares channel '{name}' more than once; channel "
3818 "names must be unique",
3819 )
3820 seen.add(name)
3821 if name in forbidden:
3822 raise RunHardFailure(
3823 "CHANNEL_BINDING_ERROR",
3824 f"Pipeline channel '{name}' collides with an engine-reserved "
3825 "name (a step id, engine state key, reserved namespace, or the "
3826 "core state channel); channel names must be unique and "
3827 "non-reserved",
3828 )
3831def _input_path_root(value: str) -> str | None:
3832 """Leading state key of a bare dot-path input string, or None.
3834 Template strings, bracket-prefixed verbatim literals, and blank strings
3835 are not state reads and yield None.
3836 """
3837 if "{{" in value:
3838 return None
3839 stripped = value.lstrip()
3840 if not stripped or stripped.startswith("["):
3841 return None
3842 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3845def _iter_input_roots(input_spec: Any):
3846 """Yield every bare dot-path root in an input spec, recursing into containers."""
3847 if isinstance(input_spec, str):
3848 root = _input_path_root(input_spec)
3849 if root:
3850 yield root
3851 elif isinstance(input_spec, dict):
3852 for value in input_spec.values():
3853 yield from _iter_input_roots(value)
3854 elif isinstance(input_spec, list):
3855 for value in input_spec:
3856 yield from _iter_input_roots(value)
3859def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3860 """Reject channel-pipeline input roots that name nothing the state exposes.
3862 The channel state is the declared channels plus the engine state, so a bare
3863 dot-path input root must be a declared channel, a known step id, a reserved
3864 namespace, or an engine state key. Anything else is a hard build error —
3865 there is no silent fallback to a missing key.
3866 """
3867 allowed = (
3868 {ch["name"] for ch in channels}
3869 | _collect_step_ids(steps)
3870 | set(_RESERVED_NAMESPACES)
3871 | _CORE_STATE_KEYS
3872 )
3873 for step in _iter_steps(steps):
3874 for root in _iter_input_roots(step.get("input")):
3875 if root not in allowed:
3876 raise RunHardFailure(
3877 "CHANNEL_BINDING_ERROR",
3878 f"Pipeline channel step '{step.get('id', '?')}' input "
3879 f"references unknown root '{root}'; in a channel pipeline "
3880 "an input root must be a declared channel, a step id, a "
3881 "reserved namespace, or an engine state key",
3882 )
3885def _flatten_channel_state(state: dict) -> dict:
3886 """Read the flat committed snapshot nested under the Core namespace."""
3887 return dict(state.get(_CORE) or {})
3890def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3891 """Drop channel-named keys so a node emits only the core state."""
3892 return {k: v for k, v in flat.items() if k not in channel_names}
3895_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3897_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3900def _envelope_content(result: Any) -> Any:
3901 """Return an envelope's optional ``content`` field (ADR-0023 D3).
3903 ``content`` is opaque and optional: a non-mapping result, an absent field,
3904 and an explicit ``null`` all read as ``None``. Callers apply their own
3905 fallback and shape checks (e.g. ``[]`` for a spawn child plan).
3906 """
3907 if isinstance(result, dict):
3908 return result.get("content")
3909 return None
3912def _is_failure_envelope(value: Any) -> bool:
3913 """True when a step output is not a successfully parsed dict.
3915 Missing output, non-dict values, and the ``error``/``parse_error``
3916 envelopes written by the LLM failure paths are not routed. A valid
3917 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3918 it carries an explicit ``"error": null``.
3919 """
3920 if not isinstance(value, dict):
3921 return True
3922 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3923 return False
3924 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3927def _declared_output_channel_writes(
3928 step: dict, output: Any, channel_names: frozenset[str]
3929) -> dict[str, Any]:
3930 """Route one step output's content to its declared output channel.
3932 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3933 the working set. Failure envelopes and steps without an ``output_channel``
3934 are not routed; an undeclared target channel is a hard error.
3935 """
3936 if _is_failure_envelope(output):
3937 return {}
3938 channel = step.get("output_channel")
3939 if not channel:
3940 return {}
3941 if channel not in channel_names:
3942 raise RunHardFailure(
3943 "CHANNEL_BINDING_ERROR",
3944 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3945 "which is not declared in spec.channels",
3946 )
3947 return {channel: _envelope_content(output)}
3950def _route_declared_channels(
3951 step: dict,
3952 pre_state: dict,
3953 core: dict,
3954 step_runner: Any,
3955 channel_names: frozenset[str],
3956) -> dict[str, Any]:
3957 """Route a step's declared output channel write to its channel.
3959 Returns ``{channel_name: value}`` for LangGraph to merge through the
3960 channel's reducer, where ``channel_name`` is the step's single declared
3961 ``output_channel`` and ``value`` is the agent envelope's ``content``. Only
3962 an actual agent invocation is routed: a ``skip_llm_if`` passthrough and a
3963 parallel fan-out envelope are excluded. A step without an ``output_channel``
3964 routes nothing (its output stays in ``core`` untouched), and failure
3965 envelopes are not routed. A spawn step is routed at the spawn seam from its
3966 node result, never here from its envelope.
3967 """
3968 step_id = step.get("id", "?")
3969 if step.get("type") == "spawn":
3970 return {}
3971 if step_id not in core:
3972 return {}
3973 agent_spec = step_runner.resolve_agent(step)
3974 if agent_spec is None:
3975 return {}
3976 if not step_runner.ran_agent(step, pre_state):
3977 return {}
3978 return _declared_output_channel_writes(step, core[step_id], channel_names)
3981def _make_step_node(
3982 step: dict,
3983 step_runner: Any,
3984 channel_names: frozenset[str],
3985 channel_types: dict[str, str],
3986) -> Any:
3987 if not channel_names:
3988 def node_fn(state: PipelineState) -> PipelineState:
3989 return step_runner.run_step(step, state)
3990 return node_fn
3992 def node_fn(state: PipelineState) -> PipelineState:
3993 committed = _flatten_channel_state(state)
3994 working_set = ChannelWorkingSet(committed, channel_types)
3995 core = _strip_channel_keys(
3996 step_runner.run_step(step, committed, working_set), channel_names
3997 )
3998 writes = _route_declared_channels(
3999 step, committed, core, step_runner, channel_names
4000 )
4001 for name, value in writes.items():
4002 working_set.stage(name, value)
4003 written_names = {name for name, _ in working_set.pending_writes()}
4004 channels = working_set.commit()
4005 emitted = {name: channels[name] for name in written_names}
4006 return {_CORE: {**core, **emitted}}
4007 return node_fn
4010def _make_subgraph_node(
4011 compiled: Any,
4012 condition: Any,
4013 step_runner: Any,
4014 channel_names: frozenset[str],
4015) -> Any:
4016 """Wrap a compiled call subgraph with the call step's entry condition.
4018 A false condition returns no state updates, so the subgraph never runs and
4019 ``state[<call_id>]`` is not written — matching the soft-skip semantics of a
4020 normal step condition. The ctx is derived exactly like ``run_step``'s,
4021 including channel-mode flattening. With no condition the compiled subgraph
4022 is the node, preserving the existing wiring.
4023 """
4024 if condition is None:
4025 return compiled
4027 if not channel_names:
4028 def node_fn(state: PipelineState) -> PipelineState:
4029 if not _eval_condition(condition, step_runner._make_ctx(state)):
4030 return {}
4031 return compiled.invoke(state)
4032 return node_fn
4034 def node_fn(state: PipelineState) -> PipelineState:
4035 committed = _flatten_channel_state(state)
4036 if not _eval_condition(condition, step_runner._make_ctx(committed)):
4037 return {}
4038 return compiled.invoke(state)
4039 return node_fn
4042def _collect_parallel_channel_writes(
4043 steps: list[dict],
4044 pre_state: dict,
4045 core: dict,
4046 step_runner: Any,
4047 channel_names: frozenset[str],
4048 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
4049) -> list[tuple[str, Any]]:
4050 """Collect one parallel node's declared writes in branch order.
4052 Every branch write is emitted, so fan-in feeds the channel-type reducer
4053 every write in deterministic declaration/branch order: Topic appends each
4054 write, LastValue keeps the last, and BinaryOperatorAggregate combines them
4055 per its accumulated shape (list concat, collection-map union, dict merge,
4056 scalar last). A branch's isolated body writes are emitted before its own
4057 declared write, mirroring the top-level node's route-after-run order.
4058 """
4059 writes: list[tuple[str, Any]] = []
4060 for step in steps:
4061 if branch_working_sets is not None:
4062 branch_working_set = branch_working_sets.get(step["id"])
4063 if branch_working_set is not None:
4064 writes.extend(branch_working_set.pending_writes())
4065 branch = _route_declared_channels(
4066 step, pre_state, core, step_runner, channel_names
4067 )
4068 for channel_name, value in branch.items():
4069 writes.append((channel_name, value))
4070 return writes
4073def _make_parallel_node(
4074 group_steps: list[dict],
4075 step_runner: Any,
4076 channel_names: frozenset[str],
4077 channel_types: dict[str, str],
4078) -> Any:
4079 def node_fn(state: PipelineState) -> PipelineState:
4080 base = (
4081 _flatten_channel_state(state)
4082 if channel_names else state
4083 )
4084 working_set = (
4085 ChannelWorkingSet(base, channel_types) if channel_names else None
4086 )
4087 # Each branch writes into its own fork; the shared set is advanced only
4088 # at fan-in, so branch-internal body writes never race.
4089 branch_working_sets = (
4090 {s["id"]: working_set.fork() for s in group_steps}
4091 if working_set is not None else None
4092 )
4094 def submit_branch(pool: Any, step: dict) -> Any:
4095 if branch_working_sets is None:
4096 return pool.submit(step_runner.run_step, step, base)
4097 return pool.submit(
4098 step_runner.run_step, step, base, branch_working_sets[step["id"]]
4099 )
4101 results: dict[str, Any] = {}
4102 branch_metrics: dict = {}
4103 branch_tool_meta: dict = {}
4104 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
4105 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
4106 for future in as_completed(futures):
4107 sid = futures[future]
4108 try:
4109 out = future.result()
4110 results[sid] = out.get(sid)
4111 branch_metrics.update(out.get("step_context_metrics") or {})
4112 branch_tool_meta.update(out.get("step_tool_meta") or {})
4113 except SkillLoadError:
4114 raise
4115 except RunHardFailure:
4116 pool.shutdown(wait=False, cancel_futures=True)
4117 raise
4118 except Exception as exc: # noqa: BLE001
4119 log.error("Step '%s' failed: %s", sid, exc)
4120 results[sid] = {"error": str(exc)}
4121 if channel_names:
4122 merged = _strip_channel_keys({**base, **results}, channel_names)
4123 if branch_metrics:
4124 merged["step_context_metrics"] = _merge_side_channel(
4125 merged.get("step_context_metrics"), branch_metrics
4126 )
4127 if branch_tool_meta:
4128 merged["step_tool_meta"] = _merge_side_channel(
4129 merged.get("step_tool_meta"), branch_tool_meta
4130 )
4131 writes = _collect_parallel_channel_writes(
4132 group_steps, base, merged, step_runner, channel_names,
4133 branch_working_sets,
4134 )
4135 written_names = {name for name, _ in writes}
4136 for name, value in writes:
4137 working_set.stage(name, value)
4138 channels = working_set.commit()
4139 emitted = {name: channels[name] for name in written_names}
4140 return {_CORE: {**merged, **emitted}}
4141 final: PipelineState = {**state, **results}
4142 if branch_metrics:
4143 final = {
4144 **final,
4145 "step_context_metrics": _merge_side_channel(
4146 state.get("step_context_metrics"), branch_metrics
4147 ),
4148 }
4149 if branch_tool_meta:
4150 final = {
4151 **final,
4152 "step_tool_meta": _merge_side_channel(
4153 state.get("step_tool_meta"), branch_tool_meta
4154 ),
4155 }
4156 return final
4157 return node_fn
4160@dataclass(frozen=True)
4161class _MapReduceParts:
4162 """The nodes and edges of one concurrent step's Send fan-out.
4164 ``dispatch_id`` reuses the step id so the step's incoming dependency edges are
4165 unchanged; ``join_id`` carries the step's outgoing edges.
4166 """
4168 dispatch_id: str
4169 worker_id: str
4170 fallback_id: str
4171 join_id: str
4172 dispatch_fn: Any
4173 route_fn: Any
4174 worker_fn: Any
4175 join_fn: Any
4178@dataclass(frozen=True)
4179class _MapReduceItemOutcome:
4180 """One worker's per-item result plus its errors, side channels and writes.
4182 ``errors`` carries the item's soft errors (and the inherited soft errors it
4183 read), so the join can fold them in item order exactly like the in-node
4184 concurrent path. ``channel_writes`` is empty for the parallel strategy (its
4185 channel writes are derived from the item result at the join) and carries the
4186 foreach body's pending writes so the join can stage them in item order.
4187 """
4189 result: Any
4190 metrics: dict | None
4191 tool_meta: dict | None
4192 channel_writes: tuple[tuple[str, Any], ...]
4193 errors: tuple[str, ...] = ()
4196@dataclass(frozen=True)
4197class _MapReducePlan:
4198 """The per-step behaviour injected into the shared Send machinery.
4200 The parallel strategy runs a single agent call per item; the foreach
4201 strategy runs the body (or agent) per item. ``item_lister`` resolves the
4202 fan-out list, ``item_runner`` executes one item, ``result_assembler``
4203 shapes the step result, and ``channel_write_stager`` stages per-item
4204 channel writes at the join.
4205 """
4207 item_lister: Any
4208 item_runner: Any
4209 result_assembler: Any
4210 channel_write_stager: Any
4213def _map_reduce_worker_payload(flat: dict, item: Any, index: int) -> dict:
4214 """Build a worker's ``Send`` payload from the resolved source state."""
4215 return {**flat, "item": item, "_item": item, _MAP_REDUCE_ITEM_INDEX: index}
4218def _map_reduce_worker_output(
4219 step_id: str,
4220 index: int,
4221 outcome: _MapReduceItemOutcome,
4222) -> PipelineState:
4223 """The accumulator record carrying every per-item fold input.
4225 The slot is ``{index: {"result", "writes", "errors", "metrics",
4226 "tool_meta"}}`` so the join can assemble the step result, stage channel
4227 writes, and fold soft errors plus side channels in item order — the same
4228 observable result as the in-node concurrent path.
4229 """
4230 record = {
4231 "result": outcome.result,
4232 "writes": list(outcome.channel_writes),
4233 "errors": list(outcome.errors),
4234 "metrics": outcome.metrics if isinstance(outcome.metrics, dict) else None,
4235 "tool_meta": (
4236 outcome.tool_meta if isinstance(outcome.tool_meta, dict) else None
4237 ),
4238 }
4239 return {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: {index: record}}}
4242def _make_map_reduce_route_fn(
4243 plan: _MapReducePlan,
4244 step_runner: Any,
4245 channel_names: frozenset[str],
4246 worker_id: str,
4247 fallback_id: str,
4248 join_id: str,
4249) -> Any:
4250 def route(state: PipelineState) -> list:
4251 flat = _flatten_channel_state(state) if channel_names else state
4252 ctx = step_runner._make_ctx(flat)
4253 items = plan.item_lister(flat, ctx)
4254 if items is None:
4255 return [Send(fallback_id, state)]
4256 if not items:
4257 return [Send(join_id, state)]
4258 return [
4259 Send(worker_id, _map_reduce_worker_payload(flat, item, index))
4260 for index, item in enumerate(items)
4261 ]
4263 return route
4266def _make_map_reduce_worker_fn(
4267 step: dict,
4268 plan: _MapReducePlan,
4269 step_runner: Any,
4270 gate: threading.Semaphore,
4271) -> Any:
4272 step_id = step.get("id", "?")
4274 def worker(state: PipelineState) -> PipelineState:
4275 index = state[_MAP_REDUCE_ITEM_INDEX]
4276 item_state = {
4277 key: value
4278 for key, value in state.items()
4279 if key != _MAP_REDUCE_ITEM_INDEX
4280 }
4281 item = item_state.get("item")
4282 item_ctx = step_runner._make_ctx(item_state)
4283 with gate:
4284 try:
4285 outcome = plan.item_runner(
4286 state, index, item, item_state, item_ctx
4287 )
4288 except SkillLoadError:
4289 raise
4290 except RunHardFailure:
4291 raise
4292 except Exception as exc: # noqa: BLE001
4293 log.error("[step] %s item failed: %s", step_id, exc)
4294 outcome = _MapReduceItemOutcome(
4295 {"error": str(exc)}, None, None, ()
4296 )
4297 return _map_reduce_worker_output(step_id, index, outcome)
4299 return worker
4302def _fold_map_reduce_errors(records: list, base_errors: list) -> list:
4303 """Fold every item's soft errors into *base_errors* in item order."""
4304 merged = list(base_errors)
4305 for record in records:
4306 if not isinstance(record, dict):
4307 continue
4308 merged = StepRunner._merge_errors(merged, record.get("errors") or [])
4309 return merged
4312def _fold_map_reduce_side_channel(records: list, record_key: str) -> dict:
4313 """Fold every item's side-channel map in item order (later item wins)."""
4314 folded: dict = {}
4315 for record in records:
4316 if not isinstance(record, dict):
4317 continue
4318 value = record.get(record_key)
4319 if isinstance(value, dict):
4320 folded.update(value)
4321 return folded
4324def _fold_map_reduce_extras(flat: dict, records: list) -> dict:
4325 """The join's per-item state extras: soft errors plus folded side channels.
4327 Mirrors the in-node concurrent path: errors are merged in item order with
4328 dedup, and the budget/side-channel counters follow the same item order.
4329 """
4330 extras: dict = {}
4331 base_errors = list(flat.get("errors") or [])
4332 errors = _fold_map_reduce_errors(records, base_errors)
4333 if errors != base_errors:
4334 extras["errors"] = errors
4335 for state_key, record_key in (
4336 ("step_context_metrics", "metrics"),
4337 ("step_tool_meta", "tool_meta"),
4338 ):
4339 folded = _fold_map_reduce_side_channel(records, record_key)
4340 if folded:
4341 extras[state_key] = _merge_side_channel(flat.get(state_key), folded)
4342 return extras
4345def _make_map_reduce_join_fn(
4346 step: dict,
4347 plan: _MapReducePlan,
4348 step_runner: Any,
4349 channel_names: frozenset[str],
4350 channel_types: dict[str, str],
4351) -> Any:
4352 step_id = step.get("id", "?")
4354 def join(state: PipelineState) -> PipelineState:
4355 flat = _flatten_channel_state(state) if channel_names else state
4356 accumulated = state.get(_MAP_REDUCE_ACCUMULATOR_KEY) or {}
4357 entries = accumulated.get(step_id) or {}
4358 clear = {_MAP_REDUCE_ACCUMULATOR_KEY: {step_id: None}}
4359 ctx = step_runner._make_ctx(flat)
4360 items = plan.item_lister(flat, ctx)
4361 if items is None:
4362 return clear
4363 records = [entries.get(index) for index in range(len(items))]
4364 results = [
4365 record.get("result") if isinstance(record, dict) else None
4366 for record in records
4367 ]
4368 step_result = plan.result_assembler(results)
4369 extras = _fold_map_reduce_extras(flat, records)
4370 if not channel_names:
4371 return {step_id: step_result, **extras, **clear}
4372 working_set = ChannelWorkingSet(flat, channel_types)
4373 plan.channel_write_stager(results, records, working_set)
4374 written_names = {name for name, _ in working_set.pending_writes()}
4375 channels = working_set.commit()
4376 emitted = {name: channels[name] for name in written_names}
4377 core = _strip_channel_keys(
4378 {**flat, step_id: step_result, **extras}, channel_names
4379 )
4380 return {_CORE: {**core, **emitted}, **clear}
4382 return join
4385def _make_parallel_map_reduce_plan(
4386 step: dict, step_runner: Any
4387) -> _MapReducePlan:
4388 """Plan for a concurrent `parallel` step: one agent call per item."""
4389 resolved_agent = step_runner.resolve_agent(step)
4391 def run_item(state, index, item, item_state, item_ctx):
4392 result, metrics, tool_meta, errors = step_runner._invoke_parallel_item(
4393 step, resolved_agent, item, item_state, item_ctx
4394 )
4395 return _MapReduceItemOutcome(
4396 result, metrics, tool_meta, (), tuple(errors)
4397 )
4399 return _MapReducePlan(
4400 item_lister=lambda flat, ctx: step_runner._parallel_fanout_item_list(
4401 step, flat, ctx
4402 ),
4403 item_runner=run_item,
4404 result_assembler=lambda results: {"results": results},
4405 channel_write_stager=lambda results, records, working_set: (
4406 step_runner._stage_fanout_channel_writes(step, results, working_set)
4407 ),
4408 )
4411def _worker_committed_flat(
4412 state: PipelineState, channel_names: frozenset[str]
4413) -> dict:
4414 """The committed channel snapshot visible to one worker.
4416 A channels-path worker receives the flattened snapshot as its payload, so
4417 its channel keys sit at the top level; the typed ``__core__`` shape (only
4418 seen by the join) is unwrapped.
4419 """
4420 if _CORE in state:
4421 return _flatten_channel_state(state)
4422 return {name: state[name] for name in channel_names if name in state}
4425def _pending_channel_writes(
4426 working_set: ChannelWorkingSet | None,
4427) -> tuple[tuple[str, Any], ...]:
4428 if working_set is None:
4429 return ()
4430 return tuple(working_set.pending_writes())
4433def _stage_map_reduce_item_channel_writes(
4434 results: list, records: list, working_set: ChannelWorkingSet
4435) -> None:
4436 """Stage each item's carried channel writes in item order."""
4437 for record in records:
4438 if not isinstance(record, dict):
4439 continue
4440 for name, value in record.get("writes") or ():
4441 working_set.stage(name, value)
4444def _make_foreach_map_reduce_plan(
4445 step: dict,
4446 step_runner: Any,
4447 channel_names: frozenset[str],
4448 channel_types: dict[str, str],
4449) -> _MapReducePlan:
4450 """Plan for a concurrent `foreach` step: the body (or agent) runs per item."""
4451 step_id = step.get("id", "?")
4452 body = step.get("body") or []
4453 declared_output = step.get("output")
4454 runs_body = bool(body)
4456 def result_assembler(results):
4457 if runs_body:
4458 return StepRunner._aggregate_body_results(results)
4459 return results
4461 def run_item(state, index, item, item_state, _item_ctx):
4462 item_working_set = (
4463 ChannelWorkingSet(
4464 _worker_committed_flat(state, channel_names), channel_types
4465 )
4466 if channel_names
4467 else None
4468 )
4469 iter_state = {
4470 **item_state, "item": item, "_item": item, "_foreach_item": item
4471 }
4472 if runs_body:
4473 for sub_step in body:
4474 iter_state = step_runner._run_body_step(
4475 sub_step, iter_state, item_working_set
4476 )
4477 return _MapReduceItemOutcome(
4478 step_runner._body_item_result(iter_state, declared_output),
4479 iter_state.get("step_context_metrics"),
4480 iter_state.get("step_tool_meta"),
4481 _pending_channel_writes(item_working_set),
4482 tuple(iter_state.get("errors") or []),
4483 )
4484 resolved = step_runner.resolve_agent(step)
4485 item_read_state = step_runner._read_state(iter_state, item_working_set)
4486 item_ctx = step_runner._make_ctx(item_read_state)
4487 resolved_step = step_runner._resolve_step_runtime(
4488 step, item_read_state, item_ctx
4489 )
4490 resolved_step = step_runner._attach_communication_context(
4491 resolved_step, item_read_state
4492 )
4493 resolved_step = step_runner._attach_resolved_tools(
4494 resolved_step, resolved
4495 )
4496 result_state = step_runner._invoke_agent(
4497 resolved, resolved_step, iter_state
4498 )
4499 return _MapReduceItemOutcome(
4500 result_state.get(step_id),
4501 result_state.get("step_context_metrics"),
4502 result_state.get("step_tool_meta"),
4503 _pending_channel_writes(item_working_set),
4504 tuple(result_state.get("errors") or []),
4505 )
4507 return _MapReducePlan(
4508 item_lister=lambda flat, ctx: step_runner._foreach_item_list(
4509 step, flat, ctx
4510 ),
4511 item_runner=run_item,
4512 result_assembler=result_assembler,
4513 channel_write_stager=_stage_map_reduce_item_channel_writes,
4514 )
4517def _make_map_reduce_parts(
4518 step: dict,
4519 plan: _MapReducePlan,
4520 step_runner: Any,
4521 channel_names: frozenset[str],
4522 channel_types: dict[str, str],
4523 max_concurrent: int,
4524) -> _MapReduceParts:
4525 step_id = step.get("id", "?")
4526 worker_id = f"{step_id}__worker"
4527 fallback_id = f"{step_id}__fallback"
4528 join_id = f"{step_id}__join"
4529 gate = threading.Semaphore(max(1, max_concurrent))
4530 return _MapReduceParts(
4531 dispatch_id=step_id,
4532 worker_id=worker_id,
4533 fallback_id=fallback_id,
4534 join_id=join_id,
4535 dispatch_fn=lambda state: {},
4536 route_fn=_make_map_reduce_route_fn(
4537 plan, step_runner, channel_names, worker_id, fallback_id, join_id
4538 ),
4539 worker_fn=_make_map_reduce_worker_fn(step, plan, step_runner, gate),
4540 join_fn=_make_map_reduce_join_fn(
4541 step, plan, step_runner, channel_names, channel_types
4542 ),
4543 )
4546def _resolved_send_concurrency(step: dict, build_ctx: dict) -> int:
4547 """The build-time resolved ``max_concurrent`` for a Send-eligible step."""
4548 return _resolve_int(step.get("max_concurrent"), build_ctx, 1)
4551def _is_send_parallel_step(
4552 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4553) -> bool:
4554 """Whether a top-level `parallel` step fans out through native ``Send``.
4556 Restricted to a singleton group's statically concurrent, agent-backed,
4557 item-sourced step with no `routes`, no `condition`/`skip_llm_if`, and an
4558 explicit `max_concurrent > 1`; everything else keeps the in-node path.
4559 """
4560 if step.get("type") != "parallel":
4561 return False
4562 if step.get("id") in route_plan.target_node_ids:
4563 return False
4564 if step.get("routes") or not step.get("agent"):
4565 return False
4566 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
4567 return False
4568 if StepRunner._declared_item_source(step) is None:
4569 return False
4570 if step.get("max_concurrent") is None:
4571 return False
4572 return _resolved_send_concurrency(step, build_ctx) > 1
4575def _is_send_foreach_step(
4576 step: dict, route_plan: "_RoutePlan", build_ctx: dict
4577) -> bool:
4578 """Whether a top-level `foreach` step fans out through native ``Send``.
4580 Restricted to a singleton group's body- or agent-backed, item-sourced step
4581 with no `routes`, no `condition`/`skip_llm_if`, no step-level
4582 `output_channel`, and no `max_output`; an explicit `max_concurrent > 1` is
4583 required. The default (absent or ``1``) keeps the sequential in-node path,
4584 and the excluded edges keep their existing in-node semantics.
4585 """
4586 if step.get("type") != "foreach":
4587 return False
4588 if step.get("id") in route_plan.target_node_ids:
4589 return False
4590 if step.get("routes") or step.get("output_channel"):
4591 return False
4592 if step.get("condition") is not None or step.get("skip_llm_if") is not None:
4593 return False
4594 if step.get("max_output") is not None:
4595 return False
4596 if StepRunner._declared_item_source(step) is None:
4597 return False
4598 if not (step.get("body") or step.get("agent")):
4599 return False
4600 if step.get("max_concurrent") is None:
4601 return False
4602 return _resolved_send_concurrency(step, build_ctx) > 1
4605class _ChannelGraphShim:
4606 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
4608 Preserves the invoke contract used by main.py and tests: a flat dict in,
4609 a flat dict out. The flat input is seeded through ``seed_committed`` and
4610 nested under ``__core__`` for the compiled graph; the committed snapshot is
4611 unwrapped back out. Streaming and state surfaces are not exposed because
4612 they would leak or accept the internal ``__core__`` shape; read-only
4613 introspection delegates to the compiled graph.
4614 """
4616 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
4617 self._compiled = compiled
4618 self._specs = specs
4620 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
4621 seeded = seed_committed(self._specs, flat_state)
4622 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
4623 return unwrap(result)
4625 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
4626 seeded = seed_committed(self._specs, flat_state)
4627 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
4628 return unwrap(result)
4630 def stream(self, *args: Any, **kwargs: Any) -> Any:
4631 self._reject_internal_shape("stream")
4633 def astream(self, *args: Any, **kwargs: Any) -> Any:
4634 self._reject_internal_shape("astream")
4636 def update_state(self, *args: Any, **kwargs: Any) -> Any:
4637 self._reject_internal_shape("update_state")
4639 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
4640 self._reject_internal_shape("aupdate_state")
4642 def get_state(self, *args: Any, **kwargs: Any) -> Any:
4643 self._reject_internal_shape("get_state")
4645 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
4646 self._reject_internal_shape("aget_state")
4648 def _reject_internal_shape(self, surface: str) -> None:
4649 raise NotImplementedError(
4650 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
4651 ".invoke()/.ainvoke() contract is exposed through the channel "
4652 "boundary shim"
4653 )
4655 def __getattr__(self, name: str) -> Any:
4656 return getattr(self._compiled, name)
4659class GraphBuilder:
4660 """
4661 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
4663 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
4664 käytetään vain {{ }} -templatejen evaluointiin.
4665 """
4667 def __init__(
4668 self,
4669 registry: ManifestRegistry,
4670 builtins: BuiltinExecutor | None = None,
4671 invoke_agent_fn: Any | None = None,
4672 run_context: dict | None = None,
4673 tracer: Any | None = None,
4674 ) -> None:
4675 self._registry = registry
4676 self._resolver = AgentResolver(registry)
4677 self._invoke_agent_fn = invoke_agent_fn
4678 self._builtins = builtins or _DEFAULT_BUILTINS
4679 if invoke_agent_fn is not None:
4680 self._builtins.invoke_agent_fn = invoke_agent_fn
4681 self._step_functions = StepFunctionExecutor(registry)
4682 self._tool_registry = build_platform_registry()
4683 self._tool_resolver = ToolResolver(self._tool_registry, self._step_functions)
4684 self._step_runner = StepRunner(
4685 self._resolver, self._builtins, invoke_agent_fn, run_context,
4686 step_functions=self._step_functions,
4687 tracer=tracer,
4688 tool_resolver=self._tool_resolver,
4689 )
4691 def build(self, pipeline: dict): # -> CompiledGraph
4692 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
4693 try:
4694 from langgraph.graph import StateGraph, END
4695 except ImportError as exc:
4696 raise ImportError(
4697 "langgraph is required to build a workflow. "
4698 "Install it with: pip install langgraph"
4699 ) from exc
4701 raw_steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
4703 channels: list[dict] = _validated_channels(
4704 (pipeline.get("spec") or {}).get("channels")
4705 )
4706 specs = parse_channel_specs(
4707 [{"name": ch["name"], "type": ch.get("type", LAST_VALUE)} for ch in channels]
4708 )
4709 channel_types = {spec.name: spec.type for spec in specs}
4710 channel_names = frozenset(channel_types)
4712 # Build-time call-step composition (ADR-0037 D4): each invoked Function
4713 # body is compiled into its own native LangGraph subgraph node, so the
4714 # runtime graph executes only native steps; a call step never survives.
4715 steps = self._compose_calls(
4716 raw_steps, channels, channel_names, channel_types
4717 )
4718 if channels:
4719 _validate_channel_names(channels, steps)
4720 _validate_channel_inputs(channels, steps)
4721 _validate_step_channel_bindings(steps, channel_names)
4722 ordered = _topological_order(steps)
4724 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
4725 envs = (pipeline.get("spec") or {}).get("envs")
4726 if envs is not None and not isinstance(envs, dict):
4727 raise RunHardFailure(
4728 "VALIDATION_ERROR",
4729 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
4730 "spec.envs must be a map of env name → image ref (ADR-0012)",
4731 )
4733 graph = self._assemble_graph(ordered, channel_names, channel_types)
4734 compiled = graph.compile()
4735 if channel_names:
4736 return _ChannelGraphShim(compiled, specs)
4737 return compiled
4739 def _compose_calls(
4740 self,
4741 steps: list[dict],
4742 channels: list[dict],
4743 channel_names: frozenset[str],
4744 channel_types: dict[str, str],
4745 ) -> list[dict]:
4746 """Replace `call` steps with compiled native subgraph nodes.
4748 Each invoked Function body is assembled through ``_assemble_graph`` and
4749 compiled, then substituted as a single ``__subgraph__`` placeholder that
4750 keeps the call id and its ``depends_on``. Downstream steps addressing the
4751 call id therefore address the subgraph node unchanged.
4752 """
4754 def assemble_subgraph(sub_steps: list[dict]):
4755 return self._assemble_graph(
4756 sub_steps, channel_names, channel_types
4757 ).compile()
4759 composer = _CallComposer(
4760 self._registry,
4761 assemble_subgraph,
4762 channels,
4763 channel_names,
4764 )
4765 return composer.compose(steps)
4767 def _assemble_graph(
4768 self,
4769 steps: list[dict],
4770 channel_names: frozenset[str],
4771 channel_types: dict[str, str],
4772 ) -> "StateGraph":
4773 from langgraph.graph import StateGraph, END
4775 # Group steps into LangGraph nodes:
4776 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
4777 nodes: list[list[dict]] = []
4778 for step in steps:
4779 dep_sig = frozenset(step.get("depends_on") or [])
4780 if nodes:
4781 last_group_sig = frozenset(
4782 dep
4783 for s in nodes[-1]
4784 for dep in (s.get("depends_on") or [])
4785 )
4786 last_types = {s.get("type") for s in nodes[-1]}
4787 if (
4788 dep_sig == last_group_sig
4789 and step.get("type") == "parallel"
4790 and last_types == {"parallel"}
4791 and not step.get("routes")
4792 and not any(s.get("routes") for s in nodes[-1])
4793 ):
4794 nodes[-1].append(step)
4795 continue
4796 nodes.append([step])
4798 state_schema = (
4799 _build_state_schema()
4800 if channel_names
4801 else Annotated[dict[str, Any], _merge_state]
4802 )
4803 graph: StateGraph = StateGraph(state_schema)
4804 step_runner = self._step_runner
4805 step_runner.set_channel_routing(channel_names)
4807 node_ids: list[str] = []
4808 for group in nodes:
4809 if len(group) == 1:
4810 node_ids.append(group[0]["id"])
4811 else:
4812 node_ids.append("parallel_" + "_".join(s["id"] for s in group))
4814 route_plan = _build_route_plan(nodes, node_ids) if nodes else None
4816 send_parts: dict[int, _MapReduceParts] = {}
4817 if route_plan is not None:
4818 build_ctx = step_runner._make_ctx({})
4819 for index, group in enumerate(nodes):
4820 if len(group) != 1:
4821 continue
4822 send_candidate = group[0]
4823 if _is_send_parallel_step(send_candidate, route_plan, build_ctx):
4824 plan = _make_parallel_map_reduce_plan(
4825 send_candidate, step_runner
4826 )
4827 elif _is_send_foreach_step(send_candidate, route_plan, build_ctx):
4828 plan = _make_foreach_map_reduce_plan(
4829 send_candidate, step_runner, channel_names, channel_types
4830 )
4831 else:
4832 continue
4833 send_parts[index] = _make_map_reduce_parts(
4834 send_candidate,
4835 plan,
4836 step_runner,
4837 channel_names,
4838 channel_types,
4839 _resolved_send_concurrency(send_candidate, build_ctx),
4840 )
4842 entry_ids: list[str] = []
4843 exit_ids: list[str] = []
4844 for index, group in enumerate(nodes):
4845 parts = send_parts.get(index)
4846 if parts is None:
4847 captured_group = list(group)
4848 if len(captured_group) == 1:
4849 first = captured_group[0]
4850 if first.get("type") == _SUBGRAPH_STEP_TYPE:
4851 graph.add_node(
4852 node_ids[index],
4853 _make_subgraph_node(
4854 first[_SUBGRAPH_GRAPH_KEY],
4855 first.get("condition"),
4856 step_runner,
4857 channel_names,
4858 ),
4859 )
4860 entry_ids.append(node_ids[index])
4861 exit_ids.append(node_ids[index])
4862 continue
4863 fn = _make_step_node(
4864 first, step_runner, channel_names, channel_types,
4865 )
4866 else:
4867 fn = _make_parallel_node(
4868 captured_group,
4869 step_runner,
4870 channel_names,
4871 channel_types,
4872 )
4873 graph.add_node(node_ids[index], fn)
4874 entry_ids.append(node_ids[index])
4875 exit_ids.append(node_ids[index])
4876 continue
4877 fallback_step = group[0]
4878 graph.add_node(parts.dispatch_id, parts.dispatch_fn)
4879 graph.add_node(parts.worker_id, parts.worker_fn)
4880 graph.add_node(
4881 parts.fallback_id,
4882 _make_step_node(
4883 fallback_step, step_runner, channel_names, channel_types
4884 ),
4885 )
4886 graph.add_node(parts.join_id, parts.join_fn)
4887 graph.add_conditional_edges(
4888 parts.dispatch_id,
4889 parts.route_fn,
4890 [parts.worker_id, parts.fallback_id],
4891 )
4892 graph.add_edge(parts.worker_id, parts.join_id)
4893 graph.add_edge(parts.fallback_id, parts.join_id)
4894 entry_ids.append(parts.dispatch_id)
4895 exit_ids.append(parts.join_id)
4897 if nodes:
4898 _wire_dependency_edges(
4899 graph, nodes, node_ids, END, route_plan, entry_ids, exit_ids
4900 )
4901 _wire_route_edges(graph, route_plan, END)
4902 else:
4903 graph.add_node("noop", lambda s: s)
4904 graph.set_entry_point("noop")
4905 graph.add_edge("noop", END)
4907 return graph
4910# ---------------------------------------------------------------------------
4911# Julkinen API
4912# ---------------------------------------------------------------------------
4914_registry_cache: ManifestRegistry | None = None
4917def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
4918 global _registry_cache
4919 if _registry_cache is None or dirs is not None:
4920 loader = ManifestLoader(dirs)
4921 manifests = loader.load_all()
4922 _registry_cache = ManifestRegistry(manifests)
4923 return _registry_cache
4926def build_workflow(
4927 pipeline_name: str,
4928 *,
4929 dirs: list[Path] | None = None,
4930 builtins: BuiltinExecutor | None = None,
4931 invoke_agent_fn: Any | None = None,
4932 run_context: dict | None = None,
4933 tracer: Any | None = None,
4934):
4935 """
4936 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
4937 LangGraph-workflown.
4939 Args:
4940 pipeline_name: Pipeline metadata.name -arvo.
4941 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
4942 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
4943 invoke_agent_fn: fn(resolved_spec, step, state) → state.
4944 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
4946 Returns:
4947 Compilattu LangGraph CompiledGraph.
4948 """
4949 registry = _get_registry(dirs)
4950 pipeline = registry.get("Pipeline", pipeline_name)
4951 if pipeline is None:
4952 available = [
4953 (m.get("metadata") or {}).get("name")
4954 for m in registry.all_of_kind("Pipeline")
4955 ]
4956 raise RunHardFailure(
4957 "VALIDATION_ERROR",
4958 f"Pipeline '{pipeline_name}' not found. Available: {available}",
4959 )
4960 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
4961 return builder.build(pipeline)