Coverage for core/langgraph/_llm.py: 86%
653 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-20 02:14 +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
31from dataclasses import dataclass, field
32from pathlib import Path
33from typing import Any
35log = logging.getLogger(__name__)
37# Workspace-root resolver for the file-access tool — sama dual-import-kuvio
38# kuin runner.py:ssa: flat sys.path (testit, discover-wrapperit) vs. qualified
39# langgraph.* (main.py, PYTHONPATH=/app).
40try:
41 from _files_output import resolve_workspace_root
42except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
43 from langgraph._files_output import resolve_workspace_root
45# Shared structural-skip set for the list_dir tool (ADR-0012). Imported from
46# _builtins to avoid duplicating the 8-entry set; _builtins does not import
47# _llm (only stdlib + exec_runner + _files_output), so this is cycle-safe.
48try:
49 from _builtins import _DEFAULT_SKIP_DIRS
50except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
51 from langgraph._builtins import _DEFAULT_SKIP_DIRS
53# Deterministic communication-context block (Phase 3): sama dual-import-kuvio
54# kuin yllä — flat sys.path (testit) vs. qualified langgraph.* (main.py).
55try:
56 from _data_structure import (
57 COMMUNICATION_CONTEXT_KEY,
58 render_communication_block,
59 )
60except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
61 from langgraph._data_structure import (
62 COMMUNICATION_CONTEXT_KEY,
63 render_communication_block,
64 )
66# Static agent output envelope contract (Phase 2): same dual-import pattern as
67# above — flat sys.path (tests) vs. qualified langgraph.* (main.py).
68try:
69 from _envelope import (
70 ENVELOPE_CONTRACT_BLOCK,
71 ENVELOPE_ERROR_CODE,
72 validate_envelope,
73 )
74except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
75 from langgraph._envelope import (
76 ENVELOPE_CONTRACT_BLOCK,
77 ENVELOPE_ERROR_CODE,
78 validate_envelope,
79 )
81# Usage-instruction block for a step's `allow` whitelist (Phase 5): same
82# dual-import pattern as above — flat sys.path (tests) vs. qualified
83# langgraph.* (main.py).
84try:
85 from _instructions import (
86 USAGE_INSTRUCTIONS_KEY,
87 render_usage_instructions_block,
88 )
89except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
90 from langgraph._instructions import (
91 USAGE_INSTRUCTIONS_KEY,
92 render_usage_instructions_block,
93 )
95SKILL_DIR = Path("/skills")
98class SkillLoadError(ValueError):
99 """Skill-latauksen kova virhe (ADR-0012) — ei hiljaista fallbackia.
101 Erottaa skill-resoluution virheet muista ValueErrorista: runnerin
102 parallel-käsittelijät re-raisaavat tämän (kova ajo-failure kaikilla
103 step-polkuilla), kun muut step-virheet pysyvät state-virheinä.
104 """
106# Kontekstin eheys -sopimus (Phase 4 doc, ## Kontekstin eheys):
107# koottu konteksti toimitetaan mallikutsulle kokonaisuudessaan tai step
108# epäonnistuu kovasti — ei hiljaista leikkausta missään muodossa.
109DEFAULT_CONTEXT_BUDGET_TOKENS = 100_000
110DEFAULT_MAX_PARALLEL = 500
111_CONTEXT_BUDGET_ERROR_CODE = "VALIDATION_ERROR"
112_TOOL_FAILURE_ERROR_CODE = "TOOL_FAILURE"
113_TIKTOKEN_ENCODING = "cl100k_base"
115# --- Opt-in file-access tool (agent spec.file_access: true) ---
116# One bounded read_file tool + a tool-execution loop. A tool round = one
117# additional inference call consumed through the run budget.
118_READ_FILE_TOOL_NAME = "read_file"
119_LIST_DIR_TOOL_NAME = "list_dir"
120_MAX_TOOL_ROUNDS = 3
121_TOOL_STOP_REASON_CAP = "max_tool_rounds"
122_TOOL_STOP_REASON_BUDGET = "budget"
123_READ_FILE_MAX_BYTES = 200_000
124_READ_FILE_MAX_LINES = 2000
125_LIST_DIR_MAX_ENTRIES = 512
126_TIKTOKEN_CACHE_DIR = Path(__file__).parent / "tiktoken_cache"
128_LANGFUSE_FLUSH_EVERY = 8
131# ---------------------------------------------------------------------------
132# RunBudget — deterministinen pipeline-ajon suoritusbudjetti
133# ---------------------------------------------------------------------------
136@dataclass
137class RunBudget:
138 max_llm_calls: int = 48
139 run_timeout_seconds: int = 240
140 start: float = field(default_factory=time.monotonic)
141 llm_calls: int = 0
142 stopped: bool = False
143 stop_reason: str = ""
144 _lock: threading.Lock = field(default_factory=threading.Lock)
146 def try_consume(self) -> bool:
147 with self._lock:
148 if self.stopped:
149 return False
150 if time.monotonic() - self.start > self.run_timeout_seconds:
151 self.stopped = True
152 self.stop_reason = "run_timeout"
153 return False
154 if self.llm_calls >= self.max_llm_calls:
155 self.stopped = True
156 self.stop_reason = "max_llm_calls"
157 return False
158 self.llm_calls += 1
159 return True
161 def as_stats(self) -> dict[str, Any]:
162 return {
163 "llm_calls": self.llm_calls,
164 "budget_stopped": self.stopped,
165 "budget_stop_reason": self.stop_reason or None,
166 }
169# ---------------------------------------------------------------------------
170# Skill loading
171# ---------------------------------------------------------------------------
174def _skill_repos_from_env() -> list[dict]:
175 """Lue SKILL_REPOS-env (JSON [{"name": ..., "url": ...}]) → lista entryistä.
177 Puuttuva env → [] (sopimus: SKILL_REPOS on valinnainen). Mikä tahansa
178 epäkelpo arvo on kova virhe (ADR-0012): malformi JSON, ei-lista, tai
179 entry ilman ei-tyhjää name- ja url-merkkijonoa → SkillLoadError.
180 """
181 raw = os.getenv("SKILL_REPOS")
182 if not raw:
183 return []
184 try:
185 repos = json.loads(raw)
186 except json.JSONDecodeError as exc:
187 raise SkillLoadError(f"SKILL_REPOS is not valid JSON: {exc}") from exc
188 if not isinstance(repos, list):
189 raise SkillLoadError(f"SKILL_REPOS must be a JSON list, got {type(repos).__name__}")
190 for idx, repo in enumerate(repos):
191 if not isinstance(repo, dict):
192 raise SkillLoadError(
193 f"SKILL_REPOS entry {idx} must be an object, got {type(repo).__name__}"
194 )
195 name = repo.get("name")
196 url = repo.get("url")
197 if not isinstance(name, str) or not name.strip():
198 raise SkillLoadError(f"SKILL_REPOS entry {idx} is missing a non-empty string 'name'")
199 if not isinstance(url, str) or not url.strip():
200 raise SkillLoadError(
201 f"SKILL_REPOS entry {idx} ('{name}') is missing a non-empty string 'url'"
202 )
203 return repos
206def locate_skills_root(repo_mount: Path) -> Path:
207 """Etsi skill-repo-mountista tasan yksi `skills`-niminen kansio (find).
209 Ajaa hardcodetun find-komennon (ei parametrisoitava):
211 find <mount> -mindepth 1 -type d -name .git -prune \\
212 -o -type d -name skills -print
214 Etsii pelkät `skills`-nimiset kansiot (ei tiedostoja, ei sisäkansioita),
215 `.git`-puu poistetaan (prune), osumat lajitellaan (find-järjestys ei ole
216 taattu → determinismi). Tarkalleen 1 osuma → palauttaa polun Path-
217 objektina. 0 tai >1 osumaa on kova virhe (ADR-0012), kuten myös itse
218 komennon epäonnistuminen (find puuttuu, returncode != 0) — ei hiljaista
219 ohitusta.
220 """
221 cmd = [
222 "find",
223 str(repo_mount),
224 "-mindepth",
225 "1",
226 "-type",
227 "d",
228 "-name",
229 ".git",
230 "-prune",
231 "-o",
232 "-type",
233 "d",
234 "-name",
235 "skills",
236 "-print",
237 ]
238 try:
239 proc = subprocess.run(cmd, capture_output=True, text=True)
240 except OSError as exc:
241 raise SkillLoadError(
242 f"failed to run find for skill mount {repo_mount}: {exc}"
243 ) from exc
244 if proc.returncode != 0:
245 raise SkillLoadError(
246 f"find failed for skill mount {repo_mount} "
247 f"(rc={proc.returncode}): {proc.stderr.strip()}"
248 )
249 hits = sorted(line.strip() for line in proc.stdout.splitlines() if line.strip())
250 if not hits:
251 raise SkillLoadError(f"no skills folder found in {repo_mount}")
252 if len(hits) > 1:
253 raise SkillLoadError(
254 f"ambiguous skills folders found in {repo_mount}: " + ", ".join(hits)
255 )
256 return Path(hits[0])
259def _normalize_skill_libs() -> list[Path]:
260 """Skill-lähdelista: koodirepon skills/-kirjasto ensin, sitten skill-repot.
262 Koodirepon implisiittinen skills/-kirjasto (`PROJECT_ROOT/skills/`,
263 oletus `/project/skills`) on aina korkein prioriteetti skill-lähteenä
264 (koodirepon ylivalta, ADR-0018 D3), ja skill-repot tulevat SKILL_REPOS-
265 järjestyksessä (ADR-0018 D4) listan loppuun: kunkin mountin kohdalla
266 etsitään `skills`-kansio locate_skills_root()-tuloksella. Jos skill-
267 reposta ei löydy skills-kansiota, ajo epäonnistuu kovasti (ADR-0012) —
268 koodirepon puuttuvan skills/-kansion hiljainen ohitus pysyy ennallaan
269 (resoluutio ohittaa puuttuvan kirjaston).
270 """
271 project_root = Path(os.getenv("PROJECT_ROOT", "/project"))
272 return [project_root / "skills"] + [
273 locate_skills_root(SKILL_DIR / repo["name"]) for repo in _skill_repos_from_env()
274 ]
277def load_skills(spec: dict, *, max_chars: int = 16000) -> str:
278 """
279 Lukee `spec.skills`-flat-listan (entryjä) ja palauttaa yhdistetyn
280 tekstin. Kunkin tiedoston yläraja on max_chars merkkiä (oletus
281 16 000): ylittävä tiedosto on kova SkillLoadError (ADR-0012) —
282 ei hiljaista katkaisua.
284 Entry on joko kansionimi (`<nimi>`): koko `<nimi>/`-kansion sisältö
285 injektoidaan otsakkeilla (SKILL.md ensin, sitten loput
286 aakkosjärjestyksessä suhteellisen polun mukaan), tai tarkka
287 tiedosto (`<skill>/<tiedostonimi>`): vain kyseinen tiedosto
288 injektoidaan. Tiedosto-entryt eivät saa sisältää alipolkuja —
289 täsmälleen yksi `/`; virheellinen muoto (johto-/loppuviiva, tyhjä
290 segmentti, `..`, absoluuttinen polku) on kova SkillLoadError.
292 Ensimmäinen skill-lähde, josta kohde löytyy, voittaa (first-match):
293 koodirepon skills/-kirjasto (`PROJECT_ROOT/skills/`, koodirepon
294 ylivalta, ADR-0018 D3) ensin, sitten skill-repot SKILL_REPOS-
295 järjestyksessä (prioriteetti, ADR-0018 D4); puuttuvat lähteet
296 ohitetaan hiljaa.
298 Resolvoitumaton entry → SkillLoadError (ADR-0012): viesti listaa
299 etsityt kirjastopolut ja saatavilla olevat skillit. Ei hiljaista
300 ohitusta.
301 """
302 skills = (spec.get("spec") or {}).get("skills") or []
303 if not skills:
304 return ""
305 libs = _normalize_skill_libs()
306 parts: list[str] = []
307 for entry in skills:
308 parts.extend(_load_bare_skill(entry, libs, max_chars=max_chars))
309 return "\n\n---\n\n".join(parts)
312def _parse_skill_entry(entry: str) -> tuple[str, str | None]:
313 """Parseroi spec.skills-entryn: kansionimi → (nimi, None), tarkka
314 tiedosto `<skill>/<tiedostonimi>` → (skill, tiedostonimi).
316 Virheellinen muoto → kova SkillLoadError: tyhjä entry, `.`/`..`,
317 tyhjä segmentti, absoluuttinen polku tai >1 vinoviivaa (alipolut
318 eivät ole sallittuja).
319 """
320 if not entry.strip():
321 raise SkillLoadError(
322 f"skill entry '{entry}' is empty; use '<skill>' "
323 "or '<skill>/<file>'"
324 )
325 if entry.startswith("/"):
326 raise SkillLoadError(
327 f"skill entry '{entry}' is an absolute path; use '<skill>' "
328 "or '<skill>/<file>'"
329 )
330 parts = entry.split("/")
331 if any(seg in ("", ".", "..") for seg in parts):
332 raise SkillLoadError(
333 f"skill entry '{entry}' contains an invalid path segment "
334 "(''|'.'|'..'); use '<skill>' or '<skill>/<file>'"
335 )
336 if len(parts) == 1:
337 return entry, None
338 if len(parts) > 2:
339 raise SkillLoadError(
340 f"skill entry '{entry}' has too many path segments; subpaths "
341 "are not allowed, use '<skill>' or '<skill>/<file>'"
342 )
343 skill, filename = parts
344 return skill, filename
347def _load_bare_skill(entry: str, libs: list[Path], *, max_chars: int) -> list[str]:
348 """Skill-entry (kansionimi tai tarkka tiedosto) ladattuna otsakkeilla.
350 Kansionimi (`<nimi>`): kansion `lib / <nimi>/` kaikki tiedostot
351 injektoidaan (SKILL.md ensin, sitten loput aakkosjärjestyksessä).
352 Tarkka tiedosto (`<skill>/<tiedostonimi>`): vain kyseinen tiedosto
353 `lib / <skill> / <tiedosto>` injektoidaan otsakkeella
354 `## Skill: <skill> — <tiedosto>`.
356 Ensimmäinen skill-lähde, josta kohde löytyy, voittaa (first-match).
357 Kansionimelle vastaava `lib / entry` tiedostona → SkillLoadError;
358 tiedosto-entrylle kansion löytyminen ilman kohdetiedostoa ohittaa
359 kyseisen lähteen hiljaa. Virheellinen entry-muoto tai kohde, jota
360 ei löydy mistään skill-lähteestä → SkillLoadError.
361 """
362 name, filename = _parse_skill_entry(entry)
363 for lib in libs:
364 folder = lib / name
365 if filename is None:
366 if folder.is_dir():
367 parts: list[str] = []
368 for path in _skill_folder_files(folder):
369 relpath = path.relative_to(folder)
370 parts.append(
371 f"## Skill: {name} — {relpath}\n\n"
372 f"{_read_skill_file(entry, path, max_chars=max_chars)}"
373 )
374 return parts
375 if folder.is_file():
376 raise SkillLoadError(
377 f"skill '{entry}' resolves to a file, expected a skill folder "
378 f"(SKILL.md etc.): {folder}"
379 )
380 else:
381 file_path = folder / filename
382 if file_path.is_file():
383 return [
384 f"## Skill: {name} — {filename}\n\n"
385 f"{_read_skill_file(entry, file_path, max_chars=max_chars)}"
386 ]
387 if folder.is_dir():
388 continue
389 raise SkillLoadError(_unresolved_skill_error(entry, libs))
392def _skill_folder_files(folder: Path) -> list[Path]:
393 """Kansion kaikki tiedostot rekursiivisesti, deterministinen järjestys:
394 juuren SKILL.md ensin, sitten loput suhteellisen polun mukaan
395 aakkosjärjestyksessä."""
396 files = [p for p in folder.rglob("*") if p.is_file()]
397 files.sort(
398 key=lambda p: (
399 p.relative_to(folder) != Path("SKILL.md"),
400 str(p.relative_to(folder)),
401 )
402 )
403 return files
406def _read_skill_file(entry: str, path: Path, *, max_chars: int) -> str:
407 try:
408 text = path.read_text(encoding="utf-8", errors="replace")
409 except OSError as exc:
410 raise SkillLoadError(
411 f"skill entry '{entry}' file unreadable: {path} ({exc})"
412 ) from exc
413 if len(text) > max_chars:
414 raise SkillLoadError(
415 f"skill entry '{entry}' file exceeds the {max_chars}-char limit: "
416 f"{path} ({len(text)} chars)"
417 )
418 return text
421def _available_skill_names(libs: list[Path]) -> list[str]:
422 """Kirjastojen tarjonta: skill-kansiot (sisältävät SKILL.md) + top-tason
423 *.md-tiedostot. Sorted + deduped."""
424 names: set[str] = set()
425 for lib in libs:
426 if not lib.is_dir():
427 continue
428 for child in lib.iterdir():
429 if child.is_dir():
430 if (child / "SKILL.md").is_file():
431 names.add(child.name)
432 elif child.is_file() and child.name.endswith(".md"):
433 names.add(child.name)
434 return sorted(names)
437def _unresolved_skill_error(entry: str, libs: list[Path]) -> str:
438 available = ", ".join(_available_skill_names(libs)) or "(none)"
439 searched = "\n".join(f" - {lib}" for lib in libs)
440 return (
441 f"skill '{entry}' not found in any skill repo "
442 f"(available skills: {available}). Searched skill repos:\n{searched}"
443 )
446# ---------------------------------------------------------------------------
447# Template substitution (simple {{ key }} replacement against a dict)
448# ---------------------------------------------------------------------------
450_TMPL_RE = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_.]*)\s*\}\}")
453def _render(text: str, ctx: dict[str, Any]) -> str:
454 if not isinstance(text, str) or "{{" not in text:
455 return text
457 def _lookup(path: str) -> str:
458 cur: Any = ctx
459 for part in path.split("."):
460 if isinstance(cur, dict):
461 cur = cur.get(part)
462 else:
463 return ""
464 if cur is None:
465 return ""
466 return str(cur) if not isinstance(cur, str) else cur
468 return _TMPL_RE.sub(lambda m: _lookup(m.group(1)), text)
471# ---------------------------------------------------------------------------
472# Token counting (deterministinen ja provider-agnostinen; Phase 4 doc)
473# ---------------------------------------------------------------------------
475_tokenizer_encoding: Any | None = None
476_tokenizer_lock = threading.Lock()
479def _tokenizer() -> Any:
480 """tiktoken-encoding lazy-singleton, joka käyttää bundlattua
481 TIKTOKEN_CACHE_DIR:ää (core/langgraph/tiktoken_cache/) — ei
482 verkkolatausta ensimmäiselläkään käyttökerralla."""
483 global _tokenizer_encoding
484 if _tokenizer_encoding is None:
485 with _tokenizer_lock:
486 if _tokenizer_encoding is None:
487 if not os.environ.get("TIKTOKEN_CACHE_DIR"):
488 os.environ["TIKTOKEN_CACHE_DIR"] = str(_TIKTOKEN_CACHE_DIR)
489 from tiktoken import get_encoding
491 _tokenizer_encoding = get_encoding(_TIKTOKEN_ENCODING)
492 return _tokenizer_encoding
495def count_context_tokens(messages: list[dict[str, str]]) -> int:
496 """Kootun kontekstin koko tokeneina: system + user -viestien sisältöjen
497 summa. Deterministinen, ei riipu providerista eikä kutsutusta mallista."""
498 enc = _tokenizer()
499 total = 0
500 for message in messages:
501 content = message.get("content")
502 if isinstance(content, str):
503 total += len(enc.encode_ordinary(content))
504 return total
507# ---------------------------------------------------------------------------
508# Langfuse trace helpers
509# ---------------------------------------------------------------------------
511_PARENT_STACK: contextvars.ContextVar[tuple[str, ...]] = contextvars.ContextVar(
512 "langfuse_parent_stack", default=()
513)
516class LangfuseTracer:
517 """
518 Per-ajo Langfuse-tracen alustus + generation-spanit per LLM-kutsu.
519 Käyttää singleton-Langfuse-clientiä env-muuttujista.
520 """
522 def __init__(self, project_slug: str, pipeline_name: str, run_id: str):
523 self.project_slug = project_slug
524 self.pipeline_name = pipeline_name
525 self.run_id = run_id or "unknown"
526 self._lf: Any = None
527 self._trace_ctx: dict[str, Any] | None = None
528 self._flush_counter = 0
529 self._init_done = False
531 def _client(self) -> Any:
532 if self._lf is not None:
533 return self._lf
534 try:
535 from langfuse import Langfuse # type: ignore
536 except ImportError:
537 log.warning("[langfuse] package not installed — tracing disabled")
538 return None
539 prev = os.environ.get("LANGFUSE_TRACING_ENABLED")
540 os.environ["LANGFUSE_TRACING_ENABLED"] = "true"
541 try:
542 self._lf = Langfuse()
543 except Exception as exc: # noqa: BLE001
544 log.warning("[langfuse] init failed: %s — tracing disabled", exc)
545 self._lf = None
546 finally:
547 if prev is None:
548 os.environ["LANGFUSE_TRACING_ENABLED"] = "false"
549 else:
550 os.environ["LANGFUSE_TRACING_ENABLED"] = prev
551 return self._lf
553 @property
554 def trace_name(self) -> str:
555 return f"{self.project_slug}:{self.pipeline_name}"
557 @property
558 def session_id(self) -> str:
559 return f"{self.project_slug}-{self.pipeline_name}"
561 def _ctx(self) -> dict[str, Any] | None:
562 lf = self._client()
563 if lf is None:
564 return None
565 if self._trace_ctx is None:
566 self._trace_ctx = {"trace_id": lf.create_trace_id(seed=self.run_id)}
567 return self._trace_ctx
569 def _ctx_with_parent(self) -> dict[str, Any] | None:
570 ctx = self._ctx()
571 if ctx is None:
572 return None
573 stack = _PARENT_STACK.get()
574 if stack:
575 return {**ctx, "parent_span_id": stack[-1]}
576 return ctx
578 def open_parent_observation(
579 self,
580 *,
581 name: str,
582 input: dict[str, Any] | None = None,
583 metadata: dict[str, Any] | None = None,
584 ) -> Any:
585 lf = self._client()
586 ctx = self._ctx_with_parent()
587 if lf is None or ctx is None:
588 return _NoopObservation()
589 try:
590 obs = lf.start_observation(
591 trace_context=ctx,
592 name=name,
593 as_type="span",
594 input=input or {},
595 metadata=metadata or {},
596 )
597 except Exception as exc: # noqa: BLE001
598 log.warning("[langfuse] open_parent_observation failed: %s", exc)
599 return _NoopObservation()
600 stack = _PARENT_STACK.get()
601 _PARENT_STACK.set((*stack, obs.id))
602 return obs
604 def close_parent_observation(self, obs: Any) -> None:
605 if isinstance(obs, _NoopObservation):
606 return
607 stack = _PARENT_STACK.get()
608 if stack and stack[-1] == obs.id:
609 _PARENT_STACK.set(stack[:-1])
610 obs.end()
612 def init_trace(self) -> None:
613 if self._init_done:
614 return
615 lf = self._client()
616 ctx = self._ctx()
617 if lf is None or ctx is None:
618 self._init_done = True
619 return
620 try:
621 obs = lf.start_observation(
622 trace_context=ctx,
623 name=self.trace_name,
624 as_type="span",
625 input={
626 "pipeline": self.pipeline_name,
627 "project": self.project_slug,
628 "run_id": self.run_id,
629 },
630 )
631 self._apply_meta(obs)
632 obs.end()
633 self.flush(force=True)
634 except Exception as exc: # noqa: BLE001
635 log.warning("[langfuse] init_trace failed: %s", exc)
636 self._init_done = True
638 def _apply_meta(self, obs: Any) -> None:
639 try:
640 obs.update_trace(
641 name=self.trace_name,
642 user_id=self.project_slug,
643 session_id=self.session_id,
644 tags=[f"pipeline:{self.pipeline_name}"],
645 )
646 except Exception: # noqa: BLE001
647 pass
649 def generation(
650 self,
651 name: str,
652 *,
653 model: str,
654 temperature: float,
655 messages: list[dict[str, str]],
656 metadata: dict[str, Any] | None = None,
657 ) -> Any:
658 lf = self._client()
659 ctx = self._ctx()
660 if lf is None or ctx is None:
661 return _NoopObservation()
662 try:
663 obs = lf.start_observation(
664 trace_context=self._ctx_with_parent(),
665 name=name,
666 as_type="generation",
667 model=model,
668 model_parameters={"temperature": temperature},
669 input=messages,
670 metadata=metadata,
671 )
672 self._apply_meta(obs)
673 return obs
674 except Exception as exc: # noqa: BLE001
675 log.warning("[langfuse] generation failed: %s", exc)
676 return _NoopObservation()
678 def flush(self, *, force: bool = False) -> None:
679 lf = self._client()
680 if lf is None:
681 return
682 self._flush_counter += 1
683 if force or self._flush_counter % _LANGFUSE_FLUSH_EVERY == 0:
684 try:
685 lf.flush()
686 except Exception: # noqa: BLE001
687 pass
690class _NoopObservation:
691 def update(self, **_: Any) -> None: ...
692 def end(self) -> None: ...
695# ---------------------------------------------------------------------------
696# JSON output parsing — LLM-vastauksen JSON-rakenteiden jäsennys
697# ---------------------------------------------------------------------------
700def _try_parse_json(text: str) -> tuple[Any, str | None]:
701 """Parse the response as strict JSON. Palauttaa (data, error).
703 No fence stripping, no embedded-object extraction, and no repair/retry:
704 the whole stripped text must decode. data=None on failure.
705 """
706 if not text or not text.strip():
707 return None, "empty"
708 try:
709 return json.loads(text.strip()), None
710 except json.JSONDecodeError as e:
711 return None, str(e)
714# ---------------------------------------------------------------------------
715# File-access tool (agent spec.file_access: true)
716# ---------------------------------------------------------------------------
719@dataclass
720class ReadFileResult:
721 content: str
722 truncated: bool
723 error: str | None
725 def as_tool_message(self) -> str:
726 payload: dict[str, Any] = {
727 "name": _READ_FILE_TOOL_NAME,
728 "ok": self.error is None,
729 "content": self.content,
730 "truncated": self.truncated,
731 }
732 if self.error is not None:
733 payload["error"] = self.error
734 return json.dumps(payload, ensure_ascii=False)
737_READ_FILE_TOOL: dict[str, Any] = {
738 "type": "function",
739 "function": {
740 "name": _READ_FILE_TOOL_NAME,
741 "description": (
742 "Read a text file from the project root. The path is resolved "
743 "against the project root; any read that resolves outside the "
744 "project root is rejected. Content is truncated and flagged when "
745 "it exceeds the caps, and can be windowed by offset/limit (lines)."
746 ),
747 "parameters": {
748 "type": "object",
749 "properties": {
750 "path": {
751 "type": "string",
752 "description": "Path to the file, relative to the project root.",
753 },
754 "offset": {
755 "type": "integer",
756 "description": "Number of lines to skip (0-based). Optional.",
757 },
758 "limit": {
759 "type": "integer",
760 "description": "Max number of lines to return. Optional.",
761 },
762 },
763 "required": ["path"],
764 "additionalProperties": False,
765 },
766 },
767}
770def _is_within(root: Path, candidate: Path) -> bool:
771 return candidate == root or root in candidate.parents
774def _read_file(
775 project_root: Path,
776 path: str,
777 offset: int | None,
778 limit: int | None,
779) -> ReadFileResult:
780 """Realpath-bound read of a text file inside the project root (ADR-0012).
782 Every rejected read returns a clear error — no silent fallback. Truncation
783 over `_READ_FILE_MAX_BYTES` / `_READ_FILE_MAX_LINES` is explicit via
784 `truncated`.
785 """
786 if not path or not str(path).strip():
787 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: 'path' is required (ADR-0012)")
788 try:
789 root = project_root.resolve(strict=False)
790 except OSError as exc:
791 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot resolve project root: {exc}")
792 try:
793 candidate = (root / path).resolve(strict=False)
794 except OSError as exc:
795 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot resolve '{path}': {exc}")
796 if not _is_within(root, candidate):
797 return ReadFileResult(
798 "",
799 False,
800 f"{_READ_FILE_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
801 )
802 if not candidate.is_file():
803 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: '{path}' is not a file or does not exist")
804 try:
805 text = candidate.read_text(encoding="utf-8")
806 except (OSError, UnicodeDecodeError) as exc:
807 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot read '{path}': {exc}")
809 lines = text.splitlines()
810 total_lines = len(lines)
811 try:
812 start = int(offset) if offset is not None and int(offset) > 0 else 0
813 except (TypeError, ValueError):
814 start = 0
815 if limit is not None:
816 # A requested window is not an engine cap — no line-level truncation flag.
817 try:
818 max_lines = int(limit)
819 except (TypeError, ValueError):
820 max_lines = _READ_FILE_MAX_LINES
821 truncated = False
822 end = start + max_lines if max_lines >= 0 else total_lines
823 else:
824 max_lines = _READ_FILE_MAX_LINES
825 truncated = (start + max_lines) < total_lines
826 end = start + max_lines
827 content = "\n".join(lines[start:end])
829 if len(content.encode("utf-8")) > _READ_FILE_MAX_BYTES:
830 content = content[:_READ_FILE_MAX_BYTES]
831 truncated = True
832 return ReadFileResult(content, truncated, None)
835@dataclass
836class ListDirEntry:
837 path: str
838 name: str
839 size: int | None
840 kind: str # "dir" | "file"
843@dataclass
844class ListDirResult:
845 entries: list[ListDirEntry]
846 truncated: bool
847 root: str
848 error: str | None
850 def as_tool_message(self) -> str:
851 payload: dict[str, Any] = {
852 "name": _LIST_DIR_TOOL_NAME,
853 "ok": self.error is None,
854 "entries": [
855 {"path": e.path, "name": e.name, "size": e.size, "kind": e.kind}
856 for e in self.entries
857 ],
858 "truncated": self.truncated,
859 "root": self.root,
860 }
861 if self.error is not None:
862 payload["error"] = self.error
863 return json.dumps(payload, ensure_ascii=False)
866_LIST_DIR_TOOL: dict[str, Any] = {
867 "type": "function",
868 "function": {
869 "name": _LIST_DIR_TOOL_NAME,
870 "description": (
871 "List ONE level of a directory inside the project root. The path is "
872 "resolved against the project root; any listing that resolves "
873 "outside the project root is rejected. Non-recursive: nested "
874 "subdirectories are not expanded. Only structural skip dirs are "
875 "omitted. The listing is capped and flagged when it exceeds the cap."
876 ),
877 "parameters": {
878 "type": "object",
879 "properties": {
880 "path": {
881 "type": "string",
882 "description": "Directory path, relative to the project root.",
883 },
884 },
885 "required": ["path"],
886 "additionalProperties": False,
887 },
888 },
889}
892def _execute_list_dir_tool(project_root: Path, path: str) -> ListDirResult:
893 """Realpath-bound ONE-level directory listing inside the project root
894 (ADR-0012). Every rejected listing returns a clear error — no silent
895 fallback. Only structural skip dirs are omitted; entries over
896 `_LIST_DIR_MAX_ENTRIES` are capped and flagged via `truncated`.
897 """
898 if not path or not str(path).strip():
899 return ListDirResult(
900 [], False, "", f"{_LIST_DIR_TOOL_NAME}: 'path' is required (ADR-0012)"
901 )
902 try:
903 root = project_root.resolve(strict=False)
904 except OSError as exc:
905 return ListDirResult(
906 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot resolve project root: {exc}"
907 )
908 try:
909 candidate = (root / str(path)).resolve(strict=False)
910 except OSError as exc:
911 return ListDirResult(
912 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot resolve '{path}': {exc}"
913 )
914 if not _is_within(root, candidate):
915 return ListDirResult(
916 [],
917 False,
918 "",
919 f"{_LIST_DIR_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
920 )
921 if not candidate.is_dir():
922 return ListDirResult(
923 [],
924 False,
925 "",
926 f"{_LIST_DIR_TOOL_NAME}: '{path}' is not a directory or does not exist",
927 )
928 try:
929 children = sorted(candidate.iterdir(), key=lambda p: p.name)
930 except OSError as exc:
931 return ListDirResult(
932 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot list '{path}': {exc}"
933 )
935 entries: list[ListDirEntry] = []
936 for child in children:
937 if child.name in _DEFAULT_SKIP_DIRS:
938 continue
939 is_dir = child.is_dir()
940 size: int | None = None
941 if not is_dir:
942 try:
943 size = child.stat().st_size
944 except OSError:
945 size = None
946 entries.append(
947 ListDirEntry(
948 path=str(child.relative_to(root)),
949 name=child.name,
950 size=size,
951 kind="dir" if is_dir else "file",
952 )
953 )
954 truncated = len(entries) > _LIST_DIR_MAX_ENTRIES
955 if truncated:
956 entries = entries[:_LIST_DIR_MAX_ENTRIES]
957 return ListDirResult(entries, truncated, str(root), None)
960def _invalid_tool_arguments_result(
961 name: str, reason: str
962) -> ReadFileResult | ListDirResult:
963 """Malformed tool arguments (unparseable JSON, non-object JSON, wrong
964 types) are an expected domain error: a soft, model-visible tool result —
965 never the technical-failure marker (ADR-0012, ADR-0030 D8)."""
966 if name == _LIST_DIR_TOOL_NAME:
967 return ListDirResult([], False, "", f"{_LIST_DIR_TOOL_NAME}: {reason}")
968 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: {reason}")
971# ---------------------------------------------------------------------------
972# OpenAIInvoker
973# ---------------------------------------------------------------------------
976class OpenAIInvoker:
977 """
978 Default LLM invoker for agent-platform pipelines.
980 Yksi instanssi per pipeline-ajo. Thread-safe (semafori rinnakkaisuudelle).
981 """
983 def __init__(
984 self,
985 *,
986 providers: dict[str, dict[str, Any]],
987 max_parallel: int,
988 budget: RunBudget,
989 tracer: LangfuseTracer,
990 ) -> None:
991 self._providers = providers
992 self._clients: dict[str, Any] = {}
993 self.budget = budget
994 self.tracer = tracer
995 self._semaphore = threading.Semaphore(max(1, max_parallel))
997 def _openai_client(self, base_url: str, api_key: str) -> Any:
998 key = f"{base_url}|{api_key}"
999 if key not in self._clients:
1000 from openai import OpenAI # type: ignore
1001 self._clients[key] = OpenAI(base_url=base_url, api_key=api_key)
1002 return self._clients[key]
1004 # ---- Public callable -------------------------------------------------
1006 def __call__(
1007 self,
1008 resolved_spec: dict,
1009 resolved_step: dict,
1010 state: dict,
1011 ) -> dict:
1012 """
1013 StepRunner-yhteensopiva: (spec, step, state) → uusi state.
1015 `resolved_step` = step jossa params/input on JO resolvoitu
1016 StepRunnerissa template-engineä käyttäen. Tämä invoker ei enää
1017 evaluoi `{{ }}`-lausekkeita state/run_context-vastaan, vain
1018 substituoi `step.params` ja sisäänrakennetut feature-arvot
1019 agentin `system_prompt`/`user`-mallineisiin.
1020 """
1021 step_id = resolved_step.get("id", "?")
1022 agent_spec = resolved_spec.get("spec") or {}
1024 model_spec = agent_spec.get("model") or {}
1025 provider_name = model_spec.get("provider")
1027 if not provider_name:
1028 if self._providers:
1029 provider_name = next(iter(self._providers))
1030 else:
1031 raise ValueError("No LLM provider configured and no provider specified in agent spec")
1033 provider_cfg = self._providers.get(provider_name)
1034 if provider_cfg is None:
1035 raise ValueError(
1036 f"Unknown LLM provider '{provider_name}'. "
1037 f"Available: {', '.join(sorted(self._providers.keys()))}"
1038 )
1040 model_override = model_spec.get("model")
1041 model = model_override if model_override else provider_cfg["model"]
1043 # Budget check
1044 if not self.budget.try_consume():
1045 return {
1046 **state,
1047 step_id: _with_failure_flags(
1048 {"error": f"budget:{self.budget.stop_reason}"}
1049 ),
1050 }
1052 # ---- Build system + user messages --------------------------------
1053 skills_text = load_skills(resolved_spec)
1054 step_params = resolved_step.get("params") or {}
1055 input_value = resolved_step.get("input")
1057 # Render system_prompt template with params, then inject the loaded
1058 # skills after the rendered prompt (ADR-0036 D1/D6).
1059 prompt_ctx: dict[str, Any] = {
1060 k: v for k, v in step_params.items() if isinstance(k, str)
1061 }
1062 rendered_prompt = _render(
1063 str(agent_spec.get("system_prompt") or ""),
1064 prompt_ctx,
1065 ).strip()
1066 communication = resolved_step.get(COMMUNICATION_CONTEXT_KEY)
1067 communication_text = (
1068 render_communication_block(communication) if communication else ""
1069 )
1070 instructions_text = render_usage_instructions_block(
1071 resolved_step.get(USAGE_INSTRUCTIONS_KEY)
1072 )
1073 envelope_text = ENVELOPE_CONTRACT_BLOCK
1074 system_text = "\n\n".join(
1075 part
1076 for part in (
1077 rendered_prompt,
1078 skills_text,
1079 envelope_text,
1080 instructions_text,
1081 communication_text,
1082 )
1083 if part
1084 )
1086 # User message: serialise input + step params
1087 user_text = _format_user_message(input_value, step_params)
1089 # Temperature: agent spec → provider default → step params
1090 temperature = float(
1091 model_spec.get("temperature", provider_cfg.get("temperature", 0.0))
1092 )
1093 if "temperature" in step_params:
1094 try:
1095 temperature = float(step_params["temperature"])
1096 except (TypeError, ValueError):
1097 pass
1099 messages = [
1100 {"role": "system", "content": system_text},
1101 {"role": "user", "content": user_text},
1102 ]
1104 # ---- Context-integrity check (Phase 4 doc, ## Kontekstin eheys) --
1105 agent_name = (resolved_spec.get("metadata") or {}).get("name", "agent")
1106 context_budget = _effective_context_budget(resolved_spec, agent_name)
1107 context_tokens = count_context_tokens(messages)
1108 if context_tokens > context_budget:
1109 return _with_context_metrics(
1110 state,
1111 step_id,
1112 _context_overflow_result(step_id, context_tokens, context_budget),
1113 {"context_size_tokens": context_tokens, "context_budget_tokens": context_budget},
1114 )
1115 measurement = {
1116 "context_size_tokens": context_tokens,
1117 "context_budget_tokens": context_budget,
1118 }
1120 # ---- LLM call with Langfuse generation ---------------------------
1121 gen_name = f"{step_id}:{agent_name}"
1123 with self._semaphore:
1124 obs = self.tracer.generation(
1125 gen_name,
1126 model=model,
1127 temperature=temperature,
1128 messages=messages,
1129 metadata=measurement,
1130 )
1131 content = ""
1132 usage: dict[str, int] | None = None
1133 err: str | None = None
1134 tool_meta: dict[str, Any] | None = None
1135 try:
1136 if bool(agent_spec.get("file_access")):
1137 content, usage, tool_meta = self._stream_completion_with_tools(
1138 messages,
1139 temperature,
1140 model,
1141 provider_cfg["base_url"],
1142 provider_cfg["api_key"],
1143 project_root=resolve_workspace_root(state),
1144 )
1145 else:
1146 content, usage = self._stream_completion(messages, temperature, model, provider_cfg["base_url"], provider_cfg["api_key"])
1147 except Exception as exc: # noqa: BLE001
1148 err = f"{type(exc).__name__}: {exc}"
1149 log.warning("[invoke] %s LLM call failed: %s", step_id, err)
1150 try:
1151 obs.update(
1152 output=content[:8000] if len(content) > 8000 else content,
1153 usage_details=usage,
1154 )
1155 except Exception: # noqa: BLE001
1156 pass
1157 try:
1158 obs.end()
1159 except Exception: # noqa: BLE001
1160 pass
1161 self.tracer.flush()
1163 # ---- Parse output ------------------------------------------------
1164 if err:
1165 return _with_context_metrics(
1166 state,
1167 step_id,
1168 _with_failure_flags({"error": err, "raw": content}),
1169 measurement,
1170 )
1172 if tool_meta and tool_meta.get("tool_failure"):
1173 return _with_context_metrics(
1174 state,
1175 step_id,
1176 _tool_failure_result(step_id, tool_meta["tool_failure"]),
1177 measurement,
1178 )
1180 parsed, perr = _try_parse_json(content)
1181 if parsed is not None:
1182 verr = validate_envelope(parsed)
1183 if verr is not None:
1184 return _with_context_metrics(
1185 state,
1186 step_id,
1187 {
1188 "error": f"ENVELOPE_ERROR: {verr}",
1189 "details": {
1190 "error_code": ENVELOPE_ERROR_CODE,
1191 "step_id": step_id,
1192 },
1193 },
1194 measurement,
1195 )
1196 return _with_context_metrics(
1197 state, step_id, _apply_tool_meta(parsed, tool_meta), measurement
1198 )
1199 # Strict parse: a non-JSON response is an error envelope, not raw text.
1200 return _with_context_metrics(
1201 state,
1202 step_id,
1203 _apply_tool_meta(
1204 {
1205 "status": "error",
1206 "error": {"code": ENVELOPE_ERROR_CODE, "message": perr},
1207 },
1208 tool_meta,
1209 ),
1210 measurement,
1211 )
1213 # ---- Streaming completion -------------------------------------------
1215 def _stream_completion(
1216 self,
1217 messages: list[dict[str, str]],
1218 temperature: float,
1219 model: str,
1220 base_url: str,
1221 api_key: str,
1222 ) -> tuple[str, dict[str, int] | None]:
1223 client = self._openai_client(base_url, api_key)
1224 resp = client.chat.completions.create(
1225 model=model,
1226 temperature=temperature,
1227 messages=messages,
1228 stream=True,
1229 stream_options={"include_usage": True},
1230 )
1231 content = ""
1232 usage: dict[str, int] | None = None
1233 for chunk in resp:
1234 if chunk.choices and chunk.choices[0].delta.content:
1235 content += chunk.choices[0].delta.content
1236 if getattr(chunk, "usage", None):
1237 u = chunk.usage
1238 usage = {
1239 "input": u.prompt_tokens or 0,
1240 "output": u.completion_tokens or 0,
1241 "total": u.total_tokens or 0,
1242 }
1243 return content, usage
1245 def _execute_file_tool(
1246 self, tool_call: Any, project_root: Path
1247 ) -> ReadFileResult | ListDirResult:
1248 """Execute a read_file or list_dir tool call against the project root."""
1249 name = getattr(tool_call.function, "name", "")
1250 try:
1251 args = json.loads(getattr(tool_call.function, "arguments", "{}"))
1252 except (TypeError, json.JSONDecodeError) as exc:
1253 return _invalid_tool_arguments_result(name, f"invalid arguments: {exc}")
1254 if not isinstance(args, dict):
1255 return _invalid_tool_arguments_result(
1256 name,
1257 f"invalid arguments: expected a JSON object, got {type(args).__name__}",
1258 )
1259 if name == _LIST_DIR_TOOL_NAME:
1260 return _execute_list_dir_tool(project_root, args.get("path"))
1261 return _read_file(project_root, args.get("path"), args.get("offset"), args.get("limit"))
1263 # ---- Tool-looping completion (file_access: true) --------------------
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 ) -> tuple[str, dict[str, int] | None, dict[str, Any]]:
1275 """Non-streaming completion carrying the read_file/list_dir tool loop.
1277 Each tool round consumes ONE additional inference through
1278 `self.budget.try_consume`; the loop is hard-capped at `_MAX_TOOL_ROUNDS`
1279 and stops instead of looping indefinitely. A cap-stop issues exactly ONE
1280 final tool-free completion (budget-gated) so the agent can answer from
1281 the gathered context; a budget-stop never does. A technical tool
1282 failure is surfaced via `tool_meta["tool_failure"]` for the caller to
1283 turn into a hard error. The non-flagged path never reaches here.
1284 """
1285 client = self._openai_client(base_url, api_key)
1286 working = list(messages)
1287 content = ""
1288 usage: dict[str, int] | None = None
1289 tool_rounds = 0
1290 truncated = False
1291 tool_errors: list[str] = []
1292 tool_failure: str | None = None
1293 stop_reason: str | None = None
1295 while True:
1296 resp = client.chat.completions.create(
1297 model=model,
1298 temperature=temperature,
1299 messages=working,
1300 tools=[_READ_FILE_TOOL, _LIST_DIR_TOOL],
1301 stream=False,
1302 )
1303 usage = _usage_from_response(resp) or usage
1304 msg = resp.choices[0].message
1305 piece = getattr(msg, "content", None) or ""
1306 if piece:
1307 content += piece
1308 tool_calls = getattr(msg, "tool_calls", None)
1309 if not tool_calls:
1310 break
1311 working.append(
1312 {
1313 "role": "assistant",
1314 "content": piece or "",
1315 "tool_calls": [
1316 {
1317 "id": tc.id,
1318 "type": "function",
1319 "function": {"name": tc.function.name, "arguments": tc.function.arguments},
1320 }
1321 for tc in tool_calls
1322 ],
1323 }
1324 )
1325 for tc in tool_calls:
1326 try:
1327 result = self._execute_file_tool(tc, project_root)
1328 except Exception as exc: # noqa: BLE001
1329 tool_failure = f"{type(exc).__name__}: {exc}"
1330 break
1331 truncated = truncated or result.truncated
1332 if result.error:
1333 tool_errors.append(result.error)
1334 working.append(
1335 {"role": "tool", "tool_call_id": tc.id, "content": result.as_tool_message()}
1336 )
1337 if tool_failure is not None:
1338 break
1339 tool_rounds += 1
1340 if tool_rounds >= _MAX_TOOL_ROUNDS:
1341 stop_reason = _TOOL_STOP_REASON_CAP
1342 break
1343 if not self.budget.try_consume():
1344 stop_reason = _TOOL_STOP_REASON_BUDGET
1345 break
1347 final_completion = False
1348 if stop_reason == _TOOL_STOP_REASON_CAP and self.budget.try_consume():
1349 final_content, final_usage = self._final_tool_free_completion(
1350 client, working, model, temperature
1351 )
1352 content += final_content
1353 usage = final_usage or usage
1354 final_completion = True
1356 tool_meta: dict[str, Any] = {
1357 "tool_rounds": tool_rounds,
1358 "truncated": truncated,
1359 "tool_stopped": stop_reason is not None,
1360 "tool_stop_reason": stop_reason,
1361 "final_completion": final_completion,
1362 "tool_errors": tool_errors,
1363 }
1364 if tool_failure is not None:
1365 tool_meta["tool_failure"] = tool_failure
1366 return content, usage, tool_meta
1368 def _final_tool_free_completion(
1369 self,
1370 client: Any,
1371 working: list[dict[str, Any]],
1372 model: str,
1373 temperature: float,
1374 ) -> tuple[str, dict[str, int] | None]:
1375 """One final completion on the gathered conversation, no tools."""
1376 resp = client.chat.completions.create(
1377 model=model,
1378 temperature=temperature,
1379 messages=working,
1380 stream=False,
1381 )
1382 return getattr(resp.choices[0].message, "content", None) or "", _usage_from_response(resp)
1385# ---------------------------------------------------------------------------
1386# Helpers
1387# ---------------------------------------------------------------------------
1390def _usage_from_response(resp: Any) -> dict[str, int] | None:
1391 """OpenAI-vastauksen usage-dict (input/output/total) tai None, kun
1392 provider ei palauta usagea."""
1393 u = getattr(resp, "usage", None)
1394 if not u:
1395 return None
1396 return {
1397 "input": getattr(u, "prompt_tokens", 0) or 0,
1398 "output": getattr(u, "completion_tokens", 0) or 0,
1399 "total": getattr(u, "total_tokens", 0) or 0,
1400 }
1403def _effective_context_budget(resolved_spec: dict, agent_name: str) -> int:
1404 """Efektiivinen kontekstibudjetti: spec.limits.context_budget (merge-
1405 järjestyksen tulos) tai oletus 100 000 tokenia. Epäkelpo arvo on kova
1406 virhe (ADR-0012) — ei hiljaista fallbackia."""
1407 limits = (resolved_spec.get("spec") or {}).get("limits") or {}
1408 value = limits.get("context_budget")
1409 if value is None:
1410 return DEFAULT_CONTEXT_BUDGET_TOKENS
1411 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1412 raise ValueError(
1413 f"agent '{agent_name}': spec.limits.context_budget must be a "
1414 f"positive integer, got {value!r} (ADR-0012)"
1415 )
1416 return value
1419def _context_overflow_result(step_id: str, context_tokens: int, context_budget: int) -> dict:
1420 """Budjetin ylitys → kova virhe ennen mallikutsua (fail-fast)."""
1421 return {
1422 "error": (
1423 f"{_CONTEXT_BUDGET_ERROR_CODE}: step '{step_id}' assembled "
1424 f"context is {context_tokens} tokens, exceeding the context "
1425 f"budget of {context_budget} tokens"
1426 ),
1427 "details": {
1428 "error_code": _CONTEXT_BUDGET_ERROR_CODE,
1429 "step_id": step_id,
1430 "assembled_tokens": context_tokens,
1431 "context_budget": context_budget,
1432 },
1433 }
1436def _tool_failure_result(step_id: str, failure: str) -> dict:
1437 """Tekninen työkaluvika → kova virhe (HARD_ERROR_CODES: TOOL_FAILURE).
1439 Odotetut domain-virheet (not found, path-traversal, decode, malformed
1440 args) palaavat mallille näkyvinä työkalutuloksina eivätkä kulje tästä
1441 (ADR-0012, ADR-0030 D8).
1442 """
1443 return {
1444 "error": (
1445 f"{_TOOL_FAILURE_ERROR_CODE}: step '{step_id}' tool execution "
1446 f"failed: {failure}"
1447 ),
1448 "details": {
1449 "error_code": _TOOL_FAILURE_ERROR_CODE,
1450 "step_id": step_id,
1451 },
1452 }
1455def _apply_tool_meta(value: Any, tool_meta: dict[str, Any] | None) -> Any:
1456 """Attach tool-loop metadata onto a dict-valued step result.
1458 The non-flagged path passes `tool_meta=None` and returns `value` unchanged
1459 (byte-identical). List-valued results (e.g. a parsed JSON array) are left
1460 as-is so the shape contract is preserved; tool_meta is simply not surfaced
1461 there.
1462 """
1463 if not tool_meta:
1464 return value
1465 if isinstance(value, dict):
1466 return {**value, "tool_meta": tool_meta}
1467 return value
1470def _with_failure_flags(value: dict) -> dict:
1471 """Tag an LLM-failure step result with the keys strict condition gates read.
1473 A failing step result carries `escalate`/`verdict` so a strict gate like
1474 `{{ research.escalate }}` resolves instead of raising an unresolved-
1475 reference ValueError, and the failure routes to the escalation/qb surface.
1476 Returns a new dict; the caller's `error`/`raw`/`parse_error` are preserved.
1477 """
1478 return {**value, "escalate": True, "verdict": "ERROR"}
1481def _with_context_metrics(
1482 state: dict,
1483 step_id: str,
1484 step_value: Any,
1485 measurement: dict[str, int],
1486) -> dict:
1487 """Kirjoita step-tulos state[step_id]-polkuun sellaisenaan — sama arvo,
1488 polku ja tyyppi kuin ennen Phase 5:tä, ei mittausavaimia sisällä (Phase
1489 5b) — ja kontekstimittaus rinnakkaiseen step_context_metrics-envelope-
1490 avaimen alle (step_id → {context_size_tokens, context_budget_tokens})."""
1491 metrics = dict(state.get("step_context_metrics") or {})
1492 metrics[step_id] = measurement
1493 return {**state, step_id: step_value, "step_context_metrics": metrics}
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)