AFI parity: close all 35 gaps — every adapter wires every AFI-common capability

The conformance board (tests/afi/test_capability_parity.py) is now fully green:
90 capability cells + 4 meta-locks + 3 codegen byte-parity = 97 passed. The
gaps the prose table used to launder as "Django-only" / "out of scope" are
wired, against the pinned-spec model (single-authored spec, byte-identical
conformance across languages) — never per-language reimplementation.

FastAPI — edge_manifest + PSR (logic single-sourced in mizan_core.manifest),
WebSocket RPC (/ws/ through the shared dispatch), SSR (the framework-agnostic
SSRBridge relocated to mizan_core.ssr; Django rides it from there), Shapes
(SQLAlchemy projection, same declaration surface as django-readers), Forms
(Pydantic schema/validate/submit).

Rust (Axum + Tauri + cores/mizan-rust) — X-Mizan-Invalidate header, auth=
enforcement, origin HMAC cache, edge manifest + PSR, WebSocket handler / IPC
subscription channel, multipart upload, SSR bridge, Shapes, Forms; JWT/MWT
mint+verify and cache-key derivation byte-pinned to the Python reference
(cache_keys_pin, token_pin, invalidate_header_pin).

TypeScript — a KDL IR emitter byte-identical to the Python build_ir (so a TS
backend can feed the codegen — the largest gap), multipart upload, session-init,
WebSocket transport, SSR bridge, JWT/MWT mint (pinned to Python), Shapes, Forms.

Verified in the merged tree: core 25, fastapi 74, django 353/21-skip,
mizan-rust (incl. cross-language pins) green, axum 10, tauri 8, mizan-ts 103/2-skip.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-04 13:44:35 -04:00
parent 58d2cb2848
commit 6c5f6f1fba
81 changed files with 9893 additions and 463 deletions

View File

@@ -0,0 +1,167 @@
"""
Edge-manifest + PSR behavior — the genuine capability behind the
`edge_manifest` and `psr` probes.
Proves the FastAPI adapter emits the manifest the spec defines (contexts,
mutations, params, user_scoped, render_strategy, page_routes) by deriving it from
a real registry, and that `render_strategy` falls out of the user-scoped-param
rule: a context whose params overlap {user_id, user, owner_id, account_id} is
`dynamic_cached`, otherwise `psr`.
"""
from __future__ import annotations
import json
import subprocess
import sys
import pytest
from fastapi.responses import Response
import mizan_fastapi # registers the Starlette Response base for view-path detection
from mizan_core.client.function import client
from mizan_core.registry import clear_registry, register
from mizan_fastapi import edge_manifest, generate_edge_manifest
from mizan_fastapi.manifest import render_strategies
@pytest.fixture(autouse=True)
def _clean_registry():
clear_registry()
yield
clear_registry()
def _register(fn, name):
register(fn, name)
def test_user_scoped_context_is_dynamic_cached():
@client(context="user")
def user_profile(request, user_id: int) -> dict:
return {"id": user_id}
_register(user_profile, "user_profile")
manifest = edge_manifest()
ctx = manifest["contexts"]["user"]
assert ctx["user_scoped"] is True
assert ctx["render_strategy"] == "dynamic_cached"
assert ctx["params"] == ["user_id"]
assert ctx["endpoints"] == ["/api/mizan/ctx/user/"]
def test_non_user_scoped_context_is_psr():
@client(context="catalog")
def catalog_items(request, category: str) -> list[dict]:
return [{"category": category}]
_register(catalog_items, "catalog_items")
ctx = edge_manifest()["contexts"]["catalog"]
assert ctx["user_scoped"] is False
assert ctx["render_strategy"] == "psr"
def test_render_strategies_maps_each_context():
@client(context="user")
def me(request, user_id: int) -> dict:
return {"id": user_id}
@client(context="catalog")
def items(request) -> list[dict]:
return []
_register(me, "me")
_register(items, "items")
strategies = render_strategies()
assert strategies == {"user": "dynamic_cached", "catalog": "psr"}
def test_mutation_records_affects_and_auto_scope():
@client(context="user")
def user_profile(request, user_id: int) -> dict:
return {"id": user_id}
@client(affects="user")
def rename(request, user_id: int, name: str) -> dict:
return {"ok": True}
_register(user_profile, "user_profile")
_register(rename, "rename")
mutation = edge_manifest()["mutations"]["rename"]
assert mutation["affects"] == ["user"]
# user_id matches the context's param → auto-scoped
assert mutation["auto_scoped_params"] == ["user_id"]
def test_private_and_route_mutation_carried():
@client(affects="subscription", private=True, route="/webhooks/stripe/", methods=["POST"])
def stripe_webhook(request) -> Response:
return Response(status_code=200)
@client(context="subscription")
def subscription(request, user_id: int) -> dict:
return {"id": user_id}
_register(stripe_webhook, "stripe_webhook")
_register(subscription, "subscription")
mutation = edge_manifest()["mutations"]["stripe_webhook"]
assert mutation["private"] is True
assert mutation["route"] == "/webhooks/stripe/"
assert mutation["methods"] == ["POST"]
def test_view_path_function_records_route_and_page_routes():
@client(context="profile", route="/profile/<user_id>/")
def profile_page(request, user_id: int) -> Response:
return Response(status_code=200)
_register(profile_page, "profile_page")
ctx = edge_manifest()["contexts"]["profile"]
assert ctx["page_routes"] == ["/profile/<user_id>/"]
fn_entry = next(f for f in ctx["functions"] if f["name"] == "profile_page")
assert fn_entry["path"] == "view"
assert fn_entry["route"] == "/profile/<user_id>/"
def test_fastapi_manifest_matches_core_derivation():
"""The adapter callable is a thin pass-through to the shared core derivation."""
@client(context="user")
def user_profile(request, user_id: int) -> dict:
return {"id": user_id}
_register(user_profile, "user_profile")
assert edge_manifest() == generate_edge_manifest(base_url="/api/mizan")
def test_cli_entry_emits_manifest_json(tmp_path):
"""`mizan-fastapi-edge-manifest <module>` imports the module then prints JSON."""
app_module = tmp_path / "manifest_app.py"
app_module.write_text(
"from mizan_core.client.function import client\n"
"from mizan_core.registry import register\n"
"@client(context='user')\n"
"def user_profile(request, user_id: int) -> dict:\n"
" return {'id': user_id}\n"
"register(user_profile, 'user_profile')\n",
encoding="utf-8",
)
result = subprocess.run(
[sys.executable, "-m", "mizan_fastapi.manifest", "manifest_app", "--indent", "0"],
cwd=tmp_path,
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0, result.stderr
manifest = json.loads(result.stdout)
assert manifest["contexts"]["user"]["render_strategy"] == "dynamic_cached"