skills/vercel__eve
1
0
Fork 0
mirror of https://github.com/vercel/eve.git synced 2026-09-20 05:35:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(eve)!: cancel durable turns from useEveAgent() (#1401)

Browse Source 
Signed-off-by: Timo Lins <me@timo.sh>
This commit is contained in:
Timo Lins
2026-08-14 18:35:15 +02:00
committed by GitHub
parent e306fc3422
commit 48c11057e2
23 changed files with 557 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/durable-agent-cancel.md
+5
View File
@@ -0,0 +1,5 @@
---
"eve": minor
---
Replace `stop()` on frontend agent bindings with `cancel()`. Cancellation now targets the exact durable turn through `MessageResponse.cancel()` while the binding stays attached through settlement.
apps/docs/registry/channel/web/app/_components/agent-chat.tsx
+8 -86
View File
@@ -1,10 +1,9 @@
"use client";
import type { UserContent } from "ai";
import { Client, type MessageStreamEvent } from "eve/client";
import { useEveAgent } from "eve/react";
import { AlertCircleIcon } from "lucide-react";
import { useCallback, useRef, useState } from "react";
import { useState } from "react";
import {
Conversation,
ConversationContent,
@@ -22,103 +21,26 @@ import { AgentMessage } from "./agent-message";
const AGENT_NAME = "eve-agent";
type AgentStatus = ReturnType<typeof useEveAgent>["status"];
type CancellationState = "idle" | "requested" | "cancelling";
type Cancellation = {
requested: boolean;
sentTurnId?: string;
turnId?: string;
};
export function AgentChat() {
const [client] = useState(() => new Client({ host: "" }));
const sessionIdRef = useRef<string | undefined>(undefined);
const cancellationRef = useRef<Cancellation>({ requested: false });
const [cancellationError, setCancellationError] = useState<string>();
const [cancellationState, setCancellationState] = useState<CancellationState>("idle");
const cancelTurn = useCallback(
(turnId: string) => {
const cancellation = cancellationRef.current;
if (!cancellation.requested || cancellation.sentTurnId === turnId) {
return;
}
cancellation.sentTurnId = turnId;
setCancellationState("cancelling");
const sessionId = sessionIdRef.current;
if (sessionId === undefined) {
return;
}
void client.sessions
.attach(sessionId)
.cancel({ turnId })
.catch((error: unknown) => {
if (cancellationRef.current !== cancellation) {
return;
}
cancellation.requested = false;
cancellation.sentTurnId = undefined;
setCancellationError(toErrorMessage(error));
setCancellationState("idle");
});
},
[client],
);
const handleEvent = useCallback(
(event: MessageStreamEvent) => {
if (event.type !== "turn.started") {
return;
}
const cancellation = cancellationRef.current;
cancellation.turnId = event.data.turnId;
cancelTurn(event.data.turnId);
},
[cancelTurn],
);
const agent = useEveAgent({
onEvent: handleEvent,
onSessionChange(session) {
sessionIdRef.current = session?.sessionId;
},
});
const agent = useEveAgent();
const isBusy = agent.status === "submitted" || agent.status === "streaming";
const isEmpty = agent.data.messages.length === 0;
const errorMessage = cancellationError ?? agent.error?.message;
const submitStatus = isBusy && cancellationState !== "idle" ? "submitted" : agent.status;
const prepareTurn = () => {
cancellationRef.current = { requested: false };
setCancellationError(undefined);
setCancellationState("idle");
};
const requestCancellation = () => {
if (!isBusy || cancellationState !== "idle") {
return;
}
const cancellation = cancellationRef.current;
cancellation.requested = true;
setCancellationError(undefined);
setCancellationState("requested");
if (cancellation.turnId !== undefined) {
cancelTurn(cancellation.turnId);
}
void agent.cancel().catch((error: unknown) => {
setCancellationError(toErrorMessage(error));
});
};
const handleSubmit = async (message: PromptInputMessage) => {
const text = message.text.trim();
if ((text.length === 0 && message.files.length === 0) || isBusy) return;
prepareTurn();
setCancellationError(undefined);
if (message.files.length === 0) {
await agent.send(text);
@@ -144,7 +66,7 @@ export function AgentChat() {
const composer = (
<PromptInput onSubmit={handleSubmit}>
<PromptInputTextarea placeholder="Send a message…" />
<PromptInputSubmit onStop={requestCancellation} status={submitStatus} />
<PromptInputSubmit onStop={requestCancellation} status={agent.status} />
</PromptInput>
);
@@ -183,7 +105,7 @@ export function AgentChat() {
key={message.id}
message={message}
onInputResponses={(inputResponses) => {
prepareTurn();
setCancellationError(undefined);
return agent.respond(inputResponses);
}}
/>
apps/frameworks/nuxt/app/components/EveAgentConsole.vue
+18 -4
View File
@@ -1,12 +1,14 @@
<script setup lang="ts">
import type { EveDynamicToolPart, EveMessagePart } from "eve/vue";
const { data, status, error, respond, send, stop } = useEveAgent();
const { cancel, data, status, error, respond, send } = useEveAgent();
type EveFilePart = Extract<EveMessagePart, { type: "file" }>;
const isBusy = computed(() => status.value === "submitted" || status.value === "streaming");
const isEmpty = computed(() => data.value.messages.length === 0);
const cancellationError = ref<string>();
const errorMessage = computed(() => cancellationError.value ?? error.value?.message);
const messagesEl = useTemplateRef("messagesEl");
@@ -36,10 +38,21 @@ const messageText = ref("");
function submitMessage() {
const text = messageText.value.trim();
if (!text || isBusy.value) return;
cancellationError.value = undefined;
messageText.value = "";
void send(text);
}
async function requestCancellation() {
cancellationError.value = undefined;
try {
await cancel();
} catch (cause) {
cancellationError.value =
cause instanceof Error ? cause.message : "The cancellation request failed.";
}
}
function onKeydown(e: KeyboardEvent) {
if (e.key === "Enter" && !e.shiftKey) {
e.preventDefault();
@@ -54,6 +67,7 @@ function handleInputResponses(
readonly text?: string;
}[],
) {
cancellationError.value = undefined;
void respond(responses);
}
@@ -97,13 +111,13 @@ function formatBytes(size: number | undefined): string | undefined {
<section class="mx-auto flex min-h-0 w-full max-w-3xl flex-1 flex-col px-4 sm:px-6">
<div
v-if="error"
v-if="errorMessage"
class="mt-4 flex items-start gap-3 rounded-md border border-destructive/30 bg-destructive/5 px-3 py-2.5 text-sm"
>
<div>
<p class="font-medium">Request failed</p>
<p class="mt-0.5 text-muted-foreground">
{{ error.message }}
{{ errorMessage }}
</p>
</div>
</div>
@@ -253,7 +267,7 @@ function formatBytes(size: number | undefined): string | undefined {
v-if="isBusy"
type="button"
class="flex size-8 shrink-0 items-center justify-center rounded-md bg-primary text-primary-foreground hover:bg-primary/90"
@click="stop()"
@click="requestCancellation"
>
<svg class="size-3.5" fill="currentColor" viewBox="0 0 24 24">
<rect x="4" y="4" width="16" height="16" rx="2" />
apps/frameworks/sveltekit/src/lib/EveAgentConsole.svelte
+17 -3
View File
@@ -12,6 +12,8 @@
let isBusy = $derived(agent.status === "submitted" || agent.status === "streaming");
let isEmpty = $derived(agent.data.messages.length === 0);
let cancellationError = $state<string>();
let errorMessage = $derived(cancellationError ?? agent.error?.message);
let messagesEl = $state<HTMLDivElement>();
let isNearBottom = $state(true);
@@ -44,10 +46,21 @@
function submitMessage() {
const text = messageText.trim();
if (!text || isBusy) return;
cancellationError = undefined;
messageText = "";
void agent.send(text);
}
async function requestCancellation() {
cancellationError = undefined;
try {
await agent.cancel();
} catch (cause) {
cancellationError =
cause instanceof Error ? cause.message : "The cancellation request failed.";
}
}
function onKeydown(e: KeyboardEvent) {
if (e.key === "Enter" && !e.shiftKey) {
e.preventDefault();
@@ -62,6 +75,7 @@
readonly text?: string;
}[],
) {
cancellationError = undefined;
void agent.respond(responses);
}
@@ -103,14 +117,14 @@
</header>
<section class="mx-auto flex min-h-0 w-full max-w-3xl flex-1 flex-col px-4 sm:px-6">
{#if agent.error}
{#if errorMessage}
<div
class="mt-4 flex items-start gap-3 rounded-md border border-destructive/30 bg-destructive/5 px-3 py-2.5 text-sm"
>
<div>
<p class="font-medium">Request failed</p>
<p class="mt-0.5 text-muted-foreground">
{agent.error.message}
{errorMessage}
</p>
</div>
</div>
@@ -254,7 +268,7 @@
type="button"
aria-label="Stop response"
class="flex size-8 shrink-0 items-center justify-center rounded-md bg-primary text-primary-foreground hover:bg-primary/90"
onclick={() => agent.stop()}
onclick={requestCancellation}
>
<svg class="size-3.5" fill="currentColor" viewBox="0 0 24 24">
<rect x="4" y="4" width="16" height="16" rx="2" />
docs/guides/client/streaming.mdx
+6 -5
View File
@@ -5,23 +5,24 @@ description: "Consume eve client stream events live, reconnect by event index, a
Every `ClientSession.send()` call posts the turn, then reads the session's NDJSON (newline-delimited JSON) event stream. `MessageResponse` gives you two ways to consume that stream, aggregating it with `result()` or iterating it live.
Once `send()` is accepted, the session exposes its assigned `sessionId` and can request cooperative cancellation with `session.cancel()`, even while the response stream is still running. The result status is `accepted` when the live session durably queued the command, including when a parked session consumes it as a no-op. An unknown or terminal session returns `no_active_turn`:
Once `send()` is accepted, `response.cancel()` requests cooperative cancellation of that exact turn. Start consuming the response first; cancellation waits for the stream to identify the turn, guards the request with its ID, and never targets a later turn. The result status is `accepted` when the live session durably queues the command. A response that settles before a turn starts returns `no_active_turn`:
```ts
const { session, response } = await client.sessions.create({ message: "Run the long operation." });
const { response } = await client.sessions.create({ message: "Run the long operation." });
const cancellation = await session.cancel();
const resultPromise = response.result();
const cancellation = await response.cancel();
if (cancellation.status === "accepted") {
console.log(cancellation.sessionId);
}
const result = await response.result();
const result = await resultPromise;
```
The cancellation result is discriminated by `status`: only `accepted` includes
`sessionId`; `no_active_turn` has no session identity field.
Cancellation does not replace stream consumption. Continue reading the response to observe its terminal `turn.cancelled` and `session.waiting` boundary and to advance the client session cursor normally.
Cancellation does not replace stream consumption. Continue reading the response to observe its terminal `turn.cancelled` and `session.waiting` boundary and to advance the client session cursor normally. Use `session.cancel({ turnId })` instead when you have only a fixed session handle and an observed turn ID.
Between turns, `session.compact()` queues context compaction without sending model input. An accepted request reports the session id; consume the durable stream through the following `session.waiting` boundary before sending the next turn. `compaction.completed` confirms that summarization succeeded; without it, eve preserves the previous history. A never-started session returns `no_active_session` as a successful no-op.
docs/guides/frontend/overview.mdx
+10 -8
View File
@@ -78,7 +78,7 @@ export function Chat() {
| `session` | Serializable fixed session cursor (`sessionId`, `streamIndex`). |
| `send` | Send text or a multi-part message, with per-turn options. |
| `respond` | Answer pending HITL input requests, with per-turn options. |
| `stop` | Abort the client's in-flight stream. The server-side turn keeps running. |
| `cancel` | Request durable cancellation of the active turn. |
| `reset` | Clear local events, data, errors, and the local session cursor. |
Most chat UIs only need `data.messages` and `status`. Drop down to `events` when you need the authoritative wire events directly, for example to persist an audit log or build a custom projection.
@@ -105,15 +105,17 @@ await agent.send([
]);
```
Assistant text, reasoning, tool calls, and tool results stream into `data` as they arrive, and `status` moves from `ready` to `submitted` to `streaming` and back. Call `stop()` to abort the client's in-flight stream, and `reset()` to clear local state so the next send starts a fresh durable session.
Assistant text, reasoning, tool calls, and tool results stream into `data` as they arrive, and `status` moves from `ready` to `submitted` to `streaming` and back. Call `cancel()` to stop the durable server-side turn, and `reset()` to clear local state so the next send starts a fresh durable session.
`stop()` is local: turns are resumable across disconnects, so detaching the stream never cancels the work — the turn keeps running and billing on the server. To stop the turn itself, POST the session's [cancel route](../../channels/eve#routes); the turn then settles on the stream as `turn.cancelled` followed by `session.waiting`.
`cancel()` can be called as soon as `status` is `"submitted"`; the hook waits for the active response to identify its turn when necessary, sends one guarded cancellation request, and keeps the event stream attached. The promise resolves when eve accepts the request or reports that no turn is active, and rejects if the cancellation request fails. The turn then settles on the same stream as `turn.cancelled` followed by `session.waiting`, so the session is safe to continue.
A chat UI with a server-side Cancel button can use the `sessionId` exposed by the
hook, attach a fixed `ClientSession`, and call `session.cancel({ turnId })` after
observing that turn's `turn.started` event. The `turnId` guard makes a late click
a harmless no-op instead of cancelling a newer turn. Keep the stream open until
the cancellation boundary arrives; do not use `stop()` for this interaction.
```tsx
if (agent.status === "submitted" || agent.status === "streaming") {
await agent.cancel();
}
```
Unmounting the component or closing the page disconnects the local stream but does not cancel server execution. Call `cancel()` before detaching when the user intends to stop the durable turn.
After eve confirms an attachment turn with `message.received`, the default reducer projects each
received attachment as a `file` part on the user message. The part includes `mediaType`, optional
docs/guides/frontend/use-eve-agent-svelte.mdx
+10 -10
View File
@@ -44,14 +44,14 @@ Call the binding once for a conversation and read its reactive getters directly.
The state fields are reactive getters; the commands are ordinary methods:
| Property | Svelte shape |
| ---------------------------------- | --------------------------------- |
| `data` | `TData` |
| `status` | `UseEveAgentStatus` |
| `error` | `Error \| undefined` |
| `events` | `readonly MessageStreamEvent[]` |
| `session` | `ClientSessionState \| undefined` |
| `send`, `respond`, `stop`, `reset` | Methods |
| Property | Svelte shape |
| ------------------------------------ | --------------------------------- |
| `data` | `TData` |
| `status` | `UseEveAgentStatus` |
| `error` | `Error \| undefined` |
| `events` | `readonly MessageStreamEvent[]` |
| `session` | `ClientSessionState \| undefined` |
| `send`, `respond`, `cancel`, `reset` | Methods |
Read the getters directly in templates, `$derived`, or `$effect`. The [shared returned-state reference](./overview#returned-state) describes what each value and command does.
@@ -104,9 +104,9 @@ Pending requests appear in `agent.data.messages`. Use the Svelte exports when na
See [Human-in-the-loop prompts](./overview#human-in-the-loop-prompts) for request semantics and rendering guidance.
## Stop, reset, and resume
## Cancel, reset, and resume
Call `agent.stop()` to detach the browser stream without cancelling server-side work, and `agent.reset()` to clear local state and start a new session. Pass `initialSession` and `initialEvents` to restore a saved conversation. See [Resumable sessions](./overview#resumable-sessions) for the persistence contract and [Sending and streaming](./overview#sending-and-streaming) for cancellation behavior.
Call `agent.cancel()` to stop the durable server-side turn while the binding remains attached through settlement. Destroying the component only disconnects its local stream; it does not cancel server execution. Call `agent.reset()` to clear local state and start a new session. Pass `initialSession` and `initialEvents` to restore a saved conversation. See [Resumable sessions](./overview#resumable-sessions) for the persistence contract and [Sending and streaming](./overview#sending-and-streaming) for cancellation behavior.
## Custom host and credentials
docs/guides/frontend/use-eve-agent-vue.mdx
+10 -10
View File
@@ -43,14 +43,14 @@ async function handleSubmit() {
The state fields are computed refs; the commands are ordinary methods:
| Property | Vue shape |
| ---------------------------------- | ---------------------------------------------- |
| `data` | `ComputedRef<TData>` |
| `status` | `ComputedRef<UseEveAgentStatus>` |
| `error` | `ComputedRef<Error \| undefined>` |
| `events` | `ComputedRef<readonly MessageStreamEvent[]>` |
| `session` | `ComputedRef<ClientSessionState \| undefined>` |
| `send`, `respond`, `stop`, `reset` | Methods |
| Property | Vue shape |
| ------------------------------------ | ---------------------------------------------- |
| `data` | `ComputedRef<TData>` |
| `status` | `ComputedRef<UseEveAgentStatus>` |
| `error` | `ComputedRef<Error \| undefined>` |
| `events` | `ComputedRef<readonly MessageStreamEvent[]>` |
| `session` | `ComputedRef<ClientSessionState \| undefined>` |
| `send`, `respond`, `cancel`, `reset` | Methods |
Destructuring preserves reactivity because each state value remains a ref. Read refs with `.value` in `<script>` and without `.value` in a template. The [shared returned-state reference](./overview#returned-state) describes what each value and command does.
@@ -106,9 +106,9 @@ const pendingRequests = computed(() =>
See [Human-in-the-loop prompts](./overview#human-in-the-loop-prompts) for request semantics and rendering guidance.
## Stop, reset, and resume
## Cancel, reset, and resume
Call `stop()` to detach the browser stream without cancelling server-side work, and `reset()` to clear local state and start a new session. Pass `initialSession` and `initialEvents` to restore a saved conversation. See [Resumable sessions](./overview#resumable-sessions) for the persistence contract and [Sending and streaming](./overview#sending-and-streaming) for cancellation behavior.
Call `cancel()` to stop the durable server-side turn while the composable remains attached through settlement. Disposing the component only disconnects its local stream; it does not cancel server execution. Call `reset()` to clear local state and start a new session. Pass `initialSession` and `initialEvents` to restore a saved conversation. See [Resumable sessions](./overview#resumable-sessions) for the persistence contract and [Sending and streaming](./overview#sending-and-streaming) for cancellation behavior.
## Custom host and credentials
packages/eve/src/cli/dev/tui/runner.test.ts
+3
View File
@@ -73,6 +73,7 @@ function messageResponseOf(events: readonly unknown[]): MessageResponse {
isStamped(event) ? event : stampTestEvent(event as UnstampedMessageStreamEvent, index),
);
return new MessageResponse({
cancelTurn: async () => ({ status: "no_active_turn" }),
createStream: async function* () {
for (const event of stamped) yield event;
},
@@ -3610,6 +3611,7 @@ describe("EveTUIRunner mid-turn message queue", () => {
vi.spyOn(session, "send").mockImplementation(
async () =>
new MessageResponse({
cancelTurn: async (turnId) => await session.cancel({ turnId }),
createStream: async function* () {
yield stampTestEvent(
{
@@ -3733,6 +3735,7 @@ describe("EveTUIRunner session id reporting", () => {
vi.spyOn(session, "send").mockImplementation(
async () =>
new MessageResponse({
cancelTurn: async (turnId) => await session.cancel({ turnId }),
createStream: async function* () {
yield { type: "turn.started", data: { turnId: "turn-1", sequence: 1 } } as never;
await gate.promise;
packages/eve/src/client/eve-agent-store.test.ts
+122 -1
View File
@@ -1,12 +1,14 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { EveAgentStore } from "#client/eve-agent-store.js";
import { detachEveAgentStore, EveAgentStore } from "#client/eve-agent-store.js";
import { defaultMessageReducer } from "#client/message-reducer.js";
import { stampTestEvents } from "#internal/testing/events.js";
import {
createMessageCompletedEvent,
createMessageReceivedEvent,
createSessionWaitingEvent,
createTurnCancelledEvent,
createTurnStartedEvent,
EVE_SESSION_ID_HEADER,
type UnstampedMessageStreamEvent,
type MessageStreamEvent,
@@ -47,6 +49,34 @@ function streamResponse(events: readonly MessageStreamEvent[]): Response {
);
}
function controlledStreamResponse() {
const encoder = new TextEncoder();
let controller: ReadableStreamDefaultController<Uint8Array> | undefined;
const response = new Response(
new ReadableStream<Uint8Array>({
start(nextController) {
controller = nextController;
},
}),
);
return {
close: () => controller?.close(),
emit: (event: MessageStreamEvent) => {
controller?.enqueue(encoder.encode(`${JSON.stringify(event)}\n`));
},
response,
};
}
function acceptedCancellationResponse(): Response {
return Response.json({
ok: true,
sessionId: "session_1",
status: "accepted",
});
}
function preV20MessageCompletedEvent(): MessageStreamEvent {
return {
...createMessageCompletedEvent({
@@ -171,3 +201,94 @@ describe("EveAgentStore stream overlap", () => {
);
});
});
describe("EveAgentStore cancellation", () => {
it("queues cancellation until the turn id arrives and keeps streaming", async () => {
const stream = controlledStreamResponse();
const [turnStarted, turnCancelled, boundary] = stampTestEvents([
createTurnStartedEvent({ sequence: 0, turnId: "turn_1" }),
createTurnCancelledEvent({ sequence: 1, turnId: "turn_1" }),
createSessionWaitingEvent(),
] as UnstampedMessageStreamEvent[]);
const fetchMock = vi
.spyOn(globalThis, "fetch")
.mockResolvedValueOnce(startedResponse())
.mockResolvedValueOnce(stream.response)
.mockResolvedValueOnce(acceptedCancellationResponse());
const store = new EveAgentStore({ optimistic: false, reducer: defaultMessageReducer() });
const sending = store.send({ message: "Hello" });
const cancellation = store.cancel();
const duplicateCancellation = store.cancel();
await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2));
stream.emit(turnStarted!);
await expect(cancellation).resolves.toEqual({
sessionId: "session_1",
status: "accepted",
});
await expect(duplicateCancellation).resolves.toEqual({
sessionId: "session_1",
status: "accepted",
});
expect(fetchMock).toHaveBeenCalledTimes(3);
expect(fetchMock.mock.calls[2]?.[0]).toBe("/eve/v1/session/session_1/cancel");
expect(JSON.parse(String(fetchMock.mock.calls[2]?.[1]?.body))).toEqual({
turnId: "turn_1",
});
expect(store.snapshot.status).toBe("streaming");
stream.emit(turnCancelled!);
stream.emit(boundary!);
stream.close();
await sending;
expect(store.snapshot.status).toBe("ready");
expect(store.snapshot.events).toEqual([turnStarted, turnCancelled, boundary]);
});
it("returns no_active_turn when idle", async () => {
const store = new EveAgentStore({ reducer: defaultMessageReducer() });
await expect(store.cancel()).resolves.toEqual({ status: "no_active_turn" });
});
it("resolves a queued cancellation when reset wins before dispatch", async () => {
const fetchMock = vi.spyOn(globalThis, "fetch");
const store = new EveAgentStore({ reducer: defaultMessageReducer() });
const sending = store.send({ message: "Hello" });
const cancellation = store.cancel();
store.reset();
await expect(cancellation).resolves.toEqual({ status: "no_active_turn" });
await sending;
expect(fetchMock).not.toHaveBeenCalled();
expect(store.snapshot.status).toBe("ready");
});
it("detaches local transport without cancelling durable server work", async () => {
let signal: AbortSignal | undefined;
const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementationOnce((_input, init) => {
signal = init?.signal ?? undefined;
return new Promise<Response>((_resolve, reject) => {
signal?.addEventListener(
"abort",
() => reject(new DOMException("The operation was aborted.", "AbortError")),
{ once: true },
);
});
});
const store = new EveAgentStore({ reducer: defaultMessageReducer() });
const sending = store.send({ message: "Hello" });
await vi.waitFor(() => expect(signal).toBeDefined());
detachEveAgentStore(store);
await sending;
expect(signal?.aborted).toBe(true);
expect(fetchMock).toHaveBeenCalledOnce();
expect(store.snapshot.status).toBe("ready");
});
});
packages/eve/src/client/eve-agent-store.ts
+58 -29
View File
@@ -1,10 +1,12 @@
import { Client } from "#client/client.js";
import type { MessageResponse } from "#client/message-response.js";
import type { EveAgentReducer, EveAgentReducerEvent } from "#client/reducer.js";
import type { ClientSession } from "#client/session.js";
import { createEventDeduper } from "#protocol/event-dedupe.js";
import type { MessageStreamEvent } from "#protocol/message.js";
import { toError } from "#shared/errors.js";
import type {
CancelSessionResult,
ClientAuth,
HeadersValue,
SendTurnPayload,
@@ -88,6 +90,14 @@ interface PendingMessageSubmission {
readonly message: string;
}
const detachStore = Symbol("detachEveAgentStore");
interface ActiveTurn {
readonly abortController: AbortController;
readonly response: Promise<MessageResponse | undefined>;
readonly resolveResponse: (response: MessageResponse | undefined) => void;
}
/**
* Framework-agnostic state machine for an eve agent session.
*
@@ -98,7 +108,8 @@ interface PendingMessageSubmission {
* Drives one turn at a time: `send` rejects if a turn is already submitted or
* streaming. Read the latest projection via the `snapshot` getter, observe
* changes with `subscribe`, register lifecycle hooks with `setCallbacks`,
* abort the in-flight turn with `stop`, and discard all state with `reset`.
* cancel the durable in-flight turn with `cancel`, and discard all state with
* `reset`.
*/
export class EveAgentStore<TData> {
readonly #client: Client | undefined;
@@ -110,12 +121,11 @@ export class EveAgentStore<TData> {
/** Ids already folded into the projection: `initialEvents` and a reconnect can overlap. */
#seenEvents = createEventDeduper();
#abortController: AbortController | undefined;
#activeTurn: ActiveTurn | undefined;
#callbacks: EveAgentStoreCallbacks<TData> = {};
#data: TData;
#error: Error | undefined;
#events: readonly MessageStreamEvent[];
#operationId = 0;
#pendingMessageSubmission: PendingMessageSubmission | undefined;
#projectionEvents: readonly EveAgentReducerEvent[];
#session: ClientSession | undefined;
@@ -173,9 +183,13 @@ export class EveAgentStore<TData> {
throw new Error("eve session is already processing a turn.");
}
const operationId = this.#startOperation();
const abortController = new AbortController();
this.#abortController = abortController;
const response = Promise.withResolvers<MessageResponse | undefined>();
const turn: ActiveTurn = {
abortController: new AbortController(),
response: response.promise,
resolveResponse: response.resolve,
};
this.#activeTurn = turn;
this.#error = undefined;
this.#status = "submitted";
this.#publish();
@@ -184,7 +198,7 @@ export class EveAgentStore<TData> {
const preparedInput = (await this.#callbacks.prepareSend?.(input)) ?? input;
assertExclusiveTurnInput(preparedInput);
if (!this.#isCurrentOperation(operationId)) {
if (!this.#isActiveTurn(turn)) {
return;
}
@@ -194,13 +208,16 @@ export class EveAgentStore<TData> {
const turnInput = {
...preparedInput,
signal: createAbortSignal(preparedInput.signal, abortController.signal),
signal: createAbortSignal(preparedInput.signal, turn.abortController.signal),
};
const response = await this.#dispatchTurn(turnInput);
if (!this.#isActiveTurn(turn)) return;
turn.resolveResponse(response);
let sawEvent = false;
for await (const event of response) {
if (!this.#isCurrentOperation(operationId)) {
if (!this.#isActiveTurn(turn)) {
return;
}
@@ -220,13 +237,13 @@ export class EveAgentStore<TData> {
this.#publish();
}
if (!this.#isCurrentOperation(operationId)) {
if (!this.#isActiveTurn(turn)) {
return;
}
this.#status = this.#error === undefined ? "ready" : "error";
} catch (error) {
if (!this.#isCurrentOperation(operationId)) {
if (!this.#isActiveTurn(turn)) {
return;
}
@@ -240,8 +257,9 @@ export class EveAgentStore<TData> {
this.#callbacks.onError?.(this.#error);
}
} finally {
if (this.#isCurrentOperation(operationId)) {
this.#abortController = undefined;
if (this.#isActiveTurn(turn)) {
turn.resolveResponse(undefined);
this.#activeTurn = undefined;
this.#callbacks.onSessionChange?.(this.#session?.state);
this.#publish();
this.#callbacks.onFinish?.(this.#snapshot);
@@ -249,14 +267,29 @@ export class EveAgentStore<TData> {
}
}
stop(): void {
this.#abortController?.abort();
/**
* Requests cooperative cancellation of the active durable turn.
*
* If the server has not emitted `turn.started` yet, the request waits for
* that turn ID. The event stream stays attached until the turn settles.
*/
cancel(): Promise<CancelSessionResult> {
const turn = this.#activeTurn;
if (turn === undefined) return Promise.resolve({ status: "no_active_turn" });
return turn.response.then<CancelSessionResult>((response) =>
response === undefined ? { status: "no_active_turn" } : response.cancel(),
);
}
[detachStore](): void {
this.#activeTurn?.abortController.abort();
}
reset(): void {
this.#invalidateOperation();
this.stop();
this.#abortController = undefined;
const turn = this.#activeTurn;
this.#activeTurn = undefined;
turn?.resolveResponse(undefined);
turn?.abortController.abort();
if (!this.#externalSession) this.#session = undefined;
this.#events = [];
this.#seenEvents = createEventDeduper();
@@ -297,17 +330,8 @@ export class EveAgentStore<TData> {
return await this.#session.respond(inputResponses, options);
}
#startOperation(): number {
this.#operationId += 1;
return this.#operationId;
}
#invalidateOperation(): void {
this.#operationId += 1;
}
#isCurrentOperation(operationId: number): boolean {
return this.#operationId === operationId;
#isActiveTurn(turn: ActiveTurn): boolean {
return this.#activeTurn === turn;
}
#projectOptimisticMessage(input: SendTurnPayload): void {
@@ -452,6 +476,11 @@ export class EveAgentStore<TData> {
}
}
/** @internal Detaches local transport without cancelling durable server work. */
export function detachEveAgentStore<TData>(store: EveAgentStore<TData>): void {
store[detachStore]();
}
function assertExclusiveTurnInput(input: SendTurnPayload): void {
const hasMessage = input.message !== undefined;
const hasResponses = input.inputResponses !== undefined;
packages/eve/src/client/message-response.test.ts
+110
View File
@@ -0,0 +1,110 @@
import { describe, expect, it, vi } from "vitest";
import { MessageResponse } from "#client/message-response.js";
import { stampTestEvents } from "#internal/testing/events.js";
import {
createSessionWaitingEvent,
createTurnStartedEvent,
type MessageStreamEvent,
type UnstampedMessageStreamEvent,
} from "#protocol/message.js";
function createDeferred<T>() {
return Promise.withResolvers<T>();
}
async function consume(response: MessageResponse): Promise<MessageStreamEvent[]> {
const events: MessageStreamEvent[] = [];
for await (const event of response) events.push(event);
return events;
}
function acceptedCancellation() {
return { sessionId: "session_1", status: "accepted" as const };
}
describe("MessageResponse cancellation", () => {
it("queues one guarded cancellation until this response identifies its turn", async () => {
const start = createDeferred<void>();
const settle = createDeferred<void>();
const [turnStarted, boundary] = stampTestEvents([
createTurnStartedEvent({ sequence: 0, turnId: "turn_1" }),
createSessionWaitingEvent(),
] as UnstampedMessageStreamEvent[]);
const cancelTurn = vi.fn(async () => acceptedCancellation());
const response = new MessageResponse({
cancelTurn,
createStream: async function* () {
await start.promise;
yield turnStarted!;
await settle.promise;
yield boundary!;
},
sessionId: "session_1",
});
const consumed = consume(response);
const cancellation = response.cancel();
expect(response.cancel()).toBe(cancellation);
expect(cancelTurn).not.toHaveBeenCalled();
start.resolve();
await expect(cancellation).resolves.toEqual(acceptedCancellation());
expect(cancelTurn).toHaveBeenCalledOnce();
expect(cancelTurn).toHaveBeenCalledWith("turn_1");
settle.resolve();
await expect(consumed).resolves.toEqual([turnStarted, boundary]);
await expect(response.cancel()).resolves.toEqual({ status: "no_active_turn" });
});
it("allows a failed cancellation request to be retried for the same live turn", async () => {
const settle = createDeferred<void>();
const [turnStarted, boundary] = stampTestEvents([
createTurnStartedEvent({ sequence: 0, turnId: "turn_1" }),
createSessionWaitingEvent(),
] as UnstampedMessageStreamEvent[]);
const cancelTurn = vi
.fn<(turnId: string) => Promise<ReturnType<typeof acceptedCancellation>>>()
.mockRejectedValueOnce(new Error("Cancel unavailable"))
.mockResolvedValueOnce(acceptedCancellation());
const response = new MessageResponse({
cancelTurn,
createStream: async function* () {
yield turnStarted!;
await settle.promise;
yield boundary!;
},
sessionId: "session_1",
});
const consumed = consume(response);
await expect(response.cancel()).rejects.toThrow("Cancel unavailable");
await expect(response.cancel()).resolves.toEqual(acceptedCancellation());
expect(cancelTurn).toHaveBeenCalledTimes(2);
expect(cancelTurn).toHaveBeenNthCalledWith(1, "turn_1");
expect(cancelTurn).toHaveBeenNthCalledWith(2, "turn_1");
settle.resolve();
await consumed;
});
it("drops a queued cancellation when the response settles without starting a turn", async () => {
const [boundary] = stampTestEvents([
createSessionWaitingEvent(),
] as UnstampedMessageStreamEvent[]);
const cancelTurn = vi.fn(async () => acceptedCancellation());
const response = new MessageResponse({
cancelTurn,
createStream: async function* () {
yield boundary!;
},
sessionId: "session_1",
});
const consumed = consume(response);
await expect(response.cancel()).resolves.toEqual({ status: "no_active_turn" });
await consumed;
expect(cancelTurn).not.toHaveBeenCalled();
});
});
packages/eve/src/client/message-response.ts
+45 -3
View File
@@ -1,12 +1,13 @@
import type { MessageStreamEvent } from "#protocol/message.js";
import { isCurrentTurnBoundaryEvent, type MessageStreamEvent } from "#protocol/message.js";
import { extractCompletedResult } from "#client/output-schema.js";
import { summarizeTurnEvents } from "#client/session-utils.js";
import type { MessageResult } from "#client/types.js";
import type { CancelSessionResult, MessageResult } from "#client/types.js";
/**
* Internal configuration passed to construct a {@link MessageResponse}.
*/
interface MessageResponseInput {
readonly cancelTurn: (turnId: string) => Promise<CancelSessionResult>;
readonly createStream: () => AsyncGenerator<MessageStreamEvent>;
readonly sessionId: string;
}
@@ -24,15 +25,40 @@ export class MessageResponse<TOutput = unknown> implements AsyncIterable<Message
*/
readonly sessionId: string;
readonly #cancelTurn: (turnId: string) => Promise<CancelSessionResult>;
#cancellation: Promise<CancelSessionResult> | undefined;
#consumed = false;
readonly #createStream: () => AsyncGenerator<MessageStreamEvent>;
#settled = false;
readonly #turnId = Promise.withResolvers<string | undefined>();
/** @internal */
constructor(input: MessageResponseInput) {
this.#cancelTurn = input.cancelTurn;
this.sessionId = input.sessionId;
this.#createStream = input.createStream;
}
/**
* Requests cooperative cancellation of this exact turn.
*
* The request waits for the response stream to identify the turn when
* necessary. Continue consuming the stream to observe its durable boundary.
*/
cancel(): Promise<CancelSessionResult> {
if (this.#settled) return Promise.resolve({ status: "no_active_turn" });
if (this.#cancellation !== undefined) return this.#cancellation;
const cancellation = this.#turnId.promise.then<CancelSessionResult>((turnId) =>
turnId === undefined ? { status: "no_active_turn" } : this.#cancelTurn(turnId),
);
this.#cancellation = cancellation;
void cancellation.catch(() => {
if (!this.#settled && this.#cancellation === cancellation) this.#cancellation = undefined;
});
return cancellation;
}
/**
* Consumes the full event stream and returns the aggregated
* {@link MessageResult}.
@@ -66,6 +92,22 @@ export class MessageResponse<TOutput = unknown> implements AsyncIterable<Message
}
this.#consumed = true;
return this.#createStream();
return this.#observeStream();
}
async *#observeStream(): AsyncGenerator<MessageStreamEvent> {
try {
for await (const event of this.#createStream()) {
if (event.type === "turn.started") {
this.#turnId.resolve(event.data.turnId);
} else if (isCurrentTurnBoundaryEvent(event)) {
this.#settled = true;
this.#turnId.resolve(undefined);
}
yield event;
}
} finally {
this.#turnId.resolve(undefined);
}
}
}
packages/eve/src/client/session.ts
+1
View File
@@ -170,6 +170,7 @@ export class ClientSession {
): MessageResponse<TOutput> {
response.body?.cancel().catch(() => {});
return new MessageResponse<TOutput>({
cancelTurn: async (turnId) => await this.cancel({ turnId }),
createStream: () => this.#createEventStream(initialStreamIndex, input),
sessionId: this.#state.sessionId,
});
packages/eve/src/react/use-eve-agent.test.ts
+41
View File
@@ -277,6 +277,47 @@ describe("useEveAgent", () => {
);
});
it("detaches locally on unmount without cancelling the durable turn", async () => {
let requestSignal: AbortSignal | undefined;
const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementationOnce((_input, init) => {
requestSignal = init?.signal ?? undefined;
return new Promise<Response>((_resolve, reject) => {
requestSignal?.addEventListener("abort", () => reject(createAbortError()), {
once: true,
});
});
});
let helpers: UseEveAgentHelpers<EveMessageData> | undefined;
function TestComponent() {
helpers = useEveAgent();
return null;
}
let renderer: ReturnType<typeof create> | undefined;
await act(async () => {
renderer = create(createElement(TestComponent));
});
let sendPromise: Promise<void> | undefined;
await act(async () => {
sendPromise = helpers?.send("Hello");
await Promise.resolve();
});
await vi.waitFor(() => expect(requestSignal).toBeDefined());
await act(async () => {
renderer?.unmount();
});
await act(async () => {
await sendPromise;
});
expect(requestSignal?.aborted).toBe(true);
expect(fetchMock).toHaveBeenCalledOnce();
});
it("prepares fresh clientContext before sending without projecting it optimistically", async () => {
const startResponse = createDeferred<Response>();
const fetchMock = vi
packages/eve/src/react/use-eve-agent.ts
+10 -9
View File
@@ -1,6 +1,7 @@
import { useCallback, useEffect, useMemo, useRef, useSyncExternalStore } from "react";
import {
detachEveAgentStore,
EveAgentStore,
type EveAgentStoreCallbacks,
type EveAgentStoreSnapshot,
@@ -14,6 +15,7 @@ import { defaultMessageReducer, type EveMessageData } from "#client/message-redu
import type { MessageStreamEvent } from "#protocol/message.js";
import type { UserContent } from "ai";
import type {
CancelSessionResult,
ClientAuth,
HeadersValue,
RespondTurnOptions,
@@ -44,7 +46,9 @@ export type UseEveAgentSnapshot<TData> = EveAgentStoreSnapshot<TData>;
* Snapshot plus commands returned by `useEveAgent`.
*/
export interface UseEveAgentHelpers<TData> extends UseEveAgentSnapshot<TData> {
/** Resets the session: aborts any in-flight turn, recreates the owned session, and clears events and projected data. */
/** Requests durable cancellation of the active turn while continuing to receive its events. */
readonly cancel: () => Promise<CancelSessionResult>;
/** Resets the session: detaches any local stream, recreates the owned session, and clears events and projected data. */
readonly reset: () => void;
/** Sends a message. Rejects if a turn is already in flight. */
readonly send: <TOutput = unknown>(
@@ -56,8 +60,6 @@ export interface UseEveAgentHelpers<TData> extends UseEveAgentSnapshot<TData> {
inputResponses: Parameters<ClientSession["respond"]>[0],
options?: RespondTurnOptions<TOutput>,
) => Promise<void>;
/** Aborts the in-flight turn's stream, if any. */
readonly stop: () => void;
}
/**
@@ -119,7 +121,7 @@ export function useEveAgent<TData>(
* React hook that drives an eve session and projects its event stream into UI data.
*
* Returns the current snapshot (`data`, `events`, `session`, `status`, `error`)
* plus the commands `send`, `respond`, `stop`, and `reset`. With no reducer, `data` is the
* plus the commands `send`, `respond`, `cancel`, and `reset`. With no reducer, `data` is the
* built-in `UIMessage` projection from {@link defaultMessageReducer} (`TData`
* is {@link EveMessageData}); pass a reducer to project into your own shape and
* infer `TData`.
@@ -168,8 +170,9 @@ export function useEveAgent<TData>(
() => store.snapshot,
);
useEffect(() => () => store.stop(), [store]);
useEffect(() => () => detachEveAgentStore(store), [store]);
const cancel = useCallback(() => store.cancel(), [store]);
const reset = useCallback(() => store.reset(), [store]);
const send = useCallback(
<TOutput = unknown>(message: string | UserContent, options?: SendTurnOptions<TOutput>) => {
@@ -184,16 +187,14 @@ export function useEveAgent<TData>(
) => store.send({ ...options, inputResponses }),
[store],
);
const stop = useCallback(() => store.stop(), [store]);
return useMemo(
() => ({
...snapshot,
cancel,
reset,
respond,
send,
stop,
}),
[reset, respond, send, snapshot, stop],
[cancel, reset, respond, send, snapshot],
);
}
packages/eve/src/setup/scaffold/create/web-template.ts
+1 -1
View File
File diff suppressed because one or more lines are too long
packages/eve/src/setup/scaffold/index.integration.test.ts
+4 -6
View File
@@ -431,7 +431,7 @@ describe("ensureChannel", () => {
expect(normalizeEol(channelSource)).toBe(normalizeEol(sourceChannel));
});
test("scaffolds a Web Chat Stop button that cancels the active durable turn", async () => {
test("scaffolds a Web Chat Stop button with the agent cancellation API", async () => {
const projectRoot = await createTempDir();
await mkdir(join(projectRoot, "agent"), { recursive: true });
await writeFile(
@@ -450,11 +450,9 @@ describe("ensureChannel", () => {
join(projectRoot, "app/_components/agent-chat.tsx"),
"utf8",
);
expect(agentChatSource).toContain(".attach(sessionId)");
expect(agentChatSource).toContain(".cancel({ turnId })");
expect(agentChatSource).toContain("onSessionChange(session)");
expect(agentChatSource).toContain("cancellation.sentTurnId === turnId");
expect(agentChatSource).not.toContain("onStop={agent.stop}");
expect(agentChatSource).toContain("agent.cancel()");
expect(agentChatSource).not.toContain(".attach(sessionId)");
expect(agentChatSource).not.toContain('event.type !== "turn.started"');
});
test("writes npm dist-tags for Web Chat without semver range decoration", async () => {
packages/eve/src/svelte/use-eve-agent.ts
+9 -7
View File
@@ -2,6 +2,7 @@ import { createSubscriber } from "svelte/reactivity";
import type { UserContent } from "ai";
import {
detachEveAgentStore,
EveAgentStore,
type EveAgentStoreCallbacks,
type EveAgentStoreSnapshot,
@@ -13,6 +14,7 @@ import { defaultMessageReducer, type EveMessageData } from "#client/message-redu
import type { EveAgentReducer } from "#client/reducer.js";
import type { ClientSession } from "#client/session.js";
import type {
CancelSessionResult,
ClientAuth,
HeadersValue,
RespondTurnOptions,
@@ -45,6 +47,8 @@ export type UseEveAgentSnapshot<TData> = EveAgentStoreSnapshot<TData>;
* new events.
*/
export interface UseEveAgentReturn<TData> {
/** Request durable cancellation of the active turn while continuing to receive its events. */
readonly cancel: () => Promise<CancelSessionResult>;
/** Projected state built by reducing every stream event through the reducer. */
readonly data: TData;
/** Last transport-level error, or `undefined` when healthy. */
@@ -67,8 +71,6 @@ export interface UseEveAgentReturn<TData> {
readonly session: ClientSessionState | undefined;
/** Lifecycle phase: `"ready"` (idle), `"submitted"` (request sent, awaiting first event), `"streaming"` (events arriving), or `"error"`. */
readonly status: UseEveAgentStatus;
/** Abort the in-flight request. */
readonly stop: () => void;
}
/**
@@ -148,7 +150,7 @@ class SvelteEveAgent<TData> implements UseEveAgentReturn<TData> {
return () => {
unsubscribe();
store.stop();
detachEveAgentStore(store);
};
});
}
@@ -178,6 +180,10 @@ class SvelteEveAgent<TData> implements UseEveAgentReturn<TData> {
return this.#snapshot.status;
}
cancel = (): Promise<CancelSessionResult> => {
return this.#store.cancel();
};
reset = (): void => {
this.#store.reset();
};
@@ -195,10 +201,6 @@ class SvelteEveAgent<TData> implements UseEveAgentReturn<TData> {
): Promise<void> => {
return this.#store.send({ ...options, message });
};
stop = (): void => {
this.#store.stop();
};
}
export function useEveAgent(
packages/eve/src/vue/use-eve-agent.test.ts
+1 -1
View File
@@ -378,7 +378,7 @@ describe("useEveAgent (Vue composable wiring)", () => {
scope.stop();
});
it("unsubscribes and stops the session when the scope is disposed", async () => {
it("unsubscribes and detaches the local stream when the scope is disposed", async () => {
vi.stubGlobal("window", {});
vi.spyOn(globalThis, "fetch")
.mockResolvedValueOnce(createStartedMessageResponse("session_1", "http:session_1"))
packages/eve/src/vue/use-eve-agent.ts
+8 -6
View File
@@ -2,6 +2,7 @@ import { shallowRef, computed, onScopeDispose, type ComputedRef } from "vue";
import type { UserContent } from "ai";
import {
detachEveAgentStore,
EveAgentStore,
type EveAgentStoreCallbacks,
type EveAgentStoreSnapshot,
@@ -14,6 +15,7 @@ import type { ClientSession } from "#client/session.js";
import { defaultMessageReducer, type EveMessageData } from "#client/message-reducer.js";
import type { MessageStreamEvent } from "#protocol/message.js";
import type {
CancelSessionResult,
ClientAuth,
HeadersValue,
RespondTurnOptions,
@@ -43,6 +45,8 @@ export type UseEveAgentSnapshot<TData> = EveAgentStoreSnapshot<TData>;
* Reactive return value from `useEveAgent`.
*/
export interface UseEveAgentReturn<TData> {
/** Request durable cancellation of the active turn while continuing to receive its events. */
readonly cancel: () => Promise<CancelSessionResult>;
/** Projected state: the reducer folds every stream event into this value. */
readonly data: ComputedRef<TData>;
/** Last transport-level error, or `undefined` when healthy. */
@@ -65,8 +69,6 @@ export interface UseEveAgentReturn<TData> {
readonly session: ComputedRef<ClientSessionState | undefined>;
/** Lifecycle phase: `"ready"` (idle), `"submitted"` (request sent, awaiting first event), `"streaming"` (events arriving), or `"error"`. */
readonly status: ComputedRef<UseEveAgentStatus>;
/** Abort the in-flight request. */
readonly stop: () => void;
}
/**
@@ -146,8 +148,8 @@ export function useEveAgent<TData>(
* Without a `reducer`, events project into `EveMessageData` via
* `defaultMessageReducer()`; pass `reducer` to project into a custom `TData`.
* Returns reactive refs (`data`, `error`, `events`, `session`, `status`) plus
* `send`, `respond`, `stop`, and `reset`. Configuration is read once on store creation;
* remount to change it. On scope dispose, the in-flight request is aborted and
* `send`, `respond`, `cancel`, and `reset`. Configuration is read once on store creation;
* remount to change it. On scope dispose, the in-flight request is detached and
* the store unsubscribed.
*/
export function useEveAgent<TData>(
@@ -183,11 +185,12 @@ export function useEveAgent<TData>(
onScopeDispose(() => {
unsubscribe();
store.stop();
detachEveAgentStore(store);
});
}
return {
cancel: () => store.cancel(),
data: computed(() => snapshot.value.data),
error: computed(() => snapshot.value.error),
events: computed(() => snapshot.value.events),
@@ -200,6 +203,5 @@ export function useEveAgent<TData>(
store.send({ ...options, message }),
session: computed(() => snapshot.value.session),
status: computed(() => snapshot.value.status),
stop: () => store.stop(),
};
}
packages/eve/test/tui-client/tui-connection-auth-states.ts
+1
View File
@@ -55,6 +55,7 @@ class FakeSession extends ClientSession {
const events = this.#turns[this.#turnIndex] ?? [];
this.#turnIndex += 1;
return new MessageResponse<TOutput>({
cancelTurn: async () => ({ status: "no_active_turn" }),
sessionId: "fake-session",
createStream: () => pacedEvents(events),
});
research/frontend-durable-turn-cancellation.md
+59
View File
@@ -0,0 +1,59 @@
---
issue: "483"
status: implemented
last_updated: "2026-08-14"
---
# Durable frontend turn cancellation
## Summary
Frontend bindings exposed `stop()`, which aborted the browser stream without stopping durable server execution. Templates that needed a real Stop button rebuilt cancellation manually by tracking the session ID and waiting for `turn.started` to reveal a guarded turn ID.
Replace `stop()` with `cancel()` on the React, Vue, and Svelte bindings. Put exact-turn cancellation on `MessageResponse`, the client abstraction that already owns one accepted turn and its stream, so every frontend binding delegates to the same lifecycle.
## Authoring API
```ts
const agent = useEveAgent();
const sending = agent.send("Run the analysis");
await agent.cancel();
await sending;
```
Lower-level clients can cancel the exact response while continuing to consume its stream:
```ts
const response = await session.send("Run the analysis");
const result = response.result();
await response.cancel();
await result;
```
## Semantics
```mermaid
flowchart LR
UI["cancel()"] --> Response["Active MessageResponse"]
Response --> Started["Observe turn.started"]
Started --> Guarded["session.cancel({ turnId })"]
Guarded --> Boundary["turn.cancelled then session.waiting"]
```
- Cancellation can be requested while a send is still submitted. The frontend binding waits for its accepted `MessageResponse`; the response then waits for its own `turn.started` event before dispatching one guarded request.
- Concurrent cancellation calls for the same live turn share one request. A failed request can be retried while the turn remains active.
- Cancellation acceptance and turn settlement are separate. The stream remains attached until the durable boundary so the session cursor advances normally.
- Component disposal, page closure, `AbortSignal`, and `reset()` detach local transport only. They do not imply durable cancellation.
- A response that reaches a boundary without starting a turn, or a binding with no active response, returns `no_active_turn`.
## Scope
The existing session cancel route and `ClientSession.cancel({ turnId })` remain available for consumers that have only a session handle. `detach` stays an internal framework-adapter operation rather than a second public stopping primitive.
## Validation
- Unit coverage exercises queued cancellation, exact turn-ID guarding, single-flight calls, rejection retry, reset races, boundary races, and local detach behavior.
- React, Vue, and Svelte typechecks prove the shared public surface, and their example applications handle cancellation request failures.
- Existing cancellation e2e fixtures continue to prove durable `turn.cancelled` followed by `session.waiting` and successful session continuation.