From 55118c1ad981c7d8b71d5c232ec59bb5c17bd657 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Fri, 24 Jul 2026 03:02:21 -0400 Subject: [PATCH 1/3] test: add hypothesis coverage for state json round trips --- tests/test_properties.py | 104 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) diff --git a/tests/test_properties.py b/tests/test_properties.py index 59939e0..b797cfc 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -1,3 +1,4 @@ +import json import re from unicodedata import normalize as unicode_normalize @@ -45,6 +46,59 @@ def _contains_canonical_start_fragment(value: str) -> bool: return False +def _sanitize_premise_like_engine(value: str) -> str: + sanitized = unicode_normalize("NFKC", value) + sanitized = sanitized.replace("’", "'").replace("`", "'") + return re.sub(r"\s+", " ", sanitized).strip() + + +NORMALIZATION_SENSITIVE_TEXT = st.text( + alphabet=st.characters( + blacklist_categories=("Cs",), + blacklist_characters="\x00", + ) + | st.sampled_from([" ", "\t", "’", "`"]), + min_size=1, + max_size=20, +) + +POLICY_VALUE = st.sampled_from(["use", "prohibit"]) + +VALID_STATE_PAYLOADS = st.builds( + lambda premise, pairs: { + "premise": premise, + "policies": dict(pairs), + "version": 2, + }, + premise=st.one_of(st.none(), NORMALIZATION_SENSITIVE_TEXT), + pairs=st.lists( + st.tuples(NORMALIZATION_SENSITIVE_TEXT, POLICY_VALUE), + min_size=0, + max_size=8, + ), +).filter(lambda payload: all(_normalize_item_like_engine(key) != "" for key in payload["policies"])) + + +def _canonicalize_state_payload(payload: dict[str, object]) -> State: + premise = payload["premise"] + assert premise is None or isinstance(premise, str) + + raw_policies = payload["policies"] + assert isinstance(raw_policies, dict) + + normalized_policies = { + _normalize_item_like_engine(key): value + for key, value in raw_policies.items() + if isinstance(key, str) + } + + return { + "premise": None if premise is None else _sanitize_premise_like_engine(premise), + "policies": dict(sorted(normalized_policies.items())), + "version": 2, + } + + @given(st.lists(st.text(max_size=40), min_size=0, max_size=20)) def test_determinism_same_input_sequence_same_state(inputs: list[str]) -> None: assert _run_sequence(inputs) == _run_sequence(inputs) @@ -177,3 +231,53 @@ def test_contradiction_prohibit_after_use_always_clarifies(item: str) -> None: decision = engine.step(f"prohibit {item}") assert decision["kind"] == "clarify" assert engine.state == before + + +@given(VALID_STATE_PAYLOADS) +def test_export_import_round_trip_preserves_authoritative_state_for_generated_payloads( + payload: dict[str, object], +) -> None: + source = create_engine() + source.import_json(json.dumps(payload)) + canonical_state = source.state + + target = create_engine() + target.import_json(source.export_json()) + + assert target.state == canonical_state + + +@given(VALID_STATE_PAYLOADS, st.integers(min_value=1, max_value=5)) +def test_repeated_export_import_cycles_remain_stable( + payload: dict[str, object], cycles: int +) -> None: + engine = create_engine() + engine.import_json(json.dumps(payload)) + + expected_state = engine.state + expected_json = engine.export_json() + + for _ in range(cycles): + next_engine = create_engine() + next_engine.import_json(expected_json) + assert next_engine.state == expected_state + assert next_engine.export_json() == expected_json + expected_state = next_engine.state + expected_json = next_engine.export_json() + + +@given(VALID_STATE_PAYLOADS) +def test_generated_valid_states_canonicalize_deterministically( + payload: dict[str, object], +) -> None: + expected_state = _canonicalize_state_payload(payload) + + engine1 = create_engine() + engine1.import_json(json.dumps(payload)) + + engine2 = create_engine() + engine2.import_json(json.dumps(payload)) + + assert engine1.state == expected_state + assert engine2.state == expected_state + assert engine1.export_json() == engine2.export_json() From 419dfeec0154594c5e313bbbe1378cf110b62e57 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Fri, 24 Jul 2026 03:13:04 -0400 Subject: [PATCH 2/3] test: add hypothesis core invariants --- tests/test_properties.py | 145 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 145 insertions(+) diff --git a/tests/test_properties.py b/tests/test_properties.py index b797cfc..a9dffc3 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -6,6 +6,7 @@ from hypothesis import strategies as st from context_compiler import create_engine +from context_compiler.controller import preview, state_diff from context_compiler.engine import _CANONICAL_DIRECTIVE_STARTS, State from context_compiler.grammar import validate_directive @@ -99,6 +100,70 @@ def _canonicalize_state_payload(payload: dict[str, object]) -> State: } +VALID_NONEMPTY_ITEM_TEXT = NORMALIZATION_SENSITIVE_TEXT.filter( + lambda value: ( + _normalize_item_like_engine(value) != "" + and not _contains_canonical_start_fragment(value) + and " instead of " not in value + and not value.startswith("instead of ") + and not value.endswith(" instead of") + ) +) + +VALID_USE_ITEM_TEXT = VALID_NONEMPTY_ITEM_TEXT.filter( + lambda value: validate_directive(f"use {value}") is not None +) + +VALID_PROHIBIT_ITEM_TEXT = VALID_NONEMPTY_ITEM_TEXT.filter( + lambda value: validate_directive(f"prohibit {value}") is not None +) + +VALID_PREMISE_TEXT = NORMALIZATION_SENSITIVE_TEXT.filter( + lambda value: _sanitize_premise_like_engine(value) != "" +) + +PREVIEW_INPUTS = st.one_of( + st.text(max_size=40), + VALID_USE_ITEM_TEXT.map(lambda item: f"use {item}"), + VALID_PROHIBIT_ITEM_TEXT.map(lambda item: f"prohibit {item}"), + VALID_NONEMPTY_ITEM_TEXT.map(lambda item: f"remove policy {item}").filter( + lambda text: validate_directive(text) is not None + ), + VALID_PREMISE_TEXT.map(lambda value: f"set premise {value}").filter( + lambda text: validate_directive(text) is not None + ), + VALID_PREMISE_TEXT.map(lambda value: f"change premise to {value}").filter( + lambda text: validate_directive(text) is not None + ), + st.sampled_from(["clear premise", "reset policies", "clear state"]), + st.tuples(VALID_USE_ITEM_TEXT, VALID_NONEMPTY_ITEM_TEXT) + .filter( + lambda pair: _normalize_item_like_engine(pair[0]) != _normalize_item_like_engine(pair[1]) + ) + .map(lambda pair: f"use {pair[0]} instead of {pair[1]}") + .filter(lambda text: validate_directive(text) is not None), +) + +DETERMINISTIC_REPLACEMENT_CASES = st.builds( + lambda payload, new_item, old_item, old_present: { + "payload": payload, + "new_item": new_item, + "old_item": old_item, + "old_present": old_present, + }, + payload=VALID_STATE_PAYLOADS, + new_item=VALID_USE_ITEM_TEXT, + old_item=VALID_NONEMPTY_ITEM_TEXT, + old_present=st.booleans(), +).filter( + lambda case: ( + _normalize_item_like_engine(case["new_item"]) + != _normalize_item_like_engine(case["old_item"]) + and validate_directive(f"use {case['new_item']} instead of {case['old_item']}") is not None + ) +) + + @given(st.lists(st.text(max_size=40), min_size=0, max_size=20)) def test_determinism_same_input_sequence_same_state(inputs: list[str]) -> None: assert _run_sequence(inputs) == _run_sequence(inputs) @@ -281,3 +346,83 @@ def test_generated_valid_states_canonicalize_deterministically( assert engine1.state == expected_state assert engine2.state == expected_state assert engine1.export_json() == engine2.export_json() + + +@given(DETERMINISTIC_REPLACEMENT_CASES) +def test_deterministic_replacement_matches_equivalent_explicit_transition( + case: dict[str, object], +) -> None: + payload = case["payload"] + new_item = case["new_item"] + old_item = case["old_item"] + old_present = case["old_present"] + + assert isinstance(payload, dict) + assert isinstance(new_item, str) + assert isinstance(old_item, str) + assert isinstance(old_present, bool) + + initial_state = _canonicalize_state_payload(payload) + new_key = _normalize_item_like_engine(new_item) + old_key = _normalize_item_like_engine(old_item) + + policies = dict(initial_state["policies"]) + policies.pop(new_key, None) + if old_present: + policies[old_key] = "use" + else: + policies.pop(old_key, None) + + initial_state = { + "premise": initial_state["premise"], + "policies": dict(sorted(policies.items())), + "version": 2, + } + assume(initial_state["policies"].get(new_key) != "prohibit") + + expected_state = { + "premise": initial_state["premise"], + "policies": dict(sorted({**initial_state["policies"], new_key: "use"}.items())), + "version": 2, + } + expected_state["policies"].pop(old_key, None) + expected_state["policies"][new_key] = "use" + expected_state["policies"] = dict(sorted(expected_state["policies"].items())) + + engine = create_engine(state=initial_state) + decision = engine.step(f"use {new_item} instead of {old_item}") + + assert decision == {"kind": "update", "state": expected_state, "prompt_to_user": None} + assert engine.state == expected_state + assert engine.has_pending_clarification() is False + + if not old_present: + followup = engine.step("yes") + assert followup == {"kind": "passthrough", "state": None, "prompt_to_user": None} + assert engine.has_pending_clarification() is False + assert engine.state == expected_state + + +@given(VALID_STATE_PAYLOADS, PREVIEW_INPUTS) +def test_preview_matches_isolated_execution_without_mutating_live_engine( + payload: dict[str, object], user_input: str +) -> None: + live_engine = create_engine() + live_engine.import_json(json.dumps(payload)) + before = live_engine.state + + preview_result = preview(live_engine, user_input) + + isolated_engine = create_engine() + isolated_engine.import_json(json.dumps(before)) + isolated_decision = isolated_engine.step(user_input) + isolated_after = isolated_engine.state + isolated_diff = state_diff(before, isolated_after) + + assert live_engine.state == before + assert preview_result["decision"] == isolated_decision + assert preview_result["state_before"] == before + assert preview_result["state_after"] == isolated_after + assert preview_result["diff"] == isolated_diff + assert preview_result["would_mutate"] is (before != isolated_after) + assert preview_result["would_mutate"] is preview_result["diff"]["changed"] From aa9ab013c0710523d9737f3da43ffcd68404b162 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Fri, 24 Jul 2026 03:27:20 -0400 Subject: [PATCH 3/3] test: add grammar helper property --- tests/test_properties.py | 153 ++++++++++++++++++++++++++------------- 1 file changed, 103 insertions(+), 50 deletions(-) diff --git a/tests/test_properties.py b/tests/test_properties.py index a9dffc3..6d425de 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -1,5 +1,6 @@ import json import re +from copy import deepcopy from unicodedata import normalize as unicode_normalize from hypothesis import assume, given @@ -8,7 +9,7 @@ from context_compiler import create_engine from context_compiler.controller import preview, state_diff from context_compiler.engine import _CANONICAL_DIRECTIVE_STARTS, State -from context_compiler.grammar import validate_directive +from context_compiler.grammar import DirectiveKind, render_directive, validate_directive def _run_sequence(inputs: list[str]) -> State: @@ -80,26 +81,6 @@ def _sanitize_premise_like_engine(value: str) -> str: ).filter(lambda payload: all(_normalize_item_like_engine(key) != "" for key in payload["policies"])) -def _canonicalize_state_payload(payload: dict[str, object]) -> State: - premise = payload["premise"] - assert premise is None or isinstance(premise, str) - - raw_policies = payload["policies"] - assert isinstance(raw_policies, dict) - - normalized_policies = { - _normalize_item_like_engine(key): value - for key, value in raw_policies.items() - if isinstance(key, str) - } - - return { - "premise": None if premise is None else _sanitize_premise_like_engine(premise), - "policies": dict(sorted(normalized_policies.items())), - "version": 2, - } - - VALID_NONEMPTY_ITEM_TEXT = NORMALIZATION_SENSITIVE_TEXT.filter( lambda value: ( _normalize_item_like_engine(value) != "" @@ -122,6 +103,25 @@ def _canonicalize_state_payload(payload: dict[str, object]) -> State: lambda value: _sanitize_premise_like_engine(value) != "" ) +CANONICAL_GRAMMAR_PREMISE_TEXT = NORMALIZATION_SENSITIVE_TEXT.map( + _sanitize_premise_like_engine +).filter( + lambda value: ( + value != "" + and validate_directive(f"set premise {value}") is not None + and validate_directive(f"change premise to {value}") is not None + ) +) + +CANONICAL_GRAMMAR_ITEM_TEXT = NORMALIZATION_SENSITIVE_TEXT.map(_normalize_item_like_engine).filter( + lambda value: ( + value != "" + and validate_directive(f"use {value}") is not None + and validate_directive(f"prohibit {value}") is not None + and validate_directive(f"remove policy {value}") is not None + ) +) + PREVIEW_INPUTS = st.one_of( st.text(max_size=40), VALID_USE_ITEM_TEXT.map(lambda item: f"use {item}"), @@ -164,11 +164,81 @@ def _canonicalize_state_payload(payload: dict[str, object]) -> State: ) +def _payload_has_stable_export_import_cycle(payload: dict[str, object]) -> bool: + engine = create_engine() + engine.import_json(json.dumps(payload)) + exported = engine.export_json() + + restored = create_engine() + try: + restored.import_json(exported) + except ValueError: + return False + + return restored.state == engine.state + + +GRAMMAR_RENDER_CASES = st.one_of( + CANONICAL_GRAMMAR_PREMISE_TEXT.map( + lambda value: {"kind": DirectiveKind.SET_PREMISE, "operands": {"value": value}} + ), + CANONICAL_GRAMMAR_PREMISE_TEXT.map( + lambda value: {"kind": DirectiveKind.CHANGE_PREMISE, "operands": {"value": value}} + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: {"kind": DirectiveKind.USE_ITEM, "operands": {"item": item}} + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: {"kind": DirectiveKind.PROHIBIT_ITEM, "operands": {"item": item}} + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: {"kind": DirectiveKind.REMOVE_POLICY, "operands": {"item": item}} + ), + st.tuples(CANONICAL_GRAMMAR_ITEM_TEXT, CANONICAL_GRAMMAR_ITEM_TEXT) + .filter( + lambda pair: _normalize_item_like_engine(pair[0]) != _normalize_item_like_engine(pair[1]) + ) + .map( + lambda pair: { + "kind": DirectiveKind.REPLACE_USE, + "operands": {"new_item": pair[0], "old_item": pair[1]}, + } + ), + st.sampled_from( + [ + {"kind": DirectiveKind.CLEAR_PREMISE, "operands": {}}, + {"kind": DirectiveKind.RESET_POLICIES, "operands": {}}, + {"kind": DirectiveKind.CLEAR_STATE, "operands": {}}, + ] + ), +) + + @given(st.lists(st.text(max_size=40), min_size=0, max_size=20)) def test_determinism_same_input_sequence_same_state(inputs: list[str]) -> None: assert _run_sequence(inputs) == _run_sequence(inputs) +@given(GRAMMAR_RENDER_CASES) +def test_grammar_helper_render_validate_round_trip_is_stable( + case: dict[str, DirectiveKind | dict[str, str]], +) -> None: + kind = case["kind"] + operands = case["operands"] + + assert isinstance(kind, DirectiveKind) + assert isinstance(operands, dict) + + rendered = render_directive(kind, **operands) + validated = validate_directive(rendered) + + assert validated is not None + assert validated.kind is kind + assert validated.text == rendered + assert validate_directive(validated.text) == validated + assert render_directive(kind, **operands) == rendered + + @given(st.text(min_size=1, max_size=30)) def test_idempotent_use_item_is_update_and_stable_state(item: str) -> None: assume(" instead of " not in item) @@ -331,23 +401,6 @@ def test_repeated_export_import_cycles_remain_stable( expected_json = next_engine.export_json() -@given(VALID_STATE_PAYLOADS) -def test_generated_valid_states_canonicalize_deterministically( - payload: dict[str, object], -) -> None: - expected_state = _canonicalize_state_payload(payload) - - engine1 = create_engine() - engine1.import_json(json.dumps(payload)) - - engine2 = create_engine() - engine2.import_json(json.dumps(payload)) - - assert engine1.state == expected_state - assert engine2.state == expected_state - assert engine1.export_json() == engine2.export_json() - - @given(DETERMINISTIC_REPLACEMENT_CASES) def test_deterministic_replacement_matches_equivalent_explicit_transition( case: dict[str, object], @@ -362,7 +415,9 @@ def test_deterministic_replacement_matches_equivalent_explicit_transition( assert isinstance(old_item, str) assert isinstance(old_present, bool) - initial_state = _canonicalize_state_payload(payload) + initial_state_engine = create_engine() + initial_state_engine.import_json(json.dumps(payload)) + initial_state = initial_state_engine.state new_key = _normalize_item_like_engine(new_item) old_key = _normalize_item_like_engine(old_item) @@ -379,20 +434,18 @@ def test_deterministic_replacement_matches_equivalent_explicit_transition( "version": 2, } assume(initial_state["policies"].get(new_key) != "prohibit") + assume(initial_state["policies"].get(old_key) != "prohibit") - expected_state = { - "premise": initial_state["premise"], - "policies": dict(sorted({**initial_state["policies"], new_key: "use"}.items())), - "version": 2, - } - expected_state["policies"].pop(old_key, None) - expected_state["policies"][new_key] = "use" - expected_state["policies"] = dict(sorted(expected_state["policies"].items())) + oracle_engine = create_engine(state=deepcopy(initial_state)) + oracle_engine.step(f"remove policy {old_item}") + expected_decision = oracle_engine.step(f"use {new_item}") + expected_state = oracle_engine.state engine = create_engine(state=initial_state) decision = engine.step(f"use {new_item} instead of {old_item}") - assert decision == {"kind": "update", "state": expected_state, "prompt_to_user": None} + assert expected_decision == {"kind": "update", "state": expected_state, "prompt_to_user": None} + assert decision == expected_decision assert engine.state == expected_state assert engine.has_pending_clarification() is False @@ -403,13 +456,13 @@ def test_deterministic_replacement_matches_equivalent_explicit_transition( assert engine.state == expected_state -@given(VALID_STATE_PAYLOADS, PREVIEW_INPUTS) +@given(VALID_STATE_PAYLOADS.filter(_payload_has_stable_export_import_cycle), PREVIEW_INPUTS) def test_preview_matches_isolated_execution_without_mutating_live_engine( payload: dict[str, object], user_input: str ) -> None: live_engine = create_engine() live_engine.import_json(json.dumps(payload)) - before = live_engine.state + before = deepcopy(live_engine.state) preview_result = preview(live_engine, user_input)