diff --git a/plugins/ndf/scripts/context-policy.py b/plugins/ndf/scripts/context-policy.py index 134e815bb..215b1a37b 100755 --- a/plugins/ndf/scripts/context-policy.py +++ b/plugins/ndf/scripts/context-policy.py @@ -31,6 +31,7 @@ from typing import Iterable, Mapping sys.path.insert(0, str(Path(__file__).resolve().parent / "lib")) +import context_calibrate # noqa: E402 import context_decisions as cd # noqa: E402 import context_policy as cp # noqa: E402 import step_result as sr # noqa: E402 @@ -58,24 +59,11 @@ def busy_seconds(times: Iterable[float | None]) -> float: def ledger_calls(row: Mapping) -> dict | None: """使用量の帳簿の 1 行を 1 つの呼び出しの形にする。usage が取れなければ None。""" - from usage_ledger import cache_writes - u = row.get("usage") if not isinstance(u, dict): return None - w5, w1h = cache_writes(u) - inp, read = int(u.get("input_tokens") or 0), int(u.get("cache_read_input_tokens") or 0) - return { - "ts": cd.parse_ts(row.get("at")), - "model": row.get("model") or None, - "inp": inp, - "out": int(u.get("output_tokens") or 0), - "read": read, - "w5": w5, - "w1h": w1h, - "context": inp + read + w5 + w1h, - "seconds": row.get("seconds") if isinstance(row.get("seconds"), (int, float)) else None, - } + seconds = row.get("seconds") if isinstance(row.get("seconds"), (int, float)) else None + return cd.usage_call(u, cd.parse_ts(row.get("at")), row.get("model") or None) | {"seconds": seconds} def _bucket_of(ts: float | None, restore_end: float | None, handoff_from: float | None) -> str: @@ -131,6 +119,43 @@ def _within(ts: float | None, windows: list[tuple[float, float]]) -> bool: return ts is not None and any(a <= ts < b for a, b in windows) +def _mark_bounds(marks: list[dict], by_next: bool) -> tuple[float | None, float | None]: + """(復元の区間の終わり, 引継ぎの区間の始まり)。復元は ndf-next で始まったセッションの最初の「復元完了」、引継ぎは最後の「引継ぎ開始」。""" + restored = [cd.parse_ts(m.get("at")) for m in marks if m.get("event") == cd.MARK_RESTORED] + handoffs = [cd.parse_ts(m.get("at")) for m in marks if m.get("event") == cd.MARK_HANDOFF] + restore_end = min((t for t in restored if t is not None), default=None) if by_next else None + handoff_from = max((t for t in handoffs if t is not None), default=None) + return restore_end, handoff_from + + +def _add_subagents(ndf: dict, transcript: str, excluded: set, seen: set, windows: list[tuple[float, float]]) -> None: + """会話の記録の隣の `subagents/` の呼び出しと秒を、NDF が起動した分へ足す。""" + sub = Path(str(transcript).removesuffix(".jsonl")) / "subagents" + for f in sorted(sub.glob("*.jsonl")) if sub.is_dir() else []: + if f.stem.removeprefix("agent-") in excluded: + continue # SubagentStop が帳簿へ書いたサブエージェント（session_id = agent_id）は帳簿の行で数える + sub_calls, _, _ = cd.read_calls(f, seen) + sub_calls = [c for c in sub_calls if _within(c["ts"], windows)] + for c in sub_calls: + _add_call(ndf, c) + ndf["seconds"] += busy_seconds(c["ts"] for c in sub_calls) + + +def _add_ledger(ndf: dict, sid: str, span: tuple, ledger_rows: list[dict], seen: set, windows: list[tuple[float, float]]) -> None: + """このセッションが親か、親の無い行で時刻がセッションの範囲に入る帳簿の行を、NDF が起動した分へ 1 回だけ足す。""" + for row in ledger_rows: + at = cd.parse_ts(row.get("at")) + mine = row.get("parent_session_id") == sid or ( + not row.get("parent_session_id") and span[0] is not None and at is not None and span[0] <= at <= span[1] + ) + key = ("ledger", row.get("session_id"), row.get("at"), row.get("step")) + c = ledger_calls(row) if mine and key not in seen and _within(at, windows) else None + if c is not None: + seen.add(key) + _add_call(ndf, c) + ndf["seconds"] += c["seconds"] or 0.0 + + def session_parts( sid: str, transcript: str | None, @@ -144,10 +169,7 @@ def session_parts( windows = ALL_TIME if windows is None else windows calls, compacts, by_next = cd.read_calls(transcript, seen) if transcript and sid not in excluded else ([], None, False) calls = [c for c in calls if _within(c["ts"], windows)] - restored = [cd.parse_ts(m.get("at")) for m in marks if m.get("event") == cd.MARK_RESTORED] - handoffs = [cd.parse_ts(m.get("at")) for m in marks if m.get("event") == cd.MARK_HANDOFF] - restore_end = min((t for t in restored if t is not None), default=None) if by_next else None - handoff_from = max((t for t in handoffs if t is not None), default=None) + restore_end, handoff_from = _mark_bounds(marks, by_next) parts = {b: _empty_totals() for b in BUCKETS} times: dict[str, list] = defaultdict(list) for c in calls: @@ -155,27 +177,9 @@ def session_parts( _add_call(parts[b], c) times[b].append(c["ts"]) if transcript: - sub = Path(str(transcript).removesuffix(".jsonl")) / "subagents" - for f in sorted(sub.glob("*.jsonl")) if sub.is_dir() else []: - if f.stem.removeprefix("agent-") in excluded: - continue # SubagentStop が帳簿へ書いたサブエージェント（session_id = agent_id）は下の帳簿の行で数える - sub_calls, _, _ = cd.read_calls(f, seen) - sub_calls = [c for c in sub_calls if _within(c["ts"], windows)] - for c in sub_calls: - _add_call(parts["ndf"], c) - parts["ndf"]["seconds"] += busy_seconds(c["ts"] for c in sub_calls) + _add_subagents(parts["ndf"], transcript, excluded, seen, windows) span = (min((c["ts"] for c in calls if c["ts"]), default=None), max((c["ts"] for c in calls if c["ts"]), default=None)) - for row in ledger_rows: - at = cd.parse_ts(row.get("at")) - mine = row.get("parent_session_id") == sid or ( - not row.get("parent_session_id") and span[0] is not None and at is not None and span[0] <= at <= span[1] - ) - key = ("ledger", row.get("session_id"), row.get("at"), row.get("step")) - c = ledger_calls(row) if mine and key not in seen and _within(at, windows) else None - if c is not None: - seen.add(key) - _add_call(parts["ndf"], c) - parts["ndf"]["seconds"] += c["seconds"] or 0.0 + _add_ledger(parts["ndf"], sid, span, ledger_rows, seen, windows) for b, ts in times.items(): parts[b]["seconds"] += busy_seconds(ts) return { @@ -188,27 +192,24 @@ def session_parts( } -def report_work(work: str, rows: list[dict], ledger_rows: list[dict] | None = None) -> dict: - """作業の識別子 1 つの集計（受け入れ条件 12・13、I12〜I14）。""" - ledger_rows = ledger_rows or [] - excluded = {r.get("session_id") for r in ledger_rows if r.get("session_id")} - filled = cd.fill_work(rows) - windows = work_windows(filled, work) - rows = [r for r in filled if (r.get("work") or "") == work] - sessions = cd.sessions_of(rows) - seen: set = set() +def _session_spans(sessions: dict, excluded: set, windows: dict) -> dict[str, tuple[float, float]]: + """セッションごとの (最初, 最後) の時刻。作業の区間の呼び出しと判定の記録の行から取る（印の割り当てに使う）。""" spans = {} - marks = [r for r in rows if r.get("event") in cd.MARKS] for sid, s in sessions.items(): calls, _, _ = cd.read_calls(s["transcript"], set()) if s["transcript"] and sid not in excluded else ([], None, False) calls = [c for c in calls if _within(c["ts"], windows.get(sid, ALL_TIME))] ts = [c["ts"] for c in calls if c["ts"]] + [cd.parse_ts(r.get("at")) for r in s["rows"] if cd.parse_ts(r.get("at"))] if ts: spans[sid] = (min(ts), max(ts)) - by_session = cd.attach_marks(marks, spans) + return spans + + +def _sum_sessions(sessions: dict, by_session: dict, ledger_rows: list[dict], excluded: set, windows: dict) -> dict: + """セッションごとの区分を合算し、自動圧縮の数と印の無い区間の数を数える。""" totals = {b: _empty_totals() for b in BUCKETS} compacts: int | None = 0 no_restore_mark = no_handoff_mark = 0 + seen: set = set() for sid, s in sessions.items(): summary = session_parts(sid, s["transcript"], by_session.get(sid, []), ledger_rows, excluded, seen, windows.get(sid)) for b in BUCKETS: @@ -218,33 +219,50 @@ def report_work(work: str, rows: list[dict], ledger_rows: list[dict] | None = No no_restore_mark += 0 if summary["restore_known"] else 1 restarted = any(r.get("verdict") == cp.RESTART for r in s["rows"]) no_handoff_mark += 1 if restarted and not summary["handoff_known"] else 0 + for b in BUCKETS: + totals[b]["cost"] = round(totals[b]["cost"]) + totals[b]["seconds"] = round(totals[b]["seconds"]) + return {"parts": totals, "compacts": compacts, "unknown_parts": {"restore": no_restore_mark, "handoff": no_handoff_mark}} + + +def _restarts_by_reason(rows: list[dict]) -> dict[str, int]: restarts: dict[str, int] = defaultdict(int) for r in rows: if r.get("verdict") == cp.RESTART: restarts[r.get("reason") or cd.UNKNOWN] += 1 - keep = defaultdict(int) + return dict(restarts) + + +def _rework(rows: list[dict]) -> int | str: + """「継続」の判定が 2 回以上ある工程の数（やり直し）。「継続」の判定が無ければ「不明」。""" + keep: dict[str, int] = defaultdict(int) for r in rows: if r.get("event", cd.EVENT_DECISION) == cd.EVENT_DECISION and r.get("verdict") == cp.CONTINUE and r.get("stage"): keep[r["stage"]] += 1 - for b in BUCKETS: - totals[b]["cost"] = round(totals[b]["cost"]) - totals[b]["seconds"] = round(totals[b]["seconds"]) + return sum(1 for n in keep.values() if n >= 2) if keep else cd.UNKNOWN + + +def report_work(work: str, rows: list[dict], ledger_rows: list[dict] | None = None) -> dict: + """作業の識別子 1 つの集計（受け入れ条件 12・13、I12〜I14）。""" + ledger_rows = ledger_rows or [] + excluded = {r.get("session_id") for r in ledger_rows if r.get("session_id")} + filled = cd.fill_work(rows) + windows = work_windows(filled, work) + rows = [r for r in filled if (r.get("work") or "") == work] + sessions = cd.sessions_of(rows) + by_session = cd.attach_marks([r for r in rows if r.get("event") in cd.MARKS], _session_spans(sessions, excluded, windows)) + summed = _sum_sessions(sessions, by_session, ledger_rows, excluded, windows) + restarts = _restarts_by_reason(rows) + compacts = summed["compacts"] return { "work": work, "sessions": len(sessions), - "parts": totals, - "unknown_parts": { - "restore": no_restore_mark, - "handoff": no_handoff_mark, - }, - "restarts": dict(restarts), + "parts": summed["parts"], + "unknown_parts": summed["unknown_parts"], + "restarts": restarts, "restart_count": sum(restarts.values()), "auto_compacts": cd.UNKNOWN if compacts is None or not sessions else compacts, - "quality": { - "rework": sum(1 for n in keep.values() if n >= 2) if keep else cd.UNKNOWN, - "review_fixes": cd.UNKNOWN, - "missed_requirements": cd.UNKNOWN, - }, + "quality": {"rework": _rework(rows), "review_fixes": cd.UNKNOWN, "missed_requirements": cd.UNKNOWN}, "no_policy_rows": sum(1 for r in rows if "policy" not in r), } @@ -291,24 +309,14 @@ def cmd_record(a) -> None: if a.verdict is not None and (a.verdict != cp.RESTART or a.reason not in cp.REASONS): sr.emit(sr.result(TOOL, "stopped", f"--verdict 再起動 と、理由（{' / '.join(cp.REASONS)}）を渡す"), sr.EXIT_UNREADABLE) root = cd.toplevel(None) - sid = os.environ.get("CLAUDE_CODE_SESSION_ID") if a.runtime == "claude" else None - row = cli_row(a.runtime, "record", sid or None, a.work, root, a.context) - if a.event: - row.update(event=a.event) - else: - row.update(stage="cut", verdict=cp.RESTART, reason=a.reason) + row = cd.record_row(a.runtime, "record", a.work, cp.resolve_policy(root), a.context, a.event, a.reason) if not cd.append_decision(row, cd.decisions_repo(root)): sr.emit(sr.result(TOOL, "stopped", "判定の記録へ書けない", metrics={"dir": str(cd.context_state_dir())}), sr.EXIT_VIOLATION) sr.emit(sr.result(TOOL, "ok", f"{a.event or a.verdict} を記録した（{a.work}）", [row])) -def cli_row(runtime: str, source: str, sid: str | None, work: str, root: str | None, context: int | None) -> dict: - """CLI と sprint-state.py が書く行（判定をしない。`band` は null）。""" - return cd.boundary_row(runtime=runtime, source=source, session_id=sid, work=work, context=context, applied=cp.resolve_policy(root)) - - def cmd_calibrate(a) -> None: - body, ok = cd.calibrate_repo(cd.toplevel(None), a.days) + body, ok = context_calibrate.calibrate_repo(cd.toplevel(None), a.days) if not ok: sr.emit(sr.result(TOOL, "stopped", "見込みのファイルへ書けない"), sr.EXIT_VIOLATION) sr.emit(sr.result(TOOL, "ok", f"見込みを較正した（標本 {body['samples']}）", [body], {"samples": body["samples"]})) diff --git a/plugins/ndf/scripts/hook_lib/token_guard.py b/plugins/ndf/scripts/hook_lib/token_guard.py index 6e3e8d3ec..8507bf310 100644 --- a/plugins/ndf/scripts/hook_lib/token_guard.py +++ b/plugins/ndf/scripts/hook_lib/token_guard.py @@ -22,7 +22,6 @@ import json import os import re -import subprocess import time from collections import deque from contextlib import contextmanager @@ -330,12 +329,14 @@ def _context_deny(total: int, limit: int, nxt: str, notice: str | None, cause: s def _trial_cause(applied, verdict) -> str | None: """試行方式の拒否の句。現行方式は None（文面を変えない。I1）。""" - if applied.policy != "trial": + import context_policy as cp + + if applied.policy != cp.TRIAL: return None tag = "再起動の方針: 試行方式" - if verdict.reason == "容量": + if verdict.reason == cp.REASON_CAPACITY: return f"容量（窓 {verdict.effective_limit + applied.margin} − 余白 {applied.margin} = {verdict.effective_limit}）を超えた（{tag}、理由: 容量）" - if verdict.reason == "費用": + if verdict.reason == cp.REASON_COST: e = verdict.estimate.to_dict() return f"判断開始の値 {applied.decide_from} を超え、再起動のほうが安い（{tag}、理由: 費用（続ける {e['continue']} / 再起動 {e['restart']}、入力の単価に換算））" return f"上限 {applied.limit}（{tag}、理由: 運用上限）を超えた" @@ -451,14 +452,10 @@ def plan_hint(raw: dict) -> dict | None: return None if _field(raw, "tool_input", "subagent_type") not in ("ndf:supervisor", "ndf:supervisor-waits"): return None - try: - p = subprocess.run( - ["git", "-C", _field(raw, "cwd") or ".", "rev-parse", "--show-toplevel"], capture_output=True, text=True, timeout=5 - ) - except (OSError, subprocess.SubprocessError): - return None - root = p.stdout.strip() - if p.returncode != 0 or not root: + import context_decisions as cd + + root = cd.toplevel(_field(raw, "cwd") or None) + if not root: return None if not (Path(root, ".ndf", "supervise.json").is_file() or Path(root, ".ndf", "worktree.json").is_file()): return None diff --git a/plugins/ndf/scripts/lib/clock.py b/plugins/ndf/scripts/lib/clock.py index e20067bb4..ef3e8938b 100644 --- a/plugins/ndf/scripts/lib/clock.py +++ b/plugins/ndf/scripts/lib/clock.py @@ -73,6 +73,12 @@ def parse(value: Any, naive: str = "local") -> Optional[_dt.datetime]: return None +def epoch_seconds(value: Any, naive: str = "local") -> Optional[float]: + """`parse` で読んだ時刻の UNIX 時刻（秒）。読めなければ `None`。""" + parsed = parse(value, naive) + return None if parsed is None else parsed.timestamp() + + def seconds_between(start: Any, end: Any, naive: str = "local") -> Optional[float]: """`end − start` の秒。どちらかが読めなければ `None`。""" s, e = parse(start, naive), parse(end, naive) diff --git a/plugins/ndf/scripts/lib/context_calibrate.py b/plugins/ndf/scripts/lib/context_calibrate.py new file mode 100644 index 000000000..a22947637 --- /dev/null +++ b/plugins/ndf/scripts/lib/context_calibrate.py @@ -0,0 +1,109 @@ +"""費用の見込みの較正（`calibrate`。#2262 の決定 5・6）。 + +判定の記録と会話の記録の全体を読む重い側で、hook は使わない（hook が使うのは `context_decisions` の記録の追記と +小さな JSON の読み書きだけ。決定 6）。 +""" + +from __future__ import annotations + +import statistics +import time +from collections import defaultdict + +import context_decisions as cd +import context_policy as cp +from usage_ledger import now_utc + +CACHE_1H = 3600 # 呼び出しの間隔がこれを超えたら、1 時間のキャッシュが切れたとみなす + + +def _median_of(values: list[float]) -> float | None: + return float(statistics.median(values)) if values else None + + +def _mark_times(marks: list[dict], event: str) -> list[float]: + return sorted(t for m in marks if m.get("event") == event and (t := cd.parse_ts(m.get("at")))) + + +def _restore_sample(calls: list[dict], marks: list[dict], by_next: bool) -> tuple[int, int] | None: + """ndf-next で始まり復元の印があるセッションの、最初の復元までの (直前のコンテキスト量, 呼び出し数)。""" + restored = _mark_times(marks, cd.MARK_RESTORED) + if not (by_next and restored): + return None + before = [c for c in calls if c["ts"] is not None and c["ts"] <= restored[0]] + return (before[-1]["context"], len(before)) if before else None + + +def _handoff_sample(calls: list[dict], marks: list[dict]) -> tuple[int, int] | None: + """最後の引継ぎの印から後の (呼び出し数, 出力 token の合計)。印が無ければ None。""" + handoffs = _mark_times(marks, cd.MARK_HANDOFF) + if not handoffs: + return None + after = [c for c in calls if c["ts"] is not None and c["ts"] >= handoffs[-1]] + return len(after), sum(c["out"] for c in after) + + +def _stage_segments(calls: list[dict], rows: list[dict]) -> list[tuple[str, int, int]]: + """「継続」の判定で区切った工程ごとの (工程, 呼び出し数, キャッシュ切れの数)。最後の区間と空の区間は除く。""" + launches = sorted( + ((t, r["stage"]) for r in rows if r.get("stage") and r.get("verdict") == cp.CONTINUE and (t := cd.parse_ts(r.get("at")))), + ) + out = [] + for (start, stage), nxt in zip(launches, [t for t, _ in launches[1:]] + [None]): + seg = [c for c in calls if c["ts"] is not None and c["ts"] >= start and (nxt is None or c["ts"] < nxt)] + if not seg or nxt is None: + continue + gaps = sum(1 for a, b in zip(seg, seg[1:]) if b["ts"] - a["ts"] > CACHE_1H) + out.append((stage, len(seg), gaps)) + return out + + +def calibrate(rows: list[dict], since: float | None = None) -> dict: + """判定の記録と会話の記録から費用の見込みを作る（決定 5・6）。標本が 3 未満の値は None。""" + rows = [r for r in cd.fill_work(rows) if since is None or (cd.parse_ts(r.get("at")) or 0) >= since] + sessions = cd.sessions_of(rows) + restore_q, restore_n, handoff_h, handoff_out = [], [], [], [] + stage_k: dict[str, list] = defaultdict(list) + stage_m: dict[str, list] = defaultdict(list) + spans = {} + cache: dict[str, tuple] = {} + for sid, s in sessions.items(): + cache[sid] = cd.read_calls(s["transcript"], set()) if s["transcript"] else ([], None, False) + ts = [c["ts"] for c in cache[sid][0] if c["ts"]] + if ts: + spans[sid] = (min(ts), max(ts)) + by_session = cd.attach_marks([r for r in rows if r.get("event") in cd.MARKS], spans) + for sid, s in sessions.items(): + calls, _, by_next = cache[sid] + marks = by_session.get(sid, []) + if (restore := _restore_sample(calls, marks, by_next)) is not None: + restore_q.append(restore[0]) + restore_n.append(restore[1]) + if (handoff := _handoff_sample(calls, marks)) is not None: + handoff_h.append(handoff[0]) + handoff_out.append(handoff[1]) + for stage, n, gaps in _stage_segments(calls, s["rows"]): + for key in (stage, "*"): + stage_k[key].append(n) + stage_m[key].append(gaps) + + def med(values): + return _median_of(values) if len(values) >= cp.MIN_SAMPLES else None + + return { + "restore_context": med(restore_q), + "restore_calls": med(restore_n), + "handoff_calls": med(handoff_h), + "handoff_output": med(handoff_out), + "stage_calls": {k: _median_of(v) for k, v in stage_k.items()}, + "stage_expiries": {k: _median_of(v) for k, v in stage_m.items()}, + "samples": {"restore": len(restore_q), "handoff": len(handoff_h)} | {f"stage:{k}": len(v) for k, v in stage_k.items()}, + "at": now_utc(), + } + + +def calibrate_repo(root: str | None, days: int = 30) -> tuple[dict, bool]: + """そのリポジトリの直近 `days` 日の記録から見込みを較正して書く。(見込み, 書けたか)。""" + repo = cd.decisions_repo(root) + body = calibrate(cd.read_decisions(repo), since=time.time() - days * 86400) + return body, cd.write_cost_estimates(body, repo) diff --git a/plugins/ndf/scripts/lib/context_decisions.py b/plugins/ndf/scripts/lib/context_decisions.py index ad0da5352..8cd3670fd 100644 --- a/plugins/ndf/scripts/lib/context_decisions.py +++ b/plugins/ndf/scripts/lib/context_decisions.py @@ -19,23 +19,21 @@ import json import os import re -import statistics import time from collections import defaultdict import subprocess -from datetime import datetime from pathlib import Path -from typing import Mapping +from typing import Iterator, Mapping +import clock import context_policy as cp import jsonio from context_policy import REASONS # noqa: F401 sprint-state.py の --reason の選択肢 -from usage_ledger import now_utc +from usage_ledger import cache_writes, now_utc SCHEMA = 1 SESSION_DAYS = 7 LOCAL = "_local" -CACHE_1H = 3600 # 呼び出しの間隔がこれを超えたら、1 時間のキャッシュが切れたとみなす MARK_HANDOFF, MARK_RESTORED = "引継ぎ開始", "復元完了" MARKS = (MARK_HANDOFF, MARK_RESTORED) EVENT_DECISION = "判定" @@ -94,13 +92,7 @@ def stage_of(key: str) -> str: return "plan" -def parse_ts(value) -> float | None: - if not isinstance(value, str) or not value: - return None - try: - return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() - except ValueError: - return None +parse_ts = clock.epoch_seconds # 記録の時刻（ISO 8601）を UNIX 時刻（秒）にする。読めなければ None # --- 判定の記録 ---------------------------------------------------------------------------------------- @@ -237,7 +229,7 @@ def session_file(sid: str, kind: str, env: Mapping | None = None) -> Path: def read_window(sid: str, env: Mapping | None = None) -> int | None: """statusline が書いた窓の大きさ。無い・形が違えば None（推測で埋めない。I7）。""" w = jsonio.read(session_file(sid, "window", env), missing={}, broken={}, want=dict).get("window") - return w if isinstance(w, int) and not isinstance(w, bool) and w > 0 else None + return w if cp.is_positive_int(w) else None def read_applied(sid: str, env: Mapping | None = None) -> dict: @@ -267,39 +259,60 @@ def write_applied(sid: str, applied, env: Mapping | None = None) -> bool: # --- 会話の記録 --------------------------------------------------------------------------------------- -def read_calls(path, seen: set | None = None) -> tuple[list[dict], int | None, bool]: - """会話の記録の API 呼び出し（同じ `message.id` は 1 回。I13）と、自動圧縮の回数と、ndf-next で始まったか。 +def usage_call(usage: Mapping, ts: float | None, model) -> dict: + """API 呼び出し 1 回の usage を {ts, model, inp, out, read, w5, w1h, context} にする（会話の記録と使用量の帳簿で共通）。""" + w5, w1h = cache_writes(usage) + inp, out, read = (int(usage.get(k) or 0) for k in ("input_tokens", "output_tokens", "cache_read_input_tokens")) + return {"ts": ts, "model": model, "inp": inp, "out": out, "read": read, "w5": w5, "w1h": w1h, "context": inp + read + w5 + w1h} - 同じ `message.id` の行は最後の usage を採る（ストリーミングで後の行ほど output_tokens が増える。scripts/token-usage.py と同じ）。 - 時刻は最初に現れた行のもの。`seen` に前のファイルで数えた id があれば、このファイルでは数えない。 - 呼び出しは {ts, model, inp, out, read, w5, w1h, context}。読めなければ ([], None, False)。""" - from usage_ledger import cache_writes +def _is_auto_compact(e: dict) -> bool: + meta = e.get("compactMetadata") + return e.get("type") == "system" and e.get("subtype") == "compact_boundary" and isinstance(meta, dict) and meta.get("trigger") == "auto" - seen = set() if seen is None else seen - calls: list[dict] = [] - here: dict = {} # このファイルで数えた message.id → calls の位置 - compacts = 0 - first_user: str | None = None + +def _user_text(e: dict) -> str | None: + """利用者の発言（メタの行を除く）の本文。文字列でなければ JSON にする。発言でなければ None。""" + if e.get("type") != "user" or e.get("isMeta"): + return None + content = (e.get("message") if isinstance(e.get("message"), dict) else {}).get("content") + return content if isinstance(content, str) else json.dumps(content, ensure_ascii=False) + + +def _json_lines(path) -> Iterator[dict] | None: + """JSONL の辞書の行を 1 行ずつ返す（全行を辞書のまま保持しない）。読めなければ None。壊れた行と辞書でない行は飛ばす。""" try: lines = Path(path).read_text(encoding="utf-8", errors="replace").splitlines() except OSError: + return None + return (e for e in map(_loads_or_none, lines) if isinstance(e, dict)) + + +def _loads_or_none(line: str): + try: + return json.loads(line) + except ValueError: + return None + + +def read_calls(path, seen: set | None = None) -> tuple[list[dict], int | None, bool]: + """会話の記録の API 呼び出し（同じ `message.id` は 1 回。I13）と、自動圧縮の回数と、ndf-next で始まったか。 + 同じ `message.id` の行は最後の usage を採る（ストリーミングで後の行ほど output_tokens が増える。scripts/token-usage.py と同じ）。 + 時刻は最初に現れた行のもの。`seen` に前のファイルで数えた id があれば、このファイルでは数えない。 + 呼び出しは {ts, model, inp, out, read, w5, w1h, context}。読めなければ ([], None, False)。""" + seen = set() if seen is None else seen + entries = _json_lines(path) + if entries is None: return [], None, False - for line in lines: - try: - e = json.loads(line) - except ValueError: - continue - if not isinstance(e, dict): - continue - if e.get("type") == "system" and e.get("subtype") == "compact_boundary": - meta = e.get("compactMetadata") - compacts += 1 if isinstance(meta, dict) and meta.get("trigger") == "auto" else 0 + calls, here = [], {} # here はこのファイルで数えた message.id → calls の位置 + compacts, first_user = 0, None + for e in entries: # 呼び出し・最初の利用者発言・自動圧縮を 1 回の走査で集める + if _is_auto_compact(e): + compacts += 1 continue + if first_user is None: + first_user = _user_text(e) msg = e.get("message") if isinstance(e.get("message"), dict) else {} - if e.get("type") == "user" and first_user is None and not e.get("isMeta"): - content = msg.get("content") - first_user = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False) usage = msg.get("usage") if e.get("type") != "assistant" or not isinstance(usage, dict): continue @@ -307,18 +320,7 @@ def read_calls(path, seen: set | None = None) -> tuple[list[dict], int | None, b if mid in seen and mid not in here: continue seen.add(mid) - w5, w1h = cache_writes(usage) - inp, read = int(usage.get("input_tokens") or 0), int(usage.get("cache_read_input_tokens") or 0) - call = { - "ts": parse_ts(e.get("timestamp")), - "model": msg.get("model"), - "inp": inp, - "out": int(usage.get("output_tokens") or 0), - "read": read, - "w5": w5, - "w1h": w1h, - "context": inp + read + w5 + w1h, - } + call = usage_call(usage, parse_ts(e.get("timestamp")), msg.get("model")) if mid in here: call["ts"] = calls[here[mid]]["ts"] or call["ts"] calls[here[mid]] = call @@ -357,99 +359,11 @@ def attach_marks(marks: list[dict], spans: dict[str, tuple[float, float]]) -> di return out -# --- 較正 --------------------------------------------------------------------------------------------- - - -def _median_of(values: list[float]) -> float | None: - return float(statistics.median(values)) if values else None - - -def _mark_times(marks: list[dict], event: str) -> list[float]: - return sorted(t for m in marks if m.get("event") == event and (t := parse_ts(m.get("at")))) - - -def _restore_sample(calls: list[dict], marks: list[dict], by_next: bool) -> tuple[int, int] | None: - """ndf-next で始まり復元の印があるセッションの、最初の復元までの (直前のコンテキスト量, 呼び出し数)。""" - restored = _mark_times(marks, MARK_RESTORED) - if not (by_next and restored): - return None - before = [c for c in calls if c["ts"] is not None and c["ts"] <= restored[0]] - return (before[-1]["context"], len(before)) if before else None - - -def _handoff_sample(calls: list[dict], marks: list[dict]) -> tuple[int, int] | None: - """最後の引継ぎの印から後の (呼び出し数, 出力 token の合計)。印が無ければ None。""" - handoffs = _mark_times(marks, MARK_HANDOFF) - if not handoffs: - return None - after = [c for c in calls if c["ts"] is not None and c["ts"] >= handoffs[-1]] - return len(after), sum(c["out"] for c in after) - - -def _stage_segments(calls: list[dict], rows: list[dict]) -> list[tuple[str, int, int]]: - """「継続」の判定で区切った工程ごとの (工程, 呼び出し数, キャッシュ切れの数)。最後の区間と空の区間は除く。""" - launches = sorted( - ((t, r["stage"]) for r in rows if r.get("stage") and r.get("verdict") == cp.CONTINUE and (t := parse_ts(r.get("at")))), - ) - out = [] - for (start, stage), nxt in zip(launches, [t for t, _ in launches[1:]] + [None]): - seg = [c for c in calls if c["ts"] is not None and c["ts"] >= start and (nxt is None or c["ts"] < nxt)] - if not seg or nxt is None: - continue - gaps = sum(1 for a, b in zip(seg, seg[1:]) if b["ts"] - a["ts"] > CACHE_1H) - out.append((stage, len(seg), gaps)) - return out - - -def calibrate(rows: list[dict], since: float | None = None) -> dict: - """判定の記録と会話の記録から費用の見込みを作る（決定 5・6）。標本が 3 未満の値は None。""" - rows = [r for r in fill_work(rows) if since is None or (parse_ts(r.get("at")) or 0) >= since] - sessions = sessions_of(rows) - restore_q, restore_n, handoff_h, handoff_out = [], [], [], [] - stage_k: dict[str, list] = defaultdict(list) - stage_m: dict[str, list] = defaultdict(list) - spans = {} - cache: dict[str, tuple] = {} - for sid, s in sessions.items(): - cache[sid] = read_calls(s["transcript"], set()) if s["transcript"] else ([], None, False) - ts = [c["ts"] for c in cache[sid][0] if c["ts"]] - if ts: - spans[sid] = (min(ts), max(ts)) - by_session = attach_marks([r for r in rows if r.get("event") in MARKS], spans) - for sid, s in sessions.items(): - calls, _, by_next = cache[sid] - marks = by_session.get(sid, []) - if (restore := _restore_sample(calls, marks, by_next)) is not None: - restore_q.append(restore[0]) - restore_n.append(restore[1]) - if (handoff := _handoff_sample(calls, marks)) is not None: - handoff_h.append(handoff[0]) - handoff_out.append(handoff[1]) - for stage, n, gaps in _stage_segments(calls, s["rows"]): - for key in (stage, "*"): - stage_k[key].append(n) - stage_m[key].append(gaps) - - def med(values): - return _median_of(values) if len(values) >= cp.MIN_SAMPLES else None - - return { - "restore_context": med(restore_q), - "restore_calls": med(restore_n), - "handoff_calls": med(handoff_h), - "handoff_output": med(handoff_out), - "stage_calls": {k: _median_of(v) for k, v in stage_k.items()}, - "stage_expiries": {k: _median_of(v) for k, v in stage_m.items()}, - "samples": {"restore": len(restore_q), "handoff": len(handoff_h)} | {f"stage:{k}": len(v) for k, v in stage_k.items()}, - "at": now_utc(), - } - - -def calibrate_repo(root: str | None, days: int = 30) -> tuple[dict, bool]: - """そのリポジトリの直近 `days` 日の記録から見込みを較正して書く。(見込み, 書けたか)。""" - repo = decisions_repo(root) - body = calibrate(read_decisions(repo), since=time.time() - days * 86400) - return body, write_cost_estimates(body, repo) +def record_row(runtime: str, source: str, work: str, applied, context: int | None = None, event: str | None = None, reason=None) -> dict: + """CLI と sprint-state.py が書く行（判定をしない。`band` は null）。`event` を渡せば印の行、渡さなければ再起動の行。""" + sid = os.environ.get("CLAUDE_CODE_SESSION_ID") if runtime == "claude" else None + row = boundary_row(runtime=runtime, source=source, session_id=sid or None, work=work, context=context, applied=applied) + return row | ({"event": event} if event else {"stage": "cut", "verdict": cp.RESTART, "reason": reason}) def record_restart(work: str, reason: str | None = None, cwd: str | None = None, runtime: str = "claude") -> bool: @@ -464,11 +378,10 @@ def record_restart(work: str, reason: str | None = None, cwd: str | None = None, if reason is None: prev = last_denial(read_decisions(repo), work) reason = (prev or {}).get("reason") or (cp.REASON_CUTPOINT if applied.policy == cp.CURRENT else cp.REASON_MANUAL) - sid = os.environ.get("CLAUDE_CODE_SESSION_ID") if runtime == "claude" else None - row = boundary_row(source="sprint-state", runtime=runtime, session_id=sid or None, work=work, applied=applied) - row.update(stage="cut", verdict=cp.RESTART, reason=reason) - ok = append_decision(row, repo) - calibrate_repo(root) + ok = append_decision(record_row(runtime, "sprint-state", work, applied, reason=reason), repo) + import context_calibrate # 較正は会話の記録の全体を読む重い側。循環を避けて関数の中で読む + + context_calibrate.calibrate_repo(root) return ok except Exception: # noqa: BLE001 — 記録の失敗で ndf-next を止めない return False diff --git a/plugins/ndf/scripts/lib/context_policy.py b/plugins/ndf/scripts/lib/context_policy.py index 88ffd7701..29529a5ae 100644 --- a/plugins/ndf/scripts/lib/context_policy.py +++ b/plugins/ndf/scripts/lib/context_policy.py @@ -130,7 +130,7 @@ def to_dict(self) -> dict: # --- 適用値の解決 ------------------------------------------------------------------------------------ -def _positive_int(v) -> bool: +def is_positive_int(v) -> bool: return isinstance(v, int) and not isinstance(v, bool) and v > 0 @@ -147,7 +147,7 @@ def parse_context_setting(data: dict) -> tuple[dict, str | None]: if not isinstance(trial, dict): return {}, f".ndf/{SETTING}: trial がオブジェクトでない" decide_from, limit = trial.get("decide_from", TRIAL_DECIDE_FROM), trial.get("limit", TRIAL_LIMIT) - if not (_positive_int(decide_from) and _positive_int(limit)): + if not (is_positive_int(decide_from) and is_positive_int(limit)): return {}, f".ndf/{SETTING}: trial.decide_from / trial.limit が正の整数でない" if decide_from > limit: return {}, f".ndf/{SETTING}: trial.decide_from（{decide_from}）が trial.limit（{limit}）より大きい" @@ -170,27 +170,24 @@ def read_context_setting(root) -> tuple[dict, str | None]: return (({}, why) if why else parse_context_setting(data)) if data or why else ({}, None) -def resolve_policy(root=None, env: Mapping | None = None, cli: Mapping | None = None) -> AppliedPolicy: - """適用値を解決する（I2・I3）。cli は {"policy": ..., "limit": ...}（値が None のキーは無視）。 - - `NDF_CONTEXT_LIMIT` は適用中の方針の運用上限を上書きし、判断開始の値がそれを上回れば同じ値へ下げる（決定 2）。 - モデル名と窓の大きさは方針を選ばない（I15）。""" - if env is None: # 読む変数を名前で書く（契約の検査が読みを拾う） - env = {"NDF_CONTEXT_POLICY": os.environ.get("NDF_CONTEXT_POLICY"), "NDF_CONTEXT_LIMIT": os.environ.get("NDF_CONTEXT_LIMIT")} - cli = cli or {} - setting, error = read_context_setting(root) - sources = {"policy": "default", "decide_from": "default", "limit": "default", "margin": "default"} - policy = CURRENT +def _resolve_policy_name(setting: dict, env: Mapping, cli: Mapping) -> tuple[str, str, str | None]: + """方針名を 既定 < config < env < CLI で重ね、(方針名, 出所, 無視した環境変数の記述) を返す。""" + policy, source, ignored = CURRENT, "default", None if setting: - policy = setting["policy"] - sources["policy"] = "config" + policy, source = setting["policy"], "config" env_policy = (env.get("NDF_CONTEXT_POLICY") or "").strip() if env_policy in POLICIES: - policy, sources["policy"] = env_policy, "env" + policy, source = env_policy, "env" elif env_policy: - sources["ignored"] = f"NDF_CONTEXT_POLICY={env_policy}" + ignored = f"NDF_CONTEXT_POLICY={env_policy}" if cli.get("policy") in POLICIES: - policy, sources["policy"] = cli["policy"], "cli" + policy, source = cli["policy"], "cli" + return policy, source, ignored + + +def _resolve_thresholds(policy: str, setting: dict, env: Mapping, cli: Mapping) -> tuple[int, int, int, dict]: + """方針に応じた判断開始・運用上限・余白を重ね、(decide_from, limit, margin, 出所) を返す。""" + sources = {"decide_from": "default", "limit": "default", "margin": "default"} margin = setting.get("margin", MARGIN) if "margin" in setting and policy == TRIAL: sources["margin"] = "config" @@ -203,11 +200,28 @@ def resolve_policy(root=None, env: Mapping | None = None, cli: Mapping | None = env_limit = env.get("NDF_CONTEXT_LIMIT") or "" if env_limit.isdigit() and int(env_limit) > 0: limit, sources["limit"] = int(env_limit), "env" - if _positive_int(cli.get("limit")): + if is_positive_int(cli.get("limit")): limit, sources["limit"] = cli["limit"], "cli" if policy == CURRENT or decide_from > limit: decide_from = limit sources["decide_from"] = sources["limit"] + return decide_from, limit, margin, sources + + +def resolve_policy(root=None, env: Mapping | None = None, cli: Mapping | None = None) -> AppliedPolicy: + """適用値を解決する（I2・I3）。cli は {"policy": ..., "limit": ...}（値が None のキーは無視）。 + + `NDF_CONTEXT_LIMIT` は適用中の方針の運用上限を上書きし、判断開始の値がそれを上回れば同じ値へ下げる（決定 2）。 + モデル名と窓の大きさは方針を選ばない（I15）。""" + if env is None: # 読む変数を名前で書く（契約の検査が読みを拾う） + env = {"NDF_CONTEXT_POLICY": os.environ.get("NDF_CONTEXT_POLICY"), "NDF_CONTEXT_LIMIT": os.environ.get("NDF_CONTEXT_LIMIT")} + cli = cli or {} + setting, error = read_context_setting(root) + policy, policy_source, ignored = _resolve_policy_name(setting, env, cli) + decide_from, limit, margin, threshold_sources = _resolve_thresholds(policy, setting, env, cli) + sources = {"policy": policy_source, **threshold_sources} + if ignored: + sources["ignored"] = ignored return AppliedPolicy(policy, decide_from, limit, margin, sources, error) diff --git a/plugins/ndf/scripts/lib/sprint_loop_select.py b/plugins/ndf/scripts/lib/sprint_loop_select.py index fb9521815..1ba067b1e 100644 --- a/plugins/ndf/scripts/lib/sprint_loop_select.py +++ b/plugins/ndf/scripts/lib/sprint_loop_select.py @@ -244,42 +244,37 @@ def add_env_exclusions(rec: dict, n: int) -> None: rec.setdefault("excluded", []).append({"number": num, "reason": EXCLUDE_REASON, "lap": n}) -def run_selection(env: Env, rec: dict, before: dict, n: int, lap: dict | None = None) -> dict: - add_env_exclusions(rec, n) - code, rank = env.rank(env.main_dir()) - if code not in (0, PAUSE_RANK): - return refusal_order(f"upkeep.py rank が {code} を返した。選定を推さない: {rank.get('summary', '')}") - try: - descriptions = env.milestones(rec.get("repo") or "") - except OSError as e: - return refusal_order(str(e)) - got = choose_issues(env, rec, rank, n, descriptions, set((lap or {}).get("created") or [])) - if got["milestone"] is None: - if lap is not None: - rec["laps"].remove(lap) # スプリントを組む前の周回は取り下げ、除外の一覧（excluded の lap）を次の選定に残す - if has_held_work(rec): # 選べる課題が無く答えを待つ作業だけが残る: no_issue で止まらず、見直しも挟まない（異常系 6） - return stop_and_report( - env, rec, before, waiting_only(rec, "no_selectable"), "手元の作業が待ちだけになった", {"selection": got["selection"]} - ) - cond = "no_issue" if got["candidates"] == 0 else "outside_mvv" - if cond == "no_issue" and rec.get("review_first") != len(rec["laps"]): - return review_first(env, rec, before, got) - reasons: dict = {} - for r in got["selection"]: - reasons[r["reason"]] = reasons.get(r["reason"], 0) + 1 - values = ( - {"milestones": rec.get("milestones") or (rank.get("metrics") or {}).get("order", [])} - if cond == "no_issue" - else {"reasons": reasons} - ) +def _stop_without_issue(env: Env, rec: dict, before: dict, got: dict, rank: dict, lap: dict | None) -> dict: + """選べる課題が無いときの止まり方（待ちだけ・見直しを先に・no_issue・outside_mvv）。""" + if lap is not None: + rec["laps"].remove(lap) # スプリントを組む前の周回は取り下げ、除外の一覧（excluded の lap）を次の選定に残す + if has_held_work(rec): # 選べる課題が無く答えを待つ作業だけが残る: no_issue で止まらず、見直しも挟まない（異常系 6） return stop_and_report( - env, - rec, - before, - {"condition": cond, "values": values}, - "着手できる課題が無い" if cond == "no_issue" else "候補がすべて MVV の外・人の判断待ちとして外れた", - {"selection": got["selection"]}, + env, rec, before, waiting_only(rec, "no_selectable"), "手元の作業が待ちだけになった", {"selection": got["selection"]} ) + cond = "no_issue" if got["candidates"] == 0 else "outside_mvv" + if cond == "no_issue" and rec.get("review_first") != len(rec["laps"]): + return review_first(env, rec, before, got) + reasons: dict = {} + for r in got["selection"]: + reasons[r["reason"]] = reasons.get(r["reason"], 0) + 1 + values = ( + {"milestones": rec.get("milestones") or (rank.get("metrics") or {}).get("order", [])} + if cond == "no_issue" + else {"reasons": reasons} + ) + return stop_and_report( + env, + rec, + before, + {"condition": cond, "values": values}, + "着手できる課題が無い" if cond == "no_issue" else "候補がすべて MVV の外・人の判断待ちとして外れた", + {"selection": got["selection"]}, + ) + + +def _open_lap(env: Env, rec: dict, before: dict, got: dict, rank: dict, n: int, lap: dict | None) -> dict: + """周回 n の行を作って（または取り直した周回を書き換えて）記録する。選んだ課題の直しを済みにし、スプリントの MVV を置く。""" d = loop_dir(env, rec["name"]) / f"lap-{n}" new = lap or {"n": n, "outcome": IN_PROGRESS, "started_at": env.now(), "created": [], "gates": [], "deferred_issues": []} new.update( @@ -300,6 +295,22 @@ def run_selection(env: Env, rec: dict, before: dict, n: int, lap: dict | None = d.mkdir(parents=True, exist_ok=True) (d / "mvv.md").write_text(got["sprint_mvv"], encoding="utf-8") write_record(env, rec, before) + return new + + +def run_selection(env: Env, rec: dict, before: dict, n: int, lap: dict | None = None) -> dict: + add_env_exclusions(rec, n) + code, rank = env.rank(env.main_dir()) + if code not in (0, PAUSE_RANK): + return refusal_order(f"upkeep.py rank が {code} を返した。選定を推さない: {rank.get('summary', '')}") + try: + descriptions = env.milestones(rec.get("repo") or "") + except OSError as e: + return refusal_order(str(e)) + got = choose_issues(env, rec, rank, n, descriptions, set((lap or {}).get("created") or [])) + if got["milestone"] is None: + return _stop_without_issue(env, rec, before, got, rank, lap) + new = _open_lap(env, rec, before, got, rank, n, lap) chosen = selected_issues(new) nxt = f"モードと版を決め、対象外は loop exclude で外し、python3 {shlex.quote(str(SPRINT_STATE))} loop sprint {rec['name']} --mode <モード> --dev <開発版> を打つ" update_handoff(env, rec, f"周回 {n} の課題を選んだ: {' '.join(f'#{x}' for x in chosen)}", resume_command(rec), nxt) diff --git a/plugins/ndf/scripts/tests/test_context_policy_hook.py b/plugins/ndf/scripts/tests/test_context_policy_hook.py index 6becaf95b..4cf4fba54 100644 --- a/plugins/ndf/scripts/tests/test_context_policy_hook.py +++ b/plugins/ndf/scripts/tests/test_context_policy_hook.py @@ -15,6 +15,7 @@ SCRIPTS = Path(__file__).resolve().parents[1] sys.path.insert(0, str(SCRIPTS / "lib")) +import context_calibrate # noqa: E402 import context_decisions as cd # noqa: E402 CLI = SCRIPTS / "context-policy.py" @@ -203,6 +204,45 @@ def test_record_marks_and_restart(tmp_path, ctx): assert (restart["verdict"], restart["reason"]) == ("再起動", "独立性") +def test_record_row_columns_and_stdout(tmp_path, ctx): + # CLI の記録の行の列（現状固定。#2422）。標準出力の行と記録の行は同じ + out = cli( + "record", + "--verdict", + "再起動", + "--reason", + "独立性", + "--work", + "#7", + "--context", + "1234", + ctx=ctx, + cwd=str(tmp_path), + CLAUDE_CODE_SESSION_ID="s9", + ) + (row,) = rows(ctx) + assert json.loads(out.stdout)["items"] == [row] + picked = { + k: row[k] for k in ("runtime", "source", "session_id", "work", "context", "stage", "verdict", "reason", "band", "event", "policy") + } + assert picked == { + "runtime": "claude", + "source": "record", + "session_id": "s9", + "work": "#7", + "context": 1234, + "stage": "cut", + "verdict": "再起動", + "reason": "独立性", + "band": None, + "event": "判定", + "policy": "current", + } + cli("record", "--event", "復元完了", "--work", "#7", "--runtime", "codex", ctx=ctx, cwd=str(tmp_path), CLAUDE_CODE_SESSION_ID="s9") + mark = rows(ctx)[-1] + assert (mark["runtime"], mark["session_id"], mark["event"], mark["stage"], mark["reason"]) == ("codex", None, "復元完了", None, None) + + def call(ts: str, mid: str, out: int = 10, read: int = 1_000, model: str = "claude-opus-5-5") -> dict: usage = {"input_tokens": 100, "output_tokens": out, "cache_read_input_tokens": read, "cache_creation_input_tokens": 0} return {"type": "assistant", "timestamp": ts, "message": {"id": mid, "model": model, "usage": usage, "content": []}} @@ -333,10 +373,10 @@ def test_calibrate_needs_three_samples(tmp_path): mark("復元完了", f"2026-11-0{i + 1}T00:00:30Z", f"s{i}"), mark("引継ぎ開始", f"2026-11-0{i + 1}T00:02:30Z", f"s{i}"), ] - body = cd.calibrate(recs) + body = context_calibrate.calibrate(recs) assert body["samples"]["restore"] == 3 and body["restore_calls"] == 1.0 assert body["handoff_calls"] == 1.0 and body["handoff_output"] == 50.0 - assert cd.calibrate(recs[:8])["restore_context"] is None + assert context_calibrate.calibrate(recs[:8])["restore_context"] is None def test_sprint_state_next_records_restart(tmp_path, ctx): @@ -356,7 +396,7 @@ def test_sprint_state_next_records_restart(tmp_path, ctx): def _calibrated(recs, since=None) -> dict: - body = cd.calibrate(recs, since=since) + body = context_calibrate.calibrate(recs, since=since) assert isinstance(body.pop("at"), str) return body @@ -430,3 +470,105 @@ def test_calibrate_characterizes_since_and_unmatched_sessions(tmp_path): } since = cd.parse_ts("2026-11-02T00:00:00Z") assert _calibrated(recs, since=since)["samples"] == {"restore": 0, "handoff": 0, "stage:plan": 2, "stage:*": 2} + + +def _cached(ts: str, mid: str, **over) -> dict: + """キャッシュの書き込みを 5 分と 1 時間に分けた usage を持つ呼び出し。""" + c = call(ts, mid) + c["message"]["usage"].update( + {"cache_creation_input_tokens": 70, "cache_creation": {"ephemeral_5m_input_tokens": 30, "ephemeral_1h_input_tokens": 40}}, **over + ) + return c + + +def test_transcript_and_ledger_build_the_same_call(tmp_path): + # 会話の記録と使用量の帳簿は、同じ usage から同じキーと値の呼び出しを作る（現状固定。#2422） + m = load_cli() + a = _cached("2026-11-01T00:00:00Z", "m1") + calls, compacts, by_next = cd.read_calls(write_transcript(tmp_path / "s1.jsonl", [a])) + want = { + "ts": cd.parse_ts("2026-11-01T00:00:00Z"), + "model": "claude-opus-5-5", + "inp": 100, + "out": 10, + "read": 1000, + "w5": 30, + "w1h": 40, + "context": 1170, + } + assert (calls, compacts, by_next) == ([want], 0, False) + row = {"at": "2026-11-01T00:00:00Z", "model": "claude-opus-5-5", "usage": a["message"]["usage"], "seconds": 4} + assert m.ledger_calls(row) == want | {"seconds": 4} + assert m.ledger_calls({**row, "model": "", "seconds": "x"}) == want | {"model": None, "seconds": None} + + +def test_transcript_counts_auto_compacts_and_the_handoff_start(tmp_path): + lines = [ + {"type": "user", "isMeta": True, "message": {"content": "meta"}}, + {"type": "user", "message": {"content": [{"type": "text", "text": "cat .ndf/handoff/x.md"}]}}, + {"type": "system", "subtype": "compact_boundary", "compactMetadata": {"trigger": "auto"}}, + {"type": "system", "subtype": "compact_boundary", "compactMetadata": {"trigger": "manual"}}, + _cached("2026-11-01T00:00:00Z", "m1"), + ] + calls, compacts, by_next = cd.read_calls(write_transcript(tmp_path / "s1.jsonl", lines)) + assert (len(calls), compacts, by_next) == (1, 1, True) + assert cd.read_calls(tmp_path / "missing.jsonl") == ([], None, False) + + +def test_work_report_pins_every_part(tmp_path): + # 作業の報告の区分ごとの token・秒・費用・再起動・やり直しの数（現状固定。#2422） + m = load_cli() + first = {"type": "user", "message": {"content": "続き .ndf/handoff/a.md"}} + tp = write_transcript( + tmp_path / "s1.jsonl", + [ + first, + _cached("2026-11-01T00:00:00Z", "m1"), + call("2026-11-01T00:10:00Z", "m2"), + call("2026-11-01T00:20:00Z", "m3"), + call("2026-11-01T01:30:00Z", "m4"), + ], + ) + write_transcript( + tmp_path / "s1" / "subagents" / "agent-x.jsonl", [call("2026-11-01T00:11:00Z", "m5"), call("2026-11-01T00:12:00Z", "m6")] + ) + ledger = [ + { + "session_id": "ag9", + "parent_session_id": "s1", + "at": "2026-11-01T00:15:00Z", + "usage": call("t", "m7")["message"]["usage"], + "seconds": 12, + } + ] + recs = [ + hook_row("s1", tp, "2026-11-01T00:00:00Z", verdict="継続", stage="skill:a"), + mark("復元完了", "2026-11-01T00:05:00Z"), + hook_row("s1", tp, "2026-11-01T00:06:00Z", verdict="継続", stage="skill:a"), + mark("引継ぎ開始", "2026-11-01T01:00:00Z", sid="s1"), + hook_row("s1", tp, "2026-11-01T01:40:00Z", verdict="再起動", reason="容量", stage="cut"), + hook_row("s2", None, "2026-11-01T02:00:00Z", verdict="再起動"), + ] + out = m.report_work("#7", recs, ledger) + parts = {b: (t["calls"], t["inp"], t["out"], t["read"], t["w5"], t["w1h"], t["seconds"], t["cost"]) for b, t in out["parts"].items()} + assert parts == PINNED_PARTS + rest = {k: v for k, v in out.items() if k != "parts"} + assert rest == PINNED_REST + + +PINNED_PARTS = { + "conductor": (2, 200, 20, 2000, 0, 0, 600, 400), + "ndf": (3, 300, 30, 3000, 0, 0, 72, 650), + "handoff": (1, 100, 10, 1000, 0, 0, 0, 200), + "restore": (1, 100, 10, 1000, 30, 40, 0, 318), +} +PINNED_REST = { + "work": "#7", + "sessions": 2, + "unknown_parts": {"restore": 0, "handoff": 1}, + "restarts": {"容量": 1, "不明": 1}, + "restart_count": 2, + "auto_compacts": "不明", + "quality": {"rework": 1, "review_fixes": "不明", "missed_requirements": "不明"}, + "no_policy_rows": 0, +} diff --git a/plugins/ndf/scripts/tests/test_lib_clock.py b/plugins/ndf/scripts/tests/test_lib_clock.py index 5c225e6e2..2b6bc5500 100644 --- a/plugins/ndf/scripts/tests/test_lib_clock.py +++ b/plugins/ndf/scripts/tests/test_lib_clock.py @@ -76,3 +76,13 @@ def test_parse_accepts_datetimes(): def test_seconds_between(): assert clock.seconds_between("2026-09-26T07:00:00Z", "2026-09-26T16:01:00+09:00") == 60 assert clock.seconds_between("x", "2026-09-26T07:00:00Z") is None + + +def test_epoch_seconds_reads_the_same_instant_as_parse(): + assert clock.epoch_seconds("2026-09-26T07:00:00Z") == T.replace(microsecond=0).timestamp() + assert clock.epoch_seconds("2026-09-26T16:00:00+09:00") == clock.epoch_seconds("2026-09-26T07:00:00+00:00") + + +@pytest.mark.parametrize("value", [None, "", "not a time", 12]) +def test_epoch_seconds_returns_none_for_unreadable(value): + assert clock.epoch_seconds(value) is None diff --git a/plugins/ndf/scripts/tests/test_sprint_loop.py b/plugins/ndf/scripts/tests/test_sprint_loop.py index 7b2ad986f..a0c12237f 100644 --- a/plugins/ndf/scripts/tests/test_sprint_loop.py +++ b/plugins/ndf/scripts/tests/test_sprint_loop.py @@ -1548,3 +1548,16 @@ def test_gate_1_of_a_lane_sends_the_request_to_the_organizer(env, monkeypatch): out = sl_gate.cmd_gate(env, "l1", write_done(env, [item])) w = rec(env)["waiting"][0] assert out["metrics"]["waiting"] == [11] and sent == [11] and w["request"] == "lroot-g501-x" and w["plan"] == "p1" + + +def test_selecting_a_returned_issue_marks_its_fix(env): + # 直しの周回: 差し戻された周回の課題を選ぶと、その直しを済みにする。選ばなかった直しと済みの直しは変えない（現状固定。#2422） + start(env) + with sl_record.open_record(env, "l1") as (r, before): + r["fixes"] = [{"lap": 0, "issues": [12]}, {"lap": 0, "issues": [99]}, {"lap": 0, "issues": [11], "fixed_by": 7}] + sl_record.write_record(env, r, before) + got = sl_select.cmd_select(env, "l1") + lap = rec(env)["laps"][-1] + assert got["metrics"]["selected"] == [12, 11] + assert (lap["sprint"], lap["milestone"], lap["rank_digest"], lap["state_path"]) == (f"l1-{lap['n']}", "01 近い", "dg", None) + assert [f.get("fixed_by") for f in rec(env)["fixes"]] == [lap["n"], None, 7] diff --git a/plugins/ndf/skills/backlog-refinement/scripts/upkeep_depth.py b/plugins/ndf/skills/backlog-refinement/scripts/upkeep_depth.py index 61e1f5f7c..077b2cf0d 100644 --- a/plugins/ndf/skills/backlog-refinement/scripts/upkeep_depth.py +++ b/plugins/ndf/skills/backlog-refinement/scripts/upkeep_depth.py @@ -102,14 +102,23 @@ def append_depth_row(path: Path, row: dict) -> None: locks.append_locked(path, json.dumps(row, ensure_ascii=False)) +def _is_ran(row: dict, depth: str) -> bool: + """`depth` を実行した（`event` が `ran` の）行か。""" + return row.get("depth") == depth and row.get("event") == "ran" + + +def _last_ran(rows: list[dict], depth: str) -> int: + """最後に `depth` を実行した行の位置。無ければ -1。""" + return max((i for i, r in enumerate(rows) if _is_ran(r, depth)), default=-1) + + def _ran(rows: list[dict], depth: str) -> list[dict]: - return [r for r in rows if r.get("depth") == depth and r.get("event") == "ran"] + return [r for r in rows if _is_ran(r, depth)] def diffs_since(rows: list[dict], depth: str) -> int: """最後の `depth` の `ran` の行より後の差分の棚卸しの `ran` の行の数（前提 1 の数え方）。""" - last = max((i for i, r in enumerate(rows) if r.get("depth") == depth and r.get("event") == "ran"), default=-1) - return sum(1 for r in rows[last + 1 :] if r.get("depth") == "diff" and r.get("event") == "ran") + return len(_ran(rows[_last_ran(rows, depth) + 1 :], "diff")) def cause_share(causes: list[dict], types: list[dict], open_numbers: set | None) -> tuple[str | None, float]: @@ -156,8 +165,7 @@ def _due_overhaul(rows: list[dict], cfg: dict, causes, types, open_numbers) -> t values["count"] = diffs_since(rows, "overhaul") if values["count"] >= c["every"]: trig.append("count") - last_ov = max((i for i, r in enumerate(rows) if r.get("depth") == "overhaul" and r.get("event") == "ran"), default=-1) - reviews = _ran(rows[last_ov + 1 :], "review")[-c["streak"] :] + reviews = _ran(rows[_last_ran(rows, "overhaul") + 1 :], "review")[-c["streak"] :] if len(reviews) >= c["streak"]: if all((r.get("created") or 0) > (r.get("closed") or 0) for r in reviews): trig.append("created_over_closed") diff --git a/plugins/ndf/skills/backlog-refinement/scripts/upkeep_overhaul.py b/plugins/ndf/skills/backlog-refinement/scripts/upkeep_overhaul.py index a71168cf0..0bc0f844a 100644 --- a/plugins/ndf/skills/backlog-refinement/scripts/upkeep_overhaul.py +++ b/plugins/ndf/skills/backlog-refinement/scripts/upkeep_overhaul.py @@ -62,42 +62,57 @@ def _read_input(path, what: str): return got +def _issue_types(issues: list[dict], by_num: dict, ids: list[str]) -> dict[int, str]: + """課題の番号 → 型。分類が無いか型の一覧に無ければ `I`（無ければ最後の型）。""" + other = "I" if "I" in ids else ids[-1] + kind = {} + for i in issues: + t = (by_num.get(i["number"]) or {}).get("type") + kind[i["number"]] = t if t in ids else other + return kind + + +def _components(issues: list[dict]) -> Counter: + """題と本文に現れる部品の名前ごとの、課題の数（1 つの課題で 1 回）。""" + comps = Counter() + for i in issues: + comps.update(set(_COMPONENT.findall(f"{i.get('title')}\n{i.get('body')}"))) + return comps + + +def _months(issues: list[dict], closed: list[dict]) -> dict: + """月（`YYYY-MM`）→ {起票の数, 完了の数}。月の順。""" + months: dict = {} + for i in list(issues) + list(closed): + if i.get("created_at"): + months.setdefault(i["created_at"][:7], {"created": 0, "closed": 0})["created"] += 1 + for c in closed: + if c.get("closed_at"): + months.setdefault(c["closed_at"][:7], {"created": 0, "closed": 0})["closed"] += 1 + return dict(sorted(months.items())) + + def count_types(snap: dict, classes: list[dict], types: list[dict]) -> dict: """型ごとの件数・部品・マイルストーンごとの割合・月ごとの起票と完了。""" ids = [t["id"] for t in types] - other = "I" if "I" in ids else ids[-1] by_num = {c["number"]: c for c in classes if isinstance(c, dict) and isinstance(c.get("number"), int)} issues = snap.get("open") or [] - kind = { - i["number"]: (by_num.get(i["number"]) or {}).get("type") if (by_num.get(i["number"]) or {}).get("type") in ids else other - for i in issues - } + kind = _issue_types(issues, by_num, ids) by_type = {t: sum(1 for v in kind.values() if v == t) for t in ids} defect = {t["id"] for t in types if t.get("defect", True)} n_defect = sum(by_type[t] for t in defect) - comps = Counter() - for i in issues: - comps.update(set(_COMPONENT.findall(f"{i.get('title')}\n{i.get('body')}"))) per_ms: dict = {} for i in issues: - row = per_ms.setdefault(i.get("milestone") or "（なし）", Counter()) - row[kind[i["number"]]] += 1 - months: dict = {} - for i in list(issues) + list(snap.get("closed") or []): - if i.get("created_at"): - months.setdefault(i["created_at"][:7], {"created": 0, "closed": 0})["created"] += 1 - for c in snap.get("closed") or []: - if c.get("closed_at"): - months.setdefault(c["closed_at"][:7], {"created": 0, "closed": 0})["closed"] += 1 + per_ms.setdefault(i.get("milestone") or "（なし）", Counter())[kind[i["number"]]] += 1 return { "issues": len(issues), "unclassified": sum(1 for i in issues if i["number"] not in by_num), "by_type": by_type, "defects": n_defect, "defect_share": {t: round(by_type[t] / n_defect, 4) for t in ids if t in defect and n_defect}, - "components": comps.most_common(10), + "components": _components(issues).most_common(10), "milestones": {m: dict(c) for m, c in sorted(per_ms.items())}, - "months": dict(sorted(months.items())), + "months": _months(issues, snap.get("closed") or []), } diff --git a/plugins/ndf/skills/backlog-refinement/scripts/upkeep_research.py b/plugins/ndf/skills/backlog-refinement/scripts/upkeep_research.py index 02ddbe773..14a757910 100644 --- a/plugins/ndf/skills/backlog-refinement/scripts/upkeep_research.py +++ b/plugins/ndf/skills/backlog-refinement/scripts/upkeep_research.py @@ -15,7 +15,6 @@ from __future__ import annotations -import datetime as _dt import hashlib import json import os @@ -101,25 +100,37 @@ def read_research_item(root, item_id: str) -> dict | None: return rec if rec is not None and rec.get("id", item_id) == item_id else None +def _usage_rows(needs: dict) -> tuple[list[dict], list[dict]]: + """needs.usage の行を (使える行, 飛ばした側面) に分ける。集計だけを返す行に限って読む。""" + usage, skipped = [], [] + for row in needs.get("usage") or []: + scope = row.get("scope") or {} + if scope.get("aggregate_only") is True and row.get("command"): + usage.append(row) + else: + skipped.append( + {"aspect": "needs.usage", "reason": f"{row.get('name')}: aggregate_only が true でない行は読まない（個票を読まない）"} + ) + return usage, skipped + + +def _undeclared(items: list[dict], skipped: list[dict]) -> list[dict]: + """項目にも飛ばした側面にも現れない側面を、宣言の無い側面として返す。""" + have = {i["aspect"] for i in items} | {s["aspect"] for s in skipped} + return [{"aspect": a, "reason": UNDECLARED} for a in ASPECTS if a not in have] + + def research_items(mat: dict | None) -> tuple[list[dict], list[dict]]: """(調査の項目, 飛ばした側面)。項目は {id, aspect, target, decl}。""" if mat is None: return [], [{"aspect": a, "reason": UNDECLARED} for a in ASPECTS] - items, skipped = [], [] + items = [] needs, seeds = mat.get("needs") or {}, mat.get("seeds") or {} def add(aspect, target, decl): items.append({"id": f"{aspect}:{target}", "aspect": aspect, "target": target, "decl": decl}) - usage = [] - for row in needs.get("usage") or []: - scope = row.get("scope") or {} - if scope.get("aggregate_only") is True and row.get("command"): - usage.append(row) - else: - skipped.append( - {"aspect": "needs.usage", "reason": f"{row.get('name')}: aggregate_only が true でない行は読まない（個票を読まない）"} - ) + usage, skipped = _usage_rows(needs) for row in usage: add("needs.usage", row.get("name") or row["command"], row) for row in needs.get("market") or []: @@ -133,29 +144,17 @@ def add(aspect, target, decl): add("method.unused", "usage", {"usage": [u.get("name") for u in usage]}) add("method.hotspot", "git", {}) add("method.failures", "loop", {"failures": mat.get("failures")}) - have = {i["aspect"] for i in items} | {s["aspect"] for s in skipped} - for a in ASPECTS: - if a not in have: - skipped.append({"aspect": a, "reason": UNDECLARED}) + skipped.extend(_undeclared(items, skipped)) return items, skipped -def _iso_epoch(value) -> float | None: - if not isinstance(value, str) or not value: - return None - try: - return _dt.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() - except ValueError: - return None - - def plan(root, since: str | None) -> dict: items, skipped = research_items(D.materials(root)) - cut = _iso_epoch(since) + cut = clock.epoch_seconds(since) todo, fresh = [], [] for it in items: rec = read_research_item(root, it["id"]) - checked = _iso_epoch((rec or {}).get("checked_at")) + checked = clock.epoch_seconds((rec or {}).get("checked_at")) if checked is None or (cut is not None and checked < cut): todo.append(it["id"]) else: @@ -298,9 +297,9 @@ def research_one(root, it: dict, cfg: dict, product, run: str, loop, lap, tick) extra = "# 変更の多いファイル（回数・パス）\n" + hotspots(root) if it["aspect"] == "method.hotspot" else extra got = call_llm(root, prompt_of(it, product, extra), int(lim["item_seconds"]), tick, tools_of(it["aspect"])) usage = {"cost": got["cost"], "seconds": got["seconds"]} - ok = got["ok"] and got["cost"] <= lim["item_usd"] and got["seconds"] <= lim["item_seconds"] and valid_record(got["data"]) - log_call(root, it["id"], ok, usage, run, loop, lap) - if not got["ok"] or got["cost"] > lim["item_usd"] or got["seconds"] > lim["item_seconds"] or not valid_record(got["data"]): + accepted = got["ok"] and got["cost"] <= lim["item_usd"] and got["seconds"] <= lim["item_seconds"] and valid_record(got["data"]) + log_call(root, it["id"], accepted, usage, run, loop, lap) + if not accepted: return None, usage rec = { "id": it["id"], @@ -318,6 +317,35 @@ def research_one(root, it: dict, cfg: dict, product, run: str, loop, lap, tick) return rec, usage +def _ticker(root, run_id, until: str | None, cfg: dict): + """呼び出しの間に呼ぶ関数。持ち主の印を `owner.beat` 秒ごとに書き、待っていたプランの完了から `research.grace` 秒を過ぎたら GraceOver を投げる。""" + beat = {"at": time.time(), "until_seen": None} + + def tick(*_): + if time.time() - beat["at"] >= cfg["owner"]["beat"]: + O.touch(root, run_id) + beat["at"] = time.time() + # 調査中も待っていたプランの完了を見て、research.grace 秒を過ぎた呼び出しを止める（持ち主を手放し、後の見直しへ渡す） + if until and Path(until).exists(): + seen = beat["until_seen"] = beat["until_seen"] or time.time() + if time.time() - seen >= float(cfg["research"].get("grace", 300)): + raise GraceOver + + return tick + + +def _wait_turn(root, run_id, item_id: str, until: str | None, listing: str | None, max_: int, tick, sleep) -> str | None: + """並列の空きを待ち、次の項目を始めてよければ None、止めるなら止まる理由を返す。""" + while not (until and Path(until).exists()) and not free_slot(listing, max_): + tick() + sleep(POLL) + if until and Path(until).exists(): # 待っていたプランが終わったら新しい項目を始めない（I13） + return "待っていたプランが終わった" + if O.touch(root, run_id, item_id): + return "持ち主でなくなった" + return None + + def run_research( root, until: str | None, @@ -345,31 +373,14 @@ def run_research( pl = plan(root, since) mat = D.materials(root) or {} done, cost, secs, fails, why = [], 0.0, 0.0, 0, "項目が無い" - beat = {"at": time.time(), "until_seen": None} - - def tick(*_): - if time.time() - beat["at"] >= cfg["owner"]["beat"]: - O.touch(root, run_id) - beat["at"] = time.time() - # 調査中も待っていたプランの完了を見て、research.grace 秒を過ぎた呼び出しを止める（持ち主を手放し、後の見直しへ渡す） - if until and Path(until).exists(): - seen = beat["until_seen"] = beat["until_seen"] or time.time() - if time.time() - seen >= float(cfg["research"].get("grace", 300)): - raise GraceOver - + tick = _ticker(root, run_id, until, cfg) try: for item_id in pl["todo"]: if len(done) >= max_items: why = f"1 回の上限 {max_items} 項目に当たった" break - while not (until and Path(until).exists()) and not free_slot(listing, max_): - tick() - sleep(POLL) - if until and Path(until).exists(): # 待っていたプランが終わったら新しい項目を始めない（I13） - why = "待っていたプランが終わった" - break - if O.touch(root, run_id, item_id): - why = "持ち主でなくなった" + if stop := _wait_turn(root, run_id, item_id, until, listing, max_, tick, sleep): + why = stop break try: rec, usage = research_one(root, pl["by_id"][item_id], cfg, mat.get("product"), run_id, loop, lap, tick) diff --git a/plugins/ndf/skills/skill-stats/scripts/skill-stats.py b/plugins/ndf/skills/skill-stats/scripts/skill-stats.py index 8ab030628..fd8b5c282 100755 --- a/plugins/ndf/skills/skill-stats/scripts/skill-stats.py +++ b/plugins/ndf/skills/skill-stats/scripts/skill-stats.py @@ -27,7 +27,7 @@ import sys from collections import Counter, defaultdict from datetime import datetime, timedelta -from typing import Iterable +from typing import Iterable, NamedTuple # 会話の記録を層の単位で読む部品は共通層にある（#550 の決定 13）。Kiro CLI が Skill を # symlink にするため、`.resolve()` を通してからプラグインルートへ登る（`scripts/lib/README.md`）。 @@ -375,6 +375,16 @@ def _lookahead_hit(tl: list, i: int, qualified: str, cap: int) -> bool: return False +def _count_user_text(tl: list, i: int, text, skill_triggers: list, trig_h: Counter, hits: Counter, cap: int) -> None: + """利用者の発言 1 件（tl の i 番目）に現れたトリガーを Skill ごとに数え、続く呼び出しで当たったらヒットに数える。""" + text_l = str(text).lower() + for qualified, trs in skill_triggers: + if trs and any(t in text_l for t in trs): + trig_h[qualified] += 1 + if _lookahead_hit(tl, i, qualified, cap): + hits[qualified] += 1 + + def aggregate_by_project( transcripts: list[pathlib.Path], skills: list[dict], @@ -398,20 +408,10 @@ def aggregate_by_project( for i, (kind, data) in enumerate(tl): if kind == "skill": auto[data] += 1 - continue - if kind == "slash": + elif kind == "slash": explicit[data] += 1 - continue - if kind != "user": - continue - text_l = str(data).lower() - for qualified, trs in skill_triggers: - if not trs: - continue - if any(t in text_l for t in trs): - trig_h[qualified] += 1 - if _lookahead_hit(tl, i, qualified, lookahead_cap): - hits[qualified] += 1 + elif kind == "user": + _count_user_text(tl, i, data, skill_triggers, trig_h, hits, lookahead_cap) return result @@ -747,6 +747,32 @@ def filter_projects( return {k: v for k, v in per_project.items() if needle in k.lower()} +class AgentReport(NamedTuple): + """`--agents` の報告（記録の行・要約・層ごとの合計・総計・役割ごとの使用量・要約から除いた数・割る候補の上限と出所）。""" + + agents: list + summary: list[dict] + totals_rows: list[dict] + totals: dict + usage: list[dict] + excluded: int + limit_meta: dict + + +def agent_report(args: argparse.Namespace) -> AgentReport: + """`--agents` のときだけ記録を読んで報告を作る。割る候補の上限と出所は `limit_meta` に返す。""" + if not args.agents: + return AgentReport([], [], [], {}, [], 0, {}) + agents, skipped = collect_agents(args.session, args.layer) + if skipped: + print(f"[skill-stats] 読めない行を飛ばした: {skipped} 件", file=sys.stderr) + limit_of, limit_meta = context_decisions.record_limits([r.session for r in agents], args.window_limit) + summary, excluded = summarize_agents(agents, limit_of) + totals_rows, totals = layer_totals(agents) + usage = role_usage(agents) if args.session else [] + return AgentReport(agents, summary, totals_rows, totals, usage, excluded, limit_meta) + + def emit_json( skills: list[dict], per_project: dict[str, tuple[Counter, Counter, Counter, Counter]], @@ -754,12 +780,7 @@ def emit_json( effective_days: int | None, plugin_root: pathlib.Path, transcripts: list[pathlib.Path], - agents: list, - agent_summary: list[dict], - totals_rows: list[dict], - totals: dict, - usage: list[dict], - excluded: int, + report: AgentReport, ) -> None: projects_json = [] for project, (auto, explicit, trig, hits) in sorted(per_project.items()): @@ -787,16 +808,16 @@ def emit_json( "projects": projects_json, } if args.agents: - out["meta"].update(args.limit_meta) # 割る候補の上限と出所（main が記録から決めた値） + out["meta"].update(report.limit_meta) # 割る候補の上限と出所（agent_report が記録から決めた値） out.update( { - "agents": [r.as_row() for r in agents], - "agent_summary": agent_summary, - "layer_totals": totals_rows, - "totals": totals, - "role_usage": usage, - "excluded": excluded, - "unphased_supervisors": transcript_agents.unphased_supervisors(agents), + "agents": [r.as_row() for r in report.agents], + "agent_summary": report.summary, + "layer_totals": report.totals_rows, + "totals": report.totals, + "role_usage": report.usage, + "excluded": report.excluded, + "unphased_supervisors": transcript_agents.unphased_supervisors(report.agents), } ) print(json.dumps(out, ensure_ascii=False, indent=2)) @@ -806,12 +827,7 @@ def emit_markdown( skills: list[dict], per_project: dict[str, tuple[Counter, Counter, Counter, Counter]], args: argparse.Namespace, - agents: list, - agent_summary: list[dict], - totals_rows: list[dict], - totals: dict, - usage: list[dict], - excluded: int, + report: AgentReport, ) -> None: if args.by_project: for project, (auto, explicit, trig, hits) in sorted(per_project.items()): @@ -838,12 +854,12 @@ def emit_markdown( print() print( format_agents_markdown( - agents, - agent_summary, - excluded, - totals_rows, - totals, - usage, + report.agents, + report.summary, + report.excluded, + report.totals_rows, + report.totals, + report.usage, with_session=bool(args.session), ) ) @@ -895,20 +911,7 @@ def main() -> int: transcripts = select_transcripts(args, date_from, date_to, effective_days) - agents: list = [] - agent_summary: list[dict] = [] - totals_rows: list[dict] = [] - totals: dict = {} - usage: list[dict] = [] - excluded = 0 - if args.agents: - agents, skipped = collect_agents(args.session, args.layer) - if skipped: - print(f"[skill-stats] 読めない行を飛ばした: {skipped} 件", file=sys.stderr) - limit_of, args.limit_meta = context_decisions.record_limits([r.session for r in agents], args.window_limit) - agent_summary, excluded = summarize_agents(agents, limit_of) - totals_rows, totals = layer_totals(agents) - usage = role_usage(agents) if args.session else [] + report = agent_report(args) # Header summary window = [] @@ -931,32 +934,9 @@ def main() -> int: return 0 if args.format == "json": - emit_json( - skills, - per_project, - args, - effective_days, - plugin_root, - transcripts, - agents, - agent_summary, - totals_rows, - totals, - usage, - excluded, - ) + emit_json(skills, per_project, args, effective_days, plugin_root, transcripts, report) else: - emit_markdown( - skills, - per_project, - args, - agents, - agent_summary, - totals_rows, - totals, - usage, - excluded, - ) + emit_markdown(skills, per_project, args, report) return 0 diff --git a/plugins/ndf/skills/skill-stats/tests/test_agents_report.py b/plugins/ndf/skills/skill-stats/tests/test_agents_report.py index f008ea937..14f7135eb 100644 --- a/plugins/ndf/skills/skill-stats/tests/test_agents_report.py +++ b/plugins/ndf/skills/skill-stats/tests/test_agents_report.py @@ -279,3 +279,21 @@ def test_a_session_narrows_the_skill_statistics_to_the_conductor() -> None: assert p.returncode == 0, p.stderr out = json.loads(p.stdout) assert out["meta"]["transcripts"] == 1 + + +def test_the_json_keys_keep_their_order(report) -> None: + # 結果 JSON のキーの順（現状固定。#2422）。エージェントの報告をまとめて渡しても順を変えない + assert list(report) == [ + "meta", + "total", + "grand_skills", + "projects", + "agents", + "agent_summary", + "layer_totals", + "totals", + "role_usage", + "excluded", + "unphased_supervisors", + ] + assert list(report["meta"])[-3:] == ["window_limit", "window_limit_source", "records_without_decisions"] diff --git a/plugins/ndf/skills/skill-stats/tests/test_skill_stats_wrappers.py b/plugins/ndf/skills/skill-stats/tests/test_skill_stats_wrappers.py index 8bc02822c..4a7f4a54a 100644 --- a/plugins/ndf/skills/skill-stats/tests/test_skill_stats_wrappers.py +++ b/plugins/ndf/skills/skill-stats/tests/test_skill_stats_wrappers.py @@ -82,3 +82,31 @@ def test_the_skill_table_uses_spaced_rules_and_an_empty_total_cell(ss): "| ndf:fix | explicit | 11 | 3 | 8 | 8 | 3 | 37.5% |", "| **合計** | | **23** | **5** | **18** | **8** | **3** | **37.5%** |", ] + + +def test_aggregate_counts_auto_explicit_triggers_and_hits(ss, monkeypatch): + """発言 1 件ごとのトリガーとヒットの数（現状固定。#2422）。ヒットは次の発言・スラッシュより前に同じ Skill が呼ばれたとき。""" + tl = [ + ("user", "Please REVIEW this"), + ("skill", "ndf:review"), + ("user", "review again and plan"), + ("slash", "ndf:review"), + ("skill", "ndf:review"), + ("user", "nothing here"), + ("skill", "ndf:plan"), + ("user", "plan it"), + ("assistant", "x"), + ("skill", "ndf:plan"), + ] + monkeypatch.setattr(ss, "build_timeline", lambda path, names: (tl, "proj")) + skills = [ + {"name": "review", "qualified": "ndf:review", "triggers": ["review", ""]}, + {"name": "plan", "qualified": "ndf:plan", "triggers": ["plan"]}, + {"name": "quiet", "qualified": "ndf:quiet", "triggers": []}, + ] + got = ss.aggregate_by_project([pathlib.Path("a.jsonl"), pathlib.Path("b.jsonl")], skills) + auto, explicit, trig, hits = got["proj"] + assert (dict(auto), dict(explicit)) == ({"ndf:review": 4, "ndf:plan": 4}, {"ndf:review": 2}) + assert (dict(trig), dict(hits)) == ({"ndf:review": 4, "ndf:plan": 4}, {"ndf:review": 2, "ndf:plan": 2}) + capped = ss.aggregate_by_project([pathlib.Path("a.jsonl")], skills, lookahead_cap=1)["proj"] + assert dict(capped[3]) == {"ndf:review": 1} diff --git a/scripts/token-usage.py b/scripts/token-usage.py index 931d36b58..54d307420 100755 --- a/scripts/token-usage.py +++ b/scripts/token-usage.py @@ -469,47 +469,60 @@ class External: usage: Usage | None = None # 呼び出しの並びが分かる席だけ（claude・codex）。P・k・書き直しに使う +def _codex_lines(path: Path, until: float | None) -> tuple[str, str | None, dict | None, list[float], list[tuple]]: + """codex の記録 1 本の (作業ディレクトリ, モデル, 最後の累計, 時刻の並び, 呼び出しごとの (時刻, 入力, キャッシュ, 出力))。 + + `until` より後の行は読まない。同じ累計の token_count は同じ呼び出しの再掲として数えない。""" + cwd, model, total, times, calls = "", None, None, [], [] + with open(path, encoding="utf-8", errors="replace") as fh: + lines = fh.readlines() + for line in lines: + try: + d = json.loads(line) + except ValueError: + continue + if not isinstance(d, dict): + continue + t = parse_ts(d.get("timestamp")) + if until is not None and t is not None and t > until: + continue + if t is not None: + times.append(t) + p = d.get("payload") or {} + if d.get("type") == "session_meta": + cwd = p.get("cwd") or "" + elif d.get("type") == "turn_context" and not model: + model = p.get("model") + elif p.get("type") == "token_count" and p.get("info"): + new_total = p["info"].get("total_token_usage") or total + last = p["info"].get("last_token_usage") or {} + if last and new_total != total: # 同じ累計の token_count は同じ呼び出しの再掲 + inp, cached = last.get("input_tokens", 0), last.get("cached_input_tokens", 0) + calls.append((t, inp, cached, last.get("output_tokens", 0))) + total = new_total + return cwd, model, total, times, calls + + +def _codex_usage(calls: list[tuple]) -> Usage | None: + """呼び出しごとの値から使用量を組む。codex の input_tokens は cached を含むため、キャッシュに当たらなかった分を書き込みとみなす。 + + 呼び出しごとの値（last_token_usage）が無い記録は、P・k を取れないものとして None を返す。""" + if not calls: + return None + usage = Usage() + for _, inp, cached, out_t in calls: + usage.add(Usage(1, inp - cached, cached, 0, 0, out_t)) + usage.record_calls([(t, inp, inp - cached, cached) for t, inp, cached, _ in calls]) + return usage + + def read_codex(root: Path, idle_cap: int, until: float | None = None) -> list[External]: out = [] for path in sorted(root.glob("**/rollout-*.jsonl")): - cwd, model, total, times, calls = "", None, None, [], [] - with open(path, encoding="utf-8", errors="replace") as fh: - lines = fh.readlines() - for line in lines: - try: - d = json.loads(line) - except ValueError: - continue - if not isinstance(d, dict): - continue - t = parse_ts(d.get("timestamp")) - if until is not None and t is not None and t > until: - continue - if t is not None: - times.append(t) - p = d.get("payload") or {} - if d.get("type") == "session_meta": - cwd = p.get("cwd") or "" - elif d.get("type") == "turn_context" and not model: - model = p.get("model") - elif p.get("type") == "token_count" and p.get("info"): - new_total = p["info"].get("total_token_usage") or total - last = p["info"].get("last_token_usage") or {} - if last and new_total != total: # 同じ累計の token_count は同じ呼び出しの再掲 - inp, cached = last.get("input_tokens", 0), last.get("cached_input_tokens", 0) - calls.append((t, inp, cached, last.get("output_tokens", 0))) - total = new_total + cwd, model, total, times, calls = _codex_lines(path, until) m = WT_RE.search(cwd + "/") if not m or not times or total is None: # token_count の無い記録（起動に失敗した席）は数えない continue - # codex の input_tokens は cached を含む。キャッシュに当たらなかった分を書き込みとみなす。 - # 呼び出しごとの値（last_token_usage）が無い記録は、P・k を取れないものとして usage を持たない - usage = None - if calls: - usage = Usage() - for _, inp, cached, out_t in calls: - usage.add(Usage(1, inp - cached, cached, 0, 0, out_t)) - usage.record_calls([(t, inp, inp - cached, cached) for t, inp, cached, _ in calls]) out.append( External( "codex", @@ -523,7 +536,7 @@ def read_codex(root: Path, idle_cap: int, until: float | None = None) -> list[Ex "cached": total.get("cached_input_tokens", 0), "out": total.get("output_tokens", 0), }, - usage=usage, + usage=_codex_usage(calls), ) ) return out