Disposer pendant la phase — coupe-circuit et masquage
Quatre compteurs qui coupent une phase qui dérape (tours, appels d'outils, coût, boucle) et un filtre qui masque les secrets avant qu'ils ne partent au fournisseur : le code dispose désormais pendant la phase, pas seulement après. Le roster gagne ses limites, le runner apprend à relancer avec le motif exact.
Hier, vous avez fermé la porte typée : l’enveloppe arrive par report_phase, validée par schéma,
et le runner lit stopReason et les jetons. Mais entre le premier tour et cette enveloppe, que
possède le code ? Un seul mur, posé au chapitre 15 : HarnessRequest.timeout = 600. Un builder
qui relance bun test six fois de suite en espérant un autre résultat, un scout qui relit le
même fichier à chaque tour, un modèle qui cat un .env et l’envoie au fournisseur dans le
résultat d’outil : rien ne les arrête avant la dixième minute, et quand le mur tombe, il tombe
sans enveloppe, sans motif, et la facture est déjà réglée. À la fin de ce chapitre, la loi du
livre s’appliquera pendant la phase : quatre compteurs (tours, appels d’outils, coût cumulé,
boucle) coupent le nœud pi au moment exact où il dérape, le motif remonte au runner qui relance
dans la même session avec ce motif comme consigne, et ce qui sort des outils est filtré des
secrets avant de quitter la machine. La pièce du jour ajoute une extension du socle,
factory-guard.ts, à côté de factory-report.ts (A6), et fait évoluer deux modules Python : le
roster (chapitre 27) apprend les limites par agent, et la fabrique agent_action (chapitre 11)
apprend à lire un arrêt.
Coupe-circuit : tours, outils, coût, boucle
L’idée en une phrase
Un coupe-circuit est un ensemble de compteurs tenus par le code déterministe à l’intérieur même de la phase (côté pi, dans une extension du socle, alimentés par les événements du cycle de vie), qui arrête l’agent quand un seuil posé par le roster est franchi, et qui remonte le motif au runner par la trace. La couture ne bouge pas, mais le code y dispose désormais en continu, et plus seulement quand l’agent a fini.
Points clés
- Quatre horloges plutôt qu’un mur.
turn_endcompte les tours,tool_callcompte les appels d’outils,message_end(rôle assistant) cumuleusage.cost.total, et une fenêtre glissante de hachés(outil, arguments triés)détecte la boucle : le même appel répété cinq fois dans les vingt derniers est une boucle, pas de la persévérance. Le mur de temps du roster reste côté Python, en dernier recours. - Les seuils viennent du roster, par l’environnement. Chaque agent déclare
limits:ettimeout:dansfactory.config.yaml, et le runner les transmet à pi en variablesFACTORY_LIMIT_*, jamais par le prompt. L’extension lit l’environnement : sans variable (votre session interactive), aucun plafond de tours ni de coût, et seules la détection de boucle et le délai par appel restent armés. - Deux gestes, un seul déclenchement. Sur un plafond de tours ou de coût,
ctx.abort(): le message d’assistant suivant sort enstopReason: "aborted", que l’adaptateur v4 du chapitre A6 transforme enHarnessErrorporteuse de la session. Sur une boucle ou un plafond d’appels, mieux :block: true, terminate: truedanstool_call, où l’appel est refusé et le tour s’arrête là, sans un appel LLM de plus. Le coupe-circuit ne se déclenche qu’une fois. - Le motif voyage par deux canaux, jamais par la sortie du modèle. Une ligne
guarddans le journal harnais du chapitre A4 (le même fichier JSONL, doncfactory.dbaprès le miroir), et un petit fichier de rapport désigné parFACTORY_GUARD_REPORT, queagent_actionlit après l’erreur pour écrire la relance : « ta tentative a appelébashsix fois avec les mêmes arguments ». La reprise repart dans la même session, motif en tête. - Le délai par appel manquait. L’outil
bashde pi accepte un paramètretimeout(en secondes) mais n’a aucun délai par défaut : unbun testqui attend une entrée bloque la phase jusqu’au mur de 600 s. L’extension injectetimeoutdans les arguments s’il est absent (FACTORY_TOOL_TIMEOUT_S, 300 par défaut) : le modèle ne renseigne jamais ce champ, le code le fait pour lui.
Exemple concret
Reprenez le builder du chapitre 13 sur le workhorse du roster (z-ai/glm-5.3, relevé ce jour
sur openrouter.ai/models : 1,40 $ le million de jetons en entrée, 4,40 $ en sortie). La spec
demande un export Markdown, un test échoue sur une fin de ligne. Sans coupe-circuit : le builder
relance bun test, même commande et même résultat, à chaque tour, en reformulant sa surprise.
Chaque tour relit un contexte de 50 à 80 k jetons : une dizaine de centimes par tour, quinze
à vingt tours avant le mur de 600 s, soit un à deux dollars brûlés et une HarnessError
sans motif. La reprise repart dans une session neuve, parce que l’ancien _run_pi perdait
l’identifiant. Avec la pièce du jour : au cinquième bash(bun test) identique dans la fenêtre,
tool_call refuse l’appel et termine le tour, pour zéro jeton de plus. Le journal porte
guard / boucle, le rapport dit « bash × 5 avec les mêmes arguments », et la tentative
suivante (même session, donc le builder garde ses fichiers ouverts et sa compréhension de la
spec) commence par : « ta tentative a été interrompue : bash appelé 5 fois avec les mêmes
arguments, lis la sortie du test au lieu de le relancer ». Sur ce run, l’économie se compte en
dizaines de centimes et en minutes. Sur un best-of-N du chapitre 25, où cinq boîtes
peuvent boucler en parallèle, en dollars.
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 |
Config — les limites dans le roster
Ce bloc s’ajoute à un agent (ou aux defaults:) de factory.config.yaml. Sans lui, le roster
applique les valeurs par défaut de roster.py, et votre roster du chapitre 27 tourne tel quel.
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 qui a coupé (armed à l’ouverture, puis tours,
outils, cout ou boucle) et son motif exact, et la vue harness_by_phase du chapitre A4 la
relie au run et à la phase. Sans client sqlite3, uv run adws/adw_modules/harness_trace.py --last montre la même ligne.
Piège courant : « un plafond d’appels d’outils serré économise des jetons » est inexact. Ce sont les tours et le coût qui bornent la dépense, et un appel d’outil ne coûte rien au fournisseur, c’est le tour qui l’entoure qui coûte. Un plafond d’outils trop bas coupe un builder en pleine série d’
editlégitimes et vous fait payer une reprise pour rien. Gardez-le large (trois appels par tour, en ordre de grandeur) : il n’est là que pour la boucle que la fenêtre n’aurait pas vue, jamais pour discipliner un agent qui travaille.
Filtrer ce qui sort des outils (tool_result)
L’idée en une phrase
Le résultat d’un outil est du contexte que le nœud pi s’apprête à envoyer au fournisseur
de modèles. L’événement tool_result est un intergiciel (middleware) où le code peut
le réécrire avant l’envoi : l’usine y masque les motifs de secrets et journalise chaque
masquage, parce qu’un .env lu par un scout curieux ne doit jamais quitter la machine, quel que
soit le modèle et quel que soit le prompt.
Points clés
tool_resultchaîne comme un intergiciel. Les extensions s’exécutent dans l’ordre de chargement, chacune voit le résultat laissé par la précédente et peut rendre un correctif partiel (content,details,isError,usage), et un champ omis garde sa valeur. C’est le seul point du cycle de vie où le contenu est déjà produit et pas encore envoyé.- Masquer, pas bloquer. Un
cat .envrefusé par damage control (A3) est le premier rempart, mais la clé peut arriver par ungrep -r, ungit show, un fichier de config mal rangé. Le second rempart ne juge pas la commande : il remplace tout ce qui ressemble à un secret (sk-…,OPENROUTER_API_KEY=…,Bearer …, blocs PEM) par«redacted»et laisse le reste intact, et le modèle continue de travailler. - La troncature et le code retour, pi les fait déjà. Au moment d’écrire, les outils intégrés
tronquent leur sortie à quelques dizaines de kilo-octets (tête conservée, fichier complet sur
le disque) et
bashrend une erreur d’outil dès qu’une commande sort avec un code différent de zéro : le modèle ne peut pas ignorer unbun testrouge. L’extension n’a rien à ajouter ici, et l’écrire deux fois serait présenter un pattern de la méthode comme natif, ou l’inverse. - Chaque masquage laisse une trace. Une ligne
redactdans le journal harnais, avec le nom de l’outil et le nombre de motifs remplacés, jamais le secret. Un run qui masque trois fois signale un agent qui fouille là où il ne devrait pas : c’est une donnée pour le chapitre 19, pas un incident. - Deux harnais, une seule pièce. Claude Code offre des crochets comparables, mais l’usine ne
filtre que son nœud pi : c’est lui qui sert les modèles ouverts via la passerelle, là où le
masquage a le plus de valeur. Côté Claude Code, le roster garde les outils de lecture sur des
chemins gouvernés par
protected_files. Le dire une fois suffit.
Exemple concret
Un scout sur le modèle léger du roster cherche « où est configurée la clé de la passerelle ».
Il lance grep -rn OPENROUTER . --include=* : la sortie contient .env.sample (inoffensif) et,
si votre .gitignore a une faille, .env avec la vraie clé. Sans filtre : la clé part au
fournisseur dans le résultat d’outil, puis dans chaque tour suivant tant qu’elle reste dans le
contexte, et pour un scout de six tours la même ligne voyage six fois. Avec la pièce du jour :
tool_result remplace OPENROUTER_API_KEY=sk-or-v1-… par OPENROUTER_API_KEY=«redacted», le
scout rend son enveloppe (« la clé vit dans .env, chargée par harness._load_env() »), et le
journal porte redact / bash / 1. Coût : zéro jeton (le remplacement raccourcit même le
contexte d’une trentaine de caractères) et quelques microsecondes par résultat.
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 |
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 : « avec damage control, le masquage est redondant » est inexact. Damage control juge une commande (
cat .envest bloqué), le masquage juge un contenu (la sortie d’ungrep -rlégitime). Les deux remparts voient des choses différentes, à des moments différents. Le masquage ne protège pas non plus d’un secret que vous auriez collé dans un prompt ou unAGENTS.md: celui-là part au fournisseur avant tout outil.
Fil rouge — la pièce posée aujourd’hui
Sur le plan de l’usine, la pièce du jour s’insère dans le socle pi, quatrième extension
après damage-control.ts (A3), factory-obs.ts (A4) et factory-report.ts (A6), et remonte
jusqu’au roster (chapitre 27) et à la fabrique de phase agent_action (chapitre 11). La
couture ne bouge pas : le runner Python possède le graphe, l’agent reste un nœud borné dans une
phase nommée. Ce qui change, c’est quand le code dispose : jusqu’ici, après la phase (gates,
état des lieux, parse), et désormais aussi pendant, par des compteurs que l’agent ne voit
pas et des seuils qu’il ne négocie pas. Déterministe : les seuils (roster), leur transport
(environnement), les compteurs, le haché des appels, le masquage, la ligne de trace. Délégué :
tout le travail entre deux compteurs. L’enveloppe qui traverse quand ça coupe est un rapport de
garde (compteur, seuil, motif) que agent_action transforme en consigne de reprise, dans la
même session. Coût d’usage : zéro jeton pour les compteurs et le masquage, et une boucle
coupée par block + terminate économise tous les tours qu’elle aurait encore brûlés, soit sur un
builder qui dérape, dizaines de centimes à plusieurs dollars par run. Note pour le chapitre
17 : le coût que cumule le coupe-circuit est celui du catalogue de pi (usage.cost), une
estimation, et la facture de la passerelle fait foi. Note pour le chapitre 13 : la phase build de
adw_sdlc.py construit encore sa requête elle-même, avec le mur de 600 s, et elle sera alignée
sur la fabrique du jour avec les autres évolutions du runner, plus loin dans l’annexe. En
attendant, le coupe-circuit y est armé avec les valeurs par défaut de roster.py. Note pour le
chapitre A6 : l’extension d’injection de contexte annoncée là-bas n’est pas posée aujourd’hui non
plus : la fabrique exporte dès maintenant FACTORY_PHASE, FACTORY_ROLE et FACTORY_ADW_ID,
les variables qu’elle lira, et elle prendra place avec l’inventaire du socle.
Travaux pratiques — la pièce du jour
Trois fichiers à poser dans plume-factory, qui devient, chapitre après chapitre, votre usine
logicielle agentique : une extension du socle qui porte les deux sous-thèmes (le budget de
fichiers du jour impose de la livrer en un seul module, où coupe-circuit et masquage sont deux
blocs de fonctions pures, exportées et testées), et deux modules Python qui remplacent leurs
versions précédentes. factory.config.yaml ne change pas : les valeurs par défaut s’appliquent.
Pièce — .pi/extensions/factory-guard.ts
Le coupe-circuit et le masquage. Les seuils viennent de l’environnement (readLimits), la
boucle d’un LoopDetector à fenêtre glissante, le masquage de redact : trois fonctions pures,
exportées, dont le fichier porte la gate sous Bun. Les effets de bord se limitent au journal
harnais du chapitre A4 (même fichier, même format de ligne) et au rapport de garde désigné par
FACTORY_GUARD_REPORT. 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 ce qui existait reste : mêmes dataclasses,
même fusion, même résolution du payload, mêmes validations, même gate. S’ajoutent Limits (six
plafonds, chacun avec sa variable d’environnement), les champs limits et timeout de
AgentSpec avec leurs valeurs par défaut (votre factory.config.yaml ne change pas d’une
ligne), la fusion defaults.limits → limits de l’agent, clé par clé, et leur validation à zéro
jeton : 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 échouent avant tout lancement. Le profil d’authentification du chapitre 17bis (auth:, route, coffre) est conservé.
"""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 lui-même ne change pas : même brief,
même enveloppe, mêmes phases. C’est la fabrique agent_action, que adw_plan.py et
adw_build.py importent, qui évolue : elle transmet au nœud pi les limites et l’identité de la
phase par l’environnement, applique le mur de temps de l’agent, mémorise la session même quand le
port échoue, lit le rapport de garde pour rédiger la relance, crédite enfin run.tokens, et lit
l’enveloppe de la porte typée avant le texte. Un --max-turns sert de bouton d’essai au
coupe-circuit, sans roster de test. Le profil d’authentification du chapitre 17bis (auth:, route, coffre) 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
Depuis la racine de plume-factory. Cinq commandes, une par ligne, identiques dans bash et
PowerShell. Les deux premières ne coûtent rien, la troisième lance un scout avec un plafond de
deux tours et aucune reprise (le run doit finir rouge, proprement), les deux
dernières lisent la trace.
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 variables FACTORY_*. Le
scout bridé s’arrête après son deuxième tour avec, sur la sortie d’erreur, echec — coupe-circuit (tours) : 2 tours atteints (plafond 2) et un code de retour non nul, sans aucune
HarnessError muette, et la session est dans run.sessions. La requête montre une ligne guard / armed puis guard / tours avec ce même motif (et une ligne redact si un résultat d’outil
contenait un secret, ce qui ne devrait pas arriver). Le dernier scout, sans bride, rend son enveloppe
verte comme au chapitre 11. Sans client sqlite3, uv run adws/adw_modules/harness_trace.py --last liste les mêmes événements. Coût : les deux gates à sec ne dépensent rien, le scout
bridé coûte moins d’un centime (deux tours sur le modèle léger, ~20 s), et le scout complet
un centime ou moins en 30 à 60 s sur le roster par défaut.
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.