diff --git a/src/context_compiler/controller.py b/src/context_compiler/controller.py index 1cbe9e6..8a8b23e 100644 --- a/src/context_compiler/controller.py +++ b/src/context_compiler/controller.py @@ -122,19 +122,12 @@ def step(engine: Engine, user_input: str) -> StepResult: def preview(engine: Engine, user_input: str) -> PreviewResult: - state_json = engine.export_json() state_before = engine.state - - decision: Decision | None = None - state_after: State | None = None - try: - decision = engine.step(user_input) - state_after = engine.state - finally: - engine.import_json(state_json) - - assert decision is not None - assert state_after is not None + # Preview intentionally consumes the engine's private evaluator so preview and + # committed execution share one transition path without making evaluation public. + evaluated = engine._evaluate_transition(state_before, user_input) # noqa: SLF001 + decision = evaluated.decision + state_after = evaluated.next_state diff = state_diff(state_before, state_after) would_mutate = diff["changed"] diff --git a/src/context_compiler/engine.py b/src/context_compiler/engine.py index 3bddcb1..78adc88 100644 --- a/src/context_compiler/engine.py +++ b/src/context_compiler/engine.py @@ -68,6 +68,12 @@ class Action: old_item: str | None = None +@dataclass(frozen=True) +class _EvaluatedTransition: + decision: Decision + next_state: State + + _NO_DIRECTIVE: NoDirectiveDecision = {"kind": DECISION_NO_DIRECTIVE} @@ -99,20 +105,27 @@ def import_json(self, payload: str) -> None: self._replace_state(_load_state_json(payload)) def step(self, user_input: str) -> Decision: + evaluated = self._evaluate_transition(self._state, user_input) + self._replace_state(evaluated.next_state) + return evaluated.decision + + def _evaluate_transition(self, state: State, user_input: str) -> _EvaluatedTransition: action = _parse_directive(user_input) if action is None: - return _NO_DIRECTIVE.copy() + return _EvaluatedTransition(decision=_NO_DIRECTIVE.copy(), next_state=deepcopy(state)) - error_decision = self._pre_mutation_error(action) + error_decision = self._pre_mutation_error(action, state=state) if error_decision is not None: - return error_decision + return _EvaluatedTransition(decision=error_decision, next_state=deepcopy(state)) - return self._apply_action(action) + next_state = self._apply_action(action, state=state) + return _EvaluatedTransition(decision=_update_decision(next_state), next_state=next_state) def _replace_state(self, state: State) -> None: self._state = state - def _pre_mutation_error(self, action: Action) -> Decision | None: + def _pre_mutation_error(self, action: Action, *, state: State | None = None) -> Decision | None: + candidate_state = self._state if state is None else state # Single error path: all error outcomes are detected before any mutation. if action.kind in {"set_premise", "change_premise"}: assert action.value is not None @@ -149,16 +162,16 @@ def _pre_mutation_error(self, action: Action) -> Decision | None: "Policy item cannot be empty.\nUse 'prohibit ' with a non-empty value." ) - if action.kind == "set_premise" and self._state[STATE_PREMISE] is not None: + if action.kind == "set_premise" and candidate_state[STATE_PREMISE] is not None: return _error("Premise already set.\nUse 'change premise to ' to modify it.") - if action.kind == "change_premise" and self._state[STATE_PREMISE] is None: + if action.kind == "change_premise" and candidate_state[STATE_PREMISE] is None: return _error("No premise is set.\nUse 'set premise ' to define one.") if action.kind == "use_item": assert action.item is not None item_key = _normalize_item(action.item) - if self._state[STATE_POLICIES].get(item_key) == POLICY_PROHIBIT: + if candidate_state[STATE_POLICIES].get(item_key) == POLICY_PROHIBIT: return _error( f'"{item_key}" is currently prohibited.\nRemove or replace it before using it.' ) @@ -166,7 +179,7 @@ def _pre_mutation_error(self, action: Action) -> Decision | None: if action.kind == "prohibit_item": assert action.item is not None item_key = _normalize_item(action.item) - if self._state[STATE_POLICIES].get(item_key) == POLICY_USE: + if candidate_state[STATE_POLICIES].get(item_key) == POLICY_USE: return _error( f'"{item_key}" is currently in use.\n' "Remove or replace it before prohibiting it." @@ -180,8 +193,8 @@ def _pre_mutation_error(self, action: Action) -> Decision | None: if new_key == old_key: return None - old_state = self._state[STATE_POLICIES].get(old_key) - new_state = self._state[STATE_POLICIES].get(new_key) + old_state = candidate_state[STATE_POLICIES].get(old_key) + new_state = candidate_state[STATE_POLICIES].get(new_key) if old_state == POLICY_PROHIBIT: return _error( f'"{action.old_item}" is currently prohibited.\n' @@ -200,65 +213,65 @@ def _pre_mutation_error(self, action: Action) -> Decision | None: return None - def _apply_action(self, action: Action) -> Decision: + def _apply_action(self, action: Action, *, state: State) -> State: + next_state = deepcopy(state) kind = action.kind if kind == "set_premise": assert action.value is not None - self._state[STATE_PREMISE] = _sanitize_premise_value(action.value) - return _update_decision(self._state) + next_state[STATE_PREMISE] = _sanitize_premise_value(action.value) + return next_state if kind == "change_premise": assert action.value is not None - self._state[STATE_PREMISE] = _sanitize_premise_value(action.value) - return _update_decision(self._state) + next_state[STATE_PREMISE] = _sanitize_premise_value(action.value) + return next_state if kind == "use_item": assert action.item is not None item_key = _normalize_item(action.item) # Idempotent directives are updates even if state does not change. - self._state[STATE_POLICIES][item_key] = POLICY_USE - return _update_decision(self._state) + next_state[STATE_POLICIES][item_key] = POLICY_USE + return next_state if kind == "prohibit_item": assert action.item is not None item_key = _normalize_item(action.item) # Idempotent directives are updates even if state does not change. - self._state[STATE_POLICIES][item_key] = POLICY_PROHIBIT - return _update_decision(self._state) + next_state[STATE_POLICIES][item_key] = POLICY_PROHIBIT + return next_state if kind == "replace_use": assert action.new_item is not None assert action.old_item is not None - self._apply_replacement_explicit(action.new_item, action.old_item) - return _update_decision(self._state) + self._apply_replacement_explicit(next_state, action.new_item, action.old_item) + return next_state if kind == "remove_policy_item": assert action.item is not None item_key = _normalize_item(action.item) - self._state[STATE_POLICIES].pop(item_key, None) - return _update_decision(self._state) + next_state[STATE_POLICIES].pop(item_key, None) + return next_state if kind == "clear_premise": - self._state[STATE_PREMISE] = None - return _update_decision(self._state) + next_state[STATE_PREMISE] = None + return next_state if kind == "reset_policies": - self._state[STATE_POLICIES] = {} - return _update_decision(self._state) + next_state[STATE_POLICIES] = {} + return next_state - self._state = _initial_state() - return _update_decision(self._state) + return _initial_state() - def _apply_replacement_explicit(self, new_item: str, old_item: str) -> None: + def _apply_replacement_explicit(self, state: State, new_item: str, old_item: str) -> None: new_key = _normalize_item(new_item) old_key = _normalize_item(old_item) if new_key == old_key: return - self._state[STATE_POLICIES].pop(old_key, None) - self._state[STATE_POLICIES][new_key] = POLICY_USE + state[STATE_POLICIES].pop(old_key, None) + state[STATE_POLICIES][new_key] = POLICY_USE def _parse_directive(user_input: str) -> Action | None: diff --git a/tests/test_controller.py b/tests/test_controller.py index e2bf6dc..171ddff 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -160,17 +160,24 @@ def test_state_diff_policy_removed_and_value_changed() -> None: assert diff["policies"]["added"] == {} -def test_preview_fails_when_state_restore_fails( +def test_preview_does_not_depend_on_persistence_round_trip( monkeypatch: pytest.MonkeyPatch, ) -> None: engine = create_engine() - def _boom(_: object) -> None: - raise RuntimeError("restore failed") + def _export_boom() -> str: + raise RuntimeError("preview should not export state") - monkeypatch.setattr(engine, "import_json", _boom) - with pytest.raises(RuntimeError, match="restore failed"): - preview(engine, "set premise concise replies") + def _import_boom(_: object) -> None: + raise RuntimeError("preview should not import state") + + monkeypatch.setattr(engine, "export_json", _export_boom) + monkeypatch.setattr(engine, "import_json", _import_boom) + + result = preview(engine, "set premise concise replies") + + assert result["decision"]["kind"] == DECISION_UPDATE + assert engine.state == {"premise": None, "policies": {}, "version": 2} def test_controller_preview_fixtures() -> None: