From d4d23fe95454c2b73df29f34f4bb147ebf7101bc Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 06:43:27 +0100 Subject: [PATCH] fix(mattermost): preserve websocket posts across restarts (#110386) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(mattermost): adopt durable ingress drain at the websocket chokepoint Posted events processed detached from the websocket receive with only a 5-minute in-memory guard; a crash lost the post and reconnect never replays it. Raw posted envelopes now journal durably (event_id = post.id per the upstream Post model, lane per channel_id, one row per post) before handler scheduling; dispatch runs through the core drain with deferred claims through debounce, merged-flush fan-out adoption, gated-turn settlement, and 30d/20k tombstones covering the old 5min/2k guard, which is deleted after its parity test. post_edited stays excluded and cannot be swallowed by posted tombstones. Cold-gap limitation stated: Mattermost cannot replay posts missed while disconnected. Autoreview blocked by codex sandbox network in the build stage; full manual review performed (updated one websocket test asserting the pre-adoption parsed-post contract to the raw-envelope contract). Part of #109657 wave 2. * style(mattermost): keep ingress monitor type internal * fix(mattermost): retry then loudly escalate a failed durable append Landing autoreview caught a real loss path: a durable enqueue failure at the websocket chokepoint was logged and swallowed — the raw envelope discarded, the connection kept running against a broken store, and reconnect never replays, so a transient SQLite failure silently lost the post. The append now retries with short backoff for transient blips; a persistent failure propagates and the websocket terminates loudly so the outage is operator-visible instead of silently dropping every subsequent post. Regression test covers both the absorbed-transient and escalation paths. * style(mattermost): format rebased ingress handler * docs(mattermost): document bounded auth-failure retries under deferred claims * fix(mattermost): guard the drain pump against stop racing the async prune stop() disposing before the startup pump finished pruning let the pump lazily create a fresh undisposed drain and dispatch after shutdown. The pump now re-checks running after the prune, and stop() disposes again after awaiting the pump so a drain created mid-race is torn down. Regression test blocks the prune across stop and asserts no dispatch. * fix(mattermost): serialize durable admissions to preserve lane order Concurrent websocket callbacks let a post in append-retry backoff be overtaken by its successor, inverting same-channel arrival order in the queue. Admissions now chain (order over latency, mirroring the iMessage admission tail); regression proves a retried post still lands ahead of a concurrently received one. * test(mattermost): assert lane order via dispatch sequence * fix(mattermost): stop() awaits in-flight admissions before disposal * fix(mattermost): satisfy ingress lint checks * fix(mattermost): honor envelope-level channel ids in the durable inspector Posts can carry their channel id on the post, the event data, or the broadcast envelope — the monitor dispatch honors all three, but the ingress inspector and claim-side validator required the nested field, rejecting valid posts as permanent and (via the storage-failure escalation) tearing down the socket for a failure that never happened, losing posts reconnect cannot replay. Both sites accept the three shapes; regression proves an envelope-level post dispatches. --- .../src/mattermost/monitor-ingress.test.ts | 420 ++++ .../src/mattermost/monitor-ingress.ts | 413 ++++ .../src/mattermost/monitor-replay.ts | 41 - .../src/mattermost/monitor-websocket.test.ts | 58 +- .../src/mattermost/monitor-websocket.ts | 43 +- .../monitor.inbound-system-event.test.ts | 32 + .../mattermost/src/mattermost/monitor.ts | 1773 +++++++++-------- 7 files changed, 1821 insertions(+), 959 deletions(-) create mode 100644 extensions/mattermost/src/mattermost/monitor-ingress.test.ts create mode 100644 extensions/mattermost/src/mattermost/monitor-ingress.ts delete mode 100644 extensions/mattermost/src/mattermost/monitor-replay.ts diff --git a/extensions/mattermost/src/mattermost/monitor-ingress.test.ts b/extensions/mattermost/src/mattermost/monitor-ingress.test.ts new file mode 100644 index 000000000000..edc2f74bd147 --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-ingress.test.ts @@ -0,0 +1,420 @@ +// Mattermost durable ingress tests cover append, recovery, tombstones, and merged adoption. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + buildMattermostFlushIngressLifecycle, + createMattermostIngressMonitor, + type MattermostIngressLifecycle, +} from "./monitor-ingress.js"; + +type MattermostIngressQueue = NonNullable< + Parameters[0]["queue"] +>; +type MattermostIngressPayload = Parameters[1]; +type MattermostIngressDispatch = Parameters[0]["dispatch"]; + +function postedEvent(params?: { + postId?: string; + channelId?: string; + message?: string; + event?: "posted" | "post_edited"; +}): string { + return JSON.stringify({ + event: params?.event ?? "posted", + data: { + post: JSON.stringify({ + id: params?.postId ?? "post-1", + channel_id: params?.channelId ?? "channel-1", + user_id: "user-1", + message: params?.message ?? "hello", + }), + }, + }); +} + +function startMonitor(queue: MattermostIngressQueue, dispatch: MattermostIngressDispatch) { + return createMattermostIngressMonitor({ + accountId: "default", + queue, + dispatch, + runtime: { error: vi.fn(), log: vi.fn() }, + pollIntervalMs: 60_000, + adoptionStallTimeoutMs: 5_000, + }); +} + +async function withQueue(fn: (queue: MattermostIngressQueue) => Promise): Promise { + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-mattermost-ingress-")); + const stateDir = await fs.realpath(created); + const queue = createChannelIngressQueueForTests({ + channelId: "mattermost", + accountId: "default", + stateDir, + }); + try { + return await fn(queue); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +function testLifecycle() { + const calls = { + adopted: vi.fn(async () => {}), + deferred: vi.fn(), + finalizing: vi.fn(), + abandoned: vi.fn(async () => {}), + }; + const lifecycle: MattermostIngressLifecycle = { + abortSignal: new AbortController().signal, + onAdopted: calls.adopted, + onDeferred: calls.deferred, + onAdoptionFinalizing: calls.finalizing, + onAbandoned: calls.abandoned, + }; + return { calls, lifecycle }; +} + +afterEach(() => { + closeOpenClawStateDatabaseForTest(); + vi.restoreAllMocks(); +}); + +describe("Mattermost durable ingress", () => { + it("propagates durable append failure before handler scheduling", async () => { + await withQueue(async (queue) => { + const appendError = new Error("sqlite unavailable"); + const failingQueue = { + ...queue, + enqueue: vi.fn().mockRejectedValue(appendError), + } satisfies MattermostIngressQueue; + const dispatch = vi.fn(); + const monitor = startMonitor(failingQueue, dispatch); + try { + await expect(monitor.receive(postedEvent())).rejects.toBe(appendError); + expect(dispatch).not.toHaveBeenCalled(); + } finally { + await monitor.stop(); + } + }); + }); + + it("recovers an uncompleted post with a fresh drain and dispatches exactly once", async () => { + await withQueue(async (queue) => { + const interruptedDispatch = vi.fn((_post, _payload, lifecycle) => { + lifecycle.onDeferred(); + return { kind: "deferred" } as const; + }); + const interrupted = startMonitor(queue, interruptedDispatch); + await interrupted.receive(postedEvent({ postId: "post-restart" })); + await interrupted.waitForIdle(); + expect(await queue.listClaims()).toHaveLength(1); + await interrupted.stop(); + + const recoveredDispatch = vi.fn(async (_post, _payload, lifecycle) => { + await lifecycle.onAdopted(); + }); + const recovered = startMonitor(queue, recoveredDispatch); + try { + await recovered.waitForIdle(); + expect(recoveredDispatch).toHaveBeenCalledTimes(1); + } finally { + await recovered.stop(); + } + }); + }); + + it("retains completion so a duplicate post id cannot dispatch twice", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (_post, _payload, lifecycle) => { + await lifecycle.onAdopted(); + }); + const monitor = startMonitor(queue, dispatch); + try { + const event = postedEvent({ postId: "post-completed" }); + await monitor.receive(event); + await monitor.waitForIdle(); + await monitor.receive(event); + await monitor.waitForIdle(); + expect(dispatch).toHaveBeenCalledTimes(1); + } finally { + await monitor.stop(); + } + }); + }); + + it("preserves every id from the retired merged-message guard key space", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (_post, _payload, lifecycle) => { + await lifecycle.onAdopted(); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive(postedEvent({ postId: "post-batch-first", message: "first" })); + await monitor.receive(postedEvent({ postId: "post-batch-second", message: "second" })); + await monitor.waitForIdle(); + + await monitor.receive(postedEvent({ postId: "post-batch-first", message: "first" })); + await monitor.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(2); + expect(dispatch.mock.calls.map(([post]) => post.id)).toEqual([ + "post-batch-first", + "post-batch-second", + ]); + } finally { + await monitor.stop(); + } + }); + }); + + it("stores the exact raw envelope in a per-channel lane", async () => { + await withQueue(async (queue) => { + const rawEvent = postedEvent({ postId: "post-raw", channelId: "channel-raw" }); + const dispatch = vi.fn((_post, _payload, lifecycle) => { + lifecycle.onDeferred(); + return { kind: "deferred" } as const; + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive(rawEvent); + await monitor.waitForIdle(); + expect(await queue.listClaims()).toEqual([ + expect.objectContaining({ + id: "post-raw", + laneKey: "channel:channel-raw", + payload: expect.objectContaining({ rawEvent }), + }), + ]); + } finally { + await monitor.stop(); + } + }); + }); + + it("does not let ignored post_edited events tombstone a later posted event", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (_post, _payload, lifecycle) => { + await lifecycle.onAdopted(); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive(postedEvent({ postId: "post-edit", event: "post_edited" })); + await monitor.waitForIdle(); + await monitor.receive(postedEvent({ postId: "post-edit", event: "posted" })); + await monitor.waitForIdle(); + expect(dispatch).toHaveBeenCalledTimes(1); + } finally { + await monitor.stop(); + } + }); + }); + + it("preserves same-channel arrival order across an append retry backoff", async () => { + await withQueue(async (queue) => { + let failFirst = true; + const enqueue = queue.enqueue.bind(queue); + queue.enqueue = async (...args) => { + if (failFirst) { + failFirst = false; + throw new Error("sqlite busy"); + } + return await enqueue(...args); + }; + const dispatched: string[] = []; + const ingress = startMonitor(queue, async (post, _payload, lifecycle) => { + dispatched.push(post.id); + await lifecycle.onAdopted(); + }); + try { + // A's first append fails into backoff; B arrives concurrently. The + // admission chain must hold B until A commits, or lane order inverts. + const admitA = ingress.receive(postedEvent({ postId: "post-A" })); + const admitB = ingress.receive(postedEvent({ postId: "post-B" })); + await Promise.all([admitA, admitB]); + await vi.waitFor(() => expect(dispatched).toEqual(["post-A", "post-B"]), { + timeout: 5_000, + }); + } finally { + await ingress.stop(); + } + }); + }); + + it("never dispatches from a pump racing stop through the async prune", async () => { + await withQueue(async (queue) => { + let releasePrune = () => {}; + const pruneGate = new Promise((resolve) => { + releasePrune = resolve; + }); + const prune = queue.prune.bind(queue); + queue.prune = async (...args) => { + await pruneGate; + return await prune(...args); + }; + const dispatch = vi.fn(); + const ingress = startMonitor(queue, dispatch); + await ingress.receive(postedEvent({ postId: "post-stop-race" })); + + const stopping = ingress.stop(); + releasePrune(); + await stopping; + + expect(dispatch).not.toHaveBeenCalled(); + }); + }); + + it("retries a transient append failure and escalates a persistent one", async () => { + await withQueue(async (queue) => { + let failures = 2; + const enqueue = queue.enqueue.bind(queue); + queue.enqueue = async (...args) => { + if (failures > 0) { + failures -= 1; + throw new Error("sqlite busy"); + } + return await enqueue(...args); + }; + const ingress = startMonitor(queue, vi.fn()); + try { + // Two transient failures absorb into the bounded retry. + await ingress.receive(postedEvent({ postId: "post-retry" })); + expect(await queue.listPending({ limit: "all" })).toHaveLength(1); + + // A persistent failure escalates so the websocket can tear down loudly. + failures = Number.POSITIVE_INFINITY; + await expect(ingress.receive(postedEvent({ postId: "post-lost" }))).rejects.toThrow( + "sqlite busy", + ); + } finally { + await ingress.stop(); + } + }); + }); + + it("accepts a post whose channel id arrives at the envelope level", async () => { + await withQueue(async (queue) => { + const dispatched: string[] = []; + const ingress = startMonitor(queue, async (post, _payload, lifecycle) => { + dispatched.push(post.id); + await lifecycle.onAdopted(); + }); + try { + // The monitor dispatch honors post/data/broadcast channel ids; the + // durable inspector must not reject the envelope-level shapes. + await ingress.receive( + JSON.stringify({ + event: "posted", + data: { + channel_id: "channel-envelope", + post: JSON.stringify({ id: "post-envelope", user_id: "user-1", message: "hi" }), + }, + }), + ); + await vi.waitFor(() => expect(dispatched).toEqual(["post-envelope"])); + } finally { + await ingress.stop(); + } + }); + }); + + it("dead-letters malformed persisted payloads without retry", async () => { + await withQueue(async (queue) => { + await queue.enqueue( + "post-malformed", + { version: 1, receivedAt: 1, rawEvent: "{" }, + { receivedAt: 1, laneKey: "channel:channel-1" }, + ); + const dispatch = vi.fn(); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.waitForIdle(); + expect((await queue.enqueue("post-malformed", {} as MattermostIngressPayload)).kind).toBe( + "failed", + ); + expect(dispatch).not.toHaveBeenCalled(); + } finally { + await monitor.stop(); + } + }); + }); + + it("dead-letters permanent Mattermost auth failures", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => { + throw new Error("Mattermost API 401 Unauthorized: invalid token"); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive(postedEvent({ postId: "post-auth" })); + await monitor.waitForIdle(); + expect((await queue.enqueue("post-auth", {} as MattermostIngressPayload)).kind).toBe( + "failed", + ); + expect(dispatch).toHaveBeenCalledTimes(1); + } finally { + await monitor.stop(); + } + }); + }); + + it("leaves transient Mattermost failures retryable", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => { + throw new Error("Mattermost API 503 Service Unavailable"); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive(postedEvent({ postId: "post-transient" })); + await monitor.waitForIdle(); + expect(await queue.listPending({ limit: "all" })).toEqual([ + expect.objectContaining({ id: "post-transient" }), + ]); + expect(dispatch).toHaveBeenCalledTimes(1); + } finally { + await monitor.stop(); + } + }); + }); +}); + +describe("Mattermost merged ingress lifecycle", () => { + it("fans adoption out to every constituent claim", async () => { + const first = testLifecycle(); + const second = testLifecycle(); + const merged = buildMattermostFlushIngressLifecycle([ + { turnAdoptionLifecycle: first.lifecycle }, + { turnAdoptionLifecycle: second.lifecycle }, + ]); + + merged.lifecycle?.onDeferred(); + await merged.lifecycle?.onAdopted(); + await merged.settle(); + + expect(first.calls.deferred).toHaveBeenCalledTimes(1); + expect(second.calls.deferred).toHaveBeenCalledTimes(1); + expect(first.calls.adopted).toHaveBeenCalledTimes(1); + expect(second.calls.adopted).toHaveBeenCalledTimes(1); + }); + + it("completes all claims when a gated flush never dispatches", async () => { + const first = testLifecycle(); + const second = testLifecycle(); + const merged = buildMattermostFlushIngressLifecycle([ + { turnAdoptionLifecycle: first.lifecycle }, + { turnAdoptionLifecycle: second.lifecycle }, + ]); + + await merged.settle(); + + expect(first.calls.adopted).toHaveBeenCalledTimes(1); + expect(second.calls.adopted).toHaveBeenCalledTimes(1); + }); +}); diff --git a/extensions/mattermost/src/mattermost/monitor-ingress.ts b/extensions/mattermost/src/mattermost/monitor-ingress.ts new file mode 100644 index 000000000000..cb6883d03bac --- /dev/null +++ b/extensions/mattermost/src/mattermost/monitor-ingress.ts @@ -0,0 +1,413 @@ +// Mattermost plugin module owns raw WebSocket durable ingress mapping and draining. +import { + createChannelIngressDrain, + DEFAULT_INGRESS_ADOPTION_STALL_MS, + DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + type ChannelIngressDrain, + type ChannelIngressQueue, +} from "openclaw/plugin-sdk/channel-outbound"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; +import { getMattermostRuntime } from "../runtime.js"; +import type { MattermostPost } from "./client.js"; +import { + parseMattermostEventPayload, + parseMattermostPost, + type MattermostEventPayload, +} from "./monitor-websocket.js"; + +const MATTERMOST_INGRESS_PAYLOAD_VERSION = 1; +const MATTERMOST_INGRESS_POLL_INTERVAL_MS = 1_000; +const MATTERMOST_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000; +const MATTERMOST_INGRESS_COMPLETED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const MATTERMOST_INGRESS_COMPLETED_MAX_ENTRIES = 20_000; +const MATTERMOST_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +const MATTERMOST_INGRESS_FAILED_MAX_ENTRIES = 20_000; + +export type MattermostIngressLifecycle = { + abortSignal: AbortSignal; + onAdopted: () => void | Promise; + onDeferred: () => void; + onAdoptionFinalizing: () => void; + onAbandoned: () => void | Promise; +}; + +/** Fan one merged Mattermost turn's adoption lifecycle across every source claim. */ +export function buildMattermostFlushIngressLifecycle( + entries: ReadonlyArray<{ turnAdoptionLifecycle?: MattermostIngressLifecycle }>, +): { + lifecycle: MattermostIngressLifecycle | undefined; + settle: () => Promise; +} { + const lifecycles = entries + .map((entry) => entry.turnAdoptionLifecycle) + .filter((lifecycle) => lifecycle !== undefined); + const [firstLifecycle] = lifecycles; + if (!firstLifecycle) { + return { lifecycle: undefined, settle: async () => {} }; + } + let handedOff = false; + const adoptAll = async () => { + for (const lifecycle of lifecycles) { + await lifecycle.onAdopted(); + } + }; + return { + lifecycle: { + abortSignal: + lifecycles.length === 1 + ? firstLifecycle.abortSignal + : AbortSignal.any(lifecycles.map((lifecycle) => lifecycle.abortSignal)), + onAdopted: async () => { + handedOff = true; + await adoptAll(); + }, + onDeferred: () => { + handedOff = true; + for (const lifecycle of lifecycles) { + lifecycle.onDeferred(); + } + }, + onAdoptionFinalizing: () => { + for (const lifecycle of lifecycles) { + lifecycle.onAdoptionFinalizing(); + } + }, + onAbandoned: async () => { + handedOff = true; + await Promise.all( + lifecycles.map(async (lifecycle) => { + await lifecycle.onAbandoned(); + }), + ); + }, + }, + // Gated/no-dispatch turns are terminal and must not leave source claims deferred. + settle: async () => { + if (!handedOff) { + await adoptAll(); + } + }, + }; +} + +type MattermostIngressPayload = { + version: 1; + receivedAt: number; + rawEvent: string; +}; + +type MattermostIngressDispatchResult = + | { kind: "completed" } + | { kind: "deferred" } + | { kind: "failed-retryable"; error: unknown }; + +type MattermostIngressDispatch = ( + post: MattermostPost, + payload: MattermostEventPayload, + lifecycle: MattermostIngressLifecycle, +) => Promise | MattermostIngressDispatchResult | void; + +class MattermostIngressPermanentError extends Error { + constructor( + readonly reason: "invalid-event" | "mattermost-auth", + message: string, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "MattermostIngressPermanentError"; + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function parseRawObject(raw: string, subject: string): Record { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (error) { + throw new MattermostIngressPermanentError( + "invalid-event", + `${subject} contains invalid JSON.`, + { cause: error }, + ); + } + if (!isRecord(parsed)) { + throw new MattermostIngressPermanentError("invalid-event", `${subject} must be a JSON object.`); + } + return parsed; +} + +function parseRawPost(value: unknown): Record { + if (typeof value === "string") { + return parseRawObject(value, "Mattermost posted event post"); + } + if (isRecord(value)) { + return value; + } + throw new MattermostIngressPermanentError( + "invalid-event", + "Mattermost posted event is missing its post object.", + ); +} + +function requiredString(value: unknown, field: string): string { + if (typeof value === "string" && value.trim()) { + return value.trim(); + } + throw new MattermostIngressPermanentError( + "invalid-event", + `Mattermost posted event is missing ${field}.`, + ); +} + +function inspectMattermostIngressEvent(rawEvent: string): { + eventId: string; + laneKey: string; +} | null { + const envelope = parseRawObject(rawEvent, "Mattermost WebSocket event"); + if (envelope.event !== "posted") { + return null; + } + const data = isRecord(envelope.data) ? envelope.data : null; + const post = parseRawPost(data?.post); + const eventId = requiredString(post.id, "post.id"); + // Mattermost can carry the channel id on the post, the event data, or the + // broadcast envelope (the monitor dispatch honors all three). Rejecting the + // envelope-level shapes as permanent would drop valid posts and tear the + // socket down for a storage failure that never happened. + const broadcast = isRecord(envelope.broadcast) ? envelope.broadcast : null; + const channelId = + typeof post.channel_id === "string" && post.channel_id.trim() + ? post.channel_id.trim() + : typeof data?.channel_id === "string" && data.channel_id.trim() + ? data.channel_id.trim() + : requiredString(broadcast?.channel_id, "channel_id"); + return { eventId, laneKey: `channel:${channelId}` }; +} + +function parseClaimedEvent( + rawEvent: string, + eventId: string, +): { + post: MattermostPost; + payload: MattermostEventPayload; +} { + const payload = parseMattermostEventPayload(rawEvent); + if (!payload || payload.event !== "posted") { + throw new MattermostIngressPermanentError( + "invalid-event", + `Mattermost ingress row ${eventId} is not a posted event.`, + ); + } + const post = parseMattermostPost(payload.data?.post); + // Channel id may live on the post, the event data, or the broadcast — the + // durable inspector accepted all three, so the claim-side check must too. + const claimedChannelId = + post?.channel_id?.trim() || + payload.data?.channel_id?.trim() || + payload.broadcast?.channel_id?.trim(); + if (!post || post.id !== eventId || !claimedChannelId) { + throw new MattermostIngressPermanentError( + "invalid-event", + `Mattermost ingress row ${eventId} has invalid post identity.`, + ); + } + return { post, payload }; +} + +function resolveMattermostIngressNonRetryableFailure(error: unknown) { + if (error instanceof MattermostIngressPermanentError) { + return { reason: error.reason, message: error.message }; + } + const message = formatErrorMessage(error); + return /Mattermost API (?:401|403)\b/.test(message) + ? { reason: "mattermost-auth", message } + : null; +} + +type MattermostIngressMonitor = { + receive: (rawEvent: string) => Promise; + stop: () => Promise; + waitForIdle: () => Promise; +}; + +export function createMattermostIngressMonitor(options: { + accountId: string; + queue?: ChannelIngressQueue; + dispatch: MattermostIngressDispatch; + runtime: Pick; + pollIntervalMs?: number; + adoptionStallTimeoutMs?: number; + abortSignal?: AbortSignal; +}): MattermostIngressMonitor { + let queue = options.queue; + let drain: ChannelIngressDrain | undefined; + let running = true; + let requested = false; + let pumping: Promise | undefined; + let lastPrunedAt = 0; + + const getQueue = (): ChannelIngressQueue => { + queue ??= getMattermostRuntime().state.openChannelIngressQueue({ + accountId: options.accountId, + }); + return queue; + }; + + const getDrain = (): ChannelIngressDrain => { + drain ??= createChannelIngressDrain({ + queue: getQueue(), + adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS, + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + }, + resolveNonRetryableFailure: resolveMattermostIngressNonRetryableFailure, + ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + onLog: (message) => options.runtime.log?.(`mattermost ${message}`), + dispatchClaimedEvent: async (record, lifecycle) => { + if (record.payload.version !== MATTERMOST_INGRESS_PAYLOAD_VERSION) { + throw new MattermostIngressPermanentError( + "invalid-event", + `Mattermost ingress row ${record.id} has an unsupported version.`, + ); + } + const { post, payload } = parseClaimedEvent(record.payload.rawEvent, record.id); + return await options.dispatch(post, payload, lifecycle); + }, + }); + return drain; + }; + + const pruneIfDue = async (): Promise => { + const now = Date.now(); + if (now - lastPrunedAt < MATTERMOST_INGRESS_PRUNE_INTERVAL_MS) { + return; + } + await getQueue().prune({ + completedTtlMs: MATTERMOST_INGRESS_COMPLETED_TTL_MS, + completedMaxEntries: MATTERMOST_INGRESS_COMPLETED_MAX_ENTRIES, + failedTtlMs: MATTERMOST_INGRESS_FAILED_TTL_MS, + failedMaxEntries: MATTERMOST_INGRESS_FAILED_MAX_ENTRIES, + now, + }); + lastPrunedAt = now; + }; + + const runPump = async (): Promise => { + try { + for (;;) { + requested = false; + await pruneIfDue(); + // stop() may have run during the async prune; creating the lazy drain + // now would leave an undisposed instance dispatching after stop. + if (!running) { + break; + } + const activeDrain = getDrain(); + const { started } = await activeDrain.drainOnce(); + await activeDrain.waitForIdle(); + if (!running || (!requested && started === 0)) { + break; + } + } + } catch (error) { + options.runtime.error?.(`mattermost ingress drain failed: ${formatErrorMessage(error)}`); + } finally { + pumping = undefined; + if (running && requested) { + requestDrain(); + } + } + }; + + const requestDrain = (): void => { + requested = true; + if (!running || pumping) { + return; + } + pumping = runPump(); + }; + + const timer = setInterval( + requestDrain, + options.pollIntervalMs ?? MATTERMOST_INGRESS_POLL_INTERVAL_MS, + ); + timer.unref?.(); + requestDrain(); + + // Serialize admissions: WS message callbacks run concurrently, and a post + // stuck in append-retry backoff must delay its successors or same-channel + // arrival order inverts in the queue (order over latency). + let admissionTail: Promise = Promise.resolve(); + + const admitOnce = async (rawEvent: string): Promise => { + const facts = inspectMattermostIngressEvent(rawEvent); + if (!facts) { + return; + } + const receivedAt = Date.now(); + // Websockets have no nack: a dropped append is a lost post (reconnect + // never replays). Retry transient failures before letting the error + // escalate to the socket teardown in monitor-websocket. + let lastError: unknown; + for (const delayMs of [0, 100, 300]) { + if (delayMs > 0) { + await new Promise((resolve) => { + setTimeout(resolve, delayMs); + }); + } + try { + await getQueue().enqueue( + facts.eventId, + { + version: MATTERMOST_INGRESS_PAYLOAD_VERSION, + receivedAt, + rawEvent, + }, + { receivedAt, laneKey: facts.laneKey }, + ); + requestDrain(); + return; + } catch (error) { + lastError = error; + } + } + throw lastError; + }; + + return { + receive: (rawEvent) => { + const admission = admissionTail.then(() => admitOnce(rawEvent)); + admissionTail = admission.catch(() => undefined); + return admission; + }, + stop: async () => { + running = false; + clearInterval(timer); + // A caller returning from stop() must know every accepted envelope is + // durably committed; an in-flight admission racing process exit would + // otherwise lose the post. + await admissionTail; + drain?.dispose(); + await pumping; + // The pump may have lazily created the drain after the first dispose. + drain?.dispose(); + await drain?.waitForIdle(); + }, + waitForIdle: async () => { + for (;;) { + const activePump = pumping; + if (!activePump) { + break; + } + await activePump; + } + await drain?.waitForIdle(); + }, + }; +} diff --git a/extensions/mattermost/src/mattermost/monitor-replay.ts b/extensions/mattermost/src/mattermost/monitor-replay.ts deleted file mode 100644 index 130e6b02e9f2..000000000000 --- a/extensions/mattermost/src/mattermost/monitor-replay.ts +++ /dev/null @@ -1,41 +0,0 @@ -// Mattermost plugin module owns replay-guarded post processing. -import { createChannelReplayGuard } from "openclaw/plugin-sdk/persistent-dedupe"; - -const RECENT_MATTERMOST_MESSAGE_TTL_MS = 5 * 60_000; -const RECENT_MATTERMOST_MESSAGE_MAX = 2000; -function buildMattermostInboundReplayKeys(params: { - accountId: string; - messageIds: string[]; -}): string[] { - return params.messageIds.map((id) => (id.trim() ? `${params.accountId}:${id.trim()}` : "")); -} - -function createMattermostInboundReplayGuard() { - return createChannelReplayGuard<{ accountId: string; messageIds: string[] }>({ - dedupe: { - ttlMs: RECENT_MATTERMOST_MESSAGE_TTL_MS, - memoryMaxSize: RECENT_MATTERMOST_MESSAGE_MAX, - }, - buildReplayKey: buildMattermostInboundReplayKeys, - }); -} - -type MattermostInboundReplayGuard = ReturnType; -const recentInboundMessages = createMattermostInboundReplayGuard(); - -export async function processMattermostReplayGuardedPost(params: { - accountId: string; - messageIds: string[]; - handlePost: () => Promise; - replayGuard?: MattermostInboundReplayGuard; -}): Promise<"processed" | "duplicate"> { - const replayGuard = params.replayGuard ?? recentInboundMessages; - const event = { - accountId: params.accountId, - messageIds: params.messageIds, - }; - const result = await replayGuard.processGuarded(event, params.handlePost, { - onError: "commit", - }); - return result.kind === "processed" ? "processed" : "duplicate"; -} diff --git a/extensions/mattermost/src/mattermost/monitor-websocket.test.ts b/extensions/mattermost/src/mattermost/monitor-websocket.test.ts index ebcdc543a112..0ebf5db1f1fe 100644 --- a/extensions/mattermost/src/mattermost/monitor-websocket.test.ts +++ b/extensions/mattermost/src/mattermost/monitor-websocket.test.ts @@ -284,14 +284,10 @@ describe("mattermost websocket monitor", () => { await once(server, "close"); } - expect(onPosted).toHaveBeenCalledWith( - expect.objectContaining({ id: "post-1", message: "normal Mattermost post" }), - expect.any(Object), - ); - expect(onPosted).toHaveBeenCalledWith( - expect.objectContaining({ id: "post-large", props: largeProps }), - expect.any(Object), - ); + // onPosted now receives the raw envelope; the post rides inside it as a + // nested JSON string, so its fields appear escaped. + expect(onPosted).toHaveBeenCalledWith(expect.stringContaining('\\"id\\":\\"post-1\\"')); + expect(onPosted).toHaveBeenCalledWith(expect.stringContaining('\\"id\\":\\"post-large\\"')); expect(runtime.error).toHaveBeenCalledWith( expect.stringContaining("Max payload size exceeded"), ); @@ -347,6 +343,52 @@ describe("mattermost websocket monitor", () => { }); }); + it("hands posted envelopes to ingress raw and keeps post_edited out", async () => { + const socket = new FakeWebSocket(); + const onPosted = vi.fn(async () => {}); + const connectOnce = createMattermostConnectOnce({ + wsUrl: "wss://example.invalid/api/v4/websocket", + botToken: "token", + runtime: testRuntime(), + nextSeq: () => 1, + onPosted, + webSocketFactory: () => socket, + }); + const posted = { + event: "posted", + data: { + post: JSON.stringify({ + id: "post-raw", + channel_id: "channel-raw", + unexpected_transport_field: true, + }), + }, + }; + + const connected = connectOnce(); + socket.emitOpen(); + socket.emitMessage( + Buffer.from( + JSON.stringify({ + ...posted, + event: "post_edited", + }), + ), + ); + await new Promise((resolve) => { + queueMicrotask(resolve); + }); + expect(onPosted).not.toHaveBeenCalled(); + socket.emitMessage(Buffer.from(JSON.stringify(posted))); + await vi.waitFor(() => { + expect(onPosted).toHaveBeenCalledTimes(1); + }); + socket.emitClose(1000); + await connected; + + expect(onPosted).toHaveBeenCalledWith(JSON.stringify(posted)); + }); + it("terminates when bot update_at changes (disable/enable cycle)", async () => { vi.useFakeTimers(); const socket = new FakeWebSocket(); diff --git a/extensions/mattermost/src/mattermost/monitor-websocket.ts b/extensions/mattermost/src/mattermost/monitor-websocket.ts index b3418f5028d7..5c8417602ed1 100644 --- a/extensions/mattermost/src/mattermost/monitor-websocket.ts +++ b/extensions/mattermost/src/mattermost/monitor-websocket.ts @@ -15,7 +15,7 @@ import type { ChannelAccountSnapshot, RuntimeEnv } from "./runtime-api.js"; export type MattermostEventPayload = { event?: string; data?: { - post?: string | MattermostPost; + post?: unknown; reaction?: string | Record; channel_id?: string; channel_name?: string; @@ -58,7 +58,8 @@ const MattermostEventPayloadSchema = z.object({ event: z.string().optional(), data: z .object({ - post: z.union([z.string(), MattermostPostSchema]).optional(), + // Durable ingress validates the post only after claiming the raw envelope. + post: z.unknown().optional(), reaction: z.union([z.string(), z.record(z.string(), z.unknown())]).optional(), channel_id: z.string().optional(), channel_name: z.string().optional(), @@ -77,11 +78,11 @@ const MattermostEventPayloadSchema = z.object({ .optional(), }) as z.ZodType; -function parseMattermostEventPayload(raw: string): MattermostEventPayload | null { +export function parseMattermostEventPayload(raw: string): MattermostEventPayload | null { return safeParseJsonWithSchema(MattermostEventPayloadSchema, raw); } -function parseMattermostPost(value: unknown): MattermostPost | null { +export function parseMattermostPost(value: unknown): MattermostPost | null { if (typeof value === "string") { return safeParseJsonWithSchema(MattermostPostSchema, value); } @@ -105,7 +106,7 @@ type CreateMattermostConnectOnceOpts = { statusSink?: (patch: Partial) => void; runtime: RuntimeEnv; nextSeq: () => number; - onPosted: (post: MattermostPost, payload: MattermostEventPayload) => Promise; + onPosted: (rawEvent: string) => Promise; onReaction?: (payload: MattermostEventPayload) => Promise; webSocketFactory?: MattermostWebSocketFactory; /** @@ -129,23 +130,6 @@ const defaultMattermostWebSocketFactory: MattermostWebSocketFactory = (url, opti }) as MattermostWebSocketLike; }; -function parsePostedPayload( - payload: MattermostEventPayload, -): { payload: MattermostEventPayload; post: MattermostPost } | null { - if (payload.event !== "posted") { - return null; - } - const postData = payload.data?.post; - if (!postData) { - return null; - } - const post = parseMattermostPost(postData); - if (!post) { - return null; - } - return { payload, post }; -} - export function createMattermostConnectOnce( opts: CreateMattermostConnectOnceOpts, ): () => Promise { @@ -375,14 +359,17 @@ export function createMattermostConnectOnce( if (payload.event !== "posted") { return; } - const parsed = parsePostedPayload(payload); - if (!parsed) { - return; - } try { - await opts.onPosted(parsed.post, parsed.payload); + await opts.onPosted(raw); } catch (err) { - opts.runtime.error?.(`mattermost handler failed: ${String(err)}`); + // Durable admission failed after retries: this post is lost and the + // websocket cannot nack or replay. Tear the connection down loudly + // so the outage is operator-visible instead of silently dropping + // every subsequent post against a broken store. + opts.runtime.error?.( + `mattermost durable admission failed; terminating websocket: ${String(err)}`, + ); + ws.terminate(); } }); diff --git a/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts b/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts index dcc3be00bd6d..a2832ef51fba 100644 --- a/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts +++ b/extensions/mattermost/src/mattermost/monitor.inbound-system-event.test.ts @@ -1,6 +1,8 @@ // Mattermost tests cover monitor.inbound system event plugin behavior. import { createInboundDebouncer } from "openclaw/plugin-sdk/channel-inbound-debounce"; import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { MattermostPost } from "./client.js"; +import type { MattermostEventPayload } from "./monitor-websocket.js"; import { monitorMattermostProvider } from "./monitor.js"; import type { OpenClawConfig, ReplyPayload, RuntimeEnv } from "./runtime-api.js"; @@ -144,6 +146,36 @@ vi.mock("./monitor-resources.js", async (importOriginal) => ({ }), })); +vi.mock("./monitor-ingress.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + createMattermostIngressMonitor: ( + options: Parameters[0], + ) => ({ + receive: async (rawEvent: string) => { + const payload = JSON.parse(rawEvent) as MattermostEventPayload; + const post = + typeof payload.data?.post === "string" + ? (JSON.parse(payload.data.post) as MattermostPost) + : (payload.data?.post as MattermostPost | undefined); + if (payload.event !== "posted" || !post) { + return; + } + await options.dispatch(post, payload, { + abortSignal: new AbortController().signal, + onAdopted: async () => {}, + onDeferred: () => {}, + onAdoptionFinalizing: () => {}, + onAbandoned: async () => {}, + }); + }, + stop: async () => {}, + waitForIdle: async () => {}, + }), + }; +}); + vi.mock("./monitor-slash.js", () => ({ registerMattermostMonitorSlashCommands: mockState.registerMattermostMonitorSlashCommands, })); diff --git a/extensions/mattermost/src/mattermost/monitor.ts b/extensions/mattermost/src/mattermost/monitor.ts index 255d641b25a8..66bb571e3e6d 100644 --- a/extensions/mattermost/src/mattermost/monitor.ts +++ b/extensions/mattermost/src/mattermost/monitor.ts @@ -6,6 +6,7 @@ import { type ChannelInboundTurnPlan, } from "openclaw/plugin-sdk/channel-inbound"; import { + bindIngressLifecycleToReplyOptions, buildChannelProgressDraftLineForEntry, createChannelProgressDraftCompositor, } from "openclaw/plugin-sdk/channel-outbound"; @@ -77,8 +78,12 @@ import { normalizeMention, shouldDropEmptyMattermostBody, } from "./monitor-helpers.js"; +import { + buildMattermostFlushIngressLifecycle, + createMattermostIngressMonitor, + type MattermostIngressLifecycle, +} from "./monitor-ingress.js"; import { resolveOncharPrefixes, stripOncharPrefix } from "./monitor-onchar.js"; -import { processMattermostReplayGuardedPost } from "./monitor-replay.js"; import { createMattermostMonitorResources, formatMattermostInboundMediaText, @@ -974,6 +979,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} const handlePost = async ( post: MattermostPost, payload: MattermostEventPayload, + turnAdoptionLifecycle?: MattermostIngressLifecycle, messageIds?: string[], ) => { const channelId = post.channel_id ?? payload.data?.channel_id ?? payload.broadcast?.channel_id; @@ -982,897 +988,873 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} return; } - const allMessageIds = messageIds?.length ? messageIds : post.id ? [post.id] : []; - if (allMessageIds.length === 0) { + if (!post.id) { logVerboseMessage("mattermost: drop post (missing message id)"); return; } - const replayResult = await processMattermostReplayGuardedPost({ - accountId: account.accountId, - messageIds: allMessageIds, - handlePost: async () => { - const senderId = post.user_id ?? payload.broadcast?.user_id; - if (!senderId) { - logVerboseMessage("mattermost: drop post (missing sender id)"); - return; - } - if (senderId === botUserId) { - logVerboseMessage(`mattermost: drop post (self sender=${senderId})`); - return; - } - if (isSystemPost(post)) { - logVerboseMessage(`mattermost: drop post (system post type=${post.type ?? "unknown"})`); - return; - } + const allMessageIds = messageIds?.length ? messageIds : [post.id]; + const senderId = post.user_id ?? payload.broadcast?.user_id; + if (!senderId) { + logVerboseMessage("mattermost: drop post (missing sender id)"); + return; + } + if (senderId === botUserId) { + logVerboseMessage(`mattermost: drop post (self sender=${senderId})`); + return; + } + if (isSystemPost(post)) { + logVerboseMessage(`mattermost: drop post (system post type=${post.type ?? "unknown"})`); + return; + } - const channelInfo = await resolveChannelInfo(channelId); - const channelType = - normalizeOptionalString(channelInfo?.type) ?? - normalizeOptionalString(payload.data?.channel_type); - if (!channelType) { - logVerboseMessage(`mattermost: drop post (cannot resolve channel type for ${channelId})`); + const channelInfo = await resolveChannelInfo(channelId); + const channelType = + normalizeOptionalString(channelInfo?.type) ?? + normalizeOptionalString(payload.data?.channel_type); + if (!channelType) { + logVerboseMessage(`mattermost: drop post (cannot resolve channel type for ${channelId})`); + return; + } + const kind = resolveMattermostTrustedChatKind({ + channelType, + }); + const chatType = channelChatType(kind); + + const senderName = + normalizeOptionalString(payload.data?.sender_name) ?? + normalizeOptionalString((await resolveUserInfo(senderId))?.username) ?? + senderId; + const rawPostText = typeof post.message === "string" ? post.message : ""; + const rawText = normalizeOptionalString(rawPostText) ?? ""; + const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ + cfg, + surface: "mattermost", + }); + const isControlCommand = + allowTextCommands && core.channel.commands.isControlCommandMessage(rawText, cfg); + const accessDecision = await resolveMattermostMonitorInboundAccess({ + account, + cfg, + senderId, + senderName, + channelId, + kind, + groupPolicy, + readStoreAllowFrom: pairing.readAllowFromStore, + allowTextCommands, + hasControlCommand: isControlCommand, + eventKind: "message", + mayPair: true, + }); + const commandAuthorized = accessDecision.commandAccess.authorized; + + if (accessDecision.ingress.decision !== "allow") { + if (kind === "direct") { + if (accessDecision.ingress.reasonCode === "dm_policy_disabled") { + logVerboseMessage(`mattermost: drop dm (dmPolicy=disabled sender=${senderId})`); return; } - const kind = resolveMattermostTrustedChatKind({ - channelType, - }); - const chatType = channelChatType(kind); - - const senderName = - normalizeOptionalString(payload.data?.sender_name) ?? - normalizeOptionalString((await resolveUserInfo(senderId))?.username) ?? - senderId; - const rawPostText = typeof post.message === "string" ? post.message : ""; - const rawText = normalizeOptionalString(rawPostText) ?? ""; - const allowTextCommands = core.channel.commands.shouldHandleTextCommands({ - cfg, - surface: "mattermost", - }); - const isControlCommand = - allowTextCommands && core.channel.commands.isControlCommandMessage(rawText, cfg); - const accessDecision = await resolveMattermostMonitorInboundAccess({ - account, - cfg, - senderId, - senderName, - channelId, - kind, - groupPolicy, - readStoreAllowFrom: pairing.readAllowFromStore, - allowTextCommands, - hasControlCommand: isControlCommand, - eventKind: "message", - mayPair: true, - }); - const commandAuthorized = accessDecision.commandAccess.authorized; - - if (accessDecision.ingress.decision !== "allow") { - if (kind === "direct") { - if (accessDecision.ingress.reasonCode === "dm_policy_disabled") { - logVerboseMessage(`mattermost: drop dm (dmPolicy=disabled sender=${senderId})`); - return; - } - if (accessDecision.ingress.decision === "pairing") { - const { code, created } = await pairing.upsertPairingRequest({ - id: senderId, - meta: { name: senderName }, - }); - logVerboseMessage( - `mattermost: pairing request sender=${senderId} created=${created}`, + if (accessDecision.ingress.decision === "pairing") { + const { code, created } = await pairing.upsertPairingRequest({ + id: senderId, + meta: { name: senderName }, + }); + logVerboseMessage(`mattermost: pairing request sender=${senderId} created=${created}`); + if (created) { + try { + await sendMessageMattermost( + `user:${senderId}`, + core.channel.pairing.buildPairingReply({ + channel: "mattermost", + idLine: `Your Mattermost user id: ${senderId}`, + code, + }), + { cfg, accountId: account.accountId }, ); - if (created) { - try { - await sendMessageMattermost( - `user:${senderId}`, - core.channel.pairing.buildPairingReply({ - channel: "mattermost", - idLine: `Your Mattermost user id: ${senderId}`, - code, - }), - { cfg, accountId: account.accountId }, - ); - opts.statusSink?.({ lastOutboundAt: Date.now() }); - } catch (err) { - logVerboseMessage( - `mattermost: pairing reply failed for ${senderId}: ${String(err)}`, - ); - } - } - return; + opts.statusSink?.({ lastOutboundAt: Date.now() }); + } catch (err) { + logVerboseMessage(`mattermost: pairing reply failed for ${senderId}: ${String(err)}`); } - logVerboseMessage( - formatMattermostDirectMessageDropLog({ - senderId, - dmPolicy, - reasonCode: accessDecision.senderAccess.reasonCode, - }), - ); - return; } - if (accessDecision.ingress.reasonCode === "group_policy_disabled") { - logVerboseMessage("mattermost: drop group message (groupPolicy=disabled)"); - return; - } - if (accessDecision.ingress.reasonCode === "group_policy_empty_allowlist") { - logVerboseMessage("mattermost: drop group message (no group allowlist)"); - return; - } - if (accessDecision.ingress.reasonCode === "group_policy_not_allowlisted") { - logVerboseMessage(`mattermost: drop group sender=${senderId} (not in groupAllowFrom)`); - return; - } - logVerboseMessage( - `mattermost: drop group message (groupPolicy=${groupPolicy} reason=${accessDecision.senderAccess.reasonCode})`, - ); return; } - - if (kind !== "direct" && accessDecision.commandAccess.shouldBlockControlCommand) { - logInboundDrop({ - log: logVerboseMessage, - channel: "mattermost", - reason: "control command (unauthorized)", - target: senderId, - }); - return; - } - - const teamId = payload.data?.team_id ?? channelInfo?.team_id ?? undefined; - const channelName = payload.data?.channel_name ?? channelInfo?.name ?? ""; - const channelDisplay = - payload.data?.channel_display_name ?? channelInfo?.display_name ?? channelName; - const roomLabel = channelName ? `#${channelName}` : channelDisplay || `#${channelId}`; - - const route = core.channel.routing.resolveAgentRoute({ - cfg, - channel: "mattermost", - accountId: account.accountId, - teamId, - peer: { - kind, - id: kind === "direct" ? senderId : channelId, - }, - }); - - const baseSessionKey = route.sessionKey; - const threadRootId = normalizeOptionalString(post.root_id); - const replyToMode = resolveMattermostReplyToMode(account, kind); - const threadContext = resolveMattermostThreadSessionContext({ - baseSessionKey, - kind, - postId: post.id, - replyToMode, - threadRootId, - }); - const { effectiveReplyToId, sessionKey, parentSessionKey } = threadContext; - const historyKey = resolveMattermostPendingHistoryKey({ kind, sessionKey }); - - const mentionRegexes = core.channel.mentions.buildMentionRegexes(cfg, route.agentId); - const wasMentioned = - kind !== "direct" && - ((botUsername - ? normalizeLowercaseStringOrEmpty(rawText).includes( - `@${normalizeLowercaseStringOrEmpty(botUsername)}`, - ) - : false) || - core.channel.mentions.matchesMentionPatterns(rawText, mentionRegexes)); - const pendingBody = - rawText || - (post.file_ids?.length - ? `[Mattermost ${post.file_ids.length === 1 ? "file" : "files"}]` - : ""); - const pendingSender = senderName; - const recordPendingHistory = () => { - const trimmed = pendingBody.trim(); - createChannelHistoryWindow({ historyMap: channelHistories }).record({ - limit: historyLimit, - historyKey: historyKey ?? "", - entry: - historyKey && trimmed - ? { - sender: pendingSender, - body: trimmed, - timestamp: typeof post.create_at === "number" ? post.create_at : undefined, - messageId: post.id ?? undefined, - } - : null, - }); - }; - - const oncharEnabled = account.chatmode === "onchar" && kind !== "direct"; - const oncharPrefixes = oncharEnabled ? resolveOncharPrefixes(account.oncharPrefixes) : []; - const oncharResult = oncharEnabled - ? stripOncharPrefix(rawText, oncharPrefixes) - : { triggered: false, stripped: rawText }; - const oncharTriggered = oncharResult.triggered; - const canDetectMention = Boolean(botUsername) || mentionRegexes.length > 0; - // Threads the bot already replied in auto-engage: follow-ups resume without - // a re-mention even under requireMention. Keyed by the thread root id. - const threadAlreadyEngaged = - kind !== "direct" && effectiveReplyToId - ? await hasMattermostThreadParticipationWithPersistence({ - accountId: account.accountId, - channelId, - threadRootId: effectiveReplyToId, - }) - : false; - const shouldRequireMention = - kind !== "direct" && - core.channel.groups.resolveRequireMention({ - cfg, - channel: "mattermost", - accountId: account.accountId, - groupId: channelId, - requireMentionOverride: account.requireMention, - }); - const implicitMentionKinds = implicitMentionKindWhen( - "bot_thread_participant", - threadAlreadyEngaged, + logVerboseMessage( + formatMattermostDirectMessageDropLog({ + senderId, + dmPolicy, + reasonCode: accessDecision.senderAccess.reasonCode, + }), ); - const mentionDecision = resolveMattermostInboundMentionDecision({ - cfg, - accountId: account.accountId, - kind, - requireMention: shouldRequireMention || oncharEnabled, - canDetectMention: canDetectMention || oncharEnabled, - wasMentioned: wasMentioned || oncharTriggered, - implicitMentionKinds, - allowTextCommands, - hasControlCommand: isControlCommand, - commandAuthorized, - }); - const { shouldBypassMention } = mentionDecision; + return; + } + if (accessDecision.ingress.reasonCode === "group_policy_disabled") { + logVerboseMessage("mattermost: drop group message (groupPolicy=disabled)"); + return; + } + if (accessDecision.ingress.reasonCode === "group_policy_empty_allowlist") { + logVerboseMessage("mattermost: drop group message (no group allowlist)"); + return; + } + if (accessDecision.ingress.reasonCode === "group_policy_not_allowlisted") { + logVerboseMessage(`mattermost: drop group sender=${senderId} (not in groupAllowFrom)`); + return; + } + logVerboseMessage( + `mattermost: drop group message (groupPolicy=${groupPolicy} reason=${accessDecision.senderAccess.reasonCode})`, + ); + return; + } - if ( - mentionDecision.shouldSkip && - oncharEnabled && - !oncharTriggered && - !wasMentioned && - !shouldBypassMention - ) { - logVerboseMessage( - `mattermost: drop group message (onchar not triggered channel=${channelId} sender=${senderId})`, - ); - recordPendingHistory(); - return; - } + if (kind !== "direct" && accessDecision.commandAccess.shouldBlockControlCommand) { + logInboundDrop({ + log: logVerboseMessage, + channel: "mattermost", + reason: "control command (unauthorized)", + target: senderId, + }); + return; + } - if (mentionDecision.shouldSkip) { - logVerboseMessage( - `mattermost: drop group message (missing mention channel=${channelId} sender=${senderId} requireMention=${shouldRequireMention} bypass=${shouldBypassMention} canDetectMention=${canDetectMention})`, - ); - recordPendingHistory(); - return; - } - const fileIds = uniqueStrings(normalizeTrimmedStringList(post.file_ids ?? [])); - const mediaList = await resolveMattermostMedia(fileIds); - const mediaPlaceholder = buildMattermostAttachmentPlaceholder(mediaList); - const bodySource = oncharTriggered ? oncharResult.stripped : rawText; - const downloadedText = [bodySource, mediaPlaceholder].filter(Boolean).join("\n").trim(); - const baseText = formatMattermostInboundMediaText({ - body: downloadedText, - mediaPlaceholder, - expectedCount: fileIds.length, - mediaCount: mediaList.length, - }); - const bodyText = normalizeMention(baseText, botUsername); - if (shouldDropEmptyMattermostBody({ bodyText, rawText: rawPostText, botUsername })) { - logVerboseMessage( - `mattermost: drop message (empty body after normalization channel=${channelId} sender=${senderId} wasMentioned=${wasMentioned})`, - ); - return; - } - // Mention-only turns need non-empty agent text; the shared reply runner rejects empty - // bodies before model invocation. The guard above ensures this fallback is a bot mention. - const bodyForAgent = bodyText || rawText.trim(); + const teamId = payload.data?.team_id ?? channelInfo?.team_id ?? undefined; + const channelName = payload.data?.channel_name ?? channelInfo?.name ?? ""; + const channelDisplay = + payload.data?.channel_display_name ?? channelInfo?.display_name ?? channelName; + const roomLabel = channelName ? `#${channelName}` : channelDisplay || `#${channelId}`; - core.channel.activity.record({ - channel: "mattermost", - accountId: account.accountId, - direction: "inbound", - }); + const route = core.channel.routing.resolveAgentRoute({ + cfg, + channel: "mattermost", + accountId: account.accountId, + teamId, + peer: { + kind, + id: kind === "direct" ? senderId : channelId, + }, + }); - const fromLabel = formatInboundFromLabel({ - isGroup: kind !== "direct", - groupLabel: channelDisplay || roomLabel, - groupId: channelId, - groupFallback: roomLabel || "Channel", - directLabel: senderName, - directId: senderId, - }); + const baseSessionKey = route.sessionKey; + const threadRootId = normalizeOptionalString(post.root_id); + const replyToMode = resolveMattermostReplyToMode(account, kind); + const threadContext = resolveMattermostThreadSessionContext({ + baseSessionKey, + kind, + postId: post.id, + replyToMode, + threadRootId, + }); + const { effectiveReplyToId, sessionKey, parentSessionKey } = threadContext; + const historyKey = resolveMattermostPendingHistoryKey({ kind, sessionKey }); - const textWithId = `${bodyText}\n[mattermost message id: ${post.id ?? "unknown"} channel: ${channelId}]`; - const body = formatInboundEnvelope({ - channel: "Mattermost", - from: fromLabel, - timestamp: typeof post.create_at === "number" ? post.create_at : undefined, - body: textWithId, - chatType, - sender: { name: senderName, id: senderId }, - }); - let combinedBody = body; - if (historyKey) { - const channelHistory = createChannelHistoryWindow({ historyMap: channelHistories }); - combinedBody = channelHistory.buildPendingContext({ + const mentionRegexes = core.channel.mentions.buildMentionRegexes(cfg, route.agentId); + const wasMentioned = + kind !== "direct" && + ((botUsername + ? normalizeLowercaseStringOrEmpty(rawText).includes( + `@${normalizeLowercaseStringOrEmpty(botUsername)}`, + ) + : false) || + core.channel.mentions.matchesMentionPatterns(rawText, mentionRegexes)); + const pendingBody = + rawText || + (post.file_ids?.length + ? `[Mattermost ${post.file_ids.length === 1 ? "file" : "files"}]` + : ""); + const pendingSender = senderName; + const recordPendingHistory = () => { + const trimmed = pendingBody.trim(); + createChannelHistoryWindow({ historyMap: channelHistories }).record({ + limit: historyLimit, + historyKey: historyKey ?? "", + entry: + historyKey && trimmed + ? { + sender: pendingSender, + body: trimmed, + timestamp: typeof post.create_at === "number" ? post.create_at : undefined, + messageId: post.id ?? undefined, + } + : null, + }); + }; + + const oncharEnabled = account.chatmode === "onchar" && kind !== "direct"; + const oncharPrefixes = oncharEnabled ? resolveOncharPrefixes(account.oncharPrefixes) : []; + const oncharResult = oncharEnabled + ? stripOncharPrefix(rawText, oncharPrefixes) + : { triggered: false, stripped: rawText }; + const oncharTriggered = oncharResult.triggered; + const canDetectMention = Boolean(botUsername) || mentionRegexes.length > 0; + // Threads the bot already replied in auto-engage: follow-ups resume without + // a re-mention even under requireMention. Keyed by the thread root id. + const threadAlreadyEngaged = + kind !== "direct" && effectiveReplyToId + ? await hasMattermostThreadParticipationWithPersistence({ + accountId: account.accountId, + channelId, + threadRootId: effectiveReplyToId, + }) + : false; + const shouldRequireMention = + kind !== "direct" && + core.channel.groups.resolveRequireMention({ + cfg, + channel: "mattermost", + accountId: account.accountId, + groupId: channelId, + requireMentionOverride: account.requireMention, + }); + const implicitMentionKinds = implicitMentionKindWhen( + "bot_thread_participant", + threadAlreadyEngaged, + ); + const mentionDecision = resolveMattermostInboundMentionDecision({ + cfg, + accountId: account.accountId, + kind, + requireMention: shouldRequireMention || oncharEnabled, + canDetectMention: canDetectMention || oncharEnabled, + wasMentioned: wasMentioned || oncharTriggered, + implicitMentionKinds, + allowTextCommands, + hasControlCommand: isControlCommand, + commandAuthorized, + }); + const { shouldBypassMention } = mentionDecision; + + if ( + mentionDecision.shouldSkip && + oncharEnabled && + !oncharTriggered && + !wasMentioned && + !shouldBypassMention + ) { + logVerboseMessage( + `mattermost: drop group message (onchar not triggered channel=${channelId} sender=${senderId})`, + ); + recordPendingHistory(); + return; + } + + if (mentionDecision.shouldSkip) { + logVerboseMessage( + `mattermost: drop group message (missing mention channel=${channelId} sender=${senderId} requireMention=${shouldRequireMention} bypass=${shouldBypassMention} canDetectMention=${canDetectMention})`, + ); + recordPendingHistory(); + return; + } + const fileIds = uniqueStrings(normalizeTrimmedStringList(post.file_ids ?? [])); + const mediaList = await resolveMattermostMedia(fileIds); + const mediaPlaceholder = buildMattermostAttachmentPlaceholder(mediaList); + const bodySource = oncharTriggered ? oncharResult.stripped : rawText; + const downloadedText = [bodySource, mediaPlaceholder].filter(Boolean).join("\n").trim(); + const baseText = formatMattermostInboundMediaText({ + body: downloadedText, + mediaPlaceholder, + expectedCount: fileIds.length, + mediaCount: mediaList.length, + }); + const bodyText = normalizeMention(baseText, botUsername); + if (shouldDropEmptyMattermostBody({ bodyText, rawText: rawPostText, botUsername })) { + logVerboseMessage( + `mattermost: drop message (empty body after normalization channel=${channelId} sender=${senderId} wasMentioned=${wasMentioned})`, + ); + return; + } + // Mention-only turns need non-empty agent text; the shared reply runner rejects empty + // bodies before model invocation. The guard above ensures this fallback is a bot mention. + const bodyForAgent = bodyText || rawText.trim(); + + core.channel.activity.record({ + channel: "mattermost", + accountId: account.accountId, + direction: "inbound", + }); + + const fromLabel = formatInboundFromLabel({ + isGroup: kind !== "direct", + groupLabel: channelDisplay || roomLabel, + groupId: channelId, + groupFallback: roomLabel || "Channel", + directLabel: senderName, + directId: senderId, + }); + + const textWithId = `${bodyText}\n[mattermost message id: ${post.id ?? "unknown"} channel: ${channelId}]`; + const body = formatInboundEnvelope({ + channel: "Mattermost", + from: fromLabel, + timestamp: typeof post.create_at === "number" ? post.create_at : undefined, + body: textWithId, + chatType, + sender: { name: senderName, id: senderId }, + }); + let combinedBody = body; + if (historyKey) { + const channelHistory = createChannelHistoryWindow({ historyMap: channelHistories }); + combinedBody = channelHistory.buildPendingContext({ + historyKey, + limit: historyLimit, + currentMessage: combinedBody, + formatEntry: (entry) => + formatInboundEnvelope({ + channel: "Mattermost", + from: fromLabel, + timestamp: entry.timestamp, + body: `${entry.body}${ + entry.messageId ? ` [id:${entry.messageId} channel:${channelId}]` : "" + }`, + chatType, + senderLabel: entry.sender, + }), + }); + } + + const to = kind === "direct" ? `user:${senderId}` : `channel:${channelId}`; + const mediaPayload = buildAgentMediaPayload(mediaList); + const commandBody = rawText.trim(); + const inboundHistory = + historyKey && historyLimit > 0 + ? createChannelHistoryWindow({ historyMap: channelHistories }).buildInboundHistory({ historyKey, limit: historyLimit, - currentMessage: combinedBody, - formatEntry: (entry) => - formatInboundEnvelope({ - channel: "Mattermost", - from: fromLabel, - timestamp: entry.timestamp, - body: `${entry.body}${ - entry.messageId ? ` [id:${entry.messageId} channel:${channelId}]` : "" - }`, - chatType, - senderLabel: entry.sender, - }), - }); - } + }) + : undefined; + const ctxPayload = finalizeInboundContext({ + Body: combinedBody, + BodyForAgent: bodyForAgent, + InboundHistory: inboundHistory, + RawBody: commandBody, + CommandBody: commandBody, + BodyForCommands: commandBody, + From: + kind === "direct" + ? `mattermost:${senderId}` + : kind === "group" + ? `mattermost:group:${channelId}` + : `mattermost:channel:${channelId}`, + To: to, + SessionKey: sessionKey, + DmScope: route.dmScope, + ParentSessionKey: parentSessionKey, + AccountId: route.accountId, + ChatType: chatType, + ConversationLabel: fromLabel, + GroupSubject: kind !== "direct" ? channelDisplay || roomLabel : undefined, + GroupChannel: channelName ? `#${channelName}` : undefined, + GroupSpace: teamId, + SenderName: senderName, + SenderId: senderId, + Provider: "mattermost" as const, + Surface: "mattermost" as const, + MessageSid: post.id ?? undefined, + MessageSids: allMessageIds.length > 1 ? allMessageIds : undefined, + MessageSidFirst: allMessageIds.length > 1 ? allMessageIds[0] : undefined, + MessageSidLast: + allMessageIds.length > 1 ? allMessageIds[allMessageIds.length - 1] : undefined, + ReplyToId: effectiveReplyToId, + MessageThreadId: effectiveReplyToId, + Timestamp: typeof post.create_at === "number" ? post.create_at : undefined, + WasMentioned: kind !== "direct" ? mentionDecision.effectiveWasMentioned : undefined, + CommandAuthorized: commandAuthorized, + // Tag typed text-slash control commands (e.g. ` /new`, ` /reset` sent via the regular + // post path rather than Mattermost's native slash UI) so the explicit-command turn + // exception in source-reply-delivery-mode.ts surfaces their acknowledgements under + // message_tool_only delivery modes (e.g. Codex harness DMs). Mirrors iMessage #82642. + CommandSource: commandAuthorized && isControlCommand ? ("text" as const) : undefined, + OriginatingChannel: "mattermost" as const, + OriginatingTo: to, + ...mediaPayload, + }); + const pinnedMainDmOwner = + kind === "direct" + ? resolvePinnedMainDmOwnerFromAllowlist({ + dmScope: cfg.session?.dmScope, + allowFrom: account.config.allowFrom, + normalizeEntry: normalizeMattermostAllowEntry, + }) + : null; - const to = kind === "direct" ? `user:${senderId}` : `channel:${channelId}`; - const mediaPayload = buildAgentMediaPayload(mediaList); - const commandBody = rawText.trim(); - const inboundHistory = - historyKey && historyLimit > 0 - ? createChannelHistoryWindow({ historyMap: channelHistories }).buildInboundHistory({ - historyKey, - limit: historyLimit, - }) - : undefined; - const ctxPayload = finalizeInboundContext({ - Body: combinedBody, - BodyForAgent: bodyForAgent, - InboundHistory: inboundHistory, - RawBody: commandBody, - CommandBody: commandBody, - BodyForCommands: commandBody, - From: - kind === "direct" - ? `mattermost:${senderId}` - : kind === "group" - ? `mattermost:group:${channelId}` - : `mattermost:channel:${channelId}`, - To: to, - SessionKey: sessionKey, - DmScope: route.dmScope, - ParentSessionKey: parentSessionKey, - AccountId: route.accountId, - ChatType: chatType, - ConversationLabel: fromLabel, - GroupSubject: kind !== "direct" ? channelDisplay || roomLabel : undefined, - GroupChannel: channelName ? `#${channelName}` : undefined, - GroupSpace: teamId, - SenderName: senderName, - SenderId: senderId, - Provider: "mattermost" as const, - Surface: "mattermost" as const, - MessageSid: post.id ?? undefined, - MessageSids: allMessageIds.length > 1 ? allMessageIds : undefined, - MessageSidFirst: allMessageIds.length > 1 ? allMessageIds[0] : undefined, - MessageSidLast: - allMessageIds.length > 1 ? allMessageIds[allMessageIds.length - 1] : undefined, - ReplyToId: effectiveReplyToId, - MessageThreadId: effectiveReplyToId, - Timestamp: typeof post.create_at === "number" ? post.create_at : undefined, - WasMentioned: kind !== "direct" ? mentionDecision.effectiveWasMentioned : undefined, - CommandAuthorized: commandAuthorized, - // Tag typed text-slash control commands (e.g. ` /new`, ` /reset` sent via the regular - // post path rather than Mattermost's native slash UI) so the explicit-command turn - // exception in source-reply-delivery-mode.ts surfaces their acknowledgements under - // message_tool_only delivery modes (e.g. Codex harness DMs). Mirrors iMessage #82642. - CommandSource: commandAuthorized && isControlCommand ? ("text" as const) : undefined, - OriginatingChannel: "mattermost" as const, - OriginatingTo: to, - ...mediaPayload, - }); - const pinnedMainDmOwner = - kind === "direct" - ? resolvePinnedMainDmOwnerFromAllowlist({ - dmScope: cfg.session?.dmScope, - allowFrom: account.config.allowFrom, - normalizeEntry: normalizeMattermostAllowEntry, - }) - : null; + const previewLine = truncateUtf16Safe(bodyText, 200).replace(/\n/g, "\\n"); + logVerboseMessage( + `mattermost inbound: from=${ctxPayload.From} len=${bodyText.length} preview="${previewLine}"`, + ); - const previewLine = truncateUtf16Safe(bodyText, 200).replace(/\n/g, "\\n"); - logVerboseMessage( - `mattermost inbound: from=${ctxPayload.From} len=${bodyText.length} preview="${previewLine}"`, - ); + const textLimit = core.channel.text.resolveTextChunkLimit( + cfg, + "mattermost", + account.accountId, + { + fallbackLimit: account.textChunkLimit ?? 4000, + }, + ); + const tableMode = core.channel.text.resolveMarkdownTableMode({ + cfg, + channel: "mattermost", + accountId: account.accountId, + }); + const chunkMode = core.channel.text.resolveChunkMode(cfg, "mattermost", account.accountId); - const textLimit = core.channel.text.resolveTextChunkLimit( - cfg, - "mattermost", - account.accountId, - { - fallbackLimit: account.textChunkLimit ?? 4000, - }, - ); - const tableMode = core.channel.text.resolveMarkdownTableMode({ - cfg, - channel: "mattermost", - accountId: account.accountId, - }); - const chunkMode = core.channel.text.resolveChunkMode(cfg, "mattermost", account.accountId); - - const { onModelSelected, typingCallbacks, resolveResponsePrefix, ...replyPipeline } = - createChannelMessageReplyPipeline({ - cfg, - agentId: route.agentId, - channel: "mattermost", - accountId: account.accountId, - typing: { - start: () => sendTypingIndicator(channelId, effectiveReplyToId), - onStartError: (err) => { - logTypingFailure({ - log: (message) => logger.debug?.(message), - channel: "mattermost", - target: channelId, - error: err, - }); - }, - }, - }); - const draftPreviewEnabled = account.streamingMode !== "off"; - const draftToolProgressEnabled = shouldUpdateMattermostDraftToolProgress(account); - const suppressDefaultToolProgressMessages = - shouldSuppressMattermostDefaultToolProgressMessages(account); - const draftStream = draftPreviewEnabled - ? createMattermostDraftStream({ - client, - channelId, - rootId: effectiveReplyToId, - throttleMs: 1200, - chunkText: (value) => - core.channel.text.chunkMarkdownTextWithMode( - core.channel.text.convertMarkdownTables(value, tableMode), - textLimit, - chunkMode, - ), - log: logVerboseMessage, - warn: logVerboseMessage, - }) - : createDisabledMattermostDraftStream(); - const previewBoundaryController = createMattermostDraftPreviewBoundaryController({ - enabled: draftPreviewEnabled && account.streamingMode === "block", - forceNewMessage: async () => { - await draftStream.forceNewMessage(); - }, - }); - let lastPartialText = ""; - let firstAssistantPreviewPrefix: string | undefined; - let firstAssistantPreviewPrefixPending = true; - let currentAssistantPreviewUsesPrefix = false; - let blockPreviewActivity: "none" | "reasoning" | "text" | "tool" = "none"; - let blockPreviewAssistantMessagePending = false; - const progressDraft = createChannelProgressDraftCompositor({ - entry: account.config, - mode: account.streamingMode, - active: draftPreviewEnabled, - seed: `${account.accountId}:${channelId}`, - update: async (previewText, options) => { - draftStream.update(previewText); - if (options?.flush) { - await draftStream.flush(); - } - }, - }); - const enterBlockPreviewActivity = (activity: "reasoning" | "text" | "tool") => { - if (account.streamingMode !== "block") { - return undefined; - } - const continuingToolActivity = activity === "tool" && blockPreviewActivity === "tool"; - const continuingTextActivity = - activity === "text" && - blockPreviewActivity === "text" && - !blockPreviewAssistantMessagePending; - const continuingReasoningActivity = - activity === "reasoning" && blockPreviewActivity === "reasoning"; - const continuesCurrentActivity = - continuingToolActivity || continuingTextActivity || continuingReasoningActivity; - // Reasoning placeholders are transient. A visible successor replaces the same draft; - // only entering reasoning from a durable text/tool block rotates generations. - const reusesReasoningGeneration = blockPreviewActivity === "reasoning"; - const startsNewGeneration = !continuesCurrentActivity && !reusesReasoningGeneration; - if (startsNewGeneration) { - currentAssistantPreviewUsesPrefix = false; - } - const boundarySettled = startsNewGeneration - ? previewBoundaryController.noteBoundary() - : undefined; - // Message-start is only a candidate boundary: consecutive tool-only turns stay in the - // same activity post, while the first visible text or reasoning starts a new block. - if (!continuesCurrentActivity) { - progressDraft.reset(); - } - blockPreviewActivity = activity; - blockPreviewAssistantMessagePending = false; - if (activity === "tool") { - lastPartialText = ""; - } - return boundarySettled; - }; - const previewState: MattermostDraftPreviewState = { - finalizedViaPreviewPost: false, - }; - - const resolveFinalDeliveryText = (text?: string) => { - if (typeof text !== "string") { - return undefined; - } - const resolution = draftStream.resolveFinalText(text); - return resolution.kind === "already-delivered" ? "" : resolution.text; - }; - - const resolvePreviewFinalText = (text?: string) => { - const deliveryText = resolveFinalDeliveryText(text); - if (typeof deliveryText !== "string") { - return undefined; - } - const formatted = core.channel.text.convertMarkdownTables(deliveryText, tableMode); - const chunks = core.channel.text.chunkMarkdownTextWithMode( - formatted, - textLimit, - chunkMode, - ); - if (!chunks.length && formatted) { - chunks.push(formatted); - } - if (chunks.length != 1) { - return undefined; - } - const trimmed = chunks[0]?.trim(); - if (!trimmed) { - return undefined; - } - if ( - lastPartialText && - lastPartialText.startsWith(trimmed) && - trimmed.length < lastPartialText.length - ) { - return undefined; - } - return trimmed; - }; - - const updateDraftFromPartial = (text?: string) => { - const cleaned = text?.trim(); - if (!cleaned) { - return undefined; - } - if (cleaned === lastPartialText) { - return undefined; - } - if ( - lastPartialText && - lastPartialText.startsWith(cleaned) && - cleaned.length < lastPartialText.length - ) { - return undefined; - } - const boundarySettled = enterBlockPreviewActivity("text"); - lastPartialText = cleaned; - if (firstAssistantPreviewPrefixPending) { - firstAssistantPreviewPrefix = resolveResponsePrefix?.(); - firstAssistantPreviewPrefixPending = false; - currentAssistantPreviewUsesPrefix = Boolean(firstAssistantPreviewPrefix); - } - const previewText = - currentAssistantPreviewUsesPrefix && firstAssistantPreviewPrefix - ? cleaned.startsWith(firstAssistantPreviewPrefix) - ? cleaned - : `${firstAssistantPreviewPrefix} ${cleaned}` - : cleaned; - draftStream.updateAssistantText(previewText); - previewBoundaryController.noteUpdate(); - return boundarySettled; - }; - - const deliveryBarrier = createMattermostReplyDeliveryBarrier({ - isDirect: kind === "direct", - dmRetryOptions: account.config.dmChannelRetry, - }); - const dispatcherOptions: NonNullable = { - ...replyPipeline, - resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, - onDeliverySettled: deliveryBarrier.markDeliverySettled, - humanDelay: resolveHumanDelayConfig(cfg, route.agentId), - typingCallbacks, - }; - const delivery: ChannelInboundTurnPlan["delivery"] = { - deliver: async (payloadEntry: ReplyPayload, info) => { - if (info.kind === "final") { - await enterBlockPreviewActivity("text"); - // Final text resolution uses only generations confirmed visible. Join prior - // boundary work before the synchronous final-edit decision. - await draftStream.settleBoundaries(); - progressDraft.markFinalReplyStarted(); - } - // A visible same-thread final arrives either via a normal send or by editing - // the draft preview in place; record participation on whichever path fires. - const markThreadParticipation = () => { - if (kind !== "direct" && effectiveReplyToId) { - recordMattermostThreadParticipation( - account.accountId, - channelId, - effectiveReplyToId, - { agentId: route.agentId }, - ); - } - }; - await deliverMattermostReplyWithDraftPreview({ - payload: payloadEntry, - info, - kind, - client, - draftStream, - effectiveReplyToId, - resolvePreviewFinalText, - previewState, - logVerboseMessage, - recordThreadParticipation: markThreadParticipation, - deliverPayload: async (payloadToDeliver) => { - const finalTextResolution = - info.kind === "final" && - !payloadToDeliver.isError && - typeof payloadToDeliver.text === "string" - ? draftStream.resolveFinalText(payloadToDeliver.text) - : undefined; - const resolvedPayload = finalTextResolution - ? { - ...payloadToDeliver, - text: - finalTextResolution.kind === "already-delivered" - ? "" - : finalTextResolution.text, - } - : payloadToDeliver; - const outcome = await deliverMattermostReplyPayload({ - core, - cfg, - payload: resolvedPayload, - to, - accountId: account.accountId, - agentId: route.agentId, - replyToId: resolveMattermostReplyRootId({ - kind, - threadRootId: effectiveReplyToId, - replyToId: payloadToDeliver.replyToId, - }), - textLimit, - tableMode, - sendMessage: sendMessageMattermost, - onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, - }); - // Record only on a visible send so threads we merely observed - // (reasoning-only/empty/suppressed) do not auto-engage later. - if (outcome === "text" || outcome === "media") { - markThreadParticipation(); - } else if ( - outcome === "empty" && - finalTextResolution?.kind === "already-delivered" - ) { - // The terminal payload confirms the already-published assistant block as - // the visible final reply even though this delivery has no remaining text. - markThreadParticipation(); - } - const deliveryLog = formatMattermostFinalDeliveryOutcomeLog({ - outcome, - payload: resolvedPayload, - to, - accountId: account.accountId, - agentId: route.agentId, - }); - if (deliveryLog) { - runtime.log?.(deliveryLog); - } - }, + const { onModelSelected, typingCallbacks, resolveResponsePrefix, ...replyPipeline } = + createChannelMessageReplyPipeline({ + cfg, + agentId: route.agentId, + channel: "mattermost", + accountId: account.accountId, + typing: { + start: () => sendTypingIndicator(channelId, effectiveReplyToId), + onStartError: (err) => { + logTypingFailure({ + log: (message) => logger.debug?.(message), + channel: "mattermost", + target: channelId, + error: err, }); - if (info.kind === "final") { - progressDraft.markFinalReplyDelivered(); - } }, - onError: (err, info) => { - runtime.error?.(`mattermost ${info.kind} reply failed: ${String(err)}`); - }, - }; - - const inboundLastRouteSessionKey = resolveInboundLastRouteSessionKey({ - route, - sessionKey: route.sessionKey, - }); - - try { - await core.channel.inbound.run({ - channel: "mattermost", - accountId: route.accountId, - raw: post, - adapter: { - ingest: () => ({ - id: post.id ?? `${to}:${Date.now()}`, - timestamp: post.create_at ?? undefined, - rawText, - textForAgent: ctxPayload.BodyForAgent, - textForCommands: ctxPayload.CommandBody, - raw: post, - }), - resolveTurn: () => ({ - cfg, - channel: "mattermost", - accountId: route.accountId, - route: { - agentId: route.agentId, - dmScope: route.dmScope, - sessionKey: route.sessionKey, - }, - ctxPayload, - record: { - updateLastRoute: - kind === "direct" - ? { - sessionKey: inboundLastRouteSessionKey, - channel: "mattermost", - to, - accountId: route.accountId, - mainDmOwnerPin: - inboundLastRouteSessionKey === route.mainSessionKey && pinnedMainDmOwner - ? { - ownerRecipient: pinnedMainDmOwner, - senderRecipient: normalizeMattermostAllowEntry(senderId), - onSkip: ({ - ownerRecipient, - senderRecipient, - }: { - ownerRecipient: string; - senderRecipient: string; - }) => { - logVerboseMessage( - `mattermost: skip main-session last route for ${senderRecipient} (pinned owner ${ownerRecipient})`, - ); - }, - } - : undefined, - } - : undefined, - onRecordError: (err) => { - logVerboseMessage( - `mattermost: failed updating session meta id=${post.id ?? "unknown"}: ${String(err)}`, - ); - }, - }, - history: { - isGroup: Boolean(historyKey), - historyKey: historyKey ?? undefined, - historyMap: channelHistories, - limit: historyLimit, - }, - dispatcherOptions, - delivery, - replyOptions: { - allowProgressCallbacksWhenSourceDeliverySuppressed: draftToolProgressEnabled - ? true - : undefined, - preserveProgressCallbackStartOrder: draftPreviewEnabled ? true : undefined, - onObservedReplyDelivery: draftToolProgressEnabled - ? () => draftStream.clear() - : undefined, - disableBlockStreaming: draftPreviewEnabled - ? true - : typeof account.blockStreaming === "boolean" - ? !account.blockStreaming - : undefined, - ...(suppressDefaultToolProgressMessages - ? { suppressDefaultToolProgressMessages: true } - : {}), - onModelSelected, - onPartialReply: (payloadResult) => { - if (account.streamingMode !== "progress") { - return updateDraftFromPartial(payloadResult.text); - } - return undefined; - }, - onAssistantMessageStart: () => { - lastPartialText = ""; - progressDraft.resetReasoningProgress(); - if (account.streamingMode === "block") { - blockPreviewAssistantMessagePending = true; - return; - } - if (account.streamingMode !== "progress") { - progressDraft.reset(); - } - }, - onReasoningEnd: () => { - // Hidden reasoning has no visible boundary. Only transitions that - // actually render text, reasoning, or tools rotate preview posts. - lastPartialText = ""; - progressDraft.resetReasoningProgress(); - if (account.streamingMode !== "block" && account.streamingMode !== "progress") { - progressDraft.reset(); - } - }, - onReasoningStream: async (payloadResult) => { - if (account.streamingMode === "progress") { - await progressDraft.pushReasoningProgress(payloadResult.text || "Thinking…", { - snapshot: payloadResult.isReasoningSnapshot === true, - }); - return; - } - if (!lastPartialText) { - const boundarySettled = enterBlockPreviewActivity("reasoning"); - draftStream.update("Thinking…"); - previewBoundaryController.noteUpdate(); - await boundarySettled; - } - }, - onToolStart: async (payloadValue) => { - if (!draftToolProgressEnabled) { - return; - } - const boundarySettled = enterBlockPreviewActivity("tool"); - // Boundary detach and progress staging both happen synchronously before - // their first await; agent callbacks may be dispatched fire-and-forget. - const progressSettled = progressDraft.pushToolProgress( - buildChannelProgressDraftLineForEntry( - account.config, - { - event: "tool", - itemId: payloadValue.itemId, - toolCallId: payloadValue.toolCallId, - name: payloadValue.name, - phase: payloadValue.phase, - args: payloadValue.args, - }, - payloadValue.detailMode - ? { detailMode: payloadValue.detailMode } - : undefined, - ), - { startImmediately: true }, - ); - previewBoundaryController.noteUpdate(); - await Promise.all([boundarySettled, progressSettled]); - }, - onItemEvent: async (payloadLocal) => { - if (!draftToolProgressEnabled) { - return; - } - const boundarySettled = enterBlockPreviewActivity("tool"); - const progressSettled = progressDraft.pushToolProgress( - buildChannelProgressDraftLineForEntry(account.config, { - event: "item", - itemId: payloadLocal.itemId, - itemKind: payloadLocal.kind, - title: payloadLocal.title, - name: payloadLocal.name, - phase: payloadLocal.phase, - status: payloadLocal.status, - summary: payloadLocal.summary, - progressText: payloadLocal.progressText, - meta: payloadLocal.meta, - }), - { startImmediately: true }, - ); - previewBoundaryController.noteUpdate(); - await Promise.all([boundarySettled, progressSettled]); - }, - }, - }), - }, - }); - } finally { - try { - await draftStream.stop(); - } catch (err) { - logVerboseMessage(`mattermost draft preview cleanup failed: ${String(err)}`); - } + }, + }); + const draftPreviewEnabled = account.streamingMode !== "off"; + const draftToolProgressEnabled = shouldUpdateMattermostDraftToolProgress(account); + const suppressDefaultToolProgressMessages = + shouldSuppressMattermostDefaultToolProgressMessages(account); + const draftStream = draftPreviewEnabled + ? createMattermostDraftStream({ + client, + channelId, + rootId: effectiveReplyToId, + throttleMs: 1200, + chunkText: (value) => + core.channel.text.chunkMarkdownTextWithMode( + core.channel.text.convertMarkdownTables(value, tableMode), + textLimit, + chunkMode, + ), + log: logVerboseMessage, + warn: logVerboseMessage, + }) + : createDisabledMattermostDraftStream(); + const previewBoundaryController = createMattermostDraftPreviewBoundaryController({ + enabled: draftPreviewEnabled && account.streamingMode === "block", + forceNewMessage: async () => { + await draftStream.forceNewMessage(); + }, + }); + let lastPartialText = ""; + let firstAssistantPreviewPrefix: string | undefined; + let firstAssistantPreviewPrefixPending = true; + let currentAssistantPreviewUsesPrefix = false; + let blockPreviewActivity: "none" | "reasoning" | "text" | "tool" = "none"; + let blockPreviewAssistantMessagePending = false; + const progressDraft = createChannelProgressDraftCompositor({ + entry: account.config, + mode: account.streamingMode, + active: draftPreviewEnabled, + seed: `${account.accountId}:${channelId}`, + update: async (previewText, options) => { + draftStream.update(previewText); + if (options?.flush) { + await draftStream.flush(); } }, }); - if (replayResult === "duplicate") { - logVerboseMessage( - `mattermost: drop post (dedupe account=${account.accountId} ids=${allMessageIds.length})`, - ); + const enterBlockPreviewActivity = (activity: "reasoning" | "text" | "tool") => { + if (account.streamingMode !== "block") { + return undefined; + } + const continuingToolActivity = activity === "tool" && blockPreviewActivity === "tool"; + const continuingTextActivity = + activity === "text" && + blockPreviewActivity === "text" && + !blockPreviewAssistantMessagePending; + const continuingReasoningActivity = + activity === "reasoning" && blockPreviewActivity === "reasoning"; + const continuesCurrentActivity = + continuingToolActivity || continuingTextActivity || continuingReasoningActivity; + // Reasoning placeholders are transient. A visible successor replaces the same draft; + // only entering reasoning from a durable text/tool block rotates generations. + const reusesReasoningGeneration = blockPreviewActivity === "reasoning"; + const startsNewGeneration = !continuesCurrentActivity && !reusesReasoningGeneration; + if (startsNewGeneration) { + currentAssistantPreviewUsesPrefix = false; + } + const boundarySettled = startsNewGeneration + ? previewBoundaryController.noteBoundary() + : undefined; + // Message-start is only a candidate boundary: consecutive tool-only turns stay in the + // same activity post, while the first visible text or reasoning starts a new block. + if (!continuesCurrentActivity) { + progressDraft.reset(); + } + blockPreviewActivity = activity; + blockPreviewAssistantMessagePending = false; + if (activity === "tool") { + lastPartialText = ""; + } + return boundarySettled; + }; + const previewState: MattermostDraftPreviewState = { + finalizedViaPreviewPost: false, + }; + + const resolveFinalDeliveryText = (text?: string) => { + if (typeof text !== "string") { + return undefined; + } + const resolution = draftStream.resolveFinalText(text); + return resolution.kind === "already-delivered" ? "" : resolution.text; + }; + + const resolvePreviewFinalText = (text?: string) => { + const deliveryText = resolveFinalDeliveryText(text); + if (typeof deliveryText !== "string") { + return undefined; + } + const formatted = core.channel.text.convertMarkdownTables(deliveryText, tableMode); + const chunks = core.channel.text.chunkMarkdownTextWithMode(formatted, textLimit, chunkMode); + if (!chunks.length && formatted) { + chunks.push(formatted); + } + if (chunks.length != 1) { + return undefined; + } + const trimmed = chunks[0]?.trim(); + if (!trimmed) { + return undefined; + } + if ( + lastPartialText && + lastPartialText.startsWith(trimmed) && + trimmed.length < lastPartialText.length + ) { + return undefined; + } + return trimmed; + }; + + const updateDraftFromPartial = (text?: string) => { + const cleaned = text?.trim(); + if (!cleaned) { + return undefined; + } + if (cleaned === lastPartialText) { + return undefined; + } + if ( + lastPartialText && + lastPartialText.startsWith(cleaned) && + cleaned.length < lastPartialText.length + ) { + return undefined; + } + const boundarySettled = enterBlockPreviewActivity("text"); + lastPartialText = cleaned; + if (firstAssistantPreviewPrefixPending) { + firstAssistantPreviewPrefix = resolveResponsePrefix?.(); + firstAssistantPreviewPrefixPending = false; + currentAssistantPreviewUsesPrefix = Boolean(firstAssistantPreviewPrefix); + } + const previewText = + currentAssistantPreviewUsesPrefix && firstAssistantPreviewPrefix + ? cleaned.startsWith(firstAssistantPreviewPrefix) + ? cleaned + : `${firstAssistantPreviewPrefix} ${cleaned}` + : cleaned; + draftStream.updateAssistantText(previewText); + previewBoundaryController.noteUpdate(); + return boundarySettled; + }; + + const deliveryBarrier = createMattermostReplyDeliveryBarrier({ + isDirect: kind === "direct", + dmRetryOptions: account.config.dmChannelRetry, + }); + const dispatcherOptions: NonNullable = { + ...replyPipeline, + resolveFollowupAdmissionBarrierTimeoutPolicy: deliveryBarrier.resolveTimeoutPolicy, + onDeliverySettled: deliveryBarrier.markDeliverySettled, + humanDelay: resolveHumanDelayConfig(cfg, route.agentId), + typingCallbacks, + }; + const delivery: ChannelInboundTurnPlan["delivery"] = { + deliver: async (payloadEntry: ReplyPayload, info) => { + if (info.kind === "final") { + await enterBlockPreviewActivity("text"); + // Final text resolution uses only generations confirmed visible. Join prior + // boundary work before the synchronous final-edit decision. + await draftStream.settleBoundaries(); + progressDraft.markFinalReplyStarted(); + } + // A visible same-thread final arrives either via a normal send or by editing + // the draft preview in place; record participation on whichever path fires. + const markThreadParticipation = () => { + if (kind !== "direct" && effectiveReplyToId) { + recordMattermostThreadParticipation(account.accountId, channelId, effectiveReplyToId, { + agentId: route.agentId, + }); + } + }; + await deliverMattermostReplyWithDraftPreview({ + payload: payloadEntry, + info, + kind, + client, + draftStream, + effectiveReplyToId, + resolvePreviewFinalText, + previewState, + logVerboseMessage, + recordThreadParticipation: markThreadParticipation, + deliverPayload: async (payloadToDeliver) => { + const finalTextResolution = + info.kind === "final" && + !payloadToDeliver.isError && + typeof payloadToDeliver.text === "string" + ? draftStream.resolveFinalText(payloadToDeliver.text) + : undefined; + const resolvedPayload = finalTextResolution + ? { + ...payloadToDeliver, + text: + finalTextResolution.kind === "already-delivered" + ? "" + : finalTextResolution.text, + } + : payloadToDeliver; + const outcome = await deliverMattermostReplyPayload({ + core, + cfg, + payload: resolvedPayload, + to, + accountId: account.accountId, + agentId: route.agentId, + replyToId: resolveMattermostReplyRootId({ + kind, + threadRootId: effectiveReplyToId, + replyToId: payloadToDeliver.replyToId, + }), + textLimit, + tableMode, + sendMessage: sendMessageMattermost, + onDmChannelResolution: deliveryBarrier.trackDmChannelResolution, + }); + // Record only on a visible send so threads we merely observed + // (reasoning-only/empty/suppressed) do not auto-engage later. + if (outcome === "text" || outcome === "media") { + markThreadParticipation(); + } else if (outcome === "empty" && finalTextResolution?.kind === "already-delivered") { + // The terminal payload confirms the already-published assistant block as + // the visible final reply even though this delivery has no remaining text. + markThreadParticipation(); + } + const deliveryLog = formatMattermostFinalDeliveryOutcomeLog({ + outcome, + payload: resolvedPayload, + to, + accountId: account.accountId, + agentId: route.agentId, + }); + if (deliveryLog) { + runtime.log?.(deliveryLog); + } + }, + }); + if (info.kind === "final") { + progressDraft.markFinalReplyDelivered(); + } + }, + onError: (err, info) => { + runtime.error?.(`mattermost ${info.kind} reply failed: ${String(err)}`); + }, + }; + + const inboundLastRouteSessionKey = resolveInboundLastRouteSessionKey({ + route, + sessionKey: route.sessionKey, + }); + + try { + await core.channel.inbound.run({ + channel: "mattermost", + accountId: route.accountId, + raw: post, + adapter: { + ingest: () => ({ + id: post.id ?? `${to}:${Date.now()}`, + timestamp: post.create_at ?? undefined, + rawText, + textForAgent: ctxPayload.BodyForAgent, + textForCommands: ctxPayload.CommandBody, + raw: post, + }), + resolveTurn: () => ({ + cfg, + channel: "mattermost", + accountId: route.accountId, + route: { + agentId: route.agentId, + dmScope: route.dmScope, + sessionKey: route.sessionKey, + }, + ctxPayload, + record: { + updateLastRoute: + kind === "direct" + ? { + sessionKey: inboundLastRouteSessionKey, + channel: "mattermost", + to, + accountId: route.accountId, + mainDmOwnerPin: + inboundLastRouteSessionKey === route.mainSessionKey && pinnedMainDmOwner + ? { + ownerRecipient: pinnedMainDmOwner, + senderRecipient: normalizeMattermostAllowEntry(senderId), + onSkip: ({ + ownerRecipient, + senderRecipient, + }: { + ownerRecipient: string; + senderRecipient: string; + }) => { + logVerboseMessage( + `mattermost: skip main-session last route for ${senderRecipient} (pinned owner ${ownerRecipient})`, + ); + }, + } + : undefined, + } + : undefined, + onRecordError: (err) => { + logVerboseMessage( + `mattermost: failed updating session meta id=${post.id ?? "unknown"}: ${String(err)}`, + ); + }, + }, + history: { + isGroup: Boolean(historyKey), + historyKey: historyKey ?? undefined, + historyMap: channelHistories, + limit: historyLimit, + }, + dispatcherOptions, + delivery, + replyOptions: { + ...(turnAdoptionLifecycle + ? bindIngressLifecycleToReplyOptions(turnAdoptionLifecycle) + : {}), + allowProgressCallbacksWhenSourceDeliverySuppressed: draftToolProgressEnabled + ? true + : undefined, + preserveProgressCallbackStartOrder: draftPreviewEnabled ? true : undefined, + onObservedReplyDelivery: draftToolProgressEnabled + ? () => draftStream.clear() + : undefined, + disableBlockStreaming: draftPreviewEnabled + ? true + : typeof account.blockStreaming === "boolean" + ? !account.blockStreaming + : undefined, + ...(suppressDefaultToolProgressMessages + ? { suppressDefaultToolProgressMessages: true } + : {}), + onModelSelected, + onPartialReply: (payloadResult) => { + if (account.streamingMode !== "progress") { + return updateDraftFromPartial(payloadResult.text); + } + return undefined; + }, + onAssistantMessageStart: () => { + lastPartialText = ""; + progressDraft.resetReasoningProgress(); + if (account.streamingMode === "block") { + blockPreviewAssistantMessagePending = true; + return; + } + if (account.streamingMode !== "progress") { + progressDraft.reset(); + } + }, + onReasoningEnd: () => { + // Hidden reasoning has no visible boundary. Only transitions that + // actually render text, reasoning, or tools rotate preview posts. + lastPartialText = ""; + progressDraft.resetReasoningProgress(); + if (account.streamingMode !== "block" && account.streamingMode !== "progress") { + progressDraft.reset(); + } + }, + onReasoningStream: async (payloadResult) => { + if (account.streamingMode === "progress") { + await progressDraft.pushReasoningProgress(payloadResult.text || "Thinking…", { + snapshot: payloadResult.isReasoningSnapshot === true, + }); + return; + } + if (!lastPartialText) { + const boundarySettled = enterBlockPreviewActivity("reasoning"); + draftStream.update("Thinking…"); + previewBoundaryController.noteUpdate(); + await boundarySettled; + } + }, + onToolStart: async (payloadValue) => { + if (!draftToolProgressEnabled) { + return; + } + const boundarySettled = enterBlockPreviewActivity("tool"); + // Boundary detach and progress staging both happen synchronously before + // their first await; agent callbacks may be dispatched fire-and-forget. + const progressSettled = progressDraft.pushToolProgress( + buildChannelProgressDraftLineForEntry( + account.config, + { + event: "tool", + itemId: payloadValue.itemId, + toolCallId: payloadValue.toolCallId, + name: payloadValue.name, + phase: payloadValue.phase, + args: payloadValue.args, + }, + payloadValue.detailMode ? { detailMode: payloadValue.detailMode } : undefined, + ), + { startImmediately: true }, + ); + previewBoundaryController.noteUpdate(); + await Promise.all([boundarySettled, progressSettled]); + }, + onItemEvent: async (payloadLocal) => { + if (!draftToolProgressEnabled) { + return; + } + const boundarySettled = enterBlockPreviewActivity("tool"); + const progressSettled = progressDraft.pushToolProgress( + buildChannelProgressDraftLineForEntry(account.config, { + event: "item", + itemId: payloadLocal.itemId, + itemKind: payloadLocal.kind, + title: payloadLocal.title, + name: payloadLocal.name, + phase: payloadLocal.phase, + status: payloadLocal.status, + summary: payloadLocal.summary, + progressText: payloadLocal.progressText, + meta: payloadLocal.meta, + }), + { startImmediately: true }, + ); + previewBoundaryController.noteUpdate(); + await Promise.all([boundarySettled, progressSettled]); + }, + }, + }), + }, + }); + } finally { + try { + await draftStream.stop(); + } catch (err) { + logVerboseMessage(`mattermost draft preview cleanup failed: ${String(err)}`); + } } }; @@ -1990,6 +1972,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} const debouncer = core.channel.debounce.createInboundDebouncer<{ post: MattermostPost; payload: MattermostEventPayload; + turnAdoptionLifecycle: MattermostIngressLifecycle; }>({ debounceMs: inboundDebounceMs, buildKey: (entry) => { @@ -2019,27 +2002,54 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} if (!last) { return; } - if (entries.length === 1) { - await handlePost(last.post, last.payload); - return; + const { lifecycle, settle } = buildMattermostFlushIngressLifecycle(entries); + try { + if (entries.length === 1) { + await handlePost(last.post, last.payload, lifecycle); + await settle(); + return; + } + const combinedText = entries + .map((entry) => normalizeOptionalString(entry.post.message) ?? "") + .filter(Boolean) + .join("\n"); + const mergedPost: MattermostPost = { + ...last.post, + message: combinedText, + file_ids: [], + }; + await handlePost( + mergedPost, + last.payload, + lifecycle, + entries.map((entry) => entry.post.id), + ); + await settle(); + } catch (error) { + await lifecycle?.onAbandoned(); + throw error; } - const combinedText = entries - .map((entry) => normalizeOptionalString(entry.post.message) ?? "") - .filter(Boolean) - .join("\n"); - const mergedPost: MattermostPost = { - ...last.post, - message: combinedText, - file_ids: [], - }; - const ids = entries.map((entry) => entry.post.id).filter(Boolean); - await handlePost(mergedPost, last.payload, ids.length > 0 ? ids : undefined); }, onError: (err) => { runtime.error?.(`mattermost debounce flush failed: ${String(err)}`); }, }); + const ingress = createMattermostIngressMonitor({ + accountId: account.accountId, + runtime, + abortSignal: opts.abortSignal, + dispatch: async (post, payload, turnAdoptionLifecycle) => { + // Deferred claims settle through lifecycle callbacks, so terminal flush + // errors (401/403 included) abandon rather than hit the non-retryable + // classifier; the drain's attempt/age retry policy still dead-letters + // them — auth failures just spend the bounded retry budget first. + // Accepted tradeoff over threading a fail() channel through deferral. + await debouncer.enqueue({ post, payload, turnAdoptionLifecycle }); + return { kind: "deferred" }; + }, + }); + const wsUrl = buildMattermostWsUrl(baseUrl); let seq = 1; const connectOnce = createMattermostConnectOnce({ @@ -2054,9 +2064,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} const me = await fetchMattermostMe(client); return me.update_at ?? 0; }, - onPosted: async (post, payload) => { - await debouncer.enqueue({ post, payload }); - }, + onPosted: ingress.receive, onReaction: async (payload) => { await handleReactionEvent(payload); }, @@ -2106,6 +2114,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} }, }); } finally { + await ingress.stop(); unregisterInteractions?.(); }