From 309ff3ba9b7f6ce30e885dacf6287dca2697bc9b Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Fri, 14 Aug 2026 00:24:08 -0400 Subject: [PATCH 1/2] fix: require active source for replace use --- README.md | 11 +- demos/08_llm_replacement_precondition.py | 48 +-- demos/09_llm_confirmation_no_directive.py | 51 ++- docs/DirectiveGrammarSpec.md | 25 +- docs/api-reference.md | 5 +- docs/architecture.md | 10 +- src/context_compiler/engine.py | 2 +- ...ollowup_no_directive_normalized_token.json | 4 +- ...llowup_no_directive_punctuation_token.json | 4 +- ...wup_no_directive_after_replace_update.json | 4 +- ...ollowup_no_directive_normalized_token.json | 4 +- ...llowup_no_directive_punctuation_token.json | 4 +- ...p_replace_missing_source_error_prompt.json | 8 +- .../expected/replacement_error.json | 8 +- .../scenarios/replacement_error.json | 2 +- tests/test_04_grammar_edge_cases.py | 7 +- tests/test_engine.py | 81 ++-- tests/test_properties.py | 358 +++++++++++++++++- 18 files changed, 488 insertions(+), 148 deletions(-) diff --git a/README.md b/README.md index 71ba701..55d236e 100644 --- a/README.md +++ b/README.md @@ -392,10 +392,9 @@ Replacement: User: use podman instead of docker ``` -If `docker` is absent from saved state, that does not make the directive -pending. The user's intended resulting state is still unambiguous, so the -replacement follows the deterministic `use podman` transition when otherwise -semantically valid. +If `docker` is absent from saved state, that is a semantic `error`. +Canonical replacement requires an active existing source `use` policy and +does not degrade to plain `use podman`. Removal and reset: @@ -413,8 +412,8 @@ evaluation against authoritative state. Pending continuation is a separate runtime layer. It may exist only after a canonical directive reaches a supported semantic `error` case. It never repairs malformed syntax or reinterprets non-canonical input as a directive. -An absent source item in a canonical replacement directive is not, by itself, -such a `error` case. +An absent source item in a canonical replacement directive is itself a +semantic `error` case and does not authorize degradation to plain `use`. Examples: diff --git a/demos/08_llm_replacement_precondition.py b/demos/08_llm_replacement_precondition.py index 2b60cef..2a9aa88 100644 --- a/demos/08_llm_replacement_precondition.py +++ b/demos/08_llm_replacement_precondition.py @@ -1,10 +1,10 @@ -"""Demo 8: missing-source replacement applies deterministically from authoritative state.""" +"""Demo 8: missing-source replacement fails without mutating authoritative state.""" from collections.abc import Mapping from context_compiler import ( Engine, - is_update, + is_error, ) from demos.common import ( build_baseline_messages, @@ -22,7 +22,7 @@ ) from demos.llm_client import complete_messages -DEMO_NAME = "08_replacement_precondition — missing-source replacement applies deterministically" +DEMO_NAME = "08_replacement_precondition — missing-source replacement requires active source" USER_INPUT = "use podman instead of docker" @@ -69,38 +69,38 @@ def main() -> None: reinjected_output = complete_messages(reinjected_messages) print_model_output("Reinjected-state", reinjected_output) - if is_update(decision): + if is_error(decision): print_messages("compiler-mediated (full)", []) - mediated_output = "[no call] authoritative state applied deterministic replacement update" + mediated_output = "[no call] authoritative state blocked replacement without source use" print_model_output("Compiler-mediated (full)", mediated_output) else: print_messages("compiler-mediated (full)", []) - mediated_output = "[no call] expected update was not produced" + mediated_output = "[no call] expected semantic replacement error was not produced" print_model_output("Compiler-mediated (full)", mediated_output) compacted_turns, compacted_state, compacted_prompt = compact_user_turns(user_inputs) - if compacted_prompt is None: + if compacted_prompt is not None: print_messages("compiler-mediated + compact", []) - compact_output = "[no call] compaction preserved deterministic state update" + compact_output = "[no call] unexpected error was produced during compaction" print_model_output("Compiler-mediated + compact", compact_output) else: print_messages("compiler-mediated + compact", []) - compact_output = "[no call] unexpected error was produced during compaction" + compact_output = "[no call] compaction preserved replacement error without state mutation" print_model_output("Compiler-mediated + compact", compact_output) premise, policies = observe_engine(engine) compacted_premise, compacted_policies = state_observations(compacted_state) - state_applied = not _is_initial_authoritative_state(premise=premise, policies=policies) - compact_state_applied = not _is_initial_authoritative_state( + state_preserved = _is_initial_authoritative_state(premise=premise, policies=policies) + compact_state_preserved = _is_initial_authoritative_state( premise=compacted_premise, policies=compacted_policies, ) - compact_no_pending = compacted_prompt is None + compact_error_preserved = compacted_prompt is not None baseline_has_authoritative_precondition = False reinjected_has_authoritative_precondition = False - compiler_pass = is_update(decision) and state_applied - compact_pass = compacted_prompt is None and compact_state_applied and compact_no_pending + compiler_pass = is_error(decision) and state_preserved + compact_pass = compact_error_preserved and compact_state_preserved print_host_check( "BASELINE_AUTHORITATIVE_PRECONDITION", @@ -114,12 +114,12 @@ def main() -> None: ) print_host_check( "COMPILER_BLOCKED_INVALID_REPLACEMENT", - yes_no(is_update(decision)), + yes_no(is_error(decision)), context="compiler-mediated", ) print_host_check( - "COMPILER_STATE_APPLIED", - yes_no(state_applied), + "COMPILER_STATE_PRESERVED", + yes_no(state_preserved), context="compiler-mediated", ) @@ -130,18 +130,18 @@ def main() -> None: compiler_pass=compiler_pass, compiler_compact_pass=compact_pass, expected=( - "missing-source replacement should deterministically apply the resulting use update " - "without pending continuation" + "missing-source replacement should return semantic error without mutating " + "authoritative state" ), actual=( - "compiler applied deterministic replacement update; baseline and reinjected paths " - "still lack authoritative state enforcement" + "compiler blocked missing-source replacement without mutating state; baseline and " + "reinjected paths still lack authoritative state enforcement" if compiler_pass and compact_pass - else "compiler did not consistently apply deterministic replacement behavior" + else "compiler did not consistently enforce the replacement source precondition" ), passed=compiler_pass and compact_pass, - result_pass="missing-source replacement applied deterministically", - result_fail="missing-source replacement did not apply deterministically", + result_pass="missing-source replacement was rejected without mutation", + result_fail="missing-source replacement was not rejected correctly", ) diff --git a/demos/09_llm_confirmation_no_directive.py b/demos/09_llm_confirmation_no_directive.py index f5eb745..09f6c43 100644 --- a/demos/09_llm_confirmation_no_directive.py +++ b/demos/09_llm_confirmation_no_directive.py @@ -1,11 +1,11 @@ -"""Demo 9: confirmation-style followups remain ordinary no_directive.""" +"""Demo 9: replacement errors do not create confirmation-style followup state.""" from collections.abc import Mapping from context_compiler import ( Engine, + is_error, is_no_directive, - is_update, ) from demos.common import ( build_baseline_messages, @@ -24,8 +24,7 @@ from demos.llm_client import complete_messages DEMO_NAME = ( - "09_confirmation_no_directive_boundary — " - "missing-source replacement does not create a confirmation state" + "09_confirmation_no_directive_boundary — replacement errors do not create a confirmation state" ) TURN_1 = "use podman instead of docker" TURN_2 = "maybe" @@ -34,10 +33,6 @@ INITIAL_POLICIES: dict[str, str] = {} -def _has_podman_use(policies: Mapping[str, str]) -> bool: - return policies.get("podman") == "use" - - def _is_initial_authoritative_state(*, premise: str | None, policies: Mapping[str, str]) -> bool: return premise == INITIAL_PREMISE and dict(policies) == INITIAL_POLICIES @@ -50,12 +45,17 @@ def main() -> None: first = engine.step(TURN_1) premise, policies = observe_engine(engine) print_decision("turn 1", first, premise=premise, policies=policies) - state_applied_after_first = _has_podman_use(policies) + state_preserved_after_first = _is_initial_authoritative_state( + premise=premise, policies=policies + ) second = engine.step(TURN_2) premise, policies = observe_engine(engine) print_decision("turn 2", second, premise=premise, policies=policies) - state_preserved_after_second = _has_podman_use(policies) + state_preserved_after_second = _is_initial_authoritative_state( + premise=premise, + policies=policies, + ) third = engine.step(TURN_3) premise, policies = observe_engine(engine) @@ -96,7 +96,7 @@ def main() -> None: compacted_turns, compacted_state, compacted_prompt = compact_user_turns(user_inputs) if compacted_prompt is not None: print_messages("compiler-mediated + compact", []) - compact_output = f"[no call] error required: {compacted_prompt}" + compact_output = f"[no call] replacement error preserved: {compacted_prompt}" print_model_output("Compiler-mediated + compact", compact_output) else: print_messages("compiler-mediated + compact", []) @@ -105,32 +105,30 @@ def main() -> None: ) print_model_output("Compiler-mediated + compact", compact_output) - deterministic_initial_update = is_update(first) and state_applied_after_first + deterministic_initial_error = is_error(first) and state_preserved_after_first unrelated_followup_no_directive = is_no_directive(second) and state_preserved_after_second confirmation_token_not_consumed = is_no_directive(third) premise, policies = observe_engine(engine) - deterministic_final_state = _has_podman_use(policies) + deterministic_final_state = _is_initial_authoritative_state(premise=premise, policies=policies) _, compacted_policies = state_observations(compacted_state) baseline_has_confirmation_state_machine = False reinjected_has_confirmation_state_machine = False compiler_pass = ( - deterministic_initial_update + deterministic_initial_error and unrelated_followup_no_directive and confirmation_token_not_consumed and deterministic_final_state ) compact_pass = ( - compacted_prompt is None - and compacted_turns == [TURN_2, TURN_3] - and _has_podman_use(compacted_policies) + compacted_prompt is not None and compacted_turns == [TURN_1] and compacted_policies == {} ) print_host_check( - "DETERMINISTIC_INITIAL_UPDATE", - yes_no(deterministic_initial_update), + "DETERMINISTIC_INITIAL_ERROR", + yes_no(deterministic_initial_error), context="compiler-mediated", ) print_host_check( @@ -144,7 +142,7 @@ def main() -> None: context="compiler-mediated", ) print_host_check( - "FINAL_POLICY_PODMAN_PRESENT", + "FINAL_STATE_UNCHANGED", yes_no(deterministic_final_state), context="compiler-mediated", ) @@ -156,19 +154,18 @@ def main() -> None: compiler_pass=compiler_pass, compiler_compact_pass=compact_pass, expected=( - "missing-source replacement should apply without creating an engine-owned " - "confirmation state, and later yes/no-style input should remain ordinary " - "no_directive" + "replacement error should not create an engine-owned confirmation state, " + "and later yes/no-style input should remain ordinary no_directive" ), actual=( - "compiler applied deterministic replacement update and treated later inputs as " - "ordinary no_directive" + "compiler returned semantic error and treated later inputs as ordinary " + "no_directive without mutating state" if compiler_pass and compact_pass else "compiler did not consistently preserve the confirmation-no_directive boundary" ), passed=compiler_pass and compact_pass, - result_pass="missing-source replacement stayed outside engine-owned confirmation state", - result_fail="missing-source replacement still behaved like engine-owned confirmation state", + result_pass="replacement error stayed outside engine-owned confirmation state", + result_fail="replacement error still behaved like engine-owned confirmation state", ) diff --git a/docs/DirectiveGrammarSpec.md b/docs/DirectiveGrammarSpec.md index 27a8788..8bad852 100644 --- a/docs/DirectiveGrammarSpec.md +++ b/docs/DirectiveGrammarSpec.md @@ -564,14 +564,12 @@ Let `kx` be the policy identity key for `REPLACE_NEW` under Section 10.1 and The replacement-specific `error` cases are state-dependent and belong to semantic evaluation, not parsing. -Normative classification for the historical missing-source case: - -- if `ky` is absent and applying `use ` is otherwise semantically valid, - `use instead of ` is not a error case; -- core applies the deterministic resulting transition of asserting - `REPLACE_NEW` as `use`; -- the user's incorrect assumption about the current presence of `` does - not by itself create semantic ambiguity or pending continuation; +Normative classification for the missing-source case: + +- if `ky` is absent, `use instead of ` is an `error` case; +- replacement requires an active existing `use` policy for `REPLACE_OLD`; +- core must not degrade the canonical replacement directive into plain + `use `; - malformed replacement syntax remains invalid grammar, and other semantic conflicts for a canonical replacement may still return `error`. @@ -633,11 +631,10 @@ This specification preserves the grammar hardening established after `0.8.x`: Within semantic evaluation, pending continuation is intended only for deterministic blocked transitions that do not expand authority beyond the -parsed canonical operation. In particular, the historical missing-source -replacement case (`use instead of ` when `` is absent) is not -a pending or error case under this specification; it deterministically -applies the resulting `use ` transition when otherwise semantically -valid. +parsed canonical operation. In particular, the missing-source replacement +case (`use instead of ` when `` is absent) is an `error` +case under this specification and must not be reinterpreted as plain +`use `. ## 11. Storage Normalization @@ -719,7 +716,7 @@ source material for later conformance fixtures. | `prohibit peanuts` | canonical directive | prohibit item | may apply, no-op, or error | | `remove policy docker` | canonical directive | remove policy | may apply or no-op | | `use podman instead of docker` | canonical directive | replace use | may apply, no-op, or error | -| `use podman instead of docker` when `docker` is absent and `use podman` is otherwise valid | canonical directive | replace use | applies deterministically as the resulting `use podman` transition; not a pending/error case | +| `use podman instead of docker` when `docker` is absent | canonical directive | replace use | semantic error; replacement requires an active source `use` policy | | `clear premise` | canonical directive | clear premise | may apply or no-op | | `reset policies` | canonical directive | reset policies | may apply or no-op | | `clear state` | canonical directive | clear state | may apply or no-op | diff --git a/docs/api-reference.md b/docs/api-reference.md index c866f00..c3720f2 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -114,9 +114,8 @@ Boundary notes: code should send canonical directives when it wants deterministic mutation - failed replacement requests are not reinterpreted by core into different directives -- `use instead of ` with an absent `` is not a pending or - error-only runtime category; it follows the deterministic semantic - rules defined in the specification +- `use instead of ` with an absent `` is a semantic `error`; + core does not degrade it into plain `use ` `CanonicalDirective.operands` preserves the grammar-recognized operand text. Core does not lowercase operands, collapse internal operand whitespace, or diff --git a/docs/architecture.md b/docs/architecture.md index 68722f0..1b8710d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,17 +37,15 @@ Boundary: canonical operations already established by core - pending continuation is runtime state, not grammar - malformed or non-canonical input must never create pending continuation -- a missing source item in `use instead of ` is evaluated as a - deterministic state transition question, not as a justification for pending - continuation +- a missing source item in `use instead of ` is a semantic + replacement error, not a justification for pending continuation Current repository note: - the intended contract allows semantic pending continuation for supported deterministic blocked transitions -- that continuation boundary is independent from the historical replacement - case where the requested old item is absent from state -- the current runtime implementation does not yet fully restore that contract +- that continuation boundary remains independent from replacement errors where + the requested old item is absent from state and should be treated as lagging the updated specification until runtime work lands diff --git a/src/context_compiler/engine.py b/src/context_compiler/engine.py index d8ccf97..fad19c8 100644 --- a/src/context_compiler/engine.py +++ b/src/context_compiler/engine.py @@ -176,7 +176,7 @@ def _pre_mutation_error( f'"{new_item}" is currently prohibited.\n' "Submit explicit directive(s) to remove it or use a different item." ) - if old_state not in {None, POLICY_USE}: + if old_state != POLICY_USE: return _error( f'"{old_item}" is not currently in use.\n' "Replacement requires an active 'use' policy." diff --git a/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_normalized_token.json b/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_normalized_token.json index b93745f..2a471c6 100644 --- a/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_normalized_token.json +++ b/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_normalized_token.json @@ -17,9 +17,7 @@ }, "state": { "premise": null, - "policies": { - "kubectl": "use" - }, + "policies": {}, "version": 2 } } diff --git a/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_punctuation_token.json b/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_punctuation_token.json index 7dbe818..4d1237d 100644 --- a/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_punctuation_token.json +++ b/tests/fixtures/conformance/step/step_affirmative_followup_no_directive_punctuation_token.json @@ -17,9 +17,7 @@ }, "state": { "premise": null, - "policies": { - "kubectl": "use" - }, + "policies": {}, "version": 2 } } diff --git a/tests/fixtures/conformance/step/step_independent_followup_no_directive_after_replace_update.json b/tests/fixtures/conformance/step/step_independent_followup_no_directive_after_replace_update.json index cf915c8..34a0bec 100644 --- a/tests/fixtures/conformance/step/step_independent_followup_no_directive_after_replace_update.json +++ b/tests/fixtures/conformance/step/step_independent_followup_no_directive_after_replace_update.json @@ -17,9 +17,7 @@ }, "state": { "premise": null, - "policies": { - "kubectl": "use" - }, + "policies": {}, "version": 2 } } diff --git a/tests/fixtures/conformance/step/step_negative_followup_no_directive_normalized_token.json b/tests/fixtures/conformance/step/step_negative_followup_no_directive_normalized_token.json index edc15ce..e87c3cd 100644 --- a/tests/fixtures/conformance/step/step_negative_followup_no_directive_normalized_token.json +++ b/tests/fixtures/conformance/step/step_negative_followup_no_directive_normalized_token.json @@ -17,9 +17,7 @@ }, "state": { "premise": null, - "policies": { - "kubectl": "use" - }, + "policies": {}, "version": 2 } } diff --git a/tests/fixtures/conformance/step/step_negative_followup_no_directive_punctuation_token.json b/tests/fixtures/conformance/step/step_negative_followup_no_directive_punctuation_token.json index df5ebea..e1d8ccb 100644 --- a/tests/fixtures/conformance/step/step_negative_followup_no_directive_punctuation_token.json +++ b/tests/fixtures/conformance/step/step_negative_followup_no_directive_punctuation_token.json @@ -17,9 +17,7 @@ }, "state": { "premise": null, - "policies": { - "kubectl": "use" - }, + "policies": {}, "version": 2 } } diff --git a/tests/fixtures/conformance/step/step_replace_missing_source_error_prompt.json b/tests/fixtures/conformance/step/step_replace_missing_source_error_prompt.json index 7b326cd..cfcd537 100644 --- a/tests/fixtures/conformance/step/step_replace_missing_source_error_prompt.json +++ b/tests/fixtures/conformance/step/step_replace_missing_source_error_prompt.json @@ -9,14 +9,12 @@ "input": "use kubectl instead of docker", "expected": { "decision": { - "kind": "update", - "message": null + "kind": "error", + "message": "\"docker\" is not currently in use.\nReplacement requires an active 'use' policy." }, "state": { "premise": null, - "policies": { - "kubectl": "use" - }, + "policies": {}, "version": 2 } } diff --git a/tests/fixtures/engine-regression/structured/expected/replacement_error.json b/tests/fixtures/engine-regression/structured/expected/replacement_error.json index aa9ea0c..445c133 100644 --- a/tests/fixtures/engine-regression/structured/expected/replacement_error.json +++ b/tests/fixtures/engine-regression/structured/expected/replacement_error.json @@ -3,15 +3,13 @@ "turns": [ { "state": { - "policies": { - "podman": "use" - }, + "policies": {}, "premise": null, "version": 2 }, "decision": { - "kind": "update", - "message": null + "kind": "error", + "message": "\"docker\" is not currently in use.\nReplacement requires an active 'use' policy." }, "input": "use podman instead of docker" } diff --git a/tests/fixtures/engine-regression/structured/scenarios/replacement_error.json b/tests/fixtures/engine-regression/structured/scenarios/replacement_error.json index 3704dab..076e1e5 100644 --- a/tests/fixtures/engine-regression/structured/scenarios/replacement_error.json +++ b/tests/fixtures/engine-regression/structured/scenarios/replacement_error.json @@ -1,5 +1,5 @@ { - "description": "Replacement from missing old item applies as deterministic use update", + "description": "Replacement from missing old item returns semantic error without mutation", "id": "replacement_error", "initial_state": null, "inputs": [ diff --git a/tests/test_04_grammar_edge_cases.py b/tests/test_04_grammar_edge_cases.py index ef63c19..246786d 100644 --- a/tests/test_04_grammar_edge_cases.py +++ b/tests/test_04_grammar_edge_cases.py @@ -1,4 +1,5 @@ from context_compiler import ( + DECISION_ERROR, DECISION_NO_DIRECTIVE, DECISION_UPDATE, Engine, @@ -107,14 +108,14 @@ def test_remove_policy_missing_or_whitespace_payload_remains_no_directive() -> N def test_invalid_replacement_does_not_block_following_directives() -> None: engine = Engine() first = engine.step("use kubectl instead of docker") - assert first["kind"] == DECISION_UPDATE + assert first["kind"] == DECISION_ERROR second = engine.step("set premise concise") assert second == { "kind": DECISION_UPDATE, "message": None, } - assert _observations(engine) == ("concise", {"kubectl": "use"}) + assert _observations(engine) == ("concise", {}) def test_replace_update_independent_followup_is_no_directive() -> None: @@ -122,5 +123,5 @@ def test_replace_update_independent_followup_is_no_directive() -> None: first = engine.step("use kubectl instead of docker") second = engine.step("sounds good") - assert first["kind"] == DECISION_UPDATE + assert first["kind"] == DECISION_ERROR assert second == {"kind": DECISION_NO_DIRECTIVE, "message": None} diff --git a/tests/test_engine.py b/tests/test_engine.py index 4a7fbcc..5cf30d6 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -882,15 +882,17 @@ def test_replace_use_identity_case_variant_is_noop_update() -> None: _assert_observations(engine, premise=None, policies={"docker": "use"}) -def test_replace_use_missing_source_applies_as_use_update() -> None: +def test_replace_use_missing_source_returns_error_without_mutation() -> None: engine = Engine() d1 = engine.step("use kubectl instead of docker") assert d1 == { - "kind": "update", - "message": None, + "kind": "error", + "message": ( + "\"docker\" is not currently in use.\nReplacement requires an active 'use' policy." + ), } - _assert_observations(engine, premise=None, policies={"kubectl": "use"}) + _assert_observations(engine, premise=None, policies={}) def test_replace_use_missing_source_yes_followup_is_no_directive() -> None: @@ -898,14 +900,16 @@ def test_replace_use_missing_source_yes_followup_is_no_directive() -> None: first = engine.step("use kubectl instead of docker") assert first == { - "kind": "update", - "message": None, + "kind": "error", + "message": ( + "\"docker\" is not currently in use.\nReplacement requires an active 'use' policy." + ), } - _assert_observations(engine, premise=None, policies={"kubectl": "use"}) + _assert_observations(engine, premise=None, policies={}) second = engine.step("yes") assert second == {"kind": DECISION_NO_DIRECTIVE, "message": None} - _assert_observations(engine, premise=None, policies={"kubectl": "use"}) + _assert_observations(engine, premise=None, policies={}) def test_replace_use_missing_source_no_followup_has_no_mutation() -> None: @@ -934,39 +938,50 @@ def test_replace_use_missing_source_still_reports_target_prohibit_when_new_item_ } -def test_replace_use_missing_source_ignores_unrelated_existing_policies() -> None: +def test_replace_use_missing_source_preserves_unrelated_existing_policies() -> None: engine = Engine() engine.step("use python and docker") decision = engine.step("use kubectl instead of python") - assert decision["kind"] == DECISION_UPDATE - assert dict(engine.policies) == {"kubectl": "use", "python and docker": "use"} + assert decision == { + "kind": "error", + "message": ( + "\"python\" is not currently in use.\nReplacement requires an active 'use' policy." + ), + } + assert dict(engine.policies) == {"python and docker": "use"} -def test_replace_use_missing_source_ignores_other_conflicting_entries() -> None: +def test_replace_use_missing_source_preserves_other_conflicting_entries() -> None: engine = Engine() engine.step("use python and docker") engine.step("prohibit python tooling") decision = engine.step("use kubectl instead of python") - assert decision["kind"] == DECISION_UPDATE - assert dict(engine.policies) == { - "kubectl": "use", - "python and docker": "use", - "python tooling": "prohibit", + assert decision == { + "kind": "error", + "message": ( + "\"python\" is not currently in use.\nReplacement requires an active 'use' policy." + ), } + assert dict(engine.policies) == {"python and docker": "use", "python tooling": "prohibit"} -def test_replace_use_missing_source_with_empty_probe_uses_invalid_prompt() -> None: +def test_replace_use_missing_source_with_empty_probe_returns_error() -> None: engine = Engine() engine.step("use python and docker") decision = engine.step("use kubectl instead of the") - assert decision["kind"] == DECISION_UPDATE + assert decision == { + "kind": "error", + "message": ( + "\"the\" is not currently in use.\nReplacement requires an active 'use' policy." + ), + } _assert_observations( engine, premise=None, - policies={"kubectl": "use", "python and docker": "use"}, + policies={"python and docker": "use"}, ) @@ -1067,21 +1082,21 @@ def test_replace_use_kx_prohibit_no_followup_has_no_mutation() -> None: def test_missing_source_replacement_does_not_block_following_directives() -> None: engine = Engine() first = engine.step("use kubectl instead of docker") - assert first["kind"] == "update" + assert first["kind"] == "error" second = engine.step("use docker") assert second["kind"] == "update" - assert dict(engine.policies) == {"docker": "use", "kubectl": "use"} + assert dict(engine.policies) == {"docker": "use"} third = engine.step("yes") assert third == {"kind": DECISION_NO_DIRECTIVE, "message": None} - assert dict(engine.policies) == {"docker": "use", "kubectl": "use"} + assert dict(engine.policies) == {"docker": "use"} def test_missing_source_replacement_does_not_suspend_admin_commands() -> None: engine = Engine() engine.step("use kubectl instead of docker") - before = (None, {"kubectl": "use"}) + before = (None, {}) assert _observations(engine) == before @@ -1101,7 +1116,7 @@ def test_missing_source_replacement_negative_followup_is_no_directive() -> None: decision = engine.step("no") assert decision == {"kind": DECISION_NO_DIRECTIVE, "message": None} - assert dict(engine.policies) == {"kubectl": "use"} + assert dict(engine.policies) == {} def test_missing_source_replacement_affirmative_followup_tokens_are_no_directive() -> None: @@ -1110,7 +1125,7 @@ def test_missing_source_replacement_affirmative_followup_tokens_are_no_directive decision = engine.step(" YES!!! ") assert decision["kind"] == DECISION_NO_DIRECTIVE - assert dict(engine.policies) == {"kubectl": "use"} + assert dict(engine.policies) == {} def test_missing_source_replacement_affirmative_token_variants_are_no_directive() -> None: @@ -1119,7 +1134,7 @@ def test_missing_source_replacement_affirmative_token_variants_are_no_directive( engine.step("use kubectl instead of docker") decision = engine.step(token) assert decision["kind"] == DECISION_NO_DIRECTIVE - assert dict(engine.policies) == {"kubectl": "use"} + assert dict(engine.policies) == {} def test_missing_source_replacement_negative_tokens_are_no_directive() -> None: @@ -1189,7 +1204,7 @@ def test_prohibited_replacement_yes_cannot_override_conflicting_target_polarity( def test_import_json_does_not_change_independent_yes_no_followup_behavior() -> None: engine = Engine() first = engine.step("use kubectl instead of docker") - assert first["kind"] == DECISION_UPDATE + assert first["kind"] == DECISION_ERROR imported = {"premise": "baseline", "policies": {"pytest": "use"}, "version": 2} engine.import_json(json.dumps(imported)) @@ -1541,15 +1556,17 @@ def test_all_canonical_directive_starts_remain_single_directive_when_valid( assert decision["kind"] != DECISION_NO_DIRECTIVE -def test_compound_no_directive_after_prior_missing_source_replacement_update() -> None: +def test_compound_no_directive_after_prior_missing_source_replacement_error() -> None: engine = Engine() first = engine.step("use kubectl instead of docker") assert first == { - "kind": DECISION_UPDATE, - "message": None, + "kind": DECISION_ERROR, + "message": ( + "\"docker\" is not currently in use.\nReplacement requires an active 'use' policy." + ), } decision = engine.step("use docker and prohibit peanuts") assert decision == {"kind": DECISION_NO_DIRECTIVE, "message": None} - _assert_observations(engine, premise=None, policies={"kubectl": "use"}) + _assert_observations(engine, premise=None, policies={}) diff --git a/tests/test_properties.py b/tests/test_properties.py index c3cd063..c894856 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -16,7 +16,10 @@ from context_compiler.grammar import ( CanonicalDirective, DirectiveKind, + DirectiveSyntaxFailure, + InvalidDirectiveSyntax, decompose_directive, + get_directive_metadata, ) @@ -61,6 +64,12 @@ def _sanitize_premise_like_engine(value: str) -> str: return re.sub(r"\s+", " ", sanitized).strip() +def _canonical_directive_from_text(text: str) -> CanonicalDirective: + directive = decompose_directive(text) + assert isinstance(directive, CanonicalDirective) + return directive + + NORMALIZATION_SENSITIVE_TEXT = st.text( alphabet=st.characters( blacklist_categories=("Cs",), @@ -183,9 +192,107 @@ def _build_deterministic_replacement_case( ) ) ) + .filter(lambda args: args[4]) .map(lambda args: _build_deterministic_replacement_case(*args)) ) +ERROR_CASES = st.one_of( + CANONICAL_GRAMMAR_PREMISE_TEXT.map( + lambda value: ( + {"premise": "existing", "policies": {}, "version": 2}, + _canonical_directive_from_text(f"set premise {value}"), + ) + ), + CANONICAL_GRAMMAR_PREMISE_TEXT.map( + lambda value: ( + {"premise": None, "policies": {}, "version": 2}, + _canonical_directive_from_text(f"change premise to {value}"), + ) + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: ( + { + "premise": None, + "policies": {_normalize_item_like_engine(item): "prohibit"}, + "version": 2, + }, + _canonical_directive_from_text(f"use {item}"), + ) + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: ( + {"premise": None, "policies": {_normalize_item_like_engine(item): "use"}, "version": 2}, + _canonical_directive_from_text(f"prohibit {item}"), + ) + ), + 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: ( + {"premise": None, "policies": {}, "version": 2}, + _canonical_directive_from_text(f"use {pair[0]} instead of {pair[1]}"), + ) + ), +) + +REPLACEMENT_ERROR_CASES = st.one_of( + 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: ( + { + "premise": None, + "policies": {_normalize_item_like_engine(pair[1]): "prohibit"}, + "version": 2, + }, + pair[0], + pair[1], + "source_prohibited", + ) + ), + 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: ( + { + "premise": None, + "policies": {_normalize_item_like_engine(pair[0]): "prohibit"}, + "version": 2, + }, + pair[0], + pair[1], + "target_prohibited", + ) + ), + 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: ( + {"premise": None, "policies": {}, "version": 2}, + pair[0], + pair[1], + "source_absent", + ) + ), +) + +POLICY_MACHINE_OPERATIONS = st.one_of( + CANONICAL_GRAMMAR_ITEM_TEXT.map(lambda item: ("use", item)), + CANONICAL_GRAMMAR_ITEM_TEXT.map(lambda item: ("prohibit", item)), + CANONICAL_GRAMMAR_ITEM_TEXT.map(lambda item: ("remove", item)), + st.tuples(CANONICAL_GRAMMAR_ITEM_TEXT, CANONICAL_GRAMMAR_ITEM_TEXT).map( + lambda pair: ("replace", pair[0], pair[1]) + ), +) + def _payload_has_stable_export_import_cycle(payload: dict[str, object]) -> bool: engine = Engine() @@ -236,6 +343,20 @@ def _payload_has_stable_export_import_cycle(payload: dict[str, object]) -> bool: ), ) +REPLACEMENT_NEAR_MISS_CASES = st.one_of( + VALID_NONEMPTY_ITEM_TEXT.map(lambda old_item: f"use instead of {old_item}"), + VALID_USE_ITEM_TEXT.map(lambda new_item: f"use {new_item} instead of"), + st.tuples(VALID_USE_ITEM_TEXT, VALID_NONEMPTY_ITEM_TEXT, VALID_NONEMPTY_ITEM_TEXT) + .filter( + lambda parts: ( + _normalize_item_like_engine(parts[0]) != _normalize_item_like_engine(parts[1]) + and _normalize_item_like_engine(parts[0]) != _normalize_item_like_engine(parts[2]) + and _normalize_item_like_engine(parts[1]) != _normalize_item_like_engine(parts[2]) + ) + ) + .map(lambda parts: f"use {parts[0]} instead of {parts[1]} instead of {parts[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: @@ -263,6 +384,66 @@ def test_grammar_helper_render_decompose_round_trip_is_stable( assert grammar_module._render_directive(kind, **operands) == rendered +@given(st.sampled_from(get_directive_metadata())) +def test_public_directive_metadata_matches_internal_rendering_contract( + metadata: grammar_module.DirectiveMetadata, +) -> None: + spec = grammar_module._DIRECTIVE_SPECS[metadata.kind] + + assert metadata.canonical_start == spec.canonical_start + assert metadata.operand_names == spec.operand_names + + if metadata.kind is DirectiveKind.SET_PREMISE: + rendered = grammar_module._render_directive(metadata.kind, value="concise replies") + elif metadata.kind is DirectiveKind.CHANGE_PREMISE: + rendered = grammar_module._render_directive(metadata.kind, value="formal tone") + elif metadata.kind is DirectiveKind.USE_ITEM: + rendered = grammar_module._render_directive(metadata.kind, item="docker") + elif metadata.kind is DirectiveKind.PROHIBIT_ITEM: + rendered = grammar_module._render_directive(metadata.kind, item="peanuts") + elif metadata.kind is DirectiveKind.REMOVE_POLICY: + rendered = grammar_module._render_directive(metadata.kind, item="docker") + elif metadata.kind is DirectiveKind.REPLACE_USE: + rendered = grammar_module._render_directive( + metadata.kind, + new_item="podman", + old_item="docker", + ) + elif ( + metadata.kind is DirectiveKind.CLEAR_PREMISE + or metadata.kind is DirectiveKind.RESET_POLICIES + ): + rendered = grammar_module._render_directive(metadata.kind) + else: + assert metadata.kind is DirectiveKind.CLEAR_STATE + rendered = grammar_module._render_directive(metadata.kind) + + directive = decompose_directive(rendered) + assert isinstance(directive, CanonicalDirective) + assert directive.kind is metadata.kind + assert tuple(directive.operands) == metadata.operand_names + assert _normalize_item_like_engine(rendered.split()[0]) == _normalize_item_like_engine( + metadata.canonical_start.split()[0] + ) + + +def test_public_directive_metadata_only_collides_on_canonical_start_for_use_families() -> None: + starts_by_kind = { + metadata.kind: metadata.canonical_start for metadata in get_directive_metadata() + } + + assert starts_by_kind[DirectiveKind.USE_ITEM] == starts_by_kind[DirectiveKind.REPLACE_USE] + + inverse: dict[str, set[DirectiveKind]] = {} + for kind, start in starts_by_kind.items(): + inverse.setdefault(start, set()).add(kind) + + assert inverse["use"] == {DirectiveKind.USE_ITEM, DirectiveKind.REPLACE_USE} + for start, kinds in inverse.items(): + if start != "use": + assert len(kinds) == 1, start + + @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) @@ -416,12 +597,10 @@ def test_deterministic_replacement_matches_equivalent_explicit_transition( initial_state = case["initial_state"] new_item = case["new_item"] old_item = case["old_item"] - old_present = case["old_present"] assert isinstance(initial_state, dict) assert isinstance(new_item, str) assert isinstance(old_item, str) - assert isinstance(old_present, bool) oracle_engine = Engine() oracle_engine.import_json( @@ -442,7 +621,174 @@ def test_deterministic_replacement_matches_equivalent_explicit_transition( assert decision == expected_decision assert _observations(engine) == expected_state - if not old_present: - followup = engine.step("yes") - assert followup == {"kind": DECISION_NO_DIRECTIVE, "message": None} - assert _observations(engine) == expected_state + +@given(ERROR_CASES) +def test_apply_directive_semantic_errors_never_partially_mutate_state( + case: tuple[dict[str, object], CanonicalDirective], +) -> None: + initial_state, directive = case + engine = Engine() + engine.import_json(json.dumps(initial_state, sort_keys=True, separators=(",", ":"))) + before = _observations(engine) + + decision = engine.apply_directive(directive) + + assert decision == {"kind": DECISION_ERROR, "message": decision["message"]} + assert decision["message"] is not None + assert _observations(engine) == before + + +@given(REPLACEMENT_ERROR_CASES) +def test_apply_directive_replacement_error_cases_preserve_state( + case: tuple[dict[str, object], str, str, str], +) -> None: + initial_state, new_item, old_item, _reason = case + engine = Engine() + engine.import_json(json.dumps(initial_state, sort_keys=True, separators=(",", ":"))) + directive = _canonical_directive_from_text(f"use {new_item} instead of {old_item}") + before = _observations(engine) + + decision = engine.apply_directive(directive) + + assert decision["kind"] == DECISION_ERROR + assert decision["message"] is not None + assert _observations(engine) == before + + +@given(CANONICAL_GRAMMAR_ITEM_TEXT) +def test_apply_directive_replacement_with_normalized_equivalent_keys_is_noop_update( + item: str, +) -> None: + normalized = _normalize_item_like_engine(item) + assume(normalized != "") + engine = Engine() + engine.import_json( + json.dumps( + {"premise": None, "policies": {normalized: "use"}, "version": 2}, + sort_keys=True, + separators=(",", ":"), + ) + ) + before = _observations(engine) + directive = _canonical_directive_from_text(f"use {item.upper()} instead of {item}") + + decision = engine.apply_directive(directive) + + assert decision == {"kind": DECISION_UPDATE, "message": None} + assert _observations(engine) == before + + +@given(DETERMINISTIC_REPLACEMENT_CASES) +def test_apply_directive_valid_replacement_performs_expected_transition( + case: dict[str, object], +) -> None: + initial_state = case["initial_state"] + new_item = case["new_item"] + old_item = case["old_item"] + + assert isinstance(initial_state, dict) + assert isinstance(new_item, str) + assert isinstance(old_item, str) + + engine = Engine() + engine.import_json(json.dumps(initial_state, sort_keys=True, separators=(",", ":"))) + directive = _canonical_directive_from_text(f"use {new_item} instead of {old_item}") + before_premise, before_policies = _observations(engine) + + decision = engine.apply_directive(directive) + + expected_policies = dict(before_policies) + expected_policies.pop(_normalize_item_like_engine(old_item), None) + expected_policies[_normalize_item_like_engine(new_item)] = "use" + + assert decision == {"kind": DECISION_UPDATE, "message": None} + assert _observations(engine) == (before_premise, expected_policies) + + +@given(st.lists(POLICY_MACHINE_OPERATIONS, min_size=1, max_size=25)) +def test_apply_directive_policy_lifecycle_matches_simple_state_model( + operations: list[tuple[str, ...]], +) -> None: + engine = Engine() + model: dict[str, str] = {} + + for operation in operations: + before = _observations(engine) + before_model = dict(model) + + if operation[0] == "use": + item = operation[1] + directive = _canonical_directive_from_text(f"use {item}") + key = _normalize_item_like_engine(item) + expected_error = model.get(key) == "prohibit" + if not expected_error: + model[key] = "use" + elif operation[0] == "prohibit": + item = operation[1] + directive = _canonical_directive_from_text(f"prohibit {item}") + key = _normalize_item_like_engine(item) + expected_error = model.get(key) == "use" + if not expected_error: + model[key] = "prohibit" + elif operation[0] == "remove": + item = operation[1] + directive = _canonical_directive_from_text(f"remove policy {item}") + key = _normalize_item_like_engine(item) + expected_error = False + model.pop(key, None) + else: + assert operation[0] == "replace" + new_item = operation[1] + old_item = operation[2] + directive = _canonical_directive_from_text(f"use {new_item} instead of {old_item}") + new_key = _normalize_item_like_engine(new_item) + old_key = _normalize_item_like_engine(old_item) + if new_key == old_key: + expected_error = False + else: + expected_error = model.get(old_key) != "use" or model.get(new_key) == "prohibit" + if not expected_error: + model.pop(old_key, None) + model[new_key] = "use" + + decision = engine.apply_directive(directive) + + if expected_error: + assert decision["kind"] == DECISION_ERROR + assert _observations(engine) == before + assert model == before_model + else: + assert decision == {"kind": DECISION_UPDATE, "message": None} + assert dict(engine.policies) == model + assert all(value in {"use", "prohibit"} for value in model.values()) + + +@given(REPLACEMENT_NEAR_MISS_CASES) +def test_replacement_near_misses_never_parse_as_canonical_or_mutate_state(text: str) -> None: + engine = Engine() + before = _observations(engine) + parsed = decompose_directive(text) + decision = engine.step(text) + + assert not (isinstance(parsed, CanonicalDirective) and parsed.kind is DirectiveKind.REPLACE_USE) + assert decision == {"kind": DECISION_NO_DIRECTIVE, "message": None} + assert _observations(engine) == before + + +@given( + st.one_of( + VALID_NONEMPTY_ITEM_TEXT.map(lambda old_item: (f"use instead of {old_item}", "new_item")), + VALID_USE_ITEM_TEXT.map(lambda new_item: (f"use {new_item} instead of", "old_item")), + ) +) +def test_incomplete_replacement_forms_report_replace_use_family( + case: tuple[str, str], +) -> None: + text, missing_operand = case + parsed = decompose_directive(text) + + assert parsed == InvalidDirectiveSyntax( + failure=DirectiveSyntaxFailure.MISSING_REQUIRED_OPERAND, + directive_kind=DirectiveKind.REPLACE_USE, + missing_operand=missing_operand, + ) From bd5ffa1a7ad847c890837bbdc648361cc3162c63 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Fri, 14 Aug 2026 00:32:27 -0400 Subject: [PATCH 2/2] test: add engine property coverage --- tests/test_properties.py | 175 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 168 insertions(+), 7 deletions(-) diff --git a/tests/test_properties.py b/tests/test_properties.py index c894856..b7ef4fb 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -229,6 +229,11 @@ def _build_deterministic_replacement_case( .filter( lambda pair: _normalize_item_like_engine(pair[0]) != _normalize_item_like_engine(pair[1]) ) + .filter( + lambda pair: _is_canonical_directive( + decompose_directive(f"use {pair[0]} instead of {pair[1]}") + ) + ) .map( lambda pair: ( {"premise": None, "policies": {}, "version": 2}, @@ -293,6 +298,37 @@ def _build_deterministic_replacement_case( ), ) +PREMISE_MACHINE_OPERATIONS = st.one_of( + CANONICAL_GRAMMAR_PREMISE_TEXT.map(lambda value: ("set", value)), + CANONICAL_GRAMMAR_PREMISE_TEXT.map(lambda value: ("change", value)), + st.sampled_from([("clear_premise",), ("clear_state",)]), +) + +CANONICAL_DIRECTIVE_TEXT_CASES = st.one_of( + CANONICAL_GRAMMAR_PREMISE_TEXT.map(lambda value: f"set premise {value}"), + CANONICAL_GRAMMAR_PREMISE_TEXT.map(lambda value: f"change premise to {value}"), + CANONICAL_GRAMMAR_ITEM_TEXT.map(lambda item: f"use {item}"), + CANONICAL_GRAMMAR_ITEM_TEXT.map(lambda item: f"prohibit {item}"), + CANONICAL_GRAMMAR_ITEM_TEXT.map(lambda item: f"remove policy {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: f"use {pair[0]} instead of {pair[1]}"), + st.sampled_from(["clear premise", "reset policies", "clear state"]), +) + +NONEMPTY_NORMALIZED_KEY_TEXT = NORMALIZATION_SENSITIVE_TEXT.filter( + lambda value: _normalize_item_like_engine(value) != "" +) + +INVALID_EMPTY_NORMALIZED_KEY_TEXT = st.text(alphabet=" \t", min_size=1, max_size=6) + +EQUIVALENT_NORMALIZED_KEY_PAIRS = st.builds( + lambda item: (item, " " + item.upper().replace("'", "’") + " "), + CANONICAL_GRAMMAR_ITEM_TEXT, +) + def _payload_has_stable_export_import_cycle(payload: dict[str, object]) -> bool: engine = Engine() @@ -363,6 +399,28 @@ def test_determinism_same_input_sequence_same_state(inputs: list[str]) -> None: assert _run_sequence(inputs) == _run_sequence(inputs) +@given(VALID_STATE_PAYLOADS, CANONICAL_DIRECTIVE_TEXT_CASES) +def test_step_and_apply_directive_are_equivalent_for_canonical_inputs( + initial_state: dict[str, object], + text: str, +) -> None: + directive = decompose_directive(text) + assert isinstance(directive, CanonicalDirective) + + step_engine = Engine() + step_engine.import_json(json.dumps(initial_state, sort_keys=True, separators=(",", ":"))) + step_decision = step_engine.step(text) + step_state = _observations(step_engine) + + apply_engine = Engine() + apply_engine.import_json(json.dumps(initial_state, sort_keys=True, separators=(",", ":"))) + apply_decision = apply_engine.apply_directive(directive) + apply_state = _observations(apply_engine) + + assert step_decision == apply_decision + assert step_state == apply_state + + @given(GRAMMAR_RENDER_CASES) def test_grammar_helper_render_decompose_round_trip_is_stable( case: dict[str, DirectiveKind | dict[str, str]], @@ -659,8 +717,11 @@ def test_apply_directive_replacement_error_cases_preserve_state( def test_apply_directive_replacement_with_normalized_equivalent_keys_is_noop_update( item: str, ) -> None: - normalized = _normalize_item_like_engine(item) - assume(normalized != "") + original_normalized = _normalize_item_like_engine(item) + upper_normalized = _normalize_item_like_engine(item.upper()) + assume(original_normalized != "") + assume(original_normalized == upper_normalized) + normalized = original_normalized engine = Engine() engine.import_json( json.dumps( @@ -763,6 +824,107 @@ def test_apply_directive_policy_lifecycle_matches_simple_state_model( assert all(value in {"use", "prohibit"} for value in model.values()) +@given(st.lists(PREMISE_MACHINE_OPERATIONS, min_size=1, max_size=25)) +def test_apply_directive_premise_lifecycle_matches_simple_model( + operations: list[tuple[str, ...]], +) -> None: + engine = Engine() + model_premise: str | None = None + + for operation in operations: + before_premise, before_policies = _observations(engine) + before_model_premise = model_premise + + if operation[0] == "set": + value = operation[1] + directive = _canonical_directive_from_text(f"set premise {value}") + expected_error = model_premise is not None + if not expected_error: + model_premise = _sanitize_premise_like_engine(value) + elif operation[0] == "change": + value = operation[1] + directive = _canonical_directive_from_text(f"change premise to {value}") + expected_error = model_premise is None + if not expected_error: + model_premise = _sanitize_premise_like_engine(value) + elif operation[0] == "clear_premise": + directive = _canonical_directive_from_text("clear premise") + expected_error = False + model_premise = None + else: + assert operation[0] == "clear_state" + directive = _canonical_directive_from_text("clear state") + expected_error = False + model_premise = None + + decision = engine.apply_directive(directive) + after_premise, after_policies = _observations(engine) + + if expected_error: + assert decision["kind"] == DECISION_ERROR + assert after_premise == before_premise + assert after_policies == before_policies + assert model_premise == before_model_premise + else: + assert decision == {"kind": DECISION_UPDATE, "message": None} + assert after_premise == model_premise + if operation[0] == "clear_state": + assert after_policies == {} + else: + assert after_policies == before_policies + + +@given(EQUIVALENT_NORMALIZED_KEY_PAIRS) +def test_import_json_normalization_converges_equivalent_policy_keys( + pair: tuple[str, str], +) -> None: + raw_a, raw_b = pair + + payload = { + "premise": None, + "policies": {raw_a: "use", raw_b: "prohibit"}, + "version": 2, + } + engine = Engine() + engine.import_json(json.dumps(payload, sort_keys=True, separators=(",", ":"))) + + normalized_key = _normalize_item_like_engine(raw_b) + expected_value = { + _normalize_item_like_engine(raw_key): value + for raw_key, value in sorted(payload["policies"].items()) + }[normalized_key] + assert _observations(engine) == (None, {normalized_key: expected_value}) + + +@given(INVALID_EMPTY_NORMALIZED_KEY_TEXT) +def test_import_json_rejects_policy_keys_that_normalize_to_empty(key: str) -> None: + engine = Engine() + before = _observations(engine) + payload = {"premise": None, "policies": {key: "use"}, "version": 2} + + try: + engine.import_json(json.dumps(payload, sort_keys=True, separators=(",", ":"))) + except ValueError as exc: + assert str(exc) == "Invalid state payload." + else: + raise AssertionError("Expected ValueError for empty normalized policy key") + + assert _observations(engine) == before + + +@given(VALID_STATE_PAYLOADS) +def test_import_json_preserves_authoritative_invariants_for_generated_payloads( + payload: dict[str, object], +) -> None: + engine = Engine() + engine.import_json(json.dumps(payload, sort_keys=True, separators=(",", ":"))) + premise, policies = _observations(engine) + + assert premise is None or premise == _sanitize_premise_like_engine(premise) + assert all(key == _normalize_item_like_engine(key) for key in policies) + assert all(value in {"use", "prohibit"} for value in policies.values()) + + @given(REPLACEMENT_NEAR_MISS_CASES) def test_replacement_near_misses_never_parse_as_canonical_or_mutate_state(text: str) -> None: engine = Engine() @@ -787,8 +949,7 @@ def test_incomplete_replacement_forms_report_replace_use_family( text, missing_operand = case parsed = decompose_directive(text) - assert parsed == InvalidDirectiveSyntax( - failure=DirectiveSyntaxFailure.MISSING_REQUIRED_OPERAND, - directive_kind=DirectiveKind.REPLACE_USE, - missing_operand=missing_operand, - ) + assert isinstance(parsed, InvalidDirectiveSyntax) + assert parsed.failure is DirectiveSyntaxFailure.MISSING_REQUIRED_OPERAND + assert parsed.missing_operand == missing_operand + assert parsed.directive_kind in {DirectiveKind.REPLACE_USE, DirectiveKind.USE_ITEM}