Annexe — Harnais pi Chapitre A8 / 43

Vérifier le socle, pas y croire

Le runner exige la preuve que le socle pi est chargé et refuse la phase sinon. Puis des tests qui exercent les extensions dans le vrai pipeline de pi, sur un fournisseur factice, sans réseau ni jeton. Comparé à Claude Code.

Depuis A6, le runner lance pi avec --approve. Mais rien ne prouve, run après run, que le damage control, la trace, la porte typée et le coupe-circuit sont chargés. Un fichier renommé, un pi update qui casse un import ou une boîte livrée sans .pi/ suffisent. Le nœud tourne alors nu, et rend parfois quand même une enveloppe.

Ce chapitre fait refuser au runner une phase dont le socle n’est pas prouvé. Puis il teste les extensions dans le vrai pipeline de pi, sur un fournisseur factice, sans réseau ni jeton.

Inventaire fail-closed du nœud pi

Prenez un modèle factice qui appelle bash rm -rf adws, puis rend une enveloppe success. Avec --approve, le damage control refuse le rm. Avec --no-approve, aucune extension ne se charge. Le rm s’exécute, le dossier de l’usine disparaît, et pi sort en code 0 avec une enveloppe valide. Le runner v4 compte ce run comme vert.

Un run sans socle ressemble à un run sain qui n’avait rien à refuser. Le runner v5 exige donc une preuve positive. C’est le principe fail-closed d’A3, appliqué au socle.

L’inventaire

La preuve est une entrée typée que factory-inventory.ts dépose dans le flux de pi à chaque session. Voici ce que _read_pi_stream lit sur stdout (données abrégées) :

{"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"]}}}}

C’est une enveloppe au sens du livre. Du code l’écrit, du code la lit, le modèle ne la voit jamais.

Le runner nomme ce qu’il attend. La liste vit dans harness.py (SOCLE) et part au nœud par la variable FACTORY_EXPECTED. socle=() n’exige rien, pour un run de test.
L’inventaire se dresse à session_start et se dépose à agent_start. En --mode json, pi ne relaie sur stdout que les événements émis après son abonnement, et session_start arrive avant. Le runner ne verrait jamais une entrée déposée là.
La provenance sert de preuve en plus. pi.getAllTools() et pi.getCommands() portent sourceInfo.path, le fichier qui a enregistré chaque outil ou commande. L’inventaire les range dans proofs.

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)

Le code 1 ne couvre qu’une extension chargée qui échoue. Une extension jamais chargée ne fait aucun bruit. Seul l’inventaire distingue les deux.

Le verdict

harness._judge juge le socle avant la porte. Sans inventaire, ou avec un manquant, la phase est refusée, même avec une enveloppe valide. La session est conservée.

# 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)

Le runner lit le flux quand pi a rendu la main. La phase a donc déjà coûté ses jetons. L’inventaire garantit seulement qu’un nœud nu ne passe jamais pour un run vert. La vérification avant tout run viendra avec adw_doctor en A10.

Côté Claude Code : cet inventaire existe d’office. En --output-format stream-json, l’événement system/init liste le modèle, les outils, les serveurs MCP et les plugins chargés. Ses champs plugin_errors et mcp_server_errors servent à faire échouer une CI quand un plugin ou un serveur ne se charge pas. En interactif, /hooks liste les hooks actifs de la session.

Tester les extensions hors réseau avec le SDK

Comment vérifier que le coupe-circuit coupe au troisième appel identique ? Avec un vrai modèle, il faut d’abord qu’il veuille bien boucler. Un fournisseur factice boucle sur commande.

Le SDK de pi monte une session complète en mémoire. Extensions, outils et événements sont ceux de la production. Seul le modèle est remplacé par une file de réponses scriptées. Les huit scénarios du jour passent sous bun test en une demi-seconde. Sur un vrai modèle, ils prendraient quelques minutes, avec un résultat variable.

Le banc

// .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");

Le fournisseur factice est un vrai fournisseur. fauxProvider() vient de @earendil-works/pi-ai, le paquet que pi utilise lui-même. Ses réponses se scriptent avec fauxAssistantMessage, fauxText, fauxToolCall(nom, args) et un stopReason. L’usage est estimé (un jeton pour quatre caractères), ce qui exerce aussi le compteur de coût d’A7.
Rien ne vient de votre poste. Clés, réglages et session restent en mémoire, et agentDir pointe hors de ~/.pi pour écarter vos extensions personnelles.
L’ordre compte : bindExtensions, puis subscribe, puis prompt. bindExtensions émet session_start. Sans lui, le damage control n’a aucune règle et le rm -rf passe.
Les outils sont réels, donc le dépôt doit être jetable. Le test copie le socle dans un dossier temporaire, y pointe cwd et le supprime après.

Ce que le test vérifie

La logique pure de chaque extension a déjà sa gate (bun .pi/extensions/nom.ts, A3 à A7). Le test SDK vérifie le branchement de chacune sur les événements de pi. Si une montée de version de pi renomme un événement, c’est le test qui casse, pas un run.

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é

Lancer la batterie

Le ./ compte, car bun test ignore les dossiers cachés quand il cherche les tests.

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

Piège courant : les tests exercent le SDK de node_modules, pas le binaire pi que lance le runner. Les deux peuvent diverger d’une version. Leur épinglage viendra en A10.

Côté Claude Code : un hook est une commande shell qui reçoit un JSON sur stdin et bloque l’appel d’outil en sortant en code 2. La doc conseille de le tester en lui envoyant un JSON d’exemple par un tube, l’équivalent de la gate de fichier. L’Agent SDK, en Python et en TypeScript, donne les mêmes outils et la même boucle que Claude Code.

pi et Claude Code côte à côte

	pi	Claude Code
Preuve de ce qui est chargé	à construire (`factory-inventory.ts`)	événement `system/init` en `stream-json`
Échec de chargement signalé	extension qui jette : code 1. Fichier ignoré : rien	`plugin_errors`, `mcp_server_errors`
Voir les protections actives	`/factory-inventory` (pièce du jour)	`/hooks`
Tester une protection seule	`bun .pi/extensions/nom.ts`	un JSON d’exemple envoyé au hook
Piloter la boucle depuis du code	SDK de pi (`createAgentSession`)	Agent SDK (Python, TypeScript)

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

factory-inventory.ts ferme le socle pi, après damage-control.ts, factory-obs.ts, factory-report.ts et factory-guard.ts. .pi/tests/ devient le banc d’essai du nœud. La couture ne bouge pas. Le runner ne suppose plus son nœud protégé, il le vérifie, pour zéro jeton. Un refus coûte la phase qui l’a révélé. Le miroir bloquant de factory-obs.ts passera côté runner en A9, et l’extension d’injection de contexte annoncée en A6 n’est pas encore posée.

Travaux pratiques — la pièce du jour

Trois fichiers à poser dans plume-factory. Prérequis : Bun (chapitre 2) et pi (A1).

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

Trois fonctions pures exportées (expectedFrom, listExtensionFiles, takeInventory), l’extension qui dresse à session_start et dépose à agent_start, une commande /factory-inventory pour le poste, et la gate du fichier.

// .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`

Huit scénarios, un par garantie que le runner attend du socle. Le fichier pose dans l’environnement ce que le runner y pose (FACTORY_ENVELOPE_SCHEMA, FACTORY_LOOP_*, FACTORY_EXPECTED).

// .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 l’existant reste, profil 17bis compris. S’ajoutent SOCLE, FACTORY_EXPECTED, le champ socle de HarnessRequest, la lecture de l’inventaire, et _judge, verdict devenu fonction pure que la gate éprouve sur cinq flux.

"""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, identique dans bash et PowerShell. Les quatre dernières lignes renomment factory-guard.ts, lancent un scout qui doit être refusé, remettent le fichier et relancent le scout.

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) ;
harness OK — … verdict v5 : socle prouve (5 extensions), refus sans inventaire, refus sur manquant … ;
8 pass, 0 fail en une demi-seconde ;
le scout amputé s’arrête sur echec — socle pi incomplet : manquants ['factory-guard'] — pi 0.84.x n'a charge que [...], avec un code non nul et la session conservée dans run.sessions ;
le dernier scout rend son enveloppe verte, comme au chapitre 11.

Coût : rien pour les quatre premières lignes, moins d’un centime pour le scout refusé, un centime ou moins pour le scout complet.

Quiz — teste tes connaissances

Annexe — Harnais pi 7 questions Objectif : 5/7 minimum

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.