KiloStar/tests/unit/test_individual_nodes.py

"""Regulatory / Consciousness / Control 三个核心节点的 working/_run 分支逻辑测试。

绕过 ``@ray.remote`` 装饰，直接通过 ``__ray_actor_class__`` 取出原始类，
mock 掉内部的 pydantic-ai Agent，验证节点对各类输入的分发与异常吞吐。
"""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest


# ─── RegulatoryNode ─────────────────────────────────────────────────────────


@pytest.fixture
def regulatory_instance():
    from kilostar.core.individual.regulatory_node.regulatory_node import (
        RegulatoryNode,
    )
    cls = RegulatoryNode.__ray_actor_class__
    obj = cls.__new__(cls)
    from kilostar.utils.logger import get_logger
    obj.logger = get_logger("regulatory_node")
    obj.agent = None
    return obj


@pytest.mark.asyncio
async def test_regulatory_run_returns_response_with_platform_filled(
    regulatory_instance,
):
    from kilostar.core.individual.regulatory_node.template import (
        MessageRequest,
        MessageResponse,
    )

    fake_response = MessageResponse(
        platform=None, platform_id=None, reply_message="hi"
    )
    agent_run_result = SimpleNamespace(output=fake_response)
    regulatory_instance.agent = MagicMock()
    regulatory_instance.agent.run = AsyncMock(return_value=agent_run_result)

    req = MessageRequest(
        platform="client",
        user_name="alice",
        platform_id="abc",
        message="hello",
    )
    out = await regulatory_instance.working(req)

    assert out is fake_response
    assert out.platform == "client"
    assert out.platform_id == "abc"
    regulatory_instance.agent.run.assert_awaited_once()


@pytest.mark.asyncio
async def test_regulatory_run_swallows_exception_returns_none(regulatory_instance):
    from kilostar.core.individual.regulatory_node.template import MessageRequest

    regulatory_instance.agent = MagicMock()
    regulatory_instance.agent.run = AsyncMock(side_effect=RuntimeError("boom"))

    req = MessageRequest(
        platform="onebot",
        user_name="bob",
        platform_id="x",
        message="hello",
    )
    out = await regulatory_instance.working(req)
    assert out is None


# ─── ControlNode ────────────────────────────────────────────────────────────


@pytest.fixture
def control_instance():
    from kilostar.core.individual.control_node.control_node import ControlNode
    cls = ControlNode.__ray_actor_class__
    obj = cls.__new__(cls)
    from kilostar.utils.logger import get_logger
    obj.logger = get_logger("control_node")
    obj.agent = None
    return obj


def _make_workflow_step():
    from kilostar.core.work.workflow.workflow import WorkflowStep

    return WorkflowStep(
        step=1,
        name="do something",
        action="execute the thing",
        inputs=None,
        outputs="result",
    )


@pytest.mark.asyncio
async def test_control_working_returns_for_workflow_output(control_instance):
    from kilostar.core.individual.control_node.template import (
        ForWorkflow,
        ForWorkflowInput,
    )

    step = _make_workflow_step()
    expected = ForWorkflow(output="done")
    agent_run_result = SimpleNamespace(output=expected)

    control_instance.agent = MagicMock()
    control_instance.agent.run = AsyncMock(return_value=agent_run_result)

    out = await control_instance.working(ForWorkflowInput(workflow_step=step))
    assert out is expected


@pytest.mark.asyncio
async def test_control_working_swallows_exception_returns_none(control_instance):
    from kilostar.core.individual.control_node.template import ForWorkflowInput

    step = _make_workflow_step()
    control_instance.agent = MagicMock()
    control_instance.agent.run = AsyncMock(side_effect=RuntimeError("boom"))

    out = await control_instance.working(ForWorkflowInput(workflow_step=step))
    assert out is None


# ─── ConsciousnessNode ──────────────────────────────────────────────────────


@pytest.fixture
def consciousness_instance():
    from kilostar.core.individual.consciousness_node.consciousness_node import (
        ConsciousnessNode,
    )
    cls = ConsciousnessNode.__ray_actor_class__
    obj = cls.__new__(cls)
    from kilostar.utils.logger import get_logger
    obj.logger = get_logger("consciousness_node")
    obj.agent = None
    return obj


@pytest.mark.asyncio
async def test_consciousness_working_dispatches_workflow_engine_input(
    consciousness_instance,
):
    from kilostar.core.individual.consciousness_node.template import (
        ForWorkflowEngine,
        ForWorkflowEngineInput,
    )
    from kilostar.core.work.workflow.workflow import KiloStarWorkflow
    from kilostar.core.work.workflow.model import WorkflowMetadata

    workflow = KiloStarWorkflow(
        title="t",
        work_link=[],
        workflow_metadata=WorkflowMetadata(),
    )
    expected = ForWorkflowEngine(workflow=workflow, reasoning="r")
    agent_run_result = SimpleNamespace(output=expected)

    consciousness_instance.agent = MagicMock()
    consciousness_instance.agent.run = AsyncMock(return_value=agent_run_result)

    out = await consciousness_instance.working(
        ForWorkflowEngineInput(original_command="cmd", available_skills=[])
    )
    assert out is expected
    assert isinstance(out, ForWorkflowEngine)


@pytest.mark.asyncio
async def test_consciousness_working_returns_none_on_unknown_output(
    consciousness_instance,
):
    """Agent 返回的不是三种已知 ForXxx 类型时，working 应返回 None。"""
    from kilostar.core.individual.consciousness_node.template import (
        ForWorkflowEngineInput,
    )

    agent_run_result = SimpleNamespace(output="unexpected string")
    consciousness_instance.agent = MagicMock()
    consciousness_instance.agent.run = AsyncMock(return_value=agent_run_result)

    out = await consciousness_instance.working(
        ForWorkflowEngineInput(original_command="cmd", available_skills=[])
    )
    assert out is None


@pytest.mark.asyncio
async def test_consciousness_working_swallows_exception(consciousness_instance):
    from kilostar.core.individual.consciousness_node.template import (
        ForWorkflowEngineInput,
    )

    consciousness_instance.agent = MagicMock()
    consciousness_instance.agent.run = AsyncMock(side_effect=RuntimeError("boom"))

    out = await consciousness_instance.working(
        ForWorkflowEngineInput(original_command="cmd", available_skills=[])
    )
    assert out is None