Coverage for core/langgraph/runner.py: 90%
1304 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 02:20 +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 logging
20import os
21import re
22import threading
23from concurrent.futures import ThreadPoolExecutor, as_completed
24from pathlib import Path
25from typing import Any
27import yaml
29# SkillLoadError-import (ei sirkulaari-riskiä — _llm importtaa vain stdlibin):
30# testit ja discover-wrapperit lataavat _llm:n tasaisena moduulina (sys.path
31# insert core/langgraph/), main.py qualified (langgraph._llm, PYTHONPATH=/app).
32# Kummatkin pinnat käännetään samaan _llm-tiedostoon; try/except varmistaa,
33# että luokka on sama objekti kuin nostavalla puolella molemmissa konteksteissa.
34try:
35 from _llm import SkillLoadError # noqa: F401
36except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
37 from langgraph._llm import SkillLoadError # noqa: F401
39# Same dual-import pattern as _llm (flat sys.path and qualified PYTHONPATH=/app)
40try:
41 from exec_runner import CommandExecutor # noqa: F401
42except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
43 from langgraph.exec_runner import CommandExecutor # noqa: F401
45log = logging.getLogger(__name__)
48class RunHardFailure(Exception):
49 """Run-tason kova virhe: ajo pysähtyy heti (phase failed, exit 1).
51 Nostetaan kun step-tulos kantaa kovan virheen: TRUNCATION_ERROR-koodi
52 (kuratoidun kontekstin katkaisu) tai
53 kontekstiylityksen signatuurin kantava VALIDATION_ERROR
54 (assembled_tokens + context_budget, _llm._context_overflow_result).
55 Kaikki muut virheluokat (budget:*, parse_error, timeout, LLM-virheet)
56 säilyvät nykyisessä partial-semantiikassa. Stepin details kulkeutuu
57 attribuuttina loppu-PATCHin error.details-kenttään.
58 """
60 HARD_ERROR_CODES: frozenset[str] = frozenset({"TRUNCATION_ERROR"})
62 def __init__(self, code: str, message: str, details: dict | None = None) -> None:
63 super().__init__(message)
64 self.code = code
65 self.message = message
66 self.details = details
69class SpawnBudget:
70 """Per-tree agent-call budget shared across all spawn nodes.
72 The whole spawn tree draws from a single counter: every agent invocation
73 (the director included) consumes one unit, so the total number of
74 agent invocations the tree may make is bounded by `max_agent_calls`. This is a
75 per-agent-run budget, NOT an LLM-roundtrip budget: a single agent run may
76 make several LLM calls (e.g. an agent-side tool loop via file_access);
77 those LLM roundtrips are bounded by the run's `max_llm_calls`, not by
78 `max_agent_calls`. There is no engine-level agent retry loop. Thread-safe so
79 a concurrent caller cannot over-commit the shared counter.
80 """
82 def __init__(self, max_agent_calls: int | None) -> None:
83 self._remaining: int | None = (
84 None if max_agent_calls is None else max(0, int(max_agent_calls))
85 )
86 self._lock = threading.Lock()
88 def consume(self) -> bool:
89 """Reserve one agent call. Returns False when the budget is exhausted."""
90 with self._lock:
91 if self._remaining is None:
92 return True
93 if self._remaining <= 0:
94 return False
95 self._remaining -= 1
96 return True
99_SPAWN_SOFT_FAIL_KEYS: frozenset[str] = frozenset({"error", "parse_error", "escalate"})
101_SPAWN_DEFAULT_MAX_AGENT_CALLS: int = 64
102_SPAWN_DEFAULT_MAX_DEPTH: int = 1
103_SPAWN_DEFAULT_MAX_SPAWN_PER_NODE: int = 16
105# ---------------------------------------------------------------------------
106# Polut
107# ---------------------------------------------------------------------------
109PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", "/project"))
110MANIFEST_DIRS = [
111 PROJECT_ROOT / ".agent-platform" / "workflows",
112 PROJECT_ROOT / ".agent-platform" / "agents",
113]
114SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "manifest.schema.json"
117def resolve_manifest_dirs(clone_root: Path) -> list[Path]:
118 """Palauttaa olemassa olevat manifestihakemistot kloonatusta manifest-reposta.
120 Skannaa ainoastaan `.agent-platform/workflows` ja `.agent-platform/agents` —
121 ei koskaan koko kloonia. [] jos kumpaakaan ei ole (kutsuja → fail-fast).
122 """
123 return [
124 d for d in (
125 clone_root / ".agent-platform" / "workflows",
126 clone_root / ".agent-platform" / "agents",
127 )
128 if d.is_dir()
129 ]
131# ---------------------------------------------------------------------------
132# Template-resoluutio
133# ---------------------------------------------------------------------------
135_TMPL_RE = re.compile(r'\{\{\s*(.*?)\s*\}\}')
137# Sentinel for an unresolvable dot-path segment (distinct from a resolved None).
138_UNRESOLVED = object()
141def _split_key_indices(part: str) -> tuple[str, list[int]]:
142 """Split a dot-path segment into a base key and a list of indices.
144 ``'a[0][1]'`` → ``('a', [0, 1])``; ``'a'`` → ``('a', [])``; ``'[0]'`` →
145 ``('', [0])``. Only non-negative indices are supported (list access).
146 """
147 m = re.fullmatch(r'([^\[]*)((\[\d+\])*)', part)
148 base = m.group(1)
149 indices = [int(x) for x in re.findall(r'\d+', m.group(2))]
150 return base, indices
153def _apply_indices(obj: Any, indices: list[int]) -> Any:
154 """Index a (possibly chained) list/tuple; ``_UNRESOLVED`` on out-of-range."""
155 for idx in indices:
156 if not isinstance(obj, (list, tuple)) or not (0 <= idx < len(obj)):
157 return _UNRESOLVED
158 obj = obj[idx]
159 return obj
162def _dot_get_segment(obj: Any, part: str) -> Any:
163 """Resolve one dot-path segment against a dict or an indexed sequence.
165 A segment may be a bare key (``'a'``), a keyed list-index (``'a[0]'``,
166 ``'a[0][1]'``), or a bare index into a sequence. Returns the resolved value
167 or the ``_UNRESOLVED`` sentinel when the segment cannot be resolved (missing
168 key, non-dict/non-sequence mid-path, or out-of-range index).
169 """
170 base, indices = _split_key_indices(part)
171 if isinstance(obj, dict):
172 if base:
173 if base not in obj:
174 return _UNRESOLVED
175 obj = obj[base]
176 return _apply_indices(obj, indices)
177 if isinstance(obj, (list, tuple)):
178 if indices:
179 return _apply_indices(obj, indices)
180 if base and base.isdigit():
181 return _apply_indices(obj, [int(base)])
182 return _UNRESOLVED
183 return _UNRESOLVED
186def _dot_get(obj: Any, path: str) -> Any:
187 """Navigate nested dicts/lists via dot-path: 'a.b.c' → obj['a']['b']['c'].
189 Supports list-index segments (``results[1].field``) and dot-into-aggregate
190 traversal (missing key / non-dict-non-list mid-path / out-of-range index
191 resolve to None — lenient semantics for non-output refs).
192 """
193 for part in path.split('.'):
194 obj = _dot_get_segment(obj, part)
195 if obj is _UNRESOLVED:
196 return None
197 return obj
200def _dot_get_strict(obj: Any, path: str) -> Any:
201 """Dot-path navigation that raises on an unresolved reference.
203 A reference is unresolved when a key on the path is absent or a segment is
204 not a mapping — used by condition evaluation so an undefined or skipped
205 step's field is never silently evaluated to falsy (ADR-0012 spirit). Also
206 used by ``__call_emit`` to distinguish a resolved output source from an
207 unresolved one (required output → hard error, ADR-0012).
208 """
209 for part in path.split('.'):
210 if obj is None:
211 raise ValueError(
212 f"Unresolved reference '{path}': '{part}' is not a mapping"
213 )
214 obj = _dot_get_segment(obj, part)
215 if obj is _UNRESOLVED:
216 raise ValueError(
217 f"Unresolved reference '{path}': '{part}' is not a mapping"
218 )
219 return obj
222def materialize_pipeline_output(pipeline_spec: dict, final_state: dict) -> Any:
223 """Resolve a Pipeline's declared run-end output.
225 Reads ``spec.output.source`` (a dot-path) out of ``final_state`` via
226 ``_dot_get`` and returns the surfaced value. Returns ``None`` when no
227 ``spec.output`` is declared, when it has no ``source``, or when the source
228 is unresolved — this is a run-end surfacing (distinct from the emit-time
229 hard-error used for function-interface outputs): an optional run output
230 that cannot be resolved surfaces as None, never a run-end failure.
231 """
232 out_spec = (pipeline_spec or {}).get("output")
233 if not out_spec:
234 return None
235 source = out_spec.get("source")
236 if not source:
237 return None
238 return _dot_get(final_state, source)
241def _merge_metrics(base: dict | None, additions: dict) -> dict:
242 """Yhdistä step_context_metrics-envelopet: myöhempi kirjoitus voittaa
243 (sama semantiikka kuin state-merge {**state, ...})."""
244 merged = dict(base or {})
245 merged.update(additions)
246 return merged
249_MULTI_OPS = ('<=', '>=', '==', '!=')
250_SINGLE_OPS = '+-<>'
251_WORD_BREAK = set("()[],+<=>!'-")
252_CMP_OPS = frozenset(('==', '!=', '<=', '>=', '<', '>'))
255def _tokenize_expr(expr: str) -> list[tuple[str, Any]]:
256 """Split an expression into (kind, value) tokens.
258 Kinds: 'op' (+ - <= >= == != < >), 'punct' ( ) [ ] ,, 'num', 'str',
259 'name'. A 'name' is a dot-path reference or a bare word (e.g. an unquoted
260 comparison literal, true/false/None, and/or/not).
261 """
262 tokens: list[tuple[str, Any]] = []
263 i, n = 0, len(expr)
264 while i < n:
265 ch = expr[i]
266 if ch.isspace():
267 i += 1
268 continue
269 if ch in '()[],':
270 tokens.append(('punct', ch))
271 i += 1
272 continue
273 if ch in '+-<=>!':
274 two = expr[i:i + 2]
275 if two in _MULTI_OPS:
276 tokens.append(('op', two))
277 i += 2
278 continue
279 if ch in _SINGLE_OPS:
280 tokens.append(('op', ch))
281 i += 1
282 continue
283 raise ValueError(f"Invalid operator in expression: {two!r}")
284 if ch in ('\'', '"'):
285 j = i + 1
286 buf: list[str] = []
287 while j < n and expr[j] != ch:
288 buf.append(expr[j])
289 j += 1
290 if j >= n:
291 raise ValueError(f"Unterminated string literal in expression: {expr!r}")
292 tokens.append(('str', ''.join(buf)))
293 i = j + 1
294 continue
295 if ch.isdigit() or (ch == '.' and i + 1 < n and expr[i + 1].isdigit()):
296 j = i
297 if ch == '.':
298 j += 1
299 while j < n and expr[j].isdigit():
300 j += 1
301 tokens.append(('num', float(expr[i:j])))
302 else:
303 while j < n and expr[j].isdigit():
304 j += 1
305 if j < n and expr[j] == '.' and j + 1 < n and expr[j + 1].isdigit():
306 j += 1
307 while j < n and expr[j].isdigit():
308 j += 1
309 tokens.append(('num', float(expr[i:j])))
310 else:
311 tokens.append(('num', int(expr[i:j])))
312 i = j
313 continue
314 j = i
315 while j < n and not expr[j].isspace() and expr[j] not in _WORD_BREAK:
316 j += 1
317 if j == i:
318 raise ValueError(f"Unexpected character in expression: {expr[i:]!r}")
319 tokens.append(('name', expr[i:j]))
320 i = j
321 return tokens
324# AST node shapes:
325# ('lit', value) literal (str/num/bool/None)
326# ('ref', name) dot-path reference
327# ('list', [node, ...]) [a, b, c] literal
328# ('neg', node) unary minus
329# ('not', node) not X
330# ('cmp', op, left, right) comparison
331# ('binop', 'and'|'or'|'+', l, r) binary operator
334class _ExprParser:
335 """Recursive-descent parser for template/condition expressions.
337 Precedence (loosest first): or < and < not < comparison < '+' < unary.
338 Grammar:
339 expr := or_expr
340 or_expr := and_expr ('or' and_expr)*
341 and_expr := not_expr ('and' not_expr)*
342 not_expr := 'not' not_expr | cmp_expr
343 cmp_expr := add_expr (cmp_op add_expr)?
344 add_expr := unary_expr ('+' unary_expr)*
345 unary_expr:= '-' unary_expr | primary
346 primary := num | str | true/false/None | list | '(' expr ')' | name
347 `and`/`or` parse into a full AST but EVALUATE with left-to-right
348 short-circuit, so the right operand is only resolved if needed and an
349 unresolved read there never raises.
350 """
352 def __init__(self, tokens: list[tuple[str, Any]]) -> None:
353 self._tokens = tokens
354 self._i = 0
356 def parse(self) -> tuple:
357 node = self._or_expr()
358 if self._i != len(self._tokens):
359 tok = self._tokens[self._i]
360 raise ValueError(f"Unexpected token in expression: {tok[1]!r}")
361 return node
363 def _peek(self) -> tuple[str, Any] | None:
364 return self._tokens[self._i] if self._i < len(self._tokens) else None
366 def _next(self) -> tuple[str, Any]:
367 tok = self._tokens[self._i]
368 self._i += 1
369 return tok
371 def _peek_name(self, kw: str) -> bool:
372 tok = self._peek()
373 return tok is not None and tok[0] == 'name' and tok[1] == kw
375 def _peek_op(self, op: str) -> bool:
376 tok = self._peek()
377 return tok is not None and tok[0] == 'op' and tok[1] == op
379 def _expect_punct(self, value: str) -> None:
380 tok = self._next()
381 if tok != ('punct', value):
382 raise ValueError(f"Expected '{value}' in expression, got {tok[1]!r}")
384 def _or_expr(self) -> tuple:
385 node = self._and_expr()
386 while self._peek_name('or'):
387 self._next()
388 right = self._and_expr()
389 node = ('binop', 'or', node, right)
390 return node
392 def _and_expr(self) -> tuple:
393 node = self._not_expr()
394 while self._peek_name('and'):
395 self._next()
396 right = self._not_expr()
397 node = ('binop', 'and', node, right)
398 return node
400 def _not_expr(self) -> tuple:
401 if self._peek_name('not'):
402 self._next()
403 return ('not', self._not_expr())
404 return self._cmp_expr()
406 def _cmp_expr(self) -> tuple:
407 node = self._add_expr()
408 tok = self._peek()
409 if tok is not None and tok[0] == 'op' and tok[1] in _CMP_OPS:
410 self._next()
411 right = self._add_expr()
412 node = ('cmp', tok[1], node, right)
413 return node
415 def _add_expr(self) -> tuple:
416 node = self._unary_expr()
417 while self._peek_op('+'):
418 self._next()
419 right = self._unary_expr()
420 node = ('binop', '+', node, right)
421 return node
423 def _unary_expr(self) -> tuple:
424 if self._peek_op('-'):
425 self._next()
426 node = self._unary_expr()
427 if node[0] == 'lit' and isinstance(node[1], (int, float)) and not isinstance(node[1], bool):
428 return ('lit', -node[1])
429 return ('neg', node)
430 return self._primary()
432 def _primary(self) -> tuple:
433 tok = self._peek()
434 if tok is None:
435 raise ValueError("Unexpected end of expression")
436 kind, value = tok
437 if kind == 'num':
438 self._next()
439 return ('lit', value)
440 if kind == 'str':
441 self._next()
442 return ('lit', value)
443 if kind == 'punct':
444 if value == '(':
445 self._next()
446 node = self._or_expr()
447 self._expect_punct(')')
448 return node
449 if value == '[':
450 self._next()
451 items: list[tuple] = []
452 if self._peek() != ('punct', ']'):
453 while True:
454 items.append(self._or_expr())
455 if self._peek() == ('punct', ','):
456 self._next()
457 continue
458 break
459 self._expect_punct(']')
460 return ('list', items)
461 raise ValueError(f"Unexpected punctuation in expression: {value!r}")
462 if kind == 'name':
463 self._next()
464 lowered = value.lower()
465 if lowered == 'true':
466 return ('lit', True)
467 if lowered == 'false':
468 return ('lit', False)
469 if lowered in ('none', 'null'):
470 return ('lit', None)
471 if value in ('and', 'or', 'not'):
472 raise ValueError(f"Unexpected keyword in expression: {value!r}")
473 return ('ref', value)
474 raise ValueError(f"Unexpected token in expression: {value!r}")
477def _is_number(value: Any) -> bool:
478 return isinstance(value, (int, float)) and not isinstance(value, bool)
481def _add_values(left: Any, right: Any) -> Any:
482 """Binary `+`. Lists concatenate, numbers add, strings concatenate."""
483 if isinstance(left, list) and isinstance(right, list):
484 return left + right
485 if _is_number(left) and _is_number(right):
486 return left + right
487 if isinstance(left, str) and isinstance(right, str):
488 return left + right
489 raise ValueError(
490 f"Unsupported operands for '+': {type(left).__name__} + {type(right).__name__}"
491 )
494def _eval_comparison_rhs(node: tuple, ctx: dict, strict: bool) -> Any:
495 """Comparison right-hand side. A bare non-dotted reference compares as a
496 string literal (preserving `profile == PR`); everything else evaluates
497 normally (numbers, quoted strings, dotted refs, lists, grouped exprs)."""
498 if node[0] == 'ref' and '.' not in node[1]:
499 return node[1]
500 return _eval_ast(node, ctx, strict)
503def _eval_cmp(op: str, left_node: tuple, right_node: tuple, ctx: dict, strict: bool) -> bool:
504 left = _eval_ast(left_node, ctx, strict)
505 right = _eval_comparison_rhs(right_node, ctx, strict)
506 try:
507 if op == '==':
508 return left == right
509 if op == '!=':
510 return left != right
511 if op == '<=':
512 return left <= right
513 if op == '>=':
514 return left >= right
515 if op == '<':
516 return left < right
517 if op == '>':
518 return left > right
519 except TypeError:
520 return False
521 raise ValueError(f"Unknown comparison operator: {op!r}")
524def _eval_ast(node: tuple, ctx: dict, strict: bool) -> Any:
525 """Evaluate a parsed AST. `strict=True` raises on an unresolved reference
526 read (condition/exit_condition); `and`/`or` short-circuit left-to-right."""
527 kind = node[0]
528 if kind == 'lit':
529 return node[1]
530 if kind == 'ref':
531 return _dot_get_strict(ctx, node[1]) if strict else _dot_get(ctx, node[1])
532 if kind == 'list':
533 return [_eval_ast(item, ctx, strict) for item in node[1]]
534 if kind == 'neg':
535 return -_eval_ast(node[1], ctx, strict)
536 if kind == 'not':
537 return not bool(_eval_ast(node[1], ctx, strict))
538 if kind == 'cmp':
539 return _eval_cmp(node[1], node[2], node[3], ctx, strict)
540 if kind == 'binop':
541 op, left_node, right_node = node[1], node[2], node[3]
542 if op == 'and':
543 if not bool(_eval_ast(left_node, ctx, strict)):
544 return False
545 return bool(_eval_ast(right_node, ctx, strict))
546 if op == 'or':
547 if bool(_eval_ast(left_node, ctx, strict)):
548 return True
549 return bool(_eval_ast(right_node, ctx, strict))
550 if op == '+':
551 left = _eval_ast(left_node, ctx, strict)
552 right = _eval_ast(right_node, ctx, strict)
553 return _add_values(left, right)
554 raise ValueError(f"Unknown binary operator: {op!r}")
555 raise ValueError(f"Unknown expression node: {kind!r}")
558def _eval_expr(expr: str, ctx: dict, strict: bool = False) -> Any:
559 """Evaluate a template expression against context dict.
561 `strict` is used by condition evaluation: dot-path access to an unresolved
562 reference raises (ValueError) instead of silently resolving to falsy, and
563 and/or short-circuit left-to-right so a value is never read when the left
564 side already determines the result.
565 """
566 expr = expr.strip()
567 tokens = _tokenize_expr(expr)
568 if not tokens:
569 return _dot_get_strict(ctx, expr) if strict else _dot_get(ctx, expr)
570 return _eval_ast(_ExprParser(tokens).parse(), ctx, strict)
573def _resolve_template(value: Any, ctx: dict, strict: bool = False) -> Any:
574 """Recursively resolve {{ expr }} templates in strings / dicts / lists."""
575 if isinstance(value, str):
576 stripped = value.strip()
577 # Entire string is one template → replace with actual type (not string)
578 m = _TMPL_RE.fullmatch(stripped)
579 if m:
580 return _eval_expr(m.group(1), ctx, strict)
581 # Inline substitution → always returns string
582 def _replace(m: re.Match) -> str:
583 result = _eval_expr(m.group(1), ctx, strict)
584 return '' if result is None else str(result)
585 return _TMPL_RE.sub(_replace, value)
586 if isinstance(value, dict):
587 return {k: _resolve_template(v, ctx, strict) for k, v in value.items()}
588 if isinstance(value, list):
589 return [_resolve_template(v, ctx, strict) for v in value]
590 return value
593def _resolve_input(input_spec: Any, state: dict, ctx: dict) -> Any:
594 """
595 Resolve step input spec to a concrete value:
596 "{{ item }}" → template
597 "[host-ai-context-map-path]" → literal placeholder (verbatim pass-through)
598 "gather.doc_files" → dot-path into state
599 {"k": "a.b"} → dict with each value resolved
600 None → None
601 """
602 if input_spec is None:
603 return None
604 if isinstance(input_spec, str):
605 if '{{' in input_spec:
606 return _resolve_template(input_spec, ctx)
607 # A bracket-prefixed literal (e.g. a host-ai-context-map-path
608 # placeholder) is NOT a dot-path reference into state — hand it to the
609 # builtin verbatim so it resolves the placeholder itself.
610 if input_spec.lstrip().startswith('['):
611 return input_spec
612 # Dot-path into state (no braces)
613 return _dot_get(state, input_spec)
614 if isinstance(input_spec, dict):
615 return {k: _resolve_input(v, state, ctx) for k, v in input_spec.items()}
616 if isinstance(input_spec, list):
617 return [_resolve_input(v, state, ctx) for v in input_spec]
618 return input_spec
621def _eval_condition(cond: Any, ctx: dict) -> bool:
622 """Evaluate a step condition. None → True (always run).
624 Strict (short-circuit + hard error on unresolved reads) per R1: an
625 undefined or skipped-step field read without short-circuit protection
626 raises, never silently evaluates to falsy.
627 """
628 if cond is None:
629 return True
630 resolved = _resolve_template(cond, ctx, strict=True)
631 if isinstance(resolved, bool):
632 return resolved
633 if isinstance(resolved, str):
634 return resolved.lower() not in ('false', '0', '', 'none', 'null')
635 return bool(resolved)
638def _resolve_int(value: Any, ctx: dict, default: int) -> int:
639 v = _resolve_template(value, ctx) if isinstance(value, str) else value
640 try:
641 return int(v)
642 except (TypeError, ValueError):
643 return default
646# ---------------------------------------------------------------------------
647# ManifestLoader
648# ---------------------------------------------------------------------------
651def _parse_manifest_file(path: Path) -> list[dict]:
652 """Palauttaa kaikki apiVersion: agent-platform/v1 -dokumentit tiedostosta."""
653 try:
654 content = path.read_text(encoding="utf-8")
655 except OSError as exc:
656 log.warning("Cannot read %s: %s", path, exc)
657 return []
659 results: list[dict] = []
660 for raw in content.split("\n---"):
661 raw_clean = "\n".join(
662 line for line in raw.splitlines()
663 if not line.strip().startswith("#")
664 )
665 try:
666 doc = yaml.safe_load(raw_clean)
667 except yaml.YAMLError as exc:
668 log.warning("YAML parse error in %s: %s", path, exc)
669 continue
670 if isinstance(doc, dict) and doc.get("apiVersion") == "agent-platform/v1":
671 results.append(doc)
672 return results
675class ManifestLoader:
676 """Skannaa hakemistot ja lataa kaikki YAML-manifestit."""
678 def __init__(self, dirs: list[Path] | None = None) -> None:
679 self._dirs = dirs or MANIFEST_DIRS
681 def load_all(self) -> list[dict]:
682 manifests: list[dict] = []
683 for d in self._dirs:
684 if not d.exists():
685 log.debug("Manifest dir not found: %s", d)
686 continue
687 for path in sorted(d.glob("**/*.yaml")):
688 docs = _parse_manifest_file(path)
689 manifests.extend(docs)
690 if docs:
691 log.debug("Loaded %d manifest(s) from %s", len(docs), path)
692 return manifests
695# ---------------------------------------------------------------------------
696# ManifestRegistry
697# ---------------------------------------------------------------------------
699ResourceKey = tuple[str, str] # (kind, name)
702class ManifestRegistry:
703 """Indeksoi ladatut manifestit kind+name avaimella."""
705 def __init__(self, manifests: list[dict]) -> None:
706 self._store: dict[ResourceKey, dict] = {}
707 for doc in manifests:
708 kind = doc.get("kind", "")
709 name = (doc.get("metadata") or {}).get("name", "")
710 if kind and name:
711 key: ResourceKey = (kind, name)
712 if key in self._store:
713 log.warning("Duplicate manifest %s/%s — last one wins", kind, name)
714 self._store[key] = doc
716 def get(self, kind: str, name: str) -> dict | None:
717 return self._store.get((kind, name))
719 def all_of_kind(self, kind: str) -> list[dict]:
720 return [v for (k, _), v in self._store.items() if k == kind]
723# ---------------------------------------------------------------------------
724# AgentResolver
725# ---------------------------------------------------------------------------
727def _deep_merge(base: dict, override: dict) -> dict:
728 """Syvä merge: override voittaa konflikteissa."""
729 result = dict(base)
730 for k, v in override.items():
731 if isinstance(v, dict) and isinstance(result.get(k), dict):
732 result[k] = _deep_merge(result[k], v)
733 else:
734 result[k] = v
735 return result
738class AgentResolver:
739 """
740 Palauttaa fully-resolved agent spec:n instanssille.
741 Jos instanssilla on based_on, mergee Agent-base + AgentInstance-overridet.
742 """
744 def __init__(self, registry: ManifestRegistry) -> None:
745 self._registry = registry
746 self._cache: dict[str, dict] = {}
748 def resolve(self, instance_name: str) -> dict:
749 if instance_name in self._cache:
750 return self._cache[instance_name]
752 instance = self._registry.get("AgentInstance", instance_name)
753 if instance is None:
754 agent = self._registry.get("Agent", instance_name)
755 if agent is None:
756 raise KeyError(f"No Agent or AgentInstance named '{instance_name}'")
757 resolved = agent
758 else:
759 based_on = (instance.get("spec") or {}).get("based_on")
760 if based_on:
761 base_agent = self._registry.get("Agent", based_on)
762 if base_agent is None:
763 raise KeyError(
764 f"AgentInstance '{instance_name}' references unknown Agent '{based_on}'"
765 )
766 base_spec = base_agent.get("spec") or {}
767 inst_spec = {k: v for k, v in (instance.get("spec") or {}).items()
768 if k != "based_on"}
769 merged_spec = _deep_merge(base_spec, inst_spec)
770 else:
771 merged_spec = {k: v for k, v in (instance.get("spec") or {}).items()
772 if k != "based_on"}
774 resolved = {
775 "apiVersion": "agent-platform/v1",
776 "kind": "Agent",
777 "metadata": {"name": instance_name},
778 "spec": merged_spec,
779 }
781 self._cache[instance_name] = resolved
782 return resolved
785# ---------------------------------------------------------------------------
786# State type
787# ---------------------------------------------------------------------------
789PipelineState = dict[str, Any]
791# Sentinel for optional kwargs distinguishing "not provided" from None
792_UNSET: Any = object()
794# ---------------------------------------------------------------------------
795# BuiltinExecutor
796# ---------------------------------------------------------------------------
798class BuiltinExecutor:
799 """
800 Ajaa builtin-komponentit.
801 Uusia komponentteja voi rekisteröidä register()-metodilla.
802 Nimet normalisoidaan: viiva → alaviiva (esim. 'path-verifier' == 'path_verifier').
803 """
805 # Lazy modules are imported on first use (they need httpx / heavier deps).
806 # Profile-specific builtins live here: the PR-profile github-pr-comment
807 # and the TICKET-profile ticket-publish (per-run local feature branch; push/PR disabled).
808 _LAZY_MODULES: tuple[str, ...] = ("_builtins_pr", "_builtins_ticket")
810 def __init__(self) -> None:
811 self._handlers: dict[str, Any] = {
812 "passthrough": self._passthrough,
813 # Engine-internal call-expansion scaffolding (ADR-0025 D4): these are
814 # synthesized at build time by expand_calls(), never authored in a
815 # manifest. The steps that generate them are removed from the flat
816 # runtime graph — only their body + these two op steps remain.
817 "__call_bind": self._call_bind,
818 "__call_emit": self._call_emit,
819 }
820 # Pipeline's LLM invoker (spec, step, state) → state, bound by
821 # GraphBuilder. Profile builtins that make LLM decisions
822 # (ticket-dor / ticket-review) read it via their register_all
823 # wrapper — the executor must have it set before the builtin
824 # executes (binding happens at build_workflow time).
825 self.invoke_agent_fn: Any | None = None
827 def register(self, name: str, fn: Any) -> None:
828 self._handlers[name] = fn
829 # Also register with normalised name (hyphen → underscore)
830 normalised = name.replace("-", "_")
831 if normalised != name:
832 self._handlers[normalised] = fn
834 def get(self, name: str) -> Any:
835 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
836 if fn is not None:
837 return fn
838 import importlib
839 for module_name in self._LAZY_MODULES:
840 try:
841 mod = importlib.import_module(f".{module_name}", package="langgraph")
842 except ImportError:
843 continue
844 mod.register_all(self)
845 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
846 if fn is None:
847 raise ValueError(f"Unknown builtin component '{name}'")
848 return fn
850 def execute(self, component: str, step: dict, state: PipelineState) -> PipelineState:
851 fn = self.get(component)
852 return fn(step, state)
854 # --- default builtins ---
856 @staticmethod
857 def _passthrough(step: dict, state: PipelineState) -> PipelineState:
858 return state
860 @staticmethod
861 def _call_bind(step: dict, state: PipelineState) -> PipelineState:
862 """Call-scope entry (synthesized): bind the function call's params.
864 The caller's function params are resolved against the current state and
865 stored at ``state[<call_id>] = {"params": {...}}`` so the (rewritten)
866 function-body templates ``{{ <call_id>.params.<name> }}`` resolve. The
867 call's reserved ``__call_id`` key is stripped from the bound params.
868 """
869 call_id = (step.get("params") or {}).get("__call_id")
870 if not call_id:
871 raise ValueError("__call_bind missing __call_id (not a synthesized step)")
872 bound = {
873 k: v for k, v in (step.get("params") or {}).items() if k != "__call_id"
874 }
875 return {**state, call_id: {"params": bound}}
877 @staticmethod
878 def _call_emit(step: dict, state: PipelineState) -> PipelineState:
879 """Call-scope exit (synthesized): surface the function's outputs.
881 Per the call step-result contract (00-process-contract r.70) the caller
882 sees only ``state[<call_id>] = {outputs: {<name>: <source value>}}`` —
883 the function body step results are invisible to the caller. Each output
884 name resolves to the manifest interface ``outputs[].source`` dot-path.
886 Optionality (ADR-0031): an output flagged ``optional: true`` is skipped
887 when its source-path is unresolved; an output that is NOT optional must
888 resolve or a RunHardFailure is raised (no silent None — ADR-0012).
889 """
890 params = step.get("params") or {}
891 call_id = params.get("__call_id")
892 if not call_id:
893 raise ValueError("__call_emit missing __call_id (not a synthesized step)")
894 outputs_spec = params.get("__outputs") or []
895 outputs: dict[str, Any] = {}
896 for spec in outputs_spec:
897 if not isinstance(spec, dict):
898 continue
899 name = spec.get("name")
900 source = spec.get("source")
901 if name is None:
902 continue
903 optional = bool(spec.get("optional", False))
904 if source is None:
905 # No declared source-path: treat as unresolved against the
906 # optionality gate (ADR-0012 — no silent None for required).
907 if optional:
908 continue
909 raise RunHardFailure(
910 "UNRESOLVED_OUTPUT",
911 f"call '{call_id}' output '{name}' has no source "
912 "(required output — no silent None, ADR-0012)",
913 {"call_id": call_id, "name": name, "source": source},
914 )
915 try:
916 value = _dot_get_strict(state, source)
917 except ValueError:
918 # Unresolved source-path: required → hard error; optional → skip.
919 if optional:
920 continue
921 raise RunHardFailure(
922 "UNRESOLVED_OUTPUT",
923 f"call '{call_id}' output '{name}' unresolved: source "
924 f"'{source}' not found in state (required output — "
925 "no silent None, ADR-0012)",
926 {"call_id": call_id, "name": name, "source": source},
927 ) from None
928 outputs[name] = value
929 return {**state, call_id: {"outputs": outputs}}
932_DEFAULT_BUILTINS = BuiltinExecutor()
935def _register_domain_builtins(executor: BuiltinExecutor) -> None:
936 """Rekisteröi v1-domain-builtinit (path-verifier, ...).
938 Erotettu omaan moduliin (`_builtins.py`) — runner.py pysyy generic-engine,
939 builtinit voivat olla pipeline-spesifisiäkin v1:ssä.
940 """
941 try:
942 from langgraph import _builtins # type: ignore
943 except ImportError:
944 # Kun runner.py ajetaan ilman pakettirakennetta (esim. /app/langgraph)
945 import importlib.util
946 spec = importlib.util.spec_from_file_location(
947 "_builtins", Path(__file__).parent / "_builtins.py"
948 )
949 if spec and spec.loader:
950 _builtins = importlib.util.module_from_spec(spec)
951 spec.loader.exec_module(_builtins) # type: ignore[union-attr]
952 else:
953 log.warning("Cannot load _builtins.py — domain builtins disabled")
954 return
955 _builtins.register_all(executor)
956 _register_function_builtins(executor)
959def _register_function_builtins(executor: BuiltinExecutor) -> None:
960 """Rekisteröi function-engine builtinit (coverage-check, question-block-form).
962 Erotettu omaan moduliin (`_builtins_functions.py`) — runner.py pysyy
963 generic-engine, function-builtinit pysyvät puhtaina ja erillisinä.
964 """
965 try:
966 from langgraph import _builtins_functions # type: ignore
967 except ImportError:
968 import importlib.util
969 spec = importlib.util.spec_from_file_location(
970 "_builtins_functions", Path(__file__).parent / "_builtins_functions.py"
971 )
972 if spec and spec.loader:
973 _builtins_functions = importlib.util.module_from_spec(spec)
974 spec.loader.exec_module(_builtins_functions) # type: ignore[union-attr]
975 else:
976 log.warning("Cannot load _builtins_functions.py — function builtins disabled")
977 return
978 _builtins_functions.register_all(executor)
981_register_domain_builtins(_DEFAULT_BUILTINS)
983# ---------------------------------------------------------------------------
984# StepRunner
985# ---------------------------------------------------------------------------
987class StepRunner:
988 """
989 Ajaa yksittäisen step:n sen typin mukaan.
991 run_context: dict joka lisätään template-kontekstiin. Esim:
992 {"budget": {"max_parallel": 16, ...}, "config": {...}}
993 """
995 def __init__(
996 self,
997 resolver: AgentResolver,
998 builtins: BuiltinExecutor | None = None,
999 invoke_agent_fn: Any | None = None,
1000 run_context: dict | None = None,
1001 *,
1002 envs: dict | None = None,
1003 command_executor: CommandExecutor | None = None,
1004 tracer: Any | None = None,
1005 ) -> None:
1006 self._resolver = resolver
1007 self._builtins = builtins or _DEFAULT_BUILTINS
1008 self._invoke_agent = self._wrap_hard_failure_gate(
1009 invoke_agent_fn or self._dummy_invoke
1010 )
1011 self._run_context: dict = run_context or {}
1012 self._envs: dict = envs or {}
1013 self._command_executor = command_executor or CommandExecutor()
1014 self._tracer = tracer
1016 def set_envs(self, envs: dict) -> None:
1017 """Bind the pipeline's spec.envs map (env names → image refs) for
1018 command-step env validation. Called by GraphBuilder per pipeline."""
1019 self._envs = envs
1021 @staticmethod
1022 def _should_skip_llm(step: dict, ctx: dict) -> bool:
1023 """Truthiness of a step's `skip_llm_if` expression (ENG-12, Q4).
1025 Lenient (non-strict) evaluation: a missing/undefined field resolves to
1026 falsy, so the step behaves normally (LLM invoked). Only an explicit
1027 truthy expression skips the LLM.
1028 """
1029 raw = step.get("skip_llm_if")
1030 if raw is None:
1031 return False
1032 resolved = _resolve_template(raw, ctx)
1033 if isinstance(resolved, bool):
1034 return resolved
1035 if isinstance(resolved, str):
1036 return resolved.lower() not in ('false', '0', '', 'none', 'null')
1037 return bool(resolved)
1039 @staticmethod
1040 def _skip_llm_passthrough(step: dict, state: PipelineState, ctx: dict) -> PipelineState:
1041 """Deterministic passthrough of a step's resolved input (no LLM).
1043 `state[step_id]` becomes the resolved `input` value (the parallel
1044 fan-out envelope `{results: [...]}` when the input resolves to a list).
1045 """
1046 step_id = step.get("id", "?")
1047 input_spec = step.get("input")
1048 resolved = _resolve_input(input_spec, state, ctx) if input_spec is not None else None
1049 if isinstance(resolved, list):
1050 log.debug("[step] %s skip_llm_if passthrough (fanout, no LLM)", step_id)
1051 return {**state, step_id: {"results": resolved}}
1052 log.debug("[step] %s skip_llm_if passthrough (no LLM)", step_id)
1053 return {**state, step_id: resolved}
1055 def _make_ctx(self, state: dict, **extras: Any) -> dict:
1056 """Rakentaa template-kontekstin: state + run_context + extras."""
1057 ctx: dict = {}
1058 ctx.update(state)
1059 ctx.update(self._run_context)
1060 ctx.update(extras)
1061 return ctx
1063 def _resolve_step_runtime(
1064 self,
1065 step: dict,
1066 state: dict,
1067 ctx: dict,
1068 *,
1069 input_override: Any = _UNSET,
1070 extra_params: dict | None = None,
1071 ) -> dict:
1072 """
1073 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
1074 templaattien osalta state + run_context -kontekstia vasten.
1076 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
1077 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
1078 """
1079 resolved: dict = dict(step)
1081 if input_override is not _UNSET:
1082 resolved["input"] = input_override
1083 else:
1084 input_spec = step.get("input")
1085 if input_spec is not None:
1086 resolved["input"] = _resolve_input(input_spec, state, ctx)
1088 params_spec = step.get("params")
1089 if params_spec or extra_params:
1090 base = _resolve_template(params_spec, ctx) if params_spec else {}
1091 if not isinstance(base, dict):
1092 base = {}
1093 if extra_params:
1094 base = {**base, **extra_params}
1095 resolved["params"] = base
1097 return resolved
1099 @staticmethod
1100 def _dummy_invoke(
1101 resolved_spec: dict, step: dict, state: PipelineState
1102 ) -> PipelineState:
1103 """Placeholder — ei kutsu oikeaa LLM:ää."""
1104 step_id = step.get("id", "?")
1105 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
1106 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
1108 @staticmethod
1109 def _merge_errors(base: list, additions: list) -> list:
1110 """Return a new list containing all items from *base* plus any items from
1111 *additions* that are not already present. Order is preserved and no
1112 in-place mutation occurs — safe to call from any context."""
1113 if not additions:
1114 return base
1115 merged = list(base)
1116 for err in additions:
1117 if err not in merged:
1118 merged.append(err)
1119 return merged
1121 @staticmethod
1122 def _wrap_hard_failure_gate(invoke_fn: Any) -> Any:
1123 """Kietoo invokerin: jokaiseen agent-step-tulokseen sovelletaan
1124 kova-virhe-portti (RunHardFailure details.error_code-perusteella).
1125 Suppein funnel joka näkee KAIKKI agent-tulokset: sequential,
1126 parallel (single + fanout), foreach (body + agent-polku) ja while."""
1127 def invoke_with_gate(
1128 resolved_spec: dict, step: dict, state: PipelineState
1129 ) -> PipelineState:
1130 result = invoke_fn(resolved_spec, step, state)
1131 step_id = step.get("id", "?")
1132 StepRunner._raise_on_hard_error(result.get(step_id), step_id)
1133 return result
1134 return invoke_with_gate
1136 @staticmethod
1137 def _is_hard_fail(details: dict) -> bool:
1138 """Kova virhe vain kun: error_code on HARD_ERROR_CODESissa
1139 (TRUNCATION_ERROR) tai VALIDATION_ERROR kantaa kontekstiylityksen
1140 signatuurin (assembled_tokens + context_budget —
1141 _llm._context_overflow_result). Generic-VALIDATION_ERROR ei kaada
1142 ajoa."""
1143 code = details.get("error_code")
1144 if code in RunHardFailure.HARD_ERROR_CODES:
1145 return True
1146 if code == "VALIDATION_ERROR":
1147 return {"assembled_tokens", "context_budget"} <= details.keys()
1148 return False
1150 @staticmethod
1151 def _raise_on_hard_error(step_result: Any, step_id: str) -> None:
1152 """Nosta RunHardFailure jos step-tulos kantaa kovan virheen
1153 (_is_hard_fail); stepin details kulkeutuu poikkeuksen mukana."""
1154 if not isinstance(step_result, dict):
1155 return
1156 details = step_result.get("details")
1157 if not isinstance(details, dict):
1158 return
1159 if not StepRunner._is_hard_fail(details):
1160 return
1161 code = details.get("error_code")
1162 err = step_result.get("error", "")
1163 message = err if isinstance(err, str) and err else f"{code}: step '{step_id}' hard failure"
1164 raise RunHardFailure(code, message, details)
1166 @staticmethod
1167 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
1168 """
1169 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
1170 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
1171 epäonnistuneeksi eikä hiljennä virhettä.
1173 Kaikki virhetyypit propagoidaan:
1174 - budget:max_llm_calls / budget:run_timeout → budjettiloki
1175 - parse_error:*, timeout:*, ja muut → error-loki
1176 """
1177 step_result = result.get(step_id)
1178 if not isinstance(step_result, dict):
1179 return result
1180 err = step_result.get("error", "")
1181 if not isinstance(err, str) or not err:
1182 return result
1183 errors = list(result.get("errors") or [])
1184 message = f"{step_id}: {err}"
1185 if message not in errors:
1186 errors.append(message)
1187 if err.startswith("budget:"):
1188 log.warning("[budget] %s", message)
1189 else:
1190 log.error("[step-error] %s", message)
1191 return {**result, "errors": errors}
1193 def run_step(self, step: dict, state: PipelineState) -> PipelineState:
1194 step_id = step.get("id", "?")
1195 ctx = self._make_ctx(state)
1196 step_type = step.get("type", "sequential")
1198 # Evaluate condition — if false, skip step. A `while` whose entry
1199 # condition is false writes a first-class skip result (r.68) instead of
1200 # returning state unchanged, so `_run_while` owns its skip path.
1201 cond = step.get("condition")
1202 if cond is not None and not _eval_condition(cond, ctx) and step_type != "while":
1203 log.debug("[step] %s skipped (condition false)", step_id)
1204 return state
1206 if step_type == "builtin":
1207 component = step.get("component", "passthrough")
1208 log.debug("[step] %s builtin:%s", step_id, component)
1209 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
1210 # template-arvot (esim. "{{ budget.min_severity_for_ticket }}")
1211 resolved_step = self._resolve_step_runtime(step, state, ctx)
1212 builtin_state = self._builtins.execute(component, resolved_step, state)
1213 self._raise_on_hard_error(builtin_state.get(step_id), step_id)
1214 return builtin_state
1216 # skip_llm_if (ENG-12, Q4): when the step's skip_llm_if evaluates truthy,
1217 # do NOT invoke the LLM (no inference budget consumed) and set the step
1218 # result to a deterministic passthrough of the step's resolved input.
1219 if (
1220 step_type in ("sequential", "parallel")
1221 and self._should_skip_llm(step, ctx)
1222 ):
1223 return self._skip_llm_passthrough(step, state, ctx)
1225 if step_type == "parallel":
1226 # Fan-out: if input resolves to a list, call agent once per item
1227 input_spec = step.get("input")
1228 if input_spec is not None:
1229 input_val = _resolve_input(input_spec, state, ctx)
1230 if isinstance(input_val, list):
1231 return self._run_parallel_fanout(step, state, input_val, ctx)
1232 # Fallback: single agent call
1233 agent_name = step.get("agent")
1234 if not agent_name:
1235 log.warning("[step] %s has no agent — skipping", step_id)
1236 return state
1237 resolved = self._resolver.resolve(agent_name)
1238 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
1239 resolved_step = self._resolve_step_runtime(step, state, ctx)
1240 result = self._invoke_agent(resolved, resolved_step, state)
1241 return self._propagate_step_errors(result, step_id)
1243 if step_type == "sequential":
1244 agent_name = step.get("agent")
1245 if not agent_name:
1246 log.warning("[step] %s has no agent — skipping", step_id)
1247 return state
1248 resolved = self._resolver.resolve(agent_name)
1249 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
1250 resolved_step = self._resolve_step_runtime(step, state, ctx)
1251 result = self._invoke_agent(resolved, resolved_step, state)
1252 return self._propagate_step_errors(result, step_id)
1254 if step_type == "foreach":
1255 return self._run_foreach(step, state)
1257 if step_type == "while":
1258 return self._run_while(step, state)
1260 if step_type == "command":
1261 return self._run_command_step(step, state)
1263 if step_type == "call":
1264 # Build-time expansion (ADR-0025 D4) is a later phase; a call step
1265 # reaching the runtime is wiring that is not yet expanded.
1266 raise ValueError(
1267 f"call step '{step_id}' reached runtime without build-time "
1268 "expansion (ADR-0025) — not yet supported"
1269 )
1271 if step_type == "spawn":
1272 return self._run_spawn(step, state)
1274 raise ValueError(f"Unknown step type '{step_type}' in step '{step_id}'")
1276 def _run_parallel_fanout(
1277 self,
1278 step: dict,
1279 state: PipelineState,
1280 items: list,
1281 ctx: dict,
1282 ) -> PipelineState:
1283 """
1284 Fan-out: call agent once per item, concurrently.
1285 Results stored as state[step_id] = {"results": [item_result, ...]}.
1286 Each item is available as state["item"] inside the invoke call.
1287 """
1288 step_id = step.get("id", "?")
1289 agent_name = step.get("agent")
1290 if not agent_name:
1291 log.warning("[step] %s parallel fanout: no agent", step_id)
1292 return {**state, step_id: {"results": []}}
1294 resolved = self._resolver.resolve(agent_name)
1296 # Apply max_items cap
1297 max_items_raw = step.get("max_items")
1298 if max_items_raw is not None:
1299 max_items = _resolve_int(max_items_raw, ctx, len(items))
1300 items = items[:max_items]
1302 # Concurrency limit
1303 max_conc_raw = step.get("max_concurrent")
1304 if max_conc_raw is not None:
1305 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
1306 else:
1307 max_conc = len(items) or 1
1308 max_conc = max(1, min(max_conc, len(items) or 1))
1310 log.debug(
1311 "[step] %s parallel fanout: %d items, %d concurrent",
1312 step_id, len(items), max_conc,
1313 )
1315 # Pre-fetch path verification map if present in state (from path-verifier builtin)
1316 verify_by_path: dict = {}
1317 verify_step = state.get("verify_paths")
1318 if isinstance(verify_step, dict):
1319 vbp = verify_step.get("by_path")
1320 if isinstance(vbp, dict):
1321 verify_by_path = vbp
1323 def call_one(item: Any) -> tuple[Any, dict | None]:
1324 item_state = {**state, "_item": item, "item": item}
1325 item_ctx = self._make_ctx(item_state)
1326 # Per-item path verification block injection (matches doc_files[*].path)
1327 extra: dict = {}
1328 if isinstance(item, dict) and verify_by_path:
1329 p = item.get("path")
1330 if p and p in verify_by_path:
1331 extra["path_verification_block"] = verify_by_path[p]
1332 resolved_step = self._resolve_step_runtime(
1333 step, item_state, item_ctx,
1334 input_override=item,
1335 extra_params=extra or None,
1336 )
1337 result_state = self._invoke_agent(resolved, resolved_step, item_state)
1338 return result_state.get(step_id), result_state.get("step_context_metrics")
1340 # Pre-allocate results to preserve input order regardless of completion
1341 # order (ENG-07), matching the concurrent foreach body-polku.
1342 results: list = [None] * len(items)
1343 item_metrics: dict = {}
1344 with ThreadPoolExecutor(max_workers=max_conc) as pool:
1345 futures = {pool.submit(call_one, item): i for i, item in enumerate(items)}
1346 for future in as_completed(futures):
1347 idx = futures[future]
1348 try:
1349 item_result, item_metrics_map = future.result()
1350 results[idx] = item_result
1351 if isinstance(item_metrics_map, dict):
1352 item_metrics.update(item_metrics_map)
1353 except SkillLoadError:
1354 raise
1355 except RunHardFailure:
1356 pool.shutdown(wait=False, cancel_futures=True)
1357 raise
1358 except Exception as exc: # noqa: BLE001
1359 log.error("[step] %s item failed: %s", step_id, exc)
1360 results[idx] = {"error": str(exc)}
1362 final: PipelineState = {**state, step_id: {"results": results}}
1363 if item_metrics:
1364 final = {
1365 **final,
1366 "step_context_metrics": _merge_metrics(
1367 state.get("step_context_metrics"), item_metrics
1368 ),
1369 }
1370 return final
1372 def _run_foreach(self, step: dict, state: PipelineState) -> PipelineState:
1373 """
1374 Foreach: iterate over input list, run body steps per item.
1375 Input from step.input (dot-path or template) or step.over (state key).
1377 When max_concurrent > 1, body iterations run concurrently via
1378 ThreadPoolExecutor — each item gets its own worker thread. The
1379 max_concurrent value sets the thread pool size (mirrors parallel
1380 fan-out semantics). Results list preserves original item order.
1382 max_output budget in concurrent mode: a stop flag prevents new workers
1383 from starting after the budget is exhausted. Workers already executing
1384 when the flag is set will complete naturally — at most max_concurrent
1385 extra items may run past the limit. This is accepted behaviour; the
1386 sequential early-exit guarantee cannot be reproduced without sacrificing
1387 true concurrency.
1388 """
1389 step_id = step.get("id", "?")
1390 ctx = self._make_ctx(state)
1392 # Determine items
1393 input_spec = step.get("input")
1394 over_key = step.get("over")
1396 if input_spec is not None:
1397 items: list = _resolve_input(input_spec, state, ctx) or []
1398 if not isinstance(items, list):
1399 items = [items] if items is not None else []
1400 elif over_key:
1401 items = state.get(over_key, [None])
1402 if not isinstance(items, list):
1403 items = [items]
1404 else:
1405 items = [None]
1407 # max_iterations cap
1408 max_iter_raw = step.get("max_iterations")
1409 if max_iter_raw is not None:
1410 max_iter = _resolve_int(max_iter_raw, ctx, len(items))
1411 items = items[:max_iter]
1413 # max_output cap — stop once N iterations have produced a successful
1414 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
1415 max_output_raw = step.get("max_output")
1416 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
1418 # max_concurrent — when > 1, enables parallel body execution
1419 max_conc_raw = step.get("max_concurrent")
1420 if max_conc_raw is not None:
1421 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
1422 max_conc = max(1, min(max_conc, len(items) or 1))
1423 else:
1424 max_conc = 1
1426 body = step.get("body") or []
1427 agent_name = step.get("agent")
1429 if max_conc <= 1:
1430 return self._run_foreach_sequential(
1431 step_id, step, state, items, body, agent_name, max_output,
1432 )
1433 return self._run_foreach_concurrent(
1434 step_id, step, state, items, body, agent_name, max_output, max_conc,
1435 )
1437 @staticmethod
1438 def _body_item_result(iter_state: PipelineState, declared_output: str | None) -> Any:
1439 """body-polku per-iteration result (r.67): the declared body-output
1440 step's result; else the iteration's `publish` result; else None."""
1441 if declared_output:
1442 return iter_state.get(declared_output)
1443 return iter_state.get("publish")
1445 @staticmethod
1446 def _body_item_passed(item: Any) -> bool:
1447 """Derive a body-polku item verdict from its key: verdict PASS/FAIL or a
1448 `passed` bool; an item lacking such a key is not counted as failed."""
1449 if isinstance(item, dict):
1450 verdict = item.get("verdict")
1451 if verdict == "FAIL":
1452 return False
1453 if verdict == "PASS":
1454 return True
1455 passed = item.get("passed")
1456 if isinstance(passed, bool):
1457 return passed
1458 return True
1460 @staticmethod
1461 def _aggregate_body_results(item_results: list) -> dict:
1462 """body-polku aggregate (r.67): {results, passed, failing, failed}."""
1463 failing = [r for r in item_results if not StepRunner._body_item_passed(r)]
1464 return {
1465 "results": item_results,
1466 "passed": not failing,
1467 "failing": failing,
1468 "failed": bool(failing),
1469 }
1471 def _run_foreach_sequential(
1472 self,
1473 step_id: str,
1474 step: dict,
1475 state: PipelineState,
1476 items: list,
1477 body: list,
1478 agent_name: str | None,
1479 max_output: int,
1480 ) -> PipelineState:
1481 """Sequential foreach — default path when max_concurrent is absent or 1."""
1482 body_polku = bool(body)
1483 declared_output = step.get("output")
1484 results: list = []
1485 produced_outputs = 0
1486 accumulated_errors: list = list(state.get("errors") or [])
1487 accumulated_metrics: dict = {}
1489 for item in items:
1490 if max_output and produced_outputs >= max_output:
1491 log.info("[foreach] %s max_output=%d reached — stopping", step_id, max_output)
1492 break
1493 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
1495 if body_polku:
1496 for sub_step in body:
1497 iter_state = self.run_step(sub_step, iter_state)
1498 # Propagate any budget errors from this iteration to accumulated list
1499 accumulated_errors = self._merge_errors(
1500 accumulated_errors, iter_state.get("errors") or []
1501 )
1502 publish_out = iter_state.get("publish")
1503 results.append(self._body_item_result(iter_state, declared_output))
1504 if isinstance(publish_out, dict) and publish_out.get("path"):
1505 produced_outputs += 1
1506 elif agent_name:
1507 resolved = self._resolver.resolve(agent_name)
1508 iter_state = self._invoke_agent(resolved, step, iter_state)
1509 results.append(iter_state.get(step_id))
1510 item_metrics_map = iter_state.get("step_context_metrics")
1511 if isinstance(item_metrics_map, dict):
1512 accumulated_metrics.update(item_metrics_map)
1514 step_result: Any = self._aggregate_body_results(results) if body_polku else results
1515 final: PipelineState = {**state, step_id: step_result}
1516 if accumulated_metrics:
1517 final = {
1518 **final,
1519 "step_context_metrics": _merge_metrics(
1520 state.get("step_context_metrics"), accumulated_metrics
1521 ),
1522 }
1523 if accumulated_errors != list(state.get("errors") or []):
1524 final = {**final, "errors": accumulated_errors}
1525 return final
1527 def _run_foreach_concurrent(
1528 self,
1529 step_id: str,
1530 step: dict,
1531 state: PipelineState,
1532 items: list,
1533 body: list,
1534 agent_name: str | None,
1535 max_output: int,
1536 max_conc: int,
1537 ) -> PipelineState:
1538 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
1539 log.info(
1540 "[foreach] %s concurrent: %d items, max_concurrent=%d",
1541 step_id, len(items), max_conc,
1542 )
1544 # Pre-allocate results list to preserve item order across threads.
1545 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
1546 # legitimate None results (body produced no output) are kept.
1547 body_polku = bool(body)
1548 declared_output = step.get("output")
1549 _SKIPPED = object()
1550 results: list = [_SKIPPED] * len(items)
1551 lock = threading.Lock()
1552 produced_outputs = [0] # mutable counter shared across threads
1553 stop_flag = threading.Event() # set when max_output budget is exhausted
1554 item_errors: list = [] # budget errors collected from worker threads
1555 accumulated_metrics: dict = {} # per-item envelopes from worker threads
1557 def run_item(idx: int, item: Any) -> None:
1558 nonlocal item_errors
1559 # Honour stop flag — skip if budget already exhausted before we start
1560 if max_output and stop_flag.is_set():
1561 return
1563 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
1565 if body_polku:
1566 for sub_step in body:
1567 iter_state = self.run_step(sub_step, iter_state)
1568 # Collect budget errors from this iteration (thread-safe merge)
1569 new_errs = iter_state.get("errors") or []
1570 if new_errs:
1571 with lock:
1572 item_errors = self._merge_errors(item_errors, new_errs)
1573 publish_out = iter_state.get("publish")
1574 results[idx] = self._body_item_result(iter_state, declared_output)
1575 if isinstance(publish_out, dict) and publish_out.get("path"):
1576 with lock:
1577 produced_outputs[0] += 1
1578 if max_output and produced_outputs[0] >= max_output:
1579 stop_flag.set()
1580 elif agent_name:
1581 resolved = self._resolver.resolve(agent_name)
1582 iter_state = self._invoke_agent(resolved, step, iter_state)
1583 results[idx] = iter_state.get(step_id)
1584 item_metrics_map = iter_state.get("step_context_metrics")
1585 if isinstance(item_metrics_map, dict):
1586 with lock:
1587 accumulated_metrics.update(item_metrics_map)
1589 with ThreadPoolExecutor(max_workers=max_conc) as pool:
1590 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
1591 for future in as_completed(futures):
1592 try:
1593 future.result()
1594 except SkillLoadError:
1595 raise
1596 except RunHardFailure:
1597 pool.shutdown(wait=False, cancel_futures=True)
1598 raise
1599 except Exception as exc: # noqa: BLE001
1600 idx = futures[future]
1601 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
1602 results[idx] = {"error": str(exc)}
1604 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
1605 ordered = [r for r in results if r is not _SKIPPED]
1606 step_result: Any = self._aggregate_body_results(ordered) if body_polku else ordered
1607 final: PipelineState = {**state, step_id: step_result}
1608 if accumulated_metrics:
1609 final = {
1610 **final,
1611 "step_context_metrics": _merge_metrics(
1612 state.get("step_context_metrics"), accumulated_metrics
1613 ),
1614 }
1615 if item_errors:
1616 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
1617 return final
1619 def _run_while(self, step: dict, state: PipelineState) -> PipelineState:
1620 """
1621 While loop with exit_condition and optional entry condition.
1623 Fields:
1624 condition: "{{ triage.approved }}" — if false, skip entirely
1625 input: initial value for loop.current
1626 exit_condition: "{{ review.pass }}" — break when true
1627 max_iterations: int or template
1628 output: body step id whose result feeds the next round's
1629 `loop.current` (envelope-unwrapped when dict with `output`)
1630 body: list of sub-steps
1632 Result shape (r.68): {passed, iterations, output, exhausted, skipped}.
1633 The feed comes ONLY from the declared `output` body step — there is no
1634 hardcoded `refine` feed; a body step named `refine` is not special.
1635 """
1636 step_id = step.get("id", "?")
1637 ctx = self._make_ctx(state)
1639 # Entry condition — skip is a first-class state (r.68): a skipped
1640 # while writes a result with output = the resolved input so downstream
1641 # `.output` reads are non-null on the skipped path.
1642 entry_cond = step.get("condition")
1643 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
1644 input_spec = step.get("input")
1645 skip_input = _resolve_input(input_spec, state, ctx) if input_spec is not None else None
1646 return {**state, step_id: {
1647 "passed": True,
1648 "iterations": 0,
1649 "output": skip_input,
1650 "exhausted": False,
1651 "skipped": True,
1652 }}
1654 # Initial current value from input
1655 input_spec = step.get("input")
1656 if input_spec is not None:
1657 current = _resolve_input(input_spec, state, ctx)
1658 else:
1659 current = None
1661 exit_condition = step.get("exit_condition")
1662 max_iter = _resolve_int(step.get("max_iterations", 2), ctx, 2)
1663 body = step.get("body") or []
1664 declared_output = step.get("output")
1666 passed = False
1667 iteration = 0
1668 loop_state: PipelineState = dict(state)
1670 while iteration < max_iter:
1671 # Inject loop context
1672 loop_state = {
1673 **state,
1674 "loop": {"current": current, "iteration": iteration},
1675 step_id: {"current": current, "passed": False},
1676 }
1677 # Add current item context (from outer foreach if any)
1678 if "item" in state:
1679 loop_state["item"] = state["item"]
1681 # Run body steps
1682 for sub_step in body:
1683 loop_state = self.run_step(sub_step, loop_state)
1685 iteration += 1
1687 # Feed: next round's loop.current comes from the declared body-output
1688 # step (envelope-unwrapped); otherwise current is unchanged.
1689 # Envelopes: a `{output: ...}` (r.68) and a `{results: [...]}`
1690 # aggregate (parallel/foreach body-step) are unwrapped to their
1691 # inner value so the next round's consumer receives the list, not
1692 # the envelope. A raw value (plain list/dict) feeds as-is.
1693 if declared_output:
1694 feed_value = loop_state.get(declared_output)
1695 if isinstance(feed_value, dict) and "output" in feed_value:
1696 current = feed_value["output"]
1697 elif isinstance(feed_value, dict) and isinstance(feed_value.get("results"), list):
1698 current = feed_value["results"]
1699 else:
1700 current = feed_value
1702 # Check exit condition
1703 if exit_condition:
1704 exit_ctx = self._make_ctx(loop_state)
1705 if _eval_condition(exit_condition, exit_ctx):
1706 passed = True
1707 break
1708 else:
1709 passed = True
1710 break
1712 skipped = False
1713 exhausted = not passed
1714 final: PipelineState = {**state, step_id: {
1715 "passed": passed,
1716 "iterations": iteration,
1717 "output": current,
1718 "exhausted": exhausted,
1719 "skipped": skipped,
1720 }}
1721 # ENG-11: a while that ends exhausted (ran max_iterations without
1722 # exit_condition passing, not skipped) and whose `fail_if_exhausted`
1723 # evaluates true is a hard run failure — nothing downstream runs. A
1724 # skipped or passed while must never raise.
1725 fail_if_exhausted = step.get("fail_if_exhausted")
1726 if (
1727 fail_if_exhausted is not None
1728 and exhausted
1729 and _eval_condition(fail_if_exhausted, self._make_ctx(state))
1730 ):
1731 raise RunHardFailure(
1732 "EXHAUSTED",
1733 f"while step '{step_id}' exhausted max_iterations={max_iter} "
1734 "without exit_condition passing (fail_if_exhausted)",
1735 {"step_id": step_id, "exhausted": True},
1736 )
1737 # Propagate any errors accumulated in the loop body back to the outer state
1738 loop_errors = loop_state.get("errors") or []
1739 if loop_errors:
1740 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), loop_errors)}
1741 return final
1743 def _run_spawn(self, step: dict, state: PipelineState) -> PipelineState:
1744 """Run a spawn step: a recursive node tree of agent calls (S2).
1746 The `agent` runs once as director at the root and may emit a `children`
1747 plan; every deeper node re-runs the `child` role (default = `agent`)
1748 with that node's own `input`. Depth and fan are bounded by the
1749 spawn_policy, and the whole tree draws from one shared SpawnBudget.
1750 """
1751 step_id = step.get("id", "?")
1752 ctx = self._make_ctx(state)
1753 policy = step.get("spawn_policy") or {}
1755 max_depth_raw = policy.get("max_depth")
1756 if max_depth_raw is None:
1757 max_depth = _SPAWN_DEFAULT_MAX_DEPTH
1758 else:
1759 max_depth = _resolve_int(max_depth_raw, ctx, _SPAWN_DEFAULT_MAX_DEPTH)
1761 max_spawn_raw = policy.get("max_spawn_per_node")
1762 if max_spawn_raw is None:
1763 max_spawn_per_node = _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE
1764 else:
1765 max_spawn_per_node = _resolve_int(max_spawn_raw, ctx, _SPAWN_DEFAULT_MAX_SPAWN_PER_NODE)
1767 max_calls_raw = policy.get("max_agent_calls")
1768 if max_calls_raw is None:
1769 budget = SpawnBudget(_SPAWN_DEFAULT_MAX_AGENT_CALLS)
1770 else:
1771 budget = SpawnBudget(_resolve_int(max_calls_raw, ctx, _SPAWN_DEFAULT_MAX_AGENT_CALLS))
1773 run_max_parallel = int((self._run_context.get("budget") or {}).get("max_parallel", 8))
1774 max_concurrent_raw = policy.get("max_concurrent_children")
1775 if max_concurrent_raw is None:
1776 max_concurrent_children = run_max_parallel
1777 else:
1778 max_concurrent_children = _resolve_int(max_concurrent_raw, ctx, run_max_parallel)
1780 director = step.get("agent")
1781 if not director:
1782 log.warning("[step] %s spawn has no agent — skipping", step_id)
1783 return state
1784 child_role = step.get("child") or director
1786 envelope, metrics = self._spawn_node(
1787 node_id=step_id,
1788 role=director,
1789 depth=0,
1790 step=step,
1791 input_override=_UNSET,
1792 base_state=state,
1793 base_ctx=ctx,
1794 child_role=child_role,
1795 budget=budget,
1796 max_depth=max_depth,
1797 max_spawn_per_node=max_spawn_per_node,
1798 max_concurrent_children=max_concurrent_children,
1799 )
1800 final: PipelineState = {**state, step_id: envelope}
1801 if metrics:
1802 final = {
1803 **final,
1804 "step_context_metrics": _merge_metrics(
1805 state.get("step_context_metrics"), metrics
1806 ),
1807 }
1808 return final
1810 def _spawn_node(
1811 self,
1812 *,
1813 node_id: str,
1814 role: str,
1815 depth: int,
1816 step: dict,
1817 input_override: Any,
1818 base_state: PipelineState,
1819 base_ctx: dict,
1820 child_role: str,
1821 budget: SpawnBudget,
1822 max_depth: int,
1823 max_spawn_per_node: int | None,
1824 max_concurrent_children: int,
1825 ) -> tuple[dict, dict]:
1826 """Run one spawn-tree node; returns (node envelope, subtree metrics)."""
1827 tracer = self._tracer
1828 parent_obs = (
1829 tracer.open_parent_observation(
1830 name=f"spawn:{node_id}",
1831 input={"node_id": node_id, "role": role, "depth": depth},
1832 )
1833 if tracer
1834 else None
1835 )
1836 try:
1837 resolved_role = self._resolver.resolve(role)
1838 resolved_step = self._resolve_step_runtime(
1839 step, base_state, base_ctx, input_override=input_override,
1840 )
1841 if not budget.consume():
1842 raise RunHardFailure(
1843 "SPAWN_BUDGET_EXHAUSTED",
1844 f"spawn tree budget exhausted at node '{node_id}'",
1845 {"node_id": node_id, "depth": depth},
1846 )
1847 result_state = self._invoke_agent(resolved_role, resolved_step, base_state)
1848 this_result = result_state.get(node_id)
1849 metrics = dict(result_state.get("step_context_metrics") or {})
1851 children = self._spawn_children_from_result(this_result)
1852 envelope: dict = {
1853 "result": this_result,
1854 "children": {},
1855 "passed": not self._spawn_is_soft_fail(this_result),
1856 "failing": [],
1857 "exhausted": False,
1858 }
1860 if not children:
1861 return envelope, metrics
1862 if depth + 1 > max_depth:
1863 envelope["exhausted"] = True
1864 return envelope, metrics
1865 if max_spawn_per_node is not None and len(children) > max_spawn_per_node:
1866 raise RunHardFailure(
1867 "SPAWN_OVER_FLIGHT",
1868 f"spawn node '{node_id}' emitted {len(children)} children, "
1869 f"exceeds max_spawn_per_node={max_spawn_per_node}",
1870 {
1871 "node_id": node_id,
1872 "depth": depth,
1873 "children": len(children),
1874 "max_spawn_per_node": max_spawn_per_node,
1875 },
1876 )
1878 pool_size = self._spawn_pool_size(max_spawn_per_node, max_concurrent_children)
1879 if len(children) <= 1 or pool_size <= 1:
1880 for idx, child in enumerate(children):
1881 child_id, child_envelope, child_metrics = self._spawn_child(
1882 node_id=node_id,
1883 child_role=child_role,
1884 depth=depth,
1885 step=step,
1886 base_state=base_state,
1887 base_ctx=base_ctx,
1888 budget=budget,
1889 max_depth=max_depth,
1890 max_spawn_per_node=max_spawn_per_node,
1891 max_concurrent_children=max_concurrent_children,
1892 child=child,
1893 idx=idx,
1894 )
1895 metrics = _merge_metrics(metrics, child_metrics)
1896 envelope["children"][child_id] = child_envelope
1897 if not child_envelope["passed"]:
1898 envelope["passed"] = False
1899 envelope["failing"].append(child_id)
1900 return envelope, metrics
1902 pool = ThreadPoolExecutor(max_workers=pool_size)
1903 cancel = threading.Event()
1904 futures = []
1905 for idx, child in enumerate(children):
1906 ctx = contextvars.copy_context()
1907 futures.append(
1908 pool.submit(
1909 ctx.run,
1910 self._spawn_child,
1911 node_id=node_id,
1912 child_role=child_role,
1913 depth=depth,
1914 step=step,
1915 base_state=base_state,
1916 base_ctx=base_ctx,
1917 budget=budget,
1918 max_depth=max_depth,
1919 max_spawn_per_node=max_spawn_per_node,
1920 max_concurrent_children=max_concurrent_children,
1921 child=child,
1922 idx=idx,
1923 )
1924 )
1925 hard_error: Exception | None = None
1926 try:
1927 for fut in as_completed(futures):
1928 try:
1929 child_id, child_envelope, child_metrics = fut.result()
1930 except (RunHardFailure, SkillLoadError) as exc:
1931 hard_error = exc
1932 cancel.set()
1933 for pending in futures:
1934 pending.cancel()
1935 break
1936 metrics = _merge_metrics(metrics, child_metrics)
1937 envelope["children"][child_id] = child_envelope
1938 if not child_envelope["passed"]:
1939 envelope["passed"] = False
1940 envelope["failing"].append(child_id)
1941 finally:
1942 pool.shutdown(wait=False, cancel_futures=True)
1943 if hard_error is not None:
1944 raise hard_error
1946 return envelope, metrics
1947 finally:
1948 if tracer is not None and parent_obs is not None:
1949 tracer.close_parent_observation(parent_obs)
1951 @staticmethod
1952 def _spawn_pool_size(max_spawn_per_node: int | None, max_concurrent_children: int) -> int:
1953 if max_spawn_per_node is None:
1954 return max(1, max_concurrent_children)
1955 return max(1, min(max_spawn_per_node, max_concurrent_children))
1957 def _spawn_child(
1958 self,
1959 *,
1960 node_id: str,
1961 child_role: str,
1962 depth: int,
1963 step: dict,
1964 base_state: PipelineState,
1965 base_ctx: dict,
1966 budget: SpawnBudget,
1967 max_depth: int,
1968 max_spawn_per_node: int | None,
1969 max_concurrent_children: int,
1970 child: Any,
1971 idx: int,
1972 ) -> tuple[str, dict, dict]:
1973 if not self._spawn_has_input(child):
1974 raise RunHardFailure(
1975 "INVALID_CHILD_PLAN",
1976 f"spawn child #{idx} of node '{node_id}' must be a dict "
1977 "with a non-empty 'input'",
1978 {"node_id": node_id, "child_index": idx, "child": repr(child)[:200]},
1979 )
1980 child_id = child["id"] if child.get("id") else f"{node_id}.{idx}"
1981 child_input = child["input"]
1982 child_envelope, child_metrics = self._spawn_node(
1983 node_id=child_id,
1984 role=child_role,
1985 depth=depth + 1,
1986 step={"id": child_id, "input": child_input},
1987 input_override=child_input,
1988 base_state=base_state,
1989 base_ctx=base_ctx,
1990 child_role=child_role,
1991 budget=budget,
1992 max_depth=max_depth,
1993 max_spawn_per_node=max_spawn_per_node,
1994 max_concurrent_children=max_concurrent_children,
1995 )
1996 return child_id, child_envelope, child_metrics
1998 @staticmethod
1999 def _spawn_children_from_result(result: Any) -> list:
2000 """The node's child plan: [] when absent or empty; raises on a malformed plan."""
2001 if not isinstance(result, dict):
2002 return []
2003 raw = result.get("children")
2004 if raw is None:
2005 return []
2006 if not isinstance(raw, list):
2007 raise RunHardFailure(
2008 "INVALID_CHILD_PLAN",
2009 "spawn agent 'children' must be a list",
2010 {"children": repr(raw)[:200]},
2011 )
2012 return raw
2014 @staticmethod
2015 def _spawn_has_input(child: Any) -> bool:
2016 """A valid child plan item: a dict carrying a non-empty 'input'."""
2017 if not isinstance(child, dict):
2018 return False
2019 input_val = child.get("input")
2020 if input_val is None:
2021 return False
2022 if isinstance(input_val, str) and not input_val.strip():
2023 return False
2024 if isinstance(input_val, (list, dict)) and not input_val:
2025 return False
2026 return True
2028 @staticmethod
2029 def _spawn_is_soft_fail(result: Any) -> bool:
2030 """A soft/gap agent outcome (escalate, parse_error, error, budget:*) — not a hard error."""
2031 if not isinstance(result, dict):
2032 return False
2033 if _SPAWN_SOFT_FAIL_KEYS & result.keys():
2034 return True
2035 return any(key.startswith("budget:") for key in result)
2037 def _run_command_step(self, step: dict, state: PipelineState) -> PipelineState:
2038 """Execute a manifest command step (type: command).
2040 Validation (ADR-0012 — no silent assumptions, fail-fast):
2041 - 'command' is required and must be a non-empty string
2042 - 'env' (optional) must name a key in the pipeline's spec.envs map;
2043 an unknown name is a hard error listing the available names
2044 Execution:
2045 - env absent → engine container (subprocess, `sh -c`) — documented
2046 default, recorded explicitly in the step result (invariant b7)
2047 - env present → env container via Kubernetes exec (mechanic A):
2048 own pod (hostname), container slug(env), `sh -c <command>`
2049 Result stored at state[step_id]: {exit_code, stdout, stderr, env,
2050 executed_in}. A non-zero exit code is a result, not a pipeline
2051 failure — LLM steps consume it (TDD-loop building block).
2052 """
2053 step_id = step.get("id", "?")
2054 command = step.get("command")
2055 if not isinstance(command, str) or not command.strip():
2056 raise ValueError(
2057 f"command step '{step_id}': field 'command' is required and "
2058 "must be a non-empty string (ADR-0012)"
2059 )
2060 env_name = step.get("env")
2061 if env_name is not None:
2062 if not isinstance(env_name, str) or not env_name.strip():
2063 raise ValueError(
2064 f"command step '{step_id}': field 'env' must be a non-empty string (ADR-0012)"
2065 )
2066 if env_name not in self._envs:
2067 available = ", ".join(sorted(self._envs)) or "(none)"
2068 raise ValueError(
2069 f"command step '{step_id}': unknown env '{env_name}' "
2070 f"(available envs: {available}) (ADR-0012)"
2071 )
2072 result = self._command_executor.run_in_env(command, env_name)
2073 step_result: dict = {**result, "env": env_name, "executed_in": "env"}
2074 else:
2075 result = self._command_executor.run_in_engine(command)
2076 step_result = {**result, "env": None, "executed_in": "engine"}
2077 return {**state, step_id: step_result}
2080# ---------------------------------------------------------------------------
2081# GraphBuilder
2082# ---------------------------------------------------------------------------
2084def _topological_order(steps: list[dict]) -> list[dict]:
2085 """
2086 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
2087 """
2088 id_to_step = {s["id"]: s for s in steps}
2089 visited: set[str] = set()
2090 order: list[dict] = []
2092 def visit(step_id: str) -> None:
2093 if step_id in visited:
2094 return
2095 visited.add(step_id)
2096 for dep in (id_to_step[step_id].get("depends_on") or []):
2097 if dep in id_to_step:
2098 visit(dep)
2099 order.append(id_to_step[step_id])
2101 for s in steps:
2102 visit(s["id"])
2103 return order
2106# ---------------------------------------------------------------------------
2107# Call-step build-time expansion (ADR-0025 D3/D4)
2108# ---------------------------------------------------------------------------
2109#
2110# A `type: call` step is expanded at build time into the function's inlined body
2111# steps plus two synthesized op steps (`__call_bind` / `__call_emit`). The
2112# runtime executes only this flat graph; LangGraph never sees a call step. The
2113# caller reads the function outputs via the call result shape
2114# `state[<call_id>] = {outputs: {<name>: value}}` (00-process-contract r.70).
2115# The function body's `{{ params.<name> }}` references are rewritten to
2116# `{{ <call_id>.params.<name> }}` so the bind step's params namespace resolves.
2117#
2118# Isolation/purity (ADR-0025 D5) and recursion (D7/D9): `command` in a body is a
2119# build error; a cycle in the function call graph (self or mutual) is a build
2120# error. Bounded while loops are runtime constructs and are not cycle-flagged.
2122# Call step result namespace keys (internal scaffolding).
2123def _scope_params_value(value: Any, call_id: str) -> Any:
2124 """Rewrite `params.` template references to the call-scoped namespace.
2126 Operates on the function body's string values (templates). Dict keys are
2127 untouched — only templated string VALUES are scoped, so a function body's
2128 `{{ params.<name> }}` becomes `{{ <call_id>.params.<name> }}`.
2129 """
2130 if isinstance(value, str):
2131 return re.sub(r"\bparams\.", f"{call_id}.params.", value)
2132 if isinstance(value, dict):
2133 return {k: _scope_params_value(v, call_id) for k, v in value.items()}
2134 if isinstance(value, list):
2135 return [_scope_params_value(v, call_id) for v in value]
2136 return value
2139def _scope_params(step: dict, call_id: str) -> dict:
2140 """Deep-scope a function body step's params references for a call id."""
2141 scoped: dict = dict(step)
2142 for field in ("input", "params", "condition", "exit_condition", "agent", "command", "function"):
2143 if field in scoped and isinstance(scoped[field], (str, dict, list)):
2144 scoped[field] = _scope_params_value(scoped[field], call_id)
2145 if "body" in scoped and isinstance(scoped["body"], list):
2146 scoped["body"] = [_scope_params(b, call_id) for b in scoped["body"]]
2147 return scoped
2150def _validate_function_purity(fn_name: str, body: list[dict]) -> None:
2151 """function-body purity (ADR-0025 D5): `command` is a hard build error."""
2152 for step in body:
2153 if step.get("type") == "command":
2154 raise ValueError(
2155 f"function '{fn_name}' step '{step['id']}': command is forbidden "
2156 "in a function body (ADR-0025 D5 — purity; no side-effect execution)"
2157 )
2158 if "body" in step and isinstance(step["body"], list):
2159 _validate_function_purity(fn_name, step["body"])
2162def _expand_call(
2163 step: dict,
2164 registry: ManifestRegistry,
2165 stack: list[str],
2166 call_map: dict[str, str],
2167 expanded_functions: dict[str, str],
2168) -> list[dict]:
2169 """Expand a call step into bind + inlined body + emit steps."""
2170 fn_name = step["function"]
2171 call_id = step["id"]
2173 if fn_name in stack:
2174 chain = " -> ".join(stack + [fn_name])
2175 raise ValueError(
2176 f"cycle detected in function expansion: {chain} "
2177 "(ADR-0025 D7/D9 — recursion must be a bounded while loop or idempotent re-run)"
2178 )
2180 # ADR-0025 D7: inlining the same function twice in one pipeline build
2181 # collides its body step-ids (flat graph). This is a hard build error until
2182 # id-prefixing lands; naming the function and both call sites pins the bug.
2183 if fn_name in expanded_functions:
2184 raise ValueError(
2185 f"duplicate call expansion: function '{fn_name}' called at both "
2186 f"'{expanded_functions[fn_name]}' and '{call_id}' in one pipeline — "
2187 "inlined body step ids collide (ADR-0025 D7)"
2188 )
2189 expanded_functions[fn_name] = call_id
2191 function = registry.get("Function", fn_name)
2192 if function is None:
2193 raise ValueError(
2194 f"call step '{call_id}' references unknown function '{fn_name}' "
2195 "(no kind: Function manifest in the workflows tree)"
2196 )
2198 body = (function.get("spec") or {}).get("steps") or []
2199 _validate_function_purity(fn_name, body)
2201 sub_stack = list(stack) + [fn_name]
2202 outputs_spec = ((function.get("spec") or {}).get("interface") or {}).get("outputs") or []
2204 bind_id = f"_call__{call_id}__bind"
2205 emit_id = f"_call__{call_id}__emit"
2206 call_map[call_id] = emit_id
2208 bind_step: dict = {
2209 "id": bind_id,
2210 "type": "builtin",
2211 "component": "__call_bind",
2212 "params": {**(step.get("params") or {}), "__call_id": call_id},
2213 "depends_on": list(step.get("depends_on") or []),
2214 }
2216 expanded_body: list[dict] = []
2217 for body_step in body:
2218 scoped = _scope_params(body_step, call_id)
2219 for sub in _expand_step(scoped, registry, sub_stack, call_map, expanded_functions):
2220 dep = list(sub.get("depends_on") or [])
2221 if bind_id not in dep:
2222 sub["depends_on"] = [bind_id, *dep]
2223 expanded_body.append(sub)
2225 emit_step: dict = {
2226 "id": emit_id,
2227 "type": "builtin",
2228 "component": "__call_emit",
2229 "params": {"__call_id": call_id, "__outputs": outputs_spec},
2230 "depends_on": [bind_id] + [s["id"] for s in expanded_body],
2231 }
2233 return [bind_step, *expanded_body, emit_step]
2236def _expand_step(
2237 step: dict,
2238 registry: ManifestRegistry,
2239 stack: list[str],
2240 call_map: dict[str, str],
2241 expanded_functions: dict[str, str],
2242) -> list[dict]:
2243 """Expand one step (recursing into while/foreach bodies)."""
2244 if step.get("type") == "call":
2245 return _expand_call(step, registry, stack, call_map, expanded_functions)
2246 if "body" in step and isinstance(step["body"], list):
2247 expanded_body: list[dict] = []
2248 for sub in step["body"]:
2249 expanded_body.extend(
2250 _expand_step(sub, registry, stack, call_map, expanded_functions)
2251 )
2252 return [{**step, "body": expanded_body}]
2253 return [step]
2256def _rewrite_call_deps(steps: list[dict], call_map: dict[str, str]) -> None:
2257 """Rewrite `depends_on` references to call ids → the corresponding emit id.
2259 After a call is expanded away, a downstream step that depends_on the call id
2260 must instead wait on the emit step (which writes state[<call_id>]). This
2261 guarantees the caller's `{{ <call_id>.outputs.<name> }}` reads resolve after
2262 the function body ran.
2263 """
2264 for step in steps:
2265 deps = step.get("depends_on") or []
2266 if deps:
2267 step["depends_on"] = [call_map.get(d, d) for d in deps]
2268 if "body" in step and isinstance(step["body"], list):
2269 _rewrite_call_deps(step["body"], call_map)
2272def expand_calls(steps: list[dict], registry: ManifestRegistry) -> list[dict]:
2273 """Expand all `call` steps into the flat runtime graph (ADR-0025 D4).
2275 Returns a new step list with each call replaced by its bind + inlined body +
2276 emit steps; a call step never survives to the runtime graph. Cycles and
2277 command-in-body are hard errors.
2278 """
2279 call_map: dict[str, str] = {}
2280 expanded_functions: dict[str, str] = {}
2281 expanded: list[dict] = []
2282 for step in steps:
2283 expanded.extend(
2284 _expand_step(step, registry, [], call_map, expanded_functions)
2285 )
2286 _rewrite_call_deps(expanded, call_map)
2287 return expanded
2290class GraphBuilder:
2291 """
2292 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
2294 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
2295 käytetään vain {{ }} -templatejen evaluointiin.
2296 """
2298 def __init__(
2299 self,
2300 registry: ManifestRegistry,
2301 builtins: BuiltinExecutor | None = None,
2302 invoke_agent_fn: Any | None = None,
2303 run_context: dict | None = None,
2304 command_executor: CommandExecutor | None = None,
2305 tracer: Any | None = None,
2306 ) -> None:
2307 self._registry = registry
2308 self._resolver = AgentResolver(registry)
2309 if invoke_agent_fn is not None:
2310 (builtins or _DEFAULT_BUILTINS).invoke_agent_fn = invoke_agent_fn
2311 self._step_runner = StepRunner(
2312 self._resolver, builtins, invoke_agent_fn, run_context,
2313 command_executor=command_executor,
2314 tracer=tracer,
2315 )
2317 def build(self, pipeline: dict): # -> CompiledGraph
2318 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
2319 try:
2320 from langgraph.graph import StateGraph, END
2321 except ImportError as exc:
2322 raise ImportError(
2323 "langgraph is required to build a workflow. "
2324 "Install it with: pip install langgraph"
2325 ) from exc
2327 steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
2328 # Build-time call-step expansion (ADR-0025 D4): the runtime executes only
2329 # the flat expanded graph. LangGraph never sees a call step.
2330 steps = expand_calls(steps, self._registry)
2331 ordered = _topological_order(steps)
2333 # Bind the pipeline's env map to command steps (env validation at
2334 # step-execution time, spec.envs must be a map — ADR-0012).
2335 envs = (pipeline.get("spec") or {}).get("envs")
2336 if envs is not None and not isinstance(envs, dict):
2337 raise ValueError(
2338 f"Pipeline '{((pipeline.get('metadata') or {}).get('name') or '?')}': "
2339 "spec.envs must be a map of env name → image ref (ADR-0012)"
2340 )
2341 self._step_runner.set_envs(envs or {})
2343 # Group steps into LangGraph nodes:
2344 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
2345 nodes: list[list[dict]] = []
2346 for step in ordered:
2347 dep_sig = frozenset(step.get("depends_on") or [])
2348 if nodes:
2349 last_group_sig = frozenset(
2350 dep
2351 for s in nodes[-1]
2352 for dep in (s.get("depends_on") or [])
2353 )
2354 last_types = {s.get("type") for s in nodes[-1]}
2355 if (
2356 dep_sig == last_group_sig
2357 and step.get("type") == "parallel"
2358 and last_types == {"parallel"}
2359 ):
2360 nodes[-1].append(step)
2361 continue
2362 nodes.append([step])
2364 graph: StateGraph = StateGraph(PipelineState)
2365 step_runner = self._step_runner
2367 prev_node_id: str | None = None
2368 for group in nodes:
2369 if len(group) == 1:
2370 node_id = group[0]["id"]
2371 else:
2372 node_id = "parallel_" + "_".join(s["id"] for s in group)
2374 captured_group = list(group)
2376 if len(captured_group) == 1:
2377 def make_node(step: dict):
2378 def node_fn(state: PipelineState) -> PipelineState:
2379 return step_runner.run_step(step, state)
2380 return node_fn
2381 fn = make_node(captured_group[0])
2382 else:
2383 def make_parallel_node(group_steps: list[dict]):
2384 def node_fn(state: PipelineState) -> PipelineState:
2385 results: dict[str, Any] = {}
2386 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
2387 futures = {
2388 pool.submit(step_runner.run_step, s, state): s["id"]
2389 for s in group_steps
2390 }
2391 for future in as_completed(futures):
2392 sid = futures[future]
2393 try:
2394 out = future.result()
2395 results[sid] = out.get(sid)
2396 except SkillLoadError:
2397 raise
2398 except RunHardFailure:
2399 pool.shutdown(wait=False, cancel_futures=True)
2400 raise
2401 except Exception as exc: # noqa: BLE001
2402 log.error("Step '%s' failed: %s", sid, exc)
2403 results[sid] = {"error": str(exc)}
2404 return {**state, **results}
2405 return node_fn
2406 fn = make_parallel_node(captured_group)
2408 graph.add_node(node_id, fn)
2410 if prev_node_id is None:
2411 graph.set_entry_point(node_id)
2412 else:
2413 graph.add_edge(prev_node_id, node_id)
2415 prev_node_id = node_id
2417 if prev_node_id:
2418 graph.add_edge(prev_node_id, END)
2419 else:
2420 graph.add_node("noop", lambda s: s)
2421 graph.set_entry_point("noop")
2422 graph.add_edge("noop", END)
2424 return graph.compile()
2427# ---------------------------------------------------------------------------
2428# Julkinen API
2429# ---------------------------------------------------------------------------
2431_registry_cache: ManifestRegistry | None = None
2434def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
2435 global _registry_cache
2436 if _registry_cache is None or dirs is not None:
2437 loader = ManifestLoader(dirs)
2438 manifests = loader.load_all()
2439 _registry_cache = ManifestRegistry(manifests)
2440 return _registry_cache
2443def build_workflow(
2444 pipeline_name: str,
2445 *,
2446 dirs: list[Path] | None = None,
2447 builtins: BuiltinExecutor | None = None,
2448 invoke_agent_fn: Any | None = None,
2449 run_context: dict | None = None,
2450 tracer: Any | None = None,
2451):
2452 """
2453 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
2454 LangGraph-workflown.
2456 Args:
2457 pipeline_name: Pipeline metadata.name -arvo.
2458 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
2459 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
2460 invoke_agent_fn: fn(resolved_spec, step, state) → state.
2461 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
2463 Returns:
2464 Compilattu LangGraph CompiledGraph.
2465 """
2466 registry = _get_registry(dirs)
2467 pipeline = registry.get("Pipeline", pipeline_name)
2468 if pipeline is None:
2469 available = [
2470 (m.get("metadata") or {}).get("name")
2471 for m in registry.all_of_kind("Pipeline")
2472 ]
2473 raise KeyError(
2474 f"Pipeline '{pipeline_name}' not found. Available: {available}"
2475 )
2476 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context, tracer=tracer)
2477 return builder.build(pipeline)