# WEKID ASK (Architecture Style Kit) Master Context Manifest
## Status: Implementation Specification & Reference Blueprint | Version 1.3

This single, self-contained reference file is the system context for Large Language Models (LLMs) and coding assistants (e.g., Cursor, Claude Projects, Custom GPTs). It sets out the architectural requirements, infrastructure blueprints, dependency set-up and code patterns needed to implement the **WEKID-STD-001 v1.2** governance framework with the **wekid-sdk 1.2.0** reference implementation. All code in Sections 02–05 is copied verbatim from the SDK source and covered by its test suite. The SDK is open source under the Apache License 2.0; the WEKID™ name and marks are not licensed by it.

---

## 1. Architectural Mandate & Strategy
The **WEKID ASK** is a deterministic, fail-closed runtime gate for probabilistic AI outputs. No AI agent or system may execute downstream infrastructure changes, data operations or high-risk business logic without presenting verifiable evidence to the gate, and no authority is granted unless the decision has first been committed to the evidence vault.

### The Runtime Lifecycle Execution
1. **Parse:** The AI output is broken into isolated, testable `Claims`, each bound to one of the five epistemic layers (Data, Information, Knowledge, Experience, Wisdom) and to a verifier registered for that same layer.
2. **Score:** The `VerifierRegistry` checks each claim against live systems and produces measured evidence. The five layers are scored over the evidence actually supplied; a layer with no evidence supports nothing. Self-declared evidence is labelled as such.
3. **Gate:** The Trust Bridge applies the Decision Matrix (first match governs) to set a maturity ceiling, consequence sets a stakes cap, and the Authority Level (AL-0 to AL-4) is whichever is more conservative.
4. **Commit:** Every attempt, granted or denied, is fingerprinted and written to the WORM evidence vault after the decision and before authority is returned. If the write fails, the gate grants no authority.

### Authority Levels
| Level | Meaning | Integration behaviour |
|---|---|---|
| AL-3 Augmented | Autonomous within scope; human on the loop | Execute (sample for review when `enhanced_sampling` is true) |
| AL-2 Supervised | AI proposes and stages; a human releases each action | Stage; route `record_id` to the accountable human |
| AL-1 Constrained | Whitelisted, reversible, impact-capped actions only | Execute only reversible, in-cap actions |
| AL-0 No Authority | Informational only | Do not execute; re-plan from the governing triggers |
| AL-4 Human Authority | Reserved to an accountable human (from consequence, never maturity) | Do not act; inform the human decision-maker |

### Decision Matrix (WEKID-STD-001 v1.2)
| Row | Outcome | Condition | Maturity ceiling |
|---|---|---|---|
| 1 | Rejected | Hard gate: fabrication, Data < 2, high-stakes judgment failure, failed Data-layer claim, policy hard gate | AL-0 |
| 2 | Insufficient Maturity | A layer without evidence, or Maturity Score < 50 | AL-0 |
| 3 | Oversight Decay | Engagement gating active and the human loop not exercised | At most AL-2 |
| 4 | Constrained | Any layer < 3, or Wisdom / Experience below the profile floor | AL-2 |
| 5 | Remediation Required | Maturity Score 50–69, or a soft gate fired | AL-1 |
| 6 | Approved with Monitoring | Maturity Score 70–79, gate-free | AL-3 + enhanced sampling |
| 7 | Approved for Autonomous Use | Maturity Score ≥ 80, gate-free | AL-3 |
| 8 | Policy-Reserved | Safety-critical, legally reserved, reserved decision class, or irreversible and high-risk | AL-4 |

---

## 2. Infrastructure Layer (AWS S3 Compliance Vault)

### Terraform Blueprint (`infra/terraform/main.tf`)
The immutable evidence repository. Object Lock runs in **COMPLIANCE mode**, so no one, including the root user, can delete or alter a record during retention. Retention is a variable (default 2,555 days, about seven years); public access is blocked, the bucket accepts TLS only, and separate least-privilege policies are issued for the gate and for auditors. Variables and outputs are in `variables.tf` and `outputs.tf`.

```hcl
# SPDX-License-Identifier: Apache-2.0
# Copyright 2025-2026 WEKID LLC
# WEKID evidence vault — S3 with Object Lock (COMPLIANCE mode).
#
# Changes from the ASK v1.2 manifest blueprint:
#   * retention is a variable (default 2555 days ≈ 7 years; the manifest's 365 days is
#     usually too short for SOX-relevant records — align with your retention schedule)
#   * public access fully blocked; TLS-only bucket policy; bucket-owner-enforced ownership
#   * optional SSE-KMS (customer-managed key) instead of SSE-S3
#   * separate least-privilege policies for the gate (append-only) and auditors (read-only)
#
# COMPLIANCE mode cannot be shortened or removed by anyone, including the root user, for the
# retention period. Test with a short retention in a sandbox account first.

terraform {
  required_version = ">= 1.5"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = ">= 5.0"
    }
  }
}

provider "aws" {
  region = var.region
}

resource "aws_s3_bucket" "wekid_corpus" {
  bucket              = var.bucket_name
  object_lock_enabled = true
  tags                = merge(var.tags, { Framework = "WEKID-Governance", Compliance = "Immutable-Evidence-Logs" })
}

resource "aws_s3_bucket_ownership_controls" "wekid" {
  bucket = aws_s3_bucket.wekid_corpus.id
  rule {
    object_ownership = "BucketOwnerEnforced"
  }
}

resource "aws_s3_bucket_public_access_block" "wekid" {
  bucket                  = aws_s3_bucket.wekid_corpus.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_versioning" "wekid" {
  bucket = aws_s3_bucket.wekid_corpus.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_object_lock_configuration" "wekid" {
  bucket = aws_s3_bucket.wekid_corpus.id
  rule {
    default_retention {
      mode = "COMPLIANCE"
      days = var.retention_days
    }
  }
  depends_on = [aws_s3_bucket_versioning.wekid]
}

resource "aws_s3_bucket_server_side_encryption_configuration" "wekid" {
  bucket = aws_s3_bucket.wekid_corpus.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = var.kms_key_arn == "" ? "AES256" : "aws:kms"
      kms_master_key_id = var.kms_key_arn == "" ? null : var.kms_key_arn
    }
    bucket_key_enabled = var.kms_key_arn != ""
  }
}

data "aws_iam_policy_document" "bucket" {
  statement {
    sid       = "DenyInsecureTransport"
    effect    = "Deny"
    actions   = ["s3:*"]
    resources = [aws_s3_bucket.wekid_corpus.arn, "${aws_s3_bucket.wekid_corpus.arn}/*"]
    principals {
      type        = "*"
      identifiers = ["*"]
    }
    condition {
      test     = "Bool"
      variable = "aws:SecureTransport"
      values   = ["false"]
    }
  }
}

resource "aws_s3_bucket_policy" "wekid" {
  bucket     = aws_s3_bucket.wekid_corpus.id
  policy     = data.aws_iam_policy_document.bucket.json
  depends_on = [aws_s3_bucket_public_access_block.wekid]
}

# --- least-privilege policies ---------------------------------------------------------

data "aws_iam_policy_document" "gate_writer" {
  statement {
    sid       = "WekidAppendOnlyWrite"
    actions   = ["s3:PutObject", "s3:PutObjectRetention"]
    resources = ["${aws_s3_bucket.wekid_corpus.arn}/${var.prefix}*"]
  }
  dynamic "statement" {
    for_each = var.kms_key_arn == "" ? [] : [1]
    content {
      sid       = "WekidKmsEncrypt"
      actions   = ["kms:GenerateDataKey", "kms:Encrypt"]
      resources = [var.kms_key_arn]
    }
  }
}

data "aws_iam_policy_document" "gate_reader" {
  # Only needed if the gate serves /v1/records (record retrieval and overrides).
  statement {
    sid       = "WekidReadRecords"
    actions   = ["s3:GetObject"]
    resources = ["${aws_s3_bucket.wekid_corpus.arn}/${var.prefix}*"]
  }
  statement {
    sid       = "WekidListOverrides"
    actions   = ["s3:ListBucket"]
    resources = [aws_s3_bucket.wekid_corpus.arn]
    condition {
      test     = "StringLike"
      variable = "s3:prefix"
      values   = ["${var.prefix}overrides/*"]
    }
  }
  dynamic "statement" {
    for_each = var.kms_key_arn == "" ? [] : [1]
    content {
      sid       = "WekidKmsDecrypt"
      actions   = ["kms:Decrypt"]
      resources = [var.kms_key_arn]
    }
  }
}

data "aws_iam_policy_document" "auditor" {
  statement {
    sid       = "WekidAuditRead"
    actions   = ["s3:GetObject", "s3:GetObjectVersion", "s3:GetObjectRetention", "s3:GetObjectAttributes"]
    resources = ["${aws_s3_bucket.wekid_corpus.arn}/*"]
  }
  statement {
    sid = "WekidAuditList"
    actions = ["s3:ListBucket", "s3:ListBucketVersions", "s3:GetBucketObjectLockConfiguration",
    "s3:GetBucketVersioning", "s3:GetEncryptionConfiguration", "s3:GetBucketPolicy"]
    resources = [aws_s3_bucket.wekid_corpus.arn]
  }
  dynamic "statement" {
    for_each = var.kms_key_arn == "" ? [] : [1]
    content {
      sid       = "WekidKmsDecrypt"
      actions   = ["kms:Decrypt"]
      resources = [var.kms_key_arn]
    }
  }
}

resource "aws_iam_policy" "gate_writer" {
  name   = "${var.name_prefix}-gate-writer"
  policy = data.aws_iam_policy_document.gate_writer.json
}

resource "aws_iam_policy" "gate_reader" {
  name   = "${var.name_prefix}-gate-reader"
  policy = data.aws_iam_policy_document.gate_reader.json
}

resource "aws_iam_policy" "auditor" {
  name   = "${var.name_prefix}-auditor-readonly"
  policy = data.aws_iam_policy_document.auditor.json
}
```

### IAM Security Profile — gate writer (`infra/iam/gate-writer-policy.json`)
The gate writes blind and append-only. It cannot read, delete or shorten retention.

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "WekidAppendOnlyWrite",
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:PutObjectRetention"],
      "Resource": "arn:aws:s3:::YOUR-WEKID-BUCKET/evidence/*"
    }
  ]
}
```

### IAM Security Profile — auditor (`infra/iam/auditor-readonly-policy.json`)
A separate read-only role for internal audit (audit check 17).

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "WekidAuditRead",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:GetObjectVersion", "s3:GetObjectRetention", "s3:GetObjectAttributes"],
      "Resource": "arn:aws:s3:::YOUR-WEKID-BUCKET/*"
    },
    {
      "Sid": "WekidAuditList",
      "Effect": "Allow",
      "Action": ["s3:ListBucket", "s3:ListBucketVersions", "s3:GetBucketObjectLockConfiguration",
                 "s3:GetBucketVersioning", "s3:GetEncryptionConfiguration", "s3:GetBucketPolicy"],
      "Resource": "arn:aws:s3:::YOUR-WEKID-BUCKET"
    }
  ]
}
```

---

## 3. Dependency Configuration (`pyproject.toml`)
The SDK is a single pip-installable package. The core engine, gate, vaults, client and CLI use only the Python standard library; the HTTP service and S3 vault are optional extras. Install with `pip install "wekid_sdk-1.2.0-py3-none-any.whl[server,aws]"` and verify with `wekid scenarios`.

```toml
[build-system]
requires = ["setuptools>=77", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "wekid-sdk"
version = "1.2.0"
description = "WEKID Architecture Style Kit (ASK) runtime gate and SDK — implements WEKID-STD-001 v1.2: five-layer scoring, the Trust Bridge, AI Decision Authority Levels AL-0..AL-4, and a WORM evidence vault."
readme = "README.md"
requires-python = ">=3.10"
license = "Apache-2.0"
license-files = ["LICENSE", "NOTICE"]
authors = [{ name = "WEKID LLC" }]
maintainers = [{ name = "James Madigan Judge" }]
keywords = ["wekid", "ai-governance", "trust-bridge", "authority-levels", "audit", "worm"]
classifiers = [
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Programming Language :: Python :: 3.13",
    "Operating System :: OS Independent",
]
# The core engine, gate, vaults (memory/file), client and CLI use the standard library only.
dependencies = []

[project.optional-dependencies]
server = ["fastapi>=0.110", "uvicorn>=0.29", "pydantic>=2.6"]
aws = ["boto3>=1.28"]
test = ["httpx>=0.27", "fastapi>=0.110", "pydantic>=2.6"]
all = ["fastapi>=0.110", "uvicorn>=0.29", "pydantic>=2.6", "boto3>=1.28", "httpx>=0.27"]

[project.scripts]
wekid = "wekid.cli:main"

[project.urls]
Homepage = "https://wekid.org"
Framework = "https://wekid.org/framework.php"
Standards = "https://wekid.org/standards.php"
Blueprint = "https://wekid.org/blueprint/ask/index.php"

[tool.setuptools.packages.find]
where = ["src"]

[tool.setuptools.package-data]
wekid = ["data/*.json", "py.typed"]
```

---

## 4. Software Application Architecture (Python)

### Module Map (`src/wekid/`)
| Module | Responsibility |
|---|---|
| `models.py` | Data contract: AuthorityLevel, Outcome, Provenance, request and Determination types |
| `calibration.py` | CalibrationProfile; refuses any profile that would weaken the protected core |
| `scoring.py` | Five-layer scoring over supplied evidence; Maturity Score 0–100 |
| `bridge.py` | Gates, floors, Decision Matrix, provenance cap, stakes floor, authority resolution |
| `verifiers.py` | Layer-bound, fail-closed, freezable VerifierRegistry; built-in `sql_safety` |
| `sqlsafety.py` | Static SQL safety checks |
| `vault.py` | MemoryVault, FileVault (hash-chained), S3ObjectLockVault (WORM) |
| `gate.py` | WekidGate: assess, permits, guard, governed, overrides |
| `api.py` / `client.py` / `cli.py` | HTTP service, zero-dependency client, `wekid` command |

### Modular System Verifiers (`examples/custom_verifiers.py`)
Each verifier checks one claim against a live system and returns measured evidence for one layer. Modules expose `register(registry)`; the service loads them at start-up (`WEKID_VERIFIER_MODULES`) and then freezes the registry, so verifiers change only through a deployment. Unknown verifiers, errors and the manifest's original `DB_METADATA_API` and `CI_BUILD_LOGS` placeholders all fail closed.

```python
# SPDX-License-Identifier: Apache-2.0
# Copyright 2025-2026 WEKID LLC
"""
examples/custom_verifiers.py — how to plug live systems into the gate.

Load it into the service with:

    WEKID_VERIFIER_MODULES=custom_verifiers  wekid serve      # module must be importable

The service calls ``register(registry)`` at start-up and then freezes the
registry, so verifiers can only change through a deployment (change control).

Each verifier returns measured evidence for ONE layer. Replace the stub
functions below with calls to your real systems.
"""

from __future__ import annotations

from wekid import Claim, VerifierRegistry, VerifierResult, http_health, metric_threshold


# --- Data layer: is the target database healthy and within limits? -----------
def _replica_lag_seconds(claim: Claim) -> float:
    # e.g. query CloudWatch / pg_stat_replication for the cluster named in the claim
    return 1.4


def _orders_db_metadata(content: str, claim: Claim) -> VerifierResult:
    lag = _replica_lag_seconds(claim)
    storage_used = 0.62  # e.g. from your metadata API
    if lag > 5.0 or storage_used > 0.85:
        return VerifierResult(False, 0.0, f"replica lag {lag}s / storage {storage_used:.0%} outside limits")
    # Per-component evidence (0-5) is better than a single score when you have it.
    return VerifierResult(True, 1.0, f"replica lag {lag}s, storage {storage_used:.0%}",
                          signals={"source_integrity": 4.8, "freshness": 4.6 if lag < 2 else 3.6,
                                   "anomaly_control": 4.5, "access_control": 4.7,
                                   "completeness": 4.5, "lineage": 4.4})


# --- Experience layer: has this change passed CI / staging before? -----------
def _ci_status(content: str, claim: Claim) -> VerifierResult:
    passed, total = 48, 50  # e.g. from your CI API for the change referenced in the claim
    rate = passed / total
    return VerifierResult(rate >= 0.95, rate, f"{passed}/{total} staging runs passed", score=round(rate * 5, 2))


def register(registry: VerifierRegistry) -> None:
    registry.register("orders_db_metadata", _orders_db_metadata, layer="data", owner="data-platform",
                      description="Orders cluster replica lag and storage headroom")
    registry.register("ci_pipeline_status", _ci_status, layer="experience", owner="platform-eng",
                      description="Staging pass rate for the change under review (>= 95%)")
    registry.register("payments_api_health", http_health("https://payments.internal/healthz", timeout=2.0),
                      layer="data", owner="payments", description="Payments API health endpoint returns 200")
    registry.register("error_budget", metric_threshold(lambda c: 0.31, "<", 0.8, "error budget burn"),
                      layer="experience", owner="sre", description="Error-budget burn below 80%")
```

### Core Governance Interceptor: the Decision Matrix (`src/wekid/bridge.py`, excerpt)
Gates and floors are evaluated first, then the matrix rows in order. A high Maturity Score can never override a failure beneath it.

```python
# ---- Decision Matrix --------------------------------------------------
s = scores.maturity_score
sampling = False

def govern(ts: List[Trigger]) -> None:
    for t in ts:
        t.governing = True

if hard:
    outcome, ceiling = Outcome.REJECTED, AL.AL0
    govern(hard)
elif no_ceiling:
    outcome, ceiling = Outcome.INSUFFICIENT_MATURITY, AL.AL0
    govern(no_ceiling)
elif s < p.insufficient_below:
    outcome, ceiling = Outcome.INSUFFICIENT_MATURITY, AL.AL0
    t = Trigger("band:insufficient_maturity", "band", f"maturity score {s:.1f} < {p.insufficient_below:.0f}", True)
    other.append(t)
else:
    if floors:
        outcome, ceiling = Outcome.CONSTRAINED, AL.AL2
        govern(floors)
    elif s < p.remediation_below or soft:
        outcome, ceiling = Outcome.REMEDIATION_REQUIRED, AL.AL1
        if soft:
            govern(soft)
        if s < p.remediation_below:
            other.append(Trigger("band:remediation_required", "band",
                                 f"maturity score {s:.1f} in {p.insufficient_below:.0f}-{p.remediation_below - 1:.0f}",
                                 not soft))
    elif s < p.autonomous_from:
        outcome, ceiling, sampling = Outcome.APPROVED_WITH_MONITORING, AL.AL3, True
        other.append(Trigger("band:approved_with_monitoring", "band",
                             f"maturity score {s:.1f} in {p.remediation_below:.0f}-{p.autonomous_from - 1:.0f}", True))
    else:
        outcome, ceiling = Outcome.APPROVED_AUTONOMOUS, AL.AL3
    if req.oversight.decayed:
        outcome = Outcome.OVERSIGHT_DECAY
        ceiling = min(ceiling, AL.AL2)
        sampling = False
        other.append(Trigger("oversight:decay", "oversight",
                             "engagement gating active and the human loop is not demonstrably exercised", True))

# Self-declared evidence cannot support the upper tiers.
self_declared = [n for n, ls in L.items() if ls.provenance == Provenance.SELF_DECLARED]
if self_declared and ceiling > p.self_declared_ceiling:
    ceiling = p.self_declared_ceiling
    sampling = False
    other.append(Trigger("provenance:self_declared_cap", "provenance",
                         f"self-declared evidence ({', '.join(sorted(self_declared))}) caps the ceiling at "
                         f"{p.self_declared_ceiling.code}", True))
maturity_outcome = outcome

# ---- stakes floor (Model Two) ----------------------------------------
c = req.consequence
if c.risk_level not in p.stakes_cap_by_risk:
    raise InvalidRequest(f"consequence.risk_level '{c.risk_level}' not in {sorted(p.stakes_cap_by_risk)}")
reasons: List[str] = []
if c.safety_critical:
    reasons.append("safety-critical")
if c.legally_reserved:
    reasons.append("legally reserved")
if c.decision_class and c.decision_class in p.reserved_decision_classes:
    reasons.append(f"decision class '{c.decision_class}' reserved by policy")
if p.reserve_irreversible_high_risk and not c.reversible and c.risk_level == "high":
    reasons.append("irreversible and high-risk")
stakes: List[Trigger] = []
if reasons:
    cap = AL.AL4
    stakes.append(Trigger("stakes:policy_reserved", "stakes", "; ".join(reasons), True))
else:
    cap = p.stakes_cap_by_risk[c.risk_level]
    if cap < AL.AL3:
        stakes.append(Trigger(f"stakes:risk:{c.risk_level}", "stakes", f"{c.risk_level} risk caps at {cap.code}"))
    if not c.reversible and p.irreversible_cap < cap:
        cap = p.irreversible_cap
        stakes.append(Trigger("stakes:irreversible", "stakes", f"irreversible action caps at {cap.code}"))
    if policy:
        for cond, lvl in policy.caps.items():
            lvl = AL.parse(lvl)
            if _fired(cond, req, scores, p) and lvl < cap:
                cap = lvl
                stakes.append(Trigger(f"stakes:policy_cap:{cond}", "stakes", f"policy caps at {lvl.code}"))

# ---- resolution -------------------------------------------------------
if cap == AL.AL4:
    final, binding, outcome = AL.AL4, BindingLimit.POLICY_RESERVATION, Outcome.POLICY_RESERVED
    sampling = False
else:
    final = min(ceiling, cap)
    if ceiling < cap:
        binding = BindingLimit.MATURITY_CEILING
    elif cap < ceiling:
        binding = BindingLimit.STAKES_FLOOR
        sampling = False
        for t in stakes:
            t.governing = True
    else:
        binding = BindingLimit.BOTH
    outcome = maturity_outcome
```

### Evidence Commit: S3 Object Lock (`src/wekid/vault.py`, excerpt)
Records are fingerprinted with SHA-256 over canonical JSON, uploaded with the integrity checksum that Object Lock requires, and optionally given a per-object COMPLIANCE retain-until date.

```python
    def commit(self, record: Dict[str, Any]) -> Dict[str, Any]:
        rec = dict(record)
        rec["fingerprint"] = compute_fingerprint(rec)
        body = json.dumps(rec, indent=2, sort_keys=True, ensure_ascii=False).encode("utf-8")
        params: Dict[str, Any] = {
            "Bucket": self.bucket,
            "Key": self._key(rec),
            "Body": body,
            "ContentType": "application/json",
            "ChecksumAlgorithm": "SHA256",
            "ChecksumSHA256": base64.b64encode(hashlib.sha256(body).digest()).decode("ascii"),
            "Metadata": {"wekid-fingerprint": rec["fingerprint"], "wekid-schema": EVIDENCE_SCHEMA},
        }
        if self.retention_days:
            params["ObjectLockMode"] = "COMPLIANCE"
            params["ObjectLockRetainUntilDate"] = utc_now() + _dt.timedelta(days=int(self.retention_days))
        if self.kms_key_id:
            params["ServerSideEncryption"] = "aws:kms"
            params["SSEKMSKeyId"] = self.kms_key_id
        try:
            self.client.put_object(**params)
        except Exception as e:
            raise EvidenceCommitError(f"S3 evidence write failed: {type(e).__name__}: {e}") from e
        rec["location"] = f"s3://{self.bucket}/{params['Key']}"
        return rec
```

### Fail-Closed Gate (`src/wekid/gate.py`, excerpt)
The determination is committed before authority is returned; commit failures are retried and then raise `EvidenceCommitError`. `permits()` never grants authority to an uncommitted determination.

```python
    def _write(self, record: Dict[str, Any]) -> Dict[str, Any]:
        last: Optional[Exception] = None
        for attempt in range(self.commit_retries + 1):
            try:
                return self.vault.commit(record)
            except Exception as e:  # retry, then fail closed
                last = e
                log.error("evidence commit failed (attempt %d/%d): %s", attempt + 1, self.commit_retries + 1, e)
                if attempt < self.commit_retries:
                    time.sleep(self.retry_backoff_s * (2 ** attempt))
        raise EvidenceCommitError(f"evidence record {record.get('record_id')} not committed; "
                                  f"authority withheld: {last}") from last

    def _commit(self, req: AssessmentRequest, det: Determination, extra: Optional[Dict[str, Any]] = None) -> None:
        d = det.to_dict()
        for k in ("record_id", "committed", "fingerprint"):
            d.pop(k, None)
        body = {"request": request_to_dict(req, self.redact_content), "determination": d}
        if extra:
            body.update(extra)
        stored = self._write(self._envelope("determination", det.record_id, body))
        det.committed = True
        det.fingerprint = stored.get("fingerprint")

    # ------------------------------------------------------------------ enforcement
    def permits(self, det: Determination, action: Optional[ActionSpec] = None) -> bool:
        """Does this determination permit the action to run *now*, without a human?"""
        if not det.committed:
            return False
        al = det.authority_level
        if al == AuthorityLevel.AL3:
            return True
        if al == AuthorityLevel.AL1:
            if action is None or not action.reversible:
                return False
            if self.profile.al1_impact_cap and action.impact > self.profile.al1_impact_cap:
                return False
            if self.profile.al1_allowlist and action.name not in self.profile.al1_allowlist:
                return False
            return True
        return False  # AL-0 informational, AL-2 needs release, AL-4 reserved to a human

    def guard(self, req: AssessmentRequest, action: Optional[ActionSpec] = None) -> Determination:
        """Assess and raise unless the action may run now.

        Raises ``HumanReleaseRequired`` for AL-2 / AL-4 (stage it for a human),
        ``AuthorityDenied`` otherwise, ``EvidenceCommitError`` if the record
        could not be written.
        """
        det = self.assess(req)
        if self.permits(det, action):
            return det
        if det.authority_level in (AuthorityLevel.AL2, AuthorityLevel.AL4):
            raise HumanReleaseRequired(f"{det.authority_level.code}: human release required "
                                       f"(record {det.record_id})", det)
        raise AuthorityDenied(f"{det.authority_level.code} {det.outcome.value}: action not permitted "
                              f"(record {det.record_id})", det)

    def governed(self, build_request: Callable[..., AssessmentRequest],
                 action: Optional[Callable[..., ActionSpec]] = None):
        """Decorator: run the function only if the gate permits it.

            @gate.governed(lambda order_id, amount: make_request(order_id, amount),
                           action=lambda order_id, amount: ActionSpec("refund", True, amount))
            def refund(order_id, amount): ...
        """
        def deco(fn):
            @functools.wraps(fn)
            def wrapper(*args, **kwargs):
                req = build_request(*args, **kwargs)
                spec = action(*args, **kwargs) if action else None
                self.guard(req, spec)
                return fn(*args, **kwargs)
            return wrapper
        return deco
```

---

## 5. End-to-End Runtime Blueprint (FastAPI Integration)
The original Section 5 endpoint, rebuilt on the SDK (`integrations/fastapi_governed_action.py`). Verifiers are registered and frozen at start-up, every claim is bound to a verifier of its own layer, the decision is committed before anything executes, and the response distinguishes execute (200), staged for human release, denied (403) and evidence-not-committed (503).

```python
# SPDX-License-Identifier: Apache-2.0
# Copyright 2025-2026 WEKID LLC
"""
integrations/fastapi_governed_action.py — embed the gate in your own FastAPI service.

This is the ASK v1.2 manifest's Section 5 endpoint, rebuilt on the SDK:

* verifiers are registered at start-up and the registry is frozen;
* every claim is bound to a verifier of the same layer;
* the determination is committed to the evidence vault *after* the decision and
  *before* anything executes; a failed commit returns 503 and nothing runs;
* the response distinguishes 200 (execute), 202 (human release) and 403 (denied).

Run:  uvicorn fastapi_governed_action:app --app-dir integrations --port 8100
(needs: pip install 'wekid-sdk[server]'; set WEKID_VAULT=s3 + WEKID_CORPUS_BUCKET for WORM storage)
"""

from __future__ import annotations

import os
import sys

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "examples"))

from wekid import (ActionSpec, AuthorityDenied, EvidenceCommitError, HumanReleaseRequired, VerifierRegistry,
                   WekidGate, vault_from_env)

import custom_verifiers  # your verifiers: register(registry)

registry = VerifierRegistry()
custom_verifiers.register(registry)
registry.freeze()
gate = WekidGate(registry=registry, vault=vault_from_env())

app = FastAPI(title="Governed actions")


class GovernedSQL(BaseModel):
    request_id: str
    sql_query: str
    cluster: str
    change_ref: str
    reviewed_wisdom: dict       # signals from your independent review step
    reviewed_information: dict  # signals from your grounding checks


def run_sql(sql: str) -> None:  # your executor
    print("EXECUTING:", sql)


@app.post("/api/v1/execute-governed-action")
def execute_action(body: GovernedSQL):
    request = gate.parse({
        "output": {"output_id": body.request_id, "output_type": "tool_call", "content": body.sql_query},
        "consequence": {"risk_level": "medium", "domain": "data_operations"},
        "claims": [
            {"claim_id": "C1", "layer": "knowledge", "verifier": "sql_safety", "content": body.sql_query},
            {"claim_id": "C2", "layer": "data", "verifier": "orders_db_metadata", "content": body.cluster},
            {"claim_id": "C3", "layer": "experience", "verifier": "ci_pipeline_status", "content": body.change_ref},
        ],
        "evidence": {
            "wisdom": {"provenance": "measured", "evaluator_id": "review-panel", "signals": body.reviewed_wisdom},
            "information": {"provenance": "measured", "evaluator_id": "grounding", "signals": body.reviewed_information},
        },
    })
    try:
        det = gate.guard(request, action=ActionSpec("sql_write", reversible=False))
    except EvidenceCommitError as e:
        raise HTTPException(status_code=503, detail=f"evidence not committed; nothing executed: {e}")
    except HumanReleaseRequired as e:
        return {"status": "STAGED", "authority_level": e.determination.authority_level.code,
                "record_id": e.determination.record_id}
    except AuthorityDenied as e:
        d = e.determination
        raise HTTPException(status_code=403, detail={
            "authority_level": d.authority_level.code, "outcome": d.outcome.value, "record_id": d.record_id,
            "governing_triggers": [t.code for t in d.triggers if t.governing]})
    run_sql(body.sql_query)
    return {"status": "EXECUTED", "authority_level": det.authority_level.code, "record_id": det.record_id}
```

### Request Contract (`POST /v1/authorize`)
The same JSON shape is accepted by `WekidGate.assess()`, the HTTP service, the client and `wekid assess`. Evidence for layers not covered by claims comes from measured sources such as grounding checks and independent review.

```json
{
  "output": {
    "output_id": "SQL-7781",
    "output_type": "tool_call",
    "content": "UPDATE accounts SET status = 'closed' WHERE account_id = 4417"
  },
  "consequence": {
    "risk_level": "medium",
    "domain": "data_operations",
    "reversible": true
  },
  "claims": [
    {
      "claim_id": "C1",
      "layer": "knowledge",
      "verifier": "sql_safety",
      "content": "UPDATE accounts SET status = 'closed' WHERE account_id = 4417"
    },
    {
      "claim_id": "C2",
      "layer": "data",
      "verifier": "orders_db_metadata",
      "content": "accounts cluster"
    },
    {
      "claim_id": "C3",
      "layer": "experience",
      "verifier": "ci_pipeline_status",
      "content": "change 8812"
    }
  ],
  "evidence": {
    "information": {
      "provenance": "measured",
      "evaluator_id": "grounding-checker",
      "signals": {
        "context_accuracy": 4.3,
        "relevance": 4.4,
        "faithfulness": 4.2,
        "completeness": 4.0
      }
    },
    "wisdom": {
      "provenance": "measured",
      "evaluator_id": "ops-review-panel",
      "signals": {
        "authority_calibration": 4.2,
        "contextual_awareness": 4.1,
        "consequence_awareness": 4.3,
        "ethical_alignment": 4.6,
        "judgment_under_ambiguity": 4.0,
        "tradeoff_optimization": 4.0,
        "restraint_escalation": 4.2
      }
    }
  }
}
```

### Determination (abridged)
Generated by running the request above through the SDK with the example verifiers. Enforce `authority_level`; never act when `committed` is false.

```json
{
  "record_id": "wkd-20261025T141511Z-eca014903e49",
  "authority_level": "AL-3",
  "authority_title": "Augmented Authority",
  "outcome": "APPROVED_AUTONOMOUS",
  "maturity_ceiling": "AL-3",
  "stakes_cap": "AL-3",
  "binding_limit": "both",
  "maturity_score": 91.15,
  "enhanced_sampling": false,
  "committed": true,
  "fingerprint": "25edf01934941de9613e7c581474bea4a4f98ea80c5dea7336d2acae763a4ee6",
  "permissions": {
    "may_execute_autonomously": true,
    "requires_human_release": false,
    "constrained_only": false,
    "informational_only": false,
    "reserved_to_human": false
  },
  "claims": [
    {
      "claim_id": "C1",
      "verifier": "sql_safety",
      "valid": true,
      "reason": "UPDATE statement passed static safety checks"
    },
    {
      "claim_id": "C2",
      "verifier": "orders_db_metadata",
      "valid": true,
      "reason": "replica lag 1.4s, storage 62%"
    },
    {
      "claim_id": "C3",
      "verifier": "ci_pipeline_status",
      "valid": true,
      "reason": "48/50 staging runs passed"
    }
  ],
  "rationale": "Maturity score 91.2; matrix outcome APPROVED_AUTONOMOUS; maturity ceiling AL-3. Stakes cap AL-3. Authority AL-3 Augmented Authority (bound by both)."
}
```

### Service Configuration
Set `WEKID_VAULT=s3` with `WEKID_CORPUS_BUCKET`, `WEKID_RETENTION_DAYS` and optionally `WEKID_KMS_KEY_ID` for WORM storage; `WEKID_API_KEYS` for bearer authentication; `WEKID_VERIFIER_MODULES` to load verifiers; and `WEKID_PROFILE` for an approved calibration profile. Run with `wekid serve` or `uvicorn wekid.api:app`. Full installation steps and every method are in `INSTALL.md` and `docs/SDK_REFERENCE.md`.

---

© 2026 WEKID LLC. All rights reserved. WEKID™ is a trademark of WEKID LLC.