fix(node): make creation-route rate limit configurable; fix(gl): surface real errors on pr create failure (#560)

- gitlawb-node: rate limiter was hardcoded to 10 req/hour per DID across all
  creation routes (repo/issue/PR create, register, fork). Adds
  GITLAWB_RATE_LIMIT_MAX and GITLAWB_RATE_LIMIT_WINDOW_SECS config fields
  (clap env-var pattern matching the rest of Config), defaults unchanged.
  Lets node operators tune this for real agent/bot usage without recompiling.

- gl pr create: previously called resp.json() before checking response
  status, so any non-2xx response with an empty or non-JSON body crashed
  with a generic "invalid JSON" error instead of surfacing the real cause.
  Now reads the raw response body first and includes it in the error message
  on failure.

Co-authored-by: carlitos973 <[email protected]>

.env.example

@@ -326,6 +326,19 @@ GITLAWB_IPFS_RESOLVE_BUDGET_SECS=10
 # from the concurrency caps above). 0 disables. Default 600.
 GITLAWB_IPFS_RATE_LIMIT=600
 
+# Per-client-IP rate limit for the anonymous task read routes
+# (GET /api/v1/tasks, GET /api/v1/tasks/{id}), in requests per hour.
+# Both are publicly reachable (optional_signature). GET /api/v1/tasks runs
+# collect_visible_tasks and returns a visibility-filtered page;
+# GET /api/v1/tasks/{id} runs get_visible_task and returns an opaque 404 when
+# the task is hidden or missing. These reads can require task, repository,
+# and visibility-rule queries even when no task is returned, so the brake
+# bounds the cost of anonymous probes. Keyed on the resolved client IP via
+# GITLAWB_TRUSTED_PROXY. 0 disables. Default: 1200 (a list page
+# followed by per-task reads is a normal client pattern, so this sits above
+# the /ipfs budget). The GraphQL/WS task-read brake shares this budget.
+GITLAWB_TASK_READ_RATE_LIMIT=1200
+
 # ── Creation rate limiting (repo/agent/issue/PR flood brake) ──────────────
 # Max creation requests (POST /api/v1/repos, /api/register, fork, issues,
 # pulls) per client IP per hour, in addition to the per-DID limit. The per-DID
@@ -334,6 +347,14 @@ GITLAWB_IPFS_RATE_LIMIT=600
 # brake caps a single-source flood. Uses GITLAWB_TRUSTED_PROXY below to resolve
 # the client IP. 0 disables. Default 120.
 GITLAWB_CREATE_RATE_LIMIT=120
+# Per-DID creation requests allowed inside the window below (repo create,
+# issue create, PR create, register, fork). 0 is rejected at startup: it would
+# silently disable the limiter. Default 10.
+GITLAWB_RATE_LIMIT_MAX=10
+# Window the per-DID creation count above applies over, in seconds. 0 is
+# rejected at startup: every timestamp would prune immediately, disabling the
+# limiter. Default 3600.
+GITLAWB_RATE_LIMIT_WINDOW_SECS=3600
 
 # ── Peer-sync rate limiting (per client IP, uses GITLAWB_TRUSTED_PROXY below) ─
 # /api/v1/peers/announce and /api/v1/sync/notify accept unsigned requests from
@@ -359,6 +380,21 @@ GITLAWB_TRUSTED_PROXY=
 # Enable automatic background sync from known peers
 GITLAWB_AUTO_SYNC=false
 
+# ── External upstream mirroring ───────────────────────────────────────────
+# Default-off scheduled fetch worker for repositories configured as INBOUND.
+# Enabling requires GITLAWB_ENFORCE_OWNER_PUSH=true; the node refuses to start
+# otherwise. The worker updates branches and tags atomically while preserving
+# Gitlawb-owned review/internal refs. Enabling also requires Git >= 2.37 on
+# the host for http.curloptResolve DNS pinning; the node refuses to start
+# when the installed Git is older.
+GITLAWB_UPSTREAM_MIRROR_ENABLED=false
+# Pause after each complete scan. Must be 5..=86400 seconds.
+GITLAWB_UPSTREAM_MIRROR_INTERVAL_SECS=300
+# Deadline for the external fetch child only; the preflight probes run on
+# their own short bound. Must be 1..=3153600000 seconds.
+GITLAWB_UPSTREAM_MIRROR_FETCH_TIMEOUT_SECS=600
+# Keyset page size; every page is processed before the next scan delay.
+GITLAWB_UPSTREAM_MIRROR_PAGE_SIZE=100
 # ── Reconciliation sweep ─────────────────────────────────────────────────
 # Periodic durability sweep: re-derives the public pin set and the withheld-blob
 # recovery set each hour and fills gaps so a dropped replication job never means

README.md

@@ -70,7 +70,7 @@ Known limitations:
 - Repository write authorization is secure by default: `GITLAWB_ENFORCE_OWNER_PUSH` defaults to `true`, so a valid HTTP Signature identifies a pusher and the node additionally requires that pusher to be the repository owner. Set `GITLAWB_ENFORCE_OWNER_PUSH=false` only for the duration of a rolling upgrade whose pushers are not yet the repo owner. Authentication is not authorization on its own (anyone can mint a `did:key` and sign), so owner enforcement is the only thing that prevents every signed caller from pushing to every repository, private ones included.
 - UCAN proof chains are validated when supplied, but UCAN capabilities are not consulted by write authorization and the root issuer is not independently trust-anchored. UCANs therefore do not yet grant scoped collaborator access.
 - Agent lifecycle revocation is not enforced by HTTP Signature authorization; do not rely on removing or revoking an agent record to block a compromised signer.
-- Read visibility is not a blanket data-classification boundary: task, IPFS-pin, and Arweave-anchor listings are not repository-gated; withheld path names can be visible to a root reader; and later visibility changes cannot retract content already announced or externally anchored.
+- Task list/get reads (REST and GraphQL) are repository/task-gated and omit `ucan_token`. Slash-form `repo_id` rows (mirror-hosted tasks) can appear on list for the delegator or trimmed assignee, but `GET /api/v1/tasks/{id}` and GraphQL `task` fail closed for every caller because there is no local repository backing to serve. Read visibility is not a blanket data-classification boundary: IPFS-pin and Arweave-anchor listings are not repository-gated; withheld path names can be visible to a root reader; and later visibility changes cannot retract content already announced or externally anchored.
 - Peer writes are signed by upgraded nodes, but strict signed-peer enforcement is opt-in during rolling upgrades.
 - GraphQL mutations require a signed identity (a verified signer bound to the acting DID), are limited to the agent-task queue, and are covered by a CI completeness fence that requires signer acquisition and actor binding on every registered mutation; repo writes are not exposed over GraphQL and remain REST-only.
 - Pull-request review comments do not yet have threaded line-level anchors, and merges do not enforce approval requirements.
@@ -422,6 +422,7 @@ Important node settings:
 | `GITLAWB_IPFS_REQUEST_BUDGET_SECS` | Absolute wall-clock budget for one admitted `/ipfs/{cid}` request's acquire+walk lifetime. Per-stage clamps bound the acquire and walk stages to the remaining budget, and no stage starts once it is exhausted; the scan then stops with a retryable 503. The object-type probe and content-read `cat-file` subprocesses are budget-checked before starting and each also run under their own deadline (the lesser of `GITLAWB_GIT_SERVICE_TIMEOUT_SECS` and the remaining budget), reaped via process-group teardown, so a hung `cat-file` cannot hold the request's walk slot past it. One hang path is still unbounded: the probe's object-store readability check is a plain filesystem sweep with nothing to reap, so a wedged filesystem can hold the slot past the deadline. Default 600. Accepted range is 1 to 3153600000 (100 years), since the node derives a deadline from this value and a larger one cannot be represented. |
 | `GITLAWB_IPFS_RESOLVE_BUDGET_SECS` | Shorter budget for the pre-walk CID resolve inside an admitted `/ipfs/{cid}` request: the lookup that maps the requested CID to its git oid(s), which runs while the scarce walk admission is already held. A well-formed CID with no pin row does no probe and no walk work, so without this it could hold a walk slot for the whole request budget while nothing walked, and enough such requests shed every real retrieval at admission. The effective deadline is the lesser of this and the remaining request budget, so a value above `GITLAWB_IPFS_REQUEST_BUDGET_SECS` degrades to the request budget. Only the resolve is on this clock; walk and probe work stay on the request budget, so a slow but progressing scan is never shed by it. Default 10. Accepted range is 1 to 3153600000 (100 years). |
 | `GITLAWB_IPFS_RATE_LIMIT` | Max `/ipfs/{cid}` requests per client IP per hour (route flood brake). 0 disables. Default 600. |
+| `GITLAWB_TASK_READ_RATE_LIMIT` | Max anonymous task-read requests (`GET /api/v1/tasks`, `GET /api/v1/tasks/{id}`) per client IP per hour, sharing the budget with the GraphQL/WS task-read brake. Uses `GITLAWB_TRUSTED_PROXY` to resolve the client IP. 0 disables. Default 1200 (above the `/ipfs` budget to allow list pages followed by per-task reads). |
 | `GITLAWB_TIGRIS_BUCKET` | Optional S3/Tigris shared repo storage bucket. |
 | `GITLAWB_PINATA_JWT` | Optional Pinata/IPFS warm-storage pinning. |
 | `GITLAWB_IRYS_URL` | Optional Irys/Arweave permanent anchoring. |

SECURITY.md

@@ -26,7 +26,7 @@ We will acknowledge receipt within 48 hours and aim to release a fix within 14 d
 
 **Content addressing**
 - Every git object is content-addressed via CIDv1 (SHA-256)
-- Tamper-evident by construction — a modified object changes its CID
+- Tamper-evident by construction: a modified object changes its CID
 
 **UCAN token validation**
 - Bootstrap UCAN tokens are issued at registration.
@@ -35,8 +35,8 @@ We will acknowledge receipt within 48 hours and aim to release a fix within 14 d
 - Capability grants are not yet consulted by repository write authorization; see the limitations below.
 
 **Smart contracts (Base Sepolia testnet)**
-- `GitlawbDIDRegistry` — on-chain DID → document registry
-- `GitlawbNameRegistry` — human name → DID registry
+- `GitlawbDIDRegistry`: on-chain DID → document registry
+- `GitlawbNameRegistry`: human name → DID registry
 - Both auditable on-chain, no admin keys
 
 ---
@@ -55,7 +55,7 @@ These are documented limitations of the current live release. They should be pri
 
 ### Repository write authorization defaults
 - `git-receive-pack` verifies HTTP Signatures and, with the default configuration, also requires the pusher DID to be the repository owner: `GITLAWB_ENFORCE_OWNER_PUSH` defaults to `true`.
-- **Impact:** With the default setting, a valid signature authenticates the pusher *and* the pusher must be the repository owner. Authentication is not authorization on its own — anyone can mint a `did:key` and sign — so owner enforcement is the only thing that prevents every signed caller from pushing to every repository, including private ones.
+- **Impact:** With the default setting, a valid signature authenticates the pusher, and the pusher must be the repository owner. Authentication is not authorization on its own, since anyone can mint a `did:key` and sign, so owner enforcement is the only thing that prevents every signed caller from pushing to every repository, including private ones.
 - **Mitigation:** `GITLAWB_ENFORCE_OWNER_PUSH=false` is the explicit rolling-upgrade compatibility override. Disable it only for the duration of an upgrade whose pushers are not yet the repo owner, then turn it back on. Delegated and CI keys count as non-owners: a UCAN `git/push` capability is verified but not yet honored for authorization, so they cannot push while this is on.
 
 ### UCAN delegation and revocation
@@ -73,7 +73,7 @@ These are documented limitations of the current live release. They should be pri
 ### Private repository reads
 - Repository and path-scoped visibility checks are enforced for repository API and Git content reads. A denied whole-repository or root read returns the same 404 shape as a missing repository, so the denial does not reveal private-repository existence.
 - Sparse-clone support exposes withheld path globs to callers who may read the repository root. Do not put sensitive information in withheld path names.
-- `GET /api/v1/tasks`, `/api/v1/ipfs/pins`, and `/api/v1/arweave/anchors` are not repository-gated. Task records include a UCAN token; pin and anchor listings expose object and ref metadata.
+- Task list/get reads (`GET /api/v1/tasks`, `GET /api/v1/tasks/{id}`, and GraphQL `tasks` / `task` over HTTP and WebSocket) are repository/task-gated through shared visibility collectors and omit `ucan_token`. Slash-form `repo_id` rows (mirror-hosted tasks) can appear on list for the delegator or trimmed assignee, but `GET /api/v1/tasks/{id}` and GraphQL `task` fail closed for every caller because there is no local repository backing to serve. All of those transports share the same per-IP `GITLAWB_TASK_READ_RATE_LIMIT` budget. `/api/v1/ipfs/pins` and `/api/v1/arweave/anchors` are not repository-gated; pin and anchor listings expose object and ref metadata.
 - Changing a repository's visibility controls future serving, but cannot retract ref metadata or configured external pins and anchors already announced while the repository was public. Do not push secrets to an announceable repository.
 - **Impact:** Visibility policies protect the repository and Git content routes they gate, not every metadata endpoint or previously published content.
 - **Remaining boundary:** This read control does not address the independent write-authorization and UCAN-delegation limitations described above.
@@ -119,10 +119,10 @@ These are documented limitations of the current live release. They should be pri
 ## Threat Model
 
 Twigpine is designed to be secure against:
-- **Unauthenticated writes** — HTTP Signature auth on protected write endpoints; see Known Limitations for owner and capability authorization gaps
-- **Tampered git objects** — CIDv1 content addressing detects modification
-- **Identity spoofing** — DIDs derived from public keys, unforgeable without the private key
-- **Centralized takedown** — no single point of control; data on IPFS + Arweave
+- **Unauthenticated writes:** HTTP Signature auth on protected write endpoints; see Known Limitations for owner and capability authorization gaps
+- **Tampered git objects:** CIDv1 content addressing detects modification
+- **Identity spoofing:** DIDs derived from public keys, unforgeable without the private key
+- **Centralized takedown:** no single point of control; data on IPFS + Arweave
 
 Twigpine is **not yet** designed to defend against:
 - A compromised node operator (node operators are trusted for their own node)

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

@@ -22,6 +22,7 @@ pub mod replicas;
 pub mod repos;
 pub mod resolve;
 pub mod stars;
+pub mod task_cursor;
 pub mod tasks;
 pub mod visibility;
 pub mod webhooks;

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

@@ -84,7 +84,7 @@ pub(crate) const PUBLIC_HTTP_URL_REQUIREMENT: &str = "must be a public http(s) U
 /// sits at a prefix-length-dependent offset (RFC 6052 §2.2) — so they return
 /// `None`. Any caller that needs them blocked must reject the wider
 /// `64:ff9b::/32` itself; `is_public_http_url` does this in its native-v6 arm.
-fn embedded_ipv4(v6: std::net::Ipv6Addr) -> Option<std::net::Ipv4Addr> {
+pub(crate) fn embedded_ipv4(v6: std::net::Ipv6Addr) -> Option<std::net::Ipv4Addr> {
     if let Some(v4) = v6.to_ipv4_mapped().or_else(|| v6.to_ipv4()) {
         return Some(v4);
     }
@@ -104,6 +104,116 @@ fn embedded_ipv4(v6: std::net::Ipv6Addr) -> Option<std::net::Ipv4Addr> {
     None
 }
 
+/// Whether an IP literal is outside the local/private ranges that an
+/// attacker-controlled outbound URL must never reach. Kept separate from URL
+/// parsing so callers that resolve a DNS name themselves can apply the same
+/// classification to every answer and then pin the approved addresses.
+pub(crate) fn is_public_ip(ip: std::net::IpAddr) -> bool {
+    // Reject loopback/unspecified on the literal as given — catches `::1` and
+    // `::` before the IPv4-folding below (`::1`.to_ipv4() would otherwise map
+    // to a non-loopback `0.0.0.1`).
+    if ip.is_loopback() || ip.is_unspecified() {
+        return false;
+    }
+    // Fold any IPv6 literal that embeds an IPv4 address (mapped, compatible,
+    // 6to4, NAT64) down to that IPv4 so the v4 range checks catch
+    // loopback/private addresses smuggled in via an IPv6 encoding, then
+    // re-check loopback/unspecified.
+    let ip = match ip {
+        std::net::IpAddr::V6(v6) => embedded_ipv4(v6).map(std::net::IpAddr::V4).unwrap_or(ip),
+        v4 => v4,
+    };
+    if ip.is_loopback() || ip.is_unspecified() {
+        return false;
+    }
+    match ip {
+        std::net::IpAddr::V4(v4) => {
+            let o = v4.octets();
+            // RFC1918 private, link-local, CGNAT (100.64.0.0/10), or the
+            // RFC1122 "this host" block 0.0.0.0/8 (never a valid destination;
+            // 0.0.0.0 itself is already caught by the is_unspecified check).
+            if v4.is_private()
+                || v4.is_link_local()
+                || (o[0] == 100 && (o[1] & 0xc0) == 64)
+                || o[0] == 0
+            {
+                return false;
+            }
+        }
+        std::net::IpAddr::V6(v6) => {
+            let s = v6.segments();
+            // fc00::/7 (unique-local), fe80::/10 (link-local), or fec0::/10
+            // (site-local, deprecated by RFC 3879; the link-local mask does
+            // not cover it).
+            if (s[0] & 0xfe00) == 0xfc00 || (s[0] & 0xffc0) == 0xfe80 || (s[0] & 0xffc0) == 0xfec0 {
+                return false;
+            }
+            // Any NAT64 address (64:ff9b::/32) that is not the cleanly
+            // decodable well-known /96 — e.g. the RFC 8215 local-use
+            // 64:ff9b:1::/48 — carries a translated target whose embedded v4
+            // sits at a prefix-length-dependent offset (RFC 6052 §2.2).
+            // Rather than risk a wrong decode across every prefix length we
+            // reject the whole NAT64 space here. The well-known /96 was already
+            // folded to its v4 above and never reaches this arm.
+            if s[0] == 0x0064 && s[1] == 0xff9b {
+                return false;
+            }
+        }
+    }
+    true
+}
+
+/// Mirror-upstream variant of [`is_public_ip`]: in addition to the shared
+/// peer classification, rejects IANA special-purpose ranges that are
+/// non-global, locally meaningful, or unsafe as owner-controlled outbound
+/// fetch endpoints. For IP-literal upstreams, configure-time URL validation
+/// and the fetch-time egress policy share this classifier, so a stored
+/// literal can never fail the worker it is scheduled for. Hostname upstreams
+/// are only shape-checked at configure time: each fetch resolves DNS and
+/// applies this classifier to the answers, so a hostname whose answers turn
+/// non-public still fails at fetch time.
+pub(crate) fn mirror_upstream_ip_permitted(ip: std::net::IpAddr) -> bool {
+    if !is_public_ip(ip) {
+        return false;
+    }
+    let classified = match ip {
+        std::net::IpAddr::V6(v6) => embedded_ipv4(v6).map(std::net::IpAddr::V4).unwrap_or(ip),
+        _ => ip,
+    };
+    match classified {
+        std::net::IpAddr::V4(v4) => {
+            let [a, b, c, _] = v4.octets();
+            // IANA special-purpose ranges that are not globally reachable but
+            // may be routed inside labs, clouds, or operator networks. They are
+            // unsuitable for an owner-controlled outbound fetch even though
+            // `Ipv4Addr::is_private` does not classify them as RFC1918.
+            !(a >= 224
+                || (a == 192 && b == 0 && c == 0)
+                || (a == 192 && b == 0 && c == 2)
+                || (a == 192 && b == 88 && c == 99)
+                || (a == 198 && matches!(b, 18 | 19))
+                || (a == 198 && b == 51 && c == 100)
+                || (a == 203 && b == 0 && c == 113))
+        }
+        std::net::IpAddr::V6(v6) => {
+            let s = v6.segments();
+            // IANA special-purpose ranges that are non-global, locally
+            // meaningful, or unsafe to treat as ordinary forge endpoints.
+            // Reject the complete IETF protocol-assignment /23 rather than
+            // trying to maintain its narrow globally reachable exceptions;
+            // a public forge has no reason to depend on those anycast ranges.
+            !(v6.is_multicast()
+                || (s[0] & 0xffc0) == 0xfec0
+                || (s[0] == 0x0100 && s[1] == 0 && s[2] == 0 && s[3] == 0)
+                || (s[0] == 0x0100 && s[1] == 0 && s[2] == 0 && s[3] == 1)
+                || (s[0] == 0x2001 && (s[1] & 0xfe00) == 0)
+                || (s[0] == 0x2001 && s[1] == 0x0db8)
+                || (s[0] == 0x3fff && (s[1] & 0xf000) == 0)
+                || s[0] == 0x5f00)
+        }
+    }
+}
+
 /// Whether a peer `http_url` is a public http(s) endpoint safe to register.
 /// Rejects non-http(s) schemes, loopback/unspecified/private/link-local IPs,
 /// and `localhost` / `.localhost` / `.local` / `.internal` hostnames. Used at
@@ -137,60 +247,9 @@ pub fn is_public_http_url(raw: &str) -> bool {
     // before parsing as an IP.
     let ip_candidate = host.trim_start_matches('[').trim_end_matches(']');
     if let Ok(ip) = ip_candidate.parse::<std::net::IpAddr>() {
-        // Reject loopback/unspecified on the literal as given — catches `::1`
-        // and `::` before the IPv4-folding below (`::1`.to_ipv4() would
-        // otherwise map to a non-loopback `0.0.0.1`).
-        if ip.is_loopback() || ip.is_unspecified() {
-            return false;
-        }
-        // Fold any IPv6 literal that embeds an IPv4 address (mapped, compatible,
-        // 6to4, NAT64) down to that IPv4 so the v4 range checks catch
-        // loopback/private addresses smuggled in via an IPv6 encoding, then
-        // re-check loopback/unspecified.
-        let ip = match ip {
-            std::net::IpAddr::V6(v6) => embedded_ipv4(v6).map(std::net::IpAddr::V4).unwrap_or(ip),
-            v4 => v4,
-        };
-        if ip.is_loopback() || ip.is_unspecified() {
-            return false;
-        }
-        match ip {
-            std::net::IpAddr::V4(v4) => {
-                let o = v4.octets();
-                // RFC1918 private, link-local, CGNAT (100.64.0.0/10), or the
-                // RFC1122 "this host" block 0.0.0.0/8 (never a valid destination;
-                // 0.0.0.0 itself is already caught by the is_unspecified check).
-                if v4.is_private()
-                    || v4.is_link_local()
-                    || (o[0] == 100 && (o[1] & 0xc0) == 64)
-                    || o[0] == 0
-                {
-                    return false;
-                }
-            }
-            std::net::IpAddr::V6(v6) => {
-                let s = v6.segments();
-                // fc00::/7 (unique-local), fe80::/10 (link-local), or fec0::/10
-                // (site-local, deprecated by RFC 3879 and so never a legitimate
-                // peer address, but still routable on networks that kept it).
-                if (s[0] & 0xfe00) == 0xfc00
-                    || (s[0] & 0xffc0) == 0xfe80
-                    || (s[0] & 0xffc0) == 0xfec0
-                {
-                    return false;
-                }
-                // Any NAT64 address (64:ff9b::/32) that is not the cleanly
-                // decodable well-known /96 — e.g. the RFC 8215 local-use
-                // 64:ff9b:1::/48 — carries a translated target whose embedded v4
-                // sits at a prefix-length-dependent offset (RFC 6052 §2.2).
-                // Rather than risk a wrong decode across every prefix length we
-                // reject the whole NAT64 space here. The well-known /96 was
-                // already folded to its v4 above and never reaches this arm.
-                if s[0] == 0x0064 && s[1] == 0xff9b {
-                    return false;
-                }
-            }
-        }
+        // IP literals share the single DNS/literal classification in
+        // `is_public_ip`; hostname-only rules stay local to this function.
+        return is_public_ip(ip);
     } else if !host.contains('.') {
         // A name with no dot at all. The `.local` / `.internal` rules above are
         // suffix checks, so a single label slips both, and the resolver decides
@@ -616,6 +675,8 @@ mod tests {
     fn rejects_deprecated_site_local_v6() {
         // fec0::/10, deprecated by RFC 3879. The link-local mask (0xffc0 ==
         // 0xfe80) does not cover it: 0xfec0 & 0xffc0 is 0xfec0.
+        assert!(!super::is_public_ip("fec0::1".parse().unwrap()));
+        assert!(!super::is_public_ip("feff:ffff::1".parse().unwrap()));
         for bad in ["http://[fec0::1]/", "http://[feff:ffff::1]:7545"] {
             assert!(!is_public_http_url(bad), "{bad:?} must be rejected");
         }

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

@@ -0,0 +1,782 @@
+//! Opaque, integrity-protected continuation tokens for the task read surfaces.
+//!
+//! The task list pages by keyset over `(created_at, id)`, but the rows a
+//! caller may *see* are a filtered subset of the rows the query has to
+//! *examine*. Handing the caller a raw `(created_at, id)` cursor therefore
+//! forced a choice between two broken options (#327 review): anchor the cursor
+//! on the last visible row, and a window of denied tasks longer than the scan
+//! budget stalls paging forever; or anchor it on the last examined row, and a
+//! denied read leaks the id and timestamp of a task `GET /tasks/{id}` other-
+//! wise 404s.
+//!
+//! A server-issued token removes the choice. The position it carries is the
+//! last *examined* candidate, so paging always advances, and the payload is
+//! opaque and MAC'd, so the caller learns nothing from it and cannot forge one
+//! naming a row of their choosing.
+//!
+//! The position must be *confidential*, not merely unforgeable: a token that
+//! merely signed a base64 payload would still let its holder read the id and
+//! timestamp of the denied row it names, which is the disclosure the token
+//! exists to prevent. The payload is therefore encrypted under a
+//! synthetic-IV construction (SIV): the tag is an HMAC over the filter and the
+//! plaintext, and it doubles as the IV seeding the keystream the plaintext is
+//! XORed with. That needs no randomness source and no dependency beyond the
+//! `hmac`/`sha2` pair already used for webhook signatures and blob recipient
+//! tags, and it is decrypt-last: nothing is parsed until the tag verifies.
+//!
+//! Making the token the *only* accepted cursor also fixes the ordering-domain
+//! bug the same review found. `agent_tasks.created_at` is TEXT and compared as
+//! TEXT, so `...Z` and `...+00:00` denote one instant but sort differently. A
+//! caller-typed timestamp could silently skip or repeat same-time rows. The
+//! token instead carries the stored string verbatim, so the value compared is
+//! always a value the server wrote.
+//!
+//! A token is bound to the *caller* as well as the filter. The position it
+//! names is the last examined candidate, not the last visible one, so it
+//! encodes how far a scan got under one caller's visibility. Resuming it as a
+//! different caller would start the scan past rows that caller is entitled to
+//! read, silently dropping them from the answer (#327 review). The presenting
+//! caller's normalized DID (empty and flagged absent when anonymous) is
+//! therefore part of the MAC input, so a token presented by anyone else fails
+//! to verify rather than under-reporting.
+//!
+//! Wire form: `v1.<iv>.<body>`, both parts base64url unpadded: the 16-byte
+//! synthetic IV (the truncated MAC, which also seeds the keystream) and the
+//! padded, encrypted JSON payload.
+
+use base64::engine::general_purpose::URL_SAFE_NO_PAD;
+use base64::Engine;
+use hmac::{Hmac, Mac};
+use serde::{Deserialize, Serialize};
+use sha2::Sha256;
+
+use crate::error::AppError;
+
+type HmacSha256 = Hmac<Sha256>;
+
+const CURSOR_PREFIX: &str = "v1";
+const KEY_DERIVATION_LABEL: &[u8] = b"gitlawb/tasks-cursor-key/v1";
+const TAG_DOMAIN: &[u8] = b"gitlawb/tasks-cursor-tag/v1";
+const STREAM_DOMAIN: &[u8] = b"gitlawb/tasks-cursor-stream/v1";
+/// Truncated MAC length, and the synthetic IV width. 128 bits is far beyond
+/// forgery reach for a token that carries no authority of its own (every page
+/// re-runs the visibility gate against the presenting caller), and keeps the
+/// token short enough to sit in a query string.
+const TAG_LEN: usize = 16;
+
+/// Padding bucket for the serialized cursor payload, in bytes. The keystream
+/// XORs the plaintext byte for byte, so token length would otherwise track
+/// payload length and leak the named row's field widths to a holder who
+/// cannot decrypt anything. 128 comfortably covers the widest realistic
+/// payload: a 35-char RFC3339 timestamp (`to_rfc3339()` renders 29 chars at
+/// millisecond precision and 32 at microsecond, 35 only with all nine
+/// fractional digits), a 36-char server-minted UUID, and the fixed JSON
+/// envelope plus a 10-digit expiry, which together total 101 bytes. Should a future payload ever exceed the bucket, the length
+/// rounds up to the next multiple instead of failing, so `encode` never
+/// breaks — it just mints a longer token.
+const PAYLOAD_BUCKET: usize = 128;
+
+/// How long an issued cursor stays acceptable. Keyset positions never go stale
+/// on their own — `created_at`/`id` are immutable — so this is not a
+/// correctness bound. It bounds how long a token stays valid across a node
+/// restart-and-rotate and keeps an abandoned page from being resumed
+/// indefinitely.
+const CURSOR_TTL_SECS: i64 = 24 * 60 * 60;
+
+/// Node-keyed MAC key for continuation tokens, derived from the node keypair
+/// seed so it needs no configuration and survives restarts of the same node.
+/// Derived rather than used directly so a token forgery oracle could not
+/// bear on the signing key itself.
+#[derive(Clone)]
+pub struct TaskCursorKey([u8; 32]);
+
+impl std::fmt::Debug for TaskCursorKey {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("TaskCursorKey(<redacted>)")
+    }
+}
+
+impl TaskCursorKey {
+    pub fn derive(node_seed: &[u8; 32]) -> Self {
+        let mut mac = HmacSha256::new_from_slice(node_seed).expect("HMAC accepts any key length");
+        mac.update(KEY_DERIVATION_LABEL);
+        let mut key = [0u8; 32];
+        key.copy_from_slice(&mac.finalize().into_bytes());
+        Self(key)
+    }
+}
+
+/// The keyset position a token carries: the last candidate row the previous
+/// request examined, whether or not the caller was allowed to see it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TaskPosition {
+    pub created_at: String,
+    pub id: String,
+}
+
+impl TaskPosition {
+    pub fn new(created_at: impl Into<String>, id: impl Into<String>) -> Self {
+        Self {
+            created_at: created_at.into(),
+            id: id.into(),
+        }
+    }
+
+    pub fn as_pair(&self) -> (&str, &str) {
+        (self.created_at.as_str(), self.id.as_str())
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+struct CursorPayload<'a> {
+    /// `created_at` of the last examined candidate, verbatim as stored.
+    t: &'a str,
+    /// `id` of the last examined candidate.
+    i: &'a str,
+    /// Unix expiry.
+    e: i64,
+}
+
+/// The filter a page was issued under. A cursor is only meaningful against the
+/// same `status`/`assignee_did` filter that produced it: resuming a filtered
+/// scan from an unfiltered position (or the reverse) silently skips rows.
+/// Bound into the MAC rather than stored in the payload, so it costs no token
+/// length and cannot be edited without invalidating the tag.
+#[derive(Debug, Clone, Copy)]
+pub struct TaskFilter<'a> {
+    pub status: Option<&'a str>,
+    pub assignee_did: Option<&'a str>,
+}
+
+/// The visibility identity a token was issued to. Normalized through
+/// `normalize_owner_key` so the two spellings of one `did:key` identity
+/// (`did:key:X` and bare `X`) bind identically, matching `did_matches` on the
+/// read path: a caller who authenticates with the other spelling of their own
+/// DID must be able to resume their own page.
+fn caller_binding(caller: Option<&str>) -> &str {
+    caller.map(crate::db::normalize_owner_key).unwrap_or("")
+}
+
+/// The assignee filter a token was issued under. Normalized through
+/// `normalize_owner_key` so the two spellings of one `did:key` identity
+/// (`did:key:X` and bare `X`) bind identically, matching `list_tasks_keyset`
+/// in SQL: minting under `did:key:X` and resuming with bare `X` (or the reverse)
+/// hits the same rows and must verify under the MAC.
+fn assignee_binding(assignee: Option<&str>) -> &str {
+    assignee.map(crate::db::normalize_owner_key).unwrap_or("")
+}
+
+fn cursor_mac(
+    key: &TaskCursorKey,
+    filter: TaskFilter<'_>,
+    caller: Option<&str>,
+    plaintext: &[u8],
+) -> HmacSha256 {
+    let mut mac = HmacSha256::new_from_slice(&key.0).expect("HMAC accepts any key length");
+    mac.update(TAG_DOMAIN);
+    // Length-prefix every field so no two distinct filter/caller/plaintext
+    // tuples can produce the same MAC input.
+    for field in [
+        filter.status.unwrap_or("").as_bytes(),
+        assignee_binding(filter.assignee_did).as_bytes(),
+        caller_binding(caller).as_bytes(),
+        plaintext,
+    ] {
+        mac.update(&(field.len() as u64).to_be_bytes());
+        mac.update(field);
+    }
+    // Presence is distinct from emptiness for all three optional fields, so an
+    // anonymous token cannot collide with one issued to a caller whose
+    // normalized DID is the empty string.
+    mac.update(&[
+        u8::from(filter.status.is_some()),
+        u8::from(filter.assignee_did.is_some()),
+        u8::from(caller.is_some()),
+    ]);
+    mac
+}
+
+/// Synthetic IV: the authentication tag over the filter and plaintext, which
+/// also seeds the keystream. Deterministic by construction, so no randomness
+/// source is needed and two tokens for the same page are byte-identical.
+fn siv(
+    key: &TaskCursorKey,
+    filter: TaskFilter<'_>,
+    caller: Option<&str>,
+    plaintext: &[u8],
+) -> [u8; TAG_LEN] {
+    let mut out = [0u8; TAG_LEN];
+    out.copy_from_slice(
+        &cursor_mac(key, filter, caller, plaintext)
+            .finalize()
+            .into_bytes()[..TAG_LEN],
+    );
+    out
+}
+
+/// XOR `buf` with the keystream for `iv`. Its own inverse, so encrypt and
+/// decrypt are the same call.
+fn apply_keystream(key: &TaskCursorKey, iv: &[u8; TAG_LEN], buf: &mut [u8]) {
+    for (block_index, chunk) in buf.chunks_mut(32).enumerate() {
+        let mut mac = HmacSha256::new_from_slice(&key.0).expect("HMAC accepts any key length");
+        mac.update(STREAM_DOMAIN);
+        mac.update(iv);
+        mac.update(&(block_index as u64).to_be_bytes());
+        let block = mac.finalize().into_bytes();
+        for (byte, k) in chunk.iter_mut().zip(block.iter()) {
+            *byte ^= k;
+        }
+    }
+}
+
+/// Mint a token resuming at `position` for `filter`, usable only by `caller`.
+///
+/// The serialized payload is padded with trailing whitespace to a fixed
+/// bucket before the SIV tag is computed: the keystream preserves plaintext
+/// length, so an unpadded payload would let the token's length leak the
+/// named row's field widths (a 25-char `to_rfc3339()` rendering mints a
+/// shorter token than a 35-char one). `decode` tolerates the padding because
+/// `serde_json` ignores trailing whitespace, and the tag covers the padded
+/// bytes, so a token minted before padding landed still verifies.
+pub fn encode(
+    key: &TaskCursorKey,
+    filter: TaskFilter<'_>,
+    caller: Option<&str>,
+    position: &TaskPosition,
+) -> String {
+    let mut payload = serde_json::to_vec(&CursorPayload {
+        t: &position.created_at,
+        i: &position.id,
+        e: chrono::Utc::now().timestamp() + CURSOR_TTL_SECS,
+    })
+    .expect("cursor payload is plain strings and an integer");
+    let padded_len = payload.len().next_multiple_of(PAYLOAD_BUCKET);
+    payload.resize(padded_len, b' ');
+    let iv = siv(key, filter, caller, &payload);
+    apply_keystream(key, &iv, &mut payload);
+    format!(
+        "{CURSOR_PREFIX}.{}.{}",
+        URL_SAFE_NO_PAD.encode(iv),
+        URL_SAFE_NO_PAD.encode(&payload)
+    )
+}
+
+/// One rejection message for every way a token can fail to verify. A caller
+/// who mangled a token, replayed an expired one, or tried to move one to a
+/// different filter or a different presenting identity learns only that the
+/// cursor is not usable — never which of those it was, and never anything
+/// about the row it named.
+const INVALID_CURSOR: &str = "invalid or expired cursor";
+
+fn reject() -> AppError {
+    AppError::BadRequest(INVALID_CURSOR.into())
+}
+
+/// Verify a token against `filter` and the presenting `caller`, and return the
+/// position it carries.
+pub fn decode(
+    key: &TaskCursorKey,
+    filter: TaskFilter<'_>,
+    caller: Option<&str>,
+    token: &str,
+) -> crate::error::Result<TaskPosition> {
+    let mut parts = token.split('.');
+    let (Some(version), Some(iv_b64), Some(body_b64), None) =
+        (parts.next(), parts.next(), parts.next(), parts.next())
+    else {
+        return Err(reject());
+    };
+    if version != CURSOR_PREFIX {
+        return Err(reject());
+    }
+    let iv_bytes = URL_SAFE_NO_PAD.decode(iv_b64).map_err(|_| reject())?;
+    let mut plaintext = URL_SAFE_NO_PAD.decode(body_b64).map_err(|_| reject())?;
+    let iv: [u8; TAG_LEN] = iv_bytes.try_into().map_err(|_| reject())?;
+
+    apply_keystream(key, &iv, &mut plaintext);
+    // Authenticate before parsing: until the tag matches, `plaintext` is just
+    // attacker-chosen bytes run through a keystream. `verify_truncated_left`
+    // is the constant-time compare, so a forgery attempt cannot be steered by
+    // timing the first differing byte.
+    cursor_mac(key, filter, caller, &plaintext)
+        .verify_truncated_left(&iv)
+        .map_err(|_| reject())?;
+
+    let decoded: CursorPayload<'_> = serde_json::from_slice(&plaintext).map_err(|_| reject())?;
+    if decoded.e < chrono::Utc::now().timestamp() {
+        return Err(reject());
+    }
+    Ok(TaskPosition::new(decoded.t, decoded.i))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn key() -> TaskCursorKey {
+        TaskCursorKey::derive(&[7u8; 32])
+    }
+
+    fn unfiltered() -> TaskFilter<'static> {
+        TaskFilter {
+            status: None,
+            assignee_did: None,
+        }
+    }
+
+    #[test]
+    fn round_trips_position_verbatim() {
+        let k = key();
+        // A stored timestamp the server wrote, fractional digits and all.
+        let pos = TaskPosition::new("2026-01-03T00:00:00.123456789+00:00", "task-a");
+        let token = encode(&k, unfiltered(), None, &pos);
+        assert_eq!(decode(&k, unfiltered(), None, &token).unwrap(), pos);
+        // A sub-bucket payload pads to exactly one bucket: the padding
+        // claim is an exact width, and every other width assertion here is
+        // bucket-relative, so only an absolute check catches a pad that
+        // skips "next multiple" or a drifted bucket constant.
+        let body = token.split('.').nth(2).expect("token has a body part");
+        let raw = URL_SAFE_NO_PAD.decode(body).unwrap();
+        assert_eq!(
+            raw.len(),
+            PAYLOAD_BUCKET,
+            "normal token must pad to exactly one bucket"
+        );
+    }
+
+    /// The whole point of the token is that the caller may hold it without
+    /// learning the denied row it names. A signed-but-plaintext payload would
+    /// pass every other test here and still fail this one.
+    #[test]
+    fn token_does_not_expose_the_row_it_names() {
+        let k = key();
+        let token = encode(
+            &k,
+            unfiltered(),
+            None,
+            &TaskPosition::new("2026-01-03T00:00:00+00:00", "denied-task-id"),
+        );
+        let body = token.split('.').nth(2).expect("token has a body part");
+        let raw = URL_SAFE_NO_PAD.decode(body).unwrap();
+        let as_text = String::from_utf8_lossy(&raw);
+        for secret in ["denied-task-id", "2026-01-03", "\"t\"", "\"i\""] {
+            assert!(
+                !as_text.contains(secret),
+                "token body must not carry {secret:?} in the clear: {as_text:?}"
+            );
+        }
+        // Also assert it is not merely reordered or whitespace-mangled JSON.
+        assert!(
+            serde_json::from_slice::<serde_json::Value>(&raw).is_err(),
+            "token body must not be parseable as JSON"
+        );
+    }
+
+    /// The keystream preserves plaintext length, so without the fixed-width
+    /// padding a token naming a 35-char timestamp would be measurably longer
+    /// than one naming a 32-char one: the holder would learn the withheld
+    /// row's field widths without decrypting anything. Mint the same shape
+    /// both ways and require identical token length.
+    #[test]
+    fn token_length_does_not_vary_with_position_width() {
+        let k = key();
+        // Ids are server-minted fixed-length UUIDs; the timestamp is the
+        // width that varies in production (`to_rfc3339()` renders 35 chars
+        // with nanoseconds, 25 without fractional seconds).
+        let id = "b3f7a1c2-4d5e-4f6a-8b9c-0d1e2f3a4b5c";
+        let narrow = encode(
+            &k,
+            unfiltered(),
+            None,
+            &TaskPosition::new("2026-01-03T00:00:00+00:00", id),
+        );
+        let wide = encode(
+            &k,
+            unfiltered(),
+            None,
+            &TaskPosition::new("2026-01-03T00:00:00.123456789+00:00", id),
+        );
+        assert_eq!(
+            narrow.len(),
+            wide.len(),
+            "tokens must not leak the named row's field widths: {narrow} vs {wide}"
+        );
+        // The padded token still decodes to the exact stored string.
+        assert_eq!(
+            decode(&k, unfiltered(), None, &wide).unwrap(),
+            TaskPosition::new("2026-01-03T00:00:00.123456789+00:00", id)
+        );
+    }
+
+    /// A token minted before the fixed-width padding landed (unpadded payload,
+    /// tag over the raw bytes) must still verify: `decode` MACs whatever
+    /// plaintext the keystream recovers and `serde_json` tolerates the
+    /// missing trailing whitespace. This pins the back-compat claim in the
+    /// `encode` docstring with a frozen v1 artifact, so a change to the v1
+    /// recipe pins (domains, MAC input, keystream, segment layout, field
+    /// names) breaks this test instead of silently re-deriving under the
+    /// new internals. It does not pin the padding: `decode` never consults
+    /// `PAYLOAD_BUCKET` and the fixture is unpadded, so the padding-width
+    /// claim belongs to `oversized_payload_rounds_up_to_next_bucket`.
+    ///
+    /// Fixture expiry is fixed at 2100-01-01 (`4102444800`): it must stay in
+    /// the future whenever the test runs.
+    #[test]
+    fn unpadded_legacy_token_still_verifies() {
+        let k = key();
+        let filter = unfiltered();
+        let position = TaskPosition::new("2026-01-03T00:00:00+00:00", "legacy-task");
+        // Pre-padding recipe minted once and frozen: serialize with no
+        // padding, tag, and encrypt. Deterministic by construction (SIV needs
+        // no randomness), so this literal is stable across runs.
+        let token = "v1.ILJyy7JZScm_h-HM1-CggA.l4kJUoHTbqSGTVKL0DP4hORnbOS1_gHYV2fwJYXd2QtM-6TBMCKh-wvluIiaEYC9rUSUPVSteAKkp7jBCmK_bR2D";
+        assert_eq!(
+            decode(&k, filter, None, token).unwrap(),
+            position,
+            "pre-padding token must verify under the current decode"
+        );
+        // The fixture is a genuine pre-padding artifact, so its decoded
+        // body must stay below one bucket: a literal re-minted under the
+        // padded recipe would retire this back-compat pin silently.
+        let body = token.split('.').nth(2).expect("token has a body part");
+        let raw = URL_SAFE_NO_PAD.decode(body).unwrap();
+        assert!(
+            raw.len() < PAYLOAD_BUCKET,
+            "legacy fixture must be the unpadded shape: {} bytes",
+            raw.len()
+        );
+    }
+
+    /// The `PAYLOAD_BUCKET` docstring promises an oversized payload rounds up
+    /// to the next multiple instead of failing. Mint a position whose id
+    /// pushes the serialized payload past one bucket and require the token
+    /// to round-trip (at two buckets).
+    #[test]
+    fn oversized_payload_rounds_up_to_next_bucket() {
+        let k = key();
+        // ~140-char id: payload crosses the 128-byte bucket.
+        let long_id = "a".repeat(140);
+        let position = TaskPosition::new("2026-01-03T00:00:00+00:00", long_id.clone());
+        let probe = serde_json::to_vec(&CursorPayload {
+            t: &position.created_at,
+            i: &position.id,
+            e: 1_700_000_000,
+        })
+        .expect("cursor payload is plain strings and an integer");
+        assert!(
+            probe.len() > PAYLOAD_BUCKET,
+            "fixture payload must exceed one bucket: {} bytes",
+            probe.len()
+        );
+        assert!(
+            probe.len() <= 2 * PAYLOAD_BUCKET,
+            "fixture payload must fit in two buckets: {} bytes",
+            probe.len()
+        );
+        let token = encode(&k, unfiltered(), None, &position);
+        assert_eq!(
+            decode(&k, unfiltered(), None, &token).unwrap(),
+            position,
+            "oversized payload must round-trip at two buckets"
+        );
+        // The padded body must be exactly two buckets, not merely longer
+        // than one: the padding claim is an exact width, not a direction.
+        let body = token.split('.').nth(2).expect("token has a body part");
+        let raw = URL_SAFE_NO_PAD.decode(body).unwrap();
+        assert_eq!(
+            raw.len(),
+            2 * PAYLOAD_BUCKET,
+            "oversized payload must pad to exactly two buckets"
+        );
+    }
+
+    /// Callers paste the token straight into a query string, so it must carry
+    /// no character that needs percent-encoding.
+    #[test]
+    fn token_is_url_safe_verbatim() {
+        let token = encode(
+            &key(),
+            TaskFilter {
+                status: Some("pending"),
+                assignee_did: Some("did:key:z6MkAssignee"),
+            },
+            None,
+            &TaskPosition::new("2026-01-03T00:00:00.123456789+00:00", "task-a"),
+        );
+        assert!(
+            token
+                .chars()
+                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')),
+            "token must be query-safe without encoding: {token}"
+        );
+    }
+
+    /// Flipping any byte of a token must fail the tag, not decode to a
+    /// different position: the keystream is malleable on its own, the tag is
+    /// what stops a chosen-position forgery.
+    #[test]
+    fn rejects_any_tampered_byte() {
+        let k = key();
+        let token = encode(
+            &k,
+            unfiltered(),
+            None,
+            &TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a"),
+        );
+        let parts: Vec<&str> = token.split('.').collect();
+        for part in [1usize, 2] {
+            for position in [0usize, 3] {
+                let mut bytes = parts[part].as_bytes().to_vec();
+                bytes[position] = if bytes[position] == b'A' { b'B' } else { b'A' };
+                let mut mangled: Vec<String> = parts.iter().map(|p| p.to_string()).collect();
+                mangled[part] = String::from_utf8(bytes).unwrap();
+                assert!(
+                    decode(&k, unfiltered(), None, &mangled.join(".")).is_err(),
+                    "byte {position} of part {part} must not be malleable"
+                );
+            }
+        }
+        // Sanity: the untouched token still decodes, so the loop above is not
+        // passing because every token is rejected.
+        assert!(decode(&k, unfiltered(), None, &token).is_ok());
+    }
+
+    #[test]
+    fn rejects_token_from_another_node() {
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let token = encode(&TaskCursorKey::derive(&[1u8; 32]), unfiltered(), None, &pos);
+        assert!(decode(
+            &TaskCursorKey::derive(&[2u8; 32]),
+            unfiltered(),
+            None,
+            &token
+        )
+        .is_err());
+    }
+
+    #[test]
+    fn rejects_cursor_moved_to_a_different_filter() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let token = encode(
+            &k,
+            TaskFilter {
+                status: Some("pending"),
+                assignee_did: None,
+            },
+            None,
+            &pos,
+        );
+        assert!(decode(&k, unfiltered(), None, &token).is_err());
+        assert!(decode(
+            &k,
+            TaskFilter {
+                status: Some("claimed"),
+                assignee_did: None
+            },
+            None,
+            &token
+        )
+        .is_err());
+        assert!(decode(
+            &k,
+            TaskFilter {
+                status: Some("pending"),
+                assignee_did: None
+            },
+            None,
+            &token
+        )
+        .is_ok());
+    }
+
+    /// A filter of `Some("")` must not verify a token minted with `None`.
+    #[test]
+    fn distinguishes_absent_filter_from_empty_filter() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let token = encode(&k, unfiltered(), None, &pos);
+        assert!(decode(
+            &k,
+            TaskFilter {
+                status: Some(""),
+                assignee_did: None
+            },
+            None,
+            &token
+        )
+        .is_err());
+    }
+
+    /// A cursor names how far a scan got under one caller's visibility. Moved
+    /// to a different presenting identity it would start that caller's scan
+    /// past rows they are entitled to read, so it must fail to verify rather
+    /// than silently under-report (#327 review).
+    #[test]
+    fn rejects_cursor_moved_to_a_different_caller() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let alice = Some("did:key:z6MkAlice");
+        let bob = Some("did:key:z6MkBob");
+
+        let anon_token = encode(&k, unfiltered(), None, &pos);
+        assert!(
+            decode(&k, unfiltered(), alice, &anon_token).is_err(),
+            "an anonymously minted cursor must not resume an authenticated scan"
+        );
+        assert!(decode(&k, unfiltered(), None, &anon_token).is_ok());
+
+        let alice_token = encode(&k, unfiltered(), alice, &pos);
+        assert!(
+            decode(&k, unfiltered(), bob, &alice_token).is_err(),
+            "one caller's cursor must not resume another caller's scan"
+        );
+        assert!(
+            decode(&k, unfiltered(), None, &alice_token).is_err(),
+            "an authenticated cursor must not resume an anonymous scan"
+        );
+        assert!(decode(&k, unfiltered(), alice, &alice_token).is_ok());
+    }
+
+    /// The binding is on identity, not spelling: `did:key:X` and bare `X` are
+    /// one caller everywhere else on the read path (`did_matches`), so a caller
+    /// who presents the other form of their own DID must keep their own page.
+    #[test]
+    fn caller_binding_is_normalized_across_did_key_spellings() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let token = encode(&k, unfiltered(), Some("did:key:z6MkAlice"), &pos);
+        assert_eq!(
+            decode(&k, unfiltered(), Some("z6MkAlice"), &token).unwrap(),
+            pos
+        );
+        // A different method sharing the base58 tail is a different principal.
+        assert!(decode(&k, unfiltered(), Some("did:web:z6MkAlice"), &token).is_err());
+    }
+
+    /// The filter binding is on identity, not spelling: `assignee_did` matches
+    /// normalized keys in SQL (`normalize_owner_key`), so minting under
+    /// `did:key:X` and resuming under bare `X` (or vice versa) must succeed,
+    /// while `did:web:X` remains distinct.
+    #[test]
+    fn assignee_filter_binding_is_normalized_across_did_key_spellings() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let filter_did_key = TaskFilter {
+            status: Some("pending"),
+            assignee_did: Some("did:key:z6MkAssignee"),
+        };
+        let filter_bare = TaskFilter {
+            status: Some("pending"),
+            assignee_did: Some("z6MkAssignee"),
+        };
+        let filter_web = TaskFilter {
+            status: Some("pending"),
+            assignee_did: Some("did:web:z6MkAssignee"),
+        };
+
+        let token_from_did_key = encode(&k, filter_did_key, None, &pos);
+        assert_eq!(
+            decode(&k, filter_bare, None, &token_from_did_key).unwrap(),
+            pos
+        );
+        assert!(decode(&k, filter_web, None, &token_from_did_key).is_err());
+
+        let token_from_bare = encode(&k, filter_bare, None, &pos);
+        assert_eq!(
+            decode(&k, filter_did_key, None, &token_from_bare).unwrap(),
+            pos
+        );
+        assert!(decode(&k, filter_web, None, &token_from_bare).is_err());
+    }
+
+    /// Anonymous is a distinct binding from a caller whose normalized DID is
+    /// the empty string, the same way `Some("")` is distinct from `None` for
+    /// the filter fields.
+    #[test]
+    fn distinguishes_anonymous_from_empty_caller() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let token = encode(&k, unfiltered(), None, &pos);
+        assert!(decode(&k, unfiltered(), Some(""), &token).is_err());
+    }
+
+    /// Length-prefixing must stop a status/assignee pair from being re-split.
+    #[test]
+    fn rejects_field_boundary_shift() {
+        let k = key();
+        let pos = TaskPosition::new("2026-01-03T00:00:00+00:00", "task-a");
+        let token = encode(
+            &k,
+            TaskFilter {
+                status: Some("pend"),
+                assignee_did: Some("ing"),
+            },
+            None,
+            &pos,
+        );
+        assert!(decode(
+            &k,
+            TaskFilter {
+                status: Some("pending"),
+                assignee_did: Some("")
+            },
+            None,
+            &token
+        )
+        .is_err());
+    }
+
+    #[test]
+    fn rejects_expired_token() {
+        let k = key();
+        // Minted the same way `encode` does, but already past its expiry.
+        let mut payload = serde_json::to_vec(&CursorPayload {
+            t: "2026-01-03T00:00:00+00:00",
+            i: "task-a",
+            e: chrono::Utc::now().timestamp() - 1,
+        })
+        .unwrap();
+        let iv = siv(&k, unfiltered(), None, &payload);
+        apply_keystream(&k, &iv, &mut payload);
+        let expired = format!(
+            "v1.{}.{}",
+            URL_SAFE_NO_PAD.encode(iv),
+            URL_SAFE_NO_PAD.encode(&payload)
+        );
+        let err = decode(&k, unfiltered(), None, &expired).unwrap_err();
+        assert!(err.to_string().contains(INVALID_CURSOR));
+    }
+
+    #[test]
+    fn rejects_malformed_shapes() {
+        let k = key();
+        for bad in [
+            "",
+            "v1",
+            "v1.",
+            "v1.abc",
+            "v2.abc.def",
+            "v1.abc.def.ghi",
+            "v1.!!!.def",
+        ] {
+            assert!(
+                decode(&k, unfiltered(), None, bad).is_err(),
+                "must reject {bad:?}"
+            );
+        }
+    }
+
+    /// Every rejection path renders the same text, so a caller cannot tell a
+    /// forged token from an expired one from a filter mismatch.
+    #[test]
+    fn every_rejection_is_indistinguishable() {
+        let k = key();
+        for bad in ["v1.abc.def", "not-a-cursor", "v1..", "v9.a.b"] {
+            assert_eq!(
+                decode(&k, unfiltered(), None, bad).unwrap_err().to_string(),
+                format!("invalid request: {INVALID_CURSOR}")
+            );
+        }
+    }
+}

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

@@ -634,10 +634,12 @@ mod tests {
             .connect_lazy("postgres://localhost/gitlawb_test_placeholder")
             .expect("lazy pool creation should not fail");
         let db = Arc::new(crate::db::Db::for_testing(pool.clone()));
+        let task_cursor_key = crate::api::task_cursor::TaskCursorKey::derive(&keypair.to_seed());
         let schema = Arc::new(graphql::build_schema(
             db.clone(),
             ref_tx.clone(),
             task_tx.clone(),
+            task_cursor_key.clone(),
         ));
         crate::state::AppState {
             config: Arc::new(Config::parse_from(["gitlawb-node"])),
@@ -649,6 +651,7 @@ mod tests {
             ref_update_tx: ref_tx,
             task_event_tx: task_tx,
             graphql_schema: schema,
+            task_cursor_key,
             machine_id: None,
             repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
             rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
@@ -684,6 +687,7 @@ mod tests {
             git_ipfs_walk_semaphore: Arc::new(tokio::sync::Semaphore::new(64)),
             git_ipfs_walk_per_caller:
                 crate::rate_limit::PerCallerConcurrency::with_default_max_keys(16),
+            task_read_rate_limiter: RateLimiter::new(1200, Duration::from_secs(3600)),
             git_bin: "git".to_string(),
         }
     }

crates/gitlawb-node/src/config.rs

@@ -152,6 +152,46 @@ pub struct Config {
     #[arg(long, env = "GITLAWB_AUTO_SYNC", default_value_t = false)]
     pub auto_sync: bool,
 
+    /// Periodically fetch configured INBOUND repositories from their external
+    /// HTTPS upstream. Disabled by default: enabling this also requires
+    /// `GITLAWB_ENFORCE_OWNER_PUSH=true`, so an upstream-authoritative mirror
+    /// cannot accept writes from arbitrary signed identities between fetches.
+    #[arg(long, env = "GITLAWB_UPSTREAM_MIRROR_ENABLED", default_value_t = false)]
+    pub upstream_mirror_enabled: bool,
+
+    /// Pause between complete upstream-mirror scans. A scan is sequential and
+    /// paginated; the pause begins only after the prior scan has finished, so a
+    /// slow forge cannot create overlapping worker cycles.
+    #[arg(
+        long,
+        env = "GITLAWB_UPSTREAM_MIRROR_INTERVAL_SECS",
+        default_value_t = 300,
+        value_parser = clap::value_parser!(u64).range(5..=86_400)
+    )]
+    pub upstream_mirror_interval_secs: u64,
+
+    /// Maximum wall-clock time for one external `git fetch`. The child runs in
+    /// its own process group and is fully reaped on timeout. The preflight
+    /// URL-rewrite and local-config probes run on their own short bounds and
+    /// do not consume this budget.
+    #[arg(
+        long,
+        env = "GITLAWB_UPSTREAM_MIRROR_FETCH_TIMEOUT_SECS",
+        default_value_t = 600,
+        value_parser = clap::value_parser!(u64).range(1..=GIT_SERVICE_TIMEOUT_SECS_MAX)
+    )]
+    pub upstream_mirror_fetch_timeout_secs: u64,
+
+    /// Number of durable INBOUND rows read per database page. Every page is
+    /// processed before the scan sleeps; this bounds query memory without
+    /// permanently starving repositories beyond the first page.
+    #[arg(
+        long,
+        env = "GITLAWB_UPSTREAM_MIRROR_PAGE_SIZE",
+        default_value_t = 100,
+        value_parser = clap::builder::RangedU64ValueParser::<usize>::new().range(1..=1_000)
+    )]
+    pub upstream_mirror_page_size: usize,
     /// Enable the periodic reconciliation sweep that re-derives pin/seal sets
     /// and fills durability gaps.  Defaults to true; set to false to disable
     /// the sweep even when a pin backend (IPFS/Pinata) is configured.
@@ -244,6 +284,29 @@ pub struct Config {
     #[arg(long, env = "GITLAWB_SHUTDOWN_GRACE_SECS", default_value_t = 30)]
     pub shutdown_grace_secs: u64,
 
+    /// Maximum number of creation requests (repo create, issue create, PR
+    /// create, register, fork) allowed per DID within the rate-limit window.
+    /// 0 is rejected at parse time: it would silently disable the limiter
+    /// (RateLimiter treats max_requests == 0 as allow-all), not block every
+    /// creation request.
+    #[arg(
+        long,
+        env = "GITLAWB_RATE_LIMIT_MAX",
+        default_value_t = 10,
+        value_parser = clap::value_parser!(u32).range(1..)
+    )]
+    pub rate_limit_max_requests: u32,
+
+    /// Rate-limit window size, in seconds, over which
+    /// `rate_limit_max_requests` applies. 0 is rejected at parse time: every
+    /// timestamp would prune immediately, silently disabling the limiter.
+    #[arg(
+        long,
+        env = "GITLAWB_RATE_LIMIT_WINDOW_SECS",
+        default_value_t = 3600,
+        value_parser = clap::value_parser!(u64).range(1..)
+    )]
+    pub rate_limit_window_secs: u64,
     /// Maximum wall-clock time a single served git operation (upload-pack /
     /// receive-pack through `run_git_service`) may run before it is aborted and
     /// its process group torn down, in seconds. Bounds a git that neither
@@ -773,6 +836,21 @@ pub struct Config {
         value_parser = clap::builder::RangedU64ValueParser::<u64>::new().range(0..=86_400)
     )]
     pub pin_repair_sweep_delay_secs: u64,
+
+    /// Per-client-IP rate limit for the anonymous task read routes
+    /// (`GET /api/v1/tasks`, `GET /api/v1/tasks/{id}`), in requests per hour.
+    /// Both are publicly reachable (`optional_signature`). `GET /api/v1/tasks`
+    /// runs `collect_visible_tasks` and returns a visibility-filtered page;
+    /// `GET /api/v1/tasks/{id}` runs `get_visible_task` and returns an opaque 404
+    /// when the task is hidden or missing. These reads can require task,
+    /// repository, and visibility-rule queries even when no task is returned,
+    /// so the brake bounds the cost of anonymous probes. Keyed on the resolved
+    /// client IP via `GITLAWB_TRUSTED_PROXY`. `0` disables. Default: 1200 (a list page
+    /// followed by per-task reads is a normal client pattern, so this sits above
+    /// the `/ipfs` budget). GraphQL `tasks` / `task` queries, including WebSocket
+    /// operations, share this per-IP budget with REST task reads.
+    #[arg(long, env = "GITLAWB_TASK_READ_RATE_LIMIT", default_value_t = 1200)]
+    pub task_read_rate_limit: usize,
 }
 
 impl Config {
@@ -843,6 +921,13 @@ impl Config {
                 Self::DB_LOCK_POOL_NON_PUSH_HEADROOM
             ));
         }
+        if self.upstream_mirror_enabled && !self.enforce_owner_push {
+            return Err(
+                "GITLAWB_UPSTREAM_MIRROR_ENABLED requires GITLAWB_ENFORCE_OWNER_PUSH=true: \
+                 an upstream-authoritative mirror must not accept writes from arbitrary signed identities"
+                    .to_string(),
+            );
+        }
         let main_floor = Self::DB_POOL_APP_HEADROOM as u64;
         if (self.db_max_connections as u64) < main_floor {
             return Err(format!(
@@ -1042,6 +1127,51 @@ pub(crate) fn validate_web_url(raw: &str) -> Result<(), String> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn upstream_mirror_worker_is_default_off_and_bounded() {
+        let mut config = Config::parse_from(["gitlawb-node"]);
+        assert!(!config.upstream_mirror_enabled);
+        assert_eq!(config.upstream_mirror_interval_secs, 300);
+        assert_eq!(config.upstream_mirror_fetch_timeout_secs, 600);
+        assert_eq!(config.upstream_mirror_page_size, 100);
+        config.validate().unwrap();
+
+        assert!(
+            Config::try_parse_from(["gitlawb-node", "--upstream-mirror-interval-secs", "4",])
+                .is_err()
+        );
+        assert!(
+            Config::try_parse_from(["gitlawb-node", "--upstream-mirror-page-size", "0",]).is_err()
+        );
+    }
+
+    #[test]
+    fn upstream_mirror_worker_requires_owner_push_enforcement() {
+        // enforce_owner_push defaults true on main, so the only unsafe shape
+        // is the explicit opt-out: mirror enabled plus enforcement refused.
+        let mut unsafe_config = Config::parse_from([
+            "gitlawb-node",
+            "--upstream-mirror-enabled",
+            "--enforce-owner-push",
+            "false",
+        ]);
+        let error = unsafe_config.validate().unwrap_err();
+        assert!(error.contains("GITLAWB_ENFORCE_OWNER_PUSH=true"));
+
+        Config::parse_from([
+            "gitlawb-node",
+            "--upstream-mirror-enabled",
+            "--enforce-owner-push",
+            "true",
+        ])
+        .validate()
+        .unwrap();
+        // The default posture alone also validates: enforcement is on.
+        Config::parse_from(["gitlawb-node", "--upstream-mirror-enabled"])
+            .validate()
+            .unwrap();
+    }
+
     #[test]
     fn lock_pool_size_defaults_to_40_and_rejects_zero() {
         assert_eq!(

crates/gitlawb-node/src/error.rs

@@ -12,6 +12,9 @@ pub enum AppError {
     #[error("repo already exists: {0}")]
     RepoExists(String),
 
+    #[error("conflict: {0}")]
+    Conflict(String),
+
     #[error("not found: {0}")]
     NotFound(String),
 
@@ -189,6 +192,7 @@ impl IntoResponse for AppError {
                 "repo_exists",
                 format!("repository '{r}' already exists"),
             ),
+            AppError::Conflict(msg) => (StatusCode::CONFLICT, "conflict", msg.clone()),
             AppError::NotFound(msg) => (StatusCode::NOT_FOUND, "not_found", msg.clone()),
             AppError::Unauthorized(msg) => (StatusCode::UNAUTHORIZED, "not_an_agent", msg.clone()),
             AppError::Forbidden(msg) => (StatusCode::FORBIDDEN, "forbidden", msg.clone()),
@@ -321,6 +325,16 @@ pub type Result<T> = std::result::Result<T, AppError>;
 mod tests {
     use super::*;
 
+    #[test]
+    fn conflict_maps_to_409() {
+        assert_eq!(
+            AppError::Conflict("task not claimable".into())
+                .into_response()
+                .status(),
+            StatusCode::CONFLICT
+        );
+    }
+
     #[test]
     fn timeout_maps_to_504_distinct_from_git_500() {
         assert_eq!(

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

@@ -43,6 +43,53 @@ const WALK_TIMEOUT: Duration = Duration::from_secs(600);
 #[cfg(unix)]
 const WATCHDOG_TERM_GRACE: Duration = Duration::from_secs(1);
 
+/// Build a Git child command, optionally with an isolated environment for an
+/// owner-selected HTTPS destination. The isolated form intentionally drops
+/// ambient proxy, credential-helper, askpass, and Git config environment that
+/// could otherwise forward node-operator secrets to the selected upstream.
+fn git_child_command(
+    git_bin: &str,
+    args: &[&str],
+    repo_path: &Path,
+    isolated_https: bool,
+    isolated_config: &[(&str, &str)],
+) -> std::process::Command {
+    let mut command = std::process::Command::new(git_bin);
+    command
+        .args(args)
+        .current_dir(repo_path)
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
+    if isolated_https {
+        let path = std::env::var_os("PATH");
+        command.env_clear();
+        if let Some(path) = path {
+            command.env("PATH", path);
+        }
+        command
+            .env("GIT_CONFIG_NOSYSTEM", "1")
+            .env(
+                "GIT_CONFIG_GLOBAL",
+                if cfg!(windows) { "NUL" } else { "/dev/null" },
+            )
+            .env("GIT_TERMINAL_PROMPT", "0")
+            .env("GIT_ASKPASS", "false")
+            .env("SSH_ASKPASS", "false")
+            .env("LANG", "C")
+            .env("LC_ALL", "C");
+        if !isolated_config.is_empty() {
+            command.env("GIT_CONFIG_COUNT", isolated_config.len().to_string());
+            for (index, (key, value)) in isolated_config.iter().enumerate() {
+                command
+                    .env(format!("GIT_CONFIG_KEY_{index}"), key)
+                    .env(format!("GIT_CONFIG_VALUE_{index}"), value);
+            }
+        }
+    }
+    command
+}
+
 /// Run one git child under a shared `deadline` with process-group teardown,
 /// BLOCKING, and return its stdout. The child runs in its own process group; a
 /// watchdog thread SIGTERMs (lets git clean up its `*.lock` files), then SIGKILLs,
@@ -81,24 +128,22 @@ fn child_terminated_without_reaping(pid: i32) -> bool {
 }
 
 #[cfg(unix)]
-pub(crate) fn run_bounded_git_raw(
+fn run_bounded_git_raw_impl(
     git_bin: &str,
     args: &[&str],
     repo_path: &Path,
     stdin_bytes: &[u8],
     deadline: Instant,
+    isolated_https: bool,
+    isolated_config: &[(&str, &str)],
 ) -> Result<(std::process::ExitStatus, Vec<u8>, Vec<u8>)> {
     use std::io::{Read, Write};
     use std::os::unix::process::CommandExt;
     use std::sync::mpsc::RecvTimeoutError;
 
     let label = args.first().copied().unwrap_or("git");
-    let mut child = std::process::Command::new(git_bin)
-        .args(args)
-        .current_dir(repo_path)
-        .stdin(Stdio::piped())
-        .stdout(Stdio::piped())
-        .stderr(Stdio::piped())
+    let mut command = git_child_command(git_bin, args, repo_path, isolated_https, isolated_config);
+    let mut child = command
         .process_group(0)
         .spawn()
         .with_context(|| format!("failed to spawn git {label}"))?;
@@ -208,6 +253,51 @@ pub(crate) fn run_bounded_git_raw(
     Ok((status, out, err))
 }
 
+pub(crate) fn run_bounded_git_raw(
+    git_bin: &str,
+    args: &[&str],
+    repo_path: &Path,
+    stdin_bytes: &[u8],
+    deadline: Instant,
+) -> Result<(std::process::ExitStatus, Vec<u8>, Vec<u8>)> {
+    run_bounded_git_raw_impl(git_bin, args, repo_path, stdin_bytes, deadline, false, &[])
+}
+
+/// Run a bounded Git child without inheriting ambient Git/proxy/credential
+/// configuration. Reserved for owner-selected external HTTPS destinations;
+/// ordinary local Git operations retain their existing environment.
+pub(crate) fn run_bounded_git_raw_isolated_https(
+    git_bin: &str,
+    args: &[&str],
+    repo_path: &Path,
+    stdin_bytes: &[u8],
+    deadline: Instant,
+) -> Result<(std::process::ExitStatus, Vec<u8>, Vec<u8>)> {
+    run_bounded_git_raw_impl(git_bin, args, repo_path, stdin_bytes, deadline, true, &[])
+}
+
+/// Isolated HTTPS runner with command-scope Git configuration supplied as
+/// separate key/value environment entries. Unlike `git -c key=value`, this
+/// preserves valid URL subsection keys whose path contains `=`.
+pub(crate) fn run_bounded_git_raw_isolated_https_with_config(
+    git_bin: &str,
+    args: &[&str],
+    repo_path: &Path,
+    stdin_bytes: &[u8],
+    deadline: Instant,
+    config: &[(&str, &str)],
+) -> Result<(std::process::ExitStatus, Vec<u8>, Vec<u8>)> {
+    run_bounded_git_raw_impl(
+        git_bin,
+        args,
+        repo_path,
+        stdin_bytes,
+        deadline,
+        true,
+        config,
+    )
+}
+
 /// Bounded git returning only stdout, `bail!`ing on any nonzero exit. The thin
 /// wrapper the walk callers use. Probes that must distinguish exit classes —
 /// `git cat-file` absence vs an object-store access failure — call
@@ -241,23 +331,20 @@ pub(crate) fn run_bounded_git(
 /// the Unix version's signature and result semantics so every caller compiles on all
 /// targets (#174).
 #[cfg(not(unix))]
-pub(crate) fn run_bounded_git_raw(
+fn run_bounded_git_raw_impl(
     git_bin: &str,
     args: &[&str],
     repo_path: &Path,
     stdin_bytes: &[u8],
     deadline: Instant,
+    isolated_https: bool,
+    isolated_config: &[(&str, &str)],
 ) -> Result<(std::process::ExitStatus, Vec<u8>, Vec<u8>)> {
     use std::io::{Read, Write};
     use std::sync::mpsc::RecvTimeoutError;
 
     let label = args.first().copied().unwrap_or("git");
-    let mut child = std::process::Command::new(git_bin)
-        .args(args)
-        .current_dir(repo_path)
-        .stdin(Stdio::piped())
-        .stdout(Stdio::piped())
-        .stderr(Stdio::piped())
+    let mut child = git_child_command(git_bin, args, repo_path, isolated_https, isolated_config)
         .spawn()
         .with_context(|| format!("failed to spawn git {label}"))?;
 

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

@@ -76,6 +76,7 @@ pub(crate) fn graphql_app_err(e: crate::error::AppError) -> async_graphql::Error
         // Curated client-safe variants — `Display` is intentional API text.
         safe @ (crate::error::AppError::RepoNotFound(_)
         | crate::error::AppError::RepoExists(_)
+        | crate::error::AppError::Conflict(_)
         | crate::error::AppError::NotFound(_)
         | crate::error::AppError::Unauthorized(_)
         | crate::error::AppError::Forbidden(_)
@@ -92,15 +93,81 @@ pub(crate) fn graphql_app_err(e: crate::error::AppError) -> async_graphql::Error
     }
 }
 
+/// Classify a failed `claim_task` for GraphQL exactly as REST's claim handler
+/// classifies it: a lost race is a client-safe conflict, a real sqlx fault
+/// stays opaque. Lives here rather than at the three call sites so the
+/// classification cannot drift between transports (#327 review), and so
+/// `every_graphql_map_err_uses_opaque_helpers` can keep whitelisting by name.
+pub(crate) fn graphql_claim_conflict(e: anyhow::Error) -> async_graphql::Error {
+    graphql_app_err(crate::api::tasks::task_write_conflict(
+        e,
+        "task not claimable: not found or already claimed",
+    ))
+}
+
+/// The `finish_task` half of [`graphql_claim_conflict`], covering both
+/// `completeTask` and `failTask`.
+pub(crate) fn graphql_finish_conflict(e: anyhow::Error) -> async_graphql::Error {
+    graphql_app_err(crate::api::tasks::task_write_conflict(
+        e,
+        "task not found or not in claimed state",
+    ))
+}
+
+pub struct TaskReadBrakeExtension;
+
+impl async_graphql::extensions::ExtensionFactory for TaskReadBrakeExtension {
+    fn create(&self) -> Arc<dyn async_graphql::extensions::Extension> {
+        Arc::new(TaskReadBrakeExtensionImpl)
+    }
+}
+
+use async_graphql::async_trait::async_trait;
+
+struct TaskReadBrakeExtensionImpl;
+
+#[async_trait]
+impl async_graphql::extensions::Extension for TaskReadBrakeExtensionImpl {
+    async fn prepare_request(
+        &self,
+        ctx: &async_graphql::extensions::ExtensionContext<'_>,
+        mut request: async_graphql::Request,
+        next: async_graphql::extensions::NextPrepareRequest<'_>,
+    ) -> async_graphql::ServerResult<async_graphql::Request> {
+        if let Some(session_brake) = ctx
+            .session_data
+            .get(&std::any::TypeId::of::<crate::rate_limit::TaskReadBrake>())
+            .and_then(|d| d.downcast_ref::<crate::rate_limit::TaskReadBrake>())
+        {
+            request = request.data(crate::rate_limit::TaskReadBrake {
+                limiter: session_brake.limiter.clone(),
+                key: session_brake.key.clone(),
+                request_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+            });
+        }
+        if let Some(did) = ctx
+            .session_data
+            .get(&std::any::TypeId::of::<crate::auth::AuthenticatedDid>())
+            .and_then(|d| d.downcast_ref::<crate::auth::AuthenticatedDid>())
+        {
+            request = request.data(did.clone());
+        }
+        next.run(ctx, request).await
+    }
+}
+
 pub fn build_schema(
     db: Arc<Db>,
     ref_update_tx: tokio::sync::broadcast::Sender<RefUpdateBroadcast>,
     task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
+    task_cursor_key: crate::api::task_cursor::TaskCursorKey,
 ) -> GitlawbSchema {
     apply_query_limits(Schema::build(QueryRoot, MutationRoot, SubscriptionRoot))
         .data(db)
         .data(ref_update_tx)
         .data(task_event_tx)
+        .data(task_cursor_key)
+        .extension(TaskReadBrakeExtension)
         .finish()
 }
 
@@ -192,6 +259,7 @@ mod tests {
             db,
             tokio::sync::broadcast::channel(1).0,
             tokio::sync::broadcast::channel(1).0,
+            crate::api::task_cursor::TaskCursorKey::derive(&[42u8; 32]),
         )
     }
 
@@ -217,6 +285,7 @@ mod tests {
             Arc::new(Db::for_testing(pool)),
             ref_tx.clone(),
             task_tx.clone(),
+            crate::api::task_cursor::TaskCursorKey::derive(&[42u8; 32]),
         );
         for count in [7, 8] {
             let fields = (0..count)
@@ -313,7 +382,10 @@ mod tests {
                 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(),
+                    db.list_tasks_keyset(Some("pending"), None, 8, None)
+                        .await
+                        .unwrap()
+                        .len(),
                     8
                 );
                 continue;
@@ -330,7 +402,7 @@ mod tests {
                 7
             );
             assert_eq!(
-                db.list_tasks(Some("claimed"), Some(caller), 8)
+                db.list_tasks_keyset(Some("claimed"), Some(caller), 8, None)
                     .await
                     .unwrap()
                     .len(),
@@ -352,7 +424,7 @@ mod tests {
         // 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),
+            ("tasks", "items { id }", MAX_VISIBLE_TASK_PAGE_SIZE),
             (
                 "reposPage",
                 "nodes { name }",
@@ -595,7 +667,12 @@ mod tests {
         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);
+        let schema = build_schema(
+            Arc::new(Db::for_testing(pool)),
+            ref_tx,
+            task_tx,
+            crate::api::task_cursor::TaskCursorKey::derive(&[42u8; 32]),
+        );
 
         // refUpdates(limit: MAX_VISIBLE_REF_UPDATES):
         // Single-field selection has child_complexity 1 -> cost 100 + 200 + 1 = 301 <= 400.
@@ -649,12 +726,12 @@ mod tests {
         );
 
         // 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.
+        // Selection items { id } has child_complexity 2, so the cost is
+        // 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 2 <= 400.
         // It passes validation and reaches the resolver.
         let single_task = schema
             .execute(format!(
-                "{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id }} }}"
+                "{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ items {{ id }} }} }}"
             ))
             .await;
         assert!(
@@ -674,14 +751,14 @@ mod tests {
             single_task.errors
         );
 
-        // Two-field selection has child_complexity 2, so the cost is
-        // 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 2 <= 400.
+        // Selection items { id status } has child_complexity 3, so the cost is
+        // 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 3 <= 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 }} }}"
+                "{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ items {{ id status }} }} }}"
             ))
             .await;
         assert!(
@@ -843,8 +920,10 @@ mod tests {
     }
 
     /// 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).
+    /// through one of the curated helpers in this module, or discard the error
+    /// (`|_|`). Same source-scrape pattern as `api::authz_guard` (#255
+    /// review). The list is deliberately a whitelist of names rather than a
+    /// prefix match, so adding a new mapper is a decision a reviewer sees.
     #[test]
     fn every_graphql_map_err_uses_opaque_helpers() {
         for (file, src) in [
@@ -859,6 +938,8 @@ mod tests {
                 let after = code[idx + ".map_err(".len()..].trim_start();
                 let ok = after.starts_with("crate::graphql::graphql_db_err")
                     || after.starts_with("crate::graphql::graphql_app_err")
+                    || after.starts_with("crate::graphql::graphql_claim_conflict")
+                    || after.starts_with("crate::graphql::graphql_finish_conflict")
                     || after.starts_with("|_|")
                     || after.starts_with("|_ ");
                 assert!(

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

@@ -106,25 +106,37 @@ impl MutationRoot {
         let assignee_did = caller.to_string();
         let db = ctx.data_unchecked::<Arc<Db>>();
         let tx = ctx.data_unchecked::<tokio::sync::broadcast::Sender<TaskEventBroadcast>>();
+        crate::api::tasks::get_claimable_task(db, &id, caller)
+            .await
+            .map_err(crate::graphql::graphql_app_err)?
+            .ok_or_else(|| async_graphql::Error::new("task not found"))?;
+        // Same classification REST's claim handler applies: a claim race is a
+        // client-safe conflict, a genuine sqlx failure stays opaque. Routing
+        // both transports through `task_write_conflict` is what stops a normal
+        // race from reaching a GraphQL caller as a generic database error
+        // (#327 review).
         let task = db
             .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.
+            .map_err(crate::graphql::graphql_claim_conflict)?;
         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,
-            at: Utc::now().to_rfc3339(),
-        });
+        crate::api::tasks::announce_task_event(
+            db,
+            tx,
+            TaskEventBroadcast {
+                task_id: id,
+                old_status: "pending".to_string(),
+                new_status: "claimed".to_string(),
+                by_did,
+                at: Utc::now().to_rfc3339(),
+            },
+        )
+        .await;
         Ok(AgentTaskType::from(task))
     }
 
@@ -145,12 +157,12 @@ impl MutationRoot {
         let by_did = caller.to_string();
         let db = ctx.data_unchecked::<Arc<Db>>();
         let tx = ctx.data_unchecked::<tokio::sync::broadcast::Sender<TaskEventBroadcast>>();
-        // Authorize the actor: binding by_did to the signer is necessary but not
-        // sufficient — only the task's assignee may finish it.
-        let existing = db
-            .get_task(&id)
+        // Authorize the actor: the task must be visible to the caller (returning
+        // not found for invisible tasks so existence is not leaked), and only
+        // the task's assignee may finish it.
+        let existing = crate::api::tasks::get_visible_task(db, &id, Some(caller))
             .await
-            .map_err(crate::graphql::graphql_db_err)?
+            .map_err(crate::graphql::graphql_app_err)?
             .ok_or_else(|| async_graphql::Error::new("task not found"))?;
         if !crate::api::did_matches(
             caller,
@@ -163,20 +175,25 @@ impl MutationRoot {
         let task = db
             .finish_task(&id, "completed", input.result.as_deref(), caller)
             .await
-            .map_err(crate::graphql::graphql_db_err)?;
+            .map_err(crate::graphql::graphql_finish_conflict)?;
         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(),
-            new_status: "completed".to_string(),
-            by_did,
-            at: Utc::now().to_rfc3339(),
-        });
+        crate::api::tasks::announce_task_event(
+            db,
+            tx,
+            TaskEventBroadcast {
+                task_id: id,
+                old_status: "claimed".to_string(),
+                new_status: "completed".to_string(),
+                by_did,
+                at: Utc::now().to_rfc3339(),
+            },
+        )
+        .await;
         Ok(AgentTaskType::from(task))
     }
 
@@ -197,11 +214,12 @@ impl MutationRoot {
         let by_did = caller.to_string();
         let db = ctx.data_unchecked::<Arc<Db>>();
         let tx = ctx.data_unchecked::<tokio::sync::broadcast::Sender<TaskEventBroadcast>>();
-        // Authorize the actor: only the task's assignee may fail it.
-        let existing = db
-            .get_task(&id)
+        // Authorize the actor: the task must be visible to the caller (returning
+        // not found for invisible tasks so existence is not leaked), and only
+        // the task's assignee may fail it.
+        let existing = crate::api::tasks::get_visible_task(db, &id, Some(caller))
             .await
-            .map_err(crate::graphql::graphql_db_err)?
+            .map_err(crate::graphql::graphql_app_err)?
             .ok_or_else(|| async_graphql::Error::new("task not found"))?;
         if !crate::api::did_matches(
             caller,
@@ -215,20 +233,25 @@ impl MutationRoot {
         let task = db
             .finish_task(&id, "failed", Some(&reason), caller)
             .await
-            .map_err(crate::graphql::graphql_db_err)?;
+            .map_err(crate::graphql::graphql_finish_conflict)?;
         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(),
-            new_status: "failed".to_string(),
-            by_did,
-            at: Utc::now().to_rfc3339(),
-        });
+        crate::api::tasks::announce_task_event(
+            db,
+            tx,
+            TaskEventBroadcast {
+                task_id: id,
+                old_status: "claimed".to_string(),
+                new_status: "failed".to_string(),
+                by_did,
+                at: Utc::now().to_rfc3339(),
+            },
+        )
+        .await;
         Ok(AgentTaskType::from(task))
     }
 }
@@ -326,7 +349,7 @@ mod tests {
             );
             if *name == "claimTask" {
                 assert!(
-                    errs.contains("task not claimable"),
+                    errs.contains("task not found"),
                     "claim race / missing task must keep its business message: {errs}"
                 );
                 assert!(
@@ -370,6 +393,210 @@ mod tests {
         );
     }
 
+    /// #327 review: `claimTask`/`completeTask`/`failTask` called
+    /// `graphql_db_err` directly on db-layer business failures, so an ordinary
+    /// claim race or stale finish reached GraphQL clients as a generic
+    /// database error while REST clients got an actionable conflict. All three
+    /// now route through the shared `task_write_conflict` classifier.
+    ///
+    /// Deleting either `map_err` from a mutation turns this red: the message
+    /// becomes the raw anyhow text instead of the fixed conflict form.
+    #[sqlx::test]
+    async fn task_write_races_surface_as_conflicts_not_db_errors(pool: PgPool) {
+        let state = crate::test_support::test_state(pool).await;
+        let schema = state.graphql_schema.as_ref();
+        let assignee = "did:key:zGQLRACEASSIGNEEAAAAAAAAAAAAAAAAAAAAAAA";
+        let rival = "did:key:zGQLRACERIVALBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let delegator = "did:key:zGQLRACEDELEGATORCCCCCCCCCCCCCCCCCCCCC";
+        let now = chrono::Utc::now().to_rfc3339();
+
+        state
+            .db
+            .create_repo(&crate::db::RepoRecord {
+                id: "race-repo".into(),
+                name: "race".into(),
+                owner_did: delegator.into(),
+                description: None,
+                is_public: true,
+                default_branch: "main".into(),
+                created_at: chrono::Utc::now(),
+                updated_at: chrono::Utc::now(),
+                disk_path: "/tmp/race-repo".into(),
+                forked_from: None,
+                machine_id: None,
+            })
+            .await
+            .unwrap();
+        state
+            .db
+            .create_task(&crate::db::AgentTask {
+                id: "race-task".into(),
+                repo_id: Some("race-repo".into()),
+                kind: "build".into(),
+                status: "pending".into(),
+                delegator_did: delegator.into(),
+                assignee_did: None,
+                capability: "repo:write".into(),
+                ucan_token: None,
+                payload: None,
+                result: None,
+                created_at: now.clone(),
+                updated_at: now,
+                deadline: None,
+            })
+            .await
+            .unwrap();
+
+        // The assignee wins the claim.
+        let claim = |did: &str| {
+            format!(r#"mutation {{ claimTask(id: "race-task", assigneeDid: "{did}") {{ id }} }}"#)
+        };
+        let resp = schema
+            .execute(Request::new(claim(assignee)).data(AuthenticatedDid(assignee.into())))
+            .await;
+        assert!(resp.errors.is_empty(), "first claim: {}", errors(&resp));
+
+        // The rival loses the race: ineligible caller receives opaque not-found.
+        let resp = schema
+            .execute(Request::new(claim(rival)).data(AuthenticatedDid(rival.into())))
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.contains("task not found"),
+            "ineligible caller after task is claimed must receive opaque not-found: {errs}"
+        );
+        assert!(
+            !errs.contains(crate::graphql::GRAPHQL_DB_ERROR_MESSAGE),
+            "ineligible claim is not a database failure: {errs}"
+        );
+
+        // A second claim that reaches the conditional SQL write (e.g. assignee claiming
+        // again on an already-claimed task) exercises the conflict mapper:
+        let resp = schema
+            .execute(Request::new(claim(assignee)).data(AuthenticatedDid(assignee.into())))
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.contains("task not claimable: not found or already claimed"),
+            "a lost claim race must render the fixed conflict message: {errs}"
+        );
+        assert!(
+            !errs.contains(crate::graphql::GRAPHQL_DB_ERROR_MESSAGE),
+            "a claim race is not a database failure: {errs}"
+        );
+
+        // A stale finish on an already-completed task is the same class.
+        let complete = format!(
+            r#"mutation {{ completeTask(id: "race-task", byDid: "{assignee}", input: {{ result: "ok" }}) {{ id }} }}"#
+        );
+        let resp = schema
+            .execute(Request::new(&complete).data(AuthenticatedDid(assignee.into())))
+            .await;
+        assert!(resp.errors.is_empty(), "first complete: {}", errors(&resp));
+
+        let resp = schema
+            .execute(Request::new(&complete).data(AuthenticatedDid(assignee.into())))
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.contains("task not found or not in claimed state"),
+            "a stale complete must render the fixed conflict message: {errs}"
+        );
+        assert!(
+            !errs.contains(crate::graphql::GRAPHQL_DB_ERROR_MESSAGE),
+            "a stale complete is not a database failure: {errs}"
+        );
+
+        let fail = format!(
+            r#"mutation {{ failTask(id: "race-task", byDid: "{assignee}", input: {{ reason: "nope" }}) {{ id }} }}"#
+        );
+        let resp = schema
+            .execute(Request::new(&fail).data(AuthenticatedDid(assignee.into())))
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.contains("task not found or not in claimed state"),
+            "a stale fail must render the fixed conflict message: {errs}"
+        );
+        assert!(
+            !errs.contains(crate::graphql::GRAPHQL_DB_ERROR_MESSAGE),
+            "a stale fail is not a database failure: {errs}"
+        );
+    }
+
+    /// The other half of the same classification: a genuine SQL fault on the
+    /// write itself must stay opaque. Without this, routing conflicts through
+    /// `task_write_conflict` could be "fixed" by making every db failure a
+    /// client-visible conflict message.
+    #[sqlx::test]
+    async fn task_write_sql_faults_stay_opaque(pool: PgPool) {
+        let state = crate::test_support::test_state(pool.clone()).await;
+        let assignee = "did:key:zGQLOPAQUEASSIGNEEAAAAAAAAAAAAAAAAAAAAA";
+        let delegator = "did:key:zGQLOPAQUEDELEGATORBBBBBBBBBBBBBBBBBBB";
+        let now = chrono::Utc::now().to_rfc3339();
+        state
+            .db
+            .create_task(&crate::db::AgentTask {
+                id: "opaque-task".into(),
+                repo_id: None,
+                kind: "build".into(),
+                status: "pending".into(),
+                delegator_did: delegator.into(),
+                assignee_did: Some(assignee.into()),
+                capability: "repo:write".into(),
+                ucan_token: None,
+                payload: None,
+                result: None,
+                created_at: now.clone(),
+                updated_at: now,
+                deadline: None,
+            })
+            .await
+            .unwrap();
+
+        // Fault the write itself, not the schema. Dropping a column the
+        // visibility pre-check also SELECTs would fail in `get_visible_task`
+        // and never reach `graphql_claim_conflict`, so this test would pass
+        // against a claim mapper that leaks (#327 review). A BEFORE UPDATE
+        // trigger keeps every read valid and faults only inside
+        // `Db::claim_task`.
+        sqlx::raw_sql(
+            "CREATE FUNCTION gl_test_fault_task_update() RETURNS trigger
+             LANGUAGE plpgsql AS $fn$
+             BEGIN RAISE EXCEPTION 'gl_test_forced_update_fault'; END;
+             $fn$;
+             CREATE TRIGGER gl_test_fault_task_update
+             BEFORE UPDATE ON agent_tasks
+             FOR EACH ROW EXECUTE FUNCTION gl_test_fault_task_update();",
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        let resp = state
+            .graphql_schema
+            .execute(
+                Request::new(format!(
+                    r#"mutation {{ claimTask(id: "opaque-task", assigneeDid: "{assignee}") {{ id }} }}"#
+                ))
+                .data(AuthenticatedDid(assignee.into())),
+            )
+            .await;
+        let errs = errors(&resp);
+        assert!(
+            errs.contains(crate::graphql::GRAPHQL_DB_ERROR_MESSAGE),
+            "a real sqlx fault on the write must stay opaque: {errs}"
+        );
+        assert!(
+            !errs.contains("task not claimable"),
+            "a write-time sqlx fault must not be reclassified as a claim race: {errs}"
+        );
+        assert!(
+            !errs.contains("gl_test_forced_update_fault") && !errs.contains("trigger"),
+            "database text leaked through the conflict mapper: {errs}"
+        );
+    }
+
     /// Adversarial-review GATE-1 (GraphQL): completing a task requires being its
     /// assignee, not merely signing as the by_did you pass. A signer who is not
     /// the assignee is rejected even though the by_did binding passes; the
@@ -392,7 +619,7 @@ mod tests {
             payload: None,
             result: None,
             created_at: now.clone(),
-            updated_at: now,
+            updated_at: now.clone(),
             deadline: None,
         };
         state.db.create_task(&task).await.expect("seed task");
@@ -409,14 +636,69 @@ mod tests {
             )
         };
 
-        // Stranger signs as themselves and passes byDid=self (so the signer
-        // binding passes), but is not the assignee → rejected by authorization.
+        // Stranger signs as themselves on a repo-less task they cannot see:
+        // invisible task returns "task not found" so existence is not leaked.
         let resp = schema
             .execute(Request::new(q(stranger)).data(AuthenticatedDid(stranger.into())))
             .await;
+        assert!(
+            errors(&resp).contains("task not found"),
+            "an invisible task must return not found, got: {}",
+            errors(&resp)
+        );
+
+        // Seed a task on a public repo that stranger CAN see, but is not assignee of:
+        let pub_repo = crate::db::RepoRecord {
+            id: "pub-r".into(),
+            name: "pub-r".into(),
+            owner_did: "did:key:zGQLDELEGATORCCCCCCCCCCCCCCCCCCCCCCCCCC".into(),
+            description: None,
+            is_public: true,
+            default_branch: "main".into(),
+            created_at: chrono::Utc::now(),
+            updated_at: chrono::Utc::now(),
+            disk_path: "/tmp/pub-r".into(),
+            forked_from: None,
+            machine_id: None,
+        };
+        state.db.create_repo(&pub_repo).await.expect("create repo");
+        let pub_task = crate::db::AgentTask {
+            id: "task-pub".into(),
+            repo_id: Some("pub-r".into()),
+            kind: "build".into(),
+            status: "pending".into(),
+            delegator_did: "did:key:zGQLDELEGATORCCCCCCCCCCCCCCCCCCCCCCCCCC".into(),
+            assignee_did: None,
+            capability: "repo:write".into(),
+            ucan_token: None,
+            payload: None,
+            result: None,
+            created_at: now.clone(),
+            updated_at: now,
+            deadline: None,
+        };
+        state
+            .db
+            .create_task(&pub_task)
+            .await
+            .expect("seed pub task");
+        state
+            .db
+            .claim_task("task-pub", assignee)
+            .await
+            .expect("claim pub task");
+
+        let q_pub = |actor: &str| {
+            format!(
+                r#"mutation {{ completeTask(id: "task-pub", byDid: "{actor}", input: {{}}) {{ id status }} }}"#
+            )
+        };
+        let resp = schema
+            .execute(Request::new(q_pub(stranger)).data(AuthenticatedDid(stranger.into())))
+            .await;
         assert!(
             errors(&resp).contains("assignee"),
-            "a non-assignee signer must be rejected: {}",
+            "a non-assignee signer on a visible task must be rejected: {}",
             errors(&resp)
         );
 
@@ -482,7 +764,7 @@ mod tests {
         // repo id afterwards.
         let stored = state
             .db
-            .list_tasks(None, None, 50)
+            .list_tasks_keyset(None, None, 50, None)
             .await
             .expect("list tasks");
         assert!(
@@ -644,9 +926,11 @@ mod tests {
             .execute(Request::new(q(thief)).data(AuthenticatedDid(thief.into())))
             .await;
         let errs = errors(&resp);
+        // The claim gate hides existence: an ineligible claimer gets the same
+        // not-found as a missing task, not a reserved deny that would leak it.
         assert!(
-            errs.contains("reserved"),
-            "stranger claim must surface reserved deny distinctly: {errs}"
+            errs.contains("task not found"),
+            "stranger claim must read as not-found, got: {errs}"
         );
         assert!(
             !errs.contains("gql-ucan-secret"),

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

@@ -5,7 +5,7 @@ use std::sync::Arc;
 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, RepoPageType, RepoType};
+use super::types::{AgentTaskReadType, RefUpdateType, RepoPageType, RepoType, TaskPageType};
 
 fn repo_type(repo: RepoRecord) -> RepoType {
     RepoType {
@@ -45,6 +45,23 @@ fn parse_repo_cursor(cursor: &str) -> Result<(String, String)> {
 
 pub struct QueryRoot;
 
+/// Debit the per-IP task-read brake for one task field, mirroring the
+/// `rate_limit_by_ip` layer on `task_read_routes`. Both surfaces run the same
+/// #268 visibility gate and pay the same queries before answering, so an
+/// anonymous prober must not get an unbraked lane by asking over GraphQL
+/// (#327 review).
+///
+/// A schema built without the brake (unit tests, subscriptions) is not braked,
+/// exactly as `rate_limit_by_ip` is a no-op without its extension.
+async fn task_read_brake(ctx: &Context<'_>, field: &str) -> Result<()> {
+    match ctx.data::<crate::rate_limit::TaskReadBrake>() {
+        Ok(brake) if !brake.check(field).await => Err(async_graphql::Error::new(
+            crate::rate_limit::RATE_LIMIT_MESSAGE,
+        )),
+        _ => Ok(()),
+    }
+}
+
 #[Object]
 impl QueryRoot {
     // DB-backed roots carry a base cost so aliases consume the request budget
@@ -217,30 +234,74 @@ impl QueryRoot {
         assignee_did: Option<String>,
         #[graphql(
             default = 50,
-            desc = "Requests above the documented maximum are clamped to it (no error). Negative values clamp to 0."
+            desc = "Max 200; larger requests are clamped to 200 (no error). Negative values clamp to 0. Use `nextCursor` to read past the clamp."
         )]
         limit: i64,
-    ) -> Result<Vec<AgentTaskType>> {
+        #[graphql(
+            desc = "Opaque continuation token from a previous page's `nextCursor`. Must be presented with the same `status`/`assigneeDid` filter that issued it."
+        )]
+        cursor: Option<String>,
+    ) -> Result<TaskPageType> {
+        use crate::api::task_cursor::{self, TaskCursorKey, TaskFilter};
+
+        task_read_brake(ctx, "tasks").await?;
         let db = ctx.data_unchecked::<Arc<Db>>();
-        // 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, MAX_VISIBLE_TASK_PAGE_SIZE);
-        let tasks = db
-            .list_tasks(status.as_deref(), assignee_did.as_deref(), limit)
-            .await
-            .map_err(crate::graphql::graphql_db_err)?;
-        Ok(tasks.into_iter().map(AgentTaskType::from).collect())
+        // #268: gate rows via the same collector the REST list route uses (like
+        // `ref_updates` shares `collect_visible_ref_updates` with its REST feed),
+        // so the two surfaces cannot drift. The collector clamps `limit` itself,
+        // including the negative-LIMIT case #250 called out for this resolver.
+        let caller = ctx
+            .data::<crate::auth::AuthenticatedDid>()
+            .ok()
+            .map(|d| d.0.as_str());
+        let key = ctx.data_unchecked::<TaskCursorKey>();
+        let filter = TaskFilter {
+            status: status.as_deref(),
+            assignee_did: assignee_did.as_deref(),
+        };
+        let resume = cursor
+            .as_deref()
+            .map(|token| task_cursor::decode(key, filter, caller, token))
+            .transpose()
+            .map_err(crate::graphql::graphql_app_err)?;
+        let result = crate::api::tasks::collect_visible_tasks(
+            db,
+            status.as_deref(),
+            assignee_did.as_deref(),
+            limit,
+            resume.as_ref(),
+            caller,
+        )
+        .await
+        .map_err(crate::graphql::graphql_app_err)?;
+        Ok(TaskPageType {
+            next_cursor: result
+                .next_position
+                .as_ref()
+                .map(|pos| task_cursor::encode(key, filter, caller, pos)),
+            items: result
+                .tasks
+                .into_iter()
+                .map(AgentTaskReadType::from)
+                .collect(),
+            has_more: result.has_more,
+            incomplete: result.incomplete,
+        })
     }
 
     #[graphql(complexity = "50 + child_complexity")]
-    async fn task(&self, ctx: &Context<'_>, id: String) -> Result<Option<AgentTaskType>> {
+    async fn task(&self, ctx: &Context<'_>, id: String) -> Result<Option<AgentTaskReadType>> {
+        task_read_brake(ctx, "task").await?;
         let db = ctx.data_unchecked::<Arc<Db>>();
-        let t = db
-            .get_task(&id)
+        // #268: same gate as the REST get route, via the shared helper.
+        let caller = ctx
+            .data::<crate::auth::AuthenticatedDid>()
+            .ok()
+            .map(|d| d.0.as_str());
+        let t = crate::api::tasks::get_visible_task(db, &id, caller)
             .await
-            .map_err(crate::graphql::graphql_db_err)?;
-        Ok(t.map(AgentTaskType::from))
+            .map_err(crate::graphql::graphql_app_err)?;
+        Ok(t.map(AgentTaskReadType::from))
     }
 }
 
@@ -889,10 +950,14 @@ mod tests {
         Arc::new(db)
     }
 
+    fn cursor_key() -> crate::api::task_cursor::TaskCursorKey {
+        crate::api::task_cursor::TaskCursorKey::derive(&[42u8; 32])
+    }
+
     fn schema(db: Arc<Db>) -> super::super::GitlawbSchema {
         let (ref_tx, _) = tokio::sync::broadcast::channel(16);
         let (task_tx, _) = tokio::sync::broadcast::channel(16);
-        super::super::build_schema(db, ref_tx, task_tx)
+        super::super::build_schema(db, ref_tx, task_tx, cursor_key())
     }
 
     fn repo(id: &str, owner_did: &str, name: &str, is_public: bool) -> RepoRecord {
@@ -1202,7 +1267,7 @@ mod tests {
     async fn tasks_negative_limit_clamped(pool: PgPool) {
         let db = db(pool).await;
         let schema = schema(db);
-        let resp = anon(&schema, "{ tasks(limit: -1) { id } }").await;
+        let resp = anon(&schema, "{ tasks(limit: -1) { items { id } } }").await;
         assert!(
             resp.errors.is_empty(),
             "negative limit must clamp, not fail: {:?}",
@@ -1237,12 +1302,12 @@ mod tests {
             .unwrap();
         }
         let schema = schema(db);
-        let resp = anon(&schema, "{ tasks(limit: 5000) { id } }").await;
-        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"
-        );
+        // Queried as the delegator, not anonymously: since #268 the task read
+        // surface is visibility-gated, and an anonymous caller sees none of
+        // these repo-less tasks at all. The clamp is what this test pins, so it
+        // needs a caller who can legitimately see all 201 rows.
+        let resp = authed(&schema, "{ tasks(limit: 5000) { items { id } } }", OWNER).await;
+        assert_eq!(count_tasks(&resp), 200, "limit above 200 must clamp to 200");
     }
 
     /// A limit above MAX_VISIBLE_REF_UPDATES must reach the resolver clamped,
@@ -1283,14 +1348,401 @@ mod tests {
         );
     }
 
-    fn count_tasks(resp: &async_graphql::Response) -> usize {
+    /// Seed one repo-less task carrying a `ucan_token`, so a leak on any read
+    /// surface is visible in the response body.
+    async fn seed_task(db: &Db, id: &str, delegator: &str) {
+        let now = Utc::now().to_rfc3339();
+        db.create_task(&crate::db::AgentTask {
+            id: id.into(),
+            repo_id: None,
+            kind: "build".into(),
+            status: "pending".into(),
+            delegator_did: delegator.into(),
+            assignee_did: None,
+            capability: "repo:write".into(),
+            ucan_token: Some("SECRET-UCAN-TOKEN".into()),
+            payload: None,
+            result: None,
+            created_at: now.clone(),
+            updated_at: now,
+            deadline: None,
+        })
+        .await
+        .unwrap();
+    }
+
+    /// #268: the `tasks` resolver must delegate to the gated collector, not
+    /// query the DB directly. A repo-less task belonging to someone else is
+    /// invisible to an anonymous caller. `tasks_negative_limit_clamped` cannot
+    /// catch a resolver that stops calling `collect_visible_tasks` because it
+    /// seeds no rows — this seeds one, so the gate is load-bearing here.
+    #[sqlx::test]
+    async fn tasks_repo_less_task_hidden_from_anon(pool: PgPool) {
+        let db = db(pool).await;
+        seed_task(&db, "t1", OWNER).await;
+        let schema = schema(db);
+        let resp = anon(&schema, "{ tasks { items { id } } }").await;
+        assert_eq!(
+            count_tasks(&resp),
+            0,
+            "anon must not enumerate another party's repo-less task"
+        );
+        assert!(
+            !format!("{:?}", resp.data).contains("SECRET-UCAN-TOKEN"),
+            "no ucan token may reach an anonymous caller"
+        );
+    }
+
+    /// #268 sibling for the single-task resolver: an invisible task reads as
+    /// `null`, indistinguishable from one that does not exist.
+    #[sqlx::test]
+    async fn task_by_id_is_null_for_anon(pool: PgPool) {
+        let db = db(pool).await;
+        seed_task(&db, "t1", OWNER).await;
+        let schema = schema(db);
+        let resp = anon(&schema, r#"{ task(id: "t1") { id } }"#).await;
         assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
         let async_graphql::Value::Object(obj) = &resp.data else {
             panic!("data not an object: {:?}", resp.data);
         };
-        let async_graphql::Value::List(rows) = obj.get("tasks").expect("tasks key") else {
-            panic!("tasks not a list");
+        assert_eq!(
+            obj.get("task"),
+            Some(&async_graphql::Value::Null),
+            "an invisible task must read as null, got {:?}",
+            obj.get("task")
+        );
+    }
+
+    /// #268: `ucanToken` is absent from the read type's schema entirely, so the
+    /// delegator cannot request it either. Asking for it is a validation error,
+    /// which pins the redaction at the schema level rather than per-resolver.
+    #[sqlx::test]
+    async fn task_read_schema_has_no_ucan_token_field(pool: PgPool) {
+        let db = db(pool).await;
+        seed_task(&db, "t1", OWNER).await;
+        let schema = schema(db);
+        let resp = authed(&schema, r#"{ task(id: "t1") { id ucanToken } }"#, OWNER).await;
+        assert!(
+            !resp.errors.is_empty(),
+            "ucanToken must not exist on the task read type"
+        );
+    }
+
+    #[sqlx::test]
+    async fn tasks_find_older_visible_row_behind_denied_window(pool: PgPool) {
+        let db = db(pool).await;
+        db.create_repo(&repo("public-repo", OWNER, "public", true))
+            .await
+            .unwrap();
+        let visible = crate::db::AgentTask {
+            id: "visible".into(),
+            repo_id: Some("public-repo".into()),
+            kind: "build".into(),
+            status: "pending".into(),
+            delegator_did: OWNER.into(),
+            assignee_did: None,
+            capability: "repo:write".into(),
+            ucan_token: None,
+            payload: None,
+            result: None,
+            created_at: "2026-01-01T00:00:00Z".into(),
+            updated_at: "2026-01-01T00:00:00Z".into(),
+            deadline: None,
         };
-        rows.len()
+        db.create_task(&visible).await.unwrap();
+        for i in 0..200 {
+            let mut hidden = visible.clone();
+            hidden.id = format!("hidden-{i:03}");
+            hidden.repo_id = None;
+            hidden.created_at = "2026-01-02T00:00:00Z".into();
+            hidden.updated_at = hidden.created_at.clone();
+            db.create_task(&hidden).await.unwrap();
+        }
+
+        let schema = schema(db);
+        let resp = anon(&schema, "{ tasks(limit: 1) { items { id } incomplete } }").await;
+        assert_eq!(count_tasks(&resp), 1);
+        assert!(!task_incomplete(&resp));
+        assert!(format!("{:?}", resp.data).contains("visible"));
+    }
+
+    /// The GraphQL surface must page past a denied window on server-issued
+    /// cursors alone, exactly as REST does (#327 review). Before this, the
+    /// only cursor a caller could hold named the last row they saw, so the
+    /// query stalled at `incomplete: true` forever and never reached
+    /// `past-ceiling`.
+    #[sqlx::test]
+    async fn tasks_page_past_candidate_ceiling_on_server_cursors(pool: PgPool) {
+        let db = db(pool).await;
+        db.create_repo(&repo("public-repo", OWNER, "public", true))
+            .await
+            .unwrap();
+        let visible_newer = crate::db::AgentTask {
+            id: "newer-visible".into(),
+            repo_id: Some("public-repo".into()),
+            kind: "build".into(),
+            status: "pending".into(),
+            delegator_did: OWNER.into(),
+            assignee_did: None,
+            capability: "repo:write".into(),
+            ucan_token: None,
+            payload: None,
+            result: None,
+            created_at: "2026-01-03T00:00:00Z".into(),
+            updated_at: "2026-01-03T00:00:00Z".into(),
+            deadline: None,
+        };
+        db.create_task(&visible_newer).await.unwrap();
+        for i in 0..1500 {
+            let mut hidden = visible_newer.clone();
+            hidden.id = format!("hidden-{i:04}");
+            hidden.repo_id = None;
+            hidden.created_at = "2026-01-02T00:00:00Z".into();
+            hidden.updated_at = hidden.created_at.clone();
+            db.create_task(&hidden).await.unwrap();
+        }
+
+        let mut visible_older = visible_newer.clone();
+        visible_older.id = "past-ceiling".into();
+        visible_older.created_at = "2026-01-01T00:00:00Z".into();
+        visible_older.updated_at = visible_older.created_at.clone();
+        db.create_task(&visible_older).await.unwrap();
+
+        let schema = schema(db);
+        let mut seen: Vec<String> = Vec::new();
+        let mut cursor: Option<String> = None;
+        for request in 0..10 {
+            let query = match &cursor {
+                Some(c) => format!(
+                    r#"{{ tasks(limit: 1, cursor: "{c}") {{ items {{ id }} hasMore incomplete nextCursor }} }}"#
+                ),
+                None => {
+                    "{ tasks(limit: 1) { items { id } hasMore incomplete nextCursor } }".to_string()
+                }
+            };
+            let resp = anon(&schema, &query).await;
+            let rendered = format!("{:?}", resp.data);
+            assert!(
+                !rendered.contains("hidden-"),
+                "request {request} disclosed a denied row: {rendered}"
+            );
+            for item in task_items(&resp) {
+                let async_graphql::Value::Object(row) = item else {
+                    panic!("task item not an object");
+                };
+                seen.push(row.get("id").unwrap().to_string().replace('"', ""));
+            }
+            // Short page with more behind it is the scan wall, and says so.
+            if task_page_bool(&resp, "hasMore") && task_items(&resp).is_empty() {
+                assert!(
+                    task_incomplete(&resp),
+                    "request {request}: an empty page with rows behind it is a paused scan"
+                );
+            }
+            match task_page_cursor(&resp) {
+                Some(c) => cursor = Some(c),
+                None => {
+                    assert!(!task_page_bool(&resp, "hasMore"));
+                    assert!(!task_incomplete(&resp));
+                    break;
+                }
+            }
+        }
+
+        assert_eq!(
+            seen,
+            vec!["newer-visible".to_string(), "past-ceiling".to_string()],
+            "both visible rows must be reachable using only server-issued cursors"
+        );
+    }
+
+    /// REST and GraphQL mint the same tokens from the same node key, so a
+    /// cursor must not be usable against a filter it was not issued for, and
+    /// must render the same single rejection every other bad cursor renders.
+    #[sqlx::test]
+    async fn tasks_rejects_unusable_cursor(pool: PgPool) {
+        use crate::api::task_cursor::{self, TaskFilter, TaskPosition};
+
+        let db = db(pool).await;
+        let schema = schema(db);
+
+        let wrong_filter = task_cursor::encode(
+            &cursor_key(),
+            TaskFilter {
+                status: Some("pending"),
+                assignee_did: None,
+            },
+            None,
+            &TaskPosition::new("2026-01-01T00:00:00Z", "t1"),
+        );
+        let foreign = task_cursor::encode(
+            &crate::api::task_cursor::TaskCursorKey::derive(&[9u8; 32]),
+            TaskFilter {
+                status: None,
+                assignee_did: None,
+            },
+            None,
+            &TaskPosition::new("2026-01-01T00:00:00Z", "t1"),
+        );
+
+        for (label, query) in [
+            (
+                "garbage",
+                r#"{ tasks(cursor: "not-a-cursor") { items { id } } }"#.to_string(),
+            ),
+            (
+                "another node's key",
+                format!(r#"{{ tasks(cursor: "{foreign}") {{ items {{ id }} }} }}"#),
+            ),
+            (
+                "different filter",
+                format!(r#"{{ tasks(cursor: "{wrong_filter}") {{ items {{ id }} }} }}"#),
+            ),
+        ] {
+            let resp = anon(&schema, &query).await;
+            assert_eq!(
+                resp.errors.len(),
+                1,
+                "{label}: must be rejected, not treated as page one"
+            );
+            assert!(
+                resp.errors[0].message.contains("invalid or expired cursor"),
+                "{label}: unexpected message {:?}",
+                resp.errors[0].message
+            );
+        }
+
+        let resp = anon(
+            &schema,
+            &format!(
+                r#"{{ tasks(status: "pending", cursor: "{wrong_filter}") {{ items {{ id }} }} }}"#
+            ),
+        )
+        .await;
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+    }
+
+    #[sqlx::test]
+    async fn tasks_anonymous_denial_hides_repoless_task_and_leaks_no_token(pool: PgPool) {
+        let db = db(pool).await;
+        let task = crate::db::AgentTask {
+            id: "t1".into(),
+            repo_id: None,
+            kind: "code-review".into(),
+            status: "pending".into(),
+            delegator_did: "did:key:z6MkDelegator".into(),
+            assignee_did: None,
+            capability: "agent:task".into(),
+            ucan_token: Some("secret-ucan-token".into()),
+            payload: Some("payload".into()),
+            result: None,
+            created_at: "2026-01-01T00:00:00Z".into(),
+            updated_at: "2026-01-01T00:00:00Z".into(),
+            deadline: None,
+        };
+        db.create_task(&task).await.unwrap();
+
+        let schema = schema(db);
+        let query = "{ tasks { items { id } } }";
+        let resp = anon(&schema, query).await;
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+        assert_eq!(count_tasks(&resp), 0);
+        let rendered = format!("{:?}", resp.data);
+        assert!(!rendered.contains("secret-ucan-token"));
+    }
+
+    #[sqlx::test]
+    async fn task_anonymous_denial_returns_null_and_leaks_no_token(pool: PgPool) {
+        let db = db(pool).await;
+        let task = crate::db::AgentTask {
+            id: "t1".into(),
+            repo_id: None,
+            kind: "code-review".into(),
+            status: "pending".into(),
+            delegator_did: "did:key:z6MkDelegator".into(),
+            assignee_did: None,
+            capability: "agent:task".into(),
+            ucan_token: Some("secret-ucan-token".into()),
+            payload: Some("payload".into()),
+            result: None,
+            created_at: "2026-01-01T00:00:00Z".into(),
+            updated_at: "2026-01-01T00:00:00Z".into(),
+            deadline: None,
+        };
+        db.create_task(&task).await.unwrap();
+
+        let schema = schema(db);
+        let query = r#"{ task(id: "t1") { id } }"#;
+        let resp = anon(&schema, query).await;
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+        let async_graphql::Value::Object(obj) = &resp.data else {
+            panic!("data not an object: {:?}", resp.data);
+        };
+        assert_eq!(obj.get("task"), Some(&async_graphql::Value::Null));
+        let rendered = format!("{:?}", resp.data);
+        assert!(!rendered.contains("secret-ucan-token"));
+    }
+
+    fn task_page_bool(resp: &async_graphql::Response, field: &str) -> bool {
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+        let async_graphql::Value::Object(obj) = &resp.data else {
+            panic!("data not an object: {:?}", resp.data);
+        };
+        let async_graphql::Value::Object(page) = obj.get("tasks").expect("tasks key") else {
+            panic!("tasks not an object");
+        };
+        let async_graphql::Value::Boolean(value) = page.get(field).expect("field present") else {
+            panic!("{field} not a bool");
+        };
+        *value
+    }
+
+    fn task_page_cursor(resp: &async_graphql::Response) -> Option<String> {
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+        let async_graphql::Value::Object(obj) = &resp.data else {
+            panic!("data not an object: {:?}", resp.data);
+        };
+        let async_graphql::Value::Object(page) = obj.get("tasks").expect("tasks key") else {
+            panic!("tasks not an object");
+        };
+        match page.get("nextCursor").expect("nextCursor key") {
+            async_graphql::Value::Null => None,
+            async_graphql::Value::String(c) => Some(c.clone()),
+            other => panic!("nextCursor not a string: {other:?}"),
+        }
+    }
+
+    fn task_items(resp: &async_graphql::Response) -> &Vec<async_graphql::Value> {
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+        let async_graphql::Value::Object(obj) = &resp.data else {
+            panic!("data not an object: {:?}", resp.data);
+        };
+        let async_graphql::Value::Object(page) = obj.get("tasks").expect("tasks key") else {
+            panic!("tasks not an object");
+        };
+        let async_graphql::Value::List(rows) = page.get("items").expect("items key") else {
+            panic!("items not a list");
+        };
+        rows
+    }
+
+    fn count_tasks(resp: &async_graphql::Response) -> usize {
+        task_items(resp).len()
+    }
+
+    fn task_incomplete(resp: &async_graphql::Response) -> bool {
+        assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
+        let async_graphql::Value::Object(obj) = &resp.data else {
+            panic!("data not an object: {:?}", resp.data);
+        };
+        let async_graphql::Value::Object(page) = obj.get("tasks").expect("tasks key") else {
+            panic!("tasks not an object");
+        };
+        let async_graphql::Value::Boolean(incomplete) =
+            page.get("incomplete").expect("incomplete key")
+        else {
+            panic!("incomplete not a bool");
+        };
+        *incomplete
     }
 }

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

@@ -11,9 +11,9 @@ pub struct SubscriptionRoot;
 
 #[Subscription]
 impl SubscriptionRoot {
-    /// Live ref-update stream. `/graphql/ws` is mounted outside the
-    /// `optional_signature` layer, so this resolver has NO caller identity and
-    /// cannot gate per-subscriber — it relays whatever enters the broadcast
+    /// Live ref-update stream. `/graphql/ws` is mounted under the
+    /// `optional_signature` layer and carries optional caller identity, but this
+    /// resolver does not inspect it — it relays whatever enters the broadcast
     /// channel to any anonymous client. Its visibility safety therefore rests
     /// entirely on the WRITE side: the push handler only sends a
     /// `RefUpdateBroadcast` for repos the anonymous public may read (inside its

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

@@ -56,6 +56,65 @@ impl From<AgentTask> for AgentTaskType {
     }
 }
 
+/// Read-only projection of `AgentTask` for the `tasks`/`task` queries, as
+/// opposed to `AgentTaskType`, which the task mutations (`createTask`,
+/// `claimTask`, `completeTask`, `failTask` — all `require_signer`-gated)
+/// return. Identical except for the missing `ucan_token` (#268): a read
+/// surface never needs to echo it back, since the assignee already received it
+/// at delegation/claim time via the mutation response.
+#[derive(SimpleObject, Clone)]
+pub struct AgentTaskReadType {
+    pub id: String,
+    pub repo_id: Option<String>,
+    pub kind: String,
+    pub status: String,
+    pub delegator_did: String,
+    pub assignee_did: Option<String>,
+    pub capability: String,
+    pub payload: Option<String>,
+    pub result: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+    pub deadline: Option<String>,
+}
+
+/// Wraps a `tasks` page with the same completion signals the REST list route
+/// exposes, so the two surfaces answer pagination identically.
+///
+/// `hasMore` and `incomplete` are separate because they are separate facts
+/// (#327 review): `hasMore` says more candidates remain, `incomplete` says
+/// this page is short *only* because the authorization scan hit its safety
+/// wall. `nextCursor` is present exactly when `hasMore` is true and is an
+/// opaque MAC'd token — it can name the last examined candidate, denied or
+/// not, without disclosing it, which is what lets a caller page past a denied
+/// window instead of stalling on it.
+#[derive(SimpleObject, Clone)]
+pub struct TaskPageType {
+    pub items: Vec<AgentTaskReadType>,
+    pub has_more: bool,
+    pub incomplete: bool,
+    pub next_cursor: Option<String>,
+}
+
+impl From<AgentTask> for AgentTaskReadType {
+    fn from(t: AgentTask) -> Self {
+        Self {
+            id: t.id,
+            repo_id: t.repo_id,
+            kind: t.kind,
+            status: t.status,
+            delegator_did: t.delegator_did,
+            assignee_did: t.assignee_did,
+            capability: t.capability,
+            payload: t.payload,
+            result: t.result,
+            created_at: t.created_at,
+            updated_at: t.updated_at,
+            deadline: t.deadline,
+        }
+    }
+}
+
 #[derive(SimpleObject, Clone)]
 pub struct RefUpdateType {
     pub repo: String,

crates/gitlawb-node/src/main.rs

@@ -24,6 +24,7 @@ mod state;
 mod sync;
 #[cfg(test)]
 mod test_support;
+mod upstream_mirror;
 mod visibility;
 mod webhooks;
 
@@ -307,10 +308,15 @@ async fn main() -> Result<()> {
     let (ref_update_tx, _) = tokio::sync::broadcast::channel::<state::RefUpdateBroadcast>(256);
     let (task_event_tx, _) = tokio::sync::broadcast::channel::<state::TaskEventBroadcast>(256);
 
+    // Node-keyed, so continuation tokens this node mints are only accepted by
+    // this node and survive its restarts without any configured secret.
+    let task_cursor_key = api::task_cursor::TaskCursorKey::derive(&keypair.to_seed());
+
     let graphql_schema = Arc::new(graphql::build_schema(
         Arc::clone(&db),
         ref_update_tx.clone(),
         task_event_tx.clone(),
+        task_cursor_key.clone(),
     ));
 
     let machine_id = std::env::var("FLY_MACHINE_ID").ok();
@@ -351,9 +357,14 @@ async fn main() -> Result<()> {
     );
 
     // Per-DID limiter for the creation endpoints. Keyed on the authenticated
-    // DID (attacker-varied), so bound its key set to cap memory.
-    let rate_limiter =
-        rate_limit::RateLimiter::new_bounded(10, std::time::Duration::from_secs(3600), 200_000);
+    // DID (attacker-varied), so bound its key set to cap memory. Max and window
+    // are operator-configurable via GITLAWB_RATE_LIMIT_MAX /
+    // GITLAWB_RATE_LIMIT_WINDOW_SECS; both reject 0 at parse time.
+    let rate_limiter = rate_limit::RateLimiter::new_bounded(
+        config.rate_limit_max_requests as usize,
+        std::time::Duration::from_secs(config.rate_limit_window_secs),
+        200_000,
+    );
 
     // Per-client-IP flood brake for the creation endpoints. The per-DID limiter
     // above is bypassed by a DID farm (one throwaway did:key per repo), which is
@@ -451,6 +462,7 @@ async fn main() -> Result<()> {
         db,
         node_did: node_did.clone(),
         node_keypair: keypair,
+        task_cursor_key,
         p2p: p2p_handle,
         http_client,
         ref_update_tx,
@@ -549,11 +561,19 @@ async fn main() -> Result<()> {
             std::time::Duration::from_secs(3600),
             200_000,
         ),
+        task_read_rate_limiter: rate_limit::RateLimiter::new_bounded(
+            config.task_read_rate_limit,
+            std::time::Duration::from_secs(3600),
+            200_000,
+        ),
         git_bin: "git".to_string(),
     };
     if config.ipfs_rate_limit == 0 {
         tracing::warn!("GITLAWB_IPFS_RATE_LIMIT=0 — per-IP /ipfs rate limiting disabled");
     }
+    if config.task_read_rate_limit == 0 {
+        tracing::warn!("GITLAWB_TASK_READ_RATE_LIMIT=0 — per-IP task read rate limiting disabled");
+    }
 
     // Periodic peer-count poll for the metrics gauge. If p2p is disabled
     // we still set the gauge to 0 so dashboards don't show "no data".
@@ -667,6 +687,22 @@ async fn main() -> Result<()> {
         info!("auto-sync worker started");
     }
 
+    // External-forge mirroring is a separate, default-off worker. Startup
+    // validation already requires owner-only pushes when it is enabled; the
+    // worker additionally verifies the Git runtime can pin DNS answers before
+    // any URL reaches `git fetch`.
+    if config.upstream_mirror_enabled {
+        upstream_mirror::start(
+            Arc::clone(&state.db),
+            Arc::clone(&state.config),
+            state.repo_store.clone(),
+            state.repo_write_leases.clone(),
+            Arc::clone(&state.git_write_semaphore),
+            state.subscribe_shutdown(),
+        )
+        .context("starting upstream mirror worker")?;
+    }
+
     // Periodic reconciliation sweep: re-derives pin/seal sets and fills gaps
     // so a dropped replication job never means data loss.
     {
@@ -1212,6 +1248,8 @@ mod rate_limiter_sweep_tests {
         state.peer_write_rate_limiter = RateLimiter::new(10, window);
         state.federated_rate_limiter = RateLimiter::new(10, window);
         state.ipfs_rate_limiter = RateLimiter::new(10, window);
+        state.ipfs_work_rate_limiter = RateLimiter::new(10, window);
+        state.task_read_rate_limiter = RateLimiter::new(10, window);
 
         let limiters = |s: &crate::state::AppState| {
             [
@@ -1223,6 +1261,8 @@ mod rate_limiter_sweep_tests {
                 s.peer_write_rate_limiter.clone(),
                 s.federated_rate_limiter.clone(),
                 s.ipfs_rate_limiter.clone(),
+                s.ipfs_work_rate_limiter.clone(),
+                s.task_read_rate_limiter.clone(),
             ]
         };
         for l in limiters(&state) {

crates/gitlawb-node/src/rate_limit.rs

@@ -70,6 +70,14 @@ impl RateLimiter {
         self.window.min(Duration::from_secs(1))
     }
 
+    /// The configured window, exposed so a 429 can carry an honest
+    /// `retry-after`: a hardcoded value lies when the operator retunes the
+    /// window, and a client that honors the header would retry into another
+    /// 429 until the real window elapses.
+    pub fn window(&self) -> Duration {
+        self.window
+    }
+
     pub(crate) async fn check(&self, key: &str) -> bool {
         // max_requests == 0 means the limiter is disabled, not "block all".
         if self.max_requests == 0 {
@@ -293,12 +301,7 @@ pub async fn rate_limit_by_did(request: Request, next: Next) -> Response {
 
     if let (Some(limiter), Some(did)) = (limiter, did) {
         if !limiter.check(&did).await {
-            return (
-                StatusCode::TOO_MANY_REQUESTS,
-                [("retry-after", "60")],
-                "rate limit exceeded — try again later",
-            )
-                .into_response();
+            return too_many_requests(limiter.window().as_secs());
         }
     }
 
@@ -401,18 +404,72 @@ impl<S: Send + Sync> axum::extract::FromRequestParts<S> for PeerAddr {
     }
 }
 
+/// The message every per-IP brake answers with, shared so the GraphQL surface
+/// (which cannot return a 429 status inside a 200 GraphQL envelope) says the
+/// same thing as the REST routes.
+pub const RATE_LIMIT_MESSAGE: &str = "rate limit exceeded — try again later";
+
 /// The shared 429 response for the per-IP flood brakes. Route-agnostic: this
 /// middleware now serves the push path AND the peer-sync routes, so the message
 /// stays generic (the offending path is recorded in the warn log below).
-pub fn too_many_requests() -> Response {
+pub fn too_many_requests(retry_after_secs: u64) -> Response {
     (
         StatusCode::TOO_MANY_REQUESTS,
-        [("retry-after", "60")],
-        "rate limit exceeded — try again later",
+        [("retry-after", retry_after_secs.to_string())],
+        RATE_LIMIT_MESSAGE,
     )
         .into_response()
 }
 
+/// The per-IP task-read brake carried as GraphQL request data: the limiter plus
+/// the client key the transport already resolved.
+///
+/// `/graphql` is a single POST endpoint, so this brake cannot sit in middleware
+/// the way `task_read_routes` mounts `rate_limit_by_ip`: that would charge every
+/// unrelated query and every mutation against the task-read bucket. The two task
+/// resolvers debit it instead, which also prices an aliased query honestly — ten
+/// aliased `tasks` fields run the visibility gate ten times and pay ten slots,
+/// where one request-scoped debit would pay one (#327 review).
+/// Default cap on the number of task read fields (across all aliases) permitted
+/// in a single GraphQL request before rejecting further field executions (#327 review).
+pub const MAX_GRAPHQL_TASK_READS_PER_REQUEST: usize = 5;
+
+#[derive(Clone)]
+pub struct TaskReadBrake {
+    pub limiter: RateLimiter,
+    /// `None` when no key could be resolved at all (no trusted header and no
+    /// `ConnectInfo`, e.g. a synthetic test request). Never braked, matching
+    /// [`rate_limit_by_ip`].
+    pub key: Option<String>,
+    pub request_count: Arc<std::sync::atomic::AtomicUsize>,
+}
+
+impl TaskReadBrake {
+    /// Debit one slot for one task-read field. `true` = proceed.
+    pub async fn check(&self, field: &str) -> bool {
+        let Some(key) = self.key.as_deref() else {
+            return true;
+        };
+        let count = self
+            .request_count
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        if count >= MAX_GRAPHQL_TASK_READS_PER_REQUEST {
+            tracing::warn!(
+                key = %key,
+                field = %field,
+                count = count + 1,
+                "per-request GraphQL task field limit exceeded"
+            );
+            return false;
+        }
+        if self.limiter.check(key).await {
+            return true;
+        }
+        tracing::warn!(key = %key, field = %field, "per-IP rate limit exceeded");
+        false
+    }
+}
+
 /// Throttle the git push path by resolved client IP. The socket peer address is
 /// read from `ConnectInfo` (see `into_make_service_with_connect_info` in
 /// `main`). Only skips the limiter when no key can be resolved at all.
@@ -427,7 +484,7 @@ pub async fn rate_limit_by_ip(request: Request, next: Next) -> Response {
         if let Some(key) = client_key(request.headers(), peer, limiter.trust) {
             if !limiter.limiter.check(&key).await {
                 tracing::warn!(key = %key, path = %request.uri().path(), "per-IP rate limit exceeded");
-                return too_many_requests();
+                return too_many_requests(limiter.limiter.window().as_secs());
             }
         }
     }

crates/gitlawb-node/src/server.rs

@@ -1,5 +1,6 @@
-use async_graphql_axum::{GraphQLRequest, GraphQLResponse, GraphQLSubscription};
-use axum::extract::DefaultBodyLimit;
+use async_graphql::http::ALL_WEBSOCKET_PROTOCOLS;
+use async_graphql_axum::{GraphQLProtocol, GraphQLRequest, GraphQLResponse, GraphQLWebSocket};
+use axum::extract::{DefaultBodyLimit, WebSocketUpgrade};
 use axum::{
     extract::State,
     middleware,
@@ -23,6 +24,8 @@ use crate::state::AppState;
 async fn graphql_handler(
     State(state): State<AppState>,
     auth: Option<axum::Extension<crate::auth::AuthenticatedDid>>,
+    headers: axum::http::HeaderMap,
+    rate_limit::PeerAddr(peer): rate_limit::PeerAddr,
     req: GraphQLRequest,
 ) -> GraphQLResponse {
     // `optional_signature` attaches the verified DID when a signature is present.
@@ -32,9 +35,49 @@ async fn graphql_handler(
     if let Some(axum::Extension(did)) = auth {
         inner = inner.data(did);
     }
+    // The anonymous `tasks`/`task` resolvers run the same #268 visibility gate
+    // as the REST read routes and cost the node the same queries, so they carry
+    // the same per-IP brake. It rides as request data rather than a router layer
+    // because /graphql is one endpoint for every operation — see `TaskReadBrake`
+    // (#327 review). It debits before the gate runs, but unlike the REST layer
+    // it sits inside `optional_signature`, so it brakes the gate's query cost
+    // and not signature verification.
+    inner = inner.data(rate_limit::TaskReadBrake {
+        limiter: state.task_read_rate_limiter.clone(),
+        key: rate_limit::client_key(&headers, peer, state.push_limiter_trust),
+        request_count: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+    });
     state.graphql_schema.execute(inner).await.into()
 }
 
+async fn graphql_ws_handler(
+    State(state): State<AppState>,
+    headers: axum::http::HeaderMap,
+    rate_limit::PeerAddr(peer): rate_limit::PeerAddr,
+    auth: Option<axum::Extension<auth::AuthenticatedDid>>,
+    protocol: GraphQLProtocol,
+    upgrade: WebSocketUpgrade,
+) -> axum::response::Response {
+    let mut data = async_graphql::Data::default();
+    if let Some(axum::Extension(did)) = auth {
+        data.insert(did);
+    }
+    data.insert(rate_limit::TaskReadBrake {
+        limiter: state.task_read_rate_limiter.clone(),
+        key: rate_limit::client_key(&headers, peer, state.push_limiter_trust),
+        request_count: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+    });
+    let schema = state.graphql_schema.as_ref().clone();
+    upgrade
+        .protocols(ALL_WEBSOCKET_PROTOCOLS)
+        .on_upgrade(move |stream| {
+            GraphQLWebSocket::new(stream, schema, protocol)
+                .with_data(data)
+                .serve()
+        })
+        .into_response()
+}
+
 async fn graphql_playground() -> impl IntoResponse {
     axum::response::Html(async_graphql::http::playground_source(
         async_graphql::http::GraphQLPlaygroundConfig::new("/graphql")
@@ -77,17 +120,14 @@ fn add_auth_layers(
 
 pub fn build_router(state: AppState) -> Router {
     // ── GraphQL routes ─────────────────────────────────────────────────────
-    let schema = state.graphql_schema.as_ref().clone();
     let graphql_routes = Router::new()
         .route("/graphql", get(graphql_playground).post(graphql_handler))
-        // Attach the verified DID to /graphql when a signature is present. The
-        // layer covers only routes added before it, so /graphql/ws (added after,
-        // read-only subscriptions) stays open.
+        .route("/graphql/ws", get(graphql_ws_handler))
+        // Attach the verified DID to /graphql and /graphql/ws when a signature is present.
         .layer(middleware::from_fn_with_state(
             auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
             auth::optional_signature,
-        ))
-        .route_service("/graphql/ws", GraphQLSubscription::new(schema));
+        ));
 
     // ── Task routes (write — require HTTP Signature) ───────────────────────
     let task_write_routes = add_auth_layers(
@@ -100,10 +140,31 @@ pub fn build_router(state: AppState) -> Router {
         SIGNED_BODY_LIMIT,
     );
 
-    // ── Task routes (read — open) ──────────────────────────────────────────
+    // ── Task routes (read — open, but scoped) ──────────────────────────────
+    // `optional_signature` attaches the verified DID when a signature is present
+    // so the handlers can identify the caller; the routes stay anonymous-reachable,
+    // but each task/row is gated to its delegator, its assignee, or (for a
+    // repo-scoped task) whoever can read that repo (#268 — these routes previously
+    // carried no gate and no identity at all).
+    // Both routes also carry a per-IP flood brake, mirroring `/ipfs/{cid}`: they are
+    // anon-reachable and the gate above costs a task lookup plus deduped-repo and
+    // visibility-rule queries *before* the opaque 404, so a prober pays nothing and
+    // the node pays per request. The limiter is the outermost layer so a flood is
+    // rejected before signature verification and the visibility queries run. The
+    // extension MUST be attached or `rate_limit_by_ip` is a silent no-op.
+    let task_read_limiter = rate_limit::IpRateLimiter {
+        limiter: state.task_read_rate_limiter.clone(),
+        trust: state.push_limiter_trust,
+    };
     let task_read_routes = Router::new()
         .route("/api/v1/tasks", get(tasks::list_tasks))
-        .route("/api/v1/tasks/{id}", get(tasks::get_task));
+        .route("/api/v1/tasks/{id}", get(tasks::get_task))
+        .layer(middleware::from_fn_with_state(
+            auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
+            auth::optional_signature,
+        ))
+        .layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
+        .layer(axum::Extension(task_read_limiter));
 
     // ── Rate-limited creation routes — require HTTP Signature, plus a per-DID
     // throttle AND a per-IP flood brake. The per-DID limiter (inner) caps a
@@ -715,13 +776,351 @@ async fn p2p_info(State(state): State<AppState>) -> Json<serde_json::Value> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::test_support::test_state;
+    use sqlx::PgPool;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    async fn ws_send_text(stream: &mut tokio::net::TcpStream, text: &str) {
+        let payload = text.as_bytes();
+        let len = payload.len();
+        let mut frame = Vec::new();
+        frame.push(0x81);
+        let mask = [0x12, 0x34, 0x56, 0x78];
+        if len <= 125 {
+            frame.push(0x80 | (len as u8));
+        } else if len <= 65535 {
+            frame.push(0x80 | 126);
+            frame.extend_from_slice(&(len as u16).to_be_bytes());
+        } else {
+            frame.push(0x80 | 127);
+            frame.extend_from_slice(&(len as u64).to_be_bytes());
+        }
+        frame.extend_from_slice(&mask);
+        for (i, b) in payload.iter().enumerate() {
+            frame.push(b ^ mask[i % 4]);
+        }
+        stream.write_all(&frame).await.unwrap();
+        stream.flush().await.unwrap();
+    }
+
+    async fn ws_recv_text(stream: &mut tokio::net::TcpStream) -> String {
+        let mut header = [0u8; 2];
+        stream.read_exact(&mut header).await.unwrap();
+        let b1 = header[1];
+        let masked = (b1 & 0x80) != 0;
+        let mut len = (b1 & 0x7f) as usize;
+        if len == 126 {
+            let mut ext = [0u8; 2];
+            stream.read_exact(&mut ext).await.unwrap();
+            len = u16::from_be_bytes(ext) as usize;
+        } else if len == 127 {
+            let mut ext = [0u8; 8];
+            stream.read_exact(&mut ext).await.unwrap();
+            len = u64::from_be_bytes(ext) as usize;
+        }
+        let mask = if masked {
+            let mut m = [0u8; 4];
+            stream.read_exact(&mut m).await.unwrap();
+            Some(m)
+        } else {
+            None
+        };
+        let mut payload = vec![0u8; len];
+        stream.read_exact(&mut payload).await.unwrap();
+        if let Some(m) = mask {
+            for (i, b) in payload.iter_mut().enumerate() {
+                *b ^= m[i % 4];
+            }
+        }
+        String::from_utf8(payload).unwrap()
+    }
+
+    async fn ws_recv_op_frames(stream: &mut tokio::net::TcpStream, op_id: &str) -> Vec<String> {
+        let mut frames = Vec::new();
+        loop {
+            let text = ws_recv_text(stream).await;
+            frames.push(text.clone());
+            if let Ok(v) = serde_json::from_str::<serde_json::Value>(&text) {
+                if v.get("id").and_then(|id| id.as_str()) == Some(op_id) {
+                    let msg_type = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
+                    if msg_type == "complete" || msg_type == "error" {
+                        break;
+                    }
+                }
+            }
+        }
+        frames
+    }
+
+    async fn connect_ws(addr: std::net::SocketAddr) -> tokio::net::TcpStream {
+        let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
+        let req = format!(
+            "GET /graphql/ws HTTP/1.1\r\n\
+             Host: {}\r\n\
+             Upgrade: websocket\r\n\
+             Connection: Upgrade\r\n\
+             Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
+             Sec-WebSocket-Version: 13\r\n\
+             Sec-WebSocket-Protocol: graphql-transport-ws\r\n\r\n",
+            addr
+        );
+        stream.write_all(req.as_bytes()).await.unwrap();
+        stream.flush().await.unwrap();
+
+        let mut buf = [0u8; 1024];
+        let n = stream.read(&mut buf).await.unwrap();
+        let resp = String::from_utf8_lossy(&buf[..n]);
+        assert!(resp.starts_with("HTTP/1.1 101 Switching Protocols"));
+
+        // Init connection
+        ws_send_text(&mut stream, r#"{"type":"connection_init"}"#).await;
+        let ack = ws_recv_text(&mut stream).await;
+        assert!(ack.contains("connection_ack"));
+
+        stream
+    }
+
+    #[sqlx::test]
+    async fn graphql_ws_task_query_enforces_per_request_field_limit(pool: PgPool) {
+        let state = test_state(pool).await;
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = build_router(state);
+        tokio::spawn(async move {
+            axum::serve(
+                listener,
+                app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
+            )
+            .await
+            .unwrap();
+        });
+
+        let mut stream = connect_ws(addr).await;
+
+        // Query with 6 aliased task fields (exceeding MAX_GRAPHQL_TASK_READS_PER_REQUEST = 5).
+        // limit: 1 keeps the document under GRAPHQL_MAX_COMPLEXITY so the 6th
+        // field reaches the brake instead of failing complexity first.
+        let query = r#"{"id":"1","type":"subscribe","payload":{"query":"query { f1: tasks(limit: 1) { items { id } } f2: tasks(limit: 1) { items { id } } f3: tasks(limit: 1) { items { id } } f4: tasks(limit: 1) { items { id } } f5: tasks(limit: 1) { items { id } } f6: tasks(limit: 1) { items { id } } }"}}"#;
+        ws_send_text(&mut stream, query).await;
+        let frames = ws_recv_op_frames(&mut stream, "1").await;
+        let resp = frames.join("\n");
+        assert!(
+            resp.contains("rate limit exceeded"),
+            "6th task field over WS must be braked: {resp}"
+        );
+    }
+
+    #[sqlx::test]
+    async fn graphql_ws_task_query_enforces_per_ip_rate_limit(pool: PgPool) {
+        let mut state = test_state(pool).await;
+        // Restrict task read rate limiter to 1 request per 60s
+        state.task_read_rate_limiter =
+            crate::rate_limit::RateLimiter::new(1, std::time::Duration::from_secs(60));
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = build_router(state);
+        tokio::spawn(async move {
+            axum::serve(
+                listener,
+                app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
+            )
+            .await
+            .unwrap();
+        });
+
+        let mut stream = connect_ws(addr).await;
+
+        // First task query succeeds (or returns valid data)
+        let query1 = r#"{"id":"1","type":"subscribe","payload":{"query":"query { tasks { items { id } } }"}}"#;
+        ws_send_text(&mut stream, query1).await;
+        let frames1 = ws_recv_op_frames(&mut stream, "1").await;
+        let resp1 = frames1.join("\n");
+        assert!(!resp1.contains("rate limit exceeded"));
+
+        // Second task query on the same connection hits per-IP limiter
+        let query2 = r#"{"id":"2","type":"subscribe","payload":{"query":"query { tasks { items { id } } }"}}"#;
+        ws_send_text(&mut stream, query2).await;
+        let frames2 = ws_recv_op_frames(&mut stream, "2").await;
+        let resp2 = frames2.join("\n");
+        assert!(
+            resp2.contains("rate limit exceeded"),
+            "exceeded per-IP limiter over WS must return rate limit message: {resp2}"
+        );
+    }
+
+    #[sqlx::test]
+    async fn graphql_ws_task_query_resets_field_budget_per_operation(pool: PgPool) {
+        let state = test_state(pool).await;
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = build_router(state);
+        tokio::spawn(async move {
+            axum::serve(
+                listener,
+                app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
+            )
+            .await
+            .unwrap();
+        });
+
+        let mut stream = connect_ws(addr).await;
+
+        // Execute 6 sequential 1-field queries on the same connection.
+        // Each operation must receive its own 5-field budget, so none are braked by the per-request limit.
+        for i in 1..=6 {
+            let id = i.to_string();
+            let query = format!(
+                r#"{{"id":"{id}","type":"subscribe","payload":{{"query":"query {{ tasks {{ items {{ id }} }} }}"}}}}"#
+            );
+            ws_send_text(&mut stream, &query).await;
+            let frames = ws_recv_op_frames(&mut stream, &id).await;
+            let resp = frames.join("\n");
+            assert!(
+                !resp.contains("rate limit exceeded"),
+                "operation {id} on the same WS connection must have a fresh field budget: {resp}"
+            );
+        }
+    }
+
+    async fn connect_ws_signed(
+        addr: std::net::SocketAddr,
+        keypair: &gitlawb_core::identity::Keypair,
+    ) -> tokio::net::TcpStream {
+        let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
+        let signed = gitlawb_core::http_sig::sign_request(keypair, "GET", "/graphql/ws", b"");
+        let req = format!(
+            "GET /graphql/ws HTTP/1.1\r\n\
+             Host: {}\r\n\
+             Upgrade: websocket\r\n\
+             Connection: Upgrade\r\n\
+             Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
+             Sec-WebSocket-Version: 13\r\n\
+             Sec-WebSocket-Protocol: graphql-transport-ws\r\n\
+             content-digest: {}\r\n\
+             signature-input: {}\r\n\
+             signature: {}\r\n\r\n",
+            addr, signed.content_digest, signed.signature_input, signed.signature
+        );
+        stream.write_all(req.as_bytes()).await.unwrap();
+        stream.flush().await.unwrap();
+
+        let mut buf = [0u8; 1024];
+        let n = stream.read(&mut buf).await.unwrap();
+        let resp = String::from_utf8_lossy(&buf[..n]);
+        assert!(resp.starts_with("HTTP/1.1 101 Switching Protocols"));
+
+        // Init connection
+        ws_send_text(&mut stream, r#"{"type":"connection_init"}"#).await;
+        let ack = ws_recv_text(&mut stream).await;
+        assert!(ack.contains("connection_ack"));
+
+        stream
+    }
+
+    #[sqlx::test]
+    async fn graphql_ws_authenticated_query_accesses_private_task(pool: PgPool) {
+        let state = test_state(pool).await;
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let app = build_router(state.clone());
+        tokio::spawn(async move {
+            axum::serve(
+                listener,
+                app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
+            )
+            .await
+            .unwrap();
+        });
+
+        let keypair = gitlawb_core::identity::Keypair::generate();
+        let delegator_did = keypair.did().to_string();
+
+        let task = crate::db::AgentTask {
+            id: "ws-auth-task-01".into(),
+            repo_id: None,
+            delegator_did: delegator_did.clone(),
+            kind: "test".into(),
+            capability: "read".into(),
+            status: "pending".into(),
+            assignee_did: Some(delegator_did.clone()),
+            ucan_token: None,
+            payload: None,
+            result: None,
+            created_at: "2026-01-01T00:00:00Z".into(),
+            updated_at: "2026-01-01T00:00:00Z".into(),
+            deadline: None,
+        };
+        state.db.create_task(&task).await.unwrap();
+
+        // 1. Unauthenticated WS connection cannot see the private task
+        let mut unauth_stream = connect_ws(addr).await;
+        let query = r#"{"id":"1","type":"subscribe","payload":{"query":"query { tasks { items { id } } }"}}"#;
+        ws_send_text(&mut unauth_stream, query).await;
+        let unauth_frames = ws_recv_op_frames(&mut unauth_stream, "1").await;
+        let unauth_resp = unauth_frames.join("\n");
+        assert!(
+            !unauth_resp.contains("ws-auth-task-01"),
+            "unauthenticated WS must not disclose private task: {unauth_resp}"
+        );
+
+        // 2. Forged signature claiming the delegator DID must be rejected at handshake
+        let attacker_keypair = gitlawb_core::identity::Keypair::generate();
+        let attacker_signed =
+            gitlawb_core::http_sig::sign_request(&attacker_keypair, "GET", "/graphql/ws", b"");
+        let forged_sig_input = attacker_signed
+            .signature_input
+            .replace(&attacker_keypair.did().to_string(), &delegator_did);
+        let mut forged_stream = tokio::net::TcpStream::connect(addr).await.unwrap();
+        let forged_req = format!(
+            "GET /graphql/ws HTTP/1.1\r\n\
+             Host: {}\r\n\
+             Upgrade: websocket\r\n\
+             Connection: Upgrade\r\n\
+             Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
+             Sec-WebSocket-Version: 13\r\n\
+             Sec-WebSocket-Protocol: graphql-transport-ws\r\n\
+             content-digest: {}\r\n\
+             signature-input: {}\r\n\
+             signature: {}\r\n\r\n",
+            addr, attacker_signed.content_digest, forged_sig_input, attacker_signed.signature
+        );
+        forged_stream
+            .write_all(forged_req.as_bytes())
+            .await
+            .unwrap();
+        forged_stream.flush().await.unwrap();
+        let mut forged_bytes = Vec::new();
+        let mut chunk = [0u8; 1024];
+        while !forged_bytes.windows(4).any(|w| w == b"\r\n\r\n") {
+            let n = forged_stream.read(&mut chunk).await.unwrap();
+            if n == 0 {
+                break;
+            }
+            forged_bytes.extend_from_slice(&chunk[..n]);
+        }
+        let forged_resp = String::from_utf8_lossy(&forged_bytes);
+        assert!(
+            forged_resp.starts_with("HTTP/1.1 401 Unauthorized"),
+            "forged WS signature must be rejected with 401: {forged_resp}"
+        );
+
+        // 3. Signed WS connection authenticates caller and retrieves the private task
+        let mut auth_stream = connect_ws_signed(addr, &keypair).await;
+        ws_send_text(&mut auth_stream, query).await;
+        let auth_frames = ws_recv_op_frames(&mut auth_stream, "1").await;
+        let auth_resp = auth_frames.join("\n");
+        assert!(
+            auth_resp.contains("ws-auth-task-01"),
+            "signed WS must authenticate delegator and return private task: {auth_resp}"
+        );
+    }
+
     use axum::body::Body;
     use axum::http::{Request, StatusCode};
-    use sqlx::PgPool;
     use tower::ServiceExt;
 
     use crate::db::RepoRecord;
-    use crate::test_support::test_state;
 
     /// Regression: anonymous callers must not see the pin/anchor index (#121, #134).
     #[sqlx::test]

crates/gitlawb-node/src/state.rs

@@ -59,6 +59,10 @@ pub struct AppState {
     pub task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
     /// GraphQL schema (queries + mutations + subscriptions)
     pub graphql_schema: Arc<crate::graphql::GitlawbSchema>,
+    /// MAC key for task-list continuation tokens, derived from the node
+    /// keypair seed. Held here (and handed to the GraphQL schema) so REST and
+    /// GraphQL mint and accept the same tokens.
+    pub task_cursor_key: crate::api::task_cursor::TaskCursorKey,
     /// Fly.io machine ID — used for fly-replay routing in multi-machine deployments
     pub machine_id: Option<String>,
     /// Centralized repo storage: local disk cache + optional Tigris backend
@@ -326,6 +330,16 @@ pub struct AppState {
     /// (`with_default_max_keys`, reject-before-insert) so a source-key farm cannot grow
     /// it (INV-15).
     pub git_ipfs_walk_per_caller: crate::rate_limit::PerCallerConcurrency,
+    /// Per-client-IP rate limiter for the task read routes. `GET /api/v1/tasks`
+    /// and `GET /api/v1/tasks/{id}` are anonymously reachable and run the #268
+    /// visibility gate (a task lookup plus deduped-repo and visibility-rule
+    /// queries) before returning the opaque 404, so an unauthenticated prober
+    /// costs the node real work per request whether or not anything is visible.
+    /// Keyed on the resolved client IP via `push_limiter_trust`. Layered on
+    /// `task_read_routes` via `rate_limit_by_ip`.
+    /// The same per-IP hourly budget applies to GraphQL `tasks` / `task`
+    /// queries and WebSocket task queries via `TaskReadBrake`.
+    pub task_read_rate_limiter: RateLimiter,
     /// The `git` executable the served-git withheld-blob walk spawns. Production is
     /// `"git"` (resolved via PATH); injectable so a fake `git` can drive the walk's
     /// process-group teardown in handler tests without mutating the process-global
@@ -354,6 +368,7 @@ impl AppState {
         self.ipfs_work_rate_limiter.cleanup().await;
         self.sync_trigger_rate_limiter.cleanup().await;
         self.peer_write_rate_limiter.cleanup().await;
+        self.task_read_rate_limiter.cleanup().await;
         self.federated_rate_limiter.cleanup().await;
     }
 

crates/gitlawb-node/src/test_support.rs

@@ -86,10 +86,12 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
     let node_did = keypair.did();
     let (ref_tx, _) = tokio::sync::broadcast::channel(1);
     let (task_tx, _) = tokio::sync::broadcast::channel(1);
+    let task_cursor_key = crate::api::task_cursor::TaskCursorKey::derive(&keypair.to_seed());
     let schema = Arc::new(graphql::build_schema(
         db.clone(),
         ref_tx.clone(),
         task_tx.clone(),
+        task_cursor_key.clone(),
     ));
     AppState {
         config: Arc::new(Config::parse_from(["gitlawb-node"])),
@@ -101,6 +103,7 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         ref_update_tx: ref_tx,
         task_event_tx: task_tx,
         graphql_schema: schema,
+        task_cursor_key,
         machine_id: None,
         repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
         rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
@@ -139,6 +142,7 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
         git_ipfs_walk_per_caller: crate::rate_limit::PerCallerConcurrency::with_default_max_keys(
             16,
         ),
+        task_read_rate_limiter: RateLimiter::new(1200, Duration::from_secs(3600)),
         git_bin: "git".to_string(),
     }
 }
@@ -1262,7 +1266,7 @@ mod tests {
         // repo id afterwards — a 403 that still writes is the injection.
         let stored = state
             .db
-            .list_tasks(None, None, 50)
+            .list_tasks_keyset(None, None, 50, None)
             .await
             .expect("list tasks");
         assert!(
@@ -1621,11 +1625,11 @@ mod tests {
         let assignee = "did:key:zTASKASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
         let stranger = "did:key:zTASKSTRANGERCCCCCCCCCCCCCCCCCCCCCCCCCCC";
         let state = test_state(pool).await;
-        state
-            .db
-            .create_task(&seed_task("task-1", delegator))
-            .await
-            .expect("seed task");
+        let repo = seed_repo(delegator, "task-pub-repo");
+        state.db.create_repo(&repo).await.expect("seed repo");
+        let mut t1 = seed_task("task-1", delegator);
+        t1.repo_id = Some(repo.id);
+        state.db.create_task(&t1).await.expect("seed task");
         // Assignee claims it: pending -> claimed, assignee_did = assignee.
         state
             .db
@@ -1644,8 +1648,25 @@ mod tests {
         let uri = "/api/v1/tasks/task-1/complete";
         let body = || Body::from("{}");
 
-        // Stranger (not the assignee) is rejected by the authorization gate, even
-        // with the empty body that previously bypassed the binding. Exact 403.
+        // Stranger on an invisible (repo-less) task receives opaque 404 (no existence leak).
+        let inv_task = seed_task("task-inv", delegator);
+        state.db.create_task(&inv_task).await.unwrap();
+        let inv_resp = router()
+            .oneshot(signed_request_as(
+                stranger,
+                Method::POST,
+                "/api/v1/tasks/task-inv/complete",
+                body(),
+            ))
+            .await
+            .unwrap();
+        assert_eq!(
+            inv_resp.status(),
+            StatusCode::NOT_FOUND,
+            "an invisible task must 404 so existence is not leaked"
+        );
+
+        // Stranger on a visible task is rejected by the authorization gate with exact 403.
         let resp = router()
             .oneshot(signed_request_as(stranger, Method::POST, uri, body()))
             .await
@@ -1714,9 +1735,11 @@ mod tests {
             ))
             .await
             .unwrap();
+        // The claim gate hides existence: an ineligible claimer gets the same
+        // 404 as a missing task, not a 403 that would reveal the reservation.
         assert_eq!(
             resp.status(),
-            StatusCode::FORBIDDEN,
+            StatusCode::NOT_FOUND,
             "stranger must not steal a pre-assigned task, got {}",
             resp.status()
         );
@@ -1882,7 +1905,7 @@ mod tests {
         );
         let listed_bare = state
             .db
-            .list_tasks(Some("claimed"), Some(bare), 10)
+            .list_tasks_keyset(Some("claimed"), Some(bare), 10, None)
             .await
             .unwrap();
         assert!(
@@ -1891,12 +1914,14 @@ mod tests {
         );
         let listed_full = state
             .db
-            .list_tasks(Some("claimed"), Some(&full), 10)
+            .list_tasks_keyset(Some("claimed"), Some(&full), 10, None)
             .await
             .unwrap();
+        // The keyset filter normalizes to the did:key short form, so the
+        // claimer's full DID now matches the bare stored form too.
         assert!(
-            !listed_full.iter().any(|t| t.id == "task-form"),
-            "exact list by claimer's full DID must miss the bare stored form"
+            listed_full.iter().any(|t| t.id == "task-form"),
+            "normalized list by claimer's full DID must match the bare stored form"
         );
     }
 
@@ -1933,7 +1958,7 @@ mod tests {
         );
         let listed = state
             .db
-            .list_tasks(Some("claimed"), Some(&padded), 10)
+            .list_tasks_keyset(Some("claimed"), Some(&padded), 10, None)
             .await
             .unwrap();
         assert!(

crates/gitlawb-node/src/upstream_mirror.rs

@@ -0,0 +1,968 @@
+//! Scheduled external-forge fetches for canonical INBOUND mirrors.
+//!
+//! This worker is deliberately separate from `sync`, which mirrors repositories
+//! between Gitlawb peers. External upstream URLs are owner-controlled data that
+//! eventually reach `git fetch`, so each cycle resolves DNS itself, validates
+//! every address, and pins the approved answers into Git/libcurl with
+//! `http.curloptResolve`. Redirects, proxies, credential helpers, submodule
+//! recursion, and non-HTTPS protocols are disabled at the command boundary.
+
+use anyhow::{Context, Result};
+use std::collections::BTreeSet;
+use std::net::IpAddr;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tracing::{info, warn};
+
+use crate::config::Config;
+use crate::db::{Db, InboundMirrorTarget, MirrorStatus, MirrorTransitionPhase};
+use crate::git::repo_store::RepoStore;
+use crate::state::{repo_identity_key, RepoWriteLeases};
+
+const DNS_TIMEOUT: Duration = Duration::from_secs(10);
+const MAX_PINNED_ADDRESSES: usize = 32;
+const MIN_CURL_OPT_RESOLVE_GIT: (u64, u64) = (2, 37);
+
+#[derive(Debug, Clone, Copy)]
+struct EgressPolicy;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct PinnedUpstream {
+    url: String,
+    curlopt_resolve: Option<String>,
+}
+
+impl EgressPolicy {
+    async fn resolve(&self, url: reqwest::Url) -> Result<PinnedUpstream> {
+        let raw_host = url.host_str().context("mirror upstream URL has no host")?;
+        let connect_host = raw_host
+            .trim_start_matches('[')
+            .trim_end_matches(']')
+            .to_ascii_lowercase();
+        let policy_host = normalize_policy_host(&connect_host);
+        reject_always_local_hostname(&policy_host)?;
+        let port = url
+            .port_or_known_default()
+            .context("mirror upstream URL has no known port")?;
+
+        if let Ok(ip) = connect_host.parse::<IpAddr>() {
+            self.validate_addresses(&policy_host, &[ip])?;
+            return Ok(PinnedUpstream {
+                url: url.to_string(),
+                // An IP literal performs no DNS lookup, so there is no rebinding
+                // window to pin. TLS still verifies the literal from the URL.
+                curlopt_resolve: None,
+            });
+        }
+
+        let resolved = tokio::time::timeout(
+            DNS_TIMEOUT,
+            tokio::net::lookup_host((connect_host.as_str(), port)),
+        )
+        .await
+        .context("mirror upstream DNS lookup timed out")?
+        .context("resolving mirror upstream host")?;
+        let addresses: BTreeSet<IpAddr> = resolved.map(|addr| addr.ip()).collect();
+        if addresses.is_empty() {
+            anyhow::bail!("mirror upstream host resolved to no addresses");
+        }
+        let addresses: Vec<IpAddr> = addresses.into_iter().collect();
+        self.validate_addresses(&policy_host, &addresses)?;
+
+        let pinned = addresses
+            .iter()
+            .map(|ip| match ip {
+                IpAddr::V4(v4) => v4.to_string(),
+                IpAddr::V6(v6) => format!("[{v6}]"),
+            })
+            .collect::<Vec<_>>()
+            .join(",");
+        Ok(PinnedUpstream {
+            url: url.to_string(),
+            curlopt_resolve: Some(format!("+{connect_host}:{port}:{pinned}")),
+        })
+    }
+
+    fn validate_addresses(&self, host: &str, addresses: &[IpAddr]) -> Result<()> {
+        if addresses.is_empty() || addresses.len() > MAX_PINNED_ADDRESSES {
+            anyhow::bail!(
+                "mirror upstream host must resolve to 1..={MAX_PINNED_ADDRESSES} addresses"
+            );
+        }
+        for &ip in addresses {
+            if !address_is_permitted(ip) {
+                anyhow::bail!("mirror upstream host {host:?} resolved to disallowed address {ip}");
+            }
+        }
+        Ok(())
+    }
+}
+
+fn normalize_policy_host(host: &str) -> String {
+    host.trim_end_matches('.').to_ascii_lowercase()
+}
+
+fn reject_always_local_hostname(host: &str) -> Result<()> {
+    if host.is_empty()
+        || host == "localhost"
+        || host.ends_with(".localhost")
+        || host.ends_with(".local")
+        || host.ends_with(".internal")
+    {
+        anyhow::bail!("mirror upstream host is local-only and cannot be fetched");
+    }
+    Ok(())
+}
+
+fn address_is_permitted(ip: IpAddr) -> bool {
+    crate::api::peers::mirror_upstream_ip_permitted(ip)
+}
+
+fn parse_git_version(output: &str) -> Option<(u64, u64)> {
+    let version = output.split_whitespace().find(|part| {
+        part.as_bytes()
+            .first()
+            .is_some_and(|byte| byte.is_ascii_digit())
+    })?;
+    let mut parts = version.split('.');
+    Some((parts.next()?.parse().ok()?, parts.next()?.parse().ok()?))
+}
+
+fn ensure_safe_git_runtime(git_bin: &str) -> Result<()> {
+    // Bounded runner like every other git spawn on this path: --version does
+    // no I/O of its own, but the child still takes a permit and the deadline.
+    let (status, stdout, stderr) = crate::git::visibility_pack::run_bounded_git_raw_isolated_https(
+        git_bin,
+        &["--version"],
+        &std::env::temp_dir(),
+        b"",
+        Instant::now() + Duration::from_secs(10),
+    )?;
+    if !status.success() {
+        anyhow::bail!(
+            "{git_bin} --version failed: {}",
+            String::from_utf8_lossy(&stderr)
+                .chars()
+                .take(256)
+                .collect::<String>()
+        );
+    }
+    let stdout = String::from_utf8_lossy(&stdout);
+    let version = parse_git_version(&stdout).context("parsing git version")?;
+    if version < MIN_CURL_OPT_RESOLVE_GIT {
+        anyhow::bail!(
+            "upstream mirroring requires Git 2.37+ for DNS pinning; found {}.{}",
+            version.0,
+            version.1
+        );
+    }
+    Ok(())
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct FetchCommand {
+    config: Vec<(String, String)>,
+    args: Vec<String>,
+}
+
+fn fetch_command(upstream: &PinnedUpstream) -> FetchCommand {
+    let http_scope = |name: &str| format!("http.{}.{name}", upstream.url);
+    let credential_scope = |name: &str| format!("credential.{}.{name}", upstream.url);
+    let mut config = vec![
+        ("http.proxy".to_string(), "".to_string()),
+        ("http.followRedirects".to_string(), "false".to_string()),
+        // Reset any inherited multi-valued resolver entries before adding the
+        // address set validated for this exact fetch.
+        ("http.curloptResolve".to_string(), "".to_string()),
+        // Never forward operator or repository-scoped HTTP credentials to an
+        // owner-selected forge. Empty values reset Git's multi-valued headers
+        // and disable cookie state for this invocation.
+        ("http.extraHeader".to_string(), "".to_string()),
+        ("http.cookieFile".to_string(), "".to_string()),
+        ("http.saveCookies".to_string(), "false".to_string()),
+        ("http.sslVerify".to_string(), "true".to_string()),
+        // URL-specific configuration outranks generic `http.*` values even
+        // when the generic value is command-scoped. Repeat every security
+        // control at the exact validated URL so a repository-local scoped
+        // setting cannot re-enable redirects/proxies/headers/cookies or disable
+        // TLS. Key/value environment entries avoid `git -c` splitting a valid
+        // URL path that contains `=`.
+        (http_scope("proxy"), "".to_string()),
+        (http_scope("followRedirects"), "false".to_string()),
+        (http_scope("curloptResolve"), "".to_string()),
+        (http_scope("extraHeader"), "".to_string()),
+        (http_scope("cookieFile"), "".to_string()),
+        (http_scope("saveCookies"), "false".to_string()),
+        (http_scope("sslVerify"), "true".to_string()),
+    ];
+    if let Some(resolve) = &upstream.curlopt_resolve {
+        config.extend([
+            ("http.curloptResolve".to_string(), resolve.clone()),
+            (http_scope("curloptResolve"), resolve.clone()),
+        ]);
+    }
+    config.extend([
+        ("credential.helper".to_string(), "".to_string()),
+        (credential_scope("helper"), "".to_string()),
+        ("credential.interactive".to_string(), "false".to_string()),
+        ("core.askPass".to_string(), "false".to_string()),
+        ("fetch.recurseSubmodules".to_string(), "false".to_string()),
+        // Do not let a smart server or repository-local config introduce a
+        // second, unvalidated pack/bundle URL outside the pinned upstream.
+        ("fetch.uriprotocols".to_string(), "".to_string()),
+        ("fetch.bundleURI".to_string(), "".to_string()),
+        ("protocol.allow".to_string(), "never".to_string()),
+        ("protocol.https.allow".to_string(), "always".to_string()),
+    ]);
+    let args = vec![
+        "fetch".to_string(),
+        "--atomic".to_string(),
+        "--force".to_string(),
+        "--prune".to_string(),
+        "--prune-tags".to_string(),
+        "--no-auto-maintenance".to_string(),
+        "--no-recurse-submodules".to_string(),
+        "--no-write-fetch-head".to_string(),
+        upstream.url.clone(),
+        "+refs/heads/*:refs/heads/*".to_string(),
+        "+refs/tags/*:refs/tags/*".to_string(),
+    ];
+    FetchCommand { config, args }
+}
+
+fn reject_local_url_rewrite(
+    git_bin: &str,
+    repo_path: &Path,
+    upstream_url: &str,
+    deadline: Instant,
+) -> Result<()> {
+    let args = ["ls-remote", "--get-url", upstream_url];
+    let (status, stdout, stderr) = crate::git::visibility_pack::run_bounded_git_raw_isolated_https(
+        git_bin, &args, repo_path, b"", deadline,
+    )?;
+    if !status.success() {
+        let stderr = String::from_utf8_lossy(&stderr);
+        anyhow::bail!(
+            "checking mirror upstream URL rewrite failed: {}",
+            stderr.chars().take(4_096).collect::<String>()
+        );
+    }
+    let expanded = std::str::from_utf8(&stdout)
+        .context("Git returned a non-UTF-8 expanded mirror upstream URL")?
+        .trim_end_matches(['\r', '\n']);
+    if expanded != upstream_url {
+        anyhow::bail!("repository-local Git config attempted to rewrite the mirror upstream URL");
+    }
+    Ok(())
+}
+
+fn reject_sensitive_local_config(git_bin: &str, repo_path: &Path, deadline: Instant) -> Result<()> {
+    let args = [
+        "config",
+        "--local",
+        "--includes",
+        "--get-regexp",
+        r"^(http|credential|url)\.",
+    ];
+    let (status, stdout, stderr) = crate::git::visibility_pack::run_bounded_git_raw_isolated_https(
+        git_bin, &args, repo_path, b"", deadline,
+    )?;
+    match status.code() {
+        Some(1) if stdout.is_empty() => Ok(()),
+        Some(0) => anyhow::bail!(
+            "repository-local network or credential Git config is not allowed for upstream mirrors"
+        ),
+        _ => {
+            let stderr = String::from_utf8_lossy(&stderr);
+            anyhow::bail!(
+                "checking repository-local Git config failed: {}",
+                stderr.chars().take(4_096).collect::<String>()
+            )
+        }
+    }
+}
+
+/// Budget for each preflight probe inside [`run_fetch`] (the URL-rewrite
+/// check and the sensitive-local-config scan). Both run local git children
+/// against the on-disk repo, so a short bound is generous; keeping them on
+/// their own deadlines means `upstream_mirror_fetch_timeout_secs` covers
+/// only the external pack transfer it documents.
+const PREFLIGHT_PROBE_TIMEOUT: Duration = Duration::from_secs(30);
+
+fn run_fetch(
+    git_bin: &str,
+    repo_path: &Path,
+    upstream: &PinnedUpstream,
+    timeout: Duration,
+) -> Result<()> {
+    let command = fetch_command(upstream);
+    let args: Vec<&str> = command.args.iter().map(String::as_str).collect();
+    let config: Vec<(&str, &str)> = command
+        .config
+        .iter()
+        .map(|(key, value)| (key.as_str(), value.as_str()))
+        .collect();
+    let rewrite_deadline = Instant::now()
+        .checked_add(PREFLIGHT_PROBE_TIMEOUT)
+        .context("mirror URL-rewrite probe timeout is not representable")?;
+    reject_local_url_rewrite(git_bin, repo_path, &upstream.url, rewrite_deadline)?;
+    let config_deadline = Instant::now()
+        .checked_add(PREFLIGHT_PROBE_TIMEOUT)
+        .context("mirror local-config probe timeout is not representable")?;
+    reject_sensitive_local_config(git_bin, repo_path, config_deadline)?;
+    let deadline = Instant::now()
+        .checked_add(timeout)
+        .context("upstream mirror fetch timeout is not representable")?;
+    let (status, _stdout, stderr) =
+        crate::git::visibility_pack::run_bounded_git_raw_isolated_https_with_config(
+            git_bin, &args, repo_path, b"", deadline, &config,
+        )?;
+    if !status.success() {
+        let stderr = String::from_utf8_lossy(&stderr);
+        let stderr = stderr.chars().take(4_096).collect::<String>();
+        anyhow::bail!("git upstream fetch failed: {stderr}");
+    }
+    Ok(())
+}
+
+#[derive(Clone)]
+struct Worker {
+    db: Arc<Db>,
+    config: Arc<Config>,
+    repo_store: RepoStore,
+    repo_write_leases: RepoWriteLeases,
+    git_write_semaphore: Arc<tokio::sync::Semaphore>,
+    policy: EgressPolicy,
+    git_bin: String,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FetchOutcome {
+    Updated,
+    SkippedStale,
+}
+
+impl Worker {
+    async fn fetch_target(&self, target: InboundMirrorTarget) -> Result<FetchOutcome> {
+        let url = crate::db::validate_mirror_upstream_url(&target.upstream_url)?;
+        let upstream = self.policy.resolve(url).await?;
+
+        let repo_identity = repo_identity_key(&target.owner_did, &target.name);
+        let steal_after = Duration::from_secs(
+            self.config
+                .upstream_mirror_fetch_timeout_secs
+                .saturating_mul(2)
+                .saturating_add(60),
+        );
+        let _lease = self
+            .repo_write_leases
+            .acquire(&repo_identity, steal_after)
+            .await
+            .context("upstream mirror write-lease waiter cap reached")?;
+
+        let _write_permit = Arc::clone(&self.git_write_semaphore)
+            .acquire_owned()
+            .await
+            .context("upstream mirror write semaphore closed")?;
+        let acquire_timeout = Duration::from_secs(self.config.git_acquire_timeout_secs);
+        let guard = tokio::time::timeout(
+            acquire_timeout,
+            self.repo_store
+                .acquire_write(&target.owner_did, &target.name),
+        )
+        .await
+        .context("upstream mirror write-lock acquisition timed out")??;
+        let repo_path = guard.path().to_path_buf();
+
+        // A row can leave INBOUND state after selection. Recheck only after
+        // both the process-local lease and the cluster-wide advisory write lock
+        // are held. Every future transition path must take these same locks
+        // before changing authority, so no other node can commit a transition
+        // between this check and the fetch.
+        let current = match self.db.get_repo_mirror_state(&target.repo_id).await {
+            Ok(Some(current)) => current,
+            Ok(None) => {
+                let _ = guard.release(false).await;
+                anyhow::bail!("inbound mirror state disappeared before fetch");
+            }
+            Err(error) => {
+                let _ = guard.release(false).await;
+                return Err(error);
+            }
+        };
+        // Both stable shapes the CHECK admits: never transitioned, or a
+        // completed transition whose metadata is retained as audit. Only an
+        // in-flight transition makes the target stale.
+        let stable = (current.transition_id.is_none() && current.transition_phase.is_none())
+            || (current.transition_id.is_some()
+                && current.transition_phase == Some(MirrorTransitionPhase::Completed));
+        if current.status != MirrorStatus::Inbound
+            || !stable
+            || current.upstream_url != target.upstream_url
+        {
+            let _ = guard.release(false).await;
+            info!(repo_id = %target.repo_id, "upstream mirror state changed before fetch; skipping stale target");
+            return Ok(FetchOutcome::SkippedStale);
+        }
+
+        let git_bin = self.git_bin.clone();
+        let fetch_timeout = Duration::from_secs(self.config.upstream_mirror_fetch_timeout_secs);
+        let fetch = tokio::task::spawn_blocking(move || {
+            run_fetch(&git_bin, &repo_path, &upstream, fetch_timeout)
+        })
+        .await
+        .context("upstream mirror fetch task panicked")?;
+        let success = fetch.is_ok();
+        let outcome = guard.release(success).await;
+        // A refused publish (fenced or unknowable upload) must not report
+        // Updated; surface it as the failure it is before mapping the fetch.
+        outcome.into_result()?;
+        fetch.map(|()| FetchOutcome::Updated)
+    }
+
+    async fn scan_once(&self) {
+        let mut cursor: Option<String> = None;
+        loop {
+            let targets = match self
+                .db
+                .list_inbound_mirror_targets(
+                    cursor.as_deref(),
+                    self.config.upstream_mirror_page_size as i64,
+                )
+                .await
+            {
+                Ok(targets) => targets,
+                Err(error) => {
+                    warn!(err = %error, "failed to list inbound mirror targets");
+                    return;
+                }
+            };
+            if targets.is_empty() {
+                return;
+            }
+            let page_len = targets.len();
+            cursor = targets.last().map(|target| target.repo_id.clone());
+            for target in targets {
+                let repo_id = target.repo_id.clone();
+                let upstream_url = target.upstream_url.clone();
+                match self.fetch_target(target).await {
+                    Ok(FetchOutcome::Updated) => {
+                        info!(repo_id, upstream = %upstream_url, "upstream mirror fetch completed")
+                    }
+                    Ok(FetchOutcome::SkippedStale) => {}
+                    Err(error) => {
+                        warn!(repo_id, upstream = %upstream_url, err = %error, "upstream mirror fetch failed")
+                    }
+                }
+            }
+            if page_len < self.config.upstream_mirror_page_size {
+                return;
+            }
+        }
+    }
+}
+
+/// Validate the opt-in runtime and spawn the scheduled worker. The worker runs
+/// one scan immediately, sleeps only after a complete scan, and exits between
+/// scans when graceful shutdown is signalled. An in-flight Git child is bounded
+/// and reaped before the worker observes shutdown.
+pub fn start(
+    db: Arc<Db>,
+    config: Arc<Config>,
+    repo_store: RepoStore,
+    repo_write_leases: RepoWriteLeases,
+    git_write_semaphore: Arc<tokio::sync::Semaphore>,
+    mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
+) -> Result<()> {
+    ensure_safe_git_runtime("git")?;
+    let policy = EgressPolicy;
+    let worker = Worker {
+        db,
+        config: Arc::clone(&config),
+        repo_store,
+        repo_write_leases,
+        git_write_semaphore,
+        policy,
+        git_bin: "git".to_string(),
+    };
+    tokio::spawn(async move {
+        info!(
+            interval_secs = config.upstream_mirror_interval_secs,
+            "upstream mirror worker started"
+        );
+        loop {
+            worker.scan_once().await;
+            tokio::select! {
+                _ = tokio::time::sleep(Duration::from_secs(config.upstream_mirror_interval_secs)) => {}
+                changed = shutdown_rx.changed() => {
+                    if changed.is_err() || *shutdown_rx.borrow() {
+                        info!("upstream mirror worker stopped");
+                        return;
+                    }
+                }
+            }
+        }
+    });
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use clap::Parser;
+    use std::path::PathBuf;
+
+    async fn test_worker(
+        pool: sqlx::PgPool,
+        git_bin: PathBuf,
+        page_size: usize,
+    ) -> (Worker, Arc<Db>, tempfile::TempDir) {
+        let db = Arc::new(Db::for_testing(pool.clone()));
+        db.run_migrations().await.unwrap();
+        let repos = tempfile::tempdir().unwrap();
+        let mut config = Config::parse_from([
+            "gitlawb-node".to_string(),
+            "--upstream-mirror-enabled".to_string(),
+            "--enforce-owner-push".to_string(),
+            "--upstream-mirror-fetch-timeout-secs".to_string(),
+            "5".to_string(),
+            "--upstream-mirror-page-size".to_string(),
+            page_size.to_string(),
+        ]);
+        config.validate().unwrap();
+        let config = Arc::new(config);
+        let worker = Worker {
+            db: Arc::clone(&db),
+            config,
+            repo_store: RepoStore::for_testing(repos.path().to_path_buf(), pool),
+            repo_write_leases: RepoWriteLeases::new(8),
+            git_write_semaphore: Arc::new(tokio::sync::Semaphore::new(4)),
+            policy: EgressPolicy,
+            git_bin: git_bin.to_string_lossy().into_owned(),
+        };
+        (worker, db, repos)
+    }
+
+    async fn add_inbound_mirror(
+        worker: &Worker,
+        db: &Db,
+        id: &str,
+        name: &str,
+        upstream_url: &str,
+    ) -> InboundMirrorTarget {
+        let now = chrono::Utc::now();
+        let record = crate::db::RepoRecord {
+            id: id.to_string(),
+            name: name.to_string(),
+            owner_did: "did:key:z6MkMirrorWorker".to_string(),
+            description: None,
+            is_public: true,
+            default_branch: "main".to_string(),
+            created_at: now,
+            updated_at: now,
+            disk_path: format!("/test/{id}.git"),
+            forked_from: None,
+            machine_id: None,
+        };
+        db.create_repo(&record).await.unwrap();
+        worker
+            .repo_store
+            .init(&record.owner_did, &record.name)
+            .await
+            .unwrap();
+        let state = db
+            .configure_inbound_mirror(&record.id, upstream_url)
+            .await
+            .unwrap();
+        InboundMirrorTarget {
+            repo_id: record.id,
+            owner_did: record.owner_did,
+            name: record.name,
+            upstream_url: state.upstream_url,
+        }
+    }
+
+    #[cfg(unix)]
+    fn fake_git(repos: &tempfile::TempDir) -> (PathBuf, PathBuf) {
+        use std::os::unix::fs::PermissionsExt;
+
+        let git = repos.path().join("fake-git");
+        let calls = repos.path().join("fetch-calls");
+        let script = format!(
+            r#"#!/bin/sh
+if [ "$1" = "ls-remote" ]; then
+  printf '%s\n' "$3"
+  exit 0
+fi
+if [ "$1" = "config" ]; then
+  # Match real Git when no sensitive repository-local keys are configured.
+  exit 1
+fi
+case "$*" in
+  *fail.git*) printf '%s\n' fail >> '{}'; exit 7 ;;
+  *) printf '%s\n' ok >> '{}'; exit 0 ;;
+esac
+"#,
+            calls.display(),
+            calls.display()
+        );
+        std::fs::write(&git, script).unwrap();
+        let mut permissions = std::fs::metadata(&git).unwrap().permissions();
+        permissions.set_mode(0o700);
+        std::fs::set_permissions(&git, permissions).unwrap();
+        (git, calls)
+    }
+
+    #[test]
+    fn git_version_gate_requires_curlopt_resolve_support() {
+        assert_eq!(parse_git_version("git version 2.37.0"), Some((2, 37)));
+        assert_eq!(
+            parse_git_version("git version 2.39.5 (Apple Git-154)"),
+            Some((2, 39))
+        );
+        assert_eq!(parse_git_version("not git"), None);
+        assert!((2, 36) < MIN_CURL_OPT_RESOLVE_GIT);
+    }
+
+    #[test]
+    fn every_resolved_address_must_be_public() {
+        let policy = EgressPolicy;
+        let public_v4: IpAddr = "8.8.8.8".parse().unwrap();
+        let private_v4: IpAddr = "10.2.3.4".parse().unwrap();
+        let private_v6: IpAddr = "fd00::20".parse().unwrap();
+        assert!(policy
+            .validate_addresses("public.example", &[public_v4])
+            .is_ok());
+        assert!(policy
+            .validate_addresses("private.example", &[private_v4, private_v6])
+            .is_err());
+        assert!(policy
+            .validate_addresses("mixed.example", &[public_v4, private_v4])
+            .is_err());
+        let too_many = (1..=MAX_PINNED_ADDRESSES + 1)
+            .map(|last| IpAddr::V4(std::net::Ipv4Addr::new(8, 8, 0, last as u8)))
+            .collect::<Vec<_>>();
+        assert!(policy
+            .validate_addresses("oversized.example", &too_many)
+            .is_err());
+
+        for never in [
+            "127.0.0.1",
+            "169.254.169.254",
+            "192.0.2.1",
+            "198.18.0.1",
+            "198.51.100.1",
+            "203.0.113.10",
+            "::1",
+            "::ffff:198.18.0.1",
+            "2002:c612:1::",
+            "64:ff9b::c612:1",
+            "100::1",
+            "100:0:0:1::1",
+            "2001::1",
+            "2001:2::1",
+            "2001:db8::1",
+            "fec0::1",
+            "fe80::1",
+            "3fff::1",
+            "5f00::1",
+        ] {
+            let ip = never.parse().unwrap();
+            assert!(!address_is_permitted(ip), "{never} must stay denied");
+        }
+    }
+
+    #[test]
+    fn local_only_hostnames_are_always_rejected() {
+        for invalid in [
+            "",
+            "localhost",
+            "forge.localhost",
+            "forge.local",
+            "ghe.internal",
+        ] {
+            assert!(
+                reject_always_local_hostname(invalid).is_err(),
+                "{invalid:?}"
+            );
+        }
+        assert!(reject_always_local_hostname("github.example").is_ok());
+    }
+
+    #[test]
+    fn fetch_command_pins_dns_and_only_updates_branches_and_tags() {
+        let upstream = PinnedUpstream {
+            url: "https://github.example/org/repo.git".to_string(),
+            curlopt_resolve: Some("+github.example:443:8.8.8.8,[2606:4700:4700::1111]".to_string()),
+        };
+        let command = fetch_command(&upstream);
+        let has_config = |key: &str, value: &str| {
+            command
+                .config
+                .contains(&(key.to_string(), value.to_string()))
+        };
+        assert!(has_config("http.followRedirects", "false"));
+        assert!(has_config("http.proxy", ""));
+        assert!(has_config("http.extraHeader", ""));
+        assert!(has_config("http.cookieFile", ""));
+        assert!(has_config("http.saveCookies", "false"));
+        assert!(has_config("http.sslVerify", "true"));
+        assert!(has_config(
+            "http.curloptResolve",
+            "+github.example:443:8.8.8.8,[2606:4700:4700::1111]"
+        ));
+        assert!(has_config(
+            "http.https://github.example/org/repo.git.followRedirects",
+            "false"
+        ));
+        assert!(has_config(
+            "http.https://github.example/org/repo.git.proxy",
+            ""
+        ));
+        assert!(has_config(
+            "http.https://github.example/org/repo.git.extraHeader",
+            ""
+        ));
+        assert!(has_config(
+            "http.https://github.example/org/repo.git.curloptResolve",
+            "+github.example:443:8.8.8.8,[2606:4700:4700::1111]"
+        ));
+        assert!(has_config(
+            "credential.https://github.example/org/repo.git.helper",
+            ""
+        ));
+        assert!(has_config("protocol.allow", "never"));
+        assert!(has_config("protocol.https.allow", "always"));
+        assert!(has_config("fetch.uriprotocols", ""));
+        assert!(has_config("fetch.bundleURI", ""));
+        assert!(!command
+            .config
+            .iter()
+            .any(|(key, _)| key.ends_with("sslCert") || key.ends_with("sslKey")));
+        assert!(command.args.contains(&"--atomic".to_string()));
+        assert!(command
+            .args
+            .contains(&"+refs/heads/*:refs/heads/*".to_string()));
+        assert!(command
+            .args
+            .contains(&"+refs/tags/*:refs/tags/*".to_string()));
+        assert!(!command.args.iter().any(|arg| arg == "+refs/*:refs/*"));
+    }
+
+    /// Spawn `git` inside a fixture repo with inherited repository-location
+    /// and config environment scrubbed, so a test run inside an outer Git
+    /// checkout or custom config can never leak into the fixture.
+    fn fixture_git(dir: &std::path::Path) -> std::process::Command {
+        let mut cmd = std::process::Command::new("git");
+        cmd.current_dir(dir);
+        for var in [
+            "GIT_DIR",
+            "GIT_WORK_TREE",
+            "GIT_CONFIG",
+            "GIT_CONFIG_GLOBAL",
+            "GIT_CONFIG_SYSTEM",
+            "GIT_CONFIG_NOSYSTEM",
+            "GIT_CONFIG_COUNT",
+            "GIT_EXEC_PATH",
+            "GIT_SSH",
+            "GIT_SSH_COMMAND",
+            "GIT_ASKPASS",
+            "SSH_ASKPASS",
+        ] {
+            cmd.env_remove(var);
+        }
+        cmd
+    }
+
+    #[test]
+    fn repository_local_instead_of_rewrite_is_rejected() {
+        let repo = tempfile::tempdir().unwrap();
+        let init = fixture_git(repo.path())
+            .args(["init", "--bare"])
+            .output()
+            .unwrap();
+        assert!(init.status.success());
+        let config = fixture_git(repo.path())
+            .args([
+                "config",
+                "url.https://evil.example/.insteadOf",
+                "https://github.example/",
+            ])
+            .output()
+            .unwrap();
+        assert!(config.status.success());
+
+        let error = reject_local_url_rewrite(
+            "git",
+            repo.path(),
+            "https://github.example/org/repo.git",
+            Instant::now() + Duration::from_secs(5),
+        )
+        .unwrap_err();
+        assert!(error.to_string().contains("attempted to rewrite"));
+    }
+
+    #[test]
+    fn repository_local_network_and_credential_config_is_rejected() {
+        let repo = tempfile::tempdir().unwrap();
+        let init = fixture_git(repo.path())
+            .args(["init", "--bare"])
+            .output()
+            .unwrap();
+        assert!(init.status.success());
+
+        reject_sensitive_local_config("git", repo.path(), Instant::now() + Duration::from_secs(5))
+            .unwrap();
+
+        let config = fixture_git(repo.path())
+            .args(["config", "http.sslCert", "/tmp/untrusted-client.pem"])
+            .output()
+            .unwrap();
+        assert!(config.status.success());
+
+        let error = reject_sensitive_local_config(
+            "git",
+            repo.path(),
+            Instant::now() + Duration::from_secs(5),
+        )
+        .unwrap_err();
+        assert!(error.to_string().contains("network or credential"));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn hardened_real_git_reaches_https_transport_for_equals_path() {
+        let repo = tempfile::tempdir().unwrap();
+        let init = fixture_git(repo.path())
+            .args(["init", "--bare"])
+            .output()
+            .unwrap();
+        assert!(init.status.success());
+
+        // This is intentionally a loopback test seam around `run_fetch`, below
+        // the egress-policy layer. Reaching the listener proves the generated
+        // Git config neither dies on an empty client certificate nor splits the
+        // valid `=` path at the command-config boundary. The listener closes
+        // immediately, so the expected final result is a TLS transport error.
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        listener.set_nonblocking(true).unwrap();
+        let accepted = std::thread::spawn(move || {
+            let deadline = Instant::now() + Duration::from_secs(3);
+            while Instant::now() < deadline {
+                match listener.accept() {
+                    Ok((_stream, _address)) => return true,
+                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                        std::thread::sleep(Duration::from_millis(10));
+                    }
+                    Err(error) => panic!("accepting real-Git transport probe: {error}"),
+                }
+            }
+            false
+        });
+        let upstream = PinnedUpstream {
+            url: format!("https://127.0.0.1:{port}/org/repo=mirror.git"),
+            curlopt_resolve: None,
+        };
+        let error = run_fetch("git", repo.path(), &upstream, Duration::from_secs(5)).unwrap_err();
+        assert!(
+            accepted.join().unwrap(),
+            "real Git must reach the HTTPS transport; got {error:#}"
+        );
+        let error = error.to_string();
+        assert!(!error.contains("credential missing host field"));
+        assert!(!error.contains("invalid key"));
+    }
+
+    #[tokio::test]
+    async fn literal_addresses_are_checked_without_dns() {
+        let policy = EgressPolicy;
+        let allowed = policy
+            .resolve(reqwest::Url::parse("https://8.8.8.8/org/repo.git").unwrap())
+            .await
+            .unwrap();
+        assert_eq!(allowed.curlopt_resolve, None);
+
+        for denied in [
+            "127.0.0.1",
+            "10.2.3.4",
+            "169.254.169.254",
+            "192.0.2.1",
+            "198.18.0.1",
+            "203.0.113.10",
+            "[::1]",
+            "[100::1]",
+            "[100:0:0:1::1]",
+            "[2001::1]",
+            "[2001:db8::1]",
+            "[fec0::1]",
+            "[fd00::20]",
+            "[3fff::1]",
+            "[5f00::1]",
+        ] {
+            let url = reqwest::Url::parse(&format!("https://{denied}/org/repo.git")).unwrap();
+            assert!(
+                policy.resolve(url).await.is_err(),
+                "{denied} must stay denied"
+            );
+        }
+    }
+
+    #[sqlx::test]
+    async fn stale_target_is_rechecked_under_the_write_locks(pool: sqlx::PgPool) {
+        let missing_git = PathBuf::from("/definitely/missing/git");
+        let (worker, db, _repos) = test_worker(pool, missing_git, 1).await;
+        let target = add_inbound_mirror(
+            &worker,
+            &db,
+            "00000000-0000-0000-0000-000000000001",
+            "stale",
+            "https://8.8.8.8/org/stale.git",
+        )
+        .await;
+        sqlx::query(
+            "UPDATE repos
+             SET mirror_status = 'outbound'
+             WHERE id = $1",
+        )
+        .bind(&target.repo_id)
+        .execute(db.pool())
+        .await
+        .unwrap();
+
+        let outcome = worker.fetch_target(target).await.unwrap();
+        assert_eq!(outcome, FetchOutcome::SkippedStale);
+        assert!(worker.repo_write_leases.is_empty());
+    }
+
+    #[cfg(unix)]
+    #[sqlx::test]
+    async fn scan_pages_past_one_failure_and_fetches_the_next_repo(pool: sqlx::PgPool) {
+        let bootstrap = tempfile::tempdir().unwrap();
+        let (git, calls) = fake_git(&bootstrap);
+        let (worker, db, _repos) = test_worker(pool, git, 1).await;
+        add_inbound_mirror(
+            &worker,
+            &db,
+            "00000000-0000-0000-0000-000000000001",
+            "fails",
+            "https://8.8.8.8/org/fail.git",
+        )
+        .await;
+        add_inbound_mirror(
+            &worker,
+            &db,
+            "00000000-0000-0000-0000-000000000002",
+            "succeeds",
+            "https://8.8.8.8/org/ok.git",
+        )
+        .await;
+
+        worker.scan_once().await;
+
+        let calls = std::fs::read_to_string(calls).unwrap();
+        assert_eq!(calls.lines().collect::<Vec<_>>(), ["fail", "ok"]);
+        assert!(worker.repo_write_leases.is_empty());
+    }
+}

crates/gl/src/identity.rs

@@ -99,6 +99,27 @@ pub fn load_keypair_from_dir(dir: Option<&std::path::Path>) -> Result<Keypair> {
     Keypair::from_pem(&pem).context("failed to load keypair from PEM")
 }
 
+/// Load keypair from an optional directory override, distinguishing missing default
+/// identity (clean anonymous) from an explicit directory error or corrupt PEM.
+pub fn load_optional_keypair(dir: Option<&std::path::Path>) -> Result<Option<Keypair>> {
+    if let Some(d) = dir {
+        return load_keypair_from_dir(Some(d)).map(Some);
+    }
+    let base = match dirs::home_dir() {
+        Some(h) => h.join(".gitlawb"),
+        None => return Ok(None),
+    };
+    let path = key_path(&base);
+    match fs::read_to_string(&path) {
+        Ok(pem) => Keypair::from_pem(&pem)
+            .context("failed to load keypair from PEM")
+            .map(Some),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
+        Err(e) => Err(anyhow::Error::from(e)
+            .context(format!("failed to read identity from {}", path.display()))),
+    }
+}
+
 async fn cmd_new(dir: Option<PathBuf>, force: bool) -> Result<()> {
     cmd_new_with_reader(dir, force, &mut std::io::stdin().lock()).await
 }

crates/gl/src/mcp.rs

@@ -43,7 +43,6 @@ use std::io::{BufRead, Write};
 use std::path::PathBuf;
 
 use crate::http::NodeClient;
-use crate::identity::load_keypair_from_dir;
 
 #[derive(Args)]
 pub struct McpArgs {
@@ -516,7 +515,8 @@ fn tool_definitions() -> Value {
                 "properties": {
                     "status": { "type": "string", "description": "Filter by status: pending, claimed, completed, failed" },
                     "assignee_did": { "type": "string", "description": "Filter by assignee DID" },
-                    "limit": { "type": "integer", "description": "Max results (default: 50)", "default": 50 }
+                    "limit": { "type": "integer", "minimum": 1, "description": "Total results to return (default: 50). Must be positive; a non-positive limit is rejected rather than answered with an empty list. Values above the node's 200-row page cap are gathered by following continuation tokens.", "default": 50 },
+                    "cursor": { "type": "string", "description": "Resume from a previous call's next_cursor. Must be paired with the same status/assignee_did filter that produced it." }
                 }
             }
         },
@@ -640,7 +640,14 @@ async fn call_tool(
     node: &str,
     dir: Option<&std::path::Path>,
 ) -> Result<String> {
-    let keypair = load_keypair_from_dir(dir).ok();
+    // An explicit dir with no identity file is a clean anonymous caller
+    // (main's unsigned `webhook_list` path lets the node answer the denial);
+    // a present but unloadable identity is an error, never a silent
+    // downgrade to anonymous (this PR's `load_optional_keypair` contract).
+    let keypair = match dir {
+        Some(d) if !d.join("identity.pem").exists() => None,
+        _ => crate::identity::load_optional_keypair(dir)?,
+    };
     let client = NodeClient::new(node, keypair.clone());
 
     match name {
@@ -1114,16 +1121,31 @@ async fn call_tool(
 
         // ── Task tools ────────────────────────────────────────────────────
         "task_list" => {
-            let limit = args["limit"].as_i64().unwrap_or(50);
-            let mut path = format!("/api/v1/tasks?limit={limit}");
-            if let Some(s) = args.get("status").and_then(|v| v.as_str()) {
-                path.push_str(&format!("&status={}", urlencoding::encode(s)));
+            // Follows the node's opaque `next_cursor` so a limit above the
+            // 200-row server page cap returns what was asked for instead of a
+            // silently truncated page, and reports an explicit incomplete
+            // result when a guard or the node's scan ceiling stops it (#327).
+            let limit = match args.get("limit") {
+                None | Some(Value::Null) => 50,
+                Some(val) => val.as_i64().context("invalid limit: expected an integer")?,
+            };
+            let cursor = match args.get("cursor") {
+                None | Some(Value::Null) => None,
+                Some(val) => Some(val.as_str().context("invalid cursor: expected a string")?),
+            };
+            let result = crate::task::fetch_tasks(
+                &client,
+                args.get("status").and_then(|v| v.as_str()),
+                args.get("assignee_did").and_then(|v| v.as_str()),
+                limit,
+                cursor,
+            )
+            .await?;
+            let mut out = result.to_json();
+            if let Some(warning) = result.truncation_warning() {
+                out["warning"] = Value::String(warning);
             }
-            if let Some(a) = args.get("assignee_did").and_then(|v| v.as_str()) {
-                path.push_str(&format!("&assignee_did={}", urlencoding::encode(a)));
-            }
-            let resp = crate::http::read_json(client.get(&path).await?, "list tasks").await?;
-            Ok(serde_json::to_string_pretty(&resp)?)
+            Ok(serde_json::to_string_pretty(&out)?)
         }
 
         "task_create" => {
@@ -1726,7 +1748,7 @@ mod tests {
             )
             .with_status(200)
             .with_header("content-type", "application/json")
-            .with_body(r#"{"tasks":[{"id":"t1","kind":"test","status":"pending"}]}"#)
+            .with_body(r#"{"tasks":[{"id":"t1","kind":"test","status":"pending"}],"has_more":false,"incomplete":false,"next_cursor":null}"#)
             .create_async()
             .await;
 
@@ -1740,6 +1762,275 @@ mod tests {
         .unwrap();
         let parsed: Value = serde_json::from_str(&result).unwrap();
         assert_eq!(parsed["tasks"][0]["id"], "t1");
+        assert_eq!(parsed["complete"], json!(true));
+    }
+
+    #[tokio::test]
+    async fn test_task_list_via_mcp_uses_loaded_identity() {
+        let mut server = mockito::Server::new_async().await;
+        let dir = tempfile::TempDir::new().unwrap();
+        let kp = gitlawb_core::identity::Keypair::generate();
+        std::fs::write(
+            dir.path().join("identity.pem"),
+            kp.to_pem().unwrap().as_bytes(),
+        )
+        .unwrap();
+        let _m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"/api/v1/tasks\?".to_string()),
+            )
+            .match_header("signature", mockito::Matcher::Any)
+            .match_header("signature-input", mockito::Matcher::Any)
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(
+                r#"{"tasks":[{"id":"t1"}],"has_more":false,"incomplete":false,"next_cursor":null}"#,
+            )
+            .create_async()
+            .await;
+
+        let result = call_tool("task_list", json!({}), &server.url(), Some(dir.path()))
+            .await
+            .unwrap();
+        let parsed: Value = serde_json::from_str(&result).unwrap();
+        assert_eq!(parsed["tasks"][0]["id"], "t1");
+        assert_eq!(parsed["complete"], json!(true));
+    }
+
+    /// #327 review: MCP `task_list` issued one request and exposed no
+    /// continuation, so a limit above the node's 200-row page cap returned a
+    /// silently truncated result the model had no way to detect. It now
+    /// follows cursors and, when a guard stops it, says so in the payload.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_follows_cursors() {
+        let mut server = mockito::Server::new_async().await;
+        let first = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"^/api/v1/tasks\?limit=200$".to_string()),
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(
+                r#"{"tasks":[{"id":"t1"}],"has_more":true,"incomplete":false,"next_cursor":"c1"}"#,
+            )
+            .create_async()
+            .await;
+        let second = server
+            .mock("GET", mockito::Matcher::Regex(r"cursor=c1".to_string()))
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"tasks":[{"id":"t2"}],"has_more":false,"next_cursor":null}"#)
+            .create_async()
+            .await;
+
+        let result = call_tool("task_list", json!({"limit": 500}), &server.url(), None)
+            .await
+            .unwrap();
+        first.assert_async().await;
+        second.assert_async().await;
+        let parsed: Value = serde_json::from_str(&result).unwrap();
+        assert_eq!(parsed["count"], 2);
+        assert_eq!(parsed["tasks"][1]["id"], "t2");
+        assert_eq!(parsed["complete"], json!(true));
+        assert!(parsed.get("warning").is_none());
+    }
+
+    /// A truncated MCP result must carry an explicit warning and a resume
+    /// cursor, so the model cannot read it as the whole answer.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_flags_incomplete_results() {
+        let mut server = mockito::Server::new_async().await;
+        let _m = server
+            .mock("GET", mockito::Matcher::Any)
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(
+                r#"{"tasks":[{"id":"t1"}],"has_more":true,"incomplete":true,"next_cursor":"resume-me"}"#,
+            )
+            .create_async()
+            .await;
+
+        let result = call_tool("task_list", json!({"limit": 1}), &server.url(), None)
+            .await
+            .unwrap();
+        let parsed: Value = serde_json::from_str(&result).unwrap();
+        assert_eq!(parsed["complete"], json!(false));
+        assert_eq!(parsed["incomplete"], json!(true));
+        assert_eq!(parsed["next_cursor"], "resume-me");
+        assert!(
+            parsed["warning"]
+                .as_str()
+                .unwrap()
+                .contains("result incomplete"),
+            "{parsed}"
+        );
+    }
+
+    /// A legacy response without pagination metadata is reported as incomplete via MCP.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_legacy_response_is_incomplete() {
+        let mut server = mockito::Server::new_async().await;
+        let _m = server
+            .mock("GET", mockito::Matcher::Any)
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"tasks":[{"id":"t1"}],"count":1}"#)
+            .create_async()
+            .await;
+
+        let result = call_tool("task_list", json!({"limit": 50}), &server.url(), None)
+            .await
+            .unwrap();
+        let parsed: Value = serde_json::from_str(&result).unwrap();
+        assert_eq!(parsed["complete"], json!(false));
+        assert_eq!(parsed["incomplete"], json!(true));
+        assert!(parsed["next_cursor"].is_null());
+        assert!(
+            parsed["warning"]
+                .as_str()
+                .unwrap()
+                .contains("node does not support pagination metadata"),
+            "{parsed}"
+        );
+    }
+
+    /// A legacy response with limit > 200 stops after 1 page and does not claim completeness via MCP.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_legacy_limit_above_page_cap() {
+        let mut server = mockito::Server::new_async().await;
+        let ids: Vec<String> = (0..200).map(|i| format!("t{i}")).collect();
+        let tasks_json: Vec<Value> = ids.iter().map(|id| json!({ "id": id })).collect();
+        let body = json!({ "tasks": tasks_json, "count": 200 }).to_string();
+
+        let m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"^/api/v1/tasks\?limit=200$".to_string()),
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(body)
+            .expect(1)
+            .create_async()
+            .await;
+
+        let result = call_tool("task_list", json!({"limit": 500}), &server.url(), None)
+            .await
+            .unwrap();
+        m.assert_async().await;
+        let parsed: Value = serde_json::from_str(&result).unwrap();
+        assert_eq!(parsed["count"], 200);
+        assert_eq!(parsed["complete"], json!(false));
+        assert_eq!(parsed["incomplete"], json!(true));
+        assert!(parsed["next_cursor"].is_null());
+        assert!(
+            parsed["warning"]
+                .as_str()
+                .unwrap()
+                .contains("node does not support pagination metadata"),
+            "{parsed}"
+        );
+    }
+
+    /// A cursor the model passes back must reach the node.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_forwards_cursor_argument() {
+        let mut server = mockito::Server::new_async().await;
+        let m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"cursor=given-cursor".to_string()),
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"tasks":[],"has_more":false,"incomplete":false,"next_cursor":null}"#)
+            .expect(1)
+            .create_async()
+            .await;
+
+        call_tool(
+            "task_list",
+            json!({"cursor": "given-cursor"}),
+            &server.url(),
+            None,
+        )
+        .await
+        .unwrap();
+        m.assert_async().await;
+    }
+
+    #[tokio::test]
+    async fn test_task_list_via_mcp_returns_http_errors() {
+        let mut server = mockito::Server::new_async().await;
+        let _m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"/api/v1/tasks\?".to_string()),
+            )
+            .with_status(500)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"error":"failed"}"#)
+            .create_async()
+            .await;
+
+        let err = call_tool("task_list", json!({}), &server.url(), None)
+            .await
+            .unwrap_err();
+        assert!(err.to_string().contains("500"));
+    }
+
+    /// Malformed server responses fail visibly in MCP.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_malformed_response_errors() {
+        let mut server = mockito::Server::new_async().await;
+        let _m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"/api/v1/tasks\?".to_string()),
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"tasks":"not-an-array"}"#)
+            .create_async()
+            .await;
+
+        let err = call_tool("task_list", json!({}), &server.url(), None)
+            .await
+            .unwrap_err();
+        assert!(
+            err.to_string().contains("malformed") || err.to_string().contains("invalid JSON"),
+            "{err}"
+        );
+    }
+
+    /// #327 review: a model that sends `limit: 0` used to get an empty list
+    /// marked complete, which reads as "this node has no tasks". The shared
+    /// helper rejects it, so the model sees an invalid argument instead.
+    #[tokio::test]
+    async fn test_task_list_via_mcp_rejects_non_positive_limit() {
+        let mut server = mockito::Server::new_async().await;
+        let m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"/api/v1/tasks\?".to_string()),
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"tasks":[],"has_more":false,"incomplete":false,"next_cursor":null}"#)
+            .expect(0)
+            .create_async()
+            .await;
+
+        let err = call_tool("task_list", json!({"limit": 0}), &server.url(), None)
+            .await
+            .unwrap_err();
+        assert!(
+            err.to_string().contains("limit must be a positive"),
+            "{err}"
+        );
+        m.assert_async().await;
     }
 
     #[tokio::test]
@@ -2208,6 +2499,86 @@ mod tests {
         assert!(err.to_string().contains("no identity found"), "got: {err}");
     }
 
+    #[tokio::test]
+    async fn test_task_list_invalid_limit_rejected() {
+        let server = mockito::Server::new_async().await;
+        let dir = tempfile::TempDir::new().unwrap();
+        let kp = gitlawb_core::identity::Keypair::generate();
+        std::fs::write(
+            dir.path().join("identity.pem"),
+            kp.to_pem().unwrap().as_bytes(),
+        )
+        .unwrap();
+
+        let err = call_tool(
+            "task_list",
+            json!({"limit": "50"}),
+            &server.url(),
+            Some(dir.path()),
+        )
+        .await
+        .expect_err("string limit must be rejected");
+        assert!(
+            err.to_string()
+                .contains("invalid limit: expected an integer"),
+            "got: {err}"
+        );
+
+        let err_float = call_tool(
+            "task_list",
+            json!({"limit": 50.5}),
+            &server.url(),
+            Some(dir.path()),
+        )
+        .await
+        .expect_err("float limit must be rejected");
+        assert!(
+            err_float
+                .to_string()
+                .contains("invalid limit: expected an integer"),
+            "got: {err_float}"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_task_list_invalid_cursor_rejected() {
+        let server = mockito::Server::new_async().await;
+        for invalid in [json!(123), json!({}), json!([]), json!(true)] {
+            let err = call_tool("task_list", json!({"cursor": invalid}), &server.url(), None)
+                .await
+                .expect_err("non-string cursor must be rejected");
+            assert!(
+                err.to_string()
+                    .contains("invalid cursor: expected a string"),
+                "for {invalid}: got {err}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_task_list_null_or_absent_cursor_accepted() {
+        let mut server = mockito::Server::new_async().await;
+        let _m = server
+            .mock(
+                "GET",
+                mockito::Matcher::Regex(r"/api/v1/tasks\?".to_string()),
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"tasks":[],"has_more":false,"incomplete":false,"next_cursor":null}"#)
+            .expect(2)
+            .create_async()
+            .await;
+
+        call_tool("task_list", json!({}), &server.url(), None)
+            .await
+            .unwrap();
+
+        call_tool("task_list", json!({"cursor": null}), &server.url(), None)
+            .await
+            .unwrap();
+    }
+
     #[test]
     fn test_tool_count_is_42() {
         let tools = tool_definitions();