control: compose File source operator commands

STATUS.md

@@ -115,6 +115,7 @@ Follow the ADR for its exact evidence boundary.
 | [0060](./docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md) | Reclaim expired File imports with bounded retries |
 | [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 |
 
 ADR-0065 extends the active File Provider boundary from a flat root to
 deterministic recursive discovery of canonical nested Markdown paths. Each

applications/control.py

@@ -3,15 +3,45 @@
 from __future__ import annotations
 
 import argparse
+import json
 import os
 from collections.abc import Sequence
+from datetime import UTC, datetime, timedelta
+from uuid import UUID, uuid4
 
 from applications.operator_authentication import (
+    CONTROL_OPERATOR_SECRET_ENV,
     LocalOperatorAuthorities,
     LocalOperatorConfiguration,
 )
+from engine.control import (
+    ActivateFileChangeFeed,
+    ActivateFileDeleteObservations,
+    ContextControl,
+    ControlOperation,
+    FileRootRef,
+    RegisterFileSource,
+    SourceManifest,
+    SourceNotAvailable,
+    SourceRef,
+)
+from engine.persistence import (
+    DatabasePurpose,
+    PostgreSQLControlStore,
+    create_database_engine,
+    load_database_configuration,
+)
 from engine.persistence.migrations import migrate_to_head
 
+_OPERATOR_SUBCOMMANDS = frozenset(
+    {
+        "register-file-source",
+        "read-source",
+        "activate-change-feed",
+        "activate-delete-observations",
+    }
+)
+
 
 def _parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(prog="context-engine-control")
@@ -20,19 +50,50 @@ def _parser() -> argparse.ArgumentParser:
         "migrate",
         help="upgrade the configured database to the current schema head",
     )
+    register = subcommands.add_parser(
+        "register-file-source",
+        help="register one logical File root",
+    )
+    _organization_argument(register)
+    register.add_argument("--display-name", required=True)
+    register.add_argument("--root-ref", required=True)
+    register.add_argument("--idempotency-key", required=True)
+    for name, help_text in (
+        ("read-source", "read one registered File source"),
+        ("activate-change-feed", "activate one File source change feed"),
+        (
+            "activate-delete-observations",
+            "activate one File source delete-observation capability",
+        ),
+    ):
+        source_command = subcommands.add_parser(name, help=help_text)
+        _organization_argument(source_command)
+        source_command.add_argument("--source-ref", required=True)
     return parser
 
 
+def _organization_argument(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--organization-id", required=True)
+
+
 def main(argv: Sequence[str] | None = None) -> None:
     parser = _parser()
     arguments = parser.parse_args(argv)
-    if arguments.subcommand != "migrate":
+    if arguments.subcommand == "migrate":
+        try:
+            revision = migrate_to_head()
+        except Exception:  # The local process must never render connection details.
+            parser.exit(1, "context-engine-control: migration refused\n")
+        print(revision, flush=True)
+        return
+    if arguments.subcommand not in _OPERATOR_SUBCOMMANDS:
         parser.error("unknown operation")
     try:
-        revision = migrate_to_head()
-    except Exception:  # The local process must never render connection details.
-        parser.exit(1, "context-engine-control: migration refused\n")
-    print(revision, flush=True)
+        manifest = _run_operator_subcommand(arguments)
+        rendered = _manifest_json(manifest)
+    except Exception:  # Operator refusals disclose no supplied or trusted facts.
+        parser.exit(1, "context-engine-control: operation refused\n")
+    print(rendered, flush=True)
 
 
 def local_operator_authorities() -> LocalOperatorAuthorities | None:
@@ -44,5 +105,98 @@ def local_operator_authorities() -> LocalOperatorAuthorities | None:
     return configuration.authorities()
 
 
+def _run_operator_subcommand(arguments: argparse.Namespace) -> SourceManifest:
+    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:
+        control = ContextControl(
+            store=PostgreSQLControlStore(engine, clock=clock),
+            authority=authorities.control,
+            clock=clock,
+        )
+        with authorities.control.authorize(
+            opaque_credential=opaque_credential,
+            operation=operation,
+            request_id=f"local-{arguments.subcommand}-{uuid4().hex}",
+        ) as call:
+            if call.organization_id != organization_id:
+                raise SourceNotAvailable
+            if operation is ControlOperation.REGISTER_SOURCE:
+                return control.register_source(
+                    call,
+                    RegisterFileSource(
+                        display_name=arguments.display_name,
+                        root_ref=FileRootRef(arguments.root_ref),
+                        idempotency_key=arguments.idempotency_key,
+                    ),
+                )
+            source_ref = SourceRef(UUID(arguments.source_ref))
+            if operation is ControlOperation.READ_SOURCE:
+                return control.read_source(call, source_ref)
+            if operation is ControlOperation.ACTIVATE_FILE_CHANGE_FEED:
+                return control.activate_file_change_feed(
+                    call,
+                    ActivateFileChangeFeed(source_ref),
+                )
+            if operation is ControlOperation.ACTIVATE_FILE_DELETE_OBSERVATIONS:
+                return control.activate_file_delete_observations(
+                    call,
+                    ActivateFileDeleteObservations(source_ref),
+                )
+            raise SourceNotAvailable
+    finally:
+        engine.dispose()
+
+
+def _operation(subcommand: str) -> ControlOperation:
+    operations = {
+        "register-file-source": ControlOperation.REGISTER_SOURCE,
+        "read-source": ControlOperation.READ_SOURCE,
+        "activate-change-feed": ControlOperation.ACTIVATE_FILE_CHANGE_FEED,
+        "activate-delete-observations": (
+            ControlOperation.ACTIVATE_FILE_DELETE_OBSERVATIONS
+        ),
+    }
+    try:
+        return operations[subcommand]
+    except KeyError:
+        raise SourceNotAvailable from None
+
+
+def _manifest_json(manifest: SourceManifest) -> str:
+    if type(manifest) is not SourceManifest:
+        raise SourceNotAvailable
+    document = {
+        "activeVersion": {
+            "capabilities": manifest.active_version.capabilities.document(),
+            "createdAt": _timestamp(manifest.active_version.created_at),
+            "kind": manifest.active_version.kind.value,
+            "rootRef": manifest.active_version.root_ref.value,
+            "versionRef": str(manifest.active_version.version_ref),
+        },
+        "createdAt": _timestamp(manifest.created_at),
+        "displayName": manifest.display_name,
+        "kind": manifest.kind.value,
+        "sourceRef": str(manifest.source_ref.value),
+    }
+    return json.dumps(document, separators=(",", ":"), sort_keys=True)
+
+
+def _timestamp(value: datetime) -> str:
+    if type(value) is not datetime or value.utcoffset() != timedelta(0):
+        raise SourceNotAvailable
+    return value.isoformat().replace("+00:00", "Z")
+
+
 if __name__ == "__main__":
     main()

docs/decisions/0070-activate-file-change-feed-from-registration.md

@@ -0,0 +1,69 @@
+---
+name: adr-0070-activate-file-change-feed-from-registration
+version: "1.0.0"
+description: >
+  Permit the existing change-feed Control operation to advance either a
+  registered v1 or import-enabled v2 File source to the same immutable v3
+  capability manifest.
+---
+
+# 0070. Activate a File change feed from registration
+
+- Status: accepted
+- Date: 2026-07-27
+- Refines: ADR-0035, ADR-0037, ADR-0054, ADR-0069
+
+## Context
+
+File source registration creates the immutable v1 capability manifest. The
+first manual import operation can advance v1 to v2 while creating an exact
+audience-bound import job, and change-feed activation historically accepted
+only v2 before creating v3. That ordering reflected the implementation
+sequence, not a security dependency: v3 contains the v2 import capabilities
+plus the change-provider carriers, while activation itself creates no import,
+audience, job, lease, or content.
+
+The local operator workflow exposed the mismatch. A maintainer can register a
+source, but cannot activate its change feed without first naming and scheduling
+one manual file. An operator-invoked initial scan cannot provide that missing
+step because the File provider correctly refuses `readChanges` until v3 is
+active. Requiring a fabricated bootstrap import would create unrelated durable
+work and make an empty registered root impossible to scan.
+
+## Decision
+
+`ACTIVATE_FILE_CHANGE_FEED` may atomically advance an active File source from
+either the exact v1 registration manifest or the exact v2 import manifest to
+the existing server-owned immutable v3 manifest. An already-active exact v3
+remains an idempotent replay. Every other manifest, a disabled or foreign
+source, and every non-Control database caller continue to receive the existing
+generic refusal.
+
+The transition remains inside the existing SECURITY DEFINER database function
+and runs through `ControlOperatorAuthority`, one operation-bound
+`TrustedControlCall`, `ContextControl`, the non-owner Control role, and FORCE
+RLS. The command supplies only a `SourceRef`; it cannot construct a manifest or
+version. Direct v1-to-v3 activation creates only one `SourceVersion` and updates
+the active pointer. It does not read the filesystem or create an acquisition,
+job, audience, WorkerLease, checkpoint, Resource, Revision, or Fragment.
+
+V2 remains valid for manual-import-first sources. It is no longer a mandatory
+ceremonial waypoint for sources whose first acquisition is a change scan.
+
+## Consequences
+
+- A registered empty or populated File root can become scan-capable without a
+  fake manual import.
+- Manual-import-first and scan-first sources converge on the same v3 manifest
+  and retain the same downstream authorization and scheduling boundaries.
+- Change-feed activation does not prove filesystem reachability. The provider's
+  separately configured anchored root registry remains responsible for that
+  check when a scan actually runs.
+- Downgrade restores the v2-only precondition for future calls without
+  rewriting retained immutable v3 source history.
+
+## Revisit trigger
+
+Revisit before change-feed activation creates durable work, accepts a manifest
+other than exact v1/v2/v3, or derives filesystem or audience authority from a
+logical `FileRootRef`.

docs/decisions/README.md

@@ -171,3 +171,4 @@ touched:
 - [0067 — Discover vector candidates in the current Runtime transaction](0067-discover-vector-candidates-in-the-current-runtime-transaction.md)
 - [0068 — Activate the loopback dogfood Runtime](0068-activate-loopback-dogfood-runtime.md)
 - [0069 — Admit an explicit local operator composition](0069-admit-an-explicit-local-operator-composition.md)
+- [0070 — Activate a File change feed from registration](0070-activate-file-change-feed-from-registration.md)

migrations/versions/20260727_0037_direct_file_change_activation.py

@@ -0,0 +1,68 @@
+"""Activate a registered File source directly for change scanning.
+
+Revision ID: 20260727_0037
+Revises: 20260726_0036
+Create Date: 2026-07-27
+"""
+
+# ruff: noqa: E501
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260727_0037"
+down_revision: str | None = "20260726_0036"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_DEFINER = "context_engine_worker_lease_definer"
+_FUNCTION = "context_control_activate_file_change_feed"
+_REGPROCEDURE = f"{_FUNCTION}(uuid,uuid,uuid)"
+_V1 = """{"aclEvidenceMode":"mirrored","authorizeAndProject":"unavailable","batchLimits":"unavailable","checkpoint":"unavailable","checkpointSemantics":"unavailable","consistencyGuarantees":"unavailable","contentKinds":["markdown"],"cursorSemantics":"unavailable","declarationVersion":"file-capabilities-v1","deletion":"unavailable","describeCapabilities":"unavailable","discover":"unavailable","fileSourceAccess":"unavailable","freshness":"unavailable","ingestionJobs":"unavailable","projectionFields":[],"readChanges":"unavailable","resourceKinds":["markdown_document"],"sourceMode":"materialized"}"""
+_V2 = """{"aclEvidenceMode":"mirrored","authorizeAndProject":"unavailable","batchLimits":"unavailable","checkpoint":"unavailable","checkpointSemantics":"unavailable","consistencyGuarantees":"unavailable","contentKinds":["markdown"],"cursorSemantics":"unavailable","declarationVersion":"file-capabilities-v2","deletion":"unavailable","describeCapabilities":"unavailable","discover":"unavailable","fileSourceAccess":"available","freshness":"unavailable","ingestionJobs":"available","projectionFields":[],"readChanges":"unavailable","resourceKinds":["markdown_document"],"sourceMode":"materialized"}"""
+_V2_ONLY = f"""            IF selected_capabilities <> '{_V2}'::jsonb THEN RETURN; END IF;"""
+_V1_OR_V2 = f"""            IF selected_capabilities NOT IN (
+                '{_V1}'::jsonb, '{_V2}'::jsonb
+            ) THEN RETURN; END IF;"""
+
+
+def _function_definition() -> str:
+    definition = (
+        op.get_bind()
+        .execute(
+            sa.text(
+                "SELECT pg_catalog.pg_get_functiondef("
+                f"'public.{_REGPROCEDURE}'::regprocedure)"
+            )
+        )
+        .scalar_one()
+    )
+    if not isinstance(definition, str):
+        raise RuntimeError("File change activation function is unavailable")
+    return definition
+
+
+def _replace_exact(searched: str, replacement: str) -> None:
+    definition = _function_definition()
+    if definition.count(searched) != 1:
+        raise RuntimeError("File change activation function shape was not recognized")
+    replacement_definition = definition.replace(searched, replacement)
+    op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}")
+    op.execute(f"SET LOCAL ROLE {_DEFINER}")
+    op.execute(replacement_definition)
+    op.execute("RESET ROLE")
+    op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}")
+
+
+def upgrade() -> None:
+    """Allow exact v1 or v2 state to advance to the existing immutable v3."""
+
+    _replace_exact(_V2_ONLY, _V1_OR_V2)
+
+
+def downgrade() -> None:
+    """Restore the former v2-only activation precondition."""
+
+    _replace_exact(_V1_OR_V2, _V2_ONLY)