cadrumo.core.paths module

Shared path normalization helpers.

Centralises the small set of Path primitives that every other Cadrumo module needs: resolving relative operator paths against the application-data anchor via resolve_project_path(), normalising settings with normalize_project_relative_path().

These helpers validate and compose paths only. They do not read, write, create, or secure files; persistence adapters that need registered storage errors own their typed containment layer.

is_windows_long_path_error(), windows_long_paths_enabled(), and windows_storage_root_long_path_margin() are the Windows MAX_PATH (260-character) hardening surface: classifying an OSError that legacy Windows raises once a resolved path exceeds the limit, reading the machine-wide long-path opt-in, and computing whether a candidate storage root leaves enough headroom for the deepest object path the caller’s storage layout can produce. The worst-case suffix that last helper measures against is supplied by its caller: this module owns the arithmetic, never the on-disk grammar of a layer above it.

WINDOWS_MAX_PATH: int

Legacy Windows CreateFileW path-length ceiling in UTF-16 code units. Windows 10 1607+ can lift this per-application via the LongPathsEnabled registry value combined with a longPathAware application manifest; a workstation that has neither still enforces this ceiling.

is_windows_long_path_error(exc)[source]

Return whether exc is a Windows path-length-ceiling failure.

Classifies WinError 3 (ERROR_PATH_NOT_FOUND) and WinError 206 (ERROR_FILENAME_EXCED_RANGE) — the two concrete Windows API error codes a legacy (non long-path-aware) workstation raises once a resolved path walks past WINDOWS_MAX_PATH. Always False on non-Windows platforms and for any other OSError, so callers can unconditionally probe every caught OSError without a platform guard of their own.

Parameters:

exc (OSError) – The caught OSError (or subclass, e.g. FileNotFoundError) to classify.

Return type:

bool

Returns:

True when exc.winerror names a known long-path failure.

windows_long_paths_enabled()[source]

Report the machine-wide Windows long-path opt-in, if determinable.

Reads HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\FileSystem \LongPathsEnabled — the registry value Windows 10 1607+ consults to lift WINDOWS_MAX_PATH for manifest-declared long-path-aware applications (this CLI is built with a longPathAware manifest via its packaging).

Return type:

bool | None

Returns:

True when the value is present and non-zero, False when present and zero (or when the platform is Windows but the value is absent — the pre-1607 / not-yet-opted-in default), and None on a non-Windows platform where the concept does not apply, or when the registry cannot be read at all (a probe is best-effort; it never raises).

windows_storage_root_long_path_margin(root, *, object_path_suffix_length)[source]

Return the headroom, in characters, before an object write risks MAX_PATH.

Computes WINDOWS_MAX_PATH - len(str(root.resolve())) - object_path_suffix_length. A positive result is the number of characters of slack remaining; zero or negative means the deepest object the caller’s layout can write already meets or exceeds the legacy MAX_PATH ceiling from this root. Platform-independent by design: the margin is informative on every OS, but only Windows without the long-path opt-in enforces the ceiling it measures against.

The worst-case suffix is a required argument rather than a constant of this module. It was previously a literal here, spelling one hand-picked <namespace> sample into the arithmetic because core sits below the layers that own the on-disk grammar. That sample was drawn from the wrong vocabulary entirely: the <namespace> segment of a provider object path is a registered secure-object storage namespace, and the literal named a bucket-event object type, none of whose values is a registered namespace. The real longest namespace is materially longer, so the resulting ceiling understated every margin this function returned. Taking the budget from the caller lets the owning layer derive it from its own registry — structurally, over the whole domain — without inverting the dependency graph. See windows_worst_case_object_path_suffix_length().

Parameters:

  • root (Path) – The candidate storage root (cadrumo_local_storage_root or an outbound-storage provider root).

  • object_path_suffix_length (int) – Length of the deepest path suffix (leading separator through file extension) the caller’s layout can append below root.

Return type:

int

Returns:

The signed character margin described above.

Raises:

CoreValidationError – When object_path_suffix_length is not a positive length, which would silently inflate the margin.

resolve_project_path(value, *, state_root_inputs=None)[source]

Resolve a relative path against the application-data anchor.

Absolute paths are returned as absolute resolved paths. Relative paths are interpreted against _relative_path_anchor() — never the process cwd, which keeps config defaults stable regardless of where the CLI process starts, and never a bare checkout-root walk, which keeps an installed run’s relative override out of a virtualenv or ephemeral package cache. This helper is not a containment guard.

Parameters:

  • value (str | Path) – An absolute or relative path; user-style ~ references are expanded.

  • state_root_inputs (StateRootInputs | None) – Optional injectable StateRootInputs seam forwarded to _relative_path_anchor(). None (the default) captures the live process’s inputs.

Return type:

Path

Returns:

The fully resolved absolute pathlib.Path.

normalize_project_relative_path(value, *, state_root_inputs=None)[source]

Normalise an optional path setting to an absolute, data-root-anchored path.

Used by settings validators for optional path fields. It preserves None and delegates path semantics to resolve_project_path(); it does not verify that the resulting path exists.

Parameters:
Return type:

Path | None

Returns:

None when value is None; otherwise the resolved absolute path produced by resolve_project_path().

effective_storage_root(root=None, *, settings=None, state_root_inputs=None)[source]

Return the effective Cadrumo storage root: a caller override, or the settings default.

The single accessor for the “an explicit root override wins, otherwise fall back to Settings.cadrumo_local_storage_root” fallback that six call sites across application/user_profile and application/config_reset_repository.py each re-implemented inline. Comparing the six copies surfaced real drift, not just duplication: one normalised an override by calling bare resolve(), which for a relative override resolves against the process’s current working directory rather than anchoring it the way every other relative operator path in this codebase does; the other four returned an explicit override completely unnormalised — a relative override, or one carrying ~, passed straight through un-expanded and un-resolved. Both are defects: a login or profile-repository root must compare identically regardless of the directory the process happened to start from, and an unnormalised override risks a cross-platform identity mismatch (a differently-cased or non-canonical path failing to compare equal to itself resolved a second time elsewhere).

An explicit root is therefore always normalised through resolve_project_path(): an absolute override resolves as-is (including Windows on-disk casing); a relative override anchors under the platform user-data root — one level above the settings default’s own storage/ — never under the current working directory and never nested under an already-derived storage root. root is None falls back to Settings.cadrumo_local_storage_root, which the settings field validator already normalises identically on load, so no second resolve is spent on the common (no-override) path.

Parameters:

  • root (Path | None) – Optional caller-supplied storage-root override. None (the default) resolves Settings.cadrumo_local_storage_root.

  • settings (Settings | None) – Optional already-resolved Settings, read only when root is None, so a caller that already holds a Settings instance need not trigger a second load_settings(). None (the default) loads the live settings.

  • state_root_inputs (StateRootInputs | None) – Optional injectable StateRootInputs seam forwarded to resolve_project_path() when root is supplied. None (the default) captures the live process’s inputs.

Return type:

Path

Returns:

The resolved absolute storage root.

file_stat_fingerprint(path)[source]

Return a cache-key fingerprint triple for a single file.

The triple (name, size_bytes, mtime_ns) is a stable, low-cost proxy for file identity used by file-backed loader caches. Any in-place modification that changes size or mtime invalidates the cache without requiring a full content hash.

This is not an integrity hash or evidence digest. For byte-level verification use cadrumo.core.hashing.hash_file() or cadrumo.core.hashing.sha256_file().

Parameters:

path (Path) – The file to fingerprint. Must be an existing, stat-able path.

Return type:

tuple[str, int, int]

Returns:

(path.name, stat.st_size, stat.st_mtime_ns). path.stat() propagates OSError when the file is unreadable or disappears.

path_stat_fingerprint(path)[source]

Return a cache-key fingerprint triple for a single file, keyed by its full path.

The path-keyed sibling of file_stat_fingerprint(). That function’s triple starts with the bare file name, which is correct only when every candidate is known to come from the same directory (a tree-walk fingerprint); reused as a global cache key spanning multiple directories, a name-only first element collides across unrelated files that happen to share a filename. This triple starts with str(path) instead, so it stays unique across directories. Use file_stat_fingerprint() for a same-directory tree fingerprint (its shorter name-only key is the correct, and cheaper, choice there); use this one for a single-file loader cache keyed on a resolved, possibly cross-directory path.

Parameters:

path (Path) – The file to fingerprint. Must be an existing, stat-able path.

Return type:

tuple[str, int, int]

Returns:

(str(path), stat.st_size, stat.st_mtime_ns). path.stat() propagates OSError when the file is unreadable or disappears.

directory_byte_total(directory, *, tolerate_errors=False, entries=None)[source]

Return (total_bytes, file_count) for every regular file under directory.

A missing (or not-a-directory) directory reports (0, 0) rather than raising, mirroring the shape every prior hand-rolled walker used.

Parameters:

  • directory (Path) – The directory to sum. Only used to short-circuit on absence when entries is not supplied; the actual walk reads entries (or a fresh recursive scan of directory).

  • tolerate_errors (bool) –

    When False (the default), an OSError raised while statting a candidate propagates, as does one raised while advancing a caller-supplied entries iterator. When True, the failing candidate — or the rest of an interrupted walk — is skipped and the total reflects only what was successfully stat’d, so a blob write racing this read never crashes the caller.

    The flag does not reach the internal scan’s own descent: an unreadable subdirectory is skipped silently whatever it is set to, the scan reporting what it could read rather than raising from beneath the root. That predates the move off Path.rglob, which swallowed the same errors — so an injected entries is the only walk advance this flag governs.

  • entries (Iterable[Path] | None) – Pre-enumerated candidates to sum instead of a fresh recursive scan of directory. Injectable so a caller that already holds an enumeration can reuse it, and so a test can reproduce a file vanishing mid-walk deterministically (list the candidate, delete it on disk, then call this function with that stale reference).

Return type:

tuple[int, int]

Returns:

(total_bytes, file_count) summed over every regular file encountered. Directories and other non-regular entries among the candidates are silently skipped (not an error).

select_filesystem_retention_survivors(entries, *, timestamp, cutoff=None, max_count=None, max_total_bytes=None, size_fn=None, combine='sequential', protect_newest=0)[source]

Select survivors/removals under composable age, count, and byte bounds.

Mirrors select_retention_removal_keys()’s pure rank-and-bound shape, generalized to a filesystem entry (a run directory, a dump file, a telemetry file, a compiled-cache pickle) instead of a secure-object key, and widened from that primitive’s fixed cutoff-then-count pipeline to composable, independently-optional bounds. Deletion stays with the caller: this function only decides who survives.

Entries are ranked newest-first by timestamp (a stable sort, so entries sharing a timestamp keep their relative input order — pass entries pre-sorted by a secondary key when that tie-break matters, e.g. filename). protect_newest exempts the top-ranked N entries from removal under every bound, while still counting them toward the max_total_bytes running total and toward rank positions in combine="union" mode.

Parameters:

  • entries (Sequence[TypeVar(EntryT)]) – Candidates to rank and select over. Order does not matter; the function re-ranks internally.

  • timestamp (Callable[[TypeVar(EntryT)], TypeVar(TimestampT, bound= _RetentionTimestamp)]) – Projection returning each entry’s retention timestamp.

  • cutoff (Optional[TypeVar(TimestampT, bound= _RetentionTimestamp)]) – Age boundary. An entry strictly older than cutoff is a removal candidate; an entry exactly at cutoff survives.

  • max_count (int | None) – Maximum surviving entries under the rank/count bound.

  • max_total_bytes (int | None) – Total-size ceiling; requires size_fn. Oldest surviving entries are dropped until the running total fits.

  • size_fn (Callable[[TypeVar(EntryT)], int] | None) – Per-entry byte size, required when max_total_bytes is set.

  • combine (Literal['sequential', 'union']) – "sequential" (default) applies cutoff, then max_count, then max_total_bytes as successive stages, each narrowing the previous stage’s survivors — mirrors select_retention_removal_keys()’s cutoff-then-count shape, extended with a byte-total stage. "union" evaluates cutoff and max_count independently against the full (unstaged) ranking and removes an entry matching either — the shape a “keep the newest N sessions, otherwise prune by age or rank” policy needs, where a rank-3 entry within the age window still gets removed for being beyond the count bound. "union" does not support max_total_bytes (a byte ceiling has no rank-independent per-entry membership test).

  • protect_newest (int) – The top-ranked N entries survive unconditionally (still counted toward max_total_bytes and toward rank positions in "union" mode).

Return type:

tuple[list[TypeVar(EntryT)], list[TypeVar(EntryT)]]

Returns:

(keep, remove), both drawn from entries with no entry in both.

Raises:

ValueError – When no bound is supplied, when max_total_bytes is supplied without size_fn, or when combine="union" is paired with max_total_bytes.