diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index f7100b579a..a7336c3a87 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,8 @@ ## [Unreleased] +- Fixed empty "ghost" session files being created on disk before the first assistant message by preventing `session_state` entries from triggering file persistence. +- Added daemon startup sweeps for stale session leases and orphaned ghost session files. - Added privacy-safe pseudonymous product analytics for onboarding, command use, execution modes, run outcomes, TTFT, latency, usage, tools, retries, and compactions, with disclosure and opt-out controls ([ENG-4682](https://linear.app/primeintellect/issue/ENG-4682/add-privacy-safe-posthog-analytics-to-prime-agent)). - Changed sent agent messages in the IPython cell UI to show only the message text with a `╰─` gutter when expanded, matching received messages, and hid the raw `agent_message.send` receipt dictionary. - Fixed Homebrew installs attempting to self-update their versioned Cellar keg instead of directing users to `brew upgrade prime-agent` ([#844](https://github.com/PrimeIntellect-ai/prime-agent/issues/844)) diff --git a/packages/coding-agent/src/core/session-file-actions.ts b/packages/coding-agent/src/core/session-file-actions.ts index adcec4c012..be341fdb2c 100644 --- a/packages/coding-agent/src/core/session-file-actions.ts +++ b/packages/coding-agent/src/core/session-file-actions.ts @@ -1,5 +1,5 @@ import { spawnSync } from "node:child_process"; -import { existsSync } from "node:fs"; +import { existsSync, readdirSync, readFileSync } from "node:fs"; import { rm, unlink } from "node:fs/promises"; import { basename, dirname, join } from "node:path"; @@ -72,3 +72,57 @@ export async function deleteSessionFile( } return result; } + +// Entry types that are always present when a session is created. A file that +// contains only these plus session_state holds nothing worth keeping. +const BOOTSTRAP_ENTRY_TYPES = new Set(["session", "model_change", "thinking_level_change", "service_tier_change"]); + +/** + * True when a session file is an empty draft: it has no messages and its only + * entries are the bootstrap prefix (session header, model/thinking/tier changes) + * optionally followed by a daemon-written session_state. Such files are ghost + * sessions — created on disk for crash recovery but never receiving a message. + */ +function isEmptyDraftSessionFile(sessionPath: string): boolean { + let content: string; + try { + content = readFileSync(sessionPath, "utf8"); + } catch { + return false; + } + for (const line of content.split("\n")) { + if (!line.trim()) continue; + let entry: { type?: string }; + try { + entry = JSON.parse(line); + } catch { + return false; + } + const type = entry.type; + if (!type) return false; + if (BOOTSTRAP_ENTRY_TYPES.has(type)) continue; + if (type === "session_state") continue; + // Any other entry type means the session has real content. + return false; + } + return true; +} + +/** + * Scan a session directory for ghost session files — empty drafts left behind + * when a daemon shuts down or crashes before the user sends their first message. + * Returns the count of files removed. + */ +export async function sweepGhostSessionFiles(sessionDir: string): Promise { + if (!existsSync(sessionDir)) return 0; + let swept = 0; + for (const entry of readdirSync(sessionDir)) { + if (!entry.endsWith(".jsonl")) continue; + const filePath = join(sessionDir, entry); + if (isEmptyDraftSessionFile(filePath)) { + await deleteSessionFile(filePath).catch(() => undefined); + swept++; + } + } + return swept; +} diff --git a/packages/coding-agent/src/core/session-lease.ts b/packages/coding-agent/src/core/session-lease.ts index 6c4e2975cf..e3ec6b4014 100644 --- a/packages/coding-agent/src/core/session-lease.ts +++ b/packages/coding-agent/src/core/session-lease.ts @@ -1,6 +1,15 @@ import { execFileSync } from "node:child_process"; import { createHash, randomUUID } from "node:crypto"; -import { existsSync, mkdirSync, readFileSync, realpathSync, renameSync, rmSync, writeFileSync } from "node:fs"; +import { + existsSync, + mkdirSync, + readdirSync, + readFileSync, + realpathSync, + renameSync, + rmSync, + writeFileSync, +} from "node:fs"; import { basename, dirname, join, resolve } from "node:path"; import { lockSync } from "proper-lockfile"; @@ -229,6 +238,33 @@ function reclaimStaleLease(directory: string): boolean { return true; } +/** + * Remove all session-lease directories whose owner process is no longer alive. + * Called at daemon startup to reclaim leases left behind by crashed or killed + * processes. Safe because a live process can always re-acquire a swept lease. + */ +export function sweepStaleSessionLeases(agentDir: string): number { + const root = join(agentDir, "session-leases"); + if (!existsSync(root)) return 0; + let swept = 0; + for (const entry of readdirSync(root)) { + if (!entry.endsWith(".lock")) continue; + const directory = join(root, entry); + const owner = readLeaseOwner(directory); + if (!owner) { + // Malformed or incomplete lease directory: reclaim it. + reclaimStaleLease(directory); + swept++; + continue; + } + if (!isLeaseOwnerAlive(owner)) { + reclaimStaleLease(directory); + swept++; + } + } + return swept; +} + export function acquireSessionLease( sessionPath: string | undefined, agentDir: string, diff --git a/packages/coding-agent/src/core/session-manager.ts b/packages/coding-agent/src/core/session-manager.ts index f9161168e5..bba9badb0f 100644 --- a/packages/coding-agent/src/core/session-manager.ts +++ b/packages/coding-agent/src/core/session-manager.ts @@ -1457,7 +1457,7 @@ export class SessionManager { if (!this.persist || !this.sessionFile) return; const hasAssistant = this.fileEntries.some((e) => e.type === "message" && e.message.role === "assistant"); - const shouldPersistWithoutAssistant = entry.type === "session_state" || entry.type === "session_info"; + const shouldPersistWithoutAssistant = entry.type === "session_info"; if (!hasAssistant && !shouldPersistWithoutAssistant) { // Mark as not flushed so when assistant arrives, all entries get written this.flushed = false; diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index dfebcdf61e..bb64a02230 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -30,6 +30,7 @@ import { getCronJobsPath, getDaemonLogPath, getDaemonUpdateRestartManifestPath, + getSessionsDir, VERSION, } from "../../config.js"; import { @@ -106,8 +107,13 @@ import { type IdleEvictionMinutes, type SessionPassivationSnapshot, } from "../../core/session-action-store.js"; -import { deleteSessionFile } from "../../core/session-file-actions.js"; -import { acquireSessionLease, canonicalSessionPath, type SessionLease } from "../../core/session-lease.js"; +import { deleteSessionFile, sweepGhostSessionFiles } from "../../core/session-file-actions.js"; +import { + acquireSessionLease, + canonicalSessionPath, + type SessionLease, + sweepStaleSessionLeases, +} from "../../core/session-lease.js"; import { readSessionInfo, resolveSessionRlmDepth, @@ -617,6 +623,7 @@ export class AgentDaemon { this.registerSignalHandlers(); this.summarizer.start(); + this.sweepStaleStartupState(); this.log(`Prime Agent daemon listening on ${this.socketPath}`); // No startup restore: on-disk sessions return only via --resume or the agents view. if (!this.shuttingDown) { @@ -625,6 +632,30 @@ export class AgentDaemon { this.startSupervisorMonitor(); } + /** + * Best-effort cleanup at daemon startup: remove stale session leases left by + * crashed processes and ghost session files (empty drafts that were never + * sent a message). These accumulate when a daemon exits without cleaning up. + */ + private sweepStaleStartupState(): void { + try { + const staleLeases = sweepStaleSessionLeases(this.agentDir); + if (staleLeases > 0) { + this.log(`swept ${staleLeases} stale session lease(s) at startup`); + } + } catch (error) { + this.log(`session lease sweep failed: ${error instanceof Error ? error.message : String(error)}`); + } + const sessionDir = this.options.defaultSessionConfig.sessionDir ?? getSessionsDir(this.agentDir); + void sweepGhostSessionFiles(sessionDir) + .then((count) => { + if (count > 0) this.log(`swept ${count} ghost session file(s) at startup`); + }) + .catch((error) => { + this.log(`ghost session sweep failed: ${error instanceof Error ? error.message : String(error)}`); + }); + } + private startSupervisorMonitor(): void { const supervisorSocketPath = process.env[DAEMON_WORKER_SUPERVISOR_SOCKET_ENV]; if (!this.options.worker || !supervisorSocketPath) { @@ -6035,10 +6066,11 @@ export class AgentDaemon { ? await this.closeChildSessions(state, reason, waitForAbort, descendants) : undefined; // Empty draft (no messages, config, or jobs): discard rather than persist an - // empty session file. Mirrors the detach-time discard so a config-bearing - // draft closed via kill/completed is never wiped. + // empty session file. A busy session is never discarded even if empty — the + // activity may produce content. Mirrors isDiscardableDraft so all close + // reasons (including shutdown/update) agree on what an abandoned draft is. const keepsResumeEntry = this.closeKeepsResumeEntry(reason); - const isEmptyDraftSession = !keepsResumeEntry && this.isEmptyDraftContent(state); + const isEmptyDraftSession = this.isEmptyDraftContent(state) && !isActiveSessionBusy(state); let persistError: unknown; // Clean shutdown leaves the session un-archived so it stays in the resume list. if (!keepsResumeEntry && !isEmptyDraftSession) { diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index e8e6aed07b..a21ef9463d 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -1287,6 +1287,8 @@ describe("daemon mode helpers", () => { session: { sessionId: "session-child", sessionFile: undefined, + isSessionActive: false, + hasRunningRlmChildren: () => false, abort: vi.fn(() => new Promise(() => {})), }, } as unknown as ActiveSessionState["runtime"]; @@ -4543,6 +4545,8 @@ describe("daemon mode helpers", () => { sessionId: "session-active", sessionFile: undefined, isBashRunning: false, + isSessionActive: false, + hasRunningRlmChildren: () => false, abort: vi.fn(async () => {}), sessionManager: { appendSessionState: vi.fn() }, }, @@ -9536,6 +9540,113 @@ describe("daemon mode helpers", () => { }), ).rejects.toThrow("Unknown active session: missing"); }); + + it("deletes an empty draft session file when closed via shutdown", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-ghost-shutdown-")); + try { + const sessionDir = join(tempDir, "sessions"); + const manager = SessionManager.create(tempDir, sessionDir); + manager.newSession(); + manager.appendSessionState({ status: "active" }); + manager.flushNow(); + const sessionFile = manager.getSessionFile(); + if (!sessionFile) throw new Error("Missing session file"); + + const createRuntime = vi.fn(async (options: Parameters[0]) => { + const session = makeRuntimeSession(options.sessionManager); + Object.assign(session, { + isStreaming: false, + isCompacting: false, + isSessionActive: false, + isBashRunning: false, + isRetrying: false, + unfinishedActionCount: 0, + hasRunningRlmChildren: () => false, + }); + return { + session, + extensionsResult: { extensions: [], errors: [], runtime: {} } as unknown as Awaited< + ReturnType + >["extensionsResult"], + services: { cwd: options.cwd, agentDir: options.agentDir } as Awaited< + ReturnType + >["services"], + diagnostics: [], + }; + }); + const daemon = new AgentDaemon(join(tempDir, "daemon.sock"), { + defaultSessionConfig: { agentDir: tempDir, cwd: tempDir, sessionDir }, + createRuntime, + }); + const internals = daemon as unknown as { + createRuntime(command: Extract): Promise; + closeSession(state: ActiveSessionState, reason: "shutdown"): Promise; + }; + + const state = await internals.createRuntime({ type: "create", sessionPath: sessionFile }); + expect(existsSync(sessionFile)).toBe(true); + + await internals.closeSession(state, "shutdown"); + + expect(existsSync(sessionFile)).toBe(false); + } finally { + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("preserves a session with user content when closed via shutdown", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-shutdown-content-")); + try { + const sessionDir = join(tempDir, "sessions"); + const manager = SessionManager.create(tempDir, sessionDir); + manager.newSession(); + manager.appendMessage({ role: "user", content: "hello world", timestamp: 1 }); + manager.appendSessionState({ status: "active" }); + manager.flushNow(); + const sessionFile = manager.getSessionFile(); + if (!sessionFile) throw new Error("Missing session file"); + + const createRuntime = vi.fn(async (options: Parameters[0]) => { + const session = makeRuntimeSession(options.sessionManager); + Object.assign(session, { + isStreaming: false, + isCompacting: false, + isSessionActive: false, + isBashRunning: false, + isRetrying: false, + unfinishedActionCount: 0, + hasRunningRlmChildren: () => false, + }); + return { + session, + extensionsResult: { extensions: [], errors: [], runtime: {} } as unknown as Awaited< + ReturnType + >["extensionsResult"], + services: { cwd: options.cwd, agentDir: options.agentDir } as Awaited< + ReturnType + >["services"], + diagnostics: [], + }; + }); + const daemon = new AgentDaemon(join(tempDir, "daemon.sock"), { + defaultSessionConfig: { agentDir: tempDir, cwd: tempDir, sessionDir }, + createRuntime, + }); + const internals = daemon as unknown as { + createRuntime(command: Extract): Promise; + closeSession(state: ActiveSessionState, reason: "shutdown"): Promise; + }; + + const state = await internals.createRuntime({ type: "create", sessionPath: sessionFile }); + expect(existsSync(sessionFile)).toBe(true); + + await internals.closeSession(state, "shutdown"); + + expect(existsSync(sessionFile)).toBe(true); + } finally { + rmSync(tempDir, { recursive: true, force: true }); + } + }); }); type CronAdmissionActivity = Partial<{ diff --git a/packages/coding-agent/test/session-artifacts-delete.test.ts b/packages/coding-agent/test/session-artifacts-delete.test.ts index 2f2a370c50..dcd688e912 100644 --- a/packages/coding-agent/test/session-artifacts-delete.test.ts +++ b/packages/coding-agent/test/session-artifacts-delete.test.ts @@ -2,7 +2,7 @@ import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node: import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { deleteSessionFile } from "../src/core/session-file-actions.js"; +import { deleteSessionFile, sweepGhostSessionFiles } from "../src/core/session-file-actions.js"; let root = ""; @@ -71,4 +71,43 @@ describe("deleteSessionFile removes the session artifact directory", () => { const result = await deleteSessionFile(sessionPath); expect(result.ok).toBe(true); }); + + it("sweepGhostSessionFiles removes empty draft sessions and preserves real ones", async () => { + const sessionsDir = join(root, "sessions"); + mkdirSync(sessionsDir, { recursive: true }); + + // Ghost: only bootstrap entries + session_state. + const ghostPath = join(sessionsDir, "ghost.jsonl"); + writeFileSync( + ghostPath, + `${[ + '{"type":"session","version":3,"id":"ghost"}', + '{"type":"model_change","id":"a","parentId":null}', + '{"type":"thinking_level_change","id":"b","parentId":"a"}', + '{"type":"service_tier_change","id":"c","parentId":"b"}', + '{"type":"session_state","id":"d","parentId":"c","state":{"status":"active"}}', + ].join("\n")}\n`, + ); + + // Real session: has a user message. + const realPath = join(sessionsDir, "real.jsonl"); + writeFileSync( + realPath, + `${[ + '{"type":"session","version":3,"id":"real"}', + '{"type":"model_change","id":"a","parentId":null}', + '{"type":"message","id":"b","parentId":"a","message":{"role":"user","content":"hi"}}', + ].join("\n")}\n`, + ); + + const swept = await sweepGhostSessionFiles(sessionsDir); + expect(swept).toBe(1); + expect(existsSync(ghostPath)).toBe(false); + expect(existsSync(realPath)).toBe(true); + }); + + it("sweepGhostSessionFiles is a no-op for a missing directory", async () => { + const swept = await sweepGhostSessionFiles(join(root, "does-not-exist")); + expect(swept).toBe(0); + }); }); diff --git a/packages/coding-agent/test/session-lease.test.ts b/packages/coding-agent/test/session-lease.test.ts index 87338b6866..24ef311cfd 100644 --- a/packages/coding-agent/test/session-lease.test.ts +++ b/packages/coding-agent/test/session-lease.test.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { lockSync } from "proper-lockfile"; @@ -11,6 +11,7 @@ import { SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV, SessionAlreadyActiveError, + sweepStaleSessionLeases, } from "../src/core/session-lease.js"; const tempDirs: string[] = []; @@ -186,4 +187,52 @@ describe("session leases", () => { const agentDir = createTempDir(); expect(acquireSessionLease(join(agentDir, "session.jsonl"), agentDir, {})).toBeUndefined(); }); + + it("sweeps stale leases at startup while keeping live ones", () => { + const agentDir = createTempDir(); + + // Stale lease: owner PID does not exist. + const stalePath = canonicalSessionPath(resolve(agentDir, "stale.jsonl")); + const staleKey = createHash("sha256").update(stalePath).digest("hex"); + const staleDir = join(agentDir, "session-leases", `${staleKey}.lock`); + mkdirSync(staleDir, { recursive: true }); + writeFileSync( + join(staleDir, "owner.json"), + JSON.stringify({ + version: 1, + token: "stale", + pid: 2_147_483_647, + activeSessionId: "dead-owner", + sessionPath: stalePath, + createdAt: new Date(0).toISOString(), + }), + ); + + // Live lease: owner is this process. + const livePath = canonicalSessionPath(resolve(agentDir, "live.jsonl")); + const liveKey = createHash("sha256").update(livePath).digest("hex"); + const liveDir = join(agentDir, "session-leases", `${liveKey}.lock`); + mkdirSync(liveDir, { recursive: true }); + writeFileSync( + join(liveDir, "owner.json"), + JSON.stringify({ + version: 1, + token: "live", + pid: process.pid, + activeSessionId: "this-process", + sessionPath: livePath, + createdAt: new Date(0).toISOString(), + }), + ); + + const swept = sweepStaleSessionLeases(agentDir); + expect(swept).toBe(1); + expect(existsSync(staleDir)).toBe(false); + expect(existsSync(liveDir)).toBe(true); + }); + + it("handles a missing session-leases directory gracefully", () => { + const agentDir = createTempDir(); + expect(sweepStaleSessionLeases(agentDir)).toBe(0); + }); }); diff --git a/packages/coding-agent/test/session-manager-flush.test.ts b/packages/coding-agent/test/session-manager-flush.test.ts index 418b52b2e4..eeeb9e49bd 100644 --- a/packages/coding-agent/test/session-manager-flush.test.ts +++ b/packages/coding-agent/test/session-manager-flush.test.ts @@ -437,3 +437,66 @@ describe("SessionManager.appendCustomMessageEntryWithRollback", () => { expect(readFileSync(file)).toEqual(before); }); }); + +describe("SessionManager _persist gate: session_state must not create ghost files", () => { + it("does not write a file when appendSessionState(active) is called before any assistant message", () => { + const dir = createTempDir(); + const sessionDir = join(dir, "sessions"); + const mgr = SessionManager.create(dir, sessionDir); + + // Daemon addRuntime() calls appendSessionState("active") immediately. + mgr.appendSessionState({ status: "active" }); + + const file = mgr.getSessionFile()!; + expect(existsSync(file)).toBe(false); + }); + + it("writes the file — including the session_state entry — once the first assistant message arrives", () => { + const dir = createTempDir(); + const sessionDir = join(dir, "sessions"); + const mgr = SessionManager.create(dir, sessionDir); + + // Simulate daemon addRuntime: state written in-memory but NOT to disk yet. + mgr.appendSessionState({ status: "active" }); + const file = mgr.getSessionFile()!; + expect(existsSync(file)).toBe(false); + + // User message alone is still not enough to trigger a write. + mgr.appendMessage({ role: "user", content: "hello", timestamp: Date.now() }); + expect(existsSync(file)).toBe(false); + + // First assistant message triggers _rewriteFile, which writes ALL + // accumulated entries at once — including the deferred session_state. + mgr.appendMessage({ + role: "assistant", + content: [{ type: "text", text: "hi" }], + api: "openai-completions", + provider: "openai", + model: "test", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }); + + expect(existsSync(file)).toBe(true); + + const entries = readFileSync(file, "utf8") + .trim() + .split("\n") + .map((line) => JSON.parse(line)); + + // Order: header, session_state, user message, assistant message. + expect(entries[0]!.type).toBe("session"); + expect(entries[1]!.type).toBe("session_state"); + expect(entries[1]!.state).toEqual({ status: "active" }); + expect(entries[2]!.message.role).toBe("user"); + expect(entries[3]!.message.role).toBe("assistant"); + }); +});