Skip to content

feat: add fail-closed runtime evidence aggregate #143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Pigbibi wants to merge 1 commit into from
Closed

feat: add fail-closed runtime evidence aggregate #143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
212 changes: 212 additions & 0 deletions runtime_support.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import hashlib
import os
import re
import time
from collections.abc import Mapping
from dataclasses import dataclass, field
Expand All @@ -11,6 +12,23 @@
# Binance rate limits (public API: 1200 weight/min, order placement: 50 orders/10s)
_BINANCE_ORDER_RATE_LIMIT_INTERVAL_SEC = 0.25 # max ~4 orders/sec
_LAST_API_CALL_TS: float = 0.0
RUNTIME_EVIDENCE_CONTRACT_VERSION = "qsl.runtime_evidence_aggregate.v1"
RECONCILIATION_STATUSES = frozenset({"MISSING", "MATCHED", "MISMATCHED"})
_RUNTIME_EVIDENCE_FORBIDDEN_FIELDS = frozenset(
{
"api_key",
"api_secret",
"authorization",
"balances",
"credentials",
"headers",
"orders",
"positions",
"provider_rows",
"secret",
"token",
}
)


def _rate_limit_pause():
Expand All @@ -22,6 +40,200 @@ def _rate_limit_pause():
_LAST_API_CALL_TS = time.monotonic()


def _is_sha256(value: Any) -> bool:
return isinstance(value, str) and bool(re.fullmatch(r"[0-9a-f]{64}", value.strip()))
Comment on lines +43 to +44

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Validate the stored digest instead of a stripped copy

When a digest has leading or trailing whitespace, the regex succeeds against value.strip(), but the original malformed value remains in the aggregate. Consumers performing exact identity or reconciliation comparisons can consequently receive a non-digest even though validation reported success; either reject whitespace by matching value directly or normalize the value before storing it.

Useful? React with 👍 / 👎.



def _is_git_revision(value: Any) -> bool:
return isinstance(value, str) and bool(re.fullmatch(r"[0-9a-f]{40}", value.strip()))


def _is_utc_timestamp(value: Any) -> bool:
if not isinstance(value, str) or not value.endswith("Z"):
return False
try:
datetime.fromisoformat(value.replace("Z", "+00:00"))
Comment on lines +51 to +55

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Require a time component in UTC timestamps

When input_timestamp is date-only, such as 2026-03-13Z, replacing Z produces 2026-03-13+00:00, which datetime.fromisoformat() accepts as a naive midnight datetime. The helper therefore accepts a value that identifies neither an explicit time nor an aware UTC instant, allowing ambiguous release provenance; require a time component and verify that the parsed datetime is UTC-aware.

Useful? React with 👍 / 👎.

except ValueError:
return False
return True


def _append_missing_fields(payload: Mapping[str, Any], fields: tuple[str, ...], errors: list[str], label: str) -> None:
for field_name in fields:
if field_name not in payload:
errors.append(f"{label} missing field: {field_name}")


def _append_forbidden_field_errors(value: Any, errors: list[str]) -> None:
if isinstance(value, Mapping):
for field, nested_value in value.items():
if str(field).lower() in _RUNTIME_EVIDENCE_FORBIDDEN_FIELDS:
errors.append(f"runtime_evidence_aggregate contains forbidden field: {field}")
Comment on lines +70 to +71

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Reject credential aliases in redaction checks

When a component mapping contains this repository's tg_token field—or variants such as apiKey or private_key—the exact lowercase lookup does not match token or api_key. Because the component validators also permit unknown keys, build_runtime_evidence_aggregate() returns a supposedly redacted aggregate containing the live secret; use closed per-component field allowlists or normalize and reject credential aliases.

Useful? React with 👍 / 👎.

_append_forbidden_field_errors(nested_value, errors)
elif isinstance(value, (list, tuple)):
for item in value:
_append_forbidden_field_errors(item, errors)


def _validate_release_identity(identity: Any, errors: list[str]) -> None:
label = "runtime_evidence_aggregate release_identity"
if not isinstance(identity, Mapping):
errors.append(f"{label} must be an object")
return
_append_missing_fields(
identity,
(
"strategy_profile",
"mode",
"source_revision",
"input_timestamp",
"artifact_contract",
"artifact_version",
"artifacts",
),
errors,
label,
)
for field_name in ("strategy_profile", "mode", "artifact_contract", "artifact_version"):
if not isinstance(identity.get(field_name), str) or not identity[field_name].strip():
errors.append(f"{label} {field_name} must be a non-empty string")
if not _is_git_revision(identity.get("source_revision")):
errors.append(f"{label} source_revision must be a 40-character lowercase git SHA")
if not _is_utc_timestamp(identity.get("input_timestamp")):
errors.append(f"{label} input_timestamp must be a UTC timestamp")
artifacts = identity.get("artifacts")
if not isinstance(artifacts, Mapping) or not artifacts:
errors.append(f"{label} artifacts must be a non-empty object")
return
for artifact_name, artifact in artifacts.items():
if not isinstance(artifact_name, str) or not artifact_name.strip() or not isinstance(artifact, Mapping):
errors.append(f"{label} artifacts must contain named objects")
continue
if not _is_sha256(artifact.get("sha256")):
errors.append(f"{label} artifacts.{artifact_name}.sha256 must be a SHA-256 digest")


def _validate_reconciliation(reconciliation: Any, errors: list[str]) -> None:
label = "runtime_evidence_aggregate reconciliation"
if not isinstance(reconciliation, Mapping):
errors.append(f"{label} must be an object")
return
status = reconciliation.get("status")
if status not in RECONCILIATION_STATUSES:
errors.append(f"{label} status must be one of MISSING, MATCHED, MISMATCHED")
return
if status == "MATCHED":
Comment on lines +121 to +125

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Reject digest fields on missing reconciliations

When status is MISSING, this function skips all validation of reconciliation digest fields, so an object such as {"status": "MISSING", "durable_receipt_sha256": "invalid"} is accepted as valid evidence. This permits malformed or contradictory receipt provenance to survive validation; reject digest fields for MISSING, or validate every supplied digest and define which ones are allowed for that status.

Useful? React with 👍 / 👎.

for field in ("durable_receipt_sha256", "identity_sha256"):
if not _is_sha256(reconciliation.get(field)):
errors.append(f"{label}.MATCHED requires {field}")
errors.append(f"{label}.MATCHED is not valid for static acceptance")
elif status == "MISMATCHED":
for field in ("durable_receipt_sha256", "identity_sha256", "observed_identity_sha256"):
if not _is_sha256(reconciliation.get(field)):
errors.append(f"{label}.MISMATCHED requires {field}")
if reconciliation.get("identity_sha256") == reconciliation.get("observed_identity_sha256"):
errors.append(f"{label}.MISMATCHED identity digests must differ")


def validate_runtime_evidence_aggregate(aggregate: Any) -> dict[str, Any]:
"""Validate a redacted, static-only runtime evidence aggregate."""
errors: list[str] = []
label = "runtime_evidence_aggregate"
if not isinstance(aggregate, Mapping):
return {"ok": False, "errors": [f"{label} must be an object"]}

_append_forbidden_field_errors(aggregate, errors)
_append_missing_fields(
aggregate,
(
"contract_version",
"release_identity",
"risk_engine",
"effective_exposure_cap",
"stop_breaker_evaluation",
"reconciliation",
"static_validation_only",
"execution_permitted",
"verified_active",
"fills_verified",
"capital_use_verified",
),
errors,
label,
)
if aggregate.get("contract_version") != RUNTIME_EVIDENCE_CONTRACT_VERSION:
errors.append(f"{label} contract_version must be {RUNTIME_EVIDENCE_CONTRACT_VERSION}")
_validate_release_identity(aggregate.get("release_identity"), errors)

risk_engine = aggregate.get("risk_engine")
if not isinstance(risk_engine, Mapping):
errors.append(f"{label} risk_engine must be an object")
else:
if risk_engine.get("outcome") != "APPROVE":
errors.append(f"{label} risk_engine.outcome must be APPROVE")
if not isinstance(risk_engine.get("policy_version"), str) or not risk_engine["policy_version"].strip():
errors.append(f"{label} risk_engine.policy_version must be a non-empty string")

cap = aggregate.get("effective_exposure_cap")
if not isinstance(cap, Mapping):
errors.append(f"{label} effective_exposure_cap must be an object")
else:
value = cap.get("value")
if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 < value <= 1:
errors.append(f"{label} effective_exposure_cap.value must be in (0, 1]")
for field in ("mandate_version", "source"):
if not isinstance(cap.get(field), str) or not cap[field].strip():
errors.append(f"{label} effective_exposure_cap.{field} must be a non-empty string")

stop_breaker = aggregate.get("stop_breaker_evaluation")
if not isinstance(stop_breaker, Mapping):
errors.append(f"{label} stop_breaker_evaluation must be an object")
else:
if stop_breaker.get("stop_evaluated") is not True:
errors.append(f"{label} stop_breaker_evaluation.stop_evaluated must be true")
if stop_breaker.get("breaker_evaluated") is not True:
errors.append(f"{label} stop_breaker_evaluation.breaker_evaluated must be true")
if stop_breaker.get("outcome") != "CLEAR":
errors.append(f"{label} stop_breaker_evaluation.outcome must be CLEAR")
if not isinstance(stop_breaker.get("policy_version"), str) or not stop_breaker["policy_version"].strip():
errors.append(f"{label} stop_breaker_evaluation.policy_version must be a non-empty string")

_validate_reconciliation(aggregate.get("reconciliation"), errors)
for field in ("static_validation_only", "execution_permitted", "verified_active", "fills_verified", "capital_use_verified"):
expected = field == "static_validation_only"
if aggregate.get(field) is not expected:
errors.append(f"{label} {field} must be {str(expected).lower()} for static acceptance")
return {"ok": not errors, "errors": errors}


def build_runtime_evidence_aggregate(
*,
release_identity: Mapping[str, Any],
risk_engine: Mapping[str, Any],
effective_exposure_cap: Mapping[str, Any],
stop_breaker_evaluation: Mapping[str, Any],
reconciliation: Mapping[str, Any],
) -> dict[str, Any]:
"""Build a fail-closed aggregate that cannot claim runtime activity."""
aggregate = {
"contract_version": RUNTIME_EVIDENCE_CONTRACT_VERSION,
"release_identity": dict(release_identity),

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Detach nested inputs before returning the aggregate

When the caller retains and later updates a nested source object, especially release_identity["artifacts"][name], this outer dict() copy leaves that object shared with the returned aggregate. A routine update to the source after construction can therefore replace a validated digest or otherwise invalidate the aggregate without touching the returned value or triggering revalidation; deep-copy the inputs or reconstruct the closed validated schema.

Useful? React with 👍 / 👎.

"risk_engine": dict(risk_engine),
"effective_exposure_cap": dict(effective_exposure_cap),
"stop_breaker_evaluation": dict(stop_breaker_evaluation),
"reconciliation": dict(reconciliation),
"static_validation_only": True,
"execution_permitted": False,
"verified_active": False,
"fills_verified": False,
"capital_use_verified": False,
}
validation = validate_runtime_evidence_aggregate(aggregate)
if not validation["ok"]:
raise ValueError("Runtime evidence aggregate validation failed: " + "; ".join(validation["errors"]))
return aggregate


@dataclass
class ExecutionRuntime:
dry_run: bool = False
Expand Down
74 changes: 74 additions & 0 deletions tests/test_runtime_support.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,15 +14,89 @@

from runtime_support import (
ExecutionRuntime,
build_runtime_evidence_aggregate,
build_execution_report,
finalize_notification_delivery,
record_gating_event,
runtime_notify,
validate_runtime_evidence_aggregate,
)
from quant_platform_kit.common.runtime_target import build_runtime_target


class TestBuildExecutionReport(unittest.TestCase):
@staticmethod
def runtime_evidence_inputs():
return {
"release_identity": {
"strategy_profile": "crypto_live_pool_rotation",
"mode": "core_major",
"source_revision": "a" * 40,
"input_timestamp": "2026-03-13T00:00:00Z",
"artifact_contract": "crypto_live_pool_rotation.live_pool.v1",
"artifact_version": "2026-03-13-core_major",
"artifacts": {"live_pool": {"sha256": "b" * 64}},
},
"risk_engine": {"outcome": "APPROVE", "policy_version": "bootstrap_small_account_v2"},
"effective_exposure_cap": {
"value": 0.5,
"mandate_version": "bootstrap_small_account_v2",
"source": "approved_risk_mandate",
},
"stop_breaker_evaluation": {
"stop_evaluated": True,
"breaker_evaluated": True,
"outcome": "CLEAR",
"policy_version": "bootstrap_small_account_v2",
},
"reconciliation": {"status": "MISSING"},
}

def test_runtime_evidence_aggregate_is_redacted_and_static_only(self):
aggregate = build_runtime_evidence_aggregate(**self.runtime_evidence_inputs())

self.assertTrue(validate_runtime_evidence_aggregate(aggregate)["ok"])
self.assertFalse(aggregate["verified_active"])
self.assertFalse(aggregate["fills_verified"])
self.assertFalse(aggregate["capital_use_verified"])
self.assertNotIn("orders", str(aggregate))

def test_runtime_evidence_aggregate_fails_closed_for_risk_reconciliation_and_sensitive_fields(self):
aggregate = build_runtime_evidence_aggregate(**self.runtime_evidence_inputs())
aggregate["risk_engine"]["outcome"] = "REJECT"
aggregate["reconciliation"] = {"status": "MATCHED"}
aggregate["positions"] = [{"symbol": "BTCUSDT"}]
aggregate["release_identity"]["headers"] = {"authorization": "redacted"}

validation = validate_runtime_evidence_aggregate(aggregate)

self.assertFalse(validation["ok"])
self.assertIn("runtime_evidence_aggregate risk_engine.outcome must be APPROVE", validation["errors"])
self.assertIn(
"runtime_evidence_aggregate reconciliation.MATCHED requires durable_receipt_sha256",
validation["errors"],
)
self.assertIn("runtime_evidence_aggregate contains forbidden field: positions", validation["errors"])
self.assertIn("runtime_evidence_aggregate contains forbidden field: headers", validation["errors"])

def test_runtime_evidence_aggregate_rejects_static_matched_reconciliation(self):
matched_inputs = self.runtime_evidence_inputs()
matched_inputs["reconciliation"] = {
"status": "MATCHED",
"durable_receipt_sha256": "c" * 64,
"identity_sha256": "d" * 64,
}
mismatched_inputs = self.runtime_evidence_inputs()
mismatched_inputs["reconciliation"] = {
"status": "MISMATCHED",
"durable_receipt_sha256": "c" * 64,
"identity_sha256": "d" * 64,
"observed_identity_sha256": "e" * 64,
}

with self.assertRaisesRegex(ValueError, "MATCHED is not valid for static acceptance"):
build_runtime_evidence_aggregate(**matched_inputs)
self.assertTrue(validate_runtime_evidence_aggregate(build_runtime_evidence_aggregate(**mismatched_inputs))["ok"])
def test_report_contains_enrichment_fields(self):
runtime = ExecutionRuntime(dry_run=True, run_id="test-001")
report = build_execution_report(runtime)
Expand Down
Loading