diff --git a/src/auto-reply/reply/agent-runner-execution.cli-persistence.test.ts b/src/auto-reply/reply/agent-runner-execution.cli-persistence.test.ts new file mode 100644 index 000000000..8ea61ca46 --- /dev/null +++ b/src/auto-reply/reply/agent-runner-execution.cli-persistence.test.ts @@ -0,0 +1,285 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { + appendAssistantMessageToTranscript, + appendMessageToTranscript, + appendMessageToTranscriptAsync, +} from "../../gateway/session-utils.fs.js"; + +describe("CLI backend transcript persistence", () => { + let tmpDir: string; + let storePath: string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "clawdbot-cli-persist-test-")); + storePath = path.join(tmpDir, "sessions.json"); + }); + + afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + test("user message persisted before CLI run", () => { + const sessionId = "test-user-persist"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + const result = appendMessageToTranscript({ + message: "Hello from user", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + expect(result.ok).toBe(true); + expect(fs.existsSync(transcriptPath)).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBeGreaterThanOrEqual(2); + + const msgLine = JSON.parse(lines[1]); + expect(msgLine.message.role).toBe("user"); + expect(msgLine.message.content[0].text).toBe("Hello from user"); + }); + + test("assistant response persisted after successful CLI run with provider/model", () => { + const sessionId = "test-assistant-persist"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + // First persist user message + appendMessageToTranscript({ + message: "User question", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + // Then persist assistant response with provider/model + const result = appendAssistantMessageToTranscript({ + message: "Assistant response", + sessionId, + storePath, + provider: "claude-cli", + model: "claude-3-opus", + }); + + expect(result.ok).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBe(3); + + const msgLine = JSON.parse(lines[2]); + expect(msgLine.message.role).toBe("assistant"); + expect(msgLine.message.content[0].text).toBe("Assistant response"); + expect(msgLine.message.provider).toBe("claude-cli"); + expect(msgLine.message.model).toBe("claude-3-opus"); + }); + + test("error response persisted when CLI fails", () => { + const sessionId = "test-error-persist"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + // First persist user message + appendMessageToTranscript({ + message: "User question", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + // Simulate error being persisted + const errorResult = appendAssistantMessageToTranscript({ + message: "[CLI error: Connection timeout]", + sessionId, + storePath, + provider: "claude-cli", + model: "claude-3-opus", + }); + + expect(errorResult.ok).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBe(3); + + const msgLine = JSON.parse(lines[2]); + expect(msgLine.message.role).toBe("assistant"); + expect(msgLine.message.content[0].text).toContain("[CLI error:"); + }); + + test("persistence failure returns error but does not throw", () => { + const sessionId = "test-persist-fail"; + // Use invalid path that cannot be created + const invalidStorePath = "/nonexistent/deeply/nested/sessions.json"; + + const result = appendMessageToTranscript({ + message: "Test message", + role: "user", + sessionId, + storePath: invalidStorePath, + createIfMissing: true, + }); + + expect(result.ok).toBe(false); + expect(result.error).toBeDefined(); + }); + + test("empty CLI response still recorded to maintain turn consistency", () => { + const sessionId = "test-empty-response"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + // First persist user message + appendMessageToTranscript({ + message: "User question", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + // Persist empty response + const result = appendAssistantMessageToTranscript({ + message: "", + sessionId, + storePath, + provider: "claude-cli", + model: "claude-3-opus", + }); + + expect(result.ok).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBe(3); + + const msgLine = JSON.parse(lines[2]); + expect(msgLine.message.role).toBe("assistant"); + expect(msgLine.message.content[0].text).toBe(""); + }); + + test("persistence preserves role alternation (user -> assistant)", () => { + const sessionId = "test-role-alternation"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + // Simulate multiple turns + appendMessageToTranscript({ + message: "First user message", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + appendAssistantMessageToTranscript({ + message: "First assistant response", + sessionId, + storePath, + }); + + appendMessageToTranscript({ + message: "Second user message", + role: "user", + sessionId, + storePath, + }); + + appendAssistantMessageToTranscript({ + message: "Second assistant response", + sessionId, + storePath, + }); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBe(5); // header + 4 messages + + const roles = lines.slice(1).map((line) => JSON.parse(line).message.role); + expect(roles).toEqual(["user", "assistant", "user", "assistant"]); + }); + + test("partial failure: user persists but assistant fails leaves orphaned message", () => { + const sessionId = "test-partial-failure"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + // User message persists successfully + const userResult = appendMessageToTranscript({ + message: "User question", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + expect(userResult.ok).toBe(true); + + // Make file read-only to simulate assistant persistence failure + fs.chmodSync(transcriptPath, 0o444); + + try { + // Assistant message should fail + const assistantResult = appendAssistantMessageToTranscript({ + message: "Response", + sessionId, + storePath, + }); + + expect(assistantResult.ok).toBe(false); + expect(assistantResult.error).toBeDefined(); + + // Verify orphaned user message exists + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBe(2); // header + user message only (orphaned) + + const roles = lines.slice(1).map((line) => JSON.parse(line).message.role); + expect(roles).toEqual(["user"]); // No assistant response + } finally { + // Restore permissions for cleanup + fs.chmodSync(transcriptPath, 0o644); + } + }); + + test("concurrent writes with async version maintain transcript integrity", async () => { + const sessionId = "test-concurrent-async"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + // Create initial file + appendMessageToTranscript({ + message: "Initial", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + // Launch 10 concurrent writes using async version with locking + const writes = Array.from({ length: 10 }, (_, i) => + appendMessageToTranscriptAsync({ + message: `Concurrent message ${i}`, + role: i % 2 === 0 ? "user" : "assistant", + sessionId, + storePath, + }), + ); + + const results = await Promise.all(writes); + + // All writes should succeed + expect(results.every((r) => r.ok)).toBe(true); + + // Verify all messages are present and valid JSON + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines.length).toBe(12); // header + initial + 10 concurrent + + // All lines should be valid JSON + for (const line of lines) { + expect(() => JSON.parse(line)).not.toThrow(); + } + }); +}); diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index 47c45b09d..fa2ba2438 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/src/auto-reply/reply/agent-runner-execution.ts @@ -19,6 +19,10 @@ import { type SessionEntry, updateSessionStore, } from "../../config/sessions.js"; +import { + appendAssistantMessageToTranscript, + appendMessageToTranscript, +} from "../../gateway/session-utils.fs.js"; import { logVerbose } from "../../globals.js"; import { emitAgentEvent, registerAgentRunContext } from "../../infra/agent-events.js"; import { defaultRuntime } from "../../runtime.js"; @@ -160,6 +164,21 @@ export async function runAgentTurnWithFallback(params: { startedAt, }, }); + // CLI backends bypass SessionManager, so persist user message to transcript + // directly to ensure it's available on TUI reload. + const userPersistResult = appendMessageToTranscript({ + message: params.commandBody, + role: "user", + sessionId: params.followupRun.run.sessionId, + sessionFile: params.followupRun.run.sessionFile, + storePath: params.storePath, + createIfMissing: true, + }); + if (!userPersistResult.ok) { + logVerbose( + `CLI transcript: failed to persist user message: ${userPersistResult.error}`, + ); + } const cliSessionId = getCliSessionId(params.getActiveSessionEntry(), provider); return runCliAgent({ sessionId: params.followupRun.run.sessionId, @@ -183,12 +202,28 @@ export async function runAgentTurnWithFallback(params: { // emit one with the final text so server-chat can populate its buffer // and send the response to TUI/WebSocket clients. const cliText = result.payloads?.[0]?.text?.trim(); - if (cliText) { - emitAgentEvent({ - runId, - stream: "assistant", - data: { text: cliText }, - }); + // Always emit assistant event (even if empty) so lifecycle tracking is consistent + emitAgentEvent({ + runId, + stream: "assistant", + data: { text: cliText || "" }, + }); + // CLI backends bypass SessionManager, so persist response to transcript + // directly to ensure it's available on TUI reload. + // Record even empty responses to maintain message turn consistency. + const assistantPersistResult = appendAssistantMessageToTranscript({ + message: cliText || "", + sessionId: params.followupRun.run.sessionId, + sessionFile: params.followupRun.run.sessionFile, + storePath: params.storePath, + createIfMissing: true, + provider, + model, + }); + if (!assistantPersistResult.ok) { + logVerbose( + `CLI transcript: failed to persist assistant message: ${assistantPersistResult.error}`, + ); } emitAgentEvent({ runId, @@ -202,6 +237,22 @@ export async function runAgentTurnWithFallback(params: { return result; }) .catch((err) => { + const errorMessage = err instanceof Error ? err.message : String(err); + // Persist error as assistant response to avoid orphaned user message + const errorPersistResult = appendAssistantMessageToTranscript({ + message: `[CLI error: ${errorMessage}]`, + sessionId: params.followupRun.run.sessionId, + sessionFile: params.followupRun.run.sessionFile, + storePath: params.storePath, + createIfMissing: true, + provider, + model, + }); + if (!errorPersistResult.ok) { + logVerbose( + `CLI transcript: failed to persist error message: ${errorPersistResult.error}`, + ); + } emitAgentEvent({ runId, stream: "lifecycle", @@ -209,7 +260,7 @@ export async function runAgentTurnWithFallback(params: { phase: "error", startedAt, endedAt: Date.now(), - error: err instanceof Error ? err.message : String(err), + error: errorMessage, }, }); throw err; diff --git a/src/gateway/server-methods/chat.ts b/src/gateway/server-methods/chat.ts index 50f441779..2e6fcfc7c 100644 --- a/src/gateway/server-methods/chat.ts +++ b/src/gateway/server-methods/chat.ts @@ -2,7 +2,6 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; import path from "node:path"; -import { CURRENT_SESSION_VERSION } from "@mariozechner/pi-coding-agent"; import { resolveSessionAgentId } from "../../agents/agent-scope.js"; import { resolveEffectiveMessagesConfig, resolveIdentityName } from "../../agents/identity.js"; import { resolveThinkingDefault } from "../../agents/model-selection.js"; @@ -39,6 +38,7 @@ import { readSessionMessages, resolveSessionModelRef, } from "../session-utils.js"; +import { ensureTranscriptFile } from "../session-utils.fs.js"; import { stripEnvelopeFromMessages } from "../chat-sanitize.js"; import { formatForLog } from "../ws-log.js"; import type { GatewayRequestContext, GatewayRequestHandlers } from "./types.js"; @@ -61,27 +61,6 @@ function resolveTranscriptPath(params: { return path.join(path.dirname(storePath), `${sessionId}.jsonl`); } -function ensureTranscriptFile(params: { transcriptPath: string; sessionId: string }): { - ok: boolean; - error?: string; -} { - if (fs.existsSync(params.transcriptPath)) return { ok: true }; - try { - fs.mkdirSync(path.dirname(params.transcriptPath), { recursive: true }); - const header = { - type: "session", - version: CURRENT_SESSION_VERSION, - id: params.sessionId, - timestamp: new Date().toISOString(), - cwd: process.cwd(), - }; - fs.writeFileSync(params.transcriptPath, `${JSON.stringify(header)}\n`, "utf-8"); - return { ok: true }; - } catch (err) { - return { ok: false, error: err instanceof Error ? err.message : String(err) }; - } -} - function appendAssistantTranscriptMessage(params: { message: string; label?: string; diff --git a/src/gateway/session-utils.fs.test.ts b/src/gateway/session-utils.fs.test.ts index a85dc9f69..c81f9619a 100644 --- a/src/gateway/session-utils.fs.test.ts +++ b/src/gateway/session-utils.fs.test.ts @@ -3,6 +3,8 @@ import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { + appendMessageToTranscript, + ensureTranscriptFile, readFirstUserMessageFromTranscript, readLastMessagePreviewFromTranscript, readSessionPreviewItemsFromTranscript, @@ -404,3 +406,233 @@ describe("readSessionPreviewItemsFromTranscript", () => { expect(result[0]?.text.endsWith("...")).toBe(true); }); }); + +describe("appendMessageToTranscript", () => { + let tmpDir: string; + let storePath: string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "clawdbot-append-msg-test-")); + storePath = path.join(tmpDir, "sessions.json"); + }); + + afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + test("appends user message to existing transcript", () => { + const sessionId = "test-append-user"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + const header = JSON.stringify({ type: "session", version: 1, id: sessionId }); + fs.writeFileSync(transcriptPath, header + "\n", "utf-8"); + + const result = appendMessageToTranscript({ + message: "Hello from user", + role: "user", + sessionId, + storePath, + }); + + expect(result.ok).toBe(true); + expect(result.messageId).toBeDefined(); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines).toHaveLength(2); + const msg = JSON.parse(lines[1]); + expect(msg.message.role).toBe("user"); + expect(msg.message.content[0].text).toBe("Hello from user"); + expect(msg.message.stopReason).toBeUndefined(); + }); + + test("appends assistant message with stopReason", () => { + const sessionId = "test-append-assistant"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + const header = JSON.stringify({ type: "session", version: 1, id: sessionId }); + fs.writeFileSync(transcriptPath, header + "\n", "utf-8"); + + const result = appendMessageToTranscript({ + message: "Hello from assistant", + role: "assistant", + sessionId, + storePath, + }); + + expect(result.ok).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + const msg = JSON.parse(lines[1]); + expect(msg.message.role).toBe("assistant"); + expect(msg.message.stopReason).toBe("cli_backend"); + expect(msg.message.usage).toBeDefined(); + }); + + test("creates transcript file when createIfMissing is true", () => { + const sessionId = "test-append-create"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + expect(fs.existsSync(transcriptPath)).toBe(false); + + const result = appendMessageToTranscript({ + message: "First message", + role: "user", + sessionId, + storePath, + createIfMissing: true, + }); + + expect(result.ok).toBe(true); + expect(fs.existsSync(transcriptPath)).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + expect(lines).toHaveLength(2); + const header = JSON.parse(lines[0]); + expect(header.type).toBe("session"); + expect(header.id).toBe(sessionId); + }); + + test("fails when transcript does not exist and createIfMissing is false", () => { + const sessionId = "test-append-no-create"; + + const result = appendMessageToTranscript({ + message: "Message", + role: "user", + sessionId, + storePath, + createIfMissing: false, + }); + + expect(result.ok).toBe(false); + expect(result.error).toContain("not found"); + }); + + test("prefers sessionFile over storePath", () => { + const sessionId = "test-append-sessionfile"; + const customPath = path.join(tmpDir, "custom-transcript.jsonl"); + const header = JSON.stringify({ type: "session", version: 1, id: sessionId }); + fs.writeFileSync(customPath, header + "\n", "utf-8"); + + const result = appendMessageToTranscript({ + message: "Custom file message", + role: "user", + sessionId, + sessionFile: customPath, + storePath, + }); + + expect(result.ok).toBe(true); + + const content = fs.readFileSync(customPath, "utf-8"); + expect(content).toContain("Custom file message"); + }); + + test("includes provider and model in assistant message metadata", () => { + const sessionId = "test-append-metadata"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + const header = JSON.stringify({ type: "session", version: 1, id: sessionId }); + fs.writeFileSync(transcriptPath, header + "\n", "utf-8"); + + const result = appendMessageToTranscript({ + message: "Response with metadata", + role: "assistant", + sessionId, + storePath, + provider: "claude-cli", + model: "claude-3-opus", + }); + + expect(result.ok).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + const msg = JSON.parse(lines[1]); + expect(msg.message.provider).toBe("claude-cli"); + expect(msg.message.model).toBe("claude-3-opus"); + }); + + test("omits provider/model when not provided", () => { + const sessionId = "test-append-no-metadata"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + const header = JSON.stringify({ type: "session", version: 1, id: sessionId }); + fs.writeFileSync(transcriptPath, header + "\n", "utf-8"); + + const result = appendMessageToTranscript({ + message: "Response without metadata", + role: "assistant", + sessionId, + storePath, + }); + + expect(result.ok).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const lines = content.trim().split("\n"); + const msg = JSON.parse(lines[1]); + expect(msg.message.provider).toBeUndefined(); + expect(msg.message.model).toBeUndefined(); + }); +}); + +describe("ensureTranscriptFile", () => { + let tmpDir: string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "clawdbot-ensure-transcript-test-")); + }); + + afterEach(() => { + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + test("creates transcript file with valid header", () => { + const sessionId = "test-ensure-create"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + + const result = ensureTranscriptFile({ transcriptPath, sessionId }); + + expect(result.ok).toBe(true); + expect(fs.existsSync(transcriptPath)).toBe(true); + + const content = fs.readFileSync(transcriptPath, "utf-8"); + const header = JSON.parse(content.trim()); + expect(header.type).toBe("session"); + expect(header.id).toBe(sessionId); + expect(header.timestamp).toBeDefined(); + }); + + test("returns ok when file already exists", () => { + const sessionId = "test-ensure-exists"; + const transcriptPath = path.join(tmpDir, `${sessionId}.jsonl`); + fs.writeFileSync(transcriptPath, "existing content\n", "utf-8"); + + const result = ensureTranscriptFile({ transcriptPath, sessionId }); + + expect(result.ok).toBe(true); + // Content should not be modified + const content = fs.readFileSync(transcriptPath, "utf-8"); + expect(content).toBe("existing content\n"); + }); + + test("creates nested directories if needed", () => { + const sessionId = "test-ensure-nested"; + const transcriptPath = path.join(tmpDir, "nested", "deeply", `${sessionId}.jsonl`); + + const result = ensureTranscriptFile({ transcriptPath, sessionId }); + + expect(result.ok).toBe(true); + expect(fs.existsSync(transcriptPath)).toBe(true); + }); + + test("returns error for invalid path", () => { + const sessionId = "test-ensure-invalid"; + // Path that cannot be created (null byte in path) + const transcriptPath = "/dev/null/\0/invalid.jsonl"; + + const result = ensureTranscriptFile({ transcriptPath, sessionId }); + + expect(result.ok).toBe(false); + expect(result.error).toBeDefined(); + }); +}); diff --git a/src/gateway/session-utils.fs.ts b/src/gateway/session-utils.fs.ts index d6453ace6..32af4e0a4 100644 --- a/src/gateway/session-utils.fs.ts +++ b/src/gateway/session-utils.fs.ts @@ -1,11 +1,253 @@ +import crypto from "node:crypto"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import { CURRENT_SESSION_VERSION } from "@mariozechner/pi-coding-agent"; import { resolveSessionTranscriptPath } from "../config/sessions.js"; import { stripEnvelope } from "./chat-sanitize.js"; import type { SessionPreviewItem } from "./session-utils.types.js"; +export type TranscriptAppendResult = { + ok: boolean; + messageId?: string; + error?: string; +}; + +/** + * In-process lock for session transcript writes. + * Ensures atomic user+assistant message pairs within the same Node process. + * Similar protection to what SessionManager provides for Pi agents. + */ +const sessionWriteLocks = new Map>(); + +async function withSessionLock(sessionId: string, fn: () => T | Promise): Promise { + // Wait for any pending write to this session + const pending = sessionWriteLocks.get(sessionId); + if (pending) { + await pending.catch(() => {}); // Ignore errors from previous writes + } + + // Create a new lock for this write + let resolve: () => void; + const lock = new Promise((r) => { + resolve = r; + }); + sessionWriteLocks.set(sessionId, lock); + + try { + return await fn(); + } finally { + resolve!(); + // Clean up if this is still our lock + if (sessionWriteLocks.get(sessionId) === lock) { + sessionWriteLocks.delete(sessionId); + } + } +} + +/** + * Ensures a transcript file exists with a valid session header. + * Creates the file and parent directories if needed. + * Uses atomic file creation (flag: 'wx') to prevent TOCTOU race conditions. + * Exported for reuse by other modules (e.g., server-methods/chat.ts). + */ +export function ensureTranscriptFile(params: { transcriptPath: string; sessionId: string }): { + ok: boolean; + error?: string; +} { + try { + fs.mkdirSync(path.dirname(params.transcriptPath), { recursive: true }); + const header = { + type: "session", + version: CURRENT_SESSION_VERSION, + id: params.sessionId, + timestamp: new Date().toISOString(), + cwd: process.cwd(), + }; + // Use 'wx' flag for atomic creation - fails if file exists (prevents TOCTOU race) + fs.writeFileSync(params.transcriptPath, `${JSON.stringify(header)}\n`, { + encoding: "utf-8", + flag: "wx", + }); + return { ok: true }; + } catch (err) { + // EEXIST means file was created by another process - that's fine + if ((err as NodeJS.ErrnoException).code === "EEXIST") { + return { ok: true }; + } + return { ok: false, error: err instanceof Error ? err.message : String(err) }; + } +} + +/** + * Resolves transcript path from params. + * @param params.sessionFile - Direct path to transcript file (preferred) + * @param params.storePath - Fallback: derives transcript path from store location + * @returns Transcript path or null if neither sessionFile nor storePath provided + */ +function resolveTranscriptPathFromParams(params: { + sessionId: string; + sessionFile?: string; + storePath?: string; +}): string | null { + if (params.sessionFile) return params.sessionFile; + if (params.storePath) { + return path.join(path.dirname(params.storePath), `${params.sessionId}.jsonl`); + } + return null; +} + +/** + * Internal implementation of message append (no locking). + */ +function appendMessageToTranscriptImpl( + params: { + message: string; + role: "user" | "assistant"; + sessionId: string; + sessionFile?: string; + storePath?: string; + createIfMissing?: boolean; + stopReason?: string; + provider?: string; + model?: string; + }, + transcriptPath: string, +): TranscriptAppendResult { + if (!fs.existsSync(transcriptPath)) { + if (!params.createIfMissing) { + return { ok: false, error: "transcript file not found" }; + } + const ensured = ensureTranscriptFile({ + transcriptPath, + sessionId: params.sessionId, + }); + if (!ensured.ok) { + return { ok: false, error: ensured.error ?? "failed to create transcript file" }; + } + } + + const now = Date.now(); + const messageId = crypto.randomUUID().slice(0, 8); + const messageBody: Record = { + role: params.role, + content: [{ type: "text", text: params.message }], + timestamp: now, + }; + if (params.role === "assistant") { + messageBody.stopReason = params.stopReason ?? "cli_backend"; + messageBody.usage = { input: 0, output: 0, totalTokens: 0 }; + if (params.provider) messageBody.provider = params.provider; + if (params.model) messageBody.model = params.model; + } + const transcriptEntry = { + type: "message", + id: messageId, + timestamp: new Date(now).toISOString(), + message: messageBody, + }; + + try { + fs.appendFileSync(transcriptPath, `${JSON.stringify(transcriptEntry)}\n`, "utf-8"); + } catch (err) { + return { ok: false, error: err instanceof Error ? err.message : String(err) }; + } + + return { ok: true, messageId }; +} + +/** + * Appends a message to a session transcript file. + * Used for CLI backend responses which bypass SessionManager. + * + * Note: This is a synchronous function. For concurrent-safe writes + * (e.g., multiple CLI runs on same session), use appendMessageToTranscriptAsync(). + */ +export function appendMessageToTranscript(params: { + message: string; + role: "user" | "assistant"; + sessionId: string; + sessionFile?: string; + storePath?: string; + createIfMissing?: boolean; + stopReason?: string; + /** Provider that generated the response (for assistant messages). */ + provider?: string; + /** Model that generated the response (for assistant messages). */ + model?: string; +}): TranscriptAppendResult { + const transcriptPath = resolveTranscriptPathFromParams(params); + if (!transcriptPath) { + return { ok: false, error: "transcript path not resolved" }; + } + return appendMessageToTranscriptImpl(params, transcriptPath); +} + +/** + * Async version of appendMessageToTranscript with session-level locking. + * Ensures atomic writes when multiple concurrent writes target the same session. + * Provides similar protection to what SessionManager offers for Pi agents. + */ +export async function appendMessageToTranscriptAsync(params: { + message: string; + role: "user" | "assistant"; + sessionId: string; + sessionFile?: string; + storePath?: string; + createIfMissing?: boolean; + stopReason?: string; + provider?: string; + model?: string; +}): Promise { + const transcriptPath = resolveTranscriptPathFromParams(params); + if (!transcriptPath) { + return { ok: false, error: "transcript path not resolved" }; + } + return withSessionLock(params.sessionId, () => + appendMessageToTranscriptImpl(params, transcriptPath), + ); +} + +/** + * Appends an assistant message to a session transcript file. + * Used for CLI backend responses which bypass SessionManager. + */ +export function appendAssistantMessageToTranscript(params: { + message: string; + sessionId: string; + sessionFile?: string; + storePath?: string; + createIfMissing?: boolean; + /** Provider that generated the response. */ + provider?: string; + /** Model that generated the response. */ + model?: string; +}): TranscriptAppendResult { + return appendMessageToTranscript({ + ...params, + role: "assistant", + }); +} + +/** + * Async version with session-level locking for concurrent-safe writes. + */ +export async function appendAssistantMessageToTranscriptAsync(params: { + message: string; + sessionId: string; + sessionFile?: string; + storePath?: string; + createIfMissing?: boolean; + provider?: string; + model?: string; +}): Promise { + return appendMessageToTranscriptAsync({ + ...params, + role: "assistant", + }); +} + export function readSessionMessages( sessionId: string, storePath: string | undefined,