fix: harden live updates and user-facing workflow errors

Author: buger

src/agent-protocol/task-live-update-slack.ts

@@ -8,6 +8,8 @@ export class SlackTaskLiveUpdateSink implements TaskLiveUpdateSink {
   private messageTs?: string;
   /** Serialize all publish calls to prevent concurrent postMessage races */
   private publishQueue: Promise<{ ref?: Record<string, unknown> } | null> = Promise.resolve(null);
+  /** Count consecutive chat.update failures for observability and final-message fallback decisions */
+  private consecutiveUpdateFailures = 0;
 
   constructor(
     private readonly slack: SlackClient,
@@ -65,10 +67,23 @@ export class SlackTaskLiveUpdateSink implements TaskLiveUpdateSink {
         ts: this.messageTs,
         text: formatSlackText(text),
       });
-      if (updated?.ok) return null;
+      if (updated?.ok) {
+        this.consecutiveUpdateFailures = 0;
+        return null;
+      }
+      this.consecutiveUpdateFailures++;
       logger.warn(
-        `[TaskLiveUpdates][Slack] chat.update failed for ts=${this.messageTs} error=${updated?.error || 'unknown_error'}; falling back to chat.postMessage`
+        `[TaskLiveUpdates][Slack] chat.update failed for ts=${this.messageTs} error=${updated?.error || 'unknown_error'} (failure #${this.consecutiveUpdateFailures})`
       );
+      // Never create a new progress message when an update fails.
+      // Keeping the existing live-update message in place is less noisy than
+      // attempting a post+delete recovery that can leave orphaned messages.
+      if (mode === 'progress') {
+        logger.warn(
+          `[TaskLiveUpdates][Slack] Preserving existing live update ts=${this.messageTs}; suppressing progress post fallback to avoid Slack spam`
+        );
+        return null;
+      }
     }
     logger.info(
       `[TaskLiveUpdates][Slack] Posting live update message in channel=${this.channel} thread=${this.threadTs}`
@@ -81,7 +96,8 @@ export class SlackTaskLiveUpdateSink implements TaskLiveUpdateSink {
     if (posted?.ok && posted.ts) {
       const previousTs = this.messageTs;
       this.messageTs = posted.ts;
-      if (mode === 'final' && previousTs && previousTs !== posted.ts) {
+      // Clean up the old stale message to avoid orphaned progress messages
+      if (previousTs && previousTs !== posted.ts) {
         const deleted = await this.slack.chat.delete({
           channel: this.channel,
           ts: previousTs,
@@ -92,7 +108,7 @@ export class SlackTaskLiveUpdateSink implements TaskLiveUpdateSink {
           );
         } else {
           logger.info(
-            `[TaskLiveUpdates][Slack] Removed stale live update message ts=${previousTs} after final fallback post`
+            `[TaskLiveUpdates][Slack] Removed stale live update message ts=${previousTs} after fallback post`
           );
         }
       }

src/agent-protocol/task-live-updates.ts

@@ -208,6 +208,8 @@ export class TaskLiveUpdateManager {
   private metadataRefreshTimer?: ReturnType<typeof setInterval>;
   private running = false;
   private inflightTick?: Promise<void>;
+  private inflightMetadataRefresh?: Promise<void>;
+  private readonly inflightPublishes = new Set<Promise<unknown>>();
   private started = false;
   private completed = false;
   private readonly startedAt = new Date();
@@ -259,12 +261,7 @@ export class TaskLiveUpdateManager {
     if (this.completed) return;
     this.completed = true;
     this.stop();
-    // Wait for any in-flight tick to finish so the sink's publish queue is drained
-    if (this.running && this.inflightTick) {
-      try {
-        await this.inflightTick;
-      } catch {}
-    }
+    await this.waitForInflightSinkPublishes();
     try {
       logger.info(`[TaskLiveUpdates] Publishing final success update for task ${this.ctx.taskId}`);
       const result = await this.ctx.sink.complete(this.decorateText(finalText));
@@ -283,12 +280,7 @@ export class TaskLiveUpdateManager {
     if (this.completed) return;
     this.completed = true;
     this.stop();
-    // Wait for any in-flight tick to finish so the sink's publish queue is drained
-    if (this.running && this.inflightTick) {
-      try {
-        await this.inflightTick;
-      } catch {}
-    }
+    await this.waitForInflightSinkPublishes();
     try {
       logger.info(`[TaskLiveUpdates] Publishing final failure update for task ${this.ctx.taskId}`);
       const result = await this.ctx.sink.fail(this.decorateText(finalText));
@@ -319,6 +311,17 @@ export class TaskLiveUpdateManager {
   }
 
   async tick(): Promise<void> {
+    if (this.inflightTick) return this.inflightTick;
+    const promise = this.runTick().finally(() => {
+      if (this.inflightTick === promise) {
+        this.inflightTick = undefined;
+      }
+    });
+    this.inflightTick = promise;
+    return promise;
+  }
+
+  private async runTick(): Promise<void> {
     if (this.completed || this.running) return;
     const traceState = this.getTraceState();
     if (!traceState.traceRef && !traceState.traceId) {
@@ -402,7 +405,13 @@ export class TaskLiveUpdateManager {
         },
         traceState.traceId
       );
-      const result = await this.ctx.sink.update(message);
+      if (this.completed) {
+        logger.debug(
+          `[TaskLiveUpdates] Aborting progress publish for task ${this.ctx.taskId}: task already completed before sink update`
+        );
+        return;
+      }
+      const result = await this.trackPublish(this.ctx.sink.update(message));
       this.recordSinkRef(result);
       this.ctx.appendHistory?.(cleaned, 'progress');
       this.lastUpdateText = cleaned;
@@ -424,27 +433,43 @@ export class TaskLiveUpdateManager {
   private async runFirstTick(): Promise<void> {
     if (this.completed) return;
     logger.debug(`[TaskLiveUpdates] Running first scheduled tick for task ${this.ctx.taskId}`);
-    this.inflightTick = this.tick();
-    await this.inflightTick;
+    await this.tick();
     if (this.completed) return;
     this.timer = setInterval(() => {
-      this.inflightTick = this.tick();
+      void this.tick();
     }, this.ctx.config.intervalSeconds * 1000);
     if (typeof (this.timer as any)?.unref === 'function') {
       (this.timer as any).unref();
     }
     this.metadataRefreshTimer = setInterval(() => {
-      void this.refreshProgressMetadata();
+      void this.scheduleMetadataRefresh();
     }, DEFAULT_TASK_LIVE_UPDATE_METADATA_REFRESH_SECONDS * 1000);
     if (typeof (this.metadataRefreshTimer as any)?.unref === 'function') {
       (this.metadataRefreshTimer as any).unref();
     }
   }
 
+  private async waitForInflightSinkPublishes(): Promise<void> {
+    const pending = [...this.inflightPublishes];
+    if (pending.length === 0) return;
+    try {
+      await Promise.allSettled(pending);
+    } catch {}
+  }
+
   private recordSinkRef(result: { ref?: Record<string, unknown> } | null | undefined): void {
     if (result?.ref) this.ctx.onPostedRef?.(result.ref);
   }
 
+  private async trackPublish<T>(publish: Promise<T>): Promise<T> {
+    this.inflightPublishes.add(publish);
+    try {
+      return await publish;
+    } finally {
+      this.inflightPublishes.delete(publish);
+    }
+  }
+
   private getTraceState(): { traceRef?: string; traceId?: string } {
     const resolved = this.ctx.resolveTraceState?.();
     return {
@@ -462,7 +487,7 @@ export class TaskLiveUpdateManager {
   private decorateProgressText(
     text: string,
     timing: ProgressTimingMetadata,
-    traceId?: string
+    _traceId?: string
   ): string {
     const normalized = normalizeProgressSummary(text);
     const taskSummary = formatTaskSummary(timing.taskSummary);
@@ -473,8 +498,21 @@ export class TaskLiveUpdateManager {
       taskSummary,
       normalized,
       formatProgressMetadata(timing),
+      `\`task_id: ${this.ctx.taskId}\``,
+      '`live_update: true`',
     ].filter(Boolean);
-    return this.decorateText(blocks.join('\n\n'), traceId);
+    return blocks.join('\n\n');
+  }
+
+  private scheduleMetadataRefresh(): Promise<void> {
+    if (this.inflightMetadataRefresh) return this.inflightMetadataRefresh;
+    const promise = this.refreshProgressMetadata().finally(() => {
+      if (this.inflightMetadataRefresh === promise) {
+        this.inflightMetadataRefresh = undefined;
+      }
+    });
+    this.inflightMetadataRefresh = promise;
+    return promise;
   }
 
   private async refreshProgressMetadata(): Promise<void> {
@@ -499,7 +537,7 @@ export class TaskLiveUpdateManager {
       logger.debug(
         `[TaskLiveUpdates] Refreshing metadata-only live update for task ${this.ctx.taskId}`
       );
-      const result = await this.ctx.sink.update(message);
+      const result = await this.trackPublish(this.ctx.sink.update(message));
       this.recordSinkRef(result);
       this.lastPostedMessage = message;
     } catch (err) {
@@ -559,7 +597,8 @@ export class TaskLiveUpdateManager {
       this.lastStallFallbackAt = now;
       return;
     }
-    const result = await this.ctx.sink.update(message);
+    if (this.completed) return;
+    const result = await this.trackPublish(this.ctx.sink.update(message));
     this.recordSinkRef(result);
     if (!this.lastUpdateText) {
       this.ctx.appendHistory?.(baseText, 'progress');
@@ -698,20 +737,28 @@ function formatTaskLabel(task: { id: string; title: string; status: string }): s
   return title || task.id || task.status;
 }
 
+function scopeHasRenderableTasks(scope: NonNullable<ProbeTaskSummary['scopes']>[number]): boolean {
+  if (scope.tasks.some(task => !task.synthetic)) return true;
+  return scope.children.some(child => scopeHasRenderableTasks(child));
+}
+
 function appendTaskScopeLines(
   lines: string[],
   scope: NonNullable<ProbeTaskSummary['scopes']>[number],
   depth = 0
 ): void {
+  if (!scopeHasRenderableTasks(scope)) return;
   const prefix = '  '.repeat(depth);
   if (depth > 0 || scope.label !== 'Main Agent') {
     lines.push(`${prefix}${scope.label}`);
   }
-  for (const task of scope.tasks.slice(0, 8)) {
+  const visibleTasks = scope.tasks.filter(task => !task.synthetic).slice(0, 8);
+  for (const task of visibleTasks) {
     lines.push(`${prefix}• ${formatTaskStatusMarker(task.status)} ${formatTaskLabel(task)}`);
   }
-  if (scope.tasks.length > 8) {
-    lines.push(`${prefix}• ... ${scope.tasks.length - 8} more`);
+  const hiddenCount = scope.tasks.filter(task => !task.synthetic).length - visibleTasks.length;
+  if (hiddenCount > 0) {
+    lines.push(`${prefix}• ... ${hiddenCount} more`);
   }
   for (const child of scope.children) {
     appendTaskScopeLines(lines, child, depth + 1);
@@ -724,14 +771,16 @@ function formatTaskSummary(summary?: ProbeTaskSummary): string | undefined {
   if (summary.scopes.length > 0) {
     for (const scope of summary.scopes) appendTaskScopeLines(lines, scope);
   } else {
-    for (const task of summary.tasks.slice(0, 8)) {
+    const visibleTasks = summary.tasks.filter(task => !task.synthetic).slice(0, 8);
+    for (const task of visibleTasks) {
       lines.push(`• ${formatTaskStatusMarker(task.status)} ${formatTaskLabel(task)}`);
     }
-    if (summary.tasks.length > 8) {
-      lines.push(`• ... ${summary.tasks.length - 8} more`);
+    const hiddenCount = summary.tasks.filter(task => !task.synthetic).length - visibleTasks.length;
+    if (hiddenCount > 0) {
+      lines.push(`• ... ${hiddenCount} more`);
     }
   }
-  return lines.join('\n');
+  return lines.length > 1 ? lines.join('\n') : undefined;
 }
 
 function formatSkillList(skills: string[]): string {

src/agent-protocol/trace-serializer.ts

@@ -114,6 +114,7 @@ export interface ProbeTaskSummaryItem {
   id: string;
   title: string;
   status: string;
+  synthetic?: boolean;
 }
 
 export interface ProbeTaskScopeSummary {
@@ -1256,7 +1257,7 @@ function buildProbeTaskScopeSummary(
       source: 'events',
       eventCount: 0,
       snapshotCount: 0,
-      tasks: [{ id: `__scope_${node.span.spanId}`, title: scopeLabel, status }],
+      tasks: [{ id: `__scope_${node.span.spanId}`, title: scopeLabel, status, synthetic: true }],
       children: [],
     };
   }
@@ -1312,6 +1313,7 @@ function buildProbeTaskScopeSummary(
           id: `__scope_${n.span.spanId}`,
           title: scopeLabel,
           status,
+          synthetic: true,
         });
       }
     }
@@ -1387,6 +1389,7 @@ function flattenTaskScopes(scopes: ProbeTaskScopeSummary[]): ProbeTaskSummaryIte
   const items: ProbeTaskSummaryItem[] = [];
   const visit = (scope: ProbeTaskScopeSummary) => {
     for (const task of scope.tasks) {
+      if (task.synthetic) continue;
       if (seen.has(task.id)) continue;
       seen.add(task.id);
       items.push(task);
@@ -1578,12 +1581,18 @@ function formatTaskLabel(task: { id: string; title: string; status: string }): s
   return title || task.id || task.status;
 }
 
+function scopeHasRenderableTasks(scope: ProbeTaskScopeSummary): boolean {
+  if (scope.tasks.some(task => !task.synthetic)) return true;
+  return scope.children.some(child => scopeHasRenderableTasks(child));
+}
+
 function appendTaskScopeLines(lines: string[], scope: ProbeTaskScopeSummary, depth = 0): void {
+  if (!scopeHasRenderableTasks(scope)) return;
   const prefix = '  '.repeat(depth);
   if (depth > 0 || scope.label !== ROOT_TASK_SCOPE_LABEL) {
     lines.push(`${prefix}${scope.label}`);
   }
-  for (const task of scope.tasks) {
+  for (const task of scope.tasks.filter(task => !task.synthetic)) {
     lines.push(`${prefix}  ${formatTaskStatus(task.status)} ${formatTaskLabel(task)}`);
   }
   for (const child of scope.children) {
@@ -1596,7 +1605,7 @@ export function buildTraceHeaderLines(
   taskSummary: ProbeTaskSummary | null
 ): string[] {
   const lines = [`Trace source: ${traceData.source || 'unknown'}`];
-  if (!taskSummary) {
+  if (!taskSummary || taskSummary.tasks.length === 0) {
     lines.push('Tasks: no task telemetry found in this trace');
     return lines;
   }
@@ -1883,10 +1892,12 @@ function renderYamlNode(
     const action = attrs['dedup.action'] || '?';
     const reason = attrs['dedup.reason'] || '';
     const rewritten = attrs['dedup.rewritten'] || '';
+    const error = attrs['dedup.error'] || '';
     const prevCount = attrs['dedup.previous_count'] || '0';
     let detail = `${action}`;
     if (rewritten) detail += ` → "${truncate(String(rewritten), 60)}"`;
     if (reason) detail += ` (${truncate(String(reason), 80)})`;
+    if (error) detail += ` [error: ${truncate(String(error), 120)}]`;
     lines.push(
       `${pad}dedup("${truncate(String(query), 60)}") [${prevCount} prior]: ${detail} — ${duration}`
     );
@@ -2630,9 +2641,11 @@ function formatSpanLine(
     const action = attrs['dedup.action'] || '?';
     const reason = attrs['dedup.reason'] || '';
     const rewritten = attrs['dedup.rewritten'] || '';
+    const error = attrs['dedup.error'] || '';
     let detail = `${action}`;
     if (rewritten) detail += ` → "${truncate(String(rewritten), 50)}"`;
     if (reason) detail += ` — ${truncate(String(reason), 60)}`;
+    if (error) detail += ` [error: ${truncate(String(error), 80)}]`;
     return { line: `dedup("${truncate(String(query), 50)}") ${detail} (${duration})` };
   }
 

src/agent-protocol/track-execution.ts

@@ -228,10 +228,9 @@ export async function trackExecution<T>(
         // best-effort — don't fail the task over metadata
       }
 
-      // Extract AI response text from the result.
+      // Extract response text from the result.
       // result.reviewSummary.history is keyed by checkId with arrays of outputs.
-      // We want the LAST check's text output (the final AI response), not the
-      // first (which is typically the intent router).
+      // We want the LAST check's text output (the final AI response or workflow text output).
       let responseText = 'Execution completed';
       try {
         const history = (result as any)?.reviewSummary?.history as
@@ -245,7 +244,9 @@ export async function trackExecution<T>(
             if (!Array.isArray(outputs)) continue;
             // Within a check, look at the last output first too
             for (let j = outputs.length - 1; j >= 0; j--) {
-              const text = (outputs[j] as any)?.text;
+              const out = outputs[j] as any;
+              // Direct text field (AI response or workflow text output)
+              const text = out?.text;
               if (typeof text === 'string' && text.trim().length > 0) {
                 responseText = text.trim();
                 break;