HAL9000
2dd920078a
The PR added resolve_actor_options() to StrategyActor, LLMStrategizeActor, and LLMExecuteActor, but the three mock lifecycle SimpleNamespace objects used by tests did not include this method, causing AttributeError at runtime: - features/mocks/mock_strategy_llm.py: make_mock_lifecycle() used by robot/helper_strategy_actor.py (strategy_actor.robot llm-json test) and features/steps/strategy_actor_llm_steps.py - features/steps/llm_actors_coverage_steps.py: _make_mock_lifecycle() used by llm_actors_coverage.feature scenarios - robot/helper_m5_e2e_context.py: inline lifecycle SimpleNamespace used by m5_e2e_verification.robot Execute Phase LLM Uses ACMS Context All three now include resolve_actor_options=MagicMock(return_value=None) matching the None-return contract the actors use when no custom backend is configured. ISSUES CLOSED: #11256
511 lines
19 KiB
Python
511 lines
19 KiB
Python
"""Context-policy verification commands for M5 Robot E2E tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
from unittest.mock import MagicMock
|
|
|
|
from helper_m5_e2e_support import _create_project, _fail, _setup_db
|
|
from helpers_common import reset_global_state
|
|
|
|
from cleveragents.application.services.context_phase_analysis import (
|
|
analyze_phase_summaries,
|
|
)
|
|
from cleveragents.application.services.context_tiers import ContextTierService
|
|
from cleveragents.application.services.execute_phase_context_assembler import (
|
|
ACMSExecutePhaseContextAssembler,
|
|
)
|
|
from cleveragents.application.services.llm_actors import LLMExecuteActor
|
|
from cleveragents.application.services.plan_executor import StrategyDecision
|
|
from cleveragents.cli.commands.project_context import _read_policy, _write_policy
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
|
|
from cleveragents.domain.models.core.context_policy import (
|
|
VALID_PHASES,
|
|
ContextView,
|
|
ProjectContextPolicy,
|
|
)
|
|
from cleveragents.domain.models.core.project import ContextConfig, NamespacedProject
|
|
|
|
|
|
def context_tier_config() -> None:
|
|
"""Verify ContextConfig with hot/warm/cold tier management."""
|
|
reset_global_state()
|
|
default_cfg = ContextConfig()
|
|
if default_cfg.hot_max_tokens is not None:
|
|
_fail(f"default hot_max_tokens should be None: {default_cfg.hot_max_tokens}")
|
|
if default_cfg.warm_max_decisions is not None:
|
|
_fail(
|
|
f"default warm_max_decisions should be None: "
|
|
f"{default_cfg.warm_max_decisions}"
|
|
)
|
|
if default_cfg.cold_max_decisions is not None:
|
|
_fail(
|
|
f"default cold_max_decisions should be None: "
|
|
f"{default_cfg.cold_max_decisions}"
|
|
)
|
|
|
|
tier_cfg = ContextConfig(
|
|
hot_max_tokens=128_000,
|
|
warm_max_decisions=50,
|
|
cold_max_decisions=200,
|
|
max_file_size=2_000_000,
|
|
max_total_size=100_000_000,
|
|
)
|
|
if tier_cfg.hot_max_tokens != 128_000:
|
|
_fail(f"hot_max_tokens mismatch: {tier_cfg.hot_max_tokens}")
|
|
if tier_cfg.warm_max_decisions != 50:
|
|
_fail(f"warm_max_decisions mismatch: {tier_cfg.warm_max_decisions}")
|
|
if tier_cfg.cold_max_decisions != 200:
|
|
_fail(f"cold_max_decisions mismatch: {tier_cfg.cold_max_decisions}")
|
|
if tier_cfg.max_file_size != 2_000_000:
|
|
_fail(f"max_file_size mismatch: {tier_cfg.max_file_size}")
|
|
if tier_cfg.max_total_size != 100_000_000:
|
|
_fail(f"max_total_size mismatch: {tier_cfg.max_total_size}")
|
|
|
|
proj = NamespacedProject(
|
|
name="tier-test",
|
|
namespace="local",
|
|
context_config=tier_cfg,
|
|
)
|
|
if proj.context_config.hot_max_tokens != 128_000:
|
|
_fail("project did not preserve hot_max_tokens")
|
|
if proj.context_config.warm_max_decisions != 50:
|
|
_fail("project did not preserve warm_max_decisions")
|
|
if proj.context_config.cold_max_decisions != 200:
|
|
_fail("project did not preserve cold_max_decisions")
|
|
|
|
print("m5-context-tier-config-ok")
|
|
|
|
|
|
def context_policy_set_show() -> None:
|
|
"""Set a context policy via SQL helpers and read it back."""
|
|
reset_global_state()
|
|
proj_repo, _link_repo, _svc, sf = _setup_db()
|
|
_create_project(proj_repo, "local/policy-test")
|
|
|
|
policy = ProjectContextPolicy(
|
|
default_view=ContextView(
|
|
include_resources=["local/repo-*"],
|
|
exclude_paths=["*.log", "*.tmp"],
|
|
max_file_size=1_048_576,
|
|
max_total_size=52_428_800,
|
|
),
|
|
strategize_view=ContextView(
|
|
include_resources=["local/core-*"],
|
|
include_paths=["src/**/*.py"],
|
|
max_file_size=500_000,
|
|
),
|
|
)
|
|
|
|
_write_policy(sf, "local/policy-test", policy)
|
|
loaded = _read_policy(sf, "local/policy-test")
|
|
|
|
if loaded.default_view.include_resources != ["local/repo-*"]:
|
|
_fail(
|
|
f"default include_resources mismatch: "
|
|
f"{loaded.default_view.include_resources}"
|
|
)
|
|
if loaded.default_view.exclude_paths != ["*.log", "*.tmp"]:
|
|
_fail(f"default exclude_paths mismatch: {loaded.default_view.exclude_paths}")
|
|
if loaded.default_view.max_file_size != 1_048_576:
|
|
_fail(f"default max_file_size mismatch: {loaded.default_view.max_file_size}")
|
|
if loaded.strategize_view is None:
|
|
_fail("strategize_view is None after round-trip")
|
|
if loaded.strategize_view.include_resources != ["local/core-*"]:
|
|
_fail(
|
|
f"strategize include_resources mismatch: "
|
|
f"{loaded.strategize_view.include_resources}"
|
|
)
|
|
if loaded.strategize_view.include_paths != ["src/**/*.py"]:
|
|
_fail(
|
|
f"strategize include_paths mismatch: {loaded.strategize_view.include_paths}"
|
|
)
|
|
if loaded.execute_view is not None:
|
|
_fail("execute_view should be None")
|
|
if loaded.apply_view is not None:
|
|
_fail("apply_view should be None")
|
|
|
|
print("m5-context-policy-set-show-ok")
|
|
|
|
|
|
def acms_phase_inheritance() -> None:
|
|
"""Verify ACMS view inheritance: default -> strategize -> execute -> apply."""
|
|
reset_global_state()
|
|
policy = ProjectContextPolicy(
|
|
default_view=ContextView(
|
|
include_resources=["*"],
|
|
max_total_size=50_000_000,
|
|
),
|
|
strategize_view=ContextView(
|
|
include_resources=["core-*"],
|
|
max_file_size=1_000_000,
|
|
),
|
|
)
|
|
|
|
rv_default = policy.resolve_view("default")
|
|
if rv_default.include_resources != ["*"]:
|
|
_fail(f"default resolution wrong: {rv_default.include_resources}")
|
|
if rv_default.max_total_size != 50_000_000:
|
|
_fail(f"default max_total_size wrong: {rv_default.max_total_size}")
|
|
|
|
rv_strat = policy.resolve_view("strategize")
|
|
if rv_strat.include_resources != ["core-*"]:
|
|
_fail(f"strategize resolution wrong: {rv_strat.include_resources}")
|
|
if rv_strat.max_file_size != 1_000_000:
|
|
_fail(f"strategize max_file_size wrong: {rv_strat.max_file_size}")
|
|
|
|
rv_exec = policy.resolve_view("execute")
|
|
if rv_exec.include_resources != ["core-*"]:
|
|
_fail(f"execute should inherit strategize: {rv_exec.include_resources}")
|
|
|
|
rv_apply = policy.resolve_view("apply")
|
|
if rv_apply.include_resources != ["core-*"]:
|
|
_fail(f"apply should inherit via chain: {rv_apply.include_resources}")
|
|
|
|
policy2 = ProjectContextPolicy(
|
|
default_view=ContextView(include_resources=["*"]),
|
|
strategize_view=ContextView(include_resources=["core-*"]),
|
|
execute_view=ContextView(
|
|
include_resources=["exec-*"],
|
|
exclude_paths=["tests/**"],
|
|
),
|
|
)
|
|
rv_apply2 = policy2.resolve_view("apply")
|
|
if rv_apply2.include_resources != ["exec-*"]:
|
|
_fail(f"apply should inherit from execute: {rv_apply2.include_resources}")
|
|
if rv_apply2.exclude_paths != ["tests/**"]:
|
|
_fail(f"apply exclude_paths wrong: {rv_apply2.exclude_paths}")
|
|
|
|
print("m5-acms-phase-inheritance-ok")
|
|
|
|
|
|
def acms_scoped_context() -> None:
|
|
"""Verify ACMS produces scoped context output per phase."""
|
|
reset_global_state()
|
|
policy = ProjectContextPolicy(
|
|
default_view=ContextView(
|
|
include_resources=["*"],
|
|
max_file_size=2_000_000,
|
|
max_total_size=100_000_000,
|
|
),
|
|
strategize_view=ContextView(
|
|
include_resources=["local/large-repo"],
|
|
include_paths=["src/**/*.py", "docs/**/*.md"],
|
|
exclude_paths=["**/test_*.py"],
|
|
max_file_size=1_000_000,
|
|
),
|
|
execute_view=ContextView(
|
|
include_resources=["local/large-repo"],
|
|
include_paths=["src/**/*.py"],
|
|
exclude_paths=["**/test_*.py", "**/conftest.py"],
|
|
max_file_size=500_000,
|
|
max_total_size=1_200_000,
|
|
),
|
|
apply_view=ContextView(
|
|
include_resources=["local/large-repo"],
|
|
include_paths=["src/**/*.py"],
|
|
exclude_paths=["docs/**", "**/test_*.py", "**/conftest.py"],
|
|
max_file_size=250_000,
|
|
max_total_size=700_000,
|
|
),
|
|
)
|
|
|
|
for phase in ("default", "strategize", "execute", "apply"):
|
|
rv = policy.resolve_view(phase)
|
|
expected_attr = getattr(policy, f"{phase}_view")
|
|
if expected_attr is None:
|
|
_fail(f"phase {phase} view should not be None")
|
|
if rv.include_resources != expected_attr.include_resources:
|
|
_fail(
|
|
f"{phase} include_resources mismatch: "
|
|
f"{rv.include_resources} != {expected_attr.include_resources}"
|
|
)
|
|
|
|
strat_view = policy.resolve_view("strategize")
|
|
exec_view = policy.resolve_view("execute")
|
|
apply_view = policy.resolve_view("apply")
|
|
|
|
if strat_view.max_file_size is None:
|
|
_fail("strategize max_file_size is None")
|
|
if exec_view.max_file_size is None:
|
|
_fail("execute max_file_size is None")
|
|
if apply_view.max_file_size is None:
|
|
_fail("apply max_file_size is None")
|
|
if not (
|
|
strat_view.max_file_size > exec_view.max_file_size > apply_view.max_file_size
|
|
):
|
|
_fail(
|
|
f"size limits should narrow: "
|
|
f"{strat_view.max_file_size} > {exec_view.max_file_size} > "
|
|
f"{apply_view.max_file_size}"
|
|
)
|
|
|
|
sample_fragments = [
|
|
TieredFragment(
|
|
fragment_id="f-core",
|
|
content="x" * 900,
|
|
tier=ContextTier.HOT,
|
|
resource_id="local/large-repo",
|
|
project_name="local/scoped-test",
|
|
token_count=900,
|
|
metadata={"path": "src/core.py", "byte_size": 900_000},
|
|
),
|
|
TieredFragment(
|
|
fragment_id="f-engine",
|
|
content="x" * 500,
|
|
tier=ContextTier.HOT,
|
|
resource_id="local/large-repo",
|
|
project_name="local/scoped-test",
|
|
token_count=500,
|
|
metadata={"path": "src/engine.py", "byte_size": 450_000},
|
|
),
|
|
TieredFragment(
|
|
fragment_id="f-test",
|
|
content="x" * 120,
|
|
tier=ContextTier.WARM,
|
|
resource_id="local/large-repo",
|
|
project_name="local/scoped-test",
|
|
token_count=120,
|
|
metadata={"path": "src/test_core.py", "byte_size": 100_000},
|
|
),
|
|
TieredFragment(
|
|
fragment_id="f-doc",
|
|
content="x" * 180,
|
|
tier=ContextTier.WARM,
|
|
resource_id="local/large-repo",
|
|
project_name="local/scoped-test",
|
|
token_count=180,
|
|
metadata={"path": "docs/design.md", "byte_size": 200_000},
|
|
),
|
|
TieredFragment(
|
|
fragment_id="f-conftest",
|
|
content="x" * 90,
|
|
tier=ContextTier.COLD,
|
|
resource_id="local/large-repo",
|
|
project_name="local/scoped-test",
|
|
token_count=90,
|
|
metadata={"path": "src/conftest.py", "byte_size": 80_000},
|
|
),
|
|
TieredFragment(
|
|
fragment_id="f-huge",
|
|
content="x" * 1500,
|
|
tier=ContextTier.COLD,
|
|
resource_id="local/large-repo",
|
|
project_name="local/scoped-test",
|
|
token_count=1500,
|
|
metadata={"path": "src/huge.py", "byte_size": 1_500_000},
|
|
),
|
|
]
|
|
analysis = analyze_phase_summaries(policy, sample_fragments, budget_tokens=2_000)
|
|
|
|
for phase in ("strategize", "execute", "apply"):
|
|
phase_row = analysis["phases"][phase]
|
|
if not phase_row["summary"]:
|
|
_fail(f"{phase} summary should be non-empty")
|
|
if "resource_count" not in phase_row or "total_bytes" not in phase_row:
|
|
_fail(f"{phase} summary missing resource/size metrics")
|
|
|
|
narrowing = analysis["narrowing"]
|
|
if not narrowing["overall"]:
|
|
_fail(f"narrowing should hold across phases: {narrowing}")
|
|
proj_repo, _link_repo, _svc, sf = _setup_db()
|
|
_create_project(proj_repo, "local/scoped-test")
|
|
_write_policy(sf, "local/scoped-test", policy)
|
|
loaded = _read_policy(sf, "local/scoped-test")
|
|
|
|
for phase in ("default", "strategize", "execute", "apply"):
|
|
rv_loaded = loaded.resolve_view(phase)
|
|
rv_orig = policy.resolve_view(phase)
|
|
if rv_loaded.include_resources != rv_orig.include_resources:
|
|
_fail(f"{phase} round-trip include_resources mismatch")
|
|
if rv_loaded.max_file_size != rv_orig.max_file_size:
|
|
_fail(f"{phase} round-trip max_file_size mismatch")
|
|
|
|
print("m5-acms-scoped-context-ok")
|
|
|
|
|
|
def context_policy_clear() -> None:
|
|
"""Verify clearing a phase view causes inheritance fallback."""
|
|
reset_global_state()
|
|
policy = ProjectContextPolicy(
|
|
default_view=ContextView(include_resources=["all-*"]),
|
|
strategize_view=ContextView(include_resources=["strat-*"]),
|
|
execute_view=ContextView(include_resources=["exec-*"]),
|
|
apply_view=ContextView(include_resources=["apply-*"]),
|
|
)
|
|
|
|
rv = policy.resolve_view("execute")
|
|
if rv.include_resources != ["exec-*"]:
|
|
_fail(f"before clear, execute wrong: {rv.include_resources}")
|
|
|
|
cleared = policy.model_copy(update={"execute_view": None})
|
|
rv_after = cleared.resolve_view("execute")
|
|
if rv_after.include_resources != ["strat-*"]:
|
|
_fail(
|
|
f"after clear, execute should inherit strategize: "
|
|
f"{rv_after.include_resources}"
|
|
)
|
|
|
|
rv_apply = cleared.resolve_view("apply")
|
|
if rv_apply.include_resources != ["apply-*"]:
|
|
_fail(f"apply should still be its own: {rv_apply.include_resources}")
|
|
|
|
cleared2 = cleared.model_copy(update={"apply_view": None})
|
|
rv_apply2 = cleared2.resolve_view("apply")
|
|
if rv_apply2.include_resources != ["strat-*"]:
|
|
_fail(
|
|
f"after clearing apply+execute, apply should inherit strategize: "
|
|
f"{rv_apply2.include_resources}"
|
|
)
|
|
|
|
proj_repo, _link_repo, _svc, sf = _setup_db()
|
|
_create_project(proj_repo, "local/clear-test")
|
|
_write_policy(sf, "local/clear-test", cleared2)
|
|
loaded = _read_policy(sf, "local/clear-test")
|
|
|
|
if loaded.execute_view is not None:
|
|
_fail("execute_view should be None after round-trip")
|
|
if loaded.apply_view is not None:
|
|
_fail("apply_view should be None after round-trip")
|
|
|
|
rv_loaded = loaded.resolve_view("execute")
|
|
if rv_loaded.include_resources != ["strat-*"]:
|
|
_fail(f"round-trip execute fallback wrong: {rv_loaded.include_resources}")
|
|
|
|
print("m5-context-policy-clear-ok")
|
|
|
|
|
|
def context_view_validation() -> None:
|
|
"""Verify ContextView validation rules."""
|
|
reset_global_state()
|
|
cv = ContextView(max_file_size=1, max_total_size=1)
|
|
if cv.max_file_size != 1:
|
|
_fail(f"max_file_size should be 1: {cv.max_file_size}")
|
|
|
|
try:
|
|
ContextView(max_file_size=0)
|
|
_fail("max_file_size=0 should be rejected")
|
|
except ValueError:
|
|
pass
|
|
|
|
try:
|
|
ContextView(max_total_size=-1)
|
|
_fail("max_total_size=-1 should be rejected")
|
|
except ValueError:
|
|
pass
|
|
|
|
cv_none = ContextView(max_file_size=None, max_total_size=None)
|
|
if cv_none.max_file_size is not None:
|
|
_fail("max_file_size should be None")
|
|
|
|
expected_phases = {"default", "strategize", "execute", "apply"}
|
|
if expected_phases != VALID_PHASES:
|
|
_fail(f"VALID_PHASES mismatch: {VALID_PHASES} != {expected_phases}")
|
|
|
|
policy = ProjectContextPolicy()
|
|
try:
|
|
policy.resolve_view("invalid-phase")
|
|
_fail("resolve_view should reject invalid phase")
|
|
except ValueError:
|
|
pass
|
|
|
|
print("m5-context-view-validation-ok")
|
|
|
|
|
|
def llm_execute_acms_context() -> None:
|
|
"""Verify execute-phase LLM prompt uses ACMS assembled execute context."""
|
|
reset_global_state()
|
|
proj_repo, _link_repo, _svc, sf = _setup_db()
|
|
_create_project(proj_repo, "local/exec-ctx")
|
|
|
|
policy = ProjectContextPolicy(
|
|
default_view=ContextView(include_paths=["**/*"]),
|
|
strategize_view=ContextView(include_paths=["docs/**"]),
|
|
execute_view=ContextView(
|
|
include_paths=["src/**"],
|
|
max_file_size=100_000,
|
|
max_total_size=100_000,
|
|
),
|
|
)
|
|
_write_policy(sf, "local/exec-ctx", policy)
|
|
|
|
tier_service = ContextTierService(settings=Settings())
|
|
tier_service.store(
|
|
TieredFragment(
|
|
fragment_id="frag-docs",
|
|
content="documentation fragment",
|
|
tier=ContextTier.HOT,
|
|
resource_id="local/large-repo",
|
|
project_name="local/exec-ctx",
|
|
token_count=20,
|
|
metadata={"path": "docs/readme.md", "relevance_score": "0.6"},
|
|
)
|
|
)
|
|
tier_service.store(
|
|
TieredFragment(
|
|
fragment_id="frag-src",
|
|
content="def run():\n return 'ok'",
|
|
tier=ContextTier.HOT,
|
|
resource_id="local/large-repo",
|
|
project_name="local/exec-ctx",
|
|
token_count=16,
|
|
metadata={"path": "src/main.py", "relevance_score": "0.9"},
|
|
)
|
|
)
|
|
|
|
assembler = ACMSExecutePhaseContextAssembler(
|
|
context_tier_service=tier_service,
|
|
project_repository=proj_repo,
|
|
hot_max_tokens=1024,
|
|
)
|
|
|
|
mock_llm = MagicMock()
|
|
mock_llm.invoke.return_value = SimpleNamespace(
|
|
content="FILE: src/out.py\n```python\nprint('ok')\n```\n"
|
|
)
|
|
registry = cast(Any, SimpleNamespace(create_llm=MagicMock(return_value=mock_llm)))
|
|
plan = SimpleNamespace(
|
|
identity=SimpleNamespace(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
action_name="local/action",
|
|
project_links=[SimpleNamespace(project_name="local/exec-ctx")],
|
|
)
|
|
action = SimpleNamespace(execution_actor="openai/gpt-4")
|
|
lifecycle = SimpleNamespace(
|
|
get_plan=MagicMock(return_value=plan),
|
|
get_action=MagicMock(return_value=action),
|
|
resolve_actor_provider_model=MagicMock(return_value=None),
|
|
resolve_actor_options=MagicMock(return_value=None),
|
|
)
|
|
|
|
actor = LLMExecuteActor(
|
|
provider_registry=registry,
|
|
lifecycle_service=lifecycle,
|
|
context_assembler=assembler,
|
|
)
|
|
actor.execute(
|
|
plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
|
|
decisions=[
|
|
StrategyDecision(
|
|
decision_id="d1",
|
|
step_text="Implement feature",
|
|
sequence=0,
|
|
parent_id=None,
|
|
)
|
|
],
|
|
stream_callback=None,
|
|
)
|
|
|
|
call_args = mock_llm.invoke.call_args
|
|
if call_args is None:
|
|
_fail("LLM invoke was not called")
|
|
prompt = call_args.args[0][0].content
|
|
if "ACMS Execute-Phase Context" not in prompt:
|
|
_fail("prompt missing ACMS execute-phase context section")
|
|
if "src/main.py" not in prompt:
|
|
_fail("prompt missing execute-view source fragment")
|
|
if "docs/readme.md" in prompt:
|
|
_fail("prompt should exclude strategize-only docs fragment in execute view")
|
|
print("m5-llm-execute-acms-context-ok")
|