test_claude_sdk_side_question_isolated.py
Rama main · réplica de lectura del repositorio original.
"""Native side-question leases; no server import, process, or provider I/O."""
from __future__ import annotations
import asyncio
import unittest
from types import SimpleNamespace
from unittest.mock import patch
from claude_sdk_client import (
ClaudeSDKConfigurationConflict,
ClaudeSDKGenerationChanged,
ClaudeSDKSupervisorManager,
ClaudeSDKUnavailable,
)
from side_questions import SideQuestionError
from tests.test_claude_sdk_client import FakeClaudeClient, FakeFactory
class ClaudeSDKSideQuestionTests(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
self.factory = FakeFactory()
self.manager = ClaudeSDKSupervisorManager(
client_factory=self.factory, idle_ttl_seconds=None,
disconnect_timeout_seconds=0.1,
)
self.options = {"resume": "existing-provider-session"}
self.tasks: list[asyncio.Task] = []
async def asyncTearDown(self) -> None:
for task in self.tasks:
if not task.done():
task.cancel()
await asyncio.gather(*self.tasks, return_exceptions=True)
await self.manager.close_all()
def ask(self, **kwargs) -> asyncio.Task:
task = asyncio.create_task(self.manager.ask_side_question(
"chat", "Why?", options=self.options,
configuration_key="profile", **kwargs,
))
self.tasks.append(task)
return task
async def start_main(self):
handle = await asyncio.wait_for(self.manager.start_run(
"chat", "Main task", run_id="main", options=self.options,
configuration_key="profile",
), 1)
await asyncio.wait_for(handle.wait_acknowledged(), 1)
return handle
async def test_cold_resume_uses_exact_client_without_main_turn(self) -> None:
history = [{"question": "Before?", "response": "Previous answer"}]
answer = {"answer": "Native context", "context_note": "Native"}
async def native(client, question, **kwargs):
self.assertIs(client, self.factory.clients[0])
self.assertTrue(client.connected)
self.assertEqual(client.options["resume"], "existing-provider-session")
self.assertEqual(question, "Why?")
self.assertEqual(kwargs, {"history": history, "timeout_seconds": 4})
self.assertEqual(self.manager._pins, {"chat": 1})
return answer
with patch("claude_side_question.ask_native_side_question", native):
self.assertEqual(await self.ask(history=history, timeout_seconds=4,
expected_provider_id="existing-provider-session"), answer)
client = self.factory.clients[0]
self.assertEqual([call[0] for call in client.calls], ["connect", "receive_messages"])
self.assertFalse(self.manager._supervisors["chat"].is_active)
self.assertEqual(self.manager._pins, {})
async def test_pending_side_answer_does_not_block_main_start_messages_or_stop(self) -> None:
started = asyncio.Event()
release = asyncio.Event()
async def native(*args, **kwargs):
started.set()
await release.wait()
return {"answer": "Side answer"}
with patch("claude_side_question.ask_native_side_question", native):
side = self.ask()
await asyncio.wait_for(started.wait(), 1)
main = await self.start_main()
client = self.factory.clients[0]
message = {"type": "assistant", "content": [{"type": "text", "text": "Main progress"}]}
await client.emit(message)
self.assertEqual(await asyncio.wait_for(main.__anext__(), 1), message)
self.assertTrue(await asyncio.wait_for(main.interrupt(), 1))
self.assertFalse(side.done())
self.assertFalse(client.disconnected)
release.set()
self.assertEqual(await side, {"answer": "Side answer"})
self.assertEqual([call[0] for call in client.calls].count("query"), 1)
self.assertEqual([call[0] for call in client.calls].count("interrupt"), 1)
self.assertEqual(self.manager._pins, {})
async def test_cancel_and_provider_error_leave_active_main_unchanged(self) -> None:
main = await self.start_main()
client = self.factory.clients[0]
original_calls = list(client.calls)
started = asyncio.Event()
async def pending(*args, **kwargs):
started.set()
await asyncio.Future()
with patch("claude_side_question.ask_native_side_question", pending):
side = self.ask()
await asyncio.wait_for(started.wait(), 1)
side.cancel()
with self.assertRaises(asyncio.CancelledError):
await side
self.assertEqual(self.manager._pins, {})
self.assertFalse(main.done)
self.assertEqual(client.calls, original_calls)
with patch("claude_side_question.ask_native_side_question", side_effect=SideQuestionError(503, "Unavailable")):
with self.assertRaises(SideQuestionError):
await self.ask()
self.assertEqual(client.calls, original_calls)
self.assertEqual(self.manager._supervisors["chat"].active_run_id, "main")
self.assertEqual(self.manager._pins, {})
async def test_saved_effort_change_uses_live_parent_without_reconfiguration(self) -> None:
self.options["effort"] = "ultracode"
main = await self.start_main()
client = self.factory.clients[0]
supervisor = self.manager._supervisors["chat"]
original_calls = list(client.calls)
saved_options = {**self.options, "effort": "max"}
with patch("claude_side_question.ask_native_side_question",
return_value={"answer": "Live context"}) as native:
answer = await self.manager.ask_side_question(
"chat", "Status?", options=saved_options,
configuration_key="saved-max-profile",
expected_provider_id="existing-provider-session",
)
self.assertEqual(answer, {"answer": "Live context"})
self.assertIs(native.call_args.args[0], client)
self.assertFalse(main.done)
self.assertEqual(client.calls, original_calls)
self.assertEqual(len(self.factory.clients), 1)
self.assertIs(self.manager._supervisors["chat"], supervisor)
self.assertEqual(supervisor.configuration_key, "profile")
self.assertEqual(client.options["effort"], "ultracode")
self.assertEqual(self.manager._pins, {})
with self.assertRaises(ClaudeSDKConfigurationConflict):
await self.manager.get_mcp_status(
"chat", options=saved_options,
configuration_key="saved-max-profile",
)
async def test_lease_blocks_idle_eviction_and_configuration_replacement(self) -> None:
started = asyncio.Event()
release = asyncio.Event()
async def native(*args, **kwargs):
started.set()
await release.wait()
return {"answer": "Done"}
with patch("claude_side_question.ask_native_side_question", native):
side = self.ask()
await asyncio.wait_for(started.wait(), 1)
self.assertFalse(await self.manager.evict("chat"))
with self.assertRaises(ClaudeSDKConfigurationConflict):
await self.manager.get("chat", options={}, configuration_key="different")
self.assertEqual(len(self.factory.clients), 1)
release.set()
await side
self.assertTrue(await self.manager.evict("chat"))
async def test_retired_side_reply_is_rejected_and_cannot_unpin_replacement(self) -> None:
first_started = asyncio.Event()
second_started = asyncio.Event()
old_cancelled = asyncio.Event()
allow_old_cleanup = asyncio.Event()
finish_new = asyncio.Event()
async def native(client, *args, **kwargs):
if client is self.factory.clients[0]:
first_started.set()
try:
await asyncio.Future()
except asyncio.CancelledError:
old_cancelled.set()
await allow_old_cleanup.wait()
return {"answer": "Stale"}
second_started.set()
await finish_new.wait()
return {"answer": "Replacement"}
with patch("claude_side_question.ask_native_side_question", native):
old = self.ask()
await asyncio.wait_for(first_started.wait(), 1)
self.assertTrue(await self.manager.evict("chat", force=True))
await asyncio.wait_for(old_cancelled.wait(), 1)
new = self.ask()
await asyncio.wait_for(second_started.wait(), 1)
allow_old_cleanup.set()
with self.assertRaises(ClaudeSDKGenerationChanged):
await old
self.assertEqual(self.manager._pins, {"chat": 1})
self.assertFalse(self.factory.clients[1].disconnected)
finish_new.set()
self.assertEqual(await new, {"answer": "Replacement"})
self.assertEqual(self.manager._pins, {})
async def test_cancel_during_cold_connect_leaves_connection_for_waiting_main(self) -> None:
connecting = asyncio.Event()
connected = asyncio.Event()
factory = self.factory
class SlowClient(FakeClaudeClient):
async def connect(self):
connecting.set()
await connected.wait()
await super().connect()
def make_client(options):
client = SlowClient(options)
factory.clients.append(client)
return client
self.manager._client_factory = make_client
with patch("claude_side_question.ask_native_side_question") as native:
side = self.ask()
await asyncio.wait_for(connecting.wait(), 1)
side.cancel()
with self.assertRaises(asyncio.CancelledError):
await side
self.assertEqual(self.manager._pins, {})
connected.set()
main = await self.start_main()
self.assertFalse(main.done)
native.assert_not_called()
self.assertEqual(len(factory.clients), 1)
self.assertFalse(factory.clients[0].disconnected)
async def test_cold_connection_failure_unpins_without_retiring_supervisor(self) -> None:
self.factory.connect_error = RuntimeError("Synthetic connect failure")
with patch("claude_side_question.ask_native_side_question") as native:
with self.assertRaises(ClaudeSDKUnavailable):
await self.ask()
native.assert_not_called()
self.assertEqual(self.manager._pins, {})
self.assertFalse(self.manager._supervisors["chat"].closed)
self.factory.connect_error = None
await self.start_main()
self.assertEqual(len(self.factory.clients), 2)
async def test_cold_resume_rejects_missing_or_different_native_parent(self) -> None:
for resume in (None, "", "different-provider"):
with self.subTest(resume=resume):
self.options = {"resume": resume}
with patch("claude_side_question.ask_native_side_question") as native:
with self.assertRaises(ClaudeSDKUnavailable):
await self.ask(expected_provider_id="existing-provider-session")
native.assert_not_called()
self.assertEqual(self.factory.clients, [])
self.assertEqual(self.manager._pins, {})
async def test_connected_main_without_resume_option_retains_its_native_context(self) -> None:
self.options = {"resume": None}
main = await self.start_main()
original_calls = list(self.factory.clients[0].calls)
with patch("claude_side_question.ask_native_side_question", return_value={"answer": "Live context"}) as native:
self.assertEqual(await self.ask(expected_provider_id="now-bound-provider"),
{"answer": "Live context"})
self.assertIs(native.call_args.args[0], self.factory.clients[0])
self.assertFalse(main.done)
self.assertEqual(self.factory.clients[0].calls, original_calls)
async def test_side_control_progress_never_enters_main_run_projection(self) -> None:
main = await self.start_main()
client = self.factory.clients[0]
side_id = "agentsdock_side_" + "a" * 32
await client.emit({"type": "system", "subtype": "control_request_progress",
"request_id": side_id, "status": "started"})
await client.emit(SimpleNamespace(subtype="control_request_progress", data={
"request_id": side_id, "status": "api_retry"}))
other_control = {"type": "system", "subtype": "control_request_progress",
"request_id": "main-control", "status": "started"}
await client.emit(other_control)
main_text = {"type": "assistant", "content": [{"type": "text", "text": "Main answer"}]}
await client.emit(main_text)
result = {"type": "result", "subtype": "success", "result": "Main answer"}
await client.emit(result)
projected = await asyncio.wait_for(self.collect(main), 1)
self.assertEqual(projected, [other_control, main_text, result])
self.assertEqual([call[0] for call in client.calls].count("query"), 1)
@staticmethod
async def collect(handle):
return [message async for message in handle]
if __name__ == "__main__":
unittest.main()