diff --git a/apps/tui/lib/coding-agent/src/core/agent-observe.ts b/apps/tui/lib/coding-agent/src/core/agent-observe.ts deleted file mode 100644 index f3df7679..00000000 --- a/apps/tui/lib/coding-agent/src/core/agent-observe.ts +++ /dev/null @@ -1,207 +0,0 @@ -import type { AgentMessage } from "@earendil-works/pi-agent-core"; -import type { AgentFamilyRelationship } from "./agent-messages.js"; - -export const AGENT_OBSERVE_SKILL_NAME = "agent-observe"; -/** Shared cap for the message previews carried by roster rows. */ -export const AGENT_OBSERVE_PREVIEW_MAX_CHARS = 240; -export const AGENT_OBSERVE_IMPORT_NAME = "agent_observe"; -export const ORCHESTRATION_HEARTBEAT_SKILL_NAME = "orchestration-heartbeat"; - -export interface AgentObserveAgentSummary { - /** Absent for family members that have no live session in this daemon. */ - activeSessionId?: string; - sessionId: string; - sessionName?: string; - /** Absent only for the current agent. */ - relationship?: AgentFamilyRelationship; - runtimeKind?: "top-level" | "subagent"; - cwd?: string; - status: string; - isCurrent: boolean; - isStreaming: boolean; - isCompacting: boolean; - attachedClients: number; - messageCount?: number; - queuedCount: number; - isSessionActive: boolean; - repliedSinceTask?: boolean; - parentActiveSessionId?: string; - parentSessionId?: string; - rlmChildId?: string; - rlmParentNodeId?: string; - firstMessage?: string; - latestMessage?: AgentObserveMessagePreview; -} - -export interface AgentObserveListResult { - current: AgentObserveAgentSummary; - agents: AgentObserveAgentSummary[]; -} - -export interface AgentObserveAgentSnapshot { - agent: AgentObserveAgentSummary; -} - -export interface AgentObserveRecentMessagesInput { - target: string; - limit?: number; - maxChars?: number; -} - -export interface AgentObserveRecentMessagesResult { - agent: AgentObserveAgentSummary; - messages: AgentObserveMessagePreview[]; - limit: number; - maxChars: number; - truncated: boolean; -} - -export interface AgentObserveMessagePreview { - index: number; - role: string; - timestamp?: number; - text: string; - truncated: boolean; - toolCalls?: string[]; - customType?: string; -} - -export interface AgentObserveController { - listAgents(): AgentObserveListResult | Promise; - getAgent(target: string): AgentObserveAgentSnapshot | Promise; - recentMessages( - input: AgentObserveRecentMessagesInput, - ): AgentObserveRecentMessagesResult | Promise; -} - -export function createAgentObserveHostHandlers(controller: AgentObserveController) { - return { - "agent_observe.list": async () => controller.listAgents() as unknown as Record, - "agent_observe.get": async (payload: Record = {}) => { - if (typeof payload.target !== "string") { - throw new Error("agent_observe.get target must be a string"); - } - return (await controller.getAgent(payload.target)) as unknown as Record; - }, - "agent_observe.recent": async (payload: Record = {}) => { - if (typeof payload.target !== "string") { - throw new Error("agent_observe.recent target must be a string"); - } - return (await controller.recentMessages({ - target: payload.target, - limit: normalizeOptionalInteger(payload.limit, "agent_observe.recent limit"), - maxChars: normalizeOptionalInteger(payload.max_chars ?? payload.maxChars, "agent_observe.recent max_chars"), - })) as unknown as Record; - }, - }; -} - -export function normalizeObserveLimit(limit: number | undefined, defaultLimit = 8): number { - return clampInteger(limit ?? defaultLimit, 1, 50, "agent_observe limit"); -} - -export function normalizeObserveMaxChars(maxChars: number | undefined, defaultMaxChars = 800): number { - return clampInteger(maxChars ?? defaultMaxChars, 80, 2_000, "agent_observe max_chars"); -} - -export function createAgentObserveMessagePreview( - message: AgentMessage, - index: number, - maxChars: number, -): AgentObserveMessagePreview { - const text = messageText(message); - const clipped = truncate(text, maxChars); - const toolCalls = message.role === "assistant" ? assistantToolCalls(message) : undefined; - return { - index, - role: message.role, - ...(message.timestamp ? { timestamp: message.timestamp } : {}), - text: clipped.text, - truncated: clipped.truncated, - ...(toolCalls && toolCalls.length > 0 ? { toolCalls } : {}), - ...(message.role === "custom" ? { customType: message.customType } : {}), - }; -} - -function normalizeOptionalInteger(value: unknown, label: string): number | undefined { - if (value === undefined) { - return undefined; - } - if (typeof value !== "number" || !Number.isInteger(value)) { - throw new Error(`${label} must be an integer when provided`); - } - return value; -} - -function clampInteger(value: number, min: number, max: number, label: string): number { - if (!Number.isInteger(value)) { - throw new Error(`${label} must be an integer`); - } - if (value < min || value > max) { - throw new Error(`${label} must be between ${min} and ${max}`); - } - return value; -} - -function truncate(text: string, maxChars: number): { text: string; truncated: boolean } { - if (text.length <= maxChars) { - return { text, truncated: false }; - } - return { text: text.slice(0, maxChars), truncated: true }; -} - -function messageText(message: AgentMessage): string { - switch (message.role) { - case "user": - case "assistant": - return contentText(message.content); - case "toolResult": - return contentText(message.content); - case "bashExecution": - return [message.command, message.output].filter(Boolean).join("\n"); - case "custom": - return typeof message.content === "string" ? message.content : contentText(message.content); - case "branchSummary": - return message.summary; - case "compactionSummary": - return message.summary; - default: { - const exhaustive: never = message; - return JSON.stringify(exhaustive); - } - } -} - -function contentText(content: unknown): string { - if (typeof content === "string") { - return content; - } - if (!Array.isArray(content)) { - return ""; - } - return content - .map((block) => { - if (!block || typeof block !== "object" || !("type" in block)) { - return ""; - } - if (block.type === "text" && "text" in block && typeof block.text === "string") { - return block.text; - } - if (block.type === "thinking" && "thinking" in block && typeof block.thinking === "string") { - return block.thinking; - } - if (block.type === "image") { - return "[image]"; - } - if (block.type === "toolCall" && "name" in block && typeof block.name === "string") { - return `[tool_call:${block.name}]`; - } - return ""; - }) - .filter(Boolean) - .join("\n"); -} - -function assistantToolCalls(message: Extract): string[] { - return message.content.filter((block) => block.type === "toolCall").map((block) => block.name); -} diff --git a/apps/tui/lib/coding-agent/src/core/output-guard.ts b/apps/tui/lib/coding-agent/src/core/output-guard.ts deleted file mode 100644 index c0783760..00000000 --- a/apps/tui/lib/coding-agent/src/core/output-guard.ts +++ /dev/null @@ -1,74 +0,0 @@ -interface StdoutTakeoverState { - rawStdoutWrite: (chunk: string, callback?: (error?: Error | null) => void) => boolean; - rawStderrWrite: (chunk: string, callback?: (error?: Error | null) => void) => boolean; - originalStdoutWrite: typeof process.stdout.write; -} - -let stdoutTakeoverState: StdoutTakeoverState | undefined; - -export function takeOverStdout(): void { - if (stdoutTakeoverState) { - return; - } - - const rawStdoutWrite = process.stdout.write.bind(process.stdout) as StdoutTakeoverState["rawStdoutWrite"]; - const rawStderrWrite = process.stderr.write.bind(process.stderr) as StdoutTakeoverState["rawStderrWrite"]; - const originalStdoutWrite = process.stdout.write; - - process.stdout.write = (( - chunk: string | Uint8Array, - encodingOrCallback?: BufferEncoding | ((error?: Error | null) => void), - callback?: (error?: Error | null) => void, - ): boolean => { - if (typeof encodingOrCallback === "function") { - return rawStderrWrite(String(chunk), encodingOrCallback); - } - return rawStderrWrite(String(chunk), callback); - }) as typeof process.stdout.write; - - stdoutTakeoverState = { - rawStdoutWrite, - rawStderrWrite, - originalStdoutWrite, - }; -} - -export function restoreStdout(): void { - if (!stdoutTakeoverState) { - return; - } - - process.stdout.write = stdoutTakeoverState.originalStdoutWrite; - stdoutTakeoverState = undefined; -} - -export function isStdoutTakenOver(): boolean { - return stdoutTakeoverState !== undefined; -} - -export function writeRawStdout(text: string): void { - if (stdoutTakeoverState) { - stdoutTakeoverState.rawStdoutWrite(text); - return; - } - process.stdout.write(text); -} - -export async function flushRawStdout(): Promise { - if (stdoutTakeoverState) { - await new Promise((resolve, reject) => { - stdoutTakeoverState?.rawStdoutWrite("", (err) => { - if (err) reject(err); - else resolve(); - }); - }); - return; - } - - await new Promise((resolve, reject) => { - process.stdout.write("", (err) => { - if (err) reject(err); - else resolve(); - }); - }); -} diff --git a/apps/tui/lib/coding-agent/src/core/resource-loader.ts b/apps/tui/lib/coding-agent/src/core/resource-loader.ts deleted file mode 100644 index 5cb03a4b..00000000 --- a/apps/tui/lib/coding-agent/src/core/resource-loader.ts +++ /dev/null @@ -1,938 +0,0 @@ -import { existsSync, readdirSync, readFileSync, statSync } from "node:fs"; -import { homedir } from "node:os"; -import { join, resolve, sep } from "node:path"; -import chalk from "chalk"; -import { CONFIG_DIR_NAME, getBundledSkillsDir } from "../config.js"; -import { loadThemeFromPath, type Theme } from "../modes/interactive/theme/theme.js"; -import type { ResourceDiagnostic } from "./diagnostics.js"; - -export type { ResourceCollision, ResourceDiagnostic } from "./diagnostics.js"; - -import { canonicalizePath, isLocalPath } from "../utils/paths.js"; -import { createEventBus, type EventBus } from "./event-bus.js"; -import { createExtensionRuntime, loadExtensionFromFactory, loadExtensions } from "./extensions/loader.js"; -import type { Extension, ExtensionFactory, ExtensionRuntime, LoadExtensionsResult } from "./extensions/types.js"; -import { DefaultPackageManager, type PathMetadata } from "./package-manager.js"; -import type { PromptTemplate } from "./prompt-templates.js"; -import { loadPromptTemplates } from "./prompt-templates.js"; -import { SettingsManager } from "./settings-manager.js"; -import type { Skill } from "./skills.js"; -import { loadSkills } from "./skills.js"; -import { createSourceInfo, type SourceInfo } from "./source-info.js"; - -export interface ResourceExtensionPaths { - skillPaths?: Array<{ path: string; metadata: PathMetadata }>; - promptPaths?: Array<{ path: string; metadata: PathMetadata }>; - themePaths?: Array<{ path: string; metadata: PathMetadata }>; -} - -export interface ResourceLoader { - getExtensions(): LoadExtensionsResult; - getSkills(): { skills: Skill[]; diagnostics: ResourceDiagnostic[] }; - getPrompts(): { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] }; - getThemes(): { themes: Theme[]; diagnostics: ResourceDiagnostic[] }; - getAgentsFiles(): { agentsFiles: Array<{ path: string; content: string }> }; - getSystemPrompt(): string | undefined; - getAppendSystemPrompt(): string[]; - extendResources(paths: ResourceExtensionPaths): void; - reload(): Promise; -} - -function resolvePromptInput(input: string | undefined, description: string): string | undefined { - if (!input) { - return undefined; - } - - if (existsSync(input)) { - try { - return readFileSync(input, "utf-8"); - } catch (error) { - console.error(chalk.yellow(`Warning: Could not read ${description} file ${input}: ${error}`)); - return input; - } - } - - return input; -} - -function loadContextFileFromDir(dir: string): { path: string; content: string } | null { - const candidates = ["AGENTS.md", "AGENTS.MD", "CLAUDE.md", "CLAUDE.MD"]; - for (const filename of candidates) { - const filePath = join(dir, filename); - if (existsSync(filePath)) { - try { - return { - path: filePath, - content: readFileSync(filePath, "utf-8"), - }; - } catch (error) { - console.error(chalk.yellow(`Warning: Could not read ${filePath}: ${error}`)); - } - } - } - return null; -} - -export function loadProjectContextFiles(options: { - cwd: string; - agentDir: string; -}): Array<{ path: string; content: string }> { - const resolvedCwd = options.cwd; - const resolvedAgentDir = options.agentDir; - - const contextFiles: Array<{ path: string; content: string }> = []; - const seenPaths = new Set(); - - const globalContext = loadContextFileFromDir(resolvedAgentDir); - if (globalContext) { - contextFiles.push(globalContext); - seenPaths.add(globalContext.path); - } - - const ancestorContextFiles: Array<{ path: string; content: string }> = []; - - let currentDir = resolvedCwd; - const root = resolve("/"); - - while (true) { - const contextFile = loadContextFileFromDir(currentDir); - if (contextFile && !seenPaths.has(contextFile.path)) { - ancestorContextFiles.unshift(contextFile); - seenPaths.add(contextFile.path); - } - - if (currentDir === root) break; - - const parentDir = resolve(currentDir, ".."); - if (parentDir === currentDir) break; - currentDir = parentDir; - } - - contextFiles.push(...ancestorContextFiles); - - return contextFiles; -} - -export interface DefaultResourceLoaderOptions { - cwd: string; - agentDir: string; - settingsManager?: SettingsManager; - eventBus?: EventBus; - additionalExtensionPaths?: string[]; - additionalSkillPaths?: string[]; - additionalPromptTemplatePaths?: string[]; - additionalThemePaths?: string[]; - extensionFactories?: ExtensionFactory[]; - noExtensions?: boolean; - noSkills?: boolean; - noPromptTemplates?: boolean; - noThemes?: boolean; - noContextFiles?: boolean; - /** Directory of built-in skills shipped with the package. Defaults to the bundled skills dir; pass null to disable. */ - bundledSkillsDir?: string | null; - /** Extra force-exclude patterns for built-in skills (e.g. unauthenticated MCP integrations). */ - extraBuiltinSkillOverrides?: () => string[]; - systemPrompt?: string; - appendSystemPrompt?: string[]; - extensionsOverride?: (base: LoadExtensionsResult) => LoadExtensionsResult; - skillsOverride?: (base: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }) => { - skills: Skill[]; - diagnostics: ResourceDiagnostic[]; - }; - promptsOverride?: (base: { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] }) => { - prompts: PromptTemplate[]; - diagnostics: ResourceDiagnostic[]; - }; - themesOverride?: (base: { themes: Theme[]; diagnostics: ResourceDiagnostic[] }) => { - themes: Theme[]; - diagnostics: ResourceDiagnostic[]; - }; - agentsFilesOverride?: (base: { agentsFiles: Array<{ path: string; content: string }> }) => { - agentsFiles: Array<{ path: string; content: string }>; - }; - systemPromptOverride?: (base: string | undefined) => string | undefined; - appendSystemPromptOverride?: (base: string[]) => string[]; -} - -export class DefaultResourceLoader implements ResourceLoader { - private cwd: string; - private agentDir: string; - private settingsManager: SettingsManager; - private eventBus: EventBus; - private packageManager: DefaultPackageManager; - private bundledSkillsDir: string | null; - private additionalExtensionPaths: string[]; - private additionalSkillPaths: string[]; - private additionalPromptTemplatePaths: string[]; - private additionalThemePaths: string[]; - private extensionFactories: ExtensionFactory[]; - private noExtensions: boolean; - private noSkills: boolean; - private noPromptTemplates: boolean; - private noThemes: boolean; - private noContextFiles: boolean; - private systemPromptSource?: string; - private appendSystemPromptSource?: string[]; - private extensionsOverride?: (base: LoadExtensionsResult) => LoadExtensionsResult; - private skillsOverride?: (base: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }) => { - skills: Skill[]; - diagnostics: ResourceDiagnostic[]; - }; - private promptsOverride?: (base: { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] }) => { - prompts: PromptTemplate[]; - diagnostics: ResourceDiagnostic[]; - }; - private themesOverride?: (base: { themes: Theme[]; diagnostics: ResourceDiagnostic[] }) => { - themes: Theme[]; - diagnostics: ResourceDiagnostic[]; - }; - private agentsFilesOverride?: (base: { agentsFiles: Array<{ path: string; content: string }> }) => { - agentsFiles: Array<{ path: string; content: string }>; - }; - private systemPromptOverride?: (base: string | undefined) => string | undefined; - private appendSystemPromptOverride?: (base: string[]) => string[]; - - private extensionsResult: LoadExtensionsResult; - private loadedExtensionPaths: string[] = []; - private skills: Skill[]; - private skillDiagnostics: ResourceDiagnostic[]; - private prompts: PromptTemplate[]; - private promptDiagnostics: ResourceDiagnostic[]; - private themes: Theme[]; - private themeDiagnostics: ResourceDiagnostic[]; - private agentsFiles: Array<{ path: string; content: string }>; - private systemPrompt?: string; - private appendSystemPrompt: string[]; - private lastSkillPaths: string[]; - private extensionSkillSourceInfos: Map; - private extensionPromptSourceInfos: Map; - private extensionThemeSourceInfos: Map; - private lastPromptPaths: string[]; - private lastThemePaths: string[]; - - constructor(options: DefaultResourceLoaderOptions) { - this.cwd = options.cwd; - this.agentDir = options.agentDir; - this.settingsManager = options.settingsManager ?? SettingsManager.create(this.cwd, this.agentDir); - this.eventBus = options.eventBus ?? createEventBus(); - this.bundledSkillsDir = options.bundledSkillsDir === undefined ? getBundledSkillsDir() : options.bundledSkillsDir; - this.packageManager = new DefaultPackageManager({ - cwd: this.cwd, - agentDir: this.agentDir, - settingsManager: this.settingsManager, - bundledSkillsDir: this.bundledSkillsDir, - extraBuiltinSkillOverrides: options.extraBuiltinSkillOverrides, - }); - this.additionalExtensionPaths = options.additionalExtensionPaths ?? []; - this.additionalSkillPaths = options.additionalSkillPaths ?? []; - this.additionalPromptTemplatePaths = options.additionalPromptTemplatePaths ?? []; - this.additionalThemePaths = options.additionalThemePaths ?? []; - this.extensionFactories = options.extensionFactories ?? []; - this.noExtensions = options.noExtensions ?? false; - this.noSkills = options.noSkills ?? false; - this.noPromptTemplates = options.noPromptTemplates ?? false; - this.noThemes = options.noThemes ?? false; - this.noContextFiles = options.noContextFiles ?? false; - this.systemPromptSource = options.systemPrompt; - this.appendSystemPromptSource = options.appendSystemPrompt; - this.extensionsOverride = options.extensionsOverride; - this.skillsOverride = options.skillsOverride; - this.promptsOverride = options.promptsOverride; - this.themesOverride = options.themesOverride; - this.agentsFilesOverride = options.agentsFilesOverride; - this.systemPromptOverride = options.systemPromptOverride; - this.appendSystemPromptOverride = options.appendSystemPromptOverride; - - this.extensionsResult = { extensions: [], errors: [], runtime: createExtensionRuntime() }; - this.skills = []; - this.skillDiagnostics = []; - this.prompts = []; - this.promptDiagnostics = []; - this.themes = []; - this.themeDiagnostics = []; - this.agentsFiles = []; - this.appendSystemPrompt = []; - this.lastSkillPaths = []; - this.extensionSkillSourceInfos = new Map(); - this.extensionPromptSourceInfos = new Map(); - this.extensionThemeSourceInfos = new Map(); - this.lastPromptPaths = []; - this.lastThemePaths = []; - } - - getExtensions(): LoadExtensionsResult { - return this.extensionsResult; - } - - /** Extension file paths the last reload actually loaded (after settings overrides). */ - getLoadedExtensionPaths(): string[] { - return this.loadedExtensionPaths; - } - - getSkills(): { skills: Skill[]; diagnostics: ResourceDiagnostic[] } { - return { skills: this.skills, diagnostics: this.skillDiagnostics }; - } - - getPrompts(): { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] } { - return { prompts: this.prompts, diagnostics: this.promptDiagnostics }; - } - - getThemes(): { themes: Theme[]; diagnostics: ResourceDiagnostic[] } { - return { themes: this.themes, diagnostics: this.themeDiagnostics }; - } - - getAgentsFiles(): { agentsFiles: Array<{ path: string; content: string }> } { - return { agentsFiles: this.agentsFiles }; - } - - getSystemPrompt(): string | undefined { - return this.systemPrompt; - } - - getAppendSystemPrompt(): string[] { - return this.appendSystemPrompt; - } - - extendResources(paths: ResourceExtensionPaths): void { - const skillPaths = this.normalizeExtensionPaths(paths.skillPaths ?? []); - const promptPaths = this.normalizeExtensionPaths(paths.promptPaths ?? []); - const themePaths = this.normalizeExtensionPaths(paths.themePaths ?? []); - - for (const entry of skillPaths) { - this.extensionSkillSourceInfos.set(entry.path, createSourceInfo(entry.path, entry.metadata)); - } - for (const entry of promptPaths) { - this.extensionPromptSourceInfos.set(entry.path, createSourceInfo(entry.path, entry.metadata)); - } - for (const entry of themePaths) { - this.extensionThemeSourceInfos.set(entry.path, createSourceInfo(entry.path, entry.metadata)); - } - - if (skillPaths.length > 0) { - this.lastSkillPaths = this.mergePaths( - this.lastSkillPaths, - skillPaths.map((entry) => entry.path), - ); - this.updateSkillsFromPaths(this.lastSkillPaths); - } - - if (promptPaths.length > 0) { - this.lastPromptPaths = this.mergePaths( - this.lastPromptPaths, - promptPaths.map((entry) => entry.path), - ); - this.updatePromptsFromPaths(this.lastPromptPaths); - } - - if (themePaths.length > 0) { - this.lastThemePaths = this.mergePaths( - this.lastThemePaths, - themePaths.map((entry) => entry.path), - ); - this.updateThemesFromPaths(this.lastThemePaths); - } - } - - async reload(): Promise { - await this.settingsManager.reload(); - const resolvedPaths = await this.packageManager.resolve(); - const cliExtensionPaths = await this.packageManager.resolveExtensionSources(this.additionalExtensionPaths, { - temporary: true, - }); - const metadataByPath = new Map(); - - this.extensionSkillSourceInfos = new Map(); - this.extensionPromptSourceInfos = new Map(); - this.extensionThemeSourceInfos = new Map(); - - // Helper to extract enabled paths and store metadata - const getEnabledResources = ( - resources: Array<{ path: string; enabled: boolean; metadata: PathMetadata }>, - ): Array<{ path: string; enabled: boolean; metadata: PathMetadata }> => { - for (const r of resources) { - if (!metadataByPath.has(r.path)) { - metadataByPath.set(r.path, r.metadata); - } - } - return resources.filter((r) => r.enabled); - }; - - const getEnabledPaths = ( - resources: Array<{ path: string; enabled: boolean; metadata: PathMetadata }>, - ): string[] => getEnabledResources(resources).map((r) => r.path); - const enabledExtensions = getEnabledPaths(resolvedPaths.extensions); - const enabledSkillResources = getEnabledResources(resolvedPaths.skills); - const enabledPrompts = getEnabledPaths(resolvedPaths.prompts); - const enabledThemes = getEnabledPaths(resolvedPaths.themes); - - const mapSkillPath = (resource: { path: string; metadata: PathMetadata }): string => { - if (resource.metadata.source !== "auto" && resource.metadata.origin !== "package") { - return resource.path; - } - try { - const stats = statSync(resource.path); - if (!stats.isDirectory()) { - return resource.path; - } - } catch { - return resource.path; - } - const skillFile = join(resource.path, "SKILL.md"); - if (existsSync(skillFile)) { - if (!metadataByPath.has(skillFile)) { - metadataByPath.set(skillFile, resource.metadata); - } - return skillFile; - } - return resource.path; - }; - - const enabledSkills = enabledSkillResources.map(mapSkillPath); - - // Add CLI paths metadata - for (const r of cliExtensionPaths.extensions) { - if (!metadataByPath.has(r.path)) { - metadataByPath.set(r.path, { source: "cli", scope: "temporary", origin: "top-level" }); - } - } - for (const r of cliExtensionPaths.skills) { - if (!metadataByPath.has(r.path)) { - metadataByPath.set(r.path, { source: "cli", scope: "temporary", origin: "top-level" }); - } - } - - const cliEnabledExtensions = getEnabledPaths(cliExtensionPaths.extensions); - const cliEnabledSkills = getEnabledPaths(cliExtensionPaths.skills); - const cliEnabledPrompts = getEnabledPaths(cliExtensionPaths.prompts); - const cliEnabledThemes = getEnabledPaths(cliExtensionPaths.themes); - - const extensionPaths = this.noExtensions - ? cliEnabledExtensions - : this.mergePaths(cliEnabledExtensions, enabledExtensions); - - const extensionsResult = await loadExtensions(extensionPaths, this.cwd, this.eventBus); - // Set before inline factories run so a factory can see which file-based - // extensions actually loaded this cycle (e.g. the built-in Herdr reporter - // defers to Herdr's own file-based integration only when it is active). - this.loadedExtensionPaths = extensionPaths; - const inlineExtensions = await this.loadExtensionFactories(extensionsResult.runtime); - extensionsResult.extensions.push(...inlineExtensions.extensions); - extensionsResult.errors.push(...inlineExtensions.errors); - - // Detect extension conflicts (tools, commands, flags with same names from different extensions) - // Keep all extensions loaded. Conflicts are reported as diagnostics, and precedence is handled by load order. - const conflicts = this.detectExtensionConflicts(extensionsResult.extensions); - for (const conflict of conflicts) { - extensionsResult.errors.push({ path: conflict.path, error: conflict.message }); - } - - for (const p of this.additionalExtensionPaths) { - if (isLocalPath(p) && !existsSync(p)) { - extensionsResult.errors.push({ path: p, error: `Extension path does not exist: ${p}` }); - } - } - this.extensionsResult = this.extensionsOverride ? this.extensionsOverride(extensionsResult) : extensionsResult; - this.applyExtensionSourceInfo(this.extensionsResult.extensions, metadataByPath); - - const skillPaths = this.noSkills - ? this.mergePaths(cliEnabledSkills, this.additionalSkillPaths) - : this.mergePaths([...cliEnabledSkills, ...this.additionalSkillPaths], enabledSkills); - - this.lastSkillPaths = skillPaths; - this.updateSkillsFromPaths(skillPaths, metadataByPath); - // Surface resolution-time skill warnings (e.g. missing bundled skills dir). - this.skillDiagnostics.push(...resolvedPaths.diagnostics); - for (const p of this.additionalSkillPaths) { - if (isLocalPath(p) && !existsSync(p) && !this.skillDiagnostics.some((d) => d.path === p)) { - this.skillDiagnostics.push({ type: "error", message: "Skill path does not exist", path: p }); - } - } - - const promptPaths = this.noPromptTemplates - ? this.mergePaths(cliEnabledPrompts, this.additionalPromptTemplatePaths) - : this.mergePaths([...cliEnabledPrompts, ...enabledPrompts], this.additionalPromptTemplatePaths); - - this.lastPromptPaths = promptPaths; - this.updatePromptsFromPaths(promptPaths, metadataByPath); - for (const p of this.additionalPromptTemplatePaths) { - if (isLocalPath(p) && !existsSync(p) && !this.promptDiagnostics.some((d) => d.path === p)) { - this.promptDiagnostics.push({ type: "error", message: "Prompt template path does not exist", path: p }); - } - } - - const themePaths = this.noThemes - ? this.mergePaths(cliEnabledThemes, this.additionalThemePaths) - : this.mergePaths([...cliEnabledThemes, ...enabledThemes], this.additionalThemePaths); - - this.lastThemePaths = themePaths; - this.updateThemesFromPaths(themePaths, metadataByPath); - for (const p of this.additionalThemePaths) { - if (!existsSync(p) && !this.themeDiagnostics.some((d) => d.path === p)) { - this.themeDiagnostics.push({ type: "error", message: "Theme path does not exist", path: p }); - } - } - - const agentsFiles = { - agentsFiles: this.noContextFiles ? [] : loadProjectContextFiles({ cwd: this.cwd, agentDir: this.agentDir }), - }; - const resolvedAgentsFiles = this.agentsFilesOverride ? this.agentsFilesOverride(agentsFiles) : agentsFiles; - this.agentsFiles = resolvedAgentsFiles.agentsFiles; - - const baseSystemPrompt = resolvePromptInput( - this.systemPromptSource ?? this.discoverSystemPromptFile(), - "system prompt", - ); - this.systemPrompt = this.systemPromptOverride ? this.systemPromptOverride(baseSystemPrompt) : baseSystemPrompt; - - const appendSources = - this.appendSystemPromptSource ?? - (this.discoverAppendSystemPromptFile() ? [this.discoverAppendSystemPromptFile()!] : []); - const baseAppend = appendSources - .map((s) => resolvePromptInput(s, "append system prompt")) - .filter((s): s is string => s !== undefined); - this.appendSystemPrompt = this.appendSystemPromptOverride - ? this.appendSystemPromptOverride(baseAppend) - : baseAppend; - } - - private normalizeExtensionPaths( - entries: Array<{ path: string; metadata: PathMetadata }>, - ): Array<{ path: string; metadata: PathMetadata }> { - return entries.map((entry) => ({ - path: this.resolveResourcePath(entry.path), - metadata: entry.metadata, - })); - } - - private updateSkillsFromPaths(skillPaths: string[], metadataByPath?: Map): void { - let skillsResult: { skills: Skill[]; diagnostics: ResourceDiagnostic[] }; - if (this.noSkills && skillPaths.length === 0) { - skillsResult = { skills: [], diagnostics: [] }; - } else { - skillsResult = loadSkills({ - cwd: this.cwd, - agentDir: this.agentDir, - skillPaths, - includeDefaults: false, - }); - } - const resolvedSkills = this.skillsOverride ? this.skillsOverride(skillsResult) : skillsResult; - this.skills = resolvedSkills.skills.map((skill) => ({ - ...skill, - sourceInfo: - this.findSourceInfoForPath(skill.filePath, this.extensionSkillSourceInfos, metadataByPath) ?? - skill.sourceInfo ?? - this.getDefaultSourceInfoForPath(skill.filePath), - })); - this.skillDiagnostics = resolvedSkills.diagnostics; - } - - private updatePromptsFromPaths(promptPaths: string[], metadataByPath?: Map): void { - let promptsResult: { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] }; - if (this.noPromptTemplates && promptPaths.length === 0) { - promptsResult = { prompts: [], diagnostics: [] }; - } else { - const allPrompts = loadPromptTemplates({ - cwd: this.cwd, - agentDir: this.agentDir, - promptPaths, - includeDefaults: false, - }); - promptsResult = this.dedupePrompts(allPrompts); - } - const resolvedPrompts = this.promptsOverride ? this.promptsOverride(promptsResult) : promptsResult; - this.prompts = resolvedPrompts.prompts.map((prompt) => ({ - ...prompt, - sourceInfo: - this.findSourceInfoForPath(prompt.filePath, this.extensionPromptSourceInfos, metadataByPath) ?? - prompt.sourceInfo ?? - this.getDefaultSourceInfoForPath(prompt.filePath), - })); - this.promptDiagnostics = resolvedPrompts.diagnostics; - } - - private updateThemesFromPaths(themePaths: string[], metadataByPath?: Map): void { - let themesResult: { themes: Theme[]; diagnostics: ResourceDiagnostic[] }; - if (this.noThemes && themePaths.length === 0) { - themesResult = { themes: [], diagnostics: [] }; - } else { - const loaded = this.loadThemes(themePaths, false); - const deduped = this.dedupeThemes(loaded.themes); - themesResult = { themes: deduped.themes, diagnostics: [...loaded.diagnostics, ...deduped.diagnostics] }; - } - const resolvedThemes = this.themesOverride ? this.themesOverride(themesResult) : themesResult; - this.themes = resolvedThemes.themes.map((theme) => { - const sourcePath = theme.sourcePath; - theme.sourceInfo = sourcePath - ? (this.findSourceInfoForPath(sourcePath, this.extensionThemeSourceInfos, metadataByPath) ?? - theme.sourceInfo ?? - this.getDefaultSourceInfoForPath(sourcePath)) - : theme.sourceInfo; - return theme; - }); - this.themeDiagnostics = resolvedThemes.diagnostics; - } - - private applyExtensionSourceInfo(extensions: Extension[], metadataByPath: Map): void { - for (const extension of extensions) { - extension.sourceInfo = - this.findSourceInfoForPath(extension.path, undefined, metadataByPath) ?? - this.getDefaultSourceInfoForPath(extension.path); - for (const command of extension.commands.values()) { - command.sourceInfo = extension.sourceInfo; - } - for (const tool of extension.tools.values()) { - tool.sourceInfo = extension.sourceInfo; - } - } - } - - private findSourceInfoForPath( - resourcePath: string, - extraSourceInfos?: Map, - metadataByPath?: Map, - ): SourceInfo | undefined { - if (!resourcePath) { - return undefined; - } - - if (resourcePath.startsWith("<")) { - return this.getDefaultSourceInfoForPath(resourcePath); - } - - const normalizedResourcePath = resolve(resourcePath); - if (extraSourceInfos) { - for (const [sourcePath, sourceInfo] of extraSourceInfos.entries()) { - const normalizedSourcePath = resolve(sourcePath); - if ( - normalizedResourcePath === normalizedSourcePath || - normalizedResourcePath.startsWith(`${normalizedSourcePath}${sep}`) - ) { - return { ...sourceInfo, path: resourcePath }; - } - } - } - - if (metadataByPath) { - const exact = metadataByPath.get(normalizedResourcePath) ?? metadataByPath.get(resourcePath); - if (exact) { - return createSourceInfo(resourcePath, exact); - } - - for (const [sourcePath, metadata] of metadataByPath.entries()) { - const normalizedSourcePath = resolve(sourcePath); - if ( - normalizedResourcePath === normalizedSourcePath || - normalizedResourcePath.startsWith(`${normalizedSourcePath}${sep}`) - ) { - return createSourceInfo(resourcePath, metadata); - } - } - } - - return undefined; - } - - private getDefaultSourceInfoForPath(filePath: string): SourceInfo { - if (filePath.startsWith("<") && filePath.endsWith(">")) { - return { - path: filePath, - source: filePath.slice(1, -1).split(":")[0] || "temporary", - scope: "temporary", - origin: "top-level", - }; - } - - const normalizedPath = resolve(filePath); - const agentRoots = [ - join(this.agentDir, "skills"), - join(this.agentDir, "prompts"), - join(this.agentDir, "themes"), - join(this.agentDir, "extensions"), - ]; - const projectRoots = [ - join(this.cwd, CONFIG_DIR_NAME, "skills"), - join(this.cwd, CONFIG_DIR_NAME, "prompts"), - join(this.cwd, CONFIG_DIR_NAME, "themes"), - join(this.cwd, CONFIG_DIR_NAME, "extensions"), - ]; - - for (const root of agentRoots) { - if (this.isUnderPath(normalizedPath, root)) { - return { path: filePath, source: "local", scope: "user", origin: "top-level", baseDir: root }; - } - } - - for (const root of projectRoots) { - if (this.isUnderPath(normalizedPath, root)) { - return { path: filePath, source: "local", scope: "project", origin: "top-level", baseDir: root }; - } - } - - return { - path: filePath, - source: "local", - scope: "temporary", - origin: "top-level", - baseDir: statSync(normalizedPath).isDirectory() ? normalizedPath : resolve(normalizedPath, ".."), - }; - } - - private mergePaths(primary: string[], additional: string[]): string[] { - const merged: string[] = []; - const seen = new Set(); - - for (const p of [...primary, ...additional]) { - const resolved = this.resolveResourcePath(p); - const canonicalPath = canonicalizePath(resolved); - if (seen.has(canonicalPath)) continue; - seen.add(canonicalPath); - merged.push(resolved); - } - - return merged; - } - - private resolveResourcePath(p: string): string { - const trimmed = p.trim(); - let expanded = trimmed; - if (trimmed === "~") { - expanded = homedir(); - } else if (trimmed.startsWith("~/")) { - expanded = join(homedir(), trimmed.slice(2)); - } else if (trimmed.startsWith("~")) { - expanded = join(homedir(), trimmed.slice(1)); - } - return resolve(this.cwd, expanded); - } - - private loadThemes( - paths: string[], - includeDefaults: boolean = true, - ): { - themes: Theme[]; - diagnostics: ResourceDiagnostic[]; - } { - const themes: Theme[] = []; - const diagnostics: ResourceDiagnostic[] = []; - if (includeDefaults) { - const defaultDirs = [join(this.agentDir, "themes"), join(this.cwd, CONFIG_DIR_NAME, "themes")]; - - for (const dir of defaultDirs) { - this.loadThemesFromDir(dir, themes, diagnostics); - } - } - - for (const p of paths) { - const resolved = resolve(this.cwd, p); - if (!existsSync(resolved)) { - diagnostics.push({ type: "warning", message: "theme path does not exist", path: resolved }); - continue; - } - - try { - const stats = statSync(resolved); - if (stats.isDirectory()) { - this.loadThemesFromDir(resolved, themes, diagnostics); - } else if (stats.isFile() && resolved.endsWith(".json")) { - this.loadThemeFromFile(resolved, themes, diagnostics); - } else { - diagnostics.push({ type: "warning", message: "theme path is not a json file", path: resolved }); - } - } catch (error) { - const message = error instanceof Error ? error.message : "failed to read theme path"; - diagnostics.push({ type: "warning", message, path: resolved }); - } - } - - return { themes, diagnostics }; - } - - private loadThemesFromDir(dir: string, themes: Theme[], diagnostics: ResourceDiagnostic[]): void { - if (!existsSync(dir)) { - return; - } - - try { - const entries = readdirSync(dir, { withFileTypes: true }); - for (const entry of entries) { - let isFile = entry.isFile(); - if (entry.isSymbolicLink()) { - try { - isFile = statSync(join(dir, entry.name)).isFile(); - } catch { - continue; - } - } - if (!isFile) { - continue; - } - if (!entry.name.endsWith(".json")) { - continue; - } - this.loadThemeFromFile(join(dir, entry.name), themes, diagnostics); - } - } catch (error) { - const message = error instanceof Error ? error.message : "failed to read theme directory"; - diagnostics.push({ type: "warning", message, path: dir }); - } - } - - private loadThemeFromFile(filePath: string, themes: Theme[], diagnostics: ResourceDiagnostic[]): void { - try { - themes.push(loadThemeFromPath(filePath)); - } catch (error) { - const message = error instanceof Error ? error.message : "failed to load theme"; - diagnostics.push({ type: "warning", message, path: filePath }); - } - } - - private async loadExtensionFactories(runtime: ExtensionRuntime): Promise<{ - extensions: Extension[]; - errors: Array<{ path: string; error: string }>; - }> { - const extensions: Extension[] = []; - const errors: Array<{ path: string; error: string }> = []; - - for (const [index, factory] of this.extensionFactories.entries()) { - const extensionPath = ``; - try { - const extension = await loadExtensionFromFactory(factory, this.cwd, this.eventBus, runtime, extensionPath); - extensions.push(extension); - } catch (error) { - const message = error instanceof Error ? error.message : "failed to load extension"; - errors.push({ path: extensionPath, error: message }); - } - } - - return { extensions, errors }; - } - - private dedupePrompts(prompts: PromptTemplate[]): { prompts: PromptTemplate[]; diagnostics: ResourceDiagnostic[] } { - const seen = new Map(); - const diagnostics: ResourceDiagnostic[] = []; - - for (const prompt of prompts) { - const existing = seen.get(prompt.name); - if (existing) { - diagnostics.push({ - type: "collision", - message: `name "/${prompt.name}" collision`, - path: prompt.filePath, - collision: { - resourceType: "prompt", - name: prompt.name, - winnerPath: existing.filePath, - loserPath: prompt.filePath, - }, - }); - } else { - seen.set(prompt.name, prompt); - } - } - - return { prompts: Array.from(seen.values()), diagnostics }; - } - - private dedupeThemes(themes: Theme[]): { themes: Theme[]; diagnostics: ResourceDiagnostic[] } { - const seen = new Map(); - const diagnostics: ResourceDiagnostic[] = []; - - for (const t of themes) { - const name = t.name ?? "unnamed"; - const existing = seen.get(name); - if (existing) { - diagnostics.push({ - type: "collision", - message: `name "${name}" collision`, - path: t.sourcePath, - collision: { - resourceType: "theme", - name, - winnerPath: existing.sourcePath ?? "", - loserPath: t.sourcePath ?? "", - }, - }); - } else { - seen.set(name, t); - } - } - - return { themes: Array.from(seen.values()), diagnostics }; - } - - private discoverSystemPromptFile(): string | undefined { - const projectPath = join(this.cwd, CONFIG_DIR_NAME, "SYSTEM.md"); - if (existsSync(projectPath)) { - return projectPath; - } - - const globalPath = join(this.agentDir, "SYSTEM.md"); - if (existsSync(globalPath)) { - return globalPath; - } - - return undefined; - } - - private discoverAppendSystemPromptFile(): string | undefined { - const projectPath = join(this.cwd, CONFIG_DIR_NAME, "APPEND_SYSTEM.md"); - if (existsSync(projectPath)) { - return projectPath; - } - - const globalPath = join(this.agentDir, "APPEND_SYSTEM.md"); - if (existsSync(globalPath)) { - return globalPath; - } - - return undefined; - } - - private isUnderPath(target: string, root: string): boolean { - const normalizedRoot = resolve(root); - if (target === normalizedRoot) { - return true; - } - const prefix = normalizedRoot.endsWith(sep) ? normalizedRoot : `${normalizedRoot}${sep}`; - return target.startsWith(prefix); - } - - private detectExtensionConflicts(extensions: Extension[]): Array<{ path: string; message: string }> { - const conflicts: Array<{ path: string; message: string }> = []; - - // Track which extension registered each tool and flag - const toolOwners = new Map(); - const flagOwners = new Map(); - - for (const ext of extensions) { - // Check tools - for (const toolName of ext.tools.keys()) { - const existingOwner = toolOwners.get(toolName); - if (existingOwner && existingOwner !== ext.path) { - conflicts.push({ - path: ext.path, - message: `Tool "${toolName}" conflicts with ${existingOwner}`, - }); - } else { - toolOwners.set(toolName, ext.path); - } - } - - // Check flags - for (const flagName of ext.flags.keys()) { - const existingOwner = flagOwners.get(flagName); - if (existingOwner && existingOwner !== ext.path) { - conflicts.push({ - path: ext.path, - message: `Flag "--${flagName}" conflicts with ${existingOwner}`, - }); - } else { - flagOwners.set(flagName, ext.path); - } - } - } - - return conflicts; - } -} diff --git a/apps/tui/lib/coding-agent/src/core/side-question.ts b/apps/tui/lib/coding-agent/src/core/side-question.ts deleted file mode 100644 index 0b7ab103..00000000 --- a/apps/tui/lib/coding-agent/src/core/side-question.ts +++ /dev/null @@ -1,215 +0,0 @@ -import { Agent, type AgentMessage } from "@earendil-works/pi-agent-core"; -import type { AssistantMessage, UserMessage } from "@earendil-works/pi-ai"; -import { - completeWithProviderRetry, - DEFAULT_PROVIDER_RETRY_POLICY, - type ProviderRetryPolicy, -} from "./provider-retry.js"; -import { unwrapSemanticEdgeStreamFn } from "./semantic-edges.js"; - -export type SideQuestionStatus = "running" | "complete" | "cancelled" | "error"; - -export interface SideQuestionEvent { - id: string; - question: string; - answer: string; - status: SideQuestionStatus; - errorMessage?: string; -} - -export interface SideQuestionTurn { - question: string; - answer: string; -} - -export interface SideQuestionRun { - done: Promise; - abort(): void; -} - -const SIDE_QUESTION_INSTRUCTION = - "The user asked this via `/btw` — a temporary side thread cloned from the main conversation to answer a question without interrupting the main work. Tools (including `ipython`) are deactivated in this side thread and return an error if called; answer using only the conversation context above. The user may send follow-up side questions. Nothing here is added to the main session, so don't start or plan main-session work from this thread."; - -const SIDE_QUESTION_TOOL_BLOCKED = "Tools are deactivated in this side thread. Answer from the conversation context."; - -/** Backstop for a model that keeps calling deactivated tools instead of answering. */ -const SIDE_QUESTION_MAX_TURNS = 3; - -function sideQuestionPrompt(question: string, isFirstTurn: boolean): string { - const body = isFirstTurn ? `${SIDE_QUESTION_INSTRUCTION}\n\n${question}` : question; - return `\n${body}\n`; -} - -function readAssistantText(message: AgentMessage): string { - if (message.role !== "assistant") { - return ""; - } - return message.content - .filter((block) => block.type === "text") - .map((block) => block.text) - .join(""); -} - -export function startSideQuestion( - parent: Agent, - id: string, - question: string, - onEvent: (event: SideQuestionEvent) => void | Promise, - previousTurns: SideQuestionTurn[] = [], - retry: ProviderRetryPolicy = DEFAULT_PROVIDER_RETRY_POLICY, -): SideQuestionRun { - const model = parent.state.model; - if (!model) { - throw new Error("Select a model before asking a side question"); - } - - // Each turn re-clones the live main conversation, so follow-ups always see - // the newest main-thread context; earlier side turns are replayed after it. - const previousTurnMessages: AgentMessage[] = previousTurns.flatMap((turn, index) => [ - { - role: "user", - content: [{ type: "text", text: sideQuestionPrompt(turn.question, index === 0) }], - timestamp: Date.now(), - } satisfies UserMessage, - { - role: "assistant", - content: [{ type: "text", text: turn.answer }], - api: model.api, - provider: model.provider, - model: model.id, - usage: { - input: 0, - output: 0, - cacheRead: 0, - cacheWrite: 0, - totalTokens: 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, - }, - stopReason: "stop", - timestamp: Date.now(), - } satisfies AssistantMessage, - ]); - - let turnCount = 0; - const sideAgent = new Agent({ - initialState: { - model, - systemPrompt: parent.state.systemPrompt, - messages: [...structuredClone(parent.state.messages), ...previousTurnMessages], - // Anthropic message-level caching keys on the thinking parameters, so a - // different level here would re-read the whole cloned conversation. - thinkingLevel: parent.state.thinkingLevel, - serviceTier: parent.state.serviceTier, - // Providers serialize tool declarations ahead of the cached prefix, so an - // empty list would miss the main cache; execution is blocked in beforeToolCall. - tools: parent.state.tools, - }, - convertToLlm: parent.convertToLlm, - transformContext: parent.transformContext, - // Side questions are excluded from session history; their calls carry no provenance. - streamFn: unwrapSemanticEdgeStreamFn(parent.streamFn), - getApiKey: parent.getApiKey, - onPayload: parent.onPayload, - onResponse: parent.onResponse, - beforeToolCall: async () => ({ block: true, reason: SIDE_QUESTION_TOOL_BLOCKED }), - shouldStopAfterTurn: ({ message }) => { - turnCount += 1; - return turnCount >= SIDE_QUESTION_MAX_TURNS || !message.content.some((block) => block.type === "toolCall"); - }, - sessionId: parent.sessionId, - thinkingBudgets: parent.thinkingBudgets, - transport: "sse", - toolExecution: parent.toolExecution, - }); - - const clonedMessageCount = sideAgent.state.messages.length; - // A turn-capped run can end on tool results, so its outcome lives in the - // assistant turns it appended rather than in its last message. - const assistantTurns = () => - sideAgent.state.messages - .slice(clonedMessageCount) - .filter((message): message is AssistantMessage => message.role === "assistant"); - let answer = ""; - let abortRequested = false; - let started = false; - const retryAbortController = new AbortController(); - const emit = (status: SideQuestionStatus, errorMessage?: string) => - onEvent({ id, question, answer, status, ...(errorMessage ? { errorMessage } : {}) }); - - // Streaming events carry one partial turn at a time, so they only fill in text - // as it arrives; the answer of the whole run is derived from its finished turns. - const unsubscribe = sideAgent.subscribe(async (event) => { - if (event.type !== "message_update" && event.type !== "message_end") { - return; - } - const nextAnswer = readAssistantText(event.message); - if (!nextAnswer || nextAnswer === answer) { - return; - } - answer = nextAnswer; - await emit("running"); - }); - - const prompt = sideQuestionPrompt(question, previousTurns.length === 0); - const done = Promise.resolve() - .then(() => emit("running")) - .then(async () => { - if (abortRequested) { - await emit("cancelled"); - return; - } - started = true; - // Standalone side agents bypass the session auto-retry loop; retry here instead. - let promptedOnce = false; - const finalTurn = await completeWithProviderRetry( - async () => { - if (promptedOnce) { - // Session-loop recovery: drop the failed assistant turn and re-run. - sideAgent.state.messages = sideAgent.state.messages.slice(0, -1); - await sideAgent.continue(); - } else { - promptedOnce = true; - await sideAgent.prompt(prompt); - } - const last = assistantTurns().at(-1); - if (!last) { - throw new Error(sideAgent.state.errorMessage || "Side question produced no assistant message"); - } - return last; - }, - { policy: retry, signal: retryAbortController.signal }, - ); - if (abortRequested) { - await emit("cancelled"); - return; - } - if (sideAgent.state.errorMessage) { - await emit("error", sideAgent.state.errorMessage); - return; - } - // A run that ends on a tool turn was answered in an earlier turn; a textless - // final turn without tool calls is a genuinely empty answer. - answer = finalTurn.content.some((block) => block.type === "toolCall") - ? (assistantTurns().map(readAssistantText).filter(Boolean).at(-1) ?? "") - : readAssistantText(finalTurn); - await emit("complete"); - }) - .catch(async (error) => { - const errorMessage = error instanceof Error ? error.message : String(error); - await Promise.resolve( - emit(abortRequested ? "cancelled" : "error", abortRequested ? undefined : errorMessage), - ).catch(() => undefined); - }) - .finally(unsubscribe); - - return { - done, - abort() { - abortRequested = true; - retryAbortController.abort(); - if (started) { - sideAgent.abort(); - } - }, - }; -} diff --git a/apps/tui/lib/coding-agent/src/modes/daemon/daemon-client-env.ts b/apps/tui/lib/coding-agent/src/modes/daemon/daemon-client-env.ts deleted file mode 100644 index 857a4d12..00000000 --- a/apps/tui/lib/coding-agent/src/modes/daemon/daemon-client-env.ts +++ /dev/null @@ -1,97 +0,0 @@ -import { DAEMON_CLIENT_ENV_KEYS } from "./daemon-protocol.js"; - -/** Re-filter client-sent env to the allowlist; the socket peer is untrusted. */ -export function filterClientEnv(env?: Record): Record | undefined { - if (!env) { - return undefined; - } - const filtered: Record = {}; - for (const key of DAEMON_CLIENT_ENV_KEYS) { - if (env[key] !== undefined) { - filtered[key] = env[key]; - } - } - return Object.keys(filtered).length > 0 ? filtered : undefined; -} - -// The daemon's own allowlisted env, captured at startup before any env window -// can mutate process.env. -const baseClientEnv: Record = {}; -for (const key of DAEMON_CLIENT_ENV_KEYS) { - baseClientEnv[key] = process.env[key]; -} - -/** - * Exec env for a session's subprocesses: pins every allowlisted key to the - * session's value (unset when the client didn't send it), or to the daemon's - * startup value for env-less sessions. Pinning makes subprocess env - * independent of any env window another session has open at spawn time. - */ -export function execEnvForSession(clientEnv?: Record): Record { - const source = clientEnv ?? baseClientEnv; - const env: Record = {}; - for (const key of DAEMON_CLIENT_ENV_KEYS) { - env[key] = source[key]; - } - return env; -} - -// Shared/exclusive lock: env windows are exclusive (they mutate process.env), -// env-less loads are shared — they run concurrently with each other but never -// inside an env window, so they can't capture another session's identity. -let lastExclusive: Promise = Promise.resolve(); -const activeShared = new Set>(); - -/** - * Run fn with the client's env applied to process.env, restoring afterwards. - * Extensions capture vars like HERDR_PANE_ID synchronously at module load, so - * they must be in process.env while the session loads its extensions; after - * this window the session's exec env covers subprocess reads. - */ -export async function withClientEnv(env: Record | undefined, fn: () => Promise): Promise { - if (!env) { - const gate = lastExclusive; - const run = (async () => { - await gate.catch(() => undefined); - return fn(); - })(); - const tracked = run.catch(() => undefined); - activeShared.add(tracked); - void tracked.then(() => activeShared.delete(tracked)); - return run; - } - const prior = lastExclusive; - // Snapshot synchronously: shareds arriving later gate on this window via - // lastExclusive, so waiting for them here would deadlock. - const sharedAtRequest = [...activeShared]; - const run = (async () => { - await prior.catch(() => undefined); - await Promise.all(sharedAtRequest); - const previous = new Map(); - // Pin the full allowlist (unsetting keys the client didn't send) so a - // partially-forwarded env can't mix with the daemon's ambient values — - // mirroring execEnvForSession. - for (const key of DAEMON_CLIENT_ENV_KEYS) { - previous.set(key, process.env[key]); - const value = env[key]; - if (value === undefined) { - delete process.env[key]; - } else { - process.env[key] = value; - } - } - try { - return await fn(); - } finally { - for (const [key, value] of previous) { - if (value === undefined) { - delete process.env[key]; - } else { - process.env[key] = value; - } - } - } - })(); - lastExclusive = run.catch(() => undefined); - return run; -} diff --git a/apps/tui/lib/coding-agent/src/modes/daemon/daemon-socket.ts b/apps/tui/lib/coding-agent/src/modes/daemon/daemon-socket.ts deleted file mode 100644 index 99a3a151..00000000 --- a/apps/tui/lib/coding-agent/src/modes/daemon/daemon-socket.ts +++ /dev/null @@ -1,332 +0,0 @@ -import { chmodSync, existsSync, lstatSync, mkdirSync, unlinkSync } from "node:fs"; -import { createConnection } from "node:net"; -import { tmpdir } from "node:os"; -import { dirname, join } from "node:path"; -import lockfile from "proper-lockfile"; - -export { normalizeSocketPath } from "../../utils/daemon-socket-path.js"; - -const DAEMON_SOCKET_MODE = 0o600; -const DAEMON_SOCKET_DIR_MODE = 0o700; -const DAEMON_SOCKET_RELEASE_GRACE_MS = 1000; -const DAEMON_SOCKET_RELEASE_POLL_MS = 25; -const DAEMON_SOCKET_LOCK_STALE_MS = 5000; -const DAEMON_SOCKET_LOCK_UPDATE_MS = 1000; - -type DaemonSocketCompromiseListener = (error: Error) => void; - -export class DaemonSocketPathLease { - private released = false; - private compromisedError?: Error; - private readonly compromiseListeners = new Set(); - - constructor( - readonly socketPath: string, - private readonly releaseLock: () => Promise, - ) {} - - get compromise(): Error | undefined { - return this.compromisedError; - } - - onCompromised(listener: DaemonSocketCompromiseListener): () => void { - if (this.compromisedError) { - this.notifyCompromiseListener(listener, this.compromisedError); - return () => {}; - } - this.compromiseListeners.add(listener); - return () => this.compromiseListeners.delete(listener); - } - - recordCompromise(error: Error): void { - if (this.compromisedError) return; - this.compromisedError = error; - const listeners = [...this.compromiseListeners]; - this.compromiseListeners.clear(); - for (const listener of listeners) this.notifyCompromiseListener(listener, error); - } - - async release(): Promise { - if (this.released) return; - this.released = true; - this.compromiseListeners.clear(); - await this.releaseLock(); - } - - private notifyCompromiseListener(listener: DaemonSocketCompromiseListener, error: Error): void { - try { - listener(error); - } catch { - // A lease callback must not rethrow from proper-lockfile's refresh callback. - } - } -} - -export interface DaemonSocketIdentity { - dev: number; - ino: number; -} - -export function defaultDaemonSocketPath(): string { - if (process.platform === "win32") { - return "\\\\.\\pipe\\prime-agent-daemon"; - } - return join(defaultDaemonSocketDir(), "daemon.sock"); -} - -export async function acquireDaemonSocketPathLease(socketPath: string): Promise { - ensureDefaultDaemonSocketDir(socketPath); - if (process.platform === "win32") { - return undefined; - } - let lease: DaemonSocketPathLease | undefined; - let pendingCompromise: Error | undefined; - const releaseLock = await lockfile.lock(socketPath, { - realpath: false, - stale: DAEMON_SOCKET_LOCK_STALE_MS, - update: DAEMON_SOCKET_LOCK_UPDATE_MS, - onCompromised: (error) => { - if (lease) lease.recordCompromise(error); - else pendingCompromise = error; - }, - retries: { - retries: 600, - factor: 1, - minTimeout: DAEMON_SOCKET_RELEASE_POLL_MS, - maxTimeout: DAEMON_SOCKET_RELEASE_POLL_MS, - }, - }); - lease = new DaemonSocketPathLease(socketPath, releaseLock); - if (pendingCompromise) lease.recordCompromise(pendingCompromise); - return lease; -} - -export async function prepareDaemonSocketPath(socketPath: string, lease?: DaemonSocketPathLease): Promise { - ensureDefaultDaemonSocketDir(socketPath); - - if (process.platform === "win32") { - return; - } - if (lease) { - assertSocketLease(socketPath, lease); - assertSocketLeaseHeld(socketPath, lease); - await prepareUnixDaemonSocketPath(socketPath, lease); - return; - } - if (!existsSync(socketPath)) { - return; - } - if (await canConnectToUnixSocket(socketPath)) { - throw new Error(`Daemon socket already in use: ${socketPath}`); - } - const ownedLease = await acquireDaemonSocketPathLease(socketPath); - try { - await prepareUnixDaemonSocketPath(socketPath, ownedLease); - } finally { - await ownedLease?.release(); - } -} - -async function prepareUnixDaemonSocketPath(socketPath: string, lease?: DaemonSocketPathLease): Promise { - if (!existsSync(socketPath)) { - return; - } - - let stat: ReturnType; - try { - stat = lstatSync(socketPath); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") { - return; - } - throw error; - } - if (!stat.isSocket()) { - throw new Error(`Daemon socket path exists and is not a socket: ${socketPath}`); - } - - const staleIdentity: DaemonSocketIdentity = { dev: stat.dev, ino: stat.ino }; - if (await canConnectToUnixSocket(socketPath)) { - throw new Error(`Daemon socket already in use: ${socketPath}`); - } - const deadline = Date.now() + DAEMON_SOCKET_RELEASE_GRACE_MS; - while (Date.now() < deadline) { - await delay(DAEMON_SOCKET_RELEASE_POLL_MS); - if (!existsSync(socketPath)) { - return; - } - let currentIdentity: DaemonSocketIdentity | undefined; - try { - currentIdentity = getDaemonSocketIdentity(socketPath); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ENOENT") { - return; - } - throw error; - } - if (!currentIdentity || currentIdentity.dev !== staleIdentity.dev || currentIdentity.ino !== staleIdentity.ino) { - throw new Error(`Daemon socket changed ownership while waiting for cleanup: ${socketPath}`); - } - if (await canConnectToUnixSocket(socketPath)) { - throw new Error(`Daemon socket already in use: ${socketPath}`); - } - } - - if (lease) assertSocketLeaseHeld(socketPath, lease); - unlinkSync(socketPath); -} - -export function restrictDaemonSocketPath(socketPath: string): void { - if (process.platform === "win32") { - return; - } - chmodSync(socketPath, DAEMON_SOCKET_MODE); -} - -export function getDaemonSocketIdentity(socketPath: string): DaemonSocketIdentity | undefined { - if (process.platform === "win32") { - return undefined; - } - const stat = lstatSync(socketPath); - return { dev: stat.dev, ino: stat.ino }; -} - -export function cleanupDaemonSocketPath( - socketPath: string, - expectedIdentity?: DaemonSocketIdentity, - lease?: DaemonSocketPathLease, -): void { - if (process.platform === "win32") { - return; - } - if (lease) { - assertSocketLease(socketPath, lease); - if (lease.compromise) { - return; - } - try { - cleanupUnixDaemonSocketPath(socketPath, expectedIdentity); - } catch { - // Best effort cleanup; shutdown should not be blocked by socket unlink failures. - } - return; - } - let releaseLock: (() => void) | undefined; - let lockCompromised = false; - try { - releaseLock = lockfile.lockSync(socketPath, { - realpath: false, - stale: DAEMON_SOCKET_LOCK_STALE_MS, - update: DAEMON_SOCKET_LOCK_UPDATE_MS, - onCompromised: () => { - lockCompromised = true; - }, - retries: 0, - }); - } catch { - return; - } - if (lockCompromised) { - try { - releaseLock(); - } catch { - // Best effort release. - } - return; - } - try { - cleanupUnixDaemonSocketPath(socketPath, expectedIdentity); - } catch { - // Best effort cleanup; shutdown should not be blocked by socket unlink failures. - } finally { - try { - releaseLock(); - } catch { - // Best effort cleanup; a failed release is recoverable as a stale lock. - } - } -} - -function cleanupUnixDaemonSocketPath(socketPath: string, expectedIdentity?: DaemonSocketIdentity): void { - if (!existsSync(socketPath)) { - return; - } - if (expectedIdentity) { - const currentIdentity = getDaemonSocketIdentity(socketPath); - if ( - !currentIdentity || - currentIdentity.dev !== expectedIdentity.dev || - currentIdentity.ino !== expectedIdentity.ino - ) { - return; - } - } - unlinkSync(socketPath); -} - -function assertSocketLease(socketPath: string, lease: DaemonSocketPathLease): void { - if (lease.socketPath !== socketPath) { - throw new Error(`Daemon socket lease does not match ${socketPath}`); - } -} - -function assertSocketLeaseHeld(socketPath: string, lease: DaemonSocketPathLease): void { - if (lease.compromise) { - throw new Error(`Daemon socket lease for ${socketPath} was compromised: ${lease.compromise.message}`); - } -} - -export function defaultDaemonSocketDir(): string { - const suffix = typeof process.getuid === "function" ? String(process.getuid()) : "user"; - return join(tmpdir(), `prime-agent-${suffix}`); -} - -function ensureDefaultDaemonSocketDir(socketPath: string): void { - if (process.platform === "win32" || dirname(socketPath) !== defaultDaemonSocketDir()) { - return; - } - - if (!existsSync(defaultDaemonSocketDir())) { - mkdirSync(defaultDaemonSocketDir(), { recursive: true, mode: DAEMON_SOCKET_DIR_MODE }); - } - - const stat = lstatSync(defaultDaemonSocketDir()); - if (!stat.isDirectory()) { - throw new Error(`Daemon socket directory exists and is not a directory: ${defaultDaemonSocketDir()}`); - } - - if (typeof process.getuid === "function" && stat.uid !== process.getuid()) { - throw new Error(`Daemon socket directory is not owned by the current user: ${defaultDaemonSocketDir()}`); - } - - chmodSync(defaultDaemonSocketDir(), DAEMON_SOCKET_DIR_MODE); -} - -function canConnectToUnixSocket(socketPath: string): Promise { - return new Promise((resolveConnect) => { - const socket = createConnection(socketPath); - let settled = false; - let timeoutId: ReturnType | undefined; - - const finish = (canConnect: boolean) => { - if (settled) { - return; - } - settled = true; - if (timeoutId) { - clearTimeout(timeoutId); - } - socket.removeAllListeners(); - socket.destroy(); - resolveConnect(canConnect); - }; - - timeoutId = setTimeout(() => finish(false), 250); - socket.once("connect", () => finish(true)); - socket.once("error", () => finish(false)); - }); -} - -function delay(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} diff --git a/apps/tui/lib/coding-agent/src/modes/daemon/daemon-worker-protocol.ts b/apps/tui/lib/coding-agent/src/modes/daemon/daemon-worker-protocol.ts deleted file mode 100644 index 2a20f1e2..00000000 --- a/apps/tui/lib/coding-agent/src/modes/daemon/daemon-worker-protocol.ts +++ /dev/null @@ -1,280 +0,0 @@ -import { closeSync, readFileSync } from "node:fs"; -import type { AgentSessionMessageDeliveryMode, AgentSessionMessageSender } from "../../core/agent-messages.js"; -import type { IdleEvictionMinutes } from "../../core/session-action-store.js"; - -export { SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV } from "../../core/session-lease.js"; - -import type { WorkerRosterEntry } from "./agent-roster.js"; -import type { DaemonClientCapability, DaemonCommand, DaemonOutbound } from "./daemon-protocol.js"; - -export const DAEMON_WORKER_ROLE_ENV = "DEVO_INTERNAL_DAEMON_WORKER"; -export const DAEMON_WORKER_TOKEN_ENV = "DEVO_INTERNAL_DAEMON_WORKER_TOKEN"; -export const DAEMON_WORKER_INSTANCE_ID_ENV = "DEVO_INTERNAL_DAEMON_WORKER_INSTANCE_ID"; -export const DAEMON_WORKER_ACTIVE_SESSION_ID_ENV = "DEVO_INTERNAL_DAEMON_WORKER_ACTIVE_SESSION_ID"; -export const DAEMON_WORKER_SUPERVISOR_SOCKET_ENV = "DEVO_INTERNAL_DAEMON_SUPERVISOR_SOCKET"; -export const DAEMON_WORKER_RECOVERY_JOURNAL_ENV = "DEVO_INTERNAL_DAEMON_WORKER_RECOVERY_JOURNAL"; -export const DAEMON_WORKER_STARTUP_GATE_FD_ENV = "DEVO_INTERNAL_DAEMON_WORKER_STARTUP_GATE_FD"; -export const DAEMON_WORKER_STARTUP_GATE_COMMIT = "start\n"; -export type DaemonWorkerLifecycle = "starting" | "ready" | "recovering" | "stopping" | "failed"; - -// Worker->supervisor roster frames live outside the client-facing DaemonOutbound schema. -export type DaemonWorkerRosterOutbound = - | { - type: "roster_delta"; - entries: WorkerRosterEntry[]; - removedAgentIds?: string[]; - snapshot?: true; - } - | { type: "roster_heartbeat" }; - -/** Advertised by new workers in the worker_auth response; absent on legacy workers. */ -export const DAEMON_WORKER_ROSTER_CAPABILITY = "agent_roster"; - -/** Advertised in the worker_auth response by workers that accept peer transport grants. */ -export const DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY = "peer_transport"; - -/** Idle keepalive cadence for worker->supervisor roster frames; the supervisor staleness threshold derives from it. */ -export const ROSTER_HEARTBEAT_INTERVAL_MS = 15_000; - -export type DaemonWorkerFrameHeader = - | { - kind: "command"; - requestId: string; - commandType: string; - } - | { - kind: "outbound"; - requestId?: string; - outboundType: DaemonOutbound["type"] | DaemonWorkerRosterOutbound["type"]; - activeSessionId?: string; - snapshotId?: string; - sessionEventType?: string; - payloadEncoding?: "jsonl" | "assistant-delta"; - snapshotPurpose?: "attach" | "replacement" | "catchup"; - }; - -export type DaemonCreateCommand = Extract; - -export interface DurableDaemonCreateCommand { - type: "create"; - sessionPath?: string; - noSession?: boolean; -} - -export function durableDaemonCreateCommand(command: DaemonCreateCommand): DurableDaemonCreateCommand { - return { - type: "create", - ...(command.sessionPath !== undefined ? { sessionPath: command.sessionPath } : {}), - ...(command.noSession !== undefined ? { noSession: command.noSession } : {}), - }; -} - -/** - * A single-use, worker-memory-only admission for one direct peer, scoped to one active-session - * slot (stable across switch_session/new_session/fork) of one worker incarnation. The session pin - * is a routing/accident guard, not a privilege boundary: ticket holders already hold supervisor - * access, which can attach to, switch, or kill any session. - */ -export interface DaemonWorkerPeerGrant { - grantId: string; - token: string; - expiresAt: string; - purpose: "session_client"; - workerInstanceId: string; - activeSessionId: string; - issuerGeneration: string; -} - -/** Commands a direct peer may send before it holds an authenticated session role. */ -export type DaemonPeerCommand = { - id?: string; - type: "peer_auth"; - grantId: string; - token: string; - workerInstanceId: string; - purpose: "session_client"; -}; - -export type DaemonPeerCommandBody = Omit; - -export type DaemonWorkerCommand = - | { - id?: string; - type: "worker_auth"; - token: string; - workerInstanceId?: string; - supervisorGeneration: string; - supervisorPid: number; - supervisorProcessStartId?: string; - supervisorSocketPath: string; - } - | { - id?: string; - type: "worker_subscribe"; - activeSessionId: string; - capabilities?: readonly DaemonClientCapability[]; - supportsExtensionUi?: boolean; - } - | { id?: string; type: "worker_unsubscribe"; activeSessionId: string } - | { id?: string; type: "worker_register_peer_transport"; grant: DaemonWorkerPeerGrant } - | { id?: string; type: "worker_archive_and_shutdown" } - | { - id?: string; - type: "worker_passivate_idle_children"; - idleEvictionMinutes: IdleEvictionMinutes; - now: number; - limit: number; - } - | { - id?: string; - type: "worker_deliver_message"; - targetActiveSessionId: string; - message: string; - sender: AgentSessionMessageSender; - deliveryMode?: AgentSessionMessageDeliveryMode; - } - | { id?: string; type: "worker_prepare_update" } - | { id?: string; type: "worker_commit_update" } - | { id?: string; type: "worker_cancel_update" }; - -export type DaemonWorkerCommandBody = DaemonWorkerCommand extends infer TCommand - ? TCommand extends { id?: string } - ? Omit - : never - : never; - -export interface DaemonWorkerDescriptor { - version: 1 | 2; - workerId: string; - pid: number; - processStartId?: string; - socketPath: string; - recoveryJournalPath: string; - orphanProcessJournalPath?: string; - supervisorSocketPath: string; - authenticationToken: string; - workerInstanceId?: string; - rootActiveSessionId: string; - /** Stable protocol client that owns this worker. Omitted for resident sessions. */ - ownerClientId?: string; - rootSessionId?: string; - sessionFile?: string; - sessionDir?: string; - telemetryDisabled?: true; - createdAt: string; - updatedAt: string; - lifecycle: DaemonWorkerLifecycle; - createCommand: DurableDaemonCreateCommand; - consecutiveFailures: number; - /** Durable intent written before root termination so replacement supervisors never recover it. */ - stopRequestedAt?: string; - /** Complete the root's archived lifecycle state after its process has stopped. */ - archiveOnStop?: boolean; - lastFailureAt?: string; - lastError?: string; -} - -export function durableDaemonWorkerDescriptor(descriptor: DaemonWorkerDescriptor): DaemonWorkerDescriptor { - const versionOneCreateCommand = descriptor.createCommand as unknown as { config?: unknown }; - const versionOneConfig = - descriptor.version === 1 && - typeof versionOneCreateCommand.config === "object" && - versionOneCreateCommand.config !== null - ? (versionOneCreateCommand.config as Record) - : undefined; - const sessionDir = - descriptor.sessionDir ?? - (typeof versionOneConfig?.sessionDir === "string" ? versionOneConfig.sessionDir : undefined); - const telemetryDisabled = descriptor.telemetryDisabled === true || versionOneConfig?.telemetryDisabled === true; - return { - version: 2, - workerId: descriptor.workerId, - pid: descriptor.pid, - ...(descriptor.processStartId !== undefined ? { processStartId: descriptor.processStartId } : {}), - socketPath: descriptor.socketPath, - recoveryJournalPath: descriptor.recoveryJournalPath, - ...(descriptor.orphanProcessJournalPath !== undefined - ? { orphanProcessJournalPath: descriptor.orphanProcessJournalPath } - : {}), - supervisorSocketPath: descriptor.supervisorSocketPath, - authenticationToken: descriptor.authenticationToken, - ...(descriptor.workerInstanceId !== undefined ? { workerInstanceId: descriptor.workerInstanceId } : {}), - rootActiveSessionId: descriptor.rootActiveSessionId, - ...(descriptor.ownerClientId !== undefined ? { ownerClientId: descriptor.ownerClientId } : {}), - ...(descriptor.rootSessionId !== undefined ? { rootSessionId: descriptor.rootSessionId } : {}), - ...(descriptor.sessionFile !== undefined ? { sessionFile: descriptor.sessionFile } : {}), - ...(sessionDir !== undefined ? { sessionDir } : {}), - ...(telemetryDisabled ? { telemetryDisabled: true as const } : {}), - createdAt: descriptor.createdAt, - updatedAt: descriptor.updatedAt, - lifecycle: descriptor.lifecycle, - createCommand: durableDaemonCreateCommand(descriptor.createCommand), - consecutiveFailures: descriptor.consecutiveFailures, - ...(descriptor.stopRequestedAt !== undefined ? { stopRequestedAt: descriptor.stopRequestedAt } : {}), - ...(descriptor.archiveOnStop !== undefined ? { archiveOnStop: descriptor.archiveOnStop } : {}), - ...(descriptor.lastFailureAt !== undefined ? { lastFailureAt: descriptor.lastFailureAt } : {}), - ...(descriptor.lifecycle === "failed" ? { lastError: "Waiting for a client with fresh runtime context" } : {}), - }; -} - -export function isDaemonWorkerProcess(environment: NodeJS.ProcessEnv = process.env): boolean { - return environment[DAEMON_WORKER_ROLE_ENV] === "1"; -} - -export function waitForDaemonWorkerStartupGate(environment: NodeJS.ProcessEnv = process.env): void { - const rawFd = environment[DAEMON_WORKER_STARTUP_GATE_FD_ENV]; - if (rawFd === undefined) { - return; - } - delete environment[DAEMON_WORKER_STARTUP_GATE_FD_ENV]; - const fd = Number(rawFd); - if (!Number.isInteger(fd) || fd < 3) { - throw new Error("Daemon session worker has an invalid startup gate"); - } - let marker: string; - try { - marker = readFileSync(fd, "utf8"); - } finally { - closeSync(fd); - } - if (marker !== DAEMON_WORKER_STARTUP_GATE_COMMIT) { - throw new Error("Daemon session worker startup was cancelled"); - } -} - -export function daemonWorkerInstanceId(environment: NodeJS.ProcessEnv = process.env): string | undefined { - return environment[DAEMON_WORKER_INSTANCE_ID_ENV] || undefined; -} - -export function requireDaemonWorkerAuthenticationToken(environment: NodeJS.ProcessEnv = process.env): string { - const token = environment[DAEMON_WORKER_TOKEN_ENV]; - if (!token) { - throw new Error("Daemon session worker is missing its authentication token"); - } - return token; -} - -export function isDaemonWorkerFrameHeader(value: unknown): value is DaemonWorkerFrameHeader { - if (!value || typeof value !== "object") { - return false; - } - const candidate = value as Record; - if (candidate.kind === "command") { - return typeof candidate.requestId === "string" && typeof candidate.commandType === "string"; - } - return ( - candidate.kind === "outbound" && - typeof candidate.outboundType === "string" && - (candidate.requestId === undefined || typeof candidate.requestId === "string") && - (candidate.activeSessionId === undefined || typeof candidate.activeSessionId === "string") && - (candidate.snapshotId === undefined || typeof candidate.snapshotId === "string") && - (candidate.sessionEventType === undefined || typeof candidate.sessionEventType === "string") && - (candidate.snapshotPurpose === undefined || - candidate.snapshotPurpose === "attach" || - candidate.snapshotPurpose === "replacement" || - candidate.snapshotPurpose === "catchup") && - (candidate.payloadEncoding === undefined || - candidate.payloadEncoding === "jsonl" || - candidate.payloadEncoding === "assistant-delta") - ); -} diff --git a/apps/tui/src/host-env.test.ts b/apps/tui/src/host-env.test.ts new file mode 100644 index 00000000..fc327770 --- /dev/null +++ b/apps/tui/src/host-env.test.ts @@ -0,0 +1,53 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +for (const explicitOverrides of [false, true]) { + test(`agent directory setup is idempotent with ${explicitOverrides ? "explicit" : "default"} environment`, () => { + const directory = mkdtempSync(join(tmpdir(), "devo-host-env-")); + try { + const home = join(directory, "home"); + const env = { ...process.env }; + delete env.DEVO_CODING_AGENT_DIR; + delete env.DEVO_HOME; + delete env.PI_SKIP_VERSION_CHECK; + const expected = explicitOverrides + ? { + agentDir: join(directory, "agent-override"), + home: join(directory, "home-override"), + skipVersionCheck: "0", + } + : { agentDir: home, home, skipVersionCheck: "1" }; + if (explicitOverrides) { + env.DEVO_CODING_AGENT_DIR = expected.agentDir; + env.DEVO_HOME = expected.home; + env.PI_SKIP_VERSION_CHECK = expected.skipVersionCheck; + } + const result = spawnSync(process.execPath, [ + "--import", "tsx", "--input-type=module", "--eval", + ` + import { existsSync } from "node:fs"; + import { applyDevoAgentDirEnv } from ${JSON.stringify(new URL("./host.ts", import.meta.url).href)}; + const home = ${JSON.stringify(home)}; + const first = applyDevoAgentDirEnv(home); + const second = applyDevoAgentDirEnv(home); + console.log(JSON.stringify({ + first, second, exists: existsSync(home), + agentDir: process.env.DEVO_CODING_AGENT_DIR, + home: process.env.DEVO_HOME, + skipVersionCheck: process.env.PI_SKIP_VERSION_CHECK, + })); + `, + ], { env, encoding: "utf8" }); + assert.equal(result.status, 0, result.stderr); + assert.deepEqual(JSON.parse(result.stdout.trim()), { + first: home, second: home, exists: true, ...expected, + }); + } finally { + rmSync(directory, { recursive: true, force: true }); + } + }); +} diff --git a/apps/tui/src/host.ts b/apps/tui/src/host.ts index 6254e325..11802442 100644 --- a/apps/tui/src/host.ts +++ b/apps/tui/src/host.ts @@ -40,9 +40,6 @@ export function applyDevoAgentDirEnv(home = resolveDevoHome()): string { if (!process.env.DEVO_CODING_AGENT_DIR) { process.env.DEVO_CODING_AGENT_DIR = home; } - if (!process.env.DEVO_CODING_AGENT_DIR) { - process.env.DEVO_CODING_AGENT_DIR = home; - } if (!process.env.DEVO_HOME) { process.env.DEVO_HOME = home; } diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 85444786..af062875 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -36,7 +36,6 @@ mod prompt_command; mod prompt_host; mod upgrade_command; -use agent_command::launch_interactive_mode_client; use agent_command::run_agent; use app_exit::AppExit; use doctor_command::run_doctor; @@ -423,17 +422,6 @@ fn cli_logging_overrides(cli: &Cli) -> toml::Value { )])) } -/// TODO: Why there is a simple wrapper function, maybe should be removed. -/// Launch helper re-exported for tests / clarity — implementation lives in -/// [`agent_command::launch_interactive_mode_client`]. -#[allow(dead_code)] -fn launch_interactive_mode_client_for_tests() -> Result<()> { - launch_interactive_mode_client( - /*resume_session_id*/ None, /*dangerously_skip_permissions*/ false, - /*log_level*/ None, - ) -} - /// TODO: Move test to another single file. #[cfg(test)] mod tests { diff --git a/crates/core/src/context/mod.rs b/crates/core/src/context/mod.rs index 17458e49..e5b6d0a2 100644 --- a/crates/core/src/context/mod.rs +++ b/crates/core/src/context/mod.rs @@ -1,4 +1,3 @@ -use std::marker::PhantomData; use std::path::PathBuf; use async_trait::async_trait; @@ -9,38 +8,9 @@ use devo_protocol::approx_tokens_from_byte_count; use devo_protocol::{ContentBlock, Message, Model, Role}; // --------------------------------------------------------------------------- -// Contextual user fragment traits and registration +// Contextual user fragment traits // --------------------------------------------------------------------------- -/// Type-erased registration for a contextual user fragment. -/// -/// Implementations are used by context filtering code to recognize injected -/// fragments without constructing the concrete context payload. -#[allow(dead_code)] -pub(crate) trait FragmentRegistration: Sync { - fn matches_text(&self, text: &str) -> bool; -} - -#[allow(dead_code)] -pub(crate) struct FragmentRegistrationProxy { - _marker: PhantomData T>, -} - -#[allow(dead_code)] -impl FragmentRegistrationProxy { - pub(crate) const fn new() -> Self { - Self { - _marker: PhantomData, - } - } -} - -impl FragmentRegistration for FragmentRegistrationProxy { - fn matches_text(&self, text: &str) -> bool { - T::matches_text(text) - } -} - /// Context payload that is injected as a message fragment. /// /// Implementations own the response role and provide the exact fragment body. diff --git a/crates/server/src/runtime/items.rs b/crates/server/src/runtime/items.rs index e8bcfc1c..7c78abbb 100644 --- a/crates/server/src/runtime/items.rs +++ b/crates/server/src/runtime/items.rs @@ -16,6 +16,9 @@ use devo_protocol::native::wire_projector::{item_started_state, typed_item_envel use super::*; +#[cfg(test)] +mod tests; + /// Build a Native user-message item for first-party emit (no legacy payload bag). /// /// Images are listed before text so the TUI projects @@ -69,10 +72,6 @@ fn next_fallback_item_seq() -> u64 { NEXT.fetch_add(1, Ordering::Relaxed) } -fn legacy_session_for_lookup(session_id: &NativeSessionId) -> Option { - Some(*session_id) -} - impl ServerRuntime { /// Persist session summary to SQLite if the session is durable. /// The rollout file is the authoritative store, so failures here are @@ -121,10 +120,7 @@ impl ServerRuntime { } async fn remember_item_started_at(&self, session_id: &NativeSessionId, item_id: NativeItemId) { - let Some(legacy) = legacy_session_for_lookup(session_id) else { - return; - }; - let Some(stream) = self.active_stream_state(legacy).await else { + let Some(stream) = self.active_stream_state(*session_id).await else { return; }; let mut stream = stream.lock().await; @@ -138,8 +134,7 @@ impl ServerRuntime { session_id: &NativeSessionId, item_id: &NativeItemId, ) -> Option> { - let legacy = legacy_session_for_lookup(session_id)?; - let stream = self.active_stream_state(legacy).await?; + let stream = self.active_stream_state(*session_id).await?; let mut stream = stream.lock().await; stream .turn_inline @@ -246,16 +241,8 @@ impl ServerRuntime { use crate::persisted_native_item::PersistedNativeItem; use crate::persisted_native_item::history_entry_from_native_item; - let Some(legacy_session_id) = legacy_session_for_lookup(&session_id) else { - tracing::warn!( - session_id = %session_id, - "persist_item skipped: Native session id is not bridgeable to legacy lookup key" - ); - return; - }; - let history_entry = history_entry_from_native_item(&native_item); - if let Some(stream) = self.active_stream_state(legacy_session_id).await { + if let Some(stream) = self.active_stream_state(session_id).await { // Mutate inline state under the lock, then release before any // blocking rollout I/O so the event stream cannot pin the async // mutex across synchronous disk writes. @@ -306,13 +293,13 @@ impl ServerRuntime { item_id, parent_id, ) { - tracing::warn!(session_id = %legacy_session_id, error = %error, "failed to persist tree edge"); + tracing::warn!(session_id = %session_id, error = %error, "failed to persist tree edge"); } if let Err(error) = self .rollout_store .append_canonical_item_at(&rollout_path, envelope) { - tracing::warn!(session_id = %legacy_session_id, error = %error, "failed to persist item line"); + tracing::warn!(session_id = %session_id, error = %error, "failed to persist item line"); } if let Err(error) = self.rollout_store.append_session_leaf_at( &rollout_path, @@ -320,7 +307,7 @@ impl ServerRuntime { Some(item_id), leaf_epoch, ) { - tracing::warn!(session_id = %legacy_session_id, error = %error, "failed to persist session leaf"); + tracing::warn!(session_id = %session_id, error = %error, "failed to persist session leaf"); } } return; @@ -329,20 +316,19 @@ impl ServerRuntime { // actor is not polling its mailbox until the stream finishes, so we // must not fall through to blocking actor commands. tracing::warn!( - session_id = %legacy_session_id, + session_id = %session_id, turn_id = %turn_id, "persist_item skipped: active turn stream has no inline state" ); return; } - let Some(session_handle) = self.session(legacy_session_id).await else { + let Some(session_handle) = self.session(session_id).await else { return; }; if let Some(history_entry) = history_entry { session_handle.append_history_item(history_entry).await; } - let legacy_turn_id = turn_id; - let Some(prep) = session_handle.prepare_persist_item(legacy_turn_id).await else { + let Some(prep) = session_handle.prepare_persist_item(turn_id).await else { return; }; session_handle @@ -369,25 +355,25 @@ impl ServerRuntime { envelope.parent_id = parent_id; if let Err(error) = self.rollout_store.append_tree_edge_at( &rollout_path, - legacy_session_id, + session_id, item_id, parent_id, ) { - tracing::warn!(session_id = %legacy_session_id, error = %error, "failed to persist tree edge"); + tracing::warn!(session_id = %session_id, error = %error, "failed to persist tree edge"); } if let Err(error) = self .rollout_store .append_canonical_item_at(&rollout_path, envelope) { - tracing::warn!(session_id = %legacy_session_id, error = %error, "failed to persist item line"); + tracing::warn!(session_id = %session_id, error = %error, "failed to persist item line"); } if let Err(error) = self.rollout_store.append_session_leaf_at( &rollout_path, - legacy_session_id, + session_id, Some(item_id), leaf_epoch, ) { - tracing::warn!(session_id = %legacy_session_id, error = %error, "failed to persist session leaf"); + tracing::warn!(session_id = %session_id, error = %error, "failed to persist session leaf"); } session_handle .set_transcript_leaf( @@ -401,10 +387,7 @@ impl ServerRuntime { } pub(super) async fn allocate_item_sequence(&self, session_id: &NativeSessionId) -> u64 { - let Some(legacy_session_id) = legacy_session_for_lookup(session_id) else { - return next_fallback_item_seq(); - }; - if let Some(stream) = self.active_stream_state(legacy_session_id).await { + if let Some(stream) = self.active_stream_state(*session_id).await { let mut stream = stream.lock().await; if let Some(inline) = stream.turn_inline.as_mut() { return inline.allocate_item_seq(); @@ -413,7 +396,7 @@ impl ServerRuntime { // mailbox while its turn event stream is registered. return next_fallback_item_seq(); } - if let Some(handle) = self.session(legacy_session_id).await + if let Some(handle) = self.session(*session_id).await && let Some(item_seq) = handle.allocate_item_seq().await { return item_seq; diff --git a/crates/server/src/runtime/items/tests.rs b/crates/server/src/runtime/items/tests.rs new file mode 100644 index 00000000..743071f7 --- /dev/null +++ b/crates/server/src/runtime/items/tests.rs @@ -0,0 +1,43 @@ +use std::sync::Arc; + +use pretty_assertions::assert_eq; +use tokio::sync::Mutex; + +use super::*; +use crate::runtime::session_actor::state::SessionStreamState; +use crate::test_support::TestRuntime; + +#[tokio::test] +async fn missing_session_keeps_default_item_sequence() { + let directory = tempfile::tempdir().expect("temporary directory"); + let runtime = TestRuntime::noop().runtime(directory.path()); + + assert_eq!( + runtime + .allocate_item_sequence(&NativeSessionId::new()) + .await, + 1 + ); +} + +#[tokio::test] +async fn active_stream_without_inline_state_keeps_fallback_item_sequence() { + let directory = tempfile::tempdir().expect("temporary directory"); + let runtime = TestRuntime::noop().runtime(directory.path()); + let session_id = NativeSessionId::new(); + runtime + .active_turns + .register_stream( + session_id, + Arc::new(Mutex::new(SessionStreamState::default())), + ) + .await; + + let first = runtime.allocate_item_sequence(&session_id).await; + let second = runtime.allocate_item_sequence(&session_id).await; + assert!(first >= 1 << 32); + assert!(second > first); + + runtime.active_turns.unregister_stream(session_id).await; + assert_eq!(runtime.allocate_item_sequence(&session_id).await, 1); +} diff --git a/crates/server/src/runtime/session_actor/registry.rs b/crates/server/src/runtime/session_actor/registry.rs index e8702cf6..94d7a1dd 100644 --- a/crates/server/src/runtime/session_actor/registry.rs +++ b/crates/server/src/runtime/session_actor/registry.rs @@ -62,26 +62,6 @@ impl ServerRuntime { self.sessions.lock().await.values().cloned().collect() } - #[allow(dead_code)] - pub(crate) async fn list_session_summaries_from_actors( - &self, - ) -> Vec { - let handles = self.list_session_handles().await; - let mut summaries = Vec::with_capacity(handles.len()); - for handle in handles { - if let Some(summary) = handle.summary().await { - summaries.push(summary); - } - } - summaries.sort_by(|left, right| { - right - .last_activity_at - .cmp(&left.last_activity_at) - .then_with(|| right.updated_at.cmp(&left.updated_at)) - }); - summaries - } - /// Reads turn reservation state, preferring runtime caches while a turn /// task holds the working copy (shared queue Arcs stay usable without a /// mailbox hop). Callers may mutate `pending_turn_queue` and @@ -257,29 +237,4 @@ impl ServerRuntime { let handle = self.session(session_id).await?; handle.collaboration_mode().await } - - /// Reads a session's parent id, preferring the in-flight turn inline snapshot - /// and agent-registry hierarchy over a mailbox round-trip. - #[allow(dead_code)] - pub(crate) async fn session_parent_id_snapshot( - &self, - session_id: SessionId, - ) -> Option> { - if let Some(stream) = self.active_stream_state(session_id).await { - let stream = stream.lock().await; - if let Some(inline) = stream.turn_inline.as_ref() { - return Some(inline.summary.parent_session_id()); - } - } - { - let registries = self.agent_registries.lock().await; - for registry in registries.values() { - if let Some(parent_id) = registry.child_to_parent.get(&session_id) { - return Some(Some(*parent_id)); - } - } - } - let handle = self.session(session_id).await?; - handle.parent_session_id().await - } }