jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 22:11:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(typing): tolerate agent clock skew

Browse Source 
Keep event timestamps within the agent ordering domain while using desktop-observed time for completion retention and suppression.

Co-authored-by: Carl <acda9e433d19dcd0e6b6840f7f4b98f3a56f1fab98049d444c087019e6d36560@buzz.block.builderlab.xyz>
Co-authored-by: Mongo <5398c5fd039b963ce132b3e078e7c4af097dd997517bb5e14c2682fe68c25197@buzz.block.builderlab.xyz>
Co-authored-by: Princess Donut <0366ccd5ee09c2779a9d6bd6683daa17c16a508a51f6a7e7314018dab8fdc49b@buzz.block.builderlab.xyz>
Signed-off-by: Taylor Ho <taylorkmho@gmail.com>
This commit is contained in:
Taylor Ho
2026-08-23 15:37:33 -07:00
committed by Carl
parent f37c431b8c
commit 6898defb6c
3 changed files with 113 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
desktop/src/features/agents/activeAgentTurnsStore.test.mjs
View File

@@ -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, [

72
desktop/src/features/messages/useChannelTyping.test.mjs
View File

@@ -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 },
});
});

54
desktop/src/features/messages/useChannelTyping.ts
View File

@@ -19,13 +19,18 @@ export type TypingIndicatorEntry = {
};
type TypingEntry = {
createdAt: number;
expiresAt: number;
firstSeenAt: number;
pubkey: string;
threadHeadId: string | null;
};
type TypingState = Record<string, TypingEntry>;
type TypingCompletionWatermarks = Record<string, number>;
type TypingCompletionWatermark = {
createdAt: number;
observedAt: number;
};
type TypingCompletionWatermarks = Record<string, TypingCompletionWatermark>;
type TypingSuppressionDeadlines = Record<string, number>;
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<TypingState>({});
const normalizedCurrentPubkey = currentPubkey?.toLowerCase();
const typingSuppressUntilByPubkeyRef = useRef<Record<string, number>>({});
const latestMessageCreatedAtByPubkeyRef = useRef<Record<string, number>>({});
const latestMessageCreatedAtByPubkeyRef = useRef<TypingCompletionWatermarks>(
{},
);
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) {