From e526c7d77d4275ee9159bfa80634ed9f026e3863 Mon Sep 17 00:00:00 2001 From: Nitsu45 Date: Wed, 14 Oct 2026 14:21:21 +0200 Subject: [PATCH] Add operational telemetry with safe diagnostics and passive monitoring --- README.md | 8 +- config/telemetry.example.toml | 12 + docs/telemetry-v1.md | 162 ++++++ .../analysis/change_request_builder.py | 2 + .../analysis/diagnosis_builder.py | 2 + .../analysis/hypothesis_builder.py | 2 + .../analysis/interpreter.py | 2 + .../analysis/investigator.py | 3 + .../analysis/output_validator.py | 4 + src/viano_bug_analysis/budget/manager.py | 2 + src/viano_bug_analysis/case/builder.py | 2 + src/viano_bug_analysis/evidence/store.py | 2 + .../lifecycle/async_tests.py | 3 + .../lifecycle/checkpoints.py | 10 +- src/viano_bug_analysis/lifecycle/queue.py | 9 +- src/viano_bug_analysis/llm/runner.py | 15 +- .../schemas/service-status-v1.schema.json | 189 +++++++ src/viano_bug_analysis/telemetry/__init__.py | 15 + src/viano_bug_analysis/telemetry/__main__.py | 97 ++++ src/viano_bug_analysis/telemetry/codes.py | 4 + src/viano_bug_analysis/telemetry/config.py | 40 ++ src/viano_bug_analysis/telemetry/core.py | 495 ++++++++++++++++++ .../telemetry/decorators.py | 38 ++ .../telemetry/inspection.py | 164 ++++++ src/viano_bug_analysis/telemetry/safety.py | 118 +++++ tests/test_telemetry.py | 261 +++++++++ 26 files changed, 1655 insertions(+), 6 deletions(-) create mode 100644 config/telemetry.example.toml create mode 100644 docs/telemetry-v1.md create mode 100644 src/viano_bug_analysis/schemas/service-status-v1.schema.json create mode 100644 src/viano_bug_analysis/telemetry/__init__.py create mode 100644 src/viano_bug_analysis/telemetry/__main__.py create mode 100644 src/viano_bug_analysis/telemetry/codes.py create mode 100644 src/viano_bug_analysis/telemetry/config.py create mode 100644 src/viano_bug_analysis/telemetry/core.py create mode 100644 src/viano_bug_analysis/telemetry/decorators.py create mode 100644 src/viano_bug_analysis/telemetry/inspection.py create mode 100644 src/viano_bug_analysis/telemetry/safety.py create mode 100644 tests/test_telemetry.py diff --git a/README.md b/README.md index 7be9498..7dc2bf6 100644 --- a/README.md +++ b/README.md @@ -26,8 +26,8 @@ per-hypothesis budget accounting. An LLM-driven investigation controller connect read-only queries, durable evidence and asynchronous test handoffs. Final evaluation produces traceable diagnoses or human review. Confirmed diagnoses can produce policy-bound Change Requests, with final output validation and atomic local -outbox/review publication. A continuous worker and service-wide telemetry remain -future milestones. +outbox/review publication. Opt-in operational telemetry observes the existing APIs. A continuous worker +remains a future milestone. Expected-behavior resolution and the read-only provider contract: [expected-behavior-v1.md](docs/expected-behavior-v1.md). @@ -61,3 +61,7 @@ Final hypothesis evaluation, traceable diagnoses and controlled human review: Policy-bound coding requests and validated, resumable local output delivery: [change-request-v1.md](docs/change-request-v1.md), with [change request configuration](config/change-request.example.toml). + +Opt-in operational logs, safe error chains, metrics, heartbeat/progress and passive +queue diagnostics: [telemetry-v1.md](docs/telemetry-v1.md), with +[telemetry configuration](config/telemetry.example.toml). diff --git a/config/telemetry.example.toml b/config/telemetry.example.toml new file mode 100644 index 0000000..9dae69c --- /dev/null +++ b/config/telemetry.example.toml @@ -0,0 +1,12 @@ +[telemetry] +heartbeat_seconds = 5 +heartbeat_stale_seconds = 30 +snapshot_interval_seconds = 1 +progress_warning_seconds = 300 +inventory_interval_seconds = 60 +inventory_stale_seconds = 180 +log_max_bytes = 5242880 +log_backups = 5 +max_cases = 1000 +max_frames = 40 +max_snapshot_bytes = 4194304 diff --git a/docs/telemetry-v1.md b/docs/telemetry-v1.md new file mode 100644 index 0000000..fa9d3e0 --- /dev/null +++ b/docs/telemetry-v1.md @@ -0,0 +1,162 @@ +# Operational telemetry v1 + +Telemetry observes existing library execution, not case truth. It never resumes +BLOCKED cases, refills budgets, changes routing, retries a model, edits packages, +executes tools or deletes case artifacts. `docs/architecture.md` is unchanged. +A continuous analysis worker/scheduler remains the next integration milestone. +The observation heartbeat is not a substitute for that worker or evidence that +an analysis is progressing. + +## Enable observations + +```python +from viano_bug_analysis.lifecycle import Workspace, CheckpointStore +from viano_bug_analysis.telemetry import Telemetry, TelemetryConfig +from viano_bug_analysis.analysis import CaseInterpreter + +workspace = Workspace("work") +config = TelemetryConfig.load("config/telemetry.local.toml") +with Telemetry(workspace, config) as telemetry: + store = CheckpointStore(workspace, telemetry=telemetry) + # In an async driver, use existing APIs normally: + # await CaseInterpreter(store, gateway).interpret(case_id, package_id) +``` + +Telemetry is explicitly opt-in. Without an attached running session, existing API +behavior is unchanged. InboxProcessor also accepts telemetry=telemetry. Stores +can be attached after construction with store.telemetry=telemetry. Use one +observation session per workspace; a nonblocking OS lease prevents competing +log rotations/status writers. Concurrent threads and async tasks in that process +have independent, correlated operation spans. There is no global worker lock +across observation or model calls. Stop sessions normally to close files/threads. + +The wrappers cover intake, checkpoints, CaseBuilder, interpretation, hypotheses, +investigation, diagnosis, Change Requests, output validation/publication, +LLM runner, budget/evidence operations and asynchronous test submit/confirmation. +The session records span/parent IDs, operation, duration, known result/error code, +replay marker and hashed case/package identities. Checkpoints supply manifest +state, analysis revision and generation; actual LLM rounds supply usage metadata. +Budget observations retain current verified revision balances, spend, reward and +pending actions. Evidence publication counts exclude replay. Invocation counters +include replay; actual model-round/token counters do not double-count successful +stored calls. All counters start fresh with the observation session; service_id +and start time identify resets. Durable lifetime history remains in the case. + +## Where to inspect + +Service-owned files under runtime/telemetry/: + +- service.jsonl: correlated structured JSON events; rotated by configured byte + size and backup count (default 5 MiB plus five backups). +- status.json: atomically replaced observation status, active operations, last + confirmed progress/error, latest observed cases/budgets and queue inventory. +- metrics.prom: atomically replaced Prometheus text exposition, suitable for a + textfile collector or later monitoring adapter. No HTTP endpoint is opened. + +```powershell +python -m viano_bug_analysis.telemetry status --workspace C:/vba/work +python -m viano_bug_analysis.telemetry errors --workspace C:/vba/work --case-id CASE-ID +python -m viano_bug_analysis.telemetry events --workspace C:/vba/work --limit 100 +python -m viano_bug_analysis.telemetry events --workspace C:/vba/work --source-name incoming.zip +python -m viano_bug_analysis.telemetry metrics --workspace C:/vba/work +python -m viano_bug_analysis.telemetry scan --workspace C:/vba/work +``` + +Commands are passive. scan validates the latest managed case candidates and +checkpoint identity from their manifests; filenames only select service-owned +candidates. It does not perform intake, reconstruct service indices or run any +analysis API. A missing observer status returns NOT_STARTED. Unreadable or unsafe +files return a controlled TELEMETRY_READ and exit 2. status can successfully read +an unhealthy/stopped service (exit 0); inspect its health fields, not exit 0 alone. +errors selects ERROR events; events also includes expected warnings such as +quarantine, provider retries and rejected model output. Quarantine events retain +only hashed source names and the generated local quarantine directory key, so +operators can find validation_error.json without logging the input filename. +Readers tolerate incomplete last records and report corrupt records explicitly; +startup separates an interrupted log tail before appending new records. + +## Liveness, progress and waiting + +An independent heartbeat thread updates every five seconds by default. It does +not wait for model requests or archive inspection. A separate inventory thread +samples every 60 seconds; slow archive checks cannot directly park the heartbeat. +The sample has observed_at/complete/error fields and becomes unknown/stale rather +than pretending an incomplete scan means zero work. Case checkpoint callbacks +immediately update generation/state; inventory merging cannot regress a newer +observed generation. Current output/review/handed-off cases are distinguished +from runnable analysis, delayed retries, human feedback and manual provider waits. +Expired human/manual waits never become automatically runnable. + +observer is LIVE, STOPPED, HEARTBEAT_STALE or NOT_STARTED. Work distinguishes +WORKING, NO_RECENT_PROGRESS, INPUT_PENDING, READY_NOT_PROCESSING, +WAITING_EXTERNAL, IDLE and QUEUE_UNKNOWN. Active operations expose start time, +elapsed time, per-case progress age and a configurable stalled marker. Progress +on another case cannot hide an old active phase with no confirmed checkpoints. +A long phase is a diagnostic warning, not proof of deadlock. Nothing is restarted +or cancelled automatically. Public read_status recomputes heartbeat/inventory age +at read time, so a dead process does not remain LIVE because a stale JSON file +was written previously. + +health distinguishes observation freshness, recent errors, incomplete inventory, +no recent progress and telemetry sink failure. Last errors remain inspectable; +a recent-error warning ages out. Recovered log/snapshot channels can become +healthy again, while loss of observation integrity remains visible. This is an +operational hint, not permission to terminate a worker or alter a case. + +Nested-operation snapshot writes are throttled by snapshot_interval_seconds +(default one second). Top-level phase boundaries, heartbeat and explicit +telemetry.flush() force a snapshot. JSONL events are still emitted immediately. +All queue/progress views are sampled observations, never scheduler authority. + +## Metrics and data quality + +Metrics include operation counts/duration sums and counts, checkpoint counts, +active operations, heartbeat/start/last-progress times, fresh inventory/inbox/case +categories, telemetry failures, published evidence, observed budget units, +actual model rounds, provider failures, reported tokens and estimated monetary +costs. Monetary cost currencies are kept separate; investigation budget is not +money. Unknown token/cost fields and usage of failed provider attempts have +explicit counters. Estimates use configured rates only; they are not invoices. +No token count, cost or model capacity is inferred from response size. + +Do not alert on a stale metrics file's observer_alive=1 alone. Monitor freshness +using time() - vba_observer_heartbeat_time_seconds against your configured stale +threshold. The same issue applies to all textfile collectors after process death. +Unknown/stale inventories omit queue gauges instead of exporting fabricated zero. +Case IDs, provider/model names and arbitrary error text are not metric labels, +which bounds cardinality. Counters reset per observation session; current budget +views carry analysis_revision and must not be treated as lifetime totals. + +## Privacy and failures + +Case/package/source/provider/model identifiers are hashed. Events use fixed +operation/status fields and installed code allowlists. Unknown/custom error codes +are replaced by UNCLASSIFIED_ERROR. Error chains include suppressed context, +exception type, numeric errno, known schema keyword and bounded internal +module/function/line frames. They exclude exception messages, source lines, +locals, arguments, absolute paths, endpoint URLs, prompts, responses and case +contents. External frames are intentionally masked. A model's raw reply remains +only in the authorized case archive. Existing legacy LLM logging is replaced by +safe observations within an active session; legacy gateway provider messages are +suppressed only in the corresponding task context. Uninstrumented third-party +application loggers are outside this telemetry sink's privacy guarantee. + +After startup, log/snapshot/observer-callback failures are fail-open for case +execution, are counted, degrade observation status and produce a rate-limited +safe stderr marker. No secret-bearing exception text is used in that fallback. +A service manager can capture stderr when the normal sink is unwritable. Startup +configuration, unsafe paths and an already-held writer lease fail explicitly +before using the observation session. Snapshot/metric publication is atomic; +telemetry files are mutable operational artifacts outside the case ZIP. + +The defaults in config/telemetry.example.toml are configurable operational limits, +not case-package format boundaries. Scanner package limits can be supplied to +Telemetry(package_limits=...) or are adopted from an attached store. max_cases, +max_frames, snapshot byte limits and bounded metric series prevent uncontrolled +observation growth; partial observations are reported explicitly. + +Tests cover a complete mocked-model pipeline, preserved case bytes/budget/replay, +independent heartbeat, per-case stalls, passive/manual/human waits, privacy, +provider failover, rotation, corrupt-tail restart, cancellation, sink recovery, +callback failure, single-writer lease and CLI/status/schema behavior. Real-model +reasoning quality and the future continuous scheduler are separate concerns. diff --git a/src/viano_bug_analysis/analysis/change_request_builder.py b/src/viano_bug_analysis/analysis/change_request_builder.py index b67f871..f310ed5 100644 --- a/src/viano_bug_analysis/analysis/change_request_builder.py +++ b/src/viano_bug_analysis/analysis/change_request_builder.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import hashlib import tomllib from dataclasses import dataclass @@ -66,6 +67,7 @@ class ChangeRequestBuilder: self.diagnosis = DiagnosisBuilder(store, gateway, limits=self.limits, router=router, clock=clock) self.runner = self.diagnosis.runner + @observed("change_request_build") async def build(self, case_id, package_id, *, retry_id=None): if retry_id is not None and (not isinstance(retry_id, str) or not retry_id.strip()): raise ChangeRequestError("CHANGE_REQUEST_RETRY_ID") diff --git a/src/viano_bug_analysis/analysis/diagnosis_builder.py b/src/viano_bug_analysis/analysis/diagnosis_builder.py index 13a78e3..adc3472 100644 --- a/src/viano_bug_analysis/analysis/diagnosis_builder.py +++ b/src/viano_bug_analysis/analysis/diagnosis_builder.py @@ -1,4 +1,5 @@ """Durable final evaluation; deliberately does not create a coding task.""" +from viano_bug_analysis.telemetry.decorators import observed import json import hashlib import tomllib @@ -145,6 +146,7 @@ class DiagnosisBuilder: self.runner = CaseLLMRunner(store, gateway, clock=clock) self.clock = self.runner.clock + @observed("diagnosis_build") async def build(self, case_id, package_id, *, retry_id=None): if retry_id is not None and (not isinstance(retry_id, str) or not retry_id.strip()): raise DiagnosisError("DIAGNOSIS_RETRY_ID") diff --git a/src/viano_bug_analysis/analysis/hypothesis_builder.py b/src/viano_bug_analysis/analysis/hypothesis_builder.py index 32d0d08..f0a6852 100644 --- a/src/viano_bug_analysis/analysis/hypothesis_builder.py +++ b/src/viano_bug_analysis/analysis/hypothesis_builder.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import hashlib from dataclasses import dataclass from pathlib import Path @@ -46,6 +47,7 @@ class HypothesisBuilder: self.runner = CaseLLMRunner(store, gateway, clock=clock) self.prompt = HypothesisPromptBuilder() + @observed("hypotheses_build") async def build(self, case_id, package_id, *, attempt_id="default"): if not isinstance(attempt_id, str) or not attempt_id.strip(): raise HypothesisError("HYPOTHESIS_ATTEMPT") diff --git a/src/viano_bug_analysis/analysis/interpreter.py b/src/viano_bug_analysis/analysis/interpreter.py index d9ec931..123069b 100644 --- a/src/viano_bug_analysis/analysis/interpreter.py +++ b/src/viano_bug_analysis/analysis/interpreter.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import hashlib from dataclasses import dataclass from pathlib import Path @@ -43,6 +44,7 @@ class CaseInterpreter: self.prompt = InterpretationPromptBuilder() self.runner = CaseLLMRunner(store, gateway, clock=clock) + @observed("case_interpret") async def interpret(self, case_id: str, package_id: str, *, attempt_id: str = "default") -> InterpretationResult: if not isinstance(attempt_id, str) or not attempt_id.strip(): raise InterpretationError("INTERPRETATION_ATTEMPT") diff --git a/src/viano_bug_analysis/analysis/investigator.py b/src/viano_bug_analysis/analysis/investigator.py index cd3742a..835b125 100644 --- a/src/viano_bug_analysis/analysis/investigator.py +++ b/src/viano_bug_analysis/analysis/investigator.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import copy from dataclasses import asdict, dataclass from datetime import datetime, timedelta, timezone @@ -54,6 +55,7 @@ class Investigator: self.evidence=EvidenceStore(store,router=self.router) self.tests=AsyncTestCoordinator(store) + @observed("investigation_run") async def run(self,case_id,package_id,*,max_steps=None,retry_id=None): count=self.config.max_steps_per_dispatch if max_steps is None else max_steps if max_steps is None: @@ -68,6 +70,7 @@ class Investigator: return result return InvestigationResult("YIELDED",result.package,result.state_ref) + @observed("investigation_step") async def step(self,case_id,package_id,*,retry_id=None): if retry_id is not None and (not isinstance(retry_id,str) or not retry_id.strip()): raise InvestigationError("INVESTIGATION_RETRY_ID") diff --git a/src/viano_bug_analysis/analysis/output_validator.py b/src/viano_bug_analysis/analysis/output_validator.py index 2e74b75..88a5613 100644 --- a/src/viano_bug_analysis/analysis/output_validator.py +++ b/src/viano_bug_analysis/analysis/output_validator.py @@ -1,4 +1,5 @@ """Validate portable final output and atomically export to a local delivery queue.""" +from viano_bug_analysis.telemetry.decorators import observed import hashlib import os import tempfile @@ -39,6 +40,7 @@ class OutputValidator: self.builder = ChangeRequestBuilder(store, gateway, limits=limits, router=router) self.limits = self.builder.limits + @observed("output_validate") async def validate(self, case_id, package_id): """Read-only validation: guards prohibit initiating missing model stages.""" try: @@ -100,6 +102,7 @@ class OutputValidator: except ARCHIVE_ERRORS + (ValueError, TypeError, KeyError, IndexError, UnicodeError, RecursionError): raise OutputError("OUTPUT_INPUT") from None + @observed("output_finalize") async def finalize(self, case_id, package_id): try: with FileLock(self.store.workspace.runtime / "locks" / f"output-{key(case_id)}.lock"): @@ -126,6 +129,7 @@ class OutputValidator: except (LifecycleError, PackageReadError, CaseBuildError) as error: raise OutputError(getattr(error, "code", "OUTPUT_PACKAGE")) from None + @observed("output_publish") async def publish(self, case_id, package_id): """Copy a validated generation; local availability is not recipient ACK.""" checked = await self.finalize(case_id, package_id) diff --git a/src/viano_bug_analysis/budget/manager.py b/src/viano_bug_analysis/budget/manager.py index 9acf88c..0bcf3df 100644 --- a/src/viano_bug_analysis/budget/manager.py +++ b/src/viano_bug_analysis/budget/manager.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import copy import hashlib from dataclasses import dataclass @@ -119,6 +120,7 @@ class BudgetManager: return self._execute(case_id, package_id, {"operation":"complete","hypothesis_id":hypothesis_id, "action_id":"terminal","reason":reason}) + @observed("budget_execute") def _execute(self, case_id, package_id, request): try: with FileLock(self.store.workspace.runtime / "locks" / f"budget-{key(case_id)}.lock"): diff --git a/src/viano_bug_analysis/case/builder.py b/src/viano_bug_analysis/case/builder.py index 7d92431..c96437b 100644 --- a/src/viano_bug_analysis/case/builder.py +++ b/src/viano_bug_analysis/case/builder.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed from dataclasses import dataclass from pathlib import Path @@ -27,6 +28,7 @@ class CaseBuilder: self.store = store self.assembler = CaseAssembler(limits) + @observed("case_build") def build(self, case_id: str, package_id: str) -> CaseBuildResult: try: latest = self.store.latest(case_id, package_id) diff --git a/src/viano_bug_analysis/evidence/store.py b/src/viano_bug_analysis/evidence/store.py index 679d99e..78515b1 100644 --- a/src/viano_bug_analysis/evidence/store.py +++ b/src/viano_bug_analysis/evidence/store.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import copy import hashlib from dataclasses import asdict, dataclass @@ -164,6 +165,7 @@ class EvidenceStore: if h.hexdigest() != expected: raise EvidenceError("EVIDENCE_SOURCE_CHANGED") + @observed("evidence_execute") def _execute(self, case_id, package_id, operation_id, payload): try: payload = copy.deepcopy(payload) diff --git a/src/viano_bug_analysis/lifecycle/async_tests.py b/src/viano_bug_analysis/lifecycle/async_tests.py index 4a63910..c25e601 100644 --- a/src/viano_bug_analysis/lifecycle/async_tests.py +++ b/src/viano_bug_analysis/lifecycle/async_tests.py @@ -1,6 +1,7 @@ """Case-local async test handoff; external delivery and authentication are adapters.""" import hashlib +from viano_bug_analysis.telemetry.decorators import observed from dataclasses import dataclass from pathlib import Path @@ -38,9 +39,11 @@ class AsyncTestCoordinator: self.store = store self.max_json_bytes = max_json_bytes + @observed("async_test_submit", identity=False) def submit(self, request): return self._execute(request, None) + @observed("async_test_confirm", identity=False) def confirm_feedback(self, feedback): return self._execute(None, feedback) diff --git a/src/viano_bug_analysis/lifecycle/checkpoints.py b/src/viano_bug_analysis/lifecycle/checkpoints.py index 7ad27d2..f586c82 100644 --- a/src/viano_bug_analysis/lifecycle/checkpoints.py +++ b/src/viano_bug_analysis/lifecycle/checkpoints.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import hashlib import json from contextlib import contextmanager @@ -97,10 +98,13 @@ def checkpoint_chain(reader: PackageReader) -> list[tuple[str, dict]]: class CheckpointStore: - def __init__(self, workspace: Workspace, limits=None): + def __init__(self, workspace: Workspace, limits=None, *, telemetry=None): self.workspace = workspace + self.telemetry = telemetry self.writer = PackageWriter(limits) self.limits = self.writer.validator.limits + if telemetry is not None and telemetry.package_limits is None: + telemetry.package_limits = self.limits self._validation_cache = {} @contextmanager @@ -202,8 +206,11 @@ class CheckpointStore: manifest_updates=dict(updates or {}) | {"state": state, "artifacts": artifacts}) self._index(destination, result.manifest) atomic_json(self.workspace.runtime / "current_job.json", None) + if self.telemetry is not None: + self.telemetry.guard(self.telemetry.case_checkpoint, result.manifest, record["details"]) return destination + @observed("intake", identity=False) def intake(self, source: Path) -> tuple[str, Path]: try: with PackageReader(source, self.limits) as inbound: @@ -257,6 +264,7 @@ class CheckpointStore: value = manifest["routing"]["retry_not_before"] return value is None or datetime.fromisoformat(value.upper().replace("Z", "+00:00")) <= (now or datetime.now(timezone.utc)) + @observed("checkpoint_advance") def advance(self, case_id: str, package_id: str, target: str, *, step_id: str, reason: str | None = None, retry_not_before: str | None = None, details: dict | None = None, additions: dict[str, bytes] | None = None, diff --git a/src/viano_bug_analysis/lifecycle/queue.py b/src/viano_bug_analysis/lifecycle/queue.py index ade8afb..8931306 100644 --- a/src/viano_bug_analysis/lifecycle/queue.py +++ b/src/viano_bug_analysis/lifecycle/queue.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import hashlib import os import tempfile @@ -26,9 +27,9 @@ class IntakeResult: class InboxProcessor: """One inbox pass, no analysis execution or waits for external conditions.""" - def __init__(self, workspace: Workspace, limits=None): + def __init__(self, workspace: Workspace, limits=None, *, telemetry=None): self.workspace = workspace - self.store = CheckpointStore(workspace, limits) + self.store = CheckpointStore(workspace, limits, telemetry=telemetry) self.validator = PackageValidator(self.store.limits) def _snapshot(self, source: Path) -> tuple[Path, str]: @@ -89,6 +90,7 @@ class InboxProcessor: os.replace(source, directory / "package.zip") return IntakeResult("QUARANTINED", source, directory / "package.zip", issues[0]["code"], issues[0]["message"]) + @observed("intake", identity=False) def process(self, source: Path) -> IntakeResult: source = Path(source) try: @@ -115,8 +117,11 @@ class InboxProcessor: self._acknowledge(source, raw_hash) return IntakeResult(status, source, managed) except (LifecycleError, OSError, PackageReadError) as error: + if self.store.telemetry is not None: + self.store.telemetry.guard(self.store.telemetry.exception, error) return IntakeResult("ERROR", source, code=getattr(error, "code", "STORAGE_ERROR"), message=str(error)) + @observed("inbox_poll", identity=False) def poll(self, *, max_jobs: int = 100) -> list[IntakeResult]: if isinstance(max_jobs, bool) or not isinstance(max_jobs, int) or max_jobs <= 0: raise ValueError("max_jobs must be a positive integer") diff --git a/src/viano_bug_analysis/llm/runner.py b/src/viano_bug_analysis/llm/runner.py index 77de6aa..23b3f8c 100644 --- a/src/viano_bug_analysis/llm/runner.py +++ b/src/viano_bug_analysis/llm/runner.py @@ -1,3 +1,4 @@ +from viano_bug_analysis.telemetry.decorators import observed import hashlib from dataclasses import dataclass from datetime import datetime, timedelta, timezone @@ -33,8 +34,18 @@ class CaseLLMRunner: self.store = store self.gateway = gateway self.clock = clock or (lambda: datetime.now(timezone.utc)) - self.logger = case_logger(store.workspace) + self._legacy_logger = case_logger(store.workspace) if getattr(store, "telemetry", None) is None else None + @property + def logger(self): + telemetry = getattr(self.store, "telemetry", None) + if telemetry is not None and telemetry.started: + return telemetry.logger + if self._legacy_logger is None: + self._legacy_logger = case_logger(self.store.workspace) + return self._legacy_logger + + @observed("llm_run") async def run(self, case_id: str, package_id: str, *, operation_id: str, request: dict) -> LLMRunResult: if not isinstance(operation_id, str) or not operation_id.strip(): raise LLMError("LLM_OPERATION") @@ -147,6 +158,8 @@ class CaseLLMRunner: "case_id": case_id, "package_id": package_id, "analysis_revision": manifest["analysis_revision"], **details, "outcome": outcome} validate(record, "llm-call") + if getattr(self.store, "telemetry", None) is not None: + self.store.telemetry.guard(self.store.telemetry.llm_round, record) additions[result_ref] = json_bytes(record) committed = self.store.advance(case_id, package_id, target, step_id=f"llm:{operation}:round:{rounds}", diff --git a/src/viano_bug_analysis/schemas/service-status-v1.schema.json b/src/viano_bug_analysis/schemas/service-status-v1.schema.json new file mode 100644 index 0000000..f18e182 --- /dev/null +++ b/src/viano_bug_analysis/schemas/service-status-v1.schema.json @@ -0,0 +1,189 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "format": { + "const": "viano-service-status" + }, + "format_version": { + "const": 1 + }, + "service_id": { + "type": "string", + "pattern": "^[a-f0-9]{32}$" + }, + "pid": { + "type": "integer", + "minimum": 1 + }, + "started_at": { + "type": "string" + }, + "heartbeat_at": { + "type": "string" + }, + "stopped_at": { + "type": [ + "string", + "null" + ] + }, + "last_progress_at": { + "type": [ + "string", + "null" + ] + }, + "last_error": { + "type": [ + "object", + "null" + ] + }, + "telemetry_failures": { + "type": "integer", + "minimum": 0 + }, + "telemetry_healthy": { + "type": "boolean" + }, + "active_operations": { + "type": "array", + "items": { + "type": "object" + } + }, + "cases": { + "type": "array", + "items": { + "type": "object" + } + }, + "budgets": { + "type": "array", + "items": { + "type": "object" + } + }, + "inventory": { + "type": [ + "object", + "null" + ] + }, + "config": { + "type": "object", + "properties": { + "heartbeat_seconds": { + "type": "integer", + "minimum": 1 + }, + "heartbeat_stale_seconds": { + "type": "integer", + "minimum": 1 + }, + "progress_warning_seconds": { + "type": "integer", + "minimum": 1 + }, + "inventory_interval_seconds": { + "type": "integer", + "minimum": 1 + }, + "inventory_stale_seconds": { + "type": "integer", + "minimum": 1 + }, + "log_max_bytes": { + "type": "integer", + "minimum": 1 + }, + "log_backups": { + "type": "integer", + "minimum": 1 + }, + "max_cases": { + "type": "integer", + "minimum": 1 + }, + "max_frames": { + "type": "integer", + "minimum": 1 + }, + "max_snapshot_bytes": { + "type": "integer", + "minimum": 1 + }, + "snapshot_interval_seconds": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "heartbeat_seconds", + "heartbeat_stale_seconds", + "progress_warning_seconds", + "inventory_interval_seconds", + "inventory_stale_seconds", + "log_max_bytes", + "log_backups", + "max_cases", + "max_frames", + "max_snapshot_bytes" + ], + "additionalProperties": false + }, + "counters": { + "type": "array", + "items": { + "type": "object" + } + }, + "observer": { + "type": "string" + }, + "work": { + "type": "string" + }, + "health": { + "type": "string" + }, + "heartbeat_age_seconds": { + "type": "number", + "minimum": 0 + }, + "progress_age_seconds": { + "type": "number", + "minimum": 0 + }, + "inventory_fresh": { + "type": "boolean" + } + }, + "required": [ + "format", + "format_version", + "service_id", + "pid", + "started_at", + "heartbeat_at", + "stopped_at", + "last_progress_at", + "last_error", + "telemetry_failures", + "telemetry_healthy", + "active_operations", + "cases", + "budgets", + "inventory", + "config", + "counters", + "observer", + "work", + "health", + "heartbeat_age_seconds", + "progress_age_seconds", + "inventory_fresh" + ], + "additionalProperties": false +} diff --git a/src/viano_bug_analysis/telemetry/__init__.py b/src/viano_bug_analysis/telemetry/__init__.py new file mode 100644 index 0000000..28e9a73 --- /dev/null +++ b/src/viano_bug_analysis/telemetry/__init__.py @@ -0,0 +1,15 @@ +"""Optional operational observations, separate from authoritative case state.""" +__all__ = ["Telemetry", "TelemetryConfig", "TelemetryError", "read_status", "scan_workspace"] + + +def __getattr__(name): + if name in {"TelemetryConfig", "TelemetryError"}: + from . import config + return getattr(config, name) + if name == "Telemetry": + from .core import Telemetry + return Telemetry + if name in {"read_status", "scan_workspace"}: + from . import inspection + return getattr(inspection, name) + raise AttributeError(name) diff --git a/src/viano_bug_analysis/telemetry/__main__.py b/src/viano_bug_analysis/telemetry/__main__.py new file mode 100644 index 0000000..3c1751a --- /dev/null +++ b/src/viano_bug_analysis/telemetry/__main__.py @@ -0,0 +1,97 @@ +"""Read-only operator commands; no job retry, case mutation or remote server.""" +import argparse +import json +from collections import deque +from pathlib import Path + +from .config import TelemetryError +from .core import identity +from .inspection import read_status, scan_workspace +from .safety import EVENTS, OPERATIONS, STATUSES, detail, hashed, safe_code + + +def read_events(root, *, errors=False, case_id=None, source_name=None, limit=50, max_bytes=5 * 1024**2, backups=5): + directory = Path(root) / "runtime" / "telemetry" + if directory.is_symlink() or not 1 <= limit <= 1000: + raise TelemetryError("TELEMETRY_READ") + records = deque(maxlen=limit) + corrupt = 0 + paths = [directory / f"service.jsonl.{i}" for i in range(backups, 0, -1)] + [directory / "service.jsonl"] + try: + for path in paths: + if not path.exists(): + continue + if path.is_symlink() or path.stat().st_size > max_bytes: + raise TelemetryError("TELEMETRY_READ") + with path.open("rb") as stream: + while True: + line = stream.readline(max_bytes + 1) + if not line: + break + if len(line) > max_bytes: + raise TelemetryError("TELEMETRY_READ") + # Ignore an incomplete last record during an ongoing append. + if not line.endswith(b"\n"): + break + try: + value = json.loads(line) + except (ValueError, UnicodeError): + corrupt += 1 + continue + if not isinstance(value, dict) or value.get("event") not in EVENTS: + continue + if errors and value.get("level") != "ERROR": + continue + if case_id is not None and value.get("case_key") != identity(case_id): + continue + if source_name is not None and value.get("detail", {}).get("source_key") != identity(source_name): + continue + records.append({"timestamp": value.get("timestamp"), "service_id": hashed(value.get("service_id"), 32), + "level": value.get("level") if value.get("level") in {"INFO", "WARNING", "ERROR"} else None, + "event": value["event"], "operation": value.get("operation") if value.get("operation") in OPERATIONS else None, + "case_key": hashed(value.get("case_key")), "package_key": hashed(value.get("package_key")), + "span_id": hashed(value.get("span_id"), 32), "parent_span_id": hashed(value.get("parent_span_id"), 32), + "status": value.get("status") if value.get("status") in STATUSES else None, + "code": safe_code(value["code"]) if value.get("code") else None, + "detail": detail(value["event"], value.get("detail", {}))}) + result = list(records) + if corrupt: + result.append({"event": "telemetry_read_warning", "level": "WARNING", "code": "TELEMETRY_LOG_CORRUPT", "detail": {"corrupt_records": corrupt}}) + return result + except (OSError, ValueError, TypeError, AttributeError): + raise TelemetryError("TELEMETRY_READ") from None + + +def main(argv=None): + parser = argparse.ArgumentParser(description="VianoBugAnalysis operational observations") + parser.add_argument("command", choices=("status", "errors", "events", "metrics", "scan")) + parser.add_argument("--workspace", required=True) + parser.add_argument("--case-id") + parser.add_argument("--source-name") + parser.add_argument("--limit", type=int, default=50) + args = parser.parse_args(argv) + try: + if args.command == "status": + result = read_status(args.workspace) + elif args.command == "scan": + result = scan_workspace(args.workspace) + elif args.command in {"errors", "events"}: + status = read_status(args.workspace) + config = status.get("config", {}) + result = read_events(args.workspace, errors=args.command == "errors", case_id=args.case_id, source_name=args.source_name, limit=args.limit, + max_bytes=config.get("log_max_bytes", 5 * 1024**2), backups=config.get("log_backups", 5)) + else: + path = Path(args.workspace) / "runtime" / "telemetry" / "metrics.prom" + if path.is_symlink() or path.stat().st_size > 4 * 1024**2: + raise TelemetryError("TELEMETRY_READ") + print(path.read_text(encoding="utf-8"), end="") + return 0 + print(json.dumps(result, ensure_ascii=True, indent=2)) + return 0 + except (TelemetryError, OSError): + print(json.dumps({"code": "TELEMETRY_READ"})) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/viano_bug_analysis/telemetry/codes.py b/src/viano_bug_analysis/telemetry/codes.py new file mode 100644 index 0000000..61cd624 --- /dev/null +++ b/src/viano_bug_analysis/telemetry/codes.py @@ -0,0 +1,4 @@ +"""Installed code/module/class allowlists; unknown exception text is never echoed.""" +KNOWN_CODES = frozenset(['ACCEPTED', 'ARCHIVE_ERROR', 'ARCHIVE_INVALID', 'ARCHIVE_LIMIT', 'ARTIFACT_EXISTS', 'ARTIFACT_INVALID', 'ARTIFACT_TYPE', 'ASYNC_TEST_ATTACHMENT', 'ASYNC_TEST_CONFLICT', 'ASYNC_TEST_COST', 'ASYNC_TEST_IDENTITY', 'ASYNC_TEST_INVALID', 'ASYNC_TEST_LIMIT', 'ASYNC_TEST_NOT_WAITING', 'ASYNC_TEST_PHASE', 'ASYNC_TEST_RESERVATION', 'AWAITING_TEST_FEEDBACK', 'BLOCKED', 'BUDGET_ACCOUNTING_CHANGED', 'BUDGET_ACTION_CONFLICT', 'BUDGET_ACTION_UNKNOWN', 'BUDGET_CASE_NOT_FOUND', 'BUDGET_CONFIG', 'BUDGET_EVENT_LIMIT', 'BUDGET_EVIDENCE', 'BUDGET_EVIDENCE_BINDING', 'BUDGET_GENERATION_CHANGED', 'BUDGET_HISTORY_INVALID', 'BUDGET_HISTORY_LIMIT', 'BUDGET_HISTORY_MISSING', 'BUDGET_HYPOTHESES_NOT_READY', 'BUDGET_HYPOTHESIS_UNKNOWN', 'BUDGET_INPUT', 'BUDGET_INSUFFICIENT', 'BUDGET_LIMIT_CONFIG', 'BUDGET_NOT_INITIALIZED', 'BUDGET_NOVELTY_CAP', 'BUDGET_OPERATION', 'BUDGET_ORIGIN_CHANGED', 'BUDGET_PACKAGE', 'BUDGET_PENDING_ACTION', 'BUDGET_PHASE', 'BUDGET_POINTER_CHANGED', 'BUDGET_POLICY_CHANGED', 'BUDGET_RESULT_LIMIT', 'BUDGET_RESULT_REQUIRED', 'BUDGET_SEMANTIC_SCORE', 'BUDGET_SOURCE_CHANGED', 'BUDGET_SOURCE_REFERENCES', 'BUDGET_STEP_CONFLICT', 'BUDGET_TERMINAL', 'BUG_ID_MISMATCH', 'BUILDING_CASE', 'BUILDING_DIAGNOSIS', 'BUSY', 'CACHE_CONFLICT', 'CANCELLED', 'CASE_BLOCKED', 'CASE_BUILD_ERROR', 'CASE_IDENTITY', 'CASE_LIMIT', 'CASE_NOT_FOUND', 'CASE_OUTPUT_CHANGED', 'CASE_SCHEMA', 'CHANGE_REQUEST_CONFIG', 'CHANGE_REQUEST_CONTEXT_CHANGED', 'CHANGE_REQUEST_CONTEXT_LIMIT', 'CHANGE_REQUEST_DIAGNOSIS_NOT_READY', 'CHANGE_REQUEST_EVIDENCE', 'CHANGE_REQUEST_INCOMPLETE', 'CHANGE_REQUEST_INPUT', 'CHANGE_REQUEST_INSUFFICIENT_CONTEXT', 'CHANGE_REQUEST_LIMIT_CONFIG', 'CHANGE_REQUEST_LLM_BINDING', 'CHANGE_REQUEST_LOCAL_CONTEXT_CHANGED', 'CHANGE_REQUEST_MISSING_INFORMATION', 'CHANGE_REQUEST_OUTPUT', 'CHANGE_REQUEST_OUTPUT_CHANGED', 'CHANGE_REQUEST_OUTPUT_LIMIT', 'CHANGE_REQUEST_PACKAGE', 'CHANGE_REQUEST_PHASE', 'CHANGE_REQUEST_POINTER_CHANGED', 'CHANGE_REQUEST_POLICY_CHANGED', 'CHANGE_REQUEST_REJECTION_LIMIT', 'CHANGE_REQUEST_RETRY_ID', 'CHANGE_REQUEST_ROOT_SCOPE', 'CHANGE_REQUEST_SCOPE', 'CHANGE_REQUEST_SCOPE_EVIDENCE', 'CHECKPOINT_HISTORY_LOST', 'CHECKPOINT_INVALID', 'CHECKPOINT_LIMIT', 'CLEANUP_ERROR', 'COMPLETED', 'CONFLICTING_CLAIMS', 'CORE_REVISION_MISSING', 'CREATE_ERROR', 'DEFERRED', 'DEGRADED', 'DIAGNOSIS_BUDGET_NOT_READY', 'DIAGNOSIS_CASE_NOT_FOUND', 'DIAGNOSIS_CONFIG', 'DIAGNOSIS_CONTEXT_CHANGED', 'DIAGNOSIS_CONTEXT_LIMIT', 'DIAGNOSIS_COUNTERCHECK', 'DIAGNOSIS_EVIDENCE_REFERENCE', 'DIAGNOSIS_HYPOTHESES', 'DIAGNOSIS_IDENTITY', 'DIAGNOSIS_INCONCLUSIVE', 'DIAGNOSIS_INPUT', 'DIAGNOSIS_INVESTIGATION_NOT_READY', 'DIAGNOSIS_LIMIT_CONFIG', 'DIAGNOSIS_LLM_BINDING', 'DIAGNOSIS_OUTPUT', 'DIAGNOSIS_OUTPUT_CHANGED', 'DIAGNOSIS_OUTPUT_LIMIT', 'DIAGNOSIS_PACKAGE', 'DIAGNOSIS_PHASE', 'DIAGNOSIS_POINTER_CHANGED', 'DIAGNOSIS_POLICY_CHANGED', 'DIAGNOSIS_REJECTED_SELECTION', 'DIAGNOSIS_REJECTION_LIMIT', 'DIAGNOSIS_RETRY_ID', 'DIAGNOSIS_SELECTION', 'DIAGNOSIS_STALE_CASE', 'DIAGNOSIS_UNGROUNDED', 'DUPLICATE', 'ENTRY_COLLISION', 'ENTRY_LIMIT', 'ERROR', 'EVIDENCE_CASE_NOT_FOUND', 'EVIDENCE_CONTENT_CHANGED', 'EVIDENCE_DUPLICATE_CHANGED', 'EVIDENCE_GENERATION_CHANGED', 'EVIDENCE_HISTORY_INVALID', 'EVIDENCE_HISTORY_LIMIT', 'EVIDENCE_IDENTITY', 'EVIDENCE_INDEX_CHANGED', 'EVIDENCE_INPUT', 'EVIDENCE_LIMIT', 'EVIDENCE_LIMIT_CONFIG', 'EVIDENCE_NOVELTY_CHANGED', 'EVIDENCE_OPERATION_ID', 'EVIDENCE_ORIGIN_CHANGED', 'EVIDENCE_PHASE', 'EVIDENCE_POINTER_CHANGED', 'EVIDENCE_RECORD_INVALID', 'EVIDENCE_RESERVATION', 'EVIDENCE_RESULT_CHANGED', 'EVIDENCE_SOURCE_CHANGED', 'EVIDENCE_STEP_CONFLICT', 'EXPANSION_LIMIT', 'EXPECTED_BEHAVIOR_', 'EXTERNAL_FEEDBACK_REQUIRED', 'FAILED_TO_ANALYZE', 'FILE_LIMIT', 'FILE_NOT_FOUND', 'GENERATING_HYPOTHESES', 'GENERATION_CHANGED', 'GENERATION_CONFLICT', 'HANDED_OFF', 'HEADER_LIMIT', 'HEARTBEAT_STALE', 'HISTORY_REGRESSION', 'HYPOTHESIS_ALLOCATION_CHANGED', 'HYPOTHESIS_ATTEMPT', 'HYPOTHESIS_CONFIG', 'HYPOTHESIS_CONTEXT_CHANGED', 'HYPOTHESIS_CONTEXT_LIMIT', 'HYPOTHESIS_COUNT', 'HYPOTHESIS_DUPLICATE', 'HYPOTHESIS_EXISTING_REGISTRY', 'HYPOTHESIS_GENERATION_CHANGED', 'HYPOTHESIS_GROUNDING', 'HYPOTHESIS_HISTORY_IDENTITY', 'HYPOTHESIS_HISTORY_LIMIT', 'HYPOTHESIS_IDENTITY', 'HYPOTHESIS_INPUT', 'HYPOTHESIS_INTERPRETATION_NOT_READY', 'HYPOTHESIS_LIMIT_CONFIG', 'HYPOTHESIS_LLM_BINDING', 'HYPOTHESIS_NO_OBSERVATIONS', 'HYPOTHESIS_OUTPUT_CHANGED', 'HYPOTHESIS_OUTPUT_LIMIT', 'HYPOTHESIS_PACKAGE', 'HYPOTHESIS_PHASE', 'HYPOTHESIS_POLICY_CHANGED', 'HYPOTHESIS_PRIMARY_FOCUS', 'HYPOTHESIS_REGISTRY_CHANGED', 'HYPOTHESIS_SCHEMA', 'HYPOTHESIS_SOURCE_UNKNOWN', 'HYPOTHESIS_SUMMARY_LIMIT', 'IDLE', 'IMPORT_COLLISION', 'IMPORT_ERROR', 'IMPORT_EXISTS', 'IMPORT_LIMIT', 'IMPORT_RECEIPT', 'INCONCLUSIVE', 'INDEX_UPDATE_FORBIDDEN', 'INFO', 'INITIAL_GENERATION', 'INPUT_DUPLICATE', 'INPUT_JSON', 'INPUT_LIMIT', 'INPUT_PENDING', 'INPUT_REFERENCE', 'INPUT_SCHEMA', 'INPUT_TARGET_ALIAS', 'INTEGRITY_MISMATCH', 'INTERPRETATION_ATTEMPT', 'INTERPRETATION_CASE_IDENTITY', 'INTERPRETATION_CASE_IDS', 'INTERPRETATION_CASE_NOT_READY', 'INTERPRETATION_CONTEXT_CHANGED', 'INTERPRETATION_CONTEXT_LIMIT', 'INTERPRETATION_CONTRADICTION_SOURCES', 'INTERPRETATION_ENTITY', 'INTERPRETATION_EXPECTATION_UNKNOWN', 'INTERPRETATION_GAP_SOURCES', 'INTERPRETATION_GENERATION_CHANGED', 'INTERPRETATION_INPUT', 'INTERPRETATION_LLM_BINDING', 'INTERPRETATION_OUTPUT_CHANGED', 'INTERPRETATION_OUTPUT_LIMIT', 'INTERPRETATION_PACKAGE', 'INTERPRETATION_PHASE', 'INTERPRETATION_SCHEMA', 'INTERPRETATION_SOURCE_KIND', 'INTERPRETATION_SOURCE_LIMIT', 'INTERPRETATION_SOURCE_UNKNOWN', 'INTERPRETATION_SUMMARY_LIMIT', 'INTERPRETING_CASE', 'INVALID_BUILD_STATE', 'INVALID_GENERATION', 'INVALID_LLM_RESPONSES', 'INVALID_RETRY', 'INVALID_REVISION', 'INVALID_STATE', 'INVALID_STEP', 'INVALID_TRANSITION', 'INVESTIGATING_HYPOTHESES', 'INVESTIGATION_ASSESSMENT_BINDING', 'INVESTIGATION_ASSESSMENT_REQUIRED', 'INVESTIGATION_CASE_NOT_FOUND', 'INVESTIGATION_CLOCK', 'INVESTIGATION_CONFIG', 'INVESTIGATION_CONTEXT_CHANGED', 'INVESTIGATION_CONTEXT_LIMIT', 'INVESTIGATION_CONTRADICTORY_UPDATE', 'INVESTIGATION_DISPATCH_LIMIT', 'INVESTIGATION_EVIDENCE_UNKNOWN', 'INVESTIGATION_GENERATION_CHANGED', 'INVESTIGATION_HISTORY_INVALID', 'INVESTIGATION_HISTORY_LIMIT', 'INVESTIGATION_HISTORY_MISSING', 'INVESTIGATION_INPUT', 'INVESTIGATION_ITERATION_LIMIT', 'INVESTIGATION_LIMIT_CONFIG', 'INVESTIGATION_LLM_BINDING', 'INVESTIGATION_OUTPUT', 'INVESTIGATION_PHASE', 'INVESTIGATION_PLAN_BINDING', 'INVESTIGATION_POINTER_CHANGED', 'INVESTIGATION_POLICY_CHANGED', 'INVESTIGATION_QUESTION', 'INVESTIGATION_QUESTION_DUPLICATE', 'INVESTIGATION_QUESTION_LIMIT', 'INVESTIGATION_RETRY_ID', 'INVESTIGATION_RETRY_NOT_EXHAUSTED', 'INVESTIGATION_SCORE', 'INVESTIGATION_STALE_CASE', 'INVESTIGATION_STATE_LIMIT', 'INVESTIGATION_SUMMARY_CHANGED', 'INVESTIGATION_SUPPORT_REQUIRED', 'INVESTIGATION_TEST_COST', 'INVESTIGATION_UNOWNED_PENDING_ACTION', 'LIFECYCLE_ERROR', 'LIVE', 'LLM_ADAPTER', 'LLM_CHECKPOINT', 'LLM_CLOCK', 'LLM_CONFIG', 'LLM_CONFIGURATION', 'LLM_CONNECTION', 'LLM_CREDENTIALS', 'LLM_ERROR', 'LLM_GENERATION_CHANGED', 'LLM_HTTP_CONFIGURATION', 'LLM_HTTP_TEMPORARY', 'LLM_INCOMPLETE_OUTPUT', 'LLM_LOG_PATH', 'LLM_OPERATION', 'LLM_OPERATION_CONFLICT', 'LLM_OUTPUT_SCHEMA', 'LLM_PHASE', 'LLM_PRIVACY_NO_PROVIDER', 'LLM_PROVIDER_ERROR', 'LLM_REQUEST', 'LLM_REQUEST_LIMIT', 'LLM_RESPONSE', 'LLM_RESPONSE_LIMIT', 'LLM_RESULT_MISSING', 'LLM_RETRY_EXHAUSTED', 'LLM_SCHEMA_REFERENCE', 'LLM_TIMEOUT', 'LLM_UNAVAILABLE', 'LOCK_BUSY', 'LOCK_ERROR', 'MANIFEST_AMBIGUOUS', 'MANIFEST_CHANGED', 'MANIFEST_MISSING', 'MANIFEST_SCHEMA', 'MANIFEST_TIME', 'MANIFEST_UPDATE_FORBIDDEN', 'MEMBER_COLLISION', 'MEMBER_TRUNCATED', 'MISSING_INPUT_INDEX', 'MISSING_OBSERVATIONS', 'MISSING_REPORT', 'MISSING_REPRODUCTION', 'NEW', 'NFKC', 'NORMALIZATION_ERROR', 'NOT_STARTED', 'NO_RECENT_PROGRESS', 'OUTPUT_CHANGED', 'OUTPUT_CHANGE_REQUEST_NOT_READY', 'OUTPUT_EXPORT_CONFLICT', 'OUTPUT_INCONCLUSIVE_CODING_TASK', 'OUTPUT_INPUT', 'OUTPUT_NOT_READY', 'OUTPUT_PACKAGE', 'OUTPUT_PHASE', 'OUTPUT_READY', 'OUTPUT_REPORT_CHANGED', 'OUTPUT_REVIEW_ROUTING', 'OUTPUT_ROUTING', 'OUTPUT_STORAGE', 'OUTPUT_UNSAFE_DESTINATION', 'OUTPUT_VALIDATION_MISSING', 'PACKAGE_ERROR', 'PACKAGE_INVALID', 'PACKAGE_LIMIT', 'PRESERVATION_FAILED', 'PREVIOUS_CASE', 'PREVIOUS_IDS', 'PREVIOUS_INPUT_CHANGED', 'PREVIOUS_RECORD_LOST', 'PROGRESSED', 'PROJECT_MISMATCH', 'Q0001', 'QUARANTINED', 'QUEUE_UNKNOWN', 'RATIO_LIMIT', 'READY_NOT_PROCESSING', 'READ_ERROR', 'REFERENCE_MISSING', 'REJECTED', 'REOPENED', 'RESEARCH_EXISTS', 'RESEARCH_IDENTITY', 'RESEARCH_JSON', 'RESEARCH_LIMIT', 'RESEARCH_NOT_REQUIRED', 'RESEARCH_PROVIDER', 'RESEARCH_REQUEST_MISMATCH', 'RESEARCH_SOURCE', 'RESEARCH_STATE', 'RESEARCH_UNAVAILABLE', 'RESERVED_ARTIFACT', 'RESUMED', 'RETRY_NOT_DUE', 'RETRY_PENDING', 'REVIEW', 'REVISION_IMMUTABLE', 'REVISION_REGRESSION', 'SIZE_MISMATCH', 'SOURCE_CHANGED', 'SOURCE_ENTITY', 'SOURCE_FIELD', 'SOURCE_FORMAT', 'SOURCE_ID', 'SOURCE_INDEX', 'SOURCE_JSON', 'SOURCE_MANIFEST', 'SOURCE_NOT_COPIED', 'SOURCE_PAIR_MISMATCH', 'SOURCE_REFERENCE', 'SOURCE_RELATION', 'SOURCE_REPORTS_MISSING', 'SOURCE_ROWS', 'SOURCE_SIZE', 'SOURCE_TIME_UNQUALIFIED', 'SOURCE_WARNING', 'SPECIAL_ENTRY', 'SPECIAL_MEMBER', 'STALE_CASE', 'STEP_CONFLICT', 'STOPPED', 'STORAGE_CHECKPOINT_INVALID', 'STORAGE_CONFLICT', 'STORAGE_ERROR', 'STORAGE_PACKAGE_CHANGED', 'STORAGE_PACKAGE_INVALID', 'STREAM_LIMIT', 'STRUCTURED_INPUT_LIMIT', 'SUCCESS', 'SUMMARY_LIMIT', 'TARGET_EXISTS', 'TARGET_SYMLINK', 'TELEMETRY_CALLBACK', 'TELEMETRY_CLOCK', 'TELEMETRY_CONFIG', 'TELEMETRY_EVENT', 'TELEMETRY_LIMIT', 'TELEMETRY_LOG_CORRUPT', 'TELEMETRY_OPERATION', 'TELEMETRY_PATH', 'TELEMETRY_READ', 'TELEMETRY_SCAN_CHECKPOINT', 'TELEMETRY_SCAN_IDENTITY', 'TELEMETRY_SCAN_PATH', 'TELEMETRY_SESSION', 'TELEMETRY_STORAGE', 'TOOL_BATCH_LIMIT', 'TOOL_CASE_IDENTITY', 'TOOL_CONTEXT_RANGE', 'TOOL_DATABASE_COLUMN_UNKNOWN', 'TOOL_DATABASE_DESCRIPTOR', 'TOOL_DATABASE_JSON', 'TOOL_DATABASE_ROWS', 'TOOL_DUPLICATE_REQUEST', 'TOOL_FILE_LIMIT', 'TOOL_HYPOTHESIS_IDENTITY', 'TOOL_HYPOTHESIS_UNKNOWN', 'TOOL_IDENTITY', 'TOOL_INPUT', 'TOOL_INPUT_LIMIT', 'TOOL_LIMIT_CONFIG', 'TOOL_MATCH_LIMIT', 'TOOL_PACKAGE', 'TOOL_PHASE', 'TOOL_REQUEST_INTEGER', 'TOOL_REQUEST_LIMIT', 'TOOL_RESULT_LIMIT', 'TOOL_SOURCE_BINARY', 'TOOL_SOURCE_CHANGED', 'TOOL_SOURCE_CONFLICT', 'TOOL_SOURCE_ENCODING', 'TOOL_SOURCE_NOT_DECLARED', 'TOOL_STALE_CASE', 'TOTAL_LIMIT', 'UNCLASSIFIED_ERROR', 'UNKNOWN', 'UNSAFE_INPUT', 'UNSAFE_LOCK', 'UNSAFE_PATH', 'UNSAFE_REFERENCE', 'UNSAFE_WORKSPACE', 'UNSUPPORTED_ENTRY', 'USD', 'VALID', 'VALIDATED', 'WAITING_EXTERNAL', 'WARNING', 'WORKING', 'WRITE_ERROR', 'YIELDED']) +KNOWN_MODULES = frozenset(['viano_bug_analysis', 'viano_bug_analysis.analysis', 'viano_bug_analysis.analysis.change_request_builder', 'viano_bug_analysis.analysis.change_request_contracts', 'viano_bug_analysis.analysis.context', 'viano_bug_analysis.analysis.contracts', 'viano_bug_analysis.analysis.diagnosis_builder', 'viano_bug_analysis.analysis.errors', 'viano_bug_analysis.analysis.hypothesis_builder', 'viano_bug_analysis.analysis.hypothesis_config', 'viano_bug_analysis.analysis.hypothesis_contracts', 'viano_bug_analysis.analysis.hypothesis_prompt', 'viano_bug_analysis.analysis.hypothesis_summary', 'viano_bug_analysis.analysis.interpreter', 'viano_bug_analysis.analysis.investigation_config', 'viano_bug_analysis.analysis.investigation_prompt', 'viano_bug_analysis.analysis.investigation_summary', 'viano_bug_analysis.analysis.investigator', 'viano_bug_analysis.analysis.output_validator', 'viano_bug_analysis.analysis.prompt', 'viano_bug_analysis.analysis.summary', 'viano_bug_analysis.budget', 'viano_bug_analysis.budget.config', 'viano_bug_analysis.budget.engine', 'viano_bug_analysis.budget.manager', 'viano_bug_analysis.budget.scoring', 'viano_bug_analysis.case', 'viano_bug_analysis.case.assembler', 'viano_bug_analysis.case.builder', 'viano_bug_analysis.case.contracts', 'viano_bug_analysis.case.errors', 'viano_bug_analysis.case.expected_behavior', 'viano_bug_analysis.case.report_collector', 'viano_bug_analysis.case.reproduction_collector', 'viano_bug_analysis.case.summary', 'viano_bug_analysis.evidence', 'viano_bug_analysis.evidence.scoring', 'viano_bug_analysis.evidence.store', 'viano_bug_analysis.lifecycle', 'viano_bug_analysis.lifecycle.async_tests', 'viano_bug_analysis.lifecycle.checkpoints', 'viano_bug_analysis.lifecycle.errors', 'viano_bug_analysis.lifecycle.job', 'viano_bug_analysis.lifecycle.locking', 'viano_bug_analysis.lifecycle.queue', 'viano_bug_analysis.lifecycle.state', 'viano_bug_analysis.lifecycle.storage', 'viano_bug_analysis.llm', 'viano_bug_analysis.llm.config', 'viano_bug_analysis.llm.errors', 'viano_bug_analysis.llm.gateway', 'viano_bug_analysis.llm.logging', 'viano_bug_analysis.llm.openai_compatible', 'viano_bug_analysis.llm.prompt', 'viano_bug_analysis.llm.prompts', 'viano_bug_analysis.llm.provider_base', 'viano_bug_analysis.llm.runner', 'viano_bug_analysis.package', 'viano_bug_analysis.package.errors', 'viano_bug_analysis.package.factory', 'viano_bug_analysis.package.limits', 'viano_bug_analysis.package.manifest', 'viano_bug_analysis.package.reader', 'viano_bug_analysis.package.validator', 'viano_bug_analysis.package.writer', 'viano_bug_analysis.profiles', 'viano_bug_analysis.profiles.vcore', 'viano_bug_analysis.profiles.vcore.archive_reader', 'viano_bug_analysis.profiles.vcore.common', 'viano_bug_analysis.profiles.vcore.db_tools', 'viano_bug_analysis.profiles.vcore.errors', 'viano_bug_analysis.profiles.vcore.importer', 'viano_bug_analysis.profiles.vcore.wowdiag_adapter', 'viano_bug_analysis.profiles.vcore.wowsrc_adapter', 'viano_bug_analysis.research', 'viano_bug_analysis.research.provider', 'viano_bug_analysis.schemas', 'viano_bug_analysis.telemetry', 'viano_bug_analysis.telemetry.__main__', 'viano_bug_analysis.telemetry.config', 'viano_bug_analysis.telemetry.core', 'viano_bug_analysis.telemetry.decorators', 'viano_bug_analysis.telemetry.inspection', 'viano_bug_analysis.telemetry.safety', 'viano_bug_analysis.tools', 'viano_bug_analysis.tools.config', 'viano_bug_analysis.tools.context', 'viano_bug_analysis.tools.database_query', 'viano_bug_analysis.tools.router', 'viano_bug_analysis.tools.source_query']) +KNOWN_CLASSES = frozenset(['AdaptedInputs', 'AdapterError', 'ArchiveLimits', 'ArchiveReader', 'AsyncTestCoordinator', 'AsyncTestResult', 'BoundedTarInfo', 'BudgetConfig', 'BudgetError', 'BudgetLimits', 'BudgetManager', 'BudgetResult', 'CaseAssembler', 'CaseBuildError', 'CaseBuildResult', 'CaseBuilder', 'CaseBuilderLimits', 'CaseInterpreter', 'CaseLLMRunner', 'ChangeRequestBuilder', 'ChangeRequestConfig', 'ChangeRequestError', 'ChangeRequestLimits', 'ChangeRequestResult', 'CheckpointStore', 'DatabaseQueryTool', 'DiagnosisBuilder', 'DiagnosisConfig', 'DiagnosisError', 'DiagnosisLimits', 'DiagnosisResult', 'EvidenceError', 'EvidenceLimits', 'EvidenceResult', 'EvidenceStore', 'ExpectedBehaviorResolver', 'ExternalResearchProvider', 'FileLock', 'GatewayConfig', 'GatewayUnavailable', 'HypothesisBuildResult', 'HypothesisBuilder', 'HypothesisConfig', 'HypothesisError', 'HypothesisLimits', 'HypothesisPromptBuilder', 'ImportPlan', 'InboxProcessor', 'IntakeResult', 'InterpretationError', 'InterpretationLimits', 'InterpretationPromptBuilder', 'InterpretationResult', 'InvestigationConfig', 'InvestigationError', 'InvestigationLimits', 'InvestigationResult', 'InvestigationToolRouter', 'Investigator', 'Job', 'LLMError', 'LLMGateway', 'LLMProvider', 'LLMRunResult', 'LifecycleError', 'LockBusy', 'OpenAICompatibleProvider', 'OutputError', 'OutputResult', 'OutputValidator', 'PackageFactory', 'PackageLimits', 'PackageReadError', 'PackageReader', 'PackageValidator', 'PackageWriteError', 'PackageWriter', 'PromptBuilder', 'ProviderConfig', 'ProviderResponse', 'ProviderUnavailable', 'QueryContext', 'ResearchError', 'ResearchLimits', 'ResearchService', 'SourceQueryTool', 'State', 'Telemetry', 'TelemetryConfig', 'TelemetryError', 'ToolError', 'ToolLimits', 'ValidationIssue', 'ValidationResult', 'VcoreImporter', 'Workspace', 'WowdiagAdapter', 'WowsrcAdapter', '_GatewayFilter', '_Handler', '_JSONFormatter', '_NoRedirect', '_PhaseLogger', '_Scope']) diff --git a/src/viano_bug_analysis/telemetry/config.py b/src/viano_bug_analysis/telemetry/config.py new file mode 100644 index 0000000..676b178 --- /dev/null +++ b/src/viano_bug_analysis/telemetry/config.py @@ -0,0 +1,40 @@ +import tomllib +from dataclasses import asdict, dataclass + + +class TelemetryError(Exception): + def __init__(self, code): + self.code = code + super().__init__(code) + + +@dataclass(frozen=True) +class TelemetryConfig: + heartbeat_seconds: int = 5 + snapshot_interval_seconds: int = 1 + heartbeat_stale_seconds: int = 30 + progress_warning_seconds: int = 300 + inventory_interval_seconds: int = 60 + inventory_stale_seconds: int = 180 + log_max_bytes: int = 5 * 1024**2 + log_backups: int = 5 + max_cases: int = 1000 + max_frames: int = 40 + max_snapshot_bytes: int = 4 * 1024**2 + + def __post_init__(self): + if (any(type(v) is not int or v <= 0 for v in asdict(self).values()) + or self.heartbeat_stale_seconds <= self.heartbeat_seconds + or self.inventory_stale_seconds <= self.inventory_interval_seconds): + raise TelemetryError("TELEMETRY_CONFIG") + + @classmethod + def load(cls, path): + try: + with open(path, "rb") as stream: + config = tomllib.load(stream) + if set(config) != {"telemetry"}: + raise TelemetryError("TELEMETRY_CONFIG") + return cls(**config["telemetry"]) + except (OSError, TypeError, ValueError): + raise TelemetryError("TELEMETRY_CONFIG") from None diff --git a/src/viano_bug_analysis/telemetry/core.py b/src/viano_bug_analysis/telemetry/core.py new file mode 100644 index 0000000..751862f --- /dev/null +++ b/src/viano_bug_analysis/telemetry/core.py @@ -0,0 +1,495 @@ +import contextvars +import json +import logging +import math +import os +import re +import sys +import tempfile +import threading +import time +import uuid +from collections import Counter +from contextlib import contextmanager +from datetime import datetime, timezone +from logging.handlers import RotatingFileHandler +from pathlib import Path + +from .config import TelemetryConfig, TelemetryError +from .inspection import assess, scan_workspace, scheduling, count_cases +from .safety import EVENTS, OPERATIONS, STATUSES, diagnostic, safe_code, detail, hashed, error_code + +_context = contextvars.ContextVar("vba_telemetry_context", default=None) + + +class _GatewayFilter(logging.Filter): + def __init__(self, service_id): + super().__init__() + self.service_id = service_id + + def filter(self, record): + # Replace legacy provider-name messages only inside this observed task. + return (_context.get() or {}).get("service_id") != self.service_id + + +def identity(value): + import hashlib + return hashlib.sha256(value.encode(errors="surrogatepass")).hexdigest() if isinstance(value, str) else None + + +class _Handler(RotatingFileHandler): + def handleError(self, record): + raise TelemetryError("TELEMETRY_STORAGE") + + +class _Scope: + def __init__(self, telemetry): + self.telemetry = telemetry + self.status = "SUCCESS" + self.code = None + self.reused = False + self.failed = False + self.detail = {} + + def result(self, result): + status = getattr(result, "status", "SUCCESS") + self.status = status if status in STATUSES else "SUCCESS" + code = getattr(result, "code", None) + self.code = safe_code(code) if code else None + self.reused = getattr(result, "reused", False) is True + if hasattr(result, "source"): + self.detail["source_key"] = identity(Path(result.source).name) + if self.status == "QUARANTINED" and getattr(result, "package", None) is not None: + self.detail["quarantine_key"] = Path(result.package).parent.name + self.telemetry.guard(self.telemetry.result_metrics, result) + if isinstance(result, list) and any(getattr(item, "status", None) == "ERROR" for item in result): + self.status = "ERROR" + + +class _PhaseLogger: + def __init__(self, telemetry): + self.telemetry = telemetry + + def _log(self, level, message, args): + if not self.telemetry.started: + from viano_bug_analysis.llm.logging import case_logger + getattr(case_logger(self.telemetry.workspace), level.lower())(message, *args) + return + index = message.split("code=%s")[0].count("%s") if isinstance(message, str) and "code=%s" in message else None + code = safe_code(args[index]) if index is not None and index < len(args) else None + if code is None and isinstance(message, str) and "no reachable LLM server" in message: + code = "LLM_UNAVAILABLE" + self.telemetry.emit("phase_diagnostic", level=level, code=code) + + def info(self, message, *args): + self._log("INFO", message, args) + + def warning(self, message, *args): + self._log("WARNING", message, args) + + def error(self, message, *args): + self._log("ERROR", message, args) + + +class Telemetry: + """Single-writer observation session. Never owns case state or recovery policy.""" + def __init__(self, workspace, config=None, *, package_limits=None, clock=None, monotonic=None): + self.workspace = workspace + self.config = config or TelemetryConfig() + self.package_limits = package_limits + self.clock = clock or (lambda: datetime.now(timezone.utc)) + self.monotonic = monotonic or time.monotonic + self.directory = workspace.runtime / "telemetry" + self.started = False + self.service_id = uuid.uuid4().hex + self._lock = threading.RLock() + self._stop = threading.Event() + self._thread = self._inventory_thread = self._handler = self._lease = None + self._active, self._metrics = {}, Counter() + self._cases = {} + self._case_progress = {} + self._budgets = {} + self._inventory = None + self._started_at = self._heartbeat = self._stopped_at = self._progress = None + self._last_error = None + self._failures = 0 + self._healthy = True + self._channels = {channel: True for channel in ("log", "snapshot", "scan", "observation")} + self._last_fallback = float("-inf") + self._last_scan = float("-inf") + self._last_flush = float("-inf") + self.logger = _PhaseLogger(self) + self._gateway_filter = _GatewayFilter(self.service_id) + + def _now(self): + now = self.clock() + if now.tzinfo is None or now.utcoffset() is None: + raise TelemetryError("TELEMETRY_CLOCK") + return now.isoformat() + + def __enter__(self): + return self.start() + + def __exit__(self, *args): + self.stop() + + def start(self): + from viano_bug_analysis.lifecycle.locking import FileLock + if self.started or self._stopped_at: + raise TelemetryError("TELEMETRY_SESSION") + self._lease = FileLock(self.workspace.runtime / "locks" / "telemetry.lock") + self._lease.__enter__() + try: + if self.directory.is_symlink(): + raise TelemetryError("TELEMETRY_PATH") + self.directory.mkdir(exist_ok=True) + self._check_paths() + log = self.directory / "service.jsonl" + if log.exists() and log.stat().st_size: + with log.open("rb") as stream: + stream.seek(-1, 2) + truncated = stream.read(1) != b"\n" + if truncated: + with log.open("ab") as stream: + stream.write(b"\n") + self._handler = _Handler(self.directory / "service.jsonl", maxBytes=self.config.log_max_bytes, + backupCount=self.config.log_backups, encoding="utf-8", delay=True) + self._handler.setFormatter(logging.Formatter("%(message)s")) + self._started_at = self._heartbeat = self._now() + self.started = True + logging.getLogger("viano_bug_analysis.llm").addFilter(self._gateway_filter) + self.emit("observer_started") + self._flush(force=True) + self._thread = threading.Thread(target=self._pulse, name="vba-telemetry", daemon=True) + self._thread.start() + self.refresh_inventory(force=True) + self._inventory_thread = threading.Thread(target=self._inventory_pulse, name="vba-inventory", daemon=True) + self._inventory_thread.start() + return self + except BaseException: + self._stop.set() + self.started = False + logging.getLogger("viano_bug_analysis.llm").removeFilter(self._gateway_filter) + if self._handler: + self._handler.close() + self._lease.__exit__() + raise + + def stop(self): + if not self.started: + return + self._stop.set() + if self._thread and threading.current_thread() is not self._thread: + self._thread.join(timeout=2) + if self._inventory_thread: + self._inventory_thread.join(timeout=2) + with self._lock: + self._stopped_at = self._heartbeat = self._now() + self.emit("observer_stopped") + self._flush(force=True) + self.started = False + logging.getLogger("viano_bug_analysis.llm").removeFilter(self._gateway_filter) + self._handler.close() + self._lease.__exit__() + + def _check_paths(self): + if self.directory.is_symlink(): + raise TelemetryError("TELEMETRY_PATH") + paths = [self.directory / name for name in ("service.jsonl", "status.json", "metrics.prom")] + paths += [self.directory / f"service.jsonl.{i}" for i in range(1, self.config.log_backups + 1)] + if any(path.is_symlink() for path in paths): + raise TelemetryError("TELEMETRY_PATH") + + def _failed(self, channel="observation"): + self._failures += 1 + self._channels[channel] = False + self._healthy = False + if self.monotonic() - self._last_fallback >= 30: + self._last_fallback = self.monotonic() + try: + sys.stderr.write('{"event":"telemetry_write_failed","code":"TELEMETRY_STORAGE"}\n') + except OSError: + pass + + def _metric(self, name, amount=1, **labels): + if not math.isfinite(amount) or amount < 0: + return + key = (name, tuple(sorted(labels.items()))) + if key not in self._metrics and len(self._metrics) >= 2000: + self._failed() + return + self._metrics[key] += amount + + def emit(self, event, *, level="INFO", code=None, status=None, context=None, extra=None): + if not self.started: + return + if event not in EVENTS or level not in {"INFO", "WARNING", "ERROR"}: + raise TelemetryError("TELEMETRY_EVENT") + with self._lock: + ctx = context or _context.get() or {} + record = {"format": "viano-service-event", "format_version": 1, "timestamp": self._now(), + "service_id": self.service_id, "level": level, "event": event, + "span_id": hashed(ctx.get("span_id"), 32), "parent_span_id": hashed(ctx.get("parent_span_id"), 32), + "operation": ctx.get("operation") if ctx.get("operation") in OPERATIONS else None, + "case_key": hashed(ctx.get("case_key")), "package_key": hashed(ctx.get("package_key")), + "status": status if status in STATUSES else None, "code": safe_code(code) if code else None} + if extra: + record["detail"] = detail(event, extra) + if level == "ERROR": + self._last_error = {key: record[key] for key in ("timestamp", "event", "span_id", "case_key", "code")} + try: + self._check_paths() + text = json.dumps(record, ensure_ascii=True, allow_nan=False) + if len(text.encode()) > min(self.config.max_snapshot_bytes, self.config.log_max_bytes): + raise TelemetryError("TELEMETRY_LIMIT") + self._handler.emit(logging.LogRecord("vba.telemetry", getattr(logging, level), "", 0, text, (), None)) + self._channels["log"] = True + self._healthy = all(self._channels.values()) + except Exception: + self._failed("log") + + @contextmanager + def operation(self, operation, case_id=None, package_id=None): + if operation not in OPERATIONS: + raise TelemetryError("TELEMETRY_OPERATION") + parent = _context.get() or {} + ctx = {"service_id": self.service_id, "span_id": uuid.uuid4().hex, "parent_span_id": parent.get("span_id"), "operation": operation, + "case_key": identity(case_id) or parent.get("case_key"), "package_key": identity(package_id) or parent.get("package_key")} + token = _context.set(ctx) + scope = _Scope(self) + begun = self.monotonic() + with self._lock: + self._active[ctx["span_id"]] = {**ctx, "started_at": self._now()} + self.emit("operation_started", context=ctx) + self._flush(force=ctx["parent_span_id"] is None) + try: + yield scope + except BaseException as error: + scope.failed = True + scope.status = "CANCELLED" if type(error).__name__ in {"CancelledError", "KeyboardInterrupt", "SystemExit"} else "BUSY" if type(error).__name__ == "LockBusy" else "ERROR" + scope.code = error_code(error) + self.emit("operation_failed", level="WARNING" if scope.status in {"BUSY", "CANCELLED"} else "ERROR", status=scope.status, code=scope.code, context=ctx, + extra={"exceptions": diagnostic(error, self.config.max_frames)}) + raise + finally: + elapsed = max(0, self.monotonic() - begun) + with self._lock: + self._active.pop(ctx["span_id"], None) + self._metric("vba_operations_total", operation=operation, status=scope.status) + self._metric("vba_operation_duration_seconds_count", operation=operation) + self._metric("vba_operation_duration_seconds_sum", elapsed, operation=operation) + if not scope.failed: + self.emit("operation_finished", level="ERROR" if scope.status == "ERROR" else "WARNING" if scope.status in {"BLOCKED", "REJECTED", "QUARANTINED"} else "INFO", + status=scope.status, code=scope.code, context=ctx, extra={"duration_seconds": elapsed, "reused": scope.reused, **scope.detail}) + self._flush(force=ctx["parent_span_id"] is None) + _context.reset(token) + + def exception(self, error): + self.emit("phase_diagnostic", level="ERROR", code=error_code(error), extra={"exceptions": diagnostic(error, self.config.max_frames)}) + + def guard(self, callback, *args): + try: + return callback(*args) + except Exception as error: + with self._lock: + self._failed() + try: + self.emit("phase_diagnostic", level="ERROR", code="TELEMETRY_CALLBACK", extra={"exceptions": diagnostic(error, self.config.max_frames)}) + except Exception: + pass + + def result_metrics(self, result): + with self._lock: + ctx = _context.get() or {} + if ctx.get("operation") == "inbox_poll" and isinstance(result, list): + for item in result: + status = getattr(item, "status", None) + if status in STATUSES: + self._metric("vba_inbox_results_total", status=status) + if ctx.get("operation") == "evidence_execute" and not getattr(result, "reused", True): + self._metric("vba_evidence_records_published_total", len(result.record_refs)) + if ctx.get("operation") == "budget_execute" and hasattr(result, "state"): + ident = (ctx.get("case_key"), ctx.get("package_key")) + if ident not in self._budgets and len(self._budgets) >= self.config.max_cases: + self._failed() + return + accounts = result.state["accounts"].values() + self._budgets[ident] = {"case_key": ident[0], "package_key": ident[1], "analysis_revision": result.analysis_revision, + **{field: sum(a[field] for a in accounts) for field in ("remaining", "spent", "rewarded")}, + "pending_actions": sum(action["status"] != "settled" for a in accounts for action in a["actions"].values())} + + def case_checkpoint(self, manifest, details=None): + if not self.started: + return + with self._lock: + case_key, package_key = identity(manifest["case_id"]), identity(manifest["package_id"]) + ident = (case_key, package_key) + if ident not in self._cases and len(self._cases) >= self.config.max_cases: + self._failed() + return + item = {"case_key": case_key, "package_key": package_key, "state": manifest["state"], + "package_generation": manifest["package_generation"], "analysis_revision": manifest["analysis_revision"], + "reason": safe_code(manifest["routing"]["reason"]) if manifest["routing"]["reason"] else None, + "retry_not_before": manifest["routing"]["retry_not_before"]} + item["scheduling"] = scheduling(manifest, details or {}, details if details and details.get("operation") == "llm-round" else None, self.clock()) + if item["scheduling"] == "delayed" and details and details.get("status") == "RETRY_PENDING": + item["retry_not_before"] = details["retry_not_before"] + previous = self._cases.get(ident) + if previous and previous["package_generation"] >= item["package_generation"]: + return + self._cases[ident] = item + self._progress = self._now() + self._case_progress[ident] = self._progress + self._metric("vba_case_checkpoints_total", state=item["state"]) + self.emit("case_checkpoint", context={**(_context.get() or {}), "case_key": case_key, "package_key": package_key}, extra=item) + + def llm_round(self, record): + if not self.started: + return + with self._lock: + phase, status = record["phase"], record["status"] + self._metric("vba_llm_rounds_total", phase=phase, status=status) + metadata = record["outcome"].get("metadata", {}) + for field in ("input_tokens", "output_tokens"): + value = metadata.get(field) + if type(value) is int and value >= 0: + self._metric("vba_llm_tokens_total", value, direction=field) + else: + self._metric("vba_llm_usage_unknown_total", field=field) + cost, currency = metadata.get("estimated_cost"), metadata.get("cost_currency") + currency_valid = isinstance(currency, str) and re.fullmatch("[A-Z]{3}", currency) is not None + if type(cost) in {int, float} and math.isfinite(cost) and cost >= 0 and currency_valid: + self._metric("vba_llm_estimated_cost_total", cost, currency=currency) + else: + self._metric("vba_llm_usage_unknown_total", field="estimated_cost") + attempts = metadata.get("attempts", record["outcome"].get("attempts", [])) + self._metric("vba_llm_provider_failures_total", len(attempts), phase=phase) + self._metric("vba_llm_failed_attempt_usage_unknown_total", len(attempts)) + self.emit("llm_round", level="INFO" if status == "SUCCESS" else "WARNING", status=status, code=record.get("code"), + extra={"phase": phase, "provider_key": identity(metadata.get("provider")), "model_key": identity(metadata.get("model")), + "input_tokens": metadata.get("input_tokens"), "output_tokens": metadata.get("output_tokens"), + "estimated_cost": cost if currency_valid else None, + "currency": currency if currency_valid else None, + "failed_providers": [{"provider_key": identity(a.get("provider")), "code": safe_code(a.get("code"))} for a in attempts]}) + + def refresh_inventory(self, *, force=False): + if not self.started or not force and self.monotonic() - self._last_scan < self.config.inventory_interval_seconds: + return + try: + with self.operation("inventory_scan"): + scanned = scan_workspace(self.workspace.root, limits=self.package_limits, max_cases=self.config.max_cases, now=self.clock(), cancelled=self._stop.is_set) + with self._lock: + self._inventory = scanned + for item in scanned["cases"]: + ident = (item["case_key"], item["package_key"]) + previous = self._cases.get(ident) + if previous and item["package_generation"] > previous["package_generation"]: + self._progress = self._now() + self._case_progress[ident] = self._progress + if ident in self._cases or len(self._cases) < self.config.max_cases: + if previous is None or previous["package_generation"] <= item["package_generation"]: + self._cases[ident] = item + self._channels["scan"] = scanned["complete"] + self._healthy = all(self._channels.values()) + self._last_scan = self.monotonic() + self.emit("inventory_updated", level="INFO" if scanned["complete"] else "WARNING", extra={"complete": scanned["complete"], "counts": scanned["counts"], "inbox_files": scanned["inbox_files"]}) + except Exception: + # Observation errors never change the analysis result. + with self._lock: + self._failed("scan") + + def snapshot(self): + from dataclasses import asdict + with self._lock: + now = self.clock() + active = [] + for item in self._active.values(): + elapsed = max(0, (now - datetime.fromisoformat(item["started_at"])).total_seconds()) + progressed = self._case_progress.get((item["case_key"], item["package_key"]), item["started_at"]) + progress_age = max(0, (now - datetime.fromisoformat(progressed)).total_seconds()) + active.append({**item, "elapsed_seconds": elapsed, "progress_age_seconds": progress_age, + "stalled": elapsed > self.config.progress_warning_seconds and progress_age > self.config.progress_warning_seconds}) + inventory = json.loads(json.dumps(self._inventory)) if self._inventory else None + if inventory: + rows = {(item["case_key"], item["package_key"]): item for item in inventory["cases"]} + for ident, item in self._cases.items(): + if ident not in rows or rows[ident]["package_generation"] <= item["package_generation"]: + rows[ident] = item + inventory["cases"] = list(rows.values()) + for row in inventory["cases"]: + if row.get("scheduling") == "delayed" and row.get("retry_not_before") and datetime.fromisoformat(row["retry_not_before"].replace("Z", "+00:00")) <= now: + row["scheduling"] = "ready" + inventory["counts"] = count_cases(inventory["cases"]) + data = {"format": "viano-service-status", "format_version": 1, "service_id": self.service_id, "pid": os.getpid(), + "started_at": self._started_at, "heartbeat_at": self._heartbeat, "stopped_at": self._stopped_at, + "last_progress_at": self._progress, "last_error": self._last_error, "telemetry_failures": self._failures, + "telemetry_healthy": self._healthy, "budgets": list(self._budgets.values()), + "active_operations": active, "cases": list(self._cases.values()), "inventory": inventory, + "config": asdict(self.config), "counters": [{"name": name, "labels": dict(labels), "value": value} for (name, labels), value in sorted(self._metrics.items())]} + return assess(data, self.clock()) + + def prometheus(self): + data = self.snapshot() + lines = [f"vba_observer_alive {int(data['observer'] == 'LIVE')}", f"vba_active_operations {len(data['active_operations'])}", + f"vba_telemetry_failures_total {self._failures}", f"vba_inventory_fresh {int(data['inventory_fresh'])}", + f"vba_observer_heartbeat_time_seconds {datetime.fromisoformat(self._heartbeat).timestamp()}", + f"vba_last_progress_time_seconds {datetime.fromisoformat(self._progress).timestamp() if self._progress else 0}", + f"vba_observer_start_time_seconds {datetime.fromisoformat(self._started_at).timestamp()}"] + for kind in ("remaining", "spent", "rewarded", "pending_actions"): + lines.append('vba_observed_budget{kind="' + kind + '"} ' + str(sum(row[kind] for row in data["budgets"]))) + for metric in data["counters"]: + labels = ",".join(name + "=" + json.dumps(value) for name, value in sorted(metric["labels"].items())) + lines.append(metric["name"] + ("{" + labels + "}" if labels else "") + " " + str(metric["value"])) + if data["inventory_fresh"]: + lines.append("vba_inbox_files " + str(data["inventory"]["inbox_files"])) + for category, value in data["inventory"]["counts"].items(): + lines.append('vba_cases_observed{category="' + category + '"} ' + str(value)) + return "\n".join(lines) + "\n" + + def _write(self, name, data): + temporary = None + try: + if len(data) > self.config.max_snapshot_bytes: + raise TelemetryError("TELEMETRY_LIMIT") + with tempfile.NamedTemporaryFile(dir=self.directory, prefix=".telemetry-", delete=False) as stream: + temporary = Path(stream.name) + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, self.directory / name) + finally: + if temporary: + temporary.unlink(missing_ok=True) + + def flush(self): + self._flush(force=True) + + def _flush(self, *, force=False): + if not self.started: + return + with self._lock: + if not force and self.monotonic() - self._last_flush < self.config.snapshot_interval_seconds: + return + try: + self._check_paths() + self._write("metrics.prom", self.prometheus().encode()) + self._channels["snapshot"] = True + self._healthy = all(self._channels.values()) + self._write("status.json", json.dumps(self.snapshot(), allow_nan=False).encode()) + self._last_flush = self.monotonic() + except Exception: + self._failed("snapshot") + + def _pulse(self): + while not self._stop.wait(self.config.heartbeat_seconds): + try: + with self._lock: + self._heartbeat = self._now() + self._flush(force=True) + except Exception: + with self._lock: + self._failed("snapshot") + + def _inventory_pulse(self): + while not self._stop.wait(self.config.inventory_interval_seconds): + self.refresh_inventory(force=True) diff --git a/src/viano_bug_analysis/telemetry/decorators.py b/src/viano_bug_analysis/telemetry/decorators.py new file mode 100644 index 0000000..b1d3f7c --- /dev/null +++ b/src/viano_bug_analysis/telemetry/decorators.py @@ -0,0 +1,38 @@ +import functools +import inspect + + +def observed(operation, *, identity=True): + """Transparent when telemetry is absent; no instrumentation controls a case.""" + def decorate(function): + def observer(instance): + return getattr(instance, "telemetry", None) or getattr(getattr(instance, "store", None), "telemetry", None) + + def ids(args, kwargs): + if not identity: + return None, None + return (kwargs.get("case_id", args[0] if args else None), + kwargs.get("package_id", args[1] if len(args) > 1 else None)) + + if inspect.iscoroutinefunction(function): + @functools.wraps(function) + async def wrapped(instance, *args, **kwargs): + telemetry = observer(instance) + if telemetry is None or not telemetry.started: + return await function(instance, *args, **kwargs) + with telemetry.operation(operation, *ids(args, kwargs)) as scope: + result = await function(instance, *args, **kwargs) + telemetry.guard(scope.result, result) + return result + else: + @functools.wraps(function) + def wrapped(instance, *args, **kwargs): + telemetry = observer(instance) + if telemetry is None or not telemetry.started: + return function(instance, *args, **kwargs) + with telemetry.operation(operation, *ids(args, kwargs)) as scope: + result = function(instance, *args, **kwargs) + telemetry.guard(scope.result, result) + return result + return wrapped + return decorate diff --git a/src/viano_bug_analysis/telemetry/inspection.py b/src/viano_bug_analysis/telemetry/inspection.py new file mode 100644 index 0000000..b872115 --- /dev/null +++ b/src/viano_bug_analysis/telemetry/inspection.py @@ -0,0 +1,164 @@ +import json +import re +from datetime import datetime, timezone +from pathlib import Path + +from .config import TelemetryError +from .safety import safe_code, error_code + + +def scheduling(manifest, checkpoint, call, now): + from viano_bug_analysis.lifecycle.state import State, ACTIVE + state = manifest["state"] + if state == State.BLOCKED: + if manifest["routing"]["reason"] == "AWAITING_TEST_FEEDBACK": + return "human_wait" + if checkpoint.get("automatic_retry") is False: + return "manual_wait" + due = manifest["routing"]["retry_not_before"] + return "delayed" if due and datetime.fromisoformat(due.replace("Z", "+00:00")) > now else "ready" + if state in ACTIVE: + return "delayed" if call and call.get("status") == "RETRY_PENDING" and datetime.fromisoformat(call["retry_not_before"].replace("Z", "+00:00")) > now else "ready" + return {State.OUTPUT_READY: "output_ready", State.FAILED_TO_ANALYZE: "review", State.HANDED_OFF: "handed_off"}.get(state) + + +def count_cases(items): + from .safety import CATEGORIES + counts = {category: 0 for category in CATEGORIES} + for item in items: + if item["state"] == "BLOCKED": + counts["blocked"] += 1 + if item.get("scheduling") in counts: + counts[item["scheduling"]] += 1 + return counts + + +def utc_now(): + return datetime.now(timezone.utc) + + +def age(value, now): + if value is None: + return None + return max(0.0, (now - datetime.fromisoformat(value.replace("Z", "+00:00"))).total_seconds()) + + +def assess(snapshot, now=None): + now = now or utc_now() + result = dict(snapshot) + if not snapshot.get("heartbeat_at"): + result.update(observer="NOT_STARTED", work="QUEUE_UNKNOWN", health="UNKNOWN") + return result + heartbeat_age = age(snapshot["heartbeat_at"], now) + config = snapshot["config"] + result["heartbeat_age_seconds"] = heartbeat_age + result["progress_age_seconds"] = age(snapshot.get("last_progress_at") or snapshot["started_at"], now) + result["observer"] = "STOPPED" if snapshot.get("stopped_at") else "HEARTBEAT_STALE" if heartbeat_age > config["heartbeat_stale_seconds"] else "LIVE" + queue = snapshot.get("inventory") + queue_fresh = bool(queue and queue["complete"] and age(queue["observed_at"], now) <= config["inventory_stale_seconds"]) + result["inventory_fresh"] = queue_fresh + active = any(item["operation"] != "inventory_scan" for item in snapshot["active_operations"]) + if active: + work = "NO_RECENT_PROGRESS" if any(item.get("stalled") for item in snapshot["active_operations"] if item["operation"] != "inventory_scan") else "WORKING" + elif not queue_fresh: + work = "QUEUE_UNKNOWN" + elif queue["inbox_files"]: + work = "INPUT_PENDING" + elif queue["counts"].get("ready", 0): + work = "READY_NOT_PROCESSING" + elif queue["counts"].get("blocked", 0) or queue["counts"].get("delayed", 0): + work = "WAITING_EXTERNAL" + else: + work = "IDLE" + result["work"] = work + recent_error = snapshot["last_error"] and age(snapshot["last_error"]["timestamp"], now) <= config["progress_warning_seconds"] + result["health"] = "STOPPED" if result["observer"] == "STOPPED" else "HEARTBEAT_STALE" if result["observer"] != "LIVE" else "DEGRADED" if not snapshot["telemetry_healthy"] or recent_error or queue and not queue["complete"] else "NO_RECENT_PROGRESS" if work == "NO_RECENT_PROGRESS" else "OK" + return result + + +def read_status(root, *, now=None, limit=4 * 1024**2): + from viano_bug_analysis.package.manifest import parse_manifest + path = Path(root) / "runtime" / "telemetry" / "status.json" + if not path.exists(): + return {"format": "viano-service-status", "format_version": 1, "observer": "NOT_STARTED", "work": "QUEUE_UNKNOWN", "health": "UNKNOWN"} + try: + if path.parent.is_symlink() or path.parent.parent.is_symlink() or path.is_symlink() or path.stat().st_size > limit: + raise TelemetryError("TELEMETRY_READ") + with path.open("rb") as stream: + data = parse_manifest(stream.read(limit + 1)) + if len(json.dumps(data)) > limit * 2: + raise TelemetryError("TELEMETRY_READ") + from viano_bug_analysis.case.contracts import validate + validate(data, "service-status") + return assess(data, now) + except Exception as error: + if isinstance(error, TelemetryError): + raise + raise TelemetryError("TELEMETRY_READ") from None + + +def scan_workspace(root, *, limits=None, max_cases=1000, now=None, cancelled=None): + """Passive bounded inventory. Filenames select candidates, never identity/state.""" + from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain + from viano_bug_analysis.lifecycle.storage import key + from viano_bug_analysis.lifecycle.state import State, ACTIVE + from viano_bug_analysis.package import PackageReader + root = Path(root).resolve() + now = now or utc_now() + counts = {name: 0 for name in ("ready", "blocked", "delayed", "human_wait", "manual_wait", "output_ready", "review", "handed_off")} + result = {"observed_at": now.isoformat(), "complete": True, "inbox_files": 0, "counts": counts, "cases": [], "errors": []} + if type(max_cases) is not int or max_cases <= 0: + raise TelemetryError("TELEMETRY_CONFIG") + inbox, cases = root / "inbox", root / "cases" + if inbox.is_symlink() or cases.is_symlink(): + raise TelemetryError("TELEMETRY_SCAN_PATH") + if inbox.exists(): + result["inbox_files"] = sum(1 for p in inbox.iterdir() if p.is_file() or p.is_symlink()) + if not cases.exists(): + return result + seen = 0 + for parent in cases.iterdir(): + if parent.is_symlink() or not parent.is_dir() or not re.fullmatch("[0-9a-f]{64}", parent.name): + result["complete"] = False + continue + for directory in parent.iterdir(): + if cancelled is not None and cancelled(): + result["complete"] = False + result["counts"] = count_cases(result["cases"]) + return result + seen += 1 + if seen > max_cases: + result["complete"] = False + result["counts"] = count_cases(result["cases"]) + return result + try: + if directory.is_symlink() or not directory.is_dir() or not re.fullmatch("[0-9a-f]{64}", directory.name): + raise TelemetryError("TELEMETRY_SCAN_PATH") + candidate = max(directory.glob("g*.zip"), default=None) + if candidate is None: + continue + if candidate.is_symlink(): + raise TelemetryError("TELEMETRY_SCAN_PATH") + with PackageReader(candidate, limits) as reader: + m = reader.manifest + if (key(m["case_id"]) != parent.name or key(m["package_id"]) != directory.name + or candidate.name != f"g{m['package_generation']:08d}.zip"): + raise TelemetryError("TELEMETRY_SCAN_IDENTITY") + chain = checkpoint_chain(reader) + if not chain or chain[0][1]["package_generation"] != m["package_generation"]: + raise TelemetryError("TELEMETRY_SCAN_CHECKPOINT") + state = m["state"] + reason = safe_code(m["routing"]["reason"]) if m["routing"]["reason"] else None + item = {"case_key": parent.name, "package_key": directory.name, "state": state, + "analysis_revision": m["analysis_revision"], "package_generation": m["package_generation"], + "reason": reason, "retry_not_before": m["routing"]["retry_not_before"]} + previous = next((cp["details"] for _, cp in chain if cp["analysis_revision"] == m["analysis_revision"] and cp["details"].get("operation") == "llm-round" and cp["details"].get("phase") == state), None) + item["scheduling"] = scheduling(m, chain[0][1]["details"], previous, now) + if item["scheduling"] == "delayed" and previous and state in ACTIVE: + item["retry_not_before"] = previous["retry_not_before"] + result["cases"].append(item) + except Exception as error: + result["complete"] = False + result["errors"].append({"case_key": parent.name, "package_key": directory.name if re.fullmatch("[0-9a-f]{64}", directory.name) else None, "code": error_code(error)}) + result["counts"] = count_cases(result["cases"]) + return result diff --git a/src/viano_bug_analysis/telemetry/safety.py b/src/viano_bug_analysis/telemetry/safety.py new file mode 100644 index 0000000..2a28141 --- /dev/null +++ b/src/viano_bug_analysis/telemetry/safety.py @@ -0,0 +1,118 @@ +import errno +import re +import traceback + +from .codes import KNOWN_CODES, KNOWN_MODULES, KNOWN_CLASSES +import math + +EVENTS = frozenset({"observer_started", "observer_stopped", "operation_started", "operation_finished", "operation_failed", "telemetry_read_warning", + "case_checkpoint", "llm_round", "phase_diagnostic", "inventory_updated", "telemetry_write_failed"}) +OPERATIONS = frozenset({"case_build", "case_interpret", "hypotheses_build", "investigation_run", "investigation_step", + "diagnosis_build", "change_request_build", "output_validate", "output_finalize", "output_publish", + "llm_run", "checkpoint_advance", "intake", "inbox_poll", "inventory_scan", "budget_execute", + "evidence_execute", "async_test_submit", "async_test_confirm"}) +STATUSES = frozenset({"SUCCESS", "ERROR", "BUSY", "CANCELLED", "COMPLETED", "INCONCLUSIVE", "BLOCKED", "RETRY_PENDING", + "DEFERRED", "PROGRESSED", "REJECTED", "YIELDED", "ACCEPTED", "DUPLICATE", "RESUMED", "QUARANTINED", + "VALID", "REVIEW", "OUTPUT_READY", "HANDED_OFF", "INFO"}) + + +def safe_code(value): + own = {"TELEMETRY_CONFIG", "TELEMETRY_READ", "TELEMETRY_PATH", "TELEMETRY_STORAGE", "TELEMETRY_LIMIT", "TELEMETRY_SCAN_PATH", "TELEMETRY_SCAN_IDENTITY", "TELEMETRY_SCAN_CHECKPOINT", "TELEMETRY_CLOCK", "TELEMETRY_EVENT", "TELEMETRY_SESSION", "TELEMETRY_OPERATION"} + return value if isinstance(value, str) and value in KNOWN_CODES | own else "UNCLASSIFIED_ERROR" + + +def attribute(value, name): + try: + return getattr(value, name, None) + except Exception: + return None + + +def error_code(error): + code = attribute(error, "code") + issues = attribute(attribute(error, "result"), "issues") + if code is None and isinstance(issues, (tuple, list)) and issues: + code = attribute(issues[0], "code") + return safe_code(code) + + +def diagnostic(error, max_frames): + """No exception messages, paths, source lines, arguments or local values.""" + result, seen = [], set() + while error is not None and id(error) not in seen and len(result) < 8: + seen.add(id(error)) + module = type(error).__module__ + trusted = module == "builtins" or module in KNOWN_MODULES and type(error).__name__ in KNOWN_CLASSES + frames = [] + tb = attribute(error, "__traceback__") + while tb is not None and len(frames) < max_frames: + frame_module = tb.tb_frame.f_globals.get("__name__", "") + own = frame_module in KNOWN_MODULES + frames.append({"module": frame_module if own else "external", "function": tb.tb_frame.f_code.co_name if own else "external", + "line": tb.tb_lineno}) + tb = tb.tb_next + os_errno = attribute(error, "errno") + keyword = attribute(error, "validator") + result.append({"type": type(error).__name__ if trusted else "ExternalError", "code": error_code(error), + "schema_keyword": keyword if isinstance(keyword, str) and keyword in {"type", "required", "additionalProperties", "oneOf", "anyOf", "const", "enum", "pattern", "minLength", "maxLength", "minItems", "maxItems", "minimum", "maximum"} else None, + "errno": os_errno if type(os_errno) is int and os_errno in errno.errorcode else None, "frames": frames}) + # Suppressed context remains useful to operators without exposing its text. + error = attribute(error, "__cause__") or attribute(error, "__context__") + return result + + +STATES = frozenset({"NEW", "VALIDATED", "BUILDING_CASE", "INTERPRETING_CASE", "GENERATING_HYPOTHESES", "INVESTIGATING_HYPOTHESES", "BUILDING_DIAGNOSIS", "OUTPUT_READY", "HANDED_OFF", "BLOCKED", "FAILED_TO_ANALYZE", "REOPENED"}) +CATEGORIES = frozenset({"ready", "blocked", "delayed", "human_wait", "manual_wait", "output_ready", "review", "handed_off"}) + + +def hashed(value, length=64): + return value if isinstance(value, str) and re.fullmatch("[0-9a-f]{" + str(length) + "}", value) else None + + +def number(value): + return value if type(value) in {int, float} and math.isfinite(value) and value >= 0 else None + + +def detail(event, value): + """Strict field allowlists even for callers emitting their own events.""" + if not isinstance(value, dict): + return {} + if event == "telemetry_read_warning": + return {"corrupt_records": number(value.get("corrupt_records"))} + result = {} + if event == "operation_finished": + return {"duration_seconds": number(value.get("duration_seconds")), "reused": value.get("reused") is True, + "source_key": hashed(value.get("source_key")), "quarantine_key": hashed(value.get("quarantine_key"), 32)} + if event == "inventory_updated": + return {"complete": value.get("complete") is True, "inbox_files": number(value.get("inbox_files")), + "counts": {k: number(v) for k, v in value.get("counts", {}).items() if k in CATEGORIES}} + if event == "case_checkpoint": + return {"case_key": hashed(value.get("case_key")), "package_key": hashed(value.get("package_key")), + "state": value.get("state") if value.get("state") in STATES else None, + "package_generation": number(value.get("package_generation")), "analysis_revision": number(value.get("analysis_revision")), + "reason": safe_code(value.get("reason")) if value.get("reason") else None} + if event == "llm_round": + return {"phase": value.get("phase") if value.get("phase") in STATES else None, + "provider_key": hashed(value.get("provider_key")), "model_key": hashed(value.get("model_key")), + "input_tokens": number(value.get("input_tokens")), "output_tokens": number(value.get("output_tokens")), + "estimated_cost": number(value.get("estimated_cost")), + "currency": value.get("currency") if isinstance(value.get("currency"), str) and re.fullmatch("[A-Z]{3}", value["currency"]) else None, + "failed_providers": [{"provider_key": hashed(a.get("provider_key")), "code": safe_code(a.get("code"))} for a in value.get("failed_providers", [])[:64] if isinstance(a, dict)]} + if event in {"operation_failed", "phase_diagnostic"}: + chains = [] + for error in value.get("exceptions", [])[:8]: + if not isinstance(error, dict): + continue + frames = [] + for frame in error.get("frames", [])[:100]: + if isinstance(frame, dict): + module = frame.get("module") if frame.get("module") in KNOWN_MODULES else "external" + function = frame.get("function") + frames.append({"module": module, "function": function if module != "external" and isinstance(function, str) and re.fullmatch("[a-zA-Z_][a-zA-Z_0-9]{0,99}|", function) else "external", "line": number(frame.get("line"))}) + cls = error.get("type") + import builtins + chains.append({"type": cls if cls in KNOWN_CLASSES or isinstance(cls, str) and isinstance(getattr(builtins, cls, None), type) and issubclass(getattr(builtins, cls), BaseException) else "ExternalError", + "schema_keyword": error.get("schema_keyword") if error.get("schema_keyword") in {"type", "required", "additionalProperties", "oneOf", "anyOf", "const", "enum", "pattern", "minLength", "maxLength", "minItems", "maxItems", "minimum", "maximum"} else None, + "code": safe_code(error.get("code")), "errno": error.get("errno") if type(error.get("errno")) is int and error["errno"] in errno.errorcode else None, "frames": frames}) + return {"exceptions": chains} + return result diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py new file mode 100644 index 0000000..0c7b2a3 --- /dev/null +++ b/tests/test_telemetry.py @@ -0,0 +1,261 @@ +import asyncio,json,time +from dataclasses import replace +from datetime import datetime,timedelta,timezone +import pytest +from viano_bug_analysis.telemetry import Telemetry,TelemetryConfig,TelemetryError,read_status,scan_workspace +from viano_bug_analysis.telemetry.__main__ import main,read_events +from viano_bug_analysis.lifecycle import Workspace,CheckpointStore,State,InboxProcessor,LockBusy +from viano_bug_analysis.telemetry.decorators import observed +from test_case_builder import case_setup +from test_case_interpreter import FixtureAdapter,gateway +from viano_bug_analysis.analysis import CaseInterpreter + +@pytest.fixture +def session(tmp_path): + ws=Workspace(tmp_path/'work') + with Telemetry(ws,TelemetryConfig(heartbeat_seconds=1,heartbeat_stale_seconds=3,inventory_interval_seconds=2,inventory_stale_seconds=5)) as telemetry: + yield telemetry + +def test_live_idle_clean_stop_and_readonly_commands(session,capsys): + ws=session.workspace + status=read_status(ws.root) + assert status['observer']=='LIVE' and status['work']=='IDLE' + assert main(['status','--workspace',str(ws.root)])==0 + assert json.loads(capsys.readouterr().out)['observer']=='LIVE' + assert main(['metrics','--workspace',str(ws.root)])==0 + assert 'vba_observer_alive 1' in capsys.readouterr().out + session.stop();status=read_status(ws.root);assert status['observer']=='STOPPED' + +def test_heartbeat_independent_of_long_operation(session): + before=read_status(session.workspace.root)['heartbeat_at'] + with session.operation('case_build','private-id','private-package'): + time.sleep(1.2) + status=read_status(session.workspace.root) + assert status['heartbeat_at']!=before and status['work']=='WORKING' + assert status['active_operations'][0]['case_key']!='private-id' + +def test_error_chain_suppressed_context_and_no_sensitive_values(session): + from viano_bug_analysis.case.errors import CaseBuildError + secret='sk-private-value-123' + with pytest.raises(CaseBuildError): + with session.operation('case_build',secret,secret): + try:raise RuntimeError(secret) + except RuntimeError:raise CaseBuildError('CASE_NOT_FOUND',secret) from None + records=read_events(session.workspace.root,errors=True) + assert records[-1]['code']=='CASE_NOT_FOUND' + exceptions=records[-1]['detail']['exceptions'] + assert len(exceptions)==2 and exceptions[1]['type']=='RuntimeError' + assert secret not in (session.directory/'service.jsonl').read_text() + assert secret not in (session.directory/'status.json').read_text() + assert secret not in (session.directory/'metrics.prom').read_text() + assert read_status(session.workspace.root)['health']=='DEGRADED' + +def test_fail_open_on_sink_failure(session,monkeypatch,capsys): + original=session._write + monkeypatch.setattr(session,'_write',lambda *a:(_ for _ in ()).throw(OSError('secret filesystem details'))) + with session.operation('case_build'):pass + assert session.snapshot()['telemetry_failures']>0 + assert 'secret filesystem details' not in capsys.readouterr().err + monkeypatch.setattr(session,'_write',original);session.flush() + assert session.snapshot()['telemetry_healthy'] + +def test_passive_inventory_does_not_resume_or_change_packages(session,make_package): + store=CheckpointStore(session.workspace,telemetry=session) + status,path=store.intake(make_package());original=path.read_bytes() + scan=scan_workspace(session.workspace.root) + assert scan['counts']['ready']==1 and path.read_bytes()==original + store.advance('case-A','stable-package',State.BUILDING_CASE,step_id='build') + blocked=store.advance('case-A','stable-package',State.BLOCKED,step_id='blocked',reason='LLM_UNAVAILABLE',retry_not_before=(datetime.now(timezone.utc)+timedelta(hours=6)).isoformat(),details={'automatic_retry':False}) + before=blocked.read_bytes();session.refresh_inventory(force=True) + status=read_status(session.workspace.root) + assert status['work']=='WAITING_EXTERNAL' and status['inventory']['counts']['manual_wait']==1 + assert blocked.read_bytes()==before and store.latest('case-A','stable-package')==blocked + +def test_schema_heartbeat_stale_and_queue_unknown(session): + status=read_status(session.workspace.root,now=datetime.now(timezone.utc)+timedelta(seconds=10)) + assert status['observer']=='HEARTBEAT_STALE' and status['work']=='QUEUE_UNKNOWN' + +def test_rotation_and_read_event_filter(tmp_path): + ws=Workspace(tmp_path/'work') + with Telemetry(ws,TelemetryConfig(log_max_bytes=1500,log_backups=2)) as t: + for _ in range(15): + with t.operation('case_build','case','pack'):pass + assert (t.directory/'service.jsonl.1').exists() + assert len(list(t.directory.glob('service.jsonl*')))<=3 + assert read_events(ws.root,case_id='case',max_bytes=1500,backups=2) + assert not read_events(ws.root,case_id='other',max_bytes=1500,backups=2) + +def test_single_writer_lease(session): + with pytest.raises(LockBusy):Telemetry(session.workspace).start() + +@pytest.mark.parametrize('value',[0,-1,True]) +def test_config_rejects_bad_defaults(value): + with pytest.raises(TelemetryError):TelemetryConfig(heartbeat_seconds=value) + +def test_instrumented_interpreter_and_llm_usage_and_replay(case_setup): + store,builder,*_=case_setup + with Telemetry(store.workspace) as t: + store.telemetry=t + builder.build('case-A','stable-package') + interpreter=CaseInterpreter(store,gateway(FixtureAdapter())) + result=asyncio.run(interpreter.interpret('case-A','stable-package')) + with open(result.package,'rb') as f:before=f.read() + assert asyncio.run(interpreter.interpret('case-A','stable-package')).reused + assert result.package.read_bytes()==before + data=t.prometheus() + assert 'vba_llm_rounds_total{phase="INTERPRETING_CASE",status="SUCCESS"} 1' in data + assert 'vba_llm_tokens_total{direction="input_tokens"} 100' in data + assert 'vba_llm_usage_unknown_total{field="estimated_cost"} 1' in data + events=read_events(store.workspace.root,limit=1000) + assert any(r['event']=='case_checkpoint' and r['case_key'] for r in events) + assert not (store.workspace.runtime/'logs'/'llm.jsonl').exists() + +def test_inbox_error_is_visible_without_free_message(session,monkeypatch,tmp_path): + queue=InboxProcessor(session.workspace,telemetry=session) + source=session.workspace.inbox/'one.zip';source.write_bytes(b'x') + def fail(*a):raise OSError('api_key=SECRET') + monkeypatch.setattr(queue,'_snapshot',fail) + result=queue.process(source);assert result.status=='ERROR' + records=read_events(session.workspace.root,errors=True) + assert records and 'SECRET' not in json.dumps(records) + +def test_full_pipeline_observed_without_telemetry_artifacts_or_double_accounting(tool_case): + from viano_bug_analysis.analysis import Investigator,DiagnosisBuilder,ChangeRequestBuilder,OutputValidator + from viano_bug_analysis.budget import BudgetManager + from test_investigator import LoopAdapter,CASE + from test_diagnosis_builder import FinalAdapter + from test_change_request_builder import RequestAdapter + store=tool_case[0] + with Telemetry(store.workspace) as t: + store.telemetry=t + async def process(): + await Investigator(store,gateway(LoopAdapter())).run(*CASE,max_steps=20) + await DiagnosisBuilder(store,gateway(FinalAdapter(True))).build(*CASE) + g=gateway(RequestAdapter()) + await ChangeRequestBuilder(store,g).build(*CASE) + return await OutputValidator(store,g).publish(*CASE) + result=asyncio.run(process()) + assert result.status=='OUTPUT_READY' + from viano_bug_analysis.package import PackageReader + with PackageReader(result.package) as r: + assert not any(name.startswith('telemetry/') or name.endswith('service.jsonl') for name in r.files) + data=t.prometheus() + assert 'vba_evidence_records_published_total 3' in data + assert 'vba_llm_tokens_total{direction="input_tokens"} 800' in data + snapshot=t.snapshot() + assert snapshot['budgets'][0]['spent']==30 + assert snapshot['budgets'][0]['remaining']==BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['remaining']+180 + assert not snapshot['active_operations'] + +def test_failover_error_no_legacy_provider_identifier_output(case_setup,caplog): + from viano_bug_analysis.llm import ProviderUnavailable + from viano_bug_analysis.llm.runner import CaseLLMRunner + store,builder,*_=case_setup + with Telemetry(store.workspace) as t: + store.telemetry=t;builder.build('case-A','stable-package') + secret='provider-private-secret-value' + adapter=FixtureAdapter(ProviderUnavailable('LLM_UNAVAILABLE')) + g=gateway(adapter) + g.config=replace(g.config,providers=(replace(g.config.providers[0],id=secret),),max_attempts=1) + result=asyncio.run(CaseInterpreter(store,g).interpret('case-A','stable-package')) + assert result.status=='BLOCKED' + assert secret not in (t.directory/'service.jsonl').read_text() + assert secret not in caplog.text + assert 'vba_llm_provider_failures_total{phase="INTERPRETING_CASE"} 1' in t.prometheus() + +def test_explicit_payload_fields_are_dropped_and_unknown_code_is_masked(session): + session.emit('phase_diagnostic',level='ERROR',code='CUSTOM_SECRET_CODE',extra={'prompt':'SECRET','password':'SECRET','message':'SECRET'}) + records=read_events(session.workspace.root,errors=True) + assert records[-1]['code']=='UNCLASSIFIED_ERROR' + assert 'SECRET' not in (session.directory/'service.jsonl').read_text() + +def test_active_status_and_metrics_do_not_expose_case_labels(session): + with session.operation('case_build','CASE_SECRET','PACKAGE_SECRET'): + data=session.prometheus() + assert 'CASE_SECRET' not in data and 'PACKAGE_SECRET' not in data + assert not read_events(session.workspace.root,case_id='other') + +def test_inventory_limit_and_corrupt_package_are_explicit(session,make_package): + store=CheckpointStore(session.workspace);_,path=store.intake(make_package()) + path.write_bytes(b'broken') + scan=scan_workspace(session.workspace.root) + assert not scan['complete'] and scan['errors'] + +def test_cancelled_async_operation_has_no_stale_active_phase(session): + class App: + store=type('Store',(),{'telemetry':session})() + @observed('case_build') + async def build(self,case_id,package_id):raise asyncio.CancelledError() + with pytest.raises(asyncio.CancelledError):asyncio.run(App().build('case','package')) + assert not session.snapshot()['active_operations'] + assert 'status="CANCELLED"' in session.prometheus() + +def test_noop_without_running_session(tmp_path): + workspace=Workspace(tmp_path/'work');t=Telemetry(workspace) + class App: + store=type('Store',(),{'telemetry':t})() + @observed('case_build') + def build(self,case_id,package_id):return 42 + assert App().build('case','pack')==42 and not t.directory.exists() +from test_investigation_tools import tool_case + +def test_unfinished_log_record_survives_restart(tmp_path): + ws=Workspace(tmp_path/'work') + with Telemetry(ws):pass + path=ws.runtime/'telemetry'/'service.jsonl' + with path.open('ab') as stream:stream.write(b'{"event":') + with Telemetry(ws):pass + rows=read_events(ws.root) + assert any(r['event']=='observer_started' for r in rows) + assert rows[-1]['code']=='TELEMETRY_LOG_CORRUPT' + +def test_log_channel_failure_does_not_claim_healthy_snapshots(session,monkeypatch): + original=session._handler.emit + def fail(*args):raise OSError('SECRET') + monkeypatch.setattr(session._handler,'emit',fail) + session.emit('phase_diagnostic');session.flush() + assert not read_status(session.workspace.root)['telemetry_healthy'] + monkeypatch.setattr(session._handler,'emit',original) + session.emit('phase_diagnostic');session.flush() + assert read_status(session.workspace.root)['telemetry_healthy'] + +def test_per_case_stall_not_hidden_by_other_case_progress(tmp_path): + ws=Workspace(tmp_path/'work');now=[datetime.now(timezone.utc)] + with Telemetry(ws,TelemetryConfig(progress_warning_seconds=10),clock=lambda:now[0]) as t: + with t.operation('case_build','A','PA'): + now[0]+=timedelta(seconds=20) + t.case_checkpoint({'case_id':'B','package_id':'PB','state':'VALIDATED','package_generation':1,'analysis_revision':1,'routing':{'reason':None,'retry_not_before':None}}) + status=t.snapshot() + assert status['work']=='NO_RECENT_PROGRESS' and status['active_operations'][0]['stalled'] + +def test_checkpoint_updates_ready_queue_without_rescan(session,make_package): + store=CheckpointStore(session.workspace,telemetry=session) + store.intake(make_package()) + assert session.snapshot()['work']=='READY_NOT_PROCESSING' + +def test_quarantine_event_links_local_diagnostic_without_source_name(session): + queue=InboxProcessor(session.workspace,telemetry=session) + source=session.workspace.inbox/'SECRET-source.bin';source.write_bytes(b'invalid ZIP') + result=queue.process(source);assert result.status=='QUARANTINED' + rows=read_events(session.workspace.root,source_name='SECRET-source.bin') + assert rows[-1]['detail']['quarantine_key']==result.package.parent.name + assert 'SECRET-source.bin' not in (session.directory/'service.jsonl').read_text() + +def test_config_file_and_not_started_status(tmp_path): + config=tmp_path/'telemetry.toml';config.write_text('[telemetry]\nheartbeat_seconds=2\nheartbeat_stale_seconds=10\n') + assert TelemetryConfig.load(config).heartbeat_seconds==2 + config.write_text('[telemetry]\nunknown=true\n') + with pytest.raises(TelemetryError):TelemetryConfig.load(config) + assert read_status(tmp_path)['observer']=='NOT_STARTED' + +def test_bad_result_introspection_does_not_change_method_result(session): + class Result: + @property + def status(self):raise RuntimeError('SECRET') + result=Result() + class App: + store=type('Store',(),{'telemetry':session})() + @observed('case_build') + def build(self,*a):return result + assert App().build('A','B') is result + assert session.snapshot()['telemetry_failures']>0