diff --git a/.changeset/git-inline-diff-highlights.md b/.changeset/git-inline-diff-highlights.md new file mode 100644 index 0000000..26e3890 --- /dev/null +++ b/.changeset/git-inline-diff-highlights.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Highlight within-line changes in the Git diff viewer. diff --git a/.changeset/persist-subsession-links.md b/.changeset/persist-subsession-links.md new file mode 100644 index 0000000..8199eaa --- /dev/null +++ b/.changeset/persist-subsession-links.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Persist tracked subsession links in session history so parents can list, check, and read child sessions after the session daemon restarts, and reopened children can resume parent notifications. diff --git a/src/client/src/components/UnifiedDiffViewer.ts b/src/client/src/components/UnifiedDiffViewer.ts new file mode 100644 index 0000000..3844f52 --- /dev/null +++ b/src/client/src/components/UnifiedDiffViewer.ts @@ -0,0 +1,59 @@ +import { LitElement, css, html, type TemplateResult } from "lit"; +import { customElement, property } from "lit/decorators.js"; +import { parseUnifiedDiff, type UnifiedDiffLine, type UnifiedDiffTextSpan } from "../diff/unifiedDiff"; + +@customElement("unified-diff-viewer") +export class UnifiedDiffViewer extends LitElement { + @property() diff = ""; + + override render(): TemplateResult { + const lines = parseUnifiedDiff(this.diff); + if (lines.length === 0) return html`

No diff.

`; + return html` +
+
+ ${lines.map((line) => this.renderLine(line))} +
+
+ `; + } + + private renderLine(line: UnifiedDiffLine): TemplateResult { + const kindClass = line.kind; + return html` +
+ ${formatLineNumber(line.oldLineNumber)} + ${formatLineNumber(line.newLineNumber)} + ${line.prefix} + ${renderSpans(line.spans)} +
+ `; + } + + static override styles = css` + :host { display: block; min-height: 0; height: 100%; color: var(--pi-text); background: var(--pi-bg); } + .empty { box-sizing: border-box; margin: 0; padding: 10px; color: var(--pi-muted); } + .scroller { height: 100%; min-height: 0; overflow: auto; background: var(--pi-bg); } + .diff-grid { display: grid; grid-template-columns: max-content max-content 2ch max-content; width: max-content; min-width: 100%; padding: 6px 0; font: 12px ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; line-height: 1.45; } + .line { display: contents; } + .cell { min-height: 1.45em; white-space: pre; } + .line-number { min-width: 4ch; padding: 0 8px; border-right: 1px solid var(--pi-border-muted); color: var(--pi-dim); text-align: right; user-select: none; } + .prefix { padding: 0 4px; color: var(--pi-dim); text-align: center; user-select: none; } + .content { padding: 0 12px 0 4px; } + .meta { color: var(--pi-dim); } + .hunk { background: color-mix(in srgb, var(--pi-accent) 9%, transparent); color: var(--pi-accent); } + .add { background: color-mix(in srgb, var(--pi-success) 12%, transparent); } + .remove { background: color-mix(in srgb, var(--pi-danger) 12%, transparent); } + .marker { color: var(--pi-dim); } + .content.add .inline-change { border-radius: 2px; background: color-mix(in srgb, var(--pi-success) 36%, transparent); color: var(--pi-text); } + .content.remove .inline-change { border-radius: 2px; background: color-mix(in srgb, var(--pi-danger) 36%, transparent); color: var(--pi-text); } + `; +} + +function renderSpans(spans: UnifiedDiffTextSpan[]): TemplateResult[] { + return spans.map((span) => html`${span.text}`); +} + +function formatLineNumber(lineNumber: number | undefined): string { + return lineNumber === undefined ? "" : String(lineNumber); +} diff --git a/src/client/src/components/shared.ts b/src/client/src/components/shared.ts index 91165e5..fde76ff 100644 --- a/src/client/src/components/shared.ts +++ b/src/client/src/components/shared.ts @@ -200,7 +200,7 @@ export const workspacePanelStyles = css` .diff-section:last-child { border-bottom: 0; } .viewer-header { position: sticky; top: 0; display: flex; justify-content: space-between; gap: 8px; padding: 8px; border-bottom: 1px solid var(--pi-border-muted); background: var(--pi-bg); } .viewer-header strong { min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } - code-viewer { flex: 1 1 auto; min-height: 0; } + code-viewer, unified-diff-viewer { flex: 1 1 auto; min-height: 0; } .image-preview { flex: 1 1 auto; min-height: 0; box-sizing: border-box; display: flex; align-items: center; justify-content: center; overflow: auto; padding: 16px; } .image-preview img { display: block; max-width: 100%; max-height: 100%; object-fit: contain; border: 1px solid var(--pi-border-muted); border-radius: 8px; background-color: var(--pi-surface); background-image: linear-gradient(45deg, color-mix(in srgb, var(--pi-border-muted) 45%, transparent) 25%, transparent 25%), linear-gradient(-45deg, color-mix(in srgb, var(--pi-border-muted) 45%, transparent) 25%, transparent 25%), linear-gradient(45deg, transparent 75%, color-mix(in srgb, var(--pi-border-muted) 45%, transparent) 75%), linear-gradient(-45deg, transparent 75%, color-mix(in srgb, var(--pi-border-muted) 45%, transparent) 75%); background-position: 0 0, 0 8px, 8px -8px, -8px 0; background-size: 16px 16px; box-shadow: 0 8px 24px var(--pi-shadow-soft); } pre { margin: 0; padding: 10px; overflow: auto; font: 12px ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; line-height: 1.45; white-space: pre-wrap; overflow-wrap: anywhere; } diff --git a/src/client/src/diff/unifiedDiff.test.ts b/src/client/src/diff/unifiedDiff.test.ts new file mode 100644 index 0000000..bdebc9b --- /dev/null +++ b/src/client/src/diff/unifiedDiff.test.ts @@ -0,0 +1,117 @@ +import { describe, expect, it } from "vitest"; +import { parseUnifiedDiff, type UnifiedDiffLine, type UnifiedDiffLineKind } from "./unifiedDiff"; + +describe("parseUnifiedDiff", () => { + it("computes inline spans for paired removed and added lines", () => { + const diff = [ + "diff --git a/src/app.ts b/src/app.ts", + "index 1111111..2222222 100644", + "--- a/src/app.ts", + "+++ b/src/app.ts", + "@@ -10,2 +10,2 @@ export function demo() {", + "- const name = \"fooBar\";", + "+ const name = \"fooBaz\";", + " return name;", + ].join("\n"); + + const lines = parseUnifiedDiff(diff); + const removed = firstLineOfKind(lines, "remove"); + const added = firstLineOfKind(lines, "add"); + const context = firstLineOfKind(lines, "context"); + + expect(removed.oldLineNumber).toBe(10); + expect(removed.newLineNumber).toBeUndefined(); + expect(changedText(removed)).toEqual(["r"]); + expect(added.oldLineNumber).toBeUndefined(); + expect(added.newLineNumber).toBe(10); + expect(changedText(added)).toEqual(["z"]); + expect(context.oldLineNumber).toBe(11); + expect(context.newLineNumber).toBe(11); + }); + + it("keeps file headers as metadata before a hunk starts", () => { + const diff = [ + "diff --git a/README.md b/README.md", + "index 1111111..2222222 100644", + "--- a/README.md", + "+++ b/README.md", + "@@ -1 +1 @@", + "-old", + "+new", + ].join("\n"); + + expect(parseUnifiedDiff(diff).slice(0, 5).map((line) => line.kind)).toEqual(["meta", "meta", "meta", "meta", "hunk"]); + }); + + it("parses changed content that starts with file header markers inside hunks", () => { + const diff = [ + "diff --git a/README.md b/README.md", + "--- a/README.md", + "+++ b/README.md", + "@@ -1 +1 @@", + "---- removed heading", + "++++ added heading", + ].join("\n"); + + const removed = firstLineOfKind(parseUnifiedDiff(diff), "remove"); + const added = firstLineOfKind(parseUnifiedDiff(diff), "add"); + + expect(removed.text).toBe("--- removed heading"); + expect(added.text).toBe("+++ added heading"); + }); + + it("pairs a single removed line with the closest added line in uneven blocks", () => { + const diff = [ + "diff --git a/src/app.ts b/src/app.ts", + "--- a/src/app.ts", + "+++ b/src/app.ts", + "@@ -1 +1,2 @@", + "-const label = \"old\";", + "+const label = \"new\";", + "+const extra = true;", + ].join("\n"); + + const addedLines = linesOfKind(parseUnifiedDiff(diff), "add"); + const firstAdded = lineAt(addedLines, 0); + const secondAdded = lineAt(addedLines, 1); + + expect(changedText(firstAdded)).toEqual(["new"]); + expect(secondAdded.spans.every((span) => !span.changed)).toBe(true); + }); + + it("leaves pure additions without inline change spans", () => { + const diff = [ + "diff --git a/new.txt b/new.txt", + "new file mode 100644", + "--- /dev/null", + "+++ b/new.txt", + "@@ -0,0 +1 @@", + "+brand new", + ].join("\n"); + + const added = firstLineOfKind(parseUnifiedDiff(diff), "add"); + + expect(added.newLineNumber).toBe(1); + expect(added.spans).toEqual([{ text: "brand new", changed: false }]); + }); +}); + +function firstLineOfKind(lines: UnifiedDiffLine[], kind: UnifiedDiffLineKind): UnifiedDiffLine { + const found = lines.find((line) => line.kind === kind); + if (found === undefined) throw new Error(`Missing ${kind} line`); + return found; +} + +function linesOfKind(lines: UnifiedDiffLine[], kind: UnifiedDiffLineKind): UnifiedDiffLine[] { + return lines.filter((line) => line.kind === kind); +} + +function lineAt(lines: UnifiedDiffLine[], index: number): UnifiedDiffLine { + const line = lines[index]; + if (line === undefined) throw new Error(`Missing line at ${String(index)}`); + return line; +} + +function changedText(line: UnifiedDiffLine): string[] { + return line.spans.filter((span) => span.changed).map((span) => span.text); +} diff --git a/src/client/src/diff/unifiedDiff.ts b/src/client/src/diff/unifiedDiff.ts new file mode 100644 index 0000000..7a5a3c8 --- /dev/null +++ b/src/client/src/diff/unifiedDiff.ts @@ -0,0 +1,224 @@ +import { diffChars } from "diff"; + +export type UnifiedDiffLineKind = "meta" | "hunk" | "context" | "add" | "remove" | "marker"; + +export interface UnifiedDiffTextSpan { + text: string; + changed: boolean; +} + +export interface UnifiedDiffLine { + kind: UnifiedDiffLineKind; + prefix: string; + text: string; + spans: UnifiedDiffTextSpan[]; + oldLineNumber?: number; + newLineNumber?: number; +} + +interface InlineDiffResult { + removed: UnifiedDiffTextSpan[]; + added: UnifiedDiffTextSpan[]; +} + +interface DiffLinePair { + removed: UnifiedDiffLine; + added: UnifiedDiffLine; +} + +const hunkHeaderPattern = /^@@ -(\d+)(?:,\d+)? \+(\d+)(?:,\d+)? @@/; +const maxInlineLineLength = 5_000; +const maxInlineBlockLines = 20; +const minInlineSimilarity = 0.20; +const minPairSimilarity = 0.25; + +export function parseUnifiedDiff(diff: string): UnifiedDiffLine[] { + const parsedLines = parseUnifiedDiffLines(diff); + applyInlineDiffs(parsedLines); + return parsedLines; +} + +function parseUnifiedDiffLines(diff: string): UnifiedDiffLine[] { + const lines = splitDiffLines(diff); + const parsedLines: UnifiedDiffLine[] = []; + let oldLineNumber: number | undefined; + let newLineNumber: number | undefined; + + for (const rawLine of lines) { + const hunkMatch = hunkHeaderPattern.exec(rawLine); + if (hunkMatch !== null) { + oldLineNumber = Number(hunkMatch[1]); + newLineNumber = Number(hunkMatch[2]); + parsedLines.push(line("hunk", "", rawLine)); + continue; + } + + if (oldLineNumber !== undefined && newLineNumber !== undefined) { + if (rawLine.startsWith("+")) { + parsedLines.push(line("add", "+", rawLine.slice(1), { newLineNumber })); + newLineNumber++; + continue; + } + if (rawLine.startsWith("-")) { + parsedLines.push(line("remove", "-", rawLine.slice(1), { oldLineNumber })); + oldLineNumber++; + continue; + } + if (rawLine.startsWith(" ")) { + parsedLines.push(line("context", " ", rawLine.slice(1), { oldLineNumber, newLineNumber })); + oldLineNumber++; + newLineNumber++; + continue; + } + if (rawLine.startsWith("\\")) { + parsedLines.push(line("marker", "", rawLine)); + continue; + } + } + + oldLineNumber = undefined; + newLineNumber = undefined; + parsedLines.push(line("meta", "", rawLine)); + } + + return parsedLines; +} + +function splitDiffLines(diff: string): string[] { + if (diff === "") return []; + const lines = diff.replace(/\r\n/g, "\n").replace(/\r/g, "\n").split("\n"); + if (lines.at(-1) === "") lines.pop(); + return lines; +} + +function line(kind: UnifiedDiffLineKind, prefix: string, text: string, numbers: { oldLineNumber?: number; newLineNumber?: number } = {}): UnifiedDiffLine { + return { + kind, + prefix, + text, + spans: text === "" ? [] : [{ text, changed: false }], + ...numbers, + }; +} + +function applyInlineDiffs(lines: UnifiedDiffLine[]): void { + let index = 0; + while (index < lines.length) { + const current = lines[index]; + if (current?.kind !== "remove") { + index++; + continue; + } + + const removedStart = index; + while (lines[index]?.kind === "remove") index++; + const addedStart = index; + while (lines[index]?.kind === "add") index++; + + if (addedStart === index) continue; + const removedLines = lines.slice(removedStart, addedStart); + const addedLines = lines.slice(addedStart, index); + applyInlineDiffBlock(removedLines, addedLines); + } +} + +function applyInlineDiffBlock(removedLines: UnifiedDiffLine[], addedLines: UnifiedDiffLine[]): void { + if (removedLines.length + addedLines.length > maxInlineBlockLines) return; + for (const pair of pairChangedLines(removedLines, addedLines)) { + const inlineDiff = computeInlineDiff(pair.removed.text, pair.added.text); + if (inlineDiff === undefined) continue; + pair.removed.spans = inlineDiff.removed; + pair.added.spans = inlineDiff.added; + } +} + +function pairChangedLines(removedLines: UnifiedDiffLine[], addedLines: UnifiedDiffLine[]): DiffLinePair[] { + if (removedLines.length === addedLines.length) return removedLines.map((removed, index) => ({ removed, added: addedLines[index] })).filter(isCompletePair); + if (removedLines.length === 1) return bestPairsForSingleRemovedLine(removedLines[0], addedLines); + if (addedLines.length === 1) return bestPairsForSingleAddedLine(removedLines, addedLines[0]); + + const pairs: DiffLinePair[] = []; + const pairCount = Math.min(removedLines.length, addedLines.length); + for (let index = 0; index < pairCount; index++) { + const removed = removedLines[index]; + const added = addedLines[index]; + if (removed === undefined || added === undefined) continue; + if (lineSimilarity(removed.text, added.text) >= minPairSimilarity) pairs.push({ removed, added }); + } + return pairs; +} + +function isCompletePair(pair: { removed: UnifiedDiffLine; added: UnifiedDiffLine | undefined }): pair is DiffLinePair { + return pair.added !== undefined; +} + +function bestPairsForSingleRemovedLine(removed: UnifiedDiffLine | undefined, addedLines: UnifiedDiffLine[]): DiffLinePair[] { + if (removed === undefined) return []; + const added = bestMatchingLine(removed.text, addedLines); + return added === undefined ? [] : [{ removed, added }]; +} + +function bestPairsForSingleAddedLine(removedLines: UnifiedDiffLine[], added: UnifiedDiffLine | undefined): DiffLinePair[] { + if (added === undefined) return []; + const removed = bestMatchingLine(added.text, removedLines); + return removed === undefined ? [] : [{ removed, added }]; +} + +function bestMatchingLine(text: string, candidates: UnifiedDiffLine[]): UnifiedDiffLine | undefined { + let bestCandidate: UnifiedDiffLine | undefined; + let bestScore = minPairSimilarity; + for (const candidate of candidates) { + const score = lineSimilarity(text, candidate.text); + if (score <= bestScore) continue; + bestCandidate = candidate; + bestScore = score; + } + return bestCandidate; +} + +function computeInlineDiff(oldText: string, newText: string): InlineDiffResult | undefined { + if (oldText === newText) return undefined; + if (oldText.length > maxInlineLineLength || newText.length > maxInlineLineLength) return undefined; + + const changes = diffChars(oldText, newText); + const similarity = similarityFromChanges(changes, oldText, newText); + if (Math.max(oldText.length, newText.length) >= 20 && similarity < minInlineSimilarity) return undefined; + + const removed: UnifiedDiffTextSpan[] = []; + const added: UnifiedDiffTextSpan[] = []; + for (const change of changes) { + if (change.value === "") continue; + if (change.added) added.push({ text: change.value, changed: true }); + else if (change.removed) removed.push({ text: change.value, changed: true }); + else { + removed.push({ text: change.value, changed: false }); + added.push({ text: change.value, changed: false }); + } + } + + if (!removed.some((span) => span.changed) && !added.some((span) => span.changed)) return undefined; + return { removed: mergeAdjacentSpans(removed), added: mergeAdjacentSpans(added) }; +} + +function lineSimilarity(oldText: string, newText: string): number { + if (oldText === newText) return 1; + if (oldText.length > maxInlineLineLength || newText.length > maxInlineLineLength) return 0; + return similarityFromChanges(diffChars(oldText, newText), oldText, newText); +} + +function similarityFromChanges(changes: ReturnType, oldText: string, newText: string): number { + const maxLength = Math.max(oldText.length, newText.length); + if (maxLength === 0) return 1; + const unchangedLength = changes.reduce((total, change) => change.added || change.removed ? total : total + change.value.length, 0); + return unchangedLength / maxLength; +} + +function mergeAdjacentSpans(spans: UnifiedDiffTextSpan[]): UnifiedDiffTextSpan[] { + const merged: UnifiedDiffTextSpan[] = []; + for (const span of spans) { + const previous = merged[merged.length - 1]; + if (previous?.changed === span.changed) previous.text += span.text; + else merged.push({ ...span }); + } + return merged; +} diff --git a/src/client/src/plugins/core/panels.ts b/src/client/src/plugins/core/panels.ts index 8ba9ed8..0f0d2a3 100644 --- a/src/client/src/plugins/core/panels.ts +++ b/src/client/src/plugins/core/panels.ts @@ -146,11 +146,11 @@ function renderDiffViewer(context: WorkspacePanelContext): TemplateResult { } function renderDiffSection(diff: GitDiffResponse): TemplateResult { - loadCodeViewer(); + loadUnifiedDiffViewer(); return html`
${diff.path ?? "diff"}${diff.staged ? "staged" : "unstaged"}${diff.truncated ? " ยท truncated" : ""}
- +
`; } @@ -159,6 +159,10 @@ function loadCodeViewer(): void { void import("../../components/CodeViewer"); } +function loadUnifiedDiffViewer(): void { + void import("../../components/UnifiedDiffViewer"); +} + function loadTerminalPanel(): void { void import("../../components/TerminalPanel"); } diff --git a/src/server/sessions/piSessionService.test.ts b/src/server/sessions/piSessionService.test.ts index 9de755f..4c54fde 100644 --- a/src/server/sessions/piSessionService.test.ts +++ b/src/server/sessions/piSessionService.test.ts @@ -1,3 +1,6 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent"; import { describe, expect, it, vi } from "vitest"; import type { GlobalSessionEvent, SessionUiEvent } from "../../shared/apiTypes.js"; @@ -32,11 +35,12 @@ interface TestSession extends PiAgentSession { getFollowUpMessages: () => readonly string[]; } -function fakeSessionManager(cwd = "/workspace"): PiSessionManager { +function fakeSessionManager(cwd = "/workspace", patch: Partial = {}): PiSessionManager { return { getCwd: () => cwd, getBranch: () => [], getLeafId: () => "leaf-1", + ...patch, }; } @@ -141,6 +145,16 @@ function sessionGateway(records: ReturnType[]): SessionGat }; } +function emptyArchiveStore(): NonNullable { + return { + list: () => Promise.resolve([]), + get: () => Promise.resolve(undefined), + archive: () => Promise.reject(new Error("archive should not be called")), + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + }; +} + describe("PiSessionService", () => { it("starts sessions through an injected runtime creator", async () => { const hub = new CapturingSessionEventHub(); @@ -887,6 +901,668 @@ describe("PiSessionService", () => { await service.dispose(); }); + it("persists tracked child links in the parent and child sessions", async () => { + const parentPersisted: { customType: string; data?: unknown }[] = []; + const childPersisted: { customType: string; data?: unknown }[] = []; + const parent = fakeRuntime("parent-1", { + sessionFile: "/tmp/parent-1.jsonl", + sessionManager: fakeSessionManager("/workspace", { + appendCustomEntry: (customType, data) => { + parentPersisted.push({ customType, data }); + return "parent-entry-1"; + }, + }), + }); + const child = fakeRuntime("child-1", { + sessionFile: "/tmp/child-1.jsonl", + sessionManager: fakeSessionManager("/workspace-feature", { + appendCustomEntry: (customType, data) => { + childPersisted.push({ customType, data }); + return "child-entry-1"; + }, + }), + }); + const runtimes = [parent.runtime, child.runtime]; + let index = 0; + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? child.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: sessionGateway([]), + archiveStore: emptyArchiveStore(), + spawnTargets: { resolveSpawnTarget: () => Promise.resolve({ allowed: true, cwd: "/workspace-feature" }) }, + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + await service.spawnSubsession({ spawningCwd: "/workspace", parentSessionId: "parent-1", parentSessionFile: "/tmp/parent-1.jsonl", prompt: "do the slice", cwd: "/workspace-feature" }); + + expect(parentPersisted).toEqual([ + { + customType: "pi-web.subsession.link", + data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: "/tmp/child-1.jsonl", cwd: "/workspace-feature" }, + }, + ]); + expect(childPersisted).toEqual([ + { + customType: "pi-web.subsession.spawned", + data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" }, + }, + ]); + await service.dispose(); + }); + + it("hydrates persisted child links after a service restart so the parent can inspect them", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-")); + const parentFile = join(tempDir, "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(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", { + getBranch: () => [{ type: "message", message: { role: "assistant", content: "finished" } }], + }); + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: 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 child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const runtimes = [parent.runtime, child.runtime]; + let index = 0; + const open = vi.fn(() => childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? child.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.checkSubsession("parent-1", "child-1")).resolves.toEqual({ + sessionId: "child-1", + cwd: "/workspace-feature", + status: "idle", + finalText: "finished", + messageCount: 1, + }); + expect(open).toHaveBeenCalledWith(childFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("ignores stale persisted child links when the child no longer records the parent", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-stale-")); + const parentFile = join(tempDir, "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(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature" })}\n`, "utf8"); + + try { + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: 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 service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("does not hydrate persisted links when the exact child file is unavailable", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: "/sessions/child-1.jsonl", cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("does not hydrate parent links without a child file", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child", cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("does not invent subsession links from existing child session headers", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const childRecord = { ...sessionRecord("child-1", "/workspace-feature"), path: "/sessions/child-1.jsonl", parentSessionPath: parentFile }; + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { getEntries: () => [] }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([childRecord]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("does not hydrate copied parent links when the opened parent has a different id", async () => { + const forkedParent = fakeRuntime("parent-fork-1", { + sessionFile: "/sessions/parent-fork-1.jsonl", + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: "/sessions/child-1.jsonl", cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(forkedParent.runtime), + sessionManager: { create: () => forkedParent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-fork-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("relinks a spawned child when the child session is opened after restart", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-open-child-")); + const parentFile = join(tempDir, "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(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", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + 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 child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager }); + const runtimes = [child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => path === parentFile ? parentManager : childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...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")); + 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(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("notifies the validated parent file instead of an active prefix-matched parent id", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-prefix-parent-")); + const parentFile = join(tempDir, "parent.jsonl"); + const forkParentFile = join(tempDir, "parent-fork.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(forkParentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1-fork", 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", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + 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 forkManager = fakeSessionManager("/workspace"); + const fork = fakeRuntime("parent-1-fork", { sessionFile: forkParentFile, sessionManager: forkManager }); + const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager }); + const runtimes = [fork.runtime, child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => { + if (path === parentFile) return parentManager; + if (path === forkParentFile) return forkManager; + return childManager; + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => forkManager, + list: (cwd: string) => Promise.resolve(cwd === "/workspace" + ? [{ ...sessionRecord("parent-1-fork", "/workspace"), path: forkParentFile }] + : [{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]), + listAll: () => Promise.resolve([]), + open, + }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("parent-1-fork", "/workspace")); + await service.status(sessionRef("child-1", "/workspace-feature")); + 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(fork.calls.sendCustomMessage).toHaveLength(0); + expect(parent.calls.sendCustomMessage).toHaveLength(1); + expect(open).toHaveBeenCalledWith(parentFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("does not relink a copied child with the original session id unless the parent link names the current child file", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-copied-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 childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + 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 child = fakeRuntime("child-1", { sessionFile: copiedChildFile, sessionManager: childManager }); + const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager }); + const runtimes = [child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => path === parentFile ? parentManager : childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...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")); + 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(parent.calls.sendCustomMessage).toHaveLength(0); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + 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 () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-stale-child-header-")); + const parentFile = join(tempDir, "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(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature" })}\n`, "utf8"); + + try { + const childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + 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 child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager }); + const runtimes = [child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => path === parentFile ? parentManager : childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]), + listAll: () => Promise.resolve([]), + open, + }, + archiveStore: { + ...emptyArchiveStore(), + get: (sessionId) => Promise.resolve(sessionId === "child-1" ? { sessionId: "child-1", cwd: "/workspace-feature", archivedAt: "2026-01-01T00:00:00.000Z", parentSessionPath: parentFile } : undefined), + }, + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("child-1", "/workspace-feature")); + 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(parent.calls.sendCustomMessage).toHaveLength(0); + expect(open).not.toHaveBeenCalledWith(parentFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("does not relink a child marker when the child header points at a different parent id", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-wrong-parent-")); + const mismatchedParentFile = join(tempDir, "other-parent.jsonl"); + const actualParentFile = join(tempDir, "parent.jsonl"); + const childFile = join(tempDir, "child.jsonl"); + await writeFile(mismatchedParentFile, `${JSON.stringify({ type: "session", version: 3, id: "other-parent", 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: mismatchedParentFile })}\n`, "utf8"); + + try { + const childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: mismatchedParentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + const parent = fakeRuntime("parent-1", { sessionFile: actualParentFile, sessionManager: fakeSessionManager("/workspace") }); + const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const runtimes = [child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => path === actualParentFile ? parent.session.sessionManager : childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: mismatchedParentFile }]), + listAll: () => Promise.resolve([{ ...sessionRecord("parent-1", "/workspace"), path: actualParentFile }]), + open, + }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("child-1", "/workspace-feature")); + 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(parent.calls.sendCustomMessage).toHaveLength(0); + expect(open).not.toHaveBeenCalledWith(actualParentFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("does not relink copied child markers when the opened child has a different id", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const childFile = "/sessions/child-fork-1.jsonl"; + const childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + const child = fakeRuntime("child-fork-1", { sessionFile: childFile, sessionManager: childManager }); + const open = vi.fn(() => childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(child.runtime), + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...sessionRecord("child-fork-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]), + listAll: () => Promise.resolve([]), + open, + }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("child-fork-1", "/workspace-feature")); + 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(open).not.toHaveBeenCalledWith(parentFile); + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + it("notifies the parent once when the tracked child stops working", async () => { const { parent, child, service } = subsessionService({ allowed: true, cwd: "/workspace-feature" }); await service.start("/workspace"); @@ -947,7 +1623,7 @@ describe("PiSessionService", () => { await service.dispose(); }); - it("reports an archived child's status in the subsession list", async () => { + it("reports a missing tracked child file as unknown in the subsession list", async () => { const { service } = subsessionService({ allowed: true, cwd: "/workspace-feature" }); await service.start("/workspace"); await service.spawnSubsession({ spawningCwd: "/workspace", parentSessionId: "parent-1", parentSessionFile: "/tmp/parent-1.jsonl", prompt: "go", cwd: "/workspace-feature" }); @@ -955,7 +1631,7 @@ describe("PiSessionService", () => { await service.archive("child-1"); await expect(service.listSubsessions("parent-1")).resolves.toEqual([ - { sessionId: "child-1", cwd: "/workspace-feature", status: "archived" }, + { sessionId: "child-1", cwd: "/workspace-feature", status: "unknown" }, ]); await service.dispose(); }); diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index 3562925..35047b9 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -1,4 +1,4 @@ -import { readFile, writeFile } from "node:fs/promises"; +import { open, readFile, writeFile } from "node:fs/promises"; import type { Api, ImageContent, Model } from "@earendil-works/pi-ai"; import { AuthStorage, @@ -81,6 +81,26 @@ interface QueuedPrompt { echoUserMessage?: boolean; } +interface TrackedSubsessionLink { + parentSessionId: string; + childSessionId: string; + childSessionFile?: string; + parentSessionFile?: string; + cwd?: string; +} + +interface PersistedParentSubsessionLink { + spawnedBySessionId: string; + spawnedSessionId: string; + spawnedSessionFile?: string; + cwd?: string; +} + +interface PersistedChildSubsessionLink { + spawnedBySessionId: string; + spawnedSessionId: string; +} + function requirePromptText(value: unknown): string { if (typeof value !== "string") throw new Error("Prompt text is required"); return value; @@ -123,8 +143,10 @@ type ModelRegistryInstance = ReturnType; export interface PiSessionManager { getCwd(): string; getBranch(): unknown[]; + getEntries?(): readonly unknown[]; getLeafId(): string | null; getHeader?(): { parentSession?: string } | null | undefined; + appendCustomEntry?(customType: string, data?: unknown): string; } export interface PiSessionManagerGateway { @@ -290,6 +312,10 @@ export class PiSessionService { private readonly subsessionParents = new Map(); /** Parent session id -> the set of tracked subsession ids it spawned. */ private readonly subsessionChildren = new Map>(); + /** Tracked subsession id -> persisted recovery details for the child. */ + private readonly subsessionLinks = new Map(); + /** Parent id/file identities whose persisted links have already been loaded. */ + private readonly subsessionHydratedParents = new Set(); /** * Tracked subsession id -> whether a completion notification is armed. * Armed when the child starts working; firing on completion disarms it so a @@ -322,9 +348,9 @@ export class PiSessionService { this.spawnTargets === undefined ? undefined : (input) => this.spawnSession(input), !subsessionsActive ? undefined : { spawn: (input) => this.spawnSubsession(input), - list: (parentSessionId) => this.listSubsessions(parentSessionId), - check: (parentSessionId, sessionId) => this.checkSubsession(parentSessionId, sessionId), - read: (parentSessionId, sessionId, query) => this.readSubsession(parentSessionId, sessionId, query), + list: (parentSessionId, parentSessionFile) => this.listSubsessions(parentSessionId, parentSessionFile), + check: (parentSessionId, sessionId, parentSessionFile) => this.checkSubsession(parentSessionId, sessionId, parentSessionFile), + read: (parentSessionId, sessionId, query, parentSessionFile) => this.readSubsession(parentSessionId, sessionId, query, parentSessionFile), }, ); this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime; @@ -362,6 +388,8 @@ export class PiSessionService { this.authLossWarnings.clear(); this.subsessionParents.clear(); this.subsessionChildren.clear(); + this.subsessionLinks.clear(); + this.subsessionHydratedParents.clear(); this.subsessionNotifyArmed.clear(); await Promise.all(activeSessions.map(async (active) => { active.unsubscribe(); @@ -439,7 +467,17 @@ export class PiSessionService { const decision = await this.spawnTargets.resolveSpawnTarget(input.spawningCwd, input.cwd); if (!decision.allowed) throw spawnTargetError(decision); const created = await this.start(decision.cwd, input.parentSessionFile); - this.registerSubsession(input.parentSessionId, created.id); + const parentSessionFile = nonEmptyString(input.parentSessionFile); + const link: TrackedSubsessionLink = { + parentSessionId: input.parentSessionId, + childSessionId: created.id, + ...(created.path === "" ? {} : { childSessionFile: created.path }), + ...(parentSessionFile === undefined ? {} : { parentSessionFile }), + cwd: decision.cwd, + }; + this.registerVerifiedSubsession(link); + this.persistSubsessionLink(link); + this.persistSubsessionChildMarker(input.parentSessionId, created.id); await this.prompt(created.id, input.prompt); this.logger.info( { parentSessionId: input.parentSessionId, sessionId: created.id, cwd: decision.cwd, promptLength: input.prompt.length }, @@ -449,65 +487,272 @@ export class PiSessionService { } /** Summaries of the tracked subsessions spawned by `parentSessionId`. */ - async listSubsessions(parentSessionId: string): Promise { + async listSubsessions(parentSessionId: string, parentSessionFile?: string): Promise { + const parentFile = nonEmptyString(parentSessionFile); + await this.hydrateSubsessionsForParent(parentSessionId, parentFile); const childIds = this.subsessionChildren.get(parentSessionId); 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. */ - async checkSubsession(parentSessionId: string, sessionId: string): Promise { - const session = await this.openSubsession(parentSessionId, sessionId); + async checkSubsession(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise { + const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile); const messages = historyMessages(session); return { sessionId, cwd: session.sessionManager.getCwd(), - status: await this.subsessionStatus(session), + status: this.subsessionStatus(session), finalText: finalAssistantText(messages), messageCount: messages.length, }; } /** Filtered, paginated transcript of a subsession, scoped to the caller's children. */ - async readSubsession(parentSessionId: string, sessionId: string, query: SubsessionReadQuery): Promise { - const session = await this.openSubsession(parentSessionId, sessionId); + async readSubsession(parentSessionId: string, sessionId: string, query: SubsessionReadQuery, parentSessionFile?: string): Promise { + const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile); const view = buildTranscriptView(historyMessages(session), query); return { sessionId, cwd: session.sessionManager.getCwd(), - status: await this.subsessionStatus(session), + status: this.subsessionStatus(session), ...view, }; } /** Open a session after verifying it is one of the caller's tracked children. */ - private async openSubsession(parentSessionId: string, sessionId: string): Promise { - if (this.subsessionParents.get(sessionId) !== parentSessionId) { + private async openSubsession(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise { + const parentFile = nonEmptyString(parentSessionFile); + 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`); } - return this.getOrOpen(sessionId); + return this.getOrOpenTrackedSubsession(sessionId); } - private registerSubsession(parentSessionId: string, childSessionId: string): void { + 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 | 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 | 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 { + const { childSessionId, parentSessionId } = link; + const previousParentId = this.subsessionParents.get(childSessionId); + if (previousParentId !== undefined && previousParentId !== parentSessionId) { + const previousChildren = this.subsessionChildren.get(previousParentId); + previousChildren?.delete(childSessionId); + if (previousChildren?.size === 0) this.subsessionChildren.delete(previousParentId); + } + this.subsessionParents.set(childSessionId, parentSessionId); const children = this.subsessionChildren.get(parentSessionId) ?? new Set(); children.add(childSessionId); this.subsessionChildren.set(parentSessionId, children); - this.subsessionNotifyArmed.set(childSessionId, false); + + this.subsessionLinks.set(childSessionId, link); + if (!this.subsessionNotifyArmed.has(childSessionId)) this.subsessionNotifyArmed.set(childSessionId, false); + } + + private unregisterSubsession(childSessionId: string): void { + const parentSessionId = this.subsessionParents.get(childSessionId); + this.subsessionParents.delete(childSessionId); + this.subsessionLinks.delete(childSessionId); + this.subsessionNotifyArmed.delete(childSessionId); + if (parentSessionId === undefined) return; + const children = this.subsessionChildren.get(parentSessionId); + children?.delete(childSessionId); + if (children?.size === 0) this.subsessionChildren.delete(parentSessionId); + } + + private persistSubsessionLink(link: TrackedSubsessionLink): void { + const parent = this.activeParentForSubsessionLink(link)?.runtime.session; + if (parent === undefined) return; + if (parent.sessionManager.appendCustomEntry === undefined) return; + try { + parent.sessionManager.appendCustomEntry(SUBSESSION_LINK_CUSTOM_TYPE, persistedParentSubsessionLinkData(link)); + } catch (error: unknown) { + this.logger.info( + { parentSessionId: link.parentSessionId, sessionId: link.childSessionId, error: error instanceof Error ? error.message : String(error) }, + "failed to persist subsession link", + ); + } + } + + private persistSubsessionChildMarker(parentSessionId: string, childSessionId: string): void { + const child = this.active.get(childSessionId)?.runtime.session; + if (child === undefined) return; + if (child.sessionManager.appendCustomEntry === undefined) return; + try { + child.sessionManager.appendCustomEntry(SUBSESSION_CHILD_LINK_CUSTOM_TYPE, persistedChildSubsessionLinkData(parentSessionId, childSessionId)); + } catch (error: unknown) { + this.logger.info( + { parentSessionId, sessionId: childSessionId, error: error instanceof Error ? error.message : String(error) }, + "failed to persist subsession child marker", + ); + } + } + + private async hydrateSubsessionsForParent(parentSessionId: string, parentSessionFile?: string): Promise { + const hydrationKey = subsessionHydratedParentKey(parentSessionId, parentSessionFile); + if (this.subsessionHydratedParents.has(hydrationKey)) return; + + const activeParent = this.active.get(parentSessionId); + if (activeParent !== undefined && (parentSessionFile === undefined || activeSessionFileMatches(activeParent, parentSessionFile))) { + 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, parentManager: PiSessionManager, parentSessionFile: string | undefined): Promise { + // Parent custom links are the authoritative recovery record: verify the + // exact live child file/header before tracking. + const entries = parentManager.getEntries?.() ?? parentManager.getBranch(); + for (const entry of entries) { + const link = parsePersistedParentSubsessionLink(entry); + if (link === undefined) continue; + const verified = await this.verifiedSubsessionLinkFromParentLink(parentSessionId, parentSessionFile, link); + if (verified === undefined) continue; + this.registerVerifiedSubsession(verified); + } + } + + private async verifiedSubsessionLinkFromParentLink(parentSessionId: string, parentSessionFile: string | undefined, link: PersistedParentSubsessionLink): Promise { + if (parentSessionFile === undefined) return undefined; + if (link.spawnedBySessionId !== parentSessionId) return undefined; + if (!(await this.parentLinkHasValidChildTarget(parentSessionFile, link))) return undefined; + return trackedSubsessionLinkFromParentLink(parentSessionId, link, parentSessionFile); + } + + private async parentLinkHasValidChildTarget(parentSessionFile: string, link: PersistedParentSubsessionLink): Promise { + return link.spawnedSessionFile !== undefined + && await sessionFileHeaderMatches(link.spawnedSessionFile, { sessionId: link.spawnedSessionId, parentSessionFile }); + } + + private async recoverSubsessionTrackingForOpenedSession(session: PiAgentSession): Promise { + const link = await this.verifiedSubsessionLinkFromOpenedChild(session); + if (link === undefined) return; + this.registerVerifiedSubsession(link); + } + + private async verifiedSubsessionLinkFromOpenedChild(session: PiAgentSession): Promise { + // Child markers are only hints; the current child header and reciprocal + // parent custom link must agree on the exact ids and files before relinking. + const entries = session.sessionManager.getEntries?.() ?? session.sessionManager.getBranch(); + let marker: PersistedChildSubsessionLink | undefined; + for (const entry of entries) { + const parsed = parsePersistedChildSubsessionLink(entry); + if (parsed?.spawnedSessionId === session.sessionId) marker = parsed; + } + if (marker === undefined) return undefined; + + const childSessionFile = nonEmptyString(session.sessionFile); + if (childSessionFile === undefined) return undefined; + const childHeader = await readSessionHeaderSummary(childSessionFile); + if (childHeader?.id !== session.sessionId) return undefined; + const parentSessionFile = nonEmptyString(childHeader.parentSession); + if (parentSessionFile === undefined) return undefined; + const parentHeader = await readSessionHeaderSummary(parentSessionFile); + if (parentHeader?.id !== marker.spawnedBySessionId) return undefined; + + const parentLink = this.findReciprocalParentSubsessionLink(parentSessionFile, marker.spawnedBySessionId, session.sessionId, childSessionFile); + if (parentLink === undefined) return undefined; + return { + parentSessionId: marker.spawnedBySessionId, + childSessionId: session.sessionId, + childSessionFile, + parentSessionFile, + cwd: parentLink.cwd ?? session.sessionManager.getCwd(), + }; + } + + private findReciprocalParentSubsessionLink(parentSessionFile: string, parentSessionId: string, childSessionId: string, childSessionFile: string): PersistedParentSubsessionLink | undefined { + let parentManager: PiSessionManager; + try { + parentManager = this.sessionManager.open(parentSessionFile); + } catch { + return undefined; + } + const entries = parentManager.getEntries?.() ?? parentManager.getBranch(); + for (const entry of entries) { + const link = parsePersistedParentSubsessionLink(entry); + if (link === undefined) continue; + if (link.spawnedBySessionId !== parentSessionId || link.spawnedSessionId !== childSessionId) continue; + if (link.spawnedSessionFile === undefined || !sessionPathsEqual(link.spawnedSessionFile, childSessionFile)) continue; + return link; + } + return undefined; + } + + private async getOrOpenTrackedSubsession(sessionId: string): Promise { + 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 (link.childSessionFile !== undefined) { + if (!(await sessionFileHeaderMatches(link.childSessionFile, { sessionId, parentSessionFile: link.parentSessionFile }))) throw new Error("Session not found"); + const sessionManager = this.sessionManager.open(link.childSessionFile); + return (await this.create(sessionManager, link.cwd ?? sessionManager.getCwd())).runtime.session; + } + + throw new Error("Session not found"); } 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) { - return { cwd: active.runtime.cwd, status: await this.subsessionStatus(active.runtime.session) }; + return { cwd: active.runtime.cwd, status: this.subsessionStatus(active.runtime.session) }; } - const archived = await this.archiveStore.get(childSessionId); - if (archived !== undefined) return { cwd: archived.cwd, status: "archived" }; + if (link?.childSessionFile !== undefined && (await sessionFileHeaderMatches(link.childSessionFile, { sessionId: childSessionId, parentSessionFile: link.parentSessionFile }))) { + return { cwd: link.cwd ?? "", status: "idle" }; + } + if (link?.cwd !== undefined) return { cwd: link.cwd, status: "unknown" }; return { cwd: "", status: "unknown" }; } - private async subsessionStatus(session: PiAgentSession): Promise { - if (await this.archiveStore.isArchived(session.sessionId)) return "archived"; + private subsessionStatus(session: PiAgentSession): SubsessionStatus { if (this.hasActiveWork(session)) return "working"; if (this.activities.get(session.sessionId)?.phase === "error") return "error"; return "idle"; @@ -520,9 +765,9 @@ export class PiSessionService { * parent is busy and delivers immediately when it is idle). */ private updateSubsessionTracking(session: PiAgentSession): void { - const childId = session.sessionId; - const parentId = this.subsessionParents.get(childId); - if (parentId === undefined) return; + const link = this.subsessionLinkForActiveChild(session); + if (link === undefined) return; + const childId = link.childSessionId; if (this.hasActiveWork(session)) { this.subsessionNotifyArmed.set(childId, true); return; @@ -533,7 +778,23 @@ export class PiSessionService { const finalText = finalAssistantText(historyMessages(session)); 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.`; - void this.notifyParentOfSubsession(parentId, childId, text); + void this.notifyParentOfSubsession(link.parentSessionId, childId, text); + } + + private async getOrOpenParentForSubsession(parentSessionId: string, childSessionId: string): Promise { + 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; + + const parentSessionFile = link.parentSessionFile; + if (parentSessionFile === undefined) throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`); + if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) { + throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`); + } + const sessionManager = this.sessionManager.open(parentSessionFile); + return (await this.create(sessionManager, sessionManager.getCwd())).runtime.session; } /** @@ -545,7 +806,7 @@ export class PiSessionService { */ private async notifyParentOfSubsession(parentId: string, childId: string, text: string): Promise { try { - const session = await this.getOrOpen(parentId); + const session = await this.getOrOpenParentForSubsession(parentId, childId); await session.sendCustomMessage( { customType: SUBSESSION_NOTIFICATION_CUSTOM_TYPE, content: text, display: true, details: { sessionId: childId } }, { triggerTurn: true, deliverAs: "followUp" }, @@ -809,6 +1070,8 @@ export class PiSessionService { const sessionFile = session.sessionFile; if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); await clearParentSession(sessionFile); + clearParentSessionHeader(session.sessionManager); + this.unregisterSubsession(session.sessionId); } async abort(ref: PiSessionLookup): Promise { @@ -922,7 +1185,7 @@ export class PiSessionService { // Disarm subsession notification before teardown so the abort below cannot // 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. - this.subsessionNotifyArmed.delete(sessionId); + if (this.subsessionLinkForActiveChild(active.runtime.session) !== undefined) this.subsessionNotifyArmed.delete(sessionId); clearSessionQueue(active.runtime.session); active.unsubscribe(); try { @@ -979,8 +1242,10 @@ export class PiSessionService { runtime.setRebindSession(async (session) => { await this.bindSessionExtensions(session); this.bindRuntime(active); + await this.recoverSubsessionTrackingForOpenedSession(session); }); this.active.set(runtime.session.sessionId, active); + await this.recoverSubsessionTrackingForOpenedSession(runtime.session); this.publishStatus(runtime.session); return active; } @@ -1411,6 +1676,117 @@ function isDefined(value: T | undefined): value is T { return value !== undefined; } +function trackedSubsessionLinkFromParentLink(parentSessionId: string, link: PersistedParentSubsessionLink, parentSessionFile: string): TrackedSubsessionLink { + return { + parentSessionId, + childSessionId: link.spawnedSessionId, + ...(link.spawnedSessionFile === undefined ? {} : { childSessionFile: link.spawnedSessionFile }), + parentSessionFile, + ...(link.cwd === undefined ? {} : { cwd: link.cwd }), + }; +} + +function persistedParentSubsessionLinkData(link: TrackedSubsessionLink): Record { + return { + version: 1, + spawnedBySessionId: link.parentSessionId, + spawnedSessionId: link.childSessionId, + ...(link.childSessionFile === undefined ? {} : { spawnedSessionFile: link.childSessionFile }), + ...(link.cwd === undefined ? {} : { cwd: link.cwd }), + }; +} + +function persistedChildSubsessionLinkData(parentSessionId: string, childSessionId: string): Record { + return { + version: 1, + spawnedBySessionId: parentSessionId, + spawnedSessionId: childSessionId, + }; +} + +function parsePersistedParentSubsessionLink(entry: unknown): PersistedParentSubsessionLink | undefined { + if (!isRecord(entry) || entry["type"] !== "custom" || entry["customType"] !== SUBSESSION_LINK_CUSTOM_TYPE) return undefined; + const data = entry["data"]; + if (!isRecord(data)) return undefined; + const spawnedBySessionId = getString(data, "spawnedBySessionId"); + const spawnedSessionId = getString(data, "spawnedSessionId"); + if (spawnedBySessionId === undefined || spawnedBySessionId === "" || spawnedSessionId === undefined || spawnedSessionId === "") return undefined; + const spawnedSessionFile = getString(data, "spawnedSessionFile"); + const cwd = getString(data, "cwd"); + return { + spawnedBySessionId, + spawnedSessionId, + ...(spawnedSessionFile === undefined || spawnedSessionFile === "" ? {} : { spawnedSessionFile }), + ...(cwd === undefined || cwd === "" ? {} : { cwd }), + }; +} + +function parsePersistedChildSubsessionLink(entry: unknown): PersistedChildSubsessionLink | undefined { + if (!isRecord(entry) || entry["type"] !== "custom" || entry["customType"] !== SUBSESSION_CHILD_LINK_CUSTOM_TYPE) return undefined; + const data = entry["data"]; + if (!isRecord(data)) return undefined; + const spawnedBySessionId = getString(data, "spawnedBySessionId"); + const spawnedSessionId = getString(data, "spawnedSessionId"); + if (spawnedBySessionId === undefined || spawnedBySessionId === "" || spawnedSessionId === undefined || spawnedSessionId === "") return undefined; + return { spawnedBySessionId, spawnedSessionId }; +} + +function nonEmptyString(value: string | undefined): string | undefined { + 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 { + 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, 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 { + id: string; + parentSession?: string; +} + +async function readSessionHeaderSummary(sessionFile: string): Promise { + let file: Awaited> | undefined; + try { + file = await open(sessionFile, "r"); + const buffer = Buffer.alloc(4096); + const { bytesRead } = await file.read(buffer, 0, buffer.length, 0); + const firstLine = buffer.toString("utf8", 0, bytesRead).split("\n", 1)[0]; + if (firstLine === undefined || firstLine === "") return undefined; + const header: unknown = JSON.parse(firstLine); + if (!isRecord(header) || header["type"] !== "session" || typeof header["id"] !== "string") return undefined; + const parentSession = getString(header, "parentSession"); + return { id: header["id"], ...(parentSession === undefined ? {} : { parentSession }) }; + } catch { + return undefined; + } finally { + await file?.close().catch(() => undefined); + } +} + +async function sessionFileHeaderMatches(sessionFile: string, expected: { sessionId: string; parentSessionFile?: string | undefined }): Promise { + const header = await readSessionHeaderSummary(sessionFile); + if (header?.id !== expected.sessionId) return false; + if (expected.parentSessionFile === undefined) return true; + return header.parentSession !== undefined && sessionPathsEqual(header.parentSession, expected.parentSessionFile); +} + async function clearParentSession(sessionFile: string): Promise { const content = await readFile(sessionFile, "utf8"); const newlineIndex = content.indexOf("\n"); @@ -1423,6 +1799,11 @@ async function clearParentSession(sessionFile: string): Promise { await writeFile(sessionFile, `${JSON.stringify(header)}${rest}`, "utf8"); } +function clearParentSessionHeader(sessionManager: PiSessionManager): void { + const header = sessionManager.getHeader?.(); + if (header !== undefined && header !== null) delete header.parentSession; +} + function clearSessionQueue(session: PiAgentSession): void { session.clearQueue(); } @@ -1475,6 +1856,12 @@ function historyMessages(session: PiAgentSession): unknown[] { return messages; } +/** custom entry type used to persist parent -> child subsession links outside LLM context. */ +const SUBSESSION_LINK_CUSTOM_TYPE = "pi-web.subsession.link"; + +/** custom entry type used to mark a child as created by spawn_subsession. */ +const SUBSESSION_CHILD_LINK_CUSTOM_TYPE = "pi-web.subsession.spawned"; + /** customType marking a parent-facing subsession-completion notice. */ const SUBSESSION_NOTIFICATION_CUSTOM_TYPE = "subsession.completion"; diff --git a/src/server/sessions/spawnSubsessionTool.test.ts b/src/server/sessions/spawnSubsessionTool.test.ts index 35b4311..3bfa036 100644 --- a/src/server/sessions/spawnSubsessionTool.test.ts +++ b/src/server/sessions/spawnSubsessionTool.test.ts @@ -56,9 +56,9 @@ describe("createSubsessionToolDefinitions", () => { ])); 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: [ { sessionId: "child-1", cwd: "/repos/a", status: "working" }, { 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: 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(firstText(result.content)).toContain("all done"); }); @@ -99,9 +99,9 @@ describe("createSubsessionToolDefinitions", () => { })); 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(firstText(result.content)).toContain("the answer"); }); diff --git a/src/server/sessions/spawnSubsessionTool.ts b/src/server/sessions/spawnSubsessionTool.ts index 5ce2405..9297396 100644 --- a/src/server/sessions/spawnSubsessionTool.ts +++ b/src/server/sessions/spawnSubsessionTool.ts @@ -3,7 +3,7 @@ import { defineTool } from "@earendil-works/pi-coding-agent"; import type { TranscriptContentKind, TranscriptEntry, TranscriptRole, TranscriptView } from "./subsessionTranscript.js"; /** Lifecycle phase of a tracked subsession as seen by its parent. */ -export type SubsessionStatus = "working" | "idle" | "error" | "archived" | "unknown"; +export type SubsessionStatus = "working" | "idle" | "error" | "unknown"; export interface SpawnSubsessionResult { sessionId: string; @@ -56,9 +56,9 @@ export interface SubsessionReadQuery { export interface SubsessionToolDeps { spawn(input: SpawnSubsessionInvocation): Promise; - list(parentSessionId: string): Promise; - check(parentSessionId: string, sessionId: string): Promise; - read(parentSessionId: string, sessionId: string, query: SubsessionReadQuery): Promise; + list(parentSessionId: string, parentSessionFile?: string): Promise; + check(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise; + read(parentSessionId: string, sessionId: string, query: SubsessionReadQuery, parentSessionFile?: string): Promise; } const SpawnSubsessionParams = Type.Object({ @@ -196,7 +196,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse parameters: ListSubsessionsParams, async execute(_toolCallId, _params, _signal, _onUpdate, ctx) { 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 ? "You have not spawned any subsessions." : `Your subsessions:\n${subsessions.map(statusLine).join("\n")}`; @@ -212,7 +213,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse parameters: CheckSubsessionParams, async execute(_toolCallId, params, _signal, _onUpdate, ctx) { 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; return { 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, async execute(_toolCallId, params, _signal, _onUpdate, ctx) { const parentSessionId = ctx.sessionManager.getSessionId(); + const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined; const { sessionId, ...query } = params; - const result = await deps.read(parentSessionId, sessionId, query); + const result = await deps.read(parentSessionId, sessionId, query, parentSessionFile); return { content: [{ type: "text", text: renderTranscript(result) }], details: result,