forked from HAL9000/cleveragents-core
aditya
f0442e835d
Integrate ACMS execute-phase context assembly into LLMExecuteActor and inject assembled context into execute prompts with resilient fallback when assembly fails or returns empty output. Wire the plan CLI executor to use an ACMS-backed execute context assembler, add Behave coverage for context injection/fallback/empty context, and extend Robot M5 verification helpers to assert execute-phase ACMS context usage. ISSUES CLOSED: #850
781 lines
27 KiB
Python
781 lines
27 KiB
Python
"""Step definitions for llm_actors_coverage.feature.
|
|
|
|
These steps exercise uncovered code paths in llm_actors.py including:
|
|
- _parse_actor_name helper (lines 45-50)
|
|
- LLMStrategizeActor.__init__ validation (lines 62-73)
|
|
- LLMStrategizeActor.execute full flow (lines 75-182)
|
|
- LLMStrategizeActor._parse_decisions (lines 184-204)
|
|
- LLMExecuteActor.__init__ validation (lines 215-226)
|
|
- LLMExecuteActor.execute full flow (lines 228-333)
|
|
- LLMExecuteActor._parse_file_blocks (lines 335-356)
|
|
- LLMExecuteActor._write_to_sandbox (lines 358-384)
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.application.services.llm_actors import (
|
|
LLMExecuteActor,
|
|
LLMStrategizeActor,
|
|
_parse_actor_name,
|
|
)
|
|
from cleveragents.application.services.plan_executor import (
|
|
ExecuteResult,
|
|
StrategizeResult,
|
|
StrategyDecision,
|
|
)
|
|
from cleveragents.core.exceptions import ValidationError
|
|
from cleveragents.domain.models.acms.crp import (
|
|
AssembledContext,
|
|
ContextFragment,
|
|
FragmentProvenance,
|
|
)
|
|
from cleveragents.domain.models.core.plan import InvariantSource, PlanInvariant
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_mock_plan(action_name="test-action"):
|
|
"""Create a mock plan with the given action_name."""
|
|
plan = SimpleNamespace(action_name=action_name)
|
|
return plan
|
|
|
|
|
|
def _make_mock_action(strategy_actor=None, execution_actor=None):
|
|
"""Create a mock action with optional actor names."""
|
|
action = SimpleNamespace(
|
|
strategy_actor=strategy_actor,
|
|
execution_actor=execution_actor,
|
|
)
|
|
return action
|
|
|
|
|
|
def _make_mock_llm_response(content):
|
|
"""Create a mock LLM response with a .content attribute."""
|
|
response = SimpleNamespace(content=content)
|
|
return response
|
|
|
|
|
|
def _make_mock_llm_response_no_content(fallback_str):
|
|
"""Create a mock LLM response without .content attribute."""
|
|
|
|
class NoContentResponse:
|
|
def __str__(self):
|
|
return fallback_str
|
|
|
|
return NoContentResponse()
|
|
|
|
|
|
def _make_mock_registry(llm_response):
|
|
"""Create a mock ProviderRegistry that returns a mock LLM."""
|
|
mock_llm = MagicMock()
|
|
mock_llm.invoke.return_value = llm_response
|
|
registry = MagicMock()
|
|
registry.create_llm.return_value = mock_llm
|
|
return registry
|
|
|
|
|
|
def _make_mock_lifecycle(strategy_actor=None, execution_actor=None):
|
|
"""Create a mock lifecycle service."""
|
|
plan = _make_mock_plan()
|
|
action = _make_mock_action(
|
|
strategy_actor=strategy_actor,
|
|
execution_actor=execution_actor,
|
|
)
|
|
lifecycle = SimpleNamespace(
|
|
get_plan=MagicMock(return_value=plan),
|
|
get_action=MagicMock(return_value=action),
|
|
)
|
|
return lifecycle
|
|
|
|
|
|
def _sample_decisions():
|
|
"""Return a list of sample StrategyDecision objects."""
|
|
return [
|
|
StrategyDecision(
|
|
decision_id="DEC001",
|
|
step_text="Create the main module",
|
|
sequence=0,
|
|
parent_id=None,
|
|
),
|
|
StrategyDecision(
|
|
decision_id="DEC002",
|
|
step_text="Add unit tests",
|
|
sequence=1,
|
|
parent_id="DEC001",
|
|
),
|
|
]
|
|
|
|
|
|
class _StubContextAssembler:
|
|
"""Simple execute-context assembler test double."""
|
|
|
|
def __init__(self, assembled: AssembledContext | None = None, fail: bool = False):
|
|
self._assembled = assembled
|
|
self._fail = fail
|
|
|
|
def assemble(self, plan):
|
|
if self._fail:
|
|
raise RuntimeError("context assembly failed")
|
|
return self._assembled
|
|
|
|
|
|
LLM_NUMBERED_RESPONSE = "1. Create the module\n2. Write unit tests\n3. Update docs"
|
|
|
|
LLM_FILE_BLOCKS_RESPONSE = (
|
|
"FILE: src/main.py\n```python\nprint('hello')\n```\n\n"
|
|
"FILE: tests/test_main.py\n```python\ndef test_main(): pass\n```\n"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _parse_actor_name
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I parse actor name "{name}"')
|
|
def step_parse_actor_name(context, name):
|
|
context.parsed_provider, context.parsed_model = _parse_actor_name(name)
|
|
|
|
|
|
@when('I parse actor name ""')
|
|
def step_parse_actor_name_empty(context):
|
|
context.parsed_provider, context.parsed_model = _parse_actor_name("")
|
|
|
|
|
|
@then('the parsed provider should be "{expected}"')
|
|
def step_check_parsed_provider(context, expected):
|
|
assert context.parsed_provider == expected, (
|
|
f"Expected provider '{expected}', got '{context.parsed_provider}'"
|
|
)
|
|
|
|
|
|
@then('the parsed model should be "{expected}"')
|
|
def step_check_parsed_model(context, expected):
|
|
assert context.parsed_model == expected, (
|
|
f"Expected model '{expected}', got '{context.parsed_model}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMStrategizeActor.__init__ validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create an LLMStrategizeActor with None provider_registry")
|
|
def step_create_strategize_none_registry(context):
|
|
try:
|
|
LLMStrategizeActor(provider_registry=None, lifecycle_service=MagicMock())
|
|
context.caught_error = None
|
|
except ValidationError as exc:
|
|
context.caught_error = exc
|
|
|
|
|
|
@when("I create an LLMStrategizeActor with None lifecycle_service")
|
|
def step_create_strategize_none_lifecycle(context):
|
|
try:
|
|
LLMStrategizeActor(provider_registry=MagicMock(), lifecycle_service=None)
|
|
context.caught_error = None
|
|
except ValidationError as exc:
|
|
context.caught_error = exc
|
|
|
|
|
|
@then('lacov a ValidationError should be raised with message "{msg}"')
|
|
def step_check_validation_error(context, msg):
|
|
assert context.caught_error is not None, (
|
|
"Expected a ValidationError but none was raised"
|
|
)
|
|
assert msg in str(context.caught_error), (
|
|
f"Expected message containing '{msg}', got '{context.caught_error}'"
|
|
)
|
|
|
|
|
|
@given("a mock provider registry for strategize")
|
|
def step_mock_registry_for_strategize(context):
|
|
context.mock_registry = MagicMock()
|
|
|
|
|
|
@given("a mock lifecycle service for strategize")
|
|
def step_mock_lifecycle_for_strategize(context):
|
|
context.mock_lifecycle = MagicMock()
|
|
|
|
|
|
@when("I create an LLMStrategizeActor with valid dependencies")
|
|
def step_create_valid_strategize_actor(context):
|
|
context.strategize_actor = LLMStrategizeActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
|
|
|
|
@then("the LLMStrategizeActor should be created successfully")
|
|
def step_verify_strategize_actor_created(context):
|
|
assert context.strategize_actor is not None
|
|
assert isinstance(context.strategize_actor, LLMStrategizeActor)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMStrategizeActor.execute
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a valid LLMStrategizeActor")
|
|
def step_setup_valid_strategize_actor(context):
|
|
llm_response = _make_mock_llm_response(LLM_NUMBERED_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(strategy_actor="openai/gpt-4")
|
|
context.strategize_actor = LLMStrategizeActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
|
|
|
|
@given("the LLM returns a numbered step list for strategize")
|
|
def step_llm_returns_numbered_list(context):
|
|
llm_response = _make_mock_llm_response(LLM_NUMBERED_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(strategy_actor="anthropic/claude-3")
|
|
context.strategize_actor = LLMStrategizeActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
|
|
|
|
@given("the LLM returns a response without content attribute")
|
|
def step_llm_returns_no_content(context):
|
|
response = _make_mock_llm_response_no_content("1. Fallback step text")
|
|
mock_llm = MagicMock()
|
|
mock_llm.invoke.return_value = response
|
|
context.mock_registry = MagicMock()
|
|
context.mock_registry.create_llm.return_value = mock_llm
|
|
context.mock_lifecycle = _make_mock_lifecycle(strategy_actor="openai/gpt-4")
|
|
context.strategize_actor = LLMStrategizeActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
|
|
|
|
@when("I call strategize execute with empty plan_id")
|
|
def step_strategize_execute_empty_plan_id(context):
|
|
try:
|
|
context.strategize_actor.execute(plan_id="", definition_of_done="some task")
|
|
context.caught_error = None
|
|
except ValidationError as exc:
|
|
context.caught_error = exc
|
|
|
|
|
|
@when('I call strategize execute with plan_id "{pid}" and a stream callback')
|
|
def step_strategize_execute_with_callback(context, pid):
|
|
context.stream_events = []
|
|
|
|
def callback(event_type, data):
|
|
context.stream_events.append({"type": event_type, "data": data})
|
|
|
|
context.strategize_result = context.strategize_actor.execute(
|
|
plan_id=pid,
|
|
definition_of_done="Build a REST API",
|
|
stream_callback=callback,
|
|
)
|
|
|
|
|
|
@when('I call strategize execute with plan_id "{pid}" and no stream callback')
|
|
def step_strategize_execute_no_callback(context, pid):
|
|
context.strategize_result = context.strategize_actor.execute(
|
|
plan_id=pid,
|
|
definition_of_done="Build a REST API",
|
|
stream_callback=None,
|
|
)
|
|
|
|
|
|
@when('I call strategize execute with plan_id "{pid}" and invariants')
|
|
def step_strategize_execute_with_invariants(context, pid):
|
|
invariants = [
|
|
PlanInvariant(text="Must use Python 3.12+", source=InvariantSource.PLAN),
|
|
PlanInvariant(text="No external dependencies", source=InvariantSource.ACTION),
|
|
]
|
|
context.strategize_result = context.strategize_actor.execute(
|
|
plan_id=pid,
|
|
definition_of_done="Build a module",
|
|
invariants=invariants,
|
|
stream_callback=None,
|
|
)
|
|
|
|
|
|
@when('I call strategize execute with plan_id "{pid}" and None definition_of_done')
|
|
def step_strategize_execute_none_dod(context, pid):
|
|
context.strategize_result = context.strategize_actor.execute(
|
|
plan_id=pid,
|
|
definition_of_done=None,
|
|
stream_callback=None,
|
|
)
|
|
|
|
|
|
@then("the strategize result should contain decisions")
|
|
def step_verify_strategize_decisions(context):
|
|
assert isinstance(context.strategize_result, StrategizeResult)
|
|
assert len(context.strategize_result.decisions) > 0
|
|
assert context.strategize_result.decision_root_id is not None
|
|
|
|
|
|
@then('the stream callback should have received "{event_type}"')
|
|
def step_verify_stream_callback_event(context, event_type):
|
|
event_types = [e["type"] for e in context.stream_events]
|
|
assert event_type in event_types, f"Expected event '{event_type}' in {event_types}"
|
|
|
|
|
|
@then("the strategize result should contain invariant records")
|
|
def step_verify_invariant_records(context):
|
|
assert isinstance(context.strategize_result, StrategizeResult)
|
|
assert len(context.strategize_result.invariant_records) == 2
|
|
assert (
|
|
context.strategize_result.invariant_records[0]["text"]
|
|
== "Must use Python 3.12+"
|
|
)
|
|
assert context.strategize_result.invariant_records[0]["source"] == "plan"
|
|
assert context.strategize_result.invariant_records[1]["source"] == "action"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMStrategizeActor._parse_decisions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I parse LLM decisions from numbered list "{text}"')
|
|
def step_parse_decisions_numbered(context, text):
|
|
raw = text.replace("\\n", "\n")
|
|
context.parsed_decisions = LLMStrategizeActor._parse_decisions(raw)
|
|
|
|
|
|
@when('I parse LLM decisions from bullet list "{text}"')
|
|
def step_parse_decisions_bullets(context, text):
|
|
raw = text.replace("\\n", "\n")
|
|
context.parsed_decisions = LLMStrategizeActor._parse_decisions(raw)
|
|
|
|
|
|
@when("I parse LLM decisions from empty string")
|
|
def step_parse_decisions_empty(context):
|
|
context.parsed_decisions = LLMStrategizeActor._parse_decisions("")
|
|
|
|
|
|
@when("I parse LLM decisions from text with blank lines")
|
|
def step_parse_decisions_blank_lines(context):
|
|
text = "1. Step one\n\n\n2. Step two\n\n"
|
|
context.parsed_decisions = LLMStrategizeActor._parse_decisions(text)
|
|
|
|
|
|
@when('I parse LLM decisions from "{text}"')
|
|
def step_parse_decisions_generic(context, text):
|
|
raw = text.replace("\\n", "\n")
|
|
context.parsed_decisions = LLMStrategizeActor._parse_decisions(raw)
|
|
|
|
|
|
@then("I should get {count:d} parsed decisions")
|
|
def step_verify_parsed_decision_count(context, count):
|
|
assert len(context.parsed_decisions) == count, (
|
|
f"Expected {count} decisions, got {len(context.parsed_decisions)}: "
|
|
f"{context.parsed_decisions}"
|
|
)
|
|
|
|
|
|
@then('parsed decision {idx:d} should be "{expected}"')
|
|
def step_verify_parsed_decision_text(context, idx, expected):
|
|
assert context.parsed_decisions[idx] == expected, (
|
|
f"Expected decision[{idx}]='{expected}', got '{context.parsed_decisions[idx]}'"
|
|
)
|
|
|
|
|
|
@then('I should get the default decision "{expected}"')
|
|
def step_verify_default_decision(context, expected):
|
|
assert len(context.parsed_decisions) == 1
|
|
assert context.parsed_decisions[0] == expected
|
|
|
|
|
|
@then("blank lines should be skipped in the result")
|
|
def step_verify_blank_lines_skipped(context):
|
|
assert len(context.parsed_decisions) == 2
|
|
assert context.parsed_decisions[0] == "Step one"
|
|
assert context.parsed_decisions[1] == "Step two"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMExecuteActor.__init__ validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I create an LLMExecuteActor with None provider_registry")
|
|
def step_create_execute_none_registry(context):
|
|
try:
|
|
LLMExecuteActor(provider_registry=None, lifecycle_service=MagicMock())
|
|
context.caught_error = None
|
|
except ValidationError as exc:
|
|
context.caught_error = exc
|
|
|
|
|
|
@when("I create an LLMExecuteActor with None lifecycle_service")
|
|
def step_create_execute_none_lifecycle(context):
|
|
try:
|
|
LLMExecuteActor(provider_registry=MagicMock(), lifecycle_service=None)
|
|
context.caught_error = None
|
|
except ValidationError as exc:
|
|
context.caught_error = exc
|
|
|
|
|
|
@given("a mock provider registry for execute")
|
|
def step_mock_registry_for_execute(context):
|
|
context.mock_registry = MagicMock()
|
|
|
|
|
|
@given("a mock lifecycle service for execute")
|
|
def step_mock_lifecycle_for_execute(context):
|
|
context.mock_lifecycle = MagicMock()
|
|
|
|
|
|
@when("I create an LLMExecuteActor with valid dependencies")
|
|
def step_create_valid_execute_actor(context):
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
|
|
|
|
@then("the LLMExecuteActor should be created successfully")
|
|
def step_verify_execute_actor_created(context):
|
|
assert context.execute_actor is not None
|
|
assert isinstance(context.execute_actor, LLMExecuteActor)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMExecuteActor.execute
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a valid LLMExecuteActor")
|
|
def step_setup_valid_execute_actor(context):
|
|
llm_response = _make_mock_llm_response(LLM_FILE_BLOCKS_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(execution_actor="openai/gpt-4")
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
context.sandbox_dir = None
|
|
|
|
|
|
@given("a valid LLMExecuteActor with assembled execute-phase context")
|
|
def step_setup_execute_actor_with_context(context):
|
|
llm_response = _make_mock_llm_response(LLM_FILE_BLOCKS_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(execution_actor="openai/gpt-4")
|
|
assembled = AssembledContext(
|
|
fragments=(
|
|
ContextFragment(
|
|
uko_node="resource://src/main.py",
|
|
content="def run():\n return 'ok'",
|
|
detail_depth=2,
|
|
token_count=12,
|
|
relevance_score=0.9,
|
|
provenance=FragmentProvenance(
|
|
resource_uri="local/large-repo",
|
|
location="src/main.py",
|
|
strategy="execute_phase_context",
|
|
),
|
|
),
|
|
),
|
|
total_tokens=12,
|
|
budget_used=0.1,
|
|
strategies_used=("relevance",),
|
|
context_hash="ctxhash123",
|
|
preamble="execute phase context",
|
|
provenance_map={},
|
|
)
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
context_assembler=_StubContextAssembler(assembled=assembled),
|
|
)
|
|
|
|
|
|
@given("a valid LLMExecuteActor with failing context assembly")
|
|
def step_setup_execute_actor_with_failing_context(context):
|
|
llm_response = _make_mock_llm_response(LLM_FILE_BLOCKS_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(execution_actor="openai/gpt-4")
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
context_assembler=_StubContextAssembler(fail=True),
|
|
)
|
|
|
|
|
|
@given("a valid LLMExecuteActor with empty assembled context")
|
|
def step_setup_execute_actor_with_empty_context(context):
|
|
llm_response = _make_mock_llm_response(LLM_FILE_BLOCKS_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(execution_actor="openai/gpt-4")
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
context_assembler=_StubContextAssembler(assembled=None),
|
|
)
|
|
|
|
|
|
@given("the LLM returns file blocks for execute")
|
|
def step_llm_returns_file_blocks(context):
|
|
llm_response = _make_mock_llm_response(LLM_FILE_BLOCKS_RESPONSE)
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(execution_actor="anthropic/claude-3")
|
|
context_assembler = None
|
|
if hasattr(context, "execute_actor") and context.execute_actor is not None:
|
|
context_assembler = getattr(context.execute_actor, "_context_assembler", None)
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
context_assembler=context_assembler,
|
|
)
|
|
|
|
|
|
@given("the LLM returns empty response for execute")
|
|
def step_llm_returns_empty_for_execute(context):
|
|
llm_response = _make_mock_llm_response("No files to generate.")
|
|
context.mock_registry = _make_mock_registry(llm_response)
|
|
context.mock_lifecycle = _make_mock_lifecycle(execution_actor="openai/gpt-4")
|
|
context.execute_actor = LLMExecuteActor(
|
|
provider_registry=context.mock_registry,
|
|
lifecycle_service=context.mock_lifecycle,
|
|
)
|
|
|
|
|
|
@when("I call execute actor with empty plan_id")
|
|
def step_execute_actor_empty_plan_id(context):
|
|
try:
|
|
context.execute_actor.execute(plan_id="", decisions=_sample_decisions())
|
|
context.caught_error = None
|
|
except ValidationError as exc:
|
|
context.caught_error = exc
|
|
|
|
|
|
@when('I call execute actor with plan_id "{pid}" and a stream callback')
|
|
def step_execute_actor_with_callback(context, pid):
|
|
context.exec_stream_events = []
|
|
|
|
def callback(event_type, data):
|
|
context.exec_stream_events.append({"type": event_type, "data": data})
|
|
|
|
context.execute_result = context.execute_actor.execute(
|
|
plan_id=pid,
|
|
decisions=_sample_decisions(),
|
|
stream_callback=callback,
|
|
)
|
|
|
|
|
|
@when('I call execute actor with plan_id "{pid}" and no stream callback')
|
|
def step_execute_actor_no_callback(context, pid):
|
|
context.execute_result = context.execute_actor.execute(
|
|
plan_id=pid,
|
|
decisions=_sample_decisions(),
|
|
stream_callback=None,
|
|
)
|
|
llm = context.mock_registry.create_llm.return_value
|
|
call_args = llm.invoke.call_args
|
|
assert call_args is not None
|
|
context.execute_prompt_text = call_args.args[0][0].content
|
|
|
|
|
|
@when('I call execute actor with plan_id "{pid}" and a sandbox root')
|
|
def step_execute_actor_with_sandbox(context, pid):
|
|
context.sandbox_dir = tempfile.mkdtemp(prefix="llm_actors_test_")
|
|
|
|
def cleanup():
|
|
if context.sandbox_dir and os.path.exists(context.sandbox_dir):
|
|
shutil.rmtree(context.sandbox_dir)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
context.execute_result = context.execute_actor.execute(
|
|
plan_id=pid,
|
|
decisions=_sample_decisions(),
|
|
sandbox_root=context.sandbox_dir,
|
|
stream_callback=None,
|
|
)
|
|
|
|
|
|
@when('I call execute actor with plan_id "{pid}" sandbox root and read_only')
|
|
def step_execute_actor_sandbox_readonly(context, pid):
|
|
context.sandbox_dir = tempfile.mkdtemp(prefix="llm_actors_test_ro_")
|
|
|
|
def cleanup():
|
|
if context.sandbox_dir and os.path.exists(context.sandbox_dir):
|
|
shutil.rmtree(context.sandbox_dir)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
context.execute_result = context.execute_actor.execute(
|
|
plan_id=pid,
|
|
decisions=_sample_decisions(),
|
|
sandbox_root=context.sandbox_dir,
|
|
read_only=True,
|
|
stream_callback=None,
|
|
)
|
|
|
|
|
|
@then("the execute result should contain a changeset")
|
|
def step_verify_execute_changeset(context):
|
|
assert isinstance(context.execute_result, ExecuteResult)
|
|
assert context.execute_result.changeset is not None
|
|
assert context.execute_result.changeset_id is not None
|
|
|
|
|
|
@then('the execute stream callback should have received "{event_type}"')
|
|
def step_verify_exec_stream_event(context, event_type):
|
|
event_types = [e["type"] for e in context.exec_stream_events]
|
|
assert event_type in event_types, f"Expected event '{event_type}' in {event_types}"
|
|
|
|
|
|
@then("the execute result should have sandbox refs")
|
|
def step_verify_sandbox_refs(context):
|
|
assert isinstance(context.execute_result, ExecuteResult)
|
|
assert len(context.execute_result.sandbox_refs) > 0
|
|
|
|
|
|
@then("no files should be written to sandbox")
|
|
def step_verify_no_sandbox_writes(context):
|
|
# In read_only mode, _write_to_sandbox is not called, so the sandbox
|
|
# directory should be empty (only the dir itself exists, no generated files).
|
|
contents = os.listdir(context.sandbox_dir)
|
|
assert len(contents) == 0, (
|
|
f"Expected empty sandbox in read_only mode, found: {contents}"
|
|
)
|
|
|
|
|
|
@then("the execute result should have zero entries")
|
|
def step_verify_zero_entries(context):
|
|
assert isinstance(context.execute_result, ExecuteResult)
|
|
assert context.execute_result.tool_calls_count == 0
|
|
|
|
|
|
@then('the execute prompt should contain "{needle}"')
|
|
def step_execute_prompt_contains(context, needle):
|
|
assert needle in context.execute_prompt_text, (
|
|
f"Expected prompt to contain '{needle}', got: {context.execute_prompt_text}"
|
|
)
|
|
|
|
|
|
@then('the execute prompt should not contain "{needle}"')
|
|
def step_execute_prompt_not_contains(context, needle):
|
|
assert needle not in context.execute_prompt_text, (
|
|
f"Expected prompt not to contain '{needle}', got: {context.execute_prompt_text}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMExecuteActor._parse_file_blocks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I parse file blocks from LLM output with two files")
|
|
def step_parse_file_blocks_two_files(context):
|
|
context.parsed_entries = LLMExecuteActor._parse_file_blocks(
|
|
LLM_FILE_BLOCKS_RESPONSE, "PLAN_TEST"
|
|
)
|
|
|
|
|
|
@when("I parse file blocks from empty LLM output")
|
|
def step_parse_file_blocks_empty(context):
|
|
context.parsed_entries = LLMExecuteActor._parse_file_blocks(
|
|
"No files here.", "PLAN_EMPTY"
|
|
)
|
|
|
|
|
|
@then("I should get {count:d} changeset entries")
|
|
def step_verify_changeset_entry_count(context, count):
|
|
assert len(context.parsed_entries) == count, (
|
|
f"Expected {count} entries, got {len(context.parsed_entries)}"
|
|
)
|
|
|
|
|
|
@then('changeset entry {idx:d} path should be "{expected}"')
|
|
def step_verify_changeset_entry_path(context, idx, expected):
|
|
assert context.parsed_entries[idx].path == expected, (
|
|
f"Expected path '{expected}', got '{context.parsed_entries[idx].path}'"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMExecuteActor._write_to_sandbox
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("lacov a temporary sandbox directory")
|
|
def step_create_temp_sandbox(context):
|
|
context.sandbox_dir = tempfile.mkdtemp(prefix="llm_sandbox_test_")
|
|
|
|
def cleanup():
|
|
if context.sandbox_dir and os.path.exists(context.sandbox_dir):
|
|
shutil.rmtree(context.sandbox_dir)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I write generated files to the sandbox")
|
|
def step_write_files_to_sandbox(context):
|
|
entries = LLMExecuteActor._parse_file_blocks(LLM_FILE_BLOCKS_RESPONSE, "PLAN_WR")
|
|
LLMExecuteActor._write_to_sandbox(
|
|
entries, context.sandbox_dir, LLM_FILE_BLOCKS_RESPONSE
|
|
)
|
|
|
|
|
|
@then("the sandbox should contain the generated files")
|
|
def step_verify_sandbox_files(context):
|
|
main_path = os.path.join(context.sandbox_dir, "src", "main.py")
|
|
test_path = os.path.join(context.sandbox_dir, "tests", "test_main.py")
|
|
assert os.path.exists(main_path), f"Expected file at {main_path}"
|
|
assert os.path.exists(test_path), f"Expected file at {test_path}"
|
|
with open(main_path) as f:
|
|
content = f.read()
|
|
assert "print('hello')" in content
|
|
|
|
|
|
@given("lacov a temporary sandbox directory that is read-only")
|
|
def step_create_readonly_sandbox(context):
|
|
context.sandbox_dir = tempfile.mkdtemp(prefix="llm_sandbox_ro_")
|
|
# Pre-create the subdirectory so makedirs succeeds, then make the
|
|
# target *file* location unwritable by making the leaf dir read-only.
|
|
context.readonly_leaf = os.path.join(context.sandbox_dir, "src")
|
|
os.makedirs(context.readonly_leaf, exist_ok=True)
|
|
# Make the leaf dir read-only so open() inside it fails with OSError
|
|
os.chmod(context.readonly_leaf, 0o555)
|
|
|
|
def cleanup():
|
|
if context.sandbox_dir and os.path.exists(context.sandbox_dir):
|
|
os.chmod(context.readonly_leaf, 0o755)
|
|
shutil.rmtree(context.sandbox_dir)
|
|
|
|
context.add_cleanup(cleanup)
|
|
|
|
|
|
@when("I write generated files to the read-only sandbox")
|
|
def step_write_to_readonly_sandbox(context):
|
|
# FILE path puts the file inside "src/" which is read-only
|
|
llm_output = "FILE: src/fail.py\n```python\nprint('fail')\n```\n"
|
|
entries = LLMExecuteActor._parse_file_blocks(llm_output, "PLAN_RO")
|
|
# Should not raise - OSError is caught and logged inside _write_to_sandbox
|
|
context.write_exception = None
|
|
try:
|
|
LLMExecuteActor._write_to_sandbox(entries, context.sandbox_dir, llm_output)
|
|
except Exception as exc:
|
|
context.write_exception = exc
|
|
|
|
|
|
@then("the write should not raise an exception")
|
|
def step_verify_no_write_exception(context):
|
|
assert context.write_exception is None, (
|
|
f"Expected no exception, got: {context.write_exception}"
|
|
)
|