[claude] Implement Kimi delegation for heavy research via Gitea labels (#979) (#1085)

Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>

tests/unit/test_kimi_delegation.py

@@ -0,0 +1,463 @@
+"""Unit tests for timmy.kimi_delegation — Kimi research delegation via Gitea labels."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from timmy.kimi_delegation import (
+    KIMI_LABEL_COLOR,
+    KIMI_READY_LABEL,
+    _build_research_template,
+    _extract_action_items,
+    _slugify,
+    delegate_research_to_kimi,
+    exceeds_local_capacity,
+)
+
+
+# ── Constants ─────────────────────────────────────────────────────────────────
+
+
+def test_kimi_ready_label():
+    assert KIMI_READY_LABEL == "kimi-ready"
+
+
+def test_kimi_label_color_is_hex():
+    assert KIMI_LABEL_COLOR.startswith("#")
+    assert len(KIMI_LABEL_COLOR) == 7
+
+
+# ── exceeds_local_capacity ────────────────────────────────────────────────────
+
+
+class TestExceedsLocalCapacity:
+    def test_keyword_comprehensive(self):
+        assert exceeds_local_capacity("Do a comprehensive review of X") is True
+
+    def test_keyword_deep_research(self):
+        assert exceeds_local_capacity("deep research into neural networks") is True
+
+    def test_keyword_benchmark(self):
+        assert exceeds_local_capacity("benchmark these five models") is True
+
+    def test_keyword_exhaustive(self):
+        assert exceeds_local_capacity("exhaustive list of options") is True
+
+    def test_keyword_case_insensitive(self):
+        assert exceeds_local_capacity("COMPREHENSIVE analysis") is True
+
+    def test_keyword_survey(self):
+        assert exceeds_local_capacity("survey all available tools") is True
+
+    def test_keyword_extensive(self):
+        assert exceeds_local_capacity("extensive documentation needed") is True
+
+    def test_short_simple_task(self):
+        assert exceeds_local_capacity("fix the login bug") is False
+
+    def test_long_task_exceeds_word_threshold(self):
+        long_task = " ".join(["word"] * 55)
+        assert exceeds_local_capacity(long_task) is True
+
+    def test_exactly_at_threshold(self):
+        at_threshold = " ".join(["word"] * 50)
+        assert exceeds_local_capacity(at_threshold) is True
+
+    def test_just_below_threshold(self):
+        short = " ".join(["word"] * 49)
+        assert exceeds_local_capacity(short) is False
+
+    def test_empty_string(self):
+        assert exceeds_local_capacity("") is False
+
+
+# ── _slugify ──────────────────────────────────────────────────────────────────
+
+
+class TestSlugify:
+    def test_simple_text(self):
+        assert _slugify("Hello World") == "hello-world"
+
+    def test_special_characters_removed(self):
+        assert _slugify("Hello, World!") == "hello-world"
+
+    def test_underscores_become_dashes(self):
+        assert _slugify("hello_world") == "hello-world"
+
+    def test_multiple_spaces(self):
+        assert _slugify("hello   world") == "hello-world"
+
+    def test_truncates_to_60(self):
+        long = "a" * 80
+        result = _slugify(long)
+        assert len(result) <= 60
+
+    def test_no_leading_trailing_dashes(self):
+        result = _slugify("  hello  ")
+        assert not result.startswith("-")
+        assert not result.endswith("-")
+
+    def test_empty_string(self):
+        assert _slugify("") == ""
+
+
+# ── _build_research_template ──────────────────────────────────────────────────
+
+
+class TestBuildResearchTemplate:
+    def test_contains_task(self):
+        body = _build_research_template("My Task", "some context", "What is X?")
+        assert "My Task" in body
+
+    def test_contains_question(self):
+        body = _build_research_template("Task", "ctx", "What is the answer?")
+        assert "What is the answer?" in body
+
+    def test_contains_context(self):
+        body = _build_research_template("Task", "project background", "Q?")
+        assert "project background" in body
+
+    def test_contains_kimi_ready_label(self):
+        body = _build_research_template("Task", "ctx", "Q?")
+        assert KIMI_READY_LABEL in body
+
+    def test_default_priority_normal(self):
+        body = _build_research_template("Task", "ctx", "Q?")
+        assert "normal" in body
+
+    def test_custom_priority_high(self):
+        body = _build_research_template("Task", "ctx", "Q?", priority="high")
+        assert "high" in body
+
+    def test_contains_deliverables_section(self):
+        body = _build_research_template("Task", "ctx", "Q?")
+        assert "Deliverables" in body
+
+    def test_slug_in_artifact_path(self):
+        body = _build_research_template("My Research Task", "ctx", "Q?")
+        assert "my-research-task" in body
+
+    def test_contains_research_request_header(self):
+        body = _build_research_template("Task", "ctx", "Q?")
+        assert "## Research Request" in body
+
+
+# ── _extract_action_items ─────────────────────────────────────────────────────
+
+
+class TestExtractActionItems:
+    def test_checkbox_items(self):
+        text = "- [ ] Do thing A\n- [ ] Do thing B"
+        items = _extract_action_items(text)
+        assert "Do thing A" in items
+        assert "Do thing B" in items
+
+    def test_numbered_list(self):
+        text = "1. First step\n2. Second step\n3. Third step"
+        items = _extract_action_items(text)
+        assert "First step" in items
+        assert "Second step" in items
+        assert "Third step" in items
+
+    def test_action_prefix(self):
+        text = "Action: Implement caching layer"
+        items = _extract_action_items(text)
+        assert "Implement caching layer" in items
+
+    def test_todo_prefix(self):
+        text = "TODO: Write tests"
+        items = _extract_action_items(text)
+        assert "Write tests" in items
+
+    def test_next_step_prefix(self):
+        text = "Next step: Deploy to staging"
+        items = _extract_action_items(text)
+        assert "Deploy to staging" in items
+
+    def test_case_insensitive_prefixes(self):
+        text = "TODO: Upper\ntodo: lower\nTodo: Mixed"
+        items = _extract_action_items(text)
+        assert len(items) == 3
+
+    def test_deduplication(self):
+        text = "1. Do the thing\n2. Do the thing"
+        items = _extract_action_items(text)
+        assert items.count("Do the thing") == 1
+
+    def test_empty_text(self):
+        assert _extract_action_items("") == []
+
+    def test_no_action_items(self):
+        text = "This is just a paragraph with no action items."
+        assert _extract_action_items(text) == []
+
+    def test_returns_list(self):
+        assert isinstance(_extract_action_items("1. Item"), list)
+
+
+# ── delegate_research_to_kimi ─────────────────────────────────────────────────
+
+
+class TestDelegateResearchToKimi:
+    @pytest.mark.asyncio
+    async def test_empty_task_returns_error(self):
+        result = await delegate_research_to_kimi("", "context", "question?")
+        assert result["success"] is False
+        assert "task" in result["error"].lower()
+
+    @pytest.mark.asyncio
+    async def test_whitespace_task_returns_error(self):
+        result = await delegate_research_to_kimi("   ", "context", "question?")
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_empty_question_returns_error(self):
+        result = await delegate_research_to_kimi("Task title", "context", "")
+        assert result["success"] is False
+        assert "question" in result["error"].lower()
+
+    @pytest.mark.asyncio
+    async def test_whitespace_question_returns_error(self):
+        result = await delegate_research_to_kimi("Task", "ctx", "   ")
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_delegates_to_create_issue(self):
+        with patch(
+            "timmy.kimi_delegation.create_kimi_research_issue",
+            new_callable=AsyncMock,
+            return_value={
+                "success": True,
+                "issue_number": 42,
+                "issue_url": "http://x/42",
+                "error": None,
+            },
+        ) as mock_create:
+            result = await delegate_research_to_kimi("Task", "ctx", "What is X?", "high")
+            mock_create.assert_awaited_once_with("Task", "ctx", "What is X?", "high")
+            assert result["success"] is True
+            assert result["issue_number"] == 42
+
+    @pytest.mark.asyncio
+    async def test_passes_default_priority(self):
+        with patch(
+            "timmy.kimi_delegation.create_kimi_research_issue",
+            new_callable=AsyncMock,
+            return_value={"success": True, "issue_number": 1, "issue_url": "", "error": None},
+        ) as mock_create:
+            await delegate_research_to_kimi("Task", "ctx", "Q?")
+            _, _, _, priority = mock_create.call_args.args
+            assert priority == "normal"
+
+
+# ── create_kimi_research_issue ────────────────────────────────────────────────
+
+
+class TestCreateKimiResearchIssue:
+    @pytest.mark.asyncio
+    async def test_no_gitea_token_returns_error(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = ""
+
+        with patch("config.settings", mock_settings):
+            result = await create_kimi_research_issue("Task", "ctx", "Q?")
+        assert result["success"] is False
+        assert "not configured" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_gitea_disabled_returns_error(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = False
+        mock_settings.gitea_token = "tok"
+
+        with patch("config.settings", mock_settings):
+            result = await create_kimi_research_issue("Task", "ctx", "Q?")
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_successful_issue_creation(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "fake-token"
+        mock_settings.gitea_url = "http://gitea.local"
+        mock_settings.gitea_repo = "owner/repo"
+
+        label_resp = MagicMock()
+        label_resp.status_code = 200
+        label_resp.json.return_value = [{"name": "kimi-ready", "id": 7}]
+
+        issue_resp = MagicMock()
+        issue_resp.status_code = 201
+        issue_resp.json.return_value = {
+            "number": 101,
+            "html_url": "http://gitea.local/issues/101",
+        }
+
+        mock_client = AsyncMock()
+        mock_client.get.return_value = label_resp
+        mock_client.post.return_value = issue_resp
+
+        async_ctx = AsyncMock()
+        async_ctx.__aenter__.return_value = mock_client
+        async_ctx.__aexit__.return_value = False
+
+        with (
+            patch("config.settings", mock_settings),
+            patch("httpx.AsyncClient", return_value=async_ctx),
+        ):
+            result = await create_kimi_research_issue("Task", "ctx", "Q?")
+
+        assert result["success"] is True
+        assert result["issue_number"] == 101
+        assert result["error"] is None
+
+    @pytest.mark.asyncio
+    async def test_api_error_returns_failure(self):
+        from timmy.kimi_delegation import create_kimi_research_issue
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://gitea.local"
+        mock_settings.gitea_repo = "owner/repo"
+
+        label_resp = MagicMock()
+        label_resp.status_code = 200
+        label_resp.json.return_value = [{"name": "kimi-ready", "id": 7}]
+
+        issue_resp = MagicMock()
+        issue_resp.status_code = 500
+        issue_resp.text = "Internal Server Error"
+
+        mock_client = AsyncMock()
+        mock_client.get.return_value = label_resp
+        mock_client.post.return_value = issue_resp
+
+        async_ctx = AsyncMock()
+        async_ctx.__aenter__.return_value = mock_client
+        async_ctx.__aexit__.return_value = False
+
+        with (
+            patch("config.settings", mock_settings),
+            patch("httpx.AsyncClient", return_value=async_ctx),
+        ):
+            result = await create_kimi_research_issue("Task", "ctx", "Q?")
+
+        assert result["success"] is False
+        assert "500" in result["error"]
+
+
+# ── index_kimi_artifact ───────────────────────────────────────────────────────
+
+
+class TestIndexKimiArtifact:
+    @pytest.mark.asyncio
+    async def test_empty_artifact_returns_error(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        result = await index_kimi_artifact(42, "Title", "")
+        assert result["success"] is False
+        assert "Empty" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_whitespace_only_artifact_returns_error(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        result = await index_kimi_artifact(42, "Title", "   \n  ")
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_successful_indexing(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        mock_entry = MagicMock()
+        mock_entry.id = "mem-abc-123"
+
+        with patch("timmy.memory_system.store_memory", return_value=mock_entry) as mock_store:
+            result = await index_kimi_artifact(55, "Research Title", "Artifact content here.")
+
+        assert result["success"] is True
+        assert result["memory_id"] == "mem-abc-123"
+        mock_store.assert_called_once()
+        call_kwargs = mock_store.call_args.kwargs
+        assert call_kwargs["source"] == "kimi"
+        assert call_kwargs["context_type"] == "document"
+        assert call_kwargs["task_id"] == "55"
+
+    @pytest.mark.asyncio
+    async def test_store_memory_exception_returns_error(self):
+        from timmy.kimi_delegation import index_kimi_artifact
+
+        with patch(
+            "timmy.memory_system.store_memory",
+            side_effect=RuntimeError("DB error"),
+        ):
+            result = await index_kimi_artifact(1, "T", "Some content")
+        assert result["success"] is False
+        assert "DB error" in result["error"]
+
+
+# ── extract_and_create_followups ──────────────────────────────────────────────
+
+
+class TestExtractAndCreateFollowups:
+    @pytest.mark.asyncio
+    async def test_no_action_items_returns_empty_list(self):
+        from timmy.kimi_delegation import extract_and_create_followups
+
+        result = await extract_and_create_followups("No action items here.", 10)
+        assert result["success"] is True
+        assert result["created"] == []
+        assert result["error"] is None
+
+    @pytest.mark.asyncio
+    async def test_gitea_not_configured(self):
+        from timmy.kimi_delegation import extract_and_create_followups
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = False
+        mock_settings.gitea_token = ""
+
+        with patch("config.settings", mock_settings):
+            result = await extract_and_create_followups("1. Do the thing", 10)
+        assert result["success"] is False
+        assert result["created"] == []
+
+    @pytest.mark.asyncio
+    async def test_creates_followup_issues(self):
+        from timmy.kimi_delegation import extract_and_create_followups
+
+        mock_settings = MagicMock()
+        mock_settings.gitea_enabled = True
+        mock_settings.gitea_token = "tok"
+        mock_settings.gitea_url = "http://gitea.local"
+        mock_settings.gitea_repo = "owner/repo"
+
+        issue_resp = MagicMock()
+        issue_resp.status_code = 201
+        issue_resp.json.return_value = {"number": 200}
+
+        mock_client = AsyncMock()
+        mock_client.post.return_value = issue_resp
+
+        async_ctx = AsyncMock()
+        async_ctx.__aenter__.return_value = mock_client
+        async_ctx.__aexit__.return_value = False
+
+        with (
+            patch("config.settings", mock_settings),
+            patch("httpx.AsyncClient", return_value=async_ctx),
+        ):
+            result = await extract_and_create_followups(
+                "1. Do the thing\n2. Do another thing", 10
+            )
+
+        assert result["success"] is True
+        assert 200 in result["created"]