Channel state replacement can read the previous persisted snapshot before flush

From the repository root, save this as repro_channel_state_replacement.mts and run:

pnpm install --frozen-lockfile
pnpm exec tsx repro_channel_state_replacement.mts

import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { ChannelStatePersistence } from "./src/adapters/channel/protocol/ChannelStatePersistence.js";
import { FeishuSessionMapper } from "./src/adapters/channel/feishu/FeishuSessionMapper.js";

type FixtureState = {
  activeByChatId: Record<string, string>;
  projectByChatId: Record<string, string>;
};

const channelKey = "feishu";
const chatId = "witness-chat";

function replacementFromLoadedState(state: FixtureState | undefined): {
  sessionKey: string;
  projectKey: string | null;
} {
  const mapper = new FeishuSessionMapper(
    state ?? { activeByChatId: {}, projectByChatId: {} },
    () => "fixture-unused",
  );
  const resolved = mapper.resolve({ chatId, text: "continuation" });
  return {
    sessionKey: resolved.sessionKey,
    projectKey: resolved.projectKey ?? null,
  };
}

async function fileSnapshot(stateDir: string): Promise<FixtureState | null> {
  try {
    return JSON.parse(await readFile(join(stateDir, `${channelKey}.state.json`), "utf8")) as FixtureState;
  } catch {
    return null;
  }
}

async function loadReplacement(
  persistence: ChannelStatePersistence,
): Promise<{ loaded: FixtureState | null; replacement: ReturnType<typeof replacementFromLoadedState> }> {
  const loaded = (await persistence.load<FixtureState>(channelKey)) ?? null;
  return { loaded, replacement: replacementFromLoadedState(loaded ?? undefined) };
}

function state(sessionSuffix: string, project: string): FixtureState {
  return {
    activeByChatId: { [chatId]: `feishu:chat=${chatId}:${sessionSuffix}` },
    projectByChatId: { [chatId]: project },
  };
}

async function main(): Promise<void> {
  const root = await mkdtemp(join(tmpdir(), "pilotdeck-witness-stale-replacement-"));
  const stateDir = join(root, "channels");
  const persistence = new ChannelStatePersistence({ stateDir, debounceMs: 40 });
  const oldState = state("s_fixture-old", "fixture-project-old");
  const newState = state("s_fixture-new", "fixture-project-new");

  try {
    // Establish the old snapshot as the only durable state.
    persistence.save(channelKey, oldState);
    await persistence.flush();
    const durableOld = await fileSnapshot(stateDir);

    // This models mapper.resolve -> onStateChange -> persistence.save, followed
    // immediately by PilotDeck's load -> new Mapper(saved) replacement path.
    persistence.save(channelKey, newState);
    const immediate = await loadReplacement(persistence);
    const immediateFile = await fileSnapshot(stateDir);

    // Control: once the normal debounce has completed, the same replacement
    // path observes the new snapshot without an explicit flush call.
    await new Promise<void>((resolve) => setTimeout(resolve, 80));
    const afterDebounce = await loadReplacement(persistence);
    const afterDebounceFile = await fileSnapshot(stateDir);

    // Control: an explicit per-process flush also makes the new state visible.
    const thirdState = state("s_fixture-third", "fixture-project-third");
    persistence.save(channelKey, thirdState);
    await persistence.flush();
    const afterExplicitFlush = await loadReplacement(persistence);
    const afterExplicitFlushFile = await fileSnapshot(stateDir);

    console.log(JSON.stringify({
      fixture: "channel-state-stale-replacement-read",
      transport: "in-process ChannelStatePersistence and FeishuSessionMapper; no provider, socket, or external network",
      timing: {
        debounceMs: 40,
        immediateReplacement: "save(new) followed by load/replacement in the same turn",
        debouncedReplacement: "load/replacement after 80ms",
        explicitFlushReplacement: "save(third), flush(), then load/replacement",
      },
      durableOld: {
        fileSession: durableOld?.activeByChatId[chatId] ?? null,
        fileProject: durableOld?.projectByChatId[chatId] ?? null,
        replacement: replacementFromLoadedState(durableOld ?? undefined),
      },
      immediate: {
        loadedSession: immediate.loaded?.activeByChatId[chatId] ?? null,
        loadedProject: immediate.loaded?.projectByChatId[chatId] ?? null,
        fileSession: immediateFile?.activeByChatId[chatId] ?? null,
        fileProject: immediateFile?.projectByChatId[chatId] ?? null,
        replacement: immediate.replacement,
      },
      afterDebounce: {
        loadedSession: afterDebounce.loaded?.activeByChatId[chatId] ?? null,
        loadedProject: afterDebounce.loaded?.projectByChatId[chatId] ?? null,
        fileSession: afterDebounceFile?.activeByChatId[chatId] ?? null,
        fileProject: afterDebounceFile?.projectByChatId[chatId] ?? null,
        replacement: afterDebounce.replacement,
      },
      afterExplicitFlush: {
        loadedSession: afterExplicitFlush.loaded?.activeByChatId[chatId] ?? null,
        loadedProject: afterExplicitFlush.loaded?.projectByChatId[chatId] ?? null,
        fileSession: afterExplicitFlushFile?.activeByChatId[chatId] ?? null,
        fileProject: afterExplicitFlushFile?.projectByChatId[chatId] ?? null,
        replacement: afterExplicitFlush.replacement,
      },
    }, null, 2));
  } finally {
    await rm(root, { recursive: true, force: true });
  }
}

await main();