Coverage for core/langgraph/_llm.py: 85%
643 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 02:20 +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- yhdistää AgentInstance-spec:n tiedoista täydellisen system_promptin
8 (skill-tiedostot luettu + `{{ skills }}` korvattu + params-substituutio)
9- serialisoi `step.input`-arvon user-messageksi
10- ajaa OpenAI-streamin DeepSeek-yhteensopivalla endpointilla
11- raportoi Langfuse-generationina pipeline-trace alle
12- parsii vastauksen JSON-objektiksi jos mahdollista (audit/triage/review),
13 muuten palauttaa raw-markdownin (ticket writer/refiner)
14- valvoo RunBudgetia (max LLM calls / wall-clock timeout)
16API:
17 invoker = OpenAIInvoker(config=..., budget=..., project_slug=..., pipeline_name=...)
18 new_state = invoker(resolved_spec, resolved_step, state)
19"""
20from __future__ import annotations
22import contextvars
23import json
24import logging
25import os
26import re
27import subprocess
28import threading
29import time
30from dataclasses import dataclass, field
31from pathlib import Path
32from typing import Any
34log = logging.getLogger(__name__)
36# Files-output helper (output.type: files) — sama dual-import-kuvio kuin
37# runner.py:ssa: flat sys.path (testit, discover-wrapperit) vs. qualified
38# langgraph.* (main.py, PYTHONPATH=/app).
39try:
40 from _files_output import (
41 FilesOutputError,
42 resolve_workspace_root,
43 validate_and_write_files,
44 )
45except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
46 from langgraph._files_output import (
47 FilesOutputError,
48 resolve_workspace_root,
49 validate_and_write_files,
50 )
52# Shared structural-skip set for the list_dir tool (ADR-0012). Imported from
53# _builtins to avoid duplicating the 8-entry set; _builtins does not import
54# _llm (only stdlib + exec_runner + _files_output), so this is cycle-safe.
55try:
56 from _builtins import _DEFAULT_SKIP_DIRS
57except ImportError: # pragma: no cover — main.py-ajossa (qualified importit)
58 from langgraph._builtins import _DEFAULT_SKIP_DIRS
60SKILL_DIR = Path("/skills")
63class SkillLoadError(ValueError):
64 """Skill-latauksen kova virhe (ADR-0012) — ei hiljaista fallbackia.
66 Erottaa skill-resoluution virheet muista ValueErrorista: runnerin
67 parallel-käsittelijät re-raisaavat tämän (kova ajo-failure kaikilla
68 step-polkuilla), kun muut step-virheet pysyvät state-virheinä.
69 """
71# Kontekstin eheys -sopimus (Phase 4 doc, ## Kontekstin eheys):
72# koottu konteksti toimitetaan mallikutsulle kokonaisuudessaan tai step
73# epäonnistuu kovasti — ei hiljaista leikkausta missään muodossa.
74DEFAULT_CONTEXT_BUDGET_TOKENS = 100_000
75_CONTEXT_BUDGET_ERROR_CODE = "VALIDATION_ERROR"
76_FILES_OUTPUT_ERROR_CODE = "FILES_OUTPUT_ERROR"
77_TIKTOKEN_ENCODING = "cl100k_base"
79# --- Opt-in file-access tool (agent spec.file_access: true) ---
80# One bounded read_file tool + a tool-execution loop. A tool round = one
81# additional inference call consumed through the run budget.
82_READ_FILE_TOOL_NAME = "read_file"
83_LIST_DIR_TOOL_NAME = "list_dir"
84_MAX_TOOL_ROUNDS = 3
85_READ_FILE_MAX_BYTES = 200_000
86_READ_FILE_MAX_LINES = 2000
87_LIST_DIR_MAX_ENTRIES = 512
88_TIKTOKEN_CACHE_DIR = Path(__file__).parent / "tiktoken_cache"
90_LANGFUSE_FLUSH_EVERY = 8
93# ---------------------------------------------------------------------------
94# RunBudget — deterministinen pipeline-ajon suoritusbudjetti
95# ---------------------------------------------------------------------------
98@dataclass
99class RunBudget:
100 max_llm_calls: int = 48
101 run_timeout_seconds: int = 240
102 start: float = field(default_factory=time.monotonic)
103 llm_calls: int = 0
104 stopped: bool = False
105 stop_reason: str = ""
106 _lock: threading.Lock = field(default_factory=threading.Lock)
108 def try_consume(self) -> bool:
109 with self._lock:
110 if self.stopped:
111 return False
112 if time.monotonic() - self.start > self.run_timeout_seconds:
113 self.stopped = True
114 self.stop_reason = "run_timeout"
115 return False
116 if self.llm_calls >= self.max_llm_calls:
117 self.stopped = True
118 self.stop_reason = "max_llm_calls"
119 return False
120 self.llm_calls += 1
121 return True
123 def as_stats(self) -> dict[str, Any]:
124 return {
125 "llm_calls": self.llm_calls,
126 "budget_stopped": self.stopped,
127 "budget_stop_reason": self.stop_reason or None,
128 }
131# ---------------------------------------------------------------------------
132# Skill loading
133# ---------------------------------------------------------------------------
136def _skill_repos_from_env() -> list[dict]:
137 """Lue SKILL_REPOS-env (JSON [{"name": ..., "url": ...}]) → lista entryistä.
139 Puuttuva env → [] (sopimus: SKILL_REPOS on valinnainen). Mikä tahansa
140 epäkelpo arvo on kova virhe (ADR-0012): malformi JSON, ei-lista, tai
141 entry ilman ei-tyhjää name- ja url-merkkijonoa → SkillLoadError.
142 """
143 raw = os.getenv("SKILL_REPOS")
144 if not raw:
145 return []
146 try:
147 repos = json.loads(raw)
148 except json.JSONDecodeError as exc:
149 raise SkillLoadError(f"SKILL_REPOS is not valid JSON: {exc}") from exc
150 if not isinstance(repos, list):
151 raise SkillLoadError(f"SKILL_REPOS must be a JSON list, got {type(repos).__name__}")
152 for idx, repo in enumerate(repos):
153 if not isinstance(repo, dict):
154 raise SkillLoadError(
155 f"SKILL_REPOS entry {idx} must be an object, got {type(repo).__name__}"
156 )
157 name = repo.get("name")
158 url = repo.get("url")
159 if not isinstance(name, str) or not name.strip():
160 raise SkillLoadError(f"SKILL_REPOS entry {idx} is missing a non-empty string 'name'")
161 if not isinstance(url, str) or not url.strip():
162 raise SkillLoadError(
163 f"SKILL_REPOS entry {idx} ('{name}') is missing a non-empty string 'url'"
164 )
165 return repos
168def locate_skills_root(repo_mount: Path) -> Path:
169 """Etsi skill-repo-mountista tasan yksi `skills`-niminen kansio (find).
171 Ajaa hardcodetun find-komennon (ei parametrisoitava):
173 find <mount> -mindepth 1 -type d -name .git -prune \\
174 -o -type d -name skills -print
176 Etsii pelkät `skills`-nimiset kansiot (ei tiedostoja, ei sisäkansioita),
177 `.git`-puu poistetaan (prune), osumat lajitellaan (find-järjestys ei ole
178 taattu → determinismi). Tarkalleen 1 osuma → palauttaa polun Path-
179 objektina. 0 tai >1 osumaa on kova virhe (ADR-0012), kuten myös itse
180 komennon epäonnistuminen (find puuttuu, returncode != 0) — ei hiljaista
181 ohitusta.
182 """
183 cmd = [
184 "find",
185 str(repo_mount),
186 "-mindepth",
187 "1",
188 "-type",
189 "d",
190 "-name",
191 ".git",
192 "-prune",
193 "-o",
194 "-type",
195 "d",
196 "-name",
197 "skills",
198 "-print",
199 ]
200 try:
201 proc = subprocess.run(cmd, capture_output=True, text=True)
202 except OSError as exc:
203 raise SkillLoadError(
204 f"failed to run find for skill mount {repo_mount}: {exc}"
205 ) from exc
206 if proc.returncode != 0:
207 raise SkillLoadError(
208 f"find failed for skill mount {repo_mount} "
209 f"(rc={proc.returncode}): {proc.stderr.strip()}"
210 )
211 hits = sorted(line.strip() for line in proc.stdout.splitlines() if line.strip())
212 if not hits:
213 raise SkillLoadError(f"no skills folder found in {repo_mount}")
214 if len(hits) > 1:
215 raise SkillLoadError(
216 f"ambiguous skills folders found in {repo_mount}: " + ", ".join(hits)
217 )
218 return Path(hits[0])
221def _normalize_skill_libs() -> list[Path]:
222 """Skill-lähdelista: koodirepon skills/-kirjasto ensin, sitten skill-repot.
224 Koodirepon implisiittinen skills/-kirjasto (`PROJECT_ROOT/skills/`,
225 oletus `/project/skills`) on aina korkein prioriteetti skill-lähteenä
226 (koodirepon ylivalta, ADR-0018 D3), ja skill-repot tulevat SKILL_REPOS-
227 järjestyksessä (ADR-0018 D4) listan loppuun: kunkin mountin kohdalla
228 etsitään `skills`-kansio locate_skills_root()-tuloksella. Jos skill-
229 reposta ei löydy skills-kansiota, ajo epäonnistuu kovasti (ADR-0012) —
230 koodirepon puuttuvan skills/-kansion hiljainen ohitus pysyy ennallaan
231 (resoluutio ohittaa puuttuvan kirjaston).
232 """
233 project_root = Path(os.getenv("PROJECT_ROOT", "/project"))
234 return [project_root / "skills"] + [
235 locate_skills_root(SKILL_DIR / repo["name"]) for repo in _skill_repos_from_env()
236 ]
239def load_skills(spec: dict, *, max_chars: int = 16000) -> str:
240 """
241 Lukee `spec.skills`-flat-listan (entryjä) ja palauttaa yhdistetyn
242 tekstin. Kunkin tiedoston yläraja on max_chars merkkiä (oletus
243 16 000): ylittävä tiedosto on kova SkillLoadError (ADR-0012) —
244 ei hiljaista katkaisua.
246 Entry on joko kansionimi (`<nimi>`): koko `<nimi>/`-kansion sisältö
247 injektoidaan otsakkeilla (SKILL.md ensin, sitten loput
248 aakkosjärjestyksessä suhteellisen polun mukaan), tai tarkka
249 tiedosto (`<skill>/<tiedostonimi>`): vain kyseinen tiedosto
250 injektoidaan. Tiedosto-entryt eivät saa sisältää alipolkuja —
251 täsmälleen yksi `/`; virheellinen muoto (johto-/loppuviiva, tyhjä
252 segmentti, `..`, absoluuttinen polku) on kova SkillLoadError.
254 Ensimmäinen skill-lähde, josta kohde löytyy, voittaa (first-match):
255 koodirepon skills/-kirjasto (`PROJECT_ROOT/skills/`, koodirepon
256 ylivalta, ADR-0018 D3) ensin, sitten skill-repot SKILL_REPOS-
257 järjestyksessä (prioriteetti, ADR-0018 D4); puuttuvat lähteet
258 ohitetaan hiljaa.
260 Resolvoitumaton entry → SkillLoadError (ADR-0012): viesti listaa
261 etsityt kirjastopolut ja saatavilla olevat skillit. Ei hiljaista
262 ohitusta.
263 """
264 skills = (spec.get("spec") or {}).get("skills") or []
265 if not skills:
266 return ""
267 libs = _normalize_skill_libs()
268 parts: list[str] = []
269 for entry in skills:
270 parts.extend(_load_bare_skill(entry, libs, max_chars=max_chars))
271 return "\n\n---\n\n".join(parts)
274def _parse_skill_entry(entry: str) -> tuple[str, str | None]:
275 """Parseroi spec.skills-entryn: kansionimi → (nimi, None), tarkka
276 tiedosto `<skill>/<tiedostonimi>` → (skill, tiedostonimi).
278 Virheellinen muoto → kova SkillLoadError: tyhjä entry, `.`/`..`,
279 tyhjä segmentti, absoluuttinen polku tai >1 vinoviivaa (alipolut
280 eivät ole sallittuja).
281 """
282 if not entry.strip():
283 raise SkillLoadError(
284 f"skill entry '{entry}' is empty; use '<skill>' "
285 "or '<skill>/<file>'"
286 )
287 if entry.startswith("/"):
288 raise SkillLoadError(
289 f"skill entry '{entry}' is an absolute path; use '<skill>' "
290 "or '<skill>/<file>'"
291 )
292 parts = entry.split("/")
293 if any(seg in ("", ".", "..") for seg in parts):
294 raise SkillLoadError(
295 f"skill entry '{entry}' contains an invalid path segment "
296 "(''|'.'|'..'); use '<skill>' or '<skill>/<file>'"
297 )
298 if len(parts) == 1:
299 return entry, None
300 if len(parts) > 2:
301 raise SkillLoadError(
302 f"skill entry '{entry}' has too many path segments; subpaths "
303 "are not allowed, use '<skill>' or '<skill>/<file>'"
304 )
305 skill, filename = parts
306 return skill, filename
309def _load_bare_skill(entry: str, libs: list[Path], *, max_chars: int) -> list[str]:
310 """Skill-entry (kansionimi tai tarkka tiedosto) ladattuna otsakkeilla.
312 Kansionimi (`<nimi>`): kansion `lib / <nimi>/` kaikki tiedostot
313 injektoidaan (SKILL.md ensin, sitten loput aakkosjärjestyksessä).
314 Tarkka tiedosto (`<skill>/<tiedostonimi>`): vain kyseinen tiedosto
315 `lib / <skill> / <tiedosto>` injektoidaan otsakkeella
316 `## Skill: <skill> — <tiedosto>`.
318 Ensimmäinen skill-lähde, josta kohde löytyy, voittaa (first-match).
319 Kansionimelle vastaava `lib / entry` tiedostona → SkillLoadError;
320 tiedosto-entrylle kansion löytyminen ilman kohdetiedostoa ohittaa
321 kyseisen lähteen hiljaa. Virheellinen entry-muoto tai kohde, jota
322 ei löydy mistään skill-lähteestä → SkillLoadError.
323 """
324 name, filename = _parse_skill_entry(entry)
325 for lib in libs:
326 folder = lib / name
327 if filename is None:
328 if folder.is_dir():
329 parts: list[str] = []
330 for path in _skill_folder_files(folder):
331 relpath = path.relative_to(folder)
332 parts.append(
333 f"## Skill: {name} — {relpath}\n\n"
334 f"{_read_skill_file(entry, path, max_chars=max_chars)}"
335 )
336 return parts
337 if folder.is_file():
338 raise SkillLoadError(
339 f"skill '{entry}' resolves to a file, expected a skill folder "
340 f"(SKILL.md etc.): {folder}"
341 )
342 else:
343 file_path = folder / filename
344 if file_path.is_file():
345 return [
346 f"## Skill: {name} — {filename}\n\n"
347 f"{_read_skill_file(entry, file_path, max_chars=max_chars)}"
348 ]
349 if folder.is_dir():
350 continue
351 raise SkillLoadError(_unresolved_skill_error(entry, libs))
354def _skill_folder_files(folder: Path) -> list[Path]:
355 """Kansion kaikki tiedostot rekursiivisesti, deterministinen järjestys:
356 juuren SKILL.md ensin, sitten loput suhteellisen polun mukaan
357 aakkosjärjestyksessä."""
358 files = [p for p in folder.rglob("*") if p.is_file()]
359 files.sort(
360 key=lambda p: (
361 p.relative_to(folder) != Path("SKILL.md"),
362 str(p.relative_to(folder)),
363 )
364 )
365 return files
368def _read_skill_file(entry: str, path: Path, *, max_chars: int) -> str:
369 try:
370 text = path.read_text(encoding="utf-8", errors="replace")
371 except OSError as exc:
372 raise SkillLoadError(
373 f"skill entry '{entry}' file unreadable: {path} ({exc})"
374 ) from exc
375 if len(text) > max_chars:
376 raise SkillLoadError(
377 f"skill entry '{entry}' file exceeds the {max_chars}-char limit: "
378 f"{path} ({len(text)} chars)"
379 )
380 return text
383def _available_skill_names(libs: list[Path]) -> list[str]:
384 """Kirjastojen tarjonta: skill-kansiot (sisältävät SKILL.md) + top-tason
385 *.md-tiedostot. Sorted + deduped."""
386 names: set[str] = set()
387 for lib in libs:
388 if not lib.is_dir():
389 continue
390 for child in lib.iterdir():
391 if child.is_dir():
392 if (child / "SKILL.md").is_file():
393 names.add(child.name)
394 elif child.is_file() and child.name.endswith(".md"):
395 names.add(child.name)
396 return sorted(names)
399def _unresolved_skill_error(entry: str, libs: list[Path]) -> str:
400 available = ", ".join(_available_skill_names(libs)) or "(none)"
401 searched = "\n".join(f" - {lib}" for lib in libs)
402 return (
403 f"skill '{entry}' not found in any skill repo "
404 f"(available skills: {available}). Searched skill repos:\n{searched}"
405 )
408# ---------------------------------------------------------------------------
409# Template substitution (simple {{ key }} replacement against a dict)
410# ---------------------------------------------------------------------------
412_TMPL_RE = re.compile(r"\{\{\s*([a-zA-Z_][a-zA-Z0-9_.]*)\s*\}\}")
415def _render(text: str, ctx: dict[str, Any]) -> str:
416 if not isinstance(text, str) or "{{" not in text:
417 return text
419 def _lookup(path: str) -> str:
420 cur: Any = ctx
421 for part in path.split("."):
422 if isinstance(cur, dict):
423 cur = cur.get(part)
424 else:
425 return ""
426 if cur is None:
427 return ""
428 return str(cur) if not isinstance(cur, str) else cur
430 return _TMPL_RE.sub(lambda m: _lookup(m.group(1)), text)
433# ---------------------------------------------------------------------------
434# Token counting (deterministinen ja provider-agnostinen; Phase 4 doc)
435# ---------------------------------------------------------------------------
437_tokenizer_encoding: Any | None = None
438_tokenizer_lock = threading.Lock()
441def _tokenizer() -> Any:
442 """tiktoken-encoding lazy-singleton, joka käyttää bundlattua
443 TIKTOKEN_CACHE_DIR:ää (core/langgraph/tiktoken_cache/) — ei
444 verkkolatausta ensimmäiselläkään käyttökerralla."""
445 global _tokenizer_encoding
446 if _tokenizer_encoding is None:
447 with _tokenizer_lock:
448 if _tokenizer_encoding is None:
449 if not os.environ.get("TIKTOKEN_CACHE_DIR"):
450 os.environ["TIKTOKEN_CACHE_DIR"] = str(_TIKTOKEN_CACHE_DIR)
451 from tiktoken import get_encoding
453 _tokenizer_encoding = get_encoding(_TIKTOKEN_ENCODING)
454 return _tokenizer_encoding
457def count_context_tokens(messages: list[dict[str, str]]) -> int:
458 """Kootun kontekstin koko tokeneina: system + user -viestien sisältöjen
459 summa. Deterministinen, ei riipu providerista eikä kutsutusta mallista."""
460 enc = _tokenizer()
461 total = 0
462 for message in messages:
463 content = message.get("content")
464 if isinstance(content, str):
465 total += len(enc.encode_ordinary(content))
466 return total
469# ---------------------------------------------------------------------------
470# Langfuse trace helpers
471# ---------------------------------------------------------------------------
473_PARENT_STACK: contextvars.ContextVar[tuple[str, ...]] = contextvars.ContextVar(
474 "langfuse_parent_stack", default=()
475)
478class LangfuseTracer:
479 """
480 Per-ajo Langfuse-tracen alustus + generation-spanit per LLM-kutsu.
481 Käyttää singleton-Langfuse-clientiä env-muuttujista.
482 """
484 def __init__(self, project_slug: str, pipeline_name: str, run_id: str):
485 self.project_slug = project_slug
486 self.pipeline_name = pipeline_name
487 self.run_id = run_id or "unknown"
488 self._lf: Any = None
489 self._trace_ctx: dict[str, Any] | None = None
490 self._flush_counter = 0
491 self._init_done = False
493 def _client(self) -> Any:
494 if self._lf is not None:
495 return self._lf
496 try:
497 from langfuse import Langfuse # type: ignore
498 except ImportError:
499 log.warning("[langfuse] package not installed — tracing disabled")
500 return None
501 prev = os.environ.get("LANGFUSE_TRACING_ENABLED")
502 os.environ["LANGFUSE_TRACING_ENABLED"] = "true"
503 try:
504 self._lf = Langfuse()
505 except Exception as exc: # noqa: BLE001
506 log.warning("[langfuse] init failed: %s — tracing disabled", exc)
507 self._lf = None
508 finally:
509 if prev is None:
510 os.environ["LANGFUSE_TRACING_ENABLED"] = "false"
511 else:
512 os.environ["LANGFUSE_TRACING_ENABLED"] = prev
513 return self._lf
515 @property
516 def trace_name(self) -> str:
517 return f"{self.project_slug}:{self.pipeline_name}"
519 @property
520 def session_id(self) -> str:
521 return f"{self.project_slug}-{self.pipeline_name}"
523 def _ctx(self) -> dict[str, Any] | None:
524 lf = self._client()
525 if lf is None:
526 return None
527 if self._trace_ctx is None:
528 self._trace_ctx = {"trace_id": lf.create_trace_id(seed=self.run_id)}
529 return self._trace_ctx
531 def _ctx_with_parent(self) -> dict[str, Any] | None:
532 ctx = self._ctx()
533 if ctx is None:
534 return None
535 stack = _PARENT_STACK.get()
536 if stack:
537 return {**ctx, "parent_span_id": stack[-1]}
538 return ctx
540 def open_parent_observation(
541 self,
542 *,
543 name: str,
544 input: dict[str, Any] | None = None,
545 metadata: dict[str, Any] | None = None,
546 ) -> Any:
547 lf = self._client()
548 ctx = self._ctx_with_parent()
549 if lf is None or ctx is None:
550 return _NoopObservation()
551 try:
552 obs = lf.start_observation(
553 trace_context=ctx,
554 name=name,
555 as_type="span",
556 input=input or {},
557 metadata=metadata or {},
558 )
559 except Exception as exc: # noqa: BLE001
560 log.warning("[langfuse] open_parent_observation failed: %s", exc)
561 return _NoopObservation()
562 stack = _PARENT_STACK.get()
563 _PARENT_STACK.set((*stack, obs.id))
564 return obs
566 def close_parent_observation(self, obs: Any) -> None:
567 if isinstance(obs, _NoopObservation):
568 return
569 stack = _PARENT_STACK.get()
570 if stack and stack[-1] == obs.id:
571 _PARENT_STACK.set(stack[:-1])
572 obs.end()
574 def init_trace(self) -> None:
575 if self._init_done:
576 return
577 lf = self._client()
578 ctx = self._ctx()
579 if lf is None or ctx is None:
580 self._init_done = True
581 return
582 try:
583 obs = lf.start_observation(
584 trace_context=ctx,
585 name=self.trace_name,
586 as_type="span",
587 input={
588 "pipeline": self.pipeline_name,
589 "project": self.project_slug,
590 "run_id": self.run_id,
591 },
592 )
593 self._apply_meta(obs)
594 obs.end()
595 self.flush(force=True)
596 except Exception as exc: # noqa: BLE001
597 log.warning("[langfuse] init_trace failed: %s", exc)
598 self._init_done = True
600 def _apply_meta(self, obs: Any) -> None:
601 try:
602 obs.update_trace(
603 name=self.trace_name,
604 user_id=self.project_slug,
605 session_id=self.session_id,
606 tags=[f"pipeline:{self.pipeline_name}"],
607 )
608 except Exception: # noqa: BLE001
609 pass
611 def generation(
612 self,
613 name: str,
614 *,
615 model: str,
616 temperature: float,
617 messages: list[dict[str, str]],
618 metadata: dict[str, Any] | None = None,
619 ) -> Any:
620 lf = self._client()
621 ctx = self._ctx()
622 if lf is None or ctx is None:
623 return _NoopObservation()
624 try:
625 obs = lf.start_observation(
626 trace_context=self._ctx_with_parent(),
627 name=name,
628 as_type="generation",
629 model=model,
630 model_parameters={"temperature": temperature},
631 input=messages,
632 metadata=metadata,
633 )
634 self._apply_meta(obs)
635 return obs
636 except Exception as exc: # noqa: BLE001
637 log.warning("[langfuse] generation failed: %s", exc)
638 return _NoopObservation()
640 def flush(self, *, force: bool = False) -> None:
641 lf = self._client()
642 if lf is None:
643 return
644 self._flush_counter += 1
645 if force or self._flush_counter % _LANGFUSE_FLUSH_EVERY == 0:
646 try:
647 lf.flush()
648 except Exception: # noqa: BLE001
649 pass
652class _NoopObservation:
653 def update(self, **_: Any) -> None: ...
654 def end(self) -> None: ...
657# ---------------------------------------------------------------------------
658# JSON output parsing — LLM-vastauksen JSON-rakenteiden jäsennys
659# ---------------------------------------------------------------------------
662def _try_parse_json(text: str) -> tuple[Any, str | None]:
663 """Yritä parsia JSON. Palauttaa (data, error). data=None jos epäonnistui."""
664 if not text or not text.strip():
665 return None, "empty"
666 raw = text.strip()
667 fence = re.search(r"```(?:json)?\s*([\s\S]*?)```", raw)
668 if fence:
669 raw = fence.group(1).strip()
670 else:
671 m = re.search(r"\{[\s\S]*\}|\[[\s\S]*\]", raw)
672 if m:
673 raw = m.group(0)
674 try:
675 return json.loads(raw), None
676 except json.JSONDecodeError as e:
677 # Yritys: korjaa katkennut JSON
678 repaired = raw
679 if repaired.count('"') % 2 == 1:
680 repaired = repaired.rsplit('"', 1)[0] + '"'
681 repaired = re.sub(r",\s*$", "", repaired.rstrip())
682 for suffix in ("", "}", "]}", "\"]}"):
683 try:
684 return json.loads(repaired + suffix), None
685 except json.JSONDecodeError:
686 continue
687 return None, str(e)
690# ---------------------------------------------------------------------------
691# File-access tool (agent spec.file_access: true)
692# ---------------------------------------------------------------------------
695@dataclass
696class ReadFileResult:
697 content: str
698 truncated: bool
699 error: str | None
701 def as_tool_message(self) -> str:
702 payload: dict[str, Any] = {
703 "name": _READ_FILE_TOOL_NAME,
704 "ok": self.error is None,
705 "content": self.content,
706 "truncated": self.truncated,
707 }
708 if self.error is not None:
709 payload["error"] = self.error
710 return json.dumps(payload, ensure_ascii=False)
713_READ_FILE_TOOL: dict[str, Any] = {
714 "type": "function",
715 "function": {
716 "name": _READ_FILE_TOOL_NAME,
717 "description": (
718 "Read a text file from the project root. The path is resolved "
719 "against the project root; any read that resolves outside the "
720 "project root is rejected. Content is truncated and flagged when "
721 "it exceeds the caps, and can be windowed by offset/limit (lines)."
722 ),
723 "parameters": {
724 "type": "object",
725 "properties": {
726 "path": {
727 "type": "string",
728 "description": "Path to the file, relative to the project root.",
729 },
730 "offset": {
731 "type": "integer",
732 "description": "Number of lines to skip (0-based). Optional.",
733 },
734 "limit": {
735 "type": "integer",
736 "description": "Max number of lines to return. Optional.",
737 },
738 },
739 "required": ["path"],
740 "additionalProperties": False,
741 },
742 },
743}
746def _is_within(root: Path, candidate: Path) -> bool:
747 return candidate == root or root in candidate.parents
750def _read_file(
751 project_root: Path,
752 path: str,
753 offset: int | None,
754 limit: int | None,
755) -> ReadFileResult:
756 """Realpath-bound read of a text file inside the project root (ADR-0012).
758 Every rejected read returns a clear error — no silent fallback. Truncation
759 over `_READ_FILE_MAX_BYTES` / `_READ_FILE_MAX_LINES` is explicit via
760 `truncated`.
761 """
762 if not path or not str(path).strip():
763 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: 'path' is required (ADR-0012)")
764 try:
765 root = project_root.resolve(strict=False)
766 except OSError as exc:
767 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot resolve project root: {exc}")
768 try:
769 candidate = (root / path).resolve(strict=False)
770 except OSError as exc:
771 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot resolve '{path}': {exc}")
772 if not _is_within(root, candidate):
773 return ReadFileResult(
774 "",
775 False,
776 f"{_READ_FILE_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
777 )
778 if not candidate.is_file():
779 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: '{path}' is not a file or does not exist")
780 try:
781 text = candidate.read_text(encoding="utf-8")
782 except (OSError, UnicodeDecodeError) as exc:
783 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: cannot read '{path}': {exc}")
785 lines = text.splitlines()
786 total_lines = len(lines)
787 try:
788 start = int(offset) if offset is not None and int(offset) > 0 else 0
789 except (TypeError, ValueError):
790 start = 0
791 if limit is not None:
792 # A requested window is not an engine cap — no line-level truncation flag.
793 try:
794 max_lines = int(limit)
795 except (TypeError, ValueError):
796 max_lines = _READ_FILE_MAX_LINES
797 truncated = False
798 end = start + max_lines if max_lines >= 0 else total_lines
799 else:
800 max_lines = _READ_FILE_MAX_LINES
801 truncated = (start + max_lines) < total_lines
802 end = start + max_lines
803 content = "\n".join(lines[start:end])
805 if len(content.encode("utf-8")) > _READ_FILE_MAX_BYTES:
806 content = content[:_READ_FILE_MAX_BYTES]
807 truncated = True
808 return ReadFileResult(content, truncated, None)
811@dataclass
812class ListDirEntry:
813 path: str
814 name: str
815 size: int | None
816 kind: str # "dir" | "file"
819@dataclass
820class ListDirResult:
821 entries: list[ListDirEntry]
822 truncated: bool
823 root: str
824 error: str | None
826 def as_tool_message(self) -> str:
827 payload: dict[str, Any] = {
828 "name": _LIST_DIR_TOOL_NAME,
829 "ok": self.error is None,
830 "entries": [
831 {"path": e.path, "name": e.name, "size": e.size, "kind": e.kind}
832 for e in self.entries
833 ],
834 "truncated": self.truncated,
835 "root": self.root,
836 }
837 if self.error is not None:
838 payload["error"] = self.error
839 return json.dumps(payload, ensure_ascii=False)
842_LIST_DIR_TOOL: dict[str, Any] = {
843 "type": "function",
844 "function": {
845 "name": _LIST_DIR_TOOL_NAME,
846 "description": (
847 "List ONE level of a directory inside the project root. The path is "
848 "resolved against the project root; any listing that resolves "
849 "outside the project root is rejected. Non-recursive: nested "
850 "subdirectories are not expanded. Only structural skip dirs are "
851 "omitted. The listing is capped and flagged when it exceeds the cap."
852 ),
853 "parameters": {
854 "type": "object",
855 "properties": {
856 "path": {
857 "type": "string",
858 "description": "Directory path, relative to the project root.",
859 },
860 },
861 "required": ["path"],
862 "additionalProperties": False,
863 },
864 },
865}
868def _execute_list_dir_tool(project_root: Path, path: str) -> ListDirResult:
869 """Realpath-bound ONE-level directory listing inside the project root
870 (ADR-0012). Every rejected listing returns a clear error — no silent
871 fallback. Only structural skip dirs are omitted; entries over
872 `_LIST_DIR_MAX_ENTRIES` are capped and flagged via `truncated`.
873 """
874 if not path or not str(path).strip():
875 return ListDirResult(
876 [], False, "", f"{_LIST_DIR_TOOL_NAME}: 'path' is required (ADR-0012)"
877 )
878 try:
879 root = project_root.resolve(strict=False)
880 except OSError as exc:
881 return ListDirResult(
882 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot resolve project root: {exc}"
883 )
884 try:
885 candidate = (root / str(path)).resolve(strict=False)
886 except OSError as exc:
887 return ListDirResult(
888 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot resolve '{path}': {exc}"
889 )
890 if not _is_within(root, candidate):
891 return ListDirResult(
892 [],
893 False,
894 "",
895 f"{_LIST_DIR_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
896 )
897 if not candidate.is_dir():
898 return ListDirResult(
899 [],
900 False,
901 "",
902 f"{_LIST_DIR_TOOL_NAME}: '{path}' is not a directory or does not exist",
903 )
904 try:
905 children = sorted(candidate.iterdir(), key=lambda p: p.name)
906 except OSError as exc:
907 return ListDirResult(
908 [], False, "", f"{_LIST_DIR_TOOL_NAME}: cannot list '{path}': {exc}"
909 )
911 entries: list[ListDirEntry] = []
912 for child in children:
913 if child.name in _DEFAULT_SKIP_DIRS:
914 continue
915 is_dir = child.is_dir()
916 size: int | None = None
917 if not is_dir:
918 try:
919 size = child.stat().st_size
920 except OSError:
921 size = None
922 entries.append(
923 ListDirEntry(
924 path=str(child.relative_to(root)),
925 name=child.name,
926 size=size,
927 kind="dir" if is_dir else "file",
928 )
929 )
930 truncated = len(entries) > _LIST_DIR_MAX_ENTRIES
931 if truncated:
932 entries = entries[:_LIST_DIR_MAX_ENTRIES]
933 return ListDirResult(entries, truncated, str(root), None)
936# ---------------------------------------------------------------------------
937# OpenAIInvoker
938# ---------------------------------------------------------------------------
941class OpenAIInvoker:
942 """
943 Default LLM invoker for agent-platform pipelines.
945 Yksi instanssi per pipeline-ajo. Thread-safe (semafori rinnakkaisuudelle).
946 """
948 def __init__(
949 self,
950 *,
951 providers: dict[str, dict[str, Any]],
952 max_parallel: int,
953 budget: RunBudget,
954 tracer: LangfuseTracer,
955 ) -> None:
956 self._providers = providers
957 self._clients: dict[str, Any] = {}
958 self.budget = budget
959 self.tracer = tracer
960 self._semaphore = threading.Semaphore(max(1, max_parallel))
962 def _openai_client(self, base_url: str, api_key: str) -> Any:
963 key = f"{base_url}|{api_key}"
964 if key not in self._clients:
965 from openai import OpenAI # type: ignore
966 self._clients[key] = OpenAI(base_url=base_url, api_key=api_key)
967 return self._clients[key]
969 # ---- Public callable -------------------------------------------------
971 def __call__(
972 self,
973 resolved_spec: dict,
974 resolved_step: dict,
975 state: dict,
976 ) -> dict:
977 """
978 StepRunner-yhteensopiva: (spec, step, state) → uusi state.
980 `resolved_step` = step jossa params/input on JO resolvoitu
981 StepRunnerissa template-engineä käyttäen. Tämä invoker ei enää
982 evaluoi `{{ }}`-lausekkeita state/run_context-vastaan, vain
983 substituoi `step.params` ja sisäänrakennetut feature-arvot
984 agentin `system_prompt`/`user`-mallineisiin.
985 """
986 step_id = resolved_step.get("id", "?")
987 agent_spec = resolved_spec.get("spec") or {}
989 model_spec = agent_spec.get("model") or {}
990 provider_name = model_spec.get("provider")
992 if not provider_name:
993 if self._providers:
994 provider_name = next(iter(self._providers))
995 else:
996 raise ValueError("No LLM provider configured and no provider specified in agent spec")
998 provider_cfg = self._providers.get(provider_name)
999 if provider_cfg is None:
1000 raise ValueError(
1001 f"Unknown LLM provider '{provider_name}'. "
1002 f"Available: {', '.join(sorted(self._providers.keys()))}"
1003 )
1005 model_override = model_spec.get("model")
1006 model = model_override if model_override else provider_cfg["model"]
1008 # Budget check
1009 if not self.budget.try_consume():
1010 return {
1011 **state,
1012 step_id: _with_failure_flags(
1013 {"error": f"budget:{self.budget.stop_reason}"}
1014 ),
1015 }
1017 # ---- Build system + user messages --------------------------------
1018 skills_text = load_skills(resolved_spec)
1019 step_params = resolved_step.get("params") or {}
1020 input_value = resolved_step.get("input")
1022 # Auto-feature: path verification block (filled by builtin path-verifier
1023 # earlier in pipeline, or empty string if not present).
1024 if "path_verification_block" not in step_params:
1025 step_params = {**step_params, "path_verification_block": ""}
1027 # Render system_prompt template with skills + params
1028 prompt_ctx: dict[str, Any] = {
1029 "skills": skills_text,
1030 **{k: v for k, v in step_params.items() if isinstance(k, str)},
1031 }
1032 system_text = _render(
1033 str(agent_spec.get("system_prompt") or ""),
1034 prompt_ctx,
1035 ).strip()
1037 # User message: serialise input + step params
1038 user_text = _format_user_message(input_value, step_params)
1040 # Temperature: agent spec → provider default → step params
1041 temperature = float(
1042 model_spec.get("temperature", provider_cfg.get("temperature", 0.0))
1043 )
1044 if "temperature" in step_params:
1045 try:
1046 temperature = float(step_params["temperature"])
1047 except (TypeError, ValueError):
1048 pass
1050 messages = [
1051 {"role": "system", "content": system_text},
1052 {"role": "user", "content": user_text},
1053 ]
1055 # ---- Context-integrity check (Phase 4 doc, ## Kontekstin eheys) --
1056 agent_name = (resolved_spec.get("metadata") or {}).get("name", "agent")
1057 context_budget = _effective_context_budget(resolved_spec, agent_name)
1058 context_tokens = count_context_tokens(messages)
1059 if context_tokens > context_budget:
1060 return _with_context_metrics(
1061 state,
1062 step_id,
1063 _context_overflow_result(step_id, context_tokens, context_budget),
1064 {"context_size_tokens": context_tokens, "context_budget_tokens": context_budget},
1065 )
1066 measurement = {
1067 "context_size_tokens": context_tokens,
1068 "context_budget_tokens": context_budget,
1069 }
1071 # ---- LLM call with Langfuse generation ---------------------------
1072 gen_name = f"{step_id}:{agent_name}"
1074 with self._semaphore:
1075 obs = self.tracer.generation(
1076 gen_name,
1077 model=model,
1078 temperature=temperature,
1079 messages=messages,
1080 metadata=measurement,
1081 )
1082 content = ""
1083 usage: dict[str, int] | None = None
1084 err: str | None = None
1085 tool_meta: dict[str, Any] | None = None
1086 try:
1087 if bool(agent_spec.get("file_access")):
1088 content, usage, tool_meta = self._stream_completion_with_tools(
1089 messages,
1090 temperature,
1091 model,
1092 provider_cfg["base_url"],
1093 provider_cfg["api_key"],
1094 project_root=resolve_workspace_root(state),
1095 )
1096 else:
1097 content, usage = self._stream_completion(messages, temperature, model, provider_cfg["base_url"], provider_cfg["api_key"])
1098 except Exception as exc: # noqa: BLE001
1099 err = f"{type(exc).__name__}: {exc}"
1100 log.warning("[invoke] %s LLM call failed: %s", step_id, err)
1101 try:
1102 obs.update(
1103 output=content[:8000] if len(content) > 8000 else content,
1104 usage_details=usage,
1105 )
1106 except Exception: # noqa: BLE001
1107 pass
1108 try:
1109 obs.end()
1110 except Exception: # noqa: BLE001
1111 pass
1112 self.tracer.flush()
1114 # ---- Parse output ------------------------------------------------
1115 if err:
1116 return _with_context_metrics(
1117 state,
1118 step_id,
1119 _with_failure_flags({"error": err, "raw": content}),
1120 measurement,
1121 )
1123 output_type = ((agent_spec.get("output") or {}).get("type") or "").lower()
1124 if output_type == "files":
1125 return _handle_files_output(
1126 state, step_id, content, measurement, agent_spec, tool_meta
1127 )
1129 output_format = ((agent_spec.get("output") or {}).get("format") or "").lower()
1130 if output_format == "markdown":
1131 return _with_context_metrics(
1132 state,
1133 step_id,
1134 _apply_tool_meta({"output": content.strip()}, tool_meta),
1135 measurement,
1136 )
1138 parsed, perr = _try_parse_json(content)
1139 if parsed is not None:
1140 return _with_context_metrics(
1141 state, step_id, _apply_tool_meta(parsed, tool_meta), measurement
1142 )
1143 # Fallback: raw text
1144 return _with_context_metrics(
1145 state,
1146 step_id,
1147 _apply_tool_meta(
1148 _with_failure_flags({"output": content.strip(), "parse_error": perr}),
1149 tool_meta,
1150 ),
1151 measurement,
1152 )
1154 # ---- Streaming completion -------------------------------------------
1156 def _stream_completion(
1157 self,
1158 messages: list[dict[str, str]],
1159 temperature: float,
1160 model: str,
1161 base_url: str,
1162 api_key: str,
1163 ) -> tuple[str, dict[str, int] | None]:
1164 client = self._openai_client(base_url, api_key)
1165 resp = client.chat.completions.create(
1166 model=model,
1167 temperature=temperature,
1168 messages=messages,
1169 stream=True,
1170 stream_options={"include_usage": True},
1171 )
1172 content = ""
1173 usage: dict[str, int] | None = None
1174 for chunk in resp:
1175 if chunk.choices and chunk.choices[0].delta.content:
1176 content += chunk.choices[0].delta.content
1177 if getattr(chunk, "usage", None):
1178 u = chunk.usage
1179 usage = {
1180 "input": u.prompt_tokens or 0,
1181 "output": u.completion_tokens or 0,
1182 "total": u.total_tokens or 0,
1183 }
1184 return content, usage
1186 def _execute_file_tool(
1187 self, tool_call: Any, project_root: Path
1188 ) -> ReadFileResult | ListDirResult:
1189 """Execute a read_file or list_dir tool call against the project root."""
1190 name = getattr(tool_call.function, "name", "")
1191 try:
1192 args = json.loads(getattr(tool_call.function, "arguments", "{}"))
1193 except (TypeError, json.JSONDecodeError) as exc:
1194 if name == _LIST_DIR_TOOL_NAME:
1195 return ListDirResult([], False, "", f"{_LIST_DIR_TOOL_NAME}: invalid arguments: {exc}")
1196 return ReadFileResult("", False, f"{_READ_FILE_TOOL_NAME}: invalid arguments: {exc}")
1197 if name == _LIST_DIR_TOOL_NAME:
1198 return _execute_list_dir_tool(project_root, args.get("path"))
1199 return _read_file(project_root, args.get("path"), args.get("offset"), args.get("limit"))
1201 # ---- Tool-looping completion (file_access: true) --------------------
1203 def _stream_completion_with_tools(
1204 self,
1205 messages: list[dict[str, str]],
1206 temperature: float,
1207 model: str,
1208 base_url: str,
1209 api_key: str,
1210 *,
1211 project_root: Path,
1212 ) -> tuple[str, dict[str, int] | None, dict[str, Any]]:
1213 """Non-streaming completion carrying the read_file tool loop.
1215 Each tool round consumes ONE additional inference through
1216 `self.budget.try_consume`; the loop is hard-capped at `_MAX_TOOL_ROUNDS`
1217 and stops instead of looping indefinitely. The non-flagged path never
1218 reaches here.
1219 """
1220 client = self._openai_client(base_url, api_key)
1221 working = list(messages)
1222 content = ""
1223 usage: dict[str, int] | None = None
1224 tool_rounds = 0
1225 truncated = False
1226 tool_stopped = False
1227 tool_errors: list[str] = []
1229 while True:
1230 resp = client.chat.completions.create(
1231 model=model,
1232 temperature=temperature,
1233 messages=working,
1234 tools=[_READ_FILE_TOOL, _LIST_DIR_TOOL],
1235 stream=False,
1236 )
1237 if getattr(resp, "usage", None):
1238 u = resp.usage
1239 usage = {
1240 "input": getattr(u, "prompt_tokens", 0) or 0,
1241 "output": getattr(u, "completion_tokens", 0) or 0,
1242 "total": getattr(u, "total_tokens", 0) or 0,
1243 }
1244 msg = resp.choices[0].message
1245 piece = getattr(msg, "content", None) or ""
1246 if piece:
1247 content += piece
1248 tool_calls = getattr(msg, "tool_calls", None)
1249 if not tool_calls:
1250 break
1251 working.append(
1252 {
1253 "role": "assistant",
1254 "content": piece or "",
1255 "tool_calls": [
1256 {
1257 "id": tc.id,
1258 "type": "function",
1259 "function": {"name": tc.function.name, "arguments": tc.function.arguments},
1260 }
1261 for tc in tool_calls
1262 ],
1263 }
1264 )
1265 for tc in tool_calls:
1266 result = self._execute_file_tool(tc, project_root)
1267 truncated = truncated or result.truncated
1268 if result.error:
1269 tool_errors.append(result.error)
1270 working.append(
1271 {"role": "tool", "tool_call_id": tc.id, "content": result.as_tool_message()}
1272 )
1273 tool_rounds += 1
1274 if tool_rounds >= _MAX_TOOL_ROUNDS:
1275 tool_stopped = True
1276 break
1277 if not self.budget.try_consume():
1278 tool_stopped = True
1279 break
1281 tool_meta: dict[str, Any] = {
1282 "tool_rounds": tool_rounds,
1283 "truncated": truncated,
1284 "tool_stopped": tool_stopped,
1285 "tool_errors": tool_errors,
1286 }
1287 return content, usage, tool_meta
1290# ---------------------------------------------------------------------------
1291# Helpers
1292# ---------------------------------------------------------------------------
1295def _effective_context_budget(resolved_spec: dict, agent_name: str) -> int:
1296 """Efektiivinen kontekstibudjetti: spec.limits.context_budget (merge-
1297 järjestyksen tulos) tai oletus 100 000 tokenia. Epäkelpo arvo on kova
1298 virhe (ADR-0012) — ei hiljaista fallbackia."""
1299 limits = (resolved_spec.get("spec") or {}).get("limits") or {}
1300 value = limits.get("context_budget")
1301 if value is None:
1302 return DEFAULT_CONTEXT_BUDGET_TOKENS
1303 if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
1304 raise ValueError(
1305 f"agent '{agent_name}': spec.limits.context_budget must be a "
1306 f"positive integer, got {value!r} (ADR-0012)"
1307 )
1308 return value
1311def _context_overflow_result(step_id: str, context_tokens: int, context_budget: int) -> dict:
1312 """Budjetin ylitys → kova virhe ennen mallikutsua (fail-fast)."""
1313 return {
1314 "error": (
1315 f"{_CONTEXT_BUDGET_ERROR_CODE}: step '{step_id}' assembled "
1316 f"context is {context_tokens} tokens, exceeding the context "
1317 f"budget of {context_budget} tokens"
1318 ),
1319 "details": {
1320 "error_code": _CONTEXT_BUDGET_ERROR_CODE,
1321 "step_id": step_id,
1322 "assembled_tokens": context_tokens,
1323 "context_budget": context_budget,
1324 },
1325 }
1328def _handle_files_output(
1329 state: dict,
1330 step_id: str,
1331 content: str,
1332 measurement: dict[str, int],
1333 agent_spec: dict,
1334 tool_meta: dict[str, Any] | None = None,
1335) -> dict:
1336 """Agentin tiedostotuloste (output.type: files): parsii {path, content}
1337 -listan ja kirjoittaa sen workspaceen jaetun helperin kautta (atominen
1338 validointi, fail-fast, ADR-0012). Parsittu lista jää state[step_id]
1339 -arvoksi — sama semantiikka kuin JSON-tuloksella."""
1340 parsed, perr = _try_parse_json(content)
1341 if parsed is None:
1342 return _with_context_metrics(
1343 state,
1344 step_id,
1345 {
1346 "error": (
1347 f"{_FILES_OUTPUT_ERROR_CODE}: step '{step_id}' files output "
1348 f"is not valid JSON ({perr})"
1349 ),
1350 "details": {
1351 "error_code": _FILES_OUTPUT_ERROR_CODE,
1352 "step_id": step_id,
1353 },
1354 },
1355 measurement,
1356 )
1357 try:
1358 validate_and_write_files(
1359 parsed,
1360 workspace_root=resolve_workspace_root(state),
1361 target_path=(agent_spec.get("output") or {}).get("path"),
1362 )
1363 except FilesOutputError as exc:
1364 return _with_context_metrics(
1365 state,
1366 step_id,
1367 {
1368 "error": f"{_FILES_OUTPUT_ERROR_CODE}: {exc}",
1369 "details": {
1370 "error_code": _FILES_OUTPUT_ERROR_CODE,
1371 "step_id": step_id,
1372 },
1373 },
1374 measurement,
1375 )
1376 return _with_context_metrics(
1377 state, step_id, _apply_tool_meta(parsed, tool_meta), measurement
1378 )
1381def _apply_tool_meta(value: Any, tool_meta: dict[str, Any] | None) -> Any:
1382 """Attach tool-loop metadata onto a dict-valued step result.
1384 The non-flagged path passes `tool_meta=None` and returns `value` unchanged
1385 (byte-identical). List-valued results (e.g. files output) are left as-is so
1386 the shape contract is preserved; tool_meta is simply not surfaced there.
1387 """
1388 if not tool_meta:
1389 return value
1390 if isinstance(value, dict):
1391 return {**value, "tool_meta": tool_meta}
1392 return value
1395def _with_failure_flags(value: dict) -> dict:
1396 """Tag an LLM-failure step result with the keys strict condition gates read.
1398 A failing step result carries `escalate`/`verdict` so a strict gate like
1399 `{{ research.escalate }}` resolves instead of raising an unresolved-
1400 reference ValueError, and the failure routes to the escalation/qb surface.
1401 Returns a new dict; the caller's `error`/`raw`/`parse_error` are preserved.
1402 """
1403 return {**value, "escalate": True, "verdict": "ERROR"}
1406def _with_context_metrics(
1407 state: dict,
1408 step_id: str,
1409 step_value: Any,
1410 measurement: dict[str, int],
1411) -> dict:
1412 """Kirjoita step-tulos state[step_id]-polkuun sellaisenaan — sama arvo,
1413 polku ja tyyppi kuin ennen Phase 5:tä, ei mittausavaimia sisällä (Phase
1414 5b) — ja kontekstimittaus rinnakkaiseen step_context_metrics-envelope-
1415 avaimen alle (step_id → {context_size_tokens, context_budget_tokens})."""
1416 metrics = dict(state.get("step_context_metrics") or {})
1417 metrics[step_id] = measurement
1418 return {**state, step_id: step_value, "step_context_metrics": metrics}
1421def _format_user_message(input_value: Any, params: dict[str, Any]) -> str:
1422 """
1423 Muotoile agentin user-message resolvoidusta inputista + paramseista.
1424 """
1425 parts: list[str] = []
1427 # Inputin esitysmuoto: yritetään olla informatiivisia
1428 if input_value is None:
1429 pass
1430 elif isinstance(input_value, str):
1431 parts.append(input_value)
1432 elif isinstance(input_value, dict):
1433 # Doc-file form: {path, content, ...} → human-readable
1434 if "path" in input_value and "content" in input_value:
1435 path = input_value.get("path", "")
1436 content = input_value.get("content", "")
1437 parts.append(f"Path: {path}\n\n---\n{content}")
1438 else:
1439 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
1440 elif isinstance(input_value, list):
1441 parts.append(json.dumps(input_value, ensure_ascii=False, indent=2))
1442 else:
1443 parts.append(str(input_value))
1445 # Append path verification block (auto-feature) jos asetettu
1446 pvb = params.get("path_verification_block") or ""
1447 if pvb:
1448 parts.append(str(pvb))
1450 # Lisää muut paramit (focus, finding jne.) jos eivät jo ole system_promptissa
1451 extra: dict[str, Any] = {}
1452 skip_keys = {"path_verification_block", "skills"}
1453 for k, v in params.items():
1454 if k in skip_keys:
1455 continue
1456 # Älä toista pieniä skalaareja - ne yleensä jo system_promptissa
1457 if isinstance(v, (dict, list)):
1458 extra[k] = v
1459 if extra:
1460 parts.append("## Additional context\n\n" + json.dumps(extra, ensure_ascii=False, indent=2))
1462 return "\n\n".join(p for p in parts if p)