Add an asynchronous background job manager

docs/CONSOLE.md

@@ -34,8 +34,19 @@ The session controls are:
   saved options and `unset NAME` removes one.
 - `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.
 - `exit`, `quit`, or EOF leaves the REPL.
 
+Long-running RootsMagic queries/exports, GEDCOM operations, OCR extraction, and
+database backups run in a bounded background worker pool so the prompt remains
+responsive. The start response includes a stable job ID. Jobs move through
+`queued`, `running`, `completed`, `failed`, and (when cancellation is
+requested) `cancelled` states. Mutating jobs that target the same output,
+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.
+
 ## Multiline free text
 
 In the interactive console, omit `--question` from `rootsmagic query` or omit

docs/REPL_ARCHITECTURE.md

@@ -8,10 +8,9 @@ interface.
 
 ## Decision record
 
-**Status:** Implemented for the prompt-toolkit/Rich REPL and context-aware
-completion. Interactive console consolidation is complete; remaining future
-work is limited to features such as structured background jobs and cooperative
-cancellation.
+**Status:** Implemented for the prompt-toolkit/Rich REPL, context-aware
+completion, multiline input, and bounded background jobs. Cooperative
+cancellation and live progress presentation build on the job lifecycle.
 
 The application uses transport-neutral command specifications and
 invocation/result contracts between input adapters and application services.
@@ -27,9 +26,11 @@ flowchart LR
     Input["prompt_toolkit input\ncommands, completion, multiline, history"]
     Router["Session router\nmodule context, safe options"]
     Executor["Command executors\ntyped invocation and results"]
+    Jobs["Bounded job manager\nstate, results, resource locks"]
     Services["Application services\npolicy and use cases"]
     Presentation["Presentation adapters\nRich, text, JSON"]
     Input --> Router --> Executor --> Services
+    Executor --> Jobs --> Services
     Executor --> Presentation
 ```
 
@@ -45,7 +46,9 @@ flowchart LR
 4. **Application services** own use cases and enforce consent, endpoint
    policy, immutable source handling, and provider `none` offline behavior.
    They do not import UI libraries.
-5. **Presentation** renders DTOs, progress, and coded errors. Rich objects
+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.
 
 The dependency direction is one-way: `input -> routing -> execution ->
@@ -83,9 +86,9 @@ dependency.
   history, and interactive history is stored with owner-only permissions.
 - Provider selection and consent stay explicit. `provider=none` remains
   network-free even when keys or provider SDKs are installed.
-- Structured background jobs, progress management, and cooperative cancellation
-  are not part of the completed console consolidation. They require their own
-  lifecycle and shutdown design before implementation.
+- Background jobs expose queued, running, completed, failed, and cancelled
+  states. Progress presentation and cooperative cancellation extend this
+  lifecycle without moving policy or mutation safety into the UI.
 
 ## Migration status
 
@@ -96,7 +99,8 @@ dependency.
 | Asynchronous prompt-toolkit/Rich REPL | Implemented | Starts with no arguments and is installed by the main package |
 | 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 |
-| Background jobs and cooperative cancellation | Future work | Requires structured job lifecycle and atomic shutdown behavior |
+| Bounded background jobs | Implemented | Serializable states/results; per-resource mutation serialization |
+| Cooperative cancellation | Future work | Extends the job lifecycle around atomic sections and shutdown |
 
 ## Compatibility paths
 

src/ancestryllm/cli.py

@@ -7,7 +7,7 @@
 import json
 import sys
 from pathlib import Path
-from typing import Any, Sequence
+from typing import Any, Callable, Sequence
 
 from ancestryllm.console.presentation import PresentationAdapter
 from ancestryllm.core.config import AppConfig
@@ -122,26 +122,31 @@ def _key_values(values: list[str]) -> dict[str, str]:
     return result
 
 
-def dispatch(args: argparse.Namespace, context: AppContext) -> int:
+def dispatch(
+    args: argparse.Namespace,
+    context: AppContext,
+    *,
+    emit: Callable[[Any, bool], None] = _emit,
+) -> int:
     json_output = bool(args.json)
     if args.command == "modules":
         registry = ModuleRegistry(context)
         if args.action == "list":
-            _emit([_descriptor_payload(item) for item in registry.descriptors()], json_output)
+            emit([_descriptor_payload(item) for item in registry.descriptors()], json_output)
         elif args.action == "enable":
             registry.enable(args.module_id)
-            _emit(f"Enabled module: {args.module_id}", json_output)
+            emit(f"Enabled module: {args.module_id}", json_output)
         else:
             registry.disable(args.module_id)
-            _emit(f"Disabled module: {args.module_id}", json_output)
+            emit(f"Disabled module: {args.module_id}", json_output)
         return 0
 
     if args.command == "rootsmagic":
         from ancestryllm.rootsmagic.service import RootsMagicService
 
         rootsmagic_service = RootsMagicService(context.config, context.llm)
         if args.action == "list":
-            _emit(rootsmagic_service.list_trees(), json_output)
+            emit(rootsmagic_service.list_trees(), json_output)
         elif args.action == "query":
             query_result = (
                 rootsmagic_service.query_sql(args.tree, args.sql)
@@ -154,7 +159,7 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
                     consent=_consent(context, args.consent),
                 )
             )
-            _emit(query_result, json_output)
+            emit(query_result, json_output)
         else:
             export_result = rootsmagic_service.export(
                 args.tree,
@@ -168,7 +173,7 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
                 living=args.living,
                 report_path=args.report,
             )
-            _emit(export_result, json_output)
+            emit(export_result, json_output)
         return 0
 
     if args.command == "gedcom":
@@ -187,9 +192,9 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
                 consent=_consent(context, args.consent),
                 threshold=args.similarity_threshold,
             )
-            _emit(gedcom_result, json_output)
+            emit(gedcom_result, json_output)
         elif args.action == "subtree":
-            _emit(
+            emit(
                 gedcom_service.subtree(
                     args.input,
                     args.output,
@@ -201,7 +206,7 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
                 json_output,
             )
         elif args.action == "quality":
-            _emit(
+            emit(
                 gedcom_service.quality(args.input, args.output, root_person=args.root_person),
                 json_output,
             )
@@ -211,7 +216,7 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
 
     if args.command == "prompts":
         if args.action == "list":
-            _emit(context.prompts.list(), json_output)
+            emit(context.prompts.list(), json_output)
         elif args.action == "save":
             body = (
                 args.body if args.body is not None else args.body_file.read_text(encoding="utf-8")
@@ -221,26 +226,26 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
                 if args.schema_file
                 else None
             )
-            _emit(
+            emit(
                 context.prompts.save(
                     args.name, args.purpose, body, args.variable, schema, args.tag
                 ),
                 json_output,
             )
         elif args.action == "show":
-            _emit(context.prompts.get(args.name, args.version), json_output)
+            emit(context.prompts.get(args.name, args.version), json_output)
         else:
-            _emit(
+            emit(
                 context.prompts.render(args.name, _key_values(args.value), args.version),
                 json_output,
             )
         return 0
 
     if args.command == "people":
         if args.action == "list":
-            _emit(context.research.list_people(args.workspace), json_output)
+            emit(context.research.list_people(args.workspace), json_output)
         else:
-            _emit(
+            emit(
                 context.research.add_person(
                     args.display_name,
                     LivingStatus(args.living_status),
@@ -253,20 +258,20 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
 
     if args.command == "providers":
         if args.action == "list":
-            _emit(
+            emit(
                 {
                     "profiles": context.provider_profiles.list_profiles(),
                     "consents": context.provider_profiles.list_consents(),
                 },
                 json_output,
             )
         elif args.action == "create":
-            _emit(
+            emit(
                 context.provider_profiles.create_profile(args.name, args.provider, args.model),
                 json_output,
             )
         elif args.action == "consent":
-            _emit(
+            emit(
                 context.provider_profiles.create_consent(
                     args.name,
                     args.profile,
@@ -281,7 +286,7 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
             )
         else:
             context.provider_profiles.revoke_consent(args.name)
-            _emit(f"Revoked consent: {args.name}", json_output)
+            emit(f"Revoked consent: {args.name}", json_output)
         return 0
 
     if args.command == "secrets":
@@ -303,12 +308,12 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
             if value != confirmation:
                 raise AncestryError("SECRET_CONFIRMATION_FAILED", "Secret values did not match.")
             context.secrets.set(args.name, value)
-            _emit(f"Stored secret reference: {args.name}", json_output)
+            emit(f"Stored secret reference: {args.name}", json_output)
         elif args.action == "delete":
             context.secrets.delete(args.name)
-            _emit(f"Deleted secret reference: {args.name}", json_output)
+            emit(f"Deleted secret reference: {args.name}", json_output)
         else:
-            _emit({name: context.secrets.present(name) for name in names}, json_output)
+            emit({name: context.secrets.present(name) for name in names}, json_output)
         return 0
 
     if args.command == "ocr":
@@ -323,17 +328,17 @@ def dispatch(args: argparse.Namespace, context: AppContext) -> int:
             model=args.model,
             consent=_consent(context, args.consent),
         )
-        _emit(ocr_result, json_output)
+        emit(ocr_result, json_output)
         return 0
 
     if args.command == "database":
         if args.action == "diagnose":
             from ancestryllm.storage.diagnostics import diagnose_storage
 
-            _emit(diagnose_storage(context.database.path, context.secrets), json_output)
+            emit(diagnose_storage(context.database.path, context.secrets), json_output)
             return 0
         context.database.backup(args.destination.expanduser().resolve())
-        _emit(f"Encrypted backup created: {args.destination}", json_output)
+        emit(f"Encrypted backup created: {args.destination}", json_output)
         return 0
     raise AncestryError("COMMAND_UNKNOWN", "Unknown command.")
 

src/ancestryllm/console/completion.py

@@ -25,8 +25,19 @@
 __all__ = ["CompletionSnapshot", "create_completer"]
 
 _MAX_FILE_COMPLETIONS = 64
-_ROOT_CONTROLS = ("exit", "help", "quit", "use")
-_ACTIVE_CONTROLS = ("back", "exit", "help", "info", "quit", "run", "set", "show", "unset")
+_ROOT_CONTROLS = ("exit", "help", "jobs", "quit", "use")
+_ACTIVE_CONTROLS = (
+    "back",
+    "exit",
+    "help",
+    "info",
+    "jobs",
+    "quit",
+    "run",
+    "set",
+    "show",
+    "unset",
+)
 _DYNAMIC_SENSITIVE_KINDS = frozenset(
     {
         CompletionKind.MODEL,

src/ancestryllm/console/router.py

@@ -253,7 +253,8 @@ def _help(self, tokens: tuple[str, ...]) -> str:
             raise AncestryError("REPL_USAGE_ERROR", "Usage: help [COMMAND]", exit_code=2)
         if not tokens:
             return (
-                "Root commands: modules, use MODULE, help [COMMAND], exit, quit. "
+                "Root commands: modules, use MODULE, jobs [list|show ID], "
+                "help [COMMAND], exit, quit. "
                 "Enabled module commands can also be run directly."
             )
         command = tokens[0]