Retries and other continuing starts discard hybrid privacy holds and sticky destinations #314

  // A framework error retry starts the turn's routing afresh, so what an earlier
  // stream of the same logical turn set (a sticky destination, hybrid's
  // >>>skip_reply hold) no longer applies.
  for (const [label, mode, first, expected] of [
    ['hybrid >>>skip_reply keeps later unprefixed prose private', 'hybrid', '>>>skip_reply an aside', []],
    ['hybrid >>>#alpha is sticky', 'hybrid', '>>>#alpha first', [{ text: 'first', locus: 'chan-alpha' }, { text: 'carrying on', locus: 'chan-alpha' }]],
    ['explicit >>#alpha is sticky', 'explicit', '>>#alpha first', [{ text: 'first', locus: 'chan-alpha' }, { text: 'carrying on', locus: 'chan-alpha' }]],
  ] as const) {
  it(`repro: across an error retry, ${label}`, async () => {
    membrane.pushResponse(createMockResponse([
      { type: 'text', text: first },
      { type: 'tool_use', id: 'retry-tool', name: 'robot--move', input: { dir: 'north' } },
    ] as ContentBlock[], 'tool_use'));
    membrane.pushResponse(createMockResponse([] as ContentBlock[]));
    const framework = await createFramework(mode);
    const routed = stubRegistry(framework);
    let starts = 0;
    framework.onTrace((event) => {
      if (event.type !== 'inference:started') return;
      starts++;
      // The retry carries on without a prefix, as a model continuing its turn does.
      if (starts === 2) membrane.pushResponse(createMockResponse([{ type: 'text', text: 'carrying on' }]));
      // A bounce wakes the resident; that turn stays quiet.
      if (starts >= 3) membrane.pushResponse(createMockResponse([{ type: 'text', text: mode === 'explicit' ? '>>skip_reply' : '>>>skip_reply' }]));
    });
    const makeStream = membrane.streamYielding.bind(membrane);
    let streams = 0;
    membrane.streamYielding = (request, options) => {
      const stream = makeStream(request, options);
      const index = ++streams;
      const iterate = stream[Symbol.asyncIterator].bind(stream);
      stream[Symbol.asyncIterator] = async function* () {
        for await (const event of { [Symbol.asyncIterator]: iterate }) {
          if (event.type === 'complete' && index === 1) {
            yield { type: 'error' as const, error: new Error('synthetic transient stream error') };
            return;
          }
          yield event;
        }
      };
      return stream;
    };
    try {
      trigger(framework);
      await within(framework.runUntilIdle(), 'the turn and its retry', 10000);
      assert.ok(starts >= 2, 'the error was retried');
      // Publication as the registry stub saw it; bounce notices by their kind.
      const { bounces } = context(framework);
      assert.deepEqual(routed.map((r) => ({ text: r.text.trim(), locus: r.locus })), expected);
      assert.equal(bounces, 0, 'nothing bounced');
    } finally {
      await framework.stop();
    }
  });
  }