# All files / src/agent file-tailer.ts

3.26% Statements 3/92
0% Branches 0/53
0% Functions 0/13
3.4% Lines 3/88

Press n or j to go to the next uncovered block, b, p or k for the previous block.

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219
    220
    221
    222
    223
    224
    225
    226
    227
    228
    229
    230
    231
    232
    233
    234
    235
    236
    237
    238
    239
    240
    241
    242
    243
    244
    245
    246
    247
    248
    249
    250
    251
    252
    253
    254
    255
    256
    257
    258

    14x

    14x

    14x

     /**
     * File Tailer
     *
     * Watches an output file and emits parsed events in real-time.
     * Used for crash-resilient agent spawning where subprocesses write
     * directly to files instead of using pipes.
     *
     * Uses fs.watch() for efficient change detection with a poll fallback
     * since fs.watch isn't 100% reliable on all platforms.
     */

    import { watch, type FSWatcher } from 'node:fs';
    import { open, stat } from 'node:fs/promises';
    import type { FileHandle } from 'node:fs/promises';
    import type { StreamParser, StreamEvent } from './providers/stream-types.js';
    import { createModuleLogger } from '../logger/index.js';

    const log = createModuleLogger('file-tailer');

    /** Poll interval for fallback polling (ms) */
    const POLL_INTERVAL_MS = 500;

    /** Read buffer size (bytes) */
    const READ_BUFFER_SIZE = 64 * 1024;

    export interface FileTailerOptions {
      /** Path to the output file to watch */
      filePath: string;
      /** Agent ID for logging */
      agentId: string;
      /** Parser to convert lines to stream events */
      parser: StreamParser;
      /** Optional callback for each stream event */
      onEvent?: (event: StreamEvent) => void;
      /** If true, read from beginning of file; otherwise tail only new content (default: false) */
      startFromBeginning?: boolean;
      /** Callback for raw file content chunks (for DB persistence + event emission) */
      onRawContent?: (content: string) => void;
    }

    /**
     * FileTailer watches a file for changes and emits parsed stream events.
     *
     * Behavior:
     * - Uses fs.watch() for efficient change detection
     * - Falls back to polling every 500ms (fs.watch misses events sometimes)
     * - Reads new content incrementally, splits into lines
     * - Feeds each line to the parser, emits resulting events
     * - Handles partial lines at buffer boundaries
     */
    export class FileTailer {
      private position = 0;
      private watcher: FSWatcher | null = null;
      private pollInterval: NodeJS.Timeout | null = null;
      private fileHandle: FileHandle | null = null;
      private stopped = false;
      private partialLine = '';
      private reading = false;

      private readonly filePath: string;
      private readonly agentId: string;
      private readonly parser: StreamParser;
      private readonly onEvent?: (event: StreamEvent) => void;
      private readonly startFromBeginning: boolean;
      private readonly onRawContent?: (content: string) => void;

      constructor(options: FileTailerOptions) {
        this.filePath = options.filePath;
        this.agentId = options.agentId;
        this.parser = options.parser;
        this.onEvent = options.onEvent;
        this.startFromBeginning = options.startFromBeginning ?? false;
        this.onRawContent = options.onRawContent;
      }

      /**
       * Start watching the file for changes.
       * Initializes position, starts fs.watch, and begins poll fallback.
       */
      async start(): Promise<void> {
        if (this.stopped) return;

        log.debug({ filePath: this.filePath, agentId: this.agentId }, 'starting file tailer');

        try {
          // Open file for reading
          this.fileHandle = await open(this.filePath, 'r');

          // Set initial position
          if (this.startFromBeginning) {
            this.position = 0;
          } else {
            // Seek to end
            const stats = await stat(this.filePath);
            this.position = stats.size;
          }

          // Start fs.watch for efficient change detection
          this.watcher = watch(this.filePath, (eventType) => {
            if (eventType === 'change' && !this.stopped) {
              this.readNewContent().catch((err) => {
                log.warn({ err: err instanceof Error ? err.message : String(err), agentId: this.agentId }, 'error reading new content');
              });
            }
          });

          this.watcher.on('error', (err) => {
            log.warn({ err: err instanceof Error ? err.message : String(err), agentId: this.agentId }, 'watcher error');
          });

          // Start poll fallback (fs.watch misses events sometimes)
          this.pollInterval = setInterval(() => {
            if (!this.stopped) {
              this.readNewContent().catch((err) => {
                log.warn({ err: err instanceof Error ? err.message : String(err), agentId: this.agentId }, 'poll read error');
              });
            }
          }, POLL_INTERVAL_MS);

          // If starting from beginning, do initial read
          if (this.startFromBeginning) {
            await this.readNewContent();
          }
        } catch (err) {
          log.error({ err: err instanceof Error ? err.message : String(err), filePath: this.filePath }, 'failed to start file tailer');
        }
      }

      /**
       * Read new content from the file since last position.
       * Splits into lines, feeds to parser, emits events.
       */
      private async readNewContent(): Promise<void> {
        if (this.stopped || !this.fileHandle || this.reading) return;

        this.reading = true;
        try {
          // Check current file size
          const stats = await stat(this.filePath);
          if (stats.size <= this.position) {
            return; // No new content
          }

          // Read new bytes
          const bytesToRead = stats.size - this.position;
          const buffer = Buffer.alloc(Math.min(bytesToRead, READ_BUFFER_SIZE));
          const { bytesRead } = await this.fileHandle.read(buffer, 0, buffer.length, this.position);

          if (bytesRead === 0) return;

          this.position += bytesRead;

          // Fire raw content callback for DB persistence (before line splitting)
          const rawChunk = buffer.toString('utf-8', 0, bytesRead);
          if (this.onRawContent) {
            this.onRawContent(rawChunk);
          }

          // Convert to string and process lines
          const content = this.partialLine + rawChunk;
          const lines = content.split('\n');

          // Last element is either empty (if content ended with \n) or a partial line
          this.partialLine = lines.pop() ?? '';

          // Process complete lines
          for (const line of lines) {
            if (line.trim()) {
              this.processLine(line);
            }
          }

          // If there's more content to read, schedule another read
          if (stats.size > this.position) {
            setImmediate(() => {
              this.readNewContent().catch(() => {});
            });
          }
        } finally {
          this.reading = false;
        }
      }

      /**
       * Process a single line through the parser and emit events.
       */
      private processLine(line: string): void {
        const events = this.parser.parseLine(line);

        for (const event of events) {
          if (this.onEvent) {
            this.onEvent(event);
          }
        }
      }

      /**
       * Stop watching the file.
       * Cleans up watcher, poll timer, and file handle.
       */
      async stop(): Promise<void> {
        if (this.stopped) return;

        this.stopped = true;
        log.debug({ filePath: this.filePath, agentId: this.agentId }, 'stopping file tailer');

        // Close watcher
        if (this.watcher) {
          this.watcher.close();
          this.watcher = null;
        }

        // Clear poll timer
        if (this.pollInterval) {
          clearInterval(this.pollInterval);
          this.pollInterval = null;
        }

        // Do one final read to catch any remaining content
        try {
          await this.readNewContent();

          // Process any remaining partial line
          if (this.partialLine.trim()) {
            this.processLine(this.partialLine);
            this.partialLine = '';
          }

          // Signal end of stream to parser
          const endEvents = this.parser.end();
          for (const event of endEvents) {
            if (this.onEvent) {
              this.onEvent(event);
            }
          }
        } catch {
          // Ignore errors during cleanup
        }

        // Close file handle
        if (this.fileHandle) {
          try {
            await this.fileHandle.close();
          } catch {
            // Ignore close errors
          }
          this.fileHandle = null;
        }
      }

      /**
       * Check if the tailer has been stopped.
       */
      get isStopped(): boolean {
        return this.stopped;
      }
    }