diff --git a/gateways/web/app/src/components/ChatView.tsx b/gateways/web/app/src/components/ChatView.tsx index f542334..0288572 100644 --- a/gateways/web/app/src/components/ChatView.tsx +++ b/gateways/web/app/src/components/ChatView.tsx @@ -251,7 +251,8 @@ export const ChatView: Component = (props) => { async function handleAbort(): Promise { try { - await session.abort(); + const turn = await session.getCurrentTurn(); + await turn?.abort(); } catch { /* turn may have ended */ } diff --git a/gateways/web/test/mocks/piccolo-core.ts b/gateways/web/test/mocks/piccolo-core.ts index 13b0da5..b2f0ef3 100644 --- a/gateways/web/test/mocks/piccolo-core.ts +++ b/gateways/web/test/mocks/piccolo-core.ts @@ -38,8 +38,8 @@ export function createEventObservable(events: AgentEvent[]): IObservable { */ export interface ITurn { getCallback(): Promise; + abort(): Promise; } // ─── Context / Compaction ───────────────────────────────────────────────────── @@ -326,7 +327,6 @@ export interface ISession extends IObservable { sendUserMessage(content: string): Promise; steer(text: string): Promise; followUp(text: string): Promise; - abort(): Promise; getCurrentTurn(): Promise; // ─── Model management ──────────────────────────────────────────────────────── diff --git a/packages/core/src/agent-session-do.ts b/packages/core/src/agent-session-do.ts index dbc06c4..26a3c78 100644 --- a/packages/core/src/agent-session-do.ts +++ b/packages/core/src/agent-session-do.ts @@ -120,9 +120,6 @@ export class SessionTarget extends RpcTarget implements ISession { followUp(text: string): Promise { return this.#do.followUp(text); } - abort(): Promise { - return this.#do.abort(); - } getCurrentTurn(): Promise { return this.#do.getCurrentTurn(); } @@ -376,9 +373,11 @@ export class AgentSessionDO extends DurableObject implements ISession { // concurrent caller that runs before the microtask queue yields will // immediately see #currentTurn !== null and throw. if (this.#currentTurn !== null) { - throw new Error("A turn is already in progress. Call abort() before starting a new turn."); + throw new Error( + "A turn is already in progress. Call getCurrentTurn() and abort() that turn before starting a new turn.", + ); } - const turn = new TurnImpl(callback); + const turn = new TurnImpl(callback, () => this.#abortCurrentTurn()); this.#currentTurn = turn; try { @@ -480,7 +479,7 @@ export class AgentSessionDO extends DurableObject implements ISession { this.#followUpQueue.push(text); } - async abort(): Promise { + async #abortCurrentTurn(): Promise { this.#agentAbortController?.abort(); } @@ -1161,15 +1160,21 @@ export class AgentSessionDO extends DurableObject implements ISession { export class TurnImpl extends RpcTarget implements ITurn { readonly #callback: IGatewayCallback | undefined; + readonly #abort: () => Promise; - constructor(callback: IGatewayCallback | undefined) { + constructor(callback: IGatewayCallback | undefined, abort: () => Promise) { super(); this.#callback = callback; + this.#abort = abort; } async getCallback(): Promise { return this.#callback; } + + async abort(): Promise { + await this.#abort(); + } } // ─── Token estimation ───────────────────────────────────────────────────────── diff --git a/packages/core/test/do/agent-session.test.ts b/packages/core/test/do/agent-session.test.ts index 6953078..d693528 100644 --- a/packages/core/test/do/agent-session.test.ts +++ b/packages/core/test/do/agent-session.test.ts @@ -325,7 +325,7 @@ describe("AgentSessionDO — abort", () => { await instance._init(sid, "user-1"); const flushed = waitForEvent(instance, (e) => e.type === "turn_flushed"); const ev = await drainTurn(instance, "go"); - await instance.abort(); + await (await instance.getCurrentTurn())?.abort(); await flushed; return ev; }); @@ -857,7 +857,7 @@ describe("AgentSessionDO — abort() (inlined)", () => { await instance._init(sid, "user-1"); const flushed = waitForEvent(instance, (e) => e.type === "turn_flushed"); const promptPromise = drainTurn(instance, "go"); - await instance.abort(); + await (await instance.getCurrentTurn())?.abort(); const ev = await promptPromise; await flushed; return ev; @@ -875,7 +875,7 @@ describe("AgentSessionDO — abort() (inlined)", () => { await instance._init(sid, "user-1"); const flushed = waitForEvent(instance, (e) => e.type === "turn_flushed"); const promptPromise = drainTurn(instance, "go"); - await instance.abort(); + await (await instance.getCurrentTurn())?.abort(); await promptPromise; await flushed; const turn = await instance.getCurrentTurn(); diff --git a/packages/core/test/do/extension-runner.test.ts b/packages/core/test/do/extension-runner.test.ts index 28864c2..0687c81 100644 --- a/packages/core/test/do/extension-runner.test.ts +++ b/packages/core/test/do/extension-runner.test.ts @@ -675,7 +675,6 @@ describe("createMockSession() — all methods reachable", () => { await expect(s.sendUserMessage("hi")).resolves.toBeUndefined(); await expect(s.steer("steer")).resolves.toBeUndefined(); await expect(s.followUp("follow")).resolves.toBeUndefined(); - await expect(s.abort()).resolves.toBeUndefined(); expect(await s.getModel()).toBe("test/model"); await expect(s.setModel("x")).resolves.toBeUndefined(); expect(await s.listModels()).toEqual([]); diff --git a/packages/core/test/mocks/extension-stub.ts b/packages/core/test/mocks/extension-stub.ts index 30fd774..b84f6f6 100644 --- a/packages/core/test/mocks/extension-stub.ts +++ b/packages/core/test/mocks/extension-stub.ts @@ -278,7 +278,6 @@ export function createMockSession( sendUserMessage: async () => {}, steer: async () => {}, followUp: async () => {}, - abort: async () => {}, getCurrentTurn: async () => undefined, getModel: async () => "test/model", setModel: async () => {}, diff --git a/specs/api.md b/specs/api.md index 12a392b..688dc36 100644 --- a/specs/api.md +++ b/specs/api.md @@ -311,9 +311,6 @@ interface ISession extends IObservable { // without interrupting an active turn. followUp(text: string): Promise; - // Abort the current streaming turn immediately. - abort(): Promise; - // Return the active turn context (if a turn is in progress), or undefined if idle. // Gateways call this after getHistory() to reconnect to an in-progress turn // Returns undefined when idle; non-undefined means a turn is in progress. @@ -410,6 +407,9 @@ interface ITurn { // Tools call this to request interactive input mid-turn (select, confirm, input). // Returns undefined if the gateway did not supply a callback for this turn. getCallback(): Promise; + + // Abort this turn if it is still running. + abort(): Promise; } ``` diff --git a/specs/gateway.md b/specs/gateway.md index 5a33893..f97498a 100644 --- a/specs/gateway.md +++ b/specs/gateway.md @@ -45,9 +45,9 @@ const session: ISession = await core.getSession(storedSessionId); // All per-session operations are on the ISession stub const turn = await session.prompt("Hello", attachments); -const stream = await turn.getStream(); await session.setModel("anthropic/claude-sonnet-4-5"); -await session.abort(); +const currentTurn = await session.getCurrentTurn(); +await currentTurn?.abort(); ``` See [api.md §1–2](api.md) for `IPiccoloCore` and `ISession`. diff --git a/specs/telegram_gateway.md b/specs/telegram_gateway.md index 1cb92e0..c0ede86 100644 --- a/specs/telegram_gateway.md +++ b/specs/telegram_gateway.md @@ -165,7 +165,7 @@ Telegram `/commands` are handled by the gateway and not forwarded to the agent c | `/new` | `core.newSession()`, update KV mapping, confirm to user | | `/model ` | `session.setModel(id)`, confirm to user | | `/models` | `core.listModels()`, reply with list | -| `/abort` | `session.abort()`, confirm to user | +| `/abort` | `session.getCurrentTurn()?.abort()`, confirm to user | | `/status` | `session.getModel()` + `session.getContextUsage()`, reply with model + token usage | | `/compact` | `session.compact()`, confirm to user | | `/help` | Reply with list of available commands and descriptions |