fix(node): honor pre-assigned task assignee on claim

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,21 +284,28 @@ 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| {
             (
                 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(by_did);
     let _ = state.task_event_tx.send(TaskEventBroadcast {
         task_id: id,
         old_status: "claimed".to_string(),
@@ -325,22 +344,29 @@ 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"));
     }
     let by_did = auth.0;
     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| {
             (
                 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(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)