channels: persistent intro-DM ledger for HTTP mode (H3 onboarding fix) (#119)

mcheemaa · web-flow · commit 3e37d5012aea · 2026-05-01T15:31:46.000-07:00
Production HTTP-mode tenants stayed at /health.onboarding="pending"
forever and could re-DM the installer on every process restart. Both
behaviours trace to the in-memory firstDmSent flag in SlackHttpChannel
plus the fact that markOnboardingStarted only runs from the Socket Mode
flow (gated on a channels.yaml that is never baked in HTTP-mode rootfs).

Add an optional IntroductionLedger surface on SlackHttpChannel that
production wires against the SQLite onboarding_state table. The
receiver short-circuits the intro DM when the ledger reports sent and
stamps it only AFTER a successful Slack send, so:

  - /health.onboarding flips to "complete" on the first successful DM
  - a process restart never re-DMs the installer
  - a transient Slack failure (rate limit, no ts returned, throw) leaves
    the row clear so the next process start retries

The ledger is optional; tests and single-tenant dev fall back to the
existing in-memory flag with the prior semantics. The SlackHttpChannel
constructor stays compatible with all existing callers.

Tests: 7 new ledger-path cases on slack-http-receiver plus a
slack-channel-factory regression that pins the introductionLedger
forward, all green. Full suite 2316 pass / 0 fail.

Refs: real-vs-fake audit v2 finding H3
diff --git a/src/channels/__tests__/slack-channel-factory.test.ts b/src/channels/__tests__/slack-channel-factory.test.ts
@@ -251,6 +251,43 @@ describe("createSlackChannel", () => {
 		expect(ch).toBeInstanceOf(SlackHttpChannel);
 	});
 
+	test("transport=http forwards introductionLedger to the SlackHttpChannel", async () => {
+		// H3 fix regression guard: the SQLite-backed intro-DM ledger must
+		// reach the channel constructor so connect() can short-circuit on
+		// a process restart and stamp /health=onboarding-complete on a
+		// successful send. A future refactor that drops the forward
+		// breaks this test loud; without it, a tenant whose first DM
+		// failed would silently re-fire (or never fire) on restart.
+		const idFetcher = { get: () => Promise.resolve(HTTP_IDENTITY) };
+		const secFetcher = makeSecretFetcher();
+		let isCalled = 0;
+		let markCalled = 0;
+		const ledger = {
+			isIntroSent: () => {
+				isCalled++;
+				return true;
+			},
+			markIntroSent: () => {
+				markCalled++;
+			},
+		};
+		const ch = await createSlackChannel({
+			transport: "http",
+			channelsConfig: null,
+			identityFetcher: idFetcher,
+			secretsFetcher: secFetcher,
+			introductionLedger: ledger,
+		});
+		// Calling connect() exercises the ledger path. The mocked Bolt
+		// client (loaded by this test's @slack/bolt mock) makes auth.test
+		// resolve immediately; isIntroSent() === true short-circuits the
+		// intro DM so we know the ledger reached the channel.
+		expect(ch).toBeInstanceOf(SlackHttpChannel);
+		await (ch as InstanceType<typeof SlackHttpChannel>).connect();
+		expect(isCalled).toBeGreaterThanOrEqual(1);
+		expect(markCalled).toBe(0);
+	});
+
 	test("makeSecretFetcher fails-loud when production asks for an unknown name", async () => {
 		// This is the audit Finding 1 regression guard. If the production code
 		// in slack-channel-factory.ts ever drifts to ask for a different name
diff --git a/src/channels/__tests__/slack-http-receiver.test.ts b/src/channels/__tests__/slack-http-receiver.test.ts
@@ -83,6 +83,7 @@ mock.module("@slack/bolt", () => ({
 
 // Import the channel AFTER the module mock so the constructor uses our doubles.
 const { SlackHttpChannel } = await import("../slack-http-receiver.ts");
+type IntroductionLedger = import("../slack-http-receiver.ts").IntroductionLedger;
 
 const SECRET = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
 const TEAM_ID = "T9TK3CUKW";
@@ -778,6 +779,142 @@ describe("synthetic first DM on connect", () => {
 	});
 });
 
+// ----- introductionLedger (persistent intro-DM idempotency) ----------------
+//
+// H3 root-cause fix: the in-memory `firstDmSent` flag lost its state on
+// every process restart, so a tenant whose first intro DM failed (or
+// silently never fired) had no record of the attempt. Phantom Cloud
+// production also lacked any signal in `/health` that the intro had
+// happened: `onboarding` stayed at "pending" because `markOnboardingStarted`
+// only ran from the Socket Mode `startOnboarding` path which is gated on a
+// `channels.yaml` that is never baked in HTTP-mode rootfs.
+//
+// The ledger surface fixes both legs in one place. Production wires it
+// against the SQLite `onboarding_state` table; these tests use a tiny
+// in-memory recorder that pins the contract:
+//   1. isIntroSent() === true short-circuits before any Slack call.
+//   2. markIntroSent() fires only AFTER a successful sendIntroductionDm.
+//   3. A throwing markIntroSent does not derail connect().
+
+describe("introductionLedger persistent idempotency", () => {
+	type LedgerStub = {
+		state: { sent: boolean };
+		isCalled: { isIntroSent: number; markIntroSent: number };
+		isIntroSent: () => boolean;
+		markIntroSent: () => void;
+	};
+	function makeLedger(initial: boolean): LedgerStub {
+		const state = { sent: initial };
+		const isCalled = { isIntroSent: 0, markIntroSent: 0 };
+		return {
+			state,
+			isCalled,
+			isIntroSent: () => {
+				isCalled.isIntroSent++;
+				return state.sent;
+			},
+			markIntroSent: () => {
+				isCalled.markIntroSent++;
+				state.sent = true;
+			},
+		};
+	}
+
+	test("skips intro DM when ledger already records intro_sent (process restart case)", async () => {
+		const ledger = makeLedger(true);
+		const channel = new SlackHttpChannel({ ...baseConfig, introductionLedger: ledger });
+		await channel.connect();
+		const calls = mockPostMessage.mock.calls as unknown as Array<[{ channel?: string }]>;
+		const introCalls = calls.filter((c) => c[0].channel === "D_DM_OPEN").length;
+		expect(introCalls).toBe(0);
+		expect(ledger.isCalled.isIntroSent).toBeGreaterThanOrEqual(1);
+		expect(ledger.isCalled.markIntroSent).toBe(0);
+	});
+
+	test("fires intro DM and stamps ledger on a successful send (first boot)", async () => {
+		const ledger = makeLedger(false);
+		const channel = new SlackHttpChannel({ ...baseConfig, introductionLedger: ledger });
+		await channel.connect();
+		const calls = mockPostMessage.mock.calls as unknown as Array<[{ channel?: string }]>;
+		const introCalls = calls.filter((c) => c[0].channel === "D_DM_OPEN").length;
+		expect(introCalls).toBe(1);
+		expect(ledger.isCalled.markIntroSent).toBe(1);
+		expect(ledger.state.sent).toBe(true);
+	});
+
+	test("does NOT stamp ledger when sendIntroductionDm fails (transient Slack error preserves retry budget)", async () => {
+		// Slack rate-limit on chat.postMessage. The DM never reaches the
+		// user; the ledger must stay clear so a process restart retries.
+		mockPostMessage.mockImplementation(() => Promise.reject(new Error("ratelimited")));
+		const ledger = makeLedger(false);
+		const channel = new SlackHttpChannel({ ...baseConfig, introductionLedger: ledger });
+		await channel.connect();
+		expect(channel.getConnectionState()).toBe("connected");
+		expect(ledger.state.sent).toBe(false);
+		expect(ledger.isCalled.markIntroSent).toBe(0);
+		mockPostMessage.mockImplementation(() => Promise.resolve({ ts: "1234567890.123456" }));
+	});
+
+	test("does NOT stamp ledger when chat.postMessage returns no ts (ledger preserves retry across restart)", async () => {
+		mockPostMessage.mockImplementationOnce(() => Promise.resolve({ ts: "" } as { ts: string }));
+		const ledger = makeLedger(false);
+		const channel = new SlackHttpChannel({ ...baseConfig, introductionLedger: ledger });
+		await channel.connect();
+		expect(ledger.state.sent).toBe(false);
+		expect(ledger.isCalled.markIntroSent).toBe(0);
+	});
+
+	test("ledger stamp survives reconnect: a subsequent connect() does not re-DM", async () => {
+		const ledger = makeLedger(false);
+		const channel = new SlackHttpChannel({ ...baseConfig, introductionLedger: ledger });
+		await channel.connect();
+		await channel.disconnect();
+		mockPostMessage.mockClear();
+		await channel.connect();
+		const calls = mockPostMessage.mock.calls as unknown as Array<[{ channel?: string }]>;
+		const introCalls = calls.filter((c) => c[0].channel === "D_DM_OPEN").length;
+		expect(introCalls).toBe(0);
+	});
+
+	test("a throwing markIntroSent does not derail connect() (defense in depth)", async () => {
+		const ledger: IntroductionLedger = {
+			isIntroSent: () => false,
+			markIntroSent: () => {
+				throw new Error("disk full");
+			},
+		};
+		const warns: string[] = [];
+		const original = console.warn;
+		console.warn = (...args: unknown[]) => {
+			warns.push(args.map(String).join(" "));
+		};
+		try {
+			const channel = new SlackHttpChannel({ ...baseConfig, introductionLedger: ledger });
+			await expect(channel.connect()).resolves.toBeUndefined();
+			expect(channel.getConnectionState()).toBe("connected");
+		} finally {
+			console.warn = original;
+		}
+		const all = warns.join("\n");
+		expect(all).toContain("markIntroSent failed");
+		expect(all).toContain("disk full");
+	});
+
+	test("without a ledger (single-tenant dev) the in-memory firstDmSent fallback is used", async () => {
+		// No introductionLedger on baseConfig -> reconnect should still
+		// short-circuit via the in-memory flag. This pins the back-compat
+		// surface for self-hosted Socket Mode and unit-test fixtures.
+		const channel = new SlackHttpChannel(baseConfig);
+		await channel.connect();
+		await channel.disconnect();
+		mockPostMessage.mockClear();
+		await channel.connect();
+		const calls = mockPostMessage.mock.calls as unknown as Array<[{ channel?: string }]>;
+		const introCalls = calls.filter((c) => c[0].channel === "D_DM_OPEN").length;
+		expect(introCalls).toBe(0);
+	});
+});
+
 // ----- send and outbound API ----------------------------------------------
 
 describe("send / outbound", () => {
diff --git a/src/channels/slack-channel-factory.ts b/src/channels/slack-channel-factory.ts
@@ -6,7 +6,7 @@
 import { DEFAULT_METADATA_BASE_URL, MetadataIdentityFetcher, type SlackIdentity } from "../config/identity-fetcher.ts";
 import { MetadataSecretFetcher } from "../config/metadata-fetcher.ts";
 import type { ChannelsConfig } from "../config/schemas.ts";
-import { SlackHttpChannel } from "./slack-http-receiver.ts";
+import { type IntroductionLedger, SlackHttpChannel } from "./slack-http-receiver.ts";
 import type { SlackMetricsEmitter } from "./slack-metrics.ts";
 import type { SlackTransport } from "./slack-transport.ts";
 import { SlackChannel } from "./slack.ts";
@@ -68,6 +68,15 @@ export type CreateSlackChannelInput = {
 	 * follow-up generalizes the surface (Phase 17).
 	 */
 	metrics?: SlackMetricsEmitter;
+	/**
+	 * Persistent intro-DM ledger forwarded to the HTTP receiver. The
+	 * receiver consults it before firing the synthetic first DM so a
+	 * process restart does not re-DM the installer, and stamps it on a
+	 * successful send so /health reports onboarding complete. Socket
+	 * Mode does not use this surface (its onboarding lives in
+	 * `src/onboarding/flow.ts`).
+	 */
+	introductionLedger?: IntroductionLedger;
 };
 
 /**
@@ -125,6 +134,7 @@ export async function createSlackChannel(input: CreateSlackChannelInput): Promis
 			teamId: identity.slack.teamId,
 			installerUserId: identity.slack.installerUserId,
 			teamName: identity.slack.teamName,
+			introductionLedger: input.introductionLedger,
 		});
 	}
 
diff --git a/src/channels/slack-http-receiver.ts b/src/channels/slack-http-receiver.ts
@@ -43,12 +43,42 @@ import { redactTokens } from "./slack-http-utils.ts";
 import { sendIntroductionDm } from "./slack-introduction.ts";
 import type { Channel, ChannelCapabilities, InboundMessage, OutboundMessage, SentMessage } from "./types.ts";
 
+/**
+ * Persistent introduction ledger surface. The HTTP receiver consults
+ * `isIntroSent` before firing the synthetic first DM and calls
+ * `markIntroSent` only AFTER a successful Slack send. The wiring lives
+ * in `index.ts` against the SQLite `onboarding_state` table; this
+ * indirection keeps the channel free of a direct SQLite handle so the
+ * unit tests do not have to spin up a database.
+ *
+ * Two outcomes the ledger pins:
+ *   1. `/health.onboarding` flips from "pending" to "complete" after the
+ *      first successful intro DM, mirroring the Socket Mode flow.
+ *   2. A process restart never re-fires the intro DM once it has gone
+ *      out (the in-memory `firstDmSent` flag was lost on restart and
+ *      could re-DM the user; SQLite is now authoritative).
+ *
+ * Both callbacks are optional: when omitted (single-tenant dev / tests),
+ * the channel falls back to the in-memory `firstDmSent` flag with the
+ * same semantics it had before the ledger was introduced.
+ */
+export type IntroductionLedger = {
+	isIntroSent(): boolean;
+	markIntroSent(): void;
+};
+
 export type SlackHttpChannelConfig = {
 	botToken: string;
 	gatewaySigningSecret: string;
 	teamId: string;
 	installerUserId: string;
 	teamName: string;
+	/**
+	 * Persistent intro-DM ledger. Production wires this against
+	 * `onboarding_state` so the /health endpoint reports onboarding
+	 * complete and a process restart does not re-DM the installer.
+	 */
+	introductionLedger?: IntroductionLedger;
 };
 
 type ConnectionState = "disconnected" | "connecting" | "connected" | "error";
@@ -79,11 +109,12 @@ export class SlackHttpChannel implements Channel, EventDispatchHost {
 	private connectionState: ConnectionState = "disconnected";
 	private botUserId: string | null = null;
 	private phantomName = "Phantom";
-	// Instance-level guard against re-introducing the agent after a
-	// transient disconnect plus reconnect. A process restart resets the
-	// flag intentionally: a fresh user-visible DM beats a silent UX
-	// failure when the operator has had to restart the channel.
+	// In-memory fallback when no persistent ledger is wired (tests,
+	// single-tenant dev). Production reads/writes the SQLite ledger
+	// passed in via `introductionLedger`; once the ledger is set it
+	// is the authoritative source and this flag is unused.
 	private firstDmSent = false;
+	private readonly introductionLedger: IntroductionLedger | null;
 
 	constructor(config: SlackHttpChannelConfig) {
 		if (!config.botToken) throw new Error("SlackHttpChannel: botToken is required");
@@ -94,6 +125,7 @@ export class SlackHttpChannel implements Channel, EventDispatchHost {
 		this.installerUserId = config.installerUserId;
 		this.teamName = config.teamName;
 		this.gatewaySigningSecret = config.gatewaySigningSecret;
+		this.introductionLedger = config.introductionLedger ?? null;
 
 		// Bolt's `App` constructor calls `receiver.init(app)` synchronously
 		// to wire the App reference; we reuse that App for processEvent
@@ -201,17 +233,43 @@ export class SlackHttpChannel implements Channel, EventDispatchHost {
 		}
 
 		// Synthetic first DM. Fire after the channel state flips to connected
-		// so the user can reply immediately; gate on firstDmSent so a
-		// reconnect-after-drop does not re-introduce.
-		if (!this.firstDmSent && this.installerUserId) {
-			const result = await sendIntroductionDm({
-				phantomName: this.phantomName,
-				teamName: this.teamName,
-				installerUserId: this.installerUserId,
-				sendDm: (userId, text) => this.sendDm(userId, text),
-			});
-			if (result.sent) {
-				this.firstDmSent = true;
+		// so the user can reply immediately. Idempotency:
+		//   - production wires `introductionLedger` against the SQLite
+		//     onboarding_state table; a true ledger means the DM has
+		//     already gone out (in this process or a prior restart) and
+		//     we skip without retrying. The ledger is also what /health
+		//     reads to flip onboarding from "pending" to "complete".
+		//   - without a ledger (tests, single-tenant dev) we fall back
+		//     to the in-memory `firstDmSent` flag; a restart re-fires.
+		if (!this.installerUserId) {
+			return;
+		}
+		if (this.introductionLedger?.isIntroSent() === true) {
+			console.log(`[${LOG_TAG}] introduction ledger already records intro_sent; skipping intro DM`);
+			return;
+		}
+		if (this.introductionLedger === null && this.firstDmSent) {
+			return;
+		}
+		const result = await sendIntroductionDm({
+			phantomName: this.phantomName,
+			teamName: this.teamName,
+			installerUserId: this.installerUserId,
+			sendDm: (userId, text) => this.sendDm(userId, text),
+		});
+		if (result.sent) {
+			this.firstDmSent = true;
+			// The ledger is stamped only AFTER a successful Slack send so
+			// a transient Slack failure leaves the row clear and the next
+			// process start retries. This mirrors the Phase 12
+			// firstboot_state pattern in the Socket Mode flow.
+			if (this.introductionLedger !== null) {
+				try {
+					this.introductionLedger.markIntroSent();
+				} catch (err: unknown) {
+					const msg = err instanceof Error ? err.message : String(err);
+					console.warn(`[${LOG_TAG}] introduction ledger markIntroSent failed: ${msg}`);
+				}
 			}
 		}
 	}
diff --git a/src/index.ts b/src/index.ts
@@ -57,7 +57,7 @@ import { MemorySystem } from "./memory/system.ts";
 import { isFirstRun, isOnboardingInProgress } from "./onboarding/detection.ts";
 import { type OnboardingTarget, startOnboarding } from "./onboarding/flow.ts";
 import { buildOnboardingPrompt } from "./onboarding/prompt.ts";
-import { getOnboardingStatus } from "./onboarding/state.ts";
+import { getOnboardingStatus, markOnboardingComplete, markOnboardingStarted } from "./onboarding/state.ts";
 import { createRoleRegistry } from "./roles/registry.ts";
 import type { RoleTemplate } from "./roles/types.ts";
 import { Scheduler } from "./scheduler/service.ts";
@@ -364,11 +364,24 @@ async function main(): Promise<void> {
 	// sees both metric families. Per-emitter registries keep names from
 	// colliding across channels.
 	setMetricsRegistryProvider(() => [slackMetrics.registry, emailMetrics.registry]);
+	// HTTP-mode tenants need a persistent intro-DM ledger so /health
+	// reports onboarding=complete after the first DM lands and a process
+	// restart does not re-DM the installer. We bind it to the SQLite
+	// `onboarding_state` table that the Socket Mode flow already drives.
+	// The factory ignores this surface for the Socket Mode path; it is
+	// only consumed by SlackHttpChannel.connect().
 	const slackChannel: SlackTransport | null = await createSlackChannel({
 		transport: slackTransport,
 		channelsConfig,
 		metadataBaseUrl: process.env.METADATA_BASE_URL,
 		metrics: slackMetrics,
+		introductionLedger: {
+			isIntroSent: () => getOnboardingStatus(db).status === "complete",
+			markIntroSent: () => {
+				markOnboardingStarted(db);
+				markOnboardingComplete(db);
+			},
+		},
 	});
 
 	if (slackChannel) {
