Coverage for core/langgraph/runner.py: 92%
1766 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-20 02:14 +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 contextvars
19import hashlib
20import logging
21import os
22import re
23import threading
24import uuid
25from concurrent.futures import ThreadPoolExecutor, as_completed
26from pathlib import Path
27from typing import Any
29import yaml
31# SkillLoadError-import (ei sirkulaari-riskiä — _llm importtaa vain stdlibin):
32# testit ja discover-wrapperit lataavat _llm:n tasaisena moduulina (sys.path
33# insert core/langgraph/), main.py qualified (langgraph._llm, PYTHONPATH=/app).
34# Kummatkin pinnat käännetään samaan _llm-tiedostoon; try/except varmistaa,
35# että luokka on sama objekti kuin nostavalla puolella molemmissa konteksteissa.
36try:
37 from _llm import DEFAULT_MAX_PARALLEL, SkillLoadError # noqa: F401
38except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
39 from langgraph._llm import DEFAULT_MAX_PARALLEL, SkillLoadError # noqa: F401
41# Same dual-import pattern as _llm (flat sys.path and qualified PYTHONPATH=/app)
42try:
43 from _step_functions import StepFunctionExecutor # noqa: F401
44except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
45 from langgraph._step_functions import StepFunctionExecutor # noqa: F401
47# Core-owned prompt-text registry for whitelisted builtins — same dual-import
48# pattern as _llm.
49try:
50 from _instructions import (
51 USAGE_INSTRUCTIONS_KEY,
52 instruction_text,
53 resolve_usage_instructions,
54 )
55except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
56 from langgraph._instructions import (
57 USAGE_INSTRUCTIONS_KEY,
58 instruction_text,
59 resolve_usage_instructions,
60 )
62# COMMUNICATION-context projection (declared input.channels reads) — same
63# dual-import pattern as _llm.
64try:
65 from _data_structure import (
66 COMMUNICATION_CONTEXT_KEY,
67 resolve_communication_context,
68 )
69except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
70 from langgraph._data_structure import (
71 COMMUNICATION_CONTEXT_KEY,
72 resolve_communication_context,
73 )
75# Core-owned channel-context storage/commit/seeding primitives — same dual-import
76# pattern. The channel path commits through these; it no longer materializes
77# native LangGraph channels.
78try:
79 from _channels import (
80 CORE_KEY,
81 LAST_VALUE,
82 ChannelSpec,
83 ChannelWorkingSet,
84 parse_channel_specs,
85 seed_committed,
86 unwrap,
87 wrap,
88 )
89except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
90 from langgraph._channels import (
91 CORE_KEY,
92 LAST_VALUE,
93 ChannelSpec,
94 ChannelWorkingSet,
95 parse_channel_specs,
96 seed_committed,
97 unwrap,
98 wrap,
99 )
101# Agent output envelope error code — same dual-import pattern.
102try:
103 from _envelope import ENVELOPE_ERROR_CODE, ENVELOPE_STATUSES, is_error_envelope
104except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
105 from langgraph._envelope import ( # noqa: F401
106 ENVELOPE_ERROR_CODE,
107 ENVELOPE_STATUSES,
108 is_error_envelope,
109 )
111log = logging.getLogger(__name__)
114class RunHardFailure(Exception):
115 """Run-tason kova virhe: ajo pysähtyy heti (phase failed, exit 1).
117 Nostetaan kun step-tulos kantaa kovan virheen: TRUNCATION_ERROR-koodi
118 (kuratoidun kontekstin katkaisu), TOOL_FAILURE-koodi (työkalun
119 suorituksen tekninen vika), AGENT_ERROR-koodi (status:error-envelope,
120 ENVELOPE_ERROR_CODE) tai kontekstiylityksen signatuurin kantava
121 VALIDATION_ERROR (assembled_tokens + context_budget,
122 _llm._context_overflow_result).
123 Kaikki muut virheluokat (budget:*, parse_error, timeout, LLM-virheet)
124 säilyvät nykyisessä partial-semantiikassa. Stepin details kulkeutuu
125 attribuuttina loppu-PATCHin error.details-kenttään.
126 """
128 HARD_ERROR_CODES: frozenset[str] = frozenset(
129 {"TRUNCATION_ERROR", "TOOL_FAILURE", ENVELOPE_ERROR_CODE}
130 )
132 def __init__(self, code: str, message: str, details: dict | None = None) -> None:
133 super().__init__(message)
134 self.code = code
135 self.message = message
136 self.details = details
139class SpawnBudget:
140 """Per-tree agent-call budget shared across all spawn nodes.
142 The whole spawn tree draws from a single counter: every agent invocation
143 (the director included) consumes one unit, so the total number of
144 agent invocations the tree may make is bounded by `max_agent_calls`. This is a
145 per-agent-run budget, NOT an LLM-roundtrip budget: a single agent run may
146 make several LLM calls (e.g. an agent-side tool loop via file_access);
147 those LLM roundtrips are bounded by the run's `max_llm_calls`, not by
148 `max_agent_calls`. There is no engine-level agent retry loop. Thread-safe so
149 a concurrent caller cannot over-commit the shared counter.
150 """
152 def __init__(self, max_agent_calls: int | None) -> None:
153 self._remaining: int | None = (
154 None if max_agent_calls is None else max(0, int(max_agent_calls))
155 )
156 self._lock = threading.Lock()
158 def consume(self) -> bool:
159 """Reserve one agent call. Returns False when the budget is exhausted."""
160 with self._lock:
161 if self._remaining is None:
162 return True
163 if self._remaining <= 0:
164 return False
165 self._remaining -= 1
166 return True
169_SPAWN_SOFT_FAIL_KEYS: frozenset[str] = frozenset({"error", "parse_error", "escalate"})
171_SPAWN_DEFAULT_MAX_AGENT_CALLS: int = 64
172_SPAWN_DEFAULT_MAX_DEPTH: int = 1
173_SPAWN_DEFAULT_MAX_SPAWN_PER_NODE: int = 16
175# ---------------------------------------------------------------------------
176# Polut
177# ---------------------------------------------------------------------------
179PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", "/project"))
180MANIFEST_DIRS = [
181 PROJECT_ROOT / ".agent-platform" / "workflows",
182 PROJECT_ROOT / ".agent-platform" / "agents",
183]
184SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "manifest.schema.json"
187def resolve_manifest_dirs(clone_root: Path) -> list[Path]:
188 """Palauttaa olemassa olevat manifestihakemistot kloonatusta manifest-reposta.
190 Skannaa ainoastaan `.agent-platform/workflows` ja `.agent-platform/agents` —
191 ei koskaan koko kloonia. [] jos kumpaakaan ei ole (kutsuja → fail-fast).
192 """
193 return [
194 d for d in (
195 clone_root / ".agent-platform" / "workflows",
196 clone_root / ".agent-platform" / "agents",
197 )
198 if d.is_dir()
199 ]
201# ---------------------------------------------------------------------------
202# Template-resoluutio
203# ---------------------------------------------------------------------------
205_TMPL_RE = re.compile(r'\{\{\s*(.*?)\s*\}\}')
207# Sentinel for an unresolvable dot-path segment (distinct from a resolved None).
208_UNRESOLVED = object()
211def _split_key_indices(part: str) -> tuple[str, list[int]]:
212 """Split a dot-path segment into a base key and a list of indices.
214 ``'a[0][1]'`` → ``('a', [0, 1])``; ``'a'`` → ``('a', [])``; ``'[0]'`` →
215 ``('', [0])``. Only non-negative indices are supported (list access).
216 """
217 m = re.fullmatch(r'([^\[]*)((\[\d+\])*)', part)
218 base = m.group(1)
219 indices = [int(x) for x in re.findall(r'\d+', m.group(2))]
220 return base, indices
223def _apply_indices(obj: Any, indices: list[int]) -> Any:
224 """Index a (possibly chained) list/tuple; ``_UNRESOLVED`` on out-of-range."""
225 for idx in indices:
226 if not isinstance(obj, (list, tuple)) or not (0 <= idx < len(obj)):
227 return _UNRESOLVED
228 obj = obj[idx]
229 return obj
232def _dot_get_segment(obj: Any, part: str) -> Any:
233 """Resolve one dot-path segment against a dict or an indexed sequence.
235 A segment may be a bare key (``'a'``), a keyed list-index (``'a[0]'``,
236 ``'a[0][1]'``), or a bare index into a sequence. Returns the resolved value
237 or the ``_UNRESOLVED`` sentinel when the segment cannot be resolved (missing
238 key, non-dict/non-sequence mid-path, or out-of-range index).
239 """
240 base, indices = _split_key_indices(part)
241 if isinstance(obj, dict):
242 if base:
243 if base not in obj:
244 return _UNRESOLVED
245 obj = obj[base]
246 return _apply_indices(obj, indices)
247 if isinstance(obj, (list, tuple)):
248 if indices:
249 return _apply_indices(obj, indices)
250 if base and base.isdigit():
251 return _apply_indices(obj, [int(base)])
252 return _UNRESOLVED
253 return _UNRESOLVED
256def _dot_get(obj: Any, path: str) -> Any:
257 """Navigate nested dicts/lists via dot-path: 'a.b.c' → obj['a']['b']['c'].
259 Supports list-index segments (``results[1].field``) and dot-into-aggregate
260 traversal (missing key / non-dict-non-list mid-path / out-of-range index
261 resolve to None — lenient semantics for non-output refs).
262 """
263 for part in path.split('.'):
264 obj = _dot_get_segment(obj, part)
265 if obj is _UNRESOLVED:
266 return None
267 return obj
270def _dot_get_strict(obj: Any, path: str) -> Any:
271 """Dot-path navigation that raises on an unresolved reference.
273 A reference is unresolved when a key on the path is absent or a segment is
274 not a mapping — used by condition evaluation so an undefined or skipped
275 step's field is never silently evaluated to falsy (ADR-0012 spirit). Also
276 used by ``__call_emit`` to distinguish a resolved output source from an
277 unresolved one (required output → hard error, ADR-0012).
278 """
279 for part in path.split('.'):
280 if obj is None:
281 raise ValueError(
282 f"Unresolved reference '{path}': '{part}' is not a mapping"
283 )
284 obj = _dot_get_segment(obj, part)
285 if obj is _UNRESOLVED:
286 raise ValueError(
287 f"Unresolved reference '{path}': '{part}' is not a mapping"
288 )
289 return obj
292def materialize_pipeline_output(pipeline_spec: dict, final_state: dict) -> Any:
293 """Resolve a Pipeline's declared run-end output.
295 Reads ``spec.output.source`` (a dot-path) out of ``final_state`` via
296 ``_dot_get`` and returns the surfaced value. Returns ``None`` when no
297 ``spec.output`` is declared, when it has no ``source``, or when the source
298 is unresolved — this is a run-end surfacing (distinct from the emit-time
299 hard-error used for function-interface outputs): an optional run output
300 that cannot be resolved surfaces as None, never a run-end failure.
301 """
302 out_spec = (pipeline_spec or {}).get("output")
303 if not out_spec:
304 return None
305 source = out_spec.get("source")
306 if not source:
307 return None
308 return _dot_get(final_state, source)
311def _merge_metrics(base: dict | None, additions: dict) -> dict:
312 """Yhdistä step_context_metrics-envelopet: myöhempi kirjoitus voittaa
313 (sama semantiikka kuin state-merge {**state, ...})."""
314 merged = dict(base or {})
315 merged.update(additions)
316 return merged
319_MULTI_OPS = ('<=', '>=', '==', '!=')
320_SINGLE_OPS = '+-<>'
321_WORD_BREAK = set("()[],+<=>!'-")
322_CMP_OPS = frozenset(('==', '!=', '<=', '>=', '<', '>'))
325def _tokenize_expr(expr: str) -> list[tuple[str, Any]]:
326 """Split an expression into (kind, value) tokens.
328 Kinds: 'op' (+ - <= >= == != < >), 'punct' ( ) [ ] ,, 'num', 'str',
329 'name'. A 'name' is a dot-path reference or a bare word (e.g. an unquoted
330 comparison literal, true/false/None, and/or/not).
331 """
332 tokens: list[tuple[str, Any]] = []
333 i, n = 0, len(expr)
334 while i < n:
335 ch = expr[i]
336 if ch.isspace():
337 i += 1
338 continue
339 if ch in '()[],':
340 tokens.append(('punct', ch))
341 i += 1
342 continue
343 if ch in '+-<=>!':
344 two = expr[i:i + 2]
345 if two in _MULTI_OPS:
346 tokens.append(('op', two))
347 i += 2
348 continue
349 if ch in _SINGLE_OPS:
350 tokens.append(('op', ch))
351 i += 1
352 continue
353 raise ValueError(f"Invalid operator in expression: {two!r}")
354 if ch in ('\'', '"'):
355 j = i + 1
356 buf: list[str] = []
357 while j < n and expr[j] != ch:
358 buf.append(expr[j])
359 j += 1
360 if j >= n:
361 raise ValueError(f"Unterminated string literal in expression: {expr!r}")
362 tokens.append(('str', ''.join(buf)))
363 i = j + 1
364 continue
365 if ch.isdigit() or (ch == '.' and i + 1 < n and expr[i + 1].isdigit()):
366 j = i
367 if ch == '.':
368 j += 1
369 while j < n and expr[j].isdigit():
370 j += 1
371 tokens.append(('num', float(expr[i:j])))
372 else:
373 while j < n and expr[j].isdigit():
374 j += 1
375 if j < n and expr[j] == '.' and j + 1 < n and expr[j + 1].isdigit():
376 j += 1
377 while j < n and expr[j].isdigit():
378 j += 1
379 tokens.append(('num', float(expr[i:j])))
380 else:
381 tokens.append(('num', int(expr[i:j])))
382 i = j
383 continue
384 j = i
385 while j < n and not expr[j].isspace() and expr[j] not in _WORD_BREAK:
386 j += 1
387 if j == i:
388 raise ValueError(f"Unexpected character in expression: {expr[i:]!r}")
389 tokens.append(('name', expr[i:j]))
390 i = j
391 return tokens
394# AST node shapes:
395# ('lit', value) literal (str/num/bool/None)
396# ('ref', name) dot-path reference
397# ('list', [node, ...]) [a, b, c] literal
398# ('neg', node) unary minus
399# ('not', node) not X
400# ('cmp', op, left, right) comparison
401# ('binop', 'and'|'or'|'+', l, r) binary operator
404class _ExprParser:
405 """Recursive-descent parser for template/condition expressions.
407 Precedence (loosest first): or < and < not < comparison < '+' < unary.
408 Grammar:
409 expr := or_expr
410 or_expr := and_expr ('or' and_expr)*
411 and_expr := not_expr ('and' not_expr)*
412 not_expr := 'not' not_expr | cmp_expr
413 cmp_expr := add_expr (cmp_op add_expr)?
414 add_expr := unary_expr ('+' unary_expr)*
415 unary_expr:= '-' unary_expr | primary
416 primary := num | str | true/false/None | list | '(' expr ')' | name
417 `and`/`or` parse into a full AST but EVALUATE with left-to-right
418 short-circuit, so the right operand is only resolved if needed and an
419 unresolved read there never raises.
420 """
422 def __init__(self, tokens: list[tuple[str, Any]]) -> None:
423 self._tokens = tokens
424 self._i = 0
426 def parse(self) -> tuple:
427 node = self._or_expr()
428 if self._i != len(self._tokens):
429 tok = self._tokens[self._i]
430 raise ValueError(f"Unexpected token in expression: {tok[1]!r}")
431 return node
433 def _peek(self) -> tuple[str, Any] | None:
434 return self._tokens[self._i] if self._i < len(self._tokens) else None
436 def _next(self) -> tuple[str, Any]:
437 tok = self._tokens[self._i]
438 self._i += 1
439 return tok
441 def _peek_name(self, kw: str) -> bool:
442 tok = self._peek()
443 return tok is not None and tok[0] == 'name' and tok[1] == kw
445 def _peek_op(self, op: str) -> bool:
446 tok = self._peek()
447 return tok is not None and tok[0] == 'op' and tok[1] == op
449 def _expect_punct(self, value: str) -> None:
450 tok = self._next()
451 if tok != ('punct', value):
452 raise ValueError(f"Expected '{value}' in expression, got {tok[1]!r}")
454 def _or_expr(self) -> tuple:
455 node = self._and_expr()
456 while self._peek_name('or'):
457 self._next()
458 right = self._and_expr()
459 node = ('binop', 'or', node, right)
460 return node
462 def _and_expr(self) -> tuple:
463 node = self._not_expr()
464 while self._peek_name('and'):
465 self._next()
466 right = self._not_expr()
467 node = ('binop', 'and', node, right)
468 return node
470 def _not_expr(self) -> tuple:
471 if self._peek_name('not'):
472 self._next()
473 return ('not', self._not_expr())
474 return self._cmp_expr()
476 def _cmp_expr(self) -> tuple:
477 node = self._add_expr()
478 tok = self._peek()
479 if tok is not None and tok[0] == 'op' and tok[1] in _CMP_OPS:
480 self._next()
481 right = self._add_expr()
482 node = ('cmp', tok[1], node, right)
483 return node
485 def _add_expr(self) -> tuple:
486 node = self._unary_expr()
487 while self._peek_op('+'):
488 self._next()
489 right = self._unary_expr()
490 node = ('binop', '+', node, right)
491 return node
493 def _unary_expr(self) -> tuple:
494 if self._peek_op('-'):
495 self._next()
496 node = self._unary_expr()
497 if node[0] == 'lit' and isinstance(node[1], (int, float)) and not isinstance(node[1], bool):
498 return ('lit', -node[1])
499 return ('neg', node)
500 return self._primary()
502 def _primary(self) -> tuple:
503 tok = self._peek()
504 if tok is None:
505 raise ValueError("Unexpected end of expression")
506 kind, value = tok
507 if kind == 'num':
508 self._next()
509 return ('lit', value)
510 if kind == 'str':
511 self._next()
512 return ('lit', value)
513 if kind == 'punct':
514 if value == '(':
515 self._next()
516 node = self._or_expr()
517 self._expect_punct(')')
518 return node
519 if value == '[':
520 self._next()
521 items: list[tuple] = []
522 if self._peek() != ('punct', ']'):
523 while True:
524 items.append(self._or_expr())
525 if self._peek() == ('punct', ','):
526 self._next()
527 continue
528 break
529 self._expect_punct(']')
530 return ('list', items)
531 raise ValueError(f"Unexpected punctuation in expression: {value!r}")
532 if kind == 'name':
533 self._next()
534 lowered = value.lower()
535 if lowered == 'true':
536 return ('lit', True)
537 if lowered == 'false':
538 return ('lit', False)
539 if lowered in ('none', 'null'):
540 return ('lit', None)
541 if value in ('and', 'or', 'not'):
542 raise ValueError(f"Unexpected keyword in expression: {value!r}")
543 return ('ref', value)
544 raise ValueError(f"Unexpected token in expression: {value!r}")
547def _is_number(value: Any) -> bool:
548 return isinstance(value, (int, float)) and not isinstance(value, bool)
551def _add_values(left: Any, right: Any) -> Any:
552 """Binary `+`. Lists concatenate, numbers add, strings concatenate."""
553 if isinstance(left, list) and isinstance(right, list):
554 return left + right
555 if _is_number(left) and _is_number(right):
556 return left + right
557 if isinstance(left, str) and isinstance(right, str):
558 return left + right
559 raise ValueError(
560 f"Unsupported operands for '+': {type(left).__name__} + {type(right).__name__}"
561 )
564def _eval_comparison_rhs(node: tuple, ctx: dict, strict: bool) -> Any:
565 """Comparison right-hand side. A bare non-dotted reference compares as a
566 string literal (preserving `profile == PR`); everything else evaluates
567 normally (numbers, quoted strings, dotted refs, lists, grouped exprs)."""
568 if node[0] == 'ref' and '.' not in node[1]:
569 return node[1]
570 return _eval_ast(node, ctx, strict)
573def _eval_cmp(op: str, left_node: tuple, right_node: tuple, ctx: dict, strict: bool) -> bool:
574 left = _eval_ast(left_node, ctx, strict)
575 right = _eval_comparison_rhs(right_node, ctx, strict)
576 try:
577 if op == '==':
578 return left == right
579 if op == '!=':
580 return left != right
581 if op == '<=':
582 return left <= right
583 if op == '>=':
584 return left >= right
585 if op == '<':
586 return left < right
587 if op == '>':
588 return left > right
589 except TypeError:
590 return False
591 raise ValueError(f"Unknown comparison operator: {op!r}")
594def _eval_ast(node: tuple, ctx: dict, strict: bool) -> Any:
595 """Evaluate a parsed AST. `strict=True` raises on an unresolved reference
596 read (condition/exit_condition); `and`/`or` short-circuit left-to-right."""
597 kind = node[0]
598 if kind == 'lit':
599 return node[1]
600 if kind == 'ref':
601 return _dot_get_strict(ctx, node[1]) if strict else _dot_get(ctx, node[1])
602 if kind == 'list':
603 return [_eval_ast(item, ctx, strict) for item in node[1]]
604 if kind == 'neg':
605 return -_eval_ast(node[1], ctx, strict)
606 if kind == 'not':
607 return not bool(_eval_ast(node[1], ctx, strict))
608 if kind == 'cmp':
609 return _eval_cmp(node[1], node[2], node[3], ctx, strict)
610 if kind == 'binop':
611 op, left_node, right_node = node[1], node[2], node[3]
612 if op == 'and':
613 if not bool(_eval_ast(left_node, ctx, strict)):
614 return False
615 return bool(_eval_ast(right_node, ctx, strict))
616 if op == 'or':
617 if bool(_eval_ast(left_node, ctx, strict)):
618 return True
619 return bool(_eval_ast(right_node, ctx, strict))
620 if op == '+':
621 left = _eval_ast(left_node, ctx, strict)
622 right = _eval_ast(right_node, ctx, strict)
623 return _add_values(left, right)
624 raise ValueError(f"Unknown binary operator: {op!r}")
625 raise ValueError(f"Unknown expression node: {kind!r}")
628def _eval_expr(expr: str, ctx: dict, strict: bool = False) -> Any:
629 """Evaluate a template expression against context dict.
631 `strict` is used by condition evaluation: dot-path access to an unresolved
632 reference raises (ValueError) instead of silently resolving to falsy, and
633 and/or short-circuit left-to-right so a value is never read when the left
634 side already determines the result.
635 """
636 expr = expr.strip()
637 tokens = _tokenize_expr(expr)
638 if not tokens:
639 return _dot_get_strict(ctx, expr) if strict else _dot_get(ctx, expr)
640 return _eval_ast(_ExprParser(tokens).parse(), ctx, strict)
643def _resolve_template(value: Any, ctx: dict, strict: bool = False) -> Any:
644 """Recursively resolve {{ expr }} templates in strings / dicts / lists."""
645 if isinstance(value, str):
646 stripped = value.strip()
647 # Entire string is one template → replace with actual type (not string)
648 m = _TMPL_RE.fullmatch(stripped)
649 if m:
650 return _eval_expr(m.group(1), ctx, strict)
651 # Inline substitution → always returns string
652 def _replace(m: re.Match) -> str:
653 result = _eval_expr(m.group(1), ctx, strict)
654 return '' if result is None else str(result)
655 return _TMPL_RE.sub(_replace, value)
656 if isinstance(value, dict):
657 return {k: _resolve_template(v, ctx, strict) for k, v in value.items()}
658 if isinstance(value, list):
659 return [_resolve_template(v, ctx, strict) for v in value]
660 return value
663def _resolve_input(input_spec: Any, state: dict, ctx: dict) -> Any:
664 """
665 Resolve step input spec to a concrete value:
666 "{{ item }}" → template
667 "[host-ai-context-map-path]" → literal placeholder (verbatim pass-through)
668 "gather.doc_files" → dot-path into state
669 {"k": "a.b"} → dict with each value resolved
670 None → None
671 """
672 if input_spec is None:
673 return None
674 if isinstance(input_spec, str):
675 if '{{' in input_spec:
676 return _resolve_template(input_spec, ctx)
677 stripped = input_spec.lstrip()
678 # A bracket-prefixed literal (e.g. a host-ai-context-map-path
679 # placeholder) is NOT a dot-path reference into state — hand it to the
680 # builtin verbatim so it resolves the placeholder itself.
681 if stripped.startswith('['):
682 return input_spec
683 # Dot-path into state (no braces)
684 return _dot_get(state, stripped)
685 if isinstance(input_spec, dict):
686 return {k: _resolve_input(v, state, ctx) for k, v in input_spec.items()}
687 if isinstance(input_spec, list):
688 return [_resolve_input(v, state, ctx) for v in input_spec]
689 return input_spec
692def _eval_condition(cond: Any, ctx: dict) -> bool:
693 """Evaluate a step condition. None → True (always run).
695 Strict (short-circuit + hard error on unresolved reads) per R1: an
696 undefined or skipped-step field read without short-circuit protection
697 raises, never silently evaluates to falsy.
698 """
699 if cond is None:
700 return True
701 resolved = _resolve_template(cond, ctx, strict=True)
702 if isinstance(resolved, bool):
703 return resolved
704 if isinstance(resolved, str):
705 return resolved.lower() not in ('false', '0', '', 'none', 'null')
706 return bool(resolved)
709def _resolve_int(value: Any, ctx: dict, default: int) -> int:
710 v = _resolve_template(value, ctx) if isinstance(value, str) else value
711 try:
712 return int(v)
713 except (TypeError, ValueError):
714 return default
717# ---------------------------------------------------------------------------
718# ManifestLoader
719# ---------------------------------------------------------------------------
722def _parse_manifest_file(path: Path) -> list[dict]:
723 """Palauttaa kaikki apiVersion: agent-platform/v1 -dokumentit tiedostosta."""
724 try:
725 content = path.read_text(encoding="utf-8")
726 except OSError as exc:
727 log.warning("Cannot read %s: %s", path, exc)
728 return []
730 results: list[dict] = []
731 for raw in content.split("\n---"):
732 raw_clean = "\n".join(
733 line for line in raw.splitlines()
734 if not line.strip().startswith("#")
735 )
736 try:
737 doc = yaml.safe_load(raw_clean)
738 except yaml.YAMLError as exc:
739 log.warning("YAML parse error in %s: %s", path, exc)
740 continue
741 if isinstance(doc, dict) and doc.get("apiVersion") == "agent-platform/v1":
742 results.append(doc)
743 return results
746class ManifestLoader:
747 """Skannaa hakemistot ja lataa kaikki YAML-manifestit."""
749 def __init__(self, dirs: list[Path] | None = None) -> None:
750 self._dirs = dirs or MANIFEST_DIRS
752 def load_all(self) -> list[dict]:
753 manifests: list[dict] = []
754 for d in self._dirs:
755 if not d.exists():
756 log.debug("Manifest dir not found: %s", d)
757 continue
758 for path in sorted(d.glob("**/*.yaml")):
759 docs = _parse_manifest_file(path)
760 manifests.extend(docs)
761 if docs:
762 log.debug("Loaded %d manifest(s) from %s", len(docs), path)
763 return manifests
766# ---------------------------------------------------------------------------
767# ManifestRegistry
768# ---------------------------------------------------------------------------
770ResourceKey = tuple[str, str] # (kind, name)
773class ManifestRegistry:
774 """Indeksoi ladatut manifestit kind+name avaimella."""
776 def __init__(self, manifests: list[dict]) -> None:
777 self._store: dict[ResourceKey, dict] = {}
778 for doc in manifests:
779 kind = doc.get("kind", "")
780 name = (doc.get("metadata") or {}).get("name", "")
781 if kind and name:
782 key: ResourceKey = (kind, name)
783 if key in self._store:
784 log.warning("Duplicate manifest %s/%s — last one wins", kind, name)
785 self._store[key] = doc
787 def get(self, kind: str, name: str) -> dict | None:
788 return self._store.get((kind, name))
790 def all_of_kind(self, kind: str) -> list[dict]:
791 return [v for (k, _), v in self._store.items() if k == kind]
794# ---------------------------------------------------------------------------
795# AgentResolver
796# ---------------------------------------------------------------------------
798def _deep_merge(base: dict, override: dict) -> dict:
799 """Syvä merge: override voittaa konflikteissa."""
800 result = dict(base)
801 for k, v in override.items():
802 if isinstance(v, dict) and isinstance(result.get(k), dict):
803 result[k] = _deep_merge(result[k], v)
804 else:
805 result[k] = v
806 return result
809class AgentResolver:
810 """
811 Palauttaa fully-resolved agent spec:n instanssille.
812 Jos instanssilla on based_on, mergee Agent-base + AgentInstance-overridet.
813 """
815 def __init__(self, registry: ManifestRegistry) -> None:
816 self._registry = registry
817 self._cache: dict[str, dict] = {}
819 def resolve(self, instance_name: str) -> dict:
820 if instance_name in self._cache:
821 return self._cache[instance_name]
823 instance = self._registry.get("AgentInstance", instance_name)
824 if instance is None:
825 agent = self._registry.get("Agent", instance_name)
826 if agent is None:
827 raise KeyError(f"No Agent or AgentInstance named '{instance_name}'")
828 resolved = agent
829 else:
830 based_on = (instance.get("spec") or {}).get("based_on")
831 if based_on:
832 base_agent = self._registry.get("Agent", based_on)
833 if base_agent is None:
834 raise KeyError(
835 f"AgentInstance '{instance_name}' references unknown Agent '{based_on}'"
836 )
837 base_spec = base_agent.get("spec") or {}
838 inst_spec = {k: v for k, v in (instance.get("spec") or {}).items()
839 if k != "based_on"}
840 merged_spec = _deep_merge(base_spec, inst_spec)
841 else:
842 merged_spec = {k: v for k, v in (instance.get("spec") or {}).items()
843 if k != "based_on"}
845 resolved = {
846 "apiVersion": "agent-platform/v1",
847 "kind": "Agent",
848 "metadata": {"name": instance_name},
849 "spec": merged_spec,
850 }
852 self._cache[instance_name] = resolved
853 return resolved
856# ---------------------------------------------------------------------------
857# State type
858# ---------------------------------------------------------------------------
860PipelineState = dict[str, Any]
862# Engine-reserved channel key: the existing flat Core-managed JSON state lives
863# under this key when a pipeline declares spec.channels.
864_CORE = CORE_KEY
866# Flat-state keys the engine owns; channel names may not shadow them.
867_CORE_STATE_KEYS = frozenset({
868 "errors", "stats", "run_status", "project_root", "run_id",
869 "_stats", "step_context_metrics",
870})
872# Engine-local iteration keys the loop machinery injects into the read state.
873# They are engine plumbing, not user state, so a colliding channel key must
874# never shadow them.
875_ENGINE_LOCAL_ITERATION_KEYS = ("item", "_item", "_foreach_item", "loop")
878def _iteration_discriminator(state: dict) -> str | None:
879 """The engine-local iteration token for *state*, or None when not iterating.
881 Concurrent foreach/parallel items share a step id but differ in their
882 iteration key, so the synthesized call id uses this to stay distinct on the
883 in-cluster sidecar params file. The item is digested so the id stays a
884 bounded, filename-safe length.
885 """
886 for key in _ENGINE_LOCAL_ITERATION_KEYS:
887 if key in state:
888 digest = hashlib.sha1(repr(state[key]).encode("utf-8")).hexdigest()[:12]
889 return f"{key}-{digest}"
890 return None
894def _build_state_schema() -> Any:
895 """Build the channel-path TypedDict: a single Core-owned ``__core__`` field.
897 Channel context is Core-owned, so there are no per-channel native LangGraph
898 fields; the flat committed snapshot (core keys + channel keys) travels
899 node-to-node under ``__core__``. LangGraph stays the graph executor.
900 """
901 from typing import TypedDict
903 return TypedDict("PipelineState", {_CORE: dict[str, Any]})
906# Sentinel for optional kwargs distinguishing "not provided" from None
907_UNSET: Any = object()
909# ---------------------------------------------------------------------------
910# BuiltinExecutor
911# ---------------------------------------------------------------------------
913class BuiltinExecutor:
914 """
915 Ajaa builtin-komponentit.
916 Uusia komponentteja voi rekisteröidä register()-metodilla.
917 Nimet normalisoidaan: viiva → alaviiva (esim. 'list-dir-entries' == 'list_dir_entries').
918 """
920 # Lazy modules are imported on first use (they need httpx / heavier deps).
921 # Profile-specific builtins live here: the PR-profile github-pr-comment
922 # and the TICKET-profile ticket-publish (per-run local feature branch; push/PR disabled).
923 _LAZY_MODULES: tuple[str, ...] = ("_builtins_pr", "_builtins_ticket")
925 def __init__(self) -> None:
926 self._handlers: dict[str, Any] = {
927 "passthrough": self._passthrough,
928 # Engine-internal call-expansion scaffolding (ADR-0025 D4): these are
929 # synthesized at build time by expand_calls(), never authored in a
930 # manifest. The steps that generate them are removed from the flat
931 # runtime graph — only their body + these two op steps remain.
932 "__call_bind": self._call_bind,
933 "__call_emit": self._call_emit,
934 }
935 # Pipeline's LLM invoker (spec, step, state) → state, bound by
936 # GraphBuilder. Profile builtins that make LLM decisions
937 # (ticket-dor / ticket-review) read it via their register_all
938 # wrapper — the executor must have it set before the builtin
939 # executes (binding happens at build_workflow time).
940 self.invoke_agent_fn: Any | None = None
941 self._lazy_loaded = False
943 def register(self, name: str, fn: Any) -> None:
944 self._handlers[name] = fn
945 # Also register with normalised name (hyphen → underscore)
946 normalised = name.replace("-", "_")
947 if normalised != name:
948 self._handlers[normalised] = fn
950 def _load_lazy_modules(self) -> None:
951 if self._lazy_loaded:
952 return
953 self._lazy_loaded = True
954 import importlib
955 for module_name in self._LAZY_MODULES:
956 try:
957 mod = importlib.import_module(f".{module_name}", package="langgraph")
958 except ImportError:
959 continue
960 mod.register_all(self)
962 def has(self, name: str) -> bool:
963 """True when *name* resolves in the builtin register (lazy modules included)."""
964 self._load_lazy_modules()
965 return name in self._handlers or name.replace("-", "_") in self._handlers
967 def get(self, name: str) -> Any:
968 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
969 if fn is not None:
970 return fn
971 self._load_lazy_modules()
972 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
973 if fn is None:
974 raise ValueError(f"Unknown builtin component '{name}'")
975 return fn
977 def execute(self, component: str, step: dict, state: PipelineState) -> PipelineState:
978 fn = self.get(component)
979 return fn(step, state)
981 # --- default builtins ---
983 @staticmethod
984 def _passthrough(step: dict, state: PipelineState) -> PipelineState:
985 return state
987 @staticmethod
988 def _call_bind(step: dict, state: PipelineState) -> PipelineState:
989 """Call-scope entry (synthesized): bind the function call's params.
991 The caller's function params are resolved against the current state and
992 stored at ``state[<call_id>] = {"params": {...}}`` so the (rewritten)
993 function-body templates ``{{ <call_id>.params.<name> }}`` resolve. The
994 call's reserved ``__call_id`` key is stripped from the bound params.
995 """
996 call_id = (step.get("params") or {}).get("__call_id")
997 if not call_id:
998 raise ValueError("__call_bind missing __call_id (not a synthesized step)")
999 bound = {
1000 k: v for k, v in (step.get("params") or {}).items() if k != "__call_id"
1001 }
1002 return {**state, call_id: {"params": bound}}
1004 @staticmethod
1005 def _call_emit(step: dict, state: PipelineState) -> PipelineState:
1006 """Call-scope exit (synthesized): surface the function's outputs.
1008 Per the call step-result contract (00-process-contract r.70) the caller
1009 sees only ``state[<call_id>] = {outputs: {<name>: <source value>}}`` —
1010 the function body step results are invisible to the caller. Each output
1011 name resolves to the manifest interface ``outputs[].source`` dot-path.
1013 Optionality (ADR-0031): an output flagged ``optional: true`` is skipped
1014 when its source-path is unresolved; an output that is NOT optional must
1015 resolve or a RunHardFailure is raised (no silent None — ADR-0012).
1016 """
1017 params = step.get("params") or {}
1018 call_id = params.get("__call_id")
1019 if not call_id:
1020 raise ValueError("__call_emit missing __call_id (not a synthesized step)")
1021 outputs_spec = params.get("__outputs") or []
1022 outputs: dict[str, Any] = {}
1023 for spec in outputs_spec:
1024 if not isinstance(spec, dict):
1025 continue
1026 name = spec.get("name")
1027 source = spec.get("source")
1028 if name is None:
1029 continue
1030 optional = bool(spec.get("optional", False))
1031 if source is None:
1032 # No declared source-path: treat as unresolved against the
1033 # optionality gate (ADR-0012 — no silent None for required).
1034 if optional:
1035 continue
1036 raise RunHardFailure(
1037 "UNRESOLVED_OUTPUT",
1038 f"call '{call_id}' output '{name}' has no source "
1039 "(required output — no silent None, ADR-0012)",
1040 {"call_id": call_id, "name": name, "source": source},
1041 )
1042 try:
1043 value = _dot_get_strict(state, source)
1044 except ValueError:
1045 # Unresolved source-path: required → hard error; optional → skip.
1046 if optional:
1047 continue
1048 raise RunHardFailure(
1049 "UNRESOLVED_OUTPUT",
1050 f"call '{call_id}' output '{name}' unresolved: source "
1051 f"'{source}' not found in state (required output — "
1052 "no silent None, ADR-0012)",
1053 {"call_id": call_id, "name": name, "source": source},
1054 ) from None
1055 outputs[name] = value
1056 return {**state, call_id: {"outputs": outputs}}
1059_DEFAULT_BUILTINS = BuiltinExecutor()
1062def _register_domain_builtins(executor: BuiltinExecutor) -> None:
1063 """Rekisteröi v1-domain-builtinit (list-dir-entries, ticket-publisher, ...).
1065 Erotettu omaan moduliin (`_builtins.py`) — runner.py pysyy generic-engine,
1066 builtinit ovat generiisiä mekanismeja (ADR-0016/0017).
1067 """
1068 try:
1069 from langgraph import _builtins # type: ignore
1070 except ImportError:
1071 # Kun runner.py ajetaan ilman pakettirakennetta (esim. /app/langgraph)
1072 import importlib.util
1073 spec = importlib.util.spec_from_file_location(
1074 "_builtins", Path(__file__).parent / "_builtins.py"
1075 )
1076 if spec and spec.loader:
1077 _builtins = importlib.util.module_from_spec(spec)
1078 spec.loader.exec_module(_builtins) # type: ignore[union-attr]
1079 else:
1080 log.warning("Cannot load _builtins.py — domain builtins disabled")
1081 return
1082 _builtins.register_all(executor)
1083 _register_function_builtins(executor)
1086def _register_function_builtins(executor: BuiltinExecutor) -> None:
1087 """Rekisteröi function-engine builtinit (question-block-form).
1089 Erotettu omaan moduliin (`_builtins_functions.py`) — runner.py pysyy
1090 generic-engine, function-builtinit pysyvät puhtaina ja erillisinä.
1091 """
1092 try:
1093 from langgraph import _builtins_functions # type: ignore
1094 except ImportError:
1095 import importlib.util
1096 spec = importlib.util.spec_from_file_location(
1097 "_builtins_functions", Path(__file__).parent / "_builtins_functions.py"
1098 )
1099 if spec and spec.loader:
1100 _builtins_functions = importlib.util.module_from_spec(spec)
1101 spec.loader.exec_module(_builtins_functions) # type: ignore[union-attr]
1102 else:
1103 log.warning("Cannot load _builtins_functions.py — function builtins disabled")
1104 return
1105 _builtins_functions.register_all(executor)
1108_register_domain_builtins(_DEFAULT_BUILTINS)
1110# ---------------------------------------------------------------------------
1111# StepRunner
1112# ---------------------------------------------------------------------------
1114class StepRunner:
1115 """
1116 Ajaa yksittäisen step:n sen typin mukaan.
1118 run_context: dict joka lisätään template-kontekstiin. Esim:
1119 {"budget": {"max_parallel": 16, ...}, "config": {...}}
1120 """
1122 def __init__(
1123 self,
1124 resolver: AgentResolver,
1125 builtins: BuiltinExecutor | None = None,
1126 invoke_agent_fn: Any | None = None,
1127 run_context: dict | None = None,
1128 *,
1129 step_functions: StepFunctionExecutor | None = None,
1130 tracer: Any | None = None,
1131 ) -> None:
1132 self._resolver = resolver
1133 self._builtins = builtins or _DEFAULT_BUILTINS
1134 # Order matters: the hard-failure gate runs first (an error envelope
1135 # fail-fasts before any requested call), then the envelope's requested
1136 # deterministic-step calls execute before the result is stored/routed.
1137 self._invoke_agent = self._wrap_calls_gate(
1138 self._wrap_hard_failure_gate(
1139 invoke_agent_fn or self._dummy_invoke
1140 )
1141 )
1142 self._run_context: dict = run_context or {}
1143 self._step_functions = step_functions
1144 self._tracer = tracer
1145 self._channel_names: frozenset[str] = frozenset()
1147 def set_channel_routing(self, channel_names: frozenset[str]) -> None:
1148 """Attach the build-time routing context nested body steps route with."""
1149 self._channel_names = channel_names
1151 @staticmethod
1152 def _should_skip_llm(step: dict, ctx: dict) -> bool:
1153 """Truthiness of a step's `skip_llm_if` expression (ENG-12, Q4).
1155 Lenient (non-strict) evaluation: a missing/undefined field resolves to
1156 falsy, so the step behaves normally (LLM invoked). Only an explicit
1157 truthy expression skips the LLM.
1158 """
1159 raw = step.get("skip_llm_if")
1160 if raw is None:
1161 return False
1162 resolved = _resolve_template(raw, ctx)
1163 if isinstance(resolved, bool):
1164 return resolved
1165 if isinstance(resolved, str):
1166 return resolved.lower() not in ('false', '0', '', 'none', 'null')
1167 return bool(resolved)
1169 @staticmethod
1170 def _skip_llm_passthrough(
1171 step: dict,
1172 state: PipelineState,
1173 ctx: dict,
1174 read_state: PipelineState,
1175 ) -> PipelineState:
1176 """Deterministic passthrough of a step's resolved input (no LLM).
1178 `state[step_id]` becomes the resolved `input` value (the parallel
1179 fan-out envelope `{results: [...]}` when the input resolves to a list).
1180 ``read_state`` is the working-set overlay the input resolves against;
1181 the returned state is still built from the raw ``state``.
1182 """
1183 step_id = step.get("id", "?")
1184 input_spec = step.get("input")
1185 resolved = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
1186 if isinstance(resolved, list):
1187 log.debug("[step] %s skip_llm_if passthrough (fanout, no LLM)", step_id)
1188 return {**state, step_id: {"results": resolved}}
1189 log.debug("[step] %s skip_llm_if passthrough (no LLM)", step_id)
1190 return {**state, step_id: resolved}
1192 @staticmethod
1193 def _parallel_fanout_items(
1194 step: dict, state: PipelineState, ctx: dict
1195 ) -> list | None:
1196 """The resolved fan-out item list for a parallel step, else None.
1198 A parallel step whose resolved input is a list invokes the agent once
1199 per item and stores a ``{"results": [...]}`` envelope instead of a
1200 single agent output.
1201 """
1202 if step.get("type") != "parallel":
1203 return None
1204 input_spec = step.get("input")
1205 if input_spec is None:
1206 return None
1207 input_val = _resolve_input(input_spec, state, ctx)
1208 return input_val if isinstance(input_val, list) else None
1210 def ran_agent(self, step: dict, state: PipelineState) -> bool:
1211 """Whether a step's run invokes its agent.
1213 False for agent-less steps, a ``skip_llm_if`` passthrough, and a
1214 parallel fan-out (per-item agent calls storing a result envelope).
1215 Channel routing is restricted to actual agent outputs, so these paths
1216 are excluded from inverse projection.
1217 """
1218 if self.resolve_agent(step) is None:
1219 return False
1220 ctx = self._make_ctx(state)
1221 if self._should_skip_llm(step, ctx):
1222 return False
1223 return self._parallel_fanout_items(step, state, ctx) is None
1225 def _make_ctx(self, state: dict, **extras: Any) -> dict:
1226 """Rakentaa template-kontekstin: state + run_context + extras."""
1227 ctx: dict = {}
1228 ctx.update(state)
1229 ctx.update(self._run_context)
1230 ctx.update(extras)
1231 return ctx
1233 @staticmethod
1234 def _read_state(
1235 state: PipelineState, working_set: ChannelWorkingSet | None
1236 ) -> PipelineState:
1237 """Overlay the invocation working set's read view onto the local state.
1239 The working set carries the committed snapshot plus this invocation's
1240 pending writes; the local state contributes keys the snapshot does not
1241 hold (item context, body step results). With no working set the raw
1242 state is the read view, preserving legacy behaviour exactly. The
1243 engine-local iteration keys are re-asserted after the overlay so a
1244 declared channel with the same name cannot shadow them.
1245 """
1246 if working_set is None:
1247 return state
1248 resolved = {**state, **working_set.read_view()}
1249 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1250 if key in state:
1251 resolved[key] = state[key]
1252 return resolved
1254 def _run_body_step(
1255 self,
1256 step: dict,
1257 state: PipelineState,
1258 working_set: ChannelWorkingSet | None,
1259 ) -> PipelineState:
1260 """Run one nested body step and stage its declared channel writes.
1262 The two-argument call preserves the legacy recursion signature for the
1263 no-channels path (and for callers that monkey-patch ``run_step``).
1264 """
1265 if working_set is None:
1266 return self.run_step(step, state)
1267 result = self.run_step(step, state, working_set)
1268 self._stage_body_channel_writes(step, state, result, working_set)
1269 return result
1271 def _stage_body_channel_writes(
1272 self,
1273 step: dict,
1274 pre_state: PipelineState,
1275 result_state: PipelineState,
1276 working_set: ChannelWorkingSet,
1277 ) -> None:
1278 """Stage a body step's declared channel writes into the working set.
1280 The router early-outs for agent-less steps (builtins, call bind/emit);
1281 a step carrying an agent routes its declared writes.
1282 """
1283 if not self._channel_names:
1284 return
1285 writes = _route_declared_channels(
1286 step,
1287 self._read_state(pre_state, working_set),
1288 result_state,
1289 self,
1290 self._channel_names,
1291 )
1292 for name, value in writes.items():
1293 working_set.stage(name, value)
1295 def _resolve_step_runtime(
1296 self,
1297 step: dict,
1298 state: dict,
1299 ctx: dict,
1300 *,
1301 input_override: Any = _UNSET,
1302 extra_params: dict | None = None,
1303 ) -> dict:
1304 """
1305 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
1306 templaattien osalta state + run_context -kontekstia vasten.
1308 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
1309 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
1310 """
1311 resolved: dict = dict(step)
1313 if input_override is not _UNSET:
1314 resolved["input"] = input_override
1315 else:
1316 input_spec = step.get("input")
1317 if input_spec is not None:
1318 resolved["input"] = _resolve_input(input_spec, state, ctx)
1320 params_spec = step.get("params")
1321 if params_spec or extra_params:
1322 base = _resolve_template(params_spec, ctx) if params_spec else {}
1323 if not isinstance(base, dict):
1324 base = {}
1325 if extra_params:
1326 base = {**base, **extra_params}
1327 resolved["params"] = base
1329 return resolved
1331 def _attach_communication_context(
1332 self, resolved_step: dict, read_state: PipelineState
1333 ) -> dict:
1334 """Attach a step's projected COMMUNICATION context when declared.
1336 The projection is resolved from the invocation read overlay (the
1337 committed snapshot plus this invocation's pending writes) and carries
1338 only the step's declared input channels — never whole state. A step
1339 without a read declaration is returned unchanged.
1340 """
1341 communication = resolve_communication_context(resolved_step, read_state)
1342 if not communication:
1343 return resolved_step
1344 return {**resolved_step, COMMUNICATION_CONTEXT_KEY: communication}
1346 def _attach_usage_instructions(self, resolved_step: dict) -> dict:
1347 """Attach the resolved usage-instruction entries for a step's `allow`.
1349 The whitelist is implicit: the instruction block is derived from the
1350 step's `allow` entries, resolved builtin-first, and carried to the
1351 invoker (ADR-0036 D1 element 5). A step without an `allow` whitelist
1352 is returned unchanged.
1353 """
1354 entries = resolve_usage_instructions(
1355 resolved_step.get("allow"), self._builtins, self._step_functions
1356 )
1357 if not entries:
1358 return resolved_step
1359 return {**resolved_step, USAGE_INSTRUCTIONS_KEY: entries}
1361 @staticmethod
1362 def _dummy_invoke(
1363 resolved_spec: dict, step: dict, state: PipelineState
1364 ) -> PipelineState:
1365 """Placeholder — ei kutsu oikeaa LLM:ää."""
1366 step_id = step.get("id", "?")
1367 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
1368 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
1370 @staticmethod
1371 def _merge_errors(base: list, additions: list) -> list:
1372 """Return a new list containing all items from *base* plus any items from
1373 *additions* that are not already present. Order is preserved and no
1374 in-place mutation occurs — safe to call from any context."""
1375 if not additions:
1376 return base
1377 merged = list(base)
1378 for err in additions:
1379 if err not in merged:
1380 merged.append(err)
1381 return merged
1383 @staticmethod
1384 def _wrap_hard_failure_gate(invoke_fn: Any) -> Any:
1385 """Kietoo invokerin: jokaiseen agent-step-tulokseen sovelletaan
1386 kova-virhe-portti (RunHardFailure details.error_code-perusteella).
1387 Suppein funnel joka näkee KAIKKI agent-tulokset: sequential,
1388 parallel (single + fanout), foreach (body + agent-polku) ja while."""
1389 def invoke_with_gate(
1390 resolved_spec: dict, step: dict, state: PipelineState
1391 ) -> PipelineState:
1392 result = invoke_fn(resolved_spec, step, state)
1393 step_id = step.get("id", "?")
1394 StepRunner._raise_on_hard_error(result.get(step_id), step_id)
1395 return result
1396 return invoke_with_gate
1398 def _wrap_calls_gate(self, invoke_fn: Any) -> Any:
1399 """Kietoo invokerin: agentin pyytämät deterministic-step kutsut
1400 (envelopen `calls`) suoritetaan, kun hard-failure-portti on päästänyt
1401 tuloksen läpi ja ennen kuin step-tulos tallennetaan/reititetään."""
1402 def invoke_with_calls(
1403 resolved_spec: dict, step: dict, state: PipelineState
1404 ) -> PipelineState:
1405 result = invoke_fn(resolved_spec, step, state)
1406 self._execute_requested_calls(step, result, state)
1407 return result
1408 return invoke_with_calls
1410 def _execute_requested_calls(
1411 self, step: dict, result: PipelineState, state: PipelineState
1412 ) -> None:
1413 """Suorita agentin envelopessa pyytämät deterministic-step kutsut.
1415 No-op ellei tulos ole envelope, jossa on ei-tyhjä `calls`-lista. Kutsut
1416 suoritetaan järjestyksessä builtin-then-StepFunction-polkua pitkin;
1417 onnistunut kutsu on side-effect-only (palautettu state hylätään).
1418 """
1419 step_id = step.get("id", "?")
1420 envelope = result.get(step_id) if isinstance(result, dict) else None
1421 calls = envelope.get("calls") if isinstance(envelope, dict) else None
1422 if not calls:
1423 return
1424 allow = step.get("allow") or []
1425 for index, call in enumerate(calls):
1426 self._execute_requested_call(step_id, allow, index, call, state)
1428 def _execute_requested_call(
1429 self,
1430 step_id: str,
1431 allow: list,
1432 index: int,
1433 call: Any,
1434 state: PipelineState,
1435 ) -> None:
1436 """Validoi ja suorita yksi pyydetty kutsu, fail-fast virhekoodeilla."""
1437 name = call.get("step") if isinstance(call, dict) else None
1438 if not isinstance(name, str) or name not in allow:
1439 raise RunHardFailure(
1440 "CALL_NOT_ALLOWED",
1441 f"step '{step_id}' requested deterministic step {name!r} which "
1442 "is not in its allow whitelist",
1443 {"step_id": step_id, "requested_step": name, "allow": list(allow)},
1444 )
1445 params = self._call_params(call, step_id, name)
1446 self._validate_call_params(name, params, step_id)
1447 call_step = self._synthesize_call_step(step_id, index, name, params, state)
1448 try:
1449 self._dispatch_call_step(name, call_step, state)
1450 except RunHardFailure as exc:
1451 if exc.code == "BUILTIN_NAME_COLLISION":
1452 raise
1453 raise RunHardFailure(
1454 "CALL_FAILED",
1455 f"step '{step_id}' call to '{name}' failed: {exc.message}",
1456 {
1457 "step_id": step_id,
1458 "requested_step": name,
1459 "error_code": exc.code,
1460 },
1461 ) from exc
1462 except Exception as exc: # noqa: BLE001
1463 raise RunHardFailure(
1464 "CALL_FAILED",
1465 f"step '{step_id}' call to '{name}' failed: {exc}",
1466 {"step_id": step_id, "requested_step": name},
1467 ) from exc
1469 @staticmethod
1470 def _call_params(call: dict, step_id: str, name: str) -> dict:
1471 """The call's params object ({} when absent), else CALL_INVALID_PARAMS."""
1472 params = call.get("params")
1473 if params is None:
1474 return {}
1475 if not isinstance(params, dict):
1476 raise RunHardFailure(
1477 "CALL_INVALID_PARAMS",
1478 f"step '{step_id}' call to '{name}' has non-object params",
1479 {"step_id": step_id, "requested_step": name},
1480 )
1481 return params
1483 def _validate_call_params(self, name: str, params: dict, step_id: str) -> None:
1484 """Validoi StepFunction-kutsun paramsit interfacea vasten.
1486 Builtineilla ei ole formaalia param-interfacea — niiden params-objekti
1487 on jo strukturaalisesti validoitu envelopessa. Säännöllinen
1488 StepFunction-suoritus säilyttää nykyisen käytöksensä; tämä portti
1489 koskee vain calls-polkua.
1490 """
1491 if self._builtins.has(name) or self._step_functions is None:
1492 return
1493 declared = self._step_functions.params_for(name)
1494 if declared is None:
1495 return
1496 declared_names = {entry["name"] for entry in declared}
1497 missing = [
1498 entry["name"]
1499 for entry in declared
1500 if entry.get("required") and entry["name"] not in params
1501 ]
1502 unknown = [key for key in params if key not in declared_names]
1503 if not missing and not unknown:
1504 return
1505 raise RunHardFailure(
1506 "CALL_INVALID_PARAMS",
1507 f"step '{step_id}' call to '{name}' has invalid params "
1508 f"(missing required: {missing}; undeclared: {unknown})",
1509 {
1510 "step_id": step_id,
1511 "requested_step": name,
1512 "missing": missing,
1513 "unknown": unknown,
1514 },
1515 )
1517 @staticmethod
1518 def _synthesize_call_step(
1519 parent_step_id: str,
1520 index: int,
1521 name: str,
1522 params: dict,
1523 state: PipelineState,
1524 ) -> dict:
1525 """Rakentaa deterministisen synteettisen stepin yhdelle kutsulle.
1527 The id is unique per invocation: the engine-local iteration
1528 discriminator is included when present so concurrent foreach/parallel
1529 items get distinct ids, and a process-unique token guarantees
1530 uniqueness otherwise (and against equal discriminators). The id is
1531 not persisted — calls are side-effect-only.
1532 """
1533 discriminator = _iteration_discriminator(state)
1534 token = uuid.uuid4().hex
1535 suffix = f"{discriminator}.{token}" if discriminator else token
1536 return {
1537 "id": f"{parent_step_id}.call.{index}.{name}.{suffix}",
1538 "type": "builtin",
1539 "component": name,
1540 "params": params,
1541 }
1543 def _dispatch_call_step(
1544 self, name: str, step: dict, state: PipelineState
1545 ) -> None:
1546 """Suorita yksi deterministic-step builtin-then-StepFunction-järjestyksessä.
1548 Peilaa `run_step`-builtin-haaran resolutiöjärjestystä (ADR-0037 D2):
1549 builtin-rekisteri voittaa, ja nimi molemmissa rekistereissä on kova
1550 virhe. Palautettu state-snapshot hylätään — kutsu on side-effect-only.
1551 """
1552 is_builtin = self._builtins.has(name)
1553 is_step_function = (
1554 self._step_functions is not None and self._step_functions.has(name)
1555 )
1556 if is_builtin and is_step_function:
1557 raise RunHardFailure(
1558 "BUILTIN_NAME_COLLISION",
1559 f"component '{name}' is present in both the builtin register "
1560 "and the StepFunction registry (ADR-0037 — fail-fast)",
1561 {"component": name},
1562 )
1563 if is_step_function:
1564 self._step_functions.execute(name, step, state)
1565 return
1566 builtin_state = self._builtins.execute(name, step, state)
1567 self._raise_on_hard_error(builtin_state.get(step["id"]), step["id"])
1569 @staticmethod
1570 def _is_hard_fail(details: dict) -> bool:
1571 """Kova virhe vain kun: error_code on HARD_ERROR_CODESissa
1572 (TRUNCATION_ERROR, TOOL_FAILURE, AGENT_ERROR) tai VALIDATION_ERROR
1573 kantaa kontekstiylityksen signatuurin (assembled_tokens +
1574 context_budget — _llm._context_overflow_result).
1575 Generic-VALIDATION_ERROR ei kaada ajoa. status:error-envelope kaatuu
1576 _raise_on_error_envelope-polulla ennen tätä tarkistusta."""
1577 code = details.get("error_code")
1578 if code in RunHardFailure.HARD_ERROR_CODES:
1579 return True
1580 if code == "VALIDATION_ERROR":
1581 return {"assembled_tokens", "context_budget"} <= details.keys()
1582 return False
1584 @staticmethod
1585 def _raise_on_hard_error(step_result: Any, step_id: str) -> None:
1586 """Nosta RunHardFailure jos step-tulos kantaa kovan virheen
1587 (_is_hard_fail) tai on status:error-envelope; stepin details
1588 kulkeutuu poikkeuksen mukana."""
1589 if not isinstance(step_result, dict):
1590 return
1591 if is_error_envelope(step_result):
1592 StepRunner._raise_on_error_envelope(step_result, step_id)
1593 return
1594 details = step_result.get("details")
1595 if not isinstance(details, dict):
1596 return
1597 if not StepRunner._is_hard_fail(details):
1598 return
1599 code = details.get("error_code")
1600 err = step_result.get("error", "")
1601 message = err if isinstance(err, str) and err else f"{code}: step '{step_id}' hard failure"
1602 raise RunHardFailure(code, message, details)
1604 @staticmethod
1605 def _raise_on_error_envelope(step_result: dict, step_id: str) -> None:
1606 """Nosta RunHardFailure status:error-envelopesta; envelopen error
1607 (dict) kulkeutuu details-attribuuttina, muuten minimi details."""
1608 envelope_error = step_result.get("error")
1609 if isinstance(envelope_error, dict):
1610 details = envelope_error
1611 message = envelope_error.get("message")
1612 else:
1613 details = {"error_code": ENVELOPE_ERROR_CODE, "step_id": step_id}
1614 message = None
1615 if not isinstance(message, str) or not message:
1616 message = f"{ENVELOPE_ERROR_CODE}: step '{step_id}' envelope error"
1617 raise RunHardFailure(ENVELOPE_ERROR_CODE, message, details)
1619 @staticmethod
1620 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
1621 """
1622 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
1623 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
1624 epäonnistuneeksi eikä hiljennä virhettä.
1626 Kaikki virhetyypit propagoidaan:
1627 - budget:max_llm_calls / budget:run_timeout → budjettiloki
1628 - parse_error:*, timeout:*, ja muut → error-loki
1629 """
1630 step_result = result.get(step_id)
1631 if not isinstance(step_result, dict):
1632 return result
1633 err = step_result.get("error", "")
1634 if not isinstance(err, str) or not err:
1635 return result
1636 errors = list(result.get("errors") or [])
1637 message = f"{step_id}: {err}"
1638 if message not in errors:
1639 errors.append(message)
1640 if err.startswith("budget:"):
1641 log.warning("[budget] %s", message)
1642 else:
1643 log.error("[step-error] %s", message)
1644 return {**result, "errors": errors}
1646 def resolve_agent(self, step: dict) -> dict | None:
1647 """Resolve the agent/instance a step runs, or None for agent-less steps."""
1648 agent_name = step.get("agent")
1649 if not agent_name:
1650 return None
1651 return self._resolver.resolve(agent_name)
1653 def run_step(
1654 self,
1655 step: dict,
1656 state: PipelineState,
1657 working_set: ChannelWorkingSet | None = None,
1658 ) -> PipelineState:
1659 step_id = step.get("id", "?")
1660 read_state = self._read_state(state, working_set)
1661 ctx = self._make_ctx(read_state)
1662 step_type = step.get("type", "sequential")
1664 # Evaluate condition — if false, skip step. A `while` whose entry
1665 # condition is false writes a first-class skip result (r.68) instead of
1666 # returning state unchanged, so `_run_while` owns its skip path.
1667 cond = step.get("condition")
1668 if cond is not None and not _eval_condition(cond, ctx) and step_type != "while":
1669 log.debug("[step] %s skipped (condition false)", step_id)
1670 return state
1672 if step_type == "builtin":
1673 component = step.get("component", "passthrough")
1674 log.debug("[step] %s builtin:%s", step_id, component)
1675 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
1676 # template-arvot (esim. "{{ budget.max_parallel }}")
1677 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1678 # ADR-0037 D2: resolve the component against the platform builtin
1679 # register FIRST, then the kind: StepFunction registry. A name in
1680 # both is a hard error (fail-fast).
1681 is_builtin = self._builtins.has(component)
1682 is_step_function = (
1683 self._step_functions is not None
1684 and self._step_functions.has(component)
1685 )
1686 if is_builtin and is_step_function:
1687 raise RunHardFailure(
1688 "BUILTIN_NAME_COLLISION",
1689 f"component '{component}' is present in both the builtin "
1690 "register and the StepFunction registry (ADR-0037 — "
1691 "fail-fast)",
1692 {"component": component},
1693 )
1694 if is_step_function:
1695 return self._step_functions.execute(component, resolved_step, state)
1696 builtin_state = self._builtins.execute(component, resolved_step, state)
1697 self._raise_on_hard_error(builtin_state.get(step_id), step_id)
1698 return builtin_state
1700 # skip_llm_if (ENG-12, Q4): when the step's skip_llm_if evaluates truthy,
1701 # do NOT invoke the LLM (no inference budget consumed) and set the step
1702 # result to a deterministic passthrough of the step's resolved input.
1703 if (
1704 step_type in ("sequential", "parallel")
1705 and self._should_skip_llm(step, ctx)
1706 ):
1707 return self._skip_llm_passthrough(step, state, ctx, read_state)
1709 if step_type == "parallel":
1710 # Fan-out: if input resolves to a list, call agent once per item
1711 items = self._parallel_fanout_items(step, read_state, ctx)
1712 if items is not None:
1713 return self._run_parallel_fanout(
1714 step, state, items, ctx, working_set
1715 )
1716 # Fallback: single agent call
1717 agent_name = step.get("agent")
1718 if not agent_name:
1719 log.warning("[step] %s has no agent — skipping", step_id)
1720 return state
1721 resolved = self._resolver.resolve(agent_name)
1722 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
1723 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1724 resolved_step = self._attach_communication_context(
1725 resolved_step, read_state
1726 )
1727 resolved_step = self._attach_usage_instructions(resolved_step)
1728 result = self._invoke_agent(resolved, resolved_step, state)
1729 return self._propagate_step_errors(result, step_id)
1731 if step_type == "sequential":
1732 agent_name = step.get("agent")
1733 if not agent_name:
1734 log.warning("[step] %s has no agent — skipping", step_id)
1735 return state
1736 resolved = self._resolver.resolve(agent_name)
1737 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
1738 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1739 resolved_step = self._attach_communication_context(
1740 resolved_step, read_state
1741 )
1742 resolved_step = self._attach_usage_instructions(resolved_step)
1743 result = self._invoke_agent(resolved, resolved_step, state)
1744 return self._propagate_step_errors(result, step_id)
1746 if step_type == "foreach":
1747 return self._run_foreach(step, state, working_set)
1749 if step_type == "while":
1750 return self._run_while(step, state, working_set)
1752 if step_type == "call":
1753 # Build-time expansion (ADR-0025 D4) is a later phase; a call step
1754 # reaching the runtime is wiring that is not yet expanded.
1755 raise ValueError(
1756 f"call step '{step_id}' reached runtime without build-time "
1757 "expansion (ADR-0025) — not yet supported"
1758 )
1760 if step_type == "spawn":
1761 return self._run_spawn(step, state, working_set)
1763 raise ValueError(f"Unknown step type '{step_type}' in step '{step_id}'")
1765 def _run_parallel_fanout(
1766 self,
1767 step: dict,
1768 state: PipelineState,
1769 items: list,
1770 ctx: dict,
1771 working_set: ChannelWorkingSet | None = None,
1772 ) -> PipelineState:
1773 """
1774 Fan-out: call agent once per item, concurrently.
1775 Results stored as state[step_id] = {"results": [item_result, ...]}.
1776 Each item is available as state["item"] inside the invoke call.
1777 """
1778 step_id = step.get("id", "?")
1779 agent_name = step.get("agent")
1780 if not agent_name:
1781 log.warning("[step] %s parallel fanout: no agent", step_id)
1782 return {**state, step_id: {"results": []}}
1784 resolved = self._resolver.resolve(agent_name)
1786 # Apply max_items cap
1787 max_items_raw = step.get("max_items")
1788 if max_items_raw is not None:
1789 max_items = _resolve_int(max_items_raw, ctx, len(items))
1790 items = items[:max_items]
1792 # Concurrency limit
1793 max_conc_raw = step.get("max_concurrent")
1794 if max_conc_raw is not None:
1795 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
1796 else:
1797 max_conc = len(items) or 1
1798 max_conc = max(1, min(max_conc, len(items) or 1))
1800 log.debug(
1801 "[step] %s parallel fanout: %d items, %d concurrent",
1802 step_id, len(items), max_conc,
1803 )
1805 read_state = self._read_state(state, working_set)
1807 def call_one(item: Any) -> tuple[Any, dict | None]:
1808 item_state = {**read_state, "_item": item, "item": item}
1809 item_ctx = self._make_ctx(item_state)
1810 resolved_step = self._resolve_step_runtime(
1811 step, item_state, item_ctx,
1812 input_override=item,
1813 )
1814 resolved_step = self._attach_communication_context(
1815 resolved_step, item_state
1816 )
1817 resolved_step = self._attach_usage_instructions(resolved_step)
1818 result_state = self._invoke_agent(resolved, resolved_step, item_state)
1819 return result_state.get(step_id), result_state.get("step_context_metrics")
1821 # Pre-allocate results to preserve input order regardless of completion
1822 # order (ENG-07), matching the concurrent foreach body-polku.
1823 results: list = [None] * len(items)
1824 item_metrics: dict = {}
1825 with ThreadPoolExecutor(max_workers=max_conc) as pool:
1826 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
1827 for future in as_completed(futures):
1828 idx = futures[future]
1829 try:
1830 item_result, item_metrics_map = future.result()
1831 results[idx] = item_result
1832 if isinstance(item_metrics_map, dict):
1833 item_metrics.update(item_metrics_map)
1834 except SkillLoadError:
1835 raise
1836 except RunHardFailure:
1837 pool.shutdown(wait=False, cancel_futures=True)
1838 raise
1839 except Exception as exc: # noqa: BLE001
1840 log.error("[step] %s item failed: %s", step_id, exc)
1841 results[idx] = {"error": str(exc)}
1843 self._stage_fanout_channel_writes(step, results, working_set)
1845 final: PipelineState = {**state, step_id: {"results": results}}
1846 if item_metrics:
1847 final = {
1848 **final,
1849 "step_context_metrics": _merge_metrics(
1850 state.get("step_context_metrics"), item_metrics
1851 ),
1852 }
1853 return final
1855 def _stage_fanout_channel_writes(
1856 self,
1857 step: dict,
1858 item_outputs: list,
1859 working_set: ChannelWorkingSet | None,
1860 ) -> None:
1861 """Stage each fan-out item's declared channel writes in item order.
1863 Runs on the main thread after the item pool completes, so the shared
1864 working set is never mutated from a worker thread. The fan-out
1865 envelope itself is never routed; only each item's own agent output is.
1866 """
1867 if working_set is None or not self._channel_names:
1868 return
1869 for output in item_outputs:
1870 writes = _declared_output_channel_writes(
1871 step, output, self._channel_names
1872 )
1873 for name, value in writes.items():
1874 working_set.stage(name, value)
1876 def _run_foreach(
1877 self,
1878 step: dict,
1879 state: PipelineState,
1880 working_set: ChannelWorkingSet | None = None,
1881 ) -> PipelineState:
1882 """
1883 Foreach: iterate over input list, run body steps per item.
1884 Input from step.input (dot-path or template) or step.over (state key).
1886 When max_concurrent > 1, body iterations run concurrently via
1887 ThreadPoolExecutor — each item gets its own worker thread. The
1888 max_concurrent value sets the thread pool size (mirrors parallel
1889 fan-out semantics). Results list preserves original item order.
1891 max_output budget in concurrent mode: a stop flag prevents new workers
1892 from starting after the budget is exhausted. Workers already executing
1893 when the flag is set will complete naturally — at most max_concurrent
1894 extra items may run past the limit. This is accepted behaviour; the
1895 sequential early-exit guarantee cannot be reproduced without sacrificing
1896 true concurrency.
1898 The invocation working set is threaded into every body step so reads
1899 resolve over the committed snapshot plus this invocation's pending
1900 writes; the returned state is still built from the raw ``state``.
1901 """
1902 step_id = step.get("id", "?")
1903 read_state = self._read_state(state, working_set)
1904 ctx = self._make_ctx(read_state)
1906 # Determine items
1907 input_spec = step.get("input")
1908 over_key = step.get("over")
1910 if input_spec is not None:
1911 items: list = _resolve_input(input_spec, read_state, ctx) or []
1912 if not isinstance(items, list):
1913 items = [items] if items is not None else []
1914 elif over_key:
1915 items = read_state.get(over_key, [None])
1916 if not isinstance(items, list):
1917 items = [items]
1918 else:
1919 items = [None]
1921 # max_iterations cap
1922 max_iter_raw = step.get("max_iterations")
1923 if max_iter_raw is not None:
1924 max_iter = _resolve_int(max_iter_raw, ctx, len(items))
1925 items = items[:max_iter]
1927 # max_output cap — stop once N iterations have produced a successful
1928 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
1929 max_output_raw = step.get("max_output")
1930 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
1932 # max_concurrent — when > 1, enables parallel body execution
1933 max_conc_raw = step.get("max_concurrent")
1934 if max_conc_raw is not None:
1935 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
1936 max_conc = max(1, min(max_conc, len(items) or 1))
1937 else:
1938 max_conc = 1
1940 body = step.get("body") or []
1941 agent_name = step.get("agent")
1943 if max_conc <= 1:
1944 return self._run_foreach_sequential(
1945 step_id, step, state, items, body, agent_name, max_output,
1946 working_set,
1947 )
1948 return self._run_foreach_concurrent(
1949 step_id, step, state, items, body, agent_name, max_output, max_conc,
1950 working_set,
1951 )
1953 @staticmethod
1954 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
1955 """body-polku per-iteration result (r.67): the declared body-output
1956 step's result; else the iteration's `publish` result; else None."""
1957 if declared_output:
1958 return iter_state.get(declared_output)
1959 return iter_state.get("publish")
1961 @staticmethod
1962 def _body_item_passed(item: Any) -> bool:
1963 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
1964 `passed` bool; an item lacking such a key is not counted as failed."""
1965 if isinstance(item, dict):
1966 verdict = item.get("verdict")
1967 if verdict == "FAIL":
1968 return False
1969 if verdict == "PASS":
1970 return True
1971 passed = item.get("passed")
1972 if isinstance(passed, bool):
1973 return passed
1974 return True
1976 @staticmethod
1977 def _aggregate_body_results(item_results: list) -> dict:
1978 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
1979 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
1980 return {
1981 "results": item_results,
1982 "passed": not failing,
1983 "failing": failing,
1984 "failed": bool(failing),
1985 }
1987 def _run_foreach_sequential(
1988 self,
1989 step_id: str,
1990 step: dict,
1991 state: PipelineState,
1992 items: list,
1993 body: list,
1994 agent_name: str | None,
1995 max_output: int,
1996 working_set: ChannelWorkingSet | None = None,
1997 ) -> PipelineState:
1998 """Sequential foreach — default path when max_concurrent is absent or 1."""
1999 body_polku = bool(body)
2000 declared_output = step.get("output")
2001 results: list = []
2002 produced_outputs = 0
2003 accumulated_errors: list = list(state.get("errors") or [])
2004 accumulated_metrics: dict = {}
2006 for item in items:
2007 if max_output and produced_outputs >= max_output:
2008 log.info("[foreach] %s max_output=%d reached — stopping", step_id, max_output)
2009 break
2010 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2012 if body_polku:
2013 for sub_step in body:
2014 iter_state = self._run_body_step(sub_step, iter_state, working_set)
2015 # Propagate any budget errors from this iteration to accumulated list
2016 accumulated_errors = self._merge_errors(
2017 accumulated_errors, iter_state.get("errors") or []
2018 )
2019 publish_out = iter_state.get("publish")
2020 results.append(self._body_item_result(iter_state, declared_output))
2021 if isinstance(publish_out, dict) and publish_out.get("path"):
2022 produced_outputs += 1
2023 elif agent_name:
2024 resolved = self._resolver.resolve(agent_name)
2025 iter_read_state = self._read_state(iter_state, working_set)
2026 iter_ctx = self._make_ctx(iter_read_state)
2027 resolved_step = self._resolve_step_runtime(
2028 step, iter_read_state, iter_ctx
2029 )
2030 resolved_step = self._attach_communication_context(
2031 resolved_step, iter_read_state
2032 )
2033 resolved_step = self._attach_usage_instructions(resolved_step)
2034 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2035 results.append(iter_state.get(step_id))
2036 item_metrics_map = iter_state.get("step_context_metrics")
2037 if isinstance(item_metrics_map, dict):
2038 accumulated_metrics.update(item_metrics_map)
2040 step_result: Any = self._aggregate_body_results(results) if body_polku else results
2041 final: PipelineState = {**state, step_id: step_result}
2042 if accumulated_metrics:
2043 final = {
2044 **final,
2045 "step_context_metrics": _merge_metrics(
2046 state.get("step_context_metrics"), accumulated_metrics
2047 ),
2048 }
2049 if accumulated_errors != list(state.get("errors") or []):
2050 final = {**final, "errors": accumulated_errors}
2051 return final
2053 @staticmethod
2054 def _merge_item_working_sets(
2055 working_set: ChannelWorkingSet | None,
2056 item_working_sets: list[ChannelWorkingSet | None],
2057 ) -> None:
2058 """Stage each concurrent item's writes onto the shared set in item order."""
2059 if working_set is None:
2060 return
2061 for item_working_set in item_working_sets:
2062 if item_working_set is None:
2063 continue
2064 for name, value in item_working_set.pending_writes():
2065 working_set.stage(name, value)
2067 def _run_foreach_concurrent(
2068 self,
2069 step_id: str,
2070 step: dict,
2071 state: PipelineState,
2072 items: list,
2073 body: list,
2074 agent_name: str | None,
2075 max_output: int,
2076 max_conc: int,
2077 working_set: ChannelWorkingSet | None = None,
2078 ) -> PipelineState:
2079 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2080 log.info(
2081 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2082 step_id, len(items), max_conc,
2083 )
2085 # Pre-allocate results list to preserve item order across threads.
2086 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2087 # legitimate None results (body produced no output) are kept.
2088 body_polku = bool(body)
2089 declared_output = step.get("output")
2090 _SKIPPED = object()
2091 results: list = [_SKIPPED] * len(items)
2092 item_working_sets: list[ChannelWorkingSet | None] = [None] * len(items)
2093 lock = threading.Lock()
2094 produced_outputs = [0] # mutable counter shared across threads
2095 stop_flag = threading.Event() # set when max_output budget is exhausted
2096 item_errors: list = [] # budget errors collected from worker threads
2097 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2099 def run_item(idx: int, item: Any) -> None:
2100 nonlocal item_errors
2101 # Honour stop flag — skip if budget already exhausted before we start
2102 if max_output and stop_flag.is_set():
2103 return
2105 item_working_set = (
2106 working_set.fork() if working_set is not None else None
2107 )
2108 item_working_sets[idx] = item_working_set
2109 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2111 if body_polku:
2112 for sub_step in body:
2113 iter_state = self._run_body_step(
2114 sub_step, iter_state, item_working_set
2115 )
2116 # Collect budget errors from this iteration (thread-safe merge)
2117 new_errs = iter_state.get("errors") or []
2118 if new_errs:
2119 with lock:
2120 item_errors = self._merge_errors(item_errors, new_errs)
2121 publish_out = iter_state.get("publish")
2122 results[idx] = self._body_item_result(iter_state, declared_output)
2123 if isinstance(publish_out, dict) and publish_out.get("path"):
2124 with lock:
2125 produced_outputs[0] += 1
2126 if max_output and produced_outputs[0] >= max_output:
2127 stop_flag.set()
2128 elif agent_name:
2129 resolved = self._resolver.resolve(agent_name)
2130 item_read_state = self._read_state(iter_state, item_working_set)
2131 item_ctx = self._make_ctx(item_read_state)
2132 resolved_step = self._resolve_step_runtime(
2133 step, item_read_state, item_ctx
2134 )
2135 resolved_step = self._attach_communication_context(
2136 resolved_step, item_read_state
2137 )
2138 resolved_step = self._attach_usage_instructions(resolved_step)
2139 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2140 results[idx] = iter_state.get(step_id)
2141 item_metrics_map = iter_state.get("step_context_metrics")
2142 if isinstance(item_metrics_map, dict):
2143 with lock:
2144 accumulated_metrics.update(item_metrics_map)
2146 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2147 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2148 for future in as_completed(futures):
2149 try:
2150 future.result()
2151 except SkillLoadError:
2152 raise
2153 except RunHardFailure:
2154 pool.shutdown(wait=False, cancel_futures=True)
2155 raise
2156 except Exception as exc: # noqa: BLE001
2157 idx = futures[future]
2158 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2159 results[idx] = {"error": str(exc)}
2161 self._merge_item_working_sets(working_set, item_working_sets)
2163 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2164 ordered = [r for r in results if r is not _SKIPPED]
2165 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
2166 final: PipelineState = {**state, step_id: step_result}
2167 if accumulated_metrics:
2168 final = {
2169 **final,
2170 "step_context_metrics": _merge_metrics(
2171 state.get("step_context_metrics"), accumulated_metrics
2172 ),
2173 }
2174 if item_errors:
2175 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2176 return final
2178 def _run_while(
2179 self,
2180 step: dict,
2181 state: PipelineState,
2182 working_set: ChannelWorkingSet | None = None,
2183 ) -> PipelineState:
2184 """
2185 While loop with exit_condition and optional entry condition.
2187 Fields:
2188 condition: "{{ triage.approved }}" — if false, skip entirely
2189 input: initial value for loop.current
2190 exit_condition: "{{ review.pass }}" — break when true
2191 max_iterations: int or template
2192 output: body step id whose result feeds the next round's
2193 `loop.current` (envelope-unwrapped when dict with `output`)
2194 body: list of sub-steps
2196 Result shape (r.68): {passed, iterations, output, exhausted, skipped}.
2197 The feed comes ONLY from the declared `output` body step — there is no
2198 hardcoded `refine` feed; a body step named `refine` is not special.
2200 The invocation working set is threaded into every body step and into
2201 condition evaluation so reads resolve over the committed snapshot plus
2202 this invocation's pending writes; the returned state is still built
2203 from the raw ``state``.
2204 """
2205 step_id = step.get("id", "?")
2206 read_state = self._read_state(state, working_set)
2207 ctx = self._make_ctx(read_state)
2209 # Entry condition — skip is a first-class state (r.68): a skipped
2210 # while writes a result with output = the resolved input so downstream
2211 # `.output` reads are non-null on the skipped path.
2212 entry_cond = step.get("condition")
2213 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
2214 input_spec = step.get("input")
2215 skip_input = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
2216 return {**state, step_id: {
2217 "passed": True,
2218 "iterations": 0,
2219 "output": skip_input,
2220 "exhausted": False,
2221 "skipped": True,
2222 }}
2224 # Initial current value from input
2225 input_spec = step.get("input")
2226 if input_spec is not None:
2227 current = _resolve_input(input_spec, read_state, ctx)
2228 else:
2229 current = None
2231 exit_condition = step.get("exit_condition")
2232 max_iter = _resolve_int(step.get("max_iterations", 2), ctx, 2)
2233 body = step.get("body") or []
2234 declared_output = step.get("output")
2236 passed = False
2237 iteration = 0
2238 loop_state: PipelineState = dict(state)
2240 while iteration < max_iter:
2241 # Inject loop context
2242 loop_state = {
2243 **state,
2244 "loop": {"current": current, "iteration": iteration},
2245 step_id: {"current": current, "passed": False},
2246 }
2247 # Add current item context (from outer foreach if any)
2248 if "item" in state:
2249 loop_state["item"] = state["item"]
2251 # Run body steps
2252 for sub_step in body:
2253 loop_state = self._run_body_step(sub_step, loop_state, working_set)
2255 iteration += 1
2257 # Feed: next round's loop.current comes from the declared body-output
2258 # step (envelope-unwrapped); otherwise current is unchanged.
2259 # Envelopes: a `{output: ...}` (r.68) and a `{results: [...]}`
2260 # aggregate (parallel/foreach body-step) are unwrapped to their
2261 # inner value so the next round's consumer receives the list, not
2262 # the envelope. A raw value (plain list/dict) feeds as-is.
2263 if declared_output:
2264 feed_value = loop_state.get(declared_output)
2265 if isinstance(feed_value, dict) and "output" in feed_value:
2266 current = feed_value["output"]
2267 elif isinstance(feed_value, dict) and isinstance(feed_value.get("results"), list):
2268 current = feed_value["results"]
2269 else:
2270 current = feed_value
2272 # Check exit condition
2273 if exit_condition:
2274 exit_ctx = self._make_ctx(
2275 self._read_state(loop_state, working_set)
2276 )
2277 if _eval_condition(exit_condition, exit_ctx):
2278 passed = True
2279 break
2280 else:
2281 passed = True
2282 break
2284 skipped = False
2285 exhausted = not passed
2286 final: PipelineState = {**state, step_id: {
2287 "passed": passed,
2288 "iterations": iteration,
2289 "output": current,
2290 "exhausted": exhausted,
2291 "skipped": skipped,
2292 }}
2293 # ENG-11: a while that ends exhausted (ran max_iterations without
2294 # exit_condition passing, not skipped) and whose `fail_if_exhausted`
2295 # evaluates true is a hard run failure — nothing downstream runs. A
2296 # skipped or passed while must never raise.
2297 fail_if_exhausted = step.get("fail_if_exhausted")
2298 if (
2299 fail_if_exhausted is not None
2300 and exhausted
2301 and _eval_condition(fail_if_exhausted, self._make_ctx(read_state))
2302 ):
2303 raise RunHardFailure(
2304 "EXHAUSTED",
2305 f"while step '{step_id}' exhausted max_iterations={max_iter} "
2306 "without exit_condition passing (fail_if_exhausted)",
2307 {"step_id": step_id, "exhausted": True},
2308 )
2309 # Propagate any errors accumulated in the loop body back to the outer state
2310 loop_errors = loop_state.get("errors") or []
2311 if loop_errors:
2312 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), loop_errors)}
2313 return final
2315 def _run_spawn(
2316 self,
2317 step: dict,
2318 state: PipelineState,
2319 working_set: ChannelWorkingSet | None = None,
2320 ) -> PipelineState:
2321 """Run a spawn step: a recursive node tree of agent calls (S2).
2323 The `agent` runs once as director at the root and may emit a `children`
2324 plan; every deeper node re-runs the `child` role (default = `agent`)
2325 with that node's own `input`. Depth and fan are bounded by the
2326 spawn_policy, and the whole tree draws from one shared SpawnBudget.
2327 """
2328 step_id = step.get("id", "?")
2329 read_state = self._read_state(state, working_set)
2330 ctx = self._make_ctx(read_state)
2331 policy = step.get("spawn_policy") or {}
2333 max_depth_raw = policy.get("max_depth")
2334 if max_depth_raw is None:
2335 max_depth = _SPAWN_DEFAULT_MAX_DEPTH
2336 else:
2337 max_depth = _resolve_int(max_depth_raw, ctx, _SPAWN_DEFAULT_MAX_DEPTH)
2339 max_spawn_raw = policy.get("max_spawn_per_node")
2340 if max_spawn_raw is None:
2341 max_spawn_per_node = _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
2342 else:
2343 max_spawn_per_node = _resolve_int(max_spawn_raw, ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE)
2345 max_calls_raw = policy.get("max_agent_calls")
2346 if max_calls_raw is None:
2347 budget = SpawnBudget(_SPAWN_DEFAULT_MAX_AGENT_CALLS)
2348 else:
2349 budget = SpawnBudget(_resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS))
2351 run_max_parallel = int((self._run_context.get("budget") or {}).get("max_parallel", DEFAULT_MAX_PARALLEL))
2352 max_concurrent_raw = policy.get("max_concurrent_children")
2353 if max_concurrent_raw is None:
2354 max_concurrent_children = run_max_parallel
2355 else:
2356 max_concurrent_children = _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
2358 director = step.get("agent")
2359 if not director:
2360 log.warning("[step] %s spawn has no agent — skipping", step_id)
2361 return state
2362 child_role = step.get("child") or director
2364 envelope, metrics = self._spawn_node(
2365 node_id=step_id,
2366 role=director,
2367 depth=0,
2368 step=step,
2369 input_override=_UNSET,
2370 base_state=read_state,
2371 base_ctx=ctx,
2372 child_role=child_role,
2373 budget=budget,
2374 max_depth=max_depth,
2375 max_spawn_per_node=max_spawn_per_node,
2376 max_concurrent_children=max_concurrent_children,
2377 working_set=working_set,
2378 )
2379 final: PipelineState = {**state, step_id: envelope}
2380 if metrics:
2381 final = {
2382 **final,
2383 "step_context_metrics": _merge_metrics(
2384 state.get("step_context_metrics"), metrics
2385 ),
2386 }
2387 return final
2389 def _spawn_node(
2390 self,
2391 *,
2392 node_id: str,
2393 role: str,
2394 depth: int,
2395 step: dict,
2396 input_override: Any,
2397 base_state: PipelineState,
2398 base_ctx: dict,
2399 child_role: str,
2400 budget: SpawnBudget,
2401 max_depth: int,
2402 max_spawn_per_node: int | None,
2403 max_concurrent_children: int,
2404 working_set: ChannelWorkingSet | None,
2405 ) -> tuple[dict, dict]:
2406 """Run one spawn-tree node; returns (node envelope, subtree metrics)."""
2407 tracer = self._tracer
2408 parent_obs = (
2409 tracer.open_parent_observation(
2410 name=f"spawn:{node_id}",
2411 input={"node_id": node_id, "role": role, "depth": depth},
2412 )
2413 if tracer
2414 else None
2415 )
2416 try:
2417 resolved_role = self._resolver.resolve(role)
2418 resolved_step = self._resolve_step_runtime(
2419 step, base_state, base_ctx, input_override=input_override,
2420 )
2421 resolved_step = self._attach_communication_context(
2422 resolved_step, base_state
2423 )
2424 resolved_step = self._attach_usage_instructions(resolved_step)
2425 if not budget.consume():
2426 raise RunHardFailure(
2427 "SPAWN_BUDGET_EXHAUSTED",
2428 f"spawn tree budget exhausted at node '{node_id}'",
2429 {"node_id": node_id, "depth": depth},
2430 )
2431 result_state = self._invoke_agent(resolved_role, resolved_step, base_state)
2432 this_result = result_state.get(node_id)
2433 metrics = dict(result_state.get("step_context_metrics") or {})
2434 self._stage_spawn_channel_writes(
2435 step, this_result, working_set
2436 )
2438 children = self._spawn_children_from_result(this_result)
2439 envelope: dict = {
2440 "result": this_result,
2441 "children": {},
2442 "passed": not self._spawn_is_soft_fail(this_result),
2443 "failing": [],
2444 "exhausted": False,
2445 }
2447 if not children:
2448 return envelope, metrics
2449 if depth + 1 > max_depth:
2450 envelope["exhausted"] = True
2451 return envelope, metrics
2452 if max_spawn_per_node is not None and len(children) > max_spawn_per_node:
2453 raise RunHardFailure(
2454 "SPAWN_OVER_FLIGHT",
2455 f"spawn node '{node_id}' emitted {len(children)} children, "
2456 f"exceeds max_spawn_per_node={max_spawn_per_node}",
2457 {
2458 "node_id": node_id,
2459 "depth": depth,
2460 "children": len(children),
2461 "max_spawn_per_node": max_spawn_per_node,
2462 },
2463 )
2465 pool_size = self._spawn_pool_size(max_spawn_per_node, max_concurrent_children)
2466 if len(children) <= 1 or pool_size <= 1:
2467 for idx, child in enumerate(children):
2468 child_id, child_envelope, child_metrics = self._spawn_child(
2469 node_id=node_id,
2470 child_role=child_role,
2471 depth=depth,
2472 step=step,
2473 base_state=base_state,
2474 base_ctx=base_ctx,
2475 budget=budget,
2476 max_depth=max_depth,
2477 max_spawn_per_node=max_spawn_per_node,
2478 max_concurrent_children=max_concurrent_children,
2479 working_set=working_set,
2480 child=child,
2481 idx=idx,
2482 )
2483 metrics = _merge_metrics(metrics, child_metrics)
2484 envelope["children"][child_id] = child_envelope
2485 if not child_envelope["passed"]:
2486 envelope["passed"] = False
2487 envelope["failing"].append(child_id)
2488 return envelope, metrics
2490 pool = ThreadPoolExecutor(max_workers=pool_size)
2491 cancel = threading.Event()
2492 futures = []
2493 child_working_sets: dict[Any, ChannelWorkingSet] = {}
2494 for idx, child in enumerate(children):
2495 child_working_set = (
2496 working_set.fork() if working_set is not None else None
2497 )
2498 ctx = contextvars.copy_context()
2499 future = pool.submit(
2500 ctx.run,
2501 self._spawn_child,
2502 node_id=node_id,
2503 child_role=child_role,
2504 depth=depth,
2505 step=step,
2506 base_state=base_state,
2507 base_ctx=base_ctx,
2508 budget=budget,
2509 max_depth=max_depth,
2510 max_spawn_per_node=max_spawn_per_node,
2511 max_concurrent_children=max_concurrent_children,
2512 working_set=child_working_set,
2513 child=child,
2514 idx=idx,
2515 )
2516 futures.append(future)
2517 if child_working_set is not None:
2518 child_working_sets[future] = child_working_set
2519 hard_error: Exception | None = None
2520 child_results: dict[Any, tuple[str, dict, dict]] = {}
2521 try:
2522 for fut in as_completed(futures):
2523 try:
2524 child_results[fut] = fut.result()
2525 except (RunHardFailure, SkillLoadError) as exc:
2526 hard_error = exc
2527 cancel.set()
2528 for pending in futures:
2529 pending.cancel()
2530 break
2531 if hard_error is None:
2532 for fut in futures:
2533 child_id, child_envelope, child_metrics = child_results[fut]
2534 self._merge_spawn_fork_writes(
2535 working_set, child_working_sets.get(fut)
2536 )
2537 metrics = _merge_metrics(metrics, child_metrics)
2538 envelope["children"][child_id] = child_envelope
2539 if not child_envelope["passed"]:
2540 envelope["passed"] = False
2541 envelope["failing"].append(child_id)
2542 finally:
2543 pool.shutdown(wait=False, cancel_futures=True)
2544 if hard_error is not None:
2545 raise hard_error
2547 return envelope, metrics
2548 finally:
2549 if tracer is not None and parent_obs is not None:
2550 tracer.close_parent_observation(parent_obs)
2552 def _stage_spawn_channel_writes(
2553 self,
2554 step: dict,
2555 result: Any,
2556 working_set: ChannelWorkingSet | None,
2557 ) -> None:
2558 """Stage a spawn node's declared writes from its own agent result."""
2559 if working_set is None:
2560 return
2561 writes = _declared_output_channel_writes(step, result, self._channel_names)
2562 for name, value in writes.items():
2563 working_set.stage(name, value)
2565 @staticmethod
2566 def _merge_spawn_fork_writes(
2567 working_set: ChannelWorkingSet | None,
2568 fork: ChannelWorkingSet | None,
2569 ) -> None:
2570 """Fold a completed child's isolated fork writes into the parent set.
2572 Runs on the main thread after the child future resolves, so the shared
2573 working set is never mutated from a worker thread.
2574 """
2575 if working_set is None or fork is None:
2576 return
2577 for name, value in fork.pending_writes():
2578 working_set.stage(name, value)
2580 @staticmethod
2581 def _spawn_pool_size(max_spawn_per_node: int | None, max_concurrent_children: int) -> int:
2582 if max_spawn_per_node is None:
2583 return max(1, max_concurrent_children)
2584 return max(1, min(max_spawn_per_node, max_concurrent_children))
2586 def _spawn_child(
2587 self,
2588 *,
2589 node_id: str,
2590 child_role: str,
2591 depth: int,
2592 step: dict,
2593 base_state: PipelineState,
2594 base_ctx: dict,
2595 budget: SpawnBudget,
2596 max_depth: int,
2597 max_spawn_per_node: int | None,
2598 max_concurrent_children: int,
2599 working_set: ChannelWorkingSet | None,
2600 child: Any,
2601 idx: int,
2602 ) -> tuple[str, dict, dict]:
2603 if not self._spawn_has_input(child):
2604 raise RunHardFailure(
2605 "INVALID_CHILD_PLAN",
2606 f"spawn child #{idx} of node '{node_id}' must be a dict "
2607 "with a non-empty 'input'",
2608 {"node_id": node_id, "child_index": idx, "child": repr(child)[:200]},
2609 )
2610 child_id = child["id"] if child.get("id") else f"{node_id}.{idx}"
2611 child_input = child["input"]
2612 resolved_child_input = _resolve_input(child_input, base_state, base_ctx)
2613 child_envelope, child_metrics = self._spawn_node(
2614 node_id=child_id,
2615 role=child_role,
2616 depth=depth + 1,
2617 step={"id": child_id, "input": resolved_child_input},
2618 input_override=resolved_child_input,
2619 base_state=base_state,
2620 base_ctx=base_ctx,
2621 child_role=child_role,
2622 budget=budget,
2623 max_depth=max_depth,
2624 max_spawn_per_node=max_spawn_per_node,
2625 max_concurrent_children=max_concurrent_children,
2626 working_set=working_set,
2627 )
2628 return child_id, child_envelope, child_metrics
2630 @staticmethod
2631 def _spawn_children_from_result(result: Any) -> list:
2632 """The node's child plan: [] when absent or empty; raises on a malformed plan."""
2633 if not isinstance(result, dict):
2634 return []
2635 raw = result.get("children")
2636 if raw is None:
2637 return []
2638 if not isinstance(raw, list):
2639 raise RunHardFailure(
2640 "INVALID_CHILD_PLAN",
2641 "spawn agent 'children' must be a list",
2642 {"children": repr(raw)[:200]},
2643 )
2644 return raw
2646 @staticmethod
2647 def _spawn_has_input(child: Any) -> bool:
2648 """A valid child plan item: a dict carrying a non-empty 'input'."""
2649 if not isinstance(child, dict):
2650 return False
2651 input_val = child.get("input")
2652 if input_val is None:
2653 return False
2654 if isinstance(input_val, str) and not input_val.strip():
2655 return False
2656 if isinstance(input_val, (list, dict)) and not input_val:
2657 return False
2658 return True
2660 @staticmethod
2661 def _spawn_is_soft_fail(result: Any) -> bool:
2662 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error.
2664 An envelope-shaped result is never soft: ``status:error`` is a hard
2665 failure (the gate raises before this), and ``ok``/``partial``/
2666 ``no_success`` are valid non-fatal results.
2667 """
2668 if not isinstance(result, dict):
2669 return False
2670 if result.get("status") in ENVELOPE_STATUSES:
2671 return False
2672 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
2673 return True
2674 return any(key.startswith("budget:") for key in result)
2677# ---------------------------------------------------------------------------
2678# GraphBuilder
2679# ---------------------------------------------------------------------------
2681def _topological_order(steps: list[dict]) -> list[dict]:
2682 """
2683 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2684 """
2685 id_to_step = {s["id"]: s for s in steps}
2686 visited: set[str] = set()
2687 order: list[dict] = []
2689 def visit(step_id: str) -> None:
2690 if step_id in visited:
2691 return
2692 visited.add(step_id)
2693 for dep in (id_to_step[step_id].get("depends_on") or []):
2694 if dep in id_to_step:
2695 visit(dep)
2696 order.append(id_to_step[step_id])
2698 for s in steps:
2699 visit(s["id"])
2700 return order
2703# ---------------------------------------------------------------------------
2704# Call-step build-time expansion (ADR-0025 D3/D4)
2705# ---------------------------------------------------------------------------
2706#
2707# A `type: call` step is expanded at build time into the function's inlined body
2708# steps plus two synthesized op steps (`__call_bind` / `__call_emit`). The
2709# runtime executes only this flat graph; LangGraph never sees a call step. The
2710# caller reads the function outputs via the call result shape
2711# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
2712# The function body's `{{ params.<name> }}` references are rewritten to
2713# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves.
2714#
2715# Isolation (ADR-0025 D5 / 00-functions § Isolation) and recursion (D7/D9): a
2716# function body step must not reference a step-id outside its own body (a
2717# one-way dependency check — outward references to caller step-ids are build
2718# errors); a cycle in the function call graph (self or mutual) is a build
2719# error. Bounded while loops are runtime constructs and are not cycle-flagged.
2721# Call step result namespace keys (internal scaffolding).
2722def _scope_params_value(value: Any, call_id: str) -> Any:
2723 """Rewrite `params.` template references to the call-scoped namespace.
2725 Operates on the function body's string values (templates). Dict keys are
2726 untouched — only templated string VALUES are scoped, so a function body's
2727 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
2728 """
2729 if isinstance(value, str):
2730 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
2731 if isinstance(value, dict):
2732 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
2733 if isinstance(value, list):
2734 return [_scope_params_value(v, call_id) for v in value]
2735 return value
2738def _scope_params(step: dict, call_id: str) -> dict:
2739 """Deep-scope a function body step's params references for a call id."""
2740 scoped: dict = dict(step)
2741 for field in ("input", "params", "condition", "exit_condition", "agent", "function"):
2742 if field in scoped and isinstance(scoped[field], (str, dict, list)):
2743 scoped[field] = _scope_params_value(scoped[field], call_id)
2744 if "body" in scoped and isinstance(scoped["body"], list):
2745 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
2746 return scoped
2749def _collect_step_ids(steps: list[dict]) -> set[str]:
2750 """All step ids declared in *steps*, including nested body step ids."""
2751 ids: set[str] = set()
2752 for step in steps:
2753 step_id = step.get("id")
2754 if isinstance(step_id, str):
2755 ids.add(step_id)
2756 body = step.get("body")
2757 if isinstance(body, list):
2758 ids |= _collect_step_ids(body)
2759 return ids
2762def _iter_steps(steps: list[dict]):
2763 """Yield every step in *steps*, recursing into nested bodies."""
2764 for step in steps:
2765 yield step
2766 body = step.get("body")
2767 if isinstance(body, list):
2768 yield from _iter_steps(body)
2771_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
2772_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
2773_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
2775_REFERENCE_FIELDS = ("input", "params", "condition", "exit_condition")
2777# Engine-reserved template namespaces — never step-id references, even if a
2778# caller happens to declare a step with one of these names.
2779_RESERVED_NAMESPACES = frozenset({"params", "loop", "item", "_item", "budget", "config"})
2782def _referenced_names(value: Any) -> set[str]:
2783 """Root identifiers read by templated strings.
2785 Only the leading identifier of each reference (before the first ``.`` or
2786 ``[``) can name a step-id; nested field segments are ignored. Quoted
2787 literals are ignored.
2788 """
2789 names: set[str] = set()
2790 if isinstance(value, str):
2791 for expr in _TEMPLATE_REF_RE.findall(value):
2792 bare = _STRING_LITERAL_RE.sub("", expr)
2793 for match in _IDENTIFIER_RE.finditer(bare):
2794 if match.start() == 0 or bare[match.start() - 1] != ".":
2795 names.add(match.group())
2796 elif isinstance(value, dict):
2797 for v in value.values():
2798 names |= _referenced_names(v)
2799 elif isinstance(value, list):
2800 for v in value:
2801 names |= _referenced_names(v)
2802 return names
2805def _validate_function_purity(
2806 fn_name: str, body: list[dict], visible_ids: set[str]
2807) -> None:
2808 """One-way isolation: a function body must not reference a caller step-id.
2810 A body step may read its own body step-ids and the params namespace; a
2811 reference to a step-id that exists outside the body (``visible_ids`` but
2812 not declared in the body) is an outward dependency and a hard build error.
2813 """
2814 local_ids = _collect_step_ids(body)
2815 for step in _iter_steps(body):
2816 step_id = step.get("id", "?")
2817 for dep in step.get("depends_on") or []:
2818 if dep in visible_ids and dep not in local_ids:
2819 raise ValueError(
2820 f"function '{fn_name}' step '{step_id}': depends_on "
2821 f"'{dep}' is outside the function body (isolation — "
2822 "function bodies see only their own steps and params)"
2823 )
2824 for field in _REFERENCE_FIELDS:
2825 for ref in _referenced_names(step.get(field)):
2826 if ref in _RESERVED_NAMESPACES:
2827 continue
2828 if ref in visible_ids and ref not in local_ids:
2829 raise ValueError(
2830 f"function '{fn_name}' step '{step_id}': '{field}' "
2831 f"references '{ref}', a step-id outside the function "
2832 "body (isolation — function bodies see only their own "
2833 "steps and params)"
2834 )
2837def _expand_call(
2838 step: dict,
2839 registry: ManifestRegistry,
2840 stack: list[str],
2841 call_map: dict[str, str],
2842 expanded_functions: dict[str, str],
2843 visible_ids: set[str],
2844) -> list[dict]:
2845 """Expand a call step into bind + inlined body + emit steps."""
2846 fn_name = step["function"]
2847 call_id = step["id"]
2849 if fn_name in stack:
2850 chain = " -> ".join(stack + [fn_name])
2851 raise ValueError(
2852 f"cycle detected in function expansion: {chain} "
2853 "(ADR-0025 D7/D9 — recursion must be a bounded while loop or idempotent re-run)"
2854 )
2856 # ADR-0025 D7: inlining the same function twice in one pipeline build
2857 # collides its body step-ids (flat graph). This is a hard build error until
2858 # id-prefixing lands; naming the function and both call sites pins the bug.
2859 if fn_name in expanded_functions:
2860 raise ValueError(
2861 f"duplicate call expansion: function '{fn_name}' called at both "
2862 f"'{expanded_functions[fn_name]}' and '{call_id}' in one pipeline — "
2863 "inlined body step ids collide (ADR-0025 D7)"
2864 )
2865 expanded_functions[fn_name] = call_id
2867 function = registry.get("Function", fn_name)
2868 if function is None:
2869 raise ValueError(
2870 f"call step '{call_id}' references unknown function '{fn_name}' "
2871 "(no kind: Function manifest in the workflows tree)"
2872 )
2874 body = (function.get("spec") or {}).get("steps") or []
2875 _validate_function_purity(fn_name, body, visible_ids)
2877 sub_stack = list(stack) + [fn_name]
2878 nested_visible = visible_ids | _collect_step_ids(body)
2879 outputs_spec = ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
2881 bind_id = f"_call__{call_id}__bind"
2882 emit_id = f"_call__{call_id}__emit"
2883 call_map[call_id] = emit_id
2885 bind_step: dict = {
2886 "id": bind_id,
2887 "type": "builtin",
2888 "component": "__call_bind",
2889 "params": {**(step.get("params") or {}), "__call_id": call_id},
2890 "depends_on": list(step.get("depends_on") or []),
2891 }
2893 expanded_body: list[dict] = []
2894 for body_step in body:
2895 scoped = _scope_params(body_step, call_id)
2896 for sub in _expand_step(
2897 scoped, registry, sub_stack, call_map, expanded_functions, nested_visible
2898 ):
2899 dep = list(sub.get("depends_on") or [])
2900 if bind_id not in dep:
2901 sub["depends_on"] = [bind_id, *dep]
2902 expanded_body.append(sub)
2904 emit_step: dict = {
2905 "id": emit_id,
2906 "type": "builtin",
2907 "component": "__call_emit",
2908 "params": {"__call_id": call_id, "__outputs": outputs_spec},
2909 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
2910 }
2912 return [bind_step, *expanded_body, emit_step]
2915def _expand_step(
2916 step: dict,
2917 registry: ManifestRegistry,
2918 stack: list[str],
2919 call_map: dict[str, str],
2920 expanded_functions: dict[str, str],
2921 visible_ids: set[str],
2922) -> list[dict]:
2923 """Expand one step (recursing into while/foreach bodies)."""
2924 if step.get("type") == "call":
2925 return _expand_call(
2926 step, registry, stack, call_map, expanded_functions, visible_ids
2927 )
2928 if "body" in step and isinstance(step["body"], list):
2929 expanded_body: list[dict] = []
2930 for sub in step["body"]:
2931 expanded_body.extend(
2932 _expand_step(
2933 sub, registry, stack, call_map, expanded_functions, visible_ids
2934 )
2935 )
2936 return [{**step, "body": expanded_body}]
2937 return [step]
2940def _rewrite_call_deps(steps: list[dict], call_map: dict[str, str]) -> None:
2941 """Rewrite `depends_on` references to call ids → the corresponding emit id.
2943 After a call is expanded away, a downstream step that depends_on the call id
2944 must instead wait on the emit step (which writes state[<call_id>]). This
2945 guarantees the caller's `{{ <call_id>.outputs.<name> }}` reads resolve after
2946 the function body ran.
2947 """
2948 for step in steps:
2949 deps = step.get("depends_on") or []
2950 if deps:
2951 step["depends_on"] = [call_map.get(d, d) for d in deps]
2952 if "body" in step and isinstance(step["body"], list):
2953 _rewrite_call_deps(step["body"], call_map)
2956def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
2957 """Expand all `call` steps into the flat runtime graph (ADR-0025 D4).
2959 Returns a new step list with each call replaced by its bind + inlined body +
2960 emit steps; a call step never survives to the runtime graph. Cycles and
2961 outward references in a function body are hard errors.
2962 """
2963 call_map: dict[str, str] = {}
2964 expanded_functions: dict[str, str] = {}
2965 visible_ids = _collect_step_ids(steps)
2966 expanded: list[dict] = []
2967 for step in steps:
2968 expanded.extend(
2969 _expand_step(step, registry, [], call_map, expanded_functions, visible_ids)
2970 )
2971 _rewrite_call_deps(expanded, call_map)
2972 return expanded
2975_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
2978def _validate_channel_entry(index: int, channel: Any) -> None:
2979 """Validate one spec.channels entry against the frozen ChannelSpec shape.
2981 The engine build path does not run manifest-schema validation, so a
2982 malformed entry is rejected here — before any ``["name"]`` indexing — with
2983 an error that identifies the offending entry.
2984 """
2985 if not isinstance(channel, dict):
2986 raise ValueError(
2987 f"Pipeline spec.channels[{index}] must be a mapping; got "
2988 f"{type(channel).__name__}"
2989 )
2990 name = channel.get("name")
2991 if not isinstance(name, str) or not name:
2992 raise ValueError(
2993 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
2994 "channel requires a non-empty string 'name'"
2995 )
2996 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
2997 raise ValueError(
2998 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
2999 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}"
3000 )
3003def _validated_channels(raw_channels: Any) -> list[dict]:
3004 """Return spec.channels as a validated list, or [] when absent.
3006 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3007 """
3008 if raw_channels is None:
3009 return []
3010 if not isinstance(raw_channels, list):
3011 raise ValueError(
3012 f"Pipeline spec.channels must be a list; got "
3013 f"{type(raw_channels).__name__}"
3014 )
3015 for index, channel in enumerate(raw_channels):
3016 _validate_channel_entry(index, channel)
3017 return raw_channels
3020def _validate_step_channel_bindings(
3021 steps: list[dict], channel_names: frozenset[str]
3022) -> None:
3023 """Reject malformed or undeclared step-level channel bindings at build time.
3025 Every step's declared ``output_channel`` must be a non-empty declared
3026 channel name, and every declared ``input_channels`` entry must be a
3027 non-empty declared channel name. This runs unconditionally, so a binding
3028 to a channel absent from ``spec.channels`` fails the build even when no
3029 channels are declared.
3030 """
3031 for step in _iter_steps(steps):
3032 _validate_step_output_channel(step, channel_names)
3033 _validate_step_input_channels(step, channel_names)
3036def _validate_step_output_channel(step: dict, channel_names: frozenset[str]) -> None:
3037 channel = step.get("output_channel")
3038 if channel is None:
3039 return
3040 if not isinstance(channel, str) or not channel:
3041 raise RunHardFailure(
3042 "CHANNEL_BINDING_ERROR",
3043 f"Step '{step.get('id', '?')}' output_channel must be a non-empty "
3044 f"string; got {channel!r}",
3045 )
3046 if channel not in channel_names:
3047 raise RunHardFailure(
3048 "CHANNEL_BINDING_ERROR",
3049 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3050 "which is not declared in spec.channels",
3051 )
3054def _validate_step_input_channels(step: dict, channel_names: frozenset[str]) -> None:
3055 channels = step.get("input_channels")
3056 if channels is None:
3057 return
3058 if not isinstance(channels, list):
3059 raise RunHardFailure(
3060 "CHANNEL_BINDING_ERROR",
3061 f"Step '{step.get('id', '?')}' input_channels must be a list of "
3062 f"channel names; got {type(channels).__name__}",
3063 )
3064 for channel in channels:
3065 if not isinstance(channel, str) or not channel:
3066 raise RunHardFailure(
3067 "CHANNEL_BINDING_ERROR",
3068 f"Step '{step.get('id', '?')}' input_channels has an invalid "
3069 f"channel name '{channel}'; channel names must be non-empty "
3070 "strings",
3071 )
3072 if channel not in channel_names:
3073 raise RunHardFailure(
3074 "CHANNEL_BINDING_ERROR",
3075 f"Step '{step.get('id', '?')}' reads undeclared channel "
3076 f"'{channel}'; the channel must be declared in spec.channels",
3077 )
3080def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3081 """Reject duplicate and engine/step-reserved channel names."""
3082 forbidden = (
3083 {_CORE}
3084 | set(_RESERVED_NAMESPACES)
3085 | _CORE_STATE_KEYS
3086 | _collect_step_ids(steps)
3087 )
3088 seen: set[str] = set()
3089 for channel in channels:
3090 name = channel.get("name")
3091 if name in seen:
3092 raise ValueError(
3093 f"Pipeline declares channel '{name}' more than once; channel "
3094 "names must be unique"
3095 )
3096 seen.add(name)
3097 if name in forbidden:
3098 raise ValueError(
3099 f"Pipeline channel '{name}' collides with an engine-reserved "
3100 "name (a step id, engine state key, reserved namespace, or the "
3101 "core state channel); channel names must be unique and "
3102 "non-reserved"
3103 )
3106def _input_path_root(value: str) -> str | None:
3107 """Leading state key of a bare dot-path input string, or None.
3109 Template strings, bracket-prefixed verbatim literals, and blank strings
3110 are not state reads and yield None.
3111 """
3112 if "{{" in value:
3113 return None
3114 stripped = value.lstrip()
3115 if not stripped or stripped.startswith("["):
3116 return None
3117 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3120def _iter_input_roots(input_spec: Any):
3121 """Yield every bare dot-path root in an input spec, recursing into containers."""
3122 if isinstance(input_spec, str):
3123 root = _input_path_root(input_spec)
3124 if root:
3125 yield root
3126 elif isinstance(input_spec, dict):
3127 for value in input_spec.values():
3128 yield from _iter_input_roots(value)
3129 elif isinstance(input_spec, list):
3130 for value in input_spec:
3131 yield from _iter_input_roots(value)
3134def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3135 """Reject channel-pipeline input roots that name nothing the state exposes.
3137 The channel state is the declared channels plus the engine state, so a bare
3138 dot-path input root must be a declared channel, a known step id, a reserved
3139 namespace, or an engine state key. Anything else is a hard build error —
3140 there is no silent fallback to a missing key.
3141 """
3142 allowed = (
3143 {ch["name"] for ch in channels}
3144 | _collect_step_ids(steps)
3145 | set(_RESERVED_NAMESPACES)
3146 | _CORE_STATE_KEYS
3147 )
3148 for step in _iter_steps(steps):
3149 for root in _iter_input_roots(step.get("input")):
3150 if root not in allowed:
3151 raise ValueError(
3152 f"Pipeline channel step '{step.get('id', '?')}' input "
3153 f"references unknown root '{root}'; in a channel pipeline "
3154 "an input root must be a declared channel, a step id, a "
3155 "reserved namespace, or an engine state key"
3156 )
3159def _resolve_allow_instruction(
3160 entry: str, builtins: BuiltinExecutor, step_functions: StepFunctionExecutor | None
3161) -> str | None:
3162 """Resolve one whitelist entry's usage instruction, builtin-first.
3164 Mirrors the runtime dispatch order (ADR-0037 D2): a name in the builtin
3165 register resolves to the Core-owned registry text, otherwise to the
3166 StepFunction's ``agent_instructions``.
3167 """
3168 return instruction_text(entry, builtins, step_functions)
3171_AGENT_INVOKING_STEP_TYPES = frozenset({"sequential", "parallel", "spawn", "foreach"})
3174def _validate_allow_step_types(steps: list[dict]) -> None:
3175 """Reject ``allow`` on a step that does not actually invoke an agent.
3177 Only an agent-bearing step consumes its whitelist; on any other step the
3178 declaration would be silently dropped — notably a ``call`` step, which
3179 build-time expansion removes before the whitelist validator runs, and an
3180 agent-less ``sequential``/``spawn``/``foreach``. A container step that
3181 declares a ``body`` runs its body path, so a ``body`` wins over a declared
3182 ``agent`` and the whitelist would be inert. Runs on the raw step list,
3183 before call expansion.
3184 """
3185 for step in _iter_steps(steps):
3186 if not step.get("allow"):
3187 continue
3188 step_type = step.get("type", "sequential")
3189 agent = step.get("agent")
3190 invokes_agent = (
3191 step_type in _AGENT_INVOKING_STEP_TYPES
3192 and agent
3193 and not step.get("body")
3194 )
3195 if invokes_agent:
3196 continue
3197 step_id = step.get("id", "?")
3198 raise RunHardFailure(
3199 "ALLOW_INVALID_STEP_TYPE",
3200 f"Step '{step_id}' declares 'allow' but is not an agent-bearing "
3201 f"step that invokes an agent: type '{step_type}' with agent "
3202 f"'{agent}' does not actually invoke an agent",
3203 {"step_id": step_id, "type": step_type, "agent": agent},
3204 )
3207def _validate_allow_instructions(
3208 steps: list[dict],
3209 builtins: BuiltinExecutor,
3210 step_functions: StepFunctionExecutor | None,
3211) -> None:
3212 """Require a non-empty usage instruction for every ``allow`` entry.
3214 Walks all steps — including nested loop/parallel bodies — and fails fast
3215 when a whitelisted name resolves to no instruction: neither a whitelistable
3216 builtin registry text nor a StepFunction ``agent_instructions``.
3217 """
3218 for step in _iter_steps(steps):
3219 for entry in step.get("allow") or []:
3220 if _resolve_allow_instruction(entry, builtins, step_functions):
3221 continue
3222 step_id = step.get("id", "?")
3223 raise RunHardFailure(
3224 "AGENT_INSTRUCTION_MISSING",
3225 f"Step '{step_id}' allow entry '{entry}' has no usage "
3226 "instruction: it is neither a whitelistable builtin nor a "
3227 "StepFunction with agent_instructions",
3228 {"step_id": step_id, "allow": entry},
3229 )
3232def _flatten_channel_state(state: dict) -> dict:
3233 """Read the flat committed snapshot nested under the Core namespace."""
3234 return dict(state.get(_CORE) or {})
3237def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3238 """Drop channel-named keys so a node emits only the core state."""
3239 return {k: v for k, v in flat.items() if k not in channel_names}
3242_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3244_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3247def _is_failure_envelope(value: Any) -> bool:
3248 """True when a step output is not a successfully parsed dict.
3250 Missing output, non-dict values, and the ``error``/``parse_error``
3251 envelopes written by the LLM failure paths are not routed. A valid
3252 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3253 it carries an explicit ``"error": null``.
3254 """
3255 if not isinstance(value, dict):
3256 return True
3257 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3258 return False
3259 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3262def _declared_output_channel_writes(
3263 step: dict, output: Any, channel_names: frozenset[str]
3264) -> dict[str, Any]:
3265 """Route one step output's content to its declared output channel.
3267 Returns ``{channel_name: output["content"]}`` for the caller to stage into
3268 the working set. Failure envelopes and steps without an ``output_channel``
3269 are not routed; an undeclared target channel is a hard error.
3270 """
3271 if _is_failure_envelope(output):
3272 return {}
3273 channel = step.get("output_channel")
3274 if not channel:
3275 return {}
3276 if channel not in channel_names:
3277 raise RunHardFailure(
3278 "CHANNEL_BINDING_ERROR",
3279 f"Step '{step.get('id', '?')}' declares output_channel '{channel}' "
3280 "which is not declared in spec.channels",
3281 )
3282 return {channel: output.get("content")}
3285def _route_declared_channels(
3286 step: dict,
3287 pre_state: dict,
3288 core: dict,
3289 step_runner: Any,
3290 channel_names: frozenset[str],
3291) -> dict[str, Any]:
3292 """Route a step's declared output channel write to its channel.
3294 Returns ``{channel_name: value}`` for LangGraph to merge through the
3295 channel's reducer, where ``channel_name`` is the step's single declared
3296 ``output_channel`` and ``value`` is the agent envelope's ``content``. Only
3297 an actual agent invocation is routed: a ``skip_llm_if`` passthrough and a
3298 parallel fan-out envelope are excluded. A step without an ``output_channel``
3299 routes nothing (its output stays in ``core`` untouched), and failure
3300 envelopes are not routed. A spawn step is routed at the spawn seam from its
3301 node result, never here from its envelope.
3302 """
3303 step_id = step.get("id", "?")
3304 if step.get("type") == "spawn":
3305 return {}
3306 if step_id not in core:
3307 return {}
3308 agent_spec = step_runner.resolve_agent(step)
3309 if agent_spec is None:
3310 return {}
3311 if not step_runner.ran_agent(step, pre_state):
3312 return {}
3313 return _declared_output_channel_writes(step, core[step_id], channel_names)
3316def _make_step_node(
3317 step: dict,
3318 step_runner: Any,
3319 channel_names: frozenset[str],
3320 channel_types: dict[str, str],
3321) -> Any:
3322 if not channel_names:
3323 def node_fn(state: PipelineState) -> PipelineState:
3324 return step_runner.run_step(step, state)
3325 return node_fn
3327 def node_fn(state: PipelineState) -> PipelineState:
3328 committed = _flatten_channel_state(state)
3329 working_set = ChannelWorkingSet(committed, channel_types)
3330 core = _strip_channel_keys(
3331 step_runner.run_step(step, committed, working_set), channel_names
3332 )
3333 writes = _route_declared_channels(
3334 step, committed, core, step_runner, channel_names
3335 )
3336 for name, value in writes.items():
3337 working_set.stage(name, value)
3338 channels = working_set.commit()
3339 return {_CORE: {**core, **channels}}
3340 return node_fn
3343def _collect_parallel_channel_writes(
3344 steps: list[dict],
3345 pre_state: dict,
3346 core: dict,
3347 step_runner: Any,
3348 channel_names: frozenset[str],
3349 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
3350) -> list[tuple[str, Any]]:
3351 """Collect one parallel node's declared writes in branch order.
3353 Every branch write is emitted, so fan-in feeds the channel-type reducer
3354 every write in deterministic declaration/branch order: Topic appends each
3355 write, LastValue keeps the last, and BinaryOperatorAggregate combines them
3356 per its accumulated shape (list concat, collection-map union, dict merge,
3357 scalar last). A branch's isolated body writes are emitted before its own
3358 declared write, mirroring the top-level node's route-after-run order.
3359 """
3360 writes: list[tuple[str, Any]] = []
3361 for step in steps:
3362 if branch_working_sets is not None:
3363 branch_working_set = branch_working_sets.get(step["id"])
3364 if branch_working_set is not None:
3365 writes.extend(branch_working_set.pending_writes())
3366 branch = _route_declared_channels(
3367 step, pre_state, core, step_runner, channel_names
3368 )
3369 for channel_name, value in branch.items():
3370 writes.append((channel_name, value))
3371 return writes
3374def _make_parallel_node(
3375 group_steps: list[dict],
3376 step_runner: Any,
3377 channel_names: frozenset[str],
3378 channel_types: dict[str, str],
3379) -> Any:
3380 def node_fn(state: PipelineState) -> PipelineState:
3381 base = (
3382 _flatten_channel_state(state)
3383 if channel_names else state
3384 )
3385 working_set = (
3386 ChannelWorkingSet(base, channel_types) if channel_names else None
3387 )
3388 # Each branch writes into its own fork; the shared set is advanced only
3389 # at fan-in, so branch-internal body writes never race.
3390 branch_working_sets = (
3391 {s["id"]: working_set.fork() for s in group_steps}
3392 if working_set is not None else None
3393 )
3395 def submit_branch(pool: Any, step: dict) -> Any:
3396 if branch_working_sets is None:
3397 return pool.submit(step_runner.run_step, step, base)
3398 return pool.submit(
3399 step_runner.run_step, step, base, branch_working_sets[step["id"]]
3400 )
3402 results: dict[str, Any] = {}
3403 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
3404 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
3405 for future in as_completed(futures):
3406 sid = futures[future]
3407 try:
3408 out = future.result()
3409 results[sid] = out.get(sid)
3410 except SkillLoadError:
3411 raise
3412 except RunHardFailure:
3413 pool.shutdown(wait=False, cancel_futures=True)
3414 raise
3415 except Exception as exc: # noqa: BLE001
3416 log.error("Step '%s' failed: %s", sid, exc)
3417 results[sid] = {"error": str(exc)}
3418 if channel_names:
3419 merged = _strip_channel_keys({**base, **results}, channel_names)
3420 writes = _collect_parallel_channel_writes(
3421 group_steps, base, merged, step_runner, channel_names,
3422 branch_working_sets,
3423 )
3424 for name, value in writes:
3425 working_set.stage(name, value)
3426 channels = working_set.commit()
3427 return {_CORE: {**merged, **channels}}
3428 return {**state, **results}
3429 return node_fn
3432class _ChannelGraphShim:
3433 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
3435 Preserves the invoke contract used by main.py and tests: a flat dict in,
3436 a flat dict out. The flat input is seeded through ``seed_committed`` and
3437 nested under ``__core__`` for the compiled graph; the committed snapshot is
3438 unwrapped back out. Streaming and state surfaces are not exposed because
3439 they would leak or accept the internal ``__core__`` shape; read-only
3440 introspection delegates to the compiled graph.
3441 """
3443 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
3444 self._compiled = compiled
3445 self._specs = specs
3447 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
3448 seeded = seed_committed(self._specs, flat_state)
3449 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
3450 return unwrap(result)
3452 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
3453 seeded = seed_committed(self._specs, flat_state)
3454 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
3455 return unwrap(result)
3457 def stream(self, *args: Any, **kwargs: Any) -> Any:
3458 self._reject_internal_shape("stream")
3460 def astream(self, *args: Any, **kwargs: Any) -> Any:
3461 self._reject_internal_shape("astream")
3463 def update_state(self, *args: Any, **kwargs: Any) -> Any:
3464 self._reject_internal_shape("update_state")
3466 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
3467 self._reject_internal_shape("aupdate_state")
3469 def get_state(self, *args: Any, **kwargs: Any) -> Any:
3470 self._reject_internal_shape("get_state")
3472 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
3473 self._reject_internal_shape("aget_state")
3475 def _reject_internal_shape(self, surface: str) -> None:
3476 raise NotImplementedError(
3477 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
3478 ".invoke()/.ainvoke() contract is exposed through the channel "
3479 "boundary shim"
3480 )
3482 def __getattr__(self, name: str) -> Any:
3483 return getattr(self._compiled, name)
3486class GraphBuilder:
3487 """
3488 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
3490 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
3491 käytetään vain {{ }} -templatejen evaluointiin.
3492 """
3494 def __init__(
3495 self,
3496 registry: ManifestRegistry,
3497 builtins: BuiltinExecutor | None = None,
3498 invoke_agent_fn: Any | None = None,
3499 run_context: dict | None = None,
3500 tracer: Any | None = None,
3501 ) -> None:
3502 self._registry = registry
3503 self._resolver = AgentResolver(registry)
3504 self._invoke_agent_fn = invoke_agent_fn
3505 self._builtins = builtins or _DEFAULT_BUILTINS
3506 if invoke_agent_fn is not None:
3507 self._builtins.invoke_agent_fn = invoke_agent_fn
3508 self._step_functions = StepFunctionExecutor(registry)
3509 self._step_runner = StepRunner(
3510 self._resolver, self._builtins, invoke_agent_fn, run_context,
3511 step_functions=self._step_functions,
3512 tracer=tracer,
3513 )
3515 def build(self, pipeline: dict): # -> CompiledGraph
3516 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
3517 try:
3518 from langgraph.graph import StateGraph, END
3519 except ImportError as exc:
3520 raise ImportError(
3521 "langgraph is required to build a workflow. "
3522 "Install it with: pip install langgraph"
3523 ) from exc
3525 steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
3526 # Validate `allow` step types on the raw graph: call-expansion removes
3527 # call steps (and their dropped `allow`) before the whitelist validator.
3528 _validate_allow_step_types(steps)
3529 # Build-time call-step expansion (ADR-0025 D4): the runtime executes only
3530 # the flat expanded graph. LangGraph never sees a call step.
3531 steps = expand_calls(steps, self._registry)
3532 ordered = _topological_order(steps)
3534 channels: list[dict] = _validated_channels(
3535 (pipeline.get("spec") or {}).get("channels")
3536 )
3537 if channels:
3538 _validate_channel_names(channels, steps)
3539 _validate_channel_inputs(channels, steps)
3540 specs = parse_channel_specs(
3541 [{"name": ch["name"], "type": ch.get("type", LAST_VALUE)} for ch in channels]
3542 )
3543 channel_types = {spec.name: spec.type for spec in specs}
3544 channel_names = frozenset(channel_types)
3545 _validate_step_channel_bindings(steps, channel_names)
3546 _validate_allow_instructions(steps, self._builtins, self._step_functions)
3548 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
3549 envs = (pipeline.get("spec") or {}).get("envs")
3550 if envs is not None and not isinstance(envs, dict):
3551 raise ValueError(
3552 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
3553 "spec.envs must be a map of env name → image ref (ADR-0012)"
3554 )
3556 # Group steps into LangGraph nodes:
3557 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
3558 nodes: list[list[dict]] = []
3559 for step in ordered:
3560 dep_sig = frozenset(step.get("depends_on") or [])
3561 if nodes:
3562 last_group_sig = frozenset(
3563 dep
3564 for s in nodes[-1]
3565 for dep in (s.get("depends_on") or [])
3566 )
3567 last_types = {s.get("type") for s in nodes[-1]}
3568 if (
3569 dep_sig == last_group_sig
3570 and step.get("type") == "parallel"
3571 and last_types == {"parallel"}
3572 ):
3573 nodes[-1].append(step)
3574 continue
3575 nodes.append([step])
3577 state_schema = _build_state_schema() if channels else PipelineState
3578 graph: StateGraph = StateGraph(state_schema)
3579 step_runner = self._step_runner
3580 step_runner.set_channel_routing(channel_names)
3582 prev_node_id: str | None = None
3583 for group in nodes:
3584 if len(group) == 1:
3585 node_id = group[0]["id"]
3586 else:
3587 node_id = "parallel_" + "_".join(s["id"] for s in group)
3589 captured_group = list(group)
3591 if len(captured_group) == 1:
3592 fn = _make_step_node(
3593 captured_group[0], step_runner, channel_names,
3594 channel_types,
3595 )
3596 else:
3597 fn = _make_parallel_node(
3598 captured_group,
3599 step_runner,
3600 channel_names,
3601 channel_types,
3602 )
3604 graph.add_node(node_id, fn)
3606 if prev_node_id is None:
3607 graph.set_entry_point(node_id)
3608 else:
3609 graph.add_edge(prev_node_id, node_id)
3611 prev_node_id = node_id
3613 if prev_node_id:
3614 graph.add_edge(prev_node_id, END)
3615 else:
3616 graph.add_node("noop", lambda s: s)
3617 graph.set_entry_point("noop")
3618 graph.add_edge("noop", END)
3620 compiled = graph.compile()
3621 if channel_names:
3622 return _ChannelGraphShim(compiled, specs)
3623 return compiled
3626# ---------------------------------------------------------------------------
3627# Julkinen API
3628# ---------------------------------------------------------------------------
3630_registry_cache: ManifestRegistry | None = None
3633def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
3634 global _registry_cache
3635 if _registry_cache is None or dirs is not None:
3636 loader = ManifestLoader(dirs)
3637 manifests = loader.load_all()
3638 _registry_cache = ManifestRegistry(manifests)
3639 return _registry_cache
3642def build_workflow(
3643 pipeline_name: str,
3644 *,
3645 dirs: list[Path] | None = None,
3646 builtins: BuiltinExecutor | None = None,
3647 invoke_agent_fn: Any | None = None,
3648 run_context: dict | None = None,
3649 tracer: Any | None = None,
3650):
3651 """
3652 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
3653 LangGraph-workflown.
3655 Args:
3656 pipeline_name: Pipeline metadata.name -arvo.
3657 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
3658 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
3659 invoke_agent_fn: fn(resolved_spec, step, state) → state.
3660 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
3662 Returns:
3663 Compilattu LangGraph CompiledGraph.
3664 """
3665 registry = _get_registry(dirs)
3666 pipeline = registry.get("Pipeline", pipeline_name)
3667 if pipeline is None:
3668 available = [
3669 (m.get("metadata") or {}).get("name")
3670 for m in registry.all_of_kind("Pipeline")
3671 ]
3672 raise KeyError(
3673 f"Pipeline '{pipeline_name}' not found. Available: {available}"
3674 )
3675 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
3676 return builder.build(pipeline)