Session vivante : reprise, RPC et fork
Reprendre une session pi sans la perdre : un registre écrit avant l'appel, un refus hors de son dossier. Puis le mode RPC de pi, réservé au best-of-N par clone. Comparé à Claude Code.
Une session pi a deux faiblesses. Si le runner est tué net, son identifiant disparaît avec lui. Et
une reprise lancée depuis un autre dossier crée en silence une seconde session au même
identifiant. Ce chapitre corrige les deux, puis ouvre le mode RPC de pi : un processus qui reste
ouvert et sait cloner une session. Il sert au best-of-N, où N builds partent d’une même amorce
payée une fois. Trois pièces : harness.py (remplace la version A8), harness_rpc.py et
adw_bon.py.
Reprendre une session sans la perdre
Une session pi est un fichier JSONL. Le runner lui donne son nom avec --session-id.
Si aucun fichier ne porte cet identifiant, pi le crée et écrit un avertissement sur stderr, que le runner ignore. Sinon, pi rouvre le fichier et charge l’historique sans le ré-émettre. Une reprise ne compte donc jamais deux fois les mêmes jetons. Comportement vérifié sur pi 0.85.
Le processus pi -p meurt à chaque tentative. Une reprise est un nouveau pi -p qui rouvre le
même fichier. Rien ne vit en mémoire, d’où un kill propre et une trace bornée.
Premier trou : le runner tué net
Jusqu’ici, l’identifiant ne vivait que dans la mémoire du runner (run.sessions). Après un
Ctrl-C, la tentative suivante repartait à froid. Le port écrit désormais une ligne dans un
registre, adws/adw_data/sessions/ledger.jsonl, avant de lancer pi.
{"session_id": "5b1c…", "cwd": "/home/vous/plume-factory", "model": "z-ai/glm-5.3",
"phase": "build", "adw_id": "9f2e4a10", "declared_at": "2026-09-05T08:41:12+00:00"}La phase et l’ADW viennent de l’environnement (FACTORY_PHASE, FACTORY_ADW_ID, chapitre A7).
Après un runner tué, un grep sur le registre rend la session à reprendre. La reprise coûte
quelques centimes au lieu de tout relire.
Second trou : la reprise depuis un autre dossier
Avec un --session-dir non standard, pi filtre les sessions sur le champ cwd de leur en-tête.
Lancé depuis un autre dossier, un worktree par exemple, il ne trouve rien et crée une seconde
session au même identifiant, sans erreur. La trace est coupée en deux. Le port refuse donc cette
reprise avant tout appel, en zéro jeton.
# adws/adw_modules/harness.py — extrait : refuser AVANT d'appeler, zero jeton.
def resume_requires_cwd(session_id, cwd, ledger=LEDGER, session_dir=SESSION_DIR):
origin = session_origin(session_id, ledger, session_dir)
if origin is None:
return # inconnue : pi la creera — c'est une premiere tentative
if Path(origin).resolve() != Path(cwd).resolve():
raise HarnessError(f"reprise refusee : la session {session_id} est nee dans {origin}, "
f"demandee depuis {Path(cwd).resolve()} — pi en creerait une "
"seconde au meme id", session_id)session_origin cherche le dossier de naissance dans le registre, puis dans l’en-tête du fichier
de session. Une session née avant ce chapitre est donc protégée aussi. Le refus est un échec de
phase ordinaire, session conservée.
Quatre façons de continuer un travail
| Geste | Ce que pi fait | Quand l’usine l’emploie |
|---|---|---|
pi -p, id neuf | session vide | première tentative d’une phase |
pi -p, même id, même dossier | recharge le fichier | reprise (ch. 8, A7) |
pi -p, même id, autre dossier | seconde session, même id | jamais, refusé par resume_requires_cwd |
pi --mode rpc gardé ouvert | plusieurs tours et des clones | best-of-N par clone |
Côté Claude Code : --session-id impose aussi un identifiant choisi d’avance (un UUID), et
--resume reprend une session par son identifiant. Mais --resume la cherche dans tous les
projets de la machine (depuis la v2.1.223), donc le piège de la session jumelle n’existe pas.
RPC et fork pour le best-of-N
Lancez pi --mode rpc : le processus reste ouvert et attend vos commandes sur stdin.
Le protocole
Chaque commande est un objet JSON sur une ligne, par exemple
{"id": "c1", "type": "prompt", "message": "…"}. pi répond par un objet type: response au même
id, puis diffuse les événements du tour jusqu’à agent_settled, qui signale que rien ne suivra.
Le séparateur est \n seul, et le client découpe uniquement là-dessus. Protocole vérifié dans la
doc rpc.md de pi 0.85.
L’usine emploie sept commandes : prompt, clone, switch_session, get_state (pour lire
sessionId et sessionFile), set_model, abort et get_session_stats.
Le clone
clone copie la session courante, à sa position, dans une session neuve et bascule dessus.
switch_session revient à l’original. pi a aussi fork, qui repart d’un message antérieur, mais
l’usine n’en a pas besoin.
Chaque clone ré-émet session_start. Le coupe-circuit (A7) remet ses compteurs à zéro et
l’inventaire (A8) se redépose. Chaque bras est donc jugé comme une phase. L’environnement (schéma
de la porte, limites) reste fixé pour tout le processus : une enveloppe et une garde par
processus.
Le best-of-N par clone
Un best-of-3 sur une spec du planner. L’amorce lit la spec, les fichiers qu’elle nomme et les tests : quelques centimes. Trois clones implémentent ensuite, une vingtaine de centimes chacun. Les gates tranchent et le meilleur patch est proposé.
Avec trois adw_build séparés, l’amorce serait payée trois fois, avec trois lectures différentes
de la spec. Par clone, elle est payée une fois. Le gain est de dix à vingt pour cent, et surtout
les trois bras sont comparables, car ils partent du même état.
Les demandes de dialogue
En RPC, ctx.hasUI vaut true. Un damage control réglé sur ask (chapitre A3) émet donc une
demande extension_ui_request et bloque pi jusqu’à la réponse. Le client répond cancelled à
toute demande de dialogue et la journalise. ask vaut refus, sans humain dans la boucle.
Côté Claude Code : en -p, --permission-prompts none refuse toute demande de permission
qu’aucune règle ni aucun hook ne résout, sans attendre de réponse (v2.1.259 et plus).
Quand prendre le RPC
| Critère | pi -p par phase (ch. A6) | pi --mode rpc (ce chapitre) |
|---|---|---|
| Isolation d’une tentative | totale, le processus meurt | partielle, le clone isole la session |
| Environnement (schéma, limites) | par phase | par processus : une enveloppe, une garde |
| Reprise après une coupure | rouvrir le fichier | le processus est déjà là, abort externe possible |
| Départs répétés (best-of-N) | N contextes à reconstruire | une amorce, N clones |
| Ce que le runner doit tenir | rien entre deux appels | un tube stdin/stdout, des id, un close() |
Le RPC ne rend pas un tour moins cher. Le contexte est renvoyé à chaque tour dans les deux
modes, et seul le démarrage de pi est économisé. Ce que le RPC achète, c’est le clone. Hors de ce
cas, c’est un état de plus à tenir. D’où la règle du livre : pi -p pour les phases, RPC pour le
best-of-N.
Le client lit chaque tour avec le même _read_pi_stream et le même _judge que le port. Le RPC
change la façon de parler à pi, pas la façon de le juger.
# adws/adw_modules/harness_rpc.py — extrait : le tour, puis le verdict du port.
def prompt(self, message: str) -> Turn:
clock = time.monotonic()
_, seen = self._command({"type": "prompt", "message": message}) # accepte, ou refuse
lines, deadline = list(seen), clock + self.request.timeout
while True:
event, line = self._next(deadline) # les demandes d'interface y sont refusees
lines.append(line)
if event.get("type") == SETTLED: # agent_settled : plus rien ne suivra
break
return Turn(reading=harness._read_pi_stream(lines), lines=lines,
seconds=time.monotonic() - clock)
def judge(self, turn: Turn, request=None) -> harness.HarnessResult:
return harness._judge(turn.reading, request or self.request, self.session_id, 0, "")Côté Claude Code : l’Agent SDK offre un processus long comparable, le mode d’entrée en
streaming (ClaudeSDKClient en Python). Pour le clone, --resume avec --fork-session crée une
nouvelle session à partir d’une copie de l’historique. Ce fork se fait au lancement d’un appel, pas
dans un processus ouvert.
pi et Claude Code côte à côte
| pi | Claude Code | |
|---|---|---|
| Nommer la session d’avance | --session-id | --session-id (UUID) |
| Reprendre une session | même id, même dossier | --resume, depuis n’importe quel dossier de la machine |
| Processus long, plusieurs tours | pi --mode rpc | Agent SDK en mode streaming |
| Cloner une session | clone, dans le processus RPC | --fork-session avec --resume |
| Demande de permission sans humain | le client répond cancelled | --permission-prompts none |
Fil rouge — la pièce posée aujourd’hui
Le port harnais (adaptateur v6) déclare chaque session et refuse les reprises hors dossier.
À côté, harness_rpc.py ouvre une seconde porte vers pi pour un seul client, adw_bon.py, le
best-of-N local qui complète celui du chapitre 25. La couture ne bouge pas : le runner
choisit, clone, juge et moissonne, l’agent reste un tour borné dans chaque clone. Coût : zéro
jeton pour le registre et le refus, moins d’un dollar pour un best-of-3. Le miroir de
factory-obs.ts (A4) ne tourne qu’à la fermeture du processus RPC, donc adw_bon.py relance
l’ingestion après avoir fermé pi.
Travaux pratiques — la pièce du jour
Trois fichiers à poser dans plume-factory, qui devient, chapitre après chapitre, votre usine
logicielle agentique. Prérequis : pi (chapitre A1), Bun (chapitre 2) et un dépôt git propre.
L’ADW du jour remet l’arbre à zéro entre deux bras, il l’exige et le vérifie.
Pièce — adws/adw_modules/harness.py
Cette version remplace celle du chapitre A8. Tout reste : dataclasses, run(), _judge,
adaptateur Claude Code et profil d’authentification du chapitre 17bis. S’ajoutent le registre
(LEDGER, declare_session, session_origin), le refus resume_requires_cwd et _pi_flags,
les drapeaux pi que le client RPC réutilise tels quels.
"""harness — le port de l'usine vers ses agents.
Une frontiere, deux adaptateurs. Aucun script de l'usine n'invoque `pi` ou
`claude` directement : tout passe par run(). Une HarnessRequest entre, un
HarnessResult sort — quel que soit le harnais derriere la porte.
Version chapitre 17bis : le profil d'authentification par agent. Une
requete peut porter `auth`, les variables de credentials que le roster a
resolues pour cet agent (noms ET valeurs, jamais dans le prompt), et
`direct`, la route hors passerelle d'un profil de forfait ou d'API native.
Quand elle porte un profil, .env devient un COFFRE : le noeud ne recoit
plus tout ce que .env avait charge, seulement ce que son profil nomme — a
la place du .env global du ch. 15, meme preseance (.env puis environnement
reel). Sans `auth`, l'heritage du ch. 15 s'applique tel quel.
Version annexe A6 (v4 de l'adaptateur pi) : le runner ne fait plus confiance
au noeud pi, il le VERIFIE. Le socle .pi/ est charge a coup sur (--approve),
le prompt voyage par stdin, l'enveloppe arrive par l'outil terminal
report_phase (tool_execution_end) plutot que devinee dans la prose, les
jetons et le stopReason sont lus, le modele observe est compare au modele
demande, et l'identifiant de session est choisi AVANT l'appel — une erreur
le porte, la reprise ne perd plus sa session.
Version annexe A8 (v5) : le runner exige la PREUVE que le socle est charge
(inventaire depose dans le flux, refus fail-closed s'il manque ou s'il
declare un manquant). Le verdict est une fonction pure (_judge).
Version annexe A9 (v6) : la session survit au runner. Chaque session est
DECLAREE dans un registre (adws/adw_data/sessions/ledger.jsonl) avant que pi
demarre — identifiant, dossier de naissance, modele, phase, ADW — et une
reprise depuis un autre dossier est REFUSEE avant tout appel : pi, avec un
--session-dir non standard, filtre ses sessions sur le cwd de leur en-tete
et en creerait une seconde au meme id. Les drapeaux pi sont factorises
(_pi_flags) pour que harness_rpc.py, la session vivante du best-of-N,
parle exactement la meme langue. L'API de run() ne bouge pas : vos ADW des
chapitres 8 a 27 tournent tels quels.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from . import envelopes
# Les sessions pi et les schemas de porte vivent dans adw_data/ — couvert par
# le .gitignore du ch. 1. Chemins ABSOLUS a l'appel : pi filtre ses sessions
# sur le cwd de leur en-tete, un chemin relatif devient ambigu (module 6).
SESSION_DIR = Path("adws/adw_data/sessions")
SCHEMA_DIR = Path("adws/adw_data/schemas")
# Le registre des sessions (A9) : une ligne JSON par session ouverte, ecrite
# AVANT l'appel — l'uuid, le dossier de naissance, le modele, la phase et
# l'ADW. C'est ce qui survit a un runner tue net, et ce que lit
# resume_requires_cwd pour refuser une reprise hors de son dossier.
LEDGER = SESSION_DIR / "ledger.jsonl"
PHASE_ENV = "FACTORY_PHASE" # poses par la fabrique agent_action (A7)
ADW_ENV = "FACTORY_ADW_ID"
# La route par defaut de l'usine : le fournisseur passerelle integre de pi.
# Une seule cle (OPENROUTER_API_KEY) sert tous les moteurs du roster.
# Passer par les API directes des fournisseurs : GATEWAY = "" — et a vous
# de fournir une cle par fournisseur dans l'environnement.
GATEWAY = "openrouter"
# La variable que lit .pi/extensions/factory-report.ts pour construire
# l'outil report_phase : le chemin du schema JSON ecrit par ce module.
SCHEMA_ENV = "FACTORY_ENVELOPE_SCHEMA"
# Le socle : les extensions pi que TOUT noeud d'usine doit avoir chargees.
# Le runner les nomme (FACTORY_EXPECTED), factory-inventory.ts les compare a
# ce qu'il trouve et depose l'inventaire dans le flux (INVENTORY_ENTRY). Le
# poste (footer, chaine) n'en fait pas partie : il ne se charge qu'en TUI.
SOCLE = ("damage-control", "factory-obs", "factory-report", "factory-guard",
"factory-inventory")
EXPECTED_ENV = "FACTORY_EXPECTED"
INVENTORY_ENTRY = "factory-inventory"
# Le dialecte Claude Code : outils avec majuscules, et pas d'outil ls ni find
# dedies — Bash et Glob les couvrent. L'adaptateur absorbe l'asymetrie.
CLAUDE_TOOLS = {"read": "Read", "bash": "Bash", "edit": "Edit", "write": "Write",
"grep": "Grep", "find": "Glob", "ls": "Bash"}
# L'echelle de reflexion de pi, traduite en budget de tokens pour Claude Code
# (variable d'environnement MAX_THINKING_TOKENS).
THINKING_TOKENS = {"off": 0, "minimal": 1024, "low": 4096, "medium": 8192,
"high": 16384, "xhigh": 24576, "max": 32000}
# Les cles que .env a fournies — et que l'environnement reel ne portait pas.
# C'est le COFFRE (17bis) : ce que le port retire du noeud quand la requete
# porte un profil, pour n'y remettre que ce que le profil nomme.
ENV_FILE_KEYS: set[str] = set()
def _load_env(path: str | Path = ".env") -> None:
"""Charge .env dans l'environnement du process — une fois, au chargement.
Ni pi ni claude ne lisent .env d'eux-memes : sans ce chargement, la cle
de la passerelle n'atteindrait jamais les agents. Une variable deja
presente dans l'environnement reel gagne toujours — un export de session
ou un secret de CI ne sont jamais ecrases — et n'entre pas dans le coffre :
elle est a vous, pas a .env.
"""
env_file = Path(path)
if not env_file.is_file():
return
for line in env_file.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key = key.strip()
if key not in os.environ:
os.environ[key] = value.strip().strip("'\"")
ENV_FILE_KEYS.add(key)
_load_env()
def node_environment(base: dict[str, str], extra_env: dict[str, str] | None,
auth: dict[str, str] | None, vault: set[str]) -> dict[str, str]:
"""L'environnement d'un noeud — pure, donc testable a sec.
Sans profil (auth=None) : l'heritage du ch. 15, tout .env compris. Avec
profil : .env est un coffre — chaque cle qu'il a fournie est retiree,
puis le profil depose les siennes, avec leurs valeurs. Ce que le port ou
l'ADW pose lui-meme pour la phase passe toujours.
"""
environment = dict(base)
if auth is not None:
for key in vault:
environment.pop(key, None)
environment.update(auth)
if extra_env:
environment.update(extra_env)
return environment
class HarnessError(RuntimeError):
"""Le harnais n'a pas rendu de reponse exploitable.
Depuis A6, l'erreur porte la session qu'elle a interrompue : l'uuid est
choisi avant l'appel, donc une tentative tuee par le mur de temps a quand
meme une session — la reprise la poursuit au lieu de repartir a froid.
"""
def __init__(self, message: str, session_id: str | None = None) -> None:
super().__init__(message)
self.session_id = session_id
@dataclass(frozen=True)
class HarnessRequest:
"""Ce que l'usine a le droit de demander a un agent — rien de plus."""
prompt: str
session_id: str | None = None # None = nouvelle session
cwd: str = "."
timeout: int = 600 # un agent muet ne bloque pas l'usine
model: str | None = None # l'id du REGISTRE, tel quel dans le roster
thinking: str | None = None # off..max — None = defaut du harnais
tools: tuple[str, ...] = () # allowlist du roster — () = outils par defaut
schema: dict | None = None # schema JSON de la porte — None = Envelope de base
socle: tuple[str, ...] = SOCLE # extensions exigees du noeud pi — () = aucune exigence
auth: dict[str, str] | None = None # le profil resolu (17bis) — None = l'heritage du ch. 15
direct: bool = False # route directe (17bis) : le modele est deja une route pi
@dataclass(frozen=True)
class HarnessResult:
"""Ce qu'un agent rend a l'usine — quel que soit le harnais."""
text: str # la derniere reponse de l'agent (ou l'enveloppe en JSON)
session_id: str # de quoi poursuivre la MEME session
cost_usd: float # 0.0 si le harnais ne rapporte pas le cout
returncode: int
tokens: int = 0 # jetons factures sur cet appel, 0 si non rapportes
stop_reason: str = "stop" # stop | length | toolUse | error | aborted
envelope: dict | None = None # l'enveloppe rendue par la porte typee, sinon None
def run(harness: str, request: HarnessRequest) -> HarnessResult:
"""L'unique porte d'entree vers les agents : choisit l'adaptateur, normalise."""
try:
adapter = ADAPTERS[harness]
except KeyError:
raise HarnessError(
f"harnais inconnu {harness!r} — disponibles : {sorted(ADAPTERS)}"
) from None
return adapter(request)
def _route(model: str, direct: bool = False) -> str:
"""L'id du registre devient une route pi : prefixe du fournisseur passerelle.
Le roster parle le langage du registre (z-ai/glm-5.3) — la meme chaine
que verifie la jauge du ch. 14. Le prefixe est un detail de dialecte :
il vit ici, jamais dans le YAML ni dans vos scripts. Sous un profil a
route directe (17bis), l'identifiant est deja une route pi (zai/glm-5.3,
kimi-coding/k3) et part tel quel — la route est dans le profil, pas dans
le nom : deepseek/... est un auteur OpenRouter ET un fournisseur natif.
"""
if direct or not GATEWAY or model.startswith(GATEWAY + "/"):
return model
return f"{GATEWAY}/{model}"
def _spawn(cmd: list[str], request: HarnessRequest,
extra_env: dict[str, str] | None = None,
stdin_text: str | None = None) -> subprocess.CompletedProcess[str]:
# Resoudre l'executable via le PATH : sous Windows, les harnais sont des
# shims (pi.cmd, claude.cmd) que CreateProcess ne trouve pas par leur nom
# court — shutil.which respecte PATHEXT et regle les deux mondes d'un coup.
executable = shutil.which(cmd[0])
if executable is None:
raise HarnessError(f"{cmd[0]!r} introuvable dans le PATH — "
"le harnais est-il installe ?")
environment = node_environment(dict(os.environ), extra_env, request.auth, ENV_FILE_KEYS)
# Deux modes d'entree, jamais d'entre-deux :
# - stdin_text=None : le prompt voyage dans argv, et stdin est ferme
# (DEVNULL) — un enfant qui herite de notre stdin peut attendre
# indefiniment une entree qui ne viendra jamais : echec silencieux,
# 0 % CPU, aucune sortie.
# - stdin_text : le prompt voyage par stdin, puis le tube est referme.
# Indispensable quand le harnais est un shim .cmd Windows : cmd.exe
# tronque un argument a la premiere nouvelle ligne, et les asks de
# l'usine (brief + mission + contrat) sont multi-lignes.
io = ({"input": stdin_text} if stdin_text is not None
else {"stdin": subprocess.DEVNULL})
# Encodage explicite : les harnais emettent de l'UTF-8, mais text=True
# seul decode avec la locale — cp1252 sous Windows, qui mutile tirets
# et accents. Vaut pour la sortie ET pour le prompt ecrit sur stdin.
try:
return subprocess.run([executable, *cmd[1:]], **io,
capture_output=True, text=True,
encoding="utf-8", errors="replace",
env=environment,
timeout=request.timeout, cwd=request.cwd)
except subprocess.TimeoutExpired:
# Le timeout aussi sort par la porte normalisee : une seule exception.
raise HarnessError(f"harnais muet apres {request.timeout} s : {cmd[0]}") from None
def _text_of(message: dict) -> str:
"""Concatene les blocs de texte d'un message pi."""
return "".join(part.get("text", "") for part in message.get("content", []) or []
if isinstance(part, dict) and part.get("type") == "text")
# ---------- le registre des sessions (A9) : declarer avant d'appeler ----------
def declare_session(session_id: str, cwd: str | Path, model: str | None = None,
ledger: Path = LEDGER, env: dict[str, str] | None = None) -> dict:
"""Ecrit la session dans le registre — AVANT que pi demarre.
Ce que l'ADW a pose dans l'environnement (phase, adw_id — fabrique
agent_action, A7) est releve au passage : un runner tue net laisse ainsi
de quoi retrouver la session de chaque phase, sans passer par la trace.
env : l'environnement du noeud s'il differe de celui du runner (RPC).
"""
source = os.environ if env is None else env
entry = {"session_id": session_id, "cwd": str(Path(cwd).resolve()), "model": model,
"phase": source.get(PHASE_ENV), "adw_id": source.get(ADW_ENV),
"declared_at": datetime.now(timezone.utc).isoformat(timespec="seconds")}
ledger.parent.mkdir(parents=True, exist_ok=True)
with ledger.open("a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
return entry
def session_origin(session_id: str, ledger: Path = LEDGER,
session_dir: Path = SESSION_DIR) -> str | None:
"""Le dossier de naissance d'une session : le registre d'abord, l'en-tete pi sinon.
Une session nee avant ce chapitre n'est pas dans le registre, mais son
fichier (<horodatage>_<id>.jsonl) commence par un en-tete qui porte le
cwd : elle est protegee de la meme facon. None = session inconnue.
"""
origin = None
if ledger.is_file():
for line in ledger.read_text(encoding="utf-8").splitlines():
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
if entry.get("session_id") == session_id and entry.get("cwd"):
origin = entry["cwd"] # la derniere declaration gagne
if origin:
return origin
for path in sorted(session_dir.glob(f"*_{session_id}.jsonl")):
try:
with path.open(encoding="utf-8") as f:
header = json.loads(f.readline())
except (OSError, json.JSONDecodeError):
continue
if header.get("type") == "session" and header.get("cwd"):
return str(header["cwd"])
return None
def resume_requires_cwd(session_id: str, cwd: str | Path, ledger: Path = LEDGER,
session_dir: Path = SESSION_DIR) -> None:
"""Refuse — avant tout appel, zero jeton — une reprise hors de son dossier.
Verifie sur pi 0.85 : avec un --session-dir non standard, pi filtre les
sessions sur le cwd de leur en-tete ; depuis un autre dossier il ne
retrouve rien et CREE une seconde session au meme id, sans erreur.
"""
origin = session_origin(session_id, ledger, session_dir)
if origin is None:
return # inconnue : pi la creera — c'est une premiere tentative
if Path(origin).resolve() != Path(cwd).resolve():
raise HarnessError(f"reprise refusee : la session {session_id} est nee dans {origin}, "
f"demandee depuis {Path(cwd).resolve()} — pi en creerait une "
"seconde au meme id", session_id)
# ---------- l'adaptateur pi : des fonctions pures autour d'un seul subprocess ----------
def _write_schema(schema: dict) -> Path:
"""La porte typee se construit depuis le schema ecrit ICI, avant la phase.
Une seule source de verite, cote Python : envelopes.schema(). L'extension
ne connait aucun champ d'avance — elle lit ce fichier au chargement.
"""
SCHEMA_DIR.mkdir(parents=True, exist_ok=True)
target = SCHEMA_DIR / f"{schema.get('title', 'Envelope')}.json"
target.write_text(json.dumps(schema, indent=2, ensure_ascii=False), encoding="utf-8")
return target.resolve()
def _pi_flags(request: HarnessRequest, session_id: str) -> list[str]:
"""Les drapeaux communs aux deux modes (json et rpc) — purs, testables a sec."""
flags = [
# --approve : le socle .pi/ (damage control, trace, porte typee) est
# charge a coup sur, sans dependre d'un trust.json de poste. Cela
# vaut confiance au depot COURANT — jamais sur un depot inconnu.
"--approve",
"--session-id", session_id, "--session-dir", str(SESSION_DIR.resolve()),
# Un noeud d'usine ne lit ni les skills ni les templates du poste :
# ils s'adressent a un humain qui pilote. AGENTS.md reste charge —
# c'est du contexte gouverne par le depot.
"--no-skills", "--no-prompt-templates"]
if request.model:
flags += ["--model", _route(request.model, request.direct)]
if request.thinking:
flags += ["--thinking", request.thinking]
if request.tools:
# --tools est une allowlist qui retire AUSSI les outils d'extension
# absents de la liste : la porte doit y figurer, sinon elle disparait.
flags += ["--tools", ",".join((*request.tools, envelopes.REPORT_TOOL))]
return flags
def _pi_argv(request: HarnessRequest, session_id: str) -> list[str]:
"""La ligne de commande pi d'une phase : un tour headless, flux JSON, prompt sur stdin."""
# `--` ferme les options : plus rien de positionnel — le prompt arrive par
# stdin, quel que soit son premier caractere ou son nombre de lignes.
return ["pi", "-p", "--mode", "json", *_pi_flags(request, session_id), "--"]
@dataclass
class PiReading:
"""Ce que le runner retient du flux JSON de pi : quatre chiffres, une enveloppe."""
text: str = ""
cost_usd: float = 0.0
tokens: int = 0
stop_reason: str = "stop"
observed_model: str | None = None
envelope: dict | None = None
inventory: dict | None = None # ce que le socle declare de lui-meme (A8)
def _read_pi_stream(lines: list[str]) -> PiReading:
"""Lit le flux ligne a ligne — pure, donc testable a sec.
message_end (assistant) : texte, cout, jetons, stopReason, modele observe.
tool_execution_end de report_phase : l'enveloppe, deja validee par schema
cote pi. entry_appended de factory-inventory : la preuve du socle. Le
dernier texte gagne ; les couts et jetons s'additionnent. Le flux d'un
tour RPC (harness_rpc.py) se lit avec la meme fonction.
"""
reading = PiReading()
for line in lines:
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
kind = event.get("type")
if kind == "message_end":
message = event.get("message") or {}
if message.get("role") != "assistant":
continue
reading.text = _text_of(message) or reading.text
usage = message.get("usage") or {}
reading.cost_usd += (usage.get("cost") or {}).get("total") or 0.0
reading.tokens += int(usage.get("totalTokens") or 0)
reading.stop_reason = message.get("stopReason") or reading.stop_reason
if reading.observed_model is None and message.get("model"):
reading.observed_model = f"{message.get('provider')}/{message.get('model')}"
elif (kind == "tool_execution_end"
and event.get("toolName") == envelopes.REPORT_TOOL
and not event.get("isError")):
details = (event.get("result") or {}).get("details")
if isinstance(details, dict):
reading.envelope = details
elif kind == "entry_appended":
entry = event.get("entry") or {}
if entry.get("customType") == INVENTORY_ENTRY and isinstance(entry.get("data"), dict):
reading.inventory = entry["data"]
return reading
def _judge(reading: PiReading, request: HarnessRequest, session_id: str,
returncode: int, stderr: str) -> HarnessResult:
"""Le verdict du runner sur un flux pi — pure, donc testable a sec.
Dans l'ordre : le socle d'abord (sans preuve, pas de phase — meme avec
une enveloppe) ; puis la porte typee ; puis les arrets qui ne sont pas
une reponse (error, aborted — pi rend 0 en --mode json, le code retour
ne dit rien) ; enfin le texte, pour le chemin de secours ; et le modele
observe, qui doit etre celui du roster.
"""
evidence = stderr.strip()[-400:]
if request.socle:
if reading.inventory is None:
raise HarnessError(
"socle pi non charge : aucun inventaire dans le flux — .pi/ non approuve, "
"factory-inventory.ts absent, ou pi n'a pas demarre de tour"
+ (f" ({evidence})" if evidence else ""), session_id)
missing = [name for name in request.socle
if name not in (reading.inventory.get("present") or [])]
if missing:
raise HarnessError(f"socle pi incomplet : manquants {missing} — "
f"pi {reading.inventory.get('pi_version', '?')} n'a charge que "
f"{reading.inventory.get('present')}", session_id)
if reading.envelope is not None:
text = json.dumps(reading.envelope, ensure_ascii=False)
elif reading.stop_reason in ("error", "aborted"):
raise HarnessError(f"pi s'est arrete sur {reading.stop_reason} : "
f"{evidence or reading.text[-400:]}", session_id)
elif returncode != 0 and not reading.text:
raise HarnessError(f"pi a rendu {returncode} : {evidence}", session_id)
else:
text = reading.text
# Le modele observe doit etre celui du roster : pi resout un motif par
# sous-chaine, et une facture sur le mauvais moteur n'est pas un run vert.
if request.model and reading.observed_model \
and reading.observed_model != _route(request.model, request.direct):
raise HarnessError(f"modele observe {reading.observed_model!r} "
f"≠ modele demande {_route(request.model, request.direct)!r}", session_id)
return HarnessResult(text=text, session_id=session_id, cost_usd=reading.cost_usd,
returncode=returncode, tokens=reading.tokens,
stop_reason=reading.stop_reason, envelope=reading.envelope)
def _run_pi(request: HarnessRequest) -> HarnessResult:
# pi : c'est VOUS qui nommez la session — et vous la nommez AVANT l'appel.
# Meme id + meme dossier = meme contexte ; une erreur porte cet id.
session_id = request.session_id or str(uuid.uuid4())
SESSION_DIR.mkdir(parents=True, exist_ok=True)
if request.session_id:
# Une reprise hors de son dossier de naissance ouvrirait une session
# jumelle : refus avant tout appel, zero jeton (A9).
resume_requires_cwd(session_id, request.cwd)
schema_path = _write_schema(request.schema or envelopes.schema(envelopes.Envelope))
# Ce que le noeud doit savoir de l'usine passe par l'environnement : le
# schema de la porte, et la liste du socle qu'il devra prouver.
extra_env = {SCHEMA_ENV: str(schema_path), EXPECTED_ENV: ",".join(request.socle)}
# Le registre AVANT le processus : un runner tue pendant la phase laisse
# cette ligne, et la reprise sait quelle session poursuivre (A9).
declare_session(session_id, request.cwd, request.model)
try:
proc = _spawn(_pi_argv(request, session_id), request,
extra_env=extra_env, stdin_text=request.prompt)
except HarnessError as error:
error.session_id = session_id
raise
reading = _read_pi_stream(proc.stdout.splitlines())
return _judge(reading, request, session_id, proc.returncode, proc.stderr)
def _claude_evidence(stdout: str, stderr: str) -> str:
"""Le motif d'un echec claude — pure. Le JSON de stdout d'abord, stderr ensuite.
En --output-format json, claude ecrit son erreur dans l'objet de stdout
(result, error) et n'envoie sur stderr que des avertissements (« Ignoring
N permissions.allow entries… ») : lire stderr d'abord masquerait la vraie
cause — une cle absente, un depot non approuve, un modele inconnu.
"""
try:
payload = json.loads(stdout)
for key in ("result", "error", "message"):
if isinstance(payload, dict) and payload.get(key):
return str(payload[key]).strip()[-400:]
except (json.JSONDecodeError, TypeError):
pass
lines = [line for line in stderr.strip().splitlines() if not line.startswith("Ignoring ")]
return ("\n".join(lines).strip() or stderr.strip() or stdout.strip())[-400:]
def _run_claude(request: HarnessRequest) -> HarnessResult:
# Claude Code : c'est LUI qui nomme la session. On la poursuit en rendant
# son session_id via --resume. Pas d'outil terminal type de ce cote :
# l'enveloppe reste une convention de texte, parse() la lit en secours.
#
# Regime par defaut : natif Anthropic — sa propre authentification, des
# modeles Anthropic. Le pointer sur la passerelle est possible (trois
# variables : ANTHROPIC_BASE_URL, ANTHROPIC_AUTH_TOKEN, et
# ANTHROPIC_API_KEY explicitement vide), mais la compatibilite n'est
# garantie que sur les modeles Anthropic : l'adaptateur polyglotte de
# l'usine reste pi, et ce choix-la appartient a votre environnement,
# pas a cet adaptateur.
cmd = ["claude", "-p", "--output-format", "json"]
if request.model:
# Le dialecte claude ignore le fournisseur : provider/id -> id.
cmd += ["--model", request.model.split("/", 1)[-1]]
if request.tools:
# Traduire puis dedoublonner en gardant l'ordre : bash et ls donnent
# tous deux Bash, inutile de le declarer deux fois.
allowed = list(dict.fromkeys(
CLAUDE_TOOLS[tool] for tool in request.tools if tool in CLAUDE_TOOLS))
cmd += ["--allowedTools", ",".join(allowed)]
if request.session_id:
cmd += ["--resume", request.session_id]
# L'echelle de reflexion devient un budget de tokens — l'asymetrie reste
# dans l'adaptateur, le roster n'en sait rien.
extra_env = ({"MAX_THINKING_TOKENS": str(THINKING_TOKENS[request.thinking])}
if request.thinking in THINKING_TOKENS else None)
# Le prompt part par stdin, PAS dans argv : c'est un mode documente de
# claude -p, et le seul qui survive aux shims .cmd de Windows.
proc = _spawn(cmd, request, extra_env, stdin_text=request.prompt)
if proc.returncode != 0:
# Le JSON de stdout d'abord, stderr ensuite : les avertissements de
# claude (« Ignoring … ») ne doivent pas masquer la vraie cause.
evidence = _claude_evidence(proc.stdout, proc.stderr)
raise HarnessError(f"claude a rendu {proc.returncode} : {evidence}",
request.session_id)
try:
payload = json.loads(proc.stdout)
except json.JSONDecodeError:
raise HarnessError("claude n'a pas rendu l'objet JSON attendu "
"(--output-format json)", request.session_id) from None
usage = payload.get("usage") or {}
return HarnessResult(text=str(payload.get("result", "")),
session_id=str(payload.get("session_id", "")),
cost_usd=float(payload.get("total_cost_usd") or 0.0),
returncode=proc.returncode,
tokens=int(usage.get("input_tokens") or 0)
+ int(usage.get("output_tokens") or 0),
stop_reason="error" if payload.get("is_error") else "stop")
# Le registre des adaptateurs. Un harnais de plus = une fonction + une ligne.
ADAPTERS = {"pi": _run_pi, "claude": _run_claude}
if __name__ == "__main__":
# La gate du module — zero token, sans pi : les fonctions pures et le registre.
# Lancer depuis la racine : uv run python -m adws.adw_modules.harness
import tempfile
request = HarnessRequest(prompt="- une ligne qui commence par un tiret",
model="z-ai/glm-5.3", tools=("read", "bash"))
argv = _pi_argv(request, "sess-1")
assert "--approve" in argv and argv[-1] == "--", argv
assert Path(argv[argv.index("--session-dir") + 1]).is_absolute()
assert argv[argv.index("--tools") + 1] == "read,bash,report_phase"
assert argv[argv.index("--model") + 1] == "openrouter/z-ai/glm-5.3"
assert request.prompt not in argv # le prompt part par stdin, jamais par argv
assert argv[4:-1] == _pi_flags(request, "sess-1") # un seul jeu de drapeaux, deux modes
def inventory_line(present: list[str]) -> str:
return json.dumps({"type": "entry_appended", "entry": {
"type": "custom", "customType": INVENTORY_ENTRY,
"data": {"pi_version": "0.85.0", "present": present,
"missing": [n for n in SOCLE if n not in present]}}})
assistant = json.dumps({"type": "message_end", "message": {
"role": "assistant", "provider": "openrouter", "model": "z-ai/glm-5.3",
"content": [{"type": "text", "text": "je lis"}], "stopReason": "toolUse",
"usage": {"totalTokens": 1200, "cost": {"total": 0.002}}}})
envelope = json.dumps({"type": "tool_execution_end", "toolName": "report_phase",
"isError": False,
"result": {"details": {"status": "success", "summary": "fini"}}})
# 1. Le socle au complet : l'enveloppe passe, les chiffres sont lus.
reading = _read_pi_stream([inventory_line(list(SOCLE)), assistant, envelope, "pas du JSON"])
assert reading.inventory and reading.inventory["missing"] == []
result = _judge(reading, request, "sess-1", 0, "")
assert result.envelope == {"status": "success", "summary": "fini"}, result
assert result.tokens == 1200 and result.stop_reason == "toolUse"
assert reading.observed_model == _route(request.model, request.direct)
# 2. Pas d'inventaire : refus, MEME avec une enveloppe valide — fail-closed.
try:
_judge(_read_pi_stream([assistant, envelope]), request, "sess-2", 0, "")
raise AssertionError("un flux sans inventaire doit etre refuse")
except HarnessError as error:
assert "socle pi non charge" in str(error) and error.session_id == "sess-2"
# 3. Un manquant (factory-guard renomme) : refus motive, session conservee.
partial = [n for n in SOCLE if n != "factory-guard"]
try:
_judge(_read_pi_stream([inventory_line(partial), assistant, envelope]), request, "sess-3", 0, "")
raise AssertionError("un socle incomplet doit etre refuse")
except HarnessError as error:
assert "manquants ['factory-guard']" in str(error) and error.session_id == "sess-3"
# 4. socle=() : aucune exigence — le chemin d'un run « a sec » ou d'un harnais sans socle.
free = HarnessRequest(prompt="ping", socle=())
assert _judge(_read_pi_stream([assistant]), free, "sess-4", 0, "").text == "je lis"
# 5. Un arret aborted (coupe-circuit A7) reste une HarnessError porteuse de session.
aborted = _read_pi_stream([inventory_line(list(SOCLE)), json.dumps({"type": "message_end", "message": {
"role": "assistant", "content": [], "stopReason": "aborted", "usage": {}}})])
try:
_judge(aborted, request, "sess-5", 0, "")
raise AssertionError("aborted doit lever")
except HarnessError as error:
assert "aborted" in str(error) and error.session_id == "sess-5"
# 6. Le registre (A9) : declare avant l'appel, retrouve, et une reprise
# hors dossier refusee — dans un dossier temporaire, jamais le vrai registre.
with tempfile.TemporaryDirectory() as tmp:
ledger, sessions = Path(tmp) / "ledger.jsonl", Path(tmp) / "sessions"
home, elsewhere = Path(tmp) / "usine", Path(tmp) / "worktree"
home.mkdir(), elsewhere.mkdir(), sessions.mkdir()
os.environ[PHASE_ENV], os.environ[ADW_ENV] = "build", "9f2e4a10"
entry = declare_session("sess-6", home, "z-ai/glm-5.3", ledger=ledger)
del os.environ[PHASE_ENV], os.environ[ADW_ENV]
assert entry["phase"] == "build" and entry["adw_id"] == "9f2e4a10"
assert session_origin("sess-6", ledger, sessions) == str(home.resolve())
resume_requires_cwd("sess-6", home, ledger, sessions) # meme dossier : passe
resume_requires_cwd("sess-inconnue", elsewhere, ledger, sessions) # inconnue : passe
try:
resume_requires_cwd("sess-6", elsewhere, ledger, sessions)
raise AssertionError("une reprise hors dossier doit etre refusee")
except HarnessError as error:
assert "reprise refusee" in str(error) and error.session_id == "sess-6"
# Une session d'avant A9 : pas dans le registre, mais son en-tete pi suffit.
(sessions / "2026-09-05T08-00-00_sess-7.jsonl").write_text(
json.dumps({"type": "session", "version": 3, "id": "sess-7", "cwd": str(home)}) + "\n",
encoding="utf-8")
assert session_origin("sess-7", ledger, sessions) == str(home)
# La route directe (17bis) : un profil hors passerelle envoie l'identifiant tel quel.
assert _route("deepseek/deepseek-v4-flash-0731") == "openrouter/deepseek/deepseek-v4-flash-0731"
assert _route("deepseek/deepseek-v4-flash-0731", direct=True) == "deepseek/deepseek-v4-flash-0731"
assert _route("zai/glm-5.3", direct=True) == "zai/glm-5.3"
# Le coffre (17bis) : sans profil, tout .env passe ; avec profil, seul le profil passe —
# et ce que le port ou l'ADW pose lui-meme pour la phase passe toujours.
base = {"PATH": "/usr/bin", "OPENROUTER_API_KEY": "sk-or-env", "ZAI_API_KEY": "zai-env",
"TERM": "xterm"}
vault = {"OPENROUTER_API_KEY", "ZAI_API_KEY"}
legacy = node_environment(base, {"MAX_THINKING_TOKENS": "8192"}, None, vault)
assert legacy["OPENROUTER_API_KEY"] == "sk-or-env" and legacy["ZAI_API_KEY"] == "zai-env"
node = node_environment(base, {"MAX_THINKING_TOKENS": "8192"}, {"OPENROUTER_API_KEY": "sk-or-env"}, vault)
assert node["OPENROUTER_API_KEY"] == "sk-or-env" and "ZAI_API_KEY" not in node
assert node["PATH"] == "/usr/bin" and node["TERM"] == "xterm" and node["MAX_THINKING_TOKENS"] == "8192"
session = node_environment(base, None, {}, vault) # regime session : rien a injecter, coffre ferme
assert "OPENROUTER_API_KEY" not in session and "ZAI_API_KEY" not in session
exported = node_environment(base, None, {}, set()) # une variable de l'environnement reel reste a vous
assert exported["OPENROUTER_API_KEY"] == "sk-or-env"
print("harness OK — argv v4 (approve, stdin, --, report_phase) ; verdict v5 : socle "
f"prouve ({len(SOCLE)} extensions), refus sans inventaire, refus sur manquant, "
f"enveloppe par la porte ({result.tokens} jetons), arret 'aborted' detecte ; "
"registre v6 : declare avant l'appel, reprise hors dossier refusee"
" ; profil 17bis : route directe hors passerelle, coffre .env ferme sous profil, ouvert sans")Pièce — adws/adw_modules/harness_rpc.py
Nouveau. La classe PiRpc ouvre pi --mode rpc avec les drapeaux du port, refuse toute demande
de dialogue et rend chaque tour au même _judge. Sa gate n’a pas besoin de pi : avec --fake, le
module joue lui-même un faux pi RPC et se pilote de bout en bout (une amorce, un dialogue refusé,
deux clones).
"""harness_rpc — la session vivante : un processus pi, plusieurs tours, des clones (A9).
Le port harness.run() garde un processus par tentative : c'est l'isolation,
le kill propre, la trace bornee. Ce module ouvre l'autre porte de pi,
--mode rpc, pour le seul usage qui la justifie dans l'usine : le best-of-N
par clone (adw_bon.py) — une amorce payee une fois, N bras qui en partent.
Meme lecture du flux (_read_pi_stream), meme verdict (_judge) : le RPC
change la maniere de parler a pi, pas la maniere de le juger.
Protocole (pi 0.85, doc rpc.md) : une commande JSON par ligne sur stdin ;
une reponse {"type": "response", "command": ..., "success": ...} par
commande, correlee par "id" ; les evenements de session en JSONL sur
stdout, jusqu'a agent_settled. Le separateur est "\\n" seul. Toute demande
d'interface (extension_ui_request : select, confirm, input, editor) recoit
un refus (cancelled) et est journalisee : ask = refus en headless, la
decision du chapitre A3 tient sans humain dans la boucle.
Gate (zero token, sans pi) : uv run python -m adws.adw_modules.harness_rpc
— le module se lance lui-meme comme un faux pi RPC (--fake) et se pilote.
"""
from __future__ import annotations
import json
import os
import queue
import shutil
import subprocess
import sys
import threading
import time
import uuid
from dataclasses import dataclass, field, replace
from pathlib import Path
from . import envelopes, harness
from .harness import HarnessError, HarnessRequest, HarnessResult, PiReading
# Les methodes de dialogue du sous-protocole d'interface : elles BLOQUENT pi
# jusqu'a la reponse. notify, setStatus, setWidget... n'attendent rien.
DIALOGS = ("select", "confirm", "input", "editor")
SETTLED = "agent_settled" # plus rien ne suivra automatiquement : fin du tour
def rpc_argv(request: HarnessRequest, session_id: str) -> list[str]:
"""La ligne de commande d'une session vivante : memes drapeaux qu'une phase, autre mode."""
return ["pi", "--mode", "rpc", *harness._pi_flags(request, session_id)]
@dataclass
class Turn:
"""Ce qu'un tour RPC rend au runner : la lecture du port, le brut, la duree."""
reading: PiReading
lines: list[str] = field(default_factory=list)
seconds: float = 0.0
class PiRpc:
"""Un processus pi --mode rpc, pilote en JSONL. A fermer — `with` s'en charge.
Le processus herite de l'environnement du runner (schema de la porte,
socle attendu, limites du roster) UNE fois, a l'ouverture : une
enveloppe par processus, une garde par processus. Chaque clone rejoue
session_start — compteurs remis a zero, inventaire redepose — et se
juge avec les memes preuves qu'une phase ordinaire.
"""
def __init__(self, request: HarnessRequest, session_id: str | None = None,
extra_env: dict[str, str] | None = None,
executable: list[str] | None = None,
ledger: Path = harness.LEDGER) -> None:
self.request = request
self.session_id = session_id or request.session_id or str(uuid.uuid4())
self._argv = executable or rpc_argv(request, self.session_id)
self._extra_env = dict(extra_env or {})
self._ledger = ledger
self.proc: subprocess.Popen[str] | None = None
self._lines: queue.Queue[str | None] = queue.Queue()
self._env: dict[str, str] = {}
self._seq = 0
self.refused: list[dict] = [] # les demandes d'interface refusees (ask = refus)
def __enter__(self) -> "PiRpc":
self.start()
return self
def __exit__(self, *_exc) -> None:
self.close()
# ---------- ouverture / fermeture ----------
def start(self) -> None:
program = shutil.which(self._argv[0]) or self._argv[0]
if not Path(program).exists() and shutil.which(program) is None:
raise HarnessError(f"{self._argv[0]!r} introuvable dans le PATH — pi est-il installe ?",
self.session_id)
harness.SESSION_DIR.mkdir(parents=True, exist_ok=True)
# Le meme contexte d'environnement qu'une phase : la porte, le socle,
# puis ce que l'ADW ajoute (limites, identite de phase).
schema_path = harness._write_schema(self.request.schema
or envelopes.schema(envelopes.Envelope))
env = {**os.environ, harness.SCHEMA_ENV: str(schema_path),
harness.EXPECTED_ENV: ",".join(self.request.socle), **self._extra_env}
# Le registre AVANT le processus, comme pour une phase (A9).
self._env = env
harness.declare_session(self.session_id, self.request.cwd, self.request.model,
ledger=self._ledger, env=env)
# stderr va dans un fichier, jamais dans un tube que personne ne lit :
# un processus long qui remplit un tube ferme finit par se bloquer.
self._stderr_path = harness.SESSION_DIR / f"{self.session_id}.stderr.log"
self._stderr = self._stderr_path.open("w", encoding="utf-8")
self.proc = subprocess.Popen(
[program, *self._argv[1:]], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=self._stderr, text=True, encoding="utf-8", errors="replace",
env=env, cwd=self.request.cwd, bufsize=1)
threading.Thread(target=self._pump, daemon=True).start()
def close(self) -> None:
"""Fin d'entree = arret propre : pi emet session_shutdown (le miroir A4 tourne)."""
if self.proc is None:
return
try:
self.proc.stdin.close()
except OSError:
pass
try:
self.proc.wait(timeout=30)
except subprocess.TimeoutExpired:
self.proc.kill()
self.proc.wait()
self._stderr.close()
self.proc = None
def _pump(self) -> None:
# Le lecteur du tube, dans son fil : stdout est vide en continu, pi
# ne se bloque jamais sur une sortie pleine. "\n" seul separe.
assert self.proc is not None and self.proc.stdout is not None
for line in self.proc.stdout:
self._lines.put(line.rstrip("\r\n"))
self._lines.put(None)
def _stderr_tail(self) -> str:
try:
self._stderr.flush()
return self._stderr_path.read_text(encoding="utf-8", errors="replace")[-400:].strip()
except (OSError, ValueError, AttributeError):
return ""
# ---------- le protocole ----------
def _write(self, obj: dict) -> None:
assert self.proc is not None and self.proc.stdin is not None
self.proc.stdin.write(json.dumps(obj, ensure_ascii=False) + "\n")
self.proc.stdin.flush()
def _next(self, deadline: float) -> tuple[dict, str]:
"""La prochaine ligne JSON du flux — les demandes de dialogue sont refusees en passant."""
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise HarnessError(f"pi (rpc) muet apres {self.request.timeout} s", self.session_id)
try:
line = self._lines.get(timeout=min(remaining, 1.0))
except queue.Empty:
if self.proc is not None and self.proc.poll() is not None:
raise HarnessError(f"pi (rpc) s'est arrete ({self.proc.returncode}) : "
f"{self._stderr_tail()}", self.session_id) from None
continue
if line is None:
raise HarnessError(f"pi (rpc) a ferme sa sortie : {self._stderr_tail()}",
self.session_id)
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
if event.get("type") == "extension_ui_request" and event.get("method") in DIALOGS:
# ask = refus : personne n'est la pour repondre, et pi attendrait.
self.refused.append(event)
self._write({"type": "extension_ui_response", "id": event.get("id"),
"cancelled": True})
continue
return event, line
def _command(self, payload: dict, timeout: float = 60) -> tuple[dict, list[str]]:
"""Envoie une commande, attend SA reponse ; rend aussi les evenements croises."""
self._seq += 1
rid = f"c{self._seq}"
self._write({"id": rid, **payload})
deadline, seen = time.monotonic() + timeout, []
while True:
event, line = self._next(deadline)
if event.get("type") == "response" and event.get("id") == rid:
if not event.get("success"):
raise HarnessError(f"pi (rpc) refuse {payload['type']} : {event.get('error')}",
self.session_id)
return (event.get("data") or {}), seen
seen.append(line) # un evenement du tour, deja en route : il compte
def command(self, payload: dict, timeout: float = 60) -> dict:
return self._command(payload, timeout)[0]
# ---------- ce que l'usine emploie ----------
def prompt(self, message: str) -> Turn:
"""Un tour complet : la demande, puis tout le flux jusqu'a agent_settled — lu comme le port."""
clock = time.monotonic()
_, seen = self._command({"type": "prompt", "message": message}) # accepte, ou refuse
lines, deadline = list(seen), clock + self.request.timeout
while True:
event, line = self._next(deadline) # les demandes d'interface y sont refusees
lines.append(line)
if event.get("type") == SETTLED: # agent_settled : plus rien ne suivra
break
return Turn(reading=harness._read_pi_stream(lines), lines=lines,
seconds=time.monotonic() - clock)
def judge(self, turn: Turn, request: HarnessRequest | None = None) -> HarnessResult:
"""Le verdict du port, a l'identique : socle prouve, porte typee, modele observe."""
return harness._judge(turn.reading, request or self.request, self.session_id, 0, "")
def state(self) -> dict:
return self.command({"type": "get_state"})
def stats(self) -> dict:
"""Jetons et cout cumules de la session courante, tels que pi les tient."""
return self.command({"type": "get_session_stats"})
def clone(self) -> dict:
"""Une session neuve, copie de la branche courante a sa position : le prefixe partage.
pi bascule dessus et relie les extensions (session_start rejoue). Le
nouvel identifiant est lu dans get_state et declare au registre.
"""
data = self.command({"type": "clone"})
if data.get("cancelled"):
raise HarnessError("clone annule par une extension", self.session_id)
state = self.state()
self.session_id = str(state.get("sessionId") or self.session_id)
harness.declare_session(self.session_id, self.request.cwd, self.request.model,
ledger=self._ledger, env=self._env)
return state
def switch_session(self, session_file: str) -> dict:
"""Revenir a une session par son fichier (celui de l'amorce, entre deux clones)."""
data = self.command({"type": "switch_session", "sessionPath": session_file})
if data.get("cancelled"):
raise HarnessError("switch_session annule par une extension", self.session_id)
state = self.state()
self.session_id = str(state.get("sessionId") or self.session_id)
return state
def set_model(self, model: str) -> HarnessRequest:
"""Change le moteur de la session courante ; rend la requete a juger avec ce modele."""
provider, _, model_id = harness._route(model).partition("/")
self.command({"type": "set_model", "provider": provider, "modelId": model_id})
return replace(self.request, model=model)
def set_thinking(self, level: str) -> None:
self.command({"type": "set_thinking_level", "level": level})
def abort(self) -> None:
"""L'arret externe : pi interrompt le tour et repond quand il est inactif."""
self.command({"type": "abort"}, timeout=30)
# ---------- un faux pi RPC pour la gate : meme framing, meme protocole, zero jeton ----------
def _fake_pi() -> int:
"""Joue pi --mode rpc : repond aux commandes, diffuse un tour scripte par prompt."""
out = sys.stdout
turns, made, current = 0, 0, 0
# Le modele demande sur la ligne de commande est celui qu'on observe —
# comme le vrai pi quand l'id est exact (A6 verifie l'inverse).
route = sys.argv[sys.argv.index("--model") + 1] if "--model" in sys.argv else "openrouter/fake/model"
provider, _, model = route.partition("/")
def emit(obj: dict) -> None:
out.write(json.dumps(obj) + "\n")
out.flush()
def ok(rid, command, data=None):
response = {"id": rid, "type": "response", "command": command, "success": True}
if data is not None:
response["data"] = data
emit(response)
while True:
raw = sys.stdin.readline()
if not raw:
return 0 # fin d'entree : arret propre
cmd = json.loads(raw)
rid, kind = cmd.get("id"), cmd.get("type")
if kind == "prompt":
emit({"type": "agent_start"}) # deja en route avant la reponse : le client le garde
ok(rid, "prompt")
emit({"type": "entry_appended", "entry": {"type": "custom", "customType":
harness.INVENTORY_ENTRY, "data": {"pi_version": "fake",
"present": list(harness.SOCLE), "missing": []}}})
if turns == 0:
# Un damage control en mode ask : pi BLOQUE jusqu'a la reponse.
emit({"type": "extension_ui_request", "id": "ui-1", "method": "confirm",
"title": "rm -rf adws ?"})
answer = json.loads(sys.stdin.readline())
assert answer.get("cancelled") is True, answer
emit({"type": "message_end", "message": {
"role": "assistant", "provider": provider, "model": model,
"content": [{"type": "text", "text": cmd["message"][:24]}],
"stopReason": "toolUse", "usage": {"totalTokens": 100, "cost": {"total": 0.001}}}})
emit({"type": "tool_execution_end", "toolName": envelopes.REPORT_TOOL,
"isError": False, "result": {"details": {"status": "success",
"summary": f"tour {turns}"}}})
emit({"type": SETTLED})
turns += 1
elif kind == "clone":
made += 1
current = made
ok(rid, "clone", {"cancelled": False})
elif kind == "switch_session":
current = int(Path(cmd["sessionPath"]).stem.split("-")[-1])
ok(rid, "switch_session", {"cancelled": False})
elif kind == "get_state":
ok(rid, "get_state", {"sessionId": f"fake-{current}",
"sessionFile": f"/fake/session-{current}.jsonl"})
elif kind == "get_session_stats":
ok(rid, "get_session_stats", {"cost": round(0.001 * turns, 3),
"tokens": {"total": 100 * turns}})
elif kind in ("abort", "set_model", "set_thinking_level"):
ok(rid, kind)
else:
emit({"id": rid, "type": "response", "command": kind, "success": False,
"error": f"commande inconnue {kind}"})
if __name__ == "__main__":
if "--fake" in sys.argv:
raise SystemExit(_fake_pi())
# La gate du module — zero token, sans pi : argv pur, puis un lot complet
# pilote sur le faux pi. Lancer depuis la racine :
# uv run python -m adws.adw_modules.harness_rpc
import tempfile
request = HarnessRequest(prompt="", model="fake/model", tools=("read", "bash"), timeout=20)
argv = rpc_argv(request, "sess-rpc")
assert argv[:3] == ["pi", "--mode", "rpc"] and "--approve" in argv, argv
assert "-p" not in argv and "--" not in argv # pas de prompt positionnel en RPC
assert argv[argv.index("--tools") + 1] == "read,bash,report_phase"
fake = [sys.executable, "-m", "adws.adw_modules.harness_rpc", "--fake"]
with tempfile.TemporaryDirectory() as tmp:
with PiRpc(request, session_id="sess-rpc", executable=fake,
ledger=Path(tmp) / "ledger.jsonl") as pi:
amorce = pi.judge(pi.prompt("lis la spec")) # tour 0 : l'amorce
assert amorce.envelope == {"status": "success", "summary": "tour 0"}, amorce
assert amorce.tokens == 100 and amorce.stop_reason == "toolUse"
assert len(pi.refused) == 1 and pi.refused[0]["method"] == "confirm" # ask = refus
base = pi.state()["sessionFile"]
arms = []
for k in range(2):
if k:
pi.switch_session(base) # retour a l'amorce
pi.clone() # un bras = un clone
arms.append(pi.judge(pi.prompt(f"bras {k}")).envelope["summary"])
assert arms == ["tour 1", "tour 2"] and pi.session_id == "fake-2", (arms, pi.session_id)
assert pi.stats()["tokens"]["total"] == 300
declared = (Path(tmp) / "ledger.jsonl").read_text(encoding="utf-8").splitlines()
assert len(declared) == 3 # l'amorce et les deux clones
print("harness_rpc OK — argv rpc (approve, sans -p ni --), un tour lu et juge comme une "
f"phase ({amorce.tokens} jetons), une demande de dialogue refusee, deux clones "
"depuis la meme amorce, trois sessions au registre")Pièce — adws/adw_bon.py
Nouveau, le pendant local de sandbox_bestof.py (chapitre 25). Il réutilise sans les modifier
adw_build.py (chapitre 27), adw_scout.py (A7), les gates du chapitre 12 et le roster. Il exige
un arbre propre, specs non suivies tolérées. Une amorce, puis un bras par clone, chacun une phase
du runner avec ses jetons et son coût. La moisson classe les bras. --demo écrit une spec d’essai
et --dry éprouve le classement, sans jeton. Le processus pi est toujours fermé à la fin.
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["pyyaml"]
# ///
"""adw_bon — best-of-N par clone, dans une session vivante (annexe A9).
Usage :
uv run adws/adw_bon.py specs/ma-spec.md [--arms 3] [--models a/b,c/d] [--config ...]
uv run adws/adw_bon.py --demo # ecrit specs/bon-demo.md, la spec d'essai, puis sort
uv run adws/adw_bon.py --dry # la gate a sec : classement et lot fictifs, zero token
Un seul processus pi (--mode rpc, harness_rpc.py). D'abord l'AMORCE : le
builder lit la spec et les fichiers qu'elle nomme, n'ecrit rien, rend une
enveloppe par la porte typee — c'est le prefixe partage, paye une fois.
Puis N BRAS : chacun est un clone de l'amorce (meme contexte lu, zero
relecture), implemente, est juge par les gates du roster ; son patch est
moissonne, l'arbre est remis a zero. La MOISSON classe : vert d'abord, le
moins cher, puis le plus rapide. Le code choisit ; vous tranchez.
Le pendant local du best-of-N hors-site du chapitre 25 (N boites, N
rosters) : ici une seule variable bouge — le tirage, ou le modele par bras
avec --models — et l'amorce est commune.
"""
import argparse
import json
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass, field, replace
from pathlib import Path
from adw_build import BUILDER_BRIEF, BuildEnvelope
from adw_modules import envelopes, gates, harness, harness_rpc, harness_trace, roster
from adw_modules.envelopes import EnvelopeError
from adw_modules.runner import PhaseFailure, PhaseSpec, Run
from adw_scout import phase_env
BON_DIR = Path("adws/adw_data/bon") # un dossier par lot : patches et moisson — jamais commite
DEMO_SPEC = Path("specs/bon-demo.md")
DEMO_TEXT = """# Compteur de mots
Dans le dossier du payload, ajouter un module `src/word-count.ts` qui exporte une
fonction `wordCount(text: string): number` : le nombre de mots separes par des blancs
(espaces, tabulations, retours a la ligne), 0 pour une chaine vide ou faite de blancs.
Ajouter son test `src/word-count.test.ts` (bun test) : chaine vide, un mot, plusieurs
blancs consecutifs, retours a la ligne.
Ne toucher a aucun autre fichier. Verifier avec la suite de tests du payload.
"""
AMORCE_BRIEF = """Tu es le builder de l'usine. Cette premiere etape est une LECTURE.
- Lis la spec EN ENTIER : {spec}
- Lis chaque fichier qu'elle nomme, et les tests existants du payload ({payload}).
- N'ecris RIEN et ne lance aucune commande qui modifie le depot.
- Quand tu as tout lu, resume en cinq lignes au plus ce que tu changeras
(dans 'summary'), avec changed_files: [] — tu n'as encore rien modifie."""
BRAS_ASK = """Tu as deja lu la spec et les fichiers concernes : implemente maintenant
la spec {spec}, sans la relire, exactement, rien de plus.
"""
# ---------- git : l'arbre propre, le patch de chaque bras, la remise a zero ----------
def git(*args: str) -> str:
proc = subprocess.run(["git", *args], capture_output=True, text=True, encoding="utf-8")
if proc.returncode != 0:
raise PhaseFailure(f"git {' '.join(args)} : {proc.stderr.strip()[-300:]}")
return proc.stdout
def dirty_lines() -> list[str]:
"""Ce qui sale l'arbre — sauf les specs non suivies, que le planner depose la."""
return [line for line in git("status", "--porcelain").splitlines()
if not line.startswith("?? specs/")]
def harvest_patch(target: Path) -> int:
"""Le patch du bras — tout sauf specs/ — puis l'index remis tel qu'il etait."""
git("add", "-A", "--", ".", ":(exclude)specs")
patch = git("diff", "--cached", "--binary")
git("reset", "-q")
target.write_text(patch, encoding="utf-8")
return len(patch.encode("utf-8"))
def reset_tree() -> None:
"""L'arbre revient a HEAD : fichiers suivis restaures, nouveaux supprimes — specs/ preservee."""
git("checkout", "--", ".")
git("clean", "-qfd", "-e", "specs")
# ---------- le lot : ce que les phases partagent ----------
@dataclass
class Lot:
spec: str
builder: roster.AgentSpec
factory: roster.Roster
models: list[str]
run_dir: Path
pi: harness_rpc.PiRpc | None = None
request: harness.HarnessRequest | None = None
base_file: str = "" # le fichier de session de l'amorce
amorce: dict = field(default_factory=dict)
results: list[dict] = field(default_factory=list)
def close(self) -> None:
if self.pi is not None:
self.pi.close() # session_shutdown : le miroir A4 tourne
self.pi = None
try:
# L'ingestion cote runner, apres la fermeture : idempotente (A4),
# elle rattrape ce que le miroir n'aurait pas eu le temps d'ecrire.
harness_trace.ingest(harness_trace.connect())
except Exception as error: # best effort : le brut est deja sur le disque
print(f"[bon] ingestion differee : {error}", file=sys.stderr)
# ---------- les phases ----------
def constat(lot: Lot):
"""Phase code : l'arbre est propre, la spec existe, le dossier du lot est cree — zero token."""
def action(run: Run, attempt: int):
git("rev-parse", "--is-inside-work-tree")
dirty = dirty_lines()
if dirty:
raise PhaseFailure("arbre de travail sale — commitez ou remisez d'abord, le lot "
f"remet l'arbre a zero entre deux bras :\n" + "\n".join(dirty[:10]))
spec = Path(lot.spec)
if not spec.is_file() or spec.stat().st_size == 0:
raise PhaseFailure(f"spec introuvable ou vide : {lot.spec}")
lot.run_dir.mkdir(parents=True, exist_ok=True)
return {"spec": lot.spec, "arms": len(lot.models)}
return action
def amorce(lot: Lot):
"""Phase agent : ouvrir la session vivante, lire — le prefixe que tous les bras partagent."""
def action(run: Run, attempt: int):
agent = lot.builder
lot.request = harness.HarnessRequest(
prompt="", model=agent.model, thinking=agent.thinking, tools=agent.tools,
timeout=agent.timeout, schema=envelopes.schema(BuildEnvelope))
# Les limites du roster et l'identite de la phase, par l'environnement
# (A7) — fixes pour tout le processus : une garde par processus.
lot.pi = harness_rpc.PiRpc(lot.request, extra_env=phase_env("amorce", agent, run))
lot.pi.start()
run.sessions["amorce"] = lot.pi.session_id
ask = (AMORCE_BRIEF.format(spec=lot.spec, payload=lot.factory.payload.dir)
+ "\n\n" + envelopes.contract(BuildEnvelope))
try:
turn = lot.pi.prompt(ask)
result = lot.pi.judge(turn)
except harness.HarnessError as error:
raise PhaseFailure(str(error)) from None
run.cost_usd += result.cost_usd
run.tokens += result.tokens
lot.amorce = {"tokens": result.tokens, "cost_usd": round(result.cost_usd, 4),
"seconds": round(turn.seconds, 1), "session_id": lot.pi.session_id}
lot.base_file = str(lot.pi.state().get("sessionFile") or "")
if not lot.base_file:
raise PhaseFailure("pi n'a pas rendu le fichier de session de l'amorce")
payload = result.envelope if result.envelope is not None else result.text
try:
return envelopes.parse(payload, envelopes.Envelope) # lecture : l'enveloppe de base
except EnvelopeError as error:
raise PhaseFailure(f"enveloppe d'amorce invalide : {error}") from None
return action
def bras(lot: Lot, k: int, model: str):
"""Phase agent : un clone de l'amorce, une implementation, les gates, le patch, la remise a zero.
Un bras rouge n'arrete pas le lot : son verdict et son motif sont dans le
resultat et dans la trace (evenement bon_bras) ; la moisson classera.
"""
def action(run: Run, attempt: int):
pi, request = lot.pi, lot.request
assert pi is not None and request is not None
try:
if k > 1:
pi.switch_session(lot.base_file) # retour a l'amorce
pi.clone() # session_start rejoue : garde et inventaire a neuf
if model != lot.builder.model:
request = pi.set_model(model) # une variable par bras, tenue par le code
except harness.HarnessError as error:
raise PhaseFailure(str(error)) from None # sans clone, pas de bras : le lot s'arrete
run.sessions[f"bras_{k}"] = pi.session_id
arm = {"bras": k, "model": model, "session_id": pi.session_id, "verdict": "rouge",
"cost_usd": 0.0, "tokens": 0, "seconds": 0.0, "patch_bytes": 0, "motif": ""}
clock = time.monotonic()
try:
turn = pi.prompt(BRAS_ASK.format(spec=lot.spec) + BUILDER_BRIEF + "\n\n"
+ envelopes.contract(BuildEnvelope))
result = pi.judge(turn, request)
arm["cost_usd"], arm["tokens"] = round(result.cost_usd, 4), result.tokens
payload = result.envelope if result.envelope is not None else result.text
envelope: BuildEnvelope = envelopes.parse(payload, BuildEnvelope)
if envelope.status != "success":
arm["motif"] = f"le builder declare un echec : {envelope.summary}"
else:
# Les gates du roster, sur l'arbre tel que le bras l'a laisse — zero token.
reports = [gates.changed_files_exist(envelope)]
reports += [gates.command(cmd, cwd)(envelope)
for cmd, cwd in lot.factory.payload.commands()]
arm["motif"] = gates.motif(reports)
arm["verdict"] = "vert" if not arm["motif"] else "rouge"
except (harness.HarnessError, EnvelopeError) as error:
arm["motif"] = str(error)
finally:
arm["seconds"] = round(time.monotonic() - clock, 1)
arm["patch_bytes"] = harvest_patch(lot.run_dir / f"bras-{k}.patch")
reset_tree()
run.cost_usd += arm["cost_usd"]
run.tokens += arm["tokens"]
run.tracer.event(run.adw_id, "bon_bras", f"bras_{k}", payload=arm)
print(f"[bon] bras {k} ({model}) : {arm['verdict']} — {arm['tokens']} jetons, "
f"{arm['cost_usd']:.4f} $, {arm['seconds']} s, patch {arm['patch_bytes']} o"
+ (f" — {arm['motif'].splitlines()[0][:80]}" if arm["motif"] else ""),
file=sys.stderr)
lot.results.append(arm)
return arm
return action
# ---------- la moisson : classer, chiffrer, proposer — zero token ----------
def rank(results: list[dict]) -> list[dict]:
"""Vert d'abord, puis le moins cher, puis le plus rapide. Pas de score compose."""
return sorted(results, key=lambda a: (a["verdict"] != "vert", a["cost_usd"], a["seconds"]))
def print_table(ranked: list[dict]) -> None:
print(f"{'bras':<5}{'modele':<32}{'verdict':<8}{'jetons':>9}{'$':>9}{'s':>7}{'patch':>8}")
for a in ranked:
print(f"{a['bras']:<5}{a['model'][:31]:<32}{a['verdict']:<8}{a['tokens']:>9}"
f"{a['cost_usd']:>9.4f}{a['seconds']:>7.1f}{a['patch_bytes']:>8}")
def moisson(lot: Lot):
def action(run: Run, attempt: int):
ranked = rank(lot.results)
print_table(ranked)
n = len(lot.results)
saved = lot.amorce.get("tokens", 0) * (n - 1)
print(f"amorce : {lot.amorce.get('tokens', 0)} jetons, {lot.amorce.get('cost_usd', 0):.4f} $, "
f"payee une fois — a froid elle l'aurait ete {n} fois : ~{saved} jetons de plus "
"(estimation, hors cache du fournisseur)")
out = lot.run_dir / "harvest.json"
out.write_text(json.dumps({"adw_id": run.adw_id, "spec": lot.spec, "amorce": lot.amorce,
"results": ranked}, indent=2, ensure_ascii=False),
encoding="utf-8")
winner = ranked[0]
if winner["verdict"] != "vert":
raise PhaseFailure(f"aucun bras vert sur {n} — motifs dans {out}")
patch = lot.run_dir / f"bras-{winner['bras']}.patch"
print(f"propose : bras {winner['bras']} ({winner['model']}) — vert, le moins cher des verts")
print(f"dispose : git apply --index {patch}")
return ranked
return action
# ---------- la gate a sec : le classement sur un lot fictif ----------
def selftest() -> int:
arms = [
{"bras": 1, "model": "m", "verdict": "rouge", "cost_usd": 0.05, "tokens": 40_000, "seconds": 60.0, "patch_bytes": 900, "motif": "bun test : 1 fail"},
{"bras": 2, "model": "m", "verdict": "vert", "cost_usd": 0.22, "tokens": 150_000, "seconds": 140.0, "patch_bytes": 1800, "motif": ""},
{"bras": 3, "model": "m", "verdict": "vert", "cost_usd": 0.19, "tokens": 130_000, "seconds": 170.0, "patch_bytes": 1700, "motif": ""},
]
ranked = rank(arms)
assert [a["bras"] for a in ranked] == [3, 2, 1], ranked # vert, puis le moins cher
assert rank(arms[:1])[0]["verdict"] == "rouge" # un lot tout rouge se classe quand meme
print_table(ranked)
print("adw_bon OK — classement : vert d'abord, le moins cher, puis le plus rapide ; "
"lot fictif, zero token")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="best-of-N par clone, en session vivante")
parser.add_argument("spec", nargs="?", help="la spec a implementer (sous specs/)")
parser.add_argument("--arms", type=int, default=3, help="nombre de bras, 2 a 5")
parser.add_argument("--models", default="",
help="un modele par bras, separes par des virgules (ids du registre)")
parser.add_argument("--config", default=str(roster.DEFAULT_PATH))
parser.add_argument("--demo", action="store_true", help=f"ecrire {DEMO_SPEC} puis sortir")
parser.add_argument("--dry", action="store_true", help="la gate a sec, zero token")
args = parser.parse_args()
if args.dry:
return selftest()
if args.demo:
DEMO_SPEC.parent.mkdir(exist_ok=True)
DEMO_SPEC.write_text(DEMO_TEXT, encoding="utf-8")
print(f"spec d'essai ecrite : {DEMO_SPEC}")
return 0
if not args.spec:
parser.error("spec manquante — uv run adws/adw_bon.py specs/ma-spec.md")
factory = roster.load(args.config) # zero token : tout echec est gratuit
builder = factory.agents["builder"]
models = ([m.strip() for m in args.models.split(",") if m.strip()]
if args.models else [builder.model] * args.arms)
if not 2 <= len(models) <= 5:
print("un best-of-N va de 2 a 5 bras — au-dela, la moisson n'est plus lisible",
file=sys.stderr)
return 1
run = Run(adw_id=uuid.uuid4().hex[:8])
lot = Lot(spec=args.spec, builder=builder, factory=factory, models=models,
run_dir=BON_DIR / run.adw_id)
phases = [PhaseSpec(name="constat_bon", kind="code", action=constat(lot)),
PhaseSpec(name="amorce", kind="agent", action=amorce(lot))]
phases += [PhaseSpec(name=f"bras_{k}", kind="agent", action=bras(lot, k, model))
for k, model in enumerate(models, start=1)]
phases.append(PhaseSpec(name="moisson", kind="code", action=moisson(lot)))
try:
return run.execute(phases)
finally:
lot.close() # toujours : le processus pi ne survit pas au lot
if __name__ == "__main__":
sys.exit(main())La gate du TP
Depuis la racine de plume-factory, sur un arbre propre (git status vide, hors specs). Une
commande par ligne, identiques dans bash et PowerShell. Les trois premières ne coûtent rien, la
quatrième écrit la spec d’essai, la cinquième lance trois bras, les deux dernières lisent la
trace.
uv run python -m adws.adw_modules.harness
uv run python -m adws.adw_modules.harness_rpc
uv run adws/adw_bon.py --dry
uv run adws/adw_bon.py --demo
uv run adws/adw_bon.py specs/bon-demo.md --arms 3
sqlite3 adws/adw_data/factory.db "SELECT name, status, tokens, cost_usd FROM phases WHERE adw_id = (SELECT adw_id FROM runs WHERE adw_name = 'adw_bon' ORDER BY started_at DESC LIMIT 1) ORDER BY seq;"
sqlite3 adws/adw_data/factory.db "SELECT COUNT(DISTINCT session_id) FROM harness_by_phase WHERE adw_id = (SELECT adw_id FROM runs WHERE adw_name = 'adw_bon' ORDER BY started_at DESC LIMIT 1);"Résultat attendu : harness OK — … registre v6 : declare avant l'appel, reprise hors dossier refusee, puis harness_rpc OK — argv rpc (approve, sans -p ni --), un tour lu et juge comme une phase (100 jetons), une demande de dialogue refusee, deux clones depuis la meme amorce, trois sessions au registre, puis adw_bon OK — classement : vert d'abord, le moins cher, puis le plus rapide.
Le lot affiche une ligne par bras, le tableau de la moisson, la ligne
amorce : … payee une fois — a froid elle l'aurait ete 3 fois, puis le bras proposé et la
commande git apply --index du patch gagnant. L’arbre revient propre. La première requête montre
amorce une seule fois, puis bras_1 à bras_3. La seconde compte quatre sessions.
Coût : rien pour les gates à sec, de cinquante centimes à un dollar pour le lot, en cinq à
dix minutes (les bras passent l’un après l’autre). Variante éco : --models avec trois fois le
modèle léger du roster, ou --arms 2. Pour éprouver le refus sans jeton, relancez un ADW avec une
session du registre depuis un autre dossier : pi ne démarre pas.
Quiz — teste tes connaissances
0/7
bonnes reponses
Objectif non atteint (minimum 5/7 requis).
Remonte relire la fiche memo en pretant attention aux points manques, puis cliquer sur « Recommencer » pour retenter.
Remonte relire la fiche memo en pretant attention aux points manques, puis cliquer sur « Recommencer » pour retenter.