Suivre les exécutions avec OpenTelemetry

Observer une exécution entière

Créez un hub d’observation pour réunir les événements du workflow, des agents et des ressources. Ajoutez les fonctions qui recevront les événements, puis passez le hub dans l’option observation. Chaque événement contient son contexte d’exécution.

events.tsreview-task.tsobserve-workflow.ts

events.ts

import { reportValue } from "./reporter.ts";
import { createObservationHub } from "@elie-laloum/outpost";

export const observation = createObservationHub({
  sinks: [
    {
      observe({ seq, source, scope, event }) {
        reportValue(seq, source, scope.taskKey, event.kind);
        // Example output: 1 agent review phase
      },
    },
  ],
});

review-task.ts

import { defineIsolatedTask } from "@elie-laloum/outpost";
import { repository, sandboxProvider, coder } from "./outpost.config.ts";

export const review = defineIsolatedTask({
  key: "review",
  request: () => ({
    repository,
    sandboxProvider,
    agent: coder,
    brief: { text: "Review the public API without modifying files." },
  }),
});

observe-workflow.ts

import { defineWorkflow } from "@elie-laloum/outpost";
import { review } from "./review-task.ts";
import { observation } from "./events.ts";

export const result = await defineWorkflow("review", [review]).start({
  observation,
});
await observation.close();
result.unwrap();

Le récepteur affiche les transitions du workflow, les opérations de sandbox et de Git et les événements de l’agent, dans l’ordre de seq. dispatch() accepte la même option observation pour une tâche isolée.

L’enveloppe de l’événement inclut le contexte de l’exécution.

Référence API : Observation.

Transmettre le contexte à vos propres tâches

defineAgentTask et defineIsolatedTask rattachent leur dispatch au contexte de la tâche. Dans une tâche écrite avec defineTask, passez context.observation à chaque dispatch.

import { defineTask, dispatch } from "@elie-laloum/outpost";
import { coder, repository, sandboxProvider } from "./outpost.config.ts";

const audit = defineTask({
  key: "audit",
  perform: async (context) => {
    const result = await dispatch({
      repository,
      sandboxProvider,
      agent: coder,
      brief: { text: "List outdated dependencies without changing files." },
      signal: context.signal,
      ...(context.observation ? { observation: context.observation } : {}),
    });
    return result.text;
  },
});

Sans cela, le dispatch alimente toujours son propre observe, mais ses événements n’atteignent jamais le hub. observation.child(scope, sinks) dérive un hub qui ajoute des champs de contexte ; les récepteurs passés à un enfant ne reçoivent que les événements émis sous lui.

La spéculation accepte la même option observation, et les fonctions utilitaires de récupération et de rétention acceptent un hub en dernier argument.

Ce que reçoit le hub

Le hub réunit l’activité des agents, les transitions du workflow et les opérations sur les ressources.

Référence API : ObservationEvent.

Un événement operation associe started à finished ou failed par son id. Seul l’événement terminal porte durationMs.

Événements du harness intégré

Le harness intégré rend compte de sa boucle avec ces types. Ils atteignent aussi observe.

Référence API : AgentEvent.

tool-result ne conserve qu’un preview de 2 000 caractères ; abonnez-vous à tool-output pour le flux complet.

Attention

model-request, model-response et tool-output peuvent contenir du contenu du dépôt et des secrets lus par l’agent. Tenez-les à l’écart des journaux publics.

Livraison et erreurs

Un récepteur qui ne renvoie rien s’exécute pendant l’émission : gardez-le rapide. Un récepteur qui renvoie une promesse dispose de sa propre file ordonnée. Le flush() facultatif d’un récepteur s’exécute chaque fois que le hub se vide.

dispatch() et start() vident leurs livraisons avant de rendre la main. flush() vide le hub à tout moment ; close() le vide et cesse d’accepter des événements.

slow-observer.tsdelivery.ts

slow-observer.ts

import { reportValue } from "./reporter.ts";
import { createObservationHub } from "@elie-laloum/outpost";

export const observation = createObservationHub({
  deliveryTimeoutMs: 2_000,
  sinks: [
    {
      async observe({ seq, event }) {
        await new Promise((resolve) => setTimeout(resolve, 5));
        reportValue(seq, event.kind);
        // Example output: 1 workflow
      },
    },
  ],
});

delivery.ts

import { reportValue } from "./reporter.ts";
import { defineTask, defineWorkflow } from "@elie-laloum/outpost";
import { observation } from "./slow-observer.ts";

export const greet = defineTask({ key: "greet", perform: () => "hello" });
export const result = await defineWorkflow("greet", [greet]).start({
  observation,
});
await observation.close();
reportValue(result.status, observation.dropped, observation.errors.length);
// Example output: done 0 0

Situation	Conséquence
La file d’un récepteur contient déjà capacity événements (1 024 par défaut).	Les nouveaux événements pour ce récepteur sont perdus et dropped augmente.
Une livraison dépasse deliveryTimeoutMs (5 000 par défaut).	Le récepteur est désactivé. Sa promesse en cours continue de s’exécuter.
Un récepteur lève une exception ou rejette.	L’erreur rejoint errors et les observerErrors de l’exécution, qui n’est pas affectée.

Le récepteur affiche les événements workflow numérotés, puis le script affiche done 0 0. Vérifiez dropped et errors avant de considérer une trace comme complète.

Traiter une sortie trop volumineuse

Une ligne de protocole de plus de 16 Mio arrête un agent CLI. Le hub et observe reçoivent un événement raw avec ses 2 000 premiers caractères, bytes et truncated: true, puis stopped avec la raison oversized-event. Le dispatch échoue avec le code process (Erreurs).

Exporter vers OpenTelemetry

Installez @opentelemetry/api et un SDK OpenTelemetry, puis enregistrez le SDK et ses exportateurs avant de créer l’observateur. Son sink transforme les événements du hub en spans liés et en métriques.

telemetry.tstelemetry-review.tsobserve-telemetry.ts

telemetry.ts

import { createOpenTelemetryObserver } from "@elie-laloum/outpost/opentelemetry";
import { trace, metrics } from "@opentelemetry/api";
import { createObservationHub } from "@elie-laloum/outpost";

export const telemetry = createOpenTelemetryObserver({
  tracer: trace.getTracer("outpost"),
  meter: metrics.getMeter("outpost"),
});
export const observation = createObservationHub({ sinks: [telemetry.sink] });

telemetry-review.ts

import { defineIsolatedTask } from "@elie-laloum/outpost";
import { repository, sandboxProvider, coder } from "./outpost.config.ts";

export const review = defineIsolatedTask({
  key: "review",
  request: () => ({
    repository,
    sandboxProvider,
    agent: coder,
    brief: { text: "Review the public API without modifying files." },
  }),
});

observe-telemetry.ts

import { defineWorkflow } from "@elie-laloum/outpost";
import { review } from "./telemetry-review.ts";
import { observation, telemetry } from "./telemetry.ts";

await defineWorkflow("review", [review]).start({ observation });
await observation.close();
telemetry.close();

La trace imbrique les spans outpost.workflow, outpost.task, outpost.task.attempt et outpost.dispatch, avec un span par opération, par exemple outpost.sandbox.acquire. Sans SDK enregistré, les objets de l’API n’exportent rien.

Référence API : createOpenTelemetryObserver.

telemetry.close() termine les spans encore ouverts. Votre application vide et arrête le SDK. Passez onError pour recevoir les erreurs d’instrumentation ; elles ne changent jamais le résultat d’une exécution.

Sans hub

Passez l’observateur comme telemetry à start() pour les spans de workflow, de tâche et de tentative, ou à dispatch() pour un span de dispatch. Les spans d’opération nécessitent le hub.

Attention

Branchez l’observateur une seule fois par exécution : passer à la même exécution telemetry et un hub contenant telemetry.sink duplique ses spans et ses métriques.

Traiter les événements d’agent de façon asynchrone

createCustomReporter() construit une fonction de rappel observe à partir de traitements indexés par type d’événement. Les traitements peuvent être asynchrones ; ils passent par une file limitée, comme les récepteurs du hub.

import { appendFile } from "node:fs/promises";
import { createCustomReporter, dispatch } from "@elie-laloum/outpost";
import { coder, repository, sandboxProvider } from "./outpost.config.ts";

const report = createCustomReporter({
  async tool(event) {
    await appendFile("tools.log", `${event.at} ${event.name}\n`);
  },
});
await dispatch({
  repository,
  sandboxProvider,
  agent: coder,
  brief: { text: "Summarize the public API without changing files." },
  observe: report,
});
await report.flush();

Le dispatch attend les traitements en cours avant de rendre la main et signale la première erreur de traitement dans result.observerErrors. report.flush() relance cette erreur. Le second argument accepte onError, appelé à chaque échec, ainsi que capacity et deliveryTimeoutMs pour la file.

Limites

• Le hub est un flux en mémoire et en direct : il ne stocke rien, et un récepteur lent perd des événements. Pour relire les événements après l’exécution, utilisez le journal du dispatch, lui-même un récepteur soumis aux mêmes limites capacity et deliveryTimeoutMs.
• Un récepteur désactivé le reste pendant toute la vie du hub, et errors conserve les 100 premières erreurs.
• Un hub fermé ignore les nouveaux événements. Un hub réutilisé entre plusieurs exécutions conserve ses errors et son compteur dropped : les observerErrors de chaque exécution incluent alors les erreurs précédentes.
• Les événements émis sur un worker distant restent sur le hub de ce worker.

API : createObservationHub · ObservationHub · Observation · ObservationEvent · OperationEvent · createOpenTelemetryObserver · OpenTelemetryObserver · createCustomReporter.

Observer les décisions et sélections

Les évaluations de décision émettent des résumés de cycle de vie avec la source decision. Les harnesses routés émettent des événements d’agent model-route indiquant modèle effectif, motif et confiance native facultative. Passez observation à decide() pour une évaluation directe ; tâches et harnesses propagent les scopes workflow, tâche, passage et sous-agent. Les états et réponses complets exigent un hub verbose. L’usage valide est compté de façon synchrone, indépendamment des livraisons et erreurs des sinks.