fix(node): mirror assignee normalization across GraphQL task mutations

GraphQL createTask returned the pre-insert input, so a whitespace-only
assignee came back as a reservation the store never kept; complete and
fail compared the caller against the raw stored assignee where REST and
Db::finish_task trim; and task-event broadcasts reported the caller
string rather than the stored assignee form REST publishes. All three
paths now mirror the REST and DB contract: the create response is the
stored row, finish gates compare the trimmed stored assignee, and
by_did on claim/complete/fail broadcasts derives from the returned row.

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

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

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

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

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

@@ -4000,28 +4000,54 @@ impl Db {
 
 // ── Agent Tasks ───────────────────────────────────────────────────────────────
 
+/// Permanent authorization denial: the task is reserved for a different assignee.
+/// Handlers downcast this to return 403 (vs 409 for a lost claim race).
+#[derive(Debug, thiserror::Error)]
+#[error("task not claimable: reserved for another assignee")]
+pub struct TaskReservedForOtherAssignee;
+
+/// ASCII whitespace treated as blank for assignee slots — must stay in sync with
+/// `BTRIM(assignee_did, E' \t\n\r')` in `claim_task`'s SQL open-slot check.
+const ASSIGNEE_BLANK: &[char] = &[' ', '\t', '\n', '\r'];
+
+fn assignee_slot_blank(s: &str) -> bool {
+    s.trim_matches(ASSIGNEE_BLANK).is_empty()
+}
+
+/// Trim assignee DIDs for auth matching (same ASCII blank set as SQL/open checks).
+pub fn trim_assignee_did(s: &str) -> &str {
+    s.trim_matches(ASSIGNEE_BLANK)
+}
+
 impl Db {
-    pub async fn create_task(&self, task: &AgentTask) -> Result<()> {
+    /// Persist a task. Whitespace-only `assignee_did` is stored as `NULL` (open).
+    /// Returns the normalized row shape so REST/GraphQL create responses match GET.
+    pub async fn create_task(&self, task: &AgentTask) -> Result<AgentTask> {
+        let mut stored = task.clone();
+        stored.assignee_did = stored
+            .assignee_did
+            .take()
+            .filter(|s| !assignee_slot_blank(s));
         sqlx::query(
             "INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
              VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)",
         )
-        .bind(&task.id)
-        .bind(&task.repo_id)
-        .bind(&task.kind)
-        .bind(&task.status)
-        .bind(&task.delegator_did)
-        .bind(&task.assignee_did)
-        .bind(&task.capability)
-        .bind(&task.ucan_token)
-        .bind(&task.payload)
-        .bind(&task.result)
-        .bind(&task.created_at)
-        .bind(&task.updated_at)
-        .bind(&task.deadline)
+        .bind(&stored.id)
+        .bind(&stored.repo_id)
+        .bind(&stored.kind)
+        .bind(&stored.status)
+        .bind(&stored.delegator_did)
+        .bind(&stored.assignee_did)
+        .bind(&stored.capability)
+        .bind(&stored.ucan_token)
+        .bind(&stored.payload)
+        .bind(&stored.result)
+        .bind(&stored.created_at)
+        .bind(&stored.updated_at)
+        .bind(&stored.deadline)
         .execute(&self.pool)
         .await?;
-        Ok(())
+        Ok(stored)
     }
 
     pub async fn get_task(&self, id: &str) -> Result<Option<AgentTask>> {
@@ -4078,38 +4104,102 @@ impl Db {
         Ok(rows.into_iter().map(row_to_task).collect())
     }
 
+    /// Claim a pending task for `assignee_did`.
+    ///
+    /// If the task was created with a non-blank pre-set `assignee_did`, only that
+    /// agent (DID-normalized via [`crate::api::did_matches`]) may claim it. Open
+    /// tasks (`NULL` / blank) remain first-claimer-wins. A reserved claim keeps
+    /// the **exact** stored DID form via `COALESCE($4, $2)` (no BTRIM rewrite) so
+    /// exact-match list filters still work. The UPDATE also re-checks the
+    /// assignee slot as defense-in-depth against a future writer racing the
+    /// pre-check.
     pub async fn claim_task(&self, id: &str, assignee_did: &str) -> Result<AgentTask> {
         let now = Utc::now().to_rfc3339();
+        // Denial path only needs status + assignee — do not load payload/UCAN
+        // for a permissionless caller who will be rejected.
+        let row = sqlx::query("SELECT status, assignee_did FROM agent_tasks WHERE id = $1")
+            .bind(id)
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("task not claimable: not found or already claimed"))?;
+        let status: String = row.get("status");
+        if status != "pending" {
+            return Err(anyhow::anyhow!(
+                "task not claimable: not found or already claimed"
+            ));
+        }
+        let stored: Option<String> = row.get("assignee_did");
+        // Blank reservations are treated as open. Keep the exact stored string
+        // for the UPDATE equality check and SET (do not BTRIM the reserved form).
+        // Blank definition matches SQL `BTRIM(..., E' \t\n\r')` (not space-only).
+        let reserved_exact = stored.as_deref().filter(|r| !assignee_slot_blank(r));
+        if let Some(reserved) = reserved_exact {
+            if !crate::api::did_matches(assignee_did, trim_assignee_did(reserved)) {
+                return Err(TaskReservedForOtherAssignee.into());
+            }
+        }
+        // Keep the exact reserved form ($4); only fill assignee on open claims ($2).
+        // Treat blank stored values as open for the SQL slot check.
         let row = sqlx::query(
-            "UPDATE agent_tasks SET status='claimed', assignee_did=$2, updated_at=$3
+            "UPDATE agent_tasks SET status='claimed',
+                 assignee_did = COALESCE($4, $2),
+                 updated_at=$3
              WHERE id=$1 AND status='pending'
+               AND (
+                 ($4::text IS NULL AND (assignee_did IS NULL OR BTRIM(assignee_did, E' \t\n\r') = ''))
+                 OR assignee_did = $4
+               )
              RETURNING id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline",
         )
         .bind(id)
         .bind(assignee_did)
         .bind(&now)
+        .bind(reserved_exact)
         .fetch_optional(&self.pool)
         .await?;
         row.map(row_to_task)
             .ok_or_else(|| anyhow::anyhow!("task not claimable: not found or already claimed"))
     }
 
+    /// Transition a claimed task to `new_status` (`completed` / `failed`).
+    ///
+    /// `actor_did` must be the task's assignee (DID-normalized). The UPDATE
+    /// binds the exact stored `assignee_did` so a concurrent reassignment /
+    /// re-claim cannot finish under a check-then-act race in the handler.
     pub async fn finish_task(
         &self,
         id: &str,
         new_status: &str,
         result: Option<&str>,
+        actor_did: &str,
     ) -> Result<AgentTask> {
         let now = Utc::now().to_rfc3339();
+        let existing = self
+            .get_task(id)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("task not found or not in claimed state"))?;
+        if existing.status != "claimed" {
+            return Err(anyhow::anyhow!("task not found or not in claimed state"));
+        }
+        let Some(ref assigned) = existing.assignee_did else {
+            return Err(anyhow::anyhow!("task not found or not in claimed state"));
+        };
+        // Trim matches claim_task so tab-padded stored values can finish.
+        if !crate::api::did_matches(actor_did, trim_assignee_did(assigned)) {
+            return Err(anyhow::anyhow!(
+                "task not finishable: only the assignee may finish it"
+            ));
+        }
         let row = sqlx::query(
             "UPDATE agent_tasks SET status=$2, result=$3, updated_at=$4
-             WHERE id=$1 AND status='claimed'
+             WHERE id=$1 AND status='claimed' AND assignee_did=$5
              RETURNING id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline",
         )
         .bind(id)
         .bind(new_status)
         .bind(result)
         .bind(&now)
+        .bind(assigned.as_str())
         .fetch_optional(&self.pool)
         .await?;
         row.map(row_to_task)

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

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

crates/gitlawb-node/src/test_support.rs

@@ -1526,6 +1526,451 @@ mod tests {
         );
     }
 
+    /// Pre-assigned tasks must not be stealable: a stranger claiming a pending
+    /// task reserved for another agent must fail, and must not receive the
+    /// UCAN. The reserved assignee still claims successfully.
+    #[sqlx::test]
+    async fn claim_task_honors_preassigned_assignee(pool: PgPool) {
+        let delegator = "did:key:zCLAIMDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let reserved = "did:key:zCLAIMRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let thief = "did:key:zCLAIMTHIEFCCCCCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool).await;
+        let mut task = seed_task("task-reserved", delegator);
+        task.assignee_did = Some(reserved.to_string());
+        task.ucan_token = Some("ucan-secret-for-reserved-only".into());
+        state.db.create_task(&task).await.expect("seed");
+
+        let router = || {
+            Router::new()
+                .route(
+                    "/api/v1/tasks/{id}/claim",
+                    axum::routing::post(crate::api::tasks::claim_task),
+                )
+                .with_state(state.clone())
+        };
+        let uri = "/api/v1/tasks/task-reserved/claim";
+
+        // Thief signs as themselves and asks to claim — must fail, UCAN stays put.
+        let resp = router()
+            .oneshot(signed_request_as(
+                thief,
+                Method::POST,
+                uri,
+                Body::from(format!(r#"{{"assignee_did":"{thief}"}}"#)),
+            ))
+            .await
+            .unwrap();
+        assert_eq!(
+            resp.status(),
+            StatusCode::FORBIDDEN,
+            "stranger must not steal a pre-assigned task, got {}",
+            resp.status()
+        );
+        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let text = String::from_utf8_lossy(&body);
+        assert!(
+            !text.contains("ucan-secret-for-reserved-only"),
+            "UCAN must not leak on a failed steal: {text}"
+        );
+        let still = state.db.get_task("task-reserved").await.unwrap().unwrap();
+        assert_eq!(still.status, "pending");
+        assert_eq!(still.assignee_did.as_deref(), Some(reserved));
+
+        // Reserved assignee claims successfully and receives the UCAN.
+        let resp = router()
+            .oneshot(signed_request_as(
+                reserved,
+                Method::POST,
+                uri,
+                Body::from(format!(r#"{{"assignee_did":"{reserved}"}}"#)),
+            ))
+            .await
+            .unwrap();
+        assert!(
+            resp.status().is_success(),
+            "reserved assignee must claim, got {}",
+            resp.status()
+        );
+        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let text = String::from_utf8_lossy(&body);
+        assert!(
+            text.contains("ucan-secret-for-reserved-only"),
+            "reserved assignee must receive UCAN: {text}"
+        );
+        assert!(text.contains("\"status\":\"claimed\"") || text.contains("claimed"));
+    }
+
+    /// Open (unassigned) pending tasks remain first-claimer-wins after the
+    /// pre-assignment gate.
+    #[sqlx::test]
+    async fn claim_task_open_still_first_claimer_wins(pool: PgPool) {
+        let delegator = "did:key:zOPENDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let claimer = "did:key:zOPENCLAIMERBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let state = test_state(pool).await;
+        state
+            .db
+            .create_task(&seed_task("task-open", delegator))
+            .await
+            .expect("seed");
+
+        let router = Router::new()
+            .route(
+                "/api/v1/tasks/{id}/claim",
+                axum::routing::post(crate::api::tasks::claim_task),
+            )
+            .with_state(state.clone());
+        let resp = router
+            .oneshot(signed_request_as(
+                claimer,
+                Method::POST,
+                "/api/v1/tasks/task-open/claim",
+                Body::from(format!(r#"{{"assignee_did":"{claimer}"}}"#)),
+            ))
+            .await
+            .unwrap();
+        assert!(
+            resp.status().is_success(),
+            "open task must still be claimable, got {}",
+            resp.status()
+        );
+        let got = state.db.get_task("task-open").await.unwrap().unwrap();
+        assert_eq!(got.status, "claimed");
+        assert_eq!(got.assignee_did.as_deref(), Some(claimer));
+    }
+
+    /// ASCII-blank assignees (`ASSIGNEE_BLANK`: space, tab, LF, CR) must be open
+    /// at the claim SQL blank branch. Seeds via raw INSERT so create_task
+    /// normalization is not in the path.
+    #[sqlx::test]
+    async fn claim_task_ascii_blank_reservations_are_open(pool: PgPool) {
+        let delegator = "did:key:zBLANKDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let claimer = "did:key:zBLANKCLAIMERBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let state = test_state(pool.clone()).await;
+        let now = chrono::Utc::now().to_rfc3339();
+        for (id, blank) in [
+            ("task-blank-empty", ""),
+            ("task-blank-tab", "\t"),
+            ("task-blank-space", " "),
+            ("task-blank-lf", "\n"),
+            ("task-blank-cr", "\r"),
+        ] {
+            sqlx::query(
+                "INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
+                 VALUES ($1,NULL,'build','pending',$2,$3,'repo:write',NULL,NULL,NULL,$4,$4,NULL)",
+            )
+            .bind(id)
+            .bind(delegator)
+            .bind(blank)
+            .bind(&now)
+            .execute(&pool)
+            .await
+            .unwrap_or_else(|e| panic!("seed {id}: {e}"));
+
+            let claimed = state
+                .db
+                .claim_task(id, claimer)
+                .await
+                .unwrap_or_else(|e| panic!("{id} must be open: {e}"));
+            assert_eq!(claimed.status, "claimed", "{id}");
+            assert_eq!(claimed.assignee_did.as_deref(), Some(claimer), "{id}");
+        }
+    }
+
+    /// create_task must null whitespace-only assignees and return that shape
+    /// (fail-on-remove for the blank filter).
+    #[sqlx::test]
+    async fn create_task_normalizes_whitespace_assignee(pool: PgPool) {
+        let delegator = "did:key:zCREATENORMDELEGAAAAAAAAAAAAAAAAAAAAAAAA";
+        let state = test_state(pool).await;
+        let mut task = seed_task("task-create-norm", delegator);
+        task.assignee_did = Some(" \t\n ".into());
+        let stored = state.db.create_task(&task).await.expect("create");
+        assert!(
+            stored.assignee_did.is_none(),
+            "create_task must return normalized NULL assignee, got {:?}",
+            stored.assignee_did
+        );
+        let got = state
+            .db
+            .get_task("task-create-norm")
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(got.assignee_did.is_none(), "row must store NULL assignee");
+    }
+
+    /// Claiming a reserved task must keep the stored DID form so exact-match
+    /// list filters still find it. Also covers bare→full claim + list by bare
+    /// (claimer full DID must not be required for list equality).
+    #[sqlx::test]
+    async fn claim_task_keeps_stored_assignee_did_form(pool: PgPool) {
+        let delegator = "did:key:zFORMDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let bare = "zFORMRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let full = format!("did:key:{bare}");
+        let state = test_state(pool).await;
+        let mut task = seed_task("task-form", delegator);
+        task.assignee_did = Some(bare.to_string());
+        state.db.create_task(&task).await.expect("seed");
+
+        let claimed = state
+            .db
+            .claim_task("task-form", &full)
+            .await
+            .expect("claim with full DID");
+        assert_eq!(
+            claimed.assignee_did.as_deref(),
+            Some(bare),
+            "reserved claim must keep the stored bare-key form"
+        );
+        let listed_bare = state
+            .db
+            .list_tasks(Some("claimed"), Some(bare), 10)
+            .await
+            .unwrap();
+        assert!(
+            listed_bare.iter().any(|t| t.id == "task-form"),
+            "delegator filtering by bare key must still find the task"
+        );
+        let listed_full = state
+            .db
+            .list_tasks(Some("claimed"), Some(&full), 10)
+            .await
+            .unwrap();
+        assert!(
+            !listed_full.iter().any(|t| t.id == "task-form"),
+            "exact list by claimer's full DID must miss the bare stored form"
+        );
+    }
+
+    /// Padded non-blank reservation must survive claim unchanged (exact form).
+    #[sqlx::test]
+    async fn claim_task_keeps_padded_reservation_exact(pool: PgPool) {
+        let delegator = "did:key:zPADDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let key = "zPADRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let padded = format!(" {key}\t");
+        let full = format!("did:key:{key}");
+        let state = test_state(pool.clone()).await;
+        let now = chrono::Utc::now().to_rfc3339();
+        sqlx::query(
+            "INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
+             VALUES ($1,NULL,'build','pending',$2,$3,'repo:write',NULL,NULL,NULL,$4,$4,NULL)",
+        )
+        .bind("task-padded")
+        .bind(delegator)
+        .bind(&padded)
+        .bind(&now)
+        .execute(&pool)
+        .await
+        .expect("seed padded");
+
+        let claimed = state
+            .db
+            .claim_task("task-padded", &full)
+            .await
+            .expect("claim padded reservation");
+        assert_eq!(
+            claimed.assignee_did.as_deref(),
+            Some(padded.as_str()),
+            "claim must not BTRIM a non-blank reservation"
+        );
+        let listed = state
+            .db
+            .list_tasks(Some("claimed"), Some(&padded), 10)
+            .await
+            .unwrap();
+        assert!(
+            listed.iter().any(|t| t.id == "task-padded"),
+            "exact list by original padded form must still find the task"
+        );
+
+        let finished = state
+            .db
+            .finish_task("task-padded", "completed", None, &full)
+            .await
+            .expect("padded assignee must finish after trim match");
+        assert_eq!(finished.status, "completed");
+    }
+
+    /// Direct coverage for the DB-layer finish_task assignee Rust gate (handlers
+    /// 403 before calling in, so HTTP tests alone leave this unproven).
+    #[sqlx::test]
+    async fn finish_task_rejects_non_assignee_at_db(pool: PgPool) {
+        let delegator = "did:key:zFINISHDELEGATORAAAAAAAAAAAAAAAAAAAAAAAA";
+        let assignee = "did:key:zFINISHASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let stranger = "did:key:zFINISHSTRANGERCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool).await;
+        state
+            .db
+            .create_task(&seed_task("task-finish-gate", delegator))
+            .await
+            .expect("seed");
+        state
+            .db
+            .claim_task("task-finish-gate", assignee)
+            .await
+            .expect("claim");
+
+        let err = state
+            .db
+            .finish_task("task-finish-gate", "completed", None, stranger)
+            .await
+            .expect_err("stranger must not finish");
+        assert!(
+            err.to_string().contains("assignee") || err.to_string().contains("finishable"),
+            "expected assignee denial, got {err}"
+        );
+        let still = state
+            .db
+            .get_task("task-finish-gate")
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            still.status, "claimed",
+            "row must stay claimed after denial"
+        );
+
+        let done = state
+            .db
+            .finish_task("task-finish-gate", "completed", None, assignee)
+            .await
+            .expect("assignee finishes");
+        assert_eq!(done.status, "completed");
+    }
+
+    /// Fail-on-remove for `finish_task`'s `AND assignee_did=$5`: a concurrent writer
+    /// that changes the stored assignee between the production read and UPDATE must
+    /// leave the task claimed (production `Db::finish_task` path).
+    #[sqlx::test]
+    async fn finish_task_sql_assignee_predicate_is_load_bearing(pool: PgPool) {
+        let delegator = "did:key:zFINISHSQLDELEGAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let assignee = "did:key:zFINISHSQLASSIGNBBBBBBBBBBBBBBBBBBBBBBBB";
+        let other = "did:key:zFINISHSQLOTHERCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool.clone()).await;
+        let id = "task-finish-sql";
+        state
+            .db
+            .create_task(&seed_task(id, delegator))
+            .await
+            .expect("seed");
+        state.db.claim_task(id, assignee).await.expect("claim");
+
+        let pool_bg = pool.clone();
+        let id_bg = id.to_string();
+        let other_bg = other.to_string();
+        let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
+        let hog = tokio::spawn(async move {
+            let mut tx = pool_bg.begin().await.expect("begin");
+            sqlx::query("SELECT assignee_did FROM agent_tasks WHERE id = $1 FOR UPDATE")
+                .bind(&id_bg)
+                .fetch_one(&mut *tx)
+                .await
+                .expect("lock row");
+            locked_tx.send(()).ok();
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            sqlx::query(
+                "UPDATE agent_tasks SET assignee_did = $2 WHERE id = $1 AND status = 'claimed'",
+            )
+            .bind(&id_bg)
+            .bind(&other_bg)
+            .execute(&mut *tx)
+            .await
+            .expect("race assignee");
+            tx.commit().await.expect("commit");
+        });
+        locked_rx.await.expect("row lock held before finish");
+
+        let err = state
+            .db
+            .finish_task(id, "completed", None, assignee)
+            .await
+            .expect_err("assignee slot race must fail production finish_task");
+        hog.await.expect("hog task");
+
+        assert!(
+            err.to_string().contains("not in claimed state"),
+            "expected finish denial, got {err}"
+        );
+        let still = state.db.get_task(id).await.unwrap().unwrap();
+        assert_eq!(still.status, "claimed");
+        assert_eq!(still.assignee_did.as_deref(), Some(other));
+    }
+
+    /// Fail-on-remove for the claim UPDATE assignee-slot re-check: a concurrent
+    /// writer that changes `assignee_did` after the pre-read must block claim.
+    #[sqlx::test]
+    async fn claim_task_update_assignee_slot_recheck_is_load_bearing(pool: PgPool) {
+        let delegator = "did:key:zCLAIMSLOTDELEGAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let reserved = "did:key:zCLAIMSLOTASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBB";
+        let other = "did:key:zCLAIMSLOTOTHERCCCCCCCCCCCCCCCCCCCCCCCCC";
+        let state = test_state(pool.clone()).await;
+        let id = "task-claim-slot";
+        let now = chrono::Utc::now().to_rfc3339();
+        sqlx::query(
+            "INSERT INTO agent_tasks
+                 (id, repo_id, kind, status, delegator_did, assignee_did, capability,
+                  ucan_token, payload, result, created_at, updated_at, deadline)
+             VALUES ($1, NULL, 'build', 'pending', $2, $3, 'repo:write',
+                     NULL, NULL, NULL, $4, $4, NULL)",
+        )
+        .bind(id)
+        .bind(delegator)
+        .bind(reserved)
+        .bind(&now)
+        .execute(&pool)
+        .await
+        .expect("seed reserved");
+
+        let pool_bg = pool.clone();
+        let id_bg = id.to_string();
+        let other_bg = other.to_string();
+        let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
+        let hog = tokio::spawn(async move {
+            let mut tx = pool_bg.begin().await.expect("begin");
+            sqlx::query("SELECT assignee_did FROM agent_tasks WHERE id = $1 FOR UPDATE")
+                .bind(&id_bg)
+                .fetch_one(&mut *tx)
+                .await
+                .expect("lock row");
+            locked_tx.send(()).ok();
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            sqlx::query(
+                "UPDATE agent_tasks SET assignee_did = $2 WHERE id = $1 AND status = 'pending'",
+            )
+            .bind(&id_bg)
+            .bind(&other_bg)
+            .execute(&mut *tx)
+            .await
+            .expect("race assignee");
+            tx.commit().await.expect("commit");
+        });
+        locked_rx.await.expect("row lock held before claim");
+
+        let err = state
+            .db
+            .claim_task(id, reserved)
+            .await
+            .expect_err("assignee slot race must fail production claim_task");
+        hog.await.expect("hog task");
+
+        assert!(
+            err.to_string().contains("not found or already claimed"),
+            "expected SQL re-check denial, got {err}"
+        );
+        assert!(
+            err.downcast_ref::<crate::db::TaskReservedForOtherAssignee>()
+                .is_none(),
+            "must prove the UPDATE predicate, not the Rust pre-check: {err}"
+        );
+        let still = state.db.get_task(id).await.unwrap().unwrap();
+        assert_eq!(still.status, "pending");
+        assert_eq!(still.assignee_did.as_deref(), Some(other));
+    }
+
     /// Adversarial-review GATE-2 (create_pr): opening a PR requires read access.
     /// A non-reader is denied on a private repo before any PR is created; the
     /// owner is allowed.