+ JSON Schemas
+ 8 schemas: AI System Inventory, RSP Manifest, Decision Envelope, Control Mapping, FRIA, Incident Record, FedReg Message, Obligation Spec.
+ aiSystemInventoryEntry
{
+ "title": "AI System Inventory Entry (Annex J1)",
+ "required": [
+ "systemId",
+ "businessOwner",
+ "euAiActTier",
+ "internalTier",
+ "modelRiskTier",
+ "lastFRIA",
+ "rspVersion"
+ ],
+ "fields": {
+ "systemId": "string",
+ "businessOwner": "string",
+ "euAiActTier": "enum[Prohibited|HighRisk|Limited|Minimal]",
+ "internalTier": "enum[T0|T1|T2|T3|T4|T5]",
+ "modelRiskTier": "enum[Tier-1|Tier-2|Tier-3]",
+ "annexIIIRef": "string",
+ "lastFRIA": "ISO-8601",
+ "lastDPIA": "ISO-8601",
+ "rspVersion": "string",
+ "regulatorEngagementStatus": "enum[Filed|Pending|UnderReview|Approved|Withdrawn]"
+ }
+}
rspManifest
{
+ "title": "Regulator Submission Pack \u2014 Manifest (v2.4+)",
+ "required": [
+ "rspId",
+ "version",
+ "subjectSystemId",
+ "issuedAt",
+ "signatures",
+ "merkleRoot"
+ ],
+ "fields": {
+ "rspId": "string",
+ "version": "string",
+ "subjectSystemId": "string",
+ "issuedAt": "ISO-8601",
+ "regulators": "string[]",
+ "artefacts": "object[]",
+ "signatures": "object[]",
+ "merkleRoot": "hex",
+ "policyBundleDigest": "hex",
+ "ledgerAnchorTx": "string"
+ }
+}
decisionEnvelope
{
+ "title": "Decision Envelope (per AI decision)",
+ "required": [
+ "decisionId",
+ "subjectId",
+ "modelId",
+ "modelVersion",
+ "inputsHash",
+ "output",
+ "shapTopK",
+ "ts",
+ "signature"
+ ],
+ "fields": {
+ "decisionId": "string",
+ "subjectId": "string",
+ "modelId": "string",
+ "modelVersion": "string",
+ "inputsHash": "hex",
+ "output": "object",
+ "shapTopK": "object[]",
+ "counterfactual": "object",
+ "policyDecision": "object",
+ "ts": "ISO-8601",
+ "signature": "object"
+ }
+}
controlMapping
{
+ "title": "Control Mapping (Annex J2 SoA)",
+ "required": [
+ "controlId",
+ "category",
+ "iso42001Ref",
+ "overlays",
+ "enforcement"
+ ],
+ "fields": {
+ "controlId": "string",
+ "category": "string",
+ "iso42001Ref": "string",
+ "overlays": "object",
+ "enforcement": "object",
+ "evidenceAutomation": "enum[None|Partial|Full]",
+ "owner": "string"
+ }
+}
friaRecord
{
+ "title": "FRIA + DPIA Combined Record (Annex J3)",
+ "required": [
+ "friaId",
+ "subjectSystemId",
+ "phase",
+ "residualRisk",
+ "approvers"
+ ],
+ "fields": {
+ "friaId": "string",
+ "subjectSystemId": "string",
+ "phase": "enum[A|B|C|D|E|F]",
+ "axes": "object[]",
+ "residualRisk": "enum[Low|Medium|High|Critical]",
+ "approvers": "string[]",
+ "nextReviewAt": "ISO-8601"
+ }
+}
incidentRecord
{
+ "title": "AI Incident Record (Cl. 10.2 + EU AI Act Art. 73)",
+ "required": [
+ "incidentId",
+ "severity",
+ "detectedAt",
+ "affectedSystems",
+ "narrative"
+ ],
+ "fields": {
+ "incidentId": "string",
+ "severity": "enum[Sev-0|Sev-1|Sev-2|Sev-3]",
+ "detectedAt": "ISO-8601",
+ "affectedSystems": "string[]",
+ "regulatorNotifications": "object[]",
+ "narrative": "string",
+ "rootCause": "string",
+ "capa": "object[]"
+ }
+}
fedRegMessage
{
+ "title": "Federation Protocol Message (FedReg)",
+ "required": [
+ "messageId",
+ "fromSpiffeId",
+ "toSpiffeId",
+ "op",
+ "payloadRef",
+ "consentScope"
+ ],
+ "fields": {
+ "messageId": "string",
+ "fromSpiffeId": "string",
+ "toSpiffeId": "string",
+ "op": "enum[Disclose|Subscribe|Challenge|Attest]",
+ "payloadRef": "string",
+ "consentScope": "object",
+ "signatures": "object[]",
+ "ts": "ISO-8601"
+ }
+}
obligationSpec
{
+ "title": "Formally-Verified Obligation Spec",
+ "required": [
+ "obligationId",
+ "regulatorRef",
+ "specLanguage",
+ "specHash",
+ "refinementProof"
+ ],
+ "fields": {
+ "obligationId": "string",
+ "regulatorRef": "string",
+ "specLanguage": "enum[TLA+|Lean|Coq]",
+ "specHash": "hex",
+ "refinementProof": "string",
+ "ciJobRef": "string"
+ }
+}
Code Examples
+ 11 reference implementations: OPA RSP gate, Terraform WORM evidence, decision envelope signer (Ed25519 + PQC), fairness monitor, FedReg client, predictive drift forecaster, TLA+ obligation spec, Lean FCRA spec, self-healing playbook engine, FastAPI traceability API, Merkle anchor.
+ opaRspGate
rego · Block RSP issuance unless all required artefacts + signatures present
package rsp.gate
+
+default allow = false
+
+required := {"manifest", "model-card", "datasheet", "fria-dpia",
+ "validation-report", "monitoring-plan", "incident-plan",
+ "policy-bundle", "attestations", "hash-chain"}
+
+allow {
+ have := {a | a := input.artefacts[_].name}
+ missing := required - have
+ count(missing) == 0
+ input.signatures.cosign.verified == true
+ input.signatures.intoto.verified == true
+ input.policyBundleDigest == data.policy.expectedDigest
+}
+
terraformWormEvidence
hcl · S3 Object Lock + KMS WORM evidence bucket (10-year retention)
resource "aws_s3_bucket" "aims_evidence" {
+ bucket = "gsifi-aims-evidence-${var.env}"
+ object_lock_enabled = true
+}
+
+resource "aws_s3_bucket_object_lock_configuration" "lock" {
+ bucket = aws_s3_bucket.aims_evidence.id
+ rule {
+ default_retention {
+ mode = "COMPLIANCE"
+ years = 10
+ }
+ }
+}
+
+resource "aws_s3_bucket_server_side_encryption_configuration" "sse" {
+ bucket = aws_s3_bucket.aims_evidence.id
+ rule {
+ apply_server_side_encryption_by_default {
+ kms_master_key_id = aws_kms_key.aims.arn
+ sse_algorithm = "aws:kms"
+ }
+ }
+}
+
decisionEnvelopeSigner
python · Sign per-decision envelopes (Ed25519 + PQC dual-sign)
import hashlib, json, time
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+# pqcrypto.sign.dilithium3 illustrative
+from pqcrypto.sign.dilithium3 import generate_keypair, sign as pqc_sign
+
+def make_envelope(decision_id, subject_id, model_id, model_version,
+ inputs, output, shap_topk, ed_sk, pqc_sk):
+ inputs_hash = hashlib.sha256(json.dumps(inputs, sort_keys=True).encode()).hexdigest()
+ body = {
+ "decisionId": decision_id,
+ "subjectId": subject_id,
+ "modelId": model_id,
+ "modelVersion": model_version,
+ "inputsHash": inputs_hash,
+ "output": output,
+ "shapTopK": shap_topk,
+ "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+ }
+ payload = json.dumps(body, sort_keys=True).encode()
+ sig_ed = ed_sk.sign(payload).hex()
+ sig_pqc = pqc_sign(pqc_sk, payload).hex()
+ body["signature"] = {"ed25519": sig_ed, "dilithium3": sig_pqc}
+ return body
+
fairnessMonitor
python · Daily AIR / EOD monitor with self-healing trigger (SH-01)
import numpy as np
+
+def adverse_impact_ratio(y_pred, protected):
+ rates = {g: y_pred[protected == g].mean() for g in np.unique(protected)}
+ ref = max(rates.values())
+ return min(rates.values()) / ref if ref else 1.0
+
+def monitor(daily_predictions, protected, prev_air, prev_psi):
+ air = adverse_impact_ratio(daily_predictions, protected)
+ if air < 0.85 or prev_psi > 0.2:
+ trigger_self_heal("SH-01", reason={"air": air, "psi": prev_psi})
+ return {"air": air}
+
+def trigger_self_heal(playbook_id, reason):
+ # POST signed event to governance bus → triggers rollback + Sev-2 ticket
+ ...
+
fedRegClient
python · FedReg federation client — disclose RSP slice to supervisor
import requests, json, time
+
+def disclose(supervisor_url, rsp_slice, scope, spiffe_ctx, signer):
+ msg = {
+ "messageId": f"msg-{int(time.time()*1000)}",
+ "fromSpiffeId": spiffe_ctx.self_id,
+ "toSpiffeId": spiffe_ctx.peer_id,
+ "op": "Disclose",
+ "payloadRef": rsp_slice["uri"],
+ "consentScope": scope,
+ "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+ }
+ body = json.dumps(msg, sort_keys=True).encode()
+ msg["signatures"] = signer(body)
+ return requests.post(supervisor_url + "/fedreg/v1/messages",
+ json=msg, cert=spiffe_ctx.mtls_cert).json()
+
predictiveDriftForecaster
python · Forecast PSI breach 7 days ahead (predictive governance)
from prophet import Prophet
+import pandas as pd
+
+def forecast_psi_breach(history_df, threshold=0.2, horizon=7):
+ m = Prophet(interval_width=0.95).fit(history_df.rename(columns={"date":"ds","psi":"y"}))
+ future = m.make_future_dataframe(periods=horizon)
+ fcst = m.predict(future)
+ breach = fcst[fcst["yhat"] > threshold].head(1)
+ return None if breach.empty else {
+ "predictedBreachAt": str(breach.iloc[0]["ds"].date()),
+ "expectedPsi": float(breach.iloc[0]["yhat"]),
+ }
+
tlaPlusObligation
tla · TLA+ liveness spec for EU AI Act Art. 73 incident reporting
-------------- MODULE Art73Reporting --------------
+EXTENDS Naturals, TLC
+VARIABLES status, notifiedAt, detectedAt
+Init == /\ status = "open" /\ notifiedAt = 0 /\ detectedAt = 0
+Report == /\ status = "open" /\ status' = "reported"
+ /\ notifiedAt' = detectedAt + 15
+Liveness == <>(status = "reported" /\ notifiedAt - detectedAt <= 15)
+Spec == Init /\ [][Report]_<<status, notifiedAt, detectedAt>> /\ Liveness
+====
+
leanFcraSpec
lean · Lean spec for FCRA §615 adverse-action obligation
import data.real.basic
+
+structure Decision := (subject : string) (denied : bool) (timestamp_h : nat)
+structure Notice := (subject : string) (sent_at_h : nat) (reasons : list string)
+
+def fcra_compliant (d : Decision) (n : Notice) : Prop :=
+ d.subject = n.subject
+ ∧ d.denied = tt
+ ∧ n.sent_at_h ≤ d.timestamp_h + 30 * 24
+ ∧ n.reasons.length ≥ 1
+
+theorem fcra_demo :
+ ∀ d n, d.denied = tt → fcra_compliant d n → n.reasons.length ≥ 1 :=
+λ d n h1 hc, hc.2.2.2
+
selfHealingPlaybookEngine
python · Self-healing playbook executor with WORM-attested actions
import json, time, hashlib
+
+def execute_playbook(playbook, signals, signer, worm_writer):
+ record = {"playbook": playbook["id"], "trigger": playbook["trigger"],
+ "signals": signals, "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ")}
+ if playbook["id"] == "SH-01":
+ rollback_model(signals["modelId"])
+ open_ticket(severity="Sev-2", reason="Bias drift")
+ elif playbook["id"] == "SH-02":
+ quarantine_workload(signals["workloadId"])
+ restore_lkg_bundle()
+ payload = json.dumps(record, sort_keys=True).encode()
+ record["digest"] = hashlib.sha256(payload).hexdigest()
+ record["signature"] = signer(payload)
+ worm_writer.write(record)
+ return record
+
+def rollback_model(*a, **k): ...
+def open_ticket(*a, **k): ...
+def quarantine_workload(*a, **k): ...
+def restore_lkg_bundle(*a, **k): ...
+
rspApiFastapi
python · FastAPI decision-traceability API for RSP v2.4+
from fastapi import FastAPI, HTTPException, Depends
+app = FastAPI(title="RSP Decision Traceability API")
+
+def auth(spiffe_id: str = ""):
+ if not spiffe_id.startswith("spiffe://supervisor."):
+ raise HTTPException(401, "Supervisor SPIFFE required")
+ return spiffe_id
+
+@app.get("/rsp/{rsp_id}/decisions/{decision_id}")
+def get_decision(rsp_id: str, decision_id: str, who=Depends(auth)):
+ env = decision_store.fetch(rsp_id, decision_id)
+ if not env: raise HTTPException(404, "Decision not found")
+ return env
+
+@app.post("/rsp/{rsp_id}/challenge")
+def challenge(rsp_id: str, body: dict, who=Depends(auth)):
+ return counterfactual_engine.run(rsp_id, body)
+
merkleAnchor
python · Daily Merkle anchor of evidence WORM into public ledger
import hashlib
+
+def merkle_root(leaves):
+ layer = [bytes.fromhex(l) for l in leaves]
+ while len(layer) > 1:
+ if len(layer) % 2: layer.append(layer[-1])
+ layer = [hashlib.sha256(layer[i]+layer[i+1]).digest() for i in range(0,len(layer),2)]
+ return layer[0].hex()
+
+def anchor_today(evidence_hashes, ledger_client):
+ root = merkle_root(evidence_hashes)
+ txid = ledger_client.publish(root)
+ return {"root": root, "txid": txid, "count": len(evidence_hashes)}
+