feat(database): add judging schema and separate write roles

backend/migrations/README.md

@@ -0,0 +1,23 @@
+# 数据库所有权
+
+Alembic 是唯一的 schema 入口。先对专用 PostgreSQL 18 数据库运行 `uv run --locked alembic
+upgrade head`，API 不在启动时创建表。迁移账号拥有该数据库对象并需要 `CREATEROLE`；它不作为
+任何运行服务账号。生产配置及登录角色由部署方提供，不在源码里设置默认数据库口令。
+
+初始迁移建立两个受限的 `NOLOGIN` 权限组：`kf_oj_platform` 和 `kf_oj_coordinator`。
+平台组创建 submission / queued evaluation / outbox，不能指定或更新执行状态、attempt 或成绩。
+协调器组领取和更新评测、插入 attempt / 用例成绩及恢复通知，不能修改账号、题库或提交身份。
+Worker 没有数据库角色，也不需要数据库连接配置。
+
+服务使用各自独立的普通 `LOGIN` 角色，分别加入对应权限组；登录角色不能拥有数据库/表、
+不能加入迁移账号或其他高权限组，且必须为 `NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS`。
+初始迁移拒绝已有的高权限同名组和继承其他角色的同名组。应用启动时不运行 migration。
+降级删除本项目业务表与触发器，保留权限组及操作方配置的登录账号，避免删除集群级身份。
+
+发布是一次单向转换：先在事务内创建版本和连续测试点，再设置 `published_at`。
+数据库触发器拒绝再次发布、修改已发布版本和追加/改写已发布测试点；评测通过复合外键
+永久绑定 submission 的版本，当前 attempt 和用例成绩也由复合外键绑定同一评测及版本。
+工件引用具有数据库生成的 `ready` 状态，引用存在时不能转为 `deleting` 或删除元数据。
+
+真实数据库验收使用显式 `KF_OJ_TEST_POSTGRES` 管理连接。每个测试只创建和删除随机命名的
+`kf_oj_test_*` 数据库，不清空已有业务库。例行无数据库测试会明确跳过这些验收；跳过不算通过。

backend/migrations/env.py

@@ -5,14 +5,19 @@ from alembic import context
 from sqlalchemy import Connection
 
 from kf_oj.core.settings import Settings
+from kf_oj.infra.artifacts import models as artifact_models  # noqa: F401
 from kf_oj.infra.db.base import Base
 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.problems import models as problem_models  # noqa: F401
+from kf_oj.modules.submissions import models as submission_models  # noqa: F401
 
 config = context.config
 if config.config_file_name is not None:
     fileConfig(config.config_file_name)
 
-# Register each business module's ORM models here when its schema is implemented.
+# Explicit composition root. Infra and domains do not import foreign ORM models.
 target_metadata = Base.metadata
 
 

backend/migrations/versions/0001_judging_foundation_judging_foundation_and_role_boundaries.py

@@ -0,0 +1,588 @@
+"""judging foundation and role boundaries
+
+Revision ID: 0001_judging_foundation
+Revises:
+Create Date: 2026-10-09 00:57:34.004943
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0001_judging_foundation"
+down_revision: str | None = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def install_guards_and_permissions() -> None:
+    # Serialize case insertion against the one-way publication transition.
+    op.execute("""
+        CREATE FUNCTION public.kf_immutable_case() RETURNS trigger LANGUAGE plpgsql
+        SET search_path = pg_catalog, public AS $$
+        DECLARE published timestamptz;
+        BEGIN
+            IF TG_OP = 'UPDATE' THEN
+                RAISE EXCEPTION 'test cases are immutable' USING ERRCODE = '23514';
+            END IF;
+            SELECT published_at INTO published FROM public.problem_versions
+                WHERE id = CASE WHEN TG_OP = 'DELETE' THEN OLD.version_id ELSE NEW.version_id END
+                FOR UPDATE;
+            IF published IS NOT NULL THEN
+                RAISE EXCEPTION 'published test cases are immutable' USING ERRCODE = '23514';
+            END IF;
+            IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
+            RETURN NEW;
+        END $$
+    """)
+    op.execute("""
+        CREATE TRIGGER kf_immutable_case BEFORE INSERT OR UPDATE OR DELETE ON public.problem_cases
+        FOR EACH ROW EXECUTE FUNCTION public.kf_immutable_case()
+    """)
+    op.execute("""
+        CREATE FUNCTION public.kf_immutable_version() RETURNS trigger LANGUAGE plpgsql
+        SET search_path = pg_catalog, public AS $$
+        DECLARE count_cases integer; last_position integer;
+        BEGIN
+            IF TG_OP = 'INSERT' THEN
+                IF NEW.published_at IS NOT NULL THEN
+                    RAISE EXCEPTION 'publish after inserting cases' USING ERRCODE = '23514';
+                END IF;
+                RETURN NEW;
+            END IF;
+            IF TG_OP = 'DELETE' THEN
+                IF OLD.published_at IS NOT NULL THEN
+                    RAISE EXCEPTION 'published version is immutable' USING ERRCODE = '23514';
+                END IF;
+                RETURN OLD;
+            END IF;
+            IF OLD.published_at IS NOT NULL OR NEW.published_at IS NULL
+                OR (to_jsonb(NEW) - 'published_at') != (to_jsonb(OLD) - 'published_at') THEN
+                RAISE EXCEPTION 'version may only be published once' USING ERRCODE = '23514';
+            END IF;
+            SELECT count(*), max(position) INTO count_cases, last_position
+                FROM public.problem_cases WHERE version_id = OLD.id;
+            IF count_cases NOT BETWEEN 1 AND 256 OR last_position != count_cases - 1 THEN
+                RAISE EXCEPTION 'publication requires contiguous ordered cases'
+                    USING ERRCODE = '23514';
+            END IF;
+            RETURN NEW;
+        END $$
+    """)
+    op.execute("""
+        CREATE TRIGGER kf_immutable_version BEFORE INSERT OR UPDATE OR DELETE
+        ON public.problem_versions
+        FOR EACH ROW EXECUTE FUNCTION public.kf_immutable_version()
+    """)
+    op.execute("REVOKE ALL ON FUNCTION public.kf_immutable_case() FROM PUBLIC")
+    op.execute("REVOKE ALL ON FUNCTION public.kf_immutable_version() FROM PUBLIC")
+    # Migration owner needs CREATEROLE; runtime logins are separate, unprivileged group members.
+    for role in ("kf_oj_platform", "kf_oj_coordinator"):
+        op.execute(f"""
+            DO $$ BEGIN
+                IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN
+                    CREATE ROLE {role} NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOBYPASSRLS;
+                ELSIF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}' AND
+                    (rolcanlogin OR rolsuper OR rolcreatedb OR rolcreaterole OR rolbypassrls))
+                    OR EXISTS (SELECT 1 FROM pg_auth_members m JOIN pg_roles r ON r.oid=m.member
+                        WHERE r.rolname='{role}') THEN
+                    RAISE EXCEPTION 'unsafe existing runtime role';
+                END IF;
+            END $$
+        """)
+    tables = (
+        "accounts,sessions,login_throttles,artifacts,problems,problem_versions,problem_cases,"
+        "submissions,evaluations,judge_attempts,case_results,outbox"
+    )
+    op.execute(f"REVOKE ALL ON {tables} FROM PUBLIC, kf_oj_platform, kf_oj_coordinator")
+    op.execute("REVOKE CREATE ON SCHEMA public FROM PUBLIC")
+    op.execute("GRANT USAGE ON SCHEMA public TO kf_oj_platform, kf_oj_coordinator")
+    op.execute(f"GRANT SELECT ON {tables} TO kf_oj_platform")
+    op.execute("GRANT INSERT,UPDATE,DELETE ON accounts,sessions,login_throttles TO kf_oj_platform")
+    op.execute("GRANT INSERT,SELECT,DELETE ON artifacts TO kf_oj_platform")
+    op.execute("GRANT UPDATE(state) ON artifacts TO kf_oj_platform")
+    op.execute(
+        "GRANT INSERT ON problems,problem_versions,problem_cases,submissions,outbox "
+        "TO kf_oj_platform"
+    )
+    op.execute("GRANT UPDATE(active_version_id,enabled) ON problems TO kf_oj_platform")
+    op.execute("GRANT UPDATE(published_at) ON problem_versions TO kf_oj_platform")
+    op.execute("GRANT INSERT(id,submission_id,problem_version_id) ON evaluations TO kf_oj_platform")
+    op.execute(
+        "GRANT UPDATE(failures,next_attempt_at,locked_until,lock_owner,delivered_at,"
+        "last_error_code) ON outbox TO kf_oj_platform"
+    )
+    op.execute(
+        "GRANT SELECT ON problems,problem_versions,problem_cases,artifacts,submissions,"
+        "evaluations,judge_attempts,case_results,outbox TO kf_oj_coordinator"
+    )
+    op.execute(
+        "GRANT UPDATE(state,fencing_version,attempts_count,current_attempt_id,not_before,"
+        "cancel_requested_at,updated_at,verdict,failure_code,final_report,final_report_digest) "
+        "ON evaluations TO kf_oj_coordinator"
+    )
+    op.execute("GRANT INSERT ON judge_attempts,case_results TO kf_oj_coordinator")
+    op.execute(
+        "GRANT UPDATE(expires_at,revoked_at,sequence,stage,case_index,report,report_digest,"
+        "cleanup_complete) ON judge_attempts TO kf_oj_coordinator"
+    )
+    # Recovery may enqueue the same immutable notification, but cannot change identity fields.
+    op.execute("GRANT INSERT ON outbox TO kf_oj_coordinator")
+    op.execute("GRANT UPDATE(next_attempt_at,delivered_at) ON outbox TO kf_oj_coordinator")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "accounts",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("username", sa.String(length=24), nullable=False),
+        sa.Column("password_hash", sa.Text(), nullable=False),
+        sa.Column("enabled", sa.Boolean(), server_default="true", nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "username ~ '^[a-z0-9_]{3,24}$'", name=op.f("ck_accounts_canonical_username")
+        ),
+        sa.CheckConstraint(
+            "char_length(password_hash) BETWEEN 40 AND 256",
+            name=op.f("ck_accounts_password_hash_size"),
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_accounts")),
+        sa.UniqueConstraint("username", name=op.f("uq_accounts_username")),
+    )
+    op.create_table(
+        "artifacts",
+        sa.Column("digest", sa.String(length=64), nullable=False),
+        sa.Column("bytes", sa.BigInteger(), nullable=False),
+        sa.Column("state", sa.String(length=8), server_default="ready", nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint("digest ~ '^[0-9a-f]{64}$'", name=op.f("ck_artifacts_digest")),
+        sa.CheckConstraint("state IN ('ready', 'deleting')", name=op.f("ck_artifacts_state")),
+        sa.CheckConstraint("bytes BETWEEN 0 AND 67108864", name=op.f("ck_artifacts_bytes")),
+        sa.PrimaryKeyConstraint("digest", name=op.f("pk_artifacts")),
+        sa.UniqueConstraint("digest", "state", name=op.f("uq_artifacts_digest")),
+    )
+    op.create_table(
+        "login_throttles",
+        sa.Column("bucket_hash", sa.String(length=64), nullable=False),
+        sa.Column("attempts", sa.Integer(), nullable=False),
+        sa.Column("reset_at", sa.DateTime(timezone=True), nullable=False),
+        sa.CheckConstraint(
+            "bucket_hash ~ '^[0-9a-f]{64}$'", name=op.f("ck_login_throttles_bucket_digest")
+        ),
+        sa.CheckConstraint(
+            "attempts BETWEEN 1 AND 10", name=op.f("ck_login_throttles_attempt_budget")
+        ),
+        sa.PrimaryKeyConstraint("bucket_hash", name=op.f("pk_login_throttles")),
+    )
+    op.create_index("ix_login_throttles_reset", "login_throttles", ["reset_at"], unique=False)
+    op.create_table(
+        "problems",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("slug", sa.String(length=64), nullable=False),
+        sa.Column("active_version_id", sa.Uuid(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default="true", nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint("slug ~ '^[a-z0-9][a-z0-9_-]{0,63}$'", name=op.f("ck_problems_slug")),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_problems")),
+        sa.UniqueConstraint("slug", name=op.f("uq_problems_slug")),
+    )
+    op.create_table(
+        "problem_versions",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("problem_id", sa.Uuid(), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column("title", sa.String(length=160), nullable=False),
+        sa.Column("statement", sa.Text(), nullable=False),
+        sa.Column("languages", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("toolchain", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("compile_limits", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("run_limits", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("job_wall_time_us", sa.BigInteger(), nullable=False),
+        sa.Column("comparison", sa.String(length=16), server_default="exact_bytes", nullable=False),
+        sa.Column("public_samples", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("published_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "comparison = 'exact_bytes'", name=op.f("ck_problem_versions_comparison")
+        ),
+        sa.CheckConstraint(
+            "char_length(title) BETWEEN 1 AND 160", name=op.f("ck_problem_versions_title_size")
+        ),
+        sa.CheckConstraint(
+            "job_wall_time_us BETWEEN 1000 AND 3600000000",
+            name=op.f("ck_problem_versions_job_wall"),
+        ),
+        sa.CheckConstraint(
+            "octet_length(statement) BETWEEN 1 AND 262144",
+            name=op.f("ck_problem_versions_statement_bytes"),
+        ),
+        sa.CheckConstraint("revision > 0", name=op.f("ck_problem_versions_revision")),
+        sa.ForeignKeyConstraint(
+            ["problem_id"], ["problems.id"], name=op.f("fk_problem_versions_problem_id_problems")
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_problem_versions")),
+        sa.UniqueConstraint("id", "problem_id", name=op.f("uq_problem_versions_id")),
+        sa.UniqueConstraint("problem_id", "revision", name=op.f("uq_problem_versions_problem_id")),
+    )
+    op.create_index(
+        "ix_problem_versions_public",
+        "problem_versions",
+        ["problem_id", "published_at"],
+        unique=False,
+    )
+    op.create_table(
+        "sessions",
+        sa.Column("token_hash", sa.String(length=64), nullable=False),
+        sa.Column("account_id", sa.Uuid(), nullable=False),
+        sa.Column("csrf_hash", sa.String(length=64), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint("csrf_hash ~ '^[0-9a-f]{64}$'", name=op.f("ck_sessions_csrf_digest")),
+        sa.CheckConstraint("token_hash ~ '^[0-9a-f]{64}$'", name=op.f("ck_sessions_token_digest")),
+        sa.CheckConstraint("expires_at > created_at", name=op.f("ck_sessions_expiry")),
+        sa.ForeignKeyConstraint(
+            ["account_id"], ["accounts.id"], name=op.f("fk_sessions_account_id_accounts")
+        ),
+        sa.PrimaryKeyConstraint("token_hash", name=op.f("pk_sessions")),
+    )
+    op.create_index(
+        "ix_sessions_account_expiry", "sessions", ["account_id", "expires_at"], unique=False
+    )
+    op.create_table(
+        "problem_cases",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("version_id", sa.Uuid(), nullable=False),
+        sa.Column("position", sa.Integer(), nullable=False),
+        sa.Column("input_digest", sa.String(length=64), nullable=False),
+        sa.Column("expected_digest", sa.String(length=64), nullable=False),
+        sa.Column(
+            "artifact_state",
+            sa.String(length=8),
+            sa.Computed("'ready'", persisted=True),
+            nullable=False,
+        ),
+        sa.CheckConstraint("position BETWEEN 0 AND 255", name=op.f("ck_problem_cases_position")),
+        sa.ForeignKeyConstraint(
+            ["expected_digest", "artifact_state"],
+            ["artifacts.digest", "artifacts.state"],
+            name="fk_problem_cases_expected_digest_artifacts",
+        ),
+        sa.ForeignKeyConstraint(
+            ["input_digest", "artifact_state"],
+            ["artifacts.digest", "artifacts.state"],
+            name=op.f("fk_problem_cases_input_digest_artifacts"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["version_id"],
+            ["problem_versions.id"],
+            name=op.f("fk_problem_cases_version_id_problem_versions"),
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_problem_cases")),
+        sa.UniqueConstraint("id", "version_id", name=op.f("uq_problem_cases_id")),
+        sa.UniqueConstraint("version_id", "position", name=op.f("uq_problem_cases_version_id")),
+    )
+    op.create_table(
+        "submissions",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("account_id", sa.Uuid(), nullable=False),
+        sa.Column("problem_id", sa.Uuid(), nullable=False),
+        sa.Column("problem_version_id", sa.Uuid(), nullable=False),
+        sa.Column("language", sa.String(length=8), nullable=False),
+        sa.Column("source_digest", sa.String(length=64), nullable=False),
+        sa.Column(
+            "artifact_state",
+            sa.String(length=8),
+            sa.Computed("'ready'", persisted=True),
+            nullable=False,
+        ),
+        sa.Column("idempotency_key", sa.Uuid(), nullable=False),
+        sa.Column("request_digest", sa.String(length=64), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint("language IN ('c17', 'cpp20')", name=op.f("ck_submissions_language")),
+        sa.CheckConstraint(
+            "request_digest ~ '^[0-9a-f]{64}$'", name=op.f("ck_submissions_request_digest")
+        ),
+        sa.ForeignKeyConstraint(
+            ["account_id"], ["accounts.id"], name=op.f("fk_submissions_account_id_accounts")
+        ),
+        sa.ForeignKeyConstraint(
+            ["problem_version_id", "problem_id"],
+            ["problem_versions.id", "problem_versions.problem_id"],
+            name=op.f("fk_submissions_problem_version_id_problem_versions"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["source_digest", "artifact_state"],
+            ["artifacts.digest", "artifacts.state"],
+            name=op.f("fk_submissions_source_digest_artifacts"),
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_submissions")),
+        sa.UniqueConstraint(
+            "account_id", "idempotency_key", name=op.f("uq_submissions_account_id")
+        ),
+        sa.UniqueConstraint("id", "problem_version_id", name=op.f("uq_submissions_id")),
+    )
+    op.create_index(
+        "ix_submissions_account_history",
+        "submissions",
+        ["account_id", "created_at", "id"],
+        unique=False,
+    )
+    op.create_table(
+        "evaluations",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("submission_id", sa.Uuid(), nullable=False),
+        sa.Column("problem_version_id", sa.Uuid(), nullable=False),
+        sa.Column("state", sa.String(length=16), server_default="queued", nullable=False),
+        sa.Column("fencing_version", sa.BigInteger(), server_default="0", nullable=False),
+        sa.Column("attempts_count", sa.Integer(), server_default="0", nullable=False),
+        sa.Column("max_attempts", sa.Integer(), server_default="3", nullable=False),
+        sa.Column("current_attempt_id", sa.Uuid(), nullable=True),
+        sa.Column(
+            "not_before",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column("cancel_requested_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column("verdict", sa.String(length=32), nullable=True),
+        sa.Column("failure_code", sa.String(length=32), nullable=True),
+        sa.Column("final_report", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+        sa.Column("final_report_digest", sa.String(length=64), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "state IN ('queued','claimed','compiling','running','cancelling',"
+            "'finished','cancelled','failed')",
+            name=op.f("ck_evaluations_state"),
+        ),
+        sa.CheckConstraint(
+            "state NOT IN ('claimed','compiling','running','cancelling') "
+            "OR current_attempt_id IS NOT NULL",
+            name=op.f("ck_evaluations_active_attempt"),
+        ),
+        sa.CheckConstraint(
+            "attempts_count BETWEEN 0 AND max_attempts", name=op.f("ck_evaluations_attempts_count")
+        ),
+        sa.CheckConstraint("fencing_version >= 0", name=op.f("ck_evaluations_fence")),
+        sa.CheckConstraint(
+            "max_attempts BETWEEN 1 AND 5", name=op.f("ck_evaluations_max_attempts")
+        ),
+        sa.ForeignKeyConstraint(
+            ["submission_id", "problem_version_id"],
+            ["submissions.id", "submissions.problem_version_id"],
+            name=op.f("fk_evaluations_submission_id_submissions"),
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_evaluations")),
+        sa.UniqueConstraint("id", "problem_version_id", name=op.f("uq_evaluations_id")),
+        sa.UniqueConstraint("submission_id", name=op.f("uq_evaluations_submission_id")),
+    )
+    op.create_index(
+        "ix_evaluations_recovery", "evaluations", ["state", "not_before", "id"], unique=False
+    )
+    op.create_table(
+        "judge_attempts",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("evaluation_id", sa.Uuid(), nullable=False),
+        sa.Column("worker_id", sa.Uuid(), nullable=False),
+        sa.Column("number", sa.Integer(), nullable=False),
+        sa.Column("fencing_version", sa.BigInteger(), nullable=False),
+        sa.Column("lease_token_hash", sa.String(length=64), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("manifest", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("manifest_digest", sa.String(length=64), nullable=False),
+        sa.Column("sequence", sa.BigInteger(), server_default="0", nullable=False),
+        sa.Column("stage", sa.String(length=16), nullable=True),
+        sa.Column("case_index", sa.Integer(), nullable=True),
+        sa.Column("report", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+        sa.Column("report_digest", sa.String(length=64), nullable=True),
+        sa.Column("cleanup_complete", sa.Boolean(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "lease_token_hash ~ '^[0-9a-f]{64}$'", name=op.f("ck_judge_attempts_token_digest")
+        ),
+        sa.CheckConstraint(
+            "manifest_digest ~ '^[0-9a-f]{64}$'", name=op.f("ck_judge_attempts_manifest_digest")
+        ),
+        sa.CheckConstraint("fencing_version > 0", name=op.f("ck_judge_attempts_fence")),
+        sa.CheckConstraint("number BETWEEN 1 AND 5", name=op.f("ck_judge_attempts_number")),
+        sa.CheckConstraint("sequence >= 0", name=op.f("ck_judge_attempts_sequence")),
+        sa.ForeignKeyConstraint(
+            ["evaluation_id"],
+            ["evaluations.id"],
+            name=op.f("fk_judge_attempts_evaluation_id_evaluations"),
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_judge_attempts")),
+        sa.UniqueConstraint(
+            "evaluation_id", "fencing_version", name="uq_judge_attempts_evaluation_fence"
+        ),
+        sa.UniqueConstraint("evaluation_id", "number", name="uq_judge_attempts_evaluation_number"),
+        sa.UniqueConstraint("id", "evaluation_id", name=op.f("uq_judge_attempts_id")),
+    )
+    op.create_index(
+        "ix_judge_attempts_expiry", "judge_attempts", ["expires_at", "evaluation_id"], unique=False
+    )
+    op.create_table(
+        "outbox",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("evaluation_id", sa.Uuid(), nullable=False),
+        sa.Column("kind", sa.String(length=32), server_default="judge.evaluation", nullable=False),
+        sa.Column("failures", sa.Integer(), server_default="0", nullable=False),
+        sa.Column(
+            "next_attempt_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column("locked_until", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("lock_owner", sa.Uuid(), nullable=True),
+        sa.Column("delivered_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_error_code", sa.String(length=32), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint("kind = 'judge.evaluation'", name=op.f("ck_outbox_kind")),
+        sa.CheckConstraint("failures BETWEEN 0 AND 100", name=op.f("ck_outbox_failure_budget")),
+        sa.ForeignKeyConstraint(
+            ["evaluation_id"], ["evaluations.id"], name=op.f("fk_outbox_evaluation_id_evaluations")
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_outbox")),
+        sa.UniqueConstraint("evaluation_id", name=op.f("uq_outbox_evaluation_id")),
+    )
+    op.create_index(
+        "ix_outbox_delivery", "outbox", ["delivered_at", "next_attempt_at", "id"], unique=False
+    )
+    op.create_table(
+        "case_results",
+        sa.Column("attempt_id", sa.Uuid(), nullable=False),
+        sa.Column("case_id", sa.Uuid(), nullable=False),
+        sa.Column("evaluation_id", sa.Uuid(), nullable=False),
+        sa.Column("problem_version_id", sa.Uuid(), nullable=False),
+        sa.Column("position", sa.Integer(), nullable=False),
+        sa.Column("verdict", sa.String(length=32), nullable=False),
+        sa.Column("cpu_time_us", sa.BigInteger(), nullable=True),
+        sa.Column("memory_peak_bytes", sa.BigInteger(), nullable=True),
+        sa.CheckConstraint(
+            "cpu_time_us IS NULL OR cpu_time_us >= 0", name=op.f("ck_case_results_cpu")
+        ),
+        sa.CheckConstraint(
+            "memory_peak_bytes IS NULL OR memory_peak_bytes >= 0",
+            name=op.f("ck_case_results_memory"),
+        ),
+        sa.CheckConstraint("position BETWEEN 0 AND 255", name=op.f("ck_case_results_position")),
+        sa.ForeignKeyConstraint(
+            ["attempt_id", "evaluation_id"],
+            ["judge_attempts.id", "judge_attempts.evaluation_id"],
+            name=op.f("fk_case_results_attempt_id_judge_attempts"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["case_id", "problem_version_id"],
+            ["problem_cases.id", "problem_cases.version_id"],
+            name=op.f("fk_case_results_case_id_problem_cases"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["evaluation_id", "problem_version_id"],
+            ["evaluations.id", "evaluations.problem_version_id"],
+            name=op.f("fk_case_results_evaluation_id_evaluations"),
+        ),
+        sa.PrimaryKeyConstraint("attempt_id", "case_id", name=op.f("pk_case_results")),
+        sa.UniqueConstraint("attempt_id", "position", name=op.f("uq_case_results_attempt_id")),
+    )
+
+    op.create_foreign_key(
+        "fk_problems_active_version",
+        "problems",
+        "problem_versions",
+        ["active_version_id", "id"],
+        ["id", "problem_id"],
+        deferrable=True,
+        initially="DEFERRED",
+    )
+    op.create_foreign_key(
+        "fk_evaluations_current_attempt",
+        "evaluations",
+        "judge_attempts",
+        ["current_attempt_id", "id"],
+        ["id", "evaluation_id"],
+        deferrable=True,
+        initially="DEFERRED",
+    )
+    install_guards_and_permissions()
+
+
+def downgrade() -> None:
+    op.drop_constraint("fk_evaluations_current_attempt", "evaluations", type_="foreignkey")
+    op.drop_constraint("fk_problems_active_version", "problems", type_="foreignkey")
+    op.drop_table("case_results")
+    op.drop_index("ix_outbox_delivery", table_name="outbox")
+    op.drop_table("outbox")
+    op.drop_index("ix_judge_attempts_expiry", table_name="judge_attempts")
+    op.drop_table("judge_attempts")
+    op.drop_index("ix_evaluations_recovery", table_name="evaluations")
+    op.drop_table("evaluations")
+    op.drop_index("ix_submissions_account_history", table_name="submissions")
+    op.drop_table("submissions")
+    op.drop_table("problem_cases")
+    op.drop_index("ix_sessions_account_expiry", table_name="sessions")
+    op.drop_table("sessions")
+    op.drop_index("ix_problem_versions_public", table_name="problem_versions")
+    op.drop_table("problem_versions")
+    op.drop_table("problems")
+    op.drop_index("ix_login_throttles_reset", table_name="login_throttles")
+    op.drop_table("login_throttles")
+    op.drop_table("artifacts")
+    op.drop_table("accounts")
+    op.execute("DROP FUNCTION public.kf_immutable_version()")
+    op.execute("DROP FUNCTION public.kf_immutable_case()")
+    # NOLOGIN groups are intentionally retained: a downgrade never deletes operator roles.

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

@@ -0,0 +1,21 @@
+"""Metadata ownership and lifecycle for immutable, digest-addressed bytes."""
+
+from sqlalchemy import BigInteger, CheckConstraint, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from kf_oj.infra.db.base import Base
+from kf_oj.infra.db.columns import CreatedAt
+
+
+class Artifact(CreatedAt, Base):
+    __tablename__ = "artifacts"
+    __table_args__ = (
+        CheckConstraint("digest ~ '^[0-9a-f]{64}$'", name="digest"),
+        CheckConstraint("bytes BETWEEN 0 AND 67108864", name="bytes"),
+        CheckConstraint("state IN ('ready', 'deleting')", name="state"),
+        UniqueConstraint("digest", "state"),
+    )
+
+    digest: Mapped[str] = mapped_column(String(64), primary_key=True)
+    bytes: Mapped[int] = mapped_column(BigInteger)
+    state: Mapped[str] = mapped_column(String(8), server_default="ready")

backend/src/kf_oj/infra/db/columns.py

@@ -0,0 +1,12 @@
+"""Shared column types; business schemas are registered by composition roots."""
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+class CreatedAt:
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )

backend/src/kf_oj/infra/messaging/models.py

@@ -0,0 +1,31 @@
+"""Transactional notifications; contains IDs, never source or hidden artifacts."""
+
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from kf_oj.infra.db.base import Base
+from kf_oj.infra.db.columns import CreatedAt
+
+
+class Outbox(CreatedAt, Base):
+    __tablename__ = "outbox"
+    __table_args__ = (
+        CheckConstraint("failures BETWEEN 0 AND 100", name="failure_budget"),
+        CheckConstraint("kind = 'judge.evaluation'", name="kind"),
+        Index("ix_outbox_delivery", "delivered_at", "next_attempt_at", "id"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    evaluation_id: Mapped[UUID] = mapped_column(ForeignKey("evaluations.id"), unique=True)
+    kind: Mapped[str] = mapped_column(String(32), server_default="judge.evaluation")
+    failures: Mapped[int] = mapped_column(server_default="0")
+    next_attempt_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    locked_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    lock_owner: Mapped[UUID | None]
+    delivered_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_error_code: Mapped[str | None] = mapped_column(String(32))

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

@@ -0,0 +1,52 @@
+"""Identity-owned persistent account and revocable session records."""
+
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, Index, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from kf_oj.infra.db.base import Base
+from kf_oj.infra.db.columns import CreatedAt
+
+
+class Account(CreatedAt, Base):
+    __tablename__ = "accounts"
+    __table_args__ = (
+        CheckConstraint("username ~ '^[a-z0-9_]{3,24}$'", name="canonical_username"),
+        CheckConstraint("char_length(password_hash) BETWEEN 40 AND 256", name="password_hash_size"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    username: Mapped[str] = mapped_column(String(24), unique=True)
+    password_hash: Mapped[str] = mapped_column(Text)
+    enabled: Mapped[bool] = mapped_column(Boolean, server_default="true")
+
+
+class Session(CreatedAt, Base):
+    __tablename__ = "sessions"
+    __table_args__ = (
+        CheckConstraint("token_hash ~ '^[0-9a-f]{64}$'", name="token_digest"),
+        CheckConstraint("csrf_hash ~ '^[0-9a-f]{64}$'", name="csrf_digest"),
+        CheckConstraint("expires_at > created_at", name="expiry"),
+        Index("ix_sessions_account_expiry", "account_id", "expires_at"),
+    )
+
+    token_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
+    account_id: Mapped[UUID] = mapped_column(ForeignKey("accounts.id"))
+    csrf_hash: Mapped[str] = mapped_column(String(64))
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class LoginThrottle(Base):
+    __tablename__ = "login_throttles"
+    __table_args__ = (
+        CheckConstraint("bucket_hash ~ '^[0-9a-f]{64}$'", name="bucket_digest"),
+        CheckConstraint("attempts BETWEEN 1 AND 10", name="attempt_budget"),
+        Index("ix_login_throttles_reset", "reset_at"),
+    )
+
+    bucket_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
+    attempts: Mapped[int]
+    reset_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))

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

@@ -0,0 +1,96 @@
+"""Immutable published versions, ordered cases, and mutable public availability."""
+
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    CheckConstraint,
+    Computed,
+    DateTime,
+    ForeignKey,
+    ForeignKeyConstraint,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+)
+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 Problem(CreatedAt, Base):
+    __tablename__ = "problems"
+    __table_args__ = (
+        CheckConstraint("slug ~ '^[a-z0-9][a-z0-9_-]{0,63}$'", name="slug"),
+        ForeignKeyConstraint(
+            ["active_version_id", "id"],
+            ["problem_versions.id", "problem_versions.problem_id"],
+            name="fk_problems_active_version",
+            use_alter=True,
+            deferrable=True,
+            initially="DEFERRED",
+        ),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    slug: Mapped[str] = mapped_column(String(64), unique=True)
+    active_version_id: Mapped[UUID | None]
+    enabled: Mapped[bool] = mapped_column(Boolean, server_default="true")
+
+
+class ProblemVersion(CreatedAt, Base):
+    __tablename__ = "problem_versions"
+    __table_args__ = (
+        UniqueConstraint("id", "problem_id"),
+        UniqueConstraint("problem_id", "revision"),
+        CheckConstraint("revision > 0", name="revision"),
+        CheckConstraint("octet_length(statement) BETWEEN 1 AND 262144", name="statement_bytes"),
+        CheckConstraint("char_length(title) BETWEEN 1 AND 160", name="title_size"),
+        CheckConstraint("comparison = 'exact_bytes'", name="comparison"),
+        CheckConstraint("job_wall_time_us BETWEEN 1000 AND 3600000000", name="job_wall"),
+        Index("ix_problem_versions_public", "problem_id", "published_at"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    problem_id: Mapped[UUID] = mapped_column(ForeignKey("problems.id"))
+    revision: Mapped[int] = mapped_column(Integer)
+    title: Mapped[str] = mapped_column(String(160))
+    statement: Mapped[str] = mapped_column(Text)
+    languages: Mapped[list[str]] = mapped_column(JSONB)
+    toolchain: Mapped[dict] = mapped_column(JSONB)
+    compile_limits: Mapped[dict] = mapped_column(JSONB)
+    run_limits: Mapped[dict] = mapped_column(JSONB)
+    job_wall_time_us: Mapped[int] = mapped_column(BigInteger)
+    comparison: Mapped[str] = mapped_column(String(16), server_default="exact_bytes")
+    public_samples: Mapped[list[dict]] = mapped_column(JSONB)
+    published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class ProblemCase(Base):
+    __tablename__ = "problem_cases"
+    __table_args__ = (
+        UniqueConstraint("id", "version_id"),
+        UniqueConstraint("version_id", "position"),
+        CheckConstraint("position BETWEEN 0 AND 255", name="position"),
+        ForeignKeyConstraint(
+            ["input_digest", "artifact_state"], ["artifacts.digest", "artifacts.state"]
+        ),
+        ForeignKeyConstraint(
+            ["expected_digest", "artifact_state"],
+            ["artifacts.digest", "artifacts.state"],
+            name="fk_problem_cases_expected_digest_artifacts",
+        ),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    version_id: Mapped[UUID] = mapped_column(ForeignKey("problem_versions.id"))
+    position: Mapped[int]
+    input_digest: Mapped[str] = mapped_column(String(64))
+    expected_digest: Mapped[str] = mapped_column(String(64))
+    artifact_state: Mapped[str] = mapped_column(String(8), Computed("'ready'", persisted=True))

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

@@ -0,0 +1,163 @@
+"""Platform creates initial rows; Coordinator exclusively advances execution."""
+
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    CheckConstraint,
+    Computed,
+    DateTime,
+    ForeignKey,
+    ForeignKeyConstraint,
+    Index,
+    String,
+    UniqueConstraint,
+    func,
+)
+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 Submission(CreatedAt, Base):
+    __tablename__ = "submissions"
+    __table_args__ = (
+        UniqueConstraint("account_id", "idempotency_key"),
+        UniqueConstraint("id", "problem_version_id"),
+        CheckConstraint("language IN ('c17', 'cpp20')", name="language"),
+        CheckConstraint("request_digest ~ '^[0-9a-f]{64}$'", name="request_digest"),
+        ForeignKeyConstraint(
+            ["problem_version_id", "problem_id"],
+            ["problem_versions.id", "problem_versions.problem_id"],
+        ),
+        ForeignKeyConstraint(
+            ["source_digest", "artifact_state"], ["artifacts.digest", "artifacts.state"]
+        ),
+        Index("ix_submissions_account_history", "account_id", "created_at", "id"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    account_id: Mapped[UUID] = mapped_column(ForeignKey("accounts.id"))
+    problem_id: Mapped[UUID]
+    problem_version_id: Mapped[UUID]
+    language: Mapped[str] = mapped_column(String(8))
+    source_digest: Mapped[str] = mapped_column(String(64))
+    artifact_state: Mapped[str] = mapped_column(String(8), Computed("'ready'", persisted=True))
+    idempotency_key: Mapped[UUID]
+    request_digest: Mapped[str] = mapped_column(String(64))
+
+
+class Evaluation(CreatedAt, Base):
+    __tablename__ = "evaluations"
+    __table_args__ = (
+        UniqueConstraint("id", "problem_version_id"),
+        ForeignKeyConstraint(
+            ["submission_id", "problem_version_id"],
+            ["submissions.id", "submissions.problem_version_id"],
+        ),
+        ForeignKeyConstraint(
+            ["current_attempt_id", "id"],
+            ["judge_attempts.id", "judge_attempts.evaluation_id"],
+            name="fk_evaluations_current_attempt",
+            use_alter=True,
+            deferrable=True,
+            initially="DEFERRED",
+        ),
+        CheckConstraint(
+            "state IN ('queued','claimed','compiling','running','cancelling',"
+            "'finished','cancelled','failed')",
+            name="state",
+        ),
+        CheckConstraint("fencing_version >= 0", name="fence"),
+        CheckConstraint("max_attempts BETWEEN 1 AND 5", name="max_attempts"),
+        CheckConstraint("attempts_count BETWEEN 0 AND max_attempts", name="attempts_count"),
+        CheckConstraint(
+            "state NOT IN ('claimed','compiling','running','cancelling') "
+            "OR current_attempt_id IS NOT NULL",
+            name="active_attempt",
+        ),
+        Index("ix_evaluations_recovery", "state", "not_before", "id"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    submission_id: Mapped[UUID] = mapped_column(unique=True)
+    problem_version_id: Mapped[UUID]
+    state: Mapped[str] = mapped_column(String(16), server_default="queued")
+    fencing_version: Mapped[int] = mapped_column(BigInteger, server_default="0")
+    attempts_count: Mapped[int] = mapped_column(server_default="0")
+    max_attempts: Mapped[int] = mapped_column(server_default="3")
+    current_attempt_id: Mapped[UUID | None]
+    not_before: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    cancel_requested_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    verdict: Mapped[str | None] = mapped_column(String(32))
+    failure_code: Mapped[str | None] = mapped_column(String(32))
+    final_report: Mapped[dict | None] = mapped_column(JSONB)
+    final_report_digest: Mapped[str | None] = mapped_column(String(64))
+
+
+class JudgeAttempt(CreatedAt, Base):
+    __tablename__ = "judge_attempts"
+    __table_args__ = (
+        UniqueConstraint("id", "evaluation_id"),
+        UniqueConstraint("evaluation_id", "number", name="uq_judge_attempts_evaluation_number"),
+        UniqueConstraint(
+            "evaluation_id", "fencing_version", name="uq_judge_attempts_evaluation_fence"
+        ),
+        CheckConstraint("number BETWEEN 1 AND 5", name="number"),
+        CheckConstraint("fencing_version > 0", name="fence"),
+        CheckConstraint("lease_token_hash ~ '^[0-9a-f]{64}$'", name="token_digest"),
+        CheckConstraint("manifest_digest ~ '^[0-9a-f]{64}$'", name="manifest_digest"),
+        CheckConstraint("sequence >= 0", name="sequence"),
+        Index("ix_judge_attempts_expiry", "expires_at", "evaluation_id"),
+    )
+
+    id: Mapped[UUID] = mapped_column(primary_key=True)
+    evaluation_id: Mapped[UUID] = mapped_column(ForeignKey("evaluations.id"))
+    worker_id: Mapped[UUID]
+    number: Mapped[int]
+    fencing_version: Mapped[int] = mapped_column(BigInteger)
+    lease_token_hash: Mapped[str] = mapped_column(String(64))
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    manifest: Mapped[dict] = mapped_column(JSONB)
+    manifest_digest: Mapped[str] = mapped_column(String(64))
+    sequence: Mapped[int] = mapped_column(BigInteger, server_default="0")
+    stage: Mapped[str | None] = mapped_column(String(16))
+    case_index: Mapped[int | None]
+    report: Mapped[dict | None] = mapped_column(JSONB)
+    report_digest: Mapped[str | None] = mapped_column(String(64))
+    cleanup_complete: Mapped[bool | None] = mapped_column(Boolean)
+
+
+class CaseResult(Base):
+    __tablename__ = "case_results"
+    __table_args__ = (
+        UniqueConstraint("attempt_id", "position"),
+        ForeignKeyConstraint(
+            ["case_id", "problem_version_id"], ["problem_cases.id", "problem_cases.version_id"]
+        ),
+        ForeignKeyConstraint(
+            ["evaluation_id", "problem_version_id"],
+            ["evaluations.id", "evaluations.problem_version_id"],
+        ),
+        ForeignKeyConstraint(
+            ["attempt_id", "evaluation_id"], ["judge_attempts.id", "judge_attempts.evaluation_id"]
+        ),
+        CheckConstraint("position BETWEEN 0 AND 255", name="position"),
+        CheckConstraint("cpu_time_us IS NULL OR cpu_time_us >= 0", name="cpu"),
+        CheckConstraint("memory_peak_bytes IS NULL OR memory_peak_bytes >= 0", name="memory"),
+    )
+
+    attempt_id: Mapped[UUID] = mapped_column(primary_key=True)
+    case_id: Mapped[UUID] = mapped_column(primary_key=True)
+    evaluation_id: Mapped[UUID]
+    problem_version_id: Mapped[UUID]
+    position: Mapped[int]
+    verdict: Mapped[str] = mapped_column(String(32))
+    cpu_time_us: Mapped[int | None] = mapped_column(BigInteger)
+    memory_peak_bytes: Mapped[int | None] = mapped_column(BigInteger)

backend/tests/conftest.py

@@ -0,0 +1,53 @@
+"""Optional real PostgreSQL acceptance; never reuse or reset an existing business DB."""
+
+import asyncio
+import os
+import subprocess
+from pathlib import Path
+from uuid import uuid4
+
+import asyncpg
+import pytest
+from sqlalchemy.engine import make_url
+
+
+@pytest.fixture
+def postgres_url():
+    admin = os.environ.get("KF_OJ_TEST_POSTGRES")
+    if admin is None:
+        pytest.skip("real PostgreSQL acceptance requires KF_OJ_TEST_POSTGRES")
+    url = make_url(admin)
+    database = "kf_oj_test_" + uuid4().hex
+    assert url.drivername == "postgresql+asyncpg"
+    test_url = url.set(database=database).render_as_string(hide_password=False)
+
+    async def manage(command):
+        connection = await asyncpg.connect(
+            url.set(drivername="postgresql").render_as_string(hide_password=False)
+        )
+        try:
+            await connection.execute(command)
+        finally:
+            await connection.close()
+
+    asyncio.run(manage(f'CREATE DATABASE "{database}"'))
+    try:
+        migrate(test_url, "upgrade", "head")
+        yield test_url
+    finally:
+        asyncio.run(manage(f'DROP DATABASE "{database}"'))
+
+
+def migrate(url: str, command: str, target: str | None = None) -> None:
+    subprocess.run(
+        [str(Path(__file__).resolve().parents[1] / ".venv/bin/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,
+    )
+
+
+async def connect(url: str):
+    return await asyncpg.connect(url.replace("postgresql+asyncpg:", "postgresql:", 1))

backend/tests/test_database_foundation.py

@@ -0,0 +1,262 @@
+import asyncio
+import json
+from uuid import uuid4
+
+import asyncpg
+import pytest
+from conftest import connect, migrate
+
+
+def test_migration_roundtrip_and_role_safety(postgres_url):
+    async def inspect():
+        c = await connect(postgres_url)
+        try:
+            tables = set(await c.fetch("SELECT tablename FROM pg_tables WHERE schemaname='public'"))
+            assert len(tables) == 13  # Twelve business 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", "kf_oj_coordinator"}
+            assert all(
+                not any(
+                    r[k]
+                    for k in (
+                        "rolsuper",
+                        "rolcanlogin",
+                        "rolcreatedb",
+                        "rolcreaterole",
+                        "rolbypassrls",
+                    )
+                )
+                for r in roles
+            )
+            await c.execute("SET ROLE kf_oj_platform")
+            with pytest.raises(asyncpg.InsufficientPrivilegeError):
+                await c.execute("CREATE TABLE forbidden_table (x int)")
+            await c.execute("RESET ROLE")
+        finally:
+            await c.close()
+
+    asyncio.run(inspect())
+    migrate(postgres_url, "downgrade", "base")
+    migrate(postgres_url, "upgrade", "head")
+    migrate(postgres_url, "check")
+    asyncio.run(inspect())
+
+
+async def fixture_rows(c):
+    account, problem, version, case, submission, evaluation = (uuid4() for _ in range(6))
+    await c.execute(
+        "INSERT INTO accounts (id,username,password_hash) VALUES ($1,'fixture_user',$2)",
+        account,
+        "synthetic-irreversible-placeholder-hash-" * 2,
+    )
+    await c.execute("INSERT INTO artifacts (digest,bytes) VALUES ($1,3),($2,0)", "a" * 64, "b" * 64)
+    await c.execute("INSERT INTO problems (id,slug) VALUES ($1,'fixture-problem')", problem)
+    await c.execute(
+        """INSERT INTO problem_versions
+        (id,problem_id,revision,title,statement,languages,toolchain,compile_limits,run_limits,
+         job_wall_time_us,public_samples)
+        VALUES ($1,$2,1,'Fixture','Public statement','["c17"]','{}','{}','{}',10000,'[]')""",
+        version,
+        problem,
+    )
+    await c.execute(
+        """INSERT INTO problem_cases
+        (id,version_id,position,input_digest,expected_digest) VALUES ($1,$2,0,$3,$4)""",
+        case,
+        version,
+        "b" * 64,
+        "a" * 64,
+    )
+    await c.execute("UPDATE problem_versions SET published_at=now() WHERE id=$1", version)
+    await c.execute("UPDATE problems SET active_version_id=$1 WHERE id=$2", version, problem)
+    await c.execute(
+        """INSERT INTO submissions
+        (id,account_id,problem_id,problem_version_id,language,source_digest,idempotency_key,
+        request_digest) VALUES ($1,$2,$3,$4,'c17',$5,$6,$7)""",
+        submission,
+        account,
+        problem,
+        version,
+        "a" * 64,
+        uuid4(),
+        "c" * 64,
+    )
+    await c.execute(
+        "INSERT INTO evaluations (id,submission_id,problem_version_id) VALUES ($1,$2,$3)",
+        evaluation,
+        submission,
+        version,
+    )
+    await c.execute("INSERT INTO outbox (id,evaluation_id) VALUES ($1,$2)", uuid4(), evaluation)
+    return account, problem, version, case, submission, evaluation
+
+
+def test_database_write_authorities_and_immutable_versions(postgres_url):
+    async def run():
+        c = await connect(postgres_url)
+        try:
+            await c.execute("SET ROLE kf_oj_platform")
+            ids = await fixture_rows(c)
+            assert await c.fetchval("SELECT state FROM evaluations WHERE id=$1", ids[5]) == "queued"
+            for command in (
+                "UPDATE evaluations SET state='finished'",
+                "INSERT INTO evaluations (id,submission_id,problem_version_id,state) "
+                "SELECT gen_random_uuid(),submission_id,problem_version_id,'finished' "
+                "FROM evaluations",
+                "DELETE FROM evaluations",
+                "UPDATE problem_versions SET title='altered'",
+                "INSERT INTO judge_attempts (id) VALUES (gen_random_uuid())",
+                "UPDATE artifacts SET bytes=42",
+            ):
+                with pytest.raises(asyncpg.InsufficientPrivilegeError):
+                    await c.execute(command)
+            with pytest.raises(asyncpg.CheckViolationError):
+                await c.execute("UPDATE problem_versions SET published_at=now()")
+            with pytest.raises(asyncpg.CheckViolationError):
+                await c.execute(
+                    """INSERT INTO problem_cases
+                    (id,version_id,position,input_digest,expected_digest)
+                    VALUES ($1,$2,1,$3,$3)""",
+                    uuid4(),
+                    ids[2],
+                    "a" * 64,
+                )
+            with pytest.raises(asyncpg.ForeignKeyViolationError):
+                await c.execute("UPDATE artifacts SET state='deleting' WHERE digest=$1", "a" * 64)
+            await c.execute("RESET ROLE")
+            await c.execute("SET ROLE kf_oj_coordinator")
+            for command in (
+                "UPDATE accounts SET enabled=false",
+                "INSERT INTO problems (id,slug) VALUES (gen_random_uuid(),'forbidden')",
+                "UPDATE submissions SET language='cpp20'",
+            ):
+                with pytest.raises(asyncpg.InsufficientPrivilegeError):
+                    await c.execute(command)
+            attempt = uuid4()
+            async with c.transaction():
+                await c.execute(
+                    """INSERT INTO judge_attempts
+                    (id,evaluation_id,worker_id,number,fencing_version,lease_token_hash,
+                    expires_at,manifest,manifest_digest)
+                    VALUES ($1,$2,$3,1,1,$4,now()+interval '30 seconds',$5,$6)""",
+                    attempt,
+                    ids[5],
+                    uuid4(),
+                    "d" * 64,
+                    json.dumps({"fixture": True}),
+                    "e" * 64,
+                )
+                await c.execute(
+                    """UPDATE evaluations SET current_attempt_id=$1,state='claimed',
+                    attempts_count=1,fencing_version=1 WHERE id=$2""",
+                    attempt,
+                    ids[5],
+                )
+            assert await c.fetchval("SELECT state FROM evaluations") == "claimed"
+            with pytest.raises(asyncpg.InsufficientPrivilegeError):
+                await c.execute("UPDATE judge_attempts SET manifest='{}'")
+            # Active-attempt FK is deferred but still rejects an unrelated/absent attempt at commit.
+            with pytest.raises(asyncpg.ForeignKeyViolationError):
+                async with c.transaction():
+                    await c.execute("UPDATE evaluations SET current_attempt_id=$1", uuid4())
+        finally:
+            await c.close()
+
+    asyncio.run(run())
+
+
+def test_initial_creation_transaction_rolls_back_and_cross_version_fk_rejects(postgres_url):
+    async def run():
+        c = await connect(postgres_url)
+        try:
+            with pytest.raises(RuntimeError):
+                async with c.transaction():
+                    await c.execute("SET LOCAL ROLE kf_oj_platform")
+                    await fixture_rows(c)
+                    raise RuntimeError("simulated transaction failure")
+            assert await c.fetchval("SELECT count(*) FROM submissions") == 0
+            assert await c.fetchval("SELECT count(*) FROM outbox") == 0
+            ids = await fixture_rows(c)
+            with pytest.raises(asyncpg.ForeignKeyViolationError):
+                await c.execute(
+                    "UPDATE evaluations SET problem_version_id=$1 WHERE id=$2",
+                    uuid4(),
+                    ids[5],
+                )
+        finally:
+            await c.close()
+
+    asyncio.run(run())
+
+
+def test_publication_serializes_concurrent_case_insertion(postgres_url):
+    async def run():
+        c, other = await connect(postgres_url), await connect(postgres_url)
+        pending = None
+        transaction = None
+        try:
+            ids = await fixture_rows(c)
+            # A new draft version, separate from the immutable already published fixture.
+            draft = uuid4()
+            await c.execute(
+                """INSERT INTO problem_versions
+                (id,problem_id,revision,title,statement,languages,toolchain,compile_limits,
+                run_limits,job_wall_time_us,public_samples)
+                SELECT $1,problem_id,2,title,statement,languages,toolchain,compile_limits,
+                    run_limits,job_wall_time_us,public_samples FROM problem_versions WHERE id=$2
+                """,
+                draft,
+                ids[2],
+            )
+            await c.execute(
+                """INSERT INTO problem_cases
+                (id,version_id,position,input_digest,expected_digest)
+                VALUES ($1,$2,0,$3,$3)""",
+                uuid4(),
+                draft,
+                "a" * 64,
+            )
+            transaction = c.transaction()
+            await transaction.start()
+            await c.execute("UPDATE problem_versions SET published_at=now() WHERE id=$1", draft)
+            pending = asyncio.create_task(
+                other.execute(
+                    """INSERT INTO problem_cases
+                (id,version_id,position,input_digest,expected_digest)
+                VALUES ($1,$2,1,$3,$3)""",
+                    uuid4(),
+                    draft,
+                    "a" * 64,
+                )
+            )
+            await asyncio.sleep(0.1)
+            assert not pending.done()  # Actual row lock, not a simulated repository response.
+            await transaction.commit()
+            transaction = None
+            with pytest.raises(asyncpg.CheckViolationError):
+                await asyncio.wait_for(pending, timeout=5)
+            assert (
+                await c.fetchval("SELECT count(*) FROM problem_cases WHERE version_id=$1", draft)
+                == 1
+            )
+            with pytest.raises(asyncpg.CheckViolationError):
+                await c.execute(
+                    """INSERT INTO problem_versions
+                    (id,problem_id,revision,title,statement,languages,toolchain,compile_limits,
+                    run_limits,job_wall_time_us,public_samples,published_at)
+                    SELECT $1,problem_id,3,title,statement,languages,toolchain,compile_limits,
+                        run_limits,job_wall_time_us,public_samples,now()
+                    FROM problem_versions WHERE id=$2""",
+                    uuid4(),
+                    draft,
+                )
+        finally:
+            if transaction:
+                await transaction.rollback()
+            if pending and not pending.done():
+                pending.cancel()
+                await asyncio.gather(pending, return_exceptions=True)
+            await c.close()
+            await other.close()
+
+    asyncio.run(run())