test/fix(agentex): address round-2 audit on current_state

- Add the real use-case-level clobber regression test: it forces a stale read
  (entity fetched RUNNING, task transitioned to COMPLETED before the write) and
  asserts status stays COMPLETED. This fails on the old whole-row-merge path and
  passes on the column-scoped update — the prior service-level test could not
  catch the regression (it exercised the new primitive directly, which is
  clobber-proof by construction); its docstring is corrected accordingly.
- update_mutable_fields_on_task now raises ItemDoesNotExist when the atomic
  update reports the row vanished, instead of silently returning stale data
  (consistent with the not-found contract and sibling transition methods;
  defensive — no live hard-delete path reaches it).
- Drop max_length from the response Task.current_state field: input is already
  bounded by UpdateTaskRequest + the String(255) column, and enforcing it on the
  read path would turn a future column-widening into a 500 on every read.
- Make the repository update_mutable_fields empty-fields branch's contract honest
  in its docstring (it's an unreachable defensive no-op).
- Assert the intermediate set in the clear-to-null test so the clear is a real
  transition, not a null→null no-op.
- Re-await the cancelled collector in the stream test to avoid a dangling-task
  warning at teardown.
- Remove the now-unused `import sqlalchemy as sa` from the migration.
- Regenerated openapi.yaml (maxLength now only on the request schema).

Author: qwhex

agentex/database/migrations/alembic/versions/2026_07_22_1200_add_task_current_state_b2c3d4e5f6a7.py

@@ -8,7 +8,6 @@
 from typing import Sequence, Union
 
 from alembic import op
-import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.

agentex/openapi.yaml

@@ -6591,7 +6591,6 @@ components:
         current_state:
           anyOf:
           - type: string
-            maxLength: 255
           - type: 'null'
           title: Opaque label mirroring the agent's StateMachine current state; null
             when the agent does not emit one. Orthogonal to 'status'.
@@ -6821,7 +6820,6 @@ components:
         current_state:
           anyOf:
           - type: string
-            maxLength: 255
           - type: 'null'
           title: Opaque label mirroring the agent's StateMachine current state; null
             when the agent does not emit one. Orthogonal to 'status'.

agentex/src/api/schemas/tasks.py

@@ -68,9 +68,11 @@ class Task(BaseModel):
         None,
         title="Task metadata",
     )
+    # No max_length on this response field on purpose: input is already bounded
+    # by UpdateTaskRequest + the String(255) column, and enforcing the bound on
+    # the read path would turn a future column-widening into a 500 on every read.
     current_state: str | None = Field(
         None,
-        max_length=CURRENT_STATE_MAX_LENGTH,
         title=(
             "Opaque label mirroring the agent's StateMachine current state; "
             "null when the agent does not emit one. Orthogonal to 'status'."

agentex/src/domain/repositories/task_repository.py

@@ -264,11 +264,15 @@ async def update_mutable_fields(
         changed ``status``/``params``), this issues a single
         ``UPDATE ... SET <only these columns> WHERE id = :id RETURNING *``.
         Touching only the supplied columns means a concurrent status transition
-        or param merge is never reverted. Returns the updated entity, or
-        ``None`` if no task with ``task_id`` exists.
+        or param merge is never reverted. With a non-empty ``fields`` this returns
+        the updated entity, or ``None`` if no task with ``task_id`` exists.
 
         ``fields`` values are applied verbatim, so passing ``current_state=None``
         clears the column (callers distinguish "clear" from "omit" upstream).
+
+        An empty ``fields`` is a defensive no-op that returns the current task
+        (raising ``ItemDoesNotExist`` if absent); the sole caller never reaches it,
+        as it gates the call behind a non-empty field set.
         """
         if not fields:
             return await self.get(id=task_id)

agentex/src/domain/use_cases/tasks_use_case.py

@@ -158,8 +158,14 @@ async def update_mutable_fields_on_task(
             updated = await self.task_service.update_mutable_fields(
                 task_entity.id, fields
             )
-            if updated is not None:
-                task_entity = updated
+            if updated is None:
+                # The row vanished between the initial read and the write. No live
+                # hard-delete path reaches here today (delete is a soft status
+                # update, already guarded above), so this is defensive — but raise
+                # rather than return stale data, matching the not-found contract
+                # above and the sibling terminal-transition methods.
+                raise ItemDoesNotExist(f"Task {id or name} not found")
+            task_entity = updated
 
         return task_entity
 

agentex/tests/integration/test_task_stream.py

@@ -275,6 +275,9 @@ async def collect_stream_events():
                 await stream_task
         except (TimeoutError, asyncio.CancelledError):
             stream_task.cancel()
+            # Re-await so the task's own cancellation cleanup runs and no
+            # "Task exception was never retrieved" warning leaks at teardown.
+            await asyncio.gather(stream_task, return_exceptions=True)
 
         task_updated_events = [
             e for e in stream_events if e.get("type") == "task_updated"

agentex/tests/unit/services/test_task_service.py

@@ -991,12 +991,13 @@ async def test_update_mutable_fields_persists_and_publishes(
         assert event_data["type"] == "task_updated"
         assert event_data["task"]["current_state"] == "working"
 
-    async def test_update_mutable_fields_does_not_clobber_status(
+    async def test_update_mutable_fields_leaves_status_untouched(
         self, task_service, agent_repository, sample_agent, redis_stream_repository
     ):
-        """Regression (blocker): a current_state write must not revert a status
-        changed by another writer. update_mutable_fields is column-scoped, so a
-        task that went terminal is not resurrected to its earlier status."""
+        """The primitive is column-scoped: writing current_state does not touch
+        status. (The use-case-level clobber regression — a stale read racing a
+        status transition — is guarded in test_tasks_use_case.py; this only pins
+        that the repository UPDATE sets no columns beyond those supplied.)"""
         await create_or_get_agent(agent_repository, sample_agent)
         created_task = await task_service.create_task(
             agent=sample_agent, task_name="task-for-noclobber"

agentex/tests/unit/use_cases/test_tasks_use_case.py

@@ -639,9 +639,12 @@ async def test_update_current_state_clears_on_explicit_null(
         task = await task_service.create_task(
             agent=sample_agent, task_name="current-state-clear-test"
         )
-        await tasks_use_case.update_mutable_fields_on_task(
+        was_set = await tasks_use_case.update_mutable_fields_on_task(
             id=task.id, current_state="working"
         )
+        # Confirm it was actually set, so the clear below is a real transition
+        # (not a null→null no-op that would pass trivially).
+        assert was_set.current_state == "working"
 
         cleared = await tasks_use_case.update_mutable_fields_on_task(
             id=task.id, current_state=None
@@ -673,6 +676,48 @@ async def test_update_current_state_and_metadata_single_atomic_write(
         assert updated.task_metadata == {"stage": "two"}
         assert updated.status == TaskStatus.RUNNING
 
+    async def test_update_current_state_does_not_clobber_concurrent_status(
+        self, tasks_use_case, task_service, task_repository, agent_repository, sample_agent
+    ):
+        """Regression (round-one blocker): a current_state write must not revert a
+        status changed concurrently. Reproduces the exact staleness the old
+        whole-row-merge path needed — the use case reads the entity while RUNNING,
+        the task goes COMPLETED before the write lands, and the write must keep
+        COMPLETED. Fails on the old update_task/merge path (reverts to RUNNING),
+        passes on the column-scoped update.
+        """
+        await create_or_get_agent(agent_repository, sample_agent)
+        task = await task_service.create_task(
+            agent=sample_agent, task_name="current-state-clobber-test"
+        )
+
+        # Snapshot the entity as the use case would have read it, BEFORE the
+        # concurrent transition — this is the stale read the bug wrote back.
+        stale_entity = await task_service.get_task(id=task.id)
+        assert stale_entity.status == TaskStatus.RUNNING
+
+        # Another writer moves the task to a terminal status after that read.
+        await task_service.transition_task_status(
+            task_id=task.id,
+            expected_status=TaskStatus.RUNNING,
+            new_status=TaskStatus.COMPLETED,
+            status_reason="done",
+        )
+
+        # Force the use case to operate on the stale (pre-transition) read.
+        task_service.get_task = AsyncMock(return_value=stale_entity)
+
+        updated = await tasks_use_case.update_mutable_fields_on_task(
+            id=task.id, current_state="late"
+        )
+
+        # The write set current_state without reverting the terminal status.
+        assert updated.current_state == "late"
+        assert updated.status == TaskStatus.COMPLETED
+        persisted = await task_repository.get(id=task.id)
+        assert persisted.status == TaskStatus.COMPLETED
+        assert persisted.current_state == "late"
+
     async def test_update_current_state_on_deleted_task_raises(
         self, tasks_use_case, task_service, agent_repository, sample_agent
     ):