import { open, readFile, writeFile } from "node:fs/promises"; import type { Api, ImageContent, Model } from "@earendil-works/pi-ai"; import { AuthStorage, createAgentSessionFromServices, createAgentSessionRuntime, createAgentSessionServices, createEditToolDefinition, defineTool, getAgentDir, ModelRegistry, SessionManager, type CreateAgentSessionRuntimeFactory, type EditToolDetails, } from "@earendil-works/pi-coding-agent"; import type { ClientArchiveSessionsResponse, ClientCommand, ClientCommandResult, ClientMessagePage, ClientSession, ClientSessionModel, ClientSessionRef, ClientSessionStatus, ClientThinkingLevel, SessionUiEvent } from "../types.js"; import { pageMessagesAtSafeBoundary } from "./messagePaging.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; import { BUILTIN_COMMANDS } from "./builtinCommands.js"; import { SessionCommandService } from "./sessionCommandService.js"; import { SessionArchiveStore, type ArchivedSessionRecord, type ArchiveSessionInput } from "./sessionArchiveStore.js"; import { findArchiveCandidateByIdOrPrefix, planSessionArchiveTree, type SessionArchiveTreeCandidate } from "./sessionArchiveTree.js"; import type { ActiveSession } from "./sessionRuntimeStore.js"; import type { AuthChange } from "./authService.js"; import { fallbackSessionName, generateShortSessionName } from "./sessionNameGenerator.js"; import { computeEditPreview, type EditPreviewResult } from "./editPreview.js"; import { createPiSessionManagerGateway } from "./piSessionManagerGateway.js"; import { attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachmentService.js"; import { parsePromptAttachments } from "../../shared/promptAttachments.js"; import type { SavedPromptAttachment } from "../../shared/apiTypes.js"; import { cwdPathsEqual } from "../workingDirectory.js"; import type { WorkspaceActivityService } from "../activity/workspaceActivityService.js"; import { createSpawnSessionToolDefinition, type SpawnSessionInvocation, type SpawnSessionResult } from "./spawnSessionTool.js"; import { createSubsessionToolDefinitions, type SpawnSubsessionInvocation, type SpawnSubsessionResult, type SubsessionCheckResult, type SubsessionReadQuery, type SubsessionReadResult, type SubsessionStatus, type SubsessionSummary, type SubsessionToolDeps } from "./spawnSubsessionTool.js"; import { buildTranscriptView } from "./subsessionTranscript.js"; import type { SpawnTargetDecision, SpawnTargetResolver } from "./spawnTargetResolver.js"; /** * Minimal structured-logging seam, shaped like Fastify's logger so sessiond can * pass `app.log` directly. Defaults to a no-op so the service stays usable * without booting a server (e.g. in tests). */ export interface PiSessionLogger { info(details: Record, message: string): void; } const noopLogger: PiSessionLogger = { info() { /* no-op */ } }; function noop(): void { // Intentionally empty default unsubscribe callback. } function spawnTargetError(decision: Extract): Error { if (decision.reason === "not-registered") return new Error("Spawning session is not in a registered project"); return new Error(`cwd must be a workspace of this project. Allowed: ${decision.allowedCwds.join(", ")}`); } function authLossWarningKey(sessionId: string, provider: string, modelId: string): string { return `${sessionId}:${provider}/${modelId}`; } function sessionIdFromLookup(ref: PiSessionLookup): string { return typeof ref === "string" ? ref : ref.id; } function isPiSessionRef(ref: PiSessionLookup): ref is PiSessionRef { return typeof ref !== "string"; } function lookupMatchesActiveSession(ref: PiSessionLookup, active: ActiveSession): boolean { return !isPiSessionRef(ref) || cwdPathsEqual(active.runtime.cwd, ref.cwd); } type QueuedPromptKind = "steer" | "followUp"; interface QueuedPrompt { kind: QueuedPromptKind; text: string; images?: ImageContent[]; 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; } function parsePromptStreamingBehavior(value: unknown): QueuedPromptKind | undefined { if (value === undefined) return undefined; if (value === "steer" || value === "followUp") return value; throw new Error('Prompt streamingBehavior must be "steer" or "followUp"'); } type SessionArchiveRepository = Pick & { deleteArchived?: (sessionId: string) => Promise }; export type PiSessionRef = ClientSessionRef; type PiSessionLookup = string | PiSessionRef; export interface PiSessionListEntry { id: string; path: string; cwd: string; created: Date; modified: Date; messageCount: number; firstMessage: string; allMessagesText: string; name?: string; parentSessionPath?: string; } interface WorkspaceArchiveCandidate extends SessionArchiveTreeCandidate { cwd: string; listEntry?: PiSessionListEntry; activeSession?: PiAgentSession; } type AgentModel = Model; 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 { list(cwd: string): Promise; create(cwd: string, options?: { parentSession?: string }): PiSessionManager; /** * Legacy id-only lookup surface for older clients. This intentionally searches * only Pi's default session store, because custom session directories require * a cwd-scoped lookup. */ listAll?(): Promise; open(path: string): PiSessionManager; } interface PiExtensionError { extensionPath: string; event: string; error: string; stack?: string; } interface PiExtensionBindings { onError?: (error: PiExtensionError) => void; } export interface PiAgentSession { modelRegistry: ModelRegistryInstance; sessionManager: PiSessionManager; scopedModels: readonly { model: AgentModel; thinkingLevel?: ClientThinkingLevel }[]; sessionId: string; sessionFile: string | undefined; sessionName: string | undefined; messages: readonly unknown[]; model: AgentModel | undefined; thinkingLevel: ClientThinkingLevel; isStreaming: boolean; isCompacting: boolean; isBashRunning: boolean; pendingMessageCount: number; extensionRunner: { getRegisteredCommands(): readonly { invocationName: string; description?: string }[] }; promptTemplates: readonly { name: string; description?: string }[]; resourceLoader: { getSkills(): { skills: readonly { name: string; description?: string }[] } }; subscribe(listener: (event: unknown) => void): () => void; bindExtensions(bindings: PiExtensionBindings): Promise; compact(instructions?: string): Promise<{ summary: string; tokensBefore: number }>; getUserMessagesForForking(): readonly { entryId: string; text: string }[]; getSessionStats(): { sessionId: string; totalMessages: number; userMessages: number; assistantMessages: number; toolCalls: number; tokens: ClientSessionStatus["tokens"]; cost: number }; getContextUsage(): ClientSessionStatus["contextUsage"] | undefined; prompt(text: string, options?: { streamingBehavior?: "steer" | "followUp"; images?: ImageContent[] }): Promise; sendCustomMessage(message: { customType: string; content: string; display: boolean; details?: unknown }, options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }): Promise; executeBash(command: string, onChunk?: (chunk: string) => void, options?: { excludeFromContext?: boolean }): Promise<{ output: string; exitCode: number | undefined; cancelled: boolean; truncated: boolean; fullOutputPath?: string }>; abort(): Promise; clearQueue(): { steering: string[]; followUp: string[] }; getSteeringMessages(): readonly string[]; getFollowUpMessages(): readonly string[]; setModel(model: AgentModel): Promise; cycleModel(direction?: "forward" | "backward"): Promise<{ model: AgentModel } | undefined>; getAvailableThinkingLevels(): ClientThinkingLevel[]; setThinkingLevel(level: ClientThinkingLevel): void; cycleThinkingLevel(): ClientThinkingLevel | undefined; setSessionName(name: string): void; } export interface PiSessionRuntime { readonly cwd: string; readonly session: PiAgentSession; setRebindSession(rebindSession?: (session: PiAgentSession) => Promise): void; fork(entryId: string, options?: { position?: "before" | "at" }): Promise<{ cancelled: boolean; selectedText?: string }>; dispose(): Promise; } interface CreateAgentRuntimeOptions { cwd: string; agentDir: string; sessionManager: PiSessionManager; } type CreateAgentRuntime = (createRuntime: CreateAgentSessionRuntimeFactory, options: CreateAgentRuntimeOptions) => Promise; function defaultCreateAgentRuntime(createRuntime: CreateAgentSessionRuntimeFactory, options: CreateAgentRuntimeOptions): Promise { if (!(options.sessionManager instanceof SessionManager)) throw new Error("Default runtime creation requires an SDK SessionManager"); return createAgentSessionRuntime(createRuntime, { ...options, sessionManager: options.sessionManager }); } type SpawnSessionFn = (input: SpawnSessionInvocation) => Promise; function createDefaultRuntimeFactory(authStorage: AuthStorage, modelRegistry: ModelRegistryInstance, spawn?: SpawnSessionFn, subsessions?: SubsessionToolDeps): CreateAgentSessionRuntimeFactory { return async ({ cwd, agentDir, sessionManager, sessionStartEvent }) => { const services = await createAgentSessionServices({ cwd, agentDir, authStorage, modelRegistry }); const customTools = [ createPiWebEditToolDefinition(cwd), ...(spawn === undefined ? [] : [createSpawnSessionToolDefinition(cwd, { spawn })]), ...(subsessions === undefined ? [] : createSubsessionToolDefinitions(cwd, subsessions)), ]; const options = sessionStartEvent === undefined ? { services, sessionManager, customTools } : { services, sessionManager, sessionStartEvent, customTools }; const result = await createAgentSessionFromServices(options); return { ...result, services, diagnostics: services.diagnostics }; }; } type PiWebEditToolDetails = EditToolDetails | { preview: EditPreviewResult } | undefined; function createPiWebEditToolDefinition(cwd: string) { const editTool = createEditToolDefinition(cwd); return defineTool({ name: editTool.name, label: editTool.label, description: editTool.description, ...(editTool.promptSnippet === undefined ? {} : { promptSnippet: editTool.promptSnippet }), ...(editTool.promptGuidelines === undefined ? {} : { promptGuidelines: editTool.promptGuidelines }), parameters: editTool.parameters, ...(editTool.renderShell === undefined ? {} : { renderShell: editTool.renderShell }), ...(editTool.prepareArguments === undefined ? {} : { prepareArguments: editTool.prepareArguments }), ...(editTool.executionMode === undefined ? {} : { executionMode: editTool.executionMode }), async execute(toolCallId, params, signal, onUpdate, ctx) { const preview = await computeEditPreview(params.path, params.edits, cwd); if (signal?.aborted !== true) { onUpdate?.({ content: [{ type: "text", text: "Edit preview computed." }], details: { preview } }); } return editTool.execute(toolCallId, params, signal, onUpdate, ctx); }, }); } export interface PiSessionServiceDependencies { archiveStore?: SessionArchiveRepository; agentDir?: string; sessionManager?: PiSessionManagerGateway; createRuntime?: CreateAgentSessionRuntimeFactory; createAgentRuntime?: CreateAgentRuntime; modelRegistry?: ModelRegistryInstance; heartbeatIntervalMs?: number; workspaceActivity?: Pick; /** * When provided, the `spawn_session` tool is registered on every session, * letting the LLM start new sessions scoped to its project's workspaces. * Omit to keep the capability disabled (the tool is never registered). */ spawnTargets?: SpawnTargetResolver; /** * Beta: when true (and `spawnTargets` is provided), the tracked-subsession * tools (`spawn_subsession`, `list_subsessions`, `check_subsession`, * `read_subsession`) are * registered on every session. Off by default so the capability can ship in * main without being exposed in releases. */ subsessionsEnabled?: boolean; /** Structured logger for notable runtime events (e.g. spawns). */ logger?: PiSessionLogger; } export class PiSessionService { private readonly active = new Map>(); private readonly activities = new Map(); private readonly heartbeat: NodeJS.Timeout; private readonly commandService: SessionCommandService; private readonly compactionPromptQueues = new Map(); private readonly compactionDrainTimers = new Map(); private readonly authLossWarnings = new Set(); /** Tracked subsession id -> the parent session id that spawned it. */ 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 * child that works again (and stops again) notifies the parent each time. */ private readonly subsessionNotifyArmed = new Map(); private readonly archiveStore: SessionArchiveRepository; private readonly agentDir: string; private readonly sessionManager: PiSessionManagerGateway; private readonly createRuntime: CreateAgentSessionRuntimeFactory; private readonly createAgentRuntime: CreateAgentRuntime; private readonly modelRegistry: ModelRegistryInstance; private readonly workspaceActivity: Pick | undefined; private readonly spawnTargets: SpawnTargetResolver | undefined; private readonly logger: PiSessionLogger; constructor(private readonly events: SessionEventHub, deps: PiSessionServiceDependencies = {}) { this.archiveStore = deps.archiveStore ?? new SessionArchiveStore(); this.agentDir = deps.agentDir ?? getAgentDir(); this.sessionManager = deps.sessionManager ?? createPiSessionManagerGateway({ agentDir: this.agentDir }); this.modelRegistry = deps.modelRegistry ?? ModelRegistry.create(AuthStorage.create()); this.spawnTargets = deps.spawnTargets; this.logger = deps.logger ?? noopLogger; // Subsessions are a beta capability gated behind their own flag, and they // also require the spawn capability (they share its project-scope resolver). const subsessionsActive = this.spawnTargets !== undefined && deps.subsessionsEnabled === true; this.createRuntime = deps.createRuntime ?? createDefaultRuntimeFactory( this.modelRegistry.authStorage, this.modelRegistry, this.spawnTargets === undefined ? undefined : (input) => this.spawnSession(input), !subsessionsActive ? undefined : { spawn: (input) => this.spawnSubsession(input), 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; this.workspaceActivity = deps.workspaceActivity; this.heartbeat = setInterval(() => { this.publishHeartbeats(); }, deps.heartbeatIntervalMs ?? 2000); this.commandService = new SessionCommandService( (sessionId) => this.getActive(sessionId), (sessionId, text) => this.prompt(sessionId, text, undefined, undefined, { echoUserMessage: false }), events, { onCompactionStart: (session) => { this.publishActivity(session, "compacting", "active"); this.publishStatus(session); }, onCompactionEnd: (session, result, detail) => { this.publishActivity(session, result === "success" ? "compaction complete" : "compaction failed", result === "success" ? "idle" : "error", detail); this.publishStatus(session); }, }, { listSessionNames: (cwd) => this.listSessionNames(cwd) }, ); } activeCount(): number { return this.active.size; } async dispose(): Promise { clearInterval(this.heartbeat); this.clearCompactionDrainTimers(); const activeSessions = Array.from(new Set(this.active.values())); this.active.clear(); this.activities.clear(); this.compactionPromptQueues.clear(); 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(); this.workspaceActivity?.removeSession(active.runtime.session.sessionId, active.runtime.session.sessionManager.getCwd()); await active.runtime.session.abort(); await active.runtime.dispose(); })); } async list(cwd: string): Promise { const [sessions, archivedRecords] = await Promise.all([this.sessionManager.list(cwd), this.archiveStore.list()]); const sessionsById = new Map(sessions.map((session) => [session.id, session])); const archivedForCwd = await Promise.all( archivedRecords .filter((record) => record.cwd === cwd) .map((record) => this.ensureArchivedSessionMoved(record, sessionsById.get(record.sessionId))), ); const archivedById = new Map(archivedForCwd.map((record) => [record.sessionId, record])); const unarchivedSessions = sessions.filter((session) => !archivedById.has(session.id)).map(clientSessionFromListEntry); this.workspaceActivity?.reconcileSessionActivity(cwd, this.reconcilableSessionIds(cwd, unarchivedSessions.map((session) => session.id), archivedById)); const archivedSessions = archivedForCwd .sort(compareArchivedRecords) .map((record) => clientSessionFromArchivedRecord(record, sessionsById.get(record.sessionId))) .filter(isDefined); return [...unarchivedSessions, ...archivedSessions]; } async start(cwd: string, parentSession?: string): Promise { const active = await this.create(this.sessionManager.create(cwd, parentSession === undefined ? undefined : { parentSession }), cwd); const { session } = active.runtime; const created: ClientSession = { id: session.sessionId, path: session.sessionFile ?? "", cwd, created: new Date().toISOString(), modified: new Date().toISOString(), messageCount: session.messages.length, firstMessage: "", // Include the parent so listeners can nest the new session in the tree // immediately, instead of showing it flat until the next reload. ...(parentSession === undefined ? {} : { parentSessionPath: parentSession }), }; // Broadcast so other clients (and the spawning agent's UI) can add the new // session to their list without a manual reload. this.events.publishGlobal({ type: "session.created", session: created }); return created; } /** * Start a new session on behalf of a LLM and deliver an initial prompt to it. * The target cwd is constrained to a workspace of the same registered project * as the spawning session so the new session is visible in the web UI. */ async spawnSession(input: SpawnSessionInvocation): Promise { if (this.spawnTargets === undefined) throw new Error("Spawning sessions is disabled"); const decision = await this.spawnTargets.resolveSpawnTarget(input.spawningCwd, input.cwd); if (!decision.allowed) throw spawnTargetError(decision); const created = await this.start(decision.cwd); await this.prompt(created.id, input.prompt); this.logger.info( { spawningCwd: input.spawningCwd, sessionId: created.id, cwd: decision.cwd, promptLength: input.prompt.length }, "spawn_session started a new session", ); return { sessionId: created.id, cwd: decision.cwd }; } /** * Start a *tracked* child session on behalf of a LLM. Identical to * {@link spawnSession} in how the target cwd is resolved, but the child * records its parent (so it shows in the session tree) and is registered so * the parent is notified when it stops working and can inspect it later. */ async spawnSubsession(input: SpawnSubsessionInvocation): Promise { if (this.spawnTargets === undefined) throw new Error("Spawning sessions is disabled"); 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); 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 }, "spawn_subsession started a tracked child session", ); return { sessionId: created.id, cwd: decision.cwd }; } /** Summaries of the tracked subsessions spawned by `parentSessionId`. */ 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 []; 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, parentSessionFile?: string): Promise { const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile); const messages = historyMessages(session); return { sessionId, cwd: session.sessionManager.getCwd(), 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, parentSessionFile?: string): Promise { const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile); const view = buildTranscriptView(historyMessages(session), query); return { sessionId, cwd: session.sessionManager.getCwd(), 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, 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.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 | 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.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 link = this.subsessionLinks.get(childSessionId); const active = link === undefined ? undefined : this.activeChildForSubsessionLink(link); if (active !== undefined) { return { cwd: active.runtime.cwd, status: this.subsessionStatus(active.runtime.session) }; } 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 subsessionStatus(session: PiAgentSession): SubsessionStatus { if (this.hasActiveWork(session)) return "working"; if (this.activities.get(session.sessionId)?.phase === "error") return "error"; return "idle"; } /** * Drive parent notifications from a tracked child's status. Arms a pending * notification while the child is working, and when it stops fires a single * follow-up message to the parent via {@link prompt} (which queues if the * parent is busy and delivers immediately when it is idle). */ private updateSubsessionTracking(session: PiAgentSession): void { const link = this.subsessionLinkForActiveChild(session); if (link === undefined) return; const childId = link.childSessionId; if (this.hasActiveWork(session)) { this.subsessionNotifyArmed.set(childId, true); return; } if (this.subsessionNotifyArmed.get(childId) !== true) return; this.subsessionNotifyArmed.set(childId, false); const status: SubsessionStatus = this.activities.get(childId)?.phase === "error" ? "error" : "idle"; 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(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; } /** * Deliver a subsession-completion notice to the parent as a system-authored * custom message rather than a user message, so it is not attributed to the * human in the transcript. It still wakes an idle parent (`triggerTurn`) and * queues behind in-flight work (`deliverAs: "followUp"`), preserving the * established "queue if busy, send and act if idle" behavior. */ private async notifyParentOfSubsession(parentId: string, childId: string, text: string): Promise { try { 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" }, ); this.publishStatus(session); } catch (error: unknown) { this.logger.info( { parentSessionId: parentId, sessionId: childId, error: error instanceof Error ? error.message : String(error) }, "failed to notify parent of subsession completion", ); } } async messages(ref: PiSessionLookup, page?: { before?: number; limit?: number }): Promise { const session = await this.getOrOpen(ref); return pageMessagesAtSafeBoundary(historyMessages(session), page); } async status(ref: PiSessionLookup): Promise { return this.statusFromSession(await this.getOrOpen(ref)); } async availableModels(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); session.modelRegistry.refresh(); const models = session.scopedModels.length > 0 ? session.scopedModels.map((scoped) => scoped.model) : session.modelRegistry.getAvailable(); return models.map(modelToClientModel); } async setModel(ref: PiSessionLookup, provider: string, modelId: string): Promise { await this.assertWritable(ref); const session = await this.getOrOpen(ref); session.modelRegistry.refresh(); const candidates = session.scopedModels.length > 0 ? session.scopedModels.map((scoped) => scoped.model) : session.modelRegistry.getAvailable(); const model = candidates.find((candidate) => candidate.provider === provider && candidate.id === modelId) ?? session.modelRegistry.find(provider, modelId); if (model === undefined) throw new Error(`Model not found: ${provider}/${modelId}`); await session.setModel(model); this.publishActivity(session, `model: ${model.id}`, "idle", model.provider); this.publishStatus(session); return this.statusFromSession(session); } async cycleModel(ref: PiSessionLookup, direction: "forward" | "backward"): Promise { await this.assertWritable(ref); const session = await this.getOrOpen(ref); const result = await session.cycleModel(direction); if (result === undefined) throw new Error(session.scopedModels.length > 0 ? "Only one model in scope" : "Only one model available"); this.publishActivity(session, `model: ${result.model.id}`, "idle", result.model.provider); this.publishStatus(session); return this.statusFromSession(session); } async availableThinkingLevels(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); return session.getAvailableThinkingLevels(); } async setThinkingLevel(ref: PiSessionLookup, level: string): Promise { await this.assertWritable(ref); const session = await this.getOrOpen(ref); // pi owns the valid set; validate against the session's live levels rather // than a hardcoded union so this stays correct if pi changes the set. const available = session.getAvailableThinkingLevels(); const match = available.find((candidate) => candidate === level); if (match === undefined) throw new Error(`Invalid thinking level: ${level}`); session.setThinkingLevel(match); this.publishActivity(session, `thinking: ${session.thinkingLevel}`, "idle"); this.publishStatus(session); return this.statusFromSession(session); } async cycleThinkingLevel(ref: PiSessionLookup): Promise { await this.assertWritable(ref); const session = await this.getOrOpen(ref); const level = session.cycleThinkingLevel(); if (level === undefined) throw new Error("Current model does not support thinking"); this.publishActivity(session, `thinking: ${level}`, "idle"); this.publishStatus(session); return this.statusFromSession(session); } async commands(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); const commands: ClientCommand[] = [...BUILTIN_COMMANDS]; for (const command of session.extensionRunner.getRegisteredCommands()) { commands.push({ name: command.invocationName, ...(command.description === undefined ? {} : { description: command.description }), source: "extension" }); } for (const template of session.promptTemplates) { commands.push({ name: template.name, ...(template.description === undefined ? {} : { description: template.description }), source: "prompt" }); } for (const skill of session.resourceLoader.getSkills().skills) { commands.push({ name: `skill:${skill.name}`, ...(skill.description === undefined ? {} : { description: skill.description }), source: "skill" }); } return commands.sort((a, b) => a.name.localeCompare(b.name)); } async prompt(ref: PiSessionLookup, text: unknown, streamingBehavior?: unknown, attachments?: unknown, options?: { echoUserMessage?: boolean }): Promise { const promptText = requirePromptText(text); // Command-forwarded prompts (e.g. /skill:*) are expanded by the agent, which // streams the canonical message back. The client doesn't render the raw // command text, so the server must not echo it either, or it would show up // as a transient line that vanishes on reload. const echoUserMessage = options?.echoUserMessage !== false; const requestedBehavior = parsePromptStreamingBehavior(streamingBehavior); const parsedAttachments = parsePromptAttachments(attachments, { enforceInlineSizeLimit: false }); const images = (await attachmentsToInlineImages(parsedAttachments)).map((entry) => entry.image); await this.assertWritable(ref); const session = await this.getOrOpen(ref); this.maybeGenerateSessionName(session, promptText); const isQueued = session.isStreaming || session.isCompacting; const behavior = isQueued ? requestedBehavior ?? "followUp" : undefined; if (isQueued && images.length === 0 && this.hasQueuedMessageText(session, promptText)) { this.publishActivity(session, "duplicate queued message ignored", "active"); this.publishStatus(session); return; } if (session.isCompacting) { this.enqueuePromptDuringCompaction(session, promptText, behavior ?? "followUp", images, echoUserMessage); return; } void this.submitPrompt(session, promptText, behavior, images, echoUserMessage); } private submitPrompt(session: PiAgentSession, text: string, behavior: QueuedPromptKind | undefined, images: ImageContent[] = [], echoUserMessage = true): Promise { this.publishActivity(session, behavior === "steer" ? "steering queued" : behavior === "followUp" ? "message queued" : "prompt accepted", "active"); if (behavior === undefined && echoUserMessage) this.events.publish(session.sessionId, { type: "message.append", message: userMessage(text, images) }); const promptOptions = buildPromptOptions(behavior, images); const promptPromise = session.prompt(text, promptOptions).catch((error: unknown) => { const message = error instanceof Error ? error.message : String(error); this.publishActivity(session, "error", "error", message); this.events.publish(session.sessionId, { type: "session.error", message }); }); void promptPromise; return promptPromise; } private enqueuePromptDuringCompaction(session: PiAgentSession, text: string, kind: QueuedPromptKind, images: ImageContent[] = [], echoUserMessage = true): void { const queue = this.compactionPromptQueues.get(session.sessionId) ?? []; queue.push({ kind, text, ...(images.length > 0 ? { images } : {}), ...(echoUserMessage ? {} : { echoUserMessage: false }) }); this.compactionPromptQueues.set(session.sessionId, queue); this.publishActivity(session, "message queued during compaction", "active"); this.publishStatus(session); } async saveAttachments(ref: PiSessionLookup, attachments: unknown, folder?: string): Promise { const parsed = parsePromptAttachments(attachments, { enforceInlineSizeLimit: false, allowFileAttachments: true }); if (parsed.length === 0) return []; await this.assertWritable(ref); const active = await this.getActive(ref); return saveAttachmentsToWorkspace(active.runtime.cwd, parsed, folder === undefined ? {} : { folder }); } async shell(ref: PiSessionLookup, text: string): Promise { await this.assertWritable(ref); const active = await this.getActive(ref); const { session } = active.runtime; const isExcluded = text.startsWith("!!"); const command = (isExcluded ? text.slice(2) : text.slice(1)).trim(); if (!command) throw new Error("Usage: !"); if (session.isBashRunning) throw new Error("A bash command is already running"); this.publishActivity(session, "running bash", "active", command); this.events.publish(session.sessionId, { type: "shell.start", command, excludeFromContext: isExcluded }); void session.executeBash(command, (chunk) => { this.events.publish(session.sessionId, { type: "shell.chunk", chunk }); this.publishActivity(session, "running bash", "active", command); this.publishStatus(session); }, { excludeFromContext: isExcluded }).then((result) => { this.events.publish(session.sessionId, { type: "shell.end", output: result.output, ...(result.exitCode === undefined ? {} : { exitCode: result.exitCode }), cancelled: result.cancelled, truncated: result.truncated, ...(result.fullOutputPath === undefined ? {} : { fullOutputPath: result.fullOutputPath }), }); this.publishActivity(session, "bash complete", result.exitCode === 0 ? "idle" : "error", command); this.publishStatus(session); }).catch((error: unknown) => { const message = error instanceof Error ? error.message : String(error); this.events.publish(session.sessionId, { type: "shell.end", output: message, isError: true }); this.events.publish(session.sessionId, { type: "session.error", message }); this.publishActivity(session, "bash failed", "error", message); this.publishStatus(session); }); } async runCommand(ref: PiSessionLookup, text: string): Promise { await this.assertWritable(ref); const active = await this.getActive(ref); return this.commandService.run(active.runtime.session.sessionId, text); } async respondToCommand(ref: PiSessionLookup, requestId: string, value: string): Promise { await this.assertWritable(ref); const active = await this.getActive(ref); return this.commandService.respond(active.runtime.session.sessionId, requestId, value); } async archive(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); if (this.hasActiveWork(session)) throw new Error("Stop current session activity before archiving"); const archiveInput = await this.archiveInputForSession(session); await this.closeActive(session.sessionId); await this.archiveStore.archive(archiveInput); } async archiveTree(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); const catalog = await this.workspaceArchiveCandidates(session.sessionManager.getCwd()); const root = findArchiveCandidateByIdOrPrefix(catalog, session.sessionId) ?? archiveCandidateFromActiveSession(session, false); const plan = planSessionArchiveTree(root, catalog); const busy = plan.targets.map((target) => target.activeSession).find((target) => target !== undefined && this.hasActiveWork(target)); if (busy !== undefined) throw new Error(`Stop current session activity before archiving ${sessionDisplayName(busy)}`); const archiveInputs = plan.unarchivedTargets.map((target) => archiveInputFromCandidate(target)); for (const input of archiveInputs) await this.closeActive(input.sessionId); for (const input of archiveInputs) await this.archiveStore.archive(input); return { archived: true, sessionIds: archiveInputs.map((input) => input.sessionId), archivedCount: archiveInputs.length, skippedAlreadyArchivedCount: plan.skippedAlreadyArchivedCount, }; } async restore(ref: PiSessionLookup): Promise { const archived = await this.getArchived(ref); if (archived === undefined) throw new Error("Session not found"); await this.closeActive(archived.sessionId); await this.archiveStore.restore(archived.sessionId); } async deleteArchived(ref: PiSessionLookup): Promise { const record = await this.getArchived(ref); if (record === undefined) throw new Error("Archived session not found"); if (this.archiveStore.deleteArchived === undefined) throw new Error("Archive store does not support deletion"); await this.closeActive(record.sessionId); if (record.archivePath === undefined) await this.ensureArchivedRecordMoved(record); await this.archiveStore.deleteArchived(record.sessionId); } async reload(ref: PiSessionLookup): Promise { await this.assertWritable(ref); const session = await this.getOrOpen(ref); if (this.hasActiveWork(session)) throw new Error("Stop current session activity before reloading"); await this.closeActive(session.sessionId); const reopened = await this.getActive(ref); this.publishStatus(reopened.runtime.session); } async detachParent(ref: PiSessionLookup): Promise { const session = await this.getOrOpen(ref); 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 { const active = this.activeForLookup(ref); if (active === undefined) return; const sessionId = active.runtime.session.sessionId; this.clearCompactionPromptQueue(sessionId); clearSessionQueue(active.runtime.session); await active.runtime.session.abort(); this.publishActivity(active.runtime.session, "stopped", "idle"); this.publishStatus(active.runtime.session); } stop(ref: PiSessionLookup): void { const active = this.activeForLookup(ref); if (active === undefined) return; void this.closeActive(active.runtime.session.sessionId).catch(() => { // Best-effort shutdown; callers that need errors await closeActive directly. }); } private reconcilableSessionIds(cwd: string, listedSessionIds: string[], archivedById: Map): string[] { const sessionIds = new Set(listedSessionIds); for (const active of new Set(this.active.values())) { const session = active.runtime.session; if (session.sessionManager.getCwd() === cwd && !archivedById.has(session.sessionId)) sessionIds.add(session.sessionId); } return [...sessionIds]; } private async ensureArchivedSessionMoved(record: ArchivedSessionRecord, session: PiSessionListEntry | undefined): Promise { if (session === undefined || this.active.has(record.sessionId)) return record; try { return await this.archiveStore.archive(archiveInputFromListEntry(session)); } catch { return record; } } private async ensureArchivedRecordMoved(record: ArchivedSessionRecord): Promise { const session = (await this.sessionManager.list(record.cwd)).find((candidate) => candidate.id === record.sessionId); if (session === undefined) return record; return this.archiveStore.archive(archiveInputFromListEntry(session)); } private async archiveInputForSession(session: PiAgentSession): Promise { const cwd = session.sessionManager.getCwd(); const sessionFile = session.sessionFile; if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); const listed = (await this.sessionManager.list(cwd)).find((candidate) => candidate.id === session.sessionId); if (listed !== undefined) return archiveInputFromListEntry(listed); return archiveInputFromActiveSession(session); } private async workspaceArchiveCandidates(cwd: string): Promise { const [sessions, archivedRecords] = await Promise.all([this.sessionManager.list(cwd), this.archiveStore.list()]); const candidates = new Map(); const archivedById = new Map(); for (const record of archivedRecords) { if (record.cwd === cwd) archivedById.set(record.sessionId, record); } for (const session of sessions) { const archived = archivedById.get(session.id); if (archived === undefined) candidates.set(session.id, archiveCandidateFromListEntry(session)); else { const candidate = archiveCandidateFromArchivedRecord(archived, session); if (candidate !== undefined) candidates.set(candidate.id, candidate); } } for (const record of archivedById.values()) { if (candidates.has(record.sessionId)) continue; const candidate = archiveCandidateFromArchivedRecord(record, undefined); if (candidate !== undefined) candidates.set(candidate.id, candidate); } for (const active of new Set(this.active.values())) { const session = active.runtime.session; if (session.sessionManager.getCwd() !== cwd || archivedById.has(session.sessionId)) continue; const existing = candidates.get(session.sessionId); candidates.set(session.sessionId, { ...(existing ?? archiveCandidateFromActiveSession(session, false)), activeSession: session }); } return [...candidates.values()]; } private async listSessionNames(cwd: string): Promise { const [sessions, archivedRecords] = await Promise.all([this.sessionManager.list(cwd), this.archiveStore.list()]); const names = new Set(); for (const session of sessions) addSessionName(names, session.name); for (const record of archivedRecords) { if (record.cwd === cwd) addSessionName(names, record.name); } for (const active of new Set(this.active.values())) { const session = active.runtime.session; if (session.sessionManager.getCwd() === cwd) addSessionName(names, session.sessionName); } return [...names]; } private async closeActive(sessionId: string): Promise { const active = this.active.get(sessionId); if (!active) return; this.active.delete(sessionId); this.activities.delete(sessionId); this.workspaceActivity?.removeSession(sessionId, active.runtime.session.sessionManager.getCwd()); this.clearAuthLossWarningsForSession(sessionId); this.clearCompactionPromptQueue(sessionId); // 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. if (this.subsessionLinkForActiveChild(active.runtime.session) !== undefined) this.subsessionNotifyArmed.delete(sessionId); clearSessionQueue(active.runtime.session); active.unsubscribe(); try { await active.runtime.session.abort(); } finally { await active.runtime.dispose(); } } private async assertWritable(ref: PiSessionLookup): Promise { if (await this.getArchived(ref) !== undefined) throw new Error("Archived sessions are read-only. Restore the session to continue."); } private async getOrOpen(ref: PiSessionLookup): Promise { return (await this.getActive(ref)).runtime.session; } private async getActive(ref: PiSessionLookup): Promise> { const active = this.activeForLookup(ref); if (active !== undefined) return active; const archived = await this.getArchived(ref); if (archived?.archivePath !== undefined) return this.create(this.sessionManager.open(archived.archivePath), archived.cwd); const match = isPiSessionRef(ref) ? (await this.sessionManager.list(ref.cwd)).find((s) => s.id === ref.id || s.id.startsWith(ref.id)) : (await this.sessionManager.listAll?.() ?? []).find((s) => s.id === ref || s.id.startsWith(ref)); if (!match) throw new Error("Session not found"); return this.create(this.sessionManager.open(match.path), match.cwd); } private async getArchived(ref: PiSessionLookup): Promise { const archived = await this.archiveStore.get(sessionIdFromLookup(ref)); if (archived === undefined) return undefined; if (isPiSessionRef(ref) && archived.cwd !== ref.cwd) return undefined; return archived; } private activeForLookup(ref: PiSessionLookup): ActiveSession | undefined { const sessionId = sessionIdFromLookup(ref); const exact = this.active.get(sessionId); if (exact !== undefined && lookupMatchesActiveSession(ref, exact)) return exact; for (const [candidateId, active] of this.active.entries()) { if (candidateId.startsWith(sessionId) && lookupMatchesActiveSession(ref, active)) return active; } return undefined; } private async create(sessionManager: PiSessionManager, cwd: string): Promise> { const runtime = await this.createAgentRuntime(this.createRuntime, { cwd, agentDir: this.agentDir, sessionManager }); await this.bindSessionExtensions(runtime.session); const active: ActiveSession = { runtime, unsubscribe: noop }; this.bindRuntime(active); 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; } private async bindSessionExtensions(session: PiAgentSession): Promise { await session.bindExtensions({ onError: (error) => { const message = `${error.extensionPath}: ${error.error}`; this.publishActivity(session, "extension error", "error", message); this.events.publish(session.sessionId, { type: "session.error", message }); }, }); } private bindRuntime(active: ActiveSession): void { active.unsubscribe(); const { session } = active.runtime; for (const [sessionId, candidate] of this.active.entries()) { if (candidate === active) { this.active.delete(sessionId); if (sessionId !== session.sessionId) this.clearCompactionPromptQueue(sessionId); } } active.unsubscribe = session.subscribe((event) => { this.events.publish(session.sessionId, toClientEvent(event)); this.publishActivityForEvent(session, event); const eventType = getString(event, "type"); if (eventType === "compaction_end") this.scheduleCompactionQueueDrain(session.sessionId); if (eventType === "agent_start" || eventType === "agent_end") this.scheduleCompactionQueueDrain(session.sessionId); this.publishStatus(session); this.updateSubsessionTracking(session); }); this.active.set(session.sessionId, active); } private scheduleCompactionQueueDrain(sessionId: string, delayMs = 0): void { if (!this.compactionPromptQueues.has(sessionId) || this.compactionDrainTimers.has(sessionId)) return; const timer = setTimeout(() => { this.compactionDrainTimers.delete(sessionId); this.drainCompactionPromptQueue(sessionId); }, delayMs); this.compactionDrainTimers.set(sessionId, timer); } private drainCompactionPromptQueue(sessionId: string): void { const active = this.active.get(sessionId); if (active === undefined) return; const { session } = active.runtime; if (session.isCompacting) { this.scheduleCompactionQueueDrain(sessionId, 100); return; } if (session.isStreaming) { const queued = this.takeCompactionPromptQueue(sessionId); if (queued.length === 0) return; this.publishStatus(session); for (const prompt of queued) void this.submitPrompt(session, prompt.text, prompt.kind, prompt.images, prompt.echoUserMessage ?? true); return; } const prompt = this.shiftCompactionPrompt(sessionId); if (prompt === undefined) return; this.publishStatus(session); const submitted = this.submitPrompt(session, prompt.text, undefined, prompt.images, prompt.echoUserMessage ?? true); void submitted.finally(() => { this.scheduleCompactionQueueDrain(sessionId); }); } private takeCompactionPromptQueue(sessionId: string): QueuedPrompt[] { const queued = this.compactionPromptQueues.get(sessionId) ?? []; this.compactionPromptQueues.delete(sessionId); return queued; } private shiftCompactionPrompt(sessionId: string): QueuedPrompt | undefined { const queue = this.compactionPromptQueues.get(sessionId); const prompt = queue?.shift(); if (queue === undefined || queue.length === 0) this.compactionPromptQueues.delete(sessionId); return prompt; } private clearCompactionPromptQueue(sessionId: string): void { this.compactionPromptQueues.delete(sessionId); const timer = this.compactionDrainTimers.get(sessionId); if (timer !== undefined) { clearTimeout(timer); this.compactionDrainTimers.delete(sessionId); } } private clearCompactionDrainTimers(): void { for (const timer of this.compactionDrainTimers.values()) clearTimeout(timer); this.compactionDrainTimers.clear(); } private maybeGenerateSessionName(session: PiAgentSession, firstMessage: string): void { if (session.sessionName !== undefined || session.messages.length !== 0 || session.isStreaming || session.isCompacting) return; const model = session.model; if (model === undefined) return; void generateShortSessionName(this.modelRegistry, model, firstMessage).then((name) => { this.applyGeneratedSessionName(session, name ?? fallbackSessionName(firstMessage)); }).catch(() => { this.applyGeneratedSessionName(session, fallbackSessionName(firstMessage)); }); } private applyGeneratedSessionName(session: PiAgentSession, name: string | undefined): void { if (name === undefined || session.sessionName !== undefined) return; session.setSessionName(name); this.publishSessionName(session); } applyAuthChange(change: AuthChange = {}): void { this.modelRegistry.refresh(); for (const active of this.active.values()) { const { session } = active.runtime; session.modelRegistry.refresh(); this.syncCurrentModelAuthWarning(session, change.removedProviderId); this.publishStatus(session); } } private syncCurrentModelAuthWarning(session: PiAgentSession, removedProviderId: string | undefined): void { const model = session.model; if (model === undefined) return; if (model.provider === "unknown" && model.id === "unknown") return; const warningKey = authLossWarningKey(session.sessionId, model.provider, model.id); const registered = session.modelRegistry.find(model.provider, model.id); if (registered === undefined) return; if (session.modelRegistry.hasConfiguredAuth(registered)) { this.authLossWarnings.delete(warningKey); return; } if (removedProviderId === undefined || model.provider !== removedProviderId || this.authLossWarnings.has(warningKey)) return; this.authLossWarnings.add(warningKey); this.events.publish(session.sessionId, { type: "command.output", level: "error", message: `Authentication for ${model.provider}/${model.id} was removed. Use /model to select another model.`, }); } private clearAuthLossWarningsForSession(sessionId: string): void { const prefix = `${sessionId}:`; for (const key of this.authLossWarnings) { if (key.startsWith(prefix)) this.authLossWarnings.delete(key); } } private publishSessionName(session: PiAgentSession): void { const event = session.sessionName === undefined ? { type: "session.name", sessionId: session.sessionId } as const : { type: "session.name", sessionId: session.sessionId, name: session.sessionName } as const; this.events.publish(session.sessionId, event); this.events.publishGlobal(event); } private publishHeartbeats(): void { for (const active of this.active.values()) { const { session } = active.runtime; // Re-evaluate subsession completion here too: agent_end can arrive while // the session still reports active work transiently, so the event-driven // latch may not fire. The heartbeat re-checks once the session settles. this.updateSubsessionTracking(session); const activity = this.activities.get(session.sessionId); if (!this.hasActiveWork(session)) { if (activity?.phase === "active") this.publishStatus(session); continue; } this.publishStatus(session); if (activity?.phase === "active") this.publishActivity(session, activity.label, "active", activity.detail); else this.publishActivity(session, this.activityLabelFromStatus(session), "active"); } } private activityLabelFromStatus(session: PiAgentSession): string { if (session.isCompacting) return "compacting"; if (session.isBashRunning) return "running bash"; if (session.isStreaming) return "agent running"; if (this.pendingMessageCount(session) > 0) return "queued"; return "active"; } private hasActiveWork(session: PiAgentSession): boolean { return sessionHasActiveWork(session, this.compactionQueuedMessages(session.sessionId).length); } private publishActivityForEvent(session: PiAgentSession, event: unknown): void { const eventType = getString(event, "type"); if (eventType === undefined) return; if (eventType === "agent_start") { this.publishActivity(session, "agent running", "active"); return; } if (eventType === "agent_end") { this.publishActivity(session, "idle", "idle"); setTimeout(() => { this.publishActivity(session, "idle", "idle"); this.publishStatus(session); }, 250); return; } if (eventType === "turn_end") { this.publishActivity(session, "turn complete", "idle"); return; } if (eventType === "message_start") { this.publishActivity(session, "message started", "active"); return; } if (eventType === "message_end") { this.publishActivity(session, "message complete", "idle"); return; } if (eventType === "message_update") { this.publishActivity(session, "receiving response", "active"); return; } if (eventType === "tool_execution_start") { this.publishActivity(session, "running tool", "active", getString(event, "toolName")); return; } if (eventType === "tool_execution_end") { const isError = getBoolean(event, "isError") === true; this.publishActivity(session, isError ? "tool failed" : "tool complete", isError ? "error" : "idle", getString(event, "toolName")); return; } if (eventType === "bash_execution_start") { this.publishActivity(session, "running bash", "active"); return; } if (eventType === "bash_execution_end") { this.publishActivity(session, "bash complete", "idle"); return; } if (this.hasActiveWork(session)) this.publishActivity(session, eventType.replaceAll("_", " "), "active"); } private publishActivity(session: PiAgentSession, label: string, phase: "active" | "idle" | "error", detail?: string): void { const at = new Date().toISOString(); const stored = detail === undefined ? { phase, label, at } : { phase, label, detail, at }; this.activities.set(session.sessionId, stored); const activity = detail === undefined ? { sessionId: session.sessionId, phase, label, at } : { sessionId: session.sessionId, phase, label, detail, at }; this.workspaceActivity?.applySessionActivity(session.sessionManager.getCwd(), activity); this.events.publish(session.sessionId, { type: "activity.update", activity }); this.events.publishGlobal({ type: "activity.update", activity }); } private publishStatus(session: PiAgentSession): void { const status = this.statusFromSession(session); this.clearStaleActiveActivity(session); this.workspaceActivity?.applySessionStatus(session.sessionManager.getCwd(), status); this.events.publish(session.sessionId, { type: "status.update", status }); this.events.publishGlobal({ type: "status.update", status }); } private clearStaleActiveActivity(session: PiAgentSession): void { const current = this.activities.get(session.sessionId); if (current?.phase !== "active" || this.hasActiveWork(session)) return; const at = new Date().toISOString(); const stored = { phase: "idle" as const, label: "idle", at }; this.activities.set(session.sessionId, stored); const activity = { sessionId: session.sessionId, ...stored }; this.events.publish(session.sessionId, { type: "activity.update", activity }); this.events.publishGlobal({ type: "activity.update", activity }); } private statusFromSession(session: PiAgentSession): ClientSessionStatus { const stats = session.getSessionStats(); const model = session.model === undefined ? undefined : modelToClientModel(session.model); const contextUsage = session.getContextUsage(); return { sessionId: session.sessionId, ...(model === undefined ? {} : { model }), thinkingLevel: session.thinkingLevel, isStreaming: session.isStreaming, isCompacting: session.isCompacting, isBashRunning: session.isBashRunning, pendingMessageCount: this.pendingMessageCount(session), queuedMessages: queuedMessagesFromSession(session, this.compactionQueuedMessages(session.sessionId)), messageCount: session.messages.length, tokens: stats.tokens, cost: stats.cost, ...(contextUsage === undefined ? {} : { contextUsage }), }; } private pendingMessageCount(session: PiAgentSession): number { return session.pendingMessageCount + this.compactionQueuedMessages(session.sessionId).length; } private compactionQueuedMessages(sessionId: string): readonly QueuedPrompt[] { return this.compactionPromptQueues.get(sessionId) ?? []; } private hasQueuedMessageText(session: PiAgentSession, text: string): boolean { return queuedMessagesFromSession(session, this.compactionQueuedMessages(session.sessionId)).some((message) => message.text === text); } } function modelToClientModel(model: PiAgentSession["model"]): ClientSessionModel { if (model === undefined) return {}; const name = getString(model, "name"); const reasoning = getProperty(model, "reasoning"); return { provider: model.provider, id: model.id, ...(name === undefined ? {} : { name }), contextWindow: model.contextWindow, ...(reasoning === undefined ? {} : { reasoning }), }; } function clientSessionFromListEntry(session: PiSessionListEntry): ClientSession { return { id: session.id, path: session.path, cwd: session.cwd, ...(session.name === undefined ? {} : { name: session.name }), created: session.created.toISOString(), modified: session.modified.toISOString(), messageCount: session.messageCount, firstMessage: session.firstMessage, ...(session.parentSessionPath === undefined ? {} : { parentSessionPath: session.parentSessionPath }), }; } function archiveInputFromListEntry(session: PiSessionListEntry): ArchiveSessionInput { return { sessionId: session.id, cwd: session.cwd, path: session.path, created: session.created.toISOString(), modified: session.modified.toISOString(), messageCount: session.messageCount, firstMessage: session.firstMessage, ...(session.name === undefined ? {} : { name: session.name }), ...(session.parentSessionPath === undefined ? {} : { parentSessionPath: session.parentSessionPath }), }; } function archiveInputFromActiveSession(session: PiAgentSession): ArchiveSessionInput { const sessionFile = session.sessionFile; if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); const parentSessionPath = session.sessionManager.getHeader?.()?.parentSession; return { sessionId: session.sessionId, cwd: session.sessionManager.getCwd(), path: sessionFile, created: new Date().toISOString(), modified: new Date().toISOString(), messageCount: session.messages.length, firstMessage: "", ...(session.sessionName === undefined ? {} : { name: session.sessionName }), ...(parentSessionPath === undefined ? {} : { parentSessionPath }), }; } function archiveCandidateFromListEntry(session: PiSessionListEntry): WorkspaceArchiveCandidate { return { id: session.id, path: session.path, cwd: session.cwd, archived: false, listEntry: session, ...(session.parentSessionPath === undefined ? {} : { parentSessionPath: session.parentSessionPath }), }; } function archiveCandidateFromArchivedRecord(record: ArchivedSessionRecord, fallback: PiSessionListEntry | undefined): WorkspaceArchiveCandidate | undefined { const path = record.originalPath ?? fallback?.path; if (path === undefined) return undefined; const parentSessionPath = record.parentSessionPath ?? fallback?.parentSessionPath; return { id: record.sessionId, path, cwd: record.cwd, archived: true, ...(fallback === undefined ? {} : { listEntry: fallback }), ...(parentSessionPath === undefined ? {} : { parentSessionPath }), }; } function archiveCandidateFromActiveSession(session: PiAgentSession, archived: boolean): WorkspaceArchiveCandidate { const sessionFile = session.sessionFile; if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); const parentSessionPath = session.sessionManager.getHeader?.()?.parentSession; return { id: session.sessionId, path: sessionFile, cwd: session.sessionManager.getCwd(), archived, activeSession: session, ...(parentSessionPath === undefined ? {} : { parentSessionPath }), }; } function archiveInputFromCandidate(candidate: WorkspaceArchiveCandidate): ArchiveSessionInput { if (candidate.listEntry !== undefined) return archiveInputFromListEntry(candidate.listEntry); if (candidate.activeSession !== undefined) return archiveInputFromActiveSession(candidate.activeSession); throw new Error(`Session is not available for archiving: ${candidate.id}`); } function sessionHasActiveWork(session: PiAgentSession, extraQueuedMessageCount = 0): boolean { return session.isStreaming || session.isCompacting || session.isBashRunning || session.pendingMessageCount + extraQueuedMessageCount > 0; } function sessionDisplayName(session: PiAgentSession): string { return session.sessionName ?? session.sessionId; } function clientSessionFromArchivedRecord(record: ArchivedSessionRecord, fallback: PiSessionListEntry | undefined): ClientSession | undefined { const path = record.originalPath ?? fallback?.path; const created = record.created ?? fallback?.created.toISOString(); const modified = record.modified ?? fallback?.modified.toISOString(); const messageCount = record.messageCount ?? fallback?.messageCount; const firstMessage = record.firstMessage ?? fallback?.firstMessage; if (path === undefined || created === undefined || modified === undefined || messageCount === undefined || firstMessage === undefined) return undefined; const name = record.name ?? fallback?.name; const parentSessionPath = record.parentSessionPath ?? fallback?.parentSessionPath; return { id: record.sessionId, path, cwd: record.cwd, ...(name === undefined ? {} : { name }), created, modified, messageCount, firstMessage, ...(parentSessionPath === undefined ? {} : { parentSessionPath }), archived: true, archivedAt: record.archivedAt, }; } function addSessionName(names: Set, name: string | undefined): void { const trimmed = name?.replace(/\s+/g, " ").trim(); if (trimmed !== undefined && trimmed !== "") names.add(trimmed); } function compareArchivedRecords(a: ArchivedSessionRecord, b: ArchivedSessionRecord): number { return archivedTimestamp(b) - archivedTimestamp(a); } function archivedTimestamp(record: ArchivedSessionRecord): number { const time = Date.parse(record.archivedAt); return Number.isNaN(time) ? 0 : time; } 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"); const firstLine = newlineIndex === -1 ? content : content.slice(0, newlineIndex); const rest = newlineIndex === -1 ? "" : content.slice(newlineIndex); const header: unknown = JSON.parse(firstLine); if (!isRecord(header) || header["type"] !== "session") throw new Error("Invalid session file header"); if (header["parentSession"] === undefined) return; delete header["parentSession"]; 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(); } function queuedMessagesFromSession(session: PiAgentSession, extraQueuedMessages: readonly QueuedPrompt[] = []): { kind: "steer" | "followUp"; text: string }[] { return [ ...session.getSteeringMessages().map((text) => ({ kind: "steer" as const, text })), ...session.getFollowUpMessages().map((text) => ({ kind: "followUp" as const, text })), ...extraQueuedMessages, ]; } function userTextMessage(text: string): { role: "user"; content: string } { return { role: "user", content: text }; } /** * Build the optimistic user message echoed to clients. When images are present * we mirror pi's content-array shape (`[{type:"text"}, {type:"image"}, ...]`) so * the local echo matches what pi persists in the session branch. */ function userMessage(text: string, images: ImageContent[]): { role: "user"; content: string | (ImageContent | { type: "text"; text: string })[] } { if (images.length === 0) return userTextMessage(text); const content: (ImageContent | { type: "text"; text: string })[] = []; if (text !== "") content.push({ type: "text", text }); content.push(...images); return { role: "user", content }; } function buildPromptOptions(behavior: QueuedPromptKind | undefined, images: ImageContent[]): { streamingBehavior?: "steer" | "followUp"; images?: ImageContent[] } | undefined { const options: { streamingBehavior?: "steer" | "followUp"; images?: ImageContent[] } = {}; if (behavior !== undefined) options.streamingBehavior = behavior; if (images.length > 0) options.images = images; return Object.keys(options).length > 0 ? options : undefined; } function stringValue(value: unknown): string { return typeof value === "string" ? value : ""; } function historyMessages(session: PiAgentSession): unknown[] { const messages: unknown[] = []; for (const entry of session.sessionManager.getBranch()) { if (!isRecord(entry)) continue; if (entry["type"] === "message") messages.push(entry["message"]); else if (entry["type"] === "custom_message" && entry["display"] === true) messages.push({ role: "custom", content: entry["content"], customType: entry["customType"], details: entry["details"] }); else if (entry["type"] === "compaction") messages.push({ role: "system", source: "compaction", content: `Compacted history:\n\n${stringValue(entry["summary"])}` }); else if (entry["type"] === "branch_summary") messages.push({ role: "system", source: "branch_summary", content: `Branch summary:\n\n${stringValue(entry["summary"])}` }); } 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"; const SUBSESSION_NOTIFICATION_PREVIEW_CHARS = 2000; function truncateForNotification(text: string): string { if (text.length <= SUBSESSION_NOTIFICATION_PREVIEW_CHARS) return text; return `${text.slice(0, SUBSESSION_NOTIFICATION_PREVIEW_CHARS)}…`; } /** Most recent assistant text from a history message list, or "" if none. */ function finalAssistantText(messages: readonly unknown[]): string { for (let i = messages.length - 1; i >= 0; i--) { const message = messages[i]; if (!isRecord(message) || message["role"] !== "assistant") continue; const content = message["content"]; if (typeof content === "string") return content; if (!Array.isArray(content)) continue; const texts: string[] = []; for (const part of content) { if (isRecord(part) && part["type"] === "text" && typeof part["text"] === "string") texts.push(part["text"]); } if (texts.length > 0) return texts.join("\n").trim(); } return ""; } function toClientEvent(event: unknown): SessionUiEvent { const eventType = getString(event, "type"); const assistantMessageEvent = getProperty(event, "assistantMessageEvent"); if (eventType === "message_update" && getString(assistantMessageEvent, "type") === "text_delta") { return { type: "assistant.delta", text: getString(assistantMessageEvent, "delta") ?? "" }; } if (eventType === "message_update" && getString(assistantMessageEvent, "type") === "thinking_delta") { return { type: "assistant.thinking.delta", text: getString(assistantMessageEvent, "delta") ?? "" }; } if (eventType === "tool_execution_start") { const args = getProperty(event, "args"); return { type: "tool.start", toolName: getString(event, "toolName") ?? "", toolCallId: getString(event, "toolCallId") ?? "", summary: summarizeToolArgs(args), args }; } if (eventType === "tool_execution_update") { const partialResult = getProperty(event, "partialResult"); return { type: "tool.update", toolName: getString(event, "toolName") ?? "", toolCallId: getString(event, "toolCallId") ?? "", text: stringifyToolResult(partialResult), content: toolResultContent(partialResult), details: toolResultDetails(partialResult) }; } if (eventType === "tool_execution_end") { const result = getProperty(event, "result"); return { type: "tool.end", toolName: getString(event, "toolName") ?? "", toolCallId: getString(event, "toolCallId") ?? "", text: stringifyToolResult(result), content: toolResultContent(result), details: toolResultDetails(result), isError: getBoolean(event, "isError") === true }; } if (eventType === "agent_start") return { type: "agent.start" }; if (eventType === "agent_end") return { type: "agent.end" }; if (eventType === "message_end") { const message = getProperty(event, "message"); return message === undefined ? { type: "message.end" } : { type: "message.end", message }; } return { type: "pi.event", eventType: eventType ?? "unknown" }; } function summarizeToolArgs(args: unknown): string { if (!isRecord(args)) return stringifyPrimitive(args); const command = getString(args, "command"); if (command !== undefined) return command; const path = getString(args, "path"); if (path !== undefined) return path; if (typeof args["oldText"] === "string" && typeof args["newText"] === "string") return "edit text replacement"; const edits = args["edits"]; if (Array.isArray(edits)) return `${String(edits.length)} edit${edits.length === 1 ? "" : "s"}`; const entries = Object.entries(args).filter(([, value]) => value != null).slice(0, 3); return entries.map(([key, value]) => `${key}: ${shortToolValue(value)}`).join(" · "); } function shortToolValue(value: unknown): string { if (typeof value === "string") return value.length > 80 ? `${value.slice(0, 77)}…` : value; if (typeof value === "number" || typeof value === "boolean") return String(value); if (Array.isArray(value)) return `${String(value.length)} item${value.length === 1 ? "" : "s"}`; if (typeof value === "object" && value !== null) return "object"; return ""; } function toolResultContent(result: unknown): unknown { if (isRecord(result)) { const content = getProperty(result, "content"); if (content !== undefined) return content; const text = getString(result, "text") ?? getString(result, "output"); if (text !== undefined) return [{ type: "text", text }]; } if (typeof result === "string") return [{ type: "text", text: result }]; return result; } function toolResultDetails(result: unknown): unknown { return isRecord(result) ? getProperty(result, "details") : undefined; } function stringifyToolResult(result: unknown): string { if (typeof result === "string") return result; if (Array.isArray(result)) return result.map(stringifyToolResult).filter((text) => text !== "").join("\n"); if (isRecord(result)) { if (getString(result, "type") === "image") return "[image]"; const text = getString(result, "text") ?? getString(result, "content") ?? getString(result, "output"); if (text !== undefined) return text; const content = getProperty(result, "content"); if (Array.isArray(content)) return stringifyToolResult(content); return JSON.stringify(result, null, 2); } return stringifyPrimitive(result); } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } function getProperty(value: unknown, key: string): unknown { return isRecord(value) ? value[key] : undefined; } function getString(value: unknown, key: string): string | undefined { const property = getProperty(value, key); return typeof property === "string" ? property : undefined; } function getBoolean(value: unknown, key: string): boolean | undefined { const property = getProperty(value, key); return typeof property === "boolean" ? property : undefined; } function stringifyPrimitive(value: unknown): string { if (value == null) return ""; if (typeof value === "string") return value; if (typeof value === "number" || typeof value === "boolean" || typeof value === "bigint") return String(value); return ""; }