Disposer pendant la phase — coupe-circuit et masquage
Quatre compteurs coupent une phase pi qui dérape (tours, appels d'outils, coût, boucle) et un filtre masque les secrets avant l'envoi au fournisseur. Comparé aux drapeaux et aux hooks de Claude Code. La pièce du jour : factory-guard.ts.
Hier, l’enveloppe est arrivée par report_phase. Mais pendant la phase, le code ne possède
qu’un mur : HarnessRequest.timeout = 600. Ce chapitre ajoute deux protections dans une
extension pi, factory-guard.ts. Quatre compteurs coupent une phase qui dérape, et un filtre
masque les secrets avant l’envoi au fournisseur. Le roster et la fabrique agent_action évoluent
avec elle.
Coupe-circuit : tours, outils, coût, boucle
Un builder lance bun test, et un test échoue. Le builder relance la même commande, obtient le
même résultat, et recommence à chaque tour. Chaque tour relit 50 à 80 k jetons de contexte. Le
mur de 600 s tombe après quinze à vingt tours, avec un à deux dollars brûlés et une
HarnessError sans motif.
Un coupe-circuit arrête ce builder plus tôt. C’est un jeu de compteurs tenus par le code dans
une extension pi, avec des seuils fixés par le roster. Ici, au cinquième bash(bun test)
identique, l’appel est refusé et le tour s’arrête, pour zéro jeton de plus.
Quatre compteurs, deux gestes
| Compteur | Événement pi | Seuil (variable) | Geste | Ce que voit le runner |
|---|---|---|---|---|
| Tours | turn_end | FACTORY_LIMIT_TURNS | ctx.abort() | stopReason: aborted → HarnessError + session |
| Appels d’outils | tool_call | FACTORY_LIMIT_TOOL_CALLS | block + terminate | idem, sans appel LLM de plus |
| Coût cumulé | message_end (assistant) | FACTORY_LIMIT_COST_USD | ctx.abort() | idem |
| Boucle | tool_call (haché dans une fenêtre) | FACTORY_LOOP_WINDOW | FACTORY_LOOP_THRESHOLD | block + terminate | idem, motif « × N mêmes arguments » |
| Délai par appel | tool_call de bash | FACTORY_TOOL_TIMEOUT_S | injection de timeout | un bash muet rend une erreur d’outil, la phase continue |
Pour la boucle, l’extension hache chaque appel (outil et arguments triés) et compte les répétitions dans une fenêtre glissante. Cinq fois le même appel parmi les vingt derniers est une boucle. Le mur de temps reste côté Python, en dernier recours.
Le coupe-circuit ne se déclenche qu’une fois, par l’un de deux gestes :
- Tours ou coût :
ctx.abort(). Le message suivant sort enstopReason: "aborted", que l’adaptateur (A6) transforme enHarnessErrorporteuse de la session. - Boucle ou plafond d’appels :
block: true, terminate: truedanstool_call. L’appel est refusé et le tour s’arrête sans appel LLM de plus.
Côté Claude Code : claude -p --max-turns plafonne les tours et --max-budget-usd le coût,
lui aussi estimé côté client. Un hook PreToolUse refuse un appel avec
permissionDecision: "deny", et Claude lit la raison du refus.
Les seuils et le motif
Le runner transmet les limits: du roster à pi en variables FACTORY_LIMIT_*, jamais par le
prompt. En session interactive, sans ces variables, seuls la détection de boucle et le délai par
appel restent armés.
Le motif remonte par une ligne guard dans le journal harnais (A4) et par un rapport de garde
(FACTORY_GUARD_REPORT). agent_action lit ce rapport et relance dans la même session : « ta
tentative a été interrompue : bash appelé 5 fois avec les mêmes arguments, lis la sortie du test
au lieu de le relancer ».
Le délai par appel
L’outil bash de pi accepte un timeout en secondes, sans valeur par défaut, et le modèle ne le
renseigne jamais. Un bun test qui attend une entrée bloque donc la phase jusqu’au mur.
L’extension injecte ce champ quand il manque (FACTORY_TOOL_TIMEOUT_S, 300 par défaut).
Côté Claude Code : Bash a un délai par défaut de deux minutes (BASH_DEFAULT_TIMEOUT_MS). Un
hook PreToolUse peut réécrire les arguments d’un outil avec updatedInput.
Config — les limites dans le roster
Ce bloc s’ajoute à un agent ou aux defaults:. Sans lui, les valeurs par défaut de
roster.py s’appliquent.
agents:
- name: builder
purpose: Implementer le plan, exactement ; declarer chaque fichier modifie.
model: z-ai/glm-5.3 # workhorse : ~1,40 $/M entree, ~4,40 $/M sortie (2026-09-04)
thinking: high
tools: [read, bash, grep, find, ls, edit, write]
# Le coupe-circuit (annexe A7) : transmis a pi par variables FACTORY_LIMIT_*,
# jamais par le prompt. Plafonnez large : ce sont des filets, pas des objectifs.
limits:
turns: 40 # tours LLM avant abort — un build de Plume en prend 10 a 25
tool_calls: 120 # appels d'outils avant refus — trois par tour, en moyenne
cost_usd: 0.80 # cout cumule (catalogue pi, pas la facture) avant abort
loop_window: 20 # la fenetre glissante d'appels observes
loop_threshold: 5 # le meme appel N fois dans la fenetre = boucle
tool_timeout_s: 300 # injecte dans chaque bash sans timeout explicite
timeout: 900 # le mur de temps, cote Python — remplace le 600 fixeCommande — lire un arrêt dans la trace
sqlite3 adws/adw_data/factory.db "SELECT ts, name, json_extract(payload_json, '$.reason') FROM harness_by_phase WHERE type = 'guard' ORDER BY ts DESC LIMIT 5;"Chaque ligne guard nomme le compteur (armed à l’ouverture, puis tours, outils, cout
ou boucle) et son motif. Sans sqlite3, uv run adws/adw_modules/harness_trace.py --last
affiche la même chose.
Piège courant : un plafond d’appels d’outils serré n’économise rien. C’est le tour qui coûte, pas l’appel d’outil. Un plafond trop bas coupe un builder en pleine série d’
editet vous fait payer une reprise. Gardez-le large, de l’ordre de trois appels par tour.
Filtrer ce qui sort des outils (tool_result)
Un scout cherche où est configurée la clé de la passerelle. Il lance
grep -rn OPENROUTER . --include=*. Si votre .gitignore a une faille, la sortie contient
.env avec la vraie clé. Elle part au fournisseur, puis repart à chaque tour tant qu’elle reste
dans le contexte.
L’événement tool_result de pi est le seul point où ce contenu est produit et pas encore envoyé.
L’extension y remplace tout ce qui ressemble à un secret (sk-…, OPENROUTER_API_KEY=…,
Bearer …, blocs PEM) par «redacted». Le reste passe intact. Le scout voit
OPENROUTER_API_KEY=«redacted», finit son travail, et le journal porte redact / bash / 1.
Coût : zéro jeton.
Comment tool_result réécrit un résultat
tool_result fonctionne comme un intergiciel (middleware). Les extensions passent dans l’ordre
de chargement. Chacune voit le résultat de la précédente et peut rendre un correctif partiel
(content, details, isError, usage). Un champ omis garde sa valeur.
Chaque masquage écrit une ligne redact dans le journal, avec l’outil et le nombre de motifs,
jamais le secret. Le reste, pi le fait déjà. Les outils intégrés tronquent leur sortie à quelques
dizaines de kilo-octets, et bash rend une erreur d’outil sur un code de sortie non nul.
Côté Claude Code : un hook PostToolUse remplace la sortie d’un outil avant que Claude ne la
voie, avec updatedToolOutput. La valeur doit respecter la forme de sortie de l’outil (pour
Bash, stdout, stderr, interrupted et isImage). La télémétrie capture quand même la sortie
originale.
L’usine ne pose ce filtre que sur son nœud pi, celui qui sert les modèles ouverts via la passerelle.
Trois remparts, trois moments
| Rempart | Événement | Ce qu’il juge | Ce qu’il laisse passer |
|---|---|---|---|
| Damage control (A3) | tool_call | la commande avant exécution | une commande sûre dont la sortie contient un secret |
| Masquage (A7) | tool_result | le contenu avant envoi au fournisseur | un secret déjà dans le contexte par un autre canal |
| Fichiers protégés + état des lieux (ch. 10) | après la phase | les écritures dans le dépôt | tout ce qui n’est pas une écriture |
Côté Claude Code : une règle deny Read(./.env) tient le rôle de damage control. Elle
bloque cat .env, mais pas un grep -r pattern . qui lit le fichier sans le nommer.
Script — l’intergiciel, en dix lignes
// .pi/extensions/factory-guard.ts — extrait : masquer avant d'envoyer, journaliser sans le secret.
pi.on("tool_result", async (event, ctx) => {
let hits = 0;
const content = event.content.map((part) => {
if (part.type !== "text") return part; // images, etc. : intactes
const { text, count } = redact(part.text); // fonction pure, testée à sec
hits += count;
return count > 0 ? { ...part, text } : part;
});
if (hits === 0) return; // rien à changer : pas de correctif
journal(ctx, "redact", event.toolName, { hits }); // le nombre, jamais le motif
return { content }; // correctif partiel : isError, details inchangés
});Piège courant : le masquage ne double pas damage control. Damage control juge une commande (
cat .env), le masquage juge un contenu (la sortie d’ungrep -rlégitime). Aucun des deux ne protège un secret collé dans un prompt ou unAGENTS.md, qui part au fournisseur avant tout outil.
pi et Claude Code côte à côte
| pi (extension de l’usine) | Claude Code | |
|---|---|---|
| Plafond de tours | turn_end puis ctx.abort() | claude -p --max-turns |
| Plafond de coût | usage.cost du catalogue, estimé | claude -p --max-budget-usd, estimé |
| Refuser un appel d’outil | tool_call rend block | hook PreToolUse, permissionDecision: "deny" |
| Délai d’une commande shell | aucun par défaut, injecté par l’extension | deux minutes par défaut |
| Réécrire la sortie d’un outil | tool_result, correctif partiel | hook PostToolUse, updatedToolOutput |
Interdire la lecture de .env | damage control (A3) | règle deny Read(./.env) |
Fil rouge — la pièce posée aujourd’hui
La pièce du jour est la quatrième extension du socle pi, après damage-control.ts (A3),
factory-obs.ts (A4) et factory-report.ts (A6). La couture ne bouge pas : le runner possède le
graphe. Le code disposait après la phase. Il dispose désormais aussi pendant, par des seuils que
l’agent ne négocie pas, pour zéro jeton.
Retenez deux points. Le coût cumulé vient du catalogue de pi, et la facture de la passerelle fait foi. La
phase build de adw_sdlc.py garde son mur de 600 s jusqu’à son alignement, plus loin dans
l’annexe. D’ici là, le coupe-circuit y tourne avec les valeurs par défaut de roster.py.
Travaux pratiques — la pièce du jour
Trois fichiers à poser dans plume-factory : une extension du socle et deux modules Python qui
remplacent leurs versions précédentes. factory.config.yaml ne change pas.
Pièce — .pi/extensions/factory-guard.ts
Ce module porte les deux sous-thèmes. readLimits, LoopDetector et redact sont purs et
exportés, et la gate sous Bun les teste. Les seuls effets de bord sont le journal harnais et le
rapport de garde. Aucune dépendance npm à l’exécution.
// .pi/extensions/factory-guard.ts — le code dispose PENDANT la phase.
// Quatre compteurs (tours, appels d'outils, coût cumulé, boucle) coupent le nœud pi au premier
// seuil franchi ; un intergiciel tool_result masque les secrets avant qu'ils ne partent au
// fournisseur. Les seuils viennent du roster par variables FACTORY_* (jamais par le prompt) ;
// le motif remonte au runner par le journal harnais (A4) et par un rapport de garde.
import { createHash, randomUUID } from "node:crypto";
import { appendFileSync, mkdirSync, writeFileSync } from "node:fs";
import { dirname, join } from "node:path";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
const HARNESS_DIR = join("adws", "adw_data", "traces", "harness"); // le journal du chapitre A4
export const REPORT_ENV = "FACTORY_GUARD_REPORT"; // le rapport de garde lu par agent_action
// ---------- les seuils : depuis l'environnement posé par le runner, jamais depuis le prompt ----------
export interface Limits {
turns: number; // 0 = pas de plafond (session de poste)
toolCalls: number; // 0 = pas de plafond
costUsd: number; // 0 = pas de plafond
loopWindow: number; // taille de la fenêtre glissante d'appels observés
loopThreshold: number; // le même appel N fois dans la fenêtre = boucle
toolTimeoutS: number; // injecté dans chaque bash sans timeout explicite
}
function num(value: string | undefined, fallback: number): number {
const parsed = value === undefined || value === "" ? NaN : Number(value);
return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback;
}
export function readLimits(env: Record<string, string | undefined>): Limits {
return {
turns: num(env.FACTORY_LIMIT_TURNS, 0),
toolCalls: num(env.FACTORY_LIMIT_TOOL_CALLS, 0),
costUsd: num(env.FACTORY_LIMIT_COST_USD, 0),
loopWindow: Math.max(2, num(env.FACTORY_LOOP_WINDOW, 20)),
loopThreshold: Math.max(2, num(env.FACTORY_LOOP_THRESHOLD, 5)),
toolTimeoutS: num(env.FACTORY_TOOL_TIMEOUT_S, 300),
};
}
// ---------- la boucle : un haché (outil, arguments triés) dans une fenêtre glissante ----------
function canonical(value: unknown): string {
if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`;
if (value && typeof value === "object") {
const record = value as Record<string, unknown>;
return `{${Object.keys(record).sort().map((k) => `${JSON.stringify(k)}:${canonical(record[k])}`).join(",")}}`;
}
return JSON.stringify(value);
}
export function fingerprint(toolName: string, input: unknown): string {
return createHash("sha1").update(`${toolName}\n${canonical(input)}`).digest("hex").slice(0, 16);
}
export class LoopDetector {
private readonly recent: string[] = [];
constructor(readonly window: number, readonly threshold: number) {}
/** Observe un appel ; rend le nombre d'occurrences de ce même appel dans la fenêtre (lui compris). */
observe(toolName: string, input: unknown): number {
const key = fingerprint(toolName, input);
this.recent.push(key);
if (this.recent.length > this.window) this.recent.shift();
return this.recent.filter((k) => k === key).length;
}
looping(toolName: string, input: unknown): boolean {
return this.observe(toolName, input) >= this.threshold;
}
}
// ---------- le masquage : ce qui ressemble à un secret ne quitte pas la machine ----------
export const REDACTED = "«redacted»";
// Cinq motifs, volontairement larges : mieux vaut masquer un faux positif qu'envoyer une clé.
const SECRET_PATTERNS: RegExp[] = [
/-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----/g, // blocs PEM
/\bsk-[A-Za-z0-9_-]{16,}/g, // clés sk-…, sk-or-v1-…
/\b(?:ghp|gho|github_pat)_[A-Za-z0-9_]{20,}/g, // jetons GitHub
/\bAKIA[0-9A-Z]{16}\b/g, // identifiants AWS
/\b([A-Z0-9_]*(?:KEY|TOKEN|SECRET|PASSWORD))(\s*[=:]\s*)["']?([^\s"']{8,})["']?/g, // VAR=valeur
/\b(Bearer\s+)[A-Za-z0-9._~+/-]{20,}=*/g, // en-têtes Authorization
];
export function redact(text: string): { text: string; count: number } {
let count = 0;
let out = text;
for (const pattern of SECRET_PATTERNS) {
out = out.replace(pattern, (match: string, ...groups: unknown[]) => {
count += 1;
// Motif VAR=valeur : garder le nom et le signe, masquer la valeur seule.
if (typeof groups[0] === "string" && typeof groups[1] === "string" && typeof groups[2] === "string") {
return `${groups[0]}${groups[1]}${REDACTED}`;
}
if (typeof groups[0] === "string" && match.startsWith(groups[0])) return `${groups[0]}${REDACTED}`;
return REDACTED;
});
}
return { text: out, count };
}
// ---------- le journal (A4) et le rapport de garde : les deux canaux du motif ----------
function journalLine(cwd: string, sessionId: string, type: string, name: string, payload: Record<string, unknown>): void {
const dir = join(cwd, HARNESS_DIR);
mkdirSync(dir, { recursive: true });
const line = {
event_id: `hev_${randomUUID().replace(/-/g, "").slice(0, 12)}`,
ts: new Date().toISOString(),
session_id: sessionId,
type,
name,
payload,
};
appendFileSync(join(dir, `${sessionId}.jsonl`), `${JSON.stringify(line)}\n`, "utf8");
}
export interface GuardReport {
kind: "tours" | "outils" | "cout" | "boucle";
reason: string;
session_id: string;
counters: { turns: number; tool_calls: number; cost_usd: number };
ts: string;
}
function writeReport(path: string | undefined, report: GuardReport): void {
if (!path) return;
mkdirSync(dirname(path), { recursive: true });
writeFileSync(path, JSON.stringify(report, null, 2), "utf8");
}
// ---------- l'extension : compter, couper une fois, masquer toujours ----------
export default function (pi: ExtensionAPI) {
const limits = readLimits(process.env);
const counters = { turns: 0, tool_calls: 0, cost_usd: 0 };
let detector = new LoopDetector(limits.loopWindow, limits.loopThreshold);
let sessionId = "";
let cwd = ".";
let tripped = false;
// Un seul déclenchement : journal, rapport, entrée de session — puis le geste choisi par l'appelant.
const trip = (kind: GuardReport["kind"], reason: string): void => {
if (tripped) return;
tripped = true;
const report: GuardReport = { kind, reason, session_id: sessionId, counters: { ...counters }, ts: new Date().toISOString() };
journalLine(cwd, sessionId, "guard", kind, { reason, ...counters });
writeReport(process.env[REPORT_ENV], report);
pi.appendEntry("factory-guard", report); // persiste dans la session ; hors contexte LLM
};
pi.on("session_start", async (_event, ctx) => {
sessionId = ctx.sessionManager.getSessionId();
cwd = ctx.cwd;
counters.turns = counters.tool_calls = counters.cost_usd = 0;
detector = new LoopDetector(limits.loopWindow, limits.loopThreshold);
tripped = false;
journalLine(cwd, sessionId, "guard", "armed", { ...limits, headless: !ctx.hasUI });
if (ctx.hasUI) ctx.ui.setStatus("factory-guard", "⏚ garde");
});
// Tours : un abort — le message suivant sort en stopReason "aborted", l'adaptateur v4 le lit.
pi.on("turn_end", async (_event, ctx) => {
counters.turns += 1;
if (limits.turns > 0 && counters.turns >= limits.turns && !tripped) {
trip("tours", `${counters.turns} tours atteints (plafond ${limits.turns})`);
ctx.abort();
}
});
// Coût cumulé : le catalogue de pi, pas la facture — une estimation suffit pour un disjoncteur.
pi.on("message_end", async (event, ctx) => {
const message = event.message as { role: string; usage?: { cost?: { total?: number } } };
if (message.role !== "assistant") return;
counters.cost_usd += message.usage?.cost?.total ?? 0;
if (limits.costUsd > 0 && counters.cost_usd >= limits.costUsd && !tripped) {
trip("cout", `${counters.cost_usd.toFixed(3)} $ cumulés (plafond ${limits.costUsd} $)`);
ctx.abort();
}
});
// Appels d'outils : délai injecté, boucle et plafond — refusés AVANT exécution, tour terminé.
pi.on("tool_call", async (event) => {
if (tripped) return { block: true, reason: "[garde] phase interrompue", terminate: true };
counters.tool_calls += 1;
const input = event.input as Record<string, unknown>;
if (event.toolName === "bash" && limits.toolTimeoutS > 0 && input.timeout === undefined) {
input.timeout = limits.toolTimeoutS; // le modèle ne le renseigne jamais ; le code le fait pour lui
}
if (detector.looping(event.toolName, input)) {
const reason = `[boucle] ${event.toolName} appelé ${limits.loopThreshold} fois avec les mêmes arguments dans les ${limits.loopWindow} derniers appels`;
trip("boucle", reason);
return { block: true, reason, terminate: true };
}
if (limits.toolCalls > 0 && counters.tool_calls > limits.toolCalls) {
const reason = `[outils] ${counters.tool_calls} appels d'outils (plafond ${limits.toolCalls})`;
trip("outils", reason);
return { block: true, reason, terminate: true };
}
return undefined;
});
// Masquage : ce qui ressemble à un secret ne part pas au fournisseur — le journal compte, sans le motif.
pi.on("tool_result", async (event) => {
let hits = 0;
const content = event.content.map((part) => {
if (part.type !== "text") return part;
const { text, count } = redact(part.text);
hits += count;
return count > 0 ? { ...part, text } : part;
});
if (hits === 0) return undefined;
journalLine(cwd, sessionId, "redact", event.toolName, { hits });
return { content };
});
}
// ---------- la gate du fichier : `bun .pi/extensions/factory-guard.ts`, zéro jeton ----------
if (import.meta.main) {
const limits = readLimits({ FACTORY_LIMIT_TURNS: "2", FACTORY_LOOP_THRESHOLD: "3", FACTORY_LIMIT_COST_USD: "abc" });
const okLimits = limits.turns === 2 && limits.loopThreshold === 3 && limits.costUsd === 0 && limits.toolTimeoutS === 300;
const detector = new LoopDetector(20, 5);
let tripAt = 0;
for (let i = 1; i <= 6 && tripAt === 0; i++) {
if (detector.looping("bash", { command: "bun test", timeout: 300 })) tripAt = i;
}
const differentArgs = new LoopDetector(20, 5);
const noFalsePositive = [1, 2, 3, 4, 5, 6].every((i) => !differentArgs.looping("read", { path: `src/f${i}.ts` }));
const orderBlind = fingerprint("bash", { a: 1, b: 2 }) === fingerprint("bash", { b: 2, a: 1 });
const sample = [
"OPENROUTER_API_KEY=sk-or-v1-abcdefghijklmnopqrstuvwxyz0123456789",
"-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBg\n-----END PRIVATE KEY-----",
"Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.payload",
"const cfg = { theme: \"dark\" }; // rien de secret ici",
].join("\n");
const masked = redact(sample);
const okRedact = masked.count >= 3
&& !masked.text.includes("sk-or-v1-") && !masked.text.includes("MIIEvQIBADANBg") && !masked.text.includes("eyJhbGci")
&& masked.text.includes(`OPENROUTER_API_KEY=${REDACTED}`) && masked.text.includes("theme: \"dark\"")
&& redact("bun test : 12 pass, 0 fail").count === 0;
const ok = okLimits && tripAt === 5 && noFalsePositive && orderBlind && okRedact;
console.log(`factory-guard ${ok ? "OK" : "KO"} — seuils lus, boucle coupée au ${tripAt}e appel identique, ${masked.count} secrets masqués, texte ordinaire intact`);
process.exit(ok ? 0 : 1);
}Pièce — adws/adw_modules/roster.py
Cette version remplace celle du chapitre 27. Tout l’existant reste, y compris le profil
d’authentification du chapitre 17bis. S’ajoutent Limits (six plafonds, chacun avec sa variable
d’environnement), les champs limits et timeout de AgentSpec avec leurs valeurs par défaut,
et la fusion defaults.limits → limits de l’agent, clé par clé. La validation refuse avant
tout lancement un plafond nul, une fenêtre plus courte que le seuil, ou un mur qui tombe avant le
délai d’un seul appel d’outil.
"""roster — la feuille de distribution de l'usine : qui tourne, avec quels
moyens, sur quel payload — et jusqu'ou.
Les ADW nomment des agents, jamais des modeles. Ce module charge
factory.config.yaml, fusionne chaque agent sur les defauts (cle par cle),
et valide TOUT avant le moindre lancement : une erreur de roster coute
zero token — c'est le but.
Depuis le chapitre 27, le roster porte aussi la declaration du PAYLOAD.
Depuis l'annexe A7, chaque agent porte ses LIMITES : le coupe-circuit du
socle pi (.pi/extensions/factory-guard.ts) les lit dans l'environnement du
sous-processus (variables FACTORY_*), jamais dans le prompt ; le mur de
temps (timeout) reste cote Python. Sans bloc `limits`, les valeurs par
defaut s'appliquent : un roster d'avant ce chapitre tourne tel quel.
Depuis le chapitre 17bis, le roster porte aussi les PROFILS D'AUTHENTIFICATION
(bloc `auth:`) : par profil, le regime (cle, forfait, session), la route
(passerelle ou fournisseur direct), les NOMS des variables a injecter dans le
harnais — jamais leurs valeurs, qui vivent dans .env ou dans l'environnement
reel —, les hotes que la porte du module 6 devra ouvrir, et l'eventuel
fichier de session a monter dans une boite. Chaque agent reference un profil
(`auth:`, `gateway` par defaut). Tout est valide a zero token : profil
inconnu, variable absente (nommee dans le message), route ou regime inconnus,
montage hors du depot — avant le moindre lancement. Sans bloc `auth:`, le
profil integre `gateway` s'applique : un roster d'avant ce chapitre tourne
tel quel.
"""
from __future__ import annotations
import sys
import os
from dataclasses import dataclass, fields
from pathlib import Path, PurePosixPath, PureWindowsPath
import yaml
DEFAULT_PATH = Path("adws/adw_config/factory.config.yaml")
# Les deux dialectes du port (ch. 7), l'echelle de reflexion, et les sept
# outils integres de pi — l'adaptateur Claude Code traduit (ch. 11). L'outil
# terminal report_phase (A6) est ajoute d'office par l'adaptateur : le roster
# n'a pas a le declarer.
HARNESSES = ("pi", "claude")
THINKING = ("off", "minimal", "low", "medium", "high", "xhigh", "max")
KNOWN_TOOLS = ("read", "bash", "edit", "write", "grep", "find", "ls")
# Le mur de temps par defaut d'une phase agent, en secondes (remplace le 600
# fixe de HarnessRequest quand l'agent ne dit rien).
DEFAULT_TIMEOUT = 900
# Les trois regimes de credentials (17bis), ranges par MECANIQUE, jamais par
# marque : une cle au jeton (une variable, revocable, plafonnable) ; un
# forfait EXPOSE derriere une cle (la cle de l'abonnement, un point d'entree
# propre au fournisseur : un profil ordinaire) ; une session attachee au
# poste (rien a injecter, un fichier de session a monter — ni plafonnable ni
# revocable par run).
REGIMES = ("key", "plan", "session")
DEFAULT_AUTH_NAME = "gateway"
# Par ou passe le modele : `gateway` = la passerelle du ch. 15 (le port prefixe
# la route, l'identifiant du roster est celui du registre OpenRouter) ;
# `direct` = un fournisseur natif du harnais, l'identifiant du roster est la
# route telle quelle (zai/glm-5.3, kimi-coding/k3). La route n'est pas dans le
# nom du modele — `deepseek/...` est a la fois un auteur OpenRouter et un
# fournisseur natif de pi — elle est dans le profil.
ROUTES = ("gateway", "direct")
class RosterError(ValueError):
"""Un roster invalide — le motif exact, avant tout lancement."""
@dataclass(frozen=True)
class Limits:
"""Le coupe-circuit d'un agent : des filets larges, pas des objectifs.
Chaque champ a sa variable d'environnement, lue par factory-guard.ts.
0 = pas de plafond (jamais par defaut dans l'usine : un noeud sans
plafond est une session de poste, pas une phase).
"""
turns: int = 60 # tours LLM avant abort
tool_calls: int = 200 # appels d'outils avant refus
cost_usd: float = 2.0 # cout cumule (catalogue pi) avant abort
loop_window: int = 20 # fenetre glissante d'appels observes
loop_threshold: int = 5 # le meme appel N fois dans la fenetre = boucle
tool_timeout_s: int = 300 # injecte dans chaque bash sans timeout explicite
ENV = {"turns": "FACTORY_LIMIT_TURNS", "tool_calls": "FACTORY_LIMIT_TOOL_CALLS",
"cost_usd": "FACTORY_LIMIT_COST_USD", "loop_window": "FACTORY_LOOP_WINDOW",
"loop_threshold": "FACTORY_LOOP_THRESHOLD", "tool_timeout_s": "FACTORY_TOOL_TIMEOUT_S"}
def env(self) -> dict[str, str]:
"""Les limites telles que le sous-processus pi les recevra."""
return {self.ENV[f.name]: str(getattr(self, f.name)) for f in fields(self)}
DEFAULT_LIMITS = Limits()
@dataclass(frozen=True)
class AuthProfile:
"""Un profil d'authentification : ce qu'un agent a le droit d'emporter, et par ou.
Le profil DECRIT, il n'invente pas : `env` liste des NOMS de variables,
jamais des valeurs ; les valeurs vivent dans .env (ou l'environnement
reel, qui gagne toujours — la preseance du ch. 15). `hosts` est ce que la
porte du module 6 ouvrira pour ce profil : un profil sans hosts n'ouvre
rien. `mount` est un chemin RELATIF au depot, sous data_dir de preference
(jamais commite) : une session copiee la pour voyager dans une boite.
"""
name: str
regime: str # key | plan | session
route: str = "gateway" # gateway | direct — par ou passe le modele
env: tuple[str, ...] = () # des noms, jamais des valeurs
hosts: tuple[str, ...] = () # ce que la porte devra ouvrir (ch. 22)
mount: str | None = None # session a monter, relative au depot
note: str = "" # une phrase pour le lecteur du roster
@property
def direct(self) -> bool:
"""Vrai si le modele part tel quel vers un fournisseur natif, sans passerelle."""
return self.route == "direct"
def credentials(self, env: dict[str, str] | None = None) -> dict[str, str]:
"""Les variables du profil, resolues — ou le NOM de celle qui manque.
C'est le dictionnaire que le port injectera dans le noeud, a la
place du .env global : un noeud n'emporte que ce que son profil nomme.
"""
source = os.environ if env is None else env
missing = [name for name in self.env if not source.get(name)]
if missing:
raise RosterError(f"profil {self.name!r} : variable {missing[0]} absente de .env "
"et de l'environnement — le harnais rendrait "
"« Not logged in » au premier jeton paye")
return {name: source[name] for name in self.env}
# Le profil integre : la passerelle, une cle, tous les moteurs (ch. 15). Un
# roster sans bloc `auth:` tourne dessus ; un bloc `auth:` peut le redefinir.
GATEWAY_PROFILE = AuthProfile(name=DEFAULT_AUTH_NAME, regime="key", route="gateway",
env=("OPENROUTER_API_KEY",), hosts=("openrouter.ai",),
note="la passerelle : une cle, revocable et plafonnable par run, tous les moteurs")
@dataclass(frozen=True)
class AgentSpec:
"""Un agent du roster, defauts fusionnes : pret a etre lance tel quel."""
name: str
purpose: str
harness: str
model: str
thinking: str
tools: tuple[str, ...]
writes: tuple[str, ...] | None # None = libre · () = lecture seule · (...) = ces chemins
limits: Limits = DEFAULT_LIMITS # le coupe-circuit (A7)
timeout: int = DEFAULT_TIMEOUT # le mur de temps, cote Python (A7)
auth: AuthProfile = GATEWAY_PROFILE # le profil d'authentification, resolu (17bis)
@dataclass(frozen=True)
class Payload:
"""Le produit que l'usine travaille : ou il vit, et ce qui dit la verite sur lui."""
dir: str # relatif a la racine du depot
truth: tuple[tuple[str, ...], ...] # des argv ; exit 0 = vert (gates.command, ch. 12)
def commands(self) -> list[tuple[list[str], str]]:
"""Pret pour gates.command(cmd, cwd) : chaque commande tourne dans dir."""
return [(list(cmd), self.dir) for cmd in self.truth]
# La declaration implicite des chapitres 12 a 26 — ce que les ADW codaient en
# dur. Elle ne sert plus que si AUCUN roster ne declare de payload.
DEFAULT_PAYLOAD = Payload(dir="apps/plume", truth=(("bun", "test"),))
@dataclass(frozen=True)
class Roster:
"""Le roster charge et valide, les regles communes, et le payload."""
agents: dict[str, AgentSpec]
protected_files: tuple[str, ...]
data_dir: str
payload: Payload = DEFAULT_PAYLOAD
auth: dict[str, AuthProfile] | None = None # None = le seul profil integre (17bis)
def profile(self, name: str) -> AuthProfile:
"""Le profil par son nom — ou le refus, avec la liste des profils connus."""
return _profile(self.auth, name)
def credentials(self, agent: AgentSpec, env: dict[str, str] | None = None) -> dict[str, str]:
"""Ce que le port injectera dans le noeud de cet agent : son profil, resolu."""
return agent.auth.credentials(env)
def hosts(self) -> tuple[str, ...]:
"""L'union des hotes des profils UTILISES : ce que la porte du module 6 ouvrira."""
return tuple(sorted({host for spec in self.agents.values() for host in spec.auth.hosts}))
def _profile(profiles: dict[str, AuthProfile] | None, name: str) -> AuthProfile:
known = profiles or {DEFAULT_AUTH_NAME: GATEWAY_PROFILE}
try:
return known[name]
except KeyError:
raise RosterError(f"profil d'authentification inconnu {name!r} — "
f"declares : {sorted(known)}") from None
def load(path: str | Path = DEFAULT_PATH, env: dict[str, str] | None = None) -> Roster:
"""Charge, fusionne, valide — dans cet ordre, et tout ou rien.
env : l'environnement dans lequel les profils sont resolus. None = .env
puis l'environnement reel (la preseance du ch. 15) ; un dictionnaire
explicite sert aux gates a sec.
"""
path = Path(path)
raw = _read(path)
defaults = raw.get("defaults") or {}
entries = raw.get("agents") or []
if not entries:
raise RosterError(f"{path} : aucun agent declare")
profiles = _auth(raw, path) # les profils d'abord : les agents les referencent
agents: dict[str, AgentSpec] = {}
for entry in entries:
spec = _merge(defaults, entry, profiles)
if spec.name in agents:
raise RosterError(f"agent {spec.name!r} declare deux fois")
agents[spec.name] = spec
roster = Roster(
agents=agents,
protected_files=tuple(defaults.get("protected_files") or ()),
data_dir=str(defaults.get("data_dir", "adws/adw_data")),
payload=_payload(raw, path),
auth=profiles,
)
for spec in agents.values():
_validate(spec)
# Les profils : chaque agent en reference un qui existe, et chaque
# variable qu'il nomme est la — AVANT le premier jeton, jamais apres.
if env is None:
load_env()
for spec in agents.values():
try:
roster.credentials(spec, env)
except RosterError as error:
raise RosterError(f"agent {spec.name!r} : {error}") from None
return roster
def load_env(path: str | Path = ".env") -> None:
"""Le contrat du ch. 15 : .env dans l'environnement, sans jamais l'ecraser.
Le port fait de meme au chargement ; le roster le refait ici pour que
sa validation voie ce que le port verra — et rien de plus.
"""
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("=")
os.environ.setdefault(key.strip(), value.strip().strip("'\""))
def _auth(raw: dict, path: Path) -> dict[str, AuthProfile] | None:
"""Le bloc `auth:` : un profil par nom, valide a zero token. Absent = gateway seul."""
block = raw.get("auth")
if block is None:
return None
if not isinstance(block, dict) or not block:
raise RosterError(f"{path} : auth doit etre un bloc — un profil par nom")
profiles: dict[str, AuthProfile] = {}
for name, body in block.items():
body = body or {}
if not isinstance(body, dict):
raise RosterError(f"{path} : profil {name!r} doit etre un bloc (regime, route, env, hosts, mount)")
regime = str(body.get("regime", "key"))
if regime not in REGIMES:
raise RosterError(f"profil {name!r} : regime {regime!r} inconnu — {list(REGIMES)}")
route = str(body.get("route", "gateway"))
if route not in ROUTES:
raise RosterError(f"profil {name!r} : route {route!r} inconnue — {list(ROUTES)}")
names = tuple(str(v) for v in body.get("env") or ())
for variable in names:
if "=" in variable or not variable.isidentifier() or variable != variable.upper():
raise RosterError(f"profil {name!r} : env attend des NOMS de variables "
f"(MAJUSCULES, jamais de valeur) — recu {variable!r}")
if regime in ("key", "plan") and not names:
raise RosterError(f"profil {name!r} : un regime {regime!r} nomme au moins une variable")
mount = body.get("mount")
if mount is not None:
mount = str(mount)
# Relatif au depot, quel que soit l'OS : ni chemin absolu (POSIX ou
# Windows), ni HOME, ni remontee par `..`.
if (PurePosixPath(mount).is_absolute() or PureWindowsPath(mount).is_absolute()
or mount.startswith("~") or ".." in Path(mount).parts):
raise RosterError(f"profil {name!r} : mount {mount!r} sort du depot — une session "
"voyage sous un chemin relatif (data_dir de preference), jamais "
"depuis votre HOME")
if regime == "session" and mount is None:
raise RosterError(f"profil {name!r} : un regime 'session' nomme le fichier a monter (mount)")
profiles[str(name)] = AuthProfile(
name=str(name), regime=regime, route=route, env=names,
hosts=tuple(str(h) for h in body.get("hosts") or ()),
mount=mount, note=str(body.get("note", "")),
)
return profiles
def _read(path: Path) -> dict:
return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
def _limits(defaults: dict, entry: dict, name: str) -> Limits:
"""Les limites : les defauts du module, puis defaults.limits, puis limits de l'agent — cle par cle."""
merged = {**(defaults.get("limits") or {}), **(entry.get("limits") or {})}
known = {f.name for f in fields(Limits)}
unknown = sorted(set(merged) - known)
if unknown:
raise RosterError(f"agent {name!r} : limites inconnues {unknown} — connues : {sorted(known)}")
try:
return Limits(**{k: (float(v) if k == "cost_usd" else int(v)) for k, v in merged.items()})
except (TypeError, ValueError) as error:
raise RosterError(f"agent {name!r} : limits — {error}") from None
def _merge(defaults: dict, entry: dict, profiles: dict[str, AuthProfile] | None = None) -> AgentSpec:
"""L'agent par-dessus les defauts, cle par cle : il ne dit que ce qui differe."""
writes = entry.get("writes", None) # cle absente = None = libre dans le repo
name = str(entry.get("name", ""))
try:
profile = _profile(profiles, str(entry.get("auth", defaults.get("auth", DEFAULT_AUTH_NAME))))
except RosterError as error:
raise RosterError(f"agent {name!r} : {error}") from None
return AgentSpec(
name=name,
purpose=str(entry.get("purpose", "")),
harness=str(entry.get("harness", defaults.get("harness", "pi"))),
model=str(entry.get("model", defaults.get("model", ""))),
thinking=str(entry.get("thinking", defaults.get("thinking", "medium"))),
tools=tuple(entry.get("tools", defaults.get("tools") or ())),
writes=None if writes is None else tuple(writes),
auth=profile,
limits=_limits(defaults, entry, name),
timeout=int(entry.get("timeout", defaults.get("timeout", DEFAULT_TIMEOUT))),
)
def _payload(raw: dict, path: Path) -> Payload:
"""Le payload appartient au DEPOT, pas au roster choisi par --config.
Resolution en trois temps : le bloc `payload` du roster charge ; sinon
celui du roster par defaut ; sinon la declaration implicite d'avant le
chapitre 27.
"""
block = raw.get("payload")
if block is None and path.resolve() != DEFAULT_PATH.resolve() and DEFAULT_PATH.is_file():
block = _read(DEFAULT_PATH).get("payload")
if block is None:
return DEFAULT_PAYLOAD
if not isinstance(block, dict):
raise RosterError(f"{path} : payload doit etre un bloc avec dir et truth")
directory = str(block.get("dir") or "").strip()
if not directory:
raise RosterError(f"{path} : payload.dir manquant — ou vit le produit que l'usine travaille ?")
truth: list[tuple[str, ...]] = []
for entry in block.get("truth") or []:
# Une liste argv, jamais une chaine shell : pas de quoting, pas d'injection.
if (not isinstance(entry, list) or not entry
or not all(isinstance(part, str) and part for part in entry)):
raise RosterError(f"{path} : payload.truth — chaque commande est une liste argv "
f"non vide, jamais une chaine shell (recu {entry!r})")
truth.append(tuple(entry))
if not truth:
raise RosterError(f"{path} : payload.truth vide — sans commande de verite, "
"aucune gate ne peut dire « fini »")
return Payload(dir=directory, truth=tuple(truth))
def _validate(spec: AgentSpec) -> None:
"""Chaque miss echoue AVANT le lancement — jamais pendant, jamais en facture."""
if not spec.name:
raise RosterError("un agent sans nom n'est pas adressable")
if not spec.purpose:
raise RosterError(f"agent {spec.name!r} : purpose manquant — un agent, un role")
if spec.harness not in HARNESSES:
raise RosterError(f"agent {spec.name!r} : harnais inconnu {spec.harness!r} "
f"— disponibles : {list(HARNESSES)}")
if "/" not in spec.model:
raise RosterError(
f"agent {spec.name!r} : modele {spec.model!r} — toujours provider/id : "
"un motif nu devient ambigu des que deux fournisseurs portent le meme modele")
if spec.thinking not in THINKING:
raise RosterError(f"agent {spec.name!r} : thinking {spec.thinking!r} "
f"hors echelle {list(THINKING)}")
if not spec.tools:
raise RosterError(f"agent {spec.name!r} : aucun outil — "
"un agent sans outils ne peut rien proposer")
unknown = [tool for tool in spec.tools if tool not in KNOWN_TOOLS]
if unknown:
raise RosterError(f"agent {spec.name!r} : outils inconnus {unknown} "
f"— integres : {list(KNOWN_TOOLS)}")
# Le coupe-circuit : des plafonds positifs, une boucle detectable, un mur qui tient.
lim = spec.limits
for field_name in ("turns", "tool_calls", "cost_usd", "tool_timeout_s"):
if getattr(lim, field_name) <= 0:
raise RosterError(f"agent {spec.name!r} : limits.{field_name} doit etre > 0 — "
"un noeud d'usine a toujours un plafond")
if lim.loop_threshold < 2 or lim.loop_window < lim.loop_threshold:
raise RosterError(f"agent {spec.name!r} : loop_threshold >= 2 et loop_window >= loop_threshold "
f"(recu window={lim.loop_window}, threshold={lim.loop_threshold})")
if spec.timeout <= 0:
raise RosterError(f"agent {spec.name!r} : timeout doit etre > 0 (secondes)")
if spec.timeout <= lim.tool_timeout_s:
raise RosterError(f"agent {spec.name!r} : timeout ({spec.timeout} s) doit depasser "
f"limits.tool_timeout_s ({lim.tool_timeout_s} s) — sinon le mur tombe "
"avant le delai d'un seul appel d'outil")
if __name__ == "__main__":
# La gate du module : le roster se charge, se fusionne, se valide — et
# dit sur quel payload il travaille, avec quels plafonds. Lancer depuis la racine :
# uv run --with pyyaml python -m adws.adw_modules.roster [chemin-du-roster]
loaded = load(sys.argv[1] if len(sys.argv) > 1 else DEFAULT_PATH)
for agent_name in sorted(loaded.agents):
spec = loaded.agents[agent_name]
print(f"{spec.name:9} {spec.harness:7} {spec.model:34} "
f"thinking={spec.thinking:7} writes={spec.writes}")
profile = spec.auth
print(f"{'':9} auth : {profile.name} ({profile.regime}, route {profile.route}) — "
f"{', '.join(profile.env) or 'aucune variable'}"
+ (f" ; mount {profile.mount}" if profile.mount else ""))
print(f"{'':9} limites : {spec.limits.turns} tours, {spec.limits.tool_calls} outils, "
f"{spec.limits.cost_usd} $, boucle {spec.limits.loop_threshold}/{spec.limits.loop_window}, "
f"mur {spec.timeout} s")
print("proteges :", ", ".join(loaded.protected_files))
print("hotes :", ", ".join(loaded.hosts()) or "(aucun — la porte n'ouvrira rien)")
print("payload :", loaded.payload.dir, "—",
" ; ".join(" ".join(cmd) for cmd in loaded.payload.truth))
# Les variables que recevra pi : six, nommees, jamais dans le prompt.
sample = next(iter(loaded.agents.values())).limits.env()
assert set(sample) == set(Limits.ENV.values()) and sample["FACTORY_LOOP_THRESHOLD"].isdigit()
print("env :", " ".join(sorted(sample)))
# La gate a sec des profils (17bis) : un roster jetable, un environnement
# explicite — profil inconnu refuse, variable absente NOMMEE, route ou
# regime inconnus, montage hors du depot refuse, union des hotes calculee.
# Zero token, aucun fichier .env lu.
import tempfile
HEAD = "payload: { dir: apps/plume, truth: [[bun, test]] }\ndefaults: { tools: [read] }\n"
AUTH = ("auth:\n"
" gateway: { regime: key, env: [OPENROUTER_API_KEY], hosts: [openrouter.ai] }\n"
" glm-plan: { regime: plan, route: direct, env: [ZAI_API_KEY], hosts: [api.z.ai] }\n"
" claude-plan: { regime: plan, route: direct, env: [CLAUDE_CODE_OAUTH_TOKEN], hosts: [api.anthropic.com] }\n"
" claude-session: { regime: session, route: direct, env: [CLAUDE_CONFIG_DIR], mount: adws/adw_data/auth/claude, hosts: [api.anthropic.com] }\n")
AGENTS = ("agents:\n"
" - { name: scout, purpose: reperer, model: a/b }\n"
" - { name: builder, purpose: construire, model: zai/glm-5.3, auth: glm-plan }\n"
" - { name: reviewer, purpose: juger, model: a/b, harness: claude, auth: claude-plan }\n"
" - { name: documenter, purpose: rediger, model: a/b, harness: claude, auth: claude-session }\n")
full = {"OPENROUTER_API_KEY": "sk-or-x", "ZAI_API_KEY": "zai-x", "CLAUDE_CODE_OAUTH_TOKEN": "oat-x",
"CLAUDE_CONFIG_DIR": "/usine/adws/adw_data/auth/claude"}
def roster_file(text: str) -> Path:
handle = tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False, encoding="utf-8")
handle.write(text)
handle.close()
return Path(handle.name)
# 1. Trois regimes, resolus : la cle, le forfait derriere une cle, la session sans variable.
trio = load(roster_file(HEAD + AUTH + AGENTS), env=full)
assert trio.credentials(trio.agents["scout"], full) == {"OPENROUTER_API_KEY": "sk-or-x"}
assert trio.credentials(trio.agents["builder"], full) == {"ZAI_API_KEY": "zai-x"}
assert trio.credentials(trio.agents["reviewer"], full) == {"CLAUDE_CODE_OAUTH_TOKEN": "oat-x"}
assert trio.credentials(trio.agents["documenter"], full) == {"CLAUDE_CONFIG_DIR": "/usine/adws/adw_data/auth/claude"}
assert not trio.agents["scout"].auth.direct and trio.agents["builder"].auth.direct
assert trio.hosts() == ("api.anthropic.com", "api.z.ai", "openrouter.ai"), trio.hosts()
assert trio.profile("claude-session").mount == "adws/adw_data/auth/claude"
# 2. Sans bloc auth : gateway seul, un roster d'avant ce chapitre tourne tel quel.
plain = load(roster_file(HEAD + "agents:\n - { name: scout, purpose: reperer, model: a/b }\n"),
env={"OPENROUTER_API_KEY": "sk-or-x"})
assert plain.auth is None and plain.hosts() == ("openrouter.ai",)
# 3. Les refus, chacun avec son motif — avant tout lancement.
for text, env, expected in [
(HEAD + AUTH + "agents:\n - { name: scout, purpose: reperer, model: a/b, auth: codex }\n",
full, "profil d'authentification inconnu 'codex'"),
(HEAD + AUTH + AGENTS, {"OPENROUTER_API_KEY": "sk-or-x"}, "variable ZAI_API_KEY absente"),
(HEAD + "auth:\n side: { regime: key, route: sideways, env: [X] }\n"
+ "agents:\n - { name: scout, purpose: reperer, model: a/b, auth: side }\n", full, "route 'sideways' inconnue"),
(HEAD + "auth:\n home: { regime: session, mount: /Users/vous/.claude, hosts: [api.anthropic.com] }\n"
+ "agents:\n - { name: scout, purpose: reperer, model: a/b, auth: home }\n", full, "sort du depot"),
(HEAD + "auth:\n up: { regime: session, mount: ../ailleurs/session.json }\n"
+ "agents:\n - { name: scout, purpose: reperer, model: a/b, auth: up }\n", full, "sort du depot"),
(HEAD + "auth:\n leak: { regime: key, env: [OPENROUTER_API_KEY=sk-or-x] }\n"
+ "agents:\n - { name: scout, purpose: reperer, model: a/b, auth: leak }\n", full, "jamais de valeur"),
(HEAD + "auth:\n odd: { regime: cookie, env: [X] }\n"
+ "agents:\n - { name: scout, purpose: reperer, model: a/b, auth: odd }\n", full, "regime 'cookie' inconnu"),
]:
try:
load(roster_file(text), env=env)
raise AssertionError(f"aurait du refuser : {expected}")
except RosterError as error:
assert expected in str(error), (expected, str(error))
print("profils : 3 regimes resolus (passerelle, direct, pi et claude), gateway seul sans bloc auth, union des "
"hotes ; refus — profil inconnu, variable absente nommee, route inconnue, mount hors depot (x2), "
"valeur dans env, regime inconnu")Pièce — adws/adw_scout.py
Cette version remplace celle du chapitre 11. Le scout ne change pas. C’est la fabrique
agent_action, importée par adw_plan.py et adw_build.py, qui évolue. Elle transmet à pi les
limites et l’identité de la phase par l’environnement, applique le mur de temps de l’agent, et
garde la session même quand le port échoue. Après un coupe-circuit, elle lit le rapport de garde
et rédige la relance. Elle crédite aussi run.tokens et lit l’enveloppe de la porte typée avant
le texte. --max-turns sert à éprouver le coupe-circuit. Le profil d’authentification du
chapitre 17bis est conservé.
#!/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]
uv run adws/adw_scout.py "..." --max-turns 2 --retries 0 # eprouver le coupe-circuit
uv run adws/adw_scout.py "..." --auth glm-plan --model zai/glm-5.3 # un autre profil, pour CE run
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.
Version 17bis : la fabrique agent_action passe au port le PROFIL
D'AUTHENTIFICATION de l'agent, resolu par le roster (noms et valeurs des
variables — jamais dans le prompt, jamais dans le YAML) et sa route. Le port
ferme alors le coffre .env : le noeud ne recoit que ce que son profil nomme.
Deux boutons d'essai — --auth <profil> et --model <route> — pour essayer un
profil declare sans toucher au roster. adw_plan.py et adw_build.py heritent
du profil de leur agent sans changer d'une ligne.
Version annexe A7 : la fabrique agent_action transmet au noeud pi les
limites du roster (variables FACTORY_LIMIT_*) et l'identite de la phase
(FACTORY_PHASE, FACTORY_ROLE, FACTORY_ADW_ID), applique le mur de temps de
l'agent, lit le rapport de garde quand le coupe-circuit a coupe, et repart
dans la MEME session avec le motif exact. adw_plan.py et adw_build.py
l'importent et n'ont pas a changer.
"""
import argparse
import json
import os
import sys
import uuid
from contextlib import contextmanager
from dataclasses import asdict, dataclass, field, replace
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
# Le rapport de garde : ecrit par .pi/extensions/factory-guard.ts quand un
# compteur coupe la phase, lu ici pour rediger la relance. Un fichier par
# phase, sous le dossier de run cree par constat() — jamais commite.
GUARD_DIR = Path("adws/adw_data/runs")
GUARD_ENV = "FACTORY_GUARD_REPORT"
@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."""
# La relance apres un coupe-circuit : le motif exact, puis le contrat. Pas
# de reprise « a l'identique » : l'agent reprend ou il en etait, sans
# repeter le geste qui a coupe.
GUARD_ASK = """Ta tentative precedente a ete interrompue par l'usine : {reason}.
Reprends ou tu en etais, dans cette meme session, SANS repeter ce geste :
si une commande a deja rendu son resultat, lis-le au lieu de la relancer ;
si tu as assez d'elements, conclus. Termine par l'enveloppe demandee.
"""
def phase_env(phase_name: str, agent: roster.AgentSpec, run: Run) -> dict[str, str]:
"""Ce que le noeud pi doit savoir de la phase — par l'environnement, jamais par le prompt.
Les limites du roster (coupe-circuit), l'identite de la phase (lue par
les extensions du socle) et le chemin du rapport de garde.
"""
variables = dict(agent.limits.env())
variables.update({
"FACTORY_PHASE": phase_name,
"FACTORY_ROLE": agent.name,
"FACTORY_ADW_ID": run.adw_id,
GUARD_ENV: str((GUARD_DIR / run.adw_id / f"{phase_name}.guard.json").resolve()),
})
return variables
@contextmanager
def phase_environment(variables: dict[str, str]):
"""Pose les variables le temps de l'appel au port, puis restaure l'environnement.
Le port herite de l'environnement du runner (c'est ainsi que _spawn
transmet deja le schema de la porte) : ce contexte est le canal, borne
a la duree de la phase — un ADW n'en lance jamais deux a la fois.
"""
previous = {name: os.environ.get(name) for name in variables}
os.environ.update(variables)
try:
yield
finally:
for name, value in previous.items():
if value is None:
os.environ.pop(name, None)
else:
os.environ[name] = value
def read_guard_report(path: str) -> dict | None:
"""Le rapport de garde s'il existe — consomme (supprime) pour ne jamais resservir."""
report = Path(path)
if not report.is_file():
return None
try:
data = json.loads(report.read_text(encoding="utf-8"))
except json.JSONDecodeError:
data = None
report.unlink(missing_ok=True)
return data if isinstance(data, dict) else None
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, les outils, les
limites et le mur de temps ; le port traduit les uns en drapeaux, le
contexte d'environnement transmet les autres. La demande part a la
tentative 1, la correction motivee ensuite — dans la MEME session, y
compris apres un coupe-circuit : l'erreur du port porte la session.
"""
next_ask = envelopes.correction("enveloppe invalide", expected)
def action(run: Run, attempt: int):
nonlocal next_ask
ask = make_ask(run) if attempt == 0 else next_ask
variables = phase_env(phase_name, agent, run)
request = harness.HarnessRequest(
prompt=ask, session_id=run.sessions.get(phase_name),
model=agent.model, thinking=agent.thinking, tools=agent.tools,
timeout=agent.timeout,
# Le profil, resolu ICI, cote runner : le noeud n'emporte que ses
# variables, le coffre .env reste ferme pour lui (17bis).
auth=agent.auth.credentials(), direct=agent.auth.direct)
try:
with phase_environment(variables):
result = harness.run(agent.harness, request)
except harness.HarnessError as error:
# La session existe meme en echec (A6) : la reprise la poursuit.
if error.session_id:
run.sessions[phase_name] = error.session_id
guard = read_guard_report(variables[GUARD_ENV])
if guard:
# Le coupe-circuit a coupe : le motif exact devient la consigne.
reason = str(guard.get("reason", "plafond atteint"))
next_ask = GUARD_ASK.format(reason=reason) + envelopes.contract(expected)
raise PhaseFailure(f"coupe-circuit ({guard.get('kind', '?')}) : {reason}") from None
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
run.tokens += result.tokens # le compteur du ch. 20, enfin credite
payload = result.envelope if result.envelope is not None else result.text
try:
return envelopes.parse(payload, expected)
except EnvelopeError as error:
next_ask = envelopes.correction(str(error), expected)
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)
parser.add_argument("--max-turns", type=int, default=None,
help="abaisser le plafond de tours du scout pour CE run — "
"le bouton d'essai du coupe-circuit (A7)")
parser.add_argument("--auth", default=None,
help="un autre profil d'authentification du roster pour CE run — "
"le bouton d'essai des profils (17bis)")
parser.add_argument("--model", default=None,
help="le modele pour CE run — un profil a route directe attend une route "
"de son fournisseur (zai/glm-5.3), pas un id de la passerelle (17bis)")
args = parser.parse_args()
factory = roster.load(args.config) # zero token : tout echec est gratuit
agent = factory.agents["scout"]
if args.max_turns is not None:
# Le roster reste la verite ; ce run seul serre le filet — sans YAML de test.
agent = replace(agent, limits=replace(agent.limits, turns=max(1, args.max_turns)))
if args.auth is not None:
# Un profil declare dans le roster, resolu avant tout jeton : un profil
# inconnu ou une variable absente s'arretent ici, avec leur motif.
profile = factory.profile(args.auth)
profile.credentials()
agent = replace(agent, auth=profile)
if args.model is not None:
agent = replace(agent, model=args.model)
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())La gate du TP
Lancez ces commandes depuis la racine de plume-factory. Les deux premières commandes ne coûtent rien. La troisième
lance un scout bridé à deux tours, sans reprise : il doit finir rouge, proprement. La quatrième
lit la trace. La dernière relance le scout sans bride.
bun .pi/extensions/factory-guard.ts
uv run --with pyyaml python -m adws.adw_modules.roster
uv run adws/adw_scout.py "Liste les fichiers de tests de apps/plume et ce que chacun verifie." --max-turns 2 --retries 0
sqlite3 adws/adw_data/factory.db "SELECT type, name, json_extract(payload_json, '$.reason') FROM harness_by_phase WHERE adw_id = (SELECT adw_id FROM runs ORDER BY started_at DESC LIMIT 1) AND type IN ('guard', 'redact') ORDER BY ts;"
uv run adws/adw_scout.py "Liste les fichiers de tests de apps/plume et ce que chacun verifie."Résultat attendu :
factory-guard OK — seuils lus, boucle coupée au 5e appel identique, 4 secrets masqués, texte ordinaire intact.- Le roster affiche pour chaque agent ses limites par défaut
(
60 tours, 200 outils, 2.0 $, boucle 5/20, mur 900 s) et les six variablesFACTORY_*. - Le scout bridé s’arrête après son deuxième tour. La sortie d’erreur affiche
echec — coupe-circuit (tours) : 2 tours atteints (plafond 2), le code de retour est non nul, et la session est dansrun.sessions. - La requête montre
guard / armedpuisguard / toursavec ce motif. Une ligneredactn’apparaît que si un résultat d’outil contenait un secret. - Le dernier scout rend son enveloppe verte, comme au chapitre 11.
Coût : rien pour les deux gates à sec, moins d’un centime pour le scout bridé (environ 20 s), un centime ou moins pour le scout complet (30 à 60 s).
Quiz — teste tes connaissances
0/7
bonnes reponses
Objectif non atteint (minimum 5/7 requis).
Remonte relire la fiche memo en pretant attention aux points manques, puis cliquer sur « Recommencer » pour retenter.
Remonte relire la fiche memo en pretant attention aux points manques, puis cliquer sur « Recommencer » pour retenter.