cleveragents-core/robot/helper_container_resolve_crash.py

"""Robot Framework helper for bug #647 regression tests.

Runs `plan tree`, `plan explain`, and `plan correct` through real CLI
subprocesses with real DI/container wiring (no mocks).
"""

from __future__ import annotations

import os
import sys
import tempfile
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path
from typing import NoReturn

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

_ROBOT = str(Path(__file__).resolve().parent)
if _ROBOT not in sys.path:
    sys.path.insert(0, _ROBOT)

from helper_e2e_common import run_cli  # noqa: E402
from ulid import ULID  # noqa: E402

from cleveragents.application.container import reset_container  # noqa: E402
from cleveragents.application.services import decision_service as ds_mod  # noqa: E402
from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
    PlanLifecycleService,
)
from cleveragents.config.settings import Settings  # noqa: E402
from cleveragents.domain.models.core.decision import (  # noqa: E402
    ContextSnapshot,
    DecisionType,
    ResourceRef,
)
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork  # noqa: E402


@dataclass(frozen=True)
class DecisionContext:
    """Data required to run and verify one CLI command."""

    plan_id: str
    decision_id: str
    database_url: str
    db_path: str
    uow: UnitOfWork


def _fail(message: str) -> NoReturn:
    """Exit with expected regression-failure status."""
    print(f"FAIL: {message}", file=sys.stderr)
    sys.exit(1)


def _fail_unexpected(message: str) -> NoReturn:
    """Exit with unexpected-infrastructure-failure status."""
    print(f"UNEXPECTED: {message}", file=sys.stderr)
    sys.exit(2)


def _setup_decisions() -> DecisionContext:
    """Create a real sqlite database and seed a small decision tree."""
    fd, db_path = tempfile.mkstemp(prefix="cr647_", suffix=".db")
    os.close(fd)
    database_url = f"sqlite:///{db_path}"
    os.environ["CLEVERAGENTS_DATABASE_URL"] = database_url

    reset_container()
    uow = UnitOfWork(database_url)
    uow.init_database()

    settings = Settings(database_url=database_url, async_enabled=False)
    decision_svc = ds_mod.DecisionService(
        settings=settings,
        unit_of_work=uow,
    )

    lifecycle_svc = PlanLifecycleService(settings=settings, unit_of_work=uow)
    action_name = f"local/cr647-{ULID()!s}"
    lifecycle_svc.create_action(
        name=action_name,
        description="Container.resolve regression action",
        definition_of_done="Regression guard setup complete",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
    )
    plan = lifecycle_svc.use_action(action_name)
    plan_id = plan.identity.plan_id
    root = decision_svc.record_decision(
        plan_id=plan_id,
        decision_type=DecisionType.PROMPT_DEFINITION,
        question="What should we build?",
        chosen_option="A REST API",
        alternatives_considered=["GraphQL API", "gRPC service"],
        confidence_score=0.95,
        rationale="REST fits requirements",
        context_snapshot=ContextSnapshot(
            hot_context_hash="sha256:root",
            hot_context_ref="store://root",
            relevant_resources=[
                ResourceRef(resource_id=str(ULID()), path="src/main.py"),
            ],
            actor_state_ref="checkpoint://root",
        ),
    )
    child = decision_svc.record_decision(
        plan_id=plan_id,
        decision_type=DecisionType.STRATEGY_CHOICE,
        question="Which framework should we use?",
        chosen_option="FastAPI",
        parent_decision_id=root.decision_id,
        alternatives_considered=["Flask", "Django"],
        confidence_score=0.90,
        rationale="FastAPI fits async service requirements",
        context_snapshot=ContextSnapshot(
            hot_context_hash="sha256:child",
            hot_context_ref="store://child",
            relevant_resources=[
                ResourceRef(resource_id=str(ULID()), path="src/api.py"),
            ],
            actor_state_ref="checkpoint://child",
        ),
    )
    return DecisionContext(
        plan_id=plan_id,
        decision_id=child.decision_id,
        database_url=database_url,
        db_path=db_path,
        uow=uow,
    )


def _cleanup(ctx: DecisionContext) -> None:
    """Best-effort cleanup that does not mask later teardown steps."""
    with suppress(Exception):
        ctx.uow.engine.dispose()
    with suppress(Exception):
        reset_container()
    with suppress(Exception):
        os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
    for suffix in ("", "-journal", "-wal", "-shm"):
        with suppress(OSError):
            Path(ctx.db_path + suffix).unlink(missing_ok=True)
    # Keep reset last to avoid stale singleton recreation during cleanup.
    with suppress(Exception):
        Settings.reset()


def _assert_tree_output(ctx: DecisionContext, output: str) -> None:
    """Require tree output to include the seeded decision id."""
    if ctx.decision_id not in output:
        _fail(
            "Expected plan tree output to include seeded decision data.\n"
            f"Output:\n{output}"
        )


def _assert_explain_output(ctx: DecisionContext, output: str) -> None:
    """Require explain output to include seeded id and content."""
    if ctx.decision_id not in output:
        _fail(
            "Expected plan explain output to include seeded decision id.\n"
            f"Output:\n{output}"
        )
    if "FastAPI" not in output:
        _fail(
            "Expected plan explain output to include child chosen decision content "
            "(FastAPI).\n"
            f"Output:\n{output}"
        )


def _assert_correct_output(ctx: DecisionContext, output: str) -> None:
    """Require correct output to reference the seeded decision id and mode."""
    if ctx.decision_id not in output:
        _fail(
            "Expected plan correct output to include seeded decision id.\n"
            f"Output:\n{output}"
        )
    # Validate command-specific content: dry-run mode or revert indication.
    lowered = output.lower()
    if "revert" not in lowered and "dry" not in lowered:
        _fail(
            "Expected plan correct output to reference revert mode or dry-run.\n"
            f"Output:\n{output}"
        )


def _run_and_verify(
    *,
    label: str,
    cli_args_factory: Callable[[DecisionContext], list[str]],
    positive_assertion: Callable[[DecisionContext, str], None],
) -> None:
    """Run one command end-to-end and enforce regression invariants."""
    ctx = _setup_decisions()
    try:
        args = cli_args_factory(ctx)
        result = run_cli(
            *args,
            workspace=str(Path.cwd()),
            env_extra={"CLEVERAGENTS_DATABASE_URL": ctx.database_url},
            timeout=120,
        )
        output = (result.stdout or "") + (result.stderr or "")
        lowered = output.lower()

        if result.returncode != 0:
            if "attributeerror" in lowered and "has no attribute 'resolve'" in lowered:
                _fail(
                    f"{label} unexpectedly reproduced resolve() crash.\n"
                    f"rc={result.returncode}\n{output}"
                )
            _fail_unexpected(
                f"{label} failed with unexpected non-zero exit.\n"
                f"rc={result.returncode}\n{output}"
            )

        if "attributeerror" in lowered or "has no attribute 'resolve'" in lowered:
            _fail(
                f"{label} printed resolve() crash text despite exit 0.\n"
                f"Output:\n{output}"
            )

        positive_assertion(ctx, output)
        print(f"{label}-ok")
    finally:
        _cleanup(ctx)


def plan_tree_crash() -> None:
    """Regression guard for ``plan tree`` DI-container path."""
    _run_and_verify(
        label="plan-tree-crash",
        cli_args_factory=lambda ctx: ["plan", "tree", ctx.plan_id, "--format", "json"],
        positive_assertion=_assert_tree_output,
    )


def plan_explain_crash() -> None:
    """Regression guard for ``plan explain`` DI-container path."""
    _run_and_verify(
        label="plan-explain-crash",
        cli_args_factory=lambda ctx: [
            "plan",
            "explain",
            ctx.decision_id,
            "--show-context",
            "--show-reasoning",
            "--format",
            "json",
        ],
        positive_assertion=_assert_explain_output,
    )


def plan_correct_crash() -> None:
    """Regression guard for ``plan correct`` DI-container path."""
    _run_and_verify(
        label="plan-correct-crash",
        cli_args_factory=lambda ctx: [
            "plan",
            "correct",
            ctx.decision_id,
            "--mode",
            "revert",
            "--guidance",
            "Use Django instead",
            # NOTE: ``--plan`` is required by implementation when there is no
            # active-plan CLI context, though spec examples show it omitted.
            # See spec-vs-implementation drift tracked separately.
            "--plan",
            ctx.plan_id,
            "--dry-run",
            "--format",
            "json",
        ],
        positive_assertion=_assert_correct_output,
    )


_COMMANDS: dict[str, Callable[[], None]] = {
    "plan-tree-crash": plan_tree_crash,
    "plan-explain-crash": plan_explain_crash,
    "plan-correct-crash": plan_correct_crash,
}


def main() -> int:
    """Dispatch helper command from Robot ``Run Process``."""
    if len(sys.argv) < 2:
        print(f"Usage: helper_container_resolve_crash.py <{'|'.join(_COMMANDS)}>")
        return 1

    handler = _COMMANDS.get(sys.argv[1])
    if handler is None:
        print(f"Unknown command: {sys.argv[1]}")
        return 1

    handler()
    return 0


if __name__ == "__main__":
    sys.exit(main())