Coverage for core/langgraph/runner.py: 78%
611 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-08 05:24 +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("doc-quality")
14 result = workflow.invoke({"input": "..."})
15"""
16from __future__ import annotations
18import logging
19import os
20import re
21import threading
22from concurrent.futures import ThreadPoolExecutor, as_completed
23from pathlib import Path
24from typing import Any
26import yaml
28# SkillLoadError-import (ei sirkulaari-riskiä — _llm importtaa vain stdlibin):
29# testit ja discover-wrapperit lataavat _llm:n tasaisena moduulina (sys.path
30# insert core/langgraph/), main.py qualified (langgraph._llm, PYTHONPATH=/app).
31# Kummatkin pinnat käännetään samaan _llm-tiedostoon; try/except varmistaa,
32# että luokka on sama objekti kuin nostavalla puolella molemmissa konteksteissa.
33try:
34 from _llm import SkillLoadError # noqa: F401
35except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
36 from langgraph._llm import SkillLoadError # noqa: F401
38log = logging.getLogger(__name__)
40# ---------------------------------------------------------------------------
41# Polut
42# ---------------------------------------------------------------------------
44PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", "/project"))
45MANIFEST_DIRS = [
46 PROJECT_ROOT / ".agent-platform" / "workflows",
47 PROJECT_ROOT / ".agent-platform" / "agents",
48]
49SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "manifest.schema.json"
52def resolve_manifest_dirs(clone_root: Path) -> list[Path]:
53 """Palauttaa olemassa olevat manifestihakemistot kloonatusta manifest-reposta.
55 Skannaa ainoastaan `.agent-platform/workflows` ja `.agent-platform/agents` —
56 ei koskaan koko kloonia. [] jos kumpaakaan ei ole (kutsuja → fail-fast).
57 """
58 return [
59 d for d in (
60 clone_root / ".agent-platform" / "workflows",
61 clone_root / ".agent-platform" / "agents",
62 )
63 if d.is_dir()
64 ]
66# ---------------------------------------------------------------------------
67# Template-resoluutio
68# ---------------------------------------------------------------------------
70_TMPL_RE = re.compile(r'\{\{\s*(.*?)\s*\}\}')
73def _dot_get(obj: Any, path: str) -> Any:
74 """Navigate nested dicts via dot-path: 'a.b.c' → obj['a']['b']['c']."""
75 for part in path.split('.'):
76 if isinstance(obj, dict):
77 obj = obj.get(part)
78 else:
79 return None
80 return obj
83def _eval_expr(expr: str, ctx: dict) -> Any:
84 """Evaluate a simple template expression against context dict."""
85 expr = expr.strip()
86 # Negation: not X
87 if expr.startswith('not '):
88 return not _eval_expr(expr[4:].strip(), ctx)
89 # Comparisons
90 for op in ('<=', '>=', '==', '!=', '<', '>'):
91 if op in expr:
92 left_s, right_s = expr.split(op, 1)
93 left = _eval_expr(left_s.strip(), ctx)
94 right_raw = right_s.strip()
95 # Quoted string literal → strip quotes and compare as string
96 # (applies to all operators; numeric cast preserved for unquoted values)
97 if len(right_raw) >= 2 and right_raw[0] == right_raw[-1] and right_raw[0] in ("'", '"'):
98 right: Any = right_raw[1:-1]
99 else:
100 try:
101 if isinstance(left, (int, float)):
102 right = type(left)(right_raw)
103 elif right_raw.lstrip('-').isdigit():
104 right = int(right_raw)
105 else:
106 right = right_raw
107 except (ValueError, TypeError):
108 right = right_raw
109 try:
110 if op == '<=': return left <= right
111 if op == '>=': return left >= right
112 if op == '==': return left == right
113 if op == '!=': return left != right
114 if op == '<': return left < right
115 if op == '>': return left > right
116 except TypeError:
117 return False
118 # Dot-path lookup
119 return _dot_get(ctx, expr)
122def _resolve_template(value: Any, ctx: dict) -> Any:
123 """Recursively resolve {{ expr }} templates in strings / dicts / lists."""
124 if isinstance(value, str):
125 stripped = value.strip()
126 # Entire string is one template → replace with actual type (not string)
127 m = _TMPL_RE.fullmatch(stripped)
128 if m:
129 return _eval_expr(m.group(1), ctx)
130 # Inline substitution → always returns string
131 def _replace(m: re.Match) -> str:
132 result = _eval_expr(m.group(1), ctx)
133 return '' if result is None else str(result)
134 return _TMPL_RE.sub(_replace, value)
135 if isinstance(value, dict):
136 return {k: _resolve_template(v, ctx) for k, v in value.items()}
137 if isinstance(value, list):
138 return [_resolve_template(v, ctx) for v in value]
139 return value
142def _resolve_input(input_spec: Any, state: dict, ctx: dict) -> Any:
143 """
144 Resolve step input spec to a concrete value:
145 "{{ item }}" → template
146 "gather.doc_files" → dot-path into state
147 {"k": "a.b"} → dict with each value resolved
148 None → None
149 """
150 if input_spec is None:
151 return None
152 if isinstance(input_spec, str):
153 if '{{' in input_spec:
154 return _resolve_template(input_spec, ctx)
155 # Dot-path into state (no braces)
156 return _dot_get(state, input_spec)
157 if isinstance(input_spec, dict):
158 return {k: _resolve_input(v, state, ctx) for k, v in input_spec.items()}
159 if isinstance(input_spec, list):
160 return [_resolve_input(v, state, ctx) for v in input_spec]
161 return input_spec
164def _eval_condition(cond: Any, ctx: dict) -> bool:
165 """Evaluate a step condition. None → True (always run)."""
166 if cond is None:
167 return True
168 resolved = _resolve_template(cond, ctx)
169 if isinstance(resolved, bool):
170 return resolved
171 if isinstance(resolved, str):
172 return resolved.lower() not in ('false', '0', '', 'none', 'null')
173 return bool(resolved)
176def _resolve_int(value: Any, ctx: dict, default: int) -> int:
177 v = _resolve_template(value, ctx) if isinstance(value, str) else value
178 try:
179 return int(v)
180 except (TypeError, ValueError):
181 return default
184# ---------------------------------------------------------------------------
185# ManifestLoader
186# ---------------------------------------------------------------------------
189def _parse_manifest_file(path: Path) -> list[dict]:
190 """Palauttaa kaikki apiVersion: agent-platform/v1 -dokumentit tiedostosta."""
191 try:
192 content = path.read_text(encoding="utf-8")
193 except OSError as exc:
194 log.warning("Cannot read %s: %s", path, exc)
195 return []
197 results: list[dict] = []
198 for raw in content.split("\n---"):
199 raw_clean = "\n".join(
200 line for line in raw.splitlines()
201 if not line.strip().startswith("#")
202 )
203 try:
204 doc = yaml.safe_load(raw_clean)
205 except yaml.YAMLError as exc:
206 log.warning("YAML parse error in %s: %s", path, exc)
207 continue
208 if isinstance(doc, dict) and doc.get("apiVersion") == "agent-platform/v1":
209 results.append(doc)
210 return results
213class ManifestLoader:
214 """Skannaa hakemistot ja lataa kaikki YAML-manifestit."""
216 def __init__(self, dirs: list[Path] | None = None) -> None:
217 self._dirs = dirs or MANIFEST_DIRS
219 def load_all(self) -> list[dict]:
220 manifests: list[dict] = []
221 for d in self._dirs:
222 if not d.exists():
223 log.debug("Manifest dir not found: %s", d)
224 continue
225 for path in sorted(d.glob("**/*.yaml")):
226 docs = _parse_manifest_file(path)
227 manifests.extend(docs)
228 if docs:
229 log.debug("Loaded %d manifest(s) from %s", len(docs), path)
230 return manifests
233# ---------------------------------------------------------------------------
234# ManifestRegistry
235# ---------------------------------------------------------------------------
237ResourceKey = tuple[str, str] # (kind, name)
240class ManifestRegistry:
241 """Indeksoi ladatut manifestit kind+name avaimella."""
243 def __init__(self, manifests: list[dict]) -> None:
244 self._store: dict[ResourceKey, dict] = {}
245 for doc in manifests:
246 kind = doc.get("kind", "")
247 name = (doc.get("metadata") or {}).get("name", "")
248 if kind and name:
249 key: ResourceKey = (kind, name)
250 if key in self._store:
251 log.warning("Duplicate manifest %s/%s — last one wins", kind, name)
252 self._store[key] = doc
254 def get(self, kind: str, name: str) -> dict | None:
255 return self._store.get((kind, name))
257 def all_of_kind(self, kind: str) -> list[dict]:
258 return [v for (k, _), v in self._store.items() if k == kind]
261# ---------------------------------------------------------------------------
262# AgentResolver
263# ---------------------------------------------------------------------------
265def _deep_merge(base: dict, override: dict) -> dict:
266 """Syvä merge: override voittaa konflikteissa."""
267 result = dict(base)
268 for k, v in override.items():
269 if isinstance(v, dict) and isinstance(result.get(k), dict):
270 result[k] = _deep_merge(result[k], v)
271 else:
272 result[k] = v
273 return result
276class AgentResolver:
277 """
278 Palauttaa fully-resolved agent spec:n instanssille.
279 Jos instanssilla on based_on, mergee Agent-base + AgentInstance-overridet.
280 """
282 def __init__(self, registry: ManifestRegistry) -> None:
283 self._registry = registry
284 self._cache: dict[str, dict] = {}
286 def resolve(self, instance_name: str) -> dict:
287 if instance_name in self._cache:
288 return self._cache[instance_name]
290 instance = self._registry.get("AgentInstance", instance_name)
291 if instance is None:
292 agent = self._registry.get("Agent", instance_name)
293 if agent is None:
294 raise KeyError(f"No Agent or AgentInstance named '{instance_name}'")
295 resolved = agent
296 else:
297 based_on = (instance.get("spec") or {}).get("based_on")
298 if based_on:
299 base_agent = self._registry.get("Agent", based_on)
300 if base_agent is None:
301 raise KeyError(
302 f"AgentInstance '{instance_name}' references unknown Agent '{based_on}'"
303 )
304 base_spec = base_agent.get("spec") or {}
305 inst_spec = {k: v for k, v in (instance.get("spec") or {}).items()
306 if k != "based_on"}
307 merged_spec = _deep_merge(base_spec, inst_spec)
308 else:
309 merged_spec = {k: v for k, v in (instance.get("spec") or {}).items()
310 if k != "based_on"}
312 resolved = {
313 "apiVersion": "agent-platform/v1",
314 "kind": "Agent",
315 "metadata": {"name": instance_name},
316 "spec": merged_spec,
317 }
319 self._cache[instance_name] = resolved
320 return resolved
323# ---------------------------------------------------------------------------
324# State type
325# ---------------------------------------------------------------------------
327PipelineState = dict[str, Any]
329# Sentinel for optional kwargs distinguishing "not provided" from None
330_UNSET: Any = object()
332# ---------------------------------------------------------------------------
333# BuiltinExecutor
334# ---------------------------------------------------------------------------
336class BuiltinExecutor:
337 """
338 Ajaa builtin-komponentit.
339 Uusia komponentteja voi rekisteröidä register()-metodilla.
340 Nimet normalisoidaan: 'file-collector' == 'file_collector'.
341 """
343 _LAZY_MODULES: tuple[str, ...] = ("_builtins_pr",)
345 def __init__(self) -> None:
346 self._handlers: dict[str, Any] = {
347 "passthrough": self._passthrough,
348 "file_collector": self._file_collector,
349 "file-collector": self._file_collector,
350 }
352 def register(self, name: str, fn: Any) -> None:
353 self._handlers[name] = fn
354 # Also register with normalised name (hyphen → underscore)
355 normalised = name.replace("-", "_")
356 if normalised != name:
357 self._handlers[normalised] = fn
359 def get(self, name: str) -> Any:
360 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
361 if fn is not None:
362 return fn
363 import importlib
364 for module_name in self._LAZY_MODULES:
365 try:
366 mod = importlib.import_module(f".{module_name}", package="langgraph")
367 except ImportError:
368 continue
369 mod.register_all(self)
370 fn = self._handlers.get(name) or self._handlers.get(name.replace("-", "_"))
371 if fn is None:
372 raise ValueError(f"Unknown builtin component '{name}'")
373 return fn
375 def execute(self, component: str, step: dict, state: PipelineState) -> PipelineState:
376 fn = self.get(component)
377 return fn(step, state)
379 # --- default builtins ---
381 @staticmethod
382 def _passthrough(step: dict, state: PipelineState) -> PipelineState:
383 return state
385 @staticmethod
386 def _file_collector(step: dict, state: PipelineState) -> PipelineState:
387 """
388 Kerää tiedostot project_root/:sta.
389 Parametrit (step.params):
390 - path: glob-pattern (oletus: "**/*.md")
391 - output_key: state-avain johon lista tallennetaan (oletus: "files")
392 """
393 params = step.get("params") or {}
394 pattern = params.get("path", "**/*.md")
395 output_key = params.get("output_key", "files")
396 root = Path(state.get("project_root", str(PROJECT_ROOT)))
397 files = [str(p) for p in sorted(root.glob(pattern))]
398 return {**state, output_key: files}
401_DEFAULT_BUILTINS = BuiltinExecutor()
404def _register_domain_builtins(executor: BuiltinExecutor) -> None:
405 """Rekisteröi v1-domain-builtinit (file-collector, path-verifier, ...).
407 Erotettu omaan moduliin (`_builtins.py`) — runner.py pysyy generic-engine,
408 builtinit voivat olla doc-quality-spesifisiäkin v1:ssä.
409 """
410 try:
411 from langgraph import _builtins # type: ignore
412 except ImportError:
413 # Kun runner.py ajetaan ilman pakettirakennetta (esim. /app/langgraph)
414 import importlib.util
415 spec = importlib.util.spec_from_file_location(
416 "_builtins", Path(__file__).parent / "_builtins.py"
417 )
418 if spec and spec.loader:
419 _builtins = importlib.util.module_from_spec(spec)
420 spec.loader.exec_module(_builtins) # type: ignore[union-attr]
421 else:
422 log.warning("Cannot load _builtins.py — domain builtins disabled")
423 return
424 _builtins.register_all(executor)
427_register_domain_builtins(_DEFAULT_BUILTINS)
429# ---------------------------------------------------------------------------
430# StepRunner
431# ---------------------------------------------------------------------------
433class StepRunner:
434 """
435 Ajaa yksittäisen step:n sen typin mukaan.
437 run_context: dict joka lisätään template-kontekstiin. Esim:
438 {"budget": {"max_parallel": 16, ...}, "config": {...}}
439 """
441 def __init__(
442 self,
443 resolver: AgentResolver,
444 builtins: BuiltinExecutor | None = None,
445 invoke_agent_fn: Any | None = None,
446 run_context: dict | None = None,
447 ) -> None:
448 self._resolver = resolver
449 self._builtins = builtins or _DEFAULT_BUILTINS
450 self._invoke_agent = invoke_agent_fn or self._dummy_invoke
451 self._run_context: dict = run_context or {}
453 def _make_ctx(self, state: dict, **extras: Any) -> dict:
454 """Rakentaa template-kontekstin: state + run_context + extras."""
455 ctx: dict = {}
456 ctx.update(state)
457 ctx.update(self._run_context)
458 ctx.update(extras)
459 return ctx
461 def _resolve_step_runtime(
462 self,
463 step: dict,
464 state: dict,
465 ctx: dict,
466 *,
467 input_override: Any = _UNSET,
468 extra_params: dict | None = None,
469 ) -> dict:
470 """
471 Palauta kopio step:stä jossa `input` ja `params` on resolvoitu
472 templaattien osalta state + run_context -kontekstia vasten.
474 OpenAIInvoker olettaa että step.input ja step.params on JO resolvoitu
475 (vrt. _llm.py:n __call__ docstring) — tämä helper hoitaa sen.
476 """
477 resolved: dict = dict(step)
479 if input_override is not _UNSET:
480 resolved["input"] = input_override
481 else:
482 input_spec = step.get("input")
483 if input_spec is not None:
484 resolved["input"] = _resolve_input(input_spec, state, ctx)
486 params_spec = step.get("params")
487 if params_spec or extra_params:
488 base = _resolve_template(params_spec, ctx) if params_spec else {}
489 if not isinstance(base, dict):
490 base = {}
491 if extra_params:
492 base = {**base, **extra_params}
493 resolved["params"] = base
495 return resolved
497 @staticmethod
498 def _dummy_invoke(
499 resolved_spec: dict, step: dict, state: PipelineState
500 ) -> PipelineState:
501 """Placeholder — ei kutsu oikeaa LLM:ää."""
502 step_id = step.get("id", "?")
503 log.debug("[dummy] Step '%s' agent call skipped (no invoke_agent_fn set)", step_id)
504 return {**state, step_id: {"status": "skipped", "agent": resolved_spec["metadata"]["name"]}}
506 @staticmethod
507 def _merge_errors(base: list, additions: list) -> list:
508 """Return a new list containing all items from *base* plus any items from
509 *additions* that are not already present. Order is preserved and no
510 in-place mutation occurs — safe to call from any context."""
511 if not additions:
512 return base
513 merged = list(base)
514 for err in additions:
515 if err not in merged:
516 merged.append(err)
517 return merged
519 @staticmethod
520 def _propagate_step_errors(result: PipelineState, step_id: str) -> PipelineState:
521 """
522 Jos _invoke_agent palautti virheen (mikä tahansa ei-tyhjä 'error'-kenttä),
523 lisätään se state['errors']-listaan jotta run_finalizer raportoi ajon
524 epäonnistuneeksi eikä hiljennä virhettä.
526 Kaikki virhetyypit propagoidaan:
527 - budget:max_llm_calls / budget:run_timeout → budjettiloki
528 - parse_error:*, timeout:*, ja muut → error-loki
529 """
530 step_result = result.get(step_id)
531 if not isinstance(step_result, dict):
532 return result
533 err = step_result.get("error", "")
534 if not isinstance(err, str) or not err:
535 return result
536 errors = list(result.get("errors") or [])
537 message = f"{step_id}: {err}"
538 if message not in errors:
539 errors.append(message)
540 if err.startswith("budget:"):
541 log.warning("[budget] %s", message)
542 else:
543 log.error("[step-error] %s", message)
544 return {**result, "errors": errors}
546 def run_step(self, step: dict, state: PipelineState) -> PipelineState:
547 step_id = step.get("id", "?")
548 ctx = self._make_ctx(state)
550 # Evaluate condition — if false, skip step
551 cond = step.get("condition")
552 if cond is not None and not _eval_condition(cond, ctx):
553 log.debug("[step] %s skipped (condition false)", step_id)
554 return state
556 step_type = step.get("type", "sequential")
558 if step_type == "builtin":
559 component = step.get("component", "passthrough")
560 log.debug("[step] %s builtin:%s", step_id, component)
561 # Builtins lukevat raw step.params suoraan; resolvoidaan kuitenkin
562 # template-arvot (esim. "{{ budget.min_severity_for_ticket }}")
563 resolved_step = self._resolve_step_runtime(step, state, ctx)
564 return self._builtins.execute(component, resolved_step, state)
566 if step_type == "parallel":
567 # Fan-out: if input resolves to a list, call agent once per item
568 input_spec = step.get("input")
569 if input_spec is not None:
570 input_val = _resolve_input(input_spec, state, ctx)
571 if isinstance(input_val, list):
572 return self._run_parallel_fanout(step, state, input_val, ctx)
573 # Fallback: single agent call
574 agent_name = step.get("agent")
575 if not agent_name:
576 log.warning("[step] %s has no agent — skipping", step_id)
577 return state
578 resolved = self._resolver.resolve(agent_name)
579 log.debug("[step] %s parallel (single) → agent:%s", step_id, agent_name)
580 resolved_step = self._resolve_step_runtime(step, state, ctx)
581 result = self._invoke_agent(resolved, resolved_step, state)
582 return self._propagate_step_errors(result, step_id)
584 if step_type == "sequential":
585 agent_name = step.get("agent")
586 if not agent_name:
587 log.warning("[step] %s has no agent — skipping", step_id)
588 return state
589 resolved = self._resolver.resolve(agent_name)
590 log.debug("[step] %s sequential → agent:%s", step_id, agent_name)
591 resolved_step = self._resolve_step_runtime(step, state, ctx)
592 result = self._invoke_agent(resolved, resolved_step, state)
593 return self._propagate_step_errors(result, step_id)
595 if step_type == "foreach":
596 return self._run_foreach(step, state)
598 if step_type == "while":
599 return self._run_while(step, state)
601 log.warning("[step] Unknown step type '%s' in step '%s'", step_type, step_id)
602 return state
604 def _run_parallel_fanout(
605 self,
606 step: dict,
607 state: PipelineState,
608 items: list,
609 ctx: dict,
610 ) -> PipelineState:
611 """
612 Fan-out: call agent once per item, concurrently.
613 Results stored as state[step_id] = {"results": [item_result, ...]}.
614 Each item is available as state["item"] inside the invoke call.
615 """
616 step_id = step.get("id", "?")
617 agent_name = step.get("agent")
618 if not agent_name:
619 log.warning("[step] %s parallel fanout: no agent", step_id)
620 return {**state, step_id: {"results": []}}
622 resolved = self._resolver.resolve(agent_name)
624 # Apply max_items cap
625 max_items_raw = step.get("max_items")
626 if max_items_raw is not None:
627 max_items = _resolve_int(max_items_raw, ctx, len(items))
628 items = items[:max_items]
630 # Concurrency limit
631 max_conc_raw = step.get("max_concurrent")
632 if max_conc_raw is not None:
633 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
634 else:
635 max_conc = len(items) or 1
636 max_conc = max(1, min(max_conc, len(items) or 1))
638 log.debug(
639 "[step] %s parallel fanout: %d items, %d concurrent",
640 step_id, len(items), max_conc,
641 )
643 # Pre-fetch path verification map if present in state (from path-verifier builtin)
644 verify_by_path: dict = {}
645 verify_step = state.get("verify_paths")
646 if isinstance(verify_step, dict):
647 vbp = verify_step.get("by_path")
648 if isinstance(vbp, dict):
649 verify_by_path = vbp
651 def call_one(item: Any) -> Any:
652 item_state = {**state, "_item": item, "item": item}
653 item_ctx = self._make_ctx(item_state)
654 # Per-item path verification block injection (matches doc_files[*].path)
655 extra: dict = {}
656 if isinstance(item, dict) and verify_by_path:
657 p = item.get("path")
658 if p and p in verify_by_path:
659 extra["path_verification_block"] = verify_by_path[p]
660 resolved_step = self._resolve_step_runtime(
661 step, item_state, item_ctx,
662 input_override=item,
663 extra_params=extra or None,
664 )
665 result_state = self._invoke_agent(resolved, resolved_step, item_state)
666 return result_state.get(step_id)
668 results: list = []
669 with ThreadPoolExecutor(max_workers=max_conc) as pool:
670 futures = [pool.submit(call_one, item) for item in items]
671 for future in as_completed(futures):
672 try:
673 results.append(future.result())
674 except SkillLoadError:
675 raise
676 except Exception as exc: # noqa: BLE001
677 log.error("[step] %s item failed: %s", step_id, exc)
678 results.append({"error": str(exc)})
680 return {**state, step_id: {"results": results}}
682 def _run_foreach(self, step: dict, state: PipelineState) -> PipelineState:
683 """
684 Foreach: iterate over input list, run body steps per item.
685 Input from step.input (dot-path or template) or step.over (state key).
687 When max_concurrent > 1, body iterations run concurrently via
688 ThreadPoolExecutor — each item gets its own worker thread. The
689 max_concurrent value sets the thread pool size (mirrors parallel
690 fan-out semantics). Results list preserves original item order.
692 max_output budget in concurrent mode: a stop flag prevents new workers
693 from starting after the budget is exhausted. Workers already executing
694 when the flag is set will complete naturally — at most max_concurrent
695 extra items may run past the limit. This is accepted behaviour; the
696 sequential early-exit guarantee cannot be reproduced without sacrificing
697 true concurrency.
698 """
699 step_id = step.get("id", "?")
700 ctx = self._make_ctx(state)
702 # Determine items
703 input_spec = step.get("input")
704 over_key = step.get("over")
706 if input_spec is not None:
707 items: list = _resolve_input(input_spec, state, ctx) or []
708 if not isinstance(items, list):
709 items = [items] if items is not None else []
710 elif over_key:
711 items = state.get(over_key, [None])
712 if not isinstance(items, list):
713 items = [items]
714 else:
715 items = [None]
717 # max_iterations cap
718 max_iter_raw = step.get("max_iterations")
719 if max_iter_raw is not None:
720 max_iter = _resolve_int(max_iter_raw, ctx, len(items))
721 items = items[:max_iter]
723 # max_output cap — stop once N iterations have produced a successful
724 # publish (iter_state["publish"]["path"]). 0/absent = unlimited.
725 max_output_raw = step.get("max_output")
726 max_output = _resolve_int(max_output_raw, ctx, 0) if max_output_raw is not None else 0
728 # max_concurrent — when > 1, enables parallel body execution
729 max_conc_raw = step.get("max_concurrent")
730 if max_conc_raw is not None:
731 max_conc = _resolve_int(max_conc_raw, ctx, len(items))
732 max_conc = max(1, min(max_conc, len(items) or 1))
733 else:
734 max_conc = 1
736 body = step.get("body") or []
737 agent_name = step.get("agent")
739 if max_conc <= 1:
740 return self._run_foreach_sequential(
741 step_id, step, state, items, body, agent_name, max_output,
742 )
743 return self._run_foreach_concurrent(
744 step_id, step, state, items, body, agent_name, max_output, max_conc,
745 )
747 def _run_foreach_sequential(
748 self,
749 step_id: str,
750 step: dict,
751 state: PipelineState,
752 items: list,
753 body: list,
754 agent_name: str | None,
755 max_output: int,
756 ) -> PipelineState:
757 """Sequential foreach — default path when max_concurrent is absent or 1."""
758 results: list = []
759 produced_outputs = 0
760 accumulated_errors: list = list(state.get("errors") or [])
762 for item in items:
763 if max_output and produced_outputs >= max_output:
764 log.info("[foreach] %s max_output=%d reached — stopping", step_id, max_output)
765 break
766 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
768 if body:
769 for sub_step in body:
770 iter_state = self.run_step(sub_step, iter_state)
771 # Propagate any budget errors from this iteration to accumulated list
772 accumulated_errors = self._merge_errors(
773 accumulated_errors, iter_state.get("errors") or []
774 )
775 publish_out = iter_state.get("publish")
776 results.append(publish_out or iter_state.get(step_id))
777 if isinstance(publish_out, dict) and publish_out.get("path"):
778 produced_outputs += 1
779 elif agent_name:
780 resolved = self._resolver.resolve(agent_name)
781 iter_state = self._invoke_agent(resolved, step, iter_state)
782 results.append(iter_state.get(step_id))
784 final: PipelineState = {**state, step_id: results}
785 if accumulated_errors != list(state.get("errors") or []):
786 final = {**final, "errors": accumulated_errors}
787 return final
789 def _run_foreach_concurrent(
790 self,
791 step_id: str,
792 step: dict,
793 state: PipelineState,
794 items: list,
795 body: list,
796 agent_name: str | None,
797 max_output: int,
798 max_conc: int,
799 ) -> PipelineState:
800 """Concurrent foreach — runs body iterations in parallel, throttled by max_conc."""
801 log.info(
802 "[foreach] %s concurrent: %d items, max_concurrent=%d",
803 step_id, len(items), max_conc,
804 )
806 # Pre-allocate results list to preserve item order across threads.
807 # _SKIPPED is a sentinel that marks slots skipped via the stop_flag;
808 # legitimate None results (body produced no output) are kept.
809 _SKIPPED = object()
810 results: list = [_SKIPPED] * len(items)
811 lock = threading.Lock()
812 produced_outputs = [0] # mutable counter shared across threads
813 stop_flag = threading.Event() # set when max_output budget is exhausted
814 item_errors: list = [] # budget errors collected from worker threads
816 def run_item(idx: int, item: Any) -> None:
817 nonlocal item_errors
818 # Honour stop flag — skip if budget already exhausted before we start
819 if max_output and stop_flag.is_set():
820 return
822 iter_state = {**state, "_item": item, "item": item, "_foreach_item": item}
824 if body:
825 for sub_step in body:
826 iter_state = self.run_step(sub_step, iter_state)
827 # Collect budget errors from this iteration (thread-safe merge)
828 new_errs = iter_state.get("errors") or []
829 if new_errs:
830 with lock:
831 item_errors = self._merge_errors(item_errors, new_errs)
832 publish_out = iter_state.get("publish")
833 results[idx] = publish_out or iter_state.get(step_id)
834 if isinstance(publish_out, dict) and publish_out.get("path"):
835 with lock:
836 produced_outputs[0] += 1
837 if max_output and produced_outputs[0] >= max_output:
838 stop_flag.set()
839 elif agent_name:
840 resolved = self._resolver.resolve(agent_name)
841 iter_state = self._invoke_agent(resolved, step, iter_state)
842 results[idx] = iter_state.get(step_id)
844 with ThreadPoolExecutor(max_workers=max_conc) as pool:
845 futures = {pool.submit(run_item, i, item): i for i, item in enumerate(items)}
846 for future in as_completed(futures):
847 try:
848 future.result()
849 except SkillLoadError:
850 raise
851 except Exception as exc: # noqa: BLE001
852 idx = futures[future]
853 log.error("[foreach] %s item[%d] failed: %s", step_id, idx, exc)
854 results[idx] = {"error": str(exc)}
856 # Strip only sentinel slots (items skipped via stop_flag); keep legitimate None results
857 final: PipelineState = {**state, step_id: [r for r in results if r is not _SKIPPED]}
858 if item_errors:
859 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), item_errors)}
860 return final
862 def _run_while(self, step: dict, state: PipelineState) -> PipelineState:
863 """
864 While loop with exit_condition and optional entry condition.
866 Fields:
867 condition: "{{ triage.approved }}" — if false, skip entirely
868 input: initial value for loop.current
869 exit_condition: "{{ review.pass }}" — break when true
870 max_iterations: int or template
871 body: list of sub-steps
872 """
873 step_id = step.get("id", "?")
874 ctx = self._make_ctx(state)
876 # Entry condition (already checked by run_step, but also here for nested calls)
877 entry_cond = step.get("condition")
878 if entry_cond is not None and not _eval_condition(entry_cond, ctx):
879 return {**state, step_id: {"passed": False, "iterations": 0, "output": None, "skipped": True}}
881 # Initial current value from input
882 input_spec = step.get("input")
883 if input_spec is not None:
884 current = _resolve_input(input_spec, state, ctx)
885 else:
886 current = None
888 exit_condition = step.get("exit_condition")
889 max_iter = _resolve_int(step.get("max_iterations", 2), ctx, 2)
890 body = step.get("body") or []
892 passed = False
893 iteration = 0
895 while iteration < max_iter:
896 # Inject loop context
897 loop_state = {
898 **state,
899 "loop": {"current": current, "iteration": iteration},
900 step_id: {"current": current, "passed": False},
901 }
902 # Add current item context (from outer foreach if any)
903 if "item" in state:
904 loop_state["item"] = state["item"]
906 # Run body steps
907 for sub_step in body:
908 loop_state = self.run_step(sub_step, loop_state)
910 iteration += 1
912 # Update current from refine output (if refine ran)
913 refine_out = loop_state.get("refine")
914 if refine_out is not None:
915 if isinstance(refine_out, str):
916 current = refine_out
917 elif isinstance(refine_out, dict) and "output" in refine_out:
918 current = refine_out["output"]
920 # Check exit condition
921 if exit_condition:
922 exit_ctx = self._make_ctx(loop_state)
923 if _eval_condition(exit_condition, exit_ctx):
924 passed = True
925 break
926 else:
927 passed = True
928 break
930 final: PipelineState = {**state, step_id: {
931 "passed": passed,
932 "iterations": iteration,
933 "output": current,
934 }}
935 # Propagate any errors accumulated in the loop body back to the outer state
936 loop_errors = loop_state.get("errors") or []
937 if loop_errors:
938 final = {**final, "errors": self._merge_errors(list(state.get("errors") or []), loop_errors)}
939 return final
942# ---------------------------------------------------------------------------
943# GraphBuilder
944# ---------------------------------------------------------------------------
946def _topological_order(steps: list[dict]) -> list[dict]:
947 """
948 Palauttaa stepit topologisessa järjestyksessä depends_on-kenttien perusteella.
949 """
950 id_to_step = {s["id"]: s for s in steps}
951 visited: set[str] = set()
952 order: list[dict] = []
954 def visit(step_id: str) -> None:
955 if step_id in visited:
956 return
957 visited.add(step_id)
958 for dep in (id_to_step[step_id].get("depends_on") or []):
959 if dep in id_to_step:
960 visit(dep)
961 order.append(id_to_step[step_id])
963 for s in steps:
964 visit(s["id"])
965 return order
968class GraphBuilder:
969 """
970 Rakentaa LangGraph StateGraphin Pipeline-manifestista.
972 run_context: template-kontekstidata (budget, config jne.) — ei tallennu stateen,
973 käytetään vain {{ }} -templatejen evaluointiin.
974 """
976 def __init__(
977 self,
978 registry: ManifestRegistry,
979 builtins: BuiltinExecutor | None = None,
980 invoke_agent_fn: Any | None = None,
981 run_context: dict | None = None,
982 ) -> None:
983 self._resolver = AgentResolver(registry)
984 self._step_runner = StepRunner(
985 self._resolver, builtins, invoke_agent_fn, run_context
986 )
988 def build(self, pipeline: dict): # -> CompiledGraph
989 """Rakentaa ja palauttaa compilatun LangGraph StateGraphin."""
990 try:
991 from langgraph.graph import StateGraph, END
992 except ImportError as exc:
993 raise ImportError(
994 "langgraph is required to build a workflow. "
995 "Install it with: pip install langgraph"
996 ) from exc
998 steps: list[dict] = (pipeline.get("spec") or {}).get("steps") or []
999 ordered = _topological_order(steps)
1001 # Group steps into LangGraph nodes:
1002 # Steps with the same depends_on set AND type==parallel → one node (LangGraph-level parallel)
1003 nodes: list[list[dict]] = []
1004 for step in ordered:
1005 dep_sig = frozenset(step.get("depends_on") or [])
1006 if nodes:
1007 last_group_sig = frozenset(
1008 dep
1009 for s in nodes[-1]
1010 for dep in (s.get("depends_on") or [])
1011 )
1012 last_types = {s.get("type") for s in nodes[-1]}
1013 if (
1014 dep_sig == last_group_sig
1015 and step.get("type") == "parallel"
1016 and last_types == {"parallel"}
1017 ):
1018 nodes[-1].append(step)
1019 continue
1020 nodes.append([step])
1022 graph: StateGraph = StateGraph(PipelineState)
1023 step_runner = self._step_runner
1025 prev_node_id: str | None = None
1026 for group in nodes:
1027 if len(group) == 1:
1028 node_id = group[0]["id"]
1029 else:
1030 node_id = "parallel_" + "_".join(s["id"] for s in group)
1032 captured_group = list(group)
1034 if len(captured_group) == 1:
1035 def make_node(step: dict):
1036 def node_fn(state: PipelineState) -> PipelineState:
1037 return step_runner.run_step(step, state)
1038 return node_fn
1039 fn = make_node(captured_group[0])
1040 else:
1041 def make_parallel_node(group_steps: list[dict]):
1042 def node_fn(state: PipelineState) -> PipelineState:
1043 results: dict[str, Any] = {}
1044 with ThreadPoolExecutor(max_workers=len(group_steps)) as pool:
1045 futures = {
1046 pool.submit(step_runner.run_step, s, state): s["id"]
1047 for s in group_steps
1048 }
1049 for future in as_completed(futures):
1050 sid = futures[future]
1051 try:
1052 out = future.result()
1053 results[sid] = out.get(sid)
1054 except SkillLoadError:
1055 raise
1056 except Exception as exc: # noqa: BLE001
1057 log.error("Step '%s' failed: %s", sid, exc)
1058 results[sid] = {"error": str(exc)}
1059 return {**state, **results}
1060 return node_fn
1061 fn = make_parallel_node(captured_group)
1063 graph.add_node(node_id, fn)
1065 if prev_node_id is None:
1066 graph.set_entry_point(node_id)
1067 else:
1068 graph.add_edge(prev_node_id, node_id)
1070 prev_node_id = node_id
1072 if prev_node_id:
1073 graph.add_edge(prev_node_id, END)
1074 else:
1075 graph.add_node("noop", lambda s: s)
1076 graph.set_entry_point("noop")
1077 graph.add_edge("noop", END)
1079 return graph.compile()
1082# ---------------------------------------------------------------------------
1083# Julkinen API
1084# ---------------------------------------------------------------------------
1086_registry_cache: ManifestRegistry | None = None
1089def _get_registry(dirs: list[Path] | None = None) -> ManifestRegistry:
1090 global _registry_cache
1091 if _registry_cache is None or dirs is not None:
1092 loader = ManifestLoader(dirs)
1093 manifests = loader.load_all()
1094 _registry_cache = ManifestRegistry(manifests)
1095 return _registry_cache
1098def build_workflow(
1099 pipeline_name: str,
1100 *,
1101 dirs: list[Path] | None = None,
1102 builtins: BuiltinExecutor | None = None,
1103 invoke_agent_fn: Any | None = None,
1104 run_context: dict | None = None,
1105):
1106 """
1107 Lataa Pipeline-manifesti nimeltä *pipeline_name* ja rakentaa
1108 LangGraph-workflown.
1110 Args:
1111 pipeline_name: Pipeline metadata.name -arvo.
1112 dirs: Manifestihakemistot (oletus: MANIFEST_DIRS).
1113 builtins: Oma BuiltinExecutor (oletus: _DEFAULT_BUILTINS).
1114 invoke_agent_fn: fn(resolved_spec, step, state) → state.
1115 run_context: Template-kontekstidata, esim. {"budget": {...}, "config": {...}}.
1117 Returns:
1118 Compilattu LangGraph CompiledGraph.
1119 """
1120 registry = _get_registry(dirs)
1121 pipeline = registry.get("Pipeline", pipeline_name)
1122 if pipeline is None:
1123 available = [
1124 (m.get("metadata") or {}).get("name")
1125 for m in registry.all_of_kind("Pipeline")
1126 ]
1127 raise KeyError(
1128 f"Pipeline '{pipeline_name}' not found. Available: {available}"
1129 )
1130 builder = GraphBuilder(registry, builtins, invoke_agent_fn, run_context)
1131 return builder.build(pipeline)