diff --git a/tests/test_parallel_task.py b/tests/test_parallel_task.py
index fe48ae9d4..4b3ad9016 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,
 )
 
 
@@ -879,3 +880,264 @@ 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)
+        execution_id = response.json()["execution_id"]
+
+        poll_execution_until_done(api_client, created_agent['name'], execution_id)
+
+        # 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"
+        )
+        assert_status(sessions_resp, 200)
+        # 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
+    @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_execution_until_done(api_client, created_agent['name'], execution_id)
+
+        # Session should still exist (messages were added to it).
+        sessions_resp = api_client.get(
+            f"/api/agents/{created_agent['name']}/chat/sessions"
+        )
+        assert_status(sessions_resp, 200)
+        # 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"
+
+    @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"]
+
+        result = poll_execution_until_done(api_client, created_agent['name'], execution_id)
+
+        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:
+    """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."""
+        # 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": source_agent_name},
+            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_execution_until_done(api_client, created_agent['name'], 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,
+        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
+        # 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 5a4363f90..a7d635198 100644
--- a/tests/testing_utils/assertions.py
+++ b/tests/testing_utils/assertions.py
@@ -133,3 +133,36 @@ 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 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
+
+    # 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(
+            f"/api/agents/{agent_name}/executions/{execution_id}"
+        )
+        if poll.status_code == 200:
+            data = poll.json()
+            if data.get("status") in terminal:
+                return data
+        time.sleep(interval)
+    return None