Vérifier le socle, pas y croire

Depuis le chapitre A6, votre runner lance pi avec --approve, et vous savez pourquoi : sans approbation du dépôt, pi ignore .pi/settings.json et .pi/extensions/ sans un mot. Mais qu’est-ce qui vous prouve, run après run, que le damage control, la trace, la porte typée et le coupe-circuit sont réellement là ? Rien : un fichier renommé, un pi update qui casse un import, un dépôt jeté sur une boîte du module 6 sans son .pi/, et le nœud tourne nu. Il rend même une enveloppe, parfois. À la fin de ce chapitre, votre runner refusera une phase dont le socle n’est pas prouvé, et vous saurez tester chacune de vos extensions dans le pipeline réel de pi, sur un fournisseur factice, en une demi-seconde et zéro jeton. La pièce du jour ferme la faille la plus silencieuse du nœud pi : une cinquième extension du socle, factory-inventory.ts, un verdict plus strict dans harness.py (qui remplace la version du chapitre A6), et le premier dossier .pi/tests/ de l’usine.

Inventaire fail-closed du nœud pi

L’idée en une phrase

Un inventaire est une pièce du socle, une extension déterministe qui, à chaque session, dresse la liste des extensions attendues et trouvées, puis la dépose dans le flux de pi comme une entrée typée, plus un verdict côté runner qui lit cette entrée et refuse la phase quand elle manque ou déclare un manquant : c’est le principe fail-closed du chapitre A3 appliqué au socle lui-même.

Points clés

• L’absence n’est pas un succès. Un run sans trace harnais, sans refus de damage control et sans ligne guard / armed ressemble exactement à un run sain qui n’a rien eu à refuser. Le runner v4 ne distinguait pas les deux. Le runner v5 exige une preuve positive : l’entrée factory-inventory dans le flux JSON, avec missing vide.
• Trois niveaux de preuve, tous vérifiés le jour même sur pi 0.84. Le fichier est sur le disque (.pi/extensions/nom.ts), pi a démarré (une extension qui jette au chargement fait sortir pi en code 1, dans tous les modes), et l’extension a laissé une provenance : pi.getAllTools() et pi.getCommands() portent sourceInfo.path, le chemin du fichier qui a enregistré l’outil ou la commande. L’inventaire note les trois, et le runner juge sur les deux premiers, les preuves de provenance restent lisibles dans la trace.
• Le dépôt se fait au premier tour, pas à session_start. En --mode json, pi ne relaie sur stdout que les événements émis après son abonnement à la session, et session_start le précède. Une entrée déposée là est bien écrite dans la session, mais le runner ne la verrait jamais. L’inventaire se dresse à session_start (tout est chargé) et se dépose à agent_start, une fois par session.
• Le runner nomme ce qu’il attend. La liste du socle vit dans harness.py (SOCLE) et part au nœud par la variable FACTORY_EXPECTED : une seule source de vérité, côté Python, comme le schéma de la porte au chapitre A6. Un roster de test ou un run « à sec » peut passer socle=() et ne rien exiger. L’usine, elle, exige les cinq.
• Le refus tombe après la phase. Un processus par tentative (chapitre A6) veut dire que le runner lit le flux quand pi a rendu la main : la phase a déjà coûté ses jetons quand le verdict tombe. Ce que l’inventaire garantit, c’est qu’un nœud nu ne passe jamais pour un run vert. La vérification avant tout run, à sec, sera le travail d’adw_doctor au chapitre A10.

Exemple concret

En préparant ce chapitre, j’ai lancé le socle sur un modèle factice (celui du second sous-thème) qui joue toujours le même scénario : un bash rm -rf adws, puis une enveloppe success par report_phase. Avec --approve, tout se passe comme prévu : l’inventaire arrive en tête du flux (present : cinq noms, missing : vide), le damage control refuse le rm avec son motif, l’enveloppe passe la porte, et le runner v5 rend un HarnessResult vert. Le même scénario avec --no-approve, ce que fait un runner d’avant A6 sur un dépôt non approuvé : aucune extension chargée, le rm -rf adws est exécuté, le dossier de l’usine disparaît, et pi sort en code 0 avec une enveloppe valide. Le runner v4 aurait compté un run vert. Le runner v5 lit le flux, ne trouve pas d’inventaire et lève socle pi non charge : aucun inventaire dans le flux, session conservée. Sur votre roster, la démonstration coûte ce qu’un scout coûte sur le modèle léger (deepseek/deepseek-v4-flash-0731, relevé ce jour : ~0,07 $ le million de jetons en entrée, ~0,18 $ en sortie) : moins d’un centime, une trentaine de secondes. Sur le fournisseur factice, zéro.

Ce que prouve quoi

Question	Ce qui répond	Ce que ça ne prouve pas
Le fichier existe-t-il ?	listExtensionFiles sur .pi/extensions/	qu’il a été chargé
pi l’a-t-il chargé ?	pi a démarré (une extension qui jette ⇒ code 1) et le dépôt est approuvé (ctx.isProjectTrusted())	qu’il fait quelque chose
A-t-il pris sa place ?	sourceInfo.path des outils et commandes (proofs)	rien pour une extension qui n’enregistre ni outil ni commande
Le runner l’a-t-il vu ?	l’entrée factory-inventory sur stdout, missing vide	que le run est bon — c’est le travail des gates

Config — l’entrée que lit le runner

Voici, tel qu’il sort sur stdout en --mode json, l’événement que _read_pi_stream retient (données abrégées). C’est une enveloppe au sens du livre : typée, déposée par du code, lue par du code, jamais par le modèle.

{"type": "entry_appended", "entry": {"type": "custom", "customType": "factory-inventory",
  "data": {"pi_version": "0.84.4", "model": "openrouter/z-ai/glm-5.3",
           "trusted": true, "headless": true,
           "expected": ["damage-control", "factory-obs", "factory-report", "factory-guard", "factory-inventory"],
           "present":  ["damage-control", "factory-obs", "factory-report", "factory-guard", "factory-inventory"],
           "missing":  [],
           "files": ["damage-control", "factory-chain", "factory-footer", "factory-guard", "factory-inventory", "factory-obs", "factory-report"],
           "proofs": {"factory-report": ["tool:report_phase"], "damage-control": ["command:/damage-control"]}}}}

Et le verdict, dans l’ordre où harness._judge le rend : le socle d’abord, la porte ensuite.

# adws/adw_modules/harness.py — extrait : sans preuve, pas de phase — même avec une enveloppe.
if request.socle:
    if reading.inventory is None:
        raise HarnessError("socle pi non charge : aucun inventaire dans le flux — ...", 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} — ...", session_id)
if reading.envelope is not None:          # seulement maintenant : la porte typee
    text = json.dumps(reading.envelope, ensure_ascii=False)

Piège courant : « si pi démarre, le socle est chargé » est inexact. Pi démarre très bien sans le socle : sans approbation du dépôt, les fichiers de .pi/ sont simplement ignorés, sans diagnostic. Ce que garantit le code 1, c’est qu’une extension chargée qui échoue ne passe pas inaperçue. Une extension jamais chargée, elle, ne fait aucun bruit. Seule une preuve positive distingue les deux.

---

Tester les extensions hors réseau avec le SDK

L’idée en une phrase

Le SDK de pi monte une session complète (extensions du socle chargées depuis .pi/extensions/, outils intégrés réels, session en mémoire) sur un fournisseur factice dont vous scriptez les réponses. Vos tests, sous bun test, vérifient alors le comportement du socle branché sur pi (l’événement arrive-t-il, le refus passe-t-il, l’entrée est-elle déposée ?), à zéro jeton, en quelques centaines de millisecondes, côté code entièrement.

Points clés

• Le fournisseur factice est un vrai fournisseur. fauxProvider() (paquet @earendil-works/pi-ai, celui que pi utilise lui-même) rend un provider dont les réponses sont une file scriptée : fauxAssistantMessage(...) avec fauxText, fauxToolCall(nom, args) et un stopReason. Pi le consomme comme n’importe quel modèle, et le flux, les événements et les outils sont ceux de la production. L’usage est même estimé (un jeton pour quatre caractères), ce qui permet d’exercer le compteur de coût du chapitre A7.
• Quatre objets, tous en mémoire. ModelRuntime.create avec un InMemoryCredentialStore et modelsPath: null (aucun fichier de votre poste n’est lu), auquel on enregistre le provider factice, DefaultResourceLoader avec cwd sur un dépôt jetable et agentDir hors de ~/.pi (vos extensions personnelles ne s’invitent pas), SettingsManager.inMemory, SessionManager.inMemory. createAgentSession assemble le tout.
• bindExtensions d’abord, subscribe ensuite, prompt enfin. C’est bindExtensions qui émet session_start : sans lui, le damage control n’a pas chargé ses règles et le rm -rf passe (je l’ai vu passer). Le mode headless de pi fait exactement cette séquence, et le test la reproduit.
• Les outils sont réels : le dépôt de test doit être jetable. bash exécute vraiment la commande, donc le test copie le socle et les règles dans un dossier temporaire, y pointe cwd, et le supprime après. Les effets de bord du socle (journal harnais, rapport de garde) tombent dans ce dossier, selon la règle du plan de l’usine : le socle n’écrit que sous adws/adw_data/.
• Ce qu’on teste ici, et ce qu’on ne teste plus. La logique pure de chaque extension a déjà sa gate (bun .pi/extensions/nom.ts, chapitres A3 à A7). Le test SDK vérifie le branchement : l’inventaire arrive à agent_start, rm -rf sort en isError avec le motif, le troisième appel identique est refusé sans réponse de modèle supplémentaire, le secret est masqué dans le toolResult, l’enveloppe arrive par tool_execution_end et termine le tour. C’est aussi le filet contre les montées de version de pi : un événement renommé casse le test, pas un run.

Exemple concret

La batterie du jour compte huit scénarios, et sur mon poste bun test les passe en une demi-seconde, sans réseau. Le même contrôle « en vrai », avec un scout par scénario sur le modèle léger : huit runs à moins d’un centime, une trentaine de secondes chacun, soit quatre à cinq minutes, avec un résultat qui dépend de l’humeur du modèle : pour vérifier que la boucle est coupée au troisième appel identique, il faudrait d’abord qu’un modèle veuille bien boucler. Le fournisseur factice boucle sur commande. Le rapport est de l’ordre de un pour cinq cents en temps, et de zéro à quelques centimes en argent. Surtout, il rend le socle reproductible, ce qu’aucun run sur un vrai modèle ne sera jamais tout à fait.

Trois façons d’éprouver une extension

Banc	Ce qu’il exerce	Coût	Ce qu’il rate
Gate de fichier (bun .pi/extensions/nom.ts)	la logique pure exportée	zéro, ~50 ms	le branchement sur les événements de pi
Test SDK (bun test, fournisseur factice)	le pipeline réel de pi, réponses scriptées	zéro, ~0,5 s la batterie	un vrai modèle, un vrai fournisseur, la facture
Run réel (uv run adws/adw_scout.py)	tout, y compris le modèle et la passerelle	des centimes, des dizaines de secondes	la reproductibilité

Script — une session pi sur le fournisseur factice

Le cœur du fichier de test : tout le reste, ce sont les scénarios. Les imports viennent des deux paquets que la gate installe en développement, et aucun des deux n’est nécessaire à l’exécution des extensions.

// .pi/tests/socle.test.ts — extrait : le banc, une session complète en mémoire.
const faux = fauxProvider();                       // le modèle factice, réponses scriptées
faux.setResponses(responses);
const modelRuntime = await ModelRuntime.create({ credentials: new InMemoryCredentialStore(), modelsPath: null });
modelRuntime.registerNativeProvider(faux.provider);

const settingsManager = SettingsManager.inMemory({ compaction: { enabled: false }, retry: { enabled: false } });
const resourceLoader = new DefaultResourceLoader({ cwd, agentDir: join(cwd, "agent"), settingsManager, noSkills: true, noPromptTemplates: true });
await resourceLoader.reload();                      // charge .pi/extensions/ du dépôt jetable
expect(resourceLoader.getExtensions().errors).toEqual([]);

const { session } = await createAgentSession({
  cwd, agentDir: join(cwd, "agent"), model: faux.getModel(), modelRuntime, resourceLoader, settingsManager,
  sessionManager: SessionManager.inMemory(cwd), tools,
});
await session.bindExtensions({ mode: "json" });    // émet session_start — comme le mode headless
session.subscribe((event) => { events.push(event); });
await session.prompt("phase de test");

Commande — installer le banc, lancer la batterie

Deux paquets en développement (ils s’ajoutent au package.json posé au chapitre A5), puis la batterie. Le ./ compte : bun test ignore les dossiers cachés quand il découvre les fichiers, un chemin explicite lève l’ambiguïté. Côté Claude Code, il n’y a rien d’équivalent à poser : l’annexe équipe le nœud pi, et le socle n’existe que de ce côté de la couture.

bun add -d @earendil-works/pi-coding-agent @earendil-works/pi-ai
bun test ./.pi/tests/socle.test.ts

Piège courant : « la batterie est verte, donc mon pi est bon » est inexact. Les tests exercent la version du SDK installée dans node_modules, pas le binaire pi que lance le runner. Un pi update sur le poste et un bun add dans le dépôt peuvent diverger d’une version. Le socle tolère cet écart tant que les événements ne changent pas de nom. L’épinglage des deux versions, et le refus du runner quand elles divergent, est la pièce du chapitre A10.

---

Fil rouge — la pièce posée aujourd’hui

Sur le plan de l’usine, la pièce du jour ferme le socle pi : cinquième extension chargée dans tous les modes, après damage-control.ts (A3), factory-obs.ts (A4), factory-report.ts (A6) et factory-guard.ts (A7). Elle remonte jusqu’au port harnais (chapitre 7, adaptateur v5) et ouvre une nouvelle zone, .pi/tests/, le banc d’essai du nœud. La couture ne bouge pas : le runner Python possède le graphe, l’agent reste un nœud borné. Ce qui change, c’est que le runner ne suppose plus son nœud protégé, il le vérifie : l’inventaire est une enveloppe déposée par du code, lue par du code, et un nœud sans preuve est un échec de phase, session conservée, avant même de regarder l’enveloppe du modèle. Déterministe : la liste attendue (SOCLE), son transport (FACTORY_EXPECTED), l’inventaire, le verdict _judge, et toute la batterie de tests. Délégué : rien de nouveau. Coût d’usage : zéro jeton pour l’inventaire et pour les tests. Un refus coûte la phase qui l’a révélé, moins d’un centime sur le modèle léger et quelques dizaines de centimes sur un builder workhorse, contre, sans lui, une série de runs « verts » sans damage control, sans trace et sans coupe-circuit, dont personne n’aurait rien vu. Deux notes de continuité : le miroir bloquant de factory-obs.ts à session_shutdown (chapitre A4) reste tel quel aujourd’hui, et il passera côté runner avec les évolutions de la reprise au chapitre A9. L’extension d’injection de contexte annoncée au chapitre A6 attend toujours son tour, le budget de fichiers du jour étant pris par les trois pièces ci-dessous.

---

Travaux pratiques — la pièce du jour

Trois fichiers à poser dans plume-factory, qui devient, chapitre après chapitre, votre usine logicielle agentique : la cinquième extension du socle, le premier fichier de tests du nœud pi, et le port harnais qui apprend à refuser. Prérequis : Bun (chapitre 2) et pi (chapitre A1). La gate installe les deux paquets de développement.

Pièce — .pi/extensions/factory-inventory.ts

L’inventaire du socle. Trois fonctions pures exportées (expectedFrom, listExtensionFiles, takeInventory), une extension qui dresse à session_start et dépose à agent_start, une commande /factory-inventory pour le poste, et la gate du fichier sous Bun. Elle s’appuie sur les quatre extensions précédentes du socle sans les modifier, et lit FACTORY_EXPECTED posé par le runner. Aucune dépendance npm à l’exécution.

// .pi/extensions/factory-inventory.ts — l'inventaire du socle : vérifier, pas croire.
// Au démarrage de chaque session, cette extension dresse la liste des extensions du socle
// réellement chargées et l'écrit dans la session (appendEntry). Le runner Python la lit sur
// stdout (entry_appended, --mode json) et REFUSE la phase si l'entrée manque ou si un nom
// attendu manque : sans inventaire, le nœud pi tourne sans protection — et cela ne doit
// jamais passer en silence. Coût : zéro jeton. Aucune dépendance npm à l'exécution.
import { existsSync, readdirSync } from "node:fs";
import { basename, join } from "node:path";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { VERSION } from "@earendil-works/pi-coding-agent";

export const ENTRY_TYPE = "factory-inventory";   // le customType lu par le runner
export const EXPECTED_ENV = "FACTORY_EXPECTED";  // la liste attendue, posée par le runner
export const EXTENSIONS_DIR = join(".pi", "extensions");

// Le socle : les extensions chargées dans TOUS les modes, que le runner exige.
// factory-inventory s'y compte : si ce fichier est renommé, l'entrée n'arrive plus — et
// le runner refuse la phase de la même façon. Le poste (footer, chaîne) n'en fait pas partie.
export const DEFAULT_EXPECTED = ["damage-control", "factory-obs", "factory-report", "factory-guard", "factory-inventory"];

export interface Provenance { name: string; path: string }

export interface Inventory {
  pi_version: string;
  model: string | null;
  trusted: boolean;
  headless: boolean;
  expected: string[];
  present: string[];                    // attendus ET trouvés dans .pi/extensions/
  missing: string[];                    // attendus mais absents — le runner refuse la phase
  files: string[];                      // tout ce que .pi/extensions/ contient (socle + poste)
  proofs: Record<string, string[]>;     // par extension, ce qu'elle a enregistré (outils, commandes)
}

// ---------- les fonctions pures : lire la liste attendue, dresser l'inventaire ----------

export function expectedFrom(env: Record<string, string | undefined>): string[] {
  const raw = env[EXPECTED_ENV];
  if (raw === undefined) return DEFAULT_EXPECTED;
  return raw.split(",").map((s) => s.trim()).filter((s) => s.length > 0);
}

// Le nom d'une extension = son fichier sans suffixe.
export function extensionName(path: string): string {
  return basename(path).replace(/\.(ts|js|mjs|cjs|tsx)$/, "");
}

export function listExtensionFiles(cwd: string): string[] {
  const dir = join(cwd, EXTENSIONS_DIR);
  if (!existsSync(dir)) return [];
  return readdirSync(dir)
    .filter((f) => /\.(ts|js|mjs|cjs|tsx)$/.test(f) && !f.endsWith(".d.ts"))
    .map(extensionName)
    .sort();
}

export function takeInventory(input: {
  files: string[];
  expected: string[];
  tools: Provenance[];
  commands: Provenance[];
  pi_version: string;
  model: string | null;
  trusted: boolean;
  headless: boolean;
}): Inventory {
  const present = input.expected.filter((name) => input.files.includes(name));
  const missing = input.expected.filter((name) => !input.files.includes(name));
  // Les preuves : ce que chaque extension a enregistré, reconnu par sa provenance (sourceInfo.path).
  const proofs: Record<string, string[]> = {};
  for (const tool of input.tools) {
    const owner = extensionName(tool.path);
    if (input.files.includes(owner)) (proofs[owner] ??= []).push(`tool:${tool.name}`);
  }
  for (const command of input.commands) {
    const owner = extensionName(command.path);
    if (input.files.includes(owner)) (proofs[owner] ??= []).push(`command:/${command.name}`);
  }
  return {
    pi_version: input.pi_version,
    model: input.model,
    trusted: input.trusted,
    headless: input.headless,
    expected: input.expected,
    present,
    missing,
    files: input.files,
    proofs,
  };
}

// ---------- l'extension : dresser au démarrage, déposer au premier tour ----------

export default function (pi: ExtensionAPI) {
  let inventory: Inventory | null = null;
  let deposited = false;

  // Dresser l'inventaire quand tout est chargé : les outils et commandes des autres
  // extensions sont enregistrés avant session_start.
  pi.on("session_start", async (_event, ctx) => {
    inventory = takeInventory({
      files: listExtensionFiles(ctx.cwd),
      expected: expectedFrom(process.env),
      tools: pi.getAllTools().map((t) => ({ name: t.name, path: t.sourceInfo.path })),
      commands: pi.getCommands().filter((c) => c.source === "extension").map((c) => ({ name: c.name, path: c.sourceInfo.path })),
      pi_version: VERSION,
      model: ctx.model ? `${ctx.model.provider}/${ctx.model.id}` : null,
      trusted: ctx.isProjectTrusted(),
      headless: !ctx.hasUI,
    });
    deposited = false;
    if (ctx.hasUI) {
      const state = inventory.missing.length === 0
        ? `socle ${inventory.present.length}/${inventory.expected.length}`
        : `socle incomplet : ${inventory.missing.join(", ")}`;
      ctx.ui.setStatus("factory-inventory", `▣ ${state}`);
    }
  });

  // Déposer l'entrée au premier tour, pas à session_start : en --mode json, pi ne relaie sur
  // stdout que les événements émis après son abonnement à la session — session_start le
  // précède, agent_start le suit (vérifié sur pi 0.84). Une seule entrée par session.
  pi.on("agent_start", async () => {
    if (deposited || !inventory) return;
    deposited = true;
    pi.appendEntry(ENTRY_TYPE, inventory);
  });

  pi.registerCommand("factory-inventory", {
    description: "Afficher l'inventaire du socle de l'usine pour cette session",
    handler: async (_args, ctx) => {
      if (!ctx.hasUI || !inventory) return;
      const proofs = Object.entries(inventory.proofs).map(([name, list]) => `${name} (${list.join(", ")})`).join(" · ");
      ctx.ui.notify(
        `pi ${inventory.pi_version} · socle ${inventory.present.length}/${inventory.expected.length}` +
          (inventory.missing.length ? ` · manquants : ${inventory.missing.join(", ")}` : "") +
          (proofs ? ` · preuves : ${proofs}` : ""),
        inventory.missing.length ? "warning" : "info",
      );
    },
  });
}

// ---------- la gate du fichier : `bun .pi/extensions/factory-inventory.ts`, zéro jeton ----------

if (import.meta.main) {
  const files = listExtensionFiles(process.cwd());
  const full = takeInventory({
    files, expected: DEFAULT_EXPECTED,
    tools: [{ name: "report_phase", path: join(process.cwd(), EXTENSIONS_DIR, "factory-report.ts") }],
    commands: [{ name: "damage-control", path: join(process.cwd(), EXTENSIONS_DIR, "damage-control.ts") }],
    pi_version: VERSION, model: null, trusted: true, headless: true,
  });
  const renamed = takeInventory({
    files: files.filter((f) => f !== "factory-guard"), expected: DEFAULT_EXPECTED,
    tools: [], commands: [], pi_version: VERSION, model: null, trusted: true, headless: true,
  });
  const custom = expectedFrom({ FACTORY_EXPECTED: "damage-control, factory-obs" });
  const ok = full.missing.length === 0
    && full.proofs["factory-report"]?.includes("tool:report_phase") === true
    && full.proofs["damage-control"]?.includes("command:/damage-control") === true
    && renamed.missing.length === 1 && renamed.missing[0] === "factory-guard"
    && custom.length === 2 && expectedFrom({}).length === DEFAULT_EXPECTED.length;
  console.log(`factory-inventory ${ok ? "OK" : "KO"} — pi ${VERSION}, ${files.length} extensions sur disque, ` +
    `socle ${full.present.length}/${full.expected.length}` +
    (full.missing.length ? ` (manquants : ${full.missing.join(", ")})` : "") +
    `, un renommage détecté : ${renamed.missing.join(", ")}`);
  process.exit(ok ? 0 : 1);
}

Pièce — .pi/tests/socle.test.ts

Le banc d’essai du nœud pi : huit scénarios, un par garantie que le runner attend du socle. Le fichier copie les cinq extensions et les règles du damage control dans un dépôt jetable, pose dans l’environnement ce que le runner y pose (FACTORY_ENVELOPE_SCHEMA, FACTORY_LOOP_*, FACTORY_EXPECTED), monte une session complète sur le fournisseur factice et lit les événements. Il ne modifie aucune pièce posée.

// .pi/tests/socle.test.ts — le socle, testé dans le vrai pipeline de pi, sans réseau ni jeton.
// Le SDK de pi monte une session complète (extensions du socle chargées depuis .pi/extensions/,
// outils intégrés réels, session en mémoire) sur un FOURNISSEUR FACTICE dont on scripte les
// réponses : un appel d'outil, un texte, une enveloppe. Ce que l'on vérifie, ce n'est pas la
// logique pure de chaque extension (les gates `bun .pi/extensions/*.ts` le font déjà), c'est
// leur comportement une fois branchées sur pi : l'événement arrive-t-il, le refus passe-t-il,
// l'entrée est-elle déposée ? C'est aussi le filet contre les montées de version de pi.
//   bun test ./.pi/tests/socle.test.ts     ← zéro jeton, une demi-seconde
import { afterEach, beforeEach, describe, expect, test } from "bun:test";
import { cpSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall, InMemoryCredentialStore } from "@earendil-works/pi-ai";
import type { AgentSessionEvent } from "@earendil-works/pi-coding-agent";
import {
  createAgentSession, DefaultResourceLoader, ModelRuntime, SessionManager, SettingsManager,
} from "@earendil-works/pi-coding-agent";

const REPO = resolve(import.meta.dir, "..", "..");            // la racine de plume-factory
const SOCLE = ["damage-control", "factory-obs", "factory-report", "factory-guard", "factory-inventory"];
const ENVELOPE_SCHEMA = {
  title: "Envelope", type: "object", additionalProperties: true, required: ["status", "summary"],
  properties: {
    status: { type: "string", enum: ["success", "fail"] },
    summary: { type: "string" },
    artifacts: { type: "array", items: { type: "string" } },
  },
};

// ---------- le banc : un dépôt jetable, le socle copié tel quel, l'environnement du runner ----------

let cwd: string;
let savedEnv: Record<string, string | undefined>;

beforeEach(() => {
  cwd = mkdtempSync(join(tmpdir(), "plume-socle-"));
  mkdirSync(join(cwd, ".pi", "extensions"), { recursive: true });
  for (const name of SOCLE) {
    cpSync(join(REPO, ".pi", "extensions", `${name}.ts`), join(cwd, ".pi", "extensions", `${name}.ts`));
  }
  cpSync(join(REPO, ".pi", "damage-control-rules.yaml"), join(cwd, ".pi", "damage-control-rules.yaml"));
  writeFileSync(join(cwd, "Envelope.json"), JSON.stringify(ENVELOPE_SCHEMA));
  // Ce que le runner pose dans l'environnement d'un nœud d'usine (harness.py, roster.py).
  savedEnv = { ...process.env };
  process.env.PI_OFFLINE = "1";
  process.env.FACTORY_ENVELOPE_SCHEMA = join(cwd, "Envelope.json");
  process.env.FACTORY_LOOP_WINDOW = "10";
  process.env.FACTORY_LOOP_THRESHOLD = "3";
  process.env.FACTORY_LIMIT_TURNS = "20";
  delete process.env.FACTORY_EXPECTED;
});

afterEach(() => {
  for (const key of Object.keys(process.env)) if (!(key in savedEnv)) delete process.env[key];
  Object.assign(process.env, savedEnv);
  rmSync(cwd, { recursive: true, force: true });
});

type Step = Parameters<ReturnType<typeof fauxProvider>["setResponses"]>[0][number];

// Une session pi complète sur le fournisseur factice : les réponses sont scriptées, tout le
// reste (extensions, outils, événements) est le vrai pipeline. Rend les événements observés.
async function runSession(responses: Step[], tools = ["read", "bash", "report_phase"]) {
  const faux = fauxProvider();
  faux.setResponses(responses);
  const modelRuntime = await ModelRuntime.create({ credentials: new InMemoryCredentialStore(), modelsPath: null });
  modelRuntime.registerNativeProvider(faux.provider);

  const settingsManager = SettingsManager.inMemory({ compaction: { enabled: false }, retry: { enabled: false } });
  const resourceLoader = new DefaultResourceLoader({ cwd, agentDir: join(cwd, "agent"), settingsManager, noSkills: true, noPromptTemplates: true });
  await resourceLoader.reload();
  const loaded = resourceLoader.getExtensions();
  expect(loaded.errors).toEqual([]);                       // une extension qui jette au chargement = test rouge

  const { session } = await createAgentSession({
    cwd, agentDir: join(cwd, "agent"), model: faux.getModel(), modelRuntime, resourceLoader, settingsManager,
    sessionManager: SessionManager.inMemory(cwd), tools,
  });
  // Comme le mode headless de pi : lier les extensions (c'est ce qui émet session_start —
  // règles du damage control chargées, inventaire dressé), PUIS s'abonner, PUIS prompter.
  await session.bindExtensions({ mode: "json" });
  const events: AgentSessionEvent[] = [];
  session.subscribe((event) => { events.push(event); });
  try {
    await session.prompt("phase de test");
  } finally {
    session.dispose();
  }
  return { events, loaded: loaded.extensions.map((e) => e.path) };
}

const entries = (events: AgentSessionEvent[], customType: string) =>
  events.filter((e) => e.type === "entry_appended" && (e.entry as { customType?: string }).customType === customType)
    .map((e) => ((e as { entry: { data?: unknown } }).entry.data ?? {}) as Record<string, unknown>);
const toolEnds = (events: AgentSessionEvent[], toolName: string) =>
  events.filter((e) => e.type === "tool_execution_end" && (e as { toolName: string }).toolName === toolName) as Array<{
    toolName: string; isError: boolean; result: { content: Array<{ type: string; text?: string }>; details?: unknown };
  }>;

// ---------- les scénarios : un par garantie que le runner attend du socle ----------

describe("factory-inventory — vérifier, pas croire", () => {
  test("les cinq extensions du socle se chargent et l'inventaire est déposé au premier tour", async () => {
    const { events, loaded } = await runSession([fauxAssistantMessage("rien à faire")]);
    expect(loaded.map((p) => p.split("/").pop())).toEqual(expect.arrayContaining(SOCLE.map((n) => `${n}.ts`)));
    const [inventory] = entries(events, "factory-inventory");
    expect(inventory).toBeDefined();
    expect(inventory.missing).toEqual([]);
    expect(inventory.present).toEqual(SOCLE);
    expect(inventory.headless).toBe(true);
    expect(inventory.model).toBe("faux/faux-1");
    expect((inventory.proofs as Record<string, string[]>)["factory-report"]).toContain("tool:report_phase");
  });

  test("un nom attendu absent apparaît dans missing — le runner refusera la phase", async () => {
    process.env.FACTORY_EXPECTED = "damage-control,factory-guard,factory-renamed";
    const { events } = await runSession([fauxAssistantMessage("rien à faire")]);
    const [inventory] = entries(events, "factory-inventory");
    expect(inventory.missing).toEqual(["factory-renamed"]);
    expect(inventory.present).toEqual(["damage-control", "factory-guard"]);
  });
});

describe("damage-control — décider avant le geste", () => {
  test("rm -rf est refusé avec son motif, sans être exécuté", async () => {
    const { events } = await runSession([
      fauxAssistantMessage([fauxToolCall("bash", { command: "rm -rf ." })], { stopReason: "toolUse" }),
      fauxAssistantMessage("je n'insiste pas"),
    ]);
    const [end] = toolEnds(events, "bash");
    expect(end.isError).toBe(true);
    expect(end.result.content[0]?.text).toContain("rm récursif ou forcé");
    expect(entries(events, "damage-control")[0]?.action).toBe("blocked");
  });
});

describe("factory-guard — disposer pendant la phase", () => {
  test("le même appel répété est coupé au seuil, sans appel LLM de plus", async () => {
    const same = () => fauxAssistantMessage([fauxToolCall("bash", { command: "echo encore" })], { stopReason: "toolUse" });
    const { events } = await runSession([same(), same(), same(), same(), fauxAssistantMessage("jamais atteint")]);
    const ends = toolEnds(events, "bash");
    expect(ends.length).toBe(3);                                       // deux exécutés, le troisième refusé
    expect(ends[2].isError).toBe(true);
    expect(ends[2].result.content[0]?.text).toContain("[boucle]");
    const [guard] = entries(events, "factory-guard");
    expect(guard.kind).toBe("boucle");
    const assistantEnds = events.filter((e) => e.type === "message_end" && (e as { message: { role: string } }).message.role === "assistant");
    expect(assistantEnds.length).toBe(3);                              // terminate : pas de 4e réponse du modèle
  });

  test("un secret dans un résultat d'outil est masqué avant de partir au fournisseur", async () => {
    const { events } = await runSession([
      fauxAssistantMessage([fauxToolCall("bash", { command: "echo OPENROUTER_API_KEY=sk-or-v1-abcdefghijklmnopqrstuvwxyz0123456789" })], { stopReason: "toolUse" }),
      fauxAssistantMessage("vu"),
    ]);
    const toolResults = events.filter((e) => e.type === "message_end" && (e as { message: { role: string } }).message.role === "toolResult") as Array<{
      message: { content: Array<{ type: string; text?: string }> };
    }>;
    const text = toolResults[0]?.message.content[0]?.text ?? "";
    expect(text).toContain("OPENROUTER_API_KEY=«redacted»");
    expect(text).not.toContain("sk-or-v1-");
  });
});

describe("factory-report — l'enveloppe est une porte, pas une convention", () => {
  test("une enveloppe conforme arrive par tool_execution_end et termine le tour", async () => {
    const { events } = await runSession([
      fauxAssistantMessage([fauxToolCall("report_phase", { status: "success", summary: "fini", artifacts: [] })], { stopReason: "toolUse" }),
      fauxAssistantMessage("jamais atteint"),
    ]);
    const [end] = toolEnds(events, "report_phase");
    expect(end.isError).toBe(false);
    expect(end.result.details).toEqual({ status: "success", summary: "fini", artifacts: [] });
    expect(events.filter((e) => e.type === "message_end" && (e as { message: { role: string } }).message.role === "assistant").length).toBe(1);
  });

  test("un champ requis manquant est refusé au modèle, avec le motif", async () => {
    const { events } = await runSession([
      fauxAssistantMessage([fauxToolCall("report_phase", { summary: "sans statut" })], { stopReason: "toolUse" }),
      fauxAssistantMessage("je corrige"),
    ]);
    const [end] = toolEnds(events, "report_phase");
    expect(end.isError).toBe(true);
    expect(end.result.content[0]?.text).toMatch(/status/);
  });

  test("une réponse en prose ne produit aucune enveloppe — le runner verra l'absence", async () => {
    const { events } = await runSession([fauxAssistantMessage([fauxText('{"status": "success", "summary": "dans la prose"}')])]);
    expect(toolEnds(events, "report_phase")).toEqual([]);
  });
});

Pièce — adws/adw_modules/harness.py

Cette version remplace celle du chapitre A6. Tout ce qui existait reste : mêmes dataclasses, même run(), même argv v4, même adaptateur Claude Code. S’ajoutent la liste SOCLE et son transport FACTORY_EXPECTED, le champ socle de HarnessRequest (() = aucune exigence), la lecture de l’entrée factory-inventory dans le flux, et surtout _judge : le verdict du runner, sorti de _run_pi pour devenir une fonction pure (le socle d’abord, la porte ensuite) que la gate du module éprouve sur cinq flux synthétiques. 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.
Il pose la liste des extensions attendues dans l'environnement du noeud pi
(FACTORY_EXPECTED), lit l'inventaire que .pi/extensions/factory-inventory.ts
depose dans le flux (entry_appended, customType factory-inventory) et REFUSE
la phase si l'inventaire manque ou s'il declare un manquant — fail-closed :
un noeud sans damage control, sans trace ou sans coupe-circuit ne tourne
pas, meme s'il aurait rendu une enveloppe. Le verdict est une fonction pure
(_judge), testee a sec par la gate. L'API de run() ne bouge pas : vos ADW
des chapitres 8 a 27 tournent tels quels.
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import uuid
from dataclasses import dataclass
from 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")

# La route par defaut de l'usine : le fournisseur passerelle integre de pi.
# Une seule cle (OPENROUTER_API_KEY) sert tous les moteurs du roster.
# Passer par les API directes des fournisseurs : GATEWAY = "" — et a vous
# de fournir une cle par fournisseur dans l'environnement.
GATEWAY = "openrouter"

# La variable que lit .pi/extensions/factory-report.ts pour construire
# l'outil report_phase : le chemin du schema JSON ecrit par ce module.
SCHEMA_ENV = "FACTORY_ENVELOPE_SCHEMA"

# Le socle : les extensions pi que TOUT noeud d'usine doit avoir chargees.
# Le runner les nomme (FACTORY_EXPECTED), factory-inventory.ts les compare a
# ce qu'il trouve et depose l'inventaire dans le flux (INVENTORY_ENTRY). Le
# poste (footer, chaine) n'en fait pas partie : il ne se charge qu'en TUI.
SOCLE = ("damage-control", "factory-obs", "factory-report", "factory-guard",
         "factory-inventory")
EXPECTED_ENV = "FACTORY_EXPECTED"
INVENTORY_ENTRY = "factory-inventory"

# Le dialecte Claude Code : outils avec majuscules, et pas d'outil ls ni find
# dedies — Bash et Glob les couvrent. L'adaptateur absorbe l'asymetrie.
CLAUDE_TOOLS = {"read": "Read", "bash": "Bash", "edit": "Edit", "write": "Write",
                "grep": "Grep", "find": "Glob", "ls": "Bash"}

# L'echelle de reflexion de pi, traduite en budget de tokens pour Claude Code
# (variable d'environnement MAX_THINKING_TOKENS).
THINKING_TOKENS = {"off": 0, "minimal": 1024, "low": 4096, "medium": 8192,
                   "high": 16384, "xhigh": 24576, "max": 32000}


# Les cles que .env a fournies — et que l'environnement reel ne portait pas.
# C'est le COFFRE (17bis) : ce que le port retire du noeud quand la requete
# porte un profil, pour n'y remettre que ce que le profil nomme.
ENV_FILE_KEYS: set[str] = set()


def _load_env(path: str | Path = ".env") -> None:
    """Charge .env dans l'environnement du process — une fois, au chargement.

    Ni pi ni claude ne lisent .env d'eux-memes : sans ce chargement, la cle
    de la passerelle n'atteindrait jamais les agents. Une variable deja
    presente dans l'environnement reel gagne toujours — un export de session
    ou un secret de CI ne sont jamais ecrases — et n'entre pas dans le coffre :
    elle est a vous, pas a .env.
    """
    env_file = Path(path)
    if not env_file.is_file():
        return
    for line in env_file.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        if key not in os.environ:
            os.environ[key] = value.strip().strip("'\"")
            ENV_FILE_KEYS.add(key)


_load_env()


def node_environment(base: dict[str, str], extra_env: dict[str, str] | None,
                     auth: dict[str, str] | None, vault: set[str]) -> dict[str, str]:
    """L'environnement d'un noeud — pure, donc testable a sec.

    Sans profil (auth=None) : l'heritage du ch. 15, tout .env compris. Avec
    profil : .env est un coffre — chaque cle qu'il a fournie est retiree,
    puis le profil depose les siennes, avec leurs valeurs. Ce que le port ou
    l'ADW pose lui-meme pour la phase passe toujours.
    """
    environment = dict(base)
    if auth is not None:
        for key in vault:
            environment.pop(key, None)
        environment.update(auth)
    if extra_env:
        environment.update(extra_env)
    return environment


class HarnessError(RuntimeError):
    """Le harnais n'a pas rendu de reponse exploitable.

    Depuis A6, l'erreur porte la session qu'elle a interrompue : l'uuid est
    choisi avant l'appel, donc une tentative tuee par le mur de temps a quand
    meme une session — la reprise la poursuit au lieu de repartir a froid.
    """

    def __init__(self, message: str, session_id: str | None = None) -> None:
        super().__init__(message)
        self.session_id = session_id


@dataclass(frozen=True)
class HarnessRequest:
    """Ce que l'usine a le droit de demander a un agent — rien de plus."""
    prompt: str
    session_id: str | None = None    # None = nouvelle session
    cwd: str = "."
    timeout: int = 600               # un agent muet ne bloque pas l'usine
    model: str | None = None         # l'id du REGISTRE, tel quel dans le roster
    thinking: str | None = None      # off..max — None = defaut du harnais
    tools: tuple[str, ...] = ()      # allowlist du roster — () = outils par defaut
    schema: dict | None = None       # schema JSON de la porte — None = Envelope de base
    socle: tuple[str, ...] = SOCLE   # extensions exigees du noeud pi — () = aucune exigence
    auth: dict[str, str] | None = None # le profil resolu (17bis) — None = l'heritage du ch. 15
    direct: bool = False             # route directe (17bis) : le modele est deja une route pi


@dataclass(frozen=True)
class HarnessResult:
    """Ce qu'un agent rend a l'usine — quel que soit le harnais."""
    text: str                        # la derniere reponse de l'agent (ou l'enveloppe en JSON)
    session_id: str                  # de quoi poursuivre la MEME session
    cost_usd: float                  # 0.0 si le harnais ne rapporte pas le cout
    returncode: int
    tokens: int = 0                  # jetons factures sur cet appel, 0 si non rapportes
    stop_reason: str = "stop"        # stop | length | toolUse | error | aborted
    envelope: dict | None = None     # l'enveloppe rendue par la porte typee, sinon None


def run(harness: str, request: HarnessRequest) -> HarnessResult:
    """L'unique porte d'entree vers les agents : choisit l'adaptateur, normalise."""
    try:
        adapter = ADAPTERS[harness]
    except KeyError:
        raise HarnessError(
            f"harnais inconnu {harness!r} — disponibles : {sorted(ADAPTERS)}"
        ) from None
    return adapter(request)


def _route(model: str, direct: bool = False) -> str:
    """L'id du registre devient une route pi : prefixe du fournisseur passerelle.

    Le roster parle le langage du registre (z-ai/glm-5.3) — la meme chaine
    que verifie la jauge du ch. 14. Le prefixe est un detail de dialecte :
    il vit ici, jamais dans le YAML ni dans vos scripts. Sous un profil a
    route directe (17bis), l'identifiant est deja une route pi (zai/glm-5.3,
    kimi-coding/k3) et part tel quel — la route est dans le profil, pas dans
    le nom : deepseek/... est un auteur OpenRouter ET un fournisseur natif.
    """
    if direct or not GATEWAY or model.startswith(GATEWAY + "/"):
        return model
    return f"{GATEWAY}/{model}"


def _spawn(cmd: list[str], request: HarnessRequest,
           extra_env: dict[str, str] | None = None,
           stdin_text: str | None = None) -> subprocess.CompletedProcess[str]:
    # Resoudre l'executable via le PATH : sous Windows, les harnais sont des
    # shims (pi.cmd, claude.cmd) que CreateProcess ne trouve pas par leur nom
    # court — shutil.which respecte PATHEXT et regle les deux mondes d'un coup.
    executable = shutil.which(cmd[0])
    if executable is None:
        raise HarnessError(f"{cmd[0]!r} introuvable dans le PATH — "
                           "le harnais est-il installe ?")
    environment = node_environment(dict(os.environ), extra_env, request.auth, ENV_FILE_KEYS)
    # Deux modes d'entree, jamais d'entre-deux :
    # - stdin_text=None : le prompt voyage dans argv, et stdin est ferme
    #   (DEVNULL) — un enfant qui herite de notre stdin peut attendre
    #   indefiniment une entree qui ne viendra jamais : echec silencieux,
    #   0 % CPU, aucune sortie.
    # - stdin_text : le prompt voyage par stdin, puis le tube est referme.
    #   Indispensable quand le harnais est un shim .cmd Windows : cmd.exe
    #   tronque un argument a la premiere nouvelle ligne, et les asks de
    #   l'usine (brief + mission + contrat) sont multi-lignes.
    io = ({"input": stdin_text} if stdin_text is not None
          else {"stdin": subprocess.DEVNULL})
    # Encodage explicite : les harnais emettent de l'UTF-8, mais text=True
    # seul decode avec la locale — cp1252 sous Windows, qui mutile tirets
    # et accents. Vaut pour la sortie ET pour le prompt ecrit sur stdin.
    try:
        return subprocess.run([executable, *cmd[1:]], **io,
                              capture_output=True, text=True,
                              encoding="utf-8", errors="replace",
                              env=environment,
                              timeout=request.timeout, cwd=request.cwd)
    except subprocess.TimeoutExpired:
        # Le timeout aussi sort par la porte normalisee : une seule exception.
        raise HarnessError(f"harnais muet apres {request.timeout} s : {cmd[0]}") from None


def _text_of(message: dict) -> str:
    """Concatene les blocs de texte d'un message pi."""
    return "".join(part.get("text", "") for part in message.get("content", []) or []
                   if isinstance(part, dict) and part.get("type") == "text")


# ---------- l'adaptateur pi, v4 : trois 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_argv(request: HarnessRequest, session_id: str) -> list[str]:
    """La ligne de commande pi — pure, donc testable a sec (voir la gate)."""
    cmd = ["pi", "-p", "--mode", "json",
           # --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:
        cmd += ["--model", _route(request.model, request.direct)]
    if request.thinking:
        cmd += ["--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.
        cmd += ["--tools", ",".join((*request.tools, envelopes.REPORT_TOOL))]
    # `--` ferme les options : plus rien de positionnel — le prompt arrive par
    # stdin, quel que soit son premier caractere ou son nombre de lignes.
    cmd.append("--")
    return cmd


@dataclass
class PiReading:
    """Ce que le runner retient du flux JSON de pi : quatre chiffres, une enveloppe."""
    text: str = ""
    cost_usd: float = 0.0
    tokens: int = 0
    stop_reason: str = "stop"
    observed_model: str | None = None
    envelope: dict | None = None
    inventory: dict | None = None    # ce que le socle declare de lui-meme (A8)


def _read_pi_stream(lines: list[str]) -> PiReading:
    """Lit le flux ligne a ligne — pure, donc testable a sec.

    message_end (assistant) : texte, cout, jetons, stopReason, modele observe.
    tool_execution_end de report_phase : l'enveloppe, deja validee par schema
    cote pi. entry_appended de factory-inventory : la preuve du socle. Le
    dernier texte gagne ; les couts et jetons s'additionnent.
    """
    reading = PiReading()
    for line in lines:
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        kind = event.get("type")
        if kind == "message_end":
            message = event.get("message") or {}
            if message.get("role") != "assistant":
                continue
            reading.text = _text_of(message) or reading.text
            usage = message.get("usage") or {}
            reading.cost_usd += (usage.get("cost") or {}).get("total") or 0.0
            reading.tokens += int(usage.get("totalTokens") or 0)
            reading.stop_reason = message.get("stopReason") or reading.stop_reason
            if reading.observed_model is None and message.get("model"):
                reading.observed_model = f"{message.get('provider')}/{message.get('model')}"
        elif (kind == "tool_execution_end"
              and event.get("toolName") == envelopes.REPORT_TOOL
              and not event.get("isError")):
            details = (event.get("result") or {}).get("details")
            if isinstance(details, dict):
                reading.envelope = details
        elif kind == "entry_appended":
            entry = event.get("entry") or {}
            if entry.get("customType") == INVENTORY_ENTRY and isinstance(entry.get("data"), dict):
                reading.inventory = entry["data"]
    return reading


def _judge(reading: PiReading, request: HarnessRequest, session_id: str,
           returncode: int, stderr: str) -> HarnessResult:
    """Le verdict du runner sur un flux pi — pure, donc testable a sec.

    Dans l'ordre : le socle d'abord (sans preuve, pas de phase — meme avec
    une enveloppe) ; puis la porte typee ; puis les arrets qui ne sont pas
    une reponse (error, aborted — pi rend 0 en --mode json, le code retour
    ne dit rien) ; enfin le texte, pour le chemin de secours ; et le modele
    observe, qui doit etre celui du roster.
    """
    evidence = stderr.strip()[-400:]
    if request.socle:
        if reading.inventory is None:
            raise HarnessError(
                "socle pi non charge : aucun inventaire dans le flux — .pi/ non approuve, "
                "factory-inventory.ts absent, ou pi n'a pas demarre de tour"
                + (f" ({evidence})" if evidence else ""), session_id)
        missing = [name for name in request.socle
                   if name not in (reading.inventory.get("present") or [])]
        if missing:
            raise HarnessError(f"socle pi incomplet : manquants {missing} — "
                               f"pi {reading.inventory.get('pi_version', '?')} n'a charge que "
                               f"{reading.inventory.get('present')}", session_id)
    if reading.envelope is not None:
        text = json.dumps(reading.envelope, ensure_ascii=False)
    elif reading.stop_reason in ("error", "aborted"):
        raise HarnessError(f"pi s'est arrete sur {reading.stop_reason} : "
                           f"{evidence or reading.text[-400:]}", session_id)
    elif returncode != 0 and not reading.text:
        raise HarnessError(f"pi a rendu {returncode} : {evidence}", session_id)
    else:
        text = reading.text
    # Le modele observe doit etre celui du roster : pi resout un motif par
    # sous-chaine, et une facture sur le mauvais moteur n'est pas un run vert.
    if request.model and reading.observed_model \
            and reading.observed_model != _route(request.model, request.direct):
        raise HarnessError(f"modele observe {reading.observed_model!r} "
                           f"≠ modele demande {_route(request.model, request.direct)!r}", session_id)
    return HarnessResult(text=text, session_id=session_id, cost_usd=reading.cost_usd,
                         returncode=returncode, tokens=reading.tokens,
                         stop_reason=reading.stop_reason, envelope=reading.envelope)


def _run_pi(request: HarnessRequest) -> HarnessResult:
    # pi : c'est VOUS qui nommez la session — et vous la nommez AVANT l'appel.
    # Meme id + meme dossier = meme contexte ; une erreur porte cet id.
    session_id = request.session_id or str(uuid.uuid4())
    SESSION_DIR.mkdir(parents=True, exist_ok=True)
    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)}
    try:
        proc = _spawn(_pi_argv(request, session_id), request,
                      extra_env=extra_env, stdin_text=request.prompt)
    except HarnessError as error:
        error.session_id = session_id
        raise
    reading = _read_pi_stream(proc.stdout.splitlines())
    return _judge(reading, request, session_id, proc.returncode, proc.stderr)


def _claude_evidence(stdout: str, stderr: str) -> str:
    """Le motif d'un echec claude — pure. Le JSON de stdout d'abord, stderr ensuite.

    En --output-format json, claude ecrit son erreur dans l'objet de stdout
    (result, error) et n'envoie sur stderr que des avertissements (« Ignoring
    N permissions.allow entries… ») : lire stderr d'abord masquerait la vraie
    cause — une cle absente, un depot non approuve, un modele inconnu.
    """
    try:
        payload = json.loads(stdout)
        for key in ("result", "error", "message"):
            if isinstance(payload, dict) and payload.get(key):
                return str(payload[key]).strip()[-400:]
    except (json.JSONDecodeError, TypeError):
        pass
    lines = [line for line in stderr.strip().splitlines() if not line.startswith("Ignoring ")]
    return ("\n".join(lines).strip() or stderr.strip() or stdout.strip())[-400:]


def _run_claude(request: HarnessRequest) -> HarnessResult:
    # Claude Code : c'est LUI qui nomme la session. On la poursuit en rendant
    # son session_id via --resume. Pas d'outil terminal type de ce cote :
    # l'enveloppe reste une convention de texte, parse() la lit en secours.
    #
    # Regime par defaut : natif Anthropic — sa propre authentification, des
    # modeles Anthropic. Le pointer sur la passerelle est possible (trois
    # variables : ANTHROPIC_BASE_URL, ANTHROPIC_AUTH_TOKEN, et
    # ANTHROPIC_API_KEY explicitement vide), mais la compatibilite n'est
    # garantie que sur les modeles Anthropic : l'adaptateur polyglotte de
    # l'usine reste pi, et ce choix-la appartient a votre environnement,
    # pas a cet adaptateur.
    cmd = ["claude", "-p", "--output-format", "json"]
    if request.model:
        # Le dialecte claude ignore le fournisseur : provider/id -> id.
        cmd += ["--model", request.model.split("/", 1)[-1]]
    if request.tools:
        # Traduire puis dedoublonner en gardant l'ordre : bash et ls donnent
        # tous deux Bash, inutile de le declarer deux fois.
        allowed = list(dict.fromkeys(
            CLAUDE_TOOLS[tool] for tool in request.tools if tool in CLAUDE_TOOLS))
        cmd += ["--allowedTools", ",".join(allowed)]
    if request.session_id:
        cmd += ["--resume", request.session_id]
    # L'echelle de reflexion devient un budget de tokens — l'asymetrie reste
    # dans l'adaptateur, le roster n'en sait rien.
    extra_env = ({"MAX_THINKING_TOKENS": str(THINKING_TOKENS[request.thinking])}
                 if request.thinking in THINKING_TOKENS else None)
    # Le prompt part par stdin, PAS dans argv : c'est un mode documente de
    # claude -p, et le seul qui survive aux shims .cmd de Windows.
    proc = _spawn(cmd, request, extra_env, stdin_text=request.prompt)

    if proc.returncode != 0:
        # Le JSON de stdout d'abord, stderr ensuite : les avertissements de
        # claude (« Ignoring … ») ne doivent pas masquer la vraie cause.
        evidence = _claude_evidence(proc.stdout, proc.stderr)
        raise HarnessError(f"claude a rendu {proc.returncode} : {evidence}",
                           request.session_id)
    try:
        payload = json.loads(proc.stdout)
    except json.JSONDecodeError:
        raise HarnessError("claude n'a pas rendu l'objet JSON attendu "
                           "(--output-format json)", request.session_id) from None
    usage = payload.get("usage") or {}
    return HarnessResult(text=str(payload.get("result", "")),
                         session_id=str(payload.get("session_id", "")),
                         cost_usd=float(payload.get("total_cost_usd") or 0.0),
                         returncode=proc.returncode,
                         tokens=int(usage.get("input_tokens") or 0)
                         + int(usage.get("output_tokens") or 0),
                         stop_reason="error" if payload.get("is_error") else "stop")


# Le registre des adaptateurs. Un harnais de plus = une fonction + une ligne.
ADAPTERS = {"pi": _run_pi, "claude": _run_claude}


if __name__ == "__main__":
    # La gate du module — zero token, sans pi : les trois fonctions pures.
    # Lancer depuis la racine : uv run python -m adws.adw_modules.harness
    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

    def inventory_line(present: list[str]) -> str:
        return json.dumps({"type": "entry_appended", "entry": {
            "type": "custom", "customType": INVENTORY_ENTRY,
            "data": {"pi_version": "0.84.4", "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"


    # 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"
          " ; profil 17bis : route directe hors passerelle, coffre .env ferme sous profil, ouvert sans")

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 cinquième renomme une pièce du socle (avec git mv, qui parle les deux shells), la sixième lance un scout qui doit être refusé, la septième remet la pièce en place, la dernière relance le scout pour le voir passer.

bun add -d @earendil-works/pi-coding-agent @earendil-works/pi-ai
bun .pi/extensions/factory-inventory.ts
uv run python -m adws.adw_modules.harness
bun test ./.pi/tests/socle.test.ts
git mv .pi/extensions/factory-guard.ts .pi/extensions/factory-guard.ts.off
uv run adws/adw_scout.py "Liste les fichiers de tests de apps/plume et ce que chacun verifie." --retries 0
git mv .pi/extensions/factory-guard.ts.off .pi/extensions/factory-guard.ts
uv run adws/adw_scout.py "Liste les fichiers de tests de apps/plume et ce que chacun verifie."

Résultat attendu : factory-inventory OK — pi 0.84.4, 7 extensions sur disque, socle 5/5, un renommage détecté : factory-guard (la version est celle de votre node_modules), puis harness OK — … verdict v5 : socle prouve (5 extensions), refus sans inventaire, refus sur manquant …, puis 8 pass, 0 fail en une demi-seconde. Le scout au socle amputé s’arrête, sur la sortie d’erreur, avec echec — socle pi incomplet : manquants ['factory-guard'] — pi 0.84.x n'a charge que [...] et un code de retour non nul, session conservée dans run.sessions. Le dernier scout rend son enveloppe verte comme au chapitre 11. Coût : les quatre gates à sec ne dépensent rien (bun add télécharge une fois quelques dizaines de mégaoctets), le scout refusé coûte moins d’un centime sur le modèle léger, ce qui est le prix d’un refus après la phase, et le scout complet, un centime ou moins en 30 à 60 s.

---