Anatomie d'un ADW & les phases

Au chapitre 7, vous avez enfermé les dialectes de pi et de Claude Code derrière le port harnais : un appel à run(), un HarnessResult en retour. Mais un aller-retour unique ne fabrique pas de logiciel : un vrai travail s’étale sur plusieurs étapes, comprendre, produire, vérifier. Qui décide de l’ordre de ces étapes ? Qui décide qu’on passe à la suivante, ou qu’on retente ? Aujourd’hui, vous donnez cette autorité à une pièce de code déterministe : le runner, le squelette sur lequel tous les workflows du livre vont s’articuler. À la fin du chapitre, vous saurez ce qu’est un AI Developer Workflow, pourquoi chaque phase vit d’un côté précis de la couture, et vous aurez lancé adw_prompt.py, le plus petit ADW complet, branché sur le port posé hier.

L’AI Developer Workflow

L’idée en une phrase

Un ADW (AI Developer Workflow) est un script Python déterministe qui orchestre une séquence nommée de phases pour produire un résultat vérifiable. C’est l’unité de travail de l’usine, et elle vit entièrement côté code de la couture : les agents n’y sont que des nœuds bornés, invoqués à travers le port du chapitre 7.

Points clés

• Un ADW est un script autonome lancé par uv run (chapitre 4), nommé par son intention : adw_prompt, puis adw_scout, adw_plan, adw_build, adw_sdlc au fil du module.
• Le séquencement appartient au code : l’ordre des phases, les reprises et l’arrêt sont écrits en Python, pas décidés par un agent. C’est la loi du livre appliquée à l’échelle d’un workflow entier : l’agent propose, le code dispose.
• Un ADW rend un code retour : 0 si le run est vert, autre chose sinon. Il s’enchaîne donc dans un justfile, un && de shell ou une CI, comme n’importe quel outil déterministe.
• Le plus petit ADW possible tient en deux phases : une phase agent qui envoie votre demande par le port, une phase code qui valide ce qui en revient. C’est adw_prompt.py, la pièce du jour.
• L’embryon du chapitre 3 faisait déjà tout cela, mais en dur, sans vocabulaire. L’ADW donne des noms : chaque étape devient une phase déclarée, mesurable, et bientôt traçable (module 5).

Exemple concret

Confiez « ajoute un compteur de mots à Plume » à un agent seul, en session interactive : une session d’un quart d’heure, quelques centaines de milliers de tokens, et un résultat qui dépend de votre vigilance : c’est vous qui avez validé, à l’œil. Le même travail passé par un ADW plan → build : 4 phases nommées, chacune bornée, un coût total de l’ordre de quelques dizaines de centimes sur un modèle intermédiaire, et surtout un code retour : le run est vert ou il ne l’est pas, que vous ayez regardé l’écran ou non. La différence n’est pas la vitesse : le second se relance à l’identique demain, s’enchaîne dans une recette just, et échoue proprement au lieu d’échouer silencieusement.

Agent seul vs ADW

Aspect	Agent seul en session	ADW
Qui séquence	l’agent, au fil de l’eau	le code, phases déclarées à l’avance
« Fini » décidé par	votre lecture de l’écran	un code retour, vérifiable en CI
Reproductibilité	faible — chaque session diverge	le même script, le même ordre, à chaque run
Reprise sur incident	tout relire, tout re-expliquer	retenter la phase en échec, session vivante

Commande — lancer le plus petit ADW

Une seule commande suffit pour les deux harnais : depuis le chapitre 7, le port absorbe leurs dialectes, et l’option --harness n’est plus qu’une donnée qui traverse l’ADW jusqu’à run() :

# le plus petit ADW : votre demande entre, une enveloppe validee ressort
uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ?"

# le meme workflow, l'autre harnais — rien d'autre ne change
uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ?" --harness claude

Piège courant : « un ADW, c’est juste un script qui appelle l’agent » passe à côté de l’essentiel : l’embryon du chapitre 3 appelait déjà un agent. Ce qui fait l’ADW, c’est ce qui entoure l’appel : des phases nommées dont le code possède l’ordre, un échec motivé qui arrête le run, et un code retour qu’une machine peut consommer. Sans cela, vous avez un wrapper, pas un workflow.

---

Phases agent et phases code

L’idée en une phrase

Une phase est l’unité atomique d’un ADW (un nom, une action, un verdict) et son kind déclare de quel côté de la couture elle s’exécute : code (déterministe, instantané, gratuit) ou agent (jugement, coûteux, variable). Le runner, lui, reste toujours côté code, même quand il pilote une phase agent.

Points clés

• Le kind détermine le coût : une phase code consomme zéro token, une phase agent loue du jugement au token. Le réflexe de conception : tout ce qui peut être une phase code doit l’être.
• Le succès se mérite : dans le runner, une phase qui lève une exception est en échec, et un run n’est vert que si toutes ses phases le sont. Le chemin par défaut est l’échec, jamais l’inverse.
• L’échec d’une phase agent est motivé, une PhaseFailure qui dit pourquoi, et le runner peut alors retenter dans la même session : la tentative suivante poursuit la conversation vivante au lieu de repartir de zéro. C’est la « correction en session vivante » du chapitre 3, devenue une mécanique du squelette.
• Le résultat d’une phase est rangé dans run.results sous le nom de la phase : la phase suivante peut le lire. C’est l’ancêtre direct du handoff que le chapitre 9 va typer avec les enveloppes.
• Retenter une phase code n’a pas de sens : même entrée, même sortie, le déterminisme ne change pas d’avis. Le runner ne retente que les phases agent.

Exemple concret

Lancez adw_prompt.py et supposez que l’agent, bavard, rende sa réponse sans l’objet JSON demandé. Sans runner : vous relancez tout, l’agent redécouvre votre demande dans une session neuve, et vous payez le run entier une deuxième fois. Avec le runner : la phase agent échoue avec un motif précis (« aucun objet JSON dans la réponse »), et la tentative suivante envoie la correction seule dans la même session. L’agent a encore tout le contexte, il ne lui manque que le format. Ordre de grandeur : la correction coûte quelques milliers de tokens, contre plusieurs dizaines de milliers pour un redémarrage à froid, un facteur 10 environ pour une ligne de code dans le runner.

Les deux kinds de phases

Aspect	Phase code	Phase agent
Coût	zéro token, gratuit	loué au token — centimes à dizaines de centimes
Durée	millisecondes	dizaines de secondes à minutes
Reproductibilité	totale	variable d’un run à l’autre
Échec typique	contrat non respecté, refus motivé	enveloppe invalide, réponse hors sujet
Reprise	inutile — même entrée, même sortie	retry en session vivante

Script — le cœur du runner

Le contrat tient en deux types : un ADW déclare des PhaseSpec, le runner les exécute dans l’ordre et s’arrête à la première phase définitivement en échec (le fichier complet est dans les travaux pratiques) :

@dataclass(frozen=True)
class PhaseSpec:
    """Ce qu'un ADW declare : un nom, un cote de la couture, une action."""
    name: str
    kind: str                              # "agent" ou "code"
    action: Callable[["Run", int], Any]    # (run, tentative) -> resultat
    retries: int = 0                       # phases agent : reprises en session vivante

class PhaseFailure(Exception):
    """L'echec motive d'une phase — le runner decide s'il retente."""

Piège courant : « en cas d’échec, il suffit de relancer le script » confond deux reprises très différentes. Relancer le script, c’est un redémarrage à froid : session neuve, contexte perdu, coût plein. Le retry du runner est une poursuite de session : l’agent qui a produit le presque-bon reçoit la correction dans le contexte qui l’a produit. Le premier jette ce que vous avez déjà payé, le second le réutilise.

---

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

La pièce du jour ouvre la zone « Le squelette ADW » du plan : adws/adw_modules/runner.py, posé juste à côté du port harnais du chapitre 7 sur lequel il s’appuie, et adws/adw_prompt.py, le premier workflow qui s’articule dessus. La couture est nette : le runner est du déterminisme pur (séquencement, verdicts, reprises, zéro token) et l’agent ne vit qu’à l’intérieur d’une phase agent, derrière le port. Ce qui traverse : votre demande dans un sens, une enveloppe JSON minimale (status, summary) dans l’autre, que le chapitre 9 va promouvoir en enveloppes typées. Un run d’adw_prompt coûte quelques centimes et moins d’une minute. Ce que le squelette économise, c’est le redémarrage à froid à chaque accroc (la correction en session vivante coûte un facteur 10 de moins), et il prépare la suite : scout, plan et build ne seront que des listes de PhaseSpec de plus en plus riches sur ce même runner.

---

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 squelette (runner.py) et le premier ADW qui s’articule dessus (adw_prompt.py). L’embryon hello_factory.py des chapitres 3 et 7 reste en place, il n’est pas modifié.

Pièce — adws/adw_modules/runner.py

Le squelette de l’usine, bibliothèque standard uniquement. Il vit entièrement côté déterministe : il ne connaît ni pi, ni claude, ni même le port : il séquence des actions, mesure, retente et rend un verdict. Comme harness.py, c’est un module importé, sans en-tête PEP 723.

"""runner — le squelette de l'usine : phases, sequencement, retries.

Un ADW declare ses phases ; le runner les execute dans l'ordre, mesure,
retente les phases agent en session vivante, et rend un code retour.
Le succes se merite : toute PhaseFailure marque la tentative en echec,
et un run n'est vert que si toutes ses phases le sont.
"""
from __future__ import annotations

import sys
import time
from dataclasses import dataclass, field
from typing import Any, Callable


class PhaseFailure(Exception):
    """L'echec motive d'une phase — le runner decide s'il retente."""


@dataclass(frozen=True)
class PhaseSpec:
    """Ce qu'un ADW declare : un nom, un cote de la couture, une action."""
    name: str
    kind: str                              # "agent" ou "code"
    action: Callable[["Run", int], Any]    # (run, tentative) -> resultat
    retries: int = 0                       # phases agent : reprises en session vivante


@dataclass
class Run:
    """L'etat partage d'un run : resultats des phases, sessions, cout."""
    adw_id: str
    results: dict[str, Any] = field(default_factory=dict)
    sessions: dict[str, str] = field(default_factory=dict)  # phase -> session_id
    cost_usd: float = 0.0

    def execute(self, phases: list[PhaseSpec]) -> int:
        """Sequence les phases declarees. Arret a la premiere phase en echec definitif."""
        if not phases:
            print("aucune phase declaree — un ADW vide n'est pas un ADW", file=sys.stderr)
            return 1
        for spec in phases:
            if spec.kind not in ("agent", "code"):
                print(f"[{self.adw_id}] {spec.name} : kind inconnu {spec.kind!r}",
                      file=sys.stderr)
                return 1
            if not self._run_phase(spec):
                print(f"[{self.adw_id}] ECHEC en phase {spec.name} — arret du run",
                      file=sys.stderr)
                return 1
        print(f"[{self.adw_id}] run vert — cout total ~{self.cost_usd:.4f} $",
              file=sys.stderr)
        return 0

    def _run_phase(self, spec: PhaseSpec) -> bool:
        # Retenter une phase code n'a pas de sens : meme entree, meme sortie.
        # Seules les phases agent ont droit aux reprises — en session vivante.
        attempts = 1 + (spec.retries if spec.kind == "agent" else 0)
        for attempt in range(attempts):
            clock = time.monotonic()
            label = f"{spec.name} ({spec.kind}, tentative {attempt + 1}/{attempts})"
            try:
                # L'action recoit le run (etat partage) et le numero de tentative :
                # a la tentative 1, une phase agent envoie la demande ; ensuite,
                # elle envoie la correction dans la MEME session.
                self.results[spec.name] = spec.action(self, attempt)
            except PhaseFailure as error:
                print(f"[{self.adw_id}] {label} : echec — {error}", file=sys.stderr)
                continue
            duration = time.monotonic() - clock
            print(f"[{self.adw_id}] {label} : OK en {duration:.1f} s", file=sys.stderr)
            return True
        return False

Pièce — adws/adw_prompt.py

Le plus petit ADW complet : une phase agent qui envoie votre demande par le port et exige une enveloppe JSON minimale, une phase code qui dispose. En cas d’enveloppe invalide, le retry du runner renvoie la correction dans la même session, et vous voyez la mécanique du chapitre entier tourner en une commande. Il s’appuie sur harness.py (chapitre 7) et sur le runner ci-dessus.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# ///
"""adw_prompt — le plus petit ADW : une phase agent, une phase code.

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

La phase agent envoie la demande par le port et exige une enveloppe JSON
minimale (status, summary) ; la phase code la valide et l'affiche. Un echec
de format repart vers l'agent dans la MEME session — correction vivante.
"""
import argparse
import json
import sys
import uuid

from adw_modules import harness
from adw_modules.runner import PhaseFailure, PhaseSpec, Run

# L'exigence de format, ajoutee a votre demande : l'enveloppe minimale du jour.
ENVELOPE_ASK = (
    "Termine ta reponse par un objet JSON, sans bloc markdown autour, avec "
    "exactement ces champs : 'status' ('success' ou 'fail') et 'summary' "
    "(une phrase sur ce que tu as fait ou trouve)."
)

# La correction envoyee en session vivante : le format seul, pas la demande.
FIX_ASK = (
    "Ta derniere reponse ne contenait pas l'objet JSON demande. Reponds a "
    "nouveau avec UNIQUEMENT l'objet JSON : champs 'status' et 'summary'."
)


def extract_envelope(text: str) -> dict:
    """Isole et valide l'enveloppe minimale : status + summary, rien d'obscur."""
    start, end = text.find("{"), text.rfind("}")
    if start == -1 or end <= start:
        raise ValueError("aucun objet JSON dans la reponse")
    payload = json.loads(text[start:end + 1])
    missing = {"status", "summary"} - payload.keys()
    if missing:
        raise ValueError(f"champs manquants : {sorted(missing)}")
    if payload["status"] not in ("success", "fail"):
        raise ValueError("status doit valoir 'success' ou 'fail'")
    return payload


def agent_phase(prompt: str, harness_name: str):
    """Fabrique l'action de la phase agent : la demande, puis les corrections."""
    def action(run: Run, attempt: int) -> dict:
        ask = f"{prompt}\n\n{ENVELOPE_ASK}" if attempt == 0 else FIX_ASK
        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 extract_envelope(result.text)
        except (ValueError, json.JSONDecodeError) as error:
            raise PhaseFailure(f"enveloppe invalide : {error}") from None
    return action


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


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Le plus petit ADW : une demande entre, une enveloppe validee 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

uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ? Reponds en une phrase."

Attendu : la phase prompt en OK (avec sa durée), la phase dispose qui affiche l’enveloppe (status: success et un summary citant les fichiers de Plume), puis run vert avec le coût : ~1 à 3 centimes, 20 à 60 secondes sur le modèle par défaut du harnais. Ajoutez --harness claude pour vérifier l’autre dialecte si vous l’avez installé. Quand la gate passe, commitez : le squelette est posé, et le chapitre 9 pourra typer ce qui le traverse.

---