解决一系列上报失败的错误
This commit is contained in:
zbc
parent
2e6dd38603
commit
ae22ffd36e
|
|
@ -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<ReturnType<typeof getRepoInfo>>
|
||||
|
||||
const BATCH_INTERVAL_MS = 120_000
|
||||
|
||||
// Git repo 信息缓存，同一 directory 的多个 task 短时间内不需要重复 spawn git
|
||||
const repoInfoCache = new Map<string, { repoInfo: RepoInfo; ts: number }>()
|
||||
const REPO_CACHE_TTL_MS = 60_000
|
||||
|
||||
async function getCachedRepoInfo(directory: string): Promise<RepoInfo> {
|
||||
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<string, unknown>[]; 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<ReturnType<typeof readState>>,
|
||||
) {
|
||||
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<string, QueueTask>()
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -17,10 +17,6 @@ const log = createLogger('raw-dump')
|
|||
|
||||
let batchWorkerSpawned = false
|
||||
|
||||
// 调用频率限制：同一 session + messageID 5s 内不重复 enqueue
|
||||
const lastEnqueueMap = new Map<string, number>()
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<string, number>()
|
||||
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<void> {
|
||||
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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<QueueTask, 'enqueuedAt' | 'attemptCount'>): 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(() => {})
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
613
src/services/rawDump/session.ts
Normal file
613
src/services/rawDump/session.ts
Normal file
|
|
@ -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<string, unknown> => 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<string, unknown>)?.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<string, unknown>[],
|
||||
messageID: string,
|
||||
): Record<string, unknown> | undefined {
|
||||
return messages.find(
|
||||
m =>
|
||||
m.uuid === messageID ||
|
||||
(m.message as Record<string, unknown>)?.id === messageID,
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* 查找 assistant message 对应的父级 user message
|
||||
* 在 csc 中，user message 通常紧邻 assistant message 之前
|
||||
* 从 assistant 位置向前遍历，找到第一个 type === 'user' 的消息
|
||||
*/
|
||||
export function findParentUserMessage(
|
||||
messages: Record<string, unknown>[],
|
||||
assistantMsg: Record<string, unknown>,
|
||||
): Record<string, unknown> | 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<string, unknown>,
|
||||
user: Record<string, unknown> | 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, unknown>): string {
|
||||
const content = (msg.message as Record<string, unknown>)?.content
|
||||
if (!Array.isArray(content)) return String(content ?? '')
|
||||
return content
|
||||
.filter((block): block is Record<string, unknown> => block?.type === 'text')
|
||||
.map(block => String(block.text ?? ''))
|
||||
.join('\n')
|
||||
}
|
||||
|
||||
/**
|
||||
* 将 structured patch 格式转换为 unified diff 格式
|
||||
*/
|
||||
function structuredPatchToUnifiedDiff(
|
||||
filePath: string,
|
||||
patches: Array<Record<string, unknown>>,
|
||||
): 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<string, unknown>,
|
||||
allMessages?: Record<string, unknown>[],
|
||||
): { diff: string; diff_lines: number; files: string[] } {
|
||||
const diffs: string[] = []
|
||||
const files = new Set<string>()
|
||||
const handledToolUseIds = new Set<string>()
|
||||
|
||||
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<string, unknown>
|
||||
| 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<string, unknown> | 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<string, unknown>)?.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<Record<string, unknown>>
|
||||
| undefined
|
||||
if (Array.isArray(sp) && sp.length > 0 && filePath) {
|
||||
diffs.push(structuredPatchToUnifiedDiff(filePath, sp))
|
||||
files.add(filePath)
|
||||
const mContent = (m.message as Record<string, unknown>)?.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<string, unknown>)?.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<string, unknown> | 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<string, unknown>) {
|
||||
const usage = (msg.message as Record<string, unknown>)?.usage as
|
||||
| Record<string, number>
|
||||
| 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<string, number> = {
|
||||
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<string, unknown>): {
|
||||
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<string, unknown>
|
||||
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<string, unknown>
|
||||
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<string, unknown>)?.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<string, unknown>[],
|
||||
]
|
||||
const sessionMessagesCache = new Map<string, SessionMessagesCacheEntry>()
|
||||
|
||||
/**
|
||||
* 获取缓存的 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<string, unknown>
|
||||
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 []
|
||||
}
|
||||
|
|
@ -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<string, string> = {}
|
||||
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<string, string> = {}
|
||||
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<string, TaskRecord> = {}
|
||||
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<RawDumpState> {
|
|||
try {
|
||||
const text = await fs.readFile(STATE_FILE, 'utf-8')
|
||||
const parsed = JSON.parse(text) as Partial<RawDumpState>
|
||||
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()
|
||||
}
|
||||
|
|
|
|||
217
src/services/rawDump/statistics.ts
Normal file
217
src/services/rawDump/statistics.ts
Normal file
|
|
@ -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<void> {
|
||||
try {
|
||||
const state = await readState()
|
||||
const todayKey = getTodayKey()
|
||||
|
||||
// 过滤出今天的记录，清理掉昨天及之前的
|
||||
const cleanedStatistics: Record<string, string> = {}
|
||||
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<ReturnType<typeof readState>>,
|
||||
): 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
|
||||
}
|
||||
|
|
@ -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<string, true>
|
||||
summary: Record<string, string> // RFC3339 时间戳，如 "2024-01-01T12:00:00.000Z"
|
||||
commits: Record<string, string>
|
||||
conversation: Record<string, string> // key: "taskID:messageID", value: RFC3339 时间戳，未上报则为空字符串
|
||||
summary: Record<string, string> //key: task-id, value: 上报成功的时间戳， RFC3339格式，如 "2024-01-01T12:00:00.000Z"
|
||||
commits: Record<string, string> //key: repo-addr#branch#work-dir，value: 最后一个上报成功的commit-id
|
||||
statistics: Record<string, string> // key: YYYY/MM/DD, value: RFC3339 时间戳
|
||||
tasks: Record<string, TaskRecord> // 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<string, string>
|
||||
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
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue
Block a user