snowspeeder/pi-web
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sessions): add tracked subsessions behind a beta flag

Browse Source 
Add spawn_subsession / list_subsessions / read_subsession tools that let an
agent start child sessions it stays attached to: the child records its parent
in the session tree, the parent is notified (as a system-authored custom
message that wakes an idle parent and queues behind in-flight work) when the
child stops working, and the parent can inspect children's status and result.

Gated behind a beta flag, off by default, mirroring spawnSessions: enable via
PI_WEB_SUBSESSIONS, the subsessions config key, or the Settings toggle. Also
requires spawnSessions.

Also fix the release skill so the version step resyncs package-lock.json
(npm install --package-lock-only) and the commit step refuses a release where
package.json and package-lock.json versions disagree.
This commit is contained in:
Federico Jaramillo Martinez
2026-06-17 12:13:23 +02:00
parent ef454f2b65
commit 355ebe8cf8
17 changed files with 658 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/sessions/piSessionService.ts
+188 -4
View File
@@ -32,6 +32,7 @@ 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 SubsessionReadResult, type SubsessionStatus, type SubsessionSummary, type SubsessionToolDeps } from "./spawnSubsessionTool.js";
import type { SpawnTargetDecision, SpawnTargetResolver } from "./spawnTargetResolver.js";
/**
@@ -127,7 +128,7 @@ export interface PiSessionManager {
export interface PiSessionManagerGateway {
list(cwd: string): Promise<PiSessionListEntry[]>;
create(cwd: string): PiSessionManager;
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
@@ -172,6 +173,7 @@ export interface PiAgentSession {
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<void>;
sendCustomMessage(message: { customType: string; content: string; display: boolean; details?: unknown }, options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }): Promise<void>;
executeBash(command: string, onChunk?: (chunk: string) => void, options?: { excludeFromContext?: boolean }): Promise<{ output: string; exitCode: number | undefined; cancelled: boolean; truncated: boolean; fullOutputPath?: string }>;
abort(): Promise<void>;
clearQueue(): { steering: string[]; followUp: string[] };
@@ -208,12 +210,13 @@ function defaultCreateAgentRuntime(createRuntime: CreateAgentSessionRuntimeFacto
type SpawnSessionFn = (input: SpawnSessionInvocation) => Promise<SpawnSessionResult>;
function createDefaultRuntimeFactory(authStorage: AuthStorage, modelRegistry: ModelRegistryInstance, spawn?: SpawnSessionFn): CreateAgentSessionRuntimeFactory {
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 }
@@ -262,6 +265,13 @@ export interface PiSessionServiceDependencies {
* 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`, `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;
}
@@ -274,6 +284,16 @@ export class PiSessionService {
private readonly compactionPromptQueues = new Map<string, QueuedPrompt[]>();
private readonly compactionDrainTimers = new Map<string, NodeJS.Timeout>();
private readonly authLossWarnings = new Set<string>();
/** Tracked subsession id -> the parent session id that spawned it. */
private readonly subsessionParents = new Map<string, string>();
/** Parent session id -> the set of tracked subsession ids it spawned. */
private readonly subsessionChildren = new Map<string, Set<string>>();
/**
* 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<string, boolean>();
private readonly archiveStore: SessionArchiveRepository;
private readonly agentDir: string;
private readonly sessionManager: PiSessionManagerGateway;
@@ -291,10 +311,18 @@ export class PiSessionService {
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) => this.listSubsessions(parentSessionId),
read: (parentSessionId, sessionId) => this.readSubsession(parentSessionId, sessionId),
},
);
this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime;
this.workspaceActivity = deps.workspaceActivity;
@@ -329,6 +357,9 @@ export class PiSessionService {
this.activities.clear();
this.compactionPromptQueues.clear();
this.authLossWarnings.clear();
this.subsessionParents.clear();
this.subsessionChildren.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());
@@ -355,8 +386,8 @@ export class PiSessionService {
return [...unarchivedSessions, ...archivedSessions];
}
async start(cwd: string): Promise<ClientSession> {
const active = await this.create(this.sessionManager.create(cwd), cwd);
async start(cwd: string, parentSession?: string): Promise<ClientSession> {
const active = await this.create(this.sessionManager.create(cwd, parentSession === undefined ? undefined : { parentSession }), cwd);
const { session } = active.runtime;
const created: ClientSession = {
id: session.sessionId,
@@ -366,6 +397,9 @@ export class PiSessionService {
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.
@@ -391,6 +425,120 @@ export class PiSessionService {
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<SpawnSubsessionResult> {
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);
this.registerSubsession(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): Promise<SubsessionSummary[]> {
const childIds = this.subsessionChildren.get(parentSessionId);
if (childIds === undefined) return [];
return Promise.all([...childIds].map(async (childId) => ({ sessionId: childId, ...(await this.subsessionSummaryFields(childId)) })));
}
/** Status and final result of a subsession, scoped to the caller's children. */
async readSubsession(parentSessionId: string, sessionId: string): Promise<SubsessionReadResult> {
if (this.subsessionParents.get(sessionId) !== parentSessionId) {
throw new Error(`Session ${sessionId} is not one of your subsessions`);
}
const session = await this.getOrOpen(sessionId);
const messages = historyMessages(session);
return {
sessionId,
cwd: session.sessionManager.getCwd(),
status: await this.subsessionStatus(session),
finalText: finalAssistantText(messages),
messageCount: messages.length,
};
}
private registerSubsession(parentSessionId: string, childSessionId: string): void {
this.subsessionParents.set(childSessionId, parentSessionId);
const children = this.subsessionChildren.get(parentSessionId) ?? new Set<string>();
children.add(childSessionId);
this.subsessionChildren.set(parentSessionId, children);
this.subsessionNotifyArmed.set(childSessionId, false);
}
private async subsessionSummaryFields(childSessionId: string): Promise<{ cwd: string; status: SubsessionStatus }> {
const active = this.active.get(childSessionId);
if (active !== undefined) {
return { cwd: active.runtime.cwd, status: await this.subsessionStatus(active.runtime.session) };
}
const archived = await this.archiveStore.get(childSessionId);
if (archived !== undefined) return { cwd: archived.cwd, status: "archived" };
return { cwd: "", status: "unknown" };
}
private async subsessionStatus(session: PiAgentSession): Promise<SubsessionStatus> {
if (await this.archiveStore.isArchived(session.sessionId)) return "archived";
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 childId = session.sessionId;
const parentId = this.subsessionParents.get(childId);
if (parentId === undefined) return;
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 read_subsession with sessionId "${childId}" for the full result.`;
void this.notifyParentOfSubsession(parentId, childId, text);
}
/**
* 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<void> {
try {
const session = await this.getOrOpen(parentId);
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<unknown[] | ClientMessagePage> {
const session = await this.getOrOpen(ref);
return pageMessagesAtSafeBoundary(historyMessages(session), page);
@@ -751,6 +899,10 @@ export class PiSessionService {
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.
this.subsessionNotifyArmed.delete(sessionId);
clearSessionQueue(active.runtime.session);
active.unsubscribe();
try {
@@ -839,6 +991,7 @@ export class PiSessionService {
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);
}
@@ -969,6 +1122,10 @@ export class PiSessionService {
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);
@@ -1298,6 +1455,33 @@ function historyMessages(session: PiAgentSession): unknown[] {
return messages;
}
/** 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");