Observer le harnais de l'intérieur

Au chapitre 19, vous avez lu un run en swim lanes : dix-huit phases, leur verdict, leur coût, leur durée. Mais ouvrez la phase build qui a mis quatre minutes et coûté douze centimes : le journal de l’usine vous dit qu’elle a réussi à la deuxième tentative, et rien de plus. Combien de tours ? Combien d’appels à bun test ? Le modèle a-t-il relu la spec trois fois ? Y a-t-il eu une compaction au milieu ? Tout cela s’est passé à l’intérieur de la phase, côté harnais, là où le tracer du chapitre 18 n’a pas de capteur. À la fin de ce chapitre, vous saurez faire remonter cette histoire intérieure sans dépenser un jeton : une extension écrit chaque événement du cycle de vie de pi au moment où il arrive, et un miroir SQLite le range dans factory.db, joint aux phases du runner par la clé que vous avez toujours eue sous la main : la session. La pièce du jour prolonge le damage control d’hier : là où A3 décidait avant le geste, A4 se souvient de tout, et le runner Python reste, comme toujours, le seul propriétaire du graphe.

Streamer les événements du cycle de vie

L’idée en une phrase

Une extension abonnée aux événements observateurs de pi (session_start, turn_start, message_end, tool_execution_start / tool_execution_end, session_compact, agent_settled, session_shutdown) écrit une ligne JSONL par événement, au fil de l’eau, dans adws/adw_data/traces/harness/. C’est le capteur du nœud pi de l’usine, côté code déterministe du harnais, et il ne bloque jamais rien.

Points clés

• Les mêmes règles que le tracer du chapitre 18. Écrire quand ça se passe, jamais en fin de session : un pi -p tué par le timeout du port harnais laisse sa trace jusqu’à l’instant du gel. Une ligne par événement, typée, datée à la milliseconde, un fichier par session.
• message_end est la facture. Chaque message d’assistant porte son usage : jetons d’entrée, de sortie, de cache, et cost.total calculé par pi depuis son registre de modèles. C’est le même chiffre que le pied de page du chapitre A2 additionne, et ici vous le gardez.
• Les outils se chronomètrent par toolCallId. tool_execution_start et tool_execution_end partagent cet identifiant, et l’extension note l’heure au départ et calcule la durée à l’arrivée. En mode parallèle, les fins arrivent dans l’ordre d’achèvement, pas dans l’ordre d’appel, et l’identifiant règle la question.
• Ce que vous n’enregistrez pas compte autant. Pas le contenu des résultats d’outils (un read de 40 Ko par ligne rendrait le journal illisible), pas le prompt entier (le runner l’a déjà, dans sa demande) : des tailles, des durées, des compteurs, et 120 caractères de tête pour reconnaître un tour.
• Zéro dépendance, zéro jeton. appendFileSync sur un fichier ouvert en ajout : quelques microsecondes par événement. En -p et --mode json, aucune ligne de plus sur la sortie standard : le flux JSON que lit l’adaptateur du chapitre 7 reste intact.

Exemple concret

Phase build d’un SDLC, workhorse GLM 5.3 à 1,40 $ le million de jetons en entrée et 4,40 $ en sortie au moment d’écrire. Le runner voit : tentative 1 en échec (gate tests rouge), tentative 2 verte, ~12 centimes, 4 minutes. La trace harnais raconte le reste : à la tentative 1, sept tours, onze appels d’outils dont deux bun test, un message d’assistant à 38 000 jetons d’entrée, le modèle a relu tout apps/plume/. À la tentative 2, dans la même session, trois tours, quatre outils, et la correction en ~9 000 jetons. Vous savez maintenant où sont passés les douze centimes : les deux tiers dans la relecture initiale, pas dans la correction. Coût de ce savoir : zéro jeton, une quinzaine de lignes JSONL par tour, quelques microsecondes chacune.

Ce que chaque événement laisse dans le journal

Événement pi	Ligne écrite (type)	Ce que la charge utile porte
session_start	session_start	mode (print, json, tui), modèle actif, dossier de travail
before_agent_start	prompt	taille du prompt, ses 120 premiers caractères
turn_start / turn_end	turn_start / turn_end	numéro du tour
message_end (assistant)	assistant	modèle, jetons entrée / sortie / cache, coût, motif d’arrêt, nombre d’appels d’outils
tool_execution_start / _end	tool_start / tool_end	outil, arguments abrégés, durée, erreur ou non, taille du résultat
session_compact	compaction	motif (threshold, overflow, manual), jetons avant
agent_settled / session_shutdown	agent_settled / session_shutdown	totaux de la session : tours, outils, jetons, coût

Config — une ligne du journal harnais

Même forme que la ligne du chapitre 18 : un identifiant, une date, une clé de regroupement (ici la session au lieu du run), un type, un nom, une charge utile. Une ligne se lit seule.

{
  "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}
}

Piège courant : « le harnais garde déjà tout dans son fichier de session, inutile de tracer » est inexact. Le fichier de session de pi est fait pour reprendre une conversation, pas pour la requêter : c’est un arbre d’entrées où le coût d’un tour se recalcule en remontant les branches. La trace harnais est plate, typée et datée pour la question que vous poserez plus tard, et elle survit à un /new, à un --session-dir nettoyé, ou à une session jamais persistée.

---

Brancher la trace harnais sur factory.db

L’idée en une phrase

Le journal harnais rejoint le journal de l’usine par un miroir SQLite : une table harness_events dans factory.db, remplie par harness_trace.py depuis les JSONL, et une vue harness_by_phase qui la joint aux phases du runner grâce à l’identifiant de session que le port harnais choisit lui-même, côté code déterministe, des deux côtés de la couture.

Points clés

• La clé de jointure existait déjà. Depuis le chapitre 7, l’adaptateur pi nomme la session (--session-id) et le runner la mémorise dans run.sessions, mais seulement en mémoire. La pièce du jour fait écrire au runner un événement phase_session de plus : la phase, la tentative, la session. Rien d’autre ne change dans le runner, et vos ADW tournent tels quels.
• Brut d’abord, miroir ensuite, et le miroir est rejouable. INSERT OR IGNORE sur event_id : relancer l’ingestion dix fois ne duplique rien, et un miroir perdu se reconstruit depuis les JSONL, comme au chapitre 18.
• Le miroir se rafraîchit tout seul à la fermeture. À session_shutdown, l’extension lance uv run adws/adw_modules/harness_trace.py --ingest en arrière-plan de sa sortie : environ une seconde, sans jeton. En -p, l’événement est bien émis à la fin du run, et le runner voit pi rendre la main une seconde plus tard, c’est tout. Si uv manque, le brut est déjà écrit et 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 que pi a fermé sa session et miroité ses lignes. Stocker adw_id dans harness_events serait donc faux la plupart du temps. La vue calcule la jointure à chaque requête, et elle est toujours juste.
• Une phase sans trace harnais n’est pas un bug. Les phases code n’ouvrent pas de session, et une phase confiée à Claude Code par le roster n’est pas vue par une extension pi. Le lecteur --last le dit en clair plutôt que d’afficher zéro.

Exemple concret

Lancez uv run adws/adw_prompt.py "Quels fichiers composent apps/plume ?" : une phase agent, une session pi, moins d’un centime, une vingtaine de secondes. Pendant le run, l’extension écrit une quinzaine de lignes dans traces/harness/3f2a9c7e….jsonl. À la fermeture, elle les miroite dans harness_events, puis le runner écrit phase_session et clôt le run. Interrogez la vue : la phase prompt du run, sa session, deux tours, trois outils (ls, read, read), 6 400 jetons, 0,012 $. Le chiffre de coût du runner et la somme des factures de la trace harnais se recoupent, et quand ils divergent vous savez désormais lequel des deux regarder.

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

Une seule version suffit : le miroir est du SQL pur, et la trace harnais n’existe que côté pi (l’annexe équipe ce côté de la couture, et Claude Code garde ses propres transcripts, hors programme). 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 : « autant faire écrire l’extension directement dans factory.db » est tentant. Cela ferait dépendre le harnais d’un pilote SQLite (celui de Node n’est pas celui de Bun, et votre gate bun … ne tournerait plus), et deux processus écrivant la même base pendant qu’un troisième la lit demande plus de soin que la valeur ajoutée. Le JSONL est le contrat : n’importe quel harnais qui sait écrire une ligne peut alimenter le miroir.

---

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

Sur le plan de l’usine, la pièce se pose à cheval sur deux zones : le poste de pilotage (.pi/extensions/factory-obs.ts, à côté du pied de page et du damage control) et la salle de contrôle (adws/adw_modules/harness_trace.py, à côté du tracer du chapitre 18 et du pont OTel du chapitre 20). La couture ne bouge pas : le runner possède le graphe des phases et son journal reste la référence des verdicts et des coûts. L’extension observe le graphe des tours que pi possède à l’intérieur d’une phase, et ne décide de rien. L’enveloppe qui traverse aujourd’hui n’est pas un rapport pour l’agent, c’est une ligne JSONL pour vous, et la clé qui la relie au runner est la session, que le port harnais choisissait déjà. Coût : zéro jeton, quelques microsecondes par événement, une seconde de miroir à la fermeture de chaque session. Ce qu’elle économise : le re-run « pour voir » à plusieurs dizaines de centimes quand une phase coûte trop cher sans qu’on sache pourquoi.

---

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 : le capteur côté pi, le miroir côté Python, et le runner qui apprend à nommer sa session dans la trace.

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

Le capteur du nœud pi. Il ouvre un journal par session à session_start, écrit une ligne par événement observé, et lance le miroir à session_shutdown. Aucune dépendance npm, et le fichier porte sa propre gate sous Bun, qui écrit un journal jetable et le relit. Côté pi, côté code déterministe. En -p et --mode json, rien de plus sur la sortie standard.

// .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. Bibliothèque standard uniquement, aucun import des autres modules : il lit les JSONL du harnais, les range dans harness_events (rejouable), crée la vue harness_by_phase dès que la table events du tracer existe, et relit le dernier run phase par phase. Lancé sans argument, il porte sa propre gate. Entièrement côté déterministe.

# /// 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. Un seul ajout : après chaque tentative d’une phase agent, le runner écrit un événement phase_session qui relie la phase à la session du harnais, une fois par session, jamais deux. Tout le reste est inchangé, PhaseFailure, PhaseSpec, l’API de Run, le grain comptable et les messages stderr : vos ADW des chapitres 8 à 27 tournent tels quels.

"""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, une par ligne, identiques dans bash et PowerShell. Les deux premières ne coûtent rien, la troisième lance un ADW d’une phase, les deux dernières lisent.

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, puis harness_trace OK — 3 lignes miroitees, 0 au rejeu, 2 jointes a la phase, 1500 jetons. L’ADW rend un run vert, --last affiche la phase prompt avec ses tours, ses outils, ses jetons et deux coûts qui se recoupent (harnais et runner), et la requête compte au moins un session_start, un assistant, un tool_end et un session_shutdown pour ce run. Sans client sqlite3, --last suffit. Coût : les gates Bun et Python ne dépensent rien, l’ADW coûte moins d’un centime en une vingtaine de secondes sur le workhorse du chapitre A1, plus environ une seconde de miroir à la fermeture. Variante éco : elle est déjà dans la pièce, le modèle est celui du roster.

---