feat: Add safe daily attribution diagnostic counts

docs/unified-daily-notification.md

@@ -202,3 +202,22 @@ Private state object and submission-claim locators retain their existing identit
 format so a display change cannot bypass an existing duplicate-submission lock
 or session-maintenance marker. New state payload account fields use the two-character
 mask. State-store errors omit object locators and provider exception details.
+
+
+### Daily attribution diagnostics
+
+The daily runtime publisher's local JSON output includes `diagnostics` after
+archive projection: `files_scanned`, `files_eligible`, `files_read`,
+`files_attributable`, and `excluded_by_reason`. Reasons are fixed labels:
+`identity_unverified`, `invalid_report`, `invalid_run_time`,
+`other_business_date`, and `duplicate_run`. Failed or truncated reads leave
+attribution and exclusion totals null; successful empty listings produce zeros.
+`archive_outcome` distinguishes success, listing failure, report read failure,
+and truncation. These diagnostics do not change the receiver payload or certify
+archive completeness, identity, execution, or reconciliation.
+
+Candidate archive failures retain `candidate_archive_read_failed` and expose only
+allowlisted operation (`initialize`, `list`, `read`, `unknown`) and category
+(`permission_denied`, `not_found`, `timeout`, `invalid_json`, `integrity_failed`,
+`unavailable`, `unknown`). Provider messages, source locators, bindings and report
+bodies are excluded. Failures still exit nonzero; no retry or sending gate changes.

notifications/digest_snapshot/digest_archive.py

@@ -53,12 +53,29 @@ def read(self, args):
 
 
 def _read(args):
-    from google.cloud import storage
+    operation = "initialize"
     try:
-        return ArchiveReader(storage.Client()).read(args)
+        from google.cloud import storage
+        client = storage.Client()
+        operation = {"ls": "list", "cat": "read"}.get(args[0], "unknown")
+        return ArchiveReader(client).read(args)
     except Exception as error:
-        # Empty SDK listings return []; all failed reads remain failures.
-        raise ValueError("candidate_archive_read_failed") from error
+        from google.api_core import exceptions
+        category = "unknown"
+        if isinstance(error, (exceptions.Forbidden, exceptions.Unauthorized)):
+            category = "permission_denied"
+        elif isinstance(error, exceptions.NotFound):
+            category = "not_found"
+        elif isinstance(error, (TimeoutError, exceptions.DeadlineExceeded)):
+            category = "timeout"
+        elif isinstance(error, (json.JSONDecodeError, UnicodeError)):
+            category = "invalid_json"
+        elif isinstance(error, exceptions.PreconditionFailed):
+            category = "integrity_failed"
+        elif isinstance(error, exceptions.ServiceUnavailable):
+            category = "unavailable"
+        # Only fixed labels cross this boundary, never provider text or locators.
+        raise ValueError(f"candidate_archive_read_failed:{operation}:{category}") from None
 
 
 def emit(env, *, now=None, read=_read, upload=None):

scripts/emit_unified_digest_candidates.py

@@ -50,6 +50,14 @@ def _failure_diagnostic(error, stage):
         code = category if type(category) is str and category in _STORAGE_CATEGORIES else "storage_unknown"
         return {"stage": stage, "error_code": code, "error_category": code}
     code = error.args[0] if len(error.args) == 1 and type(error.args[0]) is str else None
+    if isinstance(error, ValueError) and type(code) is str:
+        parts = code.split(":")
+        if (len(parts) == 3 and parts[0] == "candidate_archive_read_failed"
+                and parts[1] in {"initialize", "list", "read", "unknown"}
+                and parts[2] in {"permission_denied", "not_found", "timeout", "invalid_json",
+                                 "integrity_failed", "unavailable", "unknown"}):
+            return {"stage": "archive_read", "error_code": parts[0],
+                    "operation": parts[1], "error_category": parts[2]}
     if isinstance(error, (ValueError, RuntimeError)) and code in _ERROR_STAGES:
         return {"stage": _ERROR_STAGES[code], "error_code": code}
     if isinstance(error, json.JSONDecodeError):

scripts/publish_runtime_daily.py

@@ -31,18 +31,33 @@ def _binding(env, target):
     return account, binding, binding
 
 
+class _ReadResult(tuple):
+    """Keep the reader's two-value interface with local diagnostic metadata."""
+
+    def __new__(cls, reports, error, diagnostics):
+        result = super().__new__(cls, (reports, error))
+        result.diagnostics = diagnostics
+        return result
+
+
 def _read_reports(day, now, project):
     """Use the existing report reader; never expose a URI or report body."""
     start = datetime.combine(day, time.min, tzinfo=ZoneInfo("America/New_York")).astimezone(timezone.utc)
     end = min(start + timedelta(days=2), now)
+    diagnostics = dict(files_scanned=None, files_eligible=None, files_read=None,
+                       archive_outcome="list_failed")
+    def result(reports, error):
+        diagnostics["archive_outcome"] = error or "success"
+        return _ReadResult(reports, error, diagnostics)
     try:
         entries = []
         for glob in heartbeat._report_globs(start, end):
             entries.extend(heartbeat._list_gcs_objects(glob, project=project))
         eligible = [entry for entry in entries if heartbeat._object_updated_at(entry) is not None
                     and start - timedelta(days=1) <= heartbeat._object_updated_at(entry) <= end + timedelta(days=1)]
+        diagnostics.update(files_scanned=len(entries), files_eligible=len(eligible))
         if len(eligible) > 20:
-            return [], "read_truncated"
+            return result([], "read_truncated")
         reports = []
         failed = False
         for entry in eligible:
@@ -53,9 +68,10 @@ def _read_reports(day, now, project):
                 failed = True
             else:
                 reports.append(report)
-        return reports, "report_read_error" if failed else None
+        diagnostics["files_read"] = len(reports)
+        return result(reports, "report_read_error" if failed else None)
     except Exception:
-        return [], "list_failed"
+        return result([], "list_failed")
 
 
 def _auth_config(env):
@@ -93,7 +109,7 @@ def auth_preflight(env, *, opener=None):
         return "unavailable"
 
 
-def publish_once(env, *, now=None, opener=None):
+def publish_once(env, *, now=None, opener=None, diagnostics=None):
     """Make one POST; an unconfirmed response is never retried."""
     now = now or datetime.now(timezone.utc)
     target = json.loads(env.get("RUNTIME_TARGET_JSON") or "null")
@@ -113,8 +129,19 @@ def publish_once(env, *, now=None, opener=None):
     kwargs = dict(target_key=target_key, account_binding=account_binding,
                   business_date=day.isoformat(), observed_at=now)
     project_runtime_daily([], **kwargs)
-    reports, error = _read_reports(day, now, env.get("GCP_PROJECT_ID"))
+    read_result = _read_reports(day, now, env.get("GCP_PROJECT_ID"))
+    reports, error = read_result
     projection = project_runtime_daily(reports, **kwargs)
+    if diagnostics is not None:
+        diagnostics.update(getattr(read_result, "diagnostics", {}))
+        record = projection["records"][0]
+        reasons = {key: 0 for key in ("identity_unverified", "invalid_report",
+                                     "invalid_run_time", "other_business_date", "duplicate_run")}
+        for item in projection["unmatched_reports"] + record["excluded_reports"]:
+            reasons[item["reason"]] += 1
+        reasons["duplicate_run"] = len(reports) - len(record["runs"]) - sum(reasons.values())
+        diagnostics.update(files_attributable=None if error else len(record["runs"]),
+                           excluded_by_reason={key: None for key in reasons} if error else reasons)
     if error:
         projection["read_errors"].append(error)
     if not projection["records"][0]["runs"]:
@@ -146,16 +173,17 @@ def publish_once(env, *, now=None, opener=None):
 
 
 def main():
+    diagnostics = {}
     try:
         if sys.argv[1:] == ["--preflight"]:
             status = auth_preflight(os.environ)
         elif not sys.argv[1:]:
-            status = publish_once(os.environ)
+            status = publish_once(os.environ, diagnostics=diagnostics)
         else:
             raise ValueError("invalid_mode")
     except Exception:
         status = "unavailable"
-    print(json.dumps({"status": status}, separators=(",", ":")))
+    print(json.dumps({"status": status, **({"diagnostics": diagnostics} if diagnostics else {})}, separators=(",", ":")))
     return 0 if status in {"published", "verified"} else 1
 
 

tests/test_attribution_diagnostics.py

@@ -0,0 +1,90 @@
+from datetime import datetime, timezone
+import json
+
+import pytest
+
+from scripts import publish_runtime_daily as daily
+from notifications.digest_snapshot import digest_archive as archive
+
+
+def test_empty_listing_has_measured_zero_counts(monkeypatch):
+    monkeypatch.setattr(daily.heartbeat, '_report_globs', lambda *_: ['fake'])
+    monkeypatch.setattr(daily.heartbeat, '_list_gcs_objects', lambda *_, **__: [])
+    now = datetime(2026, 11, 4, tzinfo=timezone.utc)
+    result = daily._read_reports(now.date(), now, None)
+    assert result.diagnostics == {'files_scanned': 0, 'files_eligible': 0, 'files_read': 0,
+                                  'archive_outcome': 'success'}
+
+
+def test_failed_listing_keeps_counts_unknown(monkeypatch):
+    monkeypatch.setattr(daily.heartbeat, '_report_globs', lambda *_: ['fake'])
+    def fail(*_, **__):
+        raise RuntimeError('fake-private-provider-message')
+    monkeypatch.setattr(daily.heartbeat, '_list_gcs_objects', fail)
+    now = datetime(2026, 11, 4, tzinfo=timezone.utc)
+    result = daily._read_reports(now.date(), now, None)
+    assert result[1] == 'list_failed'
+    assert result.diagnostics['files_scanned'] is None
+    assert 'fake-private' not in json.dumps(result.diagnostics)
+
+
+@pytest.mark.parametrize('operation', ['ls', 'cat'])
+def test_archive_failure_has_fixed_operation_category(monkeypatch, operation):
+    from google.cloud import storage
+    from google.api_core.exceptions import Forbidden
+    def fail(*_, **__):
+        raise Forbidden('fake-private-provider-message')
+    monkeypatch.setattr(storage, 'Client', fail)
+    with pytest.raises(ValueError) as caught:
+        archive._read([operation, 'gs://fake-bucket/fake-key'])
+    assert str(caught.value) == 'candidate_archive_read_failed:initialize:permission_denied'
+    assert caught.value.__suppress_context__
+
+
+def test_cli_reports_exclusion_counts_without_private_values(monkeypatch, capsys):
+    monkeypatch.setattr(daily, '_FACTS_URL', 'https://example.test/api/account-facts/sync')
+    monkeypatch.setattr(daily, '_binding', lambda *_: ('fake-account', 'a' * 64, 'b' * 64))
+    env = {'RUNTIME_TARGET_JSON': json.dumps({'service_name': 'fake-service',
+           'strategy_profile': 'fake-profile', 'account_scope': 'paper'}),
+           'EXECUTION_EVIDENCE_SYNC_TOKEN': 'fake-token',
+           'EXECUTION_EVIDENCE_SYNC_URL': 'https://example.test/api/execution-evidence/sync',
+           'RUNTIME_DAILY_BUSINESS_DATE': '2026-11-03'}
+    base = {'platform': 'firstrade', 'service_name': 'fake-service',
+            'strategy_profile': 'fake-profile', 'account_scope': 'paper',
+            'account_facts_source_binding_id': 'a' * 64,
+            'started_at': '2026-11-03T18:00:00Z', 'finished_at': '2026-11-03T18:01:00Z',
+            'run_id': 'fake-private-run', 'errors': []}
+    reports = [{}, {**base, 'started_at': 'bad'},
+               {**base, 'started_at': '2026-11-02T18:00:00Z'}, {**base, 'errors': None}]
+    monkeypatch.setattr(daily, '_read_reports', lambda *_: daily._ReadResult(
+        reports, None, dict(files_scanned=5, files_eligible=4, files_read=4, archive_outcome='success')))
+    diagnostics = {}
+    assert daily.publish_once(env, now=datetime(2026, 11, 4, tzinfo=timezone.utc),
+                              diagnostics=diagnostics) == 'no_attributable_runs'
+    assert diagnostics['files_attributable'] == 0
+    assert diagnostics['excluded_by_reason'] == dict(identity_unverified=1, invalid_report=1,
+        invalid_run_time=1, other_business_date=1, duplicate_run=0)
+    assert 'fake-private' not in json.dumps(diagnostics)
+    assert capsys.readouterr().out == ''
+
+
+@pytest.mark.parametrize(('error', 'category'), [(TimeoutError('fake-private'), 'timeout'),
+    (ValueError('fake-private'), 'unknown')])
+def test_archive_read_failure_never_exposes_provider_text(monkeypatch, error, category):
+    from google.cloud import storage
+    monkeypatch.setattr(storage, 'Client', lambda: object())
+    def fail(*_):
+        raise error
+    monkeypatch.setattr(archive.ArchiveReader, 'read', fail)
+    with pytest.raises(ValueError) as caught:
+        archive._read(['cat', 'gs://fake-bucket/fake-key'])
+    assert str(caught.value) == f'candidate_archive_read_failed:read:{category}'
+
+
+def test_archive_diagnostic_public_boundary_is_allowlisted():
+    from scripts.emit_unified_digest_candidates import _failure_diagnostic
+    assert _failure_diagnostic(ValueError('candidate_archive_read_failed:read:timeout'), 'projection') == {
+        'stage': 'archive_read', 'error_code': 'candidate_archive_read_failed',
+        'operation': 'read', 'error_category': 'timeout'}
+    assert _failure_diagnostic(ValueError('candidate_archive_read_failed:read:fake-private'),
+                               'projection') == {'stage': 'projection', 'error_code': 'unexpected_error'}

tests/test_project_runtime_daily.py

@@ -146,7 +146,10 @@ def test_unreadable_empty_source_keeps_safe_failure_category(monkeypatch, capsys
     ):
         monkeypatch.setattr(publisher, "_read_reports", lambda *_: ([], error))
         assert publisher.main() == 1
-        assert json.loads(capsys.readouterr().out) == {"status": expected}
+        output = json.loads(capsys.readouterr().out)
+        assert output["status"] == expected
+        assert output["diagnostics"]["files_attributable"] is None
+        assert all(value is None for value in output["diagnostics"]["excluded_by_reason"].values())
 
 
 def test_archive_listing_exception_is_redacted_and_preserved(monkeypatch, capsys):