| """๊ธฐ์ฌ ํธ์ถ์ capture ์ ์ฑ
๊ณผ process/run/stage ์ฐธ์กฐ ์์ ๊ถ์ ๊ตฌ๋ถํ๋ค. |
| |
| ๊ธฐ์กด raw result์ ์๋ฏธ ํ๋จ์ ์ด ๋ชจ๋์ด ๋ฐ๊พธ์ง ์๋๋ค. ์ง๋จ ๊ธฐ๋ก์ ๊ฐ์ผ๋ก |
| ๋ถ๋ฆฌํ๋ฉฐ, stage/run scope๋ ๋ง์ง๋ง ์๋น ๋ค ์์ ์ด ์์ ํ ์ฐธ์กฐ๋ฅผ ํด์ ํ๋ค. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from collections import Counter |
| from dataclasses import dataclass, field |
| from enum import Enum |
| import json |
| import math |
| from pathlib import Path |
| from typing import Any, Mapping |
|
|
|
|
| class CaptureLevel(str, Enum): |
| SUMMARY = "SUMMARY" |
| DECISIONS = "DECISIONS" |
| FULL = "FULL" |
|
|
|
|
| def _independent(value: Any) -> Any: |
| """์ ๊ฐ์ฒด๋ฅผ ์ก์๋์ง ์๋ JSON ๊ฐ๋ง ๋ณต์ฌํ๋ค. Tensor/์์ ๊ฐ์ฒด๋ ๊ฑฐ๋ถํ๋ค.""" |
|
|
| if value is None or isinstance(value, (str, bool, int)): |
| return value |
| if isinstance(value, float): |
| if not math.isfinite(value): |
| raise ValueError("diagnostic record must contain finite numbers") |
| return value |
| if isinstance(value, Mapping): |
| return {str(key): _independent(item) for key, item in value.items()} |
| if isinstance(value, (list, tuple)): |
| return [_independent(item) for item in value] |
| raise TypeError(f"diagnostic record cannot retain {type(value).__name__}") |
|
|
|
|
| def _stage_summary(row: Mapping[str, Any]) -> dict[str, Any]: |
| warnings = row.get("warnings", ()) |
| return { |
| "stage": str(row.get("stage", "UNKNOWN")), |
| "component": str(row.get("component", "UNKNOWN")), |
| "input_count": int(row.get("input_count", 0)), |
| "candidate_count": _independent(row.get("candidate_count", 0)), |
| "output_count": _independent(row.get("output_count", 0)), |
| "drop_reason_counts": _independent(row.get("drop_reason_counts", {})), |
| "warning_count": len(warnings) if isinstance(warnings, (list, tuple)) else int(bool(warnings)), |
| "elapsed_seconds": float(row.get("elapsed_seconds", 0.0)), |
| } |
|
|
|
|
| class DiagnosticSink: |
| """Bounded ๋
๋ฆฝ record sink; full capture๋ JSONL ์์ฐจ ๊ธฐ๋ก์ ๊ถ์ฅํ๋ค.""" |
|
|
| def __init__(self, level: CaptureLevel, *, path: Path | None = None, |
| max_records: int = 512, max_record_bytes: int = 262144) -> None: |
| if max_records <= 0 or max_record_bytes <= 0: |
| raise ValueError("diagnostic limits must be positive") |
| self.level = level |
| self.path = path |
| self.max_records = max_records |
| self.max_record_bytes = max_record_bytes |
| self.records: list[dict[str, Any]] = [] |
| self.stage_summaries: list[dict[str, Any]] = [] |
| self.counts: Counter[str] = Counter() |
| self.truncated_count = 0 |
| self.failed_reason: str | None = None |
| self._stream = None |
| self._opened = False |
|
|
| def record(self, kind: str, value: Mapping[str, Any]) -> None: |
| self.counts[kind] += 1 |
| if kind == "stage": |
| try: |
| self.stage_summaries.append(_stage_summary(value)) |
| except (TypeError, ValueError) as error: |
| self.failed_reason = f"{type(error).__name__}: {error}" |
| self.truncated_count += 1 |
| return |
| if self.level is CaptureLevel.SUMMARY: |
| return |
| if self.level is CaptureLevel.DECISIONS and kind == "candidate": |
| return |
| if len(self.records) + self.counts.get("written_to_file", 0) >= self.max_records: |
| self.truncated_count += 1 |
| return |
| try: |
| detached = {"kind": kind, "record": _independent(value)} |
| encoded = json.dumps(detached, ensure_ascii=False, allow_nan=False, separators=(",", ":")) |
| if len(encoded.encode("utf-8")) > self.max_record_bytes: |
| self.truncated_count += 1 |
| return |
| if self.path is None: |
| self.records.append(detached) |
| else: |
| if self._stream is None: |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| self._stream = self.path.open("x", encoding="utf-8") |
| self._opened = True |
| self._stream.write(encoded + "\n") |
| self.counts["written_to_file"] += 1 |
| except (OSError, TypeError, ValueError) as error: |
| self.failed_reason = f"{type(error).__name__}: {error}" |
| self.truncated_count += 1 |
|
|
| def summary(self) -> dict[str, Any]: |
| if self._stream is not None: |
| try: |
| self._stream.flush() |
| except OSError as error: |
| self.failed_reason = f"{type(error).__name__}: {error}" |
| return { |
| "capture_level": self.level.value, |
| "observed_counts": dict(self.counts), |
| "stored_in_memory": len(self.records), |
| "written_to_file": self.counts.get("written_to_file", 0), |
| "truncated_count": self.truncated_count, |
| "failed_reason": self.failed_reason, |
| "artifact_path": str(self.path) if self.path and self._opened else None, |
| "max_records": self.max_records, |
| "max_record_bytes": self.max_record_bytes, |
| } |
|
|
| def close(self) -> None: |
| if self._stream is not None: |
| try: |
| self._stream.close() |
| except OSError as error: |
| self.failed_reason = f"{type(error).__name__}: {error}" |
| finally: |
| self._stream = None |
|
|
|
|
| class NoOpDiagnosticSink(DiagnosticSink): |
| def __init__(self) -> None: |
| super().__init__(CaptureLevel.SUMMARY) |
|
|
|
|
| class DecisionDiagnosticSink(DiagnosticSink): |
| def __init__(self, *, path: Path | None = None, max_records: int = 512) -> None: |
| super().__init__(CaptureLevel.DECISIONS, path=path, max_records=max_records) |
|
|
|
|
| class FullDiagnosticSink(DiagnosticSink): |
| def __init__(self, *, path: Path | None = None, max_records: int | None = None) -> None: |
| if max_records is None: |
| max_records = 512 if path is None else 100000 |
| super().__init__(CaptureLevel.FULL, path=path, max_records=max_records) |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class ExecutionPolicy: |
| output_profile: str |
| capture_level: CaptureLevel |
| diagnostic_sink: DiagnosticSink |
|
|
| @classmethod |
| def from_request(cls, output_profile: str = "PUBLIC", *, debug_trace: bool = False, |
| diagnostic_sink: DiagnosticSink | None = None, |
| diagnostic_path: Path | None = None) -> "ExecutionPolicy": |
| profile = str(getattr(output_profile, "value", output_profile)) |
| if profile not in ("PUBLIC", "AUDIT", "DEBUG"): |
| raise ValueError(f"unsupported output profile: {profile}") |
| level = (CaptureLevel.FULL if debug_trace or profile == "DEBUG" else |
| CaptureLevel.DECISIONS if profile == "AUDIT" else CaptureLevel.SUMMARY) |
| sink = diagnostic_sink or ( |
| NoOpDiagnosticSink() if level is CaptureLevel.SUMMARY else |
| DecisionDiagnosticSink(path=diagnostic_path) if level is CaptureLevel.DECISIONS else |
| FullDiagnosticSink(path=diagnostic_path) |
| ) |
| if sink.level is not level: |
| raise ValueError("diagnostic sink capture level differs from execution policy") |
| return cls(profile, level, sink) |
|
|
|
|
| class StageScope: |
| """ํ scorer batch์ ์์ tensor/map ์ฐธ์กฐ๋ฅผ ํด๋น stage ๋์ ๋ซ๋๋ค.""" |
|
|
| def __init__(self) -> None: |
| self._owned: list[object] = [] |
|
|
| def own(self, value: Any) -> Any: |
| self._owned.append(value) |
| return value |
|
|
| def close(self) -> None: |
| self._owned.clear() |
|
|
| def __enter__(self) -> "StageScope": |
| return self |
|
|
| def __exit__(self, *_exc: object) -> None: |
| self.close() |
|
|
|
|
| class ArticleRunScope: |
| """ํ article์ ์๋ฌธ view์ request-local ํํ๋ง ์์ ํ๋ค.""" |
|
|
| def __init__(self, article: Any, policy: ExecutionPolicy) -> None: |
| self.article_version_id = str(article.article_version_id) |
| self.source_text: str | None = article.content |
| self.policy = policy |
| self.representations: dict[object, object] = {} |
| self.compact_event_closure: object | None = None |
| self._owned: list[object] = [] |
|
|
| def own(self, value: Any) -> Any: |
| self._owned.append(value) |
| return value |
|
|
| def close(self) -> None: |
| self.representations.clear() |
| self.compact_event_closure = None |
| self._owned.clear() |
| self.source_text = None |
| self.policy.diagnostic_sink.close() |
|
|
| def owner_census(self) -> dict[str, int | bool]: |
| """์์ฒญ ๋ฐํ ํ scope๊ฐ ์ง์ ์์ ํ source/feature ์ฐธ์กฐ ์๋ง ๋ณด๊ณ ํ๋ค.""" |
| return { |
| "source_text_retained": self.source_text is not None, |
| "request_representation_count": len(self.representations), |
| "compact_event_closure_retained": self.compact_event_closure is not None, |
| "owned_object_count": len(self._owned), |
| } |
|
|
| def __enter__(self) -> "ArticleRunScope": |
| return self |
|
|
| def __exit__(self, *_exc: object) -> None: |
| self.close() |
|
|
