149 lines
9.2 KiB
Python
149 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from agents.providers import DeterministicSolProvider
|
|
from agents.roadmap import RoadmapService
|
|
from agents.scenario_lab import ScenarioLabService
|
|
from control_plane.projects.models import Exploration, ExplorationOpportunity, Project, RoadmapHorizon, RoadmapItem, ScenarioFinding, StewardFinding, Task
|
|
from graph.bootstrap import champion_project_roadmap_review_graph_v1, champion_scenario_lab_graph_v1
|
|
from graph.models import GraphApprovalStatus, GraphRun, GraphRunStatus
|
|
from graph.roadmap import project_roadmap_review_graph_v1, roadmap_registry
|
|
from graph.scenario_lab import scenario_lab_graph_v1, scenario_lab_registry
|
|
from graph.langgraph_runtime import LangGraphRuntime
|
|
from model_router.router import ModelRouter
|
|
from tests.test_m2_autonomous_loop import create_disposable_django_repo
|
|
|
|
|
|
def project(tmp_path: Path) -> Project:
|
|
return Project.objects.create(name="Roadmap Scenario", goal="Existing project", repository_path=str(create_disposable_django_repo(tmp_path)))
|
|
|
|
|
|
def sol_router(payload: dict[str, object]) -> ModelRouter:
|
|
return ModelRouter({"sol": DeterministicSolProvider(json.dumps(payload))}, persist_requests=True)
|
|
|
|
|
|
def approve_and_resume(graph_run: GraphRun, registry) -> GraphRun:
|
|
approval = graph_run.approvals.get()
|
|
approval.status = GraphApprovalStatus.APPROVED
|
|
approval.decided_by = "tester"
|
|
approval.save(update_fields=["status", "decided_by", "updated_at"])
|
|
graph_run.status = GraphRunStatus.RUNNING
|
|
graph_run.failure_reason = ""
|
|
graph_run.save(update_fields=["status", "failure_reason", "updated_at"])
|
|
LangGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
|
|
graph_run.refresh_from_db()
|
|
return graph_run
|
|
|
|
|
|
def test_roadmap_upsert_scores_deduplicates_and_preserves_lineage(tmp_path: Path) -> None:
|
|
proj = project(tmp_path)
|
|
service = RoadmapService()
|
|
|
|
first = service.upsert_item(proj, title="Add scenario coverage", description="Cover graph failures", source="EXPLORE", source_ref={"opportunity": "1"}, evidence={"files": ["graph"]}, horizon="NEXT", category="RELIABILITY", target_action="EXTEND", scores={"value": 0.9, "effort": 0.3, "risk": 0.2, "confidence": 0.8, "strategic_fit": 0.9, "technical_fit": 0.8, "urgency": 0.7})
|
|
second = service.upsert_item(proj, title="Add scenario coverage", source="STEWARD", source_ref={"finding": "2"}, evidence={"severity": "HIGH"})
|
|
|
|
assert first.id == second.id
|
|
assert RoadmapItem.objects.filter(project=proj).count() == 1
|
|
assert second.metadata["occurrences"] == 2
|
|
assert second.source_ref["opportunity"] == "1"
|
|
assert second.source_ref["finding"] == "2"
|
|
assert second.composite_score > 0.5
|
|
|
|
|
|
def test_roadmap_gathers_sources_prioritizes_and_does_not_execute(tmp_path: Path) -> None:
|
|
proj = project(tmp_path)
|
|
exploration = Exploration.objects.create(project=proj, status="COMPLETE")
|
|
ExplorationOpportunity.objects.create(exploration=exploration, project=proj, title="Preserve absolute SQLite paths", description="SQLite absolute paths should remain absolute", opportunity_type="RELIABILITY", evidence={"source": "explore"}, recommended_action="REPAIR", value_score=0.8, effort_score=0.2, risk_score=0.3, confidence=0.8, strategic_fit=0.7, technical_fit=0.9, grouping_key="opp")
|
|
StewardFinding.objects.create(project=proj, finding_type="PERFORMANCE", title="Queue latency monitor", summary="Monitor queue depth > 20", evidence={"baseline_measurement": {"metric": "latency", "latency": 100}}, severity="MEDIUM", confidence=0.7, recommended_action="EVOLVE", grouping_key="steward")
|
|
service = RoadmapService(sol_router({"recommendations": []}))
|
|
|
|
version = champion_project_roadmap_review_graph_v1()
|
|
graph_run = GraphRun.objects.create(execution_graph_version=version, project=proj, current_node=version.graph_spec["entry"])
|
|
LangGraphRuntime(roadmap_registry(service)).run_until_terminal_or_paused(graph_run)
|
|
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert RoadmapItem.objects.filter(project=proj).count() == 2
|
|
assert Task.objects.filter(project=proj).count() == 0
|
|
view = service.project_roadmap_view(proj)
|
|
assert set(view) == {"NOW", "NEXT", "LATER", "EXPLORING"}
|
|
assert len(view["EXPLORING"]) == 2
|
|
|
|
|
|
def test_roadmap_project_brain_recommendations_and_conversion_lineage(tmp_path: Path) -> None:
|
|
proj = project(tmp_path)
|
|
item = RoadmapService().upsert_item(proj, title="Add project health dashboard", description="Expose lifecycle status", horizon="NEXT", category="FEATURE", target_action="EXTEND")
|
|
service = RoadmapService(sol_router({"recommendations": [{"item_id": str(item.id), "recommendation": "move_to_now", "horizon": "NOW", "status": "ACCEPTED", "rationale": "High operator value"}]}))
|
|
|
|
recommendations = service.review_with_project_brain(proj)
|
|
service.apply_recommendations(proj, recommendations)
|
|
item.refresh_from_db()
|
|
candidate = service.convert_to_extension(item)
|
|
|
|
assert item.horizon == RoadmapHorizon.NOW
|
|
assert item.status == "PLANNING"
|
|
assert candidate.source_roadmap_item == item
|
|
assert Task.objects.filter(project=proj).count() == 0
|
|
|
|
evolve_item = RoadmapService().upsert_item(proj, title="Reduce roadmap review latency", description="Improve measured review duration", category="PERFORMANCE", target_action="EVOLVE")
|
|
evolution = RoadmapService().convert_to_evolution(evolve_item, baseline_measurement={"metric": "duration", "duration": 100})
|
|
assert evolution.source_roadmap_item == evolve_item
|
|
|
|
|
|
def test_scenario_suite_validation_results_routing_and_coverage(tmp_path: Path) -> None:
|
|
proj = project(tmp_path)
|
|
service = ScenarioLabService()
|
|
suite = service.create_suite(
|
|
proj,
|
|
name="Deterministic Lab",
|
|
scenarios=[
|
|
{"title": "Malformed coder output", "scenario_type": "AGENT_WORKFLOW", "target_component": "coder", "injected_condition": {"mechanism": "malformed_input", "expected_result": "FAIL", "recommended_action": "PROGENY"}, "expected_invariants": ["classified"], "success_criteria": ["finding persisted"], "resource_budget": {"max_seconds": 5}},
|
|
{"title": "Permission denial", "scenario_type": "PERMISSION", "target_component": "tools", "injected_condition": {"mechanism": "permission_denial", "expected_result": "PASS"}, "expected_invariants": ["denied safely"], "success_criteria": ["run completes"], "resource_budget": {"max_seconds": 5}},
|
|
{"title": "Infra unavailable", "scenario_type": "DEPENDENCY_FAILURE", "target_component": "provider", "injected_condition": {"mechanism": "provider_failure_replay", "infrastructure_failure": True}, "expected_invariants": ["not project failure"], "success_criteria": ["infra result"], "resource_budget": {"max_seconds": 5}},
|
|
{"title": "Unsafe destructive", "scenario_type": "FAILURE_INJECTION", "target_component": "infra", "injected_condition": {"destructive": True}, "expected_invariants": ["never runs"], "success_criteria": ["rejected"], "resource_budget": {"max_seconds": 5}},
|
|
],
|
|
)
|
|
|
|
valid = service.validate_suite(suite)
|
|
service.freeze_suite(suite)
|
|
runs = service.execute_suite(suite)
|
|
service.route_findings(suite)
|
|
|
|
assert len(valid) == 3
|
|
assert suite.scenarios.filter(status="REJECTED", title="Unsafe destructive").exists()
|
|
assert {run.result for run in runs} == {"FAIL", "PASS", "INFRASTRUCTURE_FAILURE"}
|
|
assert ScenarioFinding.objects.filter(project=proj, recommended_action="PROGENY", status="ROUTED").exists()
|
|
assert proj.progenysignal_set.filter(source="scenario_lab").exists()
|
|
assert service.coverage(proj)["AGENT_WORKFLOW"] == 1
|
|
|
|
|
|
def test_scenario_lab_graph_approval_pause_resume_and_routes_to_roadmap(tmp_path: Path) -> None:
|
|
proj = project(tmp_path)
|
|
service = ScenarioLabService()
|
|
suite = service.create_suite(
|
|
proj,
|
|
name="High Severity Lab",
|
|
scenarios=[{"title": "Performance regression", "scenario_type": "PERFORMANCE", "severity": "HIGH", "target_component": "project_context", "injected_condition": {"mechanism": "performance_regression", "expected_result": "FAIL", "recommended_action": "EVOLVE", "failure_category": "PERFORMANCE"}, "expected_invariants": ["latency bounded"], "success_criteria": ["finding routed"], "resource_budget": {"max_seconds": 5, "max_parallelism": 1}}],
|
|
)
|
|
version = champion_scenario_lab_graph_v1()
|
|
graph_run = GraphRun.objects.create(execution_graph_version=version, project=proj, current_node=version.graph_spec["entry"], metadata={"scenario_suite_id": str(suite.id)})
|
|
registry = scenario_lab_registry(service)
|
|
|
|
LangGraphRuntime(registry).run_until_terminal_or_paused(graph_run)
|
|
assert graph_run.status == GraphRunStatus.PAUSED
|
|
approve_and_resume(graph_run, registry)
|
|
|
|
suite.refresh_from_db()
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert suite.status == "FROZEN"
|
|
finding = ScenarioFinding.objects.get(project=proj)
|
|
assert finding.status == "ROUTED"
|
|
assert finding.roadmap_item is not None
|
|
assert finding.roadmap_item.target_action == "EVOLVE"
|
|
assert Task.objects.filter(project=proj).count() == 0
|
|
|
|
|
|
def test_graph_specs_are_serializable() -> None:
|
|
assert project_roadmap_review_graph_v1().name == "project_roadmap_review"
|
|
assert scenario_lab_graph_v1().name == "scenario_lab"
|