diff --git a/.mcp.json b/.mcp.json new file mode 100644 index 0000000..4f41b7f --- /dev/null +++ b/.mcp.json @@ -0,0 +1,8 @@ +{ + "mcpServers": { + "spanledger": { + "command": "python", + "args": ["-m", "spanledger", "mcp"] + } + } +} diff --git a/DECISIONS.md b/DECISIONS.md index fbb8e42..6bbaa01 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -12,4 +12,6 @@ Changelog of targeted deviations from PROJECT_PLAN.md (per §13.3). - 2026-07-21 — **`spanledger_probe_e2e_latency_seconds` now populated** by `verify.LatencySampler`, implementing spike S1's method: poll the Query API for a freshly-emitted probe until it is visible, then record `emit→queryable` at poll resolution. Deliberately independent of `maturity_delay` (which floors loss verdicts at 30s but must not floor latency) — it polls a just-sent probe immediately. Honest by construction: a real measured delay; if a sampled probe never becomes visible within the timeout (e.g. during an outage) nothing is recorded, never a guessed value. Kept inside `verify.py` (not a new top-level module) to preserve the §6 module list. Live-verified 2026-07-21 (~6s emit→queryable). - 2026-07-21 — **M4 correlation hint built (first cut), gated behind `correlation.enabled` (default off).** On a loss finding it queries the `otelcol_*` loss counters over the gap window and attaches the largest positive delta as `correlation_hint` (`{metric, delta, hop}`); `hop` is the coarse component parsed from the metric name (receiver/processor/exporter). Requires the user to scrape collector self-metrics (`docs/setup-collector-metrics.md`), so it is opt-in — off by default to avoid wasted queries. The v5 metrics query shape (`signal: metrics`, `requestType: scalar`, `increase`/`sum` aggregation) is **validated live against the running SigNoz** (accepted, returns success), but the *hint end-to-end* is not yet exercised because the demo pipeline does not scrape otelcol metrics. Strictly off the detection path: any query failure/empty result degrades to no hint via `select_top_anomaly`, never to a false or missing loss verdict. Per-collector `hop` attribution (vs component-name) is a V2 job (the conservation ledger). - 2026-07-22 — **V2 Phase 1 (`events.py`): `ReliabilityEvent.to_json()` omits `severity`/`links` for the four V1 finding classes**, deviating from the PHASE_1_FOUNDATION.md module-3 prose ("new fields severity, links added top-level"). The two testable, higher-priority requirements in the same document — the golden-file lock (`test/golden/finding_v1.json`) and "`/findings` byte-identical to the golden file" (P1-7 gate) — both require the wire form to stay exactly the V1 §4.2 shape with no additions. `severity` and `links` remain ordinary attributes on the `ReliabilityEvent` object (used internally, e.g. `recovery.links == [loss_id]`); they just aren't injected into `to_json()`'s flattened dict. No behavior change to `/findings` or the golden fixture; this only resolves an internal doc inconsistency in favor of the frozen wire contract. -- 2026-07-22 — **`entry_refused` and `verification_stalled` event classes are declared but not constructed in Phase 1**, matching V1's own precedent (both existed only as unused constants in `findings.py` before this refactor — no runner code path ever built one). `backend_unreachable` **is** newly wired (`EventEngine.report_backend_unreachable`, called from `runner._verify_stream` on `SignozQueryError`) since Phase 1's edge case 2 explicitly requires it and has a test. Its payload is a minimal ad hoc shape (`id`, `stream`, `epoch`, `class`, `detail`) rather than the full V1 §4.2 finding schema — there is no golden fixture for it (only `loss` has one), so there is no historical byte-shape to preserve. Extending `entry_refused`/`verification_stalled` with real call sites, and giving `backend_unreachable` a fuller schema if `/api/v2/events` consumers need one, is a V2 job. +- 2026-07-24 — **M2 demo physics values pinned:** `toxic_type: bandwidth`, `proxy: gateway`, `rate: 8 KB/s`, `duration: 90s`, `agent_queue_size: 64`, `agent_max_elapsed_time: 30s`, `probe_interval: 1s`. These values guarantee deterministic overflow loss within 90s while keeping delivery ratio >= 0.80 and preserving silent loss semantics. +- 2026-07-24 — **Claude Code CLI version pinned for the demo: `2.1.204`.** Per plan Part 7 risk #6 (beta trace/log/metric span shape can change between releases): do not upgrade the Claude Code CLI used for demo recording past this version without re-running the M1 acceptance checklist. Note: this pins the CLI version only. The signal-presence portion of the M1 acceptance checklist (`claude_code.interaction` spans, `claude_code.user_prompt` events, `claude_code.session.count` in SigNoz Traces/Logs/Metrics Explorer) requires a live SigNoz instance on `signoz-network` (external, per the 2026-07-17 SigNoz entry above) and has not been run in this environment — it remains an outstanding manual step for an operator with the demo stack up. + diff --git a/README.md b/README.md index fb6cb22..bbe65a0 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,17 @@ python -m spanledger run --config spanledger.yaml ``` Status: `curl localhost:8231/status` · Findings: `curl localhost:8231/findings` · -CLI: `python -m spanledger report` +Trust API: `curl "localhost:8231/api/v2/trust?from=...&to=..."` · +MCP Server: `python -m spanledger mcp` · CLI: `python -m spanledger report` + +## Demo Quickstart (Claude Code + MCP Trust Layer) + +To run the live demo with real Claude Code telemetry and active loss verification: + +1. Enable Claude Code telemetry export: `. demo/claude-code-env.sh` +2. Start the demo stack: `docker compose -f demo/compose.yaml up -d` +3. Run spanLedger: `python -m spanledger run --config demo/spanledger.demo.yaml` +4. Follow the step-by-step 3-minute script in [demo/walkthrough.md](demo/walkthrough.md). ## Tests @@ -28,6 +38,7 @@ ruff check . ## Layout -`spanledger/` core modules (config, probe, registry, verify, findings, signoz, report, httpapi) · +`spanledger/` core modules (config, probe, registry, verify, findings, signoz, report, httpapi, mcp) · `assets/` SigNoz dashboard + alerts · `demo/` compose stack, traffic app, chaos scripts · `spikes/` M0 spike scripts · `docs/` architecture and specs. + diff --git a/demo/README.md b/demo/README.md index 54a1834..7b13450 100644 --- a/demo/README.md +++ b/demo/README.md @@ -29,3 +29,15 @@ otel-gateway → SigNoz. spanLedger injects probes at the agent and verifies in - `reset.sh` — restore the stack between takes Every number shown in the demo must be reproducible by running these scripts. + +## Real telemetry (Claude Code) + +To send live Claude Code telemetry through the audited pipeline: + +```sh +. demo/claude-code-env.sh +``` + +This sets up OTLP export to `http://localhost:14317` (the entry agent). +Ensure privacy flags remain unset to avoid logging prompts or API bodies. + diff --git a/demo/chaos/overflow.sh b/demo/chaos/overflow.sh index 4044dcf..9fc6d19 100644 --- a/demo/chaos/overflow.sh +++ b/demo/chaos/overflow.sh @@ -1,10 +1,12 @@ #!/usr/bin/env sh -# Sustained overload via toxiproxy bandwidth limit -> queue overflow -> periodic striped drops. +# Sustained overload via toxiproxy bandwidth limit -> queue overflow -> the queue stays +# full for the duration, so drops land as one contiguous run (plan Part 2 physics). set -eu -DURATION="${1:-120}" +DURATION="${1:-90}" +RATE="${2:-8}" curl -s -X POST localhost:8474/proxies/gateway/toxics \ - -d '{"name":"slow","type":"bandwidth","attributes":{"rate":8}}' >/dev/null -echo "bandwidth throttled for ${DURATION}s" + -d "{\"name\":\"slow\",\"type\":\"bandwidth\",\"attributes\":{\"rate\":${RATE}}}" >/dev/null +echo "bandwidth throttled for ${DURATION}s at ${RATE}KB/s" sleep "$DURATION" curl -s -X DELETE localhost:8474/proxies/gateway/toxics/slow >/dev/null echo "throttle removed" diff --git a/demo/chaos/run_scenarios.py b/demo/chaos/run_scenarios.py index 247e424..5cfea4e 100644 --- a/demo/chaos/run_scenarios.py +++ b/demo/chaos/run_scenarios.py @@ -97,10 +97,11 @@ def _fault_oomcrash(): def _fault_overflow(): - # Throttle the gateway link so the agent's queue backs up and overflows -> striped drops. + # Throttle the gateway link so the agent's queue backs up and stays full for the + # duration -> one contiguous run of drops (plan Part 2 physics, frozen). r = httpx.post( f"{TOXIPROXY}/proxies/gateway/toxics", - json={"type": "bandwidth", "attributes": {"rate": 1}}, + json={"type": "bandwidth", "attributes": {"rate": 8}}, timeout=10, ) name = r.json().get("name", "bandwidth_downstream") if r.status_code < 300 else None @@ -149,7 +150,7 @@ class Scenario: "SIGTERM mid-load (opentelemetry-collector #13853)"), "oomcrash": Scenario(_fault_oomcrash, Expectation("loss", ("contiguous",)), "OOM/SIGKILL with in-memory queue"), - "overflow": Scenario(_fault_overflow, Expectation("loss", ("striped", "scattered")), + "overflow": Scenario(_fault_overflow, Expectation("loss", ("contiguous",)), "sustained overload -> queue overflow"), } diff --git a/demo/claude-code-env.sh b/demo/claude-code-env.sh new file mode 100644 index 0000000..064a988 --- /dev/null +++ b/demo/claude-code-env.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env sh +# Claude Code Telemetry Configuration for spanLedger demo. +# Note: Keep privacy flags OFF (do NOT set OTEL_LOG_USER_PROMPTS, +# OTEL_LOG_ASSISTANT_RESPONSES, OTEL_LOG_TOOL_CONTENT, or OTEL_LOG_RAW_API_BODIES). + +export CLAUDE_CODE_ENABLE_TELEMETRY=1 +export CLAUDE_CODE_ENHANCED_TELEMETRY_BETA=1 # required for traces +export OTEL_EXPORTER_OTLP_PROTOCOL=grpc +export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:14317 +export OTEL_TRACES_EXPORTER=otlp +export OTEL_LOGS_EXPORTER=otlp +export OTEL_METRICS_EXPORTER=otlp +export OTEL_METRIC_EXPORT_INTERVAL=10000 # 10s, not the 60s default +export OTEL_RESOURCE_ATTRIBUTES=service.name=claude-code-demo diff --git a/demo/otel/agent.yaml b/demo/otel/agent.yaml index 3157b9f..4afbe87 100644 --- a/demo/otel/agent.yaml +++ b/demo/otel/agent.yaml @@ -12,6 +12,14 @@ exporters: endpoint: toxiproxy:24317 # gateway via toxiproxy (fault injection point) tls: insecure: true + # Bounded queue + bounded retry ceiling: makes overflow reachable and loss + # guaranteed (not probabilistic) under the Part 2 bandwidth throttle. + sending_queue: + enabled: true + queue_size: 64 + retry_on_failure: + enabled: true + max_elapsed_time: 30s service: pipelines: @@ -23,3 +31,7 @@ service: receivers: [otlp] processors: [batch] exporters: [otlp] + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlp] diff --git a/demo/otel/gateway.yaml b/demo/otel/gateway.yaml index dd12533..03f77ed 100644 --- a/demo/otel/gateway.yaml +++ b/demo/otel/gateway.yaml @@ -23,3 +23,7 @@ service: receivers: [otlp] processors: [batch] exporters: [otlp] + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlp] diff --git a/demo/walkthrough.md b/demo/walkthrough.md index a059ce9..3fd0ad9 100644 --- a/demo/walkthrough.md +++ b/demo/walkthrough.md @@ -1,93 +1,68 @@ -# Demo walkthrough - -The operator journey Phase 2 exists to tell: an alert fires, you land on the SLO -dashboard, you drill into the finding, you pivot to the actual traces, you decide -whether to roll back, and you watch recovery close the loop. Every command below is -real — reproducible by a judge running the same steps. - -Automated version: `python demo/chaos/run_scenarios.py --walkthrough` runs this exact -gateway-kill script with narration printed at each step, timed for a live recording -so the presenter talks over it instead of typing (see the script's `run_walkthrough`). - -## Prerequisites - -- Demo stack up: `docker compose -f demo/compose.yaml up -d` (SigNoz on - `signoz-network`, `SIGNOZ_API_KEY` exported). -- Dashboards and alerts rendered and imported: - ```sh - spanledger assets render --config demo/spanledger.demo.yaml --out /tmp/rendered - python demo/validate_assets.py --rendered /tmp/rendered - ``` -- `spanledger run --config demo/spanledger.demo.yaml` running against the stack. - -## 1. Alert fires - -Kill the gateway: `docker stop spanledger-demo-otel-gateway-1` (or let -`run_scenarios.py --walkthrough` do it on a timer). Within a few minutes, -`burn-fast.json` + `burn-fast-1h.json` (both labeled `spanledger.policy: fast-burn`) -should both be firing in SigNoz's alert list — narrate that the two-window pairing -(D14) is what's supposed to co-fire before you'd page on it for real. - -## 2. Land on the SLO dashboard - -Open **Delivery SLO & Error Budget** (`slo-error-budget.json`). Point at: - -- **Error budget remaining** dropping through the same 0.25/0.10/0 ladder the - `budget_warning`/`budget_exhausted` events fire on. -- **Burn rates** climbing past the 14.4 / 6.0 threshold lines. -- **Formula SLI** (panel 3) tracking panel 2's `spanledger_sli_ratio` gauge — - "computed two different ways, same answer" is the audit-the-auditor moment - (success criterion 2). -- **Unknown/low-confidence strip** staying near zero — the loss is real loss, not - probes stuck in the maturity settling window. - -## 3. Drill into the finding - -The **Findings & budget events** table on the same dashboard shows the `loss` event. -Open it — the payload has `gap_shape: "contiguous"`, a `loss_onset` timestamp, and -(new in Phase 2, D15) a **`traces_filter`** field: - -``` -spanledger.stream = 'gateway-a' AND spanledger.seq >= AND spanledger.seq <= -``` - -## 4. Ledger-less honesty (deferred) - -spanLedger V2 tells you a stream lost data and roughly when — it does not yet tell -you *which hop*. That per-hop conservation ledger (comparing an otelcol panel's own -counters against spanLedger's probe-verified floor) is explicitly a Phase 3 feature. -Say so plainly here rather than implying more precision than the product has today. - -## 5. Traces evidence - -Copy the `traces_filter` string from step 3, paste it into SigNoz's **Traces -Explorer** query bar. You'll see the surrounding real traffic around the missing -probe sequence range — the honest floor ("this many probes, this range, this -confidence") next to the real spans that were flowing at the time. - -## 6. Rollback (if this were a real regression) - -This is also where a deploy marker earns its keep: if the loss lines up with a -recent `deploy_marker` event (visible on **Deployment Regression**'s marker-lane -table), that's your rollback signal. Post one for the walkthrough itself so the -marker lane isn't empty: +# 3-Minute Demo Walkthrough Script + +Every command and number shown below is reproducible by running the demo stack. + +## Timed Run-of-Show (3 Minutes) + +### 0:00–0:45 — The Baseline & Real Workload +1. Boot the stack with Claude Code telemetry enabled: + ```sh + . demo/claude-code-env.sh + python -m spanledger run --config demo/spanledger.demo.yaml + ``` +2. Point out live `claude_code.*` spans flowing in SigNoz alongside synthetic probes. +3. Show `GET /api/v2/trust?from=...&to=...&stream=gateway-a` returning `trustworthy: true`, + `100.0% delivery`. **Scope every trust query to `stream=gateway-a`** — the demo + config also runs `gateway-b` (plan §7, non-determinism #10); an unscoped query + aggregates both streams and the numbers won't match the single-stream narrative. + +### 0:45–1:30 — Silent Loss Injection (Overflow Scenario) +1. Trigger the overflow chaos scenario: + ```sh + sh demo/chaos/overflow.sh 90 8 + ``` +2. Claude Code keeps coding with **zero error output**. +3. SigNoz continues ingesting telemetry, but backend exporter queue fills up and drops spans silently. + +### 1:30–2:15 — The MCP Trust Layer in Action +1. Ask Claude Code (or query `/api/v2/trust?...&stream=gateway-a`): *"Why are my traces + missing?"* +2. MCP Tool `spanledger_check_telemetry_trust` is invoked **first**. +3. Response returns (the exact numbers come from this run, never state them in + advance — plan §2's honesty note: guaranteed is *that* loss occurs and *that* + spanLedger reports the exact range it observed, not a predetermined figure): + - `trustworthy: false` + - `verified_ratio` below `1.0` — read the live value off the response + - `extrapolated_user_spans_lost` on the incident — a nonzero live estimate, not a + fixed number + - `probable_cause`: `contiguous` queue overflow + - Prose recommendation warning the developer before application diagnosis. + +### 2:15–2:45 — Verification & Evidence Link +1. Navigate to the incident in spanLedger UI or click the `signoz_url` deep link. +2. Click **View Evidence**. +3. SigNoz Traces Explorer opens pre-filtered with `traces_filter` and time-bounded to `affected_window`. +4. The platform confirms the absence of probe spans in the lost sequence range. + +### 2:45–3:00 — Recovery & Wrap-up +1. Throttle lifts automatically after 90 seconds. +2. A `recovery` event registers, linked back to the loss incident ID. +3. Trust status recovers to `trustworthy: true`. + +## Pre-Recording PII Checklist + +All four must be true **before any recording** — check them for real each time, not +once: + +- [ ] Use throwaway Anthropic/Groq account +- [ ] `OTEL_LOG_USER_PROMPTS` unset +- [ ] `OTEL_LOG_ASSISTANT_RESPONSES` unset +- [ ] `OTEL_LOG_RAW_API_BODIES` unset + +Verify the three env vars with: ```sh -spanledger mark-deploy --label "walkthrough recording" --api http://localhost:8231 +env | grep -c OTEL_LOG_ ``` -## 7. Recovery - -Restart the gateway: `docker start spanledger-demo-otel-gateway-1`. Within a couple -of verification cycles: - -- A `recovery` event appears, `links` pointing at the `loss` event's id it closed. -- **Error budget remaining** stops dropping; **burn rates** fall back toward 0. -- The fast-burn alert pair clears once both windows recover and stay recovered for - `2 * verify.poll_interval` (the hysteresis re-arm window, module 5). - -## Status - -Written and internally consistent (real event fields, real metric names, the real -`traces_filter` string), but **not rehearsed end-to-end by a human** in this -implementation session — see `docs/v2/PENDING_WORK.md`. +→ must print `0`. diff --git a/docs/v2/IMPL_TRUST_LAYER.md b/docs/v2/IMPL_TRUST_LAYER.md new file mode 100644 index 0000000..80dfe2a --- /dev/null +++ b/docs/v2/IMPL_TRUST_LAYER.md @@ -0,0 +1,1152 @@ +# Implementation Strategy: Real Telemetry + MCP Trust Layer + +Status: **PLAN — no code changed.** +Date: 2026-07-24 · Target: SigNoz Hackathon, Track 2 +Prereq: research complete ([RESEARCH_PRODUCT_DIRECTION.md](RESEARCH_PRODUCT_DIRECTION.md)), +Foundry migration planned ([RFC_FOUNDRY_MIGRATION.md](RFC_FOUNDRY_MIGRATION.md)). + +Architecture is fixed. Everything below is additive or configuration. + +--- + +## The one technical fact that shapes this entire document + +**spanLedger cannot verify Claude Code's spans individually. It never could, and it +must never claim to.** + +spanLedger verifies *its own* sequence-numbered probes. Claude Code's telemetry is +**user traffic** — the thing that makes the demo real, not the thing being counted. +The honest claim is: + +> "Probes riding the same pipeline, injected at the same entry point, during the same +> window, lost 3.8%. Therefore approximately N of your Claude Code spans were lost." + +The codebase already encodes this correctly: `extrapolated_user_spans_lost` in +[events.py](../../spanledger/events.py:132) and `confidence: "probe_verified"` in the +loss payload. **That field is the bridge between the probe model and the user-facing +claim, and it is the most important thing in the demo.** Every piece of copy — MCP +response, UI, narration — must preserve the distinction between *verified* (probes) +and *extrapolated* (user spans). Blur it once and the never-over-report invariant +becomes marketing. + +--- + +## Part 1 — Real Telemetry Integration + +### Routing strategy: Claude Code becomes a second traffic source at the existing entry + +No parallel pipeline. Claude Code exports to the **same OTLP entry point the probes +use** — `otel-agent:4317`, published on the host as `:14317` +([demo/compose.yaml](../../demo/compose.yaml)). Probes and Claude Code telemetry then +travel byte-for-byte the same path: agent → toxiproxy → gateway → SigNoz. + +``` +Claude Code (host) ─┐ + ├─▶ otel-agent :14317 ─▶ toxiproxy :24317 ─▶ otel-gateway ─▶ SigNoz +spanLedger probes ──┘ (fault injection point) ▲ + │ +spanLedger verify (out-of-band, direct to SigNoz Query API) ──────────────────────┘ +``` + +**Why this is correct and not merely convenient:** the audit claim is only valid if +probes share the failure domain with the traffic they speak for. Same receiver, same +batch processor, same sending queue, same exporter. If Claude Code had its own +collector, the probes would be auditing a *different* queue and the extrapolation would +be a lie. **Co-location at the entry point is a correctness requirement.** + +### Environment variables + +Set in the shell that launches the demo Claude Code session: + +```bash +export CLAUDE_CODE_ENABLE_TELEMETRY=1 +export CLAUDE_CODE_ENHANCED_TELEMETRY_BETA=1 # required for traces +export OTEL_EXPORTER_OTLP_PROTOCOL=grpc +export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:14317 +export OTEL_TRACES_EXPORTER=otlp +export OTEL_LOGS_EXPORTER=otlp +export OTEL_METRICS_EXPORTER=otlp +export OTEL_METRIC_EXPORT_INTERVAL=10000 # 10s, not the 60s default +export OTEL_RESOURCE_ATTRIBUTES=service.name=claude-code-demo +``` + +**Two non-obvious settings, both load-bearing:** + +- **`OTEL_METRIC_EXPORT_INTERVAL=10000`.** The default is 60s. In a 3-minute demo + that's three data points — the metrics signal would appear dead. 10s makes the + metrics leg visible. Documented as a debugging value; restore defaults outside demos. +- **`CLAUDE_CODE_ENHANCED_TELEMETRY_BETA=1`.** Without it there are no spans, and the + traces leg — the one the demo is built on — carries no user traffic at all. + +**Privacy: leave these OFF.** `OTEL_LOG_USER_PROMPTS`, `OTEL_LOG_ASSISTANT_RESPONSES`, +`OTEL_LOG_TOOL_CONTENT`, `OTEL_LOG_RAW_API_BODIES` are off by default — keep them off. +Standard attributes still include `user.email` and `organization.id`, so **use a +throwaway account for any recording.** Add this as a checklist item in +[demo/walkthrough.md](../../demo/walkthrough.md), not as tribal knowledge. + +### Collector configuration + +[demo/otel/agent.yaml](../../demo/otel/agent.yaml) already has `traces` and `logs` +pipelines. Two edits, both **frozen below — copy verbatim**. + +**Edit 1 — `demo/otel/agent.yaml`.** Replace the whole `exporters:` block and append a +`metrics` pipeline. Final file content: + +```yaml +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + +processors: + batch: {} + +exporters: + otlp: + endpoint: toxiproxy:24317 # gateway via toxiproxy (fault injection point) + tls: + insecure: true + # Bounded queue + bounded retry ceiling: makes overflow reachable and loss + # guaranteed (not probabilistic) under the Part 2 bandwidth throttle. + sending_queue: + enabled: true + queue_size: 64 + retry_on_failure: + enabled: true + max_elapsed_time: 30s + +service: + pipelines: + traces: + receivers: [otlp] + processors: [batch] + exporters: [otlp] + logs: + receivers: [otlp] + processors: [batch] + exporters: [otlp] + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlp] +``` + +**Edit 2 — `demo/otel/gateway.yaml`.** Append the `metrics` pipeline only. **Do not add +`sending_queue`/`retry_on_failure` here** — the throttle sits between agent and gateway, +so the agent's queue is the one that must overflow. Add exactly: + +```yaml + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlp] +``` + +**Do not modify** `demo/otel/gateway-logs-down.yaml` (Phase 3 fault config, out of scope +for this plan). + +No new services, no new ports, no routing rules. Probes and user traffic are +distinguished downstream by the `spanledger.*` resource attributes +([probe.py](../../spanledger/probe.py:29)) — the collector treats both as ordinary OTLP. + +### Execution order for Part 1 + +1. Edit `demo/otel/agent.yaml` (both blocks above). +2. Edit `demo/otel/gateway.yaml` (metrics pipeline only). +3. Create `demo/claude-code-env.sh` with the exact variable block above, prefixed with + `#!/usr/bin/env sh` and a comment naming the privacy flags that must stay unset. +4. Restart the demo stack; confirm the collectors accept the config (no restart loop). +5. Verify signals in SigNoz (M1 acceptance checklist). +6. Update `demo/README.md` — add a "Real telemetry (Claude Code)" section pointing at + `demo/claude-code-env.sh`. + +### Expected telemetry + +| Signal | Claude Code emits | Interval | Role in demo | +|---|---|---|---| +| **Traces** | `claude_code.interaction` → `llm_request` / `tool` / `hook` | 5s | **Primary.** The spans a developer would actually debug. | +| **Logs/events** | `user_prompt`, `tool_result`, `api_request`, `api_error` | 5s | Secondary. Proves multi-signal; exercises the unverified logs leg (spike S6). | +| **Metrics** | `session.count`, `token.usage`, `cost.usage`, `lines_of_code.count` | 10s (set) | Tertiary. Nice-to-have; do not build the demo on it. | + +**Architectural impact: none.** No changes to `spanledger/`. This is environment +variables plus one collector pipeline. Keep [demo/agent-app](../../demo/agent-app) — +Claude Code gives authenticity, the synthetic generator gives volume and determinism. +A benchmark needs both. + +--- + +## Part 2 — Failure Injection + +### The recommendation: **queue overflow via sustained bandwidth throttle** — `demo/chaos/overflow.sh`, already written. + +### Why this one, and why the others are wrong for this demo + +The brief's own constraints eliminate every alternative: + +| Requirement | overflow | outage.sh | sigterm.sh | oomcrash.sh | +|---|---|---|---|---| +| Nothing crashes | ✅ | ✅ | ❌ collector dies | ❌ collector dies | +| Claude keeps coding | ✅ | ✅ | ✅ | ✅ | +| **SigNoz still receives telemetry** | ✅ | ❌ **all stops** | partial | partial | +| Only *part* silently lost | ✅ | ❌ total | ✅ | ✅ | +| Deterministic | ✅ | ✅ | ⚠️ timing-sensitive | ⚠️ memory-sensitive | + +**Only queue overflow satisfies all five.** A backend outage makes the dashboard go +flat — everyone sees it, and *you don't need spanLedger to notice*. It proves the least +while looking the most dramatic. Overflow is the opposite: the graph dips slightly and +looks like less traffic. That gap between "looks fine" and "is broken" **is the +product**, and it is the most common silent failure in production — the entire +`otelcol_exporter_enqueue_failed_spans` monitoring practice exists because of it. + +### How to inject it + +`overflow.sh` posts a toxiproxy `bandwidth` toxic (`rate: 8` KB/s) on the `gateway` +proxy for N seconds. Upstream of the throttle, the agent's sending queue fills; once +full, new spans are dropped with no error surfaced to the emitter. + +**Tuning is already frozen in Part 1** (`queue_size: 64`, `max_elapsed_time: 30s`). +This makes overflow reachable in ~30–45s and — critically — makes the drop +**deterministic in effect**: with a bounded queue and a bounded retry ceiling, loss is +guaranteed, not probabilistic. + +**Frozen scenario parameters.** `demo/chaos/overflow.sh` currently hard-codes +`rate: 8` and takes duration as `$1` defaulting to `120`. Change **only** the default +duration and add a rate parameter; do not alter the toxic type or the proxy name: + +| Parameter | Frozen value | Where | +|---|---|---| +| Toxic type | `bandwidth` | `overflow.sh` (unchanged) | +| Proxy name | `gateway` | `demo/toxiproxy.json` (unchanged) | +| `rate` | `8` (KB/s) — `$2`, default `8` | `overflow.sh` | +| Duration | `90` (seconds) — `$1`, default `90` | `overflow.sh` (was `120`) | +| Agent `queue_size` | `64` | `demo/otel/agent.yaml` | +| Agent `max_elapsed_time` | `30s` | `demo/otel/agent.yaml` | +| Probe interval | `1s` (traces) | `demo/spanledger.demo.yaml` (unchanged) | + +**These seven values are the demo's physics. Once M2 acceptance passes, they are frozen +— do not tune them again for any later milestone.** Record the pinned set in +`DECISIONS.md` at the end of M2. + +**Honesty note to state on stage:** the *exact* sequence numbers lost are not +predetermined. Guaranteed is *that* loss occurs and *that* spanLedger reports the exact +gap it observed. Claiming per-seq determinism would be a lie the invariant forbids. + +### What each party sees + +**spanLedger detects:** delivery ratio drops below target → `loss` event with +contiguous `gap_runs` (`seq_from`/`seq_to` + timestamps), `gap_shape`, `confidence: +probe_verified`, `extrapolated_user_spans_lost`, and `traces_filter`; burn rate crosses +`FAST_BURN` (14.4); error budget drains; on removal, a `recovery` event with +`links: [loss_id]`. + +**SigNoz displays:** `claude_code.*` spans still arriving (nothing looks broken); a +modest dip in span rate; the spanLedger dashboards showing SLI below target and budget +draining; the fast-burn alert pair firing. The Traces Explorer, filtered by +`traces_filter`, returns **nothing** where probes should be — *the platform confirms the +absence*. + +**The audience observes:** an app that works, an agent that keeps coding, a backend +still receiving data, dashboards that look basically fine — and one system saying *"3.8% +of the last four minutes never arrived, here is the exact range."* + +--- + +## Part 3 — MCP Trust Layer + +### Minimum viable surface: **one tool.** + +`spanledger_check_telemetry_trust(from, to, stream?, signal?)` + +The research doc proposed six. For the hackathon, **five are deferred.** Justification: + +- `pipeline_status`, `telemetry_slo`, `active_incidents`, `incident_history` are all + *informational* — they tell an agent things. Only the trust check **changes what the + agent does**. A tool that alters behavior is a product; tools that return data are an + API with extra steps. +- `diagnose_pipeline`'s value (probable cause, evidence, deep link) is **collapsed into + the single tool's response**. Splitting it forces a second round-trip and creates a + path where an agent diagnoses without first checking trust — exactly the failure the + product exists to prevent. +- One tool is impossible to misuse and trivial to explain. Six invites the judge + question *"which one does Claude actually call?"* — a question with no good answer + under demo pressure. + +**One tool, one behavior change, one sentence to explain it.** That is the strongest +possible version of this idea. + +### Where the answer is computed — frozen + +The plan's own rule: *"If a field isn't available, that's a signal to extend the API +deliberately, not to let the MCP layer derive its own answers."* The trust verdict is +such a field. Therefore: + +- **`GET /api/v2/trust` is added to [httpapi.py](../../spanledger/httpapi.py).** It owns + the entire derivation. +- **`spanledger/mcp.py` is a pure passthrough.** It validates input, calls + `/api/v2/trust`, and returns the body unchanged. **It contains no arithmetic.** If an + implementer finds themselves computing a ratio in `mcp.py`, the code is in the wrong + file. + +This keeps one source of truth, makes the logic unit-testable without an MCP client, and +means the CLI/frontend can consume the identical verdict later at zero cost. + +### Request schema — FROZEN + +Both surfaces take the same four parameters. MCP tool input schema (JSON Schema draft +2020-12) and the `/api/v2/trust` query string are identical in name, type, and rules. + +| Field | Type | Required | Default | Validation | On violation | +|---|---|---|---|---|---| +| `from` | string, RFC3339 | **yes** | — | Parseable by `httpapi._parse_at`; must be `< to` | 400 `invalid from` | +| `to` | string, RFC3339 | **yes** | — | Parseable by `httpapi._parse_at`; must be `> from` | 400 `invalid to` | +| `stream` | string | no | `null` (= all streams) | Must be in `slo_engine.stream_names()` | 400 `unknown stream` | +| `signal` | enum `traces`\|`logs`\|`metrics` | no | `"traces"` | `httpapi._signal_param` | 400 `invalid signal` | + +Additional frozen rules: + +- **Unknown query parameters are a 400** — reuse `httpapi._require_params` with the + allowed set `{"from", "to", "stream", "signal"}`. This matches every other `/api/v2` + route; do not relax it. +- **Window length is capped at 24h.** `to - from > 86400s` → 400 `window too large`. + Prevents an agent from requesting a 90-day scan and blocking the handler. +- **`from`/`to` are RFC3339 strings here, not the nanosecond ints** used by + `/api/v2/events`. This is deliberate: the caller is a language model, and RFC3339 is + what it produces reliably. Convert once via `_parse_at` at the handler boundary. + +Enums stay enums, timestamps are validated, and validation **reuses these exact existing +helpers in [httpapi.py](../../spanledger/httpapi.py)** rather than reimplementing them: +`_require_params`, `_single`, `_signal_param`, `_parse_at`, `_rfc3339_ns`, and the +`HttpError` class for all error paths. One source of truth applies to validation too. + +### Response schema + +```jsonc +{ + "trustworthy": false, + "confidence": "probe_verified", + "verified_ratio": 0.962, + "unknown_ratio": 0.004, + "window": { "from": "...", "to": "..." }, + "affected_window": { "from": "2026-07-24T14:32:10Z", "to": "2026-07-24T14:36:02Z" }, + "incidents": [{ + "id": "0191...", "class": "loss", "stream": "gateway-a", "signal": "traces", + "gap_runs": [{ "seq_from": 40, "seq_to": 59, "t_from": "...", "t_to": "..." }], + "gap_shape": "contiguous", + "probes_lost": 20, + "extrapolated_user_spans_lost": 1840, + "probable_cause": { + "hypothesis": "exporter queue overflow or backend outage", + "evidence": ["gap_shape: contiguous", + "otelcol_exporter_enqueue_failed_spans +1840 at exporter"], + "confidence": "medium" + } + }], + "evidence": { + "traces_filter": "spanledger.stream = 'gateway-a' AND spanledger.seq >= 40 AND spanledger.seq <= 59", + "signoz_url": "http://localhost:8080/traces-explorer?filter=...", + "spanledger_url": "http://localhost:5173/incident/0191..." + }, + "recommendation": "Telemetry for this window is incomplete. ~1840 user spans were probably lost between 14:32:10 and 14:36:02. Conclusions drawn from this window may be unreliable. Verify against the evidence link before diagnosing application behavior." +} +``` + +**Design decisions worth defending:** + +- **`verified_ratio` and `extrapolated_user_spans_lost` are separate fields with + different names.** One is counted, one is estimated. Never merge them. +- **`unknown_ratio` is always present**, including when trustworthy. Publishing + uncertainty is the differentiator; hiding it when convenient forfeits it. +- **`probable_cause.confidence` is explicit and may be `"low"`.** The tool returns + hypotheses with evidence, never an asserted root cause. An auditor that guesses has + stopped being an auditor. +- **`recommendation` is prose.** The consumer is a language model; a natural-language + instruction is far more reliably honored than a boolean it must interpret. +- **`trustworthy: true` still returns evidence links.** Trust should be checkable. + +### Response field types — FROZEN + +Every field is **always present**. Absent data is `null` or `[]`, never an omitted key — +a model reasons far more reliably over a stable shape. + +| Field | Type | Nullable | Notes | +|---|---|---|---| +| `trustworthy` | bool | no | Never `null`. See verdict table below. | +| `confidence` | enum `probe_verified`\|`unknown`\|`no_data` | no | Verification coverage, not loss severity. | +| `verified_ratio` | float, 4dp | no | `1.0` when nothing was judged. | +| `unknown_ratio` | float, 4dp | no | `0.0` when nothing was judged. | +| `window` | `{from, to}` RFC3339 strings | no | Echoes the request, normalised. | +| `affected_window` | `{from, to}` RFC3339 strings | **yes** | `null` when `incidents` is empty. | +| `incidents` | array of incident objects | no | `[]` when none. Max 50. | +| `evidence` | object (3 keys, below) | no | Always present. | +| `recommendation` | string | no | Always non-empty. Two frozen templates below. | + +Incident object (one per `loss` event, ordered oldest → newest by `emitted_at_ns`): + +| Field | Type | Source (loss payload key) | +|---|---|---| +| `id` | string | `id` | +| `class` | string, always `"loss"` | `class` | +| `stream` | string | `stream` | +| `signal` | string | `signal` | +| `gap_runs` | array of `{seq_from:int, seq_to:int, t_from:str, t_to:str}` | `gap_runs` verbatim | +| `gap_shape` | enum `contiguous`\|`striped`\|`scattered` | `gap_shape` | +| `probes_lost` | int | `probes.missing` | +| `extrapolated_user_spans_lost` | int or `null` | `extrapolated_user_spans_lost.estimate`, else `null` | +| `probable_cause` | object, below | derived — see rules | + +`evidence` object: + +| Field | Type | Nullable | Source | +|---|---|---|---| +| `traces_filter` | string | yes | `traces_filter` of the **last** incident; `null` if no incidents | +| `signoz_url` | string | yes | Built from `traces_filter`; `null` if `traces_filter` is `null` | +| `spanledger_url` | string | yes | `null` if no incidents | + +### Derivation rules — FROZEN + +Implement exactly this in the `/api/v2/trust` handler. No step is optional. + +1. Parse and validate per the request table. Convert `from`/`to` to ns via `_parse_at`. +2. `streams = [stream] if stream else slo_engine.stream_names()`. +3. `rows = store.query_events(EventFilter(stream=None, from_ns=from_ns, to_ns=to_ns, + limit=500))` — reuse `store.query_events` and `store.EventFilter` unchanged. +4. `loss = [r for r in rows if r.class_ == CLASS_LOSS and r.signal == signal and + r.stream in streams]`, preserving query order (already `emitted_at_ns ASC`). + `unreachable = [r for r in rows if r.class_ == CLASS_BACKEND_UNREACHABLE and + r.stream in streams]`. Import both constants from `events.py` — do not inline strings. +5. Sum across `loss` payloads: `sent`, `verified`, `missing`, `unknown` from + `r.payload["probes"]`. +6. `judged = verified + missing`; + `verified_ratio = round(verified / judged, 4) if judged else 1.0`; + `unknown_ratio = round(unknown / sent, 4) if sent else 0.0`. +7. **Verdict table — exhaustive, no other combination exists:** + + | Condition (evaluated top-down, first match wins) | `trustworthy` | `confidence` | + |---|---|---| + | `loss` is non-empty | `false` | `probe_verified` | + | `unreachable` is non-empty | `false` | `unknown` | + | every stream's `slo_engine.snapshot(s, signal)["sli"]` is `None` | `false` | `no_data` | + | otherwise | `true` | `probe_verified` | + +8. `affected_window` = `{from: earliest gap_run.t_from, to: latest gap_run.t_to}` across + all `loss` incidents; `null` if `loss` is empty. +9. `probable_cause` per incident, from `gap_shape` only (frozen mapping): + + | `gap_shape` | `hypothesis` | base `confidence` | + |---|---|---| + | `contiguous` | `"exporter queue overflow or backend outage"` | `medium` | + | `striped` | `"sustained overload causing periodic queue overflow"` | `medium` | + | `scattered` | `"intermittent loss, cause unclear"` | `low` | + + `evidence` array is built by appending, in this order, only the entries that apply: + `"gap_shape: "`; `"deploy marker ' - Open in SigNoz Traces Explorer + View Evidence )} diff --git a/frontend/src/test/signoz-links.test.ts b/frontend/src/test/signoz-links.test.ts new file mode 100644 index 0000000..9882eb6 --- /dev/null +++ b/frontend/src/test/signoz-links.test.ts @@ -0,0 +1,44 @@ +import { describe, it, expect } from 'vitest' +import { + tracesExplorerUrl, + dashboardsUrl, + alertsUrl, + affectedWindowFromGapRuns, +} from '@/lib/signoz-links' + +describe('signoz-links', () => { + it('builds tracesExplorerUrl without time bounds', () => { + const url = tracesExplorerUrl("spanledger.stream = 'gateway-a'") + expect(url).toContain('/traces-explorer?filter=spanledger.stream+%3D+%27gateway-a%27') + }) + + it('builds tracesExplorerUrl with time bounds', () => { + const url = tracesExplorerUrl("spanledger.stream = 'gateway-a'", { + from: '2026-07-24T10:00:00Z', + to: '2026-07-24T11:00:00Z', + }) + expect(url).toContain('/traces-explorer?') + expect(url).toContain('filter=spanledger.stream+%3D+%27gateway-a%27') + expect(url).toContain('from=2026-07-24T10%3A00%3A00Z') + expect(url).toContain('to=2026-07-24T11%3A00%3A00Z') + }) + + it('builds dashboardsUrl and alertsUrl', () => { + expect(dashboardsUrl()).toContain('/dashboards') + expect(alertsUrl()).toContain('/alerts') + }) + + it('computes the affected window from gap runs (earliest t_from, latest t_to)', () => { + // Mirrors the backend's /api/v2/trust `affected_window` derivation (earliest + // gap_run.t_from, latest gap_run.t_to) applied to one incident's own gap_runs. + const bounds = affectedWindowFromGapRuns([ + { t_from: '2026-07-24T10:10:00Z', t_to: '2026-07-24T10:11:00Z' }, + { t_from: '2026-07-24T10:14:00Z', t_to: '2026-07-24T10:15:00Z' }, + ]) + expect(bounds).toEqual({ from: '2026-07-24T10:10:00Z', to: '2026-07-24T10:15:00Z' }) + }) + + it('returns undefined affected window when there are no gap runs', () => { + expect(affectedWindowFromGapRuns([])).toBeUndefined() + }) +}) diff --git a/spanledger/cli.py b/spanledger/cli.py index 7bd88c4..d2d274d 100644 --- a/spanledger/cli.py +++ b/spanledger/cli.py @@ -41,6 +41,15 @@ def report(api: str): click.echo(json.dumps({"status": status, "findings": findings}, indent=2)) +@main.command() +@click.option("--api", default="http://localhost:8231", show_default=True) +def mcp(api: str): + """Run the MCP stdio server.""" + from .mcp import serve_stdio + + serve_stdio(api) + + @main.command("mark-deploy") @click.option("--label", required=True, help="e.g. 'gateway v1.4'") @click.option("--stream", default=None, help="Scope the marker to one stream (default: global)") diff --git a/spanledger/config.py b/spanledger/config.py index ce3a48a..f66de07 100644 --- a/spanledger/config.py +++ b/spanledger/config.py @@ -56,6 +56,7 @@ class AlertingConfig: @dataclass class HTTPConfig: listen: str = ":8231" + ui_base_url: str = "http://localhost:5173" @dataclass @@ -259,7 +260,10 @@ def load(path: str) -> Config: lookback=_duration(v.get("lookback"), 600.0), ), alerting=AlertingConfig(min_delivery_ratio=float(a.get("min_delivery_ratio", 0.999))), - http=HTTPConfig(listen=str(h.get("listen", ":8231"))), + http=HTTPConfig( + listen=str(h.get("listen", ":8231")), + ui_base_url=str(h.get("ui_base_url", "http://localhost:5173")), + ), correlation=CorrelationConfig(enabled=bool(c.get("enabled", False))), slo=_load_slo(raw.get("slo") or {}), storage=_load_storage(raw.get("storage") or {}), diff --git a/spanledger/httpapi.py b/spanledger/httpapi.py index e9f5080..36c3ffe 100644 --- a/spanledger/httpapi.py +++ b/spanledger/httpapi.py @@ -16,7 +16,12 @@ from uuid6 import uuid7 -from .events import V1_FINDING_CLASSES +from .events import ( + CLASS_BACKEND_UNREACHABLE, + CLASS_DEPLOY_MARKER, + CLASS_LOSS, + V1_FINDING_CLASSES, +) from .store import SIGNAL_TRACES, DeployMarker, EventFilter _VALID_SIGNALS = ("traces", "logs", "metrics") @@ -160,6 +165,8 @@ def _event_row_json(row) -> dict: _EVENTS_PARAMS = {"stream", "class", "severity", "from", "to", "limit", "cursor"} _HISTORY_PARAMS = {"resolution", "signal"} _SIGNAL_PARAMS = {"signal"} +_TRUST_PARAMS = {"from", "to", "stream", "signal"} +MAX_TRUST_WINDOW_S = 86400 def _signal_param(query: dict[str, list[str]]) -> str: @@ -233,6 +240,239 @@ def _v2_slo_history(slo_engine, stream: str, query: dict[str, list[str]]) -> lis raise HttpError(400, "invalid resolution", str(e)) from e +def _evidence_links( + traces_filter: str | None, + incident_id: str | None, + signoz_url_base: str = "", + ui_base_url: str = "http://localhost:5173", +) -> dict: + from urllib.parse import quote_plus + + signoz_url = None + if traces_filter and signoz_url_base: + signoz_url = f"{signoz_url_base.rstrip('/')}/traces-explorer?filter={quote_plus(traces_filter)}" + spanledger_url = None + if incident_id: + spanledger_url = f"{ui_base_url.rstrip('/')}/incident/{incident_id}" + return { + "traces_filter": traces_filter, + "signoz_url": signoz_url, + "spanledger_url": spanledger_url, + } + + +def _v2_trust( + store, + slo_engine, + query: dict[str, list[str]], + signoz_url_base: str = "", + ui_base_url: str = "http://localhost:5173", +) -> dict: + _require_params(query, _TRUST_PARAMS) + + from_str = _single(query, "from") + to_str = _single(query, "to") + if not from_str or not to_str: + raise HttpError(400, "from and to are required") + + from_ns = _parse_at(from_str, "from") + to_ns = _parse_at(to_str, "to") + if from_ns >= to_ns: + raise HttpError(400, "invalid window") + if (to_ns - from_ns) / 1e9 > MAX_TRUST_WINDOW_S: + raise HttpError(400, "window too large") + + signal = _signal_param(query) + stream = _single(query, "stream") + if stream is not None and stream not in slo_engine.stream_names(): + raise HttpError(400, "unknown stream", stream) + + streams = [stream] if stream else slo_engine.stream_names() + + filt = EventFilter(stream=None, from_ns=from_ns, to_ns=to_ns, limit=500) + rows = store.query_events(filt) + + loss_rows = [ + r for r in rows + if r.class_ == CLASS_LOSS and r.signal == signal and r.stream in streams + ] + unreachable_rows = [ + r for r in rows + if r.class_ == CLASS_BACKEND_UNREACHABLE and r.stream in streams + ] + + sent_sum = 0 + verified_sum = 0 + missing_sum = 0 + unknown_sum = 0 + + for r in loss_rows: + probes = r.payload.get("probes", {}) if isinstance(r.payload, dict) else {} + sent_sum += probes.get("sent", 0) + verified_sum += probes.get("verified", 0) + missing_sum += probes.get("missing", 0) + unknown_sum += probes.get("unknown", 0) + + judged = verified_sum + missing_sum + verified_ratio = round(verified_sum / judged, 4) if judged else 1.0 + unknown_ratio = round(unknown_sum / sent_sum, 4) if sent_sum else 0.0 + + if loss_rows: + trustworthy = False + confidence = "probe_verified" + elif unreachable_rows: + trustworthy = False + confidence = "unknown" + else: + all_none = True + for s in streams: + snap = slo_engine.snapshot(s, signal) + if snap and snap.get("sli") is not None: + all_none = False + break + if all_none: + trustworthy = False + confidence = "no_data" + else: + trustworthy = True + confidence = "probe_verified" + + all_gap_runs = [] + for r in loss_rows: + if isinstance(r.payload, dict): + runs = r.payload.get("gap_runs", []) + if isinstance(runs, list): + all_gap_runs.extend(runs) + + if all_gap_runs: + earliest_from = min(run["t_from"] for run in all_gap_runs if "t_from" in run) + latest_to = max(run["t_to"] for run in all_gap_runs if "t_to" in run) + affected_window = {"from": earliest_from, "to": latest_to} + else: + affected_window = None + + deploy_markers_in_window = [] + if affected_window: + aff_from_ns = _parse_at(affected_window["from"]) + aff_to_ns = _parse_at(affected_window["to"]) + for r in rows: + if r.class_ == CLASS_DEPLOY_MARKER: + if aff_from_ns <= r.window_from_ns <= aff_to_ns: + label = r.payload.get("label") if isinstance(r.payload, dict) else None + if label: + deploy_markers_in_window.append((label, _rfc3339_ns(r.window_from_ns))) + + incidents = [] + total_extrapolated_user_spans = 0 + has_extrapolated_clause = False + + for r in loss_rows: + payload = r.payload if isinstance(r.payload, dict) else {} + probes = payload.get("probes", {}) if isinstance(payload.get("probes"), dict) else {} + gap_shape = payload.get("gap_shape", "contiguous") + + if gap_shape == "contiguous": + hypothesis = "exporter queue overflow or backend outage" + base_conf = "medium" + elif gap_shape == "striped": + hypothesis = "sustained overload causing periodic queue overflow" + base_conf = "medium" + else: + hypothesis = "intermittent loss, cause unclear" + base_conf = "low" + + evidence_list = [f"gap_shape: {gap_shape}"] + for dm_label, dm_time in deploy_markers_in_window: + evidence_list.append(f"deploy marker '{dm_label}' at {dm_time}") + + corr = payload.get("correlation_hint") + if isinstance(corr, dict) and corr.get("metric"): + metric = corr.get("metric") + delta = corr.get("delta") + hop = corr.get("hop") + evidence_list.append(f"{metric} +{delta} at {hop}") + + if len(evidence_list) == 1: + base_conf = "low" + + extra_obj = payload.get("extrapolated_user_spans_lost") + extra_est = None + if isinstance(extra_obj, dict): + extra_est = extra_obj.get("estimate") + if extra_est is not None: + total_extrapolated_user_spans += extra_est + has_extrapolated_clause = True + + incidents.append({ + "id": r.id, + "class": CLASS_LOSS, + "stream": r.stream, + "signal": r.signal, + "gap_runs": payload.get("gap_runs", []), + "gap_shape": gap_shape, + "probes_lost": probes.get("missing", 0), + "extrapolated_user_spans_lost": extra_est, + "probable_cause": { + "hypothesis": hypothesis, + "evidence": evidence_list, + "confidence": base_conf, + }, + }) + + # Frozen response field table: "incidents | ... | Max 50." Trim from the oldest + # end so the most recent incidents (and the "last incident" used for evidence) + # are always kept. + incidents = incidents[-50:] + + if not trustworthy: + if confidence in ("unknown", "no_data"): + recommendation = ( + "Verification was degraded for this window — spanLedger could not confirm delivery. " + "Telemetry for this window is incomplete. " + "Conclusions drawn from this window may be unreliable. Verify against the evidence link " + "before diagnosing application behavior." + ) + else: + spans_clause = ( + f" (~{total_extrapolated_user_spans} user spans estimated lost)" + if has_extrapolated_clause + else "" + ) + aff_from_str = affected_window["from"] if affected_window else _rfc3339_ns(from_ns) + aff_to_str = affected_window["to"] if affected_window else _rfc3339_ns(to_ns) + recommendation = ( + f"Telemetry for this window is incomplete. {missing_sum} probes were verified lost{spans_clause} " + f"between {aff_from_str} and {aff_to_str}. Conclusions drawn from this window may be unreliable. " + "Verify against the evidence link before diagnosing application behavior." + ) + else: + recommendation = ( + f"Telemetry for this window verified complete ({verified_ratio:.1%} delivery, " + f"{unknown_ratio:.1%} unknown). No loss incidents detected." + ) + + last_incident = incidents[-1] if incidents else None + traces_filter = None + if last_incident and loss_rows: + traces_filter = loss_rows[-1].payload.get("traces_filter") if isinstance(loss_rows[-1].payload, dict) else None + incident_id = last_incident["id"] if last_incident else None + + evidence = _evidence_links(traces_filter, incident_id, signoz_url_base=signoz_url_base, ui_base_url=ui_base_url) + + return { + "trustworthy": trustworthy, + "confidence": confidence, + "verified_ratio": verified_ratio, + "unknown_ratio": unknown_ratio, + "window": {"from": _rfc3339_ns(from_ns), "to": _rfc3339_ns(to_ns)}, + "affected_window": affected_window, + "incidents": incidents, + "evidence": evidence, + "recommendation": recommendation, + } + + + MAX_LABEL_LEN = 128 _DEPLOY_MARKER_BODY_KEYS = {"at", "scope", "stream", "label", "config_hash"} @@ -257,13 +497,13 @@ def _rfc3339_ns(ns: int) -> str: ) -def _parse_at(raw: str | None) -> int: +def _parse_at(raw: str | None, name: str = "at") -> int: if raw is None: return time.time_ns() try: dt = datetime.datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError as e: - raise HttpError(400, "invalid at", raw) from e + raise HttpError(400, f"invalid {name}", raw) from e if dt.tzinfo is None: dt = dt.replace(tzinfo=datetime.timezone.utc) return int(dt.timestamp() * 1e9) @@ -318,7 +558,7 @@ def _v2_create_deploy_marker(store, event_engine, slo_engine, body, idem_key) -> def serve( - state: State, listen: str, store=None, slo_engine=None, event_engine=None + state: State, listen: str, store=None, slo_engine=None, event_engine=None, cfg=None ) -> ThreadingHTTPServer: host, _, port = listen.rpartition(":") addr = (host or "0.0.0.0", int(port)) @@ -395,6 +635,10 @@ def _route(self, path: str, query: dict[str, list[str]]): self._send(200, _v2_slo(slo_engine, _signal_param(query))) elif (m := _SLO_HISTORY_RE.match(path)) is not None: self._send(200, _v2_slo_history(slo_engine, m.group(1), query)) + elif path == "/api/v2/trust": + sz_base = cfg.signoz.query_url if (cfg and hasattr(cfg, "signoz")) else "" + ui_base = cfg.http.ui_base_url if (cfg and hasattr(cfg, "http")) else "http://localhost:5173" + self._send(200, _v2_trust(store, slo_engine, query, signoz_url_base=sz_base, ui_base_url=ui_base)) else: raise HttpError(404, "not found", path) diff --git a/spanledger/mcp.py b/spanledger/mcp.py new file mode 100644 index 0000000..02d4754 --- /dev/null +++ b/spanledger/mcp.py @@ -0,0 +1,166 @@ +"""MCP stdio server for spanLedger (plan Part 3). + +Exposes a single read-only MCP tool: `spanledger_check_telemetry_trust`. +Communicates with the spanLedger daemon exclusively over `/api/v2/trust`. +Contains no business logic or state derivation -- it is a pure passthrough. +""" + +from __future__ import annotations + +import json +import sys + +import httpx + +TOOL_NAME = "spanledger_check_telemetry_trust" +TOOL_DESCRIPTION = ( + "Call this BEFORE analyzing any telemetry data to determine whether that " + "telemetry is complete enough to reason about. Conclusions drawn from unverified " + "telemetry may be wrong." +) + +INPUT_SCHEMA = { + "type": "object", + "properties": { + "from": { + "type": "string", + "description": "Start timestamp in RFC3339 format (e.g. 2026-07-24T14:00:00Z)", + }, + "to": { + "type": "string", + "description": "End timestamp in RFC3339 format (e.g. 2026-07-24T14:30:00Z)", + }, + "stream": { + "type": ["string", "null"], + "description": "Optional stream name (e.g. gateway-a). Null checks all streams.", + "default": None, + }, + "signal": { + "type": ["string", "null"], + "description": "Signal type: traces, logs, or metrics (default: traces).", + "enum": ["traces", "logs", "metrics", None], + "default": "traces", + }, + }, + "required": ["from", "to"], +} + + +def call_trust(api_base: str, args: dict) -> dict: + """Call /api/v2/trust on the spanLedger daemon and return the body dict. + + Raises RuntimeError if daemon is unreachable (HTTPError), or ValueError if + daemon returns non-200. + """ + url = f"{api_base.rstrip('/')}/api/v2/trust" + params = {} + for key in ("from", "to", "stream", "signal"): + val = args.get(key) + if val is not None: + params[key] = str(val) + + try: + with httpx.Client(timeout=5.0) as client: + resp = client.get(url, params=params) + except httpx.HTTPError as e: + raise RuntimeError(f"spanledger daemon unreachable at {api_base}: {e}") from e + + if resp.status_code != 200: + try: + body = resp.json() + title = body.get("title", f"HTTP {resp.status_code}") + detail = body.get("detail") + msg = f"{title}: {detail}" if detail else title + except Exception: + msg = f"HTTP {resp.status_code}: {resp.text}" + raise ValueError(msg) + + return resp.json() + + +def serve_stdio(api_base: str = "http://localhost:8231") -> None: + """Run the stdio MCP JSON-RPC 2.0 server loop.""" + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + except json.JSONDecodeError: + continue + + method = req.get("method") + req_id = req.get("id") + + if not method or req_id is None: + # Notification or invalid request + continue + + if method == "initialize": + res = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "spanledger-mcp", "version": "1.0.0"}, + }, + } + elif method == "tools/list": + res = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "tools": [ + { + "name": TOOL_NAME, + "description": TOOL_DESCRIPTION, + "inputSchema": INPUT_SCHEMA, + } + ] + }, + } + elif method == "tools/call": + params = req.get("params", {}) + tool_name = params.get("name") + tool_args = params.get("arguments", {}) + + if tool_name != TOOL_NAME: + res = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}], + "isError": True, + }, + } + else: + try: + result_data = call_trust(api_base, tool_args) + res = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [ + {"type": "text", "text": json.dumps(result_data, indent=2)} + ] + }, + } + except Exception as e: + res = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": str(e)}], + "isError": True, + }, + } + else: + res = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"}, + } + + sys.stdout.write(json.dumps(res) + "\n") + sys.stdout.flush() diff --git a/spanledger/runner.py b/spanledger/runner.py index b1e68f0..fd3b256 100644 --- a/spanledger/runner.py +++ b/spanledger/runner.py @@ -141,7 +141,7 @@ def run_forever(config_path: str): reg = registry.Registry(epoch, store=store) httpapi.serve( - state, conf.http.listen, store=store, slo_engine=slo_engine, event_engine=event_engine + state, conf.http.listen, store=store, slo_engine=slo_engine, event_engine=event_engine, cfg=conf ) client = signoz.SignozClient(conf.signoz.query_url, conf.signoz.api_key) finalizer = verify.Finalizer() diff --git a/test/test_config.py b/test/test_config.py index af78026..b9ae875 100644 --- a/test/test_config.py +++ b/test/test_config.py @@ -269,3 +269,13 @@ def test_ledger_hop_unknown_key_rejected(tmp_path): cfg_text = VALID + "ledger:\n hops:\n - name: agent\n selector: 'a'\n bogus: 1\n" with pytest.raises(ValueError): config.load(_write(tmp_path, cfg_text)) + + +def test_http_ui_base_url_default_and_override(tmp_path): + c_default = config.load(_write(tmp_path, VALID)) + assert c_default.http.ui_base_url == "http://localhost:5173" + + cfg_text = VALID + "http:\n ui_base_url: http://custom-ui:3000\n" + c_custom = config.load(_write(tmp_path, cfg_text)) + assert c_custom.http.ui_base_url == "http://custom-ui:3000" + diff --git a/test/test_httpapi_v2.py b/test/test_httpapi_v2.py index 868216e..0978e0b 100644 --- a/test/test_httpapi_v2.py +++ b/test/test_httpapi_v2.py @@ -331,3 +331,437 @@ def test_unknown_route_is_404(server): base, *_ = server r = httpx.get(f"{base}/nope") assert r.status_code == 404 + + +def test_trust_verdict_table_four_conditions(server): + base, store, state, event_engine, slo_engine = server + params = {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z", "stream": "gateway-a"} + + # Condition 1: no_data (empty store / no SLI data) + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert body["trustworthy"] is False + assert body["confidence"] == "no_data" + + # Condition 2: trustworthy: true, probe_verified (SLI data present, no loss/unreachable events) + import time as time_mod + store.bump_bucket("gateway-a", "traces", (int(time_mod.time()) - 120) // 60 * 60, good=10, bad=0, unknown=0) + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert body["trustworthy"] is True + assert body["confidence"] == "probe_verified" + + # Condition 3: loss event present -> trustworthy: false, probe_verified + t_from_ns = 1784887200 * 10**9 + t_to_ns = 1784890800 * 10**9 + store.insert_event( + EventRow( + id="loss-1", + class_="loss", + stream="gateway-a", + signal="traces", + severity="critical", + window_from_ns=t_from_ns, + window_to_ns=t_to_ns, + payload={ + "id": "loss-1", + "class": "loss", + "stream": "gateway-a", + "signal": "traces", + "probes": {"sent": 100, "verified": 80, "missing": 20, "unknown": 0}, + "delivery_ratio": 0.80, + "gap_runs": [{"seq_from": 40, "seq_to": 59, "t_from": "2026-07-24T10:10:00Z", "t_to": "2026-07-24T10:15:00Z"}], + "gap_shape": "contiguous", + "traces_filter": "spanledger.stream = 'gateway-a'", + }, + links=[], + emitted_at_ns=t_from_ns, + ) + ) + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert body["trustworthy"] is False + assert body["confidence"] == "probe_verified" + assert len(body["incidents"]) == 1 + + # Condition 4: unreachable event present (without loss) -> trustworthy: false, confidence: unknown + store2 = Store(":memory:") + slo_engine2 = SLOEngine(store2, Config( + signoz=SignozConfig(query_url="http://x", ingest_url="y:4317"), + streams=[StreamConfig(name="gateway-a", endpoint="e")], + ), _NullReporter(), None) + store2.bump_bucket("gateway-a", "traces", (int(time_mod.time()) - 120) // 60 * 60, good=10, bad=0, unknown=0) + store2.insert_event( + EventRow( + id="unreach-1", + class_="backend_unreachable", + stream="gateway-a", + signal="traces", + severity="critical", + window_from_ns=t_from_ns, + window_to_ns=t_to_ns, + payload={"id": "unreach-1", "class": "backend_unreachable"}, + links=[], + emitted_at_ns=t_from_ns, + ) + ) + srv2 = httpapi.serve(state, "127.0.0.1:0", store=store2, slo_engine=slo_engine2) + b2 = f"http://{srv2.server_address[0]}:{srv2.server_address[1]}" + try: + r = httpx.get(f"{b2}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert body["trustworthy"] is False + assert body["confidence"] == "unknown" + finally: + srv2.shutdown() + srv2.server_close() + store2.close() + + +def test_trust_error_responses(server): + base, *_ = server + valid_from = "2026-07-24T10:00:00Z" + valid_to = "2026-07-24T11:00:00Z" + + # 1. Missing from or to + r = httpx.get(f"{base}/api/v2/trust", params={"to": valid_to}) + assert r.status_code == 400 + assert r.json()["title"] == "from and to are required" + + r = httpx.get(f"{base}/api/v2/trust", params={"from": valid_from}) + assert r.status_code == 400 + assert r.json()["title"] == "from and to are required" + + # 2. Unparseable timestamp + r = httpx.get(f"{base}/api/v2/trust", params={"from": "invalid-date", "to": valid_to}) + assert r.status_code == 400 + assert r.json()["title"] == "invalid from" + + r = httpx.get(f"{base}/api/v2/trust", params={"from": valid_from, "to": "invalid-date"}) + assert r.status_code == 400 + assert r.json()["title"] == "invalid to" + + # 3. from >= to + r = httpx.get(f"{base}/api/v2/trust", params={"from": valid_to, "to": valid_from}) + assert r.status_code == 400 + assert r.json()["title"] == "invalid window" + + # 4. Window > 24h + r = httpx.get( + f"{base}/api/v2/trust", + params={"from": "2026-07-20T00:00:00Z", "to": "2026-07-22T00:00:00Z"}, + ) + assert r.status_code == 400 + assert r.json()["title"] == "window too large" + + # 5. Unknown query param + r = httpx.get( + f"{base}/api/v2/trust", + params={"from": valid_from, "to": valid_to, "bogus": "1"}, + ) + assert r.status_code == 400 + assert r.json()["title"] == "unknown query parameter(s)" + + # 6. Unknown stream + r = httpx.get( + f"{base}/api/v2/trust", + params={"from": valid_from, "to": valid_to, "stream": "nonexistent"}, + ) + assert r.status_code == 400 + assert r.json()["title"] == "unknown stream" + + # 7. Invalid signal + r = httpx.get( + f"{base}/api/v2/trust", + params={"from": valid_from, "to": valid_to, "signal": "invalid-signal"}, + ) + assert r.status_code == 400 + assert r.json()["title"] == "invalid signal" + + +def test_trust_recommendation_templates(server): + base, store, state, event_engine, slo_engine = server + import time as time_mod + store.bump_bucket("gateway-a", "traces", (int(time_mod.time()) - 120) // 60 * 60, good=10, bad=0, unknown=0) + + params = {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z", "stream": "gateway-a"} + + # 1. Trustworthy true template + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + rec = r.json()["recommendation"] + assert "Telemetry for this window verified complete" in rec + assert "100.0% delivery" in rec + + # 2. Trustworthy false template with extrapolated spans + t_from_ns = 1784887200 * 10**9 + t_to_ns = 1784890800 * 10**9 + store.insert_event( + EventRow( + id="loss-2", + class_="loss", + stream="gateway-a", + signal="traces", + severity="critical", + window_from_ns=t_from_ns, + window_to_ns=t_to_ns, + payload={ + "id": "loss-2", + "class": "loss", + "stream": "gateway-a", + "signal": "traces", + "probes": {"sent": 100, "verified": 80, "missing": 20, "unknown": 0}, + "delivery_ratio": 0.80, + "gap_runs": [{"seq_from": 40, "seq_to": 59, "t_from": "2026-07-24T10:10:00Z", "t_to": "2026-07-24T10:15:00Z"}], + "gap_shape": "contiguous", + "extrapolated_user_spans_lost": {"estimate": 1840}, + "traces_filter": "spanledger.stream = 'gateway-a'", + }, + links=[], + emitted_at_ns=t_from_ns, + ) + ) + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + rec = body["recommendation"] + assert "Telemetry for this window is incomplete" in rec + assert "20 probes were verified lost (~1840 user spans estimated lost)" in rec + assert "between 2026-07-24T10:10:00Z and 2026-07-24T10:15:00Z" in rec + + +def test_trust_recommendation_degraded_templates(server): + # Plan step 10: for confidence in (unknown, no_data), prefix the FALSE template + # with the degraded-verification sentence and omit only the counts sentence -- + # "Telemetry for this window is incomplete." must survive. + base, store, state, event_engine, slo_engine = server + params = {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z", "stream": "gateway-a"} + + # no_data: empty store, no SLI observation yet + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert body["confidence"] == "no_data" + rec = body["recommendation"] + assert rec.startswith( + "Verification was degraded for this window — spanLedger could not confirm delivery. " + ) + assert "Telemetry for this window is incomplete." in rec + assert "Conclusions drawn from this window may be unreliable." in rec + assert "Verify against the evidence link before diagnosing application behavior." in rec + assert "probes were verified lost" not in rec + + # unknown: backend_unreachable event present, no loss + t_from_ns = 1784887200 * 10**9 + t_to_ns = 1784890800 * 10**9 + store.insert_event( + EventRow( + id="unreach-degraded", + class_="backend_unreachable", + stream="gateway-a", + signal="traces", + severity="critical", + window_from_ns=t_from_ns, + window_to_ns=t_to_ns, + payload={"id": "unreach-degraded", "class": "backend_unreachable"}, + links=[], + emitted_at_ns=t_from_ns, + ) + ) + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert body["confidence"] == "unknown" + rec = body["recommendation"] + assert rec.startswith( + "Verification was degraded for this window — spanLedger could not confirm delivery. " + ) + assert "Telemetry for this window is incomplete." in rec + assert "probes were verified lost" not in rec + + +def test_evidence_links_signoz_url_encoding_matches_frontend(): + # Frozen (Part 3 file mapping): signoz_url must match signoz-links.ts exactly. + # The frontend builds it via URLSearchParams, which encodes space as '+', not + # '%20' -- see frontend/src/test/signoz-links.test.ts. + links = httpapi._evidence_links( + "spanledger.stream = 'gateway-a'", + "incident-1", + signoz_url_base="http://localhost:8080", + ui_base_url="http://localhost:5173", + ) + assert links["signoz_url"] == ( + "http://localhost:8080/traces-explorer?filter=spanledger.stream+%3D+%27gateway-a%27" + ) + + +def test_trust_incidents_capped_at_50(server): + # Frozen response field table: "incidents | ... | Max 50." + base, store, state, event_engine, slo_engine = server + base_ns = 1784887200 * 10**9 + for i in range(55): + t_ns = base_ns + i * 1_000_000_000 + store.insert_event( + EventRow( + id=f"loss-cap-{i}", + class_="loss", + stream="gateway-a", + signal="traces", + severity="critical", + window_from_ns=t_ns, + window_to_ns=t_ns + 1_000_000_000, + payload={ + "id": f"loss-cap-{i}", + "class": "loss", + "stream": "gateway-a", + "signal": "traces", + "probes": {"sent": 10, "verified": 8, "missing": 2, "unknown": 0}, + "delivery_ratio": 0.80, + "gap_runs": [{ + "seq_from": i, "seq_to": i, + "t_from": "2026-07-24T10:10:00Z", "t_to": "2026-07-24T10:10:01Z", + }], + "gap_shape": "contiguous", + "traces_filter": f"spanledger.seq = {i}", + }, + links=[], + emitted_at_ns=t_ns, + ) + ) + params = {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z", "stream": "gateway-a"} + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + assert len(body["incidents"]) == 50 + # cap keeps the most recent 50 -- oldest -> newest ordering preserved within them + assert body["incidents"][0]["id"] == "loss-cap-5" + assert body["incidents"][-1]["id"] == "loss-cap-54" + + +def test_trust_complete_response_shape(): + # Plan M3 Tests: "Integration -- test/test_httpapi_v2.py with a seeded in-memory + # Store (:memory:) containing one loss event, asserting the complete response + # shape." Reproduces the frozen Part 3 response example field-for-field. + from urllib.parse import quote_plus + + store = Store(":memory:") + config = Config( + signoz=SignozConfig(query_url="http://localhost:8080", ingest_url="y:4317"), + streams=[StreamConfig(name="gateway-a", endpoint="e")], + verify=VerifyConfig(poll_interval=15.0, lookback=0.0), + slo=SLOConfig(target=0.99, window_days=1), + ) + state = httpapi.State("epoch-1") + reporter = _NullReporter() + event_engine = EventEngine(store, reporter, state, "epoch-1") + slo_engine = SLOEngine(store, config, reporter, event_engine) + srv = httpapi.serve( + state, "127.0.0.1:0", store=store, slo_engine=slo_engine, + event_engine=event_engine, cfg=config, + ) + base = f"http://{srv.server_address[0]}:{srv.server_address[1]}" + try: + t_base_ns = 1784887200 * 10**9 # 2026-07-24T10:00:00Z + gap_from_ns = t_base_ns + 600 * 10**9 # 10:10:00Z + gap_to_ns = t_base_ns + 900 * 10**9 # 10:15:00Z + dm_ns = t_base_ns + 720 * 10**9 # 10:12:00Z -- inside the affected window + + store.insert_event( + EventRow( + id="loss-full", + class_="loss", + stream="gateway-a", + signal="traces", + severity="critical", + window_from_ns=gap_from_ns, + window_to_ns=gap_to_ns, + payload={ + "id": "loss-full", + "class": "loss", + "stream": "gateway-a", + "signal": "traces", + "probes": {"sent": 100, "verified": 80, "missing": 20, "unknown": 0}, + "delivery_ratio": 0.80, + "gap_runs": [{ + "seq_from": 40, "seq_to": 59, + "t_from": "2026-07-24T10:10:00Z", "t_to": "2026-07-24T10:15:00Z", + }], + "gap_shape": "contiguous", + "extrapolated_user_spans_lost": {"estimate": 1840}, + "traces_filter": "spanledger.stream = 'gateway-a' AND spanledger.seq >= 40 AND spanledger.seq <= 59", + "correlation_hint": { + "metric": "otelcol_exporter_enqueue_failed_spans", "delta": 1840, "hop": "exporter", + }, + }, + links=[], + emitted_at_ns=gap_from_ns, + ) + ) + store.insert_event( + EventRow( + id="dm-1", + class_="deploy_marker", + stream="gateway-a", + signal=None, + severity="info", + window_from_ns=dm_ns, + window_to_ns=dm_ns, + payload={"label": "gateway v1.4"}, + links=[], + emitted_at_ns=dm_ns, + ) + ) + + params = {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z", "stream": "gateway-a"} + r = httpx.get(f"{base}/api/v2/trust", params=params) + assert r.status_code == 200 + body = r.json() + + assert body["trustworthy"] is False + assert body["confidence"] == "probe_verified" + assert body["verified_ratio"] == 0.8 + assert body["unknown_ratio"] == 0.0 + assert body["window"] == {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z"} + assert body["affected_window"] == {"from": "2026-07-24T10:10:00Z", "to": "2026-07-24T10:15:00Z"} + + assert len(body["incidents"]) == 1 + incident = body["incidents"][0] + assert incident["id"] == "loss-full" + assert incident["class"] == "loss" + assert incident["stream"] == "gateway-a" + assert incident["signal"] == "traces" + assert incident["gap_runs"] == [{ + "seq_from": 40, "seq_to": 59, + "t_from": "2026-07-24T10:10:00Z", "t_to": "2026-07-24T10:15:00Z", + }] + assert incident["gap_shape"] == "contiguous" + assert incident["probes_lost"] == 20 + assert incident["extrapolated_user_spans_lost"] == 1840 + assert incident["probable_cause"]["hypothesis"] == "exporter queue overflow or backend outage" + assert incident["probable_cause"]["confidence"] == "medium" + assert incident["probable_cause"]["evidence"][0] == "gap_shape: contiguous" + assert "deploy marker 'gateway v1.4' at" in incident["probable_cause"]["evidence"][1] + assert incident["probable_cause"]["evidence"][2] == ( + "otelcol_exporter_enqueue_failed_spans +1840 at exporter" + ) + + traces_filter = "spanledger.stream = 'gateway-a' AND spanledger.seq >= 40 AND spanledger.seq <= 59" + assert body["evidence"]["traces_filter"] == traces_filter + assert body["evidence"]["signoz_url"] == ( + f"http://localhost:8080/traces-explorer?filter={quote_plus(traces_filter)}" + ) + assert body["evidence"]["spanledger_url"] == "http://localhost:5173/incident/loss-full" + + rec = body["recommendation"] + assert "Telemetry for this window is incomplete" in rec + assert "20 probes were verified lost (~1840 user spans estimated lost)" in rec + assert "between 2026-07-24T10:10:00Z and 2026-07-24T10:15:00Z" in rec + finally: + srv.shutdown() + srv.server_close() + store.close() + diff --git a/test/test_mcp.py b/test/test_mcp.py new file mode 100644 index 0000000..a5cdda0 --- /dev/null +++ b/test/test_mcp.py @@ -0,0 +1,157 @@ +"""Unit tests for the MCP server module (plan Part 3 / M3).""" + +from __future__ import annotations + +import io +import json + +import httpx +import pytest + +from spanledger import mcp + + +def test_input_schema_frozen_fields(): + schema = mcp.INPUT_SCHEMA + assert schema["type"] == "object" + assert set(schema["required"]) == {"from", "to"} + props = schema["properties"] + assert "from" in props + assert "to" in props + assert "stream" in props + assert "signal" in props + + +def test_call_trust_success_verbatim(monkeypatch): + expected_body = { + "trustworthy": True, + "confidence": "probe_verified", + "verified_ratio": 1.0, + "unknown_ratio": 0.0, + "window": {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z"}, + "affected_window": None, + "incidents": [], + "evidence": {"traces_filter": None, "signoz_url": None, "spanledger_url": None}, + "recommendation": "Telemetry for this window verified complete (100.0% delivery, 0.0% unknown). No loss incidents detected.", + } + + class MockResponse: + status_code = 200 + + def json(self): + return expected_body + + class MockClient: + def __init__(self, *a, **k): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def get(self, url, params=None): + assert url == "http://localhost:8231/api/v2/trust" + assert params == {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z"} + return MockResponse() + + monkeypatch.setattr(httpx, "Client", MockClient) + + res = mcp.call_trust( + "http://localhost:8231", + {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z"}, + ) + assert res == expected_body + + +def test_call_trust_non_200_raises_value_error(monkeypatch): + class MockResponse: + status_code = 400 + + def json(self): + return {"title": "invalid from", "detail": "parse error"} + + class MockClient: + def __init__(self, *a, **k): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def get(self, url, params=None): + return MockResponse() + + monkeypatch.setattr(httpx, "Client", MockClient) + + with pytest.raises(ValueError) as excinfo: + mcp.call_trust("http://localhost:8231", {"from": "bad", "to": "bad"}) + assert "invalid from: parse error" in str(excinfo.value) + + +def test_call_trust_http_error_raises_runtime_error(monkeypatch): + class MockClient: + def __init__(self, *a, **k): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def get(self, url, params=None): + raise httpx.ConnectError("Connection refused") + + monkeypatch.setattr(httpx, "Client", MockClient) + + with pytest.raises(RuntimeError) as excinfo: + mcp.call_trust("http://localhost:8231", {"from": "a", "to": "b"}) + assert "spanledger daemon unreachable at http://localhost:8231" in str(excinfo.value) + + +def test_serve_stdio_json_rpc(monkeypatch): + input_lines = [ + json.dumps({"jsonrpc": "2.0", "id": 1, "method": "initialize"}), + json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}), + json.dumps({ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "spanledger_check_telemetry_trust", + "arguments": {"from": "2026-07-24T10:00:00Z", "to": "2026-07-24T11:00:00Z"}, + }, + }), + ] + + stdin_mock = io.StringIO("\n".join(input_lines) + "\n") + stdout_mock = io.StringIO() + + monkeypatch.setattr(mcp.sys, "stdin", stdin_mock) + monkeypatch.setattr(mcp.sys, "stdout", stdout_mock) + monkeypatch.setattr( + mcp, + "call_trust", + lambda api_base, args: {"trustworthy": True, "confidence": "probe_verified"}, + ) + + mcp.serve_stdio("http://localhost:8231") + + output_lines = [line for line in stdout_mock.getvalue().split("\n") if line.strip()] + assert len(output_lines) == 3 + + resp1 = json.loads(output_lines[0]) + assert resp1["id"] == 1 + assert resp1["result"]["serverInfo"]["name"] == "spanledger-mcp" + + resp2 = json.loads(output_lines[1]) + assert resp2["id"] == 2 + assert resp2["result"]["tools"][0]["name"] == "spanledger_check_telemetry_trust" + + resp3 = json.loads(output_lines[2]) + assert resp3["id"] == 3 + assert "trustworthy" in resp3["result"]["content"][0]["text"] diff --git a/test/test_scenario.py b/test/test_scenario.py index 8000738..b175df4 100644 --- a/test/test_scenario.py +++ b/test/test_scenario.py @@ -139,3 +139,32 @@ def test_fault_logs_outage_swaps_config_and_restores(monkeypatch, tmp_path): restore() assert original_path.read_text(encoding="utf-8") == "original: true\n" assert docker_calls == [("restart", rs.GATEWAY), ("restart", rs.GATEWAY)] + + +def test_overflow_scenario_registered_and_evaluates(): + # Plan Part 2 physics (frozen): the agent's bounded queue fills and stays full for + # the throttle's duration, so overflow drops land as one contiguous run, not a + # periodic/scattered pattern -- gap_shape == "contiguous" is the M2 acceptance + # criterion (docs/v2/IMPL_TRUST_LAYER.md M2 checklist). + assert "overflow" in rs.SCENARIOS + sc = rs.SCENARIOS["overflow"] + assert "queue overflow" in sc.blurb + assert sc.expect.gap_shapes == ("contiguous",) + # seqs 0..39 sent, queue overflow drops 20..39 as one unbroken run once full + result = _reconcile(range(40), range(20)) + ok, detail = rs.evaluate(result, sc.expect) + assert ok, detail + assert "contiguous" in detail + + +def test_overflow_scenario_rejects_striped_pattern(): + # A striped/periodic drop pattern is NOT what a bounded, saturated queue produces + # under sustained overload -- guard against silently loosening the Expectation + # back to accepting shapes the physics don't predict. + sc = rs.SCENARIOS["overflow"] + arrived = [s for s in range(40) if s not in {5, 15, 25, 35}] # striped + result = _reconcile(range(40), arrived) + ok, detail = rs.evaluate(result, sc.expect) + assert not ok + assert "shape" in detail +