feat: add authenticated recorder control channel

.github/workflows/control-contract.yml

@@ -0,0 +1,45 @@
+name: Cross-platform control contract
+
+on:
+  pull_request:
+    branches: [main]
+    paths:
+      - "openadapt_capture/control.py"
+      - "openadapt_capture/recorder.py"
+      - "openadapt_capture/cli.py"
+      - "tests/test_control.py"
+      - "tests/control_recorder_process.py"
+      - ".github/workflows/control-contract.yml"
+
+concurrency:
+  group: capture-control-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  control-contract:
+    # The normal PR matrix proves Linux. This focused matrix proves the same
+    # subprocess ready -> status -> stop -> verified-complete contract on the
+    # two other supported operating systems without running the costly native
+    # video and input-injection suites on every PR.
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [macos-latest, windows-latest]
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+        with:
+          version: "latest"
+
+      - name: Set up Python
+        run: uv python install 3.12
+
+      - name: Install dependencies
+        run: uv sync --extra dev
+
+      - name: Run the authenticated control contract
+        run: uv run pytest tests/test_control.py -v --timeout=120

README.md

@@ -119,11 +119,44 @@ Record from the command line:
 
 ```bash
 capture record ./my-capture --description "Describe the workflow"
-# Press Ctrl-C to stop.
+# The ready message prints the exact session ID.
+
+# From another terminal:
+capture status --session-id SESSION_ID
+capture stop --session-id SESSION_ID
 
 capture info ./my-capture
 ```
 
+The recorder creates its control endpoint before it reports ready. The endpoint
+listens on IPv4 loopback only. Each request and response uses an authenticated
+session capability. Capture stores that capability only in an owner-only
+runtime file (`0700` directory and `0600` file on macOS/Linux; a protected
+current-user owner and DACL on Windows). Capture removes and verifies the
+absence of macOS extended ACL entries. The capability does not enter command
+arguments, logs, or the capture directory.
+
+`capture stop` binds the request to the exact session ID, process ID, and
+process start identity. It waits for producer and writer shutdown, reconciles
+producer counts with committed rows, checks database and relationship integrity,
+validates replay-relevant events, and writes atomic terminal metadata before it
+returns success. Repeated stop requests share one finalization result. A timeout,
+worker failure, invalid capability, stale process, or ambiguous set of active
+sessions returns non-success. A crash leaves `capture-state.json` incomplete;
+the next authenticated discovery removes the stale runtime descriptor only after
+it proves that the bound process instance is no longer live.
+
+Launchers and embedded clients use the public Python contract instead of
+reading recorder internals:
+
+```python
+from openadapt_capture import status_recording, stop_recording
+
+current = status_recording(session_id)
+completed = stop_recording(session_id, timeout=60)
+assert completed.complete and completed.integrity_verified
+```
+
 Or inspect processed actions in Python:
 
 ```python
@@ -140,6 +173,7 @@ A capture normally contains:
 
 ```text
 my-capture/
+├── capture-state.json
 ├── recording.db
 ├── oa_recording-*.mp4
 └── profiling.json

openadapt_capture/__init__.py

@@ -42,6 +42,15 @@
     plot_comparison,
 )
 from openadapt_capture.config import RecordingConfig
+from openadapt_capture.control import (
+    CaptureControlAuthenticationError,
+    CaptureControlError,
+    CaptureControlUnavailable,
+    RecorderStatus,
+    discover_recorders,
+    status_recording,
+    stop_recording,
+)
 from openadapt_capture.db.models import (
     ActionEvent as DBActionEvent,
 )
@@ -137,6 +146,13 @@
     # High-level APIs
     "Recorder",
     "RecordingConfig",
+    "RecorderStatus",
+    "CaptureControlError",
+    "CaptureControlUnavailable",
+    "CaptureControlAuthenticationError",
+    "discover_recorders",
+    "status_recording",
+    "stop_recording",
     "Capture",
     "CaptureSession",
     "Action",

openadapt_capture/cli.py

@@ -108,6 +108,11 @@ def record(
         if not recorder.wait_for_ready():
             print("Recording did not become ready. No successful capture was saved.")
             raise SystemExit(1)
+        print(f"Capture session: {recorder.control_session_id}")
+        print(
+            "Stop from another terminal: "
+            f"capture stop --session-id {recorder.control_session_id}"
+        )
         try:
             while recorder.is_recording:
                 time.sleep(1)
@@ -119,6 +124,66 @@ def record(
     print(f"Saved to: {output_dir}")
 
 
+def status(
+    session_id: str | None = None,
+    timeout: float = 5.0,
+    runtime_dir: str | None = None,
+) -> None:
+    """Show the authenticated status of an active recorder.
+
+    Args:
+        session_id: Exact Capture session ID. It can be omitted only when one
+            recorder is active.
+        timeout: Maximum seconds to wait for the recorder.
+        runtime_dir: Owner-only runtime directory override for an embedded
+            launcher or a test environment.
+    """
+    import json
+
+    from openadapt_capture.control import CaptureControlError, status_recording
+
+    try:
+        current = status_recording(
+            session_id,
+            timeout=timeout,
+            runtime_dir=runtime_dir,
+        )
+    except CaptureControlError as exc:
+        print(str(exc))
+        raise SystemExit(1) from exc
+    print(json.dumps(current.__dict__, sort_keys=True))
+
+
+def stop(
+    session_id: str | None = None,
+    timeout: float = 60.0,
+    runtime_dir: str | None = None,
+) -> None:
+    """Stop one recorder and confirm its finalized Capture session.
+
+    Args:
+        session_id: Exact Capture session ID. It can be omitted only when one
+            recorder is active.
+        timeout: Maximum seconds to wait for finalization and integrity checks.
+        runtime_dir: Owner-only runtime directory override for an embedded
+            launcher or a test environment.
+    """
+    import json
+
+    from openadapt_capture.control import CaptureControlError, stop_recording
+
+    try:
+        completed = stop_recording(
+            session_id,
+            timeout=timeout,
+            runtime_dir=runtime_dir,
+        )
+    except CaptureControlError as exc:
+        print(str(exc))
+        raise SystemExit(1) from exc
+    print(json.dumps(completed.__dict__, sort_keys=True))
+
+
 def visualize(
     capture_dir: str,
     output: str | None = None,
@@ -394,6 +459,8 @@ def main() -> None:
     import fire
     fire.Fire({
         "record": record,
+        "status": status,
+        "stop": stop,
         "visualize": visualize,
         "info": info,
         "transcribe": transcribe,