Merge branch '1.0-dev' into guglielmoc/refactor_utils_and_helpers

Author: guglielmo-san

.github/workflows/itk.yaml

@@ -28,4 +28,4 @@ jobs:
         run: bash run_itk.sh
         working-directory: itk
         env:
-          A2A_SAMPLES_REVISION: itk-v.0.11-alpha
+          A2A_SAMPLES_REVISION: itk-v.015-alpha

.gitignore

@@ -18,3 +18,4 @@ docs/ai/ai_learnings.md
 itk/a2a-samples/
 itk/pyproto/
 itk/instruction.proto
+itk/logs/

itk/README.md

@@ -36,7 +36,7 @@ You must set the `A2A_SAMPLES_REVISION` environment variable to specify which re
 
 Example:
 ```bash
-export A2A_SAMPLES_REVISION=itk-v.0.11-alpha
+export A2A_SAMPLES_REVISION=itk-v.015-alpha
 ```
 
 ### 2. Execute Tests
@@ -52,3 +52,22 @@ The script will:
 - Checkout the specified revision.
 - Build the ITK service Docker image.
 - Run the tests and output results.
+
+## Debugging
+
+To enable debug logging and persist logs for inspection:
+
+1.  Set the `ITK_LOG_LEVEL` environment variable to `DEBUG`:
+    ```bash
+    export ITK_LOG_LEVEL=DEBUG
+    ```
+2.  Run the test script:
+    ```bash
+    ./run_itk.sh
+    ```
+
+When run in `DEBUG` mode:
+- The `logs/` directory will be created in this directory (if it doesn't exist).
+- The `logs/` directory will be mounted to the container.
+- The test execution will produce detailed logs in `logs/` (e.g., `agent_current.log`).
+- The `logs/` directory will **not** be removed during cleanup.

itk/main.py

@@ -2,6 +2,7 @@
 import asyncio
 import base64
 import logging
+import os
 import uuid
 
 import grpc
@@ -36,7 +37,8 @@
 from a2a.utils import TransportProtocol
 
 
-logging.basicConfig(level=logging.INFO)
+log_level = os.environ.get('ITK_LOG_LEVEL', 'INFO').upper()
+logging.basicConfig(level=log_level)
 logger = logging.getLogger(__name__)
 
 
@@ -352,8 +354,9 @@ async def main_async(http_port: int, grpc_port: int) -> None:
         grpc_port,
     )
 
+    uvicorn_log_level = os.environ.get('ITK_LOG_LEVEL', 'INFO').lower()
     config = uvicorn.Config(
-        app, host='127.0.0.1', port=http_port, log_level='info'
+        app, host='127.0.0.1', port=http_port, log_level=uvicorn_log_level
     )
     uvicorn_server = uvicorn.Server(config)
 

itk/run_itk.sh

@@ -1,6 +1,9 @@
 #!/bin/bash
 set -ex
 
+# Set default log level
+export ITK_LOG_LEVEL="${ITK_LOG_LEVEL:-INFO}"
+
 # Initialize default exit code
 RESULT=1
 
@@ -63,9 +66,21 @@ ITK_DIR=$(pwd)
 # Stop existing container if any
 docker rm -f itk-service || true
 
+# Create logs directory if debug
+if [ "${ITK_LOG_LEVEL^^}" = "DEBUG" ]; then
+  mkdir -p "$ITK_DIR/logs"
+fi
+
+DOCKER_MOUNT_LOGS=""
+if [ "${ITK_LOG_LEVEL^^}" = "DEBUG" ]; then
+  DOCKER_MOUNT_LOGS="-v $ITK_DIR/logs:/app/logs"
+fi
+
 docker run -d --name itk-service \
   -v "$A2A_PYTHON_ROOT:/app/agents/repo" \
   -v "$ITK_DIR:/app/agents/repo/itk" \
+  $DOCKER_MOUNT_LOGS \
+  -e ITK_LOG_LEVEL="$ITK_LOG_LEVEL" \
   -p 8000:8000 \
   itk_service
 

src/a2a/server/agent_execution/agent_executor.py

@@ -23,20 +23,43 @@ async def execute(
         return once the agent's execution for this request is complete or
         yields control (e.g., enters an input-required state).
 
-        TODO: Document request lifecycle and AgentExecutor responsibilities:
-        - Should not close the event_queue.
-        - Guarantee single execution per request (no concurrent execution).
-        - Throwing exception will result in TaskState.TASK_STATE_ERROR (CHECK!)
-        - Once call is completed it should not access context or event_queue
-        - Before completing the call it SHOULD update task status to terminal or interrupted state.
-        - Explain AUTH_REQUIRED workflow.
-        - Explain INPUT_REQUIRED workflow.
-        - Explain how cancelation work (executor task will be canceled, cancel() is called, order of calls, etc)
-        - Explain if execute can wait for cancel and if cancel can wait for execute.
-        - Explain behaviour of streaming / not-immediate when execute() returns in active state.
-        - Possible workflows:
-            - Enqueue a SINGLE Message object
-            - Enqueue TaskStatusUpdateEvent (TASK_STATE_SUBMITTED or TASK_STATE_REJECTED) and continue with TaskStatusUpdateEvent / TaskArtifactUpdateEvent.
+        Request Lifecycle & AgentExecutor Responsibilities:
+        - **Concurrency**: The framework guarantees single execution per request;
+          `execute()` will not be called concurrently for the same request context.
+        - **Exception Handling**: Unhandled exceptions raised by `execute()` will be
+          caught by the framework and result in the task transitioning to
+          `TaskState.TASK_STATE_ERROR`.
+        - **Post-Completion**: Once `execute()` completes (returns or raises), the
+          executor must not access the `context` or `event_queue` anymore.
+        - **Terminal States**: Before completing the call normally, the executor
+          SHOULD publish a `TaskStatusUpdateEvent` to transition the task to a
+          terminal state (e.g., `TASK_STATE_COMPLETED`) or an interrupted state
+          (`TASK_STATE_INPUT_REQUIRED` or `TASK_STATE_AUTH_REQUIRED`).
+        - **Interrupted Workflows**:
+            - `TASK_STATE_INPUT_REQUIRED`: The executor publishes a `TaskStatusUpdateEvent` with
+              `TaskState.TASK_STATE_INPUT_REQUIRED` and returns to yield control.
+              The request will resume once user input is provided.
+            - `TASK_STATE_AUTH_REQUIRED`: There are in-bound and out-of-bound auth models.
+              In both scenarios, the agent publishes a `TaskStatusUpdateEvent` with
+              `TaskState.TASK_STATE_AUTH_REQUIRED`.
+                - In-bound: The agent should return from `execute()`. The framework will
+                  call `execute()` again once the user response is received.
+                - Out-of-bound: The agent should not return from `execute()`. It should wait
+                  for the out-of-band auth provider to complete the authentication and then
+                  continue execution.
+
+        - **Cancellation Workflow**: When a cancellation request is received, the
+          async task running `execute()` is cancelled (raising an `asyncio.CancelledError`),
+          and `cancel()` is explicitly called by the framework.
+
+        Allowed Workflows:
+        - Immediate response: Enqueue a SINGLE `Message` object.
+        - Asynchronous/Long-running: Enqueue a `Task` object, perform work, and emit
+          multiple `TaskStatusUpdateEvent` / `TaskArtifactUpdateEvent` objects over time.
+
+        Note that the framework waits with response to the send_message request with
+        `return_immediately=True` parameter until the first event (Message or Task)
+        is enqueued by AgentExecutor.
 
         Args:
             context: The request context containing the message, task ID, etc.
@@ -53,10 +76,6 @@ async def cancel(
         in the context and publish a `TaskStatusUpdateEvent` with state
         `TaskState.TASK_STATE_CANCELED` to the `event_queue`.
 
-        TODO: Document cancelation workflow.
-        - What if TaskState.TASK_STATE_CANCELED is not set by cancel() ?
-        - How it can interact with execute() ?
-
         Args:
             context: The request context containing the task ID to cancel.
             event_queue: The queue to publish the cancellation status update to.

tests/integration/test_scenarios.py

@@ -113,6 +113,22 @@ def agent_card():
     )
 
 
+def get_task_id(event):
+    if event.HasField('task'):
+        return event.task.id
+    if event.HasField('status_update'):
+        return event.status_update.task_id
+    assert False, f'Event {event} has no task_id'
+
+
+def get_task_context_id(event):
+    if event.HasField('task'):
+        return event.task.context_id
+    if event.HasField('status_update'):
+        return event.status_update.context_id
+    assert False, f'Event {event} has no context_id'
+
+
 def get_state(event):
     if event.HasField('task'):
         return event.task.status.state
@@ -1265,6 +1281,93 @@ async def cancel(
         )
 
 
+# Scenario: Auth required and in channel unblocking
+@pytest.mark.timeout(2.0)
+@pytest.mark.asyncio
+@pytest.mark.parametrize('use_legacy', [False, True], ids=['v2', 'legacy'])
+@pytest.mark.parametrize(
+    'streaming', [False, True], ids=['blocking', 'streaming']
+)
+async def test_scenario_auth_required_in_channel(use_legacy, streaming):
+    class AuthAgent(AgentExecutor):
+        async def execute(
+            self, context: RequestContext, event_queue: EventQueue
+        ):
+            message = context.message
+            if message and message.parts and message.parts[0].text == 'start':
+                await event_queue.enqueue_event(
+                    TaskStatusUpdateEvent(
+                        task_id=context.task_id,
+                        context_id=context.context_id,
+                        status=TaskStatus(
+                            state=TaskState.TASK_STATE_AUTH_REQUIRED
+                        ),
+                    )
+                )
+            elif (
+                message
+                and message.parts
+                and message.parts[0].text == 'credentials'
+            ):
+                await event_queue.enqueue_event(
+                    TaskStatusUpdateEvent(
+                        task_id=context.task_id,
+                        context_id=context.context_id,
+                        status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+                    )
+                )
+            else:
+                raise ValueError(f'Unexpected message {message}')
+
+        async def cancel(
+            self, context: RequestContext, event_queue: EventQueue
+        ):
+            pass
+
+    handler = create_handler(AuthAgent(), use_legacy)
+    client = await create_client(
+        handler, agent_card=agent_card(), streaming=streaming
+    )
+
+    msg1 = Message(
+        message_id='msg-start', role=Role.ROLE_USER, parts=[Part(text='start')]
+    )
+
+    it = client.send_message(
+        SendMessageRequest(
+            message=msg1,
+            configuration=SendMessageConfiguration(return_immediately=False),
+        )
+    )
+
+    events1 = [event async for event in it]
+    assert [get_state(event) for event in events1] == [
+        TaskState.TASK_STATE_AUTH_REQUIRED,
+    ]
+    task_id = get_task_id(events1[0])
+    context_id = get_task_context_id(events1[0])
+
+    # Now send another message with credentials
+    msg2 = Message(
+        task_id=task_id,
+        context_id=context_id,
+        message_id='msg-creds',
+        role=Role.ROLE_USER,
+        parts=[Part(text='credentials')],
+    )
+
+    it2 = client.send_message(
+        SendMessageRequest(
+            message=msg2,
+            configuration=SendMessageConfiguration(return_immediately=False),
+        )
+    )
+
+    assert [get_state(event) async for event in it2] == [
+        TaskState.TASK_STATE_COMPLETED,
+    ]
+
+
 # Scenario: Parallel subscribe attach detach
 # Migrated from: test_parallel_subscribe_attach_detach in test_handler_comparison
 @pytest.mark.timeout(5.0)