Reconcile admitted fragment inventory after restart

docs/artifact-admission.md

@@ -59,3 +59,40 @@ idempotent recovery, revision fencing, policy binding, conservative charges and
 private/exclusive file setup. Process-crash tests do not establish power-loss or
 multi-machine durability. Full repository/installed-package validation and
 independent review are required before publication or integration with hosting.
+
+## Read-only restart reconciliation
+
+`artifact_inventory.reconcile_inventory(ledger, selections, max_read_bytes=...)`
+rechecks every admitted request against a caller-selected absolute file path and
+manifest. `selections` is a tuple of `InventorySelection(request_id, manifest,
+path)`, with exactly one entry per admitted request, at most 192. The trusted
+ledger supplies each pinned manifest root, project, fragment index, length and
+content/transfer binding. All selections and the aggregate logical read budget
+(up to 64 MiB) are checked before any fragment is opened. The full charged amount
+must fit that budget even when some files are absent. Actual reads use the
+existing bounded 32 KiB verifier; the budget is not a total CPU/RSS or physical
+I/O claim. Work is synchronous and sequential; admission SQLite waits retain the
+existing five-second busy bound and are not asynchronously cancellable.
+
+A report identifies each request as `verified`, `missing` or `unverified`.
+Missing includes disappearance of a selected file or parent during inspection;
+unverified includes unsafe file type/path, observable mutation, bad bytes and
+filesystem errors. It is not an automatic repair or deletion decision. Cancellation
+callbacks may raise any exception: they abort the pass rather than being mistaken
+for a missing/corrupt file. No partial report is returned. A changed ledger
+snapshot also rejects the complete report, including changes committed by another
+process during the pass. Corrupt or unreadable ledger state fails closed.
+
+Prior `verified_observation` metadata is never accepted as evidence of current
+bytes. Reconciliation changes no records, revisions, charges or files, including
+stale observations. Callers must use the fresh report for their recovery decision;
+this API does not persist availability or grant a serving capability. Observations
+are made at different times and are not an atomic filesystem snapshot or a
+future retention promise. Reverify before serving or recovery use. Publisher
+approval remains separate. Trusted stable physical directories remain required;
+hostile same-user writers are outside the guarantee.
+
+Unknown/orphan files are not enumerated or charged by this API. Establishing an
+owned retained store, durable path ownership, writer fencing, crash-safe ingestion,
+orphan reconciliation, safe reclaim and bounded repair remains future work. Never
+release charges or execute downloaded bytes because this audit found a file missing.

scripts/peer_install_smoke.py

@@ -698,6 +698,10 @@ def admission_smoke(cwd: Path) -> None:
     from aethermesh_core.artifact_admission import AdmissionLimits, ArtifactAdmission
     from aethermesh_core.artifact_codec import ArtifactLimits, package_artifact
     from aethermesh_core.artifact_files import verify_fragment_file
+    from aethermesh_core.artifact_inventory import (
+        InventorySelection,
+        reconcile_inventory,
+    )
 
     package = package_artifact(
         b"synthetic catalog bytes",
@@ -734,6 +738,15 @@ def admission_smoke(cwd: Path) -> None:
         invalidated = reopened.invalidate("8" * 64, expected_revision=saved.revision)
         assert invalidated.state == "reserved"
         assert reopened.snapshot().charged_bytes == len(package.fragments[0])
+        selection = (InventorySelection("8" * 64, package.manifest, fragment),)
+        report = reconcile_inventory(reopened, selection, max_read_bytes=1024)
+        assert report.observations[0].status == "verified"
+        assert not report.retention_confirmed and not report.publisher_authorized
+        fragment.unlink()
+        missing = reconcile_inventory(reopened, selection, max_read_bytes=1024)
+        assert missing.observations[0].status == "missing"
+        assert missing.charged_bytes == report.charged_bytes
+        assert reopened.snapshot().records == (invalidated,)
     finally:
         reopened.close()
     print(

src/aethermesh_core/artifact_inventory.py

@@ -0,0 +1,133 @@
+"""Bounded read-only reconciliation of admitted, explicitly selected local files.
+
+Observations are sequential and ephemeral, not retention leases, publisher
+approval, a filesystem scan, recovery writes or permission to reclaim charges.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+from .artifact_admission import AdmissionConflict, ArtifactAdmission, FragmentAdmission
+from .artifact_codec import ArtifactManifest, CheckCancel, _cancel, _integer, _trusted
+from .artifact_files import verify_fragment_file
+
+
+@dataclass(frozen=True)
+class InventorySelection:
+    request_id: str
+    manifest: ArtifactManifest
+    path: Path
+
+
+@dataclass(frozen=True)
+class InventoryObservation:
+    admission: FragmentAdmission
+    status: Literal["verified", "missing", "unverified"]
+
+
+@dataclass(frozen=True)
+class InventoryReport:
+    ledger_revision: int
+    charged_bytes: int
+    read_budget_bytes: int
+    observations: tuple[InventoryObservation, ...]
+    publisher_authorized: Literal[False] = False
+    retention_confirmed: Literal[False] = False
+
+
+def _preflight(
+    records: tuple[FragmentAdmission, ...], selections: tuple[InventorySelection, ...]
+) -> dict[str, InventorySelection]:
+    if not isinstance(selections, tuple) or len(selections) > 192:
+        raise ValueError("inventory selections must be a bounded tuple")
+    chosen = {item.request_id: item for item in selections}
+    if len(chosen) != len(selections) or set(chosen) != {
+        record.request_id for record in records
+    }:
+        raise ValueError("select every admitted request exactly once")
+    for record in records:
+        item = chosen[record.request_id]
+        _trusted(item.manifest, record.manifest_digest)
+        if (
+            item.manifest.project != record.project
+            or item.manifest.fragment_size(record.fragment_index) != record.size
+            or item.manifest.fragments[record.fragment_index]
+            != (record.sha256, record.transfer_digest)
+        ):
+            raise ValueError("inventory manifest and admission binding mismatch")
+        if not isinstance(item.path, Path) or not item.path.is_absolute():
+            raise ValueError("inventory requires explicitly selected absolute paths")
+    return chosen
+
+
+def _observe(
+    record: FragmentAdmission,
+    item: InventorySelection,
+    check_cancel: CheckCancel | None,
+) -> InventoryObservation:
+    cancelled = False
+
+    def check() -> None:
+        nonlocal cancelled
+        try:
+            _cancel(check_cancel)
+        except BaseException:
+            cancelled = True
+            raise
+
+    try:
+        verify_fragment_file(
+            item.manifest,
+            record.fragment_index,
+            item.path,
+            expected_digest=record.manifest_digest,
+            max_read_bytes=record.size,
+            check_cancel=check,
+        )
+    except (OSError, ValueError) as error:
+        if cancelled:
+            raise
+        status: Literal["missing", "unverified"] = (
+            "missing" if isinstance(error, FileNotFoundError) else "unverified"
+        )
+        return InventoryObservation(record, status)
+    return InventoryObservation(record, "verified")
+
+
+def reconcile_inventory(
+    ledger: ArtifactAdmission,
+    selections: tuple[InventorySelection, ...],
+    *,
+    max_read_bytes: int,
+    check_cancel: CheckCancel | None = None,
+) -> InventoryReport:
+    """Recheck every admitted request without changing files or ledger metadata.
+
+    Require a complete caller-selected mapping and aggregate logical read budget
+    before any fragment open. Existing verified_observation state is never used
+    as evidence of current bytes. Missing/unsafe/corrupt/unreadable files remain
+    charged. Callback exceptions always abort; no partial report is returned.
+
+    Fail if the ledger changes during the pass. This fences report metadata, not
+    filesystem writers: file observations occur at different times and must be
+    refreshed before serving. Unselected/orphan files are outside this report.
+    """
+    _integer(max_read_bytes, 0, 64 * 1024 * 1024)
+    _cancel(check_cancel)
+    snapshot = ledger.snapshot()
+    chosen = _preflight(snapshot.records, selections)
+    if snapshot.charged_bytes > max_read_bytes:
+        raise ValueError("aggregate inventory read budget exceeded")
+    observations = tuple(
+        _observe(record, chosen[record.request_id], check_cancel)
+        for record in snapshot.records
+    )
+    _cancel(check_cancel)
+    if ledger.snapshot() != snapshot:
+        raise AdmissionConflict("inventory changed during reconciliation")
+    return InventoryReport(
+        snapshot.revision, snapshot.charged_bytes, snapshot.charged_bytes, observations
+    )