fix(intelligent-eval): atomic CAS in assign_task and complete_task (resolves §6.1)

Replace read-check-write in task_queue.assign_task with UPDATE...WHERE
status='pending' and decide on rowcount so two concurrent workers
cannot both claim the same task. Also harden complete_task with the
same CAS pattern (status='assigned') so a worker + stuck-handler
double-complete leaves the DB in one state.

The xfail guard in test_worker_task_resilience now passes (4/4).

backend/agenteval/intelligent_eval/task_queue.py

@@ -10,7 +10,7 @@ OpenClaw workers to pick up. Tasks are prioritized by:
 from datetime import timedelta
 from typing import Optional
 
-from sqlmodel import Session, select
+from sqlmodel import Session, select, update
 
 from agenteval.intelligent_eval.models import IntelligentEvalStatus
 from agenteval.storage.db import (
@@ -113,39 +113,50 @@ def get_next_task(session: Session) -> Optional[IntelligentEvalTaskQueueDB]:
 
 
 def assign_task(task_id: str, cron_id: str, session: Session) -> bool:
-    """Assign a task to a cron.
+    """Atomically assign a pending task to a cron (CAS on status).
 
-    Returns:
+    P1 真问题修复（§6.1）: use ``UPDATE ... WHERE status='pending'`` and decide
-        True if assigned successfully, False if task not found or already assigned
+    on ``rowcount`` so two concurrent workers cannot both claim the same task.
+    The previous read-check-write left a race because SQLite + two sessions
+    could each read ``status=pending`` and each commit.
     """
-    task = session.get(IntelligentEvalTaskQueueDB, task_id)
+    stmt = (
-    if task is None or task.status != "pending":
+        update(IntelligentEvalTaskQueueDB)
-        return False
+        .where(IntelligentEvalTaskQueueDB.id == task_id)
-
+        .where(IntelligentEvalTaskQueueDB.status == "pending")
-    task.status = "assigned"
+        .values(
-    task.assigned_cron_id = cron_id
+            status="assigned",
-    task.assigned_at = utc_now()
+            assigned_cron_id=cron_id,
-    task.updated_at = utc_now()
+            assigned_at=utc_now(),
+            updated_at=utc_now(),
+        )
+    )
+    result = session.exec(stmt)
     session.commit()
-    return True
+    return result.rowcount > 0
 
 
 def complete_task(task_id: str, success: bool, error: Optional[str], session: Session) -> bool:
-    """Mark a task as completed or failed.
+    """Atomically mark a task as completed/failed (CAS on status).
 
-    Returns:
+    P1 真问题修复（§6.1 审计）: guard with ``status='assigned'`` so a
-        True if completed successfully, False if task not found
+    double-complete from worker + stuck-handler leaves the DB in one state.
     """
-    task = session.get(IntelligentEvalTaskQueueDB, task_id)
+    terminal = "completed" if success else "failed"
-    if task is None:
+    stmt = (
-        return False
+        update(IntelligentEvalTaskQueueDB)
-
+        .where(IntelligentEvalTaskQueueDB.id == task_id)
-    task.status = "completed" if success else "failed"
+        .where(IntelligentEvalTaskQueueDB.status == "assigned")
-    task.completed_at = utc_now()
+        .values(
-    task.error = error
+            status=terminal,
-    task.updated_at = utc_now()
+            completed_at=utc_now(),
+            error=error,
+            updated_at=utc_now(),
+        )
+    )
+    result = session.exec(stmt)
     session.commit()
-    return True
+    return result.rowcount > 0
 
 def requeue_stuck_task(eval_id: str, cron_id: str, session: Session) -> bool:
     """Mark the cron-stuck task as failed and enqueue a retry task.

tests/integration/test_worker_task_resilience.py

@@ -41,14 +41,6 @@ def test_concurrent_assign_only_one_succeeds(db_session):
     assert task.assigned_cron_id == "cron-A"
 
 
-@pytest.mark.xfail(
-    reason=(
-        "Known race: assign_task lacks atomic CAS — two sessions can both read "
-        "status=pending and both commit. Tracked in .scratch/v111-architecture-scan.md "
-        "discoveries. Fix requires atomic UPDATE ... WHERE status=pending in assign_task."
-    ),
-    strict=False,
-)
 def test_concurrent_assign_via_two_sessions(tmp_db_path, db_session):
     task = IntelligentEvalTaskQueueDB(
         eval_id="eval1", status="pending", priority=1, reason="slot_due"