mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-05 06:24:37 +00:00
refactor: extract session scanner base class for code reuse
Create BaseSessionScanner abstract class to consolidate common session scanning logic shared between Claude and Codex session scanners. Both scanners now inherit from this base, reducing duplication and providing consistent patterns for file watching, event processing, and state management.
This commit is contained in:
weishu
@@ -1,10 +1,9 @@
|
||||
import { InvalidateSync } from "@/utils/sync";
|
||||
import { RawJSONLines, RawJSONLinesSchema } from "../types";
|
||||
import { join } from "node:path";
|
||||
import { basename, join } from "node:path";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { logger } from "@/ui/logger";
|
||||
import { startFileWatcher } from "@/modules/watcher/startFileWatcher";
|
||||
import { getProjectPath } from "./path";
|
||||
import { BaseSessionScanner, SessionFileScanEntry, SessionFileScanResult, SessionFileScanStats } from "@/modules/common/session/BaseSessionScanner";
|
||||
|
||||
/**
|
||||
* Known internal Claude Code event types that should be silently skipped.
|
||||
@@ -18,133 +17,139 @@ const INTERNAL_CLAUDE_EVENT_TYPES = new Set([
|
||||
]);
|
||||
|
||||
export async function createSessionScanner(opts: {
|
||||
sessionId: string | null,
|
||||
workingDirectory: string
|
||||
onMessage: (message: RawJSONLines) => void
|
||||
sessionId: string | null;
|
||||
workingDirectory: string;
|
||||
onMessage: (message: RawJSONLines) => void;
|
||||
}) {
|
||||
|
||||
// Resolve project directory
|
||||
const projectDir = getProjectPath(opts.workingDirectory);
|
||||
|
||||
// Finished, pending finishing and current session
|
||||
let finishedSessions = new Set<string>();
|
||||
let pendingSessions = new Set<string>();
|
||||
let currentSessionId: string | null = null;
|
||||
let watchers = new Map<string, (() => void)>();
|
||||
let processedMessageKeys = new Set<string>();
|
||||
|
||||
// Mark existing messages as processed and start watching the initial session
|
||||
if (opts.sessionId) {
|
||||
let messages = await readSessionLog(projectDir, opts.sessionId);
|
||||
logger.debug(`[SESSION_SCANNER] Marking ${messages.length} existing messages as processed from session ${opts.sessionId}`);
|
||||
for (let m of messages) {
|
||||
processedMessageKeys.add(messageKey(m));
|
||||
}
|
||||
// IMPORTANT: Also start watching the initial session file because Claude Code
|
||||
// may continue writing to it even after creating a new session with --resume
|
||||
// (agent tasks and other updates can still write to the original session file)
|
||||
currentSessionId = opts.sessionId;
|
||||
}
|
||||
|
||||
// Main sync function
|
||||
const sync = new InvalidateSync(async () => {
|
||||
// logger.debug(`[SESSION_SCANNER] Syncing...`);
|
||||
|
||||
// Collect session ids - include ALL sessions that have watchers
|
||||
// This ensures we continue processing sessions that Claude Code may still write to
|
||||
let sessions: string[] = [];
|
||||
for (let p of pendingSessions) {
|
||||
sessions.push(p);
|
||||
}
|
||||
if (currentSessionId && !pendingSessions.has(currentSessionId)) {
|
||||
sessions.push(currentSessionId);
|
||||
}
|
||||
// Also process sessions that have active watchers (they may still receive updates)
|
||||
for (let [sessionId] of watchers) {
|
||||
if (!sessions.includes(sessionId)) {
|
||||
sessions.push(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
// Process sessions
|
||||
for (let session of sessions) {
|
||||
const sessionMessages = await readSessionLog(projectDir, session);
|
||||
let skipped = 0;
|
||||
let sent = 0;
|
||||
for (let file of sessionMessages) {
|
||||
let key = messageKey(file);
|
||||
if (processedMessageKeys.has(key)) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
processedMessageKeys.add(key);
|
||||
logger.debug(`[SESSION_SCANNER] Sending new message: type=${file.type}, uuid=${file.type === 'summary' ? file.leafUuid : file.uuid}`);
|
||||
opts.onMessage(file);
|
||||
sent++;
|
||||
}
|
||||
if (sessionMessages.length > 0) {
|
||||
logger.debug(`[SESSION_SCANNER] Session ${session}: found=${sessionMessages.length}, skipped=${skipped}, sent=${sent}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Move pending sessions to finished sessions (but keep processing them via watchers)
|
||||
for (let p of sessions) {
|
||||
if (pendingSessions.has(p)) {
|
||||
pendingSessions.delete(p);
|
||||
finishedSessions.add(p);
|
||||
}
|
||||
}
|
||||
|
||||
// Update watchers for all sessions
|
||||
for (let p of sessions) {
|
||||
if (!watchers.has(p)) {
|
||||
logger.debug(`[SESSION_SCANNER] Starting watcher for session: ${p}`);
|
||||
watchers.set(p, startFileWatcher(join(projectDir, `${p}.jsonl`), () => { sync.invalidate(); }));
|
||||
}
|
||||
}
|
||||
const scanner = new ClaudeSessionScanner({
|
||||
sessionId: opts.sessionId,
|
||||
workingDirectory: opts.workingDirectory,
|
||||
onMessage: opts.onMessage
|
||||
});
|
||||
await sync.invalidateAndAwait();
|
||||
|
||||
// Periodic sync
|
||||
const intervalId = setInterval(() => { sync.invalidate(); }, 3000);
|
||||
await scanner.start();
|
||||
|
||||
// Public interface
|
||||
return {
|
||||
cleanup: async () => {
|
||||
clearInterval(intervalId);
|
||||
for (let w of watchers.values()) {
|
||||
w();
|
||||
}
|
||||
watchers.clear();
|
||||
await sync.invalidateAndAwait();
|
||||
sync.stop();
|
||||
await scanner.cleanup();
|
||||
},
|
||||
onNewSession: (sessionId: string) => {
|
||||
if (currentSessionId === sessionId) {
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId} is the same as the current session, skipping`);
|
||||
return;
|
||||
}
|
||||
if (finishedSessions.has(sessionId)) {
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId} is already finished, skipping`);
|
||||
return;
|
||||
}
|
||||
if (pendingSessions.has(sessionId)) {
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId} is already pending, skipping`);
|
||||
return;
|
||||
}
|
||||
if (currentSessionId) {
|
||||
pendingSessions.add(currentSessionId);
|
||||
}
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId}`)
|
||||
currentSessionId = sessionId;
|
||||
sync.invalidate();
|
||||
},
|
||||
}
|
||||
scanner.onNewSession(sessionId);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export type SessionScanner = ReturnType<typeof createSessionScanner>;
|
||||
|
||||
|
||||
class ClaudeSessionScanner extends BaseSessionScanner<RawJSONLines> {
|
||||
private readonly projectDir: string;
|
||||
private readonly onMessage: (message: RawJSONLines) => void;
|
||||
private readonly finishedSessions = new Set<string>();
|
||||
private readonly pendingSessions = new Set<string>();
|
||||
private currentSessionId: string | null;
|
||||
private readonly scannedSessions = new Set<string>();
|
||||
|
||||
constructor(opts: { sessionId: string | null; workingDirectory: string; onMessage: (message: RawJSONLines) => void }) {
|
||||
super({ intervalMs: 3000 });
|
||||
this.projectDir = getProjectPath(opts.workingDirectory);
|
||||
this.onMessage = opts.onMessage;
|
||||
this.currentSessionId = opts.sessionId;
|
||||
}
|
||||
|
||||
public onNewSession(sessionId: string): void {
|
||||
if (this.currentSessionId === sessionId) {
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId} is the same as the current session, skipping`);
|
||||
return;
|
||||
}
|
||||
if (this.finishedSessions.has(sessionId)) {
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId} is already finished, skipping`);
|
||||
return;
|
||||
}
|
||||
if (this.pendingSessions.has(sessionId)) {
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId} is already pending, skipping`);
|
||||
return;
|
||||
}
|
||||
if (this.currentSessionId) {
|
||||
this.pendingSessions.add(this.currentSessionId);
|
||||
}
|
||||
logger.debug(`[SESSION_SCANNER] New session: ${sessionId}`);
|
||||
this.currentSessionId = sessionId;
|
||||
this.invalidate();
|
||||
}
|
||||
|
||||
protected async initialize(): Promise<void> {
|
||||
if (!this.currentSessionId) {
|
||||
return;
|
||||
}
|
||||
const sessionFile = this.sessionFilePath(this.currentSessionId);
|
||||
const { events, totalLines } = await readSessionLog(sessionFile, 0);
|
||||
logger.debug(`[SESSION_SCANNER] Marking ${events.length} existing messages as processed from session ${this.currentSessionId}`);
|
||||
const keys = events.map((entry) => messageKey(entry.event));
|
||||
this.seedProcessedKeys(keys);
|
||||
this.setCursor(sessionFile, totalLines);
|
||||
}
|
||||
|
||||
protected async beforeScan(): Promise<void> {
|
||||
this.scannedSessions.clear();
|
||||
}
|
||||
|
||||
protected async findSessionFiles(): Promise<string[]> {
|
||||
const files = new Set<string>();
|
||||
for (const sessionId of this.pendingSessions) {
|
||||
files.add(this.sessionFilePath(sessionId));
|
||||
}
|
||||
if (this.currentSessionId && !this.pendingSessions.has(this.currentSessionId)) {
|
||||
files.add(this.sessionFilePath(this.currentSessionId));
|
||||
}
|
||||
for (const watched of this.getWatchedFiles()) {
|
||||
files.add(watched);
|
||||
}
|
||||
return [...files];
|
||||
}
|
||||
|
||||
protected async parseSessionFile(filePath: string, cursor: number): Promise<SessionFileScanResult<RawJSONLines>> {
|
||||
const sessionId = sessionIdFromPath(filePath);
|
||||
if (sessionId) {
|
||||
this.scannedSessions.add(sessionId);
|
||||
}
|
||||
const { events, totalLines } = await readSessionLog(filePath, cursor);
|
||||
return {
|
||||
events,
|
||||
nextCursor: totalLines
|
||||
};
|
||||
}
|
||||
|
||||
protected generateEventKey(event: RawJSONLines): string {
|
||||
return messageKey(event);
|
||||
}
|
||||
|
||||
protected async handleFileScan(stats: SessionFileScanStats<RawJSONLines>): Promise<void> {
|
||||
for (const message of stats.events) {
|
||||
const id = message.type === 'summary' ? message.leafUuid : message.uuid;
|
||||
logger.debug(`[SESSION_SCANNER] Sending new message: type=${message.type}, uuid=${id}`);
|
||||
this.onMessage(message);
|
||||
}
|
||||
if (stats.parsedCount > 0) {
|
||||
const sessionId = sessionIdFromPath(stats.filePath) ?? 'unknown';
|
||||
logger.debug(`[SESSION_SCANNER] Session ${sessionId}: found=${stats.parsedCount}, skipped=${stats.skippedCount}, sent=${stats.newCount}`);
|
||||
}
|
||||
}
|
||||
|
||||
protected async afterScan(): Promise<void> {
|
||||
for (const sessionId of this.scannedSessions) {
|
||||
if (this.pendingSessions.has(sessionId)) {
|
||||
this.pendingSessions.delete(sessionId);
|
||||
this.finishedSessions.add(sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private sessionFilePath(sessionId: string): string {
|
||||
return join(this.projectDir, `${sessionId}.jsonl`);
|
||||
}
|
||||
}
|
||||
|
||||
//
|
||||
// Helpers
|
||||
//
|
||||
@@ -164,22 +169,28 @@ function messageKey(message: RawJSONLines): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Read and parse session log file
|
||||
* Returns only valid conversation messages, silently skipping internal events
|
||||
* Read and parse session log file.
|
||||
* Returns only valid conversation messages, silently skipping internal events.
|
||||
*/
|
||||
async function readSessionLog(projectDir: string, sessionId: string): Promise<RawJSONLines[]> {
|
||||
const expectedSessionFile = join(projectDir, `${sessionId}.jsonl`);
|
||||
logger.debug(`[SESSION_SCANNER] Reading session file: ${expectedSessionFile}`);
|
||||
async function readSessionLog(filePath: string, startLine: number): Promise<{ events: SessionFileScanEntry<RawJSONLines>[]; totalLines: number }> {
|
||||
logger.debug(`[SESSION_SCANNER] Reading session file: ${filePath}`);
|
||||
let file: string;
|
||||
try {
|
||||
file = await readFile(expectedSessionFile, 'utf-8');
|
||||
file = await readFile(filePath, 'utf-8');
|
||||
} catch (error) {
|
||||
logger.debug(`[SESSION_SCANNER] Session file not found: ${expectedSessionFile}`);
|
||||
return [];
|
||||
logger.debug(`[SESSION_SCANNER] Session file not found: ${filePath}`);
|
||||
return { events: [], totalLines: startLine };
|
||||
}
|
||||
let lines = file.split('\n');
|
||||
let messages: RawJSONLines[] = [];
|
||||
for (let l of lines) {
|
||||
const lines = file.split('\n');
|
||||
const hasTrailingEmpty = lines.length > 0 && lines[lines.length - 1] === '';
|
||||
const totalLines = hasTrailingEmpty ? lines.length - 1 : lines.length;
|
||||
let effectiveStartLine = startLine;
|
||||
if (effectiveStartLine > totalLines) {
|
||||
effectiveStartLine = 0;
|
||||
}
|
||||
const messages: SessionFileScanEntry<RawJSONLines>[] = [];
|
||||
for (let index = effectiveStartLine; index < lines.length; index += 1) {
|
||||
const l = lines[index];
|
||||
try {
|
||||
if (l.trim() === '') {
|
||||
continue;
|
||||
@@ -194,15 +205,22 @@ async function readSessionLog(projectDir: string, sessionId: string): Promise<Ra
|
||||
|
||||
let parsed = RawJSONLinesSchema.safeParse(message);
|
||||
if (!parsed.success) {
|
||||
// Unknown message types are silently skipped
|
||||
// They will be tracked by processedMessageKeys to avoid reprocessing
|
||||
// Unknown message types are silently skipped.
|
||||
continue;
|
||||
}
|
||||
messages.push(parsed.data);
|
||||
messages.push({ event: parsed.data, lineIndex: index });
|
||||
} catch (e) {
|
||||
logger.debug(`[SESSION_SCANNER] Error processing message: ${e}`);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
return messages;
|
||||
return { events: messages, totalLines };
|
||||
}
|
||||
|
||||
function sessionIdFromPath(filePath: string): string | null {
|
||||
const base = basename(filePath);
|
||||
if (!base.endsWith('.jsonl')) {
|
||||
return null;
|
||||
}
|
||||
return base.slice(0, -'.jsonl'.length);
|
||||
}
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
import { InvalidateSync } from '@/utils/sync';
|
||||
import { startFileWatcher } from '@/modules/watcher/startFileWatcher';
|
||||
import { logger } from '@/ui/logger';
|
||||
import { join, relative, resolve, sep } from 'node:path';
|
||||
import { homedir } from 'node:os';
|
||||
import { readFile, readdir, stat } from 'node:fs/promises';
|
||||
import type { CodexSessionEvent } from './codexEventConverter';
|
||||
import { BaseSessionScanner, SessionFileScanEntry, SessionFileScanResult, SessionFileScanStats } from "@/modules/common/session/BaseSessionScanner";
|
||||
import { logger } from "@/ui/logger";
|
||||
import { join, relative, resolve, sep } from "node:path";
|
||||
import { homedir } from "node:os";
|
||||
import { readFile, readdir, stat } from "node:fs/promises";
|
||||
import type { CodexSessionEvent } from "./codexEventConverter";
|
||||
|
||||
interface CodexSessionScannerOptions {
|
||||
sessionId: string | null;
|
||||
@@ -34,33 +33,9 @@ type Candidate = {
|
||||
const DEFAULT_SESSION_START_WINDOW_MS = 2 * 60 * 1000;
|
||||
|
||||
export async function createCodexSessionScanner(opts: CodexSessionScannerOptions): Promise<CodexSessionScanner> {
|
||||
const codexHomeDir = process.env.CODEX_HOME || join(homedir(), '.codex');
|
||||
const sessionsRoot = join(codexHomeDir, 'sessions');
|
||||
|
||||
const processedLineCounts = new Map<string, number>();
|
||||
const watchers = new Map<string, () => void>();
|
||||
const sessionIdByFile = new Map<string, string>();
|
||||
const sessionCwdByFile = new Map<string, string>();
|
||||
const sessionTimestampByFile = new Map<string, number>();
|
||||
const pendingEventsByFile = new Map<string, PendingEvents>();
|
||||
const sessionMetaParsed = new Set<string>();
|
||||
|
||||
let activeSessionId: string | null = opts.sessionId;
|
||||
let reportedSessionId: string | null = opts.sessionId;
|
||||
let isClosing = false;
|
||||
let matchFailed = false;
|
||||
|
||||
const targetCwd = opts.cwd && opts.cwd.trim().length > 0 ? normalizePath(opts.cwd) : null;
|
||||
const referenceTimestampMs = opts.startupTimestampMs ?? Date.now();
|
||||
const sessionStartWindowMs = opts.sessionStartWindowMs ?? DEFAULT_SESSION_START_WINDOW_MS;
|
||||
const matchDeadlineMs = referenceTimestampMs + sessionStartWindowMs;
|
||||
const sessionDatePrefixes = targetCwd
|
||||
? getSessionDatePrefixes(referenceTimestampMs, sessionStartWindowMs)
|
||||
: null;
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Init: targetCwd=${targetCwd ?? 'none'} startupTs=${new Date(referenceTimestampMs).toISOString()} windowMs=${sessionStartWindowMs}`);
|
||||
|
||||
if (!targetCwd && !opts.sessionId) {
|
||||
matchFailed = true;
|
||||
const message = 'No cwd provided for Codex session matching; refusing to fallback.';
|
||||
logger.warn(`[CODEX_SESSION_SCANNER] ${message}`);
|
||||
opts.onSessionMatchFailed?.(message);
|
||||
@@ -70,35 +45,212 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions
|
||||
};
|
||||
}
|
||||
|
||||
function reportSessionId(sessionId: string): void {
|
||||
if (reportedSessionId === sessionId) {
|
||||
const scanner = new CodexSessionScannerImpl(opts, targetCwd);
|
||||
await scanner.start();
|
||||
|
||||
return {
|
||||
cleanup: async () => {
|
||||
await scanner.cleanup();
|
||||
},
|
||||
onNewSession: (sessionId: string) => {
|
||||
scanner.onNewSession(sessionId);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
class CodexSessionScannerImpl extends BaseSessionScanner<CodexSessionEvent> {
|
||||
private readonly sessionsRoot: string;
|
||||
private readonly onEvent: (event: CodexSessionEvent) => void;
|
||||
private readonly onSessionFound?: (sessionId: string) => void;
|
||||
private readonly onSessionMatchFailed?: (message: string) => void;
|
||||
private readonly sessionIdByFile = new Map<string, string>();
|
||||
private readonly sessionCwdByFile = new Map<string, string>();
|
||||
private readonly sessionTimestampByFile = new Map<string, number>();
|
||||
private readonly pendingEventsByFile = new Map<string, PendingEvents>();
|
||||
private readonly sessionMetaParsed = new Set<string>();
|
||||
private readonly fileEpochByPath = new Map<string, number>();
|
||||
private readonly targetCwd: string | null;
|
||||
private readonly referenceTimestampMs: number;
|
||||
private readonly sessionStartWindowMs: number;
|
||||
private readonly matchDeadlineMs: number;
|
||||
private readonly sessionDatePrefixes: Set<string> | null;
|
||||
|
||||
private activeSessionId: string | null;
|
||||
private reportedSessionId: string | null;
|
||||
private matchFailed = false;
|
||||
private bestWithinWindow: Candidate | null = null;
|
||||
|
||||
constructor(opts: CodexSessionScannerOptions, targetCwd: string | null) {
|
||||
super({ intervalMs: 2000 });
|
||||
const codexHomeDir = process.env.CODEX_HOME || join(homedir(), '.codex');
|
||||
this.sessionsRoot = join(codexHomeDir, 'sessions');
|
||||
this.onEvent = opts.onEvent;
|
||||
this.onSessionFound = opts.onSessionFound;
|
||||
this.onSessionMatchFailed = opts.onSessionMatchFailed;
|
||||
this.activeSessionId = opts.sessionId;
|
||||
this.reportedSessionId = opts.sessionId;
|
||||
this.targetCwd = targetCwd;
|
||||
this.referenceTimestampMs = opts.startupTimestampMs ?? Date.now();
|
||||
this.sessionStartWindowMs = opts.sessionStartWindowMs ?? DEFAULT_SESSION_START_WINDOW_MS;
|
||||
this.matchDeadlineMs = this.referenceTimestampMs + this.sessionStartWindowMs;
|
||||
this.sessionDatePrefixes = this.targetCwd
|
||||
? getSessionDatePrefixes(this.referenceTimestampMs, this.sessionStartWindowMs)
|
||||
: null;
|
||||
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Init: targetCwd=${this.targetCwd ?? 'none'} startupTs=${new Date(this.referenceTimestampMs).toISOString()} windowMs=${this.sessionStartWindowMs}`);
|
||||
}
|
||||
|
||||
public onNewSession(sessionId: string): void {
|
||||
if (this.activeSessionId === sessionId) {
|
||||
return;
|
||||
}
|
||||
reportedSessionId = sessionId;
|
||||
opts.onSessionFound?.(sessionId);
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Switching to new session: ${sessionId}`);
|
||||
this.setActiveSessionId(sessionId);
|
||||
this.invalidate();
|
||||
}
|
||||
|
||||
function setActiveSessionId(sessionId: string): void {
|
||||
activeSessionId = sessionId;
|
||||
reportSessionId(sessionId);
|
||||
if (targetCwd) {
|
||||
flushPendingEventsForSession(sessionId);
|
||||
} else {
|
||||
pendingEventsByFile.clear();
|
||||
protected shouldScan(): boolean {
|
||||
return !this.matchFailed;
|
||||
}
|
||||
|
||||
protected shouldWatchFile(filePath: string): boolean {
|
||||
if (!this.activeSessionId) {
|
||||
if (!this.targetCwd) {
|
||||
return false;
|
||||
}
|
||||
return this.getCandidateForFile(filePath) !== null;
|
||||
}
|
||||
const fileSessionId = this.sessionIdByFile.get(filePath);
|
||||
if (fileSessionId) {
|
||||
return fileSessionId === this.activeSessionId;
|
||||
}
|
||||
return filePath.endsWith(`-${this.activeSessionId}.jsonl`);
|
||||
}
|
||||
|
||||
protected async initialize(): Promise<void> {
|
||||
const files = await this.listSessionFiles(this.sessionsRoot);
|
||||
for (const filePath of files) {
|
||||
const { nextCursor } = await this.readSessionFile(filePath, 0);
|
||||
this.setCursor(filePath, nextCursor);
|
||||
if (this.shouldWatchFile(filePath)) {
|
||||
this.ensureWatcher(filePath);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function listSessionFiles(dir: string): Promise<string[]> {
|
||||
protected async beforeScan(): Promise<void> {
|
||||
this.bestWithinWindow = null;
|
||||
}
|
||||
|
||||
protected async findSessionFiles(): Promise<string[]> {
|
||||
const files = await this.listSessionFiles(this.sessionsRoot);
|
||||
return sortFilesByMtime(files);
|
||||
}
|
||||
|
||||
protected async parseSessionFile(filePath: string, cursor: number): Promise<SessionFileScanResult<CodexSessionEvent>> {
|
||||
if (this.shouldSkipFile(filePath)) {
|
||||
return { events: [], nextCursor: cursor };
|
||||
}
|
||||
return this.readSessionFile(filePath, cursor);
|
||||
}
|
||||
|
||||
protected generateEventKey(event: CodexSessionEvent, context: { filePath: string; lineIndex?: number }): string {
|
||||
const epoch = this.fileEpochByPath.get(context.filePath) ?? 0;
|
||||
const lineIndex = context.lineIndex ?? -1;
|
||||
return `${context.filePath}:${epoch}:${lineIndex}`;
|
||||
}
|
||||
|
||||
protected async handleFileScan(stats: SessionFileScanStats<CodexSessionEvent>): Promise<void> {
|
||||
const filePath = stats.filePath;
|
||||
const fileSessionId = this.sessionIdByFile.get(filePath) ?? null;
|
||||
|
||||
if (!this.activeSessionId && this.targetCwd) {
|
||||
this.appendPendingEvents(filePath, stats.events, fileSessionId);
|
||||
const candidate = this.getCandidateForFile(filePath);
|
||||
if (candidate) {
|
||||
if (!this.bestWithinWindow || candidate.score < this.bestWithinWindow.score) {
|
||||
this.bestWithinWindow = candidate;
|
||||
}
|
||||
}
|
||||
if (stats.newCount > 0) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Buffered ${stats.newCount} pending events from ${filePath}`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const emittedForFile = this.emitEvents(stats.events, fileSessionId);
|
||||
if (emittedForFile > 0) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emittedForFile} new events from ${filePath}`);
|
||||
}
|
||||
}
|
||||
|
||||
protected async afterScan(): Promise<void> {
|
||||
if (!this.activeSessionId && this.targetCwd) {
|
||||
if (this.bestWithinWindow) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Selected session ${this.bestWithinWindow.sessionId} within start window`);
|
||||
this.setActiveSessionId(this.bestWithinWindow.sessionId);
|
||||
} else if (Date.now() > this.matchDeadlineMs) {
|
||||
this.matchFailed = true;
|
||||
this.pendingEventsByFile.clear();
|
||||
const message = `No Codex session found within ${this.sessionStartWindowMs}ms for cwd ${this.targetCwd}; refusing fallback.`;
|
||||
logger.warn(`[CODEX_SESSION_SCANNER] ${message}`);
|
||||
this.onSessionMatchFailed?.(message);
|
||||
} else if (this.pendingEventsByFile.size > 0) {
|
||||
logger.debug('[CODEX_SESSION_SCANNER] No session candidate matched yet; pending events buffered');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private shouldSkipFile(filePath: string): boolean {
|
||||
if (!this.activeSessionId) {
|
||||
return false;
|
||||
}
|
||||
const fileSessionId = this.sessionIdByFile.get(filePath);
|
||||
if (fileSessionId && fileSessionId !== this.activeSessionId) {
|
||||
return true;
|
||||
}
|
||||
if (!fileSessionId && !filePath.endsWith(`-${this.activeSessionId}.jsonl`)) {
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private reportSessionId(sessionId: string): void {
|
||||
if (this.reportedSessionId === sessionId) {
|
||||
return;
|
||||
}
|
||||
this.reportedSessionId = sessionId;
|
||||
this.onSessionFound?.(sessionId);
|
||||
}
|
||||
|
||||
private setActiveSessionId(sessionId: string): void {
|
||||
this.activeSessionId = sessionId;
|
||||
this.reportSessionId(sessionId);
|
||||
const candidateFiles = this.getFilesForSession(sessionId);
|
||||
for (const filePath of candidateFiles) {
|
||||
if (this.shouldWatchFile(filePath)) {
|
||||
this.ensureWatcher(filePath);
|
||||
}
|
||||
}
|
||||
this.pruneWatchers(this.getWatchedFiles().filter((filePath) => this.shouldWatchFile(filePath)));
|
||||
if (this.targetCwd) {
|
||||
this.flushPendingEventsForSession(sessionId);
|
||||
} else {
|
||||
this.pendingEventsByFile.clear();
|
||||
}
|
||||
}
|
||||
|
||||
private async listSessionFiles(dir: string): Promise<string[]> {
|
||||
try {
|
||||
const entries = await readdir(dir, { withFileTypes: true });
|
||||
const results: string[] = [];
|
||||
for (const entry of entries) {
|
||||
const full = join(dir, entry.name);
|
||||
if (!shouldIncludeSessionPath(full, sessionsRoot, sessionDatePrefixes)) {
|
||||
if (!shouldIncludeSessionPath(full, this.sessionsRoot, this.sessionDatePrefixes)) {
|
||||
continue;
|
||||
}
|
||||
if (entry.isDirectory()) {
|
||||
results.push(...await listSessionFiles(full));
|
||||
results.push(...await this.listSessionFiles(full));
|
||||
} else if (entry.isFile() && entry.name.endsWith('.jsonl')) {
|
||||
results.push(full);
|
||||
}
|
||||
@@ -109,24 +261,26 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions
|
||||
}
|
||||
}
|
||||
|
||||
async function readSessionFile(filePath: string, startLine: number): Promise<{ events: CodexSessionEvent[]; totalLines: number }> {
|
||||
private async readSessionFile(filePath: string, startLine: number): Promise<SessionFileScanResult<CodexSessionEvent>> {
|
||||
let content: string;
|
||||
try {
|
||||
content = await readFile(filePath, 'utf-8');
|
||||
} catch (error) {
|
||||
return { events: [], totalLines: startLine };
|
||||
return { events: [], nextCursor: startLine };
|
||||
}
|
||||
|
||||
const events: CodexSessionEvent[] = [];
|
||||
const events: SessionFileScanEntry<CodexSessionEvent>[] = [];
|
||||
const lines = content.split('\n');
|
||||
const hasTrailingEmpty = lines.length > 0 && lines[lines.length - 1] === '';
|
||||
const totalLines = hasTrailingEmpty ? lines.length - 1 : lines.length;
|
||||
let effectiveStartLine = startLine;
|
||||
if (effectiveStartLine > totalLines) {
|
||||
effectiveStartLine = 0;
|
||||
const nextEpoch = (this.fileEpochByPath.get(filePath) ?? 0) + 1;
|
||||
this.fileEpochByPath.set(filePath, nextEpoch);
|
||||
}
|
||||
|
||||
const hasSessionMeta = sessionMetaParsed.has(filePath);
|
||||
const hasSessionMeta = this.sessionMetaParsed.has(filePath);
|
||||
const parseFrom = hasSessionMeta ? effectiveStartLine : 0;
|
||||
|
||||
for (let index = parseFrom; index < lines.length; index += 1) {
|
||||
@@ -135,66 +289,55 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
const parsed = JSON.parse(trimmed) as CodexSessionEvent;
|
||||
if (parsed?.type === 'session_meta') {
|
||||
const payload = asRecord(parsed.payload);
|
||||
const sessionId = payload ? asString(payload.id) : null;
|
||||
if (sessionId) {
|
||||
sessionIdByFile.set(filePath, sessionId);
|
||||
this.sessionIdByFile.set(filePath, sessionId);
|
||||
}
|
||||
const sessionCwd = payload ? asString(payload.cwd) : null;
|
||||
const normalizedCwd = sessionCwd ? normalizePath(sessionCwd) : null;
|
||||
if (normalizedCwd) {
|
||||
sessionCwdByFile.set(filePath, normalizedCwd);
|
||||
this.sessionCwdByFile.set(filePath, normalizedCwd);
|
||||
}
|
||||
const rawTimestamp = payload ? payload.timestamp : null;
|
||||
const sessionTimestamp = payload ? parseTimestamp(payload.timestamp) : null;
|
||||
if (sessionTimestamp !== null) {
|
||||
sessionTimestampByFile.set(filePath, sessionTimestamp);
|
||||
this.sessionTimestampByFile.set(filePath, sessionTimestamp);
|
||||
}
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Session meta: file=${filePath} cwd=${sessionCwd ?? 'none'} normalizedCwd=${normalizedCwd ?? 'none'} timestamp=${rawTimestamp ?? 'none'} parsedTs=${sessionTimestamp ?? 'none'}`);
|
||||
sessionMetaParsed.add(filePath);
|
||||
this.sessionMetaParsed.add(filePath);
|
||||
}
|
||||
if (index >= effectiveStartLine) {
|
||||
events.push(parsed);
|
||||
events.push({ event: parsed, lineIndex: index });
|
||||
}
|
||||
} catch (error) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Failed to parse line: ${error}`);
|
||||
}
|
||||
}
|
||||
|
||||
return { events, totalLines };
|
||||
return { events, nextCursor: totalLines };
|
||||
}
|
||||
|
||||
async function initializeProcessedMessages(): Promise<void> {
|
||||
const files = await listSessionFiles(sessionsRoot);
|
||||
for (const filePath of files) {
|
||||
const { totalLines } = await readSessionFile(filePath, 0);
|
||||
processedLineCounts.set(filePath, totalLines);
|
||||
if (!isClosing && !watchers.has(filePath)) {
|
||||
watchers.set(filePath, startFileWatcher(filePath, () => sync.invalidate()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function getCandidateForFile(filePath: string): Candidate | null {
|
||||
const sessionId = sessionIdByFile.get(filePath);
|
||||
private getCandidateForFile(filePath: string): Candidate | null {
|
||||
const sessionId = this.sessionIdByFile.get(filePath);
|
||||
if (!sessionId) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const fileCwd = sessionCwdByFile.get(filePath);
|
||||
if (targetCwd && fileCwd !== targetCwd) {
|
||||
const fileCwd = this.sessionCwdByFile.get(filePath);
|
||||
if (this.targetCwd && fileCwd !== this.targetCwd) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const sessionTimestamp = sessionTimestampByFile.get(filePath);
|
||||
const sessionTimestamp = this.sessionTimestampByFile.get(filePath);
|
||||
if (sessionTimestamp === undefined) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const diff = Math.abs(sessionTimestamp - referenceTimestampMs);
|
||||
if (diff > sessionStartWindowMs) {
|
||||
const diff = Math.abs(sessionTimestamp - this.referenceTimestampMs);
|
||||
if (diff > this.sessionStartWindowMs) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -204,11 +347,25 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions
|
||||
};
|
||||
}
|
||||
|
||||
function appendPendingEvents(filePath: string, events: CodexSessionEvent[], fileSessionId: string | null): void {
|
||||
private getFilesForSession(sessionId: string): string[] {
|
||||
const matches: string[] = [];
|
||||
for (const [filePath, storedSessionId] of this.sessionIdByFile.entries()) {
|
||||
if (storedSessionId === sessionId) {
|
||||
matches.push(filePath);
|
||||
}
|
||||
}
|
||||
if (matches.length > 0) {
|
||||
return matches;
|
||||
}
|
||||
const suffix = `-${sessionId}.jsonl`;
|
||||
return this.getWatchedFiles().filter((filePath) => filePath.endsWith(suffix));
|
||||
}
|
||||
|
||||
private appendPendingEvents(filePath: string, events: CodexSessionEvent[], fileSessionId: string | null): void {
|
||||
if (events.length === 0) {
|
||||
return;
|
||||
}
|
||||
const existing = pendingEventsByFile.get(filePath);
|
||||
const existing = this.pendingEventsByFile.get(filePath);
|
||||
if (existing) {
|
||||
existing.events.push(...events);
|
||||
if (!existing.fileSessionId && fileSessionId) {
|
||||
@@ -216,131 +373,47 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions
|
||||
}
|
||||
return;
|
||||
}
|
||||
pendingEventsByFile.set(filePath, {
|
||||
this.pendingEventsByFile.set(filePath, {
|
||||
events: [...events],
|
||||
fileSessionId
|
||||
});
|
||||
}
|
||||
|
||||
function emitEvents(events: CodexSessionEvent[], fileSessionId: string | null): number {
|
||||
private emitEvents(events: CodexSessionEvent[], fileSessionId: string | null): number {
|
||||
let emittedForFile = 0;
|
||||
for (const event of events) {
|
||||
const payload = asRecord(event.payload);
|
||||
const payloadSessionId = payload ? asString(payload.id) : null;
|
||||
const eventSessionId = payloadSessionId ?? fileSessionId ?? null;
|
||||
|
||||
if (activeSessionId && eventSessionId && eventSessionId !== activeSessionId) {
|
||||
if (this.activeSessionId && eventSessionId && eventSessionId !== this.activeSessionId) {
|
||||
continue;
|
||||
}
|
||||
|
||||
opts.onEvent(event);
|
||||
this.onEvent(event);
|
||||
emittedForFile += 1;
|
||||
}
|
||||
return emittedForFile;
|
||||
}
|
||||
|
||||
function flushPendingEventsForSession(sessionId: string): void {
|
||||
if (pendingEventsByFile.size === 0) {
|
||||
private flushPendingEventsForSession(sessionId: string): void {
|
||||
if (this.pendingEventsByFile.size === 0) {
|
||||
return;
|
||||
}
|
||||
let emitted = 0;
|
||||
for (const [filePath, pending] of pendingEventsByFile.entries()) {
|
||||
for (const [filePath, pending] of this.pendingEventsByFile.entries()) {
|
||||
const matches = (pending.fileSessionId && pending.fileSessionId === sessionId)
|
||||
|| filePath.endsWith(`-${sessionId}.jsonl`);
|
||||
if (!matches) {
|
||||
continue;
|
||||
}
|
||||
emitted += emitEvents(pending.events, pending.fileSessionId);
|
||||
emitted += this.emitEvents(pending.events, pending.fileSessionId);
|
||||
}
|
||||
pendingEventsByFile.clear();
|
||||
this.pendingEventsByFile.clear();
|
||||
if (emitted > 0) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emitted} pending events for session ${sessionId}`);
|
||||
}
|
||||
}
|
||||
|
||||
const sync = new InvalidateSync(async () => {
|
||||
if (isClosing || matchFailed) {
|
||||
return;
|
||||
}
|
||||
const files = await listSessionFiles(sessionsRoot);
|
||||
const sortedFiles = await sortFilesByMtime(files);
|
||||
let bestWithinWindow: Candidate | null = null;
|
||||
|
||||
for (const filePath of sortedFiles) {
|
||||
if (isClosing) {
|
||||
return;
|
||||
}
|
||||
if (!watchers.has(filePath)) {
|
||||
watchers.set(filePath, startFileWatcher(filePath, () => sync.invalidate()));
|
||||
}
|
||||
|
||||
const fileSessionId = sessionIdByFile.get(filePath);
|
||||
if (activeSessionId && fileSessionId && fileSessionId !== activeSessionId) {
|
||||
continue;
|
||||
}
|
||||
if (activeSessionId && !fileSessionId && !filePath.endsWith(`-${activeSessionId}.jsonl`)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const lastProcessedLine = processedLineCounts.get(filePath) ?? 0;
|
||||
const { events, totalLines } = await readSessionFile(filePath, lastProcessedLine);
|
||||
processedLineCounts.set(filePath, totalLines);
|
||||
const candidate = !activeSessionId && targetCwd ? getCandidateForFile(filePath) : null;
|
||||
if (!activeSessionId && targetCwd) {
|
||||
appendPendingEvents(filePath, events, fileSessionId ?? null);
|
||||
if (candidate) {
|
||||
if (!bestWithinWindow || candidate.score < bestWithinWindow.score) {
|
||||
bestWithinWindow = candidate;
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
const emittedForFile = emitEvents(events, fileSessionId ?? null);
|
||||
if (emittedForFile > 0) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emittedForFile} new events from ${filePath}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (!activeSessionId && targetCwd) {
|
||||
if (bestWithinWindow) {
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Selected session ${bestWithinWindow.sessionId} within start window`);
|
||||
setActiveSessionId(bestWithinWindow.sessionId);
|
||||
} else if (Date.now() > matchDeadlineMs) {
|
||||
matchFailed = true;
|
||||
pendingEventsByFile.clear();
|
||||
const message = `No Codex session found within ${sessionStartWindowMs}ms for cwd ${targetCwd}; refusing fallback.`;
|
||||
logger.warn(`[CODEX_SESSION_SCANNER] ${message}`);
|
||||
opts.onSessionMatchFailed?.(message);
|
||||
} else if (pendingEventsByFile.size > 0) {
|
||||
logger.debug('[CODEX_SESSION_SCANNER] No session candidate matched yet; pending events buffered');
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
await initializeProcessedMessages();
|
||||
await sync.invalidateAndAwait();
|
||||
const intervalId = setInterval(() => sync.invalidate(), 2000);
|
||||
|
||||
return {
|
||||
cleanup: async () => {
|
||||
isClosing = true;
|
||||
clearInterval(intervalId);
|
||||
sync.stop();
|
||||
for (const stop of watchers.values()) {
|
||||
stop();
|
||||
}
|
||||
watchers.clear();
|
||||
},
|
||||
onNewSession: (sessionId: string) => {
|
||||
if (activeSessionId === sessionId) {
|
||||
return;
|
||||
}
|
||||
logger.debug(`[CODEX_SESSION_SCANNER] Switching to new session: ${sessionId}`);
|
||||
setActiveSessionId(sessionId);
|
||||
sync.invalidate();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
async function sortFilesByMtime(files: string[]): Promise<string[]> {
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
import { InvalidateSync } from "@/utils/sync";
|
||||
import { startFileWatcher } from "@/modules/watcher/startFileWatcher";
|
||||
|
||||
export type SessionFileScanEntry<TEvent> = {
|
||||
event: TEvent;
|
||||
lineIndex?: number;
|
||||
};
|
||||
|
||||
export type SessionFileScanResult<TEvent> = {
|
||||
events: SessionFileScanEntry<TEvent>[];
|
||||
nextCursor: number;
|
||||
};
|
||||
|
||||
export type SessionFileScanStats<TEvent> = {
|
||||
filePath: string;
|
||||
events: TEvent[];
|
||||
parsedCount: number;
|
||||
newCount: number;
|
||||
skippedCount: number;
|
||||
cursor: number;
|
||||
nextCursor: number;
|
||||
};
|
||||
|
||||
type BaseSessionScannerOptions = {
|
||||
intervalMs: number;
|
||||
};
|
||||
|
||||
export abstract class BaseSessionScanner<TEvent> {
|
||||
private readonly sync: InvalidateSync;
|
||||
private readonly watchers = new Map<string, () => void>();
|
||||
private readonly processedEventKeys = new Set<string>();
|
||||
private readonly fileCursors = new Map<string, number>();
|
||||
private intervalId: ReturnType<typeof setInterval> | null = null;
|
||||
private stopped = false;
|
||||
private scanPromise: Promise<void> | null = null;
|
||||
|
||||
protected constructor(private readonly options: BaseSessionScannerOptions) {
|
||||
this.sync = new InvalidateSync(() => this.scan());
|
||||
}
|
||||
|
||||
protected abstract findSessionFiles(): Promise<string[]>;
|
||||
protected abstract parseSessionFile(filePath: string, cursor: number): Promise<SessionFileScanResult<TEvent>>;
|
||||
protected abstract generateEventKey(event: TEvent, context: { filePath: string; lineIndex?: number }): string;
|
||||
|
||||
protected async handleFileScan(_stats: SessionFileScanStats<TEvent>): Promise<void> {
|
||||
}
|
||||
|
||||
protected async initialize(): Promise<void> {
|
||||
}
|
||||
|
||||
protected async beforeScan(): Promise<void> {
|
||||
}
|
||||
|
||||
protected async afterScan(): Promise<void> {
|
||||
}
|
||||
|
||||
protected shouldScan(): boolean {
|
||||
return true;
|
||||
}
|
||||
|
||||
protected shouldWatchFile(_filePath: string): boolean {
|
||||
return true;
|
||||
}
|
||||
|
||||
protected ensureWatcher(filePath: string): void {
|
||||
if (this.watchers.has(filePath)) {
|
||||
return;
|
||||
}
|
||||
this.watchers.set(filePath, startFileWatcher(filePath, () => this.sync.invalidate()));
|
||||
}
|
||||
|
||||
protected invalidate(): void {
|
||||
this.sync.invalidate();
|
||||
}
|
||||
|
||||
protected getCursor(filePath: string): number {
|
||||
return this.fileCursors.get(filePath) ?? 0;
|
||||
}
|
||||
|
||||
protected setCursor(filePath: string, cursor: number): void {
|
||||
this.fileCursors.set(filePath, cursor);
|
||||
}
|
||||
|
||||
protected seedProcessedKeys(keys: Iterable<string>): void {
|
||||
for (const key of keys) {
|
||||
this.recordProcessedKey(key);
|
||||
}
|
||||
}
|
||||
|
||||
protected getWatchedFiles(): string[] {
|
||||
return [...this.watchers.keys()];
|
||||
}
|
||||
|
||||
protected pruneWatchers(keepFiles: Iterable<string>): void {
|
||||
const keep = new Set(keepFiles);
|
||||
for (const [filePath, stop] of this.watchers.entries()) {
|
||||
if (keep.has(filePath)) {
|
||||
continue;
|
||||
}
|
||||
stop();
|
||||
this.watchers.delete(filePath);
|
||||
}
|
||||
}
|
||||
|
||||
public async start(): Promise<void> {
|
||||
await this.initialize();
|
||||
await this.sync.invalidateAndAwait();
|
||||
this.intervalId = setInterval(() => this.sync.invalidate(), this.options.intervalMs);
|
||||
}
|
||||
|
||||
public async cleanup(): Promise<void> {
|
||||
this.stopped = true;
|
||||
if (this.intervalId) {
|
||||
clearInterval(this.intervalId);
|
||||
this.intervalId = null;
|
||||
}
|
||||
this.sync.stop();
|
||||
const pendingScan = this.scanPromise;
|
||||
for (const stop of this.watchers.values()) {
|
||||
stop();
|
||||
}
|
||||
this.watchers.clear();
|
||||
if (pendingScan) {
|
||||
await pendingScan.catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
private async scan(): Promise<void> {
|
||||
if (this.stopped || !this.shouldScan()) {
|
||||
return;
|
||||
}
|
||||
if (this.scanPromise) {
|
||||
return this.scanPromise;
|
||||
}
|
||||
this.scanPromise = this.runScan();
|
||||
try {
|
||||
await this.scanPromise;
|
||||
} finally {
|
||||
this.scanPromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
private async runScan(): Promise<void> {
|
||||
if (this.stopped || !this.shouldScan()) {
|
||||
return;
|
||||
}
|
||||
await this.beforeScan();
|
||||
const files = await this.findSessionFiles();
|
||||
for (const filePath of files) {
|
||||
if (this.stopped || !this.shouldScan()) {
|
||||
return;
|
||||
}
|
||||
if (this.shouldWatchFile(filePath)) {
|
||||
this.ensureWatcher(filePath);
|
||||
}
|
||||
const cursor = this.getCursor(filePath);
|
||||
const { events, nextCursor } = await this.parseSessionFile(filePath, cursor);
|
||||
const newEvents: TEvent[] = [];
|
||||
const newKeys: string[] = [];
|
||||
for (const entry of events) {
|
||||
const key = this.generateEventKey(entry.event, { filePath, lineIndex: entry.lineIndex });
|
||||
if (this.processedEventKeys.has(key)) {
|
||||
this.recordProcessedKey(key);
|
||||
continue;
|
||||
}
|
||||
newKeys.push(key);
|
||||
newEvents.push(entry.event);
|
||||
}
|
||||
await this.handleFileScan({
|
||||
filePath,
|
||||
events: newEvents,
|
||||
parsedCount: events.length,
|
||||
newCount: newEvents.length,
|
||||
skippedCount: events.length - newEvents.length,
|
||||
cursor,
|
||||
nextCursor
|
||||
});
|
||||
this.setCursor(filePath, nextCursor);
|
||||
for (const key of newKeys) {
|
||||
this.recordProcessedKey(key);
|
||||
}
|
||||
}
|
||||
await this.afterScan();
|
||||
}
|
||||
|
||||
private recordProcessedKey(key: string): void {
|
||||
this.processedEventKeys.add(key);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user