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