自動追従で追記分だけを解析するようにした

解析を SessionParser に切り出し、前回読んだ位置から先だけを読み足せる
ようにした。書きかけの行は消費せず次回に持ち越す。先頭と解析済み末尾の
バイト列で「追記の続きか」を判定し、違えば頭から読み直す。

Store はファイルごとに Entry (パーサ + ロック) を持つ。追記解析は状態を
持つので、同じファイルを複数スレッドで同時に解析しないようにしている。

実測 (追記1行あたり): 1.4MB 13.1ms -> 7.8ms、14MB 59.0ms -> 8.2ms。
残りはファイル open の実費で、ログサイズによらず一定になった。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-01 09:46:34 +09:00
co-authored by Claude Opus 5
parent 4692043106
commit efc7180317
2 changed files with 242 additions and 153 deletions
+241 -153
View File
@@ -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)]