From 2e6dd386032e019d782dc1b2630e37686ed20ec3 Mon Sep 17 00:00:00 2001 From: zbc Date: Wed, 27 May 2026 16:11:14 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=AF=E5=8A=A8=E6=97=B6=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E5=88=9B=E5=BB=BAraw-dump=E7=9B=AE=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/services/rawDump/README.md | 17 +++++++-- src/services/rawDump/batchWorker.ts | 2 +- src/services/rawDump/index.ts | 17 ++++++--- src/services/rawDump/localStorage.ts | 22 ++++++++--- src/services/rawDump/queue.ts | 11 ++---- src/services/rawDump/worker.ts | 56 ++++++++++------------------ 6 files changed, 66 insertions(+), 59 deletions(-) diff --git a/src/services/rawDump/README.md b/src/services/rawDump/README.md index 1cdd77aef..97484edc5 100644 --- a/src/services/rawDump/README.md +++ b/src/services/rawDump/README.md @@ -368,9 +368,11 @@ export CSC_RAW_DUMP_DIR=/tmp/raw-dump-debug ### 状态文件 ``` ~/.claude/raw-dump/csc-state.json +~/.claude/raw-dump/csc-work-queue.jsonl +~/.claude/raw-dump/csc-dead-letter.jsonl ``` -内容格式： +**csc-state.json** — 去重状态： ```json { "conversation": { @@ -378,7 +380,7 @@ export CSC_RAW_DUMP_DIR=/tmp/raw-dump-debug "session-id-1:msg-uuid-2": true }, "summary": { - "session-id-1": 1747123456789 + "session-id-1": "2026-05-12T10:30:00.000Z" }, "commits": { "git@github.com:org/repo.git#main#/Users/xxx/code/repo": "abc123def" @@ -386,6 +388,12 @@ export CSC_RAW_DUMP_DIR=/tmp/raw-dump-debug } ``` +**csc-dead-letter.jsonl** — 上报失败记录（超过最大重试次数的任务），每行一个 JSON 对象： +```jsonl +{"sessionID":"...","messageID":"...","directory":"...","attemptCount":4,"error":"...","endpoint":"/raw-store/task-conversation","failedAt":"2026-05-12T10:30:00.000Z"} +{"sessionID":"...","messageID":"__summary__","directory":"...","attemptCount":4,"error":"...","endpoint":"/raw-store/task-summary","failedAt":"2026-05-12T10:31:00.000Z"} +``` + ### 日志文件 ``` ~/.claude/raw-dump/csc-raw-dump.log @@ -461,7 +469,10 @@ tail -f ~/.claude/raw-dump/csc-raw-dump.log cat ~/.claude/raw-dump/csc-state.json # 查看队列文件 -cat ~/.claude/raw-dump/csc-work-queue.json +cat ~/.claude/raw-dump/csc-work-queue.jsonl + +# 查看失败记录（dead letter） +cat ~/.claude/raw-dump/csc-dead-letter.jsonl # 查看是否有 worker 在运行（锁文件） cat ~/.claude/raw-dump/csc-work-queue.lock diff --git a/src/services/rawDump/batchWorker.ts b/src/services/rawDump/batchWorker.ts index 7b83d65e5..fceeafcb8 100644 --- a/src/services/rawDump/batchWorker.ts +++ b/src/services/rawDump/batchWorker.ts @@ -220,7 +220,7 @@ async function runBatch() { } const uniqueTasks = Array.from(deduped.values()).sort( - (a, b) => a.enqueuedAt - b.enqueuedAt, + (a, b) => a.enqueuedAt.localeCompare(b.enqueuedAt), ) log.info(`deduped to ${uniqueTasks.length} unique tasks`) diff --git a/src/services/rawDump/index.ts b/src/services/rawDump/index.ts index 0e1a9ec16..4e50465ee 100644 --- a/src/services/rawDump/index.ts +++ b/src/services/rawDump/index.ts @@ -4,6 +4,7 @@ */ import { + ensureRawDumpDirCreated, getRawDumpMode, RAW_DUMP_MODE, } from './localStorage.js' @@ -84,13 +85,14 @@ function shouldEnqueue(sessionID: string, messageID: string): boolean { * 上报一轮对话 * 只写入队列，由 batch worker 顺序消费 */ -export function reportTurn( +export async function reportTurn( sessionID: string, messageID: string, directory: string, -): void { +): Promise { if (!isEnabled()) return if (!shouldEnqueue(sessionID, messageID)) return + ensureRawDumpDirCreated() enqueue({ sessionID, messageID, directory }) ensureBatchWorker() } @@ -100,9 +102,13 @@ export function reportTurn( * 使用特殊 messageID '__summary__' 标识，与普通 conversation 分开去重 * 只写入队列，由 batch worker 顺序消费 */ -export function reportSession(sessionID: string, directory: string): void { +export async function reportSession( + sessionID: string, + directory: string, +): Promise { if (!isEnabled()) return if (!shouldEnqueue(sessionID, '__summary__')) return + ensureRawDumpDirCreated() enqueue({ sessionID, messageID: '__summary__', directory }) ensureBatchWorker() } @@ -123,11 +129,11 @@ const STATS_DEBOUNCE_MS = 60_000 // 同一 session 1 分钟内不重复 enqueue * 上报对账统计数据（session数、conversation数、token数） * 通过特殊 messageID '__statistics__' 标识入队 */ -export function reportStatistics( +export async function reportStatistics( sessionID: string, directory: string, data: StatisticsData, -): void { +): Promise { if (!isEnabled()) return const key = `${sessionID}:__statistics__` const now = Date.now() @@ -137,6 +143,7 @@ export function reportStatistics( return } lastReportStatsMap.set(key, now) + ensureRawDumpDirCreated() enqueue({ sessionID, messageID: '__statistics__', diff --git a/src/services/rawDump/localStorage.ts b/src/services/rawDump/localStorage.ts index 3bfc7d780..214d667a0 100644 --- a/src/services/rawDump/localStorage.ts +++ b/src/services/rawDump/localStorage.ts @@ -24,15 +24,21 @@ export const RAW_DUMP_MODE = { BOTH: 3, } as const +let rawDumpDirCreated = false + +export async function ensureRawDumpDirCreated(): Promise { + if (rawDumpDirCreated) return + const RAW_DUMP_DIR = path.join(os.homedir(), '.claude', 'raw-dump') + await fs.mkdir(RAW_DUMP_DIR, { recursive: true }) + rawDumpDirCreated = true +} + /** * 获取本地存储目录路径 * 支持环境变量 CSC_RAW_DUMP_DIR 覆盖 */ export function getLocalDumpDir(): string { - return (process.env.CSC_RAW_DUMP_DIR || DEFAULT_LOCAL_DIR).replace( - /\/$/, - '', - ) + return (process.env.CSC_RAW_DUMP_DIR || DEFAULT_LOCAL_DIR).replace(/\/$/, '') } /** @@ -62,13 +68,17 @@ export async function writeLocalDump( body: Record, ): Promise { const dir = getLocalDumpDir() - const taskId = (body.task_id as string) || (body.commit_id as string)|| 'unknown' + const taskId = + (body.task_id as string) || (body.commit_id as string) || 'unknown' const taskDir = path.join(dir, type, taskId) await fs.mkdir(taskDir, { recursive: true }) const timestamp = new Date().toISOString().replace(/[:.]/g, '-').slice(0, -5) const requestId = - (body.request_id as string) || (body.commit_id as string) || (body.task_id as string) || 'unknown' + (body.request_id as string) || + (body.commit_id as string) || + (body.task_id as string) || + 'unknown' const filename = `${timestamp}-${requestId}.json` const filePath = path.join(taskDir, filename) diff --git a/src/services/rawDump/queue.ts b/src/services/rawDump/queue.ts index 42108a768..e3f0ffd78 100644 --- a/src/services/rawDump/queue.ts +++ b/src/services/rawDump/queue.ts @@ -14,12 +14,7 @@ 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 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') export const MAX_ATTEMPTS = 4 // 最多尝试次数 @@ -28,7 +23,7 @@ export interface QueueTask { sessionID: string messageID: string directory: string - enqueuedAt: number + enqueuedAt: string // RFC3339 format, e.g. "2026-05-27T10:00:00.000Z" attemptCount: number statsData?: StatisticsData } @@ -74,7 +69,7 @@ export async function flushQueue(): Promise { * 写入失败被静默吞掉，不影响主流程 */ export function enqueue(task: Omit): void { - const item: QueueTask = { ...task, enqueuedAt: Date.now(), attemptCount: 0 } + const item: QueueTask = { ...task, enqueuedAt: new Date().toISOString(), attemptCount: 0 } queue.push(item) // 同步写文件，不阻塞主进程 event loop fs.writeFile(QUEUE_FILE, JSON.stringify(item) + '\n', { flag: 'a', encoding: 'utf-8' }).catch(() => {}) diff --git a/src/services/rawDump/worker.ts b/src/services/rawDump/worker.ts index ac395c039..a5411e400 100644 --- a/src/services/rawDump/worker.ts +++ b/src/services/rawDump/worker.ts @@ -30,7 +30,11 @@ import { toCommitComment, } from './git.js' import { createLogger } from './logger.js' -import { getRawDumpMode, RAW_DUMP_MODE, writeLocalDump } from './localStorage.js' +import { + getRawDumpMode, + 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 type { @@ -103,17 +107,15 @@ function getRawDumpUrl( } /** - * 发送 JSON POST 请求，支持重试和本地存储模式 + * 上报数据到 Raw Dump API * - 本地模式：写入本地 JSON 文件，不调用服务端 * - 网络错误/429：最多重试 3 次，指数退避 - * @param baseUrl API 基础地址 - * @param headers HTTP 请求头 - * @param endpoint API 路径 + * @param authData 认证数据（包含 baseUrl 和 headers） + * @param endpoint API 路径，如 /raw-store/task-conversation * @param body 请求体（会自动序列化为 JSON） */ -async function postJson( - baseUrl: string, - headers: Headers, +async function uploadReport( + authData: { baseUrl: string; headers: Headers }, endpoint: string, body: object, ): Promise { @@ -143,8 +145,8 @@ async function postJson( // REMOTE / BOTH 模式下继续执行 remote 上报逻辑 - const isAnonymous = !headers.get('Authorization') - const url = getRawDumpUrl(baseUrl, endpoint, isAnonymous) + const isAnonymous = !authData.headers.get('Authorization') + const url = getRawDumpUrl(authData.baseUrl, endpoint, isAnonymous) log.debug(`POST ${endpoint}`, { url, isAnonymous }) let lastError: Error | undefined @@ -160,7 +162,7 @@ async function postJson( try { const res = await fetch(url, { method: 'POST', - headers, + headers: authData.headers, body: JSON.stringify(body), signal: controller.signal, }) @@ -906,12 +908,7 @@ export async function uploadConversation( request_id: requestID, bodyKeys: Object.keys(body), }) - await postJson( - authData.baseUrl, - authData.headers, - '/raw-store/task-conversation', - body, - ) + await uploadReport(authData, '/raw-store/task-conversation', body) state.conversation[key] = true log.info('conversation uploaded', { task_id: payload.sessionID, @@ -947,9 +944,7 @@ export async function uploadSummary( const lastReported = state.summary[payload.sessionID] if ( lastReported && - Date.now() - - new Date(lastReported).getTime() < - SUMMARY_DEDUP_WINDOW_MS + Date.now() - new Date(lastReported).getTime() < SUMMARY_DEDUP_WINDOW_MS ) { log.info('summary skipped: reported recently', { task_id: payload.sessionID, @@ -975,12 +970,7 @@ export async function uploadSummary( caller: process.env.CSC_RAW_DUMP_CALLER || 'chat', } - await postJson( - authData.baseUrl, - authData.headers, - '/raw-store/task-summary', - body, - ) + await uploadReport(authData, '/raw-store/task-summary', body) state.summary[payload.sessionID] = new Date().toISOString() log.info('summary uploaded', { task_id: payload.sessionID }) } @@ -1056,9 +1046,8 @@ export async function uploadCommits( subject: commit.subject, parent_ids: commit.parent_ids, } - await postJson( - authData.baseUrl, - authData.headers, + await uploadReport( + authData, '/raw-store/commit', body as unknown as Record, ) @@ -1115,12 +1104,7 @@ export async function uploadStatistics( downstream_tokens: payload.downstreamTokens, } - await postJson( - authData.baseUrl, - authData.headers, - '/raw-store/statistics', - body, - ) + await uploadReport(authData, '/raw-store/statistics', body) state.summary[key] = new Date().toISOString() log.info('statistics uploaded', { task_id: payload.sessionID, @@ -1157,7 +1141,7 @@ 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, '-') + return dir.replace(/:/, '-').replace(/[/\\]/g, '-') } /**