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