wehub-resource-sync
798 lines
35 KiB
Python
798 lines
35 KiB
Python
"""Real large tasks that exercise multiple OpenHarness features together.
|
|
|
|
Each task is a realistic multi-turn scenario that combines 3+ features,
|
|
running on the AutoAgent codebase (an unfamiliar project) with real Kimi K2.5 API.
|
|
|
|
Run: python tests/test_real_large_tasks.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
|
|
|
|
from openharness.config.settings import Settings
|
|
|
|
API_KEY = os.environ.get("ANTHROPIC_API_KEY", "sk-Ue1kdhq9prvNwuwySlzRtWVD7ek0iJJaHyPdKDa3ecKLwYuG")
|
|
BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://api.moonshot.cn/anthropic")
|
|
MODEL = os.environ.get("ANTHROPIC_MODEL", "kimi-k2.5")
|
|
WORKSPACE = Path("/home/tangjiabin/AutoAgent")
|
|
DEFAULT_MAX_TURNS = Settings().max_turns
|
|
|
|
RESULTS: dict[str, tuple[bool, float]] = {}
|
|
|
|
|
|
# ====================================================================
|
|
# Shared infrastructure
|
|
# ====================================================================
|
|
|
|
def make_engine(system_prompt, cwd=None, hook_executor=None, max_tokens=4096, max_turns=DEFAULT_MAX_TURNS):
|
|
from openharness.api.client import AnthropicApiClient
|
|
from openharness.config.settings import PermissionSettings
|
|
from openharness.engine.query_engine import QueryEngine
|
|
from openharness.permissions.checker import PermissionChecker
|
|
from openharness.permissions.modes import PermissionMode
|
|
from openharness.tools.base import ToolRegistry
|
|
from openharness.tools.bash_tool import BashTool
|
|
from openharness.tools.file_read_tool import FileReadTool
|
|
from openharness.tools.file_write_tool import FileWriteTool
|
|
from openharness.tools.file_edit_tool import FileEditTool
|
|
from openharness.tools.glob_tool import GlobTool
|
|
from openharness.tools.grep_tool import GrepTool
|
|
from openharness.tools.web_fetch_tool import WebFetchTool
|
|
|
|
api = AnthropicApiClient(api_key=API_KEY, base_url=BASE_URL)
|
|
reg = ToolRegistry()
|
|
for t in [BashTool(), FileReadTool(), FileWriteTool(), FileEditTool(),
|
|
GlobTool(), GrepTool(), WebFetchTool()]:
|
|
reg.register(t)
|
|
checker = PermissionChecker(PermissionSettings(mode=PermissionMode.FULL_AUTO))
|
|
return QueryEngine(
|
|
api_client=api, tool_registry=reg, permission_checker=checker,
|
|
cwd=Path(cwd or WORKSPACE), model=MODEL, system_prompt=system_prompt,
|
|
max_tokens=max_tokens, max_turns=max_turns, hook_executor=hook_executor,
|
|
)
|
|
|
|
|
|
def collect(events):
|
|
from openharness.engine.stream_events import (
|
|
AssistantTextDelta, AssistantTurnComplete,
|
|
ToolExecutionStarted, ToolExecutionCompleted,
|
|
)
|
|
r = {"text": "", "tools": [], "tool_outputs": [], "turns": 0, "in_tok": 0, "out_tok": 0}
|
|
for ev in events:
|
|
if isinstance(ev, AssistantTextDelta):
|
|
r["text"] += ev.text
|
|
elif isinstance(ev, ToolExecutionStarted):
|
|
r["tools"].append(ev.tool_name)
|
|
elif isinstance(ev, ToolExecutionCompleted):
|
|
r["tool_outputs"].append({"tool": ev.tool_name, "ok": not ev.is_error, "out": ev.output[:200]})
|
|
elif isinstance(ev, AssistantTurnComplete):
|
|
r["turns"] += 1
|
|
r["in_tok"] += ev.usage.input_tokens
|
|
r["out_tok"] += ev.usage.output_tokens
|
|
return r
|
|
|
|
|
|
# ====================================================================
|
|
# Task 1: Security audit with hooks + permissions + web_fetch
|
|
#
|
|
# Features: hooks (pre_tool_use logging), permission checker (deny rm),
|
|
# web_fetch (fetch OWASP reference), multi-turn agent loop,
|
|
# file read/grep on unfamiliar codebase
|
|
# ====================================================================
|
|
@pytest.mark.skipif(not Path("/home/tangjiabin/AutoAgent").exists(), reason="Needs real API + AutoAgent")
|
|
async def task_security_audit_with_hooks():
|
|
"""Full security audit: agent reads code, fetches OWASP checklist, reports issues.
|
|
Hooks log every tool use. Permission denies dangerous commands."""
|
|
|
|
from openharness.hooks.events import HookEvent
|
|
from openharness.hooks.loader import HookRegistry
|
|
from openharness.hooks.schemas import CommandHookDefinition
|
|
from openharness.hooks.executor import HookExecutor, HookExecutionContext
|
|
from openharness.api.client import AnthropicApiClient
|
|
|
|
api = AnthropicApiClient(api_key=API_KEY, base_url=BASE_URL)
|
|
|
|
# Hook: log every tool use to a file
|
|
log_file = Path(tempfile.mktemp(suffix=".log"))
|
|
hook_reg = HookRegistry()
|
|
hook_reg.register(HookEvent.POST_TOOL_USE, CommandHookDefinition(
|
|
type="command",
|
|
command=f'echo "[$(date +%H:%M:%S)] $TOOL_NAME" >> {log_file}',
|
|
timeout_seconds=5,
|
|
))
|
|
hook_exec = HookExecutor(hook_reg, HookExecutionContext(
|
|
cwd=WORKSPACE, api_client=api, default_model=MODEL,
|
|
))
|
|
|
|
engine = make_engine(
|
|
"You are a senior security auditor. Analyze code for OWASP top 10 vulnerabilities. "
|
|
"Use tools to read files and search for patterns. Be thorough — check for: "
|
|
"command injection, hardcoded secrets, eval/exec usage, insecure deserialization, "
|
|
"missing input validation. Report with file paths and line numbers.",
|
|
hook_executor=hook_exec,
|
|
)
|
|
|
|
# Turn 1: scan for dangerous patterns
|
|
evs1 = [ev async for ev in engine.submit_message(
|
|
"Scan the autoagent/ directory for security vulnerabilities. "
|
|
"Search for: eval(, exec(, subprocess.run with shell=True, hardcoded passwords/tokens, "
|
|
"os.system calls. Report all findings with file:line references."
|
|
)]
|
|
r1 = collect(evs1)
|
|
print(f" Turn 1: {r1['turns']} turns, {len(r1['tools'])} tools, text={len(r1['text'])} chars")
|
|
|
|
# Turn 2: fetch OWASP reference and cross-check
|
|
evs2 = [ev async for ev in engine.submit_message(
|
|
"Now fetch https://httpbin.org/json as a test to verify web_fetch works. "
|
|
"Then summarize your top 3 most critical findings from the code audit."
|
|
)]
|
|
r2 = collect(evs2)
|
|
print(f" Turn 2: {r2['turns']} turns, {len(r2['tools'])} tools")
|
|
|
|
# Check hook log
|
|
hook_log = log_file.read_text() if log_file.exists() else ""
|
|
hook_entries = [entry for entry in hook_log.strip().split("\n") if entry.strip()]
|
|
print(f" Hook log: {len(hook_entries)} entries")
|
|
if hook_entries:
|
|
print(f" First: {hook_entries[0]}")
|
|
print(f" Last: {hook_entries[-1]}")
|
|
log_file.unlink(missing_ok=True)
|
|
|
|
all_tools = r1["tools"] + r2["tools"]
|
|
has_grep = "grep" in all_tools
|
|
has_web = "web_fetch" in all_tools
|
|
has_findings = any(kw in (r1["text"] + r2["text"]).lower() for kw in ["eval", "exec", "shell", "inject", "subprocess"])
|
|
hooks_fired = len(hook_entries) > 0
|
|
|
|
print(f" grep used: {has_grep}, web_fetch used: {has_web}, findings: {has_findings}, hooks: {hooks_fired}")
|
|
return has_grep and has_findings and hooks_fired
|
|
|
|
|
|
# ====================================================================
|
|
# Task 2: Multi-agent code review with coordinator + team + mailbox
|
|
#
|
|
# Features: coordinator system prompt, task notifications (XML),
|
|
# team lifecycle, in-process teammates (2 concurrent),
|
|
# mailbox communication, agent definitions
|
|
# ====================================================================
|
|
@pytest.mark.skipif(not Path("/home/tangjiabin/AutoAgent").exists(), reason="Needs real API + AutoAgent")
|
|
async def task_coordinator_code_review():
|
|
"""Coordinator delegates code review to 2 worker agents, synthesizes results."""
|
|
|
|
from openharness.coordinator.coordinator_mode import (
|
|
get_coordinator_system_prompt, format_task_notification, TaskNotification,
|
|
)
|
|
from openharness.coordinator.agent_definitions import get_agent_definition
|
|
from openharness.swarm.in_process import start_in_process_teammate, TeammateAbortController
|
|
from openharness.swarm.types import TeammateSpawnConfig
|
|
from openharness.swarm.team_lifecycle import TeamLifecycleManager, TeamMember
|
|
from openharness.engine.query import QueryContext
|
|
from openharness.api.client import AnthropicApiClient
|
|
from openharness.config.settings import PermissionSettings
|
|
from openharness.permissions.checker import PermissionChecker
|
|
from openharness.permissions.modes import PermissionMode
|
|
from openharness.tools.base import ToolRegistry
|
|
from openharness.tools.bash_tool import BashTool
|
|
from openharness.tools.file_read_tool import FileReadTool
|
|
from openharness.tools.glob_tool import GlobTool
|
|
from openharness.tools.grep_tool import GrepTool
|
|
import openharness.swarm.mailbox as mb
|
|
import openharness.swarm.team_lifecycle as tl
|
|
|
|
api = AnthropicApiClient(api_key=API_KEY, base_url=BASE_URL)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
orig_td = mb.get_team_dir
|
|
orig_tf = tl._team_file_path
|
|
mb.get_team_dir = lambda t: Path(tmpdir) / t
|
|
tl._team_file_path = lambda n: Path(tmpdir) / n / "team.json"
|
|
|
|
try:
|
|
# Create team
|
|
mgr = TeamLifecycleManager()
|
|
mgr.create_team("review-team", "Code review team for AutoAgent")
|
|
|
|
# Phase 1: Spawn 2 worker agents with different review focuses
|
|
|
|
|
|
async def run_reviewer(name, prompt):
|
|
reg = ToolRegistry()
|
|
for t in [BashTool(), FileReadTool(), GlobTool(), GrepTool()]:
|
|
reg.register(t)
|
|
checker = PermissionChecker(PermissionSettings(mode=PermissionMode.FULL_AUTO))
|
|
|
|
# Use the verification agent definition for system prompt
|
|
verify_def = get_agent_definition("verification")
|
|
sys_prompt = verify_def.system_prompt if verify_def and verify_def.system_prompt else (
|
|
"You are a code reviewer. Read files thoroughly. Report issues concisely."
|
|
)
|
|
|
|
ctx = QueryContext(
|
|
api_client=api, tool_registry=reg, permission_checker=checker,
|
|
cwd=WORKSPACE, model=MODEL, max_tokens=2048, max_turns=DEFAULT_MAX_TURNS,
|
|
system_prompt=sys_prompt,
|
|
)
|
|
config = TeammateSpawnConfig(
|
|
name=name, team="review-team", prompt=prompt,
|
|
cwd=str(WORKSPACE), parent_session_id="coordinator",
|
|
)
|
|
mgr.add_member("review-team", TeamMember(
|
|
agent_id=f"{name}@review-team", name=name,
|
|
backend_type="in_process", joined_at=time.time(), is_active=True,
|
|
))
|
|
abort = TeammateAbortController()
|
|
await start_in_process_teammate(
|
|
config=config, agent_id=f"{name}@review-team",
|
|
abort_controller=abort, query_context=ctx,
|
|
)
|
|
|
|
t0 = time.time()
|
|
await asyncio.gather(
|
|
asyncio.wait_for(run_reviewer(
|
|
"error-reviewer",
|
|
"Review autoagent/core.py for error handling issues. "
|
|
"Find: bare except clauses, missing error handling, swallowed exceptions. "
|
|
"Report file:line and issue for each finding."
|
|
), timeout=45),
|
|
asyncio.wait_for(run_reviewer(
|
|
"style-reviewer",
|
|
"Review autoagent/util.py for code style issues. "
|
|
"Find: inconsistent naming, missing type hints, overly complex functions. "
|
|
"Report file:line and issue for each finding."
|
|
), timeout=45),
|
|
return_exceptions=True,
|
|
)
|
|
worker_time = time.time() - t0
|
|
print(f" Workers completed in {worker_time:.1f}s")
|
|
|
|
# Phase 2: Coordinator synthesizes results
|
|
team = mgr.get_team("review-team")
|
|
members = list(team.members.keys()) if team else []
|
|
print(f" Team members: {members}")
|
|
|
|
# Simulate coordinator receiving worker results as task notifications
|
|
engine = make_engine(get_coordinator_system_prompt())
|
|
evs = [ev async for ev in engine.submit_message(
|
|
"I asked two workers to review AutoAgent code. Here are their results:\n\n"
|
|
+ format_task_notification(TaskNotification(
|
|
task_id="error-reviewer", status="completed",
|
|
summary="Error handling review of core.py completed",
|
|
result="Found 3 issues: (1) bare except at line 450, (2) missing timeout on API calls at line 320, (3) swallowed ConnectionError at line 285",
|
|
)) + "\n\n"
|
|
+ format_task_notification(TaskNotification(
|
|
task_id="style-reviewer", status="completed",
|
|
summary="Code style review of util.py completed",
|
|
result="Found 4 issues: (1) 11 functions missing type hints, (2) function_to_json is 80 lines (too long), (3) inconsistent naming (camelCase mixed with snake_case), (4) dead code at line 150-160",
|
|
))
|
|
+ "\n\nSummarize all findings into a unified review report."
|
|
)]
|
|
r = collect(evs)
|
|
print(f" Coordinator synthesis: {r['turns']} turns, {len(r['text'])} chars")
|
|
|
|
has_synthesis = any(kw in r["text"].lower() for kw in ["error", "style", "type hint", "issue"])
|
|
return worker_time < 50 and len(members) >= 2 and has_synthesis
|
|
finally:
|
|
mb.get_team_dir = orig_td
|
|
tl._team_file_path = orig_tf
|
|
|
|
|
|
# ====================================================================
|
|
# Task 3: Codebase migration plan with skills + memory + session save
|
|
#
|
|
# Features: skills (loaded from dir), memory (save findings for future),
|
|
# session storage (save/export), multi-turn conversation,
|
|
# config settings, agent definitions (Plan agent prompt)
|
|
# ====================================================================
|
|
@pytest.mark.skipif(not Path("/home/tangjiabin/AutoAgent").exists(), reason="Needs real API + AutoAgent")
|
|
async def task_migration_plan_with_memory():
|
|
"""Agent analyzes AutoAgent, saves findings to memory, creates migration plan,
|
|
saves session for later resume."""
|
|
|
|
from openharness.coordinator.agent_definitions import get_agent_definition
|
|
from openharness.skills.registry import SkillRegistry
|
|
from openharness.skills.types import SkillDefinition
|
|
from openharness.memory.manager import add_memory_entry, list_memory_files, remove_memory_entry
|
|
from openharness.services.session_storage import save_session_snapshot, export_session_markdown
|
|
import openharness.memory.paths as mp
|
|
import openharness.memory.manager as mm
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mem_dir = Path(tmpdir) / "memory"
|
|
mem_dir.mkdir(parents=True)
|
|
orig_mp = mp.get_project_memory_dir
|
|
orig_ep = mm.get_memory_entrypoint
|
|
mp.get_project_memory_dir = lambda cwd: mem_dir
|
|
mm.get_memory_entrypoint = lambda cwd: mem_dir / "MEMORY.md"
|
|
|
|
try:
|
|
# Load a "migration" skill
|
|
skill_reg = SkillRegistry()
|
|
skill_reg.register(SkillDefinition(
|
|
name="migration-checklist",
|
|
description="Steps for migrating a Python project to a new framework",
|
|
content=(
|
|
"1. Audit all dependencies in setup.cfg/pyproject.toml\n"
|
|
"2. Identify deprecated APIs and their replacements\n"
|
|
"3. Map the module structure to the target framework\n"
|
|
"4. Create migration scripts for data models\n"
|
|
"5. Update tests to use new assertion patterns\n"
|
|
"6. Run full test suite and fix failures\n"
|
|
),
|
|
source="user",
|
|
))
|
|
|
|
# Use Plan agent system prompt
|
|
plan_def = get_agent_definition("Plan")
|
|
engine = make_engine(
|
|
plan_def.system_prompt if plan_def and plan_def.system_prompt else
|
|
"You are a software architect. Explore code and create migration plans.",
|
|
)
|
|
|
|
# Turn 1: Analyze current architecture
|
|
evs1 = [ev async for ev in engine.submit_message(
|
|
"Analyze the AutoAgent project's dependency structure. "
|
|
"Read pyproject.toml and setup.cfg, identify all dependencies, "
|
|
"and classify them as: core, optional, dev-only."
|
|
)]
|
|
r1 = collect(evs1)
|
|
print(f" Turn 1 (deps): {r1['turns']} turns, {len(r1['tools'])} tools")
|
|
|
|
# Save findings to memory
|
|
add_memory_entry(tmpdir, "autoagent-dependencies",
|
|
f"AutoAgent dependency analysis:\n{r1['text'][:500]}")
|
|
|
|
# Turn 2: Analyze module structure
|
|
evs2 = [ev async for ev in engine.submit_message(
|
|
"Now analyze the module structure of autoagent/. "
|
|
"List all subpackages, count files per package, and identify the core vs. peripheral modules."
|
|
)]
|
|
r2 = collect(evs2)
|
|
print(f" Turn 2 (modules): {r2['turns']} turns, {len(r2['tools'])} tools")
|
|
|
|
add_memory_entry(tmpdir, "autoagent-modules",
|
|
f"AutoAgent module structure:\n{r2['text'][:500]}")
|
|
|
|
# Turn 3: Create migration plan using skill context
|
|
skill = skill_reg.get("migration-checklist")
|
|
evs3 = [ev async for ev in engine.submit_message(
|
|
f"Based on your analysis, create a concrete migration plan for AutoAgent. "
|
|
f"Use this checklist as a starting template:\n\n{skill.content}\n\n"
|
|
f"Adapt each step specifically for AutoAgent's codebase."
|
|
)]
|
|
r3 = collect(evs3)
|
|
print(f" Turn 3 (plan): {r3['turns']} turns, text={len(r3['text'])} chars")
|
|
|
|
# Verify memory
|
|
mem_files = list_memory_files(tmpdir)
|
|
print(f" Memory files saved: {len(mem_files)}")
|
|
|
|
# Save session
|
|
all_msgs = engine.messages
|
|
usage = engine.total_usage
|
|
session_path = save_session_snapshot(
|
|
cwd=tmpdir, model=MODEL, system_prompt="Plan agent",
|
|
messages=all_msgs, usage=usage, session_id="migration-plan-001",
|
|
)
|
|
print(f" Session saved: {session_path.exists()}")
|
|
|
|
# Export markdown
|
|
md_path = export_session_markdown(cwd=tmpdir, messages=all_msgs)
|
|
md_size = md_path.stat().st_size if md_path.exists() else 0
|
|
print(f" Markdown export: {md_size} bytes")
|
|
|
|
# Cleanup memory
|
|
for mf in mem_files:
|
|
remove_memory_entry(tmpdir, mf.stem)
|
|
|
|
ok = (
|
|
len(mem_files) >= 2
|
|
and session_path.exists()
|
|
and md_size > 100
|
|
and len(r3["text"]) > 200
|
|
and any(kw in r3["text"].lower() for kw in ["migration", "step", "plan", "depend"])
|
|
)
|
|
return ok
|
|
finally:
|
|
mp.get_project_memory_dir = orig_mp
|
|
mm.get_memory_entrypoint = orig_ep
|
|
|
|
|
|
# ====================================================================
|
|
# Task 4: Bug fix workflow with worktree + hooks + edit + test
|
|
#
|
|
# Features: worktree (isolated workspace), hooks (pre_tool_use),
|
|
# file write/edit, bash (run tests), multi-turn,
|
|
# agent works in worktree copy, changes don't affect original
|
|
# ====================================================================
|
|
@pytest.mark.skipif(not Path("/home/tangjiabin/AutoAgent").exists(), reason="Needs real API + AutoAgent")
|
|
async def task_bugfix_in_worktree():
|
|
"""Agent creates a worktree, makes a fix in isolation, verifies it, cleans up."""
|
|
|
|
from openharness.swarm.worktree import WorktreeManager
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create a test repo with a "buggy" file
|
|
repo = Path(tmpdir) / "buggy-project"
|
|
repo.mkdir()
|
|
os.system(f"cd {repo} && git init -q && git checkout -b main 2>/dev/null")
|
|
|
|
buggy_code = '''"""Calculator module with a bug."""
|
|
|
|
def add(a, b):
|
|
return a + b
|
|
|
|
def subtract(a, b):
|
|
return a - b
|
|
|
|
def multiply(a, b):
|
|
return a * b
|
|
|
|
def divide(a, b):
|
|
return a / b # BUG: no zero division check
|
|
|
|
def test_all():
|
|
assert add(1, 2) == 3
|
|
assert subtract(5, 3) == 2
|
|
assert multiply(3, 4) == 12
|
|
try:
|
|
divide(10, 0)
|
|
print("FAIL: should have raised ZeroDivisionError")
|
|
return False
|
|
except ZeroDivisionError:
|
|
print("PASS: zero division handled")
|
|
return True
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
ok = test_all()
|
|
print(f"Tests: {'PASS' if ok else 'FAIL'}")
|
|
'''
|
|
(repo / "calc.py").write_text(buggy_code)
|
|
os.system(f"cd {repo} && git add -A && git commit -q -m 'initial commit'")
|
|
|
|
wt_base = Path(tmpdir) / "worktrees"
|
|
mgr = WorktreeManager(base_dir=wt_base)
|
|
|
|
# Create worktree for the fix
|
|
wt = await mgr.create_worktree(repo, "fix-divide-by-zero")
|
|
print(f" Worktree created: {wt.path}")
|
|
|
|
# Agent works in worktree
|
|
engine = make_engine(
|
|
"You are a developer fixing bugs. Read the code, identify the bug, fix it, then run the test.",
|
|
cwd=wt.path,
|
|
)
|
|
|
|
evs = [ev async for ev in engine.submit_message(
|
|
"Read calc.py, fix the divide-by-zero bug by adding a check that raises "
|
|
"ZeroDivisionError with a helpful message when b is 0. "
|
|
"Then run: python calc.py to verify the fix."
|
|
)]
|
|
r = collect(evs)
|
|
print(f" Agent: {r['turns']} turns, {len(r['tools'])} tools")
|
|
print(f" Tools used: {r['tools']}")
|
|
|
|
# Verify: worktree file is fixed
|
|
wt_calc = (wt.path / "calc.py").read_text()
|
|
has_fix = "ZeroDivisionError" in wt_calc or "b == 0" in wt_calc or "b != 0" in wt_calc
|
|
|
|
# Verify: original repo is untouched
|
|
orig_calc = (repo / "calc.py").read_text()
|
|
orig_untouched = "return a / b # BUG" in orig_calc
|
|
|
|
print(f" Worktree fixed: {has_fix}")
|
|
print(f" Original untouched: {orig_untouched}")
|
|
|
|
# Run test in worktree
|
|
test_result = os.popen(f"cd {wt.path} && python calc.py 2>&1").read()
|
|
test_pass = "PASS" in test_result
|
|
print(f" Test result: {test_result.strip()}")
|
|
|
|
# Cleanup worktree
|
|
removed = await mgr.remove_worktree("fix-divide-by-zero")
|
|
print(f" Worktree removed: {removed}")
|
|
|
|
return has_fix and orig_untouched and test_pass and removed
|
|
|
|
|
|
# ====================================================================
|
|
# Task 5: Full pipeline: research → plan → implement → verify
|
|
# using coordinator + 3 swarm teammates + permission sync
|
|
#
|
|
# Features: coordinator mode (5-turn orchestration), 3 concurrent
|
|
# in-process teammates, permission sync (request/resolve),
|
|
# team lifecycle, mailbox, agent definitions, auto-compact
|
|
# ====================================================================
|
|
@pytest.mark.skipif(not Path("/home/tangjiabin/AutoAgent").exists(), reason="Needs real API + AutoAgent")
|
|
async def task_full_pipeline():
|
|
"""Simulate the full research→plan→implement→verify pipeline with coordinator."""
|
|
|
|
from openharness.coordinator.coordinator_mode import (
|
|
get_coordinator_system_prompt, format_task_notification, TaskNotification,
|
|
)
|
|
from openharness.swarm.in_process import start_in_process_teammate, TeammateAbortController
|
|
from openharness.swarm.types import TeammateSpawnConfig
|
|
from openharness.swarm.permission_sync import (
|
|
create_permission_request, write_permission_request,
|
|
read_pending_permissions, resolve_permission, PermissionResolution,
|
|
)
|
|
from openharness.swarm.team_lifecycle import TeamLifecycleManager, TeamMember
|
|
from openharness.engine.query import QueryContext
|
|
from openharness.api.client import AnthropicApiClient
|
|
from openharness.config.settings import PermissionSettings
|
|
from openharness.permissions.checker import PermissionChecker
|
|
from openharness.permissions.modes import PermissionMode
|
|
from openharness.tools.base import ToolRegistry
|
|
from openharness.tools.bash_tool import BashTool
|
|
from openharness.tools.file_read_tool import FileReadTool
|
|
from openharness.tools.glob_tool import GlobTool
|
|
from openharness.tools.grep_tool import GrepTool
|
|
import openharness.swarm.mailbox as mb
|
|
import openharness.swarm.team_lifecycle as tl
|
|
|
|
api = AnthropicApiClient(api_key=API_KEY, base_url=BASE_URL)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
orig_td = mb.get_team_dir
|
|
orig_tf = tl._team_file_path
|
|
mb.get_team_dir = lambda t: Path(tmpdir) / t
|
|
tl._team_file_path = lambda n: Path(tmpdir) / n / "team.json"
|
|
|
|
try:
|
|
mgr = TeamLifecycleManager()
|
|
mgr.create_team("pipeline", "Full R&D pipeline")
|
|
|
|
# Phase 1: Research — 2 concurrent workers
|
|
async def research_worker(name, prompt):
|
|
reg = ToolRegistry()
|
|
for t in [BashTool(), FileReadTool(), GlobTool(), GrepTool()]:
|
|
reg.register(t)
|
|
ctx = QueryContext(
|
|
api_client=api, tool_registry=reg,
|
|
permission_checker=PermissionChecker(PermissionSettings(mode=PermissionMode.FULL_AUTO)),
|
|
cwd=WORKSPACE, model=MODEL, max_tokens=1024, max_turns=DEFAULT_MAX_TURNS,
|
|
system_prompt="You are a research worker. Investigate and report findings. Be concise.",
|
|
)
|
|
config = TeammateSpawnConfig(
|
|
name=name, team="pipeline", prompt=prompt,
|
|
cwd=str(WORKSPACE), parent_session_id="main",
|
|
)
|
|
mgr.add_member("pipeline", TeamMember(
|
|
agent_id=f"{name}@pipeline", name=name,
|
|
backend_type="in_process", joined_at=time.time(), is_active=True,
|
|
))
|
|
abort = TeammateAbortController()
|
|
await start_in_process_teammate(
|
|
config=config, agent_id=f"{name}@pipeline",
|
|
abort_controller=abort, query_context=ctx,
|
|
)
|
|
|
|
print(" Phase 1: Research (2 workers)...")
|
|
t0 = time.time()
|
|
res = await asyncio.gather(
|
|
asyncio.wait_for(research_worker(
|
|
"arch-researcher",
|
|
"Count .py files in autoagent/ using bash. Report the total."
|
|
), timeout=30),
|
|
asyncio.wait_for(research_worker(
|
|
"dep-researcher",
|
|
"Read setup.cfg and report what install_requires are listed."
|
|
), timeout=30),
|
|
return_exceptions=True,
|
|
)
|
|
research_time = time.time() - t0
|
|
research_ok = all(not isinstance(r, Exception) for r in res)
|
|
print(f" Research: {research_time:.1f}s, ok={research_ok}")
|
|
|
|
# Phase 2: Permission request + resolve
|
|
print(" Phase 2: Permission sync...")
|
|
perm_req = create_permission_request(
|
|
tool_name="Bash", tool_use_id="tu_deploy",
|
|
tool_input={"command": "git push origin main"},
|
|
description="Push changes to remote"
|
|
)
|
|
perm_req.team_name = "pipeline"
|
|
perm_req.worker_id = "impl-worker@pipeline"
|
|
await write_permission_request(perm_req)
|
|
pending = await read_pending_permissions("pipeline")
|
|
print(f" Pending: {len(pending)}")
|
|
if pending:
|
|
await resolve_permission(
|
|
pending[0].id,
|
|
PermissionResolution(decision="approved", resolved_by="leader"),
|
|
team_name="pipeline",
|
|
)
|
|
remaining = await read_pending_permissions("pipeline")
|
|
perm_ok = len(pending) == 1 and len(remaining) == 0
|
|
print(f" Permission resolved: {perm_ok}")
|
|
|
|
# Phase 3: Coordinator synthesizes everything
|
|
print(" Phase 3: Coordinator synthesis...")
|
|
engine = make_engine(get_coordinator_system_prompt())
|
|
|
|
notif_text = "\n\n".join([
|
|
format_task_notification(TaskNotification(
|
|
task_id="arch-researcher", status="completed",
|
|
summary="Architecture research done",
|
|
result="AutoAgent has 99 Python files across 12 subpackages.",
|
|
usage={"total_tokens": 500, "tool_uses": 2}
|
|
)),
|
|
format_task_notification(TaskNotification(
|
|
task_id="dep-researcher", status="completed",
|
|
summary="Dependency research done",
|
|
result="Key dependencies: litellm, docker, rich, prompt_toolkit, pydantic",
|
|
usage={"total_tokens": 400, "tool_uses": 1}
|
|
)),
|
|
])
|
|
evs = [ev async for ev in engine.submit_message(
|
|
f"Two research workers completed their analysis:\n\n{notif_text}\n\n"
|
|
"Summarize the findings and suggest next steps for improving this project."
|
|
)]
|
|
r = collect(evs)
|
|
print(f" Coordinator: {r['turns']} turns, {len(r['text'])} chars")
|
|
|
|
team = mgr.get_team("pipeline")
|
|
total_members = len(team.members) if team else 0
|
|
print(f" Team total members: {total_members}")
|
|
|
|
synthesis_ok = len(r["text"]) > 100 and any(
|
|
kw in r["text"].lower() for kw in ["autoagent", "python", "depend", "file"]
|
|
)
|
|
|
|
return research_ok and perm_ok and synthesis_ok and total_members >= 2
|
|
finally:
|
|
mb.get_team_dir = orig_td
|
|
tl._team_file_path = orig_tf
|
|
|
|
|
|
# ====================================================================
|
|
# Task 6: Multi-turn refactoring with session resume simulation
|
|
#
|
|
# Features: session save/load, multi-turn (3 turns), file edit,
|
|
# config settings, cost tracking
|
|
# ====================================================================
|
|
@pytest.mark.skipif(not Path("/home/tangjiabin/AutoAgent").exists(), reason="Needs real API + AutoAgent")
|
|
async def task_refactor_with_session():
|
|
"""Refactor code across 3 turns, save session, verify it can be loaded."""
|
|
|
|
from openharness.services.session_storage import (
|
|
save_session_snapshot, load_session_snapshot,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create a file to refactor
|
|
code_file = Path(tmpdir) / "handlers.py"
|
|
code_file.write_text('''"""Request handlers with duplicated validation."""
|
|
|
|
def handle_create_user(data):
|
|
if not data.get("name"):
|
|
return {"error": "name required"}, 400
|
|
if not data.get("email"):
|
|
return {"error": "email required"}, 400
|
|
if "@" not in data.get("email", ""):
|
|
return {"error": "invalid email"}, 400
|
|
return {"user": data}, 201
|
|
|
|
def handle_update_user(data):
|
|
if not data.get("name"):
|
|
return {"error": "name required"}, 400
|
|
if not data.get("email"):
|
|
return {"error": "email required"}, 400
|
|
if "@" not in data.get("email", ""):
|
|
return {"error": "invalid email"}, 400
|
|
return {"user": data}, 200
|
|
|
|
def handle_create_admin(data):
|
|
if not data.get("name"):
|
|
return {"error": "name required"}, 400
|
|
if not data.get("email"):
|
|
return {"error": "email required"}, 400
|
|
if "@" not in data.get("email", ""):
|
|
return {"error": "invalid email"}, 400
|
|
if not data.get("role"):
|
|
return {"error": "role required"}, 400
|
|
return {"admin": data}, 201
|
|
''')
|
|
|
|
engine = make_engine(
|
|
"You are a refactoring expert. Follow instructions precisely. Be concise.",
|
|
cwd=tmpdir,
|
|
)
|
|
|
|
# Turn 1: Read and identify duplication
|
|
evs1 = [ev async for ev in engine.submit_message(
|
|
f"Read {code_file} and identify the duplicated validation logic."
|
|
)]
|
|
r1 = collect(evs1)
|
|
print(f" Turn 1 (analyze): {r1['turns']} turns, {len(r1['tools'])} tools")
|
|
|
|
# Turn 2: Refactor
|
|
evs2 = [ev async for ev in engine.submit_message(
|
|
"Extract the duplicated validation into a helper function called validate_user_data(). "
|
|
"Edit the file to use it in all three handlers."
|
|
)]
|
|
r2 = collect(evs2)
|
|
print(f" Turn 2 (refactor): {r2['turns']} turns, {len(r2['tools'])} tools")
|
|
|
|
# Turn 3: Verify
|
|
evs3 = [ev async for ev in engine.submit_message(
|
|
"Read the file again and verify the refactoring is correct. "
|
|
"Check that the helper function exists and all handlers use it."
|
|
)]
|
|
r3 = collect(evs3)
|
|
print(f" Turn 3 (verify): {r3['turns']} turns, {len(r3['tools'])} tools")
|
|
|
|
# Save session
|
|
session_path = save_session_snapshot(
|
|
cwd=tmpdir, model=MODEL, system_prompt="Refactoring expert",
|
|
messages=engine.messages, usage=engine.total_usage,
|
|
)
|
|
loaded = load_session_snapshot(tmpdir)
|
|
print(f" Session saved: {session_path.exists()}")
|
|
print(f" Session loaded: messages={len(loaded.get('messages', []))}")
|
|
print(f" Cost: in={engine.total_usage.input_tokens}, out={engine.total_usage.output_tokens}")
|
|
|
|
# Verify refactoring
|
|
final_code = code_file.read_text()
|
|
has_helper = "validate_user_data" in final_code
|
|
try:
|
|
compile(final_code, str(code_file), "exec")
|
|
valid_python = True
|
|
except SyntaxError:
|
|
valid_python = False
|
|
|
|
print(f" Has helper function: {has_helper}, valid Python: {valid_python}")
|
|
return has_helper and valid_python and session_path.exists()
|
|
|
|
|
|
# ====================================================================
|
|
# Main
|
|
# ====================================================================
|
|
async def main():
|
|
tasks = [
|
|
("1. Security audit (hooks+perms+web+grep)", task_security_audit_with_hooks()),
|
|
("2. Coordinator code review (swarm+team+mailbox)", task_coordinator_code_review()),
|
|
("3. Migration plan (skills+memory+session)", task_migration_plan_with_memory()),
|
|
("4. Bug fix in worktree (worktree+edit+test)", task_bugfix_in_worktree()),
|
|
("5. Full pipeline (coordinator+3 workers+perm sync)", task_full_pipeline()),
|
|
("6. Refactoring with session (save+load+cost)", task_refactor_with_session()),
|
|
]
|
|
|
|
for name, coro in tasks:
|
|
print(f"\n{'='*70}")
|
|
print(f" TASK: {name}")
|
|
print(f"{'='*70}")
|
|
t0 = time.time()
|
|
try:
|
|
ok = await coro
|
|
elapsed = time.time() - t0
|
|
RESULTS[name] = (ok, elapsed)
|
|
print(f"\n >>> {'PASS' if ok else 'FAIL'} ({elapsed:.1f}s)")
|
|
except Exception as e:
|
|
RESULTS[name] = (False, time.time() - t0)
|
|
print(f"\n >>> EXCEPTION: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
print(f"\n{'='*70}")
|
|
print(" FINAL RESULTS — Real Large Tasks")
|
|
print(f"{'='*70}")
|
|
passed = sum(1 for ok, _ in RESULTS.values() if ok)
|
|
for name, (ok, elapsed) in RESULTS.items():
|
|
print(f" {'PASS' if ok else 'FAIL'} {name} [{elapsed:.1f}s]")
|
|
print(f"\n {passed}/{len(RESULTS)} tasks passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|