Show provider subagents in the subagents track (#2013)
* Add provider-owned subagent stream contract * OpenCode provider: detect child sessions, attach adopted children to the hosting server, synthesize external turns - Emit internal provider_child_session_detected / provider_session_deleted stream events from the global event stream (before the no-active-turn gate), including BFS hydration of existing descendants on subscribe - Track child sessions in a registry mapping child session id -> hosting server url; resumeSession attaches via acquireExisting() so adopted children share the parent's server generation and keep receiving live events - Synthesize external turns for adopted (externally driven) sessions so timeline, tool calls, and permission requests stream without a Paseo-initiated prompt - Honor launchContext.env in resumeSession (acquireDedicated parity with createSession) - Stop folding child tool parts into the parent's sub_agent action log; the parent row stays as a pointer * Render provider subagents in workspace timelines * fix provider subagent timeline reliability * fix provider subagent lifecycle edge cases * fix provider subagent event fidelity * Fix provider subagent history fidelity * Fix restored provider subagent state * Fix provider subagent visibility edges * Restore provider subagent timelines * Page provider subagent history * Forward subagent questions reliably * fix(subagents): preserve OpenCode child context * fix(subagents): harden provider timeline routing * fix(subagents): preserve provider page continuity * fix(subagents): retain live child prompts --------- Co-authored-by: Omer <639682+omercnet@users.noreply.github.com>
This commit is contained in:
Mohamed Boudra
@@ -2331,6 +2331,27 @@ test("importProviderSession imports the selected session without listing and pub
|
||||
timestamp: "2026-01-02T00:00:02.000Z",
|
||||
},
|
||||
],
|
||||
providerSubagentEvents: [
|
||||
{
|
||||
type: "provider_subagent" as const,
|
||||
provider: "codex" as const,
|
||||
event: {
|
||||
type: "upsert" as const,
|
||||
id: "thread-child",
|
||||
title: "Imported child",
|
||||
status: "completed" as const,
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "provider_subagent" as const,
|
||||
provider: "codex" as const,
|
||||
event: {
|
||||
type: "timeline" as const,
|
||||
id: "thread-child",
|
||||
item: { type: "assistant_message" as const, text: "Child result" },
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -2373,7 +2394,13 @@ test("importProviderSession imports the selected session without listing and pub
|
||||
},
|
||||
},
|
||||
]);
|
||||
expect(events).toHaveLength(1);
|
||||
expect(manager.listProviderSubagents(imported.id)).toEqual([
|
||||
expect.objectContaining({ id: "thread-child", title: "Imported child", status: "completed" }),
|
||||
]);
|
||||
expect(manager.fetchProviderSubagentTimeline(imported.id, "thread-child").rows).toEqual([
|
||||
expect.objectContaining({ item: { type: "assistant_message", text: "Child result" } }),
|
||||
]);
|
||||
expect(events).toHaveLength(3);
|
||||
expect(events[0]).toMatchObject({
|
||||
type: "agent_state",
|
||||
agent: {
|
||||
@@ -2553,6 +2580,150 @@ test("reloadAgentSession preserves timeline and does not force history replay",
|
||||
expect(afterHydrate).toEqual(beforeReload);
|
||||
});
|
||||
|
||||
test("reloadAgentSession clears provider children before rehydrating from disk", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-reload-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
let activeSession: TestAgentSession | null = null;
|
||||
class ProviderChildClient extends TestAgentClient {
|
||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
activeSession = new TestAgentSession(config);
|
||||
return activeSession;
|
||||
}
|
||||
|
||||
override async resumeSession(
|
||||
_handle: AgentPersistenceHandle,
|
||||
config?: Partial<AgentSessionConfig>,
|
||||
): Promise<AgentSession> {
|
||||
return new TestAgentSession({
|
||||
provider: "codex",
|
||||
cwd: config?.cwd ?? workdir,
|
||||
});
|
||||
}
|
||||
}
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new ProviderChildClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
idFactory: () => "00000000-0000-4000-8000-000000000116",
|
||||
});
|
||||
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
activeSession?.pushEvent({
|
||||
type: "provider_subagent",
|
||||
provider: "codex",
|
||||
event: { type: "upsert", id: "stale-child", title: "Stale child", status: "running" },
|
||||
});
|
||||
await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1));
|
||||
|
||||
await manager.reloadAgentSession(snapshot.id, undefined, { rehydrateFromDisk: true });
|
||||
|
||||
expect(manager.listProviderSubagents(snapshot.id)).toEqual([]);
|
||||
});
|
||||
|
||||
test("hydrateTimelineFromProvider restores and broadcasts provider children from session history", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-history-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
class ProviderChildHistorySession extends TestAgentSession {
|
||||
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
yield {
|
||||
type: "provider_subagent",
|
||||
provider: "codex",
|
||||
event: {
|
||||
type: "upsert",
|
||||
id: "restored-child",
|
||||
title: "Restored child",
|
||||
status: "completed",
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
class ProviderChildHistoryClient extends TestAgentClient {
|
||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
return new ProviderChildHistorySession(config);
|
||||
}
|
||||
}
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new ProviderChildHistoryClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
idFactory: () => "00000000-0000-4000-8000-000000000117",
|
||||
});
|
||||
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
const events: AgentManagerEvent[] = [];
|
||||
manager.subscribe((event) => events.push(event), {
|
||||
agentId: snapshot.id,
|
||||
replayState: false,
|
||||
});
|
||||
|
||||
await manager.hydrateTimelineFromProvider(snapshot.id, { broadcast: true });
|
||||
|
||||
expect(manager.listProviderSubagents(snapshot.id)).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "restored-child",
|
||||
parentAgentId: snapshot.id,
|
||||
title: "Restored child",
|
||||
status: "completed",
|
||||
}),
|
||||
]);
|
||||
expect(events).toContainEqual({
|
||||
type: "provider_subagent",
|
||||
event: {
|
||||
type: "upsert",
|
||||
subagent: expect.objectContaining({
|
||||
id: "restored-child",
|
||||
parentAgentId: snapshot.id,
|
||||
}),
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
test("force provider hydration removes children absent from current history", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-child-force-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
let session: TestAgentSession | null = null;
|
||||
class ProviderChildForceClient extends TestAgentClient {
|
||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
session = new TestAgentSession(config);
|
||||
return session;
|
||||
}
|
||||
}
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new ProviderChildForceClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
idFactory: () => "00000000-0000-4000-8000-000000000118",
|
||||
});
|
||||
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
session?.pushEvent({
|
||||
type: "provider_subagent",
|
||||
provider: "codex",
|
||||
event: { type: "upsert", id: "removed-by-rewind", status: "completed" },
|
||||
});
|
||||
await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1));
|
||||
const events: AgentManagerEvent[] = [];
|
||||
manager.subscribe((event) => events.push(event), {
|
||||
agentId: snapshot.id,
|
||||
replayState: false,
|
||||
});
|
||||
|
||||
await manager.hydrateTimelineFromProvider(snapshot.id, { force: true, broadcast: true });
|
||||
|
||||
expect(manager.listProviderSubagents(snapshot.id)).toEqual([]);
|
||||
expect(events).toContainEqual({
|
||||
type: "provider_subagent",
|
||||
event: {
|
||||
type: "remove",
|
||||
parentAgentId: snapshot.id,
|
||||
subagentId: "removed-by-rewind",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
test("reloadAgentSession preserves current title when config title is unset", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-title-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
@@ -5006,6 +5177,67 @@ test("subscribe does not emit state events for internal agents to global subscri
|
||||
expect(receivedEvents.filter((id) => id === generatedAgentIds[1]).length).toBe(0);
|
||||
});
|
||||
|
||||
test("subscribe hides provider subagents of internal parents from global subscribers", async () => {
|
||||
const internalAgentId = "00000000-0000-4000-8000-000000000117";
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-internal-provider-child-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
const sessionHolder: { current: TestAgentSession | null } = { current: null };
|
||||
class InternalProviderChildClient extends TestAgentClient {
|
||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
sessionHolder.current = new TestAgentSession(config);
|
||||
return sessionHolder.current;
|
||||
}
|
||||
}
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new InternalProviderChildClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
idFactory: () => internalAgentId,
|
||||
});
|
||||
const globalEvents: AgentManagerEvent[] = [];
|
||||
const scopedEvents: AgentManagerEvent[] = [];
|
||||
manager.subscribe((event) => globalEvents.push(event), { replayState: false });
|
||||
await manager.createAgent(
|
||||
{ provider: "codex", cwd: workdir, title: "Internal Agent", internal: true },
|
||||
undefined,
|
||||
{ workspaceId: undefined },
|
||||
);
|
||||
manager.subscribe((event) => scopedEvents.push(event), {
|
||||
agentId: internalAgentId,
|
||||
replayState: false,
|
||||
});
|
||||
|
||||
sessionHolder.current?.pushEvent({
|
||||
type: "provider_subagent",
|
||||
provider: "codex",
|
||||
event: { type: "upsert", id: "hidden-child", title: "Hidden child", status: "running" },
|
||||
});
|
||||
await manager.flush();
|
||||
|
||||
expect(globalEvents.filter((event) => event.type === "provider_subagent")).toEqual([]);
|
||||
expect(scopedEvents).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "provider_subagent",
|
||||
event: expect.objectContaining({
|
||||
type: "upsert",
|
||||
subagent: expect.objectContaining({
|
||||
id: "hidden-child",
|
||||
parentAgentId: internalAgentId,
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
);
|
||||
expect(() => manager.listProviderSubagents(internalAgentId)).toThrow(
|
||||
`Unknown agent '${internalAgentId}'`,
|
||||
);
|
||||
expect(() => manager.getProviderSubagent(internalAgentId, "hidden-child")).toThrow(
|
||||
`Unknown agent '${internalAgentId}'`,
|
||||
);
|
||||
expect(() => manager.fetchProviderSubagentTimeline(internalAgentId, "hidden-child")).toThrow(
|
||||
`Unknown agent '${internalAgentId}'`,
|
||||
);
|
||||
});
|
||||
|
||||
test("subscribe emits state events for internal agents when subscribed by agentId", async () => {
|
||||
const internalAgentId = "00000000-0000-4000-8000-000000000110";
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-"));
|
||||
|
||||
@@ -66,6 +66,11 @@ import { isSystemInjectedEnvelope } from "./agent-prompt.js";
|
||||
import { stripInternalPaseoMcpServer, withRuntimePaseoMcpServer } from "./runtime-mcp-config.js";
|
||||
import { resolveCreateAgentTitles } from "./create-agent-title.js";
|
||||
import type { PaseoToolCatalogFactory } from "./tools/types.js";
|
||||
import {
|
||||
ProviderSubagentStore,
|
||||
type ProviderSubagentDescriptor,
|
||||
type ProviderSubagentStoreEvent,
|
||||
} from "./provider-subagents/store.js";
|
||||
|
||||
const RELOAD_SESSION_CLOSE_TIMEOUT_MS = 3_000;
|
||||
const INTERRUPT_SESSION_TIMEOUT_MS = 2_000;
|
||||
@@ -145,6 +150,7 @@ export type {
|
||||
|
||||
export type AgentManagerEvent =
|
||||
| { type: "agent_state"; agent: ManagedAgent }
|
||||
| { type: "provider_subagent"; event: ProviderSubagentStoreEvent }
|
||||
| {
|
||||
type: "agent_stream";
|
||||
agentId: string;
|
||||
@@ -534,6 +540,7 @@ export class AgentManager {
|
||||
private readonly providerEnabled = new Map<AgentProvider, boolean>();
|
||||
private readonly agents = new Map<string, LiveManagedAgent>();
|
||||
private readonly timelineStore = new InMemoryAgentTimelineStore();
|
||||
private readonly providerSubagents = new ProviderSubagentStore();
|
||||
private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
|
||||
private readonly sessionEventTails = new Map<string, Promise<void>>();
|
||||
private readonly foregroundRuns = new ForegroundRunState();
|
||||
@@ -936,6 +943,28 @@ export class AgentManager {
|
||||
return this.timelineStore.fetch(id, options);
|
||||
}
|
||||
|
||||
listProviderSubagents(parentAgentId: string): ProviderSubagentDescriptor[] {
|
||||
this.requirePublicAgent(parentAgentId);
|
||||
return this.providerSubagents.list(parentAgentId);
|
||||
}
|
||||
|
||||
getProviderSubagent(
|
||||
parentAgentId: string,
|
||||
subagentId: string,
|
||||
): ProviderSubagentDescriptor | null {
|
||||
this.requirePublicAgent(parentAgentId);
|
||||
return this.providerSubagents.get(parentAgentId, subagentId);
|
||||
}
|
||||
|
||||
fetchProviderSubagentTimeline(
|
||||
parentAgentId: string,
|
||||
subagentId: string,
|
||||
options?: AgentTimelineFetchOptions,
|
||||
): AgentTimelineFetchResult {
|
||||
this.requirePublicAgent(parentAgentId);
|
||||
return this.providerSubagents.fetchTimeline(parentAgentId, subagentId, options);
|
||||
}
|
||||
|
||||
createAgent(
|
||||
config: AgentSessionConfig,
|
||||
agentId: string | undefined,
|
||||
@@ -1086,7 +1115,7 @@ export class AgentManager {
|
||||
const initialTitle = resolveImportedAgentTitle(importedConfig, timelineRows);
|
||||
|
||||
handedToRegistration = true;
|
||||
return this.registerSession(imported.session, importedConfig, resolvedAgentId, {
|
||||
const agent = await this.registerSession(imported.session, importedConfig, resolvedAgentId, {
|
||||
labels: input.labels,
|
||||
workspaceId: input.workspaceId,
|
||||
timelineRows,
|
||||
@@ -1096,6 +1125,11 @@ export class AgentManager {
|
||||
initialTitle,
|
||||
publishWhenReady: true,
|
||||
});
|
||||
for (const event of imported.providerSubagentEvents ?? []) {
|
||||
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
|
||||
this.dispatch({ type: "provider_subagent", event: update });
|
||||
}
|
||||
return agent;
|
||||
} finally {
|
||||
if (!handedToRegistration) {
|
||||
await this.closeUnregisteredSession(imported.session);
|
||||
@@ -1168,6 +1202,9 @@ export class AgentManager {
|
||||
// provider history into an empty timeline.
|
||||
await this.deleteCommittedTimeline(agentId);
|
||||
this.timelineStore.delete(agentId);
|
||||
for (const event of this.providerSubagents.deleteParent(agentId)) {
|
||||
this.dispatch({ type: "provider_subagent", event });
|
||||
}
|
||||
}
|
||||
|
||||
// Preserve existing labels and timeline during reload.
|
||||
@@ -1261,6 +1298,9 @@ export class AgentManager {
|
||||
const closedAgent = this.prepareAgentForClosure(agent, "agent closed");
|
||||
await agent.session.close();
|
||||
this.timelineStore.delete(agentId);
|
||||
for (const event of this.providerSubagents.deleteParent(agentId)) {
|
||||
this.dispatch({ type: "provider_subagent", event });
|
||||
}
|
||||
await this.persistSnapshot(closedAgent);
|
||||
this.emitClosedAgent(closedAgent, { persist: false });
|
||||
this.logger.trace(
|
||||
@@ -2837,6 +2877,11 @@ export class AgentManager {
|
||||
agent: ActiveManagedAgent,
|
||||
event: AgentStreamEvent,
|
||||
): Promise<void> {
|
||||
if (event.type === "provider_subagent") {
|
||||
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
|
||||
this.dispatch({ type: "provider_subagent", event: update });
|
||||
return;
|
||||
}
|
||||
const turnId = getAgentStreamEventTurnId(event);
|
||||
const matchingWaiters = this.foregroundRuns.getMatchingWaiters(agent, turnId);
|
||||
this.logger.trace(
|
||||
@@ -2975,49 +3020,79 @@ export class AgentManager {
|
||||
}
|
||||
|
||||
if (options?.force) {
|
||||
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
|
||||
for await (const event of agent.session.streamHistory()) {
|
||||
if (event.type === "timeline") {
|
||||
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
|
||||
continue;
|
||||
}
|
||||
historyEvents.push(event);
|
||||
}
|
||||
}
|
||||
|
||||
this.agentStreamCoalescer.flushAndDiscard(agent.id);
|
||||
await this.deleteCommittedTimeline(agent.id);
|
||||
this.timelineStore.delete(agent.id);
|
||||
this.timelineStore.initialize(agent.id, { timestamp: new Date().toISOString() });
|
||||
agent.historyPrimed = true;
|
||||
|
||||
for (const event of historyEvents) {
|
||||
const item = limitAgentTimelineItemContent(event.item);
|
||||
const row = this.recordTimeline(
|
||||
agent.id,
|
||||
item,
|
||||
event.timestamp ? { timestamp: event.timestamp } : undefined,
|
||||
);
|
||||
if (options?.broadcast) {
|
||||
this.dispatchStream(
|
||||
agent.id,
|
||||
{ ...event, item },
|
||||
{
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
this.touchUpdatedAt(agent);
|
||||
this.emitState(agent);
|
||||
await this.forceHydrateTimelineFromLegacyProviderHistory(agent, options.broadcast === true);
|
||||
return;
|
||||
}
|
||||
|
||||
await this.primeTimelineFromLegacyProviderHistory(agent, options?.broadcast === true);
|
||||
}
|
||||
|
||||
private async forceHydrateTimelineFromLegacyProviderHistory(
|
||||
agent: ActiveManagedAgent,
|
||||
broadcast: boolean,
|
||||
): Promise<void> {
|
||||
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
|
||||
const providerSubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
|
||||
for await (const event of agent.session.streamHistory()) {
|
||||
if (event.type === "timeline") {
|
||||
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
|
||||
continue;
|
||||
}
|
||||
historyEvents.push(event);
|
||||
} else if (event.type === "provider_subagent") {
|
||||
providerSubagentEvents.push(event);
|
||||
}
|
||||
}
|
||||
|
||||
this.agentStreamCoalescer.flushAndDiscard(agent.id);
|
||||
await this.deleteCommittedTimeline(agent.id);
|
||||
this.timelineStore.delete(agent.id);
|
||||
this.timelineStore.initialize(agent.id, { timestamp: new Date().toISOString() });
|
||||
agent.historyPrimed = true;
|
||||
|
||||
for (const event of this.providerSubagents.deleteParent(agent.id)) {
|
||||
if (broadcast) {
|
||||
this.dispatch({ type: "provider_subagent", event });
|
||||
}
|
||||
}
|
||||
for (const event of providerSubagentEvents) {
|
||||
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
|
||||
if (broadcast) {
|
||||
this.dispatch({ type: "provider_subagent", event: update });
|
||||
}
|
||||
}
|
||||
for (const event of historyEvents) {
|
||||
const row = this.recordTimeline(
|
||||
agent.id,
|
||||
event.item,
|
||||
event.timestamp ? { timestamp: event.timestamp } : undefined,
|
||||
);
|
||||
if (broadcast) {
|
||||
this.dispatchStream(agent.id, event, {
|
||||
seq: row.seq,
|
||||
epoch: this.timelineStore.getEpoch(agent.id),
|
||||
timestamp: row.timestamp,
|
||||
});
|
||||
}
|
||||
}
|
||||
this.touchUpdatedAt(agent);
|
||||
this.emitState(agent);
|
||||
}
|
||||
|
||||
private async primeTimelineFromLegacyProviderHistory(
|
||||
agent: ActiveManagedAgent,
|
||||
broadcast: boolean,
|
||||
): Promise<void> {
|
||||
agent.historyPrimed = true;
|
||||
try {
|
||||
for await (const event of agent.session.streamHistory()) {
|
||||
if (event.type === "provider_subagent") {
|
||||
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
|
||||
if (broadcast) {
|
||||
this.dispatch({ type: "provider_subagent", event: update });
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (event.type !== "timeline") {
|
||||
continue;
|
||||
}
|
||||
@@ -3793,22 +3868,35 @@ export class AgentManager {
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
if (
|
||||
subscriber.agentId &&
|
||||
event.type === "provider_subagent" &&
|
||||
subscriber.agentId !==
|
||||
(event.event.type === "upsert"
|
||||
? event.event.subagent.parentAgentId
|
||||
: event.event.parentAgentId)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
// Skip internal agents for global subscribers (those without a specific agentId)
|
||||
if (!subscriber.agentId) {
|
||||
if (event.type === "agent_state" && event.agent.internal) {
|
||||
continue;
|
||||
}
|
||||
if (event.type === "agent_stream") {
|
||||
const agent = this.agents.get(event.agentId);
|
||||
if (agent?.internal) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (!subscriber.agentId && this.eventBelongsToInternalAgent(event)) {
|
||||
continue;
|
||||
}
|
||||
subscriber.callback(event);
|
||||
}
|
||||
}
|
||||
|
||||
private eventBelongsToInternalAgent(event: AgentManagerEvent): boolean {
|
||||
if (event.type === "agent_state") return event.agent.internal === true;
|
||||
if (event.type === "agent_stream") return this.agents.get(event.agentId)?.internal === true;
|
||||
if (event.type !== "provider_subagent") return false;
|
||||
const parentAgentId =
|
||||
event.event.type === "upsert"
|
||||
? event.event.subagent.parentAgentId
|
||||
: event.event.parentAgentId;
|
||||
return this.agents.get(parentAgentId)?.internal === true;
|
||||
}
|
||||
|
||||
private async normalizeConfig(
|
||||
config: AgentSessionConfig,
|
||||
options: NormalizeConfigOptions = {},
|
||||
@@ -4031,6 +4119,14 @@ export class AgentManager {
|
||||
}
|
||||
return agent;
|
||||
}
|
||||
|
||||
private requirePublicAgent(id: string): LiveManagedAgent {
|
||||
const agent = this.requireAgent(id);
|
||||
if (agent.internal) {
|
||||
throw new Error(`Unknown agent '${agent.id}'`);
|
||||
}
|
||||
return agent;
|
||||
}
|
||||
}
|
||||
|
||||
export function commandMayHaveChangedExternalState(command: string): boolean {
|
||||
|
||||
@@ -426,6 +426,11 @@ export type AgentStreamEvent =
|
||||
provider: AgentProvider;
|
||||
reason: "finished" | "error" | "permission";
|
||||
timestamp: string;
|
||||
}
|
||||
| {
|
||||
type: "provider_subagent";
|
||||
provider: AgentProvider;
|
||||
event: import("./provider-subagents/store.js").ProviderSubagentInputEvent;
|
||||
};
|
||||
|
||||
export function getAgentStreamEventTurnId(event: AgentStreamEvent): string | undefined {
|
||||
@@ -541,6 +546,7 @@ export interface ImportedProviderSession {
|
||||
config: AgentSessionConfig;
|
||||
persistence: AgentPersistenceHandle;
|
||||
timeline: ImportedTimelineEntry[];
|
||||
providerSubagentEvents?: Extract<AgentStreamEvent, { type: "provider_subagent" }>[];
|
||||
}
|
||||
|
||||
export interface AgentSessionConfig {
|
||||
|
||||
@@ -33,13 +33,14 @@ export async function importSessionFromPersistence(input: {
|
||||
const persistence =
|
||||
input.persistence ?? buildImportPersistenceHandle(input.provider, input.request, storedConfig);
|
||||
const session = await input.resumeSession(persistence, config, input.context.launchContext);
|
||||
const timeline = await collectImportedTimeline(session.streamHistory());
|
||||
const history = await collectImportedHistory(session.streamHistory());
|
||||
|
||||
return {
|
||||
session,
|
||||
config: storedConfig,
|
||||
persistence,
|
||||
timeline,
|
||||
timeline: history.timeline,
|
||||
providerSubagentEvents: history.providerSubagentEvents,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -60,11 +61,17 @@ function buildImportPersistenceHandle(
|
||||
};
|
||||
}
|
||||
|
||||
async function collectImportedTimeline(
|
||||
events: AsyncGenerator<AgentStreamEvent>,
|
||||
): Promise<ImportedTimelineEntry[]> {
|
||||
async function collectImportedHistory(events: AsyncGenerator<AgentStreamEvent>): Promise<{
|
||||
timeline: ImportedTimelineEntry[];
|
||||
providerSubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[];
|
||||
}> {
|
||||
const timeline: ImportedTimelineEntry[] = [];
|
||||
const providerSubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
|
||||
for await (const event of events) {
|
||||
if (event.type === "provider_subagent") {
|
||||
providerSubagentEvents.push(event);
|
||||
continue;
|
||||
}
|
||||
if (event.type !== "timeline") {
|
||||
continue;
|
||||
}
|
||||
@@ -73,5 +80,5 @@ async function collectImportedTimeline(
|
||||
...(event.timestamp ? { timestamp: event.timestamp } : {}),
|
||||
});
|
||||
}
|
||||
return timeline;
|
||||
return { timeline, providerSubagentEvents };
|
||||
}
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { ProviderSubagentStore } from "./store.js";
|
||||
|
||||
describe("ProviderSubagentStore", () => {
|
||||
test("keeps provider children and their timelines scoped to the parent agent", () => {
|
||||
const subagents = new ProviderSubagentStore();
|
||||
|
||||
subagents.apply("parent-a", "codex", {
|
||||
type: "upsert",
|
||||
id: "child-1",
|
||||
title: "Explore",
|
||||
cwd: "/workspace/child",
|
||||
status: "running",
|
||||
timestamp: "2026-07-12T10:00:00.000Z",
|
||||
});
|
||||
subagents.apply("parent-a", "codex", {
|
||||
type: "timeline",
|
||||
id: "child-1",
|
||||
item: { type: "assistant_message", text: "Found it." },
|
||||
timestamp: "2026-07-12T10:00:01.000Z",
|
||||
});
|
||||
subagents.apply("parent-a", "codex", {
|
||||
type: "upsert",
|
||||
id: "child-1",
|
||||
status: "completed",
|
||||
timestamp: "2026-07-12T10:00:02.000Z",
|
||||
});
|
||||
subagents.apply("parent-b", "claude", {
|
||||
type: "upsert",
|
||||
id: "child-1",
|
||||
title: "Review",
|
||||
status: "running",
|
||||
timestamp: "2026-07-12T10:00:03.000Z",
|
||||
});
|
||||
|
||||
expect(subagents.list("parent-a")).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "child-1",
|
||||
parentAgentId: "parent-a",
|
||||
provider: "codex",
|
||||
title: "Explore",
|
||||
cwd: "/workspace/child",
|
||||
status: "completed",
|
||||
createdAt: "2026-07-12T10:00:00.000Z",
|
||||
updatedAt: "2026-07-12T10:00:02.000Z",
|
||||
}),
|
||||
]);
|
||||
expect(subagents.fetchTimeline("parent-a", "child-1").rows).toEqual([
|
||||
{
|
||||
seq: 1,
|
||||
timestamp: "2026-07-12T10:00:01.000Z",
|
||||
item: { type: "assistant_message", text: "Found it." },
|
||||
},
|
||||
]);
|
||||
expect(subagents.list("parent-b")[0]).toMatchObject({ provider: "claude", title: "Review" });
|
||||
expect(subagents.deleteParent("parent-a")).toEqual([
|
||||
{ type: "remove", parentAgentId: "parent-a", subagentId: "child-1" },
|
||||
]);
|
||||
expect(subagents.list("parent-a")).toEqual([]);
|
||||
expect(subagents.list("parent-b")).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("limits oversized provider child tool output before storage", () => {
|
||||
const subagents = new ProviderSubagentStore();
|
||||
const output = "x".repeat(70 * 1024);
|
||||
const update = subagents.apply("parent-a", "opencode", {
|
||||
type: "timeline",
|
||||
id: "child-1",
|
||||
item: {
|
||||
type: "tool_call",
|
||||
callId: "call-1",
|
||||
name: "shell",
|
||||
status: "completed",
|
||||
error: null,
|
||||
detail: { type: "shell", command: "print", output },
|
||||
},
|
||||
});
|
||||
|
||||
expect(update.type).toBe("timeline");
|
||||
const [row] = subagents.fetchTimeline("parent-a", "child-1").rows;
|
||||
expect(row?.item).toMatchObject({
|
||||
type: "tool_call",
|
||||
detail: { type: "shell", output: "x".repeat(64 * 1024) },
|
||||
});
|
||||
});
|
||||
|
||||
test("pages provider history on projected item boundaries", () => {
|
||||
const subagents = new ProviderSubagentStore();
|
||||
for (let index = 0; index < 101; index += 1) {
|
||||
subagents.apply("parent-a", "opencode", {
|
||||
type: "timeline",
|
||||
id: "child-1",
|
||||
item: { type: "assistant_message", text: String(index) },
|
||||
});
|
||||
}
|
||||
|
||||
const page = subagents.fetchTimeline("parent-a", "child-1", {
|
||||
direction: "tail",
|
||||
limit: 1,
|
||||
});
|
||||
expect(page.rows).toHaveLength(101);
|
||||
expect(page.rows[0]?.seq).toBe(1);
|
||||
expect(page.rows.at(-1)?.seq).toBe(101);
|
||||
expect(page.hasOlder).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,168 @@
|
||||
import type { AgentProvider, AgentTimelineItem } from "../agent-sdk-types.js";
|
||||
import { limitAgentTimelineItemContent } from "../agent-timeline-content.js";
|
||||
import { InMemoryAgentTimelineStore } from "../agent-timeline-store.js";
|
||||
import type {
|
||||
AgentTimelineFetchOptions,
|
||||
AgentTimelineFetchResult,
|
||||
AgentTimelineRow,
|
||||
} from "../agent-timeline-store-types.js";
|
||||
import { selectTimelineWindowByProjectedLimit } from "../timeline-projection.js";
|
||||
|
||||
export type ProviderSubagentStatus = "running" | "completed" | "failed" | "canceled";
|
||||
|
||||
export interface ProviderSubagentDescriptor {
|
||||
id: string;
|
||||
parentAgentId: string;
|
||||
provider: AgentProvider;
|
||||
title: string | null;
|
||||
description: string | null;
|
||||
status: ProviderSubagentStatus;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
toolCallId: string | null;
|
||||
cwd: string | null;
|
||||
}
|
||||
|
||||
export type ProviderSubagentInputEvent =
|
||||
| {
|
||||
type: "upsert";
|
||||
id: string;
|
||||
title?: string | null;
|
||||
description?: string | null;
|
||||
status: ProviderSubagentStatus;
|
||||
toolCallId?: string | null;
|
||||
cwd?: string | null;
|
||||
timestamp?: string;
|
||||
}
|
||||
| {
|
||||
type: "timeline";
|
||||
id: string;
|
||||
item: AgentTimelineItem;
|
||||
timestamp?: string;
|
||||
}
|
||||
| { type: "remove"; id: string };
|
||||
|
||||
export type ProviderSubagentStoreEvent =
|
||||
| { type: "upsert"; subagent: ProviderSubagentDescriptor }
|
||||
| {
|
||||
type: "timeline";
|
||||
parentAgentId: string;
|
||||
subagentId: string;
|
||||
provider: AgentProvider;
|
||||
row: AgentTimelineRow;
|
||||
epoch: string;
|
||||
}
|
||||
| { type: "remove"; parentAgentId: string; subagentId: string };
|
||||
|
||||
function storeKey(parentAgentId: string, subagentId: string): string {
|
||||
return `${parentAgentId}\0${subagentId}`;
|
||||
}
|
||||
|
||||
export class ProviderSubagentStore {
|
||||
private readonly descriptors = new Map<string, ProviderSubagentDescriptor>();
|
||||
private readonly timelines = new InMemoryAgentTimelineStore();
|
||||
|
||||
apply(
|
||||
parentAgentId: string,
|
||||
provider: AgentProvider,
|
||||
event: ProviderSubagentInputEvent,
|
||||
): ProviderSubagentStoreEvent {
|
||||
const key = storeKey(parentAgentId, event.id);
|
||||
if (event.type === "remove") {
|
||||
this.descriptors.delete(key);
|
||||
this.timelines.delete(key);
|
||||
return { type: "remove", parentAgentId, subagentId: event.id };
|
||||
}
|
||||
|
||||
if (event.type === "timeline") {
|
||||
if (!this.timelines.has(key)) {
|
||||
this.timelines.initialize(key);
|
||||
}
|
||||
const row = this.timelines.append(key, limitAgentTimelineItemContent(event.item), {
|
||||
timestamp: event.timestamp,
|
||||
});
|
||||
return {
|
||||
type: "timeline",
|
||||
parentAgentId,
|
||||
subagentId: event.id,
|
||||
provider,
|
||||
row,
|
||||
epoch: this.timelines.getEpoch(key),
|
||||
};
|
||||
}
|
||||
|
||||
const previous = this.descriptors.get(key);
|
||||
if (!this.timelines.has(key)) {
|
||||
this.timelines.initialize(key);
|
||||
}
|
||||
const timestamp = event.timestamp ?? new Date().toISOString();
|
||||
const subagent: ProviderSubagentDescriptor = {
|
||||
id: event.id,
|
||||
parentAgentId,
|
||||
provider,
|
||||
title: event.title === undefined ? (previous?.title ?? null) : event.title,
|
||||
description:
|
||||
event.description === undefined ? (previous?.description ?? null) : event.description,
|
||||
status: event.status,
|
||||
createdAt: previous?.createdAt ?? timestamp,
|
||||
updatedAt: timestamp,
|
||||
toolCallId:
|
||||
event.toolCallId === undefined ? (previous?.toolCallId ?? null) : event.toolCallId,
|
||||
cwd: event.cwd === undefined ? (previous?.cwd ?? null) : event.cwd,
|
||||
};
|
||||
this.descriptors.set(key, subagent);
|
||||
return { type: "upsert", subagent };
|
||||
}
|
||||
|
||||
list(parentAgentId: string): ProviderSubagentDescriptor[] {
|
||||
return [...this.descriptors.values()]
|
||||
.filter((subagent) => subagent.parentAgentId === parentAgentId)
|
||||
.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
|
||||
}
|
||||
|
||||
get(parentAgentId: string, subagentId: string): ProviderSubagentDescriptor | null {
|
||||
return this.descriptors.get(storeKey(parentAgentId, subagentId)) ?? null;
|
||||
}
|
||||
|
||||
fetchTimeline(
|
||||
parentAgentId: string,
|
||||
subagentId: string,
|
||||
options?: AgentTimelineFetchOptions,
|
||||
): AgentTimelineFetchResult {
|
||||
const direction = options?.direction ?? "tail";
|
||||
const limit = options?.limit === undefined ? 200 : Math.max(0, Math.floor(options.limit));
|
||||
const timeline = this.timelines.fetch(storeKey(parentAgentId, subagentId), {
|
||||
...options,
|
||||
limit: 0,
|
||||
});
|
||||
if (limit === 0 || timeline.rows.length === 0) {
|
||||
return timeline;
|
||||
}
|
||||
const selected = selectTimelineWindowByProjectedLimit({
|
||||
rows: timeline.rows,
|
||||
direction: timeline.reset ? "tail" : direction,
|
||||
limit,
|
||||
});
|
||||
const firstRow = selected.selectedRows[0];
|
||||
const lastRow = selected.selectedRows[selected.selectedRows.length - 1];
|
||||
return {
|
||||
...timeline,
|
||||
rows: selected.selectedRows,
|
||||
hasOlder:
|
||||
timeline.hasOlder || (firstRow !== undefined && firstRow.seq > timeline.window.minSeq),
|
||||
hasNewer:
|
||||
timeline.hasNewer || (lastRow !== undefined && lastRow.seq < timeline.window.maxSeq),
|
||||
};
|
||||
}
|
||||
|
||||
deleteParent(parentAgentId: string): ProviderSubagentStoreEvent[] {
|
||||
const events: ProviderSubagentStoreEvent[] = [];
|
||||
for (const subagent of this.list(parentAgentId)) {
|
||||
const key = storeKey(parentAgentId, subagent.id);
|
||||
this.descriptors.delete(key);
|
||||
this.timelines.delete(key);
|
||||
events.push({ type: "remove", parentAgentId, subagentId: subagent.id });
|
||||
}
|
||||
return events;
|
||||
}
|
||||
}
|
||||
@@ -211,6 +211,21 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
parent_tool_use_id: "task-call-1",
|
||||
elapsed_time_seconds: 1,
|
||||
},
|
||||
{
|
||||
type: "user",
|
||||
parent_tool_use_id: "task-call-1",
|
||||
message: {
|
||||
content: [
|
||||
{
|
||||
type: "tool_result",
|
||||
tool_use_id: "sub-read-1",
|
||||
tool_name: "Read",
|
||||
content: "README contents",
|
||||
is_error: false,
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "assistant",
|
||||
parent_tool_use_id: "task-call-1",
|
||||
@@ -307,6 +322,44 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
);
|
||||
|
||||
expect(projectedTaskCalls).toHaveLength(1);
|
||||
|
||||
const providerEvents = events.flatMap((event) =>
|
||||
event.type === "provider_subagent" ? [event.event] : [],
|
||||
);
|
||||
expect(providerEvents).toContainEqual({
|
||||
type: "timeline",
|
||||
id: "task-call-1",
|
||||
item: expect.objectContaining({
|
||||
type: "tool_call",
|
||||
callId: "sub-read-1",
|
||||
status: "running",
|
||||
}),
|
||||
});
|
||||
expect(providerEvents).toContainEqual({
|
||||
type: "timeline",
|
||||
id: "task-call-1",
|
||||
item: expect.objectContaining({
|
||||
type: "tool_call",
|
||||
callId: "sub-read-1",
|
||||
status: "completed",
|
||||
}),
|
||||
});
|
||||
expect(providerEvents).toContainEqual({
|
||||
type: "timeline",
|
||||
id: "task-call-1",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "subagent-message-1",
|
||||
text: "Sub-agent narration belongs inside the Task row, not the parent transcript.",
|
||||
},
|
||||
});
|
||||
expect(providerEvents.at(-1)).toMatchObject({
|
||||
type: "upsert",
|
||||
id: "task-call-1",
|
||||
title: "Explore",
|
||||
description: "Inspect repository structure",
|
||||
status: "completed",
|
||||
});
|
||||
});
|
||||
|
||||
test("keeps sidechain assistant text out of the parent transcript", async () => {
|
||||
@@ -349,6 +402,46 @@ describe("ClaudeAgentSession sub-agent sidechain updates", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("keeps a failed Task subagent failed when the parent turn succeeds", async () => {
|
||||
const failedEvents = buildTailScenarioEvents(1);
|
||||
const taskResult = failedEvents.find(
|
||||
(event) =>
|
||||
typeof event === "object" &&
|
||||
event !== null &&
|
||||
"type" in event &&
|
||||
event.type === "assistant",
|
||||
) as { message: Record<string, unknown> } | undefined;
|
||||
if (!taskResult) throw new Error("expected Task result fixture");
|
||||
taskResult.message = {
|
||||
...taskResult.message,
|
||||
content: [
|
||||
{
|
||||
type: "tool_result",
|
||||
tool_use_id: "task-tail-1",
|
||||
tool_name: "Task",
|
||||
content: "failed",
|
||||
is_error: true,
|
||||
},
|
||||
],
|
||||
};
|
||||
queryFactory.mockImplementation(() => buildQueryMock(failedEvents));
|
||||
const session = await new ClaudeAgentClient({
|
||||
logger,
|
||||
queryFactory,
|
||||
resolveBinary: async () => "/test/claude/bin",
|
||||
}).createSession({ provider: "claude", cwd: process.cwd() });
|
||||
|
||||
const events = await collectUntilTerminal(streamSession(session, "delegate work"));
|
||||
await session.close();
|
||||
|
||||
expect(
|
||||
events
|
||||
.filter((event) => event.type === "provider_subagent")
|
||||
.map((event) => event.event)
|
||||
.at(-1),
|
||||
).toMatchObject({ type: "upsert", id: "task-tail-1", status: "failed" });
|
||||
});
|
||||
|
||||
test("tails sub-agent actions instead of dropping latest entries at cap", async () => {
|
||||
queryFactory.mockImplementation(() => buildQueryMock(buildTailScenarioEvents(205)));
|
||||
|
||||
|
||||
@@ -1897,6 +1897,10 @@ class ClaudeAgentSession implements AgentSession {
|
||||
getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null,
|
||||
});
|
||||
private persistedHistory: PersistedTimelineEntry[] = [];
|
||||
private persistedProviderSubagentEvents: Extract<
|
||||
AgentStreamEvent,
|
||||
{ type: "provider_subagent" }
|
||||
>[] = [];
|
||||
private historyPending = false;
|
||||
private turnState: TurnState = "idle";
|
||||
private nextTurnOrdinal = 1;
|
||||
@@ -2117,11 +2121,16 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
if (!this.historyPending || this.persistedHistory.length === 0) {
|
||||
if (
|
||||
!this.historyPending ||
|
||||
(this.persistedHistory.length === 0 && this.persistedProviderSubagentEvents.length === 0)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const history = this.persistedHistory;
|
||||
const providerSubagentEvents = this.persistedProviderSubagentEvents;
|
||||
this.persistedHistory = [];
|
||||
this.persistedProviderSubagentEvents = [];
|
||||
this.historyPending = false;
|
||||
for (const entry of history) {
|
||||
yield {
|
||||
@@ -2131,6 +2140,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
timestamp: entry.timestamp,
|
||||
};
|
||||
}
|
||||
yield* providerSubagentEvents;
|
||||
}
|
||||
|
||||
async getAvailableModes(): Promise<AgentMode[]> {
|
||||
@@ -2613,6 +2623,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
this.cachedRuntimeInfo = null;
|
||||
this.queryRestartNeeded = true;
|
||||
this.persistedHistory = [];
|
||||
this.persistedProviderSubagentEvents = [];
|
||||
this.historyPending = false;
|
||||
this.userMessageIds = [];
|
||||
this.emittedUserMessageIds.clear();
|
||||
@@ -2642,6 +2653,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
this.cachedRuntimeInfo = null;
|
||||
this.queryRestartNeeded = true;
|
||||
this.persistedHistory = [];
|
||||
this.persistedProviderSubagentEvents = [];
|
||||
this.historyPending = false;
|
||||
this.userMessageIds = [];
|
||||
this.emittedUserMessageIds.clear();
|
||||
@@ -3539,6 +3551,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
this.persistence = null;
|
||||
this.persistedHistory = [];
|
||||
this.persistedProviderSubagentEvents = [];
|
||||
this.historyPending = false;
|
||||
this.cachedRuntimeInfo = null;
|
||||
this.queryRestartNeeded = false;
|
||||
@@ -3610,6 +3623,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
break;
|
||||
case "user":
|
||||
this.appendUserMessageEvents(message, events);
|
||||
this.appendSidechainResultEvents(message, events);
|
||||
break;
|
||||
case "assistant": {
|
||||
const timelineItems = this.mapBlocksToTimeline(message.message.content, {
|
||||
@@ -3619,6 +3633,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
for (const item of timelineItems) {
|
||||
events.push({ type: "timeline", item, provider: "claude" });
|
||||
}
|
||||
this.appendSidechainResultEvents(message, events);
|
||||
break;
|
||||
}
|
||||
case "stream_event":
|
||||
@@ -3634,6 +3649,18 @@ class ClaudeAgentSession implements AgentSession {
|
||||
return events;
|
||||
}
|
||||
|
||||
private appendSidechainResultEvents(message: SDKMessage, events: AgentStreamEvent[]): void {
|
||||
const content = toObjectRecord(toObjectRecord(message)?.message)?.content;
|
||||
if (!Array.isArray(content)) return;
|
||||
for (const block of content) {
|
||||
const chunk = toObjectRecord(block);
|
||||
if (chunk?.type !== "tool_result" || typeof chunk.tool_use_id !== "string") continue;
|
||||
events.push(
|
||||
...this.sidechainTracker.finish(chunk.tool_use_id, chunk.is_error ? "failed" : "completed"),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private emitSubmittedUserMessage(
|
||||
message: Extract<SDKMessage, { type: "user" }>,
|
||||
turnId: string,
|
||||
@@ -3830,6 +3857,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
): void {
|
||||
const usage = this.convertUsage(message, message.modelUsage);
|
||||
if (message.subtype === "success") {
|
||||
events.push(...this.sidechainTracker.finishAll("completed"));
|
||||
// Built-in slash commands (e.g. /voice, /usage, "Unknown command: …")
|
||||
// run client-side in the Claude CLI with no model turn — output_tokens
|
||||
// is 0 and the user-visible text is carried in `result`. Surface it only
|
||||
@@ -3855,6 +3883,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
"errors" in message && Array.isArray(message.errors) && message.errors.length > 0
|
||||
? message.errors.join("\n")
|
||||
: "Claude run failed";
|
||||
events.push(...this.sidechainTracker.finishAll("failed"));
|
||||
events.push(this.buildTurnFailedEvent(errorMessage));
|
||||
}
|
||||
|
||||
@@ -4166,7 +4195,9 @@ class ClaudeAgentSession implements AgentSession {
|
||||
if (!historyPath || !fs.existsSync(historyPath)) {
|
||||
return;
|
||||
}
|
||||
this.ingestPersistedHistory(fs.readFileSync(historyPath, "utf8"));
|
||||
const content = fs.readFileSync(historyPath, "utf8");
|
||||
this.ingestPersistedHistory(content);
|
||||
this.ingestPersistedSidechains(content, readClaudeSidechainHistory(historyPath));
|
||||
} catch {
|
||||
// ignore history load failures
|
||||
}
|
||||
@@ -4188,6 +4219,24 @@ class ClaudeAgentSession implements AgentSession {
|
||||
}
|
||||
}
|
||||
|
||||
private ingestPersistedSidechains(parentContent: string, sidechainContents: string[]): void {
|
||||
const parentEntries = parseClaudeHistoryRecords(parentContent).filter(
|
||||
(entry) => entry.isSidechain !== true,
|
||||
);
|
||||
const sidechainEntries = [parentContent, ...sidechainContents]
|
||||
.flatMap(parseClaudeHistoryRecords)
|
||||
.filter((entry) => entry.isSidechain === true && typeof entry.agentId === "string");
|
||||
if (sidechainEntries.length === 0) {
|
||||
return;
|
||||
}
|
||||
this.persistedProviderSubagentEvents.push(
|
||||
...buildClaudePersistedSidechainEvents(parentEntries, sidechainEntries, (entry) =>
|
||||
this.convertHistoryEntry(entry),
|
||||
),
|
||||
);
|
||||
this.historyPending = true;
|
||||
}
|
||||
|
||||
private ingestPersistedHistoryLine(line: string, timeline: PersistedTimelineEntry[]): void {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) {
|
||||
@@ -4465,7 +4514,6 @@ class ClaudeAgentSession implements AgentSession {
|
||||
|
||||
if (typeof block.tool_use_id === "string") {
|
||||
this.toolUseCache.delete(block.tool_use_id);
|
||||
this.sidechainTracker.delete(block.tool_use_id);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4947,11 +4995,194 @@ function normalizeHistoryBlocks(content: unknown): ClaudeContentChunk[] | null {
|
||||
return null;
|
||||
}
|
||||
|
||||
function parseClaudeHistoryRecords(content: string): ClaudeHistoryEntry[] {
|
||||
const entries: ClaudeHistoryEntry[] = [];
|
||||
for (const line of content.split(/\r?\n/)) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
try {
|
||||
const entry = toObjectRecord(JSON.parse(trimmed));
|
||||
if (entry) entries.push(entry);
|
||||
} catch {
|
||||
// Ignore individual corrupt history rows, matching the parent history replay behavior.
|
||||
}
|
||||
}
|
||||
return entries;
|
||||
}
|
||||
|
||||
function readClaudeSidechainHistory(historyPath: string): string[] {
|
||||
const sessionDirectory = path.join(
|
||||
path.dirname(historyPath),
|
||||
path.basename(historyPath, ".jsonl"),
|
||||
);
|
||||
const sidechainDirectory = path.join(sessionDirectory, "subagents");
|
||||
if (!fs.existsSync(sidechainDirectory)) return [];
|
||||
|
||||
const contents: string[] = [];
|
||||
const directories = [sidechainDirectory];
|
||||
while (directories.length > 0) {
|
||||
const directory = directories.pop();
|
||||
if (!directory) continue;
|
||||
for (const entry of fs.readdirSync(directory, { withFileTypes: true })) {
|
||||
const entryPath = path.join(directory, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
directories.push(entryPath);
|
||||
} else if (entry.isFile() && entry.name.endsWith(".jsonl")) {
|
||||
contents.push(fs.readFileSync(entryPath, "utf8"));
|
||||
}
|
||||
}
|
||||
}
|
||||
return contents;
|
||||
}
|
||||
|
||||
interface ClaudeHistoricalSubagentToolCall {
|
||||
subagentType?: string;
|
||||
description?: string;
|
||||
}
|
||||
|
||||
function readClaudeHistoricalSubagentToolCalls(
|
||||
entries: ClaudeHistoryEntry[],
|
||||
): Map<string, ClaudeHistoricalSubagentToolCall> {
|
||||
const toolCalls = new Map<string, ClaudeHistoricalSubagentToolCall>();
|
||||
for (const entry of entries) {
|
||||
const content = toObjectRecord(entry.message)?.content;
|
||||
if (!Array.isArray(content)) continue;
|
||||
for (const value of content) {
|
||||
const block = toObjectRecord(value);
|
||||
if (
|
||||
block?.type !== "tool_use" ||
|
||||
(block.name !== "Task" && block.name !== "Agent") ||
|
||||
typeof block.id !== "string"
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const input = toObjectRecord(block.input);
|
||||
const subagentType = readNonEmptyString(input?.subagent_type);
|
||||
const description = readNonEmptyString(input?.description);
|
||||
toolCalls.set(block.id, {
|
||||
...(subagentType ? { subagentType } : {}),
|
||||
...(description ? { description } : {}),
|
||||
});
|
||||
}
|
||||
}
|
||||
return toolCalls;
|
||||
}
|
||||
|
||||
function readClaudeHistoricalSubagentToolResults(
|
||||
entries: ClaudeHistoryEntry[],
|
||||
): Map<string, { toolCallId: string; failed: boolean }> {
|
||||
const results = new Map<string, { toolCallId: string; failed: boolean }>();
|
||||
for (const entry of entries) {
|
||||
const content = toObjectRecord(entry.message)?.content;
|
||||
if (!Array.isArray(content)) continue;
|
||||
for (const value of content) {
|
||||
const block = toObjectRecord(value);
|
||||
if (block?.type !== "tool_result" || typeof block.tool_use_id !== "string") continue;
|
||||
const match = /agentId:\s*([\w-]+)/.exec(JSON.stringify(block.content));
|
||||
if (!match?.[1]) continue;
|
||||
results.set(match[1], { toolCallId: block.tool_use_id, failed: block.is_error === true });
|
||||
}
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
function groupClaudeSidechainEntries(
|
||||
entries: ClaudeHistoryEntry[],
|
||||
): Map<string, ClaudeHistoryEntry[]> {
|
||||
const entriesByAgentId = new Map<string, ClaudeHistoryEntry[]>();
|
||||
for (const entry of entries) {
|
||||
if (typeof entry.agentId !== "string") continue;
|
||||
const grouped = entriesByAgentId.get(entry.agentId) ?? [];
|
||||
grouped.push(entry);
|
||||
entriesByAgentId.set(entry.agentId, grouped);
|
||||
}
|
||||
return entriesByAgentId;
|
||||
}
|
||||
|
||||
function buildClaudePersistedSidechainEvents(
|
||||
parentEntries: ClaudeHistoryEntry[],
|
||||
sidechainEntries: ClaudeHistoryEntry[],
|
||||
convertEntry: (entry: ClaudeHistoryEntry) => AgentTimelineItem[],
|
||||
): Extract<AgentStreamEvent, { type: "provider_subagent" }>[] {
|
||||
const events: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
|
||||
const toolCalls = readClaudeHistoricalSubagentToolCalls(parentEntries);
|
||||
const toolResults = readClaudeHistoricalSubagentToolResults(parentEntries);
|
||||
for (const [agentId, entries] of groupClaudeSidechainEntries(sidechainEntries)) {
|
||||
events.push(
|
||||
...buildClaudePersistedSidechainAgentEvents(
|
||||
agentId,
|
||||
entries,
|
||||
toolCalls,
|
||||
toolResults,
|
||||
convertEntry,
|
||||
),
|
||||
);
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
function buildClaudePersistedSidechainAgentEvents(
|
||||
agentId: string,
|
||||
entries: ClaudeHistoryEntry[],
|
||||
toolCalls: ReadonlyMap<string, ClaudeHistoricalSubagentToolCall>,
|
||||
toolResults: ReadonlyMap<string, { toolCallId: string; failed: boolean }>,
|
||||
convertEntry: (entry: ClaudeHistoryEntry) => AgentTimelineItem[],
|
||||
): Extract<AgentStreamEvent, { type: "provider_subagent" }>[] {
|
||||
const result = toolResults.get(agentId);
|
||||
const id = result?.toolCallId ?? agentId;
|
||||
const toolCall = result ? toolCalls.get(result.toolCallId) : undefined;
|
||||
const firstTimestamp = normalizeProviderReplayTimestamp(entries[0]?.timestamp);
|
||||
const events: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [
|
||||
{
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "upsert",
|
||||
id,
|
||||
title: toolCall?.subagentType ?? "Claude subagent",
|
||||
description: toolCall?.description ?? null,
|
||||
status: "running",
|
||||
toolCallId: result?.toolCallId ?? null,
|
||||
...(firstTimestamp ? { timestamp: firstTimestamp } : {}),
|
||||
},
|
||||
},
|
||||
];
|
||||
for (const entry of entries) {
|
||||
const timestamp = normalizeProviderReplayTimestamp(entry.timestamp);
|
||||
for (const item of convertEntry(entry)) {
|
||||
events.push({
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "timeline",
|
||||
id,
|
||||
item,
|
||||
...(timestamp ? { timestamp } : {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
const lastTimestamp = normalizeProviderReplayTimestamp(entries.at(-1)?.timestamp);
|
||||
events.push({
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "upsert",
|
||||
id,
|
||||
status: result?.failed ? "failed" : "completed",
|
||||
...(lastTimestamp ? { timestamp: lastTimestamp } : {}),
|
||||
},
|
||||
});
|
||||
return events;
|
||||
}
|
||||
|
||||
interface ClaudeHistoryEntry {
|
||||
type?: unknown;
|
||||
subtype?: unknown;
|
||||
isCompactSummary?: unknown;
|
||||
isSidechain?: unknown;
|
||||
agentId?: unknown;
|
||||
timestamp?: unknown;
|
||||
uuid?: unknown;
|
||||
message?: { content?: unknown; [key: string]: unknown };
|
||||
[key: string]: unknown;
|
||||
|
||||
+95
@@ -15,6 +15,7 @@ let lastQuery: ReturnType<typeof buildSdkQueryMock> | null = null;
|
||||
const LIVE_REPLY_MARKER = "LIVE_ONLY_REPLY_MARKER";
|
||||
const HISTORY_USER_MARKER = "HISTORY_ONLY_USER_MARKER";
|
||||
const HISTORY_ASSISTANT_MARKER = "HISTORY_ONLY_ASSISTANT_MARKER";
|
||||
const HISTORY_SIDECHAIN_MARKER = "HISTORY_ONLY_SIDECHAIN_MARKER";
|
||||
|
||||
function buildSdkQueryMock() {
|
||||
const events = [
|
||||
@@ -126,6 +127,52 @@ describe("ClaudeAgentSession history replay regression", () => {
|
||||
content: HISTORY_ASSISTANT_MARKER,
|
||||
},
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "assistant",
|
||||
uuid: "history-task-call-message",
|
||||
sessionId: "history-session",
|
||||
cwd,
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [
|
||||
{
|
||||
type: "tool_use",
|
||||
id: "history-task-call",
|
||||
name: "Agent",
|
||||
input: { description: "Inspect persisted history" },
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "assistant",
|
||||
isSidechain: true,
|
||||
agentId: "history-child",
|
||||
uuid: "history-child-message",
|
||||
timestamp: "2026-07-12T10:00:01.000Z",
|
||||
sessionId: "history-session",
|
||||
cwd,
|
||||
message: {
|
||||
id: "history-child-message",
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: HISTORY_SIDECHAIN_MARKER }],
|
||||
},
|
||||
}),
|
||||
JSON.stringify({
|
||||
type: "user",
|
||||
sessionId: "history-session",
|
||||
cwd,
|
||||
message: {
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
type: "tool_result",
|
||||
tool_use_id: "history-task-call",
|
||||
content: "done\nagentId: history-child",
|
||||
},
|
||||
],
|
||||
},
|
||||
}),
|
||||
].join("\n"),
|
||||
"utf8",
|
||||
);
|
||||
@@ -219,6 +266,54 @@ describe("ClaudeAgentSession history replay regression", () => {
|
||||
expect(timelineText).toContain(HISTORY_ASSISTANT_MARKER);
|
||||
});
|
||||
|
||||
test("replays persisted sidechains as provider subagent timelines", async () => {
|
||||
const client = new ClaudeAgentClient({
|
||||
logger: createTestLogger(),
|
||||
queryFactory,
|
||||
resolveBinary: async () => "/test/claude/bin",
|
||||
});
|
||||
const session = await client.resumeSession(
|
||||
{
|
||||
provider: "claude",
|
||||
sessionId: "history-session",
|
||||
nativeHandle: "history-session",
|
||||
metadata: { provider: "claude", cwd },
|
||||
},
|
||||
{ cwd },
|
||||
);
|
||||
const historyEvents: AgentStreamEvent[] = [];
|
||||
|
||||
try {
|
||||
for await (const event of session.streamHistory()) historyEvents.push(event);
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
|
||||
expect(historyEvents).toContainEqual({
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "timeline",
|
||||
id: "history-task-call",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
text: HISTORY_SIDECHAIN_MARKER,
|
||||
messageId: "history-child-message",
|
||||
},
|
||||
timestamp: "2026-07-12T10:00:01.000Z",
|
||||
},
|
||||
});
|
||||
expect(historyEvents).toContainEqual({
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: expect.objectContaining({
|
||||
type: "upsert",
|
||||
id: "history-task-call",
|
||||
status: "completed",
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
test("listCommands includes rewind command", async () => {
|
||||
const logger = createTestLogger();
|
||||
const client = new ClaudeAgentClient({
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
|
||||
|
||||
import { mapClaudeRunningToolCall } from "./tool-call-mapper.js";
|
||||
import {
|
||||
mapClaudeCompletedToolCall,
|
||||
mapClaudeFailedToolCall,
|
||||
mapClaudeRunningToolCall,
|
||||
} from "./tool-call-mapper.js";
|
||||
import { buildToolCallDisplayModel } from "@getpaseo/protocol/tool-call-display";
|
||||
|
||||
import type { AgentMetadata, AgentStreamEvent, AgentTimelineItem } from "../../agent-sdk-types.js";
|
||||
@@ -13,6 +17,7 @@ interface ClaudeContentChunk {
|
||||
interface SubAgentActionEntry {
|
||||
index: number;
|
||||
toolName: string;
|
||||
input: unknown;
|
||||
summary?: string;
|
||||
}
|
||||
|
||||
@@ -23,6 +28,7 @@ interface SubAgentActivityState {
|
||||
actionKeys: string[];
|
||||
nextActionIndex: number;
|
||||
actionIndexByKey: Map<string, number>;
|
||||
completedActionKeys: Set<string>;
|
||||
}
|
||||
|
||||
interface SubAgentActionCandidate {
|
||||
@@ -64,19 +70,34 @@ export class ClaudeSidechainTracker {
|
||||
actionKeys: [],
|
||||
nextActionIndex: 1,
|
||||
actionIndexByKey: new Map<string, number>(),
|
||||
completedActionKeys: new Set<string>(),
|
||||
} satisfies SubAgentActivityState);
|
||||
this.activeSidechains.set(parentToolUseId, state);
|
||||
|
||||
const contextUpdated = this.updateSubAgentContextFromTaskInput(state, parentToolUseId);
|
||||
const actionCandidates = this.extractSubAgentActionCandidates(message);
|
||||
const childTimelineItems = [
|
||||
...this.extractSubAgentTimelineItems(message),
|
||||
...this.extractSubAgentToolResults(message, state),
|
||||
];
|
||||
let actionUpdated = false;
|
||||
for (const action of actionCandidates) {
|
||||
if (state.completedActionKeys.has(action.key)) continue;
|
||||
if (this.appendSubAgentAction(state, action)) {
|
||||
actionUpdated = true;
|
||||
const toolCall = mapClaudeRunningToolCall({
|
||||
name: action.toolName,
|
||||
callId: action.key,
|
||||
input: action.input,
|
||||
output: null,
|
||||
});
|
||||
if (toolCall) {
|
||||
childTimelineItems.push(toolCall);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!contextUpdated && !actionUpdated) {
|
||||
if (!contextUpdated && !actionUpdated && childTimelineItems.length === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
@@ -103,6 +124,25 @@ export class ClaudeSidechainTracker {
|
||||
};
|
||||
|
||||
return [
|
||||
{
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "upsert",
|
||||
id: parentToolUseId,
|
||||
title: state.subAgentType ?? "Claude subagent",
|
||||
description: state.description ?? null,
|
||||
status: "running",
|
||||
toolCallId: parentToolUseId,
|
||||
},
|
||||
},
|
||||
...childTimelineItems.map(
|
||||
(item): AgentStreamEvent => ({
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: { type: "timeline", id: parentToolUseId, item },
|
||||
}),
|
||||
),
|
||||
{
|
||||
type: "timeline",
|
||||
item: {
|
||||
@@ -114,6 +154,46 @@ export class ClaudeSidechainTracker {
|
||||
];
|
||||
}
|
||||
|
||||
finishAll(status: "completed" | "failed" | "canceled"): AgentStreamEvent[] {
|
||||
const events: AgentStreamEvent[] = [];
|
||||
for (const [id, state] of this.activeSidechains) {
|
||||
events.push({
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "upsert",
|
||||
id,
|
||||
title: state.subAgentType ?? "Claude subagent",
|
||||
description: state.description ?? null,
|
||||
status,
|
||||
toolCallId: id,
|
||||
},
|
||||
});
|
||||
}
|
||||
this.activeSidechains.clear();
|
||||
return events;
|
||||
}
|
||||
|
||||
finish(id: string, status: "completed" | "failed" | "canceled"): AgentStreamEvent[] {
|
||||
const state = this.activeSidechains.get(id);
|
||||
if (!state) return [];
|
||||
this.activeSidechains.delete(id);
|
||||
return [
|
||||
{
|
||||
type: "provider_subagent",
|
||||
provider: "claude",
|
||||
event: {
|
||||
type: "upsert",
|
||||
id,
|
||||
title: state.subAgentType ?? "Claude subagent",
|
||||
description: state.description ?? null,
|
||||
status,
|
||||
toolCallId: id,
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
delete(toolUseId: string): void {
|
||||
this.activeSidechains.delete(toolUseId);
|
||||
}
|
||||
@@ -122,6 +202,66 @@ export class ClaudeSidechainTracker {
|
||||
this.activeSidechains.clear();
|
||||
}
|
||||
|
||||
private extractSubAgentTimelineItems(message: SDKMessage): AgentTimelineItem[] {
|
||||
if (message.type !== "assistant" || !Array.isArray(message.message?.content)) {
|
||||
return [];
|
||||
}
|
||||
const messageId = readTrimmedString(message.message.id);
|
||||
const items: AgentTimelineItem[] = [];
|
||||
for (const block of message.message.content) {
|
||||
if (!isClaudeContentChunk(block)) continue;
|
||||
if (block.type === "text") {
|
||||
const text = readTrimmedString(block.text);
|
||||
if (text) {
|
||||
items.push({
|
||||
type: "assistant_message",
|
||||
text,
|
||||
...(messageId ? { messageId } : {}),
|
||||
});
|
||||
}
|
||||
} else if (block.type === "thinking") {
|
||||
const text = readTrimmedString(block.thinking);
|
||||
if (text) items.push({ type: "reasoning", text });
|
||||
}
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
private extractSubAgentToolResults(
|
||||
message: SDKMessage,
|
||||
state: SubAgentActivityState,
|
||||
): AgentTimelineItem[] {
|
||||
const messageRecord = message as unknown as Record<string, unknown>;
|
||||
const messageContainer = messageRecord.message as Record<string, unknown> | undefined;
|
||||
const content = messageContainer?.content;
|
||||
if (!Array.isArray(content)) return [];
|
||||
|
||||
const items: AgentTimelineItem[] = [];
|
||||
for (const block of content) {
|
||||
if (!isClaudeContentChunk(block) || !block.type.endsWith("tool_result")) continue;
|
||||
const callId = readTrimmedString(block.tool_use_id);
|
||||
if (!callId || state.completedActionKeys.has(callId)) continue;
|
||||
const actionIndex = state.actionIndexByKey.get(callId);
|
||||
const action = actionIndex === undefined ? undefined : state.actions[actionIndex];
|
||||
const toolName = action?.toolName ?? readTrimmedString(block.tool_name);
|
||||
if (!toolName) continue;
|
||||
const params = {
|
||||
name: toolName,
|
||||
callId,
|
||||
input: action?.input ?? null,
|
||||
output: block.content ?? null,
|
||||
};
|
||||
const toolCall = block.is_error
|
||||
? mapClaudeFailedToolCall({ ...params, error: block })
|
||||
: mapClaudeCompletedToolCall(params);
|
||||
if (toolCall) {
|
||||
state.completedActionKeys.add(callId);
|
||||
items.push(toolCall);
|
||||
}
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
private updateSubAgentContextFromTaskInput(
|
||||
state: SubAgentActivityState,
|
||||
parentToolUseId: string,
|
||||
@@ -259,6 +399,7 @@ export class ClaudeSidechainTracker {
|
||||
state.actions[existingIndex] = {
|
||||
...existing,
|
||||
toolName: normalizedToolName,
|
||||
input: existing.input ?? candidate.input,
|
||||
...(nextSummary ? { summary: nextSummary } : {}),
|
||||
};
|
||||
return true;
|
||||
@@ -267,6 +408,7 @@ export class ClaudeSidechainTracker {
|
||||
state.actions.push({
|
||||
index: state.nextActionIndex,
|
||||
toolName: normalizedToolName,
|
||||
input: candidate.input,
|
||||
...(summary ? { summary } : {}),
|
||||
});
|
||||
state.nextActionIndex += 1;
|
||||
@@ -279,7 +421,8 @@ export class ClaudeSidechainTracker {
|
||||
private trimSubAgentTail(state: SubAgentActivityState): void {
|
||||
while (state.actions.length > MAX_SUB_AGENT_LOG_ENTRIES) {
|
||||
state.actions.shift();
|
||||
state.actionKeys.shift();
|
||||
const removedKey = state.actionKeys.shift();
|
||||
if (removedKey) state.completedActionKeys.delete(removedKey);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1567,7 +1567,7 @@ describe("Codex app-server provider", () => {
|
||||
asInternals(session).handleNotification("item/agentMessage/delta", {
|
||||
threadId: "child-thread-1",
|
||||
itemId: "child-message-1",
|
||||
delta: "Found the path.",
|
||||
delta: "Found",
|
||||
});
|
||||
asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "child-thread-1",
|
||||
@@ -1599,6 +1599,128 @@ describe("Codex app-server provider", () => {
|
||||
actions: [],
|
||||
},
|
||||
});
|
||||
|
||||
const providerEvents = events.flatMap((event) =>
|
||||
event.type === "provider_subagent" ? [event.event] : [],
|
||||
);
|
||||
expect(providerEvents).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "upsert",
|
||||
id: "child-thread-1",
|
||||
description: "Report findings.",
|
||||
}),
|
||||
);
|
||||
expect(providerEvents).toContainEqual({
|
||||
type: "timeline",
|
||||
id: "child-thread-1",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "child-message-1",
|
||||
text: "Found",
|
||||
},
|
||||
});
|
||||
expect(providerEvents).toContainEqual({
|
||||
type: "timeline",
|
||||
id: "child-thread-1",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "child-message-1",
|
||||
text: " the path.",
|
||||
},
|
||||
});
|
||||
expect(providerEvents.at(-1)).toMatchObject({
|
||||
type: "upsert",
|
||||
id: "child-thread-1",
|
||||
status: "completed",
|
||||
});
|
||||
});
|
||||
|
||||
test("renders child MCP image results in the provider subagent timeline", () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "test-thread",
|
||||
item: {
|
||||
type: "subAgentActivity",
|
||||
id: "spawn-image-child",
|
||||
kind: "started",
|
||||
agentThreadId: "image-child-thread",
|
||||
agentPath: "/root/image-child",
|
||||
},
|
||||
});
|
||||
|
||||
asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "image-child-thread",
|
||||
item: {
|
||||
id: "child-mcp-image",
|
||||
type: "mcpToolCall",
|
||||
status: "completed",
|
||||
server: "paseo",
|
||||
tool: "browser_screenshot",
|
||||
arguments: {},
|
||||
result: {
|
||||
content: [{ type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" }],
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const childItems = events.flatMap((event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.id === "image-child-thread"
|
||||
? [event.event.item]
|
||||
: [],
|
||||
);
|
||||
expect(childItems).toHaveLength(2);
|
||||
expect(childItems[0]).toMatchObject({ type: "tool_call", callId: "child-mcp-image" });
|
||||
expect(childItems[1]).toMatchObject({ type: "assistant_message" });
|
||||
if (childItems[1]?.type !== "assistant_message") {
|
||||
throw new Error("Expected child image markdown");
|
||||
}
|
||||
const source = markdownImageSource(childItems[1].text);
|
||||
expect(existsSync(source)).toBe(true);
|
||||
rmSync(source, { force: true });
|
||||
});
|
||||
|
||||
test("renders a child user message once across lifecycle notifications", () => {
|
||||
const session = createSession();
|
||||
const events: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => events.push(event));
|
||||
asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "test-thread",
|
||||
item: {
|
||||
type: "subAgentActivity",
|
||||
id: "spawn-user-child",
|
||||
kind: "started",
|
||||
agentThreadId: "user-child-thread",
|
||||
agentPath: "/root/user-child",
|
||||
},
|
||||
});
|
||||
const childUserMessage = {
|
||||
type: "userMessage",
|
||||
id: "child-user-message",
|
||||
content: [{ type: "text", text: "Inspect this path." }],
|
||||
};
|
||||
|
||||
asInternals(session).handleNotification("item/started", {
|
||||
threadId: "user-child-thread",
|
||||
item: childUserMessage,
|
||||
});
|
||||
asInternals(session).handleNotification("item/completed", {
|
||||
threadId: "user-child-thread",
|
||||
item: childUserMessage,
|
||||
});
|
||||
|
||||
expect(
|
||||
events.filter(
|
||||
(event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.id === "user-child-thread" &&
|
||||
event.event.item.type === "user_message",
|
||||
),
|
||||
).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("keeps the parent running when a MultiAgentV2 sub-agent finishes", async () => {
|
||||
@@ -2095,6 +2217,20 @@ describe("Codex app-server provider", () => {
|
||||
event.item.callId === "child-command",
|
||||
),
|
||||
).toBe(false);
|
||||
expect(events).toContainEqual(
|
||||
expect.objectContaining({
|
||||
type: "provider_subagent",
|
||||
event: {
|
||||
type: "timeline",
|
||||
id: "legacy-envelope-child",
|
||||
item: expect.objectContaining({
|
||||
type: "tool_call",
|
||||
callId: "child-command",
|
||||
status: "running",
|
||||
}),
|
||||
},
|
||||
}),
|
||||
);
|
||||
expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(0);
|
||||
expect(events.at(-1)).toMatchObject({
|
||||
type: "timeline",
|
||||
@@ -2343,10 +2479,28 @@ describe("Codex app-server provider", () => {
|
||||
test("loads mixed legacy and MultiAgentV2 sub-agent history", async () => {
|
||||
const session = createSession();
|
||||
session.client = {
|
||||
request: vi.fn(async (method: string) => {
|
||||
request: vi.fn(async (method: string, params: unknown) => {
|
||||
if (method !== "thread/read") {
|
||||
return {};
|
||||
}
|
||||
const threadId = (params as { threadId?: string }).threadId;
|
||||
if (threadId !== "test-thread") {
|
||||
return {
|
||||
thread: {
|
||||
turns: [
|
||||
{
|
||||
items: [
|
||||
{
|
||||
type: "agentMessage",
|
||||
id: `message-${threadId}`,
|
||||
text: `History from ${threadId}`,
|
||||
},
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
};
|
||||
}
|
||||
return {
|
||||
thread: {
|
||||
turns: [
|
||||
@@ -2382,6 +2536,38 @@ describe("Codex app-server provider", () => {
|
||||
for await (const event of session.streamHistory()) {
|
||||
history.push(event);
|
||||
}
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
|
||||
),
|
||||
).toMatchObject([
|
||||
{ type: "upsert", id: "legacy-child-thread", status: "completed" },
|
||||
{ type: "upsert", id: "v2-child-thread", status: "completed" },
|
||||
]);
|
||||
expect(
|
||||
history.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "timeline" ? [event.event] : [],
|
||||
),
|
||||
).toEqual([
|
||||
{
|
||||
type: "timeline",
|
||||
id: "legacy-child-thread",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "message-legacy-child-thread",
|
||||
text: "History from legacy-child-thread",
|
||||
},
|
||||
},
|
||||
{
|
||||
type: "timeline",
|
||||
id: "v2-child-thread",
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
messageId: "message-v2-child-thread",
|
||||
text: "History from v2-child-thread",
|
||||
},
|
||||
},
|
||||
]);
|
||||
expect(
|
||||
history
|
||||
.filter((event) => event.type === "timeline" && event.item.type === "tool_call")
|
||||
@@ -2447,10 +2633,13 @@ describe("Codex app-server provider", () => {
|
||||
test("coalesces persisted MultiAgentV2 activity for one child into one terminal card", async () => {
|
||||
const session = createSession();
|
||||
session.client = {
|
||||
request: vi.fn(async (method: string) => {
|
||||
request: vi.fn(async (method: string, params: unknown) => {
|
||||
if (method !== "thread/read") {
|
||||
return {};
|
||||
}
|
||||
if ((params as { threadId?: string }).threadId !== "test-thread") {
|
||||
return { thread: { turns: [] } };
|
||||
}
|
||||
return {
|
||||
thread: {
|
||||
turns: [
|
||||
@@ -2495,6 +2684,15 @@ describe("Codex app-server provider", () => {
|
||||
history.push(event);
|
||||
}
|
||||
expect(history).toEqual([
|
||||
{
|
||||
type: "provider_subagent",
|
||||
provider: "codex",
|
||||
event: expect.objectContaining({
|
||||
type: "upsert",
|
||||
id: "history-child-thread",
|
||||
status: "canceled",
|
||||
}),
|
||||
},
|
||||
{
|
||||
type: "timeline",
|
||||
provider: "codex",
|
||||
|
||||
@@ -3061,6 +3061,7 @@ interface CodexSubAgentCallState {
|
||||
pendingFileChangeOutputDeltas: Map<string, string[]>;
|
||||
childItemOrder: string[];
|
||||
childItems: Map<string, AgentTimelineItem>;
|
||||
childThreadIds: Set<string>;
|
||||
}
|
||||
|
||||
export class CodexAppServerAgentSession implements AgentSession {
|
||||
@@ -3081,6 +3082,8 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
private planModeEnabled = false;
|
||||
private historyPending = false;
|
||||
private persistedHistory: PersistedTimelineEntry[] = [];
|
||||
private loadingPersistedHistory = false;
|
||||
private persistedProviderSubagentEvents: AgentStreamEvent[] = [];
|
||||
private pendingPermissions = new Map<string, AgentPermissionRequest>();
|
||||
private mcpElicitationPermissionIds = new Map<number, string>();
|
||||
private pendingPermissionHandlers = new Map<
|
||||
@@ -3105,6 +3108,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
private emittedExecCommandCompletedCallIds = new Set<string>();
|
||||
private emittedItemStartedIds = new Set<string>();
|
||||
private emittedItemCompletedIds = new Set<string>();
|
||||
private emittedProviderSubagentUserMessageKeys = new Set<string>();
|
||||
private subAgentCallsByCallId = new Map<string, CodexSubAgentCallState>();
|
||||
private subAgentCallIdByChildThreadId = new Map<string, string>();
|
||||
private pendingSubAgentNotificationsByThreadId = new Map<string, ParsedCodexNotification[]>();
|
||||
@@ -3454,12 +3458,12 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.subAgentCallsByCallId.clear();
|
||||
this.subAgentCallIdByChildThreadId.clear();
|
||||
this.pendingSubAgentNotificationsByThreadId.clear();
|
||||
for (const route of subAgentRoutes) {
|
||||
this.registerSubAgentToolCall({
|
||||
timelineItem: route.toolCall,
|
||||
rawItem: { agentThreadId: route.childThreadId },
|
||||
parentCallId: null,
|
||||
});
|
||||
this.persistedProviderSubagentEvents = [];
|
||||
this.loadingPersistedHistory = true;
|
||||
try {
|
||||
await this.loadPersistedSubAgentHistories(client, subAgentRoutes);
|
||||
} finally {
|
||||
this.loadingPersistedHistory = false;
|
||||
}
|
||||
this.resetCodexUserMessageTurns();
|
||||
for (const entry of timeline) {
|
||||
@@ -3471,6 +3475,44 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.historyPending = timeline.length > 0;
|
||||
}
|
||||
|
||||
private async loadPersistedSubAgentHistories(
|
||||
client: CodexAppServerClientLike,
|
||||
rootRoutes: readonly PersistedSubAgentRoute[],
|
||||
): Promise<void> {
|
||||
const queue = rootRoutes.map((route) => ({ route, parentCallId: null as string | null }));
|
||||
const visitedThreadIds = new Set<string>();
|
||||
while (queue.length > 0 && visitedThreadIds.size < 100) {
|
||||
const next = queue.shift();
|
||||
if (!next || visitedThreadIds.has(next.route.childThreadId)) {
|
||||
continue;
|
||||
}
|
||||
visitedThreadIds.add(next.route.childThreadId);
|
||||
this.registerSubAgentToolCall({
|
||||
timelineItem: next.route.toolCall,
|
||||
rawItem: { agentThreadId: next.route.childThreadId },
|
||||
parentCallId: next.parentCallId,
|
||||
});
|
||||
try {
|
||||
const childHistory = await loadCodexThreadHistoryTimeline({
|
||||
threadId: next.route.childThreadId,
|
||||
cwd: this.config.cwd ?? null,
|
||||
requestThread: (childThreadId) => readCodexThread(client, childThreadId),
|
||||
});
|
||||
for (const entry of childHistory.timeline) {
|
||||
this.emitProviderSubagentTimeline(next.route.childThreadId, entry.item, entry.timestamp);
|
||||
}
|
||||
for (const route of childHistory.subAgentRoutes) {
|
||||
queue.push({ route, parentCallId: next.route.toolCall.callId });
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.trace(
|
||||
{ err: error, childThreadId: next.route.childThreadId },
|
||||
"Failed to load persisted Codex child history",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async ensureThreadLoaded(): Promise<void> {
|
||||
if (!this.client || !this.currentThreadId) return;
|
||||
try {
|
||||
@@ -3818,12 +3860,20 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
|
||||
if (!this.historyPending || this.persistedHistory.length === 0) {
|
||||
if (
|
||||
(!this.historyPending || this.persistedHistory.length === 0) &&
|
||||
this.persistedProviderSubagentEvents.length === 0
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const history = this.persistedHistory;
|
||||
const providerSubagents = this.persistedProviderSubagentEvents;
|
||||
this.persistedHistory = [];
|
||||
this.persistedProviderSubagentEvents = [];
|
||||
this.historyPending = false;
|
||||
for (const event of providerSubagents) {
|
||||
yield event;
|
||||
}
|
||||
for (const entry of history) {
|
||||
yield {
|
||||
type: "timeline",
|
||||
@@ -4454,6 +4504,10 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private emitEvent(event: AgentStreamEvent): void {
|
||||
if (this.loadingPersistedHistory && event.type === "provider_subagent") {
|
||||
this.persistedProviderSubagentEvents.push(event);
|
||||
return;
|
||||
}
|
||||
this.notifySubscribers(event);
|
||||
}
|
||||
|
||||
@@ -4724,6 +4778,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
pendingFileChangeOutputDeltas: new Map<string, string[]>(),
|
||||
childItemOrder: [],
|
||||
childItems: new Map<string, AgentTimelineItem>(),
|
||||
childThreadIds: new Set<string>(),
|
||||
} satisfies CodexSubAgentCallState);
|
||||
|
||||
state.toolCall = {
|
||||
@@ -4754,6 +4809,8 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
);
|
||||
for (const receiverThreadId of childThreadIds) {
|
||||
this.subAgentCallIdByChildThreadId.set(receiverThreadId, timelineItem.callId);
|
||||
state.childThreadIds.add(receiverThreadId);
|
||||
this.emitProviderSubagentUpsert(receiverThreadId, state, timelineItem.status);
|
||||
}
|
||||
return childThreadIds;
|
||||
}
|
||||
@@ -4835,12 +4892,20 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
private emitCodexToolTimelineItem(
|
||||
timelineItem: ToolCallTimelineItem,
|
||||
subAgentCallId: string | null,
|
||||
childThreadId?: string | null,
|
||||
): void {
|
||||
if (!subAgentCallId) {
|
||||
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
|
||||
return;
|
||||
}
|
||||
this.upsertSubAgentChildItem(subAgentCallId, timelineItem.callId, timelineItem);
|
||||
const state = this.subAgentCallsByCallId.get(subAgentCallId);
|
||||
if (state) {
|
||||
const targetThreadIds = childThreadId ? [childThreadId] : state.childThreadIds;
|
||||
for (const targetThreadId of targetThreadIds) {
|
||||
this.emitProviderSubagentTimeline(targetThreadId, timelineItem);
|
||||
}
|
||||
}
|
||||
this.emitSubAgentActivityUpdate(
|
||||
subAgentCallId,
|
||||
timelineItem.status === "running" ? "running" : undefined,
|
||||
@@ -4867,6 +4932,9 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
? curateAgentActivity(childTimeline, { labelAssistantMessages: true })
|
||||
: "";
|
||||
const resolvedStatus = status ?? state.toolCall.status;
|
||||
for (const childThreadId of state.childThreadIds) {
|
||||
this.emitProviderSubagentUpsert(childThreadId, state, resolvedStatus);
|
||||
}
|
||||
const baseToolCall = {
|
||||
...state.toolCall,
|
||||
detail: {
|
||||
@@ -4895,6 +4963,100 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: nextToolCall });
|
||||
}
|
||||
|
||||
private emitProviderSubagentUpsert(
|
||||
childThreadId: string,
|
||||
state: CodexSubAgentCallState,
|
||||
status: ToolCallTimelineItem["status"],
|
||||
): void {
|
||||
const detail = state.toolCall.detail;
|
||||
if (detail.type !== "sub_agent") {
|
||||
return;
|
||||
}
|
||||
let providerStatus: "running" | "completed" | "failed" | "canceled" = "running";
|
||||
if (status === "completed") {
|
||||
providerStatus = "completed";
|
||||
} else if (status === "failed") {
|
||||
providerStatus = "failed";
|
||||
} else if (status === "canceled") {
|
||||
providerStatus = "canceled";
|
||||
}
|
||||
this.emitEvent({
|
||||
type: "provider_subagent",
|
||||
provider: CODEX_PROVIDER,
|
||||
event: {
|
||||
type: "upsert",
|
||||
id: childThreadId,
|
||||
title: detail.subAgentType ?? "Codex subagent",
|
||||
description: detail.description ?? null,
|
||||
status: providerStatus,
|
||||
toolCallId: state.callId,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private emitProviderSubagentTimeline(
|
||||
childThreadId: string,
|
||||
item: AgentTimelineItem,
|
||||
timestamp?: string,
|
||||
): void {
|
||||
this.emitEvent({
|
||||
type: "provider_subagent",
|
||||
provider: CODEX_PROVIDER,
|
||||
event: {
|
||||
type: "timeline",
|
||||
id: childThreadId,
|
||||
item,
|
||||
...(timestamp ? { timestamp } : {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private emitCompletedProviderSubagentItem(
|
||||
parsed: Extract<ParsedCodexNotification, { kind: "item_completed" }>,
|
||||
timelineItem: AgentTimelineItem,
|
||||
): void {
|
||||
const itemId = parsed.item.id;
|
||||
if (!parsed.threadId) return;
|
||||
if (timelineItem.type === "assistant_message" && itemId) {
|
||||
const streamedText = this.pendingAgentMessages.get(itemId);
|
||||
if (streamedText !== undefined) {
|
||||
const suffix = this.buildMissingFinalTextSuffix(timelineItem, streamedText);
|
||||
if (suffix) this.emitProviderSubagentTimeline(parsed.threadId, suffix);
|
||||
return;
|
||||
}
|
||||
}
|
||||
if (timelineItem.type === "reasoning" && itemId) {
|
||||
const streamedText = this.pendingReasoning.get(itemId)?.join("");
|
||||
if (streamedText !== undefined) {
|
||||
const suffix = this.buildMissingFinalTextSuffix(timelineItem, streamedText);
|
||||
if (suffix) this.emitProviderSubagentTimeline(parsed.threadId, suffix);
|
||||
return;
|
||||
}
|
||||
}
|
||||
this.emitProviderSubagentTimeline(parsed.threadId, timelineItem);
|
||||
}
|
||||
|
||||
private emitStartedProviderSubagentItem(
|
||||
threadId: string | null,
|
||||
timelineItem: AgentTimelineItem,
|
||||
): void {
|
||||
if (threadId) {
|
||||
this.emitProviderSubagentTimeline(threadId, timelineItem);
|
||||
}
|
||||
}
|
||||
|
||||
private emitProviderSubagentTimelineItems(
|
||||
threadId: string | null,
|
||||
timelineItems: readonly AgentTimelineItem[],
|
||||
): void {
|
||||
if (!threadId) {
|
||||
return;
|
||||
}
|
||||
for (const timelineItem of timelineItems) {
|
||||
this.emitProviderSubagentTimeline(threadId, timelineItem);
|
||||
}
|
||||
}
|
||||
|
||||
private handleSubAgentChildItemCompleted(
|
||||
callId: string,
|
||||
itemId: string | undefined,
|
||||
@@ -4949,6 +5111,13 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.pendingAgentMessages.set(parsed.itemId, text);
|
||||
const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
|
||||
if (subAgentCallId) {
|
||||
if (parsed.threadId) {
|
||||
this.emitProviderSubagentTimeline(parsed.threadId, {
|
||||
type: "assistant_message",
|
||||
messageId: parsed.itemId,
|
||||
text: parsed.delta,
|
||||
});
|
||||
}
|
||||
this.upsertSubAgentChildItem(subAgentCallId, parsed.itemId, {
|
||||
type: "assistant_message",
|
||||
messageId: parsed.itemId,
|
||||
@@ -4981,6 +5150,12 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.pendingReasoning.set(parsed.itemId, prev);
|
||||
const subAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
|
||||
if (subAgentCallId) {
|
||||
if (parsed.threadId) {
|
||||
this.emitProviderSubagentTimeline(parsed.threadId, {
|
||||
type: "reasoning",
|
||||
text: parsed.delta,
|
||||
});
|
||||
}
|
||||
this.upsertSubAgentChildItem(subAgentCallId, parsed.itemId, {
|
||||
type: "reasoning",
|
||||
text: prev.join(""),
|
||||
@@ -5080,6 +5255,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
this.latestPlanResult = null;
|
||||
this.emittedItemStartedIds.clear();
|
||||
this.emittedItemCompletedIds.clear();
|
||||
this.emittedProviderSubagentUserMessageKeys.clear();
|
||||
this.emittedExecCommandStartedCallIds.clear();
|
||||
this.emittedExecCommandCompletedCallIds.clear();
|
||||
this.pendingAgentMessages.clear();
|
||||
@@ -5224,7 +5400,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
running: true,
|
||||
});
|
||||
if (timelineItem) {
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId);
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId, parsed.threadId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5257,7 +5433,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
if (!subAgentCallId) {
|
||||
this.emittedExecCommandCompletedCallIds.add(timelineItem.callId);
|
||||
}
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId);
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId, parsed.threadId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5306,7 +5482,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
callId: parsed.callId,
|
||||
changes: parsed.changes,
|
||||
});
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId);
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId, parsed.threadId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5336,7 +5512,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
changes: parsed.changes,
|
||||
stdout: parsed.stdout,
|
||||
});
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId);
|
||||
this.emitCodexToolTimelineItem(timelineItem, subAgentCallId, parsed.threadId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5372,8 +5548,11 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
parentCallId: childSubAgentCallId,
|
||||
})
|
||||
: [];
|
||||
const imageItems = mcpToolResultImagesToTimeline(parsed.item);
|
||||
if (childSubAgentCallId) {
|
||||
this.emitCompletedProviderSubagentItem(parsed, timelineItem);
|
||||
this.handleSubAgentChildItemCompleted(childSubAgentCallId, parsed.item.id, timelineItem);
|
||||
this.emitProviderSubagentTimelineItems(parsed.threadId, imageItems);
|
||||
this.replayPendingSubAgentNotifications(registeredChildThreadIds);
|
||||
return;
|
||||
}
|
||||
@@ -5408,7 +5587,6 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
this.warnOnIncompleteEditToolCall(timelineItem, "item_completed", parsed.item);
|
||||
}
|
||||
const imageItems = mcpToolResultImagesToTimeline(parsed.item);
|
||||
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
|
||||
if (timelineItem.type === "assistant_message") {
|
||||
this.pendingAssistantMessageBoundary = true;
|
||||
@@ -5452,26 +5630,24 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
timelineItem: Extract<AgentTimelineItem, { type: "assistant_message" | "reasoning" }>,
|
||||
streamedText: string,
|
||||
): void {
|
||||
if (!timelineItem.text.startsWith(streamedText)) {
|
||||
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem });
|
||||
return;
|
||||
}
|
||||
const item = this.buildMissingFinalTextSuffix(timelineItem, streamedText);
|
||||
if (item) this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item });
|
||||
}
|
||||
|
||||
private buildMissingFinalTextSuffix(
|
||||
timelineItem: Extract<AgentTimelineItem, { type: "assistant_message" | "reasoning" }>,
|
||||
streamedText: string,
|
||||
): AgentTimelineItem | null {
|
||||
if (!timelineItem.text.startsWith(streamedText)) return timelineItem;
|
||||
const suffix = timelineItem.text.slice(streamedText.length);
|
||||
if (!suffix) {
|
||||
return;
|
||||
}
|
||||
this.emitEvent({
|
||||
type: "timeline",
|
||||
provider: CODEX_PROVIDER,
|
||||
item:
|
||||
timelineItem.type === "assistant_message"
|
||||
? {
|
||||
type: timelineItem.type,
|
||||
text: suffix,
|
||||
...(timelineItem.messageId ? { messageId: timelineItem.messageId } : {}),
|
||||
}
|
||||
: { type: timelineItem.type, text: suffix },
|
||||
});
|
||||
if (!suffix) return null;
|
||||
return timelineItem.type === "assistant_message"
|
||||
? {
|
||||
type: timelineItem.type,
|
||||
text: suffix,
|
||||
...(timelineItem.messageId ? { messageId: timelineItem.messageId } : {}),
|
||||
}
|
||||
: { type: timelineItem.type, text: suffix };
|
||||
}
|
||||
|
||||
private applyBufferedDeltaTextToTimelineItem(
|
||||
@@ -5484,14 +5660,15 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
if (timelineItem.type === "assistant_message") {
|
||||
const buffered = this.pendingAgentMessages.get(itemId);
|
||||
if (buffered && buffered.length > 0) {
|
||||
timelineItem.text = buffered;
|
||||
if (!timelineItem.text.startsWith(buffered)) timelineItem.text = buffered;
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (timelineItem.type === "reasoning") {
|
||||
const buffered = this.pendingReasoning.get(itemId);
|
||||
if (buffered && buffered.length > 0) {
|
||||
timelineItem.text = buffered.join("");
|
||||
const streamedText = buffered.join("");
|
||||
if (!timelineItem.text.startsWith(streamedText)) timelineItem.text = streamedText;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5537,6 +5714,7 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
parentCallId: childSubAgentCallId,
|
||||
});
|
||||
if (childSubAgentCallId) {
|
||||
this.emitStartedProviderSubagentItem(parsed.threadId, timelineItem);
|
||||
if (parsed.item.id) {
|
||||
this.upsertSubAgentChildItem(childSubAgentCallId, parsed.item.id, timelineItem);
|
||||
}
|
||||
@@ -5580,6 +5758,18 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
const childSubAgentCallId = this.getSubAgentCallIdForThread(parsed.threadId);
|
||||
if (childSubAgentCallId) {
|
||||
const childMessageId = itemId ?? timelineItem.messageId;
|
||||
if (!childMessageId) {
|
||||
return;
|
||||
}
|
||||
const childMessageKey = `${parsed.threadId ?? childSubAgentCallId}:${childMessageId}`;
|
||||
if (this.emittedProviderSubagentUserMessageKeys.has(childMessageKey)) {
|
||||
return;
|
||||
}
|
||||
this.emittedProviderSubagentUserMessageKeys.add(childMessageKey);
|
||||
if (parsed.threadId) {
|
||||
this.emitProviderSubagentTimeline(parsed.threadId, timelineItem);
|
||||
}
|
||||
if (itemId) {
|
||||
this.upsertSubAgentChildItem(childSubAgentCallId, itemId, timelineItem);
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -164,6 +164,34 @@ describe("OpenCodeServerManager generations", () => {
|
||||
expect(runtime.terminatedPorts).toEqual([4474]);
|
||||
});
|
||||
|
||||
test("acquireExisting keeps a retired dedicated server alive until every reference releases", async () => {
|
||||
const { manager, runtime } = createTestManager([4475]);
|
||||
|
||||
const dedicatedAcquisition = await manager.acquireDedicated({ PASEO_AGENT_ID: "parent" });
|
||||
const existingAcquisition = manager.acquireExisting(dedicatedAcquisition.server.url);
|
||||
|
||||
expect(existingAcquisition?.server.url).toBe("http://127.0.0.1:4475");
|
||||
|
||||
dedicatedAcquisition.release();
|
||||
expect(runtime.terminatedPorts).toEqual([]);
|
||||
|
||||
existingAcquisition?.release();
|
||||
expect(runtime.terminatedPorts).toEqual([4475]);
|
||||
});
|
||||
|
||||
test("acquireExisting returns null for unknown or dead server urls", async () => {
|
||||
const { manager, runtime } = createTestManager([4476]);
|
||||
|
||||
const acquisition = await manager.acquireDedicated({ PASEO_AGENT_ID: "parent" });
|
||||
const url = acquisition.server.url;
|
||||
|
||||
expect(manager.acquireExisting("http://127.0.0.1:9999")).toBe(null);
|
||||
|
||||
acquisition.release();
|
||||
expect(runtime.terminatedPorts).toEqual([4476]);
|
||||
expect(manager.acquireExisting(url)).toBe(null);
|
||||
});
|
||||
|
||||
test("repeated rotations leave zero unreferenced retired servers", async () => {
|
||||
const { manager, runtime } = createTestManager([4501, 4502, 4503]);
|
||||
|
||||
|
||||
@@ -29,6 +29,7 @@ export interface OpenCodeServerManagerLike {
|
||||
acquireCurrent(): Promise<OpenCodeServerAcquisition>;
|
||||
acquireNew(): Promise<OpenCodeServerAcquisition>;
|
||||
acquireDedicated(env: Record<string, string>): Promise<OpenCodeServerAcquisition>;
|
||||
acquireExisting(url: string): OpenCodeServerAcquisition | null;
|
||||
shutdown(): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -164,6 +165,27 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike {
|
||||
}
|
||||
}
|
||||
|
||||
acquireExisting(url: string): OpenCodeServerAcquisition | null {
|
||||
const server = this.findLiveServerByUrl(url);
|
||||
return server ? this.acquireServer(server) : null;
|
||||
}
|
||||
|
||||
private findLiveServerByUrl(url: string): OpenCodeServerGeneration | null {
|
||||
const servers = [
|
||||
...(this.currentServer ? [this.currentServer] : []),
|
||||
...Array.from(this.retiredServers),
|
||||
];
|
||||
return servers.find((server) => server.url === url && this.isServerLive(server)) ?? null;
|
||||
}
|
||||
|
||||
private isServerLive(server: OpenCodeServerGeneration): boolean {
|
||||
return (
|
||||
!server.process.killed &&
|
||||
server.process.exitCode === null &&
|
||||
server.process.signalCode === null
|
||||
);
|
||||
}
|
||||
|
||||
private acquireServer(server: OpenCodeServerGeneration): OpenCodeServerAcquisition {
|
||||
server.refCount += 1;
|
||||
let released = false;
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import type { OpenCodeServerAcquisition, OpenCodeServerManagerLike } from "./server-manager.js";
|
||||
|
||||
export interface TestOpenCodeServerAcquisition {
|
||||
kind: "current" | "new" | "dedicated";
|
||||
kind: "current" | "new" | "dedicated" | "existing";
|
||||
env?: Record<string, string>;
|
||||
url?: string;
|
||||
released: boolean;
|
||||
}
|
||||
|
||||
@@ -28,14 +29,20 @@ export class TestOpenCodeServerManager implements OpenCodeServerManagerLike {
|
||||
return this.recordAcquisition({ kind: "dedicated", env });
|
||||
}
|
||||
|
||||
acquireExisting(url: string): OpenCodeServerAcquisition | null {
|
||||
return url === this.server.url ? this.recordAcquisition({ kind: "existing", url }) : null;
|
||||
}
|
||||
|
||||
private recordAcquisition(input: {
|
||||
kind: TestOpenCodeServerAcquisition["kind"];
|
||||
env?: Record<string, string>;
|
||||
url?: string;
|
||||
}): OpenCodeServerAcquisition {
|
||||
const acquisition: TestOpenCodeServerAcquisition = {
|
||||
kind: input.kind,
|
||||
released: false,
|
||||
...(input.env ? { env: input.env } : {}),
|
||||
...(input.url ? { url: input.url } : {}),
|
||||
};
|
||||
this.acquisitions.push(acquisition);
|
||||
return {
|
||||
|
||||
+19
-2
@@ -9,8 +9,9 @@ interface OpenCodeResponse {
|
||||
|
||||
export class TestOpenCodeHarness implements OpenCodeServerManagerLike {
|
||||
readonly acquisitions: Array<{
|
||||
kind: "current" | "new" | "dedicated";
|
||||
kind: "current" | "new" | "dedicated" | "existing";
|
||||
env?: Record<string, string>;
|
||||
url?: string;
|
||||
releaseCount: number;
|
||||
}> = [];
|
||||
readonly clientCreations: Array<{ baseUrl: string; directory: string }> = [];
|
||||
@@ -34,14 +35,20 @@ export class TestOpenCodeHarness implements OpenCodeServerManagerLike {
|
||||
return this.recordAcquisition({ kind: "dedicated", env });
|
||||
}
|
||||
|
||||
acquireExisting(url: string): OpenCodeServerAcquisition | null {
|
||||
return url === this.server.url ? this.recordAcquisition({ kind: "existing", url }) : null;
|
||||
}
|
||||
|
||||
private recordAcquisition(input: {
|
||||
kind: "current" | "new" | "dedicated";
|
||||
kind: "current" | "new" | "dedicated" | "existing";
|
||||
env?: Record<string, string>;
|
||||
url?: string;
|
||||
}): OpenCodeServerAcquisition {
|
||||
const acquisition = {
|
||||
kind: input.kind,
|
||||
releaseCount: 0,
|
||||
...(input.env ? { env: input.env } : {}),
|
||||
...(input.url ? { url: input.url } : {}),
|
||||
};
|
||||
this.acquisitions.push(acquisition);
|
||||
return {
|
||||
@@ -82,6 +89,7 @@ export class TestOpenCodeClient {
|
||||
sessionCommand: [] as unknown[],
|
||||
sessionCreate: [] as unknown[],
|
||||
sessionDelete: [] as unknown[],
|
||||
sessionChildren: [] as unknown[],
|
||||
sessionGet: [] as unknown[],
|
||||
sessionMessages: [] as unknown[],
|
||||
sessionPromptAsync: [] as unknown[],
|
||||
@@ -106,6 +114,8 @@ export class TestOpenCodeClient {
|
||||
sessionCommandResponse: OpenCodeResponse = {};
|
||||
sessionCreateResponse: OpenCodeResponse = { data: { id: "session-1" } };
|
||||
sessionDeleteResponse: OpenCodeResponse = {};
|
||||
sessionChildrenResponses: OpenCodeResponse[] = [];
|
||||
sessionChildrenImplementation: ((parameters: unknown) => Promise<OpenCodeResponse>) | null = null;
|
||||
sessionGetResponse: OpenCodeResponse = {
|
||||
data: { id: "session-1", directory: "/workspace/repo", title: null },
|
||||
};
|
||||
@@ -219,6 +229,13 @@ export class TestOpenCodeClient {
|
||||
this.calls.sessionDelete.push(parameters);
|
||||
return this.sessionDeleteResponse;
|
||||
},
|
||||
children: async (parameters: unknown) => {
|
||||
this.calls.sessionChildren.push(parameters);
|
||||
if (this.sessionChildrenImplementation) {
|
||||
return await this.sessionChildrenImplementation(parameters);
|
||||
}
|
||||
return this.sessionChildrenResponses.shift() ?? { data: [] };
|
||||
},
|
||||
get: async (parameters: unknown) => {
|
||||
this.calls.sessionGet.push(parameters);
|
||||
return this.sessionGetResponse;
|
||||
|
||||
@@ -1165,6 +1165,43 @@ export class Session {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "provider_subagent") {
|
||||
if (!this.supports(CLIENT_CAPS.providerSubagents)) {
|
||||
return;
|
||||
}
|
||||
const update = event.event;
|
||||
if (update.type === "upsert") {
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.update",
|
||||
payload: { kind: "upsert", subagent: update.subagent },
|
||||
});
|
||||
} else if (update.type === "timeline") {
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.update",
|
||||
payload: {
|
||||
kind: "timeline",
|
||||
parentAgentId: update.parentAgentId,
|
||||
subagentId: update.subagentId,
|
||||
provider: update.provider,
|
||||
item: update.row.item,
|
||||
timestamp: update.row.timestamp,
|
||||
seq: update.row.seq,
|
||||
epoch: update.epoch,
|
||||
},
|
||||
});
|
||||
} else {
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.update",
|
||||
payload: {
|
||||
kind: "remove",
|
||||
parentAgentId: update.parentAgentId,
|
||||
subagentId: update.subagentId,
|
||||
},
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (
|
||||
this.voiceSession.isActiveForAgent(event.agentId) &&
|
||||
event.event.type === "permission_requested" &&
|
||||
@@ -1450,6 +1487,10 @@ export class Session {
|
||||
switch (msg.type) {
|
||||
case "fetch_agent_timeline_request":
|
||||
return this.handleFetchAgentTimelineRequest(msg);
|
||||
case "agent.provider_subagents.list.request":
|
||||
return this.handleProviderSubagentListRequest(msg);
|
||||
case "agent.provider_subagents.timeline.get.request":
|
||||
return this.handleProviderSubagentTimelineRequest(msg);
|
||||
case "agent.fork_context.request":
|
||||
return this.handleAgentForkContextRequest(msg);
|
||||
default:
|
||||
@@ -2708,7 +2749,7 @@ export class Session {
|
||||
extractTimestamps(record),
|
||||
);
|
||||
}
|
||||
await this.agentManager.hydrateTimelineFromProvider(agentId);
|
||||
await this.agentManager.hydrateTimelineFromProvider(agentId, { broadcast: true });
|
||||
await this.agentUpdates.forwardLiveAgent(snapshot);
|
||||
const timelineSize = this.agentManager.getTimeline(agentId).length;
|
||||
if (requestId) {
|
||||
@@ -5215,6 +5256,96 @@ export class Session {
|
||||
}
|
||||
}
|
||||
|
||||
private async handleProviderSubagentListRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "agent.provider_subagents.list.request" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.list.response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
parentAgentId: msg.parentAgentId,
|
||||
subagents: this.agentManager.listProviderSubagents(msg.parentAgentId),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.list.response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
parentAgentId: msg.parentAgentId,
|
||||
subagents: [],
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async handleProviderSubagentTimelineRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "agent.provider_subagents.timeline.get.request" }>,
|
||||
): Promise<void> {
|
||||
const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail");
|
||||
try {
|
||||
const descriptor = this.agentManager.getProviderSubagent(msg.parentAgentId, msg.subagentId);
|
||||
if (!descriptor) {
|
||||
throw new Error("Provider subagent not found");
|
||||
}
|
||||
const timeline = this.agentManager.fetchProviderSubagentTimeline(
|
||||
msg.parentAgentId,
|
||||
msg.subagentId,
|
||||
{
|
||||
direction,
|
||||
cursor: msg.cursor,
|
||||
limit: msg.limit ?? (direction === "after" ? 0 : 200),
|
||||
},
|
||||
);
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.timeline.get.response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
parentAgentId: msg.parentAgentId,
|
||||
subagentId: msg.subagentId,
|
||||
provider: descriptor.provider,
|
||||
direction,
|
||||
epoch: timeline.epoch,
|
||||
reset: timeline.reset,
|
||||
staleCursor: timeline.staleCursor,
|
||||
gap: timeline.gap,
|
||||
window: timeline.window,
|
||||
hasOlder: timeline.hasOlder,
|
||||
hasNewer: timeline.hasNewer,
|
||||
rows: timeline.rows.map((row) => ({
|
||||
item: row.item,
|
||||
timestamp: row.timestamp,
|
||||
seq: row.seq,
|
||||
})),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emit({
|
||||
type: "agent.provider_subagents.timeline.get.response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
parentAgentId: msg.parentAgentId,
|
||||
subagentId: msg.subagentId,
|
||||
provider: null,
|
||||
direction,
|
||||
epoch: "",
|
||||
reset: false,
|
||||
staleCursor: false,
|
||||
gap: false,
|
||||
window: { minSeq: 0, maxSeq: 0, nextSeq: 0 },
|
||||
hasOlder: false,
|
||||
hasNewer: false,
|
||||
rows: [],
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async handleAgentForkContextRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "agent.fork_context.request" }>,
|
||||
): Promise<void> {
|
||||
|
||||
@@ -1259,6 +1259,8 @@ export class VoiceAssistantWebSocketServer {
|
||||
daemonSelfUpdate: true,
|
||||
// COMPAT(agentForkContext): added in v0.1.102, remove gate after 2026-12-28.
|
||||
agentForkContext: true,
|
||||
// COMPAT(providerSubagents): added in v0.1.107, remove gate after 2027-01-12.
|
||||
providerSubagents: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user