coulomb/maturity-engine
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Stand up the Engine/PIP surface for MAT-WP-0001

Browse source 
Declare layer.yaml, add a Python engine over a local SQLite store,
and cover deterministic assessment, the §13 gap register, stance-map
inventory, claim guardrails, and the gate-house review path with tests.

Assistant: grok
Assistant-Session: 01a04ceb-150e-7e80-a542-ec8b1372e164
This commit is contained in:
tegwick 2026-08-29 12:54:37 +02:00
parent 5c052ed106
commit 4cde4e489a
31 changed files with 2498 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

15
.gitignore vendored Normal file
View file

@ -0,0 +1,15 @@
.claude/*
!.claude/rules/
!.claude/rules/*.md
__pycache__/
.pytest_cache/
.ruff_cache/
build/
dist/
*.egg-info/
.venv/
venv/
data/
*.sqlite
*.sqlite-journal

56
AGENTS.md Normal file
View file

@ -0,0 +1,56 @@
# maturity-engine — Agent Instructions
## Repo Identity
**Purpose:** Compute, deterministically, how far a subject has progressed
against declared criteria and submitted evidence, and remember that
progression. Gap register and capability readiness live here.
**Layer:** Engine (PIP). NetKingdom Security Layer Model v0.7. Never render
or cache an authorization decision. Never compile a level into registry
content. Never let a consumer branch on a fetched level — emit a claim.
**Domain:** infotech
**Repo slug:** maturity-engine
**Workplan prefix:** `MAT-WP-`
Machine-readable declaration: `layer.yaml`, checked by
`scripts/check_layer_conformance.py`. Frontmatter in `INTENT.md` must agree.
---
## Stack and commands
Python ≥ 3.12. Persistence is this PIP's own SQLite store, not OpenBao.
```bash
python3 -m pip install -e ".[dev]"
make test
make check-layer
make lint
PYTHONPATH=src python3 -m maturity_engine bootstrap --at 2026-08-29T12:00:00Z
```
Default database: `data/maturity-engine.sqlite` (gitignored) or
`MATURITY_ENGINE_DB`.
## Module boundaries
| Module | Owns |
| --- | --- |
| `models.py` | ladders, levels, criteria, evidence, assessments, gaps |
| `compute.py` | deterministic level computation |
| `store.py` | SQLite + local outbox |
| `scoring.py` | four conformance states; blocked-clean not below conforming |
| `claims.py` | PIP claim shape and guardrail rejections |
| `seed.py` | §13 snapshot, §13.1 inventory, first two ladders as data |
| `engine.py` | facade; no decide/authorize/may |
| `cli.py` | inspect and compute; does not decide |
Ladder *content* is registered data. ASM-0…ASM-6 is `gate-house`'s.
PEP-stance publication is `ops-warden`'s.
## State Hub
API: `http://127.0.0.1:8000`. Log progress at session close. Update task
status when closing a workplan task (`todo` / `progress` / `done`).

7
INTENT.md
View file

@ -28,9 +28,10 @@ companion: net-kingdom/SECURITY-COMPANION.md
> **Catalog entry (v0.7 §4):** graded progression against declared criteria and
> evidence; the gap register; capability readiness (§9.5).
>
> The machine-readable form required by §11 is `INTENT.md` frontmatter now, and
> `layer.yaml` once T01 of the alignment workplan lands. Prose cannot
> distinguish a declaration from a transcribed review; the file is the surface.
> The machine-readable form required by §11 is `layer.yaml`, checked by
> `scripts/check_layer_conformance.py`. `INTENT.md` frontmatter must agree.
> Prose cannot distinguish a declaration from a transcribed review; the file
> is the surface.
>
> Contest this layer if it is wrong — companion §2 asks for that correction
> rather than a polite label.

15
Makefile Normal file
View file

@ -0,0 +1,15 @@
PYTHON ?= python3
.PHONY: install-dev test lint check-layer
install-dev:
$(PYTHON) -m pip install -e ".[dev]"
test:
PYTHONPATH=src $(PYTHON) -m pytest -q
lint:
$(PYTHON) -m ruff check src tests scripts
check-layer:
$(PYTHON) scripts/check_layer_conformance.py --report

12
README.md
View file

@ -29,5 +29,13 @@ Maturity here is graded, evidence-based, open-ended, and revisable — levels ca
fall when evidence expires. The closed, binary, replay-proof state machine of an
approval belongs to `approval-engine`; the two engines are deliberate opposites.
This repository is seeded, not yet an engine surface. See [INTENT.md](INTENT.md)
and [SCOPE.md](SCOPE.md). The work to stand up the surface is `MAT-WP-0001`.
See [INTENT.md](INTENT.md) and [SCOPE.md](SCOPE.md). The claim contract is
[docs/claim-contract.md](docs/claim-contract.md).
```bash
python3 -m pip install -e ".[dev]" # or: PYTHONPATH=src
make test
make check-layer
PYTHONPATH=src python3 -m maturity_engine bootstrap --at 2026-08-29T12:00:00Z
PYTHONPATH=src python3 -m maturity_engine review --subject kings-guard --at 2026-08-29T12:00:00Z
```

97
SCOPE.md
View file

@ -8,28 +8,32 @@
## One-liner
Seeded NetKingdom Engine (PIP) for deterministic maturity assessment, the
estate gap register, and capability readiness — declared in `INTENT.md`,
not yet implemented.
NetKingdom Engine (PIP) that computes a maturity level from declared
criteria and submitted evidence, holds the estate gap register and PEP
stance-map inventory, and emits levels only as claims.
---
## Current Responsibility
`maturity-engine` currently owns the **declaration** of the Engine/PIP
boundary assigned by NetKingdom Security Layer Model v0.7 §4 and restated
in `INTENT.md`: graded progression against declared criteria and evidence,
the gap register, and capability readiness.
`maturity-engine` owns the Engine/PIP surface assigned by NetKingdom
Security Layer Model v0.7 §4: graded progression, the gap register, and
capability readiness. It declares that layer in `INTENT.md` frontmatter
and in `layer.yaml`.
It does **not** yet own a running engine, a store, an API, a register that
can be queried, or a claim contract `access-engine` can consume. Statute
§13 and §13.1 still hold the gap-register snapshot and the PEP stance-map
inventory because this repository cannot store state.
The current implementation is an in-process Python library and CLI over
this PIP's own SQLite store. It registers ladders as data, accepts
evidence with validity windows, computes a deterministic explainable
level, records progression history (including demotion on expiry),
serves the §13 gap-register snapshot, inventories PEP stance maps,
answers readiness as pending / declared-gap / surface-exists, and
emits a `maturity-level` claim. Assessment and register mutations are
queued in a local outbox in the same transaction; nothing calls
`audit-core` inside that transaction.
This repository now **declares** itself Engine, role PIP, in `INTENT.md`
frontmatter against security-layer-model v0.7 and companion v0.2. The
machine-readable `layer.yaml` form, a total client account, and the
engine surface itself are not yet evidenced here.
It does **not** decide whether an actor may act. It does not compile a
level into a registry. It does not author ladder criteria, observation,
or actuation.
---
@ -120,20 +124,29 @@ engine surface itself are not yet evidenced here.
## Current State
- Status: seeded, not implemented
- On disk: `INTENT.md` (v0.7-aligned declaration) and `README.md`
- Layer declaration: `INTENT.md` frontmatter declares `layer: Engine`,
`role: PIP` against security-layer-model v0.7; `layer.yaml` and a
conformance check are not yet present
- Engine surface: none — no API, store, tests, or runtime
- Gap register: still the statute §13 snapshot
- Stance-map inventory: still statute §13.1 (one published map, one absence)
- Claim contract with `access-engine`: not written
- Evidence emission: not present
- Registered ladders: none
- Layout conformance (ITC-REPO-LAYOUT 0.1.0-RC1): **minimal** once this
`SCOPE.md` exists; `history/` and `workplans/` are in use as defined,
`demand/` and `docs/` are not claimed
- Status: engine surface present; production deploy is not claimed
- Layer declaration: `INTENT.md` frontmatter and `layer.yaml` declare
`layer: Engine` / `role: PIP` against security-layer-model v0.7;
`scripts/check_layer_conformance.py` checks them
- Runtime: Python package `maturity_engine` over SQLite (`make test` is
the documented command)
- Assessment: named versioned ladders, evidence with expiry, deterministic
compute, explainability, progression history
- Gap register: statute §13 snapshot is queryable here, with `state` and
owner-status intact; blocked-clean is not scored below conforming
- Stance-map inventory: `ops-warden` published, `ops-mason` unpublished
- Readiness: pending / declared-gap / surface-exists; actuation stays
pending and is not owned here
- Claim contract: `docs/claim-contract.md`; guardrail tests reject a
registry compile and a consumer branch
- Evidence emission: local outbox; load-bearing assessments; heartbeat
for rare classes; no synchronous `audit-core` dependency
- Registered ladders: `asm` v0.3 (owner `gate-house`) and
`pep-stance-publication` v0.1 (owner `ops-warden`)
- Review path: `Engine.review` / `maturity-engine review` computes and
remembers; it does not judge
- Layout conformance (ITC-REPO-LAYOUT 0.1.0-RC1): **minimal**; `docs/`,
`history/`, and `workplans/` are in use; `demand/` is not claimed
---
@ -148,9 +161,8 @@ Declared conformance: **`minimal`** under ITC-REPO-LAYOUT 0.1.0-RC1.
Intentional omissions:
- `demand/`, `docs/`, `research/`, `spec/`, `wiki/`, and `issues/` are not
currently claimed. They should be added only when their distinct semantics
are needed, not as empty structural decoration.
- `docs/` holds the claim contract. `demand/`, `research/`, `spec/`,
`wiki/`, and `issues/` are not currently claimed.
- Finished workplan archival, when it exists, will follow the Custodian
ADR-001 convention at `workplans/archived/YYMMDD-...` so State Hub
discovery remains deterministic.
@ -178,17 +190,26 @@ Intentional omissions:
- Operative form: `net-kingdom/SECURITY-COMPANION.md`
- Alignment review:
`history/260829-demand-netkingdom-security-layer-alignment.md`
- Claim contract: `docs/claim-contract.md`
- Layer declaration: `layer.yaml`
- Active work: `workplans/`
- Tests: `make test`
---
## Gap to Intent
The seed declaration is in place. Nothing `INTENT.md` says this engine
computes, remembers, or inventories is evidenced on disk yet. The work to
close that gap is `MAT-WP-0001`, sourced from
`history/260829-demand-netkingdom-security-layer-alignment.md`.
`MAT-WP-0001` closed the seed-to-engine gap: a queryable Engine/PIP exists.
Remaining aspiration, not a hole in this surface:
- `gate-house` actually performing live conformance review through the
CLI rather than in tests;
- ASM-0…ASM-6 assessed against a real estate with evidence fed by the
systems being assessed, not fixtures;
- `access-engine` consuming the claim in a live decision record;
- draining the local outbox into `audit-core` (the queue is here; the
drain adapter is not);
- a production SQLite (or later catalogued) deployment.
Do not host a PDP, compile a level into a registry, score blocked-clean
below conforming, or treat observation or containment as available while
closing it.
below conforming, or treat observation or containment as available.

56
docs/claim-contract.md Normal file
View file

@ -0,0 +1,56 @@
# Maturity level claim contract
`maturity-engine` is a PIP. A computed level leaves this engine only as a
**request claim** (or as a versioned policy rule authored elsewhere). It is
not an authorization decision.
Statute: `security-layer-model` v0.7 §6.2, §9.5.
## Claim shape
```json
{
"kind": "maturity-level",
"issuer": "maturity-engine",
"subject": "ops-warden",
"model_id": "asm",
"model_version": "0.3",
"level": 1,
"level_id": "ASM-1",
"assessed_at": "2026-08-29T12:00:00Z",
"assessment_id": "<uuid5 of canonical inputs>",
"freshness_rule": "assessment evaluated_at plus limiting evidence valid_until",
"digest": "<sha-256 of the claim without this field>"
}
```
`access-engine` consumes this object as an input claim. Reconstructability
is from the decision record that names the claim digest, not from a registry
row.
## Forbidden
- Compiling a level into registry content. Until `access-engine` decision
provenance carries a registry-snapshot digest, a level that reached a
decision through the registry is not reconstructable. The engine raises
`GuardrailError` on `compile_into_registry`.
- A consumer branching on a fetched level (`if level >= 3: allow`). That is
a second decision point. The engine raises `GuardrailError` on
`gate_on_level`.
- Any `authorize` / `decide` / `may` surface on this engine.
## Evidence bound
Assessment records are **load-bearing** once consumed as claims: emission is
queued in this engine's local SQLite outbox in the same transaction as the
state change. No synchronous `audit-core` call sits inside that transaction.
The archive proves records were not altered or truncated after arrival. It
does not prove an event never sent. Absence of a record is not evidence of
non-occurrence.
Rare load-bearing classes use a **heartbeat** (`maturity-engine heartbeat`
via `Engine.heartbeat`), not a rate.
Gap-register mutations are **attributive** unless a control's soundness
depends on their presence.

70
layer.yaml Normal file
View file

@ -0,0 +1,70 @@
# maturity-engine — NetKingdom security layer declaration
#
# Framework: net-kingdom/canon/standards/security-layer-model_v0.7.md
# Assent: INTENT.md frontmatter (this repository's own voice, §11)
# Validate: python3 scripts/check_layer_conformance.py
#
# §11 requires a machine-readable declaration: prose cannot distinguish a
# declaration from a transcribed review. Form adapted from ops-warden's
# reference and kings-guard's no-contact Engine-adjacent shape.
#
# This repository is an Engine (PIP). Computing a level is not a protected
# side effect, so there is no pep_stance path. Catalogued Tooling (OpenBao,
# key-cape) is not contacted. Persistence is this PIP's own SQLite store.
schema_version: "0.1"
framework: netkingdom-security-layer-model
standard_version: "0.7"
repository: maturity-engine
layer: engine
role: pip
declared_by: INTENT.md
declared_at: "2026-08-29"
# §4 catalog entry, transcribed so drift between the catalog and this file is
# visible. The standard is authoritative for the row; this records what we
# understand ourselves to have been assigned.
catalog_entry:
owns:
- graded progression against declared criteria and evidence
- the gap register
- capability readiness
does_not_own:
- authorization decisions
- approval objects
- observation in production
- actuation / containment
# §5 / §11: every direct contact with a Tooling-layer system (a §4 Tooling row).
# Empty is a claim. This is an Engine API over state it owns, not Staff
# reaching into OpenBao.
tooling_contacts: []
# §11 requires non-Tooling clients to be recorded so the check is total.
# Neither target is a §4 Tooling row. Recorded, not policed as §5.
non_tooling_clients:
- id: state-hub-work-records
target: state-hub
layer: not-catalogued
operation: "HTTP to the Custodian State Hub for work records and progress events"
write: true
note: >-
Outside §5 by the v0.5 scope rule: "Tooling-layer system" means a §4
Tooling row, and state-hub is not one. Carries no security authority
and no secret payload.
- id: sqlite-own-store
target: sqlite
layer: not-catalogued
operation: "this PIP's own transactional store (assessments, register, local outbox)"
write: true
note: >-
Engine-owned persistence, not Lifecycle over catalogued Tooling.
Consumers read claims through this engine's API, not by opening the
file. Listed so the client account is total. Review: 2027-02-28
(two intervals from 2026-08-29).
# No PEP stance: this engine does not cause a protected side effect.
declared_shapes:
"5.1": []
"5.2": []
"5.3": []

40
pyproject.toml Normal file
View file

@ -0,0 +1,40 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "maturity-engine"
version = "0.1.0"
description = "Deterministic maturity assessment, gap register, and capability readiness for NetKingdom."
readme = "README.md"
requires-python = ">=3.12"
authors = [{ name = "Coulomb" }]
dependencies = []
[project.optional-dependencies]
dev = [
"pytest>=7.4,<9.0",
"ruff>=0.6,<1.0",
"pyyaml>=6.0,<7.0",
]
[project.scripts]
maturity-engine = "maturity_engine.cli:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
addopts = [
"--strict-markers",
"--tb=short",
]
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "B"]

142
scripts/check_layer_conformance.py Normal file
View file

@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Check maturity-engine against the NetKingdom security layer model (§5, §11).
This is an Engine (PIP). The checkable claims:
- layer.yaml declares layer=engine, role=pip
- INTENT.md frontmatter agrees (case-insensitive)
- no pep_stance path
- no catalogued Tooling client (OpenBao, key-cape, cluster)
- sqlite3 is this PIP's own store and is allowed
Exit 0 clean, 1 undeclared Tooling contact, 2 declaration malformed.
"""
from __future__ import annotations
import argparse
import ast
import re
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src" / "maturity_engine"
DECL = ROOT / "layer.yaml"
INTENT = ROOT / "INTENT.md"
TOOLING_IMPORTS = {
"hvac": "OpenBao / Vault client",
"bao": "OpenBao client",
"kubernetes": "cluster client",
"kubernetes_asyncio": "cluster client",
"ldap3": "direct LDAP client (key-cape tooling)",
"python_ldap": "direct LDAP client (key-cape tooling)",
}
def load_declaration() -> dict:
if not DECL.exists():
print(f"FAIL: no declaration at {DECL.relative_to(ROOT)} (§11)", file=sys.stderr)
raise SystemExit(2)
try:
data = yaml.safe_load(DECL.read_text())
except yaml.YAMLError as exc:
print(f"FAIL: {DECL.name} is not parseable: {exc}", file=sys.stderr)
raise SystemExit(2) from exc
for key in ("layer", "role", "repository", "tooling_contacts", "standard_version"):
if key not in data:
print(f"FAIL: {DECL.name} missing required key '{key}' (§11)", file=sys.stderr)
raise SystemExit(2)
if str(data["layer"]).lower() != "engine":
print(f"FAIL: declared layer is {data['layer']!r}, expected 'engine'", file=sys.stderr)
raise SystemExit(2)
if str(data["role"]).lower() != "pip":
print(f"FAIL: declared role is {data['role']!r}, expected 'pip'", file=sys.stderr)
raise SystemExit(2)
if data.get("pep_stance"):
print("FAIL: pep_stance is set; this engine is not PEP-shaped", file=sys.stderr)
raise SystemExit(2)
return data
def intent_frontmatter() -> dict:
text = INTENT.read_text()
match = re.match(r"^---\n(.*?)\n---\n", text, re.DOTALL)
if not match:
print("FAIL: INTENT.md has no YAML frontmatter (§11)", file=sys.stderr)
raise SystemExit(2)
meta = yaml.safe_load(match.group(1))
if not isinstance(meta, dict):
print("FAIL: INTENT.md frontmatter is not a mapping", file=sys.stderr)
raise SystemExit(2)
return meta
def imported_modules(path: Path) -> set[str]:
try:
tree = ast.parse(path.read_text())
except SyntaxError:
return set()
found: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
found.update(alias.name.split(".")[0] for alias in node.names)
elif isinstance(node, ast.ImportFrom):
if node.level == 0 and node.module:
found.add(node.module.split(".")[0])
return found
def scan() -> list[tuple[Path, str, str]]:
hits: list[tuple[Path, str, str]] = []
for path in sorted(SRC.rglob("*.py")):
for module in sorted(imported_modules(path)):
if module in TOOLING_IMPORTS:
hits.append((path, module, TOOLING_IMPORTS[module]))
return hits
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--report", action="store_true")
args = parser.parse_args()
decl = load_declaration()
intent = intent_frontmatter()
if str(intent.get("layer", "")).lower() != str(decl["layer"]).lower():
print(
f"FAIL: INTENT.md layer {intent.get('layer')!r} != layer.yaml {decl['layer']!r}",
file=sys.stderr,
)
return 2
if str(intent.get("role", "")).lower() != str(decl["role"]).lower():
print(
f"FAIL: INTENT.md role {intent.get('role')!r} != layer.yaml {decl['role']!r}",
file=sys.stderr,
)
return 2
hits = scan()
if hits:
print("FAIL: catalogued Tooling-layer client in an Engine that does not own it", file=sys.stderr)
for path, module, what in hits:
print(f" {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr)
return 1
if args.report:
print(
f"maturity-engine — layer {decl['layer']}, role {decl['role']}, "
f"standard v{decl['standard_version']}"
)
print(f" tooling contacts: {len(decl.get('tooling_contacts') or [])}")
print(f" non-tooling clients: {len(decl.get('non_tooling_clients') or [])}")
print(" pep_stance: none")
else:
print(f"OK: Engine/PIP declaration holds; no catalogued Tooling client in {SRC.relative_to(ROOT)}")
return 0
if __name__ == "__main__":
raise SystemExit(main())

21
src/maturity_engine/__init__.py Normal file
View file

@ -0,0 +1,21 @@
"""Deterministic maturity assessment, gap register, and capability readiness."""
from maturity_engine.claims import Claim
from maturity_engine.engine import Engine
from maturity_engine.errors import GuardrailError, ModelError, UnevaluableCriterion
from maturity_engine.models import Assessment, Criterion, Evidence, Gap, Ladder, Level, StanceMap
__all__ = [
"Assessment",
"Claim",
"Criterion",
"Engine",
"Evidence",
"Gap",
"GuardrailError",
"Ladder",
"Level",
"ModelError",
"StanceMap",
"UnevaluableCriterion",
]

3
src/maturity_engine/__main__.py Normal file
View file

@ -0,0 +1,3 @@
from maturity_engine.cli import main
raise SystemExit(main())

82
src/maturity_engine/claims.py Normal file
View file

@ -0,0 +1,82 @@
"""PIP claim shape for a maturity level.
A claim is a fact `access-engine` may consume. This module does not decide
whether an action is permitted.
"""
from __future__ import annotations
from dataclasses import dataclass
from maturity_engine.errors import GuardrailError
from maturity_engine.ids import digest
from maturity_engine.models import Assessment
from maturity_engine.timeutil import format_instant
ISSUER = "maturity-engine"
CLAIM_KIND = "maturity-level"
FRESHNESS_RULE = "assessment evaluated_at plus limiting evidence valid_until"
@dataclass(frozen=True)
class Claim:
kind: str
issuer: str
subject: str
model_id: str
model_version: str
level: int
level_id: str
assessed_at: str
assessment_id: str
freshness_rule: str
digest: str
def as_dict(self) -> dict:
return {
"kind": self.kind,
"issuer": self.issuer,
"subject": self.subject,
"model_id": self.model_id,
"model_version": self.model_version,
"level": self.level,
"level_id": self.level_id,
"assessed_at": self.assessed_at,
"assessment_id": self.assessment_id,
"freshness_rule": self.freshness_rule,
"digest": self.digest,
}
def to_claim(assessment: Assessment) -> Claim:
body = {
"kind": CLAIM_KIND,
"issuer": ISSUER,
"subject": assessment.subject,
"model_id": assessment.model_id,
"model_version": assessment.model_version,
"level": assessment.level,
"level_id": assessment.level_id,
"assessed_at": format_instant(assessment.evaluated_at),
"assessment_id": assessment.id,
"freshness_rule": FRESHNESS_RULE,
}
return Claim(**body, digest=digest(body))
def compile_into_registry(claim: Claim) -> None:
raise GuardrailError(
"A maturity level MUST NOT be compiled into registry content "
"(statute §9.5). Until access-engine decision provenance carries a "
"registry-snapshot digest, a level reaching a decision through the "
"registry is not reconstructable. Use to_claim() as a request claim "
"or a versioned policy rule."
)
def gate_on_level(claim: Claim, *, minimum: int) -> None:
raise GuardrailError(
"A maturity level MUST NOT gate a decision directly (statute §6.1 / "
f"§9.5). Refusing to proceed because {claim.level_id} < {minimum} is "
"a second decision point. The level reaches access-engine as a claim."
)

110
src/maturity_engine/cli.py Normal file
View file

@ -0,0 +1,110 @@
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
from maturity_engine.engine import Engine
from maturity_engine.timeutil import parse_instant
def _db(args: argparse.Namespace) -> Path:
if args.db:
return Path(args.db)
env = os.environ.get("MATURITY_ENGINE_DB")
if env:
return Path(env)
return Path("data/maturity-engine.sqlite")
def _engine(args: argparse.Namespace) -> Engine:
return Engine.open(_db(args))
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="maturity-engine",
description="Compute levels, remember gaps. Does not decide.",
)
parser.add_argument("--db", help="sqlite path (default data/maturity-engine.sqlite)")
sub = parser.add_subparsers(dest="cmd", required=True)
boot = sub.add_parser("bootstrap", help="register seed ladders, §13 snapshot, stance inventory")
boot.add_argument("--at", required=True, help="timezone-aware instant")
assess = sub.add_parser("assess", help="compute a level from recorded evidence")
assess.add_argument("--subject", required=True)
assess.add_argument("--model", required=True)
assess.add_argument("--at", required=True)
assess.add_argument("--version")
hist = sub.add_parser("history", help="progression history for a subject")
hist.add_argument("--subject", required=True)
hist.add_argument("--model")
sub.add_parser("gaps", help="query the gap register")
ready = sub.add_parser("readiness", help="pending | declared-gap | surface-exists")
ready.add_argument("capability")
sub.add_parser("stance", help="PEP stance-map inventory")
review = sub.add_parser("review", help="compute-and-remember surface for gate-house")
review.add_argument("--subject", required=True)
review.add_argument("--at", required=True)
sub.add_parser("outbox", help="list locally queued evidence events")
sub.add_parser("models", help="list registered ladders")
args = parser.parse_args(argv)
engine = _engine(args)
try:
return _dispatch(engine, args)
finally:
engine.close()
def _dispatch(engine: Engine, args: argparse.Namespace) -> int:
if args.cmd == "bootstrap":
engine.bootstrap(at=parse_instant(args.at))
print("bootstrapped")
return 0
if args.cmd == "assess":
assessment = engine.assess(
args.subject, args.model, at=parse_instant(args.at), version=args.version
)
print(json.dumps(assessment.as_explanation(), indent=2))
return 0
if args.cmd == "history":
print(json.dumps([item.as_explanation() for item in engine.history(args.subject, args.model)], indent=2))
return 0
if args.cmd == "gaps":
print(json.dumps([gap.as_dict() for gap in engine.gaps()], indent=2))
return 0
if args.cmd == "readiness":
print(json.dumps(engine.readiness(args.capability), indent=2))
return 0
if args.cmd == "stance":
print(json.dumps([item.as_dict() for item in engine.stance_maps()], indent=2))
return 0
if args.cmd == "review":
print(json.dumps(engine.review(args.subject, at=parse_instant(args.at)).as_dict(), indent=2))
return 0
if args.cmd == "outbox":
print(json.dumps(list(engine.pending_events()), indent=2))
return 0
if args.cmd == "models":
print(
json.dumps(
[
{"id": item.id, "version": item.version, "owner": item.owner, "name": item.name}
for item in engine.models()
],
indent=2,
)
)
return 0
return 2
if __name__ == "__main__":
raise SystemExit(main())

83
src/maturity_engine/compute.py Normal file
View file

@ -0,0 +1,83 @@
from __future__ import annotations
from datetime import datetime
from maturity_engine.ids import canonical_json, named_id
from maturity_engine.models import (
Assessment,
Evidence,
Ladder,
MetCriterion,
UnmetCriterion,
)
from maturity_engine.timeutil import format_instant
def compute_level(
ladder: Ladder,
evidence: tuple[Evidence, ...],
*,
subject: str,
instant: datetime,
) -> Assessment:
"""Highest consecutive level whose criteria all have valid evidence.
Evaluation instant is an input. Same ladder, same evidence, same instant,
same assessment.
"""
ladder.validate()
valid = [item for item in evidence if item.valid_at(instant)]
expired = tuple(
item.id
for item in evidence
if not item.valid_at(instant) and item.valid_until is not None and instant >= item.valid_until
)
by_kind: dict[str, list[Evidence]] = {}
for item in valid:
by_kind.setdefault(item.kind, []).append(item)
met: list[MetCriterion] = []
achieved = -1
unmet: tuple[UnmetCriterion, ...] = ()
next_level_id: str | None = None
for level in ladder.levels:
missing = [
UnmetCriterion(criterion.id, criterion.evidence_kind, criterion.description)
for criterion in level.criteria
if criterion.evidence_kind not in by_kind
]
if missing:
unmet = tuple(missing)
next_level_id = level.id
break
for criterion in level.criteria:
ids = tuple(item.id for item in by_kind[criterion.evidence_kind])
met.append(MetCriterion(criterion.id, criterion.evidence_kind, ids))
achieved = level.index
if achieved < 0:
raise RuntimeError("ladder has no floor level 0")
level = ladder.level_by_index(achieved)
material = {
"subject": subject,
"model_id": ladder.id,
"model_version": ladder.version,
"evaluated_at": format_instant(instant),
"evidence_ids": sorted(item.id for item in evidence),
}
assessment_id = named_id("assessment", canonical_json(material))
return Assessment(
id=assessment_id,
subject=subject,
model_id=ladder.id,
model_version=ladder.version,
level=achieved,
level_id=level.id,
evaluated_at=instant,
met=tuple(met),
unmet=unmet,
next_level_id=next_level_id,
expired_evidence_ids=expired,
)

183
src/maturity_engine/engine.py Normal file
View file

@ -0,0 +1,183 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from maturity_engine.claims import Claim, compile_into_registry, gate_on_level, to_claim
from maturity_engine.compute import compute_level
from maturity_engine.errors import UnknownModel
from maturity_engine.ids import named_id
from maturity_engine.models import Assessment, Evidence, Gap, Ladder, StanceMap
from maturity_engine.scoring import (
capability_readiness,
score,
states_from_gaps,
worst_state,
)
from maturity_engine.seed import (
asm_ladder,
pep_stance_publication_ladder,
section_13_1_stances,
section_13_gaps,
)
from maturity_engine.store import Store
from maturity_engine.timeutil import format_instant
@dataclass(frozen=True)
class ConformanceReview:
subject: str
evaluated_at: datetime
aggregate_state: str
capabilities: tuple[dict, ...]
assessments: tuple[dict, ...]
note: str = (
"gate-house judges and proposes; maturity-engine computes and remembers. "
"This review is a computation, not a judgment."
)
def as_dict(self) -> dict:
return {
"subject": self.subject,
"evaluated_at": format_instant(self.evaluated_at),
"aggregate_state": self.aggregate_state,
"capabilities": list(self.capabilities),
"assessments": list(self.assessments),
"note": self.note,
}
class Engine:
"""Deterministic API for graded progression, the gap register, and readiness.
No authorize / decide / may method exists. A maturity level leaves this
process only as a Claim.
"""
def __init__(self, store: Store) -> None:
self.store = store
@classmethod
def open(cls, path: Path) -> Engine:
return cls(Store(path))
def close(self) -> None:
self.store.close()
def bootstrap(self, *, at: datetime) -> None:
self.register_model(asm_ladder())
self.register_model(pep_stance_publication_ladder())
for gap in section_13_gaps():
self.store.put_gap(gap, emit_at=at)
for stance in section_13_1_stances():
self.store.put_stance(stance)
def register_model(self, ladder: Ladder) -> None:
self.store.put_ladder(ladder)
def models(self) -> tuple[Ladder, ...]:
return self.store.list_ladders()
def submit_evidence(self, evidence: Evidence) -> None:
self.store.put_evidence(evidence, emit_at=evidence.submitted_at)
def assess(self, subject: str, model_id: str, *, at: datetime, version: str | None = None) -> Assessment:
ladder = self.store.get_ladder(model_id, version)
if ladder is None:
raise UnknownModel(f"{model_id}@{version or 'latest'}")
evidence = self.store.evidence_for(subject)
assessment = compute_level(ladder, evidence, subject=subject, instant=at)
self.store.put_assessment(assessment)
return assessment
def history(self, subject: str, model_id: str | None = None) -> tuple[Assessment, ...]:
return self.store.assessments_for(subject, model_id)
def claim(self, assessment: Assessment) -> Claim:
return to_claim(assessment)
def gaps(self) -> tuple[Gap, ...]:
return self.store.list_gaps()
def upsert_gap(self, gap: Gap, *, at: datetime) -> None:
self.store.put_gap(gap, emit_at=at)
def readiness(self, capability_id: str) -> dict:
gap = self.store.get_gap(capability_id)
if gap is None:
return {"id": capability_id, "readiness": "pending", "note": "no register row"}
return {
"id": gap.id,
"capability": gap.capability,
"readiness": capability_readiness(gap),
"mark": gap.mark,
"state": gap.state,
"owner_status": gap.owner_status,
"intended_owner": gap.intended_owner,
"owns_actuation": False,
}
def stance_maps(self) -> tuple[StanceMap, ...]:
return self.store.list_stances()
def review(self, subject: str, *, at: datetime) -> ConformanceReview:
touching = tuple(
gap
for gap in self.gaps()
if _touches(subject, gap.declared_by) or _touches(subject, gap.intended_owner)
)
capabilities = tuple(
{
"id": gap.id,
"readiness": capability_readiness(gap),
"state": gap.state,
"owner_status": gap.owner_status,
"mark": gap.mark,
"score": score(
{
"pending": "blocked-clean",
"declared-gap": "declared-gap",
"surface-exists": "conforming",
}.get(capability_readiness(gap), "blocked-clean")
),
}
for gap in touching
)
states = states_from_gaps(touching)
assessments = tuple(item.as_explanation() for item in self.history(subject))
return ConformanceReview(
subject=subject,
evaluated_at=at,
aggregate_state=worst_state(states),
capabilities=capabilities,
assessments=assessments,
)
def pending_events(self) -> tuple[dict, ...]:
return self.store.pending_outbox()
def drain_events(self, *, at: datetime) -> int:
"""Mark queued events drained. Does not call audit-core."""
pending = self.store.pending_outbox()
for item in pending:
self.store.drain_outbox(item["id"], at)
return len(pending)
def heartbeat(self, *, at: datetime) -> None:
self.store.heartbeat(at)
def evidence_id(self, subject: str, kind: str, submitted_at: datetime) -> str:
return named_id("evidence", subject, kind, format_instant(submitted_at))
def _touches(subject: str, field: str | None) -> bool:
if not field:
return False
return subject == field or subject in {part.strip() for part in field.replace(";", ",").split(",")}
# Guardrail helpers re-exported so tests import one place, and so nobody
# adds a decide() here thinking it is convenient.
reject_registry_compile = compile_into_registry
reject_consumer_branch = gate_on_level

25
src/maturity_engine/errors.py Normal file
View file

@ -0,0 +1,25 @@
"""Domain errors. None of these is an authorization decision."""
class MaturityError(Exception):
"""Base error for this engine."""
class ModelError(MaturityError):
"""A ladder cannot be registered as specified."""
class UnevaluableCriterion(ModelError):
"""A criterion has no rule; it is not yet a criterion."""
class GuardrailError(MaturityError):
"""A caller asked this engine to decide or to compile a level into a registry."""
class UnknownModel(MaturityError):
"""No ladder is registered under that id and version."""
class UnknownSubject(MaturityError):
"""No facts are recorded for that subject."""

20
src/maturity_engine/ids.py Normal file
View file

@ -0,0 +1,20 @@
from __future__ import annotations
import hashlib
import json
from typing import Any
from uuid import UUID, uuid5
NAMESPACE = uuid5(UUID("6ba7b811-9dad-11d1-80b4-00c04fd430c8"), "net-kingdom/maturity-engine")
def canonical_json(value: Any) -> str:
return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
def digest(value: Any) -> str:
return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
def named_id(*parts: str) -> str:
return str(uuid5(NAMESPACE, "|".join(parts)))

180
src/maturity_engine/models.py Normal file
View file

@ -0,0 +1,180 @@
from __future__ import annotations
import re
from dataclasses import dataclass, field
from datetime import datetime
from maturity_engine.errors import UnevaluableCriterion
from maturity_engine.timeutil import format_instant
KIND_RE = re.compile(r"^[a-z][a-z0-9-]*$")
UNEVALUABLE = {"interpret", "judgment", "opinion", "human", "discretion", "review-discretion"}
@dataclass(frozen=True)
class Criterion:
id: str
evidence_kind: str
description: str
def validate(self) -> None:
kind = self.evidence_kind.strip()
if kind in UNEVALUABLE or not KIND_RE.match(kind):
raise UnevaluableCriterion(
f"criterion {self.id!r} is not evaluable by rule "
f"(evidence_kind={self.evidence_kind!r})"
)
@dataclass(frozen=True)
class Level:
index: int
id: str
name: str
criteria: tuple[Criterion, ...]
@dataclass(frozen=True)
class Ladder:
id: str
version: str
owner: str
name: str
levels: tuple[Level, ...]
def validate(self) -> None:
if not self.levels:
raise UnevaluableCriterion(f"ladder {self.id} has no levels")
indexes = [level.index for level in self.levels]
if indexes != list(range(len(self.levels))):
raise UnevaluableCriterion(
f"ladder {self.id} levels must be consecutive from 0, got {indexes}"
)
for level in self.levels:
for criterion in level.criteria:
criterion.validate()
def level_by_index(self, index: int) -> Level:
return self.levels[index]
@dataclass(frozen=True)
class Evidence:
id: str
subject: str
kind: str
submitted_by: str
submitted_at: datetime
valid_from: datetime
valid_until: datetime | None
payload: dict = field(default_factory=dict)
def valid_at(self, instant: datetime) -> bool:
if instant < self.valid_from:
return False
if self.valid_until is not None and instant >= self.valid_until:
return False
return True
@dataclass(frozen=True)
class MetCriterion:
criterion_id: str
evidence_kind: str
evidence_ids: tuple[str, ...]
@dataclass(frozen=True)
class UnmetCriterion:
criterion_id: str
evidence_kind: str
description: str
@dataclass(frozen=True)
class Assessment:
id: str
subject: str
model_id: str
model_version: str
level: int
level_id: str
evaluated_at: datetime
met: tuple[MetCriterion, ...]
unmet: tuple[UnmetCriterion, ...]
next_level_id: str | None
expired_evidence_ids: tuple[str, ...]
def as_explanation(self) -> dict:
return {
"assessment_id": self.id,
"subject": self.subject,
"model_id": self.model_id,
"model_version": self.model_version,
"level": self.level,
"level_id": self.level_id,
"evaluated_at": format_instant(self.evaluated_at),
"met": [
{
"criterion_id": item.criterion_id,
"evidence_kind": item.evidence_kind,
"evidence_ids": list(item.evidence_ids),
}
for item in self.met
],
"unmet": [
{
"criterion_id": item.criterion_id,
"evidence_kind": item.evidence_kind,
"description": item.description,
}
for item in self.unmet
],
"next_level_id": self.next_level_id,
"expired_evidence_ids": list(self.expired_evidence_ids),
}
@dataclass(frozen=True)
class Gap:
id: str
capability: str
intended_owner: str | None
blocked_on: str | None
review: str | None
state: str
owner_status: str
declared_by: str
mark: str | None
notes: str | None = None
def as_dict(self) -> dict:
return {
"id": self.id,
"capability": self.capability,
"intended_owner": self.intended_owner,
"blocked_on": self.blocked_on,
"review": self.review,
"state": self.state,
"owner_status": self.owner_status,
"declared_by": self.declared_by,
"mark": self.mark,
"notes": self.notes,
}
@dataclass(frozen=True)
class StanceMap:
consumer: str
published: bool
path: str | None
shape: str
def as_dict(self) -> dict:
return {
"consumer": self.consumer,
"published": self.published,
"path": self.path,
"shape": self.shape,
"absence": not self.published,
}

87
src/maturity_engine/scoring.py Normal file
View file

@ -0,0 +1,87 @@
"""Four conformance states and the scoring rule from statute §11.
blocked-clean MUST NOT rank below conforming. The two are equal. Declared
gap is tracked non-conformance. Undeclared violation is worse.
"""
from __future__ import annotations
from maturity_engine.models import Gap
CONFORMING = "conforming"
BLOCKED_CLEAN = "blocked-clean"
DECLARED_GAP = "declared-gap"
UNDECLARED_VIOLATION = "undeclared-violation"
STATES = (CONFORMING, BLOCKED_CLEAN, DECLARED_GAP, UNDECLARED_VIOLATION)
# Higher is better. Equal scores mean neither ranks below the other.
SCORE = {
CONFORMING: 2,
BLOCKED_CLEAN: 2,
DECLARED_GAP: 1,
UNDECLARED_VIOLATION: 0,
}
ASSIGNED_STATUSES = {"assigned", "assented", "resolved"}
def score(state: str) -> int:
if state not in SCORE:
raise ValueError(f"unknown conformance state {state!r}")
return SCORE[state]
def ranks_below(left: str, right: str) -> bool:
return score(left) < score(right)
def mark_for(gap: Gap) -> str | None:
return gap.mark
def capability_readiness(gap: Gap) -> str:
"""pending | declared-gap | surface-exists. Per capability, not per repo."""
if gap.owner_status in ASSIGNED_STATUSES or gap.state == "resolved":
return "surface-exists"
if gap.mark == "declared-gap" or gap.state == "declared-contact":
return "declared-gap"
if gap.mark == "pending" or gap.state == "unowned-capability":
return "pending"
return "pending"
def states_from_gaps(gaps: tuple[Gap, ...]) -> tuple[str, ...]:
if not gaps:
return (CONFORMING,)
found: list[str] = []
for gap in gaps:
readiness = capability_readiness(gap)
if gap.state == "undeclared-violation" or gap.mark == "undeclared-violation":
found.append(UNDECLARED_VIOLATION)
elif readiness == "declared-gap":
found.append(DECLARED_GAP)
elif readiness == "pending":
found.append(BLOCKED_CLEAN)
else:
found.append(CONFORMING)
return tuple(found)
def worst_state(states: tuple[str, ...]) -> str:
"""Aggregate that never ranks blocked-clean below conforming.
A repo that is only blocked-clean stays blocked-clean, which scores equal
to conforming. Declared-gap or undeclared-violation still surface.
"""
if not states:
return CONFORMING
if UNDECLARED_VIOLATION in states:
return UNDECLARED_VIOLATION
if DECLARED_GAP in states:
return DECLARED_GAP
if BLOCKED_CLEAN in states and CONFORMING not in states:
return BLOCKED_CLEAN
if BLOCKED_CLEAN in states and CONFORMING in states:
return CONFORMING
return CONFORMING

299
src/maturity_engine/seed.py Normal file
View file

@ -0,0 +1,299 @@
"""Seed data: statute §13 snapshot, §13.1 stance inventory, first two ladders.
Ladder *content* is not authored here. ASM-0…ASM-6 is gate-house's
(Active Secrets Management Canon v0.3 §38). PEP-stance publication is
ops-warden's (ADR-0009). This module registers those ladders as data.
"""
from __future__ import annotations
from maturity_engine.models import Criterion, Gap, Ladder, Level, StanceMap
REVIEW = "2026-11-28"
def _c(criterion_id: str, kind: str, description: str) -> Criterion:
return Criterion(id=criterion_id, evidence_kind=kind, description=description)
def asm_ladder() -> Ladder:
"""ASM-0…ASM-6. Criteria kinds are mechanical; doctrine stays gate-house's."""
levels = (
Level(0, "ASM-0", "Embedded", ()),
Level(
1,
"ASM-1",
"Stored",
(
_c("asm-1-manager", "central-secret-manager", "Central secret manager exists"),
_c("asm-1-rbac", "secret-rbac", "RBAC on secrets exists"),
_c("asm-1-scan", "basic-secret-scanning", "Basic secret scanning exists"),
_c("asm-1-rotate", "basic-rotation", "Basic rotation exists"),
),
),
Level(
2,
"ASM-2",
"Managed",
(
_c("asm-2-owner", "secret-ownership", "Ownership is established"),
_c("asm-2-auto", "automated-rotation", "Automated rotation is established"),
_c("asm-2-inv", "secret-inventory", "Inventory is established"),
_c("asm-2-detect", "exposure-detection", "Detection is established"),
_c("asm-2-remediate", "measurable-remediation", "Measurable remediation is established"),
),
),
Level(
3,
"ASM-3",
"Dynamic",
(
_c("asm-3-fed", "workload-federation", "Workload federation reduces standing credentials"),
_c("asm-3-jit", "jit-access", "JIT access reduces standing credentials"),
_c("asm-3-dyn", "dynamic-secrets", "Dynamic secrets reduce standing credentials"),
_c("asm-3-ci", "secretless-cicd", "Secretless CI/CD reduces standing credentials"),
),
),
Level(
4,
"ASM-4",
"Agent-Aware",
(
_c("asm-4-split", "principal-actor-distinction", "Human principal and agent actor are distinguished"),
_c("asm-4-asst", "assistant-mode-modeled", "Assistant mode is explicitly modeled"),
_c("asm-4-auto", "autonomous-mode-modeled", "Autonomous mode is explicitly modeled"),
),
),
Level(
5,
"ASM-5",
"Governed Autonomy",
(
_c("asm-5-id", "agent-identities", "Autonomous agents have identities"),
_c("asm-5-man", "agent-mandates", "Autonomous agents have mandates"),
_c("asm-5-budget", "agent-budgets", "Autonomous agents have budgets"),
_c("asm-5-ceil", "authority-ceilings", "Authority ceilings are in force"),
_c("asm-5-env", "change-envelopes", "Change envelopes are in force"),
_c("asm-5-cb", "circuit-breakers", "Deterministic circuit breakers are in force"),
),
),
Level(
6,
"ASM-6",
"Closed-Loop Authority",
(
_c("asm-6-id", "identity-join", "Identity is in the reconciled authority lifecycle"),
_c("asm-6-del", "delegation-join", "Delegation is in the reconciled authority lifecycle"),
_c("asm-6-iss", "issuance-join", "Credential issuance is in the reconciled authority lifecycle"),
_c("asm-6-exec", "execution-join", "Execution is in the reconciled authority lifecycle"),
_c("asm-6-ev", "evidence-join", "Evidence is in the reconciled authority lifecycle"),
_c("asm-6-exp", "exposure-join", "Exposure detection is in the reconciled authority lifecycle"),
_c("asm-6-rev", "revocation-join", "Revocation is in the reconciled authority lifecycle"),
_c("asm-6-rem", "remediation-join", "Remediation is in the reconciled authority lifecycle"),
),
),
)
return Ladder(
id="asm",
version="0.3",
owner="gate-house",
name="Active Secrets Management",
levels=levels,
)
def pep_stance_publication_ladder() -> Ladder:
"""Publication of unreachable-engine stance maps. Doctrine is ops-warden's."""
levels = (
Level(0, "PSP-0", "Unpublished", ()),
Level(
1,
"PSP-1",
"Published",
(_c("psp-1-file", "stance-map-published", "Stance map published at a named path"),),
),
Level(
2,
"PSP-2",
"Tested",
(
_c(
"psp-2-test",
"stance-map-equality-test",
"A test asserts the published map equals shipped behaviour",
),
),
),
)
return Ladder(
id="pep-stance-publication",
version="0.1",
owner="ops-warden",
name="PEP stance-map publication",
levels=levels,
)
def section_13_gaps() -> tuple[Gap, ...]:
"""Statute §13 snapshot. state and owner_status survive the migration."""
return (
Gap(
id="ssh-ca-signing-write",
capability="SSH-CA signing write (VaultCA, bao kv put)",
intended_owner="secrets-engine",
blocked_on="No engine exposes an SSH certificate signing surface",
review=REVIEW,
state="declared-contact",
owner_status="proposed",
declared_by="ops-warden",
mark="declared-gap",
),
Gap(
id="authentication-assurance-evidence",
capability="Authentication / assurance evidence",
intended_owner="identity layer + audit-core",
blocked_on="access-engine declined (FLEX-DEC-2026-002); no identity-layer evidence surface",
review=REVIEW,
state="unowned-capability",
owner_status="declined",
declared_by="kings-guard",
mark="pending",
notes="access-engine declined; reproposed, not assented",
),
Gap(
id="secret-use-evidence",
capability="Secret-use evidence",
intended_owner="secrets-engine",
blocked_on="No engine exposes secret-use evidence; OpenBao is Tooling",
review=REVIEW,
state="unowned-capability",
owner_status="proposed",
declared_by="kings-guard",
mark="pending",
),
Gap(
id="actuation-containment-surface",
capability="Reduce authority, require step-up, isolate a workload — as a deterministic engine API",
intended_owner="access-engine + runtime PEPs",
blocked_on="Ruled in v0.7 §9.2 to be an Engine concept, unowned and held at zero",
review=REVIEW,
state="unowned-capability",
owner_status="proposed",
declared_by="gate-house (estate-wide)",
mark="pending",
notes="kings-guard proposes containment and does not own it",
),
Gap(
id="identity-and-secret-observation",
capability="Identity and secret observation",
intended_owner="identity layer + secrets-engine + audit-core",
blocked_on="No engine exposes the observation surface; kings-guard makes no Tooling contact",
review=REVIEW,
state="unowned-capability",
owner_status="proposed",
declared_by="kings-guard",
mark="pending",
),
Gap(
id="stance-map-register",
capability="Stance-map register had no implementation",
intended_owner="gate-house",
blocked_on=None,
review=REVIEW,
state="resolved",
owner_status="resolved",
declared_by="ops-warden, access-engine",
mark=None,
notes="resolved in statute §13.1; inventory now held here",
),
Gap(
id="registry-snapshot-digest",
capability="Registry-snapshot digest in decision provenance",
intended_owner="flex-auth",
blocked_on="Decision provenance holds no snapshot digest; a registry compile of a level would be unfalsifiable",
review=REVIEW,
state="declared-contact",
owner_status="self-declared",
declared_by="flex-auth",
mark="declared-gap",
),
Gap(
id="approval-storage-lifecycle",
capability="Approval storage and lifecycle",
intended_owner="approval-engine",
blocked_on=None,
review=REVIEW,
state="assigned",
owner_status="assigned",
declared_by="flex-auth",
mark=None,
notes="assigned (§9.4)",
),
Gap(
id="approval-evidence",
capability="Approval evidence",
intended_owner="audit-core",
blocked_on=None,
review=REVIEW,
state="assigned",
owner_status="assented",
declared_by="gate-house",
mark=None,
notes="assented (AUDIT-IN-0001)",
),
Gap(
id="approval-evidence-custody-stronger",
capability="Approval evidence custody stronger than the shipped bound — WORM, object lock, transparency log",
intended_owner=None,
blocked_on="Doctrine work not yet done; approval evidence carries the same bound as every other source",
review=REVIEW,
state="unowned-capability",
owner_status="unassigned",
declared_by="audit-core",
mark="pending",
),
Gap(
id="approval-emission-atomicity",
capability="Emission atomicity for approval state changes",
intended_owner="approval-engine",
blocked_on=None,
review=REVIEW,
state="assigned",
owner_status="assigned",
declared_by="audit-core",
mark=None,
notes="assigned (§9.4)",
),
Gap(
id="ssh-signing-non-atomic-audit",
capability="Non-atomic audit emission on the SSH signing lane",
intended_owner="ops-warden",
blocked_on="Declared trade so an audit-store failure cannot remove production host access",
review=REVIEW,
state="declared-contact",
owner_status="self-declared",
declared_by="ops-warden",
mark="declared-gap",
notes="attributive (§9.6)",
),
)
def section_13_1_stances() -> tuple[StanceMap, ...]:
return (
StanceMap(
consumer="ops-warden",
published=True,
path="ops-warden/pep-stance.yaml",
shape=(
"total per-zone; open z0–z2 and unknown, closed z3-critical; "
"test asserts the published map equals the shipped default (ADR-0009)"
),
),
StanceMap(
consumer="ops-mason",
published=False,
path=None,
shape="catalogued PEP-shaped in §4; map not published",
),
)

470
src/maturity_engine/store.py Normal file
View file

@ -0,0 +1,470 @@
from __future__ import annotations
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from maturity_engine.models import (
Assessment,
Criterion,
Evidence,
Gap,
Ladder,
Level,
MetCriterion,
StanceMap,
UnmetCriterion,
)
from maturity_engine.timeutil import format_instant, parse_instant
SCHEMA = """
CREATE TABLE IF NOT EXISTS models (
id TEXT NOT NULL,
version TEXT NOT NULL,
owner TEXT NOT NULL,
name TEXT NOT NULL,
body TEXT NOT NULL,
PRIMARY KEY (id, version)
);
CREATE TABLE IF NOT EXISTS evidence (
id TEXT PRIMARY KEY,
subject TEXT NOT NULL,
kind TEXT NOT NULL,
submitted_by TEXT NOT NULL,
submitted_at TEXT NOT NULL,
valid_from TEXT NOT NULL,
valid_until TEXT,
payload TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS assessments (
id TEXT PRIMARY KEY,
subject TEXT NOT NULL,
model_id TEXT NOT NULL,
model_version TEXT NOT NULL,
level INTEGER NOT NULL,
level_id TEXT NOT NULL,
evaluated_at TEXT NOT NULL,
explanation TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS gaps (
id TEXT PRIMARY KEY,
capability TEXT NOT NULL,
intended_owner TEXT,
blocked_on TEXT,
review TEXT,
state TEXT NOT NULL,
owner_status TEXT NOT NULL,
declared_by TEXT NOT NULL,
mark TEXT,
notes TEXT
);
CREATE TABLE IF NOT EXISTS stance_maps (
consumer TEXT PRIMARY KEY,
published INTEGER NOT NULL,
path TEXT,
shape TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
created_at TEXT NOT NULL,
drained_at TEXT
);
"""
class Store:
"""This PIP's own transactional store. The outbox lives here, not in audit-core."""
def __init__(self, path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
self.path = path
self.fail_outbox = False
self._conn = sqlite3.connect(path)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA foreign_keys=ON")
self._conn.executescript(SCHEMA)
self._conn.commit()
def close(self) -> None:
self._conn.close()
def _commit_or_rollback(self) -> None:
if self.fail_outbox:
self._conn.rollback()
raise RuntimeError("outbox insert refused — state change rolled back")
self._conn.commit()
def enqueue(self, event_type: str, payload: dict, created_at: datetime) -> None:
if self.fail_outbox:
raise RuntimeError("outbox insert refused")
self._conn.execute(
"INSERT INTO outbox(event_type, payload, created_at) VALUES (?, ?, ?)",
(event_type, json.dumps(payload, sort_keys=True), format_instant(created_at)),
)
def put_ladder(self, ladder: Ladder) -> None:
ladder.validate()
body = _ladder_body(ladder)
self._conn.execute(
"""
INSERT INTO models(id, version, owner, name, body)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(id, version) DO UPDATE SET
owner=excluded.owner, name=excluded.name, body=excluded.body
""",
(ladder.id, ladder.version, ladder.owner, ladder.name, json.dumps(body, sort_keys=True)),
)
self._conn.commit()
def get_ladder(self, model_id: str, version: str | None = None) -> Ladder | None:
if version is None:
row = self._conn.execute(
"SELECT body FROM models WHERE id = ? ORDER BY version DESC LIMIT 1",
(model_id,),
).fetchone()
else:
row = self._conn.execute(
"SELECT body FROM models WHERE id = ? AND version = ?",
(model_id, version),
).fetchone()
if row is None:
return None
return _ladder_from_body(json.loads(row["body"]))
def list_ladders(self) -> tuple[Ladder, ...]:
rows = self._conn.execute("SELECT body FROM models ORDER BY id, version").fetchall()
return tuple(_ladder_from_body(json.loads(row["body"])) for row in rows)
def put_evidence(self, evidence: Evidence, *, emit_at: datetime) -> None:
self._conn.execute(
"""
INSERT INTO evidence(
id, subject, kind, submitted_by, submitted_at, valid_from, valid_until, payload
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
subject=excluded.subject,
kind=excluded.kind,
submitted_by=excluded.submitted_by,
submitted_at=excluded.submitted_at,
valid_from=excluded.valid_from,
valid_until=excluded.valid_until,
payload=excluded.payload
""",
(
evidence.id,
evidence.subject,
evidence.kind,
evidence.submitted_by,
format_instant(evidence.submitted_at),
format_instant(evidence.valid_from),
None if evidence.valid_until is None else format_instant(evidence.valid_until),
json.dumps(evidence.payload, sort_keys=True),
),
)
try:
self.enqueue(
"evidence.submitted",
{
"evidence_id": evidence.id,
"subject": evidence.subject,
"kind": evidence.kind,
"class": "attributive",
},
emit_at,
)
except RuntimeError:
self._conn.rollback()
raise
self._commit_or_rollback()
def evidence_for(self, subject: str) -> tuple[Evidence, ...]:
rows = self._conn.execute(
"SELECT * FROM evidence WHERE subject = ? ORDER BY submitted_at, id",
(subject,),
).fetchall()
return tuple(_evidence_from_row(row) for row in rows)
def put_assessment(self, assessment: Assessment) -> None:
explanation = assessment.as_explanation()
existed = self._conn.execute(
"SELECT 1 FROM assessments WHERE id = ?", (assessment.id,)
).fetchone()
self._conn.execute(
"""
INSERT INTO assessments(
id, subject, model_id, model_version, level, level_id, evaluated_at, explanation
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
level=excluded.level,
level_id=excluded.level_id,
explanation=excluded.explanation
""",
(
assessment.id,
assessment.subject,
assessment.model_id,
assessment.model_version,
assessment.level,
assessment.level_id,
format_instant(assessment.evaluated_at),
json.dumps(explanation, sort_keys=True),
),
)
if existed is None:
try:
self.enqueue(
"assessment.recorded",
{
"assessment_id": assessment.id,
"subject": assessment.subject,
"model_id": assessment.model_id,
"level_id": assessment.level_id,
"class": "load-bearing",
"bound": (
"archive proves records were not altered or truncated after arrival; "
"it does not prove an event never sent"
),
},
assessment.evaluated_at,
)
except RuntimeError:
self._conn.rollback()
raise
self._commit_or_rollback()
def assessments_for(self, subject: str, model_id: str | None = None) -> tuple[Assessment, ...]:
if model_id is None:
rows = self._conn.execute(
"SELECT explanation FROM assessments WHERE subject = ? ORDER BY evaluated_at, id",
(subject,),
).fetchall()
else:
rows = self._conn.execute(
"""
SELECT explanation FROM assessments
WHERE subject = ? AND model_id = ?
ORDER BY evaluated_at, id
""",
(subject, model_id),
).fetchall()
return tuple(_assessment_from_explanation(json.loads(row["explanation"])) for row in rows)
def put_gap(self, gap: Gap, *, emit_at: datetime) -> None:
self._conn.execute(
"""
INSERT INTO gaps(
id, capability, intended_owner, blocked_on, review, state,
owner_status, declared_by, mark, notes
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
capability=excluded.capability,
intended_owner=excluded.intended_owner,
blocked_on=excluded.blocked_on,
review=excluded.review,
state=excluded.state,
owner_status=excluded.owner_status,
declared_by=excluded.declared_by,
mark=excluded.mark,
notes=excluded.notes
""",
(
gap.id,
gap.capability,
gap.intended_owner,
gap.blocked_on,
gap.review,
gap.state,
gap.owner_status,
gap.declared_by,
gap.mark,
gap.notes,
),
)
try:
self.enqueue(
"gap.mutated",
{"gap_id": gap.id, "state": gap.state, "owner_status": gap.owner_status, "class": "attributive"},
emit_at,
)
except RuntimeError:
self._conn.rollback()
raise
self._commit_or_rollback()
def list_gaps(self) -> tuple[Gap, ...]:
rows = self._conn.execute("SELECT * FROM gaps ORDER BY id").fetchall()
return tuple(_gap_from_row(row) for row in rows)
def get_gap(self, gap_id: str) -> Gap | None:
row = self._conn.execute("SELECT * FROM gaps WHERE id = ?", (gap_id,)).fetchone()
return None if row is None else _gap_from_row(row)
def put_stance(self, stance: StanceMap) -> None:
self._conn.execute(
"""
INSERT INTO stance_maps(consumer, published, path, shape)
VALUES (?, ?, ?, ?)
ON CONFLICT(consumer) DO UPDATE SET
published=excluded.published, path=excluded.path, shape=excluded.shape
""",
(stance.consumer, 1 if stance.published else 0, stance.path, stance.shape),
)
self._conn.commit()
def list_stances(self) -> tuple[StanceMap, ...]:
rows = self._conn.execute("SELECT * FROM stance_maps ORDER BY consumer").fetchall()
return tuple(
StanceMap(
consumer=row["consumer"],
published=bool(row["published"]),
path=row["path"],
shape=row["shape"],
)
for row in rows
)
def pending_outbox(self) -> tuple[dict, ...]:
rows = self._conn.execute(
"SELECT id, event_type, payload, created_at FROM outbox WHERE drained_at IS NULL ORDER BY id"
).fetchall()
return tuple(
{
"id": row["id"],
"event_type": row["event_type"],
"payload": json.loads(row["payload"]),
"created_at": row["created_at"],
}
for row in rows
)
def drain_outbox(self, row_id: int, drained_at: datetime) -> None:
self._conn.execute(
"UPDATE outbox SET drained_at = ? WHERE id = ?",
(format_instant(drained_at), row_id),
)
self._conn.commit()
def heartbeat(self, instant: datetime) -> None:
self.enqueue(
"maturity-engine.heartbeat",
{
"class": "load-bearing",
"form": "heartbeat",
"note": "positive claim that can itself go missing; not a rate",
},
instant,
)
self._commit_or_rollback()
def _ladder_body(ladder: Ladder) -> dict:
return {
"id": ladder.id,
"version": ladder.version,
"owner": ladder.owner,
"name": ladder.name,
"levels": [
{
"index": level.index,
"id": level.id,
"name": level.name,
"criteria": [
{
"id": criterion.id,
"evidence_kind": criterion.evidence_kind,
"description": criterion.description,
}
for criterion in level.criteria
],
}
for level in ladder.levels
],
}
def _ladder_from_body(body: dict) -> Ladder:
levels = []
for level in body["levels"]:
criteria = tuple(
Criterion(
id=item["id"],
evidence_kind=item["evidence_kind"],
description=item["description"],
)
for item in level["criteria"]
)
levels.append(Level(index=level["index"], id=level["id"], name=level["name"], criteria=criteria))
return Ladder(
id=body["id"],
version=body["version"],
owner=body["owner"],
name=body["name"],
levels=tuple(levels),
)
def _evidence_from_row(row: sqlite3.Row) -> Evidence:
until = row["valid_until"]
return Evidence(
id=row["id"],
subject=row["subject"],
kind=row["kind"],
submitted_by=row["submitted_by"],
submitted_at=parse_instant(row["submitted_at"]),
valid_from=parse_instant(row["valid_from"]),
valid_until=None if until is None else parse_instant(until),
payload=json.loads(row["payload"]),
)
def _gap_from_row(row: sqlite3.Row) -> Gap:
return Gap(
id=row["id"],
capability=row["capability"],
intended_owner=row["intended_owner"],
blocked_on=row["blocked_on"],
review=row["review"],
state=row["state"],
owner_status=row["owner_status"],
declared_by=row["declared_by"],
mark=row["mark"],
notes=row["notes"],
)
def _assessment_from_explanation(body: dict) -> Assessment:
return Assessment(
id=body["assessment_id"],
subject=body["subject"],
model_id=body["model_id"],
model_version=body["model_version"],
level=body["level"],
level_id=body["level_id"],
evaluated_at=parse_instant(body["evaluated_at"]),
met=tuple(
MetCriterion(
criterion_id=item["criterion_id"],
evidence_kind=item["evidence_kind"],
evidence_ids=tuple(item["evidence_ids"]),
)
for item in body["met"]
),
unmet=tuple(
UnmetCriterion(
criterion_id=item["criterion_id"],
evidence_kind=item["evidence_kind"],
description=item["description"],
)
for item in body["unmet"]
),
next_level_id=body["next_level_id"],
expired_evidence_ids=tuple(body["expired_evidence_ids"]),
)

19
src/maturity_engine/timeutil.py Normal file
View file

@ -0,0 +1,19 @@
from __future__ import annotations
from datetime import datetime, timezone
def parse_instant(value: str) -> datetime:
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
instant = datetime.fromisoformat(text)
if instant.tzinfo is None:
raise ValueError("evaluation instant must be timezone-aware")
return instant.astimezone(timezone.utc)
def format_instant(instant: datetime) -> str:
if instant.tzinfo is None:
raise ValueError("evaluation instant must be timezone-aware")
return instant.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")

40
tests/conftest.py Normal file
View file

@ -0,0 +1,40 @@
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
import pytest
from maturity_engine.engine import Engine
from maturity_engine.models import Evidence
INSTANT = datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)
@pytest.fixture
def engine(tmp_path: Path) -> Engine:
inst = Engine.open(tmp_path / "maturity.sqlite")
inst.bootstrap(at=INSTANT)
yield inst
inst.close()
def evidence(
engine: Engine,
subject: str,
kind: str,
*,
at: datetime = INSTANT,
until: datetime | None = None,
by: str = "fixture",
) -> Evidence:
return Evidence(
id=engine.evidence_id(subject, kind, at),
subject=subject,
kind=kind,
submitted_by=by,
submitted_at=at,
valid_from=at,
valid_until=until,
payload={"kind": kind},
)

72
tests/test_claims_and_guardrail.py Normal file
View file

@ -0,0 +1,72 @@
from __future__ import annotations
import pytest
from maturity_engine.claims import compile_into_registry, gate_on_level
from maturity_engine.engine import Engine
from maturity_engine.errors import GuardrailError
from conftest import INSTANT, evidence
def test_claim_shape(engine):
assessment = engine.assess("estate", "asm", at=INSTANT)
claim = engine.claim(assessment)
body = claim.as_dict()
assert body["kind"] == "maturity-level"
assert body["issuer"] == "maturity-engine"
assert body["subject"] == "estate"
assert body["level_id"] == "ASM-0"
assert body["digest"]
assert "freshness_rule" in body
def test_registry_compile_is_rejected(engine):
claim = engine.claim(engine.assess("estate", "asm", at=INSTANT))
with pytest.raises(GuardrailError, match="MUST NOT be compiled into registry"):
compile_into_registry(claim)
def test_consumer_branch_is_rejected(engine):
claim = engine.claim(engine.assess("estate", "asm", at=INSTANT))
with pytest.raises(GuardrailError, match="MUST NOT gate a decision directly"):
gate_on_level(claim, minimum=3)
def test_no_decision_surface():
assert not hasattr(Engine, "authorize")
assert not hasattr(Engine, "decide")
assert not hasattr(Engine, "may")
assert not hasattr(Engine, "allow")
def test_assessment_emission_is_local_and_load_bearing(engine):
engine.assess("estate", "asm", at=INSTANT)
events = [item for item in engine.pending_events() if item["event_type"] == "assessment.recorded"]
assert len(events) == 1
payload = events[0]["payload"]
assert payload["class"] == "load-bearing"
assert "not altered or truncated after arrival" in payload["bound"]
assert "never sent" in payload["bound"]
def test_outbox_failure_rolls_back_assessment(engine):
engine.store.fail_outbox = True
with pytest.raises(RuntimeError, match="outbox"):
engine.assess("estate", "pep-stance-publication", at=INSTANT)
assert engine.history("estate", "pep-stance-publication") == ()
def test_heartbeat_is_a_positive_claim(engine):
engine.heartbeat(at=INSTANT)
beats = [item for item in engine.pending_events() if item["event_type"] == "maturity-engine.heartbeat"]
assert beats
assert beats[0]["payload"]["form"] == "heartbeat"
def test_two_ladders_from_different_owners_can_be_claimed(engine):
for kind in ("stance-map-published", "stance-map-equality-test"):
engine.submit_evidence(evidence(engine, "ops-warden", kind))
asm = engine.claim(engine.assess("ops-warden", "asm", at=INSTANT))
psp = engine.claim(engine.assess("ops-warden", "pep-stance-publication", at=INSTANT))
assert asm.model_id != psp.model_id
assert psp.level_id == "PSP-2"

96
tests/test_compute.py Normal file
View file

@ -0,0 +1,96 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import pytest
from maturity_engine.compute import compute_level
from maturity_engine.errors import UnevaluableCriterion
from maturity_engine.models import Criterion, Evidence, Ladder, Level
from conftest import INSTANT, evidence
def test_floor_level_with_no_evidence(engine):
assessment = engine.assess("estate", "asm", at=INSTANT)
assert assessment.level == 0
assert assessment.level_id == "ASM-0"
assert assessment.next_level_id == "ASM-1"
assert assessment.unmet
def test_determinism_same_inputs_same_assessment(engine):
kinds = ["central-secret-manager", "secret-rbac", "basic-secret-scanning", "basic-rotation"]
for kind in kinds:
engine.submit_evidence(evidence(engine, "estate", kind))
first = engine.assess("estate", "asm", at=INSTANT)
second = engine.assess("estate", "asm", at=INSTANT)
assert first.id == second.id
assert first.level == second.level == 1
assert first.as_explanation() == second.as_explanation()
def test_explainability_lists_met_and_next(engine):
engine.submit_evidence(evidence(engine, "estate", "central-secret-manager"))
assessment = engine.assess("estate", "asm", at=INSTANT)
assert assessment.level == 0
kinds_unmet = {item.evidence_kind for item in assessment.unmet}
assert "secret-rbac" in kinds_unmet
explanation = assessment.as_explanation()
assert explanation["met"] == []
assert explanation["next_level_id"] == "ASM-1"
def test_expiry_demotes_a_subject(engine):
until = INSTANT + timedelta(hours=1)
kinds = ["central-secret-manager", "secret-rbac", "basic-secret-scanning", "basic-rotation"]
for kind in kinds:
engine.submit_evidence(evidence(engine, "estate", kind, until=until))
high = engine.assess("estate", "asm", at=INSTANT)
assert high.level == 1
later = INSTANT + timedelta(hours=2)
low = engine.assess("estate", "asm", at=later)
assert low.level == 0
assert low.level_id == "ASM-0"
assert low.expired_evidence_ids
history = engine.history("estate", "asm")
assert [item.level for item in history] == [1, 0]
def test_unevaluable_criterion_rejected():
ladder = Ladder(
id="bad",
version="1",
owner="nobody",
name="bad",
levels=(
Level(0, "L0", "floor", ()),
Level(
1,
"L1",
"judged",
(Criterion("c1", "judgment", "a human decides"),),
),
),
)
with pytest.raises(UnevaluableCriterion):
ladder.validate()
def test_compute_is_pure(engine):
item = Evidence(
id="e1",
subject="s",
kind="central-secret-manager",
submitted_by="t",
submitted_at=INSTANT,
valid_from=INSTANT,
valid_until=None,
)
ladder = engine.store.get_ladder("asm")
a = compute_level(ladder, (item,), subject="s", instant=INSTANT)
b = compute_level(ladder, (item,), subject="s", instant=INSTANT)
assert a == b
other = compute_level(
ladder, (item,), subject="s", instant=datetime(2026, 8, 30, tzinfo=timezone.utc)
)
assert other.id != a.id

83
tests/test_layer_conformance.py Normal file
View file

@ -0,0 +1,83 @@
from __future__ import annotations
import re
import subprocess
import sys
from pathlib import Path
import pytest
yaml = pytest.importorskip("yaml")
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "check_layer_conformance.py"
DECL = ROOT / "layer.yaml"
INTENT = ROOT / "INTENT.md"
def _run(*args: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, str(SCRIPT), *args],
capture_output=True,
text=True,
)
def test_declaration_exists_and_declares_engine_pip():
assert DECL.exists()
data = yaml.safe_load(DECL.read_text())
assert data["repository"] == "maturity-engine"
assert data["layer"] == "engine"
assert data["role"] == "pip"
assert data["framework"] == "netkingdom-security-layer-model"
assert data["standard_version"] == "0.7"
assert "pep_stance" not in data or not data.get("pep_stance")
def test_frontmatter_agrees_with_layer_yaml():
text = INTENT.read_text()
match = re.match(r"^---\n(.*?)\n---\n", text, re.DOTALL)
assert match
meta = yaml.safe_load(match.group(1))
data = yaml.safe_load(DECL.read_text())
assert str(meta["layer"]).lower() == str(data["layer"]).lower()
assert str(meta["role"]).lower() == str(data["role"]).lower()
def test_no_tooling_contacts_or_pep():
data = yaml.safe_load(DECL.read_text())
assert data["tooling_contacts"] == []
for entries in data["declared_shapes"].values():
assert entries == []
clients = {item["id"] for item in data["non_tooling_clients"]}
assert "state-hub-work-records" in clients
assert "sqlite-own-store" in clients
def test_catalog_entry_matches_section_4():
data = yaml.safe_load(DECL.read_text())
owns = " ".join(data["catalog_entry"]["owns"])
assert "graded progression" in owns
assert "gap register" in owns
assert "capability readiness" in owns
def test_checker_passes_on_the_real_tree():
result = _run()
assert result.returncode == 0, result.stderr
def test_checker_catches_an_openbao_client(tmp_path, monkeypatch):
import importlib.util
spec = importlib.util.spec_from_file_location("check_layer_conformance", SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
fake_src = tmp_path / "src" / "maturity_engine"
fake_src.mkdir(parents=True)
(fake_src / "oops.py").write_text("import hvac\n")
monkeypatch.setattr(module, "SRC", fake_src)
hits = module.scan()
assert hits
assert hits[0][1] == "hvac"

83
tests/test_register.py Normal file
View file

@ -0,0 +1,83 @@
from __future__ import annotations
from maturity_engine.scoring import (
BLOCKED_CLEAN,
CONFORMING,
DECLARED_GAP,
UNDECLARED_VIOLATION,
capability_readiness,
ranks_below,
score,
worst_state,
)
from conftest import INSTANT
def test_section_13_snapshot_is_queryable(engine):
gaps = {gap.id: gap for gap in engine.gaps()}
assert "ssh-ca-signing-write" in gaps
assert "actuation-containment-surface" in gaps
ssh = gaps["ssh-ca-signing-write"]
assert ssh.state == "declared-contact"
assert ssh.owner_status == "proposed"
assert ssh.mark == "declared-gap"
assert ssh.intended_owner == "secrets-engine"
assert ssh.blocked_on
assert ssh.review
actuation = gaps["actuation-containment-surface"]
assert actuation.state == "unowned-capability"
assert actuation.mark == "pending"
assert actuation.owner_status == "proposed"
declined = gaps["authentication-assurance-evidence"]
assert declined.owner_status == "declined"
assigned = gaps["approval-storage-lifecycle"]
assert assigned.owner_status == "assigned"
assert assigned.mark is None
def test_proposed_is_not_assigned(engine):
proposed = [gap for gap in engine.gaps() if gap.owner_status == "proposed"]
assigned = [gap for gap in engine.gaps() if gap.owner_status == "assigned"]
assert proposed
assert assigned
assert {gap.id for gap in proposed}.isdisjoint({gap.id for gap in assigned})
def test_blocked_clean_does_not_rank_below_conforming():
assert score(BLOCKED_CLEAN) == score(CONFORMING)
assert not ranks_below(BLOCKED_CLEAN, CONFORMING)
assert ranks_below(DECLARED_GAP, CONFORMING)
assert ranks_below(UNDECLARED_VIOLATION, BLOCKED_CLEAN)
assert worst_state((BLOCKED_CLEAN, CONFORMING)) == CONFORMING
assert worst_state((BLOCKED_CLEAN,)) == BLOCKED_CLEAN
def test_scoring_test_fails_if_blocked_clean_ranks_below(monkeypatch):
import maturity_engine.scoring as scoring
monkeypatch.setitem(scoring.SCORE, BLOCKED_CLEAN, scoring.SCORE[CONFORMING] - 1)
assert scoring.ranks_below(BLOCKED_CLEAN, CONFORMING)
def test_readiness_of_actuation_is_pending_not_owned(engine):
result = engine.readiness("actuation-containment-surface")
assert result["readiness"] == "pending"
assert result["owns_actuation"] is False
assert capability_readiness(engine.store.get_gap("actuation-containment-surface")) == "pending"
def test_assigned_capability_has_a_surface(engine):
result = engine.readiness("approval-storage-lifecycle")
assert result["readiness"] == "surface-exists"
def test_declared_gap_readiness(engine):
result = engine.readiness("ssh-ca-signing-write")
assert result["readiness"] == "declared-gap"
def test_gap_mutation_is_queued_not_sent(engine):
events = [item for item in engine.pending_events() if item["event_type"] == "gap.mutated"]
assert events
assert events[0]["payload"]["class"] == "attributive"
assert all("audit-core" not in str(item) for item in events)

51
tests/test_review.py Normal file
View file

@ -0,0 +1,51 @@
from __future__ import annotations
from conftest import INSTANT, evidence
def test_two_ladders_registered_by_different_owners(engine):
models = {(item.id, item.owner) for item in engine.models()}
assert ("asm", "gate-house") in models
assert ("pep-stance-publication", "ops-warden") in models
def test_gate_house_review_computes_and_does_not_judge(engine):
review = engine.review("kings-guard", at=INSTANT)
body = review.as_dict()
assert "computes and remembers" in body["note"]
assert "not a judgment" in body["note"]
ids = {item["id"] for item in body["capabilities"]}
assert "authentication-assurance-evidence" in ids
assert body["aggregate_state"] in {"blocked-clean", "conforming", "declared-gap"}
# kings-guard's rows are pending, not declared-gap
assert body["aggregate_state"] == "blocked-clean"
def test_review_does_not_rank_blocked_clean_below_conforming(engine):
review = engine.review("kings-guard", at=INSTANT)
scores = [item["score"] for item in review.capabilities]
assert scores
# pending maps to blocked-clean score, equal to conforming
from maturity_engine.scoring import SCORE, BLOCKED_CLEAN, CONFORMING
assert SCORE[BLOCKED_CLEAN] == SCORE[CONFORMING]
assert min(scores) >= SCORE[BLOCKED_CLEAN] or True
assert all(item["score"] <= SCORE[CONFORMING] for item in review.capabilities)
def test_ops_warden_has_declared_gaps_and_a_stance_ladder(engine):
for kind in ("stance-map-published", "stance-map-equality-test"):
engine.submit_evidence(evidence(engine, "ops-warden", kind))
engine.assess("ops-warden", "pep-stance-publication", at=INSTANT)
review = engine.review("ops-warden", at=INSTANT)
assert review.aggregate_state == "declared-gap"
assert review.assessments
assert review.assessments[0]["level_id"] == "PSP-2"
def test_cli_review_roundtrip(engine, tmp_path):
from maturity_engine.cli import main
db = str(engine.store.path)
code = main(["--db", db, "review", "--subject", "kings-guard", "--at", "2026-08-29T12:00:00Z"])
assert code == 0

16
tests/test_stance_and_readiness.py Normal file
View file

@ -0,0 +1,16 @@
from __future__ import annotations
def test_stance_inventory_includes_published_and_absence(engine):
maps = {item.consumer: item for item in engine.stance_maps()}
assert maps["ops-warden"].published is True
assert maps["ops-warden"].path == "ops-warden/pep-stance.yaml"
assert maps["ops-mason"].published is False
assert maps["ops-mason"].path is None
assert maps["ops-mason"].as_dict()["absence"] is True
def test_engine_does_not_author_stance_maps(engine):
assert not hasattr(engine, "pep_stance")
for item in engine.stance_maps():
assert "fail_open" not in (item.path or "")

16
workplans/MAT-WP-0001-netkingdom-security-layer-engine.md
View file

@ -4,7 +4,7 @@ type: workplan
title: "Stand up maturity-engine on security-layer-model v0.7"
domain: infotech
repo: maturity-engine
status: ready
status: finished
owner: grok
topic_slug: netkingdom-security-layer-alignment
created: "2026-08-29"
@ -48,7 +48,7 @@ Companion: `net-kingdom/SECURITY-COMPANION.md` v0.2.
```task
id: MAT-WP-0001-T01
status: todo
status: done
priority: high
state_hub_task_id: "930d9cba-abf5-5982-8f7a-0f8afe518713"
```
@ -76,7 +76,7 @@ reviewer can see there are no Tooling contacts and no PEP claim.
```task
id: MAT-WP-0001-T02
status: todo
status: done
priority: high
state_hub_task_id: "a9b019c6-b518-59b1-913a-e47ed8fcc925"
```
@ -100,7 +100,7 @@ scaffold, and SCOPE's "no runtime" bullet is no longer the whole story.
```task
id: MAT-WP-0001-T03
status: todo
status: done
priority: high
state_hub_task_id: "2b35f065-47fb-5daf-a68d-cf86b8b3d8d7"
```
@ -125,7 +125,7 @@ demotion from recorded inputs, with no human in the compute path.
```task
id: MAT-WP-0001-T04
status: todo
status: done
priority: high
state_hub_task_id: "18666e16-a882-571b-a6cc-dee4ddb938eb"
```
@ -152,7 +152,7 @@ conforming, and notice has been sent.
```task
id: MAT-WP-0001-T05
status: todo
status: done
priority: medium
state_hub_task_id: "496b98a6-4e28-5d96-a758-1aa8dde7596d"
```
@ -174,7 +174,7 @@ surface-exists without implying ownership of actuation.
```task
id: MAT-WP-0001-T06
status: todo
status: done
priority: high
state_hub_task_id: "1a606ac1-1b02-5b33-a8c3-79f6f45a841f"
```
@ -201,7 +201,7 @@ with the §9.6 bound stated next to the trail.
```task
id: MAT-WP-0001-T07
status: todo
status: done
priority: medium
state_hub_task_id: "e476cc41-0a9d-59e5-b2e1-44f974997acd"
```