From fad5f047a6109d07bc62d0eecda060848caec86f Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:13:11 +0100 Subject: [PATCH 01/11] feat(schema-shared): declare the ephemeral port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sibling seam to dataSource.ts, for state that is lossy, last-write-wins, and gone on reload: live presence, WebRTC signalling, cursors, work claims. Deliberately not part of QueryIR. The IR assumes durable entities in a model manifest, each with a globally-stable id; ephemeral state has none of those. The rule the port exists to enforce: if it must still be there after a refresh, it is not ephemeral — so durable messaging stays with DataSource. Shape is ephemeral(dataset).channel(tag). Dataset scoping because a call happens *in* a space; tag namespacing so a feature module gets a namespace it cannot collide with. Payloads are opaque, so presence, RTC, and cursors share one pipe without the transport knowing about any of them. The tag maps onto a Socket.io event, a Supabase channel topic, a Matrix to-device type, or a gossipsub topic. unicast is a tri-state rather than a boolean because 'emulated' (addressed-but-broadcast, every peer receives it) and 'native' differ in a way that matters for security, not efficiency. As a boolean, someone eventually builds a private feature on `unicast: true` and it silently is not private. authenticatedSender is declared for the same reason: AD4M supplies link.author, but a naive relay carries the sender in the payload where it is forgeable, and a work-claim lease must be able to tell the difference. planEphemeral mirrors planQuery — fail loudly at registration rather than mounting a consumer that cannot work. Co-Authored-By: Claude Opus 5 (1M context) --- .../shared/src/ephemeral.test.ts | 83 ++++++++ .../schema-system/shared/src/ephemeral.ts | 177 ++++++++++++++++++ packages/schema-system/shared/src/index.ts | 10 + 3 files changed, 270 insertions(+) create mode 100644 packages/schema-system/shared/src/ephemeral.test.ts create mode 100644 packages/schema-system/shared/src/ephemeral.ts diff --git a/packages/schema-system/shared/src/ephemeral.test.ts b/packages/schema-system/shared/src/ephemeral.test.ts new file mode 100644 index 00000000..e42b9023 --- /dev/null +++ b/packages/schema-system/shared/src/ephemeral.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it } from 'vitest'; + +import { type EphemeralCapabilities, planEphemeral } from './ephemeral'; + +/** AD4M today: real unicast exists upstream but `sendSignalU` is broken, so the adapter emulates it. */ +const ad4m: EphemeralCapabilities = { + fanout: true, + unicast: 'emulated', + reliability: 'send-acked', + heartbeatRequired: true, + authenticatedSender: true, +}; + +/** A server-backed transport (Socket.io, Supabase Realtime): knows who is connected, routes directly. */ +const server: EphemeralCapabilities = { + fanout: true, + unicast: 'native', + reliability: 'at-least-once', + heartbeatRequired: false, + authenticatedSender: true, +}; + +/** Yjs awareness: fan-out only, no addressing at all. */ +const awareness: EphemeralCapabilities = { + fanout: true, + unicast: 'none', + reliability: 'best-effort', + heartbeatRequired: true, + authenticatedSender: false, +}; + +describe('planEphemeral', () => { + it('lets presence run on every transport — it only needs fan-out', () => { + const req = { consumer: 'presence' }; + for (const cap of [ad4m, server, awareness]) { + expect(planEphemeral(req, cap).runnable).toBe(true); + } + }); + + it('runs a call module on AD4M, where emulated addressing is good enough for SDP/ICE', () => { + const plan = planEphemeral({ consumer: 'call module', unicast: 'emulated' }, ad4m); + expect(plan).toEqual({ runnable: true, gaps: [] }); + }); + + it('refuses a call module on a fan-out-only transport instead of hanging on the handshake', () => { + const plan = planEphemeral({ consumer: 'call module', unicast: 'emulated' }, awareness); + expect(plan.runnable).toBe(false); + expect(plan.gaps[0].feature).toBe('unicast'); + expect(plan.gaps[0].note).toContain('call module'); + }); + + it('treats native as satisfying a request for emulated, but not the reverse', () => { + expect(planEphemeral({ consumer: 'c', unicast: 'emulated' }, server).runnable).toBe(true); + expect(planEphemeral({ consumer: 'c', unicast: 'native' }, ad4m).runnable).toBe(false); + }); + + it('rejects confidential payloads over emulated addressing — addressing is not privacy', () => { + const plan = planEphemeral({ consumer: 'private notes', unicast: 'emulated', confidential: true }, ad4m); + expect(plan.runnable).toBe(false); + expect(plan.gaps.map((g) => g.feature)).toContain('unicast:confidential'); + expect(plan.gaps.find((g) => g.feature === 'unicast:confidential')?.note).toContain('not privacy'); + }); + + it('allows confidential payloads only over native unicast', () => { + expect(planEphemeral({ consumer: 'private notes', confidential: true }, server).runnable).toBe(true); + }); + + it('refuses to act on sender identity when the sender is self-asserted', () => { + const req = { consumer: 'work lease', requiresAuthenticatedSender: true }; + expect(planEphemeral(req, ad4m).runnable).toBe(true); + const plan = planEphemeral(req, awareness); + expect(plan.runnable).toBe(false); + expect(plan.gaps[0].feature).toBe('authenticatedSender'); + }); + + it('reports every unmet requirement at once, not just the first', () => { + const plan = planEphemeral( + { consumer: 'everything', unicast: 'native', confidential: true, requiresAuthenticatedSender: true }, + awareness, + ); + expect(plan.gaps.map((g) => g.feature).sort()).toEqual(['authenticatedSender', 'unicast', 'unicast:confidential']); + }); +}); diff --git a/packages/schema-system/shared/src/ephemeral.ts b/packages/schema-system/shared/src/ephemeral.ts new file mode 100644 index 00000000..77b2e6ca --- /dev/null +++ b/packages/schema-system/shared/src/ephemeral.ts @@ -0,0 +1,177 @@ +/** + * The ephemeral seam — WE's renderer ↔ backend contract for **transient** agent-to-agent state. + * + * Sibling to `dataSource.ts`. Where that port carries durable, queryable records, this one carries + * state that is lossy, last-write-wins, and gone on reload: live presence, WebRTC signalling, cursors, + * typing indicators, work claims. + * + * **Why it is not part of `QueryIR`.** The IR assumes durable entities declared in a model manifest, + * each with a globally-stable `id`. Ephemeral state has none of those: no manifest entity to name, no + * id that survives a restart, no persistence to query. Routing it through `$query` would either + * corrupt the IR's guarantees or require inventing a phantom entity type. The test, applied every time + * something is proposed for this port: + * + * > **If it must still be there after a refresh, it is not ephemeral.** + * + * Durable messaging (chat, DMs) therefore belongs to `DataSource`, never here — see + * notes/we/August-2026/presence-port.md. + * + * **Shape.** A scope is bound to one dataset (a space, a DM neighbourhood); within a scope, each + * protocol takes a named `channel(tag)`. Both are load-bearing: a call happens *in* a space, and a + * third-party feature module needs a namespace it cannot collide with. The payload is deliberately + * opaque — presence, RTC, and cursors all ride the same pipe, so the transport must not know about any + * of them. + * + * The tag maps cleanly onto every real backend: a Socket.io event name, a Supabase Realtime channel + * topic, a Matrix to-device message type, a gossipsub topic, a Yjs awareness field. + */ +import type { DatasetHandle } from './dataSource'; + +/** + * What a backend can do natively. Declared by the adapter, consumed by {@link planEphemeral} so a + * consumer that needs more than the backend offers fails loudly rather than silently misbehaving. + */ +export interface EphemeralCapabilities { + /** + * Send to everyone in the dataset. Always true — a transport that cannot fan out cannot carry + * presence, and is better modelled as absent (see {@link EphemeralPort} returning null). + */ + fanout: true; + + /** + * Send to one named peer. + * + * Tri-state rather than boolean **on purpose**, because `emulated` and `native` differ in a way that + * matters for security, not just efficiency: + * + * - `native` — the transport routes to that peer alone. + * - `emulated` — addressed but *broadcast*; every peer receives the payload and is trusted to + * discard it. The adapter filters on receipt. Acceptable for SDP/ICE (not secret in most threat + * models); **never** acceptable for anything confidential. + * - `none` — no addressing at all (e.g. Yjs awareness is fan-out only). + * + * Were this a boolean, a consumer would eventually build a private feature on `unicast: true` and it + * would silently not be private. Confidentiality comes from membrane or encryption, never from + * addressing. + */ + unicast: 'native' | 'emulated' | 'none'; + + /** + * - `best-effort` — fire and forget, no confirmation of anything. + * - `send-acked` — the send is confirmed, delivery is not (AD4M: `Promise`). + * - `at-least-once` — delivery is confirmed, possibly more than once. + */ + reliability: 'best-effort' | 'send-acked' | 'at-least-once'; + + /** + * True when peers must gossip their own liveness on a timer because the transport cannot report + * disconnects. P1/P2 backends (AD4M) set this; a P3 backend that knows who is connected sets false + * and implements a presence source directly instead of using `createHeartbeatPresence`. + */ + heartbeatRequired: boolean; + + /** + * Whether the `from` in {@link EphemeralChannel.onMessage} is asserted by the **transport** or by + * the payload. + * + * AD4M supplies `link.author`, so identity is authenticated: a peer cannot impersonate another in + * the presence map, and a work claim can be trusted enough to act on. A naive relay or unsigned + * pubsub carries the sender id in the payload, where it is forgeable — a host could implement this + * port entirely correctly and still ship spoofable presence and hijackable leases. Anything + * security-relevant (leases, moderation, call admission) must check this rather than assume it. + */ + authenticatedSender: boolean; +} + +/** A protocol's namespaced slice of a dataset's ephemeral traffic. */ +export interface EphemeralChannel { + /** + * Fire-and-forget. Passing `to.agentId` requires `unicast !== 'none'`; under `'emulated'` the + * payload still reaches every peer, so treat it as addressing, not privacy. + */ + publish(payload: unknown, to?: { agentId?: string }): void; + + /** + * Subscribe to this channel. `from` is the sender's id — see + * {@link EphemeralCapabilities.authenticatedSender} before trusting it. Returns an unsubscribe. + */ + onMessage(cb: (from: string, payload: unknown) => void): () => void; +} + +/** Ephemeral traffic for one dataset. Obtain channels from it; dispose to detach from the backend. */ +export interface EphemeralScope { + capabilities: EphemeralCapabilities; + /** Namespaced sub-channel. Repeated calls with the same tag return the same channel. */ + channel(tag: string): EphemeralChannel; + dispose(): void; +} + +/** + * The port a host injects. Returns `null` for a dataset with no transport — a personal (unshared) + * space has no neighbourhood, so there is nobody to signal. Consumers must degrade rather than throw. + */ +export type EphemeralPort = (dataset: DatasetHandle) => EphemeralScope | null; + +/** What a consumer (presence, a call module, a cursor overlay) needs from the transport. */ +export interface EphemeralRequirements { + /** Human name of the consumer, used in the failure message. */ + consumer: string; + /** Minimum addressing this consumer needs. `'emulated'` accepts emulated or native. */ + unicast?: 'native' | 'emulated'; + /** True when this consumer carries confidential payloads and so cannot accept emulated addressing. */ + confidential?: boolean; + /** True when this consumer acts on the sender's identity (leases, admission control). */ + requiresAuthenticatedSender?: boolean; +} + +export interface EphemeralGap { + /** Greppable feature name, e.g. "unicast", "unicast:confidential", "authenticatedSender". */ + feature: string; + note: string; +} + +export interface EphemeralPlan { + /** False if any requirement is unmet — the consumer must not be mounted. */ + runnable: boolean; + gaps: EphemeralGap[]; +} + +const UNICAST_RANK = { none: 0, emulated: 1, native: 2 } as const; + +/** + * Classify a consumer's requirements against a backend's capabilities. + * + * Mirrors `planQuery` in `queryCapabilities.ts`: the point is to **fail loudly at registration** + * rather than mount a feature that silently cannot work. A Yjs-backed host, for instance, has + * `unicast: 'none'` — a call module asking for `'emulated'` gets a clear refusal instead of a + * handshake that never completes. + */ +export function planEphemeral(req: EphemeralRequirements, cap: EphemeralCapabilities): EphemeralPlan { + const gaps: EphemeralGap[] = []; + + if (req.unicast && UNICAST_RANK[cap.unicast] < UNICAST_RANK[req.unicast]) { + gaps.push({ + feature: 'unicast', + note: `${req.consumer} needs unicast "${req.unicast}" but the backend offers "${cap.unicast}"`, + }); + } + + // Emulated addressing is broadcast-plus-filter, so it provides no confidentiality at all. + if (req.confidential && cap.unicast !== 'native') { + gaps.push({ + feature: 'unicast:confidential', + note: + `${req.consumer} carries confidential payloads, which requires native unicast; ` + + `the backend offers "${cap.unicast}" (addressing is not privacy)`, + }); + } + + if (req.requiresAuthenticatedSender && !cap.authenticatedSender) { + gaps.push({ + feature: 'authenticatedSender', + note: `${req.consumer} acts on sender identity, but this backend's sender id is self-asserted and forgeable`, + }); + } + + return { runnable: gaps.length === 0, gaps }; +} diff --git a/packages/schema-system/shared/src/index.ts b/packages/schema-system/shared/src/index.ts index 94b28d17..d5d8c4b2 100644 --- a/packages/schema-system/shared/src/index.ts +++ b/packages/schema-system/shared/src/index.ts @@ -127,6 +127,16 @@ export type { RendererDataBindings, RendererStores, } from './dataSource'; +export { planEphemeral } from './ephemeral'; +export type { + EphemeralCapabilities, + EphemeralChannel, + EphemeralScope, + EphemeralPort, + EphemeralRequirements, + EphemeralGap, + EphemeralPlan, +} from './ephemeral'; export { modelManifestSchema, validateManifest, getEntity, getProperty, getRelation } from './manifest'; export type { ModelManifest, From 8aba583ae40a238c5595dba1d59fb8e433c0ff48 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:13:27 +0100 Subject: [PATCH 02/11] feat(schema-shared): neutral presence over the ephemeral port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Liveness derivation, the heartbeat driver, and the selectors — all backend-agnostic and DOM-free, so they unit-test against a fake channel with no executor. Flux welds the equivalent into one AD4M-coupled composable that also owns profile hydration and an unrelated WebRTC protocol; splitting them is the point. Two shape decisions carry the design: focus is one value, activities is a list. A call is not a place you are instead of the Kanban board — it is something you are in while also being somewhere. Routes are exclusive, participation is not. Flux found this and modelled it flatly (currentRoute + callRoute + inCall + mediaSettings); as a list, concurrent calls in one space fall out for free, and a feature module adds an activity type without touching the schema. availability (declared) and liveness (measured) are separate. Flux fuses them into one union and carries a standing TODO about it. They are orthogonal: an agent can be busy and online, or available and stale. Also lifted from Flux, because they are load-bearing: adaptive scheduling (a change publishes immediately and pushes the next tick out a full interval, so navigating quickly does not double-send) and the join handshake (a joiner sets hello, peers re-announce at once, otherwise a joiner waits out a whole interval against a stateless lossy transport). Added: TTL eviction, which Flux lacks — its agent map only ever relabels to offline and grows forever. invisible stops publishing entirely rather than filtering on receipt. Flux keeps broadcasting full state including route when invisible and hides it client-side, so any modified client sees invisible agents and where they are. peersMatching takes a partial focus rather than offering peersAtPath: a route path is only meaningful within a dataset, and two spaces routinely share one. The tests caught exactly that unioning peers across spaces — a bug invisible in single-space testing, the same class as broadcasting a local uuid. Co-Authored-By: Claude Opus 5 (1M context) --- packages/schema-system/shared/src/index.ts | 25 ++ .../schema-system/shared/src/presence.test.ts | 340 ++++++++++++++ packages/schema-system/shared/src/presence.ts | 414 ++++++++++++++++++ 3 files changed, 779 insertions(+) create mode 100644 packages/schema-system/shared/src/presence.test.ts create mode 100644 packages/schema-system/shared/src/presence.ts diff --git a/packages/schema-system/shared/src/index.ts b/packages/schema-system/shared/src/index.ts index d5d8c4b2..5764c869 100644 --- a/packages/schema-system/shared/src/index.ts +++ b/packages/schema-system/shared/src/index.ts @@ -137,6 +137,31 @@ export type { EphemeralGap, EphemeralPlan, } from './ephemeral'; +export { + applyFocusDepth, + activitiesOfType, + callRosters, + createHeartbeatPresence, + derivePeers, + peersInDataset, + peersMatching, + DEFAULT_HEARTBEAT_INTERVAL, + DEFAULT_THRESHOLDS, +} from './presence'; +export type { + Activity, + Availability, + Focus, + FocusDepth, + HeartbeatOptions, + Liveness, + LivenessThresholds, + MediaSettings, + Peer, + PresenceChannel, + PresenceSource, + PresenceState, +} from './presence'; export { modelManifestSchema, validateManifest, getEntity, getProperty, getRelation } from './manifest'; export type { ModelManifest, diff --git a/packages/schema-system/shared/src/presence.test.ts b/packages/schema-system/shared/src/presence.test.ts new file mode 100644 index 00000000..dd8666a0 --- /dev/null +++ b/packages/schema-system/shared/src/presence.test.ts @@ -0,0 +1,340 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + type Activity, + applyFocusDepth, + callRosters, + createHeartbeatPresence, + DEFAULT_THRESHOLDS, + derivePeers, + type Peer, + peersInDataset, + peersMatching, + type PresenceChannel, + type PresenceState, +} from './presence'; + +const SPACE = 'neighbourhood://QmSpace'; + +function state(agentId: string, overrides: Partial = {}): PresenceState { + return { + agentId, + updatedAt: 0, + availability: 'available', + focus: { datasetUri: SPACE, path: '/kanban' }, + ...overrides, + }; +} + +/** + * A fan-out channel with no transport. `deliver` plays a message into every subscriber except the + * named sender, which is what a real broadcast does (AD4M's `sendBroadcastU` does not loop back). + */ +function createFakeChannel() { + const subscribers: Array<(from: string, payload: unknown) => void> = []; + const published: unknown[] = []; + return { + published, + channel: { + publish: (payload: unknown) => published.push(payload), + onMessage: (cb) => { + subscribers.push(cb); + return () => { + const i = subscribers.indexOf(cb); + if (i !== -1) subscribers.splice(i, 1); + }; + }, + } satisfies PresenceChannel, + deliver(from: string, payload: unknown) { + subscribers.forEach((cb) => cb(from, payload)); + }, + }; +} + +describe('derivePeers', () => { + it('maps age onto the liveness ladder', () => { + const now = 100_000; + const peers = derivePeers( + [ + state('online', { updatedAt: now - 1_000 }), + state('idle', { updatedAt: now - 20_000 }), + state('stale', { updatedAt: now - 45_000 }), + state('offline', { updatedAt: now - 120_000 }), + ], + now, + ); + expect(peers.map((p) => [p.agentId, p.liveness])).toEqual([ + ['online', 'online'], + ['idle', 'idle'], + ['stale', 'stale'], + ['offline', 'offline'], + ]); + }); + + it('evicts past evictAfter rather than keeping them forever', () => { + const now = 1_000_000; + const peers = derivePeers( + [state('recent', { updatedAt: now - 1_000 }), state('ancient', { updatedAt: now - 600_000 })], + now, + ); + expect(peers.map((p) => p.agentId)).toEqual(['recent']); + }); + + it('is pure — same inputs, same output', () => { + const input = [state('a', { updatedAt: 90_000 })]; + expect(derivePeers(input, 100_000)).toEqual(derivePeers(input, 100_000)); + expect(input[0].updatedAt).toBe(90_000); + }); + + it('treats boundaries as exclusive-below', () => { + const now = 0; + const at = (age: number) => derivePeers([state('a', { updatedAt: -age })], now)[0]?.liveness; + expect(at(DEFAULT_THRESHOLDS.idleAfter - 1)).toBe('online'); + expect(at(DEFAULT_THRESHOLDS.idleAfter)).toBe('idle'); + expect(at(DEFAULT_THRESHOLDS.staleAfter)).toBe('stale'); + expect(at(DEFAULT_THRESHOLDS.offlineAfter)).toBe('offline'); + expect(at(DEFAULT_THRESHOLDS.evictAfter)).toBeUndefined(); + }); +}); + +describe('applyFocusDepth', () => { + const focus = { datasetUri: SPACE, path: '/budget', nodeId: 'post-1' }; + + it('discloses progressively', () => { + expect(applyFocusDepth(focus, 'off')).toBeUndefined(); + expect(applyFocusDepth(focus, 'space')).toEqual({ datasetUri: SPACE }); + expect(applyFocusDepth(focus, 'route')).toEqual({ datasetUri: SPACE, path: '/budget' }); + expect(applyFocusDepth(focus, 'precise')).toEqual(focus); + }); + + it('never leaks a deeper field than the depth allows', () => { + expect(applyFocusDepth(focus, 'space')).not.toHaveProperty('path'); + expect(applyFocusDepth(focus, 'route')).not.toHaveProperty('nodeId'); + }); +}); + +describe('selectors', () => { + const now = 0; + const peers: Peer[] = derivePeers( + [ + state('here', { focus: { datasetUri: SPACE, path: '/kanban', nodeId: 'card-1' } }), + state('elsewhere-in-space', { focus: { datasetUri: SPACE, path: '/docs' } }), + state('other-space', { focus: { datasetUri: 'neighbourhood://QmOther', path: '/kanban' } }), + state('gone', { updatedAt: -90_000, focus: { datasetUri: SPACE, path: '/kanban' } }), + ], + now, + ); + + it('slices by dataset, path, and node from one hierarchical focus', () => { + expect(peersInDataset(peers, SPACE).map((p) => p.agentId)).toEqual(['here', 'elsewhere-in-space']); + expect(peersMatching(peers, { datasetUri: SPACE, path: '/kanban' }).map((p) => p.agentId)).toEqual(['here']); + expect(peersMatching(peers, { datasetUri: SPACE, nodeId: 'card-1' }).map((p) => p.agentId)).toEqual(['here']); + }); + + it('does not union the same route path across different spaces', () => { + // Two spaces can both have a "/kanban" — matching a path without its dataset is a real bug. + expect(peersMatching(peers, { path: '/kanban' }).map((p) => p.agentId)).toEqual(['here', 'other-space']); + expect(peersMatching(peers, { datasetUri: SPACE, path: '/kanban' }).map((p) => p.agentId)).toEqual(['here']); + }); + + it('excludes offline peers unless asked', () => { + const at = (includeOffline?: boolean) => + peersMatching(peers, { datasetUri: SPACE, path: '/kanban' }, includeOffline).map((p) => p.agentId); + expect(at()).not.toContain('gone'); + expect(at(true)).toContain('gone'); + }); +}); + +describe('callRosters', () => { + it('groups concurrent calls by id — something a single callRoute field cannot express', () => { + const call = (id: string): Activity => ({ type: 'call', id }); + const peers = derivePeers( + [ + state('a', { activities: [call('c1')] }), + state('b', { activities: [call('c1')] }), + state('c', { activities: [call('c2')] }), + state('d', { activities: [] }), + ], + 0, + ); + const rosters = callRosters(peers); + expect(rosters.get('c1')?.map((p) => p.agentId)).toEqual(['a', 'b']); + expect(rosters.get('c2')?.map((p) => p.agentId)).toEqual(['c']); + expect(rosters.size).toBe(2); + }); + + it('keeps a peer in a call regardless of where their focus is', () => { + const peers = derivePeers( + [ + state('on-kanban', { focus: { datasetUri: SPACE, path: '/kanban' }, activities: [{ type: 'call', id: 'c1' }] }), + state('on-docs', { focus: { datasetUri: SPACE, path: '/docs' }, activities: [{ type: 'call', id: 'c1' }] }), + ], + 0, + ); + // The point of activities-not-routes: navigating away does not leave the call. + expect(callRosters(peers).get('c1')).toHaveLength(2); + expect(peersMatching(peers, { datasetUri: SPACE, path: '/kanban' })).toHaveLength(1); + }); +}); + +describe('createHeartbeatPresence', () => { + let clock: number; + const now = () => clock; + + beforeEach(() => { + clock = 0; + vi.useFakeTimers(); + }); + + function start(overrides: Partial = {}) { + const fake = createFakeChannel(); + const source = createHeartbeatPresence(fake.channel, { now }); + source.start(state('me', overrides)); + return { ...fake, source }; + } + + it('sends a hello on start so peers re-announce', () => { + const { published } = start(); + expect(published).toEqual([{ v: 1, state: expect.objectContaining({ agentId: 'me' }), hello: true }]); + }); + + it('answers a peer hello immediately, without itself saying hello', () => { + const { deliver, published, source } = start(); + published.length = 0; + + deliver('peer', { v: 1, state: state('peer'), hello: true }); + + expect(published).toHaveLength(1); + expect(published[0]).not.toHaveProperty('hello'); + expect(source.peers().map((p) => p.agentId)).toContain('peer'); + }); + + it('heartbeats on the interval', () => { + const { published } = start(); + published.length = 0; + + clock += 5_000; + vi.advanceTimersByTime(5_000); + expect(published).toHaveLength(1); + + clock += 5_000; + vi.advanceTimersByTime(5_000); + expect(published).toHaveLength(2); + }); + + it('does not double-send when a change already published inside the window', () => { + const { published, source } = start(); + published.length = 0; + + clock += 4_000; + source.update({ focus: { datasetUri: SPACE, path: '/docs' } }); + expect(published).toHaveLength(1); // the change itself + + // The tick that was due at 5s must wait out a full interval from the change, not fire at once. + clock += 1_000; + vi.advanceTimersByTime(1_000); + expect(published).toHaveLength(1); + + clock += 4_000; + vi.advanceTimersByTime(4_000); + expect(published).toHaveLength(2); + }); + + it('stops publishing entirely when invisible, rather than filtering on receipt', () => { + const { published, source } = start(); + published.length = 0; + + source.update({ availability: 'invisible' }); + clock += 5_000; + vi.advanceTimersByTime(5_000); + clock += 5_000; + vi.advanceTimersByTime(5_000); + + expect(published).toHaveLength(0); + }); + + it('keys peer state by the transport-supplied sender, not the payload', () => { + const { deliver, source } = start(); + + // A peer claiming to be someone else must not be able to write into that agent's slot. + deliver('actual-sender', { v: 1, state: state('claimed-victim') }); + + const ids = source.peers().map((p) => p.agentId); + expect(ids).toContain('actual-sender'); + expect(ids).not.toContain('claimed-victim'); + }); + + it('ignores malformed and unversioned payloads', () => { + const { deliver, source } = start(); + const before = source.peers().length; + + deliver('peer', null); + deliver('peer', 'not an object'); + deliver('peer', { v: 2, state: state('peer') }); + deliver('peer', { v: 1 }); + + expect(source.peers()).toHaveLength(before); + }); + + it('evicts a peer that stops heartbeating', () => { + const { deliver, source } = start(); + deliver('peer', { v: 1, state: state('peer') }); + expect(source.peers().map((p) => p.agentId)).toContain('peer'); + + clock += DEFAULT_THRESHOLDS.offlineAfter + 1; + expect(source.peers().find((p) => p.agentId === 'peer')?.liveness).toBe('offline'); + + clock += DEFAULT_THRESHOLDS.evictAfter; + expect(source.peers().map((p) => p.agentId)).not.toContain('peer'); + }); + + it('replaces an activity of the same type and id, and clears it on demand', () => { + const { source } = start(); + + source.setActivity({ + type: 'call', + id: 'c1', + media: { audioEnabled: true, videoEnabled: false, screenShareEnabled: false }, + }); + source.setActivity({ + type: 'call', + id: 'c1', + media: { audioEnabled: false, videoEnabled: true, screenShareEnabled: false }, + }); + + const me = () => source.peers().find((p) => p.agentId === 'me')!; + expect(me().activities).toHaveLength(1); + expect(me().activities?.[0]).toMatchObject({ media: { videoEnabled: true } }); + + source.setActivity({ type: 'call', id: 'c2' }); + expect(me().activities).toHaveLength(2); + + source.clearActivity('call', 'c1'); + expect(me().activities?.map((a) => ('id' in a ? a.id : undefined))).toEqual(['c2']); + + source.clearActivity('call'); + expect(me().activities).toHaveLength(0); + }); + + it('an activity survives a focus change — a call is not a location', () => { + const { source } = start(); + source.setActivity({ type: 'call', id: 'c1' }); + source.update({ focus: { datasetUri: SPACE, path: '/docs' } }); + + const me = source.peers().find((p) => p.agentId === 'me')!; + expect(me.focus?.path).toBe('/docs'); + expect(me.activities).toHaveLength(1); + }); + + it('stops cleanly and publishes nothing afterwards', () => { + const { published, source } = start(); + source.stop(); + published.length = 0; + + clock += 60_000; + vi.advanceTimersByTime(60_000); + + expect(published).toHaveLength(0); + expect(source.peers()).toEqual([]); + }); +}); diff --git a/packages/schema-system/shared/src/presence.ts b/packages/schema-system/shared/src/presence.ts new file mode 100644 index 00000000..3cfb533d --- /dev/null +++ b/packages/schema-system/shared/src/presence.ts @@ -0,0 +1,414 @@ +/** + * Presence — the first consumer of the {@link EphemeralChannel} port. + * + * Everything here is backend-agnostic and DOM-free: given a channel, it gossips this agent's state on + * a timer and decays peers who go quiet. The only backend-specific piece is the transport itself. + * (Flux welds these together — its `useSignallingService` owns the timers, the decay ladder, the + * handshake, profile hydration, *and* an unrelated WebRTC protocol, all inside one AD4M-coupled + * composable. Splitting them is the point.) + * + * Two shape decisions carry the design; see notes/we/August-2026/presence-port.md for the reasoning. + * + * 1. **`focus` is one value; `activities` is a list.** A call is not a place you are *instead of* the + * Kanban board — it is something you are *in* while also being somewhere. Routes are exclusive; + * participation is not. Flux discovered this and modelled it flatly (`currentRoute` + `callRoute` + + * `inCall` + `mediaSettings`); this is the same insight, generalised, so a feature module can add + * an activity type without touching the schema. + * 2. **`availability` (declared) and `liveness` (measured) are separate.** Flux fuses them into one + * union and carries a standing TODO about it. They are orthogonal — an agent can be `busy` and + * `online`, or `available` and `stale`. + * + * Profile hydration deliberately lives elsewhere: presence carries `agentId` only, and the UI resolves + * it through the host's `$identities` directory. Flux re-fetches every peer profile on every heartbeat + * because its agent map and its profile cache are the same object; keeping them apart avoids that + * entirely. + */ + +/** Where an agent is. Hierarchical so consumers can slice at whichever depth they render. */ +export interface Focus { + /** + * The dataset, by **global** uri — never a local handle id. + * + * AD4M perspective uuids are local per-agent: the same shared neighbourhood has a different uuid on + * every peer, so a broadcast uuid is meaningless to whoever receives it. This is the local-id vs + * global-uri split from `DatasetHandle`, one layer down, and getting it wrong half-works — correct + * in single-agent testing, silently broken across peers. + */ + datasetUri?: string; + /** Route path within the template, e.g. "/kanban". */ + path?: string; + /** Optional finer grain — which post, card, or block. Enables "3 people are viewing this". */ + nodeId?: string; +} + +export interface MediaSettings { + audioEnabled: boolean; + videoEnabled: boolean; + screenShareEnabled: boolean; +} + +/** + * What an agent is participating in. Open-ended by design: a feature module contributes a variant + * rather than extending {@link PresenceState}. + * + * Activities inherit presence's self-healing property. When an agent's laptop closes, its heartbeat + * stops, it is evicted on TTL, and its activities go with it — no leave message required. A + * message-based roster ("Ana left") breaks on crash, tab close, and network partition, leaving ghosts + * in the call and work permanently claimed by a dead peer. + */ +export type Activity = + /** In a call. `anchor` is what the call is *about* ("the call on the Kanban"); absent = space-wide. */ + | { type: 'call'; id: string; anchor?: Focus; media?: MediaSettings } + /** Has the composer open on a node. */ + | { type: 'edit'; nodeId: string } + /** Typing into a node. */ + | { type: 'typing'; nodeId: string } + /** Claiming a unit of peer-elected work (AI processing, indexing) — a self-releasing lease. */ + | { type: 'processing'; anchor?: Focus; step?: number } + /** Escape hatch for feature-module activity types not known to this package. */ + | { type: string; [key: string]: unknown }; + +/** Declared by the user, as opposed to {@link Liveness}, which is measured. */ +export type Availability = 'available' | 'busy' | 'away' | 'invisible'; + +/** What actually travels. */ +export interface PresenceState { + /** DID, or whatever the host's `$identities` directory keys on. */ + agentId: string; + /** ms epoch, stamped by the sender. */ + updatedAt: number; + availability: Availability; + /** Omitted entirely when the agent hides their location — see {@link FocusDepth}. */ + focus?: Focus; + activities?: Activity[]; + /** Open payload for module state that is *not* participation (preferences, transient hints). */ + custom?: Record; +} + +/** Derived locally from `updatedAt`. Never transmitted. */ +export type Liveness = 'online' | 'idle' | 'stale' | 'offline'; + +export interface Peer extends PresenceState { + liveness: Liveness; +} + +/** + * How much of {@link Focus} to publish — a progressive-disclosure privacy control. + * + * - `off` — "this agent is online", nothing more. + * - `space` — "…is in this space". + * - `route` — "…is looking at the budget page". + * - `precise` — "…is reading your post". + * + * Distinct from `availability: 'invisible'`, which is the stronger control that stops publishing + * altogether. + */ +export type FocusDepth = 'off' | 'space' | 'route' | 'precise'; + +export interface LivenessThresholds { + /** ms since `updatedAt` after which a peer is `idle`. */ + idleAfter: number; + /** ms after which a peer is `stale`. */ + staleAfter: number; + /** ms after which a peer is `offline`. */ + offlineAfter: number; + /** ms after which a peer is dropped entirely. Flux never evicts, so its agent map grows forever. */ + evictAfter: number; +} + +/** Tuned from Flux's ladder (5s heartbeat / 30s asleep / 60s offline), with eviction added. */ +export const DEFAULT_THRESHOLDS: LivenessThresholds = { + idleAfter: 15_000, + staleAfter: 30_000, + offlineAfter: 60_000, + evictAfter: 300_000, +}; + +export const DEFAULT_HEARTBEAT_INTERVAL = 5_000; + +/** Trim a focus to the depth the agent has consented to publish. */ +export function applyFocusDepth(focus: Focus | undefined, depth: FocusDepth): Focus | undefined { + if (!focus || depth === 'off') return undefined; + if (depth === 'space') return { datasetUri: focus.datasetUri }; + if (depth === 'route') return { datasetUri: focus.datasetUri, path: focus.path }; + return focus; +} + +/** + * Map raw peer states onto {@link Peer}s, dropping anyone past `evictAfter`. + * + * Pure: same inputs, same output. This is Flux's `evaluateAgents` with the mutation, the reactive + * framework, and the `me` special-case removed — all three of which made it untestable there. + */ +export function derivePeers( + states: Iterable, + now: number, + thresholds: LivenessThresholds = DEFAULT_THRESHOLDS, +): Peer[] { + const peers: Peer[] = []; + for (const state of states) { + const age = now - state.updatedAt; + if (age >= thresholds.evictAfter) continue; + peers.push({ ...state, liveness: livenessFor(age, thresholds) }); + } + return peers; +} + +function livenessFor(age: number, t: LivenessThresholds): Liveness { + if (age < t.idleAfter) return 'online'; + if (age < t.staleAfter) return 'idle'; + if (age < t.offlineAfter) return 'stale'; + return 'offline'; +} + +// ── Selectors ──────────────────────────────────────────────────────────────── +// Space / route / node presence is not a granularity *decision* — publish the deepest focus you have +// and let each consumer slice. These are the slices. + +/** + * Peers whose focus matches every field supplied. The one primitive; the depth of the partial focus + * *is* the granularity. + * + * Deliberately not offered as a bare `peersAtPath(peers, '/kanban')`: **a route path is only + * meaningful within a dataset**, and two spaces routinely have the same one. A path-only selector + * silently unions peers across spaces, which is the kind of bug that looks correct in single-space + * testing — the same class of mistake as broadcasting a local perspective uuid. Requiring the caller + * to write the whole focus they mean removes the hazard rather than documenting it. + * + * Excludes `offline` peers unless `includeOffline`. + */ +export function peersMatching(peers: Peer[], focus: Partial, includeOffline = false): Peer[] { + const keys = Object.keys(focus) as Array; + return peers.filter((p) => { + if (!includeOffline && p.liveness === 'offline') return false; + if (!p.focus) return false; + return keys.every((k) => p.focus![k] === focus[k]); + }); +} + +/** Peers whose focus is in this dataset. Sugar for the unambiguous case. */ +export function peersInDataset(peers: Peer[], datasetUri: string, includeOffline = false): Peer[] { + return peersMatching(peers, { datasetUri }, includeOffline); +} + +/** Every activity of a type, paired with the peer performing it. */ +export function activitiesOfType( + peers: Peer[], + type: T, +): Array<{ peer: Peer; activity: Extract }> { + const out: Array<{ peer: Peer; activity: Extract }> = []; + for (const peer of peers) { + for (const activity of peer.activities ?? []) { + if (activity.type === type) out.push({ peer, activity: activity as Extract }); + } + } + return out; +} + +/** + * Group call participants by call id. + * + * Multiple concurrent calls in one space fall out for free from activities-being-a-list. Flux cannot + * express this — `callRoute` is a single value, so one call per agent, identified by channel. + */ +export function callRosters(peers: Peer[]): Map { + const rosters = new Map(); + for (const { peer, activity } of activitiesOfType(peers, 'call')) { + const roster = rosters.get(activity.id); + if (roster) roster.push(peer); + else rosters.set(activity.id, [peer]); + } + return rosters; +} + +// ── The heartbeat driver ───────────────────────────────────────────────────── + +/** The transport surface the driver needs — structurally an `EphemeralChannel`, restated so this + * module is testable with a fake and carries no import-time dependency on the port. */ +export interface PresenceChannel { + publish(payload: unknown, to?: { agentId?: string }): void; + onMessage(cb: (from: string, payload: unknown) => void): () => void; +} + +export interface HeartbeatOptions { + /** ms between heartbeats when nothing else has been published. */ + interval?: number; + thresholds?: LivenessThresholds; + /** Injected for testability and because `Date.now` is the one impurity here. */ + now?: () => number; + setTimer?: (fn: () => void, ms: number) => unknown; + clearTimer?: (handle: unknown) => void; + /** Called whenever the peer map changes, with freshly derived peers. */ + onPeersChanged?: (peers: Peer[]) => void; +} + +export interface PresenceSource { + /** Begin publishing and listening. Sends the join handshake. */ + start(state: PresenceState): void; + /** Merge a patch into this agent's state and publish immediately. */ + update(patch: Partial>): void; + /** Add or replace an activity, matched on `type` plus `id` when present. */ + setActivity(activity: Activity): void; + /** Remove activities of a type, optionally narrowed to one id. */ + clearActivity(type: string, id?: string): void; + /** Current peers, derived at call time so liveness is never stale. */ + peers(): Peer[]; + /** Stop publishing and listening. Idempotent. */ + stop(): void; +} + +/** Wire protocol. Kept minimal: a full state, plus a flag asking peers to re-announce. */ +interface PresenceMessage { + v: 1; + state: PresenceState; + /** Set on join. Peers seeing it re-broadcast so the joiner populates in one round trip. */ + hello?: true; +} + +function isPresenceMessage(payload: unknown): payload is PresenceMessage { + if (typeof payload !== 'object' || payload === null) return false; + const msg = payload as Partial; + return msg.v === 1 && typeof msg.state === 'object' && msg.state !== null; +} + +/** + * Lift a fan-out channel into a presence source by gossiping on a timer — the P1/P2 path. + * + * A P3 backend (one that reports connects and disconnects itself) should implement + * {@link PresenceSource} directly and never call this: forcing a server that already knows who is + * online to emulate 5-second gossip is exactly the lowest-common-denominator trap the capability tiers + * exist to avoid. + * + * Two behaviours lifted from Flux because they are genuinely load-bearing: + * + * - **Adaptive scheduling.** A state change publishes immediately and pushes the next tick out a full + * interval, so a peer navigating quickly does not double-send. + * - **Join handshake.** A joiner sets `hello`; peers seeing it re-announce at once. Without it a + * joiner waits a whole interval to see anyone, because the transport is stateless and lossy. + */ +export function createHeartbeatPresence(channel: PresenceChannel, options: HeartbeatOptions = {}): PresenceSource { + const interval = options.interval ?? DEFAULT_HEARTBEAT_INTERVAL; + const thresholds = options.thresholds ?? DEFAULT_THRESHOLDS; + const now = options.now ?? (() => Date.now()); + const setTimer = options.setTimer ?? ((fn, ms) => setTimeout(fn, ms)); + const clearTimer = options.clearTimer ?? ((h) => clearTimeout(h as ReturnType)); + + const states = new Map(); + let self: PresenceState | null = null; + let unsubscribe: (() => void) | null = null; + let timer: unknown = null; + let running = false; + + function notify(): void { + options.onPeersChanged?.(derivePeers(states.values(), now(), thresholds)); + } + + function send(hello?: true): void { + if (!running || !self) return; + self = { ...self, updatedAt: now() }; + states.set(self.agentId, self); + // `invisible` is a hard stop, not a client-side filter. Flux keeps broadcasting full state + // (including route) when invisible and merely hides it on receipt, so any modified client sees + // invisible agents and where they are. + if (self.availability !== 'invisible') { + const message: PresenceMessage = hello ? { v: 1, state: self, hello } : { v: 1, state: self }; + channel.publish(message); + } + notify(); + } + + function schedule(delay: number): void { + if (timer !== null) clearTimer(timer); + timer = setTimer(() => { + timer = null; + if (!running || !self) return; + // Another publish may have happened since this tick was scheduled; wait out the remainder + // rather than sending twice inside one interval. + const sinceLast = now() - self.updatedAt; + if (sinceLast < interval) schedule(interval - sinceLast); + else { + send(); + schedule(interval); + } + }, delay); + } + + function receive(from: string, payload: unknown): void { + if (!running || !isPresenceMessage(payload)) return; + if (self && from === self.agentId) return; + + // Trust the transport's sender id over the payload's — a peer must not be able to write into + // another agent's slot. Where `authenticatedSender` is false this is still the best available + // key, and the capability flag is what warns a consumer not to act on it. + states.set(from, { ...payload.state, agentId: from, updatedAt: now() }); + + // Answer a joiner immediately so they do not wait out a full interval. Not itself a hello, or + // two peers would ping-pong forever. + if (payload.hello) send(); + else notify(); + } + + function evictStale(): void { + const cutoff = now() - thresholds.evictAfter; + for (const [id, state] of states) { + if (state.updatedAt <= cutoff && id !== self?.agentId) states.delete(id); + } + } + + return { + start(state) { + if (running) return; + running = true; + self = { ...state, updatedAt: now() }; + states.set(self.agentId, self); + unsubscribe = channel.onMessage(receive); + send(true); + schedule(interval); + }, + + update(patch) { + if (!self) return; + self = { ...self, ...patch }; + send(); + schedule(interval); + }, + + setActivity(activity) { + if (!self) return; + const id = 'id' in activity ? activity.id : undefined; + const rest = (self.activities ?? []).filter( + (a) => a.type !== activity.type || ('id' in a ? a.id : undefined) !== id, + ); + self = { ...self, activities: [...rest, activity] }; + send(); + schedule(interval); + }, + + clearActivity(type, id) { + if (!self?.activities) return; + const activities = self.activities.filter( + (a) => a.type !== type || (id !== undefined && ('id' in a ? a.id : undefined) !== id), + ); + self = { ...self, activities }; + send(); + schedule(interval); + }, + + peers() { + evictStale(); + return derivePeers(states.values(), now(), thresholds); + }, + + stop() { + running = false; + if (timer !== null) { + clearTimer(timer); + timer = null; + } + unsubscribe?.(); + unsubscribe = null; + states.clear(); + self = null; + }, + }; +} From cf2fd3d8d7a4752a5ce9551716c471e47281e3f4 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:13:42 +0100 Subject: [PATCH 03/11] feat(app-framework): AD4M implementation of the ephemeral port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sibling to ad4mAdapter.ts, and deliberately the same size and shape: a capability profile plus the minimum translation, with no timers, no liveness derivation, and no profile fetching. If it grows past ~120 lines something backend-agnostic has leaked in. sendBroadcastU is an unsigned broadcast never written to the perspective — exactly right for state that must not persist. The sender arrives as link.author from the executor rather than from the payload, hence authenticatedSender: true; that is what stops a peer writing into another agent's presence slot, and what will make a work-claim lease trustworthy. unicast is 'emulated': AD4M exposes real directed send (sendSignal / sendSignalU) but it is known-broken, so this addresses over broadcast with the recipient DID in the link target and a receive-side filter. That is addressing, not privacy — every peer still receives the payload — which is why planEphemeral refuses confidential consumers rather than silently exposing them. Mirrors the drill-down predicate workaround already in ad4mAdapter.ts: a backend defect degrades to an adapter workaround, not a WE outage. When sendSignalU is fixed, flip the capability and delete the filter; grep unicast:emulated. No consumer changes. One signal handler per scope, fanned out to channels by predicate suffix — registering per-channel would mean N executor subscriptions for one stream. Returns null for a personal space (no neighbourhood, nobody to signal) so consumers degrade deliberately instead of publishing into a void. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/shared/ad4mEphemeralAdapter.ts | 142 ++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 packages/app-framework/src/shared/ad4mEphemeralAdapter.ts diff --git a/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts b/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts new file mode 100644 index 00000000..980c8fd5 --- /dev/null +++ b/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts @@ -0,0 +1,142 @@ +/** + * The AD4M ephemeral adapter — the {@link EphemeralPort} presence and (later) the call module ride. + * + * Sibling to `ad4mAdapter.ts`, and deliberately the same size and shape: a capability profile plus the + * minimum translation between the neutral port and this backend's API. No timers, no liveness + * derivation, no profile fetching — those are neutral and live in `@we/schema-shared`'s `presence.ts`. + * If this file grows past ~120 lines, something backend-agnostic has leaked into it. + * + * ## What AD4M gives us + * + * `NeighbourhoodProxy.sendBroadcastU({ links })` — an *unsigned* broadcast that is never written to + * the perspective, which is exactly right for state that must not persist. Inbound arrives through + * `addSignalHandler`, and crucially the sender arrives as `link.author`, supplied by the executor + * rather than by the payload. That is why `authenticatedSender` is true: a peer cannot write into + * another agent's presence slot, which in turn is what makes a work-claim lease trustworthy. + * + * ## Why unicast is emulated + * + * AD4M *does* expose real directed send (`sendSignal` / `sendSignalU`), but it is known-broken, so + * this adapter addresses over broadcast: the recipient DID goes in the link's `target` and receivers + * drop anything not addressed to them. That is `unicast: 'emulated'` — **addressing, not privacy**: + * every peer still receives the payload. Consumers that need confidentiality are refused by + * `planEphemeral` rather than silently exposed. + * + * This mirrors the drill-down predicate workaround in `ad4mAdapter.ts`: a backend defect degrades to + * an adapter workaround, not a WE outage. When `sendSignalU` is fixed, flip `unicast` to `'native'`, + * route `publish(..., { agentId })` through it, and delete `TARGET_ALL` plus the receive-side filter — + * grep `unicast:emulated`. No consumer changes. + */ +import type { PerspectiveExpression, PerspectiveProxy } from '@coasys/ad4m'; +import type { EphemeralCapabilities, EphemeralChannel, EphemeralPort, EphemeralScope } from '@we/schema-shared'; + +/** Namespaces this traffic so it can never be confused with another protocol's links. */ +const PREDICATE_PREFIX = 'we://ephemeral/'; + +/** `target` value meaning "everyone" — see `unicast:emulated` above. */ +const TARGET_ALL = '*'; + +export const ad4mEphemeralCapabilities: EphemeralCapabilities = { + fanout: true, + // Real unicast exists upstream but `sendSignalU` is broken; we address over broadcast instead. + unicast: 'emulated', + // `sendBroadcastU` resolves `Promise` — the send is confirmed, delivery is not. + reliability: 'send-acked', + // No connect/disconnect feed, so peers must gossip their own liveness. + heartbeatRequired: true, + // `link.author` is supplied by the executor, not the payload. + authenticatedSender: true, +}; + +/** + * Build the port. `getMyDid` is used only to drop our own echo — AD4M's `sendBroadcastU` takes a + * `loopback` flag that defaults to false, but a self-addressed signal can still arrive through other + * paths, and a presence source that ingests its own state as a peer double-counts. + */ +export function createAd4mEphemeralPort(getMyDid: () => string | undefined): EphemeralPort { + return (dataset) => { + const perspective = dataset as PerspectiveProxy | null; + // A personal (unshared) space has no neighbourhood, so there is nobody to signal. Null rather + // than a no-op scope, so consumers degrade deliberately instead of silently publishing into a void. + if (!perspective?.sharedUrl) return null; + + const neighbourhood = perspective.getNeighbourhoodProxy?.(); + if (!neighbourhood) return null; + + const channels = new Map(); + const subscribers = new Map void>>(); + + // One AD4M handler for the whole scope, fanned out to channels by predicate suffix. Registering + // per-channel would mean N executor subscriptions for what is one stream. + const handler = (signal: PerspectiveExpression) => { + const link = signal?.data?.links?.[0]; + if (!link?.author) return; + + const { source, predicate, target } = link.data ?? {}; + if (typeof predicate !== 'string' || !predicate.startsWith(PREDICATE_PREFIX)) return; + + // unicast:emulated — the payload reached us either way; honour the addressing. + const myDid = getMyDid(); + if (target && target !== TARGET_ALL && target !== myDid) return; + if (link.author === myDid) return; + + const listeners = subscribers.get(predicate.slice(PREDICATE_PREFIX.length)); + if (!listeners?.size) return; + + let payload: unknown; + try { + payload = JSON.parse(source as string); + } catch { + // A peer on a newer/older protocol, or a corrupted signal. Dropping one message is correct — + // presence is idempotent, so the next heartbeat repairs the gap. + return; + } + + listeners.forEach((cb) => cb(link.author as string, payload)); + }; + + void neighbourhood.addSignalHandler(handler); + + const scope: EphemeralScope = { + capabilities: ad4mEphemeralCapabilities, + + channel(tag) { + const existing = channels.get(tag); + if (existing) return existing; + + const predicate = PREDICATE_PREFIX + tag; + const channel: EphemeralChannel = { + publish(payload, to) { + neighbourhood + .sendBroadcastU({ + links: [{ source: JSON.stringify(payload), predicate, target: to?.agentId ?? TARGET_ALL }], + }) + // Presence is idempotent and heartbeats again shortly, so a failed send is not worth + // escalating to the user — but it must not be silent either. + .catch((error: unknown) => console.warn(`ephemeral: broadcast failed on "${tag}"`, error)); + }, + onMessage(cb) { + let listeners = subscribers.get(tag); + if (!listeners) { + listeners = new Set(); + subscribers.set(tag, listeners); + } + listeners.add(cb); + return () => listeners!.delete(cb); + }, + }; + + channels.set(tag, channel); + return channel; + }, + + dispose() { + neighbourhood.removeSignalHandler(handler); + subscribers.clear(); + channels.clear(); + }, + }; + + return scope; + }; +} From ec54f0634c289e4bd86b9d6e6010cdacc5a2da08 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:13:42 +0100 Subject: [PATCH 04/11] feat(app-framework): elect one tab per origin to publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without this every open tab heartbeats independently: N times the broadcast traffic, and peers see one agent flapping between whatever each tab happens to be showing. Leadership follows window focus, so the tab the user is actually looking at is the one whose location gets published, and a tab holding something uninterruptible can pin leadership and refuse to yield. Followers stay fully subscribed — only publishing is restricted, so every tab's UI stays live. Lives in app-framework/shared rather than schema-shared because BroadcastChannel is a DOM API: schema-shared is DOM-free and is consumed by the we-validate-schemas CLI under Node. This is host wiring, like $onError and $useQueryIR. Degrades to a permanent sole leader under electron/tauri and anywhere BroadcastChannel is unavailable, which is the correct behaviour for a single window. Adapted from Flux's useTabCoordinator, minus the Vue coupling. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/shared/tabCoordinator.ts | 207 ++++++++++++++++++ 1 file changed, 207 insertions(+) create mode 100644 packages/app-framework/src/shared/tabCoordinator.ts diff --git a/packages/app-framework/src/shared/tabCoordinator.ts b/packages/app-framework/src/shared/tabCoordinator.ts new file mode 100644 index 00000000..d38a83b4 --- /dev/null +++ b/packages/app-framework/src/shared/tabCoordinator.ts @@ -0,0 +1,207 @@ +/** + * Tab coordinator — elects one tab per origin to do the talking. + * + * Without this, every open tab heartbeats independently: N× the broadcast traffic, and peers see one + * agent flapping between whatever each tab happens to be looking at. Leadership follows **window + * focus**, so the tab the user is actually looking at is the one whose location gets published; a tab + * holding something uninterruptible (an active call) pins leadership and refuses to yield. + * + * Followers stay fully subscribed — they receive everything and their UI stays live. Only *publishing* + * is restricted to the leader. + * + * Lives in `app-framework/shared` rather than `@we/schema-shared` because `BroadcastChannel` is a DOM + * API: schema-shared is DOM-free and is consumed by the `we-validate-schemas` CLI under Node. This is + * host wiring, like `$onError` and `$useQueryIR`. Under electron/tauri single-window it degrades to + * "always leader", which is correct. + * + * Adapted from Flux's `useTabCoordinator`, minus the Vue coupling. + */ + +const CHANNEL_NAME = 'we-tab-coordinator'; +const HEARTBEAT_INTERVAL = 5_000; +/** Long enough to survive a missed beat or two; short enough that a crashed leader is replaced fast. */ +const LEADER_TIMEOUT = 15_000; + +type Message = + | { type: 'claim'; tabId: string; at: number } + /** The current leader refusing to yield because it is pinned. */ + | { type: 'pinned'; tabId: string; at: number } + | { type: 'heartbeat'; tabId: string; at: number } + /** Leaving cleanly — a successor can claim at once instead of waiting out LEADER_TIMEOUT. */ + | { type: 'resign'; tabId: string; at: number }; + +export interface TabCoordinator { + isLeader(): boolean; + /** Refuse to yield leadership while true (an active call). */ + setPinned(pinned: boolean): void; + onBecomeLeader(cb: () => void): () => void; + onLoseLeadership(cb: () => void): () => void; + dispose(): void; +} + +/** Per-tab id. `sessionStorage` survives a reload but not a new tab, which is exactly the scope. */ +function getTabId(): string { + const KEY = 'we-tab-id'; + try { + const existing = sessionStorage.getItem(KEY); + if (existing) return existing; + const id = crypto.randomUUID(); + sessionStorage.setItem(KEY, id); + return id; + } catch { + // Private mode or a non-browser host: a per-instance id is still correct, it just doesn't + // survive a reload. + return crypto.randomUUID(); + } +} + +/** Single-window / no-BroadcastChannel fallback: permanently the leader, nothing to coordinate. */ +function soleLeader(): TabCoordinator { + const becameLeader = new Set<() => void>(); + let disposed = false; + return { + isLeader: () => !disposed, + setPinned: () => {}, + onBecomeLeader(cb) { + if (!disposed) cb(); + becameLeader.add(cb); + return () => becameLeader.delete(cb); + }, + onLoseLeadership() { + return () => {}; + }, + dispose() { + disposed = true; + becameLeader.clear(); + }, + }; +} + +export function createTabCoordinator(): TabCoordinator { + if (typeof BroadcastChannel === 'undefined' || typeof window === 'undefined') return soleLeader(); + + const tabId = getTabId(); + const channel = new BroadcastChannel(CHANNEL_NAME); + const becameLeader = new Set<() => void>(); + const lostLeadership = new Set<() => void>(); + + let leader = false; + let pinned = false; + let heartbeatTimer: ReturnType | null = null; + let timeoutTimer: ReturnType | null = null; + + const post = (type: Message['type']) => channel.postMessage({ type, tabId, at: Date.now() } as Message); + + function becomeLeader(): void { + if (leader) return; + leader = true; + stopWatchingLeader(); + heartbeatTimer = setInterval(() => post('heartbeat'), HEARTBEAT_INTERVAL); + post('heartbeat'); + becameLeader.forEach((cb) => cb()); + } + + function stepDown(): void { + if (heartbeatTimer) { + clearInterval(heartbeatTimer); + heartbeatTimer = null; + } + if (!leader) return; + leader = false; + lostLeadership.forEach((cb) => cb()); + } + + /** Assume the leader is gone if it goes quiet for LEADER_TIMEOUT, and take over. */ + function watchLeader(): void { + if (timeoutTimer) clearTimeout(timeoutTimer); + timeoutTimer = setTimeout(becomeLeader, LEADER_TIMEOUT); + } + + function stopWatchingLeader(): void { + if (timeoutTimer) { + clearTimeout(timeoutTimer); + timeoutTimer = null; + } + } + + channel.onmessage = (event: MessageEvent) => { + const msg = event.data; + if (!msg || msg.tabId === tabId) return; + + switch (msg.type) { + case 'claim': + // Yield to a focused tab unless pinned — then tell it why, so it stops claiming. + if (leader) { + if (pinned) post('pinned'); + else { + stepDown(); + watchLeader(); + } + } + break; + case 'pinned': + // Someone else is holding leadership deliberately; back off and watch. + stepDown(); + watchLeader(); + break; + case 'heartbeat': + if (leader) stepDown(); // two leaders — the other one just proved it's alive; defer. + watchLeader(); + break; + case 'resign': + if (!leader) becomeLeader(); + break; + } + }; + + function onFocus(): void { + if (!leader) post('claim'); + } + + function onVisibility(): void { + if (document.visibilityState === 'visible' && !leader) post('claim'); + } + + function onUnload(): void { + if (leader) post('resign'); + } + + window.addEventListener('focus', onFocus); + document.addEventListener('visibilitychange', onVisibility); + window.addEventListener('pagehide', onUnload); + + // Claim on creation when this tab is the one being looked at; otherwise wait for the incumbent to + // go quiet. + if (document.visibilityState === 'visible' && document.hasFocus()) post('claim'); + watchLeader(); + + return { + isLeader: () => leader, + setPinned(next) { + pinned = next; + // Becoming pinned while not leader means this tab holds the thing that must not be interrupted + // — claim so publishing follows it. + if (pinned && !leader) post('claim'); + }, + onBecomeLeader(cb) { + if (leader) cb(); + becameLeader.add(cb); + return () => becameLeader.delete(cb); + }, + onLoseLeadership(cb) { + lostLeadership.add(cb); + return () => lostLeadership.delete(cb); + }, + dispose() { + if (leader) post('resign'); + stepDown(); + stopWatchingLeader(); + window.removeEventListener('focus', onFocus); + document.removeEventListener('visibilitychange', onVisibility); + window.removeEventListener('pagehide', onUnload); + channel.close(); + becameLeader.clear(); + lostLeadership.clear(); + }, + }; +} From 2b0f91abcda6232da24ff5ee6aad2cf85c8373e5 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:13:59 +0100 Subject: [PATCH 05/11] feat(app-framework): PresenceStore, and presenceStore in the stores bag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Solid binding over the neutral core. It owns the three things the core deliberately does not: when to start and stop (following the current perspective), what to publish as focus (following the route), and the join from a bare agentId to a displayable profile. App-lifetime, not view-lifetime — mounted in StoreProvider rather than inside a view. Flux creates a signalling service per community view and tears it down on unmount, so leaving the view loses every peer with no path back except remounting. Publishing and subscribing are asymmetric and easy to conflate. This agent has one location, so it publishes once per heartbeat to the space it is in, and subscribes only to that same space. Our own dot needs no transport at all — it is routeStore.currentPath, read locally. Presence for every joined space would let the sidebar show occupancy everywhere, but inbound traffic is (spaces x members / heartbeat) signals per second, each crossing the executor's GraphQL boundary into reactive state on the main thread; widening this is a deliberate later decision. On space change the source is torn down rather than retained. Retention without subscription only preserves state already past its TTL, and the join handshake repopulates in one round trip anyway. Flux defines deleteCommunityService and never calls it, so services accumulate holding frozen agent maps. Focus publishes datasetUri from currentPerspectiveSharedUrl, never perspective.uuid: AD4M uuids are local per-agent, so a broadcast uuid is meaningless to whoever receives it. This fails silently across peers while looking correct locally. Tab-leader gating wraps channel.publish at the store rather than living in the driver, keeping the neutral core unaware that browser tabs exist. Profiles come from adamStore.agents(), the cache $identities and $agent already use — presence never fetches one, so it cannot repeat Flux's N-peer Promise.all every five seconds. Co-Authored-By: Claude Opus 5 (1M context) --- .../solid/providers/StoreProvider.tsx | 8 +- .../solid/providers/TemplateProvider.tsx | 3 + .../frameworks/solid/stores/PresenceStore.tsx | 227 ++++++++++++++++++ .../src/frameworks/solid/stores/index.ts | 1 + .../src/frameworks/solid/types.ts | 12 +- 5 files changed, 249 insertions(+), 2 deletions(-) create mode 100644 packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx diff --git a/packages/app-framework/src/frameworks/solid/providers/StoreProvider.tsx b/packages/app-framework/src/frameworks/solid/providers/StoreProvider.tsx index 3ef42caf..23960550 100644 --- a/packages/app-framework/src/frameworks/solid/providers/StoreProvider.tsx +++ b/packages/app-framework/src/frameworks/solid/providers/StoreProvider.tsx @@ -2,6 +2,7 @@ import { AdamStoreProvider, AiStoreProvider, AppStoreProvider, + PresenceStoreProvider, RouteStoreProvider, SpaceStoreProvider, TemplateStoreProvider, @@ -17,7 +18,12 @@ export default function StoreProvider(props: ParentProps) { - {props.children} + + {/* Innermost: presence follows the current perspective and the route, so it needs + AdamStore and RouteStore above it. App-lifetime, not view-lifetime — it must + outlive navigation rather than being torn down with a view. */} + {props.children} + diff --git a/packages/app-framework/src/frameworks/solid/providers/TemplateProvider.tsx b/packages/app-framework/src/frameworks/solid/providers/TemplateProvider.tsx index 21905c68..b0726597 100644 --- a/packages/app-framework/src/frameworks/solid/providers/TemplateProvider.tsx +++ b/packages/app-framework/src/frameworks/solid/providers/TemplateProvider.tsx @@ -8,6 +8,7 @@ import { useAdamStore, useAiStore, useAppStore, + usePresenceStore, useRouteStore, useSpaceStore, useTemplateStore, @@ -34,6 +35,7 @@ export default function TemplateProvider() { const themeStore = useThemeStore(); const templateStore = useTemplateStore(); const routeStore = useRouteStore(); + const presenceStore = usePresenceStore(); // Set CSS custom property on :root so position:fixed elements (e.g. CesiumGlobe canvas) // can consume the sidebar width without hard-coding it. @@ -76,6 +78,7 @@ export default function TemplateProvider() { themeStore, templateStore, routeStore, + presenceStore, consoleStore, model: modelStore, // Host wiring, not backend adaptation — any backend would wire these the same way, so they stay diff --git a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx new file mode 100644 index 00000000..8344b956 --- /dev/null +++ b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx @@ -0,0 +1,227 @@ +/** + * PresenceStore — live presence for the current space. + * + * The Solid binding over the neutral presence core in `@we/schema-shared`. It owns three things the + * core deliberately does not: **when** to start and stop (following the current perspective), **what** + * to publish as this agent's focus (following the route), and the **join** from a bare `agentId` to a + * displayable profile. + * + * ## Lifecycle: app-lifetime, not view-lifetime + * + * Mounted alongside the other stores in `StoreProvider`, not inside a view. Flux creates a signalling + * service per community view and tears it down on unmount, so leaving the view loses every peer with + * no path back except remounting. Here the store outlives navigation and simply re-scopes when the + * perspective changes. + * + * ## Scope: current space only + * + * Presence for *every* joined space would let the sidebar show live occupancy everywhere, but the + * traffic is (spaces × members ÷ heartbeat) inbound signals per second — uncomfortable at modest + * numbers, since each crosses the executor's GraphQL boundary and lands in reactive state on the main + * thread. Current-space-only is the conservative default; widening it is a deliberate later decision + * (and realistically wants a backend that reports presence server-side). + * + * ## Publishing vs subscribing + * + * Asymmetric, and easy to conflate. This agent has exactly one location, so it **publishes** once per + * heartbeat to the space it is in. It **subscribes** only to that same space. Our own dot needs no + * transport at all — it is `routeStore.currentPath`, read locally. + * + * ## What it never does + * + * Fetch profiles. Presence carries `agentId` only; profiles come from `adamStore.agents()`, the cache + * `$identities` and the `$agent` block already use. Flux's presence map *is* its profile cache, so it + * re-hydrates every peer profile on every heartbeat — an N-peer `Promise.all` every five seconds. + */ +import { createAd4mEphemeralPort } from '@shared/ad4mEphemeralAdapter'; +import type { AgentProfileSummary } from '@shared/agentHelpers'; +import { createTabCoordinator } from '@shared/tabCoordinator'; +import { useAdamStore } from '@solid/stores/AdamStore'; +import { useRouteStore } from '@solid/stores/RouteStore'; +import type { Activity, Focus, FocusDepth, Peer, PresenceSource } from '@we/schema-shared'; +import { applyFocusDepth, callRosters, createHeartbeatPresence, peersInDataset } from '@we/schema-shared'; +import { + type Accessor, + createContext, + createEffect, + createMemo, + createSignal, + onCleanup, + type ParentProps, + useContext, +} from 'solid-js'; + +/** A peer joined with whatever profile the agent cache holds. `did` mirrors `agentId` for template ergonomics. */ +export type PresentAgent = Peer & Partial & { did: string }; + +export interface PresenceStore { + /** Every peer we know of in the current space, liveness-derived, offline included. */ + peers: Accessor; + /** Peers in the current space who are not offline — the "who's here" list. */ + online: Accessor; + /** Peers at this agent's exact route path. */ + onlineHere: Accessor; + /** Concurrent calls in this space, keyed by call id. */ + calls: Accessor>; + /** True when a transport exists — false in a personal space, where presence is unavailable. */ + available: Accessor; + + /** How much of this agent's location to publish. Persisted per-agent later; in-memory for now. */ + focusDepth: Accessor; + setFocusDepth: (depth: FocusDepth) => void; + /** `invisible` stops publishing entirely, rather than asking peers not to look. */ + setAvailability: (availability: 'available' | 'busy' | 'away' | 'invisible') => void; + + /** Add or replace an activity (a call, an edit, a work claim). */ + setActivity: (activity: Activity) => void; + clearActivity: (type: string, id?: string) => void; +} + +const PresenceContext = createContext(); + +export function PresenceStoreProvider(props: ParentProps) { + const adamStore = useAdamStore(); + const routeStore = useRouteStore(); + + const [rawPeers, setRawPeers] = createSignal([]); + const [focusDepth, setFocusDepth] = createSignal('route'); + const [availability, setAvailabilitySignal] = createSignal<'available' | 'busy' | 'away' | 'invisible'>('available'); + const [available, setAvailable] = createSignal(false); + + let source: PresenceSource | null = null; + + // One coordinator for the app: only the focused tab publishes, but every tab subscribes so each + // one's UI stays live. + const tabs = createTabCoordinator(); + onCleanup(() => tabs.dispose()); + + const ephemeralPort = createAd4mEphemeralPort(() => adamStore.me()?.did); + + /** + * The space, by its **global** uri. Never `perspective.uuid`: AD4M perspective uuids are local + * per-agent, so the same neighbourhood has a different one on every peer — broadcasting it produces + * a focus nobody else can interpret. Fails silently across peers while looking fine locally. + */ + const datasetUri = createMemo(() => adamStore.currentPerspectiveSharedUrl()); + + const myFocus = createMemo(() => + applyFocusDepth({ datasetUri: datasetUri(), path: routeStore.currentPath?.() }, focusDepth()), + ); + + // ── Lifecycle ────────────────────────────────────────────────────────────── + // Re-scope whenever the current space changes: tear the old source down completely rather than + // retaining its peers. Retention without subscription only preserves state that is already past its + // TTL, and the join handshake repopulates in one round trip on return anyway. + createEffect(() => { + const perspective = adamStore.currentPerspective(); + const did = adamStore.me()?.did; + + source?.stop(); + source = null; + setRawPeers([]); + setAvailable(false); + + if (!perspective || !did) return; + + const scope = ephemeralPort(perspective); + if (!scope) return; // personal space — no neighbourhood, no presence + + // Only the focused tab publishes; every tab still subscribes, so each one's UI stays live. + // Gating here rather than inside the driver keeps the neutral core unaware of browser tabs — + // and a follower tab's suppressed heartbeat is harmless, because the leader is publishing the + // same agent's state. + const raw = scope.channel('presence'); + const channel = { + publish: (payload: unknown, to?: { agentId?: string }) => { + if (tabs.isLeader()) raw.publish(payload, to); + }, + onMessage: raw.onMessage, + }; + const presence = createHeartbeatPresence(channel, { onPeersChanged: setRawPeers }); + + source = presence; + setAvailable(true); + presence.start({ + agentId: did, + updatedAt: Date.now(), + availability: availability(), + focus: myFocus(), + }); + + // Publish as soon as this tab takes over, so leadership changing mid-session doesn't leave + // peers waiting out a full interval for the new leader's first heartbeat. + const unsubLeader = tabs.onBecomeLeader(() => presence.update({})); + + onCleanup(() => { + unsubLeader(); + presence.stop(); + scope.dispose(); + }); + }); + + // Republish on navigation. Immediate rather than waiting for the next tick — the heartbeat driver + // pushes its timer out by a full interval so this does not cause a double-send. + createEffect(() => { + const focus = myFocus(); + source?.update({ focus }); + }); + + createEffect(() => { + source?.update({ availability: availability() }); + }); + + // ── Derived ──────────────────────────────────────────────────────────────── + // Join against the shared agent cache. `find` over `agents()` matches how SpaceStore.members + // resolves its dids; both read the same cache that `fetchAgent` populates. + const peers = createMemo(() => { + const cached = adamStore.agents(); + return rawPeers().map((peer) => { + const profile = cached.find((a) => a.did === peer.agentId); + return { ...peer, ...profile, did: peer.agentId }; + }); + }); + + // Ask AD4M for any peer whose profile we have not cached. The effect re-runs as peers arrive, and + // `fetchAgent` already deduplicates in-flight requests, so this is safe to call repeatedly. + createEffect(() => { + const cached = adamStore.agents(); + for (const peer of rawPeers()) { + if (!cached.some((a) => a.did === peer.agentId)) void adamStore.fetchAgent(peer.agentId); + } + }); + + const online = createMemo(() => { + const uri = datasetUri(); + if (!uri) return []; + return peersInDataset(peers(), uri) as PresentAgent[]; + }); + + const onlineHere = createMemo(() => { + const path = routeStore.currentPath?.(); + if (!path) return []; + return online().filter((p) => p.focus?.path === path); + }); + + const calls = createMemo(() => callRosters(online()) as Map); + + const store: PresenceStore = { + peers, + online, + onlineHere, + calls, + available, + focusDepth, + setFocusDepth, + setAvailability: setAvailabilitySignal, + setActivity: (activity) => source?.setActivity(activity), + clearActivity: (type, id) => source?.clearActivity(type, id), + }; + + return {props.children}; +} + +export function usePresenceStore(): PresenceStore { + const store = useContext(PresenceContext); + if (!store) throw new Error('usePresenceStore must be used within a PresenceStoreProvider'); + return store; +} diff --git a/packages/app-framework/src/frameworks/solid/stores/index.ts b/packages/app-framework/src/frameworks/solid/stores/index.ts index dc986ddc..bed37959 100644 --- a/packages/app-framework/src/frameworks/solid/stores/index.ts +++ b/packages/app-framework/src/frameworks/solid/stores/index.ts @@ -5,4 +5,5 @@ export { type TemplateStore, useTemplateStore, TemplateStoreProvider } from './T export { type RouteStore, useRouteStore, RouteStoreProvider } from './RouteStore'; export { type AiStore, useAiStore, AiStoreProvider } from './AiStore'; export { type AppStore, useAppStore, AppStoreProvider } from './AppStore'; +export { type PresenceStore, type PresentAgent, usePresenceStore, PresenceStoreProvider } from './PresenceStore'; export { useShellRouteStore, ShellRouteStoreProvider, ShellRouterRoot } from './ShellRouteStore'; diff --git a/packages/app-framework/src/frameworks/solid/types.ts b/packages/app-framework/src/frameworks/solid/types.ts index f2bdc6ef..e709defe 100644 --- a/packages/app-framework/src/frameworks/solid/types.ts +++ b/packages/app-framework/src/frameworks/solid/types.ts @@ -1,5 +1,14 @@ import type { Ad4mModel } from '@coasys/ad4m'; -import type { AdamStore, AiStore, AppStore, RouteStore, SpaceStore, TemplateStore, ThemeStore } from '@solid/stores'; +import type { + AdamStore, + AiStore, + AppStore, + PresenceStore, + RouteStore, + SpaceStore, + TemplateStore, + ThemeStore, +} from '@solid/stores'; import type { RendererStores } from '@we/schema-shared'; export type ModelStoreOptions = { @@ -44,6 +53,7 @@ export interface Stores extends RendererStores { themeStore: ThemeStore; templateStore: TemplateStore; routeStore: RouteStore; + presenceStore: PresenceStore; model?: ModelStore; /** Neutral identity — the current agent (templates read `$me.did`). Backed by `adamStore.me`; * typed `unknown` so the seam stays backend-agnostic. Host-specific: not part of the data contract. */ From d05d08f88a35432bdbbca43bac10f0f95a9eb8c4 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:13:59 +0100 Subject: [PATCH 06/11] feat(app-framework): show who is online in the space header MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit "N online now" plus an avatar row, to the right of the route buttons — the nav row already has ax: 'between', so a second child lands there. Reuses AvatarStack exactly as the members row below the title does. Hidden when nobody else is around rather than rendering "0 online now", and absent entirely in a personal space, where there is no neighbourhood and so no presence. Reads presenceStore through $store, so this needs no renderer or schema-shared changes. A $presence block would be sugar on top and touches schema-solid; keeping it out means the first consumer of the port is proved by an ordinary template. Co-Authored-By: Claude Opus 5 (1M context) --- .../HeaderLayout/SpaceHeader.ts | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts b/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts index 11e1c26f..edc12b12 100644 --- a/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts +++ b/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts @@ -234,6 +234,40 @@ export const spaceNavBar: SchemaNode = { }, ], }, + // Live presence — who else is in this space right now. Hidden entirely in a personal + // space (no neighbourhood, so `presenceStore.available` is false) and when nobody else + // is around, rather than rendering "0 online now". + { + type: '$if', + props: { + condition: { $gt: [{ $count: { items: { $store: 'presenceStore.online' } } }, 0] }, + then: { + type: 'Row', + props: { gap: '200', ay: 'center' }, + children: [ + { + type: 'we-number', + props: { value: { $count: { items: { $store: 'presenceStore.online' } } }, shorten: true }, + }, + { type: 'we-text', props: { color: 'neutral-800' }, children: ['online now'] }, + { + type: 'AvatarStack', + props: { + avatars: { + $map: { + items: { $store: 'presenceStore.online' }, + select: { image: '$item.avatar', hash: '$item.did' }, + }, + }, + max: 5, + size: 'sm', + ring: '0 0 0 2px var(--we-ring-color)', + }, + }, + ], + }, + }, + }, ], }, ], From bc3d65457c72262c9b87f4839f981d143f7bf551 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:35:40 +0100 Subject: [PATCH 07/11] fix(presence): coalesce in-flight sends and back off failure logging MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On an unhealthy executor `sendBroadcast` hangs until a 30s RPC timeout while presence heartbeats every 5s, so six stuck calls accumulate at steady state — each one adding load to the backend that is already the problem — and the console fills with a warning every five seconds, burying whatever the real fault is. Adds an opt-in `coalesce` channel option: drop a publish while a previous one is still in flight. Correct only for idempotent last-write-wins traffic, where the next beat carries the same state; explicitly wrong for a handshake, where a dropped RTC offer is simply lost. Presence opts in; the default is off so the call module's rtc channel is unaffected. Failure logging now backs off geometrically (1st, 2nd, 4th, 8th…) and logs once on recovery with the count, so an unreachable neighbourhood reports itself without drowning the console. Neither of these makes presence work against a neighbourhood that is not syncing — they stop presence making it worse, and stop it hiding the cause. Co-Authored-By: Claude Opus 5 (1M context) --- .../frameworks/solid/stores/PresenceStore.tsx | 5 ++- .../src/shared/ad4mEphemeralAdapter.ts | 34 ++++++++++++++++--- .../schema-system/shared/src/ephemeral.ts | 23 +++++++++++-- 3 files changed, 55 insertions(+), 7 deletions(-) diff --git a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx index 8344b956..0e165e60 100644 --- a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx +++ b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx @@ -130,7 +130,10 @@ export function PresenceStoreProvider(props: ParentProps) { // Gating here rather than inside the driver keeps the neutral core unaware of browser tabs — // and a follower tab's suppressed heartbeat is harmless, because the leader is publishing the // same agent's state. - const raw = scope.channel('presence'); + // coalesce: presence is last-write-wins, so a beat dropped because the previous send is still + // in flight costs nothing — and on an unhealthy executor it is the difference between one + // pending broadcast and six. + const raw = scope.channel('presence', { coalesce: true }); const channel = { publish: (payload: unknown, to?: { agentId?: string }) => { if (tabs.isLeader()) raw.publish(payload, to); diff --git a/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts b/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts index 980c8fd5..6cc82a5a 100644 --- a/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts +++ b/packages/app-framework/src/shared/ad4mEphemeralAdapter.ts @@ -100,20 +100,46 @@ export function createAd4mEphemeralPort(getMyDid: () => string | undefined): Eph const scope: EphemeralScope = { capabilities: ad4mEphemeralCapabilities, - channel(tag) { + channel(tag, options) { const existing = channels.get(tag); if (existing) return existing; const predicate = PREDICATE_PREFIX + tag; + let inFlight = false; + let failures = 0; + const channel: EphemeralChannel = { publish(payload, to) { + // Backpressure for idempotent traffic. When the executor is unhealthy `sendBroadcast` + // hangs until a 30s RPC timeout, so a 5s heartbeat accumulates six stuck calls — piling + // load onto the thing that is already failing. Dropping a beat costs nothing here: the + // next one carries the same state. + if (options?.coalesce && inFlight) return; + inFlight = true; + neighbourhood .sendBroadcastU({ links: [{ source: JSON.stringify(payload), predicate, target: to?.agentId ?? TARGET_ALL }], }) - // Presence is idempotent and heartbeats again shortly, so a failed send is not worth - // escalating to the user — but it must not be silent either. - .catch((error: unknown) => console.warn(`ephemeral: broadcast failed on "${tag}"`, error)); + .then(() => { + if (failures > 0) { + console.info(`ephemeral: "${tag}" recovered after ${failures} failed send(s)`); + failures = 0; + } + }) + // A failed send is not worth escalating to the user — presence repairs itself on the + // next beat — but it must not be silent either. Log the first, then back off + // geometrically: an unreachable neighbourhood otherwise emits a warning every 5s and + // buries whatever the actual problem is. + .catch((error: unknown) => { + failures += 1; + if ((failures & (failures - 1)) === 0) { + console.warn(`ephemeral: send failed on "${tag}" (${failures} consecutive)`, error); + } + }) + .finally(() => { + inFlight = false; + }); }, onMessage(cb) { let listeners = subscribers.get(tag); diff --git a/packages/schema-system/shared/src/ephemeral.ts b/packages/schema-system/shared/src/ephemeral.ts index 77b2e6ca..84942a8c 100644 --- a/packages/schema-system/shared/src/ephemeral.ts +++ b/packages/schema-system/shared/src/ephemeral.ts @@ -98,11 +98,30 @@ export interface EphemeralChannel { onMessage(cb: (from: string, payload: unknown) => void): () => void; } +export interface ChannelOptions { + /** + * Drop a publish while a previous one is still in flight, rather than letting sends pile up. + * + * Correct only for **idempotent last-write-wins** traffic — presence, cursors, typing — where a + * dropped message costs nothing because the next one carries the same information. It is wrong for + * a handshake: an RTC offer dropped because the previous send is slow is simply lost. + * + * Earns its place from a real failure. On a struggling AD4M executor `sendBroadcast` hangs until a + * 30s RPC timeout while presence heartbeats every 5s, so six stuck calls accumulate at steady + * state, each adding load to the backend that is already the problem. Coalescing turns that into + * one in-flight send. + */ + coalesce?: boolean; +} + /** Ephemeral traffic for one dataset. Obtain channels from it; dispose to detach from the backend. */ export interface EphemeralScope { capabilities: EphemeralCapabilities; - /** Namespaced sub-channel. Repeated calls with the same tag return the same channel. */ - channel(tag: string): EphemeralChannel; + /** + * Namespaced sub-channel. Repeated calls with the same tag return the same channel; options are + * read on first creation. + */ + channel(tag: string, options?: ChannelOptions): EphemeralChannel; dispose(): void; } From 10481d05924b68a04d389bd3c3b6ce48c37d091a Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 18:55:39 +0100 Subject: [PATCH 08/11] fix(tabCoordinator): make leadership conflict resolution total, and test it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refactors the coordinator to take its channel, focus source, and tab id as dependencies — the pattern createHeartbeatPresence already uses for now/timers — so leader election is testable without a DOM. Adds 20 tests. The public interface is unchanged; PresenceStore is untouched. Writing them surfaced two real defects, both of the silent kind: no exception, no log, just doubled heartbeats or none, which would present as an AD4M problem. Symmetric conflict resolution never terminates. "Hear another leader, step down" makes *both* step down, so nobody publishes until the timeout, then both take over again — a stable oscillation. Now the lower tab id survives. Deterministic and total. Focus still expresses preference; the id comparison only resolves conflict. Pinning was a local exemption, which is not a total order. A pinned leader skipped the comparison for itself but never told the other leader to yield, so when the pinned tab held the higher id neither stepped down and both published forever. Pinning is now asserted with a `pinned` message, and a `pinned` received while also pinned falls back to the id comparison — otherwise two tabs each holding a call would both defer and leave nobody publishing. Also fixes a genuine crash-recovery hole in the test harness rather than the code: a killed tab must stop sending as well as receiving, or its heartbeats keep holding peers off after it is gone. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/shared/tabCoordinator.ts | 192 +++++++--- .../tests/tabCoordinator.test.ts | 357 ++++++++++++++++++ 2 files changed, 507 insertions(+), 42 deletions(-) create mode 100644 packages/app-framework/tests/tabCoordinator.test.ts diff --git a/packages/app-framework/src/shared/tabCoordinator.ts b/packages/app-framework/src/shared/tabCoordinator.ts index d38a83b4..e89ac483 100644 --- a/packages/app-framework/src/shared/tabCoordinator.ts +++ b/packages/app-framework/src/shared/tabCoordinator.ts @@ -14,6 +14,16 @@ * host wiring, like `$onError` and `$useQueryIR`. Under electron/tauri single-window it degrades to * "always leader", which is correct. * + * ## Conflict resolution + * + * Focus expresses *preference*; a deterministic tab-id comparison resolves *conflict*. The split is + * load-bearing. Two tabs can briefly both believe they lead — after a resign, or if two windows each + * report focus — and a symmetric rule ("hear another leader → step down") makes **both** step down, + * leaving nobody publishing until the timeout, then both take over again: a stable oscillation with + * no error anywhere. Comparing ids breaks the symmetry so exactly one survives, and because + * `becomeLeader` heartbeats immediately, the collision resolves in one round trip rather than a + * timeout. + * * Adapted from Flux's `useTabCoordinator`, minus the Vue coupling. */ @@ -23,12 +33,37 @@ const HEARTBEAT_INTERVAL = 5_000; const LEADER_TIMEOUT = 15_000; type Message = - | { type: 'claim'; tabId: string; at: number } + /** "I have focus and want to publish." */ + | { type: 'claim'; tabId: string } /** The current leader refusing to yield because it is pinned. */ - | { type: 'pinned'; tabId: string; at: number } - | { type: 'heartbeat'; tabId: string; at: number } - /** Leaving cleanly — a successor can claim at once instead of waiting out LEADER_TIMEOUT. */ - | { type: 'resign'; tabId: string; at: number }; + | { type: 'pinned'; tabId: string } + | { type: 'heartbeat'; tabId: string } + /** Leaving cleanly — a successor can take over at once instead of waiting out LEADER_TIMEOUT. */ + | { type: 'resign'; tabId: string }; + +/** + * The cross-tab transport. Narrower than `BroadcastChannel` on purpose: this is the whole surface the + * coordinator needs, so it can be driven by an in-memory bus in tests without a DOM. + */ +export interface CoordinatorChannel { + post(message: unknown): void; + /** Must NOT deliver a tab its own messages — `BroadcastChannel` already behaves this way. */ + subscribe(cb: (message: unknown) => void): () => void; + close(): void; +} + +/** Focus signals. Injected so the focus-follows-leader behaviour is testable without a window. */ +export interface FocusSource { + hasFocus(): boolean; + onFocusGained(cb: () => void): () => void; + onHide(cb: () => void): () => void; +} + +export interface TabCoordinatorDeps { + channel?: CoordinatorChannel | null; + focus?: FocusSource; + tabId?: string; +} export interface TabCoordinator { isLeader(): boolean; @@ -40,7 +75,7 @@ export interface TabCoordinator { } /** Per-tab id. `sessionStorage` survives a reload but not a new tab, which is exactly the scope. */ -function getTabId(): string { +function defaultTabId(): string { const KEY = 'we-tab-id'; try { const existing = sessionStorage.getItem(KEY); @@ -49,13 +84,49 @@ function getTabId(): string { sessionStorage.setItem(KEY, id); return id; } catch { - // Private mode or a non-browser host: a per-instance id is still correct, it just doesn't + // Private mode or a non-browser host: a per-instance id is still correct, it just does not // survive a reload. return crypto.randomUUID(); } } -/** Single-window / no-BroadcastChannel fallback: permanently the leader, nothing to coordinate. */ +function defaultChannel(): CoordinatorChannel | null { + if (typeof BroadcastChannel === 'undefined') return null; + const bc = new BroadcastChannel(CHANNEL_NAME); + return { + post: (message) => bc.postMessage(message), + subscribe: (cb) => { + const listener = (event: MessageEvent) => cb(event.data); + bc.addEventListener('message', listener); + return () => bc.removeEventListener('message', listener); + }, + close: () => bc.close(), + }; +} + +function defaultFocus(): FocusSource | null { + if (typeof window === 'undefined' || typeof document === 'undefined') return null; + return { + hasFocus: () => document.visibilityState === 'visible' && document.hasFocus(), + onFocusGained: (cb) => { + const onVisibility = () => { + if (document.visibilityState === 'visible') cb(); + }; + window.addEventListener('focus', cb); + document.addEventListener('visibilitychange', onVisibility); + return () => { + window.removeEventListener('focus', cb); + document.removeEventListener('visibilitychange', onVisibility); + }; + }, + onHide: (cb) => { + window.addEventListener('pagehide', cb); + return () => window.removeEventListener('pagehide', cb); + }, + }; +} + +/** Single-window / no-transport fallback: permanently the leader, nothing to coordinate. */ function soleLeader(): TabCoordinator { const becameLeader = new Set<() => void>(); let disposed = false; @@ -77,11 +148,18 @@ function soleLeader(): TabCoordinator { }; } -export function createTabCoordinator(): TabCoordinator { - if (typeof BroadcastChannel === 'undefined' || typeof window === 'undefined') return soleLeader(); +function isMessage(value: unknown): value is Message { + if (typeof value !== 'object' || value === null) return false; + const msg = value as Partial; + return typeof msg.tabId === 'string' && typeof msg.type === 'string'; +} + +export function createTabCoordinator(deps: TabCoordinatorDeps = {}): TabCoordinator { + const channel = deps.channel !== undefined ? deps.channel : defaultChannel(); + const focus = deps.focus ?? defaultFocus(); + if (!channel || !focus) return soleLeader(); - const tabId = getTabId(); - const channel = new BroadcastChannel(CHANNEL_NAME); + const tabId = deps.tabId ?? defaultTabId(); const becameLeader = new Set<() => void>(); const lostLeadership = new Set<() => void>(); @@ -89,14 +167,17 @@ export function createTabCoordinator(): TabCoordinator { let pinned = false; let heartbeatTimer: ReturnType | null = null; let timeoutTimer: ReturnType | null = null; + let disposed = false; - const post = (type: Message['type']) => channel.postMessage({ type, tabId, at: Date.now() } as Message); + const post = (type: Message['type']) => channel.post({ type, tabId } as Message); function becomeLeader(): void { - if (leader) return; + if (leader || disposed) return; leader = true; stopWatchingLeader(); heartbeatTimer = setInterval(() => post('heartbeat'), HEARTBEAT_INTERVAL); + // Immediately, not on the first interval: this is what makes a split brain resolve in one round + // trip instead of waiting up to HEARTBEAT_INTERVAL to be noticed. post('heartbeat'); becameLeader.forEach((cb) => cb()); } @@ -113,6 +194,7 @@ export function createTabCoordinator(): TabCoordinator { /** Assume the leader is gone if it goes quiet for LEADER_TIMEOUT, and take over. */ function watchLeader(): void { + if (disposed) return; if (timeoutTimer) clearTimeout(timeoutTimer); timeoutTimer = setTimeout(becomeLeader, LEADER_TIMEOUT); } @@ -124,13 +206,13 @@ export function createTabCoordinator(): TabCoordinator { } } - channel.onmessage = (event: MessageEvent) => { - const msg = event.data; - if (!msg || msg.tabId === tabId) return; + const unsubscribeChannel = channel.subscribe((raw) => { + if (disposed || !isMessage(raw) || raw.tabId === tabId) return; - switch (msg.type) { + switch (raw.type) { case 'claim': - // Yield to a focused tab unless pinned — then tell it why, so it stops claiming. + // Focus is preference: yield to a focused tab unless pinned, and if pinned say so, so the + // claimant stops asking rather than retrying into a wall. if (leader) { if (pinned) post('pinned'); else { @@ -139,66 +221,92 @@ export function createTabCoordinator(): TabCoordinator { } } break; + case 'pinned': - // Someone else is holding leadership deliberately; back off and watch. - stepDown(); - watchLeader(); + // Someone is holding leadership deliberately; back off and watch — unless we are pinned + // too (two tabs each in a call, which WE makes perfectly possible). Deferring + // unconditionally would then have *both* step down and leave nobody publishing, so fall + // back to the id comparison, which is total. + if (leader && pinned) { + if (tabId > raw.tabId) { + stepDown(); + watchLeader(); + } + } else { + stepDown(); + watchLeader(); + } break; + case 'heartbeat': - if (leader) stepDown(); // two leaders — the other one just proved it's alive; defer. - watchLeader(); + if (leader) { + // Split brain. Deterministic tie-break, NOT "whoever hears the other steps down" — that is + // symmetric, so both would step down and then both take over again on timeout, forever. + if (pinned) { + // Outrank the id comparison, but *assert* it rather than just exempting ourselves. A + // silent local exemption is not a total order: when the pinned tab holds the higher id, + // neither side steps down and both publish forever. + post('pinned'); + } else if (tabId > raw.tabId) { + stepDown(); + watchLeader(); + } + } else { + watchLeader(); + } break; + case 'resign': - if (!leader) becomeLeader(); + // Take over at once rather than waiting out the timeout. Several followers may do this + // together; `becomeLeader` heartbeats immediately, so the tie-break above settles it on the + // next tick rather than leaving a gap. + becomeLeader(); break; } - }; + }); - function onFocus(): void { + const unsubscribeFocus = focus.onFocusGained(() => { if (!leader) post('claim'); - } - - function onVisibility(): void { - if (document.visibilityState === 'visible' && !leader) post('claim'); - } + }); - function onUnload(): void { + const unsubscribeHide = focus.onHide(() => { if (leader) post('resign'); - } - - window.addEventListener('focus', onFocus); - document.addEventListener('visibilitychange', onVisibility); - window.addEventListener('pagehide', onUnload); + }); // Claim on creation when this tab is the one being looked at; otherwise wait for the incumbent to // go quiet. - if (document.visibilityState === 'visible' && document.hasFocus()) post('claim'); + if (focus.hasFocus()) post('claim'); watchLeader(); return { isLeader: () => leader, + setPinned(next) { pinned = next; // Becoming pinned while not leader means this tab holds the thing that must not be interrupted // — claim so publishing follows it. if (pinned && !leader) post('claim'); }, + onBecomeLeader(cb) { if (leader) cb(); becameLeader.add(cb); return () => becameLeader.delete(cb); }, + onLoseLeadership(cb) { lostLeadership.add(cb); return () => lostLeadership.delete(cb); }, + dispose() { if (leader) post('resign'); stepDown(); stopWatchingLeader(); - window.removeEventListener('focus', onFocus); - document.removeEventListener('visibilitychange', onVisibility); - window.removeEventListener('pagehide', onUnload); + disposed = true; + unsubscribeChannel(); + unsubscribeFocus(); + unsubscribeHide(); channel.close(); becameLeader.clear(); lostLeadership.clear(); diff --git a/packages/app-framework/tests/tabCoordinator.test.ts b/packages/app-framework/tests/tabCoordinator.test.ts new file mode 100644 index 00000000..7b82053d --- /dev/null +++ b/packages/app-framework/tests/tabCoordinator.test.ts @@ -0,0 +1,357 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + type CoordinatorChannel, + createTabCoordinator, + type FocusSource, + type TabCoordinator, +} from '../src/shared/tabCoordinator'; + +const HEARTBEAT_INTERVAL = 5_000; +const LEADER_TIMEOUT = 15_000; + +/** + * An in-memory stand-in for `BroadcastChannel`, matching the one behaviour the coordinator depends + * on: **a tab never receives its own messages**. + */ +function createBus() { + const ports = new Map void>(); + const dead = new Set(); + return { + channelFor(tabId: string): CoordinatorChannel { + return { + post(message) { + if (dead.has(tabId)) return; + for (const [id, cb] of ports) if (id !== tabId) cb(message); + }, + subscribe(cb) { + ports.set(tabId, cb); + return () => ports.delete(tabId); + }, + close() { + ports.delete(tabId); + }, + }; + }, + /** + * Simulate a crash: the tab stops sending *and* receiving, with no `resign`. Closing its + * receive port alone is not enough — a killed tab that kept posting would still hold peers off + * with heartbeats it can no longer honour. + */ + kill(tabId: string) { + dead.add(tabId); + ports.delete(tabId); + }, + }; +} + +function createFocus(initial = false) { + let focused = initial; + const gained: Array<() => void> = []; + const hidden: Array<() => void> = []; + return { + source: { + hasFocus: () => focused, + onFocusGained: (cb) => { + gained.push(cb); + return () => {}; + }, + onHide: (cb) => { + hidden.push(cb); + return () => {}; + }, + } satisfies FocusSource, + gainFocus() { + focused = true; + gained.forEach((cb) => cb()); + }, + hide() { + hidden.forEach((cb) => cb()); + }, + }; +} + +describe('createTabCoordinator', () => { + const bus = { current: createBus() }; + const open: TabCoordinator[] = []; + + function tab(id: string, focused = false) { + const focus = createFocus(focused); + const coordinator = createTabCoordinator({ + channel: bus.current.channelFor(id), + focus: focus.source, + tabId: id, + }); + open.push(coordinator); + return { id, coordinator, ...focus }; + } + + beforeEach(() => { + vi.useFakeTimers(); + bus.current = createBus(); + open.length = 0; + }); + + afterEach(() => { + open.forEach((c) => c.dispose()); + vi.useRealTimers(); + }); + + describe('with no transport', () => { + it('is permanently the leader — correct for a single-window electron/tauri host', () => { + const solo = createTabCoordinator({ channel: null }); + expect(solo.isLeader()).toBe(true); + + const seen = vi.fn(); + solo.onBecomeLeader(seen); + expect(seen).toHaveBeenCalledOnce(); + + solo.dispose(); + expect(solo.isLeader()).toBe(false); + }); + }); + + describe('a lone tab', () => { + it('takes leadership after the incumbent timeout when it starts unfocused', () => { + const a = tab('a'); + expect(a.coordinator.isLeader()).toBe(false); + + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + }); + + it('fires onBecomeLeader immediately when already leading', () => { + const a = tab('a'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + + const seen = vi.fn(); + a.coordinator.onBecomeLeader(seen); + expect(seen).toHaveBeenCalledOnce(); + }); + }); + + describe('focus expresses preference', () => { + it('hands leadership to the tab the user looks at', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + // 'a' registered first, so its timeout fires first and 'b' defers to its heartbeat. + expect(a.coordinator.isLeader()).toBe(true); + expect(b.coordinator.isLeader()).toBe(false); + + const lost = vi.fn(); + a.coordinator.onLoseLeadership(lost); + + b.gainFocus(); + + expect(a.coordinator.isLeader()).toBe(false); + expect(lost).toHaveBeenCalledOnce(); + + // 'a' yielded but 'b' does not lead until it takes over on timeout. + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(b.coordinator.isLeader()).toBe(true); + expect(a.coordinator.isLeader()).toBe(false); + }); + + it('lets a pinned leader refuse to yield, and the claimant backs off', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + + a.coordinator.setPinned(true); + b.gainFocus(); + + expect(a.coordinator.isLeader()).toBe(true); + + // 'b' was told 'pinned', so it must not take over on its own timeout while 'a' keeps beating. + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(b.coordinator.isLeader()).toBe(false); + expect(a.coordinator.isLeader()).toBe(true); + }); + + it('claims when a non-leader becomes pinned, so publishing follows the call', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + + b.coordinator.setPinned(true); + expect(a.coordinator.isLeader()).toBe(false); + }); + }); + + describe('conflict resolution', () => { + /** + * Produce a genuine split brain. A leader's `resign` promotes *every* follower at once, so both + * end up believing they lead — the real scenario the tie-break exists for. (Staggered timeouts + * do not collide: the first to fire heartbeats and the rest defer, so they prove nothing here.) + */ + function splitBrain(followerIds: [string, string]) { + const incumbent = tab('incumbent'); + const first = tab(followerIds[0]); + const second = tab(followerIds[1]); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(incumbent.coordinator.isLeader()).toBe(true); + + incumbent.hide(); // resign → both followers promote themselves + return { first, second }; + } + + it('actually produces two leaders before converging — the scenario is real', () => { + const { first, second } = splitBrain(['a', 'b']); + expect([first, second].filter((t) => t.coordinator.isLeader())).toHaveLength(2); + }); + + it('converges to exactly one leader within a heartbeat', () => { + const { first, second } = splitBrain(['a', 'b']); + vi.advanceTimersByTime(HEARTBEAT_INTERVAL + 1); + expect([first, second].filter((t) => t.coordinator.isLeader())).toHaveLength(1); + }); + + it('breaks the tie deterministically rather than symmetrically', () => { + // A symmetric "hear another leader → step down" rule makes BOTH step down, leaving nobody + // publishing until the timeout, then both take over again — a silent stable oscillation. + // The lower tab id must survive, whichever order they were created in. + const { first: b, second: a } = splitBrain(['b', 'a']); + vi.advanceTimersByTime(HEARTBEAT_INTERVAL + 1); + + expect(a.coordinator.isLeader()).toBe(true); + expect(b.coordinator.isLeader()).toBe(false); + }); + + it('does not oscillate — the winner stays won', () => { + const { first: a, second: b } = splitBrain(['a', 'b']); + vi.advanceTimersByTime(HEARTBEAT_INTERVAL * 20); + + expect(a.coordinator.isLeader()).toBe(true); + expect(b.coordinator.isLeader()).toBe(false); + }); + + it('lets a pinned tab outrank a lower id', () => { + // 'z' loses the id comparison but is holding a call, so it must keep leadership — and 'a' + // must actually step down. Exempting the pinned tab locally is not enough: nothing would then + // tell 'a' to yield, and both would publish forever. + const { first: a, second: z } = splitBrain(['a', 'z']); + z.coordinator.setPinned(true); + vi.advanceTimersByTime(HEARTBEAT_INTERVAL * 4); + + expect(z.coordinator.isLeader()).toBe(true); + expect(a.coordinator.isLeader()).toBe(false); + }); + + it('still converges when both tabs are pinned', () => { + // Two calls in two tabs. "Pinned outranks" cannot apply symmetrically or both step down and + // nobody publishes, so it has to fall back to the id comparison. + const { first: a, second: z } = splitBrain(['a', 'z']); + a.coordinator.setPinned(true); + z.coordinator.setPinned(true); + vi.advanceTimersByTime(HEARTBEAT_INTERVAL * 4); + + expect([a, z].filter((t) => t.coordinator.isLeader())).toHaveLength(1); + expect(a.coordinator.isLeader()).toBe(true); + }); + + it('never leaves the origin without a leader once settled', () => { + const a = tab('a'); + const b = tab('b'); + const c = tab('c'); + + vi.advanceTimersByTime(LEADER_TIMEOUT + HEARTBEAT_INTERVAL * 5); + + expect([a, b, c].filter((t) => t.coordinator.isLeader())).toHaveLength(1); + }); + }); + + describe('leader departure', () => { + it('promotes a successor immediately on a clean resign, not after the timeout', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + + a.hide(); // pagehide → resign + + expect(b.coordinator.isLeader()).toBe(true); + }); + + it('promotes a successor on dispose', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + + a.coordinator.dispose(); + + expect(b.coordinator.isLeader()).toBe(true); + }); + + it('takes over after the timeout when the leader dies without resigning', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + + bus.current.kill('a'); // crash: no resign, and no further heartbeats + + expect(b.coordinator.isLeader()).toBe(false); + vi.advanceTimersByTime(LEADER_TIMEOUT + 1); + expect(b.coordinator.isLeader()).toBe(true); + }); + + it('keeps a follower from taking over while the leader is still beating', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(a.coordinator.isLeader()).toBe(true); + + // Well past the timeout, but 'a' keeps heartbeating, so 'b' must keep deferring. + vi.advanceTimersByTime(LEADER_TIMEOUT * 4); + expect(b.coordinator.isLeader()).toBe(false); + expect(a.coordinator.isLeader()).toBe(true); + }); + }); + + describe('disposal', () => { + it('stops heartbeating and stops responding', () => { + const a = tab('a'); + const b = tab('b'); + vi.advanceTimersByTime(LEADER_TIMEOUT); + + a.coordinator.dispose(); + expect(a.coordinator.isLeader()).toBe(false); + + // 'b' took over on the resign; 'a' must not resurrect itself on its old timeout. + vi.advanceTimersByTime(LEADER_TIMEOUT * 3); + expect(a.coordinator.isLeader()).toBe(false); + expect(b.coordinator.isLeader()).toBe(true); + }); + + it('unsubscribes its callbacks', () => { + const a = tab('a'); + const gained = vi.fn(); + const unsub = a.coordinator.onBecomeLeader(gained); + unsub(); + + vi.advanceTimersByTime(LEADER_TIMEOUT); + expect(gained).not.toHaveBeenCalled(); + }); + }); + + describe('message hygiene', () => { + it('ignores malformed traffic on the channel', () => { + const a = tab('a'); + const channel = bus.current.channelFor('intruder'); + channel.subscribe(() => {}); + + expect(() => { + channel.post(null); + channel.post('nonsense'); + channel.post({ type: 'resign' }); // no tabId + channel.post({ tabId: 'x' }); // no type + }).not.toThrow(); + + expect(a.coordinator.isLeader()).toBe(false); + }); + }); +}); From 4588c9aeb03fe45c077306730f5f6e25827ab5b3 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 19:16:15 +0100 Subject: [PATCH 09/11] Global spaces address updates --- we-seed.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/we-seed.json b/we-seed.json index 1ddec38e..426af281 100644 --- a/we-seed.json +++ b/we-seed.json @@ -16,8 +16,8 @@ "repoPath": "../ad4m" }, - "globalSpaceUrl": "neighbourhood://QmzSYwdaNQ5mYPy33drfzfCsvBRVb5XCaeHxgPX3BSKZTt156c3", - "marketplaceUrl": "neighbourhood://QmzSYwdqGZ6GH2hXJZAWVXnodZA3pUtVZ7BuQzMys5jQWNHkrwA", + "globalSpaceUrl": "neighbourhood://QmzSYwdZnsgCrgG4bGzmLP21AMoQTbxfwfNQeGhCXvuKqV4zXCv", + "marketplaceUrl": "neighbourhood://QmzSYwdjBtfHQgMBAFCi5nX1fxS3SQ7prh5CcysSrc2yngqL51H", "electron": { "appDistPath": "dist", From cbee81abe3e6b1d866a9a8e679cd2ca2fb17d0d5 Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 19:51:09 +0100 Subject: [PATCH 10/11] feat(presence): announce departures, and show state in the avatar ring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three changes that together make decay legible and stop it firing when it shouldn't. **bye on departure.** Leaving a space was indistinguishable from a crashed laptop: the agent simply stopped heartbeating and ran the full ladder, so it lingered for a full offlineAfter (60s) in a space it left instantly. Arrival was already one round trip via hello; departure is now symmetric. Best-effort by design — the transport is lossy, so TTL decay remains the backstop and this is only the fast path. An invisible agent stays silent: peers hold no state to retract, and a bye would disclose the departure and therefore the presence. **Two visual channels, because there are two independent facts.** tone (ring colour) carries declared availability — available/away/busy → success/warning/ danger. emphasis (opacity) carries measured liveness — online/idle/stale → full/muted/faded. Folding both into ring colour would force a false choice between showing that someone is on Do Not Disturb and showing that we are losing contact with them, and it would spend amber on a connection state when every user reads amber as "away". **Stable sort.** Most-present first, tiebroken on agentId. Not cosmetic: peers come out of a Map, so equal-liveness peers would otherwise order by insertion and the avatar row would reshuffle on every heartbeat, making a settled group look like it was churning. AvatarStack gains per-avatar tone and emphasis. Both are generic — any stack may want to distinguish or de-emphasise members — so no presence semantics leak into the design system, and tone maps to DS colour tokens rather than letting templates or stores name colours. The ring is never removed, only recoloured: it is what separates overlapping avatars. Co-Authored-By: Claude Opus 5 (1M context) --- .../frameworks/solid/stores/PresenceStore.tsx | 26 +++-- .../HeaderLayout/SpaceHeader.ts | 11 ++- .../people/AvatarStack/AvatarStack.solid.tsx | 32 ++++++- .../people/AvatarStack/AvatarStack.types.ts | 15 +++ packages/schema-system/shared/src/index.ts | 3 + .../schema-system/shared/src/presence.test.ts | 88 +++++++++++++++++ packages/schema-system/shared/src/presence.ts | 95 ++++++++++++++++++- 7 files changed, 256 insertions(+), 14 deletions(-) diff --git a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx index 0e165e60..e434d7a8 100644 --- a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx +++ b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx @@ -38,8 +38,15 @@ import type { AgentProfileSummary } from '@shared/agentHelpers'; import { createTabCoordinator } from '@shared/tabCoordinator'; import { useAdamStore } from '@solid/stores/AdamStore'; import { useRouteStore } from '@solid/stores/RouteStore'; -import type { Activity, Focus, FocusDepth, Peer, PresenceSource } from '@we/schema-shared'; -import { applyFocusDepth, callRosters, createHeartbeatPresence, peersInDataset } from '@we/schema-shared'; +import type { Activity, Focus, FocusDepth, Peer, PeerAppearance, PresenceSource } from '@we/schema-shared'; +import { + applyFocusDepth, + callRosters, + createHeartbeatPresence, + peerAppearance, + peersInDataset, + sortByPresence, +} from '@we/schema-shared'; import { type Accessor, createContext, @@ -51,8 +58,13 @@ import { useContext, } from 'solid-js'; -/** A peer joined with whatever profile the agent cache holds. `did` mirrors `agentId` for template ergonomics. */ -export type PresentAgent = Peer & Partial & { did: string }; +/** + * A peer joined with whatever profile the agent cache holds, plus its derived appearance. + * + * `did` mirrors `agentId`, and `tone`/`emphasis` are flattened rather than nested, so a template can + * reach them with a plain `$item.tone` in a `$map` — no nested path resolution needed. + */ +export type PresentAgent = Peer & Partial & { did: string } & PeerAppearance; export interface PresenceStore { /** Every peer we know of in the current space, liveness-derived, offline included. */ @@ -178,9 +190,11 @@ export function PresenceStoreProvider(props: ParentProps) { // resolves its dids; both read the same cache that `fetchAgent` populates. const peers = createMemo(() => { const cached = adamStore.agents(); - return rawPeers().map((peer) => { + // Sorted most-present-first, with a stable tiebreak — without it, equal-liveness peers reorder + // as the underlying Map iterates and the avatar row reshuffles on every heartbeat. + return sortByPresence(rawPeers()).map((peer) => { const profile = cached.find((a) => a.did === peer.agentId); - return { ...peer, ...profile, did: peer.agentId }; + return { ...peer, ...profile, ...peerAppearance(peer), did: peer.agentId }; }); }); diff --git a/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts b/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts index edc12b12..f4afb2b5 100644 --- a/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts +++ b/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts @@ -256,12 +256,19 @@ export const spaceNavBar: SchemaNode = { avatars: { $map: { items: { $store: 'presenceStore.online' }, - select: { image: '$item.avatar', hash: '$item.did' }, + select: { + image: '$item.avatar', + hash: '$item.did', + // tone = declared availability (green/amber/red), emphasis = measured + // liveness (fades as heartbeats go missing). Two channels, because a + // peer can be busy *and* fading, and one ring colour can't say both. + tone: '$item.tone', + emphasis: '$item.emphasis', + }, }, }, max: 5, size: 'sm', - ring: '0 0 0 2px var(--we-ring-color)', }, }, ], diff --git a/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx b/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx index d0f44698..797cd3ad 100644 --- a/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx +++ b/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx @@ -2,7 +2,31 @@ export type * from './AvatarStack.types'; import { createMemo, For } from 'solid-js'; -import type { AvatarStackProps } from './AvatarStack.types'; +import type { AvatarEmphasis, AvatarInfo, AvatarStackProps, AvatarTone } from './AvatarStack.types'; + +/** + * The default ring is not decoration — it is what separates overlapping avatars in the stack. So a + * toned avatar swaps the ring's *colour* and never removes it; dropping it would let faces merge. + */ +const DEFAULT_RING = '0 0 0 2px var(--we-color-neutral-0, white)'; + +const TONE_RING: Record = { + success: '0 0 0 2px var(--we-color-success-500)', + warning: '0 0 0 2px var(--we-color-warning-500)', + danger: '0 0 0 2px var(--we-color-danger-500)', + primary: '0 0 0 2px var(--we-color-primary-500)', + neutral: DEFAULT_RING, +}; + +const EMPHASIS_OPACITY: Record = { + full: '1', + muted: '0.65', + faded: '0.35', +}; + +function ringFor(avatar: AvatarInfo, fallback?: string): string { + return avatar.tone ? TONE_RING[avatar.tone] : (fallback ?? DEFAULT_RING); +} export function AvatarStack(props: AvatarStackProps) { const visible = createMemo(() => (props.avatars ?? []).slice(0, props.max ?? 5)); @@ -17,6 +41,10 @@ export function AvatarStack(props: AvatarStackProps) { display: 'flex', 'margin-left': i() > 0 ? overlapPx() : '0', 'flex-shrink': '0', + opacity: EMPHASIS_OPACITY[avatar.emphasis ?? 'full'], + // Fade rather than snap: emphasis tracks a decaying signal, so a step change reads as + // a glitch where a fade reads as "we're losing them". + transition: 'opacity var(--we-transition-400, 500ms) ease', }} > )} diff --git a/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.types.ts b/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.types.ts index c7adf93f..d8b13057 100644 --- a/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.types.ts +++ b/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.types.ts @@ -1,8 +1,22 @@ +/** Ring colour, as a semantic token rather than a CSS value, so themes stay in control. */ +export type AvatarTone = 'success' | 'warning' | 'danger' | 'primary' | 'neutral'; + +/** + * How prominent an avatar is. Separate from {@link AvatarTone} on purpose: an avatar can carry a + * status *and* be de-emphasised, and callers routinely need both at once (e.g. "on Do Not Disturb, + * and we are losing contact"). Folding them together would force a false choice. + */ +export type AvatarEmphasis = 'full' | 'muted' | 'faded'; + export interface AvatarInfo { image?: string; hash?: string; initials?: string; icon?: string; + /** Per-avatar ring colour. Overrides the stack-level `ring`. */ + tone?: AvatarTone; + /** Per-avatar prominence. Defaults to `full`. */ + emphasis?: AvatarEmphasis; } export interface AvatarStackProps { @@ -10,6 +24,7 @@ export interface AvatarStackProps { max?: number; size?: 'xxs' | 'xs' | 'sm' | 'md' | 'lg' | 'xl' | 'xxl'; overlap?: number; + /** Ring for every avatar. A per-avatar `tone` takes precedence. */ ring?: string; styles?: Record; } diff --git a/packages/schema-system/shared/src/index.ts b/packages/schema-system/shared/src/index.ts index 5764c869..9c7a79eb 100644 --- a/packages/schema-system/shared/src/index.ts +++ b/packages/schema-system/shared/src/index.ts @@ -143,8 +143,10 @@ export { callRosters, createHeartbeatPresence, derivePeers, + peerAppearance, peersInDataset, peersMatching, + sortByPresence, DEFAULT_HEARTBEAT_INTERVAL, DEFAULT_THRESHOLDS, } from './presence'; @@ -158,6 +160,7 @@ export type { LivenessThresholds, MediaSettings, Peer, + PeerAppearance, PresenceChannel, PresenceSource, PresenceState, diff --git a/packages/schema-system/shared/src/presence.test.ts b/packages/schema-system/shared/src/presence.test.ts index dd8666a0..f76e559f 100644 --- a/packages/schema-system/shared/src/presence.test.ts +++ b/packages/schema-system/shared/src/presence.test.ts @@ -8,10 +8,12 @@ import { DEFAULT_THRESHOLDS, derivePeers, type Peer, + peerAppearance, peersInDataset, peersMatching, type PresenceChannel, type PresenceState, + sortByPresence, } from './presence'; const SPACE = 'neighbourhood://QmSpace'; @@ -145,6 +147,60 @@ describe('selectors', () => { }); }); +describe('peerAppearance', () => { + const at = (overrides: Partial, age = 0) => + peerAppearance(derivePeers([state('a', { ...overrides, updatedAt: -age })], 0)[0]); + + it('takes tone from what the agent declared, not from their connection', () => { + expect(at({ availability: 'available' }).tone).toBe('success'); + expect(at({ availability: 'away' }).tone).toBe('warning'); + expect(at({ availability: 'busy' }).tone).toBe('danger'); + }); + + it('takes emphasis from the connection, not from what they declared', () => { + expect(at({}, 0).emphasis).toBe('full'); + expect(at({}, DEFAULT_THRESHOLDS.idleAfter).emphasis).toBe('muted'); + expect(at({}, DEFAULT_THRESHOLDS.staleAfter).emphasis).toBe('faded'); + }); + + it('keeps the two axes independent — a busy agent can also be fading', () => { + // The whole reason for two channels: one ring colour could not express both at once. + const busyAndFading = at({ availability: 'busy' }, DEFAULT_THRESHOLDS.staleAfter); + expect(busyAndFading).toEqual({ tone: 'danger', emphasis: 'faded' }); + }); +}); + +describe('sortByPresence', () => { + it('puts the most present first', () => { + const peers = derivePeers( + [ + state('stale', { updatedAt: -DEFAULT_THRESHOLDS.staleAfter }), + state('online', { updatedAt: 0 }), + state('idle', { updatedAt: -DEFAULT_THRESHOLDS.idleAfter }), + ], + 0, + ); + expect(sortByPresence(peers).map((p) => p.agentId)).toEqual(['online', 'idle', 'stale']); + }); + + it('orders equal-liveness peers stably, so the row does not reshuffle every heartbeat', () => { + // Peers arrive from a Map, so without a tiebreak their order follows insertion and a settled + // group of people appears to churn on each beat. + const forward = derivePeers([state('c'), state('a'), state('b')], 0); + const backward = derivePeers([state('b'), state('c'), state('a')], 0); + + expect(sortByPresence(forward).map((p) => p.agentId)).toEqual(['a', 'b', 'c']); + expect(sortByPresence(backward).map((p) => p.agentId)).toEqual(['a', 'b', 'c']); + }); + + it('does not mutate its input', () => { + const peers = derivePeers([state('b'), state('a')], 0); + const before = peers.map((p) => p.agentId); + sortByPresence(peers); + expect(peers.map((p) => p.agentId)).toEqual(before); + }); +}); + describe('callRosters', () => { it('groups concurrent calls by id — something a single callRoute field cannot express', () => { const call = (id: string): Activity => ({ type: 'call', id }); @@ -326,6 +382,38 @@ describe('createHeartbeatPresence', () => { expect(me.activities).toHaveLength(1); }); + it('announces a departure on stop, so leaving does not look like a crash', () => { + const { published, source } = start(); + published.length = 0; + + source.stop(); + + expect(published).toHaveLength(1); + expect(published[0]).toMatchObject({ v: 1, bye: true }); + }); + + it('drops a departing peer at once rather than letting it decay', () => { + const { deliver, source } = start(); + deliver('peer', { v: 1, state: state('peer') }); + expect(source.peers().map((p) => p.agentId)).toContain('peer'); + + deliver('peer', { v: 1, state: state('peer'), bye: true }); + + // Not merely 'offline' — gone. Decay is the backstop for a lost bye, not the normal path. + expect(source.peers().map((p) => p.agentId)).not.toContain('peer'); + }); + + it('stays silent on departure when invisible', () => { + // An invisible agent published nothing, so peers hold no state to retract — and a bye would + // disclose the departure, and therefore the presence. + const { published, source } = start({ availability: 'invisible' }); + published.length = 0; + + source.stop(); + + expect(published).toHaveLength(0); + }); + it('stops cleanly and publishes nothing afterwards', () => { const { published, source } = start(); source.stop(); diff --git a/packages/schema-system/shared/src/presence.ts b/packages/schema-system/shared/src/presence.ts index 3cfb533d..1a362eef 100644 --- a/packages/schema-system/shared/src/presence.ts +++ b/packages/schema-system/shared/src/presence.ts @@ -191,6 +191,65 @@ export function peersInDataset(peers: Peer[], datasetUri: string, includeOffline return peersMatching(peers, { datasetUri }, includeOffline); } +// ── Presentation ───────────────────────────────────────────────────────────── +// Semantic only — no colours, no opacity values. The design system owns how these render; this owns +// what they mean. + +/** + * How a peer should be presented, split across **two independent channels because there are two + * independent facts**. + * + * Cramming both into one (a single ring colour, say) forces a false choice — you could show that + * someone is on Do Not Disturb, or that we are losing contact with them, never both. Worse, it + * collides with convention: amber universally reads as *away*, so spending it on a connection state + * makes every user misread it. + */ +export interface PeerAppearance { + /** From `availability` — what the agent declared about themselves. */ + tone: 'success' | 'warning' | 'danger' | 'neutral'; + /** From `liveness` — how confident we are that they are still there. */ + emphasis: 'full' | 'muted' | 'faded'; +} + +const TONE_BY_AVAILABILITY: Record = { + available: 'success', + away: 'warning', + busy: 'danger', + // An invisible agent does not publish at all, so this is unreachable in practice — declared for + // totality rather than as a case that renders. + invisible: 'neutral', +}; + +const EMPHASIS_BY_LIVENESS: Record = { + online: 'full', + idle: 'muted', + stale: 'faded', + // Offline peers are filtered out before rendering; 'faded' is the safe answer if one slips through. + offline: 'faded', +}; + +export function peerAppearance(peer: Peer): PeerAppearance { + return { + tone: TONE_BY_AVAILABILITY[peer.availability] ?? 'neutral', + emphasis: EMPHASIS_BY_LIVENESS[peer.liveness] ?? 'faded', + }; +} + +const LIVENESS_RANK: Record = { online: 0, idle: 1, stale: 2, offline: 3 }; + +/** + * Most-present first. + * + * The `agentId` tiebreak is not cosmetic: peers arrive from a `Map`, so equal-liveness peers would + * otherwise order by insertion and **reshuffle on every heartbeat**, making a stable group of people + * look like it is constantly churning. Returns a new array; does not mutate. + */ +export function sortByPresence(peers: Peer[]): Peer[] { + return [...peers].sort( + (a, b) => LIVENESS_RANK[a.liveness] - LIVENESS_RANK[b.liveness] || a.agentId.localeCompare(b.agentId), + ); +} + /** Every activity of a type, paired with the peer performing it. */ export function activitiesOfType( peers: Peer[], @@ -257,12 +316,23 @@ export interface PresenceSource { stop(): void; } -/** Wire protocol. Kept minimal: a full state, plus a flag asking peers to re-announce. */ +/** Wire protocol. Kept minimal: a full state, plus two one-bit lifecycle hints. */ interface PresenceMessage { v: 1; state: PresenceState; /** Set on join. Peers seeing it re-broadcast so the joiner populates in one round trip. */ hello?: true; + /** + * Set on leave. Peers drop the agent immediately instead of watching it decay. + * + * Without this, leaving a space is indistinguishable from a crashed laptop: the agent simply stops + * heartbeating and runs the full ladder, so it lingers for a full `offlineAfter` in a space it left + * instantly. Arrival was already one round trip via `hello`; this makes departure symmetric. + * + * An **optimisation, not a guarantee** — the transport is lossy and fire-and-forget, so a lost + * `bye` just means the peer decays as before. TTL remains the backstop; this is the fast path. + */ + bye?: true; } function isPresenceMessage(payload: unknown): payload is PresenceMessage { @@ -303,15 +373,20 @@ export function createHeartbeatPresence(channel: PresenceChannel, options: Heart options.onPeersChanged?.(derivePeers(states.values(), now(), thresholds)); } - function send(hello?: true): void { + function send(lifecycle?: 'hello' | 'bye'): void { if (!running || !self) return; self = { ...self, updatedAt: now() }; states.set(self.agentId, self); // `invisible` is a hard stop, not a client-side filter. Flux keeps broadcasting full state // (including route) when invisible and merely hides it on receipt, so any modified client sees // invisible agents and where they are. + // + // `bye` is the one exception: an invisible agent has published nothing, so peers hold no state + // to retract, and sending one would disclose their departure — and therefore their presence. if (self.availability !== 'invisible') { - const message: PresenceMessage = hello ? { v: 1, state: self, hello } : { v: 1, state: self }; + const message: PresenceMessage = { v: 1, state: self }; + if (lifecycle === 'hello') message.hello = true; + if (lifecycle === 'bye') message.bye = true; channel.publish(message); } notify(); @@ -337,6 +412,13 @@ export function createHeartbeatPresence(channel: PresenceChannel, options: Heart if (!running || !isPresenceMessage(payload)) return; if (self && from === self.agentId) return; + // A departure retracts the agent outright rather than leaving it to decay. + if (payload.bye) { + states.delete(from); + notify(); + return; + } + // Trust the transport's sender id over the payload's — a peer must not be able to write into // another agent's slot. Where `authenticatedSender` is false this is still the best available // key, and the capability flag is what warns a consumer not to act on it. @@ -362,7 +444,7 @@ export function createHeartbeatPresence(channel: PresenceChannel, options: Heart self = { ...state, updatedAt: now() }; states.set(self.agentId, self); unsubscribe = channel.onMessage(receive); - send(true); + send('hello'); schedule(interval); }, @@ -400,6 +482,11 @@ export function createHeartbeatPresence(channel: PresenceChannel, options: Heart }, stop() { + // Announce the departure *before* tearing down, and while `running` is still true so `send` + // does not short-circuit. Peers drop us at once instead of watching a space change decay like + // a crash. Best-effort by design — see `PresenceMessage.bye`. + send('bye'); + running = false; if (timer !== null) { clearTimer(timer); From fbbf850047f819c72eb837818600cf6a1ed04d3a Mon Sep 17 00:00:00 2001 From: jhweir Date: Fri, 31 Jul 2026 20:04:56 +0100 Subject: [PATCH 11/11] fix(presence): colour the ring by liveness, drop the opacity channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reverts the two-channel encoding added in the previous commit. Ring colour now tracks liveness directly: green active, amber idle, red stale. The opacity channel was wrong for this component, for a reason that is structural rather than a matter of tuning: **avatars in a stack overlap**, so lowering opacity lets the avatar behind show through the one in front. The opaque separator ring is precisely what prevents that, so opacity and overlap cannot coexist. It also read as more confusing than it was worth — two axes encoded at once is more than a glanceable indicator should carry. Removes `emphasis` from AvatarStack rather than leaving it in the design system unused and quietly broken in the component's primary layout. `tone` stays; it is sound and generic. `availability` is no longer mapped to colour. Nothing can set it today — no idle detector, no manual control — so every peer reports `available` and the channel would be dead weight. A test records the current behaviour so that when that UI lands, deciding how the two axes combine is a deliberate change rather than an accident. Co-Authored-By: Claude Opus 5 (1M context) --- .../frameworks/solid/stores/PresenceStore.tsx | 12 ++-- .../HeaderLayout/SpaceHeader.ts | 7 +-- .../people/AvatarStack/AvatarStack.solid.tsx | 12 +--- .../people/AvatarStack/AvatarStack.types.ts | 17 +++--- packages/schema-system/shared/src/index.ts | 4 +- .../schema-system/shared/src/presence.test.ts | 33 +++++----- packages/schema-system/shared/src/presence.ts | 60 ++++++++----------- 7 files changed, 59 insertions(+), 86 deletions(-) diff --git a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx index e434d7a8..23647e49 100644 --- a/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx +++ b/packages/app-framework/src/frameworks/solid/stores/PresenceStore.tsx @@ -38,13 +38,13 @@ import type { AgentProfileSummary } from '@shared/agentHelpers'; import { createTabCoordinator } from '@shared/tabCoordinator'; import { useAdamStore } from '@solid/stores/AdamStore'; import { useRouteStore } from '@solid/stores/RouteStore'; -import type { Activity, Focus, FocusDepth, Peer, PeerAppearance, PresenceSource } from '@we/schema-shared'; +import type { Activity, Focus, FocusDepth, Peer, PresenceSource, PresenceTone } from '@we/schema-shared'; import { applyFocusDepth, callRosters, createHeartbeatPresence, - peerAppearance, peersInDataset, + peerTone, sortByPresence, } from '@we/schema-shared'; import { @@ -61,10 +61,10 @@ import { /** * A peer joined with whatever profile the agent cache holds, plus its derived appearance. * - * `did` mirrors `agentId`, and `tone`/`emphasis` are flattened rather than nested, so a template can - * reach them with a plain `$item.tone` in a `$map` — no nested path resolution needed. + * `did` mirrors `agentId`, and `tone` is flattened onto the peer so a template can reach it with a + * plain `$item.tone` in a `$map` — no nested path resolution needed. */ -export type PresentAgent = Peer & Partial & { did: string } & PeerAppearance; +export type PresentAgent = Peer & Partial & { did: string; tone: PresenceTone }; export interface PresenceStore { /** Every peer we know of in the current space, liveness-derived, offline included. */ @@ -194,7 +194,7 @@ export function PresenceStoreProvider(props: ParentProps) { // as the underlying Map iterates and the avatar row reshuffles on every heartbeat. return sortByPresence(rawPeers()).map((peer) => { const profile = cached.find((a) => a.did === peer.agentId); - return { ...peer, ...profile, ...peerAppearance(peer), did: peer.agentId }; + return { ...peer, ...profile, did: peer.agentId, tone: peerTone(peer) }; }); }); diff --git a/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts b/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts index f4afb2b5..8cdc8e55 100644 --- a/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts +++ b/packages/app-framework/src/shared/schemas/DefaultTemplate/HeaderLayout/SpaceHeader.ts @@ -259,11 +259,10 @@ export const spaceNavBar: SchemaNode = { select: { image: '$item.avatar', hash: '$item.did', - // tone = declared availability (green/amber/red), emphasis = measured - // liveness (fades as heartbeats go missing). Two channels, because a - // peer can be busy *and* fading, and one ring colour can't say both. + // Ring colour tracks liveness: green active, amber idle, red stale. + // Colour rather than opacity because these avatars overlap — a + // translucent one shows the avatar behind it through itself. tone: '$item.tone', - emphasis: '$item.emphasis', }, }, }, diff --git a/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx b/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx index 797cd3ad..632e01cc 100644 --- a/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx +++ b/packages/design-system/4-components/src/components/people/AvatarStack/AvatarStack.solid.tsx @@ -2,7 +2,7 @@ export type * from './AvatarStack.types'; import { createMemo, For } from 'solid-js'; -import type { AvatarEmphasis, AvatarInfo, AvatarStackProps, AvatarTone } from './AvatarStack.types'; +import type { AvatarInfo, AvatarStackProps, AvatarTone } from './AvatarStack.types'; /** * The default ring is not decoration — it is what separates overlapping avatars in the stack. So a @@ -18,12 +18,6 @@ const TONE_RING: Record = { neutral: DEFAULT_RING, }; -const EMPHASIS_OPACITY: Record = { - full: '1', - muted: '0.65', - faded: '0.35', -}; - function ringFor(avatar: AvatarInfo, fallback?: string): string { return avatar.tone ? TONE_RING[avatar.tone] : (fallback ?? DEFAULT_RING); } @@ -41,10 +35,6 @@ export function AvatarStack(props: AvatarStackProps) { display: 'flex', 'margin-left': i() > 0 ? overlapPx() : '0', 'flex-shrink': '0', - opacity: EMPHASIS_OPACITY[avatar.emphasis ?? 'full'], - // Fade rather than snap: emphasis tracks a decaying signal, so a step change reads as - // a glitch where a fade reads as "we're losing them". - transition: 'opacity var(--we-transition-400, 500ms) ease', }} > { }); }); -describe('peerAppearance', () => { - const at = (overrides: Partial, age = 0) => - peerAppearance(derivePeers([state('a', { ...overrides, updatedAt: -age })], 0)[0]); +describe('peerTone', () => { + const at = (age: number, overrides: Partial = {}) => + peerTone(derivePeers([state('a', { ...overrides, updatedAt: -age })], 0)[0]); - it('takes tone from what the agent declared, not from their connection', () => { - expect(at({ availability: 'available' }).tone).toBe('success'); - expect(at({ availability: 'away' }).tone).toBe('warning'); - expect(at({ availability: 'busy' }).tone).toBe('danger'); + it('colours by liveness — green active, amber idle, red stale', () => { + expect(at(0)).toBe('success'); + expect(at(DEFAULT_THRESHOLDS.idleAfter)).toBe('warning'); + expect(at(DEFAULT_THRESHOLDS.staleAfter)).toBe('danger'); }); - it('takes emphasis from the connection, not from what they declared', () => { - expect(at({}, 0).emphasis).toBe('full'); - expect(at({}, DEFAULT_THRESHOLDS.idleAfter).emphasis).toBe('muted'); - expect(at({}, DEFAULT_THRESHOLDS.staleAfter).emphasis).toBe('faded'); - }); - - it('keeps the two axes independent — a busy agent can also be fading', () => { - // The whole reason for two channels: one ring colour could not express both at once. - const busyAndFading = at({ availability: 'busy' }, DEFAULT_THRESHOLDS.staleAfter); - expect(busyAndFading).toEqual({ tone: 'danger', emphasis: 'faded' }); + it('does not yet vary with declared availability', () => { + // Nothing can set availability today, so showing it would be dead weight. Recorded as a test so + // that when the control lands, deciding how the two axes combine is a deliberate change here + // rather than an accident. + for (const availability of ['available', 'away', 'busy'] as const) { + expect(at(0, { availability })).toBe('success'); + } }); }); diff --git a/packages/schema-system/shared/src/presence.ts b/packages/schema-system/shared/src/presence.ts index 1a362eef..716e37eb 100644 --- a/packages/schema-system/shared/src/presence.ts +++ b/packages/schema-system/shared/src/presence.ts @@ -195,44 +195,34 @@ export function peersInDataset(peers: Peer[], datasetUri: string, includeOffline // Semantic only — no colours, no opacity values. The design system owns how these render; this owns // what they mean. +/** Semantic colour for a peer. The design system maps these to tokens; this only says what they mean. */ +export type PresenceTone = 'success' | 'warning' | 'danger' | 'neutral'; + +const TONE_BY_LIVENESS: Record = { + online: 'success', + idle: 'warning', + stale: 'danger', + // Offline peers are filtered before rendering; neutral is the safe answer if one slips through. + offline: 'neutral', +}; + /** - * How a peer should be presented, split across **two independent channels because there are two - * independent facts**. + * Colour a peer by **liveness** — the decaying signal, which is what a viewer needs to read at a + * glance: green here, amber going, red nearly gone. + * + * An earlier attempt split this across two channels — ring colour for declared `availability`, + * opacity for measured `liveness` — so both could be read at once. It was rejected in practice for a + * concrete reason worth recording: **avatars in a stack overlap**, so reducing opacity lets the + * avatar behind show through the one in front. The opaque separator ring is exactly what stops that. + * Opacity and overlap are incompatible, and the two-axis encoding was harder to read besides. * - * Cramming both into one (a single ring colour, say) forces a false choice — you could show that - * someone is on Do Not Disturb, or that we are losing contact with them, never both. Worse, it - * collides with convention: amber universally reads as *away*, so spending it on a connection state - * makes every user misread it. + * `availability` is therefore not shown yet. Nothing can set it today (no idle detector, no manual + * control), so every peer reports `available` and the channel would be dead weight. When that UI + * lands, the likely rule is that availability wins while a peer is `online` and liveness takes over + * as they degrade — but that is a decision to make with the feature, not to guess at now. */ -export interface PeerAppearance { - /** From `availability` — what the agent declared about themselves. */ - tone: 'success' | 'warning' | 'danger' | 'neutral'; - /** From `liveness` — how confident we are that they are still there. */ - emphasis: 'full' | 'muted' | 'faded'; -} - -const TONE_BY_AVAILABILITY: Record = { - available: 'success', - away: 'warning', - busy: 'danger', - // An invisible agent does not publish at all, so this is unreachable in practice — declared for - // totality rather than as a case that renders. - invisible: 'neutral', -}; - -const EMPHASIS_BY_LIVENESS: Record = { - online: 'full', - idle: 'muted', - stale: 'faded', - // Offline peers are filtered out before rendering; 'faded' is the safe answer if one slips through. - offline: 'faded', -}; - -export function peerAppearance(peer: Peer): PeerAppearance { - return { - tone: TONE_BY_AVAILABILITY[peer.availability] ?? 'neutral', - emphasis: EMPHASIS_BY_LIVENESS[peer.liveness] ?? 'faded', - }; +export function peerTone(peer: Peer): PresenceTone { + return TONE_BY_LIVENESS[peer.liveness] ?? 'neutral'; } const LIVENESS_RANK: Record = { online: 0, idle: 1, stale: 2, offline: 3 };