Add lawful-collapse universe overlay

.github/workflows/rootfile-runtime.yml

@@ -28,6 +28,9 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install -r requirements.txt
 
+      - name: Rootfile manifest validation
+        run: python -m tools.validate_rootfile --repo-root .
+
       - name: Rootfile validators and token flow
         run: python -m pytest tests\rootfile -q
 

core/economics/__init__.py

@@ -0,0 +1,5 @@
+"""Economics overlays for lawful-collapse artifacts."""
+
+from core.economics.qpt_token import mint_qpt_if_applicable
+
+__all__ = ["mint_qpt_if_applicable"]

core/economics/qpt_token.py

@@ -0,0 +1,81 @@
+"""Disabled-by-default QPT minting for accepted lawful-collapse reports."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import time
+from pathlib import Path
+from typing import Optional
+
+from core.meta import OperatorMeta
+from core.orchestration.reconciliation import ReconciliationReport
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.economics",
+    operator_type="qpt_minter",
+    canonical_law="H12",
+)
+
+
+def _ledger_path() -> Path:
+    return Path(os.getenv("APEX_QPT_LEDGER", "data/qpt_ledger.jsonl"))
+
+
+def _append_jsonl(path: Path, record: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    line = json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n"
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(line)
+        handle.flush()
+        os.fsync(handle.fileno())
+
+
+def _eligible(report: ReconciliationReport) -> bool:
+    threshold = float(os.getenv("QPT_INFORMATION_GAIN_THRESHOLD", "0.0") or 0.0)
+    return (
+        report.accepted
+        and report.admissible
+        and report.information_gain >= threshold
+        and report.scheduler_authorized
+        and report.evidence_valid
+    )
+
+
+def mint_qpt_if_applicable(report: ReconciliationReport) -> Optional[str]:
+    """Persist a local QPT token ID only when all collapse proof gates pass."""
+    if os.getenv("ENABLE_QPT", "0") != "1":
+        return None
+    if not _eligible(report):
+        return None
+
+    timestamp = time.time()
+    identity = {
+        "broker": report.broker,
+        "execution_id": report.execution_id,
+        "information_gain": report.information_gain,
+        "realized_pnl": report.realized_pnl,
+        "status": report.status,
+        "symbol": report.symbol,
+        "timestamp": timestamp,
+    }
+    token_id = "qpt_" + hashlib.sha256(
+        json.dumps(identity, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()[:24]
+    _append_jsonl(
+        _ledger_path(),
+        {
+            "token_id": token_id,
+            **identity,
+            "conditions": {
+                "admissible": report.admissible,
+                "evidence_valid": report.evidence_valid,
+                "reconciliation_accepted": report.accepted,
+                "scheduler_authorized": report.scheduler_authorized,
+            },
+        },
+    )
+    return token_id

core/meta.py

@@ -0,0 +1,48 @@
+"""Operator metadata helpers for rootfile law introspection."""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Mapping, NamedTuple, TypeVar
+
+
+class OperatorMeta(NamedTuple):
+    """Canonical operator metadata attached to modules, classes, or functions."""
+
+    tier: str
+    layer: str
+    operator_type: str
+    canonical_law: str = ""
+
+
+T = TypeVar("T")
+
+
+def normalize_meta(meta: OperatorMeta | Mapping[str, Any]) -> OperatorMeta:
+    """Convert legacy dict metadata or OperatorMeta into OperatorMeta."""
+    if isinstance(meta, OperatorMeta):
+        return meta
+    return OperatorMeta(
+        tier=str(meta.get("tier", "")),
+        layer=str(meta.get("layer", "")),
+        operator_type=str(meta.get("operator_type", "")),
+        canonical_law=str(meta.get("canonical_law", "")),
+    )
+
+
+def declare_operator(meta: OperatorMeta | Mapping[str, Any]) -> Callable[[T], T]:
+    """Attach canonical rootfile metadata to a function or class."""
+    normalized = normalize_meta(meta)
+
+    def decorator(obj: T) -> T:
+        setattr(obj, "__operator_meta__", normalized)
+        return obj
+
+    return decorator
+
+
+def get_declared_meta(obj: Any) -> OperatorMeta | None:
+    """Return metadata declared with @declare_operator, if present."""
+    meta = getattr(obj, "__operator_meta__", None)
+    if meta is None:
+        return None
+    return normalize_meta(meta)

core/orchestration/__init__.py

@@ -4,11 +4,15 @@
 from trading.kernel.apex_engine import ApexEngine, ExecutionMode, ExecutionOutcome, ExecutionResult
 from trading.kernel.scheduler import CollapseDecision, ExecutionToken, Scheduler
 
-META = {
-    "tier": "rootfile",
-    "layer": "core.orchestration",
-    "operator_type": "orchestration_adapter",
-}
+from core.meta import OperatorMeta
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="orchestration_adapter",
+    canonical_law="H10",
+)
 
 __all__ = [
     "ApexEngine",

core/orchestration/apex_engine.py

@@ -2,9 +2,13 @@
 
 from trading.kernel.apex_engine import *  # noqa: F401,F403
 
-META = {
-    "tier": "rootfile",
-    "layer": "core.orchestration",
-    "operator_type": "engine_adapter",
-}
+from core.meta import OperatorMeta
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="engine_adapter",
+    canonical_law="H10",
+)
 

core/orchestration/constraints.py

@@ -2,9 +2,13 @@
 
 from trading.kernel.H_constraints import *  # noqa: F401,F403
 
-META = {
-    "tier": "rootfile",
-    "layer": "core.orchestration",
-    "operator_type": "constraint_adapter",
-}
+from core.meta import OperatorMeta
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="constraint_adapter",
+    canonical_law="H8",
+)
 

core/orchestration/evidence.py

@@ -0,0 +1,58 @@
+"""Unified H13 evidence facade."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+from core.meta import OperatorMeta
+from tachyonic_chain.audit_log import append_execution_evidence
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="evidence_facade",
+    canonical_law="H13",
+)
+
+
+@dataclass
+class EvidenceEvent:
+    """Canonical evidence event accepted by the facade."""
+
+    event_type: str
+    execution_id: str
+    operation: str
+    payload: Dict[str, Any] = field(default_factory=dict)
+    symbol: Optional[str] = None
+    outcome: str = "recorded"
+    token_status: str = "not_applicable"
+
+
+def _external_anchor_status(payload: Dict[str, Any]) -> Dict[str, Any]:
+    """Return optional external anchor status without blocking pipeline work."""
+    if os.getenv("ENABLE_ANCHORING", "0") != "1":
+        return {"anchor_status": "disabled"}
+    endpoint = os.getenv("APEX_ANCHOR_ENDPOINT", "").strip()
+    if not endpoint:
+        return {"anchor_status": "failed", "anchor_error": "missing_endpoint"}
+    # Network anchoring is deliberately not performed in this offline-safe adapter.
+    return {"anchor_status": "configured", "anchor_endpoint": endpoint, "anchor_txid": None}
+
+
+def emit_evidence(event: EvidenceEvent, *, log_path: str | None = None) -> str:
+    """Append runtime evidence and include optional non-blocking anchor status."""
+    payload = dict(event.payload)
+    payload.update(_external_anchor_status(payload))
+    return append_execution_evidence(
+        event_type=event.event_type,
+        execution_id=event.execution_id,
+        operation=event.operation,
+        symbol=event.symbol,
+        outcome=event.outcome,
+        token_status=event.token_status,
+        payload=payload,
+        log_path=log_path,
+    )

core/orchestration/meta.py

@@ -0,0 +1,42 @@
+"""H-meta registry for rootfile operator jurisdiction."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Iterable
+
+from core.meta import OperatorMeta, normalize_meta
+from core.rootfile_manifest import LAWS
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="operator_registry",
+    canonical_law="H10",
+)
+
+
+@dataclass
+class OperatorRegistry:
+    """In-memory registry of operator metadata keyed by stable name."""
+
+    operators: Dict[str, OperatorMeta] = field(default_factory=dict)
+
+    def register(self, name: str, meta: OperatorMeta | dict) -> OperatorMeta:
+        normalized = normalize_meta(meta)
+        if normalized.canonical_law and normalized.canonical_law not in LAWS:
+            raise ValueError(f"{name} declares unknown canonical law {normalized.canonical_law}")
+        self.operators[name] = normalized
+        return normalized
+
+    def register_many(self, items: Iterable[tuple[str, OperatorMeta | dict]]) -> None:
+        for name, meta in items:
+            self.register(name, meta)
+
+    def assert_jurisdiction(self, name: str, law_id: str) -> None:
+        meta = self.operators.get(name)
+        if meta is None:
+            raise KeyError(f"operator not registered: {name}")
+        if meta.canonical_law and meta.canonical_law != law_id:
+            raise PermissionError(f"{name} belongs to {meta.canonical_law}, not {law_id}")

core/orchestration/reconciliation.py

@@ -0,0 +1,51 @@
+"""Broker-neutral H12 reconciliation facade."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List
+
+from core.meta import OperatorMeta
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="reconciliation_facade",
+    canonical_law="H12",
+)
+
+
+@dataclass
+class ReconciliationReport:
+    """Canonical reconciliation summary consumed by QPT/evidence layers."""
+
+    broker: str
+    accepted: bool
+    status: str
+    realized_pnl: float | None = None
+    evidence_valid: bool = False
+    admissible: bool = False
+    information_gain: float = 0.0
+    scheduler_authorized: bool = False
+    execution_id: str = ""
+    symbol: str = ""
+    payload: Dict[str, Any] = field(default_factory=dict)
+
+
+def summarize_settlement_report(broker: str, report: Any) -> ReconciliationReport:
+    """Convert existing MT5/Deriv settlement report objects into a canonical report."""
+    records: List[Any] = list(getattr(report, "records", []) or [])
+    closed = [record for record in records if getattr(record, "status", "") == "closed"]
+    realized = [getattr(record, "realized_pnl", None) for record in closed]
+    realized_values = [float(value) for value in realized if value is not None]
+    return ReconciliationReport(
+        broker=broker,
+        accepted=bool(closed),
+        status="accepted" if closed else "no_closed_records",
+        realized_pnl=sum(realized_values) if realized_values else None,
+        payload={
+            "record_count": len(records),
+            "closed_count": len(closed),
+        },
+    )

core/orchestration/root.py

@@ -0,0 +1,34 @@
+"""H-root runtime setup helpers for lawful-collapse state."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Iterable
+
+from core.meta import OperatorMeta
+
+
+META = OperatorMeta(
+    tier="rootfile",
+    layer="core.orchestration",
+    operator_type="root_runtime_setup",
+    canonical_law="H1",
+)
+
+
+DEFAULT_RUNTIME_DIRS = (
+    "logs",
+    "data/models",
+    "trading_data/pnl",
+)
+
+
+def setup_runtime(root: str | Path = ".", dirs: Iterable[str] = DEFAULT_RUNTIME_DIRS) -> list[Path]:
+    """Create required runtime directories and return the created/resolved paths."""
+    base = Path(root)
+    paths = []
+    for rel in dirs:
+        path = base / rel
+        path.mkdir(parents=True, exist_ok=True)
+        paths.append(path)
+    return paths