control: compose bounded File scan cycles

README.md

@@ -215,6 +215,97 @@ or the current authority rejects that exact failure transition.
 Activation boundaries for File dispatch, reclaim, and delete execution are
 recorded in [STATUS.md](./STATUS.md).
 
+### Scan a local File source
+
+The local operator can run one bounded File acquisition cycle and hand its
+scheduled upserts to the existing worker. This remains an explicitly configured
+local process; it adds no HTTP operation, polling daemon, publication path, or
+delete authority.
+
+Load the generated harness database environment first, then configure the
+local operator composition described by
+[ADR-0069](./docs/decisions/0069-admit-an-explicit-local-operator-composition.md).
+The Control operation allowlist for this workflow is:
+
+```text
+register_source,read_source,read_source_progress,activate_file_change_feed,activate_file_delete_observations,accept_file_change_page,schedule_file_change_page
+```
+
+The scan and worker share the same server-owned root registry and byte ceiling.
+They additionally require one durable File-import receiver, the current private
+dogfood audience, and two distinct persistent Ed25519 proof keys:
+
+```text
+CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON
+CONTEXT_ENGINE_WORKER_MAX_FILE_BYTES                 # optional
+CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID
+CONTEXT_ENGINE_DOGFOOD_PRINCIPAL_REF
+CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID
+CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_VERSION
+CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX
+CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX
+CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX
+```
+
+Each proof-key value is exactly 32 random bytes encoded as 64 lowercase or
+uppercase hexadecimal characters. Keep both in the same local secret source
+across process restarts and never print or commit them. They must be distinct
+from each other and from the Control, release, dogfood, and worker secrets. The
+worker signing key is already required by the explicit local operator
+composition and is checked here only to preserve that cross-plane separation.
+Seed the receiver together with the dogfood identity (the command is
+idempotent for the exact same bindings):
+
+```bash
+uv run context-engine-dogfood-seed \
+  --organization-id "$CONTEXT_ENGINE_DOGFOOD_ORGANIZATION_ID" \
+  --user-id "$CONTEXT_ENGINE_DOGFOOD_USER_ID" \
+  --membership-id "$CONTEXT_ENGINE_DOGFOOD_MEMBERSHIP_ID" \
+  --file-import-service-principal-id \
+    "$CONTEXT_ENGINE_WORKER_SERVICE_PRINCIPAL_ID"
+```
+
+Register the logical root, copy the returned `sourceRef` into
+`CONTEXT_ENGINE_FILE_SOURCE_REF`, activate its two existing immutable
+capability transitions, and run the cycle:
+
+```bash
+uv run context-engine-control register-file-source \
+  --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \
+  --display-name "Maintainer notes" \
+  --root-ref "maintainer-notes" \
+  --idempotency-key "maintainer-notes-v1"
+
+uv run context-engine-control activate-change-feed \
+  --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \
+  --source-ref "$CONTEXT_ENGINE_FILE_SOURCE_REF"
+
+uv run context-engine-control activate-delete-observations \
+  --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \
+  --source-ref "$CONTEXT_ENGINE_FILE_SOURCE_REF"
+
+uv run context-engine-control scan \
+  --organization-id "$CONTEXT_ENGINE_OPERATOR_ORGANIZATION_ID" \
+  --source-ref "$CONTEXT_ENGINE_FILE_SOURCE_REF"
+
+uv run context-engine-worker --dispatch-file-once
+```
+
+`scan` requires that exact delete-observation activation because its complete
+durable baseline is also what makes unchanged-path scheduling decisions
+idempotent. A v1, v2, or v3 source is refused generically.
+
+Repeat the final worker command until it reports `no_work`, or run the existing
+long-lived dispatcher. The scan prints deterministic, content-free JSON counts.
+`advancedCursor` is the accepted durable checkpoint reference; an exact
+unchanged replay reports zero accepted changes and scheduled imports while
+retaining that already-advanced checkpoint when no accepted page is missing its
+schedule. Before returning, scan idempotently schedules any accepted current-
+scan upsert page that has no durable acquisition. Those counts are baseline deltas.
+Compilation refusals are counted before handoff using the worker's exact active
+Markdown configuration, but the worker remains the only publication path and
+makes the authoritative terminal transition for each scheduled import.
+
 ### Development commands
 
 ```bash

STATUS.md

@@ -116,6 +116,7 @@ Follow the ADR for its exact evidence boundary.
 | [0065](./docs/decisions/0065-recurse-file-discovery-with-anchored-descriptors.md) | Recurse File discovery through anchored descriptors under one bounded byte ceiling |
 | [0066](./docs/decisions/0066-embed-fragments-before-publication.md) | Embed newly published Fragments before activation through an explicit provider |
 | [0070](./docs/decisions/0070-activate-file-change-feed-from-registration.md) | Advance an exact registered v1 or import-enabled v2 File source to the existing immutable v3 change-feed manifest |
+| [0071](./docs/decisions/0071-compose-bounded-file-scan-cycles.md) | Compose a bounded local File scan from operation-exact accept and schedule calls with checkpoint idempotence |
 
 ADR-0065 extends the active File Provider boundary from a flat root to
 deterministic recursive discovery of canonical nested Markdown paths. Each
@@ -138,6 +139,17 @@ future candidate-discovery implementation detail and has no authorization role.
 This does **not** activate vector retrieval, query embedding, historical
 backfill, or any Runtime/AuthorizationKernel change.
 
+ADR-0071 composes the opt-in ADR-0069 local operator process to drive one
+bounded File scan over an explicitly configured anchored root, accept every new
+provider page, schedule only changed upserts, reconcile accepted current-scan
+upsert pages missing durable jobs, and hand those jobs to the existing
+autonomous worker. Real-PostgreSQL fixture evidence covers exact unchanged
+replay, interrupted scheduling recovery, one-note addition, aggregate
+compilation refusal, delete observation without delete execution, and
+384-dimensional Fragment publication. This does not claim that the maintainer's
+private corpus has run; it activates no watcher, alternate publisher, new
+tombstone authority, or network operation.
+
 ### Wire contract, SDK, and trusted delivery
 
 | ADR | Activates |

applications/control.py

@@ -9,6 +9,8 @@
 from datetime import UTC, datetime, timedelta
 from uuid import UUID, uuid4
 
+from applications.file_root_configuration import file_roots
+from applications.file_scan import FileScanReport, scan_file_source
 from applications.operator_authentication import (
     CONTROL_OPERATOR_SECRET_ENV,
     LocalOperatorAuthorities,
@@ -39,6 +41,7 @@
         "read-source",
         "activate-change-feed",
         "activate-delete-observations",
+        "scan",
     }
 )
 
@@ -65,6 +68,7 @@ def _parser() -> argparse.ArgumentParser:
             "activate-delete-observations",
             "activate one File source delete-observation capability",
         ),
+        ("scan", "scan one registered File source and schedule changed upserts"),
     ):
         source_command = subcommands.add_parser(name, help=help_text)
         _organization_argument(source_command)
@@ -89,8 +93,13 @@ def main(argv: Sequence[str] | None = None) -> None:
     if arguments.subcommand not in _OPERATOR_SUBCOMMANDS:
         parser.error("unknown operation")
     try:
-        manifest = _run_operator_subcommand(arguments)
-        rendered = _manifest_json(manifest)
+        outcome = _run_operator_subcommand(arguments)
+        if type(outcome) is FileScanReport:
+            rendered = _scan_report_json(outcome)
+        elif type(outcome) is SourceManifest:
+            rendered = _manifest_json(outcome)
+        else:  # pragma: no cover - closed application union
+            raise SourceNotAvailable
     except Exception:  # Operator refusals disclose no supplied or trusted facts.
         parser.exit(1, "context-engine-control: operation refused\n")
     print(rendered, flush=True)
@@ -105,20 +114,33 @@ def local_operator_authorities() -> LocalOperatorAuthorities | None:
     return configuration.authorities()
 
 
-def _run_operator_subcommand(arguments: argparse.Namespace) -> SourceManifest:
+def _run_operator_subcommand(
+    arguments: argparse.Namespace,
+) -> SourceManifest | FileScanReport:
     authorities = local_operator_authorities()
     if authorities is None:
         raise SourceNotAvailable
     organization_id = UUID(arguments.organization_id)
     opaque_credential = os.environ[CONTROL_OPERATOR_SECRET_ENV]
-    operation = _operation(arguments.subcommand)
     configuration = load_database_configuration(DatabasePurpose.CONTROL_PLANE)
     engine = create_database_engine(configuration)
 
     def clock() -> datetime:
         return datetime.now(UTC)
 
     try:
+        if arguments.subcommand == "scan":
+            with file_roots() as roots:
+                return scan_file_source(
+                    organization_id=organization_id,
+                    source_ref=SourceRef(UUID(arguments.source_ref)),
+                    authority=authorities.control,
+                    opaque_credential=opaque_credential,
+                    engine=engine,
+                    clock=clock,
+                    roots=roots,
+                )
+        operation = _operation(arguments.subcommand)
         control = ContextControl(
             store=PostgreSQLControlStore(engine, clock=clock),
             authority=authorities.control,
@@ -192,6 +214,24 @@ def _manifest_json(manifest: SourceManifest) -> str:
     return json.dumps(document, separators=(",", ":"), sort_keys=True)
 
 
+def _scan_report_json(report: FileScanReport) -> str:
+    if type(report) is not FileScanReport:
+        raise SourceNotAvailable
+    return json.dumps(
+        {
+            "advancedCursor": report.advanced_cursor,
+            "changesAccepted": report.changes_accepted,
+            "compilationRefusals": report.compilation_refusals,
+            "deletesObserved": report.deletes_observed,
+            "importsScheduled": report.imports_scheduled,
+            "pathsObserved": report.paths_observed,
+            "sourceRef": str(report.source_ref.value),
+        },
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+
+
 def _timestamp(value: datetime) -> str:
     if type(value) is not datetime or value.utcoffset() != timedelta(0):
         raise SourceNotAvailable

applications/dogfood.py

@@ -1,4 +1,4 @@
-"""Explicit local identity seeding for the dogfood composition."""
+"""Explicit local identity and optional File receiver dogfood seeding."""
 
 from __future__ import annotations
 
@@ -26,12 +26,16 @@ def _uuid(value: str) -> UUID:
 
 def main(argv: Sequence[str] | None = None) -> None:
     parser = argparse.ArgumentParser(
-        description="Seed one local Organization/User/current Membership"
+        description=(
+            "Seed one local Organization/User/current Membership and an "
+            "optional File-import receiver"
+        )
     )
     parser.add_argument("--organization-id", required=True, type=_uuid)
     parser.add_argument("--user-id", required=True, type=_uuid)
     parser.add_argument("--membership-id", required=True, type=_uuid)
     parser.add_argument("--membership-version", default=1, type=int)
+    parser.add_argument("--file-import-service-principal-id", type=_uuid)
     args = parser.parse_args(argv)
     if not 1 <= args.membership_version < (1 << 63):
         parser.error("--membership-version must be a positive signed bigint")
@@ -84,6 +88,28 @@ def main(argv: Sequence[str] | None = None) -> None:
                     "valid_from": seeded_at,
                 },
             )
+            if args.file_import_service_principal_id is not None:
+                connection.execute(
+                    text(
+                        """
+                        INSERT INTO service_principal (
+                            organization_id, service_principal_id, workload,
+                            worker_audience, operation, enabled
+                        ) VALUES (
+                            :organization_id, :service_principal_id,
+                            'supply.file-import', 'context-engine-worker',
+                            'file.import', true
+                        )
+                        ON CONFLICT (
+                            organization_id, service_principal_id
+                        ) DO NOTHING
+                        """
+                    ),
+                    {
+                        "organization_id": args.organization_id,
+                        "service_principal_id": (args.file_import_service_principal_id),
+                    },
+                )
             exact = connection.execute(
                 text(
                     """
@@ -112,13 +138,46 @@ def main(argv: Sequence[str] | None = None) -> None:
             ).scalar_one()
             if exact is not True:
                 raise RuntimeError("dogfood identity conflicts with durable ownership")
+            if args.file_import_service_principal_id is not None:
+                exact_receiver = connection.execute(
+                    text(
+                        """
+                        SELECT EXISTS (
+                            SELECT 1
+                            FROM service_principal
+                            WHERE organization_id = :organization_id
+                              AND service_principal_id = :service_principal_id
+                              AND workload = 'supply.file-import'
+                              AND worker_audience = 'context-engine-worker'
+                              AND operation = 'file.import'
+                              AND enabled IS TRUE
+                        )
+                        """
+                    ),
+                    {
+                        "organization_id": args.organization_id,
+                        "service_principal_id": (args.file_import_service_principal_id),
+                    },
+                ).scalar_one()
+                if exact_receiver is not True:
+                    raise RuntimeError(
+                        "dogfood receiver conflicts with durable ownership"
+                    )
     finally:
         engine.dispose()
     print(
         "dogfood identity ready: "
         f"organization={args.organization_id} "
         f"user={args.user_id} membership={args.membership_id} "
-        f"version={args.membership_version}",
+        f"version={args.membership_version}"
+        + (
+            ""
+            if args.file_import_service_principal_id is None
+            else (
+                " file_import_service_principal="
+                f"{args.file_import_service_principal_id}"
+            )
+        ),
         flush=True,
     )