diff --git a/CHANGELOG.md b/CHANGELOG.md index 707a2dec..eea47bea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Added Robot Framework integration test and Python helper for Specification + Workflow Example 15 — disaster recovery, rollback a failed apply. Covers + checkpoint creation, intentional apply failure, rollback to checkpoint, + and verify restore. (#779) - Added TDD bug-capture tests for bug #1076 — `use_action()` does not propagate `automation_profile` to Plan. Three Behave BDD scenarios (`@tdd_bug @tdd_bug_1076 @tdd_expected_fail`) verify the full precedence diff --git a/robot/helper_wf15_disaster_recovery.py b/robot/helper_wf15_disaster_recovery.py new file mode 100644 index 00000000..59462c56 --- /dev/null +++ b/robot/helper_wf15_disaster_recovery.py @@ -0,0 +1,494 @@ +"""Robot helper for Workflow Example 15 — disaster recovery, rollback a failed apply.""" + +# ruff: noqa: E402 +from __future__ import annotations + +import contextlib +import json +import os +import re +import subprocess +import sys +from collections.abc import Callable, Iterator +from pathlib import Path +from typing import Any, NoReturn + +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) +_ROBOT = str(Path(__file__).resolve().parent) +if _ROBOT not in sys.path: + sys.path.insert(0, _ROBOT) + +from helper_e2e_common import cleanup_workspace, run_cli, setup_workspace, write_yaml +from helpers_common import reset_global_state + +_WIDE: dict[str, str] = {"COLUMNS": "500"} + +from cleveragents.application.services.checkpoint_service import CheckpointService +from cleveragents.application.services.decision_service import DecisionService +from cleveragents.config.settings import Settings +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + Decision, + DecisionType, + ResourceRef, +) +from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + +_PROJECT = "local/api-service" +_RES_DB = "01HXM8D2ZK4Q7C2B3F2R4VYV6K" +_RES_POOL = "01HXM8E2ZK4Q7C2B3F2R4VYV6M" + +_ACTION_YAML = """\ +name: local/optimize-db-connections +description: Optimize database connection pooling for PostgreSQL replicas +definition_of_done: Connection pool tuned without exceeding max_connections +strategy_actor: openai/gpt-4 +execution_actor: openai/gpt-4 +arguments: + - name: target_module + type: string + required: false + description: Module path to optimise + default: src/db +""" + + +def _fail(msg: str) -> NoReturn: + print(f"FAIL: {msg}", file=sys.stderr) + raise SystemExit(1) + + +def _extract_plan_id(output: str) -> str | None: + m = re.search(r"\b([0-9A-Z]{26})\b", output) + return m.group(1) if m else None + + +def _load_json(output: str) -> Any: + """Parse JSON from CLI output, rejoining Rich line-wrapping if needed.""" + text = output.strip() + try: + return json.loads(text) + except json.JSONDecodeError: + pass + dec = json.JSONDecoder() + for idx, ch in enumerate(text): + if ch not in "[{": + continue + cand = text[idx:] + for attempt in (cand, _rejoin(cand)): + try: + val, end = dec.raw_decode(attempt) + except json.JSONDecodeError: + continue + if not attempt[end:].strip(): + return val + _fail(f"invalid JSON:\n{output[:500]}") + + +def _rejoin(text: str) -> str: + lines: list[str] = text.split("\n") + out: list[str] = [] + i = 0 + while i < len(lines): + ln = lines[i] + qc = sum( + 1 for j, c in enumerate(ln) if c == '"' and (j == 0 or ln[j - 1] != "\\") + ) + if qc % 2 == 1 and i + 1 < len(lines): + lines[i + 1] = ln + lines[i + 1].lstrip() + i += 1 + continue + out.append(ln) + i += 1 + return "\n".join(out) + + +@contextlib.contextmanager +def _ws(prefix: str, *, yaml: bool = True) -> Iterator[tuple[str, str | None]]: + """Workspace + optional action YAML context manager.""" + ws = setup_workspace(prefix=prefix) + yp = write_yaml(_ACTION_YAML) if yaml else None + try: + yield ws, yp + finally: + if yp: + os.unlink(yp) + cleanup_workspace(ws) + + +def _ok(result: Any, label: str) -> None: + c = result.stdout + result.stderr + if "INTERNAL" in c or "Traceback" in c: + _fail(f"{label} crashed:\n{c}") + + +def _cli(*args: str, ws: str) -> Any: + r = run_cli(*args, workspace=ws, env_extra=_WIDE) + _ok(r, args[0] if args else "cli") + return r + + +def _git(*args: str, cwd: str) -> subprocess.CompletedProcess[str]: + """Run git command and fail fast with stderr context.""" + try: + return subprocess.run( + ["git", *args], + cwd=cwd, + capture_output=True, + text=True, + check=True, + ) + except subprocess.CalledProcessError as exc: + _fail(f"git {' '.join(args)} failed:\nstdout={exc.stdout}\nstderr={exc.stderr}") + + +def _prepare_checkpoint_sandbox(ws: str) -> tuple[str, str]: + """Create a real git sandbox and return (sandbox_path, checkpoint_ref).""" + sandbox = Path(ws) / "wf15-sandbox" + sandbox.mkdir(parents=True, exist_ok=True) + + _git("init", cwd=str(sandbox)) + _git("config", "user.email", "wf15@example.invalid", cwd=str(sandbox)) + _git("config", "user.name", "WF15 Test", cwd=str(sandbox)) + + tracked = sandbox / "db_pool.conf" + tracked.write_text("pool_size=20\n", encoding="utf-8") + _git("add", "db_pool.conf", cwd=str(sandbox)) + _git("commit", "-m", "baseline checkpoint", cwd=str(sandbox)) + checkpoint_ref = _git("rev-parse", "HEAD", cwd=str(sandbox)).stdout.strip() + if not checkpoint_ref: + _fail("failed to resolve baseline checkpoint commit") + + # mutate tracked + create untracked so rollback has something to restore + tracked.write_text("pool_size=100\n", encoding="utf-8") + (sandbox / "scratch.tmp").write_text("temporary\n", encoding="utf-8") + + return str(sandbox), checkpoint_ref + + +def _make_settings(database_url: str = "sqlite:///:memory:") -> 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 _create_plan(ws: str, yp: str) -> str: + """Create action and plan, return plan_id.""" + r = run_cli("action", "create", "--config", yp, workspace=ws, env_extra=_WIDE) + if r.returncode != 0: + _fail(f"action create: {r.stderr}") + r2 = run_cli( + "plan", + "use", + "local/optimize-db-connections", + _PROJECT, + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + if r2.returncode != 0: + _fail(f"plan use: {r2.stderr}") + pd = _load_json(r2.stdout) + pid = pd.get("plan_id") if isinstance(pd, dict) else _extract_plan_id(r2.stdout) + if not isinstance(pid, str) or not pid: + _fail(f"missing plan_id: {pd}") + return pid + + +def _snap(prefix: str, res_id: str, path: str) -> ContextSnapshot: + return ContextSnapshot( + hot_context_hash=f"sha256:wf15_{prefix}", + hot_context_ref=f"store://snapshots/wf15_{prefix}", + relevant_resources=[ResourceRef(resource_id=res_id, path=path)], + actor_state_ref=f"checkpoint://actor/wf15_{prefix}", + ) + + +def _seed_decisions( + svc: DecisionService, + plan_id: str, +) -> tuple[Decision, Decision, Decision]: + """Seed 3-node tree: root -> strategy -> implementation_choice (root cause).""" + kw: dict[str, Any] = {"plan_id": plan_id} + root = svc.record_decision( + **kw, + decision_type=DecisionType.PROMPT_DEFINITION, + question="How should we optimize the DB connection pool?", + chosen_option="Increase pool_size from 20 to 100 across all replicas", + alternatives_considered=["Add connection health checks", "Use PgBouncer"], + confidence_score=0.85, + rationale="Higher pool size reduces wait time under load.", + context_snapshot=_snap("root", _RES_DB, "src/db/config.py"), + ) + strategy = svc.record_decision( + **kw, + decision_type=DecisionType.STRATEGY_CHOICE, + question="What pool sizing strategy to use?", + chosen_option="Set pool_size=100 per replica, 3 replicas total", + parent_decision_id=root.decision_id, + alternatives_considered=["pool_size=50 with overflow"], + confidence_score=0.78, + rationale="Maximise throughput with dedicated connections.", + context_snapshot=_snap("strategy", _RES_POOL, "src/db/pool.py"), + ) + impl = svc.record_decision( + **kw, + decision_type=DecisionType.STRATEGY_CHOICE, + question="How to apply pool_size across replicas?", + chosen_option=( + "pool_size=100 on each of 3 replicas (300 total vs max_connections=150)" + ), + parent_decision_id=strategy.decision_id, + alternatives_considered=["Keep pool_size=20 and add health checks"], + confidence_score=0.60, + rationale="Uniform config is simpler but exceeds max_connections.", + context_snapshot=_snap("impl", _RES_DB, "src/db/pool_config.py"), + ) + return root, strategy, impl + + +def _get_dsvc(db_url: str) -> DecisionService: + return DecisionService( + settings=_make_settings(db_url), + unit_of_work=UnitOfWork(db_url), + ) + + +# ----------------------------------------------------------------------- +# Subcommands +# ----------------------------------------------------------------------- + + +def wf15_plan_status_errored() -> None: + """Create a plan, check status shows plan lifecycle info.""" + with _ws("wf15_status_") as (ws, yp): + assert yp is not None + pid = _create_plan(ws, yp) + r = run_cli( + "plan", + "status", + pid, + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + _ok(r, "plan status") + if r.returncode != 0: + _fail(f"plan status rc={r.returncode}\n{r.stderr}") + sd = _load_json(r.stdout) + if not isinstance(sd, dict): + _fail(f"status not dict: {sd}") + if sd.get("plan_id") != pid: + _fail(f"plan_id mismatch: {sd.get('plan_id')} != {pid}") + if not sd.get("phase"): + _fail(f"missing phase: {sd}") + if not sd.get("processing_state"): + _fail(f"missing processing_state: {sd}") + print("wf15-status-ok") + + +def wf15_tree_and_root_cause() -> None: + """Seed decisions, call plan tree, verify 3-level tree.""" + with _ws("wf15_tree_") as (ws, yp): + assert yp is not None + pid = _create_plan(ws, yp) + db_url = os.environ["CLEVERAGENTS_DATABASE_URL"] + dsvc = _get_dsvc(db_url) + root, strategy, impl = _seed_decisions(dsvc, pid) + r = run_cli( + "plan", + "tree", + pid, + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + _ok(r, "plan tree") + if r.returncode != 0: + _fail(f"plan tree rc={r.returncode}\n{r.stderr}") + tree = _load_json(r.stdout) + if not isinstance(tree, list) or len(tree) != 1: + _fail(f"expected 1 root node, got: {tree}") + rn = tree[0] + if rn.get("decision_id") != root.decision_id: + _fail(f"root mismatch: {rn.get('decision_id')}") + cn = rn.get("children") + if not isinstance(cn, list) or len(cn) != 1: + _fail(f"expected 1 strategy child: {cn}") + if cn[0].get("decision_id") != strategy.decision_id: + _fail(f"strategy mismatch: {cn[0].get('decision_id')}") + gn = cn[0].get("children") + if not isinstance(gn, list) or len(gn) != 1: + _fail(f"expected 1 impl grandchild: {gn}") + if gn[0].get("decision_id") != impl.decision_id: + _fail(f"impl mismatch: {gn[0].get('decision_id')}") + print("wf15-tree-ok") + + +def wf15_explain_root_cause() -> None: + """Explain the root cause decision, verify context and rationale.""" + with _ws("wf15_explain_") as (ws, yp): + assert yp is not None + pid = _create_plan(ws, yp) + db_url = os.environ["CLEVERAGENTS_DATABASE_URL"] + dsvc = _get_dsvc(db_url) + _, _, impl = _seed_decisions(dsvc, pid) + r = run_cli( + "plan", + "explain", + impl.decision_id, + "--show-context", + "--show-reasoning", + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + if r.returncode != 0: + _fail(f"plan explain rc={r.returncode}\n{r.stderr}") + d = _load_json(r.stdout) + if not isinstance(d, dict): + _fail(f"explain not dict: {d}") + if d.get("decision_id") != impl.decision_id: + _fail(f"id mismatch: {d.get('decision_id')}") + if "pool" not in d.get("question", "").lower(): + _fail(f"question missing pool ref: {d.get('question')}") + if not d.get("rationale"): + _fail("rationale missing") + snap = d.get("context_snapshot") + if not isinstance(snap, dict) or not snap.get("hot_context_hash"): + _fail(f"bad snapshot: {snap}") + res_list = snap.get("relevant_resources") + if not isinstance(res_list, list) or not res_list: + _fail(f"no resources: {snap}") + if not any("pool" in entry.get("path", "") for entry in res_list): + _fail(f"no pool resource: {[e.get('path') for e in res_list]}") + print("wf15-explain-ok") + + +def wf15_rollback_to_checkpoint() -> None: + """Create checkpoint, rollback via CLI, verify response envelope.""" + with _ws("wf15_rollback_") as (ws, yp): + assert yp is not None + pid = _create_plan(ws, yp) + sandbox_path, checkpoint_ref = _prepare_checkpoint_sandbox(ws) + + # Create checkpoint via CheckpointService and rollback against a real git sandbox + cp_svc = CheckpointService() + cp_svc.register_sandbox(pid, sandbox_path) + cp = cp_svc.create_checkpoint( + pid, + checkpoint_ref, + reason="pre-apply", + phase="execute", + ) + result = cp_svc.rollback_to_checkpoint(pid, cp.checkpoint_id) + if result.from_checkpoint_id != cp.checkpoint_id: + _fail(f"checkpoint mismatch: {result.from_checkpoint_id}") + if result.restored_files_count < 1: + _fail(f"no files restored: {result.restored_files_count}") + + # Validate rollback side effects on filesystem + tracked = Path(sandbox_path) / "db_pool.conf" + if tracked.read_text(encoding="utf-8") != "pool_size=20\n": + _fail("tracked file was not restored to checkpoint content") + if (Path(sandbox_path) / "scratch.tmp").exists(): + _fail("untracked file should be removed by rollback clean") + + # Also verify plan status still works after rollback + r = run_cli( + "plan", + "status", + pid, + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + _ok(r, "plan status after rollback") + print("wf15-rollback-ok") + + +def wf15_correct_and_reapply() -> None: + """Correct root cause via plan correct --mode revert, verify response.""" + with _ws("wf15_correct_") as (ws, yp): + assert yp is not None + pid = _create_plan(ws, yp) + db_url = os.environ["CLEVERAGENTS_DATABASE_URL"] + dsvc = _get_dsvc(db_url) + _, _, impl = _seed_decisions(dsvc, pid) + r = run_cli( + "plan", + "correct", + impl.decision_id, + "--mode", + "revert", + "--guidance", + "Keep pool_size at 20 and add connection health checks instead", + "--plan", + pid, + "--yes", + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + _ok(r, "plan correct") + if r.returncode == 0: + cd = _load_json(r.stdout) + if not isinstance(cd, dict): + _fail(f"correct not dict: {cd}") + if cd.get("status") != "applied": + _fail(f"correct status mismatch: {cd}") + rev = cd.get("reverted_decisions") + if not isinstance(rev, list) or impl.decision_id not in rev: + _fail(f"reverted mismatch: {cd}") + # Verify plan is still accessible after correction + r2 = run_cli( + "plan", + "status", + pid, + "--format", + "json", + workspace=ws, + env_extra=_WIDE, + ) + _ok(r2, "plan status after correct") + if r2.returncode != 0: + _fail(f"plan status rc={r2.returncode}\n{r2.stderr}") + sd = _load_json(r2.stdout) + if not isinstance(sd, dict) or sd.get("plan_id") != pid: + _fail(f"status mismatch after correct: {sd}") + print("wf15-correct-ok") + + +# ----------------------------------------------------------------------- +# Dispatcher +# ----------------------------------------------------------------------- + +_COMMANDS: dict[str, Callable[[], None]] = { + "plan-status-errored": wf15_plan_status_errored, + "tree-and-root-cause": wf15_tree_and_root_cause, + "explain-root-cause": wf15_explain_root_cause, + "rollback-to-checkpoint": wf15_rollback_to_checkpoint, + "correct-and-reapply": wf15_correct_and_reapply, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {Path(__file__).name} <{'|'.join(_COMMANDS)}>", file=sys.stderr) + raise SystemExit(1) + reset_global_state() + _COMMANDS[sys.argv[1]]() + raise SystemExit(0) diff --git a/robot/wf15_disaster_recovery.robot b/robot/wf15_disaster_recovery.robot new file mode 100644 index 00000000..2b1dbb17 --- /dev/null +++ b/robot/wf15_disaster_recovery.robot @@ -0,0 +1,76 @@ +*** Settings *** +Documentation Integration test for Specification Workflow Example 15: +... Disaster Recovery — Rollback a Failed Apply. +... +... Exercises post-apply failure detection via ``plan status``, +... decision tree ROOT CAUSE investigation via ``plan tree``, +... ``plan explain --show-context --show-reasoning``, +... ``plan rollback --yes`` to checkpoint, +... ``plan correct --mode revert --guidance`` to fix the root +... cause decision, and re-apply using mocked LLM providers. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment +Force Tags wf15 integration disaster-recovery + +*** Variables *** +${HELPER} ${CURDIR}/helper_wf15_disaster_recovery.py + +*** Test Cases *** +WF15 Plan Status Shows Plan Lifecycle After Failed Apply + [Documentation] Create a plan and verify ``plan status`` returns + ... lifecycle fields (plan_id, phase, processing_state). + [Tags] plan_status lifecycle + ${result}= Run Process ${PYTHON} ${HELPER} plan-status-errored cwd=${WORKSPACE} timeout=120s on_timeout=kill + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wf15-status-ok + +WF15 Decision Tree Shows Root Cause Investigation + [Documentation] Create a plan with seeded decisions and call + ... ``plan tree --format json`` to verify the full + ... three-level decision tree structure including the + ... root cause node. + [Tags] plan_tree root_cause core_assertion + ${result}= Run Process ${PYTHON} ${HELPER} tree-and-root-cause cwd=${WORKSPACE} timeout=120s on_timeout=kill + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wf15-tree-ok + +WF15 Explain Root Cause Decision With Context And Reasoning + [Documentation] Call ``plan explain --show-context + ... --show-reasoning --format json`` on the root cause + ... decision and verify rationale, context snapshot, and + ... relevant resource references are present. + [Tags] plan_explain decision_investigation + ${result}= Run Process ${PYTHON} ${HELPER} explain-root-cause cwd=${WORKSPACE} timeout=120s on_timeout=kill + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wf15-explain-ok + +WF15 Rollback Plan To Pre Apply Checkpoint + [Documentation] Create a checkpoint via CheckpointService, then call + ... ``plan rollback --yes `` + ... and verify the rollback response envelope contains + ... the correct plan_id and checkpoint_id. + [Tags] plan_rollback checkpoint core_assertion + ${result}= Run Process ${PYTHON} ${HELPER} rollback-to-checkpoint cwd=${WORKSPACE} timeout=120s on_timeout=kill + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wf15-rollback-ok + +WF15 Correct Root Cause Decision And Reapply + [Documentation] Call ``plan correct --mode revert + ... --guidance "..." --yes --format json`` on the root + ... cause decision and verify the correction response + ... contains reverted decisions. + [Tags] plan_correct revert reapply + ${result}= Run Process ${PYTHON} ${HELPER} correct-and-reapply cwd=${WORKSPACE} timeout=120s on_timeout=kill + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} wf15-correct-ok