fix(agents): stabilize onboarding state transitions and command routing

Author: 0xTomDaniel

typescript/clients/web-ag-ui/apps/agent-clmm/src/agent.ts

@@ -1,7 +1,6 @@
 import { pathToFileURL } from 'node:url';
 
 import { END, InMemoryStore, START, StateGraph } from '@langchain/langgraph';
-import { projectCycleCommandThread } from 'agent-workflow-core';
 import { v7 as uuidv7 } from 'uuid';
 import { privateKeyToAccount } from 'viem/accounts';
 import { z } from 'zod';
@@ -138,47 +137,6 @@ async function parseJsonResponse<T>(response: Response, schema: z.ZodSchema<T>):
   return schema.parse(payload);
 }
 
-type ThreadStateValues = Record<string, unknown>;
-
-const isRecord = (value: unknown): value is Record<string, unknown> =>
-  typeof value === 'object' && value !== null && !Array.isArray(value);
-
-function extractThreadStateValues(payload: unknown): ThreadStateValues | null {
-  if (!isRecord(payload)) {
-    return null;
-  }
-
-  const values = payload['values'];
-  if (isRecord(values)) {
-    return values;
-  }
-
-  const state = payload['state'];
-  if (isRecord(state)) {
-    return state;
-  }
-
-  const data = payload['data'];
-  if (isRecord(data)) {
-    return data;
-  }
-
-  if (isRecord(payload['thread'])) {
-    return payload;
-  }
-
-  return null;
-}
-
-async function fetchThreadStateValues(
-  baseUrl: string,
-  threadId: string,
-): Promise<ThreadStateValues | null> {
-  const response = await fetch(`${baseUrl}/threads/${threadId}/state`);
-  const payload = await parseJsonResponse(response, z.unknown());
-  return extractThreadStateValues(payload);
-}
-
 async function ensureThread(baseUrl: string, threadId: string, graphId: string) {
   const metadata = { graph_id: graphId };
   const response = await fetch(`${baseUrl}/threads`, {
@@ -197,24 +155,11 @@ async function ensureThread(baseUrl: string, threadId: string, graphId: string)
 }
 
 async function updateCycleState(baseUrl: string, threadId: string, runMessage: { id: string; role: 'user'; content: string }) {
-  let existingThread: Record<string, unknown> | null = null;
-  try {
-    const currentState = await fetchThreadStateValues(baseUrl, threadId);
-    if (currentState && isRecord(currentState['thread'])) {
-      existingThread = currentState['thread'];
-    }
-  } catch (error: unknown) {
-    const message =
-      error instanceof Error ? error.message : typeof error === 'string' ? error : 'Unknown error';
-    console.warn('[cron] Unable to fetch thread state before cycle update', { threadId, error: message });
-  }
-
-  const thread = projectCycleCommandThread(existingThread);
   const response = await fetch(`${baseUrl}/threads/${threadId}/state`, {
     method: 'POST',
     headers: { 'Content-Type': 'application/json' },
     body: JSON.stringify({
-      values: { messages: [runMessage], thread },
+      values: { messages: [runMessage] },
       as_node: 'runCommand',
     }),
   });
@@ -300,10 +245,11 @@ export async function runGraphOnce(
   const startedAt = Date.now();
   console.info(`[cron] Starting CLMM graph run via API (thread=${threadId})`);
 
+  const clientMutationId = uuidv7();
   const runMessage = {
     id: uuidv7(),
     role: 'user' as const,
-    content: JSON.stringify({ command: 'cycle' }),
+    content: JSON.stringify({ command: 'cycle', clientMutationId }),
   };
 
   const baseUrl = resolveLangGraphDeploymentUrl();

typescript/clients/web-ag-ui/apps/agent-clmm/src/agentCron.unit.test.ts

@@ -0,0 +1,59 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { runGraphOnce } from './agent.js';
+
+const ORIGINAL_FETCH = globalThis.fetch;
+
+function jsonResponse(payload: unknown, status = 200): Response {
+  return new Response(JSON.stringify(payload), {
+    status,
+    headers: { 'Content-Type': 'application/json' },
+  });
+}
+
+describe('runGraphOnce cron state update payload', () => {
+  beforeEach(() => {
+    process.env['A2A_TEST_AGENT_NODE_PRIVATE_KEY'] ??= `0x${'1'.repeat(64)}`;
+  });
+
+  afterEach(() => {
+    globalThis.fetch = ORIGINAL_FETCH;
+    vi.restoreAllMocks();
+  });
+
+  it('does not replay thread state into the cycle command update payload', async () => {
+    const threadId = 'thread-1';
+    const fetchMock = vi.fn();
+    globalThis.fetch = fetchMock as unknown as typeof fetch;
+
+    fetchMock.mockResolvedValueOnce(jsonResponse({ thread_id: threadId }));
+    fetchMock.mockResolvedValueOnce(jsonResponse({ thread_id: threadId }));
+
+    let stateUpdateBody: Record<string, unknown> | undefined;
+    fetchMock.mockImplementationOnce((_url: string | URL | globalThis.Request, init?: RequestInit) => {
+      const requestBody = typeof init?.body === 'string' ? init.body : '{}';
+      stateUpdateBody = JSON.parse(requestBody) as Record<string, unknown>;
+      return jsonResponse({ checkpoint_id: 'cp-1' });
+    });
+
+    fetchMock.mockResolvedValueOnce(jsonResponse({ run_id: 'run-1', status: 'running' }));
+    fetchMock.mockResolvedValueOnce(new Response(null, { status: 200 }));
+    fetchMock.mockResolvedValueOnce(jsonResponse({ run_id: 'run-1', status: 'success' }));
+
+    await runGraphOnce(threadId);
+
+    expect(stateUpdateBody).toBeDefined();
+    expect(stateUpdateBody).toMatchObject({
+      as_node: 'runCommand',
+      values: {
+        messages: [expect.objectContaining({ role: 'user' })],
+      },
+    });
+    expect((stateUpdateBody?.['values'] as Record<string, unknown>) ?? {}).not.toHaveProperty('thread');
+
+    const calledUrls = fetchMock.mock.calls
+      .map(([url]) => String(url))
+      .filter((url) => url.includes(`/threads/${threadId}/state`));
+    expect(calledUrls).toEqual([`http://localhost:8124/threads/${threadId}/state`]);
+  });
+});

typescript/clients/web-ag-ui/apps/agent-clmm/src/hire.ts

@@ -84,10 +84,11 @@ export async function startClmmHire(
   operatorInput: OperatorConfigInput,
   options?: { durability?: LangGraphDurability },
 ) {
+  const clientMutationId = uuidv7();
   const hireMessage = {
     id: uuidv7(),
     role: 'user' as const,
-    content: JSON.stringify({ command: 'hire' }),
+    content: JSON.stringify({ command: 'hire', clientMutationId }),
   };
 
   type ClmmInvokeInput = Parameters<typeof clmmGraph.invoke>[0];