After an error-policy retry, the failed stream's teardown stops the retry's typing indicator, and nothing shows it again #650

// Probe (Imogen-2490): a turn's typing indicator through an error-policy
// retry. Stream 1 fails at its end; the policy retries; stream 2 (the
// successor) streams for 400 ms. Logs each startTyping/stopTyping with what
// was running then. Harness as test/present-while-acting.test.ts's.
// Run with tsx from the agent-framework checkout's root.
import { tmpdir } from 'node:os';
import { mkdtempSync, rmSync } from 'node:fs';
import { join } from 'node:path';
import { AgentFramework } from '../src/index.js';
import type { Module, ModuleContext, ProcessEvent, ProcessState, EventResponse, ToolDefinition, ToolCall, ToolResult } from '../src/index.js';
import { MockMembrane, createMockResponse } from '../test/helpers/mock-membrane.js';
import type { ContentBlock } from '@animalabs/membrane';

console.log(`node ${process.version}, ${process.platform}`);

class Chat implements Module {
  readonly name = 'chat';
  async start(_ctx: ModuleContext): Promise<void> {}
  async stop(): Promise<void> {}
  getTools(): ToolDefinition[] { return []; }
  async handleToolCall(_call: ToolCall): Promise<ToolResult> { return { success: true }; }
  async onProcess(event: ProcessEvent, _state: ProcessState): Promise<EventResponse> {
    if (event.type !== 'external-message') return {};
    const text = String((event as { content?: unknown }).content);
    return { addMessages: [{ participant: 'Antra', content: [{ type: 'text', text }], metadata: { channelId: 'chan-1' } }], requestInference: text === 'go' };
  }
}

const t0 = Date.now();
const at = () => `${String(Date.now() - t0).padStart(4)} ms`;
const log: string[] = [];
let phase = 'before any stream';
let stoppedDuringStream2 = false;
let restartedAfterStop = false;

const tempDir = mkdtempSync(join(tmpdir(), 'f286-'));
const membrane = new MockMembrane();
membrane.pushResponse(createMockResponse([{ type: 'text', text: 'First try.' }] as ContentBlock[]));

// Stream 1 ends with an error instead of completing; stream 2 waits 400 ms before streaming.
const makeStream = membrane.streamYielding.bind(membrane);
let streams = 0;
membrane.streamYielding = (request, options) => {
  const stream = makeStream(request, options);
  const n = ++streams;
  const iterate = stream[Symbol.asyncIterator].bind(stream);
  stream[Symbol.asyncIterator] = async function* () {
    if (n === 2) {
      phase = 'stream 2 streaming';
      log.push(`${at()} stream 2 begins, and streams for 400 ms`);
      await new Promise((r) => setTimeout(r, 400));
    }
    for await (const event of { [Symbol.asyncIterator]: iterate }) {
      if (event.type === 'complete' && n === 1) {
        phase = 'stream 1 failed, before its retry';
        log.push(`${at()} stream 1 fails`);
        yield { type: 'error' as const, error: new Error('synthetic transient stream error') };
        return;
      }
      yield event;
    }
    if (n === 2) { log.push(`${at()} stream 2 ends`); phase = 'stream 2 ended'; }
  };
  return stream;
};

const framework = await AgentFramework.create({
  storePath: join(tempDir, 'test.chronicle'),
  membrane: membrane.asMembrane(),
  agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'You are helpful.' }],
  modules: [new Chat()],
});

let starts = 0;
framework.onTrace((event) => {
  if (event.type === 'inference:started') {
    starts++;
    phase = `stream ${starts} started`;
    log.push(`${at()} inference:started #${starts}`);
    if (starts === 2) membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Answering after the retry.' }] as ContentBlock[]));
  }
  if (event.type === 'inference:failed' || event.type === 'inference:completed') log.push(`${at()} ${event.type}`);
});

// The channel registry, with typing recorded and a fixed locus.
const explicit: Record<string, unknown> = {
  resolveLocus: () => 'chan-1',
  routeSpeech: async (_agent: string, text: string, locus?: string | null) => { log.push(`${at()} speech routed to ${locus}: ${JSON.stringify(text)}`); return locus ? { delivered: true, channelId: locus } : null; },
  getDefaultPublishChannel: () => null,
  isChannelOpen: () => true,
  getDescriptor: () => undefined,
  getChannelTools: () => [],
  startTyping: (channelId: string) => {
    log.push(`${at()} startTyping(${channelId})   [${phase}]`);
    if (stoppedDuringStream2 && phase === 'stream 2 streaming') restartedAfterStop = true;
  },
  stopTyping: (channelId?: string) => {
    // The caller, from the stack: the first frame in framework.ts, with its line.
    const caller = (new Error().stack ?? '').split('\n').find((l) => l.includes('src/framework.ts'))?.trim() ?? 'unknown';
    log.push(`${at()} stopTyping(${channelId ?? 'every channel'})   [${phase}]   called ${caller.replace(/\(.*\/src\//, '(src/')}`);
    if (phase === 'stream 2 streaming') stoppedDuringStream2 = true;
  },
};
(framework as unknown as { channelRegistry: unknown }).channelRegistry = new Proxy(explicit, {
  get: (target, prop: string) => (prop in target ? target[prop] : () => undefined),
});

framework.pushEvent({ type: 'external-message', source: 'test', content: 'go', metadata: {} } as unknown as ProcessEvent);
await framework.runUntilIdle();
await new Promise((r) => setTimeout(r, 800));
await framework.runUntilIdle();
console.log(`streams: ${streams}, provider calls: ${membrane.calls.length}`);
for (const line of log) console.log(line);
console.log(`typing stopped while stream 2 streamed: ${stoppedDuringStream2}; started again before stream 2 ended: ${restartedAfterStop}`);
await framework.stop();
rmSync(tempDir, { recursive: true, force: true });
process.exit(0);