From 0eb3cca579e3f6d0db44ebdc78c0e18c26053fa9 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sun, 19 Apr 2026 01:39:34 +0100 Subject: [PATCH 1/6] test: add async task parity tests for issue #95 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds three test classes covering behaviors not exercised by the upstream TestAsyncModeUnifiedExecutor (landed via #320): - TestAsyncSessionPersistence (3 tests) — save_to_session creates user+assistant messages, explicit session_id is respected, WebSocket chat_response_ready fires. - TestAsyncCollaborationActivity (1 test) — agent-to-agent async task completes the collaboration activity end-to-end. - TestAsyncSafetyNet (1 test) — execution record is created before the background task is spawned (no silent launch failure). Closes #95 test coverage gaps. Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/test_parallel_task.py | 267 ++++++++++++++++++++++++++++++++++++ 1 file changed, 267 insertions(+) diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py index fe48ae9d4..f5e759515 100644 --- a/tests/test_parallel_task.py +++ b/tests/test_parallel_task.py @@ -879,3 +879,270 @@ def test_async_mode_activity_has_parallel_mode_flag( # doesn't serialize subscription_id, so it can't be asserted at the API # surface without a new DB-direct fixture. Covered by code inspection at # src/backend/routers/chat.py:688 where subscription_id is now passed. + + + +class TestAsyncSessionPersistence: + """Issue #95: Tests for save_to_session via TaskExecutionService delegation.""" + + @pytest.mark.slow + @pytest.mark.requires_agent + def test_async_save_to_session_creates_chat_messages( + self, + api_client: TrinityApiClient, + created_agent + ): + """Async task with save_to_session=true creates user + assistant messages in a new session.""" + response = api_client.post( + f"/api/agents/{created_agent['name']}/task", + json={ + "message": "What is 3+3? Reply with just the number.", + "async_mode": True, + "save_to_session": True, + "create_new_session": True, + }, + timeout=10.0, + ) + + if response.status_code == 503: + pytest.skip("Agent server not ready") + if response.status_code == 429: + pytest.skip("Agent at capacity") + + assert_status(response, 200) + data = response.json() + execution_id = data["execution_id"] + + # Poll until task completes + max_wait = 120 + start = time.time() + while time.time() - start < max_wait: + poll = api_client.get( + f"/api/agents/{created_agent['name']}/executions/{execution_id}" + ) + if poll.status_code == 200: + exec_data = poll.json() + if exec_data.get("status") in ["success", "failed"]: + break + time.sleep(2) + + # Verify chat sessions contain messages + sessions_resp = api_client.get( + f"/api/agents/{created_agent['name']}/chat/sessions" + ) + if sessions_resp.status_code == 200: + sessions = sessions_resp.json() + assert len(sessions) > 0, "Should have at least one chat session after save_to_session" + + @pytest.mark.slow + @pytest.mark.requires_agent + def test_async_save_to_session_with_explicit_session_id( + self, + api_client: TrinityApiClient, + created_agent + ): + """Async task with explicit chat_session_id adds messages to existing session.""" + # First, create a session by sending a sync task with save_to_session + sync_resp = api_client.post( + f"/api/agents/{created_agent['name']}/task", + json={ + "message": "What is 1+1? Reply with just the number.", + "save_to_session": True, + "create_new_session": True, + }, + timeout=120.0, + ) + + if sync_resp.status_code == 503: + pytest.skip("Agent server not ready") + if sync_resp.status_code == 429: + pytest.skip("Agent at capacity") + + if sync_resp.status_code != 200: + pytest.skip(f"Sync task failed with {sync_resp.status_code}") + + sync_data = sync_resp.json() + session_id = sync_data.get("chat_session_id") + if not session_id: + pytest.skip("Sync task did not return chat_session_id") + + # Now send async task targeting the same session + async_resp = api_client.post( + f"/api/agents/{created_agent['name']}/task", + json={ + "message": "What is 2+2? Reply with just the number.", + "async_mode": True, + "save_to_session": True, + "chat_session_id": session_id, + }, + timeout=10.0, + ) + + if async_resp.status_code == 429: + pytest.skip("Agent at capacity") + + assert_status(async_resp, 200) + execution_id = async_resp.json()["execution_id"] + + # Poll until complete + max_wait = 120 + start = time.time() + while time.time() - start < max_wait: + poll = api_client.get( + f"/api/agents/{created_agent['name']}/executions/{execution_id}" + ) + if poll.status_code == 200 and poll.json().get("status") in ["success", "failed"]: + break + time.sleep(2) + + # Session should still exist (messages were added to it) + sessions_resp = api_client.get( + f"/api/agents/{created_agent['name']}/chat/sessions" + ) + if sessions_resp.status_code == 200: + sessions = sessions_resp.json() + matching = [s for s in sessions if s.get("id") == session_id] + assert len(matching) > 0, f"Session {session_id} should still exist after async task" + + @pytest.mark.slow + @pytest.mark.requires_agent + def test_async_save_to_session_broadcasts_websocket( + self, + api_client: TrinityApiClient, + created_agent + ): + """Async task with save_to_session broadcasts chat_response_ready (verified via session existence).""" + response = api_client.post( + f"/api/agents/{created_agent['name']}/task", + json={ + "message": "What is 4+4? Reply with just the number.", + "async_mode": True, + "save_to_session": True, + "create_new_session": True, + }, + timeout=10.0, + ) + + if response.status_code == 503: + pytest.skip("Agent server not ready") + if response.status_code == 429: + pytest.skip("Agent at capacity") + + assert_status(response, 200) + execution_id = response.json()["execution_id"] + + # Poll until complete + max_wait = 120 + start = time.time() + final_status = None + while time.time() - start < max_wait: + poll = api_client.get( + f"/api/agents/{created_agent['name']}/executions/{execution_id}" + ) + if poll.status_code == 200: + final_status = poll.json().get("status") + if final_status in ["success", "failed"]: + break + time.sleep(2) + + # If task succeeded, session should exist (WebSocket broadcast happened) + if final_status == "success": + sessions_resp = api_client.get( + f"/api/agents/{created_agent['name']}/chat/sessions" + ) + assert sessions_resp.status_code == 200, "Should be able to list sessions" + + +class TestAsyncCollaborationActivity: + """Issue #95: Tests for collaboration activity completion via TaskExecutionService.""" + + @pytest.mark.slow + @pytest.mark.requires_agent + def test_async_collaboration_activity_completed( + self, + api_client: TrinityApiClient, + created_agent + ): + """Async task with X-Source-Agent header creates and completes collaboration activity.""" + response = api_client.post( + f"/api/agents/{created_agent['name']}/task", + json={ + "message": "What is 5+5? Reply with just the number.", + "async_mode": True, + }, + headers={"X-Source-Agent": "test-source-agent"}, + timeout=10.0, + ) + + if response.status_code == 503: + pytest.skip("Agent server not ready") + if response.status_code == 429: + pytest.skip("Agent at capacity") + + assert_status(response, 200) + execution_id = response.json()["execution_id"] + + # Poll until complete + max_wait = 120 + start = time.time() + while time.time() - start < max_wait: + poll = api_client.get( + f"/api/agents/{created_agent['name']}/executions/{execution_id}" + ) + if poll.status_code == 200 and poll.json().get("status") in ["success", "failed"]: + break + time.sleep(2) + + # Check activities timeline for collaboration activity + time.sleep(1) # Brief wait for activity completion + activities_resp = api_client.get( + "/api/activities/timeline", + params={"activity_types": "agent_collaboration"} + ) + if activities_resp.status_code == 200: + activities = activities_resp.json() + collab_activities = [ + a for a in activities.get("activities", []) + if a.get("details", {}).get("execution_id") == execution_id + ] + # Should have a collaboration activity for this execution + assert len(collab_activities) > 0, \ + f"Should have collaboration activity for execution {execution_id}" + + +class TestAsyncSafetyNet: + """Issue #95: Tests for safety net error handling in background task.""" + + def test_async_mode_execution_record_exists_before_background( + self, + api_client: TrinityApiClient, + created_agent + ): + """Async task creates execution record immediately (before background task runs).""" + response = api_client.post( + f"/api/agents/{created_agent['name']}/task", + json={ + "message": "Quick test", + "async_mode": True, + }, + timeout=10.0, + ) + + if response.status_code == 503: + pytest.skip("Agent server not ready") + if response.status_code == 429: + pytest.skip("Agent at capacity") + + assert_status(response, 200) + execution_id = response.json()["execution_id"] + + # Execution record should exist immediately + poll = api_client.get( + f"/api/agents/{created_agent['name']}/executions/{execution_id}" + ) + assert_status(poll, 200) + exec_data = poll.json() + assert exec_data["id"] == execution_id + # Status should be pending/running/dispatched (not failed/cancelled yet) + assert exec_data["status"] not in ["cancelled"], \ + f"Execution should not be cancelled immediately, got {exec_data['status']}" From fb631d5e2b79bc96cdb9f68e0b55000704ce4b45 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sun, 19 Apr 2026 01:39:39 +0100 Subject: [PATCH 2/6] test: extract poll_execution_until_done helper Shared helper in tests/testing_utils/assertions.py to remove duplicated polling loops across the async task tests added in the previous commit. Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/test_parallel_task.py | 55 ++++--------------------------- tests/testing_utils/assertions.py | 23 +++++++++++++ 2 files changed, 30 insertions(+), 48 deletions(-) diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py index f5e759515..2dc15a97d 100644 --- a/tests/test_parallel_task.py +++ b/tests/test_parallel_task.py @@ -15,6 +15,7 @@ assert_status_in, assert_json_response, assert_has_fields, + poll_execution_until_done, ) @@ -910,21 +911,9 @@ def test_async_save_to_session_creates_chat_messages( pytest.skip("Agent at capacity") assert_status(response, 200) - data = response.json() - execution_id = data["execution_id"] + execution_id = response.json()["execution_id"] - # Poll until task completes - max_wait = 120 - start = time.time() - while time.time() - start < max_wait: - poll = api_client.get( - f"/api/agents/{created_agent['name']}/executions/{execution_id}" - ) - if poll.status_code == 200: - exec_data = poll.json() - if exec_data.get("status") in ["success", "failed"]: - break - time.sleep(2) + poll_execution_until_done(api_client, created_agent['name'], execution_id) # Verify chat sessions contain messages sessions_resp = api_client.get( @@ -984,16 +973,7 @@ def test_async_save_to_session_with_explicit_session_id( assert_status(async_resp, 200) execution_id = async_resp.json()["execution_id"] - # Poll until complete - max_wait = 120 - start = time.time() - while time.time() - start < max_wait: - poll = api_client.get( - f"/api/agents/{created_agent['name']}/executions/{execution_id}" - ) - if poll.status_code == 200 and poll.json().get("status") in ["success", "failed"]: - break - time.sleep(2) + poll_execution_until_done(api_client, created_agent['name'], execution_id) # Session should still exist (messages were added to it) sessions_resp = api_client.get( @@ -1031,22 +1011,10 @@ def test_async_save_to_session_broadcasts_websocket( assert_status(response, 200) execution_id = response.json()["execution_id"] - # Poll until complete - max_wait = 120 - start = time.time() - final_status = None - while time.time() - start < max_wait: - poll = api_client.get( - f"/api/agents/{created_agent['name']}/executions/{execution_id}" - ) - if poll.status_code == 200: - final_status = poll.json().get("status") - if final_status in ["success", "failed"]: - break - time.sleep(2) + result = poll_execution_until_done(api_client, created_agent['name'], execution_id) # If task succeeded, session should exist (WebSocket broadcast happened) - if final_status == "success": + if result and result.get("status") == "success": sessions_resp = api_client.get( f"/api/agents/{created_agent['name']}/chat/sessions" ) @@ -1082,16 +1050,7 @@ def test_async_collaboration_activity_completed( assert_status(response, 200) execution_id = response.json()["execution_id"] - # Poll until complete - max_wait = 120 - start = time.time() - while time.time() - start < max_wait: - poll = api_client.get( - f"/api/agents/{created_agent['name']}/executions/{execution_id}" - ) - if poll.status_code == 200 and poll.json().get("status") in ["success", "failed"]: - break - time.sleep(2) + poll_execution_until_done(api_client, created_agent['name'], execution_id) # Check activities timeline for collaboration activity time.sleep(1) # Brief wait for activity completion diff --git a/tests/testing_utils/assertions.py b/tests/testing_utils/assertions.py index 5a4363f90..be6a68822 100644 --- a/tests/testing_utils/assertions.py +++ b/tests/testing_utils/assertions.py @@ -133,3 +133,26 @@ def assert_credential_fields(cred: Dict[str, Any]): CRED-002 uses file-based credentials. Kept for backward compatibility. """ pass + + +def poll_execution_until_done( + api_client, + agent_name: str, + execution_id: str, + max_wait: int = 120, + interval: float = 2.0, +) -> Optional[Dict[str, Any]]: + """Poll execution endpoint until terminal status. Returns final execution data or None on timeout.""" + import time + + start = time.time() + while time.time() - start < max_wait: + poll = api_client.get( + f"/api/agents/{agent_name}/executions/{execution_id}" + ) + if poll.status_code == 200: + data = poll.json() + if data.get("status") in ["success", "failed"]: + return data + time.sleep(interval) + return None From d2ec150cd409e146d987e167a17d64f68ae31c0f Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 18:15:56 +0100 Subject: [PATCH 3/6] test: harden async parity tests per review feedback - Replace fixed time.sleep(1) in TestAsyncCollaborationActivity with a bounded retry loop on /api/activities/timeline so the assertion is not fragile when the activity write lags the execution finish. - Document poll_execution_until_done as @pytest.mark.slow-only and warn against use from non-slow contexts (busy-poll with time.sleep). - Mark TestAsyncSafetyNet test with @pytest.mark.requires_agent so it is gated correctly by the integration runner. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_parallel_task.py | 38 +++++++++++++++++++------------ tests/testing_utils/assertions.py | 9 +++++++- 2 files changed, 31 insertions(+), 16 deletions(-) diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py index 2dc15a97d..0f6f65ad3 100644 --- a/tests/test_parallel_task.py +++ b/tests/test_parallel_task.py @@ -1052,26 +1052,34 @@ def test_async_collaboration_activity_completed( poll_execution_until_done(api_client, created_agent['name'], execution_id) - # Check activities timeline for collaboration activity - time.sleep(1) # Brief wait for activity completion - activities_resp = api_client.get( - "/api/activities/timeline", - params={"activity_types": "agent_collaboration"} - ) - if activities_resp.status_code == 200: - activities = activities_resp.json() - collab_activities = [ - a for a in activities.get("activities", []) - if a.get("details", {}).get("execution_id") == execution_id - ] - # Should have a collaboration activity for this execution - assert len(collab_activities) > 0, \ - f"Should have collaboration activity for execution {execution_id}" + # The collaboration activity is written asynchronously after the + # execution finishes, so poll the timeline briefly instead of relying + # on a single fixed sleep. + deadline = time.time() + 10.0 + collab_activities: list = [] + while time.time() < deadline: + activities_resp = api_client.get( + "/api/activities/timeline", + params={"activity_types": "agent_collaboration"} + ) + if activities_resp.status_code == 200: + activities = activities_resp.json().get("activities", []) + collab_activities = [ + a for a in activities + if a.get("details", {}).get("execution_id") == execution_id + ] + if collab_activities: + break + time.sleep(0.5) + + assert len(collab_activities) > 0, \ + f"Should have collaboration activity for execution {execution_id}" class TestAsyncSafetyNet: """Issue #95: Tests for safety net error handling in background task.""" + @pytest.mark.requires_agent def test_async_mode_execution_record_exists_before_background( self, api_client: TrinityApiClient, diff --git a/tests/testing_utils/assertions.py b/tests/testing_utils/assertions.py index be6a68822..39caa52ca 100644 --- a/tests/testing_utils/assertions.py +++ b/tests/testing_utils/assertions.py @@ -142,7 +142,14 @@ def poll_execution_until_done( max_wait: int = 120, interval: float = 2.0, ) -> Optional[Dict[str, Any]]: - """Poll execution endpoint until terminal status. Returns final execution data or None on timeout.""" + """Poll the execution endpoint until terminal status. + + Returns the final execution payload, or ``None`` on timeout. + + Intended for ``@pytest.mark.slow`` integration tests against a running + backend — blocks the calling thread for up to ``max_wait`` seconds via + ``time.sleep(interval)``. Do not call from non-slow tests. + """ import time start = time.time() From 9bb342605890d7498b6e3af339c15431e6818f88 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 18:26:27 +0100 Subject: [PATCH 4/6] test: tighten async parity tests per /review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - poll_execution_until_done: include CANCELLED and SKIPPED in the terminal set per TaskExecutionStatus (models.py). Previously polled to timeout whenever an execution was cancelled or skipped. - TestAsyncSessionPersistence: replace silent `if status_code == 200` guards with assert_status(...) so a backend regression on the sessions endpoint fails the test instead of passing without assertions. - TestAsyncSessionPersistence (websocket test): require the helper to return a terminal status; skip explicitly when the execution did not succeed (broadcast only fires on success). - TestAsyncSafetyNet: tighten the "execution exists immediately" status check to an explicit allow-list (queued/running/success) — the prior `not in ["cancelled"]` permitted failed/skipped too. - Drop one extra blank line between top-level class definitions (PEP 8). Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_parallel_task.py | 47 ++++++++++++++++++------------- tests/testing_utils/assertions.py | 5 +++- 2 files changed, 32 insertions(+), 20 deletions(-) diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py index 0f6f65ad3..b0ed078be 100644 --- a/tests/test_parallel_task.py +++ b/tests/test_parallel_task.py @@ -882,7 +882,6 @@ def test_async_mode_activity_has_parallel_mode_flag( # src/backend/routers/chat.py:688 where subscription_id is now passed. - class TestAsyncSessionPersistence: """Issue #95: Tests for save_to_session via TaskExecutionService delegation.""" @@ -915,13 +914,14 @@ def test_async_save_to_session_creates_chat_messages( poll_execution_until_done(api_client, created_agent['name'], execution_id) - # Verify chat sessions contain messages + # Verify chat sessions contain messages — fail loudly on non-200 so a + # backend regression doesn't slip through silently. sessions_resp = api_client.get( f"/api/agents/{created_agent['name']}/chat/sessions" ) - if sessions_resp.status_code == 200: - sessions = sessions_resp.json() - assert len(sessions) > 0, "Should have at least one chat session after save_to_session" + assert_status(sessions_resp, 200) + sessions = sessions_resp.json() + assert len(sessions) > 0, "Should have at least one chat session after save_to_session" @pytest.mark.slow @pytest.mark.requires_agent @@ -975,14 +975,14 @@ def test_async_save_to_session_with_explicit_session_id( poll_execution_until_done(api_client, created_agent['name'], execution_id) - # Session should still exist (messages were added to it) + # Session should still exist (messages were added to it). sessions_resp = api_client.get( f"/api/agents/{created_agent['name']}/chat/sessions" ) - if sessions_resp.status_code == 200: - sessions = sessions_resp.json() - matching = [s for s in sessions if s.get("id") == session_id] - assert len(matching) > 0, f"Session {session_id} should still exist after async task" + assert_status(sessions_resp, 200) + sessions = sessions_resp.json() + matching = [s for s in sessions if s.get("id") == session_id] + assert len(matching) > 0, f"Session {session_id} should still exist after async task" @pytest.mark.slow @pytest.mark.requires_agent @@ -1013,12 +1013,18 @@ def test_async_save_to_session_broadcasts_websocket( result = poll_execution_until_done(api_client, created_agent['name'], execution_id) - # If task succeeded, session should exist (WebSocket broadcast happened) - if result and result.get("status") == "success": - sessions_resp = api_client.get( - f"/api/agents/{created_agent['name']}/chat/sessions" - ) - assert sessions_resp.status_code == 200, "Should be able to list sessions" + assert result is not None, \ + f"Execution {execution_id} did not reach a terminal status before timeout" + if result.get("status") != "success": + pytest.skip(f"Execution finished with status={result.get('status')}; " + "WebSocket broadcast only fires on success") + + # Session should be listable — proxy for the chat_response_ready broadcast + # having fired (the broadcast persists the session before notifying). + sessions_resp = api_client.get( + f"/api/agents/{created_agent['name']}/chat/sessions" + ) + assert_status(sessions_resp, 200) class TestAsyncCollaborationActivity: @@ -1110,6 +1116,9 @@ def test_async_mode_execution_record_exists_before_background( assert_status(poll, 200) exec_data = poll.json() assert exec_data["id"] == execution_id - # Status should be pending/running/dispatched (not failed/cancelled yet) - assert exec_data["status"] not in ["cancelled"], \ - f"Execution should not be cancelled immediately, got {exec_data['status']}" + # Realistic states for a fresh async submission: queued (backlog), running + # (slot acquired, executing), or success (very fast agent finished before + # the GET returned). Anything else (failed/cancelled/skipped) means the + # safety net leaked — record was created but the task died on launch. + assert exec_data["status"] in {"queued", "running", "success"}, \ + f"Execution should be in an early state, got {exec_data['status']}" diff --git a/tests/testing_utils/assertions.py b/tests/testing_utils/assertions.py index 39caa52ca..a7d635198 100644 --- a/tests/testing_utils/assertions.py +++ b/tests/testing_utils/assertions.py @@ -152,6 +152,9 @@ def poll_execution_until_done( """ import time + # Terminal statuses per src/backend/models.py::TaskExecutionStatus — + # SUCCESS, FAILED, CANCELLED, SKIPPED. Polling past these wastes test time. + terminal = {"success", "failed", "cancelled", "skipped"} start = time.time() while time.time() - start < max_wait: poll = api_client.get( @@ -159,7 +162,7 @@ def poll_execution_until_done( ) if poll.status_code == 200: data = poll.json() - if data.get("status") in ["success", "failed"]: + if data.get("status") in terminal: return data time.sleep(interval) return None From d111205e78d3a29f3bd26501abb5cc756298b1e0 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 20:50:45 +0100 Subject: [PATCH 5/6] test: extract sessions list from /chat/sessions wrapped response The /api/agents/{name}/chat/sessions endpoint returns {agent_name, session_count, sessions: [...]} (a wrapped object), not a bare list. Two of the three TestAsyncSessionPersistence tests treated it as a bare list: - test_async_save_to_session_creates_chat_messages was a false positive: iterating the dict yielded 3 keys, so len(sessions) > 0 was always true regardless of whether any session existed. - test_async_save_to_session_with_explicit_session_id raised AttributeError when calling .get('id') on a string (dict key) instead of a session dict. Both now extract sessions_resp.json()['sessions'] before iterating, so the assertions actually verify session presence. --- tests/test_parallel_task.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py index b0ed078be..ebf9d9fdc 100644 --- a/tests/test_parallel_task.py +++ b/tests/test_parallel_task.py @@ -920,7 +920,8 @@ def test_async_save_to_session_creates_chat_messages( f"/api/agents/{created_agent['name']}/chat/sessions" ) assert_status(sessions_resp, 200) - sessions = sessions_resp.json() + # Endpoint returns {agent_name, session_count, sessions: [...]}. + sessions = sessions_resp.json()["sessions"] assert len(sessions) > 0, "Should have at least one chat session after save_to_session" @pytest.mark.slow @@ -980,7 +981,8 @@ def test_async_save_to_session_with_explicit_session_id( f"/api/agents/{created_agent['name']}/chat/sessions" ) assert_status(sessions_resp, 200) - sessions = sessions_resp.json() + # Endpoint returns {agent_name, session_count, sessions: [...]}. + sessions = sessions_resp.json()["sessions"] matching = [s for s in sessions if s.get("id") == session_id] assert len(matching) > 0, f"Session {session_id} should still exist after async task" From 91bddd7b87a406136ed7cc5580e169305e8d78b4 Mon Sep 17 00:00:00 2001 From: Andrii Pasternak Date: Sat, 25 Apr 2026 20:50:58 +0100 Subject: [PATCH 6/6] test: pick a real, distinct agent as X-Source-Agent in collab test The original TestAsyncCollaborationActivity test used the literal string 'test-source-agent' as the X-Source-Agent header value. That fails against a real backend for two compounding reasons: 1. /api/activities/timeline filters activities by accessible-agent ownership (routers/activities.py:47-50). Activities owned by a non-existent source agent are silently hidden, so the test could never observe the collaboration activity it created. 2. Even if the source name resolved, using the same name as the target would trigger the SELF-EXEC-001 short-circuit in chat.py:790 that classifies the call as activity_type=self_task instead of agent_collaboration, so the agent_collaboration filter would still match nothing. The test now picks the first real agent from /api/agents that is not the target, and pytest.skips when no second agent is available. The production constraints are documented inline so this doesn't get re-broken. --- tests/test_parallel_task.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py index ebf9d9fdc..4b3ad9016 100644 --- a/tests/test_parallel_task.py +++ b/tests/test_parallel_task.py @@ -1040,13 +1040,30 @@ def test_async_collaboration_activity_completed( created_agent ): """Async task with X-Source-Agent header creates and completes collaboration activity.""" + # The source agent must satisfy two production constraints: + # 1. It must be a real, accessible agent — /api/activities/timeline + # filters activities by `agent_name ∈ accessible_agents` + # (routers/activities.py:47-50), so non-existent sources are hidden. + # 2. It must NOT equal the target — chat.py:790 classifies + # `source == target` as SELF_TASK (feature SELF-EXEC-001), which + # writes a different activity_type and skips agent_collaboration. + agents_resp = api_client.get("/api/agents") + assert_status(agents_resp, 200) + other_agents = [ + a['name'] for a in agents_resp.json() + if a['name'] != created_agent['name'] + ] + if not other_agents: + pytest.skip("Need a second accessible agent to act as collaboration source") + source_agent_name = other_agents[0] + response = api_client.post( f"/api/agents/{created_agent['name']}/task", json={ "message": "What is 5+5? Reply with just the number.", "async_mode": True, }, - headers={"X-Source-Agent": "test-source-agent"}, + headers={"X-Source-Agent": source_agent_name}, timeout=10.0, )