89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
|
"""The agentic realization scenario: one semantic action, on a real surface.
|
||
|
|
|
||
|
|
Scope is deliberately one action. `grant_access(Bob, R, READ)` is realized
|
||
|
|
through the browser UI by an actor runtime that must *find* the control, while
|
||
|
|
everything else — setup, revocation, and every oracle — stays deterministic.
|
||
|
|
|
||
|
|
That split is the design: agentic flexibility at the realization layer only,
|
||
|
|
deterministic truth everywhere else.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import threading
|
||
|
|
from contextlib import contextmanager
|
||
|
|
|
||
|
|
from testdriver import (
|
||
|
|
Actor, Cast, DirectDriver, Oracle, Scenario, SemanticAction, StateObserver,
|
||
|
|
Step, VerificationAsset, World,
|
||
|
|
)
|
||
|
|
from testdriver.browser import BrowserDriver, Session, record_baseline_selectors
|
||
|
|
from testdriver.agentic import DiscoveryRuntime
|
||
|
|
from lab.http_api import serve
|
||
|
|
from lab.mutations import ObservationChannel, build_lab
|
||
|
|
from scenarios.alice_bob_carol import RESOURCE, USE_CASE
|
||
|
|
|
||
|
|
BROWSER = frozenset({"browser"})
|
||
|
|
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def lab_server(*mutations: str):
|
||
|
|
"""A live lab on a free port. The surface is real HTTP, not a function call."""
|
||
|
|
app, tokens = build_lab(*mutations)
|
||
|
|
app.request(tokens["alice"], "create_resource",
|
||
|
|
resource_id=RESOURCE, content="the secret")
|
||
|
|
server = serve(app)
|
||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
|
|
thread.start()
|
||
|
|
base_url = f"http://127.0.0.1:{server.server_address[1]}"
|
||
|
|
try:
|
||
|
|
yield app, tokens, base_url
|
||
|
|
finally:
|
||
|
|
server.shutdown()
|
||
|
|
server.server_close()
|
||
|
|
|
||
|
|
|
||
|
|
def baseline_recordings() -> dict[str, str]:
|
||
|
|
"""Capture the control arm's selectors once, against the unmutated surface."""
|
||
|
|
with lab_server() as (app, tokens, base_url):
|
||
|
|
session = Session(base_url, tokens["alice"])
|
||
|
|
_, html = session.get(f"/resources/{RESOURCE}/view")
|
||
|
|
return record_baseline_selectors(html)
|
||
|
|
|
||
|
|
|
||
|
|
def build_agentic(app, tokens, base_url, runtime=None):
|
||
|
|
"""One agentic step; deterministic oracles unchanged."""
|
||
|
|
cast = Cast()
|
||
|
|
for name in ("alice", "bob", "carol"):
|
||
|
|
cast.add(Actor(name, name.title(), credentials={"token": tokens[name]}))
|
||
|
|
|
||
|
|
world = World(id="w-browser", sut=app, sut_version=app.version, cast=cast)
|
||
|
|
|
||
|
|
scenario = Scenario(
|
||
|
|
id="sc-grant-via-browser",
|
||
|
|
use_case=USE_CASE,
|
||
|
|
variant=f"browser/{'+'.join(app.applied_mutations) or 'baseline'}",
|
||
|
|
watches=(), # filled from the reference scenario below
|
||
|
|
steps=(
|
||
|
|
Step("s2-grant", "alice", SemanticAction(
|
||
|
|
"grant_access",
|
||
|
|
{"subject_id": "bob", "permission": "READ"},
|
||
|
|
permitted_surfaces=BROWSER,
|
||
|
|
postcondition=lambda obs: obs["state_permission:bob:R"] == "READ",
|
||
|
|
)),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
from scenarios.alice_bob_carol import build as build_reference
|
||
|
|
_, _, reference_observer, _, _ = build_reference()
|
||
|
|
scenario = Scenario(
|
||
|
|
id=scenario.id, use_case=scenario.use_case, steps=scenario.steps,
|
||
|
|
watches=reference_observer.watches, variant=scenario.variant,
|
||
|
|
)
|
||
|
|
|
||
|
|
driver = BrowserDriver(base_url, tokens, runtime or DiscoveryRuntime(), RESOURCE)
|
||
|
|
observer = StateObserver(ObservationChannel(app), scenario.watches)
|
||
|
|
asset = VerificationAsset(
|
||
|
|
id="va-grant-via-browser", scenario=scenario, maturity="T1"
|
||
|
|
)
|
||
|
|
return world, driver, observer, asset, Oracle()
|