feat(eve): cancellation - add active turn cancellation

.changeset/calm-turns-cancel.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Turn cancellation now propagates through active local and remote subagents, allowing parent turns to settle cleanly after descendant work is cancelled. Evals can start a cancellable turn with `startTurn()` and observe its settled result.

docs/concepts/sessions-runs-and-streaming.md

@@ -56,6 +56,7 @@ The stream is newline-delimited JSON (NDJSON), one event per line:
 | `authorization.completed` | A connection's authorization resolved; carries `outcome`.                                                        |
 | `step.completed`          | A model step finished; carries `finishReason` and usage.                                                         |
 | `step.failed`             | A model step failed; carries `{ code, message, details? }`.                                                      |
+| `turn.cancelled`          | The active turn was intentionally cancelled; followed by `session.waiting`.                                      |
 | `turn.completed`          | The turn finished.                                                                                               |
 | `turn.failed`             | The turn failed; carries `{ code, message, details? }`.                                                          |
 | `session.waiting`         | The session parked, waiting for the next input (a message, an answer).                                           |
@@ -86,6 +87,20 @@ The follow-up reuses the same durable session: same history, same state.
 
 For deterministic ordering, send one follow-up at a time and wait for the next `session.waiting` event before sending another message to the same session. See [message delivery and queueing](./execution-model-and-durability#message-delivery-and-queueing) for the current runtime contract.
 
+## Cancel the active turn
+
+Use the current continuation token to cancel active work without ending the session:
+
+```bash
+curl -X POST http://127.0.0.1:3000/eve/v1/session/<sessionId>/cancel \
+  -H 'content-type: application/json' \
+  -d '{"scope":"turn","continuationToken":"<token>"}'
+```
+
+Accepted cancellation returns `202`. A stale token or a continuation with no active turn returns
+`409`. The stream emits `turn.cancelled` followed by `session.waiting`, and the same session can
+accept a later follow-up.
+
 ## Reconnect and rewind
 
 The stream is durable. Every event is recorded before a step completes, so the whole stream is replayable. Pass `startIndex` to reconnect by event count and pick up where you dropped off, or rewind to the start:

docs/guides/client/messages.mdx

@@ -127,6 +127,22 @@ await resumed.result();
 
 You can send `message`, `inputResponses`, and `clientContext` together when the resumed turn needs both a human answer and follow-up text.
 
+## Cancel an active turn
+
+Call `cancel()` on the response whose turn should stop:
+
+```ts
+const response = await session.send("Run the long analysis.");
+
+await response.cancel();
+const result = await response.result();
+```
+
+Cancellation stops server-side model, tool, and descendant work. The stream settles with
+`turn.cancelled` followed by `session.waiting`, so the same `ClientSession` can send a later
+follow-up. Aborting a request or stream with an `AbortSignal` only stops the local transport; it
+does not request server-side cancellation.
+
 ## Single-use responses
 
 `MessageResponse` is single-use. Either aggregate it:

e2e/fixtures/agent-cancellation/.gitignore

@@ -0,0 +1,11 @@
+node_modules
+.env*
+.eve
+.vercel
+.workflow-data
+.next
+.output
+.nitro
+dist
+.DS_Store
+*.tsbuildinfo

e2e/fixtures/agent-cancellation/.vercelignore

@@ -0,0 +1,7 @@
+.env*.local
+node_modules
+.eve
+.next
+.output
+.nitro
+dist

e2e/fixtures/agent-cancellation/agent/agent.ts

@@ -0,0 +1,20 @@
+import { defineAgent } from "eve";
+import { mockModel } from "eve/evals";
+
+const WAIT_TOOL_NAME = "wait-for-cancellation";
+
+export default defineAgent({
+  model: mockModel(({ lastUserMessage }) =>
+    lastUserMessage?.includes(WAIT_TOOL_NAME) === true
+      ? {
+          toolCalls: [
+            {
+              id: "call_wait_for_cancellation",
+              name: WAIT_TOOL_NAME,
+            },
+          ],
+        }
+      : `cancellation-follow-up-ok:${lastUserMessage ?? ""}`,
+  ),
+  modelContextWindowTokens: 100_000,
+});

e2e/fixtures/agent-cancellation/agent/instructions.md

@@ -0,0 +1 @@
+You are a deterministic cancellation test agent.

e2e/fixtures/agent-cancellation/agent/tools/wait-for-cancellation.ts

@@ -0,0 +1,21 @@
+import { defineTool } from "eve/tools";
+import { z } from "zod";
+
+export default defineTool({
+  description: "Blocks until the active turn is cancelled.",
+  inputSchema: z.object({}),
+  async execute(_input, { abortSignal }) {
+    await new Promise<never>((_resolve, reject) => {
+      const onAbort = () => {
+        reject(abortSignal.reason ?? new Error("Turn cancelled."));
+      };
+
+      if (abortSignal.aborted) {
+        onAbort();
+        return;
+      }
+
+      abortSignal.addEventListener("abort", onAbort, { once: true });
+    });
+  },
+});

e2e/fixtures/agent-cancellation/evals/evals.config.ts

@@ -0,0 +1,6 @@
+import { defineEvalConfig } from "eve/evals";
+
+export default defineEvalConfig({
+  maxConcurrency: 1,
+  timeoutMs: 120_000,
+});

e2e/fixtures/agent-cancellation/evals/turn-cancellation.eval.ts

@@ -0,0 +1,26 @@
+import { defineEval } from "eve/evals";
+import { equals } from "eve/evals/expect";
+
+const WAIT_TOOL_NAME = "wait-for-cancellation";
+
+export default defineEval({
+  description:
+    "Turn cancellation aborts active work and settles the session at a waiting boundary.",
+  tags: ["cancellation", "workflow"],
+
+  async test(t) {
+    const activeTurn = await t.startTurn(
+      `Call ${WAIT_TOOL_NAME} and wait for it to finish before replying.`,
+    );
+
+    await t.sleep(5_000);
+    await activeTurn.cancel();
+
+    const cancelledTurn = await activeTurn.result();
+    await t.require(cancelledTurn.status, equals("waiting"));
+    cancelledTurn.calledTool(WAIT_TOOL_NAME, { count: 1, status: "pending" });
+    cancelledTurn.eventOrder([{ type: "turn.cancelled" }, { type: "session.waiting" }]);
+    cancelledTurn.notEvent("turn.failed");
+    cancelledTurn.notEvent("session.failed");
+  },
+});

e2e/fixtures/agent-cancellation/package.json

@@ -0,0 +1,24 @@
+{
+  "name": "agent-cancellation",
+  "version": "0.0.0",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "build": "eve build",
+    "dev": "eve dev",
+    "start": "eve start",
+    "typecheck": "eve build && tsc",
+    "test:e2e": "eve eval --strict"
+  },
+  "dependencies": {
+    "@opentelemetry/core": "2.6.1",
+    "@opentelemetry/sdk-trace-base": "2.6.1",
+    "@vercel/otel": "2.1.2",
+    "eve": "workspace:*",
+    "zod": "catalog:"
+  },
+  "devDependencies": {
+    "@types/node": "catalog:",
+    "typescript": "catalog:"
+  }
+}

e2e/fixtures/agent-cancellation/tsconfig.json

@@ -0,0 +1,17 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "outDir": "dist",
+    "rootDir": ".",
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "forceConsistentCasingInFileNames": true,
+    "declaration": true,
+    "noEmit": true,
+    "types": ["node"]
+  },
+  "include": ["agent/**/*.ts", "evals/**/*.ts", ".eve/**/*.d.ts"]
+}

packages/eve/src/channel/cross-channel-receive.test.ts

@@ -10,6 +10,7 @@ import type { Runtime } from "#channel/types.js";
 
 function makeRuntime(): Runtime {
   return {
+    cancelTurn: vi.fn(),
     deliver: vi.fn(),
     getEventStream: vi.fn(),
     run: vi.fn(),

packages/eve/src/channel/routes.ts

@@ -12,12 +12,14 @@ type WebSocketHeaders = Headers | readonly (readonly [string, string])[] | Recor
 
 /**
  * Second argument passed to every route handler. `send` starts or continues a
- * session on this channel; `getSession` looks one up by id; `receive` hands
- * inbound work to a different channel; `params` contains the matched path
- * parameters; `waitUntil` keeps background work alive past the response;
- * `requestIp` is the client IP, or `null` when the host cannot provide it.
+ * session on this channel; `cancelTurn` cancels its active turn; `getSession`
+ * looks one up by id; `receive` hands inbound work to a different channel;
+ * `params` contains the matched path parameters; `waitUntil` keeps background
+ * work alive past the response; `requestIp` is the client IP, or `null` when
+ * the host cannot provide it.
  */
 export interface RouteHandlerArgs<TState = undefined> {
+  cancelTurn: CancelTurnFn;
   send: SendFn<TState>;
   getSession: GetSessionFn;
   /**
@@ -50,6 +52,9 @@ export interface SendPayload {
   readonly outputSchema?: JsonObject;
 }
 
+/** Cancels the active turn addressed by a channel-local continuation token. */
+export type CancelTurnFn = (continuationToken: string) => Promise<void>;
+
 /**
  * Starts or continues a session on this channel. Accepts a plain string,
  * `UserContent`, or a {@link SendPayload}, plus {@link SendOptions} (auth,

packages/eve/src/channel/schedule.test.ts

@@ -23,6 +23,7 @@ function createMockRunHandle(): RunHandle {
 
 function createMockRuntime(): Runtime {
   return {
+    cancelTurn: vi.fn(),
     deliver: vi.fn().mockRejectedValue(new Error("no parked session")),
     run: vi.fn().mockResolvedValue(createMockRunHandle()),
     getEventStream: vi.fn().mockResolvedValue(new ReadableStream<HandleMessageStreamEvent>()),

packages/eve/src/channel/send.test.ts

@@ -16,6 +16,7 @@ function createMockRunHandle(): RunHandle {
 
 function createRuntime(deliverError: unknown): Runtime {
   return {
+    cancelTurn: vi.fn(),
     deliver: vi.fn().mockRejectedValue(deliverError),
     run: vi.fn().mockResolvedValue(createMockRunHandle()),
     getEventStream: vi.fn().mockResolvedValue(new ReadableStream<HandleMessageStreamEvent>()),
@@ -84,6 +85,7 @@ describe("createSendFn", () => {
   it("forwards context through deliver and run payloads", async () => {
     const context = ["thread background"];
     const deliverRuntime: Runtime = {
+      cancelTurn: vi.fn(),
       deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }),
       run: vi.fn().mockResolvedValue(createMockRunHandle()),
       getEventStream: vi.fn().mockResolvedValue(new ReadableStream<HandleMessageStreamEvent>()),
@@ -115,6 +117,7 @@ describe("createSendFn", () => {
 
   it("adds channel request ids to deliver and run inputs when provided", async () => {
     const deliverRuntime: Runtime = {
+      cancelTurn: vi.fn(),
       deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }),
       run: vi.fn().mockResolvedValue(createMockRunHandle()),
       getEventStream: vi.fn().mockResolvedValue(new ReadableStream<HandleMessageStreamEvent>()),
@@ -141,6 +144,7 @@ describe("createSendFn", () => {
       type: "object",
     } as const;
     const deliverRuntime: Runtime = {
+      cancelTurn: vi.fn(),
       deliver: vi.fn().mockResolvedValue({ sessionId: "existing-session-id" }),
       run: vi.fn().mockResolvedValue(createMockRunHandle()),
       getEventStream: vi.fn().mockResolvedValue(new ReadableStream<HandleMessageStreamEvent>()),

packages/eve/src/channel/types.ts

@@ -318,6 +318,11 @@ export interface RunHandle {
  * Runtime interface consumed by routes and the subagent tool wrapper.
  */
 export interface Runtime {
+  /**
+   * Cancels the active turn addressed by a continuation token.
+   */
+  cancelTurn(continuationToken: string): Promise<void>;
+
   /**
    * Starts a new run from a flat platform-shape input.
    *

packages/eve/src/cli/dev/tui/runner.test.ts

@@ -55,6 +55,7 @@ function stubSession(): ClientSession {
 /** Wraps literal stream events in a real `MessageResponse`. */
 function messageResponseOf(events: readonly unknown[]): MessageResponse {
   return new MessageResponse({
+    cancelTurn: async () => undefined,
     continuationToken: "eve:test",
     createStream: async function* () {
       for (const event of events) yield event as HandleMessageStreamEvent;

packages/eve/src/client/message-response.ts

@@ -11,6 +11,7 @@ import type { MessageResult } from "#client/types.js";
  * Internal configuration passed to construct a {@link MessageResponse}.
  */
 interface MessageResponseInput {
+  readonly cancelTurn: () => Promise<void>;
   readonly continuationToken?: string;
   readonly createStream: () => AsyncGenerator<HandleMessageStreamEvent>;
   readonly sessionId: string;
@@ -35,15 +36,28 @@ export class MessageResponse<TOutput = unknown> implements AsyncIterable<HandleM
   readonly sessionId: string;
 
   #consumed = false;
+  readonly #cancelTurn: () => Promise<void>;
   readonly #createStream: () => AsyncGenerator<HandleMessageStreamEvent>;
 
   /** @internal */
   constructor(input: MessageResponseInput) {
+    this.#cancelTurn = input.cancelTurn;
     this.continuationToken = input.continuationToken;
     this.sessionId = input.sessionId;
     this.#createStream = input.createStream;
   }
 
+  /**
+   * Requests server-side cancellation of this response's active turn.
+   *
+   * This does not merely stop local stream consumption: the server aborts
+   * active model, tool, and descendant work and settles the session at
+   * `turn.cancelled` followed by `session.waiting`.
+   */
+  async cancel(): Promise<void> {
+    await this.#cancelTurn();
+  }
+
   /**
    * Consumes the full event stream and returns the aggregated
    * {@link MessageResult}.