pikasTech-unidesk/src/components/microservices/code-queue/src/task-output.ts

// 重构前 index.ts 只读参考：commit 6a04144d3f5103014f75b637d7e6bc2f45bf007f，blob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58；可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。

import { appendFileSync, existsSync, mkdirSync, readFileSync, statSync } from "node:fs";
import { resolve } from "node:path";
import type { ArchivedLiveOutput, JsonValue, LiveOutput, OutputChannel, QueueTask, RuntimeConfig } from "./types";

export interface TaskOutputContext {
  config: Pick<RuntimeConfig, "maxInMemoryOutputRecords" | "outputArchiveDir">;
  allocateSeq: () => number;
  errorToJson: (error: unknown) => JsonValue;
  logger: (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue) => void;
  markTaskDirty: (taskId: string) => void;
  nowIso: () => string;
  onOutputAppended?: (task: QueueTask, output: LiveOutput, op: ArchivedLiveOutput["op"]) => void;
  schedulePersistState: () => void;
}

const outputArchiveSeededTasks = new Set<string>();
const archivedOutputCache = new Map<string, { signature: string; output: LiveOutput[] }>();
let context: TaskOutputContext | null = null;

export function configureTaskOutput(runtimeContext: TaskOutputContext): void {
  context = runtimeContext;
}

function ctx(): TaskOutputContext {
  if (context === null) throw new Error("task-output module is not configured");
  return context;
}

function taskOutputArchivePath(taskId: string): string {
  return resolve(ctx().config.outputArchiveDir, `${taskId}.jsonl`);
}

function serializeArchivedOutput(output: LiveOutput, op: ArchivedLiveOutput["op"], text: string): string {
  const record: ArchivedLiveOutput = {
    seq: output.seq,
    at: output.at,
    channel: output.channel,
    text,
    ...(output.method === undefined ? {} : { method: output.method }),
    ...(output.itemId === undefined ? {} : { itemId: output.itemId }),
    op,
  };
  return `${JSON.stringify(record)}\n`;
}

function ensureTaskOutputArchiveSeeded(task: QueueTask): void {
  if (outputArchiveSeededTasks.has(task.id)) return;
  mkdirSync(ctx().config.outputArchiveDir, { recursive: true });
  const path = taskOutputArchivePath(task.id);
  if (!existsSync(path) && task.output.length > 0) {
    const seed = task.output.map((output) => serializeArchivedOutput(output, "set", output.text)).join("");
    appendFileSync(path, seed, "utf8");
  }
  outputArchiveSeededTasks.add(task.id);
}

function appendOutputArchive(task: QueueTask, output: LiveOutput, op: ArchivedLiveOutput["op"], text: string): void {
  try {
    archivedOutputCache.delete(task.id);
    mkdirSync(ctx().config.outputArchiveDir, { recursive: true });
    appendFileSync(taskOutputArchivePath(task.id), serializeArchivedOutput(output, op, text), "utf8");
  } catch (error) {
    ctx().logger("error", "codex_output_archive_write_failed", { taskId: task.id, error: ctx().errorToJson(error) });
  }
}

function archiveRecordToOutput(value: unknown): ArchivedLiveOutput | null {
  if (typeof value !== "object" || value === null || Array.isArray(value)) return null;
  const record = value as Record<string, unknown>;
  const seq = Number(record.seq);
  if (!Number.isFinite(seq)) return null;
  const channel = String(record.channel || "system") as OutputChannel;
  if (!["system", "user", "assistant", "reasoning", "command", "diff", "tool", "error"].includes(channel)) return null;
  return {
    seq,
    at: typeof record.at === "string" ? record.at : ctx().nowIso(),
    channel,
    text: typeof record.text === "string" ? record.text : "",
    ...(typeof record.method === "string" ? { method: record.method } : {}),
    ...(typeof record.itemId === "string" ? { itemId: record.itemId } : {}),
    op: record.op === "append" ? "append" : "set",
  };
}

function archivedTaskOutput(task: QueueTask): LiveOutput[] {
  const path = taskOutputArchivePath(task.id);
  if (!existsSync(path)) return [];
  const signature = outputArchiveSignature(task);
  const cached = archivedOutputCache.get(task.id);
  if (cached?.signature === signature) return cached.output;
  const bySeq = new Map<number, LiveOutput>();
  try {
    const text = readFileSync(path, "utf8");
    for (const line of text.split(/\r?\n/u)) {
      if (line.trim().length === 0) continue;
      const record = archiveRecordToOutput(JSON.parse(line) as unknown);
      if (record === null) continue;
      const existing = bySeq.get(record.seq);
      if (record.op === "append" && existing !== undefined) {
        existing.text += record.text;
        existing.at = record.at;
        existing.channel = record.channel;
        existing.method = record.method;
        existing.itemId = record.itemId;
      } else {
        const { op: _op, ...output } = record;
        bySeq.set(record.seq, output);
      }
    }
  } catch (error) {
    ctx().logger("warn", "codex_output_archive_read_failed", { taskId: task.id, error: ctx().errorToJson(error) });
    return [];
  }
  const output = Array.from(bySeq.values()).sort((left, right) => Number(left.seq) - Number(right.seq));
  archivedOutputCache.set(task.id, { signature, output });
  while (archivedOutputCache.size > 80) {
    const firstKey = archivedOutputCache.keys().next().value;
    if (typeof firstKey === "string") archivedOutputCache.delete(firstKey);
  }
  return output;
}

function taskFullOutput(task: QueueTask): LiveOutput[] {
  const bySeq = new Map<number, LiveOutput>();
  for (const output of archivedTaskOutput(task)) bySeq.set(output.seq, output);
  for (const output of task.output) bySeq.set(output.seq, output);
  return Array.from(bySeq.values()).sort((left, right) => Number(left.seq) - Number(right.seq));
}

function outputArchiveSignature(task: QueueTask): string {
  try {
    const stat = statSync(taskOutputArchivePath(task.id));
    return `${stat.size}:${Math.floor(stat.mtimeMs)}`;
  } catch {
    return "none";
  }
}

function appendOutput(task: QueueTask, channel: OutputChannel, text: string, method?: string, itemId?: string, append = false): LiveOutput | null {
  if (text.length === 0) return null;
  try {
    ensureTaskOutputArchiveSeeded(task);
  } catch (error) {
    ctx().logger("error", "codex_output_archive_seed_failed", { taskId: task.id, error: ctx().errorToJson(error) });
  }
  const last = task.output[task.output.length - 1];
  let output: LiveOutput;
  let archiveOp: ArchivedLiveOutput["op"] = "set";
  let archiveText = text;
  if (append && last !== undefined && last.channel === channel && last.itemId === itemId && last.method === method && last.text.length < 24_000) {
    last.text += text;
    last.at = ctx().nowIso();
    output = last;
    archiveOp = "append";
  } else {
    output = { seq: ctx().allocateSeq(), at: ctx().nowIso(), channel, text, method, itemId };
    task.output.push(output);
  }
  appendOutputArchive(task, output, archiveOp, archiveText);
  if (ctx().config.maxInMemoryOutputRecords > 0 && task.output.length > ctx().config.maxInMemoryOutputRecords) task.output.splice(0, task.output.length - ctx().config.maxInMemoryOutputRecords);
  task.updatedAt = ctx().nowIso();
  ctx().markTaskDirty(task.id);
  ctx().schedulePersistState();
  ctx().onOutputAppended?.(task, output, archiveOp);
  return output;
}


export { appendOutput, appendOutputArchive, archivedTaskOutput, outputArchiveSignature, taskFullOutput };