feat(agentex): materialized Task.current_state for reactive state observability

CLAUDE.md

@@ -358,7 +358,7 @@ Use the escape hatch for "this needs a maintenance window with traffic shifted a
 
 ##### Anti-pattern → linter rule reference
 
-The migration linter at `agentex/scripts/lint_migrations.py` enforces these rules at PR time via `.github/workflows/migration-lint.yml`. It only checks files changed vs the PR base, so existing migrations are not retro-flagged. The mapping below is what the linter catches:
+The migration linter at `agentex/scripts/ci_tools/migration_lint.py` enforces these rules at PR time via `.github/workflows/migration-lint.yml`. It only checks files changed vs the PR base, so existing migrations are not retro-flagged. The mapping below is what the linter catches:
 
 | Anti-pattern | Linter rule |
 |---|---|
@@ -371,7 +371,7 @@ The migration linter at `agentex/scripts/lint_migrations.py` enforces these rule
 Run the linter locally before pushing:
 
 ```bash
-agentex/scripts/lint_migrations.py --base-ref origin/main
+agentex/scripts/ci_tools/migration_lint.py --base origin/main
 ```
 
 ##### Other rules

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

@@ -0,0 +1,28 @@
+"""add task current_state
+
+Revision ID: b2c3d4e5f6a7
+Revises: a1b2c3d4e5f6
+Create Date: 2026-07-22 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b2c3d4e5f6a7'
+down_revision: Union[str, None] = 'a1b2c3d4e5f6'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Nullable additive column; idempotent, metadata-only, non-blocking.
+    op.execute(
+        "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS current_state VARCHAR(255)"
+    )
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE tasks DROP COLUMN IF EXISTS current_state")

agentex/openapi.yaml

@@ -6588,6 +6588,12 @@ components:
             type: object
           - type: 'null'
           title: Task metadata
+        current_state:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Opaque label mirroring the agent's StateMachine current state; null
+            when the agent does not emit one. Orthogonal to 'status'.
       type: object
       required:
       - id
@@ -6811,6 +6817,12 @@ components:
             type: object
           - type: 'null'
           title: Task metadata
+        current_state:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Opaque label mirroring the agent's StateMachine current state; null
+            when the agent does not emit one. Orthogonal to 'status'.
         agents:
           anyOf:
           - items:
@@ -7396,6 +7408,13 @@ components:
           - type: 'null'
           title: Optional shallow-merge patch applied to the task's params column.
             Top-level keys overwrite; pass full nested objects to change subfields.
+        current_state:
+          anyOf:
+          - type: string
+            maxLength: 255
+          - type: 'null'
+          title: If provided, replaces the task's current_state label; pass null to
+            clear it, omit to leave it unchanged.
       type: object
       title: UpdateTaskRequest
     ValidationError:

agentex/src/adapters/orm.py

@@ -24,6 +24,7 @@
 from src.domain.entities.deployments import DeploymentStatus
 from src.domain.entities.tasks import TaskStatus
 from src.utils.ids import orm_id
+from src.utils.task_constants import CURRENT_STATE_MAX_LENGTH
 
 BaseORM = declarative_base()
 
@@ -75,6 +76,8 @@ class TaskORM(BaseORM):
     cleaned_at = Column(DateTime(timezone=True), nullable=True)
     params = Column(JSONB, nullable=True)
     task_metadata = Column(JSONB, nullable=True)
+    # Opaque agent-state label, orthogonal to `status`; capped since it rides every task_updated SSE payload.
+    current_state = Column(String(CURRENT_STATE_MAX_LENGTH), nullable=True)
     # Many-to-Many relationship with agents
     agents = relationship("AgentORM", secondary="task_agents", back_populates="tasks")
 

agentex/src/api/routes/tasks.py

@@ -22,7 +22,7 @@
 from src.domain.entities.tasks import TaskStatus as DomainTaskStatus
 from src.domain.services.authorization_service import DAuthorizationService
 from src.domain.use_cases.streams_use_case import DStreamsUseCase
-from src.domain.use_cases.tasks_use_case import DTaskUseCase
+from src.domain.use_cases.tasks_use_case import UNSET, DTaskUseCase
 from src.utils.authorization_shortcuts import (
     DAuthorizedId,
     DAuthorizedName,
@@ -196,6 +196,12 @@ async def update_task(
         id=task_id,
         task_metadata=request.task_metadata,
         merge_params=request.merge_params,
+        # Forward only when the client sent the key: explicit null clears, omitted leaves unchanged.
+        current_state=(
+            request.current_state
+            if "current_state" in request.model_fields_set
+            else UNSET
+        ),
     )
     return Task.model_validate(updated_task_entity)
 
@@ -217,6 +223,12 @@ async def update_task_by_name(
         name=task_name,
         task_metadata=request.task_metadata,
         merge_params=request.merge_params,
+        # Forward only when the client sent the key: explicit null clears, omitted leaves unchanged.
+        current_state=(
+            request.current_state
+            if "current_state" in request.model_fields_set
+            else UNSET
+        ),
     )
     return Task.model_validate(updated_task_entity)
 

agentex/src/api/schemas/tasks.py

@@ -6,6 +6,7 @@
 
 from src.api.schemas.agents import Agent
 from src.utils.model_utils import BaseModel
+from src.utils.task_constants import CURRENT_STATE_MAX_LENGTH
 
 
 class TaskRelationships(str, Enum):
@@ -63,6 +64,14 @@ class Task(BaseModel):
         None,
         title="Task metadata",
     )
+    # No bound on the read path: enforcing it would 500 if the column is ever widened (writes are already bounded).
+    current_state: str | None = Field(
+        None,
+        title=(
+            "Opaque label mirroring the agent's StateMachine current state; "
+            "null when the agent does not emit one. Orthogonal to 'status'."
+        ),
+    )
 
 
 class TaskResponse(Task):
@@ -87,6 +96,14 @@ class UpdateTaskRequest(BaseModel):
             "subfields."
         ),
     )
+    current_state: str | None = Field(
+        None,
+        max_length=CURRENT_STATE_MAX_LENGTH,
+        title=(
+            "If provided, replaces the task's current_state label; "
+            "pass null to clear it, omit to leave it unchanged."
+        ),
+    )
 
 
 class TaskStatusReasonRequest(BaseModel):

agentex/src/domain/entities/tasks.py

@@ -66,6 +66,13 @@ class TaskEntity(BaseModel):
         None,
         title="Task metadata",
     )
+    current_state: str | None = Field(
+        None,
+        title=(
+            "Opaque label mirroring the agent's StateMachine current state; "
+            "null when the agent does not emit one. Orthogonal to 'status'."
+        ),
+    )
 
     # allow extra fields for agents relationships
     model_config = ConfigDict(extra="allow")
@@ -84,4 +91,5 @@ def convert_task_to_entity(task: Task) -> TaskEntity:
         cleaned_at=task.cleaned_at,
         params=task.params,
         task_metadata=task.task_metadata,
+        current_state=task.current_state,
     )

agentex/src/domain/repositories/task_repository.py

@@ -1,6 +1,6 @@
 from collections.abc import Sequence
 from datetime import UTC, datetime, timedelta
-from typing import Annotated, Literal
+from typing import Annotated, Any, Literal
 
 from fastapi import Depends
 from sqlalchemy import cast, distinct, func, select, update
@@ -21,6 +21,10 @@
 
 logger = make_logger(__name__)
 
+# Columns update_mutable_fields is allowed to set — a code guardrail (not a comment)
+# so a future caller can't route status/params through it and bypass their atomic CAS.
+_MUTABLE_TASK_COLUMNS = frozenset({"task_metadata", "current_state"})
+
 
 class TaskRepository(PostgresCRUDRepository[TaskORM, TaskEntity, TaskRelationships]):
     """Repository for Task entity with relationship loading support"""
@@ -231,20 +235,42 @@ async def merge_params(self, task_id: str, patch: dict) -> TaskEntity | None:
         general-purpose updater.
         """
 
+        # ``COALESCE(params, '{}'::jsonb)`` so a NULL existing value doesn't poison the
+        # concat; explicit JSONB casts so Postgres picks the jsonb ``||`` (not text concat).
+        existing = func.coalesce(TaskORM.params, cast({}, JSONB))
+        merged = existing.op("||", return_type=JSONB)(cast(patch, JSONB))
+        return await self._update_returning(task_id, {"params": merged})
+
+    async def update_mutable_fields(
+        self, task_id: str, fields: dict[str, Any]
+    ) -> TaskEntity | None:
+        """Atomically set the given caller-mutable columns on one task row via
+        ``UPDATE ... RETURNING`` — column-scoped, so (unlike ``update``'s whole-row merge) it
+        can't clobber a concurrently changed ``status``/``params``. Values apply verbatim
+        (``current_state=None`` clears). Returns the updated entity, or ``None`` if absent.
+        """
+        unknown = fields.keys() - _MUTABLE_TASK_COLUMNS
+        if unknown:
+            raise ValueError(
+                f"update_mutable_fields may only set {sorted(_MUTABLE_TASK_COLUMNS)}; "
+                f"got disallowed columns {sorted(unknown)}"
+            )
+        return await self._update_returning(task_id, fields)
+
+    async def _update_returning(
+        self, task_id: str, values: dict[str, Any]
+    ) -> TaskEntity | None:
+        """``UPDATE tasks SET <values> WHERE id`` → the updated entity, or ``None`` if no
+        such task exists. Shared by merge_params and update_mutable_fields.
+        """
         async with (
             self.start_async_db_session(True) as session,
             async_sql_exception_handler(),
         ):
-            # ``COALESCE(params, '{}'::jsonb)`` so a NULL existing value
-            # doesn't poison the concat to NULL. Both operands cast to
-            # JSONB explicitly so Postgres picks the JSONB ``||`` operator
-            # (not the text concat overload).
-            existing = func.coalesce(TaskORM.params, cast({}, JSONB))
-            merged = existing.op("||", return_type=JSONB)(cast(patch, JSONB))
             stmt = (
                 update(TaskORM)
                 .where(TaskORM.id == task_id)
-                .values(params=merged)
+                .values(**values)
                 .returning(TaskORM)
             )
             result = await session.execute(stmt)

agentex/src/domain/services/task_service.py

@@ -242,6 +242,32 @@ async def update_task(self, task: TaskEntity) -> TaskEntity:
 
         return updated_task
 
+    async def update_mutable_fields(
+        self, task_id: str, fields: dict[str, Any]
+    ) -> TaskEntity | None:
+        """Column-scoped atomic update of the given columns, then publish task_updated.
+        Returns the updated entity, or ``None`` if the task no longer exists.
+        """
+        updated_task = await self.task_repository.update_mutable_fields(task_id, fields)
+        if updated_task is None:
+            return None
+
+        try:
+            topic = get_task_event_stream_topic(task_id=task_id)
+            await self.stream_repository.send_data(
+                topic,
+                TaskStreamTaskUpdatedEventEntity(
+                    type="task_updated", task=updated_task
+                ).model_dump(mode="json"),
+            )
+            logger.info(f"task_updated event published to topic: {topic}")
+        except Exception as e:
+            logger.error(
+                f"Error sending task_updated event to stream: {e}", exc_info=True
+            )
+
+        return updated_task
+
     async def merge_task_params(self, task_id: str, patch: dict) -> TaskEntity | None:
         """Atomically shallow-merge ``patch`` into ``tasks.params``. Returns
         the updated entity, or ``None`` if no task with ``task_id`` exists.
@@ -374,7 +400,11 @@ async def cancel_task(
             new_status=TaskStatus.CANCELED,
             status_reason="Task canceled by user",
         )
-        return updated if updated is not None else await self.task_repository.get(id=task.id)
+        return (
+            updated
+            if updated is not None
+            else await self.task_repository.get(id=task.id)
+        )
 
     async def interrupt_task(
         self, agent: AgentEntity, task: TaskEntity, acp_url: str

agentex/src/domain/use_cases/tasks_use_case.py

@@ -11,6 +11,13 @@
 logger = make_logger(__name__)
 
 
+class _Unset:
+    """Sentinel for partial updates: field omitted (untouched) vs. explicitly null (cleared)."""
+
+
+UNSET: Any = _Unset()
+
+
 class TasksUseCase:
     """
     Use case for managing tasks. Handles CRUD operations and delegates task operations to ACP servers.
@@ -95,12 +102,17 @@ async def update_mutable_fields_on_task(
         name: str | None = None,
         task_metadata: dict[str, Any] | None = None,
         merge_params: dict[str, Any] | None = None,
+        current_state: str | None | _Unset = UNSET,
     ) -> TaskEntity:
-        """Update mutable fields on a task entity. This is used by our API since not all fields should be mutable."""
+        """Update mutable fields on a task. ``current_state`` uses the UNSET sentinel
+        (explicit null clears, omitted leaves it); ``task_metadata`` None means "not supplied".
+        """
 
         if not id and not name:
             raise ClientError("Either id or name must be provided")
 
+        current_state_supplied = not isinstance(current_state, _Unset)
+
         # todo: make this a transaction?
         task_entity = await self.task_service.get_task(id=id, name=name)
         if task_entity.status == TaskStatus.DELETED:
@@ -109,26 +121,36 @@ async def update_mutable_fields_on_task(
             else:
                 raise ItemDoesNotExist(f"Task {name} not found")
 
-        # No-op if neither field was supplied.
-        if task_metadata is None and merge_params is None:
+        # No-op if no mutable field was supplied.
+        if (
+            task_metadata is None
+            and merge_params is None
+            and not current_state_supplied
+        ):
             return task_entity
 
-        # `merge_params` is a separate atomic JSONB shallow-merge so concurrent
-        # callers don't overwrite each other's fields (vs reading→mutating→writing
-        # the whole params dict on task_entity). Run it first so the refreshed
-        # entity it returns becomes the base we apply `task_metadata` on top of;
-        # otherwise the `task_entity = merged` reassignment would discard an
-        # in-memory metadata change made before the merge.
+        # Atomic JSONB shallow-merge; run first so its refreshed entity is the fallback return.
         if merge_params:
             merged = await self.task_service.merge_task_params(
                 task_entity.id, merge_params
             )
             if merged is not None:
                 task_entity = merged
 
+        # Single column-scoped write → one task_updated publish, no whole-row clobber.
+        fields: dict[str, Any] = {}
         if task_metadata is not None:
-            task_entity.task_metadata = task_metadata
-            task_entity = await self.task_service.update_task(task=task_entity)
+            fields["task_metadata"] = task_metadata
+        if current_state_supplied:
+            fields["current_state"] = current_state
+        if fields:
+            updated = await self.task_service.update_mutable_fields(
+                task_entity.id, fields
+            )
+            if updated is None:
+                # Row vanished mid-flight (defensive; no live hard-delete path). Raise, don't return stale.
+                raise ItemDoesNotExist(f"Task {id or name} not found")
+            task_entity = updated
 
         return task_entity
 

agentex/src/utils/task_constants.py

@@ -0,0 +1,3 @@
+# Single source for the current_state bound: the request-validation limit
+# (UpdateTaskRequest) and the storage limit (TaskORM.current_state column width).
+CURRENT_STATE_MAX_LENGTH = 255