From 1676eb0e4c2d028ffbf5fddab05f95f691a485b3 Mon Sep 17 00:00:00 2001 From: Ayush7614 Date: Mon, 10 Aug 2026 17:14:45 +0530 Subject: [PATCH 1/4] fix(node): honor pre-assigned task assignee on claim Stop strangers from overwriting a reserved assignee_did (and receiving the UCAN). Bind finish_task to the stored assignee; return 403 on reserved steals; keep exact stored DID forms; normalize blank assignees; pin production claim/finish SQL race guards with FOR UPDATE tests; trim broadcast by_did; and tighten blank/SQL re-check regressions. --- crates/gitlawb-node/src/api/tasks.rs | 38 +- crates/gitlawb-node/src/db/mod.rs | 124 +++++- crates/gitlawb-node/src/graphql/mutation.rs | 136 +++++- crates/gitlawb-node/src/test_support.rs | 445 ++++++++++++++++++++ 4 files changed, 710 insertions(+), 33 deletions(-) diff --git a/crates/gitlawb-node/src/api/tasks.rs b/crates/gitlawb-node/src/api/tasks.rs index de22134a..5139373a 100644 --- a/crates/gitlawb-node/src/api/tasks.rs +++ b/crates/gitlawb-node/src/api/tasks.rs @@ -117,7 +117,7 @@ pub async fn create_task( updated_at: now, deadline: body.deadline, }; - state.db.create_task(&task).await.map_err(|e| { + let task = state.db.create_task(&task).await.map_err(|e| { ( StatusCode::INTERNAL_SERVER_ERROR, Json(json!({ "error": e.to_string() })), @@ -175,16 +175,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::() + .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))) @@ -219,14 +231,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| { ( @@ -234,6 +246,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(), @@ -272,7 +291,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")); } @@ -280,7 +299,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| { ( @@ -288,6 +307,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(), diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 99c5d8c6..3e332889 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -2815,28 +2815,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 { + 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> { @@ -2893,38 +2919,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 { 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 = 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 { 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) diff --git a/crates/gitlawb-node/src/graphql/mutation.rs b/crates/gitlawb-node/src/graphql/mutation.rs index 7fb7a1dc..50d823b0 100644 --- a/crates/gitlawb-node/src/graphql/mutation.rs +++ b/crates/gitlawb-node/src/graphql/mutation.rs @@ -52,7 +52,8 @@ impl MutationRoot { updated_at: now, deadline: input.deadline, }; - db.create_task(&task) + let task = db + .create_task(&task) .await .map_err(crate::graphql::graphql_db_err)?; Ok(AgentTaskType::from(task)) @@ -73,15 +74,29 @@ impl MutationRoot { let assignee_did = caller.to_string(); let db = ctx.data_unchecked::>(); let tx = ctx.data_unchecked::>(); - let task = db - .claim_task(&id, &assignee_did) - .await - .map_err(crate::graphql::graphql_db_err)?; + let task = match db.claim_task(&id, &assignee_did).await { + Err(e) + if e.downcast_ref::() + .is_some() => + { + return Err(async_graphql::Error::new( + "task not claimable: reserved for another assignee", + )); + } + other => other.map_err(crate::graphql::graphql_db_err)?, + }; + 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(assignee_did); 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)) @@ -110,15 +125,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 complete it", )); } let task = db - .finish_task(&id, "completed", input.result.as_deref()) + .finish_task(&id, "completed", input.result.as_deref(), &by_did) .await .map_err(crate::graphql::graphql_db_err)?; + 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 _ = tx.send(TaskEventBroadcast { task_id: id, old_status: "claimed".to_string(), @@ -151,16 +176,26 @@ 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), &by_did) .await .map_err(crate::graphql::graphql_db_err)?; + 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 _ = tx.send(TaskEventBroadcast { task_id: id, old_status: "claimed".to_string(), @@ -332,4 +367,85 @@ mod tests { errors(&resp) ); } + + /// 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"); + } } diff --git a/crates/gitlawb-node/src/test_support.rs b/crates/gitlawb-node/src/test_support.rs index 2b5aef95..c4d13ea1 100644 --- a/crates/gitlawb-node/src/test_support.rs +++ b/crates/gitlawb-node/src/test_support.rs @@ -707,6 +707,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::() + .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. From 1e6b0657249cf57264253100053c839d0b981842 Mon Sep 17 00:00:00 2001 From: Ayush7614 Date: Tue, 11 Aug 2026 02:15:45 +0530 Subject: [PATCH 2/4] test(git-remote-gitlawb): wait for shim before real_git_fetch on Windows Windows CI can race: git fetch connects before the shim accept loop is polling, yielding connection aborted (os error 10053) with 0 POSTs. Probe the listener after spawn so multi-round and withheld-path tests start only once the shim is ready. --- crates/git-remote-gitlawb/tests/real_git_fetch.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/crates/git-remote-gitlawb/tests/real_git_fetch.rs b/crates/git-remote-gitlawb/tests/real_git_fetch.rs index 26d88f27..e1183192 100644 --- a/crates/git-remote-gitlawb/tests/real_git_fetch.rs +++ b/crates/git-remote-gitlawb/tests/real_git_fetch.rs @@ -128,6 +128,19 @@ impl Drop for Shim { } } +/// Wait until the shim accept loop is polling. Without this, Windows CI can +/// race: `git fetch` connects before the spawned thread enters `accept()`. +fn wait_for_shim_ready(addr: std::net::SocketAddr) { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if TcpStream::connect_timeout(&addr, Duration::from_millis(200)).is_ok() { + return; + } + std::thread::sleep(Duration::from_millis(25)); + } + panic!("shim at {addr} did not accept connections within 10s"); +} + /// Start a minimal smart-HTTP shim serving `repo` on 127.0.0.1. Handles the /// upload-pack advertisement (GET) and pack negotiation (POST), counting POSTs. fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim { @@ -154,6 +167,8 @@ fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim { } }); + wait_for_shim_ready(addr); + Shim { base_url, posts, From 0484740638b5ba10e5aab0a3f628f59f4431ea3d Mon Sep 17 00:00:00 2001 From: Ayush7614 Date: Tue, 11 Aug 2026 02:28:12 +0530 Subject: [PATCH 3/4] test(git-remote-gitlawb): retry transient Windows fetch connection errors Windows CI can still abort localhost connections (os error 10053) after the shim is ready. Retry fetch_with_helper up to three times on transient connection failures before asserting. --- .../tests/real_git_fetch.rs | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/crates/git-remote-gitlawb/tests/real_git_fetch.rs b/crates/git-remote-gitlawb/tests/real_git_fetch.rs index e1183192..7013d3e1 100644 --- a/crates/git-remote-gitlawb/tests/real_git_fetch.rs +++ b/crates/git-remote-gitlawb/tests/real_git_fetch.rs @@ -262,6 +262,26 @@ fn write_response(mut stream: TcpStream, status: &str, content_type: &str, body: /// Run `git fetch` in `clone` through the helper, with a hard timeout so a /// regression to the deadlock fails fast instead of hanging the suite. fn fetch_with_helper(clone: &Path, node_url: &str) -> (bool, std::process::Output) { + let mut last = fetch_with_helper_once(clone, node_url); + for attempt in 1..3u32 { + if last.0 && last.1.status.success() { + return last; + } + let stderr = String::from_utf8_lossy(&last.1.stderr); + let transient = stderr.contains("10053") + || stderr.contains("connection error") + || stderr.contains("connection was aborted") + || stderr.contains("error sending request"); + if !transient { + return last; + } + std::thread::sleep(Duration::from_millis(250 * attempt as u64)); + last = fetch_with_helper_once(clone, node_url); + } + last +} + +fn fetch_with_helper_once(clone: &Path, node_url: &str) -> (bool, std::process::Output) { let helper_bin = PathBuf::from(env!("CARGO_BIN_EXE_git-remote-gitlawb")); let helper_dir = helper_bin.parent().unwrap().to_path_buf(); let path_env = match std::env::var_os("PATH") { From 68133409d51638c5122868c320cfcf35b2167322 Mon Sep 17 00:00:00 2001 From: Ayush7614 Date: Sat, 15 Aug 2026 02:14:44 +0530 Subject: [PATCH 4/4] test(git-remote-gitlawb): revert Windows fetch workarounds The retry-on-10053 and wait-for-shim probes papered over the accepted-socket inheriting the listener's non-blocking state. #312 fixed that at the source (normalize_accepted_stream) and is merged, so take the workarounds back out and let the real fix carry the Windows lane. --- .../tests/real_git_fetch.rs | 35 ------------------- 1 file changed, 35 deletions(-) diff --git a/crates/git-remote-gitlawb/tests/real_git_fetch.rs b/crates/git-remote-gitlawb/tests/real_git_fetch.rs index 7013d3e1..26d88f27 100644 --- a/crates/git-remote-gitlawb/tests/real_git_fetch.rs +++ b/crates/git-remote-gitlawb/tests/real_git_fetch.rs @@ -128,19 +128,6 @@ impl Drop for Shim { } } -/// Wait until the shim accept loop is polling. Without this, Windows CI can -/// race: `git fetch` connects before the spawned thread enters `accept()`. -fn wait_for_shim_ready(addr: std::net::SocketAddr) { - let deadline = Instant::now() + Duration::from_secs(10); - while Instant::now() < deadline { - if TcpStream::connect_timeout(&addr, Duration::from_millis(200)).is_ok() { - return; - } - std::thread::sleep(Duration::from_millis(25)); - } - panic!("shim at {addr} did not accept connections within 10s"); -} - /// Start a minimal smart-HTTP shim serving `repo` on 127.0.0.1. Handles the /// upload-pack advertisement (GET) and pack negotiation (POST), counting POSTs. fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim { @@ -167,8 +154,6 @@ fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim { } }); - wait_for_shim_ready(addr); - Shim { base_url, posts, @@ -262,26 +247,6 @@ fn write_response(mut stream: TcpStream, status: &str, content_type: &str, body: /// Run `git fetch` in `clone` through the helper, with a hard timeout so a /// regression to the deadlock fails fast instead of hanging the suite. fn fetch_with_helper(clone: &Path, node_url: &str) -> (bool, std::process::Output) { - let mut last = fetch_with_helper_once(clone, node_url); - for attempt in 1..3u32 { - if last.0 && last.1.status.success() { - return last; - } - let stderr = String::from_utf8_lossy(&last.1.stderr); - let transient = stderr.contains("10053") - || stderr.contains("connection error") - || stderr.contains("connection was aborted") - || stderr.contains("error sending request"); - if !transient { - return last; - } - std::thread::sleep(Duration::from_millis(250 * attempt as u64)); - last = fetch_with_helper_once(clone, node_url); - } - last -} - -fn fetch_with_helper_once(clone: &Path, node_url: &str) -> (bool, std::process::Output) { let helper_bin = PathBuf::from(env!("CARGO_BIN_EXE_git-remote-gitlawb")); let helper_dir = helper_bin.parent().unwrap().to_path_buf(); let path_env = match std::env::var_os("PATH") {