Annexe — Harnais pi Chapitre A6 / 43

L'enveloppe devient une porte typée

Un outil terminal report_phase dont le schéma vient de l'usine, un adaptateur pi v4 qui vérifie au lieu de croire, et l'injection déterministe du contexte de phase.

Au chapitre 9, parse() cherche l’enveloppe entre le premier { et le dernier } de la réponse. Un builder cite const cfg = { theme: "dark" } avant l’objet final, et l’extraction échoue. L’usine paie une reprise pour un travail déjà fini. Même échec avec un modèle ouvert qui écrit son appel d’outil en texte (<tool_call>).

Ce chapitre remplace cette convention par une porte typée : un appel d’outil validé par un schéma que l’usine écrit. La pièce du jour fait évoluer envelopes.py et harness.py, et ajoute au socle pi l’extension factory-report.ts.

Un outil terminal pour rendre l’enveloppe

En fin de phase, le modèle n’écrit plus son JSON. Il appelle l’outil report_phase, enregistré par une extension pi, avec les champs de l’enveloppe en arguments. Ensuite :

pi valide les arguments contre le schéma, avant d’exécuter l’outil. Un appel non conforme est refusé avec un motif exact, et le modèle corrige dans le même tour.
L’outil rend terminate: true. C’est un outil terminal : le tour s’arrête là.
Le runner lit l’enveloppe dans l’événement tool_execution_end du flux JSON.

Le schéma vient du type Python

envelopes.schema(cls) transforme la dataclass en schéma JSON : types, champs requis, enum de status, et phrases ask comme descriptions. Avant la phase, l’adaptateur pi écrit ce schéma dans adws/adw_data/schemas/ et passe son chemin par FACTORY_ENVELOPE_SCHEMA. L’extension le lit au chargement. Aucun champ n’est recopié en TypeScript.

Le fichier produit pour l’enveloppe de base :

{
  "title": "Envelope",
  "type": "object",
  "properties": {
    "status": { "type": "string", "description": "'success' ou 'fail'", "enum": ["success", "fail"] },
    "summary": { "type": "string", "description": "une phrase : ce que tu as fait ou trouve" },
    "artifacts": { "type": "array", "items": { "type": "string" },
                   "description": "chemins des fichiers ecrits pour la suite, [] sinon" },
    "notes_for_next_agent": { "type": "string", "description": "ce que l'agent suivant doit savoir, '' sinon" }
  },
  "required": ["status"],
  "additionalProperties": true
}

additionalProperties: true est voulu. Vos ADW passent leur sous-type à parse(), pas au port. Des champs comme changed_files traversent donc la porte de base sans être tronqués. Le champ HarnessRequest.schema permettra une porte plus stricte au chapitre A7.

La forme à la porte, le verdict au code

pi vérifie la forme. Puis parse() reçoit le dict et juge les invariants de check(), par exemple qu’un build réussi déclare au moins un fichier.

Le schéma ne garantit pas la vérité. Un builder peut rendre un changed_files bien typé qui oublie un fichier modifié. Les gates et l’état des lieux restent là pour ça.

Seul, en dernier, une seule fois

terminate: true évite l’appel LLM suivant seulement si tous les outils du lot sont terminants. Un bash lancé dans le même lot ferait repartir le modèle pour un tour de plus, facturé. D’où la consigne du contrat : appeler report_phase seul, en dernier, une seule fois.

La porte doit être dans l’allowlist

--tools retire aussi les outils d’extension absents de la liste. L’adaptateur ajoute donc report_phase d’office aux outils du roster. Le YAML n’a rien à déclarer et KNOWN_TOOLS ne change pas.

Le chemin de secours

Un pi lancé sans son socle n’a pas l’outil, et l’adaptateur _run_claude ne passe pas de schéma. Le même contract() dit donc « appelle l’outil, et s’il n’existe pas, termine par l’objet JSON ». parse() accepte encore un texte. Un appel d’outil écrit en texte (<tool_call>, <function=) devient un échec de contrat avec son motif, au lieu d’un « aucun objet JSON » obscur.

Deux portes, un contrat

	Porte typée (`report_phase`)	Convention de texte (`find` / `rfind`)
Qui définit la forme	`envelopes.schema()` → schéma JSON	`envelopes.contract()` → prose
Une accolade dans la prose	sans effet	casse l’extraction
Coût d’un écart de forme	correction dans le même tour	une reprise (`correction()`)
Harnais	pi avec le socle chargé	Claude Code, pi sans socle

Ce que la porte économise

Chaque reprise évitée épargne un tour sur un contexte de 40 à 60 k jetons, soit quelques centimes et une trentaine de secondes. Sur un SDLC complet, le gain se compte en dizaines de centimes et en minutes. Et le motif d’un écart devient exact : « changed_files : array attendu, string reçu ».

Commande : ce que le runner lance désormais

pi -p --mode json --approve --session-id <uuid> --session-dir /abs/plume-factory/adws/adw_data/sessions --no-skills --no-prompt-templates --model openrouter/z-ai/glm-5.3 --thinking medium --tools read,bash,edit,write,grep,find,ls,report_phase --

Le prompt arrive par stdin. Un tiret initial n’est plus refusé, et un shim .cmd sous Windows ne coupe plus le prompt à la première nouvelle ligne.

--approve charge le socle .pi/ à coup sûr en headless. Sans lui, un run sans trust.json ignorait en silence damage control et la trace. Ce drapeau vaut confiance au dépôt courant : jamais sur un dépôt que vous n’avez pas lu.

Côté Claude Code : l’équivalent de la porte est la sortie structurée. claude -p --output-format json --json-schema rend l’objet validé dans le champ structured_output, et le modèle est relancé en cas d’écart. Un outil sur mesure, lui, passe par un serveur MCP (claude mcp add, .mcp.json ou --mcp-config).

Injecter le contexte de phase de façon déterministe

Aujourd’hui, SCOUT_BRIEF, PLANNER_BRIEF et BUILDER_BRIEF sont des constantes Python collées en tête de chaque make_ask. Ces briefs ne changent pas d’un run à l’autre, pas plus que les chemins autorisés. C’est du contexte déterministe : le code doit le poser dans la session. Le prompt ne garde que ce qui varie : la demande, la passation, le contrat.

Les briefs deviennent des fichiers versionnés, adws/adw_config/briefs/<role>.md.
Le runner transmet l’identité de la phase par l’environnement : FACTORY_PHASE, FACTORY_ROLE, FACTORY_ADW_ID, comme il le fait pour FACTORY_ENVELOPE_SCHEMA.
Une extension lit cet environnement à before_agent_start et injecte un message factory-context : le brief, les écritures autorisées, les fichiers protégés.

Une extension qui lit l’environnement est déterministe. Une extension qui devinerait le rôle dans le prompt ne l’est pas.

Le crochet `before_agent_start`

Il se déclenche après la soumission du prompt et avant la boucle d’agent. Il peut rendre un message custom, stocké dans la session et envoyé au modèle, et modifier le prompt système du tour.

L’extension complète, factory-context.ts, et les briefs en fichiers seront posés au chapitre A7. Cet extrait montre ce qu’elle fera, pour que vous lisiez harness.py v4 en sachant à quoi servent ses variables.

// .pi/extensions/factory-context.ts — extrait : le contexte déterministe, posé par le code.
pi.on("before_agent_start", async () => {
  const role = process.env.FACTORY_ROLE;                   // posé par le runner, jamais deviné
  if (!role) return;                                       // session de poste : rien à injecter
  const brief = readFileSync(join("adws", "adw_config", "briefs", `${role}.md`), "utf8");
  return {
    message: { customType: "factory-context", content: brief, display: true },
  };
});

Pourquoi pas le prompt système

Le prompt système de pi est reconstruit à chaque tour depuis ses options (outils actifs, AGENTS.md, skills). Un --append-system-prompt par phase vit dans la ligne de commande, pas dans la session. Un message injecté à before_agent_start est stocké dans la session. La reprise en session vivante le retrouve, le rechargement ne le ré-émet pas, et la trace du chapitre A4 le voit passer.

Exemple : le builder

Le prompt du builder fait trois parties : le brief (~400 jetons), la mission (~50), le contrat (~150). Avec l’injection, l’extension lit briefs/builder.md, ajoute « écritures autorisées : apps/plume/**, protégés : adws/, .env, PLAN.md » et l’injecte avant le premier tour. Le prompt tombe à ~200 jetons.

Le coût bouge peu. Le gain est ailleurs. Une correction() n’a plus à répéter le brief. Et le brief est le même fichier pour pi (injecté par l’extension) et pour Claude Code (collé par le runner en tête de prompt).

Élaguer, c’est aussi du contexte

L’événement context permet de remplacer, avant l’appel LLM, un gros résultat d’outil par sa tête et sa taille, sauf dans les derniers tours. Sur une phase build de quarante tours, cela retire des dizaines de milliers de jetons d’entrée à chaque tour suivant.

Trois canaux pour le contexte de phase

Canal	Qui l’écrit	Quand	Ce qu’il porte
Variables d’environnement (`FACTORY_*`)	le runner, dans `_spawn`	avant le lancement	identité : phase, rôle, id de run, chemin du schéma
Message custom (`before_agent_start`)	l’extension, depuis les briefs	avant le premier tour	brief du rôle, chemins autorisés, fichiers protégés
Prompt utilisateur (stdin)	l’ADW (`make_ask`)	à chaque tentative	la demande, la passation, le contrat

Côté Claude Code : un hook SessionStart joue le rôle de before_agent_start. Son champ additionalContext entre dans le contexte au début de la conversation et reste dans le transcript. Claude Code le réinjecte à la reprise s’il manque, et après une compaction.

Côté Claude Code : un hook PostToolUse peut remplacer le résultat d’un outil avant que Claude le voie, avec updatedToolOutput. Il agit quand l’outil rend, alors que l’événement context de pi agit avant chaque appel LLM.

pi et Claude Code côte à côte

	pi	Claude Code
Enveloppe typée	outil terminal `report_phase`, validé par schéma	`--json-schema`, objet dans `structured_output`
Outil sur mesure	extension TypeScript (`registerTool`)	serveur MCP
Contexte posé par le code	message custom à `before_agent_start`	hook `SessionStart` (`additionalContext`)
Réduire un gros résultat d’outil	événement `context`, avant chaque appel LLM	hook `PostToolUse` (`updatedToolOutput`)

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

La pièce touche le port harnais et le contrat d’enveloppe, et ajoute factory-report.ts au socle pi. La couture ne bouge pas : le runner possède le graphe, l’agent reste un nœud borné. Seule change la porte par laquelle sa proposition traverse. Déterministe : le schéma, sa validation, la lecture du flux, le contrôle du modèle observé, l’identifiant de session choisi avant l’appel. Délégué : le contenu des champs. Coût : une centaine de jetons par tour pour décrire l’outil.

Travaux pratiques — la pièce du jour

Trois fichiers à poser dans plume-factory, votre usine logicielle agentique. Deux modules Python remplacent leurs versions précédentes, une extension rejoint le socle pi. Aucun ADW ne change, et tous passent désormais par la porte.

Pièce — `adws/adw_modules/envelopes.py`

Cette version remplace celle du chapitre 9. Tout ce qui existait reste. S’ajoutent schema(), REPORT_TOOL, un contract() qui nomme l’outil et garde le chemin de secours, et un parse() qui accepte un dict ou un texte. Les sous-types des chapitres 11 à 27 ne changent pas.

"""envelopes — les enveloppes JSON typees : le contrat de sortie des agents.

Le contexte ne traverse la couture agent -> code que sous une forme declaree
ici. Le type est la seule source de verite : contract() fabrique la demande
envoyee a l'agent, schema() fabrique le schema JSON de l'outil terminal
report_phase (annexe A6), parse() valide la reponse — les trois lisent les
memes champs, la demande ne peut pas deriver de la verification.

Version annexe A6 : l'enveloppe arrive de preference par une PORTE TYPEE —
un appel d'outil valide par schema cote pi — et parse() accepte un dict
(chemin rapide) ou un texte (chemin de secours : Claude Code, ou un pi sans
l'outil). Les sous-types des chapitres 11 a 27 ne changent pas.
"""
from __future__ import annotations

import json
import typing
from dataclasses import MISSING, asdict, dataclass, field, fields

# Le nom de l'outil terminal cote pi (.pi/extensions/factory-report.ts).
# L'adaptateur pi l'ajoute a l'allowlist ; contract() le nomme a l'agent.
REPORT_TOOL = "report_phase"

# Un modele ouvert qui « imite » un appel d'outil en texte : ce n'est pas une
# enveloppe, c'est un echec de contrat — avec son motif exact.
LEAKED_TOOL_CALL = ("<tool_call>", "<function=", "<invoke ")


class EnvelopeError(ValueError):
    """Le motif precis d'une enveloppe invalide — de quoi rediger la correction."""


@dataclass(frozen=True)
class Envelope:
    """La base de toute enveloppe : ce que chaque agent doit au code.

    Chaque champ porte dans ses metadonnees la phrase que contract() mettra
    dans le prompt — et que schema() mettra dans la description du champ de
    l'outil. Les sous-types (scout, plan, build...) etendent cette base avec
    leurs champs propres, et check() avec leurs invariants.
    """
    status: str = field(metadata={"ask": "'success' ou 'fail'",
                                  "choices": ("success", "fail")})
    summary: str = field(default="", metadata={
        "ask": "une phrase : ce que tu as fait ou trouve"})
    artifacts: list[str] = field(default_factory=list, metadata={
        "ask": "chemins des fichiers ecrits pour la suite, [] sinon"})
    notes_for_next_agent: str = field(default="", metadata={
        "ask": "ce que l'agent suivant doit savoir, '' sinon"})

    def check(self) -> None:
        """Les invariants du contrat — les sous-types etendent puis appellent super()."""
        if self.status not in ("success", "fail"):
            raise EnvelopeError("status doit valoir 'success' ou 'fail'")
        if not isinstance(self.summary, str):
            raise EnvelopeError("summary doit etre une chaine")
        if not isinstance(self.artifacts, list) or not all(
                isinstance(item, str) for item in self.artifacts):
            raise EnvelopeError("artifacts doit etre une liste de chemins (str)")
        if not isinstance(self.notes_for_next_agent, str):
            raise EnvelopeError("notes_for_next_agent doit etre une chaine")


def _required(expected: type[Envelope]) -> list[str]:
    return [f.name for f in fields(expected)
            if f.default is MISSING and f.default_factory is MISSING]


def contract(expected: type[Envelope] = Envelope) -> str:
    """La demande a coller dans le prompt — derivee du MEME type que parse().

    Deux portes, une seule demande : l'outil terminal quand il existe (pi,
    socle charge), l'objet JSON en fin de reponse sinon (Claude Code). Le
    contrat ne change pas selon le harnais — c'est l'adaptateur qui sait.
    """
    shape = {f.name: f.metadata.get("ask", "") for f in fields(expected)}
    return (
        f"Quand tu as fini, appelle l'outil {REPORT_TOOL} — seul, en dernier, "
        "une seule fois — avec exactement ces champs :\n"
        + json.dumps(shape, indent=2, ensure_ascii=False)
        + f"\nSi l'outil {REPORT_TOOL} n'existe pas dans ta session, termine ta "
        "reponse par cet objet JSON, sans bloc markdown autour."
    )


# ---------- le schema JSON de la porte : genere depuis le type, jamais recopie ----------

_SCALARS = {str: "string", int: "integer", float: "number", bool: "boolean"}


def _json_type(hint: object) -> dict:
    """Un indice de type Python -> un fragment de schema JSON. Inconnu = libre."""
    origin = typing.get_origin(hint) or hint
    if origin in _SCALARS:
        return {"type": _SCALARS[origin]}
    if origin is list:
        args = typing.get_args(hint)
        return {"type": "array", "items": _json_type(args[0]) if args else {}}
    if origin is dict:
        return {"type": "object"}
    return {}


def schema(expected: type[Envelope] = Envelope) -> dict:
    """Le schema JSON de l'outil report_phase — la meme source que contract().

    additionalProperties reste vrai : un sous-type (BuildEnvelope...) passe la
    porte du type de base sans etre tronque, et parse() juge ses invariants
    cote code. La porte garantit la FORME ; le code garde le VERDICT.
    """
    hints = typing.get_type_hints(expected)
    properties: dict[str, dict] = {}
    for f in fields(expected):
        prop = _json_type(hints.get(f.name, str))
        if f.metadata.get("ask"):
            prop["description"] = f.metadata["ask"]
        if f.metadata.get("choices"):
            prop["enum"] = list(f.metadata["choices"])
        properties[f.name] = prop
    return {"title": expected.__name__, "type": "object",
            "properties": properties, "required": _required(expected),
            "additionalProperties": True}


# ---------- la validation : un dict ou un texte, le meme contrat ----------

def _extract(text: str) -> dict:
    """Le chemin de secours : isoler l'objet JSON d'une reponse en prose."""
    if any(marker in text for marker in LEAKED_TOOL_CALL):
        raise EnvelopeError(f"appel d'outil rendu en texte au lieu d'un vrai appel "
                            f"— utilise l'outil {REPORT_TOOL}")
    start, end = text.find("{"), text.rfind("}")
    if start == -1 or end <= start:
        raise EnvelopeError("aucun objet JSON dans la reponse")
    try:
        payload = json.loads(text[start:end + 1])
    except json.JSONDecodeError as error:
        raise EnvelopeError(f"JSON invalide : {error}") from None
    if not isinstance(payload, dict):
        raise EnvelopeError("la reponse doit etre un objet JSON, pas une liste")
    return payload


def parse(payload: str | dict, expected: type[Envelope] = Envelope) -> Envelope:
    """Verifie le contrat, rend une enveloppe typee.

    Un dict (l'enveloppe rendue par l'outil terminal, deja validee par schema
    cote pi) entre directement ; un texte passe par _extract(). Strict sur le
    contrat dans les deux cas : champs requis, invariants de check().
    """
    data = payload if isinstance(payload, dict) else _extract(payload)
    spec = {f.name: f for f in fields(expected)}
    missing = [name for name in _required(expected) if name not in data]
    if missing:
        raise EnvelopeError(f"champs requis manquants : {missing}")
    envelope = expected(**{k: v for k, v in data.items() if k in spec})
    envelope.check()
    return envelope


def correction(motif: str, expected: type[Envelope] = Envelope) -> str:
    """La relance en session vivante : le motif exact, puis le contrat — rien d'autre."""
    return (
        f"Ta derniere reponse n'a pas passe la validation : {motif}.\n"
        "Rends a nouveau UNIQUEMENT l'enveloppe demandee.\n\n"
        + contract(expected)
    )


def handoff(previous: Envelope) -> str:
    """La passation : l'enveloppe precedente, prete a entrer dans le prompt suivant."""
    return (
        "### previous_envelope\n\n"
        "L'enveloppe rendue par l'agent precedent :\n"
        + json.dumps(asdict(previous), indent=2, ensure_ascii=False)
    )


if __name__ == "__main__":
    # La gate du module : contract(), schema() et parse() lisent le meme type —
    # zero token, instantane.
    print(contract(Envelope))
    generated = schema(Envelope)
    assert generated["required"] == ["status"], generated["required"]
    assert generated["properties"]["status"]["enum"] == ["success", "fail"]
    assert generated["properties"]["artifacts"] == {
        "type": "array", "items": {"type": "string"},
        "description": "chemins des fichiers ecrits pour la suite, [] sinon"}
    by_tool = parse({"status": "success", "summary": "ok"})
    by_text = parse('bla {"status": "success", "summary": "ok"} bla')
    assert by_tool == by_text and by_tool.artifacts == []
    for bad in ('{"status": "peut-etre"}', '<tool_call>{"name": "report_phase"}</tool_call>'):
        try:
            parse(bad)
        except EnvelopeError as error:
            print("rejet motive :", error)
        else:
            raise SystemExit("gate en echec : l'enveloppe invalide aurait du etre rejetee")
    print("contrat verifie : contract(), schema() et parse() lisent le meme type")

Pièce — `adws/adw_modules/harness.py`

Cette version remplace celle du chapitre 15. HarnessRequest gagne schema. HarnessResult gagne tokens, stop_reason et envelope. HarnessError porte la session interrompue. L’adaptateur pi tient en trois fonctions pures autour d’un seul subprocess, ce qui lui donne une gate à sec. L’adaptateur Claude Code remplit seulement les nouveaux champs. Le profil d’authentification du chapitre 17bis 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. 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 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
    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


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. 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
    return reading


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))
    try:
        proc = _spawn(_pi_argv(request, session_id), request,
                      extra_env={SCHEMA_ENV: str(schema_path)},
                      stdin_text=request.prompt)
    except HarnessError as error:
        error.session_id = session_id
        raise
    reading = _read_pi_stream(proc.stdout.splitlines())

    # Le verdict, dans l'ordre : la porte typee d'abord ; 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.
    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"{proc.stderr.strip()[-400:] or reading.text[-400:]}", session_id)
    elif proc.returncode != 0 and not reading.text:
        raise HarnessError(f"pi a rendu {proc.returncode} : {proc.stderr.strip()[-400:]}",
                           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=proc.returncode, tokens=reading.tokens,
                         stop_reason=reading.stop_reason, envelope=reading.envelope)


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 deux 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
    stream = [
        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}}}}),
        json.dumps({"type": "tool_execution_end", "toolName": "report_phase", "isError": False,
                    "result": {"details": {"status": "success", "summary": "fini"}}}),
        "pas du JSON — ignore",
    ]
    reading = _read_pi_stream(stream)
    assert reading.envelope == {"status": "success", "summary": "fini"}, reading
    assert reading.tokens == 1200 and reading.stop_reason == "toolUse"
    assert reading.observed_model == _route(request.model, request.direct)
    aborted = _read_pi_stream([json.dumps({"type": "message_end", "message": {
        "role": "assistant", "content": [], "stopReason": "aborted", "usage": {}}})])
    assert aborted.envelope is None and aborted.stop_reason == "aborted"

    # 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), flux lu : "
          f"{reading.tokens} jetons, enveloppe par la porte, arret 'aborted' detecte"
          " ; profil 17bis : route directe hors passerelle, coffre .env ferme sous profil, ouvert sans")

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

La porte typée. Au chargement, l’extension lit le schéma désigné par FACTORY_ENVELOPE_SCHEMA et enregistre report_phase avec. Sans schéma, comme dans votre session interactive, elle n’enregistre rien. Le fichier porte sa gate sous Bun, sans dépendance npm.

// .pi/extensions/factory-report.ts — la porte typée : l'outil terminal report_phase.
// L'enveloppe ne se devine plus dans la prose ; elle passe par un appel d'outil dont le schéma
// est celui qu'écrit l'usine (adws/adw_modules/envelopes.py → adws/adw_data/schemas/*.json),
// désigné par la variable FACTORY_ENVELOPE_SCHEMA. L'outil termine le tour (terminate: true) :
// pas d'appel LLM de plus une fois l'enveloppe rendue. Aucune dépendance npm à l'exécution.
import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import type { TSchema } from "typebox";

export const TOOL_NAME = "report_phase";
export const SCHEMA_ENV = "FACTORY_ENVELOPE_SCHEMA";

// Le sous-ensemble de JSON Schema que produit envelopes.schema() — rien de plus.
export interface JsonSchema {
  title?: string;
  type?: "object" | "array" | "string" | "integer" | "number" | "boolean";
  description?: string;
  enum?: unknown[];
  properties?: Record<string, JsonSchema>;
  required?: string[];
  items?: JsonSchema;
  additionalProperties?: boolean;
}

// ---------- lire le schéma : au chargement, depuis l'environnement posé par le runner ----------

export function loadSchema(path: string | undefined): JsonSchema | null {
  if (!path || !existsSync(path)) return null;
  const schema = JSON.parse(readFileSync(path, "utf8")) as JsonSchema;
  return schema.type === "object" && schema.properties ? schema : null;
}

// ---------- vérifier : la forme, champ par champ, avec un motif exact par écart ----------

function typeOf(value: unknown): string {
  if (Array.isArray(value)) return "array";
  if (value === null) return "null";
  if (typeof value === "number") return Number.isInteger(value) ? "integer" : "number";
  return typeof value;
}

export function violations(schema: JsonSchema, value: unknown, path = "enveloppe"): string[] {
  const found: string[] = [];
  const kind = typeOf(value);
  if (schema.type && !(kind === schema.type || (schema.type === "number" && kind === "integer"))) {
    return [`${path} : ${schema.type} attendu, ${kind} recu`];
  }
  if (schema.enum && !schema.enum.includes(value)) {
    return [`${path} : valeur hors de ${JSON.stringify(schema.enum)}`];
  }
  if (schema.type === "object" && value && typeof value === "object") {
    const record = value as Record<string, unknown>;
    for (const name of schema.required ?? []) {
      if (!(name in record)) found.push(`${path}.${name} : champ requis manquant`);
    }
    for (const [name, sub] of Object.entries(schema.properties ?? {})) {
      if (name in record) found.push(...violations(sub, record[name], `${path}.${name}`));
    }
  }
  if (schema.type === "array" && schema.items && Array.isArray(value)) {
    value.forEach((item, i) => found.push(...violations(schema.items as JsonSchema, item, `${path}[${i}]`)));
  }
  return found;
}

// La description lue par le modèle : une ligne par champ, depuis le schéma — jamais recopiée.
export function describe(schema: JsonSchema): string {
  const lines = Object.entries(schema.properties ?? {}).map(([name, sub]) => {
    const required = (schema.required ?? []).includes(name) ? " (requis)" : "";
    const choices = sub.enum ? ` — ${sub.enum.map((v) => JSON.stringify(v)).join(" | ")}` : "";
    return `- ${name}${required} : ${sub.description ?? sub.type ?? "libre"}${choices}`;
  });
  return [
    `Rends l'enveloppe de fin de phase à l'usine (${schema.title ?? "Envelope"}).`,
    "Appelle cet outil SEUL, en DERNIER, UNE seule fois : il termine ton travail.",
    "Champs :",
    ...lines,
  ].join("\n");
}

// ---------- l'extension : une porte, ouverte seulement quand l'usine l'a demandée ----------

export default function (pi: ExtensionAPI) {
  const schema = loadSchema(process.env[SCHEMA_ENV]);
  // Sans schéma dans l'environnement (session interactive du poste), pas d'outil : la porte
  // n'existe que pour un nœud d'usine. Le socle reste chargé — l'inventaire (A8) le verra.
  if (!schema) return;

  pi.registerTool({
    name: TOOL_NAME,
    label: "Enveloppe de phase",
    description: describe(schema),
    promptSnippet: "Rendre l'enveloppe de fin de phase à l'usine (dernier geste, une seule fois)",
    promptGuidelines: [
      `Termine toujours par un appel à ${TOOL_NAME}, seul dans son tour, jamais suivi d'un autre message.`,
    ],
    // Le schéma JSON tel quel : pi le compile et refuse un appel non conforme AVANT execute().
    parameters: schema as unknown as TSchema,

    async execute(_toolCallId, params) {
      // Ceinture et bretelles : le même schéma, relu ici — un motif exact par écart,
      // renvoyé au modèle comme erreur d'outil (il corrige dans le même tour, pas de reprise).
      const problems = violations(schema, params);
      if (problems.length > 0) throw new Error(`enveloppe refusée : ${problems.join(" ; ")}`);
      return {
        content: [{ type: "text", text: "Enveloppe transmise à l'usine. Ne réponds plus." }],
        details: params as Record<string, unknown>,   // ce que lit le runner (tool_execution_end)
        terminate: true,                                // pas d'appel LLM de suite : la phase est finie
      };
    },
  });
}

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

if (import.meta.main) {
  const dir = mkdtempSync(join(tmpdir(), "factory-report-"));
  try {
    const file = join(dir, "Envelope.json");
    writeFileSync(file, JSON.stringify({
      title: "Envelope", type: "object", additionalProperties: true, required: ["status"],
      properties: {
        status: { type: "string", enum: ["success", "fail"], description: "'success' ou 'fail'" },
        summary: { type: "string", description: "une phrase" },
        artifacts: { type: "array", items: { type: "string" }, description: "chemins" },
      },
    }));
    const schema = loadSchema(file);
    const accepted = schema ? violations(schema, { status: "success", summary: "ok", artifacts: ["a.md"], extra: 1 }) : ["schéma non chargé"];
    const refused = schema ? violations(schema, { summary: "sans statut", artifacts: [3] }) : [];
    const wrongEnum = schema ? violations(schema, { status: "peut-etre" }) : [];
    const ok = schema !== null && accepted.length === 0 && refused.length === 2 && wrongEnum.length === 1
      && describe(schema).includes("status (requis)") && loadSchema(undefined) === null;
    console.log(`factory-report ${ok ? "OK" : "KO"} — schéma chargé, enveloppe valide acceptée, ${refused.length} écarts motivés : ${refused.join(" ; ")}`);
    process.exit(ok ? 0 : 1);
  } finally {
    rmSync(dir, { recursive: true, force: true });
  }
}

La gate du TP

Depuis la racine de plume-factory, identique dans bash et PowerShell. La quatrième commande lance un ADW avec un prompt qui commence par un tiret. La dernière lit la trace.

uv run adws/adw_modules/envelopes.py
uv run python -m adws.adw_modules.harness
bun .pi/extensions/factory-report.ts
uv run adws/adw_prompt.py -- "- Quels fichiers composent apps/plume ? - Reponds en une phrase."
sqlite3 adws/adw_data/factory.db "SELECT name, COUNT(*) FROM harness_by_phase WHERE adw_id = (SELECT adw_id FROM runs ORDER BY started_at DESC LIMIT 1) AND type = 'tool_end' GROUP BY name ORDER BY name;"

Résultat attendu : contrat verifie : contract(), schema() et parse() lisent le meme type, puis harness OK — argv v4 (approve, stdin, --, report_phase)…, puis factory-report OK — schéma chargé, enveloppe valide acceptée, 2 écarts motivés. L’ADW rend un run vert, et la requête compte un report_phase. Sans client sqlite3, uv run adws/adw_modules/harness_trace.py --last montre le même outil. Coût : rien pour les trois gates à sec, moins d’un centime pour l’ADW.

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.