diff --git a/docs/architecture.md b/docs/architecture.md index 83fe2ed41..f039d396c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -92,6 +92,7 @@ Cross-platform React Native app that connects to one or more daemons. - `SessionContext` wraps the daemon client for the active session - Composer UI and submit/draft behavior live in `packages/app/src/composer/`; screens and panels should integrate it from there instead of dropping composer internals into `components/`, `hooks/`, or `screens/workspace/` - Timeline reducers in `timeline/session-stream-reducers.ts` handle compaction, gap detection, sequence-based deduplication +- Timeline sync correctness is documented in [docs/timeline-sync.md](timeline-sync.md): live streams are for immediacy, `fetch_agent_timeline_request` is authoritative, and catch-up is paged but complete. - Voice features: dictation (STT) and voice agent (realtime) ### `packages/cli` — Command-line client @@ -222,7 +223,7 @@ initializing → idle ⇄ running - **AgentManager** is the source of truth for agent state and broadcasts updates to all subscribers - Timeline is append-only with epochs (each run starts a new epoch). Storage uses sequence numbers for client-side dedup; the default fetch page is 200 items - Timeline row `timestamp` values are canonical daemon-owned timestamps. Providers may supply original replay timestamps, but clients must not guess timestamp trust or hide time UI based on local clock heuristics. -- Events stream to all subscribed clients in real time +- Events stream to connected clients in real time; correctness is backed by authoritative timeline fetches and paged-to-completion catch-up. - Agent state persists to `$PASEO_HOME/agents/{cwd-with-dashes}/{agent-id}.json` (timeline rows live alongside the record) ## Agent providers diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md new file mode 100644 index 000000000..5caab5bef --- /dev/null +++ b/docs/timeline-sync.md @@ -0,0 +1,42 @@ +# Timeline sync + +Agent chat delivery has two paths: + +1. **Live stream** — `agent_stream` WebSocket messages for immediacy. +2. **Authoritative history** — `fetch_agent_timeline_request` for correctness. + +The invariant is: + +> If the daemon has committed timeline rows for an agent, any connected client that opens or resumes that agent eventually displays every row through the daemon's current tail. + +## Presence is not delivery + +Client heartbeat reports presence: + +- device type +- app visibility +- focused agent +- last activity time + +Heartbeat is used for notification routing. It must not be used as a correctness gate for `agent_stream` delivery. A stale mobile focus heartbeat may affect whether the user gets notified; it must not make timeline rows disappear from the live stream. + +## Catch-up is paged but complete + +Large unbounded timeline responses can exceed relay frame limits, so catch-up uses bounded pages. Bounded does not mean partial. + +When the app fetches `direction: "after"` and the daemon responds with `hasNewer: true`, the app must immediately fetch the next page from `endCursor`. The catch-up is complete only when `hasNewer: false`. + +The first load of an agent without a local cursor is different: it fetches a bounded latest tail page. Older history remains user-driven by scrolling upward. + +## Resume behavior + +When a client resumes with a known cursor, it catches up after that cursor to completion. It does not replace the view with a latest tail page, because tail pagination can skip the middle of a long background run. + +When a client resumes without a cursor, it fetches the latest tail page. + +## Relevant code + +- Server live stream forwarding: `packages/server/src/server/session.ts` +- App sync planning: `packages/app/src/timeline/timeline-sync-plan.ts` +- App stream/timeline reducer: `packages/app/src/timeline/session-stream-reducers.ts` +- Session wiring: `packages/app/src/contexts/session-context.tsx` diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 86f286240..5adae71e7 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -14,7 +14,12 @@ import { type TimelineReducerSideEffect, } from "@/timeline/session-stream-reducers"; import { useCreateFlowStore } from "@/stores/create-flow-store"; -import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import { + isTimelineCatchUpComplete, + planResumeTimelineSync, + planTimelineCatchUpAfter, + planTimelineCatchUpFollowUp, +} from "@/timeline/timeline-sync-plan"; import type { AgentAttachment, SessionOutboundMessage } from "@getpaseo/protocol/messages"; import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages"; import { @@ -500,6 +505,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const revalidationTimerRef = useRef | null>(null); const revalidationInFlightRef = useRef | null>(null); const revalidationQueuedRef = useRef(false); + const timelineCatchUpInFlightRef = useRef>(new Set()); const wasConnectedRef = useRef(isConnected); const audioOutputBuffersRef = useRef>(new Map()); const activeAudioGroupsRef = useRef>(new Set()); @@ -746,6 +752,27 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider }, AUTHORITATIVE_REVALIDATION_DEBOUNCE_MS); }, [client, flushAuthoritativeRevalidation, isConnected]); + const requestCanonicalCatchUp = useCallback( + (agentId: string, cursor: { epoch: string; endSeq: number }) => { + const request = planTimelineCatchUpAfter({ epoch: cursor.epoch, seq: cursor.endSeq }); + const key = `${agentId}:${request.cursor.epoch}:${request.cursor.seq}`; + const inFlight = timelineCatchUpInFlightRef.current; + if (inFlight.has(key)) { + return; + } + inFlight.add(key); + void client + .fetchAgentTimeline(agentId, request) + .catch((error) => { + console.warn("[Session] failed to fetch canonical catch-up timeline", agentId, error); + }) + .finally(() => { + inFlight.delete(key); + }); + }, + [client], + ); + const handleAppResumed = useCallback( (awayMs: number) => { scheduleAuthoritativeRevalidation(); @@ -754,15 +781,19 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const session = useSessionStore.getState().sessions[serverId]; const agentId = session?.focusedAgentId; if (agentId) { - void client - .fetchAgentTimeline(agentId, { - direction: "tail", - limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", - }) - .catch((error) => { + const plan = planResumeTimelineSync({ + cursor: session?.agentTimelineCursor.get(agentId), + }); + if (plan.direction === "after") { + requestCanonicalCatchUp(agentId, { + epoch: plan.cursor.epoch, + endSeq: plan.cursor.seq, + }); + } else { + void client.fetchAgentTimeline(agentId, plan).catch((error) => { console.warn("[Session] failed to fetch tail timeline on resume", agentId, error); }); + } } } @@ -771,7 +802,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } bumpHistorySyncGeneration(serverId); }, - [bumpHistorySyncGeneration, client, scheduleAuthoritativeRevalidation, serverId], + [ + bumpHistorySyncGeneration, + client, + requestCanonicalCatchUp, + scheduleAuthoritativeRevalidation, + serverId, + ], ); // Client activity tracking (heartbeat, push token registration) @@ -1018,22 +1055,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider [serverId, upsertWorkspaceSetupProgress], ); - const requestCanonicalCatchUp = useCallback( - (agentId: string, cursor: { epoch: string; endSeq: number }) => { - void client - .fetchAgentTimeline(agentId, { - direction: "after", - cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, - limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", - }) - .catch((error) => { - console.warn("[Session] failed to fetch canonical catch-up timeline", agentId, error); - }); - }, - [client], - ); - const applyTimelineResponse = useCallback( ( payload: Extract< @@ -1043,8 +1064,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider ) => { const agentId = payload.agentId; const initKey = getInitKey(serverId, agentId); + const catchUpComplete = isTimelineCatchUpComplete({ + direction: payload.direction, + hasNewer: payload.hasNewer, + error: payload.error, + }); const shouldMarkAuthoritativeHistoryApplied = - payload.direction === "tail" || payload.direction === "after"; + payload.direction === "tail" || (payload.direction === "after" && catchUpComplete); // Read current store state const session = useSessionStore.getState().sessions[serverId]; @@ -1114,6 +1140,19 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider applyAgentUpdatePayload, }); + const followUp = planTimelineCatchUpFollowUp({ + direction: payload.direction, + hasNewer: payload.hasNewer, + endCursor: payload.endCursor, + error: payload.error, + }); + if (followUp?.direction === "after") { + requestCanonicalCatchUp(agentId, { + epoch: followUp.cursor.epoch, + endSeq: followUp.cursor.seq, + }); + } + finalizeTimelineApplication({ result, agentId, diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index 8e1e72deb..1e0389363 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -8,7 +8,7 @@ import { getInitKey, rejectInitDeferred, } from "@/utils/agent-initialization"; -import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import { planInitialAgentTimelineSync, planTimelineTailFetch } from "@/timeline/timeline-sync-plan"; export const INIT_TIMEOUT_MS = 30_000; @@ -32,19 +32,7 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput): const session = useSessionStore.getState().sessions[serverId]; const cursor = session?.agentTimelineCursor.get(agentId); const hasAuthoritativeHistory = session?.agentAuthoritativeHistoryApplied.get(agentId) === true; - const timelineRequest = - hasAuthoritativeHistory && cursor - ? { - direction: "after" as const, - cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, - limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical" as const, - } - : { - direction: "tail" as const, - limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical" as const, - }; + const timelineRequest = planInitialAgentTimelineSync({ cursor, hasAuthoritativeHistory }); const deferred = createInitDeferred(key, timelineRequest.direction); const timeoutId = setTimeout(() => { @@ -87,11 +75,7 @@ export async function refreshAgent(input: RefreshAgentInput): Promise { try { await client.refreshAgent(agentId); - await client.fetchAgentTimeline(agentId, { - direction: "tail", - limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", - }); + await client.fetchAgentTimeline(agentId, planTimelineTailFetch()); } catch (error) { setAgentInitializing(agentId, false); throw error; diff --git a/packages/app/src/hooks/use-load-older-agent-history.ts b/packages/app/src/hooks/use-load-older-agent-history.ts index d9260d2ab..52e3e412e 100644 --- a/packages/app/src/hooks/use-load-older-agent-history.ts +++ b/packages/app/src/hooks/use-load-older-agent-history.ts @@ -1,7 +1,7 @@ import { useCallback } from "react"; import type { ToastApi } from "@/components/toast-host"; import { useSessionStore, type AgentTimelineCursorState } from "@/stores/session-store"; -import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import { planTimelineOlderFetch } from "@/timeline/timeline-sync-plan"; export interface LoadOlderAgentHistoryClient { fetchAgentTimeline: ( @@ -40,12 +40,10 @@ export async function loadOlderAgentHistory( setInFlight(true); try { - await client.fetchAgentTimeline(agentId, { - direction: "before", - cursor: { epoch: cursor.epoch, seq: cursor.startSeq }, - limit: TIMELINE_FETCH_PAGE_SIZE, - projection: "canonical", - }); + await client.fetchAgentTimeline( + agentId, + planTimelineOlderFetch({ epoch: cursor.epoch, seq: cursor.startSeq }), + ); } catch (error) { (logger ?? console).warn("[Timeline] failed to load older agent history", agentId, error); toast?.show("Couldn't load older history", { diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 350e6dddd..e17244b3b 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -919,6 +919,24 @@ describe("processTimelineResponse", () => { expect(result.clearInitializing).toBe(true); }); + it("keeps init open while an after catch-up page has newer rows", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: true, + initRequestDirection: "after", + payload: { + ...baseTimelineInput.payload, + direction: "after", + hasNewer: true, + entries: [], + }, + }); + + expect(result.initResolution).toBe(null); + expect(result.clearInitializing).toBe(false); + }); + it("does not resolve init when directions differ (before vs after)", () => { const result = processTimelineResponse({ ...baseTimelineInput, diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 9c1253ab8..d3723532a 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -167,12 +167,14 @@ function deriveBootstrapTailTimelinePolicy({ function shouldResolveTimelineInit({ hasActiveInitDeferred, + hasNewer, isInitializing, initRequestDirection, responseDirection, reset, }: { hasActiveInitDeferred: boolean; + hasNewer: boolean; isInitializing: boolean; initRequestDirection: InitRequestDirection; responseDirection: TimelineDirection; @@ -184,6 +186,9 @@ function shouldResolveTimelineInit({ if (reset) { return true; } + if (responseDirection === "after" && hasNewer) { + return false; + } return responseDirection === initRequestDirection; } @@ -652,12 +657,16 @@ export function processTimelineResponse( // ------------------------------------------------------------------ const shouldResolveDeferredInit = shouldResolveTimelineInit({ hasActiveInitDeferred, + hasNewer: payload.hasNewer, isInitializing, initRequestDirection, responseDirection: payload.direction, reset: payload.reset, }); - const clearInitializing = shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred); + const timelineResponseComplete = payload.direction !== "after" || !payload.hasNewer; + const clearInitializing = + (shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred)) && + timelineResponseComplete; const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null; diff --git a/packages/app/src/timeline/timeline-sync-plan.test.ts b/packages/app/src/timeline/timeline-sync-plan.test.ts new file mode 100644 index 000000000..54b121a1c --- /dev/null +++ b/packages/app/src/timeline/timeline-sync-plan.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, test } from "vitest"; +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; +import { + isTimelineCatchUpComplete, + planInitialAgentTimelineSync, + planResumeTimelineSync, + planTimelineCatchUpFollowUp, + planTimelineOlderFetch, +} from "./timeline-sync-plan"; + +describe("timeline sync planning", () => { + test("initial open without an authoritative cursor loads a bounded tail page", () => { + const plan = planInitialAgentTimelineSync({ + cursor: undefined, + hasAuthoritativeHistory: false, + }); + + expect(plan).toEqual({ + direction: "tail", + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + }); + + test("initial open with an authoritative cursor catches up after the cursor", () => { + const plan = planInitialAgentTimelineSync({ + cursor: { epoch: "epoch-1", startSeq: 1, endSeq: 42 }, + hasAuthoritativeHistory: true, + }); + + expect(plan).toEqual({ + direction: "after", + cursor: { epoch: "epoch-1", seq: 42 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + }); + + test("resume with a cursor catches up after the cursor", () => { + const plan = planResumeTimelineSync({ + cursor: { epoch: "epoch-1", startSeq: 1, endSeq: 100 }, + }); + + expect(plan).toEqual({ + direction: "after", + cursor: { epoch: "epoch-1", seq: 100 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + }); + + test("resume without a cursor loads a bounded tail page", () => { + const plan = planResumeTimelineSync({ cursor: undefined }); + + expect(plan).toEqual({ + direction: "tail", + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + }); + + test("older history loads one bounded page before the start cursor", () => { + const plan = planTimelineOlderFetch({ epoch: "epoch-1", seq: 25 }); + + expect(plan).toEqual({ + direction: "before", + cursor: { epoch: "epoch-1", seq: 25 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + }); + + test("catch-up keeps paging while the daemon reports newer rows", () => { + const plan = planTimelineCatchUpFollowUp({ + direction: "after", + hasNewer: true, + endCursor: { epoch: "epoch-1", seq: 200 }, + error: null, + }); + + expect(plan).toEqual({ + direction: "after", + cursor: { epoch: "epoch-1", seq: 200 }, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + }); + }); + + test("catch-up finishes when the daemon reports no newer rows", () => { + const plan = planTimelineCatchUpFollowUp({ + direction: "after", + hasNewer: false, + endCursor: { epoch: "epoch-1", seq: 200 }, + error: null, + }); + + expect(plan).toBeNull(); + expect(isTimelineCatchUpComplete({ direction: "after", hasNewer: false, error: null })).toBe( + true, + ); + }); +}); diff --git a/packages/app/src/timeline/timeline-sync-plan.ts b/packages/app/src/timeline/timeline-sync-plan.ts new file mode 100644 index 000000000..a9c5083f1 --- /dev/null +++ b/packages/app/src/timeline/timeline-sync-plan.ts @@ -0,0 +1,113 @@ +import { TIMELINE_FETCH_PAGE_SIZE } from "@/timeline/timeline-fetch-policy"; + +export interface TimelineSyncCursor { + epoch: string; + seq: number; +} + +export interface AgentTimelineCursorRange { + epoch: string; + startSeq: number; + endSeq: number; +} + +export interface CanonicalTimelineTailFetchPlan { + direction: "tail"; + limit: number; + projection: "canonical"; +} + +export interface CanonicalTimelineAfterFetchPlan { + direction: "after"; + cursor: TimelineSyncCursor; + limit: number; + projection: "canonical"; +} + +export interface CanonicalTimelineBeforeFetchPlan { + direction: "before"; + cursor: TimelineSyncCursor; + limit: number; + projection: "canonical"; +} + +export type CanonicalTimelineFetchPlan = + | CanonicalTimelineTailFetchPlan + | CanonicalTimelineAfterFetchPlan + | CanonicalTimelineBeforeFetchPlan; + +export type CanonicalTimelineForwardFetchPlan = + | CanonicalTimelineTailFetchPlan + | CanonicalTimelineAfterFetchPlan; + +export function planInitialAgentTimelineSync(input: { + cursor: AgentTimelineCursorRange | undefined; + hasAuthoritativeHistory: boolean; +}): CanonicalTimelineForwardFetchPlan { + if (input.hasAuthoritativeHistory && input.cursor) { + return planTimelineCatchUpAfter({ epoch: input.cursor.epoch, seq: input.cursor.endSeq }); + } + + return planTimelineTailFetch(); +} + +export function planResumeTimelineSync(input: { + cursor: AgentTimelineCursorRange | undefined; +}): CanonicalTimelineForwardFetchPlan { + if (input.cursor) { + return planTimelineCatchUpAfter({ epoch: input.cursor.epoch, seq: input.cursor.endSeq }); + } + + return planTimelineTailFetch(); +} + +export function planTimelineCatchUpAfter(cursor: TimelineSyncCursor) { + return { + direction: "after", + cursor, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + } as const; +} + +export function planTimelineTailFetch() { + return { + direction: "tail", + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + } as const; +} + +export function planTimelineOlderFetch(cursor: TimelineSyncCursor) { + return { + direction: "before", + cursor, + limit: TIMELINE_FETCH_PAGE_SIZE, + projection: "canonical", + } as const; +} + +export function planTimelineCatchUpFollowUp(input: { + direction: "tail" | "before" | "after"; + hasNewer: boolean; + endCursor: TimelineSyncCursor | null; + error: string | null; +}): CanonicalTimelineAfterFetchPlan | null { + if (input.error || input.direction !== "after" || !input.hasNewer || !input.endCursor) { + return null; + } + + return planTimelineCatchUpAfter(input.endCursor); +} + +export function isTimelineCatchUpComplete(input: { + direction: "tail" | "before" | "after"; + hasNewer: boolean; + error: string | null; +}): boolean { + if (input.error) { + return false; + } + + return input.direction !== "after" || !input.hasNewer; +} diff --git a/packages/server/src/server/client-activity.e2e.test.ts b/packages/server/src/server/client-activity.e2e.test.ts index c572654d5..eaf5ae054 100644 --- a/packages/server/src/server/client-activity.e2e.test.ts +++ b/packages/server/src/server/client-activity.e2e.test.ts @@ -96,6 +96,31 @@ describe("client activity tracking", () => { }); } + function waitForAssistantTimeline( + client: DaemonClient, + agentId: string, + timeout = 60000, + ): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + cleanup(); + reject(new Error(`Timeout waiting for assistant timeline (${timeout}ms)`)); + }, timeout); + + const cleanup = client.on("agent_stream", (msg) => { + if (msg.type !== "agent_stream") return; + if (msg.payload.agentId !== agentId) return; + if (msg.payload.event.type !== "timeline") return; + const item = msg.payload.event.item; + if (item.type !== "assistant_message") return; + + clearTimeout(timer); + cleanup(); + resolve(item.text); + }); + }); + } + // =========================================================================== // SINGLE CLIENT SCENARIOS // =========================================================================== @@ -355,6 +380,31 @@ describe("client activity tracking", () => { // =========================================================================== describe("web and mobile clients", () => { + test("mobile stream delivery does not depend on heartbeat focus", async () => { + client1 = await createClient(); // web + client2 = await createClient(); // mobile + + const agent = await createAgent({ + client: client1, + title: "Mobile Stale Focus Stream Test", + }); + + client2.sendHeartbeat({ + deviceType: "mobile", + focusedAgentId: null, + lastActivityAt: new Date(Date.now() - PRESENCE_THRESHOLD_MS - 60_000).toISOString(), + appVisible: false, + appVisibilityChangedAt: new Date(Date.now() - 120_000).toISOString(), + }); + + await new Promise((r) => setTimeout(r, 100)); + + const mobileStream = waitForAssistantTimeline(client2, agent.id); + await client1.sendMessage(agent.id, "Say 'hello' and nothing else"); + + await expect(mobileStream).resolves.toMatch(/hello/i); + }, 120000); + test("no notification to either when user actively on agent (web)", async () => { client1 = await createClient(); // web client2 = await createClient(); // mobile diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index f030a23f8..23778fb6f 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -783,7 +783,6 @@ export class Session { appVisible: boolean; appVisibilityChangedAt: Date; } | null = null; - private readonly MOBILE_BACKGROUND_STREAM_GRACE_MS = 60_000; private readonly terminalManager: TerminalManager | null; private readonly providerSnapshotManager: ProviderSnapshotManager; private unsubscribeProviderSnapshotEvents: (() => void) | null = null; @@ -1302,13 +1301,6 @@ export class Session { }); } - // Reduce bandwidth/CPU on mobile: only forward high-frequency agent stream events - // for the focused agent, with a short grace window while backgrounded. - // History catch-up is handled via pull-based `fetch_agent_timeline_request`. - if (this.shouldSkipAgentStreamForward(event.agentId)) { - return; - } - const serializedEvent = serializeAgentStreamEvent(event.event); if (!serializedEvent) { return; @@ -1355,21 +1347,6 @@ export class Session { ); } - private shouldSkipAgentStreamForward(agentId: string): boolean { - const activity = this.clientActivity; - if (activity?.deviceType !== "mobile") { - return false; - } - if (!activity.focusedAgentId || activity.focusedAgentId !== agentId) { - return true; - } - if (activity.appVisible) { - return false; - } - const hiddenForMs = Date.now() - activity.appVisibilityChangedAt.getTime(); - return hiddenForMs >= this.MOBILE_BACKGROUND_STREAM_GRACE_MS; - } - private buildAgentStreamPayload( event: Extract, serializedEvent: Extract["payload"]["event"],