cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

test(integration): workflow example 1 — Hello World, fix a single bug (manual profile) #798

Closed
brent.edwards wants to merge 2 commits from test/int-wf01-hello-world into master
pull from: test/int-wf01-hello-world
merge into: cleveragents:master
Conversation 17 Commits 2 Files Changed 6
+1018 -25
6 changed files with 1018 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3
View File
@@ -2,6 +2,8 @@
## Unreleased
- Added Robot Framework integration test for Specification Workflow Example 1
(Hello World, manual profile). (#765)
- Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
wired into CLI. Four Robot Framework E2E tests prove ContextTierService starts
empty on every CLI invocation. Tests use ``@tdd_expected_fail`` until the bug
@@ -102,6 +104,7 @@
- Added tool-level execution environment preferences with NONE, REQUIRED,
PREFERRED, and SPECIFIC modes. ToolRunner routes tool execution based on
preference mode with caller-override precedence. (#879)
- Added TDD bug-capture tests for #969 — `plan correct` expects `decision_id`
but M3 acceptance test passes `plan_id`. Behave BDD scenarios (revert and
append modes) and Robot Framework integration tests verify that
robot/helper_e2e_common.py
+100 -19
View File
@@ -14,6 +14,7 @@ import subprocess
import sys
import tempfile
from pathlib import Path
from typing import NoReturn
def run_cli(
@@ -109,7 +110,7 @@ def cleanup_workspace(workspace: str) -> None:
os.environ.pop("CLEVERAGENTS_DATABASE_URL", None)
def fail(msg: str) -> None:
def fail(msg: str) -> NoReturn:
"""Print failure message and exit with code 1."""
print(f"FAIL: {msg}", file=sys.stderr)
raise SystemExit(1)
@@ -136,32 +137,112 @@ def write_yaml(content: str) -> str:
return path
def init_bare_git_repo() -> str:
"""Create a bare git repository with an initial commit.
# ---------------------------------------------------------------------------
# Shared constants & context used by WF01 helpers
# ---------------------------------------------------------------------------
ACTION_YAML = """\
name: local/hello-world-fix
description: Fix a single bug in the hello-world project
strategy_actor: openai/gpt-4
execution_actor: openai/gpt-4
definition_of_done: The bug is fixed, tests pass, and the change is committed
automation_profile: manual
arguments:
- name: bug_description
type: string
required: true
description: Description of the bug to fix
- name: affected_file
type: string
required: false
description: Path to the file affected by the bug
"""
class WorkflowCtx:
"""Holds shared state across workflow steps."""
def __init__(self) -> None:
self.workspace: str = ""
self.repo_dir: str = ""
self.yaml_path: str = ""
self.plan_id: str = ""
def setup(self) -> None:
self.workspace = setup_workspace(prefix="wf01_")
self.repo_dir = init_test_git_repo()
self.yaml_path = write_yaml(ACTION_YAML)
def teardown(self) -> None:
if self.yaml_path and os.path.exists(self.yaml_path):
os.unlink(self.yaml_path)
if self.repo_dir:
shutil.rmtree(self.repo_dir, ignore_errors=True)
if self.workspace:
cleanup_workspace(self.workspace)
def init_test_git_repo() -> str:
"""Create a non-bare git repository with an initial commit on the ``main`` branch.
Requires git >= 2.28 (``git init -b`` flag).
Returns the path to the repository.
"""
repo_dir = tempfile.mkdtemp(prefix="e2e_git_")
cmds = [
["git", "init"],
["git", "config", "user.email", "test@example.com"],
["git", "config", "user.name", "Test"],
]
for cmd in cmds:
subprocess.run(cmd, cwd=repo_dir, capture_output=True, check=True)
try:
cmds = [
# git init -b requires git >= 2.28
["git", "init", "-b", "main"],
["git", "config", "user.email", "test@example.com"],
["git", "config", "user.name", "Test"],
]
for cmd in cmds:
subprocess.run(
cmd,
cwd=repo_dir,
capture_output=True,
check=True,
timeout=30,
)
readme = Path(repo_dir) / "README.md"
readme.write_text("# Test repo\n")
readme = Path(repo_dir) / "README.md"
readme.write_text("# Test repo\n")
subprocess.run(
["git", "add", "."],
cwd=repo_dir,
capture_output=True,
check=True,
timeout=30,
)
subprocess.run(
["git", "commit", "-m", "Initial commit"],
cwd=repo_dir,
capture_output=True,
check=True,
timeout=30,
)
except BaseException:
shutil.rmtree(repo_dir, ignore_errors=True)
raise
return repo_dir
def init_bare_git_repo() -> str:
"""Create a bare git repository with ``main`` as the default branch.
Requires git >= 2.28 (``git init -b`` flag).
Returns the path to the bare repository.
"""
repo_dir = tempfile.mkdtemp(prefix="e2e_bare_git_")
subprocess.run(
["git", "add", "."],
cwd=repo_dir,
capture_output=True,
check=True,
)
subprocess.run(
["git", "commit", "-m", "Initial commit"],
# git init -b requires git >= 2.28
["git", "init", "-b", "main", "--bare"],
cwd=repo_dir,
capture_output=True,
check=True,
timeout=30,
)
return repo_dir
robot/helper_m1_e2e_verification.py
+6 -6
View File
@@ -42,7 +42,7 @@ if _ROBOT not in sys.path:
from helper_e2e_common import ( # noqa: E402
cleanup_workspace,
init_bare_git_repo,
init_test_git_repo,
is_expected_provider_unavailable,
run_cli,
setup_workspace,
@@ -193,7 +193,7 @@ def action_create_from_yaml() -> None:
def resource_register_git_checkout() -> None:
"""Register a git-checkout resource via real CLI subprocess."""
workspace = setup_workspace(prefix="m1_resource_")
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
result = run_cli(
"resource",
@@ -237,7 +237,7 @@ def resource_register_git_checkout() -> None:
def project_create_and_link() -> None:
"""Create a project and link a resource via real CLI subprocess."""
workspace = setup_workspace(prefix="m1_project_")
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
# First register a resource (dependency for --resource flag)
r1 = run_cli(
@@ -307,7 +307,7 @@ def plan_full_lifecycle() -> None:
(``changeset-invocations``), not this CLI lifecycle test.
"""
workspace = setup_workspace(prefix="m1_plan_")
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
yaml_path = write_yaml(_VALID_ACTION_YAML)
try:
# Setup: register resource + create project
@@ -536,7 +536,7 @@ def sandbox_isolation_check() -> None:
)
from cleveragents.infrastructure.sandbox.protocol import SandboxStatus
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
# Create sandbox
sandbox = GitWorktreeSandbox(
@@ -596,7 +596,7 @@ def post_apply_commit_check() -> None:
GitWorktreeSandbox,
)
repo_dir = init_bare_git_repo()
repo_dir = init_test_git_repo()
try:
sandbox = GitWorktreeSandbox(
resource_id="res-commit-check",
robot/helper_wf01_hello_world.py
+387
View File
@@ -0,0 +1,387 @@
"""Robot helper for Workflow Example 1 — Hello World, fix a single bug.
Exercises the ``manual`` automation profile through standalone subcommand
tests: init, resource registration, project creation, validation
registration and attachment, action creation, and post-apply commit
verification.
Each subcommand is fully independent: it provisions and tears down its own
workspace, so tests may be run in any order or subset.
Each subcommand prints a sentinel on success.
Exit code 0 = pass, 1 = failure.
"""
from __future__ import annotations
import contextlib
import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from pathlib import Path
# Ensure the src directory is on the import path.
_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 ( # noqa: E402
WorkflowCtx as _WorkflowCtx,
)
from helper_e2e_common import ( # noqa: E402
fail as _fail,
)
from helper_e2e_common import ( # noqa: E402
init_test_git_repo,
run_cli,
write_yaml,
)
from cleveragents.infrastructure.sandbox.git_worktree import ( # noqa: E402
GitWorktreeSandbox,
)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_VALIDATION_YAML = """\
name: local/unit-tests
description: Run unit tests before apply
source: custom
mode: required
code: |
def run(inputs):
return {"passed": True}
"""
# ---------------------------------------------------------------------------
# Subcommand: init
# ---------------------------------------------------------------------------
def wf01_init() -> None:
"""Verify agents init via real CLI subprocess."""
ctx = _WorkflowCtx()
try:
ctx.setup()
result = run_cli(
"init",
"--yes",
"--path",
ctx.workspace,
workspace=ctx.workspace,
)
if result.returncode != 0:
_fail(f"init rc={result.returncode}\n{result.stderr}")
print("wf01-init-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: resource-register
# ---------------------------------------------------------------------------
def wf01_resource_register() -> None:
"""Register a git-checkout resource via real CLI subprocess."""
ctx = _WorkflowCtx()
try:
ctx.setup()
result = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
ctx.repo_dir,
"--branch",
"main",
"--format",
"plain",
workspace=ctx.workspace,
)
if result.returncode != 0:
_fail(f"resource add rc={result.returncode}\n{result.stderr}")
if "hello-repo" not in result.stdout.lower():
_fail(f"resource name not in output:\n{result.stdout}")
# Check for error indicators in output — only flag lines that look
# like actual error messages (starts with "error:" or "Error:"),
# not incidental use of the word "error" in field names.
for line in result.stdout.splitlines():
stripped = line.strip().lower()
if stripped.startswith("error:") or "not found" in stripped:
_fail(f"resource registration appears to have failed:\n{result.stdout}")
print("wf01-resource-register-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: project-create
# ---------------------------------------------------------------------------
def wf01_project_create() -> None:
"""Create project, link resource, add invariant."""
ctx = _WorkflowCtx()
try:
ctx.setup()
# Register resource first
r1 = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
ctx.repo_dir,
"--branch",
"main",
workspace=ctx.workspace,
)
if r1.returncode != 0:
_fail(f"resource add: {r1.stderr}")
# Create project with linked resource and invariant
r2 = run_cli(
"project",
"create",
"local/hello-project",
"--description",
"Hello World project",
"--resource",
"local/hello-repo",
"--invariant",
"All tests must pass before apply",
"--format",
"plain",
workspace=ctx.workspace,
)
if r2.returncode != 0:
_fail(f"project create rc={r2.returncode}\n{r2.stderr}")
if "hello-project" not in r2.stdout.lower():
_fail(f"project name not in output:\n{r2.stdout}")
print("wf01-project-create-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: validation-register (C2)
# ---------------------------------------------------------------------------
def wf01_validation_register() -> None:
"""Register a validation and attach it to the project (spec §1d-e)."""
ctx = _WorkflowCtx()
val_path = ""
try:
ctx.setup()
val_path = write_yaml(_VALIDATION_YAML)
# Setup: resource + project
r_res = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
ctx.repo_dir,
"--branch",
"main",
workspace=ctx.workspace,
)
if r_res.returncode != 0:
_fail(f"resource add: {r_res.stderr}")
r_proj = run_cli(
"project",
"create",
"local/hello-project",
"--resource",
"local/hello-repo",
workspace=ctx.workspace,
)
if r_proj.returncode != 0:
_fail(f"project create: {r_proj.stderr}")
# Validation add
r_val = run_cli(
"validation",
"add",
"--config",
val_path,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_val.returncode != 0:
_fail(f"validation add rc={r_val.returncode}\n{r_val.stderr}")
if "unit-tests" not in r_val.stdout.lower():
_fail(f"validation name not in output:\n{r_val.stdout}")
# Validation attach to project
r_attach = run_cli(
"validation",
"attach",
"--project",
"local/hello-project",
"local/hello-repo",
"local/unit-tests",
"--format",
"plain",
workspace=ctx.workspace,
)
if r_attach.returncode != 0:
_fail(f"validation attach rc={r_attach.returncode}\n{r_attach.stderr}")
print("wf01-validation-register-ok")
finally:
if val_path and os.path.exists(val_path):
os.unlink(val_path)
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: action-create
# ---------------------------------------------------------------------------
def wf01_action_create() -> None:
"""Create an action from YAML config."""
ctx = _WorkflowCtx()
try:
ctx.setup()
result = run_cli(
"action",
"create",
"--config",
ctx.yaml_path,
workspace=ctx.workspace,
)
if result.returncode != 0:
_fail(f"action create rc={result.returncode}\n{result.stderr}")
# Verify action is retrievable
show = run_cli(
"action",
"show",
"local/hello-world-fix",
"--format",
"plain",
workspace=ctx.workspace,
)
if show.returncode != 0:
_fail(f"action show rc={show.returncode}\n{show.stderr}")
if "hello-world-fix" not in show.stdout:
_fail(f"action not found:\n{show.stdout}")
print("wf01-action-create-ok")
finally:
ctx.teardown()
# ---------------------------------------------------------------------------
# Subcommand: post-apply-commit (sandbox-level check)
# ---------------------------------------------------------------------------
def wf01_post_apply_commit() -> None:
"""Verify sandbox commit creates a git commit in the target repo.
Exercises the GitWorktreeSandbox directly (same as M1 verification)
to confirm the fundamental apply mechanism works.
Note: intentionally skips _WorkflowCtx.setup() / workspace setup
because this test operates at the sandbox layer, not the CLI layer.
It only needs a bare git repo and the GitWorktreeSandbox — no
CLEVERAGENTS_HOME, database, or CLI invocations are involved.
"""
repo_dir = init_test_git_repo()
sandbox = None
try:
sandbox = GitWorktreeSandbox(
resource_id="res-wf01",
original_path=repo_dir,
)
plan_id = "01HWFTEST000000000000000WF"
sandbox.create(plan_id=plan_id)
# Write a "bug fix" file in the sandbox
fix_file = sandbox.get_path("src/bugfix.py")
os.makedirs(os.path.dirname(fix_file), exist_ok=True)
with open(fix_file, "w") as f:
f.write(
"# Bug fix for hello-world issue\n"
"def fixed_function():\n"
" return True\n"
)
# Commit the sandbox changes
result = sandbox.commit("fix: resolve hello-world bug")
if not result.success:
_fail(f"sandbox commit failed: {result.error}")
if result.commit_ref is None:
_fail("no commit ref after sandbox commit")
# Verify commit in original repo's git log
log = subprocess.run(
["git", "log", "--oneline", "-5"],
cwd=repo_dir,
capture_output=True,
text=True,
check=True,
timeout=30,
)
if "hello-world bug" not in log.stdout:
_fail(f"commit not found in git log:\n{log.stdout}")
# Verify file on disk in original
if not os.path.exists(os.path.join(repo_dir, "src", "bugfix.py")):
_fail("bugfix.py not present in original repo after apply")
print("wf01-post-apply-commit-ok")
finally:
if sandbox is not None:
with contextlib.suppress(OSError, FileNotFoundError):
sandbox.cleanup()
shutil.rmtree(repo_dir, ignore_errors=True)
# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------
_COMMANDS: dict[str, Callable[[], None]] = {
"init": wf01_init,
"resource-register": wf01_resource_register,
"project-create": wf01_project_create,
"validation-register": wf01_validation_register,
"action-create": wf01_action_create,
"post-apply-commit": wf01_post_apply_commit,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <{'|'.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())
robot/helper_wf01_plan_tests.py
+426
View File
@@ -0,0 +1,426 @@
"""Robot helper for WF01 plan lifecycle integration tests.
Extracted from ``helper_wf01_hello_world.py`` (M5) to stay under the
500-line limit. Contains: plan-lifecycle, plan-state-transitions,
tree-explain-output, diff-output tests. Each subcommand is independent
(own workspace), prints a sentinel on success, exit 0 = pass / 1 = fail.
"""
from __future__ import annotations
import json
import re
import sys
from collections.abc import Callable
from pathlib import Path
# Ensure the src directory is on the import path.
_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 ( # noqa: E402
WorkflowCtx as _WorkflowCtx,
)
from helper_e2e_common import ( # noqa: E402
fail as _fail,
)
from helper_e2e_common import ( # noqa: E402
run_cli,
)
_ULID_RE = re.compile(r"\b([0-9A-HJKMNP-TV-Z]{26})\b")
def _indent(text: str, prefix: str = " ") -> str:
"""Indent every line of *text* for diagnostic log readability."""
return "\n".join(f"{prefix}{line}" for line in text.splitlines())
def _extract_plan_id(output: str) -> str | None:
"""Extract a ULID plan_id from plain CLI output."""
match = _ULID_RE.search(output)
return match.group(1) if match else None
class _PlanWorkflowCtx(_WorkflowCtx):
"""Extends WorkflowCtx with plan-specific setup."""
def setup_with_plan(self) -> str:
"""Run shared resource+project+action+plan-use setup, return plan_id."""
r_res = run_cli(
"resource",
"add",
"git-checkout",
"local/hello-repo",
"--path",
self.repo_dir,
"--branch",
"main",
workspace=self.workspace,
)
if r_res.returncode != 0:
_fail(f"resource add: {r_res.stderr}")
r_proj = run_cli(
"project",
"create",
"local/hello-project",
"--resource",
"local/hello-repo",
workspace=self.workspace,
)
if r_proj.returncode != 0:
_fail(f"project create: {r_proj.stderr}")
r_act = run_cli(
"action",
"create",
"--config",
self.yaml_path,
workspace=self.workspace,
)
if r_act.returncode != 0:
_fail(f"action create: {r_act.stderr}")
r_use = run_cli(
"plan",
"use",
"local/hello-world-fix",
"local/hello-project",
"--automation-profile",
"manual",
"--arg",
"bug_description=Fix the hello-world greeting function",
"--arg",
"affected_file=src/hello.py",
"--format",
"plain",
workspace=self.workspace,
)
if r_use.returncode != 0:
_fail(f"plan use rc={r_use.returncode}\n{r_use.stderr}")
plan_id = _extract_plan_id(r_use.stdout)
if not plan_id:
_fail(f"could not extract plan_id:\n{r_use.stdout}")
self.plan_id = plan_id
# Verify the plan use output acknowledged the manual profile (n2).
# The CLI sets the profile on the returned plan object; verify it
# appears in the use output. The profile may not persist to the DB
# in all code paths, so we check the creation output, not status.
if "manual" not in r_use.stdout.lower():
_fail(f"'manual' profile not in plan use output:\n{r_use.stdout}")
return plan_id
def wf01_plan_lifecycle() -> None:
"""Full plan lifecycle: use -> execute -> tree -> diff -> apply.
Under mock AI, commands may return graceful "not ready" messages.
The test verifies no crashes (no Traceback / INTERNAL errors).
"""
ctx = _PlanWorkflowCtx()
try:
ctx.setup()
plan_id = ctx.setup_with_plan()
# --- Plan Status (verify plan exists) ---
r_status = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_status.returncode != 0:
_fail(f"plan status rc={r_status.returncode}\n{r_status.stderr}")
if plan_id not in r_status.stdout:
_fail(f"plan_id not in status output:\n{r_status.stdout}")
if "strategize" not in r_status.stdout.lower():
_fail(f"'strategize' not in status:\n{r_status.stdout}")
# --- Plan Execute (strategize phase) ---
r_exec1 = run_cli(
"plan",
"execute",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_exec1 = r_exec1.stdout + r_exec1.stderr
lower_exec1 = combined_exec1.lower()
if "internal error" in lower_exec1 or "traceback" in lower_exec1:
_fail(f"plan execute (strategize) crashed:\n{combined_exec1}")
if r_exec1.returncode not in (0, 1):
_fail(
f"plan execute (strategize) unexpected rc={r_exec1.returncode}"
f"\n{combined_exec1}"
)
# --- Plan Execute (execute phase) ---
r_exec2 = run_cli(
"plan",
"execute",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_exec2 = r_exec2.stdout + r_exec2.stderr
lower_exec2 = combined_exec2.lower()
if "internal error" in lower_exec2 or "traceback" in lower_exec2:
_fail(f"plan execute (execute) crashed:\n{combined_exec2}")
if r_exec2.returncode not in (0, 1):
_fail(
f"plan execute (execute) unexpected rc={r_exec2.returncode}"
f"\n{combined_exec2}"
)
# --- Plan Tree ---
r_tree = run_cli(
"plan",
"tree",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_tree = r_tree.stdout + r_tree.stderr
lower_tree = combined_tree.lower()
if "internal error" in lower_tree or "traceback" in lower_tree:
_fail(f"plan tree crashed:\n{combined_tree}")
if r_tree.returncode not in (0, 1):
_fail(f"plan tree unexpected rc={r_tree.returncode}\n{combined_tree}")
# --- Plan Diff ---
r_diff = run_cli(
"plan",
"diff",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_diff = r_diff.stdout + r_diff.stderr
lower_diff = combined_diff.lower()
if "internal error" in lower_diff or "traceback" in lower_diff:
_fail(f"plan diff crashed:\n{combined_diff}")
if r_diff.returncode not in (0, 1):
_fail(f"plan diff unexpected rc={r_diff.returncode}\n{combined_diff}")
# --- Plan Apply --yes ---
r_apply = run_cli(
"plan",
"apply",
"--yes",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_apply = r_apply.stdout + r_apply.stderr
lower_apply = combined_apply.lower()
if "internal error" in lower_apply or "traceback" in lower_apply:
_fail(f"plan apply crashed:\n{combined_apply}")
if r_apply.returncode not in (0, 1):
_fail(f"plan apply unexpected rc={r_apply.returncode}\n{combined_apply}")
print("wf01-plan-lifecycle-ok")
finally:
ctx.teardown()
def wf01_plan_state_transitions() -> None:
"""Verify plan state transitions: create, check initial state, execute,
re-check status. Under mock AI the plan may not progress, which is OK.
"""
ctx = _PlanWorkflowCtx()
try:
ctx.setup()
plan_id = ctx.setup_with_plan()
# Check initial state — should be in strategize phase
r_pre = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_pre.returncode != 0:
_fail(f"plan status (pre-execute): {r_pre.stderr}")
if "strategize" not in r_pre.stdout.lower():
_fail(f"expected 'strategize' in status:\n{r_pre.stdout}")
if "queued" not in r_pre.stdout.lower():
_fail(f"expected 'queued' in status:\n{r_pre.stdout}")
# Execute to attempt state transition
r_exec = run_cli(
"plan",
"execute",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined_exec = r_exec.stdout + r_exec.stderr
lower_exec = combined_exec.lower()
if "internal error" in lower_exec or "traceback" in lower_exec:
_fail(f"plan execute crashed:\n{combined_exec}")
if r_exec.returncode not in (0, 1):
_fail(f"plan execute unexpected rc={r_exec.returncode}\n{combined_exec}")
# Re-check status — plan may still be in strategize under mock AI.
r_post = run_cli(
"plan",
"status",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
if r_post.returncode != 0:
_fail(f"plan status (post-execute): {r_post.stderr}")
if plan_id not in r_post.stdout:
_fail(f"plan_id not in post-execute status:\n{r_post.stdout}")
# Log status for diagnostic visibility.
pre_block = r_pre.stdout.strip() or "unknown"
post_block = r_post.stdout.strip() or "unknown"
print(
f" state transition:\n pre:\n{_indent(pre_block)}"
f"\n post:\n{_indent(post_block)}",
file=sys.stderr,
)
print("wf01-plan-state-transitions-ok")
finally:
ctx.teardown()
def wf01_tree_explain_output() -> None:
"""Verify plan tree (JSON parseable) and plan explain (graceful error).
Under mock AI no decisions exist, so explain verifies graceful handling.
"""
ctx = _PlanWorkflowCtx()
try:
ctx.setup()
plan_id = ctx.setup_with_plan()
# Plan tree — verify it runs without crash
r_tree = run_cli(
"plan",
"tree",
plan_id,
"--format",
"json",
workspace=ctx.workspace,
)
combined_tree = r_tree.stdout + r_tree.stderr
lower_tree = combined_tree.lower()
if "traceback" in lower_tree or "internal error" in lower_tree:
_fail(f"plan tree crashed:\n{combined_tree}")
if r_tree.returncode not in (0, 1):
_fail(f"plan tree unexpected rc={r_tree.returncode}\n{combined_tree}")
if r_tree.returncode == 0 and r_tree.stdout.strip():
# Under mock AI, output may contain "No decisions found" mixed
# with debug log lines rather than valid JSON. Only fail if the
# output looks like it *should* be JSON (starts with [ or {).
stripped = r_tree.stdout.strip()
if stripped.startswith(("{", "[")):
try:
json.loads(stripped)
except json.JSONDecodeError:
_fail(f"plan tree JSON not parseable:\n{r_tree.stdout}")
# Plan explain — uses synthetic decision ULID; expects graceful error.
r_explain = run_cli(
"plan",
"explain",
"01HWFTEST00000000000DECDE0",
"--format",
"plain",
workspace=ctx.workspace,
)
combined_explain = r_explain.stdout + r_explain.stderr
lower_explain = combined_explain.lower()
if "traceback" in lower_explain or "internal error" in lower_explain:
_fail(f"plan explain crashed:\n{combined_explain}")
if r_explain.returncode not in (0, 1):
_fail(
f"plan explain unexpected rc={r_explain.returncode}\n{combined_explain}"
)
print("wf01-tree-explain-output-ok")
finally:
ctx.teardown()
def wf01_diff_output() -> None:
"""Verify plan diff runs without crashing.
Under mock AI the plan hasn't executed, so diff returns a graceful
message or empty output rather than actual changeset content.
"""
ctx = _PlanWorkflowCtx()
try:
ctx.setup()
plan_id = ctx.setup_with_plan()
# Plan diff — verify no internal errors
r_diff = run_cli(
"plan",
"diff",
plan_id,
"--format",
"plain",
workspace=ctx.workspace,
)
combined = r_diff.stdout + r_diff.stderr
lower_combined = combined.lower()
if "internal error" in lower_combined or "traceback" in lower_combined:
_fail(f"plan diff crashed:\n{combined}")
if r_diff.returncode not in (0, 1):
_fail(f"plan diff unexpected rc={r_diff.returncode}\n{combined}")
print("wf01-diff-output-ok")
finally:
ctx.teardown()
_COMMANDS: dict[str, Callable[[], None]] = {
"plan-lifecycle": wf01_plan_lifecycle,
"plan-state-transitions": wf01_plan_state_transitions,
"tree-explain-output": wf01_tree_explain_output,
"diff-output": wf01_diff_output,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <{'|'.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())
robot/wf01_hello_world.robot
+96
View File
@@ -0,0 +1,96 @@
*** Settings ***
Documentation Workflow Example 1 — Hello World: fix a single bug using the manual automation profile.
... Tests standalone CLI subcommands (init, resource add, project create,
... validation add/attach, action create) and plan lifecycle smoke tests
... (use, execute, tree, explain, diff, apply --yes) under mock AI.
... Also verifies post-apply commit via GitWorktreeSandbox.
... Mock AI means plan commands verify CLI wiring and graceful error
... handling rather than full end-to-end execution.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_wf01_hello_world.py
${PLAN_HELPER} ${CURDIR}/helper_wf01_plan_tests.py
*** Test Cases ***
WF01 Init Workspace
[Documentation] Verify agents init creates a workspace configuration
${result}= Run Process ${PYTHON} ${HELPER} init cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-init-ok
WF01 Register Git Checkout Resource
[Documentation] Register a git-checkout resource for the hello-world project
${result}= Run Process ${PYTHON} ${HELPER} resource-register cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-resource-register-ok
WF01 Create Project With Invariant
[Documentation] Create a project, link a resource, and register an invariant
${result}= Run Process ${PYTHON} ${HELPER} project-create cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-project-create-ok
WF01 Register And Attach Validation
[Documentation] Register a validation via YAML and attach it to the project (spec §1d-e)
${result}= Run Process ${PYTHON} ${HELPER} validation-register cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-validation-register-ok
WF01 Create Action From YAML
[Documentation] Create a hello-world fix action from YAML config with manual profile
${result}= Run Process ${PYTHON} ${HELPER} action-create cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-action-create-ok
WF01 Full Plan Lifecycle Manual Profile
[Documentation] Full plan lifecycle: use -> execute -> tree -> diff -> apply (manual profile)
${result}= Run Process ${PYTHON} ${PLAN_HELPER} plan-lifecycle cwd=${WORKSPACE} timeout=300s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-plan-lifecycle-ok
WF01 Plan State Transitions
[Documentation] Verify plan state transitions through the manual-profile lifecycle
${result}= Run Process ${PYTHON} ${PLAN_HELPER} plan-state-transitions cwd=${WORKSPACE} timeout=300s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-plan-state-transitions-ok
WF01 Tree And Explain Output Structure
[Documentation] Verify plan tree and plan explain produce structured output
${result}= Run Process ${PYTHON} ${PLAN_HELPER} tree-explain-output cwd=${WORKSPACE} timeout=300s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-tree-explain-output-ok
WF01 Diff Output
[Documentation] Verify plan diff runs without internal errors
${result}= Run Process ${PYTHON} ${PLAN_HELPER} diff-output cwd=${WORKSPACE} timeout=300s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-diff-output-ok
WF01 Post Apply Commit Exists
[Documentation] Verify post-apply commit exists in the target repository
${result}= Run Process ${PYTHON} ${HELPER} post-apply-commit cwd=${WORKSPACE} timeout=120s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} wf01-post-apply-commit-ok