puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-05 16:04:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert #4421 and narrow the cache restore fix (#4436)

Browse Source 
* Revert "Restore cached conversations before reconnecting (#4421)"

This reverts commit 1c51b52d07.

* fix(app): restore provider loading on native

* fix(app): preserve native navigators across appearance changes

Reapply the isolated appearance boundaries and native menu transform origin from #4421 so appearance hydration preserves native navigator and gesture host lifetimes.

* fix(app): retain cached baselines during live updates

Read the timeline concurrently with agent preparation and reconcile display-only cache rows beneath the live head using the existing reducer. Merge cached directory entries beneath live state while preserving the existing checkpoint race guard.
This commit is contained in:
Mohamed Boudra
2026-09-07 19:18:54 +02:00
committed by GitHub
parent 9c8d775895
commit c46aee4e6a
52 changed files with 3422 additions and 8638 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/runtime/directory-sync/agent-replica.test.ts
-14
View File
@@ -61,7 +61,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
() => undefined,
() => undefined,
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
replica.commitSnapshot([entry(payload("live"))], []);
replica.applyTurnLiveness("agent", {
@@ -85,7 +84,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
() => undefined,
() => undefined,
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
expect(replica.submitTimelineAgent(replica.captureTimeline("agent"), payload("network"))).toBe(
@@ -106,7 +104,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
() => undefined,
(mutations) => commits.push([...mutations]),
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
replica.commitSnapshot([entry(payload("agent"))], []);
commits.length = 0;
@@ -147,7 +144,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
(agentId) => stopped.push(agentId),
() => undefined,
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
replica.commitSnapshot([entry(payload("agent"))], []);
replica.applyTurnLiveness("agent", {
@@ -170,7 +166,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
() => undefined,
() => undefined,
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
replica.commitSnapshot(
[
@@ -224,7 +219,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
(agentId) => stopped.push(agentId),
() => undefined,
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
replica.commitSnapshot(
[
@@ -268,7 +262,6 @@ describe("AgentDirectoryReplica", () => {
serverId,
() => undefined,
() => undefined,
(agentId) => useSessionStore.getState().removeAgentTimeline(serverId, agentId),
);
replica.commitSnapshot(
[
@@ -296,16 +289,10 @@ describe("AgentDirectoryReplica", () => {
const serverId = "agent-replica";
const store = useSessionStore.getState();
store.initializeSession(serverId, null as unknown as DaemonClient);
const timelineLifetimes: string[] = [];
const replica = new AgentDirectoryReplica(
serverId,
() => undefined,
() => undefined,
(agentId) => {
timelineLifetimes.push(`remove:${agentId}`);
useSessionStore.getState().removeAgentTimeline(serverId, agentId);
},
(agentId) => timelineLifetimes.push(`accept:${agentId}`),
);
replica.commitSnapshot([entry(payload("directory"))], []);
const directoryPlacement = useSessionStore
@@ -332,7 +319,6 @@ describe("AgentDirectoryReplica", () => {
expect(
useSessionStore.getState().sessions[serverId]?.agents.get("agent")?.projectPlacement,
).toEqual(directoryPlacement);
expect(timelineLifetimes).toEqual(["accept:agent", "remove:agent", "accept:agent"]);
store.clearSession(serverId);
});
});
packages/app/src/runtime/directory-sync/agent-replica.ts
+28 -42
View File
@@ -4,7 +4,7 @@ import { clearArchiveAgentPending } from "@/hooks/use-archive-agent";
import { queryClient } from "@/data/query-client";
import type { Agent } from "@/stores/session-store";
import { normalizeAgentSnapshot, projectAgentSnapshot } from "@/utils/agent-snapshots";
import { type AgentDirectoryDelta, type AgentRemovalReason } from "@/utils/agent-directory-sync";
import { type AgentDirectoryDelta } from "@/utils/agent-directory-sync";
import { reconcileAgentDirectory } from "@/utils/agent-directory-reconciliation";
import { applyLegacyDaemonWorkspaceOwnership } from "@/workspace/legacy-daemon-workspaces";
import type { DirectoryReplicaMutation } from "@/runtime/replica-cache";
@@ -25,47 +25,52 @@ export interface AgentLifecycleToken {
export class AgentDirectoryReplica {
private readonly lifecycleVersions = new Map<string, number>();
private readonly members = new Set<string>();
private readonly pendingCacheReads = new Set<string>();
private readonly storeProjection: AgentStoreProjection;
constructor(
private readonly serverId: string,
private readonly onStoppedRunning: (agentId: string) => void,
private readonly persist: (mutations: readonly DirectoryReplicaMutation[]) => void,
removeTimeline: (agentId: string, reason: AgentRemovalReason) => void,
private readonly acceptTimeline: (agentId: string) => void = () => undefined,
) {
this.storeProjection = new AgentStoreProjection(serverId, removeTimeline);
this.storeProjection = new AgentStoreProjection(serverId);
}
captureTimeline(agentId: string): AgentLifecycleToken {
return { agentId, version: this.lifecycleVersions.get(agentId) ?? 0 };
}
isTimelineCurrent(token: AgentLifecycleToken): boolean {
return token.version === (this.lifecycleVersions.get(token.agentId) ?? 0);
}
snapshot(): Map<string, Agent> {
return this.storeProjection.snapshot();
}
// Saved rows sit under whatever the live stream already delivered for this connection.
captureCache(agentId: string): AgentLifecycleToken {
this.pendingCacheReads.add(agentId);
return this.captureTimeline(agentId);
}
commitCached(agents: Map<string, Agent>): void {
const current = new Map(
[...agents].filter(([id]) => !this.lifecycleVersions.has(id) || this.members.has(id)),
);
const merged = this.storeProjection.commitCached(current);
const merged = this.storeProjection.commitCached(agents);
this.members.clear();
for (const agentId of merged.keys()) {
this.members.add(agentId);
}
}
commitCachedAgent(token: AgentLifecycleToken, agent: Agent): boolean {
this.pendingCacheReads.delete(token.agentId);
if (token.version !== (this.lifecycleVersions.get(token.agentId) ?? 0)) return false;
if (this.members.has(agent.id)) return false;
this.members.add(agent.id);
this.storeProjection.accept(agent);
this.storeProjection.publishActivity(agent);
return true;
}
submitTimelineAgent(token: AgentLifecycleToken, payload: AgentSnapshotPayload): boolean {
if (!this.isTimelineCurrent(token)) {
if (token.version !== (this.lifecycleVersions.get(token.agentId) ?? 0)) {
return false;
}
const startsLifetime = !this.members.has(token.agentId);
const existing = this.storeProjection.get(token.agentId);
const timelineAgent = applyLegacyDaemonWorkspaceOwnership({
serverId: this.serverId,
@@ -76,10 +81,6 @@ export class AgentDirectoryReplica {
projectPlacement: timelineAgent.projectPlacement ?? existing?.projectPlacement,
};
const accepted = this.storeProjection.accept(normalized);
if (startsLifetime) {
this.resume(accepted.id);
this.acceptTimeline(accepted.id);
}
this.members.add(accepted.id);
this.storeProjection.replacePendingPermissions(accepted);
this.storeProjection.publishActivity(accepted);
@@ -98,10 +99,7 @@ export class AgentDirectoryReplica {
this.advance(delta.agentId);
} else {
this.members.add(delta.agent.id);
if (!before) {
this.resume(delta.agent.id);
this.acceptTimeline(delta.agent.id);
}
if (!before) this.advance(delta.agent.id);
}
if (result.stoppedRunning) this.onStoppedRunning(result.agentId);
this.persist(
@@ -112,12 +110,7 @@ export class AgentDirectoryReplica {
}
accept(agent: Agent): Agent {
const startsLifetime = !this.members.has(agent.id);
const accepted = this.storeProjection.accept(agent);
if (startsLifetime) {
this.resume(accepted.id);
this.acceptTimeline(accepted.id);
}
this.members.add(accepted.id);
this.persist([this.agentUpsert(accepted)]);
return accepted;
@@ -131,23 +124,22 @@ export class AgentDirectoryReplica {
const previous = this.storeProjection.snapshot();
const reconciled = reconcileAgentDirectory({ snapshot: entries, deltas });
const nextIds = new Set(reconciled.map((entry) => entry.agent.id));
const startedLifetimes = new Set<string>();
for (const agentId of this.pendingCacheReads) {
if (!nextIds.has(agentId)) this.advance(agentId);
}
for (const agentId of this.members) {
if (!nextIds.has(agentId)) this.advance(agentId);
}
for (const agentId of nextIds) {
if (!this.members.has(agentId)) {
this.resume(agentId);
startedLifetimes.add(agentId);
}
if (!this.members.has(agentId)) this.advance(agentId);
}
for (const agentId of previous.keys()) {
if (!nextIds.has(agentId)) this.storeProjection.remove(agentId, "scope");
if (!nextIds.has(agentId)) this.storeProjection.remove(agentId);
}
this.members.clear();
this.pendingCacheReads.clear();
for (const agentId of nextIds) this.members.add(agentId);
const agents = this.storeProjection.replaceFetched(reconciled);
for (const agentId of startedLifetimes) this.acceptTimeline(agentId);
for (const [agentId, previousAgent] of previous) {
if (previousAgent.turn.phase === "open" && agents.get(agentId)?.turn.phase === "idle") {
this.onStoppedRunning(agentId);
@@ -209,7 +201,7 @@ export class AgentDirectoryReplica {
remove(agentId: string): void {
this.members.delete(agentId);
this.advance(agentId);
this.storeProjection.remove(agentId, "deleted");
this.storeProjection.remove(agentId);
this.persist([{ kind: "agent", type: "delete", id: agentId }]);
}
@@ -233,12 +225,6 @@ export class AgentDirectoryReplica {
};
}
private resume(agentId: string): void {
// First admission keeps requests already in flight. Re-entry invalidates work issued
// during the excluded lifetime as well as work issued before its removal.
if (this.lifecycleVersions.has(agentId)) this.advance(agentId);
}
private advance(agentId: string): void {
this.lifecycleVersions.set(agentId, (this.lifecycleVersions.get(agentId) ?? 0) + 1);
}
packages/app/src/runtime/directory-sync/index.test.ts
+325 -544
View File
File diff suppressed because it is too large Load Diff
packages/app/src/runtime/directory-sync/index.ts
+166 -86
View File
@@ -3,6 +3,7 @@ import type {
FetchAgentsEntry,
FetchAgentsOptions,
} from "@getpaseo/client/internal/daemon-client";
import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once";
import {
normalizeProjectDescriptor,
normalizeWorkspaceDescriptor,
@@ -17,7 +18,7 @@ import {
shouldUseLegacyDaemonWorkspaceDirectory,
stampLegacyWorkspaceIds,
} from "@/workspace/legacy-daemon-workspaces";
import type { AgentDirectoryDelta, AgentRemovalReason } from "@/utils/agent-directory-sync";
import type { AgentDirectoryDelta } from "@/utils/agent-directory-sync";
import { AgentDirectoryReplica } from "./agent-replica";
import {
WorkspaceDirectoryReplica,
@@ -32,14 +33,13 @@ import {
import { workspaceLabels } from "@/workspace-labels";
import type {
CachedDirectory,
CachedWorkspace,
DirectoryCheckpoint,
DirectoryCursor,
DirectoryReplicaMutation,
} from "@/runtime/replica-cache";
import type { TurnLivenessTransition } from "@/timeline/turn-liveness";
type TimelinePage = Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>;
const PAGE_LIMIT = 200;
const AGENT_SORT: NonNullable<FetchAgentsOptions["sort"]> = [
{ key: "updated_at", direction: "desc" },
@@ -108,6 +108,8 @@ export interface DirectoryConnection {
}
export interface DirectoryCheckpointStorage {
readAgent(serverId: string, agentId: string): Promise<Agent | undefined>;
readWorkspace(serverId: string, workspaceId: string): Promise<CachedWorkspace | undefined>;
readDirectory(serverId: string): Promise<CachedDirectory>;
commitDirectoryMutations(
serverId: string,
@@ -137,7 +139,6 @@ export class DirectorySync {
WorkspaceDirectorySnapshot,
WorkspaceDirectoryDelta
>();
private readonly timelineRequests = new Map<string, Promise<TimelinePage>>();
private readonly agents: AgentDirectoryReplica;
private readonly workspaces: WorkspaceDirectoryReplica;
private connection: DirectoryConnection = {
@@ -147,10 +148,10 @@ export class DirectorySync {
};
private unsubscribe: (() => void) | null = null;
private readonly abortSessionWaits = new Set<() => void>();
// The saved baseline is restored once, at construction. Live mutations apply as they arrive and
// the baseline merges under them; network refreshes wait for it so their transactions and
// cursors start from the saved state.
private readonly baseline: Promise<void>;
private cacheLoad: Promise<void> | null = null;
private cacheAccepted = false;
private revision = 0;
private workspaceRevision = 0;
private readonly routeDemandIds = new Set<string>();
private readonly fullDemandSources = new Set<object>();
private demandRefresh: Promise<void> | null = null;
@@ -160,10 +161,7 @@ export class DirectorySync {
constructor(
private readonly serverId: string,
private readonly callbacks: {
onTimelineRequest: (agentId: string, replaceTail: boolean) => (page: TimelinePage) => void;
onAgentStoppedRunning: (agentId: string) => void;
onAgentRemoved: (agentId: string, reason: AgentRemovalReason) => void;
onAgentAccepted?: (agentId: string) => void;
markAgentLoading: () => void;
markAgentReady: () => void;
markAgentError: (error: string) => void;
@@ -172,19 +170,8 @@ export class DirectorySync {
) {
const persist = (mutations: readonly DirectoryReplicaMutation[]) =>
this.checkpoints?.commitDirectoryMutations(this.serverId, mutations);
this.agents = new AgentDirectoryReplica(
serverId,
callbacks.onAgentStoppedRunning,
persist,
callbacks.onAgentRemoved,
callbacks.onAgentAccepted,
);
this.agents = new AgentDirectoryReplica(serverId, callbacks.onAgentStoppedRunning, persist);
this.workspaces = new WorkspaceDirectoryReplica(serverId);
this.baseline = this.restoreBaseline().catch(() => undefined);
}
ready(): Promise<void> {
return this.baseline;
}
connectionChanged(connection: DirectoryConnection): boolean {
@@ -198,7 +185,6 @@ export class DirectorySync {
return false;
}
this.flushAbortedTransactions();
this.timelineRequests.clear();
this.unsubscribe?.();
this.unsubscribe = null;
workspaceLabels.disconnect(this.serverId);
@@ -210,39 +196,63 @@ export class DirectorySync {
const subscriptions = [
client.on("agent_update", (message) => {
if (message.type !== "agent_update" || !this.isCurrent(client, source)) return;
if (!this.agentTransactions.record(source, message.payload)) {
this.revision += 1;
const recorded = this.agentTransactions.record(source, message.payload);
if (!recorded) {
this.agents.applyDelta(message.payload);
this.noteLiveCursor("agents", message.payload);
this.persistCheckpoint();
}
}),
client.on("workspace_update", (message) => {
if (message.type !== "workspace_update" || !this.isCurrent(client, source)) return;
if (!this.workspaceTransactions.record(source, message.payload)) {
this.revision += 1;
this.workspaceRevision += 1;
const recorded = this.workspaceTransactions.record(source, message.payload);
if (!recorded) {
this.applyWorkspaceDelta(message.payload);
this.noteLiveCursor("workspaces", message.payload);
this.persistCheckpoint();
}
}),
client.on("project.update", (message) => {
if (message.type !== "project.update" || !this.isCurrent(client, source)) return;
if (!this.workspaceTransactions.record(source, message.payload)) {
this.revision += 1;
this.workspaceRevision += 1;
const recorded = this.workspaceTransactions.record(source, message.payload);
if (!recorded) {
this.applyWorkspaceDelta(message.payload);
this.noteLiveCursor("projects", message.payload);
this.persistCheckpoint();
}
}),
client.on("script_status_update", (message) => {
if (message.type !== "script_status_update" || !this.isCurrent(client, source)) return;
this.revision += 1;
this.workspaceRevision += 1;
const delta: WorkspaceDirectoryDelta = {
kind: "script_status",
update: message.payload,
};
if (!this.workspaceTransactions.record(source, delta)) {
const recorded = this.workspaceTransactions.record(source, delta);
if (!recorded) {
this.applyWorkspaceDelta(delta);
this.persistCheckpoint();
}
}),
client.on("agent_deleted", (message) => {
if (message.type !== "agent_deleted" || !this.isCurrent(client, source)) return;
this.agents.remove(message.payload.agentId);
if (message.type === "agent_deleted" && this.isCurrent(client, source)) {
this.revision += 1;
this.agents.remove(message.payload.agentId);
this.persistCheckpoint();
}
}),
client.on("agent_archived", (message) => {
if (message.type !== "agent_archived" || !this.isCurrent(client, source)) return;
this.agents.archive(message.payload.agentId, message.payload.archivedAt);
if (message.type === "agent_archived" && this.isCurrent(client, source)) {
this.revision += 1;
this.agents.archive(message.payload.agentId, message.payload.archivedAt);
this.persistCheckpoint();
}
}),
];
this.unsubscribe = () => {
@@ -256,8 +266,9 @@ export class DirectorySync {
const wasDemanded = this.fullDemandSources.size > 0;
if (demanded) this.fullDemandSources.add(source);
else this.fullDemandSources.delete(source);
if (!wasDemanded && this.fullDemandSources.size > 0 && this.getOnlineConnection()) {
void this.requestDemandRefresh().catch(() => undefined);
if (!wasDemanded && this.fullDemandSources.size > 0) {
void this.loadCachedDirectory().catch(() => undefined);
if (this.getOnlineConnection()) void this.requestDemandRefresh().catch(() => undefined);
}
}
@@ -277,8 +288,6 @@ export class DirectorySync {
}
dispose(): void {
this.connection = { ...this.connection, status: "offline" };
this.timelineRequests.clear();
this.flushAbortedTransactions();
this.abortPendingSessionWaits();
this.unsubscribe?.();
@@ -295,7 +304,9 @@ export class DirectorySync {
private requestDemandRefresh(force = false): Promise<void> {
if (this.demandRefresh) return this.demandRefresh;
if (!this.hasDemand()) return Promise.resolve();
if (!this.getOnlineConnection()) return this.baseline;
if (!this.getOnlineConnection()) {
return this.fullDemandSources.size > 0 ? this.loadCachedDirectory() : Promise.resolve();
}
const source = this.connection.source;
if (
!force &&
@@ -308,8 +319,8 @@ export class DirectorySync {
this.fullDemandSources.size > 0
? this.refreshAll()
: Promise.all([
this.refreshAgents({ subscribe: {} }),
this.refreshWorkspaces({ subscribe: true }),
this.refreshAgentsInternal({ subscribe: {} }, false),
this.refreshWorkspacesInternal({ subscribe: true }, false),
]).then(() => undefined);
this.demandRefresh = refresh
.then(() => {
@@ -334,13 +345,57 @@ export class DirectorySync {
return this.requestDemandRefresh(true);
}
private async restoreBaseline(): Promise<void> {
async loadCachedAgent(agentId: string): Promise<void> {
if (!this.checkpoints) return;
const cached = await this.checkpoints.readDirectory(this.serverId);
if (!useSessionStore.getState().sessions[this.serverId]) return;
this.agents.commitCached(cached.agents);
this.workspaces.commitCached(cached);
this.cursors = cached.checkpoint ?? {};
if (useSessionStore.getState().sessions[this.serverId]?.agents.has(agentId)) return;
const token = this.agents.captureCache(agentId);
const agent = await this.checkpoints.readAgent(this.serverId, agentId);
if (!agent) return;
const session = useSessionStore.getState().sessions[this.serverId];
if (!session || session.agents.has(agentId)) return;
this.agents.commitCachedAgent(token, agent);
}
async prepareAgentRoute(agentId: string): Promise<void> {
await this.loadCachedAgent(agentId);
const agent = useSessionStore.getState().sessions[this.serverId]?.agents.get(agentId);
if (agent?.workspaceId) await this.loadCachedWorkspace(agent.workspaceId);
}
async prepareWorkspaceRoute(workspaceId: string): Promise<void> {
await this.loadCachedWorkspace(workspaceId);
}
private async loadCachedWorkspace(workspaceId: string): Promise<void> {
if (!this.checkpoints) return;
if (useSessionStore.getState().sessions[this.serverId]?.workspaces.has(workspaceId)) return;
const revision = this.workspaceRevision;
const cached = await this.checkpoints.readWorkspace(this.serverId, workspaceId);
if (!cached) return;
if (this.workspaceRevision !== revision) return;
const session = useSessionStore.getState().sessions[this.serverId];
if (!session) return;
this.workspaces.commitCachedWorkspace(cached.workspace, cached.project);
}
private loadCachedDirectory(): Promise<void> {
const checkpoints = this.checkpoints;
if (!checkpoints) return Promise.resolve();
this.cacheLoad ??= (async () => {
const revision = this.revision;
const cached = await checkpoints.readDirectory(this.serverId);
if (this.cacheAccepted) return;
if (!useSessionStore.getState().sessions[this.serverId]) return;
this.agents.commitCached(cached.agents);
this.workspaces.commitCached(cached);
if (this.revision === revision) this.cursors = cached.checkpoint ?? {};
this.cacheAccepted = true;
})();
return this.cacheLoad;
}
restoreCachedDirectory(): Promise<void> {
return this.loadCachedDirectory();
}
applyAgentTurnLiveness(
@@ -357,36 +412,33 @@ export class DirectorySync {
async fetchTimeline(
agentId: string,
request: Parameters<DaemonClient["fetchAgentTimeline"]>[1],
replaceTail = false,
): Promise<TimelinePage> {
const { client, source } = this.requireOnline();
): Promise<Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>> {
const { client } = this.requireOnline();
const token = this.agents.captureTimeline(agentId);
const key = JSON.stringify({ agentId, lifetime: token.version, request, replaceTail });
const existing = this.timelineRequests.get(key);
if (existing) return existing;
const apply = this.callbacks.onTimelineRequest(agentId, replaceTail);
const fetching = client.fetchAgentTimeline(agentId, request).then((page) => {
if (!this.isCurrent(client, source) || !this.agents.isTimelineCurrent(token)) {
throw new DirectoryRefreshSupersededError("timeline page no longer current");
}
if (page.agent) this.agents.submitTimelineAgent(token, page.agent);
apply(page);
return page;
});
this.timelineRequests.set(key, fetching);
const clear = () => {
if (this.timelineRequests.get(key) === fetching) this.timelineRequests.delete(key);
};
void fetching.then(clear, clear);
return fetching;
const page = await fetchAgentTimelineOnce(client, agentId, request);
if (page.agent && this.agents.submitTimelineAgent(token, page.agent)) {
this.revision += 1;
this.persistCheckpoint();
}
return page;
}
async refreshAgents(
input: RefreshAgentDirectoryInput = {},
): Promise<RefreshAgentDirectoryResult> {
return this.refreshAgentsInternal(input, true);
}
private async refreshAgentsInternal(
input: RefreshAgentDirectoryInput,
loadDirectoryCache: boolean,
): Promise<RefreshAgentDirectoryResult> {
if (loadDirectoryCache) {
this.primeAgentTransactionForCacheLoad();
await this.loadCachedDirectory();
}
const onlineConnection = this.getOnlineConnection();
if (!onlineConnection) {
await this.baseline;
this.callbacks.markAgentReady();
return {
agents: new Map(useSessionStore.getState().sessions[this.serverId]?.agents),
@@ -394,8 +446,6 @@ export class DirectorySync {
};
}
const { client, source } = onlineConnection;
// The transaction spans the whole refresh, so a live delta that arrives while the saved
// baseline or the session is still pending is replayed on top of whatever snapshot lands.
const transaction = this.agentTransactions.begin(source, () => ({
entries: [],
subscriptionId: null,
@@ -404,7 +454,6 @@ export class DirectorySync {
}));
this.callbacks.markAgentLoading();
try {
await this.baseline;
await this.waitForSession(client, source);
const session = useSessionStore.getState().sessions[this.serverId];
if (!input.filter && shouldUseLegacyDaemonWorkspaceDirectory(session?.serverInfo)) {
@@ -461,8 +510,9 @@ export class DirectorySync {
: delta,
)
: completion.deltas;
this.revision += 1;
const agents = this.commitAgentSnapshot(completion.snapshot, deltas);
this.persistAgentCursors(completion.snapshot);
this.persistAgentCursors(completion.snapshot, completion.deltas);
this.persistCheckpoint();
this.callbacks.markAgentReady();
return { agents, subscriptionId: completion.snapshot.subscriptionId };
@@ -477,27 +527,26 @@ export class DirectorySync {
}
async refreshWorkspaces(input?: { subscribe?: boolean }): Promise<void> {
return this.refreshWorkspacesInternal(input, true);
}
private async refreshWorkspacesInternal(
input: { subscribe?: boolean } | undefined,
loadDirectoryCache: boolean,
): Promise<void> {
if (loadDirectoryCache) await this.loadCachedDirectory();
const onlineConnection = this.getOnlineConnection();
if (!onlineConnection) {
await this.baseline;
return;
}
if (!onlineConnection) return;
const { client, source } = onlineConnection;
// The transaction spans the whole refresh. Its base for a changes-mode page is the store once
// the saved baseline is in it, so it is captured after that wait rather than at begin.
const transaction = this.workspaceTransactions.begin(source, () => ({
workspaces: new Map(),
projects: new Map(),
workspaces: new Map(useSessionStore.getState().sessions[this.serverId]?.workspaces),
projects: new Map(useSessionStore.getState().sessions[this.serverId]?.projects),
syncCursors: {},
syncModes: {},
touchedWorkspaceIds: new Set(),
touchedProjectIds: new Set(),
}));
try {
await this.baseline;
const session = useSessionStore.getState().sessions[this.serverId];
transaction.snapshot.workspaces = new Map(session?.workspaces);
transaction.snapshot.projects = new Map(session?.projects);
await this.waitForSessionMetadata(client, source);
const serverInfo = useSessionStore.getState().sessions[this.serverId]?.serverInfo;
if (serverInfo?.features?.workspaceMultiplicity !== true) {
@@ -676,6 +725,17 @@ export class DirectorySync {
return { client: this.connection.client, source: this.connection.source };
}
private primeAgentTransactionForCacheLoad(): void {
const connection = this.getOnlineConnection();
if (!connection) return;
this.agentTransactions.begin(connection.source, () => ({
entries: [],
subscriptionId: null,
legacy: false,
syncRemovals: [],
}));
}
private supportsDirectorySync(): boolean {
return (
useSessionStore.getState().sessions[this.serverId]?.serverInfo?.features?.directorySync ===
@@ -692,8 +752,12 @@ export class DirectorySync {
: this.agents.commitSnapshot(snapshot.entries, deltas);
}
private persistAgentCursors(snapshot: AgentSnapshot): void {
private persistAgentCursors(
snapshot: AgentSnapshot,
deltas: readonly AgentDirectoryDelta[],
): void {
if (snapshot.syncCursor) this.writeCursor("agents", snapshot.syncCursor);
for (const delta of deltas) this.noteLiveCursor("agents", delta);
}
private async fetchProjectSnapshot(
@@ -737,6 +801,8 @@ export class DirectorySync {
if (completion.kind === "stale") {
throw new DirectoryRefreshSupersededError("workspace completion was superseded");
}
this.revision += 1;
this.workspaceRevision += 1;
const previous = this.readWorkspaceState();
const deltaMutations = this.workspaces.commitSnapshot(completion.snapshot, completion.deltas);
const next = this.workspaces.snapshot();
@@ -756,11 +822,14 @@ export class DirectorySync {
for (const [entity, cursor] of Object.entries(completion.snapshot.syncCursors ?? {})) {
if (cursor) this.writeCursor(entity as "projects" | "workspaces", cursor);
}
for (const delta of completion.deltas) {
if (delta.kind === "script_status") continue;
const entity = "projectId" in delta || "project" in delta ? "projects" : "workspaces";
this.noteLiveCursor(entity, delta);
}
this.persistCheckpoint();
}
// Only a completed catch-up may advance the saved cursor. A live sequence can follow an offline
// gap, so saving it would claim intermediate changes the client never received.
private persistCheckpoint(): void {
this.checkpoints?.commitDirectoryMutations(this.serverId, [], this.cursors);
}
@@ -805,6 +874,17 @@ export class DirectorySync {
snapshot.syncCursor = { generation: sync.generation, afterSeq: sync.headSeq };
}
private noteLiveCursor(
entity: keyof DirectoryCheckpoint,
payload: { generation?: string; seq?: number },
): void {
if (!payload.generation || payload.seq === undefined) return;
this.writeCursor(entity, {
generation: payload.generation,
afterSeq: payload.seq,
});
}
private readCursors(): DirectoryCheckpoint {
return this.cursors;
}
packages/app/src/runtime/directory-sync/internal/agent-store.ts
+8 -12
View File
@@ -2,11 +2,7 @@ import equal from "fast-deep-equal";
import type { FetchAgentsEntry } from "@getpaseo/client/internal/daemon-client";
import { useSessionStore, type Agent } from "@/stores/session-store";
import { acceptAgentDirectoryUpdate } from "@/utils/agent-directory-update-policy";
import {
buildAgentDirectoryState,
type AgentDirectoryDelta,
type AgentRemovalReason,
} from "@/utils/agent-directory-sync";
import { buildAgentDirectoryState, type AgentDirectoryDelta } from "@/utils/agent-directory-sync";
import { derivePendingPermissionKey, normalizeAgentSnapshot } from "@/utils/agent-snapshots";
import { resolveProjectPlacement } from "@/utils/project-placement";
import { clearArchiveAgentPending } from "@/hooks/use-archive-agent";
@@ -27,10 +23,7 @@ function mergeSnapshotTurn(previous: Agent | undefined, incoming: Agent): Agent
}
export class AgentStoreProjection {
constructor(
private readonly serverId: string,
private readonly removeTimeline: (agentId: string, reason: AgentRemovalReason) => void,
) {}
constructor(private readonly serverId: string) {}
snapshot(): Map<string, Agent> {
return useSessionStore.getState().sessions[this.serverId]?.agents ?? new Map();
@@ -72,7 +65,7 @@ export class AgentStoreProjection {
agent?: Agent;
} {
if (delta.kind === "remove") {
this.remove(delta.agentId, "scope");
this.remove(delta.agentId);
return { agentId: delta.agentId, stoppedRunning: false };
}
const normalized = normalizeAgentSnapshot(delta.agent, this.serverId);
@@ -148,7 +141,7 @@ export class AgentStoreProjection {
useSessionStore.getState().setPendingPermissions(this.serverId, pending);
}
remove(agentId: string, reason: AgentRemovalReason): void {
remove(agentId: string): void {
const store = useSessionStore.getState();
const removeKey = <T>(current: Map<string, T>): Map<string, T> => {
if (!current.has(agentId)) return current;
@@ -160,13 +153,16 @@ export class AgentStoreProjection {
store.setAgents(this.serverId, removeKey);
store.setAgentDetails(this.serverId, removeKey);
store.setQueuedMessages(this.serverId, removeKey);
this.removeTimeline(agentId, reason);
store.setAgentTimelineCursor(this.serverId, removeKey);
store.setInitializingAgents(this.serverId, removeKey);
store.setPendingPermissions(this.serverId, (current) => {
const next = new Map(current);
for (const [key, pending] of next) if (pending.agentId === agentId) next.delete(key);
return next.size === current.size ? current : next;
});
store.setAgentAuthoritativeHistoryApplied(this.serverId, agentId, false);
store.setAgentStreamTail(this.serverId, removeKey);
store.clearAgentStreamHead(this.serverId, agentId);
useSessionStore.setState((state) => {
if (!state.agentLastActivity.has(agentId)) return state;
const agentLastActivity = new Map(state.agentLastActivity);
packages/app/src/runtime/directory-sync/workspace-replica.test.ts
+2 -5
View File
@@ -213,7 +213,7 @@ it("does not invent a null-key project from a workspace update", () => {
store.clearSession(serverId);
});
it("does not restore a cached workspace while its archive is pending", () => {
it("does not restore a targeted cached workspace while its archive is pending", () => {
const serverId = "cached-workspace-during-archive";
const workspaceId = "archived-workspace";
const store = useSessionStore.getState();
@@ -222,10 +222,7 @@ it("does not restore a cached workspace while its archive is pending", () => {
markWorkspaceArchivePending({ serverId, workspaceId });
try {
replica.commitCached({
workspaces: new Map([[workspaceId, normalizeWorkspaceDescriptor(workspace(workspaceId))]]),
projects: new Map(),
});
replica.commitCachedWorkspace(normalizeWorkspaceDescriptor(workspace(workspaceId)), undefined);
expect(useSessionStore.getState().sessions[serverId]?.workspaces.has(workspaceId)).toBe(false);
} finally {
packages/app/src/runtime/directory-sync/workspace-replica.ts
+9
View File
@@ -58,6 +58,15 @@ export class WorkspaceDirectoryReplica {
useSessionStore.getState().setHasWorkspaceDirectorySnapshot(this.serverId, true);
}
commitCachedWorkspace(
workspace: WorkspaceDescriptor,
project: ProjectDescriptor | undefined,
): void {
if (shouldSuppressWorkspaceForLocalArchive({ serverId: this.serverId, workspace })) return;
if (project) this.setProject(project);
this.setWorkspace(workspace);
}
commitSnapshot(
snapshot: WorkspaceDirectorySnapshot,
deltas: readonly WorkspaceDirectoryDelta[],
packages/app/src/runtime/host-runtime.test.ts
+23 -326
View File
@@ -22,8 +22,6 @@ import {
type HostRuntimeStorage,
} from "./host-runtime";
import type { ReplicaRow, ReplicaRowStore } from "./replica-cache/row-store";
import { ReplicaCache } from "./replica-cache";
import { planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
class FakeDaemonClient {
private state: ConnectionState = { status: "idle" };
@@ -39,13 +37,6 @@ class FakeDaemonClient {
public fetchAgentsResponses: Array<
Awaited<ReturnType<DaemonClient["fetchAgents"]>> | ReturnType<DaemonClient["fetchAgents"]>
> = [];
public timelineCalls = 0;
public timelineResponse: ReturnType<DaemonClient["fetchAgentTimeline"]> | null = null;
fetchAgentTimeline(): ReturnType<DaemonClient["fetchAgentTimeline"]> {
this.timelineCalls++;
if (!this.timelineResponse) throw new Error("Missing timeline response");
return this.timelineResponse;
}
public sentAgentMessages: Array<Parameters<DaemonClient["sendAgentMessage"]>> = [];
public sendAgentMessageFailures: Error[] = [];
public sendAgentMessageResponses: Promise<void>[] = [];
@@ -475,13 +466,16 @@ function createMemoryReplicaRowStore(): ReplicaRowStore {
`${row.serverId}:${row.kind}:${row.id}`;
return {
open: async () => undefined,
read: async (serverIds, kinds, ids) =>
[...rows.values()].filter(
read: async (serverId, kinds, ids) => {
const acceptedKinds = new Set(kinds);
const acceptedIds = ids ? new Set(ids) : null;
return [...rows.values()].filter(
(row) =>
serverIds.includes(row.serverId) &&
kinds.includes(row.kind) &&
(!ids || ids.includes(row.id)),
),
row.serverId === serverId &&
acceptedKinds.has(row.kind) &&
(!acceptedIds || acceptedIds.has(row.id)),
);
},
readAll: async () => {
const hosts = new Map<string, ReplicaRow[]>();
for (const row of rows.values()) {
@@ -1605,74 +1599,15 @@ describe("HostRuntimeStore", () => {
expect(store.getHosts()).toEqual([]);
});
it("exposes timeline demand and paints saved history before a network client exists", async () => {
const host = makeHost();
const storage = createMemoryHostRuntimeStorage({
"@paseo:daemon-registry": JSON.stringify([host]),
"@paseo:e2e": "1",
});
const replicaRowStore = createMemoryReplicaRowStore();
const disk = new ReplicaCache(replicaRowStore, { clearLegacyCache: async () => undefined });
disk.setHosts([host.serverId]);
disk.commitTimeline(host.serverId, "offline-agent", {
agentId: "offline-agent",
hasOlder: false,
range: { epoch: "offline-epoch", startSeq: 1, endSeq: 1 },
items: [
{
kind: "assistant_message",
id: "saved-message",
messageId: "saved-message",
text: "Saved history is available offline",
timestamp: new Date("2026-09-06T10:00:00.000Z"),
timelineCursor: { epoch: "offline-epoch", seq: 1 },
source: { startSeq: 1 },
},
],
});
await disk.flush();
const store = new HostRuntimeStore({
storage,
replicaRowStore,
deps: {
createClient: () => {
throw new Error("Offline demand must not need a client");
},
connectToDaemon: async () => {
throw new Error("Offline demand must not need a connection");
},
getClientId: async () => "cid_test_runtime",
},
});
try {
const registryLoaded = onceHostListMatches(store, () => store.isHostRegistryLoaded());
store.boot();
await registryLoaded;
expect(useSessionStore.getState().sessions[host.serverId]?.client).toBeNull();
const demand = useSessionStore.getState().sessions[host.serverId]?.viewedTimelineSync;
expect(demand).not.toBeNull();
if (!demand) throw new Error("Registered hosts must expose offline timeline demand");
demand.registerVisibleAgentIds("workspace", ["offline-agent"]);
await vi.waitFor(() => {
expect(
useSessionStore.getState().sessions[host.serverId]?.agentStreamTail.get("offline-agent"),
).toMatchObject([{ text: "Saved history is available offline" }]);
});
} finally {
store.syncHosts([]);
useSessionStore.getState().clearSession(host.serverId);
}
});
it("loads the host registry and restores its cached directory", async () => {
const host = makeHost();
const storage = createMemoryHostRuntimeStorage();
let directoryReads = 0;
let fullScans = 0;
const backingStore = createMemoryReplicaRowStore();
const replicaRowStore: ReplicaRowStore = {
...backingStore,
read: async (...args) => {
directoryReads += 1;
fullScans += 1;
return backingStore.read(...args);
},
};
@@ -1697,7 +1632,7 @@ describe("HostRuntimeStore", () => {
store.boot();
await registryLoaded;
await vi.waitFor(() => expect(directoryReads).toBe(1));
await vi.waitFor(() => expect(fullScans).toBe(1));
expect(useSessionStore.getState().sessions[host.serverId]).toMatchObject({
client: null,
hasHydratedAgents: false,
@@ -2394,15 +2329,10 @@ describe("HostRuntimeStore", () => {
host.serverId,
new Map([[recoveredAgent.agent.id, replicaAgent(recoveredAgent.agent, host.serverId)]]),
);
sessionStore.applyAgentTimelineResponseState(host.serverId, recoveredAgent.agent.id, {
items: [],
head: [],
range: { epoch: "epoch", startSeq: 10, endSeq: 20 },
older: "none",
newer: false,
synchronized: false,
submissions: [],
});
sessionStore.setAgentTimelineCursor(
host.serverId,
new Map([[recoveredAgent.agent.id, { epoch: "epoch", startSeq: 10, endSeq: 20 }]]),
);
store.syncHosts([host]);
await waitForHostOnline(store, host.serverId);
const load = store.refreshAgentDirectory({
@@ -2825,9 +2755,8 @@ describe("HostRuntimeStore", () => {
getClientId: async () => "cid_drain_submission",
},
});
store.syncHosts([{ ...host, connections: [] }]);
const sessionStore = useSessionStore.getState();
sessionStore.updateSessionClient(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.updateSessionServerInfo(host.serverId, {
serverId: host.serverId,
hostname: null,
@@ -2873,7 +2802,7 @@ describe("HostRuntimeStore", () => {
expect(session?.messageSubmissions.get("agent")).toBeDefined();
send.resolve();
store.syncHosts([]);
useSessionStore.getState().clearSession(host.serverId);
});
it("restores an automatically drained message when sending fails", async () => {
@@ -2891,9 +2820,8 @@ describe("HostRuntimeStore", () => {
getClientId: async () => "cid_failed_queue_drain",
},
});
store.syncHosts([{ ...host, connections: [] }]);
const sessionStore = useSessionStore.getState();
sessionStore.updateSessionClient(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.setQueuedMessages(
host.serverId,
new Map([
@@ -2919,7 +2847,7 @@ describe("HostRuntimeStore", () => {
]);
});
store.syncHosts([]);
useSessionStore.getState().clearSession(host.serverId);
});
it("serializes queued-message drains for the same agent", async () => {
@@ -2938,9 +2866,8 @@ describe("HostRuntimeStore", () => {
getClientId: async () => "cid_serialized_queue_drain",
},
});
store.syncHosts([{ ...host, connections: [] }]);
const sessionStore = useSessionStore.getState();
sessionStore.updateSessionClient(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.setQueuedMessages(
host.serverId,
new Map([["agent", [{ id: "first", text: "send once", attachments: [] }]]]),
@@ -2957,7 +2884,7 @@ describe("HostRuntimeStore", () => {
useSessionStore.getState().sessions[host.serverId]?.queuedMessages.get("agent"),
).toEqual([]);
});
store.syncHosts([]);
useSessionStore.getState().clearSession(host.serverId);
});
it("uses legacy GitHub attachments when draining a queue for an old daemon", async () => {
@@ -2974,9 +2901,8 @@ describe("HostRuntimeStore", () => {
getClientId: async () => "cid_legacy_queue_attachment",
},
});
store.syncHosts([{ ...host, connections: [] }]);
const sessionStore = useSessionStore.getState();
sessionStore.updateSessionClient(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
sessionStore.updateSessionServerInfo(host.serverId, {
serverId: host.serverId,
hostname: null,
@@ -3215,235 +3141,6 @@ describe("HostRuntimeStore", () => {
useSessionStore.getState().clearSession(host.serverId);
});
it.each(["coalesced", "late-tail", "rewind", "daemon-reset"] as const)(
"owns timeline request application (%s)",
async (scenario) => {
const host = makeHost({
serverId: `request-${scenario}`,
connections: [makeHost().connections[0]!],
});
const client = new FakeDaemonClient();
client.setConnectionState({ status: "connected" });
const store = new HostRuntimeStore({
replicaRowStore: createMemoryReplicaRowStore(),
deps: {
createClient: () => client as unknown as DaemonClient,
connectToDaemon: async ({ host: current }) => ({
client: client as unknown as DaemonClient,
serverId: current.serverId,
hostname: null,
}),
getClientId: async () => "test",
},
});
store.syncHosts([host]);
await waitForHostOnline(store, host.serverId);
const page: Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>> = {
requestId: "page",
agentId: "agent",
agent: null,
direction: "tail",
projection: "projected",
reset: false,
staleCursor: false,
gap: false,
epoch: "epoch",
window: { minSeq: 1, maxSeq: 1, nextSeq: 2 },
startCursor: { epoch: "epoch", seq: 1 },
endCursor: { epoch: "epoch", seq: 1 },
hasOlder: false,
hasNewer: false,
error: null,
entries: [
{
provider: "codex",
item: { type: "assistant_message", text: "first", messageId: "first" },
timestamp: "2026-09-06T00:00:00Z",
seqStart: 1,
seqEnd: 1,
sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }],
collapsed: [],
},
],
};
const viewed = store.getViewedTimelineOwner(host.serverId);
if (scenario !== "coalesced") viewed.applyTimelineResponse(page);
const response = createDeferred<Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>>();
client.timelineResponse = response.promise;
let publications = 0;
const unsubscribe = useSessionStore.subscribe((state, previous) => {
if (
state.sessions[host.serverId]?.agentStreamTail !==
previous.sessions[host.serverId]?.agentStreamTail
)
publications++;
});
const appendLive = () => {
viewed.enqueueStreamEvent("agent", {
epoch: "epoch",
seq: 2,
timestamp: new Date("2026-09-06T00:00:01Z"),
event: {
type: "timeline",
provider: "codex",
item: { type: "assistant_message", text: "new live", messageId: "new-live" },
},
});
viewed.flushStreamAgent("agent");
};
if (scenario === "rewind") appendLive();
const request = planTimelineTailFetch();
const first = store.fetchAgentTimeline(
host.serverId,
"agent",
request,
scenario === "late-tail",
);
const second =
scenario === "coalesced"
? store.fetchAgentTimeline(host.serverId, "agent", request)
: first;
if (scenario === "late-tail" || scenario === "daemon-reset") appendLive();
if (scenario === "daemon-reset") page.reset = true;
response.resolve(page);
await expect(Promise.all([first, second])).resolves.toEqual([page, page]);
expect(client.timelineCalls).toBe(1);
if (scenario === "coalesced") expect(publications).toBe(1);
else {
const session = useSessionStore.getState().sessions[host.serverId];
expect(
[
...(session?.agentStreamTail.get("agent") ?? []),
...(session?.agentStreamHead.get("agent") ?? []),
].map((item) => item.id),
).toEqual(scenario === "late-tail" ? ["first", "new-live"] : ["first"]);
expect(store.getTimelineReplica(host.serverId).readCursor("agent")?.endSeq).toBe(
scenario === "late-tail" ? 2 : 1,
);
}
unsubscribe();
store.syncHosts([]);
},
);
it.each(["settled", "captured", "renamed"] as const)(
"retains scoped-out history after host identity reconciliation (%s) and deletes it only explicitly",
async (ordering) => {
const host = makeHost({
serverId: "srv_temporary",
connections: [makeHost().connections[0]!],
});
const resolvedId = "srv_resolved";
const rows = createMemoryReplicaRowStore();
const seed = new ReplicaCache(rows, { clearLegacyCache: async () => undefined });
seed.setHosts([host.serverId]);
const saved = {
agentId: "agent",
hasOlder: false,
range: { epoch: "epoch", startSeq: 1, endSeq: 1 },
items: [
{
kind: "assistant_message" as const,
id: "saved",
text: "saved history",
timestamp: new Date("2026-09-06T00:00:00Z"),
timelineCursor: { epoch: "epoch", seq: 1 },
source: { startSeq: 1 },
},
],
};
seed.commitTimeline(host.serverId, "agent", saved);
await seed.flush();
const client = new FakeDaemonClient();
client.setConnectionState({ status: "connected" });
const store = new HostRuntimeStore({
replicaRowStore: rows,
deps: {
createClient: () => client as unknown as DaemonClient,
connectToDaemon: async ({ host: current }) => ({
client: client as unknown as DaemonClient,
serverId: current.serverId,
hostname: null,
}),
getClientId: async () => "test",
},
});
const storedRows = async () =>
(await rows.readAll()).find((entry) => entry.serverId === resolvedId)?.rows ?? [];
const hasStoredKind = async (kind: ReplicaRow["kind"]) =>
(await storedRows()).some((row) => row.kind === kind);
try {
store.syncHosts([host]);
await waitForHostOnline(store, host.serverId);
const replica = store.getTimelineReplica(host.serverId);
const started = Promise.withResolvers<void>();
const release = Promise.withResolvers<void>();
const readRows = rows.read.bind(rows);
let reads = 0;
rows.read = async (...args) => {
if (!args[1].includes("timeline")) return readRows(...args);
reads++;
if (ordering === "settled" || reads !== 1) return readRows(...args);
const captured = ordering === "captured" ? await readRows(...args) : undefined;
started.resolve();
await release.promise;
return captured ?? readRows(...args);
};
const preparation = replica.prepare("agent");
if (ordering === "settled") {
await preparation;
expect(
useSessionStore.getState().sessions[host.serverId]?.agentStreamTail.get("agent"),
).toEqual(saved.items);
} else await started.promise;
store.reconcileServerId(host.serverId, resolvedId);
if (ordering !== "settled") {
await vi.waitFor(async () => expect(await hasStoredKind("timeline")).toBe(true));
release.resolve();
await preparation;
}
expect(replica.prepare("agent")).toBe(preparation);
expect(reads).toBe(ordering === "settled" ? 1 : 2);
expect(store.getTimelineReplica(resolvedId)).toBe(replica);
expect(
useSessionStore.getState().sessions[resolvedId]?.agentStreamTail.get("agent"),
).toEqual(saved.items);
expect(replica.readCursor("agent")).toEqual({ epoch: "epoch", endSeq: 1 });
await waitForHostOnline(store, resolvedId);
useSessionStore.getState().updateSessionServerInfo(resolvedId, {
serverId: resolvedId,
hostname: null,
version: "test",
features: { workspaceMultiplicity: true },
});
store.acceptAgentSnapshot(
resolvedId,
normalizeAgentSnapshot(
makeFetchAgentsEntry({ id: "agent", cwd: "/repo", updatedAt: "2026-09-06T00:00:00Z" })
.agent,
resolvedId,
),
);
await vi.waitFor(async () => expect(await hasStoredKind("agent")).toBe(true), {
timeout: 3000,
});
await store.refreshAgentDirectory({ serverId: resolvedId });
store.setAppVisible(false);
await vi.waitFor(async () => expect(await hasStoredKind("agent")).toBe(false));
const reopened = new ReplicaCache(rows, { clearLegacyCache: async () => undefined });
reopened.setHosts([resolvedId]);
expect(await reopened.readTimeline(resolvedId, "agent")).toEqual(saved);
store.restoreAgentSnapshot(resolvedId, "agent", undefined);
store.setAppVisible(false);
await vi.waitFor(async () => expect(await hasStoredKind("timeline")).toBe(false));
} finally {
store.syncHosts([]);
useSessionStore.getState().clearSession(host.serverId);
useSessionStore.getState().clearSession(resolvedId);
}
},
);
it("replaces stale active session state when active bootstrap omits an agent", async () => {
const host = makeHost({
serverId: "srv_archived_rehydrate",
packages/app/src/runtime/host-runtime.ts
+52 -50
View File
@@ -1,4 +1,3 @@
import { createTimelineReplica, type TimelineReplica } from "@/timeline/replica";
import { useSyncExternalStore, useMemo } from "react";
import AsyncStorage from "@react-native-async-storage/async-storage";
import equal from "fast-deep-equal/es6";
@@ -71,8 +70,11 @@ import { ReplicaCache } from "@/runtime/replica-cache";
import type { ReplicaRowStore } from "@/runtime/replica-cache/row-store";
import { createReplicaRowStore } from "@/runtime/replica-cache/row-store-factory";
import {
createTimelineReplica,
createViewedTimelineOwner,
type TimelineReplica,
type ViewedTimelineOwner,
type ViewedTimelineOwnerPorts,
} from "@/timeline/viewed-timeline-sync";
import { projectIconCache } from "@/projects/icon-cache";
import { nativePerformanceTrace } from "@/performance/native-trace";
@@ -1402,7 +1404,6 @@ export class HostRuntimeStore {
private directorySyncByServer = new Map<string, DirectorySync>();
private nextCancellationRequestId = 0;
private timelineReplicaByServer = new Map<string, TimelineReplica>();
private viewedTimelineByServer = new Map<string, ViewedTimelineOwner>();
private configuredOverrideBootstrapInFlight: Promise<void> | null = null;
private bootPromise: Promise<void> | null = null;
private appVisible = true;
@@ -1520,6 +1521,12 @@ export class HostRuntimeStore {
projectIconCache.setHosts(profiles.map((profile) => profile.serverId));
await projectIconCache.restore();
this.syncHosts(profiles);
for (const profile of profiles) {
void this.directorySyncByServer
.get(profile.serverId)
?.restoreCachedDirectory()
.catch(() => undefined);
}
} catch (error) {
console.error("[HostRuntime] Failed to load host registry from storage", error);
} finally {
@@ -1656,17 +1663,11 @@ export class HostRuntimeStore {
projectIconCache.reconcileServerId(oldServerId, newServerId);
this.directorySyncByServer.get(oldServerId)?.dispose();
this.directorySyncByServer.delete(oldServerId);
const timelineReplica = this.getTimelineReplica(oldServerId);
rekeyMap(this.timelineReplicaByServer, oldServerId, newServerId);
this.timelineReplicaByServer.delete(oldServerId);
const directory = new DirectorySync(
newServerId,
{
onTimelineRequest: (agentId, replaceTail) =>
this.getViewedTimelineOwner(newServerId).beginTimelineRequest(agentId, replaceTail),
onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(newServerId, agentId),
onAgentRemoved: (agentId, reason) =>
this.getViewedTimelineOwner(newServerId).removeAgent(agentId, reason),
onAgentAccepted: (agentId) => this.getViewedTimelineOwner(newServerId).acceptAgent(agentId),
markAgentLoading: () => controller.markAgentDirectorySyncLoading(),
markAgentReady: () => controller.markAgentDirectorySyncReady(),
markAgentError: (error) => controller.markAgentDirectorySyncError(error),
@@ -1674,16 +1675,18 @@ export class HostRuntimeStore {
this.replicaCache,
);
this.directorySyncByServer.set(newServerId, directory);
this.timelineReplicaByServer.set(
newServerId,
createTimelineReplica({
serverId: newServerId,
storage: this.replicaCache,
prepareAgent: (agentId) => directory.prepareAgentRoute(agentId),
}),
);
controller.adoptReconciledServerId(newServerId);
const snapshot = controller.getSnapshot();
this.viewedTimelineByServer.get(oldServerId)?.dispose();
this.viewedTimelineByServer.delete(oldServerId);
this.clearHostReplica(oldServerId);
useSessionStore
.getState()
.initializeSession(newServerId, snapshot.client, snapshot.clientGeneration);
timelineReplica.reconcileServerId(newServerId);
this.installViewedTimelineOwner(newServerId);
this.syncSessionReplica(newServerId, snapshot);
directory.connectionChanged({
client: snapshot.client,
status: snapshot.connectionStatus === "online" ? "online" : "offline",
@@ -2054,9 +2057,6 @@ export class HostRuntimeStore {
this.connectionStatusStartedAtByServer.delete(serverId);
this.directorySyncByServer.get(serverId)?.dispose();
this.directorySyncByServer.delete(serverId);
this.viewedTimelineByServer.get(serverId)?.dispose();
this.viewedTimelineByServer.delete(serverId);
this.timelineReplicaByServer.get(serverId)?.dispose();
this.timelineReplicaByServer.delete(serverId);
this.clearHostReplica(serverId);
void controller.stop();
@@ -2086,13 +2086,7 @@ export class HostRuntimeStore {
const directory = new DirectorySync(
host.serverId,
{
onTimelineRequest: (agentId, replaceTail) =>
this.getViewedTimelineOwner(host.serverId).beginTimelineRequest(agentId, replaceTail),
onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(host.serverId, agentId),
onAgentRemoved: (agentId, reason) =>
this.getViewedTimelineOwner(host.serverId).removeAgent(agentId, reason),
onAgentAccepted: (agentId) =>
this.getViewedTimelineOwner(host.serverId).acceptAgent(agentId),
markAgentLoading: () => controller.markAgentDirectorySyncLoading(),
markAgentReady: () => controller.markAgentDirectorySyncReady(),
markAgentError: (error) => controller.markAgentDirectorySyncError(error),
@@ -2105,9 +2099,9 @@ export class HostRuntimeStore {
createTimelineReplica({
serverId: host.serverId,
storage: this.replicaCache,
prepareAgent: (agentId) => directory.prepareAgentRoute(agentId),
}),
);
this.installViewedTimelineOwner(host.serverId);
const initialSnapshot = controller.getSnapshot();
this.lastConnectionStatusByServer.set(host.serverId, initialSnapshot.connectionStatus);
this.connectionStatusStartedAtByServer.set(host.serverId, Date.now());
@@ -2219,7 +2213,7 @@ export class HostRuntimeStore {
supportsForgeAttachments,
}),
encodeImages,
submission: createMessageSubmissionWriter(serverId, this.getTimelineReplica(serverId)),
submission: createMessageSubmissionWriter(serverId),
});
},
})
@@ -2349,7 +2343,6 @@ export class HostRuntimeStore {
setAppVisible(visible: boolean): void {
this.appVisible = visible;
for (const owner of this.viewedTimelineByServer.values()) owner.setActive(visible);
for (const controller of this.controllers.values()) {
controller.setReconnectEnabled(visible);
}
@@ -2398,40 +2391,49 @@ export class HostRuntimeStore {
return () => directory.setDemand(source, false);
}
getTimelineReplica(serverId: string): TimelineReplica {
const replica = this.timelineReplicaByServer.get(serverId);
if (!replica) throw new Error(`Unknown host runtime for serverId ${serverId}`);
return replica;
async prepareAgentRoute(serverId: string, agentId: string): Promise<void> {
const directory = this.directorySyncByServer.get(serverId);
if (!directory) throw new Error(`Unknown host runtime for serverId ${serverId}`);
await directory.prepareAgentRoute(agentId);
}
async fetchAgentTimeline(
async prepareWorkspaceRoute(serverId: string, workspaceId: string): Promise<void> {
const directory = this.directorySyncByServer.get(serverId);
if (!directory) throw new Error(`Unknown host runtime for serverId ${serverId}`);
await directory.prepareWorkspaceRoute(workspaceId);
}
async prepareAgentTimeline(serverId: string, agentId: string): Promise<void> {
const replica = this.timelineReplicaByServer.get(serverId);
if (!replica) throw new Error(`Unknown host runtime for serverId ${serverId}`);
await replica.prepare(agentId);
}
fetchAgentTimeline(
serverId: string,
agentId: string,
request: Parameters<DaemonClient["fetchAgentTimeline"]>[1],
replaceTail = false,
): Promise<Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>> {
const directory = this.directorySyncByServer.get(serverId);
if (!directory) throw new Error(`Unknown host runtime for serverId ${serverId}`);
return directory.fetchTimeline(agentId, request, replaceTail);
return directory.fetchTimeline(agentId, request);
}
getViewedTimelineOwner(serverId: string): ViewedTimelineOwner {
const owner = this.viewedTimelineByServer.get(serverId);
if (!owner) throw new Error(`Unknown host runtime for serverId ${serverId}`);
return owner;
}
private installViewedTimelineOwner(serverId: string): void {
const owner = createViewedTimelineOwner({
createViewedTimelineOwner(
serverId: string,
ports: ViewedTimelineOwnerPorts,
): ViewedTimelineOwner {
const directory = this.directorySyncByServer.get(serverId);
if (!directory) throw new Error(`Unknown host runtime for serverId ${serverId}`);
const replica = this.timelineReplicaByServer.get(serverId);
if (!replica) throw new Error(`Unknown host runtime for serverId ${serverId}`);
return createViewedTimelineOwner({
serverId,
replica: this.getTimelineReplica(serverId),
replaceDemandedAgentIds: (ids) =>
this.directorySyncByServer.get(serverId)?.setAgentRouteDemand(ids),
drainQueuedAgentMessage: (id) => this.drainQueuedAgentMessage(serverId, id),
replica,
replaceDemandedAgentIds: (agentIds) => directory.setAgentRouteDemand(agentIds),
drainQueuedAgentMessage: (agentId) => this.drainQueuedAgentMessage(serverId, agentId),
ports,
});
this.viewedTimelineByServer.set(serverId, owner);
owner.setActive(this.appVisible);
useSessionStore.getState().setViewedTimelineSync(serverId, owner);
}
private emit(serverId: string): void {
packages/app/src/runtime/replica-cache/index.test.ts
+78 -472
View File
@@ -25,58 +25,41 @@ class MemoryStorage implements ReplicaRowStore {
readonly rows = new Map<string, ReplicaRow>();
readonly changes: ReplicaRowChanges[] = [];
readonly reads: Array<{
serverIds: readonly string[];
serverId: string;
kinds: readonly ReplicaRow["kind"][];
ids?: readonly string[];
}> = [];
writes = 0;
readAlls = 0;
cleanups = 0;
nextWriteFailure: Error | null = null;
nextOpenFailure: Error | null = null;
nextReadAllFailure: Error | null = null;
nextDeleteFailure: Error | null = null;
readGate: Promise<void> | null = null;
readAllGate: Promise<void> | null = null;
writeGate: Promise<void> | null = null;
renameGate: Promise<void> | null = null;
onRead: (() => void) | null = null;
private key(row: Pick<ReplicaRow, "serverId" | "kind" | "id">): string {
return `${row.serverId}:${row.kind}:${row.id}`;
}
async open(): Promise<void> {
if (this.nextOpenFailure) {
const error = this.nextOpenFailure;
this.nextOpenFailure = null;
throw error;
}
}
async open(): Promise<void> {}
async read(
serverIds: readonly string[],
serverId: string,
kinds: readonly ReplicaRow["kind"][],
ids?: readonly string[],
): Promise<ReplicaRow[]> {
this.reads.push({ serverIds, kinds, ids });
const rows = [...this.rows.values()];
this.reads.push({ serverId, kinds, ...(ids ? { ids } : {}) });
this.onRead?.();
await this.readGate;
return rows.filter(
const acceptedKinds = new Set(kinds);
const acceptedIds = ids ? new Set(ids) : null;
return [...this.rows.values()].filter(
(row) =>
serverIds.includes(row.serverId) &&
kinds.includes(row.kind) &&
(!ids || ids.includes(row.id)),
row.serverId === serverId &&
acceptedKinds.has(row.kind) &&
(!acceptedIds || acceptedIds.has(row.id)),
);
}
async readAll(): Promise<ReplicaHostRows[]> {
this.readAlls += 1;
await this.readAllGate;
if (this.nextReadAllFailure) {
const error = this.nextReadAllFailure;
this.nextReadAllFailure = null;
throw error;
}
const hosts = new Map<string, ReplicaRow[]>();
for (const row of this.rows.values()) {
const rows = hosts.get(row.serverId) ?? [];
@@ -88,7 +71,6 @@ class MemoryStorage implements ReplicaRowStore {
async apply(changes: ReplicaRowChanges): Promise<void> {
this.writes += 1;
await this.writeGate;
if (this.nextWriteFailure) {
const error = this.nextWriteFailure;
this.nextWriteFailure = null;
@@ -100,16 +82,10 @@ class MemoryStorage implements ReplicaRowStore {
}
async deleteHost(serverId: string): Promise<void> {
if (this.nextDeleteFailure) {
const error = this.nextDeleteFailure;
this.nextDeleteFailure = null;
throw error;
}
for (const [key, row] of this.rows) if (row.serverId === serverId) this.rows.delete(key);
}
async renameHost(oldServerId: string, newServerId: string): Promise<void> {
await this.renameGate;
for (const [key, row] of this.rows) {
if (row.serverId !== oldServerId) continue;
this.rows.delete(key);
@@ -194,7 +170,6 @@ function timelineItem(text = "Cached"): StreamItem {
text,
timestamp: new Date("2026-07-18T08:02:00.000Z"),
timelineCursor: { epoch: "epoch-1", seq: 12 },
source: { startSeq: 12, chunks: [{ seq: 12, offset: 0 }] },
};
}
@@ -266,359 +241,6 @@ function deleteDirectory(cache: ReplicaCache, serverId: string): void {
}
describe("ReplicaCache", () => {
it("paints requested rows while the unrelated global index is unavailable", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
commitDirectory(writer, SERVER_ID, directory());
writer.commitTimeline(SERVER_ID, "agent-1", timeline("Requested"));
await writer.flush();
const release = deferred();
storage.readAllGate = release.promise;
storage.reads.length = 0;
const reader = createCache(storage);
reader.setHosts([SERVER_ID, "unrelated"]);
reader.commitTimeline("unrelated", "other", { ...timeline("Pending"), agentId: "other" });
const writing = reader.flush();
let painted = false;
const reading = Promise.all([
reader.readTimeline(SERVER_ID, "agent-1"),
reader.readDirectory(SERVER_ID),
]).then(([saved, restoredDirectory]) => {
expect(saved?.items).toEqual([timelineItem("Requested")]);
expect(restoredDirectory.agents.has("agent-1")).toBe(true);
painted = true;
return undefined;
});
try {
await expect.poll(() => painted, { timeout: 200 }).toBe(true);
expect(storage.reads).toEqual([
{ serverIds: [SERVER_ID], kinds: ["timeline"], ids: ["agent-1"] },
{
serverIds: [SERVER_ID],
kinds: ["agent", "workspace", "project", "checkpoint"],
ids: undefined,
},
]);
} finally {
release.resolve();
await Promise.all([writing, reading]);
}
});
it("overlays a commit that settles after a scoped read captures old disk rows", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
writer.commitTimeline(SERVER_ID, "agent-1", timeline("Old"));
await writer.flush();
const reader = createCache(storage);
const release = deferred();
storage.readGate = release.promise;
const reading = reader.readTimeline(SERVER_ID, "agent-1");
await expect.poll(() => storage.reads.length).toBe(1);
reader.commitTimeline(SERVER_ID, "agent-1", timeline("New"));
await reader.flush();
release.resolve();
expect((await reading)?.items).toEqual([timelineItem("New")]);
});
it("invalidates an old scoped read across removal and immediate re-registration", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
writer.commitTimeline(SERVER_ID, "agent-1", timeline("Old lifetime"));
await writer.flush();
const reader = createCache(storage);
const release = deferred();
storage.readGate = release.promise;
const reading = reader.readTimeline(SERVER_ID, "agent-1");
await expect.poll(() => storage.reads.length).toBe(1);
reader.setHosts([]);
reader.setHosts([SERVER_ID]);
release.resolve();
expect(await reading).toBeUndefined();
expect(await reader.readTimeline(SERVER_ID, "agent-1")).toBeUndefined();
await reader.flush();
});
it.each([false, true])(
"uses the same certification for accepted and persisted rows (provenance=%s)",
async (provenance) => {
const storage = new MemoryStorage();
const cache = createCache(storage);
const items = ["block-1", "block-2"].map((id) => {
const item = timelineItem("Repeated markdown");
item.id = id;
if (item.kind === "assistant_message") item.messageId = "same-message";
item.source = provenance ? { startSeq: 12, chunks: [{ seq: 12, offset: 0 }] } : undefined;
return item;
});
const value = { ...timeline(), items };
cache.commitTimeline(SERVER_ID, "agent-1", value);
const accepted = await cache.readTimeline(SERVER_ID, "agent-1");
expect(accepted?.range).toEqual(provenance ? value.range : null);
expect(accepted?.items).toEqual(items);
await cache.flush();
const persisted = await createCache(storage).readTimeline(SERVER_ID, "agent-1");
expect(persisted).toEqual(accepted);
},
);
it("serializes only the coalesced certified page, including reads of accepted commits", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
let encodings = 0;
class MeasuredDate extends Date {
override toISOString(): string {
encodings += 1;
return super.toISOString();
}
}
for (let seq = 1; seq <= 10; seq++) {
const item = {
...timelineItem(`row ${seq}`),
timestamp: new MeasuredDate("2026-07-18T08:02:00.000Z"),
timelineCursor: { epoch: "epoch", seq },
source: { startSeq: seq, chunks: [{ seq, offset: 0 }] },
};
cache.commitTimeline(
SERVER_ID,
"agent-1",
{
agentId: "agent-1",
items: [item],
range: { epoch: "epoch", startSeq: seq, endSeq: seq },
hasOlder: true,
},
true,
);
expect((await cache.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([item]);
}
expect(encodings).toBe(0);
await cache.flush();
expect(encodings).toBe(1);
expect((await createCache(storage).readTimeline(SERVER_ID, "agent-1"))?.range?.endSeq).toBe(10);
});
it("does not resurrect a removed host when its in-flight write completes", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
const release = deferred();
storage.writeGate = release.promise;
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Removed history"));
const writing = cache.flush();
await expect.poll(() => storage.writes).toBe(1);
cache.setHosts([]);
release.resolve();
await writing;
cache.setHosts([SERVER_ID]);
await cache.flush();
expect(await cache.readTimeline(SERVER_ID, "agent-1")).toBeUndefined();
expect(storage.rows.size).toBe(0);
});
it("keeps rows written under a host renamed during their write", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
const release = deferred();
storage.writeGate = release.promise;
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Renamed while writing"));
const writing = cache.flush();
await expect.poll(() => storage.writes).toBe(1);
cache.reconcileServerId(SERVER_ID, "resolved-host");
release.resolve();
await writing;
await cache.flush();
expect((await cache.readTimeline("resolved-host", "agent-1"))?.items).toEqual([
timelineItem("Renamed while writing"),
]);
expect(await cache.readTimeline(SERVER_ID, "agent-1")).toBeUndefined();
expect([...storage.rows.keys()]).toEqual(["resolved-host:timeline:agent-1"]);
});
it("writes a row replaced during its own write again", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
const release = deferred();
storage.writeGate = release.promise;
cache.commitTimeline(SERVER_ID, "agent-1", timeline("First"));
const writing = cache.flush();
await expect.poll(() => storage.writes).toBe(1);
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Second"));
release.resolve();
await writing;
await cache.flush();
expect(storage.writes).toBe(2);
expect((await createCache(storage).readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([
timelineItem("Second"),
]);
});
it("restores a timeline while another host's disk write is blocked", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
cache.setHosts([SERVER_ID, "other-host"]);
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Saved before restart"));
await cache.flush();
const release = deferred();
storage.writeGate = release.promise;
commitDirectory(cache, "other-host", directory());
const writing = cache.flush();
await expect.poll(() => storage.writes).toBe(2);
let restored: Awaited<ReturnType<ReplicaCache["readTimeline"]>>;
const reading = cache.readTimeline(SERVER_ID, "agent-1").then((value) => {
restored = value;
return value;
});
try {
await expect
.poll(() => restored?.items, { timeout: 200 })
.toEqual([timelineItem("Saved before restart")]);
} finally {
release.resolve();
await Promise.all([writing, reading]);
}
});
it("reads rows under a resolved host identity before the disk rename finishes", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
writer.commitTimeline(SERVER_ID, "agent-1", timeline("Saved under temporary host ID"));
await writer.flush();
const reader = createCache(storage);
const release = deferred();
storage.renameGate = release.promise;
reader.reconcileServerId(SERVER_ID, "resolved-host");
expect((await reader.readTimeline("resolved-host", "agent-1"))?.items).toEqual([
timelineItem("Saved under temporary host ID"),
]);
release.resolve();
await reader.flush();
expect(storage.rows.has("resolved-host:timeline:agent-1")).toBe(true);
});
it("keeps rename collision precedence and accepted baselines during lazy reads", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
writer.setHosts([SERVER_ID, "resolved-host"]);
writer.commitTimeline(SERVER_ID, "agent-1", timeline("Moved"));
writer.commitTimeline("resolved-host", "agent-1", timeline("Superseded target"));
commitDirectory(writer, SERVER_ID, directory());
await writer.flush();
const reader = createCache(storage);
reader.setHosts([SERVER_ID, "resolved-host"]);
const release = deferred();
storage.renameGate = release.promise;
reader.replaceDirectoryBaseline(SERVER_ID, {
...directory(),
workspaces: new Map(),
projects: new Map(),
});
reader.reconcileServerId(SERVER_ID, "resolved-host");
try {
expect((await reader.readTimeline("resolved-host", "agent-1"))?.items).toEqual([
timelineItem("Moved"),
]);
const accepted = await reader.readDirectory("resolved-host");
expect(accepted.agents.get("agent-1")?.serverId).toBe("resolved-host");
expect(accepted.workspaces.size).toBe(0);
} finally {
release.resolve();
await reader.flush();
}
const reopened = new ReplicaCache(storage, noLegacyCleanup);
reopened.setHosts(["resolved-host"]);
expect((await reopened.readDirectory("resolved-host")).workspaces.size).toBe(0);
expect((await reopened.readTimeline("resolved-host", "agent-1"))?.items).toEqual([
timelineItem("Moved"),
]);
});
it("returns a committed timeline before it is written", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Accepted"));
expect((await cache.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([
timelineItem("Accepted"),
]);
expect(storage.writes).toBe(0);
});
it("round-trips a row's source start and chunk provenance", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
cache.commitTimeline(SERVER_ID, "agent-1", {
...timeline(),
items: [
{
...timelineItem("Streamed"),
source: {
startSeq: 10,
chunks: [
{ seq: 10, offset: 0 },
{ seq: 12, offset: 4 },
],
},
},
],
});
await cache.flush();
const restored = await createCache(storage).readTimeline(SERVER_ID, "agent-1");
expect(restored?.items[0]).toMatchObject({ source: { startSeq: 10 } });
expect(restored?.items[0]?.source?.chunks).toEqual([
{ seq: 10, offset: 0 },
{ seq: 12, offset: 4 },
]);
});
it("persists the owner's forty-entry restart page from a long synced history", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
const items = Array.from({ length: 40 }, (_, index) => ({
...timelineItem(`Message ${index}`),
id: `message-${index}`,
timelineCursor: { epoch: "epoch-1", seq: index + 201 },
source: { startSeq: index + 201, chunks: [{ seq: index + 201, offset: 0 }] },
}));
cache.commitTimeline(SERVER_ID, "agent-1", {
agentId: "agent-1",
items,
range: { epoch: "epoch-1", startSeq: 201, endSeq: 240 },
hasOlder: true,
});
await cache.flush();
const restored = await createCache(storage).readTimeline(SERVER_ID, "agent-1");
expect(restored?.items).toHaveLength(40);
expect(restored?.range).toEqual({ epoch: "epoch-1", startSeq: 201, endSeq: 240 });
expect(restored?.hasOlder).toBe(true);
});
it("encodes the owner's certified page without a per-timeline byte ceiling", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
const items = Array.from({ length: 40 }, (_, index) => ({
...timelineItem("x".repeat(20_000)),
id: `message-${index}`,
timelineCursor: { epoch: "epoch-1", seq: index + 1 },
source: { startSeq: index + 1, chunks: [{ seq: index + 1, offset: 0 }] },
}));
cache.commitTimeline(SERVER_ID, "agent-1", {
agentId: "agent-1",
items,
range: { epoch: "epoch-1", startSeq: 1, endSeq: 40 },
hasOlder: true,
});
await cache.flush();
const restored = await createCache(storage).readTimeline(SERVER_ID, "agent-1");
expect(restored?.items.map((item) => item.id)).toEqual(items.map((item) => item.id));
expect(restored?.range).toEqual({ epoch: "epoch-1", startSeq: 1, endSeq: 40 });
expect(restored?.hasOlder).toBe(true);
});
it("does nothing until an owner explicitly commits data", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
@@ -787,10 +409,9 @@ describe("ReplicaCache", () => {
deleteDirectory(cache, SERVER_ID);
const restored = await cache.readDirectory(SERVER_ID);
expect(restored.agents.size).toBe(0);
expect(restored.workspaces.size).toBe(0);
expect(restored.projects.size).toBe(0);
expect(await cache.readAgent(SERVER_ID, "agent-1")).toBeUndefined();
expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined();
expect((await cache.readDirectory(SERVER_ID)).projects.size).toBe(0);
});
it("fails closed when an accepted deletion cannot be persisted before a read", async () => {
@@ -802,23 +423,25 @@ describe("ReplicaCache", () => {
deleteDirectory(cache, SERVER_ID);
expect((await cache.readDirectory(SERVER_ID)).workspaces.size).toBe(0);
expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined();
});
it("applies a deletion accepted while a scoped directory read is still loading", async () => {
it("discards a durable read when the host changes while it is in flight", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
commitDirectory(writer, SERVER_ID, directory());
await writer.flush();
const release = deferred();
storage.readGate = release.promise;
const cache = createCache(storage);
commitDirectory(cache, SERVER_ID, directory());
await cache.flush();
const started = deferred();
const release = deferred();
storage.onRead = started.resolve;
storage.readGate = release.promise;
const reading = cache.readDirectory(SERVER_ID);
const reading = cache.readAgent(SERVER_ID, "agent-1");
await started.promise;
deleteDirectory(cache, SERVER_ID);
release.resolve();
expect((await reading).agents.size).toBe(0);
expect(await reading).toBeUndefined();
});
it("never reads a timeline older than an accepted deferred replacement", async () => {
@@ -890,6 +513,39 @@ describe("ReplicaCache", () => {
expect(await createCache(storage).readTimeline(SERVER_ID, "agent-1")).toBeUndefined();
});
it("reads one requested agent and the focused timeline without scanning directory rows", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
commitDirectory(writer, SERVER_ID, directory());
writer.commitTimeline(SERVER_ID, "agent-1", timeline());
await writer.flush();
const reader = createCache(storage);
expect((await reader.readAgent(SERVER_ID, "agent-1"))?.id).toBe("agent-1");
expect((await reader.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([timelineItem()]);
expect(storage.reads).toEqual([
{ serverId: SERVER_ID, kinds: ["agent"], ids: ["agent-1"] },
{ serverId: SERVER_ID, kinds: ["timeline"], ids: ["agent-1"] },
]);
});
it("reads one requested workspace and its project without scanning the directory", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
commitDirectory(writer, SERVER_ID, directory());
await writer.flush();
const reader = createCache(storage);
const restored = await reader.readWorkspace(SERVER_ID, "workspace-1");
expect(restored?.workspace.id).toBe("workspace-1");
expect(restored?.project?.projectId).toBe("project-1");
expect(storage.reads).toEqual([
{ serverId: SERVER_ID, kinds: ["workspace"], ids: ["workspace-1"] },
{ serverId: SERVER_ID, kinds: ["project"], ids: ["project-1"] },
]);
});
it("treats a corrupt row as a scoped miss", async () => {
const storage = new MemoryStorage();
storage.rows.set(`${SERVER_ID}:agent:agent-1`, {
@@ -914,7 +570,6 @@ describe("ReplicaCache", () => {
const cache = createCache(storage);
const restored = await cache.readDirectory(SERVER_ID);
await cache.flush();
expect(restored.agents.size).toBe(0);
expect(restored.projects.get("project-1")?.projectDisplayName).toBe("Paseo");
@@ -946,7 +601,7 @@ describe("ReplicaCache", () => {
commitDirectory(cache, otherServerId, directory());
await cache.flush();
expect((await cache.readDirectory(SERVER_ID)).agents.has("agent-1")).toBe(false);
expect(await cache.readAgent(SERVER_ID, "agent-1")).toBeUndefined();
cache.commitTimeline(otherServerId, "agent-1", timeline("x".repeat(1_000)));
await cache.flush();
@@ -972,9 +627,7 @@ describe("ReplicaCache", () => {
payload: "{bad",
});
const cache = createCache(storage);
expect((await cache.readDirectory(SERVER_ID)).agents.has("agent-1")).toBe(false);
await cache.flush();
expect(await createCache(storage).readAgent(SERVER_ID, "agent-1")).toBeUndefined();
expect(storage.changes.at(-1)).toMatchObject({
deletes: [{ serverId: SERVER_ID, kind: "agent", id: "agent-1" }],
@@ -1004,9 +657,7 @@ describe("ReplicaCache", () => {
payload: "{bad",
});
const cache = createCache(storage);
const restored = await cache.readDirectory(SERVER_ID);
await cache.flush();
const restored = await createCache(storage).readDirectory(SERVER_ID);
expect(restored.checkpoint).toEqual({
projects: { generation: "g", afterSeq: 4 },
@@ -1037,66 +688,6 @@ describe("ReplicaCache", () => {
]);
});
it.each(["nextOpenFailure", "nextReadAllFailure"] as const)(
"persists an accepted commit automatically after %s",
async (failure) => {
const storage = new MemoryStorage();
const cache = createCache(storage);
storage[failure] = new Error("storage busy");
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Retry me"));
await cache.flush();
expect(storage.rows.size).toBe(0);
expect(await cache.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline("Retry me"));
await expect.poll(() => storage.rows.size, { timeout: 2500 }).toBe(1);
const reopened = createCache(storage);
expect(await reopened.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline("Retry me"));
},
);
it("recovers a partial index load with queued host changes and the exact surviving budget", async () => {
const storage = new MemoryStorage();
const writer = createCache(storage);
writer.setHosts([SERVER_ID, "stale", "old", "removed"]);
for (const host of [SERVER_ID, "stale", "old", "removed"]) {
writer.commitTimeline(host, "agent-1", timeline("Stored"));
}
await writer.flush();
const survivorBytes = [...storage.rows.values()]
.filter((row) => row.serverId === SERVER_ID || row.serverId === "old")
.reduce((bytes, row) => bytes + Buffer.byteLength(row.payload), 0);
const cache = createCache(storage, survivorBytes);
cache.setHosts([SERVER_ID, "old", "removed"]);
const release = deferred();
storage.readAllGate = release.promise;
storage.nextDeleteFailure = new Error("stale host cleanup busy");
const loadsBefore = storage.readAlls;
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Newest"));
const writing = cache.flush();
try {
await expect.poll(() => storage.readAlls).toBe(loadsBefore + 1);
cache.reconcileServerId("old", "renamed");
cache.setHosts([SERVER_ID, "renamed"]);
} finally {
release.resolve();
await writing;
}
expect(await cache.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline("Newest"));
expect(await cache.readTimeline("renamed", "agent-1")).toEqual(timeline("Stored"));
expect(await cache.readTimeline("removed", "agent-1")).toBeUndefined();
await expect.poll(() => storage.changes.length, { timeout: 2500 }).toBe(2);
const reopened = createCache(storage);
reopened.setHosts([SERVER_ID, "renamed"]);
expect([...storage.rows.values()].map((row) => row.serverId).sort()).toEqual([
SERVER_ID,
"renamed",
]);
expect(await reopened.readTimeline(SERVER_ID, "agent-1")).toEqual(timeline("Newest"));
expect(await reopened.readTimeline("renamed", "agent-1")).toEqual(timeline("Stored"));
});
it("retries an explicit commit after a storage failure", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
@@ -1112,6 +703,21 @@ describe("ReplicaCache", () => {
]);
});
it("retries a timeline read invalidated by a concurrent directory commit", async () => {
const storage = new MemoryStorage();
const cache = createCache(storage);
cache.commitTimeline(SERVER_ID, "agent-1", timeline("Persisted timeline"));
await cache.flush();
storage.onRead = () => {
storage.onRead = null;
commitDirectory(cache, SERVER_ID, directory());
};
expect((await cache.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([
timelineItem("Persisted timeline"),
]);
});
it("rebuilds every directory row before restoring its checkpoint after eviction", async () => {
const storage = new MemoryStorage();
const cache = new ReplicaCache(storage, { ...noLegacyCleanup, maxBytes: 2_500 });
@@ -1153,7 +759,7 @@ describe("ReplicaCache", () => {
});
cache.setHosts([SERVER_ID]);
await cache.readTimeline(SERVER_ID, "missing");
await cache.readAgent(SERVER_ID, "missing");
commitDirectory(cache, SERVER_ID, directory());
await cache.flush();
packages/app/src/runtime/replica-cache/index.ts
+435 -436
View File
File diff suppressed because it is too large Load Diff
packages/app/src/runtime/replica-cache/row-store-sqlite.ts
+10 -7
View File
@@ -108,17 +108,20 @@ export function createSqliteReplicaRowStore(
}
async function read(
serverIds: readonly string[],
serverId: string,
kinds: readonly ReplicaRow["kind"][],
ids?: readonly string[],
): Promise<ReplicaRow[]> {
if (!serverIds.length || !kinds.length || ids?.length === 0) return [];
const placeholders = (values: readonly string[]) => values.map(() => "?").join(",");
const rows = await getConnection().all<StoredRow>(
`SELECT server_id, kind, id, payload FROM rows WHERE server_id IN (${placeholders(serverIds)}) AND kind IN (${placeholders(kinds)})${ids ? ` AND id IN (${placeholders(ids)})` : ""} ORDER BY server_id, kind, id`,
[...serverIds, ...kinds, ...(ids ?? [])],
if (kinds.length === 0 || ids?.length === 0) return [];
const placeholders = kinds.map(() => "?").join(", ");
const idClause = ids ? ` AND id IN (${ids.map(() => "?").join(", ")})` : "";
const storedRows = await getConnection().all<StoredRow>(
`SELECT server_id, kind, id, payload FROM rows
WHERE server_id = ? AND kind IN (${placeholders})${idClause}
ORDER BY kind, id`,
[serverId, ...kinds, ...(ids ?? [])],
);
return rows.map((row) => ({
return storedRows.map((row) => ({
serverId: row.server_id,
kind: row.kind,
id: row.id,
packages/app/src/runtime/replica-cache/row-store.contract.ts
+9 -12
View File
@@ -55,23 +55,20 @@ export function runReplicaRowStoreContract(
]);
});
it("reads only requested hosts, kinds and identities", async () => {
const selected = row("server-a", "timeline", "wanted", "page");
it("reads only requested kinds for one host", async () => {
await store.apply({
deletes: [],
upserts: [
selected,
row("server-a", "timeline", "other", "unrelated timeline"),
row("server-a", "agent", "wanted", "directory"),
row("server-b", "timeline", "wanted", "other host"),
row("server-a", "agent", "agent-1", "agent"),
row("server-a", "timeline", "singleton", "timeline"),
row("server-b", "agent", "agent-2", "other host"),
],
deletes: [],
});
expect(await store.read(["server-a"], ["timeline"], ["wanted"])).toEqual([selected]);
expect(await store.read(["server-a"], ["agent"])).toEqual([
row("server-a", "agent", "wanted", "directory"),
expect(await store.read("server-a", ["timeline"])).toEqual([
row("server-a", "timeline", "singleton", "timeline"),
]);
expect(await store.read([], ["timeline"])).toEqual([]);
expect(await store.read(["server-a"], ["timeline"], [])).toEqual([]);
expect(await store.read("server-a", ["agent"], ["missing"])).toEqual([]);
});
it("overwrites an existing row on upsert", async () => {
packages/app/src/runtime/replica-cache/row-store.ts
+1 -1
View File
@@ -25,7 +25,7 @@ export interface ReplicaHostRows {
export interface ReplicaRowStore {
open(): Promise<void>;
read(
serverIds: readonly string[],
serverId: string,
kinds: readonly ReplicaRowKind[],
ids?: readonly string[],
): Promise<ReplicaRow[]>;
packages/app/src/runtime/replica-cache/row-store.web.test.ts
+16 -29
View File
@@ -24,41 +24,28 @@ runReplicaRowStoreContract("IndexedDB", async () => {
};
});
it("reads requested IndexedDB keys without scanning unrelated rows", async () => {
const options = {
it("uses exact IndexedDB keys for targeted rows instead of scanning the host", async () => {
const store = createIndexedDbReplicaRowStore({
databaseName: `replica-row-store-targeted-${databaseSequence++}`,
schemaVersion: 1,
};
const writer = createIndexedDbReplicaRowStore(options);
await writer.open();
await writer.apply({
});
await store.open();
await store.apply({
upserts: [
{ serverId: "host-a", kind: "timeline", id: "agent", payload: "a" },
{ serverId: "host-b", kind: "timeline", id: "agent", payload: "b" },
{ serverId: "host-a", kind: "timeline", id: "unrelated", payload: "large history" },
{ serverId: "host-a", kind: "agent", id: "agent", payload: "metadata" },
{ serverId: "server-a", kind: "agent", id: "agent-1", payload: "agent" },
{ serverId: "server-a", kind: "workspace", id: "workspace-1", payload: "workspace" },
],
deletes: [],
});
const reader = createIndexedDbReplicaRowStore(options);
await reader.open();
const get = vi.spyOn(FakeIDBObjectStore.prototype, "get");
const getAll = vi.spyOn(FakeIDBObjectStore.prototype, "getAll");
const openCursor = vi.spyOn(FakeIDBObjectStore.prototype, "openCursor");
try {
expect(await reader.read(["host-a", "host-b"], ["timeline"], ["agent"])).toEqual([
{ serverId: "host-a", kind: "timeline", id: "agent", payload: "a" },
{ serverId: "host-b", kind: "timeline", id: "agent", payload: "b" },
]);
expect(get.mock.calls).toEqual([
[["host-a", "timeline", "agent"]],
[["host-b", "timeline", "agent"]],
]);
expect(getAll).not.toHaveBeenCalled();
expect(openCursor).not.toHaveBeenCalled();
} finally {
get.mockRestore();
getAll.mockRestore();
openCursor.mockRestore();
}
expect(await store.read("server-a", ["agent"], ["agent-1"])).toEqual([
{ serverId: "server-a", kind: "agent", id: "agent-1", payload: "agent" },
]);
expect(get).toHaveBeenCalledWith(["server-a", "agent", "agent-1"]);
expect(getAll).not.toHaveBeenCalled();
get.mockRestore();
getAll.mockRestore();
});
packages/app/src/runtime/replica-cache/row-store.web.ts
+32 -33
View File
@@ -136,6 +136,38 @@ export function createIndexedDbReplicaRowStore(
return Array.from(hosts, ([serverId, hostRows]) => ({ serverId, rows: hostRows }));
}
async function read(
serverId: string,
kinds: readonly ReplicaRow["kind"][],
ids?: readonly string[],
): Promise<ReplicaRow[]> {
if (kinds.length === 0 || ids?.length === 0) return [];
const transaction = getDatabase().transaction(ROWS_STORE, "readonly");
const completion = transactionComplete(transaction);
const store = transaction.objectStore(ROWS_STORE);
const requests = ids
? kinds.flatMap((kind) =>
ids.map(async (id) => {
const row = await requestResult<ReplicaRow | undefined>(
store.get([serverId, kind, id]),
);
return row ? [row] : [];
}),
)
: kinds.map((kind) =>
requestResult<ReplicaRow[]>(
store.getAll(IDBKeyRange.bound([serverId, kind], [serverId, kind, []])),
),
);
const rows = (await Promise.all(requests)).flat();
await completion;
return rows.sort((left, right) =>
left.kind === right.kind
? left.id.localeCompare(right.id)
: left.kind.localeCompare(right.kind),
);
}
async function apply(changes: ReplicaRowChanges): Promise<void> {
await runTransaction(getDatabase(), ROWS_STORE, async (transaction) => {
const rows = transaction.objectStore(ROWS_STORE);
@@ -148,39 +180,6 @@ export function createIndexedDbReplicaRowStore(
});
}
async function read(
serverIds: readonly string[],
kinds: readonly ReplicaRow["kind"][],
ids?: readonly string[],
): Promise<ReplicaRow[]> {
const transaction = getDatabase().transaction(ROWS_STORE, "readonly");
const completion = transactionComplete(transaction);
const store = transaction.objectStore(ROWS_STORE);
const requests: Promise<ReplicaRow[]>[] = [];
for (const serverId of serverIds) {
for (const kind of kinds) {
if (ids) {
for (const id of ids) {
requests.push(
requestResult<ReplicaRow | undefined>(store.get([serverId, kind, id])).then((row) =>
row ? [row] : [],
),
);
}
} else {
requests.push(
requestResult<ReplicaRow[]>(
store.getAll(IDBKeyRange.bound([serverId, kind], [serverId, kind, []])),
),
);
}
}
}
const rows = (await Promise.all(requests)).flat();
await completion;
return rows;
}
async function deleteHost(serverId: string): Promise<void> {
await runTransaction(getDatabase(), ROWS_STORE, async (transaction) => {
const rows = transaction.objectStore(ROWS_STORE);