Coverage for core/langgraph/_llm.py: 86%
669 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-16 06:41 +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 JSON-objektiksi jos mahdollista (audit/triage/review),
14 muuten palauttaa raakatekstin (JSON-parse-fallback)
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 """Yritä parsia JSON. Palauttaa (data, error). data=None jos epäonnistui."""
702 if not text or not text.strip():
703 return None, "empty"
704 raw = text.strip()
705 fence = re.search(r"```(?:json)?\s*([\s\S]*?)```", raw)
706 if fence:
707 raw = fence.group(1).strip()
708 else:
709 m = re.search(r"\{[\s\S]*\}|\[[\s\S]*\]", raw)
710 if m:
711 raw = m.group(0)
712 try:
713 return json.loads(raw), None
714 except json.JSONDecodeError as e:
715 # Yritys: korjaa katkennut JSON
716 repaired = raw
717 if repaired.count('"') % 2 == 1:
718 repaired = repaired.rsplit('"', 1)[0] + '"'
719 repaired = re.sub(r",\s*$", "", repaired.rstrip())
720 for suffix in ("", "}", "]}", "\"]}"):
721 try:
722 return json.loads(repaired + suffix), None
723 except json.JSONDecodeError:
724 continue
725 return None, str(e)
728# ---------------------------------------------------------------------------
729# File-access tool (agent spec.file_access: true)
730# ---------------------------------------------------------------------------
733@dataclass
734class ReadFileResult:
735 content: str
736 truncated: bool
737 error: str | None
739 def as_tool_message(self) -> str:
740 payload: dict[str, Any] = {
741 "name": _READ_FILE_TOOL_NAME,
742 "ok": self.error is None,
743 "content": self.content,
744 "truncated": self.truncated,
745 }
746 if self.error is not None:
747 payload["error"] = self.error
748 return json.dumps(payload, ensure_ascii=False)
751_READ_FILE_TOOL: dict[str, Any] = {
752 "type": "function",
753 "function": {
754 "name": _READ_FILE_TOOL_NAME,
755 "description": (
756 "Read a text file from the project root. The path is resolved "
757 "against the project root; any read that resolves outside the "
758 "project root is rejected. Content is truncated and flagged when "
759 "it exceeds the caps, and can be windowed by offset/limit (lines)."
760 ),
761 "parameters": {
762 "type": "object",
763 "properties": {
764 "path": {
765 "type": "string",
766 "description": "Path to the file, relative to the project root.",
767 },
768 "offset": {
769 "type": "integer",
770 "description": "Number of lines to skip (0-based). Optional.",
771 },
772 "limit": {
773 "type": "integer",
774 "description": "Max number of lines to return. Optional.",
775 },
776 },
777 "required": ["path"],
778 "additionalProperties": False,
779 },
780 },
781}
784def _is_within(root: Path, candidate: Path) -> bool:
785 return candidate == root or root in candidate.parents
788def _read_file(
789 project_root: Path,
790 path: str,
791 offset: int | None,
792 limit: int | None,
793) -> ReadFileResult:
794 """Realpath-bound read of a text file inside the project root (ADR-0012).
796 Every rejected read returns a clear error — no silent fallback. Truncation
797 over `_READ_FILE_MAX_BYTES` / `_READ_FILE_MAX_LINES` is explicit via
798 `truncated`.
799 """
800 if not path or not str(path).strip():
801 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: 'path' is required (ADR-0012)")
802 try:
803 root = project_root.resolve(strict=False)
804 except OSError as exc:
805 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot resolve project root: {exc}")
806 try:
807 candidate = (root / path).resolve(strict=False)
808 except OSError as exc:
809 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot resolve '{path}': {exc}")
810 if not _is_within(root, candidate):
811 return ReadFileResult(
812 "",
813 False,
814 f"{_READ_FILE_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
815 )
816 if not candidate.is_file():
817 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: '{path}' is not a file or does not exist")
818 try:
819 text = candidate.read_text(encoding="utf-8")
820 except (OSError, UnicodeDecodeError) as exc:
821 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot read '{path}': {exc}")
823 lines = text.splitlines()
824 total_lines = len(lines)
825 try:
826 start = int(offset) if offset is not None and int(offset) > 0 else 0
827 except (TypeError, ValueError):
828 start = 0
829 if limit is not None:
830 # A requested window is not an engine cap — no line-level truncation flag.
831 try:
832 max_lines = int(limit)
833 except (TypeError, ValueError):
834 max_lines = _READ_FILE_MAX_LINES
835 truncated = False
836 end = start + max_lines if max_lines >= 0 else total_lines
837 else:
838 max_lines = _READ_FILE_MAX_LINES
839 truncated = (start + max_lines) < total_lines
840 end = start + max_lines
841 content = "\n".join(lines[start:end])
843 if len(content.encode("utf-8")) > _READ_FILE_MAX_BYTES:
844 content = content[:_READ_FILE_MAX_BYTES]
845 truncated = True
846 return ReadFileResult(content, truncated, None)
849@dataclass
850class ListDirEntry:
851 path: str
852 name: str
853 size: int | None
854 kind: str # "dir" | "file"
857@dataclass
858class ListDirResult:
859 entries: list[ListDirEntry]
860 truncated: bool
861 root: str
862 error: str | None
864 def as_tool_message(self) -> str:
865 payload: dict[str, Any] = {
866 "name": _LIST_DIR_TOOL_NAME,
867 "ok": self.error is None,
868 "entries": [
869 {"path": e.path, "name": e.name, "size": e.size, "kind": e.kind}
870 for e in self.entries
871 ],
872 "truncated": self.truncated,
873 "root": self.root,
874 }
875 if self.error is not None:
876 payload["error"] = self.error
877 return json.dumps(payload, ensure_ascii=False)
880_LIST_DIR_TOOL: dict[str, Any] = {
881 "type": "function",
882 "function": {
883 "name": _LIST_DIR_TOOL_NAME,
884 "description": (
885 "List ONE level of a directory inside the project root. The path is "
886 "resolved against the project root; any listing that resolves "
887 "outside the project root is rejected. Non-recursive: nested "
888 "subdirectories are not expanded. Only structural skip dirs are "
889 "omitted. The listing is capped and flagged when it exceeds the cap."
890 ),
891 "parameters": {
892 "type": "object",
893 "properties": {
894 "path": {
895 "type": "string",
896 "description": "Directory path, relative to the project root.",
897 },
898 },
899 "required": ["path"],
900 "additionalProperties": False,
901 },
902 },
903}
906def _execute_list_dir_tool(project_root: Path, path: str) -> ListDirResult:
907 """Realpath-bound ONE-level directory listing inside the project root
908 (ADR-0012). Every rejected listing returns a clear error — no silent
909 fallback. Only structural skip dirs are omitted; entries over
910 `_LIST_DIR_MAX_ENTRIES` are capped and flagged via `truncated`.
911 """
912 if not path or not str(path).strip():
913 return ListDirResult(
914 [], False, "", f"{_LIST_DIR_TOOL_NAME}: 'path' is required (ADR-0012)"
915 )
916 try:
917 root = project_root.resolve(strict=False)
918 except OSError as exc:
919 return ListDirResult(
920 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot resolve project root: {exc}"
921 )
922 try:
923 candidate = (root / str(path)).resolve(strict=False)
924 except OSError as exc:
925 return ListDirResult(
926 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot resolve '{path}': {exc}"
927 )
928 if not _is_within(root, candidate):
929 return ListDirResult(
930 [],
931 False,
932 "",
933 f"{_LIST_DIR_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
934 )
935 if not candidate.is_dir():
936 return ListDirResult(
937 [],
938 False,
939 "",
940 f"{_LIST_DIR_TOOL_NAME}: '{path}' is not a directory or does not exist",
941 )
942 try:
943 children = sorted(candidate.iterdir(), key=lambda p: p.name)
944 except OSError as exc:
945 return ListDirResult(
946 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot list '{path}': {exc}"
947 )
949 entries: list[ListDirEntry] = []
950 for child in children:
951 if child.name in _DEFAULT_SKIP_DIRS:
952 continue
953 is_dir = child.is_dir()
954 size: int | None = None
955 if not is_dir:
956 try:
957 size = child.stat().st_size
958 except OSError:
959 size = None
960 entries.append(
961 ListDirEntry(
962 path=str(child.relative_to(root)),
963 name=child.name,
964 size=size,
965 kind="dir" if is_dir else "file",
966 )
967 )
968 truncated = len(entries) > _LIST_DIR_MAX_ENTRIES
969 if truncated:
970 entries = entries[:_LIST_DIR_MAX_ENTRIES]
971 return ListDirResult(entries, truncated, str(root), None)
974def _invalid_tool_arguments_result(
975 name: str, reason: str
976) -> ReadFileResult | ListDirResult:
977 """Malformed tool arguments (unparseable JSON, non-object JSON, wrong
978 types) are an expected domain error: a soft, model-visible tool result —
979 never the technical-failure marker (ADR-0012, ADR-0030 D8)."""
980 if name == _LIST_DIR_TOOL_NAME:
981 return ListDirResult([], False, "", f"{_LIST_DIR_TOOL_NAME}: {reason}")
982 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: {reason}")
985# ---------------------------------------------------------------------------
986# OpenAIInvoker
987# ---------------------------------------------------------------------------
990class OpenAIInvoker:
991 """
992 Default LLM invoker for agent-platform pipelines.
994 Yksi instanssi per pipeline-ajo. Thread-safe (semafori rinnakkaisuudelle).
995 """
997 def __init__(
998 self,
999 *,
1000 providers: dict[str, dict[str, Any]],
1001 max_parallel: int,
1002 budget: RunBudget,
1003 tracer: LangfuseTracer,
1004 ) -> None:
1005 self._providers = providers
1006 self._clients: dict[str, Any] = {}
1007 self.budget = budget
1008 self.tracer = tracer
1009 self._semaphore = threading.Semaphore(max(1, max_parallel))
1011 def _openai_client(self, base_url: str, api_key: str) -> Any:
1012 key = f"{base_url}|{api_key}"
1013 if key not in self._clients:
1014 from openai import OpenAI # type: ignore
1015 self._clients[key] = OpenAI(base_url=base_url, api_key=api_key)
1016 return self._clients[key]
1018 # ---- Public callable -------------------------------------------------
1020 def __call__(
1021 self,
1022 resolved_spec: dict,
1023 resolved_step: dict,
1024 state: dict,
1025 ) -> dict:
1026 """
1027 StepRunner-yhteensopiva: (spec, step, state) → uusi state.
1029 `resolved_step` = step jossa params/input on JO resolvoitu
1030 StepRunnerissa template-engineä käyttäen. Tämä invoker ei enää
1031 evaluoi `{{ }}`-lausekkeita state/run_context-vastaan, vain
1032 substituoi `step.params` ja sisäänrakennetut feature-arvot
1033 agentin `system_prompt`/`user`-mallineisiin.
1034 """
1035 step_id = resolved_step.get("id", "?")
1036 agent_spec = resolved_spec.get("spec") or {}
1038 model_spec = agent_spec.get("model") or {}
1039 provider_name = model_spec.get("provider")
1041 if not provider_name:
1042 if self._providers:
1043 provider_name = next(iter(self._providers))
1044 else:
1045 raise ValueError("No LLM provider configured and no provider specified in agent spec")
1047 provider_cfg = self._providers.get(provider_name)
1048 if provider_cfg is None:
1049 raise ValueError(
1050 f"Unknown LLM provider '{provider_name}'. "
1051 f"Available: {', '.join(sorted(self._providers.keys()))}"
1052 )
1054 model_override = model_spec.get("model")
1055 model = model_override if model_override else provider_cfg["model"]
1057 # Budget check
1058 if not self.budget.try_consume():
1059 return {
1060 **state,
1061 step_id: _with_failure_flags(
1062 {"error": f"budget:{self.budget.stop_reason}"}
1063 ),
1064 }
1066 # ---- Build system + user messages --------------------------------
1067 skills_text = load_skills(resolved_spec)
1068 step_params = resolved_step.get("params") or {}
1069 input_value = resolved_step.get("input")
1071 # Render system_prompt template with params, then inject the loaded
1072 # skills after the rendered prompt (ADR-0036 D1/D6).
1073 prompt_ctx: dict[str, Any] = {
1074 k: v for k, v in step_params.items() if isinstance(k, str)
1075 }
1076 rendered_prompt = _render(
1077 str(agent_spec.get("system_prompt") or ""),
1078 prompt_ctx,
1079 ).strip()
1080 communication = resolved_step.get(COMMUNICATION_CONTEXT_KEY)
1081 communication_text = (
1082 render_communication_block(communication) if communication else ""
1083 )
1084 instructions_text = render_usage_instructions_block(
1085 resolved_step.get(USAGE_INSTRUCTIONS_KEY)
1086 )
1087 envelope_text = ENVELOPE_CONTRACT_BLOCK
1088 system_text = "\n\n".join(
1089 part
1090 for part in (
1091 rendered_prompt,
1092 skills_text,
1093 envelope_text,
1094 instructions_text,
1095 communication_text,
1096 )
1097 if part
1098 )
1100 # User message: serialise input + step params
1101 user_text = _format_user_message(input_value, step_params)
1103 # Temperature: agent spec → provider default → step params
1104 temperature = float(
1105 model_spec.get("temperature", provider_cfg.get("temperature", 0.0))
1106 )
1107 if "temperature" in step_params:
1108 try:
1109 temperature = float(step_params["temperature"])
1110 except (TypeError, ValueError):
1111 pass
1113 messages = [
1114 {"role": "system", "content": system_text},
1115 {"role": "user", "content": user_text},
1116 ]
1118 # ---- Context-integrity check (Phase 4 doc, ## Kontekstin eheys) --
1119 agent_name = (resolved_spec.get("metadata") or {}).get("name", "agent")
1120 context_budget = _effective_context_budget(resolved_spec, agent_name)
1121 context_tokens = count_context_tokens(messages)
1122 if context_tokens > context_budget:
1123 return _with_context_metrics(
1124 state,
1125 step_id,
1126 _context_overflow_result(step_id, context_tokens, context_budget),
1127 {"context_size_tokens": context_tokens, "context_budget_tokens": context_budget},
1128 )
1129 measurement = {
1130 "context_size_tokens": context_tokens,
1131 "context_budget_tokens": context_budget,
1132 }
1134 # ---- LLM call with Langfuse generation ---------------------------
1135 gen_name = f"{step_id}:{agent_name}"
1137 with self._semaphore:
1138 obs = self.tracer.generation(
1139 gen_name,
1140 model=model,
1141 temperature=temperature,
1142 messages=messages,
1143 metadata=measurement,
1144 )
1145 content = ""
1146 usage: dict[str, int] | None = None
1147 err: str | None = None
1148 tool_meta: dict[str, Any] | None = None
1149 try:
1150 if bool(agent_spec.get("file_access")):
1151 content, usage, tool_meta = self._stream_completion_with_tools(
1152 messages,
1153 temperature,
1154 model,
1155 provider_cfg["base_url"],
1156 provider_cfg["api_key"],
1157 project_root=resolve_workspace_root(state),
1158 )
1159 else:
1160 content, usage = self._stream_completion(messages, temperature, model, provider_cfg["base_url"], provider_cfg["api_key"])
1161 except Exception as exc: # noqa: BLE001
1162 err = f"{type(exc).__name__}: {exc}"
1163 log.warning("[invoke] %s LLM call failed: %s", step_id, err)
1164 try:
1165 obs.update(
1166 output=content[:8000] if len(content) > 8000 else content,
1167 usage_details=usage,
1168 )
1169 except Exception: # noqa: BLE001
1170 pass
1171 try:
1172 obs.end()
1173 except Exception: # noqa: BLE001
1174 pass
1175 self.tracer.flush()
1177 # ---- Parse output ------------------------------------------------
1178 if err:
1179 return _with_context_metrics(
1180 state,
1181 step_id,
1182 _with_failure_flags({"error": err, "raw": content}),
1183 measurement,
1184 )
1186 if tool_meta and tool_meta.get("tool_failure"):
1187 return _with_context_metrics(
1188 state,
1189 step_id,
1190 _tool_failure_result(step_id, tool_meta["tool_failure"]),
1191 measurement,
1192 )
1194 parsed, perr = _try_parse_json(content)
1195 if parsed is not None:
1196 verr = validate_envelope(parsed)
1197 if verr is not None:
1198 return _with_context_metrics(
1199 state,
1200 step_id,
1201 {
1202 "error": f"ENVELOPE_ERROR: {verr}",
1203 "details": {
1204 "error_code": ENVELOPE_ERROR_CODE,
1205 "step_id": step_id,
1206 },
1207 },
1208 measurement,
1209 )
1210 return _with_context_metrics(
1211 state, step_id, _apply_tool_meta(parsed, tool_meta), measurement
1212 )
1213 # Fallback: raw text
1214 return _with_context_metrics(
1215 state,
1216 step_id,
1217 _apply_tool_meta(
1218 _with_failure_flags({"output": content.strip(), "parse_error": perr}),
1219 tool_meta,
1220 ),
1221 measurement,
1222 )
1224 # ---- Streaming completion -------------------------------------------
1226 def _stream_completion(
1227 self,
1228 messages: list[dict[str, str]],
1229 temperature: float,
1230 model: str,
1231 base_url: str,
1232 api_key: str,
1233 ) -> tuple[str, dict[str, int] | None]:
1234 client = self._openai_client(base_url, api_key)
1235 resp = client.chat.completions.create(
1236 model=model,
1237 temperature=temperature,
1238 messages=messages,
1239 stream=True,
1240 stream_options={"include_usage": True},
1241 )
1242 content = ""
1243 usage: dict[str, int] | None = None
1244 for chunk in resp:
1245 if chunk.choices and chunk.choices[0].delta.content:
1246 content += chunk.choices[0].delta.content
1247 if getattr(chunk, "usage", None):
1248 u = chunk.usage
1249 usage = {
1250 "input": u.prompt_tokens or 0,
1251 "output": u.completion_tokens or 0,
1252 "total": u.total_tokens or 0,
1253 }
1254 return content, usage
1256 def _execute_file_tool(
1257 self, tool_call: Any, project_root: Path
1258 ) -> ReadFileResult | ListDirResult:
1259 """Execute a read_file or list_dir tool call against the project root."""
1260 name = getattr(tool_call.function, "name", "")
1261 try:
1262 args = json.loads(getattr(tool_call.function, "arguments", "{}"))
1263 except (TypeError, json.JSONDecodeError) as exc:
1264 return _invalid_tool_arguments_result(name, f"invalid arguments: {exc}")
1265 if not isinstance(args, dict):
1266 return _invalid_tool_arguments_result(
1267 name,
1268 f"invalid arguments: expected a JSON object, got {type(args).__name__}",
1269 )
1270 if name == _LIST_DIR_TOOL_NAME:
1271 return _execute_list_dir_tool(project_root, args.get("path"))
1272 return _read_file(project_root, args.get("path"), args.get("offset"), args.get("limit"))
1274 # ---- Tool-looping completion (file_access: true) --------------------
1276 def _stream_completion_with_tools(
1277 self,
1278 messages: list[dict[str, str]],
1279 temperature: float,
1280 model: str,
1281 base_url: str,
1282 api_key: str,
1283 *,
1284 project_root: Path,
1285 ) -> tuple[str, dict[str, int] | None, dict[str, Any]]:
1286 """Non-streaming completion carrying the read_file/list_dir tool loop.
1288 Each tool round consumes ONE additional inference through
1289 `self.budget.try_consume`; the loop is hard-capped at `_MAX_TOOL_ROUNDS`
1290 and stops instead of looping indefinitely. A cap-stop issues exactly ONE
1291 final tool-free completion (budget-gated) so the agent can answer from
1292 the gathered context; a budget-stop never does. A technical tool
1293 failure is surfaced via `tool_meta["tool_failure"]` for the caller to
1294 turn into a hard error. The non-flagged path never reaches here.
1295 """
1296 client = self._openai_client(base_url, api_key)
1297 working = list(messages)
1298 content = ""
1299 usage: dict[str, int] | None = None
1300 tool_rounds = 0
1301 truncated = False
1302 tool_errors: list[str] = []
1303 tool_failure: str | None = None
1304 stop_reason: str | None = None
1306 while True:
1307 resp = client.chat.completions.create(
1308 model=model,
1309 temperature=temperature,
1310 messages=working,
1311 tools=[_READ_FILE_TOOL, _LIST_DIR_TOOL],
1312 stream=False,
1313 )
1314 usage = _usage_from_response(resp) or usage
1315 msg = resp.choices[0].message
1316 piece = getattr(msg, "content", None) or ""
1317 if piece:
1318 content += piece
1319 tool_calls = getattr(msg, "tool_calls", None)
1320 if not tool_calls:
1321 break
1322 working.append(
1323 {
1324 "role": "assistant",
1325 "content": piece or "",
1326 "tool_calls": [
1327 {
1328 "id": tc.id,
1329 "type": "function",
1330 "function": {"name": tc.function.name, "arguments": tc.function.arguments},
1331 }
1332 for tc in tool_calls
1333 ],
1334 }
1335 )
1336 for tc in tool_calls:
1337 try:
1338 result = self._execute_file_tool(tc, project_root)
1339 except Exception as exc: # noqa: BLE001
1340 tool_failure = f"{type(exc).__name__}: {exc}"
1341 break
1342 truncated = truncated or result.truncated
1343 if result.error:
1344 tool_errors.append(result.error)
1345 working.append(
1346 {"role": "tool", "tool_call_id": tc.id, "content": result.as_tool_message()}
1347 )
1348 if tool_failure is not None:
1349 break
1350 tool_rounds += 1
1351 if tool_rounds >= _MAX_TOOL_ROUNDS:
1352 stop_reason = _TOOL_STOP_REASON_CAP
1353 break
1354 if not self.budget.try_consume():
1355 stop_reason = _TOOL_STOP_REASON_BUDGET
1356 break
1358 final_completion = False
1359 if stop_reason == _TOOL_STOP_REASON_CAP and self.budget.try_consume():
1360 final_content, final_usage = self._final_tool_free_completion(
1361 client, working, model, temperature
1362 )
1363 content += final_content
1364 usage = final_usage or usage
1365 final_completion = True
1367 tool_meta: dict[str, Any] = {
1368 "tool_rounds": tool_rounds,
1369 "truncated": truncated,
1370 "tool_stopped": stop_reason is not None,
1371 "tool_stop_reason": stop_reason,
1372 "final_completion": final_completion,
1373 "tool_errors": tool_errors,
1374 }
1375 if tool_failure is not None:
1376 tool_meta["tool_failure"] = tool_failure
1377 return content, usage, tool_meta
1379 def _final_tool_free_completion(
1380 self,
1381 client: Any,
1382 working: list[dict[str, Any]],
1383 model: str,
1384 temperature: float,
1385 ) -> tuple[str, dict[str, int] | None]:
1386 """One final completion on the gathered conversation, no tools."""
1387 resp = client.chat.completions.create(
1388 model=model,
1389 temperature=temperature,
1390 messages=working,
1391 stream=False,
1392 )
1393 return getattr(resp.choices[0].message, "content", None) or "", _usage_from_response(resp)
1396# ---------------------------------------------------------------------------
1397# Helpers
1398# ---------------------------------------------------------------------------
1401def _usage_from_response(resp: Any) -> dict[str, int] | None:
1402 """OpenAI-vastauksen usage-dict (input/output/total) tai None, kun
1403 provider ei palauta usagea."""
1404 u = getattr(resp, "usage", None)
1405 if not u:
1406 return None
1407 return {
1408 "input": getattr(u, "prompt_tokens", 0) or 0,
1409 "output": getattr(u, "completion_tokens", 0) or 0,
1410 "total": getattr(u, "total_tokens", 0) or 0,
1411 }
1414def _effective_context_budget(resolved_spec: dict, agent_name: str) -> int:
1415 """Efektiivinen kontekstibudjetti: spec.limits.context_budget (merge-
1416 järjestyksen tulos) tai oletus 100 000 tokenia. Epäkelpo arvo on kova
1417 virhe (ADR-0012) — ei hiljaista fallbackia."""
1418 limits = (resolved_spec.get("spec") or {}).get("limits") or {}
1419 value = limits.get("context_budget")
1420 if value is None:
1421 return DEFAULT_CONTEXT_BUDGET_TOKENS
1422 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1423 raise ValueError(
1424 f"agent '{agent_name}': spec.limits.context_budget must be a "
1425 f"positive integer, got {value!r} (ADR-0012)"
1426 )
1427 return value
1430def _context_overflow_result(step_id: str, context_tokens: int, context_budget: int) -> dict:
1431 """Budjetin ylitys → kova virhe ennen mallikutsua (fail-fast)."""
1432 return {
1433 "error": (
1434 f"{_CONTEXT_BUDGET_ERROR_CODE}: step '{step_id}' assembled "
1435 f"context is {context_tokens} tokens, exceeding the context "
1436 f"budget of {context_budget} tokens"
1437 ),
1438 "details": {
1439 "error_code": _CONTEXT_BUDGET_ERROR_CODE,
1440 "step_id": step_id,
1441 "assembled_tokens": context_tokens,
1442 "context_budget": context_budget,
1443 },
1444 }
1447def _tool_failure_result(step_id: str, failure: str) -> dict:
1448 """Tekninen työkaluvika → kova virhe (HARD_ERROR_CODES: TOOL_FAILURE).
1450 Odotetut domain-virheet (not found, path-traversal, decode, malformed
1451 args) palaavat mallille näkyvinä työkalutuloksina eivätkä kulje tästä
1452 (ADR-0012, ADR-0030 D8).
1453 """
1454 return {
1455 "error": (
1456 f"{_TOOL_FAILURE_ERROR_CODE}: step '{step_id}' tool execution "
1457 f"failed: {failure}"
1458 ),
1459 "details": {
1460 "error_code": _TOOL_FAILURE_ERROR_CODE,
1461 "step_id": step_id,
1462 },
1463 }
1466def _apply_tool_meta(value: Any, tool_meta: dict[str, Any] | None) -> Any:
1467 """Attach tool-loop metadata onto a dict-valued step result.
1469 The non-flagged path passes `tool_meta=None` and returns `value` unchanged
1470 (byte-identical). List-valued results (e.g. a parsed JSON array) are left
1471 as-is so the shape contract is preserved; tool_meta is simply not surfaced
1472 there.
1473 """
1474 if not tool_meta:
1475 return value
1476 if isinstance(value, dict):
1477 return {**value, "tool_meta": tool_meta}
1478 return value
1481def _with_failure_flags(value: dict) -> dict:
1482 """Tag an LLM-failure step result with the keys strict condition gates read.
1484 A failing step result carries `escalate`/`verdict` so a strict gate like
1485 `{{ research.escalate }}` resolves instead of raising an unresolved-
1486 reference ValueError, and the failure routes to the escalation/qb surface.
1487 Returns a new dict; the caller's `error`/`raw`/`parse_error` are preserved.
1488 """
1489 return {**value, "escalate": True, "verdict": "ERROR"}
1492def _with_context_metrics(
1493 state: dict,
1494 step_id: str,
1495 step_value: Any,
1496 measurement: dict[str, int],
1497) -> dict:
1498 """Kirjoita step-tulos state[step_id]-polkuun sellaisenaan — sama arvo,
1499 polku ja tyyppi kuin ennen Phase 5:tä, ei mittausavaimia sisällä (Phase
1500 5b) — ja kontekstimittaus rinnakkaiseen step_context_metrics-envelope-
1501 avaimen alle (step_id → {context_size_tokens, context_budget_tokens})."""
1502 metrics = dict(state.get("step_context_metrics") or {})
1503 metrics[step_id] = measurement
1504 return {**state, step_id: step_value, "step_context_metrics": metrics}
1507def _format_user_message(input_value: Any, params: dict[str, Any]) -> str:
1508 """
1509 Muotoile agentin user-message resolvoidusta inputista + paramseista.
1510 """
1511 parts: list[str] = []
1513 # Inputin esitysmuoto: yritetään olla informatiivisia
1514 if input_value is None:
1515 pass
1516 elif isinstance(input_value, str):
1517 parts.append(input_value)
1518 elif isinstance(input_value, dict):
1519 # Doc-file form: {path, content, ...} → human-readable
1520 if "path" in input_value and "content" in input_value:
1521 path = input_value.get("path", "")
1522 content = input_value.get("content", "")
1523 parts.append(f"Path: {path}\n\n---\n{content}")
1524 else:
1525 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
1526 elif isinstance(input_value, list):
1527 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
1528 else:
1529 parts.append(str(input_value))
1531 # Lisää muut paramit (focus, finding jne.) jos eivät jo ole system_promptissa
1532 extra: dict[str, Any] = {}
1533 skip_keys = {"skills"}
1534 for k, v in params.items():
1535 if k in skip_keys:
1536 continue
1537 # Älä toista pieniä skalaareja - ne yleensä jo system_promptissa
1538 if isinstance(v, (dict, list)):
1539 extra[k] = v
1540 if extra:
1541 parts.append("## Additional context\n\n" + json.dumps(extra, ensure_ascii=False, indent=2))
1543 return "\n\n".join(p for p in parts if p)