fix: harden CLI transcript writes against concurrency
- TOCTOU fix: use { flag: 'wx' } for atomic file creation
- Add session-level locking for concurrent-safe writes
- Add async API: appendMessageToTranscriptAsync, appendAssistantMessageToTranscriptAsync
- Add partial failure test (orphaned user message)
- Add concurrent write test
This commit is contained in:
Ross Morsali
Ross
parent
c1b98ffbbe
commit
7a304917d2
@ -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();
|
||||
}
|
||||
});
|
||||
});
|
||||
@ -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;
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
@ -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<string, Promise<void>>();
|
||||
|
||||
async function withSessionLock<T>(sessionId: string, fn: () => T | Promise<T>): Promise<T> {
|
||||
// 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<void>((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<string, unknown> = {
|
||||
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<TranscriptAppendResult> {
|
||||
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<TranscriptAppendResult> {
|
||||
return appendMessageToTranscriptAsync({
|
||||
...params,
|
||||
role: "assistant",
|
||||
});
|
||||
}
|
||||
|
||||
export function readSessionMessages(
|
||||
sessionId: string,
|
||||
storePath: string | undefined,
|
||||
|
||||
Loading…
Reference in New Issue
Block a user