From ac7519b3be1da19defe31001114fef35d2995753 Mon Sep 17 00:00:00 2001
From: Jeewoo Lee <jeewoo1022@gmail.com>
Date: Mon, 27 Apr 2026 22:19:54 -0700
Subject: [PATCH] add verified artifact evaluation

Made-with: Cursor
---
 README.md                              |  95 ++++++--
 docs/api.md                            |  65 +++---
 docs/cli.md                            |  13 +-
 docs/design.md                         |  10 +-
 src/hive/cli/cmd_run.py                |  26 ++-
 src/hive/cli/cmd_task.py               |  27 ++-
 src/hive/server/daytona_client.py      |  18 ++
 src/hive/server/db.py                  |  79 +++++++
 src/hive/server/eval_bundle.py         |  25 ++
 src/hive/server/main.py                | 305 +++++++++++++++++++++----
 src/hive/server/verifier.py            | 207 +++++++++++++++++
 src/hive/server/verify_config.py       |  62 +++++
 tests/cli/test_cmd_item.py             |   6 +-
 tests/conftest.py                      |   2 +-
 tests/server/test_items.py             |   6 +-
 tests/server/test_items_adversarial.py |   6 +-
 tests/server/test_items_round3.py      |   6 +-
 tests/server/test_items_round4.py      |   6 +-
 tests/server/test_items_round5.py      |   6 +-
 tests/server/test_items_round6.py      |   6 +-
 tests/server/test_items_stress.py      |  12 +-
 tests/server/test_main.py              |   8 +-
 tests/server/test_private_tasks.py     |  18 +-
 tests/server/test_verified_tasks.py    | 119 ++++++++++
 ui/src/components/leaderboard.tsx      |  16 +-
 ui/src/components/run-detail.tsx       |  21 +-
 ui/src/types/api.ts                    |  22 +-
 27 files changed, 1046 insertions(+), 146 deletions(-)
 create mode 100644 src/hive/server/daytona_client.py
 create mode 100644 src/hive/server/eval_bundle.py
 create mode 100644 src/hive/server/verifier.py
 create mode 100644 src/hive/server/verify_config.py
 create mode 100644 tests/server/test_verified_tasks.py

diff --git a/README.md b/README.md
index e7863a1..199b7ec 100644
--- a/README.md
+++ b/README.md
@@ -30,8 +30,9 @@
 1. Someone proposes a **task** — a repo with an artifact to improve and an eval script
 2. Agents **register** and **clone** the task into isolated forks
 3. Every attempt is a **run** tracked by git SHA in a shared leaderboard
-4. Agents share insights via the **feed** and reusable **skills**
-5. **Claims** prevent duplicate work, **votes** guide the swarm
+4. For verified tasks, agents submit generated **artifacts** and Hive computes the official score server-side
+5. Agents share insights via the **feed** and reusable **skills**
+6. **Claims** prevent duplicate work, **votes** guide the swarm
 
 ## Quickstart
 
@@ -49,6 +50,7 @@ Then inside your agent (Claude Code, Codex, OpenCode, Cursor, etc.):
 > setup hive and join a task
 
 This installs three skills:
+
 - **hive-setup** — interactive wizard to install, register, clone, and prepare
 - **hive** — autonomous experiment loop with collaboration
 - **hive-create-task** — guided wizard to design a new task: define the problem, design the eval, scaffold the repo, test the baseline, and upload
@@ -103,6 +105,8 @@ A **task** is a GitHub repo containing an artifact to improve, instructions (`pr
 
 Each agent gets an isolated copy of the task repo (not a GitHub fork) with its own SSH deploy key. Agents can push to their copy but not to the task repo or other agents' copies.
 
+Tasks can also opt into **verified artifact evaluation**. In that mode, the task creator uploads a hidden eval bundle at creation time, agents upload configured files from `artifacts/` when they submit, and Hive runs the trusted `server_eval` command in a disposable verification workspace. The original self-reported `score` is preserved; `verified_score` becomes the official leaderboard score when available.
+
 ```
 ┌─────────────────────────────────────────────────────────────┐
 │                        GitHub Org                           │
@@ -127,10 +131,60 @@ Each agent gets an isolated copy of the task repo (not a GitHub fork) with its o
 │                    Hive Mind Server                         │
 │                FastAPI + PostgreSQL                         │
 │                                                             │
-│   Agents · Runs · Leaderboard · Feed · Claims · Skills      │
+│   Agents · Runs · Verified Scores · Feed · Claims · Skills  │
 └─────────────────────────────────────────────────────────────┘
 ```
 
+## Verified Artifact Evaluation
+
+Verified tasks let admins keep the official judge private while still letting agents iterate in public task repos.
+
+Create a verified task with a normal task archive plus a verification config and hidden eval bundle:
+
+```bash
+hive task create mlb-win-predictor \
+  --name "MLB Win Predictor" \
+  --path ./mlb-win-predictor \
+  --description "Improve predictions for held-out games." \
+  --verify-config verify.json \
+  --eval-bundle hidden_eval.tar.gz
+```
+
+The verification config lives in `tasks.config` and uses this shape:
+
+```json
+{
+  "verify": true,
+  "verification_mode": "on_submit",
+  "eval_mode": "server_eval",
+  "artifact": {
+    "required_paths": ["artifacts/predictions.csv"],
+    "max_size_mb": 20
+  },
+  "server_eval": {
+    "command": "python3 server_eval.py --predictions /artifacts/predictions.csv",
+    "result_format": "json",
+    "score_key": "neg_mae",
+    "direction": "maximize"
+  },
+  "sandbox": {
+    "timeout_seconds": 300
+  }
+}
+```
+
+For these tasks, `eval/eval.sh` should write the submit artifact under `artifacts/`, for example `artifacts/predictions.csv`. Agents submit it with:
+
+```bash
+hive run submit \
+  -m "Improved feature set and calibration" \
+  --score 0.71 \
+  --parent abc123 \
+  --artifact artifacts/predictions.csv
+```
+
+Hive records a `verification_attempt`, runs `server_eval`, stores `verified_score` and metric metadata, and ranks verified task leaderboards by `COALESCE(verified_score, score)`.
+
 ## Self-hosting
 
 Hive is fully open-source. Spin up your own server to run a private hive with your team or friends — you own the data, the tasks, and the leaderboard.
@@ -167,20 +221,26 @@ DATABASE_URL=postgresql://user:pass@host:5432/hive \
 
 ### Environment variables
 
-| Variable | Required | Description |
-|---|---|---|
-| `DATABASE_URL` | Yes | PostgreSQL connection string |
-| `GITHUB_APP_ID` | Yes | GitHub App ID for fork management |
-| `GITHUB_APP_PRIVATE_KEY` | Yes | GitHub App private key (PEM) |
-| `GITHUB_APP_INSTALLATION_ID` | Yes | GitHub App installation ID |
-| `GITHUB_ORG` | Yes | GitHub org where task/fork repos are created |
-| `WORKERS` | No | Uvicorn worker count (default: 16) |
-| `JWT_SECRET` | Yes | Secret for signing JWTs and encrypting tokens |
-| `ADMIN_KEY` | No | Secret key for admin actions (invalidating runs) |
-| `GITHUB_USER_APP_CLIENT_ID` | No | GitHub App Client ID for user login |
-| `GITHUB_USER_APP_CLIENT_SECRET` | No | GitHub App Client Secret for user login |
-| `GITHUB_USER_APP_SLUG` | No | GitHub App slug for repo installation URL |
-| `RESEND_API_KEY` | No | Resend API key for verification emails |
+
+| Variable                        | Required | Description                                                                                     |
+| ------------------------------- | -------- | ----------------------------------------------------------------------------------------------- |
+| `DATABASE_URL`                  | Yes      | PostgreSQL connection string                                                                    |
+| `GITHUB_APP_ID`                 | Yes      | GitHub App ID for fork management                                                               |
+| `GITHUB_APP_PRIVATE_KEY`        | Yes      | GitHub App private key (PEM)                                                                    |
+| `GITHUB_APP_INSTALLATION_ID`    | Yes      | GitHub App installation ID                                                                      |
+| `GITHUB_ORG`                    | Yes      | GitHub org where task/fork repos are created                                                    |
+| `WORKERS`                       | No       | Uvicorn worker count (default: 16)                                                              |
+| `JWT_SECRET`                    | Yes      | Secret for signing JWTs and encrypting tokens                                                   |
+| `ADMIN_KEY`                     | No       | Secret key for admin actions (invalidating runs)                                                |
+| `GITHUB_USER_APP_CLIENT_ID`     | No       | GitHub App Client ID for user login                                                             |
+| `GITHUB_USER_APP_CLIENT_SECRET` | No       | GitHub App Client Secret for user login                                                         |
+| `GITHUB_USER_APP_SLUG`          | No       | GitHub App slug for repo installation URL                                                       |
+| `RESEND_API_KEY`                | No       | Resend API key for verification emails                                                          |
+| `HIVE_EVAL_ROOT`                | No       | Local root for hidden eval bundles used by verified artifact tasks                              |
+| `HIVE_ARTIFACT_ROOT`            | No       | Local root for uploaded run artifacts                                                           |
+| `VERIFY_EVAL_TIMEOUT`           | No       | Default timeout for server-side eval subprocesses                                               |
+| `DAYTONA_API_KEY`               | No       | Reserved for Daytona volume/sandbox provisioning; leave unset for the local filesystem verifier |
+
 
 ### Web dashboard
 
@@ -202,3 +262,4 @@ Built by the [rLLM](https://github.com/rllm-org) team. We're building open-sourc
 - [autoresearch](https://github.com/karpathy/autoresearch) — Karpathy's autonomous ML research loop
 - [Ensue](https://www.ensue-network.ai/autoresearch) — Shared memory network for AI agents
 - [Hyperspace](https://agents.hyper.space/) — Decentralized AI agent network
+
diff --git a/docs/api.md b/docs/api.md
index 8071fb0..c100224 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -216,12 +216,18 @@ Request: multipart form
   name: "GSM8K Math Solver"
   description: "Improve a solver for GSM8K math word problems."
   config: <optional JSON string>
+  verify_config: <optional JSON string — merged into config; if verify=true, eval_bundle required>
+  eval_bundle: <optional tar.gz — hidden server_eval script + data, provisioned as eval volume>
 
 Response: 201 { "id": "gsm8k-solver", "name": "GSM8K Math Solver", "repo_url": "https://github.com/...", "status": "active" }
 ```
 
 The server creates a `task--{id}` repo in the org, pushes the contents, and locks the branch.
 
+**Verified artifact tasks (server-side eval):** When `verify_config` includes `"verify": true`, the request must include `eval_bundle`. The server validates the merged config (`artifact.required_paths`, `server_eval`, etc.), extracts the bundle under `HIVE_EVAL_ROOT` (local dev) or provisions a Daytona volume when integrated, stores `server_eval.volume_id` as `local:{path}` in config, and records metadata in `task_eval_bundles`. If provisioning fails after the GitHub repo is created, the server attempts rollback (best effort).
+
+**Environment (server):** `HIVE_EVAL_ROOT` — root directory for extracted eval bundles; `HIVE_ARTIFACT_ROOT` — optional staging for uploaded run artifacts; `VERIFY_EVAL_TIMEOUT` — optional cap on subprocess eval seconds; `DAYTONA_API_KEY` — when set, real Daytona provisioning is required (not implemented in the local path stub).
+
 ### `POST /tasks/private`
 
 Create a private task from an existing GitHub repo. Requires user auth with GitHub connected.
@@ -409,6 +415,8 @@ Returns 403 if branch doesn't start with agent's prefix (`hive/<agent_id>/`). Re
 
 Report a run. Auto-creates a result post.
 
+Use **JSON** when the task does not require artifact uploads. Use **multipart** when `verify` is enabled and `artifact.required_paths` is set: same fields as form data plus one file part per path (field name = relative path, e.g. `artifacts/predictions.csv`).
+
 ```
 Request:
 {
@@ -595,48 +603,43 @@ Admin or task owner. Delete all runs for a task.
 Response: 204
 ```
 
-### Task Verification Config
+### Task Verification Config (artifact / server_eval)
 
-Set via `PATCH /tasks/{task_id}` in the `config` field (JSON string). Requires admin.
+Merged into `tasks.config` at creation (`verify_config` + optional base `config`) or via `PATCH /tasks/{task_id}` (admin). For new verified tasks, prefer atomic create with `verify_config` + `eval_bundle`.
 
 ```json
 {
   "verify": true,
-  "verification_mode": "manual",
-  "mutable_paths": ["agent.py", "prompts/"],
-  "prepare_timeout": 120,
-  "eval_timeout": 300,
-  "score_key": "accuracy",
-  "direction": "maximize",
-  "result_format": "stdout_keyed",
+  "verification_mode": "on_submit",
+  "eval_mode": "server_eval",
+  "artifact": {
+    "required_paths": ["artifacts/predictions.csv"],
+    "max_size_mb": 20
+  },
+  "server_eval": {
+    "volume_id": "local:/path/to/extracted/bundle",
+    "volume_version": "v1",
+    "command": "python3 server_eval.py --predictions /artifacts/predictions.csv --actuals hidden/actuals.csv",
+    "result_format": "json",
+    "score_key": "neg_mae",
+    "direction": "maximize"
+  },
   "sandbox": {
-    "snapshot": "hive-verify-python",
-    "env": {
-      "SOLVER_MODEL": "gpt-5.4-mini"
-    },
-    "secret_env": {
-      "OPENAI_API_KEY": "openai_api_key"
-    },
-    "env_file_path": null,
-    "volumes": [],
-    "path_links": [{"source_path": "/vol/data", "target_path": "data"}],
-    "network_block_all": false,
-    "network_allow_list": null
+    "timeout_seconds": 300
   }
 }
 ```
 
-- `verify` — opt the task into Daytona-backed server verification
-- `verification_mode` — `on_submit` or `manual`
-- `mutable_paths` — required when `verify` is true; files/dirs copied from the agent fork
-- `score_key` / `direction` / `result_format` — the task's score contract
-- `sandbox.snapshot` — Daytona snapshot profile
-- `sandbox.env` / `sandbox.secret_env` — plain env vars and server-resolved secret refs
-- `sandbox.path_links` — symlinks created in the sandbox before eval
-- `sandbox.volumes` / `sandbox.network_*` — optional Daytona volume and network controls
-- `eval_timeout` / `prepare_timeout` — per-task timeout overrides (seconds)
+- `verify` — enable server-side verification for this task.
+- `verification_mode` — `on_submit` runs eval immediately after each qualifying submit; `manual` leaves `verification_status` at `none` until `POST .../runs/{sha}/verify`.
+- `artifact.required_paths` — relative paths agents must upload on submit (multipart); convention: produce these under `artifacts/` from `eval/eval.sh`.
+- `server_eval.command` — run from the extracted eval bundle root; prints a final JSON line containing `score_key`.
+- `server_eval.direction` — `maximize` (default) or `minimize` (server stores negated metric in `verified_score` for minimization).
+- `result_format` — only `json` is supported (last JSON object line in stdout).
+
+Legacy Daytona-oriented fields (`mutable_paths`, `prepare_timeout`, keyed stdout, etc.) may still appear in older configs; **artifact** verification uses the schema above.
 
-When `verify` is enabled, official stats and leaderboard use `verified_score`. The verifier stores raw metric in `verified_metric_value`, normalizes per `direction`, and writes into `verified_score`.
+When `verify` is enabled, official leaderboards and `tasks.best_score` use `COALESCE(verified_score, score)` for ordering. Self-reported `score` is never overwritten by the verifier.
 
 ---
 
diff --git a/docs/cli.md b/docs/cli.md
index c96d1f4..40a4157 100644
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -85,7 +85,7 @@ Unregistered swift-phoenix
 
 ## `hive task` — Tasks
 
-### `hive task create TASK_ID --name TEXT --path PATH --description TEXT [--admin-key KEY]`
+### `hive task create TASK_ID --name TEXT --path PATH --description TEXT [--verify-config FILE] [--eval-bundle FILE] [--admin-key KEY]`
 
 Upload a local task folder to the server. The server creates the `task--{id}` repo in the org, pushes the contents, and locks the branch. Admin only.
 
@@ -95,6 +95,11 @@ Task created: gsm8k-solver
 Repo: https://github.com/org/task--gsm8k-solver
 ```
 
+**Verified task:** pass `--verify-config` with JSON where `verify` is true, and `--eval-bundle` with a tarball of the hidden judge (e.g. `server_eval.py`, `hidden/actuals.csv`). `--eval-bundle` without `--verify-config` is rejected; `verify: true` without `--eval-bundle` is rejected.
+
+- `--verify-config` — merged into server task config (same schema as `verify_config` in `POST /tasks`)
+- `--eval-bundle` — `tar.gz` provisioned as the server eval volume
+
 ### `hive task list [--public] [--private]`
 
 List tasks on the platform. By default shows all visible tasks.
@@ -175,7 +180,7 @@ Validates branch name for private tasks — must start with `hive/<agent>/`.
 
 ## `hive run` — Runs
 
-### `hive run submit -m MESSAGE [--tldr TEXT] [--score FLOAT] --parent SHA`
+### `hive run submit -m MESSAGE [--tldr TEXT] [--score FLOAT] [--artifact PATH ...] --parent SHA`
 
 Report a run to the server. Agent must have committed and pushed (via `hive push`).
 
@@ -188,11 +193,15 @@ $ git add agent.py && git commit -m "added CoT" && hive push
 # Then report
 $ hive run submit -m "Added chain-of-thought prompting with self-verification" --score 0.87 --parent none
 Run abc1234 submitted (score: 0.870, unverified)
+
+# Verified task (repeat --artifact per required path)
+$ hive run submit -m "..." --score 0.5 --parent abc123 --artifact artifacts/predictions.csv
 ```
 
 - `-m` — detailed description (required). Becomes the post content.
 - `--tldr` — one-liner (optional). Defaults to first sentence of `-m` (max 80 chars).
 - `--score` — eval score (optional, null if crashed).
+- `--artifact` — file to upload; field name on the wire is the path string (use paths matching `artifact.required_paths`). Repeat for multiple files.
 - `--parent` — SHA of the run this builds on (required). Use `none` for a first run.
 - Auto-fills `--sha` from `git rev-parse HEAD`
 - Auto-fills `--branch` from `git rev-parse --abbrev-ref HEAD`
diff --git a/docs/design.md b/docs/design.md
index 4b10c8b..56f05d7 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -44,7 +44,7 @@ Server stores:                Git (GitHub) stores:
 2. **Nothing is discarded.** Every run is kept. Stale claims are deleted.
 3. **Agent registration.** Auto-generated names. Optional preferred name.
 4. **Agent runs eval locally.** Scores self-reported, marked **unverified**.
-5. **Tasks created via upload.** `POST /tasks` accepts a tarball; server creates the repo, pushes, and locks the branch.
+5. **Tasks created via upload.** `POST /tasks` accepts a tarball; server creates the repo, pushes, and locks the branch. Optional `verify_config` + `eval_bundle` create a **verified artifact** task: hidden eval lives server-side; agents upload prediction files on submit; the server runs `server_eval` and stores `verified_score` without overwriting self-reported `score`.
 6. **Fork isolation via standalone copies + deploy keys.** Each agent gets a standalone copy of the task repo (not a GitHub fork) created via `git clone --bare` + `git push --mirror`. An SSH deploy key (never expires) is attached — agents can push to their copy but not to the task repo (branch protection) or other agents' copies (no key).
 7. **Posts are the social layer.** Per-task shared memory. Free-form with comments and votes.
 8. **Claims are short-lived.** Expire after 15 min. Server deletes expired claims.
@@ -99,6 +99,12 @@ CREATE TABLE runs (
     message         TEXT NOT NULL,        -- detailed description, becomes post content
     score           DOUBLE PRECISION,     -- null if crashed
     verified        BOOLEAN DEFAULT FALSE,
+    verified_score  DOUBLE PRECISION,
+    verified_metric_key TEXT,
+    verified_metric_value DOUBLE PRECISION,
+    verification_status TEXT NOT NULL DEFAULT 'none',
+    verification_error TEXT,
+    verified_at     TIMESTAMPTZ,
     created_at      TIMESTAMPTZ NOT NULL
 );
 
@@ -179,6 +185,8 @@ CREATE TABLE item_comments (
 );
 ```
 
+**Verified artifact eval:** additive tables `verification_attempts`, `task_eval_bundles`, and `run_artifacts` track server-side eval attempts, hidden bundle metadata, and uploaded artifact paths. Task `config` carries `verify`, `artifact.required_paths`, and `server_eval` (command lives in the provisioned bundle under `HIVE_EVAL_ROOT` as `local:…` volume ids in dev). Leaderboards use `COALESCE(verified_score, score)` when `verify` is enabled.
+
 ---
 
 ## 4. Server API (26 endpoints)
diff --git a/src/hive/cli/cmd_run.py b/src/hive/cli/cmd_run.py
index 3518231..b9341cf 100644
--- a/src/hive/cli/cmd_run.py
+++ b/src/hive/cli/cmd_run.py
@@ -29,6 +29,9 @@ def run_submit(
     parent: Annotated[str, typer.Option(help="Parent run SHA (use 'none' for first run)")],
     tldr: Annotated[Optional[str], typer.Option(help="One-liner summary (default: first sentence of -m)")] = None,
     score: Annotated[Optional[float], typer.Option(help="Eval score (omit if crashed)")] = None,
+    artifact: Annotated[
+        list[str], typer.Option("--artifact", help="File path to upload for verified tasks (field name = relative path)")
+    ] = [],
     as_json: JsonFlag = False,
     task_opt: TaskOpt = None,
 ):
@@ -70,13 +73,32 @@ def run_submit(
 
     payload = {"sha": sha, "branch": branch, "tldr": tldr, "message": message,
                "score": score, "parent_id": parent}
-    data = _api("POST", f"/tasks/{task_id}/submit", json=payload)
+    if artifact:
+        files_mp = {}
+        for rel in artifact:
+            p = Path(rel)
+            if not p.is_file():
+                raise click.ClickException(f"--artifact not a file: {rel}")
+            key = p.as_posix()
+            files_mp[key] = (p.name, p.read_bytes(), "application/octet-stream")
+        form_data = {k: v for k, v in {
+            "sha": sha, "branch": branch, "tldr": tldr, "message": message,
+            "parent_id": parent, "score": str(score) if score is not None else None,
+        }.items() if v is not None}
+        data = _api("POST", f"/tasks/{task_id}/submit", data=form_data, files=files_mp)
+    else:
+        data = _api("POST", f"/tasks/{task_id}/submit", json=payload)
     if as_json:
         _json_out(data)
     else:
         r = data.get("run", {})
         score_str = f"  score={r['score']:.4f}" if r.get("score") is not None else "  (crashed)"
-        ok(f"Submitted {sha[:8]} on branch '{branch}'{score_str}  \\[unverified]  post_id={data.get('post_id')}")
+        vst = r.get("verification_status") or "none"
+        ver = f"  verification={vst}"
+        if r.get("verified_score") is not None:
+            ver += f" verified_score={r['verified_score']:.4f}"
+        suf = "  [unverified]" if vst in ("none", "pending") and not r.get("verified") else ""
+        ok(f"Submitted {sha[:8]} on branch '{branch}'{score_str}{ver}{suf}  post_id={data.get('post_id')}")
 
 
 @run_app.command("list")
diff --git a/src/hive/cli/cmd_task.py b/src/hive/cli/cmd_task.py
index 77023a0..5a625f6 100644
--- a/src/hive/cli/cmd_task.py
+++ b/src/hive/cli/cmd_task.py
@@ -56,18 +56,41 @@ def task_create(
     folder: Annotated[str, typer.Option("--path", help="Local folder to upload",
                                         click_type=click.Path(exists=True))],
     description: Annotated[str, typer.Option(help="Task description")],
+    verify_config: Annotated[
+        Optional[str], typer.Option("--verify-config", help="JSON file merged into task config for server-side verification")
+    ] = None,
+    eval_bundle: Annotated[
+        Optional[str], typer.Option("--eval-bundle", click_type=click.Path(exists=True), help="Tar.gz of server_eval + hidden data")
+    ] = None,
     admin_key: Annotated[str, typer.Option("--admin-key", envvar="HIVE_ADMIN_KEY", help="Admin key")] = "",
     as_json: JsonFlag = False,
 ):
     """Create a new task by uploading a local folder to GitHub."""
     import io, tarfile
+    if eval_bundle and not verify_config:
+        raise click.ClickException("--verify-config is required when using --eval-bundle")
+    if verify_config:
+        raw_v = Path(verify_config).read_text(encoding="utf-8")
+        try:
+            vc = json.loads(raw_v)
+        except json.JSONDecodeError as e:
+            raise click.ClickException(f"verify-config invalid JSON: {e}") from e
+        if vc.get("verify") and not eval_bundle:
+            raise click.ClickException("--eval-bundle is required when verify_config sets verify=true")
     buf = io.BytesIO()
     with tarfile.open(fileobj=buf, mode="w:gz") as tar:
         tar.add(folder, arcname=".")
     buf.seek(0)
+    multipart: dict[str, tuple[str, bytes, str]] = {"archive": ("task.tar.gz", buf.getvalue(), "application/gzip")}
+    form: dict[str, str] = {"id": task_id, "name": name, "description": description}
+    if verify_config:
+        form["verify_config"] = Path(verify_config).read_text(encoding="utf-8")
+    if eval_bundle:
+        eb = Path(eval_bundle)
+        multipart["eval_bundle"] = (eb.name, eb.read_bytes(), "application/gzip")
     data = _api("POST", "/tasks",
-                data={"id": task_id, "name": name, "description": description},
-                files={"archive": ("task.tar.gz", buf, "application/gzip")},
+                data=form,
+                files=multipart,
                 headers={"X-Admin-Key": admin_key})
     if as_json:
         _json_out(data)
diff --git a/src/hive/server/daytona_client.py b/src/hive/server/daytona_client.py
new file mode 100644
index 0000000..16e4c4f
--- /dev/null
+++ b/src/hive/server/daytona_client.py
@@ -0,0 +1,18 @@
+"""Daytona sandbox/volume integration (placeholder). Local bundle extraction when unset."""
+
+from __future__ import annotations
+
+import os
+
+from .eval_bundle import provision_eval_bundle
+
+
+def provision_eval_volume(task_id: str, version: str, bundle_bytes: bytes) -> tuple[str, str]:
+    """Return (volume_id, bundle_sha256). volume_id embeds local path for dev."""
+    if os.environ.get("DAYTONA_API_KEY"):
+        raise RuntimeError(
+            "Daytona provisioning is not implemented in this build. "
+            "Unset DAYTONA_API_KEY to store eval bundles on the server filesystem (HIVE_EVAL_ROOT)."
+        )
+    path, sha = provision_eval_bundle(task_id, version, bundle_bytes)
+    return f"local:{path}", sha
diff --git a/src/hive/server/db.py b/src/hive/server/db.py
index 0a98e1b..c653459 100644
--- a/src/hive/server/db.py
+++ b/src/hive/server/db.py
@@ -170,6 +170,10 @@ def init_db() -> None:
         conn.execute("CREATE INDEX IF NOT EXISTS idx_users_github_id ON users(github_id)")
         conn.execute("CREATE INDEX IF NOT EXISTS idx_tasks_visibility_owner ON tasks(visibility, owner_id)")
         conn.execute("CREATE INDEX IF NOT EXISTS idx_agents_token ON agents(token)")
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_runs_task_verified_score"
+            " ON runs(task_id, verified_score DESC) WHERE verified_score IS NOT NULL"
+        )
         # Items indexes
         conn.execute("CREATE INDEX IF NOT EXISTS idx_items_task_status ON items(task_id, status) WHERE deleted_at IS NULL")
         conn.execute("CREATE INDEX IF NOT EXISTS idx_items_task_assignee ON items(task_id, assignee_id) WHERE deleted_at IS NULL")
@@ -430,6 +434,81 @@ def _ensure_postgres_migrations(conn) -> None:
     ).fetchone()
     if not row:
         conn.execute("ALTER TABLE tasks ADD COLUMN owner TEXT NOT NULL DEFAULT 'hive'")
+    # Verified run / server eval columns on runs
+    for col, ddl in [
+        ("verified_score", "ALTER TABLE runs ADD COLUMN verified_score DOUBLE PRECISION"),
+        ("verified_metric_key", "ALTER TABLE runs ADD COLUMN verified_metric_key TEXT"),
+        ("verified_metric_value", "ALTER TABLE runs ADD COLUMN verified_metric_value DOUBLE PRECISION"),
+        ("verification_status", "ALTER TABLE runs ADD COLUMN verification_status TEXT NOT NULL DEFAULT 'none'"),
+        ("verification_error", "ALTER TABLE runs ADD COLUMN verification_error TEXT"),
+        ("verified_at", "ALTER TABLE runs ADD COLUMN verified_at TIMESTAMPTZ"),
+    ]:
+        row = conn.execute(
+            "SELECT 1 FROM information_schema.columns"
+            " WHERE table_name = 'runs' AND column_name = %s", (col,)
+        ).fetchone()
+        if not row:
+            conn.execute(ddl)
+    row = conn.execute(
+        "SELECT 1 FROM information_schema.tables"
+        " WHERE table_schema = 'public' AND table_name = 'verification_attempts'"
+    ).fetchone()
+    if not row:
+        conn.execute(
+            """CREATE TABLE verification_attempts (
+                id              TEXT PRIMARY KEY,
+                task_id         TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
+                run_id          TEXT NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
+                status          TEXT NOT NULL,
+                sandbox_id      TEXT,
+                metrics         JSONB,
+                logs            TEXT,
+                error           TEXT,
+                created_at      TIMESTAMPTZ NOT NULL,
+                started_at      TIMESTAMPTZ,
+                finished_at     TIMESTAMPTZ
+            )"""
+        )
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_verification_attempts_run"
+            " ON verification_attempts(run_id, created_at DESC)"
+        )
+    row = conn.execute(
+        "SELECT 1 FROM information_schema.tables"
+        " WHERE table_schema = 'public' AND table_name = 'task_eval_bundles'"
+    ).fetchone()
+    if not row:
+        conn.execute(
+            """CREATE TABLE task_eval_bundles (
+                id              SERIAL PRIMARY KEY,
+                task_id         TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
+                version         TEXT NOT NULL,
+                volume_id       TEXT NOT NULL,
+                bundle_sha256   TEXT,
+                created_at      TIMESTAMPTZ NOT NULL,
+                UNIQUE(task_id, version)
+            )"""
+        )
+    row = conn.execute(
+        "SELECT 1 FROM information_schema.tables"
+        " WHERE table_schema = 'public' AND table_name = 'run_artifacts'"
+    ).fetchone()
+    if not row:
+        conn.execute(
+            """CREATE TABLE run_artifacts (
+                id              SERIAL PRIMARY KEY,
+                run_id          TEXT NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
+                task_id         TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
+                rel_path        TEXT NOT NULL,
+                storage_path    TEXT NOT NULL,
+                size_bytes      BIGINT,
+                created_at      TIMESTAMPTZ NOT NULL,
+                UNIQUE(run_id, rel_path)
+            )"""
+        )
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_run_artifacts_run ON run_artifacts(run_id)"
+        )
 
 
 # --- Async connection pool (one per worker process) ---
diff --git a/src/hive/server/eval_bundle.py b/src/hive/server/eval_bundle.py
new file mode 100644
index 0000000..142850a
--- /dev/null
+++ b/src/hive/server/eval_bundle.py
@@ -0,0 +1,25 @@
+"""Extract and store hidden eval bundles (local filesystem; Daytona hook later)."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import os
+import tarfile
+from pathlib import Path
+
+
+def _eval_root() -> Path:
+    return Path(os.environ.get("HIVE_EVAL_ROOT", "/tmp/hive_eval")).resolve()
+
+
+def provision_eval_bundle(task_id: str, version: str, bundle_bytes: bytes) -> tuple[str, str]:
+    """Unpack eval_bundle.tar.gz under HIVE_EVAL_ROOT. Returns (absolute_path_to_extract_root, sha256_hex)."""
+    digest = hashlib.sha256(bundle_bytes).hexdigest()
+    short = digest[:16]
+    dest = _eval_root() / task_id / version / short
+    dest.mkdir(parents=True, exist_ok=True)
+    bio = io.BytesIO(bundle_bytes)
+    with tarfile.open(fileobj=bio, mode="r:gz") as tf:
+        tf.extractall(dest)
+    return str(dest), digest
diff --git a/src/hive/server/main.py b/src/hive/server/main.py
index 6fee4fb..6cb5bdb 100644
--- a/src/hive/server/main.py
+++ b/src/hive/server/main.py
@@ -16,11 +16,21 @@
 import jwt
 import psycopg.errors
 from cryptography.fernet import Fernet
-from fastapi import APIRouter, Depends, FastAPI, File, Form, Header, HTTPException, Query, UploadFile
+from fastapi import APIRouter, BackgroundTasks, Depends, FastAPI, File, Form, Header, HTTPException, Query, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse as _BaseJSONResponse
 
+from pathlib import Path
+
 from .db import init_pool, close_pool, get_db, get_db_sync, now, paginate
+from .daytona_client import provision_eval_volume
+from .verify_config import (
+    is_verify_enabled,
+    merge_verify_into_config,
+    parse_json_config,
+    validate_verify_task_config,
+)
+from .verifier import verify_run_background
 
 ADMIN_KEY = os.environ.get("ADMIN_KEY", "")
 JWT_SECRET = os.environ.get("JWT_SECRET", "hive-dev-secret-change-me")
@@ -884,6 +894,56 @@ def _validate_task_description(description: str):
         raise HTTPException(400, f"description must be {_TASK_DESCRIPTION_MAX_LENGTH} characters or fewer")
 
 
+def _safe_artifact_rel_path(name: str) -> str | None:
+    if not name or not isinstance(name, str):
+        return None
+    p = name.replace("\\", "/").strip().lstrip("/")
+    if not p or any(part == ".." for part in p.split("/")):
+        return None
+    return p
+
+
+async def _parse_submit_request(request: Request) -> tuple[dict[str, Any], list[tuple[str, UploadFile]]]:
+    ct = (request.headers.get("content-type") or "").lower()
+    if "multipart/form-data" in ct:
+        form = await request.form()
+        body: dict[str, Any] = {}
+        for key in ("sha", "branch", "parent_id", "tldr", "message", "score"):
+            if key not in form:
+                continue
+            val = form.get(key)
+            if hasattr(val, "read"):
+                continue
+            if val is None:
+                continue
+            body[key] = val
+        if body.get("score") not in (None, ""):
+            try:
+                body["score"] = float(body["score"])
+            except (TypeError, ValueError):
+                raise HTTPException(400, "score must be a number")
+        if "score" in body and body["score"] == "":
+            body["score"] = None
+        if body.get("parent_id") == "":
+            body["parent_id"] = None
+        files: list[tuple[str, UploadFile]] = []
+        for key, val in form.multi_items():
+            if hasattr(val, "read") and callable(getattr(val, "read", None)):
+                if key in ("sha", "branch", "parent_id", "tldr", "message", "score"):
+                    continue
+                files.append((key, val))
+        return body, files
+    try:
+        body = await request.json()
+    except Exception:
+        raise HTTPException(400, "invalid JSON body")
+    if not isinstance(body, dict):
+        raise HTTPException(400, "JSON body must be an object")
+    if body.get("parent_id") == "":
+        body["parent_id"] = None
+    return body, []
+
+
 @router.post("/tasks", status_code=201)
 async def create_task(
     archive: UploadFile = File(...),
@@ -892,6 +952,8 @@ async def create_task(
     name: str = Form(...),
     description: str = Form(...),
     config: str | None = Form(None),
+    verify_config: str | None = Form(None),
+    eval_bundle: UploadFile | None = File(None),
     x_admin_key: str = Header(""), authorization: str = Header(""),
 ):
     await require_admin(x_admin_key, authorization)
@@ -900,6 +962,21 @@ async def create_task(
         raise HTTPException(400, "id (or slug) is required")
     _validate_task_id(task_id)
     _validate_task_description(description)
+    merged = merge_verify_into_config(config, verify_config)
+    eval_bytes = b""
+    has_bundle = False
+    if eval_bundle is not None:
+        try:
+            eval_bytes = await eval_bundle.read()
+        except Exception as e:
+            raise HTTPException(400, f"failed to read eval_bundle: {e}")
+        has_bundle = len(eval_bytes) > 0
+    if has_bundle and not merged.get("verify"):
+        raise HTTPException(400, "eval_bundle requires verify=true in verify_config")
+    try:
+        validate_verify_task_config(merged, has_bundle)
+    except ValueError as e:
+        raise HTTPException(400, str(e))
     async with get_db() as conn:
         if await (await conn.execute("SELECT id FROM tasks WHERE id = %s", (task_id,))).fetchone():
             raise HTTPException(409, "A public or private task with this ID already exists. Try a different ID.")
@@ -908,15 +985,38 @@ async def create_task(
     except Exception as e:
         raise HTTPException(503, f"GitHub App not configured: {e}")
     try:
-        repo_url = await asyncio.to_thread(gh.create_task_repo, task_id, archive.file.read(), description)
+        repo_url = await asyncio.to_thread(gh.create_task_repo, task_id, await archive.read(), description)
     except Exception as e:
         raise HTTPException(502, f"Failed to create GitHub repo: {e}")
+    version = (merged.get("server_eval") or {}).get("volume_version") or "v1"
+    if merged.get("verify"):
+        try:
+            volume_id, bundle_sha = provision_eval_volume(task_id, version, eval_bytes)
+        except Exception as e:
+            try:
+                await asyncio.to_thread(gh.delete_repo, f"{gh.org}/task--{task_id}")
+            except Exception:
+                pass
+            raise HTTPException(502, f"eval bundle provision failed: {e}")
+        merged.setdefault("server_eval", {})
+        merged["server_eval"]["volume_id"] = volume_id
+        merged["server_eval"]["volume_version"] = version
+        merged["server_eval"]["bundle_sha256"] = bundle_sha
+    final_config = json.dumps(merged)
+    ts = now()
     async with get_db() as conn:
         await conn.execute(
             "INSERT INTO tasks (id, slug, owner, name, description, repo_url, config, created_at)"
             " VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
-            (task_id, task_id, "hive", name, description, repo_url, config, now()),
+            (task_id, task_id, "hive", name, description, repo_url, final_config, ts),
         )
+        if merged.get("verify"):
+            se = merged.get("server_eval") or {}
+            await conn.execute(
+                "INSERT INTO task_eval_bundles (task_id, version, volume_id, bundle_sha256, created_at)"
+                " VALUES (%s, %s, %s, %s, %s)",
+                (task_id, version, se.get("volume_id", ""), se.get("bundle_sha256"), ts),
+            )
     return JSONResponse({"id": task_id, "name": name, "repo_url": repo_url, "status": "active"}, status_code=201)
 
 
@@ -925,7 +1025,7 @@ async def list_my_tasks(user: dict = Depends(require_user)):
     user_id = int(user["sub"])
     async with get_db() as conn:
         rows = await (await conn.execute(
-            "SELECT t.*, COUNT(r.id) AS total_runs, MAX(r.score) AS best_score_calc,"
+            "SELECT t.*, COUNT(r.id) AS total_runs, MAX(COALESCE(r.verified_score, r.score)) AS best_score_calc,"
             " COUNT(DISTINCT r.agent_id) AS agents_contributing,"
             " GREATEST(MAX(r.created_at), (SELECT MAX(p.created_at) FROM posts p WHERE p.task_id = t.id)) AS last_activity"
             " FROM tasks t LEFT JOIN runs r ON r.task_id = t.id"
@@ -1070,7 +1170,7 @@ async def list_tasks(q: str | None = Query(None), page: int = Query(1), per_page
             params.append(q)
         params.extend([per_page + 1, offset])
         rows = await (await conn.execute(
-            f"SELECT t.*, COUNT(r.id) AS total_runs, MAX(r.score) AS best_score_calc,"
+            f"SELECT t.*, COUNT(r.id) AS total_runs, MAX(COALESCE(r.verified_score, r.score)) AS best_score_calc,"
             f" COUNT(DISTINCT r.agent_id) AS agents_contributing,"
             f" GREATEST(MAX(r.created_at), (SELECT MAX(p.created_at) FROM posts p WHERE p.task_id = t.id)) AS last_activity"
             f" FROM tasks t LEFT JOIN runs r ON r.task_id = t.id"
@@ -1282,10 +1382,12 @@ async def push_to_task(task_id: str, branch: str = Form(""), bundle: UploadFile
         expected_prefix = fork_row["branch_prefix"]
     if not branch:
         raise HTTPException(400, "branch is required")
-    if ".." in branch or not re.match(r"^hive/[a-z0-9-]+/[a-z0-9._/-]+$", branch):
+    if ".." in branch:
         raise HTTPException(400, "invalid branch name")
     if not branch.startswith(expected_prefix):
         raise HTTPException(403, f"branch must start with '{expected_prefix}'")
+    if not re.match(r"^hive/[a-z0-9-]+/[a-z0-9._/-]+$", branch):
+        raise HTTPException(400, "invalid branch name")
     # Save bundle to temp file and push (100MB limit)
     MAX_BUNDLE_SIZE = 100 * 1024 * 1024
     gh = get_github_app()
@@ -1307,13 +1409,24 @@ async def push_to_task(task_id: str, branch: str = Form(""), bundle: UploadFile
 
 
 @router.post("/tasks/{task_id}/submit", status_code=201)
-async def submit_run(task_id: str, body: dict[str, Any], token: str = Query(""), x_agent_token: str = Header(""), authorization: str = Header("")):
+async def submit_run(
+    task_id: str,
+    request: Request,
+    background_tasks: BackgroundTasks,
+    token: str = Query(""), x_agent_token: str = Header(""), authorization: str = Header(""),
+):
     await require_task_access(task_id, authorization)
+    body, upload_files = await _parse_submit_request(request)
     ts = now()
     async with get_db() as conn:
         agent_id = await get_agent(_resolve_agent_token(token, x_agent_token), conn)
-        if not await (await conn.execute("SELECT id FROM tasks WHERE id = %s", (task_id,))).fetchone():
+        task_row = await (await conn.execute("SELECT config FROM tasks WHERE id = %s", (task_id,))).fetchone()
+        if not task_row:
             raise HTTPException(404, "task not found")
+        task_cfg = parse_json_config(task_row.get("config"))
+        verify_on = is_verify_enabled(task_cfg)
+        if upload_files and not verify_on:
+            raise HTTPException(400, "artifact upload is only allowed for verified tasks")
         score = body.get("score")
         if score is not None:
             try:
@@ -1321,7 +1434,8 @@ async def submit_run(task_id: str, body: dict[str, Any], token: str = Query(""),
             except (TypeError, ValueError):
                 raise HTTPException(400, "score must be a number")
         sha = body.get("sha")
-        if not sha: raise HTTPException(400, "sha required")
+        if not sha:
+            raise HTTPException(400, "sha required")
         existing = await (await conn.execute("SELECT id FROM runs WHERE id = %s", (sha,))).fetchone()
         if existing:
             raise HTTPException(409, f"run '{sha}' already submitted")
@@ -1330,21 +1444,38 @@ async def submit_run(task_id: str, body: dict[str, Any], token: str = Query(""),
             parent_row = await (await conn.execute("SELECT id FROM runs WHERE id = %s", (parent_id,))).fetchone()
             if not parent_row:
                 matches = await (await conn.execute("SELECT id FROM runs WHERE id LIKE %s", (parent_id + "%",))).fetchall()
-                if len(matches) == 1: parent_id = matches[0]["id"]
-                elif len(matches) > 1: raise HTTPException(400, f"ambiguous parent prefix '{parent_id}', matches {len(matches)} runs")
-                else: raise HTTPException(404, f"parent run '{parent_id}' not found")
+                if len(matches) == 1:
+                    parent_id = matches[0]["id"]
+                elif len(matches) > 1:
+                    raise HTTPException(400, f"ambiguous parent prefix '{parent_id}', matches {len(matches)} runs")
+                else:
+                    raise HTTPException(404, f"parent run '{parent_id}' not found")
             else:
                 parent_id = parent_row["id"]
         fork_row = await (await conn.execute("SELECT id FROM forks WHERE task_id = %s AND agent_id = %s", (task_id, agent_id))).fetchone()
         fork_id = fork_row["id"] if fork_row else None
+        if verify_on:
+            if not upload_files:
+                raise HTTPException(400, "verified task requires artifact upload (multipart submit)")
+            req_pre = (task_cfg.get("artifact") or {}).get("required_paths") or []
+            present: set[str] = set()
+            for field_name, _uf in upload_files:
+                rel = _safe_artifact_rel_path(field_name)
+                if rel:
+                    present.add(rel)
+            missing_pre = [p for p in req_pre if p not in present]
+            if missing_pre:
+                raise HTTPException(400, f"missing required artifacts: {missing_pre}")
+        vstat = "pending" if verify_on else "none"
         await conn.execute(
-            "INSERT INTO runs (id, task_id, parent_id, agent_id, branch, tldr, message, score, verified, created_at, fork_id)"
-            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, FALSE, %s, %s)",
+            "INSERT INTO runs (id, task_id, parent_id, agent_id, branch, tldr, message, score, verified,"
+            " verification_status, created_at, fork_id)"
+            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, FALSE, %s, %s, %s)",
             (sha, task_id, parent_id, agent_id, body.get("branch", ""),
-             body.get("tldr", ""), body.get("message", ""), score, ts, fork_id),
+             body.get("tldr", ""), body.get("message", ""), score, vstat, ts, fork_id),
         )
         await conn.execute("UPDATE agents SET total_runs = total_runs + 1 WHERE id = %s", (agent_id,))
-        if score is not None:
+        if score is not None and not verify_on:
             await conn.execute(
                 "UPDATE tasks SET"
                 " improvements = CASE WHEN %s > COALESCE(best_score, '-Infinity'::float) THEN improvements + 1 ELSE improvements END,"
@@ -1357,9 +1488,45 @@ async def submit_run(task_id: str, body: dict[str, Any], token: str = Query(""),
             " VALUES (%s, %s, %s, %s, 0, 0, %s) RETURNING id",
             (task_id, agent_id, body.get("message", ""), sha, ts),
         )).fetchone())["id"]
-    run = {"id": sha, "task_id": task_id, "agent_id": agent_id, "branch": body.get("branch", ""),
-           "parent_id": parent_id, "tldr": body.get("tldr", ""), "message": body.get("message", ""),
-           "score": score, "verified": False, "created_at": ts, "fork_id": fork_id}
+        art_cfg = task_cfg.get("artifact") or {}
+        max_mb = float(art_cfg.get("max_size_mb") or 50)
+        max_b = int(max_mb * 1024 * 1024)
+        attempt_id: str | None = None
+        if verify_on:
+            root = Path(os.environ.get("HIVE_ARTIFACT_ROOT", "/tmp/hive_artifacts")) / task_id / sha
+            root.mkdir(parents=True, exist_ok=True)
+            for field_name, uf in upload_files:
+                rel = _safe_artifact_rel_path(field_name)
+                if not rel:
+                    raise HTTPException(400, f"invalid artifact path: {field_name!r}")
+                dest = root / rel
+                dest.parent.mkdir(parents=True, exist_ok=True)
+                total = 0
+                with dest.open("wb") as out:
+                    while chunk := await uf.read(64 * 1024):
+                        total += len(chunk)
+                        if total > max_b:
+                            raise HTTPException(413, "artifact too large")
+                        out.write(chunk)
+                await conn.execute(
+                    "INSERT INTO run_artifacts (run_id, task_id, rel_path, storage_path, size_bytes, created_at)"
+                    " VALUES (%s, %s, %s, %s, %s, %s)",
+                    (sha, task_id, rel, str(dest.resolve()), total, ts),
+                )
+            attempt_id = str(uuid.uuid4())
+            await conn.execute(
+                "INSERT INTO verification_attempts (id, task_id, run_id, status, created_at)"
+                " VALUES (%s, %s, %s, %s, %s)",
+                (attempt_id, task_id, sha, "pending", ts),
+            )
+    if verify_on and attempt_id:
+        background_tasks.add_task(verify_run_background, sha, task_id, attempt_id)
+    run = {
+        "id": sha, "task_id": task_id, "agent_id": agent_id, "branch": body.get("branch", ""),
+        "parent_id": parent_id, "tldr": body.get("tldr", ""), "message": body.get("message", ""),
+        "score": score, "verified": False, "verified_score": None, "verification_status": vstat,
+        "created_at": ts, "fork_id": fork_id,
+    }
     return JSONResponse({"run": run, "post_id": post_id}, status_code=201)
 
 
@@ -1372,16 +1539,17 @@ async def list_runs(task_id: str, authorization: str = Header(""), sort: str = Q
 
         if view == "contributors":
             rows = await (await conn.execute(
-                "SELECT agent_id, COUNT(*) AS total_runs, MAX(score) AS best_score,"
+                "SELECT agent_id, COUNT(*) AS total_runs, MAX(COALESCE(verified_score, score)) AS best_score,"
                 " COUNT(*) FILTER ("
-                "   WHERE score > COALESCE("
-                "     (SELECT MAX(r2.score) FROM runs r2"
+                "   WHERE COALESCE(verified_score, score) > COALESCE("
+                "     (SELECT MAX(COALESCE(r2.verified_score, r2.score)) FROM runs r2"
                 "      WHERE r2.task_id = runs.task_id"
-                "      AND r2.created_at < runs.created_at AND r2.score IS NOT NULL),"
+                "      AND r2.created_at < runs.created_at"
+                "      AND (r2.score IS NOT NULL OR r2.verified_score IS NOT NULL)),"
                 "     '-Infinity'::float)"
                 " ) AS improvements"
                 " FROM runs"
-                " WHERE task_id = %s AND score IS NOT NULL"
+                " WHERE task_id = %s AND (score IS NOT NULL OR verified_score IS NOT NULL)"
                 " GROUP BY agent_id ORDER BY improvements DESC, best_score DESC LIMIT %s OFFSET %s",
                 (task_id, per_page + 1, offset)
             )).fetchall()
@@ -1393,10 +1561,12 @@ async def list_runs(task_id: str, authorization: str = Header(""), sort: str = Q
 
         if view == "deltas":
             rows = await (await conn.execute(
-                "SELECT r.id AS run_id, r.agent_id, r.score - p.score AS delta,"
-                " p.score AS from_score, r.score AS to_score, r.tldr"
+                "SELECT r.id AS run_id, r.agent_id,"
+                " COALESCE(r.verified_score, r.score) - COALESCE(p.verified_score, p.score) AS delta,"
+                " COALESCE(p.verified_score, p.score) AS from_score, COALESCE(r.verified_score, r.score) AS to_score, r.tldr"
                 " FROM runs r JOIN runs p ON r.parent_id = p.id"
-                " WHERE r.task_id = %s AND r.score IS NOT NULL AND p.score IS NOT NULL"
+                " WHERE r.task_id = %s AND (r.score IS NOT NULL OR r.verified_score IS NOT NULL)"
+                " AND (p.score IS NOT NULL OR p.verified_score IS NOT NULL)"
                 " ORDER BY delta DESC LIMIT %s OFFSET %s",
                 (task_id, per_page + 1, offset)
             )).fetchall()
@@ -1407,14 +1577,14 @@ async def list_runs(task_id: str, authorization: str = Header(""), sort: str = Q
         if view == "improvers":
             rows = await (await conn.execute(
                 "WITH ranked AS ("
-                " SELECT agent_id, score,"
-                " MAX(score) OVER (ORDER BY created_at"
+                " SELECT agent_id, COALESCE(verified_score, score) AS eff,"
+                " MAX(COALESCE(verified_score, score)) OVER (ORDER BY created_at"
                 " ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS prev_best"
-                " FROM runs WHERE task_id = %s AND score IS NOT NULL"
+                " FROM runs WHERE task_id = %s AND (score IS NOT NULL OR verified_score IS NOT NULL)"
                 ")"
                 " SELECT agent_id,"
-                " COUNT(*) FILTER (WHERE score > COALESCE(prev_best, '-Infinity'::float)) AS improvements_to_best,"
-                " MAX(score) AS best_score"
+                " COUNT(*) FILTER (WHERE eff > COALESCE(prev_best, '-Infinity'::float)) AS improvements_to_best,"
+                " MAX(eff) AS best_score"
                 " FROM ranked"
                 " GROUP BY agent_id"
                 " ORDER BY improvements_to_best DESC"
@@ -1425,12 +1595,16 @@ async def list_runs(task_id: str, authorization: str = Header(""), sort: str = Q
             rows = rows[:per_page]
             return {"view": "improvers", "entries": [dict(r) for r in rows], "page": page, "per_page": per_page, "has_next": has_next}
 
-        where, params = "r.task_id = %s AND r.score IS NOT NULL AND r.valid IS NOT FALSE", [task_id]
+        where, params = (
+            "r.task_id = %s AND (r.score IS NOT NULL OR r.verified_score IS NOT NULL) AND r.valid IS NOT FALSE",
+            [task_id],
+        )
         if agent: where += " AND r.agent_id = %s"; params.append(agent)
-        order = _parse_sort(sort, {"score": "r.score", "recent": "r.created_at"})
+        order = _parse_sort(sort, {"score": "COALESCE(r.verified_score, r.score)", "recent": "r.created_at"})
         params.extend([per_page + 1, offset])
         rows = await (await conn.execute(
-            f"SELECT r.id, r.agent_id, r.branch, r.parent_id, r.tldr, r.score, r.verified, r.valid, r.created_at, f.fork_url"
+            f"SELECT r.id, r.agent_id, r.branch, r.parent_id, r.tldr, r.score, r.verified_score, r.verification_status,"
+            f" r.verified, r.valid, r.created_at, f.fork_url"
             f" FROM runs r LEFT JOIN forks f ON f.id = r.fork_id WHERE {where} ORDER BY {order} LIMIT %s OFFSET %s", params
         )).fetchall()
         has_next = len(rows) > per_page
@@ -1488,12 +1662,60 @@ async def patch_run(task_id: str, sha: str, body: dict[str, Any],
             await conn.execute("UPDATE runs SET valid = %s WHERE id = %s", (valid, sha))
             # Recalculate best_score excluding invalid runs
             best = await (await conn.execute(
-                "SELECT MAX(score) AS val FROM runs WHERE task_id = %s AND valid = TRUE", (task_id,)
+                "SELECT MAX(COALESCE(verified_score, score)) AS val FROM runs WHERE task_id = %s AND valid = TRUE",
+                (task_id,),
             )).fetchone()
             await conn.execute("UPDATE tasks SET best_score = %s WHERE id = %s", (best["val"], task_id))
         return {"id": sha, "valid": body.get("valid")}
 
 
+@router.post("/tasks/{task_id}/runs/{sha}/verify")
+async def queue_run_verify(
+    task_id: str, sha: str,
+    background_tasks: BackgroundTasks,
+    x_admin_key: str = Header(""), authorization: str = Header(""),
+):
+    await require_admin_or_task_owner(task_id, x_admin_key, authorization)
+    ts = now()
+    async with get_db() as conn:
+        task_row = await (await conn.execute("SELECT config FROM tasks WHERE id = %s", (task_id,))).fetchone()
+        if not task_row:
+            raise HTTPException(404, "task not found")
+        if not is_verify_enabled(parse_json_config(task_row.get("config"))):
+            raise HTTPException(400, "task does not have verification enabled")
+        row = await (await conn.execute(
+            "SELECT id FROM runs WHERE id = %s AND task_id = %s", (sha, task_id)
+        )).fetchone()
+        if not row:
+            rows = await (await conn.execute(
+                "SELECT id FROM runs WHERE id LIKE %s AND task_id = %s", (sha + "%", task_id)
+            )).fetchall()
+            if len(rows) == 1:
+                row = rows[0]
+            elif len(rows) > 1:
+                raise HTTPException(400, f"ambiguous prefix '{sha}', matches {len(rows)} runs")
+            else:
+                raise HTTPException(404, "run not found")
+        sha = row["id"]
+        cnt = (await (await conn.execute(
+            "SELECT COUNT(*) AS c FROM run_artifacts WHERE run_id = %s", (sha,)
+        )).fetchone())["c"]
+        if not cnt:
+            raise HTTPException(400, "run has no stored artifacts to verify")
+        attempt_id = str(uuid.uuid4())
+        await conn.execute(
+            "INSERT INTO verification_attempts (id, task_id, run_id, status, created_at)"
+            " VALUES (%s, %s, %s, %s, %s)",
+            (attempt_id, task_id, sha, "pending", ts),
+        )
+        await conn.execute(
+            "UPDATE runs SET verification_status = %s, verification_error = NULL WHERE id = %s",
+            ("pending", sha),
+        )
+    background_tasks.add_task(verify_run_background, sha, task_id, attempt_id)
+    return {"id": sha, "verification_status": "pending", "attempt_id": attempt_id}
+
+
 @router.delete("/tasks/{task_id}/runs/{sha}")
 async def delete_run(task_id: str, sha: str, x_admin_key: str = Header(""), authorization: str = Header("")):
     """Delete a single run and its associated post, comments, and votes."""
@@ -1529,7 +1751,7 @@ async def delete_run(task_id: str, sha: str, x_admin_key: str = Header(""), auth
         await conn.execute("DELETE FROM runs WHERE id = %s", (sha,))
         # Recalculate task stats (exclude invalid runs)
         best = await (await conn.execute(
-            "SELECT MAX(score) AS val FROM runs WHERE task_id = %s AND valid IS NOT FALSE", (task_id,)
+            "SELECT MAX(COALESCE(verified_score, score)) AS val FROM runs WHERE task_id = %s AND valid IS NOT FALSE", (task_id,)
         )).fetchone()
         await conn.execute(
             "UPDATE tasks SET best_score = %s WHERE id = %s",
@@ -1910,9 +2132,10 @@ async def get_context(task_id: str, authorization: str = Header("")):
             "last_activity": last_activity,
         }
         leaderboard = await (await conn.execute(
-            "SELECT r.id, r.agent_id, r.score, r.tldr, r.branch, r.verified, f.fork_url"
+            "SELECT r.id, r.agent_id, r.score, r.verified_score, r.verification_status, r.tldr, r.branch, r.verified, f.fork_url"
             " FROM runs r LEFT JOIN forks f ON f.id = r.fork_id"
-            " WHERE r.task_id = %s AND r.score IS NOT NULL AND r.valid IS NOT FALSE ORDER BY r.score DESC LIMIT 5", (task_id,)
+            " WHERE r.task_id = %s AND (r.score IS NOT NULL OR r.verified_score IS NOT NULL) AND r.valid IS NOT FALSE"
+            " ORDER BY COALESCE(r.verified_score, r.score) DESC LIMIT 5", (task_id,)
         )).fetchall()
         now_ts = now()
         active_claims = await (await conn.execute(
@@ -1953,10 +2176,10 @@ async def get_graph(task_id: str, authorization: str = Header(""), max_nodes: in
             raise HTTPException(404, "task not found")
         total = (await (await conn.execute("SELECT COUNT(*) AS cnt FROM runs WHERE task_id = %s", (task_id,))).fetchone())["cnt"]
         rows = await (await conn.execute(
-            "SELECT id AS sha, agent_id, score, parent_id, tldr, created_at, valid FROM runs WHERE task_id = %s ORDER BY created_at DESC LIMIT %s",
+            "SELECT id AS sha, agent_id, score, verified_score, parent_id, tldr, created_at, valid FROM runs WHERE task_id = %s ORDER BY created_at DESC LIMIT %s",
             (task_id, max_nodes)
         )).fetchall()
-    nodes = [{"sha": r["sha"], "agent_id": r["agent_id"], "score": r["score"],
+    nodes = [{"sha": r["sha"], "agent_id": r["agent_id"], "score": r["score"], "verified_score": r.get("verified_score"),
                "parent": r["parent_id"], "is_seed": r["parent_id"] is None,
                "tldr": r["tldr"], "created_at": r["created_at"],
                "valid": r["valid"] if r["valid"] is not None else True} for r in rows]
diff --git a/src/hive/server/verifier.py b/src/hive/server/verifier.py
new file mode 100644
index 0000000..d5ad760
--- /dev/null
+++ b/src/hive/server/verifier.py
@@ -0,0 +1,207 @@
+"""Run trusted server_eval against submitted artifacts (local subprocess; Daytona later)."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from psycopg.types.json import Json
+
+from .db import get_db, now
+from .verify_config import is_verify_enabled, parse_json_config
+
+log = logging.getLogger(__name__)
+
+
+def _eval_root_from_volume_id(volume_id: str) -> Path:
+    if volume_id.startswith("local:"):
+        return Path(volume_id[6:]).resolve()
+    raise ValueError(f"unsupported volume_id (only local: paths in this build): {volume_id!r}")
+
+
+def _parse_metrics_json(stdout: str, score_key: str) -> dict[str, Any] | None:
+    for line in reversed([l.strip() for l in stdout.splitlines() if l.strip()]):
+        if line.startswith("{") and line.endswith("}"):
+            try:
+                d = json.loads(line)
+                if isinstance(d, dict) and score_key in d:
+                    return d
+            except json.JSONDecodeError:
+                continue
+    return None
+
+
+def run_server_eval_sync(
+    eval_extract_root: Path,
+    artifact_abs_paths: dict[str, Path],
+    command: str | list[str],
+    score_key: str,
+    direction: str,
+    timeout: int = 300,
+) -> dict[str, Any]:
+    work = Path(tempfile.mkdtemp(prefix="hive_verify_"))
+    try:
+        eval_root = work / "eval"
+        shutil.copytree(eval_extract_root, eval_root)
+        art_root = work / "artifacts"
+        art_root.mkdir()
+        for rel, src in artifact_abs_paths.items():
+            dest = art_root / rel
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(src, dest)
+        shell = isinstance(command, str)
+        proc = subprocess.run(
+            command,
+            cwd=str(eval_root),
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            shell=shell,
+        )
+        out = (proc.stdout or "") + ((proc.stderr or "") and f"\n{proc.stderr}" or "")
+        if proc.returncode != 0:
+            return {"ok": False, "error": f"server_eval exited {proc.returncode}", "logs": out}
+        text = proc.stdout or proc.stderr or ""
+        metrics = _parse_metrics_json(text, score_key)
+        if metrics is None:
+            return {"ok": False, "error": "could not parse JSON metrics line from server_eval output", "logs": out}
+        raw = float(metrics[score_key])
+        verified_score = raw if direction == "maximize" else -raw
+        return {
+            "ok": True,
+            "verified_score": verified_score,
+            "metric_value": raw,
+            "metrics": metrics,
+            "logs": out,
+        }
+    finally:
+        shutil.rmtree(work, ignore_errors=True)
+
+
+async def verify_run_background(run_id: str, task_id: str, attempt_id: str) -> None:
+    try:
+        async with get_db() as conn:
+            task_row = await (await conn.execute("SELECT config FROM tasks WHERE id = %s", (task_id,))).fetchone()
+            if not task_row:
+                return
+            cfg = parse_json_config(task_row.get("config"))
+            if not is_verify_enabled(cfg):
+                return
+            se = cfg.get("server_eval") or {}
+            volume_id = se.get("volume_id")
+            if not volume_id:
+                await conn.execute(
+                    "UPDATE runs SET verification_status = %s, verification_error = %s WHERE id = %s",
+                    ("error", "task missing server_eval.volume_id", run_id),
+                )
+                await conn.execute(
+                    "UPDATE verification_attempts SET status = %s, error = %s, finished_at = %s WHERE id = %s",
+                    ("error", "task missing server_eval.volume_id", now(), attempt_id),
+                )
+                return
+            extract_root = _eval_root_from_volume_id(volume_id)
+            if not extract_root.is_dir():
+                await conn.execute(
+                    "UPDATE runs SET verification_status = %s, verification_error = %s WHERE id = %s",
+                    ("error", f"eval bundle path not found: {extract_root}", run_id),
+                )
+                await conn.execute(
+                    "UPDATE verification_attempts SET status = %s, error = %s, finished_at = %s WHERE id = %s",
+                    ("error", f"eval bundle path not found: {extract_root}", now(), attempt_id),
+                )
+                return
+            rows = await (await conn.execute(
+                "SELECT rel_path, storage_path FROM run_artifacts WHERE run_id = %s ORDER BY rel_path",
+                (run_id,),
+            )).fetchall()
+            required = (cfg.get("artifact") or {}).get("required_paths") or []
+            paths_map = {r["rel_path"]: Path(r["storage_path"]) for r in rows}
+            missing = [p for p in required if p not in paths_map]
+            if missing:
+                await conn.execute(
+                    "UPDATE runs SET verification_status = %s, verification_error = %s WHERE id = %s",
+                    ("failed", f"missing artifacts: {missing}", run_id),
+                )
+                await conn.execute(
+                    "UPDATE verification_attempts SET status = %s, error = %s, finished_at = %s WHERE id = %s",
+                    ("failed", f"missing artifacts: {missing}", now(), attempt_id),
+                )
+                return
+            command = se["command"]
+            score_key = se["score_key"]
+            direction = se.get("direction", "maximize")
+            sandbox = cfg.get("sandbox") or {}
+            timeout = int(sandbox.get("timeout_seconds") or os.environ.get("VERIFY_EVAL_TIMEOUT", "300"))
+            await conn.execute(
+                "UPDATE verification_attempts SET status = %s, started_at = %s WHERE id = %s",
+                ("running", now(), attempt_id),
+            )
+
+        result = await asyncio.to_thread(
+            run_server_eval_sync,
+            extract_root,
+            paths_map,
+            command,
+            score_key,
+            direction,
+            timeout,
+        )
+        finished = now()
+        async with get_db() as conn:
+            if result.get("ok"):
+                vs = result["verified_score"]
+                mv = result["metric_value"]
+                await conn.execute(
+                    "UPDATE runs SET verified = TRUE, verified_score = %s, verified_metric_key = %s,"
+                    " verified_metric_value = %s, verification_status = %s, verification_error = NULL,"
+                    " verified_at = %s WHERE id = %s",
+                    (vs, score_key, mv, "success", finished, run_id),
+                )
+                await conn.execute(
+                    "UPDATE verification_attempts SET status = %s, metrics = %s, logs = %s, finished_at = %s"
+                    " WHERE id = %s",
+                    ("success", Json(result.get("metrics") or {}), result.get("logs", "")[:20000], finished, attempt_id),
+                )
+                await _recompute_task_best_verified(conn, task_id, cfg)
+            else:
+                err = result.get("error", "unknown")
+                await conn.execute(
+                    "UPDATE runs SET verification_status = %s, verification_error = %s WHERE id = %s",
+                    ("failed", err[:2000], run_id),
+                )
+                await conn.execute(
+                    "UPDATE verification_attempts SET status = %s, error = %s, logs = %s, finished_at = %s"
+                    " WHERE id = %s",
+                    ("failed", err[:2000], (result.get("logs") or "")[:20000], finished, attempt_id),
+                )
+    except Exception as e:
+        log.exception("verify_run_background failed")
+        finished = now()
+        async with get_db() as conn:
+            await conn.execute(
+                "UPDATE runs SET verification_status = %s, verification_error = %s WHERE id = %s",
+                ("error", str(e)[:2000], run_id),
+            )
+            await conn.execute(
+                "UPDATE verification_attempts SET status = %s, error = %s, finished_at = %s WHERE id = %s",
+                ("error", str(e)[:2000], finished, attempt_id),
+            )
+
+
+async def _recompute_task_best_verified(conn, task_id: str, cfg: dict) -> None:
+    if not is_verify_enabled(cfg):
+        return
+    row = await (await conn.execute(
+        "SELECT MAX(verified_score) AS v FROM runs WHERE task_id = %s AND verification_status = %s AND valid IS NOT FALSE",
+        (task_id, "success"),
+    )).fetchone()
+    best = row["v"] if row else None
+    if best is not None:
+        await conn.execute("UPDATE tasks SET best_score = %s WHERE id = %s", (best, task_id))
diff --git a/src/hive/server/verify_config.py b/src/hive/server/verify_config.py
new file mode 100644
index 0000000..7607577
--- /dev/null
+++ b/src/hive/server/verify_config.py
@@ -0,0 +1,62 @@
+import json
+from typing import Any
+
+
+def parse_json_config(raw: str | None) -> dict[str, Any]:
+    if not raw:
+        return {}
+    try:
+        out = json.loads(raw)
+        return out if isinstance(out, dict) else {}
+    except json.JSONDecodeError:
+        return {}
+
+
+def is_verify_enabled(cfg: dict[str, Any]) -> bool:
+    return bool(cfg.get("verify"))
+
+
+def validate_verify_task_config(cfg: dict[str, Any], has_eval_bundle: bool) -> None:
+    if not cfg.get("verify"):
+        return
+    if not has_eval_bundle:
+        raise ValueError("verify=true requires eval_bundle upload")
+    art = cfg.get("artifact")
+    if not isinstance(art, dict) or not art.get("required_paths"):
+        raise ValueError("verify=true requires artifact.required_paths (array of relative paths)")
+    paths = art["required_paths"]
+    if not isinstance(paths, list) or not all(isinstance(p, str) and p.strip() for p in paths):
+        raise ValueError("artifact.required_paths must be a non-empty list of strings")
+    se = cfg.get("server_eval")
+    if not isinstance(se, dict):
+        raise ValueError("verify=true requires server_eval object")
+    if not se.get("command"):
+        raise ValueError("server_eval.command is required")
+    if not se.get("score_key"):
+        raise ValueError("server_eval.score_key is required")
+    direction = se.get("direction", "maximize")
+    if direction not in ("maximize", "minimize"):
+        raise ValueError("server_eval.direction must be 'maximize' or 'minimize'")
+    result_format = se.get("result_format", "json")
+    if result_format != "json":
+        raise ValueError("only result_format=json is supported for now")
+
+
+def merge_verify_into_config(base_json: str | None, verify_blob: str | None) -> dict[str, Any]:
+    merged = parse_json_config(base_json)
+    if verify_blob:
+        v = parse_json_config(verify_blob)
+        merged.update(v)
+    return merged
+
+
+def leaderboard_score_expr(verify: bool) -> str:
+    if verify:
+        return "COALESCE(r.verified_score, r.score)"
+    return "r.score"
+
+
+def best_score_expr_for_task(verify: bool) -> str:
+    if verify:
+        return "COALESCE(verified_score, score)"
+    return "score"
diff --git a/tests/cli/test_cmd_item.py b/tests/cli/test_cmd_item.py
index ad70418..9afbaad 100644
--- a/tests/cli/test_cmd_item.py
+++ b/tests/cli/test_cmd_item.py
@@ -10,9 +10,9 @@
 def _post_task(task_id="cli-items"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 9729d79..2d101b8 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -11,7 +11,7 @@
 from tests.mocks import MockGitHubApp
 from hive.server.github import set_github_app
 
-_ALL_TABLES = "password_resets, oauth_states, pending_signups, item_comments, items, votes, comments, claims, skills, posts, runs, forks, agents, tasks, users"
+_ALL_TABLES = "verification_attempts, run_artifacts, task_eval_bundles, password_resets, oauth_states, pending_signups, item_comments, items, votes, comments, claims, skills, posts, runs, forks, agents, tasks, users"
 
 
 def _free_port():
diff --git a/tests/server/test_items.py b/tests/server/test_items.py
index 38ad285..85b6c5d 100644
--- a/tests/server/test_items.py
+++ b/tests/server/test_items.py
@@ -7,9 +7,9 @@
 def _post_task(client, task_id="gsm8k"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_items_adversarial.py b/tests/server/test_items_adversarial.py
index c0e28fe..55173dc 100644
--- a/tests/server/test_items_adversarial.py
+++ b/tests/server/test_items_adversarial.py
@@ -13,9 +13,9 @@
 def _post_task(client, task_id="adv-task"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_items_round3.py b/tests/server/test_items_round3.py
index 35177e2..a2752d1 100644
--- a/tests/server/test_items_round3.py
+++ b/tests/server/test_items_round3.py
@@ -12,9 +12,9 @@
 def _post_task(client, task_id="r3-task"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_items_round4.py b/tests/server/test_items_round4.py
index f6da3da..64f8d59 100644
--- a/tests/server/test_items_round4.py
+++ b/tests/server/test_items_round4.py
@@ -14,9 +14,9 @@
 def _post_task(client, task_id="r4-task"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_items_round5.py b/tests/server/test_items_round5.py
index 935ede6..c798abc 100644
--- a/tests/server/test_items_round5.py
+++ b/tests/server/test_items_round5.py
@@ -15,9 +15,9 @@
 def _post_task(client, task_id="r5-task"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_items_round6.py b/tests/server/test_items_round6.py
index 8dca3bc..336ebdd 100644
--- a/tests/server/test_items_round6.py
+++ b/tests/server/test_items_round6.py
@@ -16,9 +16,9 @@
 def _post_task(client, task_id="r6-task"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "test", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_items_stress.py b/tests/server/test_items_stress.py
index 8528c70..7d13dab 100644
--- a/tests/server/test_items_stress.py
+++ b/tests/server/test_items_stress.py
@@ -7,9 +7,9 @@
 def _post_task(client, task_id="stress-task"):
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at, item_seq)"
-            " VALUES (%s, %s, %s, %s, %s, 0)",
-            (task_id, task_id, "stress test", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at, item_seq)"
+            " VALUES (%s, %s, %s, %s, %s, %s, 0)",
+            (task_id, task_id, task_id, "stress test", "https://github.com/test", _db.now()),
         )
 
 
@@ -17,9 +17,9 @@ def _post_task_no_seq(client, task_id="no-seq-task"):
     """Insert a task row WITHOUT item_seq to test the migration path."""
     with psycopg.connect(_db.DATABASE_URL, autocommit=True) as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at)"
-            " VALUES (%s, %s, %s, %s, %s)",
-            (task_id, task_id, "no seq", "https://github.com/test", _db.now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at)"
+            " VALUES (%s, %s, %s, %s, %s, %s)",
+            (task_id, task_id, task_id, "no seq", "https://github.com/test", _db.now()),
         )
 
 
diff --git a/tests/server/test_main.py b/tests/server/test_main.py
index fd253c4..d379c21 100644
--- a/tests/server/test_main.py
+++ b/tests/server/test_main.py
@@ -1034,8 +1034,8 @@ def test_unique_agents_across_tasks(self, client):
         with get_db_sync() as conn:
             for tid in ("ta", "tb"):
                 conn.execute(
-                    "INSERT INTO tasks (id, name, description, repo_url, created_at) VALUES (%s, %s, %s, %s, %s)",
-                    (tid, tid, "desc", "https://github.com/test/test", now()),
+                    "INSERT INTO tasks (id, slug, name, description, repo_url, created_at) VALUES (%s, %s, %s, %s, %s, %s)",
+                    (tid, tid, tid, "desc", "https://github.com/test/test", now()),
                 )
         # Submit to both tasks
         client.post("/api/tasks/ta/submit", params={"token": token},
@@ -1226,6 +1226,6 @@ def _seed_task(client):
     from hive.server.db import get_db_sync, now
     with get_db_sync() as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, created_at) VALUES (%s, %s, %s, %s, %s)",
-            ("t1", "Test Task", "A test", "https://github.com/test/test", now()),
+            "INSERT INTO tasks (id, slug, name, description, repo_url, created_at) VALUES (%s, %s, %s, %s, %s, %s)",
+            ("t1", "t1", "Test Task", "A test", "https://github.com/test/test", now()),
         )
diff --git a/tests/server/test_private_tasks.py b/tests/server/test_private_tasks.py
index 57953d7..5ff9ecf 100644
--- a/tests/server/test_private_tasks.py
+++ b/tests/server/test_private_tasks.py
@@ -40,10 +40,10 @@ def _seed_private_task(client, user_id, task_id="priv-task", source_repo="testow
     """Insert a private task directly into DB."""
     with get_db_sync() as conn:
         conn.execute(
-            "INSERT INTO tasks (id, name, description, repo_url, task_type, owner_id, "
+            "INSERT INTO tasks (id, slug, name, description, repo_url, task_type, owner_id, "
             "visibility, source_repo, installation_id, created_at)"
-            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
-            (task_id, "Private Task", "A private test task",
+            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
+            (task_id, task_id, "Private Task", "A private test task",
              f"https://github.com/{source_repo}", "private", user_id,
              "private", source_repo, installation_id, now()),
         )
@@ -152,9 +152,9 @@ def test_clone_public_task_unchanged(self, client, mock_github):
         """Public task clone should still return fork mode (no mode field)."""
         with get_db_sync() as conn:
             conn.execute(
-                "INSERT INTO tasks (id, name, description, repo_url, created_at)"
-                " VALUES (%s, %s, %s, %s, %s)",
-                ("pub-task", "Public Task", "A public test", "https://github.com/test/test", now()),
+                "INSERT INTO tasks (id, slug, name, description, repo_url, created_at)"
+                " VALUES (%s, %s, %s, %s, %s, %s)",
+                ("pub-task", "pub-task", "Public Task", "A public test", "https://github.com/test/test", now()),
             )
         resp = client.post("/api/register")
         agent_token = resp.json()["token"]
@@ -225,9 +225,9 @@ def test_push_main_branch_rejected(self, client, mock_github):
     def test_push_public_task_rejected(self, client, mock_github):
         with get_db_sync() as conn:
             conn.execute(
-                "INSERT INTO tasks (id, name, description, repo_url, created_at)"
-                " VALUES (%s, %s, %s, %s, %s)",
-                ("pub-task", "Public Task", "A public test", "https://github.com/test/test", now()),
+                "INSERT INTO tasks (id, slug, name, description, repo_url, created_at)"
+                " VALUES (%s, %s, %s, %s, %s, %s)",
+                ("pub-task", "pub-task", "Public Task", "A public test", "https://github.com/test/test", now()),
             )
         resp = client.post("/api/register")
         agent_token = resp.json()["token"]
diff --git a/tests/server/test_verified_tasks.py b/tests/server/test_verified_tasks.py
new file mode 100644
index 0000000..024518f
--- /dev/null
+++ b/tests/server/test_verified_tasks.py
@@ -0,0 +1,119 @@
+import io
+import json
+import tarfile
+
+import pytest
+
+
+def _minimal_task_tar():
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        for name, content in [
+            ("program.md", b"# t\n"),
+            ("eval/eval.sh", b"#!/bin/sh\necho ok\n"),
+            ("prepare.sh", b"#!/bin/sh\n"),
+        ]:
+            ti = tarfile.TarInfo(name)
+            ti.size = len(content)
+            tar.addfile(ti, io.BytesIO(content))
+    return buf.getvalue()
+
+
+def _eval_bundle_tar():
+    server_eval = b'''import json
+print(json.dumps({"mae": 0.5, "neg_mae": -0.5}))
+'''
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        for name, content in [
+            ("server_eval.py", server_eval),
+            ("hidden/actuals.csv", b"a,b\n"),
+        ]:
+            ti = tarfile.TarInfo(name)
+            ti.size = len(content)
+            tar.addfile(ti, io.BytesIO(content))
+    return buf.getvalue()
+
+
+def test_create_verified_task_requires_bundle(client, monkeypatch, tmp_path):
+    task_tar = _minimal_task_tar()
+    vcfg = {
+        "verify": True,
+        "artifact": {"required_paths": ["artifacts/predictions.csv"]},
+        "server_eval": {"command": "python3 server_eval.py", "score_key": "neg_mae", "direction": "maximize"},
+    }
+    resp = client.post(
+        "/api/tasks",
+        data={
+            "id": "vt-no-bundle",
+            "name": "VT",
+            "description": "d",
+            "verify_config": json.dumps(vcfg),
+        },
+        files={"archive": ("t.tar.gz", task_tar, "application/gzip")},
+        headers={"X-Admin-Key": "test-key"},
+    )
+    assert resp.status_code == 400
+
+
+def test_create_verified_task_atomic_and_submit_verifies(client, monkeypatch, tmp_path):
+    monkeypatch.setenv("HIVE_EVAL_ROOT", str(tmp_path / "eval"))
+    monkeypatch.setenv("HIVE_ARTIFACT_ROOT", str(tmp_path / "art"))
+    task_tar = _minimal_task_tar()
+    bundle = _eval_bundle_tar()
+    vcfg = {
+        "verify": True,
+        "artifact": {"required_paths": ["artifacts/predictions.csv"], "max_size_mb": 5},
+        "server_eval": {
+            "volume_version": "v1",
+            "command": "python3 server_eval.py",
+            "score_key": "neg_mae",
+            "direction": "maximize",
+        },
+    }
+    resp = client.post(
+        "/api/tasks",
+        data={
+            "id": "vt-ok",
+            "name": "VT",
+            "description": "d",
+            "verify_config": json.dumps(vcfg),
+        },
+        files={
+            "archive": ("t.tar.gz", task_tar, "application/gzip"),
+            "eval_bundle": ("eval.tar.gz", bundle, "application/gzip"),
+        },
+        headers={"X-Admin-Key": "test-key"},
+    )
+    assert resp.status_code == 201, resp.text
+    r = client.post("/api/register")
+    token = r.json()["token"]
+    aid = r.json()["id"]
+    pred = b"a,b,c\n"
+    resp = client.post(
+        "/api/tasks/vt-ok/submit",
+        data={
+            "sha": "abc123def4567890123456789abcdef01234567",
+            "branch": "main",
+            "tldr": "t",
+            "message": "m",
+            "score": "0.9",
+            "parent_id": "",
+        },
+        files={"artifacts/predictions.csv": ("predictions.csv", io.BytesIO(pred), "text/csv")},
+        headers={"X-Agent-Token": token},
+    )
+    assert resp.status_code == 201, resp.text
+    row = resp.json()["run"]
+    assert row["verification_status"] in ("pending", "running", "success", "failed", "error")
+    import time
+    for _ in range(50):
+        rr = client.get(f"/api/tasks/vt-ok/runs/{row['id']}", headers={"X-Agent-Token": token})
+        if rr.json().get("verification_status") == "success":
+            assert rr.json().get("verified_score") is not None
+            return
+        if rr.json().get("verification_status") in ("failed", "error"):
+            pytest.fail(f"verify failed: {rr.json()}")
+        time.sleep(0.05)
+    pytest.fail("verification did not complete")
+
diff --git a/ui/src/components/leaderboard.tsx b/ui/src/components/leaderboard.tsx
index c2238d1..ce1fd90 100644
--- a/ui/src/components/leaderboard.tsx
+++ b/ui/src/components/leaderboard.tsx
@@ -65,20 +65,26 @@ function buildDenseRanks<T>(items: T[], getValue: (item: T) => number | null): n
   return ranks;
 }
 
+function effectiveRunScore(r: { score: number | null; verified_score?: number | null }): number | null {
+  if (r.verified_score != null) return r.verified_score;
+  return r.score;
+}
+
 function BestScoreList({
   runs,
   onRunClick,
 }: {
-  runs: Pick<Run, "id" | "agent_id" | "branch" | "parent_id" | "tldr" | "score" | "verified" | "created_at" | "fork_url">[];
+  runs: Pick<Run, "id" | "agent_id" | "branch" | "parent_id" | "tldr" | "score" | "verified_score" | "verified" | "created_at" | "fork_url">[];
   onRunClick?: (runId: string) => void;
 }) {
-  const ranks = buildDenseRanks(runs, (r) => r.score);
-  const bestScore = runs.length > 0 ? runs[0].score : null;
+  const ranks = buildDenseRanks(runs, (r) => effectiveRunScore(r));
+  const bestScore = runs.length > 0 ? effectiveRunScore(runs[0]) : null;
 
   return (
     <>
       {runs.map((run, i) => {
-        const isWinner = run.score !== null && run.score === bestScore;
+        const eff = effectiveRunScore(run);
+        const isWinner = eff !== null && eff === bestScore;
         return (
           <div
             key={run.id}
@@ -95,7 +101,7 @@ function BestScoreList({
               </div>
               <div className="text-xs text-[var(--color-text-tertiary)] mt-0.5 truncate">{run.tldr}</div>
             </div>
-            <Score value={run.score} className="text-sm shrink-0 text-[var(--color-text)]" />
+            <Score value={eff} className="text-sm shrink-0 text-[var(--color-text)]" />
           </div>
         );
       })}
diff --git a/ui/src/components/run-detail.tsx b/ui/src/components/run-detail.tsx
index 38f89bd..be732e5 100644
--- a/ui/src/components/run-detail.tsx
+++ b/ui/src/components/run-detail.tsx
@@ -21,6 +21,11 @@ interface FullRun extends Run {
   base_sha?: string;
 }
 
+function effectiveRunScore(r: Pick<Run, "score" | "verified_score">): number | null {
+  if (r.verified_score != null) return r.verified_score;
+  return r.score;
+}
+
 interface RunDetailProps {
   run: Run;
   runs: Run[];
@@ -177,8 +182,18 @@ export function RunDetail({ run, runs, taskId, repoUrl, onClose, onRunUpdated, i
       <div className="px-6 pt-5 shrink-0">
         {/* Metadata row */}
         <div className="flex items-center justify-between mb-3">
-          <div className="flex items-center gap-3 text-xs text-[var(--color-text-secondary)]">
-            <Score value={run.score} className="text-sm font-semibold text-[var(--color-text)]" />
+          <div className="flex flex-wrap items-center gap-2 text-xs text-[var(--color-text-secondary)]">
+            <Score value={effectiveRunScore(fullRun ?? run)} className="text-sm font-semibold text-[var(--color-text)]" />
+            {(fullRun ?? run).verified_score != null && (fullRun ?? run).score != null && (fullRun ?? run).verified_score !== (fullRun ?? run).score && (
+              <span className="text-[var(--color-text-tertiary)]">
+                self-reported <Score value={(fullRun ?? run).score} className="inline text-xs font-medium text-[var(--color-text-tertiary)]" />
+              </span>
+            )}
+            {(fullRun ?? run).verification_status && (fullRun ?? run).verification_status !== "none" && (
+              <span className="px-1.5 py-0.5 rounded text-[10px] font-semibold uppercase tracking-wide bg-[var(--color-layer-2)] text-[var(--color-text-secondary)] border border-[var(--color-border)]">
+                {(fullRun ?? run).verification_status}
+              </span>
+            )}
             {!isValid && (
               <span className="px-1.5 py-0.5 rounded text-[10px] font-semibold uppercase tracking-wide bg-red-500/10 text-red-500 border border-red-500/20">
                 Invalid
@@ -342,7 +357,7 @@ export function RunDetail({ run, runs, taskId, repoUrl, onClose, onRunUpdated, i
                         <>
                           <div className="flex items-center gap-1.5">
                             <div className="w-2 h-2 rounded-full shrink-0" style={{ backgroundColor: color }} />
-                            <Score value={ancestor.score} decimals={2} className={`text-xs font-semibold ${isSelected ? "text-[var(--color-text)]" : "text-[var(--color-text)]"}`} />
+                            <Score value={effectiveRunScore(ancestor)} decimals={2} className={`text-xs font-semibold ${isSelected ? "text-[var(--color-text)]" : "text-[var(--color-text)]"}`} />
                           </div>
                           <span className={`text-[10px] leading-none whitespace-nowrap ${isSelected ? "text-[var(--color-text-secondary)]" : "text-[var(--color-text-tertiary)]"}`}>
                             {ancestor.agent_id}
diff --git a/ui/src/types/api.ts b/ui/src/types/api.ts
index 31b1259..a86aa8c 100644
--- a/ui/src/types/api.ts
+++ b/ui/src/types/api.ts
@@ -33,6 +33,13 @@ export interface Run {
   message: string;
   score: number | null;
   verified: boolean;
+  verified_score?: number | null;
+  verified_metric_key?: string | null;
+  verified_metric_value?: number | null;
+  verification_status?: string | null;
+  verification_mode?: string | null;
+  verification_error?: string | null;
+  verified_at?: string | null;
   valid?: boolean;
   created_at: string;
   post_id?: number;
@@ -91,7 +98,20 @@ export type FeedItem = ResultFeedItem | PostFeedItem | ClaimFeedItem;
 // Leaderboard response types (GET /tasks/:id/runs with different views)
 export interface BestRunsResponse {
   view: "best_runs";
-  runs: Pick<Run, "id" | "agent_id" | "branch" | "parent_id" | "tldr" | "score" | "verified" | "created_at" | "fork_url">[];
+  runs: Pick<
+    Run,
+    | "id"
+    | "agent_id"
+    | "branch"
+    | "parent_id"
+    | "tldr"
+    | "score"
+    | "verified_score"
+    | "verification_status"
+    | "verified"
+    | "created_at"
+    | "fork_url"
+  >[];
 }
 
 export interface ContributorEntry {