diff --git a/packages/zulip/adapter-events.test.ts b/packages/zulip/adapter-events.test.ts index 1429553..4d76632 100644 --- a/packages/zulip/adapter-events.test.ts +++ b/packages/zulip/adapter-events.test.ts @@ -720,6 +720,102 @@ effectTest( { layer: TestContext.TestContext }, ) +// Zulip serves GET /users from a realm-wide cache that a concurrent user +// creation can leave without the new user until the next change to any user. +// GET /users?user_ids=… reads the database, so it still knows them. +const seedStaleUserList = ( + stub: StubHttpClient, + cached: ReadonlyArray, + byIds: ReadonlyArray, +): Effect.Effect => + stub.respondSequence('GET', '/api/v1/users', [ + { body: { result: 'success', members: cached } }, + { body: { result: 'success', members: byIds } }, + ]) + +effectTest( + 'inbox.events names a sender the cached user list has not caught up with', + () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedRegister(stub) + const pinnedBot = { + user_id: 777, + email: 'pinned-bot@example.com', + full_name: 'pinned-bot', + is_bot: true, + is_active: true, + role: 400, + } + yield* seedStaleUserList(stub, [HERMES, MAINTAINER], [pinnedBot]) + yield* stub.respondSequence('GET', '/api/v1/events', [ + { + body: { + result: 'success', + events: [ + messageEvent( + 5, + aZulipMessage({ + sender_id: pinnedBot.user_id, + sender_full_name: pinnedBot.full_name, + }), + ), + ], + }, + }, + { hang: true }, + ]) + const queue = yield* eventQueue(adapter) + const posted = yield* Queue.take(queue) + expect(posted.kind).toBe('message-posted') + if (posted.kind === 'message-posted') { + expect(posted.message.sender.kind).toBe('agent') + } + }), + { layer: TestContext.TestContext }, +) + +effectTest( + 'inbox.events wakes the seat on its own mention while the cached user list lacks it', + () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedRegister(stub) + yield* seedStaleUserList(stub, [MAINTAINER], [HERMES]) + yield* stub.respond('GET', '/api/v1/messages', { + body: { + result: 'success', + messages: [ + { + id: 300, + content: + '

@hermes-agent listen

', + }, + ], + }, + }) + yield* stub.respondSequence('GET', '/api/v1/events', [ + { + body: { + result: 'success', + events: [ + messageEvent(7, aZulipMessage({ id: 300, content: '@**hermes-agent** listen' })), + ], + }, + }, + { hang: true }, + ]) + const queue = yield* eventQueue(adapter) + const mention = yield* Queue.take(queue).pipe( + Effect.repeat({ until: (e: InboundEvent) => e.kind === 'mention-received' }), + ) + expect(mention.kind).toBe('mention-received') + }), + { layer: TestContext.TestContext }, +) + effectTest( 'inbox.events ignores Zulip heartbeat events but advances last_event_id', () => diff --git a/packages/zulip/adapter.test.ts b/packages/zulip/adapter.test.ts index f80bede..fef703b 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -778,6 +778,11 @@ const seedSendMessage = (stub: StubHttpClient, id: number): Effect.Effect }) }) +// A mention the directory cannot place is put to Zulip's renderer before the +// write is refused, so a test that expects a refusal seeds the rendering. +const seedRender = (stub: StubHttpClient, rendered: string): Effect.Effect => + stub.respond('POST', '/api/v1/messages/render', { body: { result: 'success', rendered } }) + effectTest( 'publisher.post sends type=channel + to=channel.name + content; defaults topic to "(no topic)" when no thread', () => @@ -1213,6 +1218,7 @@ effectTest('publisher.post fails with a tagged UnresolvedMention on a dead menti Effect.gen(function* () { const stub = yield* makeStubHttpClient yield* seedSendMessage(stub, 200) + yield* seedRender(stub, '

decision for @Alice Example to make

') const adapter = yield* buildAdapter(stub) // The directory holds HERMES only; @**Graeme Foster** resolves to nobody. const error = yield* Effect.flip( @@ -1250,6 +1256,10 @@ test.each([ Effect.gen(function* () { const stub = yield* makeStubHttpClient yield* seedSendMessage(stub, 203) + yield* seedRender( + stub, + '

cc-myproject-abcdef12 your answer

', + ) const adapter = yield* buildAdapter(stub) yield* seedUsers(stub, [ HERMES, @@ -1276,6 +1286,112 @@ test.each([ ), ) +// Zulip serves GET /users from a realm-wide cache, and a GET /users that lands +// while a user is being created re-caches the list without them until the next +// change to any user. Zulip's mention renderer and GET /users?user_ids=… read +// the database, so they still know the new user. +const PINNED = { + user_id: 777, + email: 'pinned-bot@example.com', + full_name: 'pinned-bot', + is_bot: true, + is_active: true, + role: 400, +} as const + +const pinnedIdentity: Identity = { + id: decodeIdentityIdSync('777'), + name: decodeDisplayNameSync('pinned-bot'), + kind: 'agent', +} + +const PINNED_SPAN = '@pinned-bot' + +const seedStaleUserList = ( + stub: StubHttpClient, + cached: ReadonlyArray, + byIds: ReadonlyArray, +): Effect.Effect => + stub.respondSequence('GET', '/api/v1/users', [ + { body: { result: 'success', members: cached } }, + { body: { result: 'success', members: byIds } }, + ]) + +const userListReads = (stub: StubHttpClient) => + stub.captured.pipe( + Effect.map((reqs) => + reqs.filter((r) => r.method === 'GET' && r.url.pathname === '/api/v1/users'), + ), + ) + +effectTest('identity.resolve finds a bot the cached user list has not caught up with', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* zulipAdapter(stub, yield* makeConfig()) + yield* seedStaleUserList(stub, [HERMES], [PINNED]) + yield* seedRender(stub, `

${PINNED_SPAN}

`) + const resolved = yield* adapter.identity.resolve(decodeDisplayNameSync('pinned-bot')) + expect(resolved).toEqual(Option.some(pinnedIdentity)) + const reads = yield* userListReads(stub) + expect(reads.at(-1)?.url.searchParams.get('user_ids')).toBe('[777]') + }), +) + +effectTest('identity.resolve answers none without a second read when the realm knows no one', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* zulipAdapter(stub, yield* makeConfig()) + yield* seedUsers(stub, [HERMES]) + yield* seedRender(stub, '

@nobody

') + const resolved = yield* adapter.identity.resolve(decodeDisplayNameSync('nobody')) + expect(resolved).toEqual(Option.none()) + expect(yield* userListReads(stub)).toHaveLength(1) + }), +) + +effectTest( + 'publisher.post delivers a mention of a bot the cached user list has not caught up with', + () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedSendMessage(stub, 204) + const adapter = yield* buildAdapter(stub) + yield* seedStaleUserList(stub, [HERMES], [PINNED]) + yield* seedRender(stub, `

${PINNED_SPAN} hello

`) + yield* adapter.publisher.post( + generalChannel.name, + decodeMessageBodySync('@**pinned-bot** hello'), + ) + const params = new URLSearchParams( + (yield* findRequest(stub, 'POST', '/api/v1/messages')).body, + ) + expect(params.get('content')).toBe('@**pinned-bot** hello') + }), +) + +effectTest('history.readChannel names a sender the cached user list has not caught up with', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedStaleUserList(stub, [HERMES], [PINNED]) + yield* seedMessages(stub, [ + { + id: 9001, + sender_id: PINNED.user_id, + sender_full_name: PINNED.full_name, + stream_id: 1234, + display_recipient: 'general', + subject: 'office test1', + content: 'listening', + timestamp: 1790860477, + reactions: [], + }, + ]) + const messages = yield* adapter.history.readChannel(generalChannel.name, { limit: 50 }) + expect(messages[0]?.sender).toEqual(pinnedIdentity) + }), +) + // A dead form written as an example inside a code span is literal text Zulip // never delivers as a mention — validating it would block agents from // discussing mention discipline. It must post verbatim, not fail. @@ -1314,6 +1430,7 @@ effectTest('publisher.edit fails with a tagged UnresolvedMention on a dead menti Effect.gen(function* () { const stub = yield* makeStubHttpClient yield* stub.respond('PATCH', '/api/v1/messages/42', { body: { result: 'success' } }) + yield* seedRender(stub, '

now pinging @Alice Example

') const adapter = yield* buildAdapter(stub) const target: MessageRef = { id: decodeMessageIdSync('42'), diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index 4f595c5..5e7a643 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -110,7 +110,7 @@ import { } from './mentions.ts' import { buildMessageRef, permalinkBase, withChannelPermalink } from './permalink.ts' import { type RenderedContentLookup, renderedContentForBatch } from './rendered-content.ts' -import { mentionsOfMessage } from './rendered-mentions.ts' +import { mentionedUserIds, mentionsOfMessage } from './rendered-mentions.ts' import { applyResolvedPrefix, splitTopic } from './resolved-topic.ts' import { senderNarrow, userPresencePath, ZulipUserRef } from './user-ref.ts' @@ -323,6 +323,11 @@ const sentMessageSchema = Schema.Struct({ const successSchema = Schema.Struct({ result: Schema.Literal('success') }) +const renderedSchema = Schema.Struct({ + result: Schema.Literal('success'), + rendered: Schema.String, +}) + /** * Classify a failed content edit. Zulip walls a content edit three ways — * realm editing disabled, edit-window expired, and not-the-original-sender — @@ -689,35 +694,63 @@ export const zulipAdapter = ( ZulipApiError | ParseResult.ParseError > => minterHttp.get('/users', usersSchema).pipe(Effect.map((res) => res.members)) + // The unfiltered GET /users above is served from Zulip's realm-wide user + // cache, and a read racing a user's creation re-caches the list without + // that user until the next change to any user. Filtering by `user_ids` + // reads the database instead. + const fetchMembersById = ( + userIds: ReadonlyArray, + ): Effect.Effect, ZulipApiError | ParseResult.ParseError> => + minterHttp + .get('/users', usersSchema, { user_ids: JSON.stringify(userIds) }) + .pipe(Effect.map((res) => res.members)) + interface DirectoryLookup { readonly byId: ReadonlyMap readonly byName: ReadonlyMap readonly byIdentityId: ReadonlyMap readonly active: ReadonlySet + readonly cover: ( + userIds: ReadonlyArray, + ) => Effect.Effect } + const directoryOf = ( + members: ReadonlyArray, + ): Effect.Effect => + Effect.gen(function* () { + const byId = new Map() + const byName = new Map() + const byIdentityId = new Map() + const active = new Set() + for (const u of members) { + const ident = yield* toIdentity(u) + byId.set(ZulipUserRef(u.user_id), ident) + byName.set(u.full_name, ident) + byIdentityId.set(ident.id, ZulipUserRef(u.user_id)) + if (u.is_active) active.add(ident.id) + } + const lookup: DirectoryLookup = { + byId, + byName, + byIdentityId, + active, + cover: (userIds) => { + const missing = Arr.dedupe(userIds.filter((id) => !byId.has(ZulipUserRef(id)))) + return Arr.isEmptyReadonlyArray(missing) + ? Effect.succeed(lookup) + : fetchMembersById(missing).pipe( + Effect.flatMap((found) => directoryOf([...members, ...found])), + ) + }, + } + return lookup + }) + const buildDirectoryLookup = (): Effect.Effect< DirectoryLookup, ZulipApiError | ParseResult.ParseError - > => - fetchMembers().pipe( - Effect.flatMap((members) => - Effect.gen(function* () { - const byId = new Map() - const byName = new Map() - const byIdentityId = new Map() - const active = new Set() - for (const u of members) { - const ident = yield* toIdentity(u) - byId.set(ZulipUserRef(u.user_id), ident) - byName.set(u.full_name, ident) - byIdentityId.set(ident.id, ZulipUserRef(u.user_id)) - if (u.is_active) active.add(ident.id) - } - return { byId, byName, byIdentityId, active } - }), - ), - ) + > => fetchMembers().pipe(Effect.flatMap(directoryOf)) // Recipient directory shape consumed by bot-dm-guard — narrower than // the full DirectoryLookup so the wrapping helper stays decoupled. @@ -733,6 +766,15 @@ export const zulipAdapter = ( })), ) + // Zulip's renderer resolves mention names against the database, so it + // still knows a user the cached GET /users list has lost. + const renderedMentionIds = ( + content: string, + ): Effect.Effect, ZulipApiError | ParseResult.ParseError> => + minterHttp + .post('/messages/render', renderedSchema, { content }) + .pipe(Effect.map((res) => mentionedUserIds(res.rendered))) + // A directory the shared mention helpers can resolve against — the // name-keyed map plus an id resolver for the disambiguated `@**Name|id**` // form, adapting this path's `ZulipUserRef`-keyed `byId` to a raw id. @@ -772,12 +814,22 @@ export const zulipAdapter = ( : buildDirectoryLookup().pipe( Effect.flatMap((directory) => { const dead = unresolvedMentions(body, notifiableMentionDirectory(directory)) - return dead.length === 0 + return Arr.isEmptyReadonlyArray(dead) + ? Effect.succeed(dead) + : renderedMentionIds(body).pipe( + Effect.flatMap(directory.cover), + Effect.map((covered) => + unresolvedMentions(body, notifiableMentionDirectory(covered)), + ), + ) + }), + Effect.flatMap((dead) => + Arr.isEmptyReadonlyArray(dead) ? Effect.void : Effect.fail( new UnresolvedMention({ operation, tokens: dead, substrate: 'zulip' }), - ) - }), + ), + ), ) const mapHistoricalReactions = ( @@ -829,7 +881,7 @@ export const zulipAdapter = ( mentions: yield* mentionsOfMessage( renderedFor, { id: Number(m.id), content: m.content }, - mentionDirectory(directory), + (userIds) => directory.cover(userIds).pipe(Effect.map(mentionDirectory)), ), reactions, } @@ -865,7 +917,13 @@ export const zulipAdapter = ( Effect.gen(function* () { const renderedFor = yield* renderedContentForBatch(minterHttp, query, rows) const historical = yield* Effect.forEach(rows, toHistoricalMessage) - const directory = yield* Fiber.join(directoryFiber) + const directory = yield* Fiber.join(directoryFiber).pipe( + Effect.flatMap((d) => + d.cover( + historical.flatMap((m) => [m.senderId, ...m.reactions.map((r) => r.userId)]), + ), + ), + ) return yield* Effect.forEach(historical, (m) => mapHistoricalMessage(m, directory, renderedFor), ) @@ -1155,18 +1213,33 @@ export const zulipAdapter = ( ).pipe(Effect.as([undefined, Option.none()] as const)), }), ).pipe(Effect.asVoid), - resolve: (name) => - minterHttp.get('/users', usersSchema).pipe( - Effect.flatMap((res) => - Arr.findFirst(res.members, (u) => u.is_active && u.full_name === name).pipe( - Option.match({ - onNone: () => Effect.succeed(Option.none()), - onSome: (match) => toIdentity(match).pipe(Effect.asSome), - }), - ), + resolve: (name) => { + const activeNamed = (members: ReadonlyArray) => + Arr.findFirst(members, (u) => u.is_active && u.full_name === name) + return fetchMembers().pipe( + Effect.flatMap((members) => + Option.match(activeNamed(members), { + onSome: (match) => Effect.succeedSome(match), + onNone: () => + renderedMentionIds(`@**${name}**`).pipe( + Effect.flatMap((userIds) => + Arr.isEmptyReadonlyArray(userIds) + ? Effect.succeed([]) + : fetchMembersById(userIds), + ), + Effect.map(activeNamed), + ), + }), + ), + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeed(Option.none()), + onSome: (match) => toIdentity(match).pipe(Effect.asSome), + }), ), Effect.mapError((cause) => new IdentityError({ operation: 'resolve', cause })), - ), + ) + }, } const directory: Directory = { @@ -2081,7 +2154,9 @@ export const zulipAdapter = ( onPage: (query, rows) => Effect.gen(function* () { const renderedFor = yield* renderedContentForBatch(minterHttp, query, rows) - const directory = yield* Fiber.join(directoryFiber) + const directory = yield* Fiber.join(directoryFiber).pipe( + Effect.flatMap((d) => d.cover(rows.map((row) => row.sender_id))), + ) return yield* Effect.forEach(rows, (raw) => { const { flags: _flags, ...message } = raw return messageToInboundEvents( diff --git a/packages/zulip/events.test.ts b/packages/zulip/events.test.ts index 7e5aa08..5d0cd8f 100644 --- a/packages/zulip/events.test.ts +++ b/packages/zulip/events.test.ts @@ -65,10 +65,14 @@ const MAINTAINER: Identity = { kind: 'human', } -const directoryFor = (...identities: ReadonlyArray): DirectoryLookup => ({ - byId: new Map(identities.map((i) => [Number(i.id), i])), - byName: new Map(identities.map((i) => [i.name, i])), -}) +const directoryFor = (...identities: ReadonlyArray): DirectoryLookup => { + const directory: DirectoryLookup = { + byId: new Map(identities.map((i) => [Number(i.id), i])), + byName: new Map(identities.map((i) => [i.name, i])), + cover: () => Effect.succeed(directory), + } + return directory +} const messageMentioning = (target: Identity, sender: Identity): ParsedZulipMessage => ({ id: 100, diff --git a/packages/zulip/events.ts b/packages/zulip/events.ts index 4a26ea5..c1e5e40 100644 --- a/packages/zulip/events.ts +++ b/packages/zulip/events.ts @@ -50,6 +50,7 @@ import { Stream, } from 'effect' import type { ZulipApiError, ZulipHttp } from './http.ts' +import type { MentionDirectory } from './mentions.ts' import { buildMessageRef } from './permalink.ts' import { type RenderedContentLookup, renderedContentPerMessage } from './rendered-content.ts' import { mentionsOfMessage } from './rendered-mentions.ts' @@ -58,8 +59,20 @@ import { splitTopic } from './resolved-topic.ts' export interface DirectoryLookup { readonly byId: ReadonlyMap readonly byName: ReadonlyMap + /** + * This directory extended with whichever of `userIds` it lacks and the realm + * holds. The full user list it was built from can lag the realm. + */ + readonly cover: ( + userIds: ReadonlyArray, + ) => Effect.Effect } +const mentionDirectoryOf = (directory: DirectoryLookup): MentionDirectory => ({ + byName: directory.byName, + byUserId: (userId) => directory.byId.get(userId), +}) + /** * LRU cache of MessageRefs keyed by Zulip message id. Reaction events * on Zulip's queue carry `message_id` but not stream/topic context — @@ -413,7 +426,8 @@ export const messageToInboundEvents = ( renderedFor: RenderedContentLookup, ): Effect.Effect, ZulipApiError | ParseResult.ParseError> => Effect.gen(function* () { - const sender = yield* decodeSenderIdentity(message, directory) + const covered = yield* directory.cover([message.sender_id]) + const sender = yield* decodeSenderIdentity(message, covered) const ref = yield* decodeMessageRef(message, base) const body = yield* decodeMessageBody(message.content) const ts = yield* decodeTimestamp(message.timestamp) @@ -422,10 +436,9 @@ export const messageToInboundEvents = ( sender, body, ts, - mentions: yield* mentionsOfMessage(renderedFor, message, { - byName: directory.byName, - byUserId: (userId) => directory.byId.get(userId), - }), + mentions: yield* mentionsOfMessage(renderedFor, message, (userIds) => + covered.cover(userIds).pipe(Effect.map(mentionDirectoryOf)), + ), // Reactions arrive as separate `reaction` events. A freshly-posted // message carries no reaction state — anything that exists already // (via history reads) surfaces through the HistoryReader path. @@ -511,9 +524,9 @@ export const reactionToInboundEvent = ( reaction: ParsedZulipReactionEvent, directory: DirectoryLookup, target: MessageRef, -): Effect.Effect => +): Effect.Effect => Effect.gen(function* () { - const by = yield* decodeReactionIdentity(reaction, directory) + const by = yield* decodeReactionIdentity(reaction, yield* directory.cover([reaction.user_id])) const emoji = yield* decodeEmoji(reaction.emoji_name) return { kind: reactionKind(reaction.op), @@ -707,7 +720,7 @@ const processSingleEvent = ( const reaction = parsedReaction.right const eventsForTarget = ( target: MessageRef, - ): Effect.Effect, ParseResult.ParseError> => + ): Effect.Effect, ZulipApiError | ParseResult.ParseError> => reactionToInboundEvent(reaction, directory, target).pipe(Effect.map((ev) => [ev])) const resolveTarget: Effect.Effect< ReadonlyArray, diff --git a/packages/zulip/rendered-mentions.ts b/packages/zulip/rendered-mentions.ts index 8aa1eab..c0c4dca 100644 --- a/packages/zulip/rendered-mentions.ts +++ b/packages/zulip/rendered-mentions.ts @@ -145,6 +145,18 @@ export const renderedMentions = ( Effect.map(Arr.dedupeWith((a, b) => Equal.equals(identityOf(a), identityOf(b)))), ) +/** The user ids Zulip rendered a notifying mention span for. */ +export const mentionedUserIds = (rendered: string): ReadonlyArray => + Arr.filterMap( + mentionSpans(rendered), + MentionSpan.$match({ + UserSpan: ({ userId }) => Option.some(userId), + GroupSpan: () => Option.none(), + ChannelWildcardSpan: () => Option.none(), + TopicWildcardSpan: () => Option.none(), + }), + ) + /** * Who a message mentions, for a caller that holds its raw content and can * reach its rendering. The raw body is the candidate filter and nothing more: @@ -152,18 +164,26 @@ export const renderedMentions = ( * mentioned. Because a render is a property of the message rather than of * whoever reads it, the answer is the same for the bound bot as for the * minter whose credentials fetched it. + * + * `directoryFor` is handed the rendered user ids so it can return a directory + * that names all of them. */ export const mentionsOfMessage = ( renderedFor: RenderedContentLookup, message: { readonly id: number; readonly content: string }, - directory: MentionDirectory, + directoryFor: ( + userIds: ReadonlyArray, + ) => Effect.Effect, ): Effect.Effect, ZulipApiError | ParseResult.ParseError> => mayMention(message.content) ? renderedFor(message.id).pipe( Effect.flatMap( Option.match({ onNone: () => Effect.succeed([] as ReadonlyArray), - onSome: (rendered) => renderedMentions(rendered, directory), + onSome: (rendered) => + directoryFor(mentionedUserIds(rendered)).pipe( + Effect.flatMap((directory) => renderedMentions(rendered, directory)), + ), }), ), )