Add temporary AI commands and request transcripts

tests/test_ai_tool_loop.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace
+import unittest
+
+from app.ai.service import AIService
+from app.commands import CommandRegistry
+from app.config import AIConfig, AIModelConfig, AIPromptConfig, AIProviderConfig
+
+
+def _tool_response(call_id: str) -> SimpleNamespace:
+    call = SimpleNamespace(
+        id=call_id,
+        function=SimpleNamespace(
+            name="temporary_command_create",
+            arguments=json.dumps({"name": "test", "response": "ok"}),
+        ),
+    )
+    return SimpleNamespace(
+        choices=[
+            SimpleNamespace(
+                finish_reason="tool_calls",
+                message=SimpleNamespace(content=None, tool_calls=[call]),
+            )
+        ]
+    )
+
+
+class _FakeCompletions:
+    def __init__(self, responses: list[SimpleNamespace]) -> None:
+        self.responses = responses
+        self.requests: list[dict] = []
+
+    async def create(self, **kwargs: object) -> SimpleNamespace:
+        self.requests.append(dict(kwargs))
+        return self.responses.pop(0)
+
+
+class AIToolLoopTests(unittest.IsolatedAsyncioTestCase):
+    async def test_suppresses_duplicate_tool_call_and_returns_final_answer(self) -> None:
+        with TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            config = AIConfig(
+                default_model="test",
+                context_messages=0,
+                prompt=AIPromptConfig(root=root),
+                providers={"provider": AIProviderConfig("provider", "key", "https://example.test")},
+                models={"test": AIModelConfig("test", "provider", "model")},
+                workspace_dir=root / "workspace",
+            )
+            registry = CommandRegistry()
+            service = AIService(config, registry)
+            completions = _FakeCompletions(
+                [
+                    _tool_response("one"),
+                    _tool_response("two"),
+                    SimpleNamespace(
+                        choices=[
+                            SimpleNamespace(
+                                finish_reason="stop",
+                                message=SimpleNamespace(content="Created .test", tool_calls=[]),
+                            )
+                        ]
+                    ),
+                ]
+            )
+            service._clients["provider"] = SimpleNamespace(
+                chat=SimpleNamespace(completions=completions)
+            )
+            ctx = SimpleNamespace(
+                is_owner=True,
+                actor=None,
+                message=SimpleNamespace(chat=SimpleNamespace(id=1), id=1, text=".ai test"),
+                args="test",
+                replied_message=None,
+                replied_user=None,
+            )
+
+            response = await service.generate(ctx, SimpleNamespace(context_messages=0, model_flags=[], prompt="create test"))
+            service.close()
+
+        self.assertEqual(response.text, "Created .test")
+        self.assertIsNotNone(registry.resolve("test"))
+        self.assertEqual(len(completions.requests), 3)
+        self.assertIn("already been performed", completions.requests[2]["messages"][-1]["content"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ai_transcript.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+
+from app.ai.transcript import AITranscriptLogger
+
+
+class AITranscriptLoggerTests(unittest.TestCase):
+    def test_writes_jsonl_events(self) -> None:
+        with TemporaryDirectory() as temp_dir:
+            logger = AITranscriptLogger(Path(temp_dir))
+            logger.record("request", payload={"model": "test", "messages": []})
+            logger.record("response", payload={"choices": [{"message": {"content": "ok"}}]})
+
+            events = [json.loads(line) for line in logger.path.read_text(encoding="utf-8").splitlines()]
+            logger.close()
+
+        self.assertEqual([event["event"] for event in events], ["request", "response"])
+        self.assertEqual(events[1]["payload"]["choices"][0]["message"]["content"], "ok")

tests/test_temporary_commands.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+from app.commands import CommandRegistry, TemporaryCommandStore
+
+
+class TemporaryCommandStoreTests(unittest.IsolatedAsyncioTestCase):
+    async def test_created_command_is_owner_only_and_replies_from_memory(self) -> None:
+        registry = CommandRegistry()
+        store = TemporaryCommandStore(registry)
+
+        message = store.create("test-command", "Temporary response", "Test command")
+        command = registry.resolve("test-command")
+
+        self.assertIn(".test-command", message)
+        self.assertIsNotNone(command)
+        self.assertFalse(command.is_shared)
+        ctx = SimpleNamespace(reply=AsyncMock())
+        await command.handler(ctx)
+        ctx.reply.assert_awaited_once_with("Temporary response")
+
+    def test_rejects_existing_command_name(self) -> None:
+        registry = CommandRegistry()
+        store = TemporaryCommandStore(registry)
+        store.create("test", "First")
+
+        with self.assertRaisesRegex(ValueError, "already exists"):
+            store.create("test", "Second")