feat(problems): publish immutable versions and fence artifact access

backend/src/kf_oj/api/app.py

@@ -8,11 +8,17 @@ from starlette.responses import JSONResponse
 from kf_oj import __version__
 from kf_oj.api.body_limits import BodyLimitMiddleware
 from kf_oj.api.v1.identity import router as identity_router
+from kf_oj.api.v1.problems import router as problems_router
 from kf_oj.api.v1.router import router
+from kf_oj.api.v1.worker_artifacts import router as worker_router
 from kf_oj.core.logging import configure_logging
 from kf_oj.core.settings import Settings
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.infra.artifacts.store import ArtifactStore
 from kf_oj.infra.db.session import build_engine, build_session_factory
 from kf_oj.modules.identity.service import IdentityService
+from kf_oj.modules.problems.service import ProblemService
+from kf_oj.modules.submissions.worker_access import WorkerAccess
 
 
 def create_app(settings: Settings | None = None) -> FastAPI:
@@ -24,6 +30,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
         logger = logging.getLogger("kf_oj.api")
         logger.info("Platform API started")
         engine = None
+        store = None
         try:
             if settings.database_url is not None:
                 engine = build_engine(settings)
@@ -31,11 +38,30 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                 app.state.identity = IdentityService(
                     app.state.sessions, lifetime_seconds=settings.session_lifetime_seconds
                 )
+                artifacts = None
+                if settings.artifact_root is not None:
+                    store = ArtifactStore(
+                        settings.artifact_root,
+                        max_bytes=settings.artifact_max_bytes,
+                        max_files=settings.artifact_max_files,
+                    )
+                    artifacts = ArtifactRepository(store)
+                app.state.problems = ProblemService(
+                    app.state.sessions, artifacts, settings.judge_admission
+                )
+                if artifacts is not None:
+                    app.state.worker_access = WorkerAccess(
+                        app.state.sessions, app.state.problems, artifacts
+                    )
             yield
         finally:
             app.state.identity = None
+            app.state.problems = None
+            app.state.worker_access = None
             if engine is not None:
                 await engine.dispose()
+            if store is not None:
+                store.close()
             logger.info("Platform API stopped")
 
     app = FastAPI(
@@ -56,4 +82,6 @@ def create_app(settings: Settings | None = None) -> FastAPI:
 
     app.include_router(router, prefix="/api/v1")
     app.include_router(identity_router, prefix="/api/v1")
+    app.include_router(problems_router, prefix="/api/v1")
+    app.include_router(worker_router, prefix="/api/v1")
     return app

backend/src/kf_oj/api/v1/problems.py

@@ -0,0 +1,34 @@
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException, Query, Request
+
+from kf_oj.contracts.problems import ProblemDetail, ProblemPage
+from kf_oj.modules.problems.public import ProblemUnavailable
+from kf_oj.modules.problems.service import ProblemService
+
+router = APIRouter(prefix="/problems", tags=["problems"])
+
+
+def service(request: Request) -> ProblemService:
+    component = getattr(request.app.state, "problems", None)
+    if component is None:
+        raise HTTPException(503, "problem service unavailable")
+    return component
+
+
+@router.get("", response_model=ProblemPage)
+async def list_problems(
+    request: Request,
+    after: Annotated[str | None, Query(pattern=r"^[a-z0-9][a-z0-9_-]{0,63}$")] = None,
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+) -> ProblemPage:
+    return await service(request).list(after=after, limit=limit)
+
+
+@router.get("/versions/{version_id}", response_model=ProblemDetail)
+async def problem_version(version_id: UUID, request: Request) -> ProblemDetail:
+    try:
+        return await service(request).detail(version_id)
+    except ProblemUnavailable:
+        raise HTTPException(404, "problem version unavailable") from None

backend/src/kf_oj/api/v1/worker_artifacts.py

@@ -0,0 +1,96 @@
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Header, HTTPException, Request, Response, Security
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from kf_oj.contracts.judging import JobManifest, LeaseBinding
+from kf_oj.modules.submissions.public import WorkerAccessDenied, WorkerReadBusy
+from kf_oj.modules.submissions.worker_access import WorkerAccess
+
+router = APIRouter(prefix="/internal/evaluations", tags=["worker artifacts"])
+bearer = HTTPBearer(auto_error=False, scheme_name="WorkerLease")
+
+
+def grant(
+    evaluation_id: UUID,
+    attempt_id: UUID,
+    worker: UUID,
+    fence: int,
+    credentials: HTTPAuthorizationCredentials | None,
+) -> tuple[LeaseBinding, str]:
+    if (
+        credentials is None
+        or credentials.scheme.lower() != "bearer"
+        or len(credentials.credentials) != 64
+    ):
+        raise HTTPException(401, "worker lease required")
+    try:
+        return LeaseBinding(
+            evaluation_id=evaluation_id,
+            attempt_id=attempt_id,
+            worker_id=worker,
+            fencing_version=fence,
+        ), credentials.credentials
+    except ValueError:
+        raise HTTPException(404, "worker grant rejected") from None
+
+
+def component(request: Request) -> WorkerAccess:
+    access = getattr(request.app.state, "worker_access", None)
+    if access is None:
+        raise HTTPException(503, "worker artifact service unavailable")
+    return access
+
+
+@router.get("/{evaluation_id}/attempts/{attempt_id}/manifest", response_model=JobManifest)
+async def manifest(
+    evaluation_id: UUID,
+    attempt_id: UUID,
+    request: Request,
+    worker: Annotated[UUID, Header(alias="X-Worker-ID")],
+    fence: Annotated[int, Header(alias="X-Fencing-Version", ge=1)],
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Security(bearer)],
+):
+    binding, token = grant(evaluation_id, attempt_id, worker, fence, credentials)
+    try:
+        payload = await component(request).manifest(binding, token)
+    except WorkerAccessDenied:
+        raise HTTPException(404, "worker grant rejected") from None
+    except WorkerReadBusy:
+        raise HTTPException(429, "worker reads temporarily limited") from None
+    return Response(payload, media_type="application/json", headers={"Cache-Control": "no-store"})
+
+
+@router.get(
+    "/{evaluation_id}/attempts/{attempt_id}/artifacts/{digest}",
+    response_class=Response,
+    responses={
+        200: {
+            "content": {
+                "application/octet-stream": {"schema": {"type": "string", "format": "binary"}}
+            }
+        }
+    },
+)
+async def artifact(
+    evaluation_id: UUID,
+    attempt_id: UUID,
+    digest: str,
+    request: Request,
+    worker: Annotated[UUID, Header(alias="X-Worker-ID")],
+    fence: Annotated[int, Header(alias="X-Fencing-Version", ge=1)],
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Security(bearer)],
+):
+    binding, token = grant(evaluation_id, attempt_id, worker, fence, credentials)
+    try:
+        payload = await component(request).artifact(binding, token, digest)
+    except WorkerAccessDenied:
+        raise HTTPException(404, "worker grant rejected") from None
+    except WorkerReadBusy:
+        raise HTTPException(429, "worker reads temporarily limited") from None
+    return Response(
+        payload,
+        media_type="application/octet-stream",
+        headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"},
+    )

backend/src/kf_oj/contracts/judging.py

@@ -162,6 +162,17 @@ class JobManifest(WireModel):
         return self
 
 
+def manifest_bytes(manifest: JobManifest) -> bytes:
+    """Stable UTF-8 bytes for persisted snapshots and their SHA-256 identity."""
+    return json.dumps(
+        manifest.model_dump(mode="json"),
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        allow_nan=False,
+    ).encode("utf-8")
+
+
 class LeaseBinding(WireModel):
     evaluation_id: Identifier
     attempt_id: Identifier

backend/src/kf_oj/contracts/problems.py

@@ -0,0 +1,31 @@
+from typing import Annotated, Literal
+
+from pydantic import Field
+
+from kf_oj.contracts.judging import Identifier, Label, Language, RunLimits, WireModel
+
+
+class PublicSample(WireModel):
+    encoding: Literal["base64"] = "base64"
+    input: Annotated[str, Field(max_length=87384)]
+    output: Annotated[str, Field(max_length=87384)]
+
+
+class ProblemSummary(WireModel):
+    id: Identifier
+    slug: Label
+    version_id: Identifier
+    revision: int
+    title: str
+    languages: tuple[Language, ...]
+
+
+class ProblemDetail(ProblemSummary):
+    statement: str
+    samples: tuple[PublicSample, ...]
+    run_limits: RunLimits
+
+
+class ProblemPage(WireModel):
+    items: tuple[ProblemSummary, ...]
+    next_after: Label | None

backend/src/kf_oj/core/settings.py

@@ -1,8 +1,11 @@
+from pathlib import Path
 from typing import Literal
 
 from pydantic import Field, SecretStr, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from kf_oj.contracts.judging import AdmissionPolicy
+
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="KF_OJ_", env_file=".env", extra="ignore")
@@ -13,6 +16,10 @@ class Settings(BaseSettings):
     amqp_url: SecretStr | None = None
     browser_origins: tuple[str, ...] = ("http://localhost:5173", "http://localhost:8000")
     session_lifetime_seconds: int = Field(default=43200, ge=300, le=86400)
+    artifact_root: Path | None = None
+    artifact_max_bytes: int = Field(default=536870912, ge=1, le=16 * 1024**3)
+    artifact_max_files: int = Field(default=4096, ge=1, le=65536)
+    judge_admission: AdmissionPolicy | None = None
 
     @model_validator(mode="after")
     def secure_origins(self):

backend/src/kf_oj/infra/artifacts/README.md

@@ -0,0 +1,20 @@
+# 不可变工件
+
+存储根配置于 `KF_OJ_ARTIFACT_ROOT`，由服务账号持有，目录 0700、文件 0600。
+单件至多 64 MiB；默认总字节 512 MiB / 文件 4096，可在有界配置中调整。
+发布采用摘要名称、临时文件、文件/目录 fsync 和跨进程锁；已有工件必须通过长度/摘要核验，
+拒绝静默覆盖/修复，普通读取不跟随链接。锁等待最多五秒。临时文件也计入容量。
+
+适配器执行已经授权的操作，不自行识别用户、租约或题库。公开样例由 problems 的公开 DTO
+提供；个人源码由 submissions 的已验证主体与归属端口读取；隐藏数据只经有效当前租约取得。
+没有匿名工件下载或由文件路径选择的读取接口。
+
+发布事务先取得 digest 锁，再持久化字节与元数据，业务引用最后提交；事务回滚最多留下
+无元数据的耐久孤立文件，不留下数据库引用指向未发布字节。回收使用同一 digest 锁，
+业务引用的数据库生成 ready 状态及复合外键阻止被引用元数据转为 deleting。
+
+`uv run --locked kf-artifacts --limit 100 [--after <digest>]` 执行有限回收批次：恢复 deleting、
+标记超过一天的未引用元数据、删除耐久字节/元数据及超过一天的无元数据文件/临时文件。
+每次扫描至多 100 行；显示游标，后续调用接续，避免固定前页上的保留工件阻塞扫描。
+引用工件不能被回收。文件删除后事务失败仍保持 deleting，下一批可恢复；新引用无法加入。
+需要已迁移数据库和有限平台权限；回收不遍历业务域 ORM，也不删除不可变题目版本。

backend/src/kf_oj/infra/artifacts/__init__.py

@@ -1 +1 @@
-"""Reserved for authenticated artifact access and digest verification."""
+"""Digest-addressed immutable storage; business ports enforce authorization."""

backend/src/kf_oj/infra/artifacts/cli.py

@@ -0,0 +1,48 @@
+import argparse
+import asyncio
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from kf_oj.core.settings import Settings
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.infra.artifacts.store import ArtifactStore
+from kf_oj.infra.db.session import build_engine, build_session_factory
+
+
+async def run(limit: int, after: str | None) -> None:
+    settings = Settings()
+    if settings.artifact_root is None:
+        raise ValueError("artifact root required")
+    store = ArtifactStore(
+        settings.artifact_root,
+        max_bytes=settings.artifact_max_bytes,
+        max_files=settings.artifact_max_files,
+    )
+    engine = build_engine(settings)
+    try:
+        sessions = build_session_factory(engine)
+        repository = ArtifactRepository(store)
+        recovered = await repository.finish_deleting(sessions, limit=limit)
+        marked, cursor = await repository.collect_metadata(sessions, limit=limit, after=after)
+        removed = await repository.finish_deleting(sessions, limit=limit)
+        orphans = await repository.collect_orphans(sessions, limit=limit)
+        print(f"Recovered {recovered}; marked {marked}; removed {removed}; orphans {orphans}.")
+        print(f"Next metadata scan cursor: {cursor or 'complete'}")
+    finally:
+        await engine.dispose()
+        store.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Collect bounded old unreferenced artifacts")
+    parser.add_argument("--limit", type=int, default=100, choices=range(1, 101))
+    parser.add_argument("--after")
+    args = parser.parse_args()
+    try:
+        asyncio.run(run(args.limit, args.after))
+    except (ValueError, OSError, SQLAlchemyError):
+        parser.exit(1, "Artifact collection rejected; check storage and database configuration.\n")
+
+
+if __name__ == "__main__":
+    main()

backend/src/kf_oj/infra/artifacts/repository.py

@@ -0,0 +1,144 @@
+"""Byte durability precedes metadata. Digest locks span caller reference transactions."""
+
+import asyncio
+from datetime import timedelta
+
+from sqlalchemy import func, select, text, update
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from kf_oj.contracts.judging import ArtifactReference
+from kf_oj.infra.artifacts.models import Artifact
+from kf_oj.infra.artifacts.store import ArtifactError, ArtifactStore
+
+
+async def lock_digest(session: AsyncSession, digest: str) -> None:
+    await session.execute(
+        text("SELECT pg_advisory_xact_lock(:key)"),
+        {"key": int(digest[:16], 16) & ((1 << 63) - 1)},
+    )
+
+
+class ArtifactRepository:
+    def __init__(self, store: ArtifactStore):
+        self.store = store
+
+    async def put(
+        self, session: AsyncSession, data: bytes, expected: ArtifactReference | None = None
+    ) -> ArtifactReference:
+        reference = self.store.reference(data)
+        if expected is not None and reference != expected:
+            raise ArtifactError("artifact digest or length mismatch")
+        await lock_digest(session, reference.digest)
+        existing = await session.get(Artifact, reference.digest)
+        if existing is not None and (
+            existing.state != "ready" or existing.bytes != reference.bytes
+        ):
+            raise ArtifactError("artifact metadata unavailable")
+        await asyncio.to_thread(self.store.put, data, reference)
+        await session.execute(
+            insert(Artifact)
+            .values(digest=reference.digest, bytes=reference.bytes)
+            .on_conflict_do_nothing(index_elements=[Artifact.digest])
+        )
+        return reference
+
+    async def read(self, session: AsyncSession, reference: ArtifactReference) -> bytes:
+        metadata = await session.get(Artifact, reference.digest)
+        if metadata is None or metadata.state != "ready" or metadata.bytes != reference.bytes:
+            raise ArtifactError("artifact metadata unavailable")
+        return await asyncio.to_thread(self.store.read, reference)
+
+    async def check_metadata(self, session: AsyncSession, reference: ArtifactReference) -> None:
+        metadata = await session.get(Artifact, reference.digest)
+        if metadata is None or metadata.state != "ready" or metadata.bytes != reference.bytes:
+            raise ArtifactError("artifact metadata unavailable")
+
+    async def collect_metadata(
+        self,
+        sessions: async_sessionmaker,
+        *,
+        min_age_seconds: int = 86400,
+        limit: int = 100,
+        after: str | None = None,
+    ) -> tuple[int, str | None]:
+        if min_age_seconds < 3600 or not 1 <= limit <= 100:
+            raise ArtifactError("invalid collection budget")
+        if after is not None:
+            ArtifactReference(digest=after, bytes=0)
+        async with sessions() as session:
+            query = select(Artifact.digest).where(
+                Artifact.state == "ready",
+                Artifact.created_at < func.now() - timedelta(seconds=min_age_seconds),
+            )
+            if after is not None:
+                query = query.where(Artifact.digest > after)
+            names = list(await session.scalars(query.order_by(Artifact.digest).limit(limit)))
+        marked = 0
+        for digest in names:
+            async with sessions.begin() as session:
+                try:
+                    async with session.begin_nested():
+                        changed = await self.mark_deleting(session, digest)
+                    marked += int(changed)
+                except IntegrityError:
+                    # Referenced rows are protected by their generated ready-state foreign key.
+                    # No domain ORM imports or reverse dependencies are needed for reclamation.
+                    pass
+        return marked, names[-1] if len(names) == limit else None
+
+    async def collect_orphans(
+        self, sessions: async_sessionmaker, *, min_age_seconds: int = 86400, limit: int = 100
+    ) -> int:
+        # Only filesystem orphans are collected here. Referenced metadata is never deleted.
+        # A durable row cannot become an orphan while another transaction publishes it:
+        # every writer and collector holds the same transaction-scoped digest lock.
+        names = await asyncio.to_thread(
+            self.store.old_digests, min_age_seconds=min_age_seconds, limit=limit
+        )
+        collected = 0
+        for digest in names:
+            async with sessions.begin() as session:
+                await lock_digest(session, digest)
+                if await session.get(Artifact, digest) is None:
+                    await asyncio.to_thread(self.store.remove, digest)
+                    collected += 1
+        await asyncio.to_thread(
+            self.store.clean_temporary, min_age_seconds=min_age_seconds, limit=limit
+        )
+        return collected
+
+    async def mark_deleting(self, session: AsyncSession, digest: str) -> bool:
+        # FK(digest, 'ready') on every business reference rejects this transition if referenced.
+        await lock_digest(session, digest)
+        result = await session.scalar(
+            update(Artifact)
+            .where(Artifact.digest == digest, Artifact.state == "ready")
+            .values(state="deleting")
+            .returning(Artifact.digest)
+        )
+        return result is not None
+
+    async def finish_deleting(self, sessions: async_sessionmaker, *, limit: int = 100) -> int:
+        if not 1 <= limit <= 100:
+            raise ArtifactError("invalid collection budget")
+        async with sessions() as session:
+            names = list(
+                await session.scalars(
+                    select(Artifact.digest)
+                    .where(Artifact.state == "deleting")
+                    .order_by(Artifact.digest)
+                    .limit(limit)
+                )
+            )
+        count = 0
+        for digest in names:
+            async with sessions.begin() as session:
+                await lock_digest(session, digest)
+                metadata = await session.get(Artifact, digest)
+                if metadata is not None and metadata.state == "deleting":
+                    await asyncio.to_thread(self.store.remove, digest)
+                    await session.delete(metadata)
+                    count += 1
+        return count

backend/src/kf_oj/infra/artifacts/store.py

@@ -0,0 +1,208 @@
+"""Private immutable byte store. Authorization is applied before calling this adapter."""
+
+import fcntl
+import hashlib
+import os
+import re
+import stat
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from uuid import uuid4
+
+from kf_oj.contracts.judging import MAX_PACKAGE_BYTES, ArtifactReference
+
+DIGEST = re.compile(r"[0-9a-f]{64}\Z")
+TEMP = re.compile(r"\.tmp-[0-9a-f]{32}\Z")
+
+
+class ArtifactError(ValueError):
+    """Sanitized artifact failure; never includes paths or private byte content."""
+
+
+class ArtifactStore:
+    def __init__(self, root: Path, *, max_bytes: int = 512 * 1024 * 1024, max_files: int = 4096):
+        if not 1 <= max_bytes <= 16 * 1024**3 or not 1 <= max_files <= 65536:
+            raise ArtifactError("invalid storage capacity")
+        root = root.absolute()
+        root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        if root.resolve() != root:
+            raise ArtifactError("storage root must not contain links")
+        self._directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
+        self.max_bytes = max_bytes
+        self.max_files = max_files
+        info = os.fstat(self._directory)
+        if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
+            os.close(self._directory)
+            raise ArtifactError("storage root must be private and owned")
+        self._closed = False
+
+    def close(self) -> None:
+        if not self._closed:
+            os.close(self._directory)
+            self._closed = True
+
+    @contextmanager
+    def exclusive(self) -> Iterator[None]:
+        # Open a different description for every operation: flock also serializes threads.
+        descriptor = os.open(
+            ".lock",
+            os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC,
+            0o600,
+            dir_fd=self._directory,
+        )
+        try:
+            self._regular(os.fstat(descriptor))
+            deadline = time.monotonic() + 5
+            while True:
+                try:
+                    fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                    break
+                except BlockingIOError:
+                    if time.monotonic() >= deadline:
+                        raise ArtifactError("storage lock deadline exceeded") from None
+                    time.sleep(0.01)
+            yield
+        finally:
+            os.close(descriptor)
+
+    @staticmethod
+    def _regular(info: os.stat_result) -> None:
+        if (
+            not stat.S_ISREG(info.st_mode)
+            or info.st_nlink != 1
+            or info.st_uid != os.getuid()
+            or stat.S_IMODE(info.st_mode) != 0o600
+        ):
+            raise ArtifactError("invalid storage entry")
+
+    def _entries(self) -> list[tuple[str, os.stat_result]]:
+        entries = []
+        with os.scandir(self._directory) as directory:
+            for entry in directory:
+                if entry.name == ".lock":
+                    continue
+                if not DIGEST.fullmatch(entry.name) and not TEMP.fullmatch(entry.name):
+                    raise ArtifactError("unexpected storage entry")
+                info = entry.stat(follow_symlinks=False)
+                self._regular(info)
+                entries.append((entry.name, info))
+                if len(entries) > self.max_files:
+                    raise ArtifactError("storage file capacity exceeded")
+        return entries
+
+    @staticmethod
+    def reference(data: bytes) -> ArtifactReference:
+        if type(data) is not bytes or len(data) > MAX_PACKAGE_BYTES:
+            raise ArtifactError("artifact size outside policy")
+        return ArtifactReference(digest=hashlib.sha256(data).hexdigest(), bytes=len(data))
+
+    def put(self, data: bytes, expected: ArtifactReference | None = None) -> ArtifactReference:
+        reference = self.reference(data)
+        if expected is not None and reference != expected:
+            raise ArtifactError("artifact digest or length mismatch")
+        with self.exclusive():
+            entries = self._entries()
+            if any(name == reference.digest for name, _ in entries):
+                self._read(reference)
+                return reference
+            if len(entries) + 1 > self.max_files or (
+                sum(info.st_size for _, info in entries) + len(data) > self.max_bytes
+            ):
+                raise ArtifactError("storage capacity exhausted")
+            temporary = ".tmp-" + uuid4().hex
+            descriptor = os.open(
+                temporary,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC,
+                0o600,
+                dir_fd=self._directory,
+            )
+            try:
+                with os.fdopen(descriptor, "wb") as output:
+                    output.write(data)
+                    output.flush()
+                    os.fsync(output.fileno())
+                # Locking covers every mutation. Never replace or repair an existing digest.
+                os.rename(
+                    temporary,
+                    reference.digest,
+                    src_dir_fd=self._directory,
+                    dst_dir_fd=self._directory,
+                )
+                os.fsync(self._directory)
+            finally:
+                try:
+                    os.unlink(temporary, dir_fd=self._directory)
+                except FileNotFoundError:
+                    pass
+        return reference
+
+    def _read(self, reference: ArtifactReference) -> bytes:
+        if (
+            type(reference.digest) is not str
+            or not DIGEST.fullmatch(reference.digest)
+            or type(reference.bytes) is not int
+            or not 0 <= reference.bytes <= MAX_PACKAGE_BYTES
+        ):
+            raise ArtifactError("invalid artifact reference")
+        try:
+            descriptor = os.open(
+                reference.digest,
+                os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC,
+                dir_fd=self._directory,
+            )
+            with os.fdopen(descriptor, "rb") as source:
+                info = os.fstat(source.fileno())
+                self._regular(info)
+                if info.st_size != reference.bytes:
+                    raise ArtifactError("artifact digest or length mismatch")
+                data = source.read(reference.bytes + 1)
+                if self.reference(data) != reference:
+                    raise ArtifactError("artifact digest or length mismatch")
+                return data
+        except OSError:
+            raise ArtifactError("artifact unavailable") from None
+
+    def read(self, reference: ArtifactReference) -> bytes:
+        with self.exclusive():
+            return self._read(reference)
+
+    def remove(self, digest: str) -> None:
+        if not DIGEST.fullmatch(digest):
+            raise ArtifactError("invalid artifact identifier")
+        with self.exclusive():
+            try:
+                info = os.stat(digest, dir_fd=self._directory, follow_symlinks=False)
+                self._regular(info)
+                os.unlink(digest, dir_fd=self._directory)
+                os.fsync(self._directory)
+            except FileNotFoundError:
+                pass
+
+    def old_digests(self, *, min_age_seconds: int, limit: int) -> tuple[str, ...]:
+        if not 3600 <= min_age_seconds or not 1 <= limit <= 100:
+            raise ArtifactError("invalid collection budget")
+        with self.exclusive():
+            cutoff = time.time() - min_age_seconds
+            return tuple(
+                name
+                for name, info in self._entries()
+                if DIGEST.fullmatch(name) and info.st_mtime < cutoff
+            )[:limit]
+
+    def clean_temporary(self, *, min_age_seconds: int = 86400, limit: int = 100) -> int:
+        if min_age_seconds < 3600 or not 1 <= limit <= 100:
+            raise ArtifactError("invalid collection budget")
+        with self.exclusive():
+            cutoff = time.time() - min_age_seconds
+            names = [
+                name
+                for name, info in self._entries()
+                if TEMP.fullmatch(name) and info.st_mtime < cutoff
+            ][:limit]
+            for name in names:
+                os.unlink(name, dir_fd=self._directory)
+            if names:
+                os.fsync(self._directory)
+            return len(names)

backend/src/kf_oj/modules/identity/README.md

@@ -13,7 +13,9 @@
 
 登录接口为 `/api/v1/auth/login`，退出 `/logout`，当前主体 `/me`。会话使用随机 256-bit token，
 数据库只保存 token / CSRF 摘要；默认 12 小时有效，单账号保留至多 10 个有效会话。
-账号禁用立即拒绝登录和旧会话。请求只采用实际连接 peer，不信任客户端自报身份或转发地址。
+账号禁用立即拒绝登录和旧会话。请求只采用 ASGI 连接 peer，不读取客户端转发地址。
+受控启动使用 `--no-proxy-headers`，避免服务器中间件预先替换 peer；反向代理后的用户共享
+该连接预算，引入可信代理的独立限流策略之前不能直接打开客户端转发头信任。
 每 peer 至多 10 次登录/15 分钟，预算通过真实数据库串行更新，失败尝试也消耗预算。
 摘要计算最多同时 2 次，超出直接 429；限流表至多 4096 桶且每次有界回收过期桶。
 

backend/src/kf_oj/modules/problems/README.md

@@ -1,5 +1,21 @@
 # problems
 
 负责题目逻辑身份、不可变版本、导入、标签、限制、发布状态与测试工件元数据。
-跨模块访问通过后续 `public.py`；大文件存取通过 `infra/artifacts` 的适配器。
-隐藏数据与公开样例分别授权。当前未创建题库表或导入接口。
+跨模块通过 `public.py` 的不可变判题投影；大文件存取通过 `infra/artifacts` 的适配器。
+公开题面/显式样例与隐藏输入/答案分别授权。公开 API 不返回工件摘要或存储路径。
+
+`uv run --locked kf-problems <directory>` 导入并发布一个新版本。目录必须属于当前操作账号，
+权限 0700；`problem.json` 遵循 `import_format.py`，只引用该目录下的单层普通文件名。
+不解压归档、不跟随链接、不读取任意路径、不执行导入内容。默认清单可由 `--manifest` 指定。
+要求配置 `KF_OJ_ARTIFACT_ROOT` 和可信 `KF_OJ_JUDGE_ADMISSION` JSON；同一准入配置应交付给节点，
+不能由题目放宽。结构示例来自共享 `judge-admission.valid.json`，不是当前宿主隔离验收证明。
+
+清单固定 C17/C++20、工具链/镜像版本、精确字节比较、编译/运行/整个作业预算，并声明按序
+测试点 `{key,input_file,expected_file}` 和最多十对公开样例 `{input_file,output_file}`。
+题面以 `statement_file` 读取 UTF-8。每份样例至多 64 KiB，响应采用 base64 保留二进制数据。
+同一 slug 受事务锁保护，每次发布递增 revision；发布后题面、限额、样例、测试点不可修改。
+旧评测继续绑定旧版本；当前列表指向新版本。工件先持久化，再在同一事务建立引用和发布。
+
+`GET /api/v1/problems` 以 slug 游标分页（`after` / `limit`，最多 100 项），
+`GET /api/v1/problems/versions/{uuid}` 读取已发布版本。禁用逻辑题目时其版本对公开读取不可用。
+内部判题投影仅供授权应用编排使用；外部 Worker 读取由 submissions 的有效租约端口控制。

backend/src/kf_oj/modules/problems/cli.py

@@ -0,0 +1,49 @@
+"""Local, controlled import of ordinary files; no archive extraction or program execution."""
+
+import argparse
+import asyncio
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from kf_oj.core.settings import Settings
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.infra.artifacts.store import ArtifactStore
+from kf_oj.infra.db.session import build_engine, build_session_factory
+from kf_oj.modules.problems.service import ProblemService
+
+
+async def run(root: Path, manifest: str) -> None:
+    settings = Settings()
+    if settings.artifact_root is None or settings.judge_admission is None:
+        raise ValueError("artifact root and trusted judge admission required")
+    store = ArtifactStore(
+        settings.artifact_root,
+        max_bytes=settings.artifact_max_bytes,
+        max_files=settings.artifact_max_files,
+    )
+    engine = build_engine(settings)
+    try:
+        service = ProblemService(
+            build_session_factory(engine), ArtifactRepository(store), settings.judge_admission
+        )
+        version_id = await service.import_directory(root, manifest)
+        print(f"Problem version published: {version_id}")
+    finally:
+        await engine.dispose()
+        store.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Import immutable problem version")
+    parser.add_argument("directory", type=Path)
+    parser.add_argument("--manifest", default="problem.json")
+    args = parser.parse_args()
+    try:
+        asyncio.run(run(args.directory, args.manifest))
+    except (ValueError, OSError, SQLAlchemyError):
+        parser.exit(1, "Problem import rejected; check controlled files, budgets and database.\n")
+
+
+if __name__ == "__main__":
+    main()

backend/src/kf_oj/modules/problems/import_format.py

@@ -0,0 +1,54 @@
+"""Local operator format. Filenames never cross into the judge execution protocol."""
+
+from typing import Annotated, Literal, Self
+
+from pydantic import Field, model_validator
+
+from kf_oj.contracts.judging import (
+    MAX_CASES,
+    MAX_JOB_WALL_US,
+    Label,
+    Language,
+    RunLimits,
+    Toolchain,
+    Version,
+    WireModel,
+)
+
+FileName = Annotated[str, Field(strict=True, pattern=r"^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}$")]
+
+
+class ImportCase(WireModel):
+    key: Label
+    input_file: FileName
+    expected_file: FileName
+
+
+class ImportSample(WireModel):
+    input_file: FileName
+    output_file: FileName
+
+
+class ProblemImport(WireModel):
+    schema_version: Version
+    slug: Label
+    title: Annotated[str, Field(strict=True, min_length=1, max_length=160)]
+    statement_file: FileName
+    languages: Annotated[tuple[Language, ...], Field(min_length=1, max_length=2)]
+    toolchain: Toolchain
+    comparison: Literal["exact_bytes"]
+    compile_limits: RunLimits
+    run_limits: RunLimits
+    job_wall_time_us: Annotated[int, Field(strict=True, ge=1000, le=MAX_JOB_WALL_US)]
+    cases: Annotated[tuple[ImportCase, ...], Field(min_length=1, max_length=MAX_CASES)]
+    samples: Annotated[tuple[ImportSample, ...], Field(max_length=10)]
+
+    @model_validator(mode="after")
+    def unique(self) -> Self:
+        if len(set(self.languages)) != len(self.languages):
+            raise ValueError("duplicate language")
+        if len({case.key for case in self.cases}) != len(self.cases):
+            raise ValueError("duplicate testcase key")
+        if len({(c.input_file, c.expected_file) for c in self.cases}) != len(self.cases):
+            raise ValueError("duplicate testcase reference")
+        return self

backend/src/kf_oj/modules/problems/local_files.py

@@ -0,0 +1,48 @@
+import os
+import re
+import stat
+from pathlib import Path
+
+
+class LocalImportFiles:
+    def __init__(self, root: Path):
+        root = root.absolute()
+        if root.resolve() != root:
+            raise ValueError("import directory must not contain links")
+        self.descriptor = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        info = os.fstat(self.descriptor)
+        if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
+            self.close()
+            raise ValueError("import directory must be private and owned")
+
+    def close(self):
+        os.close(self.descriptor)
+
+    def read(self, filename: str, maximum: int) -> bytes:
+        if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}", filename):
+            raise ValueError("invalid local filename")
+        try:
+            descriptor = os.open(
+                filename, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.descriptor
+            )
+            with os.fdopen(descriptor, "rb") as stream:
+                before = os.fstat(stream.fileno())
+                if (
+                    not stat.S_ISREG(before.st_mode)
+                    or before.st_nlink != 1
+                    or before.st_uid != os.getuid()
+                    or before.st_size > maximum
+                ):
+                    raise ValueError("invalid local input file")
+                data = stream.read(maximum + 1)
+                after = os.fstat(stream.fileno())
+                if (
+                    len(data) > maximum
+                    or before.st_size != len(data)
+                    or before.st_mtime_ns != after.st_mtime_ns
+                    or before.st_ctime_ns != after.st_ctime_ns
+                ):
+                    raise ValueError("local input changed during import")
+                return data
+        except OSError:
+            raise ValueError("local input unavailable") from None

backend/src/kf_oj/modules/problems/public.py

@@ -0,0 +1,27 @@
+"""Authorized immutable version projection; hidden references stay out of public DTOs."""
+
+from dataclasses import dataclass
+from typing import Protocol
+from uuid import UUID
+
+from kf_oj.contracts.judging import Language, RunLimits, TestCase, Toolchain
+
+
+class ProblemUnavailable(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class JudgingVersion:
+    version_id: UUID
+    problem_id: UUID
+    languages: tuple[Language, ...]
+    toolchain: Toolchain
+    compile_limits: RunLimits
+    run_limits: RunLimits
+    job_wall_time_us: int
+    cases: tuple[TestCase, ...]
+
+
+class ProblemPort(Protocol):
+    async def for_submission(self, session, version_id: UUID) -> JudgingVersion: ...

backend/src/kf_oj/modules/problems/repository.py

@@ -0,0 +1,59 @@
+import hashlib
+from uuid import UUID
+
+from sqlalchemy import func, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from kf_oj.modules.problems.models import Problem, ProblemCase, ProblemVersion
+
+
+async def lock_slug(session: AsyncSession, slug: str) -> None:
+    # Two-key namespace is disjoint from the one-key artifact advisory locks.
+    key = int.from_bytes(hashlib.sha256(slug.encode()).digest()[:4], signed=True)
+    await session.execute(text("SELECT pg_advisory_xact_lock(1262907468,:key)"), {"key": key})
+
+
+async def lock_problem(session: AsyncSession, slug: str) -> Problem | None:
+    return await session.scalar(select(Problem).where(Problem.slug == slug).with_for_update())
+
+
+async def next_revision(session: AsyncSession, problem_id: UUID) -> int:
+    latest = await session.scalar(
+        select(func.max(ProblemVersion.revision)).where(ProblemVersion.problem_id == problem_id)
+    )
+    return (latest or 0) + 1
+
+
+async def public_version(session: AsyncSession, version_id: UUID):
+    return (
+        await session.execute(
+            select(ProblemVersion, Problem)
+            .join(Problem, Problem.id == ProblemVersion.problem_id)
+            .where(
+                ProblemVersion.id == version_id,
+                ProblemVersion.published_at.is_not(None),
+                Problem.enabled.is_(True),
+            )
+        )
+    ).one_or_none()
+
+
+async def ordered_cases(session: AsyncSession, version_id: UUID):
+    return list(
+        await session.scalars(
+            select(ProblemCase)
+            .where(ProblemCase.version_id == version_id)
+            .order_by(ProblemCase.position)
+        )
+    )
+
+
+async def list_public(session: AsyncSession, after: str | None, limit: int):
+    query = (
+        select(Problem, ProblemVersion)
+        .join(ProblemVersion, ProblemVersion.id == Problem.active_version_id)
+        .where(Problem.enabled.is_(True), ProblemVersion.published_at.is_not(None))
+    )
+    if after is not None:
+        query = query.where(Problem.slug > after)
+    return list(await session.execute(query.order_by(Problem.slug).limit(limit)))

backend/src/kf_oj/modules/problems/service.py

@@ -0,0 +1,224 @@
+import base64
+from pathlib import Path
+from uuid import UUID, uuid4, uuid5
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from kf_oj.contracts.judging import (
+    MAX_PACKAGE_BYTES,
+    MAX_SOURCE_BYTES,
+    AdmissionPolicy,
+    ArtifactReference,
+    JobManifest,
+    Language,
+    RunLimits,
+    TestCase,
+    Toolchain,
+    decode_wire,
+)
+from kf_oj.contracts.problems import ProblemDetail, ProblemPage, ProblemSummary, PublicSample
+from kf_oj.infra.artifacts.models import Artifact
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.modules.problems import repository
+from kf_oj.modules.problems.import_format import ProblemImport
+from kf_oj.modules.problems.local_files import LocalImportFiles
+from kf_oj.modules.problems.models import Problem, ProblemCase, ProblemVersion
+from kf_oj.modules.problems.public import JudgingVersion, ProblemUnavailable
+
+
+class ProblemService:
+    def __init__(
+        self,
+        sessions: async_sessionmaker,
+        artifacts: ArtifactRepository | None = None,
+        policy: AdmissionPolicy | None = None,
+    ):
+        self.sessions, self.artifacts, self.policy = sessions, artifacts, policy
+
+    async def import_directory(self, root: Path, manifest_file: str = "problem.json") -> UUID:
+        if self.artifacts is None or self.policy is None:
+            raise ValueError("artifact store and trusted node policy required for import")
+        files = LocalImportFiles(root)
+        try:
+            manifest = decode_wire(ProblemImport, files.read(manifest_file, 256 * 1024))
+            statement = files.read(manifest.statement_file, 256 * 1024).decode("utf-8")
+            if not statement or "\x00" in statement or "\x00" in manifest.title:
+                raise ValueError("invalid public text")
+            # Read the complete bounded package before writing anything. Public samples are
+            # separately declared, at most 64 KiB each, and never inferred from hidden tests.
+            remaining = MAX_PACKAGE_BYTES - MAX_SOURCE_BYTES
+            payloads = []
+            for case in manifest.cases:
+                data = files.read(case.input_file, remaining)
+                remaining -= len(data)
+                answer = files.read(
+                    case.expected_file, min(remaining, manifest.run_limits.stdout_bytes)
+                )
+                remaining -= len(answer)
+                payloads.append((data, answer))
+            samples = tuple(
+                PublicSample(
+                    input=base64.b64encode(files.read(s.input_file, 65536)).decode("ascii"),
+                    output=base64.b64encode(files.read(s.output_file, 65536)).decode("ascii"),
+                )
+                for s in manifest.samples
+            )
+        finally:
+            files.close()
+        version_id = uuid4()
+        cases = tuple(
+            TestCase(
+                case_id=uuid5(version_id, case.key),
+                input=self.artifacts.store.reference(data),
+                expected=self.artifacts.store.reference(answer),
+            )
+            for case, (data, answer) in zip(manifest.cases, payloads, strict=True)
+        )
+        check = JobManifest(
+            schema_version=1,
+            kind="judge.manifest",
+            evaluation_id=uuid4(),
+            submission_id=uuid4(),
+            problem_version_id=version_id,
+            language=manifest.languages[0],
+            toolchain=manifest.toolchain,
+            source=ArtifactReference(digest="0" * 64, bytes=MAX_SOURCE_BYTES),
+            cases=cases,
+            comparison=manifest.comparison,
+            compile_limits=manifest.compile_limits,
+            run_limits=manifest.run_limits,
+            job_wall_time_us=manifest.job_wall_time_us,
+        )
+        self.policy.admit(check, 1)
+        async with self.sessions.begin() as session:
+            # Lock existing and not-yet-existing logical slugs before assigning revisions.
+            await repository.lock_slug(session, manifest.slug)
+            problem = await repository.lock_problem(session, manifest.slug)
+            if problem is None:
+                problem = Problem(id=uuid4(), slug=manifest.slug)
+                session.add(problem)
+                await session.flush()
+            if not problem.enabled:
+                raise ProblemUnavailable("problem disabled")
+            revision = await repository.next_revision(session, problem.id)
+            session.add(
+                ProblemVersion(
+                    id=version_id,
+                    problem_id=problem.id,
+                    revision=revision,
+                    title=manifest.title,
+                    statement=statement,
+                    languages=[lang.value for lang in manifest.languages],
+                    toolchain=manifest.toolchain.model_dump(mode="json"),
+                    compile_limits=manifest.compile_limits.model_dump(mode="json"),
+                    run_limits=manifest.run_limits.model_dump(mode="json"),
+                    job_wall_time_us=manifest.job_wall_time_us,
+                    comparison=manifest.comparison,
+                    public_samples=[s.model_dump() for s in samples],
+                )
+            )
+            await session.flush()
+            # Consistent digest lock order avoids deadlocks between overlapping imports/jobs.
+            unique = {
+                ref.digest: (data, ref)
+                for case, (input_bytes, answer_bytes) in zip(cases, payloads, strict=True)
+                for data, ref in [(input_bytes, case.input), (answer_bytes, case.expected)]
+            }
+            for digest in sorted(unique):
+                data, reference = unique[digest]
+                await self.artifacts.put(session, data, reference)
+            for position, case in enumerate(cases):
+                session.add(
+                    ProblemCase(
+                        id=case.case_id,
+                        version_id=version_id,
+                        position=position,
+                        input_digest=case.input.digest,
+                        expected_digest=case.expected.digest,
+                    )
+                )
+            await session.flush()
+            version = await session.get(ProblemVersion, version_id)
+            version.published_at = await session.scalar(select(func.now()))
+            problem.active_version_id = version_id
+        return version_id
+
+    @staticmethod
+    def _summary(problem, version) -> ProblemSummary:
+        return ProblemSummary(
+            id=problem.id,
+            slug=problem.slug,
+            version_id=version.id,
+            revision=version.revision,
+            title=version.title,
+            languages=tuple(version.languages),
+        )
+
+    async def list(self, *, after: str | None = None, limit: int = 20) -> ProblemPage:
+        if not 1 <= limit <= 100:
+            raise ValueError("invalid page limit")
+        async with self.sessions() as session:
+            rows = await repository.list_public(session, after, limit + 1)
+            visible = rows[:limit]
+            return ProblemPage(
+                items=tuple(self._summary(*row) for row in visible),
+                next_after=visible[-1][0].slug if len(rows) > limit else None,
+            )
+
+    async def detail(self, version_id: UUID) -> ProblemDetail:
+        async with self.sessions() as session:
+            row = await repository.public_version(session, version_id)
+            if row is None:
+                raise ProblemUnavailable("problem version unavailable")
+            version, problem = row
+            return ProblemDetail(
+                **self._summary(problem, version).model_dump(),
+                statement=version.statement,
+                samples=tuple(PublicSample.model_validate(s) for s in version.public_samples),
+                run_limits=RunLimits.model_validate(version.run_limits),
+            )
+
+    async def for_submission(self, session, version_id: UUID) -> JudgingVersion:
+        if self.artifacts is None:
+            raise ProblemUnavailable("artifact service unavailable")
+        row = await repository.public_version(session, version_id)
+        if row is None:
+            raise ProblemUnavailable("problem version unavailable")
+        version, problem = row
+        cases = await repository.ordered_cases(session, version.id)
+        refs = []
+        digests = {d for case in cases for d in (case.input_digest, case.expected_digest)}
+        records = {
+            a.digest: a
+            for a in await session.scalars(select(Artifact).where(Artifact.digest.in_(digests)))
+        }
+        for case in cases:
+            input_record = records.get(case.input_digest)
+            answer_record = records.get(case.expected_digest)
+            if (
+                not input_record
+                or not answer_record
+                or input_record.state != "ready"
+                or answer_record.state != "ready"
+            ):
+                raise ProblemUnavailable("problem artifacts unavailable")
+            refs.append(
+                TestCase(
+                    case_id=case.id,
+                    input=ArtifactReference(digest=case.input_digest, bytes=input_record.bytes),
+                    expected=ArtifactReference(
+                        digest=case.expected_digest, bytes=answer_record.bytes
+                    ),
+                )
+            )
+        return JudgingVersion(
+            version.id,
+            problem.id,
+            tuple(Language(v) for v in version.languages),
+            Toolchain.model_validate(version.toolchain),
+            RunLimits.model_validate(version.compile_limits),
+            RunLimits.model_validate(version.run_limits),
+            version.job_wall_time_us,
+            tuple(refs),
+        )

backend/src/kf_oj/modules/submissions/README.md

@@ -3,4 +3,16 @@
 负责提交幂等请求、不可变源码引用、创建初始评测、历史查询与学生可见结果。
 创建提交、评测和待投递事件必须处于同一事务。
 协调器拥有领取后的执行状态与最终结果写入权；本模块不能绕过协议覆盖终态。
-后续 `public.py` 提供授权后的学习证据入口。当前未实现提交接口。
+`queries.py` 提供已验证主体的私有源码读取，其他账号不可读取。
+`worker_access.py` 核对当前 evaluation / attempt / Worker / fencing、未撤销及未过期凭证、
+固定 submission 与版本及任务清单摘要；只允许读取该清单声明的工件。
+内部 API 使用 WorkerLease bearer 和 X-Worker-ID / X-Fencing-Version，不使用用户会话授权。
+允许状态为 claimed / compiling / running；取消请求、终态和旧 attempt 均拒绝。
+字节准备后再次核对当前租约；最多同时两个有限读取，超过直接 429，不排无界等待队列。
+
+任务清单为 UTF-8 JSON、对象键递归排序、无额外空白，并以这些确切字节的 SHA-256 标识。
+Python/Rust `manifest_bytes` 与共享字节样例保持一致；Coordinator 必须用同一序列化定义存快照。
+读取凭证与客户端提交的身份分开：身份和归属从实际数据库绑定获得，不接受客户端自报 user ID。
+
+当前只实现读取与授权基础。提交创建/幂等/outbox、Coordinator 正式 claim、报告及用户成绩
+查询仍在后续模块阶段，不能把测试数据库 fixture 的租约当作正式领取或实际执行证据。

backend/src/kf_oj/modules/submissions/public.py

@@ -0,0 +1,10 @@
+class SubmissionUnavailable(ValueError):
+    pass
+
+
+class WorkerAccessDenied(ValueError):
+    """Never reveal whether the evaluation, attempt or artifact exists."""
+
+
+class WorkerReadBusy(ValueError):
+    pass

backend/src/kf_oj/modules/submissions/queries.py

@@ -0,0 +1,39 @@
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from kf_oj.contracts.judging import MAX_SOURCE_BYTES, ArtifactReference
+from kf_oj.infra.artifacts.models import Artifact
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.infra.artifacts.store import ArtifactError
+from kf_oj.modules.identity.public import Principal
+from kf_oj.modules.submissions.models import Submission
+from kf_oj.modules.submissions.public import SubmissionUnavailable
+
+
+class SubmissionQueries:
+    def __init__(self, sessions: async_sessionmaker, artifacts: ArtifactRepository):
+        self.sessions, self.artifacts = sessions, artifacts
+
+    async def owned_source(self, principal: Principal, submission_id: UUID) -> bytes:
+        async with self.sessions() as session:
+            submission = await session.scalar(
+                select(Submission).where(
+                    Submission.id == submission_id, Submission.account_id == principal.account_id
+                )
+            )
+            if submission is None:
+                raise SubmissionUnavailable("submission unavailable")
+            metadata = await session.get(Artifact, submission.source_digest)
+            if (
+                metadata is None
+                or metadata.state != "ready"
+                or not 1 <= metadata.bytes <= MAX_SOURCE_BYTES
+            ):
+                raise SubmissionUnavailable("submission source unavailable")
+            reference = ArtifactReference(digest=metadata.digest, bytes=metadata.bytes)
+            try:
+                return await self.artifacts.read(session, reference)
+            except ArtifactError:
+                raise SubmissionUnavailable("submission source unavailable") from None

backend/src/kf_oj/modules/submissions/repository.py

@@ -0,0 +1,27 @@
+from sqlalchemy import func, select
+
+from kf_oj.contracts.judging import LeaseBinding
+from kf_oj.modules.submissions.models import Evaluation, JudgeAttempt, Submission
+
+
+async def leased_submission(session, binding: LeaseBinding, token_hash: str):
+    return (
+        await session.execute(
+            select(Evaluation, JudgeAttempt, Submission)
+            .join(JudgeAttempt, JudgeAttempt.id == Evaluation.current_attempt_id)
+            .join(Submission, Submission.id == Evaluation.submission_id)
+            .where(
+                Evaluation.id == binding.evaluation_id,
+                Evaluation.state.in_(("claimed", "compiling", "running")),
+                Evaluation.cancel_requested_at.is_(None),
+                Evaluation.fencing_version == binding.fencing_version,
+                JudgeAttempt.id == binding.attempt_id,
+                JudgeAttempt.evaluation_id == Evaluation.id,
+                JudgeAttempt.worker_id == binding.worker_id,
+                JudgeAttempt.fencing_version == binding.fencing_version,
+                JudgeAttempt.lease_token_hash == token_hash,
+                JudgeAttempt.expires_at > func.clock_timestamp(),
+                JudgeAttempt.revoked_at.is_(None),
+            )
+        )
+    ).one_or_none()

backend/src/kf_oj/modules/submissions/worker_access.py

@@ -0,0 +1,111 @@
+import asyncio
+import hashlib
+import hmac
+import json
+import re
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from kf_oj.contracts.judging import (
+    JobManifest,
+    LeaseBinding,
+    decode_wire,
+    manifest_bytes,
+)
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.infra.artifacts.store import ArtifactError
+from kf_oj.modules.problems.public import ProblemPort, ProblemUnavailable
+from kf_oj.modules.submissions import repository
+from kf_oj.modules.submissions.public import WorkerAccessDenied, WorkerReadBusy
+
+
+class WorkerAccess:
+    def __init__(
+        self, sessions: async_sessionmaker, problems: ProblemPort, artifacts: ArtifactRepository
+    ):
+        self.sessions, self.problems, self.artifacts = sessions, problems, artifacts
+        self._reads = asyncio.Semaphore(2)
+
+    @asynccontextmanager
+    async def _read_slot(self):
+        if self._reads.locked():
+            raise WorkerReadBusy("worker read capacity exhausted")
+        async with self._reads:
+            yield
+
+    @staticmethod
+    def _token_hash(token: str) -> str:
+        if not re.fullmatch(r"[0-9a-f]{64}", token):
+            raise WorkerAccessDenied("worker grant rejected")
+        return hashlib.sha256(token.encode("ascii")).hexdigest()
+
+    async def _manifest(self, session, binding: LeaseBinding, token: str) -> JobManifest:
+        row = await repository.leased_submission(session, binding, self._token_hash(token))
+        if row is None:
+            raise WorkerAccessDenied("worker grant rejected")
+        evaluation, attempt, submission = row
+        # Use a sanitized decoder at this trust boundary. JSONB snapshots are immutable for
+        # runtime roles; use canonical bytes for both retrieval and digest verification.
+
+        try:
+            manifest = decode_wire(JobManifest, json.dumps(attempt.manifest).encode())
+            if (
+                manifest.evaluation_id != evaluation.id
+                or manifest.submission_id != submission.id
+                or manifest.problem_version_id != evaluation.problem_version_id
+                or manifest.problem_version_id != submission.problem_version_id
+                or manifest.language.value != submission.language
+                or manifest.source.digest != submission.source_digest
+                or not hmac.compare_digest(
+                    hashlib.sha256(manifest_bytes(manifest)).hexdigest(), attempt.manifest_digest
+                )
+            ):
+                raise WorkerAccessDenied("worker snapshot rejected")
+            version = await self.problems.for_submission(session, manifest.problem_version_id)
+            await self.artifacts.check_metadata(session, manifest.source)
+            if (
+                version.problem_id != submission.problem_id
+                or manifest.language not in version.languages
+                or manifest.toolchain != version.toolchain
+                or manifest.cases != version.cases
+                or manifest.compile_limits != version.compile_limits
+                or manifest.run_limits != version.run_limits
+                or manifest.job_wall_time_us != version.job_wall_time_us
+            ):
+                raise WorkerAccessDenied("worker snapshot rejected")
+            return manifest
+        except (ValueError, ProblemUnavailable):
+            raise WorkerAccessDenied("worker snapshot rejected") from None
+
+    async def manifest(self, binding: LeaseBinding, token: str) -> bytes:
+        async with self._read_slot(), self.sessions() as session:
+            manifest = await self._manifest(session, binding, token)
+            payload = manifest_bytes(manifest)
+            # READ COMMITTED: recheck after all preparation and just before releasing bytes.
+            if (
+                await repository.leased_submission(session, binding, self._token_hash(token))
+                is None
+            ):
+                raise WorkerAccessDenied("worker grant rejected")
+            return payload
+
+    async def artifact(self, binding: LeaseBinding, token: str, digest: str) -> bytes:
+        async with self._read_slot(), self.sessions() as session:
+            manifest = await self._manifest(session, binding, token)
+            references = (manifest.source,) + tuple(
+                reference for case in manifest.cases for reference in (case.input, case.expected)
+            )
+            reference = next((ref for ref in references if ref.digest == digest), None)
+            if reference is None:
+                raise WorkerAccessDenied("artifact grant rejected")
+            try:
+                payload = await self.artifacts.read(session, reference)
+            except ArtifactError:
+                raise WorkerAccessDenied("artifact unavailable") from None
+            if (
+                await repository.leased_submission(session, binding, self._token_hash(token))
+                is None
+            ):
+                raise WorkerAccessDenied("worker grant rejected")
+            return payload