fix(trace): a dropped turn ends with inference:aborted, and Discord typing stops at a normal end

changelog.d/discord-typing-stops-at-completion.fixed.md

@@ -0,0 +1 @@
+- The Discord module's typing keepalive now stops when a turn completes normally, at `inference:completed`. It stopped only at `inference:turn_ended`, `inference:aborted`, `inference:failed` and `inference:exhausted`, so with `autoTyping` on, its default, the indicator went on refreshing after an ordinary reply until the agent's next `inference:started` or its five-minute cap. `onTrace`'s documentation now names `inference:completed` as a turn's normal end, beside the other four.

changelog.d/turn-drop-terminal-event.fixed.md

@@ -0,0 +1 @@
+- A turn the framework drops before it sends anything, because its agent was replaced or removed, or the framework was stopping or had been stopped and started again while the turn was assembled or compiled, now ends with `inference:aborted`, whose `reason` says which, and records `lastInference.endedAt`. The turn had emitted `inference:started` and then no terminal event, so a module that brackets a turn between them, such as a typing indicator, kept that turn open, and the health snapshot showed it still running. When a newer turn of the agent's name already holds its turn token, the dropped turn emits and records nothing, since both carry only the name and would end that newer turn. The web UI in `ui/` now also returns an agent to idle on `inference:turn_ended`, `inference:aborted` and `inference:exhausted`, where it had looked only at `inference:completed` and `inference:failed`.

src/framework.ts

@@ -11334,6 +11334,28 @@ export class AgentFramework {
     }
   }
 
+  /**
+   * End a turn that beginAgentTurn drops before it sends anything. The turn
+   * emitted inference:started, and listeners that bracket a turn, such as a
+   * typing loop or an activity indicator, end it on a terminal event, so it
+   * gets inference:aborted with `reason`, and lastInference records its end
+   * as driveStream's would, when the record is still this turn's. Nothing of
+   * the kind when another turn of this agent's name holds the turn token (one
+   * begun after a stop and start, or by a replacement agent object): the
+   * event and the record carry only the name, so this frame's would end that
+   * newer turn instead.
+   */
+  private abortDroppedTurn(agentName: string, turnToken: number, reason: string, startedAt: number): void {
+    const owner = this.activeTurnTokens.get(agentName);
+    if (owner !== undefined && owner !== turnToken) return;
+    const now = Date.now();
+    // Only this turn's own record ends: a name whose run was disposed meanwhile
+    // (runEphemeralToCompletion deletes its record) gets none back.
+    const last = this.lastInferenceAt.get(agentName);
+    if (last?.startedAt === startedAt) this.lastInferenceAt.set(agentName, { ...last, endedAt: now });
+    this.emitTrace({ type: 'inference:aborted', agentName, reason, durationMs: now - startedAt });
+  }
+
   /**
    * Body of startAgentStream, split out so the caller's try/finally owns the
    * turn token unconditionally. Returns true iff the token was handed off to
@@ -11548,6 +11570,7 @@ export class AgentFramework {
     const turnLocus = turnProseRouting === 'disabled' ? undefined : routing.locus;
 
     this.touchEphemeralRun(agent.name, true);
+    const turnStartedAt = Date.now();
     this.emitTrace({
       type: 'inference:started',
       agentName: agent.name,
@@ -11560,7 +11583,7 @@ export class AgentFramework {
     if (!resumesTurn) {
       this.eventGate?.onInferenceStarted(agent.name);
     }
-    this.lastInferenceAt.set(agent.name, { ...this.lastInferenceAt.get(agent.name), startedAt: Date.now() });
+    this.lastInferenceAt.set(agent.name, { ...this.lastInferenceAt.get(agent.name), startedAt: turnStartedAt });
 
     // Typing indicator, started at TURN START. It says "attending", which is
     // true from the moment the turn is accepted — but it used to start inside
@@ -11635,9 +11658,13 @@ export class AgentFramework {
       // and stop() may close provider admission. Do not create a provider
       // stream after its generation lost ownership, or once admission closed.
       if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed) {
+        const why = this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed' : 'the framework was stopping';
+        const reason = `a turn assembled while ${why} is dropped before it sends anything`;
+        console.error(`[turn] ${agent.name}: ${reason}`);
         this.endDroppedTurn(agent.name, routing);
         this.channelRegistry?.stopTyping();
         this.eventGate?.onInferenceEnded(agent.name);
+        this.abortDroppedTurn(agent.name, turnToken, reason, turnStartedAt);
         return false;
       }
 
@@ -11666,10 +11693,11 @@ export class AgentFramework {
       // Membrane's yielding stream starts its request when it is first
       // iterated, so cancelling it here, before driveStream, sends nothing.
       if (this.agents.get(agent.name) !== agent || this.providerAdmissionClosed || lifecycle !== this.lifecycleGeneration) {
-        console.error(`[turn] ${agent.name}: a turn compiled while ${
-          this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed'
-            : this.providerAdmissionClosed ? 'the framework was stopping'
-              : 'the framework was stopped and started again'} is dropped before it sends anything`);
+        const why = this.agents.get(agent.name) !== agent ? 'the agent was replaced or removed'
+          : this.providerAdmissionClosed ? 'the framework was stopping'
+            : 'the framework was stopped and started again';
+        const reason = `a turn compiled while ${why} is dropped before it sends anything`;
+        console.error(`[turn] ${agent.name}: ${reason}`);
         stream.cancel();
         agent.cancelStream();
         if (this.agents.get(agent.name) === agent) {
@@ -11682,6 +11710,7 @@ export class AgentFramework {
         this.endDroppedTurn(agent.name, routing);
         this.channelRegistry?.stopTyping();
         this.eventGate?.onInferenceEnded(agent.name);
+        this.abortDroppedTurn(agent.name, turnToken, reason, turnStartedAt);
         return false;
       }
       // A conversation agent takes no provider gate, so its own compression

src/modules/discord/index.ts

@@ -67,9 +67,11 @@ const TYPING_REFRESH_MS = 8_000;
 // stuck on forever.
 const TYPING_MAX_MS = 5 * 60_000;
 
-// Trace event types that terminate an agent's turn — any of these stops the
-// typing keepalive. They fire once per turn, after all tool-call cycles.
+// Trace event types that end an agent's turn (onTrace's contract) — any of
+// these stops the typing keepalive. A turn's tool-call cycles stay inside its
+// stream, so they come after them. inference:completed is the normal end.
 const TURN_TERMINAL_TRACES = new Set([
+  'inference:completed',
   'inference:turn_ended',
   'inference:aborted',
   'inference:failed',

src/types/module.ts

@@ -321,10 +321,14 @@ export interface ModuleContext {
    *
    * Modules use this to react to agent lifecycle without polling — e.g. an
    * adapter that wants to show a "typing" indicator can bracket it between
-   * `inference:started` and the terminal events (`inference:turn_ended`,
-   * `inference:aborted`, `inference:failed`, `inference:exhausted`), all of
-   * which carry `agentName`. The terminal events fire once per turn, after any
-   * tool-call cycles, so a single start/stop bracket spans the whole turn.
+   * `inference:started` and the terminal events: `inference:completed` (the
+   * normal end), `inference:turn_ended` (a tool result ended the turn),
+   * `inference:aborted`, `inference:failed` and `inference:exhausted`, all of
+   * which carry `agentName`. A turn's tool-call cycles stay inside its
+   * stream, so one bracket spans them. A context-budget restart emits
+   * `inference:started` again with no terminal event for the stream it
+   * replaced, and a failing turn can emit `inference:failed` and then
+   * `inference:exhausted`, so a bracket should allow a repeated start or end.
    *
    * Returns an unsubscribe function. Call it in the module's `stop()` to avoid
    * leaking listeners across teardown/recreate (e.g. session switch).

src/types/trace.ts

@@ -63,7 +63,15 @@ export type TraceEvent =
   | (TraceEventBase & {
       type: 'inference:aborted';
       agentName: string;
+      /** How long this start of the turn ran; after a context-budget restart, from the restart. */
       durationMs: number;
+      /**
+       * Why the turn ended unfinished: the reason an operator gave
+       * `abortInference`, `'quiesce_abandoned'` for a turn an operator quiesce
+       * abandoned, or, for a turn dropped before it sent anything (its agent
+       * replaced or removed, or the framework stopping or restarted), a
+       * sentence saying which.
+       */
       reason?: string;
     })
   | (TraceEventBase & {

test/discord-typing-keepalive.test.ts

@@ -0,0 +1,85 @@
+import { test, mock } from 'node:test';
+import assert from 'node:assert/strict';
+import { DiscordModule } from '../src/modules/discord/index.js';
+import type { ConversationContext, DiscordClientInterface } from '../src/modules/discord/types.js';
+import type { ModuleContext, TraceEvent, TraceEventListener } from '../src/types/index.js';
+
+// Discord's typing indicator expires after about 10 s, so the module refreshes
+// it every 8 s (TYPING_REFRESH_MS) from a turn's inference:started until an
+// event that ends the turn.
+const REFRESH_MS = 8_000;
+const agentName = 'assistant';
+
+const turnEnds: TraceEvent[] = [
+  // The normal end: the turn's stream completed.
+  { type: 'inference:completed', agentName, durationMs: 5, timestamp: 0 },
+  { type: 'inference:turn_ended', agentName, timestamp: 0 },
+  { type: 'inference:aborted', agentName, durationMs: 5, timestamp: 0 },
+  { type: 'inference:failed', agentName, error: 'provider error', timestamp: 0 },
+  { type: 'inference:exhausted', agentName, error: 'provider error', timestamp: 0 },
+];
+
+async function startModule(): Promise<{
+  module: DiscordModule;
+  emit: TraceEventListener;
+  typings: () => number;
+}> {
+  let typings = 0;
+  const client = {
+    connect: async () => {},
+    disconnect: async () => {},
+    isConnected: () => true,
+    getBotUserId: () => undefined,
+    onMessage: () => {},
+    onMessageEdit: () => {},
+    onMessageDelete: () => {},
+    sendTyping: async () => {
+      typings += 1;
+    },
+  } as unknown as DiscordClientInterface;
+  let listener: TraceEventListener | undefined;
+  const ctx = {
+    isRestart: false,
+    getState: () => undefined,
+    setState: () => {},
+    registerSpeechHandler: () => {},
+    onTrace: (subscriber: TraceEventListener) => {
+      listener = subscriber;
+      return () => {
+        listener = undefined;
+      };
+    },
+  } as unknown as ModuleContext;
+  const module = new DiscordModule(client, { token: 'test-token' });
+  await module.start(ctx);
+  assert.ok(listener, 'autoTyping, on by default, subscribes to the trace stream');
+  // Where the agent's reply goes, as an incoming Discord message would set it.
+  (module as unknown as { replyContexts: Map<string, ConversationContext> }).replyContexts.set('default', {
+    channelId: 'channel-1',
+    userId: 'user-1',
+  } as ConversationContext);
+  return { module, emit: (event) => listener?.(event), typings: () => typings };
+}
+
+for (const end of turnEnds) {
+  test(`the typing keepalive a turn started stops at ${end.type}`, async () => {
+    mock.timers.enable({ apis: ['setInterval'] });
+    try {
+      const { module, emit, typings } = await startModule();
+      try {
+        emit({ type: 'inference:started', agentName, timestamp: 0 });
+        assert.equal(typings(), 1, 'typing shows as the turn starts');
+        mock.timers.tick(REFRESH_MS);
+        assert.equal(typings(), 2, 'and is refreshed while the turn runs');
+
+        emit(end);
+        mock.timers.tick(REFRESH_MS * 3);
+        assert.equal(typings(), 2, `no refresh after ${end.type}`);
+      } finally {
+        await module.stop();
+      }
+    } finally {
+      mock.timers.reset();
+    }
+  });
+}

test/tool-result-guard.test.ts

@@ -3229,6 +3229,13 @@ test('a framework compile from before stop cannot drive after restart', async ()
     return result;
   };
   let starting: Promise<void> | undefined;
+  // The dropped turn's inference:* events, as a module's trace listener sees them.
+  const turnEvents: Array<{ type: string; reason?: string }> = [];
+  const unsubscribe = framework.onTrace((event) => {
+    if (event.type.startsWith('inference:') && (event as { agentName?: string }).agentName === 'assistant') {
+      turnEvents.push({ type: event.type, ...('reason' in event ? { reason: String(event.reason) } : {}) });
+    }
+  });
   try {
     const lines: string[] = [];
     const error = console.error;
@@ -3243,12 +3250,112 @@ test('a framework compile from before stop cannot drive after restart', async ()
     } finally { console.error = error; }
     assert.equal(driven, 0);
     assert.ok(lines.some((line) => /stopped and started again/.test(line) && /dropped before it sends anything/.test(line)), lines.join('\n'));
+    // The turn emitted inference:started before its compile, so it gets one
+    // terminal event: a listener's bracket (a typing loop) closes.
+    assert.deepEqual(turnEvents.map((event) => event.type), ['inference:started', 'inference:aborted']);
+    assert.match(turnEvents[1]?.reason ?? '', /stopped and started again.*dropped before it sends anything/);
+    unsubscribe();
     // A turn begun after the restart is admitted and drives its stream.
     await (framework as unknown as { startAgentStream(a: typeof agent): Promise<void> }).startAgentStream(agent);
     await until(() => driven === 1, 'the turn begun after the restart drives');
   } finally { compile.open(); await starting; await framework.stop(); store.close(); }
 });
 
+// A turn dropped before it sends anything, at either check: before its
+// compile (it waits in a module's gatherContext) or after it (it waits in its
+// compile). Each way it can lose its place: the framework stops, the agent is
+// removed, the agent object is replaced under its name, or the replacement's
+// own turn already holds the name's turn token.
+for (const phase of ['before', 'after'] as const) {
+  for (const how of ['stopping', 'removed', 'disposed', 'replaced', 'superseded'] as const) {
+    const title = { stopping: 'the framework stopping', removed: 'its agent removed',
+      disposed: 'its run disposed with its record', replaced: 'its agent replaced',
+      superseded: "a newer turn of its agent's name running" }[how];
+    test(`a turn dropped ${phase} its compile, with ${title}, ${how === 'superseded' ? "leaves the newer turn's bracket alone" : 'ends with inference:aborted'}`, async () => {
+      const dir = mkdtempSync(join(tmpdir(), `guard-drop-${phase}-compile-`)); dirs.push(dir);
+      const store = JsStore.openOrCreate({ path: join(dir, 'store') });
+      const membrane = new ScriptMembrane([[answer()]]);
+      const framework = await AgentFramework.create({ store, membrane: membrane.asMembrane(),
+        agents: [{ name: 'assistant', model: 'test', systemPrompt: 'system' }],
+        modules: [], syncIntervalMs: 0, maintenanceIntervalMs: 0 });
+      const agent = framework.getAgent('assistant')!;
+      const internal = framework as unknown as {
+        moduleRegistry: { gatherContext: (n: string) => Promise<unknown[]> };
+        agents: Map<string, unknown>;
+        activeTurnTokens: Map<string, number>;
+        lastInferenceAt: Map<string, { startedAt?: number; endedAt?: number }>;
+      };
+      const held = gate();
+      const entered = gate();
+      if (phase === 'before') {
+        internal.moduleRegistry.gatherContext = async () => { entered.open(); await held.opened; return []; };
+      } else {
+        const build = agent.buildActivationRequest.bind(agent);
+        agent.buildActivationRequest = async (...args) => { entered.open(); await held.opened; return build(...args); };
+      }
+      const turnEvents: Array<{ type: string; reason?: string; durationMs?: number }> = [];
+      framework.onTrace((event) => {
+        if (event.type.startsWith('inference:') && (event as { agentName?: string }).agentName === 'assistant') {
+          turnEvents.push({ type: event.type,
+            ...('reason' in event ? { reason: String(event.reason) } : {}),
+            ...('durationMs' in event ? { durationMs: Number(event.durationMs) } : {}) });
+        }
+      });
+      const lines: string[] = [];
+      const error = console.error;
+      console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); };
+      const heldMs = 25;
+      let starting: Promise<unknown> | undefined;
+      try {
+        starting = (framework as unknown as { startAgentStream(a: typeof agent): Promise<unknown> }).startAgentStream(agent);
+        await entered.opened;
+        if (how === 'stopping') await framework.stop();
+        else if (how === 'removed') internal.agents.delete('assistant');
+        else if (how === 'disposed') {
+          // As runEphemeralToCompletion's disposal does: the agent, its record and its token go.
+          internal.agents.delete('assistant');
+          internal.lastInferenceAt.delete('assistant');
+          internal.activeTurnTokens.delete('assistant');
+        } else {
+          internal.agents.set('assistant', Object.create(agent));
+          if (how === 'superseded') internal.activeTurnTokens.set('assistant', -1);
+        }
+        await new Promise((resolve) => setTimeout(resolve, heldMs));
+        held.open();
+        await starting;
+      } finally { console.error = error; held.open(); await starting; }
+      try {
+        const why = how === 'stopping' ? 'the framework was stopping' : 'the agent was replaced or removed';
+        const reason = `a turn ${phase === 'before' ? 'assembled' : 'compiled'} while ${why} is dropped before it sends anything`;
+        assert.ok(lines.some((line) => line.includes(reason)), lines.join('\n'));
+        const last = internal.lastInferenceAt.get('assistant');
+        if (how === 'superseded') {
+          // The terminal event and lastInference carry only the name, so they
+          // would end the newer turn: the dropped frame emits and records nothing.
+          assert.deepEqual(turnEvents.map((event) => event.type), ['inference:started']);
+          assert.equal(last?.endedAt, undefined);
+        } else {
+          assert.deepEqual(turnEvents.map((event) => event.type), ['inference:started', 'inference:aborted']);
+          assert.equal(turnEvents[1]?.reason, reason);
+          const durationMs = turnEvents[1]?.durationMs ?? NaN;
+          assert.ok(Number.isFinite(durationMs) && durationMs >= heldMs - 5 && durationMs < 60_000, `durationMs ${durationMs}`);
+          if (how === 'disposed') {
+            // The terminal event still closes a listener's bracket, but the
+            // disposed name gets no record back.
+            assert.equal(internal.lastInferenceAt.has('assistant'), false, JSON.stringify(last));
+          } else {
+            assert.ok(last?.endedAt !== undefined && last.endedAt >= (last.startedAt ?? Infinity), JSON.stringify(last));
+          }
+        }
+      } finally {
+        if (how === 'superseded') internal.activeTurnTokens.delete('assistant');
+        if (how !== 'stopping') internal.agents.set('assistant', agent);
+        await framework.stop(); store.close();
+      }
+    });
+  }
+}
+
 // Drain async storage/annotation continuations without advancing deadline time.
 async function flushAsync() {
   for (let i = 0; i < 8; i++) await new Promise<void>((resolve) => setImmediate(resolve));