fix: Handle empty archive months in heartbeat publishing

docs/unified-daily-notification.md

@@ -115,6 +115,12 @@ or access Secret Manager. Archive reads use the Storage SDK and the job's
 attached identity; the image does not need gcloud. Existing candidate writes,
 merge semantics and identity checks remain authoritative.
 
+The heartbeat candidate publisher uses the same bounded SDK archive reader.
+A successfully listed empty month is empty input, including when a completed
+window spans two UTC months. Permission, timeout and JSON failures still abort
+publication. An empty window produces idle evidence only after the existing
+schedule check confirms that no run was due.
+
 Use one job per existing central target, with its exact protected runtime JSON,
 archive prefix, opaque UID, target ID, destination and slot. No identifiers can
 be inferred from a display label. Supply the existing notification schedule so

notifications/digest_snapshot/digest_archive.py

@@ -8,7 +8,6 @@
 import json
 import os
 import re
-import subprocess
 from datetime import datetime, timedelta, timezone
 from zoneinfo import ZoneInfo
 
@@ -17,11 +16,49 @@
 _SAFE = re.compile(r"[a-z0-9][a-z0-9_-]{0,40}\Z")
 
 
+class ArchiveReader:
+    """Native Storage reads with the job identity; no gcloud or secret access."""
+
+    def __init__(self, client):
+        self.client = client
+
+    def read(self, args):
+        uri = args[-1]
+        if not uri.startswith("gs://"):
+            raise ValueError("candidate_archive_uri_invalid")
+        bucket, separator, key = uri[5:].partition("/")
+        if not bucket or not separator or not key:
+            raise ValueError("candidate_archive_uri_invalid")
+        if args[0] == "ls":
+            blobs = self.client.list_blobs(
+                bucket, prefix=key.split("*", 1)[0], match_glob=key,
+                max_results=10001, retry=None, timeout=30,
+            )
+            rows = []
+            for blob in blobs:
+                rows.append({"url": f"gs://{bucket}/{blob.name}",
+                             "metadata": {"updated": blob.updated.isoformat()}})
+                if len(rows) > 10000:
+                    raise ValueError("candidate_archive_budget_exceeded")
+            return rows
+        if args[0] != "cat":
+            raise ValueError("candidate_archive_operation_invalid")
+        blob = self.client.bucket(bucket).blob(key)
+        blob.reload(retry=None, timeout=30)
+        if blob.size is None or blob.size > 2 * 1024**2:
+            raise ValueError("candidate_archive_object_too_large")
+        return json.loads(blob.download_as_bytes(
+            if_generation_match=int(blob.generation), retry=None, timeout=30,
+        ))
+
+
 def _read(args):
-    result = subprocess.run(["gcloud", "storage", *args], capture_output=True, text=True, check=False)
-    if result.returncode:
-        raise ValueError("candidate_archive_read_failed")
-    return json.loads(result.stdout)
+    from google.cloud import storage
+    try:
+        return ArchiveReader(storage.Client()).read(args)
+    except Exception as error:
+        # Empty SDK listings return []; all failed reads remain failures.
+        raise ValueError("candidate_archive_read_failed") from error
 
 
 def emit(env, *, now=None, read=_read, upload=None):

scripts/publish_digest_before_read.py

@@ -8,6 +8,8 @@
 from datetime import datetime, time, timedelta, timezone
 from zoneinfo import ZoneInfo
 
+from notifications.digest_snapshot.digest_archive import ArchiveReader
+
 SHANGHAI = ZoneInfo("Asia/Shanghai")
 PLATFORM = "firstrade"
 
@@ -25,42 +27,6 @@ def publication_day(now):
     return (local.date() - timedelta(days=1)).isoformat()
 
 
-class ArchiveReader:
-    """Native Storage reads with the job identity; no gcloud or secret access."""
-
-    def __init__(self, client):
-        self.client = client
-
-    def read(self, args):
-        uri = args[-1]
-        if not uri.startswith("gs://"):
-            raise ValueError("candidate_archive_uri_invalid")
-        bucket, separator, key = uri[5:].partition("/")
-        if not bucket or not separator or not key:
-            raise ValueError("candidate_archive_uri_invalid")
-        if args[0] == "ls":
-            blobs = self.client.list_blobs(
-                bucket, prefix=key.split("*", 1)[0], match_glob=key,
-                max_results=10001, retry=None, timeout=30,
-            )
-            rows = []
-            for blob in blobs:
-                rows.append({"url": f"gs://{bucket}/{blob.name}",
-                             "metadata": {"updated": blob.updated.isoformat()}})
-                if len(rows) > 10000:
-                    raise ValueError("candidate_archive_budget_exceeded")
-            return rows
-        if args[0] != "cat":
-            raise ValueError("candidate_archive_operation_invalid")
-        blob = self.client.bucket(bucket).blob(key)
-        blob.reload(retry=None, timeout=30)
-        if blob.size is None or blob.size > 2 * 1024**2:
-            raise ValueError("candidate_archive_object_too_large")
-        return json.loads(blob.download_as_bytes(
-            if_generation_match=int(blob.generation), retry=None, timeout=30,
-        ))
-
-
 def _emit(env, *, now, read):
     from notifications.unified import emit_candidates
     return emit_candidates(env, now=now, read=read)

tests/test_heartbeat_archive_reader.py

@@ -0,0 +1,103 @@
+"""Heartbeat uses bounded SDK archive reads, including empty month partitions."""
+from datetime import datetime, timezone
+import json
+import subprocess
+
+import pytest
+from google.api_core.exceptions import Forbidden
+
+from notifications import unified
+from test_publish_digest_before_read import ENV, PLATFORM
+import test_publish_digest_before_read as prepub
+
+client = prepub.client
+
+NOW = datetime(2026, 11, 2, 1, 35, tzinfo=timezone.utc)
+SOURCE_PLATFORM = "charles_schwab" if PLATFORM == "schwab" else PLATFORM
+
+
+def emit():
+    return unified.emit_candidates(ENV, now=NOW)
+
+
+@pytest.fixture(autouse=True)
+def no_cli(monkeypatch):
+    def fail(*args, **kwargs):
+        pytest.fail("heartbeat must use the Storage SDK")
+    monkeypatch.setattr(subprocess, "run", fail)
+
+
+def candidate(client):
+    return client.blobs[f"gs://synthetic-candidates/digest/{PLATFORM}/2026-11-01/synthetic-slot.json"]
+
+
+@pytest.mark.parametrize("other_month", [None, "2026-10", "2026-11"])
+def test_cross_month_empty_partition_reaches_not_due_check(client, other_month):
+    if other_month:
+        target = {**json.loads(ENV["RUNTIME_TARGET_JSON"]), "service_name": "synthetic-other"}
+        report = {"runtime_target": target, "started_at": "2026-11-01T20:00:00Z"}
+        client.blob(f"gs://synthetic-archive/reports/{SOURCE_PLATFORM}/synthetic/{other_month}/other.json").upload_from_string(
+            json.dumps(report), retry=None)
+    result = emit()
+    assert result["uploaded"] is True
+    rows = json.loads(candidate(client).body)["runs"]
+    assert len(rows) == 1 and rows[0]["actually_ran"] is False
+    assert rows[0]["scheduled_idle"] is True
+    assert rows[0]["business_day"] == "2026-11-01"
+    assert rows[0].get("order_count") is None and rows[0].get("fill_count") is None
+    assert [kwargs["match_glob"] for _, kwargs in client.lists] == [
+        f"reports/{SOURCE_PLATFORM}/**/2026-10/*.json",
+        f"reports/{SOURCE_PLATFORM}/**/2026-11/*.json",
+    ]
+    assert all(kwargs["retry"] is None and kwargs["timeout"] == 30 for _, kwargs in client.lists)
+
+
+def test_cross_month_empty_due_window_still_fails_without_writing(client, monkeypatch):
+    monkeypatch.setitem(ENV, "DIGEST_NOTIFICATION_SCHEDULE_JSON", json.dumps({"scheduler": {
+        "main_time": "45 15 * * *", "timezone": "America/New_York"}}))
+    with pytest.raises(ValueError, match="candidate_due_source_absent"):
+        emit()
+    assert not any(blob.writes for blob in client.blobs.values())
+
+
+@pytest.mark.parametrize("error", [Forbidden("synthetic-private"), TimeoutError("synthetic-private")])
+@pytest.mark.parametrize("failed_month", ["2026-10", "2026-11"])
+def test_listing_failure_is_never_empty_even_after_an_empty_month(client, monkeypatch, error, failed_month):
+    real_list = client.list_blobs
+    def listing(*args, **kwargs):
+        if failed_month in kwargs["match_glob"]:
+            def failed_page():
+                raise error
+                yield
+            return failed_page()
+        return real_list(*args, **kwargs)
+    monkeypatch.setattr(client, "list_blobs", listing)
+    with pytest.raises(ValueError, match="candidate_archive_read_failed"):
+        emit()
+    assert not any(blob.writes for blob in client.blobs.values())
+
+
+def test_invalid_archive_json_remains_a_read_failure(client):
+    blob = client.blob(f"gs://synthetic-archive/reports/{SOURCE_PLATFORM}/synthetic/2026-10/run.json")
+    blob.upload_from_string('{"synthetic-private":', retry=None)
+    with pytest.raises(ValueError, match="candidate_archive_read_failed"):
+        emit()
+    assert all("/digest/" not in uri for uri in client.blobs)
+
+
+def test_cross_month_report_in_october_is_preserved_when_november_is_empty(client):
+    from notifications.digest_snapshot.digest_candidates import cycle_evidence
+    target = json.loads(ENV["RUNTIME_TARGET_JSON"])
+    report = {"schema_version": "runtime_report.v1", "platform": SOURCE_PLATFORM,
+              "strategy_profile": target["strategy_profile"], "service_name": target["service_name"],
+              "runtime_target": target, "run_id": "synthetic-run", "dry_run": False,
+              "status": "ok", "started_at": "2026-10-31T23:00:00+00:00",
+              "summary": {"notification_digest": cycle_evidence(
+                  {"execution_status": "no_action", "broker_submission_done": False}, [])}}
+    client.blob(f"gs://synthetic-archive/reports/{SOURCE_PLATFORM}/synthetic/2026-10/run.json").upload_from_string(
+        json.dumps(report), retry=None)
+    assert emit()["uploaded"] is True
+    rows = json.loads(candidate(client).body)["runs"]
+    assert len(rows) == 1 and rows[0]["actually_ran"] is True
+    assert rows[0]["source_run_id"] == "synthetic-run"
+    assert rows[0]["business_day"] == "2026-11-01"