feat: project runtime reports as execution evidence

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.

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())