feat: add archive client foundation

src/archive_clients/__init__.py

@@ -0,0 +1,4 @@
+"""Archive Control data-node daemon."""
+
+PROTO_COMMIT = "4ec852014dad74606d4078b3ae1aa208c814b033"
+

src/archive_clients/cli.py

@@ -0,0 +1,40 @@
+"""Archive client command-line entry point."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+from pathlib import Path
+from typing import Sequence
+
+from archive_clients.config import ClientConfig
+from archive_clients.daemon import ArchiveClientDaemon
+from archive_clients.probes import probe_root
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="archive-client")
+    parser.add_argument("--config", type=Path, required=True)
+    parser.add_argument("--mode", choices=("archive", "cache"))
+    parser.add_argument("--check-config", action="store_true")
+    arguments = parser.parse_args(argv)
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    config = ClientConfig.load(arguments.config, arguments.mode)
+    probes = [
+        probe_root(config.qbittorrent.local_root),
+        probe_root(config.syncthing.local_root),
+    ]
+    config.read_shared_token()
+    config.qbittorrent.read_password()
+    config.syncthing.read_api_key()
+    if arguments.check_config:
+        print(json.dumps({
+            "client_id": config.client_id,
+            "role": config.role,
+            "filesystems": [probe.__dict__ | {"root": str(probe.root)} for probe in probes],
+        }, sort_keys=True))
+        return 0
+    asyncio.run(ArchiveClientDaemon(config, probes).run())
+    return 0

src/archive_clients/config.py

@@ -0,0 +1,259 @@
+"""Strict TOML configuration with file-backed secrets."""
+
+from __future__ import annotations
+
+import os
+import re
+import stat
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any
+from urllib.parse import urlsplit
+
+import tomllib
+
+
+class ConfigError(ValueError):
+    pass
+
+
+_CLIENT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
+_DURATION = re.compile(r"^([1-9][0-9]*)(ms|s|m|h)$")
+_FACTORS = {"ms": 0.001, "s": 1, "m": 60, "h": 3600}
+_ENV = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
+
+
+@dataclass(frozen=True)
+class RootMapping:
+    api_root: PurePosixPath
+    local_root: Path
+
+    def api_to_local(self, api_path: str) -> Path:
+        candidate = PurePosixPath(api_path)
+        try:
+            relative = candidate.relative_to(self.api_root)
+        except ValueError as exc:
+            raise ConfigError("API path is outside its configured root") from exc
+        if any(part in {"", ".", ".."} for part in relative.parts):
+            raise ConfigError("API path contains an unsafe component")
+        return self.local_root.joinpath(*relative.parts)
+
+
+@dataclass(frozen=True)
+class ServiceConfig:
+    endpoint: str
+    api_root: PurePosixPath
+    local_root: Path
+    username: str | None = None
+    password_file: Path | None = None
+    api_key_file: Path | None = None
+    advertised_addresses: tuple[str, ...] = ()
+
+    @property
+    def roots(self) -> RootMapping:
+        return RootMapping(self.api_root, self.local_root)
+
+    def read_password(self) -> str | None:
+        return _secret(self.password_file, "service password") if self.password_file else None
+
+    def read_api_key(self) -> str | None:
+        return _secret(self.api_key_file, "service API key") if self.api_key_file else None
+
+
+@dataclass(frozen=True)
+class ConnectionConfig:
+    registration_timeout: float = 10
+    heartbeat_interval: float = 15
+    offline_timeout: float = 45
+    reconnect_initial: float = 1
+    reconnect_max: float = 60
+    reconnect_reset_after: float = 60
+    reconnect_jitter: bool = True
+
+
+@dataclass(frozen=True)
+class JobsConfig:
+    stall_after: float = 30 * 60
+
+
+@dataclass(frozen=True)
+class ClientConfig:
+    client_id: str
+    display_name: str
+    role: str
+    control_endpoint: str
+    shared_token_file: Path
+    state_db: Path
+    backup_dir: Path
+    qbittorrent: ServiceConfig
+    syncthing: ServiceConfig
+    connection: ConnectionConfig = ConnectionConfig()
+    jobs: JobsConfig = JobsConfig()
+
+    @classmethod
+    def load(cls, path: Path, mode_override: str | None = None) -> "ClientConfig":
+        try:
+            with path.open("rb") as source:
+                raw = tomllib.load(source)
+        except (OSError, tomllib.TOMLDecodeError) as exc:
+            raise ConfigError("configuration cannot be read") from exc
+        _keys(raw, {
+            "client_id", "display_name", "role", "control_endpoint",
+            "shared_token_file", "state_db", "backup_dir", "connection",
+            "jobs", "qbittorrent", "syncthing",
+        }, "root")
+        role = mode_override or raw.get("role")
+        if role not in {"archive", "cache"}:
+            raise ConfigError("role/--mode must be archive or cache")
+        client_id = raw.get("client_id")
+        if not isinstance(client_id, str) or not _CLIENT_ID.fullmatch(client_id):
+            raise ConfigError("client_id is invalid")
+        display_name = raw.get("display_name")
+        if not isinstance(display_name, str) or not 1 <= len(display_name) <= 128:
+            raise ConfigError("display_name is invalid")
+        connection = _connection(raw.get("connection", {}))
+        jobs = _jobs(raw.get("jobs", {}))
+        return cls(
+            client_id, display_name, role,
+            _endpoint(raw, "control_endpoint", {"ws", "wss"}),
+            _absolute_path(raw, "shared_token_file"),
+            _absolute_path(raw, "state_db"),
+            _absolute_path(raw, "backup_dir"),
+            _service(raw.get("qbittorrent"), "qbittorrent"),
+            _service(raw.get("syncthing"), "syncthing"), connection, jobs,
+        )
+
+    def read_shared_token(self) -> str:
+        return _secret(self.shared_token_file, "shared token")
+
+
+def _service(value: Any, name: str) -> ServiceConfig:
+    if not isinstance(value, dict):
+        raise ConfigError(f"{name} must be a table")
+    allowed = {
+        "endpoint", "api_root", "local_root", "username", "password_file",
+        "api_key_file", "advertised_addresses",
+    }
+    _keys(value, allowed, name)
+    api_root = PurePosixPath(_string(value, "api_root"))
+    if not api_root.is_absolute() or ".." in api_root.parts:
+        raise ConfigError(f"{name}.api_root must be absolute and normalized")
+    username = value.get("username")
+    if username is not None and (not isinstance(username, str) or not username):
+        raise ConfigError(f"{name}.username must be a non-empty string")
+    addresses = value.get("advertised_addresses", [])
+    if not isinstance(addresses, list) or any(
+        not isinstance(address, str) or not address for address in addresses
+    ):
+        raise ConfigError(f"{name}.advertised_addresses must be a string array")
+    return ServiceConfig(
+        _endpoint(value, "endpoint", {"http", "https"}), api_root,
+        _absolute_path(value, "local_root"), username,
+        _absolute_path(value, "password_file")
+        if "password_file" in value else None,
+        _absolute_path(value, "api_key_file")
+        if "api_key_file" in value else None,
+        tuple(addresses),
+    )
+
+
+def _connection(value: Any) -> ConnectionConfig:
+    if not isinstance(value, dict):
+        raise ConfigError("connection must be a table")
+    _keys(value, {"registration_timeout", "heartbeat_interval", "offline_timeout", "reconnect_initial", "reconnect_max", "reconnect_reset_after", "reconnect_jitter"}, "connection")
+    result = ConnectionConfig(
+        registration_timeout=_duration(value.get("registration_timeout", "10s")),
+        heartbeat_interval=_duration(value.get("heartbeat_interval", "15s")),
+        offline_timeout=_duration(value.get("offline_timeout", "45s")),
+        reconnect_initial=_duration(value.get("reconnect_initial", "1s")),
+        reconnect_max=_duration(value.get("reconnect_max", "60s")),
+        reconnect_reset_after=_duration(value.get("reconnect_reset_after", "60s")),
+        reconnect_jitter=value.get("reconnect_jitter", True),
+    )
+    if result.reconnect_initial > result.reconnect_max:
+        raise ConfigError("reconnect_initial cannot exceed reconnect_max")
+    if result.offline_timeout <= result.heartbeat_interval:
+        raise ConfigError("offline_timeout must exceed heartbeat_interval")
+    if not isinstance(result.reconnect_jitter, bool):
+        raise ConfigError("reconnect_jitter must be boolean")
+    return result
+
+
+def _jobs(value: Any) -> JobsConfig:
+    if not isinstance(value, dict):
+        raise ConfigError("jobs must be a table")
+    _keys(value, {"stall_after"}, "jobs")
+    return JobsConfig(stall_after=_duration(value.get("stall_after", "30m")))
+
+
+def _duration(value: Any) -> float:
+    if not isinstance(value, str) or not (match := _DURATION.fullmatch(value)):
+        raise ConfigError("duration must look like 15s or 30m")
+    return int(match.group(1)) * _FACTORS[match.group(2)]
+
+
+def _string(value: dict[str, Any], key: str) -> str:
+    item = value.get(key)
+    if not isinstance(item, str) or not item:
+        raise ConfigError(f"{key} must be a non-empty string")
+    return _expand(item)
+
+
+def _path(value: dict[str, Any], key: str) -> Path:
+    return Path(_string(value, key))
+
+
+def _absolute_path(value: dict[str, Any], key: str) -> Path:
+    path = _path(value, key)
+    if not path.is_absolute():
+        raise ConfigError(f"{key} must be an absolute path")
+    return path
+
+
+def _endpoint(
+    value: dict[str, Any], key: str, allowed_schemes: set[str]
+) -> str:
+    endpoint = _string(value, key)
+    parsed = urlsplit(endpoint)
+    if (
+        parsed.scheme not in allowed_schemes
+        or not parsed.hostname
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.fragment
+    ):
+        schemes = "/".join(sorted(allowed_schemes))
+        raise ConfigError(f"{key} must be a credential-free {schemes} endpoint")
+    return endpoint
+
+
+def _expand(value: str) -> str:
+    def replace(match: re.Match[str]) -> str:
+        name = match.group(1)
+        if name not in os.environ:
+            raise ConfigError(f"environment variable {name} is not set")
+        return os.environ[name]
+    expanded = _ENV.sub(replace, value)
+    if "$" in expanded:
+        raise ConfigError("only ${NAME} environment interpolation is supported")
+    return expanded
+
+
+def _secret(path: Path, name: str) -> str:
+    try:
+        metadata = path.stat()
+        exposed = metadata.st_mode & (stat.S_IRWXG | stat.S_IRWXO)
+        if not stat.S_ISREG(metadata.st_mode) or exposed:
+            raise ConfigError(f"{name} file permissions are unsafe")
+        value = path.read_text(encoding="utf-8").strip()
+    except OSError as exc:
+        raise ConfigError(f"{name} file cannot be read") from exc
+    if not value:
+        raise ConfigError(f"{name} is empty")
+    return value
+
+
+def _keys(value: dict[str, Any], allowed: set[str], name: str) -> None:
+    unknown = sorted(set(value) - allowed)
+    if unknown:
+        raise ConfigError(f"unknown {name} keys: {', '.join(unknown)}")

src/archive_clients/daemon.py

@@ -0,0 +1,198 @@
+"""Reconnectable Archive Control client transport."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import random
+import time
+import uuid
+from typing import Any
+
+from websockets.asyncio.client import connect
+
+from archive_clients.config import ClientConfig
+from archive_clients.probes import FilesystemProbe
+from archive_clients.protocol import (
+    decode,
+    decode_message,
+    encode,
+    encode_message,
+    new_envelope,
+)
+from archive_clients.state import ClientStore, CommandConflict
+from archive_control.v1 import client_pb2, common_pb2, control_pb2, job_pb2
+
+
+logger = logging.getLogger(__name__)
+
+
+class ArchiveClientDaemon:
+    def __init__(self, config: ClientConfig, probes: list[FilesystemProbe]):
+        self.config = config
+        self.probes = probes
+        self.store = ClientStore(config.state_db)
+
+    async def run(self) -> None:
+        await asyncio.to_thread(self.store.initialize)
+        delay = self.config.connection.reconnect_initial
+        while True:
+            started = time.monotonic()
+            try:
+                await self._connection()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                logger.warning("control connection ended: %s", type(exc).__name__)
+            if time.monotonic() - started >= self.config.connection.reconnect_reset_after:
+                delay = self.config.connection.reconnect_initial
+            wait = random.uniform(0, delay) if self.config.connection.reconnect_jitter else delay
+            await asyncio.sleep(wait)
+            delay = min(delay * 2, self.config.connection.reconnect_max)
+
+    async def _connection(self) -> None:
+        async with connect(
+            self.config.control_endpoint, ping_interval=None, compression=None,
+            max_size=1024 * 1024,
+        ) as websocket:
+            registration = self._registration()
+            await websocket.send(encode(registration))
+            registration.register_request.shared_token = ""
+            response = decode(await asyncio.wait_for(
+                websocket.recv(), self.config.connection.registration_timeout
+            ))
+            if response.WhichOneof("payload") != "register_response":
+                raise RuntimeError("control did not answer registration")
+            if response.correlation_id != registration.message_id:
+                raise RuntimeError("registration response correlation mismatch")
+            if response.register_response.status != client_pb2.REGISTRATION_STATUS_ACCEPTED:
+                raise RuntimeError("control rejected registration")
+            if response.register_response.negotiated_version.major != 1:
+                raise RuntimeError("control negotiated an unsupported protocol version")
+            outbound: asyncio.Queue[str] = asyncio.Queue(maxsize=100)
+            writer = asyncio.create_task(self._writer(websocket, outbound))
+            try:
+                async for frame in websocket:
+                    await self._handle(decode(frame), outbound)
+            finally:
+                writer.cancel()
+                await asyncio.gather(writer, return_exceptions=True)
+
+    def _registration(self):
+        envelope = new_envelope()
+        request = envelope.register_request
+        request.protocol_version.CopyFrom(envelope.protocol_version)
+        request.client.client_id = self.config.client_id
+        request.client.display_name = self.config.display_name
+        request.client.role = (
+            common_pb2.CLIENT_ROLE_ARCHIVE
+            if self.config.role == "archive" else common_pb2.CLIENT_ROLE_CACHE
+        )
+        request.connection_instance_id = str(uuid.uuid4())
+        request.shared_token = self.config.read_shared_token()
+        request.capabilities.max_envelope_bytes = 1024 * 1024
+        request.capabilities.syncthing_advertised_addresses.extend(
+            self.config.syncthing.advertised_addresses
+        )
+        for root_name, probe in zip(
+            ("qbittorrent", "syncthing"), self.probes, strict=True
+        ):
+            filesystem = request.capabilities.filesystems.add()
+            filesystem.root_name = root_name
+            filesystem.readable = probe.readable
+            filesystem.writable = probe.writable
+            filesystem.hard_link = probe.hard_link
+            filesystem.sparse_files = probe.sparse_files
+        if all(probe.sparse_files for probe in self.probes):
+            request.capabilities.features.append(client_pb2.CLIENT_FEATURE_SPARSE_FILES)
+        for cursor in self.store.list_active_job_cursors():
+            active = request.active_jobs.add()
+            active.job_id = str(cursor["job_id"])
+            active.job_revision = int(cursor["revision"])
+            active.last_event_sequence = int(cursor["last_event_sequence"])
+            active.state = job_pb2.JobState.Value(str(cursor["state"]))
+            active.committed = bool(cursor["committed"])
+        return envelope
+
+    async def _writer(
+        self, websocket: Any, outbound: asyncio.Queue[str]
+    ) -> None:
+        while True:
+            await websocket.send(await outbound.get())
+
+    async def _handle(
+        self, envelope: Any, outbound: asyncio.Queue[str]
+    ) -> None:
+        payload = envelope.WhichOneof("payload")
+        if payload == "heartbeat":
+            response = new_envelope()
+            response.correlation_id = envelope.message_id
+            response.heartbeat_ack.sequence = envelope.heartbeat.sequence
+            await outbound.put(encode(response))
+        elif payload == "command":
+            await self._accept_command(envelope, outbound)
+        elif payload == "protocol_error":
+            logger.warning("control reported protocol error code=%s", envelope.protocol_error.error.code)
+
+    async def _accept_command(
+        self, envelope: Any, outbound: asyncio.Queue[str]
+    ) -> None:
+        command = envelope.command
+        acknowledgement = self._initial_acknowledgement(command)
+        accepted = None
+        try:
+            accepted = await asyncio.to_thread(
+                self.store.accept_command,
+                command.command_id,
+                encode_message(command),
+                encode_message(acknowledgement),
+            )
+            if accepted.duplicate:
+                acknowledgement = decode_message(
+                    accepted.acknowledgement_json, control_pb2.CommandAck()
+                )
+                if (
+                    acknowledgement.status
+                    == control_pb2.COMMAND_ACK_STATUS_ACCEPTED
+                ):
+                    acknowledgement.status = (
+                        control_pb2.COMMAND_ACK_STATUS_DUPLICATE
+                    )
+        except CommandConflict:
+            acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
+            acknowledgement.error.code = common_pb2.ERROR_CODE_CONFLICT
+            acknowledgement.error.message = "command ID content conflict"
+        response = new_envelope()
+        response.correlation_id = envelope.message_id
+        response.command_ack.CopyFrom(acknowledgement)
+        await outbound.put(encode(response))
+
+        if (
+            accepted is not None
+            and not accepted.duplicate
+            and acknowledgement.status
+            == control_pb2.COMMAND_ACK_STATUS_ACCEPTED
+            and command.WhichOneof("payload") == "request_job_snapshot"
+        ):
+            snapshot = new_envelope()
+            snapshot.correlation_id = envelope.message_id
+            snapshot.client_state_snapshot.snapshot_id = str(uuid.uuid4())
+            snapshot.client_state_snapshot.observed_at.CopyFrom(snapshot.sent_at)
+            await outbound.put(encode(snapshot))
+
+    @staticmethod
+    def _initial_acknowledgement(command: Any) -> control_pb2.CommandAck:
+        acknowledgement = control_pb2.CommandAck(command_id=command.command_id)
+        if not command.command_id or command.WhichOneof("payload") is None:
+            acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
+            acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT
+            acknowledgement.error.message = "command ID and payload are required"
+        elif command.WhichOneof("payload") == "request_job_snapshot":
+            acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED
+        else:
+            acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
+            acknowledgement.error.code = common_pb2.ERROR_CODE_UNSUPPORTED
+            acknowledgement.error.message = (
+                "command is not supported by this client build"
+            )
+        return acknowledgement

src/archive_clients/probes.py

@@ -0,0 +1,54 @@
+"""Fail-fast root and filesystem capability probes."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class ProbeError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class FilesystemProbe:
+    root: Path
+    readable: bool
+    writable: bool
+    hard_link: bool
+    sparse_files: bool
+
+
+def probe_root(root: Path) -> FilesystemProbe:
+    if not root.is_absolute() or not root.is_dir():
+        raise ProbeError(f"configured root is not an existing absolute directory: {root}")
+    if not os.access(root, os.R_OK | os.X_OK | os.W_OK):
+        raise ProbeError(f"configured root permissions are insufficient: {root}")
+    source: Path | None = None
+    linked: Path | None = None
+    try:
+        descriptor, raw_path = tempfile.mkstemp(prefix=".archive-control-probe-", dir=root)
+        source = Path(raw_path)
+        with os.fdopen(descriptor, "wb") as probe:
+            probe.seek(1024 * 1024)
+            probe.write(b"x")
+            probe.flush()
+            os.fsync(probe.fileno())
+        allocated = source.stat().st_blocks * 512
+        sparse = allocated < source.stat().st_size
+        linked = source.with_name(f"{source.name}.link")
+        try:
+            os.link(source, linked)
+            hard_link = linked.stat().st_ino == source.stat().st_ino
+        except OSError:
+            hard_link = False
+        return FilesystemProbe(root, True, True, hard_link, sparse)
+    except OSError as exc:
+        raise ProbeError(f"filesystem capability probe failed for {root}") from exc
+    finally:
+        if linked is not None:
+            linked.unlink(missing_ok=True)
+        if source is not None:
+            source.unlink(missing_ok=True)

src/archive_clients/protocol.py

@@ -0,0 +1,93 @@
+"""Strict protobuf-JSON envelope helpers."""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from google.protobuf import json_format
+
+from archive_control.v1 import envelope_pb2
+
+
+class ProtocolError(ValueError):
+    pass
+
+
+def new_envelope() -> envelope_pb2.Envelope:
+    envelope = envelope_pb2.Envelope()
+    envelope.protocol_version.major = 1
+    envelope.message_id = str(uuid.uuid4())
+    envelope.sent_at.FromDatetime(datetime.now(timezone.utc))
+    return envelope
+
+
+def encode(envelope: envelope_pb2.Envelope) -> str:
+    return json_format.MessageToJson(
+        envelope, preserving_proto_field_name=False, indent=None,
+        sort_keys=True, ensure_ascii=False,
+    )
+
+
+def encode_message(message: Any) -> str:
+    return json_format.MessageToJson(
+        message, preserving_proto_field_name=False, indent=None,
+        sort_keys=True, ensure_ascii=False,
+    )
+
+
+def decode_message(data: str, message: Any) -> Any:
+    try:
+        raw = json.loads(data, object_pairs_hook=_unique)
+        return json_format.ParseDict(raw, message)
+    except (ValueError, json.JSONDecodeError, json_format.ParseError) as exc:
+        raise ProtocolError("invalid protobuf JSON message") from exc
+
+
+def decode(data: str | bytes, max_bytes: int = 1024 * 1024) -> envelope_pb2.Envelope:
+    if not isinstance(data, str) or len(data.encode()) > max_bytes:
+        raise ProtocolError("control envelope must be bounded text")
+    try:
+        raw = json.loads(data, object_pairs_hook=_unique)
+        envelope = json_format.ParseDict(raw, envelope_pb2.Envelope())
+    except (
+        TypeError,
+        ValueError,
+        json.JSONDecodeError,
+        json_format.ParseError,
+    ) as exc:
+        raise ProtocolError("invalid control envelope") from exc
+    if envelope.protocol_version.major != 1:
+        raise ProtocolError("unsupported protocol major version")
+    _canonical_uuid(envelope.message_id, "message_id")
+    if envelope.correlation_id:
+        _canonical_uuid(envelope.correlation_id, "correlation_id")
+    if not envelope.HasField("sent_at"):
+        raise ProtocolError("sent_at is required")
+    try:
+        envelope.sent_at.ToDatetime(tzinfo=timezone.utc)
+    except (ValueError, OverflowError) as exc:
+        raise ProtocolError("sent_at is invalid") from exc
+    if envelope.WhichOneof("payload") is None:
+        raise ProtocolError("unsupported or empty control envelope")
+    return envelope
+
+
+def _unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    result = {}
+    for key, value in pairs:
+        if key in result:
+            raise ValueError("duplicate JSON key")
+        result[key] = value
+    return result
+
+
+def _canonical_uuid(value: str, name: str) -> None:
+    try:
+        parsed = uuid.UUID(value)
+    except (ValueError, AttributeError) as exc:
+        raise ProtocolError(f"{name} must be a canonical UUID") from exc
+    if str(parsed) != value:
+        raise ProtocolError(f"{name} must be a lowercase canonical UUID")

src/archive_clients/state.py

@@ -0,0 +1,122 @@
+"""Durable command inbox and client recovery cursors."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sqlite3
+from dataclasses import dataclass
+from pathlib import Path
+
+
+class CommandConflict(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class CommandAcceptance:
+    duplicate: bool
+    acknowledgement_json: str
+
+
+class ClientStore:
+    def __init__(self, database: Path):
+        self.database = database
+
+    def initialize(self) -> None:
+        self.database.parent.mkdir(parents=True, exist_ok=True)
+        with self._connect() as connection:
+            version = connection.execute("PRAGMA user_version").fetchone()[0]
+            if version > 1:
+                raise RuntimeError(
+                    f"client database schema {version} is newer than supported"
+                )
+            connection.executescript(_SCHEMA)
+            connection.execute("PRAGMA user_version = 1")
+            if connection.execute("PRAGMA foreign_key_check").fetchall():
+                raise RuntimeError("client database foreign-key check failed")
+
+    def accept_command(
+        self, command_id: str, command_json: str, acknowledgement_json: str
+    ) -> CommandAcceptance:
+        payload = _canonical(json.loads(command_json))
+        acknowledgement = _canonical(json.loads(acknowledgement_json))
+        digest = hashlib.sha256(payload.encode()).hexdigest()
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            existing = connection.execute(
+                "SELECT payload_sha256, command_json, acknowledgement_json FROM commands WHERE command_id = ?",
+                (command_id,),
+            ).fetchone()
+            if existing:
+                if existing["payload_sha256"] != digest or existing["command_json"] != payload:
+                    raise CommandConflict("command ID was reused with different content")
+                return CommandAcceptance(True, existing["acknowledgement_json"])
+            connection.execute(
+                """
+                INSERT INTO commands (
+                    command_id, payload_sha256, command_json,
+                    acknowledgement_json, state
+                ) VALUES (?, ?, ?, ?, 'received')
+                """,
+                (command_id, digest, payload, acknowledgement),
+            )
+        return CommandAcceptance(False, acknowledgement)
+
+    def list_active_job_cursors(self) -> list[dict[str, object]]:
+        with self._connect() as connection:
+            rows = connection.execute(
+                """
+                SELECT job_id, revision, last_event_sequence, state, committed
+                FROM jobs WHERE state NOT IN ('JOB_STATE_SUCCEEDED',
+                    'JOB_STATE_FAILED', 'JOB_STATE_CANCELLED')
+                ORDER BY job_id
+                """
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.database, isolation_level=None, timeout=5)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys = ON")
+        connection.execute("PRAGMA journal_mode = WAL")
+        connection.execute("PRAGMA synchronous = FULL")
+        connection.execute("PRAGMA busy_timeout = 5000")
+        return connection
+
+
+def _canonical(value: object) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS commands (
+    command_id TEXT PRIMARY KEY,
+    payload_sha256 TEXT NOT NULL,
+    command_json TEXT NOT NULL,
+    acknowledgement_json TEXT NOT NULL,
+    state TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS jobs (
+    job_id TEXT PRIMARY KEY,
+    definition_json TEXT NOT NULL,
+    state TEXT NOT NULL,
+    revision INTEGER NOT NULL,
+    last_event_sequence INTEGER NOT NULL,
+    committed INTEGER NOT NULL CHECK(committed IN (0, 1))
+);
+CREATE TABLE IF NOT EXISTS events (
+    event_id TEXT PRIMARY KEY,
+    job_id TEXT NOT NULL REFERENCES jobs(job_id),
+    sequence INTEGER NOT NULL,
+    event_json TEXT NOT NULL,
+    UNIQUE(job_id, sequence)
+);
+CREATE TABLE IF NOT EXISTS file_journal (
+    operation_id TEXT PRIMARY KEY,
+    job_id TEXT NOT NULL,
+    intent_json TEXT NOT NULL,
+    result_json TEXT,
+    state TEXT NOT NULL
+);
+"""