From 9e9ade362e715cd461a59941e98e1554a4fe2e34 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 13:25:02 -0400 Subject: [PATCH 01/16] feat(slack): expose conversation and thread read cursors --- connectors/slack/src/slack-api.test.ts | 96 +++++++++++++++++++++++++- connectors/slack/src/slack-api.ts | 62 ++++++++++++++++- 2 files changed, 155 insertions(+), 3 deletions(-) diff --git a/connectors/slack/src/slack-api.test.ts b/connectors/slack/src/slack-api.test.ts index 02714bcb..410199be 100644 --- a/connectors/slack/src/slack-api.test.ts +++ b/connectors/slack/src/slack-api.test.ts @@ -1,6 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { + SlackApi, transformSlackThread, + syncSlackChannel, type SlackMessage, type SlackUserInfoMap, } from "./slack-api"; @@ -67,3 +69,95 @@ describe("transformSlackThread", () => { ).toBeUndefined(); }); }); + +describe("SlackApi.getConversationInfo", () => { + it("returns the caller's last_read cursor", async () => { + const api = new SlackApi("xoxp-test"); + const call = vi + .spyOn(api, "call") + .mockResolvedValue({ channel: { id: "C1", last_read: "1700000000.000001" } }); + + await expect(api.getConversationInfo("C1")).resolves.toEqual({ + lastRead: "1700000000.000001", + }); + expect(call).toHaveBeenCalledWith("conversations.info", { channel: "C1" }); + }); + + it("returns null when Slack omits last_read rather than inventing one", async () => { + const api = new SlackApi("xoxp-test"); + vi.spyOn(api, "call").mockResolvedValue({ channel: { id: "C1" } }); + + await expect(api.getConversationInfo("C1")).resolves.toEqual({ lastRead: null }); + }); +}); + +describe("SlackApi.markConversationRead", () => { + it("marks the conversation read at the given ts", async () => { + const api = new SlackApi("xoxp-test"); + const call = vi.spyOn(api, "call").mockResolvedValue({ ok: true }); + + await api.markConversationRead("D1", "1700000000.000001"); + + expect(call).toHaveBeenCalledWith("conversations.mark", { + channel: "D1", + ts: "1700000000.000001", + }); + }); +}); + +describe("syncSlackChannel thread parent", () => { + it("keeps the conversations.replies parent so its thread cursor survives", async () => { + const historyParent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + reply_count: 1, + }; + const repliesParent = { ...historyParent, unread_count: 0, subscribed: true }; + const reply = { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "reply", + }; + + const api = { + getConversationHistory: vi + .fn() + .mockResolvedValue({ messages: [historyParent], hasMore: false }), + getThread: vi.fn().mockResolvedValue([repliesParent, reply]), + getThreadReplies: vi.fn(), + }; + + const { threads } = await syncSlackChannel(api as never, { channelId: "C1" }); + + expect(threads).toHaveLength(1); + expect(threads[0]![0]!.unread_count).toBe(0); + expect(threads[0]![1]!.ts).toBe("1700000002.000000"); + expect(api.getThreadReplies).not.toHaveBeenCalled(); + }); + + it("falls back to the history parent when conversations.replies returns nothing", async () => { + const historyParent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + reply_count: 1, + }; + const api = { + getConversationHistory: vi + .fn() + .mockResolvedValue({ messages: [historyParent], hasMore: false }), + getThread: vi.fn().mockResolvedValue([]), + }; + + const { threads } = await syncSlackChannel(api as never, { channelId: "C1" }); + + expect(threads).toEqual([[historyParent]]); + }); +}); diff --git a/connectors/slack/src/slack-api.ts b/connectors/slack/src/slack-api.ts index 9758a225..1a1085da 100644 --- a/connectors/slack/src/slack-api.ts +++ b/connectors/slack/src/slack-api.ts @@ -55,6 +55,20 @@ export type SlackMessage = { }>; reply_count?: number; reply_users_count?: number; + latest_reply?: string; + /** + * Per-THREAD read state for the calling user, present on the parent message + * of a `conversations.replies` response. Slack tracks a thread's read state + * separately from its channel's: reading the channel does not advance these, + * and opening the thread does not advance the channel's `last_read`. + * + * Absent on messages from `conversations.history`, which is why the thread + * fetch in `syncSlackChannel` keeps the replies parent rather than the + * history one. + */ + subscribed?: boolean; + last_read?: string; + unread_count?: number; }; export type SlackUser = { @@ -372,6 +386,37 @@ export class SlackApi { return messages.slice(1); } + /** + * The calling user's read cursor for one conversation. + * + * Tier 3, and NOT subject to the 1 rpm non-Marketplace limit that applies to + * `conversations.history`/`conversations.replies` — so this is safe to call + * per conversation in a sweep. Returns `null` when Slack omits `last_read` + * (the caller must then abstain rather than assume a state). + */ + public async getConversationInfo( + channelId: string + ): Promise<{ lastRead: string | null }> { + const data = await this.call("conversations.info", { channel: channelId }); + const lastRead = data.channel?.last_read; + return { lastRead: typeof lastRead === "string" ? lastRead : null }; + } + + /** + * Move the calling user's read cursor for one conversation to `ts`. + * + * CONVERSATION-scoped: there is no per-thread equivalent in the public Web + * API, so this must only ever be called for a direct conversation, where the + * Plot link IS the whole conversation. Calling it for a channel would move + * that channel's cursor for every other message in it. + */ + public async markConversationRead( + channelId: string, + ts: string + ): Promise { + await this.call("conversations.mark", { channel: channelId, ts }); + } + public async postMessage( channelId: string, text: string, @@ -858,8 +903,21 @@ export async function syncSlackChannel( // that cursor indefinitely. Degrade to the parent-only path so the // rest of the channel still advances. try { - const replies = await api.getThreadReplies(state.channelId, threadTs); - threads.push([parentMessage, ...replies]); + // `getThread` (conversations.replies), not `getThreadReplies`: the + // parent it returns carries this thread's own read cursor + // (`unread_count`/`last_read`/`latest_reply`), which the + // `conversations.history` parent does not have. Same single API call + // either way — `getThreadReplies` just threw the parent away. + // + // Spread the history parent underneath so any field only that response + // carried survives, then let the replies parent's fields win. + const full = await api.getThread(state.channelId, threadTs); + const [repliesParent, ...replies] = full; + threads.push( + repliesParent + ? [{ ...parentMessage, ...repliesParent }, ...replies] + : [parentMessage] + ); } catch (error) { console.warn( `conversations.replies failed for ${state.channelId}/${threadTs}; falling back to parent-only`, From 6da49cbb031dc741e784da6f97402d18e9487b1d Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 13:32:47 -0400 Subject: [PATCH 02/16] feat(slack): add read-state projection helpers --- connectors/slack/src/slack-read-state.test.ts | 130 ++++++++++++++++++ connectors/slack/src/slack-read-state.ts | 106 ++++++++++++++ 2 files changed, 236 insertions(+) create mode 100644 connectors/slack/src/slack-read-state.test.ts create mode 100644 connectors/slack/src/slack-read-state.ts diff --git a/connectors/slack/src/slack-read-state.test.ts b/connectors/slack/src/slack-read-state.test.ts new file mode 100644 index 00000000..75ef45a2 --- /dev/null +++ b/connectors/slack/src/slack-read-state.test.ts @@ -0,0 +1,130 @@ +import { describe, expect, it } from "vitest"; +import { + channelReadVerdict, + compareSlackTs, + deriveReadAnchor, + threadReadVerdict, +} from "./slack-read-state"; +import type { SlackMessage } from "./slack-api"; + +function msg(over: Partial & { ts: string }): SlackMessage { + return { type: "message", text: "", ...over }; +} + +describe("compareSlackTs", () => { + it("orders by seconds first", () => { + expect(compareSlackTs("1700000001.000000", "1700000000.999999")).toBe(1); + expect(compareSlackTs("1700000000.999999", "1700000001.000000")).toBe(-1); + }); + + it("orders by microseconds within the same second", () => { + expect(compareSlackTs("1700000000.000002", "1700000000.000001")).toBe(1); + expect(compareSlackTs("1700000000.000001", "1700000000.000002")).toBe(-1); + expect(compareSlackTs("1700000000.000001", "1700000000.000001")).toBe(0); + }); + + it("does not lose precision on 16-significant-digit timestamps", () => { + // parseFloat("1502126650.228446") rounds; a naive numeric compare calls + // these equal. The micro halves differ by one, so the result must be -1. + expect(compareSlackTs("1502126650.228446", "1502126650.228447")).toBe(-1); + }); + + it("treats Slack's never-read sentinel as older than everything", () => { + expect(compareSlackTs("0000000000.000000", "1700000000.000001")).toBe(-1); + }); +}); + +describe("channelReadVerdict", () => { + it("is read when the cursor is at or past the newest message", () => { + expect(channelReadVerdict("1700000000.000001", "1700000000.000001")).toBe("read"); + expect(channelReadVerdict("1700000001.000000", "1700000000.000001")).toBe("read"); + }); + + it("is unread when the cursor is behind the newest message", () => { + expect(channelReadVerdict("1700000000.000000", "1700000000.000001")).toBe("unread"); + }); + + it("abstains when Slack did not return a cursor", () => { + expect(channelReadVerdict(null, "1700000000.000001")).toBe("unknown"); + expect(channelReadVerdict(undefined, "1700000000.000001")).toBe("unknown"); + expect(channelReadVerdict("", "1700000000.000001")).toBe("unknown"); + }); +}); + +describe("threadReadVerdict", () => { + it("prefers unread_count when present", () => { + expect(threadReadVerdict(msg({ ts: "1.0", unread_count: 0 }))).toBe("read"); + expect(threadReadVerdict(msg({ ts: "1.0", unread_count: 3 }))).toBe("unread"); + }); + + it("falls back to last_read vs latest_reply", () => { + expect( + threadReadVerdict( + msg({ ts: "1.0", last_read: "1700000002.000000", latest_reply: "1700000002.000000" }) + ) + ).toBe("read"); + expect( + threadReadVerdict( + msg({ ts: "1.0", last_read: "1700000001.000000", latest_reply: "1700000002.000000" }) + ) + ).toBe("unread"); + }); + + it("abstains when the parent carries no thread cursor at all", () => { + expect(threadReadVerdict(msg({ ts: "1.0" }))).toBe("unknown"); + expect(threadReadVerdict(undefined)).toBe("unknown"); + }); + + it("abstains when only one half of the fallback pair is present", () => { + expect(threadReadVerdict(msg({ ts: "1.0", last_read: "1700000002.000000" }))).toBe( + "unknown" + ); + expect(threadReadVerdict(msg({ ts: "1.0", latest_reply: "1700000002.000000" }))).toBe( + "unknown" + ); + }); +}); + +describe("deriveReadAnchor", () => { + it("anchors on the newest message ts", () => { + const anchor = deriveReadAnchor( + [msg({ ts: "1700000000.000001" }), msg({ ts: "1700000002.000000" })], + { direct: false, at: 1000 } + ); + expect(anchor).toEqual({ newest: "1700000002.000000", threaded: false, at: 1000 }); + }); + + it("marks a channel link threaded when it holds a real thread reply", () => { + const anchor = deriveReadAnchor( + [ + msg({ ts: "1700000000.000001", thread_ts: "1700000000.000001" }), + msg({ ts: "1700000002.000000", thread_ts: "1700000000.000001" }), + ], + { direct: false, at: 1000 } + ); + expect(anchor?.threaded).toBe(true); + }); + + it("does not treat a lone parent as threaded", () => { + const anchor = deriveReadAnchor( + [msg({ ts: "1700000000.000001", thread_ts: "1700000000.000001" })], + { direct: false, at: 1000 } + ); + expect(anchor?.threaded).toBe(false); + }); + + it("never marks a direct conversation threaded — a DM uses the conversation cursor", () => { + const anchor = deriveReadAnchor( + [ + msg({ ts: "1700000000.000001", thread_ts: "1700000000.000001" }), + msg({ ts: "1700000002.000000", thread_ts: "1700000000.000001" }), + ], + { direct: true, at: 1000 } + ); + expect(anchor?.threaded).toBe(false); + }); + + it("returns null for an empty message set", () => { + expect(deriveReadAnchor([], { direct: false, at: 1000 })).toBeNull(); + }); +}); diff --git a/connectors/slack/src/slack-read-state.ts b/connectors/slack/src/slack-read-state.ts new file mode 100644 index 00000000..4227f2d7 --- /dev/null +++ b/connectors/slack/src/slack-read-state.ts @@ -0,0 +1,106 @@ +import type { SlackMessage } from "./slack-api"; + +/** + * Whether Slack considers a Plot link read. + * + * `"unknown"` is distinct from `"unread"` on purpose: it means Slack did not + * give us a usable cursor, so the connector must leave Plot's state alone + * rather than assert anything. Guessing in either direction is worse than + * abstaining — a wrong `"read"` hides a message the user never saw. + */ +export type SlackReadVerdict = "read" | "unread" | "unknown"; + +/** + * Compare two Slack timestamps (`"1700000000.000001"`). + * + * Done on the two halves as integers rather than via `parseFloat`: a Slack ts + * carries 16 significant digits, which is at the edge of float64 precision, + * so `parseFloat` can round two genuinely different timestamps to the same + * value. Returns <0, 0, or >0 like a comparator. + */ +export function compareSlackTs(a: string, b: string): number { + const [aSec = "0", aMicro = "0"] = a.split("."); + const [bSec = "0", bMicro = "0"] = b.split("."); + const secDiff = Number(aSec) - Number(bSec); + if (secDiff !== 0) return secDiff < 0 ? -1 : 1; + const microDiff = + Number(aMicro.padEnd(6, "0")) - Number(bMicro.padEnd(6, "0")); + return microDiff === 0 ? 0 : microDiff < 0 ? -1 : 1; +} + +/** + * Project a CHANNEL-level cursor (`conversations.info.last_read`) onto a link. + * + * Correct only for links whose every message sits in the channel timeline — + * a direct conversation, or a channel message with no thread replies. Reading + * a channel does not advance a thread's own cursor, so a threaded link must + * use {@link threadReadVerdict} instead. + */ +export function channelReadVerdict( + lastRead: string | null | undefined, + newestTs: string +): SlackReadVerdict { + if (!lastRead) return "unknown"; + return compareSlackTs(lastRead, newestTs) >= 0 ? "read" : "unread"; +} + +/** + * Project a THREAD's own cursor onto a link, from the thread parent returned + * by `conversations.replies`. + * + * `unread_count` is the direct answer and is preferred. The + * `last_read`/`latest_reply` pair is the fallback for responses that carry + * the cursor but not the count. A parent with neither means the caller is not + * subscribed to the thread (or Slack simply omitted the state), so abstain. + */ +export function threadReadVerdict( + parent: SlackMessage | undefined +): SlackReadVerdict { + if (!parent) return "unknown"; + if (typeof parent.unread_count === "number") { + return parent.unread_count === 0 ? "read" : "unread"; + } + if (parent.last_read && parent.latest_reply) { + return compareSlackTs(parent.last_read, parent.latest_reply) >= 0 + ? "read" + : "unread"; + } + return "unknown"; +} + +/** + * The reconciliation state one saved link needs. + * + * `newest` is what a cursor is compared against. `threaded` picks which cursor + * governs: a channel link holding real thread replies is settled by the thread + * cursor on the live path, everything else by the channel cursor in the daily + * sweep. `at` is the write time, for the retention drop. + */ +export type SlackReadAnchor = { + newest: string; + threaded: boolean; + at: number; +}; + +/** + * Derive the anchor for a set of messages about to be saved as one link. + * + * A direct conversation is never `threaded`: its link flattens Slack's reply + * threads into one running conversation, so the conversation cursor is the + * only cursor that describes it. + */ +export function deriveReadAnchor( + messages: SlackMessage[], + opts: { direct: boolean; at: number } +): SlackReadAnchor | null { + let newest: string | null = null; + let threaded = false; + for (const message of messages) { + if (!newest || compareSlackTs(message.ts, newest) > 0) newest = message.ts; + if (!opts.direct && message.thread_ts && message.thread_ts !== message.ts) { + threaded = true; + } + } + if (!newest) return null; + return { newest, threaded, at: opts.at }; +} From c4264f72041996903d617ae346e3b0dc548c43f5 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 13:41:49 -0400 Subject: [PATCH 03/16] feat(slack): mirror a thread's own read state into Plot --- connectors/slack/src/slack.test.ts | 73 ++++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 16 +++++++ 2 files changed, 89 insertions(+) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 4749c8ac..9228d7aa 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -3711,3 +3711,76 @@ describe("onNoteCreated reply targeting", () => { expect(postMessage).toHaveBeenCalledWith("C123", "ack", "100.0"); }); }); + +describe("buildConversationLink — thread read cursor", () => { + function buildSlack() { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + return { slack, store }; + } + + const parent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + }; + const reply = { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "reply", + }; + + async function build(messages: unknown[]) { + const { slack } = buildSlack(); + return (slack as unknown as { + buildConversationLink: (o: unknown) => Promise<{ unread?: boolean } | null>; + }).buildConversationLink({ + channelId: "C1", + messages, + initialSync: false, + }); + } + + it("marks the link read when the thread's own cursor says read", async () => { + const link = await build([{ ...parent, unread_count: 0 }, reply]); + expect(link?.unread).toBe(false); + }); + + it("leaves unread alone when the thread cursor says unread", async () => { + const link = await build([{ ...parent, unread_count: 2 }, reply]); + expect(link).not.toHaveProperty("unread"); + }); + + it("abstains when the parent carries no thread cursor", async () => { + const link = await build([parent, reply]); + expect(link).not.toHaveProperty("unread"); + }); + + it("abstains on a root-only link — the channel cursor is the sweep's job", async () => { + const link = await build([{ ...parent, unread_count: 0 }]); + expect(link).not.toHaveProperty("unread"); + }); + + it("never sets unread true, even when the thread is unread in Slack", async () => { + const link = await build([{ ...parent, unread_count: 5 }, reply]); + expect(link?.unread).not.toBe(true); + }); +}); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index c721b529..b17ff8ff 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -46,6 +46,10 @@ import { assembleSlackDmLink, slackConversationIdentity } from "./slack-dm"; import { unicodeToSlackName } from "./slack-emoji"; import { slackFacets } from "./slack-facets"; import { mentionsUser, type MentionContext } from "./slack-mentions"; +import { + deriveReadAnchor, + threadReadVerdict, +} from "./slack-read-state"; /** * Slack integration source. @@ -829,6 +833,18 @@ export class Slack extends Connector { syncableId: channelId, }; if (messages[0]) link.facets = slackFacets(messages[0], channelId); + // Apply Slack's per-thread read cursor. `deriveReadAnchor` tells us + // whether this link is governed by the thread cursor at all: a root-only + // channel message lives purely in the channel timeline, so only the daily + // sweep's `conversations.info` can speak for it. + // + // Read from THIS response, never from a cache: the parent's + // `unread_count` is computed after the reply we are saving landed, so a + // genuine new-message unread can never be suppressed by a stale cursor. + const anchor = deriveReadAnchor(messages, { direct: false, at: Date.now() }); + if (anchor?.threaded && threadReadVerdict(messages[0]) === "read") { + link.unread = false; + } return link; } From 32ee4a43e3a81e389dbfaf208782d46ecaa9f803 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 13:53:45 -0400 Subject: [PATCH 04/16] feat(slack): track which links still need read reconciliation --- connectors/slack/src/slack.test.ts | 122 +++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 60 ++++++++++++-- 2 files changed, 176 insertions(+), 6 deletions(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 9228d7aa..93d81a33 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -3784,3 +3784,125 @@ describe("buildConversationLink — thread read cursor", () => { expect(link?.unread).not.toBe(true); }); }); + +describe("read anchors", () => { + const parent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + }; + const reply = { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "reply", + }; + + async function build(messages: unknown[]) { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + const link = await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise<{ unread?: boolean } | null>; + }).buildConversationLink({ channelId: "C1", messages, initialSync: false }); + return { store, link }; + } + + const KEY = "read_anchor:C1:1700000000.000001"; + + it("writes an anchor for a link it could not mark read", async () => { + const { store } = await build([parent]); + const anchor = store.map.get(KEY) as { newest: string; threaded: boolean }; + expect(anchor.newest).toBe("1700000000.000001"); + expect(anchor.threaded).toBe(false); + }); + + it("marks a threaded link's anchor threaded so the sweep skips it", async () => { + const { store } = await build([parent, reply]); + expect((store.map.get(KEY) as { threaded: boolean }).threaded).toBe(true); + }); + + it("deletes the anchor once the link is marked read", async () => { + const { store, link } = await build([{ ...parent, unread_count: 0 }, reply]); + expect(link?.unread).toBe(false); + expect(store.map.has(KEY)).toBe(false); + }); + + it("keys the anchor on the thread root, so a new reply replaces it", async () => { + const { store } = await build([parent, reply]); + expect([...store.map.keys()]).toEqual([KEY]); + expect((store.map.get(KEY) as { newest: string }).newest).toBe( + "1700000002.000000" + ); + }); + + it("does not write an anchor when there is nothing to save", async () => { + const { store } = await build([]); + expect([...store.map.keys()]).toEqual([]); + }); + + it("anchors a direct conversation on the conversation id, never threaded", async () => { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + vi.spyOn( + slack as unknown as { + dmCounterpartyUserId: (c: string) => Promise; + }, + "dmCounterpartyUserId" + ).mockResolvedValue("U2"); + + await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise; + }).buildConversationLink({ + channelId: "D1", + messages: [ + { type: "message", ts: "1700000000.000001", user: "U2", text: "hi" }, + { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "threaded", + }, + ], + initialSync: false, + }); + + const anchor = store.map.get("read_anchor:D1:D1") as { + newest: string; + threaded: boolean; + }; + expect(anchor.newest).toBe("1700000002.000000"); + expect(anchor.threaded).toBe(false); + }); +}); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index b17ff8ff..d03872c3 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -807,6 +807,22 @@ export class Slack extends Connector { // and rewinding what this one just wrote. await this.set(this.dmHeadKey(channelId), true); } + // Anchor the conversation for the daily sweep. A direct conversation is + // never `threaded`: this one permanent link flattens Slack's reply + // threads into a running conversation, so the CONVERSATION cursor is the + // only cursor that describes it. Keyed on the conversation id in both + // positions — there is no thread root to key on. + // + // Only when these messages ARE the conversation head. A reaction or a + // save can name a message of any age, and an anchor rewound to an older + // `newest` would make the sweep declare the conversation read on a + // cursor that has not actually reached its latest message. + if (advanceConversationHead) { + const dmAnchor = deriveReadAnchor(kept, { direct: true, at: Date.now() }); + if (dmAnchor) { + await this.set(this.readAnchorKey(channelId, channelId), dmAnchor); + } + } return link; } @@ -833,17 +849,37 @@ export class Slack extends Connector { syncableId: channelId, }; if (messages[0]) link.facets = slackFacets(messages[0], channelId); - // Apply Slack's per-thread read cursor. `deriveReadAnchor` tells us - // whether this link is governed by the thread cursor at all: a root-only - // channel message lives purely in the channel timeline, so only the daily - // sweep's `conversations.info` can speak for it. + // Apply Slack's per-thread read cursor, and record what still needs + // settling. `deriveReadAnchor` tells us which cursor governs this link: a + // root-only channel message lives purely in the channel timeline, so only + // the daily sweep's `conversations.info` can speak for it. // // Read from THIS response, never from a cache: the parent's // `unread_count` is computed after the reply we are saving landed, so a // genuine new-message unread can never be suppressed by a stale cursor. + // + // The anchor's EXISTENCE is the transition gate — it means "not yet marked + // read for this `newest` ts". Deleting it on a successful mark-read is + // what makes a later manual "mark unread" in Plot stick: the sweep has + // nothing left to act on until new content recreates the anchor. + // + // `initialSync` already asserts read (it sets `unread: false`), so those + // links need no anchor at all — the sweep would only re-assert what is + // already true. + const threadTs = + (link.meta?.threadTs as string | undefined) ?? messages[0]?.ts; const anchor = deriveReadAnchor(messages, { direct: false, at: Date.now() }); - if (anchor?.threaded && threadReadVerdict(messages[0]) === "read") { - link.unread = false; + if (anchor && threadTs) { + const anchorKey = this.readAnchorKey(channelId, threadTs); + const read = + link.unread === false || + (anchor.threaded && threadReadVerdict(messages[0]) === "read"); + if (read) { + link.unread = false; + await this.clear(anchorKey); + } else { + await this.set(anchorKey, anchor); + } } return link; } @@ -1737,6 +1773,18 @@ export class Slack extends Connector { return `sync_thread:${channelId}:${threadTs}`; } + /** + * Key for one link's read anchor. Keyed on the thread root (the same id + * `subscriptionKey` uses) so a new reply REPLACES the anchor rather than + * accumulating one per message. + * + * A direct conversation is one permanent link, so it anchors on the + * conversation id in both positions. + */ + private readAnchorKey(channelId: string, threadTs: string): string { + return `read_anchor:${channelId}:${threadTs}`; + } + private async subscribeThread( channelId: string, threadTs: string From 7fa7329758cebd9b0f9dbe361947c8c083dcb434 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 14:07:55 -0400 Subject: [PATCH 05/16] fix(slack): don't leave a read anchor for an already-read DM sync --- connectors/slack/src/slack.test.ts | 38 ++++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 8 ++++++- 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 93d81a33..ca44bc64 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -3905,4 +3905,42 @@ describe("read anchors", () => { expect(anchor.newest).toBe("1700000002.000000"); expect(anchor.threaded).toBe(false); }); + + it("leaves no anchor for a DM's initial sync — the link is already read", async () => { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + vi.spyOn( + slack as unknown as { + dmCounterpartyUserId: (c: string) => Promise; + }, + "dmCounterpartyUserId" + ).mockResolvedValue("U2"); + + const link = await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise<{ unread?: boolean } | null>; + }).buildConversationLink({ + channelId: "D1", + messages: [ + { type: "message", ts: "1700000000.000001", user: "U2", text: "hi" }, + ], + initialSync: true, + }); + + expect(link?.unread).toBe(false); + expect(store.map.has("read_anchor:D1:D1")).toBe(false); + }); }); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index d03872c3..f48ea67a 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -817,7 +817,13 @@ export class Slack extends Connector { // save can name a message of any age, and an anchor rewound to an older // `newest` would make the sweep declare the conversation read on a // cursor that has not actually reached its latest message. - if (advanceConversationHead) { + // + // Only when the link is NOT already read. `assembleSlackDmLink` sets + // `unread: false` for `initialSync`, which already asserts read — an + // anchor for it could only ever re-assert what is already true, and + // would outlive a later manual "mark unread" in Plot, letting the sweep + // silently revert it once Slack's cursor caught up. + if (advanceConversationHead && link.unread !== false) { const dmAnchor = deriveReadAnchor(kept, { direct: true, at: Date.now() }); if (dmAnchor) { await this.set(this.readAnchorKey(channelId, channelId), dmAnchor); From 52f17674d220c6e2324a5afe1c10f0d4eff882a6 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 14:20:22 -0400 Subject: [PATCH 06/16] feat(slack): reconcile Slack read state into Plot daily --- connectors/slack/src/slack.test.ts | 199 +++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 160 +++++++++++++++++++++++ 2 files changed, 359 insertions(+) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index ca44bc64..e0ae7b43 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -30,6 +30,12 @@ function makeStore(initial: Record = {}) { list: vi.fn(async (prefix: string) => [...map.keys()].filter((k) => k.startsWith(prefix)) ), + listEntries: vi.fn(async (prefix: string) => + [...map.entries()].filter(([k]) => k.startsWith(prefix)) + ), + clearMany: vi.fn(async (keys: string[]) => { + for (const key of keys) map.delete(key); + }), }; } @@ -3944,3 +3950,196 @@ describe("read anchors", () => { expect(store.map.has("read_anchor:D1:D1")).toBe(false); }); }); + +describe("reconcileReadState", () => { + const NOW = 1_700_000_000_000; + + function setup(anchors: Record, lastRead: string | null) { + const store = makeStore(anchors); + const saveLink = vi.fn().mockResolvedValue("thread-1"); + const tools = { + store, + integrations: { get: vi.fn(), saveLink }, + network: { createWebhook: vi.fn() }, + files: {}, + }; + const slack = new Slack( + "twist-instance-1" as never, + { getTools: () => tools } as never + ); + const api = { getConversationInfo: vi.fn().mockResolvedValue({ lastRead }) }; + vi.spyOn( + slack as unknown as { getApi: (c: string) => Promise }, + "getApi" + ).mockResolvedValue(api); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + // The `finally` re-arms the daily chain; neither the callbacks nor the + // tasks tool is in the test tool shed, so both are stubbed. + vi.spyOn( + slack as unknown as { callback: (...a: unknown[]) => Promise }, + "callback" + ).mockResolvedValue("cb-token"); + vi.spyOn( + slack as unknown as { scheduleRecurring: (...a: unknown[]) => Promise }, + "scheduleRecurring" + ).mockResolvedValue(undefined); + vi.spyOn(Date, "now").mockReturnValue(NOW); + return { slack, store, saveLink, api }; + } + + const anchor = (over: Partial<{ newest: string; threaded: boolean; at: number }> = {}) => ({ + newest: "1700000000.000001", + threaded: false, + at: NOW, + ...over, + }); + + it("marks a root-only link read once the channel cursor passes it", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).toHaveBeenCalledTimes(1); + const saved = saveLink.mock.calls[0][0]; + expect(saved.unread).toBe(false); + expect(saved.channelId).toBe("C1"); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); + }); + + it("never sends `created` on the reconcile upsert", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + // saveLink drops an `unread: false` save whose date predates the plan's + // sync-history limit, so a reconcile upsert must carry no date at all. + expect(saveLink.mock.calls[0][0]).not.toHaveProperty("created"); + expect(saveLink.mock.calls[0][0]).not.toHaveProperty("schedules"); + }); + + it("does not rewrite content on the reconcile upsert", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + const saved = saveLink.mock.calls[0][0]; + expect(saved).not.toHaveProperty("title"); + expect(saved).not.toHaveProperty("preview"); + expect(saved).not.toHaveProperty("notes"); + }); + + it("leaves a link alone while the channel cursor is still behind it", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000005.000000": anchor({ newest: "1700000005.000000" }) }, + "1700000000.000001" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000005.000000")).toBe(true); + }); + + it("skips threaded anchors — the channel cursor does not speak for them", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor({ threaded: true }) }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + }); + + it("abstains when Slack returns no cursor", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + null + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + }); + + it("issues one conversations.info per conversation, not per anchor", async () => { + const { slack, api } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C1:1700000000.000002": anchor({ newest: "1700000000.000002" }), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(api.getConversationInfo).toHaveBeenCalledTimes(2); + }); + + it("drops anchors past the retention window without marking them read", async () => { + const stale = NOW - 31 * 24 * 60 * 60 * 1000; + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor({ at: stale }) }, + "1700000000.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); + }); + + it("stops the pass and keeps every remaining anchor when Slack rate limits", async () => { + const { slack, store, api } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + api.getConversationInfo.mockRejectedValue( + new SlackRateLimitedError("conversations.info", 60_000) + ); + + await slack.reconcileReadState("C1"); + + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(true); + }); + + it("skips one unreachable conversation without abandoning the rest", async () => { + const { slack, store, api, saveLink } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + api.getConversationInfo + .mockRejectedValueOnce( + new SlackPermanentError("conversations.info", "channel_not_found") + ) + .mockResolvedValue({ lastRead: "1700000005.000000" }); + + await slack.reconcileReadState("C1"); + + expect(saveLink).toHaveBeenCalledTimes(1); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); + }); +}); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index f48ea67a..7f410cec 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -47,7 +47,9 @@ import { unicodeToSlackName } from "./slack-emoji"; import { slackFacets } from "./slack-facets"; import { mentionsUser, type MentionContext } from "./slack-mentions"; import { + channelReadVerdict, deriveReadAnchor, + type SlackReadAnchor, threadReadVerdict, } from "./slack-read-state"; @@ -107,6 +109,15 @@ import { */ const INCREMENTAL_SYNC_COALESCE_MS = 10_000; +/** + * How long an unresolved read anchor is swept before it is dropped. + * + * Past this point Plot's own read state is the one that matters, and a link + * the user has left unread for a month is not going to be settled by Slack's + * cursor. Dropping it keeps the sweep bounded by live work. + */ +const READ_ANCHOR_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; + /** * How far back an event-triggered incremental sync reads. Wide enough to * absorb delayed webhook delivery and the coalescing delay above, narrow @@ -440,6 +451,7 @@ export class Slack extends Connector { await this.cancelScheduledTask(`custom-emoji-sync:${channel.id}`); await this.cancelScheduledTask(`user-groups-sync:${channel.id}`); await this.cancelScheduledTask(`dm-channels-sync:${channel.id}`); + await this.cancelScheduledTask(`read-state-sync:${channel.id}`); await this.cancelDrain(`incremental-sync:${channel.id}`); await this.cancelDrain(`reaction-refresh:${channel.id}`); await this.cancelDrain(`subscribed-thread:${channel.id}`); @@ -461,6 +473,11 @@ export class Slack extends Connector { // thread that already earned its place in Plot keeps updating. const starredKeys = await this.tools.store.list(`starred:${channel.id}:`); for (const key of starredKeys) await this.clear(key); + + // Read anchors are per-conversation reconciliation state, not a record of + // what earned a place in Plot, so unlike `sync_thread:` they are swept. + const anchorKeys = await this.tools.store.list(`read_anchor:${channel.id}:`); + for (const key of anchorKeys) await this.clear(key); } /** @@ -1958,6 +1975,137 @@ export class Slack extends Connector { } } + /** + * Daily reconciliation of Slack's CHANNEL-level read cursor into Plot. + * + * Slack has no push signal for read state — `channel_marked`/`im_marked` are + * RTM-only — so this is the only way a link that has gone quiet ever learns + * the user read it. It uses `conversations.info` (Tier 3) exclusively and + * never touches `conversations.history`/`conversations.replies`, which are + * limited to 1 rpm and are what live message ingestion runs on: a sweep that + * spent that budget would starve inbound messages. + * + * Threaded anchors are skipped — a channel cursor says nothing about whether + * a thread was opened. Those are settled for free on the live path, where + * the thread's own cursor arrives with the messages. + * + * `channelId` is only used to resolve a token; Slack tokens are + * workspace-wide, and the conversations to sweep come from the anchors. + */ + async reconcileReadState(channelId: string): Promise { + let scheduleDaily = true; + try { + await this.set("readStateSyncedAt", Date.now()); + + const entries = await this.tools.store.listEntries( + "read_anchor:" + ); + if (entries.length === 0) return; + + const now = Date.now(); + + // Drop anything past retention before spending an API call on it. + const expired = entries.filter( + ([, anchor]) => now - anchor.at >= READ_ANCHOR_RETENTION_MS + ); + if (expired.length > 0) { + await this.tools.store.clearMany(expired.map(([key]) => key)); + } + + // Threaded anchors are the live path's job; grouping by conversation is + // what keeps this to one `conversations.info` per conversation rather + // than one per link. + const byConversation = new Map(); + for (const [key, anchor] of entries) { + if (anchor.threaded) continue; + if (now - anchor.at >= READ_ANCHOR_RETENTION_MS) continue; + const rest = key.slice("read_anchor:".length); + const separator = rest.indexOf(":"); + if (separator <= 0) continue; + const conversationId = rest.slice(0, separator); + const existing = byConversation.get(conversationId); + if (existing) existing.push([key, anchor]); + else byConversation.set(conversationId, [[key, anchor]]); + } + if (byConversation.size === 0) return; + + let api: SlackApi; + try { + api = await this.getApi(channelId); + } catch (error) { + console.warn("reconcileReadState: Slack token unavailable", error); + return; + } + + for (const [conversationId, anchors] of byConversation) { + let lastRead: string | null; + try { + ({ lastRead } = await api.getConversationInfo(conversationId)); + } catch (error) { + if (error instanceof SlackRateLimitedError) { + // Every remaining conversation needs the same method, so carrying + // on would just re-issue guaranteed-429s. Anchors are durable — + // tomorrow's pass (or the retry below) picks up exactly where this + // one stopped. + console.log( + `reconcileReadState: rate limited on ${error.method}; ${ + byConversation.size + } conversation(s) left for a later pass` + ); + return; + } + if (error instanceof SlackPermanentError) { + if (SLACK_AUTH_ERRORS.has(error.slackError)) { + await this.tools.integrations.markNeedsReauth(channelId); + return; + } + // A single gone/forbidden conversation must not abandon the rest. + console.warn( + `reconcileReadState: skipping ${conversationId}: ${error.method} → ${error.slackError}` + ); + continue; + } + throw error; + } + + const resolved: string[] = []; + for (const [key, anchor] of anchors) { + if (channelReadVerdict(lastRead, anchor.newest) !== "read") continue; + const threadTs = key.slice( + `read_anchor:${conversationId}:`.length + ); + // Minimal upsert. `created` is omitted deliberately: `saveLink` + // reads `unread === false` as the initial-sync signal and DROPS the + // save outright when the item's date predates the plan's sync + // history limit. Title/preview/notes are omitted so the upsert + // preserves whatever is stored rather than rewriting content. + await this.tools.integrations.saveLink({ + // Reuse the connector's own source helper — a reconcile upsert + // that guessed the key would create a second, empty thread + // instead of updating the one the save path wrote. + source: await this.conversationSource(conversationId, threadTs), + channelId: conversationId, + type: (await this.isKnownDMChannel(conversationId)) ? "dm" : "thread", + unread: false, + }); + resolved.push(key); + } + if (resolved.length > 0) await this.tools.store.clearMany(resolved); + } + } catch (error) { + scheduleDaily = false; + console.error("reconcileReadState: unexpected error", error); + throw error; + } finally { + if (scheduleDaily) { + const daily = await this.callback(this.reconcileReadState, channelId); + await this.scheduleRecurring(`read-state-sync:${channelId}`, daily, { + intervalMs: 24 * 60 * 60 * 1000, + }); + } + } + } + async onThreadToDo( thread: Thread, _actor: Actor, @@ -2342,6 +2490,18 @@ export class Slack extends Connector { const dmListCallback = await this.callback(this.listDMChannels, channelId); await this.runTask(dmListCallback); } + + const lastReadSync = await this.get("readStateSyncedAt"); + const readClaimedAt = await this.get("readStateSyncClaimedAt"); + const readClaimed = + readClaimedAt !== null && + readClaimedAt !== undefined && + now - readClaimedAt < CLAIM_TTL_MS; + if ((!lastReadSync || now - lastReadSync >= ONE_DAY_MS) && !readClaimed) { + await this.set("readStateSyncClaimedAt", now); + const readCallback = await this.callback(this.reconcileReadState, channelId); + await this.runTask(readCallback); + } } /** From 4e54106e5be9b59e724d39b616b3283a4ef08c5d Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 14:33:22 -0400 Subject: [PATCH 07/16] fix(slack): keep the read-state chain alive on unexpected errors - Stamp readStateSyncedAt on completion, not entry, matching membersSyncedAt/customEmojiSyncedAt, so a pass that did no real work doesn't suppress queueWorkspaceDailyTasks' 24h backstop. - Stop the generic catch from killing the daily reschedule chain on an unexpected error, mirroring syncMembers. - Cover the auth-shaped SlackPermanentError branch: markNeedsReauth is called and every anchor is left in place. --- connectors/slack/src/slack.test.ts | 27 +++++++++++++++++++++++++-- connectors/slack/src/slack.ts | 16 +++++++++++++--- 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index e0ae7b43..53497ad1 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -3957,9 +3957,10 @@ describe("reconcileReadState", () => { function setup(anchors: Record, lastRead: string | null) { const store = makeStore(anchors); const saveLink = vi.fn().mockResolvedValue("thread-1"); + const markNeedsReauth = vi.fn(); const tools = { store, - integrations: { get: vi.fn(), saveLink }, + integrations: { get: vi.fn(), saveLink, markNeedsReauth }, network: { createWebhook: vi.fn() }, files: {}, }; @@ -3987,7 +3988,7 @@ describe("reconcileReadState", () => { "scheduleRecurring" ).mockResolvedValue(undefined); vi.spyOn(Date, "now").mockReturnValue(NOW); - return { slack, store, saveLink, api }; + return { slack, store, saveLink, api, markNeedsReauth }; } const anchor = (over: Partial<{ newest: string; threaded: boolean; at: number }> = {}) => ({ @@ -4142,4 +4143,26 @@ describe("reconcileReadState", () => { expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); }); + + it("flags reauth and stops the pass on an auth-shaped permanent error", async () => { + const { slack, store, api, saveLink, markNeedsReauth } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + // invalid_auth is in SLACK_AUTH_ERRORS: the grant itself is bad, so this + // is not something a retry (or waiting for the user) will fix. + api.getConversationInfo.mockRejectedValue( + new SlackPermanentError("conversations.info", "invalid_auth") + ); + + await slack.reconcileReadState("C1"); + + expect(markNeedsReauth).toHaveBeenCalledWith("C1"); + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(true); + }); }); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index 7f410cec..1971456e 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -1995,8 +1995,6 @@ export class Slack extends Connector { async reconcileReadState(channelId: string): Promise { let scheduleDaily = true; try { - await this.set("readStateSyncedAt", Date.now()); - const entries = await this.tools.store.listEntries( "read_anchor:" ); @@ -2092,8 +2090,20 @@ export class Slack extends Connector { } if (resolved.length > 0) await this.tools.store.clearMany(resolved); } + + // Stamped on completion, not entry — matches `membersSyncedAt` / + // `customEmojiSyncedAt`. A pass that returned early above (nothing to + // sweep, no token, rate limited) skips this: that's correct, since + // suppressing `queueWorkspaceDailyTasks`'s backstop for 24h on a pass + // that did no real work would leave anchors unreconciled with nothing + // to re-trigger it sooner. + await this.set("readStateSyncedAt", Date.now()); } catch (error) { - scheduleDaily = false; + // Unlike the rate-limit/permanent-error branches above (which return + // early and set nothing here), an unexpected error still lets the + // `finally` re-arm the daily chain below — mirrors `syncMembers`: a + // sweep that dies silently would leave read state unreconciled forever + // with no recovery signal, which is worse than retrying tomorrow. console.error("reconcileReadState: unexpected error", error); throw error; } finally { From 7281f119decc68683dcfe16a36772911ba409f3d Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 14:39:27 -0400 Subject: [PATCH 08/16] feat(slack): mark direct conversations read in Slack --- connectors/slack/src/slack.test.ts | 74 ++++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 59 ++++++++++++++++++++++++ 2 files changed, 133 insertions(+) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 53497ad1..2c1d64e6 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -4166,3 +4166,77 @@ describe("reconcileReadState", () => { expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(true); }); }); + +describe("onThreadRead", () => { + function setup() { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + const api = { markConversationRead: vi.fn().mockResolvedValue(undefined) }; + vi.spyOn( + slack as unknown as { getApi: (c: string) => Promise }, + "getApi" + ).mockResolvedValue(api); + return { slack, api }; + } + + it("marks a direct conversation read in Slack", async () => { + const { slack, api } = setup(); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(api.markConversationRead).toHaveBeenCalledWith( + "D1", + "1700000000.000001" + ); + }); + + it("does nothing for a channel thread — conversations.mark is channel-wide", async () => { + const { slack, api } = setup(); + const thread = { meta: { channelId: "C1", threadTs: "1700000000.000001" } }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); + + it("does nothing when the thread is marked UNREAD — Slack has no un-mark", async () => { + const { slack, api } = setup(); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, true); + + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); + + it("no-ops when the meta carries no anchor message", async () => { + const { slack, api } = setup(); + const thread = { meta: { channelId: "D1", direct: true } }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); + + it("no-ops when the dms scope group was declined", async () => { + const { slack, api } = setup(); + api.markConversationRead.mockRejectedValue( + new SlackPermanentError("conversations.mark", "missing_scope") + ); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await expect( + slack.onThreadRead(thread as never, {} as never, false) + ).resolves.toBeUndefined(); + }); +}); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index 1971456e..3f2f9082 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -2165,6 +2165,65 @@ export class Slack extends Connector { } } + /** + * Write a Plot read back to Slack — direct conversations only. + * + * `conversations.mark` is CONVERSATION-scoped and Slack's public Web API has + * no per-thread equivalent, so this is only coherent where the Plot link IS + * the whole conversation. For a channel thread the same call would move that + * channel's cursor: forward, clearing unread on every other message in it; + * backward, re-unreading messages the user had already read. Both are wrong, + * so channel threads write back nothing. + * + * Marking a thread UNREAD is likewise not propagated — Slack offers no + * un-mark, and re-pointing the cursor at an older message would un-read + * unrelated conversation history. + */ + override async onThreadRead( + thread: Thread, + _actor: Actor, + unread: boolean + ): Promise { + if (unread) return; + const meta = thread.meta ?? {}; + if (meta.direct !== true) return; + const channelId = meta.channelId as string | undefined; + const threadTs = meta.threadTs as string | undefined; + if (!channelId || !threadTs) return; + + let api: SlackApi; + try { + api = await this.getApi(channelId); + } catch (error) { + // Read state already lives in Plot; a missing token is not worth failing + // the dispatch over. + console.warn("onThreadRead: Slack token unavailable", error); + return; + } + + try { + await api.markConversationRead(channelId, threadTs); + } catch (error) { + if (error instanceof SlackRateLimitedError) { + // The read is already recorded in Plot and the next inbound message + // re-establishes the cursor; a deferred write-back is not worth the + // bookkeeping for a marker the user cannot see. + console.log("onThreadRead: rate limited; skipping write-back"); + return; + } + if (error instanceof SlackPermanentError) { + // `missing_scope` here means the optional `dms` group was declined, + // which is a user decision, not a broken connection — degrade quietly + // rather than flagging re-auth. + console.warn( + `onThreadRead: ${error.method} → ${error.slackError}; skipping write-back` + ); + return; + } + throw error; + } + } + // ---- Compose new messages from Plot ---- /** From 5d4fe52bcde9b1fa6140e321ffc5e1d03c9ff2e4 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 14:47:59 -0400 Subject: [PATCH 09/16] fix(slack): flag reauth on genuinely dead tokens in onThreadRead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The blanket SlackPermanentError handler in onThreadRead treated every permanent error the same as a declined optional dms scope, including invalid_auth/token_revoked/account_inactive/no_permission — so a truly revoked token never prompted the user to reconnect. Check missing_scope first (it's itself a member of SLACK_AUTH_ERRORS), then gate markNeedsReauth on SLACK_AUTH_ERRORS membership, matching every other write-back path in this connector. --- connectors/slack/src/slack.test.ts | 37 +++++++++++++++++++++++++----- connectors/slack/src/slack.ts | 18 ++++++++++++--- 2 files changed, 46 insertions(+), 9 deletions(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 2c1d64e6..6563dbdd 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -4170,17 +4170,23 @@ describe("reconcileReadState", () => { describe("onThreadRead", () => { function setup() { const store = makeStore(); - const slack = makeSlack({ + const markNeedsReauth = vi.fn(); + const tools = { store, - integrationsGet: vi.fn(), - createWebhook: vi.fn(), - }); + integrations: { get: vi.fn(), markNeedsReauth }, + network: { createWebhook: vi.fn() }, + files: {}, + }; + const slack = new Slack( + "twist-instance-1" as never, + { getTools: () => tools } as never + ); const api = { markConversationRead: vi.fn().mockResolvedValue(undefined) }; vi.spyOn( slack as unknown as { getApi: (c: string) => Promise }, "getApi" ).mockResolvedValue(api); - return { slack, api }; + return { slack, api, markNeedsReauth }; } it("marks a direct conversation read in Slack", async () => { @@ -4227,7 +4233,7 @@ describe("onThreadRead", () => { }); it("no-ops when the dms scope group was declined", async () => { - const { slack, api } = setup(); + const { slack, api, markNeedsReauth } = setup(); api.markConversationRead.mockRejectedValue( new SlackPermanentError("conversations.mark", "missing_scope") ); @@ -4238,5 +4244,24 @@ describe("onThreadRead", () => { await expect( slack.onThreadRead(thread as never, {} as never, false) ).resolves.toBeUndefined(); + // missing_scope is a member of SLACK_AUTH_ERRORS too, but a declined + // optional group is a user decision, not a broken connection — it must + // NOT be indistinguishable from a genuinely dead token below. + expect(markNeedsReauth).not.toHaveBeenCalled(); + }); + + it("flags reauth on a genuinely dead token — auth-shaped, not missing_scope", async () => { + const { slack, api, markNeedsReauth } = setup(); + api.markConversationRead.mockRejectedValue( + new SlackPermanentError("conversations.mark", "invalid_auth") + ); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await expect( + slack.onThreadRead(thread as never, {} as never, false) + ).resolves.toBeUndefined(); + expect(markNeedsReauth).toHaveBeenCalledWith("D1"); }); }); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index 3f2f9082..f96d0aa0 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -2212,9 +2212,21 @@ export class Slack extends Connector { return; } if (error instanceof SlackPermanentError) { - // `missing_scope` here means the optional `dms` group was declined, - // which is a user decision, not a broken connection — degrade quietly - // rather than flagging re-auth. + if (error.slackError === "missing_scope") { + // The optional `dms` scope group was declined at connect time. + // That is a user decision, not a broken connection — degrade + // quietly rather than prompting a pointless reconnect. + console.warn( + `onThreadRead: ${error.method} → missing_scope; skipping write-back` + ); + return; + } + if (SLACK_AUTH_ERRORS.has(error.slackError)) { + // A genuinely dead token. Flag it the same way every other write-back + // path in this connector does, so the user is prompted to reconnect. + await this.tools.integrations.markNeedsReauth(channelId); + return; + } console.warn( `onThreadRead: ${error.method} → ${error.slackError}; skipping write-back` ); From abb524e2c166e4a18a28863f5b1279564fa1fe58 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 14:53:35 -0400 Subject: [PATCH 10/16] docs(slack): document read-state sync --- connectors/slack/README.md | 54 +++++++++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 10 +++++++ 2 files changed, 64 insertions(+) create mode 100644 connectors/slack/README.md diff --git a/connectors/slack/README.md b/connectors/slack/README.md new file mode 100644 index 00000000..d6e5042a --- /dev/null +++ b/connectors/slack/README.md @@ -0,0 +1,54 @@ +# Slack Connector for Plot + +Follow Slack channels and DMs, reply in threads, and start new conversations. + +## What it does + +- OAuth 2.0 authentication with Slack — a user-token connection only, no bot + user installed in the workspace +- Direct messages and group DMs, each as one ongoing Plot thread +- Channel threads that mention you (directly, via a user group, or through + `@here`/`@channel`) or that you've starred — never whole channels +- Starred (saved) Slack items sync as Plot to-dos +- Real-time sync via the Slack Events API +- Reactions round-trip in both directions +- Replying in Plot posts back to Slack, including file attachments + +## OAuth scopes + +Required: `channels:history`, `channels:read`, `groups:history`, +`groups:read`, `users:read`, `users:read.email`, `chat:write`, `files:write`, +`stars:read`, `stars:write`, `reactions:read`, `reactions:write`. + +Optional (connect-time toggles): custom emoji in reactions (`emoji:read`), +@-mentions of a Slack user group you belong to (`usergroups:read`), and +direct/group DMs (`im:history`, `im:write`, `im:read`, `mpim:history`, +`mpim:write`, `mpim:read`). + +## Read state + +Read state is kept in step with Slack in both directions, within the limits of +what Slack's API exposes. + +**Slack → Plot.** Slack tracks two separate read cursors: one for a channel's +timeline, and one for each thread. A conversation or channel message you have +already read in Slack is marked read in Plot, and a thread you have opened in +Slack is marked read once its own cursor moves. A channel message is settled by +a once-daily reconciliation pass; a thread settles as soon as it sees another +reply, because the thread's cursor arrives with the messages. + +Because the two cursors are independent, catching up on a channel does **not** +mark its threads read — that matches Slack, where a thread you never opened +stays unread in your Threads view. A thread you read in Slack and that then +goes permanently quiet keeps its Plot unread until you open it in Plot. + +Marking something unread in Slack is not propagated to Plot. + +**Plot → Slack.** Reading a direct message in Plot marks that conversation read +in Slack. Channel threads do not write back: Slack's API offers no per-thread +mark, and the only available call moves the entire channel's cursor, which +would clear unread on every other message in it. + +## License + +MIT diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index f96d0aa0..93f60107 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -71,6 +71,9 @@ import { * through `@here`/`@channel`) or that the user saved — never whole channels * - Star-based to-do sync against the user's saved items * - Reactions round-trip on everything that is synced + * - Read state synced from Slack: a conversation you have read in Slack is + * read in Plot, and reading a direct conversation in Plot marks it read in + * Slack * * **Required OAuth User Scopes** (each backs a shipped, user-visible feature — * kept in sync with the authoritative {@link Slack.SCOPES} array below): @@ -100,6 +103,13 @@ import { * `im:history`/`mpim:history`, which only grant reading message content * within a conversation whose id is already known — they do not grant * enumeration, so `listDMChannels` needs both. + * + * Read state uses only scopes the connector already holds: `conversations.info` + * is covered by `channels:read`/`groups:read` (required) or `im:read`/`mpim:read` + * (the optional `dms` group), and `conversations.mark` for direct conversations + * by `im:write`/`mpim:write` (also `dms`). Channel threads are read-in only — + * Slack's public API exposes no per-thread mark, and `conversations.mark` would + * move the whole channel's cursor. */ /** From c8b223f53101157c7ef9b90713065b2a7ca5f454 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 15:27:38 -0400 Subject: [PATCH 11/16] fix(slack): clear the read anchor once Plot's own read state changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit onThreadRead never cleared reconcileReadState's pending-anchor bookkeeping. For a direct conversation, marking it read called conversations.mark, which moved Slack's own cursor to the anchor's newest timestamp — but left the anchor itself in place. Marking the same thread unread again in Plot right after left that stale anchor sitting there, so the next daily sweep saw a cursor it had itself advanced, decided the thread was "read", and reverted the user's manual unread. Clear the anchor at the top of onThreadRead, before either the unread or meta.direct guard, so it covers both a channel thread and a direct conversation, and both directions (mark read or mark unread) — whichever way Plot's read state changed is the user's final word, and no leftover anchor should be able to override it later. A DM's meta.threadTs is the conversation's latest message, not its anchor key, so the anchor id is chosen accordingly (the conversation id for a DM, threadTs otherwise). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xy4JTgqSLoGCd6MYpCjjXH --- connectors/slack/src/slack.test.ts | 54 +++++++++++++++++++++++++++++- connectors/slack/src/slack.ts | 25 ++++++++++++-- 2 files changed, 76 insertions(+), 3 deletions(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 6563dbdd..d434e1bd 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -4186,7 +4186,7 @@ describe("onThreadRead", () => { slack as unknown as { getApi: (c: string) => Promise }, "getApi" ).mockResolvedValue(api); - return { slack, api, markNeedsReauth }; + return { slack, store, api, markNeedsReauth }; } it("marks a direct conversation read in Slack", async () => { @@ -4264,4 +4264,56 @@ describe("onThreadRead", () => { ).resolves.toBeUndefined(); expect(markNeedsReauth).toHaveBeenCalledWith("D1"); }); + + it("clears a channel thread's read anchor when Plot marks it read", async () => { + const { slack, store } = setup(); + store.map.set("read_anchor:C1:1700000000.000001", { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }); + const thread = { meta: { channelId: "C1", threadTs: "1700000000.000001" } }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); + }); + + it("clears a DM's read anchor keyed on the conversation id, not meta.threadTs", async () => { + const { slack, store } = setup(); + // The DM anchor lives at read_anchor:D1:D1 — keyed on the conversation id + // in both positions, never on meta.threadTs (which names the + // conversation's latest message, not its anchor). + store.map.set("read_anchor:D1:D1", { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(store.map.has("read_anchor:D1:D1")).toBe(false); + }); + + it("clears the anchor even when marking UNREAD, so a manual override can't be reverted by the next sweep", async () => { + const { slack, store, api } = setup(); + store.map.set("read_anchor:D1:D1", { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, true); + + expect(store.map.has("read_anchor:D1:D1")).toBe(false); + // Marking unread still writes back nothing to Slack — only the anchor + // bookkeeping changed. + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); }); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index 93f60107..4f289494 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -2188,17 +2188,38 @@ export class Slack extends Connector { * Marking a thread UNREAD is likewise not propagated — Slack offers no * un-mark, and re-pointing the cursor at an older message would un-read * unrelated conversation history. + * + * Whichever direction Plot's read state changed, that's the user's final + * word — any outstanding {@link reconcileReadState} anchor for this link is + * cleared FIRST, before either the `unread` or `meta.direct` guard below. + * Skipping this for a DM read used to let the very `conversations.mark` call + * a few lines down move Slack's cursor to `anchor.newest` without ever + * clearing the anchor Slack was catching up to — so marking the thread + * unread again in Plot right after left a stale anchor in place, and the + * next sweep saw a cursor it had itself advanced and reverted the user's + * unread. Clearing here — for both link shapes and both directions — keeps + * a manual mark-unread from being undone the same way even without a + * write-back to trigger it. */ override async onThreadRead( thread: Thread, _actor: Actor, unread: boolean ): Promise { - if (unread) return; const meta = thread.meta ?? {}; - if (meta.direct !== true) return; const channelId = meta.channelId as string | undefined; const threadTs = meta.threadTs as string | undefined; + if (channelId && threadTs) { + // A DM's `meta.threadTs` is the conversation's latest message, not its + // anchor key — `buildConversationLink` anchors a direct conversation on + // the conversation id in both positions (see `readAnchorKey`), so using + // `threadTs` directly here would clear nothing. + const anchorId = meta.direct === true ? channelId : threadTs; + await this.clear(this.readAnchorKey(channelId, anchorId)); + } + + if (unread) return; + if (meta.direct !== true) return; if (!channelId || !threadTs) return; let api: SlackApi; From 57318529446a3190a794eac30e621368cafb4566 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 15:28:32 -0400 Subject: [PATCH 12/16] fix(slack): gate the channel-thread read anchor write like the DM anchor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit buildConversationLink's two branches each inlined their own copy of the read-anchor write logic, and they had drifted apart: the DM branch only wrote an anchor when advanceConversationHead && link.unread !== false, but the channel-thread branch wrote (or rewrote) an anchor whenever the read verdict wasn't "read" — regardless of advanceConversationHead. refreshSlackThread calls buildConversationLink with advanceConversationHead: false for a reaction refresh. A root-only channel message has no thread cursor of its own, so its read verdict is always "unknown" — which the channel-thread branch treated as "not read" and wrote the same anchor right back, even though no new content had actually arrived. The next daily sweep then saw that anchor and re-asserted "read" over a user's manual unread. Extract the shared applyReadAnchor helper and route both branches through it, so one gating rule governs both: advanceConversationHead && link.unread !== false for the write, with a solid "read" verdict from a cursor always allowed to clear the anchor regardless of advanceConversationHead. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xy4JTgqSLoGCd6MYpCjjXH --- connectors/slack/src/slack.test.ts | 47 +++++++++ connectors/slack/src/slack.ts | 148 +++++++++++++++++++++-------- 2 files changed, 154 insertions(+), 41 deletions(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index d434e1bd..e54e8052 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -3949,6 +3949,53 @@ describe("read anchors", () => { expect(link?.unread).toBe(false); expect(store.map.has("read_anchor:D1:D1")).toBe(false); }); + + it("does not recreate a channel-thread anchor on a reaction refresh (advanceConversationHead: false)", async () => { + // Regression test: `refreshSlackThread` re-fetches with + // `advanceConversationHead: false` for a reaction refresh. A root-only + // channel message has no thread cursor of its own, so the read verdict + // is "unknown" — that must NOT fall through to rewriting the anchor, + // since no new content actually arrived. + const store = makeStore({ + "read_anchor:C1:1700000000.000001": { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }, + }); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + + await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise; + }).buildConversationLink({ + channelId: "C1", + messages: [parent], + initialSync: false, + advanceConversationHead: false, + }); + + const anchorSetCalls = store.set.mock.calls.filter( + (call) => call[0] === "read_anchor:C1:1700000000.000001" + ); + expect(anchorSetCalls).toHaveLength(0); + expect( + (store.map.get("read_anchor:C1:1700000000.000001") as { at: number }).at + ).toBe(1000); + }); }); describe("reconcileReadState", () => { diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index 4f289494..869baaac 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -834,28 +834,31 @@ export class Slack extends Connector { // and rewinding what this one just wrote. await this.set(this.dmHeadKey(channelId), true); } - // Anchor the conversation for the daily sweep. A direct conversation is - // never `threaded`: this one permanent link flattens Slack's reply - // threads into a running conversation, so the CONVERSATION cursor is the - // only cursor that describes it. Keyed on the conversation id in both - // positions — there is no thread root to key on. - // - // Only when these messages ARE the conversation head. A reaction or a - // save can name a message of any age, and an anchor rewound to an older - // `newest` would make the sweep declare the conversation read on a - // cursor that has not actually reached its latest message. - // - // Only when the link is NOT already read. `assembleSlackDmLink` sets - // `unread: false` for `initialSync`, which already asserts read — an - // anchor for it could only ever re-assert what is already true, and - // would outlive a later manual "mark unread" in Plot, letting the sweep - // silently revert it once Slack's cursor caught up. - if (advanceConversationHead && link.unread !== false) { - const dmAnchor = deriveReadAnchor(kept, { direct: true, at: Date.now() }); - if (dmAnchor) { - await this.set(this.readAnchorKey(channelId, channelId), dmAnchor); - } - } + // Anchor the conversation for the daily sweep via the shared helper + // below (see `applyReadAnchor`). A direct conversation is never + // `threaded`: this one permanent link flattens Slack's reply threads + // into a running conversation, so the CONVERSATION cursor is the only + // cursor that describes it, and it is keyed on the conversation id in + // both positions — there is no thread root to key on. There is no + // live, cursor-informed "read" verdict for a DM (the channel cursor is + // settled only by the daily `reconcileReadState` sweep), so `read` is + // always `false` here; the helper's write gate + // (`advanceConversationHead && link.unread !== false`) still applies, + // covering both "these messages are not the conversation head" (a + // reaction or save naming an older message, which must not rewind the + // anchor) and "the link is already read" (`assembleSlackDmLink` sets + // `unread: false` for `initialSync`, so an anchor for it could only + // ever re-assert what is already true and would outlive a later manual + // "mark unread" in Plot). + await this.applyReadAnchor({ + channelId, + anchorId: channelId, + messages: kept, + direct: true, + link, + read: false, + advanceConversationHead, + }); return link; } @@ -882,41 +885,104 @@ export class Slack extends Connector { syncableId: channelId, }; if (messages[0]) link.facets = slackFacets(messages[0], channelId); - // Apply Slack's per-thread read cursor, and record what still needs - // settling. `deriveReadAnchor` tells us which cursor governs this link: a - // root-only channel message lives purely in the channel timeline, so only - // the daily sweep's `conversations.info` can speak for it. + // Apply Slack's per-thread read cursor via the same shared helper the DM + // branch above uses (see `applyReadAnchor`). A root-only channel message + // has no thread cursor of its own — only the daily sweep's + // `conversations.info` can speak for it — so `threaded` and the read + // verdict below are both false, and the helper's write gate + // (`advanceConversationHead && link.unread !== false`) is what stops a + // stale anchor from being recreated when nothing new actually arrived + // (e.g. `refreshSlackThread`'s reaction refresh, which re-fetches with + // `advanceConversationHead: false`). // // Read from THIS response, never from a cache: the parent's // `unread_count` is computed after the reply we are saving landed, so a // genuine new-message unread can never be suppressed by a stale cursor. // - // The anchor's EXISTENCE is the transition gate — it means "not yet marked - // read for this `newest` ts". Deleting it on a successful mark-read is - // what makes a later manual "mark unread" in Plot stick: the sweep has - // nothing left to act on until new content recreates the anchor. - // // `initialSync` already asserts read (it sets `unread: false`), so those // links need no anchor at all — the sweep would only re-assert what is // already true. const threadTs = (link.meta?.threadTs as string | undefined) ?? messages[0]?.ts; - const anchor = deriveReadAnchor(messages, { direct: false, at: Date.now() }); - if (anchor && threadTs) { - const anchorKey = this.readAnchorKey(channelId, threadTs); + if (threadTs) { + const threaded = Boolean( + deriveReadAnchor(messages, { direct: false, at: Date.now() })?.threaded + ); const read = link.unread === false || - (anchor.threaded && threadReadVerdict(messages[0]) === "read"); - if (read) { - link.unread = false; - await this.clear(anchorKey); - } else { - await this.set(anchorKey, anchor); - } + (threaded && threadReadVerdict(messages[0]) === "read"); + await this.applyReadAnchor({ + channelId, + anchorId: threadTs, + messages, + direct: false, + link, + read, + advanceConversationHead, + }); } return link; } + /** + * Settle one link's read-anchor bookkeeping: mark it read (and drop the + * anchor) when the caller already has a cursor-backed verdict, or decide + * whether an anchor still needs to be (re)written to track it. + * + * The anchor's EXISTENCE is the transition gate for + * {@link reconcileReadState}: present means "not yet marked read for this + * `newest` ts". Deleting it on a successful mark-read is what makes a later + * manual "mark unread" in Plot stick — the sweep has nothing left to act on + * until new content recreates the anchor. + * + * Shared by both {@link buildConversationLink} branches (direct + * conversation and channel thread) so one gating rule governs both writes. + * They used to each inline their own version of this and drifted apart: the + * channel-thread copy wrote an anchor unconditionally whenever the verdict + * wasn't "read", even on a `refreshSlackThread` reaction refresh that asked + * `advanceConversationHead: false` — a root-only message has no thread + * cursor, so that verdict is always "unknown", and the anchor was + * recreated with the same `newest` even though no new content had arrived, + * letting the next sweep re-assert read over a user's manual unread. + * + * `advanceConversationHead` gates the WRITE only, never the read-clear: + * solid evidence that a cursor already says read is safe to act on no + * matter why the caller fetched these messages, but recreating a "not yet + * read" anchor is only valid when the caller is reporting the + * conversation's actual current head. + */ + private async applyReadAnchor(opts: { + channelId: string; + /** Thread root ts, or the conversation id for a direct conversation. */ + anchorId: string; + messages: SlackMessage[]; + direct: boolean; + link: { unread?: boolean }; + /** Did a cursor say this link is read? */ + read: boolean; + advanceConversationHead: boolean; + }): Promise { + const { + channelId, + anchorId, + messages, + direct, + link, + read, + advanceConversationHead, + } = opts; + const anchorKey = this.readAnchorKey(channelId, anchorId); + if (read) { + link.unread = false; + await this.clear(anchorKey); + return; + } + if (advanceConversationHead && link.unread !== false) { + const anchor = deriveReadAnchor(messages, { direct, at: Date.now() }); + if (anchor) await this.set(anchorKey, anchor); + } + } + /** * The `source` the link for this conversation is keyed on — the same value * {@link buildConversationLink} produces, so any path addressing an From b362bd7f9f931d1a9e3a285e26e99cb5188fe1b2 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 15:29:27 -0400 Subject: [PATCH 13/16] fix(slack): harden the daily read-state sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three related problems in reconcileReadState, all fixed together since they sit in the same loop: - No once-per-day guard. queueWorkspaceDailyTasks dispatches this once per ENABLED CHANNEL, but the sweep itself is workspace-scoped (it reads every read_anchor: entry regardless of channel). Slack auto-observes channels as the user composes into them, so each newly-enabled channel armed its own daily recurring chain — keyed per channel, so scheduleRecurring couldn't dedupe them — meaning N chains each doing a full workspace sweep once a day: N times the conversations.info calls and N duplicate writes per anchor that resolves. Add the same top-of-function 24h guard syncMembers uses, gated on `readStateSyncedAt`. The vestigial `scheduleDaily` flag (declared true, never reassigned) is removed along with it — the finally block now always re-arms the one chain that gets past the guard. - missing_scope and no_permission were treated as evidence the whole connection is dead. onThreadRead already special-cases missing_scope (a declined optional scope group is a user decision, not a broken connection); reconcileReadState predates that fix. A user who declined the optional dms group would hit missing_scope on the first DM anchor, get a spurious "reconnect Slack" prompt on an otherwise healthy connection, and never sweep the channels behind it. Mirror onThreadRead: missing_scope and no_permission both skip just that conversation and continue the pass; only the remaining SLACK_AUTH_ERRORS codes flag reauth and stop. - The rate-limit log reported byConversation.size (the total conversation count) as "left for a later pass" instead of the actual remainder. Iterate with an index so the count reflects what's really left. Also pass `author: null` on the reconcile upsert — it's a genuinely authorless write (only flips `unread`), and this silences saveLink's development-time unattributed-link warning that would otherwise fire on every reconciled link. Adds coverage for a DM-shaped anchor in the sweep: `read_anchor:D1:D1` parses to a threadTs of "D1", a non-timestamp fed straight into conversationSource(). Nothing previously exercised this path; a regression in isKnownDMChannel here would synthesise a bogus app_redirect URL and create a new, empty thread instead of updating the real DM link. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xy4JTgqSLoGCd6MYpCjjXH --- connectors/slack/src/slack.test.ts | 106 +++++++++++++++++++++++++++++ connectors/slack/src/slack.ts | 75 ++++++++++++++++---- 2 files changed, 167 insertions(+), 14 deletions(-) diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index e54e8052..4cb6e233 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -4212,6 +4212,112 @@ describe("reconcileReadState", () => { expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(true); }); + + it("skips a conversation whose optional dms scope was declined, without abandoning the rest", async () => { + const { slack, store, api, saveLink, markNeedsReauth } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + // missing_scope is a member of SLACK_AUTH_ERRORS too, but a declined + // optional scope group is a user decision, not a broken connection — + // mirrors onThreadRead's handling of the same error. + api.getConversationInfo + .mockRejectedValueOnce( + new SlackPermanentError("conversations.info", "missing_scope") + ) + .mockResolvedValue({ lastRead: "1700000005.000000" }); + + await slack.reconcileReadState("C1"); + + expect(markNeedsReauth).not.toHaveBeenCalled(); + expect(saveLink).toHaveBeenCalledTimes(1); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); + }); + + it("skips a conversation the token can no longer reach (no_permission), without abandoning the rest", async () => { + const { slack, store, api, saveLink, markNeedsReauth } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + // no_permission on THIS conversation means the user lost access to it + // specifically — not evidence the token itself is dead. + api.getConversationInfo + .mockRejectedValueOnce( + new SlackPermanentError("conversations.info", "no_permission") + ) + .mockResolvedValue({ lastRead: "1700000005.000000" }); + + await slack.reconcileReadState("C1"); + + expect(markNeedsReauth).not.toHaveBeenCalled(); + expect(saveLink).toHaveBeenCalledTimes(1); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); + }); + + it("resolves a DM anchor to the DM identity source, never an app_redirect URL", async () => { + // The DM anchor key (`read_anchor:D1:D1`) parses to a threadTs of "D1" — + // a non-timestamp fed straight into conversationSource(). If + // isKnownDMChannel ever returned false here, the sweep would synthesise + // a bogus app_redirect URL and create a brand-new, empty thread instead + // of updating the real DM link. + const { slack, saveLink } = setup( + { "read_anchor:D1:D1": anchor() }, + "1700000005.000000" + ); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + + await slack.reconcileReadState("C1"); + + expect(saveLink).toHaveBeenCalledTimes(1); + const saved = saveLink.mock.calls[0][0]; + expect(saved.type).toBe("dm"); + expect(saved.source).not.toMatch(/^https:\/\/slack\.com\/app_redirect/); + expect(saved.source).toBe("slack:chat:D1"); + }); + + it("does no work on a second invocation within 24h of the last sweep", async () => { + // The anchor stays behind the channel cursor (verdict "unread"), so it + // survives the first pass — proving the second call's silence comes from + // the top-of-function guard, not from there being nothing left to sweep. + const { slack, api } = setup( + { "read_anchor:C1:1700000005.000000": anchor({ newest: "1700000005.000000" }) }, + "1700000000.000001" + ); + + await slack.reconcileReadState("C1"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(1); + + await slack.reconcileReadState("C1"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(1); + }); + + it("arms a fresh 24h chain for a second channel once the guard has passed", async () => { + // Sanity check on the other side of the same guard: once 24h have + // elapsed (simulated by advancing the mocked clock), a differently-keyed + // invocation is free to sweep again. + const { slack, api } = setup( + { "read_anchor:C1:1700000005.000000": anchor({ newest: "1700000005.000000" }) }, + "1700000000.000001" + ); + + await slack.reconcileReadState("C1"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(1); + + vi.spyOn(Date, "now").mockReturnValue(NOW + 24 * 60 * 60 * 1000 + 1); + await slack.reconcileReadState("C2"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(2); + }); }); describe("onThreadRead", () => { diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index 869baaac..f5c6a94a 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -2067,17 +2067,32 @@ export class Slack extends Connector { * * `channelId` is only used to resolve a token; Slack tokens are * workspace-wide, and the conversations to sweep come from the anchors. + * + * Gated to run at most once per 24 hours per WORKSPACE (like `syncMembers`, + * placed the same way — before the `try`, so a redundant call skips the + * `finally` below too and never arms a chain of its own). The sweep itself + * is workspace-scoped (`listEntries("read_anchor:")`, no channel filter), + * but the caller, `queueWorkspaceDailyTasks`, dispatches it once per + * ENABLED CHANNEL, and Slack auto-observes new channels as the user + * composes into them. Without this guard, every newly-enabled channel would + * arm its own daily recurring chain (`read-state-sync:` is keyed + * per channel, so `scheduleRecurring` cannot dedupe them against each + * other) — N chains sweeping the same anchors once a day each means N times + * the `conversations.info` calls and N duplicate writes per anchor that + * actually resolves. */ async reconcileReadState(channelId: string): Promise { - let scheduleDaily = true; + const now = Date.now(); + const ONE_DAY_MS = 24 * 60 * 60 * 1000; + const lastReadSync = await this.get("readStateSyncedAt"); + if (lastReadSync && now - lastReadSync < ONE_DAY_MS) return; + try { const entries = await this.tools.store.listEntries( "read_anchor:" ); if (entries.length === 0) return; - const now = Date.now(); - // Drop anything past retention before spending an API call on it. const expired = entries.filter( ([, anchor]) => now - anchor.at >= READ_ANCHOR_RETENTION_MS @@ -2111,7 +2126,9 @@ export class Slack extends Connector { return; } - for (const [conversationId, anchors] of byConversation) { + const conversations = [...byConversation]; + for (let i = 0; i < conversations.length; i++) { + const [conversationId, anchors] = conversations[i]!; let lastRead: string | null; try { ({ lastRead } = await api.getConversationInfo(conversationId)); @@ -2120,15 +2137,36 @@ export class Slack extends Connector { // Every remaining conversation needs the same method, so carrying // on would just re-issue guaranteed-429s. Anchors are durable — // tomorrow's pass (or the retry below) picks up exactly where this - // one stopped. + // one stopped. `conversations.length - i` is what's actually left + // — this one included, since it never got resolved — not the + // conversation-count total the map started with. + const remaining = conversations.length - i; console.log( - `reconcileReadState: rate limited on ${error.method}; ${ - byConversation.size - } conversation(s) left for a later pass` + `reconcileReadState: rate limited on ${error.method}; ${remaining} conversation(s) left for a later pass` ); return; } if (error instanceof SlackPermanentError) { + if (error.slackError === "missing_scope") { + // The optional `dms` scope group was declined at connect time. + // That is a user decision, not a broken connection — mirrors + // `onThreadRead`'s handling of the same error, so a user who + // opted out of DM sync doesn't get a spurious reconnect prompt + // while the channels behind this DM in `byConversation` go + // unswept. + console.warn( + `reconcileReadState: skipping ${conversationId}: ${error.method} → missing_scope` + ); + continue; + } + if (error.slackError === "no_permission") { + // This conversation specifically is unreachable (e.g. the user + // was removed from it) — not evidence the token itself is dead. + console.warn( + `reconcileReadState: skipping ${conversationId}: ${error.method} → no_permission` + ); + continue; + } if (SLACK_AUTH_ERRORS.has(error.slackError)) { await this.tools.integrations.markNeedsReauth(channelId); return; @@ -2153,6 +2191,10 @@ export class Slack extends Connector { // save outright when the item's date predates the plan's sync // history limit. Title/preview/notes are omitted so the upsert // preserves whatever is stored rather than rewriting content. + // `author: null` documents that this upsert is genuinely + // authorless — it only flips `unread`, never introduces content — + // and silences `saveLink`'s development-time unattributed-link + // warning that would otherwise fire on every reconciled link. await this.tools.integrations.saveLink({ // Reuse the connector's own source helper — a reconcile upsert // that guessed the key would create a second, empty thread @@ -2161,6 +2203,7 @@ export class Slack extends Connector { channelId: conversationId, type: (await this.isKnownDMChannel(conversationId)) ? "dm" : "thread", unread: false, + author: null, }); resolved.push(key); } @@ -2183,12 +2226,16 @@ export class Slack extends Connector { console.error("reconcileReadState: unexpected error", error); throw error; } finally { - if (scheduleDaily) { - const daily = await this.callback(this.reconcileReadState, channelId); - await this.scheduleRecurring(`read-state-sync:${channelId}`, daily, { - intervalMs: 24 * 60 * 60 * 1000, - }); - } + // Unconditional: nothing in this function suppresses the daily + // reschedule (unlike `syncMembers`, which stops its own chain on a + // permanent error) — even a dead-token pass is worth retrying + // tomorrow, since the user may have reauthed by then. The top-of- + // function guard above is what stops a REDUNDANT chain from ever being + // armed; this always re-arms the one chain that got past it. + const daily = await this.callback(this.reconcileReadState, channelId); + await this.scheduleRecurring(`read-state-sync:${channelId}`, daily, { + intervalMs: ONE_DAY_MS, + }); } } From 99c669aa184dec0f4ce8c99090f3d784b867f95b Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 15:29:39 -0400 Subject: [PATCH 14/16] fix(slack): correct onChannelDisabled's read-anchor cleanup comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The comment claimed this sweep clears "read anchors" for a disabled channel without qualification, but read_anchor:${channel.id}: only ever matches a channel thread's anchor. A direct conversation's anchor is keyed on the conversation id in both positions (read_anchor::), and a DM conversation id is never an enabled channel, so this sweep can't reach it — no behavior change, just documenting the actual scope. DM anchors are left to age out via reconcileReadState's 30-day retention. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xy4JTgqSLoGCd6MYpCjjXH --- connectors/slack/src/slack.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index f5c6a94a..d075a660 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -485,7 +485,14 @@ export class Slack extends Connector { for (const key of starredKeys) await this.clear(key); // Read anchors are per-conversation reconciliation state, not a record of - // what earned a place in Plot, so unlike `sync_thread:` they are swept. + // what earned a place in Plot, so unlike `sync_thread:` they are swept — + // but only a CHANNEL THREAD's anchor is keyed on this channel id + // (`read_anchor::`) and reachable this way. A + // direct conversation's anchor is keyed on the conversation id in both + // positions (`read_anchor::`), and a DM + // conversation id is never an enabled channel, so this sweep does not + // reach it. That's fine left alone: DM anchors are bounded by + // `reconcileReadState`'s 30-day retention regardless. const anchorKeys = await this.tools.store.list(`read_anchor:${channel.id}:`); for (const key of anchorKeys) await this.clear(key); } From b960d2f3442bcfb87d4775e033c799d6c1e7f04c Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 15:29:53 -0400 Subject: [PATCH 15/16] fix(slack): abstain on a malformed Slack timestamp instead of asserting read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit compareSlackTs returned 1 (the "greater than" direction) whenever either half of a timestamp failed to parse as a number, because every JS comparison against NaN (< 0, >= 0, === 0) evaluates to false, and the old `secDiff < 0 ? -1 : 1` shape defaulted to 1 in that case. channelReadVerdict reads a compareSlackTs result of >= 0 as "read" — so a malformed cursor failed toward the UNSAFE direction, asserting a link read on a timestamp comparison that was actually meaningless. compareSlackTs now returns NaN when either half doesn't parse, and both channelReadVerdict and threadReadVerdict's last_read/latest_reply fallback check Number.isNaN and abstain ("unknown") instead of falling through to a directional default. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xy4JTgqSLoGCd6MYpCjjXH --- connectors/slack/src/slack-read-state.test.ts | 31 +++++++++++++++++++ connectors/slack/src/slack-read-state.ts | 23 +++++++++++--- 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/connectors/slack/src/slack-read-state.test.ts b/connectors/slack/src/slack-read-state.test.ts index 75ef45a2..c04dd1c4 100644 --- a/connectors/slack/src/slack-read-state.test.ts +++ b/connectors/slack/src/slack-read-state.test.ts @@ -32,6 +32,21 @@ describe("compareSlackTs", () => { it("treats Slack's never-read sentinel as older than everything", () => { expect(compareSlackTs("0000000000.000000", "1700000000.000001")).toBe(-1); }); + + it("returns NaN for a malformed timestamp rather than defaulting to a direction", () => { + // A naive `< 0 ? -1 : 1` shape resolves every NaN comparison to `1` — + // silently the UNSAFE "read" direction for a caller like + // channelReadVerdict. NaN must propagate so callers can detect it. + expect(Number.isNaN(compareSlackTs("not-a-timestamp", "1700000000.000001"))).toBe( + true + ); + expect(Number.isNaN(compareSlackTs("1700000000.000001", "not-a-timestamp"))).toBe( + true + ); + expect(Number.isNaN(compareSlackTs("1700000000.abc", "1700000000.000001"))).toBe( + true + ); + }); }); describe("channelReadVerdict", () => { @@ -49,6 +64,14 @@ describe("channelReadVerdict", () => { expect(channelReadVerdict(undefined, "1700000000.000001")).toBe("unknown"); expect(channelReadVerdict("", "1700000000.000001")).toBe("unknown"); }); + + it("abstains rather than asserting read on a malformed cursor", () => { + // Regression: compareSlackTs used to fail toward "read" (the unsafe + // direction) on a NaN half; this must abstain instead. + expect(channelReadVerdict("not-a-timestamp", "1700000000.000001")).toBe( + "unknown" + ); + }); }); describe("threadReadVerdict", () => { @@ -83,6 +106,14 @@ describe("threadReadVerdict", () => { "unknown" ); }); + + it("abstains rather than asserting read on a malformed fallback-pair cursor", () => { + expect( + threadReadVerdict( + msg({ ts: "1.0", last_read: "not-a-timestamp", latest_reply: "1700000002.000000" }) + ) + ).toBe("unknown"); + }); }); describe("deriveReadAnchor", () => { diff --git a/connectors/slack/src/slack-read-state.ts b/connectors/slack/src/slack-read-state.ts index 4227f2d7..027b6835 100644 --- a/connectors/slack/src/slack-read-state.ts +++ b/connectors/slack/src/slack-read-state.ts @@ -16,15 +16,26 @@ export type SlackReadVerdict = "read" | "unread" | "unknown"; * Done on the two halves as integers rather than via `parseFloat`: a Slack ts * carries 16 significant digits, which is at the edge of float64 precision, * so `parseFloat` can round two genuinely different timestamps to the same - * value. Returns <0, 0, or >0 like a comparator. + * value. Returns <0, 0, or >0 like a comparator — or `NaN` when either + * timestamp is malformed enough that neither half parses as a number, so a + * meaningful comparison isn't possible. + * + * Callers MUST check `Number.isNaN` on the result and treat it as "unknown", + * never fall through to a directional default: every JS comparison against + * `NaN` (`< 0`, `>= 0`, `=== 0`) evaluates to `false`, so an unguarded + * `compareSlackTs(...) >= 0 ? "read" : "unread"` silently resolves to + * "unread" — and the equally unguarded `< 0 ? -1 : 1` shape used to resolve + * to `1`, i.e. "read", the UNSAFE direction for a cursor comparison. */ export function compareSlackTs(a: string, b: string): number { const [aSec = "0", aMicro = "0"] = a.split("."); const [bSec = "0", bMicro = "0"] = b.split("."); const secDiff = Number(aSec) - Number(bSec); + if (Number.isNaN(secDiff)) return NaN; if (secDiff !== 0) return secDiff < 0 ? -1 : 1; const microDiff = Number(aMicro.padEnd(6, "0")) - Number(bMicro.padEnd(6, "0")); + if (Number.isNaN(microDiff)) return NaN; return microDiff === 0 ? 0 : microDiff < 0 ? -1 : 1; } @@ -41,7 +52,9 @@ export function channelReadVerdict( newestTs: string ): SlackReadVerdict { if (!lastRead) return "unknown"; - return compareSlackTs(lastRead, newestTs) >= 0 ? "read" : "unread"; + const cmp = compareSlackTs(lastRead, newestTs); + if (Number.isNaN(cmp)) return "unknown"; + return cmp >= 0 ? "read" : "unread"; } /** @@ -61,9 +74,9 @@ export function threadReadVerdict( return parent.unread_count === 0 ? "read" : "unread"; } if (parent.last_read && parent.latest_reply) { - return compareSlackTs(parent.last_read, parent.latest_reply) >= 0 - ? "read" - : "unread"; + const cmp = compareSlackTs(parent.last_read, parent.latest_reply); + if (Number.isNaN(cmp)) return "unknown"; + return cmp >= 0 ? "read" : "unread"; } return "unknown"; } From 3231766a63a1e91686ae9784c960710da5b23405 Mon Sep 17 00:00:00 2001 From: Kris Braun Date: Fri, 31 Jul 2026 15:36:05 -0400 Subject: [PATCH 16/16] fix(slack): preserve thread titles on the reconcile upsert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reconcileReadState's upsert deliberately sent no title, on the assumption that omitting a field always preserves the stored value. That holds for `preview` (its platform default derives from the notes, so with none it's null and COALESCE falls through) but not for `title`: `upsert_thread` takes an "archived" code path whenever the user's `thread_priority` row points at an archived priority (they archived the focus a thread was filed under) or is missing, and on that path the platform's title default is the literal string "Untitled" — never null — so an omitted title there always loses to it. A perfectly healthy connection reconciling a thread filed under an archived focus would silently destroy that thread's real title. Carry the title on the read anchor instead, so the upsert has something to re-send: - `SlackReadAnchor` gains `title?: string | null`; `deriveReadAnchor` takes it via its options rather than trying to infer it from the messages. - `applyReadAnchor` records the link's current title when it (re)writes an anchor. `assembleSlackDmLink` deliberately omits `title` when `users.info` was unavailable, to avoid permanently renaming a DM thread to a raw Slack user id — so when the link carries no title, the previously-stored anchor's title is carried forward instead of being dropped in that window. - `reconcileReadState`'s `saveLink` call includes `title` when the anchor has one, and omits it (as before) when it doesn't — never worse than today. Sending `title` also re-writes it on the normal, non-archived upsert path, which is harmless: a channel thread's title is derived deterministically from its root message and a DM's is the counterparty name, so both are re-sent unchanged by every ordinary sync anyway — this isn't a new source of truth, just re-asserting the same value on a path that already agreed with itself. `created`, `schedules`, `preview`, and `notes` remain omitted from the reconcile upsert, unchanged. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Xy4JTgqSLoGCd6MYpCjjXH --- connectors/slack/src/slack-read-state.ts | 19 ++++- connectors/slack/src/slack.test.ts | 92 +++++++++++++++++++++++- connectors/slack/src/slack.ts | 55 +++++++++++--- 3 files changed, 153 insertions(+), 13 deletions(-) diff --git a/connectors/slack/src/slack-read-state.ts b/connectors/slack/src/slack-read-state.ts index 027b6835..3b1109ee 100644 --- a/connectors/slack/src/slack-read-state.ts +++ b/connectors/slack/src/slack-read-state.ts @@ -88,11 +88,22 @@ export function threadReadVerdict( * governs: a channel link holding real thread replies is settled by the thread * cursor on the live path, everything else by the channel cursor in the daily * sweep. `at` is the write time, for the retention drop. + * + * `title` rides along so `reconcileReadState`'s upsert can re-send it — + * `upsert_thread` takes an "archived" code path whenever the user's + * `thread_priority` row points at an archived priority (or is missing), and + * on that path the platform's title default is the literal string + * "Untitled", never null, so an upsert that omitted `title` there would + * destroy the thread's real one. `undefined`/`null` when the link that wrote + * this anchor had no title to carry (see `applyReadAnchor` in slack.ts, + * which falls back to a previously-stored anchor's title rather than losing + * it in that case). */ export type SlackReadAnchor = { newest: string; threaded: boolean; at: number; + title?: string | null; }; /** @@ -101,10 +112,14 @@ export type SlackReadAnchor = { * A direct conversation is never `threaded`: its link flattens Slack's reply * threads into one running conversation, so the conversation cursor is the * only cursor that describes it. + * + * `title` is accepted rather than inferred from `messages`: only the caller + * has the link (and, for the carry-forward case, the previously-stored + * anchor) to source it from. */ export function deriveReadAnchor( messages: SlackMessage[], - opts: { direct: boolean; at: number } + opts: { direct: boolean; at: number; title?: string | null } ): SlackReadAnchor | null { let newest: string | null = null; let threaded = false; @@ -115,5 +130,5 @@ export function deriveReadAnchor( } } if (!newest) return null; - return { newest, threaded, at: opts.at }; + return { newest, threaded, at: opts.at, title: opts.title }; } diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 4cb6e233..65fc59fa 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -3907,9 +3907,70 @@ describe("read anchors", () => { const anchor = store.map.get("read_anchor:D1:D1") as { newest: string; threaded: boolean; + title?: string | null; }; expect(anchor.newest).toBe("1700000002.000000"); expect(anchor.threaded).toBe(false); + // No `userInfos` was passed, so assembleSlackDmLink omits `title` (see + // its own comment on why — the raw Slack user id would otherwise + // permanently rename a real person's thread). Nothing else supplied one + // here, so the anchor genuinely has none to carry. + expect(anchor.title).toBeUndefined(); + }); + + it("carries a DM anchor's previously-stored title forward when the new link has none", async () => { + // assembleSlackDmLink omits `title` whenever `users.info` was + // unavailable — a real, recurring window, not an edge case. The anchor + // must not lose a title it already had just because a later sync landed + // in that window; reconcileReadState needs it to survive an + // archived-priority upsert (see SlackReadAnchor's doc comment). + const store = makeStore({ + "read_anchor:D1:D1": { + newest: "1699999999.000000", + threaded: false, + at: 1000, + title: "Alice Example", + }, + }); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + vi.spyOn( + slack as unknown as { + dmCounterpartyUserId: (c: string) => Promise; + }, + "dmCounterpartyUserId" + ).mockResolvedValue("U2"); + + await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise; + }).buildConversationLink({ + channelId: "D1", + // No userInfos passed — assembleSlackDmLink omits `title` on this link. + messages: [ + { type: "message", ts: "1700000000.000001", user: "U2", text: "hi again" }, + ], + initialSync: false, + }); + + const anchor = store.map.get("read_anchor:D1:D1") as { + newest: string; + title?: string | null; + }; + expect(anchor.newest).toBe("1700000000.000001"); + expect(anchor.title).toBe("Alice Example"); }); it("leaves no anchor for a DM's initial sync — the link is already read", async () => { @@ -4038,7 +4099,14 @@ describe("reconcileReadState", () => { return { slack, store, saveLink, api, markNeedsReauth }; } - const anchor = (over: Partial<{ newest: string; threaded: boolean; at: number }> = {}) => ({ + const anchor = ( + over: Partial<{ + newest: string; + threaded: boolean; + at: number; + title: string | null; + }> = {} + ) => ({ newest: "1700000000.000001", threaded: false, at: NOW, @@ -4060,6 +4128,28 @@ describe("reconcileReadState", () => { expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); }); + it("sends the anchor's title on the reconcile upsert, so an archived-priority thread keeps its real title", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor({ title: "Real title" }) }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink.mock.calls[0][0].title).toBe("Real title"); + }); + + it("omits `title` entirely when the anchor has none — never worse than today", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink.mock.calls[0][0]).not.toHaveProperty("title"); + }); + it("never sends `created` on the reconcile upsert", async () => { const { slack, saveLink } = setup( { "read_anchor:C1:1700000000.000001": anchor() }, diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index d075a660..3a7593fa 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -957,6 +957,12 @@ export class Slack extends Connector { * matter why the caller fetched these messages, but recreating a "not yet * read" anchor is only valid when the caller is reporting the * conversation's actual current head. + * + * Also carries the link's `title` onto the anchor (see `SlackReadAnchor` + * for why `reconcileReadState` needs it). `assembleSlackDmLink` deliberately + * omits `title` when `users.info` was unavailable, so a link with no title + * here does not mean "no title exists" — it falls back to whatever title + * is already on the stored anchor rather than dropping it for that window. */ private async applyReadAnchor(opts: { channelId: string; @@ -964,7 +970,7 @@ export class Slack extends Connector { anchorId: string; messages: SlackMessage[]; direct: boolean; - link: { unread?: boolean }; + link: { unread?: boolean; title?: string | null }; /** Did a cursor say this link is read? */ read: boolean; advanceConversationHead: boolean; @@ -985,7 +991,12 @@ export class Slack extends Connector { return; } if (advanceConversationHead && link.unread !== false) { - const anchor = deriveReadAnchor(messages, { direct, at: Date.now() }); + let title = link.title; + if (!title) { + const existing = await this.get(anchorKey); + title = existing?.title ?? undefined; + } + const anchor = deriveReadAnchor(messages, { direct, at: Date.now(), title }); if (anchor) await this.set(anchorKey, anchor); } } @@ -2193,15 +2204,38 @@ export class Slack extends Connector { const threadTs = key.slice( `read_anchor:${conversationId}:`.length ); - // Minimal upsert. `created` is omitted deliberately: `saveLink` - // reads `unread === false` as the initial-sync signal and DROPS the - // save outright when the item's date predates the plan's sync - // history limit. Title/preview/notes are omitted so the upsert - // preserves whatever is stored rather than rewriting content. + // Minimal upsert. `created` and `schedules` are omitted + // deliberately: `saveLink` reads `unread === false` as the + // initial-sync signal and DROPS the save outright when the item's + // date predates the plan's sync history limit. `preview` and + // `notes` are omitted too, so the upsert preserves whatever + // content is stored rather than rewriting it — `preview`'s + // platform default derives from the notes, so with none supplied + // it resolves to null and COALESCE falls through to the stored + // value. + // + // `title` is the one field that must be RE-SENT, not omitted: + // `upsert_thread` takes an "archived" code path whenever the + // user's `thread_priority` row points at an archived priority + // (they archived the focus this thread was filed under) or is + // missing, and on that path the platform's title default is the + // literal string "Untitled" — never null — so an omitted title + // there always loses to it, destroying the thread's real title. + // Sending `title` also re-writes it on the normal, non-archived + // path, which is harmless: a channel thread's title is derived + // deterministically from its root message and a DM's is the + // counterparty name, so both are re-sent unchanged by every + // ordinary sync anyway — this isn't a new source of truth, just + // re-asserting the same value. Sourced from the anchor (carried + // there by `applyReadAnchor`, since this upsert has no access to + // the live Slack message) and omitted here when the anchor has + // none — never worse than today's behaviour. + // // `author: null` documents that this upsert is genuinely - // authorless — it only flips `unread`, never introduces content — - // and silences `saveLink`'s development-time unattributed-link - // warning that would otherwise fire on every reconciled link. + // authorless — it only flips `unread` (and now `title`), never + // introduces new content — and silences `saveLink`'s + // development-time unattributed-link warning that would otherwise + // fire on every reconciled link. await this.tools.integrations.saveLink({ // Reuse the connector's own source helper — a reconcile upsert // that guessed the key would create a second, empty thread @@ -2211,6 +2245,7 @@ export class Slack extends Connector { type: (await this.isKnownDMChannel(conversationId)) ? "dm" : "thread", unread: false, author: null, + ...(anchor.title ? { title: anchor.title } : {}), }); resolved.push(key); }