From efc7180317b6a4bcd1b7901959d2a82333fdaf5d Mon Sep 17 00:00:00 2001 From: yoshikazu Date: Tue, 1 Sep 2026 09:46:34 +0900 Subject: [PATCH] =?UTF-8?q?=E8=87=AA=E5=8B=95=E8=BF=BD=E5=BE=93=E3=81=A7?= =?UTF-8?q?=E8=BF=BD=E8=A8=98=E5=88=86=E3=81=A0=E3=81=91=E3=82=92=E8=A7=A3?= =?UTF-8?q?=E6=9E=90=E3=81=99=E3=82=8B=E3=82=88=E3=81=86=E3=81=AB=E3=81=97?= =?UTF-8?q?=E3=81=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 解析を SessionParser に切り出し、前回読んだ位置から先だけを読み足せる ようにした。書きかけの行は消費せず次回に持ち越す。先頭と解析済み末尾の バイト列で「追記の続きか」を判定し、違えば頭から読み直す。 Store はファイルごとに Entry (パーサ + ロック) を持つ。追記解析は状態を 持つので、同じファイルを複数スレッドで同時に解析しないようにしている。 実測 (追記1行あたり): 1.4MB 13.1ms -> 7.8ms、14MB 59.0ms -> 8.2ms。 残りはファイル open の実費で、ログサイズによらず一定になった。 Co-Authored-By: Claude Opus 5 --- README.md | 1 + chktime.py | 394 ++++++++++++++++++++++++++++++++--------------------- 2 files changed, 242 insertions(+), 153 deletions(-) diff --git a/README.md b/README.md index 4f48397..4e54188 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,7 @@ python chktime.py - `[Request interrupted...]`（Esc 中断）が含まれるターンには ⚠ を付けます。 - セッション再開時に書かれるメタ行（attachment / system / mode 等）は、ターンの所要時間には数えません。 - 解析結果はファイルの更新時刻＋サイズでキャッシュします。変わったファイルだけ再解析するので、自動追従がオンなら手動操作は不要です。 +- 追記されただけのファイルは**増えたぶんだけ**読み足します（書きかけの行は次回に持ち越し）。先頭と解析済み末尾のバイト列が変わっていた場合＝ログが書き換えられた場合は頭から読み直します。 ## ファイル diff --git a/chktime.py b/chktime.py index 09d35a7..b17a52b 100644 --- a/chktime.py +++ b/chktime.py @@ -138,166 +138,232 @@ def new_turn(index, kind, text, ts): } -def parse_file(path: Path): - """1セッションの jsonl を解析して dict を返す。""" - turns = [] - titles = {} - meta = {} - counts = {"user": 0, "assistant": 0, "toolResult": 0, "sidechain": 0} - start = end = None - pending_prompt = None # turn 開始待ち (queue-operation より後の user 本体を採用) +TAIL_BYTES = 1024 # 追記の続きかを確かめるために覚えておく末尾のバイト数 - with path.open("rb") as fh: - for raw in fh: - raw = raw.strip() - if not raw: - continue - # tool_result 行は巨大なので JSON パースせず時刻だけ拾う - if b'"tool_result"' in raw and b'"type":"user"' in raw: - match = TS_RE.search(raw) - ts = parse_ts(match.group(1).decode("utf-8")) if match else None - counts["toolResult"] += 1 - if ts is not None: - start = ts if start is None else min(start, ts) - end = ts if end is None else max(end, ts) - if turns: - turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts) - continue +class SessionParser: + """1セッションの jsonl を解析する。追記されたぶんだけ読み足せる。 - try: - obj = json.loads(raw) - except (ValueError, UnicodeDecodeError): - continue + read() は「改行で終わっている行」までしか消費しないので、書きかけの行が + あっても次回の read() で読み直される。ログは追記されるだけという前提だが、 + 書き換えられた場合は resumable() が False を返すので作り直せばよい。 + """ - rtype = obj.get("type") + def __init__(self, path: Path): + self.path = path + self.turns = [] + self.titles = {} + self.meta = {} + self.counts = {"user": 0, "assistant": 0, "toolResult": 0, "sidechain": 0} + self.start = None + self.end = None + self.pos = 0 # 解析済みのバイト位置 + self.head = b"" # 先頭 TAIL_BYTES バイト + self.tail = b"" # pos の直前 TAIL_BYTES バイト - if rtype == "custom-title": - titles["custom"] = obj.get("customTitle") - continue - if rtype == "ai-title": - titles["ai"] = obj.get("aiTitle") - continue - if rtype == "last-prompt": - titles["last"] = obj.get("lastPrompt") - continue + # -- 追記の読み取り ------------------------------------------------ - ts = parse_ts(obj.get("timestamp")) + def read(self, size=None): + """未読部分を読んで解析する。前回の続きでなければ False を返す。 + + 追記されるだけのログを前提に、先頭と解析済み末尾のバイト列だけを見て + 「続き」かを判定する。ファイル open は Windows では数 ms かかるので、 + 確認と読み取りは 1 回の open で済ませる。 + """ + if size is not None and size < self.pos: + return False # 縮んだ = 書き換えられた + with self.path.open("rb") as fh: + if self.pos: + if fh.read(len(self.head)) != self.head: + return False + fh.seek(self.pos - len(self.tail)) + if fh.read(len(self.tail)) != self.tail: + return False + else: + fh.seek(0) + chunk = fh.read() + if not chunk: + return True + lines = chunk.split(b"\n") + partial = lines.pop() # 改行で終わっていなければ書きかけの行 + used = len(chunk) - len(partial) + for raw in lines: + self.feed(raw) + if used: + if len(self.head) < TAIL_BYTES: + self.head = (self.head + chunk[:used])[:TAIL_BYTES] + self.tail = (self.tail + chunk[:used])[-TAIL_BYTES:] + self.pos += used + return True + + # -- 1行の解析 ---------------------------------------------------- + + def bump(self, ts): + self.start = ts if self.start is None else min(self.start, ts) + self.end = ts if self.end is None else max(self.end, ts) + + def extend_last(self, ts): + """直前のターンの終端を伸ばす (サブエージェント・ツール結果・中断)。""" + if self.turns and ts is not None: + last = self.turns[-1] + last["endTs"] = max(last["endTs"] or ts, ts) + + def feed(self, raw): + raw = raw.strip() + if not raw: + return + + # tool_result 行は巨大なので JSON パースせず時刻だけ拾う + if b'"tool_result"' in raw and b'"type":"user"' in raw: + match = TS_RE.search(raw) + ts = parse_ts(match.group(1).decode("utf-8")) if match else None + self.counts["toolResult"] += 1 if ts is not None: - start = ts if start is None else min(start, ts) - end = ts if end is None else max(end, ts) + self.bump(ts) + self.extend_last(ts) + return - for key in ("cwd", "gitBranch", "version", "slug", "sessionId"): - if not meta.get(key) and obj.get(key): - meta[key] = obj[key] + try: + obj = json.loads(raw) + except (ValueError, UnicodeDecodeError): + return - if rtype == "user": - counts["user"] += 1 - parsed = classify_user(obj) - if not parsed: - if obj.get("isSidechain"): - counts["sidechain"] += 1 - if turns and ts is not None: - turns[-1]["sidechain"] += 1 - turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts) - continue - kind, text = parsed - if kind == "interrupt": - if turns: - turns[-1]["interrupted"] = True - if ts is not None: - turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts) - continue - pending_prompt = None - turns.append(new_turn(len(turns) + 1, kind, text, ts)) - continue + rtype = obj.get("type") - if rtype == "assistant": + if rtype == "custom-title": + self.titles["custom"] = obj.get("customTitle") + return + if rtype == "ai-title": + self.titles["ai"] = obj.get("aiTitle") + return + if rtype == "last-prompt": + self.titles["last"] = obj.get("lastPrompt") + return + + ts = parse_ts(obj.get("timestamp")) + if ts is not None: + self.bump(ts) + + for key in ("cwd", "gitBranch", "version", "slug", "sessionId"): + if not self.meta.get(key) and obj.get(key): + self.meta[key] = obj[key] + + if rtype == "user": + self.counts["user"] += 1 + parsed = classify_user(obj) + if not parsed: if obj.get("isSidechain"): - counts["sidechain"] += 1 - if turns and ts is not None: - turns[-1]["sidechain"] += 1 - turns[-1]["endTs"] = max(turns[-1]["endTs"] or ts, ts) - continue - counts["assistant"] += 1 - if turns: - turn = turns[-1] - turn["assistant"] += 1 - if ts is not None: - if turn["firstReplyTs"] is None: - turn["firstReplyTs"] = ts - turn["endTs"] = max(turn["endTs"] or ts, ts) - model = (obj.get("message") or {}).get("model") - if model and model not in turn["models"]: - turn["models"].append(model) - content = (obj.get("message") or {}).get("content") - if isinstance(content, list): - for block in content: - if isinstance(block, dict) and block.get("type") == "tool_use": - name = block.get("name") or "?" - turn["tools"][name] = turn["tools"].get(name, 0) + 1 - turn["toolCalls"] += 1 - continue + self.counts["sidechain"] += 1 + if self.turns and ts is not None: + self.turns[-1]["sidechain"] += 1 + self.extend_last(ts) + return + kind, text = parsed + if kind == "interrupt": + if self.turns: + self.turns[-1]["interrupted"] = True + self.extend_last(ts) + return + self.turns.append(new_turn(len(self.turns) + 1, kind, text, ts)) + return - # その他 (system / attachment / mode / queue-operation) はセッション範囲にのみ反映。 - # 再開時に書かれる事があり、ターンの所要時間を伸ばしてしまうため無視する。 + if rtype == "assistant": + if obj.get("isSidechain"): + self.counts["sidechain"] += 1 + if self.turns and ts is not None: + self.turns[-1]["sidechain"] += 1 + self.extend_last(ts) + return + self.counts["assistant"] += 1 + if not self.turns: + return + turn = self.turns[-1] + turn["assistant"] += 1 + if ts is not None: + if turn["firstReplyTs"] is None: + turn["firstReplyTs"] = ts + turn["endTs"] = max(turn["endTs"] or ts, ts) + model = (obj.get("message") or {}).get("model") + if model and model not in turn["models"]: + turn["models"].append(model) + content = (obj.get("message") or {}).get("content") + if isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_use": + name = block.get("name") or "?" + turn["tools"][name] = turn["tools"].get(name, 0) + 1 + turn["toolCalls"] += 1 + return - _ = pending_prompt # 予約 + # その他 (system / attachment / mode / queue-operation) はセッション範囲にのみ反映。 + # 再開時に書かれる事があり、ターンの所要時間を伸ばしてしまうため無視する。 - # ターンごとの派生値 - prev_end = None - active = 0.0 - tool_total = 0 - for turn in turns: - ts = turn["ts"] - turn["waitMs"] = ms(turn["firstReplyTs"], ts) - turn["durMs"] = ms(turn["endTs"], ts) - turn["idleMs"] = ms(ts, prev_end) - turn["tsIso"] = iso(ts) - turn["endIso"] = iso(turn["endTs"]) - turn["summary"] = one_line(turn["prompt"], SUMMARY_LIMIT) - turn["toolList"] = sorted(turn["tools"].items(), key=lambda kv: (-kv[1], kv[0])) - tool_total += turn["toolCalls"] - if turn["durMs"]: - active += turn["durMs"] - prev_end = turn["endTs"] if turn["endTs"] is not None else prev_end + # -- 結果 ---------------------------------------------------------- - first_prompt = next((t["prompt"] for t in turns if t["kind"] == "prompt"), "") - title = ( - titles.get("custom") - or titles.get("ai") - or one_line(first_prompt, 120) - or titles.get("last") - or meta.get("slug") - or path.stem - ) + def result(self, stat=None): + """現在までの解析結果を dict にする。派生値はここで計算し直す。""" + stat = stat or self.path.stat() + turns = self.turns + prev_end = None + active = 0.0 + tool_total = 0 + for turn in turns: + ts = turn["ts"] + turn["waitMs"] = ms(turn["firstReplyTs"], ts) + turn["durMs"] = ms(turn["endTs"], ts) + turn["idleMs"] = ms(ts, prev_end) + turn["tsIso"] = iso(ts) + turn["endIso"] = iso(turn["endTs"]) + turn["summary"] = one_line(turn["prompt"], SUMMARY_LIMIT) + turn["toolList"] = sorted(turn["tools"].items(), key=lambda kv: (-kv[1], kv[0])) + tool_total += turn["toolCalls"] + if turn["durMs"]: + active += turn["durMs"] + prev_end = turn["endTs"] if turn["endTs"] is not None else prev_end - stat = path.stat() - return { - "id": path.stem, - "file": str(path), - "title": title, - "titleSource": ("custom" if titles.get("custom") else "ai" if titles.get("ai") else "prompt"), - "cwd": meta.get("cwd"), - "gitBranch": meta.get("gitBranch"), - "version": meta.get("version"), - "slug": meta.get("slug"), - "start": iso(start), - "end": iso(end), - "startEpoch": start, - "endEpoch": end, - "wallMs": ms(end, start), - "activeMs": int(active) if active else 0, - "turnCount": sum(1 for t in turns if t["kind"] in ("prompt", "command")), - "userMsgs": counts["user"], - "assistantMsgs": counts["assistant"], - "toolResults": counts["toolResult"], - "sidechainMsgs": counts["sidechain"], - "toolCalls": tool_total, - "bytes": stat.st_size, - "mtime": stat.st_mtime, - "turns": turns, - } + titles = self.titles + first_prompt = next((t["prompt"] for t in turns if t["kind"] == "prompt"), "") + title = ( + titles.get("custom") + or titles.get("ai") + or one_line(first_prompt, 120) + or titles.get("last") + or self.meta.get("slug") + or self.path.stem + ) + + return { + "id": self.path.stem, + "file": str(self.path), + "title": title, + "titleSource": ("custom" if titles.get("custom") else "ai" if titles.get("ai") else "prompt"), + "cwd": self.meta.get("cwd"), + "gitBranch": self.meta.get("gitBranch"), + "version": self.meta.get("version"), + "slug": self.meta.get("slug"), + "start": iso(self.start), + "end": iso(self.end), + "startEpoch": self.start, + "endEpoch": self.end, + "wallMs": ms(self.end, self.start), + "activeMs": int(active) if active else 0, + "turnCount": sum(1 for t in turns if t["kind"] in ("prompt", "command")), + "userMsgs": self.counts["user"], + "assistantMsgs": self.counts["assistant"], + "toolResults": self.counts["toolResult"], + "sidechainMsgs": self.counts["sidechain"], + "toolCalls": tool_total, + "bytes": stat.st_size, + "mtime": stat.st_mtime, + "turns": turns, + } + + +def parse_file(path: Path): + """1セッションの jsonl を頭から解析して dict を返す。""" + parser = SessionParser(path) + parser.read() + return parser.result() def ms(later, earlier): @@ -328,8 +394,24 @@ def summary_of(session): # ---------------------------------------------------------------- ストア +class Entry: + """1ファイルぶんのキャッシュ。パーサを持ち回して追記分だけ読み足す。""" + + __slots__ = ("lock", "parser", "token", "data") + + def __init__(self, path: Path): + self.lock = threading.Lock() # 同じファイルを同時に解析しない + self.parser = SessionParser(path) + self.token = None # (mtime, size) + self.data = None + + class Store: - """jsonl の解析結果を (mtime, size) でキャッシュする。""" + """jsonl の解析結果を (mtime, size) でキャッシュする。 + + 変化したファイルは、末尾に追記されただけなら追加分だけを解析する。 + 書き換えられていた場合 (縮んだ / 途中が変わった) は頭から読み直す。 + """ def __init__(self, root: Path): self.root = root @@ -368,16 +450,22 @@ class Store: return sorted(directory.glob("*.jsonl")) def session(self, path: Path): - stat = path.stat() key = str(path) with self._lock: - hit = self._cache.get(key) - if hit and hit[0] == (stat.st_mtime, stat.st_size): - return hit[1] - parsed = parse_file(path) - with self._lock: - self._cache[key] = ((stat.st_mtime, stat.st_size), parsed) - return parsed + entry = self._cache.get(key) + if entry is None: + entry = self._cache[key] = Entry(path) + with entry.lock: + stat = path.stat() + token = (stat.st_mtime, stat.st_size) + if entry.token == token and entry.data is not None: + return entry.data + if not entry.parser.read(stat.st_size): + entry.parser = SessionParser(path) # 書き換えられていた: 頭から + entry.parser.read() + entry.data = entry.parser.result(stat) + entry.token = token + return entry.data def sessions(self, project_id): out = [summary_of(self.session(p)) for p in self.session_files(project_id)]