feat(problems): publish immutable candidates with audited maintenance

backend/migrations/env.py

@@ -11,7 +11,7 @@ from kf_oj.infra.db.session import build_engine
 from kf_oj.infra.messaging import models as messaging_models  # noqa: F401
 from kf_oj.modules.identity import models as identity_models  # noqa: F401
 from kf_oj.modules.operations import models as operations_models  # noqa: F401
-from kf_oj.modules.problems import import_models  # noqa: F401
+from kf_oj.modules.problems import import_models, maintenance_models  # noqa: F401
 from kf_oj.modules.problems import models as problem_models  # noqa: F401
 from kf_oj.modules.submissions import models as submission_models  # noqa: F401
 

backend/migrations/versions/0007_problem_imports.py

@@ -1,5 +1,6 @@
 """Durable imports/candidates, append-only events and accepted-authority worker port."""
 
+import sqlalchemy as sa
 from alembic import op
 
 from kf_oj.modules.operations.models import MaintenanceEvent
@@ -50,9 +51,31 @@ def upgrade():
         ProblemImportOperation,
         ProblemCandidate,
         CandidateCase,
-        MaintenanceEvent,
     ):
         model.__table__.create(connection)
+    # Freeze the historical shape. A later operations model must not cause a
+    # fresh database to skip ahead and conflict with a subsequent ALTER migration.
+    op.create_table(
+        "problem_maintenance_events",
+        sa.Column("id", sa.Uuid(), primary_key=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()
+        ),
+        sa.Column("actor_id", sa.Uuid(), sa.ForeignKey("accounts.id"), nullable=False),
+        sa.Column("authorization_version", sa.BigInteger(), nullable=False),
+        sa.Column("operation_id", sa.Uuid(), nullable=False),
+        sa.Column("action", sa.String(24), nullable=False),
+        sa.Column("target_id", sa.Uuid()),
+        sa.CheckConstraint("authorization_version > 0", name="authority"),
+        sa.CheckConstraint(
+            "action IN ('upload_created','upload_received','import_ready','import_failed',"
+            "'import_discarded','import_expired','published','disabled','restored')",
+            name="action",
+        ),
+    )
+    op.create_index(
+        "ix_problem_maintenance_events_page", "problem_maintenance_events", ["created_at", "id"]
+    )
     op.create_foreign_key(
         "fk_problem_import_candidate",
         "problem_import_operations",

backend/migrations/versions/0008_problem_maintenance.py

@@ -0,0 +1,117 @@
+"""Immutable maintenance receipts, explicit operator audit and read-only history."""
+
+import sqlalchemy as sa
+from alembic import op
+
+from kf_oj.modules.problems.maintenance_models import ProblemMaintenanceOperation
+
+revision = "0008_problem_maintenance"
+down_revision = "0007_problem_imports"
+branch_labels = depends_on = None
+OWNER = "kf_oj_identity_authority_owner"
+
+
+def upgrade():
+    ProblemMaintenanceOperation.__table__.create(op.get_bind())
+    op.execute("GRANT SELECT,INSERT ON problem_maintenance_operations TO kf_oj_platform")
+    op.add_column(
+        "problem_maintenance_events",
+        sa.Column("actor_kind", sa.String(8), nullable=False, server_default=sa.text("'account'")),
+    )
+    op.add_column("problem_maintenance_events", sa.Column("operator_name", sa.String(63)))
+    op.alter_column("problem_maintenance_events", "actor_id", nullable=True)
+    op.drop_constraint(op.f("ck_problem_maintenance_events_action"), "problem_maintenance_events")
+    op.create_check_constraint(
+        "action",
+        "problem_maintenance_events",
+        "action IN ('upload_created','upload_received','import_ready','import_failed',"
+        "'import_discarded','import_expired','published','disabled','restored',"
+        "'authority_bootstrap','authority_grant','authority_revoke','authority_enable','authority_disable')",
+    )
+    op.create_check_constraint(
+        "actor",
+        "problem_maintenance_events",
+        "(actor_kind='account' AND actor_id IS NOT NULL AND operator_name IS NULL) OR "
+        "(actor_kind='operator' AND actor_id IS NULL AND operator_name IS NOT NULL)",
+    )
+    op.execute("""CREATE FUNCTION public.kf_append_only_maintenance() RETURNS trigger
+        LANGUAGE plpgsql SET search_path=pg_catalog,pg_temp AS $$ BEGIN
+        RAISE EXCEPTION 'append-only maintenance history' USING ERRCODE='23514'; END $$""")
+    for table in ("problem_maintenance_operations", "problem_maintenance_events"):
+        op.execute(f"""CREATE TRIGGER append_only_maintenance BEFORE UPDATE OR DELETE ON {table}
+                       FOR EACH ROW EXECUTE FUNCTION public.kf_append_only_maintenance()""")
+    # Preserve the already-verified authority implementation; its only entry
+    # point becomes a fixed audited wrapper owned by the same narrow NOLOGIN role.
+    op.execute(
+        "ALTER FUNCTION public.kf_manage_account_authority(text,text) "
+        "RENAME TO kf_manage_account_authority_unlogged"
+    )
+    op.execute(
+        "REVOKE ALL ON FUNCTION public.kf_manage_account_authority_unlogged(text,text) "
+        "FROM PUBLIC,kf_oj_identity_operator"
+    )
+    op.execute(f"GRANT INSERT ON problem_maintenance_events TO {OWNER}")
+    op.execute("""
+        CREATE FUNCTION public.kf_manage_account_authority(p_username text,p_operation text)
+        RETURNS TABLE(account_id uuid,username text,enabled boolean,is_administrator boolean,
+            authorization_version bigint,changed boolean)
+        LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,pg_temp AS $$
+        DECLARE result record; BEGIN
+            SELECT * INTO STRICT result FROM
+                public.kf_manage_account_authority_unlogged(p_username,p_operation);
+            IF result.changed THEN
+                INSERT INTO public.problem_maintenance_events
+                    (id,created_at,actor_id,actor_kind,operator_name,authorization_version,
+                     operation_id,action,target_id)
+                VALUES (gen_random_uuid(),clock_timestamp(),NULL,'operator',session_user,
+                        result.authorization_version,gen_random_uuid(),'authority_' || p_operation,
+                        result.account_id);
+            END IF;
+            RETURN QUERY SELECT result.account_id,result.username,result.enabled,
+                result.is_administrator,result.authorization_version,result.changed;
+        END $$""")
+    op.execute(f"GRANT CREATE ON SCHEMA public TO {OWNER}")
+    op.execute(f"ALTER FUNCTION public.kf_manage_account_authority(text,text) OWNER TO {OWNER}")
+    op.execute(f"REVOKE CREATE ON SCHEMA public FROM {OWNER}")
+    op.execute("REVOKE ALL ON FUNCTION public.kf_manage_account_authority(text,text) FROM PUBLIC")
+    op.execute(
+        "GRANT EXECUTE ON FUNCTION public.kf_manage_account_authority(text,text) "
+        "TO kf_oj_identity_operator"
+    )
+
+
+def downgrade():
+    if (
+        op.get_bind()
+        .exec_driver_sql(
+            "SELECT EXISTS(SELECT 1 FROM problem_maintenance_operations) OR "
+            "EXISTS(SELECT 1 FROM problem_maintenance_events WHERE actor_kind='operator')"
+        )
+        .scalar()
+    ):
+        raise RuntimeError("cannot discard committed maintenance history")
+    op.execute("DROP FUNCTION public.kf_manage_account_authority(text,text)")
+    op.execute(
+        "ALTER FUNCTION public.kf_manage_account_authority_unlogged(text,text) "
+        "RENAME TO kf_manage_account_authority"
+    )
+    op.execute(
+        "GRANT EXECUTE ON FUNCTION public.kf_manage_account_authority(text,text) "
+        "TO kf_oj_identity_operator"
+    )
+    op.execute(f"REVOKE INSERT ON problem_maintenance_events FROM {OWNER}")
+    for table in ("problem_maintenance_operations", "problem_maintenance_events"):
+        op.execute(f"DROP TRIGGER append_only_maintenance ON {table}")
+    op.execute("DROP FUNCTION public.kf_append_only_maintenance()")
+    ProblemMaintenanceOperation.__table__.drop(op.get_bind())
+    op.drop_constraint(op.f("ck_problem_maintenance_events_actor"), "problem_maintenance_events")
+    op.drop_constraint(op.f("ck_problem_maintenance_events_action"), "problem_maintenance_events")
+    op.create_check_constraint(
+        "action",
+        "problem_maintenance_events",
+        "action IN ('upload_created','upload_received','import_ready','import_failed',"
+        "'import_discarded','import_expired','published','disabled','restored')",
+    )
+    op.alter_column("problem_maintenance_events", "actor_id", nullable=False)
+    op.drop_column("problem_maintenance_events", "operator_name")
+    op.drop_column("problem_maintenance_events", "actor_kind")

backend/src/kf_oj/api/app.py

@@ -8,6 +8,7 @@ from starlette.responses import JSONResponse
 from kf_oj import __version__
 from kf_oj.api.body_limits import BodyLimitMiddleware
 from kf_oj.api.v1.admin_imports import router as import_router
+from kf_oj.api.v1.admin_problems import router as maintenance_router
 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
@@ -21,8 +22,10 @@ from kf_oj.infra.artifacts.store import ArtifactStore
 from kf_oj.infra.coordinator import CoordinatorControl
 from kf_oj.infra.db.session import build_engine, build_session_factory
 from kf_oj.modules.identity.service import IdentityService
+from kf_oj.modules.operations.queries import MaintenanceAuditQueries
 from kf_oj.modules.operations.service import MaintenanceAudit
 from kf_oj.modules.problems.imports import ProblemImports
+from kf_oj.modules.problems.maintenance import ProblemMaintenance
 from kf_oj.modules.problems.service import ProblemService
 from kf_oj.modules.submissions.performance import SubmissionPerformanceQueries
 from kf_oj.modules.submissions.queries import SubmissionQueries
@@ -69,6 +72,16 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                             MaintenanceAudit(),
                             settings.judge_admission,
                         )
+                app.state.problem_maintenance = ProblemMaintenance(
+                    app.state.sessions,
+                    app.state.identity,
+                    MaintenanceAudit(),
+                    artifacts,
+                    settings.judge_admission,
+                )
+                app.state.maintenance_audit = MaintenanceAuditQueries(
+                    app.state.sessions, app.state.identity
+                )
                 app.state.problems = ProblemService(
                     app.state.sessions, artifacts, settings.judge_admission
                 )
@@ -98,6 +111,8 @@ def create_app(settings: Settings | None = None) -> FastAPI:
             app.state.submission_queries = None
             app.state.submission_performance = None
             app.state.problem_imports = None
+            app.state.problem_maintenance = None
+            app.state.maintenance_audit = None
             if engine is not None:
                 await engine.dispose()
             if store is not None:
@@ -137,4 +152,5 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app.include_router(worker_router, prefix="/api/v1")
     app.include_router(submission_router, prefix="/api/v1")
     app.include_router(import_router, prefix="/api/v1")
+    app.include_router(maintenance_router, prefix="/api/v1")
     return app

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

@@ -0,0 +1,150 @@
+"""Live administrator projections and actor-bound immutable mutations."""
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from sqlalchemy.exc import DBAPIError
+
+from kf_oj.api.admin_dependencies import maintenance_reader, maintenance_writer
+from kf_oj.contracts.problem_maintenance import (
+    AuditAction,
+    ChangeProblemAvailability,
+    MaintainedProblem,
+    MaintainedProblemPage,
+    MaintainedVersionPage,
+    MaintenanceAuditPage,
+    MaintenanceReceipt,
+    PublishCandidate,
+)
+from kf_oj.infra.artifacts.store import ArtifactError
+from kf_oj.modules.identity.public import MaintenanceDenied, Principal
+from kf_oj.modules.problems.imports import ImportConflict, ImportMissing, ImportUnavailable
+from kf_oj.modules.problems.public import ProblemUnavailable
+
+router = APIRouter(prefix="/admin", tags=["problem-maintenance"])
+
+
+def service(request, name="problem_maintenance"):
+    result = getattr(request.app.state, name, None)
+    if result is None:
+        raise HTTPException(503, "problem maintenance unavailable")
+    return result
+
+
+async def operation(awaitable):
+    try:
+        return await awaitable
+    except MaintenanceDenied:
+        raise HTTPException(403, "problem maintenance required") from None
+    except ImportConflict:
+        raise HTTPException(409, "maintenance precondition conflict") from None
+    except ImportMissing:
+        raise HTTPException(404, "maintenance object unavailable") from None
+    except (ImportUnavailable, ArtifactError, ProblemUnavailable, OSError, DBAPIError):
+        raise HTTPException(503, "problem maintenance unavailable") from None
+    except ValueError:
+        raise HTTPException(422, "invalid maintenance request") from None
+
+
+@router.get("/problems", response_model=MaintainedProblemPage)
+async def page(
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_reader)],
+    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,
+    enabled: bool | None = None,
+):
+    return await operation(
+        service(request).page(principal, after=after, limit=limit, enabled=enabled)
+    )
+
+
+@router.get("/problem-slugs/{slug}", response_model=MaintainedProblem)
+async def by_slug(
+    slug: str, request: Request, principal: Annotated[Principal, Depends(maintenance_reader)]
+):
+    from pydantic import TypeAdapter
+
+    from kf_oj.contracts.judging import Label
+
+    try:
+        TypeAdapter(Label).validate_python(slug)
+    except ValueError:
+        raise HTTPException(422, "invalid problem slug") from None
+    return await operation(service(request).by_slug(principal, slug))
+
+
+@router.get("/problems/{problem_id}", response_model=MaintainedProblem)
+async def detail(
+    problem_id: UUID, request: Request, principal: Annotated[Principal, Depends(maintenance_reader)]
+):
+    return await operation(service(request).detail(principal, problem_id))
+
+
+@router.get("/problems/{problem_id}/versions", response_model=MaintainedVersionPage)
+async def versions(
+    problem_id: UUID,
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_reader)],
+    before: Annotated[int | None, Query(ge=1)] = None,
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+):
+    return await operation(
+        service(request).versions(principal, problem_id, before=before, limit=limit)
+    )
+
+
+@router.get("/problem-maintenance/{operation_id}", response_model=MaintenanceReceipt)
+async def receipt(
+    operation_id: UUID,
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_reader)],
+):
+    return await operation(service(request).receipt(principal, operation_id))
+
+
+@router.post("/problem-publications", response_model=MaintenanceReceipt, status_code=201)
+async def publish(
+    body: PublishCandidate,
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_writer)],
+):
+    return await operation(service(request).publish(principal, body))
+
+
+@router.post("/problems/{problem_id}/disable", response_model=MaintenanceReceipt)
+async def disable(
+    problem_id: UUID,
+    body: ChangeProblemAvailability,
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_writer)],
+):
+    return await operation(
+        service(request).availability(principal, problem_id, body, enabled=False)
+    )
+
+
+@router.post("/problems/{problem_id}/restore", response_model=MaintenanceReceipt)
+async def restore(
+    problem_id: UUID,
+    body: ChangeProblemAvailability,
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_writer)],
+):
+    return await operation(service(request).availability(principal, problem_id, body, enabled=True))
+
+
+@router.get("/maintenance-audit", response_model=MaintenanceAuditPage)
+async def audit(
+    request: Request,
+    principal: Annotated[Principal, Depends(maintenance_reader)],
+    before: Annotated[str | None, Query(max_length=100)] = None,
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+    action: AuditAction | None = None,
+):
+    return await operation(
+        service(request, "maintenance_audit").page(
+            principal, before=before, limit=limit, action=action
+        )
+    )

backend/src/kf_oj/contracts/export.py

@@ -13,6 +13,12 @@ from kf_oj.contracts.admin_problems import (
 from kf_oj.contracts.judge import JudgeTaskNotification
 from kf_oj.contracts.judging import WIRE_MODELS
 from kf_oj.contracts.judging_v2 import WIRE_MODELS as WIRE_MODELS_V2
+from kf_oj.contracts.problem_maintenance import (
+    ChangeProblemAvailability,
+    MaintenanceAuditPage,
+    MaintenanceReceipt,
+    PublishCandidate,
+)
 from kf_oj.contracts.submission_performance import SubmissionPerformance
 from kf_oj.core.settings import Settings
 
@@ -49,6 +55,10 @@ def main() -> None:
     )
     outputs[root / "contracts/schemas/submission-performance.v1.schema.json"] = performance
     for name, model in {
+        "problem-publish": PublishCandidate,
+        "problem-availability": ChangeProblemAvailability,
+        "problem-maintenance": MaintenanceReceipt,
+        "problem-maintenance-audit": MaintenanceAuditPage,
         "problem-import-create": CreateProblemImport,
         "problem-import": ProblemImportDetail,
         "problem-import-capabilities": ProblemImportCapabilities,

backend/src/kf_oj/contracts/problem_maintenance.py

@@ -0,0 +1,103 @@
+"""Bounded administrator projections and immutable operation receipts, separate from judging."""
+
+from datetime import datetime
+from typing import Annotated, Literal
+
+from pydantic import Field, StrictBool
+
+from kf_oj.contracts.admin_problems import CandidatePreview
+from kf_oj.contracts.judging import Identifier, Label, Version, WireModel
+from kf_oj.contracts.problems import ProblemSummary
+
+MaintenanceAction = Literal["publish", "disable", "restore"]
+AuditAction = Literal[
+    "upload_created",
+    "upload_received",
+    "import_ready",
+    "import_failed",
+    "import_discarded",
+    "import_expired",
+    "published",
+    "disabled",
+    "restored",
+    "authority_bootstrap",
+    "authority_grant",
+    "authority_revoke",
+    "authority_enable",
+    "authority_disable",
+]
+
+
+class MaintainedProblem(WireModel):
+    id: Identifier
+    slug: Label
+    enabled: StrictBool
+    active_version_id: Identifier | None
+    revision: Annotated[int, Field(strict=True, ge=0)]
+    title: Annotated[str | None, Field(max_length=160)]
+    created_at: datetime
+
+
+class MaintainedProblemPage(WireModel):
+    schema_version: Version = 1
+    items: Annotated[tuple[MaintainedProblem, ...], Field(max_length=100)]
+    next_after: Label | None
+
+
+class MaintainedVersion(ProblemSummary):
+    published_at: datetime
+    case_count: Annotated[int, Field(strict=True, ge=1, le=256)]
+
+
+class MaintainedVersionPage(WireModel):
+    schema_version: Version = 1
+    items: Annotated[tuple[MaintainedVersion, ...], Field(max_length=100)]
+    next_before: Annotated[int | None, Field(strict=True, ge=1)]
+
+
+class PublishCandidate(WireModel):
+    schema_version: Version = 1
+    operation_id: Identifier
+    import_operation_id: Identifier
+    candidate_id: Identifier
+    expected_active_version_id: Identifier | None
+
+
+class ChangeProblemAvailability(WireModel):
+    schema_version: Version = 1
+    operation_id: Identifier
+    expected_active_version_id: Identifier
+    expected_enabled: StrictBool
+
+
+class MaintenanceReceipt(WireModel):
+    schema_version: Version = 1
+    operation_id: Identifier
+    action: MaintenanceAction
+    completed_at: datetime
+    problem: MaintainedProblem
+    published_version_id: Identifier | None = None
+
+
+class MaintenanceAuditEvent(WireModel):
+    id: Identifier
+    created_at: datetime
+    actor_kind: Literal["account", "operator"]
+    actor_id: Identifier | None
+    operator_name: Annotated[str | None, Field(max_length=63)]
+    authorization_version: Annotated[str, Field(pattern=r"^[1-9][0-9]{0,18}$")]
+    operation_id: Identifier
+    action: AuditAction
+    target_id: Identifier | None
+    result: Literal["committed"] = "committed"
+
+
+class MaintenanceAuditPage(WireModel):
+    schema_version: Version = 1
+    items: Annotated[tuple[MaintenanceAuditEvent, ...], Field(max_length=100)]
+    next_before: Annotated[str | None, Field(max_length=100)]
+
+
+# This alias deliberately uses the existing sanitized candidate contract. No new
+# endpoint exposes the private manifest, artifact digest or hidden case names.
+PublicCandidate = CandidatePreview

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

@@ -168,6 +168,51 @@ class ArtifactStore:
         with self.exclusive():
             return self._read(reference)
 
+    def verify(self, reference: ArtifactReference) -> None:
+        """Verify publication inputs with a 64 KiB buffer, without exposing hidden bytes."""
+        ArtifactReference.model_validate(reference)
+        with self.exclusive():
+            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:
+                    before = os.fstat(source.fileno())
+                    self._regular(before)
+                    if before.st_size != reference.bytes:
+                        raise ArtifactError("artifact digest or length mismatch")
+                    digest = hashlib.sha256()
+                    remaining = reference.bytes
+                    while remaining:
+                        data = source.read(min(65536, remaining))
+                        if not data:
+                            raise ArtifactError("artifact digest or length mismatch")
+                        digest.update(data)
+                        remaining -= len(data)
+                    after = os.fstat(source.fileno())
+                    if (
+                        source.read(1)
+                        or digest.hexdigest() != reference.digest
+                        or any(
+                            getattr(before, k) != getattr(after, k)
+                            for k in (
+                                "st_dev",
+                                "st_ino",
+                                "st_nlink",
+                                "st_mode",
+                                "st_uid",
+                                "st_size",
+                                "st_mtime_ns",
+                                "st_ctime_ns",
+                            )
+                        )
+                    ):
+                        raise ArtifactError("artifact input changed")
+            except OSError:
+                raise ArtifactError("artifact unavailable") from None
+
     def put_file(self, source: int, expected: ArtifactReference) -> ArtifactReference:
         """Adopt a validated private FD without buffering a hidden testcase in memory."""
         before = os.fstat(source)

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

@@ -1,5 +1,17 @@
 # operations
 
-负责平台侧节点、任务积压、重试原因和审计查询的管理视图。
-对判题状态的操作必须调用协调器协议；观测模块不成为第二个状态写入入口。
-当前只有目录边界，节点登记、指标聚合和管理员接口待实现。
+维护审计通过 `public.py` 的追加端口写入调用方事务；本模块不成为评测状态的写入入口。
+上传、解析完成/失败、丢弃、发布、下架与恢复记录真实账号、授权版本、操作 UUID、对象与
+数据库时间。发布回执和审计同时提交；审计失败使整个业务事务回滚，不确认半发布。
+
+本机受控身份 CLI 使用已核验的窄权限数据库函数。授权变化由固定包装函数在同一事务
+追加审计，主体类型为 operator，操作者为数据库 `session_user`，对象为实际变更账号；
+不把授予/撤销冒充成该账号自己的操作。旧账号事件保留 account 主体类型。
+
+管理员 `/api/v1/admin/maintenance-audit` 按时间和 UUID 的复合游标递减读取，每页至多
+100项，可选择有限动作类型；只展示已提交事件。导入失败动作仍是失败，committed 表示
+审计事务提交。响应不含口令、会话令牌、工件引用、内部路径或任意诊断正文。
+
+应用数据库角色无审计 UPDATE/DELETE 权限，数据库触发器还阻止历史改写。
+导入 Worker 只有追加权限，不能读取整个维护审计；Coordinator 无审计写入权限。
+节点管理、部署控制及跨账号业务管理仍未开放。

backend/src/kf_oj/modules/operations/models.py

@@ -2,7 +2,7 @@
 
 from uuid import UUID
 
-from sqlalchemy import BigInteger, CheckConstraint, ForeignKey, Index, String
+from sqlalchemy import BigInteger, CheckConstraint, ForeignKey, Index, String, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from kf_oj.infra.db.base import Base
@@ -15,14 +15,24 @@ class MaintenanceEvent(CreatedAt, Base):
         CheckConstraint(
             "action IN ('upload_created','upload_received','import_ready',"
             "'import_failed','import_discarded','import_expired','published',"
-            "'disabled','restored')",
+            "'disabled','restored','authority_bootstrap','authority_grant',"
+            "'authority_revoke','authority_enable','authority_disable')",
             name="action",
         ),
         CheckConstraint("authorization_version > 0", name="authority"),
+        CheckConstraint(
+            "(actor_kind='account' AND actor_id IS NOT NULL AND operator_name IS NULL) "
+            "OR (actor_kind='operator' AND actor_id IS NULL AND operator_name IS NOT NULL)",
+            name="actor",
+        ),
         Index("ix_problem_maintenance_events_page", "created_at", "id"),
     )
     id: Mapped[UUID] = mapped_column(primary_key=True)
-    actor_id: Mapped[UUID] = mapped_column(ForeignKey("accounts.id"))
+    actor_id: Mapped[UUID | None] = mapped_column(ForeignKey("accounts.id"))
+    actor_kind: Mapped[str] = mapped_column(
+        String(8), default="account", server_default=text("'account'")
+    )
+    operator_name: Mapped[str | None] = mapped_column(String(63))
     authorization_version: Mapped[int] = mapped_column(BigInteger)
     operation_id: Mapped[UUID] = mapped_column()
     action: Mapped[str] = mapped_column(String(24))

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

@@ -0,0 +1,62 @@
+"""Only authorized administrators read bounded append-only maintenance projections."""
+
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import select, text, tuple_
+
+from kf_oj.contracts.problem_maintenance import MaintenanceAuditEvent, MaintenanceAuditPage
+from kf_oj.modules.identity.public import ProblemMaintenancePort
+from kf_oj.modules.operations.models import MaintenanceEvent
+
+
+class MaintenanceAuditQueries:
+    def __init__(self, sessions, identity: ProblemMaintenancePort):
+        self.sessions, self.identity = sessions, identity
+
+    async def page(self, principal, *, before=None, limit=20, action=None):
+        if not 1 <= limit <= 100:
+            raise ValueError("invalid page limit")
+        cursor = None
+        if before is not None:
+            stamp, identifier = before.split("|")
+            stamp = datetime.fromisoformat(stamp)
+            if stamp.tzinfo is None:
+                raise ValueError("invalid audit cursor")
+            cursor = stamp, UUID(identifier)
+        async with self.sessions.begin() as session:
+            await session.execute(text("SET LOCAL statement_timeout='5s'"))
+            await session.execute(text("SET LOCAL lock_timeout='2s'"))
+            await self.identity.require_problem_maintenance(session, principal)
+            q = select(MaintenanceEvent)
+            if cursor:
+                q = q.where(tuple_(MaintenanceEvent.created_at, MaintenanceEvent.id) < cursor)
+            if action:
+                q = q.where(MaintenanceEvent.action == action)
+            rows = list(
+                await session.scalars(
+                    q.order_by(
+                        MaintenanceEvent.created_at.desc(), MaintenanceEvent.id.desc()
+                    ).limit(limit + 1)
+                )
+            )
+            visible = rows[:limit]
+            return MaintenanceAuditPage(
+                items=tuple(
+                    MaintenanceAuditEvent(
+                        id=r.id,
+                        created_at=r.created_at,
+                        actor_kind=r.actor_kind,
+                        actor_id=r.actor_id,
+                        operator_name=r.operator_name,
+                        authorization_version=str(r.authorization_version),
+                        operation_id=r.operation_id,
+                        action=r.action,
+                        target_id=r.target_id,
+                    )
+                    for r in visible
+                ),
+                next_before=f"{visible[-1].created_at.isoformat()}|{visible[-1].id}"
+                if len(rows) > limit
+                else None,
+            )

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

@@ -60,3 +60,21 @@ Web格式固定单层ZIP的stored/deflate，清单名为`problem.json`；公开
 同一普通用户的私有目录；数据库采用各自最小角色，导入Worker只通过固定身份公开端口
 验证已受理的维护授权。PrivateNetwork模板要求Unix PostgreSQL socket；实际部署前由运维
 设置外部私有配置、专用登录与目录，不包含默认口令，也不代表本轮已经生产部署。
+
+管理员发布使用 `/api/v1/admin/problem-publications`，绑定独立操作 UUID、本人导入操作、
+候选 UUID 与预期活动版本（首次发布为 null）。`publication.py` 为 CLI 与 Web 共用的
+准入与不可变版本写入路径；Web 在事务内再次校验实时维护权限、候选摘要、工件实际字节
+和可信节点配置。CLI 保留本机受控入口与原有导入格式，不借用网页会话。
+
+成功维护回执只追加保存，UUID 同内容重试返回原快照，不重复创建 revision 或审计。
+UUID 绑定不同内容或活动版本变化返回409；前端应核对最新状态后建立新的操作凭据。
+题目状态写入同样要求预期活动版本及 enabled；下架不修改历史版本或删除任何提交。
+新提交持有逻辑题目的共享锁直到提交事务结束，下架/恢复持有排他锁，二者明确排序。
+
+管理列表按 slug、版本历史按递减 revision 分页，每页至多100条，仅返回公开元数据。
+候选仍通过本人导入操作读取；维护回执仅本人可读。隐藏测试字节、摘要和宿主路径不进入
+这些投影。所有管理读取也经实时维护权限检查，缺失发布配置明确503。
+
+`for_accepted_execution` 仅在 submissions 已核对有效租约、源码归属及不可变尝试快照后
+调用，读取已发布原版本时允许逻辑题目已下架。它没有外部 UUID 直读端点，不放宽未发布
+版本或无有效租约的访问。公开读取与新提交仍检查 enabled。

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

@@ -3,14 +3,13 @@
 import hashlib
 import json
 from dataclasses import dataclass
-from uuid import uuid4, uuid5
+from uuid import uuid5
 
-from kf_oj.contracts import judging_v2 as v2
 from kf_oj.contracts.admin_problems import CandidatePreview, LegacyImportBudget
-from kf_oj.contracts.judging import MAX_SOURCE_BYTES, ArtifactReference, TestCase
+from kf_oj.contracts.judging import ArtifactReference, TestCase
 from kf_oj.contracts.problems import PublicLanguageConfiguration
 from kf_oj.modules.problems.import_format import ImportEnvelope, ProblemImport
-from kf_oj.modules.problems.public import JudgingVersion
+from kf_oj.modules.problems.publication import admit_version
 
 
 @dataclass(frozen=True)
@@ -71,27 +70,5 @@ def validate_candidate(manifest_data, preview_data, references, policy) -> Valid
         )
         for c in manifest.cases
     )
-    projection = JudgingVersion(
-        preview.candidate_id,
-        uuid4(),
-        manifest.languages,
-        manifest.toolchain if legacy else None,
-        manifest.compile_limits if legacy else None,
-        manifest.run_limits if legacy else None,
-        manifest.job_wall_time_us if legacy else None,
-        cases,
-        manifest.schema_version,
-        () if legacy else manifest.language_configurations,
-    )
-    for language in manifest.languages:
-        v2.admit_manifest(
-            policy,
-            projection.manifest(
-                uuid4(),
-                uuid4(),
-                language,
-                ArtifactReference(digest="0" * 64, bytes=MAX_SOURCE_BYTES),
-            ),
-            1,
-        )
+    admit_version(policy, manifest, preview.candidate_id, preview.candidate_id, cases)
     return ValidatedCandidate(manifest.model_dump(mode="json"), preview, refs)

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

@@ -0,0 +1,324 @@
+"""Administrator transactions; no evaluation writes or private data in projections."""
+
+import asyncio
+import hashlib
+import json
+from contextlib import asynccontextmanager
+from uuid import uuid4, uuid5
+
+from sqlalchemy import func, select, text
+
+from kf_oj.contracts.admin_problems import CandidatePreview
+from kf_oj.contracts.judging import ArtifactReference, TestCase
+from kf_oj.contracts.problem_maintenance import (
+    MaintainedProblem,
+    MaintainedProblemPage,
+    MaintainedVersion,
+    MaintainedVersionPage,
+    MaintenanceReceipt,
+)
+from kf_oj.modules.identity.public import ProblemMaintenancePort
+from kf_oj.modules.operations.public import MaintenanceAuditPort
+from kf_oj.modules.problems import repository
+from kf_oj.modules.problems.candidate_validation import validate_candidate
+from kf_oj.modules.problems.import_format import ImportEnvelope
+from kf_oj.modules.problems.import_models import (
+    CandidateCase,
+    ProblemCandidate,
+    ProblemImportOperation,
+)
+from kf_oj.modules.problems.imports import ImportConflict, ImportMissing, ImportUnavailable
+from kf_oj.modules.problems.maintenance_models import ProblemMaintenanceOperation
+from kf_oj.modules.problems.models import Problem, ProblemCase, ProblemVersion
+from kf_oj.modules.problems.publication import admit_version, publish_version
+from kf_oj.modules.problems.service import ProblemService
+
+
+def project_problem(problem, version):
+    return MaintainedProblem(
+        id=problem.id,
+        slug=problem.slug,
+        enabled=problem.enabled,
+        active_version_id=problem.active_version_id,
+        revision=version.revision if version else 0,
+        title=version.title if version else None,
+        created_at=problem.created_at,
+    )
+
+
+class ProblemMaintenance:
+    def __init__(
+        self,
+        sessions,
+        identity: ProblemMaintenancePort,
+        audit: MaintenanceAuditPort,
+        artifacts=None,
+        policy=None,
+    ):
+        self.sessions, self.identity, self.audit = sessions, identity, audit
+        self.artifacts, self.policy = artifacts, policy
+        self._publishing = asyncio.Semaphore(2)
+
+    @asynccontextmanager
+    async def transaction(self, principal):
+        async with self.sessions.begin() as session:
+            await session.execute(text("SET LOCAL statement_timeout='5s'"))
+            await session.execute(text("SET LOCAL lock_timeout='2s'"))
+            await self.identity.require_problem_maintenance(session, principal)
+            yield session
+
+    async def page(self, principal, *, after=None, limit=20, enabled=None):
+        if not 1 <= limit <= 100:
+            raise ValueError("invalid page limit")
+        async with self.transaction(principal) as session:
+            q = select(Problem, ProblemVersion).outerjoin(
+                ProblemVersion, Problem.active_version_id == ProblemVersion.id
+            )
+            if after is not None:
+                q = q.where(Problem.slug > after)
+            if enabled is not None:
+                q = q.where(Problem.enabled == enabled)
+            rows = list(await session.execute(q.order_by(Problem.slug).limit(limit + 1)))
+            return MaintainedProblemPage(
+                items=tuple(project_problem(*r) for r in rows[:limit]),
+                next_after=rows[limit - 1][0].slug if len(rows) > limit else None,
+            )
+
+    async def by_slug(self, principal, slug):
+        async with self.transaction(principal) as session:
+            row = (
+                await session.execute(
+                    select(Problem, ProblemVersion)
+                    .outerjoin(ProblemVersion, Problem.active_version_id == ProblemVersion.id)
+                    .where(Problem.slug == slug)
+                )
+            ).one_or_none()
+            if row is None:
+                raise ImportMissing("problem unavailable")
+            return project_problem(*row)
+
+    async def detail(self, principal, problem_id):
+        async with self.transaction(principal) as session:
+            problem = await session.get(Problem, problem_id)
+            if problem is None:
+                raise ImportMissing("problem unavailable")
+            version = (
+                await session.get(ProblemVersion, problem.active_version_id)
+                if problem.active_version_id
+                else None
+            )
+            return project_problem(problem, version)
+
+    async def versions(self, principal, problem_id, *, before=None, limit=20):
+        if not 1 <= limit <= 100 or (before is not None and before < 1):
+            raise ValueError("invalid page limit")
+        async with self.transaction(principal) as session:
+            problem = await session.get(Problem, problem_id)
+            if problem is None:
+                raise ImportMissing("problem unavailable")
+            count = (
+                select(func.count())
+                .where(ProblemCase.version_id == ProblemVersion.id)
+                .scalar_subquery()
+            )
+            q = select(ProblemVersion, count).where(
+                ProblemVersion.problem_id == problem_id, ProblemVersion.published_at.is_not(None)
+            )
+            if before is not None:
+                q = q.where(ProblemVersion.revision < before)
+            rows = list(
+                await session.execute(q.order_by(ProblemVersion.revision.desc()).limit(limit + 1))
+            )
+            return MaintainedVersionPage(
+                items=tuple(
+                    MaintainedVersion(
+                        **ProblemService._summary(problem, v).model_dump(),
+                        published_at=v.published_at,
+                        case_count=n,
+                    )
+                    for v, n in rows[:limit]
+                ),
+                next_before=rows[limit - 1][0].revision if len(rows) > limit else None,
+            )
+
+    async def receipt(self, principal, operation_id):
+        async with self.transaction(principal) as session:
+            row = await session.get(ProblemMaintenanceOperation, operation_id)
+            if row is None or row.owner_id != principal.account_id:
+                raise ImportMissing("operation unavailable")
+            return MaintenanceReceipt.model_validate(row.receipt)
+
+    async def _replay(self, session, principal, action, body, target):
+        digest = hashlib.sha256(
+            json.dumps(
+                {"action": action, "target": str(target), "request": body.model_dump(mode="json")},
+                sort_keys=True,
+                separators=(",", ":"),
+            ).encode()
+        ).hexdigest()
+        key = int.from_bytes(hashlib.sha256(body.operation_id.bytes).digest()[:4], signed=True)
+        await session.execute(text("SELECT pg_advisory_xact_lock(1262907470,:key)"), {"key": key})
+        row = await session.get(ProblemMaintenanceOperation, body.operation_id)
+        if row is not None:
+            if row.owner_id != principal.account_id:
+                raise ImportMissing("operation unavailable")
+            if row.action != action or row.request_digest != digest:
+                raise ImportConflict("operation UUID already bound")
+            return digest, MaintenanceReceipt.model_validate(row.receipt)
+        return digest, None
+
+    async def _record(
+        self, session, principal, request, action, digest, problem, version, published=None
+    ):
+        receipt = MaintenanceReceipt(
+            operation_id=request.operation_id,
+            action=action,
+            completed_at=await session.scalar(select(func.clock_timestamp())),
+            problem=project_problem(problem, version),
+            published_version_id=published,
+        )
+        session.add(
+            ProblemMaintenanceOperation(
+                id=request.operation_id,
+                owner_id=principal.account_id,
+                authorization_version=principal.authorization_version,
+                action=action,
+                request_digest=digest,
+                receipt=receipt.model_dump(mode="json"),
+            )
+        )
+        await self.audit.record(
+            session,
+            principal.account_id,
+            principal.authorization_version,
+            request.operation_id,
+            {"publish": "published", "disable": "disabled", "restore": "restored"}[action],
+            published if action == "publish" else problem.id,
+        )
+        await session.flush()
+        return receipt
+
+    async def publish(self, principal, request):
+        if self.artifacts is None or self.policy is None or self._publishing.locked():
+            raise ImportUnavailable("publication unavailable")
+        async with self._publishing:
+            async with self.transaction(principal) as session:
+                digest, replay = await self._replay(
+                    session, principal, "publish", request, request.import_operation_id
+                )
+                if replay:
+                    return replay
+                op = await session.get(
+                    ProblemImportOperation, request.import_operation_id, with_for_update=True
+                )
+                if op is None or op.owner_id != principal.account_id:
+                    raise ImportMissing("candidate unavailable")
+                if (
+                    op.state != "ready"
+                    or op.candidate_id != request.candidate_id
+                    or op.authorization_version != principal.authorization_version
+                    or op.expires_at <= await session.scalar(select(func.clock_timestamp()))
+                ):
+                    raise ImportConflict("candidate no longer publishable")
+                candidate = await session.get(ProblemCandidate, op.candidate_id)
+                if candidate is None or candidate.operation_id != op.id:
+                    raise ImportUnavailable("candidate unavailable")
+                manifest = ImportEnvelope.model_validate(candidate.manifest).root
+                preview = CandidatePreview.model_validate(candidate.preview)
+                rows = list(
+                    await session.scalars(
+                        select(CandidateCase)
+                        .where(CandidateCase.candidate_id == candidate.id)
+                        .order_by(CandidateCase.position)
+                    )
+                )
+                if len(rows) != len(manifest.cases) or len(rows) != candidate.case_count:
+                    raise ImportUnavailable("candidate references unavailable")
+                refs = {}
+                for position, (row, c) in enumerate(zip(rows, manifest.cases, strict=True)):
+                    if row.position != position or row.key != c.key:
+                        raise ImportUnavailable("candidate references unavailable")
+                    for name, ref in (
+                        (
+                            c.input_file,
+                            ArtifactReference(digest=row.input_digest, bytes=row.input_bytes),
+                        ),
+                        (
+                            c.expected_file,
+                            ArtifactReference(digest=row.expected_digest, bytes=row.expected_bytes),
+                        ),
+                    ):
+                        if name in refs and refs[name] != ref:
+                            raise ImportUnavailable("candidate references unavailable")
+                        refs[name] = ref
+                validated = validate_candidate(
+                    candidate.manifest, candidate.preview, refs, self.policy
+                )
+                if (
+                    validated.digest != candidate.manifest_digest
+                    or preview.candidate_id != candidate.id
+                ):
+                    raise ImportUnavailable("candidate integrity unavailable")
+                await repository.lock_slug(session, manifest.slug)
+                problem = await repository.lock_problem(session, manifest.slug)
+                actual = problem.active_version_id if problem else None
+                if actual != request.expected_active_version_id or (
+                    problem and not problem.enabled
+                ):
+                    raise ImportConflict("active version or availability changed")
+                if problem is None:
+                    problem = Problem(id=uuid4(), slug=manifest.slug)
+                    session.add(problem)
+                    await session.flush()
+                version_id = uuid4()
+                cases = tuple(
+                    TestCase(
+                        case_id=uuid5(version_id, c.key),
+                        input=refs[c.input_file],
+                        expected=refs[c.expected_file],
+                    )
+                    for c in manifest.cases
+                )
+                admit_version(self.policy, manifest, version_id, problem.id, cases)
+                # Candidate FKs protect all referenced ready metadata from GC. The
+                # private adapter additionally re-hashes disk bytes without buffering them.
+                for ref in sorted(
+                    {r.digest: r for r in refs.values()}.values(), key=lambda r: r.digest
+                ):
+                    await self.artifacts.check_metadata(session, ref)
+                    task = asyncio.create_task(asyncio.to_thread(self.artifacts.store.verify, ref))
+                    try:
+                        await asyncio.shield(task)
+                    except BaseException:
+                        await task
+                        raise
+                version = await publish_version(
+                    session, problem, manifest, preview, version_id, cases
+                )
+                op.state, op.published_version_id = "published", version.id
+                op.updated_at = await session.scalar(select(func.clock_timestamp()))
+                return await self._record(
+                    session, principal, request, "publish", digest, problem, version, version.id
+                )
+
+    async def availability(self, principal, problem_id, request, *, enabled):
+        action = "restore" if enabled else "disable"
+        async with self.transaction(principal) as session:
+            digest, replay = await self._replay(session, principal, action, request, problem_id)
+            if replay:
+                return replay
+            problem = await session.get(Problem, problem_id, with_for_update=True)
+            if problem is None:
+                raise ImportMissing("problem unavailable")
+            if (
+                problem.active_version_id != request.expected_active_version_id
+                or problem.enabled != request.expected_enabled
+                or problem.enabled == enabled
+            ):
+                raise ImportConflict("active version or availability changed")
+            version = await session.get(ProblemVersion, problem.active_version_id)
+            if version is None or version.published_at is None:
+                raise ImportUnavailable("published version unavailable")
+            problem.enabled = enabled
+            await session.flush()
+            return await self._record(session, principal, request, action, digest, problem, version)

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

@@ -0,0 +1,26 @@
+"""Append-only actor-bound receipts; replay never re-applies a published operation."""
+
+from uuid import UUID
+
+from sqlalchemy import BigInteger, CheckConstraint, ForeignKey, String
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from kf_oj.infra.db.base import Base
+from kf_oj.infra.db.columns import CreatedAt
+
+
+class ProblemMaintenanceOperation(CreatedAt, Base):
+    __tablename__ = "problem_maintenance_operations"
+    __table_args__ = (
+        CheckConstraint("action IN ('publish','disable','restore')", name="action"),
+        CheckConstraint("request_digest ~ '^[0-9a-f]{64}$'", name="digest"),
+        CheckConstraint("authorization_version > 0", name="authority"),
+        CheckConstraint("octet_length(receipt::text) <= 4096", name="receipt"),
+    )
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    owner_id: Mapped[UUID] = mapped_column(ForeignKey("accounts.id"))
+    authorization_version: Mapped[int] = mapped_column(BigInteger)
+    action: Mapped[str] = mapped_column(String(16))
+    request_digest: Mapped[str] = mapped_column(String(64))
+    receipt: Mapped[dict] = mapped_column(JSONB)

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

@@ -78,3 +78,7 @@ class JudgingVersion:
 
 class ProblemPort(Protocol):
     async def for_submission(self, session, version_id: UUID) -> JudgingVersion: ...
+
+    async def for_accepted_execution(self, session, version_id: UUID) -> JudgingVersion:
+        """Require an already-validated live execution lease; allow a disabled published version."""
+        ...

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

@@ -0,0 +1,82 @@
+"""One admission and immutable publication path for local and web imports."""
+
+from uuid import uuid4
+
+from sqlalchemy import func, select
+
+from kf_oj.contracts import judging_v2 as v2
+from kf_oj.contracts.judging import MAX_SOURCE_BYTES, ArtifactReference
+from kf_oj.modules.problems import repository
+from kf_oj.modules.problems.import_format import ProblemImport
+from kf_oj.modules.problems.models import ProblemCase, ProblemVersion
+from kf_oj.modules.problems.public import JudgingVersion, ProblemUnavailable
+
+
+def admit_version(policy, manifest, version_id, problem_id, cases):
+    legacy = isinstance(manifest, ProblemImport)
+    projection = JudgingVersion(
+        version_id,
+        problem_id,
+        manifest.languages,
+        manifest.toolchain if legacy else None,
+        manifest.compile_limits if legacy else None,
+        manifest.run_limits if legacy else None,
+        manifest.job_wall_time_us if legacy else None,
+        cases,
+        manifest.schema_version,
+        () if legacy else manifest.language_configurations,
+    )
+    for language in manifest.languages:
+        v2.admit_manifest(
+            policy,
+            projection.manifest(
+                uuid4(),
+                uuid4(),
+                language,
+                ArtifactReference(digest="0" * 64, bytes=MAX_SOURCE_BYTES),
+            ),
+            1,
+        )
+    return projection
+
+
+async def publish_version(session, problem, manifest, preview, version_id, cases):
+    """Caller holds the logical slug/row locks and has made all artifacts durable."""
+    if not problem.enabled:
+        raise ProblemUnavailable("problem disabled")
+    legacy = isinstance(manifest, ProblemImport)
+    version = ProblemVersion(
+        id=version_id,
+        problem_id=problem.id,
+        revision=await repository.next_revision(session, problem.id),
+        title=manifest.title,
+        statement=preview.statement,
+        languages=[lang.value for lang in manifest.languages],
+        judging_schema_version=manifest.schema_version,
+        language_configurations=None
+        if legacy
+        else [c.model_dump(mode="json") for c in manifest.language_configurations],
+        toolchain=manifest.toolchain.model_dump(mode="json") if legacy else None,
+        compile_limits=manifest.compile_limits.model_dump(mode="json") if legacy else None,
+        run_limits=manifest.run_limits.model_dump(mode="json") if legacy else None,
+        job_wall_time_us=manifest.job_wall_time_us if legacy else None,
+        comparison=manifest.comparison,
+        public_samples=[s.model_dump(mode="json") for s in preview.samples],
+    )
+    session.add(version)
+    await session.flush()
+    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.published_at = await session.scalar(select(func.now()))
+    problem.active_version_id = version_id
+    await session.flush()
+    return version

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

@@ -24,7 +24,22 @@ async def next_revision(session: AsyncSession, problem_id: UUID) -> int:
     return (latest or 0) + 1
 
 
-async def public_version(session: AsyncSession, version_id: UUID):
+async def published_version(session: AsyncSession, version_id: UUID, *, admission=False):
+    query = (
+        select(ProblemVersion, Problem)
+        .join(Problem, Problem.id == ProblemVersion.problem_id)
+        .where(ProblemVersion.id == version_id, ProblemVersion.published_at.is_not(None))
+    )
+    if admission:
+        # A new submission commits while this shared problem lock is held. A
+        # concurrent disable either precedes admission or follows that commit.
+        query = query.where(Problem.enabled.is_(True)).with_for_update(read=True, of=Problem)
+    return (await session.execute(query)).one_or_none()
+
+
+async def public_version(session: AsyncSession, version_id: UUID, *, admission=False):
+    if admission:
+        return await published_version(session, version_id, admission=True)
     return (
         await session.execute(
             select(ProblemVersion, Problem)

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

@@ -1,8 +1,9 @@
 import base64
 from pathlib import Path
+from types import SimpleNamespace
 from uuid import UUID, uuid4, uuid5
 
-from sqlalchemy import func, select
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import async_sessionmaker
 
 from kf_oj.contracts import judging_v2 as v2
@@ -28,8 +29,9 @@ from kf_oj.infra.artifacts.repository import ArtifactRepository
 from kf_oj.modules.problems import repository
 from kf_oj.modules.problems.import_format import ImportEnvelope, 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.models import Problem
 from kf_oj.modules.problems.public import JudgingVersion, ProblemUnavailable
+from kf_oj.modules.problems.publication import admit_version, publish_version
 
 
 class ProblemService:
@@ -83,27 +85,7 @@ class ProblemService:
             )
             for case, (data, answer) in zip(manifest.cases, payloads, strict=True)
         )
-        legacy = isinstance(manifest, ProblemImport)
-        projection = JudgingVersion(
-            version_id,
-            uuid4(),
-            manifest.languages,
-            manifest.toolchain if legacy else None,
-            manifest.compile_limits if legacy else None,
-            manifest.run_limits if legacy else None,
-            manifest.job_wall_time_us if legacy else None,
-            cases,
-            manifest.schema_version,
-            () if legacy else manifest.language_configurations,
-        )
-        for language in manifest.languages:
-            check = projection.manifest(
-                uuid4(),
-                uuid4(),
-                language,
-                ArtifactReference(digest="0" * 64, bytes=MAX_SOURCE_BYTES),
-            )
-            v2.admit_manifest(self.policy, check, 1)
+        admit_version(self.policy, manifest, version_id, uuid4(), cases)
         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)
@@ -114,32 +96,6 @@ class ProblemService:
                 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],
-                    judging_schema_version=manifest.schema_version,
-                    language_configurations=(
-                        None
-                        if legacy
-                        else [c.model_dump(mode="json") for c in manifest.language_configurations]
-                    ),
-                    toolchain=manifest.toolchain.model_dump(mode="json") if legacy else None,
-                    compile_limits=manifest.compile_limits.model_dump(mode="json")
-                    if legacy
-                    else None,
-                    run_limits=manifest.run_limits.model_dump(mode="json") if legacy else None,
-                    job_wall_time_us=manifest.job_wall_time_us if legacy else None,
-                    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)
@@ -149,20 +105,14 @@ class ProblemService:
             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
+            await publish_version(
+                session,
+                problem,
+                manifest,
+                SimpleNamespace(statement=statement, samples=samples),
+                version_id,
+                cases,
+            )
         return version_id
 
     @staticmethod
@@ -213,7 +163,16 @@ class ProblemService:
     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)
+        row = await repository.public_version(session, version_id, admission=True)
+        return await self._judging_version(session, row)
+
+    async def for_accepted_execution(self, session, version_id: UUID) -> JudgingVersion:
+        # Only WorkerAccess calls this after validating the current lease and its
+        # immutable submission/attempt binding; this is never a public UUID lookup.
+        row = await repository.published_version(session, version_id)
+        return await self._judging_version(session, row)
+
+    async def _judging_version(self, session, row):
         if row is None:
             raise ProblemUnavailable("problem version unavailable")
         version, problem = row

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

@@ -43,3 +43,8 @@ CPU 为测试点运行 CPU 微秒之和，内存为测试点运行峰值字节
 用量与桶边界以十进制字符串传输；零合法、缺测不补零。严格优于数按原整数的 `>`
 计算，同值独立计数。响应至多24个连续桶，只有最后上界包含端点；浏览器仅格式化单位
 和绘制坐标，比例以服务端严格计数除以样本数，不按人数统计。
+
+下架只阻止新提交与公开题目读取。Worker 的版本读取通过 `ProblemPort.for_accepted_execution`，
+在本模块验证当前 Coordinator 租约/尝试/源码绑定之后才调用；生成并比较原 canonical
+快照，再次核对租约后才释放字节。无有效租约、外来工件、未发布版本和其他提交仍拒绝。
+本人历史和源码读取保留所有权检查，不依赖题目当前 enabled 状态。

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

@@ -62,7 +62,9 @@ class WorkerAccess:
                 )
             ):
                 raise WorkerAccessDenied("worker snapshot rejected")
-            version = await self.problems.for_submission(session, manifest.problem_version_id)
+            version = await self.problems.for_accepted_execution(
+                session, manifest.problem_version_id
+            )
             await self.artifacts.check_metadata(session, manifest.source)
             expected = version.manifest(
                 manifest.evaluation_id, manifest.submission_id, manifest.language, manifest.source

backend/tests/conftest.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import os
+import re
 import subprocess
 import sys
 from pathlib import Path
@@ -40,13 +41,20 @@ def postgres_url():
 
 
 def migrate(url: str, command: str, target: str | None = None) -> None:
-    subprocess.run(
-        [sys.executable, "-m", "alembic", command] + ([target] if target else []),
-        cwd=Path(__file__).resolve().parents[1],
-        env=os.environ | {"KF_OJ_DATABASE_URL": url},
-        check=True,
-        capture_output=True,
-    )
+    try:
+        subprocess.run(
+            [sys.executable, "-m", "alembic", command] + ([target] if target else []),
+            cwd=Path(__file__).resolve().parents[1],
+            env=os.environ | {"KF_OJ_DATABASE_URL": url},
+            check=True,
+            capture_output=True,
+        )
+    except subprocess.CalledProcessError as error:
+        diagnostic = ((error.stdout or b"") + (error.stderr or b"")).decode(errors="replace")[
+            -8192:
+        ]
+        diagnostic = re.sub(r"postgresql(?:\+asyncpg)?://[^\s]+", "[test database URL]", diagnostic)
+        raise AssertionError(f"test migration {command} failed: " + diagnostic) from None
 
 
 async def connect(url: str):

backend/tests/test_database_foundation.py

@@ -13,7 +13,7 @@ def test_migration_roundtrip_and_role_safety(postgres_url):
         c = await connect(postgres_url)
         try:
             tables = set(await c.fetch("SELECT tablename FROM pg_tables WHERE schemaname='public'"))
-            assert len(tables) == 20  # Nineteen domain/control tables and Alembic history.
+            assert len(tables) == 21  # Twenty domain/control tables and Alembic history.
             roles = await c.fetch("SELECT * FROM pg_roles WHERE rolname LIKE 'kf_oj_%'")
             assert {row["rolname"] for row in roles} == {
                 "kf_oj_platform",

backend/tests/test_problem_maintenance_vm.py

@@ -0,0 +1,576 @@
+"""Real maintenance concurrency/SQL/HTTP acceptance, independent capped VM only."""
+
+import asyncio
+import hashlib
+import json
+import os
+import resource
+import subprocess
+import sys
+import time
+from pathlib import Path
+from uuid import UUID, uuid4
+
+import httpx
+import pytest
+from conftest import connect
+from sqlalchemy.engine import make_url
+from test_admin_import_vm import independent_import_vm as independent_import_vm
+from test_admin_import_vm import package
+
+from kf_oj.contracts.judging_v2 import AdmissionRegistry
+from kf_oj.modules.identity.passwords import hash_password
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_actual_publication_receipts_concurrency_audit_and_disable(
+    independent_import_vm, postgres_url, tmp_path
+):
+    groups = independent_import_vm
+    policy = AdmissionRegistry.model_validate_json(
+        (ROOT / "tests/fixtures/contracts-v2/judge-registry.valid.json").read_bytes()
+    )
+    base = {
+        k: v
+        for k, v in os.environ.items()
+        if k
+        in (
+            "PATH",
+            "PYTHONPATH",
+            "LD_LIBRARY_PATH",
+            "LC_ALL",
+            "PYTHONUNBUFFERED",
+            "PYTHONDONTWRITEBYTECODE",
+        )
+    }
+    roles = {"api": "import_api_" + uuid4().hex, "worker": "import_worker_" + uuid4().hex}
+    password = "Synthetic import acceptance only 2026!"
+    users = {name: uuid4() for name in ("web_admin", "web_backup", "web_student")}
+    processes = {}
+    logs = {}
+    artifact = tmp_path / "artifacts"
+    staging = tmp_path / "staging"
+    artifact.mkdir(mode=0o700)
+    staging.mkdir(mode=0o700)
+
+    def enter(group):
+        def apply():
+            (group / "cgroup.procs").write_text(str(os.getpid()))
+            resource.setrlimit(resource.RLIMIT_FSIZE, (67108864, 67108864))
+            resource.setrlimit(resource.RLIMIT_NOFILE, (64, 64))
+            resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
+
+        return apply
+
+    def start(name, *, configured=True, pair=False):
+        group = groups["api" if name == "api" else "import-worker"]
+        env = base | {
+            "KF_OJ_DATABASE_URL": make_url(postgres_url)
+            .set(username=roles[name])
+            .render_as_string(hide_password=False),
+            "KF_OJ_ENVIRONMENT": "test",
+            "KF_OJ_ARTIFACT_ROOT": str(artifact),
+            "KF_OJ_IMPORT_STAGING_ROOT": str(staging),
+            "KF_OJ_JUDGE_ADMISSION": policy.model_dump_json(),
+            "KF_OJ_BROWSER_ORIGINS": '["http://127.0.0.1:58000"]',
+        }
+        if not configured:
+            env.pop("KF_OJ_IMPORT_STAGING_ROOT")
+        if name == "worker":
+            env["KF_OJ_IMPORT_WORKER_CGROUP"] = str(group)
+        command = (
+            [
+                sys.executable,
+                "-m",
+                "uvicorn",
+                "kf_oj.api.app:create_app",
+                "--factory",
+                "--host",
+                "127.0.0.1",
+                "--port",
+                "58000",
+                "--no-proxy-headers",
+                "--limit-concurrency",
+                "4",
+                "--backlog",
+                "16",
+            ]
+            if name == "api"
+            else [sys.executable, "-m", "kf_oj.workers.problem_import"]
+        )
+        if pair:
+            assert name == "worker"
+            # Trusted fixture composition; both coroutines call the actual ordinary Worker.
+            code = """
+import asyncio
+from kf_oj.core.settings import Settings
+from kf_oj.infra.db.session import build_engine,build_session_factory
+from kf_oj.infra.artifacts.store import ArtifactStore
+from kf_oj.infra.artifacts.staging import ImportStaging
+from kf_oj.infra.artifacts.repository import ArtifactRepository
+from kf_oj.modules.identity.service import ReservedMaintenance
+from kf_oj.modules.operations.service import MaintenanceAudit
+from kf_oj.modules.problems.import_worker import ImportWorker
+async def run():
+ s=Settings();e=build_engine(s);store=ArtifactStore(s.artifact_root)
+ staging=ImportStaging(s.import_staging_root)
+ try:
+  w=ImportWorker(build_session_factory(e),staging,ArtifactRepository(store),
+   ReservedMaintenance(),MaintenanceAudit(),s.judge_admission,s.import_worker_cgroup)
+  results=await asyncio.gather(w.once(),w.once())
+  assert sorted(results)==[False,True],results
+  print('ACTUAL_CONCURRENT_IMPORT_CLAIM_ACCEPTED',flush=True)
+ finally:
+  staging.close();store.close();await e.dispose()
+asyncio.run(run())
+"""
+            command = [sys.executable, "-c", code]
+        log = (tmp_path / f"{name}.log").open("a")
+        logs[name] = log
+        processes[name] = subprocess.Popen(
+            command, env=env, stdout=log, stderr=log, preexec_fn=enter(group)
+        )
+
+    def stop(name):
+        p = processes.pop(name, None)
+        if p and p.poll() is None:
+            p.terminate()
+            try:
+                p.wait(timeout=20)
+            except subprocess.TimeoutExpired:
+                p.kill()
+                p.wait(timeout=10)
+        if name in logs:
+            logs.pop(name).close()
+
+    async def run():
+        c = await connect(postgres_url)
+        try:
+            for name, group in (("api", "kf_oj_platform"), ("worker", "kf_oj_problem_import")):
+                await c.execute(
+                    f"CREATE ROLE {roles[name]} LOGIN NOSUPERUSER NOCREATEDB "
+                    "NOCREATEROLE NOREPLICATION NOBYPASSRLS"
+                )
+                await c.execute(f"GRANT {group} TO {roles[name]}")
+            for name, identifier in users.items():
+                await c.execute(
+                    "INSERT INTO accounts(id,username,password_hash) VALUES($1,$2,$3)",
+                    identifier,
+                    name,
+                    hash_password(name, password),
+                )
+            await c.execute("SELECT * FROM kf_manage_account_authority('web_admin','bootstrap')")
+            await c.execute("SELECT * FROM kf_manage_account_authority('web_backup','grant')")
+            start("api")
+            start("worker")
+            async with httpx.AsyncClient(base_url="http://127.0.0.1:58000", timeout=45) as client:
+                deadline = time.monotonic() + 180
+                while time.monotonic() < deadline:
+                    assert processes["api"].poll() is None, (tmp_path / "api.log").read_text()[
+                        -2000:
+                    ]
+                    assert processes["worker"].poll() is None, (
+                        tmp_path / "worker.log"
+                    ).read_text()[-2000:]
+                    try:
+                        if (
+                            await client.get("/api/v1/health/live")
+                        ).status_code == 200 and await c.fetchval(
+                            "SELECT worker_ready_until>clock_timestamp() "
+                            "FROM problem_import_pool WHERE id=1"
+                        ):
+                            break
+                    except httpx.ConnectError:
+                        pass
+                    await asyncio.sleep(0.1)
+                else:
+                    raise AssertionError("real service startup deadline")
+                admin = "/api/v1/admin"
+                imports = admin + "/problem-imports"
+                assert (await client.get(admin + "/problems")).status_code == 401
+
+                async def login(name):
+                    client.cookies.clear()
+                    r = await client.post(
+                        "/api/v1/auth/login",
+                        json={"username": name, "password": password},
+                        headers={"Origin": "http://127.0.0.1:58000"},
+                    )
+                    assert r.status_code == 200, r.text
+                    return {
+                        "Origin": "http://127.0.0.1:58000",
+                        "X-CSRF-Token": client.cookies["kf_csrf"],
+                    }
+
+                headers = await login("web_student")
+                for path in (
+                    "/problems",
+                    "/maintenance-audit",
+                    "/problem-maintenance/" + str(uuid4()),
+                ):
+                    assert (await client.get(admin + path)).status_code == 403
+                headers = await login("web_admin")
+
+                async def ready(slug, *, multi=False):
+                    blob, payload = package(
+                        policy, multi=multi, mutation=lambda m: m.update(slug=slug)
+                    )
+                    key = str(uuid4())
+                    body = {
+                        "operation_id": key,
+                        "archive_digest": hashlib.sha256(blob).hexdigest(),
+                        "archive_bytes": len(blob),
+                    }
+                    end = time.monotonic() + 20
+                    while True:
+                        r = await client.post(imports, json=body, headers=headers)
+                        if r.status_code != 503 or time.monotonic() >= end:
+                            break
+                        await asyncio.sleep(0.2)
+                    assert r.status_code == 201, r.text
+                    end = time.monotonic() + 30
+                    while True:
+                        r = await client.put(
+                            imports + "/" + key + "/archive",
+                            content=blob,
+                            headers=headers | {"Content-Type": "application/zip"},
+                        )
+                        if r.status_code != 503 or time.monotonic() >= end:
+                            break
+                        await asyncio.sleep(0.2)
+                    assert r.status_code == 200, r.text
+                    end = time.monotonic() + 120
+                    while time.monotonic() < end:
+                        r = await client.get(imports + "/" + key)
+                        assert r.status_code == 200, r.text
+                        if r.json()["state"] == "ready":
+                            return r.json(), payload
+                        assert r.json()["state"] != "failed", r.text
+                        await asyncio.sleep(0.2)
+                    raise AssertionError("candidate deadline")
+
+                async def publish(candidate, expected=None, key=None):
+                    body = {
+                        "operation_id": str(key or uuid4()),
+                        "import_operation_id": candidate["operation_id"],
+                        "candidate_id": candidate["candidate_id"],
+                        "expected_active_version_id": expected,
+                    }
+                    r = await client.post(
+                        admin + "/problem-publications", json=body, headers=headers
+                    )
+                    return r, body
+
+                first, payload = await ready("managed-synthetic")
+                for invalid in ({}, headers | {"Origin": "https://invalid.example"}):
+                    r = await client.post(
+                        admin + "/problem-publications",
+                        json={
+                            "operation_id": str(uuid4()),
+                            "import_operation_id": first["operation_id"],
+                            "candidate_id": first["candidate_id"],
+                            "expected_active_version_id": None,
+                        },
+                        headers=invalid,
+                    )
+                    assert r.status_code == 403
+                r, body = await publish(first)
+                assert r.status_code == 201, r.text
+                receipt = r.json()
+                pid = receipt["problem"]["id"]
+                vid = receipt["published_version_id"]
+                assert receipt["problem"]["revision"] == 1
+                assert (
+                    await client.post(admin + "/problem-publications", json=body, headers=headers)
+                ).json() == receipt
+                assert (
+                    await client.get(admin + "/problem-maintenance/" + body["operation_id"])
+                ).json() == receipt
+                assert (
+                    await client.post(
+                        admin + "/problem-publications",
+                        json=body | {"expected_active_version_id": vid},
+                        headers=headers,
+                    )
+                ).status_code == 409
+                assert (await client.get(imports + "/" + first["operation_id"])).json()[
+                    "state"
+                ] == "published"
+                public = await client.get("/api/v1/problems/versions/" + vid)
+                assert public.status_code == 200, public.text
+                assert public.json()["statement"] == "Public statement only"
+                assert "digest" not in public.text and "hidden" not in public.text
+                old = await c.fetchrow("SELECT * FROM problem_versions WHERE id=$1", UUID(vid))
+                second, _ = await ready("managed-synthetic")
+                third, _ = await ready("managed-synthetic")
+                b1 = {
+                    "operation_id": str(uuid4()),
+                    "import_operation_id": second["operation_id"],
+                    "candidate_id": second["candidate_id"],
+                    "expected_active_version_id": vid,
+                }
+                b2 = b1 | {
+                    "operation_id": str(uuid4()),
+                    "import_operation_id": third["operation_id"],
+                    "candidate_id": third["candidate_id"],
+                }
+                pair = await asyncio.gather(
+                    *(
+                        client.post(admin + "/problem-publications", json=b, headers=headers)
+                        for b in (b1, b2)
+                    )
+                )
+                assert sorted(r.status_code for r in pair) == [201, 409], [
+                    (r.status_code, r.text) for r in pair
+                ]
+                latest = next(r.json() for r in pair if r.status_code == 201)
+                active = latest["published_version_id"]
+                assert latest["problem"]["revision"] == 2
+                assert (
+                    await c.fetchval(
+                        "SELECT count(*) FROM problem_versions WHERE problem_id=$1", UUID(pid)
+                    )
+                    == 2
+                )
+                assert dict(
+                    await c.fetchrow("SELECT * FROM problem_versions WHERE id=$1", UUID(vid))
+                ) == dict(old)
+                # The local CLI uses the same publication/admission service; rejected
+                # toolchains do not create rows, and successful v1/v2 keep exact bytes.
+                local = tmp_path / "cli"
+                local.mkdir(mode=0o700)
+                for name, data in payload.items():
+                    (local / name).write_bytes(data)
+                env = base | {
+                    "KF_OJ_DATABASE_URL": make_url(postgres_url)
+                    .set(username=roles["api"])
+                    .render_as_string(hide_password=False),
+                    "KF_OJ_ARTIFACT_ROOT": str(artifact),
+                    "KF_OJ_JUDGE_ADMISSION": policy.model_dump_json(),
+                }
+                cli = await asyncio.to_thread(
+                    subprocess.run,
+                    [sys.executable, "-m", "kf_oj.modules.problems.cli", str(local)],
+                    env=env,
+                    capture_output=True,
+                    timeout=90,
+                )
+                assert cli.returncode == 0, cli.stderr.decode()[-1500:]
+                active = str(
+                    await c.fetchval(
+                        "SELECT active_version_id FROM problems WHERE id=$1", UUID(pid)
+                    )
+                )
+                assert (
+                    await c.fetchval(
+                        "SELECT max(revision) FROM problem_versions WHERE problem_id=$1", UUID(pid)
+                    )
+                    == 3
+                )
+                # Disable is preconditioned and idempotent; new admission and public
+                # reads close, while immutable accepted history remains authorized.
+                toggle = {
+                    "operation_id": str(uuid4()),
+                    "expected_active_version_id": active,
+                    "expected_enabled": True,
+                }
+                r = await client.post(
+                    admin + f"/problems/{pid}/disable", json=toggle, headers=headers
+                )
+                assert r.status_code == 200, r.text
+                assert not r.json()["problem"]["enabled"]
+                assert (
+                    await client.post(
+                        admin + f"/problems/{pid}/disable", json=toggle, headers=headers
+                    )
+                ).json() == r.json()
+                assert (await client.get("/api/v1/problems/versions/" + vid)).status_code == 404
+                assert not (await client.get("/api/v1/problems")).json()["items"]
+                headers = await login("web_student")
+                submission = {
+                    "problem_version_id": active,
+                    "language": "c17",
+                    "source": "synthetic source, never executed",
+                    "idempotency_key": str(uuid4()),
+                }
+                assert (
+                    await client.post("/api/v1/submissions", json=submission, headers=headers)
+                ).status_code == 404
+                headers = await login("web_admin")
+                cli = await asyncio.to_thread(
+                    subprocess.run,
+                    [sys.executable, "-m", "kf_oj.modules.problems.cli", str(local)],
+                    env=env,
+                    capture_output=True,
+                    timeout=90,
+                )
+                assert cli.returncode != 0
+                restore = {
+                    "operation_id": str(uuid4()),
+                    "expected_active_version_id": active,
+                    "expected_enabled": False,
+                }
+                r = await client.post(
+                    admin + f"/problems/{pid}/restore", json=restore, headers=headers
+                )
+                assert r.status_code == 200, r.text
+                assert r.json()["problem"]["enabled"]
+                history = (await client.get(admin + f"/problems/{pid}/versions?limit=1")).json()
+                assert history["items"][0]["revision"] == 3 and history["next_before"] == 3
+                history2 = (
+                    await client.get(admin + f"/problems/{pid}/versions?limit=1&before=3")
+                ).json()
+                assert history2["items"][0]["revision"] == 2
+                assert "statement" not in json.dumps(history) and "digest" not in json.dumps(
+                    history
+                )
+                # Inject a transaction-local audit failure through a trusted test
+                # DB trigger; no version/active pointer/import receipt survives rollback.
+                rollback, _ = await ready("rollback-synthetic")
+                await c.execute(
+                    """CREATE FUNCTION reject_publish_audit() RETURNS trigger LANGUAGE plpgsql AS $$
+                    BEGIN IF NEW.action='published' THEN
+                    RAISE EXCEPTION 'synthetic audit failure'; END IF; RETURN NEW; END $$"""
+                )
+                await c.execute(
+                    "CREATE TRIGGER reject_publish_audit BEFORE INSERT "
+                    "ON problem_maintenance_events "
+                    "FOR EACH ROW EXECUTE FUNCTION reject_publish_audit()"
+                )
+                r, rb = await publish(rollback)
+                assert r.status_code == 503, r.text
+                assert not await c.fetchval(
+                    "SELECT EXISTS(SELECT 1 FROM problems WHERE slug='rollback-synthetic')"
+                )
+                assert not await c.fetchval(
+                    "SELECT EXISTS(SELECT 1 FROM problem_maintenance_operations WHERE id=$1)",
+                    UUID(rb["operation_id"]),
+                )
+                assert (
+                    await c.fetchval(
+                        "SELECT state FROM problem_import_operations WHERE id=$1",
+                        UUID(rollback["operation_id"]),
+                    )
+                    == "ready"
+                )
+                await c.execute("DROP TRIGGER reject_publish_audit ON problem_maintenance_events")
+                await c.execute("DROP FUNCTION reject_publish_audit()")
+                r = await client.post(admin + "/problem-publications", json=rb, headers=headers)
+                assert r.status_code == 201, r.text
+                multi, _ = await ready("managed-multi", multi=True)
+                r, _ = await publish(multi)
+                assert r.status_code == 201, r.text
+                assert (
+                    len(
+                        (
+                            await client.get(
+                                "/api/v1/problems/versions/" + r.json()["published_version_id"]
+                            )
+                        ).json()["language_configurations"]
+                    )
+                    == 4
+                )
+                # Live foreign administrator cannot adopt another actor's candidate or receipt.
+                headers = await login("web_backup")
+                assert (
+                    await client.get(admin + "/problem-maintenance/" + body["operation_id"])
+                ).status_code == 404
+                assert (
+                    await client.post(admin + "/problem-publications", json=body, headers=headers)
+                ).status_code == 404
+                headers = await login("web_admin")
+                audit = (await client.get(admin + "/maintenance-audit?limit=100")).json()["items"]
+                assert any(
+                    e["action"] == "authority_bootstrap"
+                    and e["actor_kind"] == "operator"
+                    and e["actor_id"] is None
+                    for e in audit
+                )
+                assert (
+                    sum(
+                        e["operation_id"] == body["operation_id"] and e["action"] == "published"
+                        for e in audit
+                    )
+                    == 1
+                )
+                assert "hidden" not in json.dumps(audit) and "password" not in json.dumps(audit)
+                # Finite page traversal excludes duplicates and has precise committed SQL count.
+                seen = set()
+                cursor = None
+                for _ in range(100):
+                    r = await client.get(
+                        admin + "/maintenance-audit",
+                        params={"limit": 2} | ({"before": cursor} if cursor else {}),
+                    )
+                    assert r.status_code == 200, r.text
+                    page = r.json()
+                    ids = {e["id"] for e in page["items"]}
+                    assert not seen & ids
+                    seen |= ids
+                    cursor = page["next_before"]
+                    if cursor is None:
+                        break
+                else:
+                    raise AssertionError("audit page bound")
+                assert len(seen) == await c.fetchval(
+                    "SELECT count(*) FROM problem_maintenance_events"
+                )
+                for role in ("kf_oj_platform", "kf_oj_problem_import", "kf_oj_coordinator"):
+                    await c.execute("SET ROLE " + role)
+                    for table in ("problem_maintenance_operations", "problem_maintenance_events"):
+                        for command in (f"UPDATE {table} SET id=id", f"DELETE FROM {table}"):
+                            with pytest.raises(Exception) as error:
+                                await c.execute(command)
+                            assert getattr(error.value, "sqlstate", None) == "42501"
+                    await c.execute("RESET ROLE")
+                with pytest.raises(Exception) as error:
+                    await c.execute(
+                        "UPDATE problem_versions SET title=title WHERE id=$1", UUID(vid)
+                    )
+                assert getattr(error.value, "sqlstate", None) == "23514"
+                concurrent, _ = await ready("revocation-synthetic")
+                revoke_body = {
+                    "operation_id": str(uuid4()),
+                    "import_operation_id": concurrent["operation_id"],
+                    "candidate_id": concurrent["candidate_id"],
+                    "expected_active_version_id": None,
+                }
+                published_result, _ = await asyncio.gather(
+                    client.post(admin + "/problem-publications", json=revoke_body, headers=headers),
+                    c.execute("SELECT * FROM kf_manage_account_authority('web_admin','revoke')"),
+                )
+                assert published_result.status_code in (201, 401, 403, 503), published_result.text
+                committed = await c.fetchval(
+                    "SELECT EXISTS(SELECT 1 FROM problem_maintenance_operations WHERE id=$1)",
+                    UUID(revoke_body["operation_id"]),
+                )
+                assert committed == (published_result.status_code == 201)
+                assert (
+                    await c.fetchval(
+                        "SELECT EXISTS(SELECT 1 FROM problems WHERE slug='revocation-synthetic')"
+                    )
+                    == committed
+                )
+                assert await c.fetchval(
+                    "SELECT count(*) FROM problem_maintenance_events "
+                    "WHERE operation_id=$1 AND action='published'",
+                    UUID(revoke_body["operation_id"]),
+                ) == int(committed)
+                assert (await client.get(admin + "/problems")).status_code == 401
+                assert (
+                    await client.post(admin + "/problem-publications", json=body, headers=headers)
+                ).status_code == 401
+                print(
+                    "ACTUAL_MAINTENANCE_CLI_V1_V2_UUID_CONCURRENCY_AUDIT_ROLLBACK_DISABLE_ACCEPTED",
+                    flush=True,
+                )
+        finally:
+            stop("api")
+            stop("worker")
+            for role in roles.values():
+                await c.execute("DROP OWNED BY " + role)
+                await c.execute("DROP ROLE " + role)
+            await c.close()
+
+    asyncio.run(run())