feat(log): 新增 DebugLogReader 增量读服务（JSONL 按行 + CSV fgetcsv 多行安全）

app/common/service/DebugLogReaderService.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\common\service;
+
+use base\common\service\DebugLogReaderServiceBase;
+
+/**
+ * debug 日志增量读取 service（App 入口类）.
+ *
+ * JSONL（{Ymd}.jsonl）与遗留 CSV（{ymd}.csv）导入链路（T6 定时任务）经本类
+ * 入口实例化（依赖倒置：调用方直接调本类，不调 Base）。
+ *
+ * 业务侧如需定制读取行为（如加自定义行过滤、改失败样本策略、换 position
+ * 落盘目标等），重写本类对应方法即可拦截，无需改动 extend/base/。
+ */
+class DebugLogReaderService extends DebugLogReaderServiceBase
+{
+}

extend/base/common/service/DebugLogReaderServiceBase.php

@@ -0,0 +1,510 @@
+<?php
+
+declare(strict_types=1);
+
+namespace base\common\service;
+
+use app\admin\model\DebugLogPosition;
+use app\common\service\HostService;
+use think\log\DebugLogToolkit;
+
+/**
+ * debug 日志增量读取 service（Base 层）——JSONL 按行 + CSV fgetcsv 多行安全.
+ *
+ * 与 XhprofLogReaderServiceBase 的两处差异（有意为之，勿"顺手同步"回去）：
+ *   1. JSONL read() 在 reader 内联 decode：每行经 DebugLogToolkit::decodeLine()
+ *      数组化后才 yield（坏行计 parseFails 不产出）；MAX_LINE_BYTES 收紧为 8MB
+ *      ——debug 行是短文本行，远小于 xhprof 采样行（~100KB），32MB 防御阈值不必要。
+ *   2. 新增 CSV 路径 readCsv()：遗留 CSV（DebugMysql.saveByFile 产物 {ymd}.csv）
+ *      的 content 含换行是常态，一条逻辑记录跨多物理行——必须用 fgetcsv() 有状态
+ *      解析（fgetcsv 原生处理引号内嵌入换行），【禁止】按 \n 切行后 str_getcsv
+ *      （会把跨行记录撕碎成两条坏行）。
+ *
+ * 其余协议与 xhprof 版一致：按 offset 增量读、inode/size 双条件轮转检测归零、
+ * max 上限、每次产出前推进安全偏移（消费方任意时刻 getCurrentPosition() 取到的
+ * 都是"最后一条已消费行末尾"）、JSONL 尾部半行不产出（下轮写入 \n 后补齐）、
+ * 不自动 savePosition（进度落盘由 T6 导入任务在 insertBatch 同一事务内显式调用）、
+ * position 按 (node_id, file_path) 唯一定位 upsert。
+ *
+ * 平台限制：Windows 下 fstat()['ino'] 恒为 0，inode 轮转检测退化为仅 size<offset
+ * 兜底（生产 Linux 有真实 inode，无此问题）——继承自 xhprof 版协议的已知特性，
+ * 不是缺陷。
+ *
+ * 依赖倒置：model 走 app\admin\model\DebugLogPosition 入口类；节点身份走
+ * app\common\service\HostService。本表为全新表，无旧数据接管问题，不含 lazyAdopt。
+ */
+class DebugLogReaderServiceBase
+{
+    /** 单次 fread 的 buffer 大小（64KB） */
+    protected const BUFFER_SIZE = 65536;
+
+    /** 单行硬上限（8MB）：超过整行丢弃计 dropped。真实 debug 行是短文本，8MB 已是异常防御 */
+    protected const MAX_LINE_BYTES = 8388608;
+
+    /** 解析失败样本最多保留条数（超出丢弃，样本只做排查线索不做全量记录） */
+    protected const MAX_FAIL_SAMPLES = 3;
+
+    /** 单条失败样本截断长度（字符） */
+    protected const FAIL_SAMPLE_LENGTH = 200;
+
+    /** @var int 当前文件 inode（read/readCsv 期间有效；Windows 恒为 0） */
+    protected $currentInode = 0;
+
+    /** @var int 安全偏移：最后一条已消费（产出/跳过/失败均算消费）逻辑记录的末尾 */
+    protected $currentOffset = 0;
+
+    /** @var string|null 最后一条已产出行的 md5 */
+    protected $currentLineHash = null;
+
+    /** @var bool 本轮读取是否干净到达 EOF（JSONL：尾部无半行；CSV：fgetcsv 返 false） */
+    protected $reachedEof = false;
+
+    /** @var int 本轮 read 丢弃的超长行数（仅 JSONL 路径；CSV 无缓冲切分逻辑恒为 0） */
+    protected $droppedLines = 0;
+
+    /** @var int 本轮读取解析失败计数（JSONL 坏行 + CSV 字段数不符） */
+    protected $parseFails = 0;
+
+    /** @var string[] 解析失败样本（截断后的原始行，最多 MAX_FAIL_SAMPLES 条） */
+    protected $failSamples = [];
+
+    /**
+     * 增量读取 jsonl 文件，yield decodeLine 数组化后的行.
+     *
+     * 读取协议：
+     * 1. fopen + fstat 取当前 inode/size；读 position（无则视为新建 offset=0）
+     * 2. 轮转检测：inode 不一致 OR filesize < offset → 归零重读
+     * 3. fseek 到 offset，循环 fread 64KB，定位已读段内最后 \n，仅产出完整行
+     * 4. 每次 yield 前 currentOffset 推进到该行末尾（含 \n），保证事务内
+     *    savePosition 与已消费行严格对齐（坏行同样推进——毒行不重读）
+     * 5. 完整行经 DebugLogToolkit::decodeLine() 数组化，decode 返 null 计
+     *    parseFails 不产出（失败样本进 failSamples）
+     * 6. 尾部半行（无 \n）不产出，currentOffset 停在半行起点，下次重读
+     * 7. 半行缓冲超过 MAX_LINE_BYTES：整行丢弃计 dropped，扫描至下一个 \n 继续
+     * 8. yield maxLines 后退出（不自动 savePosition，由上层事务内落盘）
+     *
+     * @param string $filePath jsonl 文件绝对路径
+     * @param int    $maxLines 单次产出的最大行数
+     *
+     * @return \Generator<array> 每次迭代返回一行（键值与 DebugLogToolkit::FIELDS 一致）
+     *
+     * @throws \RuntimeException 文件无法打开或 stat/fseek/fread 失败时抛出（交由上层 try-catch）
+     */
+    public function read(string $filePath, int $maxLines = 100000): \Generator
+    {
+        // per-run 状态复位（同一个 reader 实例可串行处理多个文件）
+        $this->reachedEof = false;
+        $this->droppedLines = 0;
+        $this->currentLineHash = null;
+        $this->parseFails = 0;
+        $this->failSamples = [];
+
+        $fp = @fopen($filePath, 'rb');
+        if ($fp === false) {
+            throw new \RuntimeException("DebugLogReaderService: 无法打开文件 {$filePath}");
+        }
+
+        try {
+            clearstatcache(true, $filePath);
+            $stat = fstat($fp);
+            if ($stat === false) {
+                throw new \RuntimeException("DebugLogReaderService: fstat 失败 {$filePath}");
+            }
+            $currentInode = (int) $stat['ino'];
+            $currentSize = (int) $stat['size'];
+            $this->currentInode = $currentInode;
+
+            // 读 position（无则默认 offset=0）
+            $offset = 0;
+            $position = $this->getLastPosition($filePath);
+            if ($position !== null) {
+                $offset = (int) $position['offset'];
+
+                // 轮转检测：inode 变化（同日文件被重建）OR size < offset（防御性归零）
+                if ((int) $position['inode'] !== $currentInode || $currentSize < $offset) {
+                    $offset = 0;
+                }
+            }
+            $this->currentOffset = $offset;
+
+            if ($offset > 0 && fseek($fp, $offset) !== 0) {
+                throw new \RuntimeException("DebugLogReaderService: fseek 失败 offset={$offset} file={$filePath}");
+            }
+
+            $yielded = 0;
+            $buffer = '';
+            // $bufferOffset 始终等于 buffer[0] 对应的文件字节偏移
+            $bufferOffset = $offset;
+
+            while (true) {
+                $chunk = fread($fp, self::BUFFER_SIZE);
+                if ($chunk === false) {
+                    throw new \RuntimeException("DebugLogReaderService: fread 失败 file={$filePath}");
+                }
+                if ($chunk === '') {
+                    // EOF：buffer 为空说明干净收尾；有半行则 currentOffset 已停在最后完整行末尾
+                    $this->reachedEof = ($buffer === '');
+                    return;
+                }
+                $buffer .= $chunk;
+
+                // 内层循环：处理 buffer 中所有完整行（含末尾 \n 的）
+                $pos = 0;
+                while (($nlPos = strpos($buffer, "\n", $pos)) !== false) {
+                    $line = substr($buffer, $pos, $nlPos - $pos);
+                    $pos = $nlPos + 1;
+
+                    // 先推进安全偏移再产出/判失败：坏行也算已消费（毒行不重读），
+                    // 消费方在事务内 savePosition 拿到的 offset 与已处理行严格对齐
+                    $this->currentOffset = $bufferOffset + $pos;
+
+                    $row = DebugLogToolkit::decodeLine($line);
+                    if ($row === null) {
+                        $this->parseFails++;
+                        $this->recordFailSample($line);
+
+                        continue;
+                    }
+
+                    $this->currentLineHash = md5($line);
+
+                    yield $row;
+                    $yielded++;
+
+                    if ($yielded >= $maxLines) {
+                        return;
+                    }
+                }
+
+                // buffer 中已消费 $pos 字节，半行（buffer[$pos..]）保留
+                $bufferOffset += $pos;
+                $buffer = substr($buffer, $pos);
+
+                // 超长行防御：整行丢弃（禁止强制切行——会把一条 JSON 切成两条坏行）
+                if (strlen($buffer) >= self::MAX_LINE_BYTES) {
+                    $this->droppedLines++;
+                    $buffer = '';
+                    // 扫描至下一个换行符，将其后内容作为新 buffer 继续
+                    while (true) {
+                        $chunk = fread($fp, self::BUFFER_SIZE);
+                        if ($chunk === false || $chunk === '') {
+                            // 至 EOF 仍未等到换行符：整个尾部按已丢弃处理，偏移推到文件末尾
+                            $bufferOffset = (int) ftell($fp);
+                            $this->currentOffset = $bufferOffset;
+                            $this->reachedEof = true;
+                            return;
+                        }
+                        $nl = strpos($chunk, "\n");
+                        if ($nl !== false) {
+                            $bufferOffset = (int) ftell($fp) - (strlen($chunk) - $nl - 1);
+                            $this->currentOffset = $bufferOffset;
+                            $buffer = substr($chunk, $nl + 1);
+                            break;
+                        }
+                    }
+                }
+            }
+        } finally {
+            if (is_resource($fp)) {
+                fclose($fp);
+            }
+        }
+    }
+
+    /**
+     * 增量读取遗留 CSV 文件，yield 关联数组化的数据行.
+     *
+     * 读取协议（与 read() 的差异全因 CSV 的多物理行特性）：
+     * 1. fopen 'rb' + fstat 取 inode/size + position 轮转检测归零——与 read() 一致
+     * 2. fseek 到 offset 后循环 fgetcsv()：fgetcsv 是有状态解析，原生处理引号内
+     *    嵌入换行（一条逻辑记录跨多物理行），返回值即一条完整逻辑记录
+     *    【禁止】按 \n 切行后 str_getcsv——引号内换行会被撕碎
+     * 3. 每读出一条逻辑记录（无论产出/表头跳过/失败）都把 currentOffset 推进到
+     *    ftell($fp)（记录末尾），保证事务内 savePosition 与已消费记录对齐
+     * 4. position==0（从文件头开始）时首行经 DebugLogToolkit::isCsvHeader() 判定，
+     *    是表头则跳过不产出（offset 仍推进）；续读 offset 处不会是表头，不判定
+     * 5. 行字段数 !== 8 计 parseFails 不产出（失败样本进 failSamples）
+     * 6. 行字段按 DebugLogToolkit::FIELDS 顺序 map 成关联数组；create_time 为
+     *    数字字符串时转 int，create_time_title 保留字符串（CSV 一切皆字符串）
+     * 7. 产出 maxRows 条后退出（不自动 savePosition，由上层事务内落盘）
+     *
+     * 注：CSV 路径无"尾部半行"概念——遗留 CSV 是已封存的静态文件（非追加中），
+     * fgetcsv 读到 EOF 返 false 即干净收尾（reachedEof=true）。
+     *
+     * @param string $filePath csv 文件绝对路径
+     * @param int    $maxRows  单次产出的最大数据行数（表头/失败行不计）
+     *
+     * @return \Generator<array> 每次迭代返回一行（键值与 DebugLogToolkit::FIELDS 一致）
+     *
+     * @throws \RuntimeException 文件无法打开或 stat/fseek 失败时抛出（交由上层 try-catch）
+     */
+    public function readCsv(string $filePath, int $maxRows = 100000): \Generator
+    {
+        // per-run 状态复位（与 read() 同一套状态位，CSV 路径 droppedLines 恒 0）
+        $this->reachedEof = false;
+        $this->droppedLines = 0;
+        $this->currentLineHash = null;
+        $this->parseFails = 0;
+        $this->failSamples = [];
+
+        $fp = @fopen($filePath, 'rb');
+        if ($fp === false) {
+            throw new \RuntimeException("DebugLogReaderService: 无法打开文件 {$filePath}");
+        }
+
+        try {
+            clearstatcache(true, $filePath);
+            $stat = fstat($fp);
+            if ($stat === false) {
+                throw new \RuntimeException("DebugLogReaderService: fstat 失败 {$filePath}");
+            }
+            $currentInode = (int) $stat['ino'];
+            $currentSize = (int) $stat['size'];
+            $this->currentInode = $currentInode;
+
+            // 读 position（无则默认 offset=0）
+            $offset = 0;
+            $position = $this->getLastPosition($filePath);
+            if ($position !== null) {
+                $offset = (int) $position['offset'];
+
+                // 轮转检测：inode 变化 OR size < offset（防御性归零）
+                if ((int) $position['inode'] !== $currentInode || $currentSize < $offset) {
+                    $offset = 0;
+                }
+            }
+            $this->currentOffset = $offset;
+
+            if ($offset > 0 && fseek($fp, $offset) !== 0) {
+                throw new \RuntimeException("DebugLogReaderService: fseek 失败 offset={$offset} file={$filePath}");
+            }
+
+            // 仅从文件头（offset==0）开始读时才做表头判定：续读起点不会是表头
+            $checkHeader = ($offset === 0);
+            $yielded = 0;
+
+            while (true) {
+                // 显式传全部参数：separator/enclosure/escape 与遗留 fputcsv 写出参数一致
+                // （PHP 8.4+ 省略 escape 会触发 deprecation，显式传可前向兼容）
+                $fields = fgetcsv($fp, null, ',', '"', '\\');
+                if ($fields === false) {
+                    // fgetcsv 返 false 即 EOF（干净收尾，无半行概念）
+                    $this->reachedEof = true;
+
+                    return;
+                }
+
+                // 每读出一条逻辑记录就把安全偏移推进到记录末尾（含引号内换行占的
+                // 多物理行）——表头/失败行同样推进（毒行不重读），保证事务内
+                // savePosition 与已消费记录严格对齐
+                $this->currentOffset = (int) ftell($fp);
+
+                if ($checkHeader) {
+                    $checkHeader = false;
+                    if (DebugLogToolkit::isCsvHeader($this->normalizeCsvFields($fields))) {
+                        // 表头行：跳过不产出（offset 已推进）
+                        continue;
+                    }
+                }
+
+                if (count($fields) !== 8) {
+                    $this->parseFails++;
+                    $this->recordFailSample('csv-fields=' . count($fields) . ': ' . implode(',', $this->stringifyCsvFields($fields)));
+
+                    continue;
+                }
+
+                $row = [];
+                foreach (DebugLogToolkit::FIELDS as $i => $field) {
+                    $value = $fields[$i] ?? '';
+                    // fgetcsv 对空字段可能返回 null，统一落为 ''（与 JSONL 行的空串语义对齐）
+                    $row[$field] = $value === null ? '' : $value;
+                }
+                // create_time 数字字符串转 int（时间戳恒为非负整数）；title 保留字符串
+                if (ctype_digit((string) $row['create_time'])) {
+                    $row['create_time'] = (int) $row['create_time'];
+                }
+
+                $this->currentLineHash = md5(serialize($fields));
+
+                yield $row;
+                $yielded++;
+
+                if ($yielded >= $maxRows) {
+                    return;
+                }
+            }
+        } finally {
+            if (is_resource($fp)) {
+                fclose($fp);
+            }
+        }
+    }
+
+    /**
+     * 当前安全读取位置（read/readCsv 期间/结束后可调）.
+     *
+     * @return array{inode: int, offset: int, last_line_hash: string|null}
+     */
+    public function getCurrentPosition(): array
+    {
+        return [
+            'inode' => $this->currentInode,
+            'offset' => $this->currentOffset,
+            'last_line_hash' => $this->currentLineHash,
+        ];
+    }
+
+    /**
+     * 本轮读取是否干净到达 EOF（JSONL：尾部无半行；CSV：fgetcsv 返 false）.
+     *
+     * 导入侧据此判断"文件已导完"：reachedEof 且 offset >= filesize 才允许 unlink。
+     */
+    public function reachedEof(): bool
+    {
+        return $this->reachedEof;
+    }
+
+    /**
+     * 本轮 read 丢弃的超长行数（由导入侧并入 fails 统计；CSV 路径恒 0）.
+     */
+    public function getDroppedLines(): int
+    {
+        return $this->droppedLines;
+    }
+
+    /**
+     * 本轮读取解析失败计数（JSONL 坏行 + CSV 字段数不符，由导入侧传给 savePosition）.
+     */
+    public function getParseFails(): int
+    {
+        return $this->parseFails;
+    }
+
+    /**
+     * 解析失败样本（json 数组字符串或 null，由导入侧传给 savePosition）.
+     *
+     * 每条样本截断至 FAIL_SAMPLE_LENGTH 字符，最多 MAX_FAIL_SAMPLES 条。
+     */
+    public function getFailSamples(): ?string
+    {
+        if ($this->failSamples === []) {
+            return null;
+        }
+
+        $json = json_encode($this->failSamples, JSON_UNESCAPED_UNICODE);
+
+        return $json === false ? null : $json;
+    }
+
+    /**
+     * 查询文件的上次读取位置（按 node_id 隔离）.
+     *
+     * @return array|null 命中时返回 [inode, offset, last_line_hash, parse_fail_count, parse_fail_samples]，无记录返回 null
+     */
+    public function getLastPosition(string $filePath): ?array
+    {
+        $nodeId = HostService::getNodeId();
+
+        $row = DebugLogPosition::where('file_path', $filePath)
+            ->where('node_id', $nodeId)
+            ->find();
+        if ($row === null) {
+            return null;
+        }
+
+        return [
+            'inode' => (int) $row->getData('inode'),
+            'offset' => (int) $row->getData('offset'),
+            'last_line_hash' => $row->getData('last_line_hash'),
+            'parse_fail_count' => (int) $row->getData('parse_fail_count'),
+            'parse_fail_samples' => $row->getData('parse_fail_samples'),
+        ];
+    }
+
+    /**
+     * 保存读取位置（upsert，按 node_id 隔离）.
+     *
+     * 由导入任务在"insertBatch 同一事务"内显式调用（本类不自动保存），
+     * 保证崩溃时入库与进度要么同时生效、要么同时回滚，消重复导入窗口。
+     *
+     * @param string      $filePath     日志文件绝对路径
+     * @param int         $inode        当前文件 inode
+     * @param int         $offset       新的字节偏移（最后一条已消费记录末尾）
+     * @param string|null $lastLineHash 最后一条已产出记录 md5
+     * @param int         $failCount    本批解析失败累计次数
+     * @param string|null $failSamples  解析失败样本（json 或 null）
+     */
+    public function savePosition(string $filePath, int $inode, int $offset, ?string $lastLineHash, int $failCount = 0, ?string $failSamples = null): void
+    {
+        $now = time();
+        $nodeId = HostService::getNodeId();
+
+        $existing = DebugLogPosition::where('file_path', $filePath)
+            ->where('node_id', $nodeId)
+            ->find();
+        if ($existing !== null) {
+            $existing->save([
+                'inode' => $inode,
+                'offset' => $offset,
+                'last_read_time' => $now,
+                'last_line_hash' => $lastLineHash,
+                'parse_fail_count' => $failCount,
+                'parse_fail_samples' => $failSamples,
+            ]);
+
+            return;
+        }
+
+        DebugLogPosition::create([
+            'node_id' => $nodeId,
+            'file_path' => $filePath,
+            'inode' => $inode,
+            'offset' => $offset,
+            'last_read_time' => $now,
+            'last_line_hash' => $lastLineHash,
+            'parse_fail_count' => $failCount,
+            'parse_fail_samples' => $failSamples,
+            'create_time' => $now,
+            'update_time' => $now,
+        ]);
+    }
+
+    /**
+     * 记录一条解析失败样本（截断至 FAIL_SAMPLE_LENGTH，超量丢弃）.
+     */
+    protected function recordFailSample(string $raw): void
+    {
+        if (count($this->failSamples) >= self::MAX_FAIL_SAMPLES) {
+            return;
+        }
+
+        $this->failSamples[] = mb_substr($raw, 0, self::FAIL_SAMPLE_LENGTH);
+    }
+
+    /**
+     * CSV 字段数组归一化后交 isCsvHeader 判定（null 字段落为 ''）.
+     *
+     * @param array $fields fgetcsv 原始返回值
+     */
+    protected function normalizeCsvFields(array $fields): array
+    {
+        return array_map(static function ($value) {
+            return $value === null ? '' : $value;
+        }, $fields);
+    }
+
+    /**
+     * CSV 字段数组字符串化（失败样本拼接用，null 落为空串占位）.
+     *
+     * @param array $fields fgetcsv 原始返回值
+     *
+     * @return string[]
+     */
+    protected function stringifyCsvFields(array $fields): array
+    {
+        return array_map(static function ($value) {
+            return $value === null ? '' : (string) $value;
+        }, $fields);
+    }
+}

tests/Unit/DebugLogReaderServiceTest.php

@@ -0,0 +1,329 @@
+<?php
+
+declare(strict_types=1);
+
+namespace tests\Unit;
+
+use app\common\service\DebugLogReaderService;
+use PHPUnit\Framework\TestCase;
+use think\log\DebugLogToolkit;
+
+/**
+ * DebugLogReaderService 增量读协议单元测试（JSONL 按行 + CSV fgetcsv 多行安全，不触 DB）.
+ *
+ * getLastPosition/savePosition 的 DB 持久化通过文件底部的 StubDebugLogReaderService
+ * 替换为内存实现（返回 canned position / 记录调用），只测 read/readCsv 的文件协议部分。
+ *
+ * 平台限制（Windows 宿主机 PHP）：fstat()['ino'] 恒为 0，轮转用例只能验证
+ * size<offset 归零路径，无法验证 inode 变化路径——生产 Linux 有真实 inode，
+ * 该路径与 XhprofLogReader 逐行同款（协议镜像），由代码评审保证。
+ *
+ * 用例 2（CSV 引号内换行跨 3 物理行）是本计划的 BLOCKER 回归锚点：遗留 CSV 的
+ * content 含换行是常态，若实现退化为"按 \n 切行 + str_getcsv"该用例必红。
+ */
+class DebugLogReaderServiceTest extends TestCase
+{
+    /** 夹具根目录（系统临时目录下，不碰项目 runtime） */
+    private const TEMP_BASE = 'debug-log-reader-test';
+
+    protected function setUp(): void
+    {
+        // 写前清理（上一轮残留）
+        $base = self::baseDir();
+        if (is_dir($base)) {
+            self::rrmdir($base);
+        }
+        mkdir($base, 0777, true);
+    }
+
+    protected function tearDown(): void
+    {
+        // 测后清理
+        $base = self::baseDir();
+        if (is_dir($base)) {
+            self::rrmdir($base);
+        }
+    }
+
+    private static function baseDir(): string
+    {
+        return sys_get_temp_dir() . '/' . self::TEMP_BASE;
+    }
+
+    /**
+     * 递归删除目录.
+     */
+    private static function rrmdir(string $dir): void
+    {
+        $items = scandir($dir);
+        foreach ($items as $item) {
+            if ($item === '.' || $item === '..') {
+                continue;
+            }
+            $path = $dir . DIRECTORY_SEPARATOR . $item;
+            is_dir($path) ? self::rrmdir($path) : unlink($path);
+        }
+        rmdir($dir);
+    }
+
+    /**
+     * 构造与 FIELDS 契约一致的 8 字段行（create_time_title 由时间戳确定性推导）.
+     */
+    private static function makeRow(string $level, string $content, int $createTime): array
+    {
+        return [
+            'level' => $level,
+            'content' => $content,
+            'create_time' => $createTime,
+            'create_time_title' => date('Y-m-d H:i:s', $createTime),
+            'uid' => 'test-uid',
+            'app_name' => 'cli',
+            'controller_name' => '',
+            'action_name' => '',
+        ];
+    }
+
+    /**
+     * 用例 1：JSONL 3 条完整行 + 尾部半行 → 产出 3 条、reachedEof=false、
+     * offset 停在半行起点（半行不产出不消费，下轮补 \n 后重读）.
+     */
+    public function test_read_jsonl_yields_complete_lines_and_stops_at_half_line(): void
+    {
+        $file = self::baseDir() . '/20260816.jsonl';
+
+        $complete = '';
+        $expected = [];
+        for ($i = 1; $i <= 3; ++$i) {
+            $row = self::makeRow('info', "完整行 {$i}", 1700000000 + $i);
+            $expected[] = $row;
+            $complete .= json_encode($row, JSON_UNESCAPED_UNICODE) . "\n";
+        }
+        $halfLine = '{"level":"info","content":"尾部半行';
+        file_put_contents($file, $complete . $halfLine);
+
+        $reader = new StubDebugLogReaderService();
+        $yielded = iterator_to_array($reader->read($file, 100), false);
+
+        self::assertCount(3, $yielded, '3 条完整行应全部产出');
+        foreach ($yielded as $i => $row) {
+            self::assertSame($expected[$i], $row, "第 {$i} 行 decode 后应与夹具一致");
+            self::assertSame(DebugLogToolkit::FIELDS, array_keys($row));
+        }
+
+        self::assertFalse($reader->reachedEof(), '存在尾部半行时不得宣告 EOF');
+        self::assertSame(
+            strlen($complete),
+            $reader->getCurrentPosition()['offset'],
+            'offset 必须停在半行起点（= 完整部分字节数）'
+        );
+        self::assertSame(0, $reader->getDroppedLines());
+        self::assertSame(0, $reader->getParseFails());
+    }
+
+    /**
+     * 用例 2（BLOCKER 回归锚点）：CSV 表头 + 1 条 content 带换行的记录（引号包裹
+     * 跨 3 物理行）+ 1 条普通记录 → 产出 2 条完整记录（表头跳过），字段与夹具
+     * 一致、offset=文件末尾.
+     *
+     * 遗留 CSV 的 content 含换行是常态——fgetcsv 有状态解析是唯一正确实现，
+     * 任何"按 \n 切行"的退化实现都会在此用例红（跨行记录被撕碎/字段数错位）。
+     */
+    public function test_read_csv_multiline_content_as_single_record(): void
+    {
+        $file = self::baseDir() . '/20260815.csv';
+
+        $header = 'level,content,create_time,create_time_title,uid,app_name,controller_name,action_name';
+        // 引号内 2 个换行 + CSV 规范的双写引号（"" 解码回 "）→ 逻辑记录跨 3 物理行
+        $csvEncodedContent = "第一行\nquoted \"\"word\"\" 内嵌双引号\nthird physical line";
+        $expectedContent = "第一行\nquoted \"word\" 内嵌双引号\nthird physical line";
+        $row1 = 'info,"' . $csvEncodedContent . '",1755302401,2026-08-16 00:00:01,uid-1,admin,Index,index';
+        $row2 = 'error,plain single line content,1755302402,2026-08-16 00:00:02,uid-2,tools,Timer,run';
+        $raw = $header . "\n" . $row1 . "\n" . $row2 . "\n";
+        file_put_contents($file, $raw);
+
+        // 夹具自检：证明这是真正的多物理行夹具（朴素按 \n 切会得到 5 条"行"而非 2 条记录）
+        self::assertSame(5, substr_count($raw, "\n"), '夹具应有 5 个物理行（表头 1 + 记录 3 + 记录 1）');
+
+        $reader = new StubDebugLogReaderService();
+        $rows = iterator_to_array($reader->readCsv($file, 100), false);
+
+        self::assertCount(2, $rows, '表头跳过，产出 2 条数据记录');
+
+        $first = $rows[0];
+        self::assertSame(DebugLogToolkit::FIELDS, array_keys($first));
+        self::assertSame('info', $first['level']);
+        self::assertSame($expectedContent, $first['content'], '引号内换行与双写引号必须完整还原（BLOCKER）');
+        self::assertSame(1755302401, $first['create_time'], 'create_time 数字字符串转 int');
+        self::assertSame('2026-08-16 00:00:01', $first['create_time_title'], 'create_time_title 保留字符串');
+        self::assertSame('uid-1', $first['uid']);
+        self::assertSame('admin', $first['app_name']);
+        self::assertSame('Index', $first['controller_name']);
+        self::assertSame('index', $first['action_name']);
+
+        self::assertSame('error', $rows[1]['level']);
+        self::assertSame('plain single line content', $rows[1]['content']);
+        self::assertSame(1755302402, $rows[1]['create_time']);
+
+        self::assertTrue($reader->reachedEof());
+        self::assertSame(
+            (int) filesize($file),
+            $reader->getCurrentPosition()['offset'],
+            'offset 必须推进到文件末尾（含跨物理行记录的全部字节）'
+        );
+        self::assertSame(0, $reader->getParseFails());
+    }
+
+    /**
+     * 用例 3：轮转——重写更短文件后 offset > size → 归零重读.
+     *
+     * Windows 平台限制：fstat()['ino'] 恒为 0，无法构造"inode 变化"路径，
+     * 只能验证 size<offset 兜底归零（stub 返回原 inode=0 使 inode 检查通过，
+     * 归零由 size 条件触发）；生产 Linux 两条路径并存，语义与 xhprof 版镜像。
+     */
+    public function test_rotation_offset_beyond_size_resets_to_zero(): void
+    {
+        $file = self::baseDir() . '/rot.jsonl';
+
+        $row1 = self::makeRow('info', '旧行 1', 1700000001);
+        $row2 = self::makeRow('info', '旧行 2', 1700000002);
+        file_put_contents(
+            $file,
+            json_encode($row1, JSON_UNESCAPED_UNICODE) . "\n" . json_encode($row2, JSON_UNESCAPED_UNICODE) . "\n"
+        );
+
+        // 第一轮：全量读，拿到 offset（Windows 下 inode 恒 0）
+        $reader = new StubDebugLogReaderService();
+        $first = iterator_to_array($reader->read($file, 100), false);
+        self::assertCount(2, $first);
+        self::assertTrue($reader->reachedEof());
+
+        $position = $reader->getCurrentPosition();
+        $oldOffset = $position['offset'];
+
+        // 续读：从旧 offset 起，无新数据 → 0 条、EOF
+        $resume = new StubDebugLogReaderService();
+        $resume->stubPosition = [
+            'inode' => $position['inode'],
+            'offset' => $oldOffset,
+            'last_line_hash' => $position['last_line_hash'],
+            'parse_fail_count' => 0,
+            'parse_fail_samples' => null,
+        ];
+        $again = iterator_to_array($resume->read($file, 100), false);
+        self::assertCount(0, $again, '续读起点在文件末尾，应无新行');
+        self::assertTrue($resume->reachedEof());
+
+        // 轮转：重写更短的文件，旧 offset > 新 size → 归零重读全量
+        $newRow = self::makeRow('info', '新文件唯一行', 1700000009);
+        file_put_contents($file, json_encode($newRow, JSON_UNESCAPED_UNICODE) . "\n");
+        clearstatcache();
+        self::assertLessThan(
+            $oldOffset,
+            (int) filesize($file),
+            '前置：新文件 size 必须小于旧 offset（size<offset 归零条件成立）'
+        );
+
+        $rotated = new StubDebugLogReaderService();
+        $rotated->stubPosition = [
+            'inode' => $position['inode'], // Windows 恒 0：inode 检查恒过，归零只能靠 size 条件
+            'offset' => $oldOffset,
+            'last_line_hash' => 'stale',
+            'parse_fail_count' => 0,
+            'parse_fail_samples' => null,
+        ];
+        $reread = iterator_to_array($rotated->read($file, 100), false);
+
+        self::assertCount(1, $reread, '归零后应重读新文件全量');
+        self::assertSame('新文件唯一行', $reread[0]['content']);
+        self::assertTrue($rotated->reachedEof());
+    }
+
+    /**
+     * 用例 4：failure——坏 JSON 行 decodeLine 返 null → 计 parseFails 不产出，
+     * 坏行仍消费（offset 推进过毒行，不无限重读），样本可取.
+     */
+    public function test_read_jsonl_bad_line_counted_as_fail_not_yielded(): void
+    {
+        $file = self::baseDir() . '/bad.jsonl';
+
+        $good1 = self::makeRow('info', '好行 1', 1700000001);
+        $good2 = self::makeRow('info', '好行 2', 1700000002);
+        $badLine = '{"level":"info","content":"坏 JSON 没闭合';
+        file_put_contents(
+            $file,
+            json_encode($good1, JSON_UNESCAPED_UNICODE) . "\n" . $badLine . "\n" . json_encode($good2, JSON_UNESCAPED_UNICODE) . "\n"
+        );
+
+        $reader = new StubDebugLogReaderService();
+        $rows = iterator_to_array($reader->read($file, 100), false);
+
+        self::assertCount(2, $rows, '坏行不产出，好行全产出');
+        self::assertSame('好行 1', $rows[0]['content']);
+        self::assertSame('好行 2', $rows[1]['content']);
+
+        self::assertSame(1, $reader->getParseFails(), '坏 JSON 行计 1 次 parseFails');
+        self::assertNotSame('', (string) $reader->getFailSamples(), '失败样本应非空');
+        self::assertStringContainsString('坏 JSON', (string) $reader->getFailSamples());
+
+        // 坏行已消费：offset 推进到文件末尾，毒行不会被下一轮重读
+        self::assertSame((int) filesize($file), $reader->getCurrentPosition()['offset']);
+        self::assertTrue($reader->reachedEof());
+    }
+
+    /**
+     * 用例 5：CSV 字段数 !== 8 → 计 parseFails 不产出（readCsv 失败路径）.
+     */
+    public function test_read_csv_field_count_mismatch_counted_as_fail(): void
+    {
+        $file = self::baseDir() . '/bad.csv';
+
+        $header = 'level,content,create_time,create_time_title,uid,app_name,controller_name,action_name';
+        $good = 'info,ok content,1755302401,2026-08-16 00:00:01,uid-1,admin,Index,index';
+        $bad = 'info,only,three,fields'; // 4 列 ≠ 8
+        file_put_contents($file, $header . "\n" . $good . "\n" . $bad . "\n");
+
+        $reader = new StubDebugLogReaderService();
+        $rows = iterator_to_array($reader->readCsv($file, 100), false);
+
+        self::assertCount(1, $rows, '字段数不符的行不产出');
+        self::assertSame('ok content', $rows[0]['content']);
+        self::assertSame(1, $reader->getParseFails(), 'CSV 坏行计 1 次 parseFails');
+        self::assertSame((int) filesize($file), $reader->getCurrentPosition()['offset'], '坏行已消费推进到文件末尾');
+        self::assertTrue($reader->reachedEof());
+    }
+}
+
+/**
+ * 测试桩：把 position 持久化（DB 依赖）替换为内存实现，只测文件协议部分.
+ *
+ * - getLastPosition 返回 canned stubPosition（null 模拟无历史记录/首次读）
+ * - savePosition 仅记录调用参数，不触 DB（本任务 reader 不自动调 savePosition，
+ *   事务内显式调用是 T6 的职责）
+ */
+class StubDebugLogReaderService extends DebugLogReaderService
+{
+    /** @var array|null getLastPosition 的 canned 返回值（null=无历史 position） */
+    public ?array $stubPosition = null;
+
+    /** @var array[] savePosition 调用记录 */
+    public array $savedPositions = [];
+
+    public function getLastPosition(string $filePath): ?array
+    {
+        unset($filePath);
+
+        return $this->stubPosition;
+    }
+
+    public function savePosition(string $filePath, int $inode, int $offset, ?string $lastLineHash, int $failCount = 0, ?string $failSamples = null): void
+    {
+        $this->savedPositions[] = [
+            'file_path' => $filePath,
+            'inode' => $inode,
+            'offset' => $offset,
+            'last_line_hash' => $lastLineHash,
+            'fail_count' => $failCount,
+            'fail_samples' => $failSamples,
+        ];
+    }
+}