worker: autonomously dispatch scheduled File imports through exact leases

README.md

@@ -256,6 +256,21 @@ ContextEngine 的安全协议依据自身需求与威胁模型独立设计，零
 - [D0 Baseline Candidate](./DESIGN-BASELINE.md)：当前候选状态与尚未关闭的
   evidence gates。
 
+独立 Supply worker 的确定性单周期 File dispatch 使用
+`context-engine-worker --dispatch-file-once`。生产长运行入口是
+`context-engine-worker --dispatch-files`；它以服务端固定的一秒间隔轮询无工作结果，
+并在 `SIGTERM` / `SIGINT` 时结束。两种入口都只读取 role-specific scheduler、
+worker URL、WorkerLease signing key 和服务端 JSON root registry
+(`CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON`)；调用方不得提供
+Organization、Source、job 或 token。输出仅包含 `dispatched` / `no_work` / `refused`；Provider
+polling、过期 lease reclaim、retry/dead-letter 与 delete execution 仍未激活。Worker
+基础设施不可用会终止 dispatch，不会继续 claim 并滞留后续 job。
+文件/内容失败仅在该 job 已持久化为 terminal failed 或当前 authority 拒绝该精确
+failure transition 后返回 `refused` 并继续调度；failure recording 基础设施不可用仍会
+终止 dispatch。
+Lease 的立即验证使用 worker PostgreSQL 时钟，与数据库签发时间保持同一时间域，
+不依赖 worker host clock 对齐。
+
 当前除固定 commit 的四仓静态证据与仓库内设计拆解外，已有
 [`compose.yaml`](./compose.yaml) 固定的真实 PostgreSQL + pgvector 基础 harness，
 以及首个 Organization-owned 代表表的 RLS 动态证据。

applications/worker.py

@@ -3,23 +3,34 @@
 import argparse
 import json
 import os
+import signal
 import threading
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, field
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Protocol
 from uuid import UUID
 
+from sqlalchemy import Engine, text
+from sqlalchemy.exc import SQLAlchemyError
+
 from adapters.file_source import FileReadLimits, FileRootRegistry
 from engine import BUILD_IDENTIFIER
 from engine.control import FileImportReceiver, FileRootRef, SourceRef
 from engine.persistence import (
     DatabasePurpose,
+    FileDispatchLease,
+    FileDispatchNoWork,
     FileImportLeaseRedemption,
+    FileImportRefused,
+    FileImportUnavailable,
+    PostgreSQLFileDispatchAuthority,
     PostgreSQLFileImportWorker,
     create_database_engine,
     load_database_configuration,
 )
+from engine.persistence.role_guard import assert_worker_role
 from engine.persistence.worker_jobs import (
     WorkerLeaseRedemption,
     WorkerNoOpCompletion,
@@ -31,8 +42,11 @@
     WorkerLeaseCodec,
     WorkerLeaseKeyring,
     WorkerLeaseToken,
+    WorkNotAvailable,
 )
 
+_FILE_DISPATCH_POLL_SECONDS = 1.0
+
 
 class WorkerNoOpCompletionAuthority(Protocol):
     """Application port for one verified persistent no-op completion."""
@@ -42,6 +56,70 @@ def complete_noop(
     ) -> WorkerNoOpCompletion: ...
 
 
+class FileDispatchAuthority(Protocol):
+    """Application port for database-selected first-attempt File work."""
+
+    def claim(self) -> FileDispatchLease | FileDispatchNoWork: ...
+
+
+class FileDispatchWorker(Protocol):
+    """Existing exact File import execution seam."""
+
+    def run(self, redemption: FileImportLeaseRedemption) -> object: ...
+
+
+class FileDispatchWorkerFactory(Protocol):
+    def __call__(self, receiver: FileImportReceiver) -> FileDispatchWorker: ...
+
+
+@dataclass(frozen=True, slots=True)
+class FileDispatchCycleResult:
+    """Content-free process result for one autonomous dispatch cycle."""
+
+    outcome: str
+    status: str = field(default="complete", init=False)
+
+    def __post_init__(self) -> None:
+        if self.outcome not in {"dispatched", "no_work", "refused"}:
+            raise ValueError("File dispatch cycle outcome must remain closed")
+
+
+def dispatch_one_file_import(
+    authority: FileDispatchAuthority,
+    worker_factory: FileDispatchWorkerFactory,
+) -> FileDispatchCycleResult:
+    """Claim and run at most one exact job without caller routing input."""
+
+    claim = authority.claim()
+    if type(claim) is FileDispatchNoWork:
+        return FileDispatchCycleResult("no_work")
+    if type(claim) is not FileDispatchLease:
+        raise TypeError("File dispatch authority returned an invalid result")
+    try:
+        worker_factory(FileImportReceiver(claim.service_principal_id)).run(
+            claim.redemption
+        )
+    except (FileImportRefused, WorkNotAvailable):
+        return FileDispatchCycleResult("refused")
+    return FileDispatchCycleResult("dispatched")
+
+
+def dispatch_file_imports_until_stopped(
+    authority: FileDispatchAuthority,
+    worker_factory: FileDispatchWorkerFactory,
+    stop_event: threading.Event,
+    outcome_observer: Callable[[FileDispatchCycleResult], None] | None = None,
+) -> None:
+    """Run bounded single-job cycles until process shutdown is requested."""
+
+    while not stop_event.is_set():
+        result = dispatch_one_file_import(authority, worker_factory)
+        if outcome_observer is not None:
+            outcome_observer(result)
+        if result.outcome == "no_work":
+            stop_event.wait(_FILE_DISPATCH_POLL_SECONDS)
+
+
 def complete_persistent_noop_job(
     authority: WorkerNoOpCompletionAuthority,
     redemption: WorkerLeaseRedemption,
@@ -63,17 +141,7 @@ def _required_environment(name: str) -> str:
 def _run_one_file_import() -> int:
     """Consume one exact, signed File job in the independent Supply process."""
 
-    signing_key_hex = _required_environment(
-        "CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX"
-    )
-    if len(signing_key_hex) != 64:
-        raise ValueError("Supply worker configuration is not available")
-    try:
-        signing_key = bytes.fromhex(signing_key_hex)
-    except ValueError:
-        raise ValueError("Supply worker configuration is not available") from None
-    if len(signing_key) != 32:
-        raise ValueError("Supply worker configuration is not available")
+    signing_key = _worker_signing_key()
     configuration = load_database_configuration(DatabasePurpose.SUPPLY_WORKER)
     engine = create_database_engine(configuration)
     roots = FileRootRegistry(
@@ -143,7 +211,175 @@ def _run_one_file_import() -> int:
         engine.dispose()
 
 
-def run(*, test_mode: bool, run_file_job: bool = False) -> int:
+def _worker_signing_key() -> bytes:
+    signing_key_hex = _required_environment(
+        "CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX"
+    )
+    if len(signing_key_hex) != 64:
+        raise ValueError("Supply worker configuration is not available")
+    try:
+        signing_key = bytes.fromhex(signing_key_hex)
+    except ValueError:
+        raise ValueError("Supply worker configuration is not available") from None
+    if len(signing_key) != 32:
+        raise ValueError("Supply worker configuration is not available")
+    return signing_key
+
+
+def _file_dispatch_root_bindings() -> dict[FileRootRef, Path]:
+    """Load the server-owned registry for every root this dispatcher serves."""
+
+    raw_registry = _required_environment("CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON")
+    try:
+        document = json.loads(raw_registry)
+    except json.JSONDecodeError:
+        raise ValueError("Supply worker configuration is not available") from None
+    if type(document) is not dict or not document:
+        raise ValueError("Supply worker configuration is not available")
+    bindings: dict[FileRootRef, Path] = {}
+    for raw_ref, raw_path in document.items():
+        if (
+            type(raw_ref) is not str
+            or type(raw_path) is not str
+            or not raw_path
+            or raw_path != raw_path.strip()
+        ):
+            raise ValueError("Supply worker configuration is not available")
+        bindings[FileRootRef(raw_ref)] = Path(raw_path)
+    return bindings
+
+
+def _file_dispatch_roots() -> FileRootRegistry:
+    return FileRootRegistry(
+        _file_dispatch_root_bindings(),
+        limits=FileReadLimits(max_file_bytes=4_096),
+    )
+
+
+def _worker_database_time(engine: Engine) -> datetime:
+    """Read the worker authority's clock for immediate lease verification."""
+
+    try:
+        with engine.connect() as connection:
+            assert_worker_role(connection)
+            checked_at = connection.execute(
+                text("SELECT pg_catalog.date_trunc('second', clock_timestamp())")
+            ).scalar_one()
+    except (SQLAlchemyError, AssertionError, ValueError):
+        raise FileImportUnavailable("File import clock is unavailable") from None
+    if type(checked_at) is not datetime or checked_at.tzinfo is None:
+        raise FileImportUnavailable("File import clock is unavailable")
+    return checked_at.astimezone(UTC)
+
+
+def _run_file_dispatch(*, single_cycle: bool) -> int:
+    """Run configured autonomous File dispatch without caller routing facts."""
+
+    codec = WorkerLeaseCodec(
+        WorkerLeaseKeyring(active_version=1, keys={1: _worker_signing_key()})
+    )
+    scheduler_engine = create_database_engine(
+        load_database_configuration(DatabasePurpose.SUPPLY_SCHEDULER)
+    )
+    worker_engine = create_database_engine(
+        load_database_configuration(DatabasePurpose.SUPPLY_WORKER)
+    )
+    root_bindings = _file_dispatch_root_bindings()
+    roots = FileRootRegistry(
+        root_bindings,
+        limits=FileReadLimits(max_file_bytes=4_096),
+    )
+    try:
+        authority = PostgreSQLFileDispatchAuthority(
+            scheduler_engine,
+            codec,
+            configured_root_refs=tuple(
+                root_ref.value for root_ref in root_bindings
+            ),
+        )
+
+        def worker_factory(receiver: FileImportReceiver) -> PostgreSQLFileImportWorker:
+            return PostgreSQLFileImportWorker(
+                worker_engine,
+                codec,
+                receiver,
+                roots,
+                MarkdownCompilerConfig("markdown-config-v1"),
+                clock=lambda: _worker_database_time(worker_engine),
+            )
+
+        if single_cycle:
+            result = dispatch_one_file_import(authority, worker_factory)
+            print(
+                json.dumps(
+                    {
+                        "dispatch": "file.import",
+                        "outcome": result.outcome,
+                        "service": "context-engine-worker",
+                        "status": result.status,
+                    },
+                    sort_keys=True,
+                ),
+                flush=True,
+            )
+        else:
+            stop_event = threading.Event()
+
+            def request_stop(_signum: int, _frame: object) -> None:
+                stop_event.set()
+
+            previous_sigterm = signal.signal(signal.SIGTERM, request_stop)
+            previous_sigint = signal.signal(signal.SIGINT, request_stop)
+            try:
+                print(
+                    json.dumps(
+                        {
+                            "dispatch": "file.import",
+                            "service": "context-engine-worker",
+                            "status": "ready",
+                        },
+                        sort_keys=True,
+                    ),
+                    flush=True,
+                )
+                dispatch_file_imports_until_stopped(
+                    authority,
+                    worker_factory,
+                    stop_event,
+                    outcome_observer=lambda result: print(
+                        json.dumps(
+                            {
+                                "dispatch": "file.import",
+                                "outcome": result.outcome,
+                                "service": "context-engine-worker",
+                                "status": result.status,
+                            },
+                            sort_keys=True,
+                        ),
+                        flush=True,
+                    ),
+                )
+            finally:
+                signal.signal(signal.SIGINT, previous_sigint)
+                signal.signal(signal.SIGTERM, previous_sigterm)
+        return 0
+    finally:
+        roots.close()
+        worker_engine.dispose()
+        scheduler_engine.dispose()
+
+
+def run(
+    *,
+    test_mode: bool,
+    run_file_job: bool = False,
+    dispatch_file_once: bool = False,
+    dispatch_files: bool = False,
+) -> int:
+    if dispatch_file_once:
+        return _run_file_dispatch(single_cycle=True)
+    if dispatch_files:
+        return _run_file_dispatch(single_cycle=False)
     if run_file_job:
         return _run_one_file_import()
     Runtime(required_kernel_dependencies())
@@ -177,10 +413,33 @@ def main(argv: Sequence[str] | None = None) -> int:
         action="store_true",
         help="consume one exact configured FileImport WorkerLease and exit",
     )
+    parser.add_argument(
+        "--dispatch-file-once",
+        action="store_true",
+        help="claim and execute at most one eligible scheduled File import",
+    )
+    parser.add_argument(
+        "--dispatch-files",
+        action="store_true",
+        help="continuously claim eligible scheduled File imports until shutdown",
+    )
     args = parser.parse_args(argv)
-    if args.test_mode and args.run_file_job:
-        parser.error("--test-mode and --run-file-job are mutually exclusive")
-    return run(test_mode=args.test_mode, run_file_job=args.run_file_job)
+    selected_modes = sum(
+        (
+            args.test_mode,
+            args.run_file_job,
+            args.dispatch_file_once,
+            args.dispatch_files,
+        )
+    )
+    if selected_modes > 1:
+        parser.error("worker execution modes are mutually exclusive")
+    return run(
+        test_mode=args.test_mode,
+        run_file_job=args.run_file_job,
+        dispatch_file_once=args.dispatch_file_once,
+        dispatch_files=args.dispatch_files,
+    )
 
 
 if __name__ == "__main__":

compose.yaml

@@ -20,6 +20,8 @@ services:
       CONTEXT_ENGINE_RUNTIME_PASSWORD: ${CONTEXT_ENGINE_RUNTIME_PASSWORD:?CONTEXT_ENGINE_RUNTIME_PASSWORD is required}
       CONTEXT_ENGINE_WORKER_ROLE: ${CONTEXT_ENGINE_WORKER_ROLE:?CONTEXT_ENGINE_WORKER_ROLE is required}
       CONTEXT_ENGINE_WORKER_PASSWORD: ${CONTEXT_ENGINE_WORKER_PASSWORD:?CONTEXT_ENGINE_WORKER_PASSWORD is required}
+      CONTEXT_ENGINE_SCHEDULER_ROLE: ${CONTEXT_ENGINE_SCHEDULER_ROLE:?CONTEXT_ENGINE_SCHEDULER_ROLE is required}
+      CONTEXT_ENGINE_SCHEDULER_PASSWORD: ${CONTEXT_ENGINE_SCHEDULER_PASSWORD:?CONTEXT_ENGINE_SCHEDULER_PASSWORD is required}
       CONTEXT_ENGINE_LEARNING_ROLE: ${CONTEXT_ENGINE_LEARNING_ROLE:?CONTEXT_ENGINE_LEARNING_ROLE is required}
       CONTEXT_ENGINE_LEARNING_PASSWORD: ${CONTEXT_ENGINE_LEARNING_PASSWORD:?CONTEXT_ENGINE_LEARNING_PASSWORD is required}
       CONTEXT_ENGINE_SECURITY_OPERATOR_ROLE: ${CONTEXT_ENGINE_SECURITY_OPERATOR_ROLE:?CONTEXT_ENGINE_SECURITY_OPERATOR_ROLE is required}