From ae22ffd36e75845aefdb5c0cadb74066b1dfb5bd Mon Sep 17 00:00:00 2001 From: zbc Date: Thu, 28 May 2026 20:24:20 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E4=B8=80=E7=B3=BB=E5=88=97?= =?UTF-8?q?=E4=B8=8A=E6=8A=A5=E5=A4=B1=E8=B4=A5=E7=9A=84=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/services/rawDump/batchWorker.ts | 240 ++----- src/services/rawDump/index.ts | 84 +-- src/services/rawDump/queue.ts | 20 +- src/services/rawDump/session.ts | 613 ++++++++++++++++ src/services/rawDump/state.ts | 62 +- src/services/rawDump/statistics.ts | 217 ++++++ src/services/rawDump/types.ts | 22 +- src/services/rawDump/worker.ts | 1039 +++++++++++---------------- 8 files changed, 1378 insertions(+), 919 deletions(-) create mode 100644 src/services/rawDump/session.ts create mode 100644 src/services/rawDump/statistics.ts diff --git a/src/services/rawDump/batchWorker.ts b/src/services/rawDump/batchWorker.ts index fceeafcb8..53c6b55c5 100644 --- a/src/services/rawDump/batchWorker.ts +++ b/src/services/rawDump/batchWorker.ts @@ -6,66 +6,28 @@ * 单队列设计： * - 队列常驻内存，进程启动时一次性加载 * - 入队/移除时写文件，其他操作仅更新内存 - * - 任务成功或 attemptCount >= MAX_ATTEMPTS 时从队列移除 * - 任务失败但未超限：attemptCount++ 写回文件，下次 batch 重试 * * 每轮 batch 三道防线：进程内重入保护 → 跨进程文件锁 → 消费后清空文件 */ import { - uploadConversation, - uploadSummary, - uploadCommits, - uploadStatistics, - authWithFallback, + processIncompleteTasks, } from './worker.js' import { loadQueue, acquireLock, releaseLock, - peekTask, - removeTask, getQueue, - flushQueue, - MAX_ATTEMPTS, - type QueueTask, + clearQueue, } from './queue.js' -import { readState, writeState, appendDeadLetter } from './state.js' -import { getSessionDirectory, loadSessionMessages } from './worker.js' -import { getRepoInfo } from './git.js' +import { readState, writeState } from './state.js' import { createLogger } from './logger.js' const log = createLogger('raw-dump-batch') -/** - * 从错误信息中提取 API endpoint 路径 - * 错误格式："/raw-store/task-conversation failed: 401 ..." - */ -function extractEndpointFromError(errorMsg: string): string | undefined { - const match = errorMsg.match(/\/(raw-store\/[^\s]+)/) - return match?.[1] -} - -type RepoInfo = Awaited> - const BATCH_INTERVAL_MS = 120_000 - -// Git repo 信息缓存，同一 directory 的多个 task 短时间内不需要重复 spawn git -const repoInfoCache = new Map() -const REPO_CACHE_TTL_MS = 60_000 - -async function getCachedRepoInfo(directory: string): Promise { - const cached = repoInfoCache.get(directory) - if (cached && Date.now() - cached.ts < REPO_CACHE_TTL_MS) { - return cached.repoInfo - } - const repoInfo = await getRepoInfo(directory) - repoInfoCache.set(directory, { repoInfo, ts: Date.now() }) - return repoInfo -} - let isRunning = false - const PARENT_PID = process.ppid const IS_WORKER_PROCESS = process.argv[1]?.includes('batchWorker') || false @@ -79,111 +41,9 @@ function isParentAlive(): boolean { } } -// Session messages 缓存 -const sessionMessagesCache = new Map< - string, - { messages: Record[]; ts: number } ->() -const SESSION_CACHE_TTL_MS = 60_000 - -async function getCachedSessionMessages( - sessionDir: string, - sessionID: string, - messageID?: string, -) { - const cacheKey = `${sessionDir}:${sessionID}` - const cached = sessionMessagesCache.get(cacheKey) - if (cached && Date.now() - cached.ts < SESSION_CACHE_TTL_MS) { - return cached.messages - } - const start = Date.now() - const messages = await loadSessionMessages(sessionDir, sessionID, messageID) - const elapsed = Date.now() - start - if (elapsed > 100) { - log.info('loadSessionMessages slow', { sessionID, elapsedMs: elapsed }) - } - sessionMessagesCache.set(cacheKey, { messages, ts: Date.now() }) - return messages -} - -/** - * 处理单个上报任务 - * 成功完成后由调用方从队列中移除 - */ -async function processTask( - task: QueueTask, - state: Awaited>, -) { - log.info('processing task', { - sessionID: task.sessionID, - messageID: task.messageID, - attempt: task.attemptCount, - }) - - const sessionDir = getSessionDirectory(task.directory, task.sessionID) - const messages = await getCachedSessionMessages( - sessionDir, - task.sessionID, - task.messageID, - ) - - if (messages.length === 0) { - log.warn('no messages found', { sessionDir, sessionID: task.sessionID }) - } - - const authData = await authWithFallback() - const repoInfo = await getCachedRepoInfo(task.directory) - - if (task.messageID === '__statistics__' && task.statsData) { - await uploadStatistics( - { - sessionID: task.sessionID, - directory: task.directory, - sessionCount: task.statsData.sessionCount, - conversationCount: task.statsData.conversationCount, - upstreamTokens: task.statsData.upstreamTokens, - downstreamTokens: task.statsData.downstreamTokens, - startTime: task.statsData.startTime, - endTime: task.statsData.endTime, - }, - authData, - state, - ) - log.info('statistics task completed', { sessionID: task.sessionID }) - return - } - - const conversationUploaded = await uploadConversation( - { - sessionID: task.sessionID, - messageID: task.messageID, - directory: task.directory, - messages, - }, - authData, - state, - { repoInfo }, - ) - - await uploadSummary( - { sessionID: task.sessionID, directory: task.directory, messages }, - authData, - state, - ) - - await uploadCommits({ directory: task.directory }, authData, state, { - repoInfo, - }) - - log.info('task completed', { - sessionID: task.sessionID, - conversationUploaded, - }) -} - /** * 执行一轮 batch 处理 - * 流程：获取锁 → 读队列内存 → 去重 → 逐个处理 → 写 state + * 流程：获取锁 → 读 state 和队列 → 同步 tasks → 保存 state 和队列 → 逐个处理 */ async function runBatch() { if (isRunning) { @@ -200,68 +60,62 @@ async function runBatch() { try { const state = await readState() - const tasks = getQueue() + const newTasks = getQueue() - if (tasks.length === 0) { + if (newTasks.length === 0) { log.debug('queue empty') return } - log.info(`processing ${tasks.length} tasks`) + log.info(`processing ${newTasks.length} new tasks`) - // 去重：同一 session:messageID 只保留最新的 - const deduped = new Map() - for (const task of tasks) { + // 同步 tasks：根据 task-id:message-id 查询 state.tasks + for (const task of newTasks) { const key = `${task.sessionID}:${task.messageID}` - const existing = deduped.get(key) - if (!existing || task.enqueuedAt > existing.enqueuedAt) { - deduped.set(key, task) - } - } + const existing = state.tasks[key] - const uniqueTasks = Array.from(deduped.values()).sort( - (a, b) => a.enqueuedAt.localeCompare(b.enqueuedAt), - ) - - log.info(`deduped to ${uniqueTasks.length} unique tasks`) - - for (const task of uniqueTasks) { - const key = `${task.sessionID}:${task.messageID}` - try { - await processTask(task, state) - removeTask(key) - } catch (err) { - const errorMsg = err instanceof Error ? err.message : String(err) - task.attemptCount++ - log.error('task failed', { - error: errorMsg, - sessionID: task.sessionID, - attemptCount: task.attemptCount, - }) - - if (task.attemptCount >= MAX_ATTEMPTS) { - // 彻底失败：写入 dead letter，从队列移除 - await appendDeadLetter({ - sessionID: task.sessionID, - messageID: task.messageID, - directory: task.directory, - attemptCount: task.attemptCount, - error: errorMsg, - endpoint: extractEndpointFromError(errorMsg), - failedAt: new Date().toISOString(), + if (existing) { + // 已存在该键，比较 lastEnqueuedAt 和当前请求的时间戳 + if (task.enqueuedAt > existing.lastEnqueuedAt) { + // 当前请求更新，更新时间戳 lastEnqueuedAt，清掉 lastUploadAt，累加 taskCount + existing.lastEnqueuedAt = task.enqueuedAt + existing.lastUploadAt = '' + existing.taskCount++ + log.debug('task updated, needs re-upload', { + key, + taskCount: existing.taskCount, + lastEnqueuedAt: existing.lastEnqueuedAt, }) - removeTask(key) - log.warn('task permanently failed, removed from queue', { - sessionID: task.sessionID, - attemptCount: task.attemptCount, - }) - } else { - // 失败但未超限：attemptCount++ 写回文件，下次重试 - await flushQueue() } + // 如果当前请求没有更新，则忽略 + } else { + // 不存在，添加到 state.tasks 中，设置初值 + state.tasks[key] = { + lastEnqueuedAt: task.enqueuedAt, + lastUploadAt: '', + taskCount: 1, + attemptCount: 0, + directory: task.directory, + } + log.debug('task added to tracking', { + key, + lastEnqueuedAt: task.enqueuedAt, + }) } } + // 保存 state 到磁盘 + await writeState(state) + + // 清掉队列并且把队列内容更新到磁盘文件 + clearQueue() + + log.info(`tracking ${Object.keys(state.tasks).length} tasks total`) + + // 处理 state.tasks 中未完成的任务 + await processIncompleteTasks(state) + + // 再次保存 state（更新 lastUploadAt） await writeState(state) log.info('batch completed') } finally { diff --git a/src/services/rawDump/index.ts b/src/services/rawDump/index.ts index 4e50465ee..50edf0ab8 100644 --- a/src/services/rawDump/index.ts +++ b/src/services/rawDump/index.ts @@ -17,10 +17,6 @@ const log = createLogger('raw-dump') let batchWorkerSpawned = false -// 调用频率限制：同一 session + messageID 5s 内不重复 enqueue -const lastEnqueueMap = new Map() -const ENQUEUE_DEBOUNCE_MS = 5_000 - /** * 判断 Raw Dump 是否启用 * - mode=0 时不输出（完全禁用） @@ -60,30 +56,11 @@ function ensureBatchWorker() { } } -/** - * 判断当前 session + message 是否需要 enqueue（去重） - * 同一 key 在 5 秒内只允许 enqueue 一次，避免重复上报 - * @returns true 表示需要入队，false 表示跳过 - */ -function shouldEnqueue(sessionID: string, messageID: string): boolean { - const key = `${sessionID}:${messageID}` - const now = Date.now() - const last = lastEnqueueMap.get(key) - if (last && now - last < ENQUEUE_DEBOUNCE_MS) { - log.debug('reportTurn debounced', { - sessionID, - messageID, - lastMs: now - last, - }) - return false - } - lastEnqueueMap.set(key, now) - return true -} - /** * 上报一轮对话 * 只写入队列，由 batch worker 顺序消费 + * reportTurn 代表有新的大模型调用（即对话），有信息变化需要上报 + * 去重放在执行任务的逻辑中（batchWorker 的 runBatch） */ export async function reportTurn( sessionID: string, @@ -91,64 +68,7 @@ export async function reportTurn( directory: string, ): Promise { if (!isEnabled()) return - if (!shouldEnqueue(sessionID, messageID)) return ensureRawDumpDirCreated() enqueue({ sessionID, messageID, directory }) ensureBatchWorker() } - -/** - * 上报 session 摘要信息 - * 使用特殊 messageID '__summary__' 标识，与普通 conversation 分开去重 - * 只写入队列，由 batch worker 顺序消费 - */ -export async function reportSession( - sessionID: string, - directory: string, -): Promise { - if (!isEnabled()) return - if (!shouldEnqueue(sessionID, '__summary__')) return - ensureRawDumpDirCreated() - enqueue({ sessionID, messageID: '__summary__', directory }) - ensureBatchWorker() -} - -export interface StatisticsData { - sessionCount: number - conversationCount: number - upstreamTokens: number - downstreamTokens: number - startTime: number - endTime: number -} - -const lastReportStatsMap = new Map() -const STATS_DEBOUNCE_MS = 60_000 // 同一 session 1 分钟内不重复 enqueue - -/** - * 上报对账统计数据（session数、conversation数、token数） - * 通过特殊 messageID '__statistics__' 标识入队 - */ -export async function reportStatistics( - sessionID: string, - directory: string, - data: StatisticsData, -): Promise { - if (!isEnabled()) return - const key = `${sessionID}:__statistics__` - const now = Date.now() - const last = lastReportStatsMap.get(key) - if (last && now - last < STATS_DEBOUNCE_MS) { - log.debug('reportStatistics debounced', { sessionID, lastMs: now - last }) - return - } - lastReportStatsMap.set(key, now) - ensureRawDumpDirCreated() - enqueue({ - sessionID, - messageID: '__statistics__', - directory, - statsData: data, - }) - ensureBatchWorker() -} diff --git a/src/services/rawDump/queue.ts b/src/services/rawDump/queue.ts index e3f0ffd78..50a846fc1 100644 --- a/src/services/rawDump/queue.ts +++ b/src/services/rawDump/queue.ts @@ -12,8 +12,6 @@ import { promises as fs } from 'node:fs' import os from 'node:os' import path from 'path' -import type { StatisticsData } from './index.js' - const QUEUE_FILE = path.join(os.homedir(), '.claude', 'raw-dump', 'csc-work-queue.jsonl') const LOCK_FILE = path.join(os.homedir(), '.claude', 'raw-dump', 'csc-work-queue.lock') @@ -25,7 +23,6 @@ export interface QueueTask { directory: string enqueuedAt: string // RFC3339 format, e.g. "2026-05-27T10:00:00.000Z" attemptCount: number - statsData?: StatisticsData } // ----------- 内存中的队列 ----------- @@ -76,20 +73,11 @@ export function enqueue(task: Omit): v } /** - * 从内存队列中移除任务（成功或彻底失败时调用） - * @param key sessionID:messageID + * 清空内存中的队列 */ -export function removeTask(key: string): void { - queue = queue.filter(t => `${t.sessionID}:${t.messageID}` !== key) - flushQueue().catch(() => {}) -} - -/** - * 消费队首任务（不移除，仅返回引用） - * 用于在处理前 peek，失败时自行更新 attemptCount 再写回 - */ -export function peekTask(): QueueTask | undefined { - return queue[0] +export function clearQueue(): void { + queue = [] + fs.writeFile(QUEUE_FILE, '', 'utf-8').catch(() => {}) } /** diff --git a/src/services/rawDump/session.ts b/src/services/rawDump/session.ts new file mode 100644 index 000000000..d579cc649 --- /dev/null +++ b/src/services/rawDump/session.ts @@ -0,0 +1,613 @@ +/** + * Raw Dump Session Utilities + * 与 session 相关的工具函数，从 worker.ts 和 batchWorker.ts 中剥离 + */ + +import { promises as fs } from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { createLogger } from './logger.js' +import { countDiffLines, extractFilesFromDiff } from './git.js' + +const log = createLogger('raw-dump-session') + +/** + * 获取 CoStrict 配置目录（~/.claude），支持环境变量 CLAUDE_CONFIG_HOME 覆盖 + */ +export function getClaudeConfigHomeDir(): string { + return process.env.CLAUDE_CONFIG_HOME || path.join(os.homedir(), '.claude') +} + +/** + * 将目录路径规范化为安全的车间目录名 + * 将路径中的 / 替换为 -，避免路径分隔符在文件系统操作中引发问题 + * 如 /Users/linkai/code/csc → -Users-linkai-code-csc + */ +export function normalizeProjectPath(dir: string): string { + return dir.replace(/:/, '-').replace(/[/\\]/g, '-') +} + +/** + * 解析 session 文件所在目录 + * 按优先级尝试多个候选路径，返回第一个存在的目录： + * 1. ~/.claude/projects/{normalized-path}/ + * 2. ~/.claude/transcripts/ + * 3. ~/.claude/sessions/ + * 4. {directory}/.claude/sessions/ + * 5. {directory}/.claude/ + * 6. {directory} 本身 + * 7. 环境变量 CSC_SESSION_DIR 指定路径 + */ +export function getSessionDirectory( + directory: string, + sessionID: string, +): string { + const claudeHome = getClaudeConfigHomeDir() + const projectPath = normalizeProjectPath(directory) + const candidates = [ + path.join(claudeHome, 'projects', projectPath), + path.join(claudeHome, 'transcripts'), + path.join(claudeHome, 'sessions'), + path.join(directory, '.claude', 'sessions'), + path.join(directory, '.claude'), + directory, + process.env.CSC_SESSION_DIR || '', + ] + return candidates.find(d => d) || directory +} + +/** + * 从 JSONL 文件加载会话消息列表 + * - 扫描 sessionDir 目录下所有 .jsonl 文件 + * - 优先读取文件名包含 sessionId 的文件（减少无意义解析） + * - 通过 sessionId、messageId 匹配目标记录 + * - csc 的会话文件名可能是 ses_{hash}.jsonl 或 {uuid}.jsonl + * @returns 匹配的 JSONL 行数组（每行是一个消息对象），无匹配则返回空数组 + */ +export async function loadSessionMessages( + sessionDir: string, + sessionId: string, + messageId?: string, +) { + try { + const entries = await fs.readdir(sessionDir) + const jsonlFiles = entries.filter(f => f.endsWith('.jsonl')) + log.debug('found jsonl files', { + sessionDir, + count: jsonlFiles.length, + files: jsonlFiles.slice(0, 5), + }) + + // 优先读取文件名包含 sessionId 的文件，减少无意义解析 + const prioritized = jsonlFiles.sort((a, b) => { + const aHas = a.includes(sessionId) + const bHas = b.includes(sessionId) + if (aHas && !bHas) return -1 + if (!aHas && bHas) return 1 + return 0 + }) + + for (const file of prioritized) { + const filePath = path.join(sessionDir, file) + try { + const text = await fs.readFile(filePath, 'utf-8') + const lines = text + .split('\n') + .filter(Boolean) + .map(line => { + try { + return JSON.parse(line) + } catch { + return null + } + }) + .filter((m): m is Record => m !== null) + + // 检查是否包含目标 sessionId 或 messageId + const hasSession = lines.some( + m => + m.sessionId === sessionId || + m.session_id === sessionId || + m.uuid === sessionId, + ) + const hasMessage = messageId + ? lines.some( + m => + m.uuid === messageId || + (m.message as Record)?.id === messageId, + ) + : false + if (hasSession || hasMessage) { + log.debug('loaded messages from file', { + file, + count: lines.length, + hasSession, + hasMessage, + }) + return lines + } + } catch { + // ignore per-file errors + } + } + } catch { + // ignore dir read errors + } + return [] +} + +/** + * 在消息列表中查找指定 ID 的消息 + * 支持通过 message.uuid 或 message.id 匹配 + */ +export function findMessage( + messages: Record[], + messageID: string, +): Record | undefined { + return messages.find( + m => + m.uuid === messageID || + (m.message as Record)?.id === messageID, + ) +} + +/** + * 查找 assistant message 对应的父级 user message + * 在 csc 中，user message 通常紧邻 assistant message 之前 + * 从 assistant 位置向前遍历，找到第一个 type === 'user' 的消息 + */ +export function findParentUserMessage( + messages: Record[], + assistantMsg: Record, +): Record | undefined { + const assistantIndex = messages.indexOf(assistantMsg) + if (assistantIndex <= 0) return undefined + for (let i = assistantIndex - 1; i >= 0; i--) { + if (messages[i]?.type === 'user') return messages[i] + } + return undefined +} + +/** + * 判断本轮对话的发送者类型：agent 或 user + * 通过 assistant 的 mode、agent 字段、isSidechain 标记、user 的 isMeta 标记综合判断 + */ +export function detectSender( + assistant: Record, + user: Record | undefined, +): string { + const mode = String(assistant.mode ?? '') + if (mode === 'agent' || mode === 'auto') return 'agent' + if (assistant.agent) return 'agent' + if (assistant.isSidechain === true) return 'agent' + if (user?.isMeta === true) return 'agent' + return 'user' +} + +/** + * 从消息对象中提取纯文本内容 + * 兼容多种格式：直接 content 字符串，或 content 数组（block type === 'text'） + */ +export function extractTextContent(msg: Record): string { + const content = (msg.message as Record)?.content + if (!Array.isArray(content)) return String(content ?? '') + return content + .filter((block): block is Record => block?.type === 'text') + .map(block => String(block.text ?? '')) + .join('\n') +} + +/** + * 将 structured patch 格式转换为 unified diff 格式 + */ +function structuredPatchToUnifiedDiff( + filePath: string, + patches: Array>, +): string { + if (!patches.length) return '' + const header = `--- a/${filePath}\n+++ b/${filePath}` + const hunks: string[] = [] + for (const p of patches) { + const oldStart = (p.oldStart as number) ?? 1 + const oldLines = (p.oldLines as number) ?? 0 + const newStart = (p.newStart as number) ?? 1 + const newLines = (p.newLines as number) ?? 0 + const lines = (p.lines as string[]) ?? [] + hunks.push( + `@@ -${oldStart},${oldLines} +${newStart},${newLines} @@\n${lines.join('\n')}`, + ) + } + return header + '\n' + hunks.join('\n') + '\n' +} + +/** + * 根据旧字符串和新字符串生成 unified diff + */ +function generateStringDiff( + filePath: string, + oldStr: string, + newStr: string, +): string { + const oldLines = oldStr.split('\n') + const newLines = newStr.split('\n') + const header = `--- a/${filePath}\n+++ b/${filePath}` + const hunk = `@@ -1,${oldLines.length} +1,${newLines.length} @@` + const body = oldLines + .map(l => `-${l}`) + .concat(newLines.map(l => `+${l}`)) + .join('\n') + return header + '\n' + hunk + '\n' + body + '\n' +} + +/** + * 从 assistant message 中提取 unified diff + * 优先从子 user message 的 toolUseResult 获取 gitDiff.patch（已格式化的 unified diff） + * fallback 到 structuredPatch 转换或从 tool_use input 参数生成 diff + */ +export function extractToolDiff( + msg: Record, + allMessages?: Record[], +): { diff: string; diff_lines: number; files: string[] } { + const diffs: string[] = [] + const files = new Set() + const handledToolUseIds = new Set() + + const msgUuid = msg.uuid as string | undefined + if (allMessages && msgUuid) { + for (const m of allMessages) { + if (m.parentUuid !== msgUuid || m.type !== 'user') continue + const tur = (m.toolUseResult ?? m.tool_use_result) as + | Record + | undefined + if (!tur) continue + + const filePath = + (tur.filePath as string | undefined) ?? + (tur.file_path as string | undefined) ?? + (tur.notebook_path as string | undefined) ?? + '' + + const gitDiff = tur.gitDiff as Record | undefined + if ( + gitDiff?.patch && + typeof gitDiff.patch === 'string' && + gitDiff.patch + ) { + diffs.push(gitDiff.patch) + if (gitDiff.filename && typeof gitDiff.filename === 'string') + files.add(gitDiff.filename) + else if (filePath) files.add(filePath) + const mContent = (m.message as Record)?.content + if (Array.isArray(mContent)) { + for (const b of mContent) { + if ( + b?.type === 'tool_result' && + typeof b.tool_use_id === 'string' + ) { + handledToolUseIds.add(b.tool_use_id) + } + } + } + continue + } + + const sp = tur.structuredPatch as + | Array> + | undefined + if (Array.isArray(sp) && sp.length > 0 && filePath) { + diffs.push(structuredPatchToUnifiedDiff(filePath, sp)) + files.add(filePath) + const mContent = (m.message as Record)?.content + if (Array.isArray(mContent)) { + for (const b of mContent) { + if ( + b?.type === 'tool_result' && + typeof b.tool_use_id === 'string' + ) { + handledToolUseIds.add(b.tool_use_id) + } + } + } + continue + } + + const oldStr = + (tur.oldString as string | undefined) ?? + (tur.old_string as string | undefined) ?? + (tur.originalFile as string | undefined) ?? + (tur.original_file as string | undefined) + const newStr = + (tur.newString as string | undefined) ?? + (tur.new_string as string | undefined) ?? + (tur.content as string | undefined) ?? + (tur.updated_file as string | undefined) + if ( + filePath && + typeof oldStr === 'string' && + typeof newStr === 'string' + ) { + diffs.push(generateStringDiff(filePath, oldStr, newStr)) + files.add(filePath) + } else if (filePath && typeof newStr === 'string') { + diffs.push(generateStringDiff(filePath, '', newStr)) + files.add(filePath) + } + } + } + + const content = (msg.message as Record)?.content + if (Array.isArray(content)) { + for (const block of content) { + if (block?.type !== 'tool_use') continue + if (typeof block.id === 'string' && handledToolUseIds.has(block.id)) + continue + + const input = block.input as Record | undefined + const toolName = block.name as string | undefined + const filePath = + (input?.file_path as string | undefined) ?? + (input?.notebook_path as string | undefined) ?? + '' + + if ( + toolName === 'Edit' && + typeof input?.old_string === 'string' && + typeof input?.new_string === 'string' + ) { + diffs.push( + generateStringDiff(filePath, input.old_string, input.new_string), + ) + if (filePath) files.add(filePath) + } else if ( + toolName === 'NotebookEdit' && + typeof input?.new_source === 'string' && + filePath + ) { + diffs.push(generateStringDiff(filePath, '', input.new_source)) + files.add(filePath) + } else if (typeof input?.diff === 'string' && input.diff) { + diffs.push(input.diff) + if (filePath) files.add(filePath) + } else if (typeof input?.patch === 'string' && input.patch) { + diffs.push(input.patch) + if (filePath) files.add(filePath) + } + } + } + + const diff = diffs.join('\n') + for (const file of extractFilesFromDiff(diff)) files.add(file) + return { diff, diff_lines: countDiffLines(diff), files: Array.from(files) } +} + +/** + * 从 assistant message 中提取 token 使用量 + */ +export function extractUsage(msg: Record) { + const usage = (msg.message as Record)?.usage as + | Record + | undefined + return { + input_tokens: usage?.input_tokens ?? 0, + output_tokens: usage?.output_tokens ?? 0, + cache_read_input_tokens: usage?.cache_read_input_tokens ?? 0, + cache_creation_input_tokens: usage?.cache_creation_input_tokens ?? 0, + } +} + +const SDK_ERROR_CODE_MAP: Record = { + authentication_failed: 401, + billing_error: 402, + rate_limit: 429, + invalid_request: 400, + server_error: 500, + max_output_tokens: 413, + unknown: 500, +} + +/** + * 从 assistant message 中提取错误信息 + */ +export function extractError(msg: Record): { + error_code?: number + error_reason?: string +} { + const error = msg.error + + if (typeof error === 'string') { + const errorCode = SDK_ERROR_CODE_MAP[error] ?? 500 + const errorDetails = msg.errorDetails + const reason = + typeof errorDetails === 'string' && errorDetails ? errorDetails : error + return { error_code: errorCode, error_reason: reason } + } + + if (typeof error === 'object' && error !== null) { + const err = error as Record + const apiError = msg.apiError + if (typeof apiError === 'string' && apiError === 'max_output_tokens') { + const reason = + typeof err.message === 'string' ? err.message : 'max_output_tokens' + return { error_code: 413, error_reason: reason } + } + const name = String(err.name ?? 'UnknownError') + const errMessage = typeof err.message === 'string' ? err.message : name + const errorCode = + name === 'ProviderAuthError' + ? 401 + : name === 'ContextOverflowError' || name === 'MessageOutputLengthError' + ? 413 + : name === 'MessageAbortedError' + ? 499 + : name === 'APIError' && typeof err.statusCode === 'number' + ? err.statusCode + : 500 + return { error_code: errorCode, error_reason: errMessage } + } + + if (msg.isApiErrorMessage) { + return { error_code: 500, error_reason: 'api_error_unclassified' } + } + + return {} +} + +/** + * 从 sessionDir 目录中获取最新的 sessionId 和 messageId + * 扫描所有 .jsonl 文件，按 timestamp 降序排列，返回最新一条消息的 sessionId 和 messageId + */ +export async function getLatestSessionInfo( + sessionDir: string, +): Promise<{ sessionId: string; messageId: string } | null> { + try { + const entries = await fs.readdir(sessionDir) + const jsonlFiles = entries.filter(f => f.endsWith('.jsonl')) + if (jsonlFiles.length === 0) return null + + let latestMsg: { sessionId: string; messageId: string; ts: number } | null = null + + for (const file of jsonlFiles) { + const filePath = path.join(sessionDir, file) + try { + const text = await fs.readFile(filePath, 'utf-8') + const lines = text.split('\n').filter(Boolean) + for (const line of lines) { + try { + const msg = JSON.parse(line) as Record + const ts = (msg.timestamp as number) || 0 + if (!latestMsg || ts > latestMsg.ts) { + const sessionId = + (msg.sessionId as string) || + (msg.session_id as string) || + (msg.uuid as string) || + '' + const messageId = + (msg.uuid as string) || + ((msg.message as Record)?.id as string) || + '' + if (sessionId && messageId) { + latestMsg = { sessionId, messageId, ts } + } + } + } catch { + // ignore parse errors + } + } + } catch { + // ignore file read errors + } + } + + if (latestMsg) { + log.debug('found latest session info', { + sessionId: latestMsg.sessionId, + messageId: latestMsg.messageId, + ts: latestMsg.ts, + }) + return { sessionId: latestMsg.sessionId, messageId: latestMsg.messageId } + } + } catch { + // ignore dir read errors + } + return null +} + +// Session messages 缓存（batchWorker 使用） +// 缓存结构: [sessionJsonlFileName, file-timestamp, cache-timestamp, messages] +type SessionMessagesCacheEntry = [ + sessionJsonlFileName: string, + fileTimestamp: number, + cacheTimestamp: number, + messages: Record[], +] +const sessionMessagesCache = new Map() + +/** + * 获取缓存的 session messages + * 缓存 key: sessionDir:sessionId + * 缓存值: [sessionJsonlFileName, file-timestamp, cache-timestamp, messages] + * 当文件相比缓存时有更新（只要变化就认为有更新），则重新加载 + */ +export async function getCachedSessionMessages( + sessionDir: string, + sessionID: string, + messageID?: string, +) { + const cacheKey = `${sessionDir}:${sessionID}` + const cached = sessionMessagesCache.get(cacheKey) + + // 尝试找到当前 session 文件 + const findCurrentFile = async (): Promise<{ fileName: string; filePath: string } | null> => { + try { + const entries = await fs.readdir(sessionDir) + const jsonlFiles = entries.filter(f => f.endsWith('.jsonl')) + const prioritized = jsonlFiles.sort((a, b) => { + const aHas = a.includes(sessionID) + const bHas = b.includes(sessionID) + if (aHas && !bHas) return -1 + if (!aHas && bHas) return 1 + return 0 + }) + for (const file of prioritized) { + const filePath = path.join(sessionDir, file) + const text = await fs.readFile(filePath, 'utf-8') + const lines = text.split('\n').filter(Boolean) + const hasSession = lines.some(m => { + const msg = JSON.parse(m) as Record + return msg.sessionId === sessionID || msg.session_id === sessionID || msg.uuid === sessionID + }) + if (hasSession) { + return { fileName: file, filePath } + } + } + } catch {} + return null + } + + if (cached) { + const [cachedFileName, cachedFileTs] = cached + const currentFile = await findCurrentFile() + + if (currentFile && currentFile.fileName === cachedFileName) { + // 文件名匹配，检查文件是否更新 + try { + const stats = await fs.stat(path.join(sessionDir, cachedFileName)) + const currentFileTs = stats.mtimeMs + if (currentFileTs === cachedFileTs) { + // 文件未变化，直接使用缓存 + log.debug('using cached session messages', { sessionID, cachedFileName }) + return cached[3] + } + } catch {} + } + + // 文件有更新或文件名不匹配，需要重新加载 + log.debug('file changed, reloading session messages', { sessionID }) + } + + // 未找到缓存或文件已更新，搜索 session 文件 + const currentFile = await findCurrentFile() + + if (currentFile) { + const start = Date.now() + const messages = await loadSessionMessages(sessionDir, sessionID, messageID) + const elapsed = Date.now() - start + if (elapsed > 100) { + log.info('loadSessionMessages slow', { sessionID, elapsedMs: elapsed }) + } + try { + const stats = await fs.stat(path.join(sessionDir, currentFile.fileName)) + sessionMessagesCache.set(cacheKey, [ + currentFile.fileName, + stats.mtimeMs, + Date.now(), + messages, + ]) + } catch {} + return messages + } + + // 文件不存在，缓存空结果 + sessionMessagesCache.set(cacheKey, ['', 0, Date.now(), []]) + return [] +} diff --git a/src/services/rawDump/state.ts b/src/services/rawDump/state.ts index 4721098e7..98e0bf0f9 100644 --- a/src/services/rawDump/state.ts +++ b/src/services/rawDump/state.ts @@ -11,7 +11,7 @@ import { promises as fs, readFileSync, writeFileSync } from 'node:fs' import os from 'node:os' import path from 'node:path' -import type { RawDumpState, DeadLetterEntry } from './types.js' +import type { RawDumpState, DeadLetterEntry, TaskRecord } from './types.js' const STATE_DIR = path.join(os.homedir(), '.claude', 'raw-dump') const STATE_FILE = path.join(STATE_DIR, 'csc-state.json') @@ -27,6 +27,61 @@ function createEmptyState(): RawDumpState { conversation: {}, summary: {}, commits: {}, + statistics: {}, + tasks: {}, + } +} + +/** + * 清理昨日之前的已完成上报的 conversation、summary、tasks 记录 + * commits 记录保留 + * @param state 待清理的 state 对象 + */ +function cleanupOldRecords(state: RawDumpState): RawDumpState { + const now = Date.now() + // 获取昨日 0 点的时间戳 + const yesterday = new Date() + yesterday.setDate(yesterday.getDate() - 1) + yesterday.setHours(0, 0, 0, 0) + const yesterdayMs = yesterday.getTime() + + const cleanedConversation: Record = {} + for (const [key, ts] of Object.entries(state.conversation)) { + if (!ts) continue + const date = new Date(ts).getTime() + if (date >= yesterdayMs) { + cleanedConversation[key] = ts + } + } + + const cleanedSummary: Record = {} + for (const [key, ts] of Object.entries(state.summary)) { + if (!ts) continue + const date = new Date(ts).getTime() + if (date >= yesterdayMs) { + cleanedSummary[key] = ts + } + } + + const cleanedTasks: Record = {} + for (const [key, record] of Object.entries(state.tasks)) { + // 清理 lastUploadAt 非空且已超过昨日的记录 + if (record.lastUploadAt) { + const date = new Date(record.lastUploadAt).getTime() + if (date >= yesterdayMs) { + cleanedTasks[key] = record + } + } else { + // lastUploadAt 为空表示尚未处理完毕，保留 + cleanedTasks[key] = record + } + } + + return { + ...state, + conversation: cleanedConversation, + summary: cleanedSummary, + tasks: cleanedTasks, } } @@ -101,11 +156,14 @@ export async function readState(): Promise { try { const text = await fs.readFile(STATE_FILE, 'utf-8') const parsed = JSON.parse(text) as Partial - return { + const state: RawDumpState = { conversation: parsed.conversation ?? {}, summary: parsed.summary ?? {}, commits: parsed.commits ?? {}, + statistics: parsed.statistics ?? {}, + tasks: parsed.tasks ?? {}, } + return cleanupOldRecords(state) } catch { return createEmptyState() } diff --git a/src/services/rawDump/statistics.ts b/src/services/rawDump/statistics.ts new file mode 100644 index 000000000..19f6e18d0 --- /dev/null +++ b/src/services/rawDump/statistics.ts @@ -0,0 +1,217 @@ +/** + * Raw Dump 统计信息管理 + * 使用全局变量维护当天的统计值，按天隔离 + * 有新的 conversation、summary、commit 时更新统计值 + * 上报完成后清理今日之前的记录 + */ + +import { readState, writeState } from './state.js' + +/** + * 当天统计数据结构 + */ +export interface DailyStatistics { + dateKey: string // YYYY/MM/DD 格式 + sessionCount: number + conversationCount: number + upstreamTokens: number + downstreamTokens: number + startTime: number // 当天最早一条记录的时间戳 + endTime: number // 当天最新一条记录的时间戳 +} + +/** + * 获取当天日期 key，格式为 YYYY/MM/DD + */ +function getTodayKey(): string { + const now = new Date() + const year = now.getFullYear() + const month = String(now.getMonth() + 1).padStart(2, '0') + const day = String(now.getDate()).padStart(2, '0') + return `${year}/${month}/${day}` +} + +/** + * 格式化时间戳为 ISO 字符串 + */ +function formatIso(ms: number | undefined): string { + if (!ms) return '' + return new Date(ms).toISOString().replace(/\.\d{3}Z$/, 'Z') +} + +// 全局变量，维护当天的统计值 +let globalStats: DailyStatistics = { + dateKey: getTodayKey(), + sessionCount: 0, + conversationCount: 0, + upstreamTokens: 0, + downstreamTokens: 0, + startTime: 0, + endTime: 0, +} + +/** + * 重置全局统计值，重新初始化为当天 + */ +function resetGlobalStats(): void { + globalStats = { + dateKey: getTodayKey(), + sessionCount: 0, + conversationCount: 0, + upstreamTokens: 0, + downstreamTokens: 0, + startTime: 0, + endTime: 0, + } +} + +/** + * 检查并处理日期变更 + * 如果日期从昨天切换到今天，重置统计值 + */ +function checkAndResetForNewDay(): void { + const todayKey = getTodayKey() + if (globalStats.dateKey !== todayKey) { + resetGlobalStats() + } +} + +/** + * 更新会话统计（当有新的 session 时调用） + * @param timestamp 当前会话的时间戳 + */ +export function incrementSession(timestamp: number): void { + checkAndResetForNewDay() + globalStats.sessionCount++ + updateTimeRange(timestamp) +} + +/** + * 更新对话统计（当有新的 conversation 上报成功时调用） + * @param timestamp 当前对话的时间戳 + */ +export function incrementConversation(timestamp: number): void { + checkAndResetForNewDay() + globalStats.conversationCount++ + updateTimeRange(timestamp) +} + +/** + * 添加 token 统计（当有新的 conversation 上报成功时调用） + * @param upstream 上游 token 数量 + * @param downstream 下游 token 数量 + * @param timestamp 当前对话的时间戳 + */ +export function addTokens( + upstream: number, + downstream: number, + timestamp: number, +): void { + checkAndResetForNewDay() + globalStats.upstreamTokens += upstream + globalStats.downstreamTokens += downstream + updateTimeRange(timestamp) +} + +/** + * 更新时间范围 + * @param timestamp 时间戳 + */ +function updateTimeRange(timestamp: number): void { + if (globalStats.startTime === 0 || timestamp < globalStats.startTime) { + globalStats.startTime = timestamp + } + if (globalStats.endTime === 0 || timestamp > globalStats.endTime) { + globalStats.endTime = timestamp + } +} + +/** + * 获取当前的全局统计值 + * @returns 当天的统计值副本 + */ +export function getCurrentStatistics(): DailyStatistics { + checkAndResetForNewDay() + return { ...globalStats } +} + +/** + * 获取用于上报的统计数据 + * 用于 uploadStatistics 上报 + */ +export function getStatisticsForUpload(): { + sessionCount: number + conversationCount: number + upstreamTokens: number + downstreamTokens: number + startTime: number + endTime: number +} { + checkAndResetForNewDay() + return { + sessionCount: globalStats.sessionCount, + conversationCount: globalStats.conversationCount, + upstreamTokens: globalStats.upstreamTokens, + downstreamTokens: globalStats.downstreamTokens, + startTime: globalStats.startTime, + endTime: globalStats.endTime, + } +} + +/** + * 上报完成后清理历史记录 + * 读取 state，将 statistics 中今天之前的记录清理掉 + * 避免 state 文件无限增长 + */ +export async function cleanupOldStatistics(): Promise { + try { + const state = await readState() + const todayKey = getTodayKey() + + // 过滤出今天的记录，清理掉昨天及之前的 + const cleanedStatistics: Record = {} + for (const [key, ts] of Object.entries(state.statistics)) { + if (key === todayKey) { + cleanedStatistics[key] = ts + } + } + + if (Object.keys(cleanedStatistics).length !== Object.keys(state.statistics).length) { + state.statistics = cleanedStatistics + await writeState(state) + } + } catch { + // ignore cleanup errors + } +} + +const STATS_HOURLY_WINDOW_MS = 60 * 60 * 1000 // 同一天内，statistics 上报间隔 1 小时 + +/** + * 检查是否需要上报 statistics + * - 如果是当天：检查上次上报时间是否超过 1 小时间隔 + * - 如果不是当天（历史日期）：已上报过则不再上报 + * @param dateKey YYYY/MM/DD 格式的日期 + * @param state 当前状态 + * @returns true 表示需要上报，false 表示跳过 + */ +export function shouldReportStatistics( + dateKey: string, + state: Awaited>, +): boolean { + const todayKey = getTodayKey() + const lastReported = state.statistics[dateKey] + + if (!lastReported) { + return true // 从未上报过，需要上报 + } + + if (dateKey !== todayKey) { + return false // 历史日期已上报过，不再上报 + } + + // 当天：检查是否超过 1 小时间隔 + const lastTime = new Date(lastReported).getTime() + const now = Date.now() + return now - lastTime >= STATS_HOURLY_WINDOW_MS +} \ No newline at end of file diff --git a/src/services/rawDump/types.ts b/src/services/rawDump/types.ts index 9e6774481..c58cae472 100644 --- a/src/services/rawDump/types.ts +++ b/src/services/rawDump/types.ts @@ -20,10 +20,20 @@ export interface RawDumpEventPayload { directory: string } +export interface TaskRecord { + lastEnqueuedAt: string // RFC3339，最后一次收到该组请求的时间戳 + lastUploadAt: string // RFC3339，请求处理完毕的时间戳，为空表示尚未处理完毕，"DEAD_LETTER"表示已移至死信 + taskCount: number // 收到的该组请求总个数 + attemptCount: number // 已尝试次数，超过 MAX_ATTEMPTS 后移至 dead letter + directory: string // 工作目录，用于定位 session 文件 +} + export interface RawDumpState { - conversation: Record - summary: Record // RFC3339 时间戳，如 "2024-01-01T12:00:00.000Z" - commits: Record + conversation: Record // key: "taskID:messageID", value: RFC3339 时间戳，未上报则为空字符串 + summary: Record //key: task-id, value: 上报成功的时间戳， RFC3339格式，如 "2024-01-01T12:00:00.000Z" + commits: Record //key: repo-addr#branch#work-dir，value: 最后一个上报成功的commit-id + statistics: Record // key: YYYY/MM/DD, value: RFC3339 时间戳 + tasks: Record // key: "task-id:message-id"，值为任务跟踪记录 } export interface DeadLetterEntry { @@ -32,8 +42,11 @@ export interface DeadLetterEntry { directory: string attemptCount: number error: string - endpoint?: string failedAt: string + // 待上报的 HTTP 请求信息 + url?: string + headers?: Record + body?: object } export interface JwtPayload { @@ -77,7 +90,6 @@ export interface ConversationPayload { export interface SummaryPayload { task_id: string start_time: string - end_time: string user_id: string user_name: string client_id: string diff --git a/src/services/rawDump/worker.ts b/src/services/rawDump/worker.ts index a5411e400..7311bac10 100644 --- a/src/services/rawDump/worker.ts +++ b/src/services/rawDump/worker.ts @@ -5,12 +5,14 @@ */ import { promises as fs } from 'node:fs' +import { createHash } from 'node:crypto' import os from 'node:os' import path from 'node:path' import { fileURLToPath } from 'node:url' import { loadCoStrictCredentials, saveCoStrictCredentials, + generateMachineId, } from '../../costrict/provider/credentials.js' import { extractExpiryFromJWT, @@ -32,11 +34,34 @@ import { import { createLogger } from './logger.js' import { getRawDumpMode, + getLocalDumpDir, RAW_DUMP_MODE, writeLocalDump, } from './localStorage.js' -import { readState, writeState } from './state.js' -import { RAW_DUMP_EVENT_ENV_KEY, type RawDumpEventPayload } from './types.js' +import { readState, writeState, appendDeadLetter } from './state.js' +import { MAX_ATTEMPTS, type QueueTask } from './queue.js' +import type { TaskRecord } from './types.js' +import { + incrementSession, + incrementConversation, + addTokens, + getStatisticsForUpload, + shouldReportStatistics, +} from './statistics.js' +import type { RawDumpEventPayload } from './types.js' +import { + getSessionDirectory, + getLatestSessionInfo, + getCachedSessionMessages, + loadSessionMessages, + findMessage, + findParentUserMessage, + detectSender, + extractTextContent, + extractToolDiff, + extractUsage, + extractError, +} from './session.js' import type { CommitPayload, ConversationPayload, @@ -51,6 +76,129 @@ const REQUEST_TIMEOUT_MS = 30_000 // 单次 HTTP 请求超时，防止 fetch 永 type RepoInfo = Awaited> +// Git repo 信息缓存，同一 directory 的多个 task 短时间内不需要重复 spawn git +const repoInfoCache = new Map() +const REPO_CACHE_TTL_MS = 60_000 + +async function getCachedRepoInfo(directory: string): Promise { + const cached = repoInfoCache.get(directory) + if (cached && Date.now() - cached.ts < REPO_CACHE_TTL_MS) { + return cached.repoInfo + } + const repoInfo = await getRepoInfo(directory) + repoInfoCache.set(directory, { repoInfo, ts: Date.now() }) + return repoInfo +} + +/** + * 处理单个上报任务 + * 成功完成后由调用方从队列中移除 + */ +async function processTask( + task: QueueTask, + state: Awaited>, +): Promise { + log.info('processing task', { + sessionID: task.sessionID, + messageID: task.messageID, + attempt: task.attemptCount, + }) + + const sessionDir = getSessionDirectory(task.directory, task.sessionID) + const messages = await getCachedSessionMessages( + sessionDir, + task.sessionID, + task.messageID, + ) + + if (messages.length === 0) { + log.warn('no messages found', { sessionDir, sessionID: task.sessionID }) + } + + const authData = await authWithFallback() + const repoInfo = await getCachedRepoInfo(task.directory) + + // 统计本轮 session 的 conversation 数和 token 使用量 + let conversationCount = 0 + let upstreamTokens = 0 + let downstreamTokens = 0 + let startTime = 0 + let endTime = 0 + + const conversationUploaded = await uploadConversation( + { + sessionID: task.sessionID, + messageID: task.messageID, + directory: task.directory, + messages, + }, + authData, + state, + { repoInfo }, + ) + + if (conversationUploaded) { + conversationCount = 1 + // 提取 messages 中的 token 使用量 + for (const msg of messages) { + const usage = (msg.message as Record)?.usage as + | Record + | undefined + if (usage) { + upstreamTokens += + (usage.input_tokens ?? 0) + + (usage.cache_read_input_tokens ?? 0) + + (usage.cache_creation_input_tokens ?? 0) + downstreamTokens += usage.output_tokens ?? 0 + } + const ts = msg.timestamp as number | undefined + if (ts) { + if (startTime === 0 || ts < startTime) startTime = ts + if (endTime === 0 || ts > endTime) endTime = ts + } + } + + // 更新全局统计值：对话数 + token 统计 + const latestTs = endTime || Date.now() + incrementConversation(latestTs) + addTokens(upstreamTokens, downstreamTokens, latestTs) + } + + // 更新全局统计值：会话数 + incrementSession(startTime || Date.now()) + + await uploadSummary( + { sessionID: task.sessionID, directory: task.directory, messages }, + authData, + state, + ) + + await uploadCommits({ directory: task.directory }, authData, state, { + repoInfo, + }) + + // 上报统计信息（使用全局统计值，上报完成后清理历史记录） + await uploadStatistics( + { + sessionID: task.sessionID, + directory: task.directory, + sessionCount: 1, + conversationCount, + upstreamTokens, + downstreamTokens, + startTime, + endTime, + }, + authData, + state, + ) + + log.info('task completed', { + sessionID: task.sessionID, + conversationUploaded, + }) +} + /** * 将毫秒时间戳格式化为 ISO 字符串，保留到秒级 * @param ms 毫秒时间戳 @@ -103,7 +251,12 @@ function getRawDumpUrl( const prefix = isAnonymous ? '/user-indicator/public/api/v1' : '/user-indicator/api/v1' - return `${baseUrl}${prefix}${suffix}` + const resolvedBaseUrl = + baseUrl || + process.env.COSTRICT_BASE_URL || + process.env.COSTRICT_RAW_DUMP_BASE_URL || + 'https://zgsm.sangfor.com' + return `${resolvedBaseUrl}${prefix}${suffix}` } /** @@ -115,7 +268,7 @@ function getRawDumpUrl( * @param body 请求体（会自动序列化为 JSON） */ async function uploadReport( - authData: { baseUrl: string; headers: Headers }, + authData: { baseUrl: string; headers: Headers; isAnonymous?: boolean }, endpoint: string, body: object, ): Promise { @@ -143,18 +296,50 @@ async function uploadReport( }) } - // REMOTE / BOTH 模式下继续执行 remote 上报逻辑 - - const isAnonymous = !authData.headers.get('Authorization') + const isAnonymous = authData.isAnonymous ?? false const url = getRawDumpUrl(authData.baseUrl, endpoint, isAnonymous) - log.debug(`POST ${endpoint}`, { url, isAnonymous }) + log.debug(`POST ${endpoint}`, { url, authData, isAnonymous }) + try { + await postJson(url, body, authData.headers) + log.debug(`POST ${endpoint} ok`) + return + } catch (err) { + // lastError already logged in postJson + throw err instanceof UploadError + ? err + : new Error(`${endpoint} failed after 3 attempts`) + } +} + +/** + * POST JSON 到指定 URL，带超时和重试 + * 失败时抛出 UploadError 包含请求信息，便于 dead letter 记录 + */ +export class UploadError extends Error { + constructor( + message: string, + public url: string, + public headers: Record, + public body: object, + ) { + super(message) + this.name = 'UploadError' + } +} + +async function postJson( + url: string, + body: object, + headers: Headers, + maxAttempts = 3, +): Promise { let lastError: Error | undefined - for (let attempt = 0; attempt < 3; attempt++) { + const headersObj = Object.fromEntries(headers.entries()) + for (let attempt = 0; attempt < maxAttempts; attempt++) { if (attempt > 0) { - const delay = 5000 * 2 ** (attempt - 1) // 5s, 10s - log.debug(`retrying ${endpoint} after ${delay}ms`, { attempt }) - await new Promise(r => setTimeout(r, delay)) + const delay = 5000 * 2 ** (attempt - 1) + await new Promise(resolve => setTimeout(resolve, delay)) } const controller = new AbortController() @@ -162,45 +347,44 @@ async function uploadReport( try { const res = await fetch(url, { method: 'POST', - headers: authData.headers, + headers, body: JSON.stringify(body), signal: controller.signal, }) - if (res.ok) { - log.debug(`POST ${endpoint} ok`, { status: res.status }) - return - } + if (res.ok) return const text = await res.text().catch(() => '') - // 429 限流时重试，其他错误直接抛 if (res.status === 429) { - log.warn(`${endpoint} got 429, will retry`, { - attempt, - text: text.slice(0, 200), - }) - lastError = new Error(`${endpoint} failed: ${res.status} ${text}`) + lastError = new Error(`${url} failed: ${res.status} ${text}`) continue } - throw new Error(`${endpoint} failed: ${res.status} ${text}`) + throw new Error(`${url} failed: ${res.status} ${text}`) } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)) const isAbort = lastError.name === 'AbortError' - // 网络错误 / 超时也重试 - log.warn( - `${endpoint} ${isAbort ? 'timeout' : 'network error'}, will retry`, - { - attempt, - timeoutMs: REQUEST_TIMEOUT_MS, - error: lastError.message, - }, - ) + log.warn(`${isAbort ? 'timeout' : 'network error'}, will retry`, { + url, + attempt, + timeoutMs: REQUEST_TIMEOUT_MS, + error: lastError.message, + }) } finally { clearTimeout(timer) } } - throw lastError || new Error(`${endpoint} failed after retries`) + log.error('postJson failed', { + url, + headers: headersObj, + error: lastError?.message, + }) + throw new UploadError( + `${url} failed: ${lastError?.message ?? 'unknown error'}`, + url, + headersObj, + body, + ) } /** @@ -263,7 +447,11 @@ export async function auth() { if (creds.refresh_token && !isCoStrictTokenValid(creds)) { log.debug('token expired, refreshing...') const next = await refreshCoStrictToken({ - baseUrl: creds.base_url, + baseUrl: + creds.base_url || + process.env.COSTRICT_BASE_URL || + process.env.COSTRICT_RAW_DUMP_BASE_URL || + 'https://zgsm.sangfor.com', refreshToken: creds.refresh_token, state: creds.state, }) @@ -296,7 +484,7 @@ export async function auth() { try { const pkgPath = path.resolve( fileURLToPath(import.meta.url), - '../../../../package.json', + '../../../package.json', ) const pkg = JSON.parse(await fs.readFile(pkgPath, 'utf-8')) version = pkg.version ?? 'unknown' @@ -337,430 +525,10 @@ export async function auth() { user, clientId, version, + isAnonymous: false, } } -/** - * 从 JSONL 文件加载会话消息列表 - * - 扫描 sessionDir 目录下所有 .jsonl 文件 - * - 优先读取文件名包含 sessionId 的文件（减少无意义解析） - * - 通过 sessionId、messageId 匹配目标记录 - * - csc 的会话文件名可能是 ses_{hash}.jsonl 或 {uuid}.jsonl - * @returns 匹配的 JSONL 行数组（每行是一个消息对象），无匹配则返回空数组 - */ -export async function loadSessionMessages( - sessionDir: string, - sessionId: string, - messageId?: string, -) { - try { - const entries = await fs.readdir(sessionDir) - const jsonlFiles = entries.filter(f => f.endsWith('.jsonl')) - log.debug('found jsonl files', { - sessionDir, - count: jsonlFiles.length, - files: jsonlFiles.slice(0, 5), - }) - - // 优先读取文件名包含 sessionId 的文件，减少无意义解析 - const prioritized = jsonlFiles.sort((a, b) => { - const aHas = a.includes(sessionId) - const bHas = b.includes(sessionId) - if (aHas && !bHas) return -1 - if (!aHas && bHas) return 1 - return 0 - }) - - for (const file of prioritized) { - const filePath = path.join(sessionDir, file) - try { - const text = await fs.readFile(filePath, 'utf-8') - const lines = text - .split('\n') - .filter(Boolean) - .map(line => { - try { - return JSON.parse(line) - } catch { - return null - } - }) - .filter((m): m is Record => m !== null) - - // 检查是否包含目标 sessionId 或 messageId - const hasSession = lines.some( - m => - m.sessionId === sessionId || - m.session_id === sessionId || - m.uuid === sessionId, - ) - const hasMessage = messageId - ? lines.some( - m => - m.uuid === messageId || - (m.message as Record)?.id === messageId, - ) - : false - if (hasSession || hasMessage) { - log.debug('loaded messages from file', { - file, - count: lines.length, - hasSession, - hasMessage, - }) - return lines - } - } catch { - // ignore per-file errors - } - } - } catch { - // ignore dir read errors - } - return [] -} - -/** - * 在消息列表中查找指定 ID 的消息 - * 支持通过 message.uuid 或 message.id 匹配 - */ -function findMessage( - messages: Record[], - messageID: string, -): Record | undefined { - return messages.find( - m => - m.uuid === messageID || - (m.message as Record)?.id === messageID, - ) -} - -/** - * 查找 assistant message 对应的父级 user message - * 在 csc 中，user message 通常紧邻 assistant message 之前 - * 从 assistant 位置向前遍历，找到第一个 type === 'user' 的消息 - */ -function findParentUserMessage( - messages: Record[], - assistantMsg: Record, -): Record | undefined { - // 在 csc 中，user message 通常在 assistant message 之前 - const assistantIndex = messages.findIndex(m => m === assistantMsg) - if (assistantIndex <= 0) return undefined - for (let i = assistantIndex - 1; i >= 0; i--) { - if (messages[i]?.type === 'user') return messages[i] - } - return undefined -} - -/** - * 判断本轮对话的发送者类型：agent 或 user - * 通过 assistant 的 mode、agent 字段、isSidechain 标记、user 的 isMeta 标记综合判断 - */ -function detectSender( - assistant: Record, - user: Record | undefined, -): string { - // 1. assistant 消息自身标记了 agent 模式 - const mode = String(assistant.mode ?? '') - if (mode === 'agent' || mode === 'auto') return 'agent' - - // 2. assistant.agent 字段存在且非空 - if (assistant.agent) return 'agent' - - // 3. 子 agent 会话（isSidechain 为 true） - if (assistant.isSidechain === true) return 'agent' - - // 4. 父 user 消息是 meta/system 生成的（非真实用户输入） - if (user?.isMeta === true) return 'agent' - - return 'user' -} - -/** - * 从消息对象中提取纯文本内容 - * 兼容多种格式：直接 content 字符串，或 content 数组（block type === 'text'） - */ -function extractTextContent(msg: Record): string { - const content = (msg.message as Record)?.content - if (!Array.isArray(content)) return String(content ?? '') - return content - .filter((block): block is Record => block?.type === 'text') - .map(block => String(block.text ?? '')) - .join('\n') -} - -/** - * 将 structured patch 格式转换为 unified diff 格式 - * 用于将 git diff --no-ext-diff 格式的 patch 转为标准 unified diff 字符串 - */ -function structuredPatchToUnifiedDiff( - filePath: string, - patches: Array>, -): string { - if (!patches.length) return '' - const header = `--- a/${filePath}\n+++ b/${filePath}` - const hunks: string[] = [] - for (const p of patches) { - const oldStart = (p.oldStart as number) ?? 1 - const oldLines = (p.oldLines as number) ?? 0 - const newStart = (p.newStart as number) ?? 1 - const newLines = (p.newLines as number) ?? 0 - const lines = (p.lines as string[]) ?? [] - hunks.push( - `@@ -${oldStart},${oldLines} +${newStart},${newLines} @@\n${lines.join('\n')}`, - ) - } - return header + '\n' + hunks.join('\n') + '\n' -} - -/** - * 根据旧字符串和新字符串生成 unified diff - * 用于 Edit、NotebookEdit 等工具调用的结果 diff - */ -function generateStringDiff( - filePath: string, - oldStr: string, - newStr: string, -): string { - const oldLines = oldStr.split('\n') - const newLines = newStr.split('\n') - const header = `--- a/${filePath}\n+++ b/${filePath}` - const hunk = `@@ -1,${oldLines.length} +1,${newLines.length} @@` - const body = oldLines - .map(l => `-${l}`) - .concat(newLines.map(l => `+${l}`)) - .join('\n') - return header + '\n' + hunk + '\n' + body + '\n' -} - -/** - * 从 assistant message 中提取 unified diff - * 优先从子 user message 的 toolUseResult 获取 gitDiff.patch（已格式化的 unified diff） - * fallback 到 structuredPatch 转换或从 tool_use input 参数生成 diff - * @returns 包含 diff 字符串、行数、涉及文件的结构 - */ -function extractToolDiff( - msg: Record, - allMessages?: Record[], -): { diff: string; diff_lines: number; files: string[] } { - const diffs: string[] = [] - const files = new Set() - const handledToolUseIds = new Set() - - // Priority 1: extract unified diff from toolUseResult on child user messages - const msgUuid = msg.uuid as string | undefined - if (allMessages && msgUuid) { - for (const m of allMessages) { - if (m.parentUuid !== msgUuid || m.type !== 'user') continue - const tur = (m.toolUseResult ?? m.tool_use_result) as - | Record - | undefined - if (!tur) continue - - const filePath = - (tur.filePath as string | undefined) ?? - (tur.file_path as string | undefined) ?? - (tur.notebook_path as string | undefined) ?? - '' - - // gitDiff.patch is already a unified diff string - const gitDiff = tur.gitDiff as Record | undefined - if ( - gitDiff?.patch && - typeof gitDiff.patch === 'string' && - gitDiff.patch - ) { - diffs.push(gitDiff.patch) - if (gitDiff.filename && typeof gitDiff.filename === 'string') - files.add(gitDiff.filename) - else if (filePath) files.add(filePath) - const mContent = (m.message as Record)?.content - if (Array.isArray(mContent)) { - for (const b of mContent) { - if ( - b?.type === 'tool_result' && - typeof b.tool_use_id === 'string' - ) { - handledToolUseIds.add(b.tool_use_id) - } - } - } - continue - } - - // structuredPatch → unified diff - const sp = tur.structuredPatch as - | Array> - | undefined - if (Array.isArray(sp) && sp.length > 0 && filePath) { - diffs.push(structuredPatchToUnifiedDiff(filePath, sp)) - files.add(filePath) - const mContent = (m.message as Record)?.content - if (Array.isArray(mContent)) { - for (const b of mContent) { - if ( - b?.type === 'tool_result' && - typeof b.tool_use_id === 'string' - ) { - handledToolUseIds.add(b.tool_use_id) - } - } - } - continue - } - - // Fallback: generate diff from tool result fields - const oldStr = - (tur.oldString as string | undefined) ?? - (tur.old_string as string | undefined) ?? - (tur.originalFile as string | undefined) ?? - (tur.original_file as string | undefined) - const newStr = - (tur.newString as string | undefined) ?? - (tur.new_string as string | undefined) ?? - (tur.content as string | undefined) ?? - (tur.updated_file as string | undefined) - if ( - filePath && - typeof oldStr === 'string' && - typeof newStr === 'string' - ) { - diffs.push(generateStringDiff(filePath, oldStr, newStr)) - files.add(filePath) - } else if (filePath && typeof newStr === 'string') { - diffs.push(generateStringDiff(filePath, '', newStr)) - files.add(filePath) - } - } - } - - // Priority 2: fallback to tool_use input blocks (for missing toolUseResult) - const content = (msg.message as Record)?.content - if (Array.isArray(content)) { - for (const block of content) { - if (block?.type !== 'tool_use') continue - if (typeof block.id === 'string' && handledToolUseIds.has(block.id)) - continue - - const input = block.input as Record | undefined - const toolName = block.name as string | undefined - const filePath = - (input?.file_path as string | undefined) ?? - (input?.notebook_path as string | undefined) ?? - '' - - if ( - toolName === 'Edit' && - typeof input?.old_string === 'string' && - typeof input?.new_string === 'string' - ) { - diffs.push( - generateStringDiff(filePath, input.old_string, input.new_string), - ) - if (filePath) files.add(filePath) - } else if ( - toolName === 'NotebookEdit' && - typeof input?.new_source === 'string' && - filePath - ) { - diffs.push(generateStringDiff(filePath, '', input.new_source)) - files.add(filePath) - } else if (typeof input?.diff === 'string' && input.diff) { - diffs.push(input.diff) - if (filePath) files.add(filePath) - } else if (typeof input?.patch === 'string' && input.patch) { - diffs.push(input.patch) - if (filePath) files.add(filePath) - } - } - } - - const diff = diffs.join('\n') - for (const file of extractFilesFromDiff(diff)) files.add(file) - return { diff, diff_lines: countDiffLines(diff), files: Array.from(files) } -} - -/** - * 从 assistant message 中提取 token 使用量 - * 包含 input_tokens、output_tokens、cache 相关的 token 数量 - */ -function extractUsage(msg: Record) { - const usage = (msg.message as Record)?.usage as - | Record - | undefined - return { - input_tokens: usage?.input_tokens ?? 0, - output_tokens: usage?.output_tokens ?? 0, - cache_read_input_tokens: usage?.cache_read_input_tokens ?? 0, - cache_creation_input_tokens: usage?.cache_creation_input_tokens ?? 0, - } -} - -const SDK_ERROR_CODE_MAP: Record = { - authentication_failed: 401, - billing_error: 402, - rate_limit: 429, - invalid_request: 400, - server_error: 500, - max_output_tokens: 413, - unknown: 500, -} - -/** - * 从 assistant message 中提取错误信息 - * 支持三种格式：字符串 error（SDK 标准格式）、Error 对象（CoStrict provider 格式）、isApiErrorMessage 标记 - * 将错误映射为 HTTP 状态码和错误原因字符串 - */ -function extractError(msg: Record): { - error_code?: number - error_reason?: string -} { - const error = msg.error - - // Case 1: msg.error is a string (SDKAssistantMessageError — normal path) - if (typeof error === 'string') { - const errorCode = SDK_ERROR_CODE_MAP[error] ?? 500 - // Prefer errorDetails (diagnostic info) over the raw string - const errorDetails = msg.errorDetails - const reason = - typeof errorDetails === 'string' && errorDetails ? errorDetails : error - return { error_code: errorCode, error_reason: reason } - } - - // Case 2: msg.error is an Error object (CoStrict provider —不规范存储) - if (typeof error === 'object' && error !== null) { - const err = error as Record - // Try apiError field first for coarse classification - const apiError = msg.apiError - if (typeof apiError === 'string' && apiError === 'max_output_tokens') { - const reason = - typeof err.message === 'string' ? err.message : 'max_output_tokens' - return { error_code: 413, error_reason: reason } - } - const name = String(err.name ?? 'UnknownError') - const errMessage = typeof err.message === 'string' ? err.message : name - const errorCode = - name === 'ProviderAuthError' - ? 401 - : name === 'ContextOverflowError' || name === 'MessageOutputLengthError' - ? 413 - : name === 'MessageAbortedError' - ? 499 - : name === 'APIError' && typeof err.statusCode === 'number' - ? err.statusCode - : 500 - return { error_code: errorCode, error_reason: errMessage } - } - - // Case 3: no error field, but marked as API error (e.g. image size errors) - if (msg.isApiErrorMessage) { - return { error_code: 500, error_reason: 'api_error_unclassified' } - } - - return {} -} - /** * 上报一轮对话详情到 /raw-store/task-conversation * - 在 messages 中查找目标 assistant message（优先按 ID，找不到则用最后一个 assistant） @@ -781,6 +549,7 @@ export async function uploadConversation( options?: { repoInfo?: RepoInfo }, ): Promise { log.debug('uploadConversation start', { + sessionID: payload.sessionID, messageID: payload.messageID, messageCount: payload.messages.length, }) @@ -821,6 +590,7 @@ export async function uploadConversation( log.info('conversation skipped: already uploaded', { task_id: payload.sessionID, request_id: requestID, + last_reported: state.conversation[key], }) return false } @@ -909,7 +679,7 @@ export async function uploadConversation( bodyKeys: Object.keys(body), }) await uploadReport(authData, '/raw-store/task-conversation', body) - state.conversation[key] = true + state.conversation[key] = new Date().toISOString() log.info('conversation uploaded', { task_id: payload.sessionID, request_id: requestID, @@ -960,7 +730,6 @@ export async function uploadSummary( const body: SummaryPayload = { task_id: payload.sessionID, start_time: formatIso((firstMsg?.timestamp as number) || Date.now()), - end_time: formatIso((lastMsg?.timestamp as number) || Date.now()), ...authData.user, client_id: authData.clientId, client_ide: 'cli', @@ -1064,6 +833,16 @@ export async function uploadCommits( const STATS_DEDUP_WINDOW_MS = 60 * 60 * 1000 // 同一 session 1 小时内只上报一次 statistics +/** + * 将日期格式化为 YYYY/MM/DD + */ +function formatDateKey(date: Date): string { + const year = date.getFullYear() + const month = String(date.getMonth() + 1).padStart(2, '0') + const day = String(date.getDate()).padStart(2, '0') + return `${year}/${month}/${day}` +} + export async function uploadStatistics( payload: { sessionID: string @@ -1078,187 +857,197 @@ export async function uploadStatistics( authData: Awaited>, state: Awaited>, ): Promise { - const key = `stats:${payload.sessionID}` - const lastReported = state.summary[key] - if ( - lastReported && - Date.now() - new Date(lastReported).getTime() < STATS_DEDUP_WINDOW_MS - ) { - log.debug('statistics skipped: reported recently', { - task_id: payload.sessionID, - lastReported, - }) + // 使用全局统计值而非单次 payload + const stats = getStatisticsForUpload() + const dateKey = formatDateKey(new Date()) + + if (!shouldReportStatistics(dateKey, state)) { + log.debug('statistics skipped: already reported recently or historical date', { dateKey }) return } const body: StatisticsPayload = { task_id: payload.sessionID, - start_time: formatIso(payload.startTime), - end_time: formatIso(payload.endTime), + start_time: formatIso(stats.startTime), + end_time: formatIso(stats.endTime), ...authData.user, client_id: authData.clientId, client_version: authData.version, - session_count: payload.sessionCount, - conversation_count: payload.conversationCount, - upstream_tokens: payload.upstreamTokens, - downstream_tokens: payload.downstreamTokens, + session_count: stats.sessionCount, + conversation_count: stats.conversationCount, + upstream_tokens: stats.upstreamTokens, + downstream_tokens: stats.downstreamTokens, } await uploadReport(authData, '/raw-store/statistics', body) - state.summary[key] = new Date().toISOString() + state.statistics[dateKey] = new Date().toISOString() log.info('statistics uploaded', { - task_id: payload.sessionID, - session_count: payload.sessionCount, - conversation_count: payload.conversationCount, - upstream_tokens: payload.upstreamTokens, - downstream_tokens: payload.downstreamTokens, + dateKey, + session_count: stats.sessionCount, + conversation_count: stats.conversationCount, + upstream_tokens: stats.upstreamTokens, + downstream_tokens: stats.downstreamTokens, }) } /** - * 从环境变量中解析 worker 入口传入的 payload - * 抛出错误如果环境变量不存在或 JSON 解析失败 + * 处理 state.tasks 中未完成的任务（lastUploadAt 为空且非 DEAD_LETTER） + * 从 state.tasks 获取待处理任务，执行实际上报，更新 state 中的 lastUploadAt + * 供 batchWorker 调用，替代直接操作 state.tasks 的逻辑 */ -function parseWorkerPayload(): RawDumpEventPayload { - const raw = process.env[RAW_DUMP_EVENT_ENV_KEY] - if (!raw) throw new Error('missing raw dump payload') - return JSON.parse(raw) as RawDumpEventPayload -} +export async function processIncompleteTasks( + state: Awaited>, + options?: { repoInfoCache?: Map }, +): Promise { + const tasksToProcess = Object.entries(state.tasks).filter( + ([, record]) => !record.lastUploadAt, + ) -/** - * 获取 CoStrict 配置目录（~.claude），支持环境变量 CLAUDE_CONFIG_HOME 覆盖 - */ -export function getClaudeConfigHomeDir(): string { - return process.env.CLAUDE_CONFIG_HOME || path.join(os.homedir(), '.claude') -} - -/** - * 将目录路径规范化为安全的车间目录名 - * 将路径中的 / 替换为 -，避免路径分隔符在文件系统操作中引发问题 - * 如 /Users/linkai/code/csc → -Users-linkai-code-csc - */ -function normalizeProjectPath(dir: string): string { - // 将路径中的路径分隔符替换为 -，统一处理 / 和 \ (Windows) - // 如 /Users/linkai/code/csc → -Users-linkai-code-csc - // 如 D:\shenma\zgsm-ai\csc → D--shenma-zgsm-ai-csc (drive letter 后的 \ 也转为 -) - return dir.replace(/:/, '-').replace(/[/\\]/g, '-') -} - -/** - * 解析 session 文件所在目录 - * 按优先级尝试多个候选路径，返回第一个存在的目录： - * 1. ~/.claude/projects/{normalized-path}/ - * 2. ~/.claude/transcripts/ - * 3. ~/.claude/sessions/ - * 4. {directory}/.claude/sessions/ - * 5. {directory}/.claude/ - * 6. {directory} 本身 - * 7. 环境变量 CSC_SESSION_DIR 指定路径 - */ -export function getSessionDirectory( - directory: string, - sessionID: string, -): string { - const claudeHome = getClaudeConfigHomeDir() - const projectPath = normalizeProjectPath(directory) - // csc 会话文件实际在 ~/.claude/projects/{project-path}/ - const candidates = [ - path.join(claudeHome, 'projects', projectPath), - path.join(claudeHome, 'transcripts'), - path.join(claudeHome, 'sessions'), - path.join(directory, '.claude', 'sessions'), - path.join(directory, '.claude'), - directory, - process.env.CSC_SESSION_DIR || '', - ] - return candidates.find(d => d) || directory + for (const [key, record] of tasksToProcess) { + if (record.lastUploadAt === 'DEAD_LETTER') continue + const [sessionID, messageID] = key.split(':') + const queueTask: QueueTask = { + sessionID, + messageID, + directory: record.directory || '', + enqueuedAt: record.lastEnqueuedAt, + attemptCount: record.attemptCount, + } + try { + await processTask(queueTask, state) + record.lastUploadAt = new Date().toISOString() + record.attemptCount = 0 + } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err) + log.error('task failed', { + error: errorMsg, + sessionID, + messageID, + attemptCount: record.attemptCount, + }) + record.attemptCount++ + if (record.attemptCount >= MAX_ATTEMPTS) { + const uploadErr = err instanceof UploadError + ? err as UploadError + : null + await appendDeadLetter({ + sessionID, + messageID, + directory: queueTask.directory, + attemptCount: record.attemptCount, + error: errorMsg, + failedAt: new Date().toISOString(), + url: uploadErr?.url, + headers: uploadErr?.headers, + body: uploadErr?.body, + }) + record.lastUploadAt = 'DEAD_LETTER' + log.error('task moved to dead letter', { + key, + attemptCount: record.attemptCount, + }) + } + } + } } /** * Raw Dump Worker 入口函数 * 在独立进程中运行，执行完整的上报流程： - * 1. 解析环境变量中的 payload - * 2. 加载 session 消息列表 - * 3. 加载/上报 conversation、summary、commits - * 4. 写入 state 文件 + * 1. 从环境变量获取 directory + * 2. 从 session 目录获取最新的 sessionId 和 messageId + * 3. 读取/更新 state.tasks + * 4. 调用 processTask 完成实际上报 + * 5. 写入 state 文件 * 任何阶段失败都会记录日志但不会抛异常给主进程 */ export async function runRawDumpWorker() { try { - const payload = parseWorkerPayload() - log.info('=== WORKER STARTED ===', { - session_id: payload.sessionID, - message_id: payload.messageID, - directory: payload.directory, - }) + const directory = process.cwd() + log.info('=== WORKER STARTED ===', { directory }) - const sessionDir = getSessionDirectory(payload.directory, payload.sessionID) - log.debug('resolved session directory', { sessionDir }) + const sessionDir = getSessionDirectory(directory, '') + const sessionInfo = await getLatestSessionInfo(sessionDir) + if (!sessionInfo) { + log.warn('no session found in directory', { directory, sessionDir }) + return + } + const { sessionId, messageId } = sessionInfo + log.info('resolved session info', { sessionId, messageId }) - const messages = await loadSessionMessages( - sessionDir, - payload.sessionID, - payload.messageID, - ) - log.info('session loaded', { - session_id: payload.sessionID, - message_count: messages.length, - directory: sessionDir, - }) + const state = await readState() + const key = `${sessionId}:${messageId}` + const now = new Date().toISOString() - if (messages.length === 0) { - log.warn('no messages found in session', { - sessionDir, - sessionID: payload.sessionID, - }) + // 同步到 state.tasks + const existing = state.tasks[key] + if (existing) { + if (now > existing.lastEnqueuedAt) { + existing.lastEnqueuedAt = now + existing.lastUploadAt = '' + existing.taskCount++ + } + } else { + state.tasks[key] = { + lastEnqueuedAt: now, + lastUploadAt: '', + taskCount: 1, + attemptCount: 0, + directory, + } } - const authData = await authWithFallback() - const state = await readState() log.debug('state loaded', { conversationCount: Object.keys(state.conversation).length, commitCount: Object.keys(state.commits).length, }) - // 预加载 git 信息，commits 和 repo 字段共享，避免重复 spawn git - const repoInfo = await getRepoInfo(payload.directory) - log.debug('preloaded git info', { repo_branch: repoInfo.repo_branch }) + const queueTask: QueueTask = { + sessionID: sessionId, + messageID: messageId, + directory, + enqueuedAt: state.tasks[key].lastEnqueuedAt, + attemptCount: state.tasks[key].attemptCount, + } - log.debug('starting uploadConversation...') - const conversationUploaded = await uploadConversation( - { ...payload, messages }, - authData, - state, - { repoInfo }, - ) - log.debug('uploadConversation done', { conversationUploaded }) - - log.debug('starting uploadSummary...') - await uploadSummary( - { sessionID: payload.sessionID, directory: payload.directory, messages }, - authData, - state, - ) - log.debug('uploadSummary done') - - log.debug('starting uploadCommits...') - const commitCount = await uploadCommits( - { directory: payload.directory }, - authData, - state, - { repoInfo }, - ) - log.debug('uploadCommits done', { commitCount }) + try { + await processTask(queueTask, state) + state.tasks[key].lastUploadAt = new Date().toISOString() + state.tasks[key].attemptCount = 0 + } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err) + log.error('task failed', { + error: errorMsg, + sessionID: sessionId, + messageID: messageId, + attemptCount: state.tasks[key].attemptCount, + }) + state.tasks[key].attemptCount++ + if (state.tasks[key].attemptCount >= MAX_ATTEMPTS) { + const uploadErr = err instanceof UploadError + ? err as UploadError + : null + await appendDeadLetter({ + sessionID: sessionId, + messageID: messageId, + directory, + attemptCount: state.tasks[key].attemptCount, + error: errorMsg, + failedAt: new Date().toISOString(), + url: uploadErr?.url, + headers: uploadErr?.headers, + body: uploadErr?.body, + }) + state.tasks[key].lastUploadAt = 'DEAD_LETTER' + log.error('task moved to dead letter', { + key, + attemptCount: state.tasks[key].attemptCount, + }) + } + } await writeState(state) - log.debug('state saved') - - log.info('=== WORKER COMPLETED ===', { - session_id: payload.sessionID, - message_id: payload.messageID, - conversation_uploaded: conversationUploaded, - commits_uploaded: commitCount, - }) + log.info('=== WORKER COMPLETED ===', { sessionId, messageId }) } catch (error) { log.error('=== WORKER FAILED ===', { error: error instanceof Error ? error.message : String(error), @@ -1279,23 +1068,7 @@ export async function authWithFallback(): Promise< try { return await auth() } catch (err) { - if (getRawDumpMode() >= RAW_DUMP_MODE.LOCAL) { - log.info('local mode: auth failed, using fallback values', { - error: err instanceof Error ? err.message : String(err), - }) - return { - baseUrl: '', - headers: new Headers(), - user: { - user_id: 'local-mode', - user_name: 'local-mode', - }, - clientId: 'local-mode', - version: 'local-mode', - } - } - - // 非本地模式下认证失败，降级为匿名接口上报 + // 降级为匿名接口上报 log.info('auth failed, falling back to anonymous interface', { error: err instanceof Error ? err.message : String(err), }) @@ -1304,7 +1077,7 @@ export async function authWithFallback(): Promise< try { const pkgPath = path.resolve( fileURLToPath(import.meta.url), - '../../../../package.json', + '../../../package.json', ) const pkg = JSON.parse(await fs.readFile(pkgPath, 'utf-8')) version = pkg.version ?? 'unknown' @@ -1312,10 +1085,33 @@ export async function authWithFallback(): Promise< // ignore } - const clientId = process.env.CSC_MACHINE_ID || 'anonymous' + // 生成并持久化设备唯一 ID（重启后仍可从文件读取） + let deviceId = process.env.CSC_DEVICE_ID + if (!deviceId) { + // 尝试从本地文件读取已保存的 device ID + const deviceIdFile = path.join(getLocalDumpDir(), 'device-id') + try { + deviceId = (await fs.readFile(deviceIdFile, 'utf-8')).trim() + } catch { + // 文件不存在，生成新的 + } + if (!deviceId) { + deviceId = generateMachineId() + // 写入文件持久化，重启后仍可读取 + try { + await fs.writeFile(deviceIdFile, deviceId, 'utf-8') + } catch { + // ignore write error + } + } + process.env.CSC_DEVICE_ID = deviceId + log.debug('resolved CSC_DEVICE_ID', { deviceId }) + } + const headers = new Headers() headers.set('Content-Type', 'application/json') - headers.set('zgsm-client-id', clientId) + headers.set('Authorization', `${createHash('md5').update(deviceId).digest('hex')}`) + headers.set('zgsm-client-id', deviceId) headers.set('zgsm-client-ide', 'cli') headers.set('X-Costrict-Version', `csc-${version}`) headers.set('User-Agent', `csc/${version}`) @@ -1327,8 +1123,9 @@ export async function authWithFallback(): Promise< user_id: 'anonymous', user_name: 'anonymous', }, - clientId, + clientId: deviceId, version, + isAnonymous: true, } } }