From 1bf0790d7b7308d4b5e81812f15fba89c7909acd Mon Sep 17 00:00:00 2001 From: Vineet Vora Date: Mon, 3 Aug 2026 13:54:36 -0400 Subject: [PATCH 1/3] feat(states): add TaskStateORM and task_states table migration Empty, inert table: nothing reads or writes it until the Postgres task-state repository lands. The (task_id, agent_id) index is deferred to the write-semantics decision. --- ...08_03_1500_add_task_states_771e95623724.py | 72 +++++++++++++++++++ agentex/src/adapters/orm.py | 39 ++++++++++ 2 files changed, 111 insertions(+) create mode 100644 agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py diff --git a/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py b/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py new file mode 100644 index 00000000..5cad4ad0 --- /dev/null +++ b/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py @@ -0,0 +1,72 @@ +"""add task_states + +Revision ID: 771e95623724 +Revises: a1b2c3d4e5f6 +Create Date: 2026-08-03 15:00:00.000000 + +Creates the task_states table: the optional PostgreSQL backend for task state +(selected per deployment via TASK_STATE_STORAGE_PHASE; MongoDB remains the +default). Schema-only on a brand-new table, so creation is instant and holds +no lock against live traffic; nothing reads or writes the table until the +Postgres task-state repository lands. + +The (task_id, agent_id) index is deliberately absent: its shape is the open +write-semantics decision (a unique constraint backing an atomic upsert, or a +plain compound index mirroring MongoDB) and it ships with that decision. +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import JSONB + +# revision identifiers, used by Alembic. +revision: str = "771e95623724" +down_revision: str | None = "a1b2c3d4e5f6" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "task_states", + sa.Column("id", sa.String(), nullable=False), + sa.Column("task_id", sa.String(), nullable=False), + sa.Column("agent_id", sa.String(), nullable=False), + sa.Column("state", JSONB(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + # Cascade is a dormant safety net: no current flow hard-deletes task + # rows (API deletes are soft; retention deletes states explicitly and + # keeps the task row). + sa.ForeignKeyConstraint(["task_id"], ["tasks.id"], ondelete="CASCADE"), + # No cascade on agent_id: agents are never hard-deleted today, and + # silently dropping an agent's states if that changed would be the + # wrong default. + sa.ForeignKeyConstraint(["agent_id"], ["agents.id"]), + sa.PrimaryKeyConstraint("id"), + ) + # The index targets the table created in this same migration, so it holds + # no write-blocking lock against live traffic (the table has no rows yet). + op.create_index( + "ix_task_states_agent_id", + "task_states", + ["agent_id"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index("ix_task_states_agent_id", table_name="task_states") + op.drop_table("task_states") diff --git a/agentex/src/adapters/orm.py b/agentex/src/adapters/orm.py index e5f7b139..d0ef4c7b 100644 --- a/agentex/src/adapters/orm.py +++ b/agentex/src/adapters/orm.py @@ -147,6 +147,45 @@ class AgentTaskTrackerORM(BaseORM): ) +class TaskStateORM(BaseORM): + """Task state on PostgreSQL: one row per (task, agent) pair. + + Optional storage backend for task state (selected per deployment via + TASK_STATE_STORAGE_PHASE); MongoDB remains the default. Nothing reads or + writes this table until the Postgres task-state repository lands. + """ + + __tablename__ = "task_states" + + id = Column(String, primary_key=True, default=orm_id) + # Cascade is a dormant safety net: no current flow hard-deletes task rows + # (API deletes are soft; retention deletes states explicitly and keeps the + # task row). + task_id = Column(String, ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False) + # No cascade: agents are never hard-deleted today, and if that changed, + # silently dropping an agent's states while its messages survive would be + # the wrong default. + agent_id = Column(String, ForeignKey("agents.id"), nullable=False) + state = Column(JSONB, nullable=False) + created_at = Column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + updated_at = Column( + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + __table_args__ = ( + # The (task_id, agent_id) index is deliberately absent: its shape is + # the open write-semantics decision (a unique constraint backing an + # atomic upsert, or a plain compound index mirroring MongoDB) and it + # ships with the repository that reads this table. + Index("ix_task_states_agent_id", "agent_id"), + ) + + class SpanORM(BaseORM): __tablename__ = "spans" id = Column(String, primary_key=True, default=orm_id) # Using UUIDs for IDs From 554fad4a3305e6d4cf8df7c06a233c710015ea29 Mon Sep 17 00:00:00 2001 From: Vineet Vora Date: Tue, 4 Aug 2026 14:55:02 -0400 Subject: [PATCH 2/3] fix(states): use bare FKs on task_states, no ON DELETE CASCADE MongoDB has no cascades, so state deletion is application-driven on both backends; a Postgres-only cascade would diverge silently on a future task hard-delete. The FK alone prevents orphaned rows. --- .../2026_08_03_1500_add_task_states_771e95623724.py | 12 +++++------- agentex/src/adapters/orm.py | 12 +++++------- 2 files changed, 10 insertions(+), 14 deletions(-) diff --git a/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py b/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py index 5cad4ad0..9b8d015d 100644 --- a/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py +++ b/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py @@ -47,13 +47,11 @@ def upgrade() -> None: server_default=sa.text("now()"), nullable=False, ), - # Cascade is a dormant safety net: no current flow hard-deletes task - # rows (API deletes are soft; retention deletes states explicitly and - # keeps the task row). - sa.ForeignKeyConstraint(["task_id"], ["tasks.id"], ondelete="CASCADE"), - # No cascade on agent_id: agents are never hard-deleted today, and - # silently dropping an agent's states if that changed would be the - # wrong default. + # Both FKs are bare (no ON DELETE action), deliberately: MongoDB has no + # cascades, so state deletion is application-driven on both backends. + # The FK alone prevents orphaned rows; a future hard-delete flow must + # remove states through the repository or fail loudly here. + sa.ForeignKeyConstraint(["task_id"], ["tasks.id"]), sa.ForeignKeyConstraint(["agent_id"], ["agents.id"]), sa.PrimaryKeyConstraint("id"), ) diff --git a/agentex/src/adapters/orm.py b/agentex/src/adapters/orm.py index d0ef4c7b..934f2af5 100644 --- a/agentex/src/adapters/orm.py +++ b/agentex/src/adapters/orm.py @@ -158,13 +158,11 @@ class TaskStateORM(BaseORM): __tablename__ = "task_states" id = Column(String, primary_key=True, default=orm_id) - # Cascade is a dormant safety net: no current flow hard-deletes task rows - # (API deletes are soft; retention deletes states explicitly and keeps the - # task row). - task_id = Column(String, ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False) - # No cascade: agents are never hard-deleted today, and if that changed, - # silently dropping an agent's states while its messages survive would be - # the wrong default. + # Both FKs are bare (no ON DELETE action), deliberately: MongoDB has no + # cascades, so state deletion is application-driven on both backends. The + # FK alone prevents orphaned rows; a future hard-delete flow must remove + # states through the repository or fail loudly here. + task_id = Column(String, ForeignKey("tasks.id"), nullable=False) agent_id = Column(String, ForeignKey("agents.id"), nullable=False) state = Column(JSONB, nullable=False) created_at = Column( From a832d8a52f5ac970d1b61a8bdead1a8d00c9a413 Mon Sep 17 00:00:00 2001 From: Vineet Vora Date: Tue, 4 Aug 2026 15:34:38 -0400 Subject: [PATCH 3/3] docs(states): correct FK comment, flag timestamp default trap DELETE /tasks issues a real row delete today and already fails on the sibling child tables' bare FKs; the earlier comment implied no hard-delete path exists. Also warn that explicitly-None timestamps bypass the server default and violate NOT NULL. --- .../2026_08_03_1500_add_task_states_771e95623724.py | 9 ++++++--- agentex/src/adapters/orm.py | 12 +++++++++--- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py b/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py index 9b8d015d..48ea95e1 100644 --- a/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py +++ b/agentex/database/migrations/alembic/versions/2026_08_03_1500_add_task_states_771e95623724.py @@ -48,9 +48,12 @@ def upgrade() -> None: nullable=False, ), # Both FKs are bare (no ON DELETE action), deliberately: MongoDB has no - # cascades, so state deletion is application-driven on both backends. - # The FK alone prevents orphaned rows; a future hard-delete flow must - # remove states through the repository or fail loudly here. + # cascades, so state deletion is application-driven on both backends, + # and the FK alone prevents orphaned rows. This matches the sibling + # child tables of tasks (task_agents, events), whose bare FKs already + # reject the existing DELETE /tasks path for any task with children; + # task_states behaves identically rather than cascading on one backend + # only. sa.ForeignKeyConstraint(["task_id"], ["tasks.id"]), sa.ForeignKeyConstraint(["agent_id"], ["agents.id"]), sa.PrimaryKeyConstraint("id"), diff --git a/agentex/src/adapters/orm.py b/agentex/src/adapters/orm.py index 934f2af5..95a95963 100644 --- a/agentex/src/adapters/orm.py +++ b/agentex/src/adapters/orm.py @@ -159,12 +159,18 @@ class TaskStateORM(BaseORM): id = Column(String, primary_key=True, default=orm_id) # Both FKs are bare (no ON DELETE action), deliberately: MongoDB has no - # cascades, so state deletion is application-driven on both backends. The - # FK alone prevents orphaned rows; a future hard-delete flow must remove - # states through the repository or fail loudly here. + # cascades, so state deletion is application-driven on both backends, and + # the FK alone prevents orphaned rows. This matches the sibling child + # tables of tasks (task_agents, events), whose bare FKs already reject the + # existing DELETE /tasks path for any task with children; task_states + # behaves identically rather than cascading on one backend only. task_id = Column(String, ForeignKey("tasks.id"), nullable=False) agent_id = Column(String, ForeignKey("agents.id"), nullable=False) state = Column(JSONB, nullable=False) + # NOT NULL with a server default: the repository must OMIT unset (None) + # timestamps when constructing rows. SQLAlchemy renders an explicitly + # assigned None as a literal NULL, which violates the constraint instead + # of falling back to the default (StateEntity defaults these to None). created_at = Column( DateTime(timezone=True), server_default=func.now(), nullable=False )