Archived
fix: require exact active subsession files
This commit is contained in:
@@ -1291,6 +1291,152 @@ describe("PiSessionService", () => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("uses the verified child file instead of an active copied child with the same id", async () => {
|
||||||
|
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-active-copy-child-"));
|
||||||
|
const parentFile = join(tempDir, "parent.jsonl");
|
||||||
|
const originalChildFile = join(tempDir, "original-child.jsonl");
|
||||||
|
const copiedChildFile = join(tempDir, "copied-child.jsonl");
|
||||||
|
await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8");
|
||||||
|
await writeFile(originalChildFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
|
||||||
|
await writeFile(copiedChildFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
|
||||||
|
|
||||||
|
try {
|
||||||
|
const copiedManager = fakeSessionManager("/workspace-feature", {
|
||||||
|
getBranch: () => [{ type: "message", message: { role: "assistant", content: "copied child result" } }],
|
||||||
|
});
|
||||||
|
const originalManager = fakeSessionManager("/workspace-feature", {
|
||||||
|
getBranch: () => [{ type: "message", message: { role: "assistant", content: "original child result" } }],
|
||||||
|
});
|
||||||
|
const parentManager = fakeSessionManager("/workspace", {
|
||||||
|
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: originalChildFile, cwd: "/workspace-feature" } }],
|
||||||
|
});
|
||||||
|
const copiedChild = fakeRuntime("child-1", { sessionFile: copiedChildFile, sessionManager: copiedManager, isStreaming: true });
|
||||||
|
const originalChild = fakeRuntime("child-1", { sessionFile: originalChildFile, sessionManager: originalManager });
|
||||||
|
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
|
||||||
|
const createAgentRuntime: RuntimeCreator = (_createRuntime, options) => {
|
||||||
|
if (options.sessionManager === copiedManager) return Promise.resolve(copiedChild.runtime);
|
||||||
|
if (options.sessionManager === originalManager) return Promise.resolve(originalChild.runtime);
|
||||||
|
if (options.sessionManager === parentManager) return Promise.resolve(parent.runtime);
|
||||||
|
throw new Error("unexpected session manager");
|
||||||
|
};
|
||||||
|
const open = vi.fn((path: string) => {
|
||||||
|
if (path === copiedChildFile) return copiedManager;
|
||||||
|
if (path === originalChildFile) return originalManager;
|
||||||
|
if (path === parentFile) return parentManager;
|
||||||
|
throw new Error(`unexpected open path ${path}`);
|
||||||
|
});
|
||||||
|
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||||
|
createAgentRuntime,
|
||||||
|
sessionManager: {
|
||||||
|
create: () => parentManager,
|
||||||
|
list: (cwd: string) => Promise.resolve(cwd === "/workspace-feature" ? [{ ...sessionRecord("child-1", "/workspace-feature"), path: copiedChildFile, parentSessionPath: parentFile }] : []),
|
||||||
|
listAll: () => Promise.resolve([]),
|
||||||
|
open,
|
||||||
|
},
|
||||||
|
archiveStore: emptyArchiveStore(),
|
||||||
|
heartbeatIntervalMs: 60_000,
|
||||||
|
});
|
||||||
|
|
||||||
|
await service.status(sessionRef("child-1", "/workspace-feature"));
|
||||||
|
await service.start("/workspace");
|
||||||
|
|
||||||
|
await expect(service.listSubsessions("parent-1", parentFile)).resolves.toEqual([
|
||||||
|
{ sessionId: "child-1", cwd: "/workspace-feature", status: "idle" },
|
||||||
|
]);
|
||||||
|
|
||||||
|
copiedChild.session.isStreaming = true;
|
||||||
|
copiedChild.emit({ type: "agent_start" });
|
||||||
|
copiedChild.session.isStreaming = false;
|
||||||
|
copiedChild.emit({ type: "agent_end" });
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||||
|
expect(parent.calls.sendCustomMessage).toHaveLength(0);
|
||||||
|
|
||||||
|
await expect(service.checkSubsession("parent-1", "child-1", parentFile)).resolves.toMatchObject({
|
||||||
|
sessionId: "child-1",
|
||||||
|
cwd: "/workspace-feature",
|
||||||
|
status: "idle",
|
||||||
|
finalText: "original child result",
|
||||||
|
messageCount: 1,
|
||||||
|
});
|
||||||
|
const read = await service.readSubsession("parent-1", "child-1", { roles: ["assistant"] }, parentFile);
|
||||||
|
expect(read.entries[0]?.parts[0]).toMatchObject({ kind: "text", text: "original child result" });
|
||||||
|
expect(open).toHaveBeenCalledWith(originalChildFile);
|
||||||
|
await service.dispose();
|
||||||
|
} finally {
|
||||||
|
await rm(tempDir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("uses the verified parent file instead of an active copied parent with the same id", async () => {
|
||||||
|
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-active-copy-parent-"));
|
||||||
|
const parentFile = join(tempDir, "parent.jsonl");
|
||||||
|
const copiedParentFile = join(tempDir, "copied-parent.jsonl");
|
||||||
|
const childFile = join(tempDir, "child.jsonl");
|
||||||
|
await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8");
|
||||||
|
await writeFile(copiedParentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8");
|
||||||
|
await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
|
||||||
|
|
||||||
|
try {
|
||||||
|
const childManager = fakeSessionManager("/workspace-feature", {
|
||||||
|
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }],
|
||||||
|
getBranch: () => [{ type: "message", message: { role: "assistant", content: "child result" } }],
|
||||||
|
});
|
||||||
|
const parentManager = fakeSessionManager("/workspace", {
|
||||||
|
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" } }],
|
||||||
|
});
|
||||||
|
const copiedParentManager = fakeSessionManager("/workspace", { getEntries: () => [] });
|
||||||
|
const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager });
|
||||||
|
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
|
||||||
|
const copiedParent = fakeRuntime("parent-1", { sessionFile: copiedParentFile, sessionManager: copiedParentManager });
|
||||||
|
const createAgentRuntime: RuntimeCreator = (_createRuntime, options) => {
|
||||||
|
if (options.sessionManager === childManager) return Promise.resolve(child.runtime);
|
||||||
|
if (options.sessionManager === parentManager) return Promise.resolve(parent.runtime);
|
||||||
|
if (options.sessionManager === copiedParentManager) return Promise.resolve(copiedParent.runtime);
|
||||||
|
throw new Error("unexpected session manager");
|
||||||
|
};
|
||||||
|
const open = vi.fn((path: string) => {
|
||||||
|
if (path === childFile) return childManager;
|
||||||
|
if (path === parentFile) return parentManager;
|
||||||
|
if (path === copiedParentFile) return copiedParentManager;
|
||||||
|
throw new Error(`unexpected open path ${path}`);
|
||||||
|
});
|
||||||
|
const service = new PiSessionService(new CapturingSessionEventHub(), {
|
||||||
|
createAgentRuntime,
|
||||||
|
sessionManager: {
|
||||||
|
create: () => copiedParentManager,
|
||||||
|
list: (cwd: string) => Promise.resolve(cwd === "/workspace"
|
||||||
|
? [{ ...sessionRecord("parent-1", "/workspace"), path: copiedParentFile }]
|
||||||
|
: [{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]),
|
||||||
|
listAll: () => Promise.resolve([]),
|
||||||
|
open,
|
||||||
|
},
|
||||||
|
archiveStore: emptyArchiveStore(),
|
||||||
|
heartbeatIntervalMs: 60_000,
|
||||||
|
});
|
||||||
|
|
||||||
|
await service.status(sessionRef("child-1", "/workspace-feature"));
|
||||||
|
await service.status(sessionRef("parent-1", "/workspace"));
|
||||||
|
|
||||||
|
await expect(service.listSubsessions("parent-1", copiedParentFile)).resolves.toEqual([]);
|
||||||
|
await expect(service.checkSubsession("parent-1", "child-1", copiedParentFile)).rejects.toThrow("not one of your subsessions");
|
||||||
|
await expect(service.readSubsession("parent-1", "child-1", {}, copiedParentFile)).rejects.toThrow("not one of your subsessions");
|
||||||
|
|
||||||
|
child.session.isStreaming = true;
|
||||||
|
child.emit({ type: "agent_start" });
|
||||||
|
child.session.isStreaming = false;
|
||||||
|
child.emit({ type: "agent_end" });
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||||
|
|
||||||
|
expect(copiedParent.calls.sendCustomMessage).toHaveLength(0);
|
||||||
|
expect(parent.calls.sendCustomMessage).toHaveLength(1);
|
||||||
|
expect(parent.calls.sendCustomMessage[0]?.message.content).toContain("Subsession child-1 stopped working");
|
||||||
|
expect(open).toHaveBeenCalledWith(parentFile);
|
||||||
|
await service.dispose();
|
||||||
|
} finally {
|
||||||
|
await rm(tempDir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
it("does not relink a child marker when the current child file header no longer records the parent", async () => {
|
it("does not relink a child marker when the current child file header no longer records the parent", async () => {
|
||||||
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-stale-child-header-"));
|
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-stale-child-header-"));
|
||||||
const parentFile = join(tempDir, "parent.jsonl");
|
const parentFile = join(tempDir, "parent.jsonl");
|
||||||
|
|||||||
@@ -314,7 +314,7 @@ export class PiSessionService {
|
|||||||
private readonly subsessionChildren = new Map<string, Set<string>>();
|
private readonly subsessionChildren = new Map<string, Set<string>>();
|
||||||
/** Tracked subsession id -> persisted recovery details for the child. */
|
/** Tracked subsession id -> persisted recovery details for the child. */
|
||||||
private readonly subsessionLinks = new Map<string, TrackedSubsessionLink>();
|
private readonly subsessionLinks = new Map<string, TrackedSubsessionLink>();
|
||||||
/** Parent session ids whose persisted links have already been loaded. */
|
/** Parent id/file identities whose persisted links have already been loaded. */
|
||||||
private readonly subsessionHydratedParents = new Set<string>();
|
private readonly subsessionHydratedParents = new Set<string>();
|
||||||
/**
|
/**
|
||||||
* Tracked subsession id -> whether a completion notification is armed.
|
* Tracked subsession id -> whether a completion notification is armed.
|
||||||
@@ -348,9 +348,9 @@ export class PiSessionService {
|
|||||||
this.spawnTargets === undefined ? undefined : (input) => this.spawnSession(input),
|
this.spawnTargets === undefined ? undefined : (input) => this.spawnSession(input),
|
||||||
!subsessionsActive ? undefined : {
|
!subsessionsActive ? undefined : {
|
||||||
spawn: (input) => this.spawnSubsession(input),
|
spawn: (input) => this.spawnSubsession(input),
|
||||||
list: (parentSessionId) => this.listSubsessions(parentSessionId),
|
list: (parentSessionId, parentSessionFile) => this.listSubsessions(parentSessionId, parentSessionFile),
|
||||||
check: (parentSessionId, sessionId) => this.checkSubsession(parentSessionId, sessionId),
|
check: (parentSessionId, sessionId, parentSessionFile) => this.checkSubsession(parentSessionId, sessionId, parentSessionFile),
|
||||||
read: (parentSessionId, sessionId, query) => this.readSubsession(parentSessionId, sessionId, query),
|
read: (parentSessionId, sessionId, query, parentSessionFile) => this.readSubsession(parentSessionId, sessionId, query, parentSessionFile),
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime;
|
this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime;
|
||||||
@@ -487,16 +487,18 @@ export class PiSessionService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Summaries of the tracked subsessions spawned by `parentSessionId`. */
|
/** Summaries of the tracked subsessions spawned by `parentSessionId`. */
|
||||||
async listSubsessions(parentSessionId: string): Promise<SubsessionSummary[]> {
|
async listSubsessions(parentSessionId: string, parentSessionFile?: string): Promise<SubsessionSummary[]> {
|
||||||
await this.hydrateSubsessionsForParent(parentSessionId);
|
const parentFile = nonEmptyString(parentSessionFile);
|
||||||
|
await this.hydrateSubsessionsForParent(parentSessionId, parentFile);
|
||||||
const childIds = this.subsessionChildren.get(parentSessionId);
|
const childIds = this.subsessionChildren.get(parentSessionId);
|
||||||
if (childIds === undefined) return [];
|
if (childIds === undefined) return [];
|
||||||
return Promise.all([...childIds].map(async (childId) => ({ sessionId: childId, ...(await this.subsessionSummaryFields(childId)) })));
|
const authorizedChildIds = [...childIds].filter((childId) => this.subsessionLinkBelongsToParent(parentSessionId, parentFile, childId));
|
||||||
|
return Promise.all(authorizedChildIds.map(async (childId) => ({ sessionId: childId, ...(await this.subsessionSummaryFields(childId)) })));
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Status and final result of a subsession, scoped to the caller's children. */
|
/** Status and final result of a subsession, scoped to the caller's children. */
|
||||||
async checkSubsession(parentSessionId: string, sessionId: string): Promise<SubsessionCheckResult> {
|
async checkSubsession(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise<SubsessionCheckResult> {
|
||||||
const session = await this.openSubsession(parentSessionId, sessionId);
|
const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile);
|
||||||
const messages = historyMessages(session);
|
const messages = historyMessages(session);
|
||||||
return {
|
return {
|
||||||
sessionId,
|
sessionId,
|
||||||
@@ -508,8 +510,8 @@ export class PiSessionService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Filtered, paginated transcript of a subsession, scoped to the caller's children. */
|
/** Filtered, paginated transcript of a subsession, scoped to the caller's children. */
|
||||||
async readSubsession(parentSessionId: string, sessionId: string, query: SubsessionReadQuery): Promise<SubsessionReadResult> {
|
async readSubsession(parentSessionId: string, sessionId: string, query: SubsessionReadQuery, parentSessionFile?: string): Promise<SubsessionReadResult> {
|
||||||
const session = await this.openSubsession(parentSessionId, sessionId);
|
const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile);
|
||||||
const view = buildTranscriptView(historyMessages(session), query);
|
const view = buildTranscriptView(historyMessages(session), query);
|
||||||
return {
|
return {
|
||||||
sessionId,
|
sessionId,
|
||||||
@@ -520,14 +522,41 @@ export class PiSessionService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/** Open a session after verifying it is one of the caller's tracked children. */
|
/** Open a session after verifying it is one of the caller's tracked children. */
|
||||||
private async openSubsession(parentSessionId: string, sessionId: string): Promise<PiAgentSession> {
|
private async openSubsession(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise<PiAgentSession> {
|
||||||
await this.hydrateSubsessionsForParent(parentSessionId);
|
const parentFile = nonEmptyString(parentSessionFile);
|
||||||
if (this.subsessionParents.get(sessionId) !== parentSessionId) {
|
await this.hydrateSubsessionsForParent(parentSessionId, parentFile);
|
||||||
|
if (this.subsessionParents.get(sessionId) !== parentSessionId || !this.subsessionLinkBelongsToParent(parentSessionId, parentFile, sessionId)) {
|
||||||
throw new Error(`Session ${sessionId} is not one of your subsessions`);
|
throw new Error(`Session ${sessionId} is not one of your subsessions`);
|
||||||
}
|
}
|
||||||
return this.getOrOpenTrackedSubsession(sessionId);
|
return this.getOrOpenTrackedSubsession(sessionId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private subsessionLinkBelongsToParent(parentSessionId: string, parentSessionFile: string | undefined, childSessionId: string): boolean {
|
||||||
|
const link = this.subsessionLinks.get(childSessionId);
|
||||||
|
if (link?.parentSessionId !== parentSessionId) return false;
|
||||||
|
return parentSessionFile === undefined || trackedLinkParentFileMatches(link, parentSessionFile);
|
||||||
|
}
|
||||||
|
|
||||||
|
private activeChildForSubsessionLink(link: TrackedSubsessionLink): ActiveSession<PiSessionRuntime> | undefined {
|
||||||
|
const active = this.active.get(link.childSessionId);
|
||||||
|
if (active === undefined) return undefined;
|
||||||
|
return activeSessionFileMatches(active, link.childSessionFile) ? active : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
private activeParentForSubsessionLink(link: TrackedSubsessionLink): ActiveSession<PiSessionRuntime> | undefined {
|
||||||
|
const active = this.active.get(link.parentSessionId);
|
||||||
|
if (active === undefined) return undefined;
|
||||||
|
return activeSessionFileMatches(active, link.parentSessionFile) ? active : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
private subsessionLinkForActiveChild(session: PiAgentSession): TrackedSubsessionLink | undefined {
|
||||||
|
const childId = session.sessionId;
|
||||||
|
const parentId = this.subsessionParents.get(childId);
|
||||||
|
const link = this.subsessionLinks.get(childId);
|
||||||
|
if (parentId === undefined || link?.parentSessionId !== parentId) return undefined;
|
||||||
|
return sessionFileMatches(session, link.childSessionFile) ? link : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
private registerVerifiedSubsession(link: TrackedSubsessionLink): void {
|
private registerVerifiedSubsession(link: TrackedSubsessionLink): void {
|
||||||
const { childSessionId, parentSessionId } = link;
|
const { childSessionId, parentSessionId } = link;
|
||||||
const previousParentId = this.subsessionParents.get(childSessionId);
|
const previousParentId = this.subsessionParents.get(childSessionId);
|
||||||
@@ -558,7 +587,7 @@ export class PiSessionService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private persistSubsessionLink(link: TrackedSubsessionLink): void {
|
private persistSubsessionLink(link: TrackedSubsessionLink): void {
|
||||||
const parent = this.active.get(link.parentSessionId)?.runtime.session;
|
const parent = this.activeParentForSubsessionLink(link)?.runtime.session;
|
||||||
if (parent === undefined) return;
|
if (parent === undefined) return;
|
||||||
if (parent.sessionManager.appendCustomEntry === undefined) return;
|
if (parent.sessionManager.appendCustomEntry === undefined) return;
|
||||||
try {
|
try {
|
||||||
@@ -585,20 +614,39 @@ export class PiSessionService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private async hydrateSubsessionsForParent(parentSessionId: string): Promise<void> {
|
private async hydrateSubsessionsForParent(parentSessionId: string, parentSessionFile?: string): Promise<void> {
|
||||||
if (this.subsessionHydratedParents.has(parentSessionId)) return;
|
const hydrationKey = subsessionHydratedParentKey(parentSessionId, parentSessionFile);
|
||||||
const parent = this.active.get(parentSessionId)?.runtime.session;
|
if (this.subsessionHydratedParents.has(hydrationKey)) return;
|
||||||
if (parent === undefined) return;
|
|
||||||
|
|
||||||
const parentSessionFile = nonEmptyString(parent.sessionFile);
|
const activeParent = this.active.get(parentSessionId);
|
||||||
await this.registerPersistedSubsessionLinks(parentSessionId, parent, parentSessionFile);
|
if (activeParent !== undefined && (parentSessionFile === undefined || activeSessionFileMatches(activeParent, parentSessionFile))) {
|
||||||
this.subsessionHydratedParents.add(parentSessionId);
|
const activeParentFile = nonEmptyString(activeParent.runtime.session.sessionFile);
|
||||||
|
await this.registerPersistedSubsessionLinks(parentSessionId, activeParent.runtime.session.sessionManager, activeParentFile);
|
||||||
|
this.subsessionHydratedParents.add(hydrationKey);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (parentSessionFile === undefined) return;
|
||||||
|
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) {
|
||||||
|
this.subsessionHydratedParents.add(hydrationKey);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let parentManager: PiSessionManager;
|
||||||
|
try {
|
||||||
|
parentManager = this.sessionManager.open(parentSessionFile);
|
||||||
|
} catch {
|
||||||
|
this.subsessionHydratedParents.add(hydrationKey);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await this.registerPersistedSubsessionLinks(parentSessionId, parentManager, parentSessionFile);
|
||||||
|
this.subsessionHydratedParents.add(hydrationKey);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async registerPersistedSubsessionLinks(parentSessionId: string, parent: PiAgentSession, parentSessionFile: string | undefined): Promise<void> {
|
private async registerPersistedSubsessionLinks(parentSessionId: string, parentManager: PiSessionManager, parentSessionFile: string | undefined): Promise<void> {
|
||||||
// Parent custom links are the authoritative recovery record: verify the
|
// Parent custom links are the authoritative recovery record: verify the
|
||||||
// exact live child file/header or an exact archived child before tracking.
|
// exact live child file/header or an exact archived child before tracking.
|
||||||
const entries = parent.sessionManager.getEntries?.() ?? parent.sessionManager.getBranch();
|
const entries = parentManager.getEntries?.() ?? parentManager.getBranch();
|
||||||
for (const entry of entries) {
|
for (const entry of entries) {
|
||||||
const link = parsePersistedParentSubsessionLink(entry);
|
const link = parsePersistedParentSubsessionLink(entry);
|
||||||
if (link === undefined) continue;
|
if (link === undefined) continue;
|
||||||
@@ -684,34 +732,32 @@ export class PiSessionService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private async getOrOpenTrackedSubsession(sessionId: string): Promise<PiAgentSession> {
|
private async getOrOpenTrackedSubsession(sessionId: string): Promise<PiAgentSession> {
|
||||||
const active = this.active.get(sessionId);
|
const link = this.subsessionLinks.get(sessionId);
|
||||||
|
if (link === undefined) throw new Error("Session not found");
|
||||||
|
|
||||||
|
const active = this.activeChildForSubsessionLink(link);
|
||||||
if (active !== undefined) return active.runtime.session;
|
if (active !== undefined) return active.runtime.session;
|
||||||
|
|
||||||
const archived = await this.getArchivedExact(sessionId);
|
const archived = await this.getArchivedExact(sessionId);
|
||||||
if (archived?.archivePath !== undefined) return (await this.create(this.sessionManager.open(archived.archivePath), archived.cwd)).runtime.session;
|
if (archived?.archivePath !== undefined) return (await this.create(this.sessionManager.open(archived.archivePath), archived.cwd)).runtime.session;
|
||||||
|
|
||||||
const link = this.subsessionLinks.get(sessionId);
|
if (link.childSessionFile !== undefined) {
|
||||||
if (link?.childSessionFile !== undefined) {
|
|
||||||
if (!(await sessionFileHeaderMatches(link.childSessionFile, { sessionId, parentSessionFile: link.parentSessionFile }))) throw new Error("Session not found");
|
if (!(await sessionFileHeaderMatches(link.childSessionFile, { sessionId, parentSessionFile: link.parentSessionFile }))) throw new Error("Session not found");
|
||||||
const sessionManager = this.sessionManager.open(link.childSessionFile);
|
const sessionManager = this.sessionManager.open(link.childSessionFile);
|
||||||
return (await this.create(sessionManager, link.cwd ?? sessionManager.getCwd())).runtime.session;
|
return (await this.create(sessionManager, link.cwd ?? sessionManager.getCwd())).runtime.session;
|
||||||
}
|
}
|
||||||
|
|
||||||
const listed = link?.cwd === undefined
|
throw new Error("Session not found");
|
||||||
? (await this.sessionManager.listAll?.() ?? []).find((session) => session.id === sessionId)
|
|
||||||
: (await this.sessionManager.list(link.cwd)).find((session) => session.id === sessionId);
|
|
||||||
if (listed === undefined) throw new Error("Session not found");
|
|
||||||
return (await this.create(this.sessionManager.open(listed.path), listed.cwd)).runtime.session;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private async subsessionSummaryFields(childSessionId: string): Promise<{ cwd: string; status: SubsessionStatus }> {
|
private async subsessionSummaryFields(childSessionId: string): Promise<{ cwd: string; status: SubsessionStatus }> {
|
||||||
const active = this.active.get(childSessionId);
|
const link = this.subsessionLinks.get(childSessionId);
|
||||||
|
const active = link === undefined ? undefined : this.activeChildForSubsessionLink(link);
|
||||||
if (active !== undefined) {
|
if (active !== undefined) {
|
||||||
return { cwd: active.runtime.cwd, status: await this.subsessionStatus(active.runtime.session) };
|
return { cwd: active.runtime.cwd, status: await this.subsessionStatus(active.runtime.session) };
|
||||||
}
|
}
|
||||||
const archived = await this.getArchivedExact(childSessionId);
|
const archived = await this.getArchivedExact(childSessionId);
|
||||||
if (archived !== undefined) return { cwd: archived.cwd, status: "archived" };
|
if (archived !== undefined) return { cwd: archived.cwd, status: "archived" };
|
||||||
const link = this.subsessionLinks.get(childSessionId);
|
|
||||||
if (link?.childSessionFile !== undefined && (await sessionFileHeaderMatches(link.childSessionFile, { sessionId: childSessionId, parentSessionFile: link.parentSessionFile }))) {
|
if (link?.childSessionFile !== undefined && (await sessionFileHeaderMatches(link.childSessionFile, { sessionId: childSessionId, parentSessionFile: link.parentSessionFile }))) {
|
||||||
return { cwd: link.cwd ?? "", status: "idle" };
|
return { cwd: link.cwd ?? "", status: "idle" };
|
||||||
}
|
}
|
||||||
@@ -733,9 +779,9 @@ export class PiSessionService {
|
|||||||
* parent is busy and delivers immediately when it is idle).
|
* parent is busy and delivers immediately when it is idle).
|
||||||
*/
|
*/
|
||||||
private updateSubsessionTracking(session: PiAgentSession): void {
|
private updateSubsessionTracking(session: PiAgentSession): void {
|
||||||
const childId = session.sessionId;
|
const link = this.subsessionLinkForActiveChild(session);
|
||||||
const parentId = this.subsessionParents.get(childId);
|
if (link === undefined) return;
|
||||||
if (parentId === undefined) return;
|
const childId = link.childSessionId;
|
||||||
if (this.hasActiveWork(session)) {
|
if (this.hasActiveWork(session)) {
|
||||||
this.subsessionNotifyArmed.set(childId, true);
|
this.subsessionNotifyArmed.set(childId, true);
|
||||||
return;
|
return;
|
||||||
@@ -746,14 +792,17 @@ export class PiSessionService {
|
|||||||
const finalText = finalAssistantText(historyMessages(session));
|
const finalText = finalAssistantText(historyMessages(session));
|
||||||
const preview = finalText === "" ? "(no output)" : truncateForNotification(finalText);
|
const preview = finalText === "" ? "(no output)" : truncateForNotification(finalText);
|
||||||
const text = `Subsession ${childId} stopped working (status: ${status}). Latest output:\n\n${preview}\n\nUse check_subsession with sessionId "${childId}" for its status and latest output, or read_subsession to look through its full transcript.`;
|
const text = `Subsession ${childId} stopped working (status: ${status}). Latest output:\n\n${preview}\n\nUse check_subsession with sessionId "${childId}" for its status and latest output, or read_subsession to look through its full transcript.`;
|
||||||
void this.notifyParentOfSubsession(parentId, childId, text);
|
void this.notifyParentOfSubsession(link.parentSessionId, childId, text);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async getOrOpenParentForSubsession(parentSessionId: string, childSessionId: string): Promise<PiAgentSession> {
|
private async getOrOpenParentForSubsession(parentSessionId: string, childSessionId: string): Promise<PiAgentSession> {
|
||||||
const active = this.active.get(parentSessionId);
|
const link = this.subsessionLinks.get(childSessionId);
|
||||||
|
if (link?.parentSessionId !== parentSessionId) throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
|
||||||
|
|
||||||
|
const active = this.activeParentForSubsessionLink(link);
|
||||||
if (active !== undefined) return active.runtime.session;
|
if (active !== undefined) return active.runtime.session;
|
||||||
|
|
||||||
const parentSessionFile = this.subsessionLinks.get(childSessionId)?.parentSessionFile;
|
const parentSessionFile = link.parentSessionFile;
|
||||||
if (parentSessionFile === undefined) throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
|
if (parentSessionFile === undefined) throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
|
||||||
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) {
|
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) {
|
||||||
throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
|
throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
|
||||||
@@ -1150,7 +1199,7 @@ export class PiSessionService {
|
|||||||
// Disarm subsession notification before teardown so the abort below cannot
|
// Disarm subsession notification before teardown so the abort below cannot
|
||||||
// emit a "stopped working" event that notifies the parent (e.g. on archive).
|
// emit a "stopped working" event that notifies the parent (e.g. on archive).
|
||||||
// The parent/children link is kept so the parent can still see the child.
|
// The parent/children link is kept so the parent can still see the child.
|
||||||
this.subsessionNotifyArmed.delete(sessionId);
|
if (this.subsessionLinkForActiveChild(active.runtime.session) !== undefined) this.subsessionNotifyArmed.delete(sessionId);
|
||||||
clearSessionQueue(active.runtime.session);
|
clearSessionQueue(active.runtime.session);
|
||||||
active.unsubscribe();
|
active.unsubscribe();
|
||||||
try {
|
try {
|
||||||
@@ -1705,10 +1754,27 @@ function nonEmptyString(value: string | undefined): string | undefined {
|
|||||||
return value === undefined || value === "" ? undefined : value;
|
return value === undefined || value === "" ? undefined : value;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function subsessionHydratedParentKey(parentSessionId: string, parentSessionFile: string | undefined): string {
|
||||||
|
return `${parentSessionId}\0${parentSessionFile ?? ""}`;
|
||||||
|
}
|
||||||
|
|
||||||
function sessionPathsEqual(a: string, b: string): boolean {
|
function sessionPathsEqual(a: string, b: string): boolean {
|
||||||
return cwdPathsEqual(a, b);
|
return cwdPathsEqual(a, b);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function sessionFileMatches(session: PiAgentSession, expectedSessionFile: string | undefined): boolean {
|
||||||
|
const sessionFile = nonEmptyString(session.sessionFile);
|
||||||
|
return sessionFile !== undefined && expectedSessionFile !== undefined && sessionPathsEqual(sessionFile, expectedSessionFile);
|
||||||
|
}
|
||||||
|
|
||||||
|
function activeSessionFileMatches(active: ActiveSession<PiSessionRuntime>, expectedSessionFile: string | undefined): boolean {
|
||||||
|
return sessionFileMatches(active.runtime.session, expectedSessionFile);
|
||||||
|
}
|
||||||
|
|
||||||
|
function trackedLinkParentFileMatches(link: TrackedSubsessionLink, parentSessionFile: string): boolean {
|
||||||
|
return link.parentSessionFile !== undefined && sessionPathsEqual(link.parentSessionFile, parentSessionFile);
|
||||||
|
}
|
||||||
|
|
||||||
interface SessionHeaderSummary {
|
interface SessionHeaderSummary {
|
||||||
id: string;
|
id: string;
|
||||||
parentSession?: string;
|
parentSession?: string;
|
||||||
|
|||||||
@@ -56,9 +56,9 @@ describe("createSubsessionToolDefinitions", () => {
|
|||||||
]));
|
]));
|
||||||
const { list: listTool } = tools({ list });
|
const { list: listTool } = tools({ list });
|
||||||
|
|
||||||
const result = await listTool.execute("call-2", {}, undefined, undefined, ctxFor("parent-1", undefined));
|
const result = await listTool.execute("call-2", {}, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
|
||||||
|
|
||||||
expect(list).toHaveBeenCalledWith("parent-1");
|
expect(list).toHaveBeenCalledWith("parent-1", "/sessions/parent-1.jsonl");
|
||||||
expect(result.details).toEqual({ subsessions: [
|
expect(result.details).toEqual({ subsessions: [
|
||||||
{ sessionId: "child-1", cwd: "/repos/a", status: "working" },
|
{ sessionId: "child-1", cwd: "/repos/a", status: "working" },
|
||||||
{ sessionId: "child-2", cwd: "/repos/a", status: "idle" },
|
{ sessionId: "child-2", cwd: "/repos/a", status: "idle" },
|
||||||
@@ -76,9 +76,9 @@ describe("createSubsessionToolDefinitions", () => {
|
|||||||
const check = vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/repos/a", status: "idle" as const, finalText: "all done", messageCount: 4 }));
|
const check = vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/repos/a", status: "idle" as const, finalText: "all done", messageCount: 4 }));
|
||||||
const { check: checkTool } = tools({ check });
|
const { check: checkTool } = tools({ check });
|
||||||
|
|
||||||
const result = await checkTool.execute("call-4", { sessionId: "child-1" }, undefined, undefined, ctxFor("parent-1", undefined));
|
const result = await checkTool.execute("call-4", { sessionId: "child-1" }, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
|
||||||
|
|
||||||
expect(check).toHaveBeenCalledWith("parent-1", "child-1");
|
expect(check).toHaveBeenCalledWith("parent-1", "child-1", "/sessions/parent-1.jsonl");
|
||||||
expect(result.details).toMatchObject({ sessionId: "child-1", status: "idle", finalText: "all done" });
|
expect(result.details).toMatchObject({ sessionId: "child-1", status: "idle", finalText: "all done" });
|
||||||
expect(firstText(result.content)).toContain("all done");
|
expect(firstText(result.content)).toContain("all done");
|
||||||
});
|
});
|
||||||
@@ -99,9 +99,9 @@ describe("createSubsessionToolDefinitions", () => {
|
|||||||
}));
|
}));
|
||||||
const { read: readTool } = tools({ read });
|
const { read: readTool } = tools({ read });
|
||||||
|
|
||||||
const result = await readTool.execute("call-6", { sessionId: "child-1", roles: ["assistant"], maxChars: 200 }, undefined, undefined, ctxFor("parent-1", undefined));
|
const result = await readTool.execute("call-6", { sessionId: "child-1", roles: ["assistant"], maxChars: 200 }, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
|
||||||
|
|
||||||
expect(read).toHaveBeenCalledWith("parent-1", "child-1", { roles: ["assistant"], maxChars: 200 });
|
expect(read).toHaveBeenCalledWith("parent-1", "child-1", { roles: ["assistant"], maxChars: 200 }, "/sessions/parent-1.jsonl");
|
||||||
expect(result.details).toMatchObject({ sessionId: "child-1", matched: 1 });
|
expect(result.details).toMatchObject({ sessionId: "child-1", matched: 1 });
|
||||||
expect(firstText(result.content)).toContain("the answer");
|
expect(firstText(result.content)).toContain("the answer");
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -56,9 +56,9 @@ export interface SubsessionReadQuery {
|
|||||||
|
|
||||||
export interface SubsessionToolDeps {
|
export interface SubsessionToolDeps {
|
||||||
spawn(input: SpawnSubsessionInvocation): Promise<SpawnSubsessionResult>;
|
spawn(input: SpawnSubsessionInvocation): Promise<SpawnSubsessionResult>;
|
||||||
list(parentSessionId: string): Promise<SubsessionSummary[]>;
|
list(parentSessionId: string, parentSessionFile?: string): Promise<SubsessionSummary[]>;
|
||||||
check(parentSessionId: string, sessionId: string): Promise<SubsessionCheckResult>;
|
check(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise<SubsessionCheckResult>;
|
||||||
read(parentSessionId: string, sessionId: string, query: SubsessionReadQuery): Promise<SubsessionReadResult>;
|
read(parentSessionId: string, sessionId: string, query: SubsessionReadQuery, parentSessionFile?: string): Promise<SubsessionReadResult>;
|
||||||
}
|
}
|
||||||
|
|
||||||
const SpawnSubsessionParams = Type.Object({
|
const SpawnSubsessionParams = Type.Object({
|
||||||
@@ -196,7 +196,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
|
|||||||
parameters: ListSubsessionsParams,
|
parameters: ListSubsessionsParams,
|
||||||
async execute(_toolCallId, _params, _signal, _onUpdate, ctx) {
|
async execute(_toolCallId, _params, _signal, _onUpdate, ctx) {
|
||||||
const parentSessionId = ctx.sessionManager.getSessionId();
|
const parentSessionId = ctx.sessionManager.getSessionId();
|
||||||
const subsessions = await deps.list(parentSessionId);
|
const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
|
||||||
|
const subsessions = await deps.list(parentSessionId, parentSessionFile);
|
||||||
const text = subsessions.length === 0
|
const text = subsessions.length === 0
|
||||||
? "You have not spawned any subsessions."
|
? "You have not spawned any subsessions."
|
||||||
: `Your subsessions:\n${subsessions.map(statusLine).join("\n")}`;
|
: `Your subsessions:\n${subsessions.map(statusLine).join("\n")}`;
|
||||||
@@ -212,7 +213,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
|
|||||||
parameters: CheckSubsessionParams,
|
parameters: CheckSubsessionParams,
|
||||||
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
const parentSessionId = ctx.sessionManager.getSessionId();
|
const parentSessionId = ctx.sessionManager.getSessionId();
|
||||||
const result = await deps.check(parentSessionId, params.sessionId);
|
const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
|
||||||
|
const result = await deps.check(parentSessionId, params.sessionId, parentSessionFile);
|
||||||
const body = result.finalText === "" ? "(no output yet)" : result.finalText;
|
const body = result.finalText === "" ? "(no output yet)" : result.finalText;
|
||||||
return {
|
return {
|
||||||
content: [{ type: "text", text: `Subsession ${result.sessionId} [${result.status}]:\n\n${body}` }],
|
content: [{ type: "text", text: `Subsession ${result.sessionId} [${result.status}]:\n\n${body}` }],
|
||||||
@@ -229,8 +231,9 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
|
|||||||
parameters: ReadSubsessionParams,
|
parameters: ReadSubsessionParams,
|
||||||
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
const parentSessionId = ctx.sessionManager.getSessionId();
|
const parentSessionId = ctx.sessionManager.getSessionId();
|
||||||
|
const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
|
||||||
const { sessionId, ...query } = params;
|
const { sessionId, ...query } = params;
|
||||||
const result = await deps.read(parentSessionId, sessionId, query);
|
const result = await deps.read(parentSessionId, sessionId, query, parentSessionFile);
|
||||||
return {
|
return {
|
||||||
content: [{ type: "text", text: renderTranscript(result) }],
|
content: [{ type: "text", text: renderTranscript(result) }],
|
||||||
details: result,
|
details: result,
|
||||||
|
|||||||
Reference in New Issue
Block a user