Channel state replacement can read the previous persisted snapshot before flush

[MicroMilo]

Summary

With an existing old file, save(new) followed immediately by load and replacement still uses the old session/project. After the debounce delay or explicit flush, the new state is read. The replacement path has no flush/read-after-save barrier.

Expected behavior

An immediate replacement after save(new) must read the latest session/project, or explicitly report that state is not durable yet. It must not silently use the old disk snapshot.

Actual behavior

With an existing old file, save(new) followed immediately by load and replacement still uses the old session/project. After the debounce delay or explicit flush, the new state is read. The replacement path has no flush/read-after-save barrier.

Impact

During hot reload or replacement immediately after a session/project change, the next message can be routed to stale context for a short but externally visible window.

Reproduction

Using the channel state persistence implementation, start with an existing on-disk old session/project, save a new session/project, and immediately trigger the replacement/load path before the debounce flush. Then repeat after waiting for the debounce or forcing a flush. The immediate replacement should read the new state or report that it is not durable; the observed result is the old snapshot until the later flush.

Minimal reproduction script

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();

Relevant source locations

• src/adapters/channel/protocol/ChannelStatePersistence.ts:29-81
• src/cli/pilotdeck.ts:263-280
• src/cli/pilotdeck.ts:290-313
• src/cli/pilotdeckServer.ts:90-125
• src/adapters/channel/feishu/FeishuSessionMapper.ts:57-85

Suggested direction

Make the external-input path establish one durable, identity-bound state/receipt before returning success; propagate explicit terminal outcomes to every channel and client; and add a regression test for the reproduced boundary.

This report is about functional behavior, not security. The reproduction uses deterministic in-memory or isolated fixtures and contains no credentials or private data.