diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs index 2658469b91..f67f3791a8 100644 --- a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs +++ b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs @@ -54,6 +54,20 @@ describe("activeAgentTurnsStore", () => { resetActiveAgentTurnsStore(); }); + describe("active turn summaries", () => { + it("retains every concurrent turn id in one channel deterministically", () => { + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ seq: 1, turnId: "turn-z", channelId: "c1" }), + makeEvent({ seq: 2, turnId: "turn-a", channelId: "c1" }), + ]); + + assert.deepEqual(getActiveTurnsForAgent(AGENT)[0].turnIds, [ + "turn-a", + "turn-z", + ]); + }); + }); + describe("seq filtering", () => { it("processes events with increasing seq", () => { syncAgentTurnsFromEvents(AGENT, [ diff --git a/desktop/src/features/messages/useChannelTyping.test.mjs b/desktop/src/features/messages/useChannelTyping.test.mjs index cf357ea5f0..d028d59e37 100644 --- a/desktop/src/features/messages/useChannelTyping.test.mjs +++ b/desktop/src/features/messages/useChannelTyping.test.mjs @@ -14,12 +14,17 @@ function makeState() { }; } -function record(state, createdAt) { +function record( + state, + createdAt, + { hasActiveTyping = true, now = NOW, typingCreatedAt = createdAt } = {}, +) { return recordTypingCompletion({ createdAt, latestMessageCreatedAtByPubkey: state.latestMessageCreatedAtByPubkey, - now: NOW, + now, suppressUntilByPubkey: state.suppressUntilByPubkey, + typingCreatedAt: hasActiveTyping ? typingCreatedAt : undefined, typingKey: TYPING_KEY, }); } @@ -30,7 +35,7 @@ test("duplicate full-array replay does not reprocess completion events", () => { assert.equal(record(state, NOW_SECONDS), true); assert.equal(record(state, NOW_SECONDS), false); assert.deepEqual(state.latestMessageCreatedAtByPubkey, { - [TYPING_KEY]: NOW_SECONDS, + [TYPING_KEY]: { createdAt: NOW_SECONDS, observedAt: NOW }, }); assert.deepEqual(state.suppressUntilByPubkey, { [TYPING_KEY]: NOW + 2_000, @@ -42,7 +47,10 @@ test("newer-then-older completion arrival keeps the newer watermark", () => { assert.equal(record(state, NOW_SECONDS), true); assert.equal(record(state, NOW_SECONDS - 1), false); - assert.equal(state.latestMessageCreatedAtByPubkey[TYPING_KEY], NOW_SECONDS); + assert.equal( + state.latestMessageCreatedAtByPubkey[TYPING_KEY].createdAt, + NOW_SECONDS, + ); assert.equal(state.suppressUntilByPubkey[TYPING_KEY], NOW + 2_000); }); @@ -51,14 +59,62 @@ test("older-then-newer completion arrival advances the watermark", () => { assert.equal(record(state, NOW_SECONDS - 1), true); assert.equal(record(state, NOW_SECONDS), true); - assert.equal(state.latestMessageCreatedAtByPubkey[TYPING_KEY], NOW_SECONDS); + assert.equal( + state.latestMessageCreatedAtByPubkey[TYPING_KEY].createdAt, + NOW_SECONDS, + ); assert.equal(state.suppressUntilByPubkey[TYPING_KEY], NOW + 2_000); }); -test("historical thread load creates neither a watermark nor fresh suppression", () => { +test("historical thread load records ordering but creates no fresh suppression", () => { const state = makeState(); - assert.equal(record(state, NOW_SECONDS - 60), false); - assert.deepEqual(state.latestMessageCreatedAtByPubkey, {}); + assert.equal( + record(state, NOW_SECONDS - 60, { hasActiveTyping: false }), + false, + ); + assert.deepEqual(state.latestMessageCreatedAtByPubkey, { + [TYPING_KEY]: { createdAt: NOW_SECONDS - 60, observedAt: NOW }, + }); assert.deepEqual(state.suppressUntilByPubkey, {}); }); + +test("lagging agent completion clears typing for a matching agent-clock event", () => { + const state = makeState(); + const laggingCreatedAt = NOW_SECONDS - 60; + + assert.equal(record(state, laggingCreatedAt), true); + assert.deepEqual(state.suppressUntilByPubkey, { + [TYPING_KEY]: NOW + 2_000, + }); +}); + +test("ahead agent completion keeps suppression desktop-local", () => { + const state = makeState(); + const aheadCreatedAt = NOW_SECONDS + 60; + + assert.equal(record(state, aheadCreatedAt), true); + assert.deepEqual(state.suppressUntilByPubkey, { + [TYPING_KEY]: NOW + 2_000, + }); +}); + +test("message older than active typing advances ordering without clearing", () => { + const state = makeState(); + + assert.equal( + record(state, NOW_SECONDS - 60, { typingCreatedAt: NOW_SECONDS }), + false, + ); + assert.deepEqual(state.suppressUntilByPubkey, {}); +}); + +test("watermarks expire by desktop observation time, not agent clock", () => { + const state = makeState(); + record(state, NOW_SECONDS + 60); + + assert.equal(record(state, NOW_SECONDS + 61, { now: NOW + 8_000 }), true); + assert.deepEqual(state.latestMessageCreatedAtByPubkey, { + [TYPING_KEY]: { createdAt: NOW_SECONDS + 61, observedAt: NOW + 8_000 }, + }); +}); diff --git a/desktop/src/features/messages/useChannelTyping.ts b/desktop/src/features/messages/useChannelTyping.ts index 52d8397dd9..9a02ed9afc 100644 --- a/desktop/src/features/messages/useChannelTyping.ts +++ b/desktop/src/features/messages/useChannelTyping.ts @@ -19,13 +19,18 @@ export type TypingIndicatorEntry = { }; type TypingEntry = { + createdAt: number; expiresAt: number; firstSeenAt: number; pubkey: string; threadHeadId: string | null; }; type TypingState = Record; -type TypingCompletionWatermarks = Record; +type TypingCompletionWatermark = { + createdAt: number; + observedAt: number; +}; +type TypingCompletionWatermarks = Record; type TypingSuppressionDeadlines = Record; const TYPING_INDICATOR_TTL_MS = 8_000; @@ -33,27 +38,33 @@ const TYPING_PRUNE_INTERVAL_MS = 1_000; const TYPING_POST_MESSAGE_SUPPRESS_MS = 2_000; /** - * Advance one author + thread completion watermark when it can still affect - * an acceptable typing indicator. Returns whether matching typing state should - * be cleared. Exported so replay and out-of-order behavior stays unit-tested. + * Advance one author + thread completion watermark and start a desktop-local + * suppression window. Event timestamps are used only for ordering within the + * agent's clock domain; retention and suppression use the observing desktop's + * clock so host skew cannot disable or stretch typing cleanup. + * + * Returns whether matching typing state should be cleared. Exported so replay, + * out-of-order, and clock-skew behavior stays unit-tested. */ export function recordTypingCompletion({ createdAt, latestMessageCreatedAtByPubkey, now = Date.now(), suppressUntilByPubkey, + typingCreatedAt, typingKey, }: { createdAt: number; latestMessageCreatedAtByPubkey: TypingCompletionWatermarks; now?: number; suppressUntilByPubkey: TypingSuppressionDeadlines; + typingCreatedAt?: number; typingKey: string; }) { for (const [key, watermark] of Object.entries( latestMessageCreatedAtByPubkey, )) { - if (watermark * 1_000 + TYPING_INDICATOR_TTL_MS <= now) { + if (watermark.observedAt + TYPING_INDICATOR_TTL_MS <= now) { delete latestMessageCreatedAtByPubkey[key]; } } @@ -63,22 +74,16 @@ export function recordTypingCompletion({ } } - if (createdAt * 1_000 + TYPING_INDICATOR_TTL_MS <= now) { - return false; - } - - const latestMessageCreatedAt = latestMessageCreatedAtByPubkey[typingKey] ?? 0; + const latestMessageCreatedAt = + latestMessageCreatedAtByPubkey[typingKey]?.createdAt ?? 0; if (createdAt <= latestMessageCreatedAt) { return false; } - latestMessageCreatedAtByPubkey[typingKey] = createdAt; - - const suppressUntil = createdAt * 1_000 + TYPING_POST_MESSAGE_SUPPRESS_MS; - if (suppressUntil > now) { - suppressUntilByPubkey[typingKey] = suppressUntil; - } else { - delete suppressUntilByPubkey[typingKey]; + latestMessageCreatedAtByPubkey[typingKey] = { createdAt, observedAt: now }; + if (typingCreatedAt === undefined || createdAt < typingCreatedAt) { + return false; } + suppressUntilByPubkey[typingKey] = now + TYPING_POST_MESSAGE_SUPPRESS_MS; return true; } @@ -129,7 +134,9 @@ export function useChannelTyping( const [typingByPubkey, setTypingByPubkey] = useState({}); const normalizedCurrentPubkey = currentPubkey?.toLowerCase(); const typingSuppressUntilByPubkeyRef = useRef>({}); - const latestMessageCreatedAtByPubkeyRef = useRef>({}); + const latestMessageCreatedAtByPubkeyRef = useRef( + {}, + ); const registerTyping = useEffectEvent((event: RelayEvent) => { if (!channelId || event.kind !== KIND_TYPING_INDICATOR) { @@ -162,8 +169,15 @@ export function useChannelTyping( delete typingSuppressUntilByPubkeyRef.current[typingKey]; } + const watermark = latestMessageCreatedAtByPubkeyRef.current[typingKey]; + if ( + watermark?.observedAt && + watermark.observedAt + TYPING_INDICATOR_TTL_MS <= now + ) { + delete latestMessageCreatedAtByPubkeyRef.current[typingKey]; + } const latestMessageCreatedAt = - latestMessageCreatedAtByPubkeyRef.current[typingKey] ?? 0; + latestMessageCreatedAtByPubkeyRef.current[typingKey]?.createdAt ?? 0; if (event.created_at <= latestMessageCreatedAt) { return; } @@ -174,6 +188,7 @@ export function useChannelTyping( return { ...pruned, [typingKey]: { + createdAt: event.created_at, expiresAt: Math.min(now + TYPING_INDICATOR_TTL_MS, eventExpiresAt), firstSeenAt: existing?.firstSeenAt ?? now, pubkey: typingPubkey, @@ -211,6 +226,7 @@ export function useChannelTyping( createdAt: event.created_at, latestMessageCreatedAtByPubkey: latestMessageCreatedAtByPubkeyRef.current, suppressUntilByPubkey: typingSuppressUntilByPubkeyRef.current, + typingCreatedAt: typingByPubkey[typingKey]?.createdAt, typingKey, }); if (!shouldClearTyping) {