Enveloppes JSON typées & le handoff de contexte

Hier, adw_prompt.py a exigé de l’agent un objet JSON avec status et summary, et vous avez vu la correction en session vivante tourner. Mais regardez de plus près comment cette exigence était écrite : la demande vivait dans une chaîne de caractères (ENVELOPE_ASK), et la vérification dans une fonction à part (extract_envelope). Deux endroits, un seul contrat. Le jour où vous ajoutez un champ, lequel des deux oublierez-vous de mettre à jour ? Aujourd’hui, vous fusionnez les deux en une seule source de vérité : un type. À la fin du chapitre, vous saurez pourquoi l’enveloppe est la seule forme sous laquelle le contexte a le droit de traverser la couture, comment un type génère à la fois la demande et la vérification, et comment le handoff transporte le travail d’un agent vers le suivant, par le code et jamais par une conversation partagée. La pièce du jour, adws/adw_modules/envelopes.py, s’emboîte entre le port du chapitre 7 et le runner du chapitre 8.

L’enveloppe JSON typée

L’idée en une phrase

L’enveloppe est la seule forme sous laquelle le contexte traverse la couture de l’agent vers le code, et la rendre typée, c’est déclarer ce contrat une seule fois, dans un type Python d’où dérivent à la fois la demande envoyée à l’agent et la validation de sa réponse. Cette pièce, envelopes.py, vit entièrement côté code déterministe.

Points clés

• status porte le run : c’est le seul champ obligatoire, et il ne peut valoir que success ou fail. Une enveloppe qui parse mais déclare fail fait échouer la phase : un agent qui annonce lui-même son échec n’est pas une phase réussie.
• Les autres champs ont des valeurs par défaut : summary (une phrase sur ce qui s’est passé), artifacts (les fichiers écrits pour la suite) et notes_for_next_agent (ce que le prochain agent doit savoir). Un agent laconique rend une enveloppe valide, un agent muet, non.
• Une seule source de vérité : contract() fabrique la demande et parse() fait la vérification à partir des mêmes champs du même type. Ajoutez un champ au type, et les deux bougent ensemble : la demande ne peut plus dériver de la vérification.
• Le parseur est tolérant en entrée, strict sur le contrat : il isole l’objet JSON même si l’agent l’a entouré de prose ou d’un bloc de code, mais les champs requis, eux, ne se négocient pas. Le prompt continue de demander du JSON nu, la tolérance est un filet, pas une invitation.
• L’enveloppe est un manifeste de déclarations, pas une preuve : l’agent affirme avoir écrit tel fichier. Vérifier que les déclarations sont vraies, c’est le travail des gates, que vous poserez au chapitre 12.

Exemple concret

Supposez que vous vouliez ajouter notes_for_next_agent au contrat d’hier, version chaînes de caractères : il faut penser à modifier ENVELOPE_ASK et extract_envelope. Oubliez le second, et rien n’explose : le champ arrive, personne ne le vérifie. Oubliez le premier, et chaque run part en correction : un tour d’agent de plus par run, quelques milliers de tokens à chaque fois, jusqu’à ce que quelqu’un remarque la dérive dans les logs. Avec le type : une ligne ajoutée dans la dataclass, et la demande comme la validation sont à jour, pour zéro token et zéro dérive possible. Le typage ne coûte rien à l’usage, c’est du code pur. Ce qu’il élimine, c’est une classe entière de pannes silencieuses facturées au token.

Enveloppe libre (chapitre 8) vs enveloppe typée

Script — le contrat, une seule source

Le cœur de la pièce (fichier complet dans les travaux pratiques) : chaque champ porte, dans ses métadonnées, la phrase que contract() mettra dans le prompt. La demande et la vérification lisent la même déclaration :

@dataclass(frozen=True)
class Envelope:
    """La base de toute enveloppe : ce que chaque agent doit au code."""
    status: str = field(metadata={"ask": "'success' ou '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 contract(expected: type[Envelope] = Envelope) -> str:
    """La demande a coller dans le prompt — derivee du MEME type que parse()."""

Piège courant : « valider le JSON, c’est vérifier la syntaxe » est inexact. json.loads accepte n’importe quel objet bien formé, y compris un objet vide. La validation typée vérifie le contrat : champs requis présents, status admissible. La vérité des déclarations (les fichiers annoncés existent-ils ?) relèvera d’un troisième étage encore, les gates du chapitre 12. Syntaxe, contrat, vérité : trois vérifications, trois pièces.

Le handoff : le contexte traverse la couture

L’idée en une phrase

Le handoff est le mécanisme par lequel le travail d’un agent parvient au suivant par le code : l’agent sortant n’a que deux canaux, des fichiers de référence et son enveloppe finale, et c’est le runner qui parse, conserve, puis injecte l’enveloppe dans le prompt du prochain agent. La passation vit donc côté déterministe, jamais dans une conversation partagée.

Points clés

• Deux canaux, exactement : des fichiers écrits pour la suite (déclarés dans artifacts) et l’enveloppe finale. Tout le reste (le raisonnement, les tâtonnements, la conversation) reste dans la session de l’agent et n’en sort pas.
• Le prochain agent reçoit l’enveloppe précédente injectée dans son prompt, pas un accès à la session de l’autre : chaque agent garde sa fenêtre de contexte, propre et bornée.
• notes_for_next_agent est le champ taillé pour cela : une consigne courte, écrite par un agent pour un agent, transportée par le code.
• La symétrie a une conséquence puissante : un résultat produit par du code peut être mis en forme d’enveloppe et tendu à un agent par la même porte : l’agent consommateur ne fait pas la différence, et c’est voulu. Vous l’exploiterez quand les gates renverront leurs verdicts.
• Le run.sessions du chapitre 8 est l’embryon de cette discipline : une session par phase agent, mémorisée par le code, jamais partagée entre agents.

Exemple concret

Projetez-vous au chapitre 11 : un scout explore Plume en lecture seule, puis un planner rédige la spec. Le scout termine avec une enveloppe : summary en une phrase, artifacts pointant un fichier de notes, notes_for_next_agent signalant le fichier à ne pas toucher. Le prompt du planner reçoit cette enveloppe injectée : quelques centaines de tokens. L’alternative, rejouer au planner toute la conversation du scout, en pèserait plusieurs dizaines de milliers : un facteur 100 environ, à chaque couture de chaque run. Le handoff, c’est la compression du contexte par le contrat : on ne transmet pas ce que l’agent a vécu, on transmet ce qu’il a déclaré.

Ce qui traverse la couture, et ce qui ne la traverse pas

Script — le prompt du prochain agent

Cette mécanique ne touche pas le harnais : elle se joue avant le port, côté Python. Une seule version suffit donc, valable pour pi comme pour Claude Code. Voici le geste que tous les ADW du module vont répéter, dès le scout et le planner du chapitre 11 :

# La passation : l'enveloppe de la phase precedente entre dans le prompt
# de la suivante — par le code, jamais par une session partagee.
previous = run.results["scout"]          # l'enveloppe typee rendue par le scout
ask = (
    f"{demande}\n\n"
    f"{envelopes.handoff(previous)}\n\n"   # la feuille de quart, injectee
    f"{envelopes.contract(Envelope)}"      # le contrat de sortie de CE prompt
)

Piège courant : « le plus simple serait que les agents partagent une session » semble économique, c’est l’inverse. Une session partagée mélange les contextes : le planner hérite des tâtonnements du scout, la fenêtre gonfle, et chaque tour facture la relecture de tout ce bruit. Le handoff par enveloppe transmet peu mais transmet juste, et chaque agent travaille dans une fenêtre propre, bornée, moins chère.

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

La pièce du jour s’appelle adws/adw_modules/envelopes.py et se pose au cœur de la zone « squelette ADW », entre le port harnais du chapitre 7 et le runner du chapitre 8. adw_prompt.py est repris pour s’appuyer dessus. La couture est nette : le type, le parseur, la correction et le handoff sont du déterminisme pur, zéro token, et l’agent ne voit du contrat que deux textes générés par le code, la demande et l’éventuelle correction. Ce qui traverse désormais la couture a une forme nommée : une Envelope, validée à l’entrée, injectée à la sortie. À l’usage, la pièce ne coûte rien. Ce qu’elle économise, c’est la dérive silencieuse entre demande et vérification (un tour d’agent gaspillé par run, quelques milliers de tokens à chaque fois) et les sessions partagées qui font gonfler les fenêtres. Le chapitre 10 branchera dessus le roster et les permissions : qui a le droit de faire quoi, avec quel modèle.

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, deux fichiers : le module des enveloppes typées, et la version reprise d’adw_prompt.py qui s’appuie dessus. Le runner du chapitre 8 et le port du chapitre 7 ne changent pas.

Pièce — adws/adw_modules/envelopes.py

Le contrat de sortie des agents, en bibliothèque standard uniquement. Le module vit entièrement côté déterministe : il déclare le type, génère la demande (contract), valide la réponse (parse), rédige la correction en session vivante (correction) et prépare la passation (handoff). Comme harness.py et runner.py, c’est un module importé, sans en-tête PEP 723, mais il porte sa propre gate : lancé directement, il prouve son contrat.

"""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, parse() valide sa reponse — les deux lisent les memes
champs, la demande ne peut pas deriver de la verification.
"""
from __future__ import annotations

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


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. Les sous-types a venir (scout, plan, build...) etendront
    cette base avec leurs champs propres, et check() avec leurs invariants.
    """
    status: str = field(metadata={"ask": "'success' ou '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 contract(expected: type[Envelope] = Envelope) -> str:
    """La demande a coller dans le prompt — derivee du MEME type que parse()."""
    shape = {f.name: f.metadata.get("ask", "") for f in fields(expected)}
    return (
        "Termine ta reponse par un objet JSON, sans bloc markdown autour, "
        "avec exactement ces champs :\n"
        + json.dumps(shape, indent=2, ensure_ascii=False)
    )


def parse(text: str, expected: type[Envelope] = Envelope) -> Envelope:
    """Isole l'objet JSON, verifie le contrat, rend une enveloppe typee.

    Tolerant en entree (prose ou bloc de code autour de l'objet), strict sur
    le contrat : champs requis presents, invariants de check() respectes.
    Le bavardage hors contrat est ignore — le contrat, lui, ne se negocie pas.
    """
    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")

    spec = {f.name: f for f in fields(expected)}
    required = [name for name, f in spec.items()
                if f.default is MISSING and f.default_factory is MISSING]
    missing = [name for name in required if name not in payload]
    if missing:
        raise EnvelopeError(f"champs requis manquants : {missing}")

    envelope = expected(**{k: v for k, v in payload.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"
        "Reponds a nouveau avec UNIQUEMENT l'objet JSON demande.\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 : il prouve son propre contrat — zero token, instantane.
    # (Un test multi-lignes en ligne de commande dependrait du quoting du shell ;
    # la piece porte sa gate, la commande reste une ligne.)
    print(contract(Envelope))
    envelope = parse('bla {"status": "success", "summary": "ok"} bla')
    assert envelope.artifacts == [] and envelope.notes_for_next_agent == ""
    try:
        parse('{"status": "peut-etre"}')
    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() et parse() lisent le meme type")

Pièce — adws/adw_prompt.py

Cette version remplace celle du chapitre 8. Le workflow est identique, une phase agent et une phase code, mais le contrat vit désormais dans envelopes.py : la demande vient de contract(), la validation de parse(), et la correction cite le motif exact de l’échec au lieu d’un message générique. La session reste mémorisée avant la validation, comme hier.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# ///
"""adw_prompt — le plus petit ADW, desormais type de bout en bout.

Usage :
    uv run adws/adw_prompt.py "Votre demande" [--harness pi|claude] [--retries 2]

La phase agent envoie la demande par le port avec le contrat genere depuis le
type Envelope ; la phase code dispose sur l'enveloppe typee. Un echec de
contrat repart vers l'agent dans la MEME session, motif exact a l'appui.
"""
import argparse
import json
import sys
import uuid
from dataclasses import asdict

from adw_modules import envelopes, harness
from adw_modules.envelopes import Envelope, EnvelopeError
from adw_modules.runner import PhaseFailure, PhaseSpec, Run


def agent_phase(prompt: str, harness_name: str):
    """Fabrique l'action de la phase agent : la demande, puis les corrections."""
    last_motif = "enveloppe invalide"

    def action(run: Run, attempt: int) -> Envelope:
        nonlocal last_motif
        # Tentative 1 : la demande + le contrat. Ensuite : la correction seule,
        # avec le motif exact — l'agent a encore tout le contexte en session.
        ask = (f"{prompt}\n\n{envelopes.contract(Envelope)}" if attempt == 0
               else envelopes.correction(last_motif))
        request = harness.HarnessRequest(prompt=ask,
                                         session_id=run.sessions.get("prompt"))
        try:
            result = harness.run(harness_name, request)
        except harness.HarnessError as error:
            raise PhaseFailure(str(error)) from None
        # Memoriser la session AVANT de valider : un retry doit la poursuivre.
        run.sessions["prompt"] = result.session_id
        run.cost_usd += result.cost_usd
        try:
            return envelopes.parse(result.text, Envelope)
        except EnvelopeError as error:
            last_motif = str(error)
            raise PhaseFailure(f"enveloppe invalide : {error}") from None
    return action


def dispose(run: Run, attempt: int) -> Envelope:
    """Phase code : le code dispose — verdict deterministe, zero token."""
    envelope: Envelope = run.results["prompt"]
    if envelope.status != "success":
        raise PhaseFailure(f"l'agent declare lui-meme un echec : {envelope.summary!r}")
    print(json.dumps(asdict(envelope), indent=2, ensure_ascii=False))
    return envelope


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Le plus petit ADW : une demande entre, une enveloppe typee sort.")
    parser.add_argument("prompt", help="votre demande, en langage naturel")
    parser.add_argument("--harness", choices=sorted(harness.ADAPTERS), default="pi")
    parser.add_argument("--retries", type=int, default=2,
                        help="reprises de la phase agent, en session vivante")
    args = parser.parse_args()

    run = Run(adw_id=uuid.uuid4().hex[:8])
    return run.execute([
        PhaseSpec(name="prompt", kind="agent",
                  action=agent_phase(args.prompt, args.harness),
                  retries=args.retries),
        PhaseSpec(name="dispose", kind="code", action=dispose),
    ])


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

La gate du TP

# 1) le contrat se verifie sans agent — zero token, instantane
uv run adws/adw_modules/envelopes.py

# 2) le workflow complet, sur l'enveloppe typee
uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ? Reponds en une phrase."

Attendu : la première commande affiche le contrat généré, rejet motive : status doit valoir 'success' ou 'fail', puis contrat verifie : la preuve que demande et vérification sortent du même type, pour zéro token. La seconde rejoue le workflow d’hier, mais l’enveloppe affichée porte les quatre champs du type, pour ~1 à 3 centimes et 20 à 60 secondes sur le modèle par défaut du harnais (--harness claude pour l’autre dialecte). Quand les deux passent, commitez : ce qui traverse la couture a désormais une forme nommée.