diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index bbd9c65..c8e7539 100644 --- a/docs/unified-daily-notification.md +++ b/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. diff --git a/notifications/digest_snapshot/digest_archive.py b/notifications/digest_snapshot/digest_archive.py index 2665e01..2bec936 100644 --- a/notifications/digest_snapshot/digest_archive.py +++ b/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): diff --git a/scripts/emit_unified_digest_candidates.py b/scripts/emit_unified_digest_candidates.py index b1a0009..fdc6552 100644 --- a/scripts/emit_unified_digest_candidates.py +++ b/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): diff --git a/scripts/publish_runtime_daily.py b/scripts/publish_runtime_daily.py index ea336a6..5347338 100644 --- a/scripts/publish_runtime_daily.py +++ b/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 diff --git a/tests/test_attribution_diagnostics.py b/tests/test_attribution_diagnostics.py new file mode 100644 index 0000000..f5419e3 --- /dev/null +++ b/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'} diff --git a/tests/test_project_runtime_daily.py b/tests/test_project_runtime_daily.py index cd4ec55..8bab974 100644 --- a/tests/test_project_runtime_daily.py +++ b/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):