From 2904e08f347d4ad3230a903f24ac2add99791d92 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Sat, 19 Sep 2026 17:08:54 -0700 Subject: [PATCH] Polish interactive session takeover prompts and notices --- amplifier_app_cli/main.py | 3 +- amplifier_app_cli/session_handoff.py | 154 +++++++++++++++++++------ amplifier_app_cli/shared_root_state.py | 1 + tests/test_session_handoff.py | 45 ++++++-- 4 files changed, 159 insertions(+), 44 deletions(-) diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index a85a436e..1ebd5561 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -3852,7 +3852,8 @@ async def interactive_chat( try: initialized = await create_initialized_session(session_config, console) except SharedRootSessionBusyError as exc: - console.print(f"[red]Error:[/red] {escape_markup(exc)}") + if not exc.displayed: + console.print(f"[red]Error:[/red] {escape_markup(exc)}") raise SystemExit(1) from None session = initialized.session actual_session_id = initialized.session_id diff --git a/amplifier_app_cli/session_handoff.py b/amplifier_app_cli/session_handoff.py index 1ed05820..e73ff976 100644 --- a/amplifier_app_cli/session_handoff.py +++ b/amplifier_app_cli/session_handoff.py @@ -6,8 +6,62 @@ import sys import time import uuid +from contextlib import nullcontext import click +from rich.panel import Panel +from rich.prompt import Confirm +from rich.text import Text + + +def _app_label(value): + from .shared_root_state import _bounded_value + + label = _bounded_value(value) + return { + "amplifier-cli": "Amplifier CLI", + "amplifier-unified": "Amplifier Unified", + }.get(label, label or "another app") + + +def _notice(console, title, content, style="cyan"): + console.print() + console.print( + Panel( + content, + title=Text(title, style=f"bold {style}"), + border_style=style, + padding=(1, 2), + ) + ) + console.print() + + +def _show_owner(console, owner): + from .shared_root_state import _bounded_value + + owner = owner or {} + content = Text("This session is open in ") + content.append(_app_label(owner.get("app")), style="bold cyan") + host = _bounded_value(owner.get("hostname")) + if host: + content.append(f" on {host}", style="dim") + content.append( + ".\n\nRequest takeover to ask that app to save and close this session, " + ) + content.append("then continue here in the CLI.") + _notice(console, "Session already open", content) + + +def _show_takeover_failure(console, status, same_owner): + message = { + "unsupported": "That app does not support takeover requests. Close the session there, then try again.", + "timed_out": "The other app has not finished releasing the session. It may still finish; try again shortly.", + "cannot_release": "The other app could not finish saving and closing this session. Check that app, then try again.", + }.get(status, "The session is still in use. Check the other app, then try again.") + if not same_owner: + message = "The session owner changed while waiting. Try again to request takeover from the current app." + _notice(console, "Could not take over", Text(message), "yellow") def _takeover_option(ctx, param, value): @@ -44,42 +98,75 @@ async def acquire_root(config, console): options = context.meta if context else {} requested = config.takeover or options.get("takeover", False) timeout = options.get("handoff_timeout", config.handoff_timeout) - if not requested and config.invocation_mode == "chat" and sys.stdin.isatty(): - console.print(str(busy), markup=False) + interactive = config.invocation_mode == "chat" and sys.stdin.isatty() + if not requested and interactive: + _show_owner(console, busy.owner) requested = await asyncio.to_thread( - click.confirm, "Request takeover?", default=False + Confirm.ask, + "[bold cyan]Request takeover?[/bold cyan]", + console=console, + default=False, ) + if not requested: + console.print("[dim]Session left open in the other app.[/dim]") + busy.displayed = True if not requested or busy.store is None: raise from amplifier_foundation.session import request_release deadline = time.monotonic() + timeout - result = await request_release( - busy.store, - expected_owner=busy.owner, - request_id=uuid.uuid4().hex, - requester_app="Amplifier CLI", - timeout=timeout, + status = ( + console.status("[cyan]Requesting takeover…[/cyan]", spinner="dots") + if interactive + else nullcontext() ) - # The reply may be lost after release. Try the real lock, but never send - # a second request to a newly observed owner. - while True: - try: - return SharedRootSession.acquire(config.session_id) - except SharedRootSessionBusyError as current: - same = (current.owner or {}).get("acquisition_id") == ( - busy.owner or {} - ).get("acquisition_id") - if ( - not same - or result.status not in {"released", "unreachable"} - or time.monotonic() >= deadline - ): - current.args = ( - f"Takeover did not complete ({result.status}). {result.message} {current}", - ) - raise current from None - await asyncio.sleep(min(0.05, max(0, deadline - time.monotonic()))) + with status as progress: + + def on_progress(stage): + message = { + "accepted": "Takeover accepted. Waiting for the other app…", + "draining": "Waiting for active work to stop safely…", + "persisting": "Saving session history…", + }.get(stage, "Waiting for the other app to release the session…") + progress.update(Text(message, style="cyan")) + + result = await request_release( + busy.store, + expected_owner=busy.owner, + request_id=uuid.uuid4().hex, + requester_app="Amplifier CLI", + timeout=timeout, + on_progress=on_progress if interactive else None, + ) + # The reply may be lost after release. Try the real lock, but never + # send a second request to a newly observed owner. + while True: + try: + root = SharedRootSession.acquire(config.session_id) + break + except SharedRootSessionBusyError as current: + same = (current.owner or {}).get("acquisition_id") == ( + busy.owner or {} + ).get("acquisition_id") + if ( + not same + or result.status not in {"released", "unreachable"} + or time.monotonic() >= deadline + ): + current.args = ( + f"Takeover did not complete ({result.status}). {result.message} {current}", + ) + if interactive: + progress.stop() + _show_takeover_failure(console, result.status, same) + current.displayed = True + raise current from None + await asyncio.sleep(min(0.05, max(0, deadline - time.monotonic()))) + if interactive: + console.print( + "[green]✓[/green] [bold]Session acquired.[/bold] [dim]Continuing here in the CLI.[/dim]" + ) + return root class CLIHandoff: @@ -134,7 +221,7 @@ async def prompt(self, factory): async def finish(self, save=None, after_cleanup=None): """Complete all persistence before either normal release or handoff.""" - from amplifier_foundation.session import ReadyToRelease, CannotRelease + from amplifier_foundation.session import CannotRelease, ReadyToRelease if not self.requested.is_set() and self.registration: await self.registration.close() @@ -162,11 +249,12 @@ async def finish(self, save=None, after_cleanup=None): self.prepared.set_result(ReadyToRelease()) await asyncio.shield(self.registration.pending) if not self.root.held.active: - self.console.print( - f"CLI session closed at the request of {self.source}. " - "Session history saved. Execution ownership released.", - markup=False, + content = Text("This CLI session closed at the request of ") + content.append(_app_label(self.source), style="bold cyan") + content.append( + ".\n\nSession history saved. Execution ownership released." ) + _notice(self.console, "Session handed off", content, "green") elif self.root is not None: # Closing first prevents a new callback from entering after cleanup. if self.registration: diff --git a/amplifier_app_cli/shared_root_state.py b/amplifier_app_cli/shared_root_state.py index d70ee2ea..a7a1f69e 100644 --- a/amplifier_app_cli/shared_root_state.py +++ b/amplifier_app_cli/shared_root_state.py @@ -28,6 +28,7 @@ class SharedRootSessionBusyError(RuntimeError): """A shared root is held by another process, with bounded owner advice.""" def __init__(self, owner: object, state_root: object, *, store=None) -> None: + self.displayed = False self.store = store self.owner = owner if isinstance(owner, dict) else None self.state_root = _bounded_value(state_root, limit=240) diff --git a/tests/test_session_handoff.py b/tests/test_session_handoff.py index d663e69a..8d23c19a 100644 --- a/tests/test_session_handoff.py +++ b/tests/test_session_handoff.py @@ -5,13 +5,13 @@ from unittest.mock import AsyncMock, Mock import pytest -from rich.console import Console - from amplifier_foundation.session import ( - SharedSessionStore, SessionBusyError, + SharedSessionStore, request_release, ) +from rich.console import Console + from amplifier_app_cli.session_handoff import CLIHandoff, acquire_root from amplifier_app_cli.session_runner import SessionConfig from amplifier_app_cli.shared_root_state import ( @@ -111,7 +111,7 @@ async def test_save_failure_reports_failure_and_retains_ownership( async def test_single_mode_does_not_request_without_explicit_intent( tmp_path, monkeypatch ): - root, initialized, control, output = fixture(tmp_path, monkeypatch) + _root, _initialized, control, _output = fixture(tmp_path, monkeypatch) await control.start() config = SessionConfig( {}, [], False, session_id="test-session", invocation_mode="single" @@ -123,7 +123,7 @@ async def test_single_mode_does_not_request_without_explicit_intent( async def test_explicit_cli_takeover_acquires_before_returning(tmp_path, monkeypatch): - root, initialized, control, output = fixture(tmp_path, monkeypatch) + root, _initialized, control, _output = fixture(tmp_path, monkeypatch) await control.start() config = SessionConfig( {}, @@ -146,10 +146,10 @@ async def test_explicit_cli_takeover_acquires_before_returning(tmp_path, monkeyp async def test_interactive_busy_session_prompts_before_requesting( tmp_path, monkeypatch, accept ): - root, initialized, control, output = fixture(tmp_path, monkeypatch) + root, _initialized, control, output = fixture(tmp_path, monkeypatch) await control.start() confirm = Mock(return_value=accept) - monkeypatch.setattr("click.confirm", confirm) + monkeypatch.setattr("amplifier_app_cli.session_handoff.Confirm.ask", confirm) monkeypatch.setattr("sys.stdin.isatty", lambda: True) config = SessionConfig( {}, [], False, session_id="test-session", invocation_mode="chat" @@ -161,10 +161,35 @@ async def test_interactive_busy_session_prompts_before_requesting( successor = await asyncio.wait_for(acquisition, 2) successor.release() else: - with pytest.raises(SharedRootSessionBusyError): + with pytest.raises(SharedRootSessionBusyError) as error: await asyncio.wait_for(acquisition, 2) + assert error.value.displayed assert not control.requested.is_set() assert root.held.active await control.finish() - confirm.assert_called_once_with("Request takeover?", default=False) - assert root.held.owner["app"] in output.getvalue() + confirm.assert_called_once() + assert confirm.call_args.kwargs == {"console": control.console, "default": False} + assert "Amplifier CLI" in output.getvalue() + + +async def test_interactive_unsupported_takeover_keeps_owner_and_reports_failure( + tmp_path, monkeypatch +): + root, _initialized, control, output = fixture(tmp_path, monkeypatch) + # A real lock without a registered handoff handler represents an older app. + monkeypatch.setattr( + "amplifier_app_cli.session_handoff.Confirm.ask", lambda *a, **k: True + ) + monkeypatch.setattr("sys.stdin.isatty", lambda: True) + config = SessionConfig( + {}, [], False, session_id="test-session", invocation_mode="chat" + ) + try: + with pytest.raises(SharedRootSessionBusyError) as error: + await acquire_root(config, control.console) + assert error.value.displayed # The startup boundary must not print it again. + assert root.held.active + assert "does not support takeover" in output.getvalue() + assert "Session acquired" not in output.getvalue() + finally: + root.release()