Scout & Plan

Hier, vous avez arrêté la feuille de distribution : quatre agents déclarés dans un YAML validé, des périmètres d’écriture vérifiables par le code. Mais relisez le roster : aucun de ces agents n’a encore tourné. Le scout, le planner, leurs modèles choisis avec soin, leurs droits calibrés : tout cela n’est pour l’instant qu’une distribution affichée sur une porte de théâtre. Aujourd’hui, le rideau se lève. À la fin du chapitre, votre usine aura lancé ses deux premiers agents sous roster, un scout qui explore Plume en lecture seule et un planner qui transforme votre demande en plan, et produit son premier artefact à fort levier : une spec, posée dans specs/, que le builder du chapitre 12 implémentera. La pièce du jour branche enfin le roster sur le port, comme promis hier : harness.py apprend à traduire un AgentSpec en drapeaux, et deux nouveaux ADW, adw_scout.py et adw_plan.py, s’articulent sur le runner du chapitre 8 et les enveloppes du chapitre 9.

adw_scout : la reconnaissance en lecture seule

L’idée en une phrase

Le scout est le premier agent du roster à tourner : une phase agent bornée à la lecture (writes: [], vérifié par l’état des lieux), sur un modèle léger en thinking: low, qui rend une enveloppe typée de findings. La pièce adws/adw_scout.py qui l’encadre vit, elle, entièrement côté code déterministe.

Points clés

• Repérer n’est pas décider. Le scout cherche où vivent les choses et cite des chemins exacts. C’est pour cela que le roster lui donne un modèle léger et thinking: low : il rapporte, la réflexion coûteuse est réservée à ceux qui décident.
• Sa sortie est typée : ScoutEnvelope étend l’Envelope du chapitre 9 avec un champ findings, une liste d’objets file + note, et check() refuse tout finding sans chemin. Le contrat est déclaré une fois, demande et validation en dérivent.
• Son rapport vit dans le runtime : le scout écrit scout_findings.md sous data_dir, le seul endroit toujours ouvert. Son enveloppe le déclare dans artifacts, et le handoff du chapitre 9 le transporte vers l’agent suivant.
• La lecture seule se vérifie : snapshot() avant la phase, enforce() après. Un scout qui « corrige une coquille au passage » meurt avec le chemin nommé, comme au chapitre 10.
• Ne rien trouver est un résultat valide : une enveloppe success avec findings: [] et un summary qui le dit est un run vert. Un scout qui invente des chemins pour remplir sa liste est le vrai échec.

Exemple concret

Demandez « où vivent les tests de Plume, et qu’est-ce qu’ils couvrent ? ». Le scout lit le repo sur le modèle léger du roster, relevé du jour à ~0,07 $ le million de tokens en entrée. Quelques dizaines de milliers de tokens lus, une poignée écrite : moins d’un centime, 30 à 60 secondes. L’alternative, laisser le planner frontier faire sa propre reconnaissance, paie la même lecture au tarif de ~5 $ le million, soit un facteur 70 environ sur le prix du token lu, et pollue la fenêtre du planner avec les tâtonnements de l’exploration. Le scout, c’est la lecture au juste prix, compressée en une enveloppe de quelques centaines de tokens que le handoff injecte au suivant.

Le profil du scout, tel que le roster le déclare

Commande — ce que le port émet désormais

C’est la promesse du chapitre 10 tenue : les ADW nomment des agents, et l’adaptateur traduit l’AgentSpec dans le dialecte du harnais. Voici ce que le port émet pour le scout, dans les deux dialectes :

# version pi — les cles du roster se traduisent presque mot a mot
pi -p --mode json --model deepseek/deepseek-v4-flash-0731 \
  --thinking low --tools read,bash,grep,find,ls,write \
  --session-id <id> --session-dir adws/adw_data/sessions "<mission du scout>"

# version Claude Code — meme roster, autre vocabulaire : l'adaptateur retire le
# prefixe fournisseur, met les outils au format maison (ls est couvert par Bash),
# traduit le thinking en budget de reflexion (MAX_THINKING_TOKENS) — et la
# mission part par stdin : un argument multi-lignes ne survivrait pas aux
# shims .cmd de Windows
echo "<mission du scout>" | claude -p --output-format json \
  --model claude-opus-5 --allowedTools "Read,Bash,Grep,Glob,Write"

Vos scripts n’écrivent jamais ces lignes : ils disent scout, et le port parle.

Piège courant : « un scout, c’est une phase de plus, donc un coût de plus » compte à l’envers. La reconnaissance sera faite de toute façon, la seule question est par qui et à quel tarif. Sans scout, c’est le planner frontier qui lit le repo : mêmes tokens, prix multiplié, et sa fenêtre encombrée avant même de commencer à décider. Le scout n’ajoute pas une dépense, il déplace la lecture vers l’étage le moins cher.

adw_plan : le plan, l’artefact au plus fort levier

L’idée en une phrase

Le plan est l’artefact au plus fort levier de l’usine : quelques milliers de tokens qui gouvernent tout ce que le builder fera ensuite. adws/adw_plan.py enchaîne scout → planner par le handoff du chapitre 9, et la seule trace que le planner laisse au repo est une spec dans specs/, déclarée dans le champ spec_path de son enveloppe.

Points clés

• Le levier, c’est le rapport coût de correction / portée. Une erreur repérée dans la spec se corrige en la relisant : deux minutes, zéro token. La même erreur découverte au build coûte un re-run, découverte plus tard, des heures. Le plan est la dernière porte avant les tokens.
• C’est ici, et seulement ici, que le frontier se justifie : le roster donne au planner le modèle le plus cher et thinking: high, parce que le plan porte tout le run. Scout léger, planner frontier : le triangle coût-qualité du module 4, déjà à l’œuvre.
• PlanEnvelope ajoute spec_path : le chemin sous specs/ que le builder du chapitre 12 lira. check() exige qu’un run success pointe bien sous specs/, et la phase code vérifie que le fichier déclaré existe : première vérification de vérité, que les gates du chapitre 12 généraliseront.
• specs/ est la mémoire de l’usine : une spec ne s’écrase jamais, un nom déjà pris se suffixe. Le répertoire naît aujourd’hui, et le roster n’y autorise que le planner.
• Le handoff tourne en vrai : l’enveloppe du scout (findings, rapport, consignes) entre dans le prompt du planner par handoff(), quelques centaines de tokens au lieu d’une session partagée qui en pèserait des dizaines de milliers.

Exemple concret

Lancez adw_plan.py sur « ajoute un compteur de mots à l’éditeur de Plume ». Le run déroule sept phases, dont deux seulement sont des phases agent. Le scout lit Plume sur le léger : moins d’un centime. Le planner reçoit votre demande plus l’enveloppe du scout, réfléchit en thinking: high sur le frontier du roster (relevé du jour : ~5 $ le million en entrée, ~25 $ en sortie), lit une dizaine de milliers de tokens et en écrit quelques milliers : ~15 à 25 centimes. Total du run : ~20 à 30 centimes, 3 à 6 minutes, et un artefact durable dans specs/. La variante éco tient en une ligne : passez le model: du planner sur le workhorse du roster, et le même run tombe à ~5 centimes. À vous de juger si le plan y perd.

Où meurt une erreur : le levier chiffré

Commande — lancer le premier plan de l’usine

Une seule version suffit désormais, et c’est le progrès du jour : l’option --harness d’adw_prompt.py a disparu, car le roster décide du harnais de chaque agent. La même commande pilote pi ou Claude Code selon ce que déclare votre YAML :

# la reconnaissance seule — le scout, rien que le scout
uv run adws/adw_scout.py "Ou vivent les tests de Plume, et que couvrent-ils ?"

# la chaine complete : scout -> handoff -> planner -> une spec dans specs/
uv run adws/adw_plan.py "Ajoute un compteur de mots a l'editeur de Plume"

Piège courant : « le plan, c’est de la paperasse, autant laisser le builder se débrouiller » confond document et levier. Sans spec, chaque build redécouvre le contexte à ses frais et aucun run n’est comparable au précédent. Avec elle, le builder implémente au lieu d’explorer, le re-run devient possible, et vous disposez d’un point de relecture à coût nul avant de dépenser le moindre token de build. La spec n’est pas un rapport sur le travail : c’est l’outil qui le gouverne.

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

Trois fichiers rejoignent la zone « squelette ADW » du plan : adw_scout.py et adw_plan.py, les premiers vrais workflows de l’usine, et une évolution annoncée : harness.py version chapitre 11, dont les adaptateurs traduisent enfin le roster en drapeaux. Les voisins déjà posés travaillent tous : le runner séquence, les enveloppes typent, le roster distribue, l’état des lieux vérifie. La loi du livre s’applique à la lettre : sept phases dont cinq de pur code, deux nœuds agent bornés (le scout ne peut que lire, le planner ne peut écrire que sous specs/), et le contexte ne traverse qu’en enveloppes : ScoutEnvelope vers le code, son handoff vers le planner, PlanEnvelope vers la phase qui dispose. À l’usage : ~20 à 30 centimes et quelques minutes pour un plan complet (~5 centimes en variante éco), contre une session frontier interactive qui paierait la reconnaissance au prix fort et ne laisserait ni spec relisible ni run rejouable. Demain, chapitre 12 : le builder implémente spec_path, et les gates vérifient ce qu’il déclare.

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, trois fichiers : la version chapitre 11 du port (le roster traduit en drapeaux), le scout, et la chaîne de plan. Les sous-types d’enveloppe vivent dans l’ADW qui les rend : la base du chapitre 9 ne bouge pas, pas plus que le runner, le roster ou les permissions.

Pièce — adws/adw_modules/harness.py

Cette version remplace celle du chapitre 7. HarnessRequest gagne trois champs (model, thinking, tools) avec des défauts neutres : adw_prompt.py et hello_factory.py continuent de tourner sans modification. Chaque adaptateur traduit ces champs dans son dialecte, comme promis au chapitre 10 : la traduction vit ici, jamais dans vos scripts.

"""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 11 : la requete transporte le profil du roster (model,
thinking, tools) et chaque adaptateur le traduit dans son dialecte. Le
prompt de claude part par stdin — un argument multi-lignes ne survit pas
aux shims .cmd de Windows.
"""
from __future__ import annotations

import json
import os
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")

# 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}


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
    model: str | None = None         # provider/id du roster — None = defaut du harnais
    thinking: str | None = None      # off..max — None = defaut du harnais
    tools: tuple[str, ...] = ()      # allowlist du roster — () = outils par defaut


@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,
           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 = {**os.environ, **extra_env} if extra_env else None
    # 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")


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)
    cmd = ["pi", "-p", "--mode", "json",
           "--session-id", session_id, "--session-dir", str(SESSION_DIR)]
    # Le profil du roster, traduit presque mot a mot : c'est ici que vivait la
    # promesse du chapitre 10 — jamais dans vos scripts.
    if request.model:
        cmd += ["--model", request.model]        # pi accepte provider/id tel quel
    if request.thinking:
        cmd += ["--thinking", request.thinking]
    if request.tools:
        cmd += ["--tools", ",".join(request.tools)]
    cmd.append(request.prompt)
    proc = _spawn(cmd, 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.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:
        # 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/adw_scout.py

Le premier ADW sous roster. Tout ce qui entoure l’agent est du code : le roster fournit le profil, le constat d’entrée précède la phase agent, l’état des lieux la suit, et la phase finale dispose. Le sous-type ScoutEnvelope et la fabrique de phase agent_action vivent ici, et adw_plan.py les importera. Notez l’en-tête PEP 723 : première dépendance externe de l’usine, comme annoncé au chapitre 10.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["pyyaml"]
# ///
"""adw_scout — la reconnaissance en lecture seule.

Usage :
    uv run adws/adw_scout.py "Ou vivent les tests de Plume ?" [--config ...] [--retries 2]

Le premier ADW sous roster : le scout est declare dans factory.config.yaml,
le port traduit son profil en drapeaux, et l'etat des lieux verifie qu'il
n'a rien change. Sa sortie est une ScoutEnvelope : des findings types.
"""
import argparse
import json
import sys
import uuid
from dataclasses import asdict, dataclass, field
from pathlib import Path

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


@dataclass(frozen=True)
class ScoutEnvelope(Envelope):
    """Ce que le scout doit au code : des trouvailles nommees, ou rien."""
    findings: list = field(default_factory=list, metadata={
        "ask": "liste d'objets {'file': chemin exact, 'note': pourquoi il compte}, [] sinon"})

    def check(self) -> None:
        super().check()
        for finding in self.findings:
            if not isinstance(finding, dict) or not finding.get("file"):
                raise EnvelopeError(
                    "chaque finding doit etre un objet avec au moins un champ 'file'")


SCOUT_BRIEF = """Tu es le scout de l'usine : trouve ou vivent les choses, ne change RIEN.
- Lecture seule : cherche, lis, lance des commandes de lecture — n'ecris jamais dans le repo.
- Cite des chemins exacts, avec un indice de ligne quand c'est utile.
- Ecris tes trouvailles en markdown dans {report} pour l'agent qui te suivra,
  et declare ce fichier dans 'artifacts'.
- Ne rien trouver est un resultat valide : dis-le simplement."""


def agent_action(phase_name, agent, make_ask, expected):
    """Fabrique l'action d'une phase agent sous roster.

    L'AgentSpec fournit le harnais, le modele, le thinking et les outils ;
    le port les traduit en drapeaux. La demande part a la tentative 1, la
    correction motivee ensuite — dans la MEME session, comme au chapitre 9.
    """
    last_motif = "enveloppe invalide"

    def action(run: Run, attempt: int):
        nonlocal last_motif
        ask = make_ask(run) if attempt == 0 else envelopes.correction(last_motif, expected)
        request = harness.HarnessRequest(
            prompt=ask, session_id=run.sessions.get(phase_name),
            model=agent.model, thinking=agent.thinking, tools=agent.tools)
        try:
            result = harness.run(agent.harness, 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[phase_name] = result.session_id
        run.cost_usd += result.cost_usd
        try:
            return envelopes.parse(result.text, expected)
        except EnvelopeError as error:
            last_motif = str(error)
            raise PhaseFailure(f"enveloppe invalide : {error}") from None
    return action


def constat(phase_name, factory):
    """Phase code : le constat d'entree — et le dossier de run du runtime."""
    def action(run: Run, attempt: int):
        Path(factory.data_dir, "runs", run.adw_id).mkdir(parents=True, exist_ok=True)
        return permissions.snapshot(".")
    return action


def perimetre(phase_name, agent, factory):
    """Phase code : l'etat des lieux de sortie — la breche tue le run."""
    def action(run: Run, attempt: int):
        try:
            return permissions.enforce(".", agent, factory,
                                       run.results[f"constat_{phase_name}"])
        except PermissionBreach as breach:
            # Une breche n'est pas une gate : l'ecriture a deja eu lieu, on ne
            # re-prompte pas — la phase code meurt, et le run avec elle.
            raise PhaseFailure(str(breach)) from None
    return action


def report_path(factory, run: Run) -> str:
    return f"{factory.data_dir}/runs/{run.adw_id}/scout_findings.md"


def scout_ask(prompt, factory):
    """La mission du scout : le brief, votre demande, le contrat — dans cet ordre."""
    def make_ask(run: Run) -> str:
        return (SCOUT_BRIEF.format(report=report_path(factory, run))
                + f"\n\n### mission\n\n{prompt}\n\n"
                + envelopes.contract(ScoutEnvelope))
    return make_ask


def dispose(run: Run, attempt: int) -> ScoutEnvelope:
    """Phase code : le code dispose — verdict deterministe, zero token."""
    envelope: ScoutEnvelope = run.results["scout"]
    if envelope.status != "success":
        raise PhaseFailure(f"le scout 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="La reconnaissance en lecture seule, sous roster.")
    parser.add_argument("prompt", help="ce que le scout doit trouver")
    parser.add_argument("--config", default=str(roster.DEFAULT_PATH))
    parser.add_argument("--retries", type=int, default=2)
    args = parser.parse_args()

    factory = roster.load(args.config)          # zero token : tout echec est gratuit
    agent = factory.agents["scout"]
    run = Run(adw_id=uuid.uuid4().hex[:8])
    return run.execute([
        PhaseSpec(name="constat_scout", kind="code", action=constat("scout", factory)),
        PhaseSpec(name="scout", kind="agent",
                  action=agent_action("scout", agent, scout_ask(args.prompt, factory),
                                      ScoutEnvelope),
                  retries=args.retries),
        PhaseSpec(name="perimetre_scout", kind="code",
                  action=perimetre("scout", agent, factory)),
        PhaseSpec(name="dispose", kind="code", action=dispose),
    ])


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

Pièce — adws/adw_plan.py

La première chaîne de l’usine : scout → handoff → planner. Le script importe la mécanique posée dans adw_scout.py (un ADW est aussi un module) et n’ajoute que ce qui lui est propre : le sous-type PlanEnvelope, le brief du planner, et une phase code qui vérifie que la spec déclarée existe vraiment.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["pyyaml"]
# ///
"""adw_plan — le plan, l'artefact au plus fort levier.

Usage :
    uv run adws/adw_plan.py "Ajoute un compteur de mots a Plume" [--config ...] [--retries 2]

La premiere chaine de l'usine : le scout repere (lecture seule), le handoff
injecte son enveloppe au planner, le planner ecrit la spec sous specs/ —
la seule trace qu'il a le droit de laisser au repo.
"""
import argparse
import json
import sys
import uuid
from dataclasses import asdict, dataclass, field
from pathlib import Path

from adw_modules import envelopes, roster
from adw_modules.envelopes import Envelope, EnvelopeError
from adw_modules.runner import PhaseFailure, PhaseSpec, Run
from adw_scout import ScoutEnvelope, agent_action, constat, perimetre, scout_ask


@dataclass(frozen=True)
class PlanEnvelope(Envelope):
    """Ce que le planner doit au code : le chemin de la spec qu'il a ecrite."""
    spec_path: str = field(default="", metadata={
        "ask": "chemin de la spec ecrite sous specs/, '' si status=fail"})

    def check(self) -> None:
        super().check()
        if self.status == "success" and not self.spec_path.startswith("specs/"):
            raise EnvelopeError("spec_path doit pointer un fichier sous specs/")


PLANNER_BRIEF = """Tu es le planner de l'usine : transforme la demande en un plan que le
builder implementera sans poser de questions.
- Lis ce qu'il faut pour comprendre — l'enveloppe du scout ci-dessous t'evite
  l'exploration.
- Ecris le plan complet dans specs/{adw_id}-<deux-a-quatre-mots-kebab>.md :
  fichiers a toucher, changements a faire, comment verifier. Si le nom existe
  deja, suffixe -v2 : une spec ne s'ecrase JAMAIS.
- N'implemente rien : le plan est ta seule ecriture."""


def plan_ask(prompt, factory):
    """Le brief, la demande, la feuille de quart du scout, le contrat."""
    def make_ask(run: Run) -> str:
        scout_envelope: ScoutEnvelope = run.results["scout"]
        return (PLANNER_BRIEF.format(adw_id=run.adw_id)
                + f"\n\n### demande\n\n{prompt}\n\n"
                + envelopes.handoff(scout_envelope) + "\n\n"
                + envelopes.contract(PlanEnvelope))
    return make_ask


def dispose(run: Run, attempt: int) -> PlanEnvelope:
    """Phase code : la spec declaree doit exister et ne pas etre vide.

    Premiere verification de verite — le chapitre 12 la generalisera en gates.
    """
    envelope: PlanEnvelope = run.results["plan"]
    if envelope.status != "success":
        raise PhaseFailure(f"le planner declare lui-meme un echec : {envelope.summary!r}")
    spec = Path(envelope.spec_path)
    if not spec.is_file() or spec.stat().st_size == 0:
        raise PhaseFailure(f"spec declaree mais introuvable ou vide : {envelope.spec_path}")
    print(json.dumps(asdict(envelope), indent=2, ensure_ascii=False))
    print(f"spec posee : {envelope.spec_path}", file=sys.stderr)
    return envelope


def main() -> int:
    parser = argparse.ArgumentParser(
        description="La chaine scout -> planner : une demande entre, une spec sort.")
    parser.add_argument("prompt", help="la demande a planifier")
    parser.add_argument("--config", default=str(roster.DEFAULT_PATH))
    parser.add_argument("--retries", type=int, default=2)
    args = parser.parse_args()

    factory = roster.load(args.config)
    scout, planner = factory.agents["scout"], factory.agents["planner"]
    run = Run(adw_id=uuid.uuid4().hex[:8])
    return run.execute([
        PhaseSpec(name="constat_scout", kind="code", action=constat("scout", factory)),
        PhaseSpec(name="scout", kind="agent",
                  action=agent_action("scout", scout, scout_ask(args.prompt, factory),
                                      ScoutEnvelope),
                  retries=args.retries),
        PhaseSpec(name="perimetre_scout", kind="code",
                  action=perimetre("scout", scout, factory)),
        PhaseSpec(name="constat_plan", kind="code", action=constat("plan", factory)),
        PhaseSpec(name="plan", kind="agent",
                  action=agent_action("plan", planner, plan_ask(args.prompt, factory),
                                      PlanEnvelope),
                  retries=args.retries),
        PhaseSpec(name="perimetre_plan", kind="code",
                  action=perimetre("plan", planner, factory)),
        PhaseSpec(name="dispose", kind="code", action=dispose),
    ])


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

La gate du TP

# 1) la reconnaissance seule — le scout sous roster, moins d'un centime
uv run adws/adw_scout.py "Ou vivent les tests de Plume, et que couvrent-ils ?"

# 2) la chaine complete — une spec doit apparaitre dans specs/
uv run adws/adw_plan.py "Ajoute un compteur de mots a l'editeur de Plume" \
  && ls specs/

Attendu : le premier run est vert en 4 phases, avec une enveloppe dont les findings citent les fichiers de test de Plume, pour moins d’un centime et ~30 à 60 secondes. Le second déroule 7 phases et se termine par spec posee : specs/…, et ls specs/ montre le fichier : ~20 à 30 centimes, 3 à 6 minutes avec le planner frontier du roster (variante éco en une ligne : passez son model: sur le workhorse, ~5 centimes). Quand les deux passent, commitez, sans la spec : relisez-la d’abord, c’est tout l’intérêt du levier.