"""Turning a stable agentic realization into deterministic code. The thesis in one module: once an agent has found the same path enough times, the finding itself is the valuable part, and repeating the search is waste. What is crystallized is the *realization* — how to perform a semantic action on this surface. What is never crystallized, and never re-authored, is the judgment. The descendant imports its ancestor's claim predicates rather than restating them. That is deliberate: a generated test that paraphrases its assertions has introduced a second, unverified statement of intent, and any drift between the two is silent. Importing makes oracle preservation a fact rather than a hope. Reversibility is part of the design (`INTENT.md`: "Crystallization is reversible"). A descendant that stops matching its ancestor is evidence the surface moved again, and the asset returns to its agentic form rather than being patched. """ from __future__ import annotations import json from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Mapping, Sequence from .actions import SemanticAction from .drivers import Realization from .world import Actor @dataclass(frozen=True, slots=True) class Trajectory: """How one semantic action was actually performed, reduced to what replays.""" step_id: str action_name: str surface_id: str method: str target: str fields: tuple[str, ...] def key(self) -> str: return json.dumps( { "step": self.step_id, "action": self.action_name, "surface": self.surface_id, "method": self.method, "target": self.target, "fields": sorted(self.fields), }, sort_keys=True, ) def capture(pack: Mapping[str, Any]) -> tuple[Trajectory, ...]: """Extract the realization path from one Evidence Pack.""" out: list[Trajectory] = [] for obs in pack["observations"]: if obs["kind"] != "realization" or obs["data"].get("raised"): continue mechanics = obs["data"].get("mechanics", {}) target = mechanics.get("target") or mechanics.get("operation") or "" fields = mechanics.get("fields") or sorted(mechanics.get("arguments") or {}) out.append(Trajectory( step_id=obs["step_id"], action_name=str(mechanics.get("action", "")).split("(")[0], surface_id=obs["data"].get("surface", ""), method=("POST" if obs["data"].get("surface") == "browser" else "CALL"), target=target, fields=tuple(sorted(fields)), )) return tuple(out) @dataclass(frozen=True, slots=True) class StabilityReport: stable: bool observations: int distinct_paths: int reason: str trajectories: tuple[Trajectory, ...] = () def assess_stability(packs: Sequence[Mapping[str, Any]], minimum: int = 3) -> StabilityReport: """Is this realization settled enough to freeze? Requires the *same* path across several runs. One successful run proves the agent can find a way; it does not show the surface has stopped moving, and freezing on a single observation is how a crystallized test becomes flaky the first time a page renders differently. """ if len(packs) < minimum: return StabilityReport( False, len(packs), 0, f"need at least {minimum} runs to judge stability, have {len(packs)}", ) captured = [capture(pack) for pack in packs] keys = {tuple(t.key() for t in trajectory) for trajectory in captured} if len(keys) != 1: return StabilityReport( False, len(packs), len(keys), f"realization varied across runs ({len(keys)} distinct paths); " "the surface is still moving", ) if not captured[0]: return StabilityReport(False, len(packs), 0, "no successful realization to freeze") return StabilityReport( True, len(packs), 1, f"identical realization across {len(packs)} runs", captured[0], ) # --- the deterministic descendant ---------------------------------------- class CrystallizedDriver: """Replays a frozen trajectory. No discovery, no runtime, no model. Deliberately fails rather than searching when the recorded path no longer works. A driver that fell back to discovery would quietly turn a T5 asset back into a T1 one and hide the fact that the surface had moved — which is exactly the signal crystallization is supposed to surface. """ name = "crystallized-driver" def __init__(self, trajectories: Sequence[Trajectory], session_factory) -> None: self._by_step = {t.step_id: t for t in trajectories} self._by_action = {t.action_name: t for t in trajectories} self._session_factory = session_factory self.surface = None # set per action; a frozen path may span surfaces def realize(self, actor: Actor, action: SemanticAction) -> Realization: trajectory = self._by_action.get(action.name) if trajectory is None: return Realization( "crystallized", {"action": action.name}, raised="NotCrystallized: no frozen path for this action", ) action.check_surface(trajectory.surface_id) args = dict(action.args) fields = {name: str(args[name]) for name in trajectory.fields if name in args} session = self._session_factory(actor) mechanics: dict[str, Any] = { "action": action.describe(), "runtime": self.name, "target": trajectory.target, "fields": sorted(fields), "metrics": { "runtime": self.name, "wall_time_ms": 0.0, "candidates_considered": 0, "attempts": 1, "retries": 0, "tokens_in": 0, "tokens_out": 0, "model": None, }, "actor": actor.id, } status, body = session.post_form(trajectory.target, fields) mechanics["status"] = status if status >= 400: return Realization(trajectory.surface_id, mechanics, raised=f"Refused: {status} {body[:120]}") return Realization(trajectory.surface_id, mechanics) # --- code generation ------------------------------------------------------ _TEMPLATE = '''"""Crystallized regression test — generated, do not edit by hand. Lineage ------- ancestor asset : {ancestor_id} ancestor maturity: {ancestor_maturity} descendant : {descendant_id} (T5 Deterministic) frozen from : {runs} identical realizations surface version: {sut_version} generated : {generated_at} Why this file exists -------------------- An agent discovered this path {runs} times running and it did not change. The search is now waste, so it has been frozen. **No model is involved in running this test.** The realization below is plain HTTP with no framework dependency. The assertions are imported from the originating use case rather than restated — a generated test that paraphrases its assertions creates a second, unverified statement of intent, and drift between the two would be silent. See F-0007 for what that costs. If this test starts failing, the correct first response is **not** to update the selectors. Re-run the agentic ancestor: if it recovers, the surface moved and this file should be regenerated; if it does not, the behaviour changed and that is a finding. """ from __future__ import annotations import urllib.error import urllib.parse import urllib.request from {claims_module} import {claim_imports} TARGET = {target!r} FIELDS = {fields!r} def _post(base_url: str, token: str, path: str, fields: dict) -> int: request = urllib.request.Request( urllib.parse.urljoin(base_url, path), data=urllib.parse.urlencode(fields).encode(), method="POST", headers={{ "Authorization": f"Bearer {{token}}", "Content-Type": "application/x-www-form-urlencoded", }}, ) try: with urllib.request.urlopen(request, timeout=10) as response: return response.status except urllib.error.HTTPError as error: return error.code def realize(base_url: str, token: str) -> int: """Perform {action_name} deterministically, exactly as the agent learned to.""" return _post(base_url, token, TARGET, FIELDS) def test_{test_name}(crystallized_world): """{action_name} still works, and the claims it protects still hold.""" base_url, token, observe = crystallized_world assert realize(base_url, token) < 400, "the frozen realization no longer works" snapshot = observe() {assertions} ''' def generate_test_module( *, trajectory: Trajectory, action_args: Mapping[str, Any], ancestor_id: str, ancestor_maturity: str, descendant_id: str, runs: int, sut_version: str, claims: Sequence[Any], claims_module: str, ) -> str: """Emit an ordinary pytest module for one crystallized semantic action.""" fields = {name: str(action_args[name]) for name in trajectory.fields if name in action_args} predicate_names = [c.predicate.__name__ for c in claims] assertions = "\n".join( f" assert {c.predicate.__name__}(snapshot), {c.text!r}" for c in claims ) return _TEMPLATE.format( ancestor_id=ancestor_id, ancestor_maturity=ancestor_maturity, descendant_id=descendant_id, runs=runs, sut_version=sut_version, generated_at=datetime.now(timezone.utc).date().isoformat(), claims_module=claims_module, claim_imports=", ".join(sorted(predicate_names)), target=trajectory.target, fields=fields, action_name=trajectory.action_name, test_name=trajectory.action_name, assertions=assertions, )