[kimi] Implement token quest system for agents (#713) (#789)

tests/unit/test_quest_system.py

@@ -0,0 +1,489 @@
+"""Unit tests for the quest system.
+
+Tests quest definitions, progress tracking, completion detection,
+and token rewards.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from timmy.quest_system import (
+    QuestDefinition,
+    QuestProgress,
+    QuestStatus,
+    QuestType,
+    _is_on_cooldown,
+    claim_quest_reward,
+    evaluate_quest_progress,
+    get_or_create_progress,
+    get_quest_definition,
+    get_quest_leaderboard,
+    load_quest_config,
+    reset_quest_progress,
+    update_quest_progress,
+)
+
+
+@pytest.fixture(autouse=True)
+def clean_quest_state():
+    """Reset quest progress between tests."""
+    reset_quest_progress()
+    yield
+    reset_quest_progress()
+
+
+@pytest.fixture
+def sample_issue_count_quest():
+    """Create a sample issue_count quest definition."""
+    return QuestDefinition(
+        id="test_close_issues",
+        name="Test Issue Closer",
+        description="Close 3 test issues",
+        reward_tokens=100,
+        quest_type=QuestType.ISSUE_COUNT,
+        enabled=True,
+        repeatable=False,
+        cooldown_hours=0,
+        criteria={"target_count": 3, "issue_labels": ["test"]},
+        notification_message="Test quest complete! Earned {tokens} tokens.",
+    )
+
+
+@pytest.fixture
+def sample_daily_run_quest():
+    """Create a sample daily_run quest definition."""
+    return QuestDefinition(
+        id="test_daily_run",
+        name="Test Daily Runner",
+        description="Complete 5 sessions",
+        reward_tokens=250,
+        quest_type=QuestType.DAILY_RUN,
+        enabled=True,
+        repeatable=True,
+        cooldown_hours=24,
+        criteria={"min_sessions": 5},
+        notification_message="Daily run quest complete! Earned {tokens} tokens.",
+    )
+
+
+# ── Quest Definition Tests ───────────────────────────────────────────────
+
+
+class TestQuestDefinition:
+    def test_from_dict_minimal(self):
+        data = {"id": "test_quest", "name": "Test Quest"}
+        quest = QuestDefinition.from_dict(data)
+        assert quest.id == "test_quest"
+        assert quest.name == "Test Quest"
+        assert quest.quest_type == QuestType.CUSTOM
+        assert quest.enabled is True
+
+    def test_from_dict_full(self):
+        data = {
+            "id": "full_quest",
+            "name": "Full Quest",
+            "description": "A test quest",
+            "reward_tokens": 500,
+            "type": "issue_count",
+            "enabled": False,
+            "repeatable": True,
+            "cooldown_hours": 12,
+            "criteria": {"target_count": 5},
+            "notification_message": "Done!",
+        }
+        quest = QuestDefinition.from_dict(data)
+        assert quest.id == "full_quest"
+        assert quest.reward_tokens == 500
+        assert quest.quest_type == QuestType.ISSUE_COUNT
+        assert quest.enabled is False
+        assert quest.repeatable is True
+        assert quest.cooldown_hours == 12
+
+
+# ── Quest Progress Tests ─────────────────────────────────────────────────
+
+
+class TestQuestProgress:
+    def test_progress_creation(self):
+        progress = QuestProgress(
+            quest_id="test_quest",
+            agent_id="test_agent",
+            status=QuestStatus.NOT_STARTED,
+        )
+        assert progress.quest_id == "test_quest"
+        assert progress.agent_id == "test_agent"
+        assert progress.current_value == 0
+
+    def test_progress_to_dict(self):
+        progress = QuestProgress(
+            quest_id="test_quest",
+            agent_id="test_agent",
+            status=QuestStatus.IN_PROGRESS,
+            current_value=2,
+            target_value=5,
+        )
+        data = progress.to_dict()
+        assert data["quest_id"] == "test_quest"
+        assert data["status"] == "in_progress"
+        assert data["current_value"] == 2
+
+
+# ── Quest Loading Tests ──────────────────────────────────────────────────
+
+
+class TestQuestLoading:
+    def test_load_quest_config(self):
+        definitions, settings = load_quest_config()
+        assert isinstance(definitions, dict)
+        assert isinstance(settings, dict)
+
+    def test_get_quest_definition_exists(self):
+        # Should return None for non-existent quest in fresh state
+        quest = get_quest_definition("nonexistent")
+        # The function returns from loaded config, which may have quests
+        # or be empty if config doesn't exist
+        assert quest is None or isinstance(quest, QuestDefinition)
+
+    def test_get_quest_definition_not_found(self):
+        quest = get_quest_definition("definitely_not_a_real_quest_12345")
+        assert quest is None
+
+
+# ── Quest Progress Management Tests ─────────────────────────────────────
+
+
+class TestQuestProgressManagement:
+    def test_get_or_create_progress_new(self):
+        # First create a quest definition
+        quest = QuestDefinition(
+            id="progress_test",
+            name="Progress Test",
+            description="Test quest",
+            reward_tokens=100,
+            quest_type=QuestType.ISSUE_COUNT,
+            enabled=True,
+            repeatable=False,
+            cooldown_hours=0,
+            criteria={"target_count": 3},
+            notification_message="Done!",
+        )
+
+        # Need to inject into the definitions dict
+        from timmy.quest_system import _quest_definitions
+
+        _quest_definitions["progress_test"] = quest
+
+        progress = get_or_create_progress("progress_test", "agent1")
+        assert progress.quest_id == "progress_test"
+        assert progress.agent_id == "agent1"
+        assert progress.status == QuestStatus.NOT_STARTED
+        assert progress.target_value == 3
+
+        del _quest_definitions["progress_test"]
+
+    def test_update_quest_progress(self):
+        quest = QuestDefinition(
+            id="update_test",
+            name="Update Test",
+            description="Test quest",
+            reward_tokens=100,
+            quest_type=QuestType.ISSUE_COUNT,
+            enabled=True,
+            repeatable=False,
+            cooldown_hours=0,
+            criteria={"target_count": 3},
+            notification_message="Done!",
+        )
+
+        from timmy.quest_system import _quest_definitions
+
+        _quest_definitions["update_test"] = quest
+
+        # Create initial progress
+        progress = get_or_create_progress("update_test", "agent1")
+        assert progress.current_value == 0
+
+        # Update progress
+        updated = update_quest_progress("update_test", "agent1", 2)
+        assert updated.current_value == 2
+        assert updated.status == QuestStatus.NOT_STARTED
+
+        # Complete the quest
+        completed = update_quest_progress("update_test", "agent1", 3)
+        assert completed.current_value == 3
+        assert completed.status == QuestStatus.COMPLETED
+        assert completed.completed_at != ""
+
+        del _quest_definitions["update_test"]
+
+
+# ── Quest Evaluation Tests ───────────────────────────────────────────────
+
+
+class TestQuestEvaluation:
+    def test_evaluate_issue_count_quest(self):
+        quest = QuestDefinition(
+            id="eval_test",
+            name="Eval Test",
+            description="Test quest",
+            reward_tokens=100,
+            quest_type=QuestType.ISSUE_COUNT,
+            enabled=True,
+            repeatable=False,
+            cooldown_hours=0,
+            criteria={"target_count": 2, "issue_labels": ["test"]},
+            notification_message="Done!",
+        )
+
+        from timmy.quest_system import _quest_definitions
+
+        _quest_definitions["eval_test"] = quest
+
+        # Simulate closed issues
+        closed_issues = [
+            {"id": 1, "labels": [{"name": "test"}]},
+            {"id": 2, "labels": [{"name": "test"}, {"name": "bug"}]},
+            {"id": 3, "labels": [{"name": "other"}]},
+        ]
+
+        context = {"closed_issues": closed_issues}
+        progress = evaluate_quest_progress("eval_test", "agent1", context)
+
+        assert progress is not None
+        assert progress.current_value == 2  # Two issues with 'test' label
+
+        del _quest_definitions["eval_test"]
+
+    def test_evaluate_issue_reduce_quest(self):
+        quest = QuestDefinition(
+            id="reduce_test",
+            name="Reduce Test",
+            description="Test quest",
+            reward_tokens=200,
+            quest_type=QuestType.ISSUE_REDUCE,
+            enabled=True,
+            repeatable=False,
+            cooldown_hours=0,
+            criteria={"target_reduction": 2},
+            notification_message="Done!",
+        )
+
+        from timmy.quest_system import _quest_definitions
+
+        _quest_definitions["reduce_test"] = quest
+
+        context = {"previous_issue_count": 10, "current_issue_count": 7}
+        progress = evaluate_quest_progress("reduce_test", "agent1", context)
+
+        assert progress is not None
+        assert progress.current_value == 3  # Reduced by 3
+
+        del _quest_definitions["reduce_test"]
+
+    def test_evaluate_daily_run_quest(self):
+        quest = QuestDefinition(
+            id="daily_test",
+            name="Daily Test",
+            description="Test quest",
+            reward_tokens=250,
+            quest_type=QuestType.DAILY_RUN,
+            enabled=True,
+            repeatable=True,
+            cooldown_hours=24,
+            criteria={"min_sessions": 5},
+            notification_message="Done!",
+        )
+
+        from timmy.quest_system import _quest_definitions
+
+        _quest_definitions["daily_test"] = quest
+
+        context = {"sessions_completed": 5}
+        progress = evaluate_quest_progress("daily_test", "agent1", context)
+
+        assert progress is not None
+        assert progress.current_value == 5
+        assert progress.status == QuestStatus.COMPLETED
+
+        del _quest_definitions["daily_test"]
+
+
+# ── Quest Cooldown Tests ─────────────────────────────────────────────────
+
+
+class TestQuestCooldown:
+    def test_is_on_cooldown_no_cooldown(self):
+        quest = QuestDefinition(
+            id="cooldown_test",
+            name="Cooldown Test",
+            description="Test quest",
+            reward_tokens=100,
+            quest_type=QuestType.ISSUE_COUNT,
+            enabled=True,
+            repeatable=True,
+            cooldown_hours=24,
+            criteria={},
+            notification_message="Done!",
+        )
+
+        progress = QuestProgress(
+            quest_id="cooldown_test",
+            agent_id="agent1",
+            status=QuestStatus.CLAIMED,
+        )
+
+        # No last_completed_at means no cooldown
+        assert _is_on_cooldown(progress, quest) is False
+
+
+# ── Quest Reward Tests ───────────────────────────────────────────────────
+
+
+class TestQuestReward:
+    def test_claim_quest_reward_not_completed(self):
+        quest = QuestDefinition(
+            id="reward_test",
+            name="Reward Test",
+            description="Test quest",
+            reward_tokens=100,
+            quest_type=QuestType.ISSUE_COUNT,
+            enabled=True,
+            repeatable=False,
+            cooldown_hours=0,
+            criteria={"target_count": 3},
+            notification_message="Done!",
+        )
+
+        from timmy.quest_system import _quest_definitions, _quest_progress
+
+        _quest_definitions["reward_test"] = quest
+
+        # Create progress but don't complete
+        progress = get_or_create_progress("reward_test", "agent1")
+        _quest_progress["agent1:reward_test"] = progress
+
+        # Try to claim - should fail
+        reward = claim_quest_reward("reward_test", "agent1")
+        assert reward is None
+
+        del _quest_definitions["reward_test"]
+
+
+# ── Leaderboard Tests ────────────────────────────────────────────────────
+
+
+class TestQuestLeaderboard:
+    def test_get_quest_leaderboard_empty(self):
+        reset_quest_progress()
+        leaderboard = get_quest_leaderboard()
+        assert leaderboard == []
+
+    def test_get_quest_leaderboard_with_data(self):
+        # Create and complete a quest for two agents
+        quest = QuestDefinition(
+            id="leaderboard_test",
+            name="Leaderboard Test",
+            description="Test quest",
+            reward_tokens=100,
+            quest_type=QuestType.ISSUE_COUNT,
+            enabled=True,
+            repeatable=True,
+            cooldown_hours=0,
+            criteria={"target_count": 1},
+            notification_message="Done!",
+        )
+
+        from timmy.quest_system import _quest_definitions, _quest_progress
+
+        _quest_definitions["leaderboard_test"] = quest
+
+        # Create progress for agent1 with 2 completions
+        progress1 = QuestProgress(
+            quest_id="leaderboard_test",
+            agent_id="agent1",
+            status=QuestStatus.NOT_STARTED,
+            completion_count=2,
+        )
+        _quest_progress["agent1:leaderboard_test"] = progress1
+
+        # Create progress for agent2 with 1 completion
+        progress2 = QuestProgress(
+            quest_id="leaderboard_test",
+            agent_id="agent2",
+            status=QuestStatus.NOT_STARTED,
+            completion_count=1,
+        )
+        _quest_progress["agent2:leaderboard_test"] = progress2
+
+        leaderboard = get_quest_leaderboard()
+
+        assert len(leaderboard) == 2
+        # agent1 should be first (more tokens)
+        assert leaderboard[0]["agent_id"] == "agent1"
+        assert leaderboard[0]["total_tokens"] == 200
+        assert leaderboard[1]["agent_id"] == "agent2"
+        assert leaderboard[1]["total_tokens"] == 100
+
+        del _quest_definitions["leaderboard_test"]
+
+
+# ── Quest Reset Tests ─────────────────────────────────────────────────────
+
+
+class TestQuestReset:
+    def test_reset_quest_progress_all(self):
+        # Create some progress entries
+        progress1 = QuestProgress(
+            quest_id="quest1", agent_id="agent1", status=QuestStatus.NOT_STARTED
+        )
+        progress2 = QuestProgress(
+            quest_id="quest2", agent_id="agent2", status=QuestStatus.NOT_STARTED
+        )
+
+        from timmy.quest_system import _quest_progress
+
+        _quest_progress["agent1:quest1"] = progress1
+        _quest_progress["agent2:quest2"] = progress2
+
+        assert len(_quest_progress) == 2
+
+        count = reset_quest_progress()
+        assert count == 2
+        assert len(_quest_progress) == 0
+
+    def test_reset_quest_progress_specific_quest(self):
+        progress1 = QuestProgress(
+            quest_id="quest1", agent_id="agent1", status=QuestStatus.NOT_STARTED
+        )
+        progress2 = QuestProgress(
+            quest_id="quest2", agent_id="agent1", status=QuestStatus.NOT_STARTED
+        )
+
+        from timmy.quest_system import _quest_progress
+
+        _quest_progress["agent1:quest1"] = progress1
+        _quest_progress["agent1:quest2"] = progress2
+
+        count = reset_quest_progress(quest_id="quest1")
+        assert count == 1
+        assert "agent1:quest1" not in _quest_progress
+        assert "agent1:quest2" in _quest_progress
+
+    def test_reset_quest_progress_specific_agent(self):
+        progress1 = QuestProgress(
+            quest_id="quest1", agent_id="agent1", status=QuestStatus.NOT_STARTED
+        )
+        progress2 = QuestProgress(
+            quest_id="quest1", agent_id="agent2", status=QuestStatus.NOT_STARTED
+        )
+
+        from timmy.quest_system import _quest_progress
+
+        _quest_progress["agent1:quest1"] = progress1
+        _quest_progress["agent2:quest1"] = progress2
+
+        count = reset_quest_progress(agent_id="agent1")
+        assert count == 1
+        assert "agent1:quest1" not in _quest_progress
+        assert "agent2:quest1" in _quest_progress