supply: schedule accepted File upserts through exact import jobs

docs/decisions/0055-schedule-accepted-file-observations-explicitly.md

@@ -0,0 +1,80 @@
+---
+name: adr-0055-schedule-accepted-file-observations-explicitly
+version: "1.0.0"
+description: >
+  Bind accepted File observations to existing import jobs without inventing an
+  audience or allowing later filesystem bytes to replace checkpoint lineage.
+---
+
+# 0055. Schedule accepted File observations explicitly
+
+- Status: accepted
+- Date: 2026-07-25
+- Refines: ADR-0037, ADR-0043, ADR-0054
+
+## Context
+
+Issue #81 made File change pages durable but intentionally created no jobs: a
+content-free filesystem observation has no delivery audience. Issue #83 needs a
+narrow bridge from an already accepted page to the existing File import worker
+without turning provider metadata, worker identity, or an earlier manual import
+into delivery authority.
+
+The filesystem can also change after Control accepts a page. Publishing the
+later bytes under the earlier page checkpoint would make the accepted digest
+decorative and corrupt acquisition provenance.
+
+## Decision
+
+`ContextControl.schedule_file_change_page` is an explicit trusted operation. It
+requires the exact active v3 Organization, Source, SourceVersion, accepted page
+reference, and an explicit existing `FileImportAudience`. Page acceptance
+continues to create no job and never infers an audience.
+
+One SECURITY DEFINER transaction acquires the existing per-Source progress
+advisory lock before the active Source row lock, matching page acceptance and
+checkpoint-trigger lock order. It validates the current Membership/version and
+exact enabled File import receiver, requires the page to belong to the current
+accepted scan epoch before creating its first lineage, checks a nonempty
+contiguous stored `upsert` set, and creates one existing `file_acquisition` plus
+one existing
+`file_import_job` per change. Exact replay returns the same ordered jobs. Any
+partial lineage or changed audience returns no rows and creates no new jobs.
+
+Each scheduled acquisition has an Organization-inclusive foreign key to the
+exact accepted observation: SourceVersion, page, ordinal, path, raw SHA-256, and
+raw byte length. The existing immutable acquisition trigger prevents lineage
+rewrites. Manual imports retain null observation fields and their existing
+behavior.
+
+Lease issuance and redemption still use only `file.import`. Redemption exposes
+the optional expected raw identity to the worker. After its stable no-follow
+read and before Markdown compilation, the worker compares byte length and
+SHA-256. A missing or changed file closes the redeemed job as failed when its
+exact WorkerLease authority remains current, with no Revision, candidate,
+policy, or publication watermark. A later scan may observe the new state; this
+job never substitutes it for the accepted observation.
+
+The failure marker remains subject to the existing exact current WorkerLease
+authority. If the receiver is revoked after redemption, revocation vetoes even
+that state mutation: the job remains content-free `running`, while the worker
+still exposes only generic `FileImportUnavailable`. Retry/reclaim policy for
+that fenced state remains inactive and must be introduced by a later issue.
+
+## Consequences
+
+- Explicit scheduling reuses the current queue, WorkerLease, publication, and
+  acquisition-progress protocols.
+- Scheduling a page is all-or-none and exact replay is idempotent.
+- Provider checkpoints and worker identity remain non-authoritative for Runtime
+  delivery.
+- Automatic polling, filesystem watching, deletion execution, retry/reclaim,
+  dead-letter handling, and full resync remain inactive.
+- Downgrade is refused while any acquisition retains accepted-change lineage.
+
+## Revisit trigger
+
+Revisit before autonomous scheduling, deletion execution, retry/reclaim,
+recursive discovery, or full resync. Any revision must preserve explicit
+audience authority, exact observation binding, whole-page atomicity, and
+pre-compiler byte verification.

engine/control/__init__.py

@@ -61,6 +61,9 @@
     FileImportReceiver,
     PreparedFileImport,
     PrepareFileImport,
+    ScheduledFileChange,
+    ScheduledFileChangePage,
+    ScheduleFileChangePage,
 )
 from engine.control.file_source_offboarding import (
     FileSourceCleanupState,
@@ -119,6 +122,9 @@
     "PendingChangeCursor",
     "PrepareFileImport",
     "PreparedFileImport",
+    "ScheduleFileChangePage",
+    "ScheduledFileChange",
+    "ScheduledFileChangePage",
     "ProviderGenericDenied",
     "ProviderInvalidCheckpoint",
     "ProviderOk",

engine/control/authority.py

@@ -26,6 +26,7 @@ class ControlOperation(StrEnum):
     REGISTER_SOURCE = "register_source"
     READ_SOURCE = "read_source"
     READ_SOURCE_PROGRESS = "read_source_progress"
+    SCHEDULE_FILE_CHANGE_PAGE = "schedule_file_change_page"
     TOMBSTONE_FILE_RESOURCE = "tombstone_file_resource"
 
 

engine/control/contracts.py

@@ -46,6 +46,16 @@ def _require_utc(field_name: str, value: object) -> datetime:
     return value
 
 
+def _require_sha256(field_name: str, value: object) -> str:
+    if (
+        type(value) is not str
+        or len(value) != 64
+        or any(character not in "0123456789abcdef" for character in value)
+    ):
+        raise ValueError(f"{field_name} must be lowercase SHA-256")
+    return value
+
+
 class SourceKind(StrEnum):
     FILE = "file"
 

engine/control/file_change_pages.py

@@ -18,7 +18,12 @@
 )
 
 from engine._opaque import decode_base64url, encode_base64url
-from engine.control.contracts import SourceRef, SourceVersion, _require_utc
+from engine.control.contracts import (
+    SourceRef,
+    SourceVersion,
+    _require_sha256,
+    _require_utc,
+)
 from engine.control.file_imports import FileImportPath
 
 MAX_FILE_CHANGE_PAGE_SIZE = 100
@@ -108,9 +113,10 @@ def __post_init__(self) -> None:
             raise ValueError("File change scan head sequence is invalid")
         if type(self.complete) is not bool:
             raise TypeError("File change scan head complete must be bool")
-        if self.superseded_scan_epoch is not None and type(
-            self.superseded_scan_epoch
-        ) is not UUID:
+        if (
+            self.superseded_scan_epoch is not None
+            and type(self.superseded_scan_epoch) is not UUID
+        ):
             raise TypeError("superseded File scan epoch must be UUID or None")
 
 
@@ -127,9 +133,10 @@ def __post_init__(self) -> None:
             raise TypeError("File change source organization_id must be UUID")
         if type(self.source_version) is not SourceVersion:
             raise TypeError("File change source requires SourceVersion")
-        if self.scan_head is not None and type(
-            self.scan_head
-        ) is not FileChangeScanHead:
+        if (
+            self.scan_head is not None
+            and type(self.scan_head) is not FileChangeScanHead
+        ):
             raise TypeError("File change source scan head is invalid")
 
 
@@ -139,16 +146,6 @@ class FileChangeKind(StrEnum):
     UPSERT = "upsert"
 
 
-def _require_sha256(name: str, value: object) -> str:
-    if (
-        type(value) is not str
-        or len(value) != 64
-        or any(character not in "0123456789abcdef" for character in value)
-    ):
-        raise ValueError(f"{name} must be lowercase SHA-256")
-    return value
-
-
 @dataclass(frozen=True, slots=True)
 class SourceChange:
     """One content-free File observation bound to an exact source scan."""
@@ -230,17 +227,21 @@ def __post_init__(self) -> None:
             or not 1 <= self.predecessor_sequence <= 2**63 - 1
         ):
             raise ValueError("ChangePage predecessor_sequence is invalid")
-        if len(
-            {
-                self.predecessor_page_ref is None,
-                self.predecessor_checkpoint_ref is None,
-                self.predecessor_sequence is None,
-            }
-        ) != 1:
+        if (
+            len(
+                {
+                    self.predecessor_page_ref is None,
+                    self.predecessor_checkpoint_ref is None,
+                    self.predecessor_sequence is None,
+                }
+            )
+            != 1
+        ):
             raise ValueError("ChangePage predecessor checkpoint binding is incomplete")
-        if self.superseded_scan_epoch is not None and type(
-            self.superseded_scan_epoch
-        ) is not UUID:
+        if (
+            self.superseded_scan_epoch is not None
+            and type(self.superseded_scan_epoch) is not UUID
+        ):
             raise TypeError("ChangePage superseded_scan_epoch must be UUID or None")
         if (
             self.predecessor_page_ref is not None
@@ -300,9 +301,7 @@ def _page_document(page: ChangePage) -> dict[str, object]:
         ],
         "complete": page.complete,
         "domain": _PAGE_DOMAIN,
-        "nextCursor": (
-            None if page.next_cursor is None else page.next_cursor.value
-        ),
+        "nextCursor": (None if page.next_cursor is None else page.next_cursor.value),
         "organizationId": str(page.organization_id),
         "pageLimit": page.page_limit,
         "predecessorCheckpointRef": page.predecessor_checkpoint_ref,
@@ -430,9 +429,10 @@ def __post_init__(self) -> None:
             or not 1 <= self.page_limit <= MAX_FILE_CHANGE_PAGE_SIZE
         ):
             raise ValueError("accepted page page_limit is invalid")
-        if self.superseded_scan_epoch is not None and type(
-            self.superseded_scan_epoch
-        ) is not UUID:
+        if (
+            self.superseded_scan_epoch is not None
+            and type(self.superseded_scan_epoch) is not UUID
+        ):
             raise TypeError("accepted page superseded scan epoch is invalid")
         if self.next_cursor is not None and type(self.next_cursor) is not ChangeCursor:
             raise TypeError("accepted page next_cursor must be ChangeCursor or None")
@@ -534,9 +534,7 @@ def _sign_pending_payload(self, payload: bytes) -> bytes:
     def _seal_page(self, page: ChangePage) -> str:
         return encode_base64url(self._provider_signing_key.sign(_page_payload(page)))
 
-    def _unwrap_cursor(
-        self, cursor: ChangeCursor
-    ) -> _AcceptedCursorClaims | None:
+    def _unwrap_cursor(self, cursor: ChangeCursor) -> _AcceptedCursorClaims | None:
         try:
             encoded_payload, encoded_signature = cursor.value.split(".")
             payload = decode_base64url(encoded_payload)

engine/control/file_imports.py

@@ -6,7 +6,7 @@
 from typing import NoReturn
 from uuid import UUID
 
-from engine.control.contracts import SourceRef
+from engine.control.contracts import SourceRef, _require_sha256
 
 FILE_IMPORT_WORKLOAD = "supply.file-import"
 FILE_IMPORT_WORKER_AUDIENCE = "context-engine-worker"
@@ -90,6 +90,28 @@ def __reduce__(self) -> NoReturn:
         raise TypeError("File import command is not serializable")
 
 
+@dataclass(frozen=True, slots=True)
+class ScheduleFileChangePage:
+    """Explicitly schedule one accepted File page for an exact audience."""
+
+    source_ref: SourceRef
+    source_version_ref: UUID = field(repr=False)
+    page_ref: str = field(repr=False)
+    audience: FileImportAudience = field(repr=False)
+
+    def __post_init__(self) -> None:
+        if type(self.source_ref) is not SourceRef:
+            raise TypeError("scheduled File page source_ref must be SourceRef")
+        if type(self.source_version_ref) is not UUID:
+            raise TypeError("scheduled File page source version must be UUID")
+        _require_sha256("scheduled File page reference", self.page_ref)
+        if type(self.audience) is not FileImportAudience:
+            raise TypeError("scheduled File page audience must be FileImportAudience")
+
+    def __reduce__(self) -> NoReturn:
+        raise TypeError("File change page scheduling command is not serializable")
+
+
 @dataclass(frozen=True, slots=True)
 class PreparedFileImport:
     """Content-free exact-job locator returned by trusted ContextControl."""
@@ -117,6 +139,64 @@ def __post_init__(self) -> None:
             raise TypeError("Prepared File import service principal must be UUID")
 
 
+@dataclass(frozen=True, slots=True)
+class ScheduledFileChange:
+    """One accepted change bound to its existing prepared import job."""
+
+    ordinal: int
+    path: FileImportPath
+    content_sha256: str = field(repr=False)
+    content_length: int
+    prepared_import: PreparedFileImport
+
+    def __post_init__(self) -> None:
+        if type(self.ordinal) is not int or not 1 <= self.ordinal <= 100:
+            raise ValueError("scheduled File change ordinal is invalid")
+        if type(self.path) is not FileImportPath:
+            raise TypeError("scheduled File change path must be FileImportPath")
+        _require_sha256("scheduled File change content identity", self.content_sha256)
+        if type(self.content_length) is not int or self.content_length < 0:
+            raise ValueError("scheduled File change content length is invalid")
+        if type(self.prepared_import) is not PreparedFileImport:
+            raise TypeError("scheduled File change requires PreparedFileImport")
+
+
+@dataclass(frozen=True, slots=True)
+class ScheduledFileChangePage:
+    """Complete ordered job set created from one accepted File page."""
+
+    organization_id: UUID = field(repr=False)
+    source_ref: SourceRef = field(repr=False)
+    source_version_ref: UUID = field(repr=False)
+    page_ref: str = field(repr=False)
+    changes: tuple[ScheduledFileChange, ...]
+
+    def __post_init__(self) -> None:
+        if type(self.organization_id) is not UUID:
+            raise TypeError("scheduled File page organization must be UUID")
+        if type(self.source_ref) is not SourceRef:
+            raise TypeError("scheduled File page source must be SourceRef")
+        if type(self.source_version_ref) is not UUID:
+            raise TypeError("scheduled File page source version must be UUID")
+        _require_sha256("scheduled File page reference", self.page_ref)
+        if (
+            type(self.changes) is not tuple
+            or not self.changes
+            or any(type(change) is not ScheduledFileChange for change in self.changes)
+        ):
+            raise TypeError("scheduled File page requires a nonempty change tuple")
+        if tuple(change.ordinal for change in self.changes) != tuple(
+            range(1, len(self.changes) + 1)
+        ):
+            raise ValueError("scheduled File page changes must be contiguous")
+        if any(
+            change.prepared_import.organization_id != self.organization_id
+            or change.prepared_import.source_ref != self.source_ref
+            for change in self.changes
+        ):
+            raise ValueError("scheduled File page contains a foreign import")
+
+
 @dataclass(frozen=True, slots=True)
 class FileImportReceiver:
     """Trusted registered worker identity injected into Control composition."""