From 722a24b05297a99785c64f60e10382dfdfa76bf2 Mon Sep 17 00:00:00 2001 From: ThorstenHellert Date: Tue, 11 Aug 2026 15:13:22 +0200 Subject: [PATCH] feat(archiver): explain empty archiver reads with coverage verdicts An archiver_read that finds nothing now says why. The response carries a coverage block naming what the emptiness means -- the window precedes or follows the archive's real bounds, the channel was never recorded, or the window holds a genuine gap -- so the agent can tell an unarchived past from recorded silence instead of guessing at a bare zero. Verdicts are derived only from what the connector reports: bounds come from the store's actual oldest and newest samples, a backend that reports no bounds yields "unknown" rather than a guess, and a failed probe becomes a note on the verdict rather than the loss of whatever data did come back. Probes run only for channels that returned empty, so a fully answered query costs nothing and changes shape not at all. The archiver-world e2e now asserts the same claim at the tool surface: the bound shown for a pre-coverage window is the deployed store's true oldest sample, not a declared window. --- CHANGELOG.md | 5 + .../control_system/tools/archiver_read.py | 149 +++++++++++- tests/e2e/test_archiver_world_e2e.py | 32 ++- tests/mcp_server/test_archiver_read_tool.py | 213 +++++++++++++++++- 4 files changed, 394 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ed5bd10e..a0d46c908 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,11 @@ Compatibility is documented in release notes, not encoded in the version string. ### Added +- An archiver read that comes back empty now says why: the response carries a + coverage verdict — the window predates or postdates the archive, the channel + was never recorded, or the window holds a genuine gap — with the archive's + real bounds, so an empty answer is never a silent one. + - A virtual accelerator can now be deployed with a real archive behind it: a MongoDB store plus an archiver-recorder service that records the machine's channels as they move. Scenario history is seeded into the store when the diff --git a/src/osprey/mcp_server/control_system/tools/archiver_read.py b/src/osprey/mcp_server/control_system/tools/archiver_read.py index 0caab79b8..eaec8c4a5 100644 --- a/src/osprey/mcp_server/control_system/tools/archiver_read.py +++ b/src/osprey/mcp_server/control_system/tools/archiver_read.py @@ -2,8 +2,9 @@ import json import logging -from datetime import datetime, timedelta -from typing import Any +from collections import Counter +from datetime import UTC, datetime, timedelta +from typing import Any, NamedTuple import pandas as pd @@ -50,6 +51,137 @@ def _parse_time(time_str: str) -> datetime: return dt +class _CoverageProbe(NamedTuple): + """What the connector could say about one empty channel.""" + + available: bool | None + metadata: Any # ArchiverMetadata | None — untyped to keep the import lazy + note: str | None + + +def _pin_utc(dt: datetime | None) -> datetime | None: + """pymongo reads naive datetimes as UTC; make that explicit before comparing.""" + if dt is None: + return None + return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt.astimezone(UTC) + + +def _classify_coverage( + start_utc: datetime, end_utc: datetime, probe: _CoverageProbe +) -> dict[str, Any]: + """One empty channel's entry: a verdict, plus bounds when they are known.""" + md = probe.metadata + if probe.available is False or (md is not None and not md.is_archived): + return {"verdict": "never_recorded"} + a_start = _pin_utc(md.archival_start) if md is not None else None + a_end = _pin_utc(md.archival_end) if md is not None else None + if a_start is None or a_end is None: + entry: dict[str, Any] = {"verdict": "coverage_unknown"} + if probe.note: + entry["note"] = probe.note + return entry + if end_utc < a_start: + verdict = "window_precedes_archive" + elif start_utc > a_end: + verdict = "window_follows_archive" + else: + verdict = "gap_within_coverage" + return { + "verdict": verdict, + "archive_start": a_start.isoformat(), + "archive_end": a_end.isoformat(), + } + + +def _coverage_message( + start_utc: datetime, end_utc: datetime, overall: str, channels: dict[str, dict[str, Any]] +) -> str: + """Plain language for the agent: the asked window, the real bounds, no guessing.""" + window = f"{start_utc.isoformat()} → {end_utc.isoformat()}" + n = len(channels) + noun = "1 channel" if n == 1 else f"{n} channels" + if overall == "window_precedes_archive": + oldest = min(entry["archive_start"] for entry in channels.values()) + return ( + f"No archived data for {noun} in {window}: archiving begins {oldest}, " + f"after the queried window ends. The archive reports only what it holds — " + f"nothing is extrapolated. Data exists from {oldest} onward." + ) + if overall == "window_follows_archive": + newest = max(entry["archive_end"] for entry in channels.values()) + return ( + f"No archived data for {noun} in {window}: the archive's newest sample is " + f"{newest}, before the queried window begins. If this deployment records " + f"continuously, recording may have stopped." + ) + if overall == "never_recorded": + return ( + f"Not in this archive: {', '.join(channels)}. No history was ever " + f"recorded for {'this channel' if n == 1 else 'these channels'}." + ) + if overall == "gap_within_coverage": + return ( + f"The queried window {window} lies inside archive coverage but holds no " + f"samples for {noun}: nothing was recorded there. A gap is recorded " + f"silence — the channel was not answering — not data awaiting synthesis." + ) + if overall == "coverage_unknown": + return ( + f"No data in {window} for {noun}, and this archiver backend reports no " + f"coverage bounds, so the reason cannot be determined from here." + ) + counts = Counter(entry["verdict"] for entry in channels.values()) + parts = ", ".join(f"{count}× {verdict}" for verdict, count in counts.items()) + return ( + f"No archived data in {window} for {noun}, for differing reasons: {parts}. " + f"See coverage.channels for each channel's verdict and bounds." + ) + + +async def _probe_coverage(connector: Any, channels: list[str]) -> dict[str, _CoverageProbe]: + """Ask the connector about each empty channel; a failed probe is a note. + + Failures degrade to ``coverage_unknown`` downstream rather than raising: + the data that DID come back must never be lost to its own explanation. + """ + try: + availability = await connector.check_availability(channels) + except Exception: # noqa: BLE001 — any probe failure degrades, none propagate + availability = {} + probes: dict[str, _CoverageProbe] = {} + for ch in channels: + try: + md = await connector.get_metadata(ch) + except Exception as exc: # noqa: BLE001 + probes[ch] = _CoverageProbe( + available=availability.get(ch), metadata=None, note=f"metadata probe failed: {exc}" + ) + else: + probes[ch] = _CoverageProbe(available=availability.get(ch), metadata=md, note=None) + return probes + + +def _compose_coverage( + start_utc: datetime, end_utc: datetime, probes: dict[str, _CoverageProbe] +) -> dict[str, Any] | None: + """Explain the empty channels of a read, from facts — or admit not knowing. + + Returns ``None`` when nothing was empty (the common path adds no block, no + tokens, no schema noise). Never raises: an explanation that failed to + compose must not cost the agent the data that DID come back. + """ + if not probes: + return None + channels = {ch: _classify_coverage(start_utc, end_utc, probe) for ch, probe in probes.items()} + verdicts = {entry["verdict"] for entry in channels.values()} + overall = next(iter(verdicts)) if len(verdicts) == 1 else "mixed" + return { + "verdict": overall, + "message": _coverage_message(start_utc, end_utc, overall, channels), + "channels": channels, + } + + @mcp.tool() async def archiver_read( channels: list[str], @@ -77,7 +209,10 @@ async def archiver_read( Returns: JSON summary with per-channel point counts and stats, and the data - file path. + file path. When a requested channel has zero points in the window, + ``summary.coverage`` explains why — the window precedes/follows the + archive's real bounds, the channel was never recorded, or the window + holds an honest gap — so an empty answer is never a silent one. """ if not channels: return make_error( @@ -177,6 +312,12 @@ async def archiver_read( stats["mean"] = round(float(numeric.mean()), 6) per_channel[ch] = stats + empty_channels = [ch for ch in unique_channels if per_channel[ch]["points"] == 0] + coverage = None + if empty_channels: + probes = await _probe_coverage(connector, empty_channels) + coverage = _compose_coverage(start_dt.astimezone(UTC), end_dt.astimezone(UTC), probes) + # Full data payload goes to file; compact summary returned inline data_payload = { "query": { @@ -195,6 +336,8 @@ async def archiver_read( "time_range": {"start": str(start_dt), "end": str(end_dt)}, "per_channel": per_channel, } + if coverage is not None: + summary["coverage"] = coverage access_details = { "data_file_structure": { "root_keys": ["query", "series"], diff --git a/tests/e2e/test_archiver_world_e2e.py b/tests/e2e/test_archiver_world_e2e.py index fe69f22ed..f39396a6b 100644 --- a/tests/e2e/test_archiver_world_e2e.py +++ b/tests/e2e/test_archiver_world_e2e.py @@ -723,7 +723,7 @@ async def _write_a_new_setpoint() -> tuple[float, datetime, Any]: time.sleep(RECORDER_POLL_SEC) -def test_a_window_before_coverage_is_reported_as_empty_not_invented(archiver_world): +def test_a_window_before_coverage_is_reported_as_empty_not_invented(archiver_world, monkeypatch): """The honesty claim, stated twice: no points, and a truthful start. An archiver that answers a pre-archival question with synthesized values is @@ -759,6 +759,36 @@ def test_a_window_before_coverage_is_reported_as_empty_not_invented(archiver_wor # Reported coverage is the oldest sample really held, not a declared window. assert abs((archival_start - actual_oldest).total_seconds()) < 1.0 + # -- the same emptiness, as the AGENT is told it -------------------------- + # Through the real MCP tool against the deployed store: the connector + # reporting an empty frame and the agent being TOLD why are different + # claims, and only the second one closes the gap this feature is about. + from tests.mcp_server.conftest import extract_response_dict, get_tool_fn + + monkeypatch.chdir(archiver_world.project_dir) + from osprey.mcp_server.control_system.server_context import initialize_server_context + + initialize_server_context() + from osprey.mcp_server.control_system.tools.archiver_read import archiver_read + + tool = get_tool_fn(archiver_read) + response = asyncio.run( + tool( + channels=[channel], + start_time=before.isoformat(), + end_time=(archival_start - timedelta(hours=1)).isoformat(), + ) + ) + payload = extract_response_dict(response) + assert payload["status"] == "success" + coverage = payload["summary"]["coverage"] + assert coverage["verdict"] == "window_precedes_archive", coverage + # The bound the agent is shown is the store's true oldest sample, not a + # declared window — same claim as above, now at the tool surface. + reported_start = datetime.fromisoformat(coverage["channels"][channel]["archive_start"]) + oldest_utc = actual_oldest if actual_oldest.tzinfo else actual_oldest.replace(tzinfo=UTC) + assert abs((reported_start - oldest_utc).total_seconds()) < 1.0 + # --------------------------------------------------------------------------- # The seam diff --git a/tests/mcp_server/test_archiver_read_tool.py b/tests/mcp_server/test_archiver_read_tool.py index 6a014b0e5..9d65a12a6 100644 --- a/tests/mcp_server/test_archiver_read_tool.py +++ b/tests/mcp_server/test_archiver_read_tool.py @@ -10,7 +10,7 @@ """ import json -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta from unittest.mock import AsyncMock, patch import pandas as pd @@ -72,8 +72,19 @@ def archiver_read_tool(archiver_project): Set ``connector.get_data.return_value`` (or ``.side_effect``), then await the tool. Tests that need the real ``MockArchiverConnector`` take ``archiver_project`` instead. + + The coverage probes carry benign defaults — every channel available, + metadata archived but boundless — because a bare ``AsyncMock`` would hand + the empty-channel path awaitable mocks instead of ``ArchiverMetadata`` and + blow up on the first datetime comparison. Boundless metadata classifies as + ``coverage_unknown``, the verdict that asserts nothing about a store this + mock never had. Tests that exercise a specific verdict override them. """ + from osprey.connectors.archiver.base import ArchiverMetadata + connector = AsyncMock() + connector.check_availability.side_effect = lambda chans: dict.fromkeys(chans, True) + connector.get_metadata.side_effect = lambda ch: ArchiverMetadata(pv_name=ch, is_archived=True) with patch( "osprey.connectors.factory.ConnectorFactory.create_archiver_connector", new_callable=AsyncMock, @@ -883,3 +894,203 @@ async def channel_series(mode): # recorded timestamp, which here falls inside the bin, not on its edge. assert mean["timestamps"] == [f"2024-01-15T10:0{minute}:00+00:00" for minute in range(6)] assert raw["timestamps"] != mean["timestamps"] + + +# --------------------------------------------------------------------------- +# Coverage verdicts (pure composer) +# --------------------------------------------------------------------------- + + +def _probe(available=True, start=None, end=None, is_archived=True, note=None): + """A _CoverageProbe with an ArchiverMetadata built from bare bounds.""" + from osprey.connectors.archiver.base import ArchiverMetadata + from osprey.mcp_server.control_system.tools.archiver_read import _CoverageProbe + + md = ArchiverMetadata( + pv_name="X", is_archived=is_archived, archival_start=start, archival_end=end + ) + return _CoverageProbe(available=available, metadata=md, note=note) + + +_WIN_START = datetime(2026, 7, 5, tzinfo=UTC) +_WIN_END = datetime(2026, 7, 6, tzinfo=UTC) +_ARC_START = datetime(2026, 7, 12, 9, 0, tzinfo=UTC) +_ARC_END = datetime(2026, 8, 11, 10, 40, tzinfo=UTC) + + +class TestComposeCoverage: + """The pure composer: verdicts, bounds, degradation — no I/O anywhere.""" + + def _compose(self, probes, start=_WIN_START, end=_WIN_END): + from osprey.mcp_server.control_system.tools.archiver_read import _compose_coverage + + return _compose_coverage(start, end, probes) + + def test_no_empty_channels_composes_nothing(self): + assert self._compose({}) is None + + def test_window_before_the_oldest_sample_precedes_the_archive(self): + block = self._compose({"CH": _probe(start=_ARC_START, end=_ARC_END)}) + assert block["verdict"] == "window_precedes_archive" + entry = block["channels"]["CH"] + assert entry["verdict"] == "window_precedes_archive" + assert entry["archive_start"] == _ARC_START.isoformat() + assert entry["archive_end"] == _ARC_END.isoformat() + # The message states the real bound and rules out synthesis. + assert _ARC_START.isoformat() in block["message"] + assert "extrapolat" in block["message"] + + def test_window_after_the_newest_sample_follows_the_archive(self): + block = self._compose( + {"CH": _probe(start=_ARC_START, end=_ARC_END)}, + start=_ARC_END + timedelta(days=1), + end=_ARC_END + timedelta(days=2), + ) + assert block["verdict"] == "window_follows_archive" + assert _ARC_END.isoformat() in block["message"] + + def test_window_overlapping_coverage_is_an_honest_gap(self): + block = self._compose( + {"CH": _probe(start=_ARC_START, end=_ARC_END)}, + start=_ARC_START + timedelta(days=1), + end=_ARC_START + timedelta(days=2), + ) + assert block["verdict"] == "gap_within_coverage" + assert "recorded" in block["message"] + + def test_window_touching_the_archive_start_counts_as_overlap(self): + # end == archival_start is NOT "precedes": the boundary sample's + # instant is inside coverage, so an empty result there is a gap. + block = self._compose( + {"CH": _probe(start=_ARC_START, end=_ARC_END)}, + start=_ARC_START - timedelta(days=1), + end=_ARC_START, + ) + assert block["verdict"] == "gap_within_coverage" + + def test_unavailable_channel_was_never_recorded(self): + block = self._compose({"CH": _probe(available=False, start=_ARC_START, end=_ARC_END)}) + assert block["verdict"] == "never_recorded" + # No bounds on a channel that has none. + assert "archive_start" not in block["channels"]["CH"] + + def test_metadata_saying_not_archived_means_never_recorded(self): + block = self._compose({"CH": _probe(is_archived=False)}) + assert block["verdict"] == "never_recorded" + + def test_boundless_metadata_is_unknown_not_guessed(self): + # The EPICS/DOOCS connectors report is_archived=True with no bounds. + block = self._compose({"CH": _probe(start=None, end=None)}) + assert block["verdict"] == "coverage_unknown" + assert "archive_start" not in block["channels"]["CH"] + + def test_probe_failure_note_is_carried_not_raised(self): + from osprey.mcp_server.control_system.tools.archiver_read import _CoverageProbe + + probe = _CoverageProbe(available=None, metadata=None, note="metadata probe failed: boom") + block = self._compose({"CH": probe}) + assert block["verdict"] == "coverage_unknown" + assert block["channels"]["CH"]["note"] == "metadata probe failed: boom" + + def test_naive_metadata_bounds_are_pinned_utc(self): + # pymongo hands back naive datetimes read as UTC; comparison must not + # depend on which layer attached the zone. + block = self._compose( + {"CH": _probe(start=_ARC_START.replace(tzinfo=None), end=_ARC_END.replace(tzinfo=None))} + ) + assert block["verdict"] == "window_precedes_archive" + assert block["channels"]["CH"]["archive_start"] == _ARC_START.isoformat() + + def test_differing_verdicts_report_mixed_with_per_channel_detail(self): + block = self._compose( + { + "OLD": _probe(start=_ARC_START, end=_ARC_END), + "GONE": _probe(available=False), + } + ) + assert block["verdict"] == "mixed" + assert block["channels"]["OLD"]["verdict"] == "window_precedes_archive" + assert block["channels"]["GONE"]["verdict"] == "never_recorded" + assert "coverage.channels" in block["message"] + + +class TestCoverageInToolResponse: + """The block reaches the agent — and costs nothing when nothing is empty.""" + + async def test_empty_channel_gets_a_coverage_block(self, archiver_read_tool): + from osprey.connectors.archiver.base import ArchiverMetadata + + fn, connector = archiver_read_tool + connector.get_data.return_value = _make_archiver_df({}) # nothing at all + # The fixture's benign probe defaults are side_effects; clear them or + # these return_values are silently ignored (side_effect wins in Mock). + connector.check_availability.side_effect = None + connector.check_availability.return_value = {"SR:OLD:RB": True} + connector.get_metadata.side_effect = None + connector.get_metadata.return_value = ArchiverMetadata( + pv_name="SR:OLD:RB", + is_archived=True, + archival_start=_ARC_START, + archival_end=_ARC_END, + ) + + result = await fn( + channels=["SR:OLD:RB"], + start_time="2026-07-05T00:00:00+00:00", + end_time="2026-07-06T00:00:00+00:00", + ) + + data = extract_response_dict(result) + assert data["status"] == "success" + cov = data["summary"]["coverage"] + assert cov["verdict"] == "window_precedes_archive" + assert cov["channels"]["SR:OLD:RB"]["archive_start"] == _ARC_START.isoformat() + + async def test_fully_answered_query_has_no_block_and_no_probe(self, archiver_read_tool): + fn, connector = archiver_read_tool + connector.get_data.return_value = _make_archiver_df({"SR:CURRENT:RB": [500.0]}) + + result = await fn(channels=["SR:CURRENT:RB"], start_time="2024-01-15T10:00:00") + + data = extract_response_dict(result) + assert "coverage" not in data["summary"] + # The probes are the empty path's cost, and only the empty path's. + connector.get_metadata.assert_not_awaited() + connector.check_availability.assert_not_awaited() + + async def test_partial_result_explains_only_the_empty_channel(self, archiver_read_tool): + from osprey.connectors.archiver.base import ArchiverMetadata + + fn, connector = archiver_read_tool + connector.get_data.return_value = _make_archiver_df({"SR:CURRENT:RB": [500.0, 500.1]}) + connector.check_availability.side_effect = None + connector.check_availability.return_value = {"SR:VOID:RB": False} + connector.get_metadata.side_effect = None + connector.get_metadata.return_value = ArchiverMetadata( + pv_name="SR:VOID:RB", is_archived=False + ) + + result = await fn( + channels=["SR:CURRENT:RB", "SR:VOID:RB"], start_time="2024-01-15T10:00:00" + ) + + data = extract_response_dict(result) + cov = data["summary"]["coverage"] + assert cov["verdict"] == "never_recorded" + assert list(cov["channels"]) == ["SR:VOID:RB"] + # The answered channel's data is untouched. + assert data["summary"]["per_channel"]["SR:CURRENT:RB"]["points"] == 2 + + async def test_probe_failure_degrades_to_unknown_not_error(self, archiver_read_tool): + fn, connector = archiver_read_tool + connector.get_data.return_value = _make_archiver_df({}) + connector.check_availability.side_effect = RuntimeError("store went away") + connector.get_metadata.side_effect = RuntimeError("store went away") + + result = await fn(channels=["SR:OLD:RB"], start_time="2024-01-15T10:00:00") + + data = extract_response_dict(result) + assert data["status"] == "success" # the read itself did not fail + cov = data["summary"]["coverage"] + assert cov["verdict"] == "coverage_unknown" + assert "store went away" in cov["channels"]["SR:OLD:RB"]["note"]