baa712ba15
Position the spawn modal directly under the 'New Agent' button without a blur overlay. Uses click-outside dismissal and absolute positioning. Reduces visual disruption for quick agent spawning.
482 lines
19 KiB
Python
482 lines
19 KiB
Python
"""Tests for mixins/conversation.py edge cases.
|
|
|
|
Unit tests for conversation parsing from Claude Code and Codex JSONL files.
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from amc_server.mixins.conversation import ConversationMixin
|
|
from amc_server.mixins.parsing import SessionParsingMixin
|
|
|
|
|
|
class DummyConversationHandler(ConversationMixin, SessionParsingMixin):
|
|
"""Minimal handler for testing conversation mixin."""
|
|
|
|
def __init__(self):
|
|
self.sent_responses = []
|
|
|
|
def _send_json(self, code, payload):
|
|
self.sent_responses.append((code, payload))
|
|
|
|
|
|
class TestParseCodexArguments(unittest.TestCase):
|
|
"""Tests for _parse_codex_arguments edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_dict_input_returned_as_is(self):
|
|
result = self.handler._parse_codex_arguments({"key": "value"})
|
|
self.assertEqual(result, {"key": "value"})
|
|
|
|
def test_empty_dict_returned_as_is(self):
|
|
result = self.handler._parse_codex_arguments({})
|
|
self.assertEqual(result, {})
|
|
|
|
def test_json_string_parsed(self):
|
|
result = self.handler._parse_codex_arguments('{"key": "value"}')
|
|
self.assertEqual(result, {"key": "value"})
|
|
|
|
def test_invalid_json_string_returns_raw(self):
|
|
result = self.handler._parse_codex_arguments("not valid json")
|
|
self.assertEqual(result, {"raw": "not valid json"})
|
|
|
|
def test_empty_string_returns_raw(self):
|
|
result = self.handler._parse_codex_arguments("")
|
|
self.assertEqual(result, {"raw": ""})
|
|
|
|
def test_none_returns_empty_dict(self):
|
|
result = self.handler._parse_codex_arguments(None)
|
|
self.assertEqual(result, {})
|
|
|
|
def test_int_returns_empty_dict(self):
|
|
result = self.handler._parse_codex_arguments(42)
|
|
self.assertEqual(result, {})
|
|
|
|
def test_list_returns_empty_dict(self):
|
|
result = self.handler._parse_codex_arguments([1, 2, 3])
|
|
self.assertEqual(result, {})
|
|
|
|
|
|
class TestServeEvents(unittest.TestCase):
|
|
"""Tests for _serve_events edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_path_traversal_sanitized(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
events_dir = Path(tmpdir)
|
|
# Create a file that path traversal might try to access (unused - documents intent)
|
|
_secret_file = Path(tmpdir).parent / "secret.jsonl"
|
|
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", events_dir):
|
|
# Try path traversal
|
|
self.handler._serve_events("../secret")
|
|
|
|
# Should have served response with sanitized id
|
|
self.assertEqual(len(self.handler.sent_responses), 1)
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(payload["session_id"], "secret")
|
|
self.assertEqual(payload["events"], [])
|
|
|
|
def test_nonexistent_file_returns_empty_events(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", Path(tmpdir)):
|
|
self.handler._serve_events("nonexistent")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(payload["events"], [])
|
|
|
|
def test_empty_file_returns_empty_events(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
events_dir = Path(tmpdir)
|
|
event_file = events_dir / "session123.jsonl"
|
|
event_file.write_text("")
|
|
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", events_dir):
|
|
self.handler._serve_events("session123")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(payload["events"], [])
|
|
|
|
def test_invalid_json_lines_skipped(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
events_dir = Path(tmpdir)
|
|
event_file = events_dir / "session123.jsonl"
|
|
event_file.write_text('{"valid": "event"}\nnot json\n{"another": "event"}\n')
|
|
|
|
with patch("amc_server.mixins.conversation.EVENTS_DIR", events_dir):
|
|
self.handler._serve_events("session123")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(len(payload["events"]), 2)
|
|
self.assertEqual(payload["events"][0], {"valid": "event"})
|
|
self.assertEqual(payload["events"][1], {"another": "event"})
|
|
|
|
|
|
class TestParseClaudeConversation(unittest.TestCase):
|
|
"""Tests for _parse_claude_conversation edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_user_message_with_array_content_skipped(self):
|
|
# Array content is tool results, not human messages
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "user",
|
|
"message": {"content": [{"type": "tool_result"}]}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages, [])
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_user_message_with_string_content_included(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "user",
|
|
"timestamp": "2024-01-01T00:00:00Z",
|
|
"message": {"content": "Hello, Claude!"}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["role"], "user")
|
|
self.assertEqual(messages[0]["content"], "Hello, Claude!")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_with_text_parts(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"timestamp": "2024-01-01T00:00:00Z",
|
|
"message": {
|
|
"content": [
|
|
{"type": "text", "text": "Part 1"},
|
|
{"type": "text", "text": "Part 2"},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["content"], "Part 1\nPart 2")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_with_tool_use(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{"type": "tool_use", "name": "Read", "input": {"file_path": "/test"}},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "Read")
|
|
self.assertEqual(messages[0]["tool_calls"][0]["input"]["file_path"], "/test")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_with_thinking(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{"type": "thinking", "thinking": "Let me consider..."},
|
|
{"type": "text", "text": "Here's my answer"},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["thinking"], "Let me consider...")
|
|
self.assertEqual(messages[0]["content"], "Here's my answer")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_assistant_message_content_as_string_parts(self):
|
|
# Some entries might have string content parts instead of dicts
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": ["plain string", {"type": "text", "text": "structured"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages[0]["content"], "plain string\nstructured")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_missing_conversation_file_returns_empty(self):
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=None):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages, [])
|
|
|
|
def test_non_dict_entry_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write('"just a string"\n')
|
|
f.write('123\n')
|
|
f.write('{"type": "user", "message": {"content": "valid"}}\n')
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(len(messages), 1)
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_non_list_content_in_assistant_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "assistant",
|
|
"message": {"content": "not a list"}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_get_claude_conversation_file", return_value=path):
|
|
messages = self.handler._parse_claude_conversation("session123", "/project")
|
|
self.assertEqual(messages, [])
|
|
finally:
|
|
path.unlink()
|
|
|
|
|
|
class TestParseCodexConversation(unittest.TestCase):
|
|
"""Tests for _parse_codex_conversation edge cases."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_developer_role_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "developer",
|
|
"content": [{"text": "System instructions"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(messages, [])
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_injected_context_skipped(self):
|
|
skip_prefixes = [
|
|
"<INSTRUCTIONS>",
|
|
"<environment_context>",
|
|
"<permissions instructions>",
|
|
"# AGENTS.md instructions",
|
|
]
|
|
for prefix in skip_prefixes:
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"text": f"{prefix} more content here"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(messages, [], f"Should skip content starting with {prefix}")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_function_call_accumulated_to_next_assistant(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
# Tool call
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "function_call",
|
|
"name": "shell",
|
|
"arguments": '{"command": "ls"}'
|
|
}
|
|
}) + "\n")
|
|
# Assistant message
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "assistant",
|
|
"content": [{"text": "Here are the files"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "shell")
|
|
self.assertEqual(messages[0]["content"], "Here are the files")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_function_calls_flushed_before_user_message(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
# Tool call
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {"type": "function_call", "name": "tool1", "arguments": "{}"}
|
|
}) + "\n")
|
|
# User message (tool calls should be flushed first)
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"text": "User response"}]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
# First message should be assistant with tool_calls (flushed)
|
|
# Second should be user
|
|
self.assertEqual(len(messages), 2)
|
|
self.assertEqual(messages[0]["role"], "assistant")
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "tool1")
|
|
self.assertEqual(messages[1]["role"], "user")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_reasoning_creates_thinking_message(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {
|
|
"type": "reasoning",
|
|
"summary": [
|
|
{"type": "summary_text", "text": "Let me think..."},
|
|
{"type": "summary_text", "text": "I'll try this approach."},
|
|
]
|
|
}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["thinking"], "Let me think...\nI'll try this approach.")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_pending_tool_calls_flushed_at_end(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
# Tool call with no following message
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {"type": "function_call", "name": "final_tool", "arguments": "{}"}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
# Should flush pending tool calls at end
|
|
self.assertEqual(len(messages), 1)
|
|
self.assertEqual(messages[0]["tool_calls"][0]["name"], "final_tool")
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_non_response_item_types_skipped(self):
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
|
|
f.write('{"type": "session_meta"}\n')
|
|
f.write('{"type": "event_msg"}\n')
|
|
f.write(json.dumps({
|
|
"type": "response_item",
|
|
"payload": {"type": "message", "role": "user", "content": [{"text": "Hello"}]}
|
|
}) + "\n")
|
|
path = Path(f.name)
|
|
|
|
try:
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=path):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(len(messages), 1)
|
|
finally:
|
|
path.unlink()
|
|
|
|
def test_missing_transcript_file_returns_empty(self):
|
|
with patch.object(self.handler, "_find_codex_transcript_file", return_value=None):
|
|
messages = self.handler._parse_codex_conversation("session123")
|
|
self.assertEqual(messages, [])
|
|
|
|
|
|
class TestServeConversation(unittest.TestCase):
|
|
"""Tests for _serve_conversation routing."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummyConversationHandler()
|
|
|
|
def test_routes_to_codex_parser(self):
|
|
with patch.object(self.handler, "_parse_codex_conversation", return_value=[]) as mock:
|
|
self.handler._serve_conversation("session123", "/project", agent="codex")
|
|
mock.assert_called_once_with("session123")
|
|
|
|
def test_routes_to_claude_parser_by_default(self):
|
|
with patch.object(self.handler, "_parse_claude_conversation", return_value=[]) as mock:
|
|
self.handler._serve_conversation("session123", "/project")
|
|
mock.assert_called_once_with("session123", "/project")
|
|
|
|
def test_sanitizes_session_id(self):
|
|
with patch.object(self.handler, "_parse_claude_conversation", return_value=[]):
|
|
self.handler._serve_conversation("../../../etc/passwd", "/project")
|
|
|
|
code, payload = self.handler.sent_responses[0]
|
|
self.assertEqual(payload["session_id"], "passwd")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|