Aspire : le routeur MultiConnector — vetting en ligne, double file et traces

Ce notebook réalise l’Axe 6 de l’Epic #1210 (semantic-fleet, réactivée le 2026-08-18) : ressusciter la pièce maîtresse du repo MyIntelligenceAgency/semantic-fleet — le MultiConnector de 2023 — comme artefact pédagogique de la série Aspire, ré-instrumenté sous OpenTelemetry.

semantic-fleet vit dans ce dépôt comme sous-module, épinglé au commit 9df3603 (branche stable-from-v0343, baseline propre du tag v0.34.3), dans MyIA.AI.Notebooks/GenAI/SemanticKernel/semantic-fleet — lisible en face de ce notebook.

L’idée du MultiConnector (PR microsoft/semantic-kernel#2323, fermée sans review en janvier 2024) tient en une phrase : un connecteur primaire sert tout le trafic par défaut, et charge de vérifier en ligne (vetting) des connecteurs secondaires moins chers, pour leur déléguer progressivement (offload) les classes de prompts qu’ils savent traiter. En 2026, c’est exactement la question qu’un déploiement multi-modèles arbitre — le plus souvent à la main.

La ligne de parité 2023 → 2026 :

Ce que faisait le code 2023 Ce que ce notebook montre
Stopwatch + ILogger adossés au routeur ActivitySource + spans OTel — la trace distribuée standard
Deux Channel<T> imbriqués (collecte, analyse) Les mêmes canaux, visibles comme spans imbriqués collecte.lot → analyse.pipeline
Coûts/durées par connecteur, vetting en ligne Le POC arithmétique original rejoué fidèlement (primaire 4 opérations, 4 secondaires mono-opération)
Pool de ClientWebSocket + serveur de test HttpListener Un banc de charge loopback réel (sockets véritables, réponses pré-encodées, débit mesuré)

Frontière d’honnêteté — à lire avant toute citation de ce notebook. Tout s’exécute CPU-only, sans aucun endpoint LLM : les connecteurs sont des simulations arithmétiques déterministes, le serveur de flux est un loopback localhost. Les mesures de débit couvrent le pipeline routeur + canaux + sockets, jamais une génération de tokens. Le souvenir des « 20 000 chunks/s » de 2023 n’est pas re-mesuré ici ; le seul chiffre de débit cité dans ce notebook est celui que sa cellule de mesure produit.

Objectifs d’apprentissage

À la fin de ce notebook, vous saurez :

  1. Expliquer le vetting en ligne : comment un primaire omniscient qualifie — ou rejette — des secondaires moins chers, sans supervision humaine ;
  2. Distinguer les deux sémantiques de coalescence d’une double file : accumulation (chaque échantillon compte) contre remplacement (seule la dernière analyse survit) ;
  3. Écrire un debounce ancré sur l’horodatage de l’objet, et dire pourquoi le debounce « au réveil » est la version ratée ;
  4. Instrumenter un pipeline asynchrone avec ActivitySource et un exporteur OTel dans le notebook ;
  5. Expliquer l’architecture pool de sockets + serveur de test à réponse pré-encodée, et ce qu’un tel banc mesure honnêtement.

Prérequis

  • Kernel .net-csharp (SDK .NET 9+) — aucun service externe, aucun secret, aucun Docker ;
  • Les notebooks 03 — Observabilité (spans OTel, sampler) et 04 — Streaming Agent (System.Threading.Channels) sont les voisins directs : les pièges de kernel OTel cités ici y sont démontrés.

Durée estimée : 60 minutes

1. Ce que contient le sous-module — et ce qu’on rejoue

Le sous-module épinglé à 9df3603 est lisible directement ; quatre zones fondent ce notebook :

Zone du sous-module Fichiers Ce qu’on en tire
Mocks arithmétiques dotnet/src/Connectors/Connectors.AI.MultiConnector/ArithmeticMocks/*.cs Le POC : primaire 4 opérations, secondaires mono-opération, coûts/durées distincts
Chemin chaud + collecte Connectors.AI.MultiConnector/MultiTextCompletion.cs:265-275 Le TryWrite non bloquant et la boucle de collecte (coalescence par accumulation)
Analyse différée Connectors.AI.MultiConnector/Analysis/MultiCompletionAnalysisSettings.cs:908-947 La seconde boucle (coalescence par remplacement) et son debounce
Pool + serveur de test Connectors.AI.Oobabooga/Completion/OobaboogaCompletionBase.cs:98-148, Connectors.UnitTests/WebSocketTestServer.cs Le pool ConcurrentBag<ClientWebSocket> et le serveur HttpListener à réponse pré-encodée

Pourquoi rejouer ces patterns plutôt que référencer le paquet NuGet v0.34.3 ? Deux raisons, toutes deux assumées : (a) le livrable demandé est la ré-instrumentation sous OTel — il faut que les spans vivent dans le routeur, la collecte et l’analyse, or le code 2023 n’a aucun ActivitySource (l’axe A9/A10 du dépôt l’a mesuré : zéro hit à l’époque) ; (b) le paquet 2023 cible l’API Semantic Kernel beta, dont la restauration n’apporte rien ici. La simulation qui suit est donc fidèle aux structures citées — chaque cellule qui en reprend une le dit et la référence.

// Le SDK OpenTelemetry (traces) : producteur ActivitySource + ecoute + exporteur.
// Version epinglee 1.11.1, la meme que le notebook 03 de la serie.
// Tout le reste (Channels, HttpListener, WebSockets) est BCL .NET — aucun endpoint externe.
#r "nuget: OpenTelemetry, 1.11.1"

using System.Diagnostics;
using System.Threading;
using System.Threading.Channels;
using System.Net;
using System.Net.Sockets;
using System.Net.WebSockets;
using System.Text;
using System.Text.RegularExpressions;
using System.Collections.Concurrent;
using System.Globalization;
using OpenTelemetry;
using OpenTelemetry.Trace;

Console.WriteLine("Packages charges : OpenTelemetry 1.11.1 (+ BCL .NET : Channels, HttpListener, WebSockets)");
Installing Packages
  • OpenTelemetry
Packages charges : OpenTelemetry 1.11.1 (+ BCL .NET : Channels, HttpListener, WebSockets)

Un seul paquet à restaurer : le SDK OpenTelemetry. Le transport de test (System.Net.HttpListener, ClientWebSocket) et la double file (System.Threading.Channels) sont dans la bibliothèque de base — c’est déjà une leçon : le pipeline 2023 n’a jamais eu besoin d’un framework, et il fonctionne tel quel sur .NET moderne.

2. Le banc d’essai arithmétique : un LLM sans LLM

Le POC 2023 remplace les modèles par de l’arithmétique : un prompt Compute Add(8, 2) attend la réponse 10. Un « connecteur » y est défini par trois choses — les opérations qu’il sait faire, sa latence (simulée par Task.Delay, déterministe), et son coût par requête (un decimal, déterministe). La fidélité au comportement d’un LLM est structurelle : capacités partielles, latences et coûts différenciés — sans aucune dépendance réseau.

// Le moteur du POC, fidele a ArithmeticMocks/ArithmeticEngine.cs du sous-module.
public enum Operation { Add, Subtract, Multiply, Divide }

public static class MoteurArithmetique
{
    public static int Calculer(Operation op, int a, int b) => op switch
    {
        Operation.Add => a + b,
        Operation.Subtract => a - b,
        Operation.Multiply => a * b,
        Operation.Divide => a / b,
        _ => throw new ArgumentOutOfRangeException(nameof(op))
    };

    public static string GenererPrompt(Operation op, int a, int b) =>
        "Compute " + op + "(" + a.ToString(CultureInfo.InvariantCulture) + ", " + b.ToString(CultureInfo.InvariantCulture) + ")";

    public static (Operation op, int a, int b) ParserPrompt(string prompt)
    {
        var m = Regex.Match(prompt, @"Compute (?<operation>\w+)\((?<a>\d+), (?<b>\d+)\)");
        if (!m.Success) throw new ArgumentException("Prompt invalide : " + prompt);
        return (Enum.Parse<Operation>(m.Groups["operation"].Value),
                int.Parse(m.Groups["a"].Value, CultureInfo.InvariantCulture),
                int.Parse(m.Groups["b"].Value, CultureInfo.InvariantCulture));
    }
}

// Les operandes discriminateurs du notebook 03 du sous-module : 8 et 2 donnent
// quatre resultats TOUS distincts — une reponse fausse ne coincide jamais avec la bonne.
foreach (Operation op in Enum.GetValues<Operation>())
{
    var prompt = MoteurArithmetique.GenererPrompt(op, 8, 2);
    Console.WriteLine(prompt.PadRight(28) + " => " + MoteurArithmetique.Calculer(op, 8, 2));
}
Compute Add(8, 2)            => 10
Compute Subtract(8, 2)       => 6
Compute Multiply(8, 2)       => 16
Compute Divide(8, 2)         => 4

Lecture du résultat : pourquoi 8 et 2

Les quatre sorties sont 10, 6, 16, 4 — toutes différentes. C’est la propriété qui rend le vetting mécanique possible : quand un secondaire qui ne sait faire que Add répond 11 à Compute Add(8, 2), la bonne réponse 10 n’est égale à aucune autre réponse possible du banc — l’erreur est toujours détectable par simple comparaison. Un seul prompt par opération suffit donc à qualifier un connecteur, ce qui explique le réglage 2023 NbPromptTests = 1. Le couple (8, 2) n’est pas une convention esthétique : c’est un jeu de tests complet en quatre prompts.

La flotte : un primaire omniscient, quatre spécialistes

La flotte reprend les chiffres exacts du notebook 03 du sous-module. Le primaire sait faire les quatre opérations mais est 10 fois plus lent et 2 fois plus cher que chaque spécialiste ; chaque secondaire ne sait faire qu’une opération. La question du routeur est alors purement économique : pour chaque classe de prompt, qui peut répondre — à quelle latence, à quel prix, sans se tromper ?

// Un connecteur = capacites + latence simulee + cout, fidele a ArithmeticCompletionService.cs.
// (Pas d'annotations nullable : le contexte script du kernel n'active pas #nullable.)
public sealed class ConnecteurArithmetique
{
    public ConnecteurArithmetique(string nom, IReadOnlySet<Operation> operations, TimeSpan delai, decimal cout,
        Func<Operation, int, int, int> substitut = null)
    { this.Nom = nom; this.Operations = operations; this.Delai = delai; this.Cout = cout; this.Substitut = substitut; }

    public string Nom { get; }
    public IReadOnlySet<Operation> Operations { get; }
    public TimeSpan Delai { get; }                                  // latence simulee, determine
    public decimal Cout { get; }                                    // cout par requete, determine
    public Func<Operation, int, int, int> Substitut { get; }        // moteur corrompu (section 8)

    public bool SaitFaire(Operation op) => this.Operations.Contains(op);

    public async Task<string> RepondreAsync(string prompt)
    {
        await Task.Delay(this.Delai);                               // la "latence provider" du POC
        var (op, a, b) = MoteurArithmetique.ParserPrompt(prompt);
        if (!this.SaitFaire(op))
            throw new InvalidOperationException(this.Nom + " ne sait pas faire " + op);
        var resultat = this.Substitut is null ? MoteurArithmetique.Calculer(op, a, b) : this.Substitut(op, a, b);
        return resultat.ToString(CultureInfo.InvariantCulture);
    }
}

public static class Flotte
{
    // Les chiffres du notebook 03 du sous-module : primaire 20 ms / 0.02 u, secondaires 2 ms / 0.01 u.
    public static List<ConnecteurArithmetique> PocStandard() => new()
    {
        new("Primaire", Enum.GetValues<Operation>().ToHashSet(), TimeSpan.FromMilliseconds(20), 0.02m),
        new("Secondaire-Add",      new HashSet<Operation> { Operation.Add },      TimeSpan.FromMilliseconds(2), 0.01m),
        new("Secondaire-Subtract", new HashSet<Operation> { Operation.Subtract }, TimeSpan.FromMilliseconds(2), 0.01m),
        new("Secondaire-Multiply", new HashSet<Operation> { Operation.Multiply }, TimeSpan.FromMilliseconds(2), 0.01m),
        new("Secondaire-Divide",   new HashSet<Operation> { Operation.Divide },   TimeSpan.FromMilliseconds(2), 0.01m),
    };
}

foreach (var c in Flotte.PocStandard())
{
    var ops = string.Join(",", c.Operations.Select(o => o.ToString()).OrderBy(s => s, StringComparer.Ordinal));
    Console.WriteLine(c.Nom.PadRight(21) + ("ops=[" + ops + "]").PadRight(36)
        + " delai=" + c.Delai.TotalMilliseconds.ToString(CultureInfo.InvariantCulture) + " ms"
        + "  cout=" + c.Cout.ToString("0.00", CultureInfo.InvariantCulture) + " u/req");
}
Primaire             ops=[Add,Divide,Multiply,Subtract]   delai=20 ms  cout=0.02 u/req
Secondaire-Add       ops=[Add]                            delai=2 ms  cout=0.01 u/req
Secondaire-Subtract  ops=[Subtract]                       delai=2 ms  cout=0.01 u/req
Secondaire-Multiply  ops=[Multiply]                       delai=2 ms  cout=0.01 u/req
Secondaire-Divide    ops=[Divide]                         delai=2 ms  cout=0.01 u/req

Lecture du résultat : la structure économique du routage

Cinq connecteurs, deux régimes. Le primaire paie 0.02 u et 20 ms par requête quoi qu’il arrive ; un spécialiste paie la moitié — 0.01 u et 2 ms — mais uniquement sur son opération. Si le vetting confirme les quatre spécialistes, un cycle des quatre opérations passe de 4 x 0.02 = 0.08 u à 4 x 0.01 = 0.04 u (coût divisé par deux) et le délai simulé cumulé de 80 ms à 8 ms. Ce gain n’est pas une hypothèse : les deux passes de la section 4 le mesurent. Et le prix d’entrée est la confiance — que le primaire doit construire en vérifiant chaque secondaire, opération par opération.

3. Le routeur : plan de routage et chemin chaud

Le routeur 2023 tient en deux décisions. D’abord un plan de routage qui associe une signature de prompt — ici, l’opération — à un connecteur ; vide au départ, donc tout passe par le primaire. Ensuite un chemin chaud qui journalise chaque appel dans un canal non borné par un simple TryWrite : la complétion rend la main dès sa réponse livrée, la journalisation et le vetting vivent en aval, découplés. C’est la raison pour laquelle journaliser ne coûte rien au débit — tout ce qui est mesuré par la section 7 est en aval de cette décision.

// L'ActivitySource : le producteur de spans, nomme comme un service (continuite Aspire03).
ActivitySource source = new("Aspire07.SemanticFleet");

// Exporteur "dans le notebook" : capte chaque span a sa fermeture dans une liste que les
// cellules suivantes peuvent lire — la memoire du notebook joue le role du backend de traces.
public sealed class ExporteurNotebook : BaseExporter<Activity>
{
    private readonly object _verrou = new();
    private readonly List<Activity> _spans = new();

    public override ExportResult Export(in Batch<Activity> lot)
    {
        lock (this._verrou) { foreach (var span in lot) this._spans.Add(span); }
        return ExportResult.Success;
    }

    public List<Activity> Capturer() { lock (this._verrou) return new(this._spans); }
    public List<Activity> CapturerEtVider() { lock (this._verrou) { var copie = new List<Activity>(this._spans); this._spans.Clear(); return copie; } }
}

var exporteur = new ExporteurNotebook();

// Les deux precautions du notebook 03 restent vraies ici (contexte kernel) :
// - AlwaysOnSampler : le kernel maintient une Activity parent ambiante ; le sampler
//   ParentBased par defaut refuserait tout span local -> StartActivity rendrait null ;
// - SimpleActivityExportProcessor : export a la FERMETURE du span ; le processeur batch
//   par defaut ne flush jamais dans un kernel vivant.
var provider = Sdk.CreateTracerProviderBuilder()
    .AddSource("Aspire07.SemanticFleet")
    .SetSampler(new AlwaysOnSampler())
    .AddProcessor(new SimpleActivityExportProcessor(exporteur))
    .Build();

Console.WriteLine("Telemetrie armee : source Aspire07.SemanticFleet, exporteur memoire branche");
Telemetrie armee : source Aspire07.SemanticFleet, exporteur memoire branche

Lecture de l’instrumentation : un exporteur pour le notebook

Trois choix à remarquer. L’exporteur dérive de BaseExporter<Activity> — c’est le même contrat que l’exporteur console ou OTLP ; seules la destination (une liste en mémoire) et sa lecture (les cellules suivantes) changent. Les deux garde-fous commentés viennent du notebook 03 : sampler forcé et export à la fermeture ne sont pas du pédagogisme — ce sont les deux façons dont un kernel .NET Interactive casse silencieusement une chaîne OTel (spans à null, batch jamais flushé). Enfin — c’est le point A9/A10 du dépôt — le code 2023 de semantic-fleet n’avait rien de tout cela : ses mesures vivaient dans des Stopwatch et des journaux. La ré-instrumentation consiste précisément à remplacer chaque Stopwatch par un span qui porte ses propres tags.

// L'echantillon journalise par le chemin chaud (l'equivalent du ConnectorTest 2023),
// et le lot qui traverse le deuxieme etage.
public readonly record struct Echantillon(string Prompt, string Connecteur, string Reponse, int DureeMs, decimal Cout, DateTime Horodatage);
public sealed record LotAnalyse(IReadOnlyList<Echantillon> Source, DateTime Horodatage, ActivityContext Parent);

public sealed class RouteurMulti
{
    private readonly List<ConnecteurArithmetique> _flotte;
    private readonly Channel<Echantillon> _fileCollecte;
    private readonly ActivitySource _telemetrie;

    public RouteurMulti(List<ConnecteurArithmetique> flotte, ActivitySource telemetrie)
    {
        this._flotte = flotte;
        this._fileCollecte = Channel.CreateUnbounded<Echantillon>();   // chemin chaud : jamais d'attente
        this._telemetrie = telemetrie;
    }

    public ConnecteurArithmetique Primaire => this._flotte[0];
    public IReadOnlyList<ConnecteurArithmetique> Secondaires => this._flotte.Skip(1).ToList();
    public ChannelReader<Echantillon> Journal => this._fileCollecte.Reader;
    public ChannelWriter<Echantillon> JournalEcriture => this._fileCollecte.Writer;
    public int EnAttente => this._fileCollecte.Reader.Count;

    // Le plan de routage : signature (l'operation) -> connecteur. Vide au depart : tout au primaire.
    public ConcurrentDictionary<string, ConnecteurArithmetique> Plan { get; } = new();

    public async Task<(string Reponse, ConnecteurArithmetique Choisi)> CompleterAsync(string prompt)
    {
        var (op, _, _) = MoteurArithmetique.ParserPrompt(prompt);
        var choix = this.Plan.TryGetValue(op.ToString(), out var cible) ? cible : this.Primaire;

        using var span = this._telemetrie.StartActivity("routeur.completion");
        span?.SetTag("prompt.operation", op.ToString());
        span?.SetTag("connecteur", choix.Nom);
        span?.SetTag("offload", !ReferenceEquals(choix, this.Primaire));

        var chrono = Stopwatch.StartNew();
        var reponse = await choix.RepondreAsync(prompt);
        chrono.Stop();
        span?.SetTag("duree.ms", chrono.ElapsedMilliseconds);
        span?.SetTag("cout.unite", choix.Cout.ToString("0.00", CultureInfo.InvariantCulture));

        // LE chemin chaud : un TryWrite sur canal non borne — non bloquant, ne peut pas echouer
        // (cf. MultiTextCompletion.cs:275, AppendConnectorTest).
        this.JournalEcriture.TryWrite(new Echantillon(prompt, choix.Nom, reponse,
            (int)chrono.ElapsedMilliseconds, choix.Cout, DateTime.Now));

        return (reponse, choix);
    }
}

var routeur = new RouteurMulti(Flotte.PocStandard(), source);
Console.WriteLine("Routeur arme. Plan initial : " + routeur.Plan.Count + " entree(s) — tout le trafic passe par "
    + routeur.Primaire.Nom);
Routeur arme. Plan initial : 0 entree(s) — tout le trafic passe par Primaire

Lecture du chemin chaud

La dernière instruction de CompleterAsync est tout le secret du débit 2023 : TryWrite sur un canal CreateUnbounded ne bloque jamais et ne peut pas échouer — ni attente de place, ni backpressure sur le chemin de la réponse. L’échantillon (prompt, connecteur, réponse, durée, coût, horodatage) part en file, et l’appelant a déjà sa réponse. Comparez avec l’alternative naïve — journaliser dans l’appel — qui mettrait la latence du journal sur chaque complétion. La section 7 reprendra la même discipline côté transport — un canal non borné par appel, alimenté par la tâche de réception pendant que le client consomme — et mesurera ce qu’elle vaut en débit.

4. Vetting en ligne : première passe, analyse, deuxième passe

Le protocole du POC, en trois temps :

  1. Première passe — quatre prompts (un par opération), aucun plan : tout passe par le primaire, chaque appel journalise un échantillon. Coût plein.
  2. Analyse — les deux pompes de fond se déclenchent : la collecte draine les échantillons (coalescence par accumulation) et forme un lot ; la file d’analyse (coalescence par remplacement) rejoue chaque prompt sur le secondaire concerné ; le primaire recalcule la bonne réponse et compare — c’est le vetting. Verdict conforme → le plan délègue la signature au secondaire.
  3. Deuxième passe — les mêmes quatre prompts, sur le plan mis à jour.
// Premiere passe : les quatre operateurs discriminateurs, tout sur le primaire.
var travaux = Enum.GetValues<Operation>().Select(op => MoteurArithmetique.GenererPrompt(op, 8, 2)).ToArray();

decimal coutPasse1 = 0m;
var delaiSimule1 = TimeSpan.Zero;
foreach (var t in travaux)
{
    var (reponse, choisi) = await routeur.CompleterAsync(t);
    coutPasse1 += choisi.Cout; delaiSimule1 += choisi.Delai;
    Console.WriteLine(t.PadRight(28) + " => " + reponse.PadRight(4) + " par " + choisi.Nom);
}

Console.WriteLine("---");
Console.WriteLine("Cout passe 1 : " + coutPasse1.ToString("0.00", CultureInfo.InvariantCulture)
    + " u ; delai simule cumule : " + delaiSimule1.TotalMilliseconds.ToString(CultureInfo.InvariantCulture)
    + " ms ; echantillons en file de collecte : " + routeur.EnAttente);
Compute Add(8, 2)            => 10   par Primaire
Compute Subtract(8, 2)       => 6    par Primaire
Compute Multiply(8, 2)       => 16   par Primaire
Compute Divide(8, 2)         => 4    par Primaire
---
Cout passe 1 : 0.08 u ; delai simule cumule : 80 ms ; echantillons en file de collecte : 4

Lecture de la première passe

Les quatre réponses sont correctes (10, 6, 16, 4) et toutes portées par le primaire — le plan est vide, c’est le comportement attendu d’un routeur qui ne fait pas encore confiance. Le coût total est 0.08 u (quatre fois 0.02) pour un délai simulé cumulé de 80 ms. Le point décisif est la dernière ligne : quatre échantillons attendent en file de collecte. Rien n’a encore été analysé — le primaire a répondu, le système a observé, et l’observation attend son traitement différé. La double file entre maintenant en scène.

// Les deux pompes de fond de la double file, fideles aux boucles 2023 :
// - CollectSamplesAsync (MultiTextCompletion.cs:217-236) : coalescence par ACCUMULATION ;
// - AnalyzeDataAsync (MultiCompletionAnalysisSettings.cs:908-947) : coalescence par REMPLACEMENT.
public sealed record LigneVerite(string Connecteur, string Signature, bool Conforme, bool Offload);

public sealed class AtelierVetting : IDisposable
{
    private readonly RouteurMulti _routeur;
    private readonly ActivitySource _telemetrie;
    private readonly Channel<LotAnalyse> _fileAnalyse = Channel.CreateUnbounded<LotAnalyse>();
    private readonly CancellationTokenSource _cts = new();
    private int _analysesEnCours;
    private int _analyseEnAttente;   // un job est lu mais son debounce court encore

    public AtelierVetting(RouteurMulti routeur, ActivitySource telemetrie)
    { this._routeur = routeur; this._telemetrie = telemetrie; }

    public TimeSpan FenetreCollecte { get; set; } = TimeSpan.FromMilliseconds(50);
    public TimeSpan FenetreAnalyse  { get; set; } = TimeSpan.FromMilliseconds(50);
    public int LotsFormes { get; private set; }
    public int AnalysesExecutees { get; private set; }
    public int TestsRejoues { get; private set; }
    public decimal CoutVetting { get; private set; }
    public ConcurrentQueue<string> Journal { get; } = new();
    public List<LigneVerite> TableVerite { get; } = new();

    public void Demarrer()
    {
        _ = Task.Factory.StartNew(() => this.BoucleCollecteAsync(this._cts.Token),
            this._cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
        _ = Task.Factory.StartNew(() => this.BoucleAnalyseAsync(this._cts.Token),
            this._cts.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
    }

    // Etage 1 : ACCUMULATION — chaque echantillon compte, on les garde tous.
    private async Task BoucleCollecteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested && await this._routeur.Journal.WaitToReadAsync(ct))
        {
            using var span = this._telemetrie.StartActivity("collecte.lot");
            var serie = new List<Echantillon>();
            while (this._routeur.Journal.TryRead(out var echantillon))
            {
                // Debounce ANCRE SUR L'OBJET : la fenetre court depuis la NAISSANCE de
                // l'echantillon, pas depuis le reveil de la boucle.
                var attente = echantillon.Horodatage + this.FenetreCollecte - DateTime.Now;
                if (attente > TimeSpan.FromMilliseconds(1)) await Task.Delay(attente, ct);
                serie.Add(echantillon);
            }
            if (serie.Count == 0) continue;

            this.LotsFormes++;
            span?.SetTag("lot.numero", this.LotsFormes);
            span?.SetTag("lot.echantillons", serie.Count);
            this.Journal.Enqueue("LOT " + this.LotsFormes + " forme : " + serie.Count + " echantillons (accumulation)");
            this._fileAnalyse.Writer.TryWrite(new LotAnalyse(serie, DateTime.Now, span?.Context ?? default));
        }
    }

    // Etage 2 : REMPLACEMENT — seule la DERNIERE analyse en attente survit.
    private async Task BoucleAnalyseAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested && await this._fileAnalyse.Reader.WaitToReadAsync(ct))
        {
            LotAnalyse courant = null;
            while (this._fileAnalyse.Reader.TryRead(out var nouveau))
            {
                courant = nouveau;   // ecrase : N declencheurs s'effondrent en une execution
                Volatile.Write(ref this._analyseEnAttente, 1);
                var attente = courant.Horodatage + this.FenetreAnalyse - DateTime.Now;
                if (attente > TimeSpan.FromMilliseconds(1)) await Task.Delay(attente, ct);
            }
            if (courant is null) continue;

            Volatile.Write(ref this._analyseEnAttente, 0);   // le debounce est ferme : on execute
            Interlocked.Increment(ref this._analysesEnCours);
            try { await this.ExecuterAnalyseAsync(courant, ct); }
            finally { Interlocked.Decrement(ref this._analysesEnCours); }
        }
    }

    // Le vetting : rejouer les prompts sur les secondaires ; le PRIMAIRE recalcule et compare.
    private async Task ExecuterAnalyseAsync(LotAnalyse lot, CancellationToken ct)
    {
        using var span = this._telemetrie.StartActivity("analyse.pipeline", ActivityKind.Internal, lot.Parent);
        this.AnalysesExecutees++;
        this.Journal.Enqueue("ANALYSE " + this.AnalysesExecutees + " executee : " + lot.Source.Count
            + " echantillons en entree (remplacement)");

        var parSignature = lot.Source
            .GroupBy(e => MoteurArithmetique.ParserPrompt(e.Prompt).op.ToString())
            .ToDictionary(g => g.Key, g => g.DistinctBy(e => e.Prompt).ToList());

        foreach (var secondaire in this._routeur.Secondaires)
        {
            foreach (var (signature, serie) in parSignature)
            {
                if (!secondaire.SaitFaire(Enum.Parse<Operation>(signature))) continue;

                var conforme = true;
                foreach (var e in serie)
                {
                    this.TestsRejoues++;
                    this.CoutVetting += this._routeur.Primaire.Cout;   // le primaire paie chaque verification
                    var reponseSecondaire = await secondaire.RepondreAsync(e.Prompt);
                    var (op, a, b) = MoteurArithmetique.ParserPrompt(e.Prompt);
                    var attendu = MoteurArithmetique.Calculer(op, a, b).ToString(CultureInfo.InvariantCulture);
                    if (reponseSecondaire != attendu) conforme = false;
                }

                var offload = conforme;   // conforme et moins cher : on delegue la signature
                this.TableVerite.Add(new LigneVerite(secondaire.Nom, signature, conforme, offload));
                if (offload) this._routeur.Plan[signature] = secondaire;
            }
        }
        span?.SetTag("tests.rejoues", this.TestsRejoues);
        span?.SetTag("offloads.accordes", this._routeur.Plan.Count);
    }

    // Attente de quiescence pour le notebook : canaux vides, rien en cours, ET aucun
    // job tenant dans le debounce (sinon la cellule imprimerait "0 analyses" alors
    // qu'une execution est deja promise — le job est lu mais attend sa fenetre).
    public async Task AttendreQuiescenceAsync(TimeSpan? delaiMax = null)
    {
        bool Calme() => this._routeur.EnAttente == 0 && this._fileAnalyse.Reader.Count == 0
            && Volatile.Read(ref this._analysesEnCours) == 0
            && Volatile.Read(ref this._analyseEnAttente) == 0;

        var fin = DateTime.Now + (delaiMax ?? TimeSpan.FromSeconds(20));
        while (DateTime.Now < fin)
        {
            if (Calme())
            {
                await Task.Delay(60);
                if (Calme()) return;
            }
            await Task.Delay(20);
        }
    }

    public void Dispose() => this._cts.Cancel();
}

var atelier = new AtelierVetting(routeur, source);
atelier.Demarrer();
await atelier.AttendreQuiescenceAsync();

Console.WriteLine("Journal de la double file :");
foreach (var ligne in atelier.Journal) Console.WriteLine("  " + ligne);
Console.WriteLine("---");
Console.WriteLine("Table de verite du vetting :");
foreach (var v in atelier.TableVerite)
    Console.WriteLine("  " + v.Connecteur.PadRight(21) + " x " + v.Signature.PadRight(9)
        + " conforme=" + (v.Conforme ? "oui" : "NON") + "  offload=" + (v.Offload ? "oui" : "non"));
Console.WriteLine("---");
Console.WriteLine("Plan de routage apres analyse :");
foreach (var entree in routeur.Plan.OrderBy(kv => kv.Key, StringComparer.Ordinal))
    Console.WriteLine("  " + entree.Key.PadRight(9) + " -> " + entree.Value.Nom);
Console.WriteLine("Cout du vetting lui-meme : " + atelier.CoutVetting.ToString("0.00", CultureInfo.InvariantCulture)
    + " u (" + atelier.TestsRejoues + " tests rejoues payes au prix primaire)");
Journal de la double file :
  LOT 1 forme : 4 echantillons (accumulation)
  ANALYSE 1 executee : 4 echantillons en entree (remplacement)
---
Table de verite du vetting :
  Secondaire-Add        x Add       conforme=oui  offload=oui
  Secondaire-Subtract   x Subtract  conforme=oui  offload=oui
  Secondaire-Multiply   x Multiply  conforme=oui  offload=oui
  Secondaire-Divide     x Divide    conforme=oui  offload=oui
---
Plan de routage apres analyse :
  Add       -> Secondaire-Add
  Divide    -> Secondaire-Divide
  Multiply  -> Secondaire-Multiply
  Subtract  -> Secondaire-Subtract
Cout du vetting lui-meme : 0.08 u (4 tests rejoues payes au prix primaire)

Lecture de la double file : deux canaux, deux coalescences

Le journal montre la mécanique complète : quatre échantillons drainer en un seul lot (l’accumulation les a tous gardés), puis une seule analyse exécutée. La table de vérité dit le reste — les quatre secondaires sont conformes sur leur opération, donc les quatre signatures passent au plan de routage. Le vetting a un coût (0.08 u ici : chaque vérification est un calcul au prix primaire), et c’est honnête : l’offload se rembourse sur le volume — dès la deuxième passe.

Le point que le code 2023 avait et que peu de tutoriels montrent : le même idiome de canal, deux sémantiques de coalescence, choisies par ce que la charge utile signifie.

Étage Coalescence Code 2023 Pourquoi
Collecte (Channel<ConnectorTest>) Accumulation — la série garde chaque échantillon MultiTextCompletion.cs:217-236 Chaque échantillon est une preuve du vetting : on les garde tous
Analyse (Channel<AnalysisJob>) Remplacement — courant = nouveau écrase MultiCompletionAnalysisSettings.cs:908-928 L’analyse est idempotente sur l’état accumulé et chère (elle appelle le primaire) : la rejouer N fois serait du gaspillage

Et le debounce ancré : dans les deux boucles, le délai se calcule horodatage + fenêtre − maintenant — la fenêtre court depuis la naissance de l’échantillon, pas depuis le réveil de la boucle. Une rafale qui arrive en retard n’ajoute donc pas de latence : c’est la forme correcte du debounce, souvent ratée (l’exercice 2 vous la fait rater exprès).

Un mot d’honnêteté sur le lancement : l’option TaskCreationOptions.LongRunning (2023 : MultiTextCompletion.cs:206) fournit un thread dédié jusqu’au premier await — pas au-delà. Sans SynchronizationContext, chaque continuation après un await repart sur le thread pool : la boucle est « longue » par sa durée de vie, pas épinglée à un thread. C’est le fonctionnement attendu d’une pompe asynchrone — et une raison de plus pour que son état passe par des canaux, pas par des variables partagées sensibles au thread.

Et une précision symétrique sur les canaux : non bornés (CreateUnbounded) ne veut pas dire « régulés ». Ils découplent l’écrivain du lecteur — l’écrivain ne bloque jamais — mais n’offrent aucune backpressure : si le lecteur s’arrête, la file grandit sans limite. C’est un choix délibéré du chemin chaud 2023 (la journalisation ne doit jamais freiner une complétion), dont le contrecoût — une file non bornée en cas de panne du lecteur — est assumé par le design de la section 9 (arrêt explicite des pompes).

// Deuxieme passe : les memes quatre prompts, sur le plan mis a jour par le vetting.
decimal coutPasse2 = 0m;
var delaiSimule2 = TimeSpan.Zero;
foreach (var t in travaux)
{
    var (reponse, choisi) = await routeur.CompleterAsync(t);
    coutPasse2 += choisi.Cout; delaiSimule2 += choisi.Delai;
    Console.WriteLine(t.PadRight(28) + " => " + reponse.PadRight(4) + " par " + choisi.Nom);
}

Console.WriteLine("---");
Console.WriteLine("                    passe 1 (primaire)   passe 2 (offload)");
Console.WriteLine("cout total          " + coutPasse1.ToString("0.00", CultureInfo.InvariantCulture).PadLeft(8)
    + " u" + coutPasse2.ToString("0.00", CultureInfo.InvariantCulture).PadLeft(16) + " u");
Console.WriteLine("delai simule cumule " + delaiSimule1.TotalMilliseconds.ToString(CultureInfo.InvariantCulture).PadLeft(8)
    + " ms" + delaiSimule2.TotalMilliseconds.ToString(CultureInfo.InvariantCulture).PadLeft(14) + " ms");
Compute Add(8, 2)            => 10   par Secondaire-Add
Compute Subtract(8, 2)       => 6    par Secondaire-Subtract
Compute Multiply(8, 2)       => 16   par Secondaire-Multiply
Compute Divide(8, 2)         => 4    par Secondaire-Divide
---
                    passe 1 (primaire)   passe 2 (offload)
cout total              0.08 u            0.04 u
delai simule cumule       80 ms             8 ms

Lecture de la deuxième passe : l’offload payé

Les quatre réponses sont identiques aux bonnes valeurs de la première passe — le routage n’a rien coûté en exactitude — mais chacune est maintenant portée par son spécialiste. Le coût tombe de 0.08 u à 0.04 u (divisé par deux, exactement la structure des tarifs), et le délai simulé cumulé de 80 ms à 8 ms (divisé par dix, le rapport 20 ms / 2 ms des latences configurées). Ces deux ratios sont structurels — ils viennent des tarifs et latences déterministes de la flotte, pas d’une mesure mur à mur — alors que les durées réelles affichables en consolide incluent l’ordonnancement du kernel. Ajouté au coût du vetting payé en section précédente, le point d’équilibre est immédiat : ici, une seule passe supplémentaire rembourse l’analyse.

Exercice 1 : le comparateur pondéré durée/coût

Le code 2023 choisissait le connecteur à offloader via un comparateur pondéré (GetWeightedConnectorComparer(durationWeight, costWeight) dans MultiTextCompletionSettings.cs) : deux poids, une décision. Implémentez-le : la fonction doit classer deux connecteurs selon poidsDuree × durée normalisée + poidsCout × coût normalisé.

Cahier des charges : - // Etape 1 — normaliser la durée : Delai.TotalMilliseconds / 20 (le primaire vaut 1) ; - // Etape 2 — normaliser le coût : Cout / 0.02m ; - // Etape 3 — combiner selon les deux poids et renvoyer un Comparison<ConnecteurArithmetique> ; - // Indice — sans normalisation, les 20 ms écrasent les 0.02 u : une unité domine l’autre et un poids ne sert plus à rien.

// Exercice 1 — stub. Le notebook doit s'executer de bout en bout : pas d'erreur volontaire.
Comparison<ConnecteurArithmetique> ComparateurPondere(int poidsDuree, int poidsCout)
{
    // TODO etudiant : renvoyer le comparateur pondere (etapes 1 a 3 ci-dessus)
    return null;
}

Console.WriteLine("Exercice 1 a completer : ComparateurPondere(poidsDuree, poidsCout)");
Exercice 1 a completer : ComparateurPondere(poidsDuree, poidsCout)

5. La double file en régime de rafales

La topologie complète 2023 comptait trois canaux, pas deux — et le premier n’est pas le moindre :

Niveau Canal Écriture (2023) Drainage (2023)
Transport Channel<string> (chunks websocket, un par appel) OobaboogaCompletionBase.cs:99 :171 ProcessWebSocketMessagesAsync — réalisé à la section 7
Collecte Channel<ConnectorTest> MultiTextCompletion.cs:275 (TryWrite) :217-236 — notre étage 1
Analyse Channel<AnalysisJob> MultiCompletionAnalysisSettings.cs:307 (TryWrite) :908-931 — notre étage 2

La section 4 a vu la double file au repos (un lot, une analyse). Le régime intéressant est la rafale : beaucoup d’échantillons proches, puis plus rien. Que devient la fenêtre d’analyse quand trois rafales tombent à 250 ms d’intervalle avec une fenêtre d’analyse de 700 ms ? L’expérience ci-dessous le mesure — la fenêtre de collecte (150 ms) forme un lot par rafale, et la fenêtre d’analyse (700 ms) devrait les effondrer.

// Regime de rafales : 3 rafales de 5 appels espacees de 250 ms, fenetres 150/700 ms.
var routeurRafales = new RouteurMulti(Flotte.PocStandard(), source);
var atelierRafales = new AtelierVetting(routeurRafales, source)
    { FenetreCollecte = TimeSpan.FromMilliseconds(150), FenetreAnalyse = TimeSpan.FromMilliseconds(700) };
atelierRafales.Demarrer();

var chrono = Stopwatch.StartNew();
foreach (var numero in Enumerable.Range(1, 3))
{
    var rafale = Enumerable.Range(0, 5)
        .Select(_ => routeurRafales.CompleterAsync(MoteurArithmetique.GenererPrompt(Operation.Multiply, 8, 2)));
    await Task.WhenAll(rafale);
    Console.WriteLine("rafale " + numero + " a t=" + chrono.ElapsedMilliseconds + " ms : 5 echantillons emis");
    await Task.Delay(250);
}
await atelierRafales.AttendreQuiescenceAsync();
chrono.Stop();

Console.WriteLine("---");
Console.WriteLine("Bilan pour 15 echantillons : " + atelierRafales.LotsFormes + " lot(s) forme(s), "
    + atelierRafales.AnalysesExecutees + " analyse(s) executee(s)");
foreach (var ligne in atelierRafales.Journal) Console.WriteLine("  " + ligne);
rafale 1 a t=36 ms : 5 echantillons emis
rafale 2 a t=327 ms : 5 echantillons emis
rafale 3 a t=616 ms : 5 echantillons emis
---
Bilan pour 15 echantillons : 3 lot(s) forme(s), 1 analyse(s) executee(s)
  LOT 1 forme : 5 echantillons (accumulation)
  LOT 2 forme : 5 echantillons (accumulation)
  LOT 3 forme : 5 echantillons (accumulation)
  ANALYSE 1 executee : 5 echantillons en entree (remplacement)

Lecture des rafales : l’effondrement, chiffres en main

Le bilan se lit en face du mécanisme. Quinze échantillons entrent ; la collecte les a gardés en trois lots de 5 (un par rafale — l’accumulation n’a rien perdu) ; et une seule analyse s’exécute. La ligne du journal dit tout : ANALYSE 1 exécutée : 5 échantillons en entrée — le remplacement n’a passé à l’exécution que le dernier lot : les lots 1 et 2 ont été écrasés en file, leur rejeu n’aura jamais lieu. C’est le comportement voulu : l’analyse est coûteuse (elle rejoue les prompts et fait recalculer le primaire) et re-dérive sa décision de l’état le plus récent — l’exécuter trois fois coûterait trois fois plus pour la même décision. Le prix est assumé et documenté : la preuve de vetting portée par les lots écrasés est jetée.

Le chronomètre confirme le debounce ancré : les rafales s’émettent espacées d’environ 300 ms à l’horloge (250 ms de délai configuré entre rafales, plus le temps des complétions), et le bilan n’est imprimé qu’après l’exécution de l’analyse — c’est-à-dire après naissance du dernier lot + fenêtre d'analyse (700 ms). Si la fenêtre courait « depuis le réveil » de la boucle, la rafale 3 — qui arrive pendant que la boucle attend déjà — repousserait l’exécution d’une fenêtre entière supplémentaire : c’est exactement la version ratée que l’exercice 2 vous fait écrire.

Exercice 2 : le debounce « au réveil » — le faire rater exprès

La version ratée du debounce attend fenêtre à partir du réveil de la boucle, au lieu de l’ancrer sur l’horodatage de l’objet. Implémentez-la, puis raisonnez sur la ligne du temps de la section 5.

Cahier des charges : - // Etape 1 — écrire AttenteNaive : au réveil, attendre la pleine fenêtre avant de drainer ; - // Etape 2 — prédire sur le papier : combien d’analyses pour 3 rafales espacées de 250 ms avec fenêtre 700 ms ? Et à quelle heure s’exécute la dernière ? - // Indice — chaque réveil repart la fenêtre : une rafale tardive ajoute la pleine fenêtre à la latence, là où la version ancrée n’ajoute que le reste.

// Exercice 2 — stub.
Task AttenteNaive(TimeSpan fenetre)
{
    // TODO etudiant : implementer le debounce "au reveil", puis comparer a la version ancree
    return Task.CompletedTask;
}

Console.WriteLine("Exercice 2 a completer : AttenteNaive + prediction du nombre d'analyses");
Exercice 2 a completer : AttenteNaive + prediction du nombre d'analyses

6. La trace : lire la machine dans les spans

Toutes les sections précédentes ont émis des spans routeur.completion, collecte.lot et analyse.pipeline. L’exporteur mémoire les a capturés à leur fermeture — exportons-les maintenant en tableau : c’est exactement ce qu’un backend de traces (Aspire Dashboard, Jaeger…) montrerait d’un routeur en production, sauf que le backend est la cellule suivante.

// Lecture du backend de fortune : arbre parent/enfant, durees, tags.
var spans = exporteur.CapturerEtVider();
Console.WriteLine(spans.Count + " spans capturees (ordre de capture = ordre de FERMETURE, pas d'ouverture)\n");

var parId = spans.ToDictionary(s => s.SpanId.ToString());
int Profondeur(Activity s)
{
    int d = 0;
    var parent = s.ParentSpanId;              // non nullable : default quand le span est racine
    while (parent != default && parId.TryGetValue(parent.ToString(), out var p)) { d++; parent = p.ParentSpanId; }
    return d;
}

foreach (var s in spans.OrderBy(s => s.StartTimeUtc))
{
    var nom = new string(' ', 2 * Profondeur(s)) + s.DisplayName;
    var tags = string.Join(" ", s.TagObjects.Where(t => t.Value is not null).Select(t => t.Key + "=" + t.Value));
    Console.WriteLine(nom.PadRight(30) + s.Duration.TotalMilliseconds.ToString("0", CultureInfo.InvariantCulture).PadLeft(6)
        + " ms   " + tags);
}

var repartition = spans.GroupBy(s => s.DisplayName).ToDictionary(g => g.Key, g => g.Count());
Console.WriteLine("\nRepartition : " + string.Join(", ", repartition.OrderBy(kv => kv.Key, StringComparer.Ordinal)
    .Select(kv => kv.Key + " x" + kv.Value)));
31 spans capturees (ordre de capture = ordre de FERMETURE, pas d'ouverture)

routeur.completion                27 ms   prompt.operation=Add connecteur=Primaire offload=False duree.ms=25 cout.unite=0.02
routeur.completion                25 ms   prompt.operation=Subtract connecteur=Primaire offload=False duree.ms=25 cout.unite=0.02
routeur.completion                30 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=29 cout.unite=0.02
routeur.completion                30 ms   prompt.operation=Divide connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
collecte.lot                       1 ms   lot.numero=1 lot.echantillons=4
  analyse.pipeline                61 ms   tests.rejoues=4 offloads.accordes=4
routeur.completion                14 ms   prompt.operation=Add connecteur=Secondaire-Add offload=True duree.ms=14 cout.unite=0.01
collecte.lot                     108 ms   lot.numero=2 lot.echantillons=4
routeur.completion                15 ms   prompt.operation=Subtract connecteur=Secondaire-Subtract offload=True duree.ms=14 cout.unite=0.01
routeur.completion                15 ms   prompt.operation=Multiply connecteur=Secondaire-Multiply offload=True duree.ms=15 cout.unite=0.01
routeur.completion                15 ms   prompt.operation=Divide connecteur=Secondaire-Divide offload=True duree.ms=14 cout.unite=0.01
  analyse.pipeline                47 ms   tests.rejoues=8 offloads.accordes=4
routeur.completion                35 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=35 cout.unite=0.02
routeur.completion                35 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=34 cout.unite=0.02
routeur.completion                35 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=34 cout.unite=0.02
routeur.completion                35 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=34 cout.unite=0.02
routeur.completion                35 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=34 cout.unite=0.02
collecte.lot                     162 ms   lot.numero=1 lot.echantillons=5
routeur.completion                31 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                31 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                31 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                31 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                31 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
collecte.lot                     161 ms   lot.numero=2 lot.echantillons=5
routeur.completion                31 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                30 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                30 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                30 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
routeur.completion                30 ms   prompt.operation=Multiply connecteur=Primaire offload=False duree.ms=30 cout.unite=0.02
collecte.lot                     164 ms   lot.numero=3 lot.echantillons=5
  analyse.pipeline                16 ms   tests.rejoues=1 offloads.accordes=1

Repartition : analyse.pipeline x3, collecte.lot x5, routeur.completion x23

Lecture de la trace : ce que le Stopwatch de 2023 ne montrait pas

Quatre choses lisibles uniquement parce que l’instrumentation vit dans le pipeline.

L’imbrication, d’abord. Chaque analyse.pipeline est indentée sous son collecte.lot — trois paires visibles, dont celle des rafales (lot.numero=3 → analyse à tests.rejoues=1) : le contexte parent a traversé la frontière de thread via le ActivityContext porté par le lot. Aucun Stopwatch ne montre ça.

La fenêtre de debounce rendue visible, ensuite. Les trois lots des rafales affichent ~160 ms — leur fenêtre de collecte de 150 ms, plus le drainage des cinq échantillons ; le lot 1 de la section 4 n’affiche que ~1 ms, parce que ses échantillons étaient nés depuis plus de 50 ms quand la boucle a drainé : la fenêtre ancrée n’attend que le restant, jamais la fenêtre entière. La trace mesure le mécanisme, pas seulement le code. Les durées d’analyse.pipeline (~60 ms pour la section 4, ~15 ms pour les rafales) sont en revanche hors debounce : le span s’ouvre quand l’exécution commence, après la fermeture de la fenêtre.

La décision économique comme donnée, troisièmement. Le tag offload des routeur.completion passe de False (passe 1, quatre connecteur=Primaire) à True (passe 2, quatre Secondaire-*) — et la répartition finale résume la session : 23 complétions, 5 lots, 3 analyses. Notez aussi que la passe 2 a déclenché son propre cycle lot + analyse (tests.rejoues=8 : 4 puis 4) : le vetting est continu, pas un one-shot de démarrage.

L’écart horloge vs configuration, enfin — une leçon C.5 en acte. Les complétions secondaires mesurent 13-16 ms alors que leur Task.Delay configuré est de 2 ms : l’horloge Windows réveille un Task.Delay à la granularité de son minuteur (~15 ms). C’est la raison pour laquelle les ratios de la section 4 (÷2 en coût, ÷10 en latence) sont ancrés sur la configuration déterministe de la flotte, jamais sur les durées mur à mur.

Deux conventions de lecture, héritées du notebook 03 : l’ordre de capture est l’ordre des fermetures (un enfant s’exporte avant son parent — d’où les complétions de la passe 2 intercalées entre le lot et son analyse), et les spans racines pointent vers l’activité ambiante du kernel — d’où le sampler forcé en section 3.

7. Le pool de sockets et le serveur de test : mesurer le routeur, pas le harnais

Le connecteur 2023 était socket, pas HTTP : un ClientWebSocket par appel, mutualisés dans un pool. Les trois artefacts du sous-module :

  • Le pool (OobaboogaCompletionSettings.cs:36) : un ConcurrentBag<ClientWebSocket>. Au départ d’un appel, TryTake — si le sac est vide, on crée et connecte ; au retour, la socket retourne au sac si et seulement si elle est encore Open (OobaboogaCompletionBase.cs:98-148). Un sémaphore plafonnait la concurrence (stress test 2023 : jamais plus de 20 clients pour 10 000 appels concurrents).
  • Le canal par appel (OobaboogaCompletionBase.cs:99) : chaque complétion en streaming ouvre un Channel<string> non borné ; la tâche de réception websocket y pousse les chunks, l’appelant les consomme en await foreach — le niveau transport de la topologie de la section 5.
  • Le serveur de test (Connectors.UnitTests/WebSocketTestServer.cs) : un HttpListener — la primitive de self-host pré-ASP.NET Core, adossée à HTTP.SYS — qui fait l’upgrade websocket nativement (AcceptWebSocketAsync). Sa propriété centrale : la réponse est pré-encodée en une passe avant la boucle d’émission, et le tampon de réception (WebSocket.CreateServerBuffer(4096)) est alloué une fois par connexion. Zéro sérialisation, zéro allocation par message dans le chemin chaud — c’est ce qui rend le banc honnête : à plein débit, on mesure le routeur, pas le harnais.

Deux délais injectables (RequestProcessingDelay = time-to-first-token, SegmentMessageDelay = cadence inter-tokens) en faisaient aussi un modèle de latence de LLM sans LLM. Ci-dessous, la version miniature fidèle : réponses pré-encodées, port libre aléatoire, ordre d’arrêt qui respecte la cicatrice documentée (Stop() avant annulation — WebSocketTestServer.cs:225-230 : GetContextAsync n’a pas de surcharge annulable, l’ordre inverse deadlock le préfixe).

// Le serveur de test 2023 en miniature (WebSocketTestServer.cs) :
// HttpListener + AcceptWebSocketAsync, reponse PRE-ENCODEE en une passe, buffer reutilise.
public sealed class ServeurFluxTest : IAsyncDisposable
{
    private readonly HttpListener _listener = new();
    private readonly CancellationTokenSource _cts = new();
    private readonly ReadOnlyMemory<byte>[] _segments;
    private Task _boucle = Task.CompletedTask;

    public ServeurFluxTest(int morceauxParReponse)
    {
        // Port libre : sniff d'un port ephemere (course theorique acceptable en local).
        var sonde = new TcpListener(IPAddress.Loopback, 0);
        sonde.Start();
        this.Port = ((IPEndPoint)sonde.LocalEndpoint).Port;
        sonde.Stop();

        // Toute la reponse est encodee UNE fois, au demarrage : zero serialisation dans le chemin chaud.
        static string Cadre(int i) => "{\"event\":\"text_stream\",\"num\":" + i + ",\"text\":\"tok" + i + "\"}";
        this._segments = Enumerable.Range(0, morceauxParReponse)
            .Select(i => (ReadOnlyMemory<byte>)Encoding.UTF8.GetBytes(Cadre(i)))
            .Append(Encoding.UTF8.GetBytes("{\"event\":\"stream_end\"}"))
            .ToArray();
    }

    public int Port { get; }

    public void Demarrer()
    {
        this._listener.Prefixes.Add("http://localhost:" + this.Port + "/");
        this._listener.Start();
        this._boucle = Task.Run(() => this.BoucleAsync());
    }

    private async Task BoucleAsync()
    {
        while (!this._cts.IsCancellationRequested)
        {
            HttpListenerContext contexte;
            try { contexte = await this._listener.GetContextAsync(); }
            catch (Exception) when (this._cts.IsCancellationRequested) { break; }
            if (contexte.Request.IsWebSocketRequest)
                _ = this.ServirAsync(contexte);   // tache par client, lancee sans etre attendue (2023 :96)
        }
    }

    private async Task ServirAsync(HttpListenerContext contexte)
    {
        try
        {
            using var ws = (await contexte.AcceptWebSocketAsync(subProtocol: null)).WebSocket;
            var tampon = WebSocket.CreateServerBuffer(4096);   // alloue UNE fois par connexion (2023 :123)
            while (ws.State == WebSocketState.Open && !this._cts.IsCancellationRequested)
            {
                var recu = await ws.ReceiveAsync(tampon, this._cts.Token);
                if (recu.MessageType == WebSocketMessageType.Close)
                {
                    await ws.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, null, CancellationToken.None);
                    break;
                }
                if (recu.EndOfMessage)
                    foreach (var segment in this._segments)
                        await ws.SendAsync(segment, WebSocketMessageType.Text, true, this._cts.Token);
            }
        }
        catch (Exception) when (this._cts.IsCancellationRequested) { /* arret du serveur */ }
    }

    public async ValueTask DisposeAsync()
    {
        // L'ordre qui defait le deadlock documente (#7270, WebSocketTestServer.cs:225-230) :
        // Stop le listener AVANT d'annuler — sinon GetContextAsync garde le prefix bloque.
        this._listener.Stop();
        this._cts.Cancel();
        try { await this._boucle; } catch (Exception) { /* le thrown d'arret est attendu */ }
    }
}

var serveur = new ServeurFluxTest(morceauxParReponse: 250);
serveur.Demarrer();
Console.WriteLine("Serveur de test demarre sur ws://localhost:" + serveur.Port
    + "/ — 250 morceaux pre-encodes par reponse, zero serialisation dans le chemin chaud");
Serveur de test demarre sur ws://localhost:61178/ — 250 morceaux pre-encodes par reponse, zero serialisation dans le chemin chaud

Côté client, la symétrie 2023 est totale : le pool (TryTake au départ, retour au sac si encore Open), et le canal par appel — chaque requête ouvre son propre Channel<string> non borné, la tâche de réception y pousse les chunks, le consommateur les énumère en await foreach. Le banc lance plus de requêtes que le plafond de parallélisme : c’est la condition pour que le pool se voit — les sockets créées doivent rester très inférieures aux requêtes servies.

// Le pool 2023 (OobaboogaCompletionSettings.cs:36) : ConcurrentBag<ClientWebSocket>.
public sealed class PoolSockets
{
    private readonly ConcurrentBag<ClientWebSocket> _sac = new();
    private readonly Uri _uri;
    public int Crees { get; private set; }
    public int Emprunts { get; private set; }

    public PoolSockets(Uri uri) { this._uri = uri; }

    public async Task<ClientWebSocket> LouerAsync(CancellationToken ct)
    {
        this.Emprunts++;
        if (!this._sac.TryTake(out var ws)) ws = new ClientWebSocket();
        if (ws.State != WebSocketState.Open) { ws.Dispose(); ws = new ClientWebSocket(); }
        if (ws.State == WebSocketState.None)
        {
            await ws.ConnectAsync(this._uri, ct);
            this.Crees++;   // seule une socket NEUVE est comptee ; une socket reusee ne l'est pas
        }
        return ws;
    }

    public void Rendre(ClientWebSocket ws)
    {
        if (ws.State == WebSocketState.Open) this._sac.Add(ws); else ws.Dispose();
    }

    public void Vider()
    {
        while (this._sac.TryTake(out var ws)) ws.Dispose();
    }
}

// Un flux complet : requete -> canal par appel -> enumeration des chunks -> rendu au pool.
// Fonction LOCALE (pas de methodes au niveau top-level du kernel).
async Task<int> UnFluxAsync(PoolSockets pool, CancellationToken ct)
{
    var ws = await pool.LouerAsync(ct);
    try
    {
        var flux = Channel.CreateUnbounded<string>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = true });
        var reception = Task.Run(async () =>
        {
            var tampon = new byte[4096];
            while (true)
            {
                var recu = await ws.ReceiveAsync(tampon, ct);
                if (recu.MessageType == WebSocketMessageType.Close) { flux.Writer.Complete(); return; }
                var texte = Encoding.UTF8.GetString(tampon, 0, recu.Count);
                if (texte.Contains("\"event\":\"stream_end\"")) { flux.Writer.Complete(); return; }
                await flux.Writer.WriteAsync(texte, ct);
            }
        }, ct);

        var requete = Encoding.UTF8.GetBytes("{\"prompt\":\"Compute Add(8, 2)\"}");
        await ws.SendAsync(requete, WebSocketMessageType.Text, true, ct);

        int morceaux = 0;
        await foreach (var _ in flux.Reader.ReadAllAsync(ct)) morceaux++;
        await reception;
        return morceaux;
    }
    finally { pool.Rendre(ws); }
}

// Le banc : 24 requetes, parallelisme plafonne a 8 — le pool doit creer ~8 sockets, pas 24.
const int Requetes = 24, Parallelisme = 8;
var limite = new SemaphoreSlim(Parallelisme, Parallelisme);   // pas de "using var" en top-level kernel
var pool = new PoolSockets(new Uri("ws://localhost:" + serveur.Port + "/"));
long morceauxTotaux = 0;

using (var span = source.StartActivity("banc.socket"))
{
    span?.SetTag("banc.requetes", Requetes);
    span?.SetTag("banc.parallelisme", Parallelisme);

    var chrono = Stopwatch.StartNew();
    var taches = Enumerable.Range(0, Requetes).Select(async _ =>
    {
        await limite.WaitAsync();
        try
        {
            var recus = await UnFluxAsync(pool, CancellationToken.None);
            Interlocked.Add(ref morceauxTotaux, recus);
        }
        finally { limite.Release(); }
    });
    await Task.WhenAll(taches);
    chrono.Stop();

    span?.SetTag("banc.duree.ms", chrono.ElapsedMilliseconds);
    span?.SetTag("banc.morceaux", morceauxTotaux);
    span?.SetTag("banc.sockets_creees", pool.Crees);

    Console.WriteLine(Requetes + " requetes x 250 morceaux, parallelisme " + Parallelisme);
    Console.WriteLine("Morceaux recus au total : " + morceauxTotaux);
    Console.WriteLine("Sockets CREES : " + pool.Crees + " (emprunts : " + pool.Emprunts + " — le pool a servi " + (pool.Emprunts - pool.Crees) + " emprunts sans creer)");
    Console.WriteLine("Duree mesuree : " + chrono.Elapsed.TotalMilliseconds.ToString("0", CultureInfo.InvariantCulture) + " ms");
    Console.WriteLine("Debit mesure : " + (morceauxTotaux / chrono.Elapsed.TotalSeconds).ToString("0", CultureInfo.InvariantCulture)
        + " morceaux/s — pipeline canal+socket sur loopback localhost");
}
limite.Dispose();
24 requetes x 250 morceaux, parallelisme 8
Morceaux recus au total : 6000
Sockets CREES : 8 (emprunts : 24 — le pool a servi 16 emprunts sans creer)
Duree mesuree : 67 ms
Debit mesure : 88980 morceaux/s — pipeline canal+socket sur loopback localhost

Lecture du banc : le pool se voit, le débit est mesuré — et borné

Trois lectures, une mise en garde. Le pool d’abord : la ligne Sockets CREES affiche 8 pour 24 requêtes — la première vague (le parallélisme plafonné) chauffe le sac, chaque vague suivante réutilise : 24 emprunts, dont 16 servis sans créer de socket. C’est la seule ligne qui prouve que le pool travaille — et ces trois comptes sont structurels (24 requêtes, plafond 8), donc stables d’une exécution à l’autre. Le débit ensuite : la cellule mesure un vrai pipeline canal + socket — requêtes réelles, frames websocket réelles, énumération asynchrone réelle — et le chiffre affiché est celui de cette exécution-ci (la durée varie avec la machine ; seul son ordre de grandeur est discutable, pas sa nature). La mise en garde enfin :

Frontière d’honnêteté. Ce banc boucle sur localhost avec un serveur à réponse pré-encodée : le débit mesuré couvre le pipeline routeur + canal + socket — le transport et la découpe en flux — et rien d’autre. Aucun token n’est généré, aucun modèle n’est servi. Le souvenir 2023 de « 20 000 chunks/s » reposait sur la même architecture (serveur de test localhost, 50 000 chunks par test) et mesurait la même chose ; il n’est pas re-mesuré ici et ne doit jamais être cité comme un débit de génération.

C’est exactement le piège que la réactivation de l’Epic demande de verrouiller : un banc de transport honnête est un banc de transport étiqueté.

8. La table de vérité : refuser correctement

Les tests d’intégration 2023 (Connectors.UnitTests/../IntegrationTests/MultiConnectorTests.cs:96-110) encodaient le résultat attendu du vetting comme assertion — premier argument de chaque InlineData : succeedsOffloading. La frontière de capacité de chaque modèle local, modèle × difficulté :

Modèle local simple medium hard
microsoft_phi-1_5 ✓ ✓ ✗
TheBloke_orca_mini_3B-GGML ✓ ✓ ✗
TheBloke_Mistral-7B-OpenOrca-GGUF ✓ ✓ ✓
TheBloke_LLaMA2-13B-Tiefighter-GGUF ✓ ✓ ✓

Le ✗ n’est pas un échec du test : il vérifie que le routeur refuse correctement de déléguer au modèle cheap ce qui le dépasse. C’est la partie la plus précieuse du vetting — et exactement la question qu’un arbitrage multi-modèles répond aujourd’hui à la main. Reproduisons la mécanique : un secondaire Add silencieusement corrompu (il répond juste à côté), et voyons si le plan le vète.

// Le secondaire Add repond result + 1 : faux silencieux, indetectable sans vetting.
var routeurCorrompu = new RouteurMulti(new List<ConnecteurArithmetique>
{
    new("Primaire", Enum.GetValues<Operation>().ToHashSet(), TimeSpan.FromMilliseconds(20), 0.02m),
    new("Secondaire-Add", new HashSet<Operation> { Operation.Add }, TimeSpan.FromMilliseconds(2), 0.01m,
        substitut: (op, a, b) => MoteurArithmetique.Calculer(op, a, b) + 1),
    new("Secondaire-Subtract", new HashSet<Operation> { Operation.Subtract }, TimeSpan.FromMilliseconds(2), 0.01m),
    new("Secondaire-Multiply", new HashSet<Operation> { Operation.Multiply }, TimeSpan.FromMilliseconds(2), 0.01m),
    new("Secondaire-Divide",   new HashSet<Operation> { Operation.Divide },   TimeSpan.FromMilliseconds(2), 0.01m),
}, source);

var atelierCorrompu = new AtelierVetting(routeurCorrompu, source);
atelierCorrompu.Demarrer();

// La faute, montree AVANT le vetting : appel direct au secondaire corrompu.
var corrompu = routeurCorrompu.Secondaires.First(c => c.Nom == "Secondaire-Add");
Console.WriteLine("Appel direct : Compute Add(8, 2) => " + await corrompu.RepondreAsync(MoteurArithmetique.GenererPrompt(Operation.Add, 8, 2))
    + "  par " + corrompu.Nom + "  (attendu : 10 — le faux est silencieux, seul le vetting le verra)");
Console.WriteLine("---");

// Trafic temoin : les quatre operations, simultanees.
var temoin = Enum.GetValues<Operation>()
    .Select(op => routeurCorrompu.CompleterAsync(MoteurArithmetique.GenererPrompt(op, 8, 2)));
await Task.WhenAll(temoin);
await atelierCorrompu.AttendreQuiescenceAsync();

Console.WriteLine("Table de verite (secondaire x sa signature) — le refus est un SUCCES du routeur :");
foreach (var v in atelierCorrompu.TableVerite.OrderBy(v => v.Signature, StringComparer.Ordinal))
    Console.WriteLine("  " + v.Connecteur.PadRight(21) + " x " + v.Signature.PadRight(9)
        + " conforme=" + (v.Conforme ? "oui" : "NON") + "  offload=" + (v.Offload ? "oui" : "non"));
Console.WriteLine("---");
Console.WriteLine("Plan final : " + string.Join(" | ", routeurCorrompu.Plan
    .OrderBy(kv => kv.Key, StringComparer.Ordinal).Select(kv => kv.Key + " -> " + kv.Value.Nom)));
Console.WriteLine("Add reste au primaire : " + (!routeurCorrompu.Plan.ContainsKey("Add")));
Appel direct : Compute Add(8, 2) => 11  par Secondaire-Add  (attendu : 10 — le faux est silencieux, seul le vetting le verra)
---
Table de verite (secondaire x sa signature) — le refus est un SUCCES du routeur :
  Secondaire-Add        x Add       conforme=NON  offload=non
  Secondaire-Divide     x Divide    conforme=oui  offload=oui
  Secondaire-Multiply   x Multiply  conforme=oui  offload=oui
  Secondaire-Subtract   x Subtract  conforme=oui  offload=oui
---
Plan final : Divide -> Secondaire-Divide | Multiply -> Secondaire-Multiply | Subtract -> Secondaire-Subtract
Add reste au primaire : True

Lecture du refus : la frontière de capacité comme assertion

Le secondaire corrompu a répondu 11 là où le primaire calcule 10 — le vetting l’a vu, la table porte conforme=NON, offload=non, et le plan final n’accorde pas la signature Add : elle reste au primaire. Les trois autres secondaires, eux, sont offloadés. Relisez la table 2023 ci-dessus : structurellement, c’est la même assertion — la ligne phi-1_5 × hard → ✗ et notre Secondaire-Add × Add → NON vérifient toutes deux que le routeur connaît la frontière de ses seconds couteaux. Un routeur qui offloaderait quand même serait en échec ; le routeur qui refuse gagne. C’est la réponse mécanique — mesurée, rejouable — à une question qu’un arbitrage humain de routage multi-modèles tranche aujourd’hui au cas par cas.

Exercice 3 : la quarantaine à trois fautes

Le vetting 2023 avait des niveaux (VettingLevel, rejouage de TestEvent) : un connecteur défaillant n’était pas simplement jeté — il pouvait être re-testé plus tard. Écrivez la règle la plus simple de cette famille : après un historique de verdicts (chronologique), un connecteur part en quarantaine dès trois faux consécutifs.

Cahier des charges : - // Etape 1 — ne regarder que la fin de l’historique : trois verdicts suffisent à décider ; - // Etape 2 — trois faux consécutifs, pas trois faux au total — [vrai, faux, faux, faux] quarantaine, [faux, faux, vrai, faux] pas encore ; - // Indice — un historique plus court que trois verdicts ne peut pas mettre en quarantaine.

// Exercice 3 — stub.
string VerdictQuarantaine(IReadOnlyList<bool> historique)
{
    // TODO etudiant : "quarantaine" si les 3 derniers verdicts sont tous faux, sinon "conforme"
    return "non implemente";
}

Console.WriteLine("Exercice 3 a completer : VerdictQuarantaine(historique)");
Exercice 3 a completer : VerdictQuarantaine(historique)

9. Arrêt propre

Le kernel vit tant que le notebook est ouvert : les pompes de fond, le serveur de test et le provider OTel doivent être arrêtés explicitement — dans le bon ordre pour le serveur (sa cicatrice d’arrêt), et après avoir rendu visibles les dernières spans.

// Arret propre : serveur (Stop AVANT annulation), pompes de fond, provider OTel.
await serveur.DisposeAsync();
atelier.Dispose();
atelierRafales.Dispose();
atelierCorrompu.Dispose();
pool.Vider();
provider.Dispose();

Console.WriteLine("Serveur arrete, pompes annulees, pool vide, provider dispose.");
Console.WriteLine("Bilan de session : " + exporteur.Capturer().Count
    + " spans emises par la source Aspire07.SemanticFleet");
Serveur arrete, pompes annulees, pool vide, provider dispose.
Bilan de session : 7 spans emises par la source Aspire07.SemanticFleet

Conclusion

Le MultiConnector de 2023 anticipait, avec trois ans d’avance et un POC arithmétique, ce que tout déploiement multi-modèles arbitre aujourd’hui : qui peut répondre à quoi, à quel prix, sans se tromper — et comment le savoir sans humain dans la boucle. Ce notebook l’a ressuscité en le ré-instrumentant :

Pattern 2023 (sous-module 9df3603) Où le voir ici Ce qu’OTel ajoute
Vetting en ligne, offload par signature Sections 2-4 : deux passes, coût divisé par deux La décision (offload=true/false) devient un tag de span
Double file, coalescence accumulation / remplacement Sections 4-5 : 15 échantillons → 3 lots → 1 analyse collecte.lot et analyse.pipeline imbriqués à travers la frontière de thread
Debounce ancré sur l’horodatage Section 5 : fenêtre absolue depuis la naissance de l’objet La fenêtre est lisible dans la durée du span
Pool ClientWebSocket + serveur HttpListener pré-encodé Section 7 : ~8 sockets pour 24 requêtes, débit mesuré Le banc porte lui-même son span banc.socket

Ce que ce notebook ne mesure pas — sa frontière, une dernière fois : aucune génération de tokens, aucun endpoint LLM, un loopback localhost. Le débit cité est celui du pipeline de transport ; le vetting est arithmétique, pas sémantique. L’honnêteté d’un banc se joue dans cette étiquette.

Pour aller plus loin : - Le sous-module GenAI/SemanticKernel/semantic-fleet : les mocks (ArithmeticMocks/), les boucles (MultiTextCompletion.cs, MultiCompletionAnalysisSettings.cs), le serveur (Connectors.UnitTests/WebSocketTestServer.cs) — tous cités dans ce notebook ; - Le matcher radix (tools/radix/ du sous-module, Axe 4 de l’Epic) : l’autre morceau du routeur, pour reconnaître les signatures récurrentes de prompts ; - Les notebooks 03 — Observabilité (les trois piliers OTel) et 04 — Streaming Agent (le pattern Channels + BackgroundService dont la double file est la version productionnée) ; - L’Epic #1210 pour l’histoire complète — trois PRs amont, un manifeste, et une vision fermée par policy que ce dépôt fait vivre autrement.

Retour au sommet