feat(supply): separate source progress signals

docs/decisions/0043-separate-file-acquisition-progress-from-publication-progress.md

@@ -0,0 +1,98 @@
+---
+name: adr-0043-separate-file-acquisition-progress-from-publication-progress
+version: "1.0.0"
+description: >
+  Persist source-scoped accepted-change checkpoints separately from contiguous
+  Runtime-visibility publication watermarks.
+---
+
+# 0043. Separate File acquisition progress from publication progress
+
+- Status: accepted
+- Date: 2026-07-23
+- Refines: ADR-0037, ADR-0039, ADR-0040, ADR-0041, ADR-0042
+
+## Context
+
+A durable File import request proves that a source change was accepted, but it
+does not prove that compilation, indexing, or publication completed. Treating
+the import job state as one generic checkpoint makes operators unable to tell a
+recoverable publication pause from completed Runtime visibility. It also makes
+an out-of-order later completion capable of hiding an earlier failed change if
+the source reports only its greatest completed ordinal.
+
+Issue #29 needs two source-level progress signals without turning either signal
+into Runtime authorization or claiming the deferred provider `checkpoint`
+operation from the standard ProviderPort.
+
+## Decision
+
+Each accepted File import job and trusted File tombstone receives one immutable,
+database-ordered `file_source_acquisition_checkpoint`, scoped by Organization
+and Source. The import checkpoint is appended in the same transaction that
+creates its acquisition/job; the tombstone checkpoint is appended in the same
+transaction as tombstone visibility and cleanup intent. Opaque references are
+deterministic SHA-256-derived tokens, while positive sequence numbers remain
+database-owned ordering facts. Idempotent replay does not append another row.
+
+Each unchanged result, active Revision event, and tombstone appends one distinct
+immutable `file_source_publish_watermark` row at its existing visibility
+transaction boundary. Prepared, indexed, claimed, interrupted, or failed state
+cannot create that row. Exact foreign keys bind each completion to its accepted
+checkpoint, durable job or cleanup intent, Resource, and Revision.
+
+The read-only `ContextControl.read_file_source_progress` seam returns the latest
+accepted checkpoint and the highest *contiguous* completed sequence. If change
+3 completes while change 2 is paused, completion 3 remains durably recorded but
+the exposed watermark remains 1. When recovery completes change 2, the exposed
+watermark becomes 3 without rewriting either append-only stream. Calls are
+operation-bound and Organization-scoped through a SECURITY DEFINER function;
+all application roles are denied direct table access.
+
+Runtime receives no table privilege or dependency on this status. Current
+Resource state, active Revision, policy, audience, and AuthorizationKernel
+checks remain the only delivery authority. The File capability manifest also
+continues to report the standard provider cursor/checkpoint operations as
+unavailable: this issue activates operational progress for the manual File
+carrier, not provider `readChanges` acknowledgement.
+
+## Rationale
+
+Acceptance and visibility have different transaction boundaries and different
+failure meanings, so one mutable cursor cannot truthfully represent both. Two
+append-only streams retain the causal history needed to diagnose a pause,
+while exposing only the contiguous publication prefix prevents an
+out-of-order completion from overstating Runtime freshness. Database-owned
+ordering, deterministic opaque references, and exact durable-lineage foreign
+keys make each accepted ordering fact stable without promoting progress
+metadata into delivery authority. Existing durable lineage is ordered once
+during the initial upgrade; after any progress exists, downgrade is refused so
+concurrent online lock order can never be reconstructed differently or accept
+new changes through an older migration boundary.
+
+Keeping the operator read seam in ContextControl also preserves the existing
+ProviderPort contract: File progress can be observed now without pretending a
+provider-native change cursor has been acknowledged or making Runtime depend
+on an operational status surface.
+
+## Consequences
+
+- An accepted change remains observable through compilation, indexing,
+  interruption, failure, and recovery without falsely claiming visibility.
+- First publication, replacement, unchanged classification, and tombstone each
+  advance publication progress only inside their committed visibility outcome.
+- Concurrent or out-of-order completion cannot skip an earlier gap; recovery
+  closes the gap and deterministically catches the watermark up.
+- Both streams are append-only, forced-RLS tenant tables with no source content.
+  Downgrade is allowed only while both streams are empty; after any accepted
+  change, a forward fix is required so sequence numbers and opaque references
+  cannot be renumbered from historical timestamps.
+- Cross-source aggregation, UI, provider cursor formats, resync, and watermark-
+  based authorization remain inactive.
+
+## Revisit trigger
+
+Revisit before provider-native `readChanges`/checkpoint activation, source
+offboarding, resync, progress compaction, cross-source aggregation, or deletion
+recovery. Any change must preserve separate signals, contiguous watermarks,
+exact tenant lineage, replay idempotency, and Runtime independence.

docs/decisions/README.md

@@ -41,6 +41,7 @@ kernel, capability separation, and publication visibility model.
 | Structural Markdown compiler | [0038 — Structural Markdown units](0038-compile-and-publish-structural-markdown.md) | Explicit v2 compilation publishes one coherent Fragment per heading, paragraph, list, fenced code block, or table with exact provenance and same-Fragment heading ancestry | Silent v1 reinterpretation, item/cell fragmentation, post-authorization parent expansion, or unbudgeted context |
 | Unchanged File acquisition | [0039 — File acquisition no-op](0039-deduplicate-unchanged-file-acquisitions.md) | Tenant/source/resource-scoped canonical identity and one PostgreSQL guard lock classify a complete active artifact before publication; each observation retains an immutable digest-only outcome | Cross-tenant/global deduplication, process-local locking, partial-artifact reuse, silent version reuse, or no-op by bypassing publication validation |
 | File Resource deletion | [0042 — Tombstone before cleanup](0042-tombstone-file-resources-before-cleanup.md) | One trusted Control transaction tombstones the active File Resource, advances its Organization Policy Epoch, and records immutable pending cleanup lineage before any physical deletion | Cleanup-defined visibility, caller-authored tenant/epoch/cleanup identity, index deletion as authorization, restore, or native watcher claims |
+| File Source progress | [0043 — Separate acquisition and publication progress](0043-separate-file-acquisition-progress-from-publication-progress.md) | Append accepted changes separately from contiguous Runtime-visibility completion and expose them through an Organization/Source-scoped Control read | One ambiguous checkpoint, skipped publication gaps, Runtime authorization from watermarks, or false standard ProviderPort capability claims |
 
 Each baseline ADR is `accepted` and contains Context, Decision, Rationale,
 Consequences, and Revisit trigger sections. A revisit trigger permits review; it
@@ -130,3 +131,4 @@ touched:
 - [0040 — Atomic File replacement](0040-stage-and-atomically-activate-file-replacements.md)
 - [0041 — Durable File publication recovery](0041-recover-file-publication-by-durable-boundary.md)
 - [0042 — Tombstone File Resources before cleanup](0042-tombstone-file-resources-before-cleanup.md)
+- [0043 — Separate File acquisition and publication progress](0043-separate-file-acquisition-progress-from-publication-progress.md)

docs/design/2026-07-18-context-engine-implementation-design.md

@@ -594,6 +594,17 @@ Duplicate and older deletion observations return the original result without
 another epoch bump. Physical cleanup, restore, native watcher detection, source
 offboarding, and cleanup-job execution remain inactive.
 
+ADR-0043 separates operational File progress into two append-only,
+Organization/Source-scoped streams. Import-job or tombstone acceptance appends
+an acquisition checkpoint at the durable acceptance boundary. Unchanged
+classification, active Revision publication/replacement, or tombstone commit
+appends a distinct publish completion at the existing visibility boundary. The
+Control status seam exposes the latest accepted sequence and only the greatest
+contiguous completed sequence: a later publication cannot jump over an earlier
+interrupted job, while recovery closes the gap without rewriting history.
+Runtime has no privilege on or dependency upon either signal, and the deferred
+standard ProviderPort cursor/checkpoint operations remain unavailable.
+
 Because Runtime resolves through multiple SQL statements at `READ COMMITTED`,
 each UserActor transaction takes an Organization-scoped shared publication
 barrier and activation takes the matching exclusive transaction barrier around

engine/control/__init__.py

@@ -38,6 +38,13 @@
     PreparedFileImport,
     PrepareFileImport,
 )
+from engine.control.file_source_progress import (
+    FileSourceAcquisitionCheckpoint,
+    FileSourceChangeKind,
+    FileSourceProgress,
+    FileSourcePublishOutcome,
+    FileSourcePublishWatermark,
+)
 from engine.control.module import ContextControl, ControlStorePort
 
 __all__ = [
@@ -56,6 +63,11 @@
     "FileImportPath",
     "FileImportReceiver",
     "FileRootRef",
+    "FileSourceAcquisitionCheckpoint",
+    "FileSourceChangeKind",
+    "FileSourceProgress",
+    "FileSourcePublishOutcome",
+    "FileSourcePublishWatermark",
     "FileResourceTombstone",
     "RegisterFileSource",
     "PrepareFileImport",

engine/control/authority.py

@@ -22,6 +22,7 @@ class ControlOperation(StrEnum):
     IMPORT_FILE = "import_file"
     REGISTER_SOURCE = "register_source"
     READ_SOURCE = "read_source"
+    READ_SOURCE_PROGRESS = "read_source_progress"
     TOMBSTONE_FILE_RESOURCE = "tombstone_file_resource"
 
 

engine/control/file_source_progress.py

@@ -0,0 +1,200 @@
+"""Read-only File Source acquisition-checkpoint and publication-watermark contracts."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import StrEnum
+from uuid import UUID
+
+from engine.control.contracts import (
+    SourceRef,
+    _require_bounded_text,
+    _require_token,
+    _require_utc,
+)
+
+_MAX_BIGINT = 9_223_372_036_854_775_807
+
+
+class FileSourceChangeKind(StrEnum):
+    """Closed durable change carriers that participate in File progress."""
+
+    FILE_IMPORT = "file_import"
+    FILE_TOMBSTONE = "file_tombstone"
+
+
+class FileSourcePublishOutcome(StrEnum):
+    """Closed visibility outcomes that may advance a publish watermark."""
+
+    PUBLISHED = "published"
+    REPLACED = "replaced"
+    UNCHANGED = "unchanged"
+    TOMBSTONED = "tombstoned"
+
+
+def _require_sequence(name: str, value: object) -> int:
+    if type(value) is not int or not 1 <= value <= _MAX_BIGINT:
+        raise ValueError(f"{name} must fit a positive signed bigint")
+    return value
+
+
+def _require_resource_ref(value: object) -> str:
+    return _require_bounded_text("File Source progress ResourceRef", value, 512)
+
+
+def _require_progress_ref(name: str, value: object, prefix: str) -> str:
+    token = _require_token(name, value)
+    digest = token.removeprefix(prefix)
+    if not token.startswith(prefix) or len(digest) != 64 or any(
+        character not in "0123456789abcdef" for character in digest
+    ):
+        raise ValueError(f"{name} is not a recognized opaque reference")
+    return token
+
+
+def _require_lineage(
+    *,
+    change_kind: FileSourceChangeKind,
+    acquisition_ref: object,
+    job_ref: object,
+    cleanup_intent_ref: object,
+    resource_ref: object,
+    revision_ref: object,
+    event_ref: object,
+    event_sequence: object,
+    allow_unresolved_import_resource: bool,
+) -> None:
+    if type(change_kind) is not FileSourceChangeKind:
+        raise TypeError("File Source progress change_kind is invalid")
+    if change_kind is FileSourceChangeKind.FILE_IMPORT:
+        if type(acquisition_ref) is not UUID or type(job_ref) is not UUID:
+            raise TypeError("File import progress requires acquisition and job lineage")
+        if any(
+            value is not None
+            for value in (cleanup_intent_ref, event_ref, event_sequence)
+        ):
+            raise ValueError("File import progress cannot carry tombstone lineage")
+        if allow_unresolved_import_resource and resource_ref is revision_ref is None:
+            return
+    else:
+        if type(cleanup_intent_ref) is not UUID:
+            raise TypeError("File tombstone progress requires cleanup lineage")
+        if acquisition_ref is not None or job_ref is not None:
+            raise ValueError("File tombstone progress cannot carry import lineage")
+        _require_token("File tombstone progress event_ref", event_ref)
+        _require_sequence("File tombstone progress event_sequence", event_sequence)
+    _require_resource_ref(resource_ref)
+    if type(revision_ref) is not UUID:
+        raise TypeError("File Source progress revision_ref must be UUID")
+
+
+@dataclass(frozen=True, slots=True)
+class FileSourceAcquisitionCheckpoint:
+    """Latest source change durably accepted, irrespective of visibility."""
+
+    sequence: int
+    checkpoint_ref: str = field(repr=False)
+    change_kind: FileSourceChangeKind
+    acquisition_ref: UUID | None = field(repr=False)
+    job_ref: UUID | None = field(repr=False)
+    cleanup_intent_ref: UUID | None = field(repr=False)
+    resource_ref: str | None
+    revision_ref: UUID | None = field(repr=False)
+    event_ref: str | None = field(repr=False)
+    event_sequence: int | None
+    accepted_at: datetime
+
+    def __post_init__(self) -> None:
+        _require_sequence("File Source acquisition sequence", self.sequence)
+        _require_progress_ref(
+            "File Source acquisition checkpoint_ref", self.checkpoint_ref, "facp_"
+        )
+        _require_lineage(
+            change_kind=self.change_kind,
+            acquisition_ref=self.acquisition_ref,
+            job_ref=self.job_ref,
+            cleanup_intent_ref=self.cleanup_intent_ref,
+            resource_ref=self.resource_ref,
+            revision_ref=self.revision_ref,
+            event_ref=self.event_ref,
+            event_sequence=self.event_sequence,
+            allow_unresolved_import_resource=True,
+        )
+        _require_utc("File Source acquisition accepted_at", self.accepted_at)
+
+
+@dataclass(frozen=True, slots=True)
+class FileSourcePublishWatermark:
+    """Latest contiguous accepted change fully reflected in Runtime visibility."""
+
+    sequence: int
+    watermark_ref: str = field(repr=False)
+    checkpoint_ref: str = field(repr=False)
+    change_kind: FileSourceChangeKind
+    outcome: FileSourcePublishOutcome
+    acquisition_ref: UUID | None = field(repr=False)
+    job_ref: UUID | None = field(repr=False)
+    cleanup_intent_ref: UUID | None = field(repr=False)
+    resource_ref: str
+    revision_ref: UUID = field(repr=False)
+    event_ref: str | None = field(repr=False)
+    event_sequence: int | None
+    published_at: datetime
+
+    def __post_init__(self) -> None:
+        _require_sequence("File Source publish sequence", self.sequence)
+        _require_progress_ref(
+            "File Source publish watermark_ref", self.watermark_ref, "fpwm_"
+        )
+        _require_progress_ref(
+            "File Source publish checkpoint_ref", self.checkpoint_ref, "facp_"
+        )
+        if type(self.outcome) is not FileSourcePublishOutcome:
+            raise TypeError("File Source publish outcome is invalid")
+        _require_lineage(
+            change_kind=self.change_kind,
+            acquisition_ref=self.acquisition_ref,
+            job_ref=self.job_ref,
+            cleanup_intent_ref=self.cleanup_intent_ref,
+            resource_ref=self.resource_ref,
+            revision_ref=self.revision_ref,
+            event_ref=self.event_ref,
+            event_sequence=self.event_sequence,
+            allow_unresolved_import_resource=False,
+        )
+        if (
+            self.change_kind is FileSourceChangeKind.FILE_TOMBSTONE
+        ) is not (self.outcome is FileSourcePublishOutcome.TOMBSTONED):
+            raise ValueError("File Source publish outcome does not match its change")
+        _require_utc("File Source publish published_at", self.published_at)
+
+
+@dataclass(frozen=True, slots=True)
+class FileSourceProgress:
+    """Organization/Source-scoped read model for the two progress signals."""
+
+    organization_id: UUID = field(repr=False)
+    source_ref: SourceRef
+    acquisition_checkpoint: FileSourceAcquisitionCheckpoint | None
+    publish_watermark: FileSourcePublishWatermark | None
+
+    def __post_init__(self) -> None:
+        if type(self.organization_id) is not UUID:
+            raise TypeError("File Source progress organization_id must be UUID")
+        if type(self.source_ref) is not SourceRef:
+            raise TypeError("File Source progress source_ref must be SourceRef")
+        if self.acquisition_checkpoint is not None and type(
+            self.acquisition_checkpoint
+        ) is not FileSourceAcquisitionCheckpoint:
+            raise TypeError("File Source acquisition checkpoint is invalid")
+        if self.publish_watermark is not None and type(
+            self.publish_watermark
+        ) is not FileSourcePublishWatermark:
+            raise TypeError("File Source publish watermark is invalid")
+        if self.publish_watermark is not None and (
+            self.acquisition_checkpoint is None
+            or self.publish_watermark.sequence
+            > self.acquisition_checkpoint.sequence
+        ):
+            raise ValueError("File Source publish watermark cannot exceed checkpoint")