Render asynchronous operations with Rich Live and spinners

README.md

@@ -29,7 +29,10 @@ only makes source-code changes immediately available.
 
 Run `.venv/bin/ancestry` with no arguments for the prompt-toolkit/Rich
 interactive console. It is the only supported interactive console; the
-canonical command reference, examples, offline defaults, and privacy rules are
+prompt stays responsive while background operations render live, sanitized
+spinner or completed-unit progress above it. One-shot and JSON output never
+emit terminal animation. The canonical command reference, examples, offline
+defaults, and privacy rules are
 in [the CLI guide](docs/CLI.md); see [the console guide](docs/CONSOLE.md) for
 interactive use.
 

docs/CONSOLE.md

@@ -35,7 +35,8 @@ The session controls are:
 - `run ACTION ...` runs an action using the saved options. Direct module
   actions may also be entered at the root prompt, such as `providers list`.
 - `jobs` or `jobs list` shows background work; `jobs show JOB_ID` shows one
-  job's timestamps, result, or stable failure code.
+  job's timestamps, latest structured progress, result, or stable failure
+  code. `help jobs` summarizes these controls.
 - `exit`, `quit`, or EOF leaves the REPL.
 
 Long-running RootsMagic queries/exports, GEDCOM operations, OCR extraction, and
@@ -47,6 +48,19 @@ manifest, or database resource are serialized; independent targets may run in
 parallel. The queue rejects new work with `JOB_QUEUE_FULL` at its configured
 64-job safety limit.
 
+Active jobs render above the prompt through Rich `Live` while prompt-toolkit's
+supported stdout patch keeps asynchronous updates from overwriting input.
+Unknown-duration operations show a spinner and current operation. Structured
+events with completed/total units render a determinate progress bar. Completion,
+failure, and cancellation stop the live display, print one final sanitized
+status line, and leave a clean prompt; full results remain available through
+`jobs show JOB_ID`.
+
+Progress operation text passes through the same secret-redaction boundary as
+job failures. Live rendering is interactive-only: one-shot commands and
+`--json` output preserve their existing non-animated, machine-readable
+contracts.
+
 ## Multiline free text
 
 In the interactive console, omit `--question` from `rootsmagic query` or omit

docs/MODULE_AUTHORING.md

@@ -9,9 +9,17 @@ both one-shot CLI execution and the prompt-toolkit/Rich REPL.
 Module authors should add or update the command specification first, then wire
 the action to a thin dispatcher that delegates to an application service. The
 dispatch layer must not open storage directly, read secrets, call providers, or
-implement business rules. Services return serializable DTOs, progress events,
-or stable `AncestryError` instances so terminal, JSON, and future adapters can
-present the same result contract.
+implement business rules. Services return serializable DTOs or stable
+`AncestryError` instances so terminal, JSON, and future adapters can present the
+same result contract.
+
+Long-running REPL actions are submitted through the UI-independent job manager.
+Its `JobReporter` accepts a current operation and, when known, validated
+completed/total units. The latest redacted progress event becomes part of the
+serializable job snapshot; the Rich presentation adapter decides whether to
+render it as a spinner or progress bar. Module and service code must never
+construct Rich objects, write terminal control sequences, or depend on
+prompt-toolkit. One-shot dispatch does not create a reporter or emit animation.
 
 Interactive behavior is derived from the shared metadata:
 

docs/REPL_ARCHITECTURE.md

@@ -47,9 +47,13 @@ flowchart LR
    policy, immutable source handling, and provider `none` offline behavior.
    They do not import UI libraries.
 5. **Background jobs** run long operations in bounded worker threads, expose
-   serializable lifecycle snapshots, and serialize mutations by target resource.
-6. **Presentation** renders DTOs, progress, and coded errors. Rich objects
-   stay in this layer; JSON is a serialization of the same result contract.
+   serializable lifecycle snapshots and validated progress events, notify
+   presentation subscribers outside execution locks, and serialize mutations
+   by target resource.
+6. **Presentation** maps structured job progress to Rich Live spinners or
+   determinate bars and renders DTOs and coded errors. prompt-toolkit stdout
+   patching coordinates asynchronous output with the active prompt. Rich
+   objects stay in this layer; JSON is a serialization of the same contracts.
 
 The dependency direction is one-way: `input -> routing -> execution ->
 services`. Presentation consumes execution results and is not a service
@@ -100,6 +104,7 @@ dependency.
 | Context-aware completion | Implemented | Static/snapshot-driven, privacy-filtered, CWD-bounded |
 | Secure history and no-echo secret entry | Implemented | Owner-only history; secrets excluded and redacted |
 | Bounded background jobs | Implemented | Serializable states/results; per-resource mutation serialization |
+| Rich Live job progress | Implemented | Spinner or determinate units; stdout-patched prompt coordination |
 | Cooperative cancellation | Future work | Extends the job lifecycle around atomic sections and shutdown |
 
 ## Compatibility paths

src/ancestryllm/console/progress.py

@@ -0,0 +1,107 @@
+"""Rich Live rendering for structured background-job progress."""
+
+from __future__ import annotations
+
+import threading
+
+from rich.console import Console, Group
+from rich.live import Live
+from rich.progress_bar import ProgressBar
+from rich.spinner import Spinner
+from rich.table import Table
+from rich.text import Text
+
+from ancestryllm.core.jobs import JobSnapshot, JobState
+
+_ACTIVE_STATES = frozenset({JobState.QUEUED, JobState.RUNNING})
+_REFRESHES_PER_SECOND = 8
+
+
+class JobProgressDisplay:
+    """Keep active jobs visible without owning job execution or service state."""
+
+    def __init__(self, console: Console) -> None:
+        self.console = console
+        self._lock = threading.RLock()
+        self._active: dict[str, JobSnapshot] = {}
+        self._live: Live | None = None
+
+    @property
+    def active(self) -> bool:
+        return self._live is not None
+
+    @property
+    def renderable(self) -> Table:
+        return self._render()
+
+    def handle(self, snapshot: JobSnapshot) -> None:
+        with self._lock:
+            if snapshot.state in _ACTIVE_STATES:
+                self._active[snapshot.job_id] = snapshot
+                if self._live is None:
+                    live = Live(
+                        self._render(),
+                        console=self.console,
+                        auto_refresh=True,
+                        refresh_per_second=_REFRESHES_PER_SECOND,
+                        transient=True,
+                        redirect_stdout=False,
+                        redirect_stderr=False,
+                    )
+                    live.start(refresh=True)
+                    self._live = live
+                else:
+                    self._live.update(self._render(), refresh=True)
+                return
+
+            self._active.pop(snapshot.job_id, None)
+            if self._live is not None and self._active:
+                self._live.update(self._render(), refresh=True)
+            elif self._live is not None:
+                self._stop_live()
+            self.console.print(self._summary(snapshot))
+
+    def _render(self) -> Table:
+        table = Table(title="Background jobs", box=None, show_header=True)
+        table.add_column("Job")
+        table.add_column("Operation")
+        table.add_column("Progress")
+        for snapshot in sorted(self._active.values(), key=lambda item: item.job_id):
+            operation = snapshot.progress.operation if snapshot.progress else snapshot.name
+            if snapshot.state is JobState.QUEUED:
+                indicator: Spinner | Group = Spinner("dots", text="queued")
+            elif snapshot.progress and snapshot.progress.total is not None:
+                completed = snapshot.progress.completed or 0
+                total = snapshot.progress.total
+                indicator = Group(
+                    ProgressBar(total=total, completed=completed, width=20),
+                    Text(f"{completed}/{total}"),
+                )
+            else:
+                indicator = Spinner("dots", text="working")
+            table.add_row(snapshot.job_id, operation, indicator)
+        return table
+
+    @staticmethod
+    def _summary(snapshot: JobSnapshot) -> Text:
+        if snapshot.state is JobState.COMPLETED:
+            return Text(f"{snapshot.job_id} completed: {snapshot.name}", style="green")
+        if snapshot.state is JobState.CANCELLED:
+            return Text(f"{snapshot.job_id} cancelled: {snapshot.name}", style="yellow")
+        return Text(
+            f"{snapshot.job_id} failed ({snapshot.error_code or 'JOB_FAILED'}): {snapshot.name}",
+            style="bold red",
+        )
+
+    def close(self) -> None:
+        with self._lock:
+            try:
+                self._stop_live()
+            finally:
+                self._active.clear()
+
+    def _stop_live(self) -> None:
+        live = self._live
+        self._live = None
+        if live is not None:
+            live.stop()

src/ancestryllm/console/router.py

@@ -262,6 +262,12 @@ def _help(self, tokens: tuple[str, ...]) -> str:
             specification = COMMAND_SPECIFICATIONS[command]
             actions = ", ".join(action.name for action in specification.actions)
             return f"{specification.name}: {specification.help}\nActions: {actions}"
+        if command == "jobs":
+            return (
+                "jobs [list|show JOB_ID]: inspect background job state, timestamps, "
+                "latest progress, results, and stable failures. Active jobs also render "
+                "live above the prompt."
+            )
         if self.active_module and command in {"info", "show", "set", "unset", "run", "back"}:
             return "Module commands: info, show [actions|options], set NAME VALUE, unset NAME, run [ACTION], back."
         raise AncestryError(

src/ancestryllm/console/shell.py

@@ -13,13 +13,15 @@
 from prompt_toolkit.history import DummyHistory
 from prompt_toolkit.input import Input
 from prompt_toolkit.output import Output
+from prompt_toolkit.patch_stdout import patch_stdout
 
 from ancestryllm.cli import dispatch
 from ancestryllm.console.completion import CompletionSnapshot, create_completer
 from ancestryllm.console.history import SecureHistory
 from ancestryllm.console.multiline import AsyncPrompt, MultilineEditor
 from ancestryllm.console.parser import split_repl_input
 from ancestryllm.console.presentation import PresentationAdapter, to_plain
+from ancestryllm.console.progress import JobProgressDisplay
 from ancestryllm.console.router import RouteKind, RouteResult, SessionRouter
 from ancestryllm.console.security import (
     RedactingTextIO,
@@ -29,7 +31,7 @@
 )
 from ancestryllm.core.context import AppContext
 from ancestryllm.core.errors import AncestryError
-from ancestryllm.core.jobs import JobManager
+from ancestryllm.core.jobs import JobManager, JobReporter
 
 _BACKGROUND_ACTIONS = frozenset(
     {
@@ -91,7 +93,9 @@ def __init__(
         self.stderr = RedactingTextIO(stderr or sys.stderr, context)
         self.presenter = PresentationAdapter.for_file(cast(TextIO, self.stdout))
         self.error_presenter = PresentationAdapter.for_file(cast(TextIO, self.stderr))
+        self.progress_display = JobProgressDisplay(self.presenter.console)
         self.jobs = jobs or JobManager(redact=context.secrets.redact)
+        self._unsubscribe_progress = self.jobs.subscribe(self.progress_display.handle)
         self.history = SecureHistory(
             context.config.data_dir / "repl_history",
             is_sensitive=lambda command: history_is_sensitive(command, self.router.active_module),
@@ -139,6 +143,8 @@ async def run_async(self) -> int:
                     return 0
         finally:
             await asyncio.to_thread(self.jobs.shutdown, wait=True)
+            self._unsubscribe_progress()
+            self.progress_display.close()
 
     async def execute_line(self, command: str) -> bool:
         for value in credential_values(command):
@@ -160,9 +166,9 @@ async def execute_line(self, command: str) -> bool:
             if namespace.command == "secrets" and namespace.action == "set":
                 await self._set_secret(namespace.name)
             elif self._should_background(namespace):
-                snapshot = self.jobs.submit(
+                snapshot = self.jobs.submit_with_progress(
                     f"{namespace.command} {namespace.action}",
-                    lambda: self._dispatch_job(namespace),
+                    lambda reporter: self._dispatch_job(namespace, reporter),
                     resource_keys=self._resource_keys(namespace),
                 )
                 self.presenter.render(
@@ -218,8 +224,13 @@ def _handle_job_control(self, command: str) -> bool:
     def _should_background(namespace: argparse.Namespace) -> bool:
         return (namespace.command, namespace.action) in _BACKGROUND_ACTIONS
 
-    def _dispatch_job(self, namespace: argparse.Namespace) -> dict[str, object]:
+    def _dispatch_job(
+        self,
+        namespace: argparse.Namespace,
+        reporter: JobReporter,
+    ) -> dict[str, object]:
         output: list[object] = []
+        reporter.update(f"{namespace.command} {namespace.action}")
 
         def capture(value: object, _json_output: bool = False) -> None:
             plain = to_plain(value)
@@ -329,4 +340,8 @@ async def _set_secret(self, name: str) -> None:
 def run_repl(context: AppContext | None = None) -> int:
     """Run the asynchronous shell from the synchronous console entry point."""
 
-    return asyncio.run(ReplApplication(context or AppContext.build()).run_async())
+    async def run() -> int:
+        with patch_stdout(raw=True):
+            return await ReplApplication(context or AppContext.build()).run_async()
+
+    return asyncio.run(run())