refactor: enhance interrupt handling and debug tracing in agent flow

Author: NoOne7135

application/ports.ts

@@ -1,12 +1,27 @@
-import type { AgentModeCompletionAdapter } from "../llm/agentModels.js";
+import type { CompletionAdapter } from "adminforth";
 import type { DetectedLanguage, PreviousUserMessage } from "../domain/languageDetect.js";
 import type {
   AgentMessage,
   AgentStreamChunk,
   AgentTurnContext,
   AgentTurnObservability,
+  PendingInterrupt,
 } from "../domain/turnTypes.js";
 
+export type AgentModelPurpose = "primary" | "summary";
+
+/**
+ * The LLM provider adapter contract (the "provider port"). Extends AdminForth's
+ * CompletionAdapter with the agent-spec factory. Declared in the application layer
+ * so the infrastructure (llm/) depends on this contract — not the other way around.
+ */
+export type AgentModeCompletionAdapter = CompletionAdapter & {
+  getLangChainAgentSpec(params: {
+    maxTokens: number;
+    purpose: AgentModelPurpose;
+  }): Promise<{ model: unknown; middleware?: unknown[] }> | { model: unknown; middleware?: unknown[] };
+};
+
 /**
  * Input for a single streamed agent turn. Either a fresh set of messages, or a
  * human-in-the-loop resume payload for a previously interrupted turn.
@@ -21,9 +36,8 @@ export type LlmStreamInput = {
 /**
  * The boundary between the application layer and the concrete LLM runtime
  * (LangChain / LangGraph). Everything provider-specific — model construction,
- * the agent graph, middleware, and the raw stream shape — lives behind this
- * port. The application layer depends only on this interface, so the turn
- * orchestration is testable with a scripted fake and reusable across providers.
+ * the agent graph, middleware, the raw stream shape, and the LangGraph interrupt
+ * shape — lives behind this port and is normalized before crossing it.
  */
 export interface LlmPort {
   /**
@@ -42,12 +56,13 @@ export interface LlmPort {
 
   /**
    * Rebuild the pending human-in-the-loop interrupts for a session from persisted
-   * state (the checkpointer). Used as a fallback when the in-process interrupt
-   * cache is empty — e.g. after a restart or on a second instance — so approvals
-   * survive process boundaries. Returns raw interrupt objects (LangGraph shape).
+   * checkpoint state (used when the in-process cache is empty — after a restart or
+   * on another instance). Returns already-normalized descriptors; the LangGraph
+   * interrupt shape never leaves the llm layer. Throws on a checkpoint/runtime
+   * failure (the caller must not treat that as "no pending interrupt").
    */
   getPendingInterrupts(input: {
     completionAdapter: AgentModeCompletionAdapter;
     sessionId: string;
-  }): Promise<unknown[]>;
+  }): Promise<PendingInterrupt[]>;
 }

application/runTurnUseCase.ts

@@ -1,6 +1,5 @@
 import { logger, type IAdminForth } from "adminforth";
 import { randomUUID } from "crypto";
-import { createSequenceDebugCollector } from "../llm/middleware/sequenceDebug.js";
 import { VegaLiteStreamBuffer } from "../domain/vegaLiteStreamBuffer.js";
 import { buildAgentTurnSystemPrompt } from "../domain/systemPrompt.js";
 import { getErrorMessage, isAbortError } from "../shared/errors.js";
@@ -13,15 +12,15 @@ import type {
   AgentTurnContext,
   AgentTurnObservability,
   BaseAgentTurnInput,
+  DebugSink,
   HandleTurnInput,
+  PendingInterrupt,
   RunAndPersistAgentResponseInput,
   RunAndPersistAgentResponseResult,
 } from "../domain/turnTypes.js";
 
 type AgentMode = PluginOptions["modes"][number];
 
-type PendingInterrupt = { id: string; count: number };
-
 type PreparedTurn = {
   prompt: string;
   sessionId: string;
@@ -41,28 +40,6 @@ function getApprovalDecision(input: BaseAgentTurnInput) {
     : undefined;
 }
 
-function getInterruptItems(interrupt: unknown): unknown[] {
-  return Array.isArray(interrupt) ? interrupt : [interrupt];
-}
-
-function getHitlInterrupts(interrupt: unknown): PendingInterrupt[] {
-  return getInterruptItems(interrupt).flatMap((item) => {
-    const value = item && typeof item === "object" && "value" in item
-      ? (item as { value: unknown }).value
-      : item;
-    const actionRequests = value && typeof value === "object"
-      ? (value as { actionRequests?: unknown }).actionRequests
-      : undefined;
-    const interruptId = item && typeof item === "object"
-      ? (item as { id?: unknown }).id
-      : undefined;
-
-    return typeof interruptId === "string" && Array.isArray(actionRequests)
-      ? [{ id: interruptId, count: actionRequests.length }]
-      : [];
-  });
-}
-
 function buildHitlDecision(decision: "approve" | "reject", prompt?: string) {
   if (decision === "approve") {
     return { type: "approve" as const };
@@ -129,6 +106,10 @@ export type RunTurnUseCaseDeps = {
   modes: PluginOptions["modes"];
   getAdminforth: () => IAdminForth;
   getAgentSystemPrompt: () => Promise<string>;
+  /** True when a persistent checkpointer is configured (not the in-memory MemorySaver). */
+  hasPersistentCheckpointer: boolean;
+  /** Factory for the per-turn debug sink; the concrete impl lives in the llm layer. */
+  createDebugSink: () => DebugSink;
 };
 
 /**
@@ -161,23 +142,32 @@ export class RunTurnUseCase {
   }
 
   /**
-   * Resolve the pending HITL interrupts for a resume. Prefers the in-process cache
-   * (populated when the interrupt fired this run) and falls back to the persisted
-   * checkpoint via the LLM port when the cache is empty (restart / other instance).
+   * Resolve the pending HITL interrupts for a resume. With a persistent checkpointer the
+   * checkpoint is authoritative (correct across instances and restarts); with the in-memory
+   * MemorySaver the in-process cache is authoritative (single process, cannot be stale).
    */
   private async resolvePendingInterrupts(sessionId: string, mode: AgentMode): Promise<PendingInterrupt[]> {
-    const cached = this.pendingInterrupts.get(sessionId);
-    if (cached && cached.length > 0) {
-      return cached;
+    if (this.deps.hasPersistentCheckpointer) {
+      // The in-process cache can be stale after a cross-instance resume, so it is intentionally
+      // NOT consulted here. A failure to read the checkpoint is surfaced as a real error — never
+      // masked as "no pending approval".
+      try {
+        return await this.deps.llm.getPendingInterrupts({
+          completionAdapter: mode.completionAdapter,
+          sessionId,
+        });
+      } catch (error) {
+        logger.error(
+          `Failed to read pending approval state from checkpoint for session "${sessionId}": ${getErrorMessage(error)}`,
+        );
+        throw error;
+      }
     }
-    const raw = await this.deps.llm
-      .getPendingInterrupts({ completionAdapter: mode.completionAdapter, sessionId })
-      .catch(() => [] as unknown[]);
-    return getHitlInterrupts(raw);
+    return this.pendingInterrupts.get(sessionId) ?? [];
   }
 
   private async prepareTurn(input: RunAndPersistAgentResponseInput): Promise<PreparedTurn> {
-    const sequenceDebugSink = createSequenceDebugCollector();
+    const sequenceDebugSink = this.deps.createDebugSink();
     const mode = this.resolveMode(input.modeName);
     const approvalDecision = getApprovalDecision(input);
     const shouldResume = Boolean(approvalDecision);
@@ -249,11 +239,14 @@ export class RunTurnUseCase {
     ];
   }
 
-  private async handleInterrupt(prepared: PreparedTurn, interrupt: unknown) {
-    const interrupts = getHitlInterrupts(interrupt);
+  private async handleInterrupt(
+    prepared: PreparedTurn,
+    interrupt: unknown,
+    descriptors: PendingInterrupt[],
+  ) {
     const existing = this.pendingInterrupts.get(prepared.sessionId) ?? [];
     const merged = new Map(existing.map((item) => [item.id, item.count]));
-    for (const item of interrupts) {
+    for (const item of descriptors) {
       merged.set(item.id, item.count);
     }
     this.pendingInterrupts.set(
@@ -299,7 +292,7 @@ export class RunTurnUseCase {
 
         if (chunk.kind === "interrupt") {
           interrupted = true;
-          await this.handleInterrupt(prepared, chunk.interrupt);
+          await this.handleInterrupt(prepared, chunk.interrupt, chunk.descriptors);
           continue;
         }
 

domain/turnTypes.ts

@@ -1,12 +1,25 @@
 import type { AdminUser, AudioAdapter } from "adminforth";
-import type { Messages } from "@langchain/langgraph";
-import type { Command } from "@langchain/langgraph";
-import type { AgentChatModel, AgentMiddleware } from "../llm/agentModels.js";
-import type { SequenceDebugCollector } from "../llm/middleware/sequenceDebug.js";
 import type { PreviousUserMessage } from "./languageDetect.js";
 import type { CurrentPageContext } from "../tools/getUserLocation.js";
 import type { AgentEventEmitter } from "./agentEvents.js";
 
+/**
+ * Minimal sink the application layer uses to persist per-turn debug traces. The
+ * concrete implementation (the sequence-debug collector) lives in the llm layer;
+ * the domain/application layers depend only on this narrow interface, so the
+ * dependency direction stays llm -> application/domain.
+ */
+export type DebugSink = {
+  flush(): void;
+  getHistory(): unknown[];
+};
+
+/** A pending human-in-the-loop approval, normalized (provider-agnostic). */
+export type PendingInterrupt = {
+  id: string;
+  count: number;
+};
+
 export type BaseAgentTurnInput = {
   prompt: string;
   sessionId: string;
@@ -51,35 +64,7 @@ export type AgentTurnContext = {
 
 export type AgentTurnObservability = {
   emit?: AgentEventEmitter;
-  sequenceDebugSink: SequenceDebugCollector;
-};
-
-export type PreparedAgentTurn = {
-  prompt: string;
-  sessionId: string;
-  turnId: string;
-  previousUserMessages: PreviousUserMessage[];
-  modeName?: string | null;
-  context: AgentTurnContext;
-  observability: AgentTurnObservability;
-  resume?: {
-    decision: "approve" | "reject";
-    interrupts?: { id: string; count: number }[];
-  };
-  initialResponse?: string;
-};
-
-export type AgentTurnModels = {
-  model: AgentChatModel;
-  summaryModel: AgentChatModel;
-  modelMiddleware?: AgentMiddleware[];
-};
-
-export type AgentRuntimeRunInput = {
-  models: AgentTurnModels;
-  input: { messages: Messages } | Command;
-  context: AgentTurnContext;
-  observability: AgentTurnObservability;
+  sequenceDebugSink: DebugSink;
 };
 
 export type RunAndPersistAgentResponseInput = BaseAgentTurnInput & {
@@ -116,4 +101,4 @@ export type AgentMessage = {
 export type AgentStreamChunk =
   | { kind: "text"; delta: string }
   | { kind: "reasoning"; delta: string }
-  | { kind: "interrupt"; interrupt: unknown };
+  | { kind: "interrupt"; interrupt: unknown; descriptors: PendingInterrupt[] };

index.ts

@@ -17,6 +17,7 @@ import type { AgentEndpointsContext } from "./transport/http/context.js";
 import { AgentSessionStore } from "./persistence/sessionStore.js";
 import { ChatSurfaceService } from "./transport/surfaces/chatSurfaceService.js";
 import { RunTurnUseCase } from "./application/runTurnUseCase.js";
+import { createSequenceDebugCollector } from "./llm/middleware/sequenceDebug.js";
 import { LangGraphLlm } from "./llm/langGraphLlm.js";
 import { AgentModelFactory } from "./llm/modelFactory.js";
 import { AgentRuntime } from "./llm/agentRuntime.js";
@@ -94,6 +95,8 @@ export default class AdminForthAgentPlugin extends AdminForthPlugin {
       modes: this.options.modes,
       getAdminforth: () => this.adminforth,
       getAgentSystemPrompt: this.getAgentSystemPrompt.bind(this),
+      hasPersistentCheckpointer: Boolean(this.options.checkpointResource),
+      createDebugSink: createSequenceDebugCollector,
     });
     this.speechTurnService = new SpeechTurnService(
       this.runTurnUseCase.runAndPersistAgentResponse.bind(this.runTurnUseCase),

llm/agentModels.ts

@@ -5,26 +5,30 @@ import {
 } from "adminforth";
 import { BaseCallbackHandler } from "@langchain/core/callbacks/base";
 import type { LLMResult } from "@langchain/core/outputs";
+import type { Messages, Command } from "@langchain/langgraph";
 import {
   createSequenceDebugMiddleware,
 } from "./middleware/sequenceDebug.js";
+import type { AgentModeCompletionAdapter, AgentModelPurpose } from "../application/ports.js";
+import type { AgentTurnContext, AgentTurnObservability } from "../domain/turnTypes.js";
+
+export type { AgentModeCompletionAdapter, AgentModelPurpose } from "../application/ports.js";
 
 export type AgentChatModel = BaseChatModel<any, any>;
-export type AgentModelPurpose = "primary" | "summary";
-export type AgentModeCompletionAdapter = CompletionAdapter & {
-  getLangChainAgentSpec(params: {
-    maxTokens: number;
-    purpose: AgentModelPurpose;
-  }): Promise<{
-    model: unknown;
-    middleware?: unknown[];
-  }> | {
-    model: unknown;
-    middleware?: unknown[];
-  };
+export type AgentMiddleware = ReturnType<typeof createSequenceDebugMiddleware>;
+
+export type AgentTurnModels = {
+  model: AgentChatModel;
+  summaryModel: AgentChatModel;
+  modelMiddleware?: AgentMiddleware[];
 };
 
-export type AgentMiddleware = ReturnType<typeof createSequenceDebugMiddleware>;
+export type AgentRuntimeRunInput = {
+  models: AgentTurnModels;
+  input: { messages: Messages } | Command;
+  context: AgentTurnContext;
+  observability: AgentTurnObservability;
+};
 
 type AgentChatModelSpec = {
   model: AgentChatModel;

llm/agentRuntime.ts

@@ -2,10 +2,10 @@ import type { IAdminForth } from "adminforth";
 import { createAgent, summarizationMiddleware, humanInTheLoopMiddleware } from "langchain";
 import type { BaseCheckpointSaver } from "@langchain/langgraph";
 import { createApiBasedToolsMiddleware } from "./middleware/apiToolsMiddleware.js";
-import { createSequenceDebugMiddleware } from "./middleware/sequenceDebug.js";
+import { createSequenceDebugMiddleware, type SequenceDebugCollector } from "./middleware/sequenceDebug.js";
 import { createAgentLlmMetricsLogger } from "./agentModels.js";
 import type { AgentToolProvider } from "../tools/agentToolProvider.js";
-import type { AgentRuntimeRunInput, AgentTurnModels } from "../domain/turnTypes.js";
+import type { AgentRuntimeRunInput, AgentTurnModels } from "./agentModels.js";
 import { contextSchema, toLangchainAgentContext } from "./agentContext.js";
 import type { ApiBasedTool } from "../tools/apiBasedTools.js";
 
@@ -42,9 +42,10 @@ export class AgentRuntime {
       apiBasedTools,
       adminforth,
     );
-    const sequenceDebugMiddleware = createSequenceDebugMiddleware(
-      input.observability.sequenceDebugSink,
-    );
+    // The domain exposes only the narrow DebugSink; the concrete collector (with the
+    // model-call hooks the debug middleware needs) is an llm-layer detail.
+    const sequenceDebugSink = input.observability.sequenceDebugSink as SequenceDebugCollector;
+    const sequenceDebugMiddleware = createSequenceDebugMiddleware(sequenceDebugSink);
     const hitlMiddleware = humanInTheLoopMiddleware({
       interruptOn: createHumanInTheLoopInterrupts(apiBasedTools),
       descriptionPrefix: "Tool execution pending approval",
@@ -82,7 +83,7 @@ export class AgentRuntime {
         ...input.context,
         adminBaseUrl: adminforth.config.baseUrlSlashed,
         emit: input.observability.emit,
-        sequenceDebugSink: input.observability.sequenceDebugSink,
+        sequenceDebugSink,
       }),
     });
   }

llm/langGraphLlm.ts

@@ -3,10 +3,9 @@ import { HumanMessage, SystemMessage } from "langchain";
 import type { AgentRuntime } from "./agentRuntime.js";
 import type { AgentModelFactory } from "./modelFactory.js";
 import { detectUserLanguage, type PreviousUserMessage } from "../domain/languageDetect.js";
-import type { AgentModeCompletionAdapter } from "./agentModels.js";
-import { adaptRawStream } from "./streamAdapter.js";
-import type { LlmPort, LlmStreamInput } from "../application/ports.js";
-import type { AgentMessage } from "../domain/turnTypes.js";
+import { adaptRawStream, normalizeInterrupts } from "./streamAdapter.js";
+import type { AgentModeCompletionAdapter, LlmPort, LlmStreamInput } from "../application/ports.js";
+import type { AgentMessage, PendingInterrupt } from "../domain/turnTypes.js";
 
 function toLangchainMessages(messages: AgentMessage[]) {
   return messages.map((message) =>
@@ -59,8 +58,9 @@ export class LangGraphLlm implements LlmPort {
   async getPendingInterrupts(input: {
     completionAdapter: AgentModeCompletionAdapter;
     sessionId: string;
-  }): Promise<unknown[]> {
+  }): Promise<PendingInterrupt[]> {
     const models = await this.modelFactory.create(input.completionAdapter);
-    return this.runtime.getPendingInterrupts({ models, sessionId: input.sessionId });
+    const raw = await this.runtime.getPendingInterrupts({ models, sessionId: input.sessionId });
+    return normalizeInterrupts(raw);
   }
 }

llm/modelFactory.ts

@@ -1,6 +1,5 @@
 import type { CompletionAdapter } from "adminforth";
-import { createAgentChatModel } from "./agentModels.js";
-import type { AgentTurnModels } from "../domain/turnTypes.js";
+import { createAgentChatModel, type AgentTurnModels } from "./agentModels.js";
 
 export class AgentModelFactory {
   constructor(private readonly maxTokens: number) {}