cleveragents-core/robot/helper_plan_apply_json_envelope.py

"""Helper script for plan apply JSON envelope Robot tests.

Each scenario constructs a real ``LifecyclePlan`` domain object, calls
``_apply_output_dict`` directly, and asserts the envelope shape. Exits 0
on success and 1 on failure so Robot's ``Run Process`` keyword can check
``${result.rc}``.
"""

from __future__ import annotations

import json
import sys
from datetime import UTC, datetime


def _make_plan(phase_str="apply", state_str="applied", cost=None):
    """Build a real LifecyclePlan domain object."""
    from cleveragents.domain.models.core.cost_metadata import CostMetadata
    from cleveragents.domain.models.core.plan import (
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanPhase,
        PlanTimestamps,
        ProcessingState,
        ProjectLink,
    )

    now = datetime.now(tz=UTC)
    phase = PlanPhase(phase_str)
    state = ProcessingState(state_str)
    cost_metadata = CostMetadata(total_cost=cost) if cost is not None else None
    return Plan(
        identity=PlanIdentity(plan_id="01JAAAAAAAAAAAAAAAAAAAAAAA"),
        namespaced_name=NamespacedName(
            server=None, namespace="local", name="test-plan"
        ),
        action_name="local/test-action",
        description="Test plan for apply JSON envelope scenarios",
        definition_of_done="All tests pass",
        phase=phase,
        processing_state=state,
        project_links=[ProjectLink(project_name="local/my-project")],
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        created_by="test-user",
        reusable=True,
        read_only=False,
        timestamps=PlanTimestamps(
            created_at=now,
            updated_at=now,
            applied_at=now,
        ),
        cost_metadata=cost_metadata,
    )


def _envelope(plan=None):
    """Call ``_apply_output_dict`` with the supplied plan."""
    from cleveragents.cli.commands.plan import _apply_output_dict

    if plan is None:
        plan = _make_plan()
    return _apply_output_dict(plan)


def _ok(v="ok"):
    """Print success and exit 0."""
    print(json.dumps({"rc": 0, "stdout": v}))
    sys.exit(0)


def _err(msg):
    """Print error and exit 1."""
    print(json.dumps({"rc": 1, "stderr": msg}))
    sys.exit(1)


# ---------------------------------------------------------------------------
# Test scenarios - each checks one aspect of the envelope.
# ---------------------------------------------------------------------------


def verify_top_level_fields():
    """Verify the apply json envelope has all required top-level keys."""
    env = _envelope()
    req = {"command", "status", "exit_code", "data", "timing", "messages"}
    if req <= env.keys():
        _ok(f"keys={sorted(env.keys())}")
    _err(f"Missing: {req - env.keys()}")


def verify_command_field():
    """Verify command is 'plan apply'."""
    env = _envelope()
    cmd = str(env.get("command", ""))
    if cmd == "plan apply":
        _ok(f"cmd={cmd}")
    _err(f"Wanted 'plan apply', got {env.get('command')!r}")


def verify_status_field():
    """Verify status is 'ok'."""
    env = _envelope()
    st = str(env.get("status", ""))
    if st == "ok":
        _ok(f"status={st}")
    _err(f"Wanted 'ok', got {env.get('status')!r}")


def verify_exit_code():
    """Verify exit_code is 0."""
    env = _envelope()
    ec = env.get("exit_code", -1)
    if ec == 0:
        _ok(f"exit_code={ec}")
    _err(f"Wanted 0, got {ec!r}")


def verify_messages_non_empty():
    """Verify messages is a non-empty list."""
    env = _envelope()
    msgs = env.get("messages", [])
    if isinstance(msgs, list) and len(msgs) > 0:
        _ok(f"msgs={json.dumps(msgs)}")
    _err(f"Expected non-empty list, got {msgs!r}")


def verify_data_sub_fields():
    """Verify data field has all required sub-fields."""
    env = _envelope()
    d = env.get("data", {})
    req = {
        "artifacts",
        "changes",
        "project",
        "applied_at",
        "validation",
        "sandbox_cleanup",
        "lifecycle",
    }
    if req <= set(d.keys()):
        _ok(f"data keys={sorted(d.keys())}")
    _err(f"Missing: {req - set(d.keys())}")


def verify_validation_sub_fields():
    """Verify data.validation has test, lint, type_check."""
    env = _envelope()
    v = env.get("data", {}).get("validation", {})
    req = {"tests", "lint", "type_check"}
    if req <= set(v.keys()):
        _ok(f"validation keys={sorted(v.keys())}")
    _err(f"Missing: {req - set(v.keys())}")


def verify_lifecycle_sub_fields():
    """Verify data.lifecycle has required sub-fields."""
    env = _envelope()
    lc = env.get("data", {}).get("lifecycle", {})
    req = {
        "phase",
        "state",
        "total_duration",
        "total_cost",
        "decisions_made",
        "child_plans",
    }
    if req <= set(lc.keys()):
        _ok(f"lifecycle keys={sorted(lc.keys())}")
    _err(f"Missing: {req - set(lc.keys())}")


def verify_sandbox_cleanup_sub_fields():
    """Verify data.sandbox_cleanup has required fields."""
    env = _envelope()
    sc = env.get("data", {}).get("sandbox_cleanup", {})
    req = {"worktree", "branch", "checkpoint"}
    if req <= set(sc.keys()):
        _ok(f"sc keys={sorted(sc.keys())}")
    _err(f"Missing: {req - set(sc.keys())}")


def verify_terminal_sandbox_cleanup():
    """Verify terminal plan has removed/merged/archived."""
    env = _envelope(_make_plan(phase_str="apply", state_str="applied"))
    sc = env.get("data", {}).get("sandbox_cleanup", {})
    checks = [
        ("worktree", "removed"),
        ("branch", "merged to main"),
        ("checkpoint", "archived"),
    ]
    ok = all(str(sc.get(k)) == v for k, v in checks)
    if ok:
        _ok(f"sc={json.dumps(sc)}")
    failed = [(k, sc.get(k), v) for k, v in checks if str(sc.get(k)) != v]
    _err(f"Failed terminal checks: {failed}")


def verify_non_terminal_sandbox_cleanup():
    """Verify non-terminal plan has active/open/pending."""
    env = _envelope(_make_plan(phase_str="apply", state_str="queued"))
    sc = env.get("data", {}).get("sandbox_cleanup", {})
    checks = [
        ("worktree", "active"),
        ("branch", "open"),
        ("checkpoint", "pending"),
    ]
    ok = all(str(sc.get(k)) == v for k, v in checks)
    if ok:
        _ok(f"sc={json.dumps(sc)}")
    failed = [(k, sc.get(k), v) for k, v in checks if str(sc.get(k)) != v]
    _err(f"Failed non-terminal checks: {failed}")


def verify_legacy_fallback():
    """Verify non-Plan objects get minimal envelope."""
    from cleveragents.cli.commands.plan import _apply_output_dict

    env = _apply_output_dict("legacy-plan-string")
    cmd_ok = env.get("command") == "plan apply"
    data_ok = "plan" in env.get("data", {})
    if cmd_ok and data_ok:
        _ok(f"env={json.dumps(env)}")
    _err("Legacy fallback failed")


def verify_cost_metadata():
    """Verify cost is reflected when cost_metadata present."""
    env = _envelope(_make_plan(cost=1.23))
    tc = str(env.get("data", {}).get("lifecycle", {}).get("total_cost", ""))
    if tc.startswith("$"):
        _ok(f"cost={tc}")
    _err(f"Wanted $ prefix, got {tc!r}")


def verify_status_no_apply_envelope():
    """Verify the apply envelope is gated on the apply command only.

    The plan status path uses ``_plan_spec_dict`` (flat schema) and therefore
    must not produce ``command == "plan apply"``. We assert the envelope from
    ``_plan_spec_dict`` differs from the apply envelope.
    """
    from cleveragents.cli.commands.plan import _plan_spec_dict

    spec = _plan_spec_dict(_make_plan(phase_str="strategize", state_str="queued"))
    if spec.get("command") != "plan apply":
        _ok(f"spec_keys={sorted(spec.keys())[:5]}")
    _err(f"_plan_spec_dict leaked command='plan apply': {spec.get('command')!r}")


def verify_cancel_no_apply_envelope():
    """Verify the apply envelope is gated on apply only (cancel uses spec dict)."""
    from cleveragents.cli.commands.plan import _plan_spec_dict

    spec = _plan_spec_dict(_make_plan(phase_str="apply", state_str="cancelled"))
    if spec.get("command") != "plan apply":
        _ok(f"spec_keys={sorted(spec.keys())[:5]}")
    _err(f"_plan_spec_dict leaked command='plan apply': {spec.get('command')!r}")


# ---------------------------------------------------------------------------
# Dispatch entry point.
# ---------------------------------------------------------------------------

SCENARIOS = {
    "verify_top_level_fields": verify_top_level_fields,
    "verify_command_field": verify_command_field,
    "verify_status_field": verify_status_field,
    "verify_exit_code": verify_exit_code,
    "verify_messages_non_empty": verify_messages_non_empty,
    "verify_data_sub_fields": verify_data_sub_fields,
    "verify_validation_sub_fields": verify_validation_sub_fields,
    "verify_lifecycle_sub_fields": verify_lifecycle_sub_fields,
    "verify_sandbox_cleanup_sub_fields": verify_sandbox_cleanup_sub_fields,
    "verify_terminal_sandbox_cleanup": verify_terminal_sandbox_cleanup,
    "verify_non_terminal_sandbox_cleanup": verify_non_terminal_sandbox_cleanup,
    "verify_legacy_fallback": verify_legacy_fallback,
    "verify_cost_metadata": verify_cost_metadata,
    "verify_status_no_apply_envelope": verify_status_no_apply_envelope,
    "verify_cancel_no_apply_envelope": verify_cancel_no_apply_envelope,
}


def main() -> None:
    """Dispatch to the requested scenario."""
    if len(sys.argv) < 2 or sys.argv[1] not in SCENARIOS:
        _err(f"Unknown: {sys.argv[1] if len(sys.argv) > 1 else '(none)'}")
        return  # unreachable but type-check friendly

    SCENARIOS[sys.argv[1]]()


if __name__ == "__main__":
    main()