ref(telemetry): Give each context compaction its own span

TELEMETRY.md

@@ -208,6 +208,7 @@ attributes. Nested logs and spans inherit them.
 | Memory recall slow, empty, or noisy  | `app.memory.recall.*` on span `plugin.hook` (`userPrompt`), span `gen_ai.rerank`, `plugin.log.warn`          |
 | Attachment or web search failure     | `attachment.*`, `image.attachment.*`, `conversation.image.*`, `web.search.failed`                            |
 | Model request slow before gateway    | `app.ai.http.*`, `app.ai.event_loop.utilization` on span `gen_ai.chat`                                       |
+| Slow or failed context compaction    | span `agent.context_compaction`, `app.compaction.*`, `app.turn.compaction_wait_ms` on `chat.turn`            |
 | Slow or failing SQL                  | automatic `db` spans; group by `db.query.text` or `db.query.summary`                                         |
 
 ## Code Mode
@@ -293,6 +294,56 @@ compare the phases:
 - The same `service.instance.id` on more than one slow call: they shared one
   process.
 
+## Context Compaction
+
+A compaction replaces the history when its estimate is above the threshold.
+`chat/services/context-compaction-telemetry.ts` owns the span.
+
+- Each compaction has one `agent.context_compaction` span. Its child
+  `gen_ai.chat` span, with `gen_ai.prompt.name:junior.context_compaction`, is
+  the summary call. It has the `gen_ai.usage.*` attributes.
+- The span and all its children have `app.compaction.mode`. Group the summary
+  `gen_ai.chat` spans by mode to compare duration and cache reads.
+- `app.turn.compaction_wait_ms` on `chat.turn` is the time that the Turn
+  waited on compaction in this execution slice. It is absent when no
+  compaction ran.
+- Sentry data scrubbing removes the values of attribute keys that contain
+  `token`. So these keys do not contain it, but the estimates count tokens.
+
+Attributes on the `agent.context_compaction` span:
+
+| Attribute                             | Type   | Meaning                                       |
+| ------------------------------------- | ------ | --------------------------------------------- |
+| `app.compaction.mode`                 | string | `blocking`, `active`, or `handoff`            |
+| `app.compaction.outcome`              | string | `compacted` or `summary_failed`               |
+| `app.compaction.input.estimate`       | int    | estimated history tokens before compaction    |
+| `app.compaction.replacement.estimate` | int    | estimated history tokens after compaction     |
+| `app.compaction.threshold`            | int    | estimated tokens that start a compaction      |
+| `app.compaction.input_limit`          | int    | most estimated input tokens that a model gets |
+| `app.compaction.input_messages`       | int    | history messages before compaction            |
+| `app.compaction.retained_messages`    | int    | messages kept before the summary              |
+| `app.compaction.summary_chars`        | int    | summary length in characters                  |
+
+- `blocking` runs at Turn start. `active` runs between model calls in a Turn.
+  `handoff` runs for a model handoff.
+- A `handoff` span has only `app.compaction.mode`. A failed handoff or an
+  input above the limit sets `error.type`.
+
+Summary call latency and cache reads by mode. In queries, a numeric `app.*`
+attribute needs the `tags[<key>,number]` form.
+
+```text
+dataset=spans query='span.op:gen_ai.chat gen_ai.prompt.name:junior.context_compaction'
+fields=app.compaction.mode,count(),p50(span.duration),p95(span.duration),sum(gen_ai.usage.input_tokens),sum(gen_ai.usage.cache_read.input_tokens)
+```
+
+Turn wait on compaction.
+
+```text
+dataset=spans query='span.op:chat.turn has:tags[app.turn.compaction_wait_ms,number]'
+fields=count(),p50(tags[app.turn.compaction_wait_ms,number]),p95(tags[app.turn.compaction_wait_ms,number]),p50(span.duration)
+```
+
 ## Invariants
 
 - `slack.message.ingress` records sender, team, and `app.slack.membership`

packages/junior/src/chat/providers/slack/turn-input-telemetry.ts

@@ -4,6 +4,7 @@ import {
   type SetSpanAttributes,
 } from "@/chat/logging";
 import { continueTrace, getActiveSpan } from "@/chat/sentry";
+import { measureTurnCompactionWait } from "@/chat/services/context-compaction-telemetry";
 import type { TraceContext } from "@/chat/task-execution/state";
 
 /** Mailbox timing facts for the input messages that start a Slack turn. */
@@ -33,7 +34,8 @@ function spanLinks({ sentryTrace }: TraceContext) {
  *
  * The span links to the other trace contexts and to the span that was active,
  * such as the worker span. Without a trace context, the span is a child of the
- * active span. A malformed trace context starts a new trace.
+ * active span. A malformed trace context starts a new trace. The span records
+ * how long the Turn waited on context compaction.
  */
 export async function withTurnSpan<T>(
   context: LogContext,
@@ -42,7 +44,9 @@ export async function withTurnSpan<T>(
 ): Promise<T> {
   const [parent, ...others] = input?.traceContexts ?? [];
   if (!parent) {
-    return await withSpan("chat.turn", "chat.turn", context, callback);
+    return await withSpan("chat.turn", "chat.turn", context, (set) =>
+      measureTurnCompactionWait(set, () => callback(set)),
+    );
   }
   const activeSpan = getActiveSpan();
   const links = [
@@ -53,7 +57,9 @@ export async function withTurnSpan<T>(
   return await continueTrace({ baggage, sentryTrace }, () =>
     withSpan("chat.turn", "chat.turn", context, async (setSpanAttributes) => {
       getActiveSpan()?.addLinks(links);
-      return await callback(setSpanAttributes);
+      return await measureTurnCompactionWait(setSpanAttributes, () =>
+        callback(setSpanAttributes),
+      );
     }),
   );
 }

packages/junior/src/chat/services/context-compaction-telemetry.ts

@@ -0,0 +1,67 @@
+/**
+ * Context compaction telemetry.
+ *
+ * Each compaction gets one `agent.context_compaction` span. The span and its
+ * children, such as the summary `gen_ai.chat` span, carry
+ * `app.compaction.mode`. The `chat.turn` span sums the time that the Turn
+ * waited on compaction.
+ *
+ * Attribute keys do not contain "token": Sentry data scrubbing removes the
+ * values of keys that match it.
+ */
+import { AsyncLocalStorage } from "node:async_hooks";
+import { runWithLogAttributes } from "@/chat/log-context";
+import { withSpan, type SetSpanAttributes } from "@/chat/logging";
+
+/**
+ * Where a compaction runs: `blocking` at Turn start, `active` between model
+ * calls in a Turn, `handoff` for a model handoff.
+ */
+export type CompactionMode = "blocking" | "active" | "handoff";
+
+interface TurnCompactionWait {
+  ms: number;
+}
+
+const turnCompactionWait = new AsyncLocalStorage<TurnCompactionWait>();
+
+/** Run one compaction in its span and add its time to the Turn wait. */
+export async function withCompactionSpan<T>(
+  mode: CompactionMode,
+  callback: () => Promise<T>,
+): Promise<T> {
+  const startedAtMs = performance.now();
+  try {
+    return await runWithLogAttributes({ "app.compaction.mode": mode }, () =>
+      withSpan(
+        "agent.context_compaction",
+        "agent.context_compaction",
+        {},
+        callback,
+      ),
+    );
+  } finally {
+    // Every mode runs on the critical path of the Turn today.
+    const wait = turnCompactionWait.getStore();
+    if (wait) {
+      wait.ms += performance.now() - startedAtMs;
+    }
+  }
+}
+
+/** Run one `chat.turn` callback and record its compaction wait, if any. */
+export async function measureTurnCompactionWait<T>(
+  setTurnAttributes: SetSpanAttributes,
+  callback: () => Promise<T>,
+): Promise<T> {
+  const wait: TurnCompactionWait = { ms: 0 };
+  try {
+    return await turnCompactionWait.run(wait, callback);
+  } finally {
+    if (wait.ms > 0) {
+      setTurnAttributes({
+        "app.turn.compaction_wait_ms": Math.round(wait.ms),
+      });
+    }
+  }
+}

packages/junior/src/chat/services/context-compaction.ts

@@ -53,6 +53,7 @@ import {
   renderAgentsInstructions,
 } from "@/chat/repository-instructions";
 import { appendOpenPlan } from "@/chat/services/plan-continuation";
+import { withCompactionSpan } from "@/chat/services/context-compaction-telemetry";
 import { escapeXml } from "@/chat/xml";
 
 const RETAINED_USER_MESSAGE_TOKENS = 20_000;
@@ -570,7 +571,17 @@ async function maybeCompactWithDeps(
   if (source.estimatedTokens <= triggerTokens) {
     return { compacted: false, reason: "below_threshold" };
   }
+  return await withCompactionSpan("blocking", () =>
+    compactAboveThreshold(args, deps, source, triggerTokens),
+  );
+}
 
+async function compactAboveThreshold(
+  args: CompactContextArgs,
+  deps: ContextCompactorDeps,
+  source: { estimatedTokens: number; messages: PiMessage[] },
+  triggerTokens: number,
+): Promise<CompactContextResult> {
   await args.onCompactionStart?.();
 
   let summary: string;
@@ -588,6 +599,7 @@ async function maybeCompactWithDeps(
       "exception.message":
         error instanceof Error ? error.message : String(error),
     });
+    setSpanAttributes({ "app.compaction.outcome": "summary_failed" });
     return { compacted: false, reason: "summary_failed" };
   }
 
@@ -674,13 +686,16 @@ async function writeCompactedThreadContext(
   });
 
   setSpanAttributes({
+    "app.compaction.outcome": "compacted",
     "app.compaction.input_messages": sourceMessages.length,
     "app.compaction.retained_messages": replacement.length - 1,
     "app.compaction.summary_chars": summary.length,
+    "app.compaction.input.estimate": context.estimatedTokens,
+    "app.compaction.replacement.estimate": replacementInputTokens,
+    "app.compaction.input_limit": context.inputLimitTokens,
     ...(context.triggerTokens !== undefined
-      ? { "app.compaction.trigger_tokens": context.triggerTokens }
+      ? { "app.compaction.threshold": context.triggerTokens }
       : undefined),
-    "app.context_tokens_estimated": context.estimatedTokens,
   });
 
   return {
@@ -702,6 +717,15 @@ export function createContextCompactor(
 export async function compactContextForHandoff(
   args: HandoffContextArgs,
   deps: Pick<ContextCompactorDeps, "completeText">,
+): Promise<PiMessage[]> {
+  return await withCompactionSpan("handoff", () =>
+    compactForHandoff(args, deps),
+  );
+}
+
+async function compactForHandoff(
+  args: HandoffContextArgs,
+  deps: Pick<ContextCompactorDeps, "completeText">,
 ): Promise<PiMessage[]> {
   const retainedInstruction = await loadLastInstruction(args.conversationId);
   const contextMessage = runtimeContextMessage(
@@ -775,7 +799,18 @@ export async function compactActiveContextIfNeeded(
   if (source.estimatedTokens <= triggerTokens) {
     return { compacted: false, reason: "below_threshold" };
   }
+  return await withCompactionSpan("active", () =>
+    compactActiveAboveThreshold(args, deps, source, triggerTokens),
+  );
+}
 
+async function compactActiveAboveThreshold(
+  args: ActiveContextCompactionArgs,
+  deps: Pick<ContextCompactorDeps, "completeText">,
+  source: { estimatedTokens: number; messages: PiMessage[] },
+  triggerTokens: number,
+): Promise<CompactContextResult> {
+  const pendingMessages = args.pendingMessages ?? [];
   await args.onCompactionStart?.();
 
   const inputLimitTokens = getAgentContextInputLimitTokens(args.modelId);
@@ -801,6 +836,7 @@ export async function compactActiveContextIfNeeded(
       "exception.message":
         error instanceof Error ? error.message : String(error),
     });
+    setSpanAttributes({ "app.compaction.outcome": "summary_failed" });
     return { compacted: false, reason: "summary_failed" };
   }
 
@@ -875,13 +911,14 @@ export async function compactActiveContextIfNeeded(
     },
   });
   setSpanAttributes({
+    "app.compaction.outcome": "compacted",
     "app.compaction.input_messages": source.messages.length,
-    "app.compaction.replacement_tokens": replacementInputTokens,
     "app.compaction.retained_messages": instructionMessages.length,
     "app.compaction.summary_chars": summary.length,
-    "app.compaction.trigger_tokens": triggerTokens,
-    "app.context_input_limit_tokens": inputLimitTokens,
-    "app.context_tokens_estimated": source.estimatedTokens,
+    "app.compaction.input.estimate": source.estimatedTokens,
+    "app.compaction.replacement.estimate": replacementInputTokens,
+    "app.compaction.input_limit": inputLimitTokens,
+    "app.compaction.threshold": triggerTokens,
   });
   return { compacted: true, piMessages: messages };
 }