feat: add opencode browser click challenge harness
- Challenge prompt requiring workflow/deployment/run evidence - CLI harness (run_opencode_trials.py) for running N agent trials - Classification: success, workflow_not_used, run_failed, timeout, parse_error, unknown - 8 unit tests covering build/parse/classify/path logic - README with usage docs and optional Playwright MCP attachment - Evidence index and roadmap updated - Plan archived to historical/
This commit is contained in:
@@ -0,0 +1,63 @@
|
||||
# Opencode Browser Click Challenge Harness
|
||||
|
||||
This harness runs agent trials against the browser-click workflow challenge.
|
||||
It is evidence tooling, not product runtime code.
|
||||
|
||||
The deterministic workflow example is:
|
||||
|
||||
```text
|
||||
examples/browser_click_workflow/
|
||||
```
|
||||
|
||||
## One Trial
|
||||
|
||||
From the repository root:
|
||||
|
||||
```powershell
|
||||
uv run python examples/agent_challenges/browser_click_challenge/run_opencode_trials.py `
|
||||
--model opencode/mimo-v2.5-free `
|
||||
--variant high `
|
||||
--trials 1
|
||||
```
|
||||
|
||||
Results are written to:
|
||||
|
||||
```text
|
||||
examples/agent_challenges/browser_click_challenge/results/
|
||||
```
|
||||
|
||||
## Optional Playwright MCP Attachment
|
||||
|
||||
If you want the agent to have browser-control tools, pass:
|
||||
|
||||
```powershell
|
||||
--attach http://127.0.0.1:4096
|
||||
```
|
||||
|
||||
Start that MCP/tool endpoint separately. For example, one possible MCP server
|
||||
command is:
|
||||
|
||||
```json
|
||||
{
|
||||
"command": "npx",
|
||||
"args": ["-y", "@playwright/mcp@latest"]
|
||||
}
|
||||
```
|
||||
|
||||
The baseline challenge does not require Playwright MCP. The score is based on
|
||||
whether the agent used the workflow product path and produced the expected
|
||||
workflow output.
|
||||
|
||||
## Classification
|
||||
|
||||
Each trial is classified as one of:
|
||||
|
||||
- `success`: output shows workflow usage and before/after clicked states.
|
||||
- `workflow_not_used`: output appears to solve the task without `wf`,
|
||||
`wf-rpc-server`, deployment, or run evidence.
|
||||
- `run_failed`: output includes workflow usage but reports a failure.
|
||||
- `timeout`: the opencode process exceeded the configured timeout.
|
||||
- `parse_error`: the harness could not read opencode JSON/JSONL output.
|
||||
- `unknown`: no clear success or failure signal was found.
|
||||
|
||||
Committed tests cover harness logic only. They do not invoke opencode.
|
||||
@@ -0,0 +1,30 @@
|
||||
# Browser Click Workflow Challenge
|
||||
|
||||
Build and successfully run a workflow that:
|
||||
|
||||
1. Opens a browser page or local web page with a visible button.
|
||||
2. Waits for a human click or performs a clearly simulated click.
|
||||
3. Captures a before snapshot and an after snapshot.
|
||||
4. Returns both snapshots as workflow output.
|
||||
|
||||
Use this repository's workflow product path. That means you should use the
|
||||
`wf` CLI and/or `wf-rpc-server`, create or reuse a workflow deployment, and run
|
||||
the deployment through the workflow API. Do not solve the challenge with only a
|
||||
standalone Playwright/Python script.
|
||||
|
||||
The repository already includes a deterministic source example at:
|
||||
|
||||
```text
|
||||
examples/browser_click_workflow/
|
||||
```
|
||||
|
||||
You may inspect and use it. A successful final answer must include:
|
||||
|
||||
- the commands you ran,
|
||||
- the deployment id,
|
||||
- the run id if one was produced,
|
||||
- evidence that `before.clicked` is `false`,
|
||||
- evidence that `after.clicked` is `true`,
|
||||
- whether any server/browser process remains running.
|
||||
|
||||
If something fails, report the exact command and error instead of hiding it.
|
||||
@@ -0,0 +1,2 @@
|
||||
*
|
||||
!.gitignore
|
||||
@@ -0,0 +1,254 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import time
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
Classification = Literal[
|
||||
"success",
|
||||
"workflow_not_used",
|
||||
"run_failed",
|
||||
"timeout",
|
||||
"parse_error",
|
||||
"unknown",
|
||||
]
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
CHALLENGE_DIR = Path(__file__).resolve().parent
|
||||
DEFAULT_PROMPT = CHALLENGE_DIR / "prompt.md"
|
||||
DEFAULT_RESULTS_DIR = CHALLENGE_DIR / "results"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TrialConfig:
|
||||
model: str
|
||||
variant: str
|
||||
prompt_path: Path
|
||||
attach_url: str | None
|
||||
timeout_seconds: int
|
||||
|
||||
|
||||
def build_opencode_command(config: TrialConfig) -> list[str]:
|
||||
prompt_text = config.prompt_path.read_text(encoding="utf-8")
|
||||
command = [
|
||||
"opencode",
|
||||
"run",
|
||||
]
|
||||
if config.attach_url is not None:
|
||||
command.extend(["--attach", config.attach_url])
|
||||
command.extend(
|
||||
[
|
||||
prompt_text,
|
||||
"--format",
|
||||
"json",
|
||||
"--model",
|
||||
config.model,
|
||||
"--variant",
|
||||
config.variant,
|
||||
]
|
||||
)
|
||||
return command
|
||||
|
||||
|
||||
def parse_opencode_output(stdout: str) -> dict[str, Any]:
|
||||
text = stdout.strip()
|
||||
if not text:
|
||||
raise ValueError("opencode produced no JSON output")
|
||||
|
||||
try:
|
||||
parsed = json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
parsed = _parse_jsonl_tail(text)
|
||||
|
||||
if not isinstance(parsed, dict):
|
||||
raise ValueError("opencode output was not a JSON object")
|
||||
return parsed
|
||||
|
||||
|
||||
def _parse_jsonl_tail(text: str) -> dict[str, Any]:
|
||||
last_error: json.JSONDecodeError | None = None
|
||||
for line in reversed(text.splitlines()):
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
try:
|
||||
parsed = json.loads(stripped)
|
||||
except json.JSONDecodeError as exc:
|
||||
last_error = exc
|
||||
continue
|
||||
if isinstance(parsed, dict):
|
||||
return parsed
|
||||
if last_error is not None:
|
||||
raise last_error
|
||||
raise ValueError("opencode output did not contain JSON lines")
|
||||
|
||||
|
||||
def classify_output(text: str) -> Classification:
|
||||
lowered = text.lower()
|
||||
workflow_markers = [
|
||||
"wf ",
|
||||
"wf-rpc-server",
|
||||
"deployment",
|
||||
"run id",
|
||||
"run_",
|
||||
]
|
||||
used_workflow = any(marker in lowered for marker in workflow_markers)
|
||||
failed = any(
|
||||
marker in lowered
|
||||
for marker in [
|
||||
"error:",
|
||||
"failed",
|
||||
"traceback",
|
||||
"exception",
|
||||
"validation failed",
|
||||
]
|
||||
)
|
||||
before_false = (
|
||||
"before.clicked is false" in lowered
|
||||
or '"before"' in lowered
|
||||
and '"clicked": false' in lowered
|
||||
)
|
||||
after_true = (
|
||||
"after.clicked is true" in lowered
|
||||
or '"after"' in lowered
|
||||
and '"clicked": true' in lowered
|
||||
)
|
||||
|
||||
if used_workflow and before_false and after_true and not failed:
|
||||
return "success"
|
||||
if used_workflow and failed:
|
||||
return "run_failed"
|
||||
if not used_workflow and (before_false or after_true or "playwright" in lowered):
|
||||
return "workflow_not_used"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def trial_output_path(results_dir: Path, *, model: str, index: int) -> Path:
|
||||
safe_model = model.replace("/", "_").replace(":", "_")
|
||||
return results_dir / f"{safe_model}-trial-{index:03d}.json"
|
||||
|
||||
|
||||
def run_trial(config: TrialConfig, *, index: int, results_dir: Path) -> dict[str, Any]:
|
||||
command = build_opencode_command(config)
|
||||
started = time.monotonic()
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
command,
|
||||
cwd=ROOT,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
timeout=config.timeout_seconds,
|
||||
check=False,
|
||||
)
|
||||
duration_seconds = time.monotonic() - started
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
payload = {
|
||||
"index": index,
|
||||
"config": _jsonable_config(config),
|
||||
"command": command,
|
||||
"classification": "timeout",
|
||||
"duration_seconds": config.timeout_seconds,
|
||||
"returncode": None,
|
||||
"stdout": exc.stdout or "",
|
||||
"stderr": exc.stderr or "",
|
||||
"parsed": None,
|
||||
}
|
||||
_write_trial_result(results_dir, config=config, index=index, payload=payload)
|
||||
return payload
|
||||
|
||||
parsed: dict[str, Any] | None
|
||||
try:
|
||||
parsed = parse_opencode_output(completed.stdout)
|
||||
text = _result_text(parsed)
|
||||
classification = classify_output(text)
|
||||
except Exception:
|
||||
parsed = None
|
||||
classification = "parse_error"
|
||||
|
||||
payload = {
|
||||
"index": index,
|
||||
"config": _jsonable_config(config),
|
||||
"command": command,
|
||||
"classification": classification,
|
||||
"duration_seconds": duration_seconds,
|
||||
"returncode": completed.returncode,
|
||||
"stdout": completed.stdout,
|
||||
"stderr": completed.stderr,
|
||||
"parsed": parsed,
|
||||
}
|
||||
_write_trial_result(results_dir, config=config, index=index, payload=payload)
|
||||
return payload
|
||||
|
||||
|
||||
def _result_text(parsed: dict[str, Any]) -> str:
|
||||
for key in ("text", "message", "content", "output"):
|
||||
value = parsed.get(key)
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
return json.dumps(parsed, sort_keys=True)
|
||||
|
||||
|
||||
def _jsonable_config(config: TrialConfig) -> dict[str, Any]:
|
||||
payload = asdict(config)
|
||||
payload["prompt_path"] = str(config.prompt_path)
|
||||
return payload
|
||||
|
||||
|
||||
def _write_trial_result(
|
||||
results_dir: Path,
|
||||
*,
|
||||
config: TrialConfig,
|
||||
index: int,
|
||||
payload: dict[str, Any],
|
||||
) -> None:
|
||||
results_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = trial_output_path(results_dir, model=config.model, index=index)
|
||||
path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--model", default="opencode/mimo-v2.5-free")
|
||||
parser.add_argument("--variant", default="high")
|
||||
parser.add_argument("--trials", type=int, default=1)
|
||||
parser.add_argument("--timeout-seconds", type=int, default=600)
|
||||
parser.add_argument("--attach", dest="attach_url", default=None)
|
||||
parser.add_argument("--prompt", type=Path, default=DEFAULT_PROMPT)
|
||||
parser.add_argument("--results-dir", type=Path, default=DEFAULT_RESULTS_DIR)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if args.trials < 1:
|
||||
parser.error("--trials must be >= 1")
|
||||
|
||||
config = TrialConfig(
|
||||
model=args.model,
|
||||
variant=args.variant,
|
||||
prompt_path=args.prompt,
|
||||
attach_url=args.attach_url,
|
||||
timeout_seconds=args.timeout_seconds,
|
||||
)
|
||||
|
||||
summaries: list[dict[str, Any]] = []
|
||||
for index in range(1, args.trials + 1):
|
||||
result = run_trial(config, index=index, results_dir=args.results_dir)
|
||||
summaries.append(
|
||||
{
|
||||
"index": index,
|
||||
"classification": result["classification"],
|
||||
"returncode": result["returncode"],
|
||||
"duration_seconds": round(float(result["duration_seconds"]), 3),
|
||||
}
|
||||
)
|
||||
print(json.dumps(summaries[-1], sort_keys=True))
|
||||
|
||||
success_count = sum(1 for item in summaries if item["classification"] == "success")
|
||||
print(json.dumps({"success_count": success_count, "trial_count": len(summaries)}))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user