exlab_wizard.sync#

Sync package. Backend Spec §7.

Public re-exports for the NAS sync subsystem. Callers should depend on this package’s surface rather than reach into the sub-modules; the NASSyncClient is the only stateful object outside callers typically interact with.

exlab_wizard.sync.HandleState#: alias of SyncHandleState

class exlab_wizard.sync.NASSyncClient(*, config, queue_db, validator, cache_creation, sync_state_writer=None, keyring_store=None, worker_poll_interval_s=0.05, push_callable_factory=None, check_callable_factory=None, lsjson_callable_factory=None)[source]#

Bases: object

Durable, per-equipment NAS sync queue with Pre-Sync Gate.

Backend Spec §7.1, §7.3.

Lifecycle:

init() opens the queue DB, replays any in-flight jobs, and starts a single background worker task.
enqueue() runs the Pre-Sync Gate, gates the run if needed, and otherwise inserts a QUEUED row.
close() cancels the worker and closes the DB.

The worker loop is a simple “pick the oldest QUEUED whose next_attempt_at has passed” scheduler with at-most-one inflight job at a time. This keeps determinism for tests; production deployments can extend to per-equipment parallelism without changing the public API.

Parameters:

config (Config)
queue_db (Path)
validator (Validator)
cache_creation (CreationWriter)
sync_state_writer (SyncStateWriter | None)
keyring_store (Any)
worker_poll_interval_s (float)
push_callable_factory (Callable[[EquipmentConfig], Callable[..., Any]] | None)
check_callable_factory (Callable[[EquipmentConfig], Callable[..., Any]] | None)
lsjson_callable_factory (Callable[[EquipmentConfig], Callable[..., Any]] | None)

apply_config(config)[source]#

Swap the cached config + equipment map in place (no relaunch).

The equipment_id -> EquipmentConfig lookup is rebuilt into a local before assignment so the worker loop – which runs in the same event loop – never observes a half-built map. In-flight queued jobs carry their own captured paths; a removed equipment id simply errors that one job exactly as it would after a relaunch.

Parameters:: config (Config)
Return type:: None

async close()[source]#

Stop the worker and close the queue DB. Idempotent.

Return type:: None

async enqueue(run_path, files=None)[source]#

Pre-Sync Gate -> if hard-tier finding without override, mark sync_status='blocked_by_validation'. Otherwise insert a QUEUED row.

files (operator-free per-file NAS sync, 2026-05-21) is the per-file subset of run-relative POSIX paths eligible at enqueue time; an empty / omitted list means “the whole run”.

The queue holds one row per run_path (UNIQUE). Re-enqueue behaviour:

existing job in a terminal state (VERIFIED / CLEANUP_ELIGIBLE / CLEANED / FAILED) and a non-empty files list -> reset to QUEUED carrying the new subset. This is how a file modified after a prior verify gets re-synced.
existing job in a terminal VERIFIED / CLEANUP_ELIGIBLE / CLEANED state with an empty files list -> falls through to a no-op: there is no subset to re-sync and a successfully-verified run is not blindly re-queued. (Only a terminal FAILED row with empty files is re-armed – the manual-retry branch below.)
existing job active (QUEUED / RUNNING / AWAITING_VERIFY) -> no-op (newly settled files ride the next sweep).
existing terminal FAILED job with no files -> re-armed via reset_to_queued so the manual-retry contract holds.
no existing job -> insert a QUEUED row with files.

Returns a SyncJobHandle. The handle’s state is either SyncHandleState.BLOCKED or SyncHandleState.QUEUED.

Parameters:

run_path (Path)
files (list[str] | None)

Return type:

SyncJobHandle

async force_verify(run_path)[source]#

Re-run rclone check --download against the configured remote.

Used by the Settings “verify integrity” action. Reports only – does NOT advance the queue state and does NOT update verified_sha256 in sync_state.json (the rclone-only migration deliberately keeps Slot A SHA capture scoped to the sync-time path that has access to a freshly-read local copy).

Resolves the equipment from run_path’s first component, gathers every tracked file in sync_state.json as the --files-from subset, and asks the driver to compare. Returns a populated VerifyResult; the caller renders mismatched, missing, and errors to the operator. A run with no tracked files yields ok=True (nothing to verify).

Parameters:: run_path (Path)
Return type:: VerifyResult

async init()[source]#

Open the queue and start the worker task. Backend Spec §7.1.2.

Return type:: None

async retry(job_id)[source]#

Re-arm a FAILED job. Backend Spec §7.1.5 (Problems-tab Retry).

Parameters:: job_id (str)
Return type:: None

async status(run_path)[source]#

Return the queue state of the job for run_path.

"none" when no job exists; otherwise the underlying SyncJobState value.

Parameters:: run_path (Path)
Return type:: str

class exlab_wizard.sync.SyncHandleState(*values)[source]#

Bases: StrEnum

In-process state of a NAS sync job handle as observed by callers.

Distinct from SyncStatus (which is persisted in creation.json) and from the queue’s internal SyncJobState (which tracks the row in sync_queue.db). Backend Spec §7.1.

BLOCKED = 'blocked'#

QUEUED = 'queued'#

class exlab_wizard.sync.SyncJobHandle(job_id, state, run_path, blocking_findings=())[source]#

Bases: object

Lightweight handle returned by NASSyncClient.enqueue().

job_id is empty when the gate blocked enqueue (the on-disk sync_status will reflect the block). blocking_findings is present iff state == BLOCKED.

Parameters:

job_id (str)
state (SyncHandleState)
run_path (str)
blocking_findings (tuple[Finding, ...])

blocking_findings: tuple[Finding, ...]#

job_id: str#

run_path: str#

state: SyncHandleState#

class exlab_wizard.sync.SyncJobRow(id, run_path, equipment_id, state, attempts=0, last_attempt_at=None, next_attempt_at=None, last_error=None, verify_passes=0, verified_at=None, enqueued_at='', nas_path=None, files=())[source]#

Bases: object

One row in the jobs table. Backend Spec §7.1.1.

Parameters:

id (str)
run_path (str)
equipment_id (str)
state (SyncJobState)
attempts (int)
last_attempt_at (str | None)
next_attempt_at (str | None)
last_error (str | None)
verify_passes (int)
verified_at (str | None)
enqueued_at (str)
nas_path (str | None)
files (tuple[str, ...])

attempts: int#

enqueued_at: str#

equipment_id: str#

files: tuple[str, ...]#

id: str#

last_attempt_at: str | None#

last_error: str | None#

nas_path: str | None#

next_attempt_at: str | None#

run_path: str#

state: SyncJobState#

verified_at: str | None#

verify_passes: int#

class exlab_wizard.sync.SyncJobState(*values)[source]#

Bases: StrEnum

State machine for a sync job. Backend Spec §7.1.2.

AWAITING_VERIFY = 'awaiting_verify'#

CLEANED = 'cleaned'#

CLEANUP_ELIGIBLE = 'cleanup_eligible'#

FAILED = 'failed'#

QUEUED = 'queued'#

RUNNING = 'running'#

VERIFIED = 'verified'#

class exlab_wizard.sync.SyncQueue(db_path)[source]#

Bases: object

Async SQLite-backed durable sync queue. Backend Spec §7.1.1.

Use init() once at application startup; the database file is created on demand. After init, all CRUD methods are coroutines.

Parameters:: db_path (Path)

async close()[source]#

Close the underlying connection (idempotent).

Return type:: None

property db_path: Path#: The on-disk path of the queue database file.

async delete(job_id)[source]#

Remove a job row entirely. Used by the cleanup reaper after CLEANED.

Parameters:: job_id (str)
Return type:: None

async get_by_id(job_id)[source]#

Return the row with the given job_id or None.

Parameters:: job_id (str)
Return type:: SyncJobRow | None

async get_by_run_path(run_path)[source]#

Return the row whose run_path matches, or None.

Parameters:: run_path (Path)
Return type:: SyncJobRow | None

async init()[source]#

Open the connection, ensure the schema, and replay in-flight rows.

Replay semantics (§7.1.2): any RUNNING row at startup gets downgraded to QUEUED (the worker died mid-transfer); any AWAITING_VERIFY row is left as-is so the verifier picks it up.

Return type:: None

async insert(*, run_path, equipment_id, nas_path=None, job_id=None, files=None)[source]#

Insert a new QUEUED row for run_path.

files is the per-file subset (run-relative POSIX paths) eligible at enqueue time; an empty / omitted list means “the whole run”.

Raises aiosqlite.IntegrityError (via the UNIQUE constraint on run_path) if a row already exists for the same path.

Parameters:

run_path (Path)
equipment_id (str)
nas_path (str | None)
job_id (str | None)
files (tuple[str, ...] | list[str] | None)

Return type:

SyncJobRow

static is_terminal(state)[source]#

Return True if the state is a terminal state (no further work).

Parameters:: state (SyncJobState)
Return type:: bool

async list_all()[source]#

Return every row in the queue ordered by enqueued_at.

Return type:: list[SyncJobRow]

async list_in_state(state)[source]#

Return every row currently in state ordered by enqueued_at.

Parameters:: state (SyncJobState)
Return type:: list[SyncJobRow]

async record_failure(job_id, error, *, terminal=False, now=None)[source]#

Record a transport failure on job_id.

If terminal is True (auth failure, local file vanished) the job goes straight to FAILED with no backoff. Otherwise:

increment attempts
if attempts >= MAX_ATTEMPTS: terminal FAILED.
else: stay in QUEUED with next_attempt_at per backoff.

Parameters:

job_id (str)
error (str)
terminal (bool)
now (datetime | None)

Return type:

SyncJobRow

async requeue_with_files(job_id, files)[source]#

Reset a terminal job back to QUEUED carrying a new files list.

Operator-free per-file NAS sync design (2026-05-21): a file modified after a prior verify becomes re-eligible. When the poller observes such a file it re-arms the run’s terminal job (VERIFIED / CLEANUP_ELIGIBLE / CLEANED / FAILED) with the freshly eligible subset. attempts / last_error / verify_passes are cleared so the backoff schedule and verify counter start fresh.

Parameters:

job_id (str)
files (tuple[str, ...] | list[str])

Return type:

SyncJobRow

async reset_to_queued(job_id)[source]#

Reset a FAILED job back to QUEUED for a manual retry.

Per §7.1.5 the Problems-tab Retry action re-enqueues a failed job. attempts and last_error are cleared so the backoff schedule starts fresh.

Parameters:: job_id (str)
Return type:: SyncJobRow

async transition(job_id, new_state, *, last_error=None, increment_attempts=False, increment_verify_passes=False, verified_at=None, next_attempt_at=None, last_attempt_at=None, nas_path=None)[source]#

Transition a job to new_state and patch the auxiliary columns.

The patch is one UPDATE statement so either every column moves or none do. Raises ValueError if the job is missing.

Parameters:

job_id (str)
new_state (SyncJobState)
last_error (str | None)
increment_attempts (bool)
increment_verify_passes (bool)
verified_at (str | None)
next_attempt_at (str | None)
last_attempt_at (str | None)
nas_path (str | None)

Return type:

SyncJobRow

exception exlab_wizard.sync.TransportError(message, *, error_kind=None)[source]#

Bases: Exception

Raised when a transport probe cannot complete.

Distinct from TransportResult: this surfaces conditions where the transport’s hashsum-style probe could not produce a manifest at all – either because the upstream binary is missing (error_kind=None for the historical “binary not on PATH” case) or because the probe ran and failed in a classifiable way (AUTH, NETWORK, UNKNOWN). The queue worker uses error_kind to route the failure through the §7.1.5 retry policy:

AUTH – terminal FAILED, no retry (configuration problem).
NETWORK – non-terminal failure, exponential backoff retry.
UNKNOWN – treated as NETWORK for retry purposes.
None – legacy “binary missing” callers. Routed through the §7.1.5 HASH_MISMATCH branch: one immediate retry, then terminal FAILED on the second occurrence. The operator surfaces the binary-missing reason via last_error.

Parameters:

message (str)
error_kind (TransportErrorKind | None)

class exlab_wizard.sync.TransportErrorKind(*values)[source]#

Bases: StrEnum

Kind of failure that the queue worker uses to drive the retry policy.

Backend Spec §7.1.5.

NETWORK: timeout, ECONNRESET, transient SSH failure – retried with exponential backoff up to MAX_ATTEMPTS.
AUTH: authentication failure – terminal FAILED, no retry.
HASH_MISMATCH: post-transport hash check failed – single retry of the transport phase, then terminal.
LOCAL_FILE_VANISHED: the local file disappeared between transport and verify – terminal FAILED with local_file_vanished reason.
UNKNOWN: catch-all for transports returning a non-zero code we don’t recognize – treated as NETWORK for retry purposes.

AUTH = 'auth'#

HASH_MISMATCH = 'hash_mismatch'#

LOCAL_FILE_VANISHED = 'local_file_vanished'#

NETWORK = 'network'#

UNKNOWN = 'unknown'#

class exlab_wizard.sync.TransportResult(ok, error_kind=None, stderr='', stdout='', returncode=0)[source]#

Bases: object

Outcome of a transport push.

ok is True iff the transport reported success. On failure, error_kind selects the retry path; stderr is the raw stderr text for log surfacing; returncode is the subprocess exit code.

Parameters:

ok (bool)
error_kind (TransportErrorKind | None)
stderr (str)
stdout (str)
returncode (int)

error_kind: TransportErrorKind | None#

ok: bool#

returncode: int#

stderr: str#

stdout: str#

class exlab_wizard.sync.Verifier(driver=None)[source]#

Bases: object

Verifier: rclone check wrapper, no Python SHA pipeline.

Constructed with an RcloneDriver; production callers pass the same driver instance the push path uses so subprocess settings (binary path, etc.) stay consistent. Tests can pass a stub driver whose check returns a canned CheckResult.

Parameters:: driver (RcloneDriver | None)

async verify(run_path, remote, *, files_from)[source]#

Run rclone check --download over files_from and translate.

Raises TransportError from the driver only on a spawn failure (the rclone binary is missing); every other failure mode – auth / network / hash-mismatch – is folded into the returned VerifyResult.

Parameters:

run_path (Path)
remote (str)
files_from (Path)

Return type:

VerifyResult

class exlab_wizard.sync.VerifyResult(ok, mismatched=(), missing=(), extra=(), errors=(), error_kind=None, verified=())[source]#

Bases: object

Outcome of one rclone check --download pass against a remote.

ok is True iff no files differ, none are missing on the destination, and the rclone subprocess reported no per-file errors. Files in mismatched / missing / errors are run-relative POSIX paths drawn from the --combined output. extra lists files present on the destination but not in the source – it does not flip ok (it is informational, mirroring the pre-migration contract).

error_kind is set when the rclone subprocess itself failed (auth / network / unknown) before producing usable combined output. The queue worker keys off this field to route through the spec §7.1.5 retry policy.

Parameters:

ok (bool)
mismatched (tuple[str, ...])
missing (tuple[str, ...])
extra (tuple[str, ...])
errors (tuple[str, ...])
error_kind (TransportErrorKind | None)
verified (tuple[str, ...])

error_kind: TransportErrorKind | None#

errors: tuple[str, ...]#

extra: tuple[str, ...]#

classmethod from_check_result(check_result)[source]#

Translate a successful rclone check into a VerifyResult.

ok is True iff nothing differs, nothing is missing on the destination, and rclone reported no per-file errors. extra (present on the destination only) is carried for reporting but does not flip ok. The single source of truth for the CheckResult -> VerifyResult mapping shared by the queue worker, the force_verify path, and Verifier.verify().

Parameters:: check_result (CheckResult)
Return type:: VerifyResult

mismatched: tuple[str, ...]#

missing: tuple[str, ...]#

ok: bool#

verified: tuple[str, ...]#

exlab_wizard.sync.cleanup_interlocks_satisfied(*, job, run_path, now_utc, config, overrides_active)[source]#

Evaluate every §7.1.6 interlock; return True iff all pass.

Logs a debug entry naming the failing interlock when one fails so the operator can see why a job stayed in CLEANUP_ELIGIBLE.

The run_path parameter is accepted (but currently unused) so callers can pass the run directory through unchanged; future interlocks (e.g., size-on-disk threshold) may consult it.

Parameters:

job (SyncJobRow)
run_path (Any)
now_utc (datetime)
config (NASCleanupConfig)
overrides_active (list[dict[str, Any]])

Return type:

bool

exlab_wizard.sync.effective_bandwidth_limit_kibps(cfg, *, now_local)[source]#

Return the effective --bwlimit in KiB/s for now_local.

Decision tree per §7.1.7:

If cfg.upload_mbps is None -> unlimited (None).
Else if cfg.schedule is empty -> the cap applies always.
Else if now_local falls inside any schedule window -> the cap applies for this transfer.
Else -> unlimited (None).

Parameters:

cfg (BandwidthConfig)
now_local (datetime)

Return type:

int | None

exlab_wizard.sync.is_eligible(*, validator, creation_json_path, creation)[source]#

Evaluate the §7.3 eligibility rule for a run.

Returns (True, []) iff there is no hard-tier finding without an active override. Otherwise returns (False, blocking_findings) where blocking_findings is the list of unmasked hard-tier findings (so the caller can surface them in logs).

The creation_json_path is the path to .exlab-wizard/creation.json; the run directory is its parent’s parent. The validator runs in creation-time mode (no walk; just rules over path segments + file names + the cached creation payload).

Parameters:

validator (Validator)
creation_json_path (Path)
creation (CreationJson)

Return type:

tuple[bool, list[Finding]]

Modules

`bandwidth`	Bandwidth schedule evaluator.
`cleanup`	Cleanup safety interlocks.
`file_stability`	Pre-rclone file-stability guard: confirm files have stopped growing.
`manifest`	Parser for `rclone lsjson` output into a run-relative remote manifest.
`nas_client`	NAS sync client.
`pre_sync_gate`	Pre-Sync Gate.
`queue`	Durable SQLite-backed sync-job queue.
`run_delete`	Keep-local-aware, symlink-safe deletion of a run's staging copy.
`transports`	Sync transports package.
`verifier`	SHA-256 verifier wrapper around `rclone check --download`.