fix(agents): drive inline agent sessions via run_agent_session, not the phantom create_agent_session

ClaudeSDKClient (claude_agent_sdk) has no `create_agent_session` method, so
every inline call site raised
`'ClaudeSDKClient' object has no attribute 'create_agent_session'` at runtime.
The shared generator boilerplate (agents/_generator_base) was fixed in #339;
this covers the remaining inline call sites it called out as a follow-up:

- documentation_generator.run_documentation_generator_session
- migration_assistant.run_migration_assistant
- test_generator.run_test_generator_session (writer + coverage-improvement)
- agent_subprocess.run_agent_session (isolated-subprocess wrapper)

Each now drives the session through run_agent_session() (agents/session.py)
inside `async with client` — the same API the planner/coder/qa phases use —
and treats a returned status == "error" as failure. The agent role prompt is
delivered as the leading message since the SDK client only carries the generic
base system prompt. agent_subprocess imports the helper under an alias so it
does not shadow its own run_agent_session() wrapper.

Add tests/test_inline_generator_sessions.py (mirrors
tests/test_fixture_generator_session.py: a fake async-context-manager client
with no create_agent_session attribute, agents.session.run_agent_session
patched as an AsyncMock, asserting it is awaited and the client is entered).
Update the existing migration/test-generation suites that mocked the phantom
method to drive the shared session helper instead.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Oleg Miagkov <mrobenner@gmail.com>

Author: OBenner

apps/backend/agents/agent_subprocess.py

@@ -149,7 +149,14 @@ async def run_agent_session(
     Returns:
         Dictionary with session results
     """
+    # ClaudeSDKClient has no `create_agent_session` method. Drive the session
+    # with the shared run_agent_session() helper (agents/session.py) inside
+    # `async with client` — the same API the planner/coder/qa phases use.
+    # Imported under an alias so it does not shadow this module's own
+    # run_agent_session() wrapper (defined above).
+    from agents.session import run_agent_session as run_sdk_session
     from core.client import create_client
+    from task_logger import LogPhase
 
     _debug(f"Starting {agent_type} agent session in subprocess")
     _debug_verbose(f"Project: {project_dir}")
@@ -168,23 +175,44 @@ async def run_agent_session(
 
         _debug(f"Client created, starting agent session: {session_name}")
 
-        # Run the agent session
-        session_kwargs = {
-            "name": session_name,
-            "starting_message": starting_message,
-        }
+        # Drive the session through the shared helper. The role prompt
+        # (system_prompt) leads the message when provided; the SDK client only
+        # carries the generic base system prompt.
+        session_message = (
+            f"{system_prompt}\n\n{starting_message}"
+            if system_prompt
+            else starting_message
+        )
+        phase = (
+            LogPhase.PLANNING
+            if agent_type == "planner"
+            else LogPhase.VALIDATION
+            if agent_type in ("qa_reviewer", "qa_fixer")
+            else LogPhase.CODING
+        )
 
-        if system_prompt:
-            session_kwargs["system_prompt"] = system_prompt
+        async with client:
+            status, response_text, _usage, _decisions = await run_sdk_session(
+                client=client,
+                message=session_message,
+                spec_dir=spec_dir,
+                phase=phase,
+            )
 
-        response = await client.create_agent_session(**session_kwargs)
+        if status == "error":
+            _debug_error(f"Agent session failed: {response_text}")
+            return {
+                "success": False,
+                "output": None,
+                "error": response_text,
+            }
 
         _debug_success("Agent session completed successfully")
 
         return {
             "success": True,
             "output": {
-                "response": str(response) if response else None,
+                "response": response_text or None,
                 "agent_type": agent_type,
                 "session_name": session_name,
             },

apps/backend/agents/documentation_generator.py

@@ -146,9 +146,9 @@ async def run_documentation_generator_session(
             f"{len(analysis_results.get('classes', []))} classes for documentation",
         )
 
-    # Validate the documentation generator prompt exists
+    # Load the documentation generator prompt (delivered as the leading message)
     try:
-        get_agent_prompt("documentation_generator")
+        prompt = get_agent_prompt("documentation_generator")
     except Exception as e:
         error_msg = f"Failed to load documentation_generator prompt: {e}"
         logger.error(error_msg)
@@ -202,14 +202,34 @@ async def run_documentation_generator_session(
             task_logger.log_entry(LogEntryType.ERROR, error_msg)
         return {"generated_files": [], "success": False, "error": error_msg}
 
-    # Run the agent session
+    # Run the agent session.
+    #
+    # ClaudeSDKClient has no `create_agent_session` method — agent turns are
+    # driven by run_agent_session() (agents/session.py) inside `async with
+    # client`, the same session API the planner/coder/qa phases use. The agent
+    # role prompt leads the message since the SDK client only carries the
+    # generic base system prompt.
+    from .session import run_agent_session
+
     try:
         print_status("Starting documentation generation session...", "progress")
 
-        await client.create_agent_session(
-            name="documentation-generator-session",
-            starting_message=starting_message,
-        )
+        session_message = f"{prompt}\n\n{starting_message}"
+        async with client:
+            status, response_text, _usage, _decisions = await run_agent_session(
+                client=client,
+                message=session_message,
+                spec_dir=spec_dir,
+                verbose=verbose,
+                phase=LogPhase.CODING,
+            )
+
+        if status == "error":
+            error_msg = f"Documentation generation session failed: {response_text}"
+            logger.error(error_msg)
+            if task_logger:
+                task_logger.log_entry(LogEntryType.ERROR, error_msg)
+            return {"generated_files": [], "success": False, "error": error_msg}
 
         if task_logger:
             task_logger.log_entry(

apps/backend/agents/migration_assistant.py

@@ -272,13 +272,35 @@ async def run_migration_assistant(
     print_status("Running migration assistant...", "progress")
     print()
 
+    # ClaudeSDKClient has no `create_agent_session` method — drive the session
+    # with run_agent_session() (agents/session.py) inside `async with client`,
+    # the same session API the planner/coder/qa phases use. The agent role
+    # prompt leads the message since the SDK client only carries the generic
+    # base system prompt.
+    from .session import run_agent_session
+
     try:
+        session_message = f"{_prompt}\n\n{starting_message}"
         async with client:
-            await client.create_agent_session(
-                name="migration-assistant-session",
-                starting_message=starting_message,
+            status, response_text, _usage, _decisions = await run_agent_session(
+                client=client,
+                message=session_message,
+                spec_dir=spec_dir,
+                verbose=verbose,
+                phase=LogPhase.CODING,
             )
 
+        if status == "error":
+            error_msg = f"Migration session failed: {response_text}"
+            logger.error(error_msg)
+            if task_logger:
+                task_logger.log(error_msg, LogEntryType.ERROR)
+            return {
+                "checkpoints_created": 0,
+                "success": False,
+                "error": error_msg,
+            }
+
         # Log completion
         if task_logger:
             task_logger.end_phase(

apps/backend/agents/test_generator.py

@@ -1595,14 +1595,32 @@ async def run_test_generator_session(
             task_logger.log_error(error_msg)
         return {"generated_files": [], "success": False, "error": error_msg}
 
-    # Run the agent session
+    # Run the agent session.
+    #
+    # ClaudeSDKClient has no `create_agent_session` method — agent turns are
+    # driven by run_agent_session() (agents/session.py) inside `async with
+    # client`. The agent role prompt leads the message since the SDK client
+    # only carries the generic base system prompt.
+    from .session import run_agent_session
+
     print_status("Running Test Generator Agent...", "progress")
     try:
-        response = await client.create_agent_session(
-            name="test-generator-session",
-            starting_message=starting_message,
-            system_prompt=prompt,
-        )
+        session_message = f"{prompt}\n\n{starting_message}"
+        async with client:
+            status, response, _usage, _decisions = await run_agent_session(
+                client=client,
+                message=session_message,
+                spec_dir=spec_dir,
+                verbose=verbose,
+                phase=LogPhase.CODING,
+            )
+
+        if status == "error":
+            error_msg = f"Test Generator Agent session failed: {response}"
+            logger.error(error_msg)
+            if task_logger:
+                task_logger.log_error(error_msg)
+            return {"generated_files": [], "success": False, "error": error_msg}
 
         if verbose:
             logger.info(f"Test Generator Agent response: {response}")
@@ -1714,16 +1732,28 @@ async def run_test_generator_session(
 Generate additional test cases to improve coverage to at least {min_threshold:.0f}%.
 """
 
-            # Run improvement iteration
+            # Run improvement iteration. A fresh connected session (the prior
+            # `async with client` above already disconnected) — the
+            # improvement_message is self-contained and the agent re-reads the
+            # test files it wrote earlier from disk.
             try:
                 print_status(
                     "Running Test Generator Agent improvement iteration...", "progress"
                 )
-                response = await client.create_agent_session(
-                    name="test-generator-improvement",
-                    starting_message=improvement_message,
-                    system_prompt=prompt,
-                )
+                improvement_session_message = f"{prompt}\n\n{improvement_message}"
+                async with client:
+                    status, response, _usage, _decisions = await run_agent_session(
+                        client=client,
+                        message=improvement_session_message,
+                        spec_dir=spec_dir,
+                        verbose=verbose,
+                        phase=LogPhase.CODING,
+                    )
+
+                if status == "error":
+                    # Non-fatal: keep the tests already generated above.
+                    logger.warning(f"Test improvement iteration failed: {response}")
+                    print_status(f"Improvement iteration failed: {response}", "warning")
 
                 if verbose:
                     logger.info(f"Test Generator improvement response: {response}")

tests/e2e/test_qa_test_generation_flow.py

@@ -329,17 +329,15 @@ async def test_e2e_test_generation_flow(
     # Import after fixtures are set up
     from qa.loop import run_qa_validation_loop
 
-    # Mock the client creation and QA agent sessions to avoid actual API calls
+    # Mock the client creation and QA agent sessions to avoid actual API calls.
+    # The client is only ever used as an async context manager here — the test
+    # generation and QA sessions are mocked at the qa.loop boundary
+    # (run_test_generator_session / run_qa_agent_session below), so no real
+    # session API is invoked on it.
     mock_client = AsyncMock()
     mock_client.__aenter__ = AsyncMock(return_value=mock_client)
     mock_client.__aexit__ = AsyncMock(return_value=None)
 
-    # Mock create_agent_session to return a valid response
-    async def mock_agent_session(*args, **kwargs):
-        return {"test_files": ["test_utils.py", "test_auth.py"]}
-
-    mock_client.create_agent_session = mock_agent_session
-
     with contextlib.ExitStack() as stack:
         stack.enter_context(patch("qa.loop.create_client", return_value=mock_client))
         stack.enter_context(

tests/integration/test_migration_workflow.py

@@ -15,7 +15,7 @@
 import subprocess
 import sys
 from pathlib import Path
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -353,6 +353,21 @@ def test_plan_includes_rollback_strategy(self, sample_project_files):
 class TestMigrationWorkflowEndToEnd:
     """End-to-end tests for complete migration workflow."""
 
+    @pytest.fixture(autouse=True)
+    def mock_session_runner(self):
+        """Patch the shared session driver used by run_migration_assistant.
+
+        ``ClaudeSDKClient`` has no ``create_agent_session`` method; the inline
+        migration session is driven by ``agents.session.run_agent_session``
+        inside ``async with client``. Patch it to a success 4-tuple; tests that
+        need a failure reconfigure ``self._mock_run`` (e.g. ``side_effect``).
+        """
+        self._mock_run = AsyncMock(
+            return_value=("complete", "migration done", None, MagicMock())
+        )
+        with patch("agents.session.run_agent_session", new=self._mock_run):
+            yield self._mock_run
+
     @pytest.mark.asyncio
     @patch("agents.migration_assistant.get_phase_thinking_budget")
     @patch("agents.migration_assistant.get_phase_model")
@@ -377,7 +392,6 @@ async def test_migration_assistant_initialization_with_context(
         mock_client = AsyncMock()
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=None)
-        mock_client.create_agent_session = AsyncMock(return_value={"status": "ok"})
         mock_create_client.return_value = mock_client
 
         migration_context = {
@@ -400,10 +414,12 @@ async def test_migration_assistant_initialization_with_context(
         assert call_kwargs["project_dir"] == project_dir
         assert call_kwargs["spec_dir"] == spec_dir
 
-        # Verify session was created
-        mock_client.create_agent_session.assert_called_once()
-        session_call = mock_client.create_agent_session.call_args
-        assert "migration-assistant-session" in str(session_call)
+        # Verify the session was driven through the shared helper, with the
+        # migration context carried in the message.
+        self._mock_run.assert_awaited_once()
+        message = self._mock_run.await_args.kwargs["message"]
+        assert "React 16" in message
+        assert "React 18" in message
 
     @pytest.mark.asyncio
     @patch("agents.migration_assistant.get_phase_thinking_budget")
@@ -429,7 +445,6 @@ async def test_migration_creates_checkpoints(
         mock_client = AsyncMock()
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=None)
-        mock_client.create_agent_session = AsyncMock(return_value={"status": "ok"})
         mock_create_client.return_value = mock_client
 
         # Pre-create checkpoint structure to simulate agent creating it
@@ -479,14 +494,12 @@ async def test_migration_handles_failures(
         mock_get_phase_model.return_value = "claude-sonnet-4-5-20250929"
         mock_get_phase_thinking_budget.return_value = 4096
 
-        # Mock client that raises an exception
+        # Mock client; the shared session driver raises mid-session.
         mock_client = AsyncMock()
         mock_client.__aenter__ = AsyncMock(return_value=mock_client)
         mock_client.__aexit__ = AsyncMock(return_value=None)
-        mock_client.create_agent_session = AsyncMock(
-            side_effect=Exception("Simulated failure")
-        )
         mock_create_client.return_value = mock_client
+        self._mock_run.side_effect = Exception("Simulated failure")
 
         result = await run_migration_assistant(
             project_dir=project_dir, spec_dir=spec_dir