leetcrypt
df8f1881d8
Fourth benchmark axis: teams of LLM agents deliberate in a room, implement code in an isolated VM, and are scored deterministically on correctness/speed. - Multi-language adapter (python/js/go/rust/bash) via MultiPL-E continuation mode - Append-only JSONL ledger with status tracking (ok/dnf/killed/error) so budget-exhausted or crashed runs still record a row (fixes selection bias) - Model-aware wall-clock scaling (U-shaped by param count; 3x for reasoning) - Self-owned SIGALRM/SIGTERM watchdog (RunTimeout: BaseException so broad except Exception handlers in the infer/completion path can't swallow it) - Seed forwarded to Ollama sampler + markdown-fence stripping in completion.py
227 lines
9.3 KiB
Python
227 lines
9.3 KiB
Python
"""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)
|