Skip to content

perf(agents): reduce subagent streaming overhead #202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
badcuban merged 1 commit into from
Aug 31, 2026
Merged

perf(agents): reduce subagent streaming overhead #202

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
349 changes: 349 additions & 0 deletions apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1450,6 +1450,56 @@ describe("ProviderRuntimeIngestion", () => {
),
);

harness.emit({
type: "content.delta",
eventId: asEventId("evt-subagent-result-before-abort-1"),
provider: ProviderDriverKind.make("codex"),
threadId: asThreadId("thread-1"),
createdAt: "2026-01-01T00:00:03.500Z",
turnId: asTurnId("turn-abort"),
itemId: asItemId("item-subagent-result-before-abort"),
providerRefs: {
providerThreadId: "agent-live-before-abort",
providerTurnId: "agent-turn-before-abort",
providerItemId: asItemId("item-subagent-result-before-abort"),
},
payload: {
streamKind: "assistant_text",
delta: "First chunk. ",
},
});
await waitForThread(harness.readModel, (thread) =>
thread.activities.some((activity) => {
const payload = activity.payload as {
sourceAgentThreadId?: string;
data?: { subagentLiveText?: string };
};
return (
activity.kind === "subagent.result" &&
payload.sourceAgentThreadId === "agent-live-before-abort" &&
payload.data?.subagentLiveText === "First chunk. "
);
}),
);
harness.emit({
type: "content.delta",
eventId: asEventId("evt-subagent-result-before-abort-2"),
provider: ProviderDriverKind.make("codex"),
threadId: asThreadId("thread-1"),
createdAt: "2026-01-01T00:00:04.000Z",
turnId: asTurnId("turn-abort"),
itemId: asItemId("item-subagent-result-before-abort"),
providerRefs: {
providerThreadId: "agent-live-before-abort",
providerTurnId: "agent-turn-before-abort",
providerItemId: asItemId("item-subagent-result-before-abort"),
},
payload: {
streamKind: "assistant_text",
delta: "Second chunk.",
},
});

harness.emit({
type: "turn.aborted",
eventId: asEventId("evt-turn-aborted"),
Expand Down Expand Up @@ -1482,6 +1532,29 @@ describe("ProviderRuntimeIngestion", () => {
(subagent) => subagent.agentThreadId === "agent-completed-before-abort",
)?.status,
).toBe("completed");
const streamedResult = thread.activities.find((activity) => {
const payload = activity.payload as { sourceAgentThreadId?: string };
return (
activity.kind === "subagent.result" &&
payload.sourceAgentThreadId === "agent-live-before-abort"
);
});
expect(streamedResult?.payload).toMatchObject({
data: {
subagentLiveText: "First chunk. Second chunk.",
},
});

await Effect.runPromise(Effect.sleep("150 millis"));
await harness.drain();
const settledThread = (await harness.readModel()).threads.find(
(entry) => entry.id === "thread-1",
);
expect(
settledThread?.subagents?.find(
(subagent) => subagent.agentThreadId === "agent-live-before-abort",
)?.status,
).toBe("interrupted");
});

it("settles live subagents when turn.completed reports an interruption", async () => {
Expand Down Expand Up @@ -2171,6 +2244,282 @@ describe("ProviderRuntimeIngestion", () => {
status: "completed",
message: "The runtime path is correct.",
});

harness.emit({
type: "content.delta",
eventId: asEventId("evt-child-final-late-delta"),
provider: ProviderDriverKind.make("codex"),
createdAt: now,
threadId: asThreadId("thread-1"),
turnId: asTurnId("turn-parent"),
itemId: asItemId("child-message-final"),
providerRefs: {
providerThreadId: "child-provider-thread",
providerTurnId: "child-turn-1",
providerItemId: asItemId("child-message-final"),
},
payload: {
streamKind: "assistant_text",
delta: " Late duplicate text.",
},
});
await harness.drain();

const afterLateDelta = (await harness.readModel()).threads.find(
(entry) => entry.id === "thread-1",
);
const finalActivity = afterLateDelta?.activities.find((entry) => entry.id === activity?.id);
expect(finalActivity?.payload).toMatchObject({
status: "completed",
data: {
item: {
agentsStates: {
"child-provider-thread": {
status: "completed",
message: "The runtime path is correct.",
},
},
},
},
});
});

it("retains a child result when turn and session completion arrive before item completion", async () => {
const harness = await createHarness();
const now = "2026-01-01T00:00:00.000Z";

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-set-out-of-order-child-completion"),
threadId: ThreadId.make("thread-1"),
session: {
threadId: ThreadId.make("thread-1"),
status: "running",
providerName: "codex",
runtimeMode: "approval-required",
activeTurnId: asTurnId("turn-out-of-order-child-completion"),
providerThreadId: "parent-provider-thread",
updatedAt: now,
lastError: null,
},
createdAt: now,
}),
);

for (const [index, delta] of ["Full child ", "result."].entries()) {
harness.emit({
type: "content.delta",
eventId: asEventId(`evt-out-of-order-child-delta-${index}`),
provider: ProviderDriverKind.make("codex"),
createdAt: now,
threadId: asThreadId("thread-1"),
turnId: asTurnId("turn-out-of-order-child-completion"),
itemId: asItemId("out-of-order-child-message"),
providerRefs: {
providerThreadId: "out-of-order-child-provider-thread",
providerTurnId: "out-of-order-child-turn",
providerItemId: asItemId("out-of-order-child-message"),
},
payload: {
streamKind: "assistant_text",
delta,
},
});
}

harness.emit({
type: "turn.completed",
eventId: asEventId("evt-out-of-order-parent-turn-completed"),
provider: ProviderDriverKind.make("codex"),
createdAt: now,
threadId: asThreadId("thread-1"),
turnId: asTurnId("turn-out-of-order-child-completion"),
payload: {
state: "completed",
},
});
harness.emit({
type: "session.exited",
eventId: asEventId("evt-out-of-order-session-exited"),
provider: ProviderDriverKind.make("codex"),
threadId: asThreadId("thread-1"),
createdAt: now,
payload: {
reason: "provider exited",
exitKind: "graceful",
},
});
harness.emit({
type: "item.completed",
eventId: asEventId("evt-out-of-order-child-item-completed"),
provider: ProviderDriverKind.make("codex"),
createdAt: now,
threadId: asThreadId("thread-1"),
turnId: asTurnId("turn-out-of-order-child-completion"),
itemId: asItemId("out-of-order-child-message"),
providerRefs: {
providerThreadId: "out-of-order-child-provider-thread",
providerTurnId: "out-of-order-child-turn",
providerItemId: asItemId("out-of-order-child-message"),
},
payload: {
itemType: "assistant_message",
status: "completed",
data: {
item: {
phase: "final_answer",
},
},
},
});
await harness.drain();

const thread = (await harness.readModel()).threads.find((entry) => entry.id === "thread-1");
const result = thread?.activities.find((entry) => {
const resultPayload = entry.payload as { sourceAgentThreadId?: string; status?: string };
return (
entry.kind === "subagent.result" &&
resultPayload.sourceAgentThreadId === "out-of-order-child-provider-thread" &&
resultPayload.status === "completed"
);
});
expect(result?.payload).toMatchObject({
data: {
item: {
agentsStates: {
"out-of-order-child-provider-thread": {
status: "completed",
message: "Full child result.",
},
},
},
},
});
});

it("keeps rapid updates from three child agents independent", async () => {
const harness = await createHarness();
const now = "2026-01-01T00:00:00.000Z";
const childProviderThreadIds = ["child-provider-a", "child-provider-b", "child-provider-c"];
const readSubagentResultEvents = async () => {
const events = await Effect.runPromise(
Stream.runCollect(harness.engine.readEvents(0)).pipe(
Effect.map((chunk) => Array.from(chunk)),
),
);
return events.filter((event) => {
if (event.type !== "thread.activity-appended") {
return false;
}
const activity = event.payload.activity;
return (
activity.kind === "subagent.result" &&
childProviderThreadIds.includes(
(activity.payload as { sourceAgentThreadId?: string }).sourceAgentThreadId ?? "",
)
);
});
};

await Effect.runPromise(
harness.engine.dispatch({
type: "thread.session.set",
commandId: CommandId.make("cmd-session-set-three-streaming-children"),
threadId: ThreadId.make("thread-1"),
session: {
threadId: ThreadId.make("thread-1"),
status: "running",
providerName: "codex",
runtimeMode: "approval-required",
activeTurnId: asTurnId("turn-three-streaming-children"),
providerThreadId: "parent-provider-thread",
updatedAt: now,
lastError: null,
},
createdAt: now,
}),
);

for (const [childIndex, childProviderThreadId] of childProviderThreadIds.entries()) {
harness.emit({
type: "content.delta",
eventId: asEventId(`evt-three-streaming-children-${childIndex}-0`),
provider: ProviderDriverKind.make("codex"),
createdAt: now,
threadId: asThreadId("thread-1"),
turnId: asTurnId("turn-three-streaming-children"),
itemId: asItemId(`child-message-${childIndex}`),
providerRefs: {
providerThreadId: childProviderThreadId,
providerTurnId: `child-turn-${childIndex}`,
providerItemId: asItemId(`child-message-${childIndex}`),
},
payload: {
streamKind: "assistant_text",
delta: `${childProviderThreadId}:0`,
},
});
}

await waitForThread(harness.readModel, (thread) => {
const sources = new Set(
thread.activities
.filter((activity) => activity.kind === "subagent.result")
.map(
(activity) =>
(activity.payload as { sourceAgentThreadId?: string }).sourceAgentThreadId,
),
);
return childProviderThreadIds.every((childProviderThreadId) =>
sources.has(childProviderThreadId),
);
});
expect(await readSubagentResultEvents()).toHaveLength(3);

for (let updateIndex = 1; updateIndex < 10; updateIndex += 1) {
for (const [childIndex, childProviderThreadId] of childProviderThreadIds.entries()) {
harness.emit({
type: "content.delta",
eventId: asEventId(`evt-three-streaming-children-${childIndex}-${updateIndex}`),
provider: ProviderDriverKind.make("codex"),
createdAt: now,
threadId: asThreadId("thread-1"),
turnId: asTurnId("turn-three-streaming-children"),
itemId: asItemId(`child-message-${childIndex}`),
providerRefs: {
providerThreadId: childProviderThreadId,
providerTurnId: `child-turn-${childIndex}`,
providerItemId: asItemId(`child-message-${childIndex}`),
},
payload: {
streamKind: "assistant_text",
delta: `|${updateIndex}`,
},
});
}
}

const thread = await waitForThread(harness.readModel, (entry) =>
childProviderThreadIds.every((childProviderThreadId) =>
entry.activities.some((activity) => {
const payload = activity.payload as {
sourceAgentThreadId?: string;
data?: { subagentLiveText?: string };
};
return (
activity.kind === "subagent.result" &&
payload.sourceAgentThreadId === childProviderThreadId &&
payload.data?.subagentLiveText === `${childProviderThreadId}:0|1|2|3|4|5|6|7|8|9`
);
}),
),
);
expect(
thread.activities.filter((activity) => activity.kind === "subagent.result"),
).toHaveLength(3);

expect(await readSubagentResultEvents()).toHaveLength(6);
});

it("uses the durable child roster when the projected parent provider id is missing", async () => {
Expand Down
Loading
Loading