diff --git a/src/state/app_state.py b/src/state/app_state.py new file mode 100644 index 000000000..d13cad665 --- /dev/null +++ b/src/state/app_state.py @@ -0,0 +1,275 @@ +"""Reactive UI state — the Python analogue of ``state/AppStateStore.ts``. + +Phase 2.1 of the ch03 state refactor: a minimal, immutable-update-style +``AppState`` dataclass backed by ``src.utils.store.create_store``, with +a centralized ``on_change_app_state`` handler that bridges UI state +transitions to bootstrap-state mirrors and external persistence. + +**Two-tier separation.** This module imports from +``src.bootstrap.state`` (for the mirror writes performed in +``on_change_app_state``) but never the reverse — the dependency direction +is preserved. ``src.bootstrap.state`` remains a DAG leaf per the +import-linter contract. + +**Scope.** Phase 2.1 covers the *minimum* AppState fields that map to the +chapter's named side effects: + +* ``main_loop_model`` → mirror to ``set_main_loop_model_override`` in + bootstrap; persist to settings (placeholder until settings.json + layering is wired). +* ``verbose`` → persist to global config. +* ``expanded_view`` → persist as legacy ``showExpandedTodos`` / + ``showSpinnerTree`` (matches TS at ``onChangeAppState.ts:123-136``). +* ``permission_mode`` → notify external listeners (CCR/SDK status + stream). Today the notifiers are no-ops; they become real once + the CCR bridge is wired. + +Many more AppState fields exist in TS (~86 total per the gap analysis). +They land in this dataclass when their consumers do. The chapter's +single-file discipline applies here too: do not split ``AppState`` into +per-domain dataclasses, even when the line count grows. + +**Side-effect coverage contract.** ``_FIELD_HANDLERS`` is a registry +mapping each AppState field to a handler function. Fields with no +side effect get an explicit handler whose body is just ``return`` — +the function name (``_on__change``) is greppable and the +no-op decision is visible at the site. The unit test at +``tests/test_app_state.py::TestSideEffectCoverage`` asserts that every +field in the dataclass appears in the registry, so adding a new field +without a handler is a compile-time-equivalent failure rather than a +silent miss. This is the structural-coverage mechanism the chapter's +lesson demands. +""" + +from __future__ import annotations + +import dataclasses +import logging +from dataclasses import dataclass, field, fields as dc_fields +from typing import Any, Callable + +from src.bootstrap.state import ( + set_main_loop_model_override, +) +from src.utils.store import Store, create_store + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# AppState dataclass +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class AppState: + """Frozen dataclass: the store treats each setState as + ``(prev) -> new_instance`` where ``new_instance`` is built via + ``dataclasses.replace(prev, ...)``. + + Use ``replace_state(prev, field=value)`` to update — the store's + identity-skip check (``if next is prev``) requires a different + reference on any meaningful change. + """ + + # Model selection (TS: state/AppStateStore.ts:93) + main_loop_model: str | None = None + + # Verbose mode (TS: AppStateStore.ts:92) + verbose: bool = False + + # Expanded view: 'none' | 'tasks' | 'teammates' (TS: AppStateStore.ts:96) + expanded_view: str = "none" + + # Permission mode (slice of toolPermissionContext — TS: AppStateStore.ts:110). + # Stored as a string here to avoid a tight coupling with the existing + # permissions package's enums; the bridge handlers can normalize. + permission_mode: str = "default" + + # ``initial_message`` — set by entrypoints to queue a prompt for the REPL + # to process at startup. TS: AppStateStore.ts:406. + initial_message: str | None = None + + +def replace_state(state: AppState, **changes: Any) -> AppState: + """Return a copy of ``state`` with ``changes`` applied. Equivalent + to ``dataclasses.replace(state, ...)`` — exists as a named helper + to keep the call sites readable.""" + return dataclasses.replace(state, **changes) + + +def get_default_app_state() -> AppState: + """Mirror of TS ``getDefaultAppState`` (``AppStateStore.ts:458``).""" + return AppState() + + +# --------------------------------------------------------------------------- +# Side-effect handlers (the on_change_app_state router) +# --------------------------------------------------------------------------- + + +def _on_main_loop_model_change(old: AppState, new: AppState) -> None: + """Mirror model choice into bootstrap singleton. + + Matches TS at ``onChangeAppState.ts:97-120`` — when the user changes + the model (via /model slash command or the model picker), the + bootstrap-state override must update so the next API call reads the + new value, and settings persistence happens as a side effect. + + Settings persistence is currently no-op — the settings.json layering + lives in ``src.settings`` and the writer is not yet wired through + a single chokepoint. Plan §P2.1 left this stub here as the wiring + target. + """ + if old.main_loop_model == new.main_loop_model: + return + set_main_loop_model_override(new.main_loop_model) + logger.debug( + "AppState.main_loop_model %s -> %s — mirrored to bootstrap", + old.main_loop_model, + new.main_loop_model, + ) + # TODO: persist to user settings via ``src.settings`` once the + # writer-side has a single chokepoint. + + +def _on_verbose_change(old: AppState, new: AppState) -> None: + if old.verbose == new.verbose: + return + logger.debug("AppState.verbose %s -> %s", old.verbose, new.verbose) + # TODO: persist to global config. + + +def _on_expanded_view_change(old: AppState, new: AppState) -> None: + if old.expanded_view == new.expanded_view: + return + logger.debug( + "AppState.expanded_view %s -> %s — persist as showExpandedTodos/showSpinnerTree", + old.expanded_view, + new.expanded_view, + ) + # TODO: persist to global config as showExpandedTodos + + # showSpinnerTree (TS: onChangeAppState.ts:123-136). + + +# Permission-mode notification hooks. Real listeners (CCR bridge, SDK +# status stream) are registered via ``set_permission_mode_listener`` and +# ``set_session_metadata_listener`` — see TS ``utils/sessionState.ts``. +# Today the Python equivalents are placeholder; the slots are here so +# adding real listeners later is a one-line wiring change. + +_permission_mode_listener: Callable[[str], None] | None = None +_session_metadata_listener: Callable[[dict[str, Any]], None] | None = None + + +def set_permission_mode_listener(cb: Callable[[str], None] | None) -> None: + """Register a callback for permission-mode changes. Mirrors TS + ``setPermissionModeChangedListener`` (``utils/sessionState.ts:79``).""" + global _permission_mode_listener + _permission_mode_listener = cb + + +def set_session_metadata_listener(cb: Callable[[dict[str, Any]], None] | None) -> None: + """Register a callback for external-metadata changes. Mirrors TS + ``setSessionMetadataChangedListener`` (``utils/sessionState.ts:66``).""" + global _session_metadata_listener + _session_metadata_listener = cb + + +def _on_permission_mode_change(old: AppState, new: AppState) -> None: + """Centralized side effect for permission-mode changes. + + Mirrors TS ``onChangeAppState.ts:67-94``: notify CCR external metadata + AND the SDK status stream. The single chokepoint here is the *whole + point* of the architecture — pre-Chapter-3, this notification was + duplicated across 6+ mutation sites and was broken in most of them. + """ + if old.permission_mode == new.permission_mode: + return + logger.debug( + "AppState.permission_mode %s -> %s — notifying CCR + SDK", + old.permission_mode, + new.permission_mode, + ) + if _session_metadata_listener is not None: + try: + _session_metadata_listener({"permission_mode": new.permission_mode}) + except Exception: + logger.exception("session_metadata_listener raised") + if _permission_mode_listener is not None: + try: + _permission_mode_listener(new.permission_mode) + except Exception: + logger.exception("permission_mode_listener raised") + + +def _on_initial_message_change(old: AppState, new: AppState) -> None: + """``initial_message`` is consumed by the REPL on startup; no + centralized side effect needed.""" + # Intentional no-op — the REPL reads ``initial_message`` directly via + # the store's ``get_state`` and clears it after processing. + return + + +# Registry. EVERY field in AppState must appear here as a handler +# (function-form, including explicit no-ops). The coverage test enforces +# this — adding a new AppState field without a handler entry fails +# ``test_every_field_appears_in_handler_registry``. +_FIELD_HANDLERS: dict[str, Callable[[AppState, AppState], None]] = { + "main_loop_model": _on_main_loop_model_change, + "verbose": _on_verbose_change, + "expanded_view": _on_expanded_view_change, + "permission_mode": _on_permission_mode_change, + "initial_message": _on_initial_message_change, +} + + +def on_change_app_state(old_state: AppState, new_state: AppState) -> None: + """Route the diff to each field's handler. + + Mirrors TS ``onChangeAppState`` (``state/onChangeAppState.ts``). Each + handler is responsible for checking ``old == new`` and returning early + when the field didn't change — this means a single ``setState`` that + mutates multiple fields fires all relevant handlers exactly once. + """ + for fname in (f.name for f in dc_fields(AppState)): + handler = _FIELD_HANDLERS.get(fname) + if handler is None: + continue + try: + handler(old_state, new_state) + except Exception: + logger.exception( + "on_change_app_state handler for %r raised; continuing", + fname, + ) + + +# --------------------------------------------------------------------------- +# Store factory +# --------------------------------------------------------------------------- + + +def create_app_state_store( + initial: AppState | None = None, +) -> Store[AppState]: + """Construct an AppState store with the centralized side-effect router. + + Equivalent to TS: + ``createStore(getDefaultAppState(), onChangeAppState)``. + """ + return create_store( + initial if initial is not None else get_default_app_state(), + on_change=on_change_app_state, + ) + + +__all__ = [ + "AppState", + "create_app_state_store", + "get_default_app_state", + "on_change_app_state", + "replace_state", + "set_permission_mode_listener", + "set_session_metadata_listener", +] diff --git a/tests/test_app_state.py b/tests/test_app_state.py new file mode 100644 index 000000000..d7ec4d086 --- /dev/null +++ b/tests/test_app_state.py @@ -0,0 +1,227 @@ +"""Tests for ``src/state/app_state.py`` — Phase 2.1 AppState + onChange. + +Verifies: +* The frozen dataclass shape supports the immutable-update discipline. +* ``replace_state`` produces a distinct reference (so the store doesn't + identity-skip a real change). +* ``on_change_app_state`` mirrors model changes into bootstrap. +* The permission-mode listener fires exactly once per real change. +* The structural-coverage contract: every AppState field has a handler + (real or sentinel) in ``_FIELD_HANDLERS``. +* End-to-end: store + onChange + listener form the chapter's chokepoint. +""" + +from __future__ import annotations + +import dataclasses +import unittest +from typing import Any + +import pytest + +from src.bootstrap.state import ( + get_main_loop_model_override, + reset_state_for_tests, +) +from src.state.app_state import ( + AppState, + _FIELD_HANDLERS, + create_app_state_store, + get_default_app_state, + on_change_app_state, + replace_state, + set_permission_mode_listener, + set_session_metadata_listener, +) + + +@pytest.fixture(autouse=True) +def _reset_bootstrap_and_listeners(): + reset_state_for_tests() + set_permission_mode_listener(None) + set_session_metadata_listener(None) + yield + reset_state_for_tests() + set_permission_mode_listener(None) + set_session_metadata_listener(None) + + +class TestAppStateDataclass(unittest.TestCase): + def test_default_state_is_well_formed(self) -> None: + state = get_default_app_state() + self.assertIsNone(state.main_loop_model) + self.assertFalse(state.verbose) + self.assertEqual(state.expanded_view, "none") + self.assertEqual(state.permission_mode, "default") + self.assertIsNone(state.initial_message) + + def test_state_is_frozen(self) -> None: + state = get_default_app_state() + with self.assertRaises(dataclasses.FrozenInstanceError): + state.verbose = True # type: ignore[misc] + + def test_replace_state_returns_distinct_reference(self) -> None: + state = get_default_app_state() + new = replace_state(state, verbose=True) + self.assertIsNot(new, state) + self.assertTrue(new.verbose) + self.assertFalse(state.verbose) # original untouched + + def test_replace_state_preserves_unchanged_fields(self) -> None: + state = get_default_app_state() + state = replace_state(state, main_loop_model="claude-opus-4") + new = replace_state(state, verbose=True) + self.assertEqual(new.main_loop_model, "claude-opus-4") + self.assertTrue(new.verbose) + + +class TestOnChangeMirrorsModel(unittest.TestCase): + def test_main_loop_model_change_mirrors_to_bootstrap(self) -> None: + old = get_default_app_state() + new = replace_state(old, main_loop_model="claude-sonnet-4-6") + + on_change_app_state(old, new) + + self.assertEqual(get_main_loop_model_override(), "claude-sonnet-4-6") + + def test_clearing_main_loop_model_mirrors_to_bootstrap(self) -> None: + old = replace_state(get_default_app_state(), main_loop_model="claude-opus-4") + # Pre-condition: bootstrap is already set + on_change_app_state(get_default_app_state(), old) + self.assertEqual(get_main_loop_model_override(), "claude-opus-4") + + new = replace_state(old, main_loop_model=None) + on_change_app_state(old, new) + + self.assertIsNone(get_main_loop_model_override()) + + def test_no_change_does_not_touch_bootstrap(self) -> None: + state = replace_state(get_default_app_state(), main_loop_model="claude-opus-4") + on_change_app_state(state, state) + # bootstrap was never touched because old == new + self.assertIsNone(get_main_loop_model_override()) + + +class TestPermissionModeNotification(unittest.TestCase): + def test_permission_mode_change_fires_listener(self) -> None: + received: list[str] = [] + set_permission_mode_listener(received.append) + + old = get_default_app_state() + new = replace_state(old, permission_mode="plan") + on_change_app_state(old, new) + + self.assertEqual(received, ["plan"]) + + def test_permission_mode_change_fires_metadata_listener(self) -> None: + received: list[dict[str, Any]] = [] + set_session_metadata_listener(received.append) + + old = get_default_app_state() + new = replace_state(old, permission_mode="acceptEdits") + on_change_app_state(old, new) + + self.assertEqual(received, [{"permission_mode": "acceptEdits"}]) + + def test_listener_exception_does_not_propagate(self) -> None: + """A buggy listener must not break the dispatch.""" + + def raising_listener(_mode: str) -> None: + raise RuntimeError("listener boom") + + set_permission_mode_listener(raising_listener) + + old = get_default_app_state() + new = replace_state(old, permission_mode="plan") + # Should not raise + on_change_app_state(old, new) + + def test_no_change_does_not_fire(self) -> None: + received: list[str] = [] + set_permission_mode_listener(received.append) + + old = replace_state(get_default_app_state(), permission_mode="plan") + on_change_app_state(old, old) + + self.assertEqual(received, []) + + +class TestSideEffectCoverage(unittest.TestCase): + """The architectural contract that the chapter's lesson demands: + every AppState field has an entry in ``_FIELD_HANDLERS``. + + If a new AppState field lands without a handler (real or no-op), + this test fails — the developer must explicitly decide whether the + field needs a side effect, not implicitly skip the question.""" + + def test_every_field_appears_in_handler_registry(self) -> None: + field_names = {f.name for f in dataclasses.fields(AppState)} + registry_keys = set(_FIELD_HANDLERS.keys()) + missing = field_names - registry_keys + self.assertEqual( + missing, + set(), + f"AppState fields without handlers in _FIELD_HANDLERS: {missing}. " + f"Add a handler entry — every field in AppState must appear in _FIELD_HANDLERS.", + ) + + def test_no_handler_entries_for_unknown_fields(self) -> None: + field_names = {f.name for f in dataclasses.fields(AppState)} + registry_keys = set(_FIELD_HANDLERS.keys()) + extra = registry_keys - field_names + self.assertEqual( + extra, + set(), + f"_FIELD_HANDLERS has entries for non-existent fields: {extra}", + ) + + +class TestEndToEndStore(unittest.TestCase): + """The chapter's chokepoint: a setState triggers onChange which + mirrors to bootstrap. Verify the wiring end-to-end through + ``create_app_state_store``.""" + + def test_setstate_triggers_bootstrap_mirror(self) -> None: + store = create_app_state_store() + + store.set_state( + lambda prev: replace_state(prev, main_loop_model="claude-opus-4") + ) + + self.assertEqual(get_main_loop_model_override(), "claude-opus-4") + + def test_setstate_triggers_permission_mode_listener(self) -> None: + received: list[str] = [] + set_permission_mode_listener(received.append) + + store = create_app_state_store() + + store.set_state(lambda prev: replace_state(prev, permission_mode="plan")) + + self.assertEqual(received, ["plan"]) + + def test_subscribe_fires_after_onchange(self) -> None: + """Order discipline: bootstrap mirror writes BEFORE subscribers + re-render (the chapter's architectural property).""" + order: list[str] = [] + + # Subscribe a listener that captures bootstrap state at the time + # of notification — should already reflect the new value. + def listener() -> None: + order.append( + f"listener_sees:{get_main_loop_model_override()}", + ) + + store = create_app_state_store() + store.subscribe(listener) + + store.set_state( + lambda prev: replace_state(prev, main_loop_model="claude-opus-4") + ) + + # Listener sees the mirror — proves onChange ran first + self.assertEqual(order, ["listener_sees:claude-opus-4"]) + + +if __name__ == "__main__": + unittest.main()