From a34d7d3f4d6359da5415c07214c4f34b4307f814 Mon Sep 17 00:00:00 2001 From: cheho Date: Fri, 2 Oct 2026 01:19:11 +0800 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20fix(vscode):=20=E5=8F=AF?= =?UTF-8?q?=E9=9D=A0=E6=8D=95=E8=8E=B7=E5=B9=B6=E8=A1=A5=E5=9B=9E=E5=AE=A1?= =?UTF-8?q?=E6=9F=A5=E4=BC=9A=E8=AF=9D=E5=85=B3=E8=81=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vscode/src/cc/codexSessionWatcher.test.ts | 275 ++++++++++--- vscode/src/cc/codexSessionWatcher.ts | 367 ++++++++++++------ vscode/src/panel/KanbanPanel.ts | 8 +- vscode/src/panel/handlers/issues.ts | 2 + .../handlers/reviewSessionCapture.test.ts | 263 +++++++++++++ .../panel/handlers/reviewSessionCapture.ts | 277 +++++++++++++ vscode/src/panel/handlers/sessions.ts | 135 ++----- 7 files changed, 1054 insertions(+), 273 deletions(-) create mode 100644 vscode/src/panel/handlers/reviewSessionCapture.test.ts create mode 100644 vscode/src/panel/handlers/reviewSessionCapture.ts diff --git a/vscode/src/cc/codexSessionWatcher.test.ts b/vscode/src/cc/codexSessionWatcher.test.ts index 5389144..0c8490e 100644 --- a/vscode/src/cc/codexSessionWatcher.test.ts +++ b/vscode/src/cc/codexSessionWatcher.test.ts @@ -1,21 +1,41 @@ -import { appendFileSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { EventEmitter } from 'node:events' +import { appendFileSync, promises as fsp, mkdirSync, mkdtempSync, rmSync, symlinkSync, watch, writeFileSync } from 'node:fs' import * as os from 'node:os' import * as path from 'node:path' -import { afterEach, describe, expect, it } from 'vitest' -import { watchForCodexSession } from './codexSessionWatcher' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { findCodexSession, watchForCodexSession } from './codexSessionWatcher' -const tempDirs: string[] = [] - -afterEach(() => { - while (tempDirs.length > 0) { - const dir = tempDirs.pop() - if (dir) - rmSync(dir, { recursive: true, force: true }) - } +vi.mock('node:fs', async (importOriginal) => { + const actual = await importOriginal() + return { ...actual, watch: vi.fn() } }) -function tempDir(prefix: string): string { - const dir = mkdtempSync(path.join(os.tmpdir(), prefix)) +const tempDirs: string[] = [] +const controllers: AbortController[] = [] +let notices: (() => void) | undefined +let fakeWatcher: EventEmitter & { close: ReturnType } + +beforeEach(() => { + vi.useFakeTimers() + fakeWatcher = Object.assign(new EventEmitter(), { close: vi.fn() }) + vi.mocked(watch).mockReset().mockImplementation(((_dir: string, _opts: unknown, notify: () => void) => { + notices = notify + return fakeWatcher + }) as unknown as typeof watch) +}) + +afterEach(() => { + for (const ac of controllers.splice(0)) + ac.abort() + vi.useRealTimers() + vi.restoreAllMocks() + for (const dir of tempDirs.splice(0)) + rmSync(dir, { recursive: true, force: true }) + notices = undefined +}) + +function tempDir(): string { + const dir = mkdtempSync(path.join(os.tmpdir(), 'codex-capture-')) tempDirs.push(dir) return dir } @@ -23,71 +43,204 @@ function tempDir(prefix: string): string { const MAIN_ID = '01a0e3bb-61e3-7c31-adf3-2d2a746a3282' const SUB_ID = '01a0e3bc-2a65-7490-84a6-e963505d9bff' const OTHER_ID = '01a0e3be-4665-7991-b89e-f6881609c36b' +const OLD_ID = '01a0e3be-4665-7991-b89e-f6881609c36b' +const STARTED_AT = Date.parse('2026-10-01T16:42:25.000Z') +const PROMPT = '审查这个仓库的 PR #470。\n只提交审查意见。' -function metaLine(id: string, cwd: string, source: unknown): string { - return `${JSON.stringify({ timestamp: '2026-09-27T16:38:24.015Z', type: 'session_meta', payload: { id, cwd, source } })}\n` +function metaLine(id: string, cwd: string, source: unknown = 'cli', timestamp = '2026-10-01T16:42:27.505Z', extra = ''): string { + return `${JSON.stringify({ timestamp: '2026-10-01T16:42:28.040Z', type: 'session_meta', payload: { id, cwd, source, timestamp, extra } })}\n` +} + +function promptLine(prompt = PROMPT): string { + return `${JSON.stringify({ type: 'response_item', payload: { role: 'user', content: [{ type: 'input_text', text: prompt }] } })}\n` } function rolloutPath(baseDir: string, id: string): string { - const day = path.join(baseDir, '2026', '09', '28') + const day = path.join(baseDir, '2026', '10', '02') mkdirSync(day, { recursive: true }) - return path.join(day, `rollout-2026-09-28T00-38-23-${id}.jsonl`) + return path.join(day, `rollout-2026-10-02T00-42-27-${id}.jsonl`) } -const tick = (ms = 50): Promise => new Promise(r => setTimeout(r, ms)) +function fixture(): { baseDir: string, cwd: string, ac: AbortController } { + const ac = new AbortController() + controllers.push(ac) + return { baseDir: tempDir(), cwd: tempDir(), ac } +} + +async function prepare(f: ReturnType) { + return watchForCodexSession({ baseDir: f.baseDir, cwd: f.cwd, startedAt: STARTED_AT, prompt: PROMPT, signal: f.ac.signal }) +} describe('watchForCodexSession', () => { - it('只认本目录的主会话，跳过别的目录和子助手会话', async () => { - const baseDir = tempDir('codex-sessions-') - const cwd = tempDir('worktree-') - const ac = new AbortController() - const found = watchForCodexSession({ baseDir, cwd, signal: ac.signal }) - await tick() - - writeFileSync(rolloutPath(baseDir, OTHER_ID), metaLine(OTHER_ID, '/somewhere/else', 'cli')) - writeFileSync(rolloutPath(baseDir, SUB_ID), metaLine(SUB_ID, cwd, { subagent: { thread_spawn: { parent_thread_id: MAIN_ID } } })) - await tick() - writeFileSync(rolloutPath(baseDir, MAIN_ID), metaLine(MAIN_ID, cwd, 'cli')) - - await expect(found).resolves.toBe(MAIN_ID) + it('准备阶段扫描已落盘会话，不依赖文件通知', async () => { + const f = fixture() + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + const found = await prepare(f) + expect(found.initialSessionId).toBe(MAIN_ID) + await expect(found.sessionId).resolves.toBe(MAIN_ID) + expect(fakeWatcher.close).toHaveBeenCalledOnce() }) - it('没有时间上限，文件晚出现也能认到', async () => { - const baseDir = tempDir('codex-sessions-') - const cwd = tempDir('worktree-') - const ac = new AbortController() - const found = watchForCodexSession({ baseDir, cwd, signal: ac.signal }) - await tick(300) - - writeFileSync(rolloutPath(baseDir, MAIN_ID), metaLine(MAIN_ID, cwd, 'cli')) - - await expect(found).resolves.toBe(MAIN_ID) + it('准备尚未结束时落盘的文件也会被首次扫描发现', async () => { + const f = fixture() + const original = fsp.realpath + let release!: () => void + const paused = new Promise((resolve) => { + release = resolve + }) + const check = vi.spyOn(fsp, 'realpath').mockImplementationOnce(async (p) => { + await paused + return original(p) + }) + const preparing = prepare(f) + await vi.waitFor(() => expect(check).toHaveBeenCalled()) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + release() + await expect((await preparing).sessionId).resolves.toBe(MAIN_ID) }) - it('第一行还没写完时等它写完再认', async () => { - const baseDir = tempDir('codex-sessions-') - const cwd = tempDir('worktree-') - const ac = new AbortController() - const found = watchForCodexSession({ baseDir, cwd, signal: ac.signal }) - await tick() + it('漏掉全部文件通知时定时扫描仍捕获新会话', async () => { + const f = fixture() + const found = await prepare(f) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + await vi.advanceTimersByTimeAsync(1000) + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) - const file = rolloutPath(baseDir, MAIN_ID) - const line = metaLine(MAIN_ID, cwd, 'cli') + it('目录变化通知没有文件名时也会扫描', async () => { + const f = fixture() + const found = await prepare(f) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + notices?.() + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) + + it('跳过其他目录、子助手与同目录不同启动提示词', async () => { + const f = fixture() + writeFileSync(rolloutPath(f.baseDir, OTHER_ID), metaLine(OTHER_ID, '/somewhere/else') + promptLine()) + writeFileSync(rolloutPath(f.baseDir, SUB_ID), metaLine(SUB_ID, f.cwd, { subagent: {} }) + promptLine()) + const differentId = '01a0e3be-4665-7991-b89e-f6881609c36c' + writeFileSync(rolloutPath(f.baseDir, differentId), metaLine(differentId, f.cwd) + promptLine('其他工作')) + const found = await prepare(f) + expect(found.initialSessionId).toBeNull() + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + notices?.() + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) + + it('按会话实际创建时间排除启动前旧会话，即使刚被写入或外层时间较新', async () => { + const f = fixture() + writeFileSync(rolloutPath(f.baseDir, OLD_ID), metaLine(OLD_ID, f.cwd, 'cli', '2026-10-01T16:42:24.999Z') + promptLine()) + const found = await prepare(f) + expect(found.initialSessionId).toBeNull() + f.ac.abort() + await expect(found.sessionId).resolves.toBeNull() + }) + + it('其他主会话后续收到相同提示词时不能当作本次启动', async () => { + const f = fixture() + const assistant = `${JSON.stringify({ type: 'response_item', payload: { role: 'assistant', type: 'message', content: [] } })}\n` + writeFileSync(rolloutPath(f.baseDir, OTHER_ID), metaLine(OTHER_ID, f.cwd) + promptLine('其他工作的启动') + assistant + promptLine()) + await expect(findCodexSession({ baseDir: f.baseDir, cwd: f.cwd, startedAt: STARTED_AT, prompt: PROMPT })).resolves.toBeNull() + }) + + it('半写的第一行在后续通知丢失时仍会重读，支持大行和中文', async () => { + const f = fixture() + const file = rolloutPath(f.baseDir, MAIN_ID) + const line = metaLine(MAIN_ID, f.cwd, 'cli', undefined, '中文上下文'.repeat(20000)) writeFileSync(file, line.slice(0, 20)) - await tick() - appendFileSync(file, line.slice(20)) - - await expect(found).resolves.toBe(MAIN_ID) + const found = await prepare(f) + expect(found.initialSessionId).toBeNull() + appendFileSync(file, line.slice(20) + promptLine()) + await vi.advanceTimersByTimeAsync(1000) + await expect(found.sessionId).resolves.toBe(MAIN_ID) }) - it('中止后返回 null', async () => { - const baseDir = tempDir('codex-sessions-') - const cwd = tempDir('worktree-') - const ac = new AbortController() - const found = watchForCodexSession({ baseDir, cwd, signal: ac.signal }) - await tick() - ac.abort() + it('元数据完整但提示词尚未完整时继续等待', async () => { + const f = fixture() + const file = rolloutPath(f.baseDir, MAIN_ID) + const line = promptLine() + writeFileSync(file, metaLine(MAIN_ID, f.cwd) + promptLine('# AGENTS.md instructions') + line.slice(0, 40)) + const found = await prepare(f) + expect(found.initialSessionId).toBeNull() + appendFileSync(file, line.slice(40)) + await vi.advanceTimersByTimeAsync(1000) + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) - await expect(found).resolves.toBeNull() + it('软链接启动目录与真实目录匹配', async () => { + const f = fixture() + const link = path.join(tempDir(), 'cwd') + symlinkSync(f.cwd, link) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, await fsp.realpath(f.cwd)) + promptLine()) + await expect(findCodexSession({ baseDir: f.baseDir, cwd: link, startedAt: STARTED_AT, prompt: PROMPT })).resolves.toBe(MAIN_ID) + }) + + it('多个主会话满足同一启动信息时明确失败，不选最新', async () => { + const f = fixture() + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + writeFileSync(rolloutPath(f.baseDir, OTHER_ID), metaLine(OTHER_ID, f.cwd) + promptLine()) + await expect(prepare(f)).rejects.toThrow('多个符合本次启动信息') + expect(fakeWatcher.close).toHaveBeenCalledOnce() + }) + + it('监听建立失败时记录原因并继续通过扫描确认', async () => { + const f = fixture() + const err = new Error('watch unavailable') + vi.mocked(watch).mockImplementationOnce(() => { + throw err + }) + const onWarning = vi.fn() + const found = await watchForCodexSession({ baseDir: f.baseDir, cwd: f.cwd, startedAt: STARTED_AT, prompt: PROMPT, signal: f.ac.signal, onWarning }) + expect(onWarning).toHaveBeenCalledWith(err) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + await vi.advanceTimersByTimeAsync(1000) + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) + + it('监听运行中失败也保持扫描，并关闭失效监听', async () => { + const f = fixture() + const found = await prepare(f) + fakeWatcher.emit('error', new Error('watch stopped')) + expect(fakeWatcher.close).toHaveBeenCalledOnce() + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + await vi.advanceTimersByTimeAsync(1000) + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) + + it('会话目录尚不存在时建立目录，之后能够捕获', async () => { + const f = fixture() + f.baseDir = path.join(f.baseDir, 'new', 'sessions') + const found = await prepare(f) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + notices?.() + await expect(found.sessionId).resolves.toBe(MAIN_ID) + }) + + it('会话根路径不是目录时准备明确失败', async () => { + const f = fixture() + f.baseDir = path.join(f.baseDir, 'not-dir') + writeFileSync(f.baseDir, '') + await expect(prepare(f)).rejects.toThrow() + expect(watch).not.toHaveBeenCalled() + }) + + it('中止会关闭监听和补偿定时器，且不再捕获', async () => { + const f = fixture() + const found = await prepare(f) + f.ac.abort() + await expect(found.sessionId).resolves.toBeNull() + expect(fakeWatcher.close).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + writeFileSync(rolloutPath(f.baseDir, MAIN_ID), metaLine(MAIN_ID, f.cwd) + promptLine()) + notices?.() + await expect(found.sessionId).resolves.toBeNull() + }) + + it('准备前已中止时不创建监听', async () => { + const f = fixture() + f.ac.abort() + await expect((await prepare(f)).sessionId).resolves.toBeNull() + expect(watch).not.toHaveBeenCalled() }) }) diff --git a/vscode/src/cc/codexSessionWatcher.ts b/vscode/src/cc/codexSessionWatcher.ts index 5a2ea62..3251769 100644 --- a/vscode/src/cc/codexSessionWatcher.ts +++ b/vscode/src/cc/codexSessionWatcher.ts @@ -1,147 +1,286 @@ -/** - * 等 codex 为指定目录新开的主会话落盘，拿到它的会话 id。 - * - * 审查终端里跑的是 codex 界面，读不到它的输出，只能盯 - * `~/.codex/sessions/YYYY/MM/DD/rollout--.jsonl`。 - * codex 在第一轮对话开始时才建这个文件，启动慢（外部工具服务起得慢等） - * 可能拖好几分钟，所以不设时间上限，由调用方在终端关掉时中止。 - * - * 同一时间可能有别的 codex 也在建文件（别的工单、审查里派出的子助手）， - * 只认第一行 session_meta 里 cwd 是本目录、且不是子助手的那个。 - */ +/** 按启动时间、目录和启动提示词确认 codex 主会话，文件通知只负责提醒扫描。 */ +import type { FileHandle } from 'node:fs/promises' import { Buffer } from 'node:buffer' import * as fs from 'node:fs' import { promises as fsp } from 'node:fs' import * as path from 'node:path' -/** UUID at the tail of `rollout--.jsonl`. v7 UUIDs are still 8-4-4-4-12. */ +/** UUID at the tail of `rollout--.jsonl`. */ export const ROLLOUT_UUID_REGEX = /-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.jsonl$/i -/** 第一行 session_meta 通常几十 KB；超过这个长度还没换行就当它不是会话文件。 */ -const FIRST_LINE_MAX_BYTES = 8 * 1024 * 1024 +const LINE_MAX_BYTES = 8 * 1024 * 1024 +const SCAN_INTERVAL_MS = 1000 -/** 读文件第一行；还没写到换行时返回 null，交给下一次文件变化再读。 */ -async function readCompleteFirstLine(filePath: string): Promise { - const fh = await fsp.open(filePath, 'r') - try { - const chunks: Buffer[] = [] - let total = 0 - while (total < FIRST_LINE_MAX_BYTES) { - const buf = Buffer.alloc(64 * 1024) - const { bytesRead } = await fh.read(buf, 0, buf.length, total) - if (bytesRead === 0) - return null - const chunk = buf.subarray(0, bytesRead) - const nl = chunk.indexOf(0x0A) - if (nl !== -1) { - chunks.push(chunk.subarray(0, nl)) - return Buffer.concat(chunks).toString('utf8') - } - chunks.push(chunk) - total += bytesRead +export interface CodexSessionLaunch { + cwd: string + startedAt: number + prompt: string +} + +interface FileProgress { + offset: number + sessionId?: string + decided?: boolean + matched?: boolean +} + +/** 未写出完整的一行时保留读取位置，下次扫描继续读。 */ +async function readCompleteLine(fh: FileHandle, offset: number, signal?: AbortSignal): Promise<{ line: string, nextOffset: number } | null> { + const chunks: Buffer[] = [] + let total = 0 + while (total < LINE_MAX_BYTES) { + if (signal?.aborted) + return null + const buf = Buffer.alloc(Math.min(64 * 1024, LINE_MAX_BYTES - total)) + const { bytesRead } = await fh.read(buf, 0, buf.length, offset + total) + if (bytesRead === 0) + return null + const chunk = buf.subarray(0, bytesRead) + const nl = chunk.indexOf(0x0A) + if (nl !== -1) { + chunks.push(chunk.subarray(0, nl)) + return { line: Buffer.concat(chunks).toString('utf8'), nextOffset: offset + total + nl + 1 } } - return '' + chunks.push(chunk) + total += bytesRead } - finally { - await fh.close() + throw new Error(`codex 会话记录单行超过 ${LINE_MAX_BYTES} 字节`) +} + +interface SessionRecord { + type?: unknown + payload?: { + type?: unknown + id?: unknown + cwd?: unknown + source?: unknown + timestamp?: unknown + role?: unknown + content?: unknown } } -/** 第一行是本目录主会话的 session_meta 时返回会话 id，否则 null。 */ -function matchMainSession(firstLine: string, cwds: Set): string | null { - let parsed: unknown +function parseLine(line: string): SessionRecord | null { try { - parsed = JSON.parse(firstLine) + const obj: unknown = JSON.parse(line) + return obj !== null && typeof obj === 'object' ? obj as SessionRecord : null } catch { return null } - const obj = parsed as { type?: unknown, payload?: { id?: unknown, cwd?: unknown, source?: unknown } } | null +} + +function matchMainSession(line: string, cwds: Set, startedAt: number): string | null { + const obj = parseLine(line) if (obj?.type !== 'session_meta' || !obj.payload) return null - const { id, cwd, source } = obj.payload - if (typeof id !== 'string' || typeof cwd !== 'string') + const { id, cwd, source, timestamp } = obj.payload + if (typeof id !== 'string' || typeof cwd !== 'string' || typeof source !== 'string' || typeof timestamp !== 'string') return null - // 子助手会话的 source 是 { subagent: ... }，主会话是 "cli" 这类字符串。 - if (typeof source === 'object' && source !== null) + const createdAt = Date.parse(timestamp) + if (!Number.isFinite(createdAt) || createdAt < startedAt) return null return cwds.has(path.resolve(cwd)) ? id : null } -/** - * 盯 `baseDir` 下新出现的会话文件，返回 `cwd` 目录下主会话的 id。 - * `signal` 中止、目录不存在或盯文件出错时返回 null。 - */ -export async function watchForCodexSession(opts: { - /** 会话根目录，通常是 `~/.codex/sessions`。 */ +function matchesPrompt(obj: SessionRecord | null, prompt: string): boolean { + if (obj?.type !== 'response_item' || obj.payload?.role !== 'user' || !Array.isArray(obj.payload.content)) + return false + const text = obj.payload.content + .filter((part: { type?: unknown, text?: unknown } | null) => part?.type === 'input_text' && typeof part.text === 'string') + .map((part: { text: string }) => part.text) + .join('\n') + return text === prompt +} + +async function createScanner(baseDir: string, launch: CodexSessionLaunch, signal?: AbortSignal): Promise<() => Promise> { + if (!Number.isFinite(launch.startedAt) || !launch.prompt) + throw new Error('捕获 codex 会话必须提供有效的启动时间和提示词') + const cwds = new Set([path.resolve(launch.cwd), await fsp.realpath(launch.cwd)]) + const progress = new Map() + + async function inspect(filePath: string): Promise { + const state = progress.get(filePath) ?? { offset: 0 } + progress.set(filePath, state) + if (state.decided) + return null + if (state.matched) + return state.sessionId ?? null + let fh: FileHandle + try { + fh = await fsp.open(filePath, 'r') + } + catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') + return null + throw err + } + try { + for (;;) { + const next = await readCompleteLine(fh, state.offset, signal) + if (!next) + return null + state.offset = next.nextOffset + if (!state.sessionId) { + const id = matchMainSession(next.line, cwds, launch.startedAt) + if (!id) { + state.decided = true + return null + } + state.sessionId = id + } + else { + const obj = parseLine(next.line) + if (matchesPrompt(obj, launch.prompt)) { + state.matched = true + return state.sessionId + } + // 首轮助手已经开始工作，之后的用户消息不再是启动提示词。 + if (obj?.type === 'response_item' && (obj.payload?.role === 'assistant' + || ['reasoning', 'function_call', 'custom_tool_call'].includes(String(obj.payload?.type)))) { + state.decided = true + return null + } + } + } + } + finally { + await fh.close() + } + } + + return async () => { + const names = await fsp.readdir(baseDir, { recursive: true }) + const matches = new Set() + for (const name of names) { + if (signal?.aborted) + return null + if (!path.basename(name).startsWith('rollout-') || !ROLLOUT_UUID_REGEX.test(name)) + continue + const id = await inspect(path.join(baseDir, name)) + if (id) + matches.add(id) + } + if (matches.size > 1) + throw new Error(`存在多个符合本次启动信息的 codex 主会话：${[...matches].join(', ')}`) + return matches.values().next().value ?? null + } +} + +/** 一次磁盘扫描；与持续捕获共用相同的确认规则。 */ +export async function findCodexSession(opts: CodexSessionLaunch & { baseDir: string }): Promise { + const scan = await createScanner(opts.baseDir, opts) + return scan() +} + +/** 准备完成后才允许启动终端；已落盘的会话也会在准备阶段确认。 */ +export async function watchForCodexSession(opts: CodexSessionLaunch & { baseDir: string - /** codex 启动时所在目录。 */ - cwd: string signal: AbortSignal -}): Promise { + onWarning?: (err: unknown) => void +}): Promise<{ sessionId: Promise, initialSessionId: string | null }> { const { baseDir, signal } = opts if (signal.aborted) - return null - try { - if (!(await fsp.stat(baseDir)).isDirectory()) - return null - } - catch { - return null - } - // codex 记的是解析过软链接的路径，两种写法都认。 - const cwds = new Set([path.resolve(opts.cwd)]) - try { - cwds.add(await fsp.realpath(opts.cwd)) - } - catch { - // 目录已被删掉时只按原样比对。 - } + return { sessionId: Promise.resolve(null), initialSessionId: null } + await fsp.mkdir(baseDir, { recursive: true }) + const scan = await createScanner(baseDir, opts, signal) + // 建立监听前验证可读，避免把准备失败当成已经开始等待。 + await fsp.readdir(baseDir) + if (signal.aborted) + return { sessionId: Promise.resolve(null), initialSessionId: null } - return new Promise((resolve) => { - let settled = false - let watcher: fs.FSWatcher | null = null - // 已经看过第一行、确定不是目标的文件，不再重复读。 - const decided = new Set() - - function done(id: string | null): void { - if (settled) - return - settled = true - signal.removeEventListener('abort', onAbort) - watcher?.close() - resolve(id) - } - function onAbort(): void { - done(null) - } - signal.addEventListener('abort', onAbort) - - const inspect = async (name: string): Promise => { - const firstLine = await readCompleteFirstLine(path.join(baseDir, name)).catch(() => null) - if (settled || firstLine === null || decided.has(name)) - return - decided.add(name) - const id = matchMainSession(firstLine, cwds) - if (id) - done(id) - } - - try { - watcher = fs.watch(baseDir, { persistent: false, recursive: true }, (_event, filename) => { - if (settled || !filename) - return - const name = filename.toString() - const base = path.basename(name) - if (!base.startsWith('rollout-') || !base.endsWith('.jsonl') || decided.has(name)) - return - void inspect(name) - }) - watcher.on('error', () => done(null)) - } - catch { - done(null) - } + let resolveSession!: (id: string | null) => void + let rejectSession!: (err: unknown) => void + const sessionId = new Promise((resolve, reject) => { + resolveSession = resolve + rejectSession = reject }) + let settled = false + let watcher: fs.FSWatcher | undefined + let timer: ReturnType | undefined + let scanning = false + let scanQueued = false + + function cleanup(): void { + signal.removeEventListener('abort', onAbort) + watcher?.close() + clearInterval(timer) + } + function done(id: string | null, err?: unknown): void { + if (settled) + return + settled = true + cleanup() + if (err) + rejectSession(err) + else + resolveSession(id) + } + function onAbort(): void { + done(null) + } + async function rescan(): Promise { + if (settled) + return + scanQueued = true + if (scanning) + return + scanning = true + try { + while (scanQueued && !settled) { + scanQueued = false + const id = await scan() + if (id) + done(id) + } + } + catch (err) { + done(null, err) + } + finally { + scanning = false + } + } + + signal.addEventListener('abort', onAbort) + try { + watcher = fs.watch(baseDir, { persistent: false, recursive: true }, () => { + void rescan() + }) + watcher.on('error', (err) => { + watcher?.close() + watcher = undefined + opts.onWarning?.(err) + }) + } + catch (err) { + opts.onWarning?.(err) + } + // 准备时的扫描与通知/定时扫描不并发，读取位置只由一个扫描推进。 + scanning = true + let initialSessionId: string | null + try { + initialSessionId = await scan() + if (initialSessionId) + done(initialSessionId) + } + catch (err) { + settled = true + cleanup() + resolveSession(null) + throw err + } + finally { + scanning = false + } + if (!settled) { + timer = setInterval(() => { + void rescan() + }, SCAN_INTERVAL_MS) + timer.unref() + if (scanQueued) + void rescan() + } + return { sessionId, initialSessionId: settled && signal.aborted ? null : initialSessionId } } diff --git a/vscode/src/panel/KanbanPanel.ts b/vscode/src/panel/KanbanPanel.ts index e07e248..e4d2c88 100644 --- a/vscode/src/panel/KanbanPanel.ts +++ b/vscode/src/panel/KanbanPanel.ts @@ -1097,6 +1097,10 @@ export class KanbanWebviewPanel { loadIssues({ host, token, owner, repo, workspaceRoot, scope }), this.loadYouTrackSafe(), ]) + this.issueRefs = new Map( + [...giteaIssues, ...youtrackList].map(i => [i.number, { source: i.source ?? 'gitea', externalId: i.externalId }] as const), + ) + await sessions.recoverOpenReviewSessions(this, [...giteaIssues, ...youtrackList].map(i => i.number)) // 本机记录(会话 id / worktree 等)叠加在共享状态之上,标出本机没有 // 会话文件的会话 id,再补 live terminal 态。 const issues = this.withLiveTerminalTabState( @@ -1104,9 +1108,6 @@ export class KanbanWebviewPanel { overlayLocalIssueState(this.context, [...giteaIssues, ...youtrackList], workspaceRoot), ), ) - this.issueRefs = new Map( - issues.map(i => [i.number, { source: i.source ?? 'gitea', externalId: i.externalId }] as const), - ) this.me = me this.issueOwners = new Map( issues.map(i => [i.number, { poster: i.poster, assignees: i.assignees }] as const), @@ -1350,6 +1351,7 @@ export class KanbanWebviewPanel { } private dispose(): void { + sessions.stopReviewSessionCaptures(this) KanbanWebviewPanel.current = undefined webhookCoordinator.setActivePanel(undefined) this.gitStateDisposable?.dispose() diff --git a/vscode/src/panel/handlers/issues.ts b/vscode/src/panel/handlers/issues.ts index 3f04655..be5fd4f 100644 --- a/vscode/src/panel/handlers/issues.ts +++ b/vscode/src/panel/handlers/issues.ts @@ -2024,6 +2024,8 @@ export async function handleClearSession(panel: KanbanWebviewPanel, issueNumber: return try { + if (kind === 'review') + await sessions.cancelReviewSessionCapture(panel, issueNumber) await panel.mergeIssueState(issueNumber, { [field]: '' }) } catch (err) { diff --git a/vscode/src/panel/handlers/reviewSessionCapture.test.ts b/vscode/src/panel/handlers/reviewSessionCapture.test.ts new file mode 100644 index 0000000..80dc12d --- /dev/null +++ b/vscode/src/panel/handlers/reviewSessionCapture.test.ts @@ -0,0 +1,263 @@ +import type { ExtensionContext, Terminal } from 'vscode' +import type { KanbanWebviewPanel } from '../KanbanPanel' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { getLocalIssueState } from '../../issues/localState' +import { + cancelReviewSessionCapture, + codexProcessStartedAt, + interactiveCodexLaunch, + prepareReviewSessionCapture, + recoverOpenReviewSessions, + recoverReviewSessionCapture, + stopReviewSessionCaptures, +} from './reviewSessionCapture' + +const mocks = vi.hoisted(() => ({ + watch: vi.fn(), + exec: vi.fn(), + log: vi.fn(), + closeListeners: new Set<(terminal: Terminal) => void>(), + dispose: vi.fn(), +})) + +vi.mock('vscode', () => ({ + window: { + onDidCloseTerminal: (listener: (terminal: Terminal) => void) => { + mocks.closeListeners.add(listener) + return { + dispose: () => { + mocks.dispose() + mocks.closeListeners.delete(listener) + }, + } + }, + }, +})) +vi.mock('node:child_process', () => ({ execFile: mocks.exec })) +vi.mock('../../cc/codexSessionWatcher', () => ({ watchForCodexSession: mocks.watch })) +vi.mock('../../logging/logger', () => ({ logger: { add: mocks.log } })) +vi.mock('../../sessions/sessionFiles', () => ({ defaultCodexSessionsDir: () => '/codex/sessions' })) + +const SID = '01a0f858-8af3-7d21-9ca5-3b63413f9efa' +const STARTED_AT = Date.parse('Fri Oct 2 00:42:25 2026') +const PROMPT = '开启`ADHD`模式，审查 PR #470。\n不要合并。\n' +const LAUNCH = { cwd: '/worktree', startedAt: STARTED_AT, prompt: PROMPT } +const panels: KanbanWebviewPanel[] = [] +let resolveSession!: (id: string | null) => void + +function fixture(source: 'gitea' | 'youtrack' = 'gitea') { + const store = new Map() + const context = { + workspaceState: { + get: vi.fn((key: string) => store.get(key)), + update: vi.fn(async (key: string, value: unknown) => { + if (value === undefined) + store.delete(key) + else + store.set(key, value) + }), + }, + } as unknown as ExtensionContext + const terminal = { + creationOptions: { + name: 'issue-466-审查', + cwd: LAUNCH.cwd, + shellPath: '/bin/zsh', + shellArgs: ['-ilc', 'exec "$@"', 'zsh', 'codex', '-c', 'model=gpt-6-astra', '-c', 'model_reasoning_effort=xhigh', '--dangerously-bypass-approvals-and-sandbox', PROMPT], + }, + processId: Promise.resolve(43567), + exitStatus: undefined, + sendText: vi.fn(), + show: vi.fn(), + } as unknown as Terminal + const panel = { + context, + refFor: vi.fn((number: number) => ({ source, number })), + reviewTerminals: new Map(), + trackSessionTerminal: vi.fn(), + findExistingTerminal: vi.fn(() => terminal), + postMessage: vi.fn(), + mergeIssueState: vi.fn(() => { throw new Error('remote unavailable') }), + } as unknown as KanbanWebviewPanel + panels.push(panel) + return { panel, context, store, terminal, launchKey: `superpowers.reviewSessionLaunch.${source}:466` } +} + +beforeEach(() => { + vi.clearAllMocks() + mocks.exec.mockImplementation((_cmd, args, _options, callback) => { + callback(null, args.at(-1) === 'lstart=' ? 'Fri Oct 2 00:42:25 2026\n' : 'codex\n') + }) + mocks.watch.mockImplementation(async ({ signal }) => { + const sessionId = new Promise((resolve) => { + resolveSession = resolve + }) + signal.addEventListener('abort', () => resolveSession(null)) + return { sessionId, initialSessionId: null } + }) +}) + +afterEach(() => { + for (const panel of panels.splice(0)) + stopReviewSessionCaptures(panel) + mocks.closeListeners.clear() + vi.useRealTimers() +}) + +describe('reviewSessionCapture', () => { + it('准备完成后绑定终端，直接保存本机ID，不依赖远端', async () => { + const f = fixture('youtrack') + const prepared = await prepareReviewSessionCapture(f.panel, 466, LAUNCH) + expect(f.store.get(f.launchKey)).toEqual(LAUNCH) + expect(getLocalIssueState(f.context, 'youtrack', 466).reviewSessionId).toBeUndefined() + const captured = prepared.attach(f.terminal) + resolveSession(SID) + await expect(captured).resolves.toBe(SID) + expect(getLocalIssueState(f.context, 'youtrack', 466).reviewSessionId).toBe(SID) + expect(f.panel.mergeIssueState).not.toHaveBeenCalled() + expect(f.panel.reviewTerminals.get(SID)).toBe(f.terminal) + expect(f.panel.postMessage).toHaveBeenCalledWith({ type: 'issue/patch', issueNumber: 466, patch: { reviewSessionId: SID } }) + expect(f.store.has(f.launchKey)).toBe(false) + expect(mocks.dispose).toHaveBeenCalledOnce() + }) + + it('刷新已有终端时按进程启动信息补回ID，不发送再审消息', async () => { + const f = fixture() + mocks.watch.mockResolvedValue({ sessionId: Promise.resolve(SID), initialSessionId: SID }) + await recoverOpenReviewSessions(f.panel, [466]) + expect(mocks.watch).toHaveBeenCalledWith(expect.objectContaining({ ...LAUNCH, startedAt: STARTED_AT })) + expect(getLocalIssueState(f.context, 'gitea', 466).reviewSessionId).toBe(SID) + expect(f.terminal.sendText).not.toHaveBeenCalled() + expect(f.terminal.show).not.toHaveBeenCalled() + }) + + it('尚未落盘的恢复立即返回，之后后台补回ID', async () => { + const f = fixture() + await expect(recoverReviewSessionCapture(f.panel, 466, f.terminal)).resolves.toBeNull() + expect(getLocalIssueState(f.context, 'gitea', 466).reviewSessionId).toBeUndefined() + resolveSession(SID) + await vi.waitFor(() => expect(getLocalIssueState(f.context, 'gitea', 466).reviewSessionId).toBe(SID)) + }) + + it('用户清除留下的墓碑不得自动补回', async () => { + const f = fixture() + f.store.set('superpowers.localIssueState', { 'gitea:466': { reviewSessionId: '' } }) + await expect(recoverReviewSessionCapture(f.panel, 466, f.terminal)).resolves.toBeNull() + expect(mocks.exec).not.toHaveBeenCalled() + expect(mocks.watch).not.toHaveBeenCalled() + expect(getLocalIssueState(f.context, 'gitea', 466).reviewSessionId).toBe('') + }) + + it('用户明确新建的审查可以替换之前清除留下的墓碑', async () => { + const f = fixture() + f.store.set('superpowers.localIssueState', { 'gitea:466': { reviewSessionId: '' } }) + const prepared = await prepareReviewSessionCapture(f.panel, 466, LAUNCH) + const captured = prepared.attach(f.terminal) + resolveSession(SID) + await expect(captured).resolves.toBe(SID) + expect(getLocalIssueState(f.context, 'gitea', 466).reviewSessionId).toBe(SID) + }) + + it('清除时中止待捕获并删除启动记录，之后不写回ID', async () => { + const f = fixture() + const prepared = await prepareReviewSessionCapture(f.panel, 466, LAUNCH) + const captured = prepared.attach(f.terminal) + await cancelReviewSessionCapture(f.panel, 466) + resolveSession(SID) + await expect(captured).resolves.toBeNull() + expect(f.store.has(f.launchKey)).toBe(false) + expect(getLocalIssueState(f.context, 'gitea', 466).reviewSessionId).toBeUndefined() + }) + + it('准备期间被取消，不允许稍后继续创建终端', async () => { + const f = fixture() + let ready!: () => void + mocks.watch.mockImplementationOnce(() => new Promise((resolve) => { + ready = () => resolve({ sessionId: Promise.resolve(null), initialSessionId: null }) + })) + const preparing = prepareReviewSessionCapture(f.panel, 466, LAUNCH) + const rejected = expect(preparing).rejects.toThrow('已中止') + await vi.waitFor(() => expect(mocks.watch).toHaveBeenCalled()) + await cancelReviewSessionCapture(f.panel, 466) + ready() + await rejected + expect(f.store.has(f.launchKey)).toBe(false) + }) + + it('恢复查询进程期间被取消，不启动新的后台捕获', async () => { + const f = fixture() + const pending: Array<{ field: string, callback: (err: null, output: string) => void }> = [] + mocks.exec.mockImplementation((_cmd, args, _options, callback) => { + pending.push({ field: args.at(-1), callback }) + }) + const recovering = recoverReviewSessionCapture(f.panel, 466, f.terminal) + await vi.waitFor(() => expect(pending).toHaveLength(2)) + await cancelReviewSessionCapture(f.panel, 466) + for (const item of pending) + item.callback(null, item.field === 'lstart=' ? 'Fri Oct 2 00:42:25 2026' : 'codex') + await expect(recovering).resolves.toBeNull() + expect(mocks.watch).not.toHaveBeenCalled() + }) + + it('同时触发恢复只检查一次进程和会话', async () => { + const f = fixture() + const a = recoverReviewSessionCapture(f.panel, 466, f.terminal) + const b = recoverReviewSessionCapture(f.panel, 466, f.terminal) + expect(a).toBe(b) + await a + expect(mocks.exec).toHaveBeenCalledTimes(2) + expect(mocks.watch).toHaveBeenCalledOnce() + }) + + it('准备失败时清理启动记录并传播原因', async () => { + const f = fixture() + mocks.watch.mockRejectedValueOnce(new Error('sessions unreadable')) + await expect(prepareReviewSessionCapture(f.panel, 466, LAUNCH)).rejects.toThrow('sessions unreadable') + expect(f.store.has(f.launchKey)).toBe(false) + }) + + it('终端关闭时停止捕获并清理记录', async () => { + const f = fixture() + const prepared = await prepareReviewSessionCapture(f.panel, 466, LAUNCH) + const captured = prepared.attach(f.terminal) + for (const listener of mocks.closeListeners) + listener(f.terminal) + await expect(captured).resolves.toBeNull() + expect(f.store.has(f.launchKey)).toBe(false) + }) + + it('无法确认进程时记录原因，不误认现存会话', async () => { + const f = fixture() + mocks.exec.mockImplementation((_cmd, args, _options, callback) => { + callback(null, args.at(-1) === 'lstart=' ? 'Fri Oct 2 00:42:25 2026' : 'zsh') + }) + await expect(recoverReviewSessionCapture(f.panel, 466, f.terminal)).resolves.toBeNull() + expect(mocks.watch).not.toHaveBeenCalled() + expect(mocks.log).toHaveBeenCalledWith(expect.objectContaining({ level: 'warn', details: expect.stringContaining('不是可确认启动时间的 codex') })) + }) + + it('resume终端不按新进程时间扫描历史会话', async () => { + const f = fixture() + const terminal = { ...f.terminal, creationOptions: { cwd: '/worktree', shellArgs: ['-ilc', 'exec "$@"', 'zsh', 'codex', 'resume', '--dangerously-bypass-approvals-and-sandbox', SID] } } as unknown as Terminal + expect(interactiveCodexLaunch(terminal)).toBeNull() + await expect(recoverReviewSessionCapture(f.panel, 466, terminal)).resolves.toBeNull() + expect(mocks.exec).not.toHaveBeenCalled() + }) + + it('进程检查使用短超时与固定英文日期，非法PID直接拒绝', async () => { + await expect(codexProcessStartedAt(43567)).resolves.toBe(STARTED_AT) + expect(mocks.exec).toHaveBeenCalledWith('ps', expect.any(Array), expect.objectContaining({ timeout: 2000, env: expect.objectContaining({ LC_ALL: 'C' }) }), expect.any(Function)) + await expect(codexProcessStartedAt(-1)).rejects.toThrow('没有有效的进程编号') + }) + + it('终端始终没有提供进程编号时刷新也会及时结束', async () => { + vi.useFakeTimers() + const f = fixture() + const terminal = { ...f.terminal, processId: new Promise(() => {}) } as unknown as Terminal + const recovering = recoverReviewSessionCapture(f.panel, 466, terminal) + await vi.advanceTimersByTimeAsync(2000) + await expect(recovering).resolves.toBeNull() + expect(mocks.exec).not.toHaveBeenCalled() + expect(mocks.log).toHaveBeenCalledWith(expect.objectContaining({ details: expect.stringContaining('在 2 秒内未提供进程编号') })) + }) +}) diff --git a/vscode/src/panel/handlers/reviewSessionCapture.ts b/vscode/src/panel/handlers/reviewSessionCapture.ts new file mode 100644 index 0000000..ddb3a00 --- /dev/null +++ b/vscode/src/panel/handlers/reviewSessionCapture.ts @@ -0,0 +1,277 @@ +import type { Terminal } from 'vscode' +import type { CodexSessionLaunch } from '../../cc/codexSessionWatcher' +import type { KanbanWebviewPanel } from '../KanbanPanel' +import { execFile } from 'node:child_process' +import * as path from 'node:path' +import process from 'node:process' +import { window } from 'vscode' +import { watchForCodexSession } from '../../cc/codexSessionWatcher' +import { getLocalIssueState, mergeLocalIssueState } from '../../issues/localState' +import { logger } from '../../logging/logger' +import { defaultCodexSessionsDir } from '../../sessions/sessionFiles' + +interface ReviewCapture { + abort: AbortController + launch: CodexSessionLaunch + terminal?: Terminal + captured?: Promise +} + +const captures = new WeakMap>() +const recoveries = new WeakMap }>>() + +function captureMap(panel: KanbanWebviewPanel): Map { + let map = captures.get(panel) + if (!map) { + map = new Map() + captures.set(panel, map) + } + return map +} + +function launchKey(panel: KanbanWebviewPanel, issueNumber: number): string { + return `superpowers.reviewSessionLaunch.${panel.refFor(issueNumber).source ?? 'gitea'}:${issueNumber}` +} + +/** 用户清除关联时先中止捕获，防止稍后落盘的文件把关联补回来。 */ +export async function cancelReviewSessionCapture(panel: KanbanWebviewPanel, issueNumber: number): Promise { + const map = captureMap(panel) + recoveries.get(panel)?.get(issueNumber)?.abort.abort() + map.get(issueNumber)?.abort.abort() + map.delete(issueNumber) + await panel.context.workspaceState.update(launchKey(panel, issueNumber), undefined) +} + +export function stopReviewSessionCaptures(panel: KanbanWebviewPanel): void { + for (const recovery of recoveries.get(panel)?.values() ?? []) + recovery.abort.abort() + for (const capture of captureMap(panel).values()) + capture.abort.abort() + captures.delete(panel) + recoveries.delete(panel) +} + +function reportCaptureFailure(issueNumber: number, err: unknown): void { + logger.add({ + level: 'warn', + source: 'review', + message: `确认审查会话失败 (#${issueNumber})`, + details: err instanceof Error ? err.message : String(err), + }) +} + +/** 保存本机关联不依赖远端认证或网络。 */ +async function persistCapture(panel: KanbanWebviewPanel, issueNumber: number, capture: ReviewCapture, id: string): Promise { + if (capture.abort.signal.aborted || captureMap(panel).get(issueNumber) !== capture || !capture.terminal) + return null + await mergeLocalIssueState(panel.context, panel.refFor(issueNumber).source, issueNumber, { reviewSessionId: id }) + if (capture.abort.signal.aborted) + return null + panel.reviewTerminals.set(id, capture.terminal) + panel.trackSessionTerminal(capture.terminal, issueNumber, 'review') + panel.postMessage({ type: 'issue/patch', issueNumber, patch: { reviewSessionId: id } }) + await panel.context.workspaceState.update(launchKey(panel, issueNumber), undefined) + logger.add({ level: 'info', source: 'review', message: `已捕获审查会话 ${id} (#${issueNumber})` }) + return id +} + +/** 返回前完成目录确认和首次扫描，调用方随后才创建终端。 */ +export async function prepareReviewSessionCapture(panel: KanbanWebviewPanel, issueNumber: number, launch: CodexSessionLaunch): Promise<{ + attach: (terminal: Terminal) => Promise + initialSessionId: string | null +}> { + captureMap(panel).get(issueNumber)?.abort.abort() + const capture: ReviewCapture = { abort: new AbortController(), launch } + captureMap(panel).set(issueNumber, capture) + try { + await panel.context.workspaceState.update(launchKey(panel, issueNumber), launch) + const watch = await watchForCodexSession({ + ...launch, + baseDir: defaultCodexSessionsDir(), + signal: capture.abort.signal, + onWarning: (err) => { + logger.add({ + level: 'warn', + source: 'review', + message: `审查会话文件通知不可用，继续定时扫描 (#${issueNumber})`, + details: err instanceof Error ? err.message : String(err), + }) + }, + }) + if (capture.abort.signal.aborted) + throw new Error('审查会话捕获已中止') + // 捕获准备好后，attach 同步绑定终端和关闭事件，再异步保存结果。 + let closeListener: { dispose: () => void } | undefined + const attach = (terminal: Terminal): Promise => { + capture.terminal = terminal + closeListener = window.onDidCloseTerminal((closed) => { + if (closed === terminal) { + void cancelReviewSessionCapture(panel, issueNumber).catch(err => reportCaptureFailure(issueNumber, err)) + } + }) + capture.captured = watch.sessionId + .then(id => id ? persistCapture(panel, issueNumber, capture, id) : null) + .catch((err) => { + reportCaptureFailure(issueNumber, err) + return null + }) + .finally(() => { + closeListener?.dispose() + if (captureMap(panel).get(issueNumber) === capture) + captureMap(panel).delete(issueNumber) + }) + return capture.captured + } + // 准备与终端创建之间失败时也有拒绝处理，避免无人接收的失败 Promise。 + void watch.sessionId.catch(() => {}) + logger.add({ + level: 'info', + source: 'review', + message: `已准备审查会话捕获 (#${issueNumber}) cwd=${launch.cwd} startedAt=${new Date(launch.startedAt).toISOString()}`, + }) + return { attach, initialSessionId: watch.initialSessionId } + } + catch (err) { + capture.abort.abort() + if (captureMap(panel).get(issueNumber) === capture) + await cancelReviewSessionCapture(panel, issueNumber) + throw err + } +} + +/** 只认由登录 shell 直接启动的交互式 codex，resume 不走新会话发现。 */ +export function interactiveCodexLaunch(terminal: Pick): Pick | null { + const options = terminal.creationOptions + if (!('shellArgs' in options) || !Array.isArray(options.shellArgs) || !('cwd' in options) || !options.cwd) + return null + const args = options.shellArgs + if (args[0] !== '-ilc' || args[1] !== 'exec "$@"' || args[3] !== 'codex') + return null + let i = 4 + while (i < args.length) { + if (args[i] === '-c') + i += 2 + else if (args[i] === '--dangerously-bypass-approvals-and-sandbox') + i++ + else + break + } + if (i !== args.length - 1 || !args[i] || ['resume', 'exec', 'review'].includes(args[i])) + return null + return { cwd: typeof options.cwd === 'string' ? options.cwd : options.cwd.fsPath, prompt: args[i] } +} + +function processField(pid: number, field: string, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + execFile('ps', ['-p', String(pid), '-o', `${field}=`], { + timeout: 2000, + encoding: 'utf8', + env: { ...process.env, LC_ALL: 'C' }, + signal, + }, (err, stdout) => { + if (err) + reject(err) + else + resolve(stdout.trim()) + }) + }) +} + +/** 存活终端的真实进程边界用于恢复，不能把刷新时间当作会话启动时间。 */ +export async function codexProcessStartedAt(pid: number, signal?: AbortSignal): Promise { + if (!Number.isInteger(pid) || pid <= 0) + throw new Error('审查终端没有有效的进程编号') + const [started, command] = await Promise.all([processField(pid, 'lstart', signal), processField(pid, 'comm', signal)]) + const startedAt = Date.parse(started) + if (path.basename(command) !== 'codex' || !Number.isFinite(startedAt)) + throw new Error(`审查终端进程不是可确认启动时间的 codex：${command || '空进程'} ${started}`) + return startedAt +} + +function terminalProcessId(terminal: Terminal, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + finish(new Error('审查终端在 2 秒内未提供进程编号')) + }, 2000) + timer.unref() + function finish(err?: unknown, pid?: number): void { + clearTimeout(timer) + signal.removeEventListener('abort', onAbort) + if (err) + reject(err) + else + resolve(pid) + } + function onAbort(): void { + finish(new Error('审查会话恢复已中止')) + } + signal.addEventListener('abort', onAbort) + if (signal.aborted) + onAbort() + else + void Promise.resolve(terminal.processId).then(pid => finish(undefined, pid), err => finish(err)) + }) +} + +/** 已有文件立即补关联；尚未落盘则后台等待，刷新不会等完整审查运行。 */ +async function recoverReviewSessionCaptureInner(panel: KanbanWebviewPanel, issueNumber: number, terminal: Terminal, signal: AbortSignal): Promise { + const local = getLocalIssueState(panel.context, panel.refFor(issueNumber).source, issueNumber) + if (local.reviewSessionId === '' || captureMap(panel).has(issueNumber)) + return null + if (local.reviewSessionId) + return local.reviewSessionId + const launch = interactiveCodexLaunch(terminal) + if (!launch) + return null + try { + const pid = await terminalProcessId(terminal, signal) + if (pid === undefined) + throw new Error('审查终端未提供进程编号，无法确认本次启动边界') + const processStartedAt = await codexProcessStartedAt(pid, signal) + if (signal.aborted || terminal.exitStatus !== undefined) + return null + const stored = panel.context.workspaceState.get(launchKey(panel, issueNumber)) + const startedAt = stored?.prompt === launch.prompt && stored.cwd === launch.cwd && Number.isFinite(stored.startedAt) + ? Math.max(stored.startedAt, processStartedAt) + : processStartedAt + const prepared = await prepareReviewSessionCapture(panel, issueNumber, { ...launch, startedAt }) + if (signal.aborted) { + await cancelReviewSessionCapture(panel, issueNumber) + return null + } + const captured = prepared.attach(terminal) + return prepared.initialSessionId ? captured : null + } + catch (err) { + if (!signal.aborted) + reportCaptureFailure(issueNumber, err) + return null + } +} + +export function recoverReviewSessionCapture(panel: KanbanWebviewPanel, issueNumber: number, terminal: Terminal): Promise { + let map = recoveries.get(panel) + if (!map) { + map = new Map() + recoveries.set(panel, map) + } + const existing = map.get(issueNumber) + if (existing) + return existing.result + const abort = new AbortController() + const result = recoverReviewSessionCaptureInner(panel, issueNumber, terminal, abort.signal) + .finally(() => { + if (map.get(issueNumber)?.result === result) + map.delete(issueNumber) + }) + map.set(issueNumber, { abort, result }) + return result +} + +export async function recoverOpenReviewSessions(panel: KanbanWebviewPanel, issueNumbers: readonly number[]): Promise { + await Promise.all(issueNumbers.map(async (issueNumber) => { + const terminal = panel.findExistingTerminal(`issue-${issueNumber}-审查`) + if (terminal) + await recoverReviewSessionCapture(panel, issueNumber, terminal) + })) +} diff --git a/vscode/src/panel/handlers/sessions.ts b/vscode/src/panel/handlers/sessions.ts index 9d80bce..934b32c 100644 --- a/vscode/src/panel/handlers/sessions.ts +++ b/vscode/src/panel/handlers/sessions.ts @@ -10,7 +10,6 @@ import { loginForToken } from '../../auth/identity' import { getToken } from '../../auth/secrets' import { buildCcArgv } from '../../cc/ccCommand' import { buildCodexArgv } from '../../cc/codexCommand' -import { watchForCodexSession } from '../../cc/codexSessionWatcher' import { resolveProfilePath } from '../../cc/profiles' import { renderBrainstormPrompt, renderImplementPlanPrompt, renderPrototypePrompt, renderReviewPrompt, renderTestPrompt } from '../../cc/prompts' import { sessionLaunch } from '../../cc/sessionTerminal' @@ -19,8 +18,9 @@ import { detectRepo } from '../../git/remote' import { ensureWorktree, resolveWorktreePath } from '../../git/worktree' import { DEFAULT_WORKTREE_TEMPLATE, deriveSlug, resolveWorktreeDir } from '../../git/worktreePath' import { updateIssueAssignees } from '../../gitea/api' +import { getLocalIssueState } from '../../issues/localState' import { logger } from '../../logging/logger' -import { claudeSessionFileExists, codexSessionFileExists, defaultCodexSessionsDir } from '../../sessions/sessionFiles' +import { claudeSessionFileExists, codexSessionFileExists } from '../../sessions/sessionFiles' import { getSettings } from '../../settings/store' import { webhookCoordinator } from '../../webhook/coordinator' import { makeNonce } from '../../webview/html' @@ -28,8 +28,11 @@ import { decideAutoReviewPath } from '../autoReviewDecision' import { pickBrainstormContinuePath } from '../brainstormContinue' import { cleanupFeatureBranch } from './branchCleanup' import { planResumeCwd, resumeTerminalName } from './resumeTarget' +import { cancelReviewSessionCapture, prepareReviewSessionCapture, recoverReviewSessionCapture } from './reviewSessionCapture' import { submitToSessionTerminal } from './terminals' +export { cancelReviewSessionCapture, recoverOpenReviewSessions, stopReviewSessionCaptures } from './reviewSessionCapture' + /** * 会话入口统一的 state 读取：读失败只记 warn，按空状态继续——提示词模板里的 * 条件块会自然走 else 分支，没必要为一次网络抖动拦住开会话。 @@ -415,21 +418,13 @@ async function triggerAutoReviewTabUnlocked(panel: KanbanWebviewPanel, opts: { const terminalName = `issue-${opts.issueNumber}-审查` const existing = panel.findExistingTerminal(terminalName) - // Pre-fetch the persisted reviewSessionId from issue state JSON so we - // can pick the right branch below. Any failure (no remote / no token / - // network blip) MUST NOT break the whole review trigger — fall through - // with existingSessionId = undefined and we'll spin up a fresh codex. - let existingSessionId: string | undefined + // 本机关联优先；未保存过时才从工单状态读取，或确认存活终端的启动记录。 + const localState = getLocalIssueState(panel.context, panel.refFor(opts.issueNumber).source, opts.issueNumber) + let existingSessionId = nonEmptyString(localState.reviewSessionId) try { - const remote = await detectRepo(opts.workspaceRoot) - if (remote) { - const token = await getToken(panel.context, remote.host) - if (token) { - const state = await panel.readIssueState(opts.issueNumber) - const sid = state?.reviewSessionId - if (typeof sid === 'string' && sid.length > 0) - existingSessionId = sid - } + if (localState.reviewSessionId === undefined) { + const state = await panel.readIssueState(opts.issueNumber) + existingSessionId = nonEmptyString(state.reviewSessionId) } } catch (err) { @@ -441,6 +436,9 @@ async function triggerAutoReviewTabUnlocked(panel: KanbanWebviewPanel, opts: { }) } + if (existing && !existingSessionId && localState.reviewSessionId !== '') + existingSessionId = (await recoverReviewSessionCapture(panel, opts.issueNumber, existing)) ?? undefined + const reviewPath = decideAutoReviewPath({ tabAlive: !!existing, reviewSessionId: existingSessionId, @@ -520,45 +518,43 @@ async function triggerAutoReviewTabUnlocked(panel: KanbanWebviewPanel, opts: { const { themeColor, iconUri } = await panel.resolveIssueIcon(opts.issueNumber) const settings = getSettings(panel.context) - // 先盯会话目录再拉起 codex，免得文件在盯上之前就建好了。拿到的 id 存成 - // reviewSessionId：PR 再更新时靠它往这个窗口发「再审」，窗口关了也能接着聊。 - // codex 启动可能拖好几分钟，一直盯到这个终端关掉为止。 - const watchAbort = new AbortController() - const watcherPromise = watchForCodexSession({ - baseDir: defaultCodexSessionsDir(), + const capture = await prepareReviewSessionCapture(panel, opts.issueNumber, { cwd: effectiveCwd, - signal: watchAbort.signal, + startedAt: Date.now(), + prompt: opts.prompt, }) // Interactive TUI (`codex` without `exec`/`review` subcommands): keeps // the terminal alive after the run so users can follow up with codex, // mirroring how claude implementation/brainstorm sessions stay open. - // 提示词模板自带 /review 前缀，不在这里再拼。 - const terminal = window.createTerminal({ - name: terminalName, - cwd: effectiveCwd, - location: panel.resolveTerminalLocation(false), - iconPath: iconUri, - color: themeColor, - ...sessionLaunch(buildCodexArgv({ - mode: 'interactive', - prompt: opts.prompt, - model: settings.codexModel, - reasoningEffort: settings.codexReasoningEffort, - })), - }) - const closeListener = window.onDidCloseTerminal((closed) => { - if (closed === terminal) - watchAbort.abort() - }) - void watcherPromise.finally(() => closeListener.dispose()) + let terminal: Terminal + try { + terminal = window.createTerminal({ + name: terminalName, + cwd: effectiveCwd, + location: panel.resolveTerminalLocation(false), + iconPath: iconUri, + color: themeColor, + ...sessionLaunch(buildCodexArgv({ + mode: 'interactive', + prompt: opts.prompt, + model: settings.codexModel, + reasoningEffort: settings.codexReasoningEffort, + })), + }) + void capture.attach(terminal) + panel.trackSessionTerminal(terminal, opts.issueNumber, 'review') + terminal.show(false) + } + catch (err) { + await cancelReviewSessionCapture(panel, opts.issueNumber) + throw err + } logger.add({ level: 'info', source: 'terminal', message: `已创建审查终端 "${terminalName}" cwd=${effectiveCwd}，codex review 已启动`, }) - panel.trackSessionTerminal(terminal, opts.issueNumber, 'review') - terminal.show(false) if (cwdFallback) { panel.postMessage({ @@ -569,57 +565,6 @@ async function triggerAutoReviewTabUnlocked(panel: KanbanWebviewPanel, opts: { dismissOnTimer: 5000, }) } - - // Fire-and-forget: when the watcher resolves with a thread_id, persist - // it to state JSON + push an issue/patch so the webview shows the resume - // button. Failures here don't break the review run itself. - void watcherPromise.then(async (threadId) => { - if (!threadId) { - logger.add({ - level: 'warn', - source: 'review', - message: `审查终端关闭前没等到会话文件 (#${opts.issueNumber})`, - }) - return - } - logger.add({ - level: 'info', - source: 'review', - message: `已捕获审查会话 ${threadId} (#${opts.issueNumber})`, - }) - try { - const workspaceRoot = workspace.workspaceFolders?.[0]?.uri.fsPath - if (!workspaceRoot) - return - const remote = await detectRepo(workspaceRoot) - if (!remote) - return - const token = await getToken(panel.context, remote.host) - if (!token) - return - await panel.mergeIssueState(opts.issueNumber, { reviewSessionId: threadId }) - panel.postMessage({ - type: 'issue/patch', - issueNumber: opts.issueNumber, - patch: { reviewSessionId: threadId }, - }) - } - catch (err) { - logger.add({ - level: 'warn', - source: 'review', - message: `持久化 reviewSessionId 失败 (#${opts.issueNumber})`, - details: err instanceof Error ? err.message : String(err), - }) - } - }).catch((err) => { - logger.add({ - level: 'warn', - source: 'review', - message: `审查 session watcher 异常 (#${opts.issueNumber})`, - details: err instanceof Error ? err.message : String(err), - }) - }) } /**