Session follow-ups: run id and verified folders, aliases, declared deps, docs

README.md

@@ -135,14 +135,14 @@ Settings live in `~/.reverse-api/config.json` and can be edited via `/settings`
 
 ## CLI
 
-`reverse-api-engineer` opens a full-screen session; `agent`, `engineer` and `collector` open it too when run in a terminal (without `--json` or `--no-interactive`). Type `/` for the commands:
+`reverse-api-engineer` opens a full-screen session; `agent`, `engineer` and `collector` open it too when run in a terminal (`agent` and `engineer` stay plain with `--json` or `--no-interactive`). Type `/` for the commands:
 - `/model`, `/effort`, `/mode`: switch the model, how hard it thinks, and the mode (also Shift+Tab, ctrl+e).
 - `/settings`: SDK, agent provider, output and sync settings.
 - `/history`: past runs, with their details, messages, and re-engineer or resume.
 - `/messages [run_id]`: what the agent did in a run.
 - `/resume [run_id]`: continue a past run's conversation where it stopped, on any SDK. Works for runs that saved their SDK session: agent and engineer runs from this version on. Collector runs cannot be resumed.
 - `/output` (ctrl+o): everything from the session, unfolded. `/open`: the latest client.
-- `/new`, `/clear`, `/help`, `/exit`.
+- `/new`, `/clear`, `/help` (or `/commands`), `/exit` (or `/quit`).
 
 Scriptable subcommands (pipe to `jq`):
 

pyproject.toml

@@ -47,6 +47,8 @@ dependencies = [
     "click>=8.3.3",
     "claude-agent-sdk>=0.2.174",
     "rich>=13.0.0",
+    "markdown-it-py>=2.2.0",
+    "prompt-toolkit>=3.0.40",
     "questionary>=2.0.0",
     "watchdog>=3.0.0",
     "setproctitle>=1.3.0",

src/reverse_api/base_engineer.py

@@ -97,6 +97,13 @@ def _report_usage(self) -> None:
         if isinstance(self.ui, LedgerUI):
             self.ui.usage(self.usage_metadata)
 
+    def _report_workspace(self) -> None:
+        """Tells the full-screen session where the run's files are, for its verified step."""
+        from .repl.bridge import LedgerUI
+
+        if isinstance(self.ui, LedgerUI):
+            self.ui.workspace(self.scripts_dir, self.local_scripts_dir)
+
     async def _opening_message(self, brief: str) -> str | None:
         """The first message of the run: its brief, or when resuming, the user's next message."""
         message = await self._prompt_follow_up() if self.resume_session else brief

src/reverse_api/cli.py

@@ -1902,6 +1902,7 @@ def engineer(
             lambda: run_engineer(
                 run_id, prompt=main_prompt, additional_instructions=additional, model=model, output_dir=output_dir, is_fresh=fresh
             ),
+            lambda p, m: engineer_from_prompt(p, m or model, output_dir),
         )
         return
     if not machine_output:
@@ -1955,6 +1956,17 @@ def engineer(
     sys.exit(0 if payload["status"] == "ok" else 1)
 
 
+def engineer_from_prompt(text: str, model: str | None = None, output_dir: str | None = None) -> dict[str, Any] | None:
+    """A prompt in engineer mode: a known run id re-engineers that run; anything else is an instruction for the latest run."""
+    if session_manager.get_run(text):
+        return run_engineer(text, model=model, output_dir=output_dir)
+    # Collector runs capture no traffic; run_engineer explains any other run that has none.
+    latest = next((r for r in session_manager.get_history(limit=50) if r.get("mode") != "collector"), None)
+    if latest is None:
+        raise RuntimeError("no captured runs yet · start one in agent or manual mode")
+    return run_engineer(latest["run_id"], model=model, output_dir=output_dir, additional_instructions=text)
+
+
 def run_engineer(
     run_id: str,
     har_path: Path | None = None,

src/reverse_api/engineer.py

@@ -127,6 +127,7 @@ async def report_client_verified(args: dict[str, Any]) -> dict[str, Any]:
                     ]
                 }
             already_reported = True
+            self._report_workspace()
             # Deliberately no manual message_store/UI logging here — the
             # generic ToolUseBlock/ToolResultBlock handling in
             # _process_streaming_response already logs every tool call,

src/reverse_api/repl/app.py

@@ -75,6 +75,7 @@
     "/help": "keys and commands",
     "/exit": "quit rae",
 }
+ALIASES = {"/quit": "/exit", "/commands": "/help"}
 HISTORY_RUNS = 20
 # Commands that open a picker are echoed once a choice shows something, not when the picker opens.
 PICKER_COMMANDS = {"/model", "/effort", "/mode", "/history", "/messages", "/resume"}
@@ -214,7 +215,9 @@ def palette_active(self) -> bool:
 
     def palette_items(self) -> list[tuple[str, str]]:
         typed = self.buf.text.strip().lower()
-        return sorted(((c, d) for c, d in COMMANDS.items() if c.startswith(typed)), key=lambda item: item[0] != typed)
+        aliased = {target for alias, target in ALIASES.items() if alias.startswith(typed)}
+        items = [(c, d) for c, d in COMMANDS.items() if c.startswith(typed) or c in aliased]
+        return sorted(items, key=lambda item: item[0] != ALIASES.get(typed, typed))
 
     def choose(self) -> None:
         """Enter on an open picker: its highlighted item, or for an agent's question, whatever was typed meanwhile."""
@@ -280,6 +283,7 @@ def cycle_effort(self) -> None:
     def run_command(self, text: str) -> None:
         cmd, _, arg = text.partition(" ")
         cmd, arg = cmd.lower(), arg.strip()
+        cmd = ALIASES.get(cmd, cmd)
         st, view = self.st, self.view
         if cmd not in PICKER_COMMANDS | {"/exit", "/clear"}:
             view.command(cmd, arg)
@@ -331,7 +335,7 @@ def run_command(self, text: str) -> None:
                 view.flush_live()
                 self.transcript.clear()
                 st.scroll, st.folded, st.cost, st.tokens, st.step = 0, False, 0.0, 0, 0
-                st.last_client, st.client_name, st.host = None, "", ""
+                st.last_client, st.client_name, st.host, st.run_id = None, "", "", None
                 self.bridge.cost_base = 0.0
         elif cmd == "/settings":
             self.classic(lambda: self.cli.handle_settings(), then=self.reload_settings, pause=False)
@@ -496,7 +500,7 @@ def start(self, prompt: str, mode: str | None = None, echo: bool = True) -> None
         elif mode == "collector":
             run = lambda: cli.run_collector(prompt=prompt, model=model)  # noqa: E731
         elif mode == "engineer":
-            run = lambda: self.engineer(prompt, model)  # noqa: E731
+            run = lambda: cli.engineer_from_prompt(prompt, model)  # noqa: E731
         else:
             run = lambda: cli.run_agent_capture(prompt=prompt, model=model)  # noqa: E731
         self.launch(prompt, run, mode, echo)
@@ -532,17 +536,6 @@ def check() -> None:
 
             threading.Thread(target=check, name="rae-update-check", daemon=True).start()
 
-    def engineer(self, text: str, model: str | None) -> dict[str, Any] | None:
-        """A known run id re-engineers that run; anything else is an instruction for the latest run."""
-        sm = self.cli.session_manager
-        if sm.get_run(text):
-            return self.cli.run_engineer(text, model=model)
-        latest = sm.get_history(limit=1)
-        if not latest:
-            self.view.error("no runs yet · start one in agent or manual mode")
-            return None
-        return self.cli.run_engineer(latest[0]["run_id"], model=model, additional_instructions=text)
-
     def work(self, fn: Callable[[], Any]) -> None:
         """Run an engine in a worker thread; its UI calls stream into the conversation through the bridge."""
 

src/reverse_api/repl/bridge.py

@@ -104,6 +104,7 @@ def __init__(self, view: LedgerView, invalidate: Callable[[], None], ask_user: C
         self.pending_card: Path | None = None
         self.cost_base = 0.0
         self.verified = False  # the agent called report_client_verified in this run
+        self.workspace: tuple[Path, Path | None] | None = None  # the run's folder and its synced copy
 
     def ui(self) -> LedgerUI:
         return LedgerUI(self)
@@ -203,7 +204,7 @@ def header(self, run_id: str, prompt: str, model: str | None = None, sdk: str |
         st.run_id, st.step, st.tokens = run_id, 0, 0
         st.running, st.waiting, st.verb, st.kind = True, False, "starting", "other"
         self.b.cost_base = st.cost
-        self.b.verified = False
+        self.b.verified, self.b.workspace = False, None
         # The run's own model and SDK; the selection in st.model stays for the next run.
         st.run_model, st.run_sdk = model or st.model, sdk or st.sdk
         claude = st.run_sdk == "claude" and not predates_effort(st.run_model)
@@ -250,6 +251,8 @@ def tool_result(self, tool_name: str, is_error: bool = False, output: str | None
         if name in HIDDEN_TOOLS:
             return
         if name == "report_client_verified" and not is_error:
+            if not self.b.verified and self.b.workspace:
+                self.view.workspace(*self.b.workspace)
             self.b.verified = True  # the card promises a client that ran, so only a confirmed report counts
         text = str(output or "").strip()
         lines = text.splitlines()
@@ -305,6 +308,9 @@ def progress(self, message: str) -> None:
         self.view.notice(message)
         self._changed()
 
+    def workspace(self, folder: Path, copy: Path | None) -> None:
+        self.b.workspace = (folder.resolve(), copy.resolve() if copy else None)
+
     def success(self, script_path: str, local_path: str | None = None) -> None:
         _ = local_path
         path = Path(script_path)

src/reverse_api/repl/view.py

@@ -482,6 +482,12 @@ def block(rows: list[str], more: int, cut: bool) -> Group:
         # The full copy is only styled if ctrl+o asks for it: tool output can run to thousands of lines.
         self.out(block(lines[:OUTPUT_PREVIEW], more, cut=True), Deferred(lambda: block(lines, 0, cut=False)))
 
+    def workspace(self, folder: Path, copy: Path | None) -> None:
+        """Where a verified client's files are, absolute so a click opens them."""
+        rows = [("in", folder), ("copy", copy)] if copy else [("in", folder)]
+        for label, path in rows:
+            self.out(f"{' ' * INDENT}[{HAIRLINE}]└[/] [{DIM}]{label}[/] [{INK_SOFT}]{link(str(path), path)}[/]")
+
     def code(self, source: str, language: str) -> None:
         """Written code, previewed from its first definition; ctrl+o shows all of it."""
         total = source.count("\n") + 1
@@ -684,6 +690,9 @@ def status(self) -> FormattedText:
         else:
             hint = [(f"fg:{DIM}", "/ commands" + (" · ⌃o expand " if st.folded else " "))]
         used = sum(len(t) for _, t in left) + sum(len(t) for _, t in hint)
+        run = f"· run {st.run_id} " if st.run_id else ""
+        if run and used + 1 + len(run) <= width():  # left out rather than cut off
+            hint, used = hint + [(f"fg:{DIM}", run)], used + len(run)
         return FormattedText(left + [("", " " * max(1, width() - used))] + hint)
 
     def menu(self, rows: list[tuple[str, str, bool, bool]], title: str | None = None, limit: int = MENU_ROWS) -> FormattedText:

tests/test_engineer.py

@@ -909,6 +909,17 @@ async def test_tool_emits_client_executed_event_when_called(self, tmp_path):
         assert len(client_executed) == 1
         assert client_executed[0]["script_path"] == str(eng.scripts_dir / "api_client.py")
 
+    @pytest.mark.asyncio
+    async def test_tool_tells_the_session_where_the_files_are_once(self, tmp_path):
+        eng = self._make_engineer(tmp_path, output_language="python")
+        eng._report_workspace = MagicMock()
+
+        tool = eng._build_verification_tool()
+        await tool.handler({"summary": "first confirmation"})
+        await tool.handler({"summary": "second confirmation"})
+
+        eng._report_workspace.assert_called_once_with()
+
     @pytest.mark.asyncio
     async def test_tool_second_call_in_same_session_does_not_re_emit(self, tmp_path):
         """The tool's own description tells the agent to call it exactly

tests/test_repl.py

@@ -5,7 +5,7 @@
 import threading
 from concurrent.futures import Future
 from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
 from rich.console import Console
@@ -120,6 +120,43 @@ def test_no_card_until_the_agent_verified_the_client(self, session, tmp_path):
         ui.usage({"input_tokens": 10, "output_tokens": 5, "estimated_cost_usd": 0.01})
         assert "ready" not in plain(t), "a client path that does not exist gets no card"
 
+    def test_verified_shows_where_the_files_are(self, session):
+        _st, t, _view, b = session
+        ui = LedgerUI(b)
+        ui.header("abc123", "prompt", model="claude-sonnet-5-5")
+        folder, copy = Path("/runs/abc123"), Path("/work/scripts/jobs")
+        ui.workspace(folder, copy)
+        for _ in range(2):
+            ui.tool_start("mcp__verification__report_client_verified", {"summary": "ok"})
+            ui.tool_result("mcp__verification__report_client_verified", output="Recorded")
+        text = plain(t)
+        assert text.count(f"└ in {folder.resolve()}") == 1 and text.count(f"└ copy {copy.resolve()}") == 1
+
+    def test_no_folders_without_a_confirmed_verification(self, session, tmp_path):
+        _st, t, _view, b = session
+        ui = LedgerUI(b)
+        ui.header("abc123", "prompt", model="claude-sonnet-5-5")
+        ui.workspace(tmp_path / "runs" / "abc123", None)
+        ui.tool_result("mcp__verification__report_client_verified", is_error=True, output="failed")
+        assert "└ in" not in plain(t)
+        ui.header("def456", "prompt", model="claude-sonnet-5-5")
+        ui.tool_result("mcp__verification__report_client_verified", output="Recorded")
+        assert "└ in" not in plain(t), "a new run forgets the last run's folder"
+
+    def test_the_status_row_shows_the_run_id(self, session):
+        _st, _t, view, b = session
+        assert "run " not in "".join(text for _, text in view.status())
+        LedgerUI(b).header("abc123", "prompt", model="claude-sonnet-5-5")
+        assert "run abc123" in "".join(text for _, text in view.status())
+
+    def test_a_run_id_that_does_not_fit_is_left_out(self, session, monkeypatch):
+        from reverse_api.repl import view as view_module
+
+        _st, _t, view, b = session
+        LedgerUI(b).header("abc123", "prompt", model="claude-sonnet-5-5")
+        monkeypatch.setattr(view_module, "width", lambda: 40)
+        assert "run abc123" not in "".join(text for _, text in view.status())
+
     def test_folded_output_offers_ctrl_o(self, session):
         st, _t, view, b = session
         assert "expand" not in "".join(text for _, text in view.status())
@@ -417,6 +454,22 @@ def test_commands_are_echoed(self, app_session):
         session.run_command("/help")
         assert "› /help" in plain(session.transcript)
 
+    def test_quit_and_commands_are_aliases(self, app_session, monkeypatch):
+        session, _history = app_session
+        quits = []
+        monkeypatch.setattr(session, "quit", lambda: quits.append(True))
+        session.run_command("/quit")
+        assert quits == [True]
+        session.run_command("/commands")
+        assert "› /help" in plain(session.transcript) and "unknown command" not in plain(session.transcript)
+
+    def test_the_menu_finds_commands_by_alias(self, app_session):
+        session, _history = app_session
+        session.buf.text = "/qu"
+        assert [c for c, _ in session.palette_items()] == ["/exit"]
+        session.buf.text = "/c"
+        assert {c for c, _ in session.palette_items()} == {"/clear", "/help"}
+
     def test_an_exact_command_wins_over_longer_ones(self, app_session):
         session, _history = app_session
         session.buf.text = "/mode"
@@ -434,6 +487,12 @@ def test_pickers_echo_once_something_is_picked(self, app_session):
         pick(session.st, "abc123")
         assert "› /history abc123" in plain(session.transcript)
 
+    def test_clear_forgets_the_run_id(self, app_session):
+        session, _history = app_session
+        session.st.run_id = "abc123"
+        session.run_command("/clear")
+        assert session.st.run_id is None
+
     def test_clear_starts_over_with_no_usage(self, app_session):
         session, _history = app_session
         st = session.st
@@ -870,6 +929,34 @@ def test_agent_in_a_terminal_opens_the_session_with_every_option(self, monkeypat
         }
         assert capture.call_args_list[1].kwargs == {"prompt": "later task", "model": "claude-opus-5-5", "output_dir": None, "headless": True}
 
+    def test_engineer_keeps_its_output_dir_for_later_prompts(self, monkeypatch):
+        from click.testing import CliRunner
+
+        from reverse_api import cli
+
+        monkeypatch.setattr(cli, "_in_terminal", lambda: True)
+        with patch.object(cli, "open_session") as session, patch.object(cli, "engineer_from_prompt") as later:
+            CliRunner().invoke(cli.main, ["engineer", "abc123", "-o", "/out"])
+            runner = session.call_args.args[4]
+            runner("add pagination", None)
+        later.assert_called_once_with("add pagination", None, "/out")
+
+    def test_an_engineer_prompt_is_a_run_id_or_an_instruction_for_the_latest_capture(self, monkeypatch):
+        from reverse_api import cli
+
+        history = MagicMock()
+        history.get_run.side_effect = lambda text: {"run_id": text} if text == "abc123" else None
+        history.get_history.return_value = [{"run_id": "collect1", "mode": "collector"}, {"run_id": "capture1", "mode": "agent"}]
+        monkeypatch.setattr(cli, "session_manager", history)
+        with patch.object(cli, "run_engineer") as engineer:
+            cli.engineer_from_prompt("abc123", "m", "/out")
+            cli.engineer_from_prompt("add pagination", "m", "/out")
+        assert engineer.call_args_list[0] == call("abc123", model="m", output_dir="/out")
+        assert engineer.call_args_list[1] == call("capture1", model="m", output_dir="/out", additional_instructions="add pagination")
+        history.get_history.return_value = history.get_history.return_value[:1]
+        with pytest.raises(RuntimeError, match="no captured runs yet"):
+            cli.engineer_from_prompt("add pagination")
+
     def test_without_a_terminal_a_prompt_is_required(self, monkeypatch):
         from click.testing import CliRunner