fix(presence): keep header avatars for everyone viewing an entity

apps/web/src/lib/core/signal/tabFocus.ts

@@ -20,6 +20,16 @@ createEffect(() => {
   });
 });
 
+const [isTabVisible_, setIsTabVisible] = createSignal(
+  document.visibilityState === 'visible'
+);
+document.addEventListener('visibilitychange', () =>
+  setIsTabVisible(document.visibilityState === 'visible')
+);
+
+/** Whether the tab is on screen, focused or not (e.g. beside a focused window) */
+export const isTabVisible = isTabVisible_;
+
 /** Whether the tab is currently focused */
 export const isTabFocused = createMemo(() => isTabFocused_(), {
   equalFn: (a: boolean | undefined, b: boolean | undefined) => a! === b!,

apps/web/src/lib/core/state/liveIndicators.test.ts

@@ -0,0 +1,63 @@
+import { createRenderEffect, createRoot } from 'solid-js';
+import { expect, it, vi } from 'vitest';
+
+const mocks = vi.hoisted(() => ({
+  trackingChange: undefined as ((event: { data: string }) => void) | undefined,
+}));
+vi.mock('@core/constant/featureFlags', () => ({
+  ENABLE_LIVE_INDICATORS: true,
+}));
+vi.mock('@service-connection/websocket', () => ({ ws: {} }));
+vi.mock('@service-connection/presence-telemetry', () => ({
+  reportTrackingChange: vi.fn(),
+  reportInvalidTrackingChange: vi.fn(),
+}));
+vi.mock('@macro-inc/collaboration/websocket', () => ({
+  createWebsocketEventEffect: (
+    _ws: unknown,
+    _type: string,
+    callback: (event: { data: string }) => void
+  ) => {
+    mocks.trackingChange = callback;
+  },
+}));
+
+const { useUserIndicators } = await import('./liveIndicators');
+
+it('re-renders an entity viewers on each change to that entity only', () => {
+  const rendered: (string[] | undefined)[] = [];
+  const dispose = createRoot((dispose) => {
+    const indicators = useUserIndicators(() => 'document-id');
+    createRenderEffect(() => rendered.push(indicators()));
+    return dispose;
+  });
+
+  mocks.trackingChange!({
+    data: JSON.stringify({
+      entity_id: 'document-id',
+      entity_type: 'document',
+      user_ids: ['macro|first@example.com'],
+    }),
+  });
+  mocks.trackingChange!({
+    data: JSON.stringify({
+      entity_id: 'document-id',
+      entity_type: 'document',
+      user_ids: ['macro|first@example.com', 'macro|second@example.com'],
+    }),
+  });
+  mocks.trackingChange!({
+    data: JSON.stringify({
+      entity_id: 'other-document-id',
+      entity_type: 'document',
+      user_ids: [],
+    }),
+  });
+  dispose();
+
+  expect(rendered).toEqual([
+    undefined,
+    ['macro|first@example.com'],
+    ['macro|first@example.com', 'macro|second@example.com'],
+  ]);
+});

apps/web/src/lib/queries/messages/tests/subscription.test.ts

@@ -10,12 +10,12 @@ import { useMessageThreadQuery } from '../thread-replies';
 
 const mocks = await vi.hoisted(async () => {
   const { createSignal } = await import('solid-js');
-  const [focused, setFocused] = createSignal(true);
+  const [visible, setVisible] = createSignal(true);
   return {
     send: vi.fn(),
     clearStream: vi.fn(),
-    focused,
-    setFocused,
+    visible,
+    setVisible,
     invalidate: vi.fn().mockResolvedValue(undefined),
     updates: new Set<(event: unknown) => void>(),
     reconnects: new Set<() => void>(),
@@ -25,7 +25,11 @@ const mocks = await vi.hoisted(async () => {
 vi.mock('@macro-inc/observability', () => ({
   Telemetry: { info: vi.fn(), warn: mocks.warn, error: vi.fn() },
 }));
-vi.mock('@core/signal/tabFocus', () => ({ isTabFocused: mocks.focused }));
+// Focus is elsewhere throughout: presence follows visibility alone.
+vi.mock('@core/signal/tabFocus', () => ({
+  isTabFocused: () => false,
+  isTabVisible: mocks.visible,
+}));
 vi.mock('@service-connection/stream', () => ({
   clearStream: mocks.clearStream,
 }));
@@ -85,7 +89,7 @@ function sent(action: string, id = parent.id) {
 beforeEach(() => {
   vi.useFakeTimers();
   vi.clearAllMocks();
-  mocks.setFocused(true);
+  mocks.setVisible(true);
 });
 afterEach(() => {
   for (const dispose of disposers.splice(0)) dispose();
@@ -213,25 +217,31 @@ it('refreshes cached messages after all views close and the parent reopens', ()
   expect(mocks.invalidate).toHaveBeenCalledTimes(3);
 });
 
+it('keeps the heartbeat in a visible tab whose window is not focused', () => {
+  mountBlock();
+  vi.advanceTimersByTime(60_000);
+  expect(sent('ping')).toHaveLength(3);
+});
+
 it('suspends the shared heartbeat in a background tab', () => {
   mountBlock();
-  mocks.setFocused(false);
+  mocks.setVisible(false);
   vi.advanceTimersByTime(40_000);
   expect(sent('ping')).toHaveLength(0);
-  mocks.setFocused(true);
+  mocks.setVisible(true);
   vi.advanceTimersByTime(20_000);
   expect(sent('ping')).toHaveLength(1);
 });
 
-it('reports a presence lapse once per unfocused absence past the gateway window', () => {
+it('reports a presence lapse once per hidden absence past the gateway window', () => {
   mount(useRefreshTrackedEntitiesOnFocus);
   mountBlock();
   const lapses = () =>
     mocks.warn.mock.calls.filter(
       ([name]) => name === 'presence.heartbeat_lapsed'
     );
 
-  mocks.setFocused(false);
+  mocks.setVisible(false);
   vi.advanceTimersByTime(40_000);
   expect(lapses()).toHaveLength(0);
   vi.advanceTimersByTime(40_000);
@@ -243,13 +253,13 @@ it('reports a presence lapse once per unfocused absence past the gateway window'
   vi.advanceTimersByTime(60_000);
   expect(lapses()).toHaveLength(1);
 
-  mocks.setFocused(true);
-  mocks.setFocused(false);
+  mocks.setVisible(true);
+  mocks.setVisible(false);
   vi.advanceTimersByTime(60_000);
   expect(lapses()).toHaveLength(2);
 });
 
-it('pings on refocus and refreshes only views left unseen past the gateway window', () => {
+it('pings when visible again and refreshes only views left unseen past the gateway window', () => {
   mount(useRefreshTrackedEntitiesOnFocus);
   mount(() =>
     useMessageThreadQuery(
@@ -259,20 +269,20 @@ it('pings on refocus and refreshes only views left unseen past the gateway windo
   );
   mocks.invalidate.mockClear();
 
-  // Heartbeats keep the gateway's view fresh while focused.
+  // Heartbeats keep the gateway's view fresh while visible, focused or not.
   vi.advanceTimersByTime(80_000);
   expect(sent('ping')).toHaveLength(4);
-  mocks.setFocused(false);
+  mocks.setVisible(false);
   vi.advanceTimersByTime(30_000);
-  mocks.setFocused(true);
+  mocks.setVisible(true);
   expect(sent('ping')).toHaveLength(5);
   expect(mocks.invalidate).not.toHaveBeenCalled();
 
   // Away past the window: the gateway stopped sending this parent's events.
-  mocks.setFocused(false);
+  mocks.setVisible(false);
   vi.advanceTimersByTime(60_000);
   expect(sent('ping')).toHaveLength(5);
-  mocks.setFocused(true);
+  mocks.setVisible(true);
   expect(sent('ping')).toHaveLength(6);
   expect(mocks.invalidate).toHaveBeenCalledTimes(3);
   expect(mocks.invalidate).toHaveBeenCalledWith({

apps/web/src/lib/service-clients/service-connection/client.ts

@@ -1,4 +1,4 @@
-import { isTabFocused } from '@core/signal/tabFocus';
+import { isTabVisible } from '@core/signal/tabFocus';
 import type { EntityId } from '@core/types';
 import { fireAndForget } from '@core/util/fireAndForget';
 import { createReconnectEffect } from '@macro-inc/collaboration/websocket';
@@ -51,8 +51,10 @@ export const connectionGatewayClient = {
         trackedEntities.set(args.entity_id, {
           entityType: args.entity_type,
           count: 1,
+          // Visible, not focused: someone reading beside the window they are
+          // typing in is still looking, and should keep their presence dot.
           heartbeat: setInterval(() => {
-            if (isTabFocused())
+            if (isTabVisible())
               fireAndForget(
                 connectionGatewayClient.trackEntity({
                   ...args,
@@ -171,16 +173,16 @@ export function useReopenTrackedEntitiesOnReconnect(): void {
 }
 
 /**
- * Pings every tracked entity as soon as the tab regains focus, since the heartbeat pauses in
+ * Pings every tracked entity as soon as the tab is visible again, since the heartbeat pauses in
  * the background. A tab that went unseen longer than the gateway's activity window was left
  * out of the entity's events meanwhile, so its views refresh as they do after a reconnect.
  */
 export function useRefreshTrackedEntitiesOnFocus(): void {
   createEffect(
     on(
-      isTabFocused,
-      (focused) => {
-        if (!focused) return;
+      isTabVisible,
+      (visible) => {
+        if (!visible) return;
         const now = Date.now();
         for (const [entity_id, tracked] of trackedEntities) {
           const entity = { entity_id, entity_type: tracked.entityType };

services/agent_harness_service/src/cognition/api/context/test.rs

@@ -119,7 +119,7 @@ impl ConnectionRepo for MockConnectionRepo {
         &self,
         _entity: &EntityConnection<'_>,
         _timestamp: u64,
-    ) -> anyhow::Result<StoredConnectionEntity> {
+    ) -> anyhow::Result<Option<StoredConnectionEntity>> {
         unimplemented!()
     }
     async fn update_user_connection_last_ping(

services/connection_gateway/AGENTS.md

@@ -50,6 +50,10 @@ the socket themselves.
 - A send can target a connection on a different instance — it always goes
   through the Redis relay in `src/service/redis.rs`, never a direct in-process
   lookup; don't bypass it when adding a new send path.
+- Presence (`user_tracking_change`) counts an entity row as active for 60s
+  after its last `open` or `ping`, compared across instances' clocks. A `ping`
+  for a row that is gone re-creates it, so a lost `open` heals on the next
+  heartbeat.
 
 ## Related
 

services/connection_gateway/src/model/connection.rs

@@ -39,15 +39,20 @@ impl StoredConnectionEntity {
             .duration_since(UNIX_EPOCH)
             .unwrap()
             .as_millis() as u64;
+        self.is_active_at(current_timestamp, threshold)
+    }
 
-        match (threshold, self.last_ping) {
-            // There is a threshold and the last ping is older than the threshold
-            (Some(t), Some(last_ping)) => (current_timestamp - last_ping) < t,
-            // There is no threshold compare the current timestamp with the created_at
-            (Some(t), None) => (current_timestamp - self.created_at) < t,
-            // No threshold then just return true
-            _ => true,
-        }
+    /// Whether this entry was opened or pinged within `threshold` of `now`.
+    ///
+    /// Gateway instances stamp `last_ping` with their own clocks, so a fresh
+    /// ping can read as slightly in the future on another instance; that
+    /// counts as active rather than wrapping into a huge age.
+    fn is_active_at(&self, now: u64, threshold: Option<u64>) -> bool {
+        let Some(threshold) = threshold else {
+            return true;
+        };
+        let last_seen = self.last_ping.unwrap_or(self.created_at);
+        now.saturating_sub(last_seen) < threshold
     }
 }
 
@@ -91,3 +96,6 @@ impl AsRef<Config> for ConnectionContext<'_> {
         self.config
     }
 }
+
+#[cfg(test)]
+mod tests;

services/connection_gateway/src/model/connection/tests.rs

@@ -0,0 +1,51 @@
+use super::*;
+
+#[test]
+fn a_ping_stamped_ahead_of_this_instance_clock_is_active() {
+    let entry = StoredConnectionEntity {
+        pk: "#document#doc".to_owned(),
+        sk: "connection".to_owned(),
+        entity_type: EntityType::Document,
+        entity_id: "doc".to_owned(),
+        connection_id: "connection".to_owned(),
+        created_at: 1_000_000,
+        user_id: "macro|user@example.com".to_owned(),
+        last_ping: Some(1_000_500),
+    };
+
+    assert!(entry.is_active_at(1_000_000, Some(60_000)));
+}
+
+#[test]
+fn an_entry_not_pinged_within_the_threshold_is_inactive() {
+    let entry = StoredConnectionEntity {
+        pk: "#document#doc".to_owned(),
+        sk: "connection".to_owned(),
+        entity_type: EntityType::Document,
+        entity_id: "doc".to_owned(),
+        connection_id: "connection".to_owned(),
+        created_at: 1_000_000,
+        user_id: "macro|user@example.com".to_owned(),
+        last_ping: Some(1_000_000),
+    };
+
+    assert!(entry.is_active_at(1_059_999, Some(60_000)));
+    assert!(!entry.is_active_at(1_060_000, Some(60_000)));
+}
+
+#[test]
+fn an_open_entry_never_pinged_counts_from_creation() {
+    let entry = StoredConnectionEntity {
+        pk: "#document#doc".to_owned(),
+        sk: "connection".to_owned(),
+        entity_type: EntityType::Document,
+        entity_id: "doc".to_owned(),
+        connection_id: "connection".to_owned(),
+        created_at: 1_000_000,
+        user_id: "macro|user@example.com".to_owned(),
+        last_ping: None,
+    };
+
+    assert!(entry.is_active_at(999_000, Some(60_000)));
+    assert!(!entry.is_active_at(1_060_000, Some(60_000)));
+}