Coverage for core/langgraph/_llm.py: 86%
614 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-28 00:19 +0000
1"""
2_llm.py — OpenAI-compatible LLM invoker + Langfuse tracing + skill injection.
4OpenAIInvoker on callable, joka StepRunner kutsuu agent-stepille
5(`sequential`, `parallel`, `foreach`-body sequential/parallel). Se:
7- kokoaa system-viestin AgentInstance-spec:n tiedoista: renderöity
8 agent-prompti (params-substituutio), jonka perään engine injektoi
9 `spec.skills`-skillit automaattisesti
10- serialisoi `step.input`-arvon user-messageksi
11- ajaa OpenAI-streamin DeepSeek-yhteensopivalla endpointilla
12- raportoi Langfuse-generationina pipeline-trace alle
13- parsii vastauksen envelopeksi: strict-JSON + rajattu deterministinen
14 palautus (≤1 quote/fence-kerros objektin tai listan ympäriltä, lista →
15 iteration); mikä tahansa muu ulostulo (ei-JSON, proosa, rikkinäinen JSON,
16 objekti ilman status-kenttää) → status:"ok" ja raaka content; vain
17 status-kentän sisältävä mutta validoinnissa epäonnistuva envelope →
18 status:error ja fail-fast
19- valvoo RunBudgetia (max LLM calls / wall-clock timeout)
21API:
22 invoker = OpenAIInvoker(config=..., budget=..., project_slug=..., pipeline_name=...)
23 new_state = invoker(resolved_spec, resolved_step, state)
24"""
25from __future__ import annotations
27import contextvars
28import json
29import logging
30import os
31import re
32import subprocess
33import threading
34import time
35import uuid
36from dataclasses import dataclass, field
37from pathlib import Path
38from typing import Any
40log = logging.getLogger(__name__)
42# Workspace-root resolver for the file-access tool — sama dual-import-kuvio
43# kuin runner.py:ssa: flat sys.path (testit, discover-wrapperit) vs. qualified
44# langgraph.* (main.py, PYTHONPATH=/app).
45try:
46 from _files_output import resolve_workspace_root
47except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
48 from langgraph._files_output import resolve_workspace_root
50# Deterministic communication-context block (Phase 3): sama dual-import-kuvio
51# kuin yllä — flat sys.path (testit) vs. qualified langgraph.* (main.py).
52try:
53 from _data_structure import (
54 COMMUNICATION_CONTEXT_KEY,
55 render_communication_block,
56 )
57except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
58 from langgraph._data_structure import (
59 COMMUNICATION_CONTEXT_KEY,
60 render_communication_block,
61 )
63# Static agent output envelope contract (Phase 2): same dual-import pattern as
64# above — flat sys.path (tests) vs. qualified langgraph.* (main.py).
65try:
66 from _envelope import (
67 ENVELOPE_CONTRACT_BLOCK,
68 ENVELOPE_ERROR_CODE,
69 parse_agent_envelope,
70 )
71except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
72 from langgraph._envelope import (
73 ENVELOPE_CONTRACT_BLOCK,
74 ENVELOPE_ERROR_CODE,
75 parse_agent_envelope,
76 )
78# Native-requirement detector for tool calls expressed as assistant-content
79# markup (ADR-0041 D3): same dual-import pattern as above — flat sys.path
80# (tests) vs. qualified langgraph.* (main.py).
81try:
82 from _tool_markup import contains_tool_call_markup
83except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
84 from langgraph._tool_markup import contains_tool_call_markup
86# File-access tool name constants for native-call routing: same dual-import
87# pattern as above — flat sys.path (tests) vs. qualified langgraph.* (main.py).
88try:
89 from _instructions import (
90 LIST_DIR_TOOL_NAME as _LIST_DIR_TOOL_NAME,
91 READ_FILE_TOOL_NAME as _READ_FILE_TOOL_NAME,
92 )
93except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
94 from langgraph._instructions import (
95 LIST_DIR_TOOL_NAME as _LIST_DIR_TOOL_NAME,
96 READ_FILE_TOOL_NAME as _READ_FILE_TOOL_NAME,
97 )
99# Structural tool contract: the resolved `Tool` carrier key and the invocation
100# context handed to `Tool.execute` (Phase 3). Same dual-import pattern.
101try:
102 from tools.contract import RESOLVED_TOOLS_KEY, ToolContext, ToolResult
103except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
104 from langgraph.tools.contract import RESOLVED_TOOLS_KEY, ToolContext, ToolResult
106# Per-tool read_file module: definition + execution single source (Phase 1).
107# Same dual-import pattern as above — flat sys.path (tests) vs. qualified
108# langgraph.* (main.py). The byte/line caps are re-exported for compatibility.
109try:
110 from tools.read_file import (
111 READ_FILE_MAX_BYTES as _READ_FILE_MAX_BYTES, # noqa: F401
112 READ_FILE_MAX_LINES as _READ_FILE_MAX_LINES, # noqa: F401
113 ReadFileResult,
114 read_file as _read_file,
115 )
116except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
117 from langgraph.tools.read_file import (
118 READ_FILE_MAX_BYTES as _READ_FILE_MAX_BYTES, # noqa: F401
119 READ_FILE_MAX_LINES as _READ_FILE_MAX_LINES, # noqa: F401
120 ReadFileResult,
121 read_file as _read_file,
122 )
124# Per-tool list_dir module: definition + execution single source (Phase 2).
125# Same dual-import pattern as above — flat sys.path (tests) vs. qualified
126# langgraph.* (main.py). The cap is re-exported for compatibility.
127try:
128 from tools.list_dir import (
129 LIST_DIR_MAX_ENTRIES as _LIST_DIR_MAX_ENTRIES, # noqa: F401
130 ListDirEntry,
131 ListDirResult,
132 list_dir as _execute_list_dir_tool,
133 )
134except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
135 from langgraph.tools.list_dir import (
136 LIST_DIR_MAX_ENTRIES as _LIST_DIR_MAX_ENTRIES, # noqa: F401
137 ListDirEntry,
138 ListDirResult,
139 list_dir as _execute_list_dir_tool,
140 )
142SKILL_DIR = Path("/skills")
145class SkillLoadError(ValueError):
146 """Skill-latauksen kova virhe (ADR-0012) — ei hiljaista fallbackia.
148 Erottaa skill-resoluution virheet muista ValueErrorista: runnerin
149 parallel-käsittelijät re-raisaavat tämän (kova ajo-failure kaikilla
150 step-polkuilla), kun muut step-virheet pysyvät state-virheinä.
151 """
153# Kontekstin eheys -sopimus (Phase 4 doc, ## Kontekstin eheys):
154# koottu konteksti toimitetaan mallikutsulle kokonaisuudessaan tai step
155# epäonnistuu kovasti — ei hiljaista leikkausta missään muodossa.
156DEFAULT_CONTEXT_BUDGET_TOKENS = 100_000
157DEFAULT_MAX_PARALLEL = 500
158_CONTEXT_BUDGET_ERROR_CODE = "VALIDATION_ERROR"
159_TOOL_FAILURE_ERROR_CODE = "TOOL_FAILURE"
160_TIKTOKEN_ENCODING = "cl100k_base"
162# --- Opt-in tool loop (agent spec.tools resolved into `Tool` instances) ---
163# The invoker receives the calling agent's resolved tools under
164# RESOLVED_TOOLS_KEY and injects exactly those definitions as the request-level
165# `tools`. A tool round = one additional inference call consumed through the run
166# budget. The file-access name constants live in the Core-owned registry
167# (_instructions) and are imported under their local names; every other
168# declared tool executes through the structural `Tool.execute` contract.
169_TOOL_STOP_REASON_BUDGET = "budget"
170_TIKTOKEN_CACHE_DIR = Path(__file__).parent / "tiktoken_cache"
172_LANGFUSE_FLUSH_EVERY = 8
175# ---------------------------------------------------------------------------
176# RunBudget — deterministinen pipeline-ajon suoritusbudjetti
177# ---------------------------------------------------------------------------
180@dataclass
181class RunBudget:
182 max_llm_calls: int = 48
183 run_timeout_seconds: int = 240
184 start: float = field(default_factory=time.monotonic)
185 llm_calls: int = 0
186 stopped: bool = False
187 stop_reason: str = ""
188 _lock: threading.Lock = field(default_factory=threading.Lock)
190 def try_consume(self) -> bool:
191 with self._lock:
192 if self.stopped:
193 return False
194 if time.monotonic() - self.start > self.run_timeout_seconds:
195 self.stopped = True
196 self.stop_reason = "run_timeout"
197 return False
198 if self.llm_calls >= self.max_llm_calls:
199 self.stopped = True
200 self.stop_reason = "max_llm_calls"
201 return False
202 self.llm_calls += 1
203 return True
205 def as_stats(self) -> dict[str, Any]:
206 return {
207 "llm_calls": self.llm_calls,
208 "budget_stopped": self.stopped,
209 "budget_stop_reason": self.stop_reason or None,
210 }
213# ---------------------------------------------------------------------------
214# Skill loading
215# ---------------------------------------------------------------------------
218def _skill_repos_from_env() -> list[dict]:
219 """Lue SKILL_REPOS-env (JSON [{"name": ..., "url": ...}]) → lista entryistä.
221 Puuttuva env → [] (sopimus: SKILL_REPOS on valinnainen). Mikä tahansa
222 epäkelpo arvo on kova virhe (ADR-0012): malformi JSON, ei-lista, tai
223 entry ilman ei-tyhjää name- ja url-merkkijonoa → SkillLoadError.
224 """
225 raw = os.getenv("SKILL_REPOS")
226 if not raw:
227 return []
228 try:
229 repos = json.loads(raw)
230 except json.JSONDecodeError as exc:
231 raise SkillLoadError(f"SKILL_REPOS is not valid JSON: {exc}") from exc
232 if not isinstance(repos, list):
233 raise SkillLoadError(f"SKILL_REPOS must be a JSON list, got {type(repos).__name__}")
234 for idx, repo in enumerate(repos):
235 if not isinstance(repo, dict):
236 raise SkillLoadError(
237 f"SKILL_REPOS entry {idx} must be an object, got {type(repo).__name__}"
238 )
239 name = repo.get("name")
240 url = repo.get("url")
241 if not isinstance(name, str) or not name.strip():
242 raise SkillLoadError(f"SKILL_REPOS entry {idx} is missing a non-empty string 'name'")
243 if not isinstance(url, str) or not url.strip():
244 raise SkillLoadError(
245 f"SKILL_REPOS entry {idx} ('{name}') is missing a non-empty string 'url'"
246 )
247 return repos
250def locate_skills_root(repo_mount: Path) -> Path:
251 """Etsi skill-repo-mountista tasan yksi `skills`-niminen kansio (find).
253 Ajaa hardcodetun find-komennon (ei parametrisoitava):
255 find <mount> -mindepth 1 -type d -name .git -prune \\
256 -o -type d -name skills -print
258 Etsii pelkät `skills`-nimiset kansiot (ei tiedostoja, ei sisäkansioita),
259 `.git`-puu poistetaan (prune), osumat lajitellaan (find-järjestys ei ole
260 taattu → determinismi). Tarkalleen 1 osuma → palauttaa polun Path-
261 objektina. 0 tai >1 osumaa on kova virhe (ADR-0012), kuten myös itse
262 komennon epäonnistuminen (find puuttuu, returncode != 0) — ei hiljaista
263 ohitusta.
264 """
265 cmd = [
266 "find",
267 str(repo_mount),
268 "-mindepth",
269 "1",
270 "-type",
271 "d",
272 "-name",
273 ".git",
274 "-prune",
275 "-o",
276 "-type",
277 "d",
278 "-name",
279 "skills",
280 "-print",
281 ]
282 try:
283 proc = subprocess.run(cmd, capture_output=True, text=True)
284 except OSError as exc:
285 raise SkillLoadError(
286 f"failed to run find for skill mount {repo_mount}: {exc}"
287 ) from exc
288 if proc.returncode != 0:
289 raise SkillLoadError(
290 f"find failed for skill mount {repo_mount} "
291 f"(rc={proc.returncode}): {proc.stderr.strip()}"
292 )
293 hits = sorted(line.strip() for line in proc.stdout.splitlines() if line.strip())
294 if not hits:
295 raise SkillLoadError(f"no skills folder found in {repo_mount}")
296 if len(hits) > 1:
297 raise SkillLoadError(
298 f"ambiguous skills folders found in {repo_mount}: " + ", ".join(hits)
299 )
300 return Path(hits[0])
303def _normalize_skill_libs() -> list[Path]:
304 """Skill-lähdelista: koodirepon skills/-kirjasto ensin, sitten skill-repot.
306 Koodirepon implisiittinen skills/-kirjasto (`PROJECT_ROOT/skills/`,
307 oletus `/project/skills`) on aina korkein prioriteetti skill-lähteenä
308 (koodirepon ylivalta, ADR-0018 D3), ja skill-repot tulevat SKILL_REPOS-
309 järjestyksessä (ADR-0018 D4) listan loppuun: kunkin mountin kohdalla
310 etsitään `skills`-kansio locate_skills_root()-tuloksella. Jos skill-
311 reposta ei löydy skills-kansiota, ajo epäonnistuu kovasti (ADR-0012) —
312 koodirepon puuttuvan skills/-kansion hiljainen ohitus pysyy ennallaan
313 (resoluutio ohittaa puuttuvan kirjaston).
314 """
315 project_root = Path(os.getenv("PROJECT_ROOT", "/project"))
316 return [project_root / "skills"] + [
317 locate_skills_root(SKILL_DIR / repo["name"]) for repo in _skill_repos_from_env()
318 ]
321def load_skills(spec: dict, *, max_chars: int = 16000) -> str:
322 """
323 Lukee `spec.skills`-flat-listan (entryjä) ja palauttaa yhdistetyn
324 tekstin. Kunkin tiedoston yläraja on max_chars merkkiä (oletus
325 16 000): ylittävä tiedosto on kova SkillLoadError (ADR-0012) —
326 ei hiljaista katkaisua.
328 Entry on joko kansionimi (`<nimi>`): koko `<nimi>/`-kansion sisältö
329 injektoidaan otsakkeilla (SKILL.md ensin, sitten loput
330 aakkosjärjestyksessä suhteellisen polun mukaan), tai tarkka
331 tiedosto (`<skill>/<tiedostonimi>`): vain kyseinen tiedosto
332 injektoidaan. Tiedosto-entryt eivät saa sisältää alipolkuja —
333 täsmälleen yksi `/`; virheellinen muoto (johto-/loppuviiva, tyhjä
334 segmentti, `..`, absoluuttinen polku) on kova SkillLoadError.
336 Ensimmäinen skill-lähde, josta kohde löytyy, voittaa (first-match):
337 koodirepon skills/-kirjasto (`PROJECT_ROOT/skills/`, koodirepon
338 ylivalta, ADR-0018 D3) ensin, sitten skill-repot SKILL_REPOS-
339 järjestyksessä (prioriteetti, ADR-0018 D4); puuttuvat lähteet
340 ohitetaan hiljaa.
342 Resolvoitumaton entry → SkillLoadError (ADR-0012): viesti listaa
343 etsityt kirjastopolut ja saatavilla olevat skillit. Ei hiljaista
344 ohitusta.
345 """
346 skills = (spec.get("spec") or {}).get("skills") or []
347 if not skills:
348 return ""
349 libs = _normalize_skill_libs()
350 parts: list[str] = []
351 for entry in skills:
352 parts.extend(_load_bare_skill(entry, libs, max_chars=max_chars))
353 return "\n\n---\n\n".join(parts)
356def _parse_skill_entry(entry: str) -> tuple[str, str | None]:
357 """Parseroi spec.skills-entryn: kansionimi → (nimi, None), tarkka
358 tiedosto `<skill>/<tiedostonimi>` → (skill, tiedostonimi).
360 Virheellinen muoto → kova SkillLoadError: tyhjä entry, `.`/`..`,
361 tyhjä segmentti, absoluuttinen polku tai >1 vinoviivaa (alipolut
362 eivät ole sallittuja).
363 """
364 if not entry.strip():
365 raise SkillLoadError(
366 f"skill entry '{entry}' is empty; use '<skill>' "
367 "or '<skill>/<file>'"
368 )
369 if entry.startswith("/"):
370 raise SkillLoadError(
371 f"skill entry '{entry}' is an absolute path; use '<skill>' "
372 "or '<skill>/<file>'"
373 )
374 parts = entry.split("/")
375 if any(seg in ("", ".", "..") for seg in parts):
376 raise SkillLoadError(
377 f"skill entry '{entry}' contains an invalid path segment "
378 "(''|'.'|'..'); use '<skill>' or '<skill>/<file>'"
379 )
380 if len(parts) == 1:
381 return entry, None
382 if len(parts) > 2:
383 raise SkillLoadError(
384 f"skill entry '{entry}' has too many path segments; subpaths "
385 "are not allowed, use '<skill>' or '<skill>/<file>'"
386 )
387 skill, filename = parts
388 return skill, filename
391def _load_bare_skill(entry: str, libs: list[Path], *, max_chars: int) -> list[str]:
392 """Skill-entry (kansionimi tai tarkka tiedosto) ladattuna otsakkeilla.
394 Kansionimi (`<nimi>`): kansion `lib / <nimi>/` kaikki tiedostot
395 injektoidaan (SKILL.md ensin, sitten loput aakkosjärjestyksessä).
396 Tarkka tiedosto (`<skill>/<tiedostonimi>`): vain kyseinen tiedosto
397 `lib / <skill> / <tiedosto>` injektoidaan otsakkeella
398 `## Skill: <skill> — <tiedosto>`.
400 Ensimmäinen skill-lähde, josta kohde löytyy, voittaa (first-match).
401 Kansionimelle vastaava `lib / entry` tiedostona → SkillLoadError;
402 tiedosto-entrylle kansion löytyminen ilman kohdetiedostoa ohittaa
403 kyseisen lähteen hiljaa. Virheellinen entry-muoto tai kohde, jota
404 ei löydy mistään skill-lähteestä → SkillLoadError.
405 """
406 name, filename = _parse_skill_entry(entry)
407 for lib in libs:
408 folder = lib / name
409 if filename is None:
410 if folder.is_dir():
411 parts: list[str] = []
412 for path in _skill_folder_files(folder):
413 relpath = path.relative_to(folder)
414 parts.append(
415 f"## Skill: {name} — {relpath}\n\n"
416 f"{_read_skill_file(entry, path, max_chars=max_chars)}"
417 )
418 return parts
419 if folder.is_file():
420 raise SkillLoadError(
421 f"skill '{entry}' resolves to a file, expected a skill folder "
422 f"(SKILL.md etc.): {folder}"
423 )
424 else:
425 file_path = folder / filename
426 if file_path.is_file():
427 return [
428 f"## Skill: {name} — {filename}\n\n"
429 f"{_read_skill_file(entry, file_path, max_chars=max_chars)}"
430 ]
431 if folder.is_dir():
432 continue
433 raise SkillLoadError(_unresolved_skill_error(entry, libs))
436def _skill_folder_files(folder: Path) -> list[Path]:
437 """Kansion kaikki tiedostot rekursiivisesti, deterministinen järjestys:
438 juuren SKILL.md ensin, sitten loput suhteellisen polun mukaan
439 aakkosjärjestyksessä."""
440 files = [p for p in folder.rglob("*") if p.is_file()]
441 files.sort(
442 key=lambda p: (
443 p.relative_to(folder) != Path("SKILL.md"),
444 str(p.relative_to(folder)),
445 )
446 )
447 return files
450def _read_skill_file(entry: str, path: Path, *, max_chars: int) -> str:
451 try:
452 text = path.read_text(encoding="utf-8", errors="replace")
453 except OSError as exc:
454 raise SkillLoadError(
455 f"skill entry '{entry}' file unreadable: {path} ({exc})"
456 ) from exc
457 if len(text) > max_chars:
458 raise SkillLoadError(
459 f"skill entry '{entry}' file exceeds the {max_chars}-char limit: "
460 f"{path} ({len(text)} chars)"
461 )
462 return text
465def _available_skill_names(libs: list[Path]) -> list[str]:
466 """Kirjastojen tarjonta: skill-kansiot (sisältävät SKILL.md) + top-tason
467 *.md-tiedostot. Sorted + deduped."""
468 names: set[str] = set()
469 for lib in libs:
470 if not lib.is_dir():
471 continue
472 for child in lib.iterdir():
473 if child.is_dir():
474 if (child / "SKILL.md").is_file():
475 names.add(child.name)
476 elif child.is_file() and child.name.endswith(".md"):
477 names.add(child.name)
478 return sorted(names)
481def _unresolved_skill_error(entry: str, libs: list[Path]) -> str:
482 available = ", ".join(_available_skill_names(libs)) or "(none)"
483 searched = "\n".join(f" - {lib}" for lib in libs)
484 return (
485 f"skill '{entry}' not found in any skill repo "
486 f"(available skills: {available}). Searched skill repos:\n{searched}"
487 )
490# ---------------------------------------------------------------------------
491# Template substitution (simple {{ key }} replacement against a dict)
492# ---------------------------------------------------------------------------
494_TMPL_RE = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_.]*)\s*\}\}")
497def _render(text: str, ctx: dict[str, Any]) -> str:
498 if not isinstance(text, str) or "{{" not in text:
499 return text
501 def _lookup(path: str) -> str:
502 cur: Any = ctx
503 for part in path.split("."):
504 if isinstance(cur, dict):
505 cur = cur.get(part)
506 else:
507 return ""
508 if cur is None:
509 return ""
510 return str(cur) if not isinstance(cur, str) else cur
512 return _TMPL_RE.sub(lambda m: _lookup(m.group(1)), text)
515# ---------------------------------------------------------------------------
516# Token counting (deterministinen ja provider-agnostinen; Phase 4 doc)
517# ---------------------------------------------------------------------------
519_tokenizer_encoding: Any | None = None
520_tokenizer_lock = threading.Lock()
523def _tokenizer() -> Any:
524 """tiktoken-encoding lazy-singleton, joka käyttää bundlattua
525 TIKTOKEN_CACHE_DIR:ää (core/langgraph/tiktoken_cache/) — ei
526 verkkolatausta ensimmäiselläkään käyttökerralla."""
527 global _tokenizer_encoding
528 if _tokenizer_encoding is None:
529 with _tokenizer_lock:
530 if _tokenizer_encoding is None:
531 if not os.environ.get("TIKTOKEN_CACHE_DIR"):
532 os.environ["TIKTOKEN_CACHE_DIR"] = str(_TIKTOKEN_CACHE_DIR)
533 from tiktoken import get_encoding
535 _tokenizer_encoding = get_encoding(_TIKTOKEN_ENCODING)
536 return _tokenizer_encoding
539def count_context_tokens(messages: list[dict[str, str]]) -> int:
540 """Kootun kontekstin koko tokeneina: system + user -viestien sisältöjen
541 summa. Deterministinen, ei riipu providerista eikä kutsutusta mallista."""
542 enc = _tokenizer()
543 total = 0
544 for message in messages:
545 content = message.get("content")
546 if isinstance(content, str):
547 total += len(enc.encode_ordinary(content))
548 return total
551# ---------------------------------------------------------------------------
552# Langfuse trace helpers
553# ---------------------------------------------------------------------------
555_PARENT_STACK: contextvars.ContextVar[tuple[str, ...]] = contextvars.ContextVar(
556 "langfuse_parent_stack", default=()
557)
560class LangfuseTracer:
561 """
562 Per-ajo Langfuse-tracen alustus + generation-spanit per LLM-kutsu.
563 Käyttää singleton-Langfuse-clientiä env-muuttujista.
564 """
566 def __init__(self, project_slug: str, pipeline_name: str, run_id: str):
567 self.project_slug = project_slug
568 self.pipeline_name = pipeline_name
569 self.run_id = run_id or "unknown"
570 self._lf: Any = None
571 self._trace_ctx: dict[str, Any] | None = None
572 self._flush_counter = 0
573 self._init_done = False
575 def _client(self) -> Any:
576 if self._lf is not None:
577 return self._lf
578 try:
579 from langfuse import Langfuse # type: ignore
580 except ImportError:
581 log.warning("[langfuse] package not installed — tracing disabled")
582 return None
583 prev = os.environ.get("LANGFUSE_TRACING_ENABLED")
584 os.environ["LANGFUSE_TRACING_ENABLED"] = "true"
585 try:
586 self._lf = Langfuse()
587 except Exception as exc: # noqa: BLE001
588 log.warning("[langfuse] init failed: %s — tracing disabled", exc)
589 self._lf = None
590 finally:
591 if prev is None:
592 os.environ["LANGFUSE_TRACING_ENABLED"] = "false"
593 else:
594 os.environ["LANGFUSE_TRACING_ENABLED"] = prev
595 return self._lf
597 @property
598 def trace_name(self) -> str:
599 return f"{self.project_slug}:{self.pipeline_name}"
601 @property
602 def session_id(self) -> str:
603 return f"{self.project_slug}-{self.pipeline_name}"
605 def _ctx(self) -> dict[str, Any] | None:
606 lf = self._client()
607 if lf is None:
608 return None
609 if self._trace_ctx is None:
610 self._trace_ctx = {"trace_id": lf.create_trace_id(seed=self.run_id)}
611 return self._trace_ctx
613 def _ctx_with_parent(self) -> dict[str, Any] | None:
614 ctx = self._ctx()
615 if ctx is None:
616 return None
617 stack = _PARENT_STACK.get()
618 if stack:
619 return {**ctx, "parent_span_id": stack[-1]}
620 return ctx
622 def open_parent_observation(
623 self,
624 *,
625 name: str,
626 input: dict[str, Any] | None = None,
627 metadata: dict[str, Any] | None = None,
628 ) -> Any:
629 lf = self._client()
630 ctx = self._ctx_with_parent()
631 if lf is None or ctx is None:
632 return _NoopObservation()
633 try:
634 obs = lf.start_observation(
635 trace_context=ctx,
636 name=name,
637 as_type="span",
638 input=input or {},
639 metadata=metadata or {},
640 )
641 except Exception as exc: # noqa: BLE001
642 log.warning("[langfuse] open_parent_observation failed: %s", exc)
643 return _NoopObservation()
644 stack = _PARENT_STACK.get()
645 _PARENT_STACK.set((*stack, obs.id))
646 return obs
648 def close_parent_observation(self, obs: Any) -> None:
649 if isinstance(obs, _NoopObservation):
650 return
651 stack = _PARENT_STACK.get()
652 if stack and stack[-1] == obs.id:
653 _PARENT_STACK.set(stack[:-1])
654 obs.end()
656 def init_trace(self) -> None:
657 if self._init_done:
658 return
659 lf = self._client()
660 ctx = self._ctx()
661 if lf is None or ctx is None:
662 self._init_done = True
663 return
664 try:
665 obs = lf.start_observation(
666 trace_context=ctx,
667 name=self.trace_name,
668 as_type="span",
669 input={
670 "pipeline": self.pipeline_name,
671 "project": self.project_slug,
672 "run_id": self.run_id,
673 },
674 )
675 self._apply_meta(obs)
676 obs.end()
677 self.flush(force=True)
678 except Exception as exc: # noqa: BLE001
679 log.warning("[langfuse] init_trace failed: %s", exc)
680 self._init_done = True
682 def _apply_meta(self, obs: Any) -> None:
683 try:
684 obs.update_trace(
685 name=self.trace_name,
686 user_id=self.project_slug,
687 session_id=self.session_id,
688 tags=[f"pipeline:{self.pipeline_name}"],
689 )
690 except Exception: # noqa: BLE001
691 pass
693 def generation(
694 self,
695 name: str,
696 *,
697 model: str,
698 temperature: float,
699 messages: list[dict[str, str]],
700 metadata: dict[str, Any] | None = None,
701 ) -> Any:
702 lf = self._client()
703 ctx = self._ctx()
704 if lf is None or ctx is None:
705 return _NoopObservation()
706 try:
707 obs = lf.start_observation(
708 trace_context=self._ctx_with_parent(),
709 name=name,
710 as_type="generation",
711 model=model,
712 model_parameters={"temperature": temperature},
713 input=messages,
714 metadata=metadata,
715 )
716 self._apply_meta(obs)
717 return obs
718 except Exception as exc: # noqa: BLE001
719 log.warning("[langfuse] generation failed: %s", exc)
720 return _NoopObservation()
722 def flush(self, *, force: bool = False) -> None:
723 lf = self._client()
724 if lf is None:
725 return
726 self._flush_counter += 1
727 if force or self._flush_counter % _LANGFUSE_FLUSH_EVERY == 0:
728 try:
729 lf.flush()
730 except Exception: # noqa: BLE001
731 pass
734class _NoopObservation:
735 def update(self, **_: Any) -> None: ...
736 def end(self) -> None: ...
739# ---------------------------------------------------------------------------
740# File-access tool execution + deterministic-tool soft result
741# ---------------------------------------------------------------------------
744def _invalid_tool_arguments_result(
745 name: str, reason: str
746) -> ReadFileResult | ListDirResult:
747 """Malformed tool arguments (unparseable JSON, non-object JSON, wrong
748 types) are an expected domain error: a soft, model-visible tool result —
749 never the technical-failure marker (ADR-0012, ADR-0030 D8)."""
750 if name == _LIST_DIR_TOOL_NAME:
751 return ListDirResult([], False, "", f"{_LIST_DIR_TOOL_NAME}: {reason}")
752 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: {reason}")
755def _file_result_from_tool_result(
756 name: str, result: ToolResult
757) -> ReadFileResult | ListDirResult:
758 """Adapt a file tool's `ToolResult` back to its module result shape."""
759 state_patch = result.state_patch or {}
760 if name == _LIST_DIR_TOOL_NAME:
761 return ListDirResult(
762 _list_dir_entries(result.content),
763 bool(state_patch.get("truncated", False)),
764 str(state_patch.get("root", "")),
765 result.error,
766 )
767 return ReadFileResult(
768 result.content,
769 bool(state_patch.get("truncated", False)),
770 result.error,
771 )
774def _list_dir_entries(content: str) -> list[ListDirEntry]:
775 """Decode the list_dir tool's JSON entry array; malformed content is empty."""
776 try:
777 items = json.loads(content)
778 except (TypeError, json.JSONDecodeError):
779 return []
780 if not isinstance(items, list):
781 return []
782 return [
783 ListDirEntry(
784 path=item["path"],
785 name=item["name"],
786 size=item["size"],
787 kind=item["kind"],
788 )
789 for item in items
790 if isinstance(item, dict)
791 ]
794def _parse_deterministic_tool_arguments(name: str, tool_call: Any) -> dict:
795 """The call's arguments object, else a hard CALL_INVALID_PARAMS failure."""
796 try:
797 args = json.loads(getattr(tool_call.function, "arguments", "{}"))
798 except (TypeError, json.JSONDecodeError) as exc:
799 raise _ToolHardFailure(
800 "CALL_INVALID_PARAMS",
801 f"tool '{name}' has invalid arguments: {exc}",
802 ) from exc
803 if not isinstance(args, dict):
804 raise _ToolHardFailure(
805 "CALL_INVALID_PARAMS",
806 f"tool '{name}' arguments must be a JSON object, got "
807 f"{type(args).__name__}",
808 )
809 return args
812class _ToolHardFailure(Exception):
813 """A native tool call that must fail the run with a CALL_* hard code."""
815 def __init__(self, code: str, message: str) -> None:
816 super().__init__(message)
817 self.code = code
818 self.message = message
821def _execute_tool_or_call_failed(
822 tool: Any, context: ToolContext, parent_step_id: str, name: str
823) -> Any:
824 """Run one resolved tool, mapping technical failures to CALL_* codes.
826 A `BUILTIN_NAME_COLLISION` a tool raises is preserved; every other
827 exception becomes a hard `CALL_FAILED` (the legacy deterministic-dispatch
828 contract), so the caller's `tool_failure` mapping stays observable.
829 """
830 try:
831 return tool.execute(context)
832 except _ToolHardFailure:
833 raise
834 except Exception as exc: # noqa: BLE001
835 code = (
836 "BUILTIN_NAME_COLLISION"
837 if getattr(exc, "code", None) == "BUILTIN_NAME_COLLISION"
838 else "CALL_FAILED"
839 )
840 message = getattr(exc, "message", None) or f"{type(exc).__name__}: {exc}"
841 raise _ToolHardFailure(
842 code, f"step '{parent_step_id}' tool '{name}' failed: {message}"
843 ) from exc
846def _synthesize_call_step(
847 parent_step_id: str, index: int, name: str, params: dict, state: dict
848) -> dict:
849 """Build the synthetic call step for one deterministic tool invocation.
851 Mirrors ``StepRunner._synthesize_call_step`` so state-patch keys keep the
852 engine's id schema: the engine-local iteration discriminator (when
853 present) plus a process-unique token keep concurrent calls distinct.
854 """
855 try:
856 from runner import _iteration_discriminator
857 except ImportError: # pragma: no cover — main.py run (qualified imports)
858 from langgraph.runner import _iteration_discriminator
860 discriminator = _iteration_discriminator(state)
861 token = uuid.uuid4().hex
862 suffix = f"{discriminator}.{token}" if discriminator else token
863 return {
864 "id": f"{parent_step_id}.call.{index}.{name}.{suffix}",
865 "type": "builtin",
866 "component": name,
867 "params": params,
868 }
871@dataclass
872class DispatchedToolResult:
873 """A deterministic tool executed through the structural `Tool` contract.
875 `result` carries the tool's model-facing content envelope (the `ToolResult`
876 content), which is also the message fed back to the model.
877 """
879 name: str
880 arguments: dict
881 result: Any
882 truncated: bool = False
883 error: str | None = None
885 def as_tool_message(self) -> str:
886 if isinstance(self.result, str):
887 return self.result
888 return json.dumps(self.result, ensure_ascii=False, default=str)
891# ---------------------------------------------------------------------------
892# OpenAIInvoker
893# ---------------------------------------------------------------------------
896class OpenAIInvoker:
897 """
898 Default LLM invoker for agent-platform pipelines.
900 Yksi instanssi per pipeline-ajo. Thread-safe (semafori rinnakkaisuudelle).
901 """
903 def __init__(
904 self,
905 *,
906 providers: dict[str, dict[str, Any]],
907 max_parallel: int,
908 budget: RunBudget,
909 tracer: LangfuseTracer,
910 ) -> None:
911 self._providers = providers
912 self._clients: dict[str, Any] = {}
913 self.budget = budget
914 self.tracer = tracer
915 self._semaphore = threading.Semaphore(max(1, max_parallel))
917 def _openai_client(self, base_url: str, api_key: str) -> Any:
918 key = f"{base_url}|{api_key}"
919 if key not in self._clients:
920 from openai import OpenAI # type: ignore
921 self._clients[key] = OpenAI(base_url=base_url, api_key=api_key)
922 return self._clients[key]
924 # ---- Public callable -------------------------------------------------
926 def __call__(
927 self,
928 resolved_spec: dict,
929 resolved_step: dict,
930 state: dict,
931 ) -> dict:
932 """
933 StepRunner-yhteensopiva: (spec, step, state) → uusi state.
935 `resolved_step` = step jossa params/input on JO resolvoitu
936 StepRunnerissa template-engineä käyttäen. Tämä invoker ei enää
937 evaluoi `{{ }}`-lausekkeita state/run_context-vastaan, vain
938 substituoi `step.params` ja sisäänrakennetut feature-arvot
939 agentin `system_prompt`/`user`-mallineisiin.
940 """
941 step_id = resolved_step.get("id", "?")
942 agent_spec = resolved_spec.get("spec") or {}
943 resolved_tools = resolved_step.get(RESOLVED_TOOLS_KEY) or []
945 model_spec = agent_spec.get("model") or {}
946 provider_name = model_spec.get("provider")
948 if not provider_name:
949 if self._providers:
950 provider_name = next(iter(self._providers))
951 else:
952 raise _hard_failure(
953 "VALIDATION_ERROR",
954 "No LLM provider configured and no provider specified in agent spec",
955 )
957 provider_cfg = self._providers.get(provider_name)
958 if provider_cfg is None:
959 raise _hard_failure(
960 "VALIDATION_ERROR",
961 f"Unknown LLM provider '{provider_name}'. "
962 f"Available: {', '.join(sorted(self._providers.keys()))}",
963 )
965 model_override = model_spec.get("model")
966 model = model_override if model_override else provider_cfg["model"]
968 # Budget check
969 if not self.budget.try_consume():
970 return {
971 **state,
972 step_id: _with_failure_flags(
973 {"error": f"budget:{self.budget.stop_reason}"}
974 ),
975 }
977 # ---- Build system + user messages --------------------------------
978 skills_text = load_skills(resolved_spec)
979 step_params = resolved_step.get("params") or {}
980 input_value = resolved_step.get("input")
982 # Render system_prompt template with params, then inject the loaded
983 # skills after the rendered prompt (ADR-0036 D1/D6).
984 prompt_ctx: dict[str, Any] = {
985 k: v for k, v in step_params.items() if isinstance(k, str)
986 }
987 rendered_prompt = _render(
988 str(agent_spec.get("system_prompt") or ""),
989 prompt_ctx,
990 ).strip()
991 communication = resolved_step.get(COMMUNICATION_CONTEXT_KEY)
992 communication_text = (
993 render_communication_block(communication) if communication else ""
994 )
995 envelope_text = ENVELOPE_CONTRACT_BLOCK
996 system_text = "\n\n".join(
997 part
998 for part in (
999 rendered_prompt,
1000 skills_text,
1001 envelope_text,
1002 communication_text,
1003 )
1004 if part
1005 )
1007 # User message: serialise input + step params
1008 user_text = _format_user_message(input_value, step_params)
1010 # Temperature: agent spec → provider default → step params
1011 temperature = float(
1012 model_spec.get("temperature", provider_cfg.get("temperature", 0.0))
1013 )
1014 if "temperature" in step_params:
1015 try:
1016 temperature = float(step_params["temperature"])
1017 except (TypeError, ValueError):
1018 pass
1020 messages = [
1021 {"role": "system", "content": system_text},
1022 {"role": "user", "content": user_text},
1023 ]
1025 # ---- Context-integrity check (Phase 4 doc, ## Kontekstin eheys) --
1026 agent_name = (resolved_spec.get("metadata") or {}).get("name", "agent")
1027 context_budget = _effective_context_budget(resolved_spec, agent_name)
1028 context_tokens = count_context_tokens(messages)
1029 if context_tokens > context_budget:
1030 return _with_context_metrics(
1031 state,
1032 step_id,
1033 _context_overflow_result(step_id, context_tokens, context_budget),
1034 {"context_size_tokens": context_tokens, "context_budget_tokens": context_budget},
1035 )
1036 measurement = {
1037 "context_size_tokens": context_tokens,
1038 "context_budget_tokens": context_budget,
1039 }
1041 # ---- LLM call with Langfuse generation ---------------------------
1042 gen_name = f"{step_id}:{agent_name}"
1044 with self._semaphore:
1045 obs = self.tracer.generation(
1046 gen_name,
1047 model=model,
1048 temperature=temperature,
1049 messages=messages,
1050 metadata=measurement,
1051 )
1052 content = ""
1053 usage: dict[str, int] | None = None
1054 err: str | None = None
1055 tool_meta: dict[str, Any] | None = None
1056 try:
1057 if resolved_tools:
1058 content, usage, tool_meta = self._stream_completion_with_tools(
1059 messages,
1060 temperature,
1061 model,
1062 provider_cfg["base_url"],
1063 provider_cfg["api_key"],
1064 project_root=resolve_workspace_root(state),
1065 tools=resolved_tools,
1066 state=state,
1067 step_id=step_id,
1068 )
1069 else:
1070 content, usage = self._stream_completion(messages, temperature, model, provider_cfg["base_url"], provider_cfg["api_key"])
1071 except Exception as exc: # noqa: BLE001
1072 err = f"{type(exc).__name__}: {exc}"
1073 log.warning("[invoke] %s LLM call failed: %s", step_id, err)
1074 try:
1075 obs.update(
1076 output=content[:8000] if len(content) > 8000 else content,
1077 usage_details=usage,
1078 )
1079 except Exception: # noqa: BLE001
1080 pass
1081 try:
1082 obs.end()
1083 except Exception: # noqa: BLE001
1084 pass
1085 self.tracer.flush()
1087 # ---- Parse output ------------------------------------------------
1088 if err:
1089 return _with_context_metrics(
1090 state,
1091 step_id,
1092 _with_failure_flags({"error": err, "raw": content}),
1093 measurement,
1094 tool_meta,
1095 )
1097 if tool_meta and tool_meta.get("tool_failure"):
1098 return _with_context_metrics(
1099 state,
1100 step_id,
1101 _tool_failure_result(step_id, tool_meta["tool_failure"]),
1102 measurement,
1103 tool_meta,
1104 )
1106 envelope, parse_message = parse_agent_envelope(content)
1107 if envelope is not None:
1108 return _with_context_metrics(
1109 state, step_id, envelope, measurement, tool_meta
1110 )
1111 return _with_context_metrics(
1112 state,
1113 step_id,
1114 {
1115 "status": "error",
1116 "error": {"code": ENVELOPE_ERROR_CODE, "message": parse_message},
1117 },
1118 measurement,
1119 tool_meta,
1120 )
1122 # ---- Streaming completion -------------------------------------------
1124 def _stream_completion(
1125 self,
1126 messages: list[dict[str, str]],
1127 temperature: float,
1128 model: str,
1129 base_url: str,
1130 api_key: str,
1131 ) -> tuple[str, dict[str, int] | None]:
1132 client = self._openai_client(base_url, api_key)
1133 resp = client.chat.completions.create(
1134 model=model,
1135 temperature=temperature,
1136 messages=messages,
1137 stream=True,
1138 stream_options={"include_usage": True},
1139 )
1140 content = ""
1141 usage: dict[str, int] | None = None
1142 for chunk in resp:
1143 if chunk.choices and chunk.choices[0].delta.content:
1144 content += chunk.choices[0].delta.content
1145 if getattr(chunk, "usage", None):
1146 u = chunk.usage
1147 usage = {
1148 "input": u.prompt_tokens or 0,
1149 "output": u.completion_tokens or 0,
1150 "total": u.total_tokens or 0,
1151 }
1152 return content, usage
1154 def _run_file_tool(
1155 self,
1156 tool_call: Any,
1157 tool: Any,
1158 project_root: Path,
1159 state: dict,
1160 parent_step_id: str,
1161 ) -> ReadFileResult | ListDirResult:
1162 """Execute a read_file or list_dir call through its resolved `Tool`.
1164 File tools keep the soft malformed-arguments contract and return their
1165 module result shape, so they stay out of the dispatched-call record.
1166 """
1167 name = getattr(tool_call.function, "name", "")
1168 try:
1169 args = json.loads(getattr(tool_call.function, "arguments", "{}"))
1170 except (TypeError, json.JSONDecodeError) as exc:
1171 return _invalid_tool_arguments_result(name, f"invalid arguments: {exc}")
1172 if not isinstance(args, dict):
1173 return _invalid_tool_arguments_result(
1174 name,
1175 f"invalid arguments: expected a JSON object, got {type(args).__name__}",
1176 )
1177 context = ToolContext(
1178 args=args,
1179 state=state,
1180 root=project_root,
1181 step=_synthesize_call_step(parent_step_id, 0, name, args, state),
1182 invoker=self,
1183 )
1184 return _file_result_from_tool_result(name, tool.execute(context))
1186 def _execute_tool_call(
1187 self,
1188 tool_call: Any,
1189 project_root: Path,
1190 tools: list[Any],
1191 state: dict,
1192 step_id: str,
1193 ) -> ReadFileResult | ListDirResult | DispatchedToolResult:
1194 """Route a native tool call: declaration gate, then file-access first."""
1195 name = getattr(tool_call.function, "name", "")
1196 resolved_by_name = {tool.name: tool for tool in tools}
1197 if name not in resolved_by_name:
1198 raise _ToolHardFailure(
1199 "CALL_NOT_ALLOWED",
1200 f"tool '{name}' is not among the agent's declared tools",
1201 )
1202 if name in (_READ_FILE_TOOL_NAME, _LIST_DIR_TOOL_NAME):
1203 return self._run_file_tool(
1204 tool_call, resolved_by_name[name], project_root, state, step_id
1205 )
1206 return self._execute_deterministic_tool(
1207 tool_call, resolved_by_name[name], step_id, state
1208 )
1210 def _execute_deterministic_tool(
1211 self, tool_call: Any, tool: Any, parent_step_id: str, state: dict
1212 ) -> DispatchedToolResult:
1213 """Execute a declared deterministic tool through the `Tool` contract."""
1214 name = getattr(tool_call.function, "name", "")
1215 args = _parse_deterministic_tool_arguments(name, tool_call)
1216 context = ToolContext(
1217 args=args,
1218 state=state,
1219 root=resolve_workspace_root(state),
1220 step=_synthesize_call_step(parent_step_id, 0, name, args, state),
1221 invoker=self,
1222 )
1223 result = _execute_tool_or_call_failed(tool, context, parent_step_id, name)
1224 state_patch = result.state_patch or {}
1225 return DispatchedToolResult(
1226 name,
1227 args,
1228 result.content,
1229 truncated=bool(state_patch.get("truncated", False)),
1230 error=result.error,
1231 )
1233 # ---- Tool-looping completion (resolved tool entries) ----------------
1235 def _stream_completion_with_tools(
1236 self,
1237 messages: list[dict[str, str]],
1238 temperature: float,
1239 model: str,
1240 base_url: str,
1241 api_key: str,
1242 *,
1243 project_root: Path,
1244 tools: list[Any],
1245 state: dict,
1246 step_id: str,
1247 ) -> tuple[str, dict[str, int] | None, dict[str, Any]]:
1248 """Non-streaming completion carrying the resolved tool loop.
1250 The request-level `tools` are exactly the resolved tools' definitions in
1251 declaration order, and a tool-declaring agent's request carries them on
1252 every round. Each tool round consumes ONE additional inference through
1253 `self.budget.try_consume`; the loop is bounded only by the run budget and
1254 stops when the model returns no `tool_calls` or the budget is exhausted.
1255 A technical tool failure is surfaced via `tool_meta["tool_failure"]` for
1256 the caller to turn into a hard error. The no-tools path never reaches
1257 here.
1258 """
1259 request_tools = [tool.definition() for tool in tools]
1260 client = self._openai_client(base_url, api_key)
1261 working = list(messages)
1262 content = ""
1263 usage: dict[str, int] | None = None
1264 tool_rounds = 0
1265 truncated = False
1266 tool_errors: list[str] = []
1267 tool_failure: dict | None = None
1268 dispatched_calls: list[dict] = []
1269 stop_reason: str | None = None
1271 while True:
1272 resp = client.chat.completions.create(
1273 model=model,
1274 temperature=temperature,
1275 messages=working,
1276 tools=request_tools,
1277 stream=False,
1278 )
1279 usage = _usage_from_response(resp) or usage
1280 msg = resp.choices[0].message
1281 piece = getattr(msg, "content", None) or ""
1282 tool_calls = getattr(msg, "tool_calls", None)
1283 if not tool_calls:
1284 content = piece
1285 if contains_tool_call_markup(piece):
1286 tool_failure = {
1287 "code": _TOOL_FAILURE_ERROR_CODE,
1288 "message": (
1289 "ADR-0041 D3: assistant expressed a tool call as "
1290 "content markup while native tool_calls was absent"
1291 ),
1292 }
1293 break
1294 working.append(
1295 {
1296 "role": "assistant",
1297 "content": piece or "",
1298 "tool_calls": [
1299 {
1300 "id": tc.id,
1301 "type": "function",
1302 "function": {"name": tc.function.name, "arguments": tc.function.arguments},
1303 }
1304 for tc in tool_calls
1305 ],
1306 }
1307 )
1308 for tc in tool_calls:
1309 try:
1310 result = self._execute_tool_call(
1311 tc, project_root, tools, state, step_id
1312 )
1313 except _ToolHardFailure as exc:
1314 tool_failure = {"code": exc.code, "message": exc.message}
1315 break
1316 except Exception as exc: # noqa: BLE001
1317 tool_failure = _tool_failure_from_exception(exc)
1318 break
1319 truncated = truncated or result.truncated
1320 if result.error:
1321 tool_errors.append(result.error)
1322 working.append(
1323 {"role": "tool", "tool_call_id": tc.id, "content": result.as_tool_message()}
1324 )
1325 if isinstance(result, DispatchedToolResult):
1326 dispatched_calls.append(
1327 {
1328 "name": result.name,
1329 "arguments": result.arguments,
1330 "result": result.result,
1331 }
1332 )
1333 if tool_failure is not None:
1334 break
1335 tool_rounds += 1
1336 if not self.budget.try_consume():
1337 stop_reason = _TOOL_STOP_REASON_BUDGET
1338 break
1340 tool_meta: dict[str, Any] = {
1341 "tool_rounds": tool_rounds,
1342 "truncated": truncated,
1343 "tool_stopped": stop_reason is not None,
1344 "tool_stop_reason": stop_reason,
1345 "tool_errors": tool_errors,
1346 "dispatched_calls": dispatched_calls,
1347 }
1348 if tool_failure is not None:
1349 tool_meta["tool_failure"] = tool_failure
1350 return content, usage, tool_meta
1353# ---------------------------------------------------------------------------
1354# Helpers
1355# ---------------------------------------------------------------------------
1358def _hard_failure(code: str, message: str, details: dict | None = None) -> Any:
1359 """Build the engine's RunHardFailure without a module-level circular import."""
1360 try:
1361 from runner import RunHardFailure
1362 except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
1363 from langgraph.runner import RunHardFailure
1364 return RunHardFailure(code, message, details)
1367def _usage_from_response(resp: Any) -> dict[str, int] | None:
1368 """OpenAI-vastauksen usage-dict (input/output/total) tai None, kun
1369 provider ei palauta usagea."""
1370 u = getattr(resp, "usage", None)
1371 if not u:
1372 return None
1373 return {
1374 "input": getattr(u, "prompt_tokens", 0) or 0,
1375 "output": getattr(u, "completion_tokens", 0) or 0,
1376 "total": getattr(u, "total_tokens", 0) or 0,
1377 }
1380def _effective_context_budget(resolved_spec: dict, agent_name: str) -> int:
1381 """Efektiivinen kontekstibudjetti: spec.limits.context_budget (merge-
1382 järjestyksen tulos) tai oletus 100 000 tokenia. Epäkelpo arvo on kova
1383 virhe (ADR-0012) — ei hiljaista fallbackia."""
1384 limits = (resolved_spec.get("spec") or {}).get("limits") or {}
1385 value = limits.get("context_budget")
1386 if value is None:
1387 return DEFAULT_CONTEXT_BUDGET_TOKENS
1388 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1389 raise _hard_failure(
1390 "VALIDATION_ERROR",
1391 f"agent '{agent_name}': spec.limits.context_budget must be a "
1392 f"positive integer, got {value!r} (ADR-0012)",
1393 )
1394 return value
1397def _context_overflow_result(step_id: str, context_tokens: int, context_budget: int) -> dict:
1398 """Budjetin ylitys → kova virhe ennen mallikutsua (fail-fast)."""
1399 return {
1400 "error": (
1401 f"{_CONTEXT_BUDGET_ERROR_CODE}: step '{step_id}' assembled "
1402 f"context is {context_tokens} tokens, exceeding the context "
1403 f"budget of {context_budget} tokens"
1404 ),
1405 "details": {
1406 "error_code": _CONTEXT_BUDGET_ERROR_CODE,
1407 "step_id": step_id,
1408 "assembled_tokens": context_tokens,
1409 "context_budget": context_budget,
1410 },
1411 }
1414def _tool_failure_from_exception(exc: Exception) -> dict:
1415 """Map a dispatch exception to the hard-failure dict, preserving its code.
1417 The deterministic dispatch wraps execution errors in a RunHardFailure
1418 carrying a CALL_* code (CALL_FAILED / BUILTIN_NAME_COLLISION). ``_llm``
1419 cannot import ``runner`` without a cycle, so the code is duck-typed: any
1420 exception exposing a non-empty string ``.code`` keeps that code; anything
1421 else stays the legacy technical TOOL_FAILURE.
1422 """
1423 code = getattr(exc, "code", None)
1424 if isinstance(code, str) and code:
1425 message = getattr(exc, "message", None) or f"{type(exc).__name__}: {exc}"
1426 return {"code": code, "message": message}
1427 return {
1428 "code": _TOOL_FAILURE_ERROR_CODE,
1429 "message": f"{type(exc).__name__}: {exc}",
1430 }
1433def _tool_failure_result(step_id: str, failure: str | dict) -> dict:
1434 """Työkaluvika → kova virhe (HARD_ERROR_CODES).
1436 A dict failure carries the explicit hard code (CALL_NOT_ALLOWED /
1437 CALL_INVALID_PARAMS / CALL_FAILED); a bare string stays the legacy
1438 technical TOOL_FAILURE. Odotetut domain-virheet (not found,
1439 path-traversal, decode, malformed file-tool args) palaavat mallille
1440 näkyvinä työkalutuloksina eivätkä kulje tästä (ADR-0012, ADR-0030 D8).
1441 """
1442 if isinstance(failure, dict):
1443 code = failure.get("code") or _TOOL_FAILURE_ERROR_CODE
1444 message = failure.get("message") or ""
1445 else:
1446 code = _TOOL_FAILURE_ERROR_CODE
1447 message = failure
1448 return {
1449 "error": (
1450 f"{code}: step '{step_id}' tool execution failed: {message}"
1451 ),
1452 "details": {
1453 "error_code": code,
1454 "step_id": step_id,
1455 },
1456 }
1459def _with_failure_flags(value: dict) -> dict:
1460 """Tag an LLM-failure step result with the keys strict condition gates read.
1462 A failing step result carries `escalate`/`verdict` so a strict gate like
1463 `{{ research.escalate }}` resolves instead of raising an unresolved-
1464 reference ValueError, and the failure routes to the escalation/qb surface.
1465 Returns a new dict; the caller's `error`/`raw`/`parse_error` are preserved.
1466 """
1467 return {**value, "escalate": True, "verdict": "ERROR"}
1470def _with_context_metrics(
1471 state: dict,
1472 step_id: str,
1473 step_value: Any,
1474 measurement: dict[str, int],
1475 tool_meta: dict[str, Any] | None = None,
1476) -> dict:
1477 """Kirjoita step-tulos state[step_id]-polkuun sellaisenaan — sama arvo,
1478 polku ja tyyppi kuin ennen Phase 5:tä, ei mittausavaimia sisällä (Phase
1479 5b) — ja kontekstimittaus rinnakkaiseen step_context_metrics-envelope-
1480 avaimen alle (step_id → {context_size_tokens, context_budget_tokens}).
1482 Työkalusilmukan metadata kirjoitetaan samalla rinnakkaiseen step_tool_meta-
1483 avaimeen (step_id → tool-meta-dict), jotta state[step_id] pysyy täsmälleen
1484 output-envelopena eikä saa ylimääräistä top-level-avainta (ADR-0023 D2/D9).
1485 Aiemmat step_tool_meta-kirjoitukset säilytetään."""
1486 metrics = dict(state.get("step_context_metrics") or {})
1487 metrics[step_id] = measurement
1488 result = {**state, step_id: step_value, "step_context_metrics": metrics}
1489 if tool_meta:
1490 tool_metas = dict(state.get("step_tool_meta") or {})
1491 tool_metas[step_id] = tool_meta
1492 result["step_tool_meta"] = tool_metas
1493 return result
1496def _format_user_message(input_value: Any, params: dict[str, Any]) -> str:
1497 """
1498 Muotoile agentin user-message resolvoidusta inputista + paramseista.
1499 """
1500 parts: list[str] = []
1502 # Inputin esitysmuoto: yritetään olla informatiivisia
1503 if input_value is None:
1504 pass
1505 elif isinstance(input_value, str):
1506 parts.append(input_value)
1507 elif isinstance(input_value, dict):
1508 # Doc-file form: {path, content, ...} → human-readable
1509 if "path" in input_value and "content" in input_value:
1510 path = input_value.get("path", "")
1511 content = input_value.get("content", "")
1512 parts.append(f"Path: {path}\n\n---\n{content}")
1513 else:
1514 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
1515 elif isinstance(input_value, list):
1516 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
1517 else:
1518 parts.append(str(input_value))
1520 # Lisää muut paramit (focus, finding jne.) jos eivät jo ole system_promptissa
1521 extra: dict[str, Any] = {}
1522 skip_keys = {"skills"}
1523 for k, v in params.items():
1524 if k in skip_keys:
1525 continue
1526 # Älä toista pieniä skalaareja - ne yleensä jo system_promptissa
1527 if isinstance(v, (dict, list)):
1528 extra[k] = v
1529 if extra:
1530 parts.append("## Additional context\n\n" + json.dumps(extra, ensure_ascii=False, indent=2))
1532 return "\n\n".join(p for p in parts if p)