feat(session): singleton detached worker controls with multiprocessing

.importlinter

@@ -24,6 +24,7 @@ forbidden_modules =
     docctl.service_manifest
     docctl.service_query
     docctl.service_session
+    docctl.service_session_worker
     docctl.service_snapshot
     docctl.service_types
     docctl.services
@@ -44,6 +45,7 @@ forbidden_modules =
     docctl.service_manifest
     docctl.service_query
     docctl.service_session
+    docctl.service_session_worker
     docctl.service_snapshot
     docctl.service_types
     docctl.services
@@ -57,6 +59,7 @@ source_modules =
     docctl.service_manifest
     docctl.service_query
     docctl.service_session
+    docctl.service_session_worker
     docctl.service_snapshot
     docctl.service_types
     docctl.services
@@ -71,6 +74,7 @@ source_modules =
     docctl.service_manifest
     docctl.service_query
     docctl.service_session
+    docctl.service_session_worker
     docctl.service_snapshot
 forbidden_modules =
     docctl.chunking

ARCHITECTURE.md

@@ -20,6 +20,9 @@ Primary runtime code lives in `src/docctl/`.
 - `service_ingest.py`, `service_query.py`, `service_session.py`, `service_doctor.py`
   - Internal orchestration modules split by workflow domain.
   - Own command execution logic for ingest/query/session/doctor flows.
+- `service_session_worker.py`
+  - Singleton detached session worker lifecycle and local IPC transport orchestration.
+  - Reuses `service_session.py` request dispatch/runtime handling.
 - `service_snapshot.py`
   - Snapshot orchestration for index export/import workflows.
   - Owns zip archive validation, safe extraction, and restore policy enforcement.

README.md

@@ -84,6 +84,10 @@ docctl show <chunk_id_from_search> --allow-model-download
 | `docctl catalog` | Show index summary and per-document inventory. |
 | `docctl doctor` | Run local diagnostics for index and embedding setup. |
 | `docctl session` | Run a read-only NDJSON request session on stdin/stdout. |
+| `docctl session start` | Start singleton detached session worker (errors if already running). |
+| `docctl session status` | Show singleton detached session worker status. |
+| `docctl session exec` | Execute NDJSON requests through singleton detached worker. |
+| `docctl session stop` | Stop singleton detached session worker. |
 
 ## JSON and Session Mode
 Use `--json` for deterministic machine-readable output:
@@ -101,6 +105,22 @@ cat <<'EOF' | docctl session --allow-model-download
 EOF
 ```
 
+Use detached singleton worker mode when you want warm reuse across CLI invocations:
+
+```bash
+# Start singleton worker (default idle timeout: 900 seconds)
+docctl --json session start
+
+# Execute NDJSON requests through detached worker
+cat <<'EOF' | docctl session exec
+{"id":"q1","op":"search","query":"security gateway diagnostics","top_k":5}
+{"id":"q2","op":"catalog"}
+EOF
+
+# Stop worker explicitly
+docctl --json session stop
+```
+
 ## Configuration
 Global options:
 - `--index-path` (default: `.docctl`)

docs/product-specs/cli-contract.md

@@ -1,4 +1,4 @@
-# PSPEC-0001 CLI Contract v3 (Multi-Format Ingest, Locator-Free Retrieval)
+# PSPEC-0001 CLI Contract v4 (Singleton Session Worker Controls + Multi-Format Retrieval)
 
 ## Commands
 - `docctl ingest <path>`
@@ -10,6 +10,10 @@
 - `docctl catalog`
 - `docctl doctor`
 - `docctl session`
+- `docctl session start`
+- `docctl session status`
+- `docctl session exec`
+- `docctl session stop`
 
 ## Command Import Boundaries
 - Set A (ML-capable command path): `ingest`, `search`, `doctor`, `session`.
@@ -58,6 +62,16 @@
   - Search request accepts optional fields: `doc_id`, `source`, `title`, `min_score`, `rerank`, `rerank_candidates`.
   - Response line format: `{"id":"q1","ok":true,"result":{...}}`
   - Error response format: `{"id":"q1","ok":false,"error":{"message":"...","exit_code":NN}}`
+- `docctl session exec` uses the same NDJSON request/response contract as `docctl session`.
+
+## Singleton Session Worker
+- At most one detached worker session may run at a time.
+- `docctl session start` fails if a worker is already running.
+- `docctl session exec` auto-starts the singleton worker if none is running.
+- A running worker is bound to startup config (`index_path`, `collection`, embedding/rerank models, `allow_model_download`).
+- `docctl session exec` fails on config mismatch until `docctl session stop` is run.
+- Detached worker mode is POSIX-only.
+- Default idle timeout for detached worker mode is `900` seconds.
 
 ## Search Hit Payload
 - Base hit shape includes: `id`, `text`, `metadata`, `distance`, `score`, `rank`.
@@ -83,3 +97,5 @@
 4. Failure classes map to stable exit codes.
 5. In `--json` mode, stdout contains only deterministic JSON payloads.
 6. `session` reuses one embedding model instance across multiple search requests in one process.
+7. `session start` fails with a stable error when singleton worker is already running.
+8. `session exec` reuses the detached singleton worker and preserves NDJSON response contract.

docs/product-specs/index.md

@@ -5,4 +5,4 @@ Sorted by `id` ascending.
 
 | id | title | status | owner | last_updated | path |
 |---|---|---|---|---|---|
-| PSPEC-0001 | CLI Contract v3 (Multi-Format Ingest, Locator-Free Retrieval) | active | Engineering | 2026-03-26 | docs/product-specs/cli-contract.md |
+| PSPEC-0001 | CLI Contract v4 (Singleton Session Worker Controls + Multi-Format Retrieval) | active | Engineering | 2026-03-26 | docs/product-specs/cli-contract.md |

src/docctl/cli.py

@@ -26,22 +26,33 @@
 from .jsonio import dumps_json
 from .models import DoctorReport
 from .services import (
+    DEFAULT_SESSION_IDLE_TTL_SECONDS,
     collect_catalog,
     collect_stats,
+    exec_session_requests,
     export_snapshot,
     import_snapshot,
     ingest_path,
     run_doctor,
     run_session_requests,
     search_chunks,
+    session_worker_status,
     show_chunk,
+    start_session_worker,
+    stop_session_worker,
 )
 
 app = typer.Typer(
     add_completion=False,
     no_args_is_help=True,
     help="docctl is a CLI-first local document retrieval tool.",
 )
+session_app = typer.Typer(
+    add_completion=False,
+    invoke_without_command=True,
+    help="Session control and NDJSON session execution commands.",
+)
+app.add_typer(session_app, name="session")
 
 ALLOW_MODEL_DOWNLOAD_HELP = "Allow downloading missing embedding/reranker model artifacts."
 
@@ -396,7 +407,7 @@ def import_(
         _handle_error(error)
 
 
-@app.command(help="Run a read-only NDJSON request session on stdin/stdout.")
+@session_app.callback()
 def session(
     ctx: typer.Context,
     allow_model_download: bool = typer.Option(
@@ -405,12 +416,14 @@ def session(
         help=ALLOW_MODEL_DOWNLOAD_HELP,
     ),
 ) -> None:
-    """Run a read-only NDJSON request session on standard streams.
+    """Run legacy read-only NDJSON stream mode when no subcommand is selected.
 
     Args:
         ctx: Typer context containing resolved configuration.
         allow_model_download: Whether missing embedding models may be downloaded.
     """
+    if ctx.invoked_subcommand is not None:
+        return
     config = ctx.obj
     try:
         responses = run_session_requests(
@@ -424,6 +437,118 @@ def session(
         _handle_error(error)
 
 
+@session_app.command(help="Start singleton detached session worker.")
+def start(
+    ctx: typer.Context,
+    allow_model_download: bool = typer.Option(
+        False,
+        "--allow-model-download",
+        help=ALLOW_MODEL_DOWNLOAD_HELP,
+    ),
+    idle_ttl: int = typer.Option(
+        DEFAULT_SESSION_IDLE_TTL_SECONDS,
+        "--idle-ttl",
+        min=1,
+        help="Idle timeout in seconds before worker self-termination.",
+    ),
+) -> None:
+    """Start singleton detached session worker.
+
+    Args:
+        ctx: Typer context containing resolved configuration.
+        allow_model_download: Whether missing embedding models may be downloaded.
+        idle_ttl: Idle timeout in seconds.
+    """
+    config = ctx.obj
+    try:
+        payload = start_session_worker(
+            config=config,
+            allow_model_download=allow_model_download,
+            idle_ttl_seconds=idle_ttl,
+        )
+        _emit_success(config=config, payload=payload)
+    except Exception as error:  # noqa: BLE001
+        _handle_error(error)
+
+
+@session_app.command(help="Show singleton detached session worker status.")
+def status(
+    ctx: typer.Context,
+    allow_model_download: bool = typer.Option(
+        False,
+        "--allow-model-download",
+        help=ALLOW_MODEL_DOWNLOAD_HELP,
+    ),
+) -> None:
+    """Show singleton detached session worker status.
+
+    Args:
+        ctx: Typer context containing resolved configuration.
+        allow_model_download: Whether missing embedding models may be downloaded.
+    """
+    config = ctx.obj
+    try:
+        payload = session_worker_status(
+            config=config,
+            allow_model_download=allow_model_download,
+        )
+        _emit_success(config=config, payload=payload)
+    except Exception as error:  # noqa: BLE001
+        _handle_error(error)
+
+
+@session_app.command(name="exec", help="Execute NDJSON requests through singleton session worker.")
+def exec_(
+    ctx: typer.Context,
+    allow_model_download: bool = typer.Option(
+        False,
+        "--allow-model-download",
+        help=ALLOW_MODEL_DOWNLOAD_HELP,
+    ),
+    idle_ttl: int = typer.Option(
+        DEFAULT_SESSION_IDLE_TTL_SECONDS,
+        "--idle-ttl",
+        min=1,
+        help="Idle timeout in seconds for auto-started worker.",
+    ),
+) -> None:
+    """Execute NDJSON request lines through singleton session worker.
+
+    Args:
+        ctx: Typer context containing resolved configuration.
+        allow_model_download: Whether missing embedding models may be downloaded.
+        idle_ttl: Idle timeout in seconds for auto-started worker.
+    """
+    config = ctx.obj
+    try:
+        request_lines = list(sys.stdin)
+        responses = exec_session_requests(
+            config=config,
+            request_lines=request_lines,
+            allow_model_download=allow_model_download,
+            idle_ttl_seconds=idle_ttl,
+        )
+        for response in responses:
+            typer.echo(dumps_json(response))
+    except Exception as error:  # noqa: BLE001
+        _handle_error(error)
+
+
+@session_app.command(help="Stop singleton detached session worker.")
+def stop(ctx: typer.Context) -> None:
+    """Stop singleton detached session worker.
+
+    Args:
+        ctx: Typer context containing resolved configuration.
+    """
+    config = ctx.obj
+    try:
+        payload = stop_session_worker()
+        _emit_success(config=config, payload=payload)
+    except Exception as error:  # noqa: BLE001
+        _handle_error(error)
+
+
 def main() -> None:
     """Run the docctl CLI application entrypoint."""
     app()

src/docctl/service_session.py

@@ -390,28 +390,71 @@ def run_session_requests(
         Response dictionaries containing success results or structured errors.
     """
     runtime = SessionRuntime(request=request, deps=deps)
+    yield from run_session_requests_with_runtime(
+        runtime=runtime,
+        request_lines=request.request_lines,
+        verbose=request.config.verbose,
+    )
+
+
+def run_session_requests_with_runtime(
+    *,
+    runtime: SessionRuntime,
+    request_lines: Iterable[str],
+    verbose: bool,
+) -> Iterable[dict[str, Any]]:
+    """Process NDJSON request lines using an existing session runtime.
+
+    Args:
+        runtime: Reusable session runtime containing cached dependencies.
+        request_lines: Incoming NDJSON request lines.
+        verbose: Whether verbose mode is enabled.
+
+    Yields:
+        Response dictionaries containing success results or structured errors.
+    """
+    for raw_line in request_lines:
+        response = run_session_request_line(runtime=runtime, raw_line=raw_line, verbose=verbose)
+        if response is not None:
+            yield response
 
-    for raw_line in request.request_lines:
-        line = raw_line.strip()
-        if not line:
-            continue
-
-        request_id: Any = None
-        try:
-            payload = _parse_payload(line)
-            request_id = payload.get("id")
-            op = _parse_operation(payload)
-            handler = _OPERATION_HANDLERS.get(op)
-            if handler is None:
-                raise DocctlError(message=f"unsupported session operation: {op}", exit_code=50)
-
-            with suppress_external_output(enabled=not request.config.verbose):
-                result = handler(runtime, payload)
-
-            yield {
-                "id": request_id,
-                "ok": True,
-                "result": result,
-            }
-        except Exception as error:  # noqa: BLE001
-            yield session_error(request_id=request_id, error=error)
+
+def run_session_request_line(
+    *,
+    runtime: SessionRuntime,
+    raw_line: str,
+    verbose: bool,
+) -> dict[str, Any] | None:
+    """Process one NDJSON request line with a reusable runtime.
+
+    Args:
+        runtime: Reusable session runtime containing cached dependencies.
+        raw_line: Raw NDJSON request line.
+        verbose: Whether verbose mode is enabled.
+
+    Returns:
+        One response payload for non-empty lines, otherwise `None`.
+    """
+    line = raw_line.strip()
+    if not line:
+        return None
+
+    request_id: Any = None
+    try:
+        payload = _parse_payload(line)
+        request_id = payload.get("id")
+        op = _parse_operation(payload)
+        handler = _OPERATION_HANDLERS.get(op)
+        if handler is None:
+            raise DocctlError(message=f"unsupported session operation: {op}", exit_code=50)
+
+        with suppress_external_output(enabled=not verbose):
+            result = handler(runtime, payload)
+
+        return {
+            "id": request_id,
+            "ok": True,
+            "result": result,
+        }
+    except Exception as error:  # noqa: BLE001
+        return session_error(request_id=request_id, error=error)