Compaction structurée, fallback et doctor
Le dernier chapitre de l'annexe pi : un résumé de contexte qui garde mot pour mot ce que le runner a dit, un modèle de secours sur panne fournisseur, une version de pi épinglée, et un doctor qui vérifie tout à sec puis installe l'usine pi ailleurs.
Une session longue finit par remplir le contexte, et pi résume alors ses anciens messages. Or les plus anciens sont ceux du runner : la mission, le handoff, le contrat de la porte typée. Ce dernier chapitre de l’annexe pose trois pièces :
factory-compact.ts, sixième extension du socle, garde mot pour mot ce que le runner a dit.harness.py(remplace la version A9) refuse un pi trop vieux et passe sur un modèle de secours en cas de panne fournisseur, sans quitter la session.doctor.py(remplace la version du chapitre 4) vérifie tout cela à sec, puis installe l’usine pi dans un autre dépôt.
Préserver l’enveloppe à travers la compaction
Quand pi compacte, et ce qui se perd
Un build du chapitre 13 tourne sur le workhorse. Au quinzième tour, pi remplace les anciens
messages par un résumé : c’est la compaction. Le résumé dit « Goal : add a word counter to
the status bar ; Progress : … ; Next steps : run tests, then report ». Le contrat de la porte est
devenu « then report ». Deux tours plus tard, le modèle termine en prose, sans appeler
report_phase. Le runner renvoie correction(), et le modèle redemande la forme de
l’enveloppe, qui n’est plus dans son contexte. Cela coûte quelques centimes, ou un run rouge sur
un modèle moins docile.
Sur pi 0.85, la compaction part quand le contexte dépasse la fenêtre du modèle moins
compaction.reserveTokens (16 384 dans vos réglages du chapitre A1). pi vérifie ce seuil après
chaque lot d’outils, résume les messages anciens, garde les keepRecentTokens récents (12 000
chez vous) et reprend dans le même tour. /compact fait la même chose à la demande.
Le premier message d’une phase, celui du runner, est le plus ancien. Il part donc en premier. Le
résumé par défaut en fait une ligne « Goal ». La forme de l’enveloppe, la consigne « appelle
report_phase seul, en dernier » et les chemins autorisés sont réécrits par le modèle, ou omis.
Le bloc [FACTORY], recopié par le code
Avant de résumer, pi émet session_before_compact avec une préparation : messages à résumer,
résumé précédent, fichiers lus et modifiés. L’extension peut ne rien répondre (pi résume
lui-même), annuler par cancel, ou fournir sa propre compaction. factory-compact.ts fournit un
résumé en deux parties :
- Un bloc
[FACTORY]construit par le code, sans modèle. Phase, rôle et run viennent de l’environnement posé par la fabriqueagent_action(chapitre A7). La mission est le premier message utilisateur. Le dernier mot du runner est le dernier message utilisateur qui quitte le contexte, par exemple une correction. - Un récit du travail accompli, rédigé par le modèle de la session.
À la compaction suivante, la mission n’est plus dans les messages. L’extension la relit dans le bloc précédent, si bien qu’elle survit à N compactions, à l’identique.
pi cumule les fichiers lus et modifiés d’une compaction à l’autre, mais pas pour un résumé fourni par une extension. L’extension les relit donc dans le résumé précédent. Si le modèle ne peut pas résumer, un récit à sec, déterministe et à zéro jeton, le remplace. Le bloc reste.
La compaction coûte le même appel qu’avant, quelques centimes. Le bloc ajoute un à deux milliers de jetons par tour, une fraction de centime. Vous achetez un contrat qui ne se dégrade pas.
Deux compactions, ce qui survit
| Ce que le runner avait injecté | Compaction par défaut | Compaction structurée (socle) |
|---|---|---|
| La mission et le brief du rôle | paraphrasés en « Goal » | recopiés mot pour mot |
| Le contrat de la porte typée (forme exacte de l’enveloppe) | souvent réduit à « report at the end » | recopié mot pour mot |
| La dernière correction ou relance du runner | fondue dans « Progress » | recopiée mot pour mot |
| Fichiers lus / modifiés | cumulés par pi pour ses propres résumés | cumulés par l’extension, y compris ceux du résumé précédent |
| Le récit du travail accompli | rédigé par le modèle | rédigé par le modèle, ou inventaire à sec si le modèle échoue |
| Coût de la compaction | un appel de résumé | le même appel, plus un à deux milliers de jetons par tour suivant |
Config — le bloc [FACTORY] tel que le modèle le voit
Après compaction, le contexte du modèle commence par ce résumé. Les bornes sont des commentaires
HTML, invisibles en markdown mais lisibles par le code à la compaction suivante. Le runner n’en
voit que la ligne compaction de la trace du chapitre A4.
<!-- [FACTORY] -->
## Contrat de l'usine (recopié mot pour mot — ne pas paraphraser)
- phase : build · rôle : builder · run : 9f2e4a10 · porte : BuildEnvelope (outil report_phase)
- règle : l'agent propose, le code dispose — termine comme le contrat ci-dessous le demande.
### Mission (mot pour mot)
<!-- [FACTORY:mission] -->
Tu es le builder de l'usine. …
### mission
Ajoute un compteur de mots dans la barre d'etat de Plume. …
Quand tu as fini, appelle l'outil report_phase — seul, en dernier, une seule fois — avec exactement ces champs : …
<!-- [/FACTORY:mission] -->
### Dernier mot du runner (mot pour mot)
<!-- [FACTORY:verdict] -->
Ta derniere reponse n'a pas passe la validation : 2 tests rouges dans apps/plume. …
<!-- [/FACTORY:verdict] -->
<!-- [/FACTORY] -->
## Progress
### Done
- [x] lu specs/word-counter.md et apps/plume/src/editor.ts
…Script — le cœur de l’extension
Sans FACTORY_PHASE, la session est un poste et pi résume comme il l’entend. Avec elle, le code
construit le bloc, le modèle rédige le récit, et l’ensemble revient comme compaction fournie.
// .pi/extensions/factory-compact.ts — extrait : le bloc par le code, le récit par le modèle.
pi.on("session_before_compact", async (event, ctx) => {
const facts = readFacts(process.env);
if (!facts) return; // poste : la compaction par défaut de pi
const { preparation, reason, signal } = event;
const messages = [...preparation.messagesToSummarize, ...preparation.turnPrefixMessages];
const previous = splitBlock(preparation.previousSummary); // le bloc d'avant, s'il existe
const anchors = pickAnchors(messages, parseAnchors(previous.block));
const block = buildBlock(facts, anchors); // déterministe, zéro jeton
// … le récit : un appel au modèle de la session sur les messages sérialisés,
// remplacé par dryNarrative() si l'appel échoue — le bloc, lui, ne dépend de rien …
return { compaction: { summary: mergeSummary(block, narrative, formatFiles(files)),
firstKeptEntryId: preparation.firstKeptEntryId,
tokensBefore: preparation.tokensBefore, usage, details } };
});Côté Claude Code : vous orientez le résumé sans pouvoir le fournir. Vous passez des
instructions à /compact, ou les écrivez sous un titre # Compact instructions dans le CLAUDE.md
racine. Le hook PreCompact peut bloquer une compaction et PostCompact reçoit le résumé, mais
aucun ne le remplace. Une consigne qui doit survivre va dans le CLAUDE.md racine, relu sur disque
après chaque compaction.
Piège courant : « mon payload est petit, la compaction ne me concerne pas ». Le seuil dépend de ce que rapportent les outils, pas de la taille du dépôt. Une sortie verbeuse de
bun testsuffit à le franchir en quelques tours. Et sans compaction, le contexte déborde et la phase meurt après avoir payé ses jetons.
Fallback, doctor et package en deux couches
Trois gestes du code déterministe, côté runner : vérifier avant le run, changer de moteur pendant une panne, installer l’usine ailleurs après.
Vérifier avant de tourner
Lundi matin, just doctor affiche en rouge binaire pi 0.84.4 < 0.85.0. Le pi update a été
fait sur le portable, pas sur ce poste. Sans le doctor, le premier scout aurait payé son tour
avant d’être refusé par le port.
La version minimale, PI_MIN_VERSION, vit dans harness.py, à côté des noms d’événements que le
port lit. Elle est vérifiée deux fois. Pendant le run, l’inventaire du chapitre A8 rapporte la
version qui a tourné, et _judge refuse la phase en dessous, comme pour un socle manquant. Avant
le run, le doctor compare à sec le binaire pi et le SDK de node_modules, et signale s’ils
divergent. La correction est pi update.
Le doctor vérifie aussi les modèles. Un --model inconnu du fournisseur ne fait pas échouer pi
0.85 : il crée un modèle synthétique, cloné du défaut du fournisseur avec son prix, et
n’émet qu’un avertissement. Le port contrôle le modèle observé après coup (chapitre A6). Le
doctor exige avant le run la route exacte de chaque agent du roster, et du secours, dans
pi --list-models. pi update --models rafraîchit ce catalogue.
Côté Claude Code : le réglage requiredMinimumVersion, accepté seulement dans les managed
settings, empêche le démarrage d’une version plus ancienne.
Le secours, sur panne fournisseur seulement
Au quatorzième tour d’un build, une trentaine de centimes déjà dépensés, le fournisseur rend
503 overloaded. pi réessaie deux fois, puis renonce avec stopReason: error. Sans secours, la
reprise relance le même moteur, qui rend souvent le même 503, et le run finit rouge un quart
d’heure plus tard. Avec un secours déclaré dans .env, le port relance une fois sur un autre
moteur, dans la même session. Celui-ci relit les quatorze tours pour un à deux centimes, reprend
le travail et rend son enveloppe.
Le port ne relaie que sur une marque de panne dans l’errorMessage : 429, 5xx, surcharge,
« no endpoints ». Il relance alors une fois sur FACTORY_FALLBACK_MODEL, avec le motif. Il
ne relaie jamais sur aborted (le coupe-circuit), une erreur d’outil, le mur de temps ou une clé
invalide, qu’un humain doit voir. pi ne relaie pas lui-même, car il ne connaît ni le roster ni le
budget.
Une panne touche un fournisseur, pas un rôle. Le secours vit donc dans .env, à côté de la clé,
et vaut pour toute l’usine. Une requête peut le surcharger par fallback_model, et un secours
identique au moteur demandé est ignoré. Après un relais, le registre des sessions (chapitre A9)
a une seconde ligne et HarnessResult.model nomme le moteur qui a fini.
Côté Claude Code : --fallback-model, ou le réglage fallbackModel, déclare jusqu’à trois
modèles de secours. Claude Code bascule quand le modèle principal est surchargé ou indisponible,
jamais sur une erreur d’authentification ou de limite de débit, et pour le tour en cours
seulement. L’adaptateur _run_claude de l’usine ne passe pas ce drapeau.
Un arrêt, un geste
| Ce que le port observe | Cause probable | Geste du code |
|---|---|---|
stopReason: aborted | le coupe-circuit du chapitre A7 | HarnessError avec session ; relance motivée, même moteur |
stopReason: error, message 429 / 5xx / overloaded / no endpoints | panne du fournisseur | relais sur le secours, même session, une fois |
stopReason: error, message « invalid api key », 401 / 403 | configuration | HarnessError ; un humain corrige .env |
stopReason: length ou texte sans enveloppe | le modèle n’a pas fini | correction(), même session, même moteur |
| harnais muet après le mur de temps | phase trop longue, outil bloqué | HarnessError avec session ; jamais de secours |
inventaire avec pi_version < épinglée | poste pas à jour | refus fail-closed ; pi update |
Piège courant : « pi réessaie déjà, le secours est redondant ». Le retry de pi relance le même moteur chez le même fournisseur. Il couvre une erreur passagère, pas une surcharge qui dure une heure.
Le doctor et l’ordonnance en deux couches
just doctor (chapitre 5) prend quelques secondes et zéro jeton. Il contrôle l’outillage, les
pièces, puis le nœud pi : versions, socle sur disque, réglages, modèles du roster, batterie du
chapitre A8. Le premier rouge rend un code non nul.
Le doctor rédige aussi l’ordonnance, l’entrée qui installe l’usine pi dans un autre dépôt.
pi filtre ce qu’il charge d’un package par une entrée objet dans packages. La couche socle
prend les six extensions du nœud et zéro skill. La couche poste prend tout. La source est
relative au dossier .pi/ du dépôt cible. C’est l’erratum du chapitre A5 : pour deux dépôts
voisins, il faut ../../plume-factory, pas ../plume-factory.
Côté Claude Code : claude doctor affiche un diagnostic de l’installation et des réglages,
en lecture seule, sans ouvrir de session. En session, /doctor propose aussi des corrections que
Claude applique après confirmation.
Commande — le doctor, puis l’ordonnance
# le diagnostic complet — zéro jeton, quelques secondes (la batterie bun test comprise)
uv run adws/doctor.py
# sans la batterie — quand vous avez déjà lancé bun test
uv run adws/doctor.py --quick
# l'ordonnance : le socle seul (six extensions, zéro skill) dans un dépôt voisin
uv run adws/doctor.py --layer socle --into ../pi-essai
# ou tout le poste — footer, chaînes et skills compris
uv run adws/doctor.py --layer poste --into ../pi-essaiConfig — ce que l’ordonnance écrit, et la ligne de .env
L’entrée écrite dans ../pi-essai/.pi/settings.json après --layer socle, puis la ligne
facultative de .env qui arme le secours.
{
"packages": [
{
"source": "../../plume-factory",
"extensions": [
".pi/extensions/damage-control.ts",
".pi/extensions/factory-obs.ts",
".pi/extensions/factory-report.ts",
".pi/extensions/factory-guard.ts",
".pi/extensions/factory-inventory.ts",
".pi/extensions/factory-compact.ts"
],
"skills": [],
"prompts": [],
"themes": []
}
]
}# .env — le secours de l'usine : une decision d'infrastructure, a cote de la cle
OPENROUTER_API_KEY=sk-or-...
FACTORY_FALLBACK_MODEL=deepseek/deepseek-v4-flash-0731pi et Claude Code côte à côte
| pi (usine) | Claude Code | |
|---|---|---|
| Résumé de compaction | fourni par une extension | généré par Claude Code, orientable par /compact |
| Ce qui survit à coup sûr | le bloc [FACTORY] | le CLAUDE.md racine, relu sur disque |
| Modèle de secours | relais du port, 429 compris | --fallback-model, jamais sur une limite de débit |
| Version minimale | PI_MIN_VERSION, phase refusée | requiredMinimumVersion, démarrage refusé |
| Diagnostic | doctor.py, zéro jeton | claude doctor et /doctor |
Fil rouge — la pièce posée aujourd’hui
Le socle gagne factory-compact.ts. Le port (adaptateur v7) refuse un pi trop vieux et
relaie une panne fournisseur. Le poste de pilotage retrouve doctor.py, qui diagnostique le
nœud pi et l’installe ailleurs. L’agent rédige le récit et reprend la mission sur le secours. Le
code dispose du reste. Le runner exige toujours les cinq extensions du chapitre A8, car
l’absence de la compaction n’ouvre aucune porte. Le doctor et la couche socle en attendent six.
Travaux pratiques — la pièce du jour
Trois fichiers à poser dans plume-factory. Prérequis : pi ≥ 0.85, Bun, les deux paquets de
développement du chapitre A8 (bun add -d @earendil-works/pi-coding-agent @earendil-works/pi-ai) et le roster du chapitre 27.
Pièce — .pi/extensions/factory-compact.ts
La sixième extension du socle. Six fonctions pures exportées (readFacts, pickAnchors,
buildBlock, splitBlock, parseAnchors, mergeFiles), un récit à sec (dryNarrative), un
seul événement écouté et sa gate sous Bun. Elle ne modifie aucune pièce posée.
// .pi/extensions/factory-compact.ts — la compaction structurée : ce que le runner a dit survit au résumé.
// Quand le contexte déborde, pi résume les anciens messages et ne garde que les récents. Dans un
// nœud d'usine, les anciens messages contiennent ce que le RUNNER a injecté : la mission, le
// handoff, le contrat de la porte typée, sa dernière correction. Un résumé libre les paraphrase —
// ou les perd. Cette extension remplace le résumé par un résumé EN DEUX PARTIES : un bloc
// [FACTORY] déterministe, recopié mot pour mot (jamais résumé, jamais réécrit), puis un récit
// produit par le modèle de la session. Sur un poste (pas de FACTORY_PHASE), elle ne fait rien :
// la compaction par défaut de pi s'applique. Aucune dépendance npm à l'exécution.
import { randomUUID } from "node:crypto";
import { appendFileSync, mkdirSync } from "node:fs";
import { join } from "node:path";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { convertToLlm, serializeConversation } from "@earendil-works/pi-coding-agent";
export const ENTRY_TYPE = "factory-compact"; // l'entrée déposée dans la session
export const BLOCK_OPEN = "<!-- [FACTORY] -->"; // les bornes du bloc, invisibles en markdown
export const BLOCK_CLOSE = "<!-- [/FACTORY] -->";
export const MISSION_CAP = 6000; // caractères gardés mot pour mot
export const VERDICT_CAP = 2000;
const HARNESS_DIR = join("adws", "adw_data", "traces", "harness"); // le journal du chapitre A4
// ---------- les faits de l'usine : ce que le runner a posé dans l'environnement (A7) ----------
export interface FactoryFacts {
phase: string;
role: string;
adwId: string;
schema: string | null; // le nom du schéma de la porte typée (A6), sans son chemin
}
export function readFacts(env: Record<string, string | undefined>): FactoryFacts | null {
const phase = env.FACTORY_PHASE?.trim();
if (!phase) return null; // pas de phase = un poste : la compaction par défaut suffit
const schemaPath = env.FACTORY_ENVELOPE_SCHEMA?.trim();
const schema = schemaPath ? schemaPath.split(/[\\/]/).pop()!.replace(/\.json$/, "") : null;
return { phase, role: env.FACTORY_ROLE?.trim() || "?", adwId: env.FACTORY_ADW_ID?.trim() || "?", schema };
}
// ---------- les ancres : les mots du runner, dans les messages qui vont disparaître ----------
export interface Anchors {
mission: string | null; // le premier message utilisateur de la session : brief, mission, contrat
verdict: string | null; // le dernier message utilisateur résumé : la correction ou la relance en cours
}
interface LooseMessage { role?: string; content?: unknown }
export function textOf(message: LooseMessage): string {
const content = message.content;
if (typeof content === "string") return content;
if (!Array.isArray(content)) return "";
return content
.filter((part): part is { type: "text"; text: string } => !!part && typeof part === "object" && (part as { type?: string }).type === "text")
.map((part) => part.text)
.join("");
}
export function cap(text: string, limit: number): string {
if (text.length <= limit) return text;
return `${text.slice(0, limit)}\n… [tronqué par factory-compact : ${text.length} caractères au total]`;
}
// La mission est le premier message utilisateur jamais vu (le bloc précédent la garde) ; le
// verdict est le dernier message utilisateur qui quitte le contexte — à défaut, le précédent.
export function pickAnchors(messages: LooseMessage[], previous: Anchors | null): Anchors {
const users = messages.filter((m) => m.role === "user").map(textOf).filter((t) => t.trim().length > 0);
const mission = previous?.mission ?? users[0] ?? null;
const later = users.filter((t) => t !== mission);
const verdict = later.length > 0 ? later[later.length - 1] : (previous?.verdict ?? null);
return { mission, verdict };
}
// ---------- le bloc [FACTORY] : construit, découpé, relu — trois fonctions pures ----------
export function buildBlock(facts: FactoryFacts, anchors: Anchors): string {
const lines = [
BLOCK_OPEN,
"## Contrat de l'usine (recopié mot pour mot — ne pas paraphraser)",
`- phase : ${facts.phase} · rôle : ${facts.role} · run : ${facts.adwId}` +
(facts.schema ? ` · porte : ${facts.schema} (outil report_phase)` : ""),
"- règle : l'agent propose, le code dispose — termine comme le contrat ci-dessous le demande.",
"",
"### Mission (mot pour mot)",
...fenced("mission", anchors.mission ? cap(anchors.mission, MISSION_CAP) : null),
"",
"### Dernier mot du runner (mot pour mot)",
...fenced("verdict", anchors.verdict ? cap(anchors.verdict, VERDICT_CAP) : null),
BLOCK_CLOSE,
];
return lines.join("\n");
}
// Chaque ancre a ses propres bornes : les asks de l'usine contiennent eux-mêmes des titres
// markdown (« ### mission », « ### previous_envelope ») — un découpage par titres les casserait.
function fenced(name: string, text: string | null): string[] {
return [`<!-- [FACTORY:${name}] -->`, text ?? `(aucune ${name === "mission" ? "mission" : "correction"} dans la partie résumée)`, `<!-- [/FACTORY:${name}] -->`];
}
export function splitBlock(summary: string | undefined): { block: string | null; rest: string } {
if (!summary) return { block: null, rest: "" };
const start = summary.indexOf(BLOCK_OPEN);
const end = summary.indexOf(BLOCK_CLOSE);
if (start === -1 || end === -1 || end < start) return { block: null, rest: summary };
const block = summary.slice(start, end + BLOCK_CLOSE.length);
const rest = (summary.slice(0, start) + summary.slice(end + BLOCK_CLOSE.length)).trim();
return { block, rest };
}
function section(block: string, name: string): string | null {
const open = `<!-- [FACTORY:${name}] -->`;
const close = `<!-- [/FACTORY:${name}] -->`;
const from = block.indexOf(open);
const to = block.indexOf(close);
if (from === -1 || to === -1 || to < from) return null;
const text = block.slice(from + open.length, to).trim();
return text.startsWith("(aucune ") ? null : text;
}
export function parseAnchors(block: string | null): Anchors | null {
if (!block) return null;
return { mission: section(block, "mission"), verdict: section(block, "verdict") };
}
// ---------- les fichiers : cumulés d'une compaction à l'autre, comme pi le fait pour les siennes ----------
export interface FileLists { read: string[]; modified: string[] }
export function parseFileTags(text: string): FileLists {
const grab = (tag: string): string[] => {
const match = text.match(new RegExp(`<${tag}>\\n?([\\s\\S]*?)</${tag}>`));
return match ? match[1].split("\n").map((l) => l.trim()).filter((l) => l.length > 0) : [];
};
return { read: grab("read-files"), modified: grab("modified-files") };
}
export function mergeFiles(previous: FileLists, ops: { read: Iterable<string>; written: Iterable<string>; edited: Iterable<string> }): FileLists {
const modified = new Set([...previous.modified, ...ops.written, ...ops.edited]);
const read = new Set([...previous.read, ...ops.read].filter((f) => !modified.has(f)));
return { read: [...read].sort(), modified: [...modified].sort() };
}
export function formatFiles(files: FileLists): string {
const parts: string[] = [];
if (files.read.length) parts.push(`<read-files>\n${files.read.join("\n")}\n</read-files>`);
if (files.modified.length) parts.push(`<modified-files>\n${files.modified.join("\n")}\n</modified-files>`);
return parts.join("\n\n");
}
export function mergeSummary(block: string, narrative: string, files: string): string {
return [block, narrative.trim(), files].filter((p) => p.length > 0).join("\n\n");
}
// Le récit « à sec » : quand le modèle ne peut pas résumer (panne, abandon), un inventaire
// déterministe vaut mieux qu'une compaction qui efface le contrat. Zéro jeton.
export function dryNarrative(messages: LooseMessage[], reason: string): string {
const turns = messages.filter((m) => m.role === "assistant").length;
const tools = messages
.filter((m) => m.role === "assistant" && Array.isArray(m.content))
.flatMap((m) => (m.content as Array<{ type?: string; name?: string }>).filter((p) => p?.type === "toolCall").map((p) => p.name ?? "?"));
const counts = new Map<string, number>();
for (const name of tools) counts.set(name, (counts.get(name) ?? 0) + 1);
const used = [...counts.entries()].map(([name, n]) => `${name}×${n}`).join(", ") || "(aucun)";
return [
"## Récit indisponible (compaction à sec)",
`- motif : ${reason}`,
`- ${turns} réponses de l'assistant et ${tools.length} appels d'outils ont quitté le contexte (${used}).`,
"- Les messages récents sont conservés tels quels ; relis-les avant de poursuivre.",
].join("\n");
}
const SYSTEM = "You are a context summarization assistant. Read the conversation and produce ONLY the structured summary in the exact format requested. Do NOT continue the conversation.";
function prompt(conversation: string, previousNarrative: string): string {
const previous = previousNarrative
? `\n\n<previous-summary>\n${previousNarrative}\n</previous-summary>\n\nThe messages are NEW messages to fold into the previous summary: keep what is still true, update what changed.`
: "";
return `<conversation>\n${conversation}\n</conversation>${previous}
Create a structured context checkpoint that another LLM will use to continue the work. The task
statement and the runner's instructions are preserved verbatim elsewhere: do NOT restate them.
Use this EXACT format:
## Progress
### Done
- [x] [completed steps, with file paths]
### In Progress
- [ ] [current work]
### Blocked
- [issues, or "(none)"]
## Key Decisions
- **[decision]**: [brief rationale]
## Next Steps
1. [ordered, concrete]
## Critical Context
- [data, commands, test names or error messages needed to continue — or "(none)"]`;
}
// ---------- le journal (A4) : une ligne par compaction, jointe à la phase par la session ----------
function journalLine(cwd: string, sessionId: string, name: string, payload: Record<string, unknown>): void {
const dir = join(cwd, HARNESS_DIR);
mkdirSync(dir, { recursive: true });
const line = { event_id: `hev_${randomUUID().replace(/-/g, "").slice(0, 12)}`, ts: new Date().toISOString(), session_id: sessionId, type: "compact", name, payload };
appendFileSync(join(dir, `${sessionId}.jsonl`), `${JSON.stringify(line)}\n`, "utf8");
}
// ---------- l'extension : un seul événement, une seule décision ----------
export default function (pi: ExtensionAPI) {
pi.on("session_before_compact", async (event, ctx) => {
const facts = readFacts(process.env);
if (!facts) return; // poste : la compaction par défaut de pi
const { preparation, reason, signal } = event;
// Un tour coupé en deux (split turn) se résume ici en un seul récit : le préfixe du tour
// rejoint l'historique, le bloc [FACTORY] porte de toute façon la demande initiale.
const messages = [...preparation.messagesToSummarize, ...preparation.turnPrefixMessages] as LooseMessage[];
const previous = splitBlock(preparation.previousSummary);
const anchors = pickAnchors(messages, parseAnchors(previous.block));
const block = buildBlock(facts, anchors);
const files = mergeFiles(parseFileTags(previous.rest), preparation.fileOps);
const sessionId = ctx.sessionManager.getSessionId();
let narrative = "";
let usage: { input: number; output: number } | undefined;
let mode: "modele" | "sec" = "modele";
try {
if (!ctx.model) throw new Error("aucun modèle dans la session");
const conversation = serializeConversation(convertToLlm(messages as never));
const response = await ctx.modelRegistry.complete(
ctx.model,
{ systemPrompt: SYSTEM, messages: [{ role: "user", content: [{ type: "text", text: prompt(conversation, previous.rest) }], timestamp: Date.now() }] },
{ maxTokens: Math.max(1024, Math.min(4096, Math.floor(0.8 * preparation.settings.reserveTokens))), signal, cacheRetention: "none" },
);
if (response.stopReason === "error" || response.stopReason === "aborted") throw new Error(response.errorMessage ?? response.stopReason);
narrative = response.content.filter((c): c is { type: "text"; text: string } => c.type === "text").map((c) => c.text).join("\n").trim();
usage = response.usage as { input: number; output: number };
if (!narrative) throw new Error("résumé vide");
} catch (error) {
if (signal.aborted) return; // annulée par pi : sa propre voie s'applique
mode = "sec";
narrative = dryNarrative(messages, error instanceof Error ? error.message : String(error));
}
const summary = mergeSummary(block, narrative, formatFiles(files));
const payload = { reason, mode, tokens_before: preparation.tokensBefore, block_chars: block.length, mission_kept: anchors.mission !== null, verdict_kept: anchors.verdict !== null };
journalLine(ctx.cwd, sessionId, reason, payload);
pi.appendEntry(ENTRY_TYPE, { ...payload, phase: facts.phase, adw_id: facts.adwId });
if (ctx.hasUI) ctx.ui.notify(`factory-compact : bloc [FACTORY] conservé (${mode})`, "info");
return {
compaction: {
summary,
firstKeptEntryId: preparation.firstKeptEntryId,
tokensBefore: preparation.tokensBefore,
usage: usage as never,
details: { factory: facts, mode, readFiles: files.read, modifiedFiles: files.modified },
},
};
});
}
// ---------- la gate du fichier : `bun .pi/extensions/factory-compact.ts`, zéro jeton ----------
if (import.meta.main) {
const env = { FACTORY_PHASE: "build", FACTORY_ROLE: "builder", FACTORY_ADW_ID: "9f2e4a10", FACTORY_ENVELOPE_SCHEMA: "/usine/adws/adw_data/schemas/BuildEnvelope.json" };
const facts = readFacts(env)!;
const mission = "Tu es le builder de l'usine.\n### mission\nAjoute un compteur de mots.\nQuand tu as fini, appelle l'outil report_phase — seul, en dernier.";
const verdict = "Ta derniere reponse n'a pas passe la validation : 2 tests rouges.\nRends a nouveau UNIQUEMENT l'enveloppe demandee.";
const messages: LooseMessage[] = [
{ role: "user", content: mission },
{ role: "assistant", content: [{ type: "text", text: "je lis" }, { type: "toolCall", name: "read" }, { type: "toolCall", name: "bash" }] },
{ role: "toolResult", content: [{ type: "text", text: "ok" }] },
{ role: "user", content: [{ type: "text", text: verdict }] },
{ role: "assistant", content: [{ type: "toolCall", name: "bash" }] },
];
// 1. Première compaction : la mission et le verdict sont recopiés mot pour mot.
const first = buildBlock(facts, pickAnchors(messages, null));
const summary1 = mergeSummary(first, "## Progress\n- [x] lu", formatFiles(mergeFiles({ read: [], modified: [] }, { read: ["apps/plume/src/app.ts"], written: [], edited: ["apps/plume/src/editor.ts"] })));
const ok1 = summary1.includes(mission) && summary1.includes(verdict) && summary1.includes("porte : BuildEnvelope")
&& summary1.indexOf(BLOCK_OPEN) === 0 && summary1.includes("<modified-files>\napps/plume/src/editor.ts");
// 2. Seconde compaction : plus de mission dans les messages, elle vient du bloc précédent ; le verdict se met à jour.
const later: LooseMessage[] = [{ role: "assistant", content: [{ type: "text", text: "je corrige" }] }, { role: "user", content: "Nouveau verdict : 1 test rouge." }];
const split = splitBlock(summary1);
const second = buildBlock(facts, pickAnchors(later, parseAnchors(split.block)));
const ok2 = second.includes(mission) && second.includes("Nouveau verdict : 1 test rouge.") && !second.includes(verdict)
&& split.rest.startsWith("## Progress") && !split.rest.includes(BLOCK_OPEN);
// 3. Idempotence et cumul : les fichiers de la première compaction survivent à la seconde.
const files2 = mergeFiles(parseFileTags(split.rest), { read: ["apps/plume/src/editor.ts"], written: [], edited: [] });
const ok3 = files2.modified.includes("apps/plume/src/editor.ts") && files2.read.includes("apps/plume/src/app.ts") && !files2.read.includes("apps/plume/src/editor.ts");
// 4. Troncature et récit à sec : un contrat trop long est coupé en le disant ; sans modèle, le résumé reste utile.
const long = buildBlock(facts, { mission: "x".repeat(MISSION_CAP + 500), verdict: null });
const dry = dryNarrative(messages, "429 Too Many Requests");
const ok4 = long.includes("[tronqué par factory-compact : 6500 caractères") && long.includes("(aucune correction")
&& dry.includes("2 réponses de l'assistant et 3 appels d'outils") && dry.includes("bash×2") && readFacts({}) === null;
const ok = ok1 && ok2 && ok3 && ok4;
console.log(`factory-compact ${ok ? "OK" : "KO"} — bloc [FACTORY] en tête (${first.length} caractères), mission et verdict mot pour mot, ` +
`mission reprise du bloc précédent, verdict mis à jour, fichiers cumulés, troncature annoncée, récit à sec` +
(ok ? "" : ` [${[ok1, ok2, ok3, ok4].map((v, i) => (v ? "" : `étape ${i + 1}`)).filter(Boolean).join(", ")}]`));
process.exit(ok ? 0 : 1);
}Pièce — adws/adw_modules/harness.py
Cette version remplace celle du chapitre A9. Tout ce qui existait reste, y compris le profil
d’authentification du chapitre 17bis. S’ajoutent la version épinglée (PI_MIN_VERSION,
version_too_old, le refus dans _judge), la lecture de errorMessage, la détection de panne
(provider_failure), le choix du secours (fallback_for), le relais dans _run_pi et le champ
model de HarnessResult.
"""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.
Version annexe A10 (v7) : le runner VERIFIE la version de pi et sait
changer de moteur sans changer de session. L'inventaire du socle (A8)
porte la version de pi : en dessous de PI_MIN_VERSION, la phase est
refusee — fail-closed, comme un socle manquant. Une panne FOURNISSEUR
(429, 5xx, surcharge, modele sans point de terminaison) est distinguee de
tout autre arret : elle seule declenche UN relancement sur le modele de
secours (FACTORY_FALLBACK_MODEL, ou fallback_model de la requete), dans
la MEME session, avec le motif. Jamais sur une erreur d'outil, jamais sur
le mur de temps, jamais cote pi : le code dispose, y compris du moteur.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
import uuid
from dataclasses import dataclass, replace
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"
# La version de pi en dessous de laquelle le runner REFUSE de tourner (A10).
# Les noms d'evenements et les drapeaux que ce module lit ont ete verifies
# sur cette version ; l'inventaire du socle (A8) rapporte la version reelle
# du binaire qui a tourne, et le verdict la compare a celle-ci. Le doctor
# (adws/doctor.py) fait la meme comparaison A SEC, avant tout run.
PI_MIN_VERSION = "0.85.0"
# Le modele de secours de l'usine (A10) : une decision d'infrastructure,
# pas de role — il vit dans .env a cote de la cle, jamais dans le roster.
# Un id du REGISTRE (fournisseur/id), verifie par le doctor. Vide = pas de
# secours : une panne fournisseur reste une HarnessError ordinaire.
FALLBACK_ENV = "FACTORY_FALLBACK_MODEL"
# Ce qui, dans le message d'erreur d'un fournisseur, signe une PANNE de
# fournisseur — la seule cause qui justifie un changement de moteur. Une
# erreur d'outil, un plafond du coupe-circuit (aborted) ou le mur de temps
# ne sont jamais dans cette liste : changer de modele n'y changerait rien.
PROVIDER_FAILURE_MARKERS = ("429", "rate limit", "rate_limit", "too many requests",
"overloaded", "capacity", "500", "502", "503", "504",
"unavailable", "no endpoints", "internal server error")
# La relance sur le modele de secours : le motif, puis la consigne de
# poursuivre — le contrat (et l'enveloppe attendue) sont deja dans la session.
FALLBACK_ASK = """Le fournisseur du modele precedent est en panne : {reason}.
Tu reprends la MEME mission, dans cette meme session, sur un autre modele.
Relis les derniers messages, poursuis ou tu en etais, et termine comme le
contrat le demande."""
# 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
fallback_model: str | None = None # secours sur panne fournisseur — None = FACTORY_FALLBACK_MODEL
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
model: str | None = None # le modele qui a repondu — le secours s'il a pris le relais
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)
error_message: str | None = None # le motif du fournisseur quand stopReason == "error" (A10)
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 message.get("errorMessage"):
reading.error_message = str(message["errorMessage"])
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 version_tuple(version: str) -> tuple[int, ...]:
"""'0.85.1' -> (0, 85, 1) ; les suffixes (-beta, +build) sont ignores. Pure."""
core = version.strip().split("-", 1)[0].split("+", 1)[0]
parts = []
for piece in core.split("."):
digits = "".join(ch for ch in piece if ch.isdigit())
parts.append(int(digits) if digits else 0)
return tuple(parts) or (0,)
def version_too_old(version: str | None, minimum: str = PI_MIN_VERSION) -> bool:
"""Vrai si la version rapportee est en dessous du minimum epingle. Pure.
Une version absente n'est PAS jugee trop vieille : un inventaire d'avant
A8 n'en porte pas, et c'est l'absence d'inventaire qui refuse, pas ceci.
"""
if not version:
return False
return version_tuple(version) < version_tuple(minimum)
def provider_failure(reading: PiReading) -> str | None:
"""Le motif d'une panne FOURNISSEUR, ou None — pure, donc testable a sec.
Seul un arret en stopReason "error" dont le message porte une marque de
panne cote fournisseur compte. "aborted" (coupe-circuit), "length",
une erreur d'outil ou un texte sans enveloppe ne sont jamais des pannes
fournisseur : ils reviennent a l'agent par correction, pas par secours.
"""
if reading.stop_reason != "error":
return None
message = (reading.error_message or "").lower()
if any(marker in message for marker in PROVIDER_FAILURE_MARKERS):
return reading.error_message
return None
def fallback_for(request: HarnessRequest, env: dict[str, str] | None = None) -> str | None:
"""Le modele de secours applicable a cette requete, ou None. Pure.
La requete d'abord, l'environnement (.env) ensuite ; jamais le meme
modele que celui qui vient de tomber — ce ne serait pas un secours.
"""
source = os.environ if env is None else env
candidate = request.fallback_model or source.get(FALLBACK_ENV, "").strip() or None
if candidate and request.model and _route(candidate) == _route(request.model, request.direct):
return None
return candidate
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)
# La version epinglee (A10) : un pi plus vieux que celui sur lequel
# ce module a ete verifie ne tourne pas — fail-closed, comme un
# socle manquant. `pi update` est la correction, pas un contournement.
if version_too_old(reading.inventory.get("pi_version")):
raise HarnessError(f"pi {reading.inventory.get('pi_version')} est en dessous de la "
f"version epinglee {PI_MIN_VERSION} — lancez `pi update` "
"(et `uv run adws/doctor.py` pour verifier a sec)", 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,
model=reading.observed_model)
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())
# Le secours (A10) : UNE relance, sur panne FOURNISSEUR seulement, dans
# la meme session, avec le motif — et sur un autre moteur. Tout autre
# arret suit le chemin ordinaire du verdict : correction ou HarnessError.
reason = provider_failure(reading)
fallback = fallback_for(request)
if reason and fallback:
relay = replace(request, model=fallback, session_id=session_id, fallback_model=None)
cost_so_far, tokens_so_far = reading.cost_usd, reading.tokens
# Le registre garde la trace du relais : la derniere ligne dit quel
# moteur a fini la phase — la premiere, lequel est tombe.
declare_session(session_id, request.cwd, fallback)
print(f"harness : panne fournisseur sur {request.model} ({str(reason)[:120]}) — "
f"relais sur {fallback}, meme session {session_id}", file=sys.stderr)
try:
proc = _spawn(_pi_argv(relay, session_id), relay, extra_env=extra_env,
stdin_text=FALLBACK_ASK.format(reason=str(reason)[:300]))
except HarnessError as error:
error.session_id = session_id
raise
reading = _read_pi_stream(proc.stdout.splitlines())
reading.cost_usd += cost_so_far # la panne a pu couter quelques jetons : on les compte
reading.tokens += tokens_so_far
return _judge(reading, relay, session_id, proc.returncode, proc.stderr)
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)
# 7. La version epinglee (A10) : un inventaire trop vieux est refuse, meme socle complet.
assert version_tuple("0.85.1") > version_tuple("0.85.0") > version_tuple("0.9.12")
assert version_too_old("0.84.4") and not version_too_old("0.85.0") and not version_too_old(None)
old = json.dumps({"type": "entry_appended", "entry": {"type": "custom", "customType": INVENTORY_ENTRY,
"data": {"pi_version": "0.84.4", "present": list(SOCLE), "missing": []}}})
try:
_judge(_read_pi_stream([old, assistant, envelope]), request, "sess-8", 0, "")
raise AssertionError("un pi plus vieux que la version epinglee doit etre refuse")
except HarnessError as error:
assert "en dessous de la version epinglee" in str(error) and error.session_id == "sess-8"
# 8. La panne fournisseur (A10) : detectee sur error + motif, jamais sur aborted ni sur un outil.
def stopped(reason: str, message: str | None) -> PiReading:
return _read_pi_stream([json.dumps({"type": "message_end", "message": {
"role": "assistant", "content": [], "stopReason": reason, "usage": {},
"errorMessage": message}})])
assert provider_failure(stopped("error", "429 Too Many Requests")) == "429 Too Many Requests"
assert provider_failure(stopped("error", "Provider returned error: 503 overloaded"))
assert provider_failure(stopped("error", "No endpoints found for z-ai/glm-5.3"))
assert provider_failure(stopped("error", "invalid api key")) is None # pas une panne : une config
assert provider_failure(stopped("aborted", "429")) is None # le coupe-circuit, pas le fournisseur
assert provider_failure(stopped("stop", None)) is None
# Le secours : la requete d'abord, .env ensuite, jamais le meme moteur.
assert fallback_for(request, {}) is None
assert fallback_for(request, {FALLBACK_ENV: "deepseek/deepseek-v4-flash-0731"}) == "deepseek/deepseek-v4-flash-0731"
assert fallback_for(request, {FALLBACK_ENV: "z-ai/glm-5.3"}) is None
assert fallback_for(replace(request, fallback_model="deepseek/deepseek-v4-flash-0731"),
{FALLBACK_ENV: "autre/modele"}) == "deepseek/deepseek-v4-flash-0731"
relay = replace(request, model="deepseek/deepseek-v4-flash-0731", session_id="sess-9")
assert _pi_argv(relay, "sess-9")[_pi_argv(relay, "sess-9").index("--model") + 1] \
== "openrouter/deepseek/deepseek-v4-flash-0731"
assert result.model == "openrouter/z-ai/glm-5.3" # le verdict dit quel moteur a repondu
# 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 ; "
f"v7 : pi < {PI_MIN_VERSION} refuse, panne fournisseur distinguee, secours sans "
"changer de session"
" ; profil 17bis : route directe hors passerelle, coffre .env ferme sous profil, ouvert sans")Pièce — adws/doctor.py
Cette version remplace celle du chapitre 4. S’ajoute le tableau du nœud pi. Le doctor importe
le port et le roster pour partager leur version épinglée, leur socle et leur lecture de .env.
Ses fonctions pures (parse_model_table, layer_entry, apply_layer) portent la gate à sec.
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["rich>=13", "pyyaml"]
# ///
"""doctor — le préflight du poste de pilotage ET du nœud pi.
Vérifie, sans dépenser un token, que l'outillage, les pièces de l'usine et
le nœud pi sont en état de tourner. Déterministe de bout en bout : une gate
sur votre environnement, à lancer avant tout run — et avant tout `pi update`.
Version annexe A10 (v2) — cette version remplace celle du chapitre 4. Elle
garde le tableau des outils et des pièces, et ajoute le diagnostic du nœud pi :
version du binaire et du SDK épinglées (PI_MIN_VERSION, la même que celle du
port), socle présent sur disque, réglages et règles en place, modèles du
roster (et modèle de secours) réellement présents dans le registre de pi,
batterie du chapitre A8 verte. Le doctor sait aussi rédiger une ordonnance :
`--layer socle|poste --into <dépôt>` installe l'usine pi dans un autre dépôt,
en deux couches, par une entrée `packages` filtrée dans son .pi/settings.json.
uv run adws/doctor.py # le diagnostic complet (zéro jeton)
uv run adws/doctor.py --quick # sans la batterie bun test
uv run adws/doctor.py --layer socle --into ../pi-essai
uv run adws/doctor.py --layer poste --into ../pi-essai
uv run adws/doctor.py --selftest # la gate à sec des fonctions pures
"""
from __future__ import annotations
import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path, PurePosixPath
from rich.console import Console
from rich.table import Table
# Les modules de l'usine : le port (version épinglée, socle, .env chargé au
# passage — c'est ainsi que `pi --list-models` voit la passerelle) et le roster.
from adw_modules import harness, roster
ROOT = Path(__file__).resolve().parent.parent # la racine de plume-factory
# (nom, requis dès aujourd'hui, rôle dans l'usine) — les outils des chapitres
# 5 et 6 sont requis depuis qu'ils ont leur pièce ; herdr reste optionnel.
TOOLS = [
("uv", True, "scripts autonomes — ch. 4"),
("git", True, "le socle du dépôt — ch. 1"),
("bun", True, "sert Plume, ses tests, et la batterie du socle — ch. 2, A8"),
("just", True, "recettes de l'usine — ch. 5"),
("pi", False, "harnais n° 1 — ch. 7, annexe (requis si .pi/ existe)"),
("claude", False, "harnais n° 2 — ch. 7"),
("herdr", False, "multiplexeur d'agents — ch. 6"),
("sqlite3", False, "lire factory.db à la main — ch. 18 (facultatif)"),
]
# Les pièces dont l'absence signale un dépôt incomplet — une par zone de l'usine.
PIECES = ["PLAN.md", ".gitignore", "specs/plume-baseline.md", "apps/plume",
"adws/hello_factory.py", "justfile", "adws/adw_modules/harness.py",
"adws/adw_modules/runner.py", "adws/adw_config/factory.config.yaml"]
# Le socle du nœud pi tel que le doctor l'attend SUR DISQUE : les cinq
# extensions que le runner exige (harness.SOCLE) plus la compaction
# structurée (A10), que le runner n'exige pas mais que l'usine embarque.
SOCLE_FILES = (*harness.SOCLE, "factory-compact")
PI_DIR = Path(".pi")
PI_FILES = ["settings.json", "damage-control-rules.yaml"]
TESTS_DIR = PI_DIR / "tests"
SDK_PACKAGE = Path("node_modules/@earendil-works/pi-coding-agent/package.json")
# Les deux couches de l'usine pi (A10). `socle` : ce que le runner charge dans
# un nœud — extensions seulement, pas de skills (un nœud d'usine ne lit pas les
# skills du poste, --no-skills). `poste` : tout ce que le package déclare.
LAYERS = {
"socle": {"extensions": [f".pi/extensions/{name}.ts" for name in SOCLE_FILES],
"skills": [], "prompts": [], "themes": []},
"poste": None,
}
# ---------- les fonctions pures : versions, catalogue, ordonnance ----------
def version_of(tool: str) -> str | None:
"""Sonde un outil sans jamais planter : absent ou muet, c'est None."""
executable = shutil.which(tool)
if executable is None:
return None
try:
out = subprocess.run([executable, "--version"], capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=20)
first = (out.stdout or out.stderr).strip().splitlines()
return first[0][:40] if first else "présente"
except (OSError, subprocess.TimeoutExpired):
return None
def extract_version(text: str | None) -> str | None:
"""'pi 0.85.1' ou '0.85.1' -> '0.85.1' ; rien de numérique -> None. Pure."""
if not text:
return None
found = re.search(r"\d+\.\d+(?:\.\d+)?", text)
return found.group(0) if found else None
def sdk_version(package_json: Path = SDK_PACKAGE) -> str | None:
"""La version du SDK pi installée dans node_modules (chapitre A8) — None si absent."""
if not package_json.is_file():
return None
try:
return str(json.loads(package_json.read_text(encoding="utf-8")).get("version") or "") or None
except (OSError, json.JSONDecodeError):
return None
def parse_model_table(output: str) -> set[str]:
"""Le tableau de `pi --list-models` -> l'ensemble des 'fournisseur/id' disponibles. Pure.
Colonnes séparées par deux espaces au moins ; la première ligne est l'en-tête.
Un fournisseur n'y figure que si pi a une clé pour lui : une passerelle
absente de la liste, c'est une clé absente, pas un catalogue vide.
"""
models: set[str] = set()
for line in output.splitlines()[1:]:
columns = [c for c in line.strip().split(" ") if c.strip()]
if len(columns) >= 2:
models.add(f"{columns[0].strip()}/{columns[1].strip()}")
return models
def list_models() -> tuple[set[str], str]:
"""Interroge pi ; rend (modèles disponibles, erreur éventuelle). Zéro jeton."""
executable = shutil.which("pi")
if executable is None:
return set(), "pi introuvable"
try:
out = subprocess.run([executable, "--list-models"], capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=60)
except (OSError, subprocess.TimeoutExpired) as error:
return set(), f"pi --list-models : {error}"
return parse_model_table(out.stdout), ""
def wanted_models(factory: roster.Roster, env: dict[str, str] | None = None) -> dict[str, str]:
"""Les routes pi que l'usine demandera : par agent, plus le secours s'il est déclaré. Pure."""
source = os.environ if env is None else env
# La route suit le profil (17bis) : un forfait ou une API native ne passe pas par la passerelle.
routes = {spec.name: harness._route(spec.model, spec.auth.direct) for spec in factory.agents.values()}
fallback = source.get(harness.FALLBACK_ENV, "").strip()
if fallback:
routes["(secours)"] = harness._route(fallback)
return routes
def layer_entry(layer: str, source_root: Path, target: Path) -> str | dict:
"""L'entrée `packages` à écrire dans <target>/.pi/settings.json. Pure.
Un chemin local est résolu par pi RELATIVEMENT AU FICHIER DE RÉGLAGES —
donc au dossier .pi/ du dépôt cible, pas à sa racine. Pour deux dépôts
voisins, cela donne `../../plume-factory`. Toujours en barres obliques.
"""
if layer not in LAYERS:
raise ValueError(f"couche inconnue {layer!r} — connues : {sorted(LAYERS)}")
relative = os.path.relpath(source_root.resolve(), (target / PI_DIR).resolve())
source = str(PurePosixPath(*Path(relative).parts))
if LAYERS[layer] is None:
return source
return {"source": source, **LAYERS[layer]}
def apply_layer(settings: dict, entry: str | dict) -> dict:
"""Pose l'entrée dans `packages`, en remplaçant celle de la même source. Pure."""
source = entry if isinstance(entry, str) else entry["source"]
packages = [p for p in settings.get("packages") or []
if (p if isinstance(p, str) else p.get("source")) != source]
packages.append(entry)
return {**settings, "packages": packages}
# ---------- les vérifications : chacune ajoute ses problèmes, aucune ne lance ----------
def check_tools(table: Table, problems: list[str]) -> dict[str, str | None]:
versions: dict[str, str | None] = {}
pi_required = (PI_DIR / "settings.json").is_file()
for name, required, role in TOOLS:
versions[name] = version_of(name)
needed = required or (name == "pi" and pi_required)
if versions[name]:
status = versions[name]
elif needed:
status = "[red]MANQUANT[/red]"
problems.append(f"outil requis absent : {name}")
else:
status = "[yellow]absent[/yellow]"
table.add_row(name, status, role)
if not (versions["pi"] or versions["claude"]):
problems.append("aucun harnais disponible : installez pi ou claude")
return versions
def check_pieces(problems: list[str]) -> None:
for piece in PIECES:
if not Path(piece).exists():
problems.append(f"pièce manquante : {piece}")
def check_pi_node(table: Table, problems: list[str], pi_probe: str | None, quick: bool) -> None:
"""Le nœud pi : version épinglée, socle sur disque, réglages, modèles, batterie."""
pi_version = extract_version(pi_probe)
if pi_version is None:
table.add_row("binaire pi", "[yellow]absent[/yellow]", "section sautée")
return
# 1. La version du binaire, celle du SDK, et leur accord.
minimum = harness.PI_MIN_VERSION
if harness.version_too_old(pi_version, minimum):
table.add_row("binaire pi", f"[red]{pi_version} < {minimum}[/red]", "pi update")
problems.append(f"pi {pi_version} est en dessous de la version épinglée {minimum} "
"— le port refuserait chaque phase : lancez `pi update`")
else:
table.add_row("binaire pi", pi_version, f"≥ {minimum} épinglée dans harness.py")
sdk = sdk_version()
if sdk is None:
table.add_row("SDK pi (node_modules)", "[yellow]absent[/yellow]", "bun add -d … (ch. A8)")
elif harness.version_too_old(sdk, minimum):
table.add_row("SDK pi (node_modules)", f"[red]{sdk} < {minimum}[/red]", "bun add -d …@latest")
problems.append(f"SDK pi {sdk} en dessous de {minimum} : la batterie testerait un autre pi")
elif harness.version_tuple(sdk)[:2] != harness.version_tuple(pi_version)[:2]:
table.add_row("SDK pi (node_modules)", f"[yellow]{sdk}[/yellow]", "diverge du binaire — réaligner")
else:
table.add_row("SDK pi (node_modules)", sdk, "aligné sur le binaire")
# 2. Le socle sur disque, les réglages, les règles.
for name in PI_FILES:
if not (PI_DIR / name).is_file():
problems.append(f"nœud pi : .pi/{name} manquant")
missing = [name for name in SOCLE_FILES
if not (PI_DIR / "extensions" / f"{name}.ts").is_file()]
table.add_row("socle .pi/extensions", "[red]incomplet[/red]" if missing
else f"{len(SOCLE_FILES)}/{len(SOCLE_FILES)}",
f"manquants : {', '.join(missing)}" if missing else ", ".join(SOCLE_FILES))
if missing:
problems.append(f"nœud pi : socle incomplet sur disque — {missing}")
# 3. Les modèles du roster (et le secours) dans le registre de pi.
try:
factory = roster.load()
except (roster.RosterError, OSError) as error:
problems.append(f"roster : {error}")
factory = None
if factory is not None:
available, error = list_models()
routes = wanted_models(factory)
if error or not available:
table.add_row("modèles du roster", "[red]?[/red]", error or "pi --list-models n'a rien rendu")
problems.append("nœud pi : aucun modèle disponible — clé de la passerelle absente "
f"de .env ({harness.GATEWAY}) ?")
else:
absent = {who: route for who, route in routes.items() if route not in available}
for who, route in absent.items():
problems.append(f"modèle absent du registre de pi : {route} ({who}) — "
"pi facturerait un modèle synthétique au prix du défaut ; "
"`pi update --models`, ou corrigez le roster")
table.add_row("modèles du roster", "[red]absent[/red]" if absent
else f"{len(routes)}/{len(routes)}",
", ".join(f"{who}={route}" for who, route in absent.items())
or ", ".join(sorted(set(routes.values()))))
# 4. La batterie du socle (A8) : le socle chargé dans le vrai pipeline, à sec.
if quick:
table.add_row("batterie .pi/tests", "[yellow]sautée[/yellow]", "--quick")
elif not TESTS_DIR.is_dir() or shutil.which("bun") is None:
table.add_row("batterie .pi/tests", "[yellow]absente[/yellow]", "chapitre A8")
else:
# Les fichiers un par un, en chemins explicites : bun ignore les dossiers cachés
# quand il découvre les tests de lui-même (chapitre A8).
files = [f"./{p.as_posix()}" for p in sorted(TESTS_DIR.glob("*.test.ts"))]
run = subprocess.run([shutil.which("bun"), "test", *files],
capture_output=True, text=True, encoding="utf-8", errors="replace",
timeout=180)
if run.returncode == 0:
table.add_row("batterie .pi/tests", "verte", "socle chargé à sec, zéro jeton")
else:
tail = (run.stderr or run.stdout).strip().splitlines()[-1:] or ["?"]
table.add_row("batterie .pi/tests", "[red]rouge[/red]", tail[0][:60])
problems.append(f"nœud pi : la batterie du socle échoue — {tail[0][:200]}")
# ---------- l'ordonnance : installer une couche ailleurs ----------
def prescribe(layer: str, into: str, console: Console) -> int:
target = Path(into).resolve()
if not target.is_dir():
console.print(f"[red]✗[/red] dépôt cible introuvable : {target}")
return 2
settings_path = target / PI_DIR / "settings.json"
settings: dict = {}
if settings_path.is_file():
try:
settings = json.loads(settings_path.read_text(encoding="utf-8")) or {}
except json.JSONDecodeError as error:
console.print(f"[red]✗[/red] {settings_path} n'est pas un JSON valide : {error}")
return 2
entry = layer_entry(layer, ROOT, target)
settings_path.parent.mkdir(parents=True, exist_ok=True)
settings_path.write_text(json.dumps(apply_layer(settings, entry), indent=2, ensure_ascii=False)
+ "\n", encoding="utf-8")
console.print(f"[green]couche {layer} posée[/green] dans {settings_path}")
console.print(json.dumps(entry, indent=2, ensure_ascii=False))
console.print("approuvez le dépôt cible (`pi --approve` ou trust) : pi charge ses packages "
"manquants au démarrage, et rien avant.")
return 0
# ---------- la gate à sec : les fonctions pures, sans pi ni bun ----------
def selftest() -> None:
assert harness.version_too_old("0.84.4") and not harness.version_too_old("0.85.1")
assert extract_version("pi 0.85.1") == "0.85.1" and extract_version("0.9") == "0.9"
assert extract_version("présente") is None and extract_version(None) is None
table = ("provider model context max-out thinking images\n"
"openrouter z-ai/glm-5.3 1000K 128K yes no\n"
"openrouter deepseek/deepseek-v4-flash-0731 1000K 384K yes no\n"
"anthropic claude-opus-5 200K 32K yes yes\n")
models = parse_model_table(table)
assert models == {"openrouter/z-ai/glm-5.3", "openrouter/deepseek/deepseek-v4-flash-0731",
"anthropic/claude-opus-5"}, models
assert parse_model_table("") == set()
with tempfile.TemporaryDirectory() as tmp:
base = Path(tmp)
usine, cible = base / "plume-factory", base / "pi-essai"
usine.mkdir(), cible.mkdir()
entry = layer_entry("socle", usine, cible)
assert isinstance(entry, dict) and entry["source"] == "../../plume-factory", entry
assert entry["skills"] == [] and len(entry["extensions"]) == len(SOCLE_FILES)
assert entry["extensions"][0].startswith(".pi/extensions/") and "/" in entry["source"]
assert layer_entry("poste", usine, cible) == "../../plume-factory"
nested = base / "ateliers" / "pi-essai"
nested.mkdir(parents=True)
assert layer_entry("poste", usine, nested) == "../../../plume-factory"
# apply_layer : remplace l'entrée de la même source, garde les autres, jamais de doublon.
settings = {"defaultModel": "x", "packages": ["npm:autre", "../../plume-factory"]}
posed = apply_layer(settings, entry)
assert posed["packages"][0] == "npm:autre" and posed["packages"][-1] == entry
assert len(posed["packages"]) == 2 and posed["defaultModel"] == "x"
again = apply_layer(posed, layer_entry("poste", usine, cible))
assert again["packages"] == ["npm:autre", "../../plume-factory"]
try:
layer_entry("cuisine", usine, cible)
raise AssertionError("une couche inconnue doit être refusée")
except ValueError:
pass
print(f"doctor OK — versions comparées, catalogue lu ({len(models)} modèles), "
f"ordonnance socle ({len(SOCLE_FILES)} extensions, 0 skill) et poste, "
"chemin relatif à .pi/, pas de doublon")
def main() -> int:
parser = argparse.ArgumentParser(description="Le préflight du poste et du nœud pi — zéro jeton.")
parser.add_argument("--quick", action="store_true", help="sauter la batterie bun test")
parser.add_argument("--layer", choices=sorted(LAYERS), help="couche à installer ailleurs")
parser.add_argument("--into", help="dépôt cible de --layer")
parser.add_argument("--selftest", action="store_true", help="la gate à sec des fonctions pures")
args = parser.parse_args()
console = Console()
if args.selftest:
selftest()
return 0
if args.layer:
if not args.into:
parser.error("--layer exige --into <dépôt cible>")
return prescribe(args.layer, args.into, console)
problems: list[str] = []
table = Table(title="plume-factory — poste de pilotage")
table.add_column("Outil")
table.add_column("Version")
table.add_column("Rôle")
versions = check_tools(table, problems)
check_pieces(problems)
console.print(table)
node = Table(title="plume-factory — nœud pi")
node.add_column("Vérification")
node.add_column("État")
node.add_column("Détail")
check_pi_node(node, problems, versions["pi"], args.quick)
console.print(node)
if problems:
for problem in problems:
console.print(f"[red]✗[/red] {problem}")
return 1
console.print("[green]poste de pilotage et nœud pi : OK[/green]")
return 0
if __name__ == "__main__":
sys.exit(main())La gate du TP
Depuis la racine de plume-factory, une commande par ligne, identiques dans bash et PowerShell.
Les quatre premières sont les gates et le doctor, à zéro jeton. Les suivantes posent la couche
socle dans un dépôt voisin et vérifient qu’elle se charge.
bun .pi/extensions/factory-compact.ts
uv run python -m adws.adw_modules.harness
uv run adws/doctor.py --selftest
uv run adws/doctor.py
git init ../pi-essai
uv run adws/doctor.py --layer socle --into ../pi-essai
cd ../pi-essai
pi -p --approve --mode json "ping"
ls adws/adw_data/traces/harness
cd ../plume-factoryRésultat attendu, dans l’ordre :
factory-compact OK — bloc [FACTORY] en tête (… caractères), mission et verdict mot pour mot, mission reprise du bloc précédent, verdict mis à jour, fichiers cumulés, troncature annoncée, récit à secharness OK — … v7 : pi < 0.85.0 refuse, panne fournisseur distinguee, secours sans changer de sessiondoctor OK — versions comparées, catalogue lu (3 modèles), ordonnance socle (6 extensions, 0 skill) et poste, chemin relatif à .pi/, pas de doublon- Les deux tableaux du doctor, avec
binaire pi 0.85.x ≥ 0.85.0,socle .pi/extensions 6/6,modèles du roster 5/5(6/6 avec un secours),batterie .pi/tests verteet la ligne verteposte de pilotage et nœud pi : OK. - L’entrée
packagesécrite dans../pi-essai/.pi/settings.json, source../../plume-factory. - Un flux JSON pour le
ping, et un.jsonllisté parls: la trace du chapitre A4 a tourné sansadws/ni runner.
Pour voir le doctor rouge, ajoutez FACTORY_FALLBACK_MODEL=z-ai/glm-inexistant à votre
.env : il signale modèle absent du registre de pi : openrouter/z-ai/glm-inexistant ((secours)) et rend le code 1. Retirez ensuite la ligne et effacez ../pi-essai. Coût total :
moins d’un centime, pour le ping.
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.