fix(node): bound GraphQL query cost (#552)

* fix(node): bound GraphQL query cost

* fix(node)!: bound repo names so emitted cursors always parse

The reposPage cursor embeds base64(JSON(owner_did, name)) while
parse_repo_cursor rejects cursors over 4096 bytes, but repo names were
unbounded on every write path. A single long-named visible repo made the
server emit an endCursor it then rejected as "invalid repository cursor",
dead-ending pagination for every caller past that row.

Bound names to MAX_REPO_NAME_LEN (100 bytes) at the create_repo and
fork_repo handlers via a shared validate_repo_name helper (bad charset is
still a 400), and reject overlong peer-supplied names in
Db::upsert_mirror_repo (the sync caller already drops the error and skips
the mirror).

Also in this round:
- docs/graphql-pagination.md states the 200 page-size maximum and corrects
  the over-budget remedy: a smaller limit does not reduce the reposPage or
  repos price, which is charged at the full MAX_VISIBLE_REPO_PAGE_SIZE
  scan regardless of limit.
- Tests pin refUpdates(limit: -1) clamping to zero rows, the
  parse_repo_cursor wrong-shape-JSON arm with its exact "invalid
  repository cursor" message, the longest-admissible-name cursor
  round-trip, and the mirror write-path rejection.

BREAKING CHANGE: the GraphQL `repos` field now returns an error instead of
the full list when more than MAX_VISIBLE_REPO_PAGE_SIZE (200) repositories
are visible; larger lists must use `reposPage` with `limit`/`after`.
Repository names are additionally limited to 100 bytes at creation, fork,
and mirror-registration time.

* fix(node): address beardthelion P3 findings on name bound and input messages

- Assert each rejection's own message in the reposPage invalid-inputs
  loop: the loop no longer accepts either message for every query, so a
  misattributed message (limit arm returning the cursor message) goes red.
- Pin the create/fork name-bound wiring through the handler: new test
  drives create_repo with an overlong name via test_state_lazy and asserts
  the 400; deleting the validate_repo_name call site would proceed to the
  lazy DB pool and fail with a connection error instead.
- Document the API-side vs disk-side validate_repo_name divergence on the
  new function, and tie the store's literal 100 to MAX_REPO_NAME_LEN so
  the two validators cannot drift apart on length.

---------

Co-authored-by: Marlowe <[email protected]>
Co-authored-by: cairn-intern <[email protected]>

CONTRIBUTING.md

@@ -81,7 +81,7 @@ time once updated.
 
 **Requirements:**
 - Rust stable (≥ 1.91) — install via [rustup](https://rustup.rs)
-- PostgreSQL — required for the node. Use the bundled `docker-compose.yml` for local dev.
+- PostgreSQL 16+ — required for the node (it uses `pg_input_is_valid`, which exists only on PostgreSQL 16 and later). Use the bundled `docker-compose.yml` for local dev.
 - Docker (optional, for full-stack local testing)
 
 **Environment variables:**

README.md

@@ -53,6 +53,7 @@ Good today:
 
 - Local or Docker node startup.
 - Postgres-backed repo metadata.
+- Bounded GraphQL queries with [repository pagination](docs/graphql-pagination.md).
 - Bare git repository storage.
 - Git smart-HTTP clone/fetch/push.
 - RFC 9421-signed writes.
@@ -311,6 +312,7 @@ metadata        local disk / optional S3
 | DID | A user, agent, or node identity derived from an Ed25519 public key. |
 | HTTP Signature | RFC 9421 signature proving control of the DID key for write requests. |
 | Ref certificate | Signed record of a ref update. Useful for audit and replication. |
+| Ref certificate versioning | The wire format is versioned; new certificates are currently issued as v1. `gl cert show <repo> <id> --verify` requires `--expect-node <did>`, and the queried node's self-reported DID is **not** accepted as a trust anchor. v2 is reserved for a future payload shape that carries the version inside the signed bytes; this client refuses to verify v2+ rather than guessing. |
 | UCAN | Delegation token for future capability-based workflows. |
 | Peer announce | Node-to-node HTTP announcement of DID + public URL. |
 | Gossipsub | libp2p topic for ref-update events. |

crates/git-remote-gitlawb/tests/real_git_fetch.rs

@@ -137,6 +137,11 @@ enum ShimMode {
 struct Shim {
     base_url: String,
     posts: Arc<AtomicUsize>,
+    /// Advertisement (GET /info/refs) count. One `git fetch` performs exactly
+    /// one advertisement, so tests assert this is EXACTLY one: any fetch-level
+    /// retry that comes back re-advertises and goes red on the spot, which the
+    /// lower-bound POST assertions alone cannot see (#275 round 6).
+    gets: Arc<AtomicUsize>,
     stop: Arc<AtomicBool>,
     handle: Option<std::thread::JoinHandle<()>>,
 }
@@ -166,15 +171,17 @@ fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim {
     let addr = listener.local_addr().unwrap();
     let base_url = format!("http://{addr}");
     let posts = Arc::new(AtomicUsize::new(0));
+    let gets = Arc::new(AtomicUsize::new(0));
     let stop = Arc::new(AtomicBool::new(false));
 
     let posts_t = posts.clone();
+    let gets_t = gets.clone();
     let stop_t = stop.clone();
     let handle = std::thread::spawn(move || {
         while !stop_t.load(Ordering::SeqCst) {
             match listener.accept() {
                 Ok((stream, _)) => {
-                    handle_conn(stream, &repo, mode, &posts_t);
+                    handle_conn(stream, &repo, mode, &posts_t, &gets_t);
                 }
                 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                     std::thread::sleep(Duration::from_millis(5));
@@ -187,6 +194,7 @@ fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim {
     Shim {
         base_url,
         posts,
+        gets,
         stop,
         handle: Some(handle),
     }
@@ -212,7 +220,13 @@ fn normalize_accepted_stream(stream: &TcpStream) {
     stream.set_read_timeout(Some(Duration::from_secs(30))).ok();
 }
 
-fn handle_conn(stream: TcpStream, repo: &Path, mode: ShimMode, posts: &AtomicUsize) {
+fn handle_conn(
+    stream: TcpStream,
+    repo: &Path,
+    mode: ShimMode,
+    posts: &AtomicUsize,
+    gets: &AtomicUsize,
+) {
     normalize_accepted_stream(&stream);
     let mut reader = BufReader::new(stream);
 
@@ -263,6 +277,7 @@ fn handle_conn(stream: TcpStream, repo: &Path, mode: ShimMode, posts: &AtomicUsi
             receive_pack(repo, false, &body),
         )
     } else if method == "GET" && target.contains("/info/refs") {
+        gets.fetch_add(1, Ordering::SeqCst);
         // v0 advertisement, wrapped exactly as the node's info_refs does.
         let adv = upload_pack(repo, true, b"");
         let mut wrapped = pkt(b"# service=git-upload-pack\n");
@@ -1007,6 +1022,7 @@ fn shim_answers_a_connection_that_arrives_non_blocking() {
 
     let server = std::thread::spawn(move || {
         let posts = AtomicUsize::new(0);
+        let gets = AtomicUsize::new(0);
         let stream = loop {
             match listener.accept() {
                 Ok((s, _)) => break s,
@@ -1021,7 +1037,7 @@ fn shim_answers_a_connection_that_arrives_non_blocking() {
         // the assertions below run on every platform.
         stream.set_nonblocking(true).unwrap();
         accepted_tx.send(()).unwrap();
-        handle_conn(stream, &repo, ShimMode::Normal, &posts);
+        handle_conn(stream, &repo, ShimMode::Normal, &posts, &gets);
     });
 
     let mut client = TcpStream::connect(addr).unwrap();
@@ -1085,6 +1101,17 @@ fn real_git_multi_round_fetch_completes() {
         posts >= 2,
         "fixture did not force multi-round negotiation (observed {posts} POST(s)); the bridging path was not exercised"
     );
+    // Exactly one advertisement: the POST bound above is a floor, so a
+    // reintroduced fetch-level retry could satisfy it by running a second full
+    // fetch instead of multi-round negotiation. Re-advertising is what a
+    // second fetch cannot avoid, so this is the committed guard on the
+    // retry's absence.
+    let gets = shim.gets.load(Ordering::SeqCst);
+    assert_eq!(
+        gets, 1,
+        "expected exactly one advertisement GET per fetch; {gets} means a \
+         fetch-level retry is back"
+    );
 
     // The fetched tip is present and the clone's object graph is intact.
     let server_head = String::from_utf8(git(&server, &["rev-parse", "HEAD"])).unwrap();
@@ -1177,6 +1204,17 @@ fn real_git_withheld_shaped_first_post() {
         "helper hung on a withheld-shaped response (should forward-and-terminate, not deadlock). stderr:\n{stderr}"
     );
 
+    // Exactly one advertisement, pass or fail: the POST assertions below are
+    // floors, so a reintroduced fetch-level retry could mask the withheld
+    // shape behind a second full fetch. A retry cannot avoid re-advertising,
+    // so this is the committed guard on its absence.
+    let gets = shim.gets.load(Ordering::SeqCst);
+    assert_eq!(
+        gets, 1,
+        "expected exactly one advertisement GET per fetch; {gets} means a \
+         fetch-level retry is back"
+    );
+
     if out.status.success() {
         // Real git accepted the mid-negotiation pack: the withheld multi-round
         // path works end to end with no extra handling.

crates/gitlawb-node/src/api/certs.rs

@@ -52,6 +52,7 @@ pub async fn list_certs(
                 "node_did":   c.node_did,
                 "signature":  c.signature,
                 "issued_at":  c.issued_at,
+                "version":    c.version,
             })
         })
         .collect();
@@ -92,5 +93,6 @@ pub async fn get_cert(
         "node_did":   cert.node_did,
         "signature":  cert.signature,
         "issued_at":  cert.issued_at,
+        "version":    cert.version,
     })))
 }

crates/gitlawb-node/src/api/events.rs

@@ -11,7 +11,8 @@ use crate::state::AppState;
 
 /// Hard ceiling on rows any ref-update feed returns in one request. Shared by the
 /// shared collector's clamp and the per-handler request caps so they can't drift.
-const MAX_VISIBLE_REF_UPDATES: i64 = 200;
+/// `pub(crate)` so the GraphQL complexity meter can price against the same bound.
+pub(crate) const MAX_VISIBLE_REF_UPDATES: i64 = 200;
 
 /// Collect up to `limit` ref-update rows visible to `caller`, newest first,
 /// paging past rows the feed gate drops. Filtering after a plain SQL `LIMIT`
@@ -437,6 +438,7 @@ mod ref_updates_feed_tests {
             node_did: "did:key:z6MkNode".into(),
             signature: "sig".into(),
             issued_at: Utc::now().to_rfc3339(),
+            version: 1,
         }
     }
 

crates/gitlawb-node/src/api/ipfs.rs

@@ -8367,6 +8367,7 @@ mod tests {
     #[tokio::test]
     async fn get_by_cid_per_source_cap_sheds_same_source_admits_other() {
         let mut state = crate::test_support::test_state_lazy();
+        state.db.pool().close().await;
         // Global pool has room; the per-source cap is 1.
         state.git_ipfs_walk_semaphore = Arc::new(Semaphore::new(8));
         state.git_ipfs_walk_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100);
@@ -8393,16 +8394,23 @@ mod tests {
             "a source at its per-source /ipfs walk cap must shed 503 with global capacity free"
         );
 
+        let bytes = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
+        let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+        assert_eq!(body["error"], "overloaded");
+
         // A DIFFERENT source is NOT shed by the per-source cap: it clears admission and
-        // proceeds (then errors on the lazy DB, which is not a 503).
+        // proceeds to the closed DB, which has a distinct db_unavailable error code.
         let resp = ipfs_router(state)
             .oneshot(get_cid(&cid, Some(other)))
             .await
             .unwrap();
-        assert_ne!(
-            resp.status(),
-            StatusCode::SERVICE_UNAVAILABLE,
-            "a different source must not be shed by the per-source cap"
+        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let bytes = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
+        let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+        assert_eq!(
+            body["error"],
+            crate::error::DB_UNAVAILABLE_CODE,
+            "a different source must clear admission and reach the closed database"
         );
     }
 

crates/gitlawb-node/src/api/repos.rs

@@ -247,6 +247,40 @@ pub struct InfoRefsQuery {
 
 // ── Handlers ──────────────────────────────────────────────────────────────
 
+/// Validate a repository name at the create/fork write paths. The charset is
+/// restricted, and the length is bounded so every cursor the server emits
+/// parses back: `repo_cursor` embeds the name in a base64(JSON) cursor that
+/// `parse_repo_cursor` rejects over 4096 bytes (#465 4th round).
+///
+/// This is the API-side validator, and it deliberately differs from the
+/// stricter disk-side `validate_repo_name` in `git/repo_store.rs`: this one
+/// admits names the store rejects (empty, leading `-`, non-ASCII
+/// alphanumeric) and rejects `.`, which the store allows. The divergence
+/// predates this bound (#412 owns the fork half), so a store-rejected name
+/// still fails later at `repo_store.init` after the proof is spent. Do not
+/// "fix" the divergence here; the length bound below is the one that keeps
+/// emitted cursors parseable, and it is shared via `MAX_REPO_NAME_LEN` so the
+/// two validators cannot drift apart on length.
+fn validate_repo_name(name: &str) -> Result<()> {
+    // Sanitize name: alphanumeric, hyphens, underscores only
+    if !name
+        .chars()
+        .all(|c| c.is_alphanumeric() || c == '-' || c == '_')
+    {
+        return Err(AppError::BadRequest(
+            "repo name must contain only alphanumeric characters, hyphens, and underscores".into(),
+        ));
+    }
+    // Byte length, not char count: the cursor budget is measured in bytes.
+    if name.len() > crate::db::MAX_REPO_NAME_LEN {
+        return Err(AppError::BadRequest(format!(
+            "repo name must be at most {} bytes",
+            crate::db::MAX_REPO_NAME_LEN
+        )));
+    }
+    Ok(())
+}
+
 /// POST /api/v1/repos
 /// Create a new repository. Requires HTTP Signature auth.
 pub async fn create_repo(
@@ -261,16 +295,7 @@ pub async fn create_repo(
     // rejected request (bad name, already exists) never burns a valid proof.
     let proof = crate::icaptcha::verify_request(&headers, &auth.0)?;
 
-    // Sanitize name: alphanumeric, hyphens, underscores only
-    if !req
-        .name
-        .chars()
-        .all(|c| c.is_alphanumeric() || c == '-' || c == '_')
-    {
-        return Err(AppError::BadRequest(
-            "repo name must contain only alphanumeric characters, hyphens, and underscores".into(),
-        ));
-    }
+    validate_repo_name(&req.name)?;
 
     // default_branch is caller-supplied and becomes a PR's target_branch when the
     // PR omits one, which is interpolated into a git revision argument. Validate
@@ -3774,15 +3799,7 @@ pub async fn fork_repo(
     let fork_name = req.name.unwrap_or_else(|| source.name.clone());
     let forker_did = auth.0;
 
-    // Validate fork name
-    if !fork_name
-        .chars()
-        .all(|c| c.is_alphanumeric() || c == '-' || c == '_')
-    {
-        return Err(AppError::BadRequest(
-            "repo name must contain only alphanumeric characters, hyphens, and underscores".into(),
-        ));
-    }
+    validate_repo_name(&fork_name)?;
 
     // Same barrier every other repo-creation route goes through. The character
     // allowlist above is vacuously true on an empty name, and `repo_disk_path`
@@ -5049,6 +5066,50 @@ mod tests {
         peer_mock.assert_async().await;
     }
 
+    #[test]
+    fn repo_name_validation_rejects_bad_charset_and_overlong_names() {
+        assert!(validate_repo_name("ok-name_123").is_ok());
+        assert!(validate_repo_name(&"a".repeat(crate::db::MAX_REPO_NAME_LEN)).is_ok());
+        // The bound is in bytes: the cursor budget is measured in bytes.
+        let overlong = "a".repeat(crate::db::MAX_REPO_NAME_LEN + 1);
+        let err = validate_repo_name(&overlong).unwrap_err();
+        assert!(
+            matches!(err, AppError::BadRequest(_)),
+            "overlong name must be a 400, got: {err:?}"
+        );
+        assert!(validate_repo_name("bad name!").is_err());
+        // Empty names were already admissible before the length bound; the
+        // bound does not change that.
+        assert!(validate_repo_name("").is_ok());
+    }
+
+    #[tokio::test]
+    async fn create_repo_rejects_overlong_name_through_handler() {
+        // Pins the wiring: `create_repo` must call `validate_repo_name`
+        // before any database access. The lazy state never connects, so
+        // deleting the call site would let the handler proceed to `get_repo`
+        // and fail with a connection error instead of BadRequest.
+        let state = crate::test_support::test_state_lazy();
+        let overlong = "a".repeat(crate::db::MAX_REPO_NAME_LEN + 1);
+        let err = create_repo(
+            State(state),
+            Extension(AuthenticatedDid(OWNER_DID.to_owned())),
+            axum::http::HeaderMap::new(),
+            Json(CreateRepoRequest {
+                name: overlong,
+                description: None,
+                is_public: true,
+                default_branch: "main".into(),
+            }),
+        )
+        .await
+        .unwrap_err();
+        assert!(
+            matches!(err, AppError::BadRequest(ref msg) if msg.contains("at most")),
+            "overlong name through create_repo must be a 400, got: {err:?}"
+        );
+    }
+
     #[test]
     fn upload_pack_request_finalizes_only_with_done_pktline() {
         let want = "0032want 1111111111111111111111111111111111111111\n";
@@ -9179,6 +9240,7 @@ mod tests {
     /// A pkt-line receive-pack body carrying one branch-create ref update, so the
     /// handler's post-receive tail resolves a non-empty new-tip set (the delta
     /// scan's git stages run).
+    #[cfg(unix)]
     fn ref_update_body(new_sha: &str) -> axum::body::Bytes {
         let line = format!("{ZERO_SHA} {new_sha} refs/heads/main");
         axum::body::Bytes::from(format!("{:04x}{}0000", line.len() + 4, line))
@@ -11636,6 +11698,7 @@ mod tests {
         )
     }
 
+    #[cfg(unix)]
     fn f2a_log(log: &std::path::Path) -> String {
         std::fs::read_to_string(log).unwrap_or_default()
     }
@@ -11643,6 +11706,7 @@ mod tests {
     /// Withheld-walk children run so far. `ls-tree` is the walk's signature child
     /// (`blob_paths` lists every reachable commit's tree); the delta scan and the
     /// full-scan fallback use `rev-list` / `cat-file` instead.
+    #[cfg(unix)]
     fn f2a_walks(log: &std::path::Path) -> usize {
         f2a_log(log)
             .lines()
@@ -11654,6 +11718,7 @@ mod tests {
     /// the withheld walk actually runs rather than taking the no-rule shortcut).
     /// The repo's on-disk path is passed to the tail directly, so no repo_store or
     /// receive-pack plumbing is involved.
+    #[cfg(unix)]
     async fn f2a_state(
         pool: sqlx::PgPool,
         git_bin: &str,
@@ -11685,6 +11750,7 @@ mod tests {
         (state, rec)
     }
 
+    #[cfg(unix)]
     fn f2a_update(ref_name: &str, new_sha: &str) -> Vec<RefUpdate> {
         vec![RefUpdate {
             old_sha: ZERO_SHA.to_string(),
@@ -11693,6 +11759,7 @@ mod tests {
         }]
     }
 
+    #[cfg(unix)]
     const F2A_PUSHER: &str = "did:key:z6MkF2aPusherAAAAAAAAAAAAAAAAAAAAAAAAAA";
 
     /// When publish durability is fenced, the tail must not run walks or take
@@ -11805,6 +11872,7 @@ mod tests {
     }
     /// Poll `cond` until it holds, with a bound so a regression fails the test
     /// rather than hanging the suite.
+    #[cfg(unix)]
     async fn f2a_wait_for(mut cond: impl FnMut() -> bool, what: &str) {
         let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
         while !cond() {
@@ -11819,6 +11887,7 @@ mod tests {
     /// A `rev-list --objects` line names the tips a DELTA scan was asked to resolve,
     /// so it attributes that scan to one push's tips. The withheld walk's own
     /// `rev-list --all` / `ls-tree` lines never carry a tip as an argument this way.
+    #[cfg(unix)]
     fn f2a_delta_scanned(log: &std::path::Path, tip: &str) -> bool {
         f2a_log(log)
             .lines()
@@ -11967,6 +12036,7 @@ mod tests {
 
     /// Mount a Pinata upload endpoint that assigns every object the same CID, and
     /// point the state at it. Returns the server (kept alive by the caller) and CID.
+    #[cfg(unix)]
     async fn f2a_pinata(state: &mut AppState) -> (mockito::ServerGuard, String) {
         let cid = "bafyf2acoalescedmapping".to_string();
         let mut server = mockito::Server::new_async().await;
@@ -11986,6 +12056,7 @@ mod tests {
 
     /// Poll the branch to CID table until the push's mapping lands (the Pinata
     /// worker is detached), bounded so a regression fails rather than hangs.
+    #[cfg(unix)]
     async fn f2a_wait_for_branch_cid(
         db: &crate::db::Db,
         slug: &str,
@@ -12005,6 +12076,7 @@ mod tests {
         }
     }
 
+    #[cfg(unix)]
     fn f2a_slug(rec: &crate::db::RepoRecord) -> String {
         format!(
             "{}/{}",
@@ -12139,6 +12211,7 @@ mod tests {
     // and `z6p2fail`), and owner-only push is on by default, so the identity has to
     // follow the repo each push targets rather than being fixed for both.
 
+    #[cfg(unix)]
     fn p2_push(
         state: &AppState,
         owner: &str,
@@ -12157,6 +12230,7 @@ mod tests {
         )
     }
 
+    #[cfg(unix)]
     fn p2_logged(log: &std::path::Path, prefix: &str) -> bool {
         f2a_log(log).lines().any(|l| l.starts_with(prefix))
     }
@@ -13152,6 +13226,7 @@ exit 0
     /// runs exactly one `rev-list --all`, so this counts the walks that were attempted
     /// (the `ls-tree` counter above cannot: a walk whose enumeration fails never gets
     /// to `ls-tree`).
+    #[cfg(unix)]
     fn f2b_walk_attempts(log: &std::path::Path) -> usize {
         f2a_log(log)
             .lines()

crates/gitlawb-node/src/api/tasks.rs

@@ -175,7 +175,7 @@ pub async fn create_task(
         updated_at: now,
         deadline: body.deadline,
     };
-    state.db.create_task(&task).await.map_err(db_error)?;
+    let task = state.db.create_task(&task).await.map_err(db_error)?;
     Ok((StatusCode::CREATED, Json(task_to_json(&task))))
 }
 
@@ -228,16 +228,28 @@ pub async fn claim_task(
         return Err(forbidden("assignee_did must be the authenticated signer"));
     }
     let task = state.db.claim_task(&id, &auth.0).await.map_err(|e| {
+        if e.downcast_ref::<crate::db::TaskReservedForOtherAssignee>()
+            .is_some()
+        {
+            return forbidden("task not claimable: reserved for another assignee");
+        }
         (
             StatusCode::CONFLICT,
             Json(json!({ "error": e.to_string() })),
         )
     })?;
+    let by_did = task
+        .assignee_did
+        .as_deref()
+        .map(crate::db::trim_assignee_did)
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .unwrap_or_else(|| auth.0.clone());
     let _ = state.task_event_tx.send(TaskEventBroadcast {
         task_id: id,
         old_status: "pending".to_string(),
         new_status: "claimed".to_string(),
-        by_did: auth.0,
+        by_did,
         at: Utc::now().to_rfc3339(),
     });
     Ok(Json(task_to_json(&task)))
@@ -272,14 +284,14 @@ pub async fn complete_task(
         })?;
     if !crate::api::did_matches(
         &auth.0,
-        existing.assignee_did.as_deref().unwrap_or_default(),
+        crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
     ) {
         return Err(forbidden("only the task assignee can complete it"));
     }
     let by_did = auth.0;
     let task = state
         .db
-        .finish_task(&id, "completed", body.result.as_deref())
+        .finish_task(&id, "completed", body.result.as_deref(), &by_did)
         .await
         .map_err(|e| {
             (
@@ -287,6 +299,13 @@ pub async fn complete_task(
                 Json(json!({ "error": e.to_string() })),
             )
         })?;
+    let by_did = task
+        .assignee_did
+        .as_deref()
+        .map(crate::db::trim_assignee_did)
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .unwrap_or(by_did);
     let _ = state.task_event_tx.send(TaskEventBroadcast {
         task_id: id,
         old_status: "claimed".to_string(),
@@ -325,7 +344,7 @@ pub async fn fail_task(
         })?;
     if !crate::api::did_matches(
         &auth.0,
-        existing.assignee_did.as_deref().unwrap_or_default(),
+        crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
     ) {
         return Err(forbidden("only the task assignee can fail it"));
     }
@@ -333,7 +352,7 @@ pub async fn fail_task(
     let reason = body.reason.unwrap_or_default();
     let task = state
         .db
-        .finish_task(&id, "failed", Some(&reason))
+        .finish_task(&id, "failed", Some(&reason), &by_did)
         .await
         .map_err(|e| {
             (
@@ -341,6 +360,13 @@ pub async fn fail_task(
                 Json(json!({ "error": e.to_string() })),
             )
         })?;
+    let by_did = task
+        .assignee_did
+        .as_deref()
+        .map(crate::db::trim_assignee_did)
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .unwrap_or(by_did);
     let _ = state.task_event_tx.send(TaskEventBroadcast {
         task_id: id,
         old_status: "claimed".to_string(),

crates/gitlawb-node/src/cert.rs

@@ -11,6 +11,74 @@ use uuid::Uuid;
 use crate::db::RefCertificate;
 use crate::state::AppState;
 
+/// Build the v1 signing payload — the EXACT bytes the node signs and
+/// gl's `verify_signature` reconstructs. This is the single source
+/// of truth shared by `issue_ref_certificate` and the frozen
+/// canonical-form tests below; drift here is what would break every
+/// existing cert.
+///
+/// Field order and key set are fixed: a default `serde_json::Value`
+/// serializes `Map<String, Value>` with sorted keys, so any change to
+/// the literal is observable as a different byte sequence on the
+/// wire.
+///
+/// v1 is the unversioned legacy shape: no `version` key in the signed
+/// JSON. From v2 onward the version belongs INSIDE the signed bytes
+/// (see `v2_signing_payload`): leaving it as an unsigned sibling
+/// would let a future v2 cert be downgraded to v1 and still verify
+/// cleanly, since the bytes would be identical.
+pub(crate) fn v1_signing_payload(
+    repo_id: &str,
+    ref_name: &str,
+    old_sha: &str,
+    new_sha: &str,
+    pusher_did: &str,
+    node_did: &str,
+    issued_at: &str,
+) -> serde_json::Value {
+    serde_json::json!({
+        "repo_id": repo_id,
+        "ref":     ref_name,
+        "old":     old_sha,
+        "new":     new_sha,
+        "pusher":  pusher_did,
+        "node":    node_did,
+        "ts":      issued_at,
+    })
+}
+
+/// Build the v2 signing payload — the future shape for versioned
+/// certs. Identical to v1 except the wire `version` is part of the
+/// signed bytes, so stripping or flipping the `version` column
+/// invalidates the signature instead of verifying cleanly under the
+/// other version's path.
+///
+/// NOT yet issued: `issue_ref_certificate` still stamps v1 because
+/// the shipped `gl` verifier only supports v1. This builder exists
+/// to pin the downgrade-resistant shape now, before any v2 cert is
+/// ever signed.
+#[allow(dead_code)]
+pub(crate) fn v2_signing_payload(
+    repo_id: &str,
+    ref_name: &str,
+    old_sha: &str,
+    new_sha: &str,
+    pusher_did: &str,
+    node_did: &str,
+    issued_at: &str,
+) -> serde_json::Value {
+    serde_json::json!({
+        "repo_id": repo_id,
+        "ref":     ref_name,
+        "old":     old_sha,
+        "new":     new_sha,
+        "pusher":  pusher_did,
+        "node":    node_did,
+        "ts":      issued_at,
+        "version": 2,
+    })
+}
+
 /// Issue a signed ref-update certificate for a successful push.
 ///
 /// Builds a canonical JSON payload, signs it with the node's Ed25519 key,
@@ -26,16 +94,12 @@ pub async fn issue_ref_certificate(
     let node_did = state.node_did.to_string();
     let issued_at = Utc::now().to_rfc3339();
 
-    // Build the canonical signing payload.
-    let payload = serde_json::json!({
-        "repo_id": repo_id,
-        "ref":     ref_name,
-        "old":     old_sha,
-        "new":     new_sha,
-        "pusher":  pusher_did,
-        "node":    node_did,
-        "ts":      issued_at,
-    });
+    // Build the canonical signing payload via the shared builder so
+    // the frozen-vector test below and the live signer cannot drift
+    // apart — a regression in either side fails both.
+    let payload = v1_signing_payload(
+        repo_id, ref_name, old_sha, new_sha, pusher_did, &node_did, &issued_at,
+    );
     let payload_bytes = serde_json::to_vec(&payload)?;
 
     let signature = state.node_keypair.sign_b64(&payload_bytes);
@@ -50,9 +114,351 @@ pub async fn issue_ref_certificate(
         node_did,
         signature,
         issued_at,
+        // #26 Split PR 3: the wire-format version. New certs stamp
+        // v1 — the pre-versioning 7-field payload with no `version`
+        // key in the signed JSON — because the shipped `gl`
+        // verifier only supports v1. Stamping v2 here would issue
+        // certificates the shipped client refuses (round-3
+        // finding: node pinned `version == 2` while gl pinned
+        // "2 is refused", with nothing exercising the
+        // composition). v2 is reserved for the future shape
+        // defined by `v2_signing_payload`, which carries the
+        // version INSIDE the signed bytes; land the v2 verify
+        // path before stamping v2.
+        version: 1,
     };
 
     // Persist and return the row as it exists in the database (on a
     // conflict the existing row survives when it is newer).
     state.db.insert_ref_certificate(&cert).await
 }
+
+#[cfg(test)]
+mod v1_payload_tests {
+    //! #26 Split PR 3 — cert payload version compat.
+    //!
+    //! The v1 cert payload is the frozen canonical byte form the
+    //! node signs. gl's `verify_signature` reconstructs the same
+    //! payload and verifies. Any drift in field order, key set, or
+    //! whitespace breaks every existing cert. The tests here pin
+    //! the v1 payload shape, the round-trip, the cert id
+    //! determinism per the v1 version, and the v2 read forward
+    //! compat (an unknown version refuses to verify rather than
+    //! guessing).
+    //!
+    //! Reviewer 1 finding: the live signer (`issue_ref_certificate`)
+    //! and the frozen-vector test (`v1_payload_matches_frozen_canonical_form`)
+    //! previously held two separate `serde_json::json!` literals,
+    //! so a regression in the signer could not fail this test.
+    //! Both now go through the shared `v1_signing_payload` builder.
+    use super::{v1_signing_payload, v2_signing_payload};
+    use crate::db::RefCertificate;
+    use gitlawb_core::identity::Keypair;
+
+    /// v1 payload is the same shape the legacy code signed, byte
+    /// for byte. Reverting this assertion to the v2 shape (which
+    /// includes a `version` key) is what the versioned format
+    /// forbids: an old client reading a v1 cert must not see a
+    /// `version` key in the signed JSON.
+    ///
+    /// The literal is built via the shared `v1_signing_payload`
+    /// helper that `issue_ref_certificate` also uses, so the live
+    /// signer and the frozen vector share one source of truth.
+    #[test]
+    fn v1_payload_matches_frozen_canonical_form() {
+        let payload = v1_signing_payload(
+            "repo-1",
+            "refs/heads/main",
+            &"0".repeat(40),
+            &"a".repeat(40),
+            "did:key:z6MkPusher",
+            "did:key:z6MkNode",
+            "2026-07-22T00:00:00+00:00",
+        );
+        let frozen = format!(
+            r#"{{"new":"{}","node":"did:key:z6MkNode","old":"{}","pusher":"did:key:z6MkPusher","ref":"refs/heads/main","repo_id":"repo-1","ts":"2026-07-22T00:00:00+00:00"}}"#,
+            "a".repeat(40),
+            "0".repeat(40),
+        );
+        assert_eq!(serde_json::to_string(&payload).unwrap(), frozen);
+    }
+
+    /// Round-trip: sign the v1 payload with a keypair, build a
+    /// RefCertificate with version: 1, and verify the structure.
+    /// This is the construction the live code does on every push.
+    ///
+    /// Uses the shared builder for both halves so a regression in
+    /// the live signer (adding/removing/renaming a field) cannot
+    /// pass this test by accident.
+    #[test]
+    fn v1_ref_certificate_structure_is_well_formed() {
+        let kp = Keypair::generate();
+        let node_did = kp.did().as_str().to_string();
+        let payload = v1_signing_payload(
+            "repo-1",
+            "refs/heads/main",
+            &"0".repeat(40),
+            &"a".repeat(40),
+            "did:key:z6MkPusher",
+            &node_did,
+            "2026-07-22T00:00:00+00:00",
+        );
+        let payload_bytes = serde_json::to_vec(&payload).unwrap();
+        let sig = kp.sign_b64(&payload_bytes);
+        let cert = RefCertificate {
+            id: "cert-id".into(),
+            repo_id: "repo-1".into(),
+            ref_name: "refs/heads/main".into(),
+            old_sha: "0".repeat(40),
+            new_sha: "a".repeat(40),
+            pusher_did: "did:key:z6MkPusher".into(),
+            node_did: node_did.clone(),
+            signature: sig.clone(),
+            issued_at: "2026-07-22T00:00:00+00:00".into(),
+            version: 1,
+        };
+        assert_eq!(cert.version, 1, "new certs carry version: 1");
+        // The signed payload reconstructs identically: gl's
+        // verify_signature would build the same JSON, hash the
+        // same bytes, and verify the same signature.
+        let reconstructed = v1_signing_payload(
+            &cert.repo_id,
+            &cert.ref_name,
+            &cert.old_sha,
+            &cert.new_sha,
+            &cert.pusher_did,
+            &cert.node_did,
+            &cert.issued_at,
+        );
+        let reconstructed_bytes = serde_json::to_vec(&reconstructed).unwrap();
+        assert_eq!(
+            reconstructed_bytes, payload_bytes,
+            "the round-trip serialization must be byte-identical; \
+             this is what makes gl's verify_signature succeed"
+        );
+        // Round 3 (P3 reviewer): byte-equality alone never calls
+        // `identity::verify`, so a bad signature helper would pass.
+        // Verify the computed signature against the reconstructed
+        // bytes — the same check gl performs.
+        let vk = node_did
+            .parse::<gitlawb_core::did::Did>()
+            .unwrap()
+            .to_verifying_key()
+            .unwrap();
+        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
+        let sig_bytes: [u8; 64] = URL_SAFE_NO_PAD
+            .decode(&cert.signature)
+            .unwrap()
+            .try_into()
+            .unwrap();
+        gitlawb_core::identity::verify(&vk, &reconstructed_bytes, &sig_bytes)
+            .expect("well-formed v1 cert signature must verify");
+    }
+
+    /// Reviewer 1 finding: the live signer must share the
+    /// canonical-form literal so a regression in either side
+    /// fails both. This test pins the shared builder's output
+    /// against the literal a future v2 byte stream would have
+    /// to break — if someone adds a `version` key to
+    /// `v1_signing_payload`, every existing cert breaks verify,
+    /// and this test is one of the canaries that fires.
+    #[test]
+    fn v1_signing_payload_has_no_version_key() {
+        let payload = v1_signing_payload(
+            "repo-1",
+            "refs/heads/main",
+            "oldsha",
+            "newsha",
+            "did:key:z6MkPusher",
+            "did:key:z6MkNode",
+            "2026-07-22T00:00:00+00:00",
+        );
+        let obj = payload.as_object().expect("payload is a JSON object");
+        assert!(
+            !obj.contains_key("version"),
+            "v1 signing payload must not carry a `version` key — adding \
+             one changes the bytes and breaks every existing cert's \
+             signature. v2 certs build a different payload, not this one"
+        );
+        // Pin the exact set so a future field addition is caught.
+        let expected: std::collections::BTreeSet<&str> =
+            ["new", "node", "old", "pusher", "ref", "repo_id", "ts"]
+                .into_iter()
+                .collect();
+        let actual: std::collections::BTreeSet<&str> = obj.keys().map(String::as_str).collect();
+        assert_eq!(actual, expected, "v1 payload key set is frozen");
+    }
+
+    /// Round 3 (P2 reviewer): from v2 onward the version belongs
+    /// INSIDE the signed bytes. v1 stays the unversioned legacy
+    /// shape; v2 carries `"version": 2` in the payload so a
+    /// version flip without re-signing invalidates the signature
+    /// instead of verifying cleanly under the other path.
+    #[test]
+    fn v2_signing_payload_binds_version_and_resists_downgrade() {
+        let kp = Keypair::generate();
+        let node_did = kp.did().as_str().to_string();
+        let ts = "2026-07-22T00:00:00+00:00";
+        let v1 = v1_signing_payload(
+            "repo-1",
+            "refs/heads/main",
+            "oldsha",
+            "newsha",
+            "did:key:z6MkPusher",
+            &node_did,
+            ts,
+        );
+        let v2 = v2_signing_payload(
+            "repo-1",
+            "refs/heads/main",
+            "oldsha",
+            "newsha",
+            "did:key:z6MkPusher",
+            &node_did,
+            ts,
+        );
+        let v2_obj = v2.as_object().expect("v2 payload is a JSON object");
+        assert_eq!(
+            v2_obj.get("version"),
+            Some(&serde_json::json!(2)),
+            "v2 payload must carry the version inside the signed bytes"
+        );
+        let v1_bytes = serde_json::to_vec(&v1).unwrap();
+        let v2_bytes = serde_json::to_vec(&v2).unwrap();
+        assert_ne!(
+            v1_bytes, v2_bytes,
+            "v1 and v2 payloads must differ, otherwise a version flip is cryptographically invisible"
+        );
+
+        // A signature over v1 must NOT verify as v2 and vice versa.
+        let vk = node_did
+            .parse::<gitlawb_core::did::Did>()
+            .unwrap()
+            .to_verifying_key()
+            .unwrap();
+        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
+        let sig_v1 = kp.sign_b64(&v1_bytes);
+        let sig_v1_bytes: [u8; 64] = URL_SAFE_NO_PAD.decode(&sig_v1).unwrap().try_into().unwrap();
+        assert!(
+            gitlawb_core::identity::verify(&vk, &v1_bytes, &sig_v1_bytes).is_ok(),
+            "v1 signature must verify under v1 bytes"
+        );
+        assert!(
+            gitlawb_core::identity::verify(&vk, &v2_bytes, &sig_v1_bytes).is_err(),
+            "a v1 signature presented as v2 (downgrade/upgrade flip) must not verify"
+        );
+    }
+
+    /// The live issuer stamps the version the shipped client can
+    /// verify. The gl client refuses to verify v2 certs explicitly;
+    /// this test pins that the version the ISSUING PATH claims
+    /// agrees with the payload shape it actually signs.
+    ///
+    /// Round 2: the prior form built a `RefCertificate` literal with
+    /// `version: 1` and asserted `version == 1` — true no matter what the
+    /// signer does, so flipping the issuer's version (or its payload shape)
+    /// left it green. This one drives the actual
+    /// `issue_ref_certificate` so a regression in the issuer flips the test.
+    /// The DB row is created against `test_support::test_state`, which
+    /// gives us a real `AppState` with a real node keypair.
+    ///
+    /// Round 3: the issuer stamps v1 because the shipped `gl`
+    /// verifier only supports v1 (stamping v2 shipped certs the
+    /// client refuses). The binding to the live issuer is kept —
+    /// flipping the stamp still turns this test red — without
+    /// changing what production issues.
+    #[sqlx::test]
+    #[allow(clippy::async_yields_async)]
+    async fn issuer_stamps_v1_over_v1_payload(pool: sqlx::PgPool) {
+        let state = crate::test_support::test_state(pool.clone()).await;
+        let repo_id = "repo-issuer-observe";
+        let ref_name = "refs/heads/main";
+        let old = "0".repeat(40);
+        let new = "a".repeat(40);
+        let pusher = "did:key:z6MkPusher";
+
+        // Drive the live issuer and read the persisted row back by
+        // id, so the test covers the stored `version` column and
+        // not just the in-process return value.
+        let cert =
+            crate::cert::issue_ref_certificate(&state, repo_id, ref_name, &old, &new, pusher)
+                .await
+                .expect("issue_ref_certificate must succeed");
+
+        // The cert returned by the issuer is the source of truth
+        // for both the version claim and the signed bytes.
+        assert_eq!(cert.repo_id, repo_id);
+        assert_eq!(cert.ref_name, ref_name);
+
+        // Cross-check 1: the issuer's claimed version must match
+        // what the live function stamps. This assertion is bound
+        // to the live function, not a hand-built literal:
+        // flipping the stamp in `issue_ref_certificate` breaks it
+        // through the actual call path.
+        assert_eq!(
+            cert.version, 1,
+            "the live issuer must stamp v1 until a v2 verifier ships; \
+             flipping the stamp breaks this assertion through the actual call path"
+        );
+
+        // Cross-check 1b: the persisted row carries the same stamp.
+        let stored = state
+            .db
+            .get_ref_certificate(&cert.id)
+            .await
+            .expect("persisted cert must be readable")
+            .expect("persisted cert must exist");
+        assert_eq!(
+            stored.version, cert.version,
+            "the stored version column must match the issued claim"
+        );
+
+        // Cross-check 2: the signature on the cert must verify
+        // against the v1 payload builder's bytes, not against some
+        // other shape the issuer might have introduced. This is
+        // what gl's `verify_signature` does for a v1 cert: rebuild
+        // the payload from the cert's own fields and verify the
+        // signature. If the issuer ever signs a different shape
+        // while still claiming v1, every shipped client breaks —
+        // and so does this.
+        let rebuilt = v1_signing_payload(
+            &cert.repo_id,
+            &cert.ref_name,
+            &cert.old_sha,
+            &cert.new_sha,
+            &cert.pusher_did,
+            &cert.node_did,
+            &cert.issued_at,
+        );
+        let vk = cert
+            .node_did
+            .parse::<gitlawb_core::did::Did>()
+            .unwrap()
+            .to_verifying_key()
+            .unwrap();
+        use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
+        let sig_bytes: [u8; 64] = URL_SAFE_NO_PAD
+            .decode(&cert.signature)
+            .unwrap()
+            .try_into()
+            .unwrap();
+        gitlawb_core::identity::verify(&vk, &serde_json::to_vec(&rebuilt).unwrap(), &sig_bytes)
+            .expect("a v1 cert's signature must verify against the v1 payload shape");
+
+        // Cross-check 3: the v1 signed payload must NOT contain a
+        // version key — v2 is the future shape that adds one inside
+        // the signed bytes (see `v2_signing_payload`). The v1
+        // payload (which the cert is signed over) is frozen at 7
+        // fields. A change that adds `version` to the v1 payload
+        // breaks every existing cert and this test.
+        let obj = rebuilt
+            .as_object()
+            .expect("rebuilt v1 payload is a JSON object");
+        assert!(
+            !obj.contains_key("version"),
+            "the v1 signing payload must not carry a version key — \
+             adding one changes the bytes and breaks every existing \
+             cert's signature"
+        );
+    }
+}

crates/gitlawb-node/src/db/mod.rs

@@ -6,6 +6,20 @@ use std::time::Duration;
 use tracing::info;
 use uuid::Uuid;
 
+/// Maximum visible repositories per page, plus one internal look-ahead row.
+pub(crate) const MAX_VISIBLE_REPO_PAGE_SIZE: usize = 200;
+
+/// Maximum repository name length in bytes. `repo_cursor` embeds the name in
+/// a base64(JSON) cursor that `parse_repo_cursor` rejects over 4096 bytes, so
+/// every write path must bound the name: a cursor the server emits must
+/// always parse back (#465 4th round). 100 bytes keeps the longest possible
+/// cursor an order of magnitude under the parse limit.
+pub(crate) const MAX_REPO_NAME_LEN: usize = 100;
+
+/// Maximum agent tasks returned per query. Shared by the GraphQL `tasks`
+/// complexity meter and the resolver clamp so they can't drift.
+pub(crate) const MAX_VISIBLE_TASK_PAGE_SIZE: i64 = 200;
+
 // ── Public data types ─────────────────────────────────────────────────────────
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -278,6 +292,14 @@ pub struct RefCertificate {
     pub node_did: String,
     pub signature: String,
     pub issued_at: String,
+    /// #26 Split PR 3 — the wire-format version of this cert. v1 is
+    /// the pre-versioning 7-field payload (no `version` key in the
+    /// signed bytes). v2+ is reserved for a new signed payload shape
+    /// that carries the version inside the signed bytes; it is not
+    /// v1 plus unsigned fields, because flipping this column must
+    /// invalidate the signature. Clients refuse to verify versions
+    /// they cannot represent rather than guessing.
+    pub version: u32,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -397,6 +419,17 @@ impl Db {
         Self { pool }
     }
 
+    /// Test-only: access the underlying Postgres connection pool.
+    ///
+    /// The production accessor was removed on main (ed76adf) once every
+    /// production caller had migrated to typed `Db` methods; this restores
+    /// it for tests only, for the raw-SQL fixtures and closed-pool cases in
+    /// this change that have no typed equivalent.
+    #[cfg(test)]
+    pub fn pool(&self) -> &PgPool {
+        &self.pool
+    }
+
     /// Test-only: apply the full schema to a fresh pool. `#[sqlx::test]`
     /// provisions an empty per-test database, so DB-backed tests must run this
     /// before seeding. Reuses the production `migrate()` path (the advisory lock
@@ -1358,6 +1391,32 @@ const MIGRATIONS: &[Migration] = &[
         name: "validate_repo_upstream_mirror_state",
         stmts: &["ALTER TABLE repos VALIDATE CONSTRAINT repos_mirror_state_valid"],
     },
+    Migration {
+        version: 37,
+        name: "ref_certificates_version",
+        stmts: &[
+            // #26 Split PR 3 — certificate / CLI compatibility. The
+            // ref-cert wire format is versioned so future fields can
+            // be added without breaking old clients (which ignore
+            // the field) or old servers (which default the column
+            // to 1). New certificates are currently issued as v1,
+            // which is byte-for-byte identical to the pre-versioning
+            // shape: an old client reading a v1 cert from a new
+            // server sees the same payload, the same signature, and
+            // the same Ed25519 verify path. A new client reading an
+            // old server sees `version` defaulted to 1 and verifies
+            // against the v1 payload shape. v2 is reserved for a
+            // future shape that carries the version inside the
+            // signed bytes; it must not be stamped until a v2
+            // verifier ships.
+            //
+            // DEFAULT 1 so an existing row reads as v1 without a
+            // backfill migration. NOT NULL so a missing value is a
+            // hard error at insert time rather than a silent v0 that
+            // an old client would interpret as "no version field."
+            "ALTER TABLE ref_certificates ADD COLUMN IF NOT EXISTS version INTEGER NOT NULL DEFAULT 1",
+        ],
+    },
 ];
 
 /// Max distinct source repos recorded per pinned object (F1, #173 jatmn round 8).
@@ -1560,6 +1619,16 @@ impl Db {
         machine_id: Option<&str>,
         quarantined: bool,
     ) -> Result<()> {
+        // Peer-supplied names are bound like the local write paths: an
+        // unbounded name would make reposPage emit a cursor longer than the
+        // 4096-byte parse limit, dead-ending pagination for every caller past
+        // that row (#465 4th round). The sync caller drops the error and skips
+        // the mirror.
+        if name.len() > MAX_REPO_NAME_LEN {
+            anyhow::bail!(
+                "mirror repo name exceeds {MAX_REPO_NAME_LEN} bytes: refusing to register"
+            );
+        }
         let now = Utc::now().to_rfc3339();
         let id = format!("{owner_short}/{name}");
         // `quarantined` is set only on first insert (the admission decision).
@@ -1851,6 +1920,73 @@ impl Db {
         Ok(rows.into_iter().map(row_to_repo).collect())
     }
 
+    /// A bounded, mirror-deduplicated page ordered by owner key and repository
+    /// name. Apply root visibility before LIMIT: private rows must not consume
+    /// page slots or influence continuation metadata. The root-rule predicate
+    /// mirrors `visibility::listable_at_root`; a differential test pins it.
+    /// Malformed reader JSON denies that repo to non-owners without failing the page.
+    /// Cursors only select a position and never confer read authority.
+    ///
+    /// Performance note: the page orders and keyset-filters on
+    /// `(owner_key, name)` under `COLLATE "C"`, but the existing
+    /// `idx_repos_owner_key_name` index is built on the database default
+    /// collation and cannot serve that ordering, so each page sorts the full
+    /// deduped repo set before the keyset filter applies: O(total repos) work
+    /// per page. This is accepted (pages are capped at
+    /// `MAX_VISIBLE_REPO_PAGE_SIZE` rows over a narrow projection). A
+    /// C-collation expression index was deliberately not added in a migration:
+    /// versions 27-35 are claimed by other in-flight branches (Gitlawb/node#384
+    /// claims 27-35, #464 claims 27-28) and the runner keys applied migrations
+    /// on the version integer alone, so a colliding entry would be silently
+    /// skipped on whichever side merges second. Revisit once that range lands
+    /// or clears.
+    pub async fn list_visible_repos_page(
+        &self,
+        caller: Option<&str>,
+        after: Option<(&str, &str)>,
+        limit: usize,
+    ) -> Result<Vec<RepoRecord>> {
+        let sql = format!(
+            "{}
+             SELECT d.id, d.name, d.owner_did, d.description, d.is_public,
+                 d.default_branch, d.created_at, d.updated_at, d.disk_path,
+                 d.forked_from, d.machine_id
+             FROM deduped d
+             LEFT JOIN LATERAL (
+                 SELECT reader_dids FROM visibility_rules v
+                 WHERE v.repo_id = d.id AND v.path_glob ~ '^/*([*][*])*$'
+                 ORDER BY v.path_glob DESC LIMIT 1
+             ) root_rule ON TRUE
+             WHERE (
+                 ($2::text IS NOT NULL AND ({key}) = $2)
+                 OR CASE WHEN root_rule.reader_dids IS NULL THEN d.is_public
+                    WHEN NOT pg_input_is_valid(root_rule.reader_dids, 'jsonb') THEN FALSE
+                    WHEN jsonb_typeof(root_rule.reader_dids::jsonb) = 'array' THEN
+                        COALESCE(root_rule.reader_dids::jsonb ? $3::text, FALSE)
+                        AND NOT EXISTS (
+                            SELECT 1 FROM jsonb_array_elements(root_rule.reader_dids::jsonb) reader
+                            WHERE jsonb_typeof(reader) <> 'string'
+                        )
+                    ELSE FALSE END
+             )
+             AND ($4::text IS NULL OR (({key}) COLLATE \"C\", d.name COLLATE \"C\") > ($4, $5::text))
+             ORDER BY ({key}) COLLATE \"C\", d.name COLLATE \"C\"
+             LIMIT $6",
+            Self::dedup_cte(),
+            key = OWNER_KEY_CASE_SQL,
+        );
+        let rows = sqlx::query(&sql)
+            .bind(None::<&str>)
+            .bind(caller.map(normalize_owner_key))
+            .bind(caller)
+            .bind(after.map(|(owner, _)| normalize_owner_key(owner)))
+            .bind(after.map(|(_, name)| name))
+            .bind(limit.min(MAX_VISIBLE_REPO_PAGE_SIZE + 1) as i64)
+            .fetch_all(&self.pool)
+            .await?;
+        Ok(rows.into_iter().map(row_to_repo).collect())
+    }
+
     /// Repos currently quarantined (admitted as mirrors but withheld from every
     /// listing surface). `list_all_repos_deduped` excludes these (its `DEDUP_CTE`
     /// filters `quarantined = FALSE`), so a gate that resolves a slug against the
@@ -2647,8 +2783,8 @@ impl Db {
     pub async fn insert_ref_certificate(&self, cert: &RefCertificate) -> Result<RefCertificate> {
         let row = sqlx::query(
             "INSERT INTO ref_certificates
-             (id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at)
-             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+             (id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at, version)
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
              ON CONFLICT (repo_id, ref_name) DO UPDATE SET
                 old_sha   = CASE WHEN EXCLUDED.issued_at > ref_certificates.issued_at
                                  THEN EXCLUDED.old_sha   ELSE ref_certificates.old_sha   END,
@@ -2661,8 +2797,10 @@ impl Db {
                 signature = CASE WHEN EXCLUDED.issued_at > ref_certificates.issued_at
                                  THEN EXCLUDED.signature ELSE ref_certificates.signature END,
                 issued_at = CASE WHEN EXCLUDED.issued_at > ref_certificates.issued_at
-                                 THEN EXCLUDED.issued_at ELSE ref_certificates.issued_at END
-             RETURNING id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at",
+                                 THEN EXCLUDED.issued_at ELSE ref_certificates.issued_at END,
+                version   = CASE WHEN EXCLUDED.issued_at > ref_certificates.issued_at
+                                 THEN EXCLUDED.version   ELSE ref_certificates.version   END
+             RETURNING id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at, version",
         )
         .bind(&cert.id)
         .bind(&cert.repo_id)
@@ -2673,6 +2811,7 @@ impl Db {
         .bind(&cert.node_did)
         .bind(&cert.signature)
         .bind(&cert.issued_at)
+        .bind(cert.version as i32)
         .fetch_one(&self.pool)
         .await?;
         Ok(row_to_cert(row))
@@ -2687,7 +2826,7 @@ impl Db {
         // bounded even if a raw/negative value slips through the handler layer.
         let limit = limit.max(1);
         let rows = sqlx::query(
-            "SELECT id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at
+            "SELECT id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at, version
              FROM ref_certificates WHERE repo_id = $1 ORDER BY issued_at DESC LIMIT $2",
         )
         .bind(repo_id)
@@ -2726,7 +2865,7 @@ impl Db {
         let pattern = format!("{}%", escaped_prefix);
 
         let rows = sqlx::query(
-            "SELECT id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at
+            "SELECT id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at, version
              FROM ref_certificates WHERE repo_id = $1 AND id LIKE $2 ESCAPE '!' ORDER BY issued_at DESC LIMIT $3",
         )
         .bind(repo_id)
@@ -2739,7 +2878,7 @@ impl Db {
 
     pub async fn get_ref_certificate(&self, id: &str) -> Result<Option<RefCertificate>> {
         let row = sqlx::query(
-            "SELECT id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at
+            "SELECT id, repo_id, ref_name, old_sha, new_sha, pusher_did, node_did, signature, issued_at, version
              FROM ref_certificates WHERE id = $1",
         )
         .bind(id)
@@ -4000,28 +4139,54 @@ impl Db {
 
 // ── Agent Tasks ───────────────────────────────────────────────────────────────
 
+/// Permanent authorization denial: the task is reserved for a different assignee.
+/// Handlers downcast this to return 403 (vs 409 for a lost claim race).
+#[derive(Debug, thiserror::Error)]
+#[error("task not claimable: reserved for another assignee")]
+pub struct TaskReservedForOtherAssignee;
+
+/// ASCII whitespace treated as blank for assignee slots — must stay in sync with
+/// `BTRIM(assignee_did, E' \t\n\r')` in `claim_task`'s SQL open-slot check.
+const ASSIGNEE_BLANK: &[char] = &[' ', '\t', '\n', '\r'];
+
+fn assignee_slot_blank(s: &str) -> bool {
+    s.trim_matches(ASSIGNEE_BLANK).is_empty()
+}
+
+/// Trim assignee DIDs for auth matching (same ASCII blank set as SQL/open checks).
+pub fn trim_assignee_did(s: &str) -> &str {
+    s.trim_matches(ASSIGNEE_BLANK)
+}
+
 impl Db {
-    pub async fn create_task(&self, task: &AgentTask) -> Result<()> {
+    /// Persist a task. Whitespace-only `assignee_did` is stored as `NULL` (open).
+    /// Returns the normalized row shape so REST/GraphQL create responses match GET.
+    pub async fn create_task(&self, task: &AgentTask) -> Result<AgentTask> {
+        let mut stored = task.clone();
+        stored.assignee_did = stored
+            .assignee_did
+            .take()
+            .filter(|s| !assignee_slot_blank(s));
         sqlx::query(
             "INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
              VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)",
         )
-        .bind(&task.id)
-        .bind(&task.repo_id)
-        .bind(&task.kind)
-        .bind(&task.status)
-        .bind(&task.delegator_did)
-        .bind(&task.assignee_did)
-        .bind(&task.capability)
-        .bind(&task.ucan_token)
-        .bind(&task.payload)
-        .bind(&task.result)
-        .bind(&task.created_at)
-        .bind(&task.updated_at)
-        .bind(&task.deadline)
+        .bind(&stored.id)
+        .bind(&stored.repo_id)
+        .bind(&stored.kind)
+        .bind(&stored.status)
+        .bind(&stored.delegator_did)
+        .bind(&stored.assignee_did)
+        .bind(&stored.capability)
+        .bind(&stored.ucan_token)
+        .bind(&stored.payload)
+        .bind(&stored.result)
+        .bind(&stored.created_at)
+        .bind(&stored.updated_at)
+        .bind(&stored.deadline)
         .execute(&self.pool)
         .await?;
-        Ok(())
+        Ok(stored)
     }
 
     pub async fn get_task(&self, id: &str) -> Result<Option<AgentTask>> {
@@ -4078,38 +4243,102 @@ impl Db {
         Ok(rows.into_iter().map(row_to_task).collect())
     }
 
+    /// Claim a pending task for `assignee_did`.
+    ///
+    /// If the task was created with a non-blank pre-set `assignee_did`, only that
+    /// agent (DID-normalized via [`crate::api::did_matches`]) may claim it. Open
+    /// tasks (`NULL` / blank) remain first-claimer-wins. A reserved claim keeps
+    /// the **exact** stored DID form via `COALESCE($4, $2)` (no BTRIM rewrite) so
+    /// exact-match list filters still work. The UPDATE also re-checks the
+    /// assignee slot as defense-in-depth against a future writer racing the
+    /// pre-check.
     pub async fn claim_task(&self, id: &str, assignee_did: &str) -> Result<AgentTask> {
         let now = Utc::now().to_rfc3339();
+        // Denial path only needs status + assignee — do not load payload/UCAN
+        // for a permissionless caller who will be rejected.
+        let row = sqlx::query("SELECT status, assignee_did FROM agent_tasks WHERE id = $1")
+            .bind(id)
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("task not claimable: not found or already claimed"))?;
+        let status: String = row.get("status");
+        if status != "pending" {
+            return Err(anyhow::anyhow!(
+                "task not claimable: not found or already claimed"
+            ));
+        }
+        let stored: Option<String> = row.get("assignee_did");
+        // Blank reservations are treated as open. Keep the exact stored string
+        // for the UPDATE equality check and SET (do not BTRIM the reserved form).
+        // Blank definition matches SQL `BTRIM(..., E' \t\n\r')` (not space-only).
+        let reserved_exact = stored.as_deref().filter(|r| !assignee_slot_blank(r));
+        if let Some(reserved) = reserved_exact {
+            if !crate::api::did_matches(assignee_did, trim_assignee_did(reserved)) {
+                return Err(TaskReservedForOtherAssignee.into());
+            }
+        }
+        // Keep the exact reserved form ($4); only fill assignee on open claims ($2).
+        // Treat blank stored values as open for the SQL slot check.
         let row = sqlx::query(
-            "UPDATE agent_tasks SET status='claimed', assignee_did=$2, updated_at=$3
+            "UPDATE agent_tasks SET status='claimed',
+                 assignee_did = COALESCE($4, $2),
+                 updated_at=$3
              WHERE id=$1 AND status='pending'
+               AND (
+                 ($4::text IS NULL AND (assignee_did IS NULL OR BTRIM(assignee_did, E' \t\n\r') = ''))
+                 OR assignee_did = $4
+               )
              RETURNING id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline",
         )
         .bind(id)
         .bind(assignee_did)
         .bind(&now)
+        .bind(reserved_exact)
         .fetch_optional(&self.pool)
         .await?;
         row.map(row_to_task)
             .ok_or_else(|| anyhow::anyhow!("task not claimable: not found or already claimed"))
     }
 
+    /// Transition a claimed task to `new_status` (`completed` / `failed`).
+    ///
+    /// `actor_did` must be the task's assignee (DID-normalized). The UPDATE
+    /// binds the exact stored `assignee_did` so a concurrent reassignment /
+    /// re-claim cannot finish under a check-then-act race in the handler.
     pub async fn finish_task(
         &self,
         id: &str,
         new_status: &str,
         result: Option<&str>,
+        actor_did: &str,
     ) -> Result<AgentTask> {
         let now = Utc::now().to_rfc3339();
+        let existing = self
+            .get_task(id)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("task not found or not in claimed state"))?;
+        if existing.status != "claimed" {
+            return Err(anyhow::anyhow!("task not found or not in claimed state"));
+        }
+        let Some(ref assigned) = existing.assignee_did else {
+            return Err(anyhow::anyhow!("task not found or not in claimed state"));
+        };
+        // Trim matches claim_task so tab-padded stored values can finish.
+        if !crate::api::did_matches(actor_did, trim_assignee_did(assigned)) {
+            return Err(anyhow::anyhow!(
+                "task not finishable: only the assignee may finish it"
+            ));
+        }
         let row = sqlx::query(
             "UPDATE agent_tasks SET status=$2, result=$3, updated_at=$4
-             WHERE id=$1 AND status='claimed'
+             WHERE id=$1 AND status='claimed' AND assignee_did=$5
              RETURNING id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline",
         )
         .bind(id)
         .bind(new_status)
         .bind(result)
         .bind(&now)
+        .bind(assigned.as_str())
         .fetch_optional(&self.pool)
         .await?;
         row.map(row_to_task)
@@ -4853,6 +5082,7 @@ fn row_to_cert(r: sqlx::postgres::PgRow) -> RefCertificate {
         node_did: r.get("node_did"),
         signature: r.get("signature"),
         issued_at: r.get("issued_at"),
+        version: r.get::<i32, _>("version") as u32,
     }
 }
 
@@ -7561,6 +7791,7 @@ mod ref_certificate_tests {
             node_did: "did:key:zNODE".to_string(),
             signature: "sig".to_string(),
             issued_at: issued_at.to_string(),
+            version: 1,
         }
     }
 
@@ -8234,6 +8465,130 @@ mod ref_certificate_tests {
         );
     }
 
+    /// #26 Split PR 3 (INV-7): an existing node past v1 gets the
+    /// `ref_certificates.version` column from its OWN v37 entry, proven
+    /// by dropping the column plus its `schema_migrations` row and
+    /// re-running the real migration code. The migration is the only
+    /// place the column is added (v37 — the v1 bundle no longer
+    /// carries it), so a deletion of v37 must be visible as a missing
+    /// column on a v27 node.
+    ///
+    /// DEFAULT 1 is what makes an existing pre-v37 cert row read as
+    /// v1 without a backfill migration. An upgraded node reads
+    /// every legacy row as version 1 — the same payload the old
+    /// code signed — so a v1 verify path on a new client still
+    /// works against an upgraded database.
+    ///
+    /// MUTATION (RED): delete the v37 entry from `MIGRATIONS` and the
+    /// upgrade path leaves the column missing.
+    #[sqlx::test]
+    async fn v37_ref_certificates_version_applies_on_upgrade(pool: PgPool) {
+        async fn version_column_default(pool: &PgPool) -> Option<String> {
+            // fetch_optional (not fetch_one) so a missing column
+            // returns Ok(None) instead of RowNotFound. The whole
+            // point of the precondition assertion below is to
+            // detect a missing column, and RowNotFound would mask
+            // the difference between "the helper is wrong" and
+            // "the migration is wrong".
+            sqlx::query_scalar::<_, Option<String>>(
+                "SELECT column_default
+                   FROM information_schema.columns
+                  WHERE table_name = 'ref_certificates'
+                    AND column_name = 'version'",
+            )
+            .fetch_optional(pool)
+            .await
+            .unwrap()
+            .flatten()
+        }
+
+        async fn seed_pre_v37_cert(pool: &PgPool) {
+            // The pre-v37 schema has no `version` column, so a raw
+            // INSERT omitting it is the legacy code path. The node
+            // did not write a version value on v1..v27 — the field
+            // did not exist.
+            sqlx::query(
+                "INSERT INTO ref_certificates
+                 (id, repo_id, ref_name, old_sha, new_sha,
+                  pusher_did, node_did, signature, issued_at)
+                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+                 ON CONFLICT (repo_id, ref_name) DO NOTHING",
+            )
+            .bind("legacy-cert-1")
+            .bind("legacy-repo")
+            .bind("refs/heads/main")
+            .bind("0000")
+            .bind("1111")
+            .bind("did:key:zLEGACYPUSHER")
+            .bind("did:key:zLEGACYNODE")
+            .bind("legacy-sig")
+            .bind("2026-07-01T10:00:00+00:00")
+            .execute(pool)
+            .await
+            .unwrap();
+        }
+
+        async fn legacy_row_reads_as_v1(pool: &PgPool) -> i32 {
+            // After v37 the legacy row (inserted without an explicit
+            // version) reads back as 1 via the DEFAULT, not as 0 or
+            // NULL — the latter would be a hard insert error under
+            // NOT NULL DEFAULT, but the read-back is the actual
+            // forward-compat property the v1 verify path relies on.
+            sqlx::query_scalar::<_, i32>(
+                "SELECT version FROM ref_certificates WHERE id = 'legacy-cert-1'",
+            )
+            .fetch_one(pool)
+            .await
+            .unwrap()
+        }
+
+        // 1. Fresh chain: v37 has run, the column exists with DEFAULT 1.
+        let db = Db::for_testing(pool.clone());
+        db.run_migrations().await.unwrap();
+        assert_eq!(
+            version_column_default(&pool).await.as_deref(),
+            Some("1"),
+            "v37 must declare the version column with DEFAULT 1"
+        );
+
+        // 2. Roll back to pre-v37: drop the column AND the v37 record.
+        //    The rollback is split into two statements because ALTER
+        //    TABLE ... DROP COLUMN and DELETE run in the same DDL
+        //    surface but the column drop must complete before the
+        //    next step reads schema_migrations.
+        sqlx::query("ALTER TABLE ref_certificates DROP COLUMN version")
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query("DELETE FROM schema_migrations WHERE version = 37")
+            .execute(&pool)
+            .await
+            .unwrap();
+        assert!(
+            version_column_default(&pool).await.is_none(),
+            "precondition: column removed and its migration record removed"
+        );
+
+        // 3. Seed a legacy cert on the pre-v37 schema (no version
+        //    column ⇒ the INSERT must omit it).
+        seed_pre_v37_cert(&pool).await;
+
+        // 4. Re-run migrations: v37 must re-add the column with
+        //    DEFAULT 1, and the legacy row must read back as 1.
+        db.run_migrations().await.unwrap();
+        assert_eq!(
+            version_column_default(&pool).await.as_deref(),
+            Some("1"),
+            "v37 must recreate the version column with DEFAULT 1 on an upgrading node"
+        );
+        assert_eq!(
+            legacy_row_reads_as_v1(&pool).await,
+            1,
+            "a pre-v37 cert row reads as version 1 after v37 runs, so the \
+             v1 verify path on a new client still works against an upgraded database"
+        );
+    }
+
     /// INV-7: upgrade-path test — seed a database at v9 with duplicate
     /// ref_certificates, then let the real v10 migration fire via
     /// run_migrations().  This exercises the migration code path rather than

crates/gitlawb-node/src/git/repo_store.rs

@@ -118,7 +118,7 @@ impl RepoStore {
     /// Test-only: every guard from this store parks in `release` right before the
     /// `pg_advisory_unlock` await, until `gate` is notified. Dropping the future
     /// while it is parked reproduces a client disconnect inside `release`.
-    #[cfg(test)]
+    #[cfg(all(test, unix))]
     pub fn with_pre_unlock_gate(mut self, gate: Arc<tokio::sync::Notify>) -> Self {
         self.pre_unlock_gate = Some(gate);
         self
@@ -126,7 +126,7 @@ impl RepoStore {
 
     /// Test-only: how many write guards from this store have reached the Tigris upload
     /// site. See [`RepoWriteGuard::release`].
-    #[cfg(test)]
+    #[cfg(all(test, unix))]
     pub fn tigris_upload_site_reached(&self) -> usize {
         self.upload_site_reached
             .load(std::sync::atomic::Ordering::SeqCst)
@@ -1717,8 +1717,11 @@ fn validate_repo_name(repo_name: &str) -> Result<()> {
     if repo_name.is_empty() {
         anyhow::bail!("repo_name is empty");
     }
-    if repo_name.len() > 100 {
-        anyhow::bail!("repo_name exceeds 100 chars");
+    // Byte bound shared with the API-side validator via `MAX_REPO_NAME_LEN`:
+    // the two validators differ on charset by design, but must not drift on
+    // length, or an API-admitted name fails here after the proof is spent.
+    if repo_name.len() > crate::db::MAX_REPO_NAME_LEN {
+        anyhow::bail!("repo_name exceeds {} chars", crate::db::MAX_REPO_NAME_LEN);
     }
     // Repo names are `[A-Za-z0-9._-]+` minus path-traversal traps.
     if repo_name.contains("..") {

crates/gitlawb-node/src/graphql/mod.rs

@@ -3,7 +3,7 @@ pub mod query;
 pub mod subscription;
 pub mod types;
 
-use async_graphql::Schema;
+use async_graphql::{Schema, SchemaBuilder};
 use std::sync::Arc;
 
 use crate::db::Db;
@@ -14,6 +14,22 @@ use subscription::SubscriptionRoot;
 
 pub type GitlawbSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;
 
+// Keep ordinary schema discovery and application queries usable while
+// bounding validation work and the number of resolver selections per request.
+const GRAPHQL_MAX_COMPLEXITY: usize = 400;
+// The current public schema is shallow; this leaves headroom for composed
+// clients and the canonical getIntrospectionQuery without allowing
+// recursively nested documents to grow unchecked.
+const GRAPHQL_MAX_DEPTH: usize = 14;
+
+fn apply_query_limits<Query, Mutation, Subscription>(
+    builder: SchemaBuilder<Query, Mutation, Subscription>,
+) -> SchemaBuilder<Query, Mutation, Subscription> {
+    builder
+        .limit_complexity(GRAPHQL_MAX_COMPLEXITY)
+        .limit_depth(GRAPHQL_MAX_DEPTH)
+}
+
 /// Client-facing message for GraphQL resolver failures that wrap a real
 /// `sqlx::Error`. The real error is logged server-side; never put sqlx/Postgres
 /// detail in the GraphQL `errors` array (#250).
@@ -81,7 +97,7 @@ pub fn build_schema(
     ref_update_tx: tokio::sync::broadcast::Sender<RefUpdateBroadcast>,
     task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
 ) -> GitlawbSchema {
-    Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
+    apply_query_limits(Schema::build(QueryRoot, MutationRoot, SubscriptionRoot))
         .data(db)
         .data(ref_update_tx)
         .data(task_event_tx)
@@ -91,6 +107,10 @@ pub fn build_schema(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::api::events::MAX_VISIBLE_REF_UPDATES;
+    use crate::db::{MAX_VISIBLE_REPO_PAGE_SIZE, MAX_VISIBLE_TASK_PAGE_SIZE};
+    use async_graphql::{EmptyMutation, EmptySubscription, Object, Value};
+    use std::sync::atomic::{AtomicUsize, Ordering};
 
     #[test]
     fn graphql_db_err_opaques_sqlx_chain() {
@@ -160,6 +180,668 @@ mod tests {
         assert!(!err.message.contains("failed to open"));
     }
 
+    fn production_test_schema() -> GitlawbSchema {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        production_test_schema_with_db(Arc::new(Db::for_testing(pool)))
+    }
+
+    fn production_test_schema_with_db(db: Arc<Db>) -> GitlawbSchema {
+        build_schema(
+            db,
+            tokio::sync::broadcast::channel(1).0,
+            tokio::sync::broadcast::channel(1).0,
+        )
+    }
+
+    #[tokio::test]
+    async fn ref_update_subscription_alias_budget() {
+        assert_subscription_alias_budget("refUpdates { repo }").await;
+    }
+
+    #[tokio::test]
+    async fn task_event_subscription_alias_budget() {
+        assert_subscription_alias_budget("taskEvents { taskId }").await;
+    }
+
+    async fn assert_subscription_alias_budget(field: &str) {
+        use futures::StreamExt;
+        use std::time::Duration;
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        let (ref_tx, _) = tokio::sync::broadcast::channel(16);
+        let (task_tx, _) = tokio::sync::broadcast::channel(16);
+        let schema = build_schema(
+            Arc::new(Db::for_testing(pool)),
+            ref_tx.clone(),
+            task_tx.clone(),
+        );
+        for count in [7, 8] {
+            let fields = (0..count)
+                .map(|n| format!("r{n}: {field}"))
+                .collect::<Vec<_>>()
+                .join(" ");
+            let query = format!("subscription {{ {fields} }}");
+            let mut responses = Box::pin(schema.execute_stream(query));
+            if count == 8 {
+                let response = tokio::time::timeout(Duration::from_secs(1), responses.next())
+                    .await
+                    .expect("over-budget subscriptions must reject before waiting for events")
+                    .expect("the validation error must be returned");
+                assert_eq!(response.errors.len(), 1);
+                assert_eq!(response.errors[0].message, "Query is too complex.");
+                assert_eq!(ref_tx.receiver_count() + task_tx.receiver_count(), 0);
+                continue;
+            }
+            // Poll the accepted operation far enough to register its receivers.
+            assert!(
+                tokio::time::timeout(Duration::from_millis(30), responses.next())
+                    .await
+                    .is_err()
+            );
+            assert_eq!(ref_tx.receiver_count() + task_tx.receiver_count(), 7);
+            let _ = ref_tx.send(RefUpdateBroadcast {
+                repo: "owner/repo".into(),
+                ref_name: "refs/heads/main".into(),
+                old_sha: "0".repeat(40),
+                new_sha: "1".repeat(40),
+                pusher_did: "did:key:test".into(),
+                node_did: "did:key:node".into(),
+                timestamp: "2026-01-01T00:00:00Z".into(),
+                owner_did: "did:key:test".into(),
+            });
+            let _ = task_tx.send(TaskEventBroadcast {
+                task_id: "task".into(),
+                old_status: "pending".into(),
+                new_status: "claimed".into(),
+                by_did: "did:key:test".into(),
+                at: "2026-01-01T00:00:00Z".into(),
+            });
+            let response = tokio::time::timeout(Duration::from_secs(1), responses.next())
+                .await
+                .unwrap()
+                .unwrap();
+            assert!(response.errors.is_empty(), "{:?}", response.errors);
+            assert_ne!(response.data, Value::Null);
+            drop(responses);
+            assert_eq!(ref_tx.receiver_count() + task_tx.receiver_count(), 0);
+        }
+    }
+
+    #[sqlx::test]
+    async fn production_mutation_budget_accepts_seven_and_rejects_eight(pool: sqlx::PgPool) {
+        let db = Arc::new(Db::for_testing(pool));
+        db.run_migrations().await.unwrap();
+        let caller = "did:key:reviewer";
+        let now = chrono::Utc::now().to_rfc3339();
+        for index in 0..8 {
+            db.create_task(&crate::db::AgentTask {
+                id: format!("task-{index}"),
+                repo_id: None,
+                kind: "test".into(),
+                status: "pending".into(),
+                delegator_did: caller.into(),
+                assignee_did: None,
+                capability: "test".into(),
+                ucan_token: None,
+                payload: None,
+                result: None,
+                created_at: now.clone(),
+                updated_at: now.clone(),
+                deadline: None,
+            })
+            .await
+            .unwrap();
+        }
+        let schema = production_test_schema_with_db(db.clone());
+        for count in [8, 7] {
+            let fields = (0..count)
+                .map(|n| {
+                    format!("r{n}: claimTask(id: \"task-{n}\", assigneeDid: \"{caller}\") {{ id }}")
+                })
+                .collect::<Vec<_>>()
+                .join(" ");
+            let response = schema
+                .execute(
+                    async_graphql::Request::new(format!("mutation {{ {fields} }}"))
+                        .data(crate::auth::AuthenticatedDid(caller.into())),
+                )
+                .await;
+            if count == 8 {
+                assert_eq!(response.errors.len(), 1);
+                assert_eq!(response.errors[0].message, "Query is too complex.");
+                assert_eq!(
+                    db.list_tasks(Some("pending"), None, 8).await.unwrap().len(),
+                    8
+                );
+                continue;
+            }
+            assert!(response.errors.is_empty(), "{:?}", response.errors);
+            assert_eq!(
+                response
+                    .data
+                    .into_json()
+                    .unwrap()
+                    .as_object()
+                    .unwrap()
+                    .len(),
+                7
+            );
+            assert_eq!(
+                db.list_tasks(Some("claimed"), Some(caller), 8)
+                    .await
+                    .unwrap()
+                    .len(),
+                7
+            );
+            assert_eq!(
+                db.get_task("task-7").await.unwrap().unwrap().status,
+                "pending"
+            );
+        }
+    }
+
+    #[sqlx::test]
+    async fn production_list_cost_scales_with_the_same_alias_count(pool: sqlx::PgPool) {
+        let db = Arc::new(Db::for_testing(pool));
+        db.run_migrations().await.unwrap();
+        let schema = production_test_schema_with_db(db);
+        // Bounds are per-root named constants so the cases move with the
+        // documented maximums instead of going stale as literals.
+        for (field, selection, max) in [
+            ("refUpdates", "repo", MAX_VISIBLE_REF_UPDATES),
+            ("tasks", "id", MAX_VISIBLE_TASK_PAGE_SIZE),
+            (
+                "reposPage",
+                "nodes { name }",
+                MAX_VISIBLE_REPO_PAGE_SIZE as i64,
+            ),
+        ] {
+            // Two aliases at the maximum always exceed the budget.
+            let fields = (0..2)
+                .map(|n| format!("r{n}: {field}(limit: {max}) {{ {selection} }}"))
+                .collect::<Vec<_>>()
+                .join(" ");
+            let response = schema.execute(format!("{{ {fields} }}")).await;
+            assert_eq!(response.errors.len(), 1);
+            assert_eq!(response.errors[0].message, "Query is too complex.");
+
+            // At limit 1 two aliases fit for the row-priced roots; reposPage
+            // prices the full corpus scan, so a single alias is the most that
+            // fits under the budget.
+            let aliases = if field == "reposPage" { 1 } else { 2 };
+            let fields = (0..aliases)
+                .map(|n| format!("r{n}: {field}(limit: 1) {{ {selection} }}"))
+                .collect::<Vec<_>>()
+                .join(" ");
+            let response = schema.execute(format!("{{ {fields} }}")).await;
+            assert!(response.errors.is_empty(), "{field}: {:?}", response.errors);
+            assert_eq!(
+                response
+                    .data
+                    .into_json()
+                    .unwrap()
+                    .as_object()
+                    .unwrap()
+                    .len(),
+                aliases
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn production_limits_reject_mutation_aliases_and_large_lists() {
+        let schema = production_test_schema();
+        // Limits are derived from the MAX_* constants so the boundaries move
+        // with the documented maximums instead of going stale as literals.
+        for (prefix, count, field) in [
+            ("mutation", 8, "claimTask(id: \"missing\", assigneeDid: \"did:key:test\") { id }".to_owned()),
+            ("mutation", 8, "createTask(delegatorDid: \"did:key:test\", input: { kind: \"test\", capability: \"test\" }) { id }".to_owned()),
+            ("mutation", 8, "completeTask(id: \"missing\", byDid: \"did:key:test\", input: {}) { id }".to_owned()),
+            ("mutation", 8, "failTask(id: \"missing\", byDid: \"did:key:test\", input: {}) { id }".to_owned()),
+            ("query", 8, "task(id: \"missing\") { id }".to_owned()),
+            ("query", 2, format!("refUpdates(limit: {MAX_VISIBLE_REF_UPDATES}) {{ repo }}")),
+            ("query", 2, format!("tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id }}")),
+            ("query", 2, format!("reposPage(limit: {MAX_VISIBLE_REPO_PAGE_SIZE}) {{ nodes {{ name }} }}")),
+            ("query", 8, "reposPage(limit: 1) { nodes { name } }".to_owned()),
+        ] {
+            let fields = (0..count)
+                .map(|n| format!("r{n}: {field}"))
+                .collect::<Vec<_>>()
+                .join(" ");
+            let response = schema.execute(format!("{prefix} {{ {fields} }}")).await;
+            assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
+            assert_eq!(response.errors[0].message, "Query is too complex.");
+        }
+    }
+
+    #[sqlx::test]
+    async fn production_repos_page_serves_the_documented_maximum(pool: sqlx::PgPool) {
+        let db = Arc::new(Db::for_testing(pool));
+        db.run_migrations().await.unwrap();
+        let schema = production_test_schema_with_db(db);
+        for selection in [
+            "nodes { name }",
+            "nodes { name ownerDid } hasNextPage endCursor",
+        ] {
+            let response = schema
+                .execute(format!(
+                    "{{ reposPage(limit: {MAX_VISIBLE_REPO_PAGE_SIZE}) {{ {selection} }} }}"
+                ))
+                .await;
+            assert!(response.errors.is_empty(), "{:?}", response.errors);
+            assert_eq!(
+                response.data.into_json().unwrap()["reposPage"]["nodes"],
+                serde_json::json!([])
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn seven_root_aliases_are_accepted() {
+        struct Root(Arc<AtomicUsize>);
+        #[Object]
+        impl Root {
+            #[graphql(complexity = "50 + child_complexity")]
+            async fn repos(&self) -> Vec<Nested> {
+                self.0.fetch_add(1, Ordering::Relaxed);
+                vec![Nested]
+            }
+        }
+        let calls = Arc::new(AtomicUsize::new(0));
+        let schema = apply_query_limits(Schema::build(
+            Root(calls.clone()),
+            EmptyMutation,
+            EmptySubscription,
+        ))
+        .finish();
+        let fields = (0..7)
+            .map(|n| format!("r{n}: repos {{ value }}"))
+            .collect::<Vec<_>>()
+            .join(" ");
+        let response = schema.execute(format!("{{ {fields} }}")).await;
+        assert!(response.errors.is_empty(), "{:?}", response.errors);
+        assert_eq!(calls.load(Ordering::Relaxed), 7);
+    }
+
+    #[tokio::test]
+    async fn expensive_root_aliases_are_rejected_before_database_access() {
+        // Use the production builder with a lazy pool: rejection must precede DB access.
+        let schema = production_test_schema();
+        let fields = (0..8)
+            .map(|n| format!("r{n}: repos {{ name }}"))
+            .collect::<Vec<_>>()
+            .join(" ");
+
+        let response = schema.execute(format!("{{ {fields} }}")).await;
+
+        assert_eq!(response.data, Value::Null);
+        assert_eq!(response.errors.len(), 1);
+        assert_eq!(response.errors[0].message, "Query is too complex.");
+    }
+
+    #[tokio::test]
+    async fn ordinary_schema_introspection_remains_available() {
+        let schema = production_test_schema();
+        let canonical_introspection = r#"
+            query IntrospectionQuery {
+              __schema {
+                queryType { name }
+                mutationType { name }
+                subscriptionType { name }
+                types {
+                  ...FullType
+                }
+                directives {
+                  name
+                  description
+                  locations
+                  args {
+                    ...InputValue
+                  }
+                }
+              }
+            }
+
+            fragment FullType on __Type {
+              kind
+              name
+              description
+              fields(includeDeprecated: true) {
+                name
+                description
+                args {
+                  ...InputValue
+                }
+                type {
+                  ...TypeRef
+                }
+                isDeprecated
+                deprecationReason
+              }
+              inputFields {
+                ...InputValue
+              }
+              interfaces {
+                ...TypeRef
+              }
+              enumValues(includeDeprecated: true) {
+                name
+                description
+                isDeprecated
+                deprecationReason
+              }
+              possibleTypes {
+                ...TypeRef
+              }
+            }
+
+            fragment InputValue on __InputValue {
+              name
+              description
+              type { ...TypeRef }
+              defaultValue
+            }
+
+            fragment TypeRef on __Type {
+              kind
+              name
+              ofType {
+                kind
+                name
+                ofType {
+                  kind
+                  name
+                  ofType {
+                    kind
+                    name
+                    ofType {
+                      kind
+                      name
+                      ofType {
+                        kind
+                        name
+                        ofType {
+                          kind
+                          name
+                          ofType {
+                            kind
+                            name
+                          }
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+        "#;
+        let response = schema.execute(canonical_introspection).await;
+        assert!(
+            response.errors.is_empty(),
+            "canonical introspection query failed against production schema: {:?}",
+            response.errors
+        );
+        assert_ne!(response.data, Value::Null);
+    }
+
+    #[tokio::test]
+    async fn ref_updates_and_tasks_single_alias_max_limit_complexity_contract() {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        pool.close().await;
+        let (ref_tx, _) = tokio::sync::broadcast::channel(16);
+        let (task_tx, _) = tokio::sync::broadcast::channel(16);
+        let schema = build_schema(Arc::new(Db::for_testing(pool)), ref_tx, task_tx);
+
+        // refUpdates(limit: MAX_VISIBLE_REF_UPDATES):
+        // Single-field selection has child_complexity 1 -> cost 100 + 200 + 1 = 301 <= 400.
+        // It passes validation and reaches the resolver (which yields db error on lazy pool).
+        let single_ref = schema
+            .execute(format!(
+                "{{ refUpdates(limit: {MAX_VISIBLE_REF_UPDATES}) {{ repo }} }}"
+            ))
+            .await;
+        assert!(
+            !single_ref
+                .errors
+                .iter()
+                .any(|e| e.message == "Query is too complex."),
+            "single-field refUpdates at max limit must pass complexity validation: {:?}",
+            single_ref.errors
+        );
+        assert!(
+            single_ref
+                .errors
+                .iter()
+                .any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
+            "single-field refUpdates at max limit must reach the resolver: {:?}",
+            single_ref.errors
+        );
+
+        // Two-field selection has child_complexity 2 -> cost 100 + 200 + 2 = 302 <= 400.
+        // The documented max-200 request stays reachable with multiple fields
+        // (the old multiplicative price rejected it at 450); it passes
+        // validation and reaches the resolver like the single-field case.
+        let multi_ref = schema
+            .execute(format!(
+                "{{ refUpdates(limit: {MAX_VISIBLE_REF_UPDATES}) {{ repo refName }} }}"
+            ))
+            .await;
+        assert!(
+            !multi_ref
+                .errors
+                .iter()
+                .any(|e| e.message == "Query is too complex."),
+            "two-field refUpdates at max limit must pass complexity validation: {:?}",
+            multi_ref.errors
+        );
+        assert!(
+            multi_ref
+                .errors
+                .iter()
+                .any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
+            "two-field refUpdates at max limit must reach the resolver: {:?}",
+            multi_ref.errors
+        );
+
+        // tasks(limit: MAX_VISIBLE_TASK_PAGE_SIZE):
+        // Single-field selection has child_complexity 1, so the cost is
+        // 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 1 <= 400.
+        // It passes validation and reaches the resolver.
+        let single_task = schema
+            .execute(format!(
+                "{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id }} }}"
+            ))
+            .await;
+        assert!(
+            !single_task
+                .errors
+                .iter()
+                .any(|e| e.message == "Query is too complex."),
+            "single-field tasks at the maximum page size must pass complexity validation: {:?}",
+            single_task.errors
+        );
+        assert!(
+            single_task
+                .errors
+                .iter()
+                .any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
+            "single-field tasks at the maximum page size must reach the resolver: {:?}",
+            single_task.errors
+        );
+
+        // Two-field selection has child_complexity 2, so the cost is
+        // 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 2 <= 400.
+        // A request at the documented maximum stays reachable with multiple
+        // fields (the old multiplicative price rejected it at 450); it passes
+        // validation and reaches the resolver like the single-field case.
+        let multi_task = schema
+            .execute(format!(
+                "{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id status }} }}"
+            ))
+            .await;
+        assert!(
+            !multi_task
+                .errors
+                .iter()
+                .any(|e| e.message == "Query is too complex."),
+            "two-field tasks at the maximum page size must pass complexity validation: {:?}",
+            multi_task.errors
+        );
+        assert!(
+            multi_task
+                .errors
+                .iter()
+                .any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
+            "two-field tasks at the maximum page size must reach the resolver: {:?}",
+            multi_task.errors
+        );
+    }
+
+    #[tokio::test]
+    async fn two_repos_aliases_exceed_the_budget() {
+        // `repos` is priced as the full page it scans
+        // (50 + MAX_VISIBLE_REPO_PAGE_SIZE + child), so two `repos { name }`
+        // aliases cost 2 * 251 = 502 > 400 and the document is rejected before
+        // any resolver runs. If the price reverts to `50 + child_complexity`
+        // the aliases cost 2 * 51 = 102, validation passes, and the failure
+        // mode changes (resolver DB errors on the lazy pool): this test goes
+        // red on revert.
+        let schema = production_test_schema();
+        let response = schema
+            .execute("{ r0: repos { name } r1: repos { name } }")
+            .await;
+        assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
+        assert_eq!(response.errors[0].message, "Query is too complex.");
+    }
+
+    #[tokio::test]
+    async fn seven_repos_page_aliases_exceed_the_budget() {
+        // `reposPage` is priced as the full corpus scan
+        // (50 + MAX_VISIBLE_REPO_PAGE_SIZE + child_complexity), so seven
+        // `reposPage(limit: 1) { nodes { name } }` aliases cost
+        // 7 * (50 + 200 + 2) = 1764 > 400 and the document is rejected
+        // before any resolver runs. Under the previous row-priced formula
+        // the same aliases cost 7 * (50 + 1 + 2) = 371, validation passes,
+        // and the failure mode changes (resolver DB errors on the lazy
+        // pool): this test goes red on revert. Eight aliases are rejected
+        // under both formulas, so seven is the load-bearing boundary.
+        let schema = production_test_schema();
+        let fields = (0..7)
+            .map(|n| format!("r{n}: reposPage(limit: 1) {{ nodes {{ name }} }}"))
+            .collect::<Vec<_>>()
+            .join(" ");
+        let response = schema.execute(format!("{{ {fields} }}")).await;
+        assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
+        assert_eq!(response.errors[0].message, "Query is too complex.");
+    }
+
+    #[tokio::test]
+    async fn ref_updates_fixed_base_is_charged() {
+        // The 100 base in the refUpdates complexity formula charges for the
+        // collector's fixed work (loading the deduped repo set, the
+        // quarantined set, and the visibility rules, then walking up to 16
+        // keyset pages) even at limit: 1. Four aliases at limit: 1 cost
+        // 4 * (100 + 1 + 1) = 408 > 400 and must be rejected. If the base
+        // disappears, the cost drops to 4 * (1 + 1) = 8 and this test fails.
+        let schema = production_test_schema();
+        let fields = (0..4)
+            .map(|n| format!("r{n}: refUpdates(limit: 1) {{ repo }}"))
+            .collect::<Vec<_>>()
+            .join(" ");
+        let response = schema.execute(format!("{{ {fields} }}")).await;
+        assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
+        assert_eq!(response.errors[0].message, "Query is too complex.");
+    }
+
+    #[derive(Clone, Copy)]
+    struct Nested;
+
+    #[Object]
+    impl Nested {
+        async fn child(&self) -> Nested {
+            Nested
+        }
+
+        async fn value(&self) -> i32 {
+            1
+        }
+    }
+
+    struct CountingQuery(Arc<AtomicUsize>);
+
+    #[Object]
+    impl CountingQuery {
+        async fn nested(&self) -> Nested {
+            self.0.fetch_add(1, Ordering::Relaxed);
+            Nested
+        }
+    }
+
+    #[tokio::test]
+    async fn query_depth_limit_accepts_fourteen_and_rejects_fifteen() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let schema = apply_query_limits(Schema::build(
+            CountingQuery(Arc::clone(&calls)),
+            EmptyMutation,
+            EmptySubscription,
+        ))
+        .finish();
+        let query_at_depth = |depth: usize| {
+            let selection = (2..depth).fold("value".to_string(), |selection, _| {
+                format!("child {{ {selection} }}")
+            });
+            format!("{{ nested {{ {selection} }} }}")
+        };
+
+        let accepted = schema.execute(query_at_depth(GRAPHQL_MAX_DEPTH)).await;
+
+        assert!(
+            accepted.errors.is_empty(),
+            "depth {GRAPHQL_MAX_DEPTH} should be accepted: {:?}",
+            accepted.errors
+        );
+        assert_eq!(calls.swap(0, Ordering::Relaxed), 1);
+
+        let rejected = schema.execute(query_at_depth(GRAPHQL_MAX_DEPTH + 1)).await;
+
+        assert_eq!(rejected.data, Value::Null);
+        assert_eq!(rejected.errors.len(), 1);
+        assert_eq!(rejected.errors[0].message, "Query is nested too deep.");
+        assert_eq!(calls.load(Ordering::Relaxed), 0);
+    }
+
+    #[tokio::test]
+    async fn production_schema_enforces_depth_limit() {
+        let schema = production_test_schema();
+        // Depth 14 is accepted through the production build_schema path.
+        let of_types_14 = (0..9).fold("name".to_string(), |acc, _| format!("ofType {{ {acc} }}"));
+        let query_14 =
+            format!("{{ __schema {{ types {{ fields {{ type {{ {of_types_14} }} }} }} }} }}");
+        let accepted = schema.execute(&query_14).await;
+        assert!(accepted.errors.is_empty(), "{:?}", accepted.errors);
+
+        // Construct an introspection query of depth 15 using __schema.
+        // 1: __schema
+        // 2: types
+        // 3: fields
+        // 4: type
+        // 5..14: ofType (10 times)
+        // 15: name
+        // Total depth = 15 > GRAPHQL_MAX_DEPTH (14).
+        let of_types = (0..10).fold("name".to_string(), |acc, _| format!("ofType {{ {acc} }}"));
+        let query = format!("{{ __schema {{ types {{ fields {{ type {{ {of_types} }} }} }} }} }}");
+
+        let rejected = schema.execute(&query).await;
+        assert_eq!(rejected.data, async_graphql::Value::Null);
+        assert_eq!(rejected.errors.len(), 1);
+        assert_eq!(rejected.errors[0].message, "Query is nested too deep.");
+    }
+
     /// Every `.map_err(` in the GraphQL query/mutation resolvers must route
     /// through the opaque helpers, or discard the error (`|_|`). Same source-
     /// scrape pattern as `api::authz_guard` (#255 review).

crates/gitlawb-node/src/graphql/mutation.rs

@@ -22,6 +22,7 @@ pub struct MutationRoot;
 
 #[Object]
 impl MutationRoot {
+    #[graphql(complexity = "50 + child_complexity")]
     async fn create_task(
         &self,
         ctx: &Context<'_>,
@@ -79,12 +80,17 @@ impl MutationRoot {
             updated_at: now,
             deadline: input.deadline,
         };
-        db.create_task(&task)
+        // Return the stored row, not the pre-insert input: create_task
+        // normalizes the assignee (whitespace-only becomes NULL/open), and a
+        // response that shows the raw input lies about the reservation.
+        let stored = db
+            .create_task(&task)
             .await
             .map_err(crate::graphql::graphql_db_err)?;
-        Ok(AgentTaskType::from(task))
+        Ok(AgentTaskType::from(stored))
     }
 
+    #[graphql(complexity = "50 + child_complexity")]
     async fn claim_task(
         &self,
         ctx: &Context<'_>,
@@ -104,16 +110,25 @@ impl MutationRoot {
             .claim_task(&id, &assignee_did)
             .await
             .map_err(crate::graphql::graphql_db_err)?;
+        // Report the stored assignee form like REST, so subscribers see the
+        // same identity whichever API surface ran the transition.
+        let by_did = task
+            .assignee_did
+            .as_deref()
+            .map(crate::db::trim_assignee_did)
+            .unwrap_or(assignee_did.as_str())
+            .to_string();
         let _ = tx.send(TaskEventBroadcast {
             task_id: id,
             old_status: "pending".to_string(),
             new_status: "claimed".to_string(),
-            by_did: assignee_did,
+            by_did,
             at: Utc::now().to_rfc3339(),
         });
         Ok(AgentTaskType::from(task))
     }
 
+    #[graphql(complexity = "50 + child_complexity")]
     async fn complete_task(
         &self,
         ctx: &Context<'_>,
@@ -137,15 +152,24 @@ impl MutationRoot {
             .await
             .map_err(crate::graphql::graphql_db_err)?
             .ok_or_else(|| async_graphql::Error::new("task not found"))?;
-        if !crate::api::did_matches(caller, existing.assignee_did.as_deref().unwrap_or_default()) {
+        if !crate::api::did_matches(
+            caller,
+            crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
+        ) {
             return Err(async_graphql::Error::new(
                 "only the task assignee can complete it",
             ));
         }
         let task = db
-            .finish_task(&id, "completed", input.result.as_deref())
+            .finish_task(&id, "completed", input.result.as_deref(), caller)
             .await
             .map_err(crate::graphql::graphql_db_err)?;
+        let by_did = task
+            .assignee_did
+            .as_deref()
+            .map(crate::db::trim_assignee_did)
+            .unwrap_or(by_did.as_str())
+            .to_string();
         let _ = tx.send(TaskEventBroadcast {
             task_id: id,
             old_status: "claimed".to_string(),
@@ -156,6 +180,7 @@ impl MutationRoot {
         Ok(AgentTaskType::from(task))
     }
 
+    #[graphql(complexity = "50 + child_complexity")]
     async fn fail_task(
         &self,
         ctx: &Context<'_>,
@@ -178,16 +203,25 @@ impl MutationRoot {
             .await
             .map_err(crate::graphql::graphql_db_err)?
             .ok_or_else(|| async_graphql::Error::new("task not found"))?;
-        if !crate::api::did_matches(caller, existing.assignee_did.as_deref().unwrap_or_default()) {
+        if !crate::api::did_matches(
+            caller,
+            crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
+        ) {
             return Err(async_graphql::Error::new(
                 "only the task assignee can fail it",
             ));
         }
         let reason = input.reason.unwrap_or_default();
         let task = db
-            .finish_task(&id, "failed", Some(&reason))
+            .finish_task(&id, "failed", Some(&reason), caller)
             .await
             .map_err(crate::graphql::graphql_db_err)?;
+        let by_did = task
+            .assignee_did
+            .as_deref()
+            .map(crate::db::trim_assignee_did)
+            .unwrap_or(by_did.as_str())
+            .to_string();
         let _ = tx.send(TaskEventBroadcast {
             task_id: id,
             old_status: "claimed".to_string(),
@@ -573,4 +607,162 @@ mod tests {
             "schema text leaked through the gate: {errs}"
         );
     }
+
+    /// Pre-assigned GraphQL claimTask must reject a stranger (no UCAN leak)
+    /// and admit the reserved assignee.
+    #[sqlx::test]
+    async fn claim_task_honors_preassigned_assignee(pool: PgPool) {
+        let state = crate::test_support::test_state(pool).await;
+        let reserved = "did:key:zGQLRESERVEDAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let thief = "did:key:zGQLTHIEFBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let now = chrono::Utc::now().to_rfc3339();
+        let task = crate::db::AgentTask {
+            id: "task-gql-reserved".into(),
+            repo_id: None,
+            kind: "build".into(),
+            status: "pending".into(),
+            delegator_did: "did:key:zGQLDELEGATORCCCCCCCCCCCCCCCCCCCCCCCCCC".into(),
+            assignee_did: Some(reserved.into()),
+            capability: "repo:write".into(),
+            ucan_token: Some("gql-ucan-secret".into()),
+            payload: None,
+            result: None,
+            created_at: now.clone(),
+            updated_at: now,
+            deadline: None,
+        };
+        state.db.create_task(&task).await.expect("seed");
+        let schema = state.graphql_schema.as_ref();
+
+        let q = |actor: &str| {
+            format!(
+                r#"mutation {{ claimTask(id: "task-gql-reserved", assigneeDid: "{actor}") {{ id status ucanToken }} }}"#
+            )
+        };
+
+        let resp = schema
+            .execute(Request::new(q(thief)).data(AuthenticatedDid(thief.into())))
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.contains("reserved"),
+            "stranger claim must surface reserved deny distinctly: {errs}"
+        );
+        assert!(
+            !errs.contains("gql-ucan-secret"),
+            "UCAN must not leak in GraphQL errors: {errs}"
+        );
+        if let Ok(data) = resp.data.into_json() {
+            let s = data.to_string();
+            assert!(
+                !s.contains("gql-ucan-secret"),
+                "UCAN must not leak in GraphQL data on failed claim: {s}"
+            );
+        }
+        // Parity with REST: failed steal must leave the row untouched.
+        let still = state
+            .db
+            .get_task("task-gql-reserved")
+            .await
+            .expect("get")
+            .expect("task exists");
+        assert_eq!(
+            still.status, "pending",
+            "status must stay pending after steal"
+        );
+        assert_eq!(
+            still.assignee_did.as_deref(),
+            Some(reserved),
+            "assignee_did must stay reserved after steal"
+        );
+
+        let resp = schema
+            .execute(Request::new(q(reserved)).data(AuthenticatedDid(reserved.into())))
+            .await;
+        assert!(
+            errors(&resp).is_empty(),
+            "reserved assignee must claim: {}",
+            errors(&resp)
+        );
+        let data = resp.data.into_json().expect("json data");
+        assert_eq!(data["claimTask"]["status"], "claimed");
+        assert_eq!(data["claimTask"]["ucanToken"], "gql-ucan-secret");
+    }
+
+    /// create_task normalizes a whitespace-only assignee to NULL (open task),
+    /// so the create response must show the stored row, not echo the input: a
+    /// client trusting the echoed "reservation" would believe a claimable task
+    /// is reserved. Before the fix the response was built from the pre-insert
+    /// struct and returned the padded input verbatim.
+    #[sqlx::test]
+    async fn create_task_reports_stored_assignee_not_input(pool: PgPool) {
+        let state = crate::test_support::test_state(pool).await;
+        let caller = "did:key:zGQLCREATORAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let schema = state.graphql_schema.as_ref();
+        let q = format!(
+            r#"mutation {{ createTask(delegatorDid: "{caller}", input: {{ kind: "build", capability: "repo:write", assigneeDid: "   " }}) {{ id assigneeDid }} }}"#
+        );
+        let resp = schema
+            .execute(Request::new(q).data(AuthenticatedDid(caller.into())))
+            .await;
+        let data = resp.data.into_json().expect("json data");
+        assert!(
+            data["createTask"]["assigneeDid"].is_null(),
+            "whitespace-only assigneeDid must come back null (stored shape): {data}"
+        );
+    }
+
+    /// A task seeded with a whitespace-padded assignee must still be
+    /// completable by that assignee over GraphQL: REST and Db::finish_task
+    /// trim the stored value before comparing, and the GraphQL gate must do
+    /// the same or padded reservations finish only through REST.
+    #[sqlx::test]
+    async fn graphql_finish_accepts_padded_stored_assignee(pool: PgPool) {
+        let state = crate::test_support::test_state(pool).await;
+        let assignee = "did:key:zGQLPADDEDAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let now = chrono::Utc::now().to_rfc3339();
+        let task = crate::db::AgentTask {
+            id: "task-gql-padded".into(),
+            repo_id: None,
+            kind: "build".into(),
+            status: "pending".into(),
+            delegator_did: "did:key:zGQLDELEGATORDDDDDDDDDDDDDDDDDDDDDDDDDD".into(),
+            assignee_did: Some(format!("\t{assignee}\n")),
+            capability: "repo:write".into(),
+            ucan_token: None,
+            payload: None,
+            result: None,
+            created_at: now.clone(),
+            updated_at: now,
+            deadline: None,
+        };
+        state.db.create_task(&task).await.expect("seed");
+        let schema = state.graphql_schema.as_ref();
+
+        let claim = format!(
+            r#"mutation {{ claimTask(id: "task-gql-padded", assigneeDid: "{assignee}") {{ id status }} }}"#
+        );
+        let resp = schema
+            .execute(Request::new(claim).data(AuthenticatedDid(assignee.into())))
+            .await;
+        assert!(
+            errors(&resp).is_empty(),
+            "reserved assignee must claim a padded reservation: {}",
+            errors(&resp)
+        );
+
+        let finish = format!(
+            r#"mutation {{ completeTask(id: "task-gql-padded", byDid: "{assignee}", input: {{}}) {{ id status }} }}"#
+        );
+        let resp = schema
+            .execute(Request::new(finish).data(AuthenticatedDid(assignee.into())))
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.is_empty(),
+            "assignee must complete a padded reservation: {errs}"
+        );
+        let data = resp.data.into_json().expect("json data");
+        assert_eq!(data["completeTask"]["status"], "completed");
+    }
 }

crates/gitlawb-node/src/graphql/query.rs

@@ -1,57 +1,154 @@
 use async_graphql::{Context, Object, Result};
+use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
 use std::sync::Arc;
 
-use crate::db::Db;
+use crate::api::events::MAX_VISIBLE_REF_UPDATES;
+use crate::db::{Db, RepoRecord, MAX_VISIBLE_REPO_PAGE_SIZE, MAX_VISIBLE_TASK_PAGE_SIZE};
 
-use super::types::{AgentTaskType, RefUpdateType, RepoType};
+use super::types::{AgentTaskType, RefUpdateType, RepoPageType, RepoType};
+
+fn repo_type(repo: RepoRecord) -> RepoType {
+    RepoType {
+        name: repo.name,
+        owner_did: repo.owner_did,
+        description: repo.description,
+        default_branch: repo.default_branch,
+        created_at: repo.created_at.to_rfc3339(),
+    }
+}
+
+fn repo_cursor(repo: &RepoRecord) -> String {
+    URL_SAFE_NO_PAD.encode(
+        serde_json::to_vec(&(&repo.owner_did, &repo.name))
+            .expect("a pair of strings is JSON serializable"),
+    )
+}
+
+fn parse_repo_cursor(cursor: &str) -> Result<(String, String)> {
+    // A position contains only public response fields, not a database id or
+    // authority. Every page re-evaluates visibility for the current caller.
+    if cursor.len() > 4096 {
+        return Err(async_graphql::Error::new("invalid repository cursor"));
+    }
+    let bytes = URL_SAFE_NO_PAD
+        .decode(cursor)
+        .map_err(|_| async_graphql::Error::new("invalid repository cursor"))?;
+    let position: (String, String) = serde_json::from_slice(&bytes)
+        .map_err(|_| async_graphql::Error::new("invalid repository cursor"))?;
+    // PostgreSQL text cannot contain NUL. Reject it as malformed input before
+    // binding either cursor component, rather than reporting a database error.
+    if position.0.contains('\0') || position.1.contains('\0') {
+        return Err(async_graphql::Error::new("invalid repository cursor"));
+    }
+    Ok(position)
+}
 
 pub struct QueryRoot;
 
 #[Object]
 impl QueryRoot {
+    // DB-backed roots carry a base cost so aliases consume the request budget
+    // even when each alias selects only one inexpensive response field.
+    // `repos` performs the same full corpus scan as a max-size `reposPage`
+    // (list_visible_repos_page with MAX_VISIBLE_REPO_PAGE_SIZE + 1), so it
+    // is priced as the full page it is, not as a cheap alias.
+    #[graphql(complexity = "50 + MAX_VISIBLE_REPO_PAGE_SIZE + child_complexity")]
+    /// Complete visible repository list, up to the documented maximum number
+    /// of entries. Larger lists must use reposPage; this field returns an
+    /// error instead of truncating silently.
     async fn repos(&self, ctx: &Context<'_>) -> Result<Vec<RepoType>> {
         let db = ctx.data_unchecked::<Arc<Db>>();
-        let repos = db
-            .list_all_repos_deduped()
-            .await
-            .map_err(crate::graphql::graphql_db_err)?;
-
-        // Apply the same "/" visibility gate the REST/per-repo endpoints use so
-        // this surface does not enumerate private repos (#97). The caller DID is
-        // threaded onto the context by optional_signature; absent = anonymous.
         let caller = ctx
             .data::<crate::auth::AuthenticatedDid>()
             .ok()
             .map(|d| d.0.as_str());
-        let ids: Vec<String> = repos.iter().map(|r| r.id.clone()).collect();
-        let rules_by_repo = db
-            .list_visibility_rules_for_repos(&ids)
+        let mut repos = db
+            .list_visible_repos_page(caller, None, MAX_VISIBLE_REPO_PAGE_SIZE + 1)
             .await
             .map_err(crate::graphql::graphql_db_err)?;
-
-        Ok(repos
-            .into_iter()
-            .filter(|r| {
-                let rules = rules_by_repo.get(&r.id).map(Vec::as_slice).unwrap_or(&[]);
-                crate::visibility::listable_at_root(rules, r.is_public, &r.owner_did, caller)
-            })
-            .map(|r| RepoType {
-                name: r.name,
-                owner_did: r.owner_did,
-                description: r.description,
-                default_branch: r.default_branch,
-                created_at: r.created_at.to_rfc3339(),
-            })
-            .collect())
+        if repos.len() > MAX_VISIBLE_REPO_PAGE_SIZE {
+            return Err(async_graphql::Error::new(format!(
+                "repository list exceeds {} entries; use reposPage with limit and after",
+                MAX_VISIBLE_REPO_PAGE_SIZE
+            )));
+        }
+        // Preserve the legacy activity ordering for complete, small lists.
+        repos.sort_by_key(|repo| std::cmp::Reverse(repo.updated_at));
+        Ok(repos.into_iter().map(repo_type).collect())
     }
 
+    /// Bounded visible repositories ordered by owner and name. Continue with
+    /// endCursor while hasNextPage is true. Pages are not a database snapshot.
+    // Every page sorts the full deduplicated corpus under COLLATE "C"
+    // (list_visible_repos_page documents the accepted O(total repos) bound),
+    // so the price charges the scan, not the emitted rows: a limit:1 page
+    // costs the same as a max-size page, the way `repos` is priced.
+    #[graphql(complexity = "50 + MAX_VISIBLE_REPO_PAGE_SIZE + child_complexity")]
+    async fn repos_page(
+        &self,
+        ctx: &Context<'_>,
+        #[graphql(
+            default = 50,
+            desc = "Page size from 1 to the documented maximum; other values are rejected."
+        )]
+        limit: i64,
+        after: Option<String>,
+    ) -> Result<RepoPageType> {
+        if !(1..=MAX_VISIBLE_REPO_PAGE_SIZE as i64).contains(&limit) {
+            return Err(async_graphql::Error::new(format!(
+                "limit must be between 1 and {}",
+                MAX_VISIBLE_REPO_PAGE_SIZE
+            )));
+        }
+        let after = after.as_deref().map(parse_repo_cursor).transpose()?;
+        let caller = ctx
+            .data::<crate::auth::AuthenticatedDid>()
+            .ok()
+            .map(|d| d.0.as_str());
+        let db = ctx.data_unchecked::<Arc<Db>>();
+        let mut repos = db
+            .list_visible_repos_page(
+                caller,
+                after
+                    .as_ref()
+                    .map(|(owner, name)| (owner.as_str(), name.as_str())),
+                limit as usize + 1,
+            )
+            .await
+            .map_err(crate::graphql::graphql_db_err)?;
+        let has_next_page = repos.len() > limit as usize;
+        repos.truncate(limit as usize);
+        let end_cursor = repos.last().map(repo_cursor);
+        Ok(RepoPageType {
+            nodes: repos.into_iter().map(repo_type).collect(),
+            has_next_page,
+            end_cursor,
+        })
+    }
+
+    // Complexity is additive with a base reflecting the collector's fixed scan:
+    // every request loads the full deduped repo set, the quarantined set, and
+    // the visibility rules, then walks up to max(limit, 2048) event rows past
+    // withheld ones, regardless of `limit`. A multiplicative price both
+    // undercharged limit:1 (51 for that fixed work) and made a request at the
+    // documented maximum unreachable (two fields at max limit cost 450 > 400).
+    //
+    // Decision on the collector floor (review #465, 3rd round): the up-to-2048
+    // walk is bounded constant overhead that does not scale with `limit`, so
+    // it is priced in the fixed base, not the limit term; the limit term
+    // prices only the emitted rows. Pricing the walk at face value in the base
+    // would put every refUpdates query over the 400 budget and make the field
+    // unusable, so the base prices that fixed scan as a lump sum instead.
+    #[graphql(
+        complexity = "100 + (limit.clamp(0, MAX_VISIBLE_REF_UPDATES) as usize) + child_complexity"
+    )]
     async fn ref_updates(
         &self,
         ctx: &Context<'_>,
         repo: Option<String>,
         #[graphql(
             default = 20,
-            desc = "Max 200; larger requests return the newest 200 rows (no continuation cursor)."
+            desc = "Requests above the documented maximum return the newest rows up to that maximum (no continuation cursor)."
         )]
         limit: i64,
     ) -> Result<Vec<RefUpdateType>> {
@@ -105,6 +202,14 @@ impl QueryRoot {
         Ok(resolved)
     }
 
+    // Complexity is additive: list_tasks is a single LIMIT-bounded query whose
+    // cost does not scale with the selected fields, so the price is the base
+    // plus the clamped limit plus the child fields. A multiplicative price
+    // made a request at the documented maximum unreachable (two fields at max
+    // limit cost 450 > 400).
+    #[graphql(
+        complexity = "50 + (limit.clamp(0, MAX_VISIBLE_TASK_PAGE_SIZE) as usize) + child_complexity"
+    )]
     async fn tasks(
         &self,
         ctx: &Context<'_>,
@@ -112,7 +217,7 @@ impl QueryRoot {
         assignee_did: Option<String>,
         #[graphql(
             default = 50,
-            desc = "Max 200; larger requests are clamped to 200 (no error). Negative values clamp to 0."
+            desc = "Requests above the documented maximum are clamped to it (no error). Negative values clamp to 0."
         )]
         limit: i64,
     ) -> Result<Vec<AgentTaskType>> {
@@ -120,7 +225,7 @@ impl QueryRoot {
         // Clamp before SQL: a negative LIMIT is a client fault that Postgres
         // rejects with 2201W, which would otherwise trip the opaque DB path
         // and write an error-level log on every probe (#250 review).
-        let limit = limit.clamp(0, 200);
+        let limit = limit.clamp(0, MAX_VISIBLE_TASK_PAGE_SIZE);
         let tasks = db
             .list_tasks(status.as_deref(), assignee_did.as_deref(), limit)
             .await
@@ -128,6 +233,7 @@ impl QueryRoot {
         Ok(tasks.into_iter().map(AgentTaskType::from).collect())
     }
 
+    #[graphql(complexity = "50 + child_complexity")]
     async fn task(&self, ctx: &Context<'_>, id: String) -> Result<Option<AgentTaskType>> {
         let db = ctx.data_unchecked::<Arc<Db>>();
         let t = db
@@ -141,12 +247,642 @@ impl QueryRoot {
 #[cfg(test)]
 mod tests {
     use crate::db::{Db, ReceivedRefUpdate, RepoRecord};
+    use base64::Engine;
     use chrono::Utc;
     use sqlx::PgPool;
     use std::sync::Arc;
 
     const OWNER: &str = "did:key:z6MkOwner";
 
+    #[sqlx::test]
+    async fn repos_legacy_rejects_overflow_and_pages_reach_every_visible_repo(pool: PgPool) {
+        let db = db(pool).await;
+        let total = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE + 5;
+        for index in 0..total {
+            let name = format!("repo-{index:03}");
+            db.create_repo(&repo(&name, OWNER, &name, true))
+                .await
+                .unwrap();
+        }
+        // The SQL helper itself must bound materialization, even for a caller
+        // accidentally requesting an unlimited page.
+        assert_eq!(
+            db.list_visible_repos_page(None, None, usize::MAX)
+                .await
+                .unwrap()
+                .len(),
+            crate::db::MAX_VISIBLE_REPO_PAGE_SIZE + 1
+        );
+        let schema = schema(db);
+        let legacy = anon(&schema, "{ repos { name } }").await;
+        assert_eq!(legacy.errors.len(), 1);
+        assert!(legacy.errors[0].message.contains("use reposPage"));
+        assert_eq!(legacy.data, async_graphql::Value::Null);
+
+        let mut cursor = None;
+        let mut names = Vec::new();
+        loop {
+            let response = schema.execute(
+                async_graphql::Request::new(
+                    "query($after: String) { reposPage(limit: 50, after: $after) { nodes { name } hasNextPage endCursor } }",
+                ).variables(async_graphql::Variables::from_json(serde_json::json!({"after": cursor}))),
+            ).await;
+            assert!(response.errors.is_empty(), "{:?}", response.errors);
+            let json = response.data.into_json().unwrap();
+            let page = &json["reposPage"];
+            let rows = page["nodes"].as_array().unwrap();
+            assert!(rows.len() <= 50);
+            names.extend(
+                rows.iter()
+                    .map(|row| row["name"].as_str().unwrap().to_owned()),
+            );
+            if !page["hasNextPage"].as_bool().unwrap() {
+                break;
+            }
+            let next = page["endCursor"].as_str().unwrap().to_owned();
+            assert_ne!(cursor.as_ref(), Some(&next));
+            cursor = Some(next);
+            assert!(names.len() <= crate::db::MAX_VISIBLE_REPO_PAGE_SIZE);
+        }
+        assert_eq!(
+            names,
+            (0..total)
+                .map(|index| format!("repo-{index:03}"))
+                .collect::<Vec<_>>()
+        );
+    }
+
+    #[sqlx::test]
+    async fn repos_legacy_accepts_exactly_the_visible_bound(pool: PgPool) {
+        let db = db(pool).await;
+        let base_time = Utc::now();
+        let total = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE;
+        for index in 0..total {
+            let name = format!("repo-{index:03}");
+            let mut r = repo(&name, OWNER, &name, true);
+            // Anti-correlate updated_at with name, creation order, and index
+            // so only ordering by updated_at DESC can satisfy the expectation.
+            let offset = (index * 37) % total;
+            r.updated_at = base_time + chrono::Duration::seconds(offset as i64);
+            db.create_repo(&r).await.unwrap();
+        }
+        db.create_repo(&repo("hidden", OWNER, "hidden", false))
+            .await
+            .unwrap();
+        let response = anon(&schema(db), "{ repos { name } }").await;
+        assert!(response.errors.is_empty(), "{:?}", response.errors);
+        let repos = response.data.into_json().unwrap()["repos"]
+            .as_array()
+            .unwrap()
+            .clone();
+        assert_eq!(repos.len(), total);
+        let names: Vec<_> = repos.iter().filter_map(|r| r["name"].as_str()).collect();
+        assert!(
+            !names.contains(&"hidden"),
+            "hidden repo must be excluded from legacy repos response"
+        );
+        let mut expected_indices = (0..total).collect::<Vec<_>>();
+        expected_indices.sort_by_key(|&idx| std::cmp::Reverse((idx * 37) % total));
+        let expected_names: Vec<String> = expected_indices
+            .into_iter()
+            .map(|index| format!("repo-{index:03}"))
+            .collect();
+        assert_eq!(
+            names,
+            expected_names
+                .iter()
+                .map(|s| s.as_str())
+                .collect::<Vec<_>>(),
+            "legacy repos must retain activity ordering (most recently updated first)"
+        );
+    }
+
+    #[sqlx::test]
+    async fn repos_page_pagination_crosses_owner_boundary(pool: PgPool) {
+        let db = db(pool).await;
+        let owner1 = "did:key:z6MkaOwner1";
+        let owner2 = "did:key:z6MkbOwner2";
+        let earlier_owner = "did:web:example.com";
+        // This owner sorts first only after did:key normalization. Its name
+        // sorts past the first cursor, so an unqualified OR name predicate
+        // would incorrectly serve it again on the next page.
+        //
+        // Rows are inserted out of order and not in reverse asserted order.
+        // Assigned IDs do not track name order, and owner2 contains both
+        // "B-repo" and "a-repo" whose byte order (B < a) differs from their
+        // case-folded order (a < b), ensuring lower(d.name) cannot satisfy
+        // the keyset query. Sized to 6 entries so the last page is full-sized
+        // (limit 2), proving hasNextPage distinguishes a full terminal page.
+        db.create_repo(&repo("r5", owner2, "c-repo", true))
+            .await
+            .unwrap();
+        db.create_repo(&repo("r9", earlier_owner, "zz-repo", true))
+            .await
+            .unwrap();
+        db.create_repo(&repo("r3", owner2, "B-repo", true))
+            .await
+            .unwrap();
+        db.create_repo(&repo("r8", owner1, "z-repo", true))
+            .await
+            .unwrap();
+        db.create_repo(&repo("r2", owner1, "b-repo", true))
+            .await
+            .unwrap();
+        db.create_repo(&repo("r1", owner2, "a-repo", true))
+            .await
+            .unwrap();
+
+        let schema = schema(db);
+        let page1_resp = anon(
+            &schema,
+            "{ reposPage(limit: 2) { nodes { name ownerDid } hasNextPage endCursor } }",
+        )
+        .await;
+        assert!(page1_resp.errors.is_empty(), "{:?}", page1_resp.errors);
+        let p1 = page1_resp.data.into_json().unwrap()["reposPage"].clone();
+        assert_eq!(p1["hasNextPage"], true);
+        assert_eq!(
+            p1["nodes"],
+            serde_json::json!([
+                {"name": "zz-repo", "ownerDid": earlier_owner},
+                {"name": "b-repo", "ownerDid": owner1},
+            ])
+        );
+        let cursor = p1["endCursor"].as_str().unwrap();
+
+        let page2_query = format!(
+            "{{ reposPage(limit: 2, after: \"{cursor}\") {{ nodes {{ name ownerDid }} hasNextPage endCursor }} }}"
+        );
+        let page2_resp = anon(&schema, &page2_query).await;
+        assert!(page2_resp.errors.is_empty(), "{:?}", page2_resp.errors);
+        let p2 = page2_resp.data.into_json().unwrap()["reposPage"].clone();
+        assert_eq!(p2["hasNextPage"], true);
+        assert_eq!(
+            p2["nodes"],
+            serde_json::json!([
+                {"name": "z-repo", "ownerDid": owner1},
+                {"name": "B-repo", "ownerDid": owner2},
+            ])
+        );
+        let cursor = p2["endCursor"].as_str().unwrap();
+        let page3_query = format!(
+            "{{ reposPage(limit: 2, after: \"{cursor}\") {{ nodes {{ name ownerDid }} hasNextPage endCursor }} }}"
+        );
+        let page3_resp = anon(&schema, &page3_query).await;
+        assert!(page3_resp.errors.is_empty(), "{:?}", page3_resp.errors);
+        let p3 = page3_resp.data.into_json().unwrap()["reposPage"].clone();
+        assert_eq!(p3["hasNextPage"], false);
+        assert_eq!(
+            p3["nodes"],
+            serde_json::json!([
+                {"name": "a-repo", "ownerDid": owner2},
+                {"name": "c-repo", "ownerDid": owner2},
+            ])
+        );
+    }
+
+    #[sqlx::test]
+    async fn repos_page_at_documented_maximum_distinguishes_terminal_page(pool: PgPool) {
+        let db = db(pool).await;
+        let total = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE;
+        for index in 0..total {
+            let name = format!("repo-{index:03}");
+            db.create_repo(&repo(&name, OWNER, &name, true))
+                .await
+                .unwrap();
+        }
+        let schema = schema(db.clone());
+        let query =
+            format!("{{ reposPage(limit: {total}) {{ nodes {{ name }} hasNextPage endCursor }} }}");
+        let resp200 = anon(&schema, &query).await;
+        assert!(resp200.errors.is_empty(), "{:?}", resp200.errors);
+        let p200 = resp200.data.into_json().unwrap()["reposPage"].clone();
+        assert_eq!(p200["hasNextPage"], false);
+        let nodes200 = p200["nodes"].as_array().unwrap().clone();
+        assert_eq!(nodes200.len(), total);
+
+        // Adding row 201 keeps page 1 content identical while flipping hasNextPage to true.
+        let name201 = format!("repo-{total:03}");
+        db.create_repo(&repo(&name201, OWNER, &name201, true))
+            .await
+            .unwrap();
+        let resp201 = anon(&schema, &query).await;
+        assert!(resp201.errors.is_empty(), "{:?}", resp201.errors);
+        let p201 = resp201.data.into_json().unwrap()["reposPage"].clone();
+        assert_eq!(p201["hasNextPage"], true);
+        assert_eq!(p201["nodes"], p200["nodes"]);
+        assert_eq!(p201["endCursor"], p200["endCursor"]);
+    }
+
+    #[sqlx::test]
+    async fn repos_page_visibility_matches_the_shared_gate(pool: PgPool) {
+        use crate::db::VisibilityMode;
+        let db = db(pool).await;
+        let reader = "did:key:zReader";
+        for (id, public) in [
+            ("open", true),
+            ("private", false),
+            ("root-deny", true),
+            ("root-reader", false),
+            ("subtree", true),
+            ("root-tie", true),
+            ("odd-star", true),
+            ("malformed-readers", true),
+            ("non-array-readers", true),
+            ("non-string-reader", true),
+            ("quarantined", true),
+        ] {
+            db.create_repo(&repo(id, OWNER, id, public)).await.unwrap();
+        }
+        // Canonical and mirror copies must still collapse before pagination.
+        db.create_repo(&repo("z6MkOwner/open", "z6MkOwner", "open", true))
+            .await
+            .unwrap();
+        db.create_repo(&repo(
+            "other-method",
+            "did:web:z6MkOwner",
+            "other-method",
+            false,
+        ))
+        .await
+        .unwrap();
+        db.set_repo_quarantine("quarantined", true).await.unwrap();
+        for (id, glob, readers) in [
+            ("root-deny", "/", vec![]),
+            ("root-reader", "/**", vec![reader.to_owned()]),
+            ("subtree", "/secret/**", vec![]),
+            ("root-tie", "/", vec![reader.to_owned()]),
+            ("root-tie", "/**", vec![]),
+            ("odd-star", "/*", vec![]),
+            ("malformed-readers", "/", vec![]),
+            ("non-array-readers", "/", vec![]),
+            ("non-string-reader", "/", vec![]),
+        ] {
+            db.set_visibility_rule(id, glob, VisibilityMode::B, &readers, OWNER)
+                .await
+                .unwrap();
+        }
+        // The typed setter cannot create malformed JSON, but existing TEXT rows
+        // can contain it. Such a rule must deny this repo, not abort the page.
+        // All three deny arms of the reader_dids predicate are pinned here:
+        // (a) invalid JSON, (b) valid JSON that is not an array, (c) an array
+        // containing a non-string member. The Rust gate parses reader_dids as
+        // Vec<String> with unwrap_or_default(), so (b) and (c) also fail to
+        // parse and deny; the SQL CASE must agree on every shape.
+        let rows = sqlx::query("UPDATE visibility_rules SET reader_dids = $1 WHERE repo_id = $2")
+            .bind("not JSON")
+            .bind("malformed-readers")
+            .execute(db.pool())
+            .await
+            .unwrap();
+        assert_eq!(rows.rows_affected(), 1);
+        let rows = sqlx::query("UPDATE visibility_rules SET reader_dids = $1 WHERE repo_id = $2")
+            .bind("\"just-a-string\"")
+            .bind("non-array-readers")
+            .execute(db.pool())
+            .await
+            .unwrap();
+        assert_eq!(rows.rows_affected(), 1);
+        // The named reader is listed, but the non-string member poisons the
+        // whole list: both sides must still deny them.
+        let rows = sqlx::query("UPDATE visibility_rules SET reader_dids = $1 WHERE repo_id = $2")
+            .bind("[\"did:key:zReader\", 42]")
+            .bind("non-string-reader")
+            .execute(db.pool())
+            .await
+            .unwrap();
+        assert_eq!(rows.rows_affected(), 1);
+        let all = db.list_all_repos_deduped().await.unwrap();
+        for caller in [
+            None,
+            Some(OWNER),
+            Some("z6MkOwner"),
+            Some(reader),
+            Some("zReader"),
+            Some("did:web:z6MkOwner"),
+        ] {
+            let mut expected = Vec::new();
+            for record in &all {
+                let rules = db.list_visibility_rules(&record.id).await.unwrap();
+                if crate::visibility::listable_at_root(
+                    &rules,
+                    record.is_public,
+                    &record.owner_did,
+                    caller,
+                ) {
+                    expected.push(record.id.clone());
+                }
+            }
+            expected.sort();
+            let mut actual = db
+                .list_visible_repos_page(caller, None, usize::MAX)
+                .await
+                .unwrap()
+                .into_iter()
+                .map(|record| record.id)
+                .collect::<Vec<_>>();
+            actual.sort();
+            assert_eq!(actual, expected, "caller {caller:?}");
+        }
+        let schema = schema(db);
+        for caller in [None, Some("did:key:zUnauthorized")] {
+            let query = "{ repos { name } }";
+            let response = match caller {
+                None => anon(&schema, query).await,
+                Some(caller) => authed(&schema, query, caller).await,
+            };
+            assert!(response.errors.is_empty(), "{:?}", response.errors);
+            let mut names: Vec<String> = response.data.into_json().unwrap()["repos"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .map(|repo| repo["name"].as_str().unwrap().to_owned())
+                .collect();
+            names.sort();
+            assert_eq!(names, ["odd-star", "open", "subtree"]);
+        }
+        let query = "{ reposPage(limit: 1) { nodes { name ownerDid } hasNextPage endCursor } }";
+        let response = anon(&schema, query).await;
+        assert!(response.errors.is_empty(), "{:?}", response.errors);
+        let json = response.data.into_json().unwrap();
+        let page = &json["reposPage"];
+        assert_eq!(
+            page["nodes"],
+            serde_json::json!([{"name": "odd-star", "ownerDid": OWNER}])
+        );
+        assert_eq!(page["hasNextPage"], true);
+        let cursor = super::parse_repo_cursor(page["endCursor"].as_str().unwrap()).unwrap();
+        assert_eq!(cursor, (OWNER.to_owned(), "odd-star".to_owned()));
+        assert!(!json.to_string().contains("private"));
+        assert!(!json.to_string().contains("quarantined"));
+
+        // Route-level check with unauthorized authenticated caller:
+        // Excludes private, quarantined, and root-deny repos, matching anonymous behavior.
+        let all_query = "{ reposPage(limit: 50) { nodes { name } hasNextPage } }";
+        let unauth_response = authed(&schema, all_query, "did:key:zUnauthorized").await;
+        assert!(
+            unauth_response.errors.is_empty(),
+            "{:?}",
+            unauth_response.errors
+        );
+        let unauth_json = unauth_response.data.into_json().unwrap();
+        let unauth_names: Vec<&str> = unauth_json["reposPage"]["nodes"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .filter_map(|n| n["name"].as_str())
+            .collect();
+        assert_eq!(unauth_names, vec!["odd-star", "open", "subtree"]);
+        let unauth_str = unauth_json.to_string();
+        assert!(!unauth_str.contains("private"));
+        assert!(!unauth_str.contains("quarantined"));
+        assert!(!unauth_str.contains("root-deny"));
+        assert!(!unauth_str.contains("root-reader"));
+        assert!(!unauth_str.contains("root-tie"));
+
+        // Reader caller gets root-reader and root-tie, but still excludes private, quarantined, root-deny.
+        let reader_response = authed(&schema, all_query, reader).await;
+        assert!(
+            reader_response.errors.is_empty(),
+            "{:?}",
+            reader_response.errors
+        );
+        let reader_json = reader_response.data.into_json().unwrap();
+        let reader_names: Vec<&str> = reader_json["reposPage"]["nodes"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .filter_map(|n| n["name"].as_str())
+            .collect();
+        assert_eq!(
+            reader_names,
+            vec!["odd-star", "open", "root-reader", "subtree"]
+        );
+        let reader_str = reader_json.to_string();
+        assert!(!reader_str.contains("private"));
+        assert!(!reader_str.contains("quarantined"));
+        assert!(!reader_str.contains("root-deny"));
+        assert!(!reader_str.contains("root-tie"));
+    }
+
+    #[sqlx::test]
+    async fn repos_page_rechecks_cursor_authority_and_exact_boundary(pool: PgPool) {
+        let db = db(pool).await;
+        let visible = repo("visible", OWNER, "a-visible", true);
+        db.create_repo(&visible).await.unwrap();
+        db.create_repo(&repo("hidden", OWNER, "z-private", false))
+            .await
+            .unwrap();
+        let schema = schema(db);
+        let first = anon(
+            &schema,
+            "{ reposPage(limit: 1) { nodes { name } hasNextPage endCursor } }",
+        )
+        .await;
+        assert!(first.errors.is_empty());
+        let first = first.data.into_json().unwrap();
+        assert_eq!(first["reposPage"]["hasNextPage"], false);
+        let query = format!(
+            "{{ reposPage(limit: 1, after: \"{}\") {{ nodes {{ name }} hasNextPage endCursor }} }}",
+            super::repo_cursor(&visible)
+        );
+        let owner = authed(&schema, &query, OWNER).await;
+        assert!(owner.errors.is_empty());
+        assert_eq!(
+            owner.data.into_json().unwrap()["reposPage"]["nodes"][0]["name"],
+            "z-private"
+        );
+        let anon = anon(&schema, &query).await;
+        assert!(anon.errors.is_empty());
+        assert_eq!(
+            anon.data.into_json().unwrap()["reposPage"],
+            serde_json::json!({
+                "nodes": [], "hasNextPage": false, "endCursor": null
+            })
+        );
+        let unauth = authed(&schema, &query, "did:key:zUnauthorized").await;
+        assert!(unauth.errors.is_empty());
+        assert_eq!(
+            unauth.data.into_json().unwrap()["reposPage"],
+            serde_json::json!({
+                "nodes": [], "hasNextPage": false, "endCursor": null
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn repos_page_rejects_nul_cursors_before_database_access() {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        let schema = schema(Arc::new(Db::for_testing(pool)));
+        for position in [("did:key:zOwner\0", "repo"), ("did:key:zOwner", "repo\0")] {
+            let cursor = super::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&position).unwrap());
+            let query = format!("{{ reposPage(after: \"{cursor}\") {{ hasNextPage }} }}");
+            let response =
+                tokio::time::timeout(std::time::Duration::from_secs(1), anon(&schema, &query))
+                    .await
+                    .expect("invalid cursors must be rejected without accessing the database");
+            assert_eq!(response.errors.len(), 1);
+            assert_eq!(response.errors[0].message, "invalid repository cursor");
+            assert_eq!(response.data, async_graphql::Value::Null);
+        }
+        let position = ("did:key:zOwner", "repo");
+        let cursor = super::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&position).unwrap());
+        assert_eq!(
+            super::parse_repo_cursor(&cursor).unwrap(),
+            (position.0.to_owned(), position.1.to_owned())
+        );
+    }
+
+    #[tokio::test]
+    async fn repos_page_rejects_invalid_inputs_before_database_access() {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        let schema = schema(Arc::new(Db::for_testing(pool)));
+        let over_max = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE + 1;
+        let expected_limit_message = format!(
+            "limit must be between 1 and {}",
+            crate::db::MAX_VISIBLE_REPO_PAGE_SIZE
+        );
+        // Each input pins its own rejection message: the loop must not accept
+        // either message for every query, or a misattributed message (e.g. the
+        // limit arm returning "invalid repository cursor") stays invisible.
+        for (query, expected_message) in [
+            (
+                "{ reposPage(limit: 0) { hasNextPage } }".to_owned(),
+                expected_limit_message.clone(),
+            ),
+            (
+                "{ reposPage(limit: -1) { hasNextPage } }".to_owned(),
+                expected_limit_message.clone(),
+            ),
+            (
+                format!("{{ reposPage(limit: {over_max}) {{ hasNextPage }} }}"),
+                expected_limit_message.clone(),
+            ),
+            (
+                "{ reposPage(after: \"invalid!\") { hasNextPage } }".to_owned(),
+                "invalid repository cursor".to_owned(),
+            ),
+        ] {
+            let response =
+                tokio::time::timeout(std::time::Duration::from_secs(1), anon(&schema, &query))
+                    .await
+                    .unwrap();
+            assert_eq!(response.errors.len(), 1);
+            assert_eq!(
+                response.errors[0].message, expected_message,
+                "wrong rejection message for query: {query}"
+            );
+        }
+        // Valid base64 and valid cursor JSON: only the length guard rejects it.
+        let oversized = super::URL_SAFE_NO_PAD
+            .encode(serde_json::to_vec(&("did:key:reader", "a".repeat(3100))).unwrap());
+        assert!(oversized.len() > 4096);
+        assert!(serde_json::from_slice::<(String, String)>(
+            &super::URL_SAFE_NO_PAD.decode(&oversized).unwrap()
+        )
+        .is_ok());
+        assert!(super::parse_repo_cursor(&oversized).is_err());
+    }
+
+    #[tokio::test]
+    async fn parse_repo_cursor_rejects_wrong_shape_json() {
+        // Valid base64 and valid JSON, but not a (String, String) pair: the
+        // wrong-shape arm must report "invalid repository cursor", exactly,
+        // before any database access.
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        let schema = schema(Arc::new(Db::for_testing(pool)));
+        for payload in [
+            serde_json::json!(["only-one"]),
+            serde_json::json!(["a", "b", "c"]),
+            serde_json::json!({"owner": "x", "name": "y"}),
+            serde_json::json!(42),
+            serde_json::json!(["a", 1]),
+        ] {
+            let cursor = super::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload).unwrap());
+            let err = super::parse_repo_cursor(&cursor).unwrap_err();
+            assert_eq!(err.message, "invalid repository cursor");
+            let query = format!("{{ reposPage(after: \"{cursor}\") {{ hasNextPage }} }}");
+            let response =
+                tokio::time::timeout(std::time::Duration::from_secs(1), anon(&schema, &query))
+                    .await
+                    .expect("wrong-shape cursors must be rejected without accessing the database");
+            assert_eq!(response.errors.len(), 1);
+            assert_eq!(response.errors[0].message, "invalid repository cursor");
+            assert_eq!(response.data, async_graphql::Value::Null);
+        }
+    }
+
+    #[tokio::test]
+    async fn longest_repo_name_cursor_round_trips() {
+        // #465 4th round: a cursor the server emits must always parse back.
+        // Repo names are bounded at the create/fork/mirror write paths, so the
+        // longest admissible name must produce a cursor under the 4096-byte
+        // parse limit.
+        let name = "a".repeat(crate::db::MAX_REPO_NAME_LEN);
+        let record = repo("long", OWNER, &name, true);
+        let cursor = super::repo_cursor(&record);
+        assert!(
+            cursor.len() <= 4096,
+            "emitted cursor must stay under the parse limit: {}",
+            cursor.len()
+        );
+        assert_eq!(
+            super::parse_repo_cursor(&cursor).unwrap(),
+            (OWNER.to_owned(), name)
+        );
+    }
+
+    /// A negative refUpdates(limit) must be accepted and clamped to zero rows,
+    /// like tasks(limit: -1): the complexity formula's `limit.clamp(0, ...)`
+    /// prices it at the floor and the shared collector returns before any
+    /// database access. Mirrors tasks_negative_limit_clamped.
+    #[tokio::test]
+    async fn ref_updates_negative_limit_clamped() {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        let schema = schema(Arc::new(Db::for_testing(pool)));
+        let resp = tokio::time::timeout(
+            std::time::Duration::from_secs(1),
+            anon(&schema, "{ refUpdates(limit: -1) { repo } }"),
+        )
+        .await
+        .expect("negative limit must clamp without accessing the database");
+        assert!(
+            resp.errors.is_empty(),
+            "negative limit must clamp, not fail: {:?}",
+            resp.errors
+        );
+        assert_eq!(count(&resp), 0);
+    }
+
+    /// The mirror write path binds peer-supplied names verbatim, so it must
+    /// enforce the same bound as the local write paths: an overlong peer name
+    /// would otherwise make reposPage emit an unparseable cursor. The
+    /// rejection happens before any database access.
+    #[tokio::test]
+    async fn mirror_repo_rejects_overlong_name() {
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy("postgres://localhost/unused")
+            .unwrap();
+        let db = Db::for_testing(pool);
+        let name = "b".repeat(crate::db::MAX_REPO_NAME_LEN + 1);
+        let err = db
+            .upsert_mirror_repo("z6MkLong", &name, "/tmp/long", None, false)
+            .await
+            .unwrap_err();
+        assert!(
+            err.to_string().contains("exceeds"),
+            "unexpected error: {err}"
+        );
+    }
+
     async fn db(pool: PgPool) -> Arc<Db> {
         let db = Db::for_testing(pool);
         db.run_migrations().await.unwrap();
@@ -475,12 +1211,13 @@ mod tests {
         assert_eq!(count_tasks(&resp), 0);
     }
 
-    /// #255: ceiling of the tasks(limit) clamp must be held (schema promises Max 200).
+    /// #255: ceiling of the tasks(limit) clamp must be held (schema promises the
+    /// documented maximum).
     #[sqlx::test]
-    async fn tasks_limit_ceiling_clamped_to_200(pool: PgPool) {
+    async fn tasks_limit_ceiling_clamped_to_max(pool: PgPool) {
         let db = db(pool).await;
         let now = Utc::now().to_rfc3339();
-        for i in 0..201 {
+        for i in 0..crate::db::MAX_VISIBLE_TASK_PAGE_SIZE + 1 {
             db.create_task(&crate::db::AgentTask {
                 id: format!("task-ceil-{i}"),
                 repo_id: None,
@@ -501,7 +1238,49 @@ mod tests {
         }
         let schema = schema(db);
         let resp = anon(&schema, "{ tasks(limit: 5000) { id } }").await;
-        assert_eq!(count_tasks(&resp), 200, "limit above 200 must clamp to 200");
+        assert_eq!(
+            count_tasks(&resp),
+            crate::db::MAX_VISIBLE_TASK_PAGE_SIZE as usize,
+            "limit above the maximum must clamp to MAX_VISIBLE_TASK_PAGE_SIZE"
+        );
+    }
+
+    /// A limit above MAX_VISIBLE_REF_UPDATES must reach the resolver clamped,
+    /// not be rejected: the complexity formula clamps the priced limit (so the
+    /// document passes validation) and the shared collector clamps the rows
+    /// (so at most MAX_VISIBLE_REF_UPDATES come back). Removing the formula
+    /// clamp prices the raw limit and rejects the document; removing the
+    /// collector clamp returns unbounded rows. Mirrors
+    /// tasks_limit_ceiling_clamped_to_max.
+    #[sqlx::test]
+    async fn ref_updates_limit_ceiling_clamped_to_max(pool: PgPool) {
+        use crate::api::events::MAX_VISIBLE_REF_UPDATES;
+        let db = db(pool).await;
+        db.create_repo(&repo("r1", OWNER, "widget", true))
+            .await
+            .unwrap();
+        let total = MAX_VISIBLE_REF_UPDATES as usize + 1;
+        for i in 0..total {
+            let mut row = ref_row(&format!("ceil{i}"), "z6MkOwner/widget");
+            row.timestamp = format!("2026-07-01T10:{:02}:{:02}+00:00", i / 60, i % 60);
+            db.insert_ref_update(&row).await.unwrap();
+        }
+        let schema = schema(db);
+        let query = format!(
+            "{{ refUpdates(limit: {}) {{ repo }} }}",
+            MAX_VISIBLE_REF_UPDATES + 4800
+        );
+        let resp = anon(&schema, &query).await;
+        assert!(
+            resp.errors.is_empty(),
+            "limit above the maximum must clamp, not fail: {:?}",
+            resp.errors
+        );
+        assert_eq!(
+            count(&resp),
+            MAX_VISIBLE_REF_UPDATES as usize,
+            "limit above the maximum must clamp to MAX_VISIBLE_REF_UPDATES"
+        );
     }
 
     fn count_tasks(resp: &async_graphql::Response) -> usize {

crates/gitlawb-node/src/graphql/subscription.rs

@@ -20,6 +20,7 @@ impl SubscriptionRoot {
     /// `if announce` block, `api/repos.rs`). This is a single-point invariant —
     /// any new sender to `ref_update_tx` MUST be `announce`-gated, or private-repo
     /// ref metadata leaks here to unauthenticated subscribers (#112/#114 class).
+    #[graphql(complexity = "50 + child_complexity")]
     async fn ref_updates(
         &self,
         ctx: &Context<'_>,
@@ -46,6 +47,7 @@ impl SubscriptionRoot {
         })
     }
 
+    #[graphql(complexity = "50 + child_complexity")]
     async fn task_events(
         &self,
         ctx: &Context<'_>,

crates/gitlawb-node/src/graphql/types.rs

@@ -11,6 +11,14 @@ pub struct RepoType {
     pub created_at: String,
 }
 
+#[derive(SimpleObject)]
+pub struct RepoPageType {
+    pub nodes: Vec<RepoType>,
+    pub has_next_page: bool,
+    /// Position after the last returned visible repository; null for an empty page.
+    pub end_cursor: Option<String>,
+}
+
 #[derive(SimpleObject, Clone)]
 pub struct AgentTaskType {
     pub id: String,

crates/gitlawb-node/src/ipfs_pin.rs

@@ -2638,6 +2638,7 @@ mod tests {
     /// type probe carries its oid on stdin rather than in argv, so an oid appears in
     /// the log only once an object has already got past its probe, and a healthy
     /// object costs two invocations to a faulting one's one.
+    #[cfg(unix)]
     fn objects_attempted(log: &std::path::Path) -> usize {
         std::fs::read_to_string(log)
             .unwrap_or_default()

crates/gitlawb-node/src/pinata.rs

@@ -912,6 +912,7 @@ mod tests {
     /// probe carries its oid on stdin rather than in argv, so an oid appears in the log only
     /// once an object has already got past its probe, and a healthy object costs two
     /// invocations to a faulting one's one.
+    #[cfg(unix)]
     fn objects_attempted(log: &std::path::Path) -> usize {
         std::fs::read_to_string(log)
             .unwrap_or_default()

crates/gitlawb-node/src/sync.rs

@@ -1330,6 +1330,18 @@ mod tests {
         .await;
     }
 
+    /// Use a rooted path without a Windows drive prefix in the remote fixture.
+    /// The destination still resolves on the temp directory's current drive,
+    /// while the remote's relative path contains no illegal colon component.
+    fn absolute_slug_path(path: &Path) -> String {
+        assert!(path.is_absolute());
+        path.components()
+            .filter(|part| !matches!(part, std::path::Component::Prefix(_)))
+            .collect::<std::path::PathBuf>()
+            .to_string_lossy()
+            .replace('\\', "/")
+    }
+
     #[sqlx::test]
     async fn process_batch_rejects_slug_escaping_repos_dir(pool: PgPool) {
         // The verified escape from #272: `PathBuf::join` discards everything
@@ -1345,12 +1357,13 @@ mod tests {
         // fails on but leaves the parent it created, so assert on both.
         let outside = TempDir::new().unwrap();
         let escape_dir = outside.path().join("nest");
-        let slug = format!("a/{}/escape", escape_dir.display());
+        let escape_path = absolute_slug_path(&escape_dir);
+        let slug = format!("a/{escape_path}/escape");
         let escape_target = escape_dir.join("escape.git");
 
         // Serve the composed URL for real, so a run without the guard genuinely
         // clones outside the root rather than merely failing at git.
-        let rel = format!("a{}/escape", escape_dir.display());
+        let rel = format!("a{escape_path}/escape");
         let (_remote, peer_url) = rooted_remote(&[&rel]);
 
         let did = "did:key:z6MkAttacker";
@@ -1470,6 +1483,7 @@ mod tests {
     // ── canonical containment before the git call (issue #272) ───────────────
 
     /// Every ref in `repo`, as one string, for a before/after comparison.
+    #[cfg(unix)]
     fn refs_of(repo: &Path) -> String {
         let out = Command::new("git")
             .args(["-C", repo.to_str().unwrap(), "for-each-ref"])
@@ -1802,6 +1816,7 @@ mod tests {
     /// starvation tests is fixed rather than dependent on how fast the loop
     /// runs. Two rows enqueued in the same microsecond would otherwise order
     /// arbitrarily.
+    #[cfg(unix)]
     async fn enqueue_at(db: &Db, pool: &PgPool, repo: &str, did: &str, enqueued_at: &str) {
         enqueue(db, repo, did).await;
         sqlx::query("UPDATE sync_queue SET enqueued_at = $1 WHERE repo = $2")
@@ -1995,7 +2010,8 @@ mod tests {
 
         let outside = TempDir::new().unwrap();
         let escape_dir = outside.path().join("nest");
-        let slug = format!("a/{}/gitlawb-probe", escape_dir.display());
+        let escape_path = absolute_slug_path(&escape_dir);
+        let slug = format!("a/{escape_path}/gitlawb-probe");
         let escape_target = escape_dir.join("gitlawb-probe.git");
 
         // Two things this fixture must get right or the test is green for the
@@ -2006,7 +2022,7 @@ mod tests {
         // run without the guard genuinely clones outside the root instead of
         // just failing at git. Db::upsert_peer cannot seed this row: it gates on
         // is_public_http_url, which rejects file://.
-        let rel = format!("a{}/gitlawb-probe", escape_dir.display());
+        let rel = format!("a{escape_path}/gitlawb-probe");
         let (_remote, peer_url) = rooted_remote(&[&rel]);
         let did = "did:key:z6MkAttacker";
         seed_local_peer(&pool, did, &peer_url).await;

crates/gitlawb-node/src/test_git_shim.rs

@@ -0,0 +1,75 @@
+//! Native Git fixtures shared by timing tests on Unix and Windows.
+
+use std::ops::Deref;
+use std::path::{Path, PathBuf};
+use std::sync::OnceLock;
+
+/// The fixture delays or hangs in its own process, so the production watchdog
+/// can terminate it without depending on a shell or descendant process support.
+pub(super) enum Behavior<'a> {
+    Delay(u64),
+    Hang,
+    HangOid(&'a str),
+    HangRepo(&'a str),
+    LogTypes(&'a Path),
+}
+
+pub(super) struct GitShim {
+    _directory: tempfile::TempDir,
+    executable: PathBuf,
+}
+
+impl Deref for GitShim {
+    type Target = Path;
+
+    fn deref(&self) -> &Path {
+        &self.executable
+    }
+}
+
+pub(super) fn create(name: &str, behavior: Behavior<'_>) -> GitShim {
+    static COMPILED: OnceLock<tempfile::TempDir> = OnceLock::new();
+    let compiled = COMPILED.get_or_init(|| {
+        let directory = tempfile::tempdir().expect("native Git fixture directory");
+        let source = directory.path().join("git_shim.rs");
+        std::fs::write(&source, include_str!("../tests/fixtures/git_shim.rs"))
+            .expect("write native Git fixture source");
+        let executable = directory
+            .path()
+            .join(format!("git-shim{}", std::env::consts::EXE_SUFFIX));
+        let output =
+            std::process::Command::new(std::env::var_os("RUSTC").unwrap_or_else(|| "rustc".into()))
+                .arg("--edition=2021")
+                .arg("-Dwarnings")
+                .arg(&source)
+                .arg("-o")
+                .arg(executable)
+                .output()
+                .expect("compile native Git fixture with the installed Rust toolchain");
+        assert!(
+            output.status.success(),
+            "{}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        directory
+    });
+    let directory = tempfile::Builder::new()
+        .prefix(name)
+        .tempdir()
+        .expect("Git fixture instance");
+    let file_name = format!("git-shim{}", std::env::consts::EXE_SUFFIX);
+    let executable = directory.path().join(&file_name);
+    std::fs::copy(compiled.path().join(file_name), &executable).expect("copy native Git fixture");
+    let config = match behavior {
+        Behavior::Delay(milliseconds) => format!("delay\n{milliseconds}"),
+        Behavior::Hang => "hang\n".to_owned(),
+        Behavior::HangOid(oid) => format!("hang-oid\n{oid}"),
+        Behavior::HangRepo(repo) => format!("hang-repo\n{repo}"),
+        Behavior::LogTypes(path) => format!("log-types\n{}", path.display()),
+    };
+    std::fs::write(directory.path().join("config"), config).expect("configure native Git fixture");
+    GitShim {
+        _directory: directory,
+        executable,
+    }
+}

crates/gitlawb-node/src/test_support.rs

@@ -28,6 +28,9 @@ use gitlawb_core::identity::Keypair;
 use crate::auth::AuthenticatedDid;
 use crate::state::AppState;
 
+#[path = "test_git_shim.rs"]
+mod git_shim;
+
 /// Build an [`AppState`] over a real, migrated Postgres pool (from `#[sqlx::test]`).
 /// Runs the schema migrations first, because the per-test database starts empty.
 ///
@@ -559,11 +562,12 @@ mod tests {
     /// #174 (SC1, load-bearing): a saturated READ pool must NOT shed an
     /// authenticated push — the write pool is a separate budget. Read pool at zero,
     /// write pool with capacity: the push proceeds PAST admission (it then errors on
-    /// the placeholder DB, but crucially it is not a 503). Route git-receive-pack
+    /// the closed DB, with db_unavailable rather than overloaded). Route git-receive-pack
     /// back to the read pool and this goes red — that is the isolation proof.
     #[tokio::test]
     async fn git_receive_pack_not_shed_by_exhausted_read_pool() {
         let mut state = test_state_lazy();
+        state.db.pool().close().await;
         // Read pool exhausted as if a flood of anonymous clones held every slot.
         state.git_read_semaphore = Arc::new(tokio::sync::Semaphore::new(0));
         // Write pool keeps its default capacity from test_state_lazy.
@@ -585,10 +589,13 @@ mod tests {
             .await
             .unwrap();
 
-        assert_ne!(
-            resp.status(),
-            StatusCode::SERVICE_UNAVAILABLE,
-            "an exhausted READ pool must not shed a push — the write pool is a separate budget (#174)"
+        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let bytes = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
+        let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+        assert_eq!(
+            body["error"],
+            crate::error::DB_UNAVAILABLE_CODE,
+            "the push must clear admission and reach the deliberately closed database"
         );
     }
 
@@ -1526,6 +1533,451 @@ mod tests {
         );
     }
 
+    /// Pre-assigned tasks must not be stealable: a stranger claiming a pending
+    /// task reserved for another agent must fail, and must not receive the
+    /// UCAN. The reserved assignee still claims successfully.
+    #[sqlx::test]
+    async fn claim_task_honors_preassigned_assignee(pool: PgPool) {
+        let delegator = "did:key:zCLAIMDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let reserved = "did:key:zCLAIMRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let thief = "did:key:zCLAIMTHIEFCCCCCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool).await;
+        let mut task = seed_task("task-reserved", delegator);
+        task.assignee_did = Some(reserved.to_string());
+        task.ucan_token = Some("ucan-secret-for-reserved-only".into());
+        state.db.create_task(&task).await.expect("seed");
+
+        let router = || {
+            Router::new()
+                .route(
+                    "/api/v1/tasks/{id}/claim",
+                    axum::routing::post(crate::api::tasks::claim_task),
+                )
+                .with_state(state.clone())
+        };
+        let uri = "/api/v1/tasks/task-reserved/claim";
+
+        // Thief signs as themselves and asks to claim — must fail, UCAN stays put.
+        let resp = router()
+            .oneshot(signed_request_as(
+                thief,
+                Method::POST,
+                uri,
+                Body::from(format!(r#"{{"assignee_did":"{thief}"}}"#)),
+            ))
+            .await
+            .unwrap();
+        assert_eq!(
+            resp.status(),
+            StatusCode::FORBIDDEN,
+            "stranger must not steal a pre-assigned task, got {}",
+            resp.status()
+        );
+        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let text = String::from_utf8_lossy(&body);
+        assert!(
+            !text.contains("ucan-secret-for-reserved-only"),
+            "UCAN must not leak on a failed steal: {text}"
+        );
+        let still = state.db.get_task("task-reserved").await.unwrap().unwrap();
+        assert_eq!(still.status, "pending");
+        assert_eq!(still.assignee_did.as_deref(), Some(reserved));
+
+        // Reserved assignee claims successfully and receives the UCAN.
+        let resp = router()
+            .oneshot(signed_request_as(
+                reserved,
+                Method::POST,
+                uri,
+                Body::from(format!(r#"{{"assignee_did":"{reserved}"}}"#)),
+            ))
+            .await
+            .unwrap();
+        assert!(
+            resp.status().is_success(),
+            "reserved assignee must claim, got {}",
+            resp.status()
+        );
+        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let text = String::from_utf8_lossy(&body);
+        assert!(
+            text.contains("ucan-secret-for-reserved-only"),
+            "reserved assignee must receive UCAN: {text}"
+        );
+        assert!(text.contains("\"status\":\"claimed\"") || text.contains("claimed"));
+    }
+
+    /// Open (unassigned) pending tasks remain first-claimer-wins after the
+    /// pre-assignment gate.
+    #[sqlx::test]
+    async fn claim_task_open_still_first_claimer_wins(pool: PgPool) {
+        let delegator = "did:key:zOPENDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let claimer = "did:key:zOPENCLAIMERBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let state = test_state(pool).await;
+        state
+            .db
+            .create_task(&seed_task("task-open", delegator))
+            .await
+            .expect("seed");
+
+        let router = Router::new()
+            .route(
+                "/api/v1/tasks/{id}/claim",
+                axum::routing::post(crate::api::tasks::claim_task),
+            )
+            .with_state(state.clone());
+        let resp = router
+            .oneshot(signed_request_as(
+                claimer,
+                Method::POST,
+                "/api/v1/tasks/task-open/claim",
+                Body::from(format!(r#"{{"assignee_did":"{claimer}"}}"#)),
+            ))
+            .await
+            .unwrap();
+        assert!(
+            resp.status().is_success(),
+            "open task must still be claimable, got {}",
+            resp.status()
+        );
+        let got = state.db.get_task("task-open").await.unwrap().unwrap();
+        assert_eq!(got.status, "claimed");
+        assert_eq!(got.assignee_did.as_deref(), Some(claimer));
+    }
+
+    /// ASCII-blank assignees (`ASSIGNEE_BLANK`: space, tab, LF, CR) must be open
+    /// at the claim SQL blank branch. Seeds via raw INSERT so create_task
+    /// normalization is not in the path.
+    #[sqlx::test]
+    async fn claim_task_ascii_blank_reservations_are_open(pool: PgPool) {
+        let delegator = "did:key:zBLANKDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let claimer = "did:key:zBLANKCLAIMERBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let state = test_state(pool.clone()).await;
+        let now = chrono::Utc::now().to_rfc3339();
+        for (id, blank) in [
+            ("task-blank-empty", ""),
+            ("task-blank-tab", "\t"),
+            ("task-blank-space", " "),
+            ("task-blank-lf", "\n"),
+            ("task-blank-cr", "\r"),
+        ] {
+            sqlx::query(
+                "INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
+                 VALUES ($1,NULL,'build','pending',$2,$3,'repo:write',NULL,NULL,NULL,$4,$4,NULL)",
+            )
+            .bind(id)
+            .bind(delegator)
+            .bind(blank)
+            .bind(&now)
+            .execute(&pool)
+            .await
+            .unwrap_or_else(|e| panic!("seed {id}: {e}"));
+
+            let claimed = state
+                .db
+                .claim_task(id, claimer)
+                .await
+                .unwrap_or_else(|e| panic!("{id} must be open: {e}"));
+            assert_eq!(claimed.status, "claimed", "{id}");
+            assert_eq!(claimed.assignee_did.as_deref(), Some(claimer), "{id}");
+        }
+    }
+
+    /// create_task must null whitespace-only assignees and return that shape
+    /// (fail-on-remove for the blank filter).
+    #[sqlx::test]
+    async fn create_task_normalizes_whitespace_assignee(pool: PgPool) {
+        let delegator = "did:key:zCREATENORMDELEGAAAAAAAAAAAAAAAAAAAAAAAA";
+        let state = test_state(pool).await;
+        let mut task = seed_task("task-create-norm", delegator);
+        task.assignee_did = Some(" \t\n ".into());
+        let stored = state.db.create_task(&task).await.expect("create");
+        assert!(
+            stored.assignee_did.is_none(),
+            "create_task must return normalized NULL assignee, got {:?}",
+            stored.assignee_did
+        );
+        let got = state
+            .db
+            .get_task("task-create-norm")
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(got.assignee_did.is_none(), "row must store NULL assignee");
+    }
+
+    /// Claiming a reserved task must keep the stored DID form so exact-match
+    /// list filters still find it. Also covers bare→full claim + list by bare
+    /// (claimer full DID must not be required for list equality).
+    #[sqlx::test]
+    async fn claim_task_keeps_stored_assignee_did_form(pool: PgPool) {
+        let delegator = "did:key:zFORMDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let bare = "zFORMRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let full = format!("did:key:{bare}");
+        let state = test_state(pool).await;
+        let mut task = seed_task("task-form", delegator);
+        task.assignee_did = Some(bare.to_string());
+        state.db.create_task(&task).await.expect("seed");
+
+        let claimed = state
+            .db
+            .claim_task("task-form", &full)
+            .await
+            .expect("claim with full DID");
+        assert_eq!(
+            claimed.assignee_did.as_deref(),
+            Some(bare),
+            "reserved claim must keep the stored bare-key form"
+        );
+        let listed_bare = state
+            .db
+            .list_tasks(Some("claimed"), Some(bare), 10)
+            .await
+            .unwrap();
+        assert!(
+            listed_bare.iter().any(|t| t.id == "task-form"),
+            "delegator filtering by bare key must still find the task"
+        );
+        let listed_full = state
+            .db
+            .list_tasks(Some("claimed"), Some(&full), 10)
+            .await
+            .unwrap();
+        assert!(
+            !listed_full.iter().any(|t| t.id == "task-form"),
+            "exact list by claimer's full DID must miss the bare stored form"
+        );
+    }
+
+    /// Padded non-blank reservation must survive claim unchanged (exact form).
+    #[sqlx::test]
+    async fn claim_task_keeps_padded_reservation_exact(pool: PgPool) {
+        let delegator = "did:key:zPADDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let key = "zPADRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let padded = format!(" {key}\t");
+        let full = format!("did:key:{key}");
+        let state = test_state(pool.clone()).await;
+        let now = chrono::Utc::now().to_rfc3339();
+        sqlx::query(
+            "INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
+             VALUES ($1,NULL,'build','pending',$2,$3,'repo:write',NULL,NULL,NULL,$4,$4,NULL)",
+        )
+        .bind("task-padded")
+        .bind(delegator)
+        .bind(&padded)
+        .bind(&now)
+        .execute(&pool)
+        .await
+        .expect("seed padded");
+
+        let claimed = state
+            .db
+            .claim_task("task-padded", &full)
+            .await
+            .expect("claim padded reservation");
+        assert_eq!(
+            claimed.assignee_did.as_deref(),
+            Some(padded.as_str()),
+            "claim must not BTRIM a non-blank reservation"
+        );
+        let listed = state
+            .db
+            .list_tasks(Some("claimed"), Some(&padded), 10)
+            .await
+            .unwrap();
+        assert!(
+            listed.iter().any(|t| t.id == "task-padded"),
+            "exact list by original padded form must still find the task"
+        );
+
+        let finished = state
+            .db
+            .finish_task("task-padded", "completed", None, &full)
+            .await
+            .expect("padded assignee must finish after trim match");
+        assert_eq!(finished.status, "completed");
+    }
+
+    /// Direct coverage for the DB-layer finish_task assignee Rust gate (handlers
+    /// 403 before calling in, so HTTP tests alone leave this unproven).
+    #[sqlx::test]
+    async fn finish_task_rejects_non_assignee_at_db(pool: PgPool) {
+        let delegator = "did:key:zFINISHDELEGATORAAAAAAAAAAAAAAAAAAAAAAAA";
+        let assignee = "did:key:zFINISHASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let stranger = "did:key:zFINISHSTRANGERCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool).await;
+        state
+            .db
+            .create_task(&seed_task("task-finish-gate", delegator))
+            .await
+            .expect("seed");
+        state
+            .db
+            .claim_task("task-finish-gate", assignee)
+            .await
+            .expect("claim");
+
+        let err = state
+            .db
+            .finish_task("task-finish-gate", "completed", None, stranger)
+            .await
+            .expect_err("stranger must not finish");
+        assert!(
+            err.to_string().contains("assignee") || err.to_string().contains("finishable"),
+            "expected assignee denial, got {err}"
+        );
+        let still = state
+            .db
+            .get_task("task-finish-gate")
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            still.status, "claimed",
+            "row must stay claimed after denial"
+        );
+
+        let done = state
+            .db
+            .finish_task("task-finish-gate", "completed", None, assignee)
+            .await
+            .expect("assignee finishes");
+        assert_eq!(done.status, "completed");
+    }
+
+    /// Fail-on-remove for `finish_task`'s `AND assignee_did=$5`: a concurrent writer
+    /// that changes the stored assignee between the production read and UPDATE must
+    /// leave the task claimed (production `Db::finish_task` path).
+    #[sqlx::test]
+    async fn finish_task_sql_assignee_predicate_is_load_bearing(pool: PgPool) {
+        let delegator = "did:key:zFINISHSQLDELEGAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let assignee = "did:key:zFINISHSQLASSIGNBBBBBBBBBBBBBBBBBBBBBBBB";
+        let other = "did:key:zFINISHSQLOTHERCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool.clone()).await;
+        let id = "task-finish-sql";
+        state
+            .db
+            .create_task(&seed_task(id, delegator))
+            .await
+            .expect("seed");
+        state.db.claim_task(id, assignee).await.expect("claim");
+
+        let pool_bg = pool.clone();
+        let id_bg = id.to_string();
+        let other_bg = other.to_string();
+        let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
+        let hog = tokio::spawn(async move {
+            let mut tx = pool_bg.begin().await.expect("begin");
+            sqlx::query("SELECT assignee_did FROM agent_tasks WHERE id = $1 FOR UPDATE")
+                .bind(&id_bg)
+                .fetch_one(&mut *tx)
+                .await
+                .expect("lock row");
+            locked_tx.send(()).ok();
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            sqlx::query(
+                "UPDATE agent_tasks SET assignee_did = $2 WHERE id = $1 AND status = 'claimed'",
+            )
+            .bind(&id_bg)
+            .bind(&other_bg)
+            .execute(&mut *tx)
+            .await
+            .expect("race assignee");
+            tx.commit().await.expect("commit");
+        });
+        locked_rx.await.expect("row lock held before finish");
+
+        let err = state
+            .db
+            .finish_task(id, "completed", None, assignee)
+            .await
+            .expect_err("assignee slot race must fail production finish_task");
+        hog.await.expect("hog task");
+
+        assert!(
+            err.to_string().contains("not in claimed state"),
+            "expected finish denial, got {err}"
+        );
+        let still = state.db.get_task(id).await.unwrap().unwrap();
+        assert_eq!(still.status, "claimed");
+        assert_eq!(still.assignee_did.as_deref(), Some(other));
+    }
+
+    /// Fail-on-remove for the claim UPDATE assignee-slot re-check: a concurrent
+    /// writer that changes `assignee_did` after the pre-read must block claim.
+    #[sqlx::test]
+    async fn claim_task_update_assignee_slot_recheck_is_load_bearing(pool: PgPool) {
+        let delegator = "did:key:zCLAIMSLOTDELEGAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let reserved = "did:key:zCLAIMSLOTASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let other = "did:key:zCLAIMSLOTOTHERCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool.clone()).await;
+        let id = "task-claim-slot";
+        let now = chrono::Utc::now().to_rfc3339();
+        sqlx::query(
+            "INSERT INTO agent_tasks
+                 (id, repo_id, kind, status, delegator_did, assignee_did, capability,
+                  ucan_token, payload, result, created_at, updated_at, deadline)
+             VALUES ($1, NULL, 'build', 'pending', $2, $3, 'repo:write',
+                     NULL, NULL, NULL, $4, $4, NULL)",
+        )
+        .bind(id)
+        .bind(delegator)
+        .bind(reserved)
+        .bind(&now)
+        .execute(&pool)
+        .await
+        .expect("seed reserved");
+
+        let pool_bg = pool.clone();
+        let id_bg = id.to_string();
+        let other_bg = other.to_string();
+        let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
+        let hog = tokio::spawn(async move {
+            let mut tx = pool_bg.begin().await.expect("begin");
+            sqlx::query("SELECT assignee_did FROM agent_tasks WHERE id = $1 FOR UPDATE")
+                .bind(&id_bg)
+                .fetch_one(&mut *tx)
+                .await
+                .expect("lock row");
+            locked_tx.send(()).ok();
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            sqlx::query(
+                "UPDATE agent_tasks SET assignee_did = $2 WHERE id = $1 AND status = 'pending'",
+            )
+            .bind(&id_bg)
+            .bind(&other_bg)
+            .execute(&mut *tx)
+            .await
+            .expect("race assignee");
+            tx.commit().await.expect("commit");
+        });
+        locked_rx.await.expect("row lock held before claim");
+
+        let err = state
+            .db
+            .claim_task(id, reserved)
+            .await
+            .expect_err("assignee slot race must fail production claim_task");
+        hog.await.expect("hog task");
+
+        assert!(
+            err.to_string().contains("not found or already claimed"),
+            "expected SQL re-check denial, got {err}"
+        );
+        assert!(
+            err.downcast_ref::<crate::db::TaskReservedForOtherAssignee>()
+                .is_none(),
+            "must prove the UPDATE predicate, not the Rust pre-check: {err}"
+        );
+        let still = state.db.get_task(id).await.unwrap().unwrap();
+        assert_eq!(still.status, "pending");
+        assert_eq!(still.assignee_did.as_deref(), Some(other));
+    }
+
     /// Adversarial-review GATE-2 (create_pr): opening a PR requires read access.
     /// A non-reader is denied on a private repo before any PR is created; the
     /// owner is allowed.
@@ -4693,6 +5145,7 @@ mod tests {
                 node_did: owner.to_string(),
                 signature: "sig".to_string(),
                 issued_at: Utc::now().to_rfc3339(),
+                version: 1,
             })
             .await
             .expect("seed private cert");
@@ -10749,14 +11202,10 @@ mod tests {
         seed_legacy_pin(&pool, &bare, &fx.public_oid, Some(&repo.id)).await;
 
         // A git that takes 300ms per invocation (the read makes two: type, then content).
-        let slow_git = std::env::temp_dir().join(format!("gl-slow-git-{short}"));
-        std::fs::write(&slow_git, "#!/bin/sh\nsleep 0.3\nexec git \"$@\"\n").expect("write shim");
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::PermissionsExt;
-            std::fs::set_permissions(&slow_git, std::fs::Permissions::from_mode(0o755))
-                .expect("chmod shim");
-        }
+        let slow_git = git_shim::create(
+            &format!("gl-slow-git-{short}"),
+            git_shim::Behavior::Delay(300),
+        );
 
         let ticks = std::sync::Arc::new(AtomicUsize::new(0));
         let ticker = {
@@ -11703,19 +12152,6 @@ mod tests {
         );
     }
 
-    /// Write an executable `git` stand-in and return its path.
-    fn write_git_shim(name: &str, script: &str) -> std::path::PathBuf {
-        let path = std::env::temp_dir().join(name);
-        std::fs::write(&path, script).expect("write the git shim");
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::PermissionsExt;
-            std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
-                .expect("chmod the git shim");
-        }
-        path
-    }
-
     /// F6 scenario 1 (#173 round 13): one hung candidate must not starve the rows behind
     /// it in the same pass. `DiscoveryCtx` is loaded once per pass, so before the per-row
     /// slice every source-less row in a pass shared ONE deadline: the first row's wedged
@@ -11755,23 +12191,9 @@ mod tests {
 
         // The type stage feeds the oid on STDIN (`cat-file --batch-check`) and the
         // content stage puts it in argv, so the stand-in has to look in both places.
-        let git_bin = write_git_shim(
+        let git_bin = git_shim::create(
             &format!("gl-hung-git-{short}"),
-            &format!(
-                "#!/bin/sh\n\
-                 if [ \"$2\" = \"--batch-check\" ]; then\n\
-                 \x20 oid=$(cat)\n\
-                 \x20 case \"$oid\" in\n\
-                 \x20   {hung_oid}) sleep 30; exit 1 ;;\n\
-                 \x20 esac\n\
-                 \x20 printf '%s\\n' \"$oid\" | git \"$@\"\n\
-                 \x20 exit $?\n\
-                 fi\n\
-                 case \"$*\" in\n\
-                 \x20 *{hung_oid}*) sleep 30; exit 1 ;;\n\
-                 esac\n\
-                 exec git \"$@\"\n"
-            ),
+            git_shim::Behavior::HangOid(&hung_oid),
         );
 
         let stats = tokio::time::timeout(
@@ -11856,10 +12278,7 @@ mod tests {
 
         // Wedges on every invocation, so no row can ever be repaired and the only
         // question left is what each one COSTS.
-        let git_bin = write_git_shim(
-            &format!("gl-spent-git-{short}"),
-            "#!/bin/sh\nsleep 30\nexit 1\n",
-        );
+        let git_bin = git_shim::create(&format!("gl-spent-git-{short}"), git_shim::Behavior::Hang);
 
         let stats = tokio::time::timeout(
             std::time::Duration::from_secs(60),
@@ -12463,10 +12882,7 @@ mod tests {
             seed_legacy_pin(&pool, &src, oid, None).await;
         }
 
-        let git_bin = write_git_shim(
-            &format!("gl-starve-git-{short}"),
-            "#!/bin/sh\nsleep 30\nexit 1\n",
-        );
+        let git_bin = git_shim::create(&format!("gl-starve-git-{short}"), git_shim::Behavior::Hang);
 
         tokio::time::timeout(
             std::time::Duration::from_secs(120),
@@ -12534,9 +12950,9 @@ mod tests {
         let (raw_cid, provider_cid) = seed_legacy_pin(&pool, &src, &fx.public_oid, None).await;
 
         // Wedges only inside the position-nine repo, which the sweep enters by cwd.
-        let git_bin = write_git_shim(
+        let git_bin = git_shim::create(
             &format!("gl-mid-git-{short}"),
-            "#!/bin/sh\ncase \"$(pwd)\" in\n  */midcand9.git) sleep 30; exit 1 ;;\nesac\nexec git \"$@\"\n",
+            git_shim::Behavior::HangRepo("midcand9.git"),
         );
 
         let first = tokio::time::timeout(
@@ -12698,19 +13114,9 @@ mod tests {
 
         let log = std::env::temp_dir().join(format!("gl-ali-log-{short}"));
         let _ = std::fs::remove_file(&log);
-        let git_bin = write_git_shim(
+        let git_bin = git_shim::create(
             &format!("gl-ali-git-{short}"),
-            &format!(
-                "#!/bin/sh\n\
-                 if [ \"$2\" = \"--batch-check\" ]; then\n\
-                 \x20 oid=$(cat)\n\
-                 \x20 printf '%s %s\\n' \"$oid\" \"$(basename $(pwd))\" >> {log}\n\
-                 \x20 printf '%s\\n' \"$oid\" | git \"$@\"\n\
-                 \x20 exit $?\n\
-                 fi\n\
-                 exec git \"$@\"\n",
-                log = log.display()
-            ),
+            git_shim::Behavior::LogTypes(&log),
         );
 
         let mut traversal = crate::ipfs_pin::DiscoveryTraversalState::default();
@@ -17252,6 +17658,7 @@ mod tests {
             node_did: "did:key:zNode".into(),
             signature: "sig".into(),
             issued_at: "2026-01-01T00:00:00Z".into(),
+            version: 1,
         };
         state.db.insert_ref_certificate(&cert).await.unwrap();
 
@@ -17287,6 +17694,7 @@ mod tests {
             node_did: "did:key:zNode".into(),
             signature: "sig".into(),
             issued_at: "2026-01-01T00:00:00Z".into(),
+            version: 1,
         };
         state.db.insert_ref_certificate(&cert).await.unwrap();
 
@@ -18026,6 +18434,7 @@ mod tests {
             node_did: "did:key:zNode".into(),
             signature: "sig".into(),
             issued_at: "2026-01-01T00:00:00Z".into(),
+            version: 1,
         };
         state.db.insert_ref_certificate(&cert).await.unwrap();
 
@@ -19700,6 +20109,7 @@ mod tests {
             node_did: "did:key:zNODE".into(),
             signature: "sig".into(),
             issued_at: issued_at.to_string(),
+            version: 1,
         }
     }
 

crates/gitlawb-node/tests/fixtures/git_shim.rs

@@ -0,0 +1,59 @@
+use std::io::{Read, Write};
+use std::process::{Command, Stdio};
+use std::time::Duration;
+
+fn main() -> std::io::Result<()> {
+    let executable = std::env::current_exe()?;
+    let config = std::fs::read_to_string(executable.parent().unwrap().join("config"))?;
+    let (mode, parameter) = config.split_once('\n').unwrap_or((&config, ""));
+    let args: Vec<_> = std::env::args_os().skip(1).collect();
+    if mode == "delay" {
+        std::thread::sleep(Duration::from_millis(parameter.parse().unwrap()));
+    }
+    let batch = args.get(1).is_some_and(|arg| arg == "--batch-check");
+    let input = if batch {
+        let mut bytes = Vec::new();
+        std::io::stdin().read_to_end(&mut bytes)?;
+        Some(bytes)
+    } else {
+        None
+    };
+    let cwd = std::env::current_dir()?;
+    let hang = mode == "hang"
+        || (mode == "hang-repo" && cwd.file_name().is_some_and(|name| name == parameter))
+        || (mode == "hang-oid"
+            && (args.iter().any(|arg| arg == parameter)
+                || input
+                    .as_ref()
+                    .is_some_and(|bytes| String::from_utf8_lossy(bytes).trim() == parameter)));
+    if hang {
+        std::thread::sleep(Duration::from_secs(30));
+        std::process::exit(1);
+    }
+    if mode == "log-types" && batch {
+        let mut log = std::fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(parameter)?;
+        writeln!(
+            log,
+            "{} {}",
+            String::from_utf8_lossy(input.as_ref().unwrap()).trim(),
+            cwd.file_name().unwrap().to_string_lossy()
+        )?;
+    }
+    let mut child = Command::new("git")
+        .args(&args)
+        .stdin(if input.is_some() {
+            Stdio::piped()
+        } else {
+            Stdio::inherit()
+        })
+        .stdout(Stdio::inherit())
+        .stderr(Stdio::inherit())
+        .spawn()?;
+    if let Some(bytes) = input {
+        child.stdin.take().unwrap().write_all(&bytes)?;
+    }
+    std::process::exit(child.wait()?.code().unwrap_or(1));
+}

crates/gl/src/cert.rs

@@ -4,6 +4,7 @@
 
 use anyhow::{Context, Result};
 use clap::{Args, Subcommand};
+use serde_json::Value;
 use std::path::PathBuf;
 
 use crate::http::NodeClient;
@@ -34,7 +35,12 @@ pub enum CertCmd {
         #[arg(long)]
         dir: Option<PathBuf>,
     },
-    /// Show a specific ref certificate and verify its signature
+    /// Show a specific ref certificate and verify its signature.
+    ///
+    /// `--verify` requires `--expect-node <did>`: the queried node's
+    /// self-reported DID is not a trust anchor, because a node serving a
+    /// forged certificate also controls its own self-report. Callers
+    /// must name the issuer they trust.
     Show {
         /// Repository in <owner>/<repo> or <repo> format
         repo: String,
@@ -49,13 +55,17 @@ pub enum CertCmd {
         #[arg(long)]
         dir: Option<PathBuf>,
         /// Exit non-zero unless the Ed25519 signature verifies AND the
-        /// issuing node matches the queried node (or --expect-node)
+        /// issuing node matches --expect-node (which --verify requires:
+        /// without a caller-supplied anchor there is no trusted issuer to
+        /// verify against)
         #[arg(long)]
         verify: bool,
         /// Expected issuing node DID for --verify. A valid signature alone
         /// only proves the cert is internally consistent — signed by whatever
         /// key it names — so --verify also anchors the issuer to a DID you
-        /// trust: this value when given, else the queried node's DID.
+        /// trust. The queried node's self-reported DID is deliberately NOT
+        /// accepted as a fallback anchor: the node that served the cert can
+        /// serve a matching self-report.
         #[arg(long, requires = "verify")]
         expect_node: Option<String>,
     },
@@ -152,6 +162,24 @@ async fn cmd_show(
     let node_did = cert["node_did"].as_str().unwrap_or("?");
     let signature = cert["signature"].as_str().unwrap_or("?");
     let issued_at = cert["issued_at"].as_str().unwrap_or("?");
+    // #26 Split PR 3: read the wire-format version. A MISSING key
+    // selects legacy v1 (an old server that predates the field).
+    // A PRESENT value must be a JSON integer that fits in u32 and
+    // names a version this client supports — otherwise the response
+    // declares a format we cannot represent and we refuse to verify
+    // rather than guess the payload shape. Reviewer 2 finding:
+    // collapsing `null`, a string, a float, or an overflow integer
+    // onto v1 would let a well-signed v1 signature pass `--verify`
+    // against a server that explicitly said "version 2 (or 99, or
+    // 4294967297)" — that is the exact mismatch the field exists
+    // to prevent. parse_cert_version is the load-bearing parser
+    // that distinguishes missing-key (legacy v1) from invalid-value
+    // (unsupported).
+    let parsed_version = parse_cert_version(cert.get("version"));
+    let version_display: String = match &parsed_version {
+        Ok(v) => v.to_string(),
+        Err(reason) => format!("unsupported ({reason})"),
+    };
 
     println!("Ref Certificate: {cert_id}");
     println!("  Ref:       {ref_name}");
@@ -160,6 +188,7 @@ async fn cmd_show(
     println!("  Pusher:    {pusher}");
     println!("  Node DID:  {node_did}");
     println!("  Issued at: {issued_at}");
+    println!("  Version:   {version_display}");
     println!("  Signature: {signature}");
     println!();
 
@@ -169,9 +198,31 @@ async fn cmd_show(
     // This proves the cert is internally authentic — signed by the key it
     // names; the node-DID comparison below covers *which* node that is.
     let repo_id = cert["repo_id"].as_str().unwrap_or("");
-    let verdict = verify_signature(
-        repo_id, ref_name, old_sha, new_sha, pusher, node_did, issued_at, signature,
-    );
+    let verdict = match parsed_version {
+        // v1 is the only version this client verifies. The v1
+        // signed payload is the 7-field canonical form with no
+        // `version` key — see gitlawb-node/src/cert.rs. A future
+        // v2+ cert has a different signed payload shape (the
+        // version field becomes part of the signed JSON per
+        // `v2_signing_payload`), so a v2+ cert from a server this
+        // client does not know about must NOT be silently verified
+        // as v1.
+        // Reviewer 2: refuse rather than guess; the client and
+        // server must agree on the version.
+        // parse_cert_version admits exactly one Ok value: 1 (missing key and
+        // integer 1 both land there; 2, floats, strings, overflow and every
+        // other spelling come back Err with their own reason). A separate
+        // `Ok(v != 1)` arm existed here and was DEAD CODE — a test aimed at it
+        // could never catch drift in this match (review round 2) — so the
+        // match now mirrors the parser's real contract, which the
+        // `parse_cert_version_truth_table` test pins.
+        Ok(_) => verify_signature(
+            repo_id, ref_name, old_sha, new_sha, pusher, node_did, issued_at, signature,
+        ),
+        Err(ref reason) => Err(format!(
+            "cert declared a version this client cannot represent ({reason}); refusing to verify"
+        )),
+    };
 
     println!("Signature verification:");
     match &verdict {
@@ -181,7 +232,11 @@ async fn cmd_show(
             );
         }
         Err(reason) => {
-            println!("  INVALID — {reason}");
+            // A version refusal means no cryptographic check ran, so it must
+            // not print as an INVALID signature: operators and log scrapers
+            // would read that as a failed Ed25519 verification. INVALID is
+            // reserved for failures out of verify_signature itself.
+            println!("  {} — {reason}", signature_verdict_label(&parsed_version));
         }
     }
 
@@ -198,29 +253,39 @@ async fn cmd_show(
         },
         Err(e) => Err(e.to_string()),
     };
-    let current_node_did = current.as_ref().ok().cloned();
     for line in did_check_report(&current, node_did) {
         println!("{line}");
     }
 
     if require_valid {
         if let Err(reason) = verdict {
-            anyhow::bail!("certificate signature did not verify: {reason}");
+            // Round 3: do not claim the signature failed when it was
+            // never checked. A version refusal means the payload
+            // shape is unknown, so no signature verification ran.
+            anyhow::bail!("certificate could not be verified: {reason}");
         }
         // A valid signature proves internal consistency only: the payload was
         // signed by whatever key the certificate itself names. A hostile
         // source can mint a keypair, put its DID in node_did, and self-sign.
-        // --verify therefore also anchors the issuer to a trusted DID:
-        // --expect-node when given, else the DID of the node being queried.
-        let expected = expect_node.as_deref().or(current_node_did.as_deref());
-        match expected {
+        // --verify therefore also anchors the issuer — and the anchor must be
+        // a DID the CALLER trusts, which is exactly what --expect-node is.
+        // The DID the queried node reports at `/` is NOT an anchor: the same
+        // node that served a forged certificate can serve a matching
+        // self-report, and with that fallback a wholly forged cert printed
+        // VALID and exited 0 (review round 2, demonstrated by execution). So
+        // --verify without --expect-node refuses to claim a trusted issuer at
+        // all, rather than laundering the server's self-assertion into one.
+        match expect_node.as_deref() {
             Some(expected) if expected == node_did => {}
             Some(expected) => anyhow::bail!(
                 "certificate is signed by {node_did}, but the expected issuer is {expected} — \
                  a valid signature alone proves internal consistency, not a trusted issuer"
             ),
             None => anyhow::bail!(
-                "cannot anchor the issuer: node info is unreachable and no --expect-node was given"
+                "--verify needs --expect-node <did> to anchor the issuer: the signature is \
+                 valid, but it only proves the cert is self-consistent with the DID it \
+                 names ({node_did}); the queried node's self-reported DID is not a trust \
+                 anchor"
             ),
         }
     }
@@ -268,6 +333,18 @@ fn did_check_report(current: &std::result::Result<String, String>, node_did: &st
     }
 }
 
+/// The stdout label for a failed verification verdict. A `parsed_version`
+/// error means `verify_signature` never ran, so the refusal must print as
+/// UNSUPPORTED VERSION rather than INVALID, which operators and log
+/// scrapers would read as a failed Ed25519 check.
+fn signature_verdict_label(parsed_version: &std::result::Result<u32, String>) -> &'static str {
+    if parsed_version.is_err() {
+        "UNSUPPORTED VERSION"
+    } else {
+        "INVALID"
+    }
+}
+
 /// Rebuild the node's canonical signing payload (field order must match
 /// gitlawb-node/src/cert.rs::issue_ref_certificate exactly) and verify the
 /// certificate's Ed25519 signature against the key embedded in `node_did`.
@@ -314,6 +391,65 @@ fn verify_signature(
         .map_err(|_| "Ed25519 signature does not match the signed payload".to_string())
 }
 
+/// Parse the wire-format `version` field of a certificate response.
+///
+/// Semantics (Reviewer 2):
+///   - MISSING key (`None`) → `Ok(1)`. An old server predates the field, so
+///     the cert is by definition the v1 (pre-versioning) shape; the v1
+///     verify path is the only path that can possibly match the bytes.
+///   - PRESENT key must be a JSON integer that fits losslessly in `u32`
+///     and equals a version this client supports. `null`, a string,
+///     a float, an overflow integer, or any version other than 1 all
+///     yield `Err`. Collapsing any of these onto v1 would let a
+///     well-signed v1 signature pass `--verify` against a response
+///     that explicitly declared a format this client cannot represent
+///     — which is exactly the mismatch the version field exists to
+///     prevent.
+///
+/// The supported-version set is hard-coded to `{1}` because PR 3 ships
+/// only v1 verification; bump this set (and the verdict branch) when
+/// v2 verification lands.
+fn parse_cert_version(value: Option<&Value>) -> Result<u32, String> {
+    let v = match value {
+        None => return Ok(1),
+        Some(v) => v,
+    };
+
+    // serde_json::Value::as_u64 rejects strings, floats, booleans,
+    // nulls, arrays, and objects — but it silently truncates floats
+    // that are whole numbers (e.g. 2.0 → 2). We must reject floats
+    // explicitly so a server cannot smuggle a v2 cert through the
+    // float shape.
+    if v.is_f64() {
+        return Err(format!(
+            "version is a JSON number with a fractional part ({v}); expected an integer"
+        ));
+    }
+
+    let n = v.as_u64().ok_or_else(|| {
+        // null / string / bool / array / object — anything that is
+        // not a JSON integer.
+        format!("version is {v}; expected a JSON integer")
+    })?;
+
+    // Lossy narrowing: refuse anything that does not fit in u32.
+    // u32::MAX is 4_294_967_295; serde_json only goes up to u64.
+    if n > u32::MAX as u64 {
+        return Err(format!(
+            "version {n} does not fit in u32 (max {})",
+            u32::MAX
+        ));
+    }
+
+    let n = n as u32;
+    if n != 1 {
+        return Err(format!(
+            "this client supports cert version 1 only; server returned {n}"
+        ));
+    }
+    Ok(1)
+}
+
 async fn resolve_cert_id(client: &NodeClient, owner: &str, name: &str, id: &str) -> Result<String> {
     if id.len() >= 36 {
         return Ok(id.to_string());
@@ -811,7 +947,15 @@ mod tests {
         let mut server = mockito::Server::new_async().await;
         let (cert, _root) = cert_server(&mut server, &id, &body, Some(&info)).await;
 
-        let got = cmd_show("alice/r".to_string(), id, server.url(), None, true, None).await;
+        let got = cmd_show(
+            "alice/r".to_string(),
+            id,
+            server.url(),
+            None,
+            true,
+            Some(kp.did().as_str().to_string()),
+        )
+        .await;
         assert!(got.is_ok(), "valid sig + matching issuer must pass");
         cert.assert_async().await;
     }
@@ -827,10 +971,17 @@ mod tests {
         let mut server = mockito::Server::new_async().await;
         let (cert, _root) = cert_server(&mut server, &id, &body, Some(&info)).await;
 
-        let err = cmd_show("alice/r".to_string(), id, server.url(), None, true, None)
-            .await
-            .unwrap_err()
-            .to_string();
+        let err = cmd_show(
+            "alice/r".to_string(),
+            id,
+            server.url(),
+            None,
+            true,
+            Some(other.did().as_str().to_string()),
+        )
+        .await
+        .unwrap_err()
+        .to_string();
         assert!(err.contains("expected issuer"), "got: {err}");
         cert.assert_async().await;
     }
@@ -848,7 +999,7 @@ mod tests {
             .await
             .unwrap_err()
             .to_string();
-        assert!(err.contains("cannot anchor the issuer"), "got: {err}");
+        assert!(err.contains("--expect-node"), "got: {err}");
         cert.assert_async().await;
     }
 
@@ -889,7 +1040,7 @@ mod tests {
             .await
             .unwrap_err()
             .to_string();
-        assert!(err.contains("signature did not verify"), "got: {err}");
+        assert!(err.contains("does not match"), "got: {err}");
         cert.assert_async().await;
     }
 
@@ -973,4 +1124,373 @@ mod tests {
         assert!(report.contains("verify offline"), "got: {report}");
         assert!(!report.contains("WARNING"), "got: {report}");
     }
+
+    /// #26 Split PR 3: a missing `version` field on a cert defaults
+    /// to 1, so an old server's response is forward-compatible with
+    /// a new client. The v1 verify path is taken, and a
+    /// well-signed v1 cert round-trips through `verify_signature`.
+    ///
+    /// Reviewer 1 finding: the previous shape only parsed the field
+    /// and never called `verify_signature`, so the test could not
+    /// fail if the verdict branch was wired to the wrong path. This
+    /// version signs and verifies the full round-trip.
+    #[test]
+    fn missing_version_defaults_to_1_and_verifies() {
+        let kp = gitlawb_core::identity::Keypair::generate();
+        let node_did = kp.did().as_str().to_string();
+        let payload = serde_json::json!({
+            "repo_id": "repo-1",
+            "ref":     "refs/heads/main",
+            "old":     "0".repeat(40),
+            "new":     "a".repeat(40),
+            "pusher":  "did:key:z6MkPusher",
+            "node":    node_did,
+            "ts":      "2026-07-22T00:00:00+00:00",
+        });
+        let sig = kp.sign_b64(&serde_json::to_vec(&payload).unwrap());
+
+        // Simulate the JSON a v1 server returns: no `version` key.
+        let cert_json = serde_json::json!({
+            "id":         "test-id",
+            "repo_id":    "repo-1",
+            "ref_name":   "refs/heads/main",
+            "old_sha":    "0".repeat(40),
+            "new_sha":    "a".repeat(40),
+            "pusher_did": "did:key:z6MkPusher",
+            "node_did":   node_did,
+            "signature":  sig.clone(),
+            "issued_at":  "2026-07-22T00:00:00+00:00",
+            // no `version` field
+        });
+
+        // Forward-compat: the parse yields v1 (the only version this
+        // client verifies), and the verdict arm matching Ok(1) feeds
+        // the v1 payload through verify_signature end to end.
+        assert_eq!(parse_cert_version(cert_json.get("version")).unwrap(), 1);
+        let verdict = verify_signature(
+            "repo-1",
+            "refs/heads/main",
+            &"0".repeat(40),
+            &"a".repeat(40),
+            "did:key:z6MkPusher",
+            &node_did,
+            "2026-07-22T00:00:00+00:00",
+            &sig,
+        );
+        assert!(
+            verdict.is_ok(),
+            "missing version must take the v1 verify path: {verdict:?}"
+        );
+    }
+
+    /// #26 Split PR 3: an explicit `version: 1` on a cert is the v1
+    /// verify path. A v1 cert with an explicit version verifies the
+    /// same as a v1 cert without one — round-trip through
+    /// `verify_signature`.
+    #[test]
+    fn explicit_version_1_takes_v1_path() {
+        let kp = gitlawb_core::identity::Keypair::generate();
+        let node_did = kp.did().as_str().to_string();
+        let payload = serde_json::json!({
+            "repo_id": "repo-1",
+            "ref":     "refs/heads/main",
+            "old":     "0".repeat(40),
+            "new":     "a".repeat(40),
+            "pusher":  "did:key:z6MkPusher",
+            "node":    node_did,
+            "ts":      "2026-07-22T00:00:00+00:00",
+        });
+        let sig = kp.sign_b64(&serde_json::to_vec(&payload).unwrap());
+
+        // Simulate a v1 cert with an explicit `version: 1` field.
+        let cert_json = serde_json::json!({
+            "id":         "test-id",
+            "repo_id":    "repo-1",
+            "ref_name":   "refs/heads/main",
+            "old_sha":    "0".repeat(40),
+            "new_sha":    "a".repeat(40),
+            "pusher_did": "did:key:z6MkPusher",
+            "node_did":   node_did,
+            "signature":  sig.clone(),
+            "issued_at":  "2026-07-22T00:00:00+00:00",
+            "version":    1,
+        });
+
+        // The parse arms the Ok(1) verdict branch, which calls
+        // verify_signature on the v1 payload. A signature mismatch
+        // here would mean the verdict branch was wired to the
+        // wrong path or the payload drifted.
+        assert_eq!(parse_cert_version(cert_json.get("version")).unwrap(), 1);
+        let verdict = verify_signature(
+            "repo-1",
+            "refs/heads/main",
+            &"0".repeat(40),
+            &"a".repeat(40),
+            "did:key:z6MkPusher",
+            &node_did,
+            "2026-07-22T00:00:00+00:00",
+            &sig,
+        );
+        assert!(
+            verdict.is_ok(),
+            "explicit version: 1 must take the v1 verify path: {verdict:?}"
+        );
+    }
+
+    /// #26 Split PR 3 + Reviewer 2: parse_cert_version is the
+    /// load-bearing gate that distinguishes a missing-key (legacy
+    /// server) from a present-but-malformed value. The four cases
+    /// below pin every cell of that truth table; collapsing any
+    /// pair would let a server advertise "version 2" (or 99, or
+    /// 4294967297, or "two") while the client runs the v1
+    /// signature path on a different signed payload — a hostile
+    /// misconfiguration that `--verify` must not silently accept.
+    #[test]
+    fn parse_cert_version_truth_table() {
+        // MISSING key → legacy v1 (the v1 path is the only one that
+        // could match pre-versioning bytes).
+        assert_eq!(
+            parse_cert_version(None).unwrap(),
+            1,
+            "a missing version field is the legacy v1 path"
+        );
+
+        // Explicit 1 → v1.
+        assert_eq!(
+            parse_cert_version(Some(&serde_json::json!(1))).unwrap(),
+            1,
+            "explicit version 1 is the v1 path"
+        );
+
+        // Explicit 2 → Err, NOT Ok(1). Reviewer 2: a v1 signature
+        // must not verify against a server that said "version 2".
+        assert!(
+            parse_cert_version(Some(&serde_json::json!(2))).is_err(),
+            "explicit version 2 must not collapse to v1"
+        );
+
+        // u32 overflow (2^32 + 1) → Err, NOT Ok(1). The previous
+        // `.map(|v| v as u32)` silently truncated this to 1.
+        let overflow = serde_json::json!(u64::from(u32::MAX) + 1);
+        assert!(
+            parse_cert_version(Some(&overflow)).is_err(),
+            "a version that overflows u32 must not truncate to 1"
+        );
+
+        // Non-numeric: a string, a float, null, bool — every shape
+        // that is not a JSON integer is rejected. None of these
+        // may collapse to v1.
+        for bad in [
+            serde_json::json!("two"),
+            serde_json::json!(2.0), // serde_json::Value::is_f64 is true for this
+            serde_json::json!(2.5), // obviously fractional
+            serde_json::json!(true),
+            serde_json::json!(null),
+            serde_json::json!([2]),
+            serde_json::json!({"v": 2}),
+        ] {
+            assert!(
+                parse_cert_version(Some(&bad)).is_err(),
+                "non-integer version {bad} must not collapse to v1"
+            );
+        }
+    }
+
+    /// A version refusal must not print as INVALID — no cryptographic
+    /// check ran, so labelling it INVALID would tell operators an Ed25519
+    /// verification failed when the payload shape was the disqualifier.
+    /// Removing the label branch fails this test.
+    #[test]
+    fn signature_verdict_label_distinguishes_refusal_from_bad_signature() {
+        assert_eq!(
+            signature_verdict_label(&Err("version 2 not supported".to_string())),
+            "UNSUPPORTED VERSION"
+        );
+        assert_eq!(signature_verdict_label(&Ok(1)), "INVALID");
+    }
+
+    /// #26 Split PR 3, round 2: a `version: 2` certificate must fail
+    /// `--verify` THROUGH `cmd_show` itself, not through a copy of its
+    /// match. The prior form of this test duplicated the verdict match in
+    /// the test body and aimed at an `Ok(v != 1)` arm the real parser can
+    /// never produce, so gutting `cmd_show`'s rejection kept every test
+    /// green (demonstrated by the reviewer by execution). Driving the
+    /// command end to end is what makes this a guard: the mock serves a
+    /// cert whose SIGNATURE IS VALID for v1 bytes but which declares
+    /// version 2, and the command must exit Err with the version named —
+    /// the v1 verify path must not launder it through.
+    #[tokio::test]
+    async fn cmd_show_verify_rejects_v2_even_with_valid_v1_signature() {
+        let kp = gitlawb_core::identity::Keypair::generate();
+        let node_did = kp.did().as_str().to_string();
+
+        // Sign a v1-shaped payload so the ONLY disqualifier is the version.
+        let payload = serde_json::json!({
+            "repo_id": "repo-1",
+            "ref":     "refs/heads/main",
+            "old":     "0".repeat(40),
+            "new":     "a".repeat(40),
+            "pusher":  "did:key:z6MkPusher",
+            "node":    node_did,
+            "ts":      "2026-07-22T00:00:00+00:00",
+        });
+        let sig = kp.sign_b64(&serde_json::to_vec(&payload).unwrap());
+
+        let cert_json = serde_json::json!({
+            "id":         "0123456789abcdef0123456789abcdef0123",
+            "repo_id":    "repo-1",
+            "ref_name":   "refs/heads/main",
+            "old_sha":    "0".repeat(40),
+            "new_sha":    "a".repeat(40),
+            "pusher_did": "did:key:z6MkPusher",
+            "node_did":   node_did,
+            "signature":  sig,
+            "issued_at":  "2026-07-22T00:00:00+00:00",
+            "version":    2,
+        });
+
+        let mut server = mockito::Server::new_async().await;
+        // 36-char id short-circuits resolve_cert_id; "owner/repo" input
+        // short-circuits resolve_repo — the cert GET is the only required
+        // route. `/` may be probed for the contextual DID line; serving it
+        // keeps the test about the verdict, not about network noise.
+        let _cert = server
+            .mock(
+                "GET",
+                "/api/v1/repos/o/r/certs/0123456789abcdef0123456789abcdef0123",
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(cert_json.to_string())
+            .create_async()
+            .await;
+        let _root = server
+            .mock("GET", "/")
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(serde_json::json!({ "did": node_did }).to_string())
+            .create_async()
+            .await;
+
+        let err = cmd_show(
+            "o/r".to_string(),
+            "0123456789abcdef0123456789abcdef0123".to_string(),
+            server.url(),
+            None,
+            true,
+            Some(node_did.clone()),
+        )
+        .await
+        .expect_err("a version-2 cert must fail --verify even with a valid v1 signature");
+        let msg = format!("{err:#}");
+        assert!(
+            msg.contains("version"),
+            "the failure must name the version mismatch, not a generic invalid signature: {msg}"
+        );
+    }
+
+    /// Round 2's second demonstrated escape: with no --expect-node, the old
+    /// fallback anchored the issuer to the DID the QUERIED NODE reports at
+    /// `/` — which a forger controls along with the cert, so a wholly forged
+    /// self-signed certificate passed --verify with exit 0. `cmd_show` with
+    /// --verify and no --expect-node must now refuse to claim a trusted
+    /// issuer, even when the served cert's signature is valid and the node's
+    /// self-report matches it.
+    #[tokio::test]
+    async fn cmd_show_verify_without_expect_node_refuses_self_asserted_anchor() {
+        let kp = gitlawb_core::identity::Keypair::generate();
+        let node_did = kp.did().as_str().to_string();
+        let payload = serde_json::json!({
+            "repo_id": "repo-1",
+            "ref":     "refs/heads/main",
+            "old":     "0".repeat(40),
+            "new":     "a".repeat(40),
+            "pusher":  "did:key:z6MkPusher",
+            "node":    node_did,
+            "ts":      "2026-07-22T00:00:00+00:00",
+        });
+        let sig = kp.sign_b64(&serde_json::to_vec(&payload).unwrap());
+        let cert_json = serde_json::json!({
+            "id":         "0123456789abcdef0123456789abcdef0123",
+            "repo_id":    "repo-1",
+            "ref_name":   "refs/heads/main",
+            "old_sha":    "0".repeat(40),
+            "new_sha":    "a".repeat(40),
+            "pusher_did": "did:key:z6MkPusher",
+            "node_did":   node_did,
+            "signature":  sig,
+            "issued_at":  "2026-07-22T00:00:00+00:00",
+        });
+
+        let mut server = mockito::Server::new_async().await;
+        // P3 (reviewer round 2): both mocks are required to be
+        // hit for the test to mean anything. `mockito 1.7`'s
+        // `assert_on_drop` is off by default, so the previous
+        // `_cert` and `_root` bindings would silently let a
+        // regression that skipped the root fetch pass.
+        // `.expect(2)` pins each mock's call count (the cert mock
+        // is hit twice because the test runs `cmd_show` twice) and
+        // `.assert_async()` at the end fires if any are unmet.
+        let cert_mock = server
+            .mock(
+                "GET",
+                "/api/v1/repos/o/r/certs/0123456789abcdef0123456789abcdef0123",
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(cert_json.to_string())
+            .expect(2)
+            .create_async()
+            .await;
+        // The forger's matching self-report — the exact shape
+        // that used to convert into a trust anchor. The guard's
+        // whole point is to make the test FAIL if a future
+        // regression stops calling `/` here: `.expect(2)` +
+        // `.assert_async()` makes that dependency visible.
+        let root_mock = server
+            .mock("GET", "/")
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(serde_json::json!({ "did": node_did }).to_string())
+            .expect(2)
+            .create_async()
+            .await;
+
+        let err = cmd_show(
+            "o/r".to_string(),
+            "0123456789abcdef0123456789abcdef0123".to_string(),
+            server.url(),
+            None,
+            true,
+            None,
+        )
+        .await
+        .expect_err("--verify without --expect-node must not mint a trust anchor from the node's self-report");
+        let msg = format!("{err:#}");
+        assert!(
+            msg.contains("--expect-node"),
+            "the refusal must tell the caller how to anchor trust: {msg}"
+        );
+
+        // Positive control: the same cert with the DID explicitly trusted
+        // passes, so the guard is about the anchor, not the signature.
+        cmd_show(
+            "o/r".to_string(),
+            "0123456789abcdef0123456789abcdef0123".to_string(),
+            server.url(),
+            None,
+            true,
+            Some(node_did),
+        )
+        .await
+        .expect("explicitly anchored --verify of a valid v1 cert must pass");
+
+        // P3 (reviewer round 2): confirm the mocks were actually
+        // hit. Without this, a regression that skipped the cert or
+        // root fetches would still produce a passing test (the
+        // `expect_err` arm succeeds, the positive control succeeds)
+        // and the test would mean nothing.
+        cert_mock.assert_async().await;
+        root_mock.assert_async().await;
+    }
 }

crates/gl/src/doctor.rs

@@ -3,7 +3,7 @@
 //! Checks:
 //!   1. Identity        — ~/.gitlawb/identity.pem exists, DID parseable
 //!   2. Registration    — ~/.gitlawb/ucan.json exists (registered with a node)
-//!   3. GITLAWB_NODE    — env var is set to a non-localhost URL
+//!   3. GITLAWB_NODE    — env var points at a public node (unset is advisory)
 //!   4. Node            — GITLAWB_NODE is reachable (default: https://node.twigpine.com)
 //!   5. git helper      — git-remote-gitlawb is in PATH
 //!   6. git             — git is installed
@@ -80,7 +80,6 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
     });
 
     let mut checks = Vec::new();
-    let mut all_ok = true;
 
     // ── 1. Identity ───────────────────────────────────────────────────────
     let pem_path = dir.join("identity.pem");
@@ -141,32 +140,16 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
     }
 
     // ── 3. GITLAWB_NODE env var ───────────────────────────────────────────
-    match std::env::var("GITLAWB_NODE") {
-        // A loopback host is a legitimate setup (self-hosted node, dev
-        // harness) — the connectivity check below fails loudly if it is not
-        // actually reachable, so don't red-flag the configuration itself.
-        Ok(v) if is_loopback_url(&v) => {
-            checks.push(Check::pass(
-                "GITLAWB_NODE",
-                format!(
-                    "{v} (local node — intentional for self-hosting/dev; unset to target the public network)"
-                ),
-            ));
-        }
-        Ok(v) if !v.is_empty() => {
-            checks.push(Check::pass("GITLAWB_NODE", v.to_string()));
-        }
-        _ => {
-            checks.push(Check::fail(
-                "GITLAWB_NODE",
-                "not set — git-remote-gitlawb will fall back to http://127.0.0.1:7545",
-                "export GITLAWB_NODE=https://node.twigpine.com",
-            ));
-        }
-    }
+    checks.push(gitlawb_node_env_check(
+        std::env::var("GITLAWB_NODE").ok().as_deref(),
+    ));
 
     // ── 4. Node connectivity ──────────────────────────────────────────────
-    let client = NodeClient::new(&args.node, None);
+    // GITLAWB_NODE="" binds args.node to "" through clap, and check 3 groups
+    // empty with unset. The probe must apply the same normalization or it
+    // would dial "" and Fail while the env row only Warns.
+    let node_url = effective_node_url(&args.node);
+    let client = NodeClient::new(node_url, None);
     match client.get("/").await {
         Ok(resp) if resp.status().is_success() => {
             let info = resp.json::<serde_json::Value>().await.unwrap_or_default();
@@ -175,7 +158,7 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
             let short_did = did.chars().take(30).collect::<String>();
             checks.push(Check::pass(
                 "node",
-                format!("{} — v{version} ({short_did}…)", args.node),
+                format!("{} — v{version} ({short_did}…)", node_url),
             ));
             // Capability drift: a node newer than this CLI may require features
             // (RFC 9421 signing, iCaptcha) the CLI doesn't speak.
@@ -194,14 +177,14 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
         Ok(resp) => {
             checks.push(Check::fail(
                 "node",
-                format!("{} returned HTTP {}", args.node, resp.status()),
+                format!("{} returned HTTP {}", node_url, resp.status()),
                 "check GITLAWB_NODE env var or try gl register --node <url>",
             ));
         }
         Err(e) => {
             checks.push(Check::fail(
                 "node",
-                format!("{} unreachable: {e}", args.node),
+                format!("{} unreachable: {e}", node_url),
                 "check your internet connection or set GITLAWB_NODE",
             ));
         }
@@ -285,23 +268,21 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
             CheckState::Fail => "✗",
         };
         println!("  {icon}  {:<24}  {}", check.label, check.detail);
-        if matches!(check.state, CheckState::Fail) {
-            all_ok = false;
-        }
     }
 
     println!();
 
+    let has_failures_now = has_failures(&checks);
     let has_issues = checks
         .iter()
         .any(|c| matches!(c.state, CheckState::Fail | CheckState::Warn));
     if !has_issues {
         println!("Everything looks good. Run `gl quickstart` to create your first repo.");
     } else {
-        if all_ok {
-            println!("Setup looks good with some warnings:");
-        } else {
+        if has_failures_now {
             println!("Some checks failed. Suggested fixes:");
+        } else {
+            println!("Setup looks good with some warnings:");
         }
         for check in &checks {
             if matches!(check.state, CheckState::Fail | CheckState::Warn) {
@@ -311,14 +292,75 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
             }
         }
         println!();
-        if !all_ok {
+        if has_failures_now {
             println!("Run `gl quickstart` for a guided setup.");
         }
     }
 
+    // #357: a non-zero exit is the only signal scripting and CI have. Warn-only
+    // rows stay exit 0 (iCaptcha offline, version drift, shell-alias shadowing
+    // are all non-fatal); Fail-class rows (identity, registration, node
+    // reachable, git-remote-gitlawb present, git present) flip to exit 1.
+    // Direct `process::exit` rather than returning Err: the prose above is the
+    // user-facing summary, and an anyhow Error frame would just duplicate it.
+    if has_failures(&checks) {
+        std::process::exit(1);
+    }
+
     Ok(())
 }
 
+/// Classify the `GITLAWB_NODE` env var for `gl doctor`. Extracted so unit
+/// tests can pin the Warn tier directly (#357): an unset env is advisory, not
+/// a failure, because `--node` defaults to `https://node.twigpine.com` and the
+/// CLI works without it. Flipping this back to `Check::fail` must break a
+/// test, not just change prose.
+fn gitlawb_node_env_check(env_val: Option<&str>) -> Check {
+    match env_val {
+        // A loopback host is a legitimate setup (self-hosted node, dev
+        // harness) — the connectivity check below fails loudly if it is not
+        // actually reachable, so don't red-flag the configuration itself.
+        Some(v) if is_loopback_url(v) => Check::pass(
+            "GITLAWB_NODE",
+            format!(
+                "{v} (local node — intentional for self-hosting/dev; unset to target the public network)"
+            ),
+        ),
+        Some(v) if !v.is_empty() => Check::pass("GITLAWB_NODE", v.to_string()),
+        // `--node` defaults to `https://node.twigpine.com` and the CLI works
+        // fine without the variable, so an unset env is advisory, not a
+        // failure. Warn-class keeps the exit code 0 for a stock working
+        // install; this used to be Fail, which the obvious "return non-zero
+        // on any failure" change would have flipped to exit 1 on every
+        // default-config install (#357).
+        _ => Check::warn(
+            "GITLAWB_NODE",
+            "not set — git-remote-gitlawb will fall back to http://127.0.0.1:7545",
+            format!("export GITLAWB_NODE={PUBLIC_NODE}"),
+        ),
+    }
+}
+
+/// The URL the connectivity probe should dial. `GITLAWB_NODE` set but empty
+/// (or whitespace-only) binds `args.node` to `""` through clap rather than
+/// to the `default_value`, and check 3 already groups empty with unset, so
+/// the probe applies the same normalization: an empty effective URL means
+/// the public node, same as unset.
+fn effective_node_url(node: &str) -> &str {
+    if node.trim().is_empty() {
+        PUBLIC_NODE
+    } else {
+        node
+    }
+}
+
+/// True when at least one check is Fail-class. Warn-only runs (iCaptcha offline,
+/// version drift, shell-alias shadowing, GITLAWB_NODE unset) keep exit 0;
+/// this is the gating predicate for #357's exit-status fix.
+fn has_failures(checks: &[Check]) -> bool {
+    checks.iter().any(|c| matches!(c.state, CheckState::Fail))
+}
+
 /// Check if a binary name exists anywhere on PATH.
 /// True when the rc file contains a real `unalias` command naming `gl` —
 /// not a comment, and not a longer word like `unalias global`. Ordering
@@ -644,4 +686,103 @@ mod tests {
         assert!(matches!(check.state, CheckState::Ok));
         assert!(check.detail.contains("GitHub API returned HTTP 403"));
     }
+
+    /// #357: an all-Ok run does not flip the exit code to non-zero — a healthy
+    /// install must keep returning 0 so scripts like `gl doctor && gl push`
+    /// keep working.
+    #[test]
+    fn exit_predicate_is_false_for_all_ok() {
+        let checks = vec![
+            Check::pass("identity", "ok"),
+            Check::pass("registration", "ok"),
+            Check::pass("git", "ok"),
+        ];
+        assert!(!has_failures(&checks));
+    }
+
+    /// #357: Warn-class rows (iCaptcha offline, version drift, shell-alias
+    /// shadowing, GITLAWB_NODE unset on a default-config install) are NOT
+    /// exit-status failures — they are still printed to the user but the
+    /// process exits 0. This is the rule that the previous "any
+    /// Fail-or-Warn row → exit 1" change would have broken for stock
+    /// installs that simply have not set the env var.
+    #[test]
+    fn exit_predicate_is_false_for_warn_only() {
+        let checks = vec![
+            Check::pass("identity", "ok"),
+            Check::warn("GITLAWB_NODE", "unset", "export GITLAWB_NODE=..."),
+            Check::warn("iCaptcha", "offline", "check connectivity"),
+            Check::warn("version", "drift", "upgrade"),
+            Check::warn("shell alias", "omz git plugin", "unalias gl"),
+        ];
+        assert!(!has_failures(&checks));
+    }
+
+    /// #357: a single Fail-class row trips the exit code. The set of Fail
+    /// rows is whatever `run` writes through `Check::fail`: identity missing
+    /// or unparseable, registration missing/malformed, node unreachable,
+    /// git-remote-gitlawb absent, git absent.
+    #[test]
+    fn exit_predicate_is_true_when_a_fail_row_is_present() {
+        let checks = vec![
+            Check::pass("registration", "ok"),
+            Check::fail("identity", "missing", "gl identity new"),
+            Check::pass("git", "ok"),
+        ];
+        assert!(has_failures(&checks));
+
+        let mixed_warns = vec![
+            Check::pass("identity", "ok"),
+            Check::warn("GITLAWB_NODE", "unset", "export"),
+            Check::fail("node", "unreachable", "check network"),
+            Check::warn("iCaptcha", "offline", "check"),
+        ];
+        assert!(has_failures(&mixed_warns));
+    }
+
+    /// #357: `GITLAWB_NODE` unset must be Warn, not Fail. This pins the
+    /// re-tier so flipping it back to `Check::fail` breaks the suite — the
+    /// previous `has_failures`-only tests built their own `Check::warn` rows
+    /// and could not see the re-tier at all.
+    #[test]
+    fn gitlawb_node_env_unset_is_warn() {
+        let check = gitlawb_node_env_check(None);
+        assert!(matches!(check.state, CheckState::Warn));
+        assert!(check.detail.contains("not set"));
+        let empty = gitlawb_node_env_check(Some(""));
+        assert!(matches!(empty.state, CheckState::Warn));
+    }
+
+    /// `GITLAWB_NODE=""` binds `args.node` to `""` through clap, and check 3
+    /// groups empty with unset. The probe must dial the same URL unset
+    /// would — removing the normalization makes check 4 probe `""` and Fail
+    /// while the env row only Warns, so an empty env install exits 1.
+    #[test]
+    fn effective_node_url_treats_empty_and_blank_as_unset() {
+        assert_eq!(effective_node_url(""), PUBLIC_NODE);
+        assert_eq!(effective_node_url("   "), PUBLIC_NODE);
+        assert_eq!(effective_node_url("\t\n"), PUBLIC_NODE);
+        assert_eq!(
+            effective_node_url("http://127.0.0.1:7545"),
+            "http://127.0.0.1:7545"
+        );
+    }
+
+    #[test]
+    fn gitlawb_node_env_set_is_pass() {
+        let check = gitlawb_node_env_check(Some("https://node.twigpine.com"));
+        assert!(matches!(check.state, CheckState::Ok));
+        assert!(check.detail.contains("https://node.twigpine.com"));
+    }
+
+    #[test]
+    fn gitlawb_node_env_loopback_is_pass() {
+        for url in ["http://127.0.0.1:7545", "http://localhost:7545"] {
+            let check = gitlawb_node_env_check(Some(url));
+            assert!(
+                matches!(check.state, CheckState::Ok),
+                "loopback {url} must be Ok, not Warn/Fail"
+            );
+        }
+    }
 }

crates/gl/tests/doctor_exit.rs

@@ -0,0 +1,48 @@
+//! #357: pin the `gl doctor` exit-code wiring, not just the predicate.
+//!
+//! The unit tests for `has_failures` never touch `std::process::exit(1)` or
+//! the `GITLAWB_NODE` Warn tier. Deleting the exit gate outright kept
+//! `cargo test -p gl doctor::` green, so this binary probe is the only thing
+//! that breaks when the wiring is gutted.
+
+use std::process::Command;
+
+#[test]
+fn doctor_exits_nonzero_when_fail_rows_present() {
+    let dir = tempfile::tempdir().expect("temp dir");
+    // Use an unroutable node so the "node unreachable" check is Fail-class.
+    // The temp dir has no identity.pem / ucan.json, so identity + registration
+    // are also Fail — at least one Fail must flip the process to exit 1.
+    let bin = env!("CARGO_BIN_EXE_gl");
+    let output = Command::new(bin)
+        .args([
+            "doctor",
+            "--dir",
+            dir.path().to_str().unwrap(),
+            "--node",
+            "http://127.0.0.1:1",
+        ])
+        // Inherit PATH but neutralize GITLAWB_NODE so the test is deterministic
+        // across machines that have it set (otherwise it would be Pass).
+        .env_remove("GITLAWB_NODE")
+        // iCaptcha probes a real URL by default; keep it unreachable too so
+        // it stays Warn and does not affect the Fail gating.
+        .env("GITLAWB_ICAPTCHA_URL", "http://127.0.0.1:1")
+        .output()
+        .expect("run gl doctor");
+
+    // The command should have exited 1 because at least one Fail row exists.
+    // If the `std::process::exit(1)` gate is removed, this becomes 0 and the
+    // test fails — that is the regression it pins.
+    assert!(
+        !output.status.success(),
+        "gl doctor must exit non-zero when Fail rows are present; stdout: {}\nstderr: {}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    assert_eq!(
+        output.status.code(),
+        Some(1),
+        "exit code must be 1, not some other non-zero"
+    );
+}

docs/RUN-A-NODE.md

@@ -8,6 +8,7 @@ Step-by-step guide to staking $GITLAWB, registering your node on-chain, and earn
 
 - A wallet with at least **10,000 $GITLAWB** (minimum stake) plus a small amount of ETH on Base for gas
 - Docker or Rust 1.91+ (for running the node process)
+- PostgreSQL 16+ — the node uses `pg_input_is_valid`, which exists only on PostgreSQL 16 and later. The bundled `docker-compose.yml` pins `postgres:16-alpine`; if you supply your own `DATABASE_URL`, the server must be 16+.
 - A public HTTP URL (your-host.com) — can be a VPS, Fly.io app, or anything reachable. A Fly.io config is provided at `infra/fly/fly.toml` (deploy from the repo root with `fly deploy -c infra/fly/fly.toml`)
 
 ---

docs/graphql-pagination.md

@@ -0,0 +1,76 @@
+# GraphQL repository pagination
+
+`repos` returns the complete visible repository list when it contains at most
+200 entries. Larger lists return a GraphQL error directing the caller to
+`reposPage`; they are never silently truncated.
+
+Use `reposPage` for larger lists:
+
+```graphql
+query Repositories($after: String) {
+  reposPage(limit: 50, after: $after) {
+    nodes {
+      name
+      ownerDid
+    }
+    hasNextPage
+    endCursor
+  }
+}
+```
+
+Start with `after: null`. When `hasNextPage` is true, pass the returned
+`endCursor` as `after` on the next request. Stop when `hasNextPage` is false.
+An empty page has a null `endCursor`. The default limit is 50 and the maximum
+page size is 200; limits below 1 or above 200 are rejected (see Query budgets
+below). Malformed cursors, including decoded owner or name strings containing
+NUL, return `invalid repository cursor` before database access.
+
+Pages are ordered by normalized owner DID and repository name. Visibility and
+mirror deduplication are applied in the database before limiting the page;
+hidden and quarantined repositories do not occupy page slots or create a
+continuation signal. Cursors contain a position from the last returned visible
+repository, not a permission grant. Each request checks the current caller's
+visibility independently. Keep the same caller while traversing a list.
+Malformed reader lists in a repository's root visibility rule deny access to
+non-owners for that repository while other visible repositories remain listable.
+
+Pagination is not a snapshot: concurrent renames, ownership changes, or visibility
+changes can alter subsequent pages. Treat cursors as opaque and restart from the
+first page when a fresh complete traversal is needed. Small legacy `repos`
+queries retain their activity ordering.
+
+## Query budgets
+
+Every GraphQL document is validated against two budgets before any resolver
+runs. Both apply per document (per request), not per client: `/graphql` has
+no per-caller rate limiting, so a rejected document can always be retried
+smaller. The constants live in `crates/gitlawb-node/src/graphql/mod.rs`.
+
+- **Complexity budget: 400 per document** (`GRAPHQL_MAX_COMPLEXITY`). Each
+  root field carries a base cost plus the clamped page size plus the selected
+  child fields:
+  - `repos`: 50 + `MAX_VISIBLE_REPO_PAGE_SIZE` + child fields. `repos`
+    always scans the full visible corpus, so it is priced as the full page
+    it is.
+  - `reposPage`: 50 + `MAX_VISIBLE_REPO_PAGE_SIZE` + child fields. Every
+    page request sorts the full deduplicated corpus, so the price does not
+    depend on `limit`: seven `reposPage(limit: 1)` aliases cost the same as
+    seven max-size pages.
+  - `refUpdates`: 100 + clamped `limit` (`MAX_VISIBLE_REF_UPDATES`) + child
+    fields. The 100 base covers the collector's fixed scan (the deduplicated
+    repo set, the quarantined set, the visibility rules, and the bounded
+    up-to-2048-row walk past withheld rows); the limit term prices only the
+    rows returned.
+  - `tasks`: 50 + clamped `limit` (`MAX_VISIBLE_TASK_PAGE_SIZE`) + child
+    fields.
+  - `task`, mutations, subscriptions: 50 + child fields.
+
+  A document whose total exceeds 400 is rejected with
+  `Query is too complex.` before any resolver runs, so an over-budget
+  document needs fewer aliases or fewer child fields. A smaller `limit` does
+  not help `reposPage` or `repos`: both are priced at the full
+  `MAX_VISIBLE_REPO_PAGE_SIZE` scan regardless of `limit`.
+
+- **Depth cap: 14 per document** (`GRAPHQL_MAX_DEPTH`). Documents nested
+  deeper than 14 selections are rejected with `Query is nested too deep.`