Skip to content

Commit 7e3100a

Browse files
colmbrogancolmbrogan
authored
fix(imessage): persist echo markers before send (#88969)
Persist short-lived pending iMessage echo markers before bridge sends so self-chat reflected rows cannot race ahead of post-send echo persistence. Keep monitor cache writes post-send, keep pending text out of generic echo matching, and observe skipped from-me catchup rows for self-chat dedupe.\n\nThanks @colmbrogan.
1 parent 03a8d18 commit 7e3100a

13 files changed

Lines changed: 456 additions & 64 deletions

‎extensions/imessage/src/monitor/catchup-bridge.ts‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,12 @@ export type RunIMessageCatchupParams = {
5555
* (including non-error drops, which mirrors the live pipeline).
5656
*/
5757
dispatchPayload: (message: IMessagePayload) => Promise<void>;
58+
/**
59+
* Called for `is_from_me=true` rows that catchup intentionally does not
60+
* dispatch. The live inbound path still needs to observe those rows so
61+
* self-chat reflected companion rows can be deduped.
62+
*/
63+
observeSkippedFromMePayload?: (message: IMessagePayload) => Promise<void> | void;
5864
runtime?: RuntimeLogger;
5965
/** Override clock for tests. */
6066
now?: () => number;
@@ -277,6 +283,14 @@ export async function runIMessageCatchup(
277283
config,
278284
fetch: fetchFn,
279285
dispatch: dispatchFn,
286+
observeSkippedFromMe: async (row) => {
287+
const payload = payloadByGuid.get(row.guid);
288+
if (!payload) {
289+
warnLog(`imessage catchup: missing skipped from-me payload for guid=${row.guid}`);
290+
return;
291+
}
292+
await params.observeSkippedFromMePayload?.(payload);
293+
},
280294
log,
281295
warn: warnLog,
282296
...(params.now ? { now: params.now() } : {}),

‎extensions/imessage/src/monitor/catchup.test.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,7 @@ describe("performIMessageCatchup", () => {
261261

262262
it("skips is_from_me rows but still advances the cursor past them", async () => {
263263
const dispatch = alwaysOk();
264+
const observeSkippedFromMe = vi.fn();
264265
const fetch = fetchOf([
265266
row({ guid: "A", rowid: 10, isFromMe: true }),
266267
row({ guid: "B", rowid: 11, isFromMe: false }),
@@ -272,12 +273,16 @@ describe("performIMessageCatchup", () => {
272273
now,
273274
fetch,
274275
dispatch,
276+
observeSkippedFromMe,
275277
});
276278

277279
expect(summary.skippedFromMe).toBe(1);
278280
expect(summary.replayed).toBe(1);
279281
expect(summary.cursorAfter.lastSeenRowid).toBe(11);
280282
expect(dispatch).toHaveBeenCalledTimes(1);
283+
expect(observeSkippedFromMe).toHaveBeenCalledWith(
284+
expect.objectContaining({ guid: "A", rowid: 10, isFromMe: true }),
285+
);
281286
});
282287

283288
it("drops rows older than the maxAgeMinutes ceiling and advances past them", async () => {

‎extensions/imessage/src/monitor/catchup.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -326,6 +326,7 @@ export type PerformCatchupParams = {
326326
now?: number;
327327
fetch: CatchupFetchFn;
328328
dispatch: CatchupDispatchFn;
329+
observeSkippedFromMe?: (row: IMessageCatchupRow) => Promise<void> | void;
329330
log?: (message: string) => void;
330331
warn?: (message: string) => void;
331332
};
@@ -462,6 +463,13 @@ export async function performIMessageCatchup(
462463
continue;
463464
}
464465
if (row.isFromMe) {
466+
try {
467+
await params.observeSkippedFromMe?.(row);
468+
} catch (err) {
469+
params.warn?.(
470+
`imessage catchup: from-me observer failed for guid=${row.guid}: ${String(err)}`,
471+
);
472+
}
465473
summary.skippedFromMe += 1;
466474
highWatermarkMs = Math.max(highWatermarkMs, row.date);
467475
highWatermarkRowid = Math.max(highWatermarkRowid, row.rowid);

‎extensions/imessage/src/monitor/deliver.test.ts‎

Lines changed: 21 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -67,24 +67,24 @@ describe("deliverReplies", () => {
6767
[
6868
"chat_id:10",
6969
"first",
70-
{
70+
expect.objectContaining({
7171
config: IMESSAGE_TEST_CFG,
7272
maxBytes: 4096,
7373
client,
7474
accountId: "default",
7575
replyToId: "reply-1",
76-
},
76+
}),
7777
],
7878
[
7979
"chat_id:10",
8080
"second",
81-
{
81+
expect.objectContaining({
8282
config: IMESSAGE_TEST_CFG,
8383
maxBytes: 4096,
8484
client,
8585
accountId: "default",
8686
replyToId: "reply-1",
87-
},
87+
}),
8888
],
8989
]);
9090
});
@@ -112,26 +112,26 @@ describe("deliverReplies", () => {
112112
[
113113
"chat_id:20",
114114
"caption",
115-
{
115+
expect.objectContaining({
116116
config: IMESSAGE_TEST_CFG,
117117
mediaUrl: "https://example.com/a.jpg",
118118
maxBytes: 8192,
119119
client,
120120
accountId: "acct-2",
121121
replyToId: "reply-2",
122-
},
122+
}),
123123
],
124124
[
125125
"chat_id:20",
126126
"",
127-
{
127+
expect.objectContaining({
128128
config: IMESSAGE_TEST_CFG,
129129
mediaUrl: "https://example.com/b.jpg",
130130
maxBytes: 8192,
131131
client,
132132
accountId: "acct-2",
133133
replyToId: "reply-2",
134-
},
134+
}),
135135
],
136136
]);
137137
});
@@ -157,11 +157,11 @@ describe("deliverReplies", () => {
157157
[
158158
"chat_id:50",
159159
"durable hello",
160-
{
160+
expect.objectContaining({
161161
config: IMESSAGE_TEST_CFG,
162162
accountId: "acct-ignored",
163163
client,
164-
},
164+
}),
165165
],
166166
]);
167167
expect(remember).toHaveBeenCalledWith("acct-5:chat_id:50", {
@@ -191,11 +191,11 @@ describe("deliverReplies", () => {
191191
[
192192
"chat_id:60",
193193
"Visible reply",
194-
{
194+
expect.objectContaining({
195195
config: IMESSAGE_TEST_CFG,
196196
accountId: "acct-ignored",
197197
client,
198-
},
198+
}),
199199
],
200200
]);
201201
expect(remember).toHaveBeenCalledWith("acct-6:chat_id:60", {
@@ -204,10 +204,9 @@ describe("deliverReplies", () => {
204204
});
205205
});
206206

207-
it("records outbound text and message ids in sent-message cache (post-send only)", async () => {
208-
// Fix for #47830: remember() is called ONLY after each chunk is sent,
209-
// never with the full un-chunked text before sending begins.
210-
// Pre-send population widened the false-positive window in self-chat.
207+
it("records outbound text and message ids in sent-message cache after send", async () => {
208+
// Fix for #47830: monitor cache population remains per chunk, never the
209+
// full un-chunked text before sending begins.
211210
const remember = vi.fn();
212211
chunkTextWithModeMock.mockImplementation((text: string) => text.split("|"));
213212
sendMessageIMessageMock
@@ -226,15 +225,13 @@ describe("deliverReplies", () => {
226225
sentMessageCache: { remember },
227226
});
228227

229-
// Only the two per-chunk post-send calls — no pre-send full-text call.
230228
expect(remember).toHaveBeenCalledTimes(2);
231-
expect(remember).toHaveBeenCalledWith("acct-3:chat_id:30", {
232-
text: "first",
233-
messageId: "imsg-1",
234-
});
235-
expect(remember).toHaveBeenCalledWith("acct-3:chat_id:30", {
236-
text: "second",
237-
messageId: "imsg-2",
229+
expect(remember.mock.calls).toStrictEqual([
230+
["acct-3:chat_id:30", { text: "first", messageId: "imsg-1" }],
231+
["acct-3:chat_id:30", { text: "second", messageId: "imsg-2" }],
232+
]);
233+
expect(remember).not.toHaveBeenCalledWith("acct-3:chat_id:30", {
234+
text: "first|second",
238235
});
239236
});
240237

‎extensions/imessage/src/monitor/deliver.ts‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -55,10 +55,6 @@ export async function deliverReplies(params: {
5555
accountId,
5656
replyToId: payload.replyToId,
5757
});
58-
// Post-send cache population (#47830): caching happens after each chunk is sent,
59-
// not before. The window between send completion and cache write is sub-millisecond;
60-
// the next SQLite inbound poll is 1-2s away, so no echo can arrive before the
61-
// cache entry exists.
6258
sentMessageCache?.remember(scope, {
6359
text: sent.echoText ?? sent.sentText,
6460
messageId: sent.messageId,

‎extensions/imessage/src/monitor/echo-cache.ts‎

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,11 @@ type SentMessageLookup = {
66
messageId?: string;
77
};
88

9+
type SentMessageLookupOptions = {
10+
skipIdShortCircuit?: boolean;
11+
includePendingText?: boolean;
12+
};
13+
914
export type SentMessageCache = {
1015
remember: (scope: string, lookup: SentMessageLookup) => void;
1116
/**
@@ -17,7 +22,11 @@ export type SentMessageCache = {
1722
* that will never match the GUID outbound IDs, but text matching is still
1823
* the right way to identify agent reply echoes.
1924
*/
20-
has: (scope: string, lookup: SentMessageLookup, skipIdShortCircuit?: boolean) => boolean;
25+
has: (
26+
scope: string,
27+
lookup: SentMessageLookup,
28+
options?: boolean | SentMessageLookupOptions,
29+
) => boolean;
2130
};
2231

2332
// Echo arrival observed at ~2.2s on M4 Mac Mini (SQLite poll interval is the bottleneck).
@@ -70,9 +79,21 @@ class DefaultSentMessageCache implements SentMessageCache {
7079
this.cleanup();
7180
}
7281

73-
has(scope: string, lookup: SentMessageLookup, skipIdShortCircuit = false): boolean {
82+
has(
83+
scope: string,
84+
lookup: SentMessageLookup,
85+
options: boolean | SentMessageLookupOptions = false,
86+
): boolean {
7487
this.cleanup();
75-
if (hasPersistedIMessageEcho({ scope, ...lookup })) {
88+
const resolvedOptions =
89+
typeof options === "boolean" ? { skipIdShortCircuit: options } : options;
90+
if (
91+
hasPersistedIMessageEcho({
92+
scope,
93+
...lookup,
94+
includePendingText: resolvedOptions.includePendingText,
95+
})
96+
) {
7697
return true;
7798
}
7899
const textKey = normalizeEchoTextKey(lookup.text);
@@ -89,7 +110,7 @@ class DefaultSentMessageCache implements SentMessageCache {
89110
const hasTextOnlyMatch =
90111
typeof textTimestamp === "number" &&
91112
(!textBackedByIdTimestamp || textTimestamp > textBackedByIdTimestamp);
92-
if (!skipIdShortCircuit && !hasTextOnlyMatch) {
113+
if (!resolvedOptions.skipIdShortCircuit && !hasTextOnlyMatch) {
93114
return false;
94115
}
95116
}

‎extensions/imessage/src/monitor/inbound-processing.ts‎

Lines changed: 57 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -193,18 +193,67 @@ function resolveInboundEchoMessageIds(message: IMessagePayload): string[] {
193193
return ids;
194194
}
195195

196+
export function rememberIMessageSkippedFromMeForSelfChatDedupe(params: {
197+
accountId: string;
198+
message: IMessagePayload;
199+
bodyText: string;
200+
selfChatCache?: SelfChatCache;
201+
}): void {
202+
if (params.message.is_from_me !== true) {
203+
return;
204+
}
205+
const sender = params.message.sender?.trim();
206+
if (!sender) {
207+
return;
208+
}
209+
const chatId = params.message.chat_id ?? undefined;
210+
const isGroup = Boolean(params.message.is_group);
211+
const chatIdentifierNormalized =
212+
normalizeIMessageHandle(params.message.chat_identifier ?? "") || undefined;
213+
const destinationCallerIdNormalized =
214+
normalizeIMessageHandle(params.message.destination_caller_id ?? "") || undefined;
215+
const senderNormalized = normalizeIMessageHandle(sender);
216+
const createdAt = params.message.created_at ? Date.parse(params.message.created_at) : undefined;
217+
const lookup = {
218+
accountId: params.accountId,
219+
isGroup,
220+
chatId,
221+
sender,
222+
text: params.bodyText.trim(),
223+
createdAt,
224+
};
225+
const matchesSelfChatDestination =
226+
destinationCallerIdNormalized != null && destinationCallerIdNormalized === senderNormalized;
227+
const isSelfChat =
228+
!isGroup &&
229+
chatIdentifierNormalized != null &&
230+
senderNormalized === chatIdentifierNormalized &&
231+
matchesSelfChatDestination;
232+
const isAmbiguousSelfThread =
233+
!isGroup &&
234+
chatIdentifierNormalized != null &&
235+
senderNormalized === chatIdentifierNormalized &&
236+
destinationCallerIdNormalized == null;
237+
if (isSelfChat) {
238+
params.selfChatCache?.remember({ ...lookup, allowCreatedAtSkew: true });
239+
} else if (isAmbiguousSelfThread) {
240+
params.selfChatCache?.remember(lookup);
241+
}
242+
}
243+
196244
function hasIMessageEchoMatch(params: {
197245
echoCache: {
198246
has: (
199247
scope: string,
200248
lookup: { text?: string; messageId?: string },
201-
skipIdShortCircuit?: boolean,
249+
options?: boolean | { skipIdShortCircuit?: boolean; includePendingText?: boolean },
202250
) => boolean;
203251
};
204252
scope: string | readonly string[];
205253
text?: string;
206254
messageIds: string[];
207255
skipIdShortCircuit?: boolean;
256+
includePendingText?: boolean;
208257
}): boolean {
209258
// Outbound sends persist echo scopes keyed by whichever target shape was
210259
// used (chat_id, chat_guid, chat_identifier, or imessage:<handle>). Inbound
@@ -232,7 +281,10 @@ function hasIMessageEchoMatch(params: {
232281
params.echoCache.has(
233282
scope,
234283
{ text: params.text, messageId: fallbackMessageId },
235-
params.skipIdShortCircuit,
284+
{
285+
skipIdShortCircuit: params.skipIdShortCircuit,
286+
includePendingText: params.includePendingText,
287+
},
236288
)
237289
) {
238290
return true;
@@ -349,7 +401,7 @@ export async function resolveIMessageInboundDecision(params: {
349401
has: (
350402
scope: string,
351403
lookup: { text?: string; messageId?: string },
352-
skipIdShortCircuit?: boolean,
404+
options?: boolean | { skipIdShortCircuit?: boolean; includePendingText?: boolean },
353405
) => boolean;
354406
};
355407
selfChatCache?: SelfChatCache;
@@ -453,6 +505,7 @@ export async function resolveIMessageInboundDecision(params: {
453505
text: bodyText || undefined,
454506
messageIds: inboundMessageIds,
455507
skipIdShortCircuit: !hasInboundGuid,
508+
includePendingText: true,
456509
})
457510
) {
458511
return { kind: "drop", reason: "agent echo in self-chat" };
@@ -649,6 +702,7 @@ export async function resolveIMessageInboundDecision(params: {
649702
scope: echoScope,
650703
text: bodyText || undefined,
651704
messageIds: inboundMessageIds,
705+
includePendingText: isSelfChat,
652706
})
653707
) {
654708
params.logVerbose?.(

‎extensions/imessage/src/monitor/monitor-provider.echo-cache.test.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,23 @@ describe("iMessage sent-message echo cache", () => {
117117
expect(cache.has(scope, { messageId: "id-only" })).toBe(true);
118118
});
119119

120+
it("keeps short-lived pending persisted echoes out of generic text matching", () => {
121+
vi.useFakeTimers();
122+
vi.setSystemTime(new Date("2026-02-25T00:00:00Z"));
123+
const scope = "acct:imessage:+1555";
124+
125+
rememberPersistedIMessageEcho({ scope, text: "pending-send", ttlMs: 1_000, pending: true });
126+
expect(hasPersistedIMessageEcho({ scope, text: "pending-send" })).toBe(false);
127+
expect(
128+
hasPersistedIMessageEcho({ scope, text: "pending-send", includePendingText: true }),
129+
).toBe(true);
130+
131+
vi.advanceTimersByTime(1_001);
132+
expect(
133+
hasPersistedIMessageEcho({ scope, text: "pending-send", includePendingText: true }),
134+
).toBe(false);
135+
});
136+
120137
it("refreshes persisted echoes written after an earlier empty lookup", () => {
121138
const cache = createSentMessageCache();
122139
const scope = "acct:imessage:+1555";

0 commit comments

Comments
 (0)