"""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 """ 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 ") 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())