cleveragents-core/robot/helper_context_tiers.py

"""Robot Framework helper for ACMS context tier smoke tests.

Provides a CLI-style interface for Robot to invoke tier model creation,
service operations, and DI resolution.  Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_context_tiers.py <command>
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.application.container import (  # noqa: E402
    get_container,
    reset_container,
)
from cleveragents.application.services.context_tiers import (  # noqa: E402
    ContextTierService,
)
from cleveragents.domain.models.acms.tiers import (  # noqa: E402
    ActorContextView,
    ActorRole,
    ContextTier,
    ScopedBackendView,
    TierBudget,
    TieredFragment,
    TierMetrics,
)


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print("Usage: helper_context_tiers.py <command>")
        return 1

    command: str = sys.argv[1]

    if command == "models":
        try:
            assert ContextTier.HOT == "hot"
            assert ActorRole.STRATEGIST == "strategist"
            frag = TieredFragment(fragment_id="f1", content="hello")
            assert frag.tier == ContextTier.HOT
            budget = TierBudget()
            assert budget.max_tokens_hot == 16000
            view = ActorContextView(actor_role=ActorRole.EXECUTOR)
            assert len(view.get_effective_tiers()) == 2
            metrics = TierMetrics()
            assert metrics.total_fragments == 0
            print("context-tiers-models-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-models-fail: {exc}")
            return 1

    if command == "store-get":
        try:
            svc = ContextTierService()
            frag = TieredFragment(
                fragment_id="f1", content="test", tier=ContextTier.HOT
            )
            svc.store(frag)
            got = svc.get("f1")
            assert got is not None
            assert got.content == "test"
            assert svc.get("missing") is None
            print("context-tiers-store-get-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-store-get-fail: {exc}")
            return 1

    if command == "promote-demote":
        try:
            svc = ContextTierService()
            frag = TieredFragment(
                fragment_id="f1", content="data", tier=ContextTier.COLD
            )
            svc.store(frag)
            promoted = svc.promote("f1")
            assert promoted is not None
            assert promoted.tier == ContextTier.WARM
            promoted2 = svc.promote("f1")
            assert promoted2 is not None
            assert promoted2.tier == ContextTier.HOT
            demoted = svc.demote("f1")
            assert demoted is not None
            assert demoted.tier == ContextTier.WARM
            print("context-tiers-promote-demote-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-promote-demote-fail: {exc}")
            return 1

    if command == "eviction":
        try:
            svc = ContextTierService()
            for i in range(5):
                svc.store(
                    TieredFragment(
                        fragment_id=f"e{i}",
                        content=f"data-{i}",
                        tier=ContextTier.HOT,
                    )
                )
            evicted = svc.evict_lru(ContextTier.HOT, 3)
            assert len(evicted) == 3
            assert svc.get_metrics().hot_count == 2
            print("context-tiers-eviction-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-eviction-fail: {exc}")
            return 1

    if command == "actor-views":
        try:
            svc = ContextTierService()
            svc.store(
                TieredFragment(
                    fragment_id="h1",
                    content="hot",
                    tier=ContextTier.HOT,
                    project_name="p1",
                )
            )
            svc.store(
                TieredFragment(
                    fragment_id="w1",
                    content="warm",
                    tier=ContextTier.WARM,
                    project_name="p1",
                )
            )
            svc.store(
                TieredFragment(
                    fragment_id="c1",
                    content="cold",
                    tier=ContextTier.COLD,
                    project_name="p1",
                )
            )
            strat = svc.get_for_actor(ActorRole.STRATEGIST, ["p1"])
            assert len(strat) == 3
            exec_frags = svc.get_for_actor(ActorRole.EXECUTOR, ["p1"])
            assert len(exec_frags) == 2
            rev = svc.get_for_actor(ActorRole.REVIEWER, ["p1"])
            assert len(rev) == 1
            print("context-tiers-actor-views-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-actor-views-fail: {exc}")
            return 1

    if command == "scoping":
        try:
            svc = ContextTierService()
            svc.store(
                TieredFragment(
                    fragment_id="s1",
                    content="yes",
                    tier=ContextTier.HOT,
                    project_name="alpha",
                )
            )
            svc.store(
                TieredFragment(
                    fragment_id="s2",
                    content="no",
                    tier=ContextTier.HOT,
                    project_name="beta",
                )
            )
            scoped = svc.get_scoped_view(["alpha"])
            assert len(scoped) == 1
            assert scoped[0].fragment_id == "s1"

            # ScopedBackendView direct test
            view = ScopedBackendView(allowed_projects=["alpha"])
            frag_a = TieredFragment(fragment_id="a", content="x", project_name="alpha")
            frag_b = TieredFragment(fragment_id="b", content="x", project_name="beta")
            assert view.is_visible(frag_a)
            assert not view.is_visible(frag_b)
            print("context-tiers-scoping-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-scoping-fail: {exc}")
            return 1

    if command == "di-resolution":
        try:
            reset_container()
            container = get_container()
            svc = container.context_tier_service()
            assert isinstance(svc, ContextTierService)
            svc2 = container.context_tier_service()
            assert svc is svc2  # singleton
            print("context-tiers-di-ok")
            return 0
        except Exception as exc:
            print(f"context-tiers-di-fail: {exc}")
            return 1

    print(f"Unknown command: {command}")
    return 1


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