fix: stat signatures cover inode + ctime everywhere config identity is cached
The cherry-picked commit extends the config/profile/MCP/managed-scope/completer/ OAuth/skills-manifest signatures. This commit finishes the class and trims it: - `file_signature()` lives in `utils.py` next to the other stat/metadata helpers instead of `hermes_cli.managed_scope` (gateway/ and agent/ callers no longer reach into the managed-scope module for a generic stat helper). - `hermes_cli/config_effective.py` was left comparing 2-/4-wide prefixes against the widened `_RAW_CONFIG_CACHE` / `_load_config_cache_sig` records, so `load_user_config_effective()` re-parsed on every call (3 parses for 3 calls on an unchanged file, 1 before); index by the new widths. - Sibling caches keyed on the same (mtime, size) shape and reading the SAME files now use the helper: `load_env()` memo, `agent/skill_utils` raw-config and external-dirs caches, `hermes_cli/model_switch` alias identity, `agent/moa_loop` preset stamp, `hermes_cli/auth` global auth-store memo. - Tests trimmed to one invariant each (pinned-mtime replacement invalidates; an unchanged file still hits), both red on origin/main. Left alone on purpose: `tools/registry.py`, `tools/skills_tool_dedup.py`, `gateway/status.py`, `hermes_cli/banner.py`, `hermes_cli/main.py`, `hermes_cli/session_recovery.py` — those fingerprint source files, PID/lock files or write to persisted on-disk caches shared across processes, where an inode/ctime key would churn on every checkout/copy rather than catch a replaced config.
This commit is contained in:
teknium1
+3
-2
@@ -120,12 +120,13 @@ _preset_cache: dict[tuple, Any] = {}
|
||||
|
||||
|
||||
def _resolve_preset_cached(preset_name: str) -> tuple[dict[str, Any], Any]:
|
||||
"""``(preset, raw moa config)``; the resolved preset is cached per config mtime
|
||||
"""``(preset, raw moa config)``; the resolved preset is cached per config file signature
|
||||
(skips resolve_moa_preset's full validation of the moa block on every create())."""
|
||||
from hermes_cli.config import get_config_path, load_config
|
||||
from hermes_cli.moa_config import resolve_moa_preset
|
||||
from utils import file_signature
|
||||
try:
|
||||
cfg_stamp = get_config_path().stat().st_mtime_ns
|
||||
cfg_stamp = file_signature(get_config_path().stat())
|
||||
except OSError:
|
||||
cfg_stamp = None
|
||||
moa_raw = load_config().get("moa") or {}
|
||||
|
||||
@@ -28,7 +28,7 @@ from agent.skill_utils import (
|
||||
skill_matches_platform, skill_matches_platform_list,
|
||||
)
|
||||
from tools.threat_patterns import scan_for_threats as _scan_for_threats
|
||||
from utils import atomic_json_write
|
||||
from utils import atomic_json_write, file_signature
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -1091,8 +1091,6 @@ def clear_skills_system_prompt_cache(*, clear_snapshot: bool = False) -> None:
|
||||
def _build_skills_manifest(skills_dir: Path) -> dict[str, list[int]]:
|
||||
"""File-signature manifest of every SKILL.md and DESCRIPTION.md; only the ACTIVE org mirror participates, and
|
||||
the ``.active_org`` marker is included so switching/leaving an org invalidates the snapshot by itself."""
|
||||
from hermes_cli.managed_scope import file_signature
|
||||
|
||||
manifest: dict[str, list[int]] = {}
|
||||
skills_dir_str = str(skills_dir)
|
||||
prefix_len = len(os.path.join(skills_dir_str, ""))
|
||||
|
||||
@@ -208,7 +208,7 @@ def skill_matches_environment(frontmatter: Dict[str, Any]) -> bool:
|
||||
return any(_detect_environment(tag) for tag in tags if tag)
|
||||
|
||||
|
||||
_RAW_CONFIG_CACHE: Dict[Tuple[str, int, int], Dict[str, Any]] = {}
|
||||
_RAW_CONFIG_CACHE: Dict[Tuple[str, int, int, int, int], Dict[str, Any]] = {}
|
||||
|
||||
|
||||
def _raw_config_cache_clear() -> None:
|
||||
@@ -216,11 +216,11 @@ def _raw_config_cache_clear() -> None:
|
||||
_RAW_CONFIG_CACHE.clear()
|
||||
|
||||
|
||||
def _config_cache_key(config_path: Path) -> Optional[Tuple[str, int, int]]:
|
||||
"""``(path, mtime_ns, size)`` identity of config.yaml, or None when unreadable/absent."""
|
||||
def _config_cache_key(config_path: Path) -> Optional[Tuple[str, int, int, int, int]]:
|
||||
"""``(path, *file_signature)`` identity of config.yaml, or None when unreadable/absent."""
|
||||
try:
|
||||
stat = config_path.stat()
|
||||
return (str(config_path), stat.st_mtime_ns, stat.st_size)
|
||||
from utils import file_signature
|
||||
return (str(config_path), *file_signature(config_path.stat()))
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
@@ -316,7 +316,7 @@ def _normalize_string_set(values) -> Set[str]:
|
||||
|
||||
# config identity -> resolved external dirs. Called once per skill during
|
||||
# banner / tool-registry scans; re-resolving each time dominated cold-start.
|
||||
_EXTERNAL_DIRS_CACHE: Dict[Tuple[str, int], List[Path]] = {}
|
||||
_EXTERNAL_DIRS_CACHE: Dict[Tuple[str, int, int, int, int], List[Path]] = {}
|
||||
|
||||
|
||||
def _external_dirs_cache_clear() -> None:
|
||||
@@ -341,7 +341,7 @@ def get_external_skills_dirs() -> List[Path]:
|
||||
if not config_path.exists():
|
||||
return []
|
||||
full_key = _config_cache_key(config_path)
|
||||
cache_key = full_key[:2] if full_key is not None else None
|
||||
cache_key = full_key
|
||||
cached = _EXTERNAL_DIRS_CACHE.get(cache_key) if cache_key is not None else None
|
||||
if cached is not None:
|
||||
return list(cached) # copy so callers can't mutate the cache
|
||||
|
||||
@@ -22,7 +22,7 @@ from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from gateway.run_shutdown import _log_suppressed
|
||||
from hermes_cli.managed_scope import file_signature
|
||||
from utils import file_signature
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
+3
-3
@@ -32,7 +32,7 @@ from hermes_cli.config import (
|
||||
get_hermes_home, get_config_path, read_raw_config, require_readable_config_before_write)
|
||||
from hermes_constants import OPENROUTER_BASE_URL, hermes_home_key, secure_parent_dir
|
||||
from agent.credential_persistence import sanitize_borrowed_credential_payload
|
||||
from utils import atomic_json_write, atomic_yaml_write, env_float, is_truthy_value # noqa: F401 (env_float: agent.credential_pool reads auth_mod.env_float)
|
||||
from utils import atomic_json_write, atomic_yaml_write, env_float, file_signature, is_truthy_value # noqa: F401 (env_float: agent.credential_pool reads auth_mod.env_float)
|
||||
from hermes_cli.auth_zai_kimi import ( # noqa: F401 re-exported
|
||||
KIMI_CODE_BASE_URL, ZAI_ENDPOINTS, _normalize_lmstudio_runtime_base_url, _resolve_kimi_base_url,
|
||||
_resolve_zai_base_url, detect_zai_endpoint)
|
||||
@@ -501,8 +501,8 @@ def _load_global_auth_store() -> Dict[str, Any]:
|
||||
_global_auth_store_cache = None
|
||||
return {}
|
||||
try:
|
||||
cache_key: Optional[Tuple[str, int]] = (
|
||||
str(global_path.resolve(strict=False)), global_path.stat().st_mtime_ns)
|
||||
cache_key: Optional[Tuple[str, Tuple[int, int, int, int]]] = (
|
||||
str(global_path.resolve(strict=False)), file_signature(global_path.stat()))
|
||||
except Exception:
|
||||
cache_key = None
|
||||
cached = _global_auth_store_cache
|
||||
|
||||
@@ -12,7 +12,7 @@ import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from hermes_cli.auth_constants import _decode_jwt_claims
|
||||
from hermes_cli.managed_scope import file_signature
|
||||
from utils import file_signature
|
||||
|
||||
# Log-record parity with the origin module (caplog tests pin "hermes_cli.auth").
|
||||
logger = logging.getLogger("hermes_cli.auth")
|
||||
|
||||
@@ -30,7 +30,7 @@ def _personalities_from_cli_config() -> Dict[str, Any]:
|
||||
Falls back to a fresh load when the file cannot be stat'ed."""
|
||||
global _personalities_memo
|
||||
from cli import load_cli_config
|
||||
from hermes_cli.managed_scope import file_signature
|
||||
from utils import file_signature
|
||||
from hermes_cli.personality import available_personalities
|
||||
try:
|
||||
from hermes_cli.config import get_config_path
|
||||
|
||||
@@ -30,7 +30,7 @@ from hermes_cli.default_soul import DEFAULT_SOUL_MD, is_legacy_template_soul
|
||||
from hermes_cli.secret_prompt import masked_secret_prompt
|
||||
# Re-export from hermes_constants — canonical definition lives there.
|
||||
from hermes_constants import get_hermes_home, get_process_hermes_home # noqa: F401
|
||||
from utils import atomic_replace, atomic_yaml_write, fast_safe_load
|
||||
from utils import atomic_replace, atomic_yaml_write, fast_safe_load, file_signature
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -93,7 +93,7 @@ def _warn_config_parse_failure(
|
||||
"""
|
||||
try:
|
||||
st = config_path.stat()
|
||||
sig = managed_scope.file_signature(st)
|
||||
sig = file_signature(st)
|
||||
key = (str(config_path), *sig)
|
||||
_CONFIG_PARSE_FAILURES[str(config_path)] = (*sig, str(exc))
|
||||
except OSError:
|
||||
@@ -120,7 +120,7 @@ def get_active_config_parse_failure() -> Optional[str]:
|
||||
try:
|
||||
record = _CONFIG_PARSE_FAILURES[str(path := get_config_path())]
|
||||
st = path.stat()
|
||||
return record[4] if managed_scope.file_signature(st) == record[:4] else None
|
||||
return record[4] if file_signature(st) == record[:4] else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@@ -1901,7 +1901,7 @@ def _read_raw_config_impl(*, want_deepcopy: bool) -> Dict[str, Any]:
|
||||
try:
|
||||
config_path = get_config_path()
|
||||
st = config_path.stat()
|
||||
cache_key = managed_scope.file_signature(st)
|
||||
cache_key = file_signature(st)
|
||||
except (FileNotFoundError, OSError):
|
||||
return {}
|
||||
|
||||
@@ -2145,13 +2145,13 @@ def _load_config_cache_sig(config_path: Path) -> Tuple[Optional[Tuple[int, int,
|
||||
the merged result. ``cache_sig`` is None only when neither file exists (nothing to cache on)."""
|
||||
try:
|
||||
st = config_path.stat()
|
||||
user_sig: Optional[Tuple[int, int, int, int]] = managed_scope.file_signature(st)
|
||||
user_sig: Optional[Tuple[int, int, int, int]] = file_signature(st)
|
||||
except FileNotFoundError:
|
||||
user_sig = None
|
||||
managed_dir = managed_scope.get_managed_dir()
|
||||
try:
|
||||
mst = (managed_dir / "config.yaml").stat() if managed_dir else None
|
||||
managed_sig = managed_scope.file_signature(mst) if mst else (0, 0, 0, 0)
|
||||
managed_sig = file_signature(mst) if mst else (0, 0, 0, 0)
|
||||
except OSError:
|
||||
managed_sig = (0, 0, 0, 0)
|
||||
if user_sig is None and managed_sig == (0, 0, 0, 0):
|
||||
@@ -2390,7 +2390,7 @@ def save_config(
|
||||
_LAST_EXPANDED_CONFIG_BY_PATH[str(config_path)] = copy.deepcopy(current_normalized)
|
||||
|
||||
|
||||
# load_env() memo keyed on (path, mtime, size). Editing .env bumps mtime -> rebuild;
|
||||
# load_env() memo keyed on (path, *file_signature). Editing .env bumps mtime/inode -> rebuild;
|
||||
# invalidate_env_cache() is the explicit knob for writers on coarse-mtime filesystems.
|
||||
_env_cache: Optional[Tuple[Tuple[str, Optional[float], Optional[int]], Dict[str, str]]] = None
|
||||
|
||||
@@ -2403,9 +2403,9 @@ def load_env() -> Dict[str, str]:
|
||||
|
||||
try:
|
||||
st = env_path.stat()
|
||||
cache_key = (str(env_path), st.st_mtime, st.st_size)
|
||||
cache_key = (str(env_path), file_signature(st))
|
||||
except FileNotFoundError:
|
||||
cache_key = (str(env_path), None, None)
|
||||
cache_key = (str(env_path), None)
|
||||
except Exception:
|
||||
cache_key = None
|
||||
if cache_key is not None and _env_cache is not None and _env_cache[0] == cache_key:
|
||||
|
||||
@@ -25,8 +25,8 @@ from utils import fast_safe_load
|
||||
# path -> raw user mapping from the last successful parse in this process; served (through the
|
||||
# normal pipeline) when the file is later found mid-edit as broken YAML.
|
||||
_LAST_GOOD_USER_RAW: Dict[str, Dict[str, Any]] = {}
|
||||
# path -> (user_mtime_ns, user_size, managed_mtime_ns, managed_size, effective, env_snapshot).
|
||||
_EFFECTIVE_CACHE: Dict[str, Tuple[int, int, int, int, Dict[str, Any], Dict[str, Optional[str]]]] = {}
|
||||
# path -> (*user_signature, *managed_signature, effective, env_snapshot); see utils.file_signature.
|
||||
_EFFECTIVE_CACHE: Dict[str, Tuple[Any, ...]] = {}
|
||||
|
||||
|
||||
def _effective(raw: Dict[str, Any]) -> Dict[str, Any]:
|
||||
@@ -65,15 +65,15 @@ def load_user_config_effective(config_path: Optional[Path] = None, *, fail_close
|
||||
with _config._CONFIG_LOCK:
|
||||
user_sig, cache_sig = _config._load_config_cache_sig(config_path)
|
||||
cached = _EFFECTIVE_CACHE.get(path_key)
|
||||
if cached is not None and cache_sig is not None and cached[:4] == cache_sig:
|
||||
if all(_config._env_ref_lookup(k) == v for k, v in cached[5].items()):
|
||||
return copy.deepcopy(cached[4])
|
||||
if cached is not None and cache_sig is not None and cached[:8] == cache_sig:
|
||||
if all(_config._env_ref_lookup(k) == v for k, v in cached[9].items()):
|
||||
return copy.deepcopy(cached[8])
|
||||
|
||||
raw: Dict[str, Any] = {}
|
||||
recovered = False
|
||||
raw_hit = _config._RAW_CONFIG_CACHE.get(path_key)
|
||||
if user_sig is not None and raw_hit is not None and raw_hit[:2] == user_sig:
|
||||
raw = copy.deepcopy(raw_hit[2]) # one parse per process, shared with read_raw_config()
|
||||
if user_sig is not None and raw_hit is not None and raw_hit[:4] == user_sig:
|
||||
raw = copy.deepcopy(raw_hit[4]) # one parse per process, shared with read_raw_config()
|
||||
_LAST_GOOD_USER_RAW.setdefault(path_key, copy.deepcopy(raw))
|
||||
elif user_sig is not None:
|
||||
try:
|
||||
|
||||
@@ -13,20 +13,11 @@ import logging
|
||||
import os
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Dict, Optional, Tuple
|
||||
from typing import Dict, Optional
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
def file_signature(st: "os.stat_result") -> Tuple[int, int, int, int]:
|
||||
"""Stat signature for cache invalidation: ``(st_mtime_ns, st_size, st_ino, st_ctime_ns)``.
|
||||
|
||||
``mtime_ns`` + ``size`` alone miss replacements that preserve both (``cp -p``,
|
||||
``rsync -t``, timestamp-pinning scripts, sync clients). ``st_ino`` changes on an
|
||||
atomic replace (fresh inode); ``st_ctime_ns`` cannot be backdated via ``os.utime``,
|
||||
catching writers that pin mtime. macOS and Linux both expose these fields.
|
||||
"""
|
||||
return (st.st_mtime_ns, st.st_size, st.st_ino, st.st_ctime_ns)
|
||||
from utils import file_signature
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -34,7 +25,7 @@ logger = logging.getLogger(__name__)
|
||||
_DEFAULT_MANAGED_DIR = Path("/etc/hermes")
|
||||
|
||||
_CACHE_LOCK = threading.Lock()
|
||||
# path_key -> (mtime_ns, size, parsed)
|
||||
# path_key -> (*file_signature, parsed)
|
||||
_CONFIG_CACHE: Dict[str, tuple] = {}
|
||||
_ENV_CACHE: Dict[str, tuple] = {}
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ from hermes_cli.providers import (
|
||||
from hermes_cli.model_normalize import normalize_model_for_provider
|
||||
from agent.models_dev import (
|
||||
ModelCapabilities, ModelInfo, get_model_capabilities, get_model_info, list_provider_models)
|
||||
from utils import base_url_host_matches, base_url_hostname, base_url_origin
|
||||
from utils import base_url_host_matches, base_url_hostname, base_url_origin, file_signature
|
||||
# Re-exported: callers/tests patch hermes_cli.model_switch.<name>.
|
||||
from hermes_cli.model_switch_providers import list_authenticated_providers
|
||||
|
||||
@@ -270,10 +270,10 @@ def _direct_alias_source_identity() -> Optional[tuple]:
|
||||
stat = path.stat()
|
||||
except OSError:
|
||||
# A missing config is still a definite identity for this profile.
|
||||
return (str(path), None, None)
|
||||
return (str(path), None)
|
||||
except Exception:
|
||||
return None
|
||||
return (str(path), stat.st_mtime_ns, stat.st_size)
|
||||
return (str(path), file_signature(stat))
|
||||
|
||||
|
||||
def _ensure_direct_aliases() -> None:
|
||||
|
||||
@@ -1,27 +1,20 @@
|
||||
"""Regression: the profile re-scan watcher must detect file *replacements*.
|
||||
|
||||
``profile_serve_signature`` keyed files on ``(st_mtime_ns, st_size)`` only, so a
|
||||
replacement preserving both (``cp -p``, ``rsync -t``, timestamp-pinning) never
|
||||
triggered a profile re-scan and adapters were not rebuilt until restart.
|
||||
|
||||
See #111105.
|
||||
"""
|
||||
"""#111105: the profile re-scan watcher must rebuild adapters after a config.yaml replacement
|
||||
that keeps mtime and size (``cp -p``, ``rsync -t``, a timestamp-pinning writer)."""
|
||||
import os
|
||||
import shutil
|
||||
|
||||
from gateway.run_profile_reconcile import profile_serve_signature
|
||||
|
||||
def test_profile_serve_signature_detects_replacement_with_pinned_mtime(tmp_path):
|
||||
from gateway.run_profile_reconcile import profile_serve_signature
|
||||
|
||||
def test_profile_serve_signature_changes_on_replacement_with_pinned_mtime(tmp_path):
|
||||
cfg = tmp_path / "config.yaml"
|
||||
cfg.write_text("x" * 64, encoding="utf-8")
|
||||
before = profile_serve_signature(tmp_path)
|
||||
assert profile_serve_signature(tmp_path) == before # unchanged file: stable
|
||||
st = cfg.stat()
|
||||
other = tmp_path / "other"
|
||||
other.write_text("y" * 64, encoding="utf-8")
|
||||
shutil.copy2(other, cfg)
|
||||
os.utime(cfg, ns=(st.st_atime_ns, st.st_mtime_ns))
|
||||
after = cfg.stat()
|
||||
assert after.st_mtime_ns == st.st_mtime_ns
|
||||
assert after.st_size == st.st_size
|
||||
assert (cfg.stat().st_mtime_ns, cfg.stat().st_size) == (st.st_mtime_ns, st.st_size)
|
||||
assert profile_serve_signature(tmp_path) != before
|
||||
|
||||
@@ -1,39 +1,28 @@
|
||||
"""Regression: config cache must detect file *replacements*.
|
||||
|
||||
A replacement that preserves ``st_mtime_ns`` and ``st_size`` (``cp -p``,
|
||||
``rsync -t``, timestamp-pinning scripts, sync clients) was treated as
|
||||
unchanged, so long-lived processes kept serving stale config until restart.
|
||||
The signature now also covers ``st_ino`` (fresh inode on atomic replace) and
|
||||
``st_ctime_ns`` (cannot be backdated via ``os.utime``).
|
||||
|
||||
See #111105.
|
||||
"""
|
||||
"""#111105: a config.yaml replacement that keeps mtime and size (``cp -p``, ``rsync -t``, a
|
||||
timestamp-pinning writer) must still invalidate the load_config() cache, while an unchanged
|
||||
file keeps serving the cached object."""
|
||||
import os
|
||||
import shutil
|
||||
from unittest.mock import patch
|
||||
|
||||
from hermes_cli import config as config_mod
|
||||
from hermes_cli.config import load_config
|
||||
|
||||
A = "model:\n provider: opencode-go\n default: aaaa-route\n"
|
||||
B = "model:\n provider: opencode-go\n default: bbbb-route\n"
|
||||
|
||||
|
||||
def test_load_config_detects_replacement_with_pinned_mtime(tmp_path):
|
||||
def _replace_pinning_mtime(path, content: str) -> None:
|
||||
before = path.stat()
|
||||
other = path.with_name("other.yaml")
|
||||
other.write_text(content, encoding="utf-8")
|
||||
shutil.copy2(other, path)
|
||||
os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
|
||||
|
||||
|
||||
def test_load_config_sees_replacement_with_pinned_mtime_and_size(tmp_path):
|
||||
with patch.dict(os.environ, {"HERMES_HOME": str(tmp_path)}):
|
||||
config_mod._LOAD_CONFIG_CACHE.clear()
|
||||
config_mod._RAW_CONFIG_CACHE.clear()
|
||||
cfg = tmp_path / "config.yaml"
|
||||
cfg.write_text(A, encoding="utf-8")
|
||||
assert load_config()["model"]["default"] == "aaaa-route"
|
||||
# Replace content, keep st_mtime_ns + st_size identical.
|
||||
before = cfg.stat()
|
||||
other = tmp_path / "other.yaml"
|
||||
other.write_text(B, encoding="utf-8")
|
||||
shutil.copy2(other, cfg)
|
||||
os.utime(cfg, ns=(before.st_atime_ns, before.st_mtime_ns))
|
||||
after = cfg.stat()
|
||||
assert after.st_mtime_ns == before.st_mtime_ns
|
||||
assert after.st_size == before.st_size
|
||||
assert cfg.read_text(encoding="utf-8").split("default: ")[1].strip() == "bbbb-route"
|
||||
assert load_config()["model"]["default"] == "bbbb-route"
|
||||
cfg.write_text("model:\n default: aaaa-route\n", encoding="utf-8")
|
||||
first = config_mod._load_config_impl(want_deepcopy=False)
|
||||
assert config_mod._load_config_impl(want_deepcopy=False) is first # unchanged file: cache hit
|
||||
_replace_pinning_mtime(cfg, "model:\n default: bbbb-route\n")
|
||||
assert config_mod.load_config()["model"]["default"] == "bbbb-route"
|
||||
|
||||
@@ -35,6 +35,18 @@ def env_var_enabled(name: str, default: str = "") -> bool:
|
||||
return is_truthy_value(os.getenv(name, default), default=False)
|
||||
|
||||
|
||||
def file_signature(st: os.stat_result) -> "tuple[int, int, int, int]":
|
||||
"""Change-detection key for a stat result: ``(st_mtime_ns, st_size, st_ino, st_ctime_ns)``.
|
||||
|
||||
mtime + size alone miss a replacement that preserves both (``cp -p``, ``rsync -t``, a tar
|
||||
restore, a script pinning the timestamp with ``os.utime``). The inode changes on an atomic
|
||||
replace and ctime cannot be backdated from user space, so the pair catches those writers.
|
||||
On Windows ``st_ino`` may be 0 and ``st_ctime_ns`` is the creation time — both stable across
|
||||
an in-place rewrite, so the key degrades to mtime + size there rather than misfiring.
|
||||
"""
|
||||
return (st.st_mtime_ns, st.st_size, st.st_ino, st.st_ctime_ns)
|
||||
|
||||
|
||||
def _preserve_file_mode(path: Path) -> "int | None":
|
||||
"""Permission bits of *path* if it exists, else ``None``."""
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user