Archived
Federico Jaramillo Martinez
d17050e144
Support pasting (Ctrl/Cmd+V), drag-and-drop, and an Attach button to add PNG/JPEG/GIF/WebP images to a message, with thumbnail previews and multi-image support. Attachments are delivered to the session using pi's native ImageContent format and are run through pi's own resizeImage so they match pi's inline image limits exactly. Image content now renders inline in the transcript. A per-message delivery toggle also lets users save attachments into the workspace `.pi-web/paste` folder and reference them so the agent reads them with its own tools. The accepted HTTP upload size is configurable via PI_WEB_MAX_UPLOAD_BYTES or the maxUploadBytes config value (default 64 MB). Closes #13
1324 lines
60 KiB
TypeScript
1324 lines
60 KiB
TypeScript
import { 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";
|
|
|
|
function noop(): void {
|
|
// Intentionally empty default unsubscribe callback.
|
|
}
|
|
|
|
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<PiSessionRuntime>): boolean {
|
|
return !isPiSessionRef(ref) || cwdPathsEqual(active.runtime.cwd, ref.cwd);
|
|
}
|
|
|
|
type QueuedPromptKind = "steer" | "followUp";
|
|
|
|
interface QueuedPrompt {
|
|
kind: QueuedPromptKind;
|
|
text: string;
|
|
images?: ImageContent[];
|
|
}
|
|
|
|
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<SessionArchiveStore, "list" | "get" | "archive" | "restore" | "isArchived"> & { deleteArchived?: (sessionId: string) => Promise<void> };
|
|
|
|
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<Api>;
|
|
type ModelRegistryInstance = ReturnType<typeof ModelRegistry.create>;
|
|
|
|
export interface PiSessionManager {
|
|
getCwd(): string;
|
|
getBranch(): unknown[];
|
|
getLeafId(): string | null;
|
|
getHeader?(): { parentSession?: string } | null | undefined;
|
|
}
|
|
|
|
export interface PiSessionManagerGateway {
|
|
list(cwd: string): Promise<PiSessionListEntry[]>;
|
|
create(cwd: 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<PiSessionListEntry[]>;
|
|
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<void>;
|
|
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<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[] };
|
|
getSteeringMessages(): readonly string[];
|
|
getFollowUpMessages(): readonly string[];
|
|
setModel(model: AgentModel): Promise<void>;
|
|
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>): void;
|
|
fork(entryId: string, options?: { position?: "before" | "at" }): Promise<{ cancelled: boolean; selectedText?: string }>;
|
|
dispose(): Promise<void>;
|
|
}
|
|
|
|
interface CreateAgentRuntimeOptions {
|
|
cwd: string;
|
|
agentDir: string;
|
|
sessionManager: PiSessionManager;
|
|
}
|
|
|
|
type CreateAgentRuntime = (createRuntime: CreateAgentSessionRuntimeFactory, options: CreateAgentRuntimeOptions) => Promise<PiSessionRuntime>;
|
|
|
|
function defaultCreateAgentRuntime(createRuntime: CreateAgentSessionRuntimeFactory, options: CreateAgentRuntimeOptions): Promise<PiSessionRuntime> {
|
|
if (!(options.sessionManager instanceof SessionManager)) throw new Error("Default runtime creation requires an SDK SessionManager");
|
|
return createAgentSessionRuntime(createRuntime, { ...options, sessionManager: options.sessionManager });
|
|
}
|
|
|
|
function createDefaultRuntimeFactory(authStorage: AuthStorage, modelRegistry: ModelRegistryInstance): CreateAgentSessionRuntimeFactory {
|
|
return async ({ cwd, agentDir, sessionManager, sessionStartEvent }) => {
|
|
const services = await createAgentSessionServices({ cwd, agentDir, authStorage, modelRegistry });
|
|
const customTools = [createPiWebEditToolDefinition(cwd)];
|
|
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<typeof editTool.parameters, PiWebEditToolDetails>({
|
|
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<WorkspaceActivityService, "applySessionStatus" | "applySessionActivity" | "removeSession" | "reconcileSessionActivity">;
|
|
}
|
|
|
|
export class PiSessionService {
|
|
private readonly active = new Map<string, ActiveSession<PiSessionRuntime>>();
|
|
private readonly activities = new Map<string, { phase: "active" | "idle" | "error"; label: string; detail?: string; at: string }>();
|
|
private readonly heartbeat: NodeJS.Timeout;
|
|
private readonly commandService: SessionCommandService<PiAgentSession>;
|
|
private readonly compactionPromptQueues = new Map<string, QueuedPrompt[]>();
|
|
private readonly compactionDrainTimers = new Map<string, NodeJS.Timeout>();
|
|
private readonly authLossWarnings = new Set<string>();
|
|
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<WorkspaceActivityService, "applySessionStatus" | "applySessionActivity" | "removeSession" | "reconcileSessionActivity"> | undefined;
|
|
|
|
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.createRuntime = deps.createRuntime ?? createDefaultRuntimeFactory(this.modelRegistry.authStorage, this.modelRegistry);
|
|
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),
|
|
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<void> {
|
|
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();
|
|
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<ClientSession[]> {
|
|
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): Promise<ClientSession> {
|
|
const active = await this.create(this.sessionManager.create(cwd), cwd);
|
|
const { session } = active.runtime;
|
|
return {
|
|
id: session.sessionId,
|
|
path: session.sessionFile ?? "",
|
|
cwd,
|
|
created: new Date().toISOString(),
|
|
modified: new Date().toISOString(),
|
|
messageCount: session.messages.length,
|
|
firstMessage: "",
|
|
};
|
|
}
|
|
|
|
async messages(ref: PiSessionLookup, page?: { before?: number; limit?: number }): Promise<unknown[] | ClientMessagePage> {
|
|
const session = await this.getOrOpen(ref);
|
|
return pageMessagesAtSafeBoundary(historyMessages(session), page);
|
|
}
|
|
|
|
async status(ref: PiSessionLookup): Promise<ClientSessionStatus> {
|
|
return this.statusFromSession(await this.getOrOpen(ref));
|
|
}
|
|
|
|
async availableModels(ref: PiSessionLookup): Promise<ClientSessionModel[]> {
|
|
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<ClientSessionStatus> {
|
|
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<ClientSessionStatus> {
|
|
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<ClientThinkingLevel[]> {
|
|
const session = await this.getOrOpen(ref);
|
|
return session.getAvailableThinkingLevels();
|
|
}
|
|
|
|
async setThinkingLevel(ref: PiSessionLookup, level: ClientThinkingLevel): Promise<ClientSessionStatus> {
|
|
await this.assertWritable(ref);
|
|
const session = await this.getOrOpen(ref);
|
|
session.setThinkingLevel(level);
|
|
this.publishActivity(session, `thinking: ${session.thinkingLevel}`, "idle");
|
|
this.publishStatus(session);
|
|
return this.statusFromSession(session);
|
|
}
|
|
|
|
async cycleThinkingLevel(ref: PiSessionLookup): Promise<ClientSessionStatus> {
|
|
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<ClientCommand[]> {
|
|
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): Promise<void> {
|
|
const promptText = requirePromptText(text);
|
|
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);
|
|
return;
|
|
}
|
|
void this.submitPrompt(session, promptText, behavior, images);
|
|
}
|
|
|
|
private submitPrompt(session: PiAgentSession, text: string, behavior: QueuedPromptKind | undefined, images: ImageContent[] = []): Promise<void> {
|
|
this.publishActivity(session, behavior === "steer" ? "steering queued" : behavior === "followUp" ? "message queued" : "prompt accepted", "active");
|
|
if (behavior === undefined) 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[] = []): void {
|
|
const queue = this.compactionPromptQueues.get(session.sessionId) ?? [];
|
|
queue.push({ kind, text, ...(images.length > 0 ? { images } : {}) });
|
|
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<SavedPromptAttachment[]> {
|
|
const parsed = parsePromptAttachments(attachments, { enforceInlineSizeLimit: false });
|
|
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<void> {
|
|
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: !<shell command>");
|
|
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<ClientCommandResult> {
|
|
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<ClientCommandResult> {
|
|
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<void> {
|
|
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<ClientArchiveSessionsResponse> {
|
|
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<void> {
|
|
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<void> {
|
|
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 detachParent(ref: PiSessionLookup): Promise<void> {
|
|
const session = await this.getOrOpen(ref);
|
|
const sessionFile = session.sessionFile;
|
|
if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted");
|
|
await clearParentSession(sessionFile);
|
|
}
|
|
|
|
async abort(ref: PiSessionLookup): Promise<void> {
|
|
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, ArchivedSessionRecord>): 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<ArchivedSessionRecord> {
|
|
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<ArchivedSessionRecord> {
|
|
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<ArchiveSessionInput> {
|
|
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<WorkspaceArchiveCandidate[]> {
|
|
const [sessions, archivedRecords] = await Promise.all([this.sessionManager.list(cwd), this.archiveStore.list()]);
|
|
const candidates = new Map<string, WorkspaceArchiveCandidate>();
|
|
const archivedById = new Map<string, ArchivedSessionRecord>();
|
|
|
|
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<string[]> {
|
|
const [sessions, archivedRecords] = await Promise.all([this.sessionManager.list(cwd), this.archiveStore.list()]);
|
|
const names = new Set<string>();
|
|
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<void> {
|
|
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);
|
|
clearSessionQueue(active.runtime.session);
|
|
active.unsubscribe();
|
|
try {
|
|
await active.runtime.session.abort();
|
|
} finally {
|
|
await active.runtime.dispose();
|
|
}
|
|
}
|
|
|
|
private async assertWritable(ref: PiSessionLookup): Promise<void> {
|
|
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<PiAgentSession> {
|
|
return (await this.getActive(ref)).runtime.session;
|
|
}
|
|
|
|
private async getActive(ref: PiSessionLookup): Promise<ActiveSession<PiSessionRuntime>> {
|
|
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<ArchivedSessionRecord | undefined> {
|
|
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<PiSessionRuntime> | 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<ActiveSession<PiSessionRuntime>> {
|
|
const runtime = await this.createAgentRuntime(this.createRuntime, { cwd, agentDir: this.agentDir, sessionManager });
|
|
await this.bindSessionExtensions(runtime.session);
|
|
const active: ActiveSession<PiSessionRuntime> = { runtime, unsubscribe: noop };
|
|
this.bindRuntime(active);
|
|
runtime.setRebindSession(async (session) => {
|
|
await this.bindSessionExtensions(session);
|
|
this.bindRuntime(active);
|
|
});
|
|
this.active.set(runtime.session.sessionId, active);
|
|
this.publishStatus(runtime.session);
|
|
return active;
|
|
}
|
|
|
|
private async bindSessionExtensions(session: PiAgentSession): Promise<void> {
|
|
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<PiSessionRuntime>): 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.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);
|
|
return;
|
|
}
|
|
|
|
const prompt = this.shiftCompactionPrompt(sessionId);
|
|
if (prompt === undefined) return;
|
|
this.publishStatus(session);
|
|
const submitted = this.submitPrompt(session, prompt.text, undefined, prompt.images);
|
|
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;
|
|
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<string>, 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<T>(value: T | undefined): value is T {
|
|
return value !== undefined;
|
|
}
|
|
|
|
async function clearParentSession(sessionFile: string): Promise<void> {
|
|
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 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;
|
|
}
|
|
|
|
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<string, unknown> {
|
|
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 "";
|
|
}
|