HAL9000
f4cea72248
The PR #11143 adds action_invariants as a 4th parameter to merge_invariants() and InvariantSet.merge(), but two call sites were not updated: - benchmarks/invariant_merge_bench.py: 5 calls with 3 positional args - robot/helper_m3_e2e_verification.py: 2 calls using keyword args All call sites now pass action_invariants=[] for backward-compatible empty-action behavior.
943 lines
34 KiB
Python
943 lines
34 KiB
Python
"""Robot Framework helper for M3 acceptance-gate verification.
|
|
|
|
This helper validates issue #494 acceptance criteria by exercising the
|
|
actual CLI command paths for:
|
|
|
|
- ``agents plan use`` + ``agents plan execute``
|
|
- ``agents plan tree``
|
|
- ``agents plan explain``
|
|
- ``agents invariant add/list`` (project-scoped)
|
|
- ``agents plan correct`` (dry-run and live revert)
|
|
|
|
CLI-facing tests (1--7) invoke the real ``agents`` CLI via subprocess
|
|
without mocking the service layer. Domain-level tests (8--10) exercise
|
|
real application code directly.
|
|
|
|
Each command prints a success sentinel and exits with status 0. Any
|
|
validation failure prints diagnostics to stderr and exits with status 1.
|
|
|
|
Usage:
|
|
python robot/helper_m3_e2e_verification.py <command>
|
|
"""
|
|
|
|
# ruff: noqa: E402
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any, NoReturn
|
|
|
|
# Ensure src is importable when run from workspace root
|
|
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
|
if _SRC not in sys.path:
|
|
sys.path.insert(0, _SRC)
|
|
|
|
# Ensure robot/ is on the import path for helper_e2e_common.
|
|
_ROBOT = str(Path(__file__).resolve().parent)
|
|
if _ROBOT not in sys.path:
|
|
sys.path.insert(0, _ROBOT)
|
|
|
|
from helper_e2e_common import (
|
|
cleanup_workspace,
|
|
is_expected_provider_unavailable,
|
|
run_cli,
|
|
setup_workspace,
|
|
write_yaml,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Heavy imports are deferred to function bodies to reduce per-process
|
|
# memory consumption. When the test suite is executed with high
|
|
# parallelism (e.g. 32 processes), eager module-level imports cause
|
|
# each helper process to consume ~90 MB before any work begins.
|
|
# Deferring imports keeps CLI-only subcommands (plan-generates-decisions,
|
|
# decision-tree-view, decision-explain, invariant-add-list, etc.) lean
|
|
# and avoids OOM kills (SIGKILL / exit -9) during parallel runs.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PROJECT_NAME = "local/large-project"
|
|
_RESOURCE_MAIN = "01HXM8D2ZK4Q7C2B3F2R4VYV6K"
|
|
_RESOURCE_REQS = "01HXM8E2ZK4Q7C2B3F2R4VYV6M"
|
|
_RESOURCE_DOCKER = "01HXM8F2ZK4Q7C2B3F2R4VYV6N"
|
|
|
|
_ACTION_YAML = """\
|
|
name: local/complex-action
|
|
description: M3 acceptance-gate action
|
|
definition_of_done: Decisions are recorded and executable
|
|
strategy_actor: openai/gpt-4
|
|
execution_actor: openai/gpt-4
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _fail(message: str) -> NoReturn:
|
|
"""Print failure message to stderr and exit with code 1."""
|
|
print(f"FAIL: {message}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def _extract_plan_id(output: str) -> str | None:
|
|
"""Extract a ULID plan_id from plain CLI output."""
|
|
match = re.search(r"\b([0-9A-Z]{26})\b", output)
|
|
return match.group(1) if match else None
|
|
|
|
|
|
def _load_json(output: str) -> Any:
|
|
"""Parse JSON output from a CLI command or fail with diagnostics.
|
|
|
|
Some CLI paths emit structured logs before the JSON payload. This
|
|
parser scans forward for the first JSON object/array that extends
|
|
to end-of-output.
|
|
"""
|
|
text = output.strip()
|
|
decoder = json.JSONDecoder()
|
|
|
|
# Fast path: payload is pure JSON.
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
for index, char in enumerate(text):
|
|
if char not in "[{":
|
|
continue
|
|
candidate = text[index:]
|
|
try:
|
|
value, end = decoder.raw_decode(candidate)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if candidate[end:].strip():
|
|
continue
|
|
return value
|
|
|
|
_fail(f"invalid JSON output:\n---\n{output}\n---")
|
|
|
|
|
|
def _make_uow(database_url: str = "sqlite:///:memory:"):
|
|
"""Create and initialize a UnitOfWork for persistence tests."""
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
uow = UnitOfWork(database_url)
|
|
uow.init_database()
|
|
return uow
|
|
|
|
|
|
def _make_settings(database_url: str = "sqlite:///:memory:"):
|
|
"""Create a minimal Settings for service wiring.
|
|
|
|
Temporarily sets ``CLEVERAGENTS_DATABASE_URL`` so the Pydantic env
|
|
alias resolves ``Settings.database_url`` to the given URL.
|
|
"""
|
|
import os as _os
|
|
|
|
from cleveragents.config.settings import Settings
|
|
|
|
prev = _os.environ.get("CLEVERAGENTS_DATABASE_URL")
|
|
_os.environ["CLEVERAGENTS_DATABASE_URL"] = database_url
|
|
try:
|
|
return Settings()
|
|
finally:
|
|
if prev is None:
|
|
_os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
|
|
else:
|
|
_os.environ["CLEVERAGENTS_DATABASE_URL"] = prev
|
|
|
|
|
|
def _seed_decisions(
|
|
decision_service,
|
|
plan_id: str,
|
|
):
|
|
"""Seed a 3-node decision tree through DecisionService APIs."""
|
|
from cleveragents.domain.models.core.decision import (
|
|
ContextSnapshot,
|
|
DecisionType,
|
|
ResourceRef,
|
|
)
|
|
|
|
root = decision_service.record_decision(
|
|
plan_id=plan_id,
|
|
decision_type=DecisionType.PROMPT_DEFINITION,
|
|
question="What should we build?",
|
|
chosen_option="A REST API for user management",
|
|
alternatives_considered=["GraphQL API", "gRPC service"],
|
|
confidence_score=0.95,
|
|
rationale="REST fits integration requirements and existing tooling.",
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="sha256:root_ctx_hash",
|
|
hot_context_ref="store://snapshots/root",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=_RESOURCE_MAIN, path="src/main.py"),
|
|
],
|
|
actor_state_ref="checkpoint://actor/root",
|
|
),
|
|
)
|
|
|
|
child = decision_service.record_decision(
|
|
plan_id=plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework should we use?",
|
|
chosen_option="FastAPI",
|
|
parent_decision_id=root.decision_id,
|
|
alternatives_considered=["Flask", "Django"],
|
|
confidence_score=0.90,
|
|
rationale="FastAPI provides async support and generated API docs.",
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="sha256:child_ctx_hash",
|
|
hot_context_ref="store://snapshots/child",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=_RESOURCE_REQS, path="requirements.txt"),
|
|
],
|
|
actor_state_ref="checkpoint://actor/child",
|
|
),
|
|
)
|
|
|
|
grandchild = decision_service.record_decision(
|
|
plan_id=plan_id,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question="Which database should we use?",
|
|
chosen_option="PostgreSQL",
|
|
parent_decision_id=child.decision_id,
|
|
alternatives_considered=["SQLite", "MySQL", "MongoDB"],
|
|
confidence_score=0.85,
|
|
rationale="PostgreSQL supports relational and JSON-centric workloads.",
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="sha256:grandchild_ctx_hash",
|
|
hot_context_ref="store://snapshots/grandchild",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=_RESOURCE_DOCKER, path="docker-compose.yml"),
|
|
],
|
|
actor_state_ref="checkpoint://actor/grandchild",
|
|
),
|
|
)
|
|
|
|
return root, child, grandchild
|
|
|
|
|
|
def _seed_workspace_decisions(
|
|
workspace: str,
|
|
):
|
|
"""Seed decisions into the workspace DB and return plan_id + decisions.
|
|
|
|
Creates an action and plan via subprocess, then seeds decisions via
|
|
direct service calls against the same SQLite file.
|
|
"""
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
yaml_path = write_yaml(_ACTION_YAML)
|
|
try:
|
|
r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
|
|
if r1.returncode != 0:
|
|
_fail(f"action create: {r1.stderr}")
|
|
finally:
|
|
os.unlink(yaml_path)
|
|
|
|
r2 = run_cli(
|
|
"plan",
|
|
"use",
|
|
"local/complex-action",
|
|
"--format",
|
|
"plain",
|
|
workspace=workspace,
|
|
)
|
|
if r2.returncode != 0:
|
|
_fail(f"plan use: {r2.stderr}")
|
|
plan_id = _extract_plan_id(r2.stdout)
|
|
if not plan_id:
|
|
_fail(f"could not extract plan_id:\n{r2.stdout}")
|
|
|
|
# Connect to the same DB the subprocess uses
|
|
db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
|
|
uow = UnitOfWork(db_url)
|
|
settings = _make_settings(db_url)
|
|
decision_service = DecisionService(settings=settings, unit_of_work=uow)
|
|
root, child, grandchild = _seed_decisions(decision_service, plan_id)
|
|
|
|
return plan_id, root, child, grandchild
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: plan-generates-decisions (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def plan_generates_decisions() -> None:
|
|
"""Validate ``plan use`` + ``plan execute`` via subprocess.
|
|
|
|
Creates an action and plan via CLI, verifies the plan is in
|
|
the strategize phase, then attempts plan execute (which correctly
|
|
rejects because the plan is not yet strategize-complete without AI).
|
|
Decision recording is verified in the domain-level tests.
|
|
"""
|
|
workspace = setup_workspace(prefix="m3_decisions_")
|
|
yaml_path = write_yaml(_ACTION_YAML)
|
|
try:
|
|
r1 = run_cli("action", "create", "--config", yaml_path, workspace=workspace)
|
|
if r1.returncode != 0:
|
|
_fail(f"action create: {r1.stderr}")
|
|
|
|
r2 = run_cli(
|
|
"plan",
|
|
"use",
|
|
"local/complex-action",
|
|
_PROJECT_NAME,
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
if r2.returncode != 0:
|
|
_fail(f"plan use: {r2.stderr}")
|
|
|
|
use_data = _load_json(r2.stdout)
|
|
if not isinstance(use_data, dict):
|
|
_fail(f"plan use output is not an object: {use_data}")
|
|
plan_id = use_data.get("plan_id")
|
|
if not isinstance(plan_id, str) or not plan_id:
|
|
_fail(f"plan use output missing plan_id: {use_data}")
|
|
if use_data.get("phase") != "strategize":
|
|
_fail(f"plan use should return strategize phase: {use_data}")
|
|
|
|
# Plan execute — plan is queued, correctly rejects without crash
|
|
r3 = run_cli(
|
|
"plan",
|
|
"execute",
|
|
plan_id,
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
combined = r3.stdout + r3.stderr
|
|
if "Traceback" in combined:
|
|
_fail(f"plan execute crashed:\n{combined}")
|
|
if "INTERNAL" in combined and not is_expected_provider_unavailable(combined):
|
|
_fail(f"plan execute crashed:\n{combined}")
|
|
|
|
print("m3-plan-generates-decisions-ok")
|
|
finally:
|
|
os.unlink(yaml_path)
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: decision-tree-view (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def decision_tree_view() -> None:
|
|
"""Invoke ``plan tree`` CLI via subprocess and validate hierarchy."""
|
|
workspace = setup_workspace(prefix="m3_tree_")
|
|
try:
|
|
plan_id, root, child, grandchild = _seed_workspace_decisions(workspace)
|
|
|
|
result = run_cli(
|
|
"plan",
|
|
"tree",
|
|
plan_id,
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
if result.returncode != 0:
|
|
_fail(f"plan tree rc={result.returncode}\n{result.stderr}")
|
|
|
|
tree_data = _load_json(result.stdout)
|
|
if not isinstance(tree_data, list) or len(tree_data) != 1:
|
|
_fail(f"expected one root node from plan tree, got: {tree_data}")
|
|
|
|
root_node = tree_data[0]
|
|
if root_node.get("decision_id") != root.decision_id:
|
|
_fail(f"root node mismatch: {root_node}")
|
|
|
|
child_nodes = root_node.get("children")
|
|
if not isinstance(child_nodes, list) or len(child_nodes) != 1:
|
|
_fail(f"expected one child node under root, got: {child_nodes}")
|
|
|
|
child_node = child_nodes[0]
|
|
if child_node.get("decision_id") != child.decision_id:
|
|
_fail(f"child node mismatch: {child_node}")
|
|
|
|
grandchild_nodes = child_node.get("children")
|
|
if not isinstance(grandchild_nodes, list) or len(grandchild_nodes) != 1:
|
|
_fail(f"expected one grandchild node, got: {grandchild_nodes}")
|
|
|
|
if grandchild_nodes[0].get("decision_id") != grandchild.decision_id:
|
|
_fail(f"grandchild node mismatch: {grandchild_nodes[0]}")
|
|
|
|
print("m3-decision-tree-view-ok")
|
|
finally:
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: decision-explain (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def decision_explain() -> None:
|
|
"""Invoke ``plan explain`` CLI via subprocess and validate context."""
|
|
workspace = setup_workspace(prefix="m3_explain_")
|
|
try:
|
|
_, _, child, _ = _seed_workspace_decisions(workspace)
|
|
|
|
result = run_cli(
|
|
"plan",
|
|
"explain",
|
|
child.decision_id,
|
|
"--show-context",
|
|
"--show-reasoning",
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
if result.returncode != 0:
|
|
_fail(f"plan explain rc={result.returncode}\n{result.stderr}")
|
|
|
|
data = _load_json(result.stdout)
|
|
if not isinstance(data, dict):
|
|
_fail(f"plan explain output is not an object: {data}")
|
|
|
|
if data.get("decision_id") != child.decision_id:
|
|
_fail(f"decision_id mismatch: {data}")
|
|
if data.get("question") != child.question:
|
|
_fail(f"question mismatch: {data}")
|
|
if data.get("chosen") != child.chosen_option:
|
|
_fail(f"chosen option mismatch: {data}")
|
|
if data.get("rationale") != child.rationale:
|
|
_fail(f"rationale mismatch: {data}")
|
|
|
|
if "confidence" not in data:
|
|
_fail(f"plan explain output missing 'confidence' field: {data}")
|
|
|
|
snapshot = data.get("context_snapshot")
|
|
if not isinstance(snapshot, dict):
|
|
_fail(f"missing context_snapshot object: {data}")
|
|
required_snapshot_keys = {
|
|
"hot_context_hash",
|
|
"hot_context_ref",
|
|
"actor_state_ref",
|
|
"relevant_resources",
|
|
}
|
|
missing = required_snapshot_keys - set(snapshot.keys())
|
|
if missing:
|
|
_fail(f"context snapshot missing keys: {sorted(missing)}")
|
|
|
|
resources = snapshot.get("relevant_resources")
|
|
if not isinstance(resources, list) or len(resources) == 0:
|
|
_fail(f"context snapshot has no relevant resources: {snapshot}")
|
|
|
|
print("m3-decision-explain-ok")
|
|
finally:
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: invariant-add-list (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def invariant_add_and_list() -> None:
|
|
"""Validate invariant add/list CLI commands via subprocess.
|
|
|
|
``InvariantService`` is intentionally in-memory, so each subprocess
|
|
invocation gets a fresh store. The test verifies CLI argument
|
|
parsing, output format, and that each command succeeds individually.
|
|
"""
|
|
from cleveragents.domain.models.core.invariant import InvariantScope
|
|
|
|
workspace = setup_workspace(prefix="m3_invariant_")
|
|
try:
|
|
add_result = run_cli(
|
|
"invariant",
|
|
"add",
|
|
"--project",
|
|
_PROJECT_NAME,
|
|
"Use session cookies",
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
if add_result.returncode != 0:
|
|
_fail(f"invariant add rc={add_result.returncode}\n{add_result.stderr}")
|
|
|
|
add_data = _load_json(add_result.stdout)
|
|
if not isinstance(add_data, dict):
|
|
_fail(f"invariant add output is not an object: {add_data}")
|
|
if add_data.get("scope") != InvariantScope.PROJECT.value:
|
|
_fail(f"invariant add scope mismatch: {add_data}")
|
|
if add_data.get("source_name") != _PROJECT_NAME:
|
|
_fail(f"invariant add source_name mismatch: {add_data}")
|
|
|
|
# List is a separate process — invariants are in-memory so this
|
|
# returns an empty list or "No invariants found." message.
|
|
list_result = run_cli(
|
|
"invariant",
|
|
"list",
|
|
"--project",
|
|
_PROJECT_NAME,
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
if list_result.returncode != 0:
|
|
_fail(f"invariant list rc={list_result.returncode}\n{list_result.stderr}")
|
|
# Empty result may be "No invariants found." text or empty JSON []
|
|
combined = list_result.stdout + list_result.stderr
|
|
if "INTERNAL" in combined or "Traceback" in combined:
|
|
_fail(f"invariant list crashed:\n{combined}")
|
|
|
|
print("m3-invariant-add-list-ok")
|
|
finally:
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: correction-dry-run (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def correction_dry_run() -> None:
|
|
"""Validate dry-run correction via service logic and CLI subprocess."""
|
|
from cleveragents.application.services.correction_service import CorrectionService
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
from cleveragents.domain.models.core.correction import CorrectionMode
|
|
|
|
# Part 1: Domain-level validation of correction impact analysis
|
|
database_url = "sqlite:///:memory:"
|
|
uow = _make_uow(database_url)
|
|
settings = _make_settings(database_url)
|
|
decision_service = DecisionService(settings=settings, unit_of_work=uow)
|
|
_plan_ulid = "01HXM8C2ZK4Q7C2B3F2R4VYV6J"
|
|
root, child, grandchild = _seed_decisions(decision_service, _plan_ulid)
|
|
|
|
service = CorrectionService()
|
|
tree = {
|
|
root.decision_id: [child.decision_id],
|
|
child.decision_id: [grandchild.decision_id],
|
|
}
|
|
|
|
request = service.request_correction(
|
|
plan_id=_plan_ulid,
|
|
target_decision_id=child.decision_id,
|
|
mode=CorrectionMode.REVERT,
|
|
guidance="Use session cookies instead of JWT",
|
|
dry_run=True,
|
|
)
|
|
impact = service.analyze_impact(request.correction_id, tree)
|
|
report = service.generate_dry_run_report(request.correction_id, tree)
|
|
|
|
if impact.affected_decisions != [child.decision_id, grandchild.decision_id]:
|
|
_fail(f"unexpected dry-run impact decisions: {impact.affected_decisions}")
|
|
if report.mode != CorrectionMode.REVERT:
|
|
_fail(f"unexpected dry-run report mode: {report.mode}")
|
|
|
|
# Part 2: CLI subprocess — seed decisions, then run plan correct --dry-run
|
|
workspace = setup_workspace(prefix="m3_correct_dry_")
|
|
try:
|
|
plan_id, _, ws_child, _ = _seed_workspace_decisions(workspace)
|
|
|
|
result = run_cli(
|
|
"plan",
|
|
"correct",
|
|
ws_child.decision_id,
|
|
"--mode",
|
|
"revert",
|
|
"--guidance",
|
|
"Use session cookies instead of JWT",
|
|
"--dry-run",
|
|
"--plan",
|
|
plan_id,
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
combined = result.stdout + result.stderr
|
|
if "INTERNAL" in combined or "Traceback" in combined:
|
|
_fail(f"correct dry-run crashed:\n{combined}")
|
|
|
|
# Command may succeed or produce a controlled error — no crash is OK
|
|
if result.returncode == 0:
|
|
data = _load_json(result.stdout)
|
|
if not isinstance(data, dict):
|
|
_fail(f"dry-run output is not an object: {data}")
|
|
if data.get("target_decision") != ws_child.decision_id:
|
|
_fail(f"dry-run target decision mismatch: {data}")
|
|
|
|
print("m3-correction-dry-run-ok")
|
|
finally:
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: correction-live-revert (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def correction_live_revert() -> None:
|
|
"""Validate live revert correction via service logic and CLI subprocess."""
|
|
from cleveragents.application.services.correction_service import CorrectionService
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
from cleveragents.domain.models.core.correction import (
|
|
CorrectionMode,
|
|
CorrectionStatus,
|
|
)
|
|
|
|
# Part 1: Domain-level validation of live revert
|
|
database_url = "sqlite:///:memory:"
|
|
uow = _make_uow(database_url)
|
|
settings = _make_settings(database_url)
|
|
decision_service = DecisionService(settings=settings, unit_of_work=uow)
|
|
_plan_ulid = "01HXM8C2ZK4Q7C2B3F2R4VYV6J"
|
|
root, child, grandchild = _seed_decisions(decision_service, _plan_ulid)
|
|
|
|
service = CorrectionService()
|
|
tree = {
|
|
root.decision_id: [child.decision_id],
|
|
child.decision_id: [grandchild.decision_id],
|
|
}
|
|
|
|
request = service.request_correction(
|
|
plan_id=_plan_ulid,
|
|
target_decision_id=child.decision_id,
|
|
mode=CorrectionMode.REVERT,
|
|
guidance="Switch auth from JWT to session cookies",
|
|
dry_run=False,
|
|
)
|
|
result = service.execute_revert(request.correction_id, tree)
|
|
if result.status != CorrectionStatus.APPLIED:
|
|
_fail(f"expected applied status from execute_revert, got: {result.status}")
|
|
if (
|
|
child.decision_id not in result.reverted_decisions
|
|
or grandchild.decision_id not in result.reverted_decisions
|
|
):
|
|
_fail(f"unexpected reverted decisions: {result.reverted_decisions}")
|
|
if root.decision_id in result.reverted_decisions:
|
|
_fail(f"root decision should not be reverted: {result.reverted_decisions}")
|
|
|
|
# Part 2: CLI subprocess — seed decisions, then run plan correct --yes
|
|
workspace = setup_workspace(prefix="m3_correct_live_")
|
|
try:
|
|
plan_id, _, ws_child, _ = _seed_workspace_decisions(workspace)
|
|
|
|
cli_result = run_cli(
|
|
"plan",
|
|
"correct",
|
|
ws_child.decision_id,
|
|
"--mode",
|
|
"revert",
|
|
"--guidance",
|
|
"Switch auth from JWT to session cookies",
|
|
"--plan",
|
|
plan_id,
|
|
"--yes",
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
combined = cli_result.stdout + cli_result.stderr
|
|
if "INTERNAL" in combined or "Traceback" in combined:
|
|
_fail(f"correct live crashed:\n{combined}")
|
|
|
|
# Verify output structure if command succeeded
|
|
if cli_result.returncode == 0:
|
|
data = _load_json(cli_result.stdout)
|
|
if not isinstance(data, dict):
|
|
_fail(f"live correction output is not an object: {data}")
|
|
if data.get("status") != CorrectionStatus.APPLIED.value:
|
|
_fail(f"live correction status mismatch: {data}")
|
|
|
|
print("m3-correction-live-revert-ok")
|
|
finally:
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: decisions-context-snapshot (domain-level, no CLI)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def decisions_context_snapshot() -> None:
|
|
"""Verify decisions are recorded with full context snapshots."""
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
|
|
database_url = "sqlite:///:memory:"
|
|
uow = _make_uow(database_url)
|
|
settings = _make_settings(database_url)
|
|
decision_service = DecisionService(settings=settings, unit_of_work=uow)
|
|
_plan_ulid = "01HXM8C2ZK4Q7C2B3F2R4VYV6J"
|
|
root, child, grandchild = _seed_decisions(decision_service, _plan_ulid)
|
|
|
|
listed = decision_service.list_decisions(_plan_ulid)
|
|
expected_ids = [root.decision_id, child.decision_id, grandchild.decision_id]
|
|
actual_ids = [decision.decision_id for decision in listed]
|
|
if actual_ids != expected_ids:
|
|
_fail(f"decision ordering mismatch. expected={expected_ids}, got={actual_ids}")
|
|
|
|
for decision in listed:
|
|
snapshot = decision_service.get_snapshot(decision.decision_id)
|
|
if snapshot is None:
|
|
_fail(f"missing snapshot for decision {decision.decision_id}")
|
|
if not snapshot.hot_context_hash:
|
|
_fail(f"missing hot_context_hash for decision {decision.decision_id}")
|
|
if not snapshot.hot_context_ref:
|
|
_fail(f"missing hot_context_ref for decision {decision.decision_id}")
|
|
if not snapshot.relevant_resources:
|
|
_fail(f"missing relevant_resources for decision {decision.decision_id}")
|
|
if not snapshot.actor_state_ref:
|
|
_fail(f"missing actor_state_ref for decision {decision.decision_id}")
|
|
|
|
print("m3-decisions-context-snapshot-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: decision-tree-persistence (CLI via subprocess)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def decision_tree_persistence() -> None:
|
|
"""Verify persisted decision tree round-trip and CLI tree rendering."""
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
|
|
|
|
workspace = setup_workspace(prefix="m3_persist_")
|
|
try:
|
|
plan_id, root, child, grandchild = _seed_workspace_decisions(workspace)
|
|
|
|
# Verify round-trip via direct service (same DB file)
|
|
db_url = os.environ["CLEVERAGENTS_DATABASE_URL"]
|
|
uow = UnitOfWork(db_url)
|
|
settings = _make_settings(db_url)
|
|
reader = DecisionService(settings=settings, unit_of_work=uow)
|
|
restored = reader.list_decisions(plan_id)
|
|
if len(restored) != 3:
|
|
_fail(f"expected 3 persisted decisions, got {len(restored)}")
|
|
|
|
by_id = {d.decision_id: d for d in restored}
|
|
if root.decision_id not in by_id:
|
|
_fail("persisted tree missing root decision")
|
|
if child.decision_id not in by_id:
|
|
_fail("persisted tree missing child decision")
|
|
if grandchild.decision_id not in by_id:
|
|
_fail("persisted tree missing grandchild decision")
|
|
|
|
if by_id[child.decision_id].parent_decision_id != root.decision_id:
|
|
_fail("persisted child parent_decision_id mismatch")
|
|
if by_id[grandchild.decision_id].parent_decision_id != child.decision_id:
|
|
_fail("persisted grandchild parent_decision_id mismatch")
|
|
|
|
# Verify CLI tree reads from the same DB via subprocess
|
|
tree_result = run_cli(
|
|
"plan",
|
|
"tree",
|
|
plan_id,
|
|
"--format",
|
|
"json",
|
|
workspace=workspace,
|
|
)
|
|
if tree_result.returncode != 0:
|
|
_fail(
|
|
"plan tree after persistence "
|
|
f"rc={tree_result.returncode}\n{tree_result.stderr}"
|
|
)
|
|
tree_data = _load_json(tree_result.stdout)
|
|
if not isinstance(tree_data, list) or len(tree_data) != 1:
|
|
_fail(f"unexpected persisted tree output: {tree_data}")
|
|
if tree_data[0].get("decision_id") != root.decision_id:
|
|
_fail(f"persisted tree root mismatch: {tree_data}")
|
|
|
|
print("m3-decision-tree-persistence-ok")
|
|
finally:
|
|
cleanup_workspace(workspace)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: correction-revert-reexecutes (domain-level, no CLI)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def correction_revert_reexecutes() -> None:
|
|
"""Verify revert correction enables re-execution from the corrected node."""
|
|
from cleveragents.application.services.correction_service import CorrectionService
|
|
from cleveragents.application.services.decision_service import DecisionService
|
|
from cleveragents.domain.models.core.correction import CorrectionMode
|
|
from cleveragents.domain.models.core.decision import (
|
|
ContextSnapshot,
|
|
DecisionType,
|
|
ResourceRef,
|
|
)
|
|
|
|
database_url = "sqlite:///:memory:"
|
|
uow = _make_uow(database_url)
|
|
settings = _make_settings(database_url)
|
|
decision_service = DecisionService(settings=settings, unit_of_work=uow)
|
|
_plan_ulid = "01HXM8C2ZK4Q7C2B3F2R4VYV6J"
|
|
root, child, grandchild = _seed_decisions(decision_service, _plan_ulid)
|
|
|
|
tree = {
|
|
root.decision_id: [child.decision_id],
|
|
child.decision_id: [grandchild.decision_id],
|
|
}
|
|
correction_service = CorrectionService()
|
|
|
|
request = correction_service.request_correction(
|
|
plan_id=_plan_ulid,
|
|
target_decision_id=child.decision_id,
|
|
mode=CorrectionMode.REVERT,
|
|
guidance="Use Django instead of FastAPI",
|
|
)
|
|
result = correction_service.execute_revert(request.correction_id, tree)
|
|
if child.decision_id not in result.reverted_decisions:
|
|
_fail(f"target decision not reverted: {result.reverted_decisions}")
|
|
|
|
new_child = decision_service.record_decision(
|
|
plan_id=_plan_ulid,
|
|
decision_type=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework should we use? (corrected)",
|
|
chosen_option="Django",
|
|
parent_decision_id=root.decision_id,
|
|
is_correction=True,
|
|
corrects_decision_id=child.decision_id,
|
|
correction_reason="Need built-in admin interface",
|
|
context_snapshot=ContextSnapshot(
|
|
hot_context_hash="sha256:corrected_child",
|
|
hot_context_ref="store://snapshots/corrected_child",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=_RESOURCE_REQS, path="requirements.txt"),
|
|
],
|
|
actor_state_ref="checkpoint://actor/corrected_child",
|
|
),
|
|
)
|
|
|
|
superseded = child.with_superseded_by(new_child.decision_id)
|
|
if not superseded.is_superseded:
|
|
_fail("corrected flow should mark original decision as superseded")
|
|
if superseded.superseded_by != new_child.decision_id:
|
|
_fail("superseded_by should point to corrected decision")
|
|
|
|
print("m3-correction-revert-reexecutes-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: invariants-enforced-during-strategize (domain-level, no CLI)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def invariants_enforced_during_strategize() -> None:
|
|
"""Verify invariant merge precedence and enforcement record creation."""
|
|
from cleveragents.application.services.invariant_service import InvariantService
|
|
from cleveragents.domain.models.core.invariant import (
|
|
InvariantScope,
|
|
InvariantSet,
|
|
merge_invariants,
|
|
)
|
|
|
|
_plan_ulid = "01HXM8C2ZK4Q7C2B3F2R4VYV6J"
|
|
service = InvariantService()
|
|
|
|
global_inv = service.add_invariant(
|
|
text="Never delete production data",
|
|
scope=InvariantScope.GLOBAL,
|
|
source_name="system",
|
|
)
|
|
project_inv = service.add_invariant(
|
|
text="All API changes need tests",
|
|
scope=InvariantScope.PROJECT,
|
|
source_name=_PROJECT_NAME,
|
|
)
|
|
plan_inv = service.add_invariant(
|
|
text="Use session cookies",
|
|
scope=InvariantScope.PLAN,
|
|
source_name=_plan_ulid,
|
|
)
|
|
|
|
effective = service.get_effective_invariants(
|
|
plan_id=_plan_ulid,
|
|
project_name=_PROJECT_NAME,
|
|
)
|
|
if len(effective) != 3:
|
|
_fail(f"expected 3 effective invariants, got {len(effective)}")
|
|
|
|
merged = merge_invariants(
|
|
plan_invariants=[plan_inv],
|
|
action_invariants=[],
|
|
project_invariants=[project_inv],
|
|
global_invariants=[global_inv],
|
|
)
|
|
if [inv.text for inv in merged] != [
|
|
plan_inv.text,
|
|
project_inv.text,
|
|
global_inv.text,
|
|
]:
|
|
_fail(f"merge precedence mismatch: {[inv.text for inv in merged]}")
|
|
|
|
records = service.enforce_invariants(
|
|
plan_id=_plan_ulid,
|
|
invariants=effective,
|
|
actor_response="All constraints acknowledged",
|
|
)
|
|
if len(records) != 3:
|
|
_fail(f"expected 3 enforcement records, got {len(records)}")
|
|
for record in records:
|
|
if not record.decision_id:
|
|
_fail(f"enforcement record missing decision_id: {record}")
|
|
|
|
invariant_set = InvariantSet.merge(
|
|
plan_invariants=[plan_inv],
|
|
action_invariants=[],
|
|
project_invariants=[project_inv],
|
|
global_invariants=[global_inv],
|
|
)
|
|
if len(invariant_set.invariants) != 3:
|
|
_fail(
|
|
"InvariantSet.merge should preserve all three precedence tiers "
|
|
f"for this input, got {len(invariant_set.invariants)}"
|
|
)
|
|
|
|
print("m3-invariants-enforced-strategize-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_COMMANDS: dict[str, Callable[[], None]] = {
|
|
"plan-generates-decisions": plan_generates_decisions,
|
|
"decision-tree-view": decision_tree_view,
|
|
"decision-explain": decision_explain,
|
|
"invariant-add-list": invariant_add_and_list,
|
|
"correction-dry-run": correction_dry_run,
|
|
"correction-live-revert": correction_live_revert,
|
|
"decisions-context-snapshot": decisions_context_snapshot,
|
|
"decision-tree-persistence": decision_tree_persistence,
|
|
"correction-revert-reexecutes": correction_revert_reexecutes,
|
|
"invariants-enforced-strategize": invariants_enforced_during_strategize,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
"""Entry point called by Robot Framework ``Run Process``."""
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: helper_m3_e2e_verification.py <{'|'.join(_COMMANDS)}>")
|
|
return 1
|
|
|
|
command = sys.argv[1]
|
|
handler = _COMMANDS.get(command)
|
|
if handler is None:
|
|
print(f"Unknown command: {command}")
|
|
return 1
|
|
|
|
handler()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|