feat(AGX1-274): record task creator identity (user/service-account) on creation

Adds two nullable creator-audit columns to the tasks table — creator_user_id
and creator_service_account_id — populated from the principal context at
create time. A CHECK constraint (ck_tasks_one_creator) enforces that at most
one is set.

This replaces the earlier dual-write draft: grants are already issued
unconditionally via grant_with_retry in agents_acp_use_case.py:239, and
per-account rollout routing belongs in agentex-auth (private), not in this
public Apache-2.0 codebase.

Author: asherfink

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

@@ -0,0 +1,56 @@
+"""add_task_creator_columns
+
+Revision ID: a1f73ada66c5
+Revises: a9959ebcbe98
+Create Date: 2026-05-21 15:08:51.441535
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a1f73ada66c5"
+down_revision: str | None = "a9959ebcbe98"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("tasks", sa.Column("creator_user_id", sa.String(), nullable=True))
+    op.add_column(
+        "tasks", sa.Column("creator_service_account_id", sa.String(), nullable=True)
+    )
+    with op.get_context().autocommit_block():
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_tasks_creator_user_id "
+            "ON tasks (creator_user_id)"
+        )
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_tasks_creator_service_account_id "
+            "ON tasks (creator_service_account_id)"
+        )
+    # Add the CHECK as NOT VALID so the brief ACCESS EXCLUSIVE lock doesn't have to
+    # wait on an existence scan, then VALIDATE under SHARE UPDATE EXCLUSIVE which
+    # doesn't block concurrent reads/writes. `tasks` is a high-write table; even the
+    # short ACCESS EXCLUSIVE held during a vanilla CHECK addition queues behind
+    # in-flight transactions and blocks readers until it releases.
+    op.execute(
+        "ALTER TABLE tasks ADD CONSTRAINT ck_tasks_at_most_one_creator "
+        "CHECK ((creator_user_id IS NULL) OR (creator_service_account_id IS NULL)) "
+        "NOT VALID"
+    )
+    op.execute("ALTER TABLE tasks VALIDATE CONSTRAINT ck_tasks_at_most_one_creator")
+
+
+def downgrade() -> None:
+    op.drop_constraint("ck_tasks_at_most_one_creator", "tasks", type_="check")
+    with op.get_context().autocommit_block():
+        op.execute(
+            "DROP INDEX CONCURRENTLY IF EXISTS ix_tasks_creator_service_account_id"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_tasks_creator_user_id")
+    op.drop_column("tasks", "creator_service_account_id")
+    op.drop_column("tasks", "creator_user_id")

agentex/database/migrations/migration_history.txt

@@ -1,4 +1,6 @@
-9ff3ee32c81b -> e9c4ff9e6542 (head), add_tasks_metadata_gin_index
+a9959ebcbe98 -> a1f73ada66c5 (head), add_task_creator_columns
+e9c4ff9e6542 -> a9959ebcbe98, finalize_spans_task_id
+9ff3ee32c81b -> e9c4ff9e6542, add_tasks_metadata_gin_index
 57c5ed4f59ae -> 9ff3ee32c81b, uppercase deployment status enum labels
 4a9b7787ccd7 -> 57c5ed4f59ae, add_task_id_to_spans
 d1a6cde41b3f -> 4a9b7787ccd7, deployments

agentex/src/adapters/orm.py

@@ -2,6 +2,7 @@
     JSON,
     BigInteger,
     Boolean,
+    CheckConstraint,
     Column,
     DateTime,
     ForeignKey,
@@ -74,13 +75,17 @@ class TaskORM(BaseORM):
     )
     params = Column(JSONB, nullable=True)
     task_metadata = Column(JSONB, nullable=True)
+    creator_user_id = Column(String, nullable=True, index=True)
+    creator_service_account_id = Column(String, nullable=True, index=True)
     # Many-to-Many relationship with agents
     agents = relationship("AgentORM", secondary="task_agents", back_populates="tasks")
 
-    # Indexes for efficient querying
     __table_args__ = (
-        # Index for filtering tasks by status (used in list queries)
         Index("ix_tasks_status", "status"),
+        CheckConstraint(
+            "creator_user_id IS NULL OR creator_service_account_id IS NULL",
+            name="ck_tasks_at_most_one_creator",
+        ),
     )
 
 

agentex/src/domain/entities/tasks.py

@@ -58,6 +58,14 @@ class TaskEntity(BaseModel):
         None,
         title="Task metadata",
     )
+    creator_user_id: str | None = Field(
+        None,
+        title="Identity ID of the user who created this task",
+    )
+    creator_service_account_id: str | None = Field(
+        None,
+        title="Service identity ID of the service account that created this task",
+    )
 
     # allow extra fields for agents relationships
     model_config = ConfigDict(extra="allow")

agentex/src/domain/services/task_service.py

@@ -14,13 +14,25 @@
 from src.domain.repositories.task_repository import DTaskRepository
 from src.domain.repositories.task_state_repository import DTaskStateRepository
 from src.domain.services.agent_acp_service import DAgentACPService
+from src.domain.services.authorization_service import DAuthorizationService
 from src.utils.ids import orm_id
 from src.utils.logging import make_logger
 from src.utils.stream_topics import get_task_event_stream_topic
 
 logger = make_logger(__name__)
 
 
+def _principal_field(principal_context: Any, key: str) -> str | None:
+    """Read an attribute from the principal context, which may be either a
+    Pydantic-style object or a plain dict. The authn proxy returns a dict
+    (``response.json()`` shape), so ``getattr`` alone silently yields None."""
+    if principal_context is None:
+        return None
+    if isinstance(principal_context, dict):
+        return principal_context.get(key)
+    return getattr(principal_context, key, None)
+
+
 class AgentTaskService:
     """
     Service for managing agent tasks and forwarding operations to ACP servers.
@@ -33,12 +45,14 @@ def __init__(
         task_repository: DTaskRepository,
         event_repository: DEventRepository,
         stream_repository: DRedisStreamRepository,
+        authorization_service: DAuthorizationService,
     ):
         self.acp_client = acp_client
         self.task_state_repository = task_state_repository
         self.task_repository = task_repository
         self.event_repository = event_repository
         self.stream_repository = stream_repository
+        self.authorization_service = authorization_service
 
     async def create_task(
         self,
@@ -49,16 +63,12 @@ async def create_task(
     ) -> TaskEntity:
         """
         Create a new task record in the repository with single agent (maintains existing interface).
-
-        Args:
-            agent: The agent to create the task for
-            task_name: The name of the task to be created
-            task_params: The parameters for the task
-            task_metadata: Caller-provided metadata to persist on the task row.
-                Not forwarded to the agent.
-        Returns:
-            Task containing the created task info
         """
+        principal_context = self.authorization_service.principal_context
+        creator_user_id = _principal_field(principal_context, "user_id")
+        creator_service_account_id = _principal_field(
+            principal_context, "service_account_id"
+        )
 
         task_entity = await self.task_repository.create(
             agent_id=agent.id,
@@ -69,6 +79,8 @@ async def create_task(
                 status_reason="Task created, forwarding to ACP server",
                 params=task_params,
                 task_metadata=task_metadata,
+                creator_user_id=creator_user_id,
+                creator_service_account_id=creator_service_account_id,
             ),
         )
         return task_entity
@@ -82,16 +94,11 @@ async def create_task_and_forward_to_acp(
         """
         Create a new task record in the repository with single agent (maintains existing interface).
         Then, forward the task to the ACP server.
-
-        Args:
-            agent: The agent to create the task for
-            task_params: The parameters for the task to be sent to the ACP server
-
-        Returns:
-            Task containing the created task info
         """
         task_entity = await self.create_task(
-            agent=agent, task_name=task_name, task_params=task_params
+            agent=agent,
+            task_name=task_name,
+            task_params=task_params,
         )
 
         if agent.acp_type == ACPType.SYNC:

agentex/tests/fixtures/services.py

@@ -3,7 +3,7 @@
 Provides factory functions and specific fixtures for creating services with test repositories.
 """
 
-from unittest.mock import MagicMock, Mock
+from unittest.mock import AsyncMock, MagicMock, Mock
 
 import pytest
 
@@ -12,6 +12,21 @@
 # =============================================================================
 
 
+def make_noop_authorization_service() -> Mock:
+    """Shared noop AuthorizationService mock for tests that don't exercise authz.
+
+    ``principal_context`` is ``None`` (so creator-audit columns resolve to NULL),
+    and ``grant``/``revoke`` are async no-ops returning ``None`` — matching the
+    real service signature. Use this anywhere a test just needs to construct
+    ``AgentTaskService`` without caring about authorization behavior.
+    """
+    svc = Mock()
+    svc.principal_context = None
+    svc.grant = AsyncMock(return_value=None)
+    svc.revoke = AsyncMock(return_value=None)
+    return svc
+
+
 def create_task_message_service(task_message_repository):
     """Factory function to create TaskMessageService with given repository"""
     from src.domain.services.task_message_service import TaskMessageService
@@ -52,16 +67,21 @@ def create_task_service(
     event_repository,
     agent_acp_service,
     redis_stream_repository,
+    authorization_service=None,
 ):
-    """Factory function to create AgentTaskService with given repositories and services"""
+    """Factory function to create AgentTaskService with given repositories and services."""
     from src.domain.services.task_service import AgentTaskService
 
+    if authorization_service is None:
+        authorization_service = make_noop_authorization_service()
+
     return AgentTaskService(
         task_repository=task_repository,
         task_state_repository=task_state_repository,
         event_repository=event_repository,
         acp_client=agent_acp_service,
         stream_repository=redis_stream_repository,
+        authorization_service=authorization_service,
     )
 
 

agentex/tests/integration/fixtures/integration_client.py

@@ -22,6 +22,8 @@
 from src.config.dependencies import GlobalDependencies
 from src.config.environment_variables import EnvironmentVariables
 
+from tests.fixtures.services import make_noop_authorization_service
+
 
 @pytest.fixture(scope="session")
 def event_loop():
@@ -447,6 +449,7 @@ async def send_message(self, *args, **kwargs):
             task_repository=isolated_repositories["task_repository"],
             event_repository=isolated_repositories["event_repository"],
             stream_repository=isolated_repositories["redis_stream_repository"],
+            authorization_service=make_noop_authorization_service(),
         )
 
         return TasksUseCase(task_service=task_service)

agentex/tests/integration/test_task_stream.py

@@ -7,6 +7,8 @@
 from src.domain.use_cases.tasks_use_case import TasksUseCase
 from src.utils.ids import orm_id
 
+from tests.fixtures.services import make_noop_authorization_service
+
 
 @pytest.mark.asyncio
 @pytest.mark.integration
@@ -76,6 +78,7 @@ async def send_message(self, *args, **kwargs):
             task_repository=isolated_repositories["task_repository"],
             event_repository=isolated_repositories["event_repository"],
             stream_repository=isolated_repositories["redis_stream_repository"],
+            authorization_service=make_noop_authorization_service(),
         )
 
         return TasksUseCase(task_service=task_service)
@@ -103,6 +106,7 @@ async def send_message(self, *args, **kwargs):
             task_repository=isolated_repositories["task_repository"],
             event_repository=isolated_repositories["event_repository"],
             stream_repository=isolated_repositories["redis_stream_repository"],
+            authorization_service=make_noop_authorization_service(),
         )
 
         environment_variables = EnvironmentVariables.refresh()