Fix mobile agent timeline catch-up

This commit is contained in:
Mohamed Boudra
2026-05-29 09:31:16 +07:00
parent 2e1b907c63
commit 65ace20880
11 changed files with 410 additions and 77 deletions
+2 -1
View File
@@ -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 - `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/` - 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 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) - Voice features: dictation (STT) and voice agent (realtime)
### `packages/cli` — Command-line client ### `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 - **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 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. - 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 state persists to `$PASEO_HOME/agents/{cwd-with-dashes}/{agent-id}.json` (timeline rows live alongside the record)
## Agent providers ## Agent providers
+42
View File
@@ -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`
+65 -26
View File
@@ -14,7 +14,12 @@ import {
type TimelineReducerSideEffect, type TimelineReducerSideEffect,
} from "@/timeline/session-stream-reducers"; } from "@/timeline/session-stream-reducers";
import { useCreateFlowStore } from "@/stores/create-flow-store"; 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 type { AgentAttachment, SessionOutboundMessage } from "@getpaseo/protocol/messages";
import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages"; import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages";
import { import {
@@ -500,6 +505,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const revalidationTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null); const revalidationTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const revalidationInFlightRef = useRef<Promise<void> | null>(null); const revalidationInFlightRef = useRef<Promise<void> | null>(null);
const revalidationQueuedRef = useRef(false); const revalidationQueuedRef = useRef(false);
const timelineCatchUpInFlightRef = useRef<Set<string>>(new Set());
const wasConnectedRef = useRef(isConnected); const wasConnectedRef = useRef(isConnected);
const audioOutputBuffersRef = useRef<Map<string, BufferedAudioChunk[]>>(new Map()); const audioOutputBuffersRef = useRef<Map<string, BufferedAudioChunk[]>>(new Map());
const activeAudioGroupsRef = useRef<Set<string>>(new Set()); const activeAudioGroupsRef = useRef<Set<string>>(new Set());
@@ -746,6 +752,27 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
}, AUTHORITATIVE_REVALIDATION_DEBOUNCE_MS); }, AUTHORITATIVE_REVALIDATION_DEBOUNCE_MS);
}, [client, flushAuthoritativeRevalidation, isConnected]); }, [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( const handleAppResumed = useCallback(
(awayMs: number) => { (awayMs: number) => {
scheduleAuthoritativeRevalidation(); scheduleAuthoritativeRevalidation();
@@ -754,15 +781,19 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
const session = useSessionStore.getState().sessions[serverId]; const session = useSessionStore.getState().sessions[serverId];
const agentId = session?.focusedAgentId; const agentId = session?.focusedAgentId;
if (agentId) { if (agentId) {
void client const plan = planResumeTimelineSync({
.fetchAgentTimeline(agentId, { cursor: session?.agentTimelineCursor.get(agentId),
direction: "tail", });
limit: TIMELINE_FETCH_PAGE_SIZE, if (plan.direction === "after") {
projection: "canonical", requestCanonicalCatchUp(agentId, {
}) epoch: plan.cursor.epoch,
.catch((error) => { endSeq: plan.cursor.seq,
});
} else {
void client.fetchAgentTimeline(agentId, plan).catch((error) => {
console.warn("[Session] failed to fetch tail timeline on resume", agentId, 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(serverId);
}, },
[bumpHistorySyncGeneration, client, scheduleAuthoritativeRevalidation, serverId], [
bumpHistorySyncGeneration,
client,
requestCanonicalCatchUp,
scheduleAuthoritativeRevalidation,
serverId,
],
); );
// Client activity tracking (heartbeat, push token registration) // Client activity tracking (heartbeat, push token registration)
@@ -1018,22 +1055,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
[serverId, upsertWorkspaceSetupProgress], [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( const applyTimelineResponse = useCallback(
( (
payload: Extract< payload: Extract<
@@ -1043,8 +1064,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
) => { ) => {
const agentId = payload.agentId; const agentId = payload.agentId;
const initKey = getInitKey(serverId, agentId); const initKey = getInitKey(serverId, agentId);
const catchUpComplete = isTimelineCatchUpComplete({
direction: payload.direction,
hasNewer: payload.hasNewer,
error: payload.error,
});
const shouldMarkAuthoritativeHistoryApplied = const shouldMarkAuthoritativeHistoryApplied =
payload.direction === "tail" || payload.direction === "after"; payload.direction === "tail" || (payload.direction === "after" && catchUpComplete);
// Read current store state // Read current store state
const session = useSessionStore.getState().sessions[serverId]; const session = useSessionStore.getState().sessions[serverId];
@@ -1114,6 +1140,19 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
applyAgentUpdatePayload, 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({ finalizeTimelineApplication({
result, result,
agentId, agentId,
@@ -8,7 +8,7 @@ import {
getInitKey, getInitKey,
rejectInitDeferred, rejectInitDeferred,
} from "@/utils/agent-initialization"; } 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; export const INIT_TIMEOUT_MS = 30_000;
@@ -32,19 +32,7 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput):
const session = useSessionStore.getState().sessions[serverId]; const session = useSessionStore.getState().sessions[serverId];
const cursor = session?.agentTimelineCursor.get(agentId); const cursor = session?.agentTimelineCursor.get(agentId);
const hasAuthoritativeHistory = session?.agentAuthoritativeHistoryApplied.get(agentId) === true; const hasAuthoritativeHistory = session?.agentAuthoritativeHistoryApplied.get(agentId) === true;
const timelineRequest = const timelineRequest = planInitialAgentTimelineSync({ cursor, hasAuthoritativeHistory });
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 deferred = createInitDeferred(key, timelineRequest.direction); const deferred = createInitDeferred(key, timelineRequest.direction);
const timeoutId = setTimeout(() => { const timeoutId = setTimeout(() => {
@@ -87,11 +75,7 @@ export async function refreshAgent(input: RefreshAgentInput): Promise<void> {
try { try {
await client.refreshAgent(agentId); await client.refreshAgent(agentId);
await client.fetchAgentTimeline(agentId, { await client.fetchAgentTimeline(agentId, planTimelineTailFetch());
direction: "tail",
limit: TIMELINE_FETCH_PAGE_SIZE,
projection: "canonical",
});
} catch (error) { } catch (error) {
setAgentInitializing(agentId, false); setAgentInitializing(agentId, false);
throw error; throw error;
@@ -1,7 +1,7 @@
import { useCallback } from "react"; import { useCallback } from "react";
import type { ToastApi } from "@/components/toast-host"; import type { ToastApi } from "@/components/toast-host";
import { useSessionStore, type AgentTimelineCursorState } from "@/stores/session-store"; 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 { export interface LoadOlderAgentHistoryClient {
fetchAgentTimeline: ( fetchAgentTimeline: (
@@ -40,12 +40,10 @@ export async function loadOlderAgentHistory(
setInFlight(true); setInFlight(true);
try { try {
await client.fetchAgentTimeline(agentId, { await client.fetchAgentTimeline(
direction: "before", agentId,
cursor: { epoch: cursor.epoch, seq: cursor.startSeq }, planTimelineOlderFetch({ epoch: cursor.epoch, seq: cursor.startSeq }),
limit: TIMELINE_FETCH_PAGE_SIZE, );
projection: "canonical",
});
} catch (error) { } catch (error) {
(logger ?? console).warn("[Timeline] failed to load older agent history", agentId, error); (logger ?? console).warn("[Timeline] failed to load older agent history", agentId, error);
toast?.show("Couldn't load older history", { toast?.show("Couldn't load older history", {
@@ -919,6 +919,24 @@ describe("processTimelineResponse", () => {
expect(result.clearInitializing).toBe(true); 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)", () => { it("does not resolve init when directions differ (before vs after)", () => {
const result = processTimelineResponse({ const result = processTimelineResponse({
...baseTimelineInput, ...baseTimelineInput,
@@ -167,12 +167,14 @@ function deriveBootstrapTailTimelinePolicy({
function shouldResolveTimelineInit({ function shouldResolveTimelineInit({
hasActiveInitDeferred, hasActiveInitDeferred,
hasNewer,
isInitializing, isInitializing,
initRequestDirection, initRequestDirection,
responseDirection, responseDirection,
reset, reset,
}: { }: {
hasActiveInitDeferred: boolean; hasActiveInitDeferred: boolean;
hasNewer: boolean;
isInitializing: boolean; isInitializing: boolean;
initRequestDirection: InitRequestDirection; initRequestDirection: InitRequestDirection;
responseDirection: TimelineDirection; responseDirection: TimelineDirection;
@@ -184,6 +186,9 @@ function shouldResolveTimelineInit({
if (reset) { if (reset) {
return true; return true;
} }
if (responseDirection === "after" && hasNewer) {
return false;
}
return responseDirection === initRequestDirection; return responseDirection === initRequestDirection;
} }
@@ -652,12 +657,16 @@ export function processTimelineResponse(
// ------------------------------------------------------------------ // ------------------------------------------------------------------
const shouldResolveDeferredInit = shouldResolveTimelineInit({ const shouldResolveDeferredInit = shouldResolveTimelineInit({
hasActiveInitDeferred, hasActiveInitDeferred,
hasNewer: payload.hasNewer,
isInitializing, isInitializing,
initRequestDirection, initRequestDirection,
responseDirection: payload.direction, responseDirection: payload.direction,
reset: payload.reset, 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; const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null;
@@ -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,
);
});
});
@@ -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;
}
@@ -96,6 +96,31 @@ describe("client activity tracking", () => {
}); });
} }
function waitForAssistantTimeline(
client: DaemonClient,
agentId: string,
timeout = 60000,
): Promise<string> {
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 // SINGLE CLIENT SCENARIOS
// =========================================================================== // ===========================================================================
@@ -355,6 +380,31 @@ describe("client activity tracking", () => {
// =========================================================================== // ===========================================================================
describe("web and mobile clients", () => { 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 () => { test("no notification to either when user actively on agent (web)", async () => {
client1 = await createClient(); // web client1 = await createClient(); // web
client2 = await createClient(); // mobile client2 = await createClient(); // mobile
-23
View File
@@ -783,7 +783,6 @@ export class Session {
appVisible: boolean; appVisible: boolean;
appVisibilityChangedAt: Date; appVisibilityChangedAt: Date;
} | null = null; } | null = null;
private readonly MOBILE_BACKGROUND_STREAM_GRACE_MS = 60_000;
private readonly terminalManager: TerminalManager | null; private readonly terminalManager: TerminalManager | null;
private readonly providerSnapshotManager: ProviderSnapshotManager; private readonly providerSnapshotManager: ProviderSnapshotManager;
private unsubscribeProviderSnapshotEvents: (() => void) | null = null; 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); const serializedEvent = serializeAgentStreamEvent(event.event);
if (!serializedEvent) { if (!serializedEvent) {
return; 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( private buildAgentStreamPayload(
event: Extract<AgentManagerEvent, { type: "agent_stream" }>, event: Extract<AgentManagerEvent, { type: "agent_stream" }>,
serializedEvent: Extract<SessionOutboundMessage, { type: "agent_stream" }>["payload"]["event"], serializedEvent: Extract<SessionOutboundMessage, { type: "agent_stream" }>["payload"]["event"],