Move abort operation from ISession to ITurn

gateways/web/app/src/components/ChatView.tsx

@@ -251,7 +251,8 @@ export const ChatView: Component<Props> = (props) => {
 
   async function handleAbort(): Promise<void> {
     try {
-      await session.abort();
+      const turn = await session.getCurrentTurn();
+      await turn?.abort();
     } catch {
       /* turn may have ended */
     }

gateways/web/test/mocks/piccolo-core.ts

@@ -38,8 +38,8 @@ export function createEventObservable(events: AgentEvent[]): IObservable<AgentEv
 
 function emptyTurn(): ITurn {
   return {
-    getObservable: vi.fn().mockResolvedValue(createEventObservable([])),
     getCallback: vi.fn().mockResolvedValue(undefined),
+    abort: vi.fn().mockResolvedValue(undefined),
   } as unknown as ITurn;
 }
 
@@ -60,7 +60,6 @@ export function createMockSession(
     sendUserMessage: vi.fn().mockResolvedValue(undefined),
     steer: vi.fn().mockResolvedValue(undefined),
     followUp: vi.fn().mockResolvedValue(undefined),
-    abort: vi.fn().mockResolvedValue(undefined),
     subscribe: vi.fn().mockResolvedValue(noopSubscription),
     getCurrentTurn: vi.fn().mockResolvedValue(undefined),
     getModel: vi.fn().mockResolvedValue(DEFAULT_MODEL),

packages/api/src/index.ts

@@ -256,6 +256,7 @@ export interface IObservable<T> {
  */
 export interface ITurn {
   getCallback(): Promise<IGatewayCallback | undefined>;
+  abort(): Promise<void>;
 }
 
 // ─── Context / Compaction ─────────────────────────────────────────────────────
@@ -326,7 +327,6 @@ export interface ISession extends IObservable<AgentEvent> {
   sendUserMessage(content: string): Promise<void>;
   steer(text: string): Promise<void>;
   followUp(text: string): Promise<void>;
-  abort(): Promise<void>;
   getCurrentTurn(): Promise<ITurn | undefined>;
 
   // ─── Model management ────────────────────────────────────────────────────────

packages/core/src/agent-session-do.ts

@@ -120,9 +120,6 @@ export class SessionTarget extends RpcTarget implements ISession {
   followUp(text: string): Promise<void> {
     return this.#do.followUp(text);
   }
-  abort(): Promise<void> {
-    return this.#do.abort();
-  }
   getCurrentTurn(): Promise<ITurn | undefined> {
     return this.#do.getCurrentTurn();
   }
@@ -376,9 +373,11 @@ export class AgentSessionDO extends DurableObject<Env> 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<Env> implements ISession {
     this.#followUpQueue.push(text);
   }
 
-  async abort(): Promise<void> {
+  async #abortCurrentTurn(): Promise<void> {
     this.#agentAbortController?.abort();
   }
 
@@ -1161,15 +1160,21 @@ export class AgentSessionDO extends DurableObject<Env> implements ISession {
 
 export class TurnImpl extends RpcTarget implements ITurn {
   readonly #callback: IGatewayCallback | undefined;
+  readonly #abort: () => Promise<void>;
 
-  constructor(callback: IGatewayCallback | undefined) {
+  constructor(callback: IGatewayCallback | undefined, abort: () => Promise<void>) {
     super();
     this.#callback = callback;
+    this.#abort = abort;
   }
 
   async getCallback(): Promise<IGatewayCallback | undefined> {
     return this.#callback;
   }
+
+  async abort(): Promise<void> {
+    await this.#abort();
+  }
 }
 
 // ─── Token estimation ─────────────────────────────────────────────────────────

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();

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([]);

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 () => {},

specs/api.md

@@ -311,9 +311,6 @@ interface ISession extends IObservable<AgentEvent> {
   // without interrupting an active turn.
   followUp(text: string): Promise<void>;
 
-  // Abort the current streaming turn immediately.
-  abort(): Promise<void>;
-
   // 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<IGatewayCallback | undefined>;
+
+  // Abort this turn if it is still running.
+  abort(): Promise<void>;
 }
 ```
 

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`.

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 <id>` | `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 |