diff --git a/docs/execution_evidence_runtime_projection.md b/docs/execution_evidence_runtime_projection.md new file mode 100644 index 0000000..0418af0 --- /dev/null +++ b/docs/execution_evidence_runtime_projection.md @@ -0,0 +1,33 @@ +# Runtime report to execution-evidence projection + +`python/scripts/execution_evidence_projection.py` builds the input contract for +the Strategy Switch Console's read-only execution-evidence board. It is shared +across all platform runtimes; it has no broker adapter, trading API, account, +order, position, capital, credential, or network-publishing capability. + +Only an eligible `runtime_report.v1` record is projected. Eligibility requires: + +- an allowed platform and strategy domain; +- a self-attested `runtime_release_receipt` with a 40-character strategy + revision; +- an internally consistent target lane (`paper` or `live`) and `dry_run` + value; and +- a valid report timestamp. + +The projection deliberately copies none of `summary`, `diagnostics`, +`artifacts`, or runtime error text. Its deployment identifier is a stable +digest, so service and account-like labels are not exposed in the console. + +An eligible report proves only that this runtime loaded the displayed strategy +revision for the configured lane. It does **not** prove market-data quality, +broker acceptance, submitted orders, fills, positions, or capital usage. For +that reason every projected record sets `target_data` and `target_execution` to +`pending` and uses `parked` with +`target_execution_evidence_missing`. The projection never emits an autonomous +paper/shadow recommendation or a live approval. + +The script writes a local JSON file only. A future scheduled publisher must use +a distinct `EXECUTION_EVIDENCE_SYNC_TOKEN` and a least-privilege read identity +for the selected runtime-report prefix, then POST that file to +`/api/internal/sync-execution-evidence-source`. Those credentials must remain +in protected secret stores and must never be added to this repository. diff --git a/python/scripts/execution_evidence_projection.py b/python/scripts/execution_evidence_projection.py new file mode 100644 index 0000000..1892236 --- /dev/null +++ b/python/scripts/execution_evidence_projection.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python3 +"""Build a fail-closed, read-only execution-evidence source snapshot. + +The input is one or more ``runtime_report.v1`` documents already exported by +platform runtimes. This module deliberately does not fetch reports, accept +credentials, call a broker, or publish a network request. It only projects +the narrow, non-sensitive identity fields that the Strategy Switch Console +requires for its read-only execution-evidence board. + +An eligible report can attest that a runtime loaded a specific strategy +revision and was configured for a paper or live lane. It cannot attest data +quality, broker acceptance, fills, positions, or capital. Those fields stay +``pending`` and the resulting recommendation is always ``parked``. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +from collections.abc import Iterable, Mapping +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + + +SOURCE_SCHEMA_VERSION = "qsl_execution_evidence_source_snapshot.v1" +RUNTIME_REPORT_SCHEMA_VERSION = "runtime_report.v1" +_PLATFORM_ALIASES = { + "alpaca": "alpaca", + "binance": "binance", + "charles-schwab": "schwab", + "charles_schwab": "schwab", + "firstrade": "firstrade", + "ibkr": "ibkr", + "interactive-brokers": "ibkr", + "interactive_brokers": "ibkr", + "longbridge": "longbridge", + "qmt": "qmt", + "schwab": "schwab", +} +_DOMAINS = frozenset({"us_equity", "hk_equity", "cn_equity", "crypto"}) +_REVISION = re.compile(r"^[0-9a-f]{40}$") +_IDENTIFIER = re.compile(r"^[A-Za-z0-9._=-]{1,128}$") +_FORBIDDEN_TEXT = re.compile(r"(?:secret|token|password|credential|api[_-]?key|account|order|fill|position|capital)", re.IGNORECASE) + + +class ExecutionEvidenceProjectionError(ValueError): + """Raised when a projection request is malformed.""" + + +def build_execution_evidence_source_snapshot( + reports: Iterable[Mapping[str, Any]], + *, + source_id: str, + now: datetime | None = None, +) -> dict[str, Any]: + """Project eligible runtime reports into the Worker source schema. + + Invalid or legacy reports are represented only by bounded error codes. No + source value from a report's diagnostics, summaries, artifacts, or errors + is copied to the output. + """ + normalized_source_id = _identity(source_id, "source_id") + computed_at = _timestamp(now or datetime.now(UTC)) + latest_by_deployment: dict[str, tuple[datetime, dict[str, Any]]] = {} + errors: set[str] = set() + + for report in reports: + try: + deployment, observed_at = _project_runtime_report(report) + except ExecutionEvidenceProjectionError as exc: + errors.add(str(exc)) + continue + previous = latest_by_deployment.get(deployment["deployment_id"]) + if previous is None or observed_at > previous[0]: + latest_by_deployment[deployment["deployment_id"]] = (observed_at, deployment) + + deployments = [entry[1] for _, entry in sorted(latest_by_deployment.items())] + if not deployments: + errors.add("runtime_report_no_eligible_records") + return { + "schema_version": SOURCE_SCHEMA_VERSION, + "source_id": normalized_source_id, + "generated_at": computed_at, + "computed_at": computed_at, + "data_status": "ready" if deployments else "unavailable", + "deployments": deployments, + "errors": sorted(errors)[:20], + } + + +def load_runtime_report(path: str | Path) -> dict[str, Any]: + """Load one report while rejecting duplicate JSON keys.""" + try: + value = json.loads(Path(path).read_text(encoding="utf-8"), object_pairs_hook=_reject_duplicate_keys) + except (OSError, ValueError, json.JSONDecodeError) as exc: + raise ExecutionEvidenceProjectionError("runtime_report_input_invalid") from exc + if not isinstance(value, dict): + raise ExecutionEvidenceProjectionError("runtime_report_input_invalid") + return value + + +def _project_runtime_report(report: Mapping[str, Any]) -> tuple[dict[str, Any], datetime]: + if not isinstance(report, Mapping) or report.get("schema_version") != RUNTIME_REPORT_SCHEMA_VERSION: + raise ExecutionEvidenceProjectionError("runtime_report_schema_unsupported") + + platform = _PLATFORM_ALIASES.get(str(report.get("platform") or "").strip().lower()) + profile = _identity(report.get("strategy_profile"), "runtime_report_strategy_invalid") + domain = str(report.get("strategy_domain") or "").strip() + if platform is None or domain not in _DOMAINS: + raise ExecutionEvidenceProjectionError("runtime_report_target_invalid") + + runtime_target = _mapping(report.get("runtime_target"), "runtime_report_target_invalid") + execution_mode = str(runtime_target.get("execution_mode") or "").strip() + dry_run_only = runtime_target.get("dry_run_only") + if execution_mode not in {"paper", "live"} or not isinstance(dry_run_only, bool): + raise ExecutionEvidenceProjectionError("runtime_report_target_invalid") + report_dry_run = report.get("dry_run") + if not isinstance(report_dry_run, bool): + raise ExecutionEvidenceProjectionError("runtime_report_target_invalid") + if report_dry_run != dry_run_only or (execution_mode == "paper") != dry_run_only: + raise ExecutionEvidenceProjectionError("runtime_report_lane_mismatch") + + receipt = _mapping(report.get("runtime_release_receipt"), "runtime_report_release_unattested") + release = _mapping(receipt.get("strategy_release"), "runtime_report_release_unattested") + revision = release.get("strategy_revision") + if receipt.get("attestation_state") != "self_attested" or not isinstance(revision, str) or not _REVISION.fullmatch(revision): + raise ExecutionEvidenceProjectionError("runtime_report_release_unattested") + + observed_at = _report_timestamp(report) + deployment_id = _deployment_id( + platform=platform, + deploy_target=report.get("deploy_target"), + service_name=report.get("service_name"), + strategy_profile=profile, + environment=execution_mode, + ) + return { + "deployment_id": deployment_id, + "strategy": { + "candidate_id": profile, + "candidate_kind": "individual", + "domain": domain, + "strategy_revision": revision, + }, + "target": {"platform": platform, "environment": execution_mode}, + "capabilities": {"shadow": "unknown", "paper": "unknown"}, + "evidence": { + "strategy": "verified", + "target_data": "pending", + "target_execution": "pending", + }, + "recommendation": { + "code": "parked", + "reason_code": "target_execution_evidence_missing", + }, + }, observed_at + + +def _mapping(value: object, error_code: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise ExecutionEvidenceProjectionError(error_code) + return value + + +def _identity(value: object, error_code: str) -> str: + text = str(value or "").strip() + if not _IDENTIFIER.fullmatch(text) or _FORBIDDEN_TEXT.search(text): + raise ExecutionEvidenceProjectionError(error_code) + return text + + +def _report_timestamp(report: Mapping[str, Any]) -> datetime: + for key in ("finished_at", "started_at"): + value = report.get(key) + if not isinstance(value, str) or not value.strip(): + continue + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + continue + if parsed.tzinfo is not None and parsed.utcoffset() is not None: + return parsed.astimezone(UTC) + raise ExecutionEvidenceProjectionError("runtime_report_timestamp_invalid") + + +def _deployment_id( + *, + platform: str, + deploy_target: object, + service_name: object, + strategy_profile: str, + environment: str, +) -> str: + material = "\x1f".join( + [ + platform, + _safe_label(deploy_target), + _safe_label(service_name), + strategy_profile, + environment, + ] + ) + # A digest is stable across reports and avoids exposing service/account-like + # labels in the console-facing deployment identity. + return f"runtime.{platform}.{hashlib.sha256(material.encode('utf-8')).hexdigest()[:24]}" + + +def _safe_label(value: object) -> str: + text = str(value or "").strip() + if not text or len(text) > 256 or _FORBIDDEN_TEXT.search(text): + return "unavailable" + return text + + +def _timestamp(value: datetime) -> str: + return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate JSON key") + result[key] = value + return result + + +def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source-id", required=True, help="stable non-sensitive source identity") + parser.add_argument("--runtime-report", action="append", default=[], help="path to one runtime_report.v1 JSON document") + parser.add_argument("--output", required=True, help="path for the generated source snapshot") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(argv) + reports: list[Mapping[str, Any]] = [] + input_errors: list[str] = [] + for path in args.runtime_report: + try: + reports.append(load_runtime_report(path)) + except ExecutionEvidenceProjectionError as exc: + input_errors.append(str(exc)) + snapshot = build_execution_evidence_source_snapshot(reports, source_id=args.source_id) + snapshot["errors"] = sorted(set([*snapshot["errors"], *input_errors]))[:20] + output_path = Path(args.output) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python/tests/test_execution_evidence_projection.py b/python/tests/test_execution_evidence_projection.py new file mode 100644 index 0000000..992fc36 --- /dev/null +++ b/python/tests/test_execution_evidence_projection.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +import tempfile +import unittest +from datetime import UTC, datetime +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" + + +def _load_module(name: str): + spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +projection = _load_module("execution_evidence_projection") + + +class ExecutionEvidenceProjectionTest(unittest.TestCase): + def _report(self, *, finished_at: str = "2026-08-25T16:00:00Z") -> dict[str, object]: + return { + "schema_version": "runtime_report.v1", + "platform": "longbridge", + "deploy_target": "sg", + "service_name": "longbridge-quant-sg-service", + "strategy_profile": "soxl_soxx_trend_income", + "strategy_domain": "us_equity", + "runtime_target": {"execution_mode": "paper", "dry_run_only": True}, + "runtime_release_receipt": { + "attestation_state": "self_attested", + "strategy_release": {"strategy_revision": "a" * 40}, + }, + "dry_run": True, + "started_at": "2026-08-25T15:58:00Z", + "finished_at": finished_at, + "summary": {"account_ids": ["must-not-be-projected"]}, + "diagnostics": {"api_token": "must-not-be-projected"}, + "artifacts": {"runtime_report_cloud_uri": "gs://must-not-be-projected"}, + } + + def test_projects_only_attested_identity_and_keeps_execution_pending(self): + snapshot = projection.build_execution_evidence_source_snapshot( + [self._report()], + source_id="runtime-reports", + now=datetime(2026, 8, 25, 16, 5, tzinfo=UTC), + ) + self.assertEqual(snapshot["schema_version"], "qsl_execution_evidence_source_snapshot.v1") + self.assertEqual(snapshot["data_status"], "ready") + self.assertEqual(snapshot["generated_at"], "2026-08-25T16:05:00Z") + self.assertEqual(len(snapshot["deployments"]), 1) + deployment = snapshot["deployments"][0] + self.assertEqual(deployment["target"], {"platform": "longbridge", "environment": "paper"}) + self.assertEqual(deployment["evidence"], { + "strategy": "verified", + "target_data": "pending", + "target_execution": "pending", + }) + self.assertEqual(deployment["recommendation"], { + "code": "parked", + "reason_code": "target_execution_evidence_missing", + }) + serialized = json.dumps(snapshot, sort_keys=True) + for forbidden in ("must-not-be-projected", "account_ids", "api_token", "gs://"): + self.assertNotIn(forbidden, serialized) + + def test_rejects_unattested_or_lane_mismatched_reports_without_claiming_execution(self): + unattested = self._report() + unattested["runtime_release_receipt"] = {"attestation_state": "legacy_unattested"} + mismatched = self._report() + mismatched["dry_run"] = False + snapshot = projection.build_execution_evidence_source_snapshot( + [unattested, mismatched], + source_id="runtime-reports", + now=datetime(2026, 8, 25, 16, 5, tzinfo=UTC), + ) + self.assertEqual(snapshot["data_status"], "unavailable") + self.assertEqual(snapshot["deployments"], []) + self.assertEqual(snapshot["errors"], [ + "runtime_report_lane_mismatch", + "runtime_report_no_eligible_records", + "runtime_report_release_unattested", + ]) + + def test_keeps_only_the_latest_report_per_deployment(self): + older = self._report(finished_at="2026-08-25T15:00:00Z") + latest = self._report(finished_at="2026-08-25T16:00:00Z") + latest["runtime_release_receipt"] = { + "attestation_state": "self_attested", + "strategy_release": {"strategy_revision": "b" * 40}, + } + snapshot = projection.build_execution_evidence_source_snapshot( + [older, latest], + source_id="runtime-reports", + now=datetime(2026, 8, 25, 16, 5, tzinfo=UTC), + ) + self.assertEqual(len(snapshot["deployments"]), 1) + self.assertEqual(snapshot["deployments"][0]["strategy"]["strategy_revision"], "b" * 40) + + def test_cli_rejects_duplicate_json_keys_and_writes_a_fail_closed_snapshot(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + invalid = root / "invalid.json" + output = root / "snapshot.json" + invalid.write_text('{"schema_version":"runtime_report.v1","schema_version":"other"}', encoding="utf-8") + self.assertEqual( + projection.main([ + "--source-id", "runtime-reports", + "--runtime-report", str(invalid), + "--output", str(output), + ]), + 0, + ) + snapshot = json.loads(output.read_text(encoding="utf-8")) + self.assertEqual(snapshot["data_status"], "unavailable") + self.assertIn("runtime_report_input_invalid", snapshot["errors"]) + + +if __name__ == "__main__": + unittest.main()