"""The collaboration phase machine (SPEC §6). Phase-separated, research-backed shape for write-heavy work: a read/plan DELIBERATE phase (parallel-friendly, round-robin) followed by a single-driver IMPLEMENT/TEST phase, then SUBMIT. Members see the full shared trace each turn (Cognition: *share full traces, not just messages*). Everything is posted into the real room (arena mode) and recorded to the transcript. M1 turn-taking is strict round-robin (deterministic; the topology router is M3). Budget = whichever of max_rounds / max_tokens / wall_clock_s trips first. """ from __future__ import annotations import time from dataclasses import dataclass from . import infer, transcript as T from .. import completion from .challenge import Challenge from .scoring import EventOutcome from .team import Team from .vm import VM _PLAN_LOCK = "PLAN-LOCKED" @dataclass class Budget: deliberate_rounds: int = 2 # round-robin passes before implementing implement_attempts: int = 3 # driver code/test iterations max_tokens: int = 8000 wall_clock_s: float = 300.0 @property def max_rounds(self) -> int: # for scoring's speed normalization return self.implement_attempts def _destructive_guard(): """The agent's own DESTRUCTIVE regex, read-only (referee/fair-play hook).""" import sys from pathlib import Path repo = Path(__file__).resolve().parents[4] if str(repo) not in sys.path: sys.path.insert(0, str(repo)) from cmd_chat.agent.bridge import DESTRUCTIVE # noqa: E402 return DESTRUCTIVE def _history_block(history: list[tuple[str, str]]) -> str: if not history: return "(no messages yet)" return "\n".join(f"{who}: {msg}" for who, msg in history) def run_event(team: Team, challenge: Challenge, room, vm: VM, tx: T.Transcript, budget: Budget, *, host: str = "http://127.0.0.1:11434", temperature: float = 0.4, seed: int | None = None, progress=None) -> EventOutcome: t0 = time.time() guard = _destructive_guard() tokens = {"in": 0, "out": 0} history: list[tuple[str, str]] = [] penalties: list[dict] = [] def _emit(p): if progress: progress(p) def over_budget() -> str | None: if time.time() - t0 > budget.wall_clock_s: return "wall_clock" if tokens["in"] + tokens["out"] > budget.max_tokens: return "tokens" return None # ── BRIEF ────────────────────────────────────────────────────────────── tx.phase_change("BRIEF") brief = challenge.brief() d = room.brief(brief) tx.add(T.KIND_MESSAGE, "BRIEF", "referee", {"text": brief, "delivered": d.delivered, "substrate": d.substrate}, role="referee") _emit("BRIEF posted") # ── DELIBERATE (round-robin, share full trace) ────────────────────────── tx.phase_change("DELIBERATE") plan_locked = False deliberate_passes = 0 for rnd in range(budget.deliberate_rounds): if over_budget(): penalties.append({"kind": "budget", "detail": over_budget()}) break deliberate_passes += 1 for member in team.speaking_order(): user = (f"{brief}\n\nConversation so far:\n" f"{_history_block(history)}\n\n" f"Your turn, {member.name}. Contribute one short message " f"(a proposal, critique, or decision). If the team has " f"agreed on an approach, end with {_PLAN_LOCK} on its own line.") reply = infer.chat(member.model, member.system_prompt(), [{"role": "user", "content": user}], host=host, temperature=temperature) tx.add(T.KIND_AGENT, "DELIBERATE", member.name, {"ok": reply.ok}, role=member.role, model=member.model, tokens=reply.tokens) text = reply.text if reply.ok else f"[infer error: {reply.error}]" tokens["in"] += reply.tokens.get("in", 0) tokens["out"] += reply.tokens.get("out", 0) dd = room.post(member.name, text) tx.add(T.KIND_MESSAGE, "DELIBERATE", member.name, {"text": text, "delivered": dd.delivered, "substrate": dd.substrate}, role=member.role, model=member.model) history.append((member.name, text)) _emit(f"DELIBERATE r{rnd + 1} {member.name}") if _PLAN_LOCK in text: plan_locked = True if plan_locked: break # extract the locked plan (last substantive deliberation message) to seed # the driver — this is how the agreed plan reaches the implementation. plan = "" for who, msg in reversed(history): clean = msg.replace(_PLAN_LOCK, "").strip() if clean: plan = clean break # ── IMPLEMENT / TEST (single driver) ──────────────────────────────────── driver = team.driver() tx.acl("IMPLEMENT", {"_perm": "acl", "drivers": [driver.name], "sudoers": []}) room.acl({"_perm": "acl", "drivers": [driver.name], "sudoers": []}) tx.phase_change("IMPLEMENT", note=f"driver={driver.name}") public_passed = False rounds_to_green: int | None = None attempts = 0 last_fail = "" completion_text = "" for attempt in range(budget.implement_attempts): if over_budget(): penalties.append({"kind": "budget", "detail": over_budget()}) break attempts += 1 # build the raw continuation prompt; seed with the agreed plan and any # prior failure so the driver iterates. prefix = "" if plan: plan_c = "\n".join(f"# {ln}" for ln in plan.splitlines()[:6]) prefix += f"# Team plan:\n{plan_c}\n" if last_fail: fc = "\n".join(f"# {ln}" for ln in last_fail.splitlines()[:6]) prefix += f"# Previous attempt failed:\n{fc}\n" gen_prompt = prefix + challenge.gen_prompt() comp = completion.complete(driver.model, gen_prompt, challenge.stop_tokens(), host=host, temperature=temperature, timeout=180.0, seed=seed) tx.add(T.KIND_AGENT, "IMPLEMENT", driver.name, {"ok": comp.ok, "attempt": attempts}, role=driver.role, model=driver.model) if not comp.ok: last_fail = f"generation error: {comp.error}" continue completion_text = comp.text # fair-play guard hook (M4 referee will act on this; M1 just records) flagged = guard.search(completion_text) if flagged: tx.guard("IMPLEMENT", driver.name, {"flagged": flagged.group(0)[:60]}) penalties.append({"kind": "destructive", "detail": flagged.group(0)[:60]}) program = challenge.public_program(completion_text) res = vm.run(program, timeout=30.0, label=f"public-attempt-{attempts}") tx.tool("TEST", driver.name, {"label": f"public-attempt-{attempts}", "rc": res.rc, "ok": res.ok, "out": res.out[-400:]}, role=driver.role) _emit(f"IMPLEMENT attempt {attempts} -> {'green' if res.ok else 'red'}") if res.ok: public_passed = True rounds_to_green = attempts break last_fail = (res.note or res.out)[-300:] fb = (f"Attempt {attempts} failed public tests:\n{last_fail}\n" f"Driver, revise the implementation.") fbd = room.post("referee", fb) tx.add(T.KIND_MESSAGE, "TEST", "referee", {"text": fb, "delivered": fbd.delivered, "substrate": fbd.substrate}, role="referee") # revoke drive between phases tx.acl("SUBMIT", {"_perm": "acl", "drivers": [], "sudoers": []}) room.acl({"_perm": "acl", "drivers": [], "sudoers": []}) # ── SUBMIT + hidden grade ──────────────────────────────────────────────── tx.phase_change("SUBMIT") sd = room.post(driver.name, "SUBMIT") tx.add(T.KIND_MESSAGE, "SUBMIT", driver.name, {"text": "SUBMIT", "delivered": sd.delivered, "substrate": sd.substrate}, role=driver.role) correct = False if completion_text: hidden = vm.run(challenge.hidden_program(completion_text), timeout=30.0, label="hidden-grade") tx.tool("SUBMIT", "referee", {"label": "hidden-grade", "rc": hidden.rc, "ok": hidden.ok, "out": hidden.out[-400:]}) correct = hidden.ok submitted = bool(completion_text) tx.add(T.KIND_TOOL, "SUBMIT", "referee", {"label": "artifact", **vm.freeze()}) elapsed = time.time() - t0 return EventOutcome( team=team.id, challenge=challenge.id, submitted=submitted, correct=correct, rounds=deliberate_passes + attempts, wall_clock_s=elapsed, tokens=tokens, public_passed=public_passed, rounds_to_green=rounds_to_green, penalties=penalties)