rythazhur/mizan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AFI parity: generate the matrix from conformance probes, not prose

Browse Source 
The per-adapter parity table was hand-maintained prose. An adapter that
never wired a capability (FastAPI SSR, Axum WebSocket) got its gap
relabelled "Django-only" or "out of scope — use native equivalents," and
nothing went red. The de-scope was crystallized in five mutually-ratifying
sites: the README §Stack-extensions table, the AFI fixture docstring
("channels/forms/shapes aren't AFI-common"), the core registry's
extension-hook framing, the mizan-fastapi __init__ docstring, and a
"CSRF is Django-only" comment in two adapters' session endpoints.

Replace prose-parity with conformance-generated parity:

- tests/afi/manifest.py declares the AFI-common surface as data — one list
  of capabilities, one of adapters. Applicability ("—") is derived from
  transport, never typed.
- tests/afi/probes.py independently inspects each backend's source for the
  artifact a capability requires (comment-stripped, backend-scoped). Green
  means wired; a cell can't be set by editing a word.
- tests/afi/test_capability_parity.py asserts every (capability × applicable
  adapter) pair is wired. 35 unwired gaps are now loud red TFDD tests, each
  naming an owed binding. No xfail/skip.
- tests/afi/parity_table.py generates the README table from the probes;
  `make parity-check` fails CI on any hand-edit, like the codegen byte-parity.

Purge the five de-scope sites. The IR byte-parity gate is unchanged and green.
`make test-afi` is now intentionally red on the 35 gaps — that board is the
owed parity work, itemized; a gap turns green by being wired, never described.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ryth Azhur
2026-06-04 12:58:03 -04:00
parent b41f469bbd
commit 58d2cb2848
11 changed files with 915 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/afi/fixture.py
View File

@@ -7,7 +7,11 @@ exercise the protocol axes both backends must agree on:
- two context functions sharing a param (proves bundling + param elevation)
- a mutation declaring `affects` on the context
No channels, no forms, no shapes — those aren't AFI-common.
This fixture is deliberately minimal: it exercises the axes the *IR-shape*
parity test (`test_codegen_parity.py`) needs to compare. It intentionally omits
channels / forms / shapes — NOT because those are outside the AFI (they are
AFI-common; see `manifest.py`), but because their per-adapter wiring is gated by
the *capability* parity suite (`test_capability_parity.py`), not by IR-shape.
`register_fixture()` registers the functions with mizan_core.registry.
Backend test apps import this module and call register_fixture() during

167
tests/afi/manifest.py Normal file
View File

@@ -0,0 +1,167 @@
"""
The AFI surface, as data — the single source of truth for what every Mizan
adapter owes.
This module exists because "what is AFI-common" used to live as prose: a
README table, a fixture docstring, an adapter `__init__` comment. Prose drifts.
An adapter that didn't wire a capability got its gap relabelled "Django-only"
or "out of scope — use native equivalents," and nothing went red. Here the
surface is a list of `Capability` objects and the implementors are a list of
`Adapter` objects; the parity table and the conformance suite are both
*generated* from these two lists. A capability cannot be de-scoped by editing
a word — only by deleting it from `CAPABILITIES`, which is a reviewable diff,
not a buried cell.
Applicability ("—" in the table) is DERIVED, never asserted: a capability that
`requires={"transport": "http"}` is simply not applicable to an IPC adapter.
The header-invalidation channel does not exist over Tauri IPC because IPC has
no headers — that is a fact about the transport, computed by `applies()`, not a
parity decision an agent gets to make.
The line between AFI-common and genuinely-backend-bound:
AFI-common Every adapter owes a binding. The protocol core, plus every
`register_extension` point (WebSocket, SSR, JWT, MWT, Shapes,
Forms). A missing one is a GAP (❌), never a category.
Backend-bound Not in this manifest at all. `allauth` is a Django-ecosystem
package — legitimately Django-only. The *bindings* of common
capabilities are backend-specific (django-readers is Django's
Shapes binding; Django Forms is Django's Forms binding) — but
the *capability* is common, so it lives here and each adapter
owes its own binding, not an "N/A."
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
class Tier(str, Enum):
"""How the README groups capabilities — presentation only, not semantics."""
PROTOCOL_CORE = "Protocol core"
EDGE_CACHE = "Edge, cache & enforcement"
EXTENSION = "Extension points"
@dataclass(frozen=True)
class Capability:
"""One thing every applicable adapter owes a binding for.
`requires` declares preconditions an adapter must meet for this capability
to APPLY. The only key today is `transport` ("http" | "ipc"); a capability
that names one is inapplicable to adapters on the other, and `applies()`
renders that as a derived "—". An empty `requires` means transport-agnostic:
every adapter owes it, full stop.
"""
id: str
title: str
tier: Tier
requires: dict[str, str] = field(default_factory=dict)
note: str = ""
@dataclass(frozen=True)
class Adapter:
"""A backend adapter and the declared properties parity is computed against.
`transport` is the load-bearing property: "http" adapters have a header
channel, an edge, a session endpoint; "ipc" adapters (Tauri) do not, and
those capabilities derive to "—" rather than counting as gaps.
"""
id: str
title: str
language: str # "python" | "rust" | "typescript"
transport: str # "http" | "ipc"
# ─── The AFI-common surface ───────────────────────────────────────────────────
CAPABILITIES: list[Capability] = [
# Protocol core — the wire contract every adapter implements.
Capability("rpc_call", "RPC call dispatch (`{result, invalidate}`)", Tier.PROTOCOL_CORE),
Capability("context_bundle", "Named-context bundle fetch", Tier.PROTOCOL_CORE),
Capability("invalidate_body", "Invalidation — JSON body", Tier.PROTOCOL_CORE),
Capability(
"invalidate_header", "Invalidation — `X-Mizan-Invalidate` header",
Tier.PROTOCOL_CORE, requires={"transport": "http"},
note="The header channel is co-equal with the body channel in the spec. "
"IPC transports carry invalidation in the response envelope instead.",
),
Capability("invalidate_autoscope", "Invalidation auto-scoping (three-tier)", Tier.PROTOCOL_CORE),
Capability("registration", "Function discovery / registration", Tier.PROTOCOL_CORE),
Capability("ir_export", "Codegen IR export (KDL)", Tier.PROTOCOL_CORE),
Capability("upload", "File uploads (`Upload` type)", Tier.PROTOCOL_CORE),
# Edge, cache & enforcement.
Capability("auth_enforcement", "Auth-guard enforcement (`auth=…` rejects)", Tier.EDGE_CACHE),
Capability("origin_cache", "Origin-side HMAC cache", Tier.EDGE_CACHE),
Capability(
"edge_manifest", "Edge manifest export", Tier.EDGE_CACHE,
requires={"transport": "http"},
note="The manifest configures an HTTP/CDN edge; a desktop IPC shell has no edge.",
),
Capability(
"psr", "PSR (`render_strategy` in manifest)", Tier.EDGE_CACHE,
requires={"transport": "http"},
),
Capability(
"session_init", "Session / CSRF init endpoint", Tier.EDGE_CACHE,
requires={"transport": "http"},
),
# Extension points — the `register_extension` surface. AFI-common, every one.
Capability("websocket", "WebSocket transport (`websocket=` declared)", Tier.EXTENSION),
Capability("ssr_bridge", "SSR bridge (subprocess renderer)", Tier.EXTENSION),
Capability("jwt", "JWT auth (access / refresh)", Tier.EXTENSION),
Capability(
"mwt", "MWT (edge identity token)", Tier.EXTENSION,
requires={"transport": "http"},
note="MWT exists to key an edge cache; without an edge there is nothing to key.",
),
Capability(
"shapes", "Typed query projection (Shapes)", Tier.EXTENSION,
note="The capability is AFI-common; the binding is per-ORM "
"(django-readers on Django, the project's ORM elsewhere).",
),
Capability(
"forms", "Forms (schema / validate / submit)", Tier.EXTENSION,
note="The capability is AFI-common; the binding is per-framework "
"(Django Forms on Django, Pydantic-or-equivalent elsewhere).",
),
]
# ─── The implementors ─────────────────────────────────────────────────────────
ADAPTERS: list[Adapter] = [
Adapter("django", "Django", "python", "http"),
Adapter("fastapi", "FastAPI", "python", "http"),
Adapter("rust_axum", "Rust / Axum", "rust", "http"),
Adapter("tauri", "Tauri", "rust", "ipc"),
Adapter("typescript", "TypeScript", "typescript", "http"),
]
CAPABILITIES_BY_ID: dict[str, Capability] = {c.id: c for c in CAPABILITIES}
ADAPTERS_BY_ID: dict[str, Adapter] = {a.id: a for a in ADAPTERS}
# Every capability owes exactly one probe in probes.py. The meta-conformance
# test pins this so a capability can't be added without its gate.
PROBES_REQUIRED: int = len(CAPABILITIES)
def applies(capability: Capability, adapter: Adapter) -> bool:
"""Whether `capability` is applicable to `adapter`, from declared properties.
This is the ONLY source of "—" in the parity table. A `False` here is a
transport fact (IPC has no HTTP header channel), not a parity verdict.
"""
required_transport = capability.requires.get("transport")
if required_transport is not None and adapter.transport != required_transport:
return False
return True

141
tests/afi/parity_table.py Normal file
View File

@@ -0,0 +1,141 @@
"""
Generate the README parity table from the conformance probes.
The table in the README is *output*, not *input*. It is computed by running
every probe over every adapter and rendering the result. An agent can no longer
type "Django-only" into a cell, because no one types cells — `make parity-table`
overwrites the block between the markers, and `--check` fails CI if the
committed block has drifted from what the probes actually report (the same
forcing function the codegen byte-parity tests already use).
Glyphs:
✅ wired the probe found the artifact
◑ partial declared or stubbed, not complete — counts as RED in the suite
❌ gap AFI-common, owed, not wired
— n/a the capability does not exist over this adapter's transport
(derived from `manifest.applies`, never asserted)
Usage:
python parity_table.py --write # regenerate the README block
python parity_table.py --check # exit 1 if README block is stale
python parity_table.py # print the block to stdout
"""
from __future__ import annotations
import sys
from pathlib import Path
from manifest import ADAPTERS, CAPABILITIES, Tier, applies
from probes import run_probe
README = Path(__file__).resolve().parents[2] / "README.md"
START = "<!-- MIZAN:PARITY:START — generated by tests/afi/parity_table.py; do not edit by hand -->"
END = "<!-- MIZAN:PARITY:END -->"
_GLYPH = {"pass": "✅", "partial": "◑", "fail": "❌"}
def _cell(cap, adapter) -> str:
if not applies(cap, adapter):
return "—"
return _GLYPH[run_probe(cap.id, adapter).state]
def _tier_table(tier: Tier) -> str:
caps = [c for c in CAPABILITIES if c.tier is tier]
if not caps:
return ""
header = "| Capability | " + " | ".join(a.title for a in ADAPTERS) + " |"
sep = "|---|" + "|".join(":---:" for _ in ADAPTERS) + "|"
rows = []
for cap in caps:
cells = " | ".join(_cell(cap, a) for a in ADAPTERS)
rows.append(f"| {cap.title} | {cells} |")
return f"### {tier.value}\n\n{header}\n{sep}\n" + "\n".join(rows)
def _notes() -> str:
noted = [c for c in CAPABILITIES if c.note]
if not noted:
return ""
lines = ["**Notes**", ""]
for c in noted:
lines.append(f"- **{c.title}** — {c.note}")
return "\n".join(lines)
def generate_block() -> str:
"""The full generated parity block: legend, one table per tier, notes."""
legend = (
"Legend: ✅ wired · ◑ partial (declared/stubbed) · ❌ gap (AFI-common, owed) · "
"— not applicable to this adapter's transport\n\n"
"Every capability below is **AFI-common**: each adapter owes a binding, and a "
"❌ is a gap on the owed-work board (`tests/afi/`), never a category. "
"Backend-specific *bindings* of common capabilities (django-readers for Shapes, "
"Django Forms for Forms) and genuinely Django-ecosystem features (allauth) are "
"out of this matrix by design — see `tests/afi/manifest.py` for the line."
)
parts = [legend]
for tier in (Tier.PROTOCOL_CORE, Tier.EDGE_CACHE, Tier.EXTENSION):
table = _tier_table(tier)
if table:
parts.append(table)
notes = _notes()
if notes:
parts.append(notes)
return "\n\n".join(parts)
def _wrap(block: str) -> str:
return f"{START}\n{block}\n{END}"
def _splice(readme_text: str, block: str) -> str:
if START not in readme_text or END not in readme_text:
raise SystemExit(
f"README markers not found. Add a block bounded by:\n {START}\n {END}\n"
f"to {README} where the parity table should render."
)
pre = readme_text.split(START)[0]
post = readme_text.split(END)[1]
return pre + _wrap(block) + post
def main(argv: list[str]) -> int:
block = generate_block()
mode = argv[1] if len(argv) > 1 else "--print"
if mode == "--print":
print(block)
return 0
text = README.read_text(encoding="utf-8")
spliced = _splice(text, block)
if mode == "--write":
if spliced != text:
README.write_text(spliced, encoding="utf-8")
print(f"Wrote parity table to {README}")
else:
print("Parity table already current.")
return 0
if mode == "--check":
if spliced != text:
print(
"README parity table is STALE. The committed table does not match what "
"the conformance probes report.\nRun: make parity-table",
file=sys.stderr,
)
return 1
print("README parity table is current.")
return 0
print(f"Unknown mode: {mode}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main(sys.argv))

385
tests/afi/probes.py Normal file
View File

@@ -0,0 +1,385 @@
"""
Capability probes — the adversarial half of the conformance gate.
A probe answers one question about one (capability, adapter) pair: *is this
capability actually wired into this adapter?* It answers by inspecting the
backend's own source for the concrete artifact the capability requires — a
route registration, an exported renderer, a management command, a call into the
shared core. It never asks the adapter to self-report (a self-report is the
thing that drifts), and it never reads a parity table (that would be circular).
The single invariant: **a probe goes green only when the wiring exists.** You
cannot flip a cell by editing prose, a docstring, or a README — only by adding
the route / module / binding the probe looks for.
Two facts learned by building this and watching it lie, now load-bearing:
1. Comments are the false-positive engine. A doc-comment that says "the JS side
re-wraps it into a Promise.reject" made an auth-enforcement probe pass on the
word `reject`. So source is comment-stripped per language before matching,
and patterns target code identifiers (`MizanError::Unauthorized`, a route
macro, an exported function name), never prose-able words.
2. AFI-common logic lives in the shared core; the adapter rides it. FastAPI's
JWT/MWT support *is* `mizan_core.auth.authenticate` — there is no `jwt`
literal in the FastAPI adapter, because the adapter delegates. Axum's KDL IR
lives in `cores/mizan-rust`, not the axum crate. So capabilities that are
genuinely core-provided (`jwt`, `mwt`, `ir_export`) probe the BACKEND scope
(adapter + its language core); capabilities the adapter must wire to its own
transport (the header on the response, the route, the manifest command, the
WebSocket handler) probe the ADAPTER scope only.
Depth. These are *source-surface* probes: they confirm the artifact is present,
not that it behaves correctly under load. Runtime-behavior depth (mount, drive,
assert the wire bytes) is owed per adapter and lands with each gap's closure —
the runtime probe is written alongside the wiring it verifies. Surface-presence
is the floor that makes de-scope impossible; runtime is the ceiling each closure
raises, and the gap between them is named here, not hidden.
`partial` (◑) is a real state: Axum declares `Transport::Websocket` in the IR
but routes no handler; mizan-ts decodes a JWT but mints none. It renders ◑ — and
the conformance suite still treats it as RED, because the suite asserts
`state == "pass"`. ◑ is visible honesty, never a shippable-green hiding place.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Callable, Literal
from manifest import Adapter
REPO_ROOT = Path(__file__).resolve().parents[2]
BACKENDS = REPO_ROOT / "backends"
CORES = REPO_ROOT / "cores"
# Each adapter's own source root. Probes scoped "adapter" read only here.
ADAPTER_ROOTS: dict[str, Path] = {
"django": BACKENDS / "mizan-django" / "src" / "mizan",
"fastapi": BACKENDS / "mizan-fastapi" / "src" / "mizan_fastapi",
"rust_axum": BACKENDS / "mizan-rust-axum" / "src",
"tauri": BACKENDS / "mizan-tauri" / "src",
"typescript": BACKENDS / "mizan-ts" / "src",
}
# The shared core each language's adapters ride. Probes scoped "backend" read
# the adapter root PLUS these — that is where core-provided AFI-common logic
# (token minting, KDL IR) actually lives.
CORE_ROOTS: dict[str, list[Path]] = {
"python": [CORES / "mizan-python" / "src" / "mizan_core"],
"rust": [CORES / "mizan-rust" / "src", CORES / "mizan-rust-macros" / "src"],
"typescript": [], # mizan-ts is self-contained; no shared core
}
_SOURCE_EXTS = {".py", ".rs", ".ts", ".tsx"}
_SKIP_DIRS = {"node_modules", "target", "__pycache__", "dist", ".venv"}
State = Literal["pass", "partial", "fail"]
@dataclass(frozen=True)
class ProbeResult:
state: State
detail: str
# ─── Source collection + comment stripping ────────────────────────────────────
def _strip_comments(text: str, language: str) -> str:
"""Remove comments so prose can't satisfy a code-artifact pattern.
Imperfect by design (a `#` inside a string may be clipped); the patterns are
code identifiers that live outside strings, and the cost of a rare false
strip is far below the cost of a comment-driven false `pass`.
"""
if language in ("rust", "typescript"):
text = re.sub(r"/\*.*?\*/", "", text, flags=re.DOTALL) # block /* */ (incl. /** */)
text = re.sub(r"(?<!:)//[^\n]*", "", text) # line // (keep URL `://`)
elif language == "python":
text = re.sub(r'""".*?"""', "", text, flags=re.DOTALL) # triple-quoted
text = re.sub(r"'''.*?'''", "", text, flags=re.DOTALL)
text = re.sub(r"#[^\n]*", "", text) # line #
return text
def _read_roots(roots: tuple[Path, ...], language: str) -> str:
chunks: list[str] = []
for root in roots:
if not root.exists():
continue
for path in sorted(root.rglob("*")):
if not (path.is_file() and path.suffix in _SOURCE_EXTS):
continue
if any(part in _SKIP_DIRS for part in path.parts):
continue
chunks.append(_strip_comments(path.read_text(encoding="utf-8", errors="replace"), language))
return "\n".join(chunks)
@lru_cache(maxsize=None)
def _adapter_text(adapter_id: str, language: str) -> str:
return _read_roots((ADAPTER_ROOTS[adapter_id],), language)
@lru_cache(maxsize=None)
def _backend_text(adapter_id: str, language: str) -> str:
roots = (ADAPTER_ROOTS[adapter_id], *CORE_ROOTS.get(language, []))
return _read_roots(tuple(roots), language)
def _adapter(a: Adapter) -> str:
return _adapter_text(a.id, a.language)
def _backend(a: Adapter) -> str:
return _backend_text(a.id, a.language)
def _has_path(adapter_id: str, *relparts: str) -> bool:
return ADAPTER_ROOTS[adapter_id].joinpath(*relparts).exists()
def _hit(text: str, pattern: str) -> bool:
return re.search(pattern, text) is not None
def _wired(found: bool, what: str) -> ProbeResult:
return ProbeResult("pass", f"wired: {what}") if found else ProbeResult("fail", f"missing: {what}")
# ─── Per-capability detection ─────────────────────────────────────────────────
#
# Patterns are grounded in artifacts read directly from each adapter:
# django urls.py `path("call/", ...)`, channels/ ssr/ shapes/ forms/ dirs
# fastapi router.py `@router.post("/call/")`, header insert, /session/
# axum handlers.rs `function_call`, `compute_invalidation`, no header/cache
# tauri lib.rs `mizan_invoke`, Envelope::Call/Fetch, no auth/cache/ws
# ts dispatch.ts `handleMutationCall`, checkAuth, token.ts decode-only
def _probe_rpc_call(a: Adapter) -> ProbeResult:
pat = {
"python": r'path\("call/|/call/|function_call',
"rust": r"function_call|mizan_invoke|Envelope::Call",
"typescript": r"handleMutationCall",
}[a.language]
return _wired(_hit(_adapter(a), pat), "RPC call dispatch")
def _probe_context_bundle(a: Adapter) -> ProbeResult:
pat = {
"python": r"context_fetch|/ctx/|ctx/<",
"rust": r"context_fetch|handle_fetch|Envelope::Fetch",
"typescript": r"handleContextFetch",
}[a.language]
return _wired(_hit(_adapter(a), pat), "named-context bundle fetch")
def _probe_invalidate_body(a: Adapter) -> ProbeResult:
pat = {
"python": r'"invalidate"|invalidate=|\binvalidate\b',
"rust": r"invalidate|InvalidationTarget",
"typescript": r"\.invalidate\b|invalidate:",
}[a.language]
return _wired(_hit(_adapter(a), pat), "`invalidate` key in response body")
def _probe_invalidate_header(a: Adapter) -> ProbeResult:
return _wired(_hit(_adapter(a), r"X-Mizan-Invalidate"), "`X-Mizan-Invalidate` header emission")
def _probe_invalidate_autoscope(a: Adapter) -> ProbeResult:
# The adapter wires auto-scoping by routing through the shared resolver.
pat = {
"python": r"resolve_invalidation|dispatch_call|dispatch_context",
"rust": r"compute_invalidation",
"typescript": r"resolveInvalidation",
}[a.language]
return _wired(_hit(_adapter(a), pat), "three-tier invalidation auto-scoping (shared resolver)")
def _probe_registration(a: Adapter) -> ProbeResult:
pat = {
"python": r"\bregister\b|get_function|registry",
"rust": r"FUNCTIONS|lookup_function|linkme",
"typescript": r"getFunction|\bregister\b|registry",
}[a.language]
return _wired(_hit(_adapter(a), pat), "function discovery / registration")
def _probe_ir_export(a: Adapter) -> ProbeResult:
# Core-provided: python build_ir/export cmd (adapter), rust kdl.rs/ir.rs (core).
if a.language == "python":
return _wired(_hit(_adapter(a), r"build_ir|export_mizan_ir"), "KDL IR export")
if a.language == "rust":
return _wired(_hit(_backend(a), r"to_kdl|emit_kdl|fn build_ir|kdl::"), "KDL IR export (rust core)")
# TypeScript: emits the edge manifest (JSON), but no codegen KDL IR (note 8).
if _hit(_adapter(a), r"to_kdl|emitKdl|buildIr|\bkdl\b"):
return ProbeResult("pass", "wired: KDL IR emitter")
return ProbeResult("fail", "missing: KDL IR emitter (manifest.ts emits edge JSON, not codegen IR)")
def _probe_upload(a: Adapter) -> ProbeResult:
pat = {
"python": r"bind_uploads|UploadedFile|multipart",
"rust": r"Multipart|multipart|bind_uploads",
"typescript": r"FormData|multipart",
}[a.language]
return _wired(_hit(_adapter(a), pat), "multipart / Upload-type dispatch binding")
def _probe_auth_enforcement(a: Adapter) -> ProbeResult:
if a.language == "python":
return _wired(_hit(_adapter(a), r"enforce_auth|authenticate\(|authguard"), "auth= enforcement")
if a.language == "typescript":
return _wired(_hit(_adapter(a), r"checkAuth|authDenial|AuthDenial"), "auth= enforcement")
# Rust adapters (axum, tauri): dispatch must reject on auth=.
if _hit(_adapter(a), r"Unauthorized|Forbidden|enforce_auth"):
return ProbeResult("pass", "wired: auth enforcement in dispatch")
if _hit(_backend(a), r"\bauth\b|is_private|\bprivate\b"):
return ProbeResult("partial", "auth/private carried in FunctionSpec; dispatch does not reject")
return ProbeResult("fail", "missing: auth= enforcement")
def _probe_origin_cache(a: Adapter) -> ProbeResult:
pat = {
"python": r"CacheOrchestrator|derive_cache_key|cfg\.cache|cache_get|cache_put",
"rust": r"derive_cache_key|CacheBackend|origin_cache",
"typescript": r"cacheGet|cachePut|deriveCacheKey",
}[a.language]
return _wired(_hit(_backend(a), pat), "origin-side HMAC cache")
def _probe_edge_manifest(a: Adapter) -> ProbeResult:
pat = {
"python": r"export_edge_manifest|generate_edge_manifest|edge_manifest",
"rust": r"edge_manifest|generate_manifest",
"typescript": r"generateManifest|EdgeManifest",
}[a.language]
return _wired(_hit(_adapter(a), pat), "edge manifest export")
def _probe_psr(a: Adapter) -> ProbeResult:
return _wired(_hit(_adapter(a), r"render_strategy|renderStrategy"), "PSR render_strategy")
def _probe_session_init(a: Adapter) -> ProbeResult:
pat = {
"python": r"session/|session_init",
"rust": r"session_init|/session/",
"typescript": r"sessionInit|/session/",
}[a.language]
return _wired(_hit(_adapter(a), pat), "session / CSRF init endpoint")
def _probe_websocket(a: Adapter) -> ProbeResult:
if a.id == "django":
return _wired(_has_path(a.id, "channels"), "Django Channels consumer")
if a.id == "fastapi":
return _wired(_hit(_adapter(a), r"WebSocket|@router\.websocket|websocket_route"), "FastAPI WebSocket route")
if a.id == "rust_axum":
if _hit(_adapter(a), r"WebSocketUpgrade|on_upgrade|ws_handler"):
return ProbeResult("pass", "wired: Axum WebSocket handler")
if _hit(_backend(a), r"Websocket|WebSocket"):
return ProbeResult("partial", "Transport::Websocket declared in IR; no Axum handler routed")
return ProbeResult("fail", "missing: WebSocket transport")
if a.id == "tauri":
return _wired(_hit(_adapter(a), r"subscription|Subscribe|emit_to|Channel<"), "IPC subscription channel")
return _wired(_hit(_adapter(a), r"WebSocket|websocket"), "WebSocket transport")
def _probe_ssr_bridge(a: Adapter) -> ProbeResult:
if a.id == "django":
return _wired(_has_path(a.id, "ssr", "bridge.py"), "Bun SSR subprocess bridge")
return _wired(_hit(_adapter(a), r"SSRBridge|renderToString|ssr_bridge"), "SSR bridge (subprocess renderer)")
def _probe_jwt(a: Adapter) -> ProbeResult:
# Core-provided for Python (mizan_core.auth); TS has decode-only helpers.
if a.language == "python":
return _wired(_hit(_backend(a), r"\bjwt\b|JWT|def authenticate"), "JWT auth (rides mizan_core.auth)")
if a.id == "typescript":
if _hit(_adapter(a), r"signJwt|mintJwt|createJwt|encodeJwt"):
return ProbeResult("pass", "wired: JWT mint + verify")
if _hit(_adapter(a), r"decodeJwt|decodeJwtBearer"):
return ProbeResult("partial", "decode-only helper; mints no tokens")
return ProbeResult("fail", "missing: JWT auth")
return _wired(_hit(_backend(a), r"\bjwt\b|JWT"), "JWT auth (access / refresh)")
def _probe_mwt(a: Adapter) -> ProbeResult:
if a.language == "python":
return _wired(_hit(_backend(a), r"\bmwt\b|MWT|X-Mizan-Token"), "MWT (edge identity token)")
if a.id == "typescript":
if _hit(_adapter(a), r"signMwt|mintMwt|createMwt"):
return ProbeResult("pass", "wired: MWT mint")
if _hit(_adapter(a), r"decodeMwt|identityFromMwt"):
return ProbeResult("partial", "decode-only helper; mints no tokens")
return ProbeResult("fail", "missing: MWT")
return _wired(_hit(_backend(a), r"\bmwt\b|MWT"), "MWT (edge identity token)")
def _probe_shapes(a: Adapter) -> ProbeResult:
if a.id == "django":
return _wired(_has_path(a.id, "shapes"), "django-readers Shapes binding")
pat = {
"python": r"django_readers|QuerysetProjection|shapes\.",
"rust": r"shapes::|QueryProjection",
"typescript": r"QueryProjection|shapesBinding",
}[a.language]
return _wired(_hit(_adapter(a), pat), "typed query-projection binding")
def _probe_forms(a: Adapter) -> ProbeResult:
if a.id == "django":
return _wired(_has_path(a.id, "forms"), "Django Forms binding (schema / validate / submit)")
if a.language == "rust":
if _hit(_adapter(a), r"form_submit|form_validate|FormRole::Submit"):
return ProbeResult("pass", "wired: form validate/submit endpoint")
if _hit(_backend(a), r"is_form|form_role|FormRole"):
return ProbeResult("partial", "is_form/form_role carried; no validate/submit endpoint")
return ProbeResult("fail", "missing: forms")
pat = {
"python": r"FormSchema|form_role|forms\.",
"typescript": r"FormSchema|formRole|formSubmit",
}[a.language]
return _wired(_hit(_adapter(a), pat), "forms (schema / validate / submit)")
PROBES: dict[str, Callable[[Adapter], ProbeResult]] = {
"rpc_call": _probe_rpc_call,
"context_bundle": _probe_context_bundle,
"invalidate_body": _probe_invalidate_body,
"invalidate_header": _probe_invalidate_header,
"invalidate_autoscope": _probe_invalidate_autoscope,
"registration": _probe_registration,
"ir_export": _probe_ir_export,
"upload": _probe_upload,
"auth_enforcement": _probe_auth_enforcement,
"origin_cache": _probe_origin_cache,
"edge_manifest": _probe_edge_manifest,
"psr": _probe_psr,
"session_init": _probe_session_init,
"websocket": _probe_websocket,
"ssr_bridge": _probe_ssr_bridge,
"jwt": _probe_jwt,
"mwt": _probe_mwt,
"shapes": _probe_shapes,
"forms": _probe_forms,
}
def run_probe(capability_id: str, adapter: Adapter) -> ProbeResult:
"""Run the probe for one (capability, adapter) pair."""
probe = PROBES.get(capability_id)
if probe is None:
raise KeyError(
f"No probe registered for capability '{capability_id}'. Every capability "
f"in manifest.CAPABILITIES owes a probe — add one to PROBES."
)
return probe(adapter)

118
tests/afi/test_capability_parity.py Normal file
View File

@@ -0,0 +1,118 @@
"""
AFI capability parity — the runtime/surface conformance gate.
`test_codegen_parity.py` gates that the three backends emit byte-identical KDL.
That is necessary but narrow: it proves the IR agrees, not that an adapter
actually *implements* the capabilities the IR describes. The vacuum left by
"IR-shape only" is exactly where parity drifted — an adapter that never wired
SSR or WebSocket got its gap relabelled "Django-only" or "out of scope," and
nothing in the suite objected.
This module closes that vacuum. It parametrizes over every (capability,
applicable-adapter) pair drawn from `manifest.py` and asserts the adapter
actually wires the capability (`probes.py`). It is designed to be RED wherever a
gap is real — each failure names one owed binding. That redness is not a broken
build; it is the board of owed work, itemized and loud, that the prior prose
table hid behind false-green. A gap turns green by being *wired*, never by being
*described*.
Applicability is derived in `manifest.applies()` from the adapter's declared
transport, so a capability that simply does not exist over a transport (header
invalidation over Tauri IPC) is not parametrized here at all — it is a "—" in
the generated table, computed, not a verdict anyone typed.
"""
from __future__ import annotations
import pytest
from manifest import ADAPTERS, CAPABILITIES, CAPABILITIES_BY_ID, PROBES_REQUIRED, applies
from probes import PROBES, run_probe
def _applicable_pairs() -> list[tuple[str, str, str]]:
"""(capability_id, adapter_id, test_id) for every pair the protocol applies to."""
pairs: list[tuple[str, str, str]] = []
for cap in CAPABILITIES:
for adapter in ADAPTERS:
if applies(cap, adapter):
pairs.append((cap.id, adapter.id, f"{cap.id}::{adapter.id}"))
return pairs
_PAIRS = _applicable_pairs()
@pytest.mark.parametrize(
"capability_id,adapter_id",
[(c, a) for c, a, _ in _PAIRS],
ids=[tid for _, _, tid in _PAIRS],
)
def test_adapter_wires_capability(capability_id: str, adapter_id: str) -> None:
"""The adapter must wire the AFI-common capability the protocol declares.
A failure here is one owed binding, not a regression. The message names the
capability, the adapter, and what the probe could not find — that string is
the gap's specification until someone closes it by wiring the artifact.
"""
adapter = next(a for a in ADAPTERS if a.id == adapter_id)
cap = CAPABILITIES_BY_ID[capability_id]
result = run_probe(capability_id, adapter)
assert result.state == "pass", (
f"AFI parity gap — {adapter.title} does not wire '{cap.title}'.\n"
f" probe: {result.detail}\n"
f" state: {result.state}"
+ (" (◑ partial — declared/stubbed but not complete)" if result.state == "partial" else "")
+ f"\n This capability is AFI-common (manifest tier: {cap.tier.value}); every "
f"adapter owes a binding. Close it by wiring the artifact the probe looks for — "
f"not by editing a table."
)
# ─── Meta-conformance: the manifest and the probes must stay in lockstep ───────
def test_every_capability_has_a_probe() -> None:
"""No capability may be declared without a probe — else it is unverifiable
and silently 'passes' by never being checked, recreating the original hole."""
missing = [c.id for c in CAPABILITIES if c.id not in PROBES]
assert not missing, (
f"Capabilities declared in manifest.py with no probe in probes.py: {missing}. "
f"An unprobed capability is an un-gated parity claim — exactly the drift this "
f"suite exists to prevent."
)
def test_no_orphan_probes() -> None:
"""No probe may exist for a capability the manifest doesn't declare — that
would be dead detection code drifting from the surface it claims to check."""
orphans = [pid for pid in PROBES if pid not in CAPABILITIES_BY_ID]
assert not orphans, (
f"Probes in probes.py with no matching capability in manifest.py: {orphans}."
)
def test_probe_count_matches_required() -> None:
"""Sanity pin: the manifest's own count of required probes equals the probe set."""
assert len(PROBES) == PROBES_REQUIRED, (
f"probes.py defines {len(PROBES)} probes; manifest expects {PROBES_REQUIRED}."
)
def test_readme_parity_table_is_current() -> None:
"""The README parity table is generated output; a hand-edit must fail here.
This is the lock that makes the original lie inexpressible. The table can no
longer be edited to read 'Django-only' — it is spliced from the probe results
by `parity_table.py`, and this test asserts the committed block matches a fresh
regeneration. Drift → red → `make parity-table`.
"""
import parity_table
text = parity_table.README.read_text(encoding="utf-8")
regenerated = parity_table._splice(text, parity_table.generate_block())
assert regenerated == text, (
"README parity table is stale or hand-edited. It is generated from the "
"conformance probes — run `make parity-table` to regenerate it."
)