Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -27,3 +27,7 @@ jobs:
- name: Run tests
run: |
pytest -q

- name: Run MGTP decision artefact demo (check mode)
run: |
python examples/minimal_decision_demo.py
7 changes: 7 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
__pycache__/
*.py[cod]
*.pyo
.pytest_cache/
*.egg-info/
dist/
build/
54 changes: 54 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,60 @@ Deterministic, hash-bound commit authority gate — stdlib-only, no network, no
- **CI:** [`commit_gate_ci.yml`](https://github.com/LalaSkye/constraint-workshop/actions/workflows/commit_gate_ci.yml) (Python 3.10/3.11/3.12 matrix)
- **Proof:** Determinism + drift-fail validated across Python 3.10/3.11/3.12.

## Deterministic Decision Artefact Demo

Proves that MGTP `DecisionRecord` serialisation is:
- **Deterministic**: same inputs → byte-identical `canonical_bytes`
- **Stable**: same bytes → same `decision_hash` (sha256)
- **Replayable**: stored golden bytes reproduce the same hash on any run

### Input parameters

| Field | Value |
|---|---|
| `transition_id` | `txn-0001` |
| `risk_class` | `LOW` |
| `irreversible` | `false` |
| `resource_identifier` | `res://demo/alpha` |
| `trust_boundary_crossed` | `false` |
| `override_token` | `null` |
| `timestamp` | `2026-01-01T00:00:00Z` |
| `actor_id` | `demo-actor` |
| `authority_basis` | `OWNER` |
| `tenant_id` | `tenant-001` |
| `verdict` | `APPROVED` |

### canonical_bytes (base64)

```
eyJhY3Rvcl9pZCI6ImRlbW8tYWN0b3IiLCJhdXRob3JpdHlfYmFzaXMiOiJPV05FUiIsImlycmV2
ZXJzaWJsZSI6ZmFsc2UsIm92ZXJyaWRlX3Rva2VuIjpudWxsLCJyZXNvdXJjZV9pZGVudGlmaWVy
IjoicmVzOi8vZGVtby9hbHBoYSIsInJpc2tfY2xhc3MiOiJMT1ciLCJ0ZW5hbnRfaWQiOiJ0ZW5h
bnQtMDAxIiwidGltZXN0YW1wIjoiMjAyNi0wMS0wMVQwMDowMDowMFoiLCJ0cmFuc2l0aW9uX2lk
IjoidHhuLTAwMDEiLCJ0cnVzdF9ib3VuZGFyeV9jcm9zc2VkIjpmYWxzZSwidmVyZGljdCI6IkFQ
UFJPVEVEIN0=
```

### decision_hash

```
8523083fc724b22f80fb638a2518133bf03f8c3283fbe7a0f629e04da5e01200
```

### Reproduce

```bash
python examples/minimal_decision_demo.py
```

### Replay test

```bash
pytest tests/test_mgtp_replay.py -v
```

---

## Scope boundaries

`/prometheus` is an **observability-only island**. It must not be imported by any execution path, gate, or pipeline code. It observes and reports; it cannot allow, hold, deny, or silence anything.
Expand Down
48 changes: 48 additions & 0 deletions examples/minimal_decision_demo.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
"""Minimal MGTP decision artefact demonstration.

Proves:
1. Deterministic canonicalisation: same inputs produce byte-identical output.
2. Stable decision_hash: sha256 of canonical_bytes is invariant.
3. Replayable decision reconstruction.

No randomness. Fixed timestamps. Fixed inputs.
Run with: python examples/minimal_decision_demo.py
"""

import base64
import sys
from pathlib import Path

# Allow running from repo root without installing the package.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from mgtp.types import DecisionRecord

# ---------------------------------------------------------------------------
# Fixed inputs — never change these; they anchor the golden artefact.
# ---------------------------------------------------------------------------
RECORD = DecisionRecord(
transition_id="txn-0001",
risk_class="LOW",
irreversible=False,
resource_identifier="res://demo/alpha",
trust_boundary_crossed=False,
override_token=None,
timestamp="2026-01-01T00:00:00Z",
actor_id="demo-actor",
authority_basis="OWNER",
tenant_id="tenant-001",
verdict="APPROVED",
)

# ---------------------------------------------------------------------------
# Derive artefact
# ---------------------------------------------------------------------------
canonical = RECORD.canonical_bytes()
b64 = base64.b64encode(canonical).decode("ascii")
h = RECORD.decision_hash()

print("=== MGTP Minimal Decision Artefact ===")
print(f"verdict : {RECORD.verdict}")
print(f"canonical_bytes : {b64}")
print(f"decision_hash : {h}")
38 changes: 38 additions & 0 deletions mgtp/types.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
import dataclasses
import hashlib
import json
from dataclasses import dataclass
from enum import Enum
from typing import Optional
Expand Down Expand Up @@ -32,3 +35,38 @@ class AuthorityContext:
actor_id: str
authority_basis: str # map to Evidence enum name e.g. "OWNER"
tenant_id: str


@dataclass(frozen=True)
class DecisionRecord:
"""Immutable record of a single MGTP transition decision.

Canonical serialisation: sorted-key JSON, no whitespace, UTF-8.
decision_hash: sha256 hex (lower-case) of canonical_bytes.
All fields are plain strings/bools so serialisation is deterministic.
"""

transition_id: str
risk_class: str # RiskClass value
irreversible: bool
resource_identifier: str
trust_boundary_crossed: bool
override_token: Optional[str]
timestamp: str # injected; do not call a clock
actor_id: str
authority_basis: str
tenant_id: str
verdict: str # TransitionOutcome value

def canonical_bytes(self) -> bytes:
"""Return canonical JSON bytes (sorted keys, no whitespace, UTF-8)."""
return json.dumps(
dataclasses.asdict(self),
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")

def decision_hash(self) -> str:
"""Return sha256 hex digest (lower-case) of canonical_bytes."""
return hashlib.sha256(self.canonical_bytes()).hexdigest()
1 change: 1 addition & 0 deletions tests/golden/decision_golden.b64
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
eyJhY3Rvcl9pZCI6ImRlbW8tYWN0b3IiLCJhdXRob3JpdHlfYmFzaXMiOiJPV05FUiIsImlycmV2ZXJzaWJsZSI6ZmFsc2UsIm92ZXJyaWRlX3Rva2VuIjpudWxsLCJyZXNvdXJjZV9pZGVudGlmaWVyIjoicmVzOi8vZGVtby9hbHBoYSIsInJpc2tfY2xhc3MiOiJMT1ciLCJ0ZW5hbnRfaWQiOiJ0ZW5hbnQtMDAxIiwidGltZXN0YW1wIjoiMjAyNi0wMS0wMVQwMDowMDowMFoiLCJ0cmFuc2l0aW9uX2lkIjoidHhuLTAwMDEiLCJ0cnVzdF9ib3VuZGFyeV9jcm9zc2VkIjpmYWxzZSwidmVyZGljdCI6IkFQUFJPVkVEIn0=
65 changes: 65 additions & 0 deletions tests/test_mgtp_replay.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
"""Replay test: golden canonical bytes → decision_hash round-trip.

Proves:
- Same inputs produce byte-identical canonical_bytes (replay works).
- decision_hash computed from stored bytes equals decision_hash from live record.
- Any byte-level drift in DecisionRecord serialisation fails this test.
"""

import base64
import hashlib
from pathlib import Path

from mgtp.types import DecisionRecord

GOLDEN_B64_PATH = Path(__file__).parent / "golden" / "decision_golden.b64"

# Fixed inputs — must match examples/minimal_decision_demo.py exactly.
RECORD = DecisionRecord(
transition_id="txn-0001",
risk_class="LOW",
irreversible=False,
resource_identifier="res://demo/alpha",
trust_boundary_crossed=False,
override_token=None,
timestamp="2026-01-01T00:00:00Z",
actor_id="demo-actor",
authority_basis="OWNER",
tenant_id="tenant-001",
verdict="APPROVED",
)


def _load_golden_bytes() -> bytes:
b64 = GOLDEN_B64_PATH.read_text(encoding="ascii").strip()
return base64.b64decode(b64)


def test_canonical_bytes_match_golden():
"""Live canonical_bytes must be byte-identical to stored golden fixture."""
golden = _load_golden_bytes()
live = RECORD.canonical_bytes()
assert live == golden, (
f"canonical_bytes drift detected.\n"
f" expected : {golden!r}\n"
f" got : {live!r}"
)


def test_decision_hash_stable():
"""decision_hash is sha256 of canonical_bytes — must equal golden hash."""
golden = _load_golden_bytes()
expected_hash = hashlib.sha256(golden).hexdigest()
assert RECORD.decision_hash() == expected_hash


def test_decision_hash_replay():
"""Recomputing decision_hash from stored bytes equals live decision_hash."""
stored_bytes = _load_golden_bytes()
replayed_hash = hashlib.sha256(stored_bytes).hexdigest()
live_hash = RECORD.decision_hash()
assert replayed_hash == live_hash, (
f"Replay hash mismatch.\n"
f" replayed : {replayed_hash}\n"
f" live : {live_hash}"
)