diff --git a/scripts/convert_team_to_coop.py b/scripts/convert_team_to_coop.py new file mode 100644 index 00000000..469385f3 --- /dev/null +++ b/scripts/convert_team_to_coop.py @@ -0,0 +1,640 @@ +"""Convert cooperbench team-run logs into the coop layout from CooperData PR #98. + +Source: ``logs//team////`` (2-agent lead/member runs). +Target: ``//coop///f1_f2/`` matching the schema in +``cooperbench/CooperData`` PR #98 (``convert_swechat.write_run``). + +Mapping: +- agent1 = team lead (asymmetric: holds integration responsibility). +- agent2 = team member (single-feature implementer). +- Inter-agent ``conversation.json`` is rebuilt from ``task_log.json`` events + (create/claim/update task events with timestamps) plus ``coop-send`` / + ``coop-broadcast`` invocations extracted from each agent's bash tool calls. +- ``result.json`` keeps the cooperbench shape and folds team-specific extras + (``team_role``, ``metrics``, ``team_features``) into a ``team`` provenance + block — analogous to PR #98's ``swechat`` block. +- ``eval.json`` is marked ``verified: true`` because cooperbench runs held-out + tests (unlike SWE-chat). +- ``metadata.json`` ships the full ``tasks.json`` and ``task_log.json`` so the + coordination state is fully reconstructible. + +Only 2-agent pairs are converted; >2-agent dirs are skipped with a warning. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import re +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +logger = logging.getLogger("convert_team_to_coop") + +SOURCE = "cooperbench-team" +TASK_NAME = "cooperbench_team" +FEATURES = [1, 2] +FEATURE_STR = "f1_f2" + +# Extract Redis-backed CLI invocations from bash commands inside tool calls. +# Conservative: matches ``coop-send`` / ``coop-broadcast`` with a -m/--message +# value, which is how mini_swe agents send free-form messages in team mode. +_COOP_SEND_RE = re.compile( + r"coop-(send|broadcast)\s+(?:--to\s+(\S+)\s+)?(?:-m|--message)\s+" + r"""(?P['"])(?P.*?)(?P=q)""", + re.DOTALL, +) + + +def _read_json(path: Path) -> Any: + if not path.exists(): + return None + try: + return json.loads(path.read_text()) + except json.JSONDecodeError: + logger.warning("could not parse %s", path) + return None + + +def _dump(path: Path, obj: Any) -> None: + path.write_text(json.dumps(obj, indent=2, default=str)) + + +def _to_iso(ts: float | str | None) -> str | None: + if ts is None: + return None + if isinstance(ts, str): + return ts + try: + return datetime.fromtimestamp(float(ts), tz=timezone.utc).isoformat() + except (TypeError, ValueError, OSError): + return None + + +def _ts_float(ts: Any) -> float | None: + if ts is None: + return None + if isinstance(ts, (int, float)): + return float(ts) + try: + return datetime.fromisoformat(str(ts).replace("Z", "+00:00")).timestamp() + except ValueError: + return None + + +def _agent_index_by_role(team_result: dict[str, Any]) -> dict[str, int]: + """Return {source_agent_id: 1|2} mapping lead→1, member→2. + + Falls back to natural-sort order of agent ids if roles are absent. + """ + agents = team_result.get("agents") or {} + lead = team_result.get("lead_agent") + if lead and lead in agents: + members = [a for a in agents if a != lead] + if len(members) == 1: + return {lead: 1, members[0]: 2} + if len(agents) == 2: + ordered = sorted(agents.keys()) + return {ordered[0]: 1, ordered[1]: 2} + return {} + + +def _feature_id_for(agent_summary: dict[str, Any]) -> int | None: + fid = agent_summary.get("feature_id") + return int(fid) if fid is not None else None + + +def _flatten_messages(full_traj: dict[str, Any]) -> list[dict[str, Any]]: + """All solver-segment messages concatenated (litellm shape preserved). + + mini_swe alternates ``solver`` and ``summarizer`` segments; summarizer + output is a condensed view of the prior solver, so skipping them and + concatenating solver messages yields a faithful turn-by-turn log. + The system message from the first segment is kept once; subsequent + segments' system messages are dropped. + + Falls back to the top-level ``messages`` field for trajectories that + don't expose a segments list (codex / claude-code / other one-shot + adapters write the cooperbench traj envelope directly). + """ + segments = full_traj.get("segments") or [] + solver_segments = [s for s in segments if s.get("kind") == "solver"] + if not solver_segments: + return full_traj.get("messages") or [] + + out: list[dict[str, Any]] = [] + for i, seg in enumerate(solver_segments): + for j, m in enumerate(seg.get("messages", [])): + if j == 0 and m.get("role") == "system" and i > 0: + continue + out.append(m) + return out + + +def _load_agent_trajectory(pair_dir: Path, src_id: str) -> dict[str, Any]: + """Prefer mini_swe's segmented ``_full_traj.json``; fall back to + the plain cooperbench traj envelope written by codex / claude-code.""" + full = pair_dir / f"{src_id}_full_traj.json" + if full.exists(): + return _read_json(full) or {} + return _read_json(pair_dir / f"{src_id}_traj.json") or {} + + +def _load_sent_messages(pair_dir: Path, src_id: str) -> list[dict[str, Any]]: + """Parse cooperbench's structured ``_sent.jsonl`` send log.""" + path = pair_dir / f"{src_id}_sent.jsonl" + if not path.exists(): + return [] + out: list[dict[str, Any]] = [] + for line in path.read_text().splitlines(): + line = line.strip() + if not line: + continue + try: + out.append(json.loads(line)) + except json.JSONDecodeError: + logger.warning("malformed sent.jsonl line in %s", path) + return out + + +def _extract_coop_sends( + traj: dict[str, Any], + *, + from_agent: str, +) -> list[dict[str, Any]]: + """Pull (from, to, message, ts) tuples from ``coop-send/-broadcast`` + bash invocations inside the agent's tool calls. + + Team mode has no structured conversation log; agents message peers by + running the Redis-backed CLI. We surface that traffic so the coop-style + ``conversation.json`` isn't empty. + """ + out: list[dict[str, Any]] = [] + msgs = _flatten_messages(traj) + for m in msgs: + if m.get("role") != "assistant": + continue + ts = ((m.get("extra") or {}).get("timestamp")) or 0.0 + for tc in m.get("tool_calls") or []: + args_raw = ((tc.get("function") or {}).get("arguments")) or "" + try: + cmd = json.loads(args_raw).get("command", "") if args_raw else "" + except (json.JSONDecodeError, AttributeError): + cmd = args_raw if isinstance(args_raw, str) else "" + for match in _COOP_SEND_RE.finditer(cmd or ""): + kind = match.group(1) + to = match.group(2) or "*" # broadcast → "*" + text = match.group("msg") + out.append( + { + "from": from_agent, + "to": to, + "message": text, + "timestamp": _to_iso(ts), + "channel": "coop-cli", + "kind": kind, + } + ) + return out + + +def _conversation_from_task_log( + task_log: list[dict[str, Any]], + tasks: list[dict[str, Any]], + role_by_id: dict[str, int], +) -> list[dict[str, Any]]: + """Each task event becomes a typed broadcast on the task-log channel. + + Schema notes (intentionally drops `to` from the legacy P2P shape because + these events aren't directed messages — they're broadcasts visible to + every peer reading the shared task list): + + - ``sender`` — who emitted the event (an agent id, or "bench-runner" + for system-emitted ``create`` events). + - ``sender_role`` — ``"system"`` for bench-runner, ``"agent"`` for any + peer-emitted claim/update. Lets a downstream consumer condition on + sender type without string-matching the sender id. + - ``owner`` — only present on ``create`` events; the pre-assigned task + owner. This is the field the old ``to`` was conflating into a + recipient slot, which it never was. + - ``channel`` — always ``"task-log"``; reinforces that this is a + broadcast on a shared log, not P2P traffic. + - No ``to``. An ``update`` from the owner to themselves had to be + rendered as a self-loop under the old shape; here it's just a log + event with no recipient, which is semantically honest. + """ + title_by_task = {t.get("id"): t.get("title") for t in (tasks or [])} + assigned_by_task = { + t.get("id"): (t.get("metadata") or {}).get("assigned_to") or t.get("owner") for t in (tasks or []) + } + + out: list[dict[str, Any]] = [] + for ev in task_log or []: + actor = ev.get("by") or "" + tid = ev.get("task_id") + kind = ev.get("kind") + ts = _to_iso(ev.get("ts")) + title = ev.get("title") or title_by_task.get(tid, "") + assignee = assigned_by_task.get(tid) + + if kind == "create": + sender = actor or "bench-runner" + sender_role = "system" if sender == "bench-runner" or not actor else "agent" + msg = f"[task-create #{tid}] {title}" + if assignee: + msg += f" → assigned to {assignee}" + entry = { + "sender": sender, + "sender_role": sender_role, + "owner": assignee or None, + "message": msg, + "timestamp": ts, + "channel": "task-log", + "kind": kind, + "task_id": tid, + } + elif kind == "claim": + sender = actor or "bench-runner" + msg = f"[task-claim #{tid}] claimed: {title}" + entry = { + "sender": sender, + "sender_role": "agent" if actor else "system", + "message": msg, + "timestamp": ts, + "channel": "task-log", + "kind": kind, + "task_id": tid, + "feature_id": role_by_id.get(actor) if actor in role_by_id else None, + } + elif kind == "update": + sender = actor or "bench-runner" + status = ev.get("status") or "" + note = ev.get("note") or "" + msg = f"[task-update #{tid}] status={status}" + (f" — {note}" if note else "") + entry = { + "sender": sender, + "sender_role": "agent" if actor else "system", + "message": msg, + "timestamp": ts, + "channel": "task-log", + "kind": kind, + "task_id": tid, + "status": status or None, + "feature_id": role_by_id.get(actor) if actor in role_by_id else None, + } + else: + continue + + out.append({k: v for k, v in entry.items() if v is not None}) + return out + + +def _agent_traj_doc( + repo: str, + task_id: int, + feature_id: int, + coop_agent_id: str, + model: str, + status: str, + steps: int, + cost: float, + messages: list[dict[str, Any]], + team_block: dict[str, Any], +) -> dict[str, Any]: + """PR #98 agent{N}_traj.json shape + a ``team`` provenance block.""" + return { + "repo": repo, + "task_id": task_id, + "feature_id": feature_id, + "agent_id": coop_agent_id, + "model": model, + "status": status, + "cost": float(cost or 0.0), + "steps": int(steps or 0), + "messages": messages, + "team": team_block, + } + + +def convert_pair( + pair_dir: Path, + out_root: Path, + run_name: str, + *, + source_run_name: str, +) -> dict[str, Any] | None: + """Convert one team pair directory. Returns a summary row, or None on skip.""" + result = _read_json(pair_dir / "result.json") + if not result: + logger.warning("skip %s: no result.json", pair_dir) + return None + + agents = result.get("agents") or {} + if len(agents) != 2: + logger.warning("skip %s: only 2-agent pairs supported (got %d)", pair_dir, len(agents)) + return None + + role_by_id = _agent_index_by_role(result) + if set(role_by_id.values()) != {1, 2}: + logger.warning("skip %s: could not assign agent1/agent2 mapping", pair_dir) + return None + + inv_role = {idx: src for src, idx in role_by_id.items()} # {1: 'agent1', 2: 'agent2'} + src_agent1 = inv_role[1] + src_agent2 = inv_role[2] + + repo = result.get("repo") or pair_dir.parents[1].name + task_id = int(result.get("task_id") or 0) + # Coop SLOTS are always (1, 2), but each task in cooperbench has + # many feature pairs (f1_f3, f1_f4, f2_f3, …). Use the source pair's + # feature ids in the dir name so they don't collide. + src_f1 = _feature_id_for(agents.get(src_agent1) or {}) + src_f2 = _feature_id_for(agents.get(src_agent2) or {}) + feature_dir = f"f{src_f1}_f{src_f2}" if src_f1 and src_f2 else FEATURE_STR + out_dir = out_root / run_name / "coop" / repo / str(task_id) / feature_dir + out_dir.mkdir(parents=True, exist_ok=True) + + eval_doc = _read_json(pair_dir / "eval.json") or {} + tasks_doc = _read_json(pair_dir / "tasks.json") or [] + task_log_doc = _read_json(pair_dir / "task_log.json") or [] + + team_block_common = { + "source_run": source_run_name, + "source_pair_dir": str(pair_dir), + "lead_agent": result.get("lead_agent"), + "team_features": result.get("team_features") or {}, + "metrics": result.get("metrics") or {}, + "setting": result.get("setting") or "team", + "agent_framework": result.get("agent_framework"), + "model": result.get("model"), + "duration_seconds": result.get("duration_seconds"), + "run_id": result.get("run_id"), + } + + # Per-agent trajectories + patches. + convo: list[dict[str, Any]] = [] + agent_docs: dict[int, dict[str, Any]] = {} + for idx, src_id in inv_role.items(): + summary = agents.get(src_id) or {} + fid = _feature_id_for(summary) + # mini_swe writes per-feature_id patch files; codex/team also follow that. + patch_path = pair_dir / f"agent{fid}.patch" if fid else None + patch_text = patch_path.read_text() if patch_path and patch_path.exists() else "" + (out_dir / f"agent{idx}.patch").write_text(patch_text) + + full_traj = _load_agent_trajectory(pair_dir, src_id) + messages = _flatten_messages(full_traj) + + agent_doc = _agent_traj_doc( + repo=repo, + task_id=task_id, + feature_id=fid or idx, + coop_agent_id=f"agent{idx}", + model=result.get("model") or "", + status=summary.get("status") or "Unknown", + steps=summary.get("steps") or 0, + cost=summary.get("cost") or 0.0, + messages=messages, + team_block={ + **team_block_common, + "source_agent_id": src_id, + "team_role": summary.get("team_role"), + "source_feature_id": fid, + "input_tokens": summary.get("input_tokens", 0), + "output_tokens": summary.get("output_tokens", 0), + "cache_read_tokens": summary.get("cache_read_tokens", 0), + "cache_write_tokens": summary.get("cache_write_tokens", 0), + "patch_lines": summary.get("patch_lines", 0), + }, + ) + # Preserve the full mini_swe segments / info losslessly. + agent_doc["mini_swe_segments"] = full_traj.get("segments") + agent_doc["mini_swe_info"] = full_traj.get("info") + agent_doc["mini_swe_trajectory_format"] = full_traj.get("trajectory_format") + _dump(out_dir / f"agent{idx}_traj.json", agent_doc) + agent_docs[idx] = agent_doc + + convo.extend(_extract_coop_sends(full_traj, from_agent=f"agent{idx}")) + # Codex / claude-code adapters log structured sends to + # ``_sent.jsonl``; map ``to`` to the coop slot id. + for sent in _load_sent_messages(pair_dir, src_id): + to_src = sent.get("to") or "" + to_idx = role_by_id.get(to_src) + convo.append( + { + "from": f"agent{idx}", + "to": f"agent{to_idx}" if to_idx else (to_src or "*"), + "message": sent.get("content") or "", + "timestamp": sent.get("timestamp_iso") or _to_iso(sent.get("timestamp")), + "channel": "sent-log", + "kind": "send", + } + ) + + # Conversation = task-log events + structured sent log + coop-send/broadcast. + convo.extend(_conversation_from_task_log(task_log_doc, tasks_doc, role_by_id)) + convo.sort(key=lambda m: _ts_float(m.get("timestamp")) or 0.0) + _dump(out_dir / "conversation.json", convo) + + # result.json — coop shape + team provenance. + correct = bool(eval_doc.get("both_passed")) + apply_status = eval_doc.get("apply_status") or {} + merge = eval_doc.get("merge") or {} + + def _coop_agent_summary(idx: int) -> dict[str, Any]: + src = inv_role[idx] + summary = agents.get(src) or {} + return { + "feature_id": _feature_id_for(summary) or idx, + "status": summary.get("status") or "Unknown", + "cost": float(summary.get("cost") or 0.0), + "steps": int(summary.get("steps") or 0), + "input_tokens": int(summary.get("input_tokens") or 0), + "output_tokens": int(summary.get("output_tokens") or 0), + "cache_read_tokens": int(summary.get("cache_read_tokens") or 0), + "cache_write_tokens": int(summary.get("cache_write_tokens") or 0), + "patch_lines": int(summary.get("patch_lines") or 0), + "error": summary.get("error"), + } + + log_dir_rel = str(Path(run_name) / "coop" / repo / str(task_id) / feature_dir) + coop_result = { + "repo": repo, + "task_id": task_id, + "features": FEATURES, + "setting": "coop", + "run_id": result.get("run_id"), + "run_name": run_name, + "agent_framework": result.get("agent_framework"), + "model": result.get("model"), + "started_at": result.get("started_at"), + "ended_at": result.get("ended_at"), + "duration_seconds": result.get("duration_seconds") or 0.0, + "agents": {f"agent{idx}": _coop_agent_summary(idx) for idx in (1, 2)}, + "total_cost": float(result.get("total_cost") or 0.0), + "total_steps": int(result.get("total_steps") or 0), + "messages_sent": len(convo), + "log_dir": log_dir_rel, + "team": { + **team_block_common, + "source_features": [ + _feature_id_for(agents.get(src_agent1) or {}), + _feature_id_for(agents.get(src_agent2) or {}), + ], + "apply_status": apply_status, + "merge_status": merge.get("status"), + "merge_strategy": merge.get("strategy"), + }, + } + _dump(out_dir / "result.json", coop_result) + + # eval.json — preserve the team eval verbatim, recast to coop schema. + # ``verified`` flags a POSITIVE outcome confirmed by held-out tests — + # set True only when ``correct`` is True. Failures get + # ``verified: false`` even though cooperbench did run tests, so + # downstream filters can treat verified-positive trajectories as the + # headline-success subset. + coop_eval = { + "correct": correct, + "score": 1.0 if correct else 0.0, + "eval": "pass" if correct else "fail", + "verified": correct, + "eval_source": "cooperbench_held_out_tests", + "both_passed": correct, + "feature1": eval_doc.get("feature1"), + "feature2": eval_doc.get("feature2"), + "apply_status": apply_status, + "merge": merge, + "error": eval_doc.get("error"), + "evaluated_at": eval_doc.get("evaluated_at"), + } + _dump(out_dir / "eval.json", coop_eval) + + # metadata.json — full coordination state + provenance. + metadata = { + "source": SOURCE, + "source_run": source_run_name, + "source_pair_dir": str(pair_dir), + "task_name": TASK_NAME, + "repo": repo, + "task_id": task_id, + "features": FEATURES, + "source_features": [ + _feature_id_for(agents.get(src_agent1) or {}), + _feature_id_for(agents.get(src_agent2) or {}), + ], + "team_features": result.get("team_features") or {}, + "lead_agent": result.get("lead_agent"), + "metrics": result.get("metrics") or {}, + "agent_framework": result.get("agent_framework"), + "model": result.get("model"), + "started_at": result.get("started_at"), + "ended_at": result.get("ended_at"), + "duration_seconds": result.get("duration_seconds"), + "agent_id_mapping": {f"agent{idx}": inv_role[idx] for idx in (1, 2)}, + "tasks": tasks_doc, + "task_log": task_log_doc, + "converted_at": datetime.now(tz=timezone.utc).isoformat(), + } + _dump(out_dir / "metadata.json", metadata) + + return { + "task": f"{repo}/{task_id}/{src_f1},{src_f2}", + "source_pair_dir": str(pair_dir), + "status": "completed", + "cost": float(result.get("total_cost") or 0.0), + "eval": "pass" if correct else "fail", + "score": 1.0 if correct else 0.0, + } + + +def _iter_pair_dirs(team_root: Path): + # logs//team//// + for repo_dir in sorted(p for p in team_root.iterdir() if p.is_dir()): + for task_dir in sorted(p for p in repo_dir.iterdir() if p.is_dir()): + yield from sorted(p for p in task_dir.iterdir() if p.is_dir()) + + +def convert_run( + source_run: Path, + out_root: Path, + *, + run_name: str | None = None, +) -> dict[str, Any]: + team_root = source_run / "team" + if not team_root.is_dir(): + raise SystemExit(f"no team dir: {team_root}") + + run_name = run_name or source_run.name + source_run_name = source_run.name + started_at = datetime.now(tz=timezone.utc).isoformat() + + rows: list[dict[str, Any]] = [] + for pair_dir in _iter_pair_dirs(team_root): + row = convert_pair(pair_dir, out_root, run_name, source_run_name=source_run_name) + if row: + rows.append(row) + logger.info("converted %s → %s (%s)", pair_dir, row["task"], row["eval"]) + + # Run-level config.json + summary.json (PR #98 shape). + source_config = _read_json(source_run / "config.json") or {} + source_summary = _read_json(source_run / "summary.json") or {} + run_root = out_root / run_name + run_root.mkdir(parents=True, exist_ok=True) + _dump( + run_root / "config.json", + { + "run_name": run_name, + "agent_framework": source_config.get("agent_framework"), + "model": source_config.get("model"), + "setting": "coop", + "source": SOURCE, + "source_run": source_run_name, + "source_config": source_config, + "total_tasks": len(rows), + "started_at": started_at, + }, + ) + graded = [r for r in rows if r["eval"] in ("pass", "fail")] + passed = sum(1 for r in rows if r["eval"] == "pass") + _dump( + run_root / "summary.json", + { + "run_name": run_name, + "completed_at": datetime.now(tz=timezone.utc).isoformat(), + "source": SOURCE, + "source_run": source_run_name, + "total_tasks": len(rows), + "completed": len(rows), + "pass_rate": (passed / len(graded)) if graded else None, + "total_cost": sum(r["cost"] for r in rows), + "results": rows, + "source_summary": source_summary, + }, + ) + return {"run_name": run_name, "rows": rows, "pass_rate": (passed / len(graded)) if graded else None} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("source_run", type=Path, help="path to logs//") + parser.add_argument("--out", type=Path, default=Path("data"), help="output root (default: ./data)") + parser.add_argument( + "--run-name", type=str, default=None, help="override run name in output (default: source dir name)" + ) + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args(argv) + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(levelname)s %(message)s", + ) + res = convert_run(args.source_run, args.out, run_name=args.run_name) + print(json.dumps(res, indent=2, default=str)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/cooperbench/agents/mini_swe_agent_v2/agents/default.py b/src/cooperbench/agents/mini_swe_agent_v2/agents/default.py index 8e7e3725..3cfad774 100644 --- a/src/cooperbench/agents/mini_swe_agent_v2/agents/default.py +++ b/src/cooperbench/agents/mini_swe_agent_v2/agents/default.py @@ -224,8 +224,248 @@ def step(self) -> list[dict]: summary = poller.poll() if summary: self.add_messages(self.model.format_message(role="user", content=summary)) + notice = self._team_conflict_notice() + if notice: + self.add_messages(self.model.format_message(role="user", content=notice)) return self.execute_actions(self.query()) + def _team_conflict_notice(self) -> str | None: + """Just-in-time merge-conflict notice via an in-container 3-way + trial merge of teammates' published diffs against this agent's + working tree. Returns None on any failure so the agent loop + never breaks because the probe couldn't run.""" + try: + from cooperbench.team_harness.jit_merge import ( + build_probe_command, + format_conflict_notice, + parse_conflicts, + ) + + out = self.env.execute({"command": build_probe_command(self.agent_id)}) + conflicts = parse_conflicts(out.get("output") or "") + return format_conflict_notice(conflicts) or None + except Exception: + return None + + def _team_read_tasks(self) -> list[dict] | None: + """Return the live task list via the team poller's redis client, + or ``None`` if team mode isn't wired or Redis is unreachable. + + Centralised so the gate / prefix / blocking helpers all read the + same state and fail uniformly (returning None never breaks the + loop).""" + poller = getattr(self, "team_poller", None) + if poller is None: + return None + try: + from cooperbench.team_harness.loop_refresh import _read_tasks + + client = poller._ensure_client() # type: ignore[attr-defined] + if client is None: + return None + return _read_tasks(client, poller._run_id) # type: ignore[attr-defined] + except Exception: + return None + + def _team_required_actions(self, cmd: str) -> list[dict]: + """Coordination actions to auto-apply *before* the agent's command. + + Two cases (each fires once until its condition flips): + + - Unclaimed-own-task: if any of my tasks is still ``status=open`` + and the command isn't itself a claim, prepend a claim. + - Submit-with-in-progress: if the command is a final submit and any + of my tasks is still ``status=in_progress``, prepend an update to + mark it done first. + + Returns a list of action dicts of the form + ``{"kind": "claim"|"update", "task_id": ..., "title": ..., ...}``. + Pure decision function — does not mutate state. + """ + if not cmd: + return [] + tasks = self._team_read_tasks() + if not tasks: + return [] + + mine = [t for t in tasks if t.get("owner") == self.agent_id] + actions: list[dict] = [] + + if "coop-task-claim" not in cmd: + for t in mine: + if t.get("status") == "open": + actions.append({"kind": "claim", "task_id": t["id"], "title": t.get("title", "")}) + + is_submit = "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" in cmd + if is_submit and "coop-task-update" not in cmd: + for t in mine: + if t.get("status") == "in_progress": + title = (t.get("title") or "")[:60].replace("'", "").replace("\n", " ") + actions.append({"kind": "update", "task_id": t["id"], "status": "done", "note": f"auto: {title}"}) + + return actions + + def _team_required_prefix(self, cmd: str) -> list[str]: + """Back-compat shim: render required actions as CLI strings. + + Some tests assert the CLI-string shape directly. Production uses + ``_team_apply_prefix`` which mutates Redis server-side. + """ + out: list[str] = [] + for a in self._team_required_actions(cmd): + if a["kind"] == "claim": + out.append(f"coop-task-claim {a['task_id']}") + elif a["kind"] == "update": + out.append(f"coop-task-update {a['task_id']} {a['status']} -n '{a['note']}'") + return out + + def _team_apply_prefix(self, cmd: str) -> str | None: + """Apply required coordination actions server-side via the host + TaskListClient. Returns a synthesized tool-output string to + prepend to the agent's observation, or ``None`` if nothing was + applied. + + Bypasses the in-container ``coop-task-*`` shell CLI (which needs + the ``redis`` python module — not always available in task base + images, the install snippet silently no-ops if pip can't reach it). + The audit-log events still land in Redis so ``task_log.json`` and + downstream ``conversation.json`` see the claim/update events. + """ + actions = self._team_required_actions(cmd) + if not actions: + return None + poller = getattr(self, "team_poller", None) + if poller is None: + return None + try: + from cooperbench.team_harness.task_list import TaskListClient + + client = poller._ensure_client() # type: ignore[attr-defined] + if client is None: + return None + tlc = TaskListClient(redis_client=client, run_id=poller._run_id) # type: ignore[attr-defined] + except Exception: + return None + + chunks: list[str] = [] + for a in actions: + try: + if a["kind"] == "claim": + ok = tlc.claim(a["task_id"], by=self.agent_id) + if ok: + chunks.append( + f"$ coop-task-claim {a['task_id']}\n[auto] claimed: {a.get('title', '')}".rstrip() + ) + elif a["kind"] == "update": + tlc.update(a["task_id"], by=self.agent_id, status=a["status"], note=a["note"]) + chunks.append(f'$ coop-task-update {a["task_id"]} {a["status"]} -n "{a["note"]}"\n[auto] updated') + except Exception: + continue + return "\n".join(chunks) if chunks else None + + def _team_blocking_reason(self, cmd: str) -> str | None: + """Return a refusal message if the command must be blocked outright. + + Only one case is auto-fix-impossible: a lead trying to submit while + a peer's task is not yet ``status=done``. The lead has nothing to + auto-execute — it has to wait for the peer to update. + """ + if not cmd: + return None + is_submit = "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" in cmd + if not is_submit: + return None + tasks = self._team_read_tasks() + if not tasks: + return None + mine = [t for t in tasks if t.get("owner") == self.agent_id] + if not mine: + return None + am_lead = any("Lead-only" in (t.get("title") or "") for t in mine) + if not am_lead: + return None + peer_open = [ + t for t in tasks if t.get("owner") and t.get("owner") != self.agent_id and t.get("status") != "done" + ] + if not peer_open: + return None + lines = [ + f"{t.get('id', '?')} [{t.get('status', '?')}] owner={t.get('owner', '?')}: {t.get('title', '')}" + for t in peer_open + ] + return "[coord-gate] Cannot submit yet: peer task(s) not yet done.\n " + "\n ".join(lines) + + def _team_coord_gate(self, cmd: str) -> dict | None: + """Legacy combined gate, retained for backward-compat tests. + + Returns a refusal observation if any rule applies, or ``None``. + Equivalent to the old behavior before split into auto-prefix + + blocking-reason. Production execution uses the split helpers + directly so it can auto-execute the prefix rather than refuse. + """ + poller = getattr(self, "team_poller", None) + if poller is None or not cmd: + return None + try: + from cooperbench.team_harness.loop_refresh import _read_tasks + + client = poller._ensure_client() # type: ignore[attr-defined] + if client is None: + return None + tasks = _read_tasks(client, poller._run_id) # type: ignore[attr-defined] + except Exception: + return None + if not tasks: + return None + + mine = [t for t in tasks if t.get("owner") == self.agent_id] + others = [t for t in tasks if t.get("owner") and t.get("owner") != self.agent_id] + + # Rule 1: unclaimed-own-task gate + unclaimed = [t for t in mine if t.get("status") == "open"] + if unclaimed and "coop-task-claim" not in cmd: + ids = ", ".join(t.get("id", "?") for t in unclaimed) + msg = ( + f"[coord-gate] You have unclaimed task(s) assigned to you: {ids}. " + f"Claim before running other commands: coop-task-claim " + ) + return {"output": msg, "returncode": 1, "exception_info": ""} + + is_submit = "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" in cmd + + # Rule 2: own-task-not-done gate (on submit) + if is_submit: + in_prog = [t for t in mine if t.get("status") == "in_progress"] + if in_prog: + ids = ", ".join(t.get("id", "?") for t in in_prog) + msg = ( + f"[coord-gate] Cannot submit: your task(s) {ids} still in_progress. " + f'Mark done first: coop-task-update done -n ""' + ) + return {"output": msg, "returncode": 1, "exception_info": ""} + + # Rule 3: peer-not-done gate (on submit) — applies to lead only + # in practice; member submissions are independent of peers. + peer_open = [t for t in others if t.get("status") != "done"] + if peer_open and mine: # only gate if I'm a coordinator (have my own task) + # Heuristic: a "lead" task has "Lead-only" in its title (per the + # current task-creation prompts). If none of my tasks look like + # a lead task, don't gate on peer state — members can submit + # independently. + am_lead = any("Lead-only" in (t.get("title") or "") for t in mine) + if am_lead: + open_lines = [ + f"{t.get('id', '?')} [{t.get('status', '?')}] owner={t.get('owner', '?')}: {t.get('title', '')}" + for t in peer_open + ] + msg = ( + "[coord-gate] Cannot submit: peer task(s) not yet done. " + "Wait for them to update, then integrate:\n " + "\n ".join(open_lines) + ) + return {"output": msg, "returncode": 1, "exception_info": ""} + + return None + def _get_prompt_tokens(self, message: dict) -> int: return message.get("extra", {}).get("response", {}).get("usage", {}).get("prompt_tokens", 0) @@ -331,6 +571,17 @@ def execute_actions(self, message: dict) -> list[dict]: continue cmd = action.get("command", "") + + # Team-mode coordination gate: block what can't be auto-fixed, + # auto-execute required prefix commands (claim, update) for the + # rest. See _team_blocking_reason / _team_required_prefix. + blocked = self._team_blocking_reason(cmd) + if blocked is not None: + outputs.append({"output": blocked, "returncode": 1, "exception_info": ""}) + continue + + prefix_text = self._team_apply_prefix(cmd) or "" + if self.comm: sm_matches = _parse_send_messages(cmd) if sm_matches: @@ -341,14 +592,23 @@ def execute_actions(self, message: dict) -> list[dict]: remaining = _strip_send_message(cmd) combined = "\n".join(sm_outputs) if not remaining.strip(): - outputs.append({"output": combined, "returncode": 0, "exception_info": ""}) + output = combined + if prefix_text: + output = prefix_text + "\n" + output + outputs.append({"output": output, "returncode": 0, "exception_info": ""}) continue env_out = self.env.execute({**action, "command": remaining}) - env_out["output"] = combined + "\n" + env_out.get("output", "") + output = combined + "\n" + env_out.get("output", "") + if prefix_text: + output = prefix_text + "\n" + output + env_out["output"] = output outputs.append(env_out) continue - outputs.append(self.env.execute(action)) + env_out = self.env.execute(action) + if prefix_text: + env_out["output"] = prefix_text + "\n" + (env_out.get("output") or "") + outputs.append(env_out) return self.add_messages(*self.model.format_observation_messages(message, outputs, self.get_template_vars())) def _handle_send_message(self, action: dict) -> dict: diff --git a/tests/agents/mini_swe_agent_v2/test_team_coord_gate.py b/tests/agents/mini_swe_agent_v2/test_team_coord_gate.py new file mode 100644 index 00000000..6abfc3c4 --- /dev/null +++ b/tests/agents/mini_swe_agent_v2/test_team_coord_gate.py @@ -0,0 +1,365 @@ +"""Unit tests for the team-mode coordination gate in DefaultAgent. + +The gate sits between action extraction and ``env.execute()`` in +``DefaultAgent.execute_actions``. It reads the live task list from +Redis (via the agent's TeamPoller) and refuses commands that would +bypass the coordination protocol: + + 1. Unclaimed-own-task gate — refuse non-claim commands while the + agent owns a task in ``status=open``. + 2. Own-task-not-done gate — refuse final-submit commands while the + agent owns a task in ``status=in_progress``. + 3. Peer-not-done gate — refuse final-submit commands from the lead + while any other agent's task is not yet ``status=done``. + +All tests use ``fakeredis`` so there's no daemon dependency. +""" + +from __future__ import annotations + +import fakeredis +import pytest + +from cooperbench.agents.mini_swe_agent_v2.agents.default import DefaultAgent +from cooperbench.team_harness.task_list import TaskListClient + +SUBMIT_CMD = "echo COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" + + +class _StubPoller: + """Minimal TeamPoller stand-in: just exposes ._ensure_client and ._run_id.""" + + def __init__(self, client, run_id): + self._client = client + self._run_id = run_id + + def _ensure_client(self): + return self._client + + +def _bare_agent(agent_id, fake_redis, run_id="test"): + """Create a DefaultAgent without invoking __init__ (avoids pulling in + real Model/Environment classes that need config files). We only + exercise the _team_coord_gate method, which needs only + ``self.agent_id`` and ``self.team_poller``.""" + agent = DefaultAgent.__new__(DefaultAgent) + agent.agent_id = agent_id + agent.team_poller = _StubPoller(fake_redis, run_id) + return agent + + +@pytest.fixture +def fake_redis(): + return fakeredis.FakeRedis() + + +@pytest.fixture +def task_client(fake_redis): + return TaskListClient(redis_client=fake_redis, run_id="test") + + +# ----------------------------------------------------------------------------- +# No-team-mode fallthrough +# ----------------------------------------------------------------------------- + + +def test_no_poller_means_no_gate(fake_redis): + """Solo / coop runs have no team_poller — gate is a no-op.""" + agent = DefaultAgent.__new__(DefaultAgent) + agent.agent_id = "agent1" + agent.team_poller = None + assert agent._team_coord_gate("ls") is None + + +def test_no_tasks_means_no_gate(fake_redis): + """Team mode is wired but the task list is empty — nothing to gate.""" + agent = _bare_agent("agent1", fake_redis) + assert agent._team_coord_gate("ls") is None + + +def test_empty_cmd_means_no_gate(fake_redis, task_client): + """Whitespace/no-op command shouldn't be force-gated.""" + task_client.create(title="t", created_by="agent1", owner="agent1") + agent = _bare_agent("agent1", fake_redis) + assert agent._team_coord_gate("") is None + + +# ----------------------------------------------------------------------------- +# Rule 1: unclaimed-own-task gate +# ----------------------------------------------------------------------------- + + +def test_gate_blocks_when_own_task_is_unclaimed(fake_redis, task_client): + """Agent owns a pre-assigned task at status=open — must claim first.""" + tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + out = agent._team_coord_gate("grep -r foo .") + assert out is not None + assert out["returncode"] == 1 + assert "coop-task-claim" in out["output"] + assert tid in out["output"] + + +def test_gate_allows_claim_command_through(fake_redis, task_client): + """Even with unclaimed task, a claim command itself is permitted.""" + tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_coord_gate(f"coop-task-claim {tid}") is None + + +def test_other_agents_unclaimed_task_does_not_gate_me(fake_redis, task_client): + """Rule 1 fires only when *my* tasks are unclaimed — peer state irrelevant.""" + task_client.create(title="Feature 2", created_by="bench", owner="agent2") # peer's open task + agent = _bare_agent("agent1", fake_redis) + # agent1 has nothing assigned; should not be gated by agent2's open task + assert agent._team_coord_gate("ls") is None + + +# ----------------------------------------------------------------------------- +# Rule 2: own-task-not-done gate on submit +# ----------------------------------------------------------------------------- + + +def test_gate_blocks_submit_when_own_task_in_progress(fake_redis, task_client): + """Agent has claimed but not marked done — submit should be refused.""" + tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + out = agent._team_coord_gate(SUBMIT_CMD) + assert out is not None + assert "in_progress" in out["output"] + assert "coop-task-update" in out["output"] + assert tid in out["output"] + + +def test_gate_allows_non_submit_when_task_in_progress(fake_redis, task_client): + """While task is in_progress, regular bash work should still execute.""" + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_coord_gate("pytest tests/") is None + + +def test_gate_allows_submit_after_marking_done(fake_redis, task_client): + """Once status=done, member submit passes (no peer-dependency for non-lead).""" + tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + task_client.update(tid, by="agent2", status="done") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_coord_gate(SUBMIT_CMD) is None + + +# ----------------------------------------------------------------------------- +# Rule 3: peer-not-done gate (lead only) +# ----------------------------------------------------------------------------- + + +def test_gate_blocks_lead_submit_when_peer_still_open(fake_redis, task_client): + """Lead's task is done but member hasn't reported — lead can't submit.""" + member_tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + lead_tid = task_client.create(title="Lead-only: integrate and submit feature 1", created_by="bench", owner="agent1") + task_client.claim(lead_tid, by="agent1") + task_client.update(lead_tid, by="agent1", status="done") + agent = _bare_agent("agent1", fake_redis) + out = agent._team_coord_gate(SUBMIT_CMD) + assert out is not None + assert "peer task" in out["output"].lower() or "not yet done" in out["output"] + assert member_tid in out["output"] + + +def test_gate_allows_lead_submit_when_peer_done(fake_redis, task_client): + """Both done → lead can submit.""" + member_tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + lead_tid = task_client.create(title="Lead-only: integrate and submit feature 1", created_by="bench", owner="agent1") + task_client.claim(member_tid, by="agent2") + task_client.update(member_tid, by="agent2", status="done") + task_client.claim(lead_tid, by="agent1") + task_client.update(lead_tid, by="agent1", status="done") + agent = _bare_agent("agent1", fake_redis) + assert agent._team_coord_gate(SUBMIT_CMD) is None + + +def test_member_submit_does_not_wait_on_lead(fake_redis, task_client): + """A member whose own task is done can submit even if lead's task is open.""" + member_tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.create(title="Lead-only: integrate and submit feature 1", created_by="bench", owner="agent1") + task_client.claim(member_tid, by="agent2") + task_client.update(member_tid, by="agent2", status="done") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_coord_gate(SUBMIT_CMD) is None + + +# ----------------------------------------------------------------------------- +# Failure tolerance +# ----------------------------------------------------------------------------- + + +def test_gate_returns_none_on_broken_poller(): + """If the poller's redis client raises, gate must silently fall through.""" + + class BrokenPoller: + _run_id = "test" + + def _ensure_client(self): + raise RuntimeError("redis down") + + agent = DefaultAgent.__new__(DefaultAgent) + agent.agent_id = "agent1" + agent.team_poller = BrokenPoller() + assert agent._team_coord_gate("ls") is None + + +# ----------------------------------------------------------------------------- +# Split helpers: required_prefix + blocking_reason +# ----------------------------------------------------------------------------- + + +def test_required_prefix_adds_claim_for_open_task(fake_redis, task_client): + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + prefix = agent._team_required_prefix("pytest tests/") + assert prefix == [f"coop-task-claim {tid}"] + + +def test_required_prefix_empty_when_already_claimed(fake_redis, task_client): + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_required_prefix("pytest tests/") == [] + + +def test_required_prefix_skips_claim_when_cmd_is_claim(fake_redis, task_client): + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_required_prefix(f"coop-task-claim {tid}") == [] + + +def test_required_prefix_adds_update_on_submit_with_in_progress(fake_redis, task_client): + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + prefix = agent._team_required_prefix(SUBMIT_CMD) + assert len(prefix) == 1 + assert prefix[0].startswith(f"coop-task-update {tid} done -n ") + assert "Feature 2" in prefix[0] + + +def test_required_prefix_skips_update_when_cmd_already_updates(fake_redis, task_client): + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + cmd = f"coop-task-update {tid} done -n 'manual' && {SUBMIT_CMD}" + # rule still wants a claim? no — already claimed. and update is in cmd. -> empty + assert agent._team_required_prefix(cmd) == [] + + +def test_required_prefix_combines_claim_and_update_on_first_submit(fake_redis, task_client): + """Edge case: agent submits without ever claiming — prefix should chain both.""" + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + prefix = agent._team_required_prefix(SUBMIT_CMD) + # only claim fires here: after claim runs, status would become in_progress, + # but the prefix is computed from the snapshot we took at call time, so + # update doesn't appear (it would on the next submit attempt). Verify: + assert len(prefix) == 1 + assert prefix[0] == f"coop-task-claim {tid}" + + +def test_blocking_reason_none_for_non_submit(fake_redis, task_client): + task_client.create(title="Lead-only: integrate", created_by="bench", owner="agent1") + agent = _bare_agent("agent1", fake_redis) + assert agent._team_blocking_reason("pytest tests/") is None + + +def test_blocking_reason_blocks_lead_with_open_peer(fake_redis, task_client): + member_tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.create(title="Lead-only: integrate feature 1", created_by="bench", owner="agent1") + agent = _bare_agent("agent1", fake_redis) + reason = agent._team_blocking_reason(SUBMIT_CMD) + assert reason is not None + assert member_tid in reason + + +def test_blocking_reason_lets_member_submit(fake_redis, task_client): + """Members don't gate on peer state; only Lead-only owners do.""" + task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.create(title="Lead-only: integrate", created_by="bench", owner="agent1") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_blocking_reason(SUBMIT_CMD) is None + + +def test_blocking_reason_lets_lead_submit_when_peer_done(fake_redis, task_client): + member_tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.claim(member_tid, by="agent2") + task_client.update(member_tid, by="agent2", status="done") + task_client.create(title="Lead-only: integrate", created_by="bench", owner="agent1") + agent = _bare_agent("agent1", fake_redis) + assert agent._team_blocking_reason(SUBMIT_CMD) is None + + +# ----------------------------------------------------------------------------- +# Apply-prefix: server-side mutation (bypasses in-container CLI) +# ----------------------------------------------------------------------------- + + +def test_apply_prefix_claims_open_task_server_side(fake_redis, task_client): + """A non-claim cmd against an open-status own task → host applies the claim.""" + tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + out = agent._team_apply_prefix("pytest tests/") + assert out is not None + assert "coop-task-claim" in out and tid in out + # The Redis state actually flipped: + assert task_client.get(tid)["status"] == "in_progress" + # And the audit log saw the event: + kinds = [e.get("kind") for e in task_client.log_events()] + assert "claim" in kinds + + +def test_apply_prefix_marks_done_on_submit(fake_redis, task_client): + """Submit cmd against own in_progress task → host applies the update.""" + tid = task_client.create(title="Implement feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + out = agent._team_apply_prefix(SUBMIT_CMD) + assert out is not None + assert "coop-task-update" in out and tid in out + assert task_client.get(tid)["status"] == "done" + kinds = [e.get("kind") for e in task_client.log_events()] + assert "update" in kinds + + +def test_apply_prefix_noop_when_no_actions_needed(fake_redis, task_client): + """Already claimed in_progress task + non-submit cmd → nothing to apply.""" + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + task_client.claim(tid, by="agent2") + agent = _bare_agent("agent2", fake_redis) + assert agent._team_apply_prefix("pytest tests/") is None + + +def test_apply_prefix_is_idempotent_after_claim(fake_redis, task_client): + """Calling apply twice in a row should result in exactly one claim event.""" + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + agent._team_apply_prefix("pytest tests/") + agent._team_apply_prefix("ls /workspace") + kinds = [e.get("kind") for e in task_client.log_events()] + assert kinds.count("claim") == 1 + assert task_client.get(tid)["status"] == "in_progress" + + +def test_apply_prefix_handles_both_claim_and_update_for_unclaimed_submit(fake_redis, task_client): + """Submit before ever claiming → first apply claims, returned string only + mentions the claim (status snapshot was 'open' when actions were computed). + A subsequent apply with submit cmd would then trigger the update.""" + tid = task_client.create(title="Feature 2", created_by="bench", owner="agent2") + agent = _bare_agent("agent2", fake_redis) + out1 = agent._team_apply_prefix(SUBMIT_CMD) + assert out1 is not None + assert "coop-task-claim" in out1 + assert "coop-task-update" not in out1 + # second call now sees status=in_progress, triggers update + out2 = agent._team_apply_prefix(SUBMIT_CMD) + assert out2 is not None + assert "coop-task-update" in out2 + assert task_client.get(tid)["status"] == "done"