cleveragents-core/robot/helper_plan_diff_artifacts.py

"""Helper utilities for plan diff and artifacts Robot integration tests.

Covers:
- Diff output in multiple formats (rich, plain, JSON, YAML)
- Artifacts summary rendering
- Empty changeset guard
- Apply summary persistence
- Merge failure handling
"""

from __future__ import annotations

import json
import sys

from cleveragents.application.services.plan_apply_service import PlanApplyService
from cleveragents.application.services.plan_lifecycle_service import (
    PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.core.exceptions import PlanError
from cleveragents.domain.models.core.change import (
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
    SpecChangeSet,
)
from cleveragents.domain.models.core.plan import (
    PlanPhase,
    ProcessingState,
)


def _create_services() -> tuple[
    PlanLifecycleService, InMemoryChangeSetStore, PlanApplyService
]:
    """Create service instances for testing."""
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)
    store = InMemoryChangeSetStore()
    apply_svc = PlanApplyService(lifecycle_service=lifecycle, changeset_store=store)
    return lifecycle, store, apply_svc


def _create_plan_with_changeset(
    lifecycle: PlanLifecycleService,
    store: InMemoryChangeSetStore,
) -> str:
    """Create a test action, plan, and changeset. Return plan_id."""
    lifecycle.create_action(
        name="local/robot-diff-test",
        description="Robot diff test",
        definition_of_done="Tests pass",
        strategy_actor="local/planner",
        execution_actor="local/executor",
    )
    plan = lifecycle.use_action(action_name="local/robot-diff-test")
    plan_id = plan.identity.plan_id

    # Advance through strategize and execute
    lifecycle.start_strategize(plan_id)
    lifecycle.complete_strategize(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.STRATEGIZE:
        lifecycle.execute_plan(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.EXECUTE:
        lifecycle.start_execute(plan_id)
        lifecycle.complete_execute(plan_id)

    # Create changeset
    cs = SpecChangeSet(
        changeset_id="robot-cs-001",
        plan_id=plan_id,
        entries=[
            ChangeEntry(
                plan_id=plan_id,
                resource_id="res-001",
                tool_name="file-write",
                operation=ChangeOperation.CREATE,
                path="src/robot_new.py",
                after_hash="abc123",
            ),
            ChangeEntry(
                plan_id=plan_id,
                resource_id="res-002",
                tool_name="file-edit",
                operation=ChangeOperation.MODIFY,
                path="src/robot_existing.py",
                before_hash="old111",
                after_hash="new222",
            ),
        ],
    )
    store._store[cs.changeset_id] = cs

    plan = lifecycle.get_plan(plan_id)
    plan.changeset_id = cs.changeset_id
    plan.sandbox_refs = ["sandbox-robot-001"]
    lifecycle.commit_plan(plan)

    return plan_id


def _create_empty_changeset_plan(
    lifecycle: PlanLifecycleService,
    store: InMemoryChangeSetStore,
) -> str:
    """Create a plan with empty changeset. Return plan_id."""
    lifecycle.create_action(
        name="local/robot-empty-test",
        description="Robot empty test",
        definition_of_done="Tests pass",
        strategy_actor="local/planner",
        execution_actor="local/executor",
    )
    plan = lifecycle.use_action(action_name="local/robot-empty-test")
    plan_id = plan.identity.plan_id

    lifecycle.start_strategize(plan_id)
    lifecycle.complete_strategize(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.STRATEGIZE:
        lifecycle.execute_plan(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.EXECUTE:
        lifecycle.start_execute(plan_id)
        lifecycle.complete_execute(plan_id)

    cs = SpecChangeSet(changeset_id="robot-cs-empty", plan_id=plan_id, entries=[])
    store._store[cs.changeset_id] = cs
    plan = lifecycle.get_plan(plan_id)
    plan.changeset_id = cs.changeset_id
    lifecycle.commit_plan(plan)

    return plan_id


def _advance_to_apply_processing(lifecycle: PlanLifecycleService, plan_id: str) -> None:
    """Advance plan to apply/processing."""
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.EXECUTE and plan.state == ProcessingState.COMPLETE:
        lifecycle.apply_plan(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.APPLY and plan.state == ProcessingState.QUEUED:
        lifecycle.start_apply(plan_id)


def _diff_output() -> None:
    """Test diff output in rich format."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_plan_with_changeset(lifecycle, store)

    output = apply_svc.diff(plan_id, fmt="rich")
    assert "robot-cs-001" in output, f"Changeset ID missing: {output[:200]}"
    assert "src/robot_new.py" in output, f"Path missing: {output[:200]}"
    assert "src/robot_existing.py" in output, f"Path missing: {output[:200]}"
    print("diff-output-ok")


def _artifacts_output() -> None:
    """Test artifacts output.

    ``format_output`` for machine-readable formats writes directly to
    ``sys.stdout`` and returns ``""``.  We redirect stdout so the JSON
    payload is captured for parsing.
    """
    from io import StringIO

    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_plan_with_changeset(lifecycle, store)

    buf = StringIO()
    old_stdout = sys.stdout
    sys.stdout = buf
    try:
        output = apply_svc.artifacts(plan_id, fmt="json")
    finally:
        sys.stdout = old_stdout
    # Combine: return value or captured stdout
    effective = output or buf.getvalue().strip()
    parsed = json.loads(effective)
    assert parsed["changeset_id"] == "robot-cs-001"
    assert "sandbox-robot-001" in parsed["sandbox_refs"]
    assert len(parsed["files_changed"]) == 2
    print("artifacts-output-ok")


def _empty_guard() -> None:
    """Test empty changeset guard blocks apply."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_empty_changeset_plan(lifecycle, store)

    try:
        apply_svc.guard_empty_changeset(plan_id)
        raise AssertionError("Should have raised PlanError")
    except PlanError as e:
        assert "empty" in e.message.lower()
    print("empty-guard-ok")


def _empty_guard_allow() -> None:
    """Test empty changeset guard allows with flag."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_empty_changeset_plan(lifecycle, store)

    result = apply_svc.guard_empty_changeset(plan_id, allow_empty=True)
    assert result is True
    print("empty-guard-allow-ok")


def _apply_summary() -> None:
    """Test apply summary persistence."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_plan_with_changeset(lifecycle, store)
    _advance_to_apply_processing(lifecycle, plan_id)

    apply_svc.persist_apply_summary(plan_id, files_changed=7, validations_run=2)

    plan = lifecycle.get_plan(plan_id)
    assert plan.error_details is not None
    assert plan.error_details["apply_files_changed"] == "7"
    assert plan.error_details["apply_validations_run"] == "2"
    assert "apply_completed_at" in plan.error_details
    print("apply-summary-ok")


def _merge_failure() -> None:
    """Test merge failure handling."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_plan_with_changeset(lifecycle, store)
    _advance_to_apply_processing(lifecycle, plan_id)

    apply_svc.handle_merge_failure(plan_id, conflict_details="Conflict in src/main.py")

    plan = lifecycle.get_plan(plan_id)
    assert plan.processing_state == ProcessingState.ERRORED
    assert "Merge failed" in (plan.error_message or "")
    assert plan.error_details is not None
    assert "merge_conflict" in plan.error_details
    print("merge-failure-ok")


def _diff_json() -> None:
    """Test diff JSON format."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_plan_with_changeset(lifecycle, store)

    output = apply_svc.diff(plan_id, fmt="json")
    parsed = json.loads(output)
    assert "entries" in parsed
    assert "summary" in parsed
    assert len(parsed["entries"]) == 2
    print("diff-json-ok")


def _diff_yaml() -> None:
    """Test diff YAML format."""
    lifecycle, store, apply_svc = _create_services()
    plan_id = _create_plan_with_changeset(lifecycle, store)

    output = apply_svc.diff(plan_id, fmt="yaml")
    assert "changeset_id" in output
    assert "robot-cs-001" in output
    print("diff-yaml-ok")


COMMANDS: dict[str, object] = {
    "diff-output": _diff_output,
    "artifacts-output": _artifacts_output,
    "empty-guard": _empty_guard,
    "empty-guard-allow": _empty_guard_allow,
    "apply-summary": _apply_summary,
    "merge-failure": _merge_failure,
    "diff-json": _diff_json,
    "diff-yaml": _diff_yaml,
}


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
        print(f"Available: {', '.join(COMMANDS)}", file=sys.stderr)
        sys.exit(1)
    cmd = sys.argv[1]
    func = COMMANDS.get(cmd)
    if func is None:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)
    func()  # type: ignore[operator]