pi, Claude Code & le port harnais

Au chapitre 3, hello_factory.py appelait pi et claude directement, avec ce commentaire en forme de promesse : « appel direct encore permis ici, à partir du chapitre 7 seul adw_modules/harness.py aura ce droit ». Nous y sommes. Aujourd’hui, vous allez d’abord ouvrir le capot des deux harnais en mode headless, comprendre ce qui entre par argv et ce qui ressort en JSON, puis enfermer cette connaissance dans une pièce unique : le port harnais. À la fin du chapitre, changer d’agent dans toute l’usine se fera en changeant une chaîne de caractères, et plus jamais un script de plume-factory n’aura besoin de savoir comment pi ou claude s’invoquent. C’est la dernière pièce du poste de pilotage, celle sur laquelle le runner du chapitre 8 viendra se brancher.

pi et Claude Code en mode headless

L’idée en une phrase

Le mode headless d’un harnais est son visage pour les machines : le prompt entre par la ligne de commande, la réponse ressort structurée sur stdout, et une session se poursuit par un identifiant. C’est la forme sous laquelle l’usine, côté code déterministe, pilote ses agents, qui restent l’unique partie agentique de la couture.

Points clés

• pi : pi -p --mode json "prompt" émet un flux JSONL, un événement JSON par ligne pendant que l’agent travaille : agent_start, message_end, tool_execution_end… La première ligne est l’en-tête de session, les message_end de l’assistant font foi (texte final, usage, coût du tour).
• Claude Code : claude -p "prompt" --output-format json rend un objet JSON unique à la fin du tour, champs result, session_id, total_cost_usd. Un --output-format stream-json existe aussi pour suivre le travail en direct, sur le même modèle événementiel que pi.
• Poursuivre une session : chez pi, c’est vous qui nommez la session (--session-id + --session-dir). Même id, même dossier, même contexte, que la session existe déjà ou non. Chez Claude Code, c’est lui qui la nomme : vous capturez le session_id de la réponse et le rendez via --resume.
• La session qui se poursuit est la clé de voûte du livre : c’est elle qui rend la correction en session vivante (chapitre 3) possible depuis un script. L’échec d’une gate reviendra à l’agent dans le même contexte, sans repartir de zéro.
• Sans option de modèle, chaque harnais utilise son modèle configuré par défaut. Le roster du module 4 pilotera ce choix depuis l’usine, à travers la pièce que vous posez aujourd’hui.

Exemple concret

Posez la même question aux deux harnais depuis un script : « quelle commande lance les tests de apps/plume ? ». Côté pi, votre script lit le flux ligne à ligne et garde le dernier message_end de l’assistant : le texte y est, l’usage aussi, pour quelques milliers de tokens, ~1 à 2 centimes, 15 à 40 secondes. Côté Claude Code, votre script attend l’objet final et lit result et total_cost_usd : mêmes ordres de grandeur. Dans les deux cas, la différence avec le chapitre 2 saute aux yeux : plus personne ne « regarde l’écran ». La sortie est parsable, le coût est rapporté par le harnais lui-même, et l’identifiant de session est un objet que votre code peut stocker, rejouer, tracer. C’est exactement la matière première dont un runner a besoin.

Les deux harnais au guichet headless

Aspect	pi	Claude Code
Invocation	pi -p --mode json "…"	claude -p "…" --output-format json
Forme de sortie	flux JSONL, un événement par ligne	un objet JSON unique en fin de tour
Poursuivre la session	vous nommez : --session-id + --session-dir	il nomme : capturer session_id, rendre via --resume
Coût rapporté	par tour, dans l’usage des message_end	global, champ total_cost_usd

Commande — le tour headless, dans les deux dialectes

Les deux versions, côte à côte. C’est la dernière fois du livre que vous les tapez à la main, dès la fiche suivante le port s’en charge :

# version pi — filtrer le flux : seuls les message_end portent la reponse et l'usage
pi -p --mode json "Quelle commande lance les tests de apps/plume ?" \
  | jq -c 'select(.type == "message_end")'

# version Claude Code — un seul objet a lire, a la fin
claude -p "Quelle commande lance les tests de apps/plume ?" --output-format json \
  | jq -r '.result, .session_id, .total_cost_usd'

# poursuivre une session Claude Code : capturer l'id, puis --resume
id=$(claude -p "Audite le store de Plume" --output-format json | jq -r '.session_id')
claude -p "Continue : liste les tests manquants" --resume "$id"

Piège courant : « headless, c’est juste capturer stdout » oublie stdin. Un harnais lancé en subprocess qui hérite du stdin de son parent peut se croire branché sur un tube et attendre indéfiniment une entrée qui ne viendra jamais : échec silencieux et total, 0 % de CPU, aucune sortie, aucun message d’erreur. La parade est systématique : décider du sort de stdin à chaque appel headless. Fermé (stdin=DEVNULL en Python) quand le prompt voyage dans argv, écrit puis refermé quand le prompt voyage par le tube. Jamais hérité.

---

Le port harnais : deux adaptateurs, une frontière

L’idée en une phrase

Le port harnais est l’interface neutre, une HarnessRequest entre et un HarnessResult sort, que l’usine possède et que chaque harnais rejoint par un adaptateur. La pièce adw_modules/harness.py vit entièrement côté code déterministe, et à partir d’aujourd’hui la couture vers les agents passe par elle, et par elle seule.

Points clés

• Le port appartient à l’usine, pas au harnais : c’est vous qui décidez ce qu’un agent a le droit de recevoir (prompt, session_id, cwd, timeout) et ce qu’il doit rendre (text, session_id, cost_usd, returncode). Les harnais s’adaptent à ce contrat, jamais l’inverse.
• Les adaptateurs absorbent les asymétries : qui nomme la session, flux JSONL ou objet unique, coût par tour ou coût global, erreur en code retour ou en stderr. Derrière le port, tout cela devient un seul et même HarnessResult.
• La règle entre en vigueur aujourd’hui : plus aucun appel direct à pi ou claude hors de harness.py. Les appels directs des chapitres 2 et 3 étaient l’échafaudage, il tombe.
• Changer d’agent devient une donnée, plus une décision d’architecture : "pi" ou "claude" en argument de run(), et demain une entrée de roster dans un YAML.
• Un port se teste sans dépenser un token : un adaptateur factice qui rend des réponses connues suffira, au module 3, à tester le runner à sec.

Exemple concret

Un matin, l’un des harnais renomme un de ses drapeaux headless. Ces outils évoluent vite, cela arrivera. Sans port, le drapeau est répété dans chaque script qui invoque l’agent : l’embryon du chapitre 3, puis le scout, le planner, le builder des modules à venir. C’est une heure de grep, six fichiers modifiés, et la peur d’en avoir oublié un septième. Avec le port : une fonction à corriger dans harness.py, une gate à relancer, quelques minutes et zéro token si vous vérifiez d’abord avec l’adaptateur factice. La frontière a un deuxième dividende : le jour où un troisième harnais vous fait envie, l’essayer coûte un adaptateur d’une trentaine de lignes, pas une réécriture de l’usine.

Ce que le port normalise

Asymétrie	Côté pi	Côté Claude Code	Derrière le port
Nom de session	vous le fournissez	il le génère	result.session_id, toujours prêt à poursuivre
Sortie	flux JSONL à filtrer	objet unique à parser	result.text
Coût	par tour, dans l’usage	global, total_cost_usd	result.cost_usd
Échec	code retour + stderr	code retour + stderr	une seule exception : HarnessError

Script — le contrat du port

Le cœur de la pièce du jour tient en deux dataclasses gelées, l’enveloppe minimale qui a le droit de traverser la couture (le fichier complet est dans les travaux pratiques) :

@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

@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
    session_id: str    # de quoi poursuivre la MEME session
    cost_usd: float    # 0.0 si le harnais ne rapporte pas le cout
    returncode: int

Piège courant : « encore une couche d’abstraction, c’est de l’over-engineering » inverse la charge. Le port n’ajoute pas une indirection décorative, il épingle la couture à un endroit unique et nommé. Sans lui, chaque script réapprend les dialectes de chaque harnais, et c’est cette duplication-là qui est la complexité. Le critère reste celui du livre : une abstraction se paie quand elle supprime de la connaissance répétée, ici six scripts qui n’auront jamais à savoir comment pi s’invoque.

---

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

La pièce du jour est adws/adw_modules/harness.py, la dernière case de la zone « Poste de pilotage », voisine du justfile du chapitre 5 qui l’invoquera et du workspace herdr du chapitre 6 qui hébergera les agents qu’elle lance. La loi « l’agent propose, le code dispose » gagne aujourd’hui sa frontière physique : le port est du déterminisme pur, l’agent ne vit que de l’autre côté, et le contexte ne traverse que dans l’enveloppe HarnessRequest → HarnessResult, l’ancêtre direct des enveloppes typées du chapitre 9. Coût à l’usage : zéro token, le port ne parle pas aux modèles, il encadre ceux qui parlent. Ce qu’il économise : l’heure de chirurgie multi-fichiers à chaque évolution d’un harnais. Il ouvre aussi la porte du runner (chapitre 8), qui n’aura plus qu’à enchaîner des appels à run() sans jamais connaître les dialectes qu’il pilote.

---

Travaux pratiques — la pièce du jour

Une pièce complète à poser dans le repo compagnon plume-factory, qui devient, chapitre après chapitre, votre usine logicielle agentique. Aujourd’hui : le port harnais, et l’embryon du chapitre 3 qui passe par la porte.

Pièce — adws/adw_modules/harness.py

Le port et ses deux adaptateurs, bibliothèque standard uniquement. Il vit du côté déterministe et s’appuie sur le .gitignore du chapitre 1 : les sessions pi sont rangées dans adws/adw_data/, qui n’est jamais commité. C’est un module importé par les scripts, pas un script : il n’a pas d’en-tête PEP 723. Notez l’asymétrie d’entrée entre les deux adaptateurs : le prompt de pi part dans argv, celui de claude par stdin. Un argument multi-lignes ne survit pas aux shims .cmd de Windows, et c’est à l’adaptateur d’absorber ce genre de détail, jamais à vos scripts.

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

Une frontiere, deux adaptateurs. A partir d'aujourd'hui (chapitre 7), 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.
"""
from __future__ import annotations

import json
import shutil
import subprocess
import uuid
from dataclasses import dataclass
from pathlib import Path

# Les sessions pi vivent dans adw_data/ — couvert par le .gitignore du ch. 1.
SESSION_DIR = Path("adws/adw_data/sessions")


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


@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


@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
    session_id: str    # de quoi poursuivre la MEME session
    cost_usd: float    # 0.0 si le harnais ne rapporte pas le cout
    returncode: int


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 _spawn(cmd: list[str], request: HarnessRequest,
           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 ?")
    # 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 prompts de
    #   l'usine deviendront multi-lignes des le chapitre 9.
    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",
                              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")


def _run_pi(request: HarnessRequest) -> HarnessResult:
    # pi : c'est VOUS qui nommez la session. Meme id + meme dossier = meme
    # contexte — que la session existe deja ou non.
    session_id = request.session_id or str(uuid.uuid4())
    SESSION_DIR.mkdir(parents=True, exist_ok=True)
    proc = _spawn(["pi", "-p", "--mode", "json",
                   "--session-id", session_id,
                   "--session-dir", str(SESSION_DIR),
                   request.prompt], request)

    # Le flux JSONL : seuls les message_end de l'assistant font foi. Le dernier
    # texte gagne (l'agent a pu parler entre deux outils) — mais chaque tour a
    # coute, donc le cout s'additionne au lieu de se remplacer.
    text, cost = "", 0.0
    for line in proc.stdout.splitlines():
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if event.get("type") != "message_end":
            continue
        message = event.get("message", {})
        if message.get("role") != "assistant":
            continue
        text = _text_of(message) or text
        usage = message.get("usage", {}) or {}
        cost += (usage.get("cost", {}) or {}).get("total", 0.0) or 0.0

    if proc.returncode != 0 and not text:
        raise HarnessError(f"pi a rendu {proc.returncode} : {proc.stderr.strip()[-400:]}")
    return HarnessResult(text=text, session_id=session_id,
                         cost_usd=cost, returncode=proc.returncode)


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.
    cmd = ["claude", "-p", "--output-format", "json"]
    if request.session_id:
        cmd += ["--resume", request.session_id]
    # 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, stdin_text=request.prompt)

    if proc.returncode != 0:
        # stderr d'abord, stdout sinon : en --output-format json, claude
        # ecrit souvent son erreur en JSON sur stdout, stderr vide.
        evidence = (proc.stderr.strip() or proc.stdout.strip())[-400:]
        raise HarnessError(f"claude a rendu {proc.returncode} : {evidence}")
    try:
        payload = json.loads(proc.stdout)
    except json.JSONDecodeError:
        raise HarnessError("claude n'a pas rendu l'objet JSON attendu "
                           "(--output-format json)") from None
    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)


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

Pièce — adws/hello_factory.py

Cette version remplace celle du chapitre 3. Même contrat, même gate, mêmes codes retour. Seul le chemin vers l’agent change : plus d’argv de harnais en dur, la requête passe par le port. L’option --harness garde son interface, la recette just hello du chapitre 6 fonctionne sans modification.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# ///
"""hello_factory — l'embryon du runner, version port harnais.

Cette version remplace celle du chapitre 3 : plus aucun appel direct a pi ou
claude. L'agent propose — via le port. Le code dispose — comme avant.
"""
import argparse
import json
import sys

from adw_modules import harness

PROMPT = (
    "Reponds UNIQUEMENT avec un objet JSON, sans texte autour ni bloc markdown, "
    "de la forme suivante : un champ 'tool' (la commande qui lance les tests du "
    "projet apps/plume) et un champ 'purpose' (son role, en une phrase)."
)

REQUIRED_KEYS = {"tool", "purpose"}


def extract_json(text: str) -> str:
    """Isole le premier objet JSON de la sortie — l'agent ajoute parfois du texte autour."""
    start, end = text.find("{"), text.rfind("}")
    if start == -1 or end <= start:
        raise ValueError("aucun objet JSON dans la sortie de l'agent")
    return text[start:end + 1]


def main() -> int:
    parser = argparse.ArgumentParser(description="Un port, une sortie JSON validee.")
    parser.add_argument("--harness", choices=sorted(harness.ADAPTERS), default="pi")
    args = parser.parse_args()

    # L'agent propose — par l'unique porte de l'usine.
    try:
        result = harness.run(args.harness, harness.HarnessRequest(prompt=PROMPT))
    except harness.HarnessError as error:
        print(f"harnais en echec : {error}", file=sys.stderr)
        return 1

    # Le code dispose : JSON parsable, contrat respecte — sinon, refus motive.
    try:
        payload = json.loads(extract_json(result.text))
    except ValueError as error:
        print(f"REFUSE : {error}", file=sys.stderr)
        return 2

    missing = REQUIRED_KEYS - payload.keys()
    if missing:
        print(f"REFUSE : champs manquants {sorted(missing)}", file=sys.stderr)
        return 3

    print(json.dumps(payload, indent=2, ensure_ascii=False))
    # La signature du port : de quoi poursuivre la session, et son cout.
    print(f"session {result.session_id} — cout ~{result.cost_usd:.4f} $",
          file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())

La gate du TP

just hello && just hello claude && echo "gate : OK"

Attendu : deux fois l’objet JSON validé (le champ tool devrait mentionner bun test), chacun suivi de sa ligne session … — cout …, la preuve que les deux dialectes rendent désormais le même HarnessResult, puis gate : OK. Coût : ~2 à 4 centimes, 30 à 80 secondes pour les deux harnais, un seul suffit si vous n’en avez installé qu’un (just hello seul). Quand la gate passe, commitez et cochez le jalon « ch. 7 » dans PLAN.md : plus aucun appel direct au harnais ne subsiste dans l’usine.

---