From 50d3de5196388e5f54411180895db97486cce572 Mon Sep 17 00:00:00 2001 From: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> Date: Fri, 11 Sep 2026 21:59:00 -0700 Subject: [PATCH 1/7] Add shared receive-only channel and thread typing Signed-off-by: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> --- docs/relay-queries.md | 52 ++++++ src/features/messages/MessageComposer.tsx | 13 ++ src/features/messages/Messages.module.css | 8 + src/features/messages/TypingIndicator.tsx | 40 +++++ src/features/relay/live.test.ts | 30 ++++ src/features/relay/live.ts | 15 +- src/features/relay/session.ts | 24 ++- src/features/relay/typing.integration.test.ts | 106 +++++++++++ src/features/relay/typing.test.ts | 156 ++++++++++++++++ src/features/relay/typing.ts | 170 ++++++++++++++++++ tests/browser/fixture.mjs | 21 +++ tests/browser/typing.spec.mjs | 54 ++++++ 12 files changed, 687 insertions(+), 2 deletions(-) create mode 100644 src/features/messages/TypingIndicator.tsx create mode 100644 src/features/relay/typing.integration.test.ts create mode 100644 src/features/relay/typing.test.ts create mode 100644 src/features/relay/typing.ts create mode 100644 tests/browser/typing.spec.mjs diff --git a/docs/relay-queries.md b/docs/relay-queries.md index 62fac93a..0014a8f0 100644 --- a/docs/relay-queries.md +++ b/docs/relay-queries.md @@ -472,3 +472,55 @@ Exhausted attempts, unsupported pauses, other route/connection failures and unfinished finite roster/head failures still show a warning and recovery action. This presentation policy does not increase quotas or guarantee that another client sharing the account cannot cause a refusal. + +## Receive-only typing + +`session.typing.snapshot()` / `.subscribe()` expose immutable active entries +`{ channelId, threadRootId?, pubkey }`. The session owns one ephemeral projection; +shared `features/messages/TypingIndicator` renders it inside channel/thread +composers, including read-only connections. Page plugins consume the same UI and +session. Mounting consumers starts no reads, profile enrichment or subscriptions. +Names reuse the shared profile snapshot, with a public-key fragment fallback. +This reports signed typing activity (including agents), **not** inferred agent +execution, online presence, or a promise that an answer is coming. + +Kind 20002 joins the existing authenticated channel route, with the same signature +verification and socket/generation fencing. Typing must match that exact route, +name exactly one bounded `h`, belong to the current visible roster and pass the +existing channel access policy. Identity is the event signer, matching message +folding; arbitrary `p`/actor tags and profile display names do not confer identity. +A channel pulse has no `e`; a thread pulse has one marked canonical `reply`, with +at most one marked `root` for nested replies. Thread scope uses the existing +canonical root convention; it is not an extra target lookup or an access grant. +Malformed/ambiguous references are dropped rather than displayed at channel level. + +Only live pulses can activate typing. No typing payload enters finite views, +recent history, channel caches, unread counts, the outbox or persistent storage. +Expiry is eight seconds from the signed timestamp. Already-expired and future +activity is rejected (no clock-skew allowance). Duplicates/older pulses cannot +extend expiry. Signed content messages (9/40002) suppress their signer's same +channel/thread activity; timestamp watermarks reject older pulses and a two-second +post-message quiet period covers late activity. Old history does not clear newer +activity or repeatedly extend suppression. This does not reinterpret edits as +new messages or resolve relay-proxied author envelopes beyond current host rules. + +At most 1,024 active/suppression records and one timeout exist per session. At +capacity new identities/scopes are dropped until expiry; suppression evidence is +never evicted to admit a stale pulse. Access loss clears all typing conservatively, +before batched access notifications. Disconnect, cache clear, session disposal and +account/community replacement clear it too. Session disposal fences late callbacks; +plugins unloading only remove their UI subscriptions. There is no typing publisher, +new signer, transport, polling, durable cache, or plugin-owned ephemeral owner. + +Protocol reference: `block/buzz`'s +`crates/buzz-acp/src/relay.rs::build_typing_event` and +`desktop/src/features/messages/useChannelTyping.ts` (8-second TTL and 2-second +post-message suppression). Regression owners are `typing*.test.ts`, `live.test.ts` +and `tests/browser/typing.spec.mjs`; browser simulations use ephemeral fixture +keys through the production authenticated broker, never the deployed relay. + +**Review-sensitive:** `session.ts` (FOUNDATION admission/lifecycle and synchronous +subscriber reentrancy), `live.ts` (existing route filter/verification boundary), +`typing.ts` (timestamp, scope, bounds and suppression), and shared composer placement. +An existing development broker imports live routing at startup and needs one +coordinated restart to receive kind 20002; frontend hot reload alone is insufficient. diff --git a/src/features/messages/MessageComposer.tsx b/src/features/messages/MessageComposer.tsx index 9639d419..a9aab60f 100644 --- a/src/features/messages/MessageComposer.tsx +++ b/src/features/messages/MessageComposer.tsx @@ -1,3 +1,4 @@ +import { TypingIndicator } from "./TypingIndicator"; import { ArrowUp, X } from "lucide-react"; import { useEffect, @@ -297,6 +298,11 @@ function Composer({ if (!outbox?.supports(9)) return ( ); @@ -311,6 +317,13 @@ function Composer({ send(); }} > + {!disabled && ( + + )} diff --git a/src/features/messages/Messages.module.css b/src/features/messages/Messages.module.css index 87a1faeb..1379183d 100644 --- a/src/features/messages/Messages.module.css +++ b/src/features/messages/Messages.module.css @@ -710,3 +710,11 @@ button.avatar:focus-visible, height: 100%; object-fit: cover; } + +.typing { + color: var(--text-muted); + font-size: calc(12px * var(--buzz-text-scale, 1)); + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} diff --git a/src/features/messages/TypingIndicator.tsx b/src/features/messages/TypingIndicator.tsx new file mode 100644 index 00000000..59d05f51 --- /dev/null +++ b/src/features/messages/TypingIndicator.tsx @@ -0,0 +1,40 @@ +import { useSyncExternalStore } from "react"; +import type { RelaySession } from "../relay/session"; +import styles from "./Messages.module.css"; + +/** Shared presentation only. Mounting more consumers creates no relay work. */ +export function TypingIndicator({ + session, + channelId, + threadRootId, +}: { + session: RelaySession; + channelId: string; + threadRootId?: string | undefined; +}) { + const entries = useSyncExternalStore( + session.typing.subscribe, + session.typing.snapshot, + ); + const profiles = useSyncExternalStore( + session.profiles.subscribe, + session.profiles.snapshot, + ); + const matching = entries.filter( + (entry) => + entry.channelId === channelId && entry.threadRootId === threadRootId, + ); + if (!matching.length) return null; + // Reuse already available names; optional typing must not trigger profile reads. + const names = matching + .slice(0, 3) + .map(({ pubkey }) => profiles.get(pubkey)?.name ?? pubkey.slice(0, 10)); + const others = matching.length - names.length; + return ( +
+ {names.join(", ")} + {others > 0 ? ` and ${others} others` : ""} + {matching.length === 1 ? " is typing…" : " are typing…"} +
+ ); +} diff --git a/src/features/relay/live.test.ts b/src/features/relay/live.test.ts index 72d484ac..4821b392 100644 --- a/src/features/relay/live.test.ts +++ b/src/features/relay/live.test.ts @@ -696,3 +696,33 @@ it("observer route is optional, live-only at dispatch/retry, separately fenced a h.owner.dispose(); expect(vi.getTimerCount()).toBe(0); }); + +it("admits signed typing only on its authenticated channel route, without extra subscriptions", async () => { + vi.useFakeTimers(); + const h = setup(); + await h.first.auth(); + await vi.advanceTimersByTimeAsync(750); + const requests = h.first.requests(); + expect(requests).toHaveLength(4); + const route = requests[2]; + assert.exists(route); + expect(route[2].kinds).toContain(20002); + const event = signed(keypair(), { + kind: 20002, + content: "", + tags: [["h", "a"]], + }); + await h.first.receive(["EVENT", requests[0]?.[1], event]); + await h.first.receive(["EVENT", requests[3]?.[1], event]); + expect(h.callbacks.receive).not.toHaveBeenCalled(); + await h.first.receive(["EVENT", route[1], event]); + expect(h.callbacks.receive).toHaveBeenCalledExactlyOnceWith([event]); + await h.first.receive([ + "EVENT", + route[1], + { ...event, sig: "0".repeat(128) }, + ]); + expect(h.callbacks.receive).toHaveBeenCalledTimes(1); + h.owner.dispose(); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/src/features/relay/live.ts b/src/features/relay/live.ts index 8febfae2..d14f1c32 100644 --- a/src/features/relay/live.ts +++ b/src/features/relay/live.ts @@ -113,7 +113,9 @@ type Route = { quotaRetries: number; deadline?: ReturnType; }; -const CHANNEL_KINDS = [9, 40002, 40099, 40003, 5, 9005, 7, 39000, 39002, 39005]; +const CHANNEL_KINDS = [ + 9, 40002, 40099, 40003, 5, 9005, 7, 39000, 39002, 39005, 20002, +]; /** One authenticated socket, independently established channel routes and two explicit globals. * Recent replay is opportunistic: finite reads own catch-up and history bounds. */ export function subscribeRelayTraffic( @@ -458,6 +460,17 @@ export function subscribeRelayTraffic( fail(route, "Relay supplied invalid live traffic"); return; } + // Ephemeral channel activity must arrive on that exact authenticated + // channel route; a global or another channel is not an access grant. + if ( + incoming.kind === 20002 && + (!route.channelId || + incoming.tags.filter(([name]) => name === "h").length !== 1 || + !incoming.tags.some( + ([name, value]) => name === "h" && value === route.channelId, + )) + ) + return; if (route.status === "pending") route.count++; if (route.id === "observer") { if ( diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index 3b5831f4..8b480862 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -17,6 +17,7 @@ import { browserReadStateStorage, type ReadStateStorage, } from "./read-state-storage"; +import { createTyping } from "./typing"; import { createUnread } from "./unread"; import type { IncomingListener, IncomingMessage } from "./incoming"; import { objectBody } from "./body"; @@ -84,6 +85,14 @@ export function createRelaySession( else listener(); }; let canAccess: (id: string) => boolean = () => true; + const typing = createTyping( + transport?.viewer ?? "", + (id) => + !closed && + canAccess(id) && + channels.queries.list().channels.some((channel) => channel.id === id), + notify, + ); const recent = new ByteLru<{ event: RelayEvent; revision: number }>( 4096, 8 * 1024 * 1024, @@ -159,6 +168,7 @@ export function createRelaySession( revoking++; try { accessEpoch++; + typing.clear(); // Filters cannot tell us ownership of broad/ID/reference reads. Infrequent // authoritative access loss cancels them all, not merely explicit #h reads. requests.invalidate(); @@ -240,10 +250,13 @@ export function createRelaySession( events.some((event) => [39000, 39002].includes(event.kind)) ) channels.acceptDiscovery(events); + // Ephemeral typing and observer telemetry never enter retained content views. const visible = events - .filter((event) => event.kind !== OBSERVER_KIND) + .filter((event) => event.kind !== OBSERVER_KIND && event.kind !== 20002) .filter(visibility(events)); const epoch = accessEpoch; + typing.accept(visible); + if (closed || epoch !== accessEpoch) return []; profiling.measure( "events.reconcile", events[0]?.id ?? "empty", @@ -589,6 +602,7 @@ export function createRelaySession( incomingListeners.delete(listener); }; }, + typing: typing.capability, unread: unread.capability, sidebarPreferences: sidebarPreferences.queries, live, @@ -965,6 +979,11 @@ export function createRelaySession( ) refreshRoster(); const visible = accept(events); + if (liveSnapshot.status === "connected") + typing.accept( + events.filter((event) => event.kind === 20002), + true, + ); if (closed || !candidates.size || !provenance?.channelId) return; const epoch = accessEpoch; const delivered = new Set(); @@ -996,6 +1015,7 @@ export function createRelaySession( state(snapshot) { if (closed) return; activity.state(snapshot); + if (snapshot.status !== "connected") typing.clear(); if ( snapshot.status !== "connected" && liveSnapshot.status === "connected" @@ -1059,6 +1079,7 @@ export function createRelaySession( accessEpoch++; cacheClearEpoch++; activity.clear(); + typing.clear(); sidebarPreferences.clear(); // New windows must not yield to or receive errors from retired owners. catchups.clear(); @@ -1076,6 +1097,7 @@ export function createRelaySession( }, dispose() { closed = true; + typing.dispose(); lifetime.abort(); activity.dispose(); sidebarPreferences.dispose(); diff --git a/src/features/relay/typing.integration.test.ts b/src/features/relay/typing.integration.test.ts new file mode 100644 index 00000000..90eb0b8b --- /dev/null +++ b/src/features/relay/typing.integration.test.ts @@ -0,0 +1,106 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { createRelaySession } from "./session"; +import type { LiveCallbacks } from "./live"; +import { keypair, roster, signed, scriptedTransport } from "./testing"; + +afterEach(() => vi.useRealTimers()); +it("owns one ephemeral projection across views; purges on access, disconnect, cache and disposal", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_800_000_000_000); + const viewer = keypair(), + relay = keypair(), + agent = keypair(); + const wire = scriptedTransport(viewer.pubkey, relay.pubkey); + let live!: LiveCallbacks; + const dispose = vi.fn(), + subscribe = vi.fn((callbacks: LiveCallbacks) => { + live = callbacks; + return { update() {}, retry() {}, dispose }; + }); + const owner = createRelaySession({ ...wire.transport, subscribe }); + const pulse = signed(agent, { + kind: 20002, + content: "", + tags: [["h", "a"]], + created_at: 1_800_000_000, + }); + const view = owner.session.observe([ + { kinds: [20002], "#h": ["a"], limit: 10 }, + ]); + const stops = [1, 2].map(() => owner.session.typing.subscribe(() => {})); + live.state({ status: "connected", routes: [] }); + const snapshot = owner.session.typing.snapshot; + // No roster means no access, even for a signed event. + live.receive([pulse]); + expect(snapshot()).toEqual([]); + live.receive([roster(relay, "a", [viewer.pubkey])]); + live.receive([pulse]); + expect(snapshot()).toHaveLength(1); + expect(view.snapshot().events).toEqual([]); + expect(subscribe).toHaveBeenCalledTimes(1); + const callback = vi.fn(() => expect(snapshot()).toEqual([])); + const stop = owner.session.typing.subscribe(callback); + live.receive([roster(relay, "a", [], 1_800_000_001), pulse]); + expect(snapshot()).toEqual([]); + expect(callback).toHaveBeenCalled(); + stop(); + live.receive([roster(relay, "a", [viewer.pubkey], 1_800_000_002), pulse]); + expect(snapshot()).toHaveLength(1); + live.state({ status: "retrying", routes: [] }); + expect(snapshot()).toEqual([]); + live.receive([pulse]); + expect(snapshot()).toEqual([]); + live.state({ status: "connected", routes: [] }); + live.receive([pulse]); + expect(snapshot()).toHaveLength(1); + await owner.clearCache(); + expect(snapshot()).toEqual([]); + live.receive([roster(relay, "a", [viewer.pubkey], 1_800_000_003), pulse]); + owner.dispose(); + live.receive([pulse]); + expect(snapshot()).toEqual([]); + expect(dispose).toHaveBeenCalledTimes(1); + for (const stop of stops) stop(); + view.dispose(); + expect(vi.getTimerCount()).toBe(0); + const other = createRelaySession(null); + expect(other.session.typing.snapshot()).toEqual([]); + other.dispose(); +}); + +it("a synchronous typing listener cannot reseed retained views after disposal", () => { + const viewer = keypair(), + relay = keypair(), + agent = keypair(); + const wire = scriptedTransport(viewer.pubkey, relay.pubkey); + let live!: LiveCallbacks; + const owner = createRelaySession({ + ...wire.transport, + subscribe(callbacks) { + live = callbacks; + return { update() {}, retry() {}, dispose() {} }; + }, + }); + live.state({ status: "connected", routes: [] }); + live.receive([roster(relay, "a", [viewer.pubkey])]); + const at = Math.floor(Date.now() / 1000); + const pulse = signed(agent, { + kind: 20002, + content: "", + created_at: at, + tags: [["h", "a"]], + }); + live.receive([pulse]); + const view = owner.session.observe([{ kinds: [9], "#h": ["a"], limit: 10 }]); + owner.session.typing.subscribe(() => owner.dispose()); + live.receive([ + signed(agent, { + kind: 9, + content: "fixture", + created_at: at, + tags: [["h", "a"]], + }), + ]); + expect(view.snapshot().events).toEqual([]); + expect(owner.session.typing.snapshot()).toEqual([]); +}); diff --git a/src/features/relay/typing.test.ts b/src/features/relay/typing.test.ts new file mode 100644 index 00000000..76796501 --- /dev/null +++ b/src/features/relay/typing.test.ts @@ -0,0 +1,156 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { createTyping } from "./typing"; +import { keypair, message, signed } from "./testing"; + +const agent = keypair(), + viewer = keypair(); +const epoch = 1_800_000_000; +function setup() { + vi.useFakeTimers(); + vi.setSystemTime(epoch * 1000); + const owner = createTyping( + viewer.pubkey, + (id) => id === "a", + (fn) => fn(), + ); + return { owner, snapshot: owner.capability.snapshot }; +} +function pulse(tags = [["h", "a"]], at = epoch, key = agent) { + return signed(key, { kind: 20002, content: "", tags, created_at: at }); +} +afterEach(() => vi.useRealTimers()); +it("expires at signed time, ignores duplicates/out-of-order pulses and uses one timer", () => { + const { owner, snapshot } = setup(); + owner.accept([pulse()], true); + expect(snapshot()).toHaveLength(1); + expect(vi.getTimerCount()).toBe(1); + vi.advanceTimersByTime(3000); + owner.accept([pulse(), pulse(undefined, epoch - 1)], true); + vi.advanceTimersByTime(4999); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(1); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); +}); +it("rejects finite, expired, future, self, denied and malformed channel/thread scope", () => { + const { owner, snapshot } = setup(); + owner.accept([pulse()]); + owner.accept( + [ + pulse(undefined, epoch - 8), + pulse(undefined, epoch + 1), + pulse(undefined, epoch, viewer), + pulse([]), + pulse([["h", "denied"]]), + pulse([ + ["h", "a"], + ["h", "a"], + ]), + pulse([ + ["h", "a"], + ["e", "bad", "", "reply"], + ]), + pulse([ + ["h", "a"], + ["e", "a".repeat(64), "", "root"], + ]), + pulse([ + ["h", "a"], + ["e", "a".repeat(64)], + ]), + ], + true, + ); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); +}); +it("separates channel, canonical threads, nested roots and multiple signers", () => { + const { owner, snapshot } = setup(); + const root = "a".repeat(64), + parent = "b".repeat(64); + owner.accept( + [ + pulse(), + pulse([ + ["h", "a"], + ["e", root, "", "reply"], + ]), + pulse(undefined, epoch, keypair()), + ], + true, + ); + expect(snapshot()).toHaveLength(3); + owner.accept( + [ + pulse([ + ["h", "a"], + ["e", root, "", "root"], + ["e", parent, "", "reply"], + ]), + ], + true, + ); + expect(snapshot()).toHaveLength(3); + expect(snapshot().filter((e) => e.threadRootId === root)).toHaveLength(1); + owner.dispose(); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); +}); +it("messages win a batch, suppress late pulses for two seconds and retain timestamp watermarks", () => { + const { owner, snapshot } = setup(); + owner.accept([pulse(), message(agent, "a", "fixture", epoch)], true); + expect(snapshot()).toEqual([]); + vi.advanceTimersByTime(1000); + owner.accept([pulse(undefined, epoch + 1)], true); + expect(snapshot()).toEqual([]); + vi.advanceTimersByTime(1000); + owner.accept([pulse(), pulse(undefined, epoch + 2)], true); + expect(snapshot()).toHaveLength(1); + // Duplicate history cannot extend suppression or remove newer activity. + owner.accept([message(agent, "a", "old", epoch)]); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(8000); + expect(snapshot()).toEqual([]); +}); +it("message suppression is signer/thread scoped and clears only older activity", () => { + const { owner, snapshot } = setup(); + const tags = [ + ["h", "a"], + ["e", "a".repeat(64), "", "reply"], + ]; + owner.accept([pulse(), pulse(tags)], true); + owner.accept([message(agent, "a", "channel", epoch)]); + expect(snapshot()).toHaveLength(1); + expect(snapshot()[0]?.threadRootId).toBe("a".repeat(64)); + owner.accept([ + signed(agent, { kind: 40002, tags, content: "{}", created_at: epoch }), + ]); + expect(snapshot()).toEqual([]); +}); +it("bounds active and suppression records without eviction; teardown fences retained callbacks", () => { + const { owner, snapshot } = setup(); + // Distinct roots avoid generating 1025 signing keys. + owner.accept( + Array.from({ length: 1025 }, (_, i) => + pulse([ + ["h", "a"], + ["e", i.toString(16).padStart(64, "0"), "", "reply"], + ]), + ), + true, + ); + expect(snapshot()).toHaveLength(1024); + expect(vi.getTimerCount()).toBe(1); + const listener = vi.fn(); + const stop = owner.capability.subscribe(listener); + owner.clear(); + expect(snapshot()).toEqual([]); + expect(listener).toHaveBeenCalledTimes(1); + stop(); + owner.accept([pulse()], true); + expect(listener).toHaveBeenCalledTimes(1); + owner.dispose(); + owner.accept([pulse()], true); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/src/features/relay/typing.ts b/src/features/relay/typing.ts new file mode 100644 index 00000000..415ca268 --- /dev/null +++ b/src/features/relay/typing.ts @@ -0,0 +1,170 @@ +import type { RelayEvent } from "./events"; +import { threadReference } from "./thread-reference"; + +const TTL = 8_000; +const SUPPRESS = 2_000; +const CAPACITY = 1024; +export type TypingEntry = Readonly<{ + channelId: string; + threadRootId?: string; + pubkey: string; +}>; +type Record = { + entry: TypingEntry; + typingAt: number; + messageAt: number; + expires: number; + suppress: number; + retire: number; +}; + +/** Receive-only ephemeral projection. Input is verified by the existing host + * transport; identity is the signer, never a display name or an untrusted p tag. + * No retained event payloads, reads, persistence, or per-consumer timers. */ +export function createTyping( + viewer: string, + canAccess: (channelId: string) => boolean, + notify: (listener: () => void) => void, +) { + let closed = false; + let timer: ReturnType | undefined; + const records = new Map(); + const listeners = new Set<() => void>(); + let snapshot: readonly TypingEntry[] = Object.freeze([]); + function publish() { + clearTimeout(timer); + timer = undefined; + const now = Date.now(); + let nextWake = Infinity; + const next: TypingEntry[] = []; + for (const [key, record] of records) { + if (record.retire <= now) { + records.delete(key); + continue; + } + nextWake = Math.min(nextWake, record.retire); + if (record.expires > now) { + next.push(record.entry); + nextWake = Math.min(nextWake, record.expires); + } + } + if (!closed && nextWake < Infinity) + timer = setTimeout(publish, nextWake - now); + if ( + next.length === snapshot.length && + next.every((e, i) => e === snapshot[i]) + ) + return; + snapshot = Object.freeze(next); + for (const listener of listeners) notify(listener); + } + function accept(events: readonly RelayEvent[], live = false) { + if (closed) return; + const now = Date.now(); + // Prune before admission; at capacity drop new keys, never evict suppression + // evidence to admit an older pulse. All records retire within a bounded TTL. + for (const [key, record] of records) + if (record.retire <= now) records.delete(key); + // Completion wins independent of batch ordering (including equal seconds). + for (const event of [...events].sort( + (a, b) => Number(a.kind === 20002) - Number(b.kind === 20002), + )) { + const typing = event.kind === 20002; + if ((!typing && ![9, 40002].includes(event.kind)) || (typing && !live)) + continue; + if (event.pubkey === viewer || !/^[0-9a-f]{64}$/.test(event.pubkey)) + continue; + const at = event.created_at * 1000; + if ( + !Number.isSafeInteger(event.created_at) || + at > now || + at + TTL <= now + ) + continue; + const channels = event.tags.filter(([name]) => name === "h"); + const channelId = channels[0]?.[1]; + if ( + channels.length !== 1 || + !channelId || + !/^[a-zA-Z0-9_-]{1,128}$/.test(channelId) || + !canAccess(channelId) + ) + continue; + const refs = event.tags.filter(([name]) => name === "e"); + // Canonical reply, optionally with one marked root for nested replies. + if ( + refs.length && + (refs.length > 2 || + refs.some( + (tag) => + !/^[0-9a-f]{64}$/i.test(tag[1] ?? "") || + !["root", "reply"].includes(tag[3] ?? ""), + ) || + refs.filter((tag) => tag[3] === "reply").length !== 1 || + refs.filter((tag) => tag[3] === "root").length > 1) + ) + continue; + const threadRootId = threadReference(event)?.rootId; + const key = `${channelId}:${threadRootId ?? ""}:${event.pubkey}`; + let record = records.get(key); + if (!record) { + if (records.size >= CAPACITY) continue; + record = { + entry: Object.freeze({ + channelId, + ...(threadRootId ? { threadRootId } : {}), + pubkey: event.pubkey, + }), + typingAt: -1, + messageAt: -1, + expires: 0, + suppress: 0, + retire: 0, + }; + records.set(key, record); + } + if (typing) { + if ( + at <= record.typingAt || + at <= record.messageAt || + record.suppress > now + ) + continue; + record.typingAt = at; + record.expires = at + TTL; + } else { + if (at <= record.messageAt) continue; + record.messageAt = at; + if (at >= record.typingAt) { + record.expires = 0; + record.suppress = now + SUPPRESS; + } + } + record.retire = Math.max(record.retire, at + TTL, record.suppress); + } + publish(); + } + function clear() { + records.clear(); + publish(); + } + return { + capability: Object.freeze({ + snapshot: () => snapshot, + subscribe(listener: () => void) { + if (closed) return () => {}; + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + }), + accept, + clear, + dispose() { + closed = true; + clear(); + listeners.clear(); + }, + }; +} diff --git a/tests/browser/fixture.mjs b/tests/browser/fixture.mjs index b4ecb916..6ffed0fe 100644 --- a/tests/browser/fixture.mjs +++ b/tests/browser/fixture.mjs @@ -63,6 +63,7 @@ export const test = base.extend({ testInfo, ) => { const relayKey = generateSecretKey(); + const typingKeys = [generateSecretKey(), generateSecretKey()]; const userKey = generateSecretKey(); const viewer = getPublicKey(userKey); const membershipKeys = membershipActivity @@ -875,6 +876,26 @@ export const test = base.extend({ expect(rosterIds).toContain(id); rosterIds.splice(rosterIds.indexOf(id), 1); }, + // Signed upstream-only simulations: never a browser publication or live relay. + activity({ + channel = "alpha", + root, + author = 0, + kind = 20002, + age = 0, + } = {}) { + if (!relay) + throw new Error("Typing fixture requires production broker"); + const event = sign( + kind, + [["h", channel], ...(root ? [["e", root, "", "reply"]] : [])], + kind === 20002 ? "" : "Fixture completion", + typingKeys[author], + Math.floor(Date.now() / 1000) - age, + ); + relay.publish("primary", event); + return event; + }, edit(community, channel, target, content) { const event = sign( 40003, diff --git a/tests/browser/typing.spec.mjs b/tests/browser/typing.spec.mjs new file mode 100644 index 00000000..4645ff5f --- /dev/null +++ b/tests/browser/typing.spec.mjs @@ -0,0 +1,54 @@ +import { test, expect } from "./fixture.mjs"; +import { open } from "./timeline.mjs"; + +test.use({ productionBroker: true, readState: true, threadUnread: true }); +test("Messages receives scoped typing through authenticated live traffic and expires it without publishing", async ({ + page, + app, +}, testInfo) => { + await open(page, app); + await expect + .poll(() => + app.report.liveRequests.some((r) => r.filter?.["#h"]?.includes("alpha")), + ) + .toBe(true); + const indicator = page.getByRole("status", { name: "Typing activity" }); + app.activity({ age: 9 }); + await expect(indicator).toHaveCount(0); + app.activity(); + await expect(indicator).toContainText("is typing…"); + app.activity({ author: 1 }); + await expect(indicator).toContainText("are typing…"); + await page.screenshot({ path: testInfo.outputPath("messages-typing.png") }); + app.activity({ kind: 9 }); + await expect(indicator).toContainText("is typing…"); + app.activity({ kind: 9, author: 1 }); + await expect(indicator).toHaveCount(0); + app.activity(); // same-second late pulse cannot resurrect completion + await expect(indicator).toHaveCount(0); + const root = app.histories + .get("primary/alpha") + .find((e) => e.content === "Thread root 0"); + await page + .locator(`[data-channel-timeline] [data-message-id="${root.id}"]`) + .getByRole("button", { name: /^View thread:/ }) + .click(); + const thread = page.getByRole("complementary", { + name: "Thread", + exact: true, + }); + await expect( + thread.getByRole("textbox", { name: "Reply to thread", exact: true }), + ).toBeVisible(); + app.activity({ root: root.id }); + await expect( + thread.getByRole("status", { name: "Typing activity" }), + ).toContainText("is typing…"); + await expect(indicator).toHaveCount(1); + await page.screenshot({ + path: testInfo.outputPath("messages-thread-typing.png"), + }); + // Real browser timer, signed timestamp TTL, no polling transport or fixture cleanup. + await expect(indicator).toHaveCount(0, { timeout: 10000 }); + expect(app.report.publications).toEqual([]); +}); From f04cc34ade8084cc1bd8e9cc2f023efd674c507d Mon Sep 17 00:00:00 2001 From: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> Date: Fri, 11 Sep 2026 22:20:09 -0700 Subject: [PATCH 2/7] Fix typing completion scope and reentrant live admission Signed-off-by: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> --- src/features/relay/session.ts | 14 ++- src/features/relay/typing.integration.test.ts | 103 ++++++++++++++++++ src/features/relay/typing.test.ts | 57 ++++++++++ src/features/relay/typing.ts | 4 +- 4 files changed, 174 insertions(+), 4 deletions(-) diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index 8b480862..c54d3fc4 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -978,14 +978,22 @@ export function createRelaySession( ) ) refreshRoster(); + const epoch = accessEpoch; + const generation = liveGeneration; const visible = accept(events); - if (liveSnapshot.status === "connected") + // Completion subscribers can synchronously clear, revoke or retire this + // live delivery. Do not admit its remaining pulses into the new lifetime. + if ( + !closed && + epoch === accessEpoch && + generation === liveGeneration && + liveSnapshot.status === "connected" + ) typing.accept( events.filter((event) => event.kind === 20002), true, ); - if (closed || !candidates.size || !provenance?.channelId) return; - const epoch = accessEpoch; + if (closed || epoch !== accessEpoch || !candidates.size || !provenance?.channelId) return; const delivered = new Set(); const incoming: readonly IncomingMessage[] = Object.freeze( visible.flatMap((event) => { diff --git a/src/features/relay/typing.integration.test.ts b/src/features/relay/typing.integration.test.ts index 90eb0b8b..11be6144 100644 --- a/src/features/relay/typing.integration.test.ts +++ b/src/features/relay/typing.integration.test.ts @@ -104,3 +104,106 @@ it("a synchronous typing listener cannot reseed retained views after disposal", expect(view.snapshot().events).toEqual([]); expect(owner.session.typing.snapshot()).toEqual([]); }); + +for (const transition of ["cache", "dispose", "access", "reconnect"] as const) { + it(`fences the rest of a live callback batch after reentrant ${transition}`, async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_800_000_000_000); + const viewer = keypair(), + relay = keypair(), + agent = keypair(); + const wire = scriptedTransport(viewer.pubkey, relay.pubkey); + let live!: LiveCallbacks; + const owner = createRelaySession({ + ...wire.transport, + subscribe(callbacks) { + live = callbacks; + return { update() {}, retry() {}, dispose() {} }; + }, + }); + live.state({ status: "connected", routes: [] }); + live.receive([ + roster(relay, "a", [viewer.pubkey]), + roster(relay, "b", [viewer.pubkey]), + ]); + const pulse = signed(agent, { + kind: 20002, + content: "", + created_at: 1_800_000_000, + tags: [["h", "a"]], + }); + live.receive([pulse]); + expect(owner.session.typing.snapshot()).toHaveLength(1); + let clearing: Promise | undefined; + const listener = vi.fn(() => { + expect(owner.session.typing.snapshot()).toEqual([]); + if (transition === "cache") clearing = owner.clearCache(); + else if (transition === "dispose") owner.dispose(); + // Revoking another channel still invalidates the in-flight access epoch. + else if (transition === "access") + live.receive([roster(relay, "b", [], 1_800_000_001)]); + else { + live.state({ status: "retrying", routes: [] }); + live.state({ status: "connected", routes: [] }); + } + }); + owner.session.typing.subscribe(listener); + // Supported LiveCallbacks batch boundary; WS/SSE currently deliver singletons. + live.receive([ + signed(agent, { + kind: 9, + content: "fixture", + created_at: 1_800_000_000, + tags: [["h", "a"]], + }), + pulse, + ]); + await clearing; + expect(listener).toHaveBeenCalledTimes(1); + expect(owner.session.typing.snapshot()).toEqual([]); + owner.dispose(); + expect(vi.getTimerCount()).toBe(0); + }); +} + +it("refreshing a finite kind-20002 view neither retains nor activates typing", async () => { + const viewer = keypair(), + relay = keypair(), + agent = keypair(); + const wire = scriptedTransport(viewer.pubkey, relay.pubkey); + let live!: LiveCallbacks; + const owner = createRelaySession({ + ...wire.transport, + subscribe(callbacks) { + live = callbacks; + return { update() {}, retry() {}, dispose() {} }; + }, + }); + try { + live.state({ status: "connected", routes: [] }); + live.receive([roster(relay, "a", [viewer.pubkey])]); + const filters = [{ kinds: [20002], "#h": ["a"], limit: 10 }]; + const view = owner.session.observe(filters); + const pulse = signed(agent, { + kind: 20002, + content: "", + created_at: Math.floor(Date.now() / 1000), + tags: [["h", "a"]], + }); + const refresh = view.refresh(); + const read = wire.next(); + expect(read.filters).toEqual(filters); + read.respond([pulse]); + await refresh; + expect(view.snapshot()).toMatchObject({ status: "ready", events: [] }); + expect(owner.session.typing.snapshot()).toEqual([]); + // The same signed event is valid live, but still cannot enter finite views. + live.receive([pulse]); + expect(owner.session.typing.snapshot()).toHaveLength(1); + expect(view.snapshot().events).toEqual([]); + const later = owner.session.observe(filters); + expect(later.snapshot().events).toEqual([]); + } finally { + owner.dispose(); + } +}); diff --git a/src/features/relay/typing.test.ts b/src/features/relay/typing.test.ts index 76796501..e63db962 100644 --- a/src/features/relay/typing.test.ts +++ b/src/features/relay/typing.test.ts @@ -58,6 +58,16 @@ it("rejects finite, expired, future, self, denied and malformed channel/thread s ["h", "a"], ["e", "a".repeat(64)], ]), + pulse([ + ["h", "a"], + ["e", "a".repeat(64), "", "mention"], + ]), + pulse([ + ["h", "a"], + ["e", "a".repeat(64), "", "root"], + ["e", "b".repeat(64), "", "reply"], + ["e", "c".repeat(64), "", "mention"], + ]), ], true, ); @@ -154,3 +164,50 @@ it("bounds active and suppression records without eviction; teardown fences reta expect(snapshot()).toEqual([]); expect(vi.getTimerCount()).toBe(0); }); + +for (const kind of [9, 40002]) { + for (const scope of ["mention", "quote", "reply", "nested"] as const) { + it(`kind ${kind} ${scope} content clears and suppresses its authoritative scope`, () => { + const { owner, snapshot } = setup(); + const root = "a".repeat(64); + const threadTags = [ + ["h", "a"], + ["e", root, "", "reply"], + ]; + const references = + scope === "mention" + ? [["e", "b".repeat(64), "", "mention"]] + : scope === "quote" + ? [["e", "b".repeat(64)]] + : [ + ...(scope === "nested" ? [["e", root, "", "root"]] : []), + ["e", scope === "nested" ? "c".repeat(64) : root, "", "reply"], + ["e", "b".repeat(64), "", "mention"], + ["e", "d".repeat(64)], + ]; + const threaded = scope === "reply" || scope === "nested"; + const target = pulse(threaded ? threadTags : undefined); + const other = pulse(threaded ? undefined : threadTags); + owner.accept([target, other], true); + owner.accept([ + signed(agent, { + kind, + content: "fixture", + created_at: epoch, + tags: [["h", "a"], ...references], + }), + ]); + expect(snapshot()).toHaveLength(1); + expect(snapshot()[0]?.threadRootId).toBe(threaded ? undefined : root); + // Same-second replay and a newer pulse during the quiet period both lose. + owner.accept([target], true); + vi.advanceTimersByTime(1000); + owner.accept([pulse(threaded ? threadTags : undefined, epoch + 1)], true); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(1000); + owner.accept([pulse(threaded ? threadTags : undefined, epoch + 2)], true); + expect(snapshot()).toHaveLength(2); + owner.dispose(); + }); + } +} diff --git a/src/features/relay/typing.ts b/src/features/relay/typing.ts index 415ca268..dbec2792 100644 --- a/src/features/relay/typing.ts +++ b/src/features/relay/typing.ts @@ -91,8 +91,10 @@ export function createTyping( ) continue; const refs = event.tags.filter(([name]) => name === "e"); - // Canonical reply, optionally with one marked root for nested replies. + // Pulses require an unambiguous canonical scope. Content uses the same + // threadReference semantics as folding, including non-thread references. if ( + typing && refs.length && (refs.length > 2 || refs.some( From 60383113f5f7704b816be9fbb965a0230e15bdc5 Mon Sep 17 00:00:00 2001 From: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> Date: Sat, 12 Sep 2026 09:51:28 -0700 Subject: [PATCH 3/7] Clarify typing lifecycle and route responsibilities Signed-off-by: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> --- docs/relay-queries.md | 67 +++------ src/features/relay/live.test.ts | 82 +++++++++++- src/features/relay/live.ts | 5 +- src/features/relay/typing.test.ts | 56 ++++++++ src/features/relay/typing.ts | 216 ++++++++++++++++-------------- 5 files changed, 271 insertions(+), 155 deletions(-) diff --git a/docs/relay-queries.md b/docs/relay-queries.md index 0014a8f0..db6d0791 100644 --- a/docs/relay-queries.md +++ b/docs/relay-queries.md @@ -475,52 +475,21 @@ client sharing the account cannot cause a refusal. ## Receive-only typing -`session.typing.snapshot()` / `.subscribe()` expose immutable active entries -`{ channelId, threadRootId?, pubkey }`. The session owns one ephemeral projection; -shared `features/messages/TypingIndicator` renders it inside channel/thread -composers, including read-only connections. Page plugins consume the same UI and -session. Mounting consumers starts no reads, profile enrichment or subscriptions. -Names reuse the shared profile snapshot, with a public-key fragment fallback. -This reports signed typing activity (including agents), **not** inferred agent -execution, online presence, or a promise that an answer is coming. - -Kind 20002 joins the existing authenticated channel route, with the same signature -verification and socket/generation fencing. Typing must match that exact route, -name exactly one bounded `h`, belong to the current visible roster and pass the -existing channel access policy. Identity is the event signer, matching message -folding; arbitrary `p`/actor tags and profile display names do not confer identity. -A channel pulse has no `e`; a thread pulse has one marked canonical `reply`, with -at most one marked `root` for nested replies. Thread scope uses the existing -canonical root convention; it is not an extra target lookup or an access grant. -Malformed/ambiguous references are dropped rather than displayed at channel level. - -Only live pulses can activate typing. No typing payload enters finite views, -recent history, channel caches, unread counts, the outbox or persistent storage. -Expiry is eight seconds from the signed timestamp. Already-expired and future -activity is rejected (no clock-skew allowance). Duplicates/older pulses cannot -extend expiry. Signed content messages (9/40002) suppress their signer's same -channel/thread activity; timestamp watermarks reject older pulses and a two-second -post-message quiet period covers late activity. Old history does not clear newer -activity or repeatedly extend suppression. This does not reinterpret edits as -new messages or resolve relay-proxied author envelopes beyond current host rules. - -At most 1,024 active/suppression records and one timeout exist per session. At -capacity new identities/scopes are dropped until expiry; suppression evidence is -never evicted to admit a stale pulse. Access loss clears all typing conservatively, -before batched access notifications. Disconnect, cache clear, session disposal and -account/community replacement clear it too. Session disposal fences late callbacks; -plugins unloading only remove their UI subscriptions. There is no typing publisher, -new signer, transport, polling, durable cache, or plugin-owned ephemeral owner. - -Protocol reference: `block/buzz`'s -`crates/buzz-acp/src/relay.rs::build_typing_event` and -`desktop/src/features/messages/useChannelTyping.ts` (8-second TTL and 2-second -post-message suppression). Regression owners are `typing*.test.ts`, `live.test.ts` -and `tests/browser/typing.spec.mjs`; browser simulations use ephemeral fixture -keys through the production authenticated broker, never the deployed relay. - -**Review-sensitive:** `session.ts` (FOUNDATION admission/lifecycle and synchronous -subscriber reentrancy), `live.ts` (existing route filter/verification boundary), -`typing.ts` (timestamp, scope, bounds and suppression), and shared composer placement. -An existing development broker imports live routing at startup and needs one -coordinated restart to receive kind 20002; frontend hot reload alone is insufficient. +Typing indicators show recent activity from another participant in the current +channel or thread. They identify the signer, including agents; they do not imply +online presence, ongoing agent execution, or a promise of an answer. + +The session owns this temporary state through `session.typing`. Shared conversation +composers use the same indicator so channel and thread views agree about who is +active and where. Names reuse already loaded profiles; displaying activity does +not start extra reads or connections. + +A message clears its author's preceding activity in that conversation. Brief +post-message suppression prevents late activity from immediately bringing the +indicator back; otherwise silence lets it expire. Only current live activity can +activate it, never fetched history. Losing access, disconnecting, clearing the +cache or replacing the session clears it too, so stale activity cannot carry into +another conversation or account. + +Typing stays out of message history, unread counts and persistent storage. This +is receive-only: opening or using a composer does not publish typing activity. diff --git a/src/features/relay/live.test.ts b/src/features/relay/live.test.ts index 4821b392..5de753d6 100644 --- a/src/features/relay/live.test.ts +++ b/src/features/relay/live.test.ts @@ -5,7 +5,8 @@ import { subscribeRelayTraffic, type LiveCallbacks, } from "./live"; -import { keypair, message, signed } from "./testing"; +import { keypair, message, roster, signed, scriptedTransport } from "./testing"; +import { createRelaySession } from "./session"; class Socket { readyState = 1; onmessage?: (event: { data: string }) => Promise; @@ -726,3 +727,82 @@ it("admits signed typing only on its authenticated channel route, without extra h.owner.dispose(); expect(vi.getTimerCount()).toBe(0); }); + +it("keeps misrouted activity out of accessible conversations; session rejects ambiguous scope", async () => { + vi.useFakeTimers(); + const h = setup(); + const relay = keypair(); + const wire = scriptedTransport(h.key.pubkey, relay.pubkey); + const owner = createRelaySession({ + ...wire.transport, + subscribe(callbacks) { + h.callbacks.receive.mockImplementation(callbacks.receive); + h.callbacks.state.mockImplementation(callbacks.state); + return h.owner; + }, + }); + // Establish both accessible channels before starting their live routes. + h.callbacks.receive([ + roster(relay, "a", [h.key.pubkey]), + roster(relay, "b", [h.key.pubkey]), + ]); + await h.first.auth(); + await vi.advanceTimersByTimeAsync(750); + const requests = h.first.requests(); + const a = requests.find((r) => r[2]["#h"]?.includes("a")); + const b = requests.find((r) => r[2]["#h"]?.includes("b")); + assert.exists(a); + assert.exists(b); + const agent = keypair(); + const activity = (tags: string[][]) => + signed(agent, { + kind: 20002, + created_at: Math.floor(Date.now() / 1000), + content: "", + tags, + }); + const pulse = activity([["h", "b"]]); + const snapshot = owner.session.typing.snapshot; + for (const route of [requests[0], requests[1], a]) { + await h.first.receive(["EVENT", route?.[1], pulse]); + expect(snapshot()).toEqual([]); + } + for (const tags of [ + [], + [["h"]], + [["h", "bad channel"]], + [["h", "denied"]], + [ + ["h", "b"], + ["h", "b"], + ], + [ + ["h", "b"], + ["h", "a"], + ], + [ + ["h", "b"], + ["e", "bad", "", "reply"], + ], + ]) { + const event = activity(tags); + await h.first.receive(["EVENT", b[1], event]); + expect(snapshot()).toEqual([]); + // Even a host that has already discarded route metadata cannot activate these. + h.callbacks.receive([event]); + expect(snapshot()).toEqual([]); + } + await h.first.receive(["EVENT", b[1], pulse]); + expect(snapshot()).toEqual([{ channelId: "b", pubkey: agent.pubkey }]); + // Once route metadata is gone, the session can only use the event's own scope. + const other = activity([["h", "a"]]); + await h.first.receive(["EVENT", b[1], other]); + expect(snapshot()).toHaveLength(1); + h.callbacks.receive([other]); + expect(snapshot()).toEqual([ + { channelId: "b", pubkey: agent.pubkey }, + { channelId: "a", pubkey: agent.pubkey }, + ]); + owner.dispose(); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/src/features/relay/live.ts b/src/features/relay/live.ts index d14f1c32..7680211b 100644 --- a/src/features/relay/live.ts +++ b/src/features/relay/live.ts @@ -460,12 +460,11 @@ export function subscribeRelayTraffic( fail(route, "Relay supplied invalid live traffic"); return; } - // Ephemeral channel activity must arrive on that exact authenticated - // channel route; a global or another channel is not an access grant. + // Preserve route consistency before receive() discards the subscription ID. + // The typing owner separately checks scope shape and channel access. if ( incoming.kind === 20002 && (!route.channelId || - incoming.tags.filter(([name]) => name === "h").length !== 1 || !incoming.tags.some( ([name, value]) => name === "h" && value === route.channelId, )) diff --git a/src/features/relay/typing.test.ts b/src/features/relay/typing.test.ts index e63db962..0e12a8e8 100644 --- a/src/features/relay/typing.test.ts +++ b/src/features/relay/typing.test.ts @@ -211,3 +211,59 @@ for (const kind of [9, 40002]) { }); } } + +it("stays visible across three-second heartbeats, then expires eight seconds after the last signed pulse", () => { + const { owner, snapshot } = setup(); + owner.accept([pulse()], true); + for (const seconds of [3, 6, 9]) { + vi.advanceTimersByTime(3000); + expect(snapshot()).toHaveLength(1); + owner.accept([pulse(undefined, epoch + seconds)], true); + } + // One missed scheduled heartbeat does not flicker; a second exhausts the margin. + vi.advanceTimersByTime(7999); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(1); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); +}); + +it("rejects delayed pre-message activity after the quiet period, but admits genuinely newer activity", () => { + const { owner, snapshot } = setup(); + owner.accept([message(agent, "a", "complete", epoch)]); + vi.advanceTimersByTime(3000); + owner.accept([pulse(undefined, epoch - 1), pulse()], true); + expect(snapshot()).toEqual([]); + owner.accept([pulse(undefined, epoch + 3)], true); + expect(snapshot()).toHaveLength(1); + // A previously unseen older completion must not clear this newer activity. + owner.accept([message(agent, "a", "delayed completion", epoch + 1)]); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(8000); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); +}); + +it("retains completion evidence at capacity until stale pulses have expired", () => { + const { owner, snapshot } = setup(); + const messages = Array.from({ length: 1024 }, (_, i) => + signed(agent, { + kind: 9, + content: "complete", + created_at: epoch, + tags: [ + ["h", "a"], + ["e", i.toString(16).padStart(64, "0"), "", "reply"], + ], + }), + ); + owner.accept(messages); + vi.advanceTimersByTime(3000); + owner.accept([pulse(undefined, epoch + 3)], true); + expect(snapshot()).toEqual([]); + vi.advanceTimersByTime(5000); + owner.accept([pulse(undefined, epoch + 8)], true); + expect(snapshot()).toHaveLength(1); + owner.dispose(); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/src/features/relay/typing.ts b/src/features/relay/typing.ts index dbec2792..ad9683f5 100644 --- a/src/features/relay/typing.ts +++ b/src/features/relay/typing.ts @@ -1,26 +1,23 @@ import type { RelayEvent } from "./events"; import { threadReference } from "./thread-reference"; -const TTL = 8_000; -const SUPPRESS = 2_000; -const CAPACITY = 1024; +const ACTIVITY_LIFETIME_MS = 8_000; +const POST_MESSAGE_QUIET_MS = 2_000; +const MAX_ACTIVITY_RECORDS = 1024; export type TypingEntry = Readonly<{ channelId: string; threadRootId?: string; pubkey: string; }>; -type Record = { +type ParticipantActivity = { entry: TypingEntry; - typingAt: number; - messageAt: number; - expires: number; - suppress: number; - retire: number; + lastActivityAt: number; + lastMessageAt: number; + visibleUntil: number; + quietUntil: number; }; -/** Receive-only ephemeral projection. Input is verified by the existing host - * transport; identity is the signer, never a display name or an untrusted p tag. - * No retained event payloads, reads, persistence, or per-consumer timers. */ +/** Session-owned activity from verified events; no event payloads or persistence. */ export function createTyping( viewer: string, canAccess: (channelId: string) => boolean, @@ -28,24 +25,33 @@ export function createTyping( ) { let closed = false; let timer: ReturnType | undefined; - const records = new Map(); + const records = new Map(); const listeners = new Set<() => void>(); let snapshot: readonly TypingEntry[] = Object.freeze([]); + function retainUntil(record: ParticipantActivity) { + return Math.max( + Math.max(record.lastActivityAt, record.lastMessageAt) + + ACTIVITY_LIFETIME_MS, + record.quietUntil, + ); + } + function expireSilence(now: number) { + for (const [key, record] of records) { + if (retainUntil(record) <= now) records.delete(key); + } + } function publish() { clearTimeout(timer); timer = undefined; const now = Date.now(); let nextWake = Infinity; const next: TypingEntry[] = []; - for (const [key, record] of records) { - if (record.retire <= now) { - records.delete(key); - continue; - } - nextWake = Math.min(nextWake, record.retire); - if (record.expires > now) { + expireSilence(now); + for (const record of records.values()) { + nextWake = Math.min(nextWake, retainUntil(record)); + if (record.visibleUntil > now) { next.push(record.entry); - nextWake = Math.min(nextWake, record.expires); + nextWake = Math.min(nextWake, record.visibleUntil); } } if (!closed && nextWake < Infinity) @@ -58,91 +64,97 @@ export function createTyping( snapshot = Object.freeze(next); for (const listener of listeners) notify(listener); } + function receiveActivity( + record: ParticipantActivity, + at: number, + now: number, + ) { + if (at <= record.lastActivityAt || at <= record.lastMessageAt) return; + if (now < record.quietUntil) return; + record.lastActivityAt = at; + record.visibleUntil = at + ACTIVITY_LIFETIME_MS; + } + function recordCompletion( + record: ParticipantActivity, + at: number, + now: number, + ) { + if (at <= record.lastMessageAt) return; + record.lastMessageAt = at; + if (at < record.lastActivityAt) return; + record.visibleUntil = 0; + record.quietUntil = now + POST_MESSAGE_QUIET_MS; + } + function receive(event: RelayEvent, now: number) { + const typing = event.kind === 20002; + if (event.pubkey === viewer || !/^[0-9a-f]{64}$/.test(event.pubkey)) return; + const at = event.created_at * 1000; + if ( + !Number.isSafeInteger(event.created_at) || + at > now || + at + ACTIVITY_LIFETIME_MS <= now + ) + return; + const channels = event.tags.filter(([name]) => name === "h"); + const channelId = channels[0]?.[1]; + if ( + channels.length !== 1 || + !channelId || + !/^[a-zA-Z0-9_-]{1,128}$/.test(channelId) || + !canAccess(channelId) + ) + return; + const refs = event.tags.filter(([name]) => name === "e"); + // Pulses require an unambiguous canonical scope. Content uses the same + // threadReference semantics as folding, including non-thread references. + if ( + typing && + refs.length && + (refs.length > 2 || + refs.some( + (tag) => + !/^[0-9a-f]{64}$/i.test(tag[1] ?? "") || + !["root", "reply"].includes(tag[3] ?? ""), + ) || + refs.filter((tag) => tag[3] === "reply").length !== 1 || + refs.filter((tag) => tag[3] === "root").length > 1) + ) + return; + const threadRootId = threadReference(event)?.rootId; + const key = `${channelId}:${threadRootId ?? ""}:${event.pubkey}`; + let record = records.get(key); + if (!record) { + if (records.size >= MAX_ACTIVITY_RECORDS) return; + record = { + entry: Object.freeze({ + channelId, + ...(threadRootId ? { threadRootId } : {}), + pubkey: event.pubkey, + }), + lastActivityAt: -1, + lastMessageAt: -1, + visibleUntil: 0, + quietUntil: 0, + }; + records.set(key, record); + } + if (typing) receiveActivity(record, at, now); + else recordCompletion(record, at, now); + } function accept(events: readonly RelayEvent[], live = false) { if (closed) return; const now = Date.now(); - // Prune before admission; at capacity drop new keys, never evict suppression - // evidence to admit an older pulse. All records retire within a bounded TTL. - for (const [key, record] of records) - if (record.retire <= now) records.delete(key); - // Completion wins independent of batch ordering (including equal seconds). - for (const event of [...events].sort( - (a, b) => Number(a.kind === 20002) - Number(b.kind === 20002), - )) { - const typing = event.kind === 20002; - if ((!typing && ![9, 40002].includes(event.kind)) || (typing && !live)) - continue; - if (event.pubkey === viewer || !/^[0-9a-f]{64}$/.test(event.pubkey)) - continue; - const at = event.created_at * 1000; - if ( - !Number.isSafeInteger(event.created_at) || - at > now || - at + TTL <= now - ) - continue; - const channels = event.tags.filter(([name]) => name === "h"); - const channelId = channels[0]?.[1]; - if ( - channels.length !== 1 || - !channelId || - !/^[a-zA-Z0-9_-]{1,128}$/.test(channelId) || - !canAccess(channelId) - ) - continue; - const refs = event.tags.filter(([name]) => name === "e"); - // Pulses require an unambiguous canonical scope. Content uses the same - // threadReference semantics as folding, including non-thread references. - if ( - typing && - refs.length && - (refs.length > 2 || - refs.some( - (tag) => - !/^[0-9a-f]{64}$/i.test(tag[1] ?? "") || - !["root", "reply"].includes(tag[3] ?? ""), - ) || - refs.filter((tag) => tag[3] === "reply").length !== 1 || - refs.filter((tag) => tag[3] === "root").length > 1) - ) - continue; - const threadRootId = threadReference(event)?.rootId; - const key = `${channelId}:${threadRootId ?? ""}:${event.pubkey}`; - let record = records.get(key); - if (!record) { - if (records.size >= CAPACITY) continue; - record = { - entry: Object.freeze({ - channelId, - ...(threadRootId ? { threadRootId } : {}), - pubkey: event.pubkey, - }), - typingAt: -1, - messageAt: -1, - expires: 0, - suppress: 0, - retire: 0, - }; - records.set(key, record); - } - if (typing) { - if ( - at <= record.typingAt || - at <= record.messageAt || - record.suppress > now - ) - continue; - record.typingAt = at; - record.expires = at + TTL; - } else { - if (at <= record.messageAt) continue; - record.messageAt = at; - if (at >= record.typingAt) { - record.expires = 0; - record.suppress = now + SUPPRESS; - } + // Keep completion evidence until old pulses can no longer be current. + // At capacity, drop new scopes rather than evict that evidence. + expireSilence(now); + for (const event of events) { + if (event.kind === 9 || event.kind === 40002) receive(event, now); + } + // Completion wins even when activity came first in the batch. + if (live) { + for (const event of events) { + if (event.kind === 20002) receive(event, now); } - record.retire = Math.max(record.retire, at + TTL, record.suppress); } publish(); } From e69dfde7416889f2256a6084748bd147155873a9 Mon Sep 17 00:00:00 2001 From: Fizz <400e8babadcee6a7f420103f10a2849d84c4a9c71d5bd04f3948c814216648a3@buzz.block.builderlab.xyz> Date: Sat, 12 Sep 2026 15:01:45 -0700 Subject: [PATCH 4/7] fix: address typing viewport and suppressed replay review findings Signed-off-by: Fizz <400e8babadcee6a7f420103f10a2849d84c4a9c71d5bd04f3948c814216648a3@buzz.block.builderlab.xyz> --- src/features/messages/Messages.module.css | 3 + src/features/messages/TypingIndicator.tsx | 13 ++-- src/features/relay/typing.test.ts | 33 +++++++++ src/features/relay/typing.ts | 3 +- tests/browser/typing.spec.mjs | 81 ++++++++++++++++++++++- 5 files changed, 126 insertions(+), 7 deletions(-) diff --git a/src/features/messages/Messages.module.css b/src/features/messages/Messages.module.css index 1379183d..e8fdc0c3 100644 --- a/src/features/messages/Messages.module.css +++ b/src/features/messages/Messages.module.css @@ -714,6 +714,9 @@ button.avatar:focus-visible, .typing { color: var(--text-muted); font-size: calc(12px * var(--buzz-text-scale, 1)); + line-height: 1.5; + height: 1lh; + flex-shrink: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; diff --git a/src/features/messages/TypingIndicator.tsx b/src/features/messages/TypingIndicator.tsx index 59d05f51..0a2f50e3 100644 --- a/src/features/messages/TypingIndicator.tsx +++ b/src/features/messages/TypingIndicator.tsx @@ -24,17 +24,20 @@ export function TypingIndicator({ (entry) => entry.channelId === channelId && entry.threadRootId === threadRootId, ); - if (!matching.length) return null; // Reuse already available names; optional typing must not trigger profile reads. const names = matching .slice(0, 3) .map(({ pubkey }) => profiles.get(pubkey)?.name ?? pubkey.slice(0, 10)); const others = matching.length - names.length; return ( -
- {names.join(", ")} - {others > 0 ? ` and ${others} others` : ""} - {matching.length === 1 ? " is typing…" : " are typing…"} +
+ {matching.length > 0 && ( + + {names.join(", ")} + {others > 0 ? ` and ${others} others` : ""} + {matching.length === 1 ? " is typing…" : " are typing…"} + + )}
); } diff --git a/src/features/relay/typing.test.ts b/src/features/relay/typing.test.ts index 0e12a8e8..7a69abb0 100644 --- a/src/features/relay/typing.test.ts +++ b/src/features/relay/typing.test.ts @@ -122,6 +122,39 @@ it("messages win a batch, suppress late pulses for two seconds and retain timest vi.advanceTimersByTime(8000); expect(snapshot()).toEqual([]); }); +for (const kind of [9, 40002]) { + it(`kind ${kind} quiet suppression remembers replayed pulses without deferring activity or extending quiet`, () => { + const { owner, snapshot } = setup(); + owner.accept([ + signed(agent, { + kind, + tags: [["h", "a"]], + content: "complete", + created_at: epoch, + }), + ]); + vi.advanceTimersByTime(1000); + const suppressed = pulse(undefined, epoch + 1); + owner.accept([suppressed], true); + expect(snapshot()).toEqual([]); + vi.advanceTimersByTime(1000); + expect(snapshot()).toEqual([]); // Quiet ending never reveals a dropped pulse. + owner.accept([suppressed], true); + expect(snapshot()).toEqual([]); + // Suppression did not move the original two-second quiet deadline. + owner.accept([pulse(undefined, epoch + 2)], true); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(1000); + owner.accept([suppressed], true); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(6999); + expect(snapshot()).toHaveLength(1); + vi.advanceTimersByTime(1); + expect(snapshot()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); + }); +} + it("message suppression is signer/thread scoped and clears only older activity", () => { const { owner, snapshot } = setup(); const tags = [ diff --git a/src/features/relay/typing.ts b/src/features/relay/typing.ts index ad9683f5..0c96c7e9 100644 --- a/src/features/relay/typing.ts +++ b/src/features/relay/typing.ts @@ -70,8 +70,9 @@ export function createTyping( now: number, ) { if (at <= record.lastActivityAt || at <= record.lastMessageAt) return; - if (now < record.quietUntil) return; + // Remember suppressed pulses too: quiet ending must not admit their replays. record.lastActivityAt = at; + if (now < record.quietUntil) return; record.visibleUntil = at + ACTIVITY_LIFETIME_MS; } function recordCompletion( diff --git a/tests/browser/typing.spec.mjs b/tests/browser/typing.spec.mjs index 4645ff5f..aea07584 100644 --- a/tests/browser/typing.spec.mjs +++ b/tests/browser/typing.spec.mjs @@ -1,5 +1,5 @@ import { test, expect } from "./fixture.mjs"; -import { open } from "./timeline.mjs"; +import { open, end } from "./timeline.mjs"; test.use({ productionBroker: true, readState: true, threadUnread: true }); test("Messages receives scoped typing through authenticated live traffic and expires it without publishing", async ({ @@ -52,3 +52,82 @@ test("Messages receives scoped typing through authenticated live traffic and exp await expect(indicator).toHaveCount(0, { timeout: 10000 }); expect(app.report.publications).toEqual([]); }); + +for (const scope of ["channel", "thread"]) { + test(`${scope} typing preserves viewport bounds and the visible bottom through completion and expiry`, async ({ + page, + app, + }) => { + await open(page, app); + await expect + .poll(() => + app.report.liveRequests.some((r) => + r.filter?.["#h"]?.includes("alpha"), + ), + ) + .toBe(true); + const root = app.histories + .get("primary/alpha") + .find((e) => e.content === "Thread root 0"); + if (scope === "thread") { + // Seed enough signed upstream replies to exercise a genuinely scrolling thread. + for (let i = 0; i < 25; i++) app.reply(root.id); + await page + .locator(`[data-channel-timeline] [data-message-id="${root.id}"]`) + .getByRole("button", { name: /^View thread:/ }) + .click(); + await expect( + page.getByText("28 replies shown", { exact: true }), + ).toBeVisible(); + } else { + await end(page); + } + const history = page.getByRole("region", { + name: scope === "thread" ? "Thread messages" : "Channel message history", + exact: true, + }); + const composer = page.getByRole("form", { + name: scope === "thread" ? "Reply to thread" : "Send a message to Alpha", + exact: true, + }); + const indicator = composer.getByRole("status", { name: "Typing activity" }); + const gap = () => + history.evaluate( + (el) => el.scrollHeight - el.clientHeight - el.scrollTop, + ); + await expect.poll(gap).toBeLessThan(2); + expect( + await history.evaluate((el) => el.scrollHeight - el.clientHeight), + ).toBeGreaterThan(100); + const idle = await history.boundingBox(); + const idleComposer = await composer.boundingBox(); + const stable = async () => { + expect(await history.boundingBox()).toEqual(idle); + expect(await composer.boundingBox()).toEqual(idleComposer); + await expect.poll(gap).toBeLessThan(2); + const tail = await history + .locator("[data-message-id]") + .last() + .boundingBox(); + expect(tail.y).toBeGreaterThanOrEqual(idle.y - 2); + expect(tail.y + tail.height).toBeLessThanOrEqual( + idle.y + idle.height + 2, + ); + }; + await expect(indicator).toHaveCount(0); + const target = scope === "thread" ? { root: root.id } : {}; + app.activity(target); + await expect(indicator).toContainText("is typing…"); + await stable(); + app.activity({ ...target, kind: 9 }); + await expect(indicator).toHaveCount(0); + await stable(); + // A different signer is outside the first signer's quiet period. + app.activity({ ...target, author: 1 }); + await expect(indicator).toContainText("is typing…"); + await stable(); + await expect(indicator).toHaveCount(0, { timeout: 10000 }); + await stable(); + expect(app.report.publications).toEqual([]); + }); +} From 0680d3baacc04b007fe61b212b0e4381d17e6dba Mon Sep 17 00:00:00 2001 From: Fizz <400e8babadcee6a7f420103f10a2849d84c4a9c71d5bd04f3948c814216648a3@buzz.block.builderlab.xyz> Date: Sat, 12 Sep 2026 15:51:28 -0700 Subject: [PATCH 5/7] test: capture clipped reading anchors in tall timelines Signed-off-by: Fizz <400e8babadcee6a7f420103f10a2849d84c4a9c71d5bd04f3948c814216648a3@buzz.block.builderlab.xyz> --- docs/browser-testing.md | 6 +++- tests/browser/timeline-setup.spec.mjs | 45 ++++++++++++++++++++++++++- tests/browser/timeline.mjs | 17 +++++++--- 3 files changed, 61 insertions(+), 7 deletions(-) diff --git a/docs/browser-testing.md b/docs/browser-testing.md index 0ad69e29..1ab38d13 100644 --- a/docs/browser-testing.md +++ b/docs/browser-testing.md @@ -266,7 +266,11 @@ cursor responses for explicit paging tests; accidentally entering that path is n valid resize setup. `upper()` establishes above-bottom reading with at most four real wheel gestures, requiring progress and settled distance >400px. It does not measure exact wheel displacement. Partial-input and blocked-input controls guard -that setup; same-ID/Y <4px and bottom <4px assertions remain unchanged. No retries +that setup; same-ID/Y <4px and bottom <4px assertions remain unchanged. Anchor +capture prefers a whole paragraph, falling back to the first intersecting row +when tall messages leave only clipped paragraphs. A deterministic helper control +covers that geometry, whole-paragraph preference, offscreen rejection, and rejection +of an actual anchor displacement. No retries or additional WebKit exclusions are used. The underlying Linux WebKit single-wheel shortfall remains unattributed; this setup change does not fix or explain it. diff --git a/tests/browser/timeline-setup.spec.mjs b/tests/browser/timeline-setup.spec.mjs index c5a60772..520a45c9 100644 --- a/tests/browser/timeline-setup.spec.mjs +++ b/tests/browser/timeline-setup.spec.mjs @@ -1,5 +1,5 @@ import { test, expect } from "./fixture.mjs"; -import { open, upper, expectAnchor } from "./timeline.mjs"; +import { open, upper, anchor, expectAnchor } from "./timeline.mjs"; // Reading setup must not accidentally exercise older-page loading. test.use({ tallMessages: true }); @@ -35,6 +35,49 @@ test("reading setup handles partial wheel progress without weakening the anchor" } }); +test("reading anchor handles clipped paragraphs and still detects displacement", async ({ + page, +}) => { + // Deterministic geometry from the CI failure: both visible paragraphs are + // clipped, with no whole paragraph to select. This is a helper control, not + // a replacement for the production resize journeys. + await page.setContent(` +
+
+

Offscreen

+
+
+

Clipped at top

+
+
+

Clipped at bottom

+
+
+ `); + const saved = await anchor(page); + expect(saved).toEqual({ id: "clipped", y: -30 }); + await expectAnchor(page, saved); + await page.locator('[data-message-id="clipped"]').evaluate((row) => { + row.style.top = "-10px"; + }); + expect(await anchor(page)).toEqual({ id: saved.id, y: saved.y + 20 }); + // The unchanged oracle must reject a real jump, not merely find the same ID. + await expect(expectAnchor(page, saved)).rejects.toThrow( + "same visible message clipped at same viewport Y", + ); + await page.locator('[data-message-id="next"] p').evaluate((p) => { + p.style.height = "40px"; + }); + expect(await anchor(page)).toEqual({ id: "next", y: 130 }); + await history(page) + .locator("[data-message-id]") + .evaluateAll((rows) => { + for (const row of rows) row.style.top = "300px"; + }); + await expect(anchor(page)).rejects.toThrow("No visible message anchor"); +}); + test("reading setup rejects an immobile timeline instead of accepting a bottom anchor", async ({ page, app, diff --git a/tests/browser/timeline.mjs b/tests/browser/timeline.mjs index 7ed822e6..e56b9871 100644 --- a/tests/browser/timeline.mjs +++ b/tests/browser/timeline.mjs @@ -34,13 +34,20 @@ export async function settle(page) { export async function anchor(page) { return history(page).evaluate((element) => { const bounds = element.getBoundingClientRect(); - const row = Array.from(element.querySelectorAll("[data-message-id]")).find( - (row) => { + const rows = Array.from(element.querySelectorAll("[data-message-id]")); + // Prefer a whole paragraph, but tall messages can leave only clipped rows. + // Track the first intersecting row in that case, as the reader does. The + // same-ID/Y assertion below still detects displacement after a resize. + const row = + rows.find((row) => { const rect = row.querySelector("p").getBoundingClientRect(); return rect.top >= bounds.top && rect.bottom <= bounds.bottom; - }, - ); - if (!row) throw new Error("No fully visible message anchor"); + }) ?? + rows.find((row) => { + const rect = row.getBoundingClientRect(); + return rect.bottom > bounds.top && rect.top < bounds.bottom; + }); + if (!row) throw new Error("No visible message anchor"); return { id: row.dataset.messageId, y: row.querySelector("p").getBoundingClientRect().top - bounds.top, From a2c1c82a5d2fa3607b404f575480dd947eedd275 Mon Sep 17 00:00:00 2001 From: Fizz <400e8babadcee6a7f420103f10a2849d84c4a9c71d5bd04f3948c814216648a3@buzz.block.builderlab.xyz> Date: Mon, 14 Sep 2026 14:38:28 +0200 Subject: [PATCH 6/7] test: cover typing and incoming notifications after rebase Signed-off-by: Fizz <400e8babadcee6a7f420103f10a2849d84c4a9c71d5bd04f3948c814216648a3@buzz.block.builderlab.xyz> --- src/features/relay/live.test.ts | 5 +- src/features/relay/session.ts | 8 ++- src/features/relay/typing.integration.test.ts | 66 +++++++++++++++++++ 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/src/features/relay/live.test.ts b/src/features/relay/live.test.ts index 5de753d6..1a57032b 100644 --- a/src/features/relay/live.test.ts +++ b/src/features/relay/live.test.ts @@ -717,7 +717,10 @@ it("admits signed typing only on its authenticated channel route, without extra await h.first.receive(["EVENT", requests[3]?.[1], event]); expect(h.callbacks.receive).not.toHaveBeenCalled(); await h.first.receive(["EVENT", route[1], event]); - expect(h.callbacks.receive).toHaveBeenCalledExactlyOnceWith([event]); + expect(h.callbacks.receive).toHaveBeenCalledExactlyOnceWith([event], { + channelId: "a", + phase: "replay", + }); await h.first.receive([ "EVENT", route[1], diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index c54d3fc4..f6a1cc23 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -993,7 +993,13 @@ export function createRelaySession( events.filter((event) => event.kind === 20002), true, ); - if (closed || epoch !== accessEpoch || !candidates.size || !provenance?.channelId) return; + if ( + closed || + epoch !== accessEpoch || + !candidates.size || + !provenance?.channelId + ) + return; const delivered = new Set(); const incoming: readonly IncomingMessage[] = Object.freeze( visible.flatMap((event) => { diff --git a/src/features/relay/typing.integration.test.ts b/src/features/relay/typing.integration.test.ts index 11be6144..b8f5cefa 100644 --- a/src/features/relay/typing.integration.test.ts +++ b/src/features/relay/typing.integration.test.ts @@ -207,3 +207,69 @@ it("refreshing a finite kind-20002 view neither retains nor activates typing", a owner.dispose(); } }); + +it.each(["none", "cache", "access", "dispose"] as const)( + "preserves incoming notifications alongside typing and fences reentrant %s", + async (transition) => { + vi.useFakeTimers(); + vi.setSystemTime(1_800_000_000_000); + const viewer = keypair(), + relay = keypair(), + agent = keypair(), + peer = keypair(); + const wire = scriptedTransport(viewer.pubkey, relay.pubkey); + let live!: LiveCallbacks; + const owner = createRelaySession({ + ...wire.transport, + subscribe(callbacks) { + live = callbacks; + return { update() {}, retry() {}, dispose() {} }; + }, + }); + let clearing: Promise | undefined; + try { + live.state({ status: "connected", routes: [] }); + live.receive([roster(relay, "a", [viewer.pubkey])]); + const incoming = vi.fn(); + owner.session.subscribeIncoming(incoming); + owner.session.typing.subscribe(() => { + if (!owner.session.typing.snapshot().length) return; + if (transition === "cache") clearing = owner.clearCache(); + else if (transition === "access") + live.receive([roster(relay, "a", [], 1_800_000_001)]); + else if (transition === "dispose") owner.dispose(); + }); + const message = signed(peer, { + kind: 9, + content: "fixture", + created_at: 1_800_000_000, + tags: [["h", "a"]], + }); + const pulse = signed(agent, { + kind: 20002, + content: "", + created_at: 1_800_000_000, + tags: [["h", "a"]], + }); + live.receive([message, pulse], { phase: "live", channelId: "a" }); + await clearing; + if (transition === "none") { + expect(owner.session.typing.snapshot()).toHaveLength(1); + expect(incoming).toHaveBeenCalledExactlyOnceWith([ + expect.objectContaining({ + messageId: message.id, + authorId: peer.pubkey, + }), + ]); + live.receive([pulse], { phase: "live", channelId: "a" }); + expect(incoming).toHaveBeenCalledTimes(1); + } else { + expect(owner.session.typing.snapshot()).toEqual([]); + expect(incoming).not.toHaveBeenCalled(); + } + } finally { + owner.dispose(); + } + expect(vi.getTimerCount()).toBe(0); + }, +); From f26cfbfbd7614ac8ca0ab971a33170068c35f2f6 Mon Sep 17 00:00:00 2001 From: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> Date: Tue, 15 Sep 2026 10:02:40 +0200 Subject: [PATCH 7/7] test: establish thread bottom before typing assertions Signed-off-by: pic-worker-aa352576c7 <6838876f3610dc213d19edeb1cdf5a8d0a0b568dc9b40db674ade16725074cce@buzz.block.builderlab.xyz> --- tests/browser/typing.spec.mjs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/browser/typing.spec.mjs b/tests/browser/typing.spec.mjs index aea07584..ca282767 100644 --- a/tests/browser/typing.spec.mjs +++ b/tests/browser/typing.spec.mjs @@ -95,6 +95,13 @@ for (const scope of ["channel", "thread"]) { history.evaluate( (el) => el.scrollHeight - el.clientHeight - el.scrollTop, ); + if (scope === "thread") { + // Opening can race the final signed fixture replies under parallel load. + // Establish the bottom-reading precondition with real browser input before + // capturing geometry; the assertions below verify typing keeps it there. + await history.hover(); + await page.mouse.wheel(0, Math.max(1, await gap())); + } await expect.poll(gap).toBeLessThan(2); expect( await history.evaluate((el) => el.scrollHeight - el.clientHeight),