fix(web): disconnect agent connect streams on detail teardown

Author: 0xTomDaniel

typescript/clients/web-ag-ui/apps/agent-gmx-allora/src/workflow/nodes/collectDelegations.ts

@@ -53,6 +53,19 @@ const resolveDelegationOnboarding = (state: ClmmState): OnboardingState => {
   return { step: 2, key: DELEGATION_STEP_KEY };
 };
 
+function readConfigString(
+  config: CopilotKitConfig,
+  key: 'thread_id' | 'checkpoint_id' | 'checkpoint_ns',
+): string | undefined {
+  const root = config as { configurable?: unknown };
+  if (!root.configurable || typeof root.configurable !== 'object') {
+    return undefined;
+  }
+  const configurable = root.configurable as Record<string, unknown>;
+  const value = configurable[key];
+  return typeof value === 'string' && value.length > 0 ? value : undefined;
+}
+
 const HexSchema = z
   .string()
   .regex(/^0x[0-9a-fA-F]*$/u)
@@ -125,19 +138,33 @@ export const collectDelegationsNode = async (
   state: ClmmState,
   config: CopilotKitConfig,
 ): Promise<ClmmUpdate | Command<string, ClmmUpdate>> => {
+  const runMetadata = {
+    threadId: readConfigString(config, 'thread_id'),
+    checkpointId: readConfigString(config, 'checkpoint_id'),
+    checkpointNamespace: readConfigString(config, 'checkpoint_ns'),
+  };
   const delegationOnboarding = resolveDelegationOnboarding(state);
+  const taskState = state.thread.task?.taskStatus?.state;
+  const taskMessage = state.thread.task?.taskStatus?.message?.content;
+  const isAwaitingDelegationInput =
+    taskState === 'input-required' &&
+    typeof taskMessage === 'string' &&
+    taskMessage.toLowerCase().includes('delegation approval');
   logInfo('collectDelegations: entering node', {
+    ...runMetadata,
     delegationsBypassActive: state.thread.delegationsBypassActive === true,
     hasDelegationBundle: Boolean(state.thread.delegationBundle),
   });
   logWarn('collectDelegations: node entered', {
+    ...runMetadata,
     onboardingStatus: state.thread.onboardingFlow?.status,
     onboardingStep: state.thread.onboarding?.step,
     onboardingKey: state.thread.onboarding?.key,
     delegationsBypassActive: state.thread.delegationsBypassActive === true,
     hasDelegationBundle: Boolean(state.thread.delegationBundle),
     hasOperatorInput: Boolean(state.thread.operatorInput),
     hasFundingTokenInput: Boolean(state.thread.fundingTokenInput),
+    isAwaitingDelegationInput,
   });
 
   if (state.thread.delegationsBypassActive === true) {
@@ -235,6 +262,16 @@ export const collectDelegationsNode = async (
     nextTaskMessage: awaitingMessage,
   });
   const hasRunnableConfig = Boolean((config as { configurable?: unknown }).configurable);
+  logInfo('collectDelegations: pause transition prepared', {
+    ...runMetadata,
+    hasRunnableConfig,
+    shouldPersistPendingState,
+    isAwaitingDelegationInput,
+    currentOnboardingStep: state.thread.onboarding?.step,
+    currentOnboardingKey: state.thread.onboarding?.key,
+    nextOnboardingStep: delegationOnboarding.step,
+    nextOnboardingKey: delegationOnboarding.key,
+  });
   const pauseSnapshotView = applyThreadPatch(state, pendingView);
   if (hasRunnableConfig && shouldPersistPendingState) {
     const mergedView = pauseSnapshotView;
@@ -244,6 +281,10 @@ export const collectDelegationsNode = async (
       thread: mergedView,
       metadata: {
         pauseMechanism: 'checkpoint-and-interrupt',
+        ...runMetadata,
+        hasRunnableConfig,
+        shouldPersistPendingState,
+        isAwaitingDelegationInput,
       },
     });
     await copilotkitEmitState(config, {
@@ -272,18 +313,30 @@ export const collectDelegationsNode = async (
     thread: pauseSnapshotView,
     metadata: {
       pauseMechanism: 'interrupt',
+      ...runMetadata,
+      hasRunnableConfig,
       checkpointPersisted: shouldPersistPendingState,
+      isAwaitingDelegationInput,
     },
   });
 
   const interruptResult = await requestInterruptPayload({
     request,
     interrupt,
   });
+  logInfo('collectDelegations: interrupt payload received', {
+    ...runMetadata,
+    rawPayloadType: typeof interruptResult.raw,
+    hasDecodedPayload: interruptResult.decoded !== undefined,
+  });
   const parsed = DelegationSigningResponseSchema.safeParse(interruptResult.decoded);
   if (!parsed.success) {
     const issues = parsed.error.issues.map((issue) => issue.message).join('; ');
     const failureMessage = `Invalid delegation signing response: ${issues}`;
+    logWarn('collectDelegations: invalid delegation signing payload', {
+      ...runMetadata,
+      issues,
+    });
     const { task, statusEvent } = buildTaskStatus(awaitingInput.task, 'failed', failureMessage);
     const failedView = applyThreadPatch(state, {
       task,
@@ -304,6 +357,9 @@ export const collectDelegationsNode = async (
   }
 
   if (parsed.data.outcome === 'rejected') {
+    logWarn('collectDelegations: delegation signing rejected by user', {
+      ...runMetadata,
+    });
     const { task, statusEvent } = buildTaskStatus(
       awaitingInput.task,
       'failed',
@@ -344,6 +400,10 @@ export const collectDelegationsNode = async (
     descriptions: [...DELEGATION_DESCRIPTIONS],
     warnings,
   };
+  logInfo('collectDelegations: delegation signing accepted', {
+    ...runMetadata,
+    signedDelegationCount: delegationBundle.delegations.length,
+  });
 
   const { task, statusEvent } = buildTaskStatus(
     awaitingInput.task,

typescript/clients/web-ag-ui/apps/web/src/app/api/agent-disconnect/route.ts

@@ -0,0 +1,75 @@
+import { NextRequest } from 'next/server';
+import { z } from 'zod';
+
+type ConnectAbortHandler = () => void;
+type ConnectAbortersByThread = Map<string, Set<ConnectAbortHandler>>;
+
+type RuntimeGlobals = typeof globalThis & {
+  __copilotkitConnectAbortersByThread?: ConnectAbortersByThread;
+};
+
+const disconnectPayloadSchema = z.object({
+  agentId: z.string().min(1),
+  threadId: z.string().min(1),
+});
+
+function getConnectAbortersByThread(): ConnectAbortersByThread | null {
+  const runtimeGlobals = globalThis as RuntimeGlobals;
+  const aborters = runtimeGlobals.__copilotkitConnectAbortersByThread;
+  if (!aborters) return null;
+  return aborters;
+}
+
+export async function POST(req: NextRequest): Promise<Response> {
+  let payload: unknown;
+  try {
+    payload = await req.json();
+  } catch {
+    return Response.json(
+      {
+        ok: false,
+        error: 'Invalid disconnect payload',
+      },
+      { status: 400 },
+    );
+  }
+
+  const parsed = disconnectPayloadSchema.safeParse(payload);
+  if (!parsed.success) {
+    return Response.json(
+      {
+        ok: false,
+        error: 'Invalid disconnect payload',
+      },
+      { status: 400 },
+    );
+  }
+
+  const { agentId, threadId } = parsed.data;
+  const connectKey = `${agentId}:${threadId}`;
+  const abortersByThread = getConnectAbortersByThread();
+  const aborters = abortersByThread?.get(connectKey);
+  if (!aborters || aborters.size === 0) {
+    return Response.json({
+      ok: true,
+      abortedCount: 0,
+    });
+  }
+
+  const handlers = Array.from(aborters);
+  abortersByThread?.delete(connectKey);
+  let abortedCount = 0;
+  for (const abortHandler of handlers) {
+    try {
+      abortHandler();
+      abortedCount += 1;
+    } catch {
+      // best-effort disconnect
+    }
+  }
+
+  return Response.json({
+    ok: true,
+    abortedCount,
+  });
+}

typescript/clients/web-ag-ui/apps/web/src/app/api/agent-disconnect/route.unit.test.ts

@@ -0,0 +1,64 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { POST } from './route';
+
+type ConnectAbortHandler = () => void;
+type ConnectAbortersByThread = Map<string, Set<ConnectAbortHandler>>;
+
+type RuntimeGlobals = typeof globalThis & {
+  __copilotkitConnectAbortersByThread?: ConnectAbortersByThread;
+};
+
+function getRuntimeGlobals(): RuntimeGlobals {
+  return globalThis as RuntimeGlobals;
+}
+
+describe('/api/agent-disconnect', () => {
+  beforeEach(() => {
+    getRuntimeGlobals().__copilotkitConnectAbortersByThread = new Map();
+  });
+
+  it('aborts active connect handlers for the agent/thread key', async () => {
+    const abortA = vi.fn();
+    const abortB = vi.fn();
+    getRuntimeGlobals().__copilotkitConnectAbortersByThread?.set(
+      'agent-gmx-allora:thread-1',
+      new Set([abortA, abortB]),
+    );
+
+    const request = new Request('http://localhost/api/agent-disconnect', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: JSON.stringify({
+        agentId: 'agent-gmx-allora',
+        threadId: 'thread-1',
+      }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(200);
+    expect(await response.json()).toEqual({
+      ok: true,
+      abortedCount: 2,
+    });
+    expect(abortA).toHaveBeenCalledTimes(1);
+    expect(abortB).toHaveBeenCalledTimes(1);
+  });
+
+  it('returns 400 for invalid payloads', async () => {
+    const request = new Request('http://localhost/api/agent-disconnect', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: JSON.stringify({
+        agentId: '',
+      }),
+    });
+
+    const response = await POST(request);
+    expect(response.status).toBe(400);
+    expect(await response.json()).toEqual({
+      ok: false,
+      error: 'Invalid disconnect payload',
+    });
+  });
+});