scripts/verify-durability.mjs
Rama main · réplica de lectura del repositorio original.
#!/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);