Compaction structurée, fallback et doctor
Le dernier chapitre de l'annexe pi : un résumé de contexte qui recopie mot pour mot ce que le runner a dit, un port qui refuse un pi trop vieux et change de moteur sans changer de session sur une panne fournisseur, et un doctor qui vérifie tout cela à sec — puis installe l'usine pi ailleurs, en deux couches.
Au chapitre A9, vous avez rendu vos sessions pi indestructibles : déclarées avant l’appel, refusées
hors de leur dossier, clonées pour un best-of-N. Mais que devient une session qui dure ? Un build
de trente tours accumule des lectures de fichiers et des sorties de bun test. À un moment, pi
résume les anciens messages pour faire de la place, et les anciens messages, ce sont précisément
la mission, le handoff et le contrat que votre runner a injectés au premier tour. Un résumé libre
les paraphrase, ou les perd. À la fin de ce chapitre, ce que le runner a dit survivra mot pour mot
à toutes les compactions, votre port refusera un pi plus vieux que celui sur lequel il a été
vérifié et saura relayer une panne de fournisseur sur un moteur de secours sans quitter la session,
et un seul uv run adws/doctor.py vous dira, à sec, si l’usine peut tourner, puis l’installera
ailleurs en deux couches. Trois pièces : factory-compact.ts, sixième extension du socle,
harness.py (qui remplace la version du chapitre A9) et doctor.py (qui remplace celle du
chapitre 4). C’est le dernier chapitre de l’annexe, et il ferme la boucle ouverte au chapitre A8 :
l’épinglage des versions, promis alors, se pose ici.
Préserver l’enveloppe à travers la compaction
L’idée en une phrase
La compaction est le geste par lequel pi remplace les anciens messages d’une session par un
résumé quand le contexte approche de sa fenêtre. Dans un nœud d’usine, l’extension
factory-compact.ts (socle, côté pi) intercepte ce geste et impose un résumé en deux
parties : un bloc [FACTORY] que le code recopie mot pour mot (mission, contrat, dernière
correction du runner), puis un récit que le modèle rédige. Ce que le runner a injecté
n’est ainsi jamais paraphrasé.
Points clés
- Quand pi compacte, vérifié sur pi 0.85. Dès que les jetons du contexte dépassent la fenêtre
du modèle moins
compaction.reserveTokens(16 384 dans vos réglages du chapitre A1), pi vérifie le seuil après chaque lot d’outils, résume les messages les plus anciens jusqu’à ne garder quekeepRecentTokensrécents (12 000 chez vous), et reprend dans le même tour. Le résumé prend la place des anciens messages, et le modèle voit désormais : le prompt système, le résumé, puis les messages récents./compactfait la même chose à la demande. - Ce qui part en premier, c’est ce que le runner a dit. Le premier message utilisateur d’une
phase (brief, mission, handoff, contrat de la porte typée du chapitre A6) est le plus ancien de
tous. Le résumé par défaut en fait une ligne « Goal : … ». La forme exacte de l’enveloppe, la
consigne « appelle
report_phaseseul, en dernier », les chemins autorisés : autant de phrases que le modèle réécrit à sa manière, et qu’il peut omettre. session_before_compactrend la main à l’extension. L’événement livre une préparation (messages à résumer, résumé précédent, fichiers lus et modifiés, jetons avant, identifiant du premier message conservé, réglages) et accepte trois réponses : rien (pi résume lui-même),cancel, ou une compaction fournie (summary,firstKeptEntryId,tokensBefore, et undetailslibre). C’est cette troisième voie que prend le socle.- Le bloc est du code, le récit est délégué. Le bloc
[FACTORY]se construit sans modèle : phase, rôle et run lus dans l’environnement posé par la fabriqueagent_action(chapitre A7), mission = premier message utilisateur, dernier mot du runner = dernier message utilisateur qui quitte le contexte (une correction, une relance après coupe-circuit). Chaque ancre a ses propres bornes, car les asks de l’usine contiennent eux-mêmes des titres### mission, un découpage par titres les casserait. À la compaction suivante, la mission n’est plus dans les messages : elle est relue dans le bloc précédent. Elle survit donc à N compactions, à l’identique. - Ce que pi ne cumule pas pour vous, le socle le cumule. pi additionne les fichiers lus et
modifiés d’une compaction à l’autre, mais seulement pour ses compactions : pour un résumé
fourni par extension, le compteur repart de zéro. L’extension relit donc les balises
read-filesetmodified-filesdu résumé précédent et les fusionne. Et si le modèle ne peut pas résumer (panne, abandon), un récit à sec, déterministe et à zéro jeton, prend la place du récit : le bloc, lui, est toujours là.
Exemple concret
Un build du chapitre 13 sur le workhorse z-ai/glm-5.3, relevé ce jour sur
openrouter.ai/models : 1,40 $ le million de jetons en entrée, 4,40 $ en sortie. Le builder lit
la spec, ouvre six fichiers, lance bun test quatre fois : au quinzième tour, le contexte frôle
la fenêtre configurée et pi compacte. Résumé par défaut : « 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 par un paragraphe en prose, sans
tool_execution_end pour report_phase. Le runner renvoie correction(), un tour de plus, et
cette fois le modèle redemande la forme de l’enveloppe : elle n’est plus nulle part dans son
contexte. Un aller-retour à quelques centimes, parfois deux, et sur un modèle ouvert moins docile,
un run rouge. Avec le socle, le résumé commence par le bloc [FACTORY] : la mission entière, le
contrat entier, la dernière correction. Le récit du modèle vient après. Coût de la compaction
elle-même : identique dans les deux cas, soit un appel qui relit les messages résumés (quelques
dizaines de milliers de jetons, ou quelques centimes sur le workhorse). Le bloc ajoute
ensuite un à deux milliers de jetons à chaque tour, une fraction de centime. Ce que vous
achetez, ce n’est pas un résumé moins cher : c’est 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 dans un rendu markdown mais faciles à retrouver par le code, et c’est ce qui rend
le bloc relisible à la compaction suivante. Rien de tout cela ne traverse la couture vers le
runner : le runner ne sait même pas qu’une compaction a eu lieu, sinon par la ligne compaction
que la trace du chapitre A4 enregistre.
<!-- [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
Le gestionnaire tient en une décision : sans FACTORY_PHASE, c’est un poste, pi résume comme il
l’entend. Avec elle, le bloc est construit par le code, le récit est demandé au modèle de la session
par le registre de pi, et l’ensemble est rendu comme compaction fournie. Côté Claude Code, il n’y
a rien d’équivalent à poser : ses hooks PreCompact et PostCompact savent bloquer une
compaction ou en prendre acte, pas fournir le résumé, et c’est un argument de plus pour pi comme
nœud d’usine sur les phases longues.
// .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 } };
});Piège courant : « mon payload est petit, la compaction ne me concerne pas » est inexact. Le seuil ne dépend pas de la taille du dépôt mais de ce que les outils rapportent : une sortie de
bun testverbeuse ou trois lectures de fichiers de 20 Ko suffisent à franchirkeepRecentTokensen quelques tours. Et « il suffit de désactiver la compaction » n’est pas une issue : sans elle, le contexte déborde, le fournisseur rend une erreur, et la phase meurt au mur de temps, après avoir payé ses jetons. La compaction est ce qui garde une phase longue en vie, et la seule question est ce qui y survit.
Fallback, doctor et package en deux couches
L’idée en une phrase
Trois gestes du code déterministe, côté runner, pour une même garantie : l’usine se
vérifie avant de tourner, tient pendant la phase, et voyage après. Une version de
pi épinglée que le port refuse en dessous et que le doctor compare à sec, un modèle de
secours appliqué par le port sur une panne fournisseur seulement, dans la même session, et
une installation en deux couches, socle ou poste, par une entrée packages filtrée que le
doctor rédige.
Points clés
- La version se vérifie deux fois, jamais trois.
PI_MIN_VERSIONvit dansharness.py, à côté des noms d’événements que le port lit. L’inventaire du chapitre A8 rapporte la version du binaire qui a tourné : en dessous du minimum,_judgerefuse la phase, en fail-closed, comme un socle manquant. Le doctor fait la même comparaison à sec, avant tout run, pour le binairepiet pour le SDK denode_modules(celui que la batterie du chapitre A8 exerce), et signale quand les deux divergent. La correction s’appellepi update, jamais un contournement. - Un modèle absent du registre ne fait pas échouer pi, il facture. Vérifié dans le résolveur
de pi 0.85 : un
--modelinconnu du fournisseur donne un modèle synthétique, cloné du défaut du fournisseur avec son prix, et un simple avertissement. Le port vérifie déjà le modèle observé après coup (chapitre A6), et le doctor le vérifie avant, en lisantpi --list-modelset en exigeant la route exacte pour chaque agent du roster et pour le secours.pi update --modelsrafraîchit le catalogue. - Le secours répond à une panne de fournisseur, à rien d’autre. pi réessaie déjà les erreurs
transitoires (vos réglages
retrydu chapitre A1), et quand il renonce le dernier message portestopReason: erroret unerrorMessage. Le port y cherche une marque de panne (429, 5xx, surcharge, « no endpoints ») et seulement alors relance une fois, surFACTORY_FALLBACK_MODEL, dans la même session, avec le motif. Jamais suraborted(c’est le coupe-circuit), jamais sur une erreur d’outil (le modèle doit la lire), jamais sur le mur de temps, jamais sur une clé invalide (un humain doit la voir). Jamais côté pi : pi ne connaît ni le roster ni le budget, et le code dispose, y compris du moteur. - Le secours est une décision d’infrastructure, pas de rôle. Une panne touche un fournisseur,
pas un agent : le modèle de secours vit dans
.env, à côté de la clé, et vaut pour toute l’usine (un appel peut le surcharger parfallback_modeldans saHarnessRequest). Le registre des sessions (chapitre A9) reçoit une seconde ligne quand le relais a lieu, etHarnessResult.modeldit quel moteur a répondu. Côté Claude Code, pas de secours : l’adaptateur reste natif Anthropic. - Deux couches, une entrée de réglages. pi sait filtrer ce qu’il charge d’un package par une
entrée objet dans
packages(source, puisextensions,skills,prompts,themes). La couche socle = les six extensions du nœud, zéro skill, et la couche poste = tout. Le doctor écrit l’entrée dans le.pi/settings.jsondu dépôt cible, avec un chemin relatif au dossier.pi/, car c’est ainsi que pi résout un chemin local, et c’est l’erratum du chapitre A5 : pour deux dépôts voisins,pi install -lstocke../../plume-factory, pas../plume-factory.
Exemple concret
Lundi matin, just doctor : quatre secondes, zéro jeton. Le tableau du nœud pi dit
binaire pi 0.84.4 < 0.85.0 en rouge : un pi update a été fait sur le portable, pas sur ce
poste. Sans le doctor, le premier scout de la journée aurait tourné, payé son tour (moins d’un
centime), puis été refusé par le port. Un adw_sdlc complet aurait payé sa phase de plan
(une dizaine de centimes) avant le même refus. Même journée, un build sur le workhorse : au
quatorzième tour, une trentaine de centimes déjà dépensés, le fournisseur rend 503 overloaded.
pi réessaie deux fois, quelques secondes, puis renonce : stopReason: error. Avant ce chapitre,
HarnessError, échec de phase, reprise dans la même session… sur le même moteur, qui rend souvent
le même 503. Trois tentatives plus tard, le run est rouge et un quart d’heure est passé. Avec le
secours déclaré dans .env (disons un modèle intermédiaire d’un autre fournisseur, deepseek/ deepseek-v4-flash-0731 au moment d’écrire, quelques dizaines de centimes le million en entrée),
le port relance une fois, dans la même session : le nouveau moteur relit les quatorze tours
(quelques dizaines de milliers de jetons, un à deux centimes), reprend où l’autre s’est
arrêté, et rend son enveloppe par la porte typée. La phase coûte ce qu’elle aurait coûté, plus
une relecture, et la trace dit qui a fini le travail.
Un arrêt, un geste — et un seul
| 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 |
Commande — le doctor, puis l’ordonnance
Le doctor se lit du haut vers le bas : l’outillage, les pièces, puis le nœud pi (versions,
socle sur disque, réglages et règles, modèles du roster dans le registre, batterie du chapitre A8).
Le premier rouge suffit à rendre un code non nul, et just doctor (chapitre 5) le lance tel quel.
Les deux dernières commandes rédigent l’ordonnance : une entrée packages filtrée dans un autre
dépôt. Le doctor sonde aussi le binaire claude, mais le nœud vérifié est pi : Claude Code n’a ni
socle, ni inventaire, ni version épinglée par le port.
# 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 telle qu’elle apparaît dans ../pi-essai/.pi/settings.json après --layer socle : les
chemins sont relatifs à la racine du package, la source est relative au .pi/ du dépôt cible. Et
la ligne facultative de votre .env qui arme le secours : un identifiant du registre, vérifié par
le doctor, jamais le même que le moteur qu’il remplace.
{
"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ège courant : « pi réessaie déjà, le secours est redondant » est inexact. Le retry de pi relance le même moteur chez le même fournisseur : c’est le bon geste pour une erreur transitoire, et le mauvais pour une surcharge qui dure une heure. Le secours change de moteur, c’est une autre décision, et elle appartient au code qui connaît le roster et le budget. Inversement, « un secours par agent dans le roster serait plus fin » confond deux niveaux : la panne est celle d’un fournisseur, pas d’un rôle, et un builder relancé sur un modèle léger, choisi rôle par rôle, produirait un patch qu’aucune gate n’aurait demandé.
Fil rouge — la pièce posée aujourd’hui
Sur le plan de l’usine, le chapitre ferme l’annexe en touchant ses trois couches : le socle
gagne sa sixième extension, factory-compact.ts, à côté de l’inventaire (A8) et du coupe-circuit
(A7). Le port (chapitre 7, adaptateur v7) apprend à refuser un pi trop vieux et à relayer une
panne de fournisseur, et le poste de pilotage retrouve son doctor.py du chapitre 4, qui
diagnostique désormais le nœud pi et rédige l’ordonnance qui l’installe ailleurs. La couture ne
bouge pas : l’agent propose (un récit de compaction, une enveloppe sur un moteur de secours) et
le code dispose du bloc recopié mot pour mot, de la version admise, du moteur qui reprend, de la
couche qui voyage. Déterministe : le bloc [FACTORY], ses bornes et sa relecture, le cumul des
fichiers, la comparaison de versions, la détection d’une marque de panne, le choix du secours, la
route exacte dans le registre, l’entrée de réglages filtrée. Délégué : le récit du travail
accompli, et la reprise de la mission sur l’autre moteur. Coût d’usage : zéro jeton pour le
doctor et l’ordonnance, une compaction coûte ce qu’elle coûtait plus une fraction de centime
par tour pour le bloc, et un relais de secours coûte une relecture de session, quelques centimes, à
la place d’un run rouge après un quart d’heure. Deux notes de continuité : le runner continue
d’exiger les cinq extensions du chapitre A8 (SOCLE ne change pas, car la compaction n’est pas
une garde, son absence n’ouvre aucune porte), tandis que le doctor et la couche socle en
attendent six. L’extension d’injection de contexte annoncée au chapitre A6 reste, au terme de
l’annexe, une pièce que vous pouvez écrire vous-même sur le modèle de celle-ci : le bloc
[FACTORY] en est déjà la moitié.
Travaux pratiques — la pièce du jour
Trois fichiers à poser dans plume-factory, qui devient, chapitre après chapitre, votre usine
logicielle agentique : la compaction structurée, le port qui épingle et relaie, le doctor qui
vérifie et prescrit. Prérequis : pi ≥ 0.85 (chapitre A1, pi update sinon), Bun et 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, nouvelle. Six fonctions pures exportées (readFacts,
pickAnchors, buildBlock, splitBlock, parseAnchors, mergeFiles), un récit à sec
(dryNarrative), une extension qui n’écoute qu’un seul événement, une ligne dans le journal du
chapitre A4, une entrée de session, et la gate du fichier sous Bun. Elle ne modifie aucune pièce
posée, et sur un poste sans FACTORY_PHASE elle ne fait rien.
// .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 : mêmes dataclasses,
même run(), même registre des sessions, mêmes drapeaux factorisés, même adaptateur Claude Code,
et harness_rpc.py et adw_bon.py du chapitre A9 tournent tels quels. S’ajoutent la version
épinglée (PI_MIN_VERSION, version_too_old, le refus dans _judge), la lecture de
errorMessage dans le flux, la détection d’une panne fournisseur (provider_failure), le choix
du secours (fallback_for : la requête, puis .env, jamais le même moteur), le relais dans
_run_pi (une relance, même session, motif en tête) et le champ model de HarnessResult. Le profil d’authentification du chapitre 17bis (auth:, route, coffre) est conservé.
"""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. Le tableau des outils et la liste des pièces
restent, mis à jour, et s’ajoute le tableau du nœud pi : version du binaire et du SDK comparées à
harness.PI_MIN_VERSION, socle sur disque (six fichiers), réglages et règles présents, chaque
route du roster (et le secours) trouvée dans pi --list-models, batterie du chapitre A8 verte.
Le doctor importe le port et le roster : c’est ainsi qu’il partage leur version épinglée, leur
liste du socle et leur lecture de .env. Ses fonctions pures (parse_model_table, layer_entry,
apply_layer) portent la gate à sec, et --layer rédige l’ordonnance.
#!/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 ne coûtent rien : la gate de l’extension, celle du port, celle du doctor,
puis le doctor lui-même (la batterie du chapitre A8 comprise). Les suivantes créent un dépôt
vierge voisin, y posent la couche socle, et prouvent qu’elle se charge, pour un ping à moins d’un
centime.
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 : 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 à sec, puis harness OK — … v7 : pi < 0.85.0 refuse, panne fournisseur distinguee, secours sans changer de session, puis doctor OK — versions comparées, catalogue lu (3 modèles), ordonnance socle (6 extensions, 0 skill) et poste, chemin relatif à .pi/, pas de doublon. Viennent ensuite les deux tableaux du doctor : binaire pi 0.85.x ≥ 0.85.0, socle .pi/extensions 6/6, modèles du roster 5/5 (6/6 si votre .env déclare un secours), batterie .pi/tests verte, et la ligne verte poste de pilotage et nœud pi : OK. L’ordonnance affiche l’entrée
packages écrite dans ../pi-essai/.pi/settings.json, source ../../plume-factory. Le ping
rend un flux JSON, et le ls liste un fichier .jsonl : la trace du chapitre A4, une extension
du socle, a tourné dans un dépôt sans adws/ ni runner. Ouvrez pi dans pi-essai pour le
constater à l’œil : pas de pied de page (couche poste absente), mais /factory-inventory répond.
Pour voir le doctor rouge sans rien casser, ajoutez FACTORY_FALLBACK_MODEL=z-ai/glm-inexistant
à votre .env et relancez-le : modèle absent du registre de pi : openrouter/z-ai/glm-inexistant ((secours)), code de retour 1, puis retirez la ligne. Coût : les quatre gates et l’ordonnance
ne dépensent rien (le doctor prend quelques secondes, batterie comprise), et le ping moins
d’un centime. Effacez ../pi-essai quand vous avez fini. Variante éco : elle est déjà là, rien
dans ce chapitre n’invoque un modèle frontier, et le secours que vous déclarez est, par choix, un
moteur intermédiaire.
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.