第 10 讲内置工具集与会话持久化（JSONL 树、分支、Fork）

// 截断发生时，给模型一条可行动的续读提示（节选自 execute 内部）
if (truncation.truncated) {
    const endLineDisplay = startLineDisplay + truncation.outputLines - 1;
    const nextOffset = endLineDisplay + 1;
    outputText = truncation.content;
    if (truncation.truncatedBy === "lines") {
        outputText += `\n\n[Showing lines ${startLineDisplay}-${endLineDisplay} of ${totalFileLines}. Use offset=${nextOffset} to continue.]`;
    } else {
        outputText += `\n\n[Showing lines ${startLineDisplay}-${endLineDisplay} of ${totalFileLines} (${formatSize(DEFAULT_MAX_BYTES)} limit). Use offset=${nextOffset} to continue.]`;
    }
    details = { truncation };
}

const absolutePath = resolveToCwd(path, ctx?.cwd || cwd);
const dir = dirname(absolutePath);
return withFileMutationQueue(absolutePath, async () => {
    // Do not reject from an abort event listener here: that would release the
    // mutation queue while an in-flight filesystem operation may still finish.
    const throwIfAborted = (): void => {
        if (signal?.aborted) throw new Error("Operation aborted");
    };

    throwIfAborted();
    await ops.mkdir(dir);      // Create parent directories if needed.
    throwIfAborted();

    await ops.writeFile(absolutePath, content);
    throwIfAborted();

    return { content: [{ type: "text", text: `Successfully wrote to ${path}` }], details: undefined };
});

// Strip BOM before matching. The model will not include an invisible BOM in oldText.
const { bom, text: content } = splitBom(rawContent);
const normalizedContent = normalizeToLF(content);
const { baseContent, newContent, replacements } =
    applyEditsToNormalizedContent(normalizedContent, edits, path);
// Diff before writing, so diff computation cannot fail after the write.
const editDiff = generateEditDiff(path, baseContent, newContent, replacements);
throwIfAborted();

// Splice into the original text, so untouched lines keep their line endings.
const finalContent =
    bom + (content === normalizedContent
        ? newContent
        : applyReplacementsToOriginal(content, replacements));
await ops.writeFile(absolutePath, finalContent);

/** Output limit of `structuredContent.output`, which programmatic callers such as codemode scripts receive. */
const STRUCTURED_OUTPUT_MAX_BYTES = 1024 * 1024;  // 1MiB

const bashOutputSchema = Type.Object({
    output: Type.String({ description: "Combined stdout and stderr, possibly truncated" }),
    truncated: Type.Boolean(),
    full_output_path: Type.Optional(Type.String({ description: "Full output, when truncated" })),
    exit_code: Type.Number(),
    wall_time_seconds: Type.Number(),
});

export const DEFAULT_MAX_LINES = 2000;
export const DEFAULT_MAX_BYTES = 50 * 1024; // 50KB
export const GREP_MAX_LINE_LENGTH = 500; // Max chars per grep match line

export interface TruncationResult {
    content: string;        // 截断后的内容
    truncated: boolean;
    truncatedBy: "lines" | "bytes" | null;
    totalLines: number;       // 原始总行数
    totalBytes: number;      // 原始总字节
    outputLines: number;
    outputBytes: number;
    lastLinePartial: boolean;  // 仅 bash 尾截断可能出现半行
    firstLineExceedsLimit: boolean;
    maxLines: number;
    maxBytes: number;
}

append(data: Buffer): void {
    this.totalRawBytes += data.length;
    this.appendDecodedText(this.decoder.decode(data, { stream: true }));

    if (this.tempFileStream || this.shouldUseTempFile()) {
        this.ensureTempFile();
        this.tempFileStream?.write(data);
    } else if (data.length > 0) {
        this.rawChunks.push(data);
    }
}

private shouldUseTempFile(): boolean {
    return (
        this.totalRawBytes > this.maxBytes ||
        this.totalDecodedBytes > this.maxBytes ||
        this.totalLines > this.maxLines
    );
}

const fileMutationQueues = new Map<string, Promise<void>>();
let registrationQueue = Promise.resolve();

async function getMutationQueueKey(filePath: string): Promise<string> {
    const resolvedPath = resolve(filePath);
    try {
        return await realpath(resolvedPath);   // 符号链接归一，同一目标同一把锁
    } catch (error) {
        if (isMissingPathError(error)) return resolvedPath;  // 新文件尚不存在
        throw error;
    }
}

export async function withFileMutationQueue<T>(filePath: string, fn: () => Promise<T>): Promise<T> {
    const registration = registrationQueue.then(async () => {
        const key = await getMutationQueueKey(filePath);
        const currentQueue = fileMutationQueues.get(key) ?? Promise.resolve();
        let releaseNext!: () => void;
        const nextQueue = new Promise<void>((resolveQueue) => { releaseNext = resolveQueue; });
        const chainedQueue = currentQueue.then(() => nextQueue);
        fileMutationQueues.set(key, chainedQueue);
        return { key, currentQueue, chainedQueue, releaseNext };
    });
    registrationQueue = registration.then(() => undefined, () => undefined);

    const { key, currentQueue, chainedQueue, releaseNext } = await registration;
    await currentQueue;          // 等前面同文件的写完成
    try {
        return await fn();
    } finally {
        releaseNext();           // 唤醒下一个排队者
        if (fileMutationQueues.get(key) === chainedQueue) {
            fileMutationQueues.delete(key);  // 队列清空后回收 Map 项
        }
    }
}

export const CURRENT_SESSION_VERSION = 3;

export interface SessionHeader {
    type: "session";
    version?: number;   // v1 sessions don't have this
    id: string;          // 会话 ID（UUIDv7）
    timestamp: string;
    cwd: string;          // 会话绑定的工作目录，创建后不变
    parentSession?: string; // /fork、/clone 创建时指向源文件
}

export interface SessionEntryBase {
    type: string;
    id: string;              // Usually an 8-char hex ID
    parentId: string | null;  // Parent entry ID (null for a root entry)
    timestamp: string;        // ISO timestamp
}

// Handle session persistence
if (event.type === "message_end") {
    let entryId: string | undefined;
    if (event.message.role === "custom") {
        // Persist as CustomMessageEntry
        entryId = this.sessionManager.appendCustomMessageEntry(
            event.message.customType, event.message.content,
            event.message.display, event.message.details,
        );
    } else if (
        event.message.role === "system" || event.message.role === "user" ||
        event.message.role === "assistant" || event.message.role === "toolResult"
    ) {
        entryId = this.sessionManager.appendMessage(event.message);
    }
    if (entryId) this._entryIdsByMessage.set(event.message, entryId);
    // Other message types (bashExecution, compactionSummary, branchSummary)
    // are persisted elsewhere
}

// ~/.pi/agent/sessions/--<path>--/<timestamp>_<session-id>.jsonl
function getDefaultSessionDirPath(cwd: string, agentDir = getDefaultAgentDir()): string {
    const resolvedCwd = resolvePath(cwd);
    const resolvedAgentDir = resolvePath(agentDir);
    const safePath = `--${resolvedCwd.replace(/^[/\\]/, "").replace(/[/\\:]/g, "-")}--`;
    return join(resolvedAgentDir, "sessions", safePath);
}

export function convertToLlm(messages: AgentMessage[]): Message[] {
    return messages
        .map((m): Message | undefined => {
            switch (m.role) {
                case "bashExecution":
                    if (m.excludeFromContext) return undefined;  // “!!” 前缀：不出现在上下文
                    return { role: "user",
                        content: [{ type: "text", text: bashExecutionToText(m) }],
                        timestamp: m.timestamp };
                case "custom":
                    return { role: "user", content: ..., timestamp: m.timestamp };
                case "branchSummary":
                    return { role: "user", content: BRANCH_SUMMARY_PREFIX + m.summary + ..., ... };
                case "compactionSummary":
                    return { role: "user", content: COMPACTION_SUMMARY_PREFIX + m.summary + ..., ... };
                case "system": case "user":
                case "assistant": case "toolResult":
                    return m;               // 原样通过
                default:
                    return undefined;         // 未知角色直接剔除
            }
        })
        .filter((m) => m !== undefined);
}