superwork/vscode/src/cc/codexSessionWatcher.ts

/** 按启动时间、目录和启动提示词确认 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-<ISO>-<uuid>.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

const LINE_MAX_BYTES = 8 * 1024 * 1024
const SCAN_INTERVAL_MS = 1000

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 }
    }
    chunks.push(chunk)
    total += bytesRead
  }
  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
  }
}

function parseLine(line: string): SessionRecord | null {
  try {
    const obj: unknown = JSON.parse(line)
    return obj !== null && typeof obj === 'object' ? obj as SessionRecord : null
  }
  catch {
    return null
  }
}

function matchMainSession(line: string, cwds: Set<string>, startedAt: number): string | null {
  const obj = parseLine(line)
  if (obj?.type !== 'session_meta' || !obj.payload)
    return null
  const { id, cwd, source, timestamp } = obj.payload
  if (typeof id !== 'string' || typeof cwd !== 'string' || typeof source !== 'string' || typeof timestamp !== 'string')
    return null
  const createdAt = Date.parse(timestamp)
  if (!Number.isFinite(createdAt) || createdAt < startedAt)
    return null
  return cwds.has(path.resolve(cwd)) ? id : null
}

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<string | null>> {
  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<string, FileProgress>()

  async function inspect(filePath: string): Promise<string | null> {
    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<string>()
    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<string | null> {
  const scan = await createScanner(opts.baseDir, opts)
  return scan()
}

/** 准备完成后才允许启动终端；已落盘的会话也会在准备阶段确认。 */
export async function watchForCodexSession(opts: CodexSessionLaunch & {
  baseDir: string
  signal: AbortSignal
  onWarning?: (err: unknown) => void
}): Promise<{ sessionId: Promise<string | null>, initialSessionId: string | null }> {
  const { baseDir, signal } = opts
  if (signal.aborted)
    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 }

  let resolveSession!: (id: string | null) => void
  let rejectSession!: (err: unknown) => void
  const sessionId = new Promise<string | null>((resolve, reject) => {
    resolveSession = resolve
    rejectSession = reject
  })
  let settled = false
  let watcher: fs.FSWatcher | undefined
  let timer: ReturnType<typeof setInterval> | 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<void> {
    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 }
}