HAL9000
de1fd49594
Update integration test helpers to use definition_of_done values that satisfy the new TextMatchEvaluator gate (criterion text must appear as a substring of a plan argument key or value). Also fix _evaluate_dod to merge DoD metadata into plan.validation_summary without overwriting Execute-phase validation counts, preserving the coverage/tool-validation gate used by apply_with_validation_gate. Update CONTRIBUTORS.md. Fixes: wf02, wf04, wf06, wf07 integration test suites. ISSUES CLOSED: #7927
567 lines
20 KiB
Python
567 lines
20 KiB
Python
"""Helper for Workflow Example 6: Documentation Generation from Codebase Analysis.
|
|
|
|
Integration test exercising the ``trusted`` automation profile with:
|
|
- Context policy configuration with view-specific settings (Step 2)
|
|
- Action creation with args (doc_types, output_dir) and invariants (Step 1)
|
|
- Trusted profile behavior: auto-progress after strategize, gated apply (Step 3)
|
|
- Temp project fixture with source code files
|
|
- Documentation generation into output directory via sandbox
|
|
- Source-code invariant: no Python source files modified
|
|
|
|
Spec reference: docs/specification.md ~lines 38700-39039
|
|
Issue: #770
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared DB setup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _setup_db() -> Any:
|
|
"""Create in-memory SQLite, return session factory."""
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from cleveragents.infrastructure.database.models import Base
|
|
|
|
engine = create_engine("sqlite:///:memory:", echo=False)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, expire_on_commit=False)()
|
|
|
|
class _NoClose:
|
|
"""Keep in-memory SQLite alive across session.close() calls."""
|
|
|
|
__slots__ = ("_s",)
|
|
|
|
def __init__(self, s: object) -> None:
|
|
object.__setattr__(self, "_s", s)
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def __getattr__(self, n: str) -> object:
|
|
return getattr(object.__getattribute__(self, "_s"), n)
|
|
|
|
wrapper = _NoClose(session)
|
|
return lambda: wrapper
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Context policy with view-specific settings (Spec Step 2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _context_policy_views() -> None:
|
|
"""Configure and verify context policy with strategize/execute views."""
|
|
from cleveragents.domain.models.core.context_policy import (
|
|
ContextView,
|
|
ProjectContextPolicy,
|
|
)
|
|
|
|
# Build strategize view: generous context for code analysis
|
|
strategize_view = ContextView(
|
|
include_resources=["01HXR6A1B2C3D4E5F6G7H8J9K0"],
|
|
exclude_paths=["**/node_modules/**", "**/.git/**", "**/dist/**"],
|
|
max_file_size=2_097_152, # 2 MB
|
|
max_total_size=52_428_800, # 50 MB
|
|
)
|
|
|
|
# Build execute view: narrower context for writing docs
|
|
execute_view = ContextView(
|
|
include_resources=["01HXR6A1B2C3D4E5F6G7H8J9K0"],
|
|
max_file_size=1_048_576, # 1 MB
|
|
max_total_size=52_428_800, # 50 MB
|
|
)
|
|
|
|
policy = ProjectContextPolicy(
|
|
strategize_view=strategize_view,
|
|
execute_view=execute_view,
|
|
)
|
|
|
|
# Verify view resolution per inheritance chain
|
|
resolved_strat = policy.resolve_view("strategize")
|
|
assert resolved_strat is strategize_view
|
|
assert resolved_strat.max_file_size == 2_097_152
|
|
assert "**/node_modules/**" in resolved_strat.exclude_paths
|
|
|
|
resolved_exec = policy.resolve_view("execute")
|
|
assert resolved_exec is execute_view
|
|
assert resolved_exec.max_file_size == 1_048_576
|
|
|
|
# Apply inherits from execute (no explicit apply view)
|
|
resolved_apply = policy.resolve_view("apply")
|
|
assert resolved_apply is execute_view
|
|
|
|
# Default view has no limits (since no explicit default was set)
|
|
resolved_default = policy.resolve_view("default")
|
|
assert resolved_default.max_file_size is None
|
|
|
|
print("context-policy-views-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Budget enforcement with context fragments (Spec Step 2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _budget_enforcement() -> None:
|
|
"""Verify budget enforcement filters fragments correctly."""
|
|
from cleveragents.domain.models.core.context_fragment import (
|
|
ContextFragment,
|
|
FragmentProvenance,
|
|
)
|
|
from cleveragents.domain.models.core.context_policy import (
|
|
ContextView,
|
|
enforce_size_budget,
|
|
)
|
|
|
|
prov = FragmentProvenance(resource_uri="test://budget-test")
|
|
|
|
# Create fragments of varying sizes
|
|
small_frag = ContextFragment(
|
|
fragment_id="frag-small",
|
|
uko_node="uko://test/small",
|
|
token_count=10,
|
|
provenance=prov,
|
|
content="x" * 100,
|
|
)
|
|
medium_frag = ContextFragment(
|
|
fragment_id="frag-medium",
|
|
uko_node="uko://test/medium",
|
|
token_count=50,
|
|
provenance=prov,
|
|
content="y" * 500,
|
|
)
|
|
large_frag = ContextFragment(
|
|
fragment_id="frag-large",
|
|
uko_node="uko://test/large",
|
|
token_count=200,
|
|
provenance=prov,
|
|
content="z" * 2000,
|
|
)
|
|
|
|
# View with per-file limit of 1000 bytes
|
|
view = ContextView(max_file_size=1000, max_total_size=5000)
|
|
result = enforce_size_budget([small_frag, medium_frag, large_frag], view)
|
|
|
|
# Small and medium pass, large is excluded
|
|
assert "frag-small" in result.accepted
|
|
assert "frag-medium" in result.accepted
|
|
assert "frag-large" not in result.accepted
|
|
assert len(result.violations) == 1
|
|
assert result.violations[0].violation_type == "max_file_size"
|
|
|
|
# View with total limit of 200 bytes
|
|
view_total = ContextView(max_total_size=200)
|
|
result2 = enforce_size_budget([small_frag, medium_frag], view_total)
|
|
assert "frag-small" in result2.accepted
|
|
assert "frag-medium" not in result2.accepted
|
|
assert len(result2.violations) == 1
|
|
assert result2.violations[0].violation_type == "max_total_size"
|
|
|
|
print("budget-enforcement-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Trusted profile behavior (Spec Step 3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _trusted_profile_behavior() -> None:
|
|
"""Verify trusted profile: auto-progress after strategize, gated apply."""
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
service = PlanLifecycleService(settings=Settings())
|
|
service.create_action(
|
|
name="local/trusted-gate-test",
|
|
description="Test trusted profile gates",
|
|
definition_of_done="Gates verified",
|
|
strategy_actor="local/stub",
|
|
execution_actor="local/stub",
|
|
automation_profile="trusted",
|
|
)
|
|
plan = service.use_action(
|
|
action_name="local/trusted-gate-test",
|
|
project_links=[ProjectLink(project_name="test")],
|
|
)
|
|
pid = plan.identity.plan_id
|
|
|
|
# Trusted: decompose_task=0.0 → strategize auto-ran in try_auto_run
|
|
# create_tool=0.0 → execute auto-ran in try_auto_run
|
|
plan = service.get_plan(pid)
|
|
assert plan.phase == PlanPhase.EXECUTE, (
|
|
f"Expected auto-progress to Execute, got {plan.phase}"
|
|
)
|
|
assert plan.processing_state == ProcessingState.COMPLETE, (
|
|
f"Expected COMPLETE after auto-progress, got {plan.processing_state}"
|
|
)
|
|
|
|
# Trusted: select_tool=1.0 -> gated apply
|
|
assert plan.phase == PlanPhase.EXECUTE
|
|
assert plan.processing_state == ProcessingState.COMPLETE
|
|
assert service.should_auto_progress(plan) is False, (
|
|
"Trusted profile should NOT auto-progress to apply"
|
|
)
|
|
|
|
# Explicit apply works
|
|
service.apply_plan(pid)
|
|
service.start_apply(pid)
|
|
plan = service.complete_apply(pid)
|
|
assert plan.processing_state == ProcessingState.APPLIED
|
|
assert plan.is_terminal
|
|
|
|
print("trusted-profile-behavior-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. Action with doc-generation args and invariants (Spec Step 1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _action_with_doc_args() -> None:
|
|
"""Create generate-docs action with args and invariants."""
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.action import (
|
|
ActionArgument,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import PlanPhase, ProjectLink
|
|
|
|
service = PlanLifecycleService(settings=Settings())
|
|
action = service.create_action(
|
|
name="local/generate-docs",
|
|
description="Generate comprehensive documentation from codebase analysis",
|
|
long_description=(
|
|
"Analyze the project codebase to produce developer documentation: "
|
|
"API reference, architecture overview, module guides, and onboarding "
|
|
"material."
|
|
),
|
|
definition_of_done=(
|
|
"API reference covers all public endpoints/functions; "
|
|
"architecture document shows module dependencies; "
|
|
"each major module has a developer guide"
|
|
),
|
|
strategy_actor="anthropic/claude-3.5-sonnet",
|
|
execution_actor="anthropic/claude-3.5-sonnet",
|
|
automation_profile="trusted",
|
|
reusable=True,
|
|
arguments=[
|
|
ActionArgument(
|
|
name="doc_types",
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description=(
|
|
"Comma-separated list: api-reference, architecture, "
|
|
"module-guides, onboarding"
|
|
),
|
|
),
|
|
ActionArgument(
|
|
name="output_dir",
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description="Output directory for generated docs",
|
|
default_value="docs/",
|
|
),
|
|
],
|
|
invariants=[
|
|
"Do not modify any source code files — only create or update "
|
|
"files in the output directory",
|
|
"All code examples in documentation must be actual code from "
|
|
"the project, not fabricated",
|
|
"Architecture diagrams must reflect actual module dependencies, "
|
|
"not aspirational ones",
|
|
],
|
|
)
|
|
assert action.automation_profile == "trusted"
|
|
assert len(action.arguments) == 2
|
|
assert len(action.invariants) == 3
|
|
assert action.arguments[0].name == "doc_types"
|
|
assert action.arguments[1].name == "output_dir"
|
|
assert action.arguments[1].requirement == ArgumentRequirement.OPTIONAL
|
|
|
|
plan = service.use_action(
|
|
action_name="local/generate-docs",
|
|
project_links=[ProjectLink(project_name="local/api-service")],
|
|
arguments={
|
|
"doc_types": "api-reference,architecture,module-guides,onboarding",
|
|
"output_dir": "docs/generated/",
|
|
},
|
|
)
|
|
assert plan.phase == PlanPhase.EXECUTE
|
|
assert plan.arguments["doc_types"] == (
|
|
"api-reference,architecture,module-guides,onboarding"
|
|
)
|
|
assert plan.arguments["output_dir"] == "docs/generated/"
|
|
assert len(plan.invariants) >= 3
|
|
|
|
print("action-with-doc-args-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. Temp project fixture + doc generation + source invariant (Spec Step 3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _doc_generation_sandbox() -> None:
|
|
"""Create temp project, generate docs in sandbox, verify invariant."""
|
|
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
|
|
from cleveragents.infrastructure.sandbox.manager import SandboxManager
|
|
|
|
# Create a temporary project directory with source files
|
|
project_dir = tempfile.mkdtemp(prefix="wf06-project-")
|
|
src_dir = Path(project_dir, "src")
|
|
src_dir.mkdir()
|
|
|
|
# Create source files matching the spec scenario
|
|
auth_py = src_dir / "auth.py"
|
|
auth_py.write_text(
|
|
"def authenticate(user, password):\n"
|
|
' """Authenticate a user."""\n'
|
|
' return user == "test_user" and password == "test_password"\n'
|
|
)
|
|
models_py = src_dir / "models.py"
|
|
models_py.write_text(
|
|
"class User:\n"
|
|
' """Domain model for a user entity."""\n'
|
|
"\n"
|
|
" def __init__(self, name: str, email: str):\n"
|
|
" self.name = name\n"
|
|
" self.email = email\n"
|
|
)
|
|
utils_py = src_dir / "utils.py"
|
|
utils_py.write_text(
|
|
"def slugify(text: str) -> str:\n"
|
|
' """Convert text to a URL-friendly slug."""\n'
|
|
' return text.lower().replace(" ", "-")\n'
|
|
)
|
|
|
|
# Capture original content for invariant verification
|
|
auth_before = auth_py.read_text()
|
|
models_before = models_py.read_text()
|
|
utils_before = utils_py.read_text()
|
|
|
|
# Create sandbox from the project directory
|
|
factory = SandboxFactory()
|
|
mgr = SandboxManager(factory=factory, cleanup_on_exit=False)
|
|
sandbox = mgr.get_or_create_sandbox(
|
|
plan_id="plan-wf06-doc-gen",
|
|
resource_id="res-wf06-repo",
|
|
original_path=project_dir,
|
|
sandbox_strategy="copy_on_write",
|
|
)
|
|
assert sandbox.context is not None
|
|
sb_path = Path(sandbox.context.sandbox_path)
|
|
|
|
# Simulate documentation generation in sandbox output directory
|
|
docs_dir = sb_path / "docs" / "generated"
|
|
docs_dir.mkdir(parents=True)
|
|
modules_dir = docs_dir / "modules"
|
|
modules_dir.mkdir()
|
|
|
|
# Create documentation files (simulating mocked LLM output)
|
|
(docs_dir / "api-reference.md").write_text(
|
|
"# API Reference\n\n"
|
|
"## Authentication\n\n"
|
|
"### POST /auth/login\n\n"
|
|
"Authenticates a user with username and password.\n"
|
|
)
|
|
(docs_dir / "architecture.md").write_text(
|
|
"# Architecture Overview\n\n"
|
|
"## Module Dependency Graph\n\n"
|
|
"The system consists of auth, models, and utils modules.\n"
|
|
)
|
|
(modules_dir / "auth.md").write_text(
|
|
"# Auth Module\n\n"
|
|
"Handles user authentication.\n\n"
|
|
"## Functions\n\n"
|
|
"- `authenticate(user, password)` — Validates credentials.\n"
|
|
)
|
|
(modules_dir / "models.md").write_text(
|
|
"# Models Module\n\n"
|
|
"Domain models for the application.\n\n"
|
|
"## Classes\n\n"
|
|
"- `User` — Represents a user entity.\n"
|
|
)
|
|
(docs_dir / "onboarding.md").write_text(
|
|
"# Onboarding Guide\n\n"
|
|
"## Getting Started\n\n"
|
|
"1. Clone the repository\n"
|
|
"2. Install dependencies\n"
|
|
"3. Run the application\n"
|
|
)
|
|
|
|
# Verify documentation files were created with non-empty content
|
|
for doc_file in [
|
|
docs_dir / "api-reference.md",
|
|
docs_dir / "architecture.md",
|
|
docs_dir / "onboarding.md",
|
|
modules_dir / "auth.md",
|
|
modules_dir / "models.md",
|
|
]:
|
|
assert doc_file.exists(), f"Missing doc file: {doc_file}"
|
|
content = doc_file.read_text()
|
|
assert len(content) > 0, f"Empty doc file: {doc_file}"
|
|
|
|
# Verify source code invariant: no source files were modified
|
|
sb_src = sb_path / "src"
|
|
assert (sb_src / "auth.py").read_text() == auth_before, (
|
|
"Source invariant violated: auth.py was modified"
|
|
)
|
|
assert (sb_src / "models.py").read_text() == models_before, (
|
|
"Source invariant violated: models.py was modified"
|
|
)
|
|
assert (sb_src / "utils.py").read_text() == utils_before, (
|
|
"Source invariant violated: utils.py was modified"
|
|
)
|
|
|
|
# Cleanup
|
|
mgr.cleanup_all("plan-wf06-doc-gen")
|
|
shutil.rmtree(project_dir, ignore_errors=True)
|
|
|
|
print("doc-generation-sandbox-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. Full trusted lifecycle for doc generation (Spec Steps 1-3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _trusted_doc_lifecycle() -> None:
|
|
"""Full plan lifecycle with trusted profile for documentation action."""
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
from cleveragents.domain.models.core.action import (
|
|
ActionArgument,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
service = PlanLifecycleService(settings=Settings())
|
|
service.create_action(
|
|
name="local/generate-docs-lifecycle",
|
|
description="Generate docs (lifecycle test)",
|
|
definition_of_done="doc_types",
|
|
strategy_actor="local/stub",
|
|
execution_actor="local/stub",
|
|
automation_profile="trusted",
|
|
reusable=True,
|
|
arguments=[
|
|
ActionArgument(
|
|
name="doc_types",
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Documentation types",
|
|
),
|
|
ActionArgument(
|
|
name="output_dir",
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description="Output directory",
|
|
default_value="docs/",
|
|
),
|
|
],
|
|
invariants=[
|
|
"Do not modify any source code files",
|
|
],
|
|
)
|
|
|
|
plan = service.use_action(
|
|
action_name="local/generate-docs-lifecycle",
|
|
project_links=[ProjectLink(project_name="local/api-service")],
|
|
arguments={
|
|
"doc_types": "api-reference,architecture",
|
|
"output_dir": "docs/generated/",
|
|
},
|
|
)
|
|
pid = plan.identity.plan_id
|
|
assert plan.phase == PlanPhase.EXECUTE
|
|
|
|
# Trusted profile auto-ran strategize (decompose_task=0.0)
|
|
# and execute (create_tool=0.0) in try_auto_run.
|
|
# Plan is already at Execute/COMPLETE — skip to applying.
|
|
|
|
# Execute is already complete
|
|
|
|
# Trusted gated: execute -> apply is manual
|
|
assert plan.phase == PlanPhase.EXECUTE
|
|
assert plan.processing_state == ProcessingState.COMPLETE
|
|
assert service.should_auto_progress(plan) is False
|
|
|
|
# Manual apply
|
|
service.apply_plan(pid)
|
|
service.start_apply(pid)
|
|
plan = service.complete_apply(pid)
|
|
|
|
assert plan.phase == PlanPhase.APPLY
|
|
assert plan.processing_state == ProcessingState.APPLIED
|
|
assert plan.is_terminal
|
|
|
|
# Verify args persisted throughout lifecycle
|
|
assert plan.arguments["doc_types"] == "api-reference,architecture"
|
|
assert plan.arguments["output_dir"] == "docs/generated/"
|
|
assert len(plan.invariants) >= 1
|
|
|
|
print("trusted-doc-lifecycle-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_COMMANDS: dict[str, Any] = {
|
|
"context-policy-views": _context_policy_views,
|
|
"budget-enforcement": _budget_enforcement,
|
|
"trusted-profile-behavior": _trusted_profile_behavior,
|
|
"action-with-doc-args": _action_with_doc_args,
|
|
"doc-generation-sandbox": _doc_generation_sandbox,
|
|
"trusted-doc-lifecycle": _trusted_doc_lifecycle,
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} <command>")
|
|
print(f"Commands: {', '.join(sorted(_COMMANDS))}")
|
|
sys.exit(1)
|
|
|
|
cmd = sys.argv[1]
|
|
fn = _COMMANDS.get(cmd)
|
|
if fn is None:
|
|
print(f"Unknown command: {cmd}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
fn()
|
|
except Exception as exc:
|
|
print(f"FAIL [{cmd}]: {type(exc).__name__}: {exc}")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
sys.exit(1)
|