tests/test_openclaw.py

"""Tests for OpenClaw integration — OpenClawToTraceStep and end-to-end pipeline."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Optional

import pytest

from ace.core.context import ACEStepContext, SkillbookView
from ace.core.outputs import (
    AgentOutput,
    ReflectorOutput,
    SkillManagerOutput,
)
from ace.core.skillbook import Skillbook, UpdateBatch, UpdateOperation
from ace.integrations.openclaw import OpenClawToTraceStep
from ace.steps import learning_tail
from ace.steps.load_traces import LoadTracesStep

from pipeline import Pipeline

# ------------------------------------------------------------------ #
# Helpers — mock roles
# ------------------------------------------------------------------ #


class MockReflector:
    """Minimal mock satisfying ReflectorLike."""

    def __init__(self, output: ReflectorOutput | None = None):
        self.output = output or ReflectorOutput(
            reasoning="test reasoning",
            correct_approach="test approach",
            key_insight="test insight",
        )
        self.calls: list[dict] = []

    def reflect(
        self,
        *,
        question: str,
        agent_output: AgentOutput,
        skillbook: Any,
        ground_truth: Optional[str] = None,
        feedback: Optional[str] = None,
        **kwargs: Any,
    ) -> ReflectorOutput:
        self.calls.append(
            {
                "question": question,
                "agent_output": agent_output,
                "ground_truth": ground_truth,
                "feedback": feedback,
                **kwargs,
            }
        )
        return self.output


class MockSkillManager:
    """Minimal mock satisfying SkillManagerLike.

    The real SM mutates the skillbook directly via tool calls; this mock
    applies its pre-canned ``output`` to the incoming skillbook so
    ``UpdateStep`` behaves like the live code path.
    """

    def __init__(self, output: SkillManagerOutput | None = None):
        self.output = output or SkillManagerOutput(
            update=UpdateBatch(reasoning="test", operations=[]),
        )
        self.calls: list[dict] = []

    def update_skills(
        self,
        *,
        reflections: tuple[ReflectorOutput, ...],
        skillbook: Any,
        question_context: str,
        progress: str,
        **kwargs: Any,
    ) -> SkillManagerOutput:
        self.calls.append(
            {
                "reflections": reflections,
                "question_context": question_context,
                "progress": progress,
            }
        )
        skillbook.apply_update(self.output.update)
        return self.output


# ------------------------------------------------------------------ #
# Fixtures
# ------------------------------------------------------------------ #


@pytest.fixture
def sample_jsonl(tmp_path: Path) -> Path:
    """Create a minimal OpenClaw session JSONL file."""
    events = [
        {
            "type": "session",
            "id": "s1",
            "timestamp": "2026-01-01T00:00:00Z",
            "version": 1,
            "cwd": "/app",
        },
        {
            "type": "message",
            "id": "m1",
            "parentId": "s1",
            "timestamp": "2026-01-01T00:00:01Z",
            "message": {
                "role": "user",
                "content": [{"type": "text", "text": "Hello, help me debug this."}],
            },
        },
        {
            "type": "message",
            "id": "m2",
            "parentId": "m1",
            "timestamp": "2026-01-01T00:00:02Z",
            "message": {
                "role": "assistant",
                "content": [
                    {"type": "thinking", "thinking": "Let me analyze the issue..."},
                    {"type": "text", "text": "I'll help you debug this."},
                    {
                        "type": "toolCall",
                        "id": "tc1",
                        "name": "Read",
                        "arguments": {"file_path": "/app/main.py"},
                    },
                ],
            },
        },
        {
            "type": "message",
            "id": "m3",
            "parentId": "m2",
            "timestamp": "2026-01-01T00:00:03Z",
            "message": {
                "role": "toolResult",
                "content": [
                    {"type": "text", "text": "def main():\n    print('hello')"}
                ],
            },
        },
    ]
    path = tmp_path / "test-session.jsonl"
    path.write_text("\n".join(json.dumps(e) for e in events) + "\n")
    return path


# ------------------------------------------------------------------ #
# OpenClawToTraceStep tests
# ------------------------------------------------------------------ #


class TestOpenClawToTraceStep:
    def test_requires_provides(self):
        step = OpenClawToTraceStep()
        assert step.requires == frozenset({"trace"})
        assert step.provides == frozenset({"trace"})

    def test_converts_to_trace_dict(self):
        """Step should convert raw events into a structured trace dict."""
        raw_events = [
            {"type": "session", "id": "s1", "cwd": "/app"},
            {
                "type": "message",
                "id": "m1",
                "message": {
                    "role": "user",
                    "content": [{"type": "text", "text": "Hello"}],
                },
            },
            {
                "type": "message",
                "id": "m2",
                "message": {
                    "role": "assistant",
                    "content": [{"type": "text", "text": "Hi there"}],
                },
            },
        ]
        ctx = ACEStepContext(trace=raw_events)
        result = OpenClawToTraceStep()(ctx)

        trace = result.trace
        assert isinstance(trace, dict)
        assert trace["question"] == "User: Hello"
        assert trace["answer"] == "Hi there"
        assert trace["skill_ids"] == []
        assert trace["ground_truth"] is None
        assert "reasoning" in trace
        assert "feedback" in trace

    def test_none_trace(self):
        """Step should handle None trace gracefully."""
        ctx = ACEStepContext(trace=None)
        result = OpenClawToTraceStep()(ctx)
        assert result.trace is None

    def test_empty_list_trace(self):
        """Step should handle empty list trace gracefully."""
        ctx = ACEStepContext(trace=[])
        result = OpenClawToTraceStep()(ctx)
        assert result.trace == []


# ------------------------------------------------------------------ #
# End-to-end: LoadTracesStep → OpenClawToTraceStep → learning_tail
# ------------------------------------------------------------------ #


class TestOpenClawEndToEnd:
    def test_load_and_convert(self, sample_jsonl: Path):
        """LoadTracesStep → OpenClawToTraceStep should produce trace data."""
        load_step = LoadTracesStep()
        convert_step = OpenClawToTraceStep()

        ctx = ACEStepContext(sample=str(sample_jsonl))
        ctx = load_step(ctx)
        assert isinstance(ctx.trace, list)
        assert len(ctx.trace) == 4

        ctx = convert_step(ctx)
        # Converted to structured trace dict
        assert isinstance(ctx.trace, dict)
        assert "question" in ctx.trace
        assert "reasoning" in ctx.trace
        assert "answer" in ctx.trace
        assert ctx.trace["skill_ids"] == []
        assert ctx.trace["ground_truth"] is None

    def test_full_pipeline_with_mocks(self, sample_jsonl: Path):
        """Full pipeline: load → convert → reflect → tag → update → apply."""
        reflector = MockReflector()
        skill_manager = MockSkillManager()
        skillbook = Skillbook()

        load_step = LoadTracesStep()
        convert_step = OpenClawToTraceStep()

        steps = [
            load_step,
            convert_step,
            *learning_tail(reflector, skill_manager, skillbook),
        ]

        pipeline = Pipeline(steps)

        ctx = ACEStepContext(
            sample=str(sample_jsonl),
            skillbook=SkillbookView(skillbook),
        )

        result = pipeline.run([ctx])
        pipeline.wait_for_background()

        assert len(result) == 1
        assert len(reflector.calls) == 1
        assert len(skill_manager.calls) == 1

    def test_pipeline_with_add_operation(self, sample_jsonl: Path):
        """Pipeline with a SkillManager that adds a skill."""
        add_op = UpdateOperation(
            type="ADD",
            section="debugging",
            issue="Use structured logging for better debug traces",
            insight="Use structured logging for better debug traces",
            skill_id=None,
            metadata={"helpful": 1, "harmful": 0, "neutral": 0},
        )
        sm_output = SkillManagerOutput(
            update=UpdateBatch(reasoning="Found useful pattern", operations=[add_op]),
        )
        reflector = MockReflector()
        skill_manager = MockSkillManager(output=sm_output)
        skillbook = Skillbook()

        steps = [
            LoadTracesStep(),
            OpenClawToTraceStep(),
            *learning_tail(reflector, skill_manager, skillbook),
        ]

        pipeline = Pipeline(steps)
        ctx = ACEStepContext(
            sample=str(sample_jsonl),
            skillbook=SkillbookView(skillbook),
        )

        pipeline.run([ctx])
        pipeline.wait_for_background()

        # Skillbook should now have one skill (legacy "debugging" → "context")
        assert len(skillbook.skills()) == 1
        skill = skillbook.skills()[0]
        assert skill.section == "context"
        assert "structured logging" in skill.insight

    def test_empty_session_skipped(self, tmp_path: Path):
        """Empty JSONL should produce empty trace."""
        path = tmp_path / "empty.jsonl"
        path.write_text("")

        load_step = LoadTracesStep()
        convert_step = OpenClawToTraceStep()

        ctx = ACEStepContext(sample=str(path))
        ctx = load_step(ctx)
        assert ctx.trace == []

        ctx = convert_step(ctx)
        assert ctx.trace == []