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

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

@@ -6622,18 +6622,26 @@ components:
           - type: 'null'
           title: The timestamp when the task's content was cleaned for retention compliance;
             null when active
-        params:
+        task_metadata:
           anyOf:
           - additionalProperties: true
             type: object
           - type: 'null'
-          title: Task parameters
-        task_metadata:
+          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''. States
+            the SDK emits today: IDLE, WORKING, AWAITING_INPUT. Treat any other value
+            as unknown rather than assuming this set is closed.'
+        params:
           anyOf:
           - additionalProperties: true
             type: object
           - type: 'null'
-          title: Task metadata
+          title: Task parameters
       type: object
       required:
       - id
@@ -6845,18 +6853,26 @@ components:
           - type: 'null'
           title: The timestamp when the task's content was cleaned for retention compliance;
             null when active
-        params:
+        task_metadata:
           anyOf:
           - additionalProperties: true
             type: object
           - type: 'null'
-          title: Task parameters
-        task_metadata:
+          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''. States
+            the SDK emits today: IDLE, WORKING, AWAITING_INPUT. Treat any other value
+            as unknown rather than assuming this set is closed.'
+        params:
           anyOf:
           - additionalProperties: true
             type: object
           - type: 'null'
-          title: Task metadata
+          title: Task parameters
         agents:
           anyOf:
           - items:
@@ -6936,6 +6952,14 @@ 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''. States
+            the SDK emits today: IDLE, WORKING, AWAITING_INPUT. Treat any other value
+            as unknown rather than assuming this set is closed.'
         agents:
           anyOf:
           - items:
@@ -7504,6 +7528,14 @@ 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. Omit to leave
+            it untouched; send "" to clear it back to null (how an operator recovers
+            a task whose agent died mid-state).
       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

@@ -200,6 +200,7 @@ async def update_task(
         id=task_id,
         task_metadata=request.task_metadata,
         merge_params=request.merge_params,
+        current_state=request.current_state,
     )
     return Task.model_validate(updated_task_entity)
 
@@ -221,6 +222,7 @@ async def update_task_by_name(
         name=task_name,
         task_metadata=request.task_metadata,
         merge_params=request.merge_params,
+        current_state=request.current_state,
     )
     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_DESCRIPTION, CURRENT_STATE_MAX_LENGTH
 
 
 class TaskRelationships(str, Enum):
@@ -26,43 +27,26 @@ class TaskStatus(str, Enum):
     DELETED = "DELETED"
 
 
-class Task(BaseModel):
-    id: str = Field(
-        ...,
-        title="Unique Task ID",
-    )
-    name: str | None = Field(
-        None,
-        title="Unique name of the task",
-    )
-    status: TaskStatus | None = Field(
-        None,
-        title="The current status of the task",
-    )
-    status_reason: str | None = Field(
-        None,
-        title="The reason for the current task status",
-    )
-    created_at: datetime | None = Field(
-        None,
-        title="The timestamp when the task was created",
-    )
-    updated_at: datetime | None = Field(
-        None,
-        title="The timestamp when the task was last updated",
-    )
+class _TaskBase(BaseModel):
+    """Shared fields for Task and TaskSummary (everything except `params`)."""
+
+    id: str = Field(..., title="Unique Task ID")
+    name: str | None = Field(None, title="Unique name of the task")
+    status: TaskStatus | None = Field(None, title="The current status of the task")
+    status_reason: str | None = Field(None, title="The reason for the current task status")
+    created_at: datetime | None = Field(None, title="The timestamp when the task was created")
+    updated_at: datetime | None = Field(None, title="The timestamp when the task was last updated")
     cleaned_at: datetime | None = Field(
         None,
         title="The timestamp when the task's content was cleaned for retention compliance; null when active",
     )
-    params: dict[str, Any] | None = Field(
-        None,
-        title="Task parameters",
-    )
-    task_metadata: dict[str, Any] | None = Field(
-        None,
-        title="Task metadata",
-    )
+    task_metadata: dict[str, Any] | None = Field(None, title="Task metadata")
+    # Writes are bounded; reads are not, so widening the column won't 500.
+    current_state: str | None = Field(None, title=CURRENT_STATE_DESCRIPTION)
+
+
+class Task(_TaskBase):
+    params: dict[str, Any] | None = Field(None, title="Task parameters")
 
 
 class TaskResponse(Task):
@@ -74,43 +58,11 @@ class TaskResponse(Task):
     )
 
 
-class TaskSummary(BaseModel):
+class TaskSummary(_TaskBase):
     """Lean list-response shape. Omits `params` (the arbitrary create-time
     payload, which can carry per-caller secrets and PII); fetch GET /tasks/{id}
     for the full record."""
 
-    id: str = Field(
-        ...,
-        title="Unique Task ID",
-    )
-    name: str | None = Field(
-        None,
-        title="Unique name of the task",
-    )
-    status: TaskStatus | None = Field(
-        None,
-        title="The current status of the task",
-    )
-    status_reason: str | None = Field(
-        None,
-        title="The reason for the current task status",
-    )
-    created_at: datetime | None = Field(
-        None,
-        title="The timestamp when the task was created",
-    )
-    updated_at: datetime | None = Field(
-        None,
-        title="The timestamp when the task was last updated",
-    )
-    cleaned_at: datetime | None = Field(
-        None,
-        title="The timestamp when the task's content was cleaned for retention compliance; null when active",
-    )
-    task_metadata: dict[str, Any] | None = Field(
-        None,
-        title="Task metadata",
-    )
     agents: list["Agent"] | None = Field(
         default=None,
         title="Agents associated with this task (only populated when 'agents' view is requested)",
@@ -130,6 +82,15 @@ 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. Omit to leave it "
+            'untouched; send "" to clear it back to null (how an operator recovers a '
+            "task whose agent died mid-state)."
+        ),
+    )
 
 
 class TaskStatusReasonRequest(BaseModel):

agentex/src/domain/entities/tasks.py

@@ -6,6 +6,7 @@
 
 from src.api.schemas.tasks import Task
 from src.utils.model_utils import BaseModel
+from src.utils.task_constants import CURRENT_STATE_DESCRIPTION
 
 
 class TaskRelationships(str, Enum):
@@ -73,22 +74,12 @@ class TaskEntity(BaseModel):
         None,
         title="Task metadata",
     )
+    current_state: str | None = Field(None, title=CURRENT_STATE_DESCRIPTION)
 
     # allow extra fields for agents relationships
     model_config = ConfigDict(extra="allow")
 
 
 def convert_task_to_entity(task: Task) -> TaskEntity:
     """Converts the pydantic model from the API layer to the domain layer"""
-
-    return TaskEntity(
-        id=task.id,
-        name=task.name,
-        status=TaskStatus[task.status.value] if task.status is not None else None,
-        status_reason=task.status_reason,
-        created_at=task.created_at,
-        updated_at=task.updated_at,
-        cleaned_at=task.cleaned_at,
-        params=task.params,
-        task_metadata=task.task_metadata,
-    )
+    return TaskEntity.model_validate(task)

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,9 @@
 
 logger = make_logger(__name__)
 
+# Columns update_mutable_fields is allowed to set (status/params have their own atomic paths).
+_MUTABLE_TASK_COLUMNS = frozenset({"task_metadata", "current_state"})
+
 
 class TaskRepository(PostgresCRUDRepository[TaskORM, TaskEntity, TaskRelationships]):
     """Repository for Task entity with relationship loading support"""
@@ -231,20 +234,36 @@ 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:
+        """Column-scoped atomic update; can't clobber status/params. Returns updated entity or None."""
+        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 … SET … WHERE id → updated entity or None. 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)