Coverage for core/langgraph/runner.py: 78%
1843 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-16 06:41 +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._contracts: dict[str, dict] = {}
1146 self._channel_names: frozenset[str] = frozenset()
1148 def set_channel_routing(
1149 self, contracts: dict[str, dict], channel_names: frozenset[str]
1150 ) -> None:
1151 """Attach the build-time routing context nested body steps route with."""
1152 self._contracts = contracts
1153 self._channel_names = channel_names
1155 @staticmethod
1156 def _should_skip_llm(step: dict, ctx: dict) -> bool:
1157 """Truthiness of a step's `skip_llm_if` expression (ENG-12, Q4).
1159 Lenient (non-strict) evaluation: a missing/undefined field resolves to
1160 falsy, so the step behaves normally (LLM invoked). Only an explicit
1161 truthy expression skips the LLM.
1162 """
1163 raw = step.get("skip_llm_if")
1164 if raw is None:
1165 return False
1166 resolved = _resolve_template(raw, ctx)
1167 if isinstance(resolved, bool):
1168 return resolved
1169 if isinstance(resolved, str):
1170 return resolved.lower() not in ('false', '0', '', 'none', 'null')
1171 return bool(resolved)
1173 @staticmethod
1174 def _skip_llm_passthrough(
1175 step: dict,
1176 state: PipelineState,
1177 ctx: dict,
1178 read_state: PipelineState,
1179 ) -> PipelineState:
1180 """Deterministic passthrough of a step's resolved input (no LLM).
1182 `state[step_id]` becomes the resolved `input` value (the parallel
1183 fan-out envelope `{results: [...]}` when the input resolves to a list).
1184 ``read_state`` is the working-set overlay the input resolves against;
1185 the returned state is still built from the raw ``state``.
1186 """
1187 step_id = step.get("id", "?")
1188 input_spec = step.get("input")
1189 resolved = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
1190 if isinstance(resolved, list):
1191 log.debug("[step] %s skip_llm_if passthrough (fanout, no LLM)", step_id)
1192 return {**state, step_id: {"results": resolved}}
1193 log.debug("[step] %s skip_llm_if passthrough (no LLM)", step_id)
1194 return {**state, step_id: resolved}
1196 @staticmethod
1197 def _parallel_fanout_items(
1198 step: dict, state: PipelineState, ctx: dict
1199 ) -> list | None:
1200 """The resolved fan-out item list for a parallel step, else None.
1202 A parallel step whose resolved input is a list invokes the agent once
1203 per item and stores a ``{"results": [...]}`` envelope instead of a
1204 single agent output.
1205 """
1206 if step.get("type") != "parallel":
1207 return None
1208 input_spec = step.get("input")
1209 if input_spec is None:
1210 return None
1211 input_val = _resolve_input(input_spec, state, ctx)
1212 return input_val if isinstance(input_val, list) else None
1214 def ran_agent(self, step: dict, state: PipelineState) -> bool:
1215 """Whether a step's run invokes its agent.
1217 False for agent-less steps, a ``skip_llm_if`` passthrough, and a
1218 parallel fan-out (per-item agent calls storing a result envelope).
1219 Channel routing is restricted to actual agent outputs, so these paths
1220 are excluded from inverse projection.
1221 """
1222 if self.resolve_agent(step) is None:
1223 return False
1224 ctx = self._make_ctx(state)
1225 if self._should_skip_llm(step, ctx):
1226 return False
1227 return self._parallel_fanout_items(step, state, ctx) is None
1229 def _make_ctx(self, state: dict, **extras: Any) -> dict:
1230 """Rakentaa template-kontekstin: state + run_context + extras."""
1231 ctx: dict = {}
1232 ctx.update(state)
1233 ctx.update(self._run_context)
1234 ctx.update(extras)
1235 return ctx
1237 @staticmethod
1238 def _read_state(
1239 state: PipelineState, working_set: ChannelWorkingSet | None
1240 ) -> PipelineState:
1241 """Overlay the invocation working set's read view onto the local state.
1243 The working set carries the committed snapshot plus this invocation's
1244 pending writes; the local state contributes keys the snapshot does not
1245 hold (item context, body step results). With no working set the raw
1246 state is the read view, preserving legacy behaviour exactly. The
1247 engine-local iteration keys are re-asserted after the overlay so a
1248 declared channel with the same name cannot shadow them.
1249 """
1250 if working_set is None:
1251 return state
1252 resolved = {**state, **working_set.read_view()}
1253 for key in _ENGINE_LOCAL_ITERATION_KEYS:
1254 if key in state:
1255 resolved[key] = state[key]
1256 return resolved
1258 def _run_body_step(
1259 self,
1260 step: dict,
1261 state: PipelineState,
1262 working_set: ChannelWorkingSet | None,
1263 ) -> PipelineState:
1264 """Run one nested body step and stage its declared channel writes.
1266 The two-argument call preserves the legacy recursion signature for the
1267 no-channels path (and for callers that monkey-patch ``run_step``).
1268 """
1269 if working_set is None:
1270 return self.run_step(step, state)
1271 result = self.run_step(step, state, working_set)
1272 self._stage_body_channel_writes(step, state, result, working_set)
1273 return result
1275 def _stage_body_channel_writes(
1276 self,
1277 step: dict,
1278 pre_state: PipelineState,
1279 result_state: PipelineState,
1280 working_set: ChannelWorkingSet,
1281 ) -> None:
1282 """Stage a body step's declared channel writes into the working set.
1284 The router early-outs for agent-less steps (builtins, call bind/emit);
1285 a step carrying an agent routes its declared writes.
1286 """
1287 if not self._channel_names:
1288 return
1289 writes = _route_declared_channels(
1290 step,
1291 self._read_state(pre_state, working_set),
1292 result_state,
1293 self,
1294 self._contracts,
1295 self._channel_names,
1296 )
1297 for name, value in writes.items():
1298 working_set.stage(name, value)
1300 def _resolve_step_runtime(
1301 self,
1302 step: dict,
1303 state: dict,
1304 ctx: dict,
1305 *,
1306 input_override: Any = _UNSET,
1307 extra_params: dict | None = None,
1308 ) -> dict:
1309 """
1310 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
1311 templaattien osalta state + run_context -kontekstia vasten.
1313 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
1314 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
1315 """
1316 resolved: dict = dict(step)
1318 if input_override is not _UNSET:
1319 resolved["input"] = input_override
1320 else:
1321 input_spec = step.get("input")
1322 if input_spec is not None:
1323 resolved["input"] = _resolve_input(input_spec, state, ctx)
1325 params_spec = step.get("params")
1326 if params_spec or extra_params:
1327 base = _resolve_template(params_spec, ctx) if params_spec else {}
1328 if not isinstance(base, dict):
1329 base = {}
1330 if extra_params:
1331 base = {**base, **extra_params}
1332 resolved["params"] = base
1334 return resolved
1336 def _attach_communication_context(
1337 self, resolved_spec: dict, resolved_step: dict, read_state: PipelineState
1338 ) -> dict:
1339 """Attach a step's projected COMMUNICATION context when declared.
1341 The projection is resolved from the invocation read overlay (the
1342 committed snapshot plus this invocation's pending writes) and carries
1343 only the agent contract's declared readable channels — never whole
1344 state. A step without a read declaration is returned unchanged.
1345 """
1346 agent_spec = resolved_spec.get("spec") or {}
1347 communication = resolve_communication_context(
1348 agent_spec, self._contracts, read_state
1349 )
1350 if not communication:
1351 return resolved_step
1352 return {**resolved_step, COMMUNICATION_CONTEXT_KEY: communication}
1354 def _attach_usage_instructions(self, resolved_step: dict) -> dict:
1355 """Attach the resolved usage-instruction entries for a step's `allow`.
1357 The whitelist is implicit: the instruction block is derived from the
1358 step's `allow` entries, resolved builtin-first, and carried to the
1359 invoker (ADR-0036 D1 element 5). A step without an `allow` whitelist
1360 is returned unchanged.
1361 """
1362 entries = resolve_usage_instructions(
1363 resolved_step.get("allow"), self._builtins, self._step_functions
1364 )
1365 if not entries:
1366 return resolved_step
1367 return {**resolved_step, USAGE_INSTRUCTIONS_KEY: entries}
1369 @staticmethod
1370 def _dummy_invoke(
1371 resolved_spec: dict, step: dict, state: PipelineState
1372 ) -> PipelineState:
1373 """Placeholder — ei kutsu oikeaa LLM:ää."""
1374 step_id = step.get("id", "?")
1375 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
1376 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
1378 @staticmethod
1379 def _merge_errors(base: list, additions: list) -> list:
1380 """Return a new list containing all items from *base* plus any items from
1381 *additions* that are not already present. Order is preserved and no
1382 in-place mutation occurs — safe to call from any context."""
1383 if not additions:
1384 return base
1385 merged = list(base)
1386 for err in additions:
1387 if err not in merged:
1388 merged.append(err)
1389 return merged
1391 @staticmethod
1392 def _wrap_hard_failure_gate(invoke_fn: Any) -> Any:
1393 """Kietoo invokerin: jokaiseen agent-step-tulokseen sovelletaan
1394 kova-virhe-portti (RunHardFailure details.error_code-perusteella).
1395 Suppein funnel joka näkee KAIKKI agent-tulokset: sequential,
1396 parallel (single + fanout), foreach (body + agent-polku) ja while."""
1397 def invoke_with_gate(
1398 resolved_spec: dict, step: dict, state: PipelineState
1399 ) -> PipelineState:
1400 result = invoke_fn(resolved_spec, step, state)
1401 step_id = step.get("id", "?")
1402 StepRunner._raise_on_hard_error(result.get(step_id), step_id)
1403 return result
1404 return invoke_with_gate
1406 def _wrap_calls_gate(self, invoke_fn: Any) -> Any:
1407 """Kietoo invokerin: agentin pyytämät deterministic-step kutsut
1408 (envelopen `calls`) suoritetaan, kun hard-failure-portti on päästänyt
1409 tuloksen läpi ja ennen kuin step-tulos tallennetaan/reititetään."""
1410 def invoke_with_calls(
1411 resolved_spec: dict, step: dict, state: PipelineState
1412 ) -> PipelineState:
1413 result = invoke_fn(resolved_spec, step, state)
1414 self._execute_requested_calls(step, result, state)
1415 return result
1416 return invoke_with_calls
1418 def _execute_requested_calls(
1419 self, step: dict, result: PipelineState, state: PipelineState
1420 ) -> None:
1421 """Suorita agentin envelopessa pyytämät deterministic-step kutsut.
1423 No-op ellei tulos ole envelope, jossa on ei-tyhjä `calls`-lista. Kutsut
1424 suoritetaan järjestyksessä builtin-then-StepFunction-polkua pitkin;
1425 onnistunut kutsu on side-effect-only (palautettu state hylätään).
1426 """
1427 step_id = step.get("id", "?")
1428 envelope = result.get(step_id) if isinstance(result, dict) else None
1429 calls = envelope.get("calls") if isinstance(envelope, dict) else None
1430 if not calls:
1431 return
1432 allow = step.get("allow") or []
1433 for index, call in enumerate(calls):
1434 self._execute_requested_call(step_id, allow, index, call, state)
1436 def _execute_requested_call(
1437 self,
1438 step_id: str,
1439 allow: list,
1440 index: int,
1441 call: Any,
1442 state: PipelineState,
1443 ) -> None:
1444 """Validoi ja suorita yksi pyydetty kutsu, fail-fast virhekoodeilla."""
1445 name = call.get("step") if isinstance(call, dict) else None
1446 if not isinstance(name, str) or name not in allow:
1447 raise RunHardFailure(
1448 "CALL_NOT_ALLOWED",
1449 f"step '{step_id}' requested deterministic step {name!r} which "
1450 "is not in its allow whitelist",
1451 {"step_id": step_id, "requested_step": name, "allow": list(allow)},
1452 )
1453 params = self._call_params(call, step_id, name)
1454 self._validate_call_params(name, params, step_id)
1455 call_step = self._synthesize_call_step(step_id, index, name, params, state)
1456 try:
1457 self._dispatch_call_step(name, call_step, state)
1458 except RunHardFailure as exc:
1459 if exc.code == "BUILTIN_NAME_COLLISION":
1460 raise
1461 raise RunHardFailure(
1462 "CALL_FAILED",
1463 f"step '{step_id}' call to '{name}' failed: {exc.message}",
1464 {
1465 "step_id": step_id,
1466 "requested_step": name,
1467 "error_code": exc.code,
1468 },
1469 ) from exc
1470 except Exception as exc: # noqa: BLE001
1471 raise RunHardFailure(
1472 "CALL_FAILED",
1473 f"step '{step_id}' call to '{name}' failed: {exc}",
1474 {"step_id": step_id, "requested_step": name},
1475 ) from exc
1477 @staticmethod
1478 def _call_params(call: dict, step_id: str, name: str) -> dict:
1479 """The call's params object ({} when absent), else CALL_INVALID_PARAMS."""
1480 params = call.get("params")
1481 if params is None:
1482 return {}
1483 if not isinstance(params, dict):
1484 raise RunHardFailure(
1485 "CALL_INVALID_PARAMS",
1486 f"step '{step_id}' call to '{name}' has non-object params",
1487 {"step_id": step_id, "requested_step": name},
1488 )
1489 return params
1491 def _validate_call_params(self, name: str, params: dict, step_id: str) -> None:
1492 """Validoi StepFunction-kutsun paramsit interfacea vasten.
1494 Builtineilla ei ole formaalia param-interfacea — niiden params-objekti
1495 on jo strukturaalisesti validoitu envelopessa. Säännöllinen
1496 StepFunction-suoritus säilyttää nykyisen käytöksensä; tämä portti
1497 koskee vain calls-polkua.
1498 """
1499 if self._builtins.has(name) or self._step_functions is None:
1500 return
1501 declared = self._step_functions.params_for(name)
1502 if declared is None:
1503 return
1504 declared_names = {entry["name"] for entry in declared}
1505 missing = [
1506 entry["name"]
1507 for entry in declared
1508 if entry.get("required") and entry["name"] not in params
1509 ]
1510 unknown = [key for key in params if key not in declared_names]
1511 if not missing and not unknown:
1512 return
1513 raise RunHardFailure(
1514 "CALL_INVALID_PARAMS",
1515 f"step '{step_id}' call to '{name}' has invalid params "
1516 f"(missing required: {missing}; undeclared: {unknown})",
1517 {
1518 "step_id": step_id,
1519 "requested_step": name,
1520 "missing": missing,
1521 "unknown": unknown,
1522 },
1523 )
1525 @staticmethod
1526 def _synthesize_call_step(
1527 parent_step_id: str,
1528 index: int,
1529 name: str,
1530 params: dict,
1531 state: PipelineState,
1532 ) -> dict:
1533 """Rakentaa deterministisen synteettisen stepin yhdelle kutsulle.
1535 The id is unique per invocation: the engine-local iteration
1536 discriminator is included when present so concurrent foreach/parallel
1537 items get distinct ids, and a process-unique token guarantees
1538 uniqueness otherwise (and against equal discriminators). The id is
1539 not persisted — calls are side-effect-only.
1540 """
1541 discriminator = _iteration_discriminator(state)
1542 token = uuid.uuid4().hex
1543 suffix = f"{discriminator}.{token}" if discriminator else token
1544 return {
1545 "id": f"{parent_step_id}.call.{index}.{name}.{suffix}",
1546 "type": "builtin",
1547 "component": name,
1548 "params": params,
1549 }
1551 def _dispatch_call_step(
1552 self, name: str, step: dict, state: PipelineState
1553 ) -> None:
1554 """Suorita yksi deterministic-step builtin-then-StepFunction-järjestyksessä.
1556 Peilaa `run_step`-builtin-haaran resolutiöjärjestystä (ADR-0037 D2):
1557 builtin-rekisteri voittaa, ja nimi molemmissa rekistereissä on kova
1558 virhe. Palautettu state-snapshot hylätään — kutsu on side-effect-only.
1559 """
1560 is_builtin = self._builtins.has(name)
1561 is_step_function = (
1562 self._step_functions is not None and self._step_functions.has(name)
1563 )
1564 if is_builtin and is_step_function:
1565 raise RunHardFailure(
1566 "BUILTIN_NAME_COLLISION",
1567 f"component '{name}' is present in both the builtin register "
1568 "and the StepFunction registry (ADR-0037 — fail-fast)",
1569 {"component": name},
1570 )
1571 if is_step_function:
1572 self._step_functions.execute(name, step, state)
1573 return
1574 builtin_state = self._builtins.execute(name, step, state)
1575 self._raise_on_hard_error(builtin_state.get(step["id"]), step["id"])
1577 @staticmethod
1578 def _is_hard_fail(details: dict) -> bool:
1579 """Kova virhe vain kun: error_code on HARD_ERROR_CODESissa
1580 (TRUNCATION_ERROR, TOOL_FAILURE, AGENT_ERROR) tai VALIDATION_ERROR
1581 kantaa kontekstiylityksen signatuurin (assembled_tokens +
1582 context_budget — _llm._context_overflow_result).
1583 Generic-VALIDATION_ERROR ei kaada ajoa. status:error-envelope kaatuu
1584 _raise_on_error_envelope-polulla ennen tätä tarkistusta."""
1585 code = details.get("error_code")
1586 if code in RunHardFailure.HARD_ERROR_CODES:
1587 return True
1588 if code == "VALIDATION_ERROR":
1589 return {"assembled_tokens", "context_budget"} <= details.keys()
1590 return False
1592 @staticmethod
1593 def _raise_on_hard_error(step_result: Any, step_id: str) -> None:
1594 """Nosta RunHardFailure jos step-tulos kantaa kovan virheen
1595 (_is_hard_fail) tai on status:error-envelope; stepin details
1596 kulkeutuu poikkeuksen mukana."""
1597 if not isinstance(step_result, dict):
1598 return
1599 if is_error_envelope(step_result):
1600 StepRunner._raise_on_error_envelope(step_result, step_id)
1601 return
1602 details = step_result.get("details")
1603 if not isinstance(details, dict):
1604 return
1605 if not StepRunner._is_hard_fail(details):
1606 return
1607 code = details.get("error_code")
1608 err = step_result.get("error", "")
1609 message = err if isinstance(err, str) and err else f"{code}: step '{step_id}' hard failure"
1610 raise RunHardFailure(code, message, details)
1612 @staticmethod
1613 def _raise_on_error_envelope(step_result: dict, step_id: str) -> None:
1614 """Nosta RunHardFailure status:error-envelopesta; envelopen error
1615 (dict) kulkeutuu details-attribuuttina, muuten minimi details."""
1616 envelope_error = step_result.get("error")
1617 if isinstance(envelope_error, dict):
1618 details = envelope_error
1619 message = envelope_error.get("message")
1620 else:
1621 details = {"error_code": ENVELOPE_ERROR_CODE, "step_id": step_id}
1622 message = None
1623 if not isinstance(message, str) or not message:
1624 message = f"{ENVELOPE_ERROR_CODE}: step '{step_id}' envelope error"
1625 raise RunHardFailure(ENVELOPE_ERROR_CODE, message, details)
1627 @staticmethod
1628 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
1629 """
1630 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
1631 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
1632 epäonnistuneeksi eikä hiljennä virhettä.
1634 Kaikki virhetyypit propagoidaan:
1635 - budget:max_llm_calls / budget:run_timeout → budjettiloki
1636 - parse_error:*, timeout:*, ja muut → error-loki
1637 """
1638 step_result = result.get(step_id)
1639 if not isinstance(step_result, dict):
1640 return result
1641 err = step_result.get("error", "")
1642 if not isinstance(err, str) or not err:
1643 return result
1644 errors = list(result.get("errors") or [])
1645 message = f"{step_id}: {err}"
1646 if message not in errors:
1647 errors.append(message)
1648 if err.startswith("budget:"):
1649 log.warning("[budget] %s", message)
1650 else:
1651 log.error("[step-error] %s", message)
1652 return {**result, "errors": errors}
1654 def resolve_agent(self, step: dict) -> dict | None:
1655 """Resolve the agent/instance a step runs, or None for agent-less steps."""
1656 agent_name = step.get("agent")
1657 if not agent_name:
1658 return None
1659 return self._resolver.resolve(agent_name)
1661 def run_step(
1662 self,
1663 step: dict,
1664 state: PipelineState,
1665 working_set: ChannelWorkingSet | None = None,
1666 ) -> PipelineState:
1667 step_id = step.get("id", "?")
1668 read_state = self._read_state(state, working_set)
1669 ctx = self._make_ctx(read_state)
1670 step_type = step.get("type", "sequential")
1672 # Evaluate condition — if false, skip step. A `while` whose entry
1673 # condition is false writes a first-class skip result (r.68) instead of
1674 # returning state unchanged, so `_run_while` owns its skip path.
1675 cond = step.get("condition")
1676 if cond is not None and not _eval_condition(cond, ctx) and step_type != "while":
1677 log.debug("[step] %s skipped (condition false)", step_id)
1678 return state
1680 if step_type == "builtin":
1681 component = step.get("component", "passthrough")
1682 log.debug("[step] %s builtin:%s", step_id, component)
1683 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
1684 # template-arvot (esim. "{{ budget.max_parallel }}")
1685 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1686 # ADR-0037 D2: resolve the component against the platform builtin
1687 # register FIRST, then the kind: StepFunction registry. A name in
1688 # both is a hard error (fail-fast).
1689 is_builtin = self._builtins.has(component)
1690 is_step_function = (
1691 self._step_functions is not None
1692 and self._step_functions.has(component)
1693 )
1694 if is_builtin and is_step_function:
1695 raise RunHardFailure(
1696 "BUILTIN_NAME_COLLISION",
1697 f"component '{component}' is present in both the builtin "
1698 "register and the StepFunction registry (ADR-0037 — "
1699 "fail-fast)",
1700 {"component": component},
1701 )
1702 if is_step_function:
1703 return self._step_functions.execute(component, resolved_step, state)
1704 builtin_state = self._builtins.execute(component, resolved_step, state)
1705 self._raise_on_hard_error(builtin_state.get(step_id), step_id)
1706 return builtin_state
1708 # skip_llm_if (ENG-12, Q4): when the step's skip_llm_if evaluates truthy,
1709 # do NOT invoke the LLM (no inference budget consumed) and set the step
1710 # result to a deterministic passthrough of the step's resolved input.
1711 if (
1712 step_type in ("sequential", "parallel")
1713 and self._should_skip_llm(step, ctx)
1714 ):
1715 return self._skip_llm_passthrough(step, state, ctx, read_state)
1717 if step_type == "parallel":
1718 # Fan-out: if input resolves to a list, call agent once per item
1719 items = self._parallel_fanout_items(step, read_state, ctx)
1720 if items is not None:
1721 return self._run_parallel_fanout(
1722 step, state, items, ctx, working_set
1723 )
1724 # Fallback: single agent call
1725 agent_name = step.get("agent")
1726 if not agent_name:
1727 log.warning("[step] %s has no agent — skipping", step_id)
1728 return state
1729 resolved = self._resolver.resolve(agent_name)
1730 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
1731 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1732 resolved_step = self._attach_communication_context(
1733 resolved, resolved_step, read_state
1734 )
1735 resolved_step = self._attach_usage_instructions(resolved_step)
1736 result = self._invoke_agent(resolved, resolved_step, state)
1737 return self._propagate_step_errors(result, step_id)
1739 if step_type == "sequential":
1740 agent_name = step.get("agent")
1741 if not agent_name:
1742 log.warning("[step] %s has no agent — skipping", step_id)
1743 return state
1744 resolved = self._resolver.resolve(agent_name)
1745 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
1746 resolved_step = self._resolve_step_runtime(step, read_state, ctx)
1747 resolved_step = self._attach_communication_context(
1748 resolved, resolved_step, read_state
1749 )
1750 resolved_step = self._attach_usage_instructions(resolved_step)
1751 result = self._invoke_agent(resolved, resolved_step, state)
1752 return self._propagate_step_errors(result, step_id)
1754 if step_type == "foreach":
1755 return self._run_foreach(step, state, working_set)
1757 if step_type == "while":
1758 return self._run_while(step, state, working_set)
1760 if step_type == "call":
1761 # Build-time expansion (ADR-0025 D4) is a later phase; a call step
1762 # reaching the runtime is wiring that is not yet expanded.
1763 raise ValueError(
1764 f"call step '{step_id}' reached runtime without build-time "
1765 "expansion (ADR-0025) — not yet supported"
1766 )
1768 if step_type == "spawn":
1769 return self._run_spawn(step, state, working_set)
1771 raise ValueError(f"Unknown step type '{step_type}' in step '{step_id}'")
1773 def _run_parallel_fanout(
1774 self,
1775 step: dict,
1776 state: PipelineState,
1777 items: list,
1778 ctx: dict,
1779 working_set: ChannelWorkingSet | None = None,
1780 ) -> PipelineState:
1781 """
1782 Fan-out: call agent once per item, concurrently.
1783 Results stored as state[step_id] = {"results": [item_result, ...]}.
1784 Each item is available as state["item"] inside the invoke call.
1785 """
1786 step_id = step.get("id", "?")
1787 agent_name = step.get("agent")
1788 if not agent_name:
1789 log.warning("[step] %s parallel fanout: no agent", step_id)
1790 return {**state, step_id: {"results": []}}
1792 resolved = self._resolver.resolve(agent_name)
1794 # Apply max_items cap
1795 max_items_raw = step.get("max_items")
1796 if max_items_raw is not None:
1797 max_items = _resolve_int(max_items_raw, ctx, len(items))
1798 items = items[:max_items]
1800 # Concurrency limit
1801 max_conc_raw = step.get("max_concurrent")
1802 if max_conc_raw is not None:
1803 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
1804 else:
1805 max_conc = len(items) or 1
1806 max_conc = max(1, min(max_conc, len(items) or 1))
1808 log.debug(
1809 "[step] %s parallel fanout: %d items, %d concurrent",
1810 step_id, len(items), max_conc,
1811 )
1813 read_state = self._read_state(state, working_set)
1815 def call_one(item: Any) -> tuple[Any, dict | None]:
1816 item_state = {**read_state, "_item": item, "item": item}
1817 item_ctx = self._make_ctx(item_state)
1818 resolved_step = self._resolve_step_runtime(
1819 step, item_state, item_ctx,
1820 input_override=item,
1821 )
1822 resolved_step = self._attach_communication_context(
1823 resolved, resolved_step, item_state
1824 )
1825 resolved_step = self._attach_usage_instructions(resolved_step)
1826 result_state = self._invoke_agent(resolved, resolved_step, item_state)
1827 return result_state.get(step_id), result_state.get("step_context_metrics")
1829 # Pre-allocate results to preserve input order regardless of completion
1830 # order (ENG-07), matching the concurrent foreach body-polku.
1831 results: list = [None] * len(items)
1832 item_metrics: dict = {}
1833 with ThreadPoolExecutor(max_workers=max_conc) as pool:
1834 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
1835 for future in as_completed(futures):
1836 idx = futures[future]
1837 try:
1838 item_result, item_metrics_map = future.result()
1839 results[idx] = item_result
1840 if isinstance(item_metrics_map, dict):
1841 item_metrics.update(item_metrics_map)
1842 except SkillLoadError:
1843 raise
1844 except RunHardFailure:
1845 pool.shutdown(wait=False, cancel_futures=True)
1846 raise
1847 except Exception as exc: # noqa: BLE001
1848 log.error("[step] %s item failed: %s", step_id, exc)
1849 results[idx] = {"error": str(exc)}
1851 self._stage_fanout_channel_writes(resolved, results, step_id, working_set)
1853 final: PipelineState = {**state, step_id: {"results": results}}
1854 if item_metrics:
1855 final = {
1856 **final,
1857 "step_context_metrics": _merge_metrics(
1858 state.get("step_context_metrics"), item_metrics
1859 ),
1860 }
1861 return final
1863 def _stage_fanout_channel_writes(
1864 self,
1865 agent_spec: dict,
1866 item_outputs: list,
1867 step_id: str,
1868 working_set: ChannelWorkingSet | None,
1869 ) -> None:
1870 """Stage each fan-out item's declared channel writes in item order.
1872 Runs on the main thread after the item pool completes, so the shared
1873 working set is never mutated from a worker thread. The fan-out
1874 envelope itself is never routed; only each item's own agent output is.
1875 """
1876 if working_set is None or not self._channel_names:
1877 return
1878 for output in item_outputs:
1879 writes = _declared_output_channel_writes(
1880 agent_spec, output, self._contracts, self._channel_names, step_id
1881 )
1882 for name, value in writes.items():
1883 working_set.stage(name, value)
1885 def _run_foreach(
1886 self,
1887 step: dict,
1888 state: PipelineState,
1889 working_set: ChannelWorkingSet | None = None,
1890 ) -> PipelineState:
1891 """
1892 Foreach: iterate over input list, run body steps per item.
1893 Input from step.input (dot-path or template) or step.over (state key).
1895 When max_concurrent > 1, body iterations run concurrently via
1896 ThreadPoolExecutor — each item gets its own worker thread. The
1897 max_concurrent value sets the thread pool size (mirrors parallel
1898 fan-out semantics). Results list preserves original item order.
1900 max_output budget in concurrent mode: a stop flag prevents new workers
1901 from starting after the budget is exhausted. Workers already executing
1902 when the flag is set will complete naturally — at most max_concurrent
1903 extra items may run past the limit. This is accepted behaviour; the
1904 sequential early-exit guarantee cannot be reproduced without sacrificing
1905 true concurrency.
1907 The invocation working set is threaded into every body step so reads
1908 resolve over the committed snapshot plus this invocation's pending
1909 writes; the returned state is still built from the raw ``state``.
1910 """
1911 step_id = step.get("id", "?")
1912 read_state = self._read_state(state, working_set)
1913 ctx = self._make_ctx(read_state)
1915 # Determine items
1916 input_spec = step.get("input")
1917 over_key = step.get("over")
1919 if input_spec is not None:
1920 items: list = _resolve_input(input_spec, read_state, ctx) or []
1921 if not isinstance(items, list):
1922 items = [items] if items is not None else []
1923 elif over_key:
1924 items = read_state.get(over_key, [None])
1925 if not isinstance(items, list):
1926 items = [items]
1927 else:
1928 items = [None]
1930 # max_iterations cap
1931 max_iter_raw = step.get("max_iterations")
1932 if max_iter_raw is not None:
1933 max_iter = _resolve_int(max_iter_raw, ctx, len(items))
1934 items = items[:max_iter]
1936 # max_output cap — stop once N iterations have produced a successful
1937 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
1938 max_output_raw = step.get("max_output")
1939 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
1941 # max_concurrent — when > 1, enables parallel body execution
1942 max_conc_raw = step.get("max_concurrent")
1943 if max_conc_raw is not None:
1944 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
1945 max_conc = max(1, min(max_conc, len(items) or 1))
1946 else:
1947 max_conc = 1
1949 body = step.get("body") or []
1950 agent_name = step.get("agent")
1952 if max_conc <= 1:
1953 return self._run_foreach_sequential(
1954 step_id, step, state, items, body, agent_name, max_output,
1955 working_set,
1956 )
1957 return self._run_foreach_concurrent(
1958 step_id, step, state, items, body, agent_name, max_output, max_conc,
1959 working_set,
1960 )
1962 @staticmethod
1963 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
1964 """body-polku per-iteration result (r.67): the declared body-output
1965 step's result; else the iteration's `publish` result; else None."""
1966 if declared_output:
1967 return iter_state.get(declared_output)
1968 return iter_state.get("publish")
1970 @staticmethod
1971 def _body_item_passed(item: Any) -> bool:
1972 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
1973 `passed` bool; an item lacking such a key is not counted as failed."""
1974 if isinstance(item, dict):
1975 verdict = item.get("verdict")
1976 if verdict == "FAIL":
1977 return False
1978 if verdict == "PASS":
1979 return True
1980 passed = item.get("passed")
1981 if isinstance(passed, bool):
1982 return passed
1983 return True
1985 @staticmethod
1986 def _aggregate_body_results(item_results: list) -> dict:
1987 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
1988 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
1989 return {
1990 "results": item_results,
1991 "passed": not failing,
1992 "failing": failing,
1993 "failed": bool(failing),
1994 }
1996 def _run_foreach_sequential(
1997 self,
1998 step_id: str,
1999 step: dict,
2000 state: PipelineState,
2001 items: list,
2002 body: list,
2003 agent_name: str | None,
2004 max_output: int,
2005 working_set: ChannelWorkingSet | None = None,
2006 ) -> PipelineState:
2007 """Sequential foreach — default path when max_concurrent is absent or 1."""
2008 body_polku = bool(body)
2009 declared_output = step.get("output")
2010 results: list = []
2011 produced_outputs = 0
2012 accumulated_errors: list = list(state.get("errors") or [])
2013 accumulated_metrics: dict = {}
2015 for item in items:
2016 if max_output and produced_outputs >= max_output:
2017 log.info("[foreach] %s max_output=%d reached — stopping", step_id, max_output)
2018 break
2019 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2021 if body_polku:
2022 for sub_step in body:
2023 iter_state = self._run_body_step(sub_step, iter_state, working_set)
2024 # Propagate any budget errors from this iteration to accumulated list
2025 accumulated_errors = self._merge_errors(
2026 accumulated_errors, iter_state.get("errors") or []
2027 )
2028 publish_out = iter_state.get("publish")
2029 results.append(self._body_item_result(iter_state, declared_output))
2030 if isinstance(publish_out, dict) and publish_out.get("path"):
2031 produced_outputs += 1
2032 elif agent_name:
2033 resolved = self._resolver.resolve(agent_name)
2034 iter_read_state = self._read_state(iter_state, working_set)
2035 iter_ctx = self._make_ctx(iter_read_state)
2036 resolved_step = self._resolve_step_runtime(
2037 step, iter_read_state, iter_ctx
2038 )
2039 resolved_step = self._attach_communication_context(
2040 resolved, resolved_step, iter_read_state
2041 )
2042 resolved_step = self._attach_usage_instructions(resolved_step)
2043 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2044 results.append(iter_state.get(step_id))
2045 item_metrics_map = iter_state.get("step_context_metrics")
2046 if isinstance(item_metrics_map, dict):
2047 accumulated_metrics.update(item_metrics_map)
2049 step_result: Any = self._aggregate_body_results(results) if body_polku else results
2050 final: PipelineState = {**state, step_id: step_result}
2051 if accumulated_metrics:
2052 final = {
2053 **final,
2054 "step_context_metrics": _merge_metrics(
2055 state.get("step_context_metrics"), accumulated_metrics
2056 ),
2057 }
2058 if accumulated_errors != list(state.get("errors") or []):
2059 final = {**final, "errors": accumulated_errors}
2060 return final
2062 @staticmethod
2063 def _merge_item_working_sets(
2064 working_set: ChannelWorkingSet | None,
2065 item_working_sets: list[ChannelWorkingSet | None],
2066 ) -> None:
2067 """Stage each concurrent item's writes onto the shared set in item order."""
2068 if working_set is None:
2069 return
2070 for item_working_set in item_working_sets:
2071 if item_working_set is None:
2072 continue
2073 for name, value in item_working_set.pending_writes():
2074 working_set.stage(name, value)
2076 def _run_foreach_concurrent(
2077 self,
2078 step_id: str,
2079 step: dict,
2080 state: PipelineState,
2081 items: list,
2082 body: list,
2083 agent_name: str | None,
2084 max_output: int,
2085 max_conc: int,
2086 working_set: ChannelWorkingSet | None = None,
2087 ) -> PipelineState:
2088 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
2089 log.info(
2090 "[foreach] %s concurrent: %d items, max_concurrent=%d",
2091 step_id, len(items), max_conc,
2092 )
2094 # Pre-allocate results list to preserve item order across threads.
2095 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
2096 # legitimate None results (body produced no output) are kept.
2097 body_polku = bool(body)
2098 declared_output = step.get("output")
2099 _SKIPPED = object()
2100 results: list = [_SKIPPED] * len(items)
2101 item_working_sets: list[ChannelWorkingSet | None] = [None] * len(items)
2102 lock = threading.Lock()
2103 produced_outputs = [0] # mutable counter shared across threads
2104 stop_flag = threading.Event() # set when max_output budget is exhausted
2105 item_errors: list = [] # budget errors collected from worker threads
2106 accumulated_metrics: dict = {} # per-item envelopes from worker threads
2108 def run_item(idx: int, item: Any) -> None:
2109 nonlocal item_errors
2110 # Honour stop flag — skip if budget already exhausted before we start
2111 if max_output and stop_flag.is_set():
2112 return
2114 item_working_set = (
2115 working_set.fork() if working_set is not None else None
2116 )
2117 item_working_sets[idx] = item_working_set
2118 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
2120 if body_polku:
2121 for sub_step in body:
2122 iter_state = self._run_body_step(
2123 sub_step, iter_state, item_working_set
2124 )
2125 # Collect budget errors from this iteration (thread-safe merge)
2126 new_errs = iter_state.get("errors") or []
2127 if new_errs:
2128 with lock:
2129 item_errors = self._merge_errors(item_errors, new_errs)
2130 publish_out = iter_state.get("publish")
2131 results[idx] = self._body_item_result(iter_state, declared_output)
2132 if isinstance(publish_out, dict) and publish_out.get("path"):
2133 with lock:
2134 produced_outputs[0] += 1
2135 if max_output and produced_outputs[0] >= max_output:
2136 stop_flag.set()
2137 elif agent_name:
2138 resolved = self._resolver.resolve(agent_name)
2139 item_read_state = self._read_state(iter_state, item_working_set)
2140 item_ctx = self._make_ctx(item_read_state)
2141 resolved_step = self._resolve_step_runtime(
2142 step, item_read_state, item_ctx
2143 )
2144 resolved_step = self._attach_communication_context(
2145 resolved, resolved_step, item_read_state
2146 )
2147 resolved_step = self._attach_usage_instructions(resolved_step)
2148 iter_state = self._invoke_agent(resolved, resolved_step, iter_state)
2149 results[idx] = iter_state.get(step_id)
2150 item_metrics_map = iter_state.get("step_context_metrics")
2151 if isinstance(item_metrics_map, dict):
2152 with lock:
2153 accumulated_metrics.update(item_metrics_map)
2155 with ThreadPoolExecutor(max_workers=max_conc) as pool:
2156 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
2157 for future in as_completed(futures):
2158 try:
2159 future.result()
2160 except SkillLoadError:
2161 raise
2162 except RunHardFailure:
2163 pool.shutdown(wait=False, cancel_futures=True)
2164 raise
2165 except Exception as exc: # noqa: BLE001
2166 idx = futures[future]
2167 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
2168 results[idx] = {"error": str(exc)}
2170 self._merge_item_working_sets(working_set, item_working_sets)
2172 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
2173 ordered = [r for r in results if r is not _SKIPPED]
2174 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
2175 final: PipelineState = {**state, step_id: step_result}
2176 if accumulated_metrics:
2177 final = {
2178 **final,
2179 "step_context_metrics": _merge_metrics(
2180 state.get("step_context_metrics"), accumulated_metrics
2181 ),
2182 }
2183 if item_errors:
2184 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
2185 return final
2187 def _run_while(
2188 self,
2189 step: dict,
2190 state: PipelineState,
2191 working_set: ChannelWorkingSet | None = None,
2192 ) -> PipelineState:
2193 """
2194 While loop with exit_condition and optional entry condition.
2196 Fields:
2197 condition: "{{ triage.approved }}" — if false, skip entirely
2198 input: initial value for loop.current
2199 exit_condition: "{{ review.pass }}" — break when true
2200 max_iterations: int or template
2201 output: body step id whose result feeds the next round's
2202 `loop.current` (envelope-unwrapped when dict with `output`)
2203 body: list of sub-steps
2205 Result shape (r.68): {passed, iterations, output, exhausted, skipped}.
2206 The feed comes ONLY from the declared `output` body step — there is no
2207 hardcoded `refine` feed; a body step named `refine` is not special.
2209 The invocation working set is threaded into every body step and into
2210 condition evaluation so reads resolve over the committed snapshot plus
2211 this invocation's pending writes; the returned state is still built
2212 from the raw ``state``.
2213 """
2214 step_id = step.get("id", "?")
2215 read_state = self._read_state(state, working_set)
2216 ctx = self._make_ctx(read_state)
2218 # Entry condition — skip is a first-class state (r.68): a skipped
2219 # while writes a result with output = the resolved input so downstream
2220 # `.output` reads are non-null on the skipped path.
2221 entry_cond = step.get("condition")
2222 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
2223 input_spec = step.get("input")
2224 skip_input = _resolve_input(input_spec, read_state, ctx) if input_spec is not None else None
2225 return {**state, step_id: {
2226 "passed": True,
2227 "iterations": 0,
2228 "output": skip_input,
2229 "exhausted": False,
2230 "skipped": True,
2231 }}
2233 # Initial current value from input
2234 input_spec = step.get("input")
2235 if input_spec is not None:
2236 current = _resolve_input(input_spec, read_state, ctx)
2237 else:
2238 current = None
2240 exit_condition = step.get("exit_condition")
2241 max_iter = _resolve_int(step.get("max_iterations", 2), ctx, 2)
2242 body = step.get("body") or []
2243 declared_output = step.get("output")
2245 passed = False
2246 iteration = 0
2247 loop_state: PipelineState = dict(state)
2249 while iteration < max_iter:
2250 # Inject loop context
2251 loop_state = {
2252 **state,
2253 "loop": {"current": current, "iteration": iteration},
2254 step_id: {"current": current, "passed": False},
2255 }
2256 # Add current item context (from outer foreach if any)
2257 if "item" in state:
2258 loop_state["item"] = state["item"]
2260 # Run body steps
2261 for sub_step in body:
2262 loop_state = self._run_body_step(sub_step, loop_state, working_set)
2264 iteration += 1
2266 # Feed: next round's loop.current comes from the declared body-output
2267 # step (envelope-unwrapped); otherwise current is unchanged.
2268 # Envelopes: a `{output: ...}` (r.68) and a `{results: [...]}`
2269 # aggregate (parallel/foreach body-step) are unwrapped to their
2270 # inner value so the next round's consumer receives the list, not
2271 # the envelope. A raw value (plain list/dict) feeds as-is.
2272 if declared_output:
2273 feed_value = loop_state.get(declared_output)
2274 if isinstance(feed_value, dict) and "output" in feed_value:
2275 current = feed_value["output"]
2276 elif isinstance(feed_value, dict) and isinstance(feed_value.get("results"), list):
2277 current = feed_value["results"]
2278 else:
2279 current = feed_value
2281 # Check exit condition
2282 if exit_condition:
2283 exit_ctx = self._make_ctx(
2284 self._read_state(loop_state, working_set)
2285 )
2286 if _eval_condition(exit_condition, exit_ctx):
2287 passed = True
2288 break
2289 else:
2290 passed = True
2291 break
2293 skipped = False
2294 exhausted = not passed
2295 final: PipelineState = {**state, step_id: {
2296 "passed": passed,
2297 "iterations": iteration,
2298 "output": current,
2299 "exhausted": exhausted,
2300 "skipped": skipped,
2301 }}
2302 # ENG-11: a while that ends exhausted (ran max_iterations without
2303 # exit_condition passing, not skipped) and whose `fail_if_exhausted`
2304 # evaluates true is a hard run failure — nothing downstream runs. A
2305 # skipped or passed while must never raise.
2306 fail_if_exhausted = step.get("fail_if_exhausted")
2307 if (
2308 fail_if_exhausted is not None
2309 and exhausted
2310 and _eval_condition(fail_if_exhausted, self._make_ctx(read_state))
2311 ):
2312 raise RunHardFailure(
2313 "EXHAUSTED",
2314 f"while step '{step_id}' exhausted max_iterations={max_iter} "
2315 "without exit_condition passing (fail_if_exhausted)",
2316 {"step_id": step_id, "exhausted": True},
2317 )
2318 # Propagate any errors accumulated in the loop body back to the outer state
2319 loop_errors = loop_state.get("errors") or []
2320 if loop_errors:
2321 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), loop_errors)}
2322 return final
2324 def _run_spawn(
2325 self,
2326 step: dict,
2327 state: PipelineState,
2328 working_set: ChannelWorkingSet | None = None,
2329 ) -> PipelineState:
2330 """Run a spawn step: a recursive node tree of agent calls (S2).
2332 The `agent` runs once as director at the root and may emit a `children`
2333 plan; every deeper node re-runs the `child` role (default = `agent`)
2334 with that node's own `input`. Depth and fan are bounded by the
2335 spawn_policy, and the whole tree draws from one shared SpawnBudget.
2336 """
2337 step_id = step.get("id", "?")
2338 read_state = self._read_state(state, working_set)
2339 ctx = self._make_ctx(read_state)
2340 policy = step.get("spawn_policy") or {}
2342 max_depth_raw = policy.get("max_depth")
2343 if max_depth_raw is None:
2344 max_depth = _SPAWN_DEFAULT_MAX_DEPTH
2345 else:
2346 max_depth = _resolve_int(max_depth_raw, ctx, _SPAWN_DEFAULT_MAX_DEPTH)
2348 max_spawn_raw = policy.get("max_spawn_per_node")
2349 if max_spawn_raw is None:
2350 max_spawn_per_node = _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
2351 else:
2352 max_spawn_per_node = _resolve_int(max_spawn_raw, ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE)
2354 max_calls_raw = policy.get("max_agent_calls")
2355 if max_calls_raw is None:
2356 budget = SpawnBudget(_SPAWN_DEFAULT_MAX_AGENT_CALLS)
2357 else:
2358 budget = SpawnBudget(_resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS))
2360 run_max_parallel = int((self._run_context.get("budget") or {}).get("max_parallel", DEFAULT_MAX_PARALLEL))
2361 max_concurrent_raw = policy.get("max_concurrent_children")
2362 if max_concurrent_raw is None:
2363 max_concurrent_children = run_max_parallel
2364 else:
2365 max_concurrent_children = _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
2367 director = step.get("agent")
2368 if not director:
2369 log.warning("[step] %s spawn has no agent — skipping", step_id)
2370 return state
2371 child_role = step.get("child") or director
2373 envelope, metrics = self._spawn_node(
2374 node_id=step_id,
2375 role=director,
2376 depth=0,
2377 step=step,
2378 input_override=_UNSET,
2379 base_state=read_state,
2380 base_ctx=ctx,
2381 child_role=child_role,
2382 budget=budget,
2383 max_depth=max_depth,
2384 max_spawn_per_node=max_spawn_per_node,
2385 max_concurrent_children=max_concurrent_children,
2386 working_set=working_set,
2387 )
2388 final: PipelineState = {**state, step_id: envelope}
2389 if metrics:
2390 final = {
2391 **final,
2392 "step_context_metrics": _merge_metrics(
2393 state.get("step_context_metrics"), metrics
2394 ),
2395 }
2396 return final
2398 def _spawn_node(
2399 self,
2400 *,
2401 node_id: str,
2402 role: str,
2403 depth: int,
2404 step: dict,
2405 input_override: Any,
2406 base_state: PipelineState,
2407 base_ctx: dict,
2408 child_role: str,
2409 budget: SpawnBudget,
2410 max_depth: int,
2411 max_spawn_per_node: int | None,
2412 max_concurrent_children: int,
2413 working_set: ChannelWorkingSet | None,
2414 ) -> tuple[dict, dict]:
2415 """Run one spawn-tree node; returns (node envelope, subtree metrics)."""
2416 tracer = self._tracer
2417 parent_obs = (
2418 tracer.open_parent_observation(
2419 name=f"spawn:{node_id}",
2420 input={"node_id": node_id, "role": role, "depth": depth},
2421 )
2422 if tracer
2423 else None
2424 )
2425 try:
2426 resolved_role = self._resolver.resolve(role)
2427 resolved_step = self._resolve_step_runtime(
2428 step, base_state, base_ctx, input_override=input_override,
2429 )
2430 resolved_step = self._attach_communication_context(
2431 resolved_role, resolved_step, base_state
2432 )
2433 resolved_step = self._attach_usage_instructions(resolved_step)
2434 if not budget.consume():
2435 raise RunHardFailure(
2436 "SPAWN_BUDGET_EXHAUSTED",
2437 f"spawn tree budget exhausted at node '{node_id}'",
2438 {"node_id": node_id, "depth": depth},
2439 )
2440 result_state = self._invoke_agent(resolved_role, resolved_step, base_state)
2441 this_result = result_state.get(node_id)
2442 metrics = dict(result_state.get("step_context_metrics") or {})
2443 self._stage_spawn_channel_writes(
2444 resolved_role, this_result, node_id, working_set
2445 )
2447 children = self._spawn_children_from_result(this_result)
2448 envelope: dict = {
2449 "result": this_result,
2450 "children": {},
2451 "passed": not self._spawn_is_soft_fail(this_result),
2452 "failing": [],
2453 "exhausted": False,
2454 }
2456 if not children:
2457 return envelope, metrics
2458 if depth + 1 > max_depth:
2459 envelope["exhausted"] = True
2460 return envelope, metrics
2461 if max_spawn_per_node is not None and len(children) > max_spawn_per_node:
2462 raise RunHardFailure(
2463 "SPAWN_OVER_FLIGHT",
2464 f"spawn node '{node_id}' emitted {len(children)} children, "
2465 f"exceeds max_spawn_per_node={max_spawn_per_node}",
2466 {
2467 "node_id": node_id,
2468 "depth": depth,
2469 "children": len(children),
2470 "max_spawn_per_node": max_spawn_per_node,
2471 },
2472 )
2474 pool_size = self._spawn_pool_size(max_spawn_per_node, max_concurrent_children)
2475 if len(children) <= 1 or pool_size <= 1:
2476 for idx, child in enumerate(children):
2477 child_id, child_envelope, child_metrics = self._spawn_child(
2478 node_id=node_id,
2479 child_role=child_role,
2480 depth=depth,
2481 step=step,
2482 base_state=base_state,
2483 base_ctx=base_ctx,
2484 budget=budget,
2485 max_depth=max_depth,
2486 max_spawn_per_node=max_spawn_per_node,
2487 max_concurrent_children=max_concurrent_children,
2488 working_set=working_set,
2489 child=child,
2490 idx=idx,
2491 )
2492 metrics = _merge_metrics(metrics, child_metrics)
2493 envelope["children"][child_id] = child_envelope
2494 if not child_envelope["passed"]:
2495 envelope["passed"] = False
2496 envelope["failing"].append(child_id)
2497 return envelope, metrics
2499 pool = ThreadPoolExecutor(max_workers=pool_size)
2500 cancel = threading.Event()
2501 futures = []
2502 child_working_sets: dict[Any, ChannelWorkingSet] = {}
2503 for idx, child in enumerate(children):
2504 child_working_set = (
2505 working_set.fork() if working_set is not None else None
2506 )
2507 ctx = contextvars.copy_context()
2508 future = pool.submit(
2509 ctx.run,
2510 self._spawn_child,
2511 node_id=node_id,
2512 child_role=child_role,
2513 depth=depth,
2514 step=step,
2515 base_state=base_state,
2516 base_ctx=base_ctx,
2517 budget=budget,
2518 max_depth=max_depth,
2519 max_spawn_per_node=max_spawn_per_node,
2520 max_concurrent_children=max_concurrent_children,
2521 working_set=child_working_set,
2522 child=child,
2523 idx=idx,
2524 )
2525 futures.append(future)
2526 if child_working_set is not None:
2527 child_working_sets[future] = child_working_set
2528 hard_error: Exception | None = None
2529 child_results: dict[Any, tuple[str, dict, dict]] = {}
2530 try:
2531 for fut in as_completed(futures):
2532 try:
2533 child_results[fut] = fut.result()
2534 except (RunHardFailure, SkillLoadError) as exc:
2535 hard_error = exc
2536 cancel.set()
2537 for pending in futures:
2538 pending.cancel()
2539 break
2540 if hard_error is None:
2541 for fut in futures:
2542 child_id, child_envelope, child_metrics = child_results[fut]
2543 self._merge_spawn_fork_writes(
2544 working_set, child_working_sets.get(fut)
2545 )
2546 metrics = _merge_metrics(metrics, child_metrics)
2547 envelope["children"][child_id] = child_envelope
2548 if not child_envelope["passed"]:
2549 envelope["passed"] = False
2550 envelope["failing"].append(child_id)
2551 finally:
2552 pool.shutdown(wait=False, cancel_futures=True)
2553 if hard_error is not None:
2554 raise hard_error
2556 return envelope, metrics
2557 finally:
2558 if tracer is not None and parent_obs is not None:
2559 tracer.close_parent_observation(parent_obs)
2561 def _stage_spawn_channel_writes(
2562 self,
2563 resolved_role: dict,
2564 result: Any,
2565 node_id: str,
2566 working_set: ChannelWorkingSet | None,
2567 ) -> None:
2568 """Stage a spawn node's declared writes from its own agent result."""
2569 if working_set is None:
2570 return
2571 writes = _declared_output_channel_writes(
2572 resolved_role, result, self._contracts, self._channel_names, node_id
2573 )
2574 for name, value in writes.items():
2575 working_set.stage(name, value)
2577 @staticmethod
2578 def _merge_spawn_fork_writes(
2579 working_set: ChannelWorkingSet | None,
2580 fork: ChannelWorkingSet | None,
2581 ) -> None:
2582 """Fold a completed child's isolated fork writes into the parent set.
2584 Runs on the main thread after the child future resolves, so the shared
2585 working set is never mutated from a worker thread.
2586 """
2587 if working_set is None or fork is None:
2588 return
2589 for name, value in fork.pending_writes():
2590 working_set.stage(name, value)
2592 @staticmethod
2593 def _spawn_pool_size(max_spawn_per_node: int | None, max_concurrent_children: int) -> int:
2594 if max_spawn_per_node is None:
2595 return max(1, max_concurrent_children)
2596 return max(1, min(max_spawn_per_node, max_concurrent_children))
2598 def _spawn_child(
2599 self,
2600 *,
2601 node_id: str,
2602 child_role: str,
2603 depth: int,
2604 step: dict,
2605 base_state: PipelineState,
2606 base_ctx: dict,
2607 budget: SpawnBudget,
2608 max_depth: int,
2609 max_spawn_per_node: int | None,
2610 max_concurrent_children: int,
2611 working_set: ChannelWorkingSet | None,
2612 child: Any,
2613 idx: int,
2614 ) -> tuple[str, dict, dict]:
2615 if not self._spawn_has_input(child):
2616 raise RunHardFailure(
2617 "INVALID_CHILD_PLAN",
2618 f"spawn child #{idx} of node '{node_id}' must be a dict "
2619 "with a non-empty 'input'",
2620 {"node_id": node_id, "child_index": idx, "child": repr(child)[:200]},
2621 )
2622 child_id = child["id"] if child.get("id") else f"{node_id}.{idx}"
2623 child_input = child["input"]
2624 resolved_child_input = _resolve_input(child_input, base_state, base_ctx)
2625 child_envelope, child_metrics = self._spawn_node(
2626 node_id=child_id,
2627 role=child_role,
2628 depth=depth + 1,
2629 step={"id": child_id, "input": resolved_child_input},
2630 input_override=resolved_child_input,
2631 base_state=base_state,
2632 base_ctx=base_ctx,
2633 child_role=child_role,
2634 budget=budget,
2635 max_depth=max_depth,
2636 max_spawn_per_node=max_spawn_per_node,
2637 max_concurrent_children=max_concurrent_children,
2638 working_set=working_set,
2639 )
2640 return child_id, child_envelope, child_metrics
2642 @staticmethod
2643 def _spawn_children_from_result(result: Any) -> list:
2644 """The node's child plan: [] when absent or empty; raises on a malformed plan."""
2645 if not isinstance(result, dict):
2646 return []
2647 raw = result.get("children")
2648 if raw is None:
2649 return []
2650 if not isinstance(raw, list):
2651 raise RunHardFailure(
2652 "INVALID_CHILD_PLAN",
2653 "spawn agent 'children' must be a list",
2654 {"children": repr(raw)[:200]},
2655 )
2656 return raw
2658 @staticmethod
2659 def _spawn_has_input(child: Any) -> bool:
2660 """A valid child plan item: a dict carrying a non-empty 'input'."""
2661 if not isinstance(child, dict):
2662 return False
2663 input_val = child.get("input")
2664 if input_val is None:
2665 return False
2666 if isinstance(input_val, str) and not input_val.strip():
2667 return False
2668 if isinstance(input_val, (list, dict)) and not input_val:
2669 return False
2670 return True
2672 @staticmethod
2673 def _spawn_is_soft_fail(result: Any) -> bool:
2674 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error.
2676 An envelope-shaped result is never soft: ``status:error`` is a hard
2677 failure (the gate raises before this), and ``ok``/``partial``/
2678 ``no_success`` are valid non-fatal results.
2679 """
2680 if not isinstance(result, dict):
2681 return False
2682 if result.get("status") in ENVELOPE_STATUSES:
2683 return False
2684 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
2685 return True
2686 return any(key.startswith("budget:") for key in result)
2689# ---------------------------------------------------------------------------
2690# GraphBuilder
2691# ---------------------------------------------------------------------------
2693def _topological_order(steps: list[dict]) -> list[dict]:
2694 """
2695 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2696 """
2697 id_to_step = {s["id"]: s for s in steps}
2698 visited: set[str] = set()
2699 order: list[dict] = []
2701 def visit(step_id: str) -> None:
2702 if step_id in visited:
2703 return
2704 visited.add(step_id)
2705 for dep in (id_to_step[step_id].get("depends_on") or []):
2706 if dep in id_to_step:
2707 visit(dep)
2708 order.append(id_to_step[step_id])
2710 for s in steps:
2711 visit(s["id"])
2712 return order
2715# ---------------------------------------------------------------------------
2716# Call-step build-time expansion (ADR-0025 D3/D4)
2717# ---------------------------------------------------------------------------
2718#
2719# A `type: call` step is expanded at build time into the function's inlined body
2720# steps plus two synthesized op steps (`__call_bind` / `__call_emit`). The
2721# runtime executes only this flat graph; LangGraph never sees a call step. The
2722# caller reads the function outputs via the call result shape
2723# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
2724# The function body's `{{ params.<name> }}` references are rewritten to
2725# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves.
2726#
2727# Isolation (ADR-0025 D5 / 00-functions § Isolation) and recursion (D7/D9): a
2728# function body step must not reference a step-id outside its own body (a
2729# one-way dependency check — outward references to caller step-ids are build
2730# errors); a cycle in the function call graph (self or mutual) is a build
2731# error. Bounded while loops are runtime constructs and are not cycle-flagged.
2733# Call step result namespace keys (internal scaffolding).
2734def _scope_params_value(value: Any, call_id: str) -> Any:
2735 """Rewrite `params.` template references to the call-scoped namespace.
2737 Operates on the function body's string values (templates). Dict keys are
2738 untouched — only templated string VALUES are scoped, so a function body's
2739 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
2740 """
2741 if isinstance(value, str):
2742 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
2743 if isinstance(value, dict):
2744 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
2745 if isinstance(value, list):
2746 return [_scope_params_value(v, call_id) for v in value]
2747 return value
2750def _scope_params(step: dict, call_id: str) -> dict:
2751 """Deep-scope a function body step's params references for a call id."""
2752 scoped: dict = dict(step)
2753 for field in ("input", "params", "condition", "exit_condition", "agent", "function"):
2754 if field in scoped and isinstance(scoped[field], (str, dict, list)):
2755 scoped[field] = _scope_params_value(scoped[field], call_id)
2756 if "body" in scoped and isinstance(scoped["body"], list):
2757 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
2758 return scoped
2761def _collect_step_ids(steps: list[dict]) -> set[str]:
2762 """All step ids declared in *steps*, including nested body step ids."""
2763 ids: set[str] = set()
2764 for step in steps:
2765 step_id = step.get("id")
2766 if isinstance(step_id, str):
2767 ids.add(step_id)
2768 body = step.get("body")
2769 if isinstance(body, list):
2770 ids |= _collect_step_ids(body)
2771 return ids
2774def _iter_steps(steps: list[dict]):
2775 """Yield every step in *steps*, recursing into nested bodies."""
2776 for step in steps:
2777 yield step
2778 body = step.get("body")
2779 if isinstance(body, list):
2780 yield from _iter_steps(body)
2783_TEMPLATE_REF_RE = re.compile(r"\{\{\s*([^}]*)\}\}")
2784_STRING_LITERAL_RE = re.compile(r"'[^']*'|\"[^\"]*\"")
2785_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
2787_REFERENCE_FIELDS = ("input", "params", "condition", "exit_condition")
2789# Engine-reserved template namespaces — never step-id references, even if a
2790# caller happens to declare a step with one of these names.
2791_RESERVED_NAMESPACES = frozenset({"params", "loop", "item", "_item", "budget", "config"})
2794def _referenced_names(value: Any) -> set[str]:
2795 """Root identifiers read by templated strings.
2797 Only the leading identifier of each reference (before the first ``.`` or
2798 ``[``) can name a step-id; nested field segments are ignored. Quoted
2799 literals are ignored.
2800 """
2801 names: set[str] = set()
2802 if isinstance(value, str):
2803 for expr in _TEMPLATE_REF_RE.findall(value):
2804 bare = _STRING_LITERAL_RE.sub("", expr)
2805 for match in _IDENTIFIER_RE.finditer(bare):
2806 if match.start() == 0 or bare[match.start() - 1] != ".":
2807 names.add(match.group())
2808 elif isinstance(value, dict):
2809 for v in value.values():
2810 names |= _referenced_names(v)
2811 elif isinstance(value, list):
2812 for v in value:
2813 names |= _referenced_names(v)
2814 return names
2817def _validate_function_purity(
2818 fn_name: str, body: list[dict], visible_ids: set[str]
2819) -> None:
2820 """One-way isolation: a function body must not reference a caller step-id.
2822 A body step may read its own body step-ids and the params namespace; a
2823 reference to a step-id that exists outside the body (``visible_ids`` but
2824 not declared in the body) is an outward dependency and a hard build error.
2825 """
2826 local_ids = _collect_step_ids(body)
2827 for step in _iter_steps(body):
2828 step_id = step.get("id", "?")
2829 for dep in step.get("depends_on") or []:
2830 if dep in visible_ids and dep not in local_ids:
2831 raise ValueError(
2832 f"function '{fn_name}' step '{step_id}': depends_on "
2833 f"'{dep}' is outside the function body (isolation — "
2834 "function bodies see only their own steps and params)"
2835 )
2836 for field in _REFERENCE_FIELDS:
2837 for ref in _referenced_names(step.get(field)):
2838 if ref in _RESERVED_NAMESPACES:
2839 continue
2840 if ref in visible_ids and ref not in local_ids:
2841 raise ValueError(
2842 f"function '{fn_name}' step '{step_id}': '{field}' "
2843 f"references '{ref}', a step-id outside the function "
2844 "body (isolation — function bodies see only their own "
2845 "steps and params)"
2846 )
2849def _expand_call(
2850 step: dict,
2851 registry: ManifestRegistry,
2852 stack: list[str],
2853 call_map: dict[str, str],
2854 expanded_functions: dict[str, str],
2855 visible_ids: set[str],
2856) -> list[dict]:
2857 """Expand a call step into bind + inlined body + emit steps."""
2858 fn_name = step["function"]
2859 call_id = step["id"]
2861 if fn_name in stack:
2862 chain = " -> ".join(stack + [fn_name])
2863 raise ValueError(
2864 f"cycle detected in function expansion: {chain} "
2865 "(ADR-0025 D7/D9 — recursion must be a bounded while loop or idempotent re-run)"
2866 )
2868 # ADR-0025 D7: inlining the same function twice in one pipeline build
2869 # collides its body step-ids (flat graph). This is a hard build error until
2870 # id-prefixing lands; naming the function and both call sites pins the bug.
2871 if fn_name in expanded_functions:
2872 raise ValueError(
2873 f"duplicate call expansion: function '{fn_name}' called at both "
2874 f"'{expanded_functions[fn_name]}' and '{call_id}' in one pipeline — "
2875 "inlined body step ids collide (ADR-0025 D7)"
2876 )
2877 expanded_functions[fn_name] = call_id
2879 function = registry.get("Function", fn_name)
2880 if function is None:
2881 raise ValueError(
2882 f"call step '{call_id}' references unknown function '{fn_name}' "
2883 "(no kind: Function manifest in the workflows tree)"
2884 )
2886 body = (function.get("spec") or {}).get("steps") or []
2887 _validate_function_purity(fn_name, body, visible_ids)
2889 sub_stack = list(stack) + [fn_name]
2890 nested_visible = visible_ids | _collect_step_ids(body)
2891 outputs_spec = ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
2893 bind_id = f"_call__{call_id}__bind"
2894 emit_id = f"_call__{call_id}__emit"
2895 call_map[call_id] = emit_id
2897 bind_step: dict = {
2898 "id": bind_id,
2899 "type": "builtin",
2900 "component": "__call_bind",
2901 "params": {**(step.get("params") or {}), "__call_id": call_id},
2902 "depends_on": list(step.get("depends_on") or []),
2903 }
2905 expanded_body: list[dict] = []
2906 for body_step in body:
2907 scoped = _scope_params(body_step, call_id)
2908 for sub in _expand_step(
2909 scoped, registry, sub_stack, call_map, expanded_functions, nested_visible
2910 ):
2911 dep = list(sub.get("depends_on") or [])
2912 if bind_id not in dep:
2913 sub["depends_on"] = [bind_id, *dep]
2914 expanded_body.append(sub)
2916 emit_step: dict = {
2917 "id": emit_id,
2918 "type": "builtin",
2919 "component": "__call_emit",
2920 "params": {"__call_id": call_id, "__outputs": outputs_spec},
2921 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
2922 }
2924 return [bind_step, *expanded_body, emit_step]
2927def _expand_step(
2928 step: dict,
2929 registry: ManifestRegistry,
2930 stack: list[str],
2931 call_map: dict[str, str],
2932 expanded_functions: dict[str, str],
2933 visible_ids: set[str],
2934) -> list[dict]:
2935 """Expand one step (recursing into while/foreach bodies)."""
2936 if step.get("type") == "call":
2937 return _expand_call(
2938 step, registry, stack, call_map, expanded_functions, visible_ids
2939 )
2940 if "body" in step and isinstance(step["body"], list):
2941 expanded_body: list[dict] = []
2942 for sub in step["body"]:
2943 expanded_body.extend(
2944 _expand_step(
2945 sub, registry, stack, call_map, expanded_functions, visible_ids
2946 )
2947 )
2948 return [{**step, "body": expanded_body}]
2949 return [step]
2952def _rewrite_call_deps(steps: list[dict], call_map: dict[str, str]) -> None:
2953 """Rewrite `depends_on` references to call ids → the corresponding emit id.
2955 After a call is expanded away, a downstream step that depends_on the call id
2956 must instead wait on the emit step (which writes state[<call_id>]). This
2957 guarantees the caller's `{{ <call_id>.outputs.<name> }}` reads resolve after
2958 the function body ran.
2959 """
2960 for step in steps:
2961 deps = step.get("depends_on") or []
2962 if deps:
2963 step["depends_on"] = [call_map.get(d, d) for d in deps]
2964 if "body" in step and isinstance(step["body"], list):
2965 _rewrite_call_deps(step["body"], call_map)
2968def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
2969 """Expand all `call` steps into the flat runtime graph (ADR-0025 D4).
2971 Returns a new step list with each call replaced by its bind + inlined body +
2972 emit steps; a call step never survives to the runtime graph. Cycles and
2973 outward references in a function body are hard errors.
2974 """
2975 call_map: dict[str, str] = {}
2976 expanded_functions: dict[str, str] = {}
2977 visible_ids = _collect_step_ids(steps)
2978 expanded: list[dict] = []
2979 for step in steps:
2980 expanded.extend(
2981 _expand_step(step, registry, [], call_map, expanded_functions, visible_ids)
2982 )
2983 _rewrite_call_deps(expanded, call_map)
2984 return expanded
2987_CHANNEL_TYPES = frozenset({"LastValue", "Topic", "BinaryOperatorAggregate"})
2990def _validate_channel_entry(index: int, channel: Any) -> None:
2991 """Validate one spec.channels entry against the frozen ChannelSpec shape.
2993 The engine build path does not run manifest-schema validation, so a
2994 malformed entry is rejected here — before any ``["name"]`` indexing — with
2995 an error that identifies the offending entry.
2996 """
2997 if not isinstance(channel, dict):
2998 raise ValueError(
2999 f"Pipeline spec.channels[{index}] must be a mapping; got "
3000 f"{type(channel).__name__}"
3001 )
3002 name = channel.get("name")
3003 if not isinstance(name, str) or not name:
3004 raise ValueError(
3005 f"Pipeline spec.channels[{index}] is missing a valid 'name'; each "
3006 "channel requires a non-empty string 'name'"
3007 )
3008 if "type" in channel and channel["type"] not in _CHANNEL_TYPES:
3009 raise ValueError(
3010 f"Pipeline spec.channels[{index}] ('{name}') has invalid type "
3011 f"'{channel['type']}'; expected one of {sorted(_CHANNEL_TYPES)}"
3012 )
3015def _validated_channels(raw_channels: Any) -> list[dict]:
3016 """Return spec.channels as a validated list, or [] when absent.
3018 ``None`` (absent) and an empty list are both valid no-channel pipelines.
3019 """
3020 if raw_channels is None:
3021 return []
3022 if not isinstance(raw_channels, list):
3023 raise ValueError(
3024 f"Pipeline spec.channels must be a list; got "
3025 f"{type(raw_channels).__name__}"
3026 )
3027 for index, channel in enumerate(raw_channels):
3028 _validate_channel_entry(index, channel)
3029 return raw_channels
3032def _index_data_contracts(raw_contracts: Any) -> dict[str, dict]:
3033 """Index spec.data_contracts by name; absent/empty yields an empty map.
3035 The engine build path does not run manifest-schema validation, so a
3036 malformed entry is rejected here with an error that identifies it.
3037 """
3038 if raw_contracts is None:
3039 return {}
3040 if not isinstance(raw_contracts, list):
3041 raise ValueError(
3042 f"Pipeline spec.data_contracts must be a list; got "
3043 f"{type(raw_contracts).__name__}"
3044 )
3045 contracts: dict[str, dict] = {}
3046 for index, contract in enumerate(raw_contracts):
3047 if not isinstance(contract, dict):
3048 raise ValueError(
3049 f"Pipeline spec.data_contracts[{index}] must be a mapping; got "
3050 f"{type(contract).__name__}"
3051 )
3052 name = contract.get("name")
3053 if not isinstance(name, str) or not name:
3054 raise ValueError(
3055 f"Pipeline spec.data_contracts[{index}] is missing a valid "
3056 "'name'; each data contract requires a non-empty string 'name'"
3057 )
3058 contracts[name] = contract
3059 return contracts
3062def _validate_channel_bindings(
3063 contracts: dict[str, dict], channel_names: frozenset[str]
3064) -> None:
3065 """Reject malformed or undeclared channel bindings at build time.
3067 Every ``output.channels`` map inside ``spec.data_contracts`` must be a
3068 mapping of non-empty output-field names to declared channel names. This
3069 runs unconditionally, so a binding to a channel absent from
3070 ``spec.channels`` fails the build even when no channels are declared.
3071 """
3072 for contract_name, contract in contracts.items():
3073 output = contract.get("output")
3074 if not isinstance(output, dict):
3075 continue
3076 bindings = output.get("channels")
3077 if bindings is None:
3078 continue
3079 if not isinstance(bindings, dict):
3080 raise RunHardFailure(
3081 "CHANNEL_BINDING_ERROR",
3082 f"Data contract '{contract_name}' output.channels must be a "
3083 "mapping of output field name to channel name; got "
3084 f"{type(bindings).__name__}",
3085 )
3086 for field_name, channel_name in bindings.items():
3087 if not isinstance(field_name, str) or not field_name:
3088 raise RunHardFailure(
3089 "CHANNEL_BINDING_ERROR",
3090 f"Data contract '{contract_name}' output.channels has an "
3091 f"invalid output field key '{field_name}'; field names must "
3092 "be non-empty strings",
3093 )
3094 if not isinstance(channel_name, str) or not channel_name:
3095 raise RunHardFailure(
3096 "CHANNEL_BINDING_ERROR",
3097 f"Data contract '{contract_name}' routes output field "
3098 f"'{field_name}' to an invalid channel name "
3099 f"'{channel_name}'; channel names must be non-empty strings",
3100 )
3101 if channel_name not in channel_names:
3102 raise RunHardFailure(
3103 "CHANNEL_BINDING_ERROR",
3104 f"Data contract '{contract_name}' routes output field "
3105 f"'{field_name}' to undeclared channel '{channel_name}'; "
3106 "the channel must be declared in spec.channels",
3107 )
3110def _validate_channel_read_bindings(
3111 contracts: dict[str, dict], channel_names: frozenset[str]
3112) -> None:
3113 """Reject malformed or undeclared read bindings at build time.
3115 Every ``input.channels`` list inside ``spec.data_contracts`` must be an
3116 ordered list of non-empty channel names declared in ``spec.channels``.
3117 This runs unconditionally, so a declared read of a channel absent from
3118 ``spec.channels`` fails the build even when no channels are declared.
3119 """
3120 for contract_name, contract in contracts.items():
3121 inputs = contract.get("input")
3122 if not isinstance(inputs, dict):
3123 continue
3124 reads = inputs.get("channels")
3125 if reads is None:
3126 continue
3127 if not isinstance(reads, list):
3128 raise RunHardFailure(
3129 "CHANNEL_BINDING_ERROR",
3130 f"Data contract '{contract_name}' input.channels must be a "
3131 f"list of channel names; got {type(reads).__name__}",
3132 )
3133 for channel_name in reads:
3134 if not isinstance(channel_name, str) or not channel_name:
3135 raise RunHardFailure(
3136 "CHANNEL_BINDING_ERROR",
3137 f"Data contract '{contract_name}' input.channels has an "
3138 f"invalid channel name '{channel_name}'; channel names must "
3139 "be non-empty strings",
3140 )
3141 if channel_name not in channel_names:
3142 raise RunHardFailure(
3143 "CHANNEL_BINDING_ERROR",
3144 f"Data contract '{contract_name}' reads undeclared channel "
3145 f"'{channel_name}'; the channel must be declared in "
3146 "spec.channels",
3147 )
3150def _step_agent_refs(step: dict) -> list[str]:
3151 """Agent references a step resolves: its agent, plus a spawn's child role."""
3152 refs: list[str] = []
3153 agent_name = step.get("agent")
3154 if agent_name:
3155 refs.append(agent_name)
3156 if step.get("type") == "spawn":
3157 child_name = step.get("child")
3158 if child_name:
3159 refs.append(child_name)
3160 return refs
3163def _resolve_agent_or_none(resolver: Any, agent_name: str) -> dict | None:
3164 """Resolve an agent reference, or None when the name is unknown.
3166 Unknown agents remain a runtime concern: this build-time validator inspects
3167 only the declared data_contract of agents it can resolve, so an unknown
3168 agent name never becomes a new build error.
3169 """
3170 try:
3171 return resolver.resolve(agent_name)
3172 except KeyError:
3173 return None
3176def _validate_step_contract_bindings(
3177 steps: list[dict], contracts: dict[str, dict], resolver: Any
3178) -> None:
3179 """Reject step agents whose declared data_contract is not in the spec.
3181 Every agent-backed step — including nested loop bodies and both roles of a
3182 spawn — must reference a contract declared in ``spec.data_contracts``.
3183 """
3184 for step in _iter_steps(steps):
3185 for agent_name in _step_agent_refs(step):
3186 agent_spec = _resolve_agent_or_none(resolver, agent_name)
3187 if agent_spec is None:
3188 continue
3189 contract_name = (agent_spec.get("spec") or {}).get("data_contract")
3190 if contract_name and contract_name not in contracts:
3191 raise RunHardFailure(
3192 "CHANNEL_BINDING_ERROR",
3193 f"Step '{step.get('id', '?')}' agent declares data_contract "
3194 f"'{contract_name}' which is not declared in the pipeline "
3195 "spec.data_contracts",
3196 )
3199def _validate_channel_names(channels: list[dict], steps: list[dict]) -> None:
3200 """Reject duplicate and engine/step-reserved channel names."""
3201 forbidden = (
3202 {_CORE}
3203 | set(_RESERVED_NAMESPACES)
3204 | _CORE_STATE_KEYS
3205 | _collect_step_ids(steps)
3206 )
3207 seen: set[str] = set()
3208 for channel in channels:
3209 name = channel.get("name")
3210 if name in seen:
3211 raise ValueError(
3212 f"Pipeline declares channel '{name}' more than once; channel "
3213 "names must be unique"
3214 )
3215 seen.add(name)
3216 if name in forbidden:
3217 raise ValueError(
3218 f"Pipeline channel '{name}' collides with an engine-reserved "
3219 "name (a step id, engine state key, reserved namespace, or the "
3220 "core state channel); channel names must be unique and "
3221 "non-reserved"
3222 )
3225def _input_path_root(value: str) -> str | None:
3226 """Leading state key of a bare dot-path input string, or None.
3228 Template strings, bracket-prefixed verbatim literals, and blank strings
3229 are not state reads and yield None.
3230 """
3231 if "{{" in value:
3232 return None
3233 stripped = value.lstrip()
3234 if not stripped or stripped.startswith("["):
3235 return None
3236 return re.split(r"[.\[]", stripped, maxsplit=1)[0]
3239def _iter_input_roots(input_spec: Any):
3240 """Yield every bare dot-path root in an input spec, recursing into containers."""
3241 if isinstance(input_spec, str):
3242 root = _input_path_root(input_spec)
3243 if root:
3244 yield root
3245 elif isinstance(input_spec, dict):
3246 for value in input_spec.values():
3247 yield from _iter_input_roots(value)
3248 elif isinstance(input_spec, list):
3249 for value in input_spec:
3250 yield from _iter_input_roots(value)
3253def _validate_channel_inputs(channels: list[dict], steps: list[dict]) -> None:
3254 """Reject channel-pipeline input roots that name nothing the state exposes.
3256 The channel state is the declared channels plus the engine state, so a bare
3257 dot-path input root must be a declared channel, a known step id, a reserved
3258 namespace, or an engine state key. Anything else is a hard build error —
3259 there is no silent fallback to a missing key.
3260 """
3261 allowed = (
3262 {ch["name"] for ch in channels}
3263 | _collect_step_ids(steps)
3264 | set(_RESERVED_NAMESPACES)
3265 | _CORE_STATE_KEYS
3266 )
3267 for step in _iter_steps(steps):
3268 for root in _iter_input_roots(step.get("input")):
3269 if root not in allowed:
3270 raise ValueError(
3271 f"Pipeline channel step '{step.get('id', '?')}' input "
3272 f"references unknown root '{root}'; in a channel pipeline "
3273 "an input root must be a declared channel, a step id, a "
3274 "reserved namespace, or an engine state key"
3275 )
3278def _resolve_allow_instruction(
3279 entry: str, builtins: BuiltinExecutor, step_functions: StepFunctionExecutor | None
3280) -> str | None:
3281 """Resolve one whitelist entry's usage instruction, builtin-first.
3283 Mirrors the runtime dispatch order (ADR-0037 D2): a name in the builtin
3284 register resolves to the Core-owned registry text, otherwise to the
3285 StepFunction's ``agent_instructions``.
3286 """
3287 return instruction_text(entry, builtins, step_functions)
3290_AGENT_INVOKING_STEP_TYPES = frozenset({"sequential", "parallel", "spawn", "foreach"})
3293def _validate_allow_step_types(steps: list[dict]) -> None:
3294 """Reject ``allow`` on a step that does not actually invoke an agent.
3296 Only an agent-bearing step consumes its whitelist; on any other step the
3297 declaration would be silently dropped — notably a ``call`` step, which
3298 build-time expansion removes before the whitelist validator runs, and an
3299 agent-less ``sequential``/``spawn``/``foreach``. A container step that
3300 declares a ``body`` runs its body path, so a ``body`` wins over a declared
3301 ``agent`` and the whitelist would be inert. Runs on the raw step list,
3302 before call expansion.
3303 """
3304 for step in _iter_steps(steps):
3305 if not step.get("allow"):
3306 continue
3307 step_type = step.get("type", "sequential")
3308 agent = step.get("agent")
3309 invokes_agent = (
3310 step_type in _AGENT_INVOKING_STEP_TYPES
3311 and agent
3312 and not step.get("body")
3313 )
3314 if invokes_agent:
3315 continue
3316 step_id = step.get("id", "?")
3317 raise RunHardFailure(
3318 "ALLOW_INVALID_STEP_TYPE",
3319 f"Step '{step_id}' declares 'allow' but is not an agent-bearing "
3320 f"step that invokes an agent: type '{step_type}' with agent "
3321 f"'{agent}' does not actually invoke an agent",
3322 {"step_id": step_id, "type": step_type, "agent": agent},
3323 )
3326def _validate_allow_instructions(
3327 steps: list[dict],
3328 builtins: BuiltinExecutor,
3329 step_functions: StepFunctionExecutor | None,
3330) -> None:
3331 """Require a non-empty usage instruction for every ``allow`` entry.
3333 Walks all steps — including nested loop/parallel bodies — and fails fast
3334 when a whitelisted name resolves to no instruction: neither a whitelistable
3335 builtin registry text nor a StepFunction ``agent_instructions``.
3336 """
3337 for step in _iter_steps(steps):
3338 for entry in step.get("allow") or []:
3339 if _resolve_allow_instruction(entry, builtins, step_functions):
3340 continue
3341 step_id = step.get("id", "?")
3342 raise RunHardFailure(
3343 "AGENT_INSTRUCTION_MISSING",
3344 f"Step '{step_id}' allow entry '{entry}' has no usage "
3345 "instruction: it is neither a whitelistable builtin nor a "
3346 "StepFunction with agent_instructions",
3347 {"step_id": step_id, "allow": entry},
3348 )
3351def _flatten_channel_state(state: dict) -> dict:
3352 """Read the flat committed snapshot nested under the Core namespace."""
3353 return dict(state.get(_CORE) or {})
3356def _strip_channel_keys(flat: dict, channel_names: frozenset[str]) -> dict:
3357 """Drop channel-named keys so a node emits only the core state."""
3358 return {k: v for k, v in flat.items() if k not in channel_names}
3361_FAILURE_ENVELOPE_KEYS = frozenset({"error", "parse_error"})
3363_NON_ERROR_ENVELOPE_STATUSES: frozenset[str] = ENVELOPE_STATUSES - {"error"}
3366def _is_failure_envelope(value: Any) -> bool:
3367 """True when a step output is not a successfully parsed dict.
3369 Missing output, non-dict values, and the ``error``/``parse_error``
3370 envelopes written by the LLM failure paths are not routed. A valid
3371 non-error envelope (``ok``/``partial``/``no_success``) is routed even when
3372 it carries an explicit ``"error": null``.
3373 """
3374 if not isinstance(value, dict):
3375 return True
3376 if value.get("status") in _NON_ERROR_ENVELOPE_STATUSES:
3377 return False
3378 return bool(_FAILURE_ENVELOPE_KEYS & value.keys())
3381def _declared_channel_bindings(
3382 agent_spec: dict,
3383 contracts: dict[str, dict],
3384 channel_names: frozenset[str],
3385 step_id: str,
3386) -> dict[str, str]:
3387 """Return the agent contract's {output_field: channel_name} bindings.
3389 The mapping is the reserved top-level ``channels`` key inside the
3390 contract's free-form ``output``. A reference to an undeclared contract or
3391 an undeclared target channel is a hard error.
3392 """
3393 contract_name = (agent_spec.get("spec") or {}).get("data_contract")
3394 if not contract_name:
3395 return {}
3396 contract = contracts.get(contract_name)
3397 if contract is None:
3398 raise RunHardFailure(
3399 "CHANNEL_BINDING_ERROR",
3400 f"Step '{step_id}' agent declares data_contract '{contract_name}' "
3401 "which is not declared in the pipeline spec.data_contracts",
3402 )
3403 output = contract.get("output")
3404 if not isinstance(output, dict):
3405 return {}
3406 channels = output.get("channels")
3407 if not isinstance(channels, dict):
3408 return {}
3409 bindings: dict[str, str] = {}
3410 for field_name, channel_name in channels.items():
3411 if channel_name not in channel_names:
3412 raise RunHardFailure(
3413 "CHANNEL_BINDING_ERROR",
3414 f"Step '{step_id}' data_contract '{contract_name}' routes output "
3415 f"field '{field_name}' to undeclared channel '{channel_name}'; "
3416 "the channel must be declared in spec.channels",
3417 )
3418 bindings[field_name] = channel_name
3419 return bindings
3422def _declared_output_channel_writes(
3423 agent_spec: dict,
3424 output: Any,
3425 contracts: dict[str, dict],
3426 channel_names: frozenset[str],
3427 step_id: str,
3428) -> dict[str, Any]:
3429 """Route one agent output's declared channel-bound fields to their channels.
3431 Returns ``{channel_name: value}`` for the caller to stage into the working
3432 set. Failure envelopes are not routed; a successful dict missing a
3433 declared field is a hard error.
3434 """
3435 bindings = _declared_channel_bindings(agent_spec, contracts, channel_names, step_id)
3436 if not bindings:
3437 return {}
3438 if _is_failure_envelope(output):
3439 return {}
3440 writes: dict[str, Any] = {}
3441 for field_name, channel_name in bindings.items():
3442 if field_name not in output:
3443 raise RunHardFailure(
3444 "CHANNEL_BINDING_ERROR",
3445 f"Step '{step_id}' output is missing declared channel-bound "
3446 f"field '{field_name}' (data_contract routes it to channel "
3447 f"'{channel_name}')",
3448 )
3449 writes[channel_name] = output[field_name]
3450 return writes
3453def _route_declared_channels(
3454 step: dict,
3455 pre_state: dict,
3456 core: dict,
3457 step_runner: Any,
3458 contracts: dict[str, dict],
3459 channel_names: frozenset[str],
3460) -> dict[str, Any]:
3461 """Route a step's declared channel-bound output fields to their channels.
3463 Returns ``{channel_name: value}`` for LangGraph to merge through each
3464 channel's reducer. Only an actual agent invocation is routed: a
3465 ``skip_llm_if`` passthrough and a parallel fan-out envelope are excluded.
3466 Non-declared output fields are left untouched in ``core``
3467 (non-destructive routing); failure envelopes are not routed. A spawn step
3468 is routed at the spawn seam from its node result, never here from its
3469 envelope.
3470 """
3471 step_id = step.get("id", "?")
3472 if step.get("type") == "spawn":
3473 return {}
3474 if step_id not in core:
3475 return {}
3476 agent_spec = step_runner.resolve_agent(step)
3477 if agent_spec is None:
3478 return {}
3479 if not step_runner.ran_agent(step, pre_state):
3480 return {}
3481 return _declared_output_channel_writes(
3482 agent_spec, core[step_id], contracts, channel_names, step_id
3483 )
3486def _make_step_node(
3487 step: dict,
3488 step_runner: Any,
3489 channel_names: frozenset[str],
3490 contracts: dict[str, dict],
3491 channel_types: dict[str, str],
3492) -> Any:
3493 if not channel_names:
3494 def node_fn(state: PipelineState) -> PipelineState:
3495 return step_runner.run_step(step, state)
3496 return node_fn
3498 def node_fn(state: PipelineState) -> PipelineState:
3499 committed = _flatten_channel_state(state)
3500 working_set = ChannelWorkingSet(committed, channel_types)
3501 core = _strip_channel_keys(
3502 step_runner.run_step(step, committed, working_set), channel_names
3503 )
3504 writes = _route_declared_channels(
3505 step, committed, core, step_runner, contracts, channel_names
3506 )
3507 for name, value in writes.items():
3508 working_set.stage(name, value)
3509 channels = working_set.commit()
3510 return {_CORE: {**core, **channels}}
3511 return node_fn
3514def _collect_parallel_channel_writes(
3515 steps: list[dict],
3516 pre_state: dict,
3517 core: dict,
3518 step_runner: Any,
3519 contracts: dict[str, dict],
3520 channel_names: frozenset[str],
3521 branch_working_sets: dict[str, ChannelWorkingSet] | None = None,
3522) -> list[tuple[str, Any]]:
3523 """Collect one parallel node's declared writes in branch order.
3525 Every branch write is emitted, so fan-in feeds the channel-type reducer
3526 every write in deterministic declaration/branch order: Topic appends each
3527 write, LastValue keeps the last, and BinaryOperatorAggregate combines them
3528 per its accumulated shape (list concat, collection-map union, dict merge,
3529 scalar last). A branch's isolated body writes are emitted before its own
3530 declared write, mirroring the top-level node's route-after-run order.
3531 """
3532 writes: list[tuple[str, Any]] = []
3533 for step in steps:
3534 if branch_working_sets is not None:
3535 branch_working_set = branch_working_sets.get(step["id"])
3536 if branch_working_set is not None:
3537 writes.extend(branch_working_set.pending_writes())
3538 branch = _route_declared_channels(
3539 step, pre_state, core, step_runner, contracts, channel_names
3540 )
3541 for channel_name, value in branch.items():
3542 writes.append((channel_name, value))
3543 return writes
3546def _make_parallel_node(
3547 group_steps: list[dict],
3548 step_runner: Any,
3549 channel_names: frozenset[str],
3550 contracts: dict[str, dict],
3551 channel_types: dict[str, str],
3552) -> Any:
3553 def node_fn(state: PipelineState) -> PipelineState:
3554 base = (
3555 _flatten_channel_state(state)
3556 if channel_names else state
3557 )
3558 working_set = (
3559 ChannelWorkingSet(base, channel_types) if channel_names else None
3560 )
3561 # Each branch writes into its own fork; the shared set is advanced only
3562 # at fan-in, so branch-internal body writes never race.
3563 branch_working_sets = (
3564 {s["id"]: working_set.fork() for s in group_steps}
3565 if working_set is not None else None
3566 )
3568 def submit_branch(pool: Any, step: dict) -> Any:
3569 if branch_working_sets is None:
3570 return pool.submit(step_runner.run_step, step, base)
3571 return pool.submit(
3572 step_runner.run_step, step, base, branch_working_sets[step["id"]]
3573 )
3575 results: dict[str, Any] = {}
3576 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
3577 futures = {submit_branch(pool, s): s["id"] for s in group_steps}
3578 for future in as_completed(futures):
3579 sid = futures[future]
3580 try:
3581 out = future.result()
3582 results[sid] = out.get(sid)
3583 except SkillLoadError:
3584 raise
3585 except RunHardFailure:
3586 pool.shutdown(wait=False, cancel_futures=True)
3587 raise
3588 except Exception as exc: # noqa: BLE001
3589 log.error("Step '%s' failed: %s", sid, exc)
3590 results[sid] = {"error": str(exc)}
3591 if channel_names:
3592 merged = _strip_channel_keys({**base, **results}, channel_names)
3593 writes = _collect_parallel_channel_writes(
3594 group_steps, base, merged, step_runner, contracts, channel_names,
3595 branch_working_sets,
3596 )
3597 for name, value in writes:
3598 working_set.stage(name, value)
3599 channels = working_set.commit()
3600 return {_CORE: {**merged, **channels}}
3601 return {**state, **results}
3602 return node_fn
3605class _ChannelGraphShim:
3606 """Flat-dict boundary over a Core-owned channel-schema compiled graph.
3608 Preserves the invoke contract used by main.py and tests: a flat dict in,
3609 a flat dict out. The flat input is seeded through ``seed_committed`` and
3610 nested under ``__core__`` for the compiled graph; the committed snapshot is
3611 unwrapped back out. Streaming and state surfaces are not exposed because
3612 they would leak or accept the internal ``__core__`` shape; read-only
3613 introspection delegates to the compiled graph.
3614 """
3616 def __init__(self, compiled: Any, specs: list[ChannelSpec]) -> None:
3617 self._compiled = compiled
3618 self._specs = specs
3620 def invoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
3621 seeded = seed_committed(self._specs, flat_state)
3622 result = self._compiled.invoke(wrap(seeded), *args, **kwargs)
3623 return unwrap(result)
3625 async def ainvoke(self, flat_state: dict, *args: Any, **kwargs: Any) -> dict:
3626 seeded = seed_committed(self._specs, flat_state)
3627 result = await self._compiled.ainvoke(wrap(seeded), *args, **kwargs)
3628 return unwrap(result)
3630 def stream(self, *args: Any, **kwargs: Any) -> Any:
3631 self._reject_internal_shape("stream")
3633 def astream(self, *args: Any, **kwargs: Any) -> Any:
3634 self._reject_internal_shape("astream")
3636 def update_state(self, *args: Any, **kwargs: Any) -> Any:
3637 self._reject_internal_shape("update_state")
3639 def aupdate_state(self, *args: Any, **kwargs: Any) -> Any:
3640 self._reject_internal_shape("aupdate_state")
3642 def get_state(self, *args: Any, **kwargs: Any) -> Any:
3643 self._reject_internal_shape("get_state")
3645 def aget_state(self, *args: Any, **kwargs: Any) -> Any:
3646 self._reject_internal_shape("aget_state")
3648 def _reject_internal_shape(self, surface: str) -> None:
3649 raise NotImplementedError(
3650 f"_ChannelGraphShim.{surface}() is not supported: only the flat "
3651 ".invoke()/.ainvoke() contract is exposed through the channel "
3652 "boundary shim"
3653 )
3655 def __getattr__(self, name: str) -> Any:
3656 return getattr(self._compiled, name)
3659class GraphBuilder:
3660 """
3661 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
3663 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
3664 käytetään vain {{ }} -templatejen evaluointiin.
3665 """
3667 def __init__(
3668 self,
3669 registry: ManifestRegistry,
3670 builtins: BuiltinExecutor | None = None,
3671 invoke_agent_fn: Any | None = None,
3672 run_context: dict | None = None,
3673 tracer: Any | None = None,
3674 ) -> None:
3675 self._registry = registry
3676 self._resolver = AgentResolver(registry)
3677 self._invoke_agent_fn = invoke_agent_fn
3678 self._builtins = builtins or _DEFAULT_BUILTINS
3679 if invoke_agent_fn is not None:
3680 self._builtins.invoke_agent_fn = invoke_agent_fn
3681 self._step_functions = StepFunctionExecutor(registry)
3682 self._step_runner = StepRunner(
3683 self._resolver, self._builtins, invoke_agent_fn, run_context,
3684 step_functions=self._step_functions,
3685 tracer=tracer,
3686 )
3688 def build(self, pipeline: dict): # -> CompiledGraph
3689 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
3690 try:
3691 from langgraph.graph import StateGraph, END
3692 except ImportError as exc:
3693 raise ImportError(
3694 "langgraph is required to build a workflow. "
3695 "Install it with: pip install langgraph"
3696 ) from exc
3698 steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
3699 # Validate `allow` step types on the raw graph: call-expansion removes
3700 # call steps (and their dropped `allow`) before the whitelist validator.
3701 _validate_allow_step_types(steps)
3702 # Build-time call-step expansion (ADR-0025 D4): the runtime executes only
3703 # the flat expanded graph. LangGraph never sees a call step.
3704 steps = expand_calls(steps, self._registry)
3705 ordered = _topological_order(steps)
3707 channels: list[dict] = _validated_channels(
3708 (pipeline.get("spec") or {}).get("channels")
3709 )
3710 if channels:
3711 _validate_channel_names(channels, steps)
3712 _validate_channel_inputs(channels, steps)
3713 specs = parse_channel_specs(
3714 [{"name": ch["name"], "type": ch.get("type", LAST_VALUE)} for ch in channels]
3715 )
3716 channel_types = {spec.name: spec.type for spec in specs}
3717 channel_names = frozenset(channel_types)
3718 contracts = _index_data_contracts(
3719 (pipeline.get("spec") or {}).get("data_contracts")
3720 )
3721 _validate_channel_bindings(contracts, channel_names)
3722 _validate_channel_read_bindings(contracts, channel_names)
3723 _validate_step_contract_bindings(steps, contracts, self._resolver)
3724 _validate_allow_instructions(steps, self._builtins, self._step_functions)
3725 if self._invoke_agent_fn is not None and hasattr(
3726 self._invoke_agent_fn, "set_data_contracts"
3727 ):
3728 self._invoke_agent_fn.set_data_contracts(contracts)
3730 # Validate the pipeline's env map (spec.envs must be a map — ADR-0012).
3731 envs = (pipeline.get("spec") or {}).get("envs")
3732 if envs is not None and not isinstance(envs, dict):
3733 raise ValueError(
3734 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
3735 "spec.envs must be a map of env name → image ref (ADR-0012)"
3736 )
3738 # Group steps into LangGraph nodes:
3739 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
3740 nodes: list[list[dict]] = []
3741 for step in ordered:
3742 dep_sig = frozenset(step.get("depends_on") or [])
3743 if nodes:
3744 last_group_sig = frozenset(
3745 dep
3746 for s in nodes[-1]
3747 for dep in (s.get("depends_on") or [])
3748 )
3749 last_types = {s.get("type") for s in nodes[-1]}
3750 if (
3751 dep_sig == last_group_sig
3752 and step.get("type") == "parallel"
3753 and last_types == {"parallel"}
3754 ):
3755 nodes[-1].append(step)
3756 continue
3757 nodes.append([step])
3759 state_schema = _build_state_schema() if channels else PipelineState
3760 graph: StateGraph = StateGraph(state_schema)
3761 step_runner = self._step_runner
3762 step_runner.set_channel_routing(contracts, channel_names)
3764 prev_node_id: str | None = None
3765 for group in nodes:
3766 if len(group) == 1:
3767 node_id = group[0]["id"]
3768 else:
3769 node_id = "parallel_" + "_".join(s["id"] for s in group)
3771 captured_group = list(group)
3773 if len(captured_group) == 1:
3774 fn = _make_step_node(
3775 captured_group[0], step_runner, channel_names, contracts,
3776 channel_types,
3777 )
3778 else:
3779 fn = _make_parallel_node(
3780 captured_group,
3781 step_runner,
3782 channel_names,
3783 contracts,
3784 channel_types,
3785 )
3787 graph.add_node(node_id, fn)
3789 if prev_node_id is None:
3790 graph.set_entry_point(node_id)
3791 else:
3792 graph.add_edge(prev_node_id, node_id)
3794 prev_node_id = node_id
3796 if prev_node_id:
3797 graph.add_edge(prev_node_id, END)
3798 else:
3799 graph.add_node("noop", lambda s: s)
3800 graph.set_entry_point("noop")
3801 graph.add_edge("noop", END)
3803 compiled = graph.compile()
3804 if channel_names:
3805 return _ChannelGraphShim(compiled, specs)
3806 return compiled
3809# ---------------------------------------------------------------------------
3810# Julkinen API
3811# ---------------------------------------------------------------------------
3813_registry_cache: ManifestRegistry | None = None
3816def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
3817 global _registry_cache
3818 if _registry_cache is None or dirs is not None:
3819 loader = ManifestLoader(dirs)
3820 manifests = loader.load_all()
3821 _registry_cache = ManifestRegistry(manifests)
3822 return _registry_cache
3825def build_workflow(
3826 pipeline_name: str,
3827 *,
3828 dirs: list[Path] | None = None,
3829 builtins: BuiltinExecutor | None = None,
3830 invoke_agent_fn: Any | None = None,
3831 run_context: dict | None = None,
3832 tracer: Any | None = None,
3833):
3834 """
3835 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
3836 LangGraph-workflown.
3838 Args:
3839 pipeline_name: Pipeline metadata.name -arvo.
3840 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
3841 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
3842 invoke_agent_fn: fn(resolved_spec, step, state) → state.
3843 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
3845 Returns:
3846 Compilattu LangGraph CompiledGraph.
3847 """
3848 registry = _get_registry(dirs)
3849 pipeline = registry.get("Pipeline", pipeline_name)
3850 if pipeline is None:
3851 available = [
3852 (m.get("metadata") or {}).get("name")
3853 for m in registry.all_of_kind("Pipeline")
3854 ]
3855 raise KeyError(
3856 f"Pipeline '{pipeline_name}' not found. Available: {available}"
3857 )
3858 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
3859 return builder.build(pipeline)