Skip to content

perf(CC-573): single streaming jq pass for pmctl run-stats #547

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
screenleon merged 4 commits into from
Aug 27, 2026
Merged

perf(CC-573): single streaming jq pass for pmctl run-stats #547

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fad8f63
docs(BACKLOG): open CC-573 — run-stats per-line jq spawn
screenleon Aug 27, 2026
6eb5e1a
perf(CC-573): single streaming jq pass for pmctl run-stats
screenleon Aug 27, 2026
6ee2748
test(CC-573): golden before/after equivalence for the streaming scan
screenleon Aug 27, 2026
20579f7
docs(BACKLOG): mark CC-573 done
screenleon Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 76 additions & 0 deletions BACKLOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ CC-001/CC-002 were consumed by PR #24 fix bundle inline, with no standalone entr
| CC-570 | 🟢 someday | Fact/Case/Strategy `memory_function`／`memory_subtype` metadata 分類法：先蒐集 [[CC-567]] 的 applied/outcome 證據，再決定值不值得建分類機制——不憑直覺先建立稅務式標籤（2026-08-25 memory 架構設計討論；外部文章優先序建議相反，本 repo 刻意反過來） | memory/DX | 2026-08-25 | — | P3 | retrieval |
| CC-571 | ✅ done | `_ctx_fts_rebuild`／`_ctx_index_file` 共用的 sqlite atomic-script 缺口：DROP+CREATE+INSERT 未加 `-bail`（實測 sqlite3 CLI 預設不會在錯誤時中止，單靠 BEGIN/COMMIT 不足）、呼叫端不檢查回傳值、`_ctx_index_file` 還有第三個獨立 bug（`rm -f` 蓋掉 sqlite3 真實 exit code）；`/simplify` altitude review 抓到手足函式同缺陷，範圍已擴大涵蓋兩者（[[CC-548]] spike 的 Open risks 側面發現，非本票 tokenizer 範圍） | memory/ops | 2026-08-26 | pr:#539 | P2 | hygiene |
| CC-572 | ✅ done | pr-gate synthesis retry（sequential／parallel 兩條路徑）留下已存在但 0 bytes 的 `$OUTPUT_FILE`，executor 的 patch 工具仍可能選擇 Update File 而非 Add File 語意，對空內容找不到 context line 而崩潰（`apply_patch verification failed`）；CC-571 gate saga 連續四輪協定失敗實測發現（2026-08-26） | gate/ops | 2026-08-26 | pr:#541 | P2 | hygiene |
| CC-573 | ✅ done | `pmctl run-stats` 每個事件行 fork 一個 jq（`pmctl_run_stats_extract_line`），與 [[CC-364]] 修掉前的 `trace tail` 同形狀。實測 jq 呼叫 N+2、~34ms/event，真實 6642 行 `events.jsonl` 時 `run-stats --json` 前景 2 分鐘 timeout。改為單次 `jq -R` 串流 over 串接的 archive+active：jq 呼叫 102/302/902 → 2/2/2、牆鐘 3-30s → 0.19s 打平、輸出對 origin/main 逐位元組相同。archive+active 串接 idiom 與 `pmctl-trace.sh` 重複 ~12 行，兩 consumer 下不抽、file header 記錄理由 | ops | 2026-08-27 | pr:#547 | P2 | hygiene |

---

Expand Down Expand Up @@ -3534,3 +3535,78 @@ codex 自己的 apply_patch 工具實作（不在本 repo 控制範圍）。
驗證，非現場實戰）。`tests/bin/run-all-tests.sh` 104 passed 0 failed。

---

## CC-573 — `pmctl run-stats` 每事件行 fork 一個 jq ✅ 2026-08-27

**See**: pr:#547

**Problem**: `pmctl_run_stats_extract_line`（`runtime/lib/pmctl-run-stats.sh`）對
`events.jsonl` 的**每一行**執行一次 `jq -r`（過濾 `kind` 是否 `^run\.`、抽出 7 個
TSV 欄位）。掃描迴圈本身是純 bash（`mapfile -d $'\t'` + assoc array），沒有額外
fork，但 jq 是逐行 spawn。與 [[CC-364]] 修掉前的 `pmctl trace tail` 是**同一個
per-item subprocess 形狀**（見 [[per-item-subprocess-class]]）。

**Profile（2026-08-27，PATH jq wrapper 計數 + 牆鐘）**:

| N events | jq 呼叫數 | 牆鐘 |
|---:|---:|---:|
| 100 | 102 | 3.0s |
| 300 | 302 | 8.8s |
| 900 | 902 | 30.2s |

jq 呼叫 = N + 2（每事件一個 + 固定 2 個 setup/teardown）；牆鐘線性、
斜率約 **34ms/event**（WSL2 上 jq fork 主導）。真實 state store 的
`events.jsonl` = 6642 行時，`pmctl run-stats --json` 前景執行 **2 分鐘 timeout
（SIGTERM）**，外推約 225s。run-stats 是 v1.0 readiness 證據工具
（[[CC-358]] DoD），現在在真實資料上跑不完。

**Why now**: [[CC-364]] 剛把 `trace tail` 的同款問題修好，pattern 新鮮；
`events.jsonl` 的 archive+active 串接掃描現在有第二個 consumer。

**Requirement**:
1. 掃描階段改為**單次 `jq -R` 串流** over 串接的 archive+active 事件流：
逐行 `try fromjson catch null`，非物件或非 `run.*` 者輸出 skip 標記，
`run.*` 者輸出分隔欄位（沿用既有 7 欄：ts/kind/run_id/adapter/note/
exit_code/fallback_used），迴圈結束後**一次** decode 進 assoc array。
分隔符用 NUL 或 tab，比照 [[CC-364]] / CC-557 / CC-560 已驗證做法。
2. `--since` 過濾維持字典序 ISO-8601 比對語意；malformed row 容忍與
`episodes_malformed` 式的計數維持既有誠實回報慣例。
3. Archive-inclusive 掃描（`read_archives`）、gzip 不可用時的 active-only
fallback、`_meta` 回報，全部維持。
4. 新增 fault-sensitive perf 迴歸：PATH jq shim 計數，斷言小分區與大分區
的 jq 呼叫數相等（O(1) in event count），比照 [[CC-364]] 的
`case_trace_tail_single_jq_pass`。
5. 修正前後輸出以 `jq -S` 正規化後逐位元組比對，確認純效能修正、無行為變更。
6. 評估 `trace tail` 與 `run-stats` 的「archive 檔案發現 + 串接 + gzip fallback」
是否值得抽成共用 primitive（兩者 jq 程式不同，只有串流串接那段可共用）；
若第二個 consumer 不足以支撐抽象就記錄理由、不強抽。

**Non-goals**: 不改 run-stats 的輸出 schema 或 CLI 介面；不改 `--since` 驗證；
不動 [[CC-358]] 的 `fallback_used` event 訊號本身。

**Cross-link**: [[CC-364]]（同形狀的第一次修正，含 profiling 方法與 oracle 測試
技巧）、[[per-item-subprocess-class]]。

**Closure 2026-08-27 (pr:#547)**: 掃描階段改為單次 `jq -R` 串流 over 串接的
archive+active 事件流。新 helper `pmctl_run_stats_filter_program`（heredoc jq
程式，逐行 `try fromjson catch null`、非物件／非 `run.*`／被 `--since` 濾掉者輸出
`empty`、其餘輸出 7 欄 `@tsv`）＋ `pmctl_run_stats_scan_stream`（stdin 讀 TSV、
`mapfile -d $'\t'` 折進 `_rs_*`），取代 `extract_line`／`process_line`／
`scan_path`／`scan_gzip_path`。`--since` 謂詞下推進 jq，語意與原 shell 檢查完全
相同（僅在有界且 ts 非空且 ts < 界時丟棄）。

**Perf 佐證**：合成資料 jq 呼叫 102/302/902 → **2/2/2**，牆鐘 3.0–30.2s →
**0.19s 打平**。異質 fixture（正常 terminal／partial／nonzero exit／cancelled／
missing-terminal／fallback／pre-`--since`／2 個 malformed 行／非 run 事件／
archive-only run；3 adapter）輸出對 `origin/main` 逐位元組相同（`jq -S` 正規化，
JSON 與 human 皆是）。新增 `case_run_stats_single_jq_pass`（jq shim 計數 20 vs
200 run 相等）與 `case_run_stats_streaming_matches_reference`（golden 比對）。
`test-pmctl-run-stats.sh` 17 passed，全套 105 passed 0 failed 0 skipped。

**Req 6（共用 primitive）**：archive-glob + gzip-check + concat-then-one-jq-pass
的 ~12 行 idiom 現與 `pmctl-trace.sh` 重複。評估後**不抽**：兩者 jq 程式與輸出
consumer 不同，gzip 不可用的訊號也分歧（trace tail `read_archives=0`；run-stats
`archive_scanned=false` + `_meta`）；兩 consumer 下 callback 間接層不划算。理由寫進
`pmctl-run-stats.sh` file header，待第三個 consumer 出現再議。未立 follow-up 票。

---
170 changes: 87 additions & 83 deletions runtime/lib/pmctl-run-stats.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,16 +2,23 @@
# pmctl run-stats — per-adapter success/failure/fallback analysis over
# events.jsonl (CC-358). Read-only consumer: never calls events_append or any
# state-writer write helper. Uses the same ISO-8601 lexicographic string
# comparison and archive-inclusive scan model as pmctl trace tail
# (runtime/lib/pmctl-trace.sh) rather than inventing a second parser for the
# same file format. (trace tail itself now does a single streaming jq pass;
# this reader still extracts per line — a standalone follow-up, not CC-364.)
# comparison, archive-inclusive scan model, and single streaming jq pass as
# pmctl trace tail (runtime/lib/pmctl-trace.sh) rather than inventing a second
# parser for the same file format.
#
# The archive-glob + gzip-check + concat-then-one-jq-pass idiom is duplicated
# between this file and pmctl-trace.sh (~12 lines each). Extraction into a
# shared events-scan primitive was considered and deferred: the two jq
# programs and output consumers differ, and the gzip-unavailable signalling
# diverges (trace tail: read_archives=0; run-stats: archive_scanned=false +
# `_meta`). At two consumers the shared seam isn't clearly worth the callback
# indirection; revisit if a third consumer appears.
#
# Archive-inclusive by default, matching pmctl trace tail's read_archives=1:
# rotated archive/events-*.jsonl.gz files are scanned alongside the active
# events.jsonl so a --since window reaching past rotation still counts every
# matching run. Falls back to active-file-only (and says so in `_meta`) only
# when gzip is unavailable, same fallback pmctl_trace_tail uses.
# when gzip is unavailable, same fallback pmctl trace tail uses.

pmctl_run_stats_usage() {
printf 'usage: pmctl run-stats [--since <ISO8601-or-date>] [--by-adapter] [--json]\n' >&2
Expand All @@ -31,80 +38,68 @@ pmctl_run_stats_ensure_state_writer() {
. "$repo_root/runtime/lib/state-writer.sh"
}

# Emits one TSV row per run.* event whose kind matches ^run\. — ts, kind,
# run_id, adapter, note, exit_code, fallback_used(true/false) — or nothing
# when the line isn't a matching run event. One jq invocation per line.
pmctl_run_stats_extract_line() {
local line="${1:-}"
printf '%s\n' "$line" | jq -r '
if (.kind? // "" | test("^run\\.")) then
[
(.ts // ""),
(.kind // ""),
(.payload.run_id // .subject_id // ""),
(.payload.adapter // ""),
(.payload.note // ""),
(.payload.exit_code // 0),
((.payload.fallback_used // false) | tostring)
] | @tsv
else
empty
end
' 2>/dev/null
# Single streaming jq program over the concatenated archive+active event
# stream (raw input, one JSON object per line). Replaces the former per-line
# jq spawn: it runs once for the whole partition regardless of event count.
# Emits one TSV row per run.* event that passes the --since bound -- ts,
# kind, run_id, adapter, note, exit_code, fallback_used(true/false) -- and
# nothing for non-run, filtered-out, malformed, or non-object lines
# (run-stats has never counted malformed rows; that silent skip is kept).
# The --since predicate matches the former shell check exactly: an event is
# dropped only when a bound is set AND its ts is non-empty AND ts < bound,
# so an empty ts is always kept.
pmctl_run_stats_filter_program() {
cat <<'JQ'
(try fromjson catch null) as $o
| if ($o | type) != "object" then empty
elif (($o.kind // "") | test("^run\\.")) | not then empty
else
($o.ts // "") as $ts
| if ($since != "" and $ts != "" and $ts < $since) then empty
else
[ $ts,
($o.kind // ""),
($o.payload.run_id // $o.subject_id // ""),
($o.payload.adapter // ""),
($o.payload.note // ""),
($o.payload.exit_code // 0),
(($o.payload.fallback_used // false) | tostring)
] | @tsv
end
end
JQ
}

# Processes one raw events.jsonl(.gz) line, updating the caller's _rs_* assoc
# arrays and consulting the caller's $since -- relies on bash dynamic scoping
# (this is only ever called from within pmctl_run_stats, so those `local`
# arrays are visible here without being passed explicitly).
pmctl_run_stats_process_line() {
local line="${1:-}" ts kind run_id adapter note exit_code fallback
[[ -n "$line" ]] || return 0
local fields
fields="$(pmctl_run_stats_extract_line "$line")" || return 0
[[ -n "$fields" ]] || return 0
# NOT `IFS=$'\t' read -r ... <<<`: bash's word-splitting treats tab as
# "IFS whitespace" and collapses runs of it / trims it at the edges even
# when IFS is set to tab alone, silently eating the empty `note` field
# and shifting every field after it. `mapfile -d` splits on the literal
# byte with no such collapsing.
# Consumes the concatenated event stream on stdin (one jq pass, wired in by
# the caller), folding every emitted TSV row into the caller's _rs_* assoc
# arrays. Relies on bash dynamic scoping -- only ever called from within
# pmctl_run_stats, so the _rs_* arrays are visible here -- and on a redirect
# rather than a pipe at the call site so those arrays survive the loop.
pmctl_run_stats_scan_stream() {
local _rs_rec kind run_id adapter note exit_code fallback
local -a _rs_f=()
mapfile -d $'\t' -t _rs_f <<< "$fields"
ts="${_rs_f[0]:-}"; kind="${_rs_f[1]:-}"; run_id="${_rs_f[2]:-}"
adapter="${_rs_f[3]:-}"; note="${_rs_f[4]:-}"; exit_code="${_rs_f[5]:-}"
fallback="${_rs_f[6]:-}"; fallback="${fallback%$'\n'}"
[[ -n "$run_id" ]] || return 0
if [[ -n "$since" && -n "$ts" && "$ts" < "$since" ]]; then
return 0
fi
_rs_seen["$run_id"]=1
[[ -n "$adapter" ]] && _rs_adapter["$run_id"]="$adapter"
[[ "$fallback" == "true" ]] && _rs_fallback["$run_id"]=1
case "$kind" in
run.completed|run.failed|run.cancelled)
_rs_terminal_kind["$run_id"]="$kind"
_rs_terminal_note["$run_id"]="$note"
_rs_terminal_exit["$run_id"]="$exit_code"
;;
esac
}

# Redirect (not a pipe) so the loop runs in the current shell and the
# caller's _rs_* associative arrays survive past it.
pmctl_run_stats_scan_path() {
local path="${1:-}" line
[[ -f "$path" ]] || return 0
while IFS= read -r line || [[ -n "$line" ]]; do
pmctl_run_stats_process_line "$line"
done < "$path"
}

pmctl_run_stats_scan_gzip_path() {
local path="${1:-}" line
[[ -f "$path" ]] || return 0
while IFS= read -r line || [[ -n "$line" ]]; do
pmctl_run_stats_process_line "$line"
done < <(gzip -dc "$path" 2>/dev/null)
while IFS= read -r _rs_rec; do
# NOT `IFS=$'\t' read -r ... <<<`: bash's word-splitting treats tab as
# "IFS whitespace" and collapses runs of it / trims it at the edges even
# when IFS is set to tab alone, silently eating the empty `note` field
# and shifting every field after it. `mapfile -d` splits on the literal
# byte with no such collapsing.
mapfile -d $'\t' -t _rs_f <<< "$_rs_rec"
kind="${_rs_f[1]:-}"; run_id="${_rs_f[2]:-}"
adapter="${_rs_f[3]:-}"; note="${_rs_f[4]:-}"; exit_code="${_rs_f[5]:-}"
fallback="${_rs_f[6]:-}"; fallback="${fallback%$'\n'}"
[[ -n "$run_id" ]] || continue
_rs_seen["$run_id"]=1
[[ -n "$adapter" ]] && _rs_adapter["$run_id"]="$adapter"
[[ "$fallback" == "true" ]] && _rs_fallback["$run_id"]=1
case "$kind" in
run.completed|run.failed|run.cancelled)
_rs_terminal_kind["$run_id"]="$kind"
_rs_terminal_note["$run_id"]="$note"
_rs_terminal_exit["$run_id"]="$exit_code"
;;
esac
done
}

pmctl_run_stats() {
Expand Down Expand Up @@ -185,6 +180,7 @@ pmctl_run_stats() {
declare -A _rs_terminal_exit=() _rs_fallback=() _rs_seen=()

local -a _rs_archives=()
local _rs_archive_path _rs_program
if [[ -d "$archive_dir" ]]; then
while IFS= read -r -d '' _rs_archive_path; do
_rs_archives+=("$_rs_archive_path")
Expand All @@ -198,13 +194,21 @@ pmctl_run_stats() {
archive_scanned=false
fi

if [[ "$archive_scanned" == true ]]; then
for _rs_archive_path in "${_rs_archives[@]}"; do
pmctl_run_stats_scan_gzip_path "$_rs_archive_path"
done
unset _rs_archive_path
fi
pmctl_run_stats_scan_path "$events_file"
# One jq pass over the whole partition: archives first (in the same
# filename-sorted order find/sort produced), then the active file. A
# redirect (not a pipe) keeps pmctl_run_stats_scan_stream in this shell so
# its _rs_* writes survive.
_rs_program="$(pmctl_run_stats_filter_program)"
pmctl_run_stats_scan_stream < <(
{
if [[ "$archive_scanned" == true ]]; then
for _rs_archive_path in "${_rs_archives[@]}"; do
[[ -f "$_rs_archive_path" ]] && gzip -dc "$_rs_archive_path" 2>/dev/null
done
fi
[[ -f "$events_file" ]] && cat "$events_file"
} | jq -R -r --arg since "$since" "$_rs_program"
)

local tmp_dir agg_file
tmp_dir="$(mktemp -d)" || return 2
Expand Down
Loading