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