From 2aaca2ca39953562305612080a02e765926de71c Mon Sep 17 00:00:00 2001 From: Biraj Date: Fri, 14 Aug 2026 15:24:11 -0700 Subject: [PATCH] fix(cli): recover externally owned agent sessions --- .changeset/spotty-snails-spend.md | 6 + biome.json | 2 +- cli/src/agents/base.ts | 102 ++++++++++- cli/src/agents/codex-readonly.ts | 258 ++++++++++++++++++++++++++ cli/src/agents/codex.ts | 51 ++++++ cli/src/agents/errors.ts | 22 +++ cli/src/agents/index.ts | 79 +++++++- cli/src/agents/process-scanner.ts | 132 +++++++++++++- cli/src/agents/session-watcher.ts | 288 ++++++++++++++++++++++-------- cli/src/agents/types.ts | 3 + cli/src/connection.ts | 9 + package.json | 2 +- pnpm-lock.yaml | 74 ++++---- protocol/src/ai-legacy.ts | 37 ++++ protocol/src/base.ts | 2 + protocol/src/client/to-host.ts | 2 + protocol/src/host/to-client.ts | 2 + 17 files changed, 939 insertions(+), 132 deletions(-) create mode 100644 .changeset/spotty-snails-spend.md create mode 100644 cli/src/agents/codex-readonly.ts diff --git a/.changeset/spotty-snails-spend.md b/.changeset/spotty-snails-spend.md new file mode 100644 index 0000000..d4d314a --- /dev/null +++ b/.changeset/spotty-snails-spend.md @@ -0,0 +1,6 @@ +--- +"@shellular/protocol": patch +"shellular": patch +--- + +fix(agents): handle externally owned sessions and defer Codex resume until prompting diff --git a/biome.json b/biome.json index 6f0a417..5cdaf06 100644 --- a/biome.json +++ b/biome.json @@ -1,5 +1,5 @@ { - "$schema": "https://biomejs.dev/schemas/2.5.7/schema.json", + "$schema": "https://biomejs.dev/schemas/2.5.8/schema.json", "vcs": { "enabled": true, "clientKind": "git", diff --git a/cli/src/agents/base.ts b/cli/src/agents/base.ts index 472efc8..d6411d4 100644 --- a/cli/src/agents/base.ts +++ b/cli/src/agents/base.ts @@ -20,7 +20,11 @@ import { type ElicitationListener, type PermissionListener, } from "./client"; -import { AgentUnavailableError, UnsupportedCapabilityError } from "./errors"; +import { + AgentUnavailableError, + AiNewError, + UnsupportedCapabilityError, +} from "./errors"; import { AcpTranscript, type AcpTranscriptOptions, @@ -78,6 +82,13 @@ export interface SpawnedAgent { stream: acp.Stream; } +interface LoadSessionFallback { + response: acp.LoadSessionResponse; + updates: acp.SessionNotification[]; + /** The agent must be resumed before the next prompt can be sent. */ + requiresResume?: boolean; +} + /** * Runtime wrapper for one ACP agent process. * @@ -94,6 +105,8 @@ export class ACP { private transcripts = new Map(); private sessions = new Map(); private loadingSessions = new Map>(); + private sessionsRequiringResume = new Set(); + private sessionResumeParams = new Map(); private activePromptSessionIds = new Set(); private stderrBuffer = ""; private state: AgentConnectionState = "unavailable"; @@ -431,6 +444,8 @@ export class ACP { session, messages: this.getMessages(params.sessionId), }); + this.sessionsRequiringResume.delete(params.sessionId); + this.sessionResumeParams.delete(params.sessionId); this.getTranscript(params.sessionId); return { response, session }; } @@ -470,6 +485,8 @@ export class ACP { ); this.sessions.delete(params.sessionId); this.transcripts.delete(params.sessionId); + this.sessionsRequiringResume.delete(params.sessionId); + this.sessionResumeParams.delete(params.sessionId); this.client.cancelSessionPermissions(params.sessionId); this.client.cancelSessionElicitations(params.sessionId); return response; @@ -534,6 +551,8 @@ export class ACP { raw, ); this.transcripts.set(sessionId, transcript); + this.sessionsRequiringResume.delete(sessionId); + this.sessionResumeParams.delete(sessionId); const messages = transcript.getMessages(); logger.debug( `ACP ${this.id}: session/load ${sessionId} replayed ${updates.length} updates -> ${messages.length} messages in ${Date.now() - loadStartedAt}ms (~${JSON.stringify(messages).length} bytes)`, @@ -553,6 +572,54 @@ export class ACP { messages, }); return { response, updates, messages }; + } catch (error) { + try { + const fallback = await this.loadSessionFallback(params, error); + if (fallback) { + for (const notification of fallback.updates) { + updates.push(notification); + transcript.apply(notification); + } + + const messages = transcript.getMessages(); + this.transcripts.set(sessionId, transcript); + if (fallback.requiresResume) { + this.sessionsRequiringResume.add(sessionId); + this.sessionResumeParams.set(sessionId, params); + } else { + this.sessionsRequiringResume.delete(sessionId); + this.sessionResumeParams.delete(sessionId); + } + const existing = this.sessions.get(sessionId); + this.sessions.set(sessionId, { + session: existing?.session + ? { + ...existing.session, + configOptions: + fallback.response.configOptions ?? + existing.session.configOptions, + } + : newAiSessionFromResponse( + { sessionId, configOptions: fallback.response.configOptions }, + path.resolve(params.cwd), + ), + messages, + }); + logger.warn( + `ACP ${this.id}: session/load used read-only fallback for ${sessionId} (${messages.length} messages)`, + ); + return { + response: fallback.response, + updates, + messages, + }; + } + } catch (fallbackError) { + logger.warn( + `ACP ${this.id}: session/load fallback failed for ${sessionId}: ${this.errorMessage(fallbackError)}`, + ); + } + throw error; } finally { // When session is loaded again, this is required to show the permission prompt again. this.client.requestPendingPermission(sessionId, clientId); @@ -572,6 +639,7 @@ export class ACP { if (loading) { await loading; } + await this.ensureWritableSession(params); const transcript = this.getTranscript(params.sessionId); let permissionRequested = false; @@ -646,6 +714,9 @@ export class ACP { properties: { sessionId: params.sessionId, error: this.errorMessage(err), + ...(err instanceof AiNewError + ? { errorCode: err.code, errorDetails: err.details } + : {}), }, }); throw err; @@ -749,6 +820,35 @@ export class ACP { return {}; } + /** + * Optional recovery path for agents whose native session/load cannot read a + * session while another process owns it. The default keeps ACP behavior + * unchanged for agents that do not need a read-only protocol. + */ + protected async loadSessionFallback( + _params: acp.LoadSessionRequest, + _error: unknown, + ): Promise { + return null; + } + + /** + * A read-only history load may not create an agent-owned session. Defer the + * optional ACP resume operation until the user actually sends a prompt. + */ + private async ensureWritableSession(params: acp.PromptRequest) { + if (!this.sessionsRequiringResume.has(params.sessionId)) return; + + const resumeParams = this.sessionResumeParams.get(params.sessionId); + if (!resumeParams) { + throw new Error( + `Session ${params.sessionId} was loaded read-only but has no resume parameters`, + ); + } + + await this.resumeSession(resumeParams); + } + protected setSessionStore(sessionId: string, stored: StoredSession) { this.sessions.set(sessionId, stored); } diff --git a/cli/src/agents/codex-readonly.ts b/cli/src/agents/codex-readonly.ts new file mode 100644 index 0000000..150b6b5 --- /dev/null +++ b/cli/src/agents/codex-readonly.ts @@ -0,0 +1,258 @@ +import { spawn } from "node:child_process"; +import { createInterface } from "node:readline"; +import type * as acp from "@agentclientprotocol/sdk"; + +const READ_TIMEOUT_MS = 30_000; + +interface CodexUserInput { + type?: unknown; + text?: unknown; +} + +interface CodexThread { + turns?: unknown; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function stringValue(value: unknown): string | undefined { + return typeof value === "string" ? value : undefined; +} + +function threadFromResponse(value: unknown): CodexThread | undefined { + if (!isRecord(value) || !isRecord(value.result)) return undefined; + return isRecord(value.result.thread) + ? { turns: value.result.thread.turns } + : undefined; +} + +function errorDetails(value: unknown): string | undefined { + if (!isRecord(value) || !isRecord(value.data)) return undefined; + return stringValue(value.data.details); +} + +export function isCodexActiveWriterError(error: unknown): boolean { + if ( + error instanceof Error && + error.message.includes("already has an active writer") + ) { + return true; + } + if (!isRecord(error)) return false; + const message = stringValue(error.message); + const details = errorDetails(error); + return [message, details].some( + (value) => value?.includes("already has an active writer") ?? false, + ); +} + +/** + * Read a Codex thread without resuming it. + * + * The ACP adapter currently implements session/load with thread/resume first. + * That is correct for an idle thread, but Codex rejects it when the original + * CLI still owns the thread's writer. app-server's thread/read is explicitly + * read-only and works in both cases. + */ +export async function readCodexThread( + command: string, + sessionId: string, + cwd: string, +): Promise { + const child = spawn(command, ["app-server", "--stdio"], { + cwd, + stdio: ["pipe", "pipe", "pipe"], + }); + + let stderr = ""; + child.stderr.on("data", (chunk: Buffer) => { + stderr = `${stderr}${chunk.toString("utf8")}`.slice(-4_000); + }); + + try { + return await new Promise((resolve, reject) => { + const readline = createInterface({ input: child.stdout }); + let settled = false; + const timeout = setTimeout(() => { + finish(() => + reject( + new Error( + `Codex read-only thread request timed out after ${READ_TIMEOUT_MS}ms`, + ), + ), + ); + }, READ_TIMEOUT_MS); + + const finish = (callback: () => void) => { + if (settled) return; + settled = true; + clearTimeout(timeout); + readline.close(); + callback(); + }; + + readline.on("line", (line) => { + let message: unknown; + try { + message = JSON.parse(line); + } catch { + return; + } + + if (!isRecord(message) || message.id !== 2) return; + const error = message.error; + if (isRecord(error)) { + const details = errorDetails(error); + finish(() => + reject( + new Error( + `${stringValue(error.message) ?? "Codex thread/read failed"}${details ? `: ${details}` : ""}`, + ), + ), + ); + return; + } + + const thread = threadFromResponse(message); + if (!thread) { + finish(() => + reject(new Error("Codex thread/read returned no thread")), + ); + return; + } + finish(() => resolve(thread)); + }); + + child.once("error", (error) => finish(() => reject(error))); + child.once("exit", (code, signal) => { + if (settled) return; + const diagnostic = stderr.trim(); + finish(() => + reject( + new Error( + `Codex app-server exited before thread/read completed (${signal ?? `code ${code}`})${diagnostic ? `: ${diagnostic}` : ""}`, + ), + ), + ); + }); + + child.stdin.write( + `${JSON.stringify({ + method: "initialize", + id: 1, + params: { + clientInfo: { + name: "shellular-readonly", + version: "1.0.0", + }, + }, + })}\n`, + ); + child.stdin.write( + `${JSON.stringify({ method: "initialized", params: {} })}\n`, + ); + child.stdin.write( + `${JSON.stringify({ + method: "thread/read", + id: 2, + params: { threadId: sessionId, includeTurns: true }, + })}\n`, + ); + }); + } finally { + child.kill(); + } +} + +function textInput(input: unknown): string | null { + if (!isRecord(input)) return null; + const value: CodexUserInput = input; + return value.type === "text" && typeof value.text === "string" + ? value.text + : null; +} + +function textList(value: unknown): string[] { + if (typeof value === "string") return value ? [value] : []; + if (!Array.isArray(value)) return []; + return value.filter( + (item): item is string => typeof item === "string" && item.length > 0, + ); +} + +function notification( + sessionId: string, + update: acp.SessionNotification["update"], +): acp.SessionNotification { + return { sessionId, update }; +} + +/** Convert the durable app-server thread projection into the ACP replay shape. */ +export function codexThreadToSessionUpdates( + sessionId: string, + thread: CodexThread, +) { + const updates: acp.SessionNotification[] = []; + if (!Array.isArray(thread.turns)) return updates; + + for (const turn of thread.turns) { + if (!isRecord(turn)) continue; + const items = turn.items; + if (!Array.isArray(items)) continue; + + for (const item of items) { + if (!isRecord(item)) continue; + const id = stringValue(item.id); + + switch (item.type) { + case "userMessage": { + if (!Array.isArray(item.content)) break; + for (const input of item.content) { + const text = textInput(input); + if (text) { + updates.push( + notification(sessionId, { + sessionUpdate: "user_message_chunk", + messageId: id, + content: { type: "text", text }, + }), + ); + } + } + break; + } + case "agentMessage": { + const text = stringValue(item.text); + if (!text) break; + updates.push( + notification(sessionId, { + sessionUpdate: "agent_message_chunk", + messageId: id, + content: { type: "text", text }, + }), + ); + break; + } + case "reasoning": { + for (const text of [ + ...textList(item.summary), + ...textList(item.content), + ]) { + updates.push( + notification(sessionId, { + sessionUpdate: "agent_thought_chunk", + messageId: id, + content: { type: "text", text }, + }), + ); + } + break; + } + } + } + } + + return updates; +} diff --git a/cli/src/agents/codex.ts b/cli/src/agents/codex.ts index d6fb764..76d6eed 100644 --- a/cli/src/agents/codex.ts +++ b/cli/src/agents/codex.ts @@ -1,6 +1,16 @@ +import type * as acp from "@agentclientprotocol/sdk"; +import type { AiSessionOwner } from "@shellular/protocol"; + import { BUILTIN_AGENT_DESCRIPTORS } from "./agents"; import { ACP } from "./base"; +import { + codexThreadToSessionUpdates, + isCodexActiveWriterError, + readCodexThread, +} from "./codex-readonly"; +import { SessionOwnedByProcessError } from "./errors"; import type { AcpTranscriptOptions } from "./events"; +import { findAgentProcesses } from "./process-scanner"; import { normalizeCodexUserReplayMessage } from "./replay-normalization"; export class Codex extends ACP { @@ -13,4 +23,45 @@ export class Codex extends ACP { normalizeUserReplayMessage: normalizeCodexUserReplayMessage, }; } + + protected override async loadSessionFallback( + params: acp.LoadSessionRequest, + error: unknown, + ) { + if (!isCodexActiveWriterError(error)) return null; + + const thread = await readCodexThread( + this.descriptor.agentExecutable, + params.sessionId, + params.cwd, + ); + + return { + response: { configOptions: [] }, + updates: codexThreadToSessionUpdates(params.sessionId, thread), + requiresResume: true, + }; + } + + override async resumeSession(params: acp.ResumeSessionRequest) { + try { + return await super.resumeSession(params); + } catch (error) { + if (!isCodexActiveWriterError(error)) throw error; + + const owner = await this.findOwner(params.cwd); + throw new SessionOwnedByProcessError(params.sessionId, params.cwd, owner); + } + } + + private async findOwner(cwd: string): Promise { + const process = (await findAgentProcesses("codex", cwd))[0]; + if (!process) return undefined; + return { + pid: process.pid, + command: process.command, + cwd: process.cwd, + ...(process.startedAt > 0 ? { startedAt: process.startedAt } : {}), + }; + } } diff --git a/cli/src/agents/errors.ts b/cli/src/agents/errors.ts index 4104464..47509e6 100644 --- a/cli/src/agents/errors.ts +++ b/cli/src/agents/errors.ts @@ -1,3 +1,5 @@ +import type { AiSessionOwner } from "@shellular/protocol"; + export class AiNewError extends Error { constructor( message: string, @@ -9,6 +11,26 @@ export class AiNewError extends Error { } } +export class SessionOwnedByProcessError extends AiNewError { + constructor( + sessionId: string, + workspacePath: string, + owner?: AiSessionOwner, + ) { + super( + owner + ? `Session ${sessionId} is owned by the running agent process ${owner.pid}` + : `Session ${sessionId} is owned by another running agent process`, + "ESESSION_OWNED_BY_PROCESS", + { + sessionId, + workspacePath, + owner: owner ?? null, + }, + ); + } +} + export class UnsupportedCapabilityError extends AiNewError { constructor(agentId: string, capability: string) { super( diff --git a/cli/src/agents/index.ts b/cli/src/agents/index.ts index 37900e8..ee83aab 100644 --- a/cli/src/agents/index.ts +++ b/cli/src/agents/index.ts @@ -17,12 +17,17 @@ import type { AiSession, AiSessionConfigOption, AiSessionCreateMsg, + AiSessionOwner, AiSessionRuntimeState, AiSessionState, CustomAcpAgentInput, ManagedAcpAgentInfo, } from "@shellular/protocol"; -import { AcpContentBlockSchema, MsgType } from "@shellular/protocol"; +import { + AcpContentBlockSchema, + AiSessionOwnerSchema, + MsgType, +} from "@shellular/protocol"; import { config } from "@/config"; import type { Connection } from "@/connection"; @@ -34,13 +39,14 @@ import { ClaudeCode } from "./claude-code"; import { Codex } from "./codex"; import { Copilot } from "./copilot"; import { Cursor } from "./cursor"; -import { AgentUnavailableError } from "./errors"; +import { AgentUnavailableError, AiNewError } from "./errors"; import { GrokActiveSessionsWatcher } from "./grok-active-sessions-watcher"; import { GrokBuild } from "./grok-build"; import { Hermes } from "./hermes"; import { NotifyBridge, type NotifyEvent } from "./notify-bridge"; import { OpenCode } from "./opencode"; import { Pi } from "./pi"; +import { terminateAgentProcess } from "./process-scanner"; import { readCachedSessionConfig, writeCachedSessionConfig, @@ -189,6 +195,18 @@ function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null; } +function errorEventProperties(error: unknown): Record { + if (!(error instanceof AiNewError)) return {}; + return { errorCode: error.code, errorDetails: error.details }; +} + +function sessionOwnerFromError(error: unknown): AiSessionOwner | undefined { + if (!(error instanceof AiNewError)) return undefined; + if (!isRecord(error.details)) return undefined; + const parsed = AiSessionOwnerSchema.safeParse(error.details.owner); + return parsed.success ? parsed.data : undefined; +} + function parseIsoTimestamp(value: unknown): number | undefined { if (typeof value !== "string") return undefined; const timestamp = Date.parse(value); @@ -311,6 +329,7 @@ export class AgentsManager { // must not overwrite the newer live transcript. private sessionTurnCounts = new Map(); private promptQueues = new Map(); + private sessionOwners = new Map(); constructor() { this.reloadDescriptors(); @@ -955,6 +974,9 @@ export class AgentsManager { window?: MessageWindow, ) { const startedAt = Date.now(); + logger.log( + `AI attach requested: agent=${agentId} session=${sessionId} workspace=${cwd}`, + ); const agent = await this.connectSessionAgent(clientId, agentId, sessionId); this.attachSessionClient(agentId, sessionId, clientId); this.rememberSessionClient(agentId, sessionId, clientId); @@ -1507,8 +1529,12 @@ export class AgentsManager { try { await this.prompt(item.clientId, agentId, sessionId, item.content); } catch (err) { + const owner = sessionOwnerFromError(err); + if (owner) { + this.sessionOwners.set(this.sessionKey(agentId, sessionId), owner); + } logger.error( - `Queued agent prompt failed for ${agentId} session ${sessionId} (client ${item.clientId}): ${getErrorMessage(err)}`, + `Agent prompt failed for ${agentId} session ${sessionId} (client ${item.clientId}): ${getErrorMessage(err)}`, err, ); this.emit(item.clientId, agentId, { @@ -1516,6 +1542,7 @@ export class AgentsManager { properties: { sessionId, error: getErrorMessage(err), + ...errorEventProperties(err), }, }); } @@ -1885,6 +1912,7 @@ export class AgentsManager { this.sessionRuntimes.clear(); this.sessionRuntimeCleanupTimers.clear(); this.sessionAgents.clear(); + this.sessionOwners.clear(); this.transcriptStore.close(); } @@ -2641,6 +2669,46 @@ export class AgentsManager { } }); + conn.on(MsgType.AI_SESSION_OWNER_KILL, async (msg) => { + const key = this.sessionKey(msg.data.backend, msg.data.sessionId); + const owner = this.sessionOwners.get(key); + if (!owner) { + conn.send({ + type: MsgType.AI_SESSION_OWNER_KILL_RESULT, + clientId: msg.clientId, + respTo: msg.id, + error: "The owning agent process is no longer known to this CLI", + }); + return; + } + + try { + const terminated = await terminateAgentProcess(msg.data.backend, { + ...owner, + startedAt: owner.startedAt ?? 0, + }); + if (!terminated) { + throw new Error( + "The owning agent process did not exit after the termination request", + ); + } + this.sessionOwners.delete(key); + conn.send({ + type: MsgType.AI_SESSION_OWNER_KILL_RESULT, + clientId: msg.clientId, + respTo: msg.id, + data: { ok: true, pid: owner.pid }, + }); + } catch (error) { + conn.send({ + type: MsgType.AI_SESSION_OWNER_KILL_RESULT, + clientId: msg.clientId, + respTo: msg.id, + error: getErrorMessage(error), + }); + } + }); + conn.on(MsgType.AI_ELICITATION_REPLY, async (msg) => { try { await this.replyElicitation( @@ -2929,7 +2997,10 @@ export class AgentsManager { }); } }) - .catch(() => {}) + .catch((error) => { + const message = error instanceof Error ? error.message : String(error); + logger.warn(`AI refresh ${key} failed: ${message}`); + }) .finally(() => { // Belt and braces: a replay that threw never reached the claim check // above, and a stale claim would silence the next session's push. diff --git a/cli/src/agents/process-scanner.ts b/cli/src/agents/process-scanner.ts index eddd19c..6ef10a7 100644 --- a/cli/src/agents/process-scanner.ts +++ b/cli/src/agents/process-scanner.ts @@ -59,13 +59,16 @@ function isAgentCommand(agent: AgentId, command: string): boolean { return base === "claude" || lower.includes("/claude/versions/"); } if (agent === "codex") { - return base === "codex" || /\/codex(\/|$|-)/.test(argv0); + // Codex has shipped both `codex` and `codex-tui` executables. Keep the + // match on argv[0] so arguments cannot create false positives, but accept + // the executable prefix used by both distributions. + return base === "codex" || base.startsWith("codex-"); } return false; } /** - * Returns candidate PIDs for the agent by scanning the process table. + * Returns candidate processes for the agent by scanning the process table. * * We use `ps`, NOT `pgrep`. On macOS, `pgrep -f` matches against a process's * argv read via KERN_PROCARGS2, which fails for hardened/signed binaries — and @@ -75,8 +78,15 @@ function isAgentCommand(agent: AgentId, command: string): boolean { * check come back negative and hid live-but-idle sessions. `ps` reads the * process table directly and lists them, so we scan its output ourselves. */ -async function candidatePids(agent: AgentId): Promise> { - const result = new Map(); +type CandidateProcess = { + startedAt: number; + command: string; +}; + +async function candidateProcesses( + agent: AgentId, +): Promise> { + const result = new Map(); try { // -A: all processes; -ww: don't truncate the command column. `lstart` is a // fixed-width absolute start time ("Fri Jul 31 15:29:20 2026"); it must come @@ -99,7 +109,10 @@ async function candidatePids(agent: AgentId): Promise> { if (!Number.isFinite(pid)) continue; if (!isAgentCommand(agent, match[3])) continue; const started = Date.parse(match[2]); - result.set(pid, Number.isFinite(started) ? started : 0); + result.set(pid, { + startedAt: Number.isFinite(started) ? started : 0, + command: match[3], + }); } } catch { // ps unavailable/failed — no candidates; callers treat liveness as unknown. @@ -161,8 +174,15 @@ export type LiveAgentCwds = { unknown: boolean; }; +export type AgentProcessInfo = { + pid: number; + startedAt: number; + cwd: string; + command: string; +}; + export async function liveAgentCwds(agent: AgentId): Promise { - const candidates = await candidatePids(agent); + const candidates = await candidateProcesses(agent); // No agent process at all is a definite answer, not an unknown one: this is // the post-reboot case, and it must clear every stale session. if (candidates.size === 0) return { cwds: new Map(), unknown: false }; @@ -176,11 +196,11 @@ export async function liveAgentCwds(agent: AgentId): Promise { if (process.platform === "linux") { let readAny = false; - for (const [pid, startedAt] of candidates) { + for (const [pid, candidate] of candidates) { const cwd = pidCwdLinux(pid); if (cwd) { readAny = true; - record(cwd, startedAt); + record(cwd, candidate.startedAt); } } return { cwds, unknown: !readAny }; @@ -189,7 +209,7 @@ export async function liveAgentCwds(agent: AgentId): Promise { if (process.platform === "darwin") { const pidCwds = await pidCwdsMacos([...candidates.keys()]); for (const [pid, cwd] of pidCwds) { - record(cwd, candidates.get(pid) ?? 0); + record(cwd, candidates.get(pid)?.startedAt ?? 0); } // lsof returned nothing for live candidates — we can't attribute them. return { cwds, unknown: pidCwds.size === 0 }; @@ -199,6 +219,100 @@ export async function liveAgentCwds(agent: AgentId): Promise { return { cwds, unknown: true }; } +/** + * Find agent processes whose working directory matches a session workspace. + * This is intentionally a fresh process-table lookup: callers use the result + * for a destructive action and must not rely on an old liveness snapshot. + */ +export async function findAgentProcesses( + agent: AgentId, + cwd: string, +): Promise { + const candidates = await candidateProcesses(agent); + if (candidates.size === 0) return []; + + const matches: AgentProcessInfo[] = []; + const record = (pid: number, processCwd: string) => { + if (processCwd !== cwd) return; + const candidate = candidates.get(pid); + if (!candidate) return; + matches.push({ + pid, + startedAt: candidate.startedAt, + cwd: processCwd, + command: candidate.command, + }); + }; + + if (process.platform === "linux") { + for (const pid of candidates.keys()) { + const processCwd = pidCwdLinux(pid); + if (processCwd) record(pid, processCwd); + } + } else if (process.platform === "darwin") { + const pidCwds = await pidCwdsMacos([...candidates.keys()]); + for (const [pid, processCwd] of pidCwds) { + record(pid, processCwd); + } + } + + return matches.sort((a, b) => a.startedAt - b.startedAt); +} + +/** + * Terminate a previously identified agent process after revalidating its PID, + * executable and cwd. A PID supplied by a client is never trusted by itself. + */ +export async function terminateAgentProcess( + agent: AgentId, + owner: AgentProcessInfo, +): Promise { + const current = (await findAgentProcesses(agent, owner.cwd)).find( + (processInfo) => processInfo.pid === owner.pid, + ); + if (!current) return false; + if ( + owner.startedAt > 0 && + current.startedAt > 0 && + owner.startedAt !== current.startedAt + ) { + return false; + } + + try { + process.kill(current.pid, "SIGTERM"); + } catch (error) { + return isProcessGoneError(error); + } + + return waitForProcessExit(current.pid); +} + +const PROCESS_EXIT_TIMEOUT_MS = 5000; +const PROCESS_EXIT_POLL_MS = 100; + +function isProcessGoneError(error: unknown): boolean { + return error instanceof Error && "code" in error && error.code === "ESRCH"; +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return !isProcessGoneError(error); + } +} + +async function waitForProcessExit(pid: number): Promise { + const deadline = Date.now() + PROCESS_EXIT_TIMEOUT_MS; + while (isProcessAlive(pid)) { + if (Date.now() >= deadline) return false; + await new Promise((resolve) => setTimeout(resolve, PROCESS_EXIT_POLL_MS)); + } + return true; +} + /** * Whether a live agent process can account for a session in `cwd` whose log was * last written at `lastWriteMs`. diff --git a/cli/src/agents/session-watcher.ts b/cli/src/agents/session-watcher.ts index 27d3583..b7917ff 100644 --- a/cli/src/agents/session-watcher.ts +++ b/cli/src/agents/session-watcher.ts @@ -25,18 +25,20 @@ import { * "working", and "finished" sessions that Shellular never started. * * Surfacing gate: a session is surfaced if its log was appended to within - * ACTIVE_WINDOW_MS (actively working/just finished), OR if it was touched - * within DISCOVERY_WINDOW_MS (2h) and a live agent process exists in its - * launch cwd (idle but CLI still open). Historical sessions beyond the - * discovery window are never surfaced. Once surfaced, the session is tracked. + * ACTIVE_WINDOW_MS (actively working/just finished), OR if a live agent process + * can be attributed to its launch cwd (idle but CLI still open). Historical + * sessions are bounded by DISCOVERY_WINDOW_MS unless that live-process check + * proves the session predates Shellular's startup. Once surfaced, the session + * is tracked. * * Retention: a session that finished (authoritatively, via a Stop hook or * task_complete marker) is sticky — it stays until the user explicitly dismisses * it, even if the CLI closes, because the user needs to check the result. A - * running/permission session that goes silent for KILL_CHECK_TIMEOUT_MS is - * disambiguated with a cheap pgrep check: if the agent process is dead, the CLI - * was killed/closed mid-turn → remove; if alive, the turn finished naturally → - * decay to a sticky finished. + * Claude/permission sessions that go silent for KILL_CHECK_TIMEOUT_MS are + * disambiguated with a cheap process check: if the agent process is dead, the + * CLI was killed/closed mid-turn → remove; if alive, the turn finished + * naturally → decay to a sticky finished. Codex keeps a task_started session + * working until its terminal lifecycle marker arrives. * * Neither agent records permission/approval prompts to disk, so those are * handled separately by the notify bridge. This watcher only reports presence @@ -66,6 +68,8 @@ export type ExternalSessionUpdate = { * authoritative ones are kill-checked after KILL_CHECK_TIMEOUT_MS. */ authoritativeFinished?: boolean; + /** Codex has started a turn and has not written its terminal lifecycle event. */ + turnInProgress?: boolean; }; type WatchTarget = { @@ -82,12 +86,10 @@ const DEBOUNCE_MS = 300; // at discovery without a process check. const ACTIVE_WINDOW_MS = 30 * 1000; // How long a non-authoritative running/finished session can be silent before we -// disambiguate "idle finished" from "killed" with a pgrep check. +// disambiguate "idle finished" from "killed" with a process scan. const KILL_CHECK_TIMEOUT_MS = 60 * 1000; -// Bounding window for discoverFresh's safety-net scan and the surfacing gate. -// Sessions whose log was touched within this window but beyond ACTIVE_WINDOW_MS -// are surfaced only if a live agent process exists in their cwd (pgrep check). -// Beyond this window, sessions are considered historical and not surfaced. +// Bounding window for discoverFresh's safety-net scan and the default surfacing +// gate. A known live process may still surface an older session at startup. const DISCOVERY_WINDOW_MS = 2 * 60 * 60 * 1000; // How often we decay state, drop killed sessions, and re-discover missed files. const DECAY_INTERVAL_MS = 10 * 1000; @@ -123,14 +125,16 @@ function statusLabel(status: AiSessionRuntimeStatus): string { } } +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + function safeParseJson(line: string): Record | undefined { const trimmed = line.trim(); if (!trimmed) return undefined; try { - const value = JSON.parse(trimmed); - return value && typeof value === "object" - ? (value as Record) - : undefined; + const value: unknown = JSON.parse(trimmed); + return isRecord(value) ? value : undefined; } catch { return undefined; } @@ -152,6 +156,11 @@ function parseTimestamp(value: unknown): number | undefined { // be far bigger than a tail window. Grow the read until a newline is found. const FIRST_LINE_MAX_BYTES = 512 * 1024; +// A few Codex rollout variants append session_meta after an initial record. +// Scan just beyond the maximum first-line size to recognize those variants +// without turning every session parse into a full-file read. +const CODEX_META_SCAN_BYTES = FIRST_LINE_MAX_BYTES + TAIL_BYTES; + async function readFirstLine(filePath: string): Promise { const handle = await open(filePath, "r"); try { @@ -172,6 +181,35 @@ async function readFirstLine(filePath: string): Promise { } } +async function readCodexSessionMeta( + filePath: string, + size: number, + first?: Record, +): Promise | undefined> { + if (first?.type === "session_meta") { + return isRecord(first.payload) ? first.payload : undefined; + } + if (size <= 0) return undefined; + + const handle = await open(filePath, "r"); + try { + const length = Math.min(size, CODEX_META_SCAN_BYTES); + const buffer = Buffer.alloc(length); + const { bytesRead } = await handle.read(buffer, 0, length, 0); + const text = buffer.subarray(0, bytesRead).toString("utf8"); + for (const line of text.split("\n")) { + if (!line.includes("session_meta")) continue; + const parsed = safeParseJson(line); + if (parsed?.type !== "session_meta") continue; + return isRecord(parsed.payload) ? parsed.payload : undefined; + } + } catch { + return undefined; + } finally { + await handle.close(); + } +} + /** Reads the last complete non-empty line of a jsonl file. */ async function readLastLine( filePath: string, @@ -321,7 +359,7 @@ async function claudeHasConversation( function claudeTurnInProgress(last?: Record): boolean { if (!last) return false; const type = last.type; - const message = last.message as Record | undefined; + const message = isRecord(last.message) ? last.message : undefined; // An assistant line whose stop_reason is tool_use means the model is about // to (or is) running tools — i.e. mid-turn. if (type === "assistant" && message) { @@ -339,7 +377,8 @@ function claudeTurnInProgress(last?: Record): boolean { (block) => block && typeof block === "object" && - (block as { type?: unknown }).type === "tool_result", + isRecord(block) && + block.type === "tool_result", ); } } @@ -422,7 +461,7 @@ async function readClaudeFirstPrompt( if (!line.includes('"user"')) continue; const parsed = safeParseJson(line); if (parsed?.type !== "user") continue; - const message = parsed.message as Record | undefined; + const message = isRecord(parsed.message) ? parsed.message : undefined; const prompt = extractUserText(message?.content); if (prompt) return prompt.slice(0, TITLE_MAX_LEN); } @@ -440,10 +479,11 @@ function extractUserText(content: unknown): string | undefined { if ( block && typeof block === "object" && - (block as { type?: unknown }).type === "text" && - typeof (block as { text?: unknown }).text === "string" + isRecord(block) && + block.type === "text" && + typeof block.text === "string" ) { - const text = (block as { text: string }).text.trim(); + const text = block.text.trim(); if (text && !text.startsWith("<")) return text; } } @@ -457,6 +497,39 @@ function extractUserText(content: unknown): string | undefined { // filename). Lifecycle is explicit: event_msg/task_started (running), // task_complete (finished), turn_aborted (cancelled). +/** + * Codex persists child/background agents in the same rollout directory as + * user sessions. Their rollout ids look valid, but they are not resumable + * user conversations through ACP session/load, so exposing them produces a + * dead activity entry with an empty transcript. + * + * `source` is the authoritative marker in current Codex versions. The other + * fields cover older or partially-written metadata. + */ +function isCodexBackgroundSession(meta?: Record): boolean { + if (!meta) return false; + + const source = meta.source; + if (typeof source === "string") { + if (source === "subagent" || source.startsWith("subagent_")) return true; + } else if (source && typeof source === "object") { + if (isRecord(source) && ("subagent" in source || "internal" in source)) { + return true; + } + } + + if (meta.thread_source === "subagent") return true; + + // These fields are only written for Codex child agents. Require a parent id + // so a future root-session metadata field cannot accidentally hide a session. + return ( + typeof meta.parent_thread_id === "string" && + (typeof meta.agent_path === "string" || + typeof meta.agent_role === "string" || + typeof meta.agent_nickname === "string") + ); +} + async function parseCodexSession( filePath: string, ): Promise { @@ -471,10 +544,8 @@ async function parseCodexSession( const firstLine = await readFirstLine(filePath); const first = firstLine ? safeParseJson(firstLine) : undefined; - const meta = - first?.type === "session_meta" - ? (first.payload as Record | undefined) - : undefined; + const meta = await readCodexSessionMeta(filePath, stat.size, first); + if (isCodexBackgroundSession(meta)) return undefined; const sessionId = (typeof meta?.id === "string" && meta.id) || codexIdFromFilename(filePath); if (!sessionId) return undefined; @@ -482,12 +553,15 @@ async function parseCodexSession( const lastEvent = await readCodexLastLifecycle(filePath, stat.size); const mtime = stat.mtimeMs; - const recentlyActive = Date.now() - mtime <= ACTIVE_WINDOW_MS; let status: AiSessionRuntimeStatus; + const turnInProgress = lastEvent?.type === "task_started"; if (lastEvent?.type === "turn_aborted") { status = "cancelled"; - } else if (lastEvent?.type === "task_started" && recentlyActive) { + } else if (turnInProgress) { + // A turn can spend longer than ACTIVE_WINDOW_MS thinking or waiting on a + // tool without appending another lifecycle marker. task_complete is the + // authoritative end; mtime alone made these sessions appear finished. status = "running"; } else { status = "finished"; @@ -506,6 +580,7 @@ async function parseCodexSession( workspacePath, title, message: statusLabel(status) || undefined, + turnInProgress, authoritativeFinished: lastEvent?.type === "task_complete" || lastEvent?.type === "turn_aborted", }; @@ -528,7 +603,7 @@ async function readCodexTitle( continue; } const parsed = safeParseJson(line); - const payload = parsed?.payload as Record | undefined; + const payload = isRecord(parsed?.payload) ? parsed.payload : undefined; if (!payload) continue; // event_msg/user_message carries the prompt as a plain string. if ( @@ -558,10 +633,11 @@ function extractCodexInputText(content: unknown): string | undefined { if ( block && typeof block === "object" && - (block as { type?: unknown }).type === "input_text" && - typeof (block as { text?: unknown }).text === "string" + isRecord(block) && + block.type === "input_text" && + typeof block.text === "string" ) { - const text = (block as { text: string }).text.trim(); + const text = block.text.trim(); if (text && !text.startsWith("<")) return text; } } @@ -577,6 +653,11 @@ function codexIdFromFilename(filePath: string): string | undefined { return match?.[0]; } +// A large assistant/tool record can push the lifecycle marker more than one +// tail window away. Expand a bounded read in that case so status does not +// silently fall back to "finished" just because the marker is out of view. +const CODEX_LIFECYCLE_MAX_BYTES = 512 * 1024; + /** Finds the most recent task lifecycle event by scanning the tail. */ async function readCodexLastLifecycle( filePath: string, @@ -585,37 +666,53 @@ async function readCodexLastLifecycle( if (size <= 0) return undefined; const handle = await open(filePath, "r"); try { - const start = Math.max(0, size - TAIL_BYTES); - const length = size - start; - const buffer = Buffer.alloc(length); - const { bytesRead } = await handle.read(buffer, 0, length, start); - const text = buffer.subarray(0, bytesRead).toString("utf8"); - const lines = text.split("\n"); - let result: { type: string; timestamp?: number } | undefined; - for (const line of lines) { - if ( - !line.includes("task_started") && - !line.includes("task_complete") && - !line.includes("turn_aborted") - ) { - continue; + for ( + let window = Math.min(size, TAIL_BYTES); + ; + window = Math.min(size, window * 4) + ) { + const start = Math.max(0, size - window); + const buffer = Buffer.alloc(size - start); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, start); + const text = buffer.subarray(0, bytesRead).toString("utf8"); + const lines = text.split("\n"); + // The first line may begin in the middle of a JSON record. It cannot be + // the latest lifecycle record, so do not let a partial parse influence it. + if (start > 0) lines.shift(); + + let result: { type: string; timestamp?: number } | undefined; + for (const line of lines) { + if ( + !line.includes("task_started") && + !line.includes("task_complete") && + !line.includes("turn_aborted") + ) { + continue; + } + const parsed = safeParseJson(line); + if (parsed?.type !== "event_msg") continue; + const payload = isRecord(parsed.payload) ? parsed.payload : undefined; + const type = payload?.type; + if ( + type === "task_started" || + type === "task_complete" || + type === "turn_aborted" + ) { + result = { + type, + timestamp: parseTimestamp(parsed.timestamp), + }; + } } - const parsed = safeParseJson(line); - if (parsed?.type !== "event_msg") continue; - const payload = parsed.payload as Record | undefined; - const type = payload?.type; if ( - type === "task_started" || - type === "task_complete" || - type === "turn_aborted" + result || + start === 0 || + window >= CODEX_LIFECYCLE_MAX_BYTES || + window >= size ) { - result = { - type, - timestamp: parseTimestamp(parsed.timestamp), - }; + return result; } } - return result; } catch { return undefined; } finally { @@ -671,7 +768,9 @@ export class SessionWatcher { for (const target of this.targets) { this.watchTarget(target); } - this.seed(); + void this.seed().catch((err) => { + logger.debug("SessionWatcher: startup discovery failed:", err); + }); this.decayTimer = setInterval(() => { void this.reconcile(); @@ -765,12 +864,25 @@ export class SessionWatcher { // - Between ACTIVE_WINDOW_MS and DISCOVERY_WINDOW_MS (2h): the session // is idle but the CLI might still be open. Surface only if a live // agent process in its launch cwd can actually account for it. - // - Beyond DISCOVERY_WINDOW_MS: historical, never surface. + // - Beyond DISCOVERY_WINDOW_MS: surface only when a known live process + // started before the last write can account for the session. // Already-tracked sessions always get processed (new activity). if (!this.tracked.has(key)) { const age = now - update.updatedAt; - if (age > DISCOVERY_WINDOW_MS) return; - if (age > ACTIVE_WINDOW_MS) { + if (age > DISCOVERY_WINDOW_MS) { + const resolved = + options.live ?? (await liveAgentCwds(update.agentId)); + if ( + resolved.unknown || + !(await this.canSurfaceIdle(update, resolved)) + ) { + return; + } + } else if (age > ACTIVE_WINDOW_MS) { + // A live process is allowed to vouch for an older rollout too. This + // is what lets startup discover a CLI that was already open before + // Shellular started; the process start-time check still rejects stale + // history in the same workspace. if (!(await this.canSurfaceIdle(update, options.live))) return; } } @@ -879,7 +991,7 @@ export class SessionWatcher { * for missed fs.watch events), decay stale running -> finished, and detect * killed CLIs. Authoritative-finished sessions are sticky and never removed * here. Non-authoritative sessions that have been silent for - * KILL_CHECK_TIMEOUT_MS are disambiguated with a cheap pgrep check: alive → + * KILL_CHECK_TIMEOUT_MS are disambiguated with a cheap process scan: alive → * upgrade to sticky finished; dead → remove. */ private async reconcile() { @@ -904,8 +1016,30 @@ export class SessionWatcher { const quietMs = now - update.updatedAt; + // Codex writes task_complete when a turn ends. Until that marker arrives, + // a quiet log is still an in-progress turn: model/tool work can run for + // longer than ACTIVE_WINDOW_MS without touching the rollout. Only remove + // it when the owning CLI is definitely gone. + if ( + update.agentId === "codex" && + update.status === "running" && + update.turnInProgress + ) { + if (quietMs <= ACTIVE_WINDOW_MS) continue; + const alive = await isAgentAliveInCwd( + update.agentId, + update.workspacePath, + update.updatedAt, + ); + if (alive === "dead") { + this.tracked.delete(key); + this.onRemove(update.agentId, update.sessionId); + } + continue; + } + // Running session that went quiet: decay to finished. If it stays - // quiet past KILL_CHECK_TIMEOUT_MS, the reconcile below will pgrep. + // quiet past KILL_CHECK_TIMEOUT_MS, the reconcile below scans processes. if (update.status === "running" && quietMs > ACTIVE_WINDOW_MS) { const finished: ExternalSessionUpdate = { ...update, @@ -922,7 +1056,7 @@ export class SessionWatcher { } // Non-authoritative finished or waiting-for-permission, silent long - // enough to suspect the CLI was killed: disambiguate with pgrep. + // enough to suspect the CLI was killed: disambiguate with a process scan. if (quietMs <= KILL_CHECK_TIMEOUT_MS) continue; if ( update.status === "waiting_for_permission" || @@ -956,7 +1090,7 @@ export class SessionWatcher { } // Non-authoritative finished, silent past kill-check timeout: the CLI - // is probably gone. Disambiguate with pgrep; if alive, upgrade to + // is probably gone. Disambiguate with a process scan; if alive, upgrade to // sticky so we stop checking. if (update.status === "finished") { const alive = await isAgentAliveInCwd( @@ -977,12 +1111,16 @@ export class SessionWatcher { } } - private seed() { + private async seed() { for (const target of this.targets) { if (!existsSync(target.root)) continue; + // Share one process scan across the startup seed. Without this, every + // idle file independently scans ps/lsof, which is expensive on machines + // with many recent rollout files. + const live = await liveAgentCwds(target.agentId); const files = this.recentFiles(target.root, SEED_LIMIT); for (const filePath of files) { - void this.processFile(target, filePath); + await this.processFile(target, filePath, { live }); } } } @@ -1023,30 +1161,22 @@ export class SessionWatcher { * CLI left open) is never re-examined and can be missed if its one surfacing * check didn't happen at the right moment. Here liveness drives discovery * instead of file writes: we enumerate live agent process cwds once, then - * surface any untracked recent log those processes can account for (see + * surface any untracked log those processes can account for (see * canSurfaceIdle). Cheap on idle machines — the process scan short-circuits to * an empty set when no agent runs. */ private async discoverLiveSessions() { - const now = Date.now(); for (const target of this.targets) { if (!existsSync(target.root)) continue; const live = await liveAgentCwds(target.agentId); if (live.cwds.size === 0 && !live.unknown) continue; for (const filePath of this.recentFiles(target.root, SEED_LIMIT)) { - let mtime: number; - try { - mtime = statSync(filePath).mtimeMs; - } catch { - continue; - } - if (now - mtime > DISCOVERY_WINDOW_MS) continue; // Already-surfaced sessions are re-reported cheaply (report() dedupes), // so we only skip re-parsing when nothing about the file changed since // we last surfaced it — the common idle case. if (this.tracked.has(this.trackedKeyForFile(target, filePath))) continue; - void this.processFile(target, filePath, { live }); + await this.processFile(target, filePath, { live }); } } } diff --git a/cli/src/agents/types.ts b/cli/src/agents/types.ts index f42b915..f66c1c0 100644 --- a/cli/src/agents/types.ts +++ b/cli/src/agents/types.ts @@ -4,6 +4,7 @@ import type { AcpMessage, AgentId, AiEvent, + AiSessionOwner, } from "@shellular/protocol"; export type AgentConnectionState = @@ -84,6 +85,8 @@ export interface PromptCallbacks { onUpdate?: (notification: acp.SessionNotification) => void; } +export type SessionOwnerProcess = AiSessionOwner; + export interface LoadSessionResult { response: acp.LoadSessionResponse; updates: acp.SessionNotification[]; diff --git a/cli/src/connection.ts b/cli/src/connection.ts index 7aa2ccc..753e619 100644 --- a/cli/src/connection.ts +++ b/cli/src/connection.ts @@ -35,6 +35,7 @@ import { type AiSessionGetMsg, type AiSessionListMsg, type AiSessionModeSetMsg, + type AiSessionOwnerKillMsg, type AiSessionResumeMsg, type AiShareMsg, type AiUnrevertMsg, @@ -368,6 +369,10 @@ export class Connection extends EventEmitter { eventName: typeof MsgType.AI_ABORT, listener: (msg: AiAbortMsg) => void, ): this; + on( + eventName: typeof MsgType.AI_SESSION_OWNER_KILL, + listener: (msg: AiSessionOwnerKillMsg) => void, + ): this; on( eventName: typeof MsgType.AI_AGENTS_LIST, listener: (msg: AiAgentsListMsg) => void, @@ -696,6 +701,10 @@ export class Connection extends EventEmitter { msg: AiAttachmentWriteMsg, ): boolean; emit(eventName: typeof MsgType.AI_ABORT, msg: AiAbortMsg): boolean; + emit( + eventName: typeof MsgType.AI_SESSION_OWNER_KILL, + msg: AiSessionOwnerKillMsg, + ): boolean; emit(eventName: typeof MsgType.AI_AGENTS_LIST, msg: AiAgentsListMsg): boolean; emit( eventName: typeof MsgType.AI_ACTIVITY_LIST, diff --git a/package.json b/package.json index d574997..ab5bb2b 100644 --- a/package.json +++ b/package.json @@ -12,7 +12,7 @@ "author": "", "license": "AGPL-3.0-only", "devDependencies": { - "@biomejs/biome": "2.5.7", + "@biomejs/biome": "2.5.8", "@changesets/cli": "^2.31.0" } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 07da9c2..2ced6bc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -9,8 +9,8 @@ importers: .: devDependencies: '@biomejs/biome': - specifier: 2.5.7 - version: 2.5.7 + specifier: 2.5.8 + version: 2.5.8 '@changesets/cli': specifier: ^2.31.0 version: 2.31.0(@types/node@22.19.17) @@ -134,59 +134,59 @@ packages: resolution: {integrity: sha512-JiDShH45zKHWyGe4ZNVRrCjBz8Nh9TMmZG1kh4QTK8hCBTWBi8Da+i7s1fJw7/lYpM4ccepSNfqzZ/QvABBi5g==} engines: {node: '>=6.9.0'} - '@biomejs/biome@2.5.7': - resolution: {integrity: sha512-zr8K/DcY5tYsQOQwqMJ0AWElo6QgmgNI7idXgXLhevVszlt8RGVpesEJPqx3ThazLaOwjJ5Y8fz3BtH5fGZNsw==} + '@biomejs/biome@2.5.8': + resolution: {integrity: sha512-aeAeeJB9fSDc7Gq+2GqpQxA0qBj6gj1k2R6L1cYqGePKP/baIq1WX8y6B+D+nRsO5ViQL22K/8IwbqERW0q1nw==} engines: {node: '>=14.21.3'} hasBin: true - '@biomejs/cli-darwin-arm64@2.5.7': - resolution: {integrity: sha512-vxo/Ls3/PYdQWyLhYYcgMOCzQypAjcY+iihS8M0wW03l16TCLW4zqZzGo75gm1VdCMj38hTVZ31KBWrZ4G9dJw==} + '@biomejs/cli-darwin-arm64@2.5.8': + resolution: {integrity: sha512-mk1QON9PHllvvLN5gU3f4rMxeh4syK5p9OvKyWH6/W8ueh04uaC8TUXXByhGufWf/y5mQc03ZLM45zU+cmqMjA==} engines: {node: '>=14.21.3'} cpu: [arm64] os: [darwin] - '@biomejs/cli-darwin-x64@2.5.7': - resolution: {integrity: sha512-Cd3Ga61amT/Yl/0x8elP5hhGYaFy4bw6WuysTgf7oo8TA5tJ5A1k+DkVoJ2BHbTVil51gTX9VPzArnrlLJ3Kyg==} + '@biomejs/cli-darwin-x64@2.5.8': + resolution: {integrity: sha512-bsGwFMBNyHPyiLSsQcZJxdoRrg1V4JL+d7wEsvUBczlP9U9lwM+7mzQHxI4o1mhBsTmdOBbAb6fHU3Z3snN45w==} engines: {node: '>=14.21.3'} cpu: [x64] os: [darwin] - '@biomejs/cli-linux-arm64-musl@2.5.7': - resolution: {integrity: sha512-xPI5yB6XlpDbNkS+bm1t42olw5c4l3UrlOmLg7KtLJvjvkNF/1V4tnUgfkylGIeb3u/T+BzMGYqgQhzjAoJzuQ==} + '@biomejs/cli-linux-arm64-musl@2.5.8': + resolution: {integrity: sha512-VcJNbstduTHx83NGAdhp78/JOcP45BZHXL7yNsfI1uGzdUgegAz2s+mSoT7wK6PBNzLoqG0zDOXaz/RQYVtSiw==} engines: {node: '>=14.21.3'} cpu: [arm64] os: [linux] libc: [musl] - '@biomejs/cli-linux-arm64@2.5.7': - resolution: {integrity: sha512-rR2QE0yF2GYSuYuKIa7pKvODGJqnOH+2eDREAM8wV+mWKSkMQKdAp4zXEZfTaxY8PMoNONnpgSWcBCyLDPDOKg==} + '@biomejs/cli-linux-arm64@2.5.8': + resolution: {integrity: sha512-XmFiA0WPYFC+uiUDC8WRFzAIH9bo7vwQLav38Uoq4ETC+T/+uBi0TsYGJECkugY3r8USl3jc+Ae2/irAF6F2lQ==} engines: {node: '>=14.21.3'} cpu: [arm64] os: [linux] libc: [glibc] - '@biomejs/cli-linux-x64-musl@2.5.7': - resolution: {integrity: sha512-rE5VZi+qtmPgQH+l7jVxYoZ18b/TiHEhulhMpjmCZH1PltSbjRcxNWywC3HZ9tYottG7ORkeTtoscBilKSBm0g==} + '@biomejs/cli-linux-x64-musl@2.5.8': + resolution: {integrity: sha512-kKmiyokeISRGq2FLwvr+TzsgBusfxaZ0FZNLcOYOpCK/78tRrEjeEBLvq3xLZMpqbANgJdRPI7vZX8ZL37u9/w==} engines: {node: '>=14.21.3'} cpu: [x64] os: [linux] libc: [musl] - '@biomejs/cli-linux-x64@2.5.7': - resolution: {integrity: sha512-FQgqJhscrqJUFptGaRSUJWlXAExwWcDwLuK49dvKfkQ1bB5SEEyFssnsxQY83Xm6jR0EbbX3+8+D5bfvYqUG2Q==} + '@biomejs/cli-linux-x64@2.5.8': + resolution: {integrity: sha512-S5wcm9OBDvLHodD4PUaN488hCpco9QD/9ZxuYJiw4euWtr/oQvLR72z2ixItH8Wd5BCm6FZaeb+YNvOoM1xHtQ==} engines: {node: '>=14.21.3'} cpu: [x64] os: [linux] libc: [glibc] - '@biomejs/cli-win32-arm64@2.5.7': - resolution: {integrity: sha512-Oq4x0CCwP4jirrcTywXs5kOGZ4v5vuEP+gWrbtjApOA2CL9F3F9GlIdQIci8AKSCa/zURanMRpX/4wQ7Am6hHg==} + '@biomejs/cli-win32-arm64@2.5.8': + resolution: {integrity: sha512-nILH0mzm3Hi3iEdd7o7GpB8kBR/mSQwfQG/tyBqyNrY2GFtcgwfV9nV8xLmbtUpMNY/Oi0Ml1XgfR4flOdq+AA==} engines: {node: '>=14.21.3'} cpu: [arm64] os: [win32] - '@biomejs/cli-win32-x64@2.5.7': - resolution: {integrity: sha512-V+0wu/nrj2S+MhP4EQ0uHNolP0IALEsz45pg0WoKkHfDeh0+ItHwP/p7bX5RPoMOl9NkpHYWdYPhIcy2mACHvQ==} + '@biomejs/cli-win32-x64@2.5.8': + resolution: {integrity: sha512-I2czzXTY61f3nFJxXoMDq80t7MivxDEnCjE+8sDKoFfcKMaoQdkqhIFQ3KyY0XLzeSpUBYeNAXgD+iOV/BU0VA==} engines: {node: '>=14.21.3'} cpu: [x64] os: [win32] @@ -1869,39 +1869,39 @@ snapshots: '@babel/runtime@7.29.2': {} - '@biomejs/biome@2.5.7': + '@biomejs/biome@2.5.8': optionalDependencies: - '@biomejs/cli-darwin-arm64': 2.5.7 - '@biomejs/cli-darwin-x64': 2.5.7 - '@biomejs/cli-linux-arm64': 2.5.7 - '@biomejs/cli-linux-arm64-musl': 2.5.7 - '@biomejs/cli-linux-x64': 2.5.7 - '@biomejs/cli-linux-x64-musl': 2.5.7 - '@biomejs/cli-win32-arm64': 2.5.7 - '@biomejs/cli-win32-x64': 2.5.7 + '@biomejs/cli-darwin-arm64': 2.5.8 + '@biomejs/cli-darwin-x64': 2.5.8 + '@biomejs/cli-linux-arm64': 2.5.8 + '@biomejs/cli-linux-arm64-musl': 2.5.8 + '@biomejs/cli-linux-x64': 2.5.8 + '@biomejs/cli-linux-x64-musl': 2.5.8 + '@biomejs/cli-win32-arm64': 2.5.8 + '@biomejs/cli-win32-x64': 2.5.8 - '@biomejs/cli-darwin-arm64@2.5.7': + '@biomejs/cli-darwin-arm64@2.5.8': optional: true - '@biomejs/cli-darwin-x64@2.5.7': + '@biomejs/cli-darwin-x64@2.5.8': optional: true - '@biomejs/cli-linux-arm64-musl@2.5.7': + '@biomejs/cli-linux-arm64-musl@2.5.8': optional: true - '@biomejs/cli-linux-arm64@2.5.7': + '@biomejs/cli-linux-arm64@2.5.8': optional: true - '@biomejs/cli-linux-x64-musl@2.5.7': + '@biomejs/cli-linux-x64-musl@2.5.8': optional: true - '@biomejs/cli-linux-x64@2.5.7': + '@biomejs/cli-linux-x64@2.5.8': optional: true - '@biomejs/cli-win32-arm64@2.5.7': + '@biomejs/cli-win32-arm64@2.5.8': optional: true - '@biomejs/cli-win32-x64@2.5.7': + '@biomejs/cli-win32-x64@2.5.8': optional: true '@changesets/apply-release-plan@7.1.1': diff --git a/protocol/src/ai-legacy.ts b/protocol/src/ai-legacy.ts index 534aa78..292c9ed 100644 --- a/protocol/src/ai-legacy.ts +++ b/protocol/src/ai-legacy.ts @@ -29,6 +29,15 @@ export const AiSessionSchema = z.object({ }); export type AiSession = z.infer; +/** A locally running agent process that currently owns a session writer. */ +export const AiSessionOwnerSchema = z.object({ + pid: z.number().int().positive(), + command: z.string(), + cwd: z.string(), + startedAt: z.number().int().nonnegative().optional(), +}); +export type AiSessionOwner = z.infer; + // ─── Message parts ──────────────────────────────────────────────────────────── const AiMessagePartTextSchema = z.object({ @@ -339,6 +348,17 @@ export const AiAbortMsgSchema = z.object({ }); export type AiAbortMsg = z.infer; +export const AiSessionOwnerKillMsgSchema = z.object({ + id: z.string(), + type: z.literal(MsgType.AI_SESSION_OWNER_KILL), + clientId: z.string(), + data: z.object({ + backend: AiBackendSchema, + sessionId: z.string(), + }), +}); +export type AiSessionOwnerKillMsg = z.infer; + export const AiAgentsListMsgSchema = z.object({ id: z.string(), type: z.literal(MsgType.AI_AGENTS_LIST), @@ -633,6 +653,23 @@ export const AiAbortAckMsgSchema = z.object({ }); export type AiAbortAckMsg = z.infer; +export const AiSessionOwnerKillResultMsgSchema = z.object({ + id: z.string().optional(), + type: z.literal(MsgType.AI_SESSION_OWNER_KILL_RESULT), + clientId: z.string(), + respTo: z.string().optional(), + error: z.string().optional(), + data: z + .object({ + ok: z.boolean(), + pid: z.number().int().positive().optional(), + }) + .optional(), +}); +export type AiSessionOwnerKillResultMsg = z.infer< + typeof AiSessionOwnerKillResultMsgSchema +>; + export const AiAgentsListResultMsgSchema = z.object({ id: z.string().optional(), type: z.literal(MsgType.AI_AGENTS_LIST_RESULT), diff --git a/protocol/src/base.ts b/protocol/src/base.ts index 6ce0b4e..5745f65 100644 --- a/protocol/src/base.ts +++ b/protocol/src/base.ts @@ -154,6 +154,8 @@ export const MsgType = { AI_SESSION_CONFIG_SET_RESULT: "ai:session:config:set:result", AI_ABORT: "ai:abort", AI_ABORT_ACK: "ai:abort:ack", + AI_SESSION_OWNER_KILL: "ai:session-owner:kill", + AI_SESSION_OWNER_KILL_RESULT: "ai:session-owner:kill:result", AI_EVENT: "ai:event", AI_AGENTS_LIST: "ai:agents:list", AI_AGENTS_LIST_RESULT: "ai:agents:list:result", diff --git a/protocol/src/client/to-host.ts b/protocol/src/client/to-host.ts index 214432c..f98c76a 100644 --- a/protocol/src/client/to-host.ts +++ b/protocol/src/client/to-host.ts @@ -36,6 +36,7 @@ import { AiSessionDeleteMsgSchema, AiSessionGetMsgSchema, AiSessionListMsgSchema, + AiSessionOwnerKillMsgSchema, AiShareMsgSchema, AiUnrevertMsgSchema, } from "@/ai-legacy"; @@ -121,6 +122,7 @@ export const ClientToHostMsgSchema = z.discriminatedUnion("type", [ AiSessionConfigSetMsgSchema, AiSessionModeSetMsgSchema, AiAbortMsgSchema, + AiSessionOwnerKillMsgSchema, AiActivityDismissMsgSchema, AiActivityListMsgSchema, AiAgentsListMsgSchema, diff --git a/protocol/src/host/to-client.ts b/protocol/src/host/to-client.ts index b6b3d4c..7ffe6b2 100644 --- a/protocol/src/host/to-client.ts +++ b/protocol/src/host/to-client.ts @@ -37,6 +37,7 @@ import { AiSessionDeletedMsgSchema, AiSessionGetResultMsgSchema, AiSessionListResultMsgSchema, + AiSessionOwnerKillResultMsgSchema, AiShareResultMsgSchema, AiUnrevertAckMsgSchema, } from "@/ai-legacy"; @@ -121,6 +122,7 @@ export const HostToClientSchema = z.discriminatedUnion("type", [ AiSessionConfigSetResultMsgSchema, AiSessionModeSetResultMsgSchema, AiAbortAckMsgSchema, + AiSessionOwnerKillResultMsgSchema, AiActivityDismissResultMsgSchema, AiActivityListResultMsgSchema, AiAgentsListResultMsgSchema,