fix(sync): centralize directory replica ordering

Own agent and workspace refresh transactions in HostRuntime so reconnect epochs, buffered updates, and lifecycle invalidation share one ordering boundary. Route timeline metadata and mixed-capability stream delivery through their authoritative runtime sources.
This commit is contained in:
Mohamed Boudra
2026-07-17 14:58:21 +02:00
parent 1774ce3c4c
commit 467fb47012
24 changed files with 1245 additions and 972 deletions
@@ -4721,6 +4721,58 @@ test("unions viewed timelines across socket sources and removes detached sources
).toEqual(["agent-b"]);
});
test("keeps selective delivery scoped per socket when a retained session also has a legacy socket", async () => {
const messages: SessionOutboundMessage[] = [];
const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];
const session = createSessionForTest({
messages,
targetedMessages,
agentManager: {
subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => {
agentEventListeners.push(listener);
return () => {};
}),
},
});
const legacySocket = {};
const selectiveSocket = {};
session.updateClientCapabilities(null, legacySocket);
session.updateClientCapabilities({ selective_agent_timeline: true }, selectiveSocket);
await session.handleMessage(
{
type: "agent.timeline.set_subscription.request",
agentIds: ["viewed-agent"],
requestId: "timeline-subscription-selective",
},
selectiveSocket,
);
targetedMessages.length = 0;
const listener = agentEventListeners[0];
if (!listener) throw new Error("Agent event listener was not installed");
listener({
type: "agent_stream",
agentId: "not-viewed-agent",
event: {
type: "timeline",
provider: "mock",
item: { type: "assistant_message", messageId: "message-global", text: "global" },
},
});
expect(messages).toEqual([]);
expect(targetedMessages).toEqual([
{
source: legacySocket,
message: expect.objectContaining({
type: "agent_stream",
payload: expect.objectContaining({ agentId: "not-viewed-agent" }),
}),
},
]);
});
describe("agent config setters", () => {
test("set_agent_mode_request: success emits accepted response carrying the notice", async () => {
const messages: SessionOutboundMessage[] = [];
+56 -22
View File
@@ -1019,6 +1019,61 @@ export class Session {
return true;
}
private forwardAgentStream(
event: Extract<AgentManagerEvent, { type: "agent_stream" }>,
serializedEvent: Extract<SessionOutboundMessage, { type: "agent_stream" }>["payload"]["event"],
): void {
if (this.selectiveTimelineCapabilityBySource.size === 0 || !this.onMessageToSource) {
if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") {
this.emit({
type: "agent_attention_required",
payload: {
agentId: event.agentId,
reason: serializedEvent.reason,
timestamp: serializedEvent.timestamp,
shouldNotify: serializedEvent.shouldNotify,
...(serializedEvent.notification ? { notification: serializedEvent.notification } : {}),
},
});
} else if (
!this.usesSelectiveTimelineDelivery() ||
this.viewedTimelineAgentIds.has(event.agentId)
) {
this.emit({
type: "agent_stream",
payload: this.buildAgentStreamPayload(event, serializedEvent),
});
}
return;
}
for (const [source, supportsSelectiveDelivery] of this.selectiveTimelineCapabilityBySource) {
if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") {
this.onMessageToSource(source, {
type: "agent_attention_required",
payload: {
agentId: event.agentId,
reason: serializedEvent.reason,
timestamp: serializedEvent.timestamp,
shouldNotify: serializedEvent.shouldNotify,
...(serializedEvent.notification ? { notification: serializedEvent.notification } : {}),
},
});
continue;
}
if (
supportsSelectiveDelivery &&
!this.viewedTimelineAgentIdsBySource.get(source)?.has(event.agentId)
) {
continue;
}
this.onMessageToSource(source, {
type: "agent_stream",
payload: this.buildAgentStreamPayload(event, serializedEvent),
});
}
}
supports(capability: ClientCapability): boolean {
return this.clientCapabilities.has(capability);
}
@@ -1340,28 +1395,7 @@ export class Session {
"agent.session.forward_stream",
);
if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") {
this.emit({
type: "agent_attention_required",
payload: {
agentId: event.agentId,
reason: serializedEvent.reason,
timestamp: serializedEvent.timestamp,
shouldNotify: serializedEvent.shouldNotify,
...(serializedEvent.notification
? { notification: serializedEvent.notification }
: {}),
},
});
} else if (
!this.usesSelectiveTimelineDelivery() ||
this.viewedTimelineAgentIds.has(event.agentId)
) {
this.emit({
type: "agent_stream",
payload: this.buildAgentStreamPayload(event, serializedEvent),
});
}
this.forwardAgentStream(event, serializedEvent);
if (event.event.type === "permission_requested") {
this.emit({