Annexe — Harnais pi Chapitre A4 / 43

Observer le harnais de l'intérieur

Une extension pi journalise chaque tour, chaque appel d'outil et chaque facture du modèle. Un miroir SQLite les relie aux phases du runner par leur session. Comparé aux hooks et à la télémétrie de Claude Code.

Au chapitre 19, vous avez lu un run phase par phase. Prenez une phase build de quatre minutes, réussie à la deuxième tentative. Combien de tours ? Combien de bun test ? Le journal de l’usine ne le sait pas, car le tracer du chapitre 18 n’a pas de capteur dans le harnais.

Ce chapitre en pose un. Une extension pi écrit chaque événement au moment où il arrive. Un miroir SQLite relie ces lignes aux phases du runner par la session. Coût : zéro jeton.

Streamer les événements du cycle de vie

Voici une ligne de la trace harnais. L’extension en écrit une par événement de pi, dans un fichier par session sous adws/adw_data/traces/harness/.

{
  "event_id": "hev_7c1e9a4d2b0f",
  "ts": "2026-09-02T14:07:31.412Z",
  "session_id": "3f2a9c7e-5b1d-4e8a-9c02-6d7f8e9a0b1c",
  "type": "assistant",
  "name": "openrouter/z-ai/glm-5.3",
  "payload": {"input": 38120, "output": 1460, "cache_read": 0, "cache_write": 0,
              "total_tokens": 39580, "cost_usd": 0.0598, "stop_reason": "toolUse", "tool_calls": 2}
}

Une ligne par événement, au fil de l’eau

L’extension s’abonne aux événements observateurs de pi et ne bloque jamais rien. Elle écrit quand l’événement arrive, pas en fin de session. Un pi -p tué par un timeout garde donc sa trace jusqu’au gel.

Événement pi	Ligne écrite (`type`)	Ce que la charge utile porte
`session_start`	`session_start`	mode, modèle, dossier de travail
`before_agent_start`	`prompt`	taille et début du prompt
`turn_start` / `turn_end`	`turn_start` / `turn_end`	numéro du tour
`message_end` (assistant)	`assistant`	modèle, jetons, coût, motif d’arrêt, appels d’outils
`tool_execution_start` / `_end`	`tool_start` / `tool_end`	outil, arguments abrégés, durée, erreur, taille du résultat
`session_compact`	`compaction`	motif, jetons avant
`agent_settled` / `session_shutdown`	`agent_settled` / `session_shutdown`	totaux : tours, outils, jetons, coût

La durée d’un outil se calcule entre tool_execution_start et tool_execution_end, appariés par leur toolCallId, même quand les outils tournent en parallèle.

Côté Claude Code : l’équivalent est un hook, une commande lancée sur un événement (SessionStart, PostToolUse, SessionEnd…). Elle reçoit un JSON sur stdin avec session_id et, pour un outil, tool_use_id.

`message_end` porte la facture

Chaque message d’assistant porte son usage : jetons d’entrée, de sortie, de cache, et cost.total calculé par pi. Le pied de page du chapitre A2 additionnait ce chiffre. Ici, vous le gardez.

Côté Claude Code : avec CLAUDE_CODE_ENABLE_TELEMETRY=1, l’export OpenTelemetry émet un événement claude_code.api_request par appel au modèle (jetons, coût estimé) et un claude_code.tool_result par outil (durée). Tous portent session.id.

Ce que le journal ne garde pas

Ni le contenu des résultats d’outils, ni le prompt entier. Un read de 40 Ko par ligne rendrait le journal illisible. Le journal garde des tailles, des durées, des compteurs, et 120 caractères de tête pour reconnaître un tour. Il n’ajoute rien à la sortie standard, que lit l’adaptateur du chapitre 7.

Ce que la trace révèle

Reprenez la phase build. À la tentative 1, le modèle a fait sept tours et relu tout apps/plume/, avec un message à 38 000 jetons d’entrée. La correction a pris trois tours et ~9 000 jetons. Les deux tiers du coût sont partis dans la relecture.

Piège courant : croire que le fichier de session de pi suffit. Il sert à reprendre une conversation, pas à la requêter. C’est un arbre, où le coût d’un tour se recalcule en remontant les branches. La trace harnais est plate, et elle survit à un /new ou à un --session-dir nettoyé.

Côté Claude Code : même constat. Les transcripts JSONL sous ~/.claude/projects/ sont supprimés après 30 jours par défaut, et la doc déconseille de les parser car leur format change d’une version à l’autre.

Brancher la trace harnais sur factory.db

Lancez uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ?". À la fermeture de la session pi, ses lignes rejoignent la table harness_events de factory.db. La vue harness_by_phase montre alors la phase prompt : deux tours, trois outils, 6 400 jetons, et un coût qui recoupe celui du runner.

La clé de jointure existait déjà

Depuis le chapitre 7, l’adaptateur pi choisit l’identifiant de session (--session-id), et le runner le garde en mémoire dans run.sessions. La pièce du jour lui fait écrire un événement de plus, phase_session : la phase, la tentative, la session. Rien d’autre ne change dans le runner.

Côté Claude Code : la même clé existe. claude --session-id accepte un UUID choisi par l’appelant, et claude -p --output-format json rend l’identifiant de session, l’usage et le coût.

Un miroir rejouable

harness_trace.py --ingest insère les JSONL avec INSERT OR IGNORE sur event_id. Relancer dix fois ne duplique rien, et un miroir perdu se reconstruit depuis le brut. L’extension lance --ingest à session_shutdown : environ une seconde, sans jeton. Si uv manque, le prochain --ingest rattrape.

La vue joint au moment de la lecture

Le runner écrit phase_session après le retour du harnais, donc après le miroir. Une colonne adw_id remplie à l’ingestion serait souvent fausse. La vue calcule la jointure à chaque requête.

Une phase sans trace n’est pas un bug

Une phase code n’ouvre pas de session, et une extension pi ne voit pas une phase confiée à Claude Code. Le lecteur --last le dit au lieu d’afficher zéro.

Deux journaux, une base

	Journal du runner (ch. 18-20)	Trace harnais (aujourd’hui)
Qui écrit	`tracer.py`, depuis le runner Python	`factory-obs.ts`, depuis le processus pi
Grain	run, phase, tentative	session, tour, appel d’outil, facture
Clé	`adw_id`, `phase_id`	`session_id`
Brut	`traces/{adw_id}.jsonl`	`traces/harness/{session_id}.jsonl`
Miroir	`runs`, `phases`, `events`	`harness_events` + vue `harness_by_phase`
Jointure	—	`events` où `type = 'phase_session'`

Commande — interroger la jointure

Avec le client sqlite3, une requête par ligne. Sans lui, --last fait le même travail.

# combien d'evenements harnais par phase du dernier run
sqlite3 adws/adw_data/factory.db "SELECT phase_id, COUNT(*) AS n FROM harness_by_phase WHERE adw_id = (SELECT adw_id FROM runs ORDER BY started_at DESC LIMIT 1) GROUP BY phase_id ORDER BY phase_id;"

# les factures du modele, tour par tour, pour une phase
sqlite3 adws/adw_data/factory.db "SELECT ts, name, json_extract(payload_json,'$.total_tokens') AS tokens, json_extract(payload_json,'$.cost_usd') AS usd FROM harness_by_phase WHERE phase_id='a1b2c3d4-01' AND type='assistant' ORDER BY ts;"

# sans client sqlite3 : le lecteur embarque
uv run adws/adw_modules/harness_trace.py --last

Piège courant : faire écrire l’extension directement dans factory.db. Le harnais dépendrait d’un pilote SQLite, qui diffère entre Node et Bun, et votre gate bun … casserait. Le JSONL est le contrat : tout harnais qui sait écrire une ligne peut alimenter le miroir.

pi et Claude Code côte à côte

	pi	Claude Code
Capteur	extension TypeScript, dans le processus	hook, commande externe qui lit du JSON
Facture d’un appel	`usage` de `message_end`	événement OTel `claude_code.api_request`
Durée d’un outil	à calculer par `toolCallId`	`duration_ms` de `claude_code.tool_result`
Clé de session	`--session-id`	`--session-id` (UUID)
Historique natif	arbre fait pour reprendre	JSONL au format interne, purgé après 30 jours

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

L’extension rejoint le poste de pilotage, le miroir la salle de contrôle. Le runner reste propriétaire du graphe et la référence des verdicts et des coûts. L’extension observe sans rien décider. Coût : zéro jeton. Gain : plus de re-run « pour voir » quand une phase coûte trop cher.

Travaux pratiques — la pièce du jour

Une pièce complète à poser dans le repo compagnon plume-factory, qui devient, chapitre après chapitre, votre usine logicielle agentique. Aujourd’hui, trois fichiers.

Pièce — `.pi/extensions/factory-obs.ts`

Le capteur. Un journal par session, une ligne par événement, le miroir lancé à la fermeture. Aucune dépendance npm. Sous Bun, le fichier exécute sa propre gate.

// .pi/extensions/factory-obs.ts — la trace harnais : chaque événement, au moment où il arrive.
// Même loi que le tracer du chapitre 18 : le JSONL est le brut, écrit au fil de l'eau ;
// factory.db est le miroir, tenu par adws/adw_modules/harness_trace.py — lancé ici à la fermeture.
import { spawnSync } from "node:child_process";
import { randomUUID } from "node:crypto";
import { appendFileSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";

const HARNESS_DIR = join("adws", "adw_data", "traces", "harness");
const INGEST_SCRIPT = join("adws", "adw_modules", "harness_trace.py");
const HEAD_CHARS = 120; // ce que l'on garde d'un prompt ou d'un argument : de quoi reconnaître, pas relire

export interface HarnessEvent {
  event_id: string;
  ts: string;
  session_id: string;
  type: string;
  name: string;
  payload: Record<string, unknown>;
}

// ---------- le journal : un fichier par session, une ligne par événement ----------

export class HarnessJournal {
  readonly sessionId: string;
  readonly file: string;
  count = 0;

  constructor(sessionId: string, dir: string) {
    this.sessionId = sessionId;
    mkdirSync(dir, { recursive: true });
    this.file = join(dir, `${sessionId}.jsonl`);
  }

  write(type: string, name: string, payload: Record<string, unknown> = {}): HarnessEvent {
    const line: HarnessEvent = {
      event_id: `hev_${randomUUID().replace(/-/g, "").slice(0, 12)}`,
      ts: new Date().toISOString(),
      session_id: this.sessionId,
      type,
      name,
      payload,
    };
    // Ajout synchrone : la ligne est sur le disque avant que l'événement suivant n'arrive.
    appendFileSync(this.file, `${JSON.stringify(line)}\n`, "utf8");
    this.count += 1;
    return line;
  }
}

// ---------- ce que l'on garde d'un message : des chiffres, pas du contenu ----------

type Usage = { input?: number; output?: number; cacheRead?: number; cacheWrite?: number; totalTokens?: number; cost?: { total?: number } };

export function usageOf(usage: Usage | undefined): Record<string, number> {
  return {
    input: usage?.input ?? 0,
    output: usage?.output ?? 0,
    cache_read: usage?.cacheRead ?? 0,
    cache_write: usage?.cacheWrite ?? 0,
    total_tokens: usage?.totalTokens ?? 0,
    cost_usd: usage?.cost?.total ?? 0,
  };
}

function textLength(content: unknown): number {
  if (!Array.isArray(content)) return 0;
  return content.reduce((n: number, part: { type?: string; text?: string }) =>
    n + (part?.type === "text" ? (part.text ?? "").length : 0), 0);
}

function head(value: unknown): string {
  const text = typeof value === "string" ? value : JSON.stringify(value ?? "");
  return text.length > HEAD_CHARS ? `${text.slice(0, HEAD_CHARS)}…` : text;
}

// Le miroir : brut → factory.db, par le module Python de la salle de contrôle. Best effort :
// si uv manque ou tarde, le brut est déjà écrit et le prochain --ingest rattrape.
export function mirror(cwd: string): boolean {
  if (!existsSync(join(cwd, INGEST_SCRIPT))) return false;
  const result = spawnSync("uv", ["run", INGEST_SCRIPT, "--ingest", "--quiet"],
    { cwd, stdio: "ignore", timeout: 20_000 });
  return result.status === 0;
}

// ---------- l'extension : observer, écrire, ne rien décider ----------

export default function (pi: ExtensionAPI) {
  let journal: HarnessJournal | null = null;
  const toolClock = new Map<string, number>();
  const totals = { turns: 0, tools: 0, tokens: 0, cost_usd: 0 };

  pi.on("session_start", async (event, ctx) => {
    journal = new HarnessJournal(ctx.sessionManager.getSessionId(), join(ctx.cwd, HARNESS_DIR));
    totals.turns = totals.tools = totals.tokens = totals.cost_usd = 0;
    journal.write("session_start", event.reason, {
      mode: ctx.mode,
      model: ctx.model ? `${ctx.model.provider}/${ctx.model.id}` : null,
      cwd: ctx.cwd,
      session_file: ctx.sessionManager.getSessionFile() ?? null,
    });
    if (ctx.hasUI) ctx.ui.setStatus("factory-obs", "◉ trace");
  });

  pi.on("before_agent_start", async (event) => {
    journal?.write("prompt", "user", { chars: event.prompt.length, head: head(event.prompt) });
  });

  pi.on("turn_start", async (event) => {
    totals.turns += 1;
    journal?.write("turn_start", `turn ${event.turnIndex}`, { turn: event.turnIndex });
  });

  pi.on("turn_end", async (event) => {
    journal?.write("turn_end", `turn ${event.turnIndex}`, { turn: event.turnIndex, tool_results: event.toolResults?.length ?? 0 });
  });

  // La facture : chaque message d'assistant porte son usage, calculé par pi depuis son registre.
  pi.on("message_end", async (event) => {
    const message = event.message as { role: string; provider?: string; model?: string; usage?: Usage; stopReason?: string; content?: unknown[] };
    if (message.role !== "assistant") return;
    const usage = usageOf(message.usage);
    totals.tokens += usage.total_tokens;
    totals.cost_usd += usage.cost_usd;
    const toolCalls = Array.isArray(message.content)
      ? message.content.filter((part) => (part as { type?: string })?.type === "toolCall").length
      : 0;
    journal?.write("assistant", `${message.provider ?? "?"}/${message.model ?? "?"}`, {
      ...usage, stop_reason: message.stopReason ?? null, tool_calls: toolCalls, chars: textLength(message.content),
    });
  });

  pi.on("tool_execution_start", async (event) => {
    toolClock.set(event.toolCallId, Date.now());
    journal?.write("tool_start", event.toolName, { tool_call_id: event.toolCallId, args: head(event.args) });
  });

  pi.on("tool_execution_end", async (event) => {
    const started = toolClock.get(event.toolCallId);
    toolClock.delete(event.toolCallId);
    totals.tools += 1;
    const result = event.result as { content?: unknown } | undefined;
    journal?.write("tool_end", event.toolName, {
      tool_call_id: event.toolCallId,
      duration_ms: started ? Date.now() - started : null,
      is_error: event.isError,
      result_chars: textLength(result?.content),
    });
  });

  pi.on("session_compact", async (event) => {
    journal?.write("compaction", event.reason, { tokens_before: event.compactionEntry?.tokensBefore ?? null });
  });

  pi.on("agent_settled", async (_event, ctx) => {
    journal?.write("agent_settled", "idle", { ...totals });
    if (ctx.hasUI) ctx.ui.setStatus("factory-obs", `◉ ${totals.turns} tours · ${totals.tools} outils · ${totals.cost_usd.toFixed(3)} $`);
  });

  pi.on("session_shutdown", async (event, ctx) => {
    if (!journal) return;
    journal.write("session_shutdown", event.reason, { ...totals, events: journal.count + 1 });
    mirror(ctx.cwd); // ~1 s, zéro jeton : le miroir est à jour quand pi rend la main.
    journal = null;
  });

  pi.registerCommand("factory-obs", {
    description: "Où en est la trace harnais de cette session",
    handler: async (_args, ctx) => {
      if (!ctx.hasUI) return;
      ctx.ui.notify(journal
        ? `${journal.count} événements dans ${journal.file} · ${totals.turns} tours · ${totals.tools} outils · ${totals.tokens} jetons · ${totals.cost_usd.toFixed(4)} $`
        : "aucun journal ouvert", "info");
    },
  });
}

// ---------- la gate du fichier : `bun .pi/extensions/factory-obs.ts`, zéro jeton ----------

if (import.meta.main) {
  const dir = mkdtempSync(join(tmpdir(), "factory-obs-"));
  try {
    const journal = new HarnessJournal("selftest-session", dir);
    journal.write("session_start", "startup", { mode: "print" });
    journal.write("assistant", "openrouter/z-ai/glm-5.3",
      usageOf({ input: 1200, output: 300, totalTokens: 1500, cost: { total: 0.003 } }));
    journal.write("tool_end", "bash", { duration_ms: 42, is_error: false });
    journal.write("session_shutdown", "quit", { turns: 1, tools: 1 });
    const lines = readFileSync(journal.file, "utf8").trim().split("\n").map((l) => JSON.parse(l) as HarnessEvent);
    const ids = new Set(lines.map((l) => l.event_id));
    const ok = lines.length === 4 && ids.size === 4 && lines.every((l) => l.session_id === "selftest-session" && l.ts && l.type);
    console.log(`factory-obs ${ok ? "OK" : "KO"} — ${lines.length} événements, ${ids.size} identifiants uniques, session ${lines[0]?.session_id}`);
    process.exit(ok ? 0 : 1);
  } finally {
    rmSync(dir, { recursive: true, force: true });
  }
}

Pièce — `adws/adw_modules/harness_trace.py`

Le miroir, en bibliothèque standard. Il remplit harness_events, crée la vue harness_by_phase et relit le dernier run. Lancé sans argument, il exécute sa gate.

# /// script
# requires-python = ">=3.11"
# ///
"""harness_trace — le miroir SQLite de la trace harnais (annexe A4).

Le capteur (.pi/extensions/factory-obs.ts) ecrit le brut : un JSONL par
session pi sous adws/adw_data/traces/harness/. Ce module le miroite dans
factory.db — table harness_events — et le joint aux phases du runner par
la session, via la vue harness_by_phase. Rejouable : INSERT OR IGNORE sur
event_id, un miroir perdu se reconstruit depuis le brut.

    uv run adws/adw_modules/harness_trace.py            # auto-test, zero token
    uv run adws/adw_modules/harness_trace.py --ingest   # brut -> factory.db
    uv run adws/adw_modules/harness_trace.py --last     # le dernier run, vu du harnais
"""
from __future__ import annotations

import json
import sqlite3
import sys
import tempfile
from pathlib import Path

DB_PATH = Path("adws/adw_data/factory.db")
HARNESS_DIR = Path("adws/adw_data/traces/harness")

SCHEMA = """
CREATE TABLE IF NOT EXISTS harness_events (
  event_id     TEXT PRIMARY KEY,
  session_id   TEXT,
  type         TEXT,
  name         TEXT,
  payload_json TEXT,
  ts           TEXT
);
CREATE INDEX IF NOT EXISTS harness_events_session ON harness_events (session_id, ts);
"""

# La jointure vit dans la vue, calculee a chaque lecture : le runner ecrit
# phase_session APRES le retour du harnais — donc apres le miroir. Une
# colonne adw_id figee dans harness_events serait fausse la plupart du temps.
VIEW = """
CREATE VIEW IF NOT EXISTS harness_by_phase AS
SELECT p.adw_id, p.phase_id, h.*
FROM harness_events h
JOIN (SELECT DISTINCT adw_id, phase_id,
             json_extract(payload_json, '$.session_id') AS session_id
      FROM events WHERE type = 'phase_session') p
  ON p.session_id = h.session_id;
"""


def connect(db_path: Path = DB_PATH) -> sqlite3.Connection:
    db_path = Path(db_path)
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path, isolation_level=None)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA busy_timeout=5000;")
    conn.executescript(SCHEMA)
    # La vue ne peut exister qu'avec la table events du tracer (ch. 18).
    has_events = conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name='events'").fetchone()
    if has_events:
        conn.executescript(VIEW)
    return conn


def ingest(conn: sqlite3.Connection, harness_dir: Path = HARNESS_DIR) -> int:
    """Miroite chaque ligne JSONL du harnais ; rend le nombre de lignes nouvelles."""
    before = conn.total_changes
    for jsonl in sorted(Path(harness_dir).glob("*.jsonl")):
        rows = []
        for raw in jsonl.read_text(encoding="utf-8").splitlines():
            raw = raw.strip()
            if not raw:
                continue
            try:
                line = json.loads(raw)
            except json.JSONDecodeError:
                continue  # une ligne coupee par un kill : le reste du fichier vaut toujours
            rows.append((line.get("event_id"), line.get("session_id"), line.get("type"),
                         line.get("name"),
                         json.dumps(line.get("payload", {}), ensure_ascii=False),
                         line.get("ts")))
        if rows:
            conn.executemany(
                "INSERT OR IGNORE INTO harness_events"
                " (event_id, session_id, type, name, payload_json, ts) VALUES (?,?,?,?,?,?)",
                rows)
    return conn.total_changes - before


def last_run(conn: sqlite3.Connection) -> int:
    """Le dernier run du journal, phase par phase, avec ce que le harnais en a vu."""
    if not conn.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name='runs'").fetchone():
        print("aucun journal — lancez un ADW, puis revenez", file=sys.stderr)
        return 1
    row = conn.execute(
        "SELECT adw_id, adw_name, status, cost_usd FROM runs"
        " ORDER BY started_at DESC LIMIT 1").fetchone()
    if row is None:
        print("journal vide — lancez un ADW, puis revenez", file=sys.stderr)
        return 1
    adw_id, adw_name, status, cost = row
    print(f"run {adw_id} ({adw_name}) — {status} — ~{cost or 0:.4f} $ (runner)")
    for phase_id, seq, name, kind, pcost in conn.execute(
            "SELECT phase_id, seq, name, kind, cost_usd FROM phases"
            " WHERE adw_id=? ORDER BY seq", (adw_id,)):
        turns, tools, tokens, usd = conn.execute(
            "SELECT SUM(type='turn_start'), SUM(type='tool_end'),"
            " COALESCE(SUM(json_extract(payload_json,'$.total_tokens')),0),"
            " COALESCE(SUM(json_extract(payload_json,'$.cost_usd')),0)"
            " FROM harness_by_phase WHERE phase_id=?", (phase_id,)).fetchone()
        if not turns and not tools:
            why = "phase code" if kind == "code" else "pas de trace harnais (claude ?)"
            print(f"  {seq:02d} {name:<22} {kind:<6} —  {why}")
            continue
        print(f"  {seq:02d} {name:<22} {kind:<6} {turns or 0} tours, {tools or 0} outils,"
              f" {int(tokens)} jetons, ~{usd:.4f} $ (harnais) / ~{pcost or 0:.4f} $ (runner)")
    return 0


def selftest() -> int:
    """La gate du module : brut synthetique, miroir, jointure, rejeu — zero token."""
    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        conn = connect(tmp / "test.db")
        # Une table events minimale, comme le tracer la laisserait, avec la cle de jointure.
        conn.executescript(
            "CREATE TABLE IF NOT EXISTS events (event_id TEXT PRIMARY KEY, adw_id TEXT,"
            " phase_id TEXT, type TEXT, name TEXT, payload_json TEXT, ts TEXT);")
        conn.execute(
            "INSERT INTO events VALUES ('evt_1','selftest','selftest-01','phase_session',"
            "'prompt','{\"session_id\": \"sess-A\", \"attempt\": 1}','2026-01-01T00:00:00Z')")
        conn.executescript(VIEW)
        harness = tmp / "harness"
        harness.mkdir()
        lines = [
            {"event_id": "hev_1", "ts": "2026-01-01T00:00:01Z", "session_id": "sess-A",
             "type": "turn_start", "name": "turn 0", "payload": {"turn": 0}},
            {"event_id": "hev_2", "ts": "2026-01-01T00:00:02Z", "session_id": "sess-A",
             "type": "assistant", "name": "m", "payload": {"total_tokens": 1500, "cost_usd": 0.01}},
            {"event_id": "hev_3", "ts": "2026-01-01T00:00:03Z", "session_id": "sess-B",
             "type": "tool_end", "name": "bash", "payload": {}},
        ]
        (harness / "sess.jsonl").write_text(
            "\n".join(json.dumps(l) for l in lines) + "\n", encoding="utf-8")
        first = ingest(conn, harness)
        second = ingest(conn, harness)   # rejeu : rien de nouveau
        joined = conn.execute(
            "SELECT COUNT(*) FROM harness_by_phase WHERE adw_id='selftest'").fetchone()[0]
        tokens = conn.execute(
            "SELECT SUM(json_extract(payload_json,'$.total_tokens'))"
            " FROM harness_by_phase WHERE phase_id='selftest-01'").fetchone()[0]
        ok = first == 3 and second == 0 and joined == 2 and tokens == 1500
        print(f"harness_trace {'OK' if ok else 'KO'} — {first} lignes miroitees,"
              f" {second} au rejeu, {joined} jointes a la phase, {tokens} jetons")
        return 0 if ok else 1


if __name__ == "__main__":
    if "--ingest" in sys.argv:
        added = ingest(connect())
        if "--quiet" not in sys.argv:
            print(f"{added} evenement(s) harnais miroite(s) dans {DB_PATH}")
        raise SystemExit(0)
    if "--last" in sys.argv:
        raise SystemExit(last_run(connect()))
    raise SystemExit(selftest())

Pièce — `adws/adw_modules/runner.py`

Cette version remplace celle du chapitre 20. Seul ajout : l’événement phase_session, écrit une fois par session après chaque tentative d’une phase agent. L’API ne change pas.

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

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

Version annexe A4 : la cle de jointure. Apres chaque tentative, le runner
relie la phase a la session du harnais (evenement phase_session) — c'est
par elle que la trace harnais rejoint le journal. L'API ne bouge pas :
vos ADW des chapitres 8 a 27 tournent tels quels.
"""
from __future__ import annotations

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

from .tracer import Tracer


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


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


@dataclass
class Run:
    """L'etat partage d'un run : resultats des phases, sessions, cout, jetons."""
    adw_id: str
    results: dict[str, Any] = field(default_factory=dict)
    sessions: dict[str, str] = field(default_factory=dict)  # phase -> session_id
    cost_usd: float = 0.0
    # Le compteur de jetons voyage comme le cout : une action qui releve
    # l'usage de son harnais le credite (run.tokens += ...) ; une action
    # qui ne releve rien laisse zero — jamais un chiffre invente.
    tokens: int = 0
    tracer: Tracer | None = None    # injectable pour les tests ; None = journal standard

    def __post_init__(self) -> None:
        if self.tracer is None:
            self.tracer = Tracer()

    def execute(self, phases: list[PhaseSpec]) -> int:
        """Sequence les phases declarees. Arret a la premiere phase en echec definitif."""
        if not phases:
            print("aucune phase declaree — un ADW vide n'est pas un ADW", file=sys.stderr)
            return 1
        # Le nom de l'ADW et la demande sont releves sur la ligne de commande :
        # zero changement dans vos scripts, et la trace sait deja qui tourne.
        adw_name = Path(sys.argv[0]).stem if sys.argv and sys.argv[0] else ""
        request = " ".join(sys.argv[1:])[:500]
        self.tracer.run_start(self.adw_id, adw_name, request)
        for seq, spec in enumerate(phases, start=1):
            if spec.kind not in ("agent", "code"):
                print(f"[{self.adw_id}] {spec.name} : kind inconnu {spec.kind!r}",
                      file=sys.stderr)
                self.tracer.run_finish(self.adw_id, ok=False,
                                       cost_usd=self.cost_usd, tokens=self.tokens)
                return 1
            if not self._run_phase(seq, spec):
                print(f"[{self.adw_id}] ECHEC en phase {spec.name} — arret du run",
                      file=sys.stderr)
                self.tracer.run_finish(self.adw_id, ok=False,
                                       cost_usd=self.cost_usd, tokens=self.tokens)
                return 1
        # La ligne de bilan reste l'API de l'oeil et du banc (ch. 17) —
        # la trace s'ajoute, elle ne retire rien.
        print(f"[{self.adw_id}] run vert — cout total ~{self.cost_usd:.4f} $",
              file=sys.stderr)
        self.tracer.run_finish(self.adw_id, ok=True,
                               cost_usd=self.cost_usd, tokens=self.tokens)
        return 0

    def _bind_session(self, phase_id: str, name: str, attempt: int,
                      bound: set[str]) -> None:
        """La cle de jointure de la trace harnais (annexe A4).

        L'action memorise la session dans run.sessions avant de valider ;
        le runner la declare au journal une fois par session — une reprise
        en session vivante garde la meme cle, une session neuve en ecrit une
        autre. Sans session (phase code), rien n'est ecrit.
        """
        session_id = self.sessions.get(name)
        if not session_id or session_id in bound:
            return
        bound.add(session_id)
        self.tracer.event(self.adw_id, "phase_session", name, phase_id=phase_id,
                          payload={"session_id": session_id, "attempt": attempt})

    def _run_phase(self, seq: int, spec: PhaseSpec) -> bool:
        # Retenter une phase code n'a pas de sens : meme entree, meme sortie.
        # Seules les phases agent ont droit aux reprises — en session vivante.
        attempts = 1 + (spec.retries if spec.kind == "agent" else 0)
        phase_id = self.tracer.phase_start(self.adw_id, seq, spec.name,
                                           spec.kind, retries=attempts - 1)
        bound: set[str] = set()
        for attempt in range(attempts):
            clock = time.monotonic()
            # Le grain comptable (ch. 20) : ce que CETTE tentative ajoute aux
            # compteurs du run — mesure en delta, vos actions ne changent pas.
            cost_before, tokens_before = self.cost_usd, self.tokens
            label = f"{spec.name} ({spec.kind}, tentative {attempt + 1}/{attempts})"
            try:
                # L'action recoit le run (etat partage) et le numero de tentative :
                # a la tentative 1, une phase agent envoie la demande ; ensuite,
                # elle envoie la correction dans la MEME session.
                self.results[spec.name] = spec.action(self, attempt)
            except PhaseFailure as error:
                # La session est memorisee AVANT la validation (ch. 8) : meme en
                # echec, la cle de jointure existe et la trace harnais se relie.
                self._bind_session(phase_id, spec.name, attempt + 1, bound)
                print(f"[{self.adw_id}] {label} : echec — {error}", file=sys.stderr)
                self.tracer.phase_attempt(phase_id, self.adw_id, spec.name,
                                          attempt + 1, ok=False, error=str(error),
                                          cost_usd=self.cost_usd - cost_before,
                                          tokens=self.tokens - tokens_before)
                continue
            self._bind_session(phase_id, spec.name, attempt + 1, bound)
            duration = time.monotonic() - clock
            print(f"[{self.adw_id}] {label} : OK en {duration:.1f} s", file=sys.stderr)
            self.tracer.phase_attempt(phase_id, self.adw_id, spec.name,
                                      attempt + 1, ok=True,
                                      cost_usd=self.cost_usd - cost_before,
                                      tokens=self.tokens - tokens_before)
            return True
        return False

La gate du TP

Depuis la racine de plume-factory, projet approuvé (chapitre A1). Cinq commandes, identiques dans bash et PowerShell.

bun .pi/extensions/factory-obs.ts
uv run adws/adw_modules/harness_trace.py
uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ? Reponds en une phrase."
uv run adws/adw_modules/harness_trace.py --last
sqlite3 adws/adw_data/factory.db "SELECT type, COUNT(*) FROM harness_by_phase WHERE adw_id = (SELECT adw_id FROM runs ORDER BY started_at DESC LIMIT 1) GROUP BY type ORDER BY type;"

Résultat attendu :

factory-obs OK — 4 événements, 4 identifiants uniques
harness_trace OK — 3 lignes miroitees, 0 au rejeu, 2 jointes a la phase, 1500 jetons
un run vert, et --last qui montre la phase prompt avec deux coûts qui se recoupent
au moins un session_start, un assistant, un tool_end et un session_shutdown pour ce run

Sans client sqlite3, --last suffit. Coût : rien pour les deux gates, moins d’un centime et une vingtaine de secondes pour l’ADW.

Quiz — teste tes connaissances

Annexe — Harnais pi 7 questions Objectif : 5/7 minimum

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.