Simplify: dedupe small helpers across the codebase

aai_cli/agent_cascade/tui.py

@@ -40,6 +40,19 @@
 _STATUS_LINE = "Esc/Ctrl-C to interrupt · Ctrl-Q to quit"
 
 
+def _call_on_ui_thread(app: App[None], fn: Callable[..., None], *args: object) -> None:
+    """Hop ``fn`` onto ``app``'s UI thread, dropping the error a torn-down app raises mid-call.
+
+    The cascade runs on a worker thread, so every render/teardown call crosses back via
+    ``call_from_thread``; once the app has stopped that raises ``RuntimeError`` and the call
+    is moot, so it's suppressed rather than surfaced as an unhandled worker-thread exception.
+    """
+    if not app.is_running:
+        return
+    with contextlib.suppress(RuntimeError):
+        app.call_from_thread(fn, *args)
+
+
 class _TuiRenderer:
     """Marshals cascade :class:`~aai_cli.agent_cascade.engine.Renderer` calls onto the UI thread.
 
@@ -76,10 +89,7 @@ def reply_done(self, *, interrupted: bool) -> None:
         self._dispatch(lambda: self._app.end_reply(interrupted=interrupted))
 
     def _dispatch(self, fn: Callable[..., None], *args: object) -> None:
-        if not self._app.is_running:
-            return
-        with contextlib.suppress(RuntimeError):
-            self._app.call_from_thread(fn, *args)
+        _call_on_ui_thread(self._app, fn, *args)
 
 
 class LiveAgentApp(App[None]):
@@ -172,11 +182,8 @@ def _run(self) -> None:
         self._safely(self.exit)
 
     def _safely(self, fn: Callable[..., None], *args: object) -> None:
-        """Hop ``fn`` onto the UI thread, dropping the error a torn-down app raises mid-call."""
-        if not self.is_running:
-            return
-        with contextlib.suppress(RuntimeError):
-            self.call_from_thread(fn, *args)
+        """Hop ``fn`` onto this app's UI thread (see :func:`_call_on_ui_thread`)."""
+        _call_on_ui_thread(self, fn, *args)
 
     # --- transcript (always called on the UI thread) --------------------------
 

aai_cli/app/mediafile.py

@@ -106,6 +106,16 @@ def default_output(
     return Path.cwd() / chosen.name if downloaded else chosen
 
 
+def existing_output(source: str, namer: Callable[[Path], Path]) -> Path | None:
+    """The default output for a local ``source`` when it already exists (so a batch
+    run skips it), else ``None`` — a URL (the output name isn't known until download)
+    or a source with no prior output, both of which are processed."""
+    if "://" in source:
+        return None
+    out = namer(Path(source))
+    return out if out.exists() else None
+
+
 def validate_out(out: Path, media: Path) -> None:
     """An unwritable or self-overwriting output file must fail here, before the
     billed transcription/translation/synthesis pipeline runs.
@@ -225,6 +235,11 @@ def resolve_diarized_transcript(
     )
 
 
+def transcript_id(transcript: object) -> str:
+    """The transcript's id as a safe string (``""`` when the SDK object carries none)."""
+    return str(getattr(transcript, "id", ""))
+
+
 def _fetched_transcript(api_key: str, transcript_id: str) -> object:
     """A --transcript-id transcript, rejected unless it finished successfully —
     a queued/processing/errored one would otherwise surface much later as a

aai_cli/code_agent/_config_root.py

@@ -0,0 +1,24 @@
+"""The coding-agent config root, shared by the skills and memory backends.
+
+`assembly setup` and the agent's middleware both anchor their on-disk state under
+the coding-agent config root (`$CLAUDE_CONFIG_DIR` or `~/.claude`). Skills and
+long-term memory each root their own `FilesystemBackend` there, so the resolution
+lives here once rather than being duplicated per backend.
+
+Mirrors `aai_cli.app.coding_agent.skills_root`'s root resolution without importing
+the app layer (a feature slice stays below it).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from aai_cli.core import env
+
+_CLAUDE_CONFIG_DIR = "CLAUDE_CONFIG_DIR"
+
+
+def claude_config_root() -> Path:
+    """The coding-agent config root: ``$CLAUDE_CONFIG_DIR`` if set, else ``~/.claude``."""
+    config_dir = env.get(_CLAUDE_CONFIG_DIR)
+    return Path(config_dir) if config_dir else Path.home() / ".claude"

aai_cli/code_agent/memory.py

@@ -12,19 +12,15 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from aai_cli.core import env
+from aai_cli.code_agent._config_root import claude_config_root
 
 if TYPE_CHECKING:
     from deepagents.middleware.memory import MemoryMiddleware
 
-_CLAUDE_CONFIG_DIR = "CLAUDE_CONFIG_DIR"
-
 
 def memory_root() -> Path:
     """Directory where the agent's long-term memory files live (created on demand)."""
-    config_dir = env.get(_CLAUDE_CONFIG_DIR)
-    root = Path(config_dir) if config_dir else Path.home() / ".claude"
-    return root / "code-memory"
+    return claude_config_root() / "code-memory"
 
 
 # The single memory file the agent reads and appends learnings to. MemoryMiddleware

aai_cli/code_agent/render.py

@@ -7,13 +7,10 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
-
 from rich.markdown import Markdown
 from rich.markup import escape
 
 from aai_cli.code_agent.events import AssistantText, ErrorText, Event, ToolCall, ToolResult
-from aai_cli.code_agent.session import Approver
 from aai_cli.code_agent.summarize import summarize_call, summarize_result
 from aai_cli.ui import output
 
@@ -42,16 +39,3 @@ def __call__(self, event: Event) -> None:
     def notice(self, text: str) -> None:
         """A dim advisory on stderr (so it never pollutes piped stdout)."""
         output.error_console.print(output.hint(text))
-
-
-def make_approver(confirm: Callable[[str, dict[str, object]], bool]) -> Approver:
-    """Build the approval callback from a yes/no ``confirm`` prompt.
-
-    Kept thin so the interactive confirm (stdin read) is injected by the command and
-    a test passes a scripted decision.
-    """
-
-    def approver(name: str, args: dict[str, object]) -> bool:
-        return confirm(name, args)
-
-    return approver

aai_cli/code_agent/skills.py

@@ -18,16 +18,12 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from aai_cli.core import env
+from aai_cli.code_agent._config_root import claude_config_root
 
 if TYPE_CHECKING:
     from langchain.agents.middleware import AgentMiddleware
     from langchain_core.tools import BaseTool
 
-# Mirrors aai_cli.app.coding_agent.skills_root without importing the app layer (a
-# feature slice stays below it): the agent config root, overridable for tests/agents.
-_CLAUDE_CONFIG_DIR = "CLAUDE_CONFIG_DIR"
-
 READ_SKILL_TOOL_NAME = "read_skill"
 
 # Skills prompt fragment. Must keep the three slots deepagents substitutes at runtime
@@ -53,9 +49,7 @@
 
 def skills_root() -> Path:
     """Directory holding installed skills (one subdir per skill, each with SKILL.md)."""
-    config_dir = env.get(_CLAUDE_CONFIG_DIR)
-    root = Path(config_dir) if config_dir else Path.home() / ".claude"
-    return root / "skills"
+    return claude_config_root() / "skills"
 
 
 def _has_skills(root: Path) -> bool:

aai_cli/commands/account.py

@@ -153,7 +153,7 @@ def body(state: AppState, json_mode: bool) -> None:
         cents = jsonshape.as_float(data.get("balance_in_cents"))
         output.emit(
             data,
-            lambda _d: f"Balance: [aai.success]${cents / 100:,.2f}[/aai.success]",
+            lambda _d: f"Balance: [aai.success]{_format_dollars(cents)}[/aai.success]",
             json_mode=json_mode,
             fields=fields,
         )

aai_cli/commands/caption/_exec.py

@@ -121,7 +121,7 @@ def _fetch_srt(transcript: object, opts: CaptionOptions, *, json_mode: bool, qui
             transcript, "srt", chars_per_caption=opts.chars_per_caption
         )
     if not srt.strip():
-        transcript_id = str(getattr(transcript, "id", ""))
+        transcript_id = mediafile.transcript_id(transcript)
         raise CLIError(
             f"Transcript {transcript_id} has no captions to burn in.",
             error_type="no_captions",
@@ -179,7 +179,10 @@ def _caption_worker(
     quiet_state = dataclasses.replace(state, quiet=True)
 
     def worker(source: str) -> batch.SourceResult:
-        if not force and (existing := _existing_output(source)) is not None:
+        if (
+            not force
+            and (existing := mediafile.existing_output(source, default_out_path)) is not None
+        ):
             return batch.SourceResult(
                 payload={"source": source, "out": str(existing)},
                 summary=f"{existing} exists",
@@ -192,16 +195,6 @@ def worker(source: str) -> batch.SourceResult:
     return worker
 
 
-def _existing_output(source: str) -> Path | None:
-    """The default output for a local ``source`` when it already exists (so batch mode
-    skips it), else ``None`` — a URL (output name isn't known until download) or a
-    source with no prior output, both of which are processed."""
-    if "://" in source:
-        return None
-    out = default_out_path(Path(source))
-    return out if out.exists() else None
-
-
 def _caption_one(opts: CaptionOptions, state: AppState, *, json_mode: bool) -> batch.SourceResult:
     """Resolve ``opts.media`` to a local video, caption it, and return the result.
 
@@ -247,7 +240,7 @@ def _caption_build(
         quiet=state.quiet,
         config=aai.TranscriptionConfig(),
     )
-    transcript_id = str(getattr(transcript, "id", ""))
+    transcript_id = mediafile.transcript_id(transcript)
     srt = _fetch_srt(transcript, opts, json_mode=json_mode, quiet=state.quiet)
     captions = srt.count("-->")  # one arrow per SRT cue timing line
     with tempfile.TemporaryDirectory(prefix="aai-caption-") as tmp:

aai_cli/commands/clip/_exec.py

@@ -167,7 +167,7 @@ def _transcript_segments(
     if not _needs_transcript(opts):
         return [], None
     transcript = _resolve_transcript(opts, media, state, json_mode=json_mode)
-    transcript_id = str(getattr(transcript, "id", ""))
+    transcript_id = mediafile.transcript_id(transcript)
     utterances = jsonshape.object_list(getattr(transcript, "utterances", None))
     if not utterances:
         raise CLIError(

aai_cli/commands/code/_exec.py

@@ -29,7 +29,7 @@
 from aai_cli.code_agent.memory import build_memory_middleware
 from aai_cli.code_agent.model import build_model
 from aai_cli.code_agent.prompt import DEFAULT_MODEL
-from aai_cli.code_agent.render import RichRenderer, make_approver
+from aai_cli.code_agent.render import RichRenderer
 from aai_cli.code_agent.session import CodeSession, EventSink, run_repl
 from aai_cli.code_agent.skills import build_skills
 from aai_cli.code_agent.store import build_checkpointer
@@ -192,7 +192,7 @@ def _run_repl(agent: CompiledAgent, opts: CodeOptions, bridge: AskBridge) -> Non
     session = CodeSession(
         agent=agent,
         sink=RichRenderer(),
-        approver=make_approver(_confirm),
+        approver=_confirm,
         thread_id=opts.session,
         auto_approve=opts.auto,
     )
@@ -260,7 +260,7 @@ def _run_voice(agent: CompiledAgent, opts: CodeOptions, bridge: AskBridge, api_k
     session = CodeSession(
         agent=agent,
         sink=_voice_sink(renderer, voice),
-        approver=make_approver(_confirm),
+        approver=_confirm,
         thread_id=opts.session,
         auto_approve=opts.auto,
     )

aai_cli/commands/dub/_exec.py

@@ -162,8 +162,11 @@ def _dub_worker(
     already exists (unless ``--force``), else dub it with spinners silenced."""
     quiet_state = dataclasses.replace(state, quiet=True)
 
+    def namer(media: Path) -> Path:
+        return default_out_path(media, language)
+
     def worker(source: str) -> batch.SourceResult:
-        if not force and (existing := _existing_output(source, language)) is not None:
+        if not force and (existing := mediafile.existing_output(source, namer)) is not None:
             return batch.SourceResult(
                 payload={"source": source, "out": str(existing)},
                 summary=f"{existing} exists",
@@ -180,15 +183,6 @@ def worker(source: str) -> batch.SourceResult:
     return worker
 
 
-def _existing_output(source: str, language: str) -> Path | None:
-    """The default output for a local ``source`` when it already exists (so batch mode
-    skips it), else ``None`` — a URL or a source with no prior output, both processed."""
-    if "://" in source:
-        return None
-    out = default_out_path(Path(source), language)
-    return out if out.exists() else None
-
-
 def _dub_one(
     opts: DubOptions,
     state: AppState,
@@ -254,7 +248,7 @@ def _dub_build(
         # transcription auto-detects the source language unless --source-lang pins it.
         detect_language=opts.source_language is None,
     )
-    transcript_id = str(getattr(transcript, "id", ""))
+    transcript_id = mediafile.transcript_id(transcript)
     utterances = pipeline.utterances_of(transcript, transcript_id)
     translations = pipeline.translate(
         api_key, utterances, language, opts, json_mode=json_mode, quiet=state.quiet

aai_cli/core/config.py

@@ -84,9 +84,14 @@ def keyring_usable() -> bool:
     return keyring_store.usable()
 
 
+def _profile(profile: str) -> Profile | None:
+    """The stored profile record, or None if the profile has none yet."""
+    return config_store.load().profiles.get(profile)
+
+
 def get_profile_env(profile: str) -> str | None:
     """The backend environment recorded for a profile, if any (e.g. 'sandbox000')."""
-    prof = config_store.load().profiles.get(profile)
+    prof = _profile(profile)
     return prof.env if prof else None
 
 
@@ -99,7 +104,7 @@ def set_profile_env(profile: str, env: str) -> None:
 
 def get_profile_email(profile: str) -> str | None:
     """The login email recorded for a profile at browser login, if any."""
-    prof = config_store.load().profiles.get(profile)
+    prof = _profile(profile)
     return prof.email if prof else None
 
 
@@ -150,7 +155,7 @@ def get_session(profile: str) -> dict[str, str] | None:
 
 def get_account_id(profile: str) -> int | None:
     """The AMS account id recorded at login for a profile, if any."""
-    prof = config_store.load().profiles.get(profile)
+    prof = _profile(profile)
     return prof.account_id if prof else None
 
 

tests/test_caption_batch.py

@@ -14,7 +14,7 @@
 
 import pytest
 
-from aai_cli.app import batch
+from aai_cli.app import batch, mediafile
 from aai_cli.app.context import AppState
 from aai_cli.commands.caption import _exec as caption_exec
 from aai_cli.core import stdio
@@ -125,20 +125,21 @@ def test_batch_rejects_transcript_id(monkeypatch):
     assert "can't apply to many sources" in (exc.value.suggestion or "")
 
 
-# --- _existing_output --------------------------------------------------------
+# --- skip-on-existing-output (caption's namer through mediafile.existing_output) ---
 
 
 def test_existing_output_is_none_for_a_url():
-    assert caption_exec._existing_output("https://youtu.be/x") is None
+    assert mediafile.existing_output("https://youtu.be/x", caption_exec.default_out_path) is None
 
 
 def test_existing_output_is_none_when_missing(tmp_path):
-    assert caption_exec._existing_output(str(tmp_path / "a.mp4")) is None
+    src = str(tmp_path / "a.mp4")
+    assert mediafile.existing_output(src, caption_exec.default_out_path) is None
 
 
 def test_existing_output_returns_the_path_when_present(tmp_path):
     src = tmp_path / "a.mp4"
     src.write_bytes(b"x")
     out = tmp_path / "a.captioned.mp4"
     out.write_bytes(b"old")
-    assert caption_exec._existing_output(str(src)) == out
+    assert mediafile.existing_output(str(src), caption_exec.default_out_path) == out

tests/test_code_agent.py

@@ -28,7 +28,7 @@
 from aai_cli.code_agent.agent import MUTATING_TOOLS, build_agent
 from aai_cli.code_agent.events import AssistantText, ErrorText, ToolCall, ToolResult
 from aai_cli.code_agent.prompt import build_system_prompt
-from aai_cli.code_agent.render import RichRenderer, make_approver
+from aai_cli.code_agent.render import RichRenderer
 from aai_cli.code_agent.session import QUIT_COMMANDS, CodeSession, run_repl
 
 
@@ -273,8 +273,6 @@ def test_rich_renderer_smoke(capsys: pytest.CaptureFixture[str]) -> None:
     renderer(AssistantText("hi"))
     renderer(ToolCall(name="write_file", args={"file_path": "a"}))
     renderer(ToolResult(name="write_file", content="Updated a"))
-    approver = make_approver(lambda name, args: True)
-    assert approver("write_file", {}) is True
     out = capsys.readouterr().out
     assert "hi" in out and "write_file" in out