Skip to content

fix(wiki-worker): bound the session-events DB fallback #328

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
khustup2 merged 2 commits into from
Jul 21, 2026
Merged

fix(wiki-worker): bound the session-events DB fallback #328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0035bff
fix(wiki-worker): bound the session-events DB fallback
khustup2 Jul 21, 2026
426cc11
test(wiki-worker): cover bounded-fetch edges + error paths
khustup2 Jul 21, 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
28 changes: 18 additions & 10 deletions src/hooks/codex/wiki-worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import { buildTrailingPromptInvocation } from "../wiki-worker-spawn.js";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { finalizeSummary, releaseLock, readState } from "../summary-state.js";
import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { capLinesByBytes, newRowsFromWindow, stampOffset, WIKI_FALLBACK_MAX_ROWS, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { redactSecrets } from "../shared/redact.js";
import { uploadSummary } from "../upload-summary.js";
import { log as _log } from "../../utils/debug.js";
Expand Down Expand Up @@ -135,19 +135,27 @@ function formatExecFailure(error: any): string {

async function main(): Promise<void> {
try {
// 1. Fetch session events from sessions table
// 1. Fetch session events from the sessions table — BOUNDED to the newest
// WIKI_FALLBACK_MAX_ROWS rows (reversed to chronological) plus the true
// total. The old unbounded `ORDER BY ASC` materialized the whole fat
// `message` column (tens of MB, ~30s cold on a mega-session) even though
// only the newest un-summarized rows are consumed. `count(*)` reads no fat
// column, so `total` is cheap and keeps the offset math + stamped offset right.
wlog("fetching session events");
const rows = await query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC`
);

if (rows.length === 0) {
const likePat = esc(`/sessions/%${cfg.sessionId}%`);
const cntRows = await query(`SELECT count(*) AS n FROM "${cfg.sessionsTable}" WHERE path LIKE E'${likePat}'`);
const total = Number(cntRows[0]?.["n"] ?? 0);
if (total === 0) {
wlog("no session events found — exiting");
return;
}
const fetched = await query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${likePat}' ORDER BY creation_date DESC LIMIT ${WIKI_FALLBACK_MAX_ROWS}`
);
const rows = fetched.reverse();

const jsonlLines = rows.length;
const jsonlLines = total;
Comment on lines +138 to +158

@coderabbitai coderabbitai Bot Jul 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Locate esc() to confirm whether it accounts for E''-literal backslash semantics.
rg -nP -A8 'function\s+esc\s*\(' --type=ts
rg -nP "LIKE E'" --type=ts -C2

Repository: activeloopai/hivemind

Length of output: 159

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== files =="
git ls-files 'src/hooks/**/wiki-worker.ts' 'src/hooks/**/utils.ts' 'src/hooks/**/shared.ts' 'src/hooks/**/index.ts' | sed -n '1,120p'

echo
echo "== outline codex/wiki-worker.ts =="
ast-grep outline src/hooks/codex/wiki-worker.ts --view expanded || true

echo
echo "== outline pi/wiki-worker.ts =="
ast-grep outline src/hooks/pi/wiki-worker.ts --view expanded || true

echo
echo "== search esc definitions/usages =="
rg -n "esc\s*\(" src/hooks -g '*.ts' -C 2 || true

echo
echo "== search LIKE E' patterns =="
rg -n "LIKE E'" src/hooks -g '*.ts' -C 2 || true

Repository: activeloopai/hivemind

Length of output: 16901

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== read relevant sections =="
for f in src/hooks/codex/wiki-worker.ts src/hooks/pi/wiki-worker.ts src/hooks/wiki-worker.ts; do
  echo "--- $f ---"
  wc -l "$f"
done

echo
echo "== src/hooks/codex/wiki-worker.ts around queries =="
sed -n '1,240p' src/hooks/codex/wiki-worker.ts | nl -ba | sed -n '1,220p'

echo
echo "== src/hooks/pi/wiki-worker.ts around queries =="
sed -n '1,220p' src/hooks/pi/wiki-worker.ts | nl -ba | sed -n '1,220p'

echo
echo "== src/hooks/wiki-worker.ts around queries =="
sed -n '1,220p' src/hooks/wiki-worker.ts | nl -ba | sed -n '1,220p'

Repository: activeloopai/hivemind

Length of output: 486

🏁 Script executed:

#!/bin/bash
set -euo pipefail

python3 - <<'PY'
from pathlib import Path
for path in [
    Path('src/hooks/codex/wiki-worker.ts'),
    Path('src/hooks/pi/wiki-worker.ts'),
    Path('src/hooks/wiki-worker.ts'),
]:
    print(f'--- {path} ---')
    lines = path.read_text().splitlines()
    for i, line in enumerate(lines, 1):
        if 'function esc' in line or 'LIKE E\'' in line or "LIKE '" in line:
            start = max(1, i-3)
            end = min(len(lines), i+8)
            for j in range(start, end+1):
                print(f'{j:4}: {lines[j-1]}')
            print()
PY

Repository: activeloopai/hivemind

Length of output: 7317

🌐 Web query:

PostgreSQL LIKE operator backslash default escape E'' standard_conforming_strings documentation

💡 Result:

In PostgreSQL, the backslash character () serves two distinct functions that often cause confusion: one as a literal string escape mechanism and another as an escape character specifically for the LIKE operator's pattern matching [1][2][3]. 1. String Literal Parsing and standard_conforming_strings: The configuration parameter standard_conforming_strings controls how backslashes are interpreted within ordinary SQL string literals ('...'). - When standard_conforming_strings is on (the default in modern PostgreSQL), backslashes in ordinary string literals are treated as literal backslash characters [2][4]. - When standard_conforming_strings is off, backslashes in ordinary string literals are treated as escape characters (e.g., '\n' becomes a newline) [2][4]. - To represent backslash escape sequences reliably, you should use E-prefixed string constants (e.g., E'\'), which treat backslashes as escape characters regardless of the standard_conforming_strings setting [2][5]. 2. LIKE Operator Escape Behavior: Regardless of the standard_conforming_strings setting, the LIKE operator always uses the backslash as its default escape character for pattern matching [1][3]. This means that within a LIKE pattern, a backslash is used to escape wildcard characters like '%' and '' [1][6]. - To match a literal backslash in a LIKE clause, you must escape it with another backslash (e.g., '\') [1][5]. - Because of the two-layer interpretation (string parsing + LIKE pattern matching), searching for a literal backslash often requires multiple backslashes, such as '\\' in an ordinary string literal, or '\\' in an E-string literal [3]. - You can override the default LIKE escape character using the ESCAPE clause [1][6]. For example, LIKE '%#%' ESCAPE '#' treats '#' as the escape character instead of the backslash [1][3]. To disable the LIKE escape mechanism entirely, you can use ESCAPE '' [1][7]. In summary, standard_conforming_strings governs how string constants are parsed, whereas the LIKE operator's use of the backslash as an escape character is a fixed behavior of the pattern matching engine [1][3]. Applications requiring reliable handling of backslashes should use E-prefixed string constants and explicitly specify an ESCAPE character for LIKE patterns if needed [1][2].

Citations:

🏁 Script executed:

#!/bin/bash
set -euo pipefail

python3 - <<'PY'
from pathlib import Path

for path, ranges in {
    'src/hooks/codex/wiki-worker.ts': [(50, 170)],
    'src/hooks/pi/wiki-worker.ts': [(60, 155)],
    'src/hooks/wiki-worker.ts': [(55, 170)],
    'src/hooks/upload-summary.ts': [(52, 65)],
}.items():
    print(f'--- {path} ---')
    lines = Path(path).read_text().splitlines()
    for start, end in ranges:
        for i in range(start, min(end, len(lines)) + 1):
            print(f'{i:4}: {lines[i-1]}')
        print()
PY

Repository: activeloopai/hivemind

Length of output: 17833

🏁 Script executed:

#!/bin/bash
set -euo pipefail

python3 - <<'PY'
from pathlib import Path

for path, start, end in [
    ('src/hooks/codex/wiki-worker.ts', 54, 160),
    ('src/hooks/pi/wiki-worker.ts', 64, 150),
    ('src/hooks/wiki-worker.ts', 59, 165),
]:
    print(f'--- {path} ---')
    lines = Path(path).read_text().splitlines()
    for i in range(start, min(end, len(lines)) + 1):
        print(f'{i:4}: {lines[i-1]}')
    print()
PY

Repository: activeloopai/hivemind

Length of output: 15321

🌐 Web query:

PostgreSQL E string literal backslash LIKE pattern default escape documentation

💡 Result:

In PostgreSQL, there are two distinct mechanisms involving backslashes: string constant escaping (lexical level) and LIKE pattern escaping (pattern matching level). LIKE Pattern Escape In a LIKE pattern, the default escape character is the backslash () [1][2]. This allows you to match literal underscore (_) or percent sign (%) characters by preceding them with the backslash (e.g., '%') [1][2]. To match a literal backslash, you write two backslashes ('\') [1][2]. You can change this default behavior by using the ESCAPE clause (e.g., LIKE 'pattern' ESCAPE '!') or disable it entirely by using ESCAPE '' [1][2]. PostgreSQL's default of using the backslash as an escape character is a nonstandard extension to the SQL standard, which specifies that there should be no default escape character [1][2]. E String Literal An E string literal (e.g., E'string') is a lexical extension that enables C-style backslash escape sequences (such as \n for newline or \xHH for hexadecimal bytes) [3][4]. Because these sequences are interpreted by the parser when the query is read, a backslash intended to be part of a LIKE pattern must be doubled if it is inside an E string literal [3][4]. For example, to match a literal percent sign using an E string literal, you would write E'\%'—the first backslash escapes the second one for the string constant, leaving a single backslash for the LIKE operator to use as an escape character [1][3]. Important Configuration Note Since PostgreSQL 9.1, the default setting for the configuration parameter standard_conforming_strings is ON [3][5]. This means that backslashes in standard (non-E) string constants are treated as literal characters rather than escape sequences [3][5]. Consequently, it is recommended to use E string literals only when backslash escape sequences are actually needed [3][5].

Citations:

Use plain quotes for these LIKE patterns
esc() already returns a string for a plain SQL literal; E'...' adds an extra backslash-parse step before LIKE sees the pattern, so backslash-containing session IDs/paths can match differently here. Use the plain '${likePat}' form to match the other queries in these files.

  • src/hooks/codex/wiki-worker.ts#L145-L154
  • src/hooks/pi/wiki-worker.ts#L130-L139
📍 Affects 2 files
  • src/hooks/codex/wiki-worker.ts#L138-L158 (this comment)
  • src/hooks/pi/wiki-worker.ts#L123-L143
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/hooks/codex/wiki-worker.ts` around lines 138 - 158, Update the LIKE
predicates in the session-event queries to use plain SQL string quotes around
likePat instead of E-string literals. Apply this in
src/hooks/codex/wiki-worker.ts lines 138-158 and src/hooks/pi/wiki-worker.ts
lines 123-143, covering both count and fetch queries while preserving the
existing esc() result and query behavior.


const pathRows = await query(
`SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` +
Expand Down Expand Up @@ -201,7 +209,7 @@ async function main(): Promise<void> {
// full session on every run is what drove the ENOBUFS / 120s-timeout
// failures on long (4000+ event) sessions — a stuck offset meant every run
// re-summarized everything from scratch.
const newRows = prevOffset > 0 ? rows.slice(prevOffset) : rows;
const newRows = newRowsFromWindow(rows, total, prevOffset);
if (prevOffset > 0 && newRows.length === 0) {
wlog(`no new events since last summary (offset=${prevOffset}, total=${jsonlLines}) — skipping`);
return;
Expand Down
43 changes: 33 additions & 10 deletions src/hooks/cursor/wiki-worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import { fileURLToPath } from "node:url";
import { finalizeSummary, releaseLock, readState } from "../summary-state.js";
import { readSessionEventCache } from "../session-event-cache.js";
import { buildSessionPath } from "../../utils/session-path.js";
import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { capLinesByBytes, newRowsFromWindow, stampOffset, WIKI_FALLBACK_MAX_ROWS, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { redactSecrets } from "../shared/redact.js";
import { uploadSummary } from "../upload-summary.js";
import { log as _log } from "../../utils/debug.js";
Expand Down Expand Up @@ -126,12 +126,26 @@ async function main(): Promise<void> {
// mega-sessions). Falls back to the DB whenever the cache is absent
// (session resumed on another machine), empty, or — once the offset is
// known — shorter than the offset already summarized.
const dbFetch = () => query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC`
);
// Bounded DB fallback: the NEWEST WIKI_FALLBACK_MAX_ROWS rows (reversed to
// chronological) plus the true `total`. The old unbounded `ORDER BY ASC`
// materialized the whole fat `message` column (tens of MB, ~30s cold on a
// mega-session) even though only the newest un-summarized rows are consumed.
// `count(*)` reads no fat column, so `total` is cheap and lets the offset
// math (`newRowsFromWindow`) and the stamped offset stay correct.
const dbFetch = async (): Promise<{ rows: Record<string, unknown>[]; total: number }> => {
const like = esc(`/sessions/%${cfg.sessionId}%`);
const cnt = await query(`SELECT count(*) AS n FROM "${cfg.sessionsTable}" WHERE path LIKE E'${like}'`);
const total = Number(cnt[0]?.["n"] ?? 0);
if (total === 0) return { rows: [], total: 0 };
const r = await query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${like}' ORDER BY creation_date DESC LIMIT ${WIKI_FALLBACK_MAX_ROWS}`
);
return { rows: r.reverse(), total };
};
Comment on lines +135 to +145

@coderabbitai coderabbitai Bot Jul 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win

dbFetch's count(*)-then-SELECT ordering is a TOCTOU race that can silently drop new session events.

Both workers implement dbFetch as two independent query() round trips: count(*) first, then the bounded SELECT ... ORDER BY creation_date DESC LIMIT 2000. query() itself can retry with up to ~30s of backoff on transient 401/403/429/5xx responses, so the gap between the two queries can be substantial while the session's capture hook keeps appending rows. When that happens — especially for the common case where the whole session fits in one fetch (rows.length < WIKI_FALLBACK_MAX_ROWS) — the earlier total understates the row count that actually existed by the time the SELECT ran. newRowsFromWindow(rows, dbTotal, prevOffset) then returns a window that starts strictly after prevOffset, skipping the rows added during the race, while the offset is still stamped from that same stale dbTotal — high enough to mark those skipped rows as already summarized. They are never fed to the agent and never revisited on a later run: a silent, permanent loss of session events from the wiki summary.

  • src/hooks/cursor/wiki-worker.ts#L135-L145: swap the order in dbFetch — fetch the bounded rows first, then count(*), and set total = Math.max(count, r.length) so total can never understate the returned window.
  • src/hooks/hermes/wiki-worker.ts#L135-L145: apply the identical reordering + Math.max clamp to its dbFetch.

The same count(*)-then-SELECT pattern also appears in src/hooks/wiki-worker.ts, src/hooks/codex/wiki-worker.ts, and src/hooks/pi/wiki-worker.ts per the supplied graph evidence; those weren't in this review's file set but should get the same fix.

📍 Affects 2 files
  • src/hooks/cursor/wiki-worker.ts#L135-L145 (this comment)
  • src/hooks/hermes/wiki-worker.ts#L135-L145
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/hooks/cursor/wiki-worker.ts` around lines 135 - 145, The dbFetch
implementations can undercount rows because count(*) runs before the bounded
SELECT. In src/hooks/cursor/wiki-worker.ts lines 135-145 and
src/hooks/hermes/wiki-worker.ts lines 135-145, update dbFetch to fetch the
bounded rows first, then query the count, and set total to the maximum of the
count and returned row length; apply the same correction to the matching dbFetch
patterns in src/hooks/wiki-worker.ts, src/hooks/codex/wiki-worker.ts, and
src/hooks/pi/wiki-worker.ts.


let usedLocalCache = false;
let dbTotal = 0; // true row count when the bounded DB path was used (else 0)
let rows: Record<string, unknown>[];
const cachedLines = readSessionEventCache(cfg.sessionId);
if (cachedLines && cachedLines.length > 0) {
Expand All @@ -140,15 +154,19 @@ async function main(): Promise<void> {
wlog(`loaded ${rows.length} events from local cache`);
} else {
wlog("fetching session events");
rows = await dbFetch();
const f = await dbFetch();
rows = f.rows;
dbTotal = f.total;
}

if (rows.length === 0) {
wlog("no session events found — exiting");
return;
}

let jsonlLines = rows.length;
// The offset high-water (stamped into the summary) must be the TRUE total, not the
// bounded window length — else the next run's offset regresses and re-summarizes.
let jsonlLines = usedLocalCache ? rows.length : dbTotal;

// Derive the server path locally when using the cache (avoids a second
// self-session `SELECT DISTINCT path` scan); the DB branch keeps its lookup.
Expand Down Expand Up @@ -213,15 +231,20 @@ async function main(): Promise<void> {
// session from the DB so no genuinely-new rows get sliced to nothing.
if (usedLocalCache && rows.length < prevOffset) {
wlog(`local cache (${rows.length}) < summarized offset (${prevOffset}) — refetching from DB`);
rows = await dbFetch();
jsonlLines = rows.length;
const f = await dbFetch();
rows = f.rows;
dbTotal = f.total;
jsonlLines = dbTotal;
usedLocalCache = false;
}

// Feed the agent only the rows added since the last summary. Reprocessing
// the full session on every run is what drives ENOBUFS / 120s-timeout
// failures on long (4000+ event) sessions — a stuck offset re-summarizes
// everything from scratch.
const newRows = prevOffset > 0 ? rows.slice(prevOffset) : rows;
const newRows = usedLocalCache
? (prevOffset > 0 ? rows.slice(prevOffset) : rows)
: newRowsFromWindow(rows, dbTotal, prevOffset);
if (prevOffset > 0 && newRows.length === 0) {
wlog(`no new events since last summary (offset=${prevOffset}, total=${jsonlLines}) — skipping`);
return;
Expand Down
47 changes: 36 additions & 11 deletions src/hooks/hermes/wiki-worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import { fileURLToPath } from "node:url";
import { finalizeSummary, releaseLock, readState } from "../summary-state.js";
import { readSessionEventCache } from "../session-event-cache.js";
import { buildSessionPath } from "../../utils/session-path.js";
import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { capLinesByBytes, newRowsFromWindow, stampOffset, WIKI_FALLBACK_MAX_ROWS, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { redactSecrets } from "../shared/redact.js";
import { uploadSummary } from "../upload-summary.js";
import { log as _log } from "../../utils/debug.js";
Expand Down Expand Up @@ -126,12 +126,26 @@ async function main(): Promise<void> {
// mega-sessions). Falls back to the DB whenever the cache is absent
// (session resumed on another machine), empty, or — once the offset is
// known — shorter than the offset already summarized.
const dbFetch = () => query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC`
);
// Bounded DB fallback: the NEWEST WIKI_FALLBACK_MAX_ROWS rows (reversed to
// chronological) plus the true `total`. The old unbounded `ORDER BY ASC`
// materialized the whole fat `message` column (tens of MB, ~30s cold on a
// mega-session) even though only the newest un-summarized rows are consumed.
// `count(*)` reads no fat column, so `total` is cheap and lets the offset
// math (`newRowsFromWindow`) and the stamped offset stay correct.
const dbFetch = async (): Promise<{ rows: Record<string, unknown>[]; total: number }> => {
const like = esc(`/sessions/%${cfg.sessionId}%`);
const cnt = await query(`SELECT count(*) AS n FROM "${cfg.sessionsTable}" WHERE path LIKE E'${like}'`);
const total = Number(cnt[0]?.["n"] ?? 0);
if (total === 0) return { rows: [], total: 0 };
const r = await query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${like}' ORDER BY creation_date DESC LIMIT ${WIKI_FALLBACK_MAX_ROWS}`
);
return { rows: r.reverse(), total };
};

let usedLocalCache = false;
let dbTotal = 0; // true row count when the bounded DB path was used (else 0)
let rows: Record<string, unknown>[];
const cachedLines = readSessionEventCache(cfg.sessionId);
if (cachedLines && cachedLines.length > 0) {
Expand All @@ -140,15 +154,19 @@ async function main(): Promise<void> {
wlog(`loaded ${rows.length} events from local cache`);
} else {
wlog("fetching session events");
rows = await dbFetch();
const f = await dbFetch();
rows = f.rows;
dbTotal = f.total;
}

if (rows.length === 0) {
wlog("no session events found — exiting");
return;
}

let jsonlLines = rows.length;
// The offset high-water (stamped into the summary) must be the TRUE total, not the
// bounded window length — else the next run's offset regresses and re-summarizes.
let jsonlLines = usedLocalCache ? rows.length : dbTotal;

// Derive the server path locally when using the cache (avoids a second
// self-session `SELECT DISTINCT path` scan); the DB branch keeps its lookup.
Expand Down Expand Up @@ -213,15 +231,22 @@ async function main(): Promise<void> {
// session from the DB so no genuinely-new rows get sliced to nothing.
if (usedLocalCache && rows.length < prevOffset) {
wlog(`local cache (${rows.length}) < summarized offset (${prevOffset}) — refetching from DB`);
rows = await dbFetch();
jsonlLines = rows.length;
const f = await dbFetch();
rows = f.rows;
dbTotal = f.total;
jsonlLines = dbTotal;
usedLocalCache = false;
}

// Feed the agent only the rows added since the last summary. Reprocessing
// the full session on every run is what drives ENOBUFS / 120s-timeout
// failures on long (4000+ event) sessions — a stuck offset re-summarizes
// everything from scratch.
const newRows = prevOffset > 0 ? rows.slice(prevOffset) : rows;
// everything from scratch. The local-cache path holds the FULL session, so a
// plain slice is exact; the bounded DB window maps the full-history offset
// through newRowsFromWindow.
const newRows = usedLocalCache
? (prevOffset > 0 ? rows.slice(prevOffset) : rows)
: newRowsFromWindow(rows, dbTotal, prevOffset);
if (prevOffset > 0 && newRows.length === 0) {
wlog(`no new events since last summary (offset=${prevOffset}, total=${jsonlLines}) — skipping`);
return;
Expand Down
28 changes: 18 additions & 10 deletions src/hooks/pi/wiki-worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import { buildTrailingPromptInvocation } from "../wiki-worker-spawn.js";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { finalizeSummary, releaseLock, readState } from "../summary-state.js";
import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { capLinesByBytes, newRowsFromWindow, stampOffset, WIKI_FALLBACK_MAX_ROWS, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js";
import { redactSecrets } from "../shared/redact.js";
import { uploadSummary } from "../upload-summary.js";
import { log as _log } from "../../utils/debug.js";
Expand Down Expand Up @@ -120,19 +120,27 @@ function cleanup(): void {

async function main(): Promise<void> {
try {
// 1. Fetch session events from sessions table
// 1. Fetch session events from the sessions table — BOUNDED to the newest
// WIKI_FALLBACK_MAX_ROWS rows (reversed to chronological) plus the true
// total. The old unbounded `ORDER BY ASC` materialized the whole fat
// `message` column (tens of MB, ~30s cold on a mega-session) even though
// only the newest un-summarized rows are consumed. `count(*)` reads no fat
// column, so `total` is cheap and keeps the offset math + stamped offset right.
wlog("fetching session events");
const rows = await query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC`
);

if (rows.length === 0) {
const likePat = esc(`/sessions/%${cfg.sessionId}%`);
const cntRows = await query(`SELECT count(*) AS n FROM "${cfg.sessionsTable}" WHERE path LIKE E'${likePat}'`);
const total = Number(cntRows[0]?.["n"] ?? 0);
if (total === 0) {
wlog("no session events found — exiting");
return;
}
const fetched = await query(
`SELECT message, creation_date FROM "${cfg.sessionsTable}" ` +
`WHERE path LIKE E'${likePat}' ORDER BY creation_date DESC LIMIT ${WIKI_FALLBACK_MAX_ROWS}`
);
const rows = fetched.reverse();

const jsonlLines = rows.length;
const jsonlLines = total;

const pathRows = await query(
`SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` +
Expand Down Expand Up @@ -186,7 +194,7 @@ async function main(): Promise<void> {
// the full session on every run is what drives ENOBUFS / 120s-timeout
// failures on long (4000+ event) sessions — a stuck offset re-summarizes
// everything from scratch.
const newRows = prevOffset > 0 ? rows.slice(prevOffset) : rows;
const newRows = newRowsFromWindow(rows, total, prevOffset);
if (prevOffset > 0 && newRows.length === 0) {
wlog(`no new events since last summary (offset=${prevOffset}, total=${jsonlLines}) — skipping`);
return;
Expand Down
33 changes: 33 additions & 0 deletions src/hooks/wiki-offset.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,39 @@
/** Max bytes of session JSONL fed to the summarizer in one run. */
export const WIKI_JSONL_MAX_BYTES = 4 * 1024 * 1024;

/**
* Max session rows the DB fallback fetches, newest-first. The old fallback did an
* unbounded `ORDER BY creation_date ASC` with no limit, which materializes the WHOLE fat
* `message` column — tens of MB, ~30s cold on a mega-session — even though the summarizer
* only ever consumes the newest un-summarized rows (`capLinesByBytes` discards the rest).
* This caps the fetch to a superset of what any one run can use.
*/
export const WIKI_FALLBACK_MAX_ROWS = 2000;

/**
* Select the un-summarized "new" rows from a BOUNDED newest-N window.
*
* The workers fetch the whole session ASC and take `rows.slice(prevOffset)`, where
* `prevOffset` is a count over the FULL history. When the fetch is instead bounded to the
* newest `window.length` rows (of a `total`-row session), that full-history index no longer
* addresses the window, so the plain slice is wrong. This maps it correctly:
*
* newCount = max(0, total - prevOffset) // rows added since the last summary
* → the LAST `newCount` rows of the window are the un-summarized ones.
*
* When the window doesn't reach back to `prevOffset` (the session grew by more than the
* window since the last summary), `newCount >= window.length` and the whole window is
* returned — the older new rows fell outside the fetch, exactly the ones `capLinesByBytes`
* would drop anyway (it keeps the newest). So bounding never changes what a run summarizes
* beyond what the byte cap already does. `prevOffset <= 0` returns the whole window.
*/
export function newRowsFromWindow<T>(window: readonly T[], total: number, prevOffset: number): T[] {
if (prevOffset <= 0) return window.slice();
const newCount = Math.max(0, total - prevOffset);
if (newCount >= window.length) return window.slice();
return window.slice(window.length - newCount);
}

/** Matches the offset line in a stored summary, regardless of leading bullet.
* Single source of truth for both detection (stampOffset) and extraction
* (parseOffset) so the round-trip contract can't drift. */
Expand Down
Loading