fix(sync): preserve mixed-client delivery guarantees

This commit is contained in:
Mohamed Boudra
2026-07-17 11:12:46 +02:00
parent e132a6106b
commit 02298d226c
4 changed files with 68 additions and 12 deletions
+2 -1
View File
@@ -907,7 +907,7 @@ function ChatAgentContent({
if (!agentId) {
return;
}
if (agentState.id) {
if (agentState.id && hasAppliedAuthoritativeHistory) {
if (missingAgentState.kind === "resolving" || missingAgentState.kind === "not_found") {
setMissingAgentState(reconcileMissingAgentStateWithPresentAgent);
}
@@ -964,6 +964,7 @@ function ChatAgentContent({
});
}, [
agentState.id,
hasAppliedAuthoritativeHistory,
agentId,
client,
ensureAgentIsInitialized,
@@ -4592,6 +4592,8 @@ test("unions viewed timelines across socket sources and removes detached sources
session.updateClientCapabilities({ selective_agent_timeline: true });
const firstSocket = {};
const secondSocket = {};
session.updateClientCapabilities({ selective_agent_timeline: true }, firstSocket);
session.updateClientCapabilities({ selective_agent_timeline: true }, secondSocket);
await session.handleMessage(
{
@@ -4635,6 +4637,22 @@ test("unions viewed timelines across socket sources and removes detached sources
});
expect(messages.filter((message) => message.type === "agent_stream")).toHaveLength(2);
const legacySocket = {};
session.updateClientCapabilities(null, legacySocket);
messages.length = 0;
forward({
type: "agent_stream",
agentId: "agent-not-viewed",
event: {
type: "timeline",
provider: "mock",
item: { type: "assistant_message", messageId: "message-legacy", text: "legacy" },
},
});
expect(messages.some((message) => message.type === "agent_stream")).toBe(true);
session.clearAgentTimelineSubscription(legacySocket);
session.clearAgentTimelineSubscription(firstSocket);
messages.length = 0;
forward({
+46 -10
View File
@@ -585,6 +585,7 @@ export class Session {
private unsubscribeAgentEvents: (() => void) | null = null;
private viewedTimelineAgentIds = new Set<string>();
private readonly viewedTimelineAgentIdsBySource = new Map<object, Set<string>>();
private readonly selectiveTimelineCapabilityBySource = new Map<object, boolean>();
private readonly defaultTimelineSubscriptionSource = {};
private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null;
private readonly agentUpdates: AgentUpdatesService;
@@ -966,17 +967,25 @@ export class Session {
}
}
updateClientCapabilities(capabilities: Record<string, unknown> | null): void {
updateClientCapabilities(capabilities: Record<string, unknown> | null, source?: object): void {
this.clientCapabilities = parseClientCapabilities(capabilities);
if (!this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
if (source) {
this.selectiveTimelineCapabilityBySource.set(
source,
this.supports(CLIENT_CAPS.selectiveAgentTimeline),
);
}
if (!source && !this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
this.viewedTimelineAgentIdsBySource.clear();
this.viewedTimelineAgentIds.clear();
}
}
clearAgentTimelineSubscription(source: object): void {
if (!this.viewedTimelineAgentIdsBySource.delete(source)) return;
this.rebuildViewedTimelineAgentIds();
this.selectiveTimelineCapabilityBySource.delete(source);
if (this.viewedTimelineAgentIdsBySource.delete(source)) {
this.rebuildViewedTimelineAgentIds();
}
}
private replaceAgentTimelineSubscription(source: object | undefined, agentIds: string[]): void {
@@ -994,6 +1003,16 @@ export class Session {
this.viewedTimelineAgentIds = viewedAgentIds;
}
private usesSelectiveTimelineDelivery(): boolean {
if (this.selectiveTimelineCapabilityBySource.size === 0) {
return this.supports(CLIENT_CAPS.selectiveAgentTimeline);
}
for (const capable of this.selectiveTimelineCapabilityBySource.values()) {
if (!capable) return false;
}
return true;
}
supports(capability: ClientCapability): boolean {
return this.clientCapabilities.has(capability);
}
@@ -1308,10 +1327,7 @@ export class Session {
"agent.session.forward_stream",
);
if (
this.supports(CLIENT_CAPS.selectiveAgentTimeline) &&
serializedEvent.type === "attention_required"
) {
if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") {
this.emit({
type: "agent_attention_required",
payload: {
@@ -1325,7 +1341,7 @@ export class Session {
},
});
} else if (
!this.supports(CLIENT_CAPS.selectiveAgentTimeline) ||
!this.usesSelectiveTimelineDelivery() ||
this.viewedTimelineAgentIds.has(event.agentId)
) {
this.emit({
@@ -1587,7 +1603,12 @@ export class Session {
return this.handleProviderSubagentTimelineRequest(msg);
case "agent.timeline.set_subscription.request": {
const agentIds = [...new Set(msg.agentIds)].sort();
if (this.supports(CLIENT_CAPS.selectiveAgentTimeline)) {
if (
source
? (this.selectiveTimelineCapabilityBySource.get(source) ??
this.supports(CLIENT_CAPS.selectiveAgentTimeline))
: this.supports(CLIENT_CAPS.selectiveAgentTimeline)
) {
this.replaceAgentTimelineSubscription(source, agentIds);
}
this.emit({
@@ -4590,6 +4611,7 @@ export class Session {
"fetch_workspaces_response_ready",
);
const snapshot = this.buildBootstrapSnapshot(payload.entries);
this.seedWorkspaceSubscriptionSnapshot(subscriptionId, payload.entries);
this.emit({
type: "fetch_workspaces_response",
@@ -4650,6 +4672,20 @@ export class Session {
return { snapshotByWorkspaceId };
}
private seedWorkspaceSubscriptionSnapshot(
subscriptionId: string | null,
entries: FetchWorkspacesResponseEntry[],
): void {
const subscription = this.workspaceUpdatesSubscription;
if (!subscriptionId || subscription?.subscriptionId !== subscriptionId) return;
for (const entry of entries) {
subscription.lastEmittedByWorkspaceId.set(entry.id, {
kind: "upsert",
workspace: entry,
});
}
}
private async registerWorkspaceForImportedAgent(
workspace: PersistedWorkspaceRecord,
): Promise<void> {
@@ -1093,6 +1093,7 @@ export class VoiceAssistantWebSocketServer {
sockets: new Set([ws]),
externalDisconnectCleanupTimeout: null,
};
session.updateClientCapabilities(clientCapabilities, ws);
return connection;
}
@@ -1166,7 +1167,7 @@ export class VoiceAssistantWebSocketServer {
// COMPAT(selectiveAgentTimeline): added in v0.1.106. Every capable resumed
// hello resets membership before server_info so stale retained-session
// state cannot leak. Remove after 2027-01-12.
existing.session.updateClientCapabilities(newClientCapabilities);
existing.session.updateClientCapabilities(newClientCapabilities, ws);
if (
JSON.stringify(existing.clientCapabilities ?? null) !==
JSON.stringify(newClientCapabilities ?? null)