diff --git a/agentex/openapi.yaml b/agentex/openapi.yaml index 5e779da4..79781ac5 100644 --- a/agentex/openapi.yaml +++ b/agentex/openapi.yaml @@ -4075,6 +4075,16 @@ components: title: Task Name description: The name of the task to cancel. Either this or task_id must be provided. + end_user_id: + anyOf: + - type: string + maxLength: 256 + - type: 'null' + title: End User Id + description: Opaque identifier for the end user this call is made on behalf + of. Not persisted on the task row, so supply it on every call that should + be attributed. Forwarded to the agent inside the ACP payload, which stamps + it onto the agent's trace spans as __end_user_id__. type: object title: CancelTaskRequest CheckpointListItem: @@ -4520,6 +4530,16 @@ components: description: Caller-provided metadata to persist on the task row. Only applied at task creation; ignored if a task with this name already exists. Forwarded to the agent inside the ACP payload for backward compatibility. + end_user_id: + anyOf: + - type: string + maxLength: 256 + - type: 'null' + title: End User Id + description: Opaque identifier for the end user this call is made on behalf + of. Not persisted on the task row, so supply it on every call that should + be attributed. Forwarded to the agent inside the ACP payload, which stamps + it onto the agent's trace spans as __end_user_id__. type: object title: CreateTaskRequest DataContent: @@ -5610,6 +5630,16 @@ components: - $ref: '#/components/schemas/TaskMessageContent' - type: 'null' description: The content to send to the event + end_user_id: + anyOf: + - type: string + maxLength: 256 + - type: 'null' + title: End User Id + description: Opaque identifier for the end user this call is made on behalf + of. Not persisted on the task row, so supply it on every call that should + be attributed. Forwarded to the agent inside the ACP payload, which stamps + it onto the agent's trace spans as __end_user_id__. type: object title: SendEventRequest SendMessageRequest: @@ -5641,6 +5671,16 @@ components: - type: 'null' title: Task Params description: The parameters for the task (only used when creating new tasks) + end_user_id: + anyOf: + - type: string + maxLength: 256 + - type: 'null' + title: End User Id + description: Opaque identifier for the end user this call is made on behalf + of. Not persisted on the task row, so supply it on every call that should + be attributed. Forwarded to the agent inside the ACP payload, which stamps + it onto the agent's trace spans as __end_user_id__. type: object required: - content diff --git a/agentex/src/api/schemas/agents_rpc.py b/agentex/src/api/schemas/agents_rpc.py index 884d55d9..7ff4e675 100644 --- a/agentex/src/api/schemas/agents_rpc.py +++ b/agentex/src/api/schemas/agents_rpc.py @@ -13,6 +13,17 @@ logger = make_logger(__name__) +# Bounded because the value is forwarded to the agent on every call and, for +# Temporal agents, copied into activity headers and therefore into workflow +# history repeatedly. +END_USER_ID_MAX_LENGTH = 256 +END_USER_ID_DESCRIPTION = ( + "Opaque identifier for the end user this call is made on behalf of. Not " + "persisted on the task row, so supply it on every call that should be " + "attributed. Forwarded to the agent inside the ACP payload, which stamps it " + "onto the agent's trace spans as __end_user_id__." +) + class AgentRPCMethod(str, Enum): EVENT_SEND = "event/send" @@ -34,6 +45,11 @@ class CreateTaskRequest(BaseModel): "Forwarded to the agent inside the ACP payload for backward compatibility." ), ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) class CancelTaskRequest(BaseModel): @@ -45,6 +61,11 @@ class CancelTaskRequest(BaseModel): None, description="The name of the task to cancel. Either this or task_id must be provided.", ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) @model_validator(mode="after") def validate_task_identifiers(self): @@ -70,6 +91,11 @@ class SendMessageRequest(BaseModel): None, description="The parameters for the task (only used when creating new tasks)", ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) @model_validator(mode="after") def validate_task_identifiers(self): @@ -88,6 +114,11 @@ class SendEventRequest(BaseModel): content: TaskMessageContent | None = Field( None, description="The content to send to the event" ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) @model_validator(mode="after") def validate_task_identifiers(self): diff --git a/agentex/src/domain/entities/agents_rpc.py b/agentex/src/domain/entities/agents_rpc.py index 864c9469..bb3ff164 100644 --- a/agentex/src/domain/entities/agents_rpc.py +++ b/agentex/src/domain/entities/agents_rpc.py @@ -4,6 +4,8 @@ from pydantic import BaseModel, Field, model_validator from src.api.schemas.agents_rpc import ( + END_USER_ID_DESCRIPTION, + END_USER_ID_MAX_LENGTH, AgentRPCRequest, CancelTaskRequest, CreateTaskRequest, @@ -30,6 +32,12 @@ class AgentRPCMethod(str, Enum): EVENT_SEND = "event/send" +ACP_END_USER_ID_DESCRIPTION = ( + "The end user this call was made on behalf of, when the caller supplied one. " + "The agent SDK stamps it onto trace spans as __end_user_id__." +) + + class CreateTaskParams(BaseModel): """Parameters for task/create method""" @@ -42,6 +50,7 @@ class CreateTaskParams(BaseModel): None, description="The parameters for the task as inputted by the user", ) + end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION) class SendMessageParams(BaseModel): @@ -58,6 +67,7 @@ class SendMessageParams(BaseModel): stream: bool = Field( False, description="Whether to stream the message to the agent" ) + end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION) class SendEventParams(BaseModel): @@ -73,6 +83,7 @@ class SendEventParams(BaseModel): None, description="Additional request context including headers", ) + end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION) class CancelTaskParams(BaseModel): @@ -83,6 +94,7 @@ class CancelTaskParams(BaseModel): description="The agent that the task was sent to", ) task: TaskEntity = Field(..., description="The task that was cancelled") + end_user_id: str | None = Field(None, description=ACP_END_USER_ID_DESCRIPTION) ACP_TYPE_TO_ALLOWED_RPC_METHODS = { @@ -113,6 +125,11 @@ class CreateTaskRequestEntity(BaseModel): "Forwarded to the agent inside the ACP payload for backward compatibility." ), ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) class CancelTaskRequestEntity(BaseModel): @@ -124,6 +141,11 @@ class CancelTaskRequestEntity(BaseModel): None, description="The name of the task to cancel. Either this or task_id must be provided.", ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) @model_validator(mode="after") def validate_task_identifiers(self): @@ -149,6 +171,11 @@ class SendMessageRequestEntity(BaseModel): None, description="The parameters for the task (only used when creating new tasks)", ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) @model_validator(mode="after") def validate_task_identifiers(self): @@ -167,6 +194,11 @@ class SendEventRequestEntity(BaseModel): content: TaskMessageContentEntity | None = Field( None, description="The content to send to the event" ) + end_user_id: str | None = Field( + None, + max_length=END_USER_ID_MAX_LENGTH, + description=END_USER_ID_DESCRIPTION, + ) @model_validator(mode="after") def validate_task_identifiers(self): @@ -193,6 +225,7 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self: name=request.params.root.name, params=request.params.root.params, task_metadata=request.params.root.task_metadata, + end_user_id=request.params.root.end_user_id, ) elif request.method == AgentRPCMethod.TASK_CANCEL and isinstance( request.params.root, CancelTaskRequest @@ -200,6 +233,7 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self: params = CancelTaskRequestEntity( task_id=request.params.root.task_id, task_name=request.params.root.task_name, + end_user_id=request.params.root.end_user_id, ) elif request.method == AgentRPCMethod.MESSAGE_SEND: content_entity = convert_task_message_content_to_entity( @@ -211,6 +245,7 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self: content=content_entity, stream=request.params.root.stream, task_params=request.params.root.task_params, + end_user_id=request.params.root.end_user_id, ) elif request.method == AgentRPCMethod.EVENT_SEND: if request.params.root.content is not None: @@ -223,6 +258,7 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self: task_id=request.params.root.task_id, task_name=request.params.root.task_name, content=content_entity, + end_user_id=request.params.root.end_user_id, ) else: logger.error(f"Invalid method from request: {request}") diff --git a/agentex/src/domain/services/agent_acp_service.py b/agentex/src/domain/services/agent_acp_service.py index fd3da39f..2ce2873b 100644 --- a/agentex/src/domain/services/agent_acp_service.py +++ b/agentex/src/domain/services/agent_acp_service.py @@ -304,12 +304,14 @@ async def create_task( task: TaskEntity, acp_url: str, params: dict[str, Any] | None = None, + end_user_id: str | None = None, ) -> dict[str, Any]: """Create a new task""" params = CreateTaskParams( agent=agent, task=task, params=params, + end_user_id=end_user_id, ) headers = await self.get_headers(agent) return await self._call_jsonrpc( @@ -326,6 +328,7 @@ async def send_message( task: TaskEntity, content: TaskMessageContentEntity, acp_url: str, + end_user_id: str | None = None, ) -> TaskMessageContentEntity: """Send a message to a running task""" params = SendMessageParams( @@ -333,6 +336,7 @@ async def send_message( task=task, content=content, stream=False, + end_user_id=end_user_id, ) headers = await self.get_headers(agent) lock_key = hash((agent.id, task.id)) @@ -357,6 +361,7 @@ async def send_message_stream( task: TaskEntity, content: TaskMessageContentEntity, acp_url: str, + end_user_id: str | None = None, ) -> AsyncIterator[TaskMessageUpdateEntity]: """Send a message to a running task and stream the response""" params = SendMessageParams( @@ -364,6 +369,7 @@ async def send_message_stream( task=task, content=content, stream=True, + end_user_id=end_user_id, ) headers = await self.get_headers(agent) lock_key = hash((agent.id, task.id)) @@ -398,10 +404,14 @@ async def send_message_stream( yield self._parse_task_message_update(chunk) async def cancel_task( - self, agent: AgentEntity, task: TaskEntity, acp_url: str + self, + agent: AgentEntity, + task: TaskEntity, + acp_url: str, + end_user_id: str | None = None, ) -> dict[str, Any]: """Cancel a running task""" - params = CancelTaskParams(agent=agent, task=task) + params = CancelTaskParams(agent=agent, task=task, end_user_id=end_user_id) headers = await self.get_headers(agent) return await self._call_jsonrpc( url=acp_url, @@ -418,6 +428,7 @@ async def send_event( task: TaskEntity, acp_url: str, request_headers: dict[str, str] | None = None, + end_user_id: str | None = None, ) -> dict[str, Any]: """Send an event to a running task""" @@ -426,6 +437,7 @@ async def send_event( task=task, event=event, request=None, + end_user_id=end_user_id, ) headers = await self.get_headers(agent, request_headers) diff --git a/agentex/src/domain/services/task_service.py b/agentex/src/domain/services/task_service.py index 013c6903..37cf1d48 100644 --- a/agentex/src/domain/services/task_service.py +++ b/agentex/src/domain/services/task_service.py @@ -78,6 +78,7 @@ async def create_task_and_forward_to_acp( agent: AgentEntity, task_name: str | None = None, task_params: dict[str, Any] | None = None, + end_user_id: str | None = None, ) -> TaskEntity: """ Create a new task record in the repository with single agent (maintains existing interface). @@ -105,6 +106,7 @@ async def create_task_and_forward_to_acp( task=task_entity, acp_url=agent.acp_url, params=task_params, + end_user_id=end_user_id, ) return task_entity except Exception as e: @@ -118,6 +120,7 @@ async def forward_task_to_acp( task: TaskEntity, task_params: dict[str, Any] | None = None, acp_url: str | None = None, + end_user_id: str | None = None, ) -> None: try: await self.acp_client.create_task( @@ -125,6 +128,7 @@ async def forward_task_to_acp( task=task, acp_url=acp_url or agent.acp_url, params=task_params, + end_user_id=end_user_id, ) except Exception as e: logger.error(f"Error creating task in ACP: {e}") @@ -257,6 +261,7 @@ async def send_message( task: TaskEntity, content: TaskMessageContentEntity, acp_url: str, + end_user_id: str | None = None, ) -> TaskMessageContentEntity: """Send a message to a running task""" return await self.acp_client.send_message( @@ -264,6 +269,7 @@ async def send_message( task=task, content=content, acp_url=acp_url, + end_user_id=end_user_id, ) async def send_message_stream( @@ -272,6 +278,7 @@ async def send_message_stream( task: TaskEntity, content: TaskMessageContentEntity, acp_url: str, + end_user_id: str | None = None, ) -> AsyncIterator[TaskMessageUpdateEntity]: """Send a message to a running task and stream the response""" logger.info(f"TaskService: Sending message stream for task {task.id}") @@ -280,14 +287,21 @@ async def send_message_stream( task=task, content=content, acp_url=acp_url, + end_user_id=end_user_id, ): yield chunk async def cancel_task( - self, agent: AgentEntity, task: TaskEntity, acp_url: str + self, + agent: AgentEntity, + task: TaskEntity, + acp_url: str, + end_user_id: str | None = None, ) -> TaskEntity: """Cancel a running task""" - await self.acp_client.cancel_task(agent=agent, task=task, acp_url=acp_url) + await self.acp_client.cancel_task( + agent=agent, task=task, acp_url=acp_url, end_user_id=end_user_id + ) task = await self.task_repository.get(id=task.id) task.status = TaskStatus.CANCELED @@ -301,6 +315,7 @@ async def create_event_and_forward_to_acp( acp_url: str, content: TaskMessageContentEntity | None = None, request_headers: dict[str, str] | None = None, + end_user_id: str | None = None, ) -> EventEntity: """Create an event and forward it to the ACP server""" event = await self.event_repository.create( @@ -315,6 +330,7 @@ async def create_event_and_forward_to_acp( task=task, acp_url=acp_url, request_headers=request_headers, + end_user_id=end_user_id, ) return event diff --git a/agentex/src/domain/use_cases/agents_acp_use_case.py b/agentex/src/domain/use_cases/agents_acp_use_case.py index fc727d88..04148ffe 100644 --- a/agentex/src/domain/use_cases/agents_acp_use_case.py +++ b/agentex/src/domain/use_cases/agents_acp_use_case.py @@ -427,6 +427,7 @@ async def _handle_task_create( task=task, task_params=params.params, acp_url=acp_url, + end_user_id=params.end_user_id, ) return task @@ -503,6 +504,7 @@ async def flush_aggregated_deltas( task=task, content=params.content, acp_url=acp_url, + end_user_id=params.end_user_id, ): logger.debug( f"[message_send_stream] Received message chunk: {task_message_update}" @@ -656,6 +658,7 @@ async def flush_aggregated_deltas(task_message_index: int) -> TaskMessageEntity: task=task, content=params.content, acp_url=acp_url, + end_user_id=params.end_user_id, ): logger.debug( f"[message_send_stream] Received message chunk type: {type(task_message_update).__name__}" @@ -799,6 +802,7 @@ async def _handle_task_cancel( agent=agent, task=task, acp_url=acp_url, + end_user_id=params.end_user_id, ) async def _handle_event_send( @@ -834,6 +838,7 @@ async def _handle_event_send( content=params.content, acp_url=acp_url, request_headers=request_headers, + end_user_id=params.end_user_id, ) return event_entity diff --git a/agentex/tests/unit/domain/test_agents_rpc_end_user_id.py b/agentex/tests/unit/domain/test_agents_rpc_end_user_id.py new file mode 100644 index 00000000..7c6903ac --- /dev/null +++ b/agentex/tests/unit/domain/test_agents_rpc_end_user_id.py @@ -0,0 +1,80 @@ +"""Unit tests for the end_user_id RPC field. + +end_user_id exists so a caller can attribute a request to an end user without +that identifier having to be threaded through agent business logic. It is +deliberately available on *every* method rather than only task/create, because +task_metadata (the pre-existing carrier) is only accepted at task creation and +is ignored on get-or-create by name, which leaves message/send-driven agents and +shared named tasks with no way to attribute the current caller. + +The field is wire-only: it is not persisted on the task row, so every call that +should be attributed has to carry it. +""" + +import pytest +from pydantic import ValidationError +from src.api.schemas.agents_rpc import ( + END_USER_ID_MAX_LENGTH, + AgentRPCMethod, + AgentRPCRequest, +) +from src.domain.entities.agents_rpc import AgentRPCRequestEntity + +_TEXT_CONTENT = {"author": "user", "type": "text", "content": "hello"} + +_PARAMS_BY_METHOD = { + AgentRPCMethod.TASK_CREATE: {"name": "task-1", "params": {"a": 1}}, + AgentRPCMethod.TASK_CANCEL: {"task_id": "task-1"}, + AgentRPCMethod.MESSAGE_SEND: {"task_id": "task-1", "content": _TEXT_CONTENT}, + AgentRPCMethod.EVENT_SEND: {"task_id": "task-1", "content": _TEXT_CONTENT}, +} + + +def _request(method: AgentRPCMethod, **extra_params) -> AgentRPCRequest: + return AgentRPCRequest( + jsonrpc="2.0", + id=1, + method=method, + params={**_PARAMS_BY_METHOD[method], **extra_params}, + ) + + +class TestEndUserIdOnRequestSchemas: + @pytest.mark.parametrize("method", list(_PARAMS_BY_METHOD)) + def test_accepted_on_every_method(self, method: AgentRPCMethod): + assert ( + _request(method, end_user_id="user-a").params.root.end_user_id == "user-a" + ) + + @pytest.mark.parametrize("method", list(_PARAMS_BY_METHOD)) + def test_defaults_to_none_when_omitted(self, method: AgentRPCMethod): + assert _request(method).params.root.end_user_id is None + + @pytest.mark.parametrize("method", list(_PARAMS_BY_METHOD)) + def test_rejects_over_length_value(self, method: AgentRPCMethod): + # Bounded because Temporal agents copy the value into activity headers, + # and therefore into workflow history, on every activity invocation. + with pytest.raises(ValidationError): + _request(method, end_user_id="x" * (END_USER_ID_MAX_LENGTH + 1)) + + @pytest.mark.parametrize("method", list(_PARAMS_BY_METHOD)) + def test_accepts_value_at_length_limit(self, method: AgentRPCMethod): + value = "x" * END_USER_ID_MAX_LENGTH + assert _request(method, end_user_id=value).params.root.end_user_id == value + + +class TestEndUserIdSurvivesEntityConversion: + """from_api_request maps each method's params field-by-field, so a new field + is silently dropped unless it is added to every branch.""" + + @pytest.mark.parametrize("method", list(_PARAMS_BY_METHOD)) + def test_mapped_for_every_method(self, method: AgentRPCMethod): + entity = AgentRPCRequestEntity.from_api_request( + _request(method, end_user_id="user-a") + ) + assert entity.params.end_user_id == "user-a" + + @pytest.mark.parametrize("method", list(_PARAMS_BY_METHOD)) + def test_none_when_omitted(self, method: AgentRPCMethod): + entity = AgentRPCRequestEntity.from_api_request(_request(method)) + assert entity.params.end_user_id is None diff --git a/agentex/tests/unit/services/test_task_service.py b/agentex/tests/unit/services/test_task_service.py index eb096eb1..3220ecc4 100644 --- a/agentex/tests/unit/services/test_task_service.py +++ b/agentex/tests/unit/services/test_task_service.py @@ -281,6 +281,7 @@ async def test_create_task_and_forward_to_acp_success( task=result, # Use the actual created task acp_url=sample_agent.acp_url, params=task_params, + end_user_id=None, ) async def test_create_task_and_forward_sync_agent_skips_acp( @@ -544,6 +545,7 @@ async def test_send_message( task=sample_task, content=sample_message_content, acp_url=acp_url, + end_user_id=None, ) # @@ -625,7 +627,7 @@ async def test_cancel_task( # Verify ACP client was called to cancel the task mock_acp_client.cancel_task.assert_called_once_with( - agent=sample_agent, task=created_task, acp_url=acp_url + agent=sample_agent, task=created_task, acp_url=acp_url, end_user_id=None ) # Verify task status is updated in the database @@ -675,6 +677,7 @@ async def test_create_event_and_forward_to_acp( task=created_task, acp_url=acp_url, request_headers=None, + end_user_id=None, ) async def test_create_event_and_forward_to_acp_with_headers( @@ -721,6 +724,7 @@ async def test_create_event_and_forward_to_acp_with_headers( task=created_task, acp_url=acp_url, request_headers=request_headers, + end_user_id=None, ) async def test_create_task_with_task_metadata( diff --git a/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py b/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py index 60914adc..5c0750ad 100644 --- a/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py +++ b/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py @@ -129,6 +129,7 @@ async def test_agentic_agent_forwards_task_to_acp(self): task=task, acp_url=agentic_agent.acp_url, params={"test": "params"}, + end_user_id=None, ) assert result == task @@ -229,6 +230,7 @@ async def test_async_agent_forwards_task_to_acp(self): task=task, acp_url=async_agent.acp_url, params={"test": "params"}, + end_user_id=None, ) assert result == task diff --git a/agentex/tests/unit/use_cases/test_agents_acp_use_case.py b/agentex/tests/unit/use_cases/test_agents_acp_use_case.py index b48751a4..4f317af6 100644 --- a/agentex/tests/unit/use_cases/test_agents_acp_use_case.py +++ b/agentex/tests/unit/use_cases/test_agents_acp_use_case.py @@ -1035,6 +1035,161 @@ async def mock_async_call(*args, **kwargs): assert second.id == first.id assert second.task_metadata == {"created_by_user_id": "user-a"} + async def test_handle_task_create_forwards_end_user_id( + self, agents_acp_use_case, mock_http_gateway, agent_repository, sample_agent + ): + """end_user_id reaches the agent in the ACP payload, where the SDK stamps it + onto trace spans.""" + await create_or_get_agent(agent_repository, sample_agent) + + from src.api.schemas.agents_rpc import CreateTaskRequest + + async def mock_async_call(*args, **kwargs): + payload = kwargs.get("payload", {}) + return { + "jsonrpc": "2.0", + "result": {"status": "created"}, + "id": payload.get("id", ""), + } + + mock_http_gateway.async_call.side_effect = mock_async_call + + import uuid + + await agents_acp_use_case._handle_task_create( + agent=sample_agent, + params=CreateTaskRequest( + name=f"test-task-euid-{uuid.uuid4().hex[:8]}", + end_user_id="user-a", + ), + acp_url=sample_agent.acp_url, + ) + + sent_payload = mock_http_gateway.async_call.call_args.kwargs["payload"] + assert sent_payload["params"]["end_user_id"] == "user-a" + + async def test_handle_task_create_forwards_none_when_end_user_id_absent( + self, agents_acp_use_case, mock_http_gateway, agent_repository, sample_agent + ): + """end_user_id is optional; omitting it must not fail or invent a value.""" + await create_or_get_agent(agent_repository, sample_agent) + + from src.api.schemas.agents_rpc import CreateTaskRequest + + async def mock_async_call(*args, **kwargs): + payload = kwargs.get("payload", {}) + return { + "jsonrpc": "2.0", + "result": {"status": "created"}, + "id": payload.get("id", ""), + } + + mock_http_gateway.async_call.side_effect = mock_async_call + + import uuid + + await agents_acp_use_case._handle_task_create( + agent=sample_agent, + params=CreateTaskRequest(name=f"test-task-noeuid-{uuid.uuid4().hex[:8]}"), + acp_url=sample_agent.acp_url, + ) + + sent_payload = mock_http_gateway.async_call.call_args.kwargs["payload"] + assert sent_payload["params"]["end_user_id"] is None + + async def test_handle_message_send_forwards_end_user_id( + self, + agents_acp_use_case, + mock_http_gateway, + agent_repository, + sample_agent, + sample_text_content, + ): + """message/send is the case task_metadata cannot serve: it is only accepted at + task creation, so a one-shot sync agent had no way to attribute a caller.""" + await create_or_get_agent(agent_repository, sample_agent) + + captured_payloads = [] + + def create_mock_stream(*args, **kwargs): + payload = kwargs.get("payload", {}) + captured_payloads.append(payload) + return AsyncStreamMock( + [ + { + "jsonrpc": "2.0", + "result": { + "type": "full", + "index": 0, + "content": { + "type": "text", + "author": "agent", + "style": "static", + "format": "plain", + "content": "ok", + "attachments": None, + }, + }, + "id": payload.get("id", ""), + }, + ] + ) + + mock_http_gateway.stream_call = create_mock_stream + + await agents_acp_use_case._handle_message_send_sync( + agent=sample_agent, + params=SendMessageRequestEntity( + task_id=None, + content=sample_text_content, + stream=False, + end_user_id="user-a", + ), + acp_url=sample_agent.acp_url, + ) + + assert len(captured_payloads) == 1 + assert captured_payloads[0]["params"]["end_user_id"] == "user-a" + + async def test_handle_event_send_forwards_end_user_id( + self, + agents_acp_use_case, + mock_http_gateway, + agent_repository, + task_service, + sample_agent, + sample_text_content, + ): + """Every subsequent turn carries its own end_user_id, so attribution follows the + current caller rather than whoever created the task.""" + await create_or_get_agent(agent_repository, sample_agent) + created_task = await task_service.create_task( + agent=sample_agent, task_name=f"euid-event-task-{uuid4().hex[:8]}" + ) + + async def mock_async_call(*args, **kwargs): + payload = kwargs.get("payload", {}) + return { + "jsonrpc": "2.0", + "result": {"status": "event_sent"}, + "id": payload.get("id", ""), + } + + mock_http_gateway.async_call.side_effect = mock_async_call + + await agents_acp_use_case._handle_event_send( + agent=sample_agent, + params=SendEventRequest( + task_id=created_task.id, + content=sample_text_content, + end_user_id="user-b", + ), + acp_url=sample_agent.acp_url, + ) + + sent_payload = mock_http_gateway.async_call.call_args.kwargs["payload"] + assert sent_payload["params"]["end_user_id"] == "user-b" + # async def test_handle_message_send_sync_error_handling( self,