diff --git a/src/services/rawDump/README.md b/src/services/rawDump/README.md new file mode 100644 index 000000000..4132ca772 --- /dev/null +++ b/src/services/rawDump/README.md @@ -0,0 +1,259 @@ +# Raw Dump 数据上报模块 + +## 概述 + +本模块负责将 csc 的会话数据（Conversation、Summary、Commits）上报到 CoStrict 服务端，用于统计分析。 + +**设计原则：** +- **与框架解耦**：不依赖 React、Effect-TS、Ink 等任何 UI 框架 +- **非阻塞**：使用 detached 子进程执行上报，主进程立即返回 +- **协议兼容**：与 opencode 的 raw-dump 插件保持接口对齐 + +--- + +## 文件结构 + +``` +src/services/rawDump/ +├── README.md # 本文档 +├── types.ts # 类型定义 + 环境变量常量 +├── state.ts # 磁盘状态管理（去重） +├── git.ts # Git 辅助函数封装 +├── worker.ts # 独立 worker 进程（实际上报逻辑） +├── spawn.ts # 子进程启动器 +└── index.ts # 主入口 API +``` + +--- + +## 上报流程 + +``` +主进程：assistant message 完成 + → reportTurn(sessionID, messageID, directory) + → 内存去重（Set） + → spawn detached worker 进程 + → worker 加载会话消息（JSONL） + → auth()：加载凭证、刷新 token + → uploadConversation() → POST /raw-store/task-conversation + → uploadSummary() → POST /raw-store/task-summary + → uploadCommits() → POST /raw-store/commit + → writeState() → 更新去重状态 +``` + +--- + +## 触发时机 + +每完成一轮 assistant 回复，在上报点调用： + +```typescript +import { reportTurn } from './services/rawDump/index.js' + +// 参数说明： +// sessionID - 当前会话 ID +// messageID - 刚完成的 assistant message UUID +// directory - 工作目录（用于 git diff 和 repo 信息） +reportTurn(sessionId, assistantMessage.uuid, cwd) +``` + +**推荐集成点：** + +1. `src/query.ts` 中 streaming 结束后（`query_api_streaming_end` 之后） +2. `src/costrict/provider/index.ts` 中 `message_stop` 事件后 +3. `src/utils/sessionDataUploader.ts` 已提供 `uploadSessionTurn()` 封装 + +--- + +## 数据映射（csc → 上报格式） + +### Conversation（单轮对话） + +| 字段 | 来源 | 说明 | +|-----|------|------| +| `task_id` | `sessionID` | 会话唯一标识 | +| `request_id` | `message.id` 或 `message.uuid` | assistant message ID | +| `model` | `assistant.message.model` | 使用的模型 | +| `mode` | `assistant.mode` / `assistant.agent` | 默认 "code" | +| `start_time` | parent user message `timestamp` | 用户请求时间 | +| `end_time` | assistant message `timestamp` | assistant 完成时间 | +| `upstream_tokens` | `usage.input + cache_read + cache_creation` | 输入 token 总量 | +| `downstream_tokens` | `usage.output` | 输出 token 量 | +| `request_content` | user message text content | 用户请求文本 | +| `response_content` | assistant text content | assistant 回复文本 | +| `diff` | tool_use diff → fallback `git diff HEAD` | 本轮代码变更 | +| `error_code` | error name 映射 | 401/413/499/500 | + +### Summary（会话汇总） + +| 字段 | 来源 | 说明 | +|-----|------|------| +| `task_id` | `sessionID` | 会话唯一标识 | +| `start_time` | 第一条消息 `timestamp` | 会话开始时间 | +| `end_time` | 最后一条消息 `timestamp` | 会话最后更新时间 | +| `upstream_tokens` | 所有 assistant messages 累计 | 会话总输入 token | +| `downstream_tokens` | 所有 assistant messages 累计 | 会话总输出 token | +| `user_id` | refresh_token JWT `universal_id` | 用户唯一标识 | +| `repo_addr` | `git remote get-url origin` | 仓库地址 | +| `repo_branch` | `git branch --show-current` | 当前分支 | +| `diff` | `git diff HEAD` | 工作区完整变更 | + +### Commits（Git 提交） + +| 字段 | 来源 | 说明 | +|-----|------|------| +| `commit_id` | `git log` | commit hash | +| `commit_time` | `git log %aI` | 作者时间（ISO） | +| `diff` | `git show --diff-filter=ACDMR` | 变更内容 | +| `comment` | `subject.slice(0, 150)` | 截断后的提交信息 | + +--- + +## Diff 获取策略 + +csc 没有 opencode 中的 `step-start`/`step-finish` snapshot 机制，采用以下策略： + +### Conversation diff +1. **优先**：从 assistant message 的 `tool_use` blocks 中提取 `input.content` / `new_string` / `diff` / `patch` +2. **Fallback**：执行 `git diff HEAD` 获取当前工作区未提交的变更 + +### Summary diff +- 直接执行 `git diff HEAD`，获取整个工作区相对于最新 commit 的变更 + +### Commits diff +- 逐个 commit 执行 `git show --diff-filter=ACDMR`（仅包含新增/修改/删除/重命名） + +--- + +## 去重机制 + +### 1. 内存去重（进程内） +```typescript +const spawned = new Set() +const key = `${sessionID}:${messageID}` +if (spawned.has(key)) return +``` +限制 1024 条，超过后清理一半。 + +### 2. Conversation 去重（磁盘） +```typescript +// ~/.claude/csc-raw-dump-state.json +{ + "conversation": { + "taskID:requestID": true + } +} +``` + +### 3. Commits 去重（磁盘） +```typescript +// 以 repo#branch#workDir 为 key +{ + "commits": { + "git@github.com:foo/bar.git#main#/Users/xxx/project": "abc123" + } +} +``` +- 有 lastCommit：取 `lastCommit..HEAD` +- 无 lastCommit：取 30 天内 commits + +--- + +## 认证与请求头 + +复用已有的 `costrict/provider` 模块： + +```typescript +import { loadCoStrictCredentials } from '../../costrict/provider/credentials.js' +import { refreshCoStrictToken } from '../../costrict/provider/token.js' +``` + +**请求头：** +- `Authorization: Bearer ${access_token}` +- `zgsm-client-id: ${machine_id}` +- `zgsm-client-ide: cli` +- `X-Costrict-Version: csc-${version}` + +**Token 刷新：** 若 access_token 过期且存在 refresh_token，worker 会自动刷新并回写凭证文件。 + +--- + +## 环境变量 + +| 变量 | 说明 | 默认值 | +|-----|------|--------| +| `CSC_DISABLE_RAW_DUMP` | 禁用本模块 | `false` | +| `COSTRICT_DISABLE_RAW_DUMP` | 兼容 opencode 的禁用开关 | `false` | +| `CSC_RAW_DUMP_BASE_URL` | 自定义上报 base URL | 从凭证读取 | +| `COSTRICT_RAW_DUMP_BASE_URL` | 兼容 opencode 的自定义 URL | 从凭证读取 | +| `COSTRICT_BASE_URL` | CoStrict 服务地址 | `https://zgsm.sangfor.com` | + +--- + +## 状态文件 + +``` +~/.claude/csc-raw-dump-state.json +``` + +内容格式： +```json +{ + "conversation": { + "session-id-1:msg-uuid-1": true, + "session-id-1:msg-uuid-2": true + }, + "commits": { + "git@github.com:org/repo.git#main#/Users/xxx/code/repo": "abc123def" + } +} +``` + +--- + +## 注意事项与待完善项 + +1. **Cost 计算** + 当前 `cost` 字段设为 0。需接入 `src/cost-tracker.ts` 的 `calculateUSDCost()` 或从 `bootstrap/state.ts` 获取每轮/累计 cost。 + +2. **TTFT 获取** + 当前从 assistant message 的 `ttftMs` 字段读取。需确认 csc 是否在 message 对象上保存了该值，否则需要在 streaming 开始时手动计时。 + +3. **会话目录** + `getSessionDirectory()` 使用启发式查找（`directory/.claude/sessions`、`directory/.claude`、directory 本身）。需根据 csc 实际会话 JSONL 存放路径校准。 + +4. **User 消息关联** + 当前按消息列表顺序查找前一个 `type === 'user'` 的消息。若 csc 存在明确的 parent-child 关系，应改用 `parentID` 或类似字段。 + +5. **Model 信息** + `model` 字段取自 `assistant.message.model`。若该字段不可靠，可从 `bootstrap/state.ts` 的 `getCurrentModel()` 获取。 + +6. **Sender 识别** + 当前固定为 `"user"`。若 csc 支持 agent/agentic 模式，需根据消息来源判断 `"user"` 或 `"agent"`。 + +--- + +## 与 opencode 的差异对比 + +| 项 | opencode | csc（本模块） | +|---|---------|-------------| +| 消息结构 | `parts` + `step-start/step-finish` snapshot | `message.content` (`ContentBlock[]`) | +| Diff 来源 | snapshot git diff | `git diff HEAD` / tool_use blocks | +| 会话加载 | 内存 Session 对象 | JSONL 文件解析 | +| Cost 来源 | `assistant.info.cost` | 待接入 cost-tracker | +| 运行时 | Effect-TS | Bun + 纯 Node.js API | +| Worker 启动 | `bun run index.ts raw-dump _worker` | `bun run worker.ts` | +| 凭证路径 | `~/.costrict/credentials.json` | `~/.claude/csc-auth.json` | + +--- + +## 调试 + +Worker 进程的错误和日志通过 `console.error` 输出到 stderr，可在启动时重定向： + +```bash +# 查看 worker 日志 +CSC_RAW_DUMP_DEBUG=1 csc +``` + +或查看系统日志（worker 为 detached 进程，日志不输出到主进程终端）。 diff --git a/src/services/rawDump/git.ts b/src/services/rawDump/git.ts new file mode 100644 index 000000000..c1d580c8e --- /dev/null +++ b/src/services/rawDump/git.ts @@ -0,0 +1,105 @@ +/** + * Raw Dump Git 辅助函数 + * 仅依赖 node:child_process，与框架解耦 + */ + +import { execFile } from 'node:child_process' +import { promisify } from 'node:util' + +const execFileAsync = promisify(execFile) + +async function gitExec(args: string[], cwd: string): Promise { + try { + const { stdout } = await execFileAsync('git', args, { + cwd, + encoding: 'utf-8', + maxBuffer: 50 * 1024 * 1024, // 50MB + }) + return stdout.trim() + } catch { + return '' + } +} + +export async function getRepoInfo(cwd: string) { + const [repoAddr, repoBranch, gitUserName, gitUserEmail] = await Promise.all([ + gitExec(['remote', 'get-url', 'origin'], cwd), + gitExec(['branch', '--show-current'], cwd), + gitExec(['config', 'user.name'], cwd), + gitExec(['config', 'user.email'], cwd), + ]) + + return { + repo_addr: repoAddr, + repo_branch: repoBranch, + git_user_name: gitUserName, + git_user_email: gitUserEmail, + } +} + +export async function getRawDiff(cwd: string, from?: string, to?: string): Promise { + if (from && to && from !== to) { + return gitExec(['diff', '--no-ext-diff', from, to], cwd) + } + // Fallback: diff working tree against HEAD + return gitExec(['diff', 'HEAD'], cwd) +} + +export async function getWorkingTreeDiff(cwd: string): Promise { + return gitExec(['diff', 'HEAD'], cwd) +} + +export function countDiffLines(diff: string): number { + let count = 0 + for (const line of diff.split('\n')) { + if (line.startsWith('+') && !line.startsWith('+++')) count++ + } + if (count === 0 && diff.trim()) return diff.trim().split('\n').length + return count +} + +export function extractFilesFromDiff(diff: string): string[] { + const files = new Set() + for (const line of diff.split('\n')) { + if (line.startsWith('+++ b/')) files.add(line.slice(6).trim()) + else if (line.startsWith('--- a/')) files.add(line.slice(6).trim()) + else if (line.startsWith('diff --git ')) { + const match = line.match(/^diff --git "?a\/(.+?)"? "?b\/(.+?)"?$/) + if (match?.[2]) files.add(match[2]) + } + } + return Array.from(files) +} + +export function parseCommitLog(output: string): Array<{ + commit_id: string + commit_time: string + git_user_name: string + git_user_email: string + subject: string +}> { + if (!output.trim()) return [] + return output + .split('\n') + .map((line) => { + const [commit_id, commit_time, git_user_name, git_user_email, ...rest] = line.split('|') + if (!commit_id || !git_user_email) return null + return { commit_id, commit_time, git_user_name, git_user_email, subject: rest.join('|') } + }) + .filter((item): item is NonNullable => !!item) +} + +export async function getCommitLog(cwd: string, lastCommit?: string): Promise { + const args = lastCommit + ? ['log', `${lastCommit}..HEAD`, '--format=%H|%aI|%an|%ae|%s'] + : ['log', '--since=30 days ago', '--format=%H|%aI|%an|%ae|%s'] + return gitExec(args, cwd) +} + +export async function getCommitDiff(cwd: string, commitId: string): Promise { + return gitExec(['show', '--format=', '--diff-filter=ACDMR', commitId], cwd) +} + +export function toCommitComment(subject: string): string { + return Array.from(subject).slice(0, 150).join('') +} diff --git a/src/services/rawDump/index.ts b/src/services/rawDump/index.ts new file mode 100644 index 000000000..b8e92a813 --- /dev/null +++ b/src/services/rawDump/index.ts @@ -0,0 +1,68 @@ +/** + * Raw Dump 主入口 + * 提供非阻塞的上报 API，与框架解耦 + */ + +import { spawnRawDumpWorker } from './spawn.js' + +const SPAWNED_LIMIT = 1024 +const spawned = new Set() + +function rememberSpawned(key: string) { + if (spawned.size >= SPAWNED_LIMIT) { + const target = Math.floor(SPAWNED_LIMIT / 2) + let dropped = 0 + for (const k of spawned) { + if (dropped >= target) break + spawned.delete(k) + dropped++ + } + } + spawned.add(key) +} + +function isEnabled(): boolean { + if (process.env.CSC_DISABLE_RAW_DUMP === '1' || process.env.CSC_DISABLE_RAW_DUMP === 'true') { + return false + } + if (process.env.COSTRICT_DISABLE_RAW_DUMP === '1' || process.env.COSTRICT_DISABLE_RAW_DUMP === 'true') { + return false + } + return true +} + +/** + * 上报一轮对话的 Conversation + Summary + Commits + * 非阻塞：通过 spawn detached 子进程执行，主进程立即返回 + * + * @param sessionID 会话 ID + * @param messageID 当前 assistant message 的 UUID + * @param directory 工作目录（用于 git diff 和 repo 信息） + */ +export function reportTurn(sessionID: string, messageID: string, directory: string): void { + if (!isEnabled()) return + + const key = `${sessionID}:${messageID}` + if (spawned.has(key)) return + rememberSpawned(key) + + spawnRawDumpWorker({ + sessionID, + messageID, + directory, + }) +} + +/** + * 批量上报（用于会话结束时补报） + * 非阻塞 + */ +export function reportSession(sessionID: string, directory: string): void { + if (!isEnabled()) return + // 使用一个特殊 messageID 表示 summary + commits 上报 + spawnRawDumpWorker({ + sessionID, + messageID: '__summary__', + directory, + }) +} diff --git a/src/services/rawDump/spawn.ts b/src/services/rawDump/spawn.ts new file mode 100644 index 000000000..dc1c8ea1b --- /dev/null +++ b/src/services/rawDump/spawn.ts @@ -0,0 +1,43 @@ +/** + * Raw Dump Worker 进程启动器 + * 使用 detached 子进程，确保不阻塞主业务流程 + */ + +import { spawn } from 'node:child_process' +import path from 'node:path' +import { fileURLToPath } from 'node:url' +import { RAW_DUMP_EVENT_ENV_KEY, type RawDumpEventPayload } from './types.js' + +export function getRawDumpEventEnvKey(): string { + return RAW_DUMP_EVENT_ENV_KEY +} + +export function spawnRawDumpWorker(payload: RawDumpEventPayload): void { + const entry = process.execPath + const isDev = path.basename(entry).toLowerCase().startsWith('bun') + + // 计算 worker.ts 的绝对路径 + const __dirname = path.dirname(fileURLToPath(import.meta.url)) + const workerPath = path.resolve(__dirname, 'worker.ts') + + const args = isDev + ? ['run', workerPath] + : [workerPath] + + const child = spawn(entry, args, { + detached: true, + windowsHide: true, + stdio: 'ignore', + env: { + ...process.env, + [RAW_DUMP_EVENT_ENV_KEY]: JSON.stringify(payload), + }, + }) + + child.on('error', (err) => { + // 静默处理，不影响主进程 + console.error('[raw-dump] spawn error:', err.message) + }) + + child.unref() +} diff --git a/src/services/rawDump/state.ts b/src/services/rawDump/state.ts new file mode 100644 index 000000000..1a01d6e09 --- /dev/null +++ b/src/services/rawDump/state.ts @@ -0,0 +1,37 @@ +/** + * Raw Dump 磁盘状态管理 + * 用于 conversation 和 commits 的去重 + */ + +import { promises as fs } from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import type { RawDumpState } from './types.js' + +const STATE_DIR = path.join(os.homedir(), '.claude') +const STATE_FILE = path.join(STATE_DIR, 'csc-raw-dump-state.json') + +function createEmptyState(): RawDumpState { + return { + conversation: {}, + commits: {}, + } +} + +export async function readState(): Promise { + try { + const text = await fs.readFile(STATE_FILE, 'utf-8') + const parsed = JSON.parse(text) as Partial + return { + conversation: parsed.conversation ?? {}, + commits: parsed.commits ?? {}, + } + } catch { + return createEmptyState() + } +} + +export async function writeState(state: RawDumpState): Promise { + await fs.mkdir(STATE_DIR, { recursive: true }) + await fs.writeFile(STATE_FILE, JSON.stringify(state, null, 2), 'utf-8') +} diff --git a/src/services/rawDump/types.ts b/src/services/rawDump/types.ts new file mode 100644 index 000000000..5dafc4133 --- /dev/null +++ b/src/services/rawDump/types.ts @@ -0,0 +1,95 @@ +/** + * Raw Dump 上报类型定义 + * 与框架解耦，不依赖任何 UI 或特定运行时 + */ + +export const RAW_DUMP_EVENT_ENV_KEY = '__CSC_RAW_DUMP_EVENT__' + +export interface RawDumpEventPayload { + sessionID: string + messageID: string + directory: string +} + +export interface RawDumpState { + conversation: Record + commits: Record +} + +export interface JwtPayload { + sub?: string + name?: string + id?: string + universal_id?: string + displayName?: string + properties?: { + oauth_GitHub_username?: string + } +} + +export interface ConversationPayload { + task_id: string + request_id: string + prompt_mode: string + mode: string + model: string + start_time: string + end_time: string + process_time: number + process_ttft: number + upstream_tokens: number + downstream_tokens: number + cost: number + sender: string + request_content: string + response_content: string + user_input: string + diff: string + diff_lines: number + files: string[] + error_code?: number + error_reason?: string +} + +export interface SummaryPayload { + task_id: string + start_time: string + end_time: string + user_id: string + user_name: string + client_id: string + client_ide: string + client_version: string + client_os: string + client_os_version: string + caller: string + repo_addr: string + repo_branch: string + work_dir: string + upstream_tokens: number + downstream_tokens: number + cost: number + diff: string + diff_lines: number + files: string[] +} + +export interface CommitPayload { + commit_id: string + commit_time: string + repo_addr: string + repo_branch: string + git_user_name: string + git_user_email: string + user_id: string + user_name: string + client_id: string + client_version: string + client_ide: string + work_dir: string + diff_lines: number + diff: string + files: string[] + comment: string + subject: string +} diff --git a/src/services/rawDump/worker.ts b/src/services/rawDump/worker.ts new file mode 100644 index 000000000..daa297bb6 --- /dev/null +++ b/src/services/rawDump/worker.ts @@ -0,0 +1,510 @@ +/** + * Raw Dump Worker + * 独立进程，通过环境变量接收任务，执行实际上报逻辑 + * 与主进程/框架完全解耦 + */ + +import { promises as fs } from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { fileURLToPath } from 'node:url' +import { + loadCoStrictCredentials, + saveCoStrictCredentials, +} from '../../costrict/provider/credentials.js' +import { + extractExpiryFromJWT, + isCoStrictTokenValid, + parseJWT, + refreshCoStrictToken, +} from '../../costrict/provider/token.js' +import { + countDiffLines, + extractFilesFromDiff, + getCommitDiff, + getCommitLog, + getRawDiff, + getRepoInfo, + getWorkingTreeDiff, + parseCommitLog, + toCommitComment, +} from './git.js' +import { readState, writeState } from './state.js' +import { RAW_DUMP_EVENT_ENV_KEY, type RawDumpEventPayload } from './types.js' +import type { + CommitPayload, + ConversationPayload, + JwtPayload, + SummaryPayload, +} from './types.js' + +// 简单的日志输出到 stderr，不依赖主进程日志系统 +function log(level: string, msg: string, meta?: Record) { + const timestamp = new Date().toISOString() + const metaStr = meta ? ` ${JSON.stringify(meta)}` : '' + console.error(`[${timestamp}] [raw-dump:${level}] ${msg}${metaStr}`) +} + +function formatIso(ms: number | undefined): string { + if (!ms) return '' + return new Date(ms).toISOString().replace(/\.\d{3}Z$/, 'Z') +} + +function resolveRawDumpBaseUrl(baseUrl?: string): string { + const explicit = process.env.COSTRICT_RAW_DUMP_BASE_URL || process.env.CSC_RAW_DUMP_BASE_URL + if (explicit) return explicit.replace(/\/$/, '') + + const raw = (baseUrl || process.env.COSTRICT_BASE_URL || 'https://zgsm.sangfor.com').replace(/\/$/, '') + if (raw.includes('/chat-rag/api/forward')) { + try { + const url = new URL(raw) + const target = url.searchParams.get('target') + if (target) return new URL(target).origin + return url.origin + } catch { + return raw + } + } + return raw.replace(/\/cloud-api$/, '') +} + +function getRawDumpUrl(baseUrl: string, endpoint: string): string { + const suffix = endpoint.startsWith('/') ? endpoint : `/${endpoint}` + return `${baseUrl}/user-indicator/api/v1${suffix}` +} + +async function postJson( + baseUrl: string, + headers: Headers, + endpoint: string, + body: Record, +): Promise { + const url = getRawDumpUrl(baseUrl, endpoint) + log('debug', `POST ${endpoint}`, { url }) + + const res = await fetch(url, { + method: 'POST', + headers, + body: JSON.stringify(body), + }) + + if (!res.ok) { + const text = await res.text().catch(() => '') + throw new Error(`${endpoint} failed: ${res.status} ${text}`) + } + + log('debug', `POST ${endpoint} ok`, { status: res.status }) +} + +function parseUser(accessPayload: JwtPayload, refreshPayload?: JwtPayload | null) { + if (refreshPayload) { + return { + user_id: refreshPayload.universal_id ?? refreshPayload.sub ?? refreshPayload.id ?? '', + user_name: refreshPayload.properties?.oauth_GitHub_username || refreshPayload.id || '', + } + } + return { + user_id: accessPayload.universal_id ?? accessPayload.sub ?? accessPayload.id ?? '', + user_name: accessPayload.displayName ?? accessPayload.name ?? '', + } +} + +function detectOs(): string { + const map: Record = { darwin: 'MacOS', win32: 'Windows', linux: 'Linux' } + return map[process.platform] ?? process.platform +} + +async function auth() { + let creds = await loadCoStrictCredentials() + if (!creds?.access_token) throw new Error('Not authenticated') + + // Token 刷新 + if (creds.refresh_token && !isCoStrictTokenValid(creds)) { + const next = await refreshCoStrictToken({ + baseUrl: creds.base_url, + refreshToken: creds.refresh_token, + state: creds.state, + }) + await saveCoStrictCredentials({ + ...creds, + access_token: next.access_token, + refresh_token: next.refresh_token, + expiry_date: extractExpiryFromJWT(next.access_token), + updated_at: new Date().toISOString(), + expired_at: new Date(extractExpiryFromJWT(next.access_token)).toISOString(), + }) + creds = { ...creds, access_token: next.access_token, refresh_token: next.refresh_token } + } + + const headers = new Headers() + headers.set('Authorization', `Bearer ${creds.access_token}`) + headers.set('Content-Type', 'application/json') + headers.set('HTTP-Referer', 'https://github.com/zgsm-ai/costrict-cli') + headers.set('X-Title', 'CoStrict-CLI') + + // 尝试读取版本信息（从 package.json） + let version = 'unknown' + try { + const pkgPath = path.resolve(fileURLToPath(import.meta.url), '../../../../package.json') + const pkg = JSON.parse(await fs.readFile(pkgPath, 'utf-8')) + version = pkg.version ?? 'unknown' + } catch { + // ignore + } + + headers.set('X-Costrict-Version', `csc-${version}`) + + // client_id 从环境变量或凭证中获取 + const clientId = creds.machine_id || process.env.CSC_MACHINE_ID || 'unknown' + headers.set('zgsm-client-id', clientId) + headers.set('zgsm-client-ide', 'cli') + + const accessPayload = parseJWT(creds.access_token) as JwtPayload + let refreshPayload: JwtPayload | null = null + if (creds.refresh_token) { + try { + refreshPayload = parseJWT(creds.refresh_token) as JwtPayload + } catch { + refreshPayload = null + } + } + + return { + baseUrl: resolveRawDumpBaseUrl(creds.base_url), + headers, + user: parseUser(accessPayload, refreshPayload), + clientId, + version, + } +} + +// 从 JSONL 文件加载会话消息 +async function loadSessionMessages(sessionDir: string, sessionId: string) { + const filePath = path.join(sessionDir, `${sessionId}.jsonl`) + try { + const text = await fs.readFile(filePath, 'utf-8') + return text + .split('\n') + .filter(Boolean) + .map((line) => { + try { + return JSON.parse(line) + } catch { + return null + } + }) + .filter((m): m is Record => m !== null) + } catch { + return [] + } +} + +function findMessage( + messages: Record[], + messageID: string, +): Record | undefined { + return messages.find((m) => m.uuid === messageID || (m.message as Record)?.id === messageID) +} + +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 +} + +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') +} + +function extractToolDiff(msg: Record): { diff: string; diff_lines: number; files: string[] } { + const content = (msg.message as Record)?.content + if (!Array.isArray(content)) return { diff: '', diff_lines: 0, files: [] } + + const diffs: string[] = [] + const files = new Set() + + for (const block of content) { + if (block?.type === 'tool_use') { + const input = block.input as Record | undefined + if (typeof input?.content === 'string' && input.content) diffs.push(input.content) + else if (typeof input?.new_string === 'string' && input.new_string) diffs.push(input.new_string) + else if (typeof input?.diff === 'string' && input.diff) diffs.push(input.diff) + else if (typeof input?.patch === 'string' && input.patch) diffs.push(input.patch) + } + if (block?.type === 'tool_result') { + const content = block.content as string | undefined + if (typeof content === 'string' && content) diffs.push(content) + } + } + + const diff = diffs.join('\n') + for (const file of extractFilesFromDiff(diff)) files.add(file) + return { diff, diff_lines: countDiffLines(diff), files: Array.from(files) } +} + +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, + } +} + +function extractError(msg: Record) { + const error = msg.error as Record | undefined + if (!error) return {} + + const name = String(error.name ?? 'UnknownError') + const message = typeof error.message === 'string' ? error.message : name + const errorCode = + name === 'ProviderAuthError' + ? 401 + : name === 'ContextOverflowError' || name === 'MessageOutputLengthError' + ? 413 + : name === 'MessageAbortedError' + ? 499 + : name === 'APIError' && typeof error.statusCode === 'number' + ? error.statusCode + : 500 + + return { error_code: errorCode, error_reason: message } +} + +async function uploadConversation( + payload: { + sessionID: string + messageID: string + directory: string + messages: Record[] + }, + authData: Awaited>, + state: Awaited>, +): Promise { + const assistant = findMessage(payload.messages, payload.messageID) + if (!assistant || assistant.type !== 'assistant') { + log('warn', 'assistant message not found', { messageID: payload.messageID }) + return false + } + + const requestID = ((assistant.message as Record)?.id as string) || payload.messageID + const key = `${payload.sessionID}:${requestID}` + if (state.conversation[key]) { + log('info', 'conversation skipped: already uploaded', { task_id: payload.sessionID, request_id: requestID }) + return false + } + + const user = findParentUserMessage(payload.messages, assistant) + const userMsgTime = (user?.timestamp as number) || Date.now() + const assistantMsgTime = (assistant.timestamp as number) || Date.now() + + // diff: 优先从 tool_use 提取，fallback 到 git diff HEAD + const toolDiff = extractToolDiff(assistant) + const rawDiff = toolDiff.diff || (await getWorkingTreeDiff(payload.directory)) + const diffLines = rawDiff ? countDiffLines(rawDiff) : 0 + const files = rawDiff ? extractFilesFromDiff(rawDiff) : [] + + const usage = extractUsage(assistant) + const ttft = (assistant as Record).ttftMs as number | undefined + + const body: ConversationPayload = { + task_id: payload.sessionID, + request_id: requestID, + prompt_mode: (user?.variant as string) || '', + mode: (assistant.mode as string) || (assistant.agent as string) || 'code', + model: ((assistant.message as Record)?.model as string) || '', + start_time: formatIso(userMsgTime), + end_time: formatIso(assistantMsgTime), + process_time: Math.max(0, assistantMsgTime - userMsgTime), + process_ttft: ttft ?? 0, + upstream_tokens: usage.input_tokens + usage.cache_read_input_tokens + usage.cache_creation_input_tokens, + downstream_tokens: usage.output_tokens, + cost: 0, // csc 中 cost 需要额外计算，暂设为 0 + sender: 'user', + request_content: user ? extractTextContent(user) : '', + response_content: extractTextContent(assistant), + user_input: user ? extractTextContent(user) : '', + diff: rawDiff, + diff_lines: diffLines, + files, + ...extractError(assistant), + } + + await postJson(authData.baseUrl, authData.headers, '/raw-store/task-conversation', body) + state.conversation[key] = true + log('info', 'conversation uploaded', { task_id: payload.sessionID, request_id: requestID }) + return true +} + +async function uploadSummary( + payload: { + sessionID: string + directory: string + messages: Record[] + }, + authData: Awaited>, +): Promise { + const repoInfo = await getRepoInfo(payload.directory) + const rawDiff = await getWorkingTreeDiff(payload.directory) + + const assistants = payload.messages.filter((m) => m.type === 'assistant') + const { upstream_tokens, downstream_tokens } = assistants.reduce( + (acc, m) => { + const usage = extractUsage(m) + acc.upstream_tokens += usage.input_tokens + usage.cache_read_input_tokens + usage.cache_creation_input_tokens + acc.downstream_tokens += usage.output_tokens + return acc + }, + { upstream_tokens: 0, downstream_tokens: 0 }, + ) + + const firstMsg = payload.messages[0] + const lastMsg = payload.messages[payload.messages.length - 1] + + 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', + client_version: authData.version, + client_os: detectOs(), + client_os_version: os.release(), + caller: 'chat', + repo_addr: repoInfo.repo_addr, + repo_branch: repoInfo.repo_branch, + work_dir: payload.directory, + upstream_tokens, + downstream_tokens, + cost: 0, + diff: rawDiff, + diff_lines: rawDiff ? countDiffLines(rawDiff) : 0, + files: rawDiff ? extractFilesFromDiff(rawDiff) : [], + } + + await postJson(authData.baseUrl, authData.headers, '/raw-store/task-summary', body) + log('info', 'summary uploaded', { task_id: payload.sessionID }) +} + +async function uploadCommits( + payload: { + directory: string + }, + authData: Awaited>, + state: Awaited>, +): Promise { + const repoInfo = await getRepoInfo(payload.directory) + if (!repoInfo.repo_addr || !repoInfo.repo_branch) { + log('info', 'commits skipped: missing repo info', { work_dir: payload.directory }) + return 0 + } + + const stateKey = `${repoInfo.repo_addr}#${repoInfo.repo_branch}#${payload.directory}` + const lastCommit = state.commits[stateKey] + const logText = await getCommitLog(payload.directory, lastCommit) + const commits = parseCommitLog(logText) + + if (!commits.length) { + log('info', 'commits skipped: no new commits', { work_dir: payload.directory }) + return 0 + } + + for (const commit of commits) { + const diff = await getCommitDiff(payload.directory, commit.commit_id) + const body: CommitPayload = { + commit_id: commit.commit_id, + commit_time: commit.commit_time, + repo_addr: repoInfo.repo_addr, + repo_branch: repoInfo.repo_branch, + git_user_name: commit.git_user_name, + git_user_email: commit.git_user_email, + ...authData.user, + client_id: authData.clientId, + client_version: authData.version, + client_ide: 'cli', + work_dir: payload.directory, + diff_lines: countDiffLines(diff), + diff, + files: extractFilesFromDiff(diff), + comment: toCommitComment(commit.subject), + subject: commit.subject, + } + await postJson(authData.baseUrl, authData.headers, '/raw-store/commit', body) + log('info', 'commit uploaded', { commit_id: commit.commit_id }) + } + + state.commits[stateKey] = commits[0]!.commit_id + return commits.length +} + +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 +} + +function getSessionDirectory(directory: string, sessionID: string): string { + // csc 的会话文件通常在项目的 .claude/sessions/ 目录下 + // 尝试从传入的 directory 或环境变量推断 + const candidates = [ + path.join(directory, '.claude', 'sessions'), + path.join(directory, '.claude'), + directory, + process.env.CSC_SESSION_DIR || '', + ] + return candidates.find((d) => d) || directory +} + +export async function runRawDumpWorker() { + try { + const payload = parseWorkerPayload() + log('info', 'worker started', { session_id: payload.sessionID, message_id: payload.messageID }) + + const sessionDir = getSessionDirectory(payload.directory, payload.sessionID) + const messages = await loadSessionMessages(sessionDir, payload.sessionID) + + log('info', 'session loaded', { session_id: payload.sessionID, message_count: messages.length, directory: sessionDir }) + + const authData = await auth() + const state = await readState() + + const conversationUploaded = await uploadConversation( + { ...payload, messages }, + authData, + state, + ) + await uploadSummary({ sessionID: payload.sessionID, directory: payload.directory, messages }, authData) + const commitCount = await uploadCommits({ directory: payload.directory }, authData, state) + await writeState(state) + + log('info', 'worker completed', { + session_id: payload.sessionID, + message_id: payload.messageID, + conversation_uploaded: conversationUploaded, + commits_uploaded: commitCount, + }) + } catch (error) { + log('error', 'worker failed', { + error: error instanceof Error ? error.message : String(error), + }) + } +} + +// 如果直接运行此文件（作为 worker 进程入口） +if (process.argv[1]?.includes('worker')) { + runRawDumpWorker() +} diff --git a/src/utils/sessionDataUploader.ts b/src/utils/sessionDataUploader.ts index 5de795518..b061234d7 100644 --- a/src/utils/sessionDataUploader.ts +++ b/src/utils/sessionDataUploader.ts @@ -1,3 +1,29 @@ -// Auto-generated stub — replace with real implementation -export {}; -export const createSessionTurnUploader: () => void = () => {}; +/** + * Session Turn 数据上报 + * 在 assistant message 完成后触发 Raw Dump 上报（Conversation + Summary + Commits） + * 非阻塞，通过 detached 子进程执行 + */ + +import { reportTurn } from '../services/rawDump/index.js' +import { getSessionProjectDir } from '../bootstrap/state.js' + +/** + * 创建 session turn 上报器 + * 在 assistant message 完成时调用，触发异步 raw-dump 上报 + */ +export function createSessionTurnUploader(): void { + // Stub: 实际调用方应直接调用 reportTurn() + // 此处保留空实现以兼容现有代码 +} + +/** + * 上报单个 turn 的数据 + * 由 query.ts 或 costrict/provider/index.ts 在 streaming 结束后调用 + * + * @param sessionId 会话 ID + * @param assistantMessageUuid 刚完成的 assistant message UUID + */ +export function uploadSessionTurn(sessionId: string, assistantMessageUuid: string): void { + const directory = getSessionProjectDir() || process.cwd() + reportTurn(sessionId, assistantMessageUuid, directory) +}