fix(api): run.start on a cancelled thread must not fold input into a resume

.changeset/lucky-hands-listen.md

@@ -0,0 +1,5 @@
+---
+"@langchain/langgraph-api": patch
+---
+
+fix(api): `run.start` with input on a cancelled thread no longer folds the input into `Command(resume)`. A cancelled run shares the "interrupted" status with a genuine `interrupt()` pause, but has no pending interrupt to consume the resume value, so the submitted message was silently dropped. The input-vs-resume decision now keys on whether the thread actually has pending interrupts (both in the protocol service and the embed protocol).

libs/langgraph-api/src/experimental/embed/protocol.mts

@@ -243,16 +243,17 @@ export function registerProtocolRoutes(
         : undefined;
     })();
 
-    const currentRun = thread.currentRun;
-    const currentStatus = currentRun?.status;
     const hasPendingInterrupts =
       params.input != null
         ? await hasPendingInterruptsForThread(thread)
         : false;
-    const isResume =
-      params.input != null &&
-      ((currentRun != null && currentStatus === "interrupted") ||
-        hasPendingInterrupts);
+    // Resume only when the thread actually has a pending `interrupt()`.
+    // A cancelled run also ends with status "interrupted" but has no
+    // pending interrupt to consume the resume value, so folding the
+    // input into `Command(resume)` there would silently drop the user's
+    // message; a cancelled thread must get plain input (a fresh run from
+    // the checkpoint that applies the input to state).
+    const isResume = params.input != null && hasPendingInterrupts;
 
     if (isResume) {
       // Drop stale lifecycle events from the paused run so late-attaching

libs/langgraph-api/src/protocol/service.mts

@@ -452,23 +452,17 @@ export class ProtocolService {
     params: NormalizedRunStart
   ) {
     const assistantId = getAssistantId(params.assistant_id);
-    const currentRun =
-      record.currentRunId != null
-        ? await this.bindings.runs.get(
-            record.currentRunId,
-            record.threadId,
-            record.auth
-          )
-        : null;
-    const currentStatus = currentRun?.status;
     const hasPendingInterrupts =
       params.input != null
         ? await this.hasPendingInterruptsForThread(record.threadId, record.auth)
         : false;
-    const isResume =
-      params.input != null &&
-      ((currentRun != null && currentStatus === "interrupted") ||
-        hasPendingInterrupts);
+    // Resume only when the thread actually has a pending `interrupt()`.
+    // A cancelled run also ends with status "interrupted" but has no
+    // pending interrupt to consume the resume value, so folding the
+    // input into `Command(resume)` there would silently drop the user's
+    // message; a cancelled thread must get plain input (a fresh run from
+    // the checkpoint that applies the input to state).
+    const isResume = params.input != null && hasPendingInterrupts;
 
     /**
      * Fork/time-travel replays from `config.configurable.checkpoint_id`

libs/langgraph-api/tests/protocol-v2/run-start-cancelled.test.mts

@@ -0,0 +1,106 @@
+/**
+ * `run.start` with input must fold the input into `Command(resume)` only
+ * when the thread actually has a pending `interrupt()`.
+ *
+ * A cancelled run also ends with status "interrupted", but there is no
+ * pending interrupt to consume the resume value — LangGraph only delivers
+ * resume values to an `interrupt()` call. Treating that submit as a resume
+ * therefore silently drops the user's message and re-runs the interrupted
+ * node with stale state. The stop-then-send flow (SDK `stop()` cancels the
+ * run, then the user sends a follow-up message) hits exactly this.
+ */
+import { describe, expect, it } from "vitest";
+
+import { ProtocolService } from "../../src/protocol/service.mjs";
+import type { Run, RunKwargs } from "../../src/storage/types.mjs";
+
+const THREAD_ID = "00000000-0000-7000-8000-000000000002";
+
+const interruptedRun = {
+  run_id: "00000000-0000-7000-8000-000000000001",
+  thread_id: THREAD_ID,
+  status: "interrupted",
+  kwargs: {},
+} as unknown as Run;
+
+const createService = (options: { pendingInterrupts: boolean }) => {
+  const puts: RunKwargs[] = [];
+  const bindings = {
+    runs: {
+      get: async () => interruptedRun,
+      put: async (runId: string, _assistantId: string, kwargs: RunKwargs) => {
+        puts.push(kwargs);
+        return [
+          {
+            ...interruptedRun,
+            run_id: runId,
+            status: "pending",
+            kwargs,
+          } as unknown as Run,
+        ];
+      },
+      stream: {
+        // eslint-disable-next-line @typescript-eslint/no-empty-function
+        join: () => (async function* () {})(),
+      },
+    },
+    threads: {
+      state: {
+        get: async () => ({
+          tasks: [
+            {
+              interrupts: options.pendingInterrupts
+                ? [{ id: "interrupt-1", value: { question: "proceed?" } }]
+                : [],
+            },
+          ],
+        }),
+      },
+    },
+  };
+  const service = new ProtocolService(
+    bindings as unknown as ConstructorParameters<typeof ProtocolService>[0]
+  );
+  const record = service.ensureThread({
+    threadId: THREAD_ID,
+    transport: "sse-http",
+  });
+  record.currentRunId = interruptedRun.run_id;
+  return { service, puts };
+};
+
+const input = {
+  messages: [{ type: "human", content: "narrow the scope to hand tools" }],
+};
+
+describe("run.start on a thread whose current run is interrupted", () => {
+  it("submits the input as input when the interruption was a cancel (no pending interrupt)", async () => {
+    const { service, puts } = createService({ pendingInterrupts: false });
+
+    const response = await service.handleCommand(THREAD_ID, {
+      id: 1,
+      method: "run.start",
+      params: { assistant_id: "agent", input },
+    });
+
+    expect(response).toMatchObject({ type: "success" });
+    expect(puts).toHaveLength(1);
+    expect(puts[0].input).toEqual(input);
+    expect(puts[0].command).toBeUndefined();
+  });
+
+  it("still folds the input into Command(resume) when an interrupt is pending", async () => {
+    const { service, puts } = createService({ pendingInterrupts: true });
+
+    const response = await service.handleCommand(THREAD_ID, {
+      id: 1,
+      method: "run.start",
+      params: { assistant_id: "agent", input },
+    });
+
+    expect(response).toMatchObject({ type: "success" });
+    expect(puts).toHaveLength(1);
+    expect(puts[0].input).toBeNull();
+    expect(puts[0].command).toEqual({ resume: input });
+  });
+});