Skip to content

feat: migrate all session_state usage to capabilities #149

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bkrabach wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat: migrate all session_state usage to capabilities #149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2bd5437
feat: migrate mode_discovery and mode_hooks reads to get_capability
bkrabach Mar 21, 2026
b44f00b
feat: migrate active_mode reads/writes in app-cli to capabilities
bkrabach Mar 21, 2026
dc83983
feat: remove session_state init block from CommandProcessor.__init__
bkrabach Mar 21, 2026
78c8492
fix(tests): remove migrated active_mode key from session_state in tes…
bkrabach Mar 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 23 additions & 28 deletions amplifier_app_cli/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -316,17 +316,15 @@ class CommandProcessor:
def __init__(self, session: AmplifierSession, bundle_name: str = "unknown"):
self.session = session
self.bundle_name = bundle_name
# Initialize session_state if not present
if not hasattr(self.session.coordinator, "session_state"):
self.session.coordinator.session_state = {}
if "active_mode" not in self.session.coordinator.session_state:
self.session.coordinator.session_state["active_mode"] = None
# Initialize active_mode capability if not already set
if self.session.coordinator.get_capability("modes.active_mode") is None:
self.session.coordinator.register_capability("modes.active_mode", None)
# Populate mode shortcuts from discovery (if available)
self._populate_mode_shortcuts()

def _populate_mode_shortcuts(self) -> None:
"""Populate MODE_SHORTCUTS from mode discovery."""
discovery = self.session.coordinator.session_state.get("mode_discovery")
discovery = self.session.coordinator.get_capability("modes.discovery")
if discovery and hasattr(discovery, "get_shortcuts"):
shortcuts = discovery.get_shortcuts()
# Update class-level shortcuts dict
Expand Down Expand Up @@ -374,7 +372,7 @@ def process_input(self, user_input: str) -> tuple[str, dict[str, Any]]:
return "unknown_command", {"command": command}

# Regular prompt
active_mode = self.session.coordinator.session_state.get("active_mode")
active_mode = self.session.coordinator.get_capability("modes.active_mode")
return "prompt", {"text": user_input, "active_mode": active_mode}

def _split_mode_trailing(self, args: str) -> tuple[str, str | None]:
Expand Down Expand Up @@ -475,15 +473,14 @@ async def handle_command(self, action: str, data: dict[str, Any]) -> str:
async def _handle_mode(self, args: str) -> str:
"""Handle /mode command for setting, toggling, or clearing modes."""
args = args.strip().lower()
session_state = self.session.coordinator.session_state
current_mode = session_state.get("active_mode")
current_mode = self.session.coordinator.get_capability("modes.active_mode")

# /mode off - clear any active mode
if args == "off":
if current_mode:
session_state["active_mode"] = None
self.session.coordinator.register_capability("modes.active_mode", None)
# Reset warnings in mode hooks if present
mode_hooks = session_state.get("mode_hooks")
mode_hooks = self.session.coordinator.get_capability("modes.hooks")
if mode_hooks and hasattr(mode_hooks, "reset_warnings"):
mode_hooks.reset_warnings()
return f"Mode off: {current_mode}"
Expand All @@ -501,7 +498,7 @@ async def _handle_mode(self, args: str) -> str:
explicit_state = parts[1] if len(parts) > 1 else None

# Check if mode exists via discovery
discovery = session_state.get("mode_discovery")
discovery = self.session.coordinator.get_capability("modes.discovery")
if discovery:
mode_def = discovery.find(mode_name)
if not mode_def:
Expand All @@ -515,31 +512,31 @@ async def _handle_mode(self, args: str) -> str:
if explicit_state == "on":
if current_mode == mode_name:
return f"Already in {mode_name} mode"
session_state["active_mode"] = mode_name
mode_hooks = session_state.get("mode_hooks")
self.session.coordinator.register_capability("modes.active_mode", mode_name)
mode_hooks = self.session.coordinator.get_capability("modes.hooks")
if mode_hooks and hasattr(mode_hooks, "reset_warnings"):
mode_hooks.reset_warnings()
return f"Mode: {mode_name}" + (f" — {description}" if description else "")

if explicit_state == "off":
if current_mode != mode_name:
return f"Not in {mode_name} mode"
session_state["active_mode"] = None
mode_hooks = session_state.get("mode_hooks")
self.session.coordinator.register_capability("modes.active_mode", None)
mode_hooks = self.session.coordinator.get_capability("modes.hooks")
if mode_hooks and hasattr(mode_hooks, "reset_warnings"):
mode_hooks.reset_warnings()
return f"Mode off: {mode_name}"

# Toggle behavior (no explicit on/off)
if current_mode == mode_name:
session_state["active_mode"] = None
mode_hooks = session_state.get("mode_hooks")
self.session.coordinator.register_capability("modes.active_mode", None)
mode_hooks = self.session.coordinator.get_capability("modes.hooks")
if mode_hooks and hasattr(mode_hooks, "reset_warnings"):
mode_hooks.reset_warnings()
return f"Mode off: {mode_name}"
else:
session_state["active_mode"] = mode_name
mode_hooks = session_state.get("mode_hooks")
self.session.coordinator.register_capability("modes.active_mode", mode_name)
mode_hooks = self.session.coordinator.get_capability("modes.hooks")
if mode_hooks and hasattr(mode_hooks, "reset_warnings"):
mode_hooks.reset_warnings()
return f"Mode: {mode_name}" + (f" — {description}" if description else "")
Expand All @@ -548,8 +545,7 @@ async def _list_modes(self) -> str:
"""List available modes, grouped by source bundle."""
from collections import defaultdict

session_state = self.session.coordinator.session_state
discovery = session_state.get("mode_discovery")
discovery = self.session.coordinator.get_capability("modes.discovery")

if not discovery:
return (
Expand All @@ -560,7 +556,7 @@ async def _list_modes(self) -> str:
if not modes:
return "No modes found. Create modes in .amplifier/modes/ or include a bundle with modes."

current_mode = session_state.get("active_mode")
current_mode = self.session.coordinator.get_capability("modes.active_mode")

# Group by source (supports both 2-tuple and 3-tuple formats)
groups: dict[str, list[tuple[str, str]]] = defaultdict(list)
Expand Down Expand Up @@ -668,7 +664,7 @@ async def _get_status(self) -> str:
lines.append(f" Config: {self.bundle_name}")

# Active mode status
active_mode = self.session.coordinator.session_state.get("active_mode")
active_mode = self.session.coordinator.get_capability("modes.active_mode")
lines.append(f" Mode: {active_mode or 'none'}")

# Context size
Expand Down Expand Up @@ -847,8 +843,7 @@ def _format_help(self) -> str:
lines.append(f" {cmd:<12} - {info['description']}")

# Add dynamic modes section if modes are available
session_state = self.session.coordinator.session_state
discovery = session_state.get("mode_discovery")
discovery = self.session.coordinator.get_capability("modes.discovery")
if discovery:
modes = discovery.list_modes()
if modes:
Expand Down Expand Up @@ -1470,8 +1465,8 @@ async def interactive_chat(

# Create prompt session for history and advanced editing
prompt_session = _create_prompt_session(
get_active_mode=lambda: command_processor.session.coordinator.session_state.get(
"active_mode"
get_active_mode=lambda: command_processor.session.coordinator.get_capability(
"modes.active_mode"
)
)

Expand Down
Loading