fix: end a closed session's turns before session/close answers (#1242)

Found while reviewing the ACP v2 prompt lifecycle (#1226). It affects
ACP v1 and v2.

## Problem

`session/close` tears the session down: `teardownSession` cancels its
turns, wakes the consumer, and closes the query. It returned without
waiting for the cancelled turns to end, so the close could answer first:
- **v1:** the `session/prompt` response arrived after the
`session/close` response.
- **v2:** the turn's `idle cancelled` arrived after the close response.
v2's `session/close` cancels "as if `session/cancel` had been called",
then frees the session.

It happens when the consumer settles the turn late: for example, while
it's still sending an update to a slow client, or when it settles the
turn only at the end of the stream that the close ends.

## Change

`teardownSession` now returns once the turns it cancels have ended (each
turn's `completion`), for at most `TEARDOWN_TURN_END_TIMEOUT_MS` (5 s),
which it logs if a wedged consumer hits it. Its other callers get the
same order too: `session/delete`, the recreation of a session whose
parameters changed on resume or load, and `dispose()`.

## Found along the way, not changed here

The consumer's abort race adds an `abort` listener to the cancel signal
on each iteration. A listener added to an already-aborted signal never
fires, so an abort that lands while the consumer is busy, such as
sending an update, is missed. In production the close still ends the
turn, through the end of the stream, which is the path this PR waits
for. But the same gap can defeat the force-cancel backstop for a wedged
Claude Code (issue #680). That gets its own PR next.

## Verification

A new turn-events test has the client accept an answer chunk slowly,
closes the session while the consumer is still sending it, and checks
that the turn's end comes before the close returns. It fails without the
change: the close returns first. Build, lint and all 2389 tests pass.

src/acp-agent.ts

@@ -373,6 +373,10 @@ const DEFAULT_CONTEXT_WINDOW = 200000;
  *  pre-empt a slow-but-healthy interrupt. */
 const DEFAULT_FORCE_CANCEL_GRACE_MS = 30_000;
 const STRUCTURED_USAGE_TIMEOUT_MS = 5_000;
+/** How long a session teardown waits for the turns that it cancels to end.
+ *  The teardown wakes the consumer itself, so they end within a few ticks
+ *  unless the consumer is wedged. */
+const TEARDOWN_TURN_END_TIMEOUT_MS = 5_000;
 /** The number of settled subagents whose parent tool call a session keeps for
  *  a later resume (see `resumableSubagents`). */
 const MAX_RESUMABLE_SUBAGENTS = 256;
@@ -7479,12 +7483,20 @@ export class ClaudeAcpAgent {
   }
 
   /** Cleanly tear down a session: cancel in-flight work, release stream
-   *  resources, and remove it from the session map. */
+   *  resources, and remove it from the session map. Returns once the turns it
+   *  cancels have ended, so their end (a v1 `session/prompt` response, a v2
+   *  `idle` state) goes out before the caller answers: ACP v2's
+   *  `session/close` cancels "as if `session/cancel` had been called", then
+   *  frees the session. A wedged consumer holds it at most
+   *  {@link TEARDOWN_TURN_END_TIMEOUT_MS}. */
   private async teardownSession(sessionId: string): Promise<void> {
     const session = this.sessions[sessionId];
     if (!session) {
       return;
     }
+    const turnsEnded = Promise.all(
+      (session.turnQueue ?? []).flatMap((turn) => (turn.completion ? [turn.completion] : [])),
+    );
     try {
       await this.cancelTurns({ sessionId }, { awaitInterrupt: false });
     } catch (error) {
@@ -7511,6 +7523,16 @@ export class ClaudeAcpAgent {
     session.nativeSubagentRuntime?.clear();
     session.asyncTaskRuntime?.clear();
     delete this.sessions[sessionId];
+    const ended = await raceTimeoutAndAbort(
+      turnsEnded,
+      TEARDOWN_TURN_END_TIMEOUT_MS,
+      new AbortController().signal,
+    );
+    if (ended.type === "timeout") {
+      this.logger.error(
+        `Session ${sessionId}: its cancelled turns did not end within ${TEARDOWN_TURN_END_TIMEOUT_MS}ms of the teardown`,
+      );
+    }
   }
 
   /** Tear down all active sessions. Called when the ACP connection closes. */

src/tests/turn-events.test.ts

@@ -647,6 +647,45 @@ describe("turn events", () => {
     expect(second.log).toEqual(["ended cancelled"]);
   });
 
+  it("ends a closed session's turn before the close returns, also for a slow client", async () => {
+    // Claude Code streams an answer and runs until the close ends its stream.
+    const queryClosed = Promise.withResolvers<void>();
+    const promptUuids: string[] = [];
+    scriptQuery(
+      async function* (input, options) {
+        const { value } = await input.next();
+        promptUuids.push(value.uuid);
+        yield echo(value, options.sessionId!);
+        for (const message of streamedText("msg_answer", "Hi", options.sessionId!)) {
+          yield message;
+        }
+        await queryClosed.promise;
+      },
+      { close: () => queryClosed.resolve() },
+    );
+    const sessionId = await newSession();
+    const turn = recordEvents();
+    // The client takes the answer's text slowly, so the consumer is still
+    // sending it when the close tears the session down.
+    const chunkSeen = Promise.withResolvers<void>();
+    const chunkTaken = Promise.withResolvers<void>();
+    onSessionUpdate = async ({ update }) => {
+      if (update.sessionUpdate !== "agent_message_chunk" || update.messageId !== "msg_answer") {
+        return;
+      }
+      chunkSeen.resolve();
+      await chunkTaken.promise;
+    };
+
+    await agent.startTurn(prompt(sessionId, "hello"), turn.events);
+    await chunkSeen.promise;
+    const closed = agent.closeSession({ sessionId }).then(() => turn.log.push("closed"));
+    setTimeout(() => chunkTaken.resolve(), 10);
+    await closed;
+
+    expect(turn.log).toEqual([`inserted ${promptUuids[0]}`, "ended cancelled", "closed"]);
+  });
+
   it("fails a queued turn without inserting it when the query ends", async () => {
     scriptTurns(
       [