#!/usr/bin/env node
/**
 * Restart-durability verification via file-backed `src/db.node.ts`.
 *
 * Proves conversation state survives a process restart. The app persists the
 * agent transcript to SQLite (db.node.ts), so a second turn in the same thread —
 * served by a BRAND NEW process on the same DB file — replays the first turn's
 * assistant reply from durable storage.
 *
 * Flow (all offline, net-guarded, stub provider):
 *   1. server1 on DB_A: T1 signed mention, stub replyText = DURABILITY_MARKER.
 *      SIGKILL server1.
 *   2. server2 on DB_A (fresh process, same DB): T2 signed mention in the SAME
 *      thread. Assert (i) T2 delivers a final on the wire; (ii) T2's provider
 *      request replays the marker (T1's assistant reply, loaded from the DB).
 *   3. NEGATIVE CONTROL — server3 on a DIFFERENT fresh DB_B: the same follow-up
 *      turn's provider request must NOT contain the marker (no shared durable
 *      storage → no replay). This proves the assertion measures durability.
 *   4. DURABLE CLAIMS + THREAD REGISTRY (SqliteSlackStateStore, sibling
 *      `<db>.state` file): on yet another fresh process sharing DB_A,
 *      (i) a byte-identical redelivery of T1's event_id posts NO new final;
 *      (ii) a new-event_id message with T1's (channel, ts) posts NO new final;
 *      (iii) an implicit (mention-free) thread reply IS admitted and answered —
 *      the joined thread survived the restart. Negative control on DB_B: the
 *      same implicit reply produces nothing (thread never started there).
 *   5. Run the focused live-Agent helper: edit Agent config after a thread's
 *      first turn, restart, and prove the next reply uses the current Agent
 *      config. Agent edits intentionally apply everywhere on the next message.
 *
 * Run with Node 24.x (see .nvmrc):
 *   node scripts/verify-durability.mjs
 */
import { spawnSync } from 'node:child_process';
import { existsSync, mkdtempSync, readFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';

import {
  REPO_ROOT,
  assertNodeVersion,
  buildNodeServer,
  delay,
  loadFake,
  loadTsModule,
  postSignedEvent,
  seedOfflineDemoChannelConfig,
  seedOfflineSlackAuthority,
  spawnReadyServer,
  stopChild,
  waitForFinals,
} from './lib/offline-harness.mjs';

const DURABILITY_MARKER = 'DURABILITY_MARKER_ALPHA';
const EXEC_CHANNEL = 'C_EXEC';
const ROOT_TS = '1782770400.000100';
// Negative checks wait until Slack/provider traffic has been idle for
// quietMs. T1 calibrates it on a cold server: a wrongly admitted turn must be
// able to reach the fake well inside the window, so a slow host widens it.
const MIN_QUIET_MS = 1500;
let quietMs = MIN_QUIET_MS;

function log(line) {
  console.log(line);
}

const results = [];
function record(name, passed, detail) {
  results.push({ name, passed });
  log(`${passed ? 'PASS' : 'FAIL'}  ${name}${detail ? ` — ${detail}` : ''}`);
}

/** A signed mention-free channel thread reply (implicit continuation). */
function threadReply({ eventId, ts, threadTs }) {
  return {
    token: 'verification-token-not-a-secret',
    team_id: 'TDEMO',
    api_app_id: 'ADEMO',
    event_id: eventId,
    event_time: 1782770400,
    type: 'event_callback',
    event: {
      type: 'message',
      channel_type: 'channel',
      user: 'U_ALICE',
      text: 'and what changed since the summary?',
      ts,
      channel: EXEC_CHANNEL,
      event_ts: ts,
      thread_ts: threadTs,
    },
  };
}

/** A signed app_mention in C_EXEC. `threadTs` set → threaded follow-up (same key). */
function mention({ eventId, ts, threadTs }) {
  return {
    token: 'verification-token-not-a-secret',
    team_id: 'TDEMO',
    api_app_id: 'ADEMO',
    event_id: eventId,
    event_time: 1782770400,
    type: 'event_callback',
    event: {
      type: 'app_mention',
      user: 'U_ALICE',
      text: '<@UBOT> please use channel context and draft an exec summary',
      ts,
      channel: EXEC_CHANNEL,
      event_ts: ts,
      ...(threadTs ? { thread_ts: threadTs } : {}),
    },
  };
}

async function runServerTurn({ serverEntry, fakeUrl, dbPath, netGuardLog, payload }) {
  // Each restart probe takes a fresh port; a port lost to another process
  // between allocation and bind is retried rather than failing the gate.
  const { child, eventsUrl, getOutput } = await spawnReadyServer({
    serverEntry,
    fakeUrl,
    netGuardLog,
    env: {
      TAG_DB_PATH: dbPath,
      // Pin the state DB to the seeded `${dbPath}.state` explicitly: spawnServer
      // forwards ambient process.env, so an exported SLACK_STATE_DB_PATH (e.g.
      // from a live-Slack shell) would otherwise redirect every server here to
      // the operator's live state store and merge the three "isolated" DBs.
      SLACK_STATE_DB_PATH: `${dbPath}.state`,
      CHICKPEA_CREDENTIAL_KEYRING_PATH: `${dbPath}.credential-keyring.json`,
      CHICKPEA_AUTH_SECRET: '9d'.repeat(32),
      SLACK_TAG_MODEL: 'local-stub/parity-stub-1',
    },
  });
  try {
    await postSignedEvent(eventsUrl, payload);
  } catch (error) {
    await stopChild(child);
    throw error;
  }
  return { child, eventsUrl, getOutput };
}

/** Milliseconds from now until the fake sees a call past `index`, or `timeoutMs`. */
async function firstWireAfter(index, timeoutMs = 15_000) {
  const startedAt = Date.now();
  while (backend.wireLog.length <= index) {
    if (Date.now() - startedAt > timeoutMs) return timeoutMs;
    await delay(5);
  }
  return Date.now() - startedAt;
}

// Load every TypeScript dependency before the restart probes begin. Registering
// tsx after repeatedly spawning and SIGKILLing server processes can leave its
// esbuild loader waiting indefinitely on Linux; module loading is setup, not
// part of the durability behavior this harness is meant to exercise.
const { FakeSlackBackend } = await loadFake();
const { SqliteMemoryStateStore } = await loadTsModule('src/memory/store.ts');
const { SqliteConfigStore } = await loadTsModule('src/config/store.ts');
const { agentRoutineDeletion, SqliteRoutineStore } = await loadTsModule('src/routines/store.ts');
const { hashRoutineValue } = await loadTsModule('src/routines/ids.ts');
const backend = new FakeSlackBackend({
  slack: {
    identity: { appId: 'ADEMO', teamId: 'TDEMO', botUserId: 'UBOT' },
    channels: [{ id: EXEC_CHANNEL, name: 'exec', isMember: true }],
    channelMembers: { [EXEC_CHANNEL]: ['U_ALICE', 'UBOT'] },
    workspaceUsers: [
      { id: 'U_ALICE', teamId: 'TDEMO' },
      { id: 'UBOT', teamId: 'TDEMO', isBot: true, isAppUser: true },
    ],
  },
  provider: { mode: 'ok', replyText: DURABILITY_MARKER },
});
const fake = await backend.listen();
log(`fake backend listening at ${fake.url}`);

const netGuardLog = join(mkdtempSync(join(tmpdir(), 'flue-dur-guard-')), 'external-hosts.log');
const dbA = join(mkdtempSync(join(tmpdir(), 'flue-dur-dbA-')), 'flue.db');
const dbB = join(mkdtempSync(join(tmpdir(), 'flue-dur-dbB-')), 'flue.db');
// dbC is reserved for the thread-registry negative control: no turn ever runs
// on it before the implicit reply (dbB already saw a mention with ROOT_TS as
// its thread key, which would legitimately register the thread there).
const dbC = join(mkdtempSync(join(tmpdir(), 'flue-dur-dbC-')), 'flue.db');

try {
  const serverEntry = await buildNodeServer();
  log(`built node server: ${serverEntry}`);
  log(`node ${assertNodeVersion()}  DB_A=${dbA}  DB_B=${dbB}`);
  await seedOfflineDemoChannelConfig(`${dbA}.state`);
  await seedOfflineDemoChannelConfig(`${dbB}.state`);
  await seedOfflineDemoChannelConfig(`${dbC}.state`);
  for (const dbPath of [dbA, dbB, dbC]) {
    await seedOfflineSlackAuthority({
      stateDbPath: `${dbPath}.state`,
      keyringPath: `${dbPath}.credential-keyring.json`,
      canonicalOrigin: 'http://127.0.0.1',
      teamId: 'TDEMO',
      slackUserId: 'UALICE01',
      appId: 'ADEMO',
      botUserId: 'UBOT',
    });
  }

  // --- Turn 1 on DB_A, then kill the process. ---
  {
    const { child, getOutput } = await runServerTurn({
      serverEntry,
      fakeUrl: fake.url,
      dbPath: dbA,
      netGuardLog,
      payload: mention({ eventId: 'Ev_DUR_T1', ts: ROOT_TS }),
    });
    // No I/O runs between the ack resolving above and this line.
    const firstWire = firstWireAfter(backend.wireLog.length);
    const finals = await waitForFinals(backend, 1, 15_000);
    const t1Final = finals.at(-1);
    const firstWireMs = await firstWire;
    quietMs = Math.max(MIN_QUIET_MS, 3 * firstWireMs);
    log(`• cold-server ack to first fake call: ${firstWireMs}ms; negative checks wait for ${quietMs}ms of quiet`);
    // This case proves restart after a completed turn. A Slack HTTP effect is
    // visible before its local receipt is committed; killing at that boundary
    // instead tests ambiguous delivery recovery and can strand T2 behind T1.
    // Wait for the receipt, not an arbitrary sleep or another model response.
    const state = new DatabaseSync(`${dbA}.state`, { readOnly: true });
    let delivered = false;
    try {
      const deadline = Date.now() + 15_000;
      while (Date.now() < deadline) {
        const job = state.prepare('SELECT delivered FROM turn_jobs WHERE id = ?')
          .get(`msg:${EXEC_CHANNEL}:${ROOT_TS}`);
        if (job?.delivered === 1) { delivered = true; break; }
        await delay(25);
      }
    } finally {
      state.close();
      await stopChild(child);
    }
    if (!delivered) throw new Error(`T1 delivery receipt was not persisted:\n${getOutput()}`);
    record(
      'T1 delivers a final carrying the durability marker, then server SIGKILLed',
      finals.length === 1 && !!t1Final && t1Final.text.includes(DURABILITY_MARKER),
      `finals=${finals.length} markerInFinal=${!!t1Final && t1Final.text.includes(DURABILITY_MARKER)}`,
    );
  }

  // --- Turn 2 on DB_A: fresh process, same DB. Marker must replay. ---
  backend.reset();
  {
    const { child, getOutput } = await runServerTurn({
      serverEntry,
      fakeUrl: fake.url,
      dbPath: dbA,
      netGuardLog,
      payload: mention({ eventId: 'Ev_DUR_T2', ts: '1782770500.000100', threadTs: ROOT_TS }),
    });
    const finals = await waitForFinals(backend, 1, 15_000);
    const t2Final = finals.at(-1);
    if (finals.length === 0) log(`T2 server output:\n${getOutput()}`);

    const providerCalls = backend.providerCalls();
    const providerReplaysMarker = providerCalls.some((call) =>
      JSON.stringify(call.body).includes(DURABILITY_MARKER),
    );

    await stopChild(child);

    record(
      'T2 (new process, same DB) delivers a final',
      finals.length === 1 && !!t2Final,
      `finals=${finals.length}`,
    );
    record(
      'T2 provider request REPLAYS the marker from durable storage',
      providerReplaysMarker,
      `providerCalls=${providerCalls.length} replaysMarker=${providerReplaysMarker}`,
    );
  }

  // --- Negative control: follow-up on a DIFFERENT fresh DB → no replay. ---
  backend.reset();
  {
    const { child } = await runServerTurn({
      serverEntry,
      fakeUrl: fake.url,
      dbPath: dbB,
      netGuardLog,
      payload: mention({ eventId: 'Ev_DUR_NEG', ts: '1782770600.000100', threadTs: ROOT_TS }),
    });
    const finals = await waitForFinals(backend, 1, 15_000);
    const providerCalls = backend.providerCalls();
    const providerHasMarker = providerCalls.some((call) =>
      JSON.stringify(call.body).includes(DURABILITY_MARKER),
    );
    await stopChild(child);
    record(
      'NEGATIVE CONTROL: fresh DB → provider request does NOT contain the marker',
      finals.length === 1 && !providerHasMarker,
      `finals=${finals.length} markerLeaked=${providerHasMarker}`,
    );
  }

  // --- Durable claims + registry: yet another fresh process on DB_A. ---
  backend.reset();
  {
    const { child, eventsUrl, getOutput } = await runServerTurn({
      serverEntry,
      fakeUrl: fake.url,
      dbPath: dbA,
      netGuardLog,
      // Byte-identical redelivery of T1's event (same event_id, same ts).
      payload: mention({ eventId: 'Ev_DUR_T1', ts: ROOT_TS }),
    });
    await backend.quiesce(quietMs, quietMs + 15_000);
    const afterRedelivery = backend.finals().length;
    record(
      'DURABLE CLAIMS: redelivered event_id after restart posts NO new final',
      afterRedelivery === 0,
      `finals=${afterRedelivery}`,
    );

    // New event_id, same (channel, message-ts): the msg: claim must hold.
    const twin = await postSignedEvent(eventsUrl, mention({ eventId: 'Ev_DUR_TWIN', ts: ROOT_TS }));
    await backend.quiesce(quietMs, quietMs + 15_000);
    const afterTwin = backend.finals().length;
    record(
      'DURABLE CLAIMS: new event_id with the same (channel, ts) posts NO new final',
      twin.status === 200 && afterTwin === 0,
      `ackStatus=${twin.status} finals=${afterTwin}`,
    );

    // Mention-free thread reply: the durable registry admits it post-restart.
    await postSignedEvent(
      eventsUrl,
      threadReply({ eventId: 'Ev_DUR_IMPL', ts: '1782770700.000100', threadTs: ROOT_TS }),
    );
    const implicitFinals = await waitForFinals(backend, 1, 15_000);
    if (implicitFinals.length === 0) log(`registry server output:\n${getOutput()}`);
    await stopChild(child);
    record(
      'DURABLE REGISTRY: implicit thread reply IS admitted after restart (one final)',
      implicitFinals.length === 1,
      `finals=${implicitFinals.length}`,
    );
  }

  // --- Registry negative control: implicit reply on untouched DB_C → silence. ---
  backend.reset();
  {
    const { child } = await runServerTurn({
      serverEntry,
      fakeUrl: fake.url,
      dbPath: dbC,
      netGuardLog,
      payload: threadReply({ eventId: 'Ev_DUR_IMPL_NEG', ts: '1782770800.000100', threadTs: ROOT_TS }),
    });
    await backend.quiesce(quietMs, quietMs + 15_000);
    const finals = backend.finals().length;
    await stopChild(child);
    record(
      'NEGATIVE CONTROL: implicit reply on a DB whose thread never started posts NO final',
      finals === 0,
      `finals=${finals}`,
    );
  }

  // --- One-body Agent memory: restart, compatibility open, and scrub. ---
  {
    const memoryPath = `${dbA}.state`;
    const memorySentinel = 'MEMORY_DURABILITY_SENTINEL_ALPHA';
    const priorLegacyFlag = process.env.SLACK_TAG_MEMORY_ENABLED;
    process.env.SLACK_TAG_MEMORY_ENABLED = 'false';
    let memoryStore;
    try {
      memoryStore = new SqliteMemoryStateStore(memoryPath);
      await memoryStore.putAgentMemory({
        agentId: 'agent_default',
        body: memorySentinel,
        expectedRevision: 0,
      });
      memoryStore.close();
      memoryStore = undefined;

      // A config-only open stands in for rollback code that knows nothing about
      // the additive memory tables: it must boot and leave them untouched.
      const configOnly = new SqliteConfigStore(memoryPath, agentRoutineDeletion, { agents: [], grants: [] });
      await configOnly.listAgents();
      configOnly.close();

      memoryStore = new SqliteMemoryStateStore(memoryPath);
      const restarted = await memoryStore.getAgentMemory('agent_default');
      record(
        'MEMORY DURABILITY: Agent body survives close/restart and legacy false is inert',
        restarted.body === memorySentinel && restarted.revision === 1,
        `revision=${String(restarted.revision)}`,
      );
      await memoryStore.putAgentMemory({
        agentId: 'agent_default',
        body: '',
        expectedRevision: 1,
      });
      memoryStore.close();
      memoryStore = undefined;

      memoryStore = new SqliteMemoryStateStore(memoryPath);
      const forgotten = await memoryStore.getAgentMemory('agent_default');
      memoryStore.close();
      memoryStore = undefined;
      record(
        'MEMORY DURABILITY: clearing the Agent body survives restart without history',
        forgotten.body === '' && forgotten.revision === 2,
        `revision=${String(forgotten.revision)}`,
      );
      record(
        'MEMORY DURABILITY: deleted sentinel is absent from the raw state file',
        !readFileSync(memoryPath).toString('latin1').includes(memorySentinel),
      );
    } finally {
      memoryStore?.close();
      if (priorLegacyFlag === undefined) delete process.env.SLACK_TAG_MEMORY_ENABLED;
      else process.env.SLACK_TAG_MEMORY_ENABLED = priorLegacyFlag;
    }
  }

  // --- Scheduled work: definition/run/idempotency survive a state restart. ---
  {
    const routinePath = join(mkdtempSync(join(tmpdir(), 'flue-routine-dur-')), 'state.db');
    const routineNow = Date.now();
    const tokenHash = hashRoutineValue('routine-durability-confirmation');
    const definition = {
      name: 'Durability routine',
      description: 'Proves scheduled-work restart behavior.',
      taskText: 'Inspect current state and update the disposable acceptance record.',
      triggerKind: 'schedule',
      scheduleInput: '0 * * * *',
      scheduleJson: JSON.stringify({ version: 1, kind: 'cron', expression: '0 * * * *' }),
      timezone: 'UTC',
      outputPolicy: 'post',
      authorityMode: 'live_channel_v1',
    };
    const draft = {
      action: 'create',
      routineId: 'routine_durability',
      definition,
      nextRunAt: routineNow + 60 * 60_000,
      projectedDailyStarts: 24,
      reservations: [{ windowStart: routineNow + 60 * 60_000, count: 1 }],
    };
    const previewHash = hashRoutineValue(JSON.stringify(draft));
    let routineStore = new SqliteRoutineStore(routinePath, () => routineNow);
    try {
      await routineStore.putConfirmation({
        confirmationId: 'rconfirm_durability',
        tokenHash,
        actorId: 'U_ALICE',
        actorClass: 'member',
        workspaceId: 'TDEMO',
        channelId: EXEC_CHANNEL,
        draft,
        previewHash,
        expiresAt: routineNow + 15 * 60_000,
      });
      const routine = await routineStore.confirm({
        tokenHash,
        actorId: 'U_ALICE',
        workspaceId: 'TDEMO',
        channelId: EXEC_CHANNEL,
        previewHash,
        idempotencyKey: 'routine:durability:confirm',
      });
      const occurrenceInput = {
        runId: 'rrun_durability',
        idempotencyKey: 'routine:durability:occurrence',
        routineId: routine.id,
        routineVersion: routine.version,
        scheduledFor: routineNow + 60 * 60_000,
        triggerSource: 'schedule',
        requestedBy: null,
        queuedAt: routineNow,
        deadlineAt: routineNow + 15 * 60_000,
      };
      await routineStore.createOccurrence(occurrenceInput);
      routineStore.close();

      routineStore = new SqliteRoutineStore(routinePath, () => routineNow);
      const restarted = await routineStore.getRoutine(routine.id);
      const replay = await routineStore.createOccurrence(occurrenceInput);
      const runs = await routineStore.listRuns({ routineId: routine.id });
      record(
        'ROUTINE DURABILITY: definition and occurrence survive restart with one idempotent run',
        restarted?.taskText === definition.taskText && replay.id === 'rrun_durability' && runs.length === 1,
        `routine=${String(restarted?.id)} run=${replay.id} runs=${runs.length}`,
      );
    } finally {
      routineStore.close();
    }
  }

  // --- Net guard: zero external traffic. ---
  {
    const attempted = existsSync(netGuardLog) ? readFileSync(netGuardLog, 'utf8').trim() : '';
    record('NET_GUARD_LOG empty -> zero external traffic', attempted === '', attempted || 'none');
  }
} catch (error) {
  record('durability harness', false, error instanceof Error ? error.stack : String(error));
} finally {
  await backend.close();
}

log('\nRunning live-Agent config restart verification...');
const snapshotRun = spawnSync(
  process.execPath,
  [join(REPO_ROOT, 'scripts', 'verify-snapshot-durability.mjs')],
  {
    cwd: REPO_ROOT,
    env: process.env,
    stdio: 'inherit',
  },
);
record(
  'LIVE AGENT CONFIG: edited instructions survive restart and reach the same thread',
  snapshotRun.status === 0,
  snapshotRun.error
    ? snapshotRun.error.message
    : `exit=${snapshotRun.status ?? 'signal'}${snapshotRun.signal ? ` signal=${snapshotRun.signal}` : ''}`,
);

const failed = results.filter((result) => !result.passed);
log(`\n${results.length - failed.length}/${results.length} checks passed`);
process.exit(failed.length === 0 ? 0 : 1);