L'enveloppe devient une porte typée

Au chapitre 9, vous avez écrit parse() : la réponse de l’agent, text.find("{"), text.rfind("}"), et l’objet JSON qui se trouve entre les deux devient une enveloppe. Cela tourne depuis vingt-quatre chapitres, jusqu’au jour où un builder, fier de son travail, colle dans sa prose un extrait const cfg = { theme: "dark" } avant l’objet final : find s’arrête sur la première accolade, json.loads échoue, et l’usine paie une reprise pour une virgule. Un modèle ouvert qui rend son appel d’outil en texte (<tool_call> dans la réponse) passe pour de la prose, lui aussi. Vous avez fermé l’annexe au chapitre A5 avec un nœud pi plus sûr, plus observable et transportable. Les cinq chapitres qui commencent ici le rendent vérifiable par le runner. À la fin de celui-ci, l’enveloppe ne sera plus une convention de prose mais une porte typée : un appel d’outil validé par schéma côté pi, dont le schéma est écrit par envelopes.py avant la phase, et que le runner lit dans le flux JSON sans rien deviner. La pièce du jour fait évoluer deux modules posés aux chapitres 9 et 15, envelopes.py et harness.py, et ajoute une extension du socle, factory-report.ts, à côté de la trace du chapitre A4.

Un outil terminal pour rendre l’enveloppe

L’idée en une phrase

Un outil terminal est un outil enregistré par une extension pi dont l’appel termine le tour (terminate: true) : le modèle « rend » son enveloppe en l’appelant, pi la valide par schéma avant même d’exécuter l’outil, et le runner la lit dans l’événement tool_execution_end du flux JSON. L’enveloppe traverse la couture par une porte que le code déterministe a dessinée (le schéma vient de envelopes.py), et l’agent ne décide plus de sa forme.

Points clés

• Une seule source de vérité, côté Python. envelopes.schema(cls) transforme la dataclass en schéma JSON (types, champs requis, enum sur status, descriptions tirées des ask). L’adaptateur pi l’écrit dans adws/adw_data/schemas/<Envelope>.json avant la phase et passe son chemin à pi par FACTORY_ENVELOPE_SCHEMA, et l’extension le lit au chargement pour enregistre report_phase avec ce schéma. Aucun champ n’est recopié en TypeScript.
• Validé deux fois, décidé une fois. pi compile le schéma et refuse un appel non conforme avant execute() : le modèle voit le motif exact et corrige dans le même tour, sans reprise du runner. Puis parse() accepte le dict de l’enveloppe et juge les invariants de check() (un build réussi déclare au moins un fichier…) : la porte garantit la forme, le code garde le verdict.
• Le tour s’arrête sur l’outil. Un résultat terminate: true évite l’appel LLM de suite, quand tous les outils du lot le sont. D’où la consigne du contrat : « appelle report_phase seul, en dernier, une seule fois ». Un bash lancé dans le même lot ferait repartir le modèle pour un tour de plus, facturé.
• L’allowlist doit contenir la porte. --tools retire aussi les outils d’extension absents de la liste. L’adaptateur ajoute report_phase d’office aux outils du roster, le YAML n’a rien à déclarer, et KNOWN_TOOLS reste celui du chapitre 27.
• Le chemin de secours reste ouvert. Claude Code n’a pas d’outil terminal typé, et un pi lancé sans son socle non plus. Le même contract() dit « appelle l’outil, et s’il n’existe pas, termine par l’objet JSON », et parse() accepte un texte comme avant. Un appel d’outil fuité en texte (<tool_call>, <function=) devient un échec de contrat motivé, pas un « aucun objet JSON » énigmatique.

Exemple concret

Reprenez le run du chapitre 13 : le builder implémente la spec « export Markdown » de Plume sur le workhorse (z-ai/glm-5.3, relevé ce jour sur openrouter.ai/models : 1,40 $ le million de jetons en entrée, 4,40 $ en sortie). Sa réponse finale cite un extrait de code entre accolades, puis l’objet JSON. Sans la porte : parse() isole { theme: "dark" }…}, json.loads échoue, correction() repart dans la session vivante avec « JSON invalide », un tour de plus sur un contexte de 40 à 60 k jetons, soit quelques centimes et une trentaine de secondes, pour un builder qui avait fini. Avec la porte : le modèle appelle report_phase avec ses champs, pi valide, le tour s’arrête, le runner lit result.details dans tool_execution_end, et il n’y a aucun tour supplémentaire. Sur un SDLC de cinq phases agent à deux ou trois reprises évitées par run, l’économie se compte en dizaines de centimes et en minutes. Surtout, le motif d’échec, quand il reste, est enfin exact : « changed_files : array attendu, string reçu ».

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
Qui valide en premier	pi, avant execute(), motif exact	parse() côté Python, après coup
Une accolade dans la prose	sans effet	casse l’extraction
Appel d’outil fuité en texte	impossible : c’est un vrai appel ou rien	échec motivé depuis A6
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

Config — le schéma que pi reçoit, tel qu’écrit par l’usine

Ce fichier, vous ne l’écrivez jamais à la main : _write_schema() le produit avant chaque phase depuis le type que parse() vérifiera ensuite.

{
  "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 n’est pas une faiblesse : tant que vos ADW ne passent pas leur sous-type au port (ils passent expected à parse(), pas à HarnessRequest), la porte est celle de l’enveloppe de base, et changed_files ou findings la traversent sans être tronqués et check() les juge ensuite. Le champ HarnessRequest.schema existe dès aujourd’hui pour qu’une phase demande une porte plus stricte, et la fabrique agent_action l’utilisera au chapitre A7.

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, pas dans la ligne : plus de tiret initial refusé, plus de nouvelle ligne tronquée par un shim .cmd sous Windows, et -- ferme les options. --approve charge le socle .pi/ à coup sûr en headless : sans lui, un run sans trust.json ignorait en silence damage control (A3) et trace (A4). Cela vaut confiance au dépôt courant : jamais sur un dépôt que vous n’avez pas lu.

Piège courant : « avec un schéma, l’enveloppe est garantie correcte » est inexact. Le schéma garantit la forme (types, champs requis, valeurs de status), pas la vérité. Un builder peut rendre un changed_files bien typé qui ne liste pas le fichier qu’il a réellement modifié. C’est le travail des gates du chapitre 12 et de l’état des lieux du chapitre 10, qui ne bougent pas : la porte typée retire l’ambiguïté de forme pour que le verdict porte enfin sur le fond.

---

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

L’idée en une phrase

Le brief d’un rôle (scout, planner, builder…), les chemins autorisés et les fichiers protégés sont du contexte déterministe : ils ne changent pas d’un run à l’autre, et c’est le code qui doit les poser dans la session, par un message injecté à before_agent_start depuis un fichier versionné, désigné par des variables d’environnement (FACTORY_PHASE, FACTORY_ROLE, FACTORY_ADW_ID). Le prompt ne porte plus que ce qui varie : la demande, la passation, le contrat.

Points clés

• Ce qui est constant sort du prompt. Aujourd’hui SCOUT_BRIEF, PLANNER_BRIEF et BUILDER_BRIEF sont des constantes Python collées en tête de chaque make_ask. Déplacées dans adws/adw_config/briefs/<role>.md, elles deviennent lisibles, relues en revue de code, et identiques pour pi (injection par extension) et pour Claude Code (le runner les colle en tête de prompt). Le port reste symétrique.
• before_agent_start est le bon crochet. Il se déclenche après la soumission du prompt et avant la boucle d’agent, et il peut rendre un message custom (stocké dans la session, envoyé au modèle) et modifier le prompt système du tour. C’est là qu’un message factory-context porte le brief, les writes du roster et les fichiers protégés.
• Les variables d’environnement, pas le prompt, portent l’identité de la phase. Le runner connaît phase, adw_id et le rôle, et il les transmet à pi par l’environnement du sous-processus, comme il transmet déjà FACTORY_ENVELOPE_SCHEMA. Une extension qui lit l’environnement est déterministe. Une extension qui devinerait le rôle dans le prompt ne l’est pas.
• Élaguer, c’est aussi du contexte. L’événement context permet de remplacer, avant l’appel LLM, un résultat d’outil de plusieurs dizaines de Ko par sa tête et sa taille, sauf les derniers tours. Sur une phase build de quarante tours, c’est des dizaines de milliers de jetons d’entrée en moins à chaque tour suivant.
• Bénéfice mesurable sur les reprises. Une correction() ne recopie pas le brief : il est déjà dans la session, posé une fois. À trois reprises par phase, sur un brief de 300 à 500 jetons, l’économie est petite. La lisibilité et la cohérence entre les deux harnais sont le vrai gain.

Exemple concret

Le builder du chapitre 27 reçoit aujourd’hui, à chaque tentative 1, un prompt de trois parties : le brief (~400 jetons), la mission (chemin de spec, ~50 jetons), le contrat (~150 jetons). Avec l’injection : le runner exporte FACTORY_ROLE=builder et FACTORY_PHASE=build, l’extension lit adws/adw_config/briefs/builder.md, ajoute « écritures autorisées : apps/plume/**, protégés : adws/, .env, PLAN.md » et injecte le tout en un message custom avant le premier tour. Le prompt utilisateur tombe à ~200 jetons. Le coût par phase ne bouge presque pas (le brief est lu une fois dans les deux cas). Ce qui change, c’est que le brief du builder est le même fichier que celui que lira Claude Code si le roster change de harnais, et que l’agent relancé en correction voit encore ses consignes, sans qu’un REPAIR_ASK les répète.

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

Script — le crochet, en dix lignes

L’extension complète (factory-context.ts) et le déplacement des briefs en fichiers forment une pièce à part entière : trois fichiers sont déjà posés aujourd’hui, elle sera posée au chapitre A7. Voici 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 },
  };
});

Piège courant : « autant tout mettre dans le prompt système » est inexact. Le prompt système de pi est reconstruit à chaque tour depuis ses options (outils actifs, AGENTS.md, skills), et un --append-system-prompt par phase vit dans la ligne de commande, pas dans la session. Un message custom injecté à before_agent_start est stocké dans la session : la reprise en session vivante le retrouve, le rechargement ne le ré-émet pas deux fois, et la trace du chapitre A4 le voit passer.

---

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

Sur le plan de l’usine, la pièce du jour touche le port harnais, la zone posée au chapitre 7 et raffinée aux chapitres 11 et 15, et le contrat d’enveloppe du chapitre 9. Côté pi, elle ajoute au socle une troisième extension, factory-report.ts, à côté de damage-control.ts (A3) et de factory-obs.ts (A4). La couture ne bouge pas : le runner Python possède toujours le graphe, et l’agent reste un nœud borné dans une phase nommée. Ce qui change, c’est la porte par laquelle sa proposition traverse : un appel d’outil validé par schéma, généré depuis le type Python, à la place d’un JSON deviné dans la prose. Déterministe : le schéma, sa validation, la lecture du flux, la vérification du modèle observé, le choix de l’identifiant de session avant l’appel. Délégué : le contenu des champs. Coût d’usage : zéro jeton pour le schéma et la porte. La description de l’outil ajoute une centaine de jetons au prompt système de chaque tour, et elle évite, à chaque écart de forme, un tour de correction sur tout le contexte, soit quelques centimes et une trentaine de secondes par écart évité. Note pour le chapitre 15 : l’adaptateur _run_pi posé là-bas passait le prompt en argument et ne passait pas --approve, et la version v4 d’aujourd’hui le remplace.

---

Travaux pratiques — la pièce du jour

Trois fichiers à poser dans plume-factory, qui devient, chapitre après chapitre, votre usine logicielle agentique : deux modules Python qui remplacent leurs versions précédentes, et une extension du socle pi. Aucun ADW ne change : adw_prompt.py, adw_scout.py, adw_build.py et adw_sdlc.py tournent tels quels et 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 : mêmes dataclasses, même check(), mêmes handoff() et correction(), même gate. S’ajoutent schema() (la dataclass devient un schéma JSON), REPORT_TOOL, un contract() qui nomme l’outil et garde le chemin de secours, et un parse() qui accepte un dict ou un texte, et qui motive précisément un appel d’outil fuité en texte. Les sous-types des chapitres 11 à 27 ne changent pas d’une ligne.

"""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 un champ optionnel schema, HarnessResult gagne tokens, stop_reason et envelope, et HarnessError porte la session interrompue. L’adaptateur pi est réécrit en trois fonctions pures autour d’un seul subprocess (_write_schema, _pi_argv, _read_pi_stream), ce qui lui donne, pour la première fois, une gate à sec. L’adaptateur Claude Code ne change que pour remplir les nouveaux champs. Le crédit de run.tokens par les ADW attendra la révision de agent_action au chapitre A7, et le chiffre est disponible dès aujourd’hui dans le résultat. Le profil d’authentification du chapitre 17bis (auth:, route, coffre) est conservé.

"""harness — le port de l'usine vers ses agents.

Une frontiere, deux adaptateurs. Aucun script de l'usine n'invoque `pi` ou
`claude` directement : tout passe par run(). Une HarnessRequest entre, un
HarnessResult sort — quel que soit le harnais derriere la porte.

Version chapitre 17bis : le profil d'authentification par agent. Une
requete peut porter `auth`, les variables de credentials que le roster a
resolues pour cet agent (noms ET valeurs, jamais dans le prompt), et
`direct`, la route hors passerelle d'un profil de forfait ou d'API native.
Quand elle porte un profil, .env devient un COFFRE : le noeud ne recoit
plus tout ce que .env avait charge, seulement ce que son profil nomme — a
la place du .env global du ch. 15, meme preseance (.env puis environnement
reel). Sans `auth`, l'heritage du ch. 15 s'applique tel quel.

Version annexe A6 (v4 de l'adaptateur pi) : le runner ne fait plus confiance
au noeud pi, il le VERIFIE. Le socle .pi/ est charge a coup sur (--approve),
le prompt voyage par stdin, l'enveloppe arrive par l'outil terminal
report_phase (tool_execution_end) plutot que devinee dans la prose, les
jetons et le stopReason sont lus, le modele observe est compare au modele
demande, et l'identifiant de session est choisi AVANT l'appel — une erreur
le porte, la reprise ne perd plus sa session. 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 ce schéma tel quel : pi le compile et refuse un appel non conforme avant execute(). Sans schéma dans l’environnement (votre session interactive), elle n’enregistre rien : la porte n’existe que pour un nœud d’usine. Sa logique (loadSchema, violations, describe) est pure et exportée, et 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. Cinq commandes, une par ligne, identiques dans bash et PowerShell. Les trois premières ne coûtent rien, la quatrième lance un ADW d’une phase 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 son enveloppe, alors que le même prompt en argument aurait été refusé par pi, et la requête compte un report_phase parmi les tool_end du run. Sans client sqlite3, uv run adws/adw_modules/harness_trace.py --last montre le même outil. Coût : les trois gates à sec ne dépensent rien, et l’ADW coûte moins d’un centime en une vingtaine de secondes sur le workhorse du roster (variante éco déjà en place : le modèle est celui de factory.config.yaml).

---