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

Expose an agent's StateMachine current state to REST/SSE consumers via a
nullable, opaque `Task.current_state` label. It is written through the existing
UPDATE_TASK / PUT /tasks/{id} path, which already publishes a `task_updated` SSE
event carrying the full task, so updates are reactive (push) while GET /tasks/{id}
remains the authoritative point-read for load/reconnect reconciliation.

The field is framework-agnostic and default-null: agents opt in by writing it;
existing tasks and stream consumers are unaffected (additive, backward compatible).

- ORM column + Alembic migration (nullable add, down_revision a1b2c3d4e5f6)
- Task response schema, UpdateTaskRequest, TaskEntity + converter
- update_mutable_fields_on_task applies it in the same update_task write as
  task_metadata (single task_updated publish)
- both PUT routes thread it through
- regenerated openapi.yaml
- tests: route PUT/null/no-clobber, task_updated carries current_state,
  service-layer emit+persist, extra="ignore" forward-compat pin

Author: qwhex

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
+import sqlalchemy as sa
+
+
+# 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:
+    # Opaque label mirroring an agent's StateMachine current state. Nullable and
+    # additive; agents opt in by emitting it. Metadata-only add, non-blocking.
+    op.add_column('tasks', sa.Column('current_state', sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column('tasks', '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,11 @@ 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
+          - type: 'null'
+          title: If provided, replaces the task's current_state label.
       type: object
       title: UpdateTaskRequest
     ValidationError:

agentex/src/adapters/orm.py

@@ -75,6 +75,10 @@ class TaskORM(BaseORM):
     cleaned_at = Column(DateTime(timezone=True), nullable=True)
     params = Column(JSONB, nullable=True)
     task_metadata = Column(JSONB, nullable=True)
+    # Opaque, framework-agnostic label mirroring an agent's StateMachine current
+    # state. Written best-effort by the agent on each transition; reconciled on
+    # read. Orthogonal to `status` (Temporal workflow lifecycle).
+    current_state = Column(String, nullable=True)
     # Many-to-Many relationship with agents
     agents = relationship("AgentORM", secondary="task_agents", back_populates="tasks")
 

agentex/src/api/routes/tasks.py

@@ -196,6 +196,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)
 
@@ -217,6 +218,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

@@ -63,6 +63,13 @@ class Task(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'."
+        ),
+    )
 
 
 class TaskResponse(Task):
@@ -87,6 +94,10 @@ class UpdateTaskRequest(BaseModel):
             "subfields."
         ),
     )
+    current_state: str | None = Field(
+        None,
+        title="If provided, replaces the task's current_state label.",
+    )
 
 
 class TaskStatusReasonRequest(BaseModel):

agentex/src/domain/entities/tasks.py

@@ -66,6 +66,10 @@ class TaskEntity(BaseModel):
         None,
         title="Task metadata",
     )
+    current_state: str | None = Field(
+        None,
+        title="Opaque label mirroring the agent's StateMachine current state",
+    )
 
     # allow extra fields for agents relationships
     model_config = ConfigDict(extra="allow")
@@ -84,4 +88,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/use_cases/tasks_use_case.py

@@ -95,6 +95,7 @@ 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 = None,
     ) -> TaskEntity:
         """Update mutable fields on a task entity. This is used by our API since not all fields should be mutable."""
 
@@ -109,8 +110,8 @@ 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 current_state is None:
             return task_entity
 
         # `merge_params` is a separate atomic JSONB shallow-merge so concurrent
@@ -126,8 +127,15 @@ async def update_mutable_fields_on_task(
             if merged is not None:
                 task_entity = merged
 
-        if task_metadata is not None:
-            task_entity.task_metadata = task_metadata
+        # Apply the whole-row field updates (task_metadata, current_state) in a
+        # single update_task write so they emit one task_updated event rather than
+        # two. current_state rides the same publish that already powers reactive
+        # task_updated delivery to stream consumers.
+        if task_metadata is not None or current_state is not None:
+            if task_metadata is not None:
+                task_entity.task_metadata = task_metadata
+            if current_state is not None:
+                task_entity.current_state = current_state
             task_entity = await self.task_service.update_task(task=task_entity)
 
         return task_entity

agentex/tests/integration/api/tasks/test_tasks_api.py

@@ -683,6 +683,85 @@ async def test_update_task_endpoint_success(
         assert response_data["task_metadata"]["configuration"]["version"] == "2.0.0"
         assert response_data["task_metadata"]["metrics"]["complexity_score"] == 75
 
+    async def test_update_task_current_state(
+        self, isolated_client, isolated_repositories
+    ):
+        """PUT /tasks/{id} writes current_state; null/omitted leaves it unset."""
+        agent_repo = isolated_repositories["agent_repository"]
+        agent = AgentEntity(
+            id=orm_id(),
+            name="current-state-agent",
+            description="Agent for current_state update testing",
+            acp_url="http://test-acp:8000",
+            acp_type=ACPType.SYNC,
+        )
+        await agent_repo.create(agent)
+
+        task_repo = isolated_repositories["task_repository"]
+        task = TaskEntity(
+            id=orm_id(),
+            name="task-for-current-state",
+            status=TaskStatus.RUNNING,
+            status_reason="Test task for current_state",
+        )
+        created_task = await task_repo.create(agent_id=agent.id, task=task)
+
+        # Fresh task: current_state present in response and null by default.
+        response = await isolated_client.get(f"/tasks/{created_task.id}")
+        assert response.status_code == 200
+        assert response.json()["current_state"] is None
+
+        # Setting current_state persists and echoes back.
+        response = await isolated_client.put(
+            f"/tasks/{created_task.id}", json={"current_state": "awaiting_input"}
+        )
+        assert response.status_code == 200
+        assert response.json()["current_state"] == "awaiting_input"
+
+        # Point-read reflects the committed value (source of truth).
+        response = await isolated_client.get(f"/tasks/{created_task.id}")
+        assert response.status_code == 200
+        assert response.json()["current_state"] == "awaiting_input"
+
+        # Updating only task_metadata leaves current_state untouched (no clobber).
+        response = await isolated_client.put(
+            f"/tasks/{created_task.id}", json={"task_metadata": {"k": "v"}}
+        )
+        assert response.status_code == 200
+        assert response.json()["current_state"] == "awaiting_input"
+
+    async def test_update_task_request_ignores_unknown_fields(
+        self, isolated_client, isolated_repositories
+    ):
+        """Guards the extra="ignore" assumption a newer SDK relies on: a payload
+        carrying fields the server doesn't model must 200 (drop them), not 422.
+        Breaks loudly if UpdateTaskRequest ever switches to extra="forbid"."""
+        agent_repo = isolated_repositories["agent_repository"]
+        agent = AgentEntity(
+            id=orm_id(),
+            name="unknown-fields-agent",
+            description="Agent for unknown-field compat testing",
+            acp_url="http://test-acp:8000",
+            acp_type=ACPType.SYNC,
+        )
+        await agent_repo.create(agent)
+
+        task_repo = isolated_repositories["task_repository"]
+        task = TaskEntity(
+            id=orm_id(),
+            name="task-for-unknown-fields",
+            status=TaskStatus.RUNNING,
+            status_reason="Test task for unknown-field compat",
+        )
+        created_task = await task_repo.create(agent_id=agent.id, task=task)
+
+        response = await isolated_client.put(
+            f"/tasks/{created_task.id}",
+            json={"current_state": "working", "field_from_a_newer_sdk": "ignored"},
+        )
+        assert response.status_code == 200
+        assert response.json()["current_state"] == "working"
+
     async def test_update_task_endpoint_validation(
         self, isolated_client, isolated_repositories
     ):

agentex/tests/integration/test_task_stream.py

@@ -230,6 +230,64 @@ async def collect_stream_events():
 
         print("✅ Task metadata update successfully triggered stream event")
 
+    async def test_current_state_update_triggers_stream_event(
+        self, test_agent_and_task, tasks_use_case, streams_use_case
+    ):
+        """current_state rides the existing task_updated event, so a client
+        subscribed to the task stream is pushed the new state reactively."""
+        _agent, task = test_agent_and_task
+
+        stream_events = []
+
+        async def collect_stream_events():
+            try:
+                async for event_data in streams_use_case.stream_task_events(
+                    task_id=task.id
+                ):
+                    if event_data.startswith("data: "):
+                        import json
+
+                        event_json = event_data[6:].strip()
+                        if event_json:
+                            try:
+                                event = json.loads(event_json)
+                                stream_events.append(event)
+                                if event.get("type") == "task_updated":
+                                    break
+                            except json.JSONDecodeError:
+                                pass
+            except asyncio.CancelledError:
+                pass
+
+        stream_task = asyncio.create_task(collect_stream_events())
+        await asyncio.sleep(0.1)
+
+        updated_task = await tasks_use_case.update_mutable_fields_on_task(
+            id=task.id, current_state="awaiting_input"
+        )
+
+        await asyncio.sleep(0.5)
+
+        stream_task.cancel()
+        try:
+            await stream_task
+        except asyncio.CancelledError:
+            pass
+
+        task_updated_events = [
+            e for e in stream_events if e.get("type") == "task_updated"
+        ]
+        assert len(task_updated_events) >= 1, (
+            f"Expected task_updated event, got events: {[e.get('type') for e in stream_events]}"
+        )
+        event_task = task_updated_events[0]["task"]
+        assert event_task["id"] == task.id
+        assert event_task["current_state"] == "awaiting_input"
+
+        assert updated_task.current_state == "awaiting_input"
+
+        print("✅ current_state update successfully triggered stream event")
+
     async def test_get_task_returns_updated_metadata_after_stream_update(
         self, test_agent_and_task, tasks_use_case
     ):

agentex/tests/unit/services/test_task_service.py

@@ -938,6 +938,32 @@ async def test_update_task_with_task_metadata_changes(
         assert event_data["type"] == "task_updated"
         assert event_data["task"]["task_metadata"] == updated_metadata
 
+    async def test_update_task_current_state_publishes_stream_event(
+        self, task_service, agent_repository, sample_agent, redis_stream_repository
+    ):
+        """update_task persists current_state and carries it on the published
+        task_updated event, so subscribed clients see the new state."""
+        await create_or_get_agent(agent_repository, sample_agent)
+        created_task = await task_service.create_task(
+            agent=sample_agent, task_name="task-for-current-state"
+        )
+
+        created_task.current_state = "working"
+        redis_stream_repository.send_data = AsyncMock()
+
+        result = await task_service.update_task(created_task)
+
+        assert result.current_state == "working"
+        retrieved_task = await task_service.get_task(id=created_task.id)
+        assert retrieved_task.current_state == "working"
+
+        redis_stream_repository.send_data.assert_called_once()
+        call_args = redis_stream_repository.send_data.call_args
+        assert call_args[0][0] == f"task:{created_task.id}"
+        event_data = call_args[0][1]
+        assert event_data["type"] == "task_updated"
+        assert event_data["task"]["current_state"] == "working"
+
     async def test_get_task_preserves_task_metadata(
         self, task_service, agent_repository, sample_agent
     ):