fix(research): keep live fetch timestamps distinct from historical cutoffs

scripts/build_context_bundle.py

@@ -20,6 +20,7 @@
     write_context_bundle,
 )
 from research_signal_context_pipelines.research_context_adapter import ResearchContextAdapter  # noqa: E402
+from research_signal_context_pipelines.price_history import parse_price_date  # noqa: E402
 from research_signal_context_pipelines.theme_universe import (  # noqa: E402
     build_theme_context,
     load_symbol_theme_exposure,
@@ -70,11 +71,18 @@ def main() -> int:
 
     web_research_context = None
     if args.web_research_sources:
-        web_research_context = ResearchContextAdapter(
+        research_adapter = ResearchContextAdapter(
             Path(args.web_research_sources),
             timeout_seconds=args.web_research_timeout,
             max_entries=args.web_research_max_entries,
-        ).build_context(pit_timestamp=generated_at)
+        )
+        end_date = parse_price_date(args.end_date) if args.end_date else None
+        if end_date is not None and end_date < generated_at.date():
+            # Calendar-date marker only, not an asserted market-close timestamp.
+            historical_cutoff = dt.datetime.combine(end_date, dt.time.min, tzinfo=dt.timezone.utc)
+            web_research_context = research_adapter.build_context(pit_timestamp=historical_cutoff)
+        else:
+            web_research_context = research_adapter.build_context()
 
     try:
         bundle = build_context_from_source(

src/research_signal_context_pipelines/research_context_adapter.py

@@ -252,12 +252,18 @@ def __init__(
         self.max_entries = int(max_entries)
 
     def build_context(self, *, pit_timestamp: dt.datetime | None = None) -> dict[str, Any]:
-        fetched_at = _isoformat_utc(pit_timestamp)
+        """Fetch current observations; past cutoffs require saved, then-visible data."""
+        started_at = dt.datetime.now(dt.timezone.utc)
+        cutoff = dt.datetime.fromisoformat(_isoformat_utc(pit_timestamp)) if pit_timestamp is not None else None
         context: dict[str, Any] = {
-            "pit_timestamp": fetched_at,
+            "pit_timestamp": _isoformat_utc(cutoff or started_at),
             "research_sources": [],
+            "source_count": 0,
             "warnings": [],
         }
+        if cutoff is not None and cutoff < started_at:
+            context["warnings"].append("historical live-web research is unsupported; saved observations visible at the requested cutoff are required")
+            return context
         if not self.sources_path.exists():
             context["warnings"].append(f"research sources file not found: {self.sources_path}")
             return context
@@ -285,10 +291,15 @@ def build_context(self, *, pit_timestamp: dt.datetime | None = None) -> dict[str
                 with urlopen(request, timeout=self.timeout_seconds) as response:  # noqa: S310 - operator-controlled research fetch.
                     body = response.read()
                     content_type = response.headers.get_content_type() if hasattr(response.headers, "get_content_type") else None
+                fetched_time = dt.datetime.now(dt.timezone.utc)
+                fetched_at = _isoformat_utc(fetched_time)
             except (OSError, URLError, TimeoutError, ValueError) as exc:
                 warnings.append(f"failed to fetch research source {source.url}: {type(exc).__name__}: {exc}")
                 continue
 
+            if cutoff is not None and fetched_time > cutoff:
+                warnings.append("skipped research response fetched after the requested cutoff")
+                continue
             kind = _source_kind(source, content_type=content_type, body=body)
             try:
                 entries = _feed_entries(body, source_url=source.url, fetched_at=fetched_at) if kind == "rss" else [_html_entry(body, source_url=source.url, fetched_at=fetched_at)]
@@ -299,8 +310,18 @@ def build_context(self, *, pit_timestamp: dt.datetime | None = None) -> dict[str
             for entry in entries:
                 if len(collected) >= self.max_entries:
                     break
+                published_at = entry.get("published_at")
+                if not published_at:
+                    warnings.append("skipped research entry with missing or invalid publication time")
+                    continue
+                if dt.datetime.fromisoformat(published_at) > fetched_time:
+                    warnings.append("skipped research entry with publication time after its observed fetch time")
+                    continue
                 collected.append(entry)
 
+        if cutoff is None:
+            # Current-query cutoff is completion, not a timestamp before I/O.
+            context["pit_timestamp"] = _isoformat_utc(None)
         context["research_sources"] = collected
         context["source_count"] = len(collected)
         return context

tests/test_research_context_adapter.py

@@ -2,9 +2,47 @@
 
 import datetime as dt
 import json
+import importlib.util
+import sys
+
+import pytest
 from pathlib import Path
 
 from research_signal_context_pipelines.research_context_adapter import ResearchContextAdapter
+from research_signal_context_pipelines import research_context_adapter as adapter_module
+
+
+@pytest.fixture(autouse=True)
+def clock(monkeypatch):
+    real_datetime = dt.datetime
+
+    class Clock(real_datetime):
+        current = real_datetime(2026, 9, 5, 12, tzinfo=dt.timezone.utc)
+        observed = []
+
+        @classmethod
+        def now(cls, tz=None):
+            value = cls.current
+            cls.observed.append(value)
+            cls.current += dt.timedelta(milliseconds=1)
+            return value.astimezone(tz) if tz else value.replace(tzinfo=None)
+
+    monkeypatch.setattr(adapter_module.dt, "datetime", Clock)
+    return Clock
+
+
+def _config(tmp_path, kind="rss"):
+    path = tmp_path / "synthetic_sources.json"
+    path.write_text(json.dumps({"whitelist": ["https://synthetic.example"],
+                                "sources": [{"url": "https://synthetic.example/source", "kind": kind}]}))
+    return path
+
+
+def _rss(*dates):
+    return "<rss><channel>" + "".join(
+        f"<item><title>synthetic-{index}</title><pubDate>{date}</pubDate></item>"
+        for index, date in enumerate(dates)
+    ) + "</channel></rss>"
 
 
 class _FakeHeaders:
@@ -33,7 +71,7 @@ def __exit__(self, exc_type, exc, tb) -> bool:
         return False
 
 
-def test_research_context_adapter_respects_whitelist_and_max_entries(tmp_path: Path, monkeypatch) -> None:
+def test_research_context_adapter_respects_whitelist_and_max_entries(tmp_path: Path, monkeypatch, clock) -> None:
     config_path = tmp_path / "web_research.json"
     config_path.write_text(
         json.dumps(
@@ -75,11 +113,9 @@ def fake_urlopen(request, timeout):
 
     monkeypatch.setattr("research_signal_context_pipelines.research_context_adapter.urlopen", fake_urlopen)
 
-    context = ResearchContextAdapter(config_path, timeout_seconds=3.0, max_entries=1).build_context(
-        pit_timestamp=dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
-    )
+    context = ResearchContextAdapter(config_path, timeout_seconds=3.0, max_entries=1).build_context()
 
-    assert context["pit_timestamp"] == "2026-01-01T00:00:00Z"
+    assert context["pit_timestamp"] == clock.observed[-1].isoformat().replace("+00:00", "Z")
     assert context["source_count"] == 1
     assert [item["title"] for item in context["research_sources"]] == ["First signal"]
     assert context["research_sources"][0]["url"] == "https://allowed.example/a"
@@ -103,15 +139,13 @@ def fake_urlopen(request, timeout):  # pragma: no cover - should not be called
 
     monkeypatch.setattr("research_signal_context_pipelines.research_context_adapter.urlopen", fake_urlopen)
 
-    context = ResearchContextAdapter(config_path, timeout_seconds=3.0, max_entries=3).build_context(
-        pit_timestamp=dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
-    )
+    context = ResearchContextAdapter(config_path, timeout_seconds=3.0, max_entries=3).build_context()
 
     assert context["research_sources"] == []
     assert any("skipped non-whitelisted source" in warning for warning in context["warnings"])
 
 
-def test_research_context_adapter_extracts_html_news(tmp_path: Path, monkeypatch) -> None:
+def test_research_context_adapter_extracts_html_news(tmp_path: Path, monkeypatch, clock) -> None:
     config_path = tmp_path / "web_research.json"
     config_path.write_text(
         json.dumps(
@@ -143,9 +177,7 @@ def fake_urlopen(request, timeout):
 
     monkeypatch.setattr("research_signal_context_pipelines.research_context_adapter.urlopen", fake_urlopen)
 
-    context = ResearchContextAdapter(config_path, timeout_seconds=5.0, max_entries=3).build_context(
-        pit_timestamp=dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc)
-    )
+    context = ResearchContextAdapter(config_path, timeout_seconds=5.0, max_entries=3).build_context()
 
     assert context["source_count"] == 1
     assert context["research_sources"] == [
@@ -156,6 +188,117 @@ def fake_urlopen(request, timeout):
             "url": "https://news.example/story",
             "source_url": "https://news.example/story",
             "source_type": "news",
-            "fetched_at": "2026-01-01T00:00:00Z",
+            "fetched_at": (clock.observed[0] + dt.timedelta(milliseconds=1)).isoformat().replace("+00:00", "Z"),
         }
     ]
+
+
+@pytest.mark.parametrize("cutoff", [dt.datetime(2026, 9, 4, tzinfo=dt.timezone.utc), dt.datetime(2026, 9, 4)])
+def test_historical_request_is_empty_unsupported_and_does_not_fetch(tmp_path, monkeypatch, cutoff):
+    calls = []
+    def fetch(*args, **kwargs):
+        calls.append(args)
+        return _FakeResponse(_rss("2026-09-03T00:00:00Z"), "application/rss+xml")
+    monkeypatch.setattr(adapter_module, "urlopen", fetch)
+    context = ResearchContextAdapter(_config(tmp_path)).build_context(pit_timestamp=cutoff)
+    assert calls == []
+    assert context["research_sources"] == []
+    assert context["source_count"] == 0
+    assert context["pit_timestamp"] == "2026-09-04T00:00:00Z"
+    assert any("unsupported" in warning and "saved" in warning for warning in context["warnings"])
+
+
+def test_current_fetch_timestamps_are_observed_after_response_not_query_start(tmp_path, monkeypatch, clock):
+    class DelayedResponse(_FakeResponse):
+        def read(self):
+            clock.current += dt.timedelta(seconds=1)
+            return super().read()
+
+    monkeypatch.setattr(adapter_module, "urlopen", lambda *args, **kwargs: DelayedResponse(
+        _rss("2026-09-05T11:59:00Z"), "application/rss+xml"))
+    context = ResearchContextAdapter(_config(tmp_path)).build_context()
+    assert context["source_count"] == 1
+    entry = context["research_sources"][0]
+    fetched = dt.datetime.fromisoformat(entry["fetched_at"].replace("Z", "+00:00"))
+    cutoff = dt.datetime.fromisoformat(context["pit_timestamp"].replace("Z", "+00:00"))
+    assert clock.observed[0] < fetched <= cutoff
+    assert fetched == clock.observed[1]
+
+
+@pytest.mark.parametrize("kind", ["rss", "news"])
+@pytest.mark.parametrize("published", ["2026-09-06T00:00:00Z", "invalid", ""])
+def test_future_and_unknown_publication_times_are_not_pit_evidence(tmp_path, monkeypatch, kind, published):
+    body = _rss(published) if kind == "rss" else f'<html><title>synthetic</title><meta property="article:published_time" content="{published}"></html>'
+    monkeypatch.setattr(adapter_module, "urlopen", lambda *args, **kwargs: _FakeResponse(body, "text/html"))
+    context = ResearchContextAdapter(_config(tmp_path, kind)).build_context()
+    assert context["research_sources"] == []
+    assert context["source_count"] == 0
+    assert context["warnings"]
+
+
+def test_future_feed_item_does_not_consume_entry_limit(tmp_path, monkeypatch):
+    monkeypatch.setattr(adapter_module, "urlopen", lambda *args, **kwargs: _FakeResponse(
+        _rss("2026-09-06T00:00:00Z", "2026-09-04T00:00:00Z"), "application/rss+xml"))
+    context = ResearchContextAdapter(_config(tmp_path), max_entries=1).build_context()
+    assert [entry["title"] for entry in context["research_sources"]] == ["synthetic-1"]
+
+
+@pytest.mark.parametrize("end_date,current", [(None, True), ("2026-09-05", True), ("2026-09-06", True), ("2026-09-04", False)])
+def test_cli_routes_current_and_explicit_historical_web_without_price_or_artifact_io(tmp_path, monkeypatch, clock, end_date, current):
+    script = Path(__file__).resolve().parents[1] / "scripts/build_context_bundle.py"
+    spec = importlib.util.spec_from_file_location("synthetic_build_context_cli", script)
+    cli = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(cli)
+    calls, captured, written = [], {}, []
+
+    def fetch(*args, **kwargs):
+        calls.append(True)
+        return _FakeResponse(_rss("2026-09-04T00:00:00Z"), "application/rss+xml")
+
+    def prices(**kwargs):
+        captured.update(kwargs)
+        return {"as_of": end_date or "2026-09-05", "universe": ["SPY"], "web_research": kwargs["web_research_context"]}
+
+    monkeypatch.setattr(adapter_module, "urlopen", fetch)
+    monkeypatch.setattr(cli, "build_context_from_source", prices)
+    monkeypatch.setattr(cli, "write_context_bundle", lambda bundle, path: written.append(bundle))
+    argv = [str(script), "--symbols", "SPY", "--no-theme-context", "--web-research-sources", str(_config(tmp_path))]
+    if end_date:
+        argv.extend(["--end-date", end_date])
+    monkeypatch.setattr(sys, "argv", argv)
+    assert cli.main() == 0
+    web = captured["web_research_context"]
+    assert len(calls) == int(current)
+    assert web["source_count"] == int(current)
+    assert captured["generated_at"] == clock.observed[0]
+    assert len(written) == 1
+    if current:
+        assert dt.datetime.fromisoformat(web["research_sources"][0]["fetched_at"].replace("Z", "+00:00")) > captured["generated_at"]
+    else:
+        assert web["warnings"]
+        assert web["pit_timestamp"].startswith("2026-09-04")
+
+
+def test_explicit_future_cutoff_does_not_replace_actual_fetch_time(tmp_path, monkeypatch):
+    monkeypatch.setattr(adapter_module, "urlopen", lambda *args, **kwargs: _FakeResponse(
+        _rss("2026-09-04T00:00:00Z"), "application/rss+xml"))
+    cutoff = dt.datetime(2026, 9, 6, tzinfo=dt.timezone.utc)
+    context = ResearchContextAdapter(_config(tmp_path)).build_context(pit_timestamp=cutoff)
+    assert context["source_count"] == 1
+    assert context["pit_timestamp"] == "2026-09-06T00:00:00Z"
+    assert context["research_sources"][0]["fetched_at"].startswith("2026-09-05T12:00:00.")
+
+
+def test_response_finishing_after_explicit_cutoff_is_not_backdated(tmp_path, monkeypatch, clock):
+    class DelayedResponse(_FakeResponse):
+        def read(self):
+            clock.current += dt.timedelta(seconds=2)
+            return super().read()
+
+    monkeypatch.setattr(adapter_module, "urlopen", lambda *args, **kwargs: DelayedResponse(
+        _rss("2026-09-04T00:00:00Z"), "application/rss+xml"))
+    cutoff = clock.current + dt.timedelta(seconds=1)
+    context = ResearchContextAdapter(_config(tmp_path)).build_context(pit_timestamp=cutoff)
+    assert context["source_count"] == 0
+    assert context["pit_timestamp"] == "2026-09-05T12:00:01Z"
+    assert any("fetched after" in warning for warning in context["warnings"])