mirror of
https://github.com/garrytan/gbrain.git
synced 2026-08-14 17:02:19 +00:00
Compare commits
9
Commits
master
...
garrytan/ship
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4f582bfa2c | ||
|
|
e1963e14b4 | ||
|
|
33598d1af9 | ||
|
|
4b17cfed68 | ||
|
|
0441d199b1 | ||
|
|
0ef4b3f593 | ||
|
|
3671e2d0ee | ||
|
|
c06803284f | ||
|
|
f6797f1437 |
+173
@@ -2,6 +2,179 @@
|
||||
|
||||
All notable changes to GBrain will be documented in this file.
|
||||
|
||||
## [0.41.21.0] - 2026-05-27
|
||||
|
||||
**Five daily-driver ops pains, fixed in one wave. Your big brains stop
|
||||
silently wedging, you can see what the cycle is doing instead of
|
||||
guessing, and the 10-hour mention scan now resumes instead of restarting
|
||||
from zero.**
|
||||
|
||||
If you run a 100K+ page brain you probably hit at least three of these
|
||||
this week. The cycle hung for ten minutes printing nothing, so you
|
||||
checked the database manually to see if it was alive. A worker crashed
|
||||
mid-phase and the lock held for 30 minutes before another worker could
|
||||
take over, so you cleared it by hand. Your mention scan died at 87% and
|
||||
you had to restart it from page 0. Your cron ran two separate sync
|
||||
entries because you didn't know `sync --all --parallel` existed. And the
|
||||
extract_atoms phase burned 5 to 10 minutes per cycle on a sequence of
|
||||
7,000 SQL roundtrips before it even started extracting anything. All five
|
||||
get fixed in this release.
|
||||
|
||||
## To take advantage of v0.41.21.0
|
||||
|
||||
`gbrain upgrade` should pick this up automatically. Migration v104 adds a
|
||||
partial expression index on `pages.frontmatter->>'source_hash'` for atom
|
||||
rows. On Postgres it builds with `CREATE INDEX CONCURRENTLY` so no
|
||||
table-level lock; PGLite uses plain `CREATE INDEX`. On a 100K-page brain
|
||||
the index takes seconds to build.
|
||||
|
||||
1. **Confirm the migration applied:**
|
||||
```bash
|
||||
gbrain doctor --json | jq '.checks[] | select(.name=="schema_version")'
|
||||
```
|
||||
2. **Confirm extract_atoms got fast:**
|
||||
```bash
|
||||
time gbrain dream --phase extract_atoms --dry-run --json
|
||||
```
|
||||
The idempotency check phase should finish in under a second instead
|
||||
of taking 5 to 10 minutes.
|
||||
3. **Multi-source brains: pick up the new doctor nudge:**
|
||||
```bash
|
||||
gbrain doctor --json | jq '.checks[] | select(.name=="sync_consolidation")'
|
||||
```
|
||||
You'll see the paste-ready cron line for `sync --all --parallel`.
|
||||
|
||||
If any step fails or the numbers look wrong, file an issue at
|
||||
https://github.com/garrytan/gbrain/issues with the output of `gbrain
|
||||
doctor` and `~/.gbrain/upgrade-errors.jsonl` if it exists.
|
||||
|
||||
### What you'd see in a concrete example
|
||||
|
||||
| Pain | Before | After |
|
||||
|---|---|---|
|
||||
| `extract_atoms` startup on 7K transcripts | 5-10 min of silent overhead | <1 s, then real work |
|
||||
| `extract_atoms` mid-run feedback | "start" then silence for 10+ min | tick every ~1s with running atom count |
|
||||
| `synthesize_concepts` mid-run feedback | "start" then silence | tick every ~1s with concept count |
|
||||
| Crashed cycle lock recovery | 30 min wait, often manual `gbrain sync --break-lock` | <5 min, no manual intervention |
|
||||
| `by-mention` resume after kill at 87% | re-scan 280K of 322K pages | resume from where you stopped |
|
||||
| Multi-source cron setup | two staggered per-source entries | one `sync --all --parallel 4` line |
|
||||
|
||||
### Things to watch
|
||||
|
||||
- **Lock TTL behavior changed (30 min → 5 min).** Cron-side
|
||||
`gbrain sync --break-lock --max-age 1800` scripts that assumed the
|
||||
old 30-min TTL still work, but the number is now larger than the
|
||||
default TTL itself. Anyone who explicitly set `--max-age` against the
|
||||
old TTL should drop the value to match the new shorter window.
|
||||
- **`by-mention --dry-run` no longer claims to be resumable.** Dry-run
|
||||
intentionally skips both the checkpoint load and write so it stays an
|
||||
inspection mode. To exercise the resume path you'll need a real run.
|
||||
- **One residual silent-failure window** under the new shorter TTL: if
|
||||
a single `await chat()` call sits past 5 min wallclock, the lock can
|
||||
expire mid-await without the original phase noticing. This is the
|
||||
same silent-overwrite risk that existed before the wave, just on a
|
||||
shorter timescale. Lock-loss detection is filed as a P2 follow-up
|
||||
TODO (`DbLockHandle.refresh()` will throw on 0 rows affected, phases
|
||||
catch + abort cleanly).
|
||||
|
||||
### Itemized changes
|
||||
|
||||
#### Added
|
||||
- `atomsExistingForHashes(engine, sourceId, hashes[])` exported from
|
||||
`src/core/cycle/extract-atoms.ts` — one batched SQL roundtrip that
|
||||
returns the set of `content_hash16` values already extracted as atoms
|
||||
for this source. Replaces the prior per-hash loop that did 7K
|
||||
individual queries on big brains. Fail-open: an SQL error logs to
|
||||
stderr and returns an empty set so extraction proceeds.
|
||||
- `progress?: ProgressReporter` opt on `ExtractAtomsOpts` and
|
||||
`SynthesizeConceptsOpts`. Cycle.ts now passes its phase-level reporter
|
||||
down (NOT a child reporter — that would produce a path collision
|
||||
`cycle.extract_atoms.extract_atoms.work`). Phases only call `tick()`
|
||||
and `heartbeat()`; cycle.ts owns `start()` and `finish()`. You see
|
||||
`[cycle.extract_atoms] N (atoms_created)` ticks every ~1s during both
|
||||
long phases.
|
||||
- `yieldDuringPhase?: () => Promise<void>` opt on `ExtractAtomsOpts`
|
||||
(and `synthesize_concepts` finally wires the existing one).
|
||||
Cycle.ts builds a `buildYieldDuringPhase(lock, outer)` closure
|
||||
(also exported for tests) that calls `lock.refresh()` AND any
|
||||
external hook on every fire. Throttled to 30s inside each phase via
|
||||
`maybeYield`. Fires both inside the main work loop AND immediately
|
||||
after every `await chat(...)` LLM call so long Haiku/Sonnet calls
|
||||
don't sit past TTL.
|
||||
- `mentionsFingerprint({source, type, since, gazetteerHash})` in
|
||||
`src/core/op-checkpoint.ts`. The gazetteer hash is the load-bearing
|
||||
field — adding new entity pages mid-pause shifts the hash, gets a
|
||||
new fingerprint, and triggers a fresh scan against the new gazetteer
|
||||
instead of silently skipping previously-scanned pages.
|
||||
- `gbrain extract links --by-mention` now resumes from where it died.
|
||||
Wired through the existing `op_checkpoints` framework with a
|
||||
`flushAndCheckpoint` ordering — links flush to the DB FIRST, page
|
||||
keys commit to the checkpoint SECOND, persist THIRD. A crash between
|
||||
`batch.push()` and the flush leaves the page un-checkpointed so
|
||||
resume re-scans it. Persist cadence: every 1000 items OR every 30s,
|
||||
whichever first. Clean exit clears the checkpoint.
|
||||
- `sync_consolidation` doctor check. Multi-source brains see a
|
||||
paste-ready `gbrain sync --all --parallel 4 --workers 4
|
||||
--skip-failed` recommendation. Single-source brains get
|
||||
"not applicable." SQL errors return `warn` via the check's own
|
||||
try/catch — outer doctor catch isn't a safe assumption.
|
||||
- "Multi-source brains" recipe block in
|
||||
`skills/cron-scheduler/SKILL.md` documenting the `sync --all`
|
||||
pattern as preferred over per-source entries.
|
||||
- Migration v104 `pages_atom_source_hash_idx` — partial expression
|
||||
index on `frontmatter->>'source_hash'` for atom rows where
|
||||
`deleted_at IS NULL`. Postgres uses `CREATE INDEX CONCURRENTLY` with
|
||||
invalid-remnant pre-drop (mirrors v97 `pages_dedup_partial_index`);
|
||||
PGLite uses plain `CREATE INDEX`. Without this, the new batch
|
||||
idempotency check would seq-scan the pages table on big brains and
|
||||
defeat the perf win.
|
||||
|
||||
#### Changed
|
||||
- Cycle lock TTL dropped from 30 min to 5 min
|
||||
(`src/core/cycle.ts:LOCK_TTL_MINUTES`). Combined with active
|
||||
in-phase `lock.refresh()` via `buildYieldDuringPhase`, a healthy
|
||||
long-running cycle keeps the lock alive while a crashed cycle
|
||||
releases it 6x faster.
|
||||
- `synthesize_concepts` no longer fires `yieldDuringPhase` per-concept-
|
||||
group. Same hook, throttled to 30s via the new shared `maybeYield`
|
||||
helper — matches the actual lock-refresh budget instead of spamming
|
||||
hundreds of redundant fires per phase.
|
||||
|
||||
#### Fixed
|
||||
- The 7K-roundtrip overhead at the start of every `extract_atoms` cycle
|
||||
on brains with conversation-transcript corpora.
|
||||
- The 30-min wait after a crashed cycle before another worker could
|
||||
acquire the lock.
|
||||
- The 10+ hour `by-mention` sweep restarting from page 0 every time it
|
||||
got interrupted.
|
||||
- Two correctness bugs in the original by-mention checkpoint design
|
||||
that the codex review caught before merge: lost links if a crash
|
||||
landed between `batch.push()` and `flush()`, and silent-miss-on-new-
|
||||
entities if the gazetteer changed between paused runs. The fix
|
||||
flushes links before committing the checkpoint and folds the
|
||||
gazetteer hash into the fingerprint.
|
||||
- Multi-source brains seeing two separate cron entries with manual
|
||||
staggering instead of one `sync --all --parallel` line.
|
||||
|
||||
### For contributors
|
||||
|
||||
- 44 new unit/PGLite tests across 9 files pinning every contract:
|
||||
- `test/cycle/extract-atoms-batch.test.ts` (5 cases) — batch idempotency
|
||||
- `test/cycle/cycle-lock-ttl.test.ts` (1 case) — regression pin on `LOCK_TTL_MINUTES === 5`
|
||||
- `test/op-checkpoint-mentions-fingerprint.test.ts` (7 cases) — fingerprint sensitivity including gazetteer-hash regression guard
|
||||
- `test/cycle/extract-atoms-progress.test.ts` (4 cases) — phase doesn't call start/finish, ticks fire per item
|
||||
- `test/cycle/synthesize-concepts-progress.test.ts` (3 cases) — same shape
|
||||
- `test/cycle/yield-during-phase-refresh.test.ts` (7 cases) — buildYieldDuringPhase actually calls lock.refresh() + outer hook, throws non-fatal
|
||||
- `test/cycle/yield-during-phase-throttle.test.ts` (3 cases) — 30s throttle gate behavior
|
||||
- `test/extract-by-mention-resume.test.ts` (5 cases) — checkpoint persistence ordering, dry-run skips persist, gazetteer change invalidates, filtered pages get checkpointed
|
||||
- `test/doctor-sync-consolidation.test.ts` (6 cases) — edge case matrix for source counts + archived filtering + SQL error path
|
||||
- `LockHandle` and `buildYieldDuringPhase` exported from
|
||||
`src/core/cycle.ts` for test seam access.
|
||||
- Two new follow-up TODOs filed in `TODOS.md` under
|
||||
"v0.41.21.0 ops-fix-wave follow-ups": `gbrain sync print-cron`
|
||||
subcommand and lock-loss detection (extending
|
||||
`DbLockHandle.refresh()` to throw on 0 rows affected).
|
||||
|
||||
## [0.41.20.0] - 2026-05-26
|
||||
|
||||
**One command tells you if your brain is healthy. And `gbrain doctor`
|
||||
|
||||
@@ -1,6 +1,29 @@
|
||||
# TODOS
|
||||
|
||||
## v0.41.19.0 status + doctor-categories wave follow-ups (v0.42+)
|
||||
## v0.41.21.0 ops-fix-wave follow-ups (v0.41.22+)
|
||||
|
||||
- **TODO-OPS-1 (P2)**: `gbrain sync print-cron` subcommand. Print the canonical
|
||||
cron line based on the active source set: `gbrain sync --all --parallel N
|
||||
--workers N --skip-failed` where N defaults to `min(sourceCount, 4)`. Reads
|
||||
`sources` table for active (non-archived, `local_path IS NOT NULL`) entries.
|
||||
Ergonomic upgrade over the v0.41.19.0 `sync_consolidation` doctor message —
|
||||
operator pipes directly into `crontab -e` instead of copy-paste-massage.
|
||||
~80 LOC. Mirrors `gbrain sync --break-lock` argv shape.
|
||||
|
||||
- **TODO-OPS-2 (P2)**: Lock-loss detection — extend `DbLockHandle.refresh()`
|
||||
to throw `LockLostError` on 0 rows affected. Codex caught during the
|
||||
v0.41.19.0 plan review: `refresh()` runs `UPDATE ... WHERE holder_pid = pid`
|
||||
with no rows-affected check (`db-lock.ts:108-114`, `:151-156`). If the
|
||||
TTL expired and another worker took over, the original keeps writing
|
||||
silently. v0.41.19.0 ships TTL=5min + active in-phase refresh via
|
||||
`buildYieldDuringPhase` which makes the race window much narrower, but
|
||||
an `await chat()` call that exceeds the 5min wallclock window can still
|
||||
hit it. Fix: `RETURNING id` on the UPDATE + check `rows.length === 0` →
|
||||
throw tagged `LockLostError`. Phases catch + abort cleanly (write partial
|
||||
progress, return `status: 'fail'` with reason `'lock_lost'`). Behavioral
|
||||
contract change with phase-abort fallout; needs its own design pass.
|
||||
|
||||
## v0.41.20.0 status + doctor-categories wave follow-ups (v0.42+)
|
||||
|
||||
- **TODO-V19-A (P3)**: Persistent `cycle_runs` table. v0.41.19.0 infers
|
||||
"last full cycle" by querying `minion_jobs WHERE name = 'autopilot-cycle'`
|
||||
|
||||
File diff suppressed because one or more lines are too long
+1
-1
@@ -140,5 +140,5 @@
|
||||
"bun": ">=1.3.10"
|
||||
},
|
||||
"license": "MIT",
|
||||
"version": "0.41.20.0"
|
||||
"version": "0.41.21.0"
|
||||
}
|
||||
|
||||
@@ -53,6 +53,34 @@ Every cron job MUST be idempotent:
|
||||
|
||||
Job configuration saved. Report: "Job '{name}' scheduled at {cron expression}. Next run: {time}."
|
||||
|
||||
## Multi-source brains: use `sync --all`, not per-source entries
|
||||
|
||||
When the brain has 2+ active sources (anything `gbrain sources list` shows
|
||||
with a non-null `local_path` that isn't archived), use one consolidated
|
||||
cron line instead of N per-source entries.
|
||||
|
||||
**Preferred (multi-source)**:
|
||||
|
||||
```cron
|
||||
*/5 * * * * gbrain sync --all --parallel 4 --workers 4 --skip-failed
|
||||
```
|
||||
|
||||
This replaces N per-source lines AND auto-picks-up future sources without
|
||||
a crontab edit. Concurrency budget: `parallel × workers × 2 ≈ 32`
|
||||
connections during the wave (each per-file worker opens its own
|
||||
2-connection pool). Stay under your Postgres `max_connections` setting.
|
||||
|
||||
**Avoid (legacy)**: separate `gbrain sync --source default` and
|
||||
`gbrain sync --source zion-brain` entries staggered by 5 minutes. They
|
||||
require manual deconfliction every time a new source is added, and a
|
||||
slow source can race a fast source on the legacy global `gbrain-sync`
|
||||
lock (v0.40.3.0+ uses per-source `gbrain-sync:<sourceId>` locks but the
|
||||
per-source cron pattern doesn't benefit from the parallelism that
|
||||
`--all --parallel` actually delivers).
|
||||
|
||||
`gbrain doctor` surfaces the recommended line as a `sync_consolidation`
|
||||
check whenever it detects 2+ active sources. Paste-ready from there.
|
||||
|
||||
## Anti-Patterns
|
||||
|
||||
- Scheduling jobs at the same minute (:00 for everything)
|
||||
@@ -60,3 +88,6 @@ Job configuration saved. Report: "Job '{name}' scheduled at {cron expression}. N
|
||||
- Running cron jobs without testing on 3-5 items first
|
||||
- Jobs that produce different output on re-run (not idempotent)
|
||||
- Sending notifications during quiet hours (save to held queue instead)
|
||||
- Separate per-source `gbrain sync --source <id>` cron entries when
|
||||
`gbrain sync --all --parallel N --workers N` would replace them with
|
||||
one line that auto-picks-up future sources.
|
||||
|
||||
@@ -661,6 +661,9 @@ export async function doctorReportRemote(engine: BrainEngine): Promise<DoctorRep
|
||||
// 6. Sync freshness check
|
||||
checks.push(await checkSyncFreshness(engine));
|
||||
|
||||
// v0.41.19.0 (Issue 5): sync --all consolidation nudge for multi-source brains.
|
||||
checks.push(await checkSyncConsolidation(engine));
|
||||
|
||||
// v0.39 T7 + T9 — schema-pack health checks (3 checks per v0.38 plan):
|
||||
// schema_pack_active — active pack resolves cleanly
|
||||
// schema_pack_consistency — % of pages typed against active pack
|
||||
@@ -2598,6 +2601,60 @@ export async function checkSyncFreshness(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.41.19.0 (Issue 5 of ops-fix-wave) — surface `sync --all --parallel`
|
||||
* to operators with multi-source brains.
|
||||
*
|
||||
* Background: `gbrain sync --all --parallel N --workers N --skip-failed`
|
||||
* has existed since v0.40.3.0 but most operators still maintain separate
|
||||
* per-source cron entries with manual deconfliction. One `--all` line
|
||||
* replaces N per-source lines AND auto-picks-up future sources without
|
||||
* a crontab edit.
|
||||
*
|
||||
* Surgical scope: we can't reach into the user's crontab (host-specific,
|
||||
* portability risk). What we CAN do is surface the paste-ready command
|
||||
* inside `gbrain doctor` so the operator sees it whenever they run a
|
||||
* health check on a multi-source brain.
|
||||
*
|
||||
* Posture: never failure-state. Always `ok` with the paste-ready cmd
|
||||
* embedded in the message (matches how sync_freshness embeds fix hints).
|
||||
* Single-source brains get `ok` with a "not applicable" message.
|
||||
* SQL error → `warn` (own try/catch, not relying on the outer doctor
|
||||
* dispatcher — codex flagged this).
|
||||
*/
|
||||
export async function checkSyncConsolidation(engine: BrainEngine): Promise<Check> {
|
||||
try {
|
||||
const rows = await engine.executeRaw<{ id: string }>(
|
||||
`SELECT id FROM sources
|
||||
WHERE archived IS NOT TRUE
|
||||
AND local_path IS NOT NULL`,
|
||||
);
|
||||
const sourceCount = rows.length;
|
||||
if (sourceCount < 2) {
|
||||
return {
|
||||
name: 'sync_consolidation',
|
||||
status: 'ok',
|
||||
message: 'Single-source brain — sync --all consolidation not applicable.',
|
||||
};
|
||||
}
|
||||
return {
|
||||
name: 'sync_consolidation',
|
||||
status: 'ok',
|
||||
message:
|
||||
`${sourceCount} active sources detected. Recommended cron: ` +
|
||||
'`gbrain sync --all --parallel 4 --workers 4 --skip-failed`. ' +
|
||||
'If your crontab has separate per-source entries, replace them with one --all line — ' +
|
||||
'future sources auto-pick-up without a crontab edit.',
|
||||
};
|
||||
} catch (err) {
|
||||
return {
|
||||
name: 'sync_consolidation',
|
||||
status: 'warn',
|
||||
message: `Could not check sync consolidation: ${err instanceof Error ? err.message : String(err)}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.38 — per-source `last_full_cycle_at` freshness check.
|
||||
*
|
||||
@@ -5473,6 +5530,9 @@ export async function buildChecks(
|
||||
if (engine !== null) {
|
||||
progress.heartbeat('sync_freshness');
|
||||
checks.push(await checkSyncFreshness(engine));
|
||||
// v0.41.19.0 (Issue 5): sync --all consolidation nudge.
|
||||
progress.heartbeat('sync_consolidation');
|
||||
checks.push(await checkSyncConsolidation(engine));
|
||||
// v0.38 — full-cycle freshness, sibling to sync_freshness. Reads
|
||||
// last_full_cycle_at from sources.config; mirrors what autopilot's
|
||||
// per-source dispatch gate sees.
|
||||
|
||||
+112
-10
@@ -51,6 +51,10 @@ import { withRetry, isRetryableConnError } from '../core/retry.ts';
|
||||
export { withRetry };
|
||||
export type { WithRetryOpts } from '../core/retry.ts';
|
||||
import { buildGazetteer, findMentionedEntities } from '../core/by-mention.ts';
|
||||
import {
|
||||
loadOpCheckpoint, recordCompleted, clearOpCheckpoint, mentionsFingerprint,
|
||||
} from '../core/op-checkpoint.ts';
|
||||
import { createHash } from 'crypto';
|
||||
// v0.41.15.0 (T7, D9): --workers N for the fs-walk inner loops via the
|
||||
// shared sliding-pool helper + PGLite-clamp wrapper.
|
||||
import { runSlidingPool } from '../core/worker-pool.ts';
|
||||
@@ -1324,18 +1328,51 @@ async function extractMentionsFromDb(
|
||||
return { created: 0, pages: 0 };
|
||||
}
|
||||
|
||||
// v0.41.19.0 (T5): gazetteer hash is part of the checkpoint
|
||||
// fingerprint so adding new entity pages mid-pause invalidates the
|
||||
// checkpoint cleanly. Without it, resumed pages would skip new
|
||||
// entities silently (codex flag).
|
||||
const gazetteerHash = createHash('sha256')
|
||||
.update([...gazetteer.keys()].sort().join('|'))
|
||||
.digest('hex')
|
||||
.slice(0, 8);
|
||||
|
||||
const allRefs = sourceIdFilter
|
||||
? (await engine.listAllPageRefs()).filter(r => r.source_id === sourceIdFilter)
|
||||
: await engine.listAllPageRefs();
|
||||
|
||||
// v0.41.19.0 (T5): load checkpoint and skip already-completed
|
||||
// (source_id, slug) pairs. Dry-run does NOT load OR persist the
|
||||
// checkpoint — dry-run is an inspection mode and shouldn't pollute
|
||||
// resume state for the next non-dry-run.
|
||||
const ckptKey = {
|
||||
op: 'extract-by-mention',
|
||||
fingerprint: mentionsFingerprint({
|
||||
source: sourceIdFilter,
|
||||
type: typeFilter,
|
||||
since,
|
||||
gazetteerHash,
|
||||
}),
|
||||
};
|
||||
const completed = dryRun
|
||||
? new Set<string>()
|
||||
: new Set(await loadOpCheckpoint(engine, ckptKey));
|
||||
const remaining = completed.size > 0
|
||||
? allRefs.filter(r => !completed.has(`${r.source_id}::${r.slug}`))
|
||||
: allRefs;
|
||||
|
||||
if (completed.size > 0 && !jsonMode) {
|
||||
console.log(`[by-mention] resuming: ${completed.size}/${allRefs.length} pages already scanned, ${remaining.length} remaining`);
|
||||
}
|
||||
|
||||
let processed = 0;
|
||||
let created = 0;
|
||||
const batch: LinkBatchInput[] = [];
|
||||
|
||||
const progress = createProgress(cliOptsToProgressOptions(getCliOptions()));
|
||||
progress.start('extract.by_mention.scan', allRefs.length);
|
||||
progress.start('extract.by_mention.scan', remaining.length);
|
||||
|
||||
async function flush() {
|
||||
async function flushBatch() {
|
||||
if (batch.length === 0) return;
|
||||
try {
|
||||
created += await engine.addLinksBatch(batch, { auditSite: 'extract.by_mention' }); // gbrain-allow-direct-insert: gbrain extract --by-mention — canonical auto-link write from body-text mention scan
|
||||
@@ -1351,15 +1388,54 @@ async function extractMentionsFromDb(
|
||||
}
|
||||
}
|
||||
|
||||
// v0.41.19.0 (T5 — codex fix #1): flush links FIRST, commit pending
|
||||
// page keys to checkpoint SECOND, persist THIRD. A crash between
|
||||
// batch.push() and flushBatch() leaves pendingForFlush uncommitted —
|
||||
// resume re-scans those pages instead of silently losing their links.
|
||||
//
|
||||
// Persist cadence: every 1000 items OR every 30s, whichever first
|
||||
// (~322 persists on a 322K-page brain, ~24s total overhead). Crash
|
||||
// window is at most 1000 pages (<0.3% loss on the driver brain).
|
||||
const PERSIST_EVERY_N = 1000;
|
||||
const PERSIST_EVERY_MS = 30_000;
|
||||
const pendingForFlush: string[] = [];
|
||||
let sinceLastPersistMs = Date.now();
|
||||
let unpersistedCount = 0;
|
||||
|
||||
async function flushAndCheckpoint(force = false): Promise<void> {
|
||||
await flushBatch();
|
||||
for (const key of pendingForFlush) completed.add(key);
|
||||
pendingForFlush.length = 0;
|
||||
if (dryRun) return;
|
||||
const now = Date.now();
|
||||
if (force || unpersistedCount >= PERSIST_EVERY_N || (now - sinceLastPersistMs) >= PERSIST_EVERY_MS) {
|
||||
await recordCompleted(engine, ckptKey, [...completed]);
|
||||
unpersistedCount = 0;
|
||||
sinceLastPersistMs = now;
|
||||
}
|
||||
}
|
||||
|
||||
const sinceMs = since ? new Date(since).getTime() : null;
|
||||
|
||||
for (const { slug, source_id } of allRefs) {
|
||||
for (const { slug, source_id } of remaining) {
|
||||
const page = await engine.getPage(slug, { sourceId: source_id });
|
||||
if (!page) continue;
|
||||
if (typeFilter && page.type !== typeFilter) continue;
|
||||
// v0.41.19.0 (T5 — codex fix #4): even when we skip a page (filter
|
||||
// miss, missing row, empty body, no mentions), MARK IT COMPLETED so
|
||||
// resume doesn't re-fetch it. The decision NOT to create links is
|
||||
// itself a completed decision.
|
||||
const key = `${source_id}::${slug}`;
|
||||
if (!page || (typeFilter && page.type !== typeFilter)) {
|
||||
pendingForFlush.push(key);
|
||||
unpersistedCount++;
|
||||
continue;
|
||||
}
|
||||
if (sinceMs !== null) {
|
||||
const updatedMs = new Date(page.updated_at).getTime();
|
||||
if (Number.isFinite(updatedMs) && updatedMs <= sinceMs) continue;
|
||||
if (Number.isFinite(updatedMs) && updatedMs <= sinceMs) {
|
||||
pendingForFlush.push(key);
|
||||
unpersistedCount++;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
processed++;
|
||||
progress.tick();
|
||||
@@ -1368,14 +1444,22 @@ async function extractMentionsFromDb(
|
||||
// end-of-compiled token doesn't accidentally merge with a
|
||||
// start-of-timeline token into a false phrase match.
|
||||
const body = page.compiled_truth + '\n\n' + (page.timeline ?? '');
|
||||
if (!body.trim()) continue;
|
||||
if (!body.trim()) {
|
||||
pendingForFlush.push(key);
|
||||
unpersistedCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
const mentions = findMentionedEntities(body, gazetteer, {
|
||||
fromSlug: slug,
|
||||
fromSourceId: source_id,
|
||||
});
|
||||
|
||||
if (mentions.length === 0) continue;
|
||||
if (mentions.length === 0) {
|
||||
pendingForFlush.push(key);
|
||||
unpersistedCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
for (const m of mentions) {
|
||||
if (dryRun) {
|
||||
@@ -1399,14 +1483,32 @@ async function extractMentionsFromDb(
|
||||
from_source_id: source_id,
|
||||
to_source_id: m.source_id,
|
||||
});
|
||||
if (batch.length >= BATCH_SIZE) await flush();
|
||||
if (batch.length >= BATCH_SIZE) {
|
||||
// The page that produced these batch entries stays UN-committed
|
||||
// until flushBatch succeeds. The push below happens AFTER the
|
||||
// flushAndCheckpoint call so a crash inside flushBatch leaves
|
||||
// the page un-checkpointed and resume re-scans it.
|
||||
await flushAndCheckpoint();
|
||||
}
|
||||
}
|
||||
}
|
||||
// Page completed (whether dry-run or non-dry-run). Stage for the
|
||||
// next flushAndCheckpoint().
|
||||
pendingForFlush.push(key);
|
||||
unpersistedCount++;
|
||||
// Time-based cadence floor.
|
||||
if (!dryRun && (Date.now() - sinceLastPersistMs) >= PERSIST_EVERY_MS) {
|
||||
await flushAndCheckpoint();
|
||||
}
|
||||
}
|
||||
|
||||
if (!dryRun) await flush();
|
||||
if (!dryRun) {
|
||||
await flushAndCheckpoint(true); // final flush + force-persist
|
||||
}
|
||||
progress.finish();
|
||||
|
||||
if (!dryRun) await clearOpCheckpoint(engine, ckptKey); // clean exit
|
||||
|
||||
if (!jsonMode) {
|
||||
const label = dryRun ? '(dry run) would create' : 'created';
|
||||
console.log(`Mentions: ${label} ${created} links from ${processed} pages against gazetteer of ${gazetteer.size} first-token buckets`);
|
||||
|
||||
+66
-4
@@ -422,12 +422,19 @@ export interface CycleOpts {
|
||||
* time use this row in `gbrain_cycle_locks`.
|
||||
*/
|
||||
const LEGACY_CYCLE_LOCK_ID = 'gbrain-cycle';
|
||||
const LOCK_TTL_MS = 30 * 60 * 1000; // 30 minutes
|
||||
const LOCK_TTL_MINUTES = 30; // db-lock.ts takes minutes
|
||||
// v0.41.19.0 (T2 of ops-fix-wave): dropped from 30 min to 5 min so a
|
||||
// crashed cycle releases the lock within 5 min instead of holding it for
|
||||
// the full 30-min TTL. Wired with active in-phase refresh via
|
||||
// `buildYieldDuringPhase` (T3) — the closure passed to long phases as
|
||||
// `yieldDuringPhase` calls `lock.refresh()` every 30s, so a healthy
|
||||
// long-running cycle keeps the TTL alive while the shorter window
|
||||
// shrinks crash recovery 6×.
|
||||
const LOCK_TTL_MS = 5 * 60 * 1000; // 5 minutes (was 30)
|
||||
const LOCK_TTL_MINUTES = 5; // was 30; db-lock.ts takes minutes
|
||||
// Lazy: GBRAIN_HOME may be set after module load; resolve at call time.
|
||||
const getLockFilePathDefault = () => gbrainPath('cycle.lock');
|
||||
|
||||
interface LockHandle {
|
||||
export interface LockHandle {
|
||||
release: () => Promise<void>;
|
||||
refresh: () => Promise<void>;
|
||||
}
|
||||
@@ -560,6 +567,53 @@ function acquireFileLock(lockPath = getLockFilePathDefault()): LockHandle | null
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.41.19.0 (T3 of ops-fix-wave): build the closure that long phases
|
||||
* call to keep the cycle DB lock alive AND fire the existing cooperative
|
||||
* yield hook (Minion job-lock renewal in jobs.ts / autopilot.ts).
|
||||
*
|
||||
* Codex caught that the prior `yieldBetweenPhases` opt does NOT refresh
|
||||
* the cycle lock — it's just a `setImmediate()` from external callers,
|
||||
* and `lock.refresh()` was only ever called via the implicit final
|
||||
* `release()` path. Combined with the TTL drop 30→5min (T2), a long
|
||||
* phase like `extract_atoms` or `synthesize_concepts` would lose the
|
||||
* lock to a competing worker mid-phase.
|
||||
*
|
||||
* The returned closure does TWO things on each fire:
|
||||
* 1. `await lock.refresh()` to bump `ttl_expires_at` + `last_refreshed_at`
|
||||
* 2. `await outer()` to renew any external job-lock the caller threaded in
|
||||
*
|
||||
* Both are wrapped in try/catch — a refresh failure logs to stderr but
|
||||
* doesn't crash the phase (if the lock was truly stolen, we want this
|
||||
* run to wind down gracefully, not throw mid-LLM-call).
|
||||
*
|
||||
* Returns `undefined` when there's no lock AND no outer hook so phases
|
||||
* short-circuit via their `if (!opts.yieldDuringPhase) return;` guard.
|
||||
*/
|
||||
export function buildYieldDuringPhase(
|
||||
lock: LockHandle | null,
|
||||
outer?: () => Promise<void>,
|
||||
): (() => Promise<void>) | undefined {
|
||||
if (!lock && !outer) return undefined;
|
||||
return async () => {
|
||||
if (lock) {
|
||||
try {
|
||||
await lock.refresh();
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
// Non-fatal: a refresh error doesn't crash the phase. If the
|
||||
// lock truly expired and was stolen, the next acquire by another
|
||||
// worker has already happened — let this run wind down rather
|
||||
// than throw mid-phase.
|
||||
console.error(`[cycle] lock refresh failed (non-fatal): ${msg}`);
|
||||
}
|
||||
}
|
||||
if (outer) {
|
||||
try { await outer(); } catch { /* outer hook errors are not fatal */ }
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
// ─── Helpers ───────────────────────────────────────────────────────
|
||||
|
||||
function makeErrorFromException(e: unknown, fallbackClass = 'InternalError'): PhaseError {
|
||||
@@ -1553,6 +1607,11 @@ export async function runCycle(
|
||||
sourceId: xaSourceId,
|
||||
dryRun,
|
||||
affectedSlugs: xaAffectedSlugs,
|
||||
// v0.41.19.0 (T3): closure refreshes cycle lock + fires outer hook.
|
||||
yieldDuringPhase: buildYieldDuringPhase(lock, opts.yieldDuringPhase),
|
||||
// v0.41.19.0 (T4): pass same reporter (not a child — cycle.ts
|
||||
// owns start/finish; phase only ticks).
|
||||
progress,
|
||||
}));
|
||||
result.duration_ms = duration_ms;
|
||||
phaseResults.push(result);
|
||||
@@ -1643,7 +1702,10 @@ export async function runCycle(
|
||||
const { result, duration_ms } = await timePhase(() => runPhaseSynthesizeConcepts(engine, {
|
||||
brainDir: opts.brainDir,
|
||||
dryRun,
|
||||
yieldDuringPhase: opts.yieldDuringPhase,
|
||||
// v0.41.19.0 (T3): closure refreshes cycle lock + fires outer hook.
|
||||
yieldDuringPhase: buildYieldDuringPhase(lock, opts.yieldDuringPhase),
|
||||
// v0.41.19.0 (T4): pass same reporter (not a child).
|
||||
progress,
|
||||
}));
|
||||
result.duration_ms = duration_ms;
|
||||
phaseResults.push(result);
|
||||
|
||||
@@ -47,6 +47,7 @@
|
||||
import type { BrainEngine } from '../engine.ts';
|
||||
import type { PhaseResult } from '../cycle.ts';
|
||||
import type { GBrainConfig } from '../config.ts';
|
||||
import type { ProgressReporter } from '../progress.ts';
|
||||
import { chat as gatewayChat } from '../ai/gateway.ts';
|
||||
|
||||
const DEFAULT_BUDGET_USD = 0.3;
|
||||
@@ -88,6 +89,25 @@ export interface ExtractAtomsOpts {
|
||||
* explicitly suppresses page discovery (for transcript-only tests).
|
||||
*/
|
||||
_pages?: Array<{ slug: string; content: string; contentHash: string }>;
|
||||
/**
|
||||
* v0.41.19.0 (T3): cooperative yield hook fired from inside the work
|
||||
* loop on a 30s throttle AND immediately after every `await chat()`
|
||||
* LLM call. Cycle.ts threads `buildYieldDuringPhase(lock, outer)` so
|
||||
* each fire refreshes the cycle DB lock + the existing external hook
|
||||
* (Minion job-lock renewal). Without it a long phase loses the lock
|
||||
* after the v0.41.19.0 TTL drop 30→5min.
|
||||
*/
|
||||
yieldDuringPhase?: () => Promise<void>;
|
||||
/**
|
||||
* v0.41.19.0 (T4): progress reporter for in-phase ticks. Cycle.ts
|
||||
* passes the SAME reporter (not a child — codex caught the path-
|
||||
* collision bug where `progress.child('extract_atoms')` under parent
|
||||
* state `cycle.extract_atoms` would produce
|
||||
* `cycle.extract_atoms.extract_atoms.work`). Cycle.ts owns the
|
||||
* phase-level start/finish; phases only call `tick()` and
|
||||
* `heartbeat()` on the passed reporter.
|
||||
*/
|
||||
progress?: ProgressReporter;
|
||||
}
|
||||
|
||||
interface ExtractedAtom {
|
||||
@@ -198,36 +218,42 @@ export async function discoverExtractablePages(
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.41.2.1 — Source-hash idempotency check (D1). Returns true if ANY
|
||||
* atom row exists for the (sourceId, contentHash16) pair.
|
||||
* Batch source-hash idempotency check. Returns the set of contentHash16
|
||||
* values that already have an atom row for this source. One SQL
|
||||
* roundtrip; migration v104 adds the partial expression index that
|
||||
* keeps this O(log n) on big brains.
|
||||
*
|
||||
* Used by the transcript path to close the pre-existing date-stamp
|
||||
* duplicate bug. Page-side idempotency is folded into the discovery
|
||||
* SQL's NOT EXISTS subquery — this helper is just for transcripts
|
||||
* which don't go through that query.
|
||||
* Replaces the prior per-hash helper (`atomsExistForHash`) — for ~7K
|
||||
* conversation transcripts the per-hash loop was 7K round trips before
|
||||
* extraction began (~5-10 min of pure overhead on a 322K-page brain).
|
||||
*
|
||||
* Empty input short-circuits without a query. Fail-open on error so
|
||||
* extraction proceeds (same posture as the prior per-hash helper).
|
||||
*
|
||||
* Exported so the unit test can drive it directly without orchestrating
|
||||
* the full phase.
|
||||
*/
|
||||
async function atomsExistForHash(
|
||||
export async function atomsExistingForHashes(
|
||||
engine: BrainEngine,
|
||||
sourceId: string,
|
||||
contentHash16: string,
|
||||
): Promise<boolean> {
|
||||
contentHash16s: string[],
|
||||
): Promise<Set<string>> {
|
||||
if (contentHash16s.length === 0) return new Set();
|
||||
try {
|
||||
const rows = await engine.executeRaw<{ existing: number }>(
|
||||
`SELECT 1 AS existing FROM pages
|
||||
const rows = await engine.executeRaw<{ h: string }>(
|
||||
`SELECT frontmatter->>'source_hash' AS h
|
||||
FROM pages
|
||||
WHERE type = 'atom'
|
||||
AND source_id = $1
|
||||
AND frontmatter->>'source_hash' = $2
|
||||
AND deleted_at IS NULL
|
||||
LIMIT 1`,
|
||||
[sourceId, contentHash16],
|
||||
AND frontmatter->>'source_hash' = ANY($2::text[])`,
|
||||
[sourceId, contentHash16s],
|
||||
);
|
||||
return rows.length > 0;
|
||||
return new Set(rows.map(r => r.h));
|
||||
} catch (err) {
|
||||
// Fail-open: if the check breaks, prefer re-extraction over silent skip.
|
||||
// Cost is bounded by the daily budget cap; correctness wins over LLM cost.
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
console.error(`[extract_atoms] idempotency check failed (assuming not extracted): ${msg}`);
|
||||
return false;
|
||||
console.error(`[extract_atoms] batch idempotency check failed (assuming none extracted): ${msg}`);
|
||||
return new Set();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -289,13 +315,18 @@ export async function runPhaseExtractAtoms(
|
||||
pages = await discoverExtractablePages(engine, sourceId, opts.affectedSlugs);
|
||||
}
|
||||
|
||||
// 2. Apply transcript-side source-hash idempotency (D1 — closes the
|
||||
// pre-existing date-stamp duplicate bug). Page-side idempotency
|
||||
// lives in the discovery SQL's NOT EXISTS subquery.
|
||||
// 2. Apply transcript-side source-hash idempotency in ONE batch query
|
||||
// instead of N per-hash round trips. Page-side idempotency lives in
|
||||
// the discovery SQL's NOT EXISTS subquery (already batched).
|
||||
const transcriptsLive: typeof transcripts = [];
|
||||
let duplicatesSkipped = 0;
|
||||
const allHashes16 = transcripts.map(t => t.contentHash.slice(0, 16));
|
||||
// Surface a heartbeat before the batch query so even an instant
|
||||
// short-circuit shows a sign of life (closes Issue 2 silent-phase pain).
|
||||
opts.progress?.heartbeat(`checking existing atoms for ${allHashes16.length} transcripts`);
|
||||
const existingHashes = await atomsExistingForHashes(engine, sourceId, allHashes16);
|
||||
for (const t of transcripts) {
|
||||
if (await atomsExistForHash(engine, sourceId, t.contentHash.slice(0, 16))) {
|
||||
if (existingHashes.has(t.contentHash.slice(0, 16))) {
|
||||
duplicatesSkipped++;
|
||||
continue;
|
||||
}
|
||||
@@ -358,7 +389,31 @@ export async function runPhaseExtractAtoms(
|
||||
let estimatedSpendUsd = 0;
|
||||
const budgetCap = DEFAULT_BUDGET_USD;
|
||||
|
||||
// v0.41.19.0 (T3): throttled yield helper. Fires `opts.yieldDuringPhase`
|
||||
// every 30s. Cycle.ts threads `buildYieldDuringPhase(lock, outer)` so
|
||||
// each fire refreshes the cycle DB lock. Combined with TTL=5min: a
|
||||
// healthy long phase keeps the lock alive (10× refresh budget before
|
||||
// TTL expires); a crash releases the lock within 5min instead of 30.
|
||||
//
|
||||
// Called both inside the work loop (cheap iterations) AND immediately
|
||||
// after every `await chat()` (long LLM await is the main TTL hazard
|
||||
// codex flagged).
|
||||
let lastYieldMs = Date.now();
|
||||
async function maybeYield(): Promise<void> {
|
||||
if (!opts.yieldDuringPhase) return;
|
||||
const now = Date.now();
|
||||
if (now - lastYieldMs < 30_000) return;
|
||||
lastYieldMs = now;
|
||||
try {
|
||||
await opts.yieldDuringPhase();
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
console.error(`[extract_atoms] yieldDuringPhase failed (non-fatal): ${msg}`);
|
||||
}
|
||||
}
|
||||
|
||||
for (const item of work) {
|
||||
await maybeYield();
|
||||
if (estimatedSpendUsd >= budgetCap) {
|
||||
if (item.kind === 'transcript') transcriptsSkipped++;
|
||||
else pagesSkipped++;
|
||||
@@ -377,6 +432,10 @@ export async function runPhaseExtractAtoms(
|
||||
],
|
||||
maxTokens: 2000,
|
||||
});
|
||||
// Post-await yield: closes the "long LLM call past TTL" hazard
|
||||
// codex flagged. The 30s throttle inside maybeYield bounds the
|
||||
// actual refresh rate so this is cheap when calls are fast.
|
||||
await maybeYield();
|
||||
|
||||
// Rough cost estimate — Haiku at ~$0.80/M input + $4/M output
|
||||
estimatedSpendUsd +=
|
||||
@@ -428,6 +487,9 @@ export async function runPhaseExtractAtoms(
|
||||
}
|
||||
if (item.kind === 'transcript') transcriptsProcessed++;
|
||||
else pagesProcessed++;
|
||||
// v0.41.19.0 (T4): one tick per processed item, with a count note.
|
||||
// Reporter rate-limits to ~1 line/sec; safe to tick every iter.
|
||||
opts.progress?.tick(1, `${totalAtomsExtracted} atoms / ${duplicatesSkipped} skipped`);
|
||||
} catch (err) {
|
||||
failures.push({
|
||||
source: originLabel,
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
|
||||
import type { BrainEngine } from '../engine.ts';
|
||||
import type { PhaseResult } from '../cycle.ts';
|
||||
import type { ProgressReporter } from '../progress.ts';
|
||||
import { chat as gatewayChat } from '../ai/gateway.ts';
|
||||
|
||||
const DEFAULT_BUDGET_USD = 1.5;
|
||||
@@ -31,6 +32,13 @@ export interface SynthesizeConceptsOpts {
|
||||
brainDir?: string;
|
||||
dryRun?: boolean;
|
||||
yieldDuringPhase?: (() => Promise<void>) | undefined;
|
||||
/**
|
||||
* v0.41.19.0 (T4): progress reporter for in-phase ticks. Cycle.ts
|
||||
* passes the SAME reporter (not a child — see extract-atoms.ts for
|
||||
* the path-collision bug codex caught). Phases only call `tick()` /
|
||||
* `heartbeat()`; cycle.ts owns start/finish.
|
||||
*/
|
||||
progress?: ProgressReporter;
|
||||
/** Test seam: alternative chat function. */
|
||||
_chat?: typeof gatewayChat;
|
||||
/** Test seam: skip DB query; cluster these atoms directly. */
|
||||
@@ -139,6 +147,26 @@ export async function runPhaseSynthesizeConcepts(
|
||||
const failures: Array<{ concept: string; error: string }> = [];
|
||||
const tierCounts = { T1: 0, T2: 0, T3: 0, T4: 0 };
|
||||
|
||||
// v0.41.19.0 (T3): throttled yield helper. Fires `opts.yieldDuringPhase`
|
||||
// every 30s — cycle.ts threads `buildYieldDuringPhase(lock, outer)` so
|
||||
// each fire refreshes the cycle DB lock + the existing external hook.
|
||||
// Pre-v0.41.19 the bare `if (opts.yieldDuringPhase) await ...()` at
|
||||
// every iteration fired hundreds of times per phase; the 30s throttle
|
||||
// matches the actual lock-refresh budget.
|
||||
let lastYieldMs = Date.now();
|
||||
async function maybeYield(): Promise<void> {
|
||||
if (!opts.yieldDuringPhase) return;
|
||||
const now = Date.now();
|
||||
if (now - lastYieldMs < 30_000) return;
|
||||
lastYieldMs = now;
|
||||
try {
|
||||
await opts.yieldDuringPhase();
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
console.error(`[synthesize_concepts] yieldDuringPhase failed (non-fatal): ${msg}`);
|
||||
}
|
||||
}
|
||||
|
||||
for (const group of atomGroups) {
|
||||
tierCounts[group.tier]++;
|
||||
let narrative: string;
|
||||
@@ -164,6 +192,10 @@ export async function runPhaseSynthesizeConcepts(
|
||||
],
|
||||
maxTokens: 500,
|
||||
});
|
||||
// Post-await yield (T3): the LLM call is the main TTL hazard
|
||||
// codex flagged. Throttle inside maybeYield bounds the actual
|
||||
// refresh rate.
|
||||
await maybeYield();
|
||||
// Sonnet at ~$3/M input + $15/M output
|
||||
estimatedSpendUsd +=
|
||||
(result.usage.input_tokens * 3.0 + result.usage.output_tokens * 15.0) / 1_000_000;
|
||||
@@ -198,8 +230,13 @@ export async function runPhaseSynthesizeConcepts(
|
||||
});
|
||||
}
|
||||
conceptsWritten++;
|
||||
// v0.41.19.0 (T4): one tick per concept group with running count.
|
||||
opts.progress?.tick(1, `${conceptsWritten} concepts`);
|
||||
|
||||
if (opts.yieldDuringPhase) await opts.yieldDuringPhase();
|
||||
// v0.41.19.0 (T3): replaced bare per-iteration fire with throttled
|
||||
// helper. Same hook, same cycle-lock refresh effect, just at the
|
||||
// right cadence (30s instead of every-group).
|
||||
await maybeYield();
|
||||
}
|
||||
|
||||
return {
|
||||
|
||||
@@ -140,6 +140,7 @@ export const OPS_CHECK_NAMES: ReadonlySet<string> = new Set([
|
||||
'subagent_capability',
|
||||
'subagent_health',
|
||||
'supervisor',
|
||||
'sync_consolidation',
|
||||
'ze_embedding_health',
|
||||
]);
|
||||
|
||||
|
||||
@@ -4733,6 +4733,51 @@ export const MIGRATIONS: Migration[] = [
|
||||
}
|
||||
},
|
||||
},
|
||||
{
|
||||
version: 104,
|
||||
name: 'pages_atom_source_hash_idx',
|
||||
// Partial expression index on frontmatter->>'source_hash' for atom
|
||||
// rows. Powers `atomsExistingForHashes` in extract_atoms
|
||||
// (src/core/cycle/extract-atoms.ts), which replaces the prior
|
||||
// per-hash loop that did 7K SQL round trips per cycle on a brain
|
||||
// with ~7K conversation transcripts.
|
||||
//
|
||||
// Mirrors v97 pattern: Postgres uses CREATE INDEX CONCURRENTLY
|
||||
// (no SHARE-lock blocking concurrent writes) and pre-drops any
|
||||
// invalid remnant from a prior failed CONCURRENTLY attempt via
|
||||
// pg_index.indisvalid. PGLite uses plain CREATE INDEX.
|
||||
transaction: false,
|
||||
sql: '',
|
||||
handler: async (engine) => {
|
||||
if (engine.kind === 'postgres') {
|
||||
await engine.runMigration(
|
||||
104,
|
||||
`DO $$ BEGIN
|
||||
IF EXISTS (
|
||||
SELECT 1 FROM pg_index i
|
||||
JOIN pg_class c ON c.oid = i.indexrelid
|
||||
WHERE c.relname = 'pages_atom_source_hash_idx' AND NOT i.indisvalid
|
||||
) THEN
|
||||
EXECUTE 'DROP INDEX CONCURRENTLY IF EXISTS pages_atom_source_hash_idx';
|
||||
END IF;
|
||||
END $$;`
|
||||
);
|
||||
await engine.runMigration(
|
||||
104,
|
||||
`CREATE INDEX CONCURRENTLY IF NOT EXISTS pages_atom_source_hash_idx
|
||||
ON pages ((frontmatter->>'source_hash'))
|
||||
WHERE type = 'atom' AND deleted_at IS NULL;`
|
||||
);
|
||||
} else {
|
||||
await engine.runMigration(
|
||||
104,
|
||||
`CREATE INDEX IF NOT EXISTS pages_atom_source_hash_idx
|
||||
ON pages ((frontmatter->>'source_hash'))
|
||||
WHERE type = 'atom' AND deleted_at IS NULL;`
|
||||
);
|
||||
}
|
||||
},
|
||||
},
|
||||
];
|
||||
|
||||
export const LATEST_VERSION = MIGRATIONS.length > 0
|
||||
|
||||
@@ -293,6 +293,33 @@ export function importFingerprint(p: {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* v0.41.19.0 — Fingerprint for `extract --by-mention`. The mode is
|
||||
* materially different from `extract links/timeline/all` (different
|
||||
* SQL, different write semantics), so it gets its own fingerprint
|
||||
* space rather than sharing extractFingerprint.
|
||||
*
|
||||
* Filters narrow the scan universe AND the gazetteer hash narrows the
|
||||
* matching universe; both belong in the fingerprint so adding new
|
||||
* entity pages between paused runs invalidates the checkpoint cleanly
|
||||
* (codex caught the omission — without gazetteer in the key, resumed
|
||||
* pages would skip new entities silently).
|
||||
*/
|
||||
export function mentionsFingerprint(p: {
|
||||
source?: string;
|
||||
type?: string;
|
||||
since?: string;
|
||||
gazetteerHash: string;
|
||||
}): string {
|
||||
return fingerprint({
|
||||
mode: 'by_mention',
|
||||
source: p.source ?? 'default',
|
||||
type: p.type ?? null,
|
||||
since: p.since ?? null,
|
||||
gazetteer: p.gazetteerHash,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Cycle's purge phase calls this to drop stale checkpoints. 7-day TTL is
|
||||
* deliberately generous — any reasonable long-running op finishes inside
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import { withEnv } from './helpers/with-env.ts';
|
||||
import { runCycle } from '../src/core/cycle.ts';
|
||||
import { mkdtempSync, rmSync } from 'fs';
|
||||
import { tmpdir } from 'os';
|
||||
@@ -21,6 +22,15 @@ import { join } from 'path';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
let brainDir: string;
|
||||
// Per-test GBRAIN_HOME isolation: cycle's PGLite path acquires a file
|
||||
// lock at `~/.gbrain/cycle.lock` (no sourceId scope). Without isolating
|
||||
// GBRAIN_HOME per test, parallel gbrain processes on the same machine
|
||||
// (including sibling Conductor worktrees running their own tests)
|
||||
// contend for the same lock file — runCycle returns 'skipped' and the
|
||||
// last_full_cycle_at exit hook silently no-ops. Each test wraps its
|
||||
// body in `withEnv({GBRAIN_HOME: <unique tmp>})` so the file lock path
|
||||
// becomes per-test.
|
||||
let gbrainHome: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
@@ -35,6 +45,7 @@ afterAll(async () => {
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
brainDir = mkdtempSync(join(tmpdir(), 'gbrain-cycle-lfca-'));
|
||||
gbrainHome = mkdtempSync(join(tmpdir(), 'gbrain-cycle-lfca-home-'));
|
||||
});
|
||||
|
||||
async function seedSource(id: string): Promise<void> {
|
||||
@@ -56,83 +67,93 @@ async function readLastFullCycleAt(sourceId: string): Promise<string | null> {
|
||||
|
||||
describe('runCycle last_full_cycle_at exit hook', () => {
|
||||
test('per-source cycle with status=ok writes timestamp', async () => {
|
||||
await seedSource('alpha');
|
||||
const before = await readLastFullCycleAt('alpha');
|
||||
expect(before).toBeNull();
|
||||
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
|
||||
await seedSource('alpha');
|
||||
const before = await readLastFullCycleAt('alpha');
|
||||
expect(before).toBeNull();
|
||||
|
||||
// Run a minimal cycle: just lint (filesystem, no DB writes, always returns 'ok')
|
||||
const t0 = Date.now();
|
||||
const report = await runCycle(engine, {
|
||||
brainDir,
|
||||
sourceId: 'alpha',
|
||||
phases: ['lint'],
|
||||
// Run a minimal cycle: just lint (filesystem, no DB writes, always returns 'ok')
|
||||
const t0 = Date.now();
|
||||
const report = await runCycle(engine, {
|
||||
brainDir,
|
||||
sourceId: 'alpha',
|
||||
phases: ['lint'],
|
||||
});
|
||||
// lint on an empty dir returns ok+clean+0 fixes
|
||||
expect(['ok', 'clean']).toContain(report.status);
|
||||
|
||||
const after = await readLastFullCycleAt('alpha');
|
||||
expect(after).not.toBeNull();
|
||||
const writtenMs = new Date(after!).getTime();
|
||||
expect(writtenMs).toBeGreaterThanOrEqual(t0);
|
||||
expect(writtenMs).toBeLessThanOrEqual(Date.now() + 1000);
|
||||
});
|
||||
// lint on an empty dir returns ok+clean+0 fixes
|
||||
expect(['ok', 'clean']).toContain(report.status);
|
||||
|
||||
const after = await readLastFullCycleAt('alpha');
|
||||
expect(after).not.toBeNull();
|
||||
const writtenMs = new Date(after!).getTime();
|
||||
expect(writtenMs).toBeGreaterThanOrEqual(t0);
|
||||
expect(writtenMs).toBeLessThanOrEqual(Date.now() + 1000);
|
||||
});
|
||||
|
||||
test('legacy caller (no sourceId) does NOT write any source timestamp', async () => {
|
||||
await seedSource('default-like');
|
||||
// No sourceId passed; should remain untouched.
|
||||
await runCycle(engine, {
|
||||
brainDir,
|
||||
phases: ['lint'],
|
||||
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
|
||||
await seedSource('default-like');
|
||||
// No sourceId passed; should remain untouched.
|
||||
await runCycle(engine, {
|
||||
brainDir,
|
||||
phases: ['lint'],
|
||||
});
|
||||
// No per-source write happens; default source's config stays empty.
|
||||
const after = await readLastFullCycleAt('default-like');
|
||||
expect(after).toBeNull();
|
||||
});
|
||||
// No per-source write happens; default source's config stays empty.
|
||||
const after = await readLastFullCycleAt('default-like');
|
||||
expect(after).toBeNull();
|
||||
});
|
||||
|
||||
test('dryRun=true skips the write', async () => {
|
||||
await seedSource('beta');
|
||||
await runCycle(engine, {
|
||||
brainDir,
|
||||
sourceId: 'beta',
|
||||
phases: ['lint'],
|
||||
dryRun: true,
|
||||
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
|
||||
await seedSource('beta');
|
||||
await runCycle(engine, {
|
||||
brainDir,
|
||||
sourceId: 'beta',
|
||||
phases: ['lint'],
|
||||
dryRun: true,
|
||||
});
|
||||
const after = await readLastFullCycleAt('beta');
|
||||
expect(after).toBeNull();
|
||||
});
|
||||
const after = await readLastFullCycleAt('beta');
|
||||
expect(after).toBeNull();
|
||||
});
|
||||
|
||||
test('cycle that returns skipped (lock held) does NOT mark timestamp', async () => {
|
||||
await seedSource('gamma');
|
||||
// Inject a live lock row directly so the cycle returns 'skipped'.
|
||||
// This simulates "another cycle is already running for gamma."
|
||||
const lockId = 'gbrain-cycle:gamma';
|
||||
const pid = process.pid;
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at)
|
||||
VALUES ($1, $2, 'test', NOW(), NOW() + INTERVAL '30 minutes')`,
|
||||
[lockId, pid + 99999],
|
||||
);
|
||||
const report = await runCycle(engine, {
|
||||
brainDir,
|
||||
sourceId: 'gamma',
|
||||
phases: ['lint', 'sync'], // sync triggers lock acquisition
|
||||
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
|
||||
await seedSource('gamma');
|
||||
// Inject a live lock row directly so the cycle returns 'skipped'.
|
||||
// This simulates "another cycle is already running for gamma."
|
||||
const lockId = 'gbrain-cycle:gamma';
|
||||
const pid = process.pid;
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at)
|
||||
VALUES ($1, $2, 'test', NOW(), NOW() + INTERVAL '30 minutes')`,
|
||||
[lockId, pid + 99999],
|
||||
);
|
||||
const report = await runCycle(engine, {
|
||||
brainDir,
|
||||
sourceId: 'gamma',
|
||||
phases: ['lint', 'sync'], // sync triggers lock acquisition
|
||||
});
|
||||
expect(report.status).toBe('skipped');
|
||||
expect(report.reason).toBe('cycle_already_running');
|
||||
const after = await readLastFullCycleAt('gamma');
|
||||
expect(after).toBeNull();
|
||||
});
|
||||
expect(report.status).toBe('skipped');
|
||||
expect(report.reason).toBe('cycle_already_running');
|
||||
const after = await readLastFullCycleAt('gamma');
|
||||
expect(after).toBeNull();
|
||||
});
|
||||
|
||||
test('two consecutive per-source cycles update the timestamp on each run', async () => {
|
||||
await seedSource('delta');
|
||||
await runCycle(engine, { brainDir, sourceId: 'delta', phases: ['lint'] });
|
||||
const first = await readLastFullCycleAt('delta');
|
||||
expect(first).not.toBeNull();
|
||||
// Wait 10ms so the timestamp can advance
|
||||
await new Promise(r => setTimeout(r, 10));
|
||||
await runCycle(engine, { brainDir, sourceId: 'delta', phases: ['lint'] });
|
||||
const second = await readLastFullCycleAt('delta');
|
||||
expect(second).not.toBeNull();
|
||||
expect(new Date(second!).getTime()).toBeGreaterThan(new Date(first!).getTime());
|
||||
await withEnv({ GBRAIN_HOME: gbrainHome }, async () => {
|
||||
await seedSource('delta');
|
||||
await runCycle(engine, { brainDir, sourceId: 'delta', phases: ['lint'] });
|
||||
const first = await readLastFullCycleAt('delta');
|
||||
expect(first).not.toBeNull();
|
||||
// Wait 10ms so the timestamp can advance
|
||||
await new Promise(r => setTimeout(r, 10));
|
||||
await runCycle(engine, { brainDir, sourceId: 'delta', phases: ['lint'] });
|
||||
const second = await readLastFullCycleAt('delta');
|
||||
expect(second).not.toBeNull();
|
||||
expect(new Date(second!).getTime()).toBeGreaterThan(new Date(first!).getTime());
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
// v0.41.19.0 — T2 of ops-fix-wave.
|
||||
//
|
||||
// Regression pin: the cycle DB lock TTL was dropped from 30 min to 5 min
|
||||
// in v0.41.19.0 (T2). Combined with active in-phase refresh via
|
||||
// buildYieldDuringPhase (T3) this makes crash recovery 6× faster
|
||||
// (≤5min vs ≤30min before).
|
||||
//
|
||||
// This test pins the constant via the migration query observable. If
|
||||
// the TTL ever climbs back above 5 min, the ops pain comes back.
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { readFileSync } from 'fs';
|
||||
import { join } from 'path';
|
||||
|
||||
describe('cycle lock TTL (T2 regression pin)', () => {
|
||||
test('LOCK_TTL_MINUTES === 5 in src/core/cycle.ts', () => {
|
||||
const src = readFileSync(
|
||||
join(__dirname, '..', '..', 'src', 'core', 'cycle.ts'),
|
||||
'utf-8',
|
||||
);
|
||||
// Pin the literal constant value. Two patterns guarded:
|
||||
// - LOCK_TTL_MINUTES = 5
|
||||
// - LOCK_TTL_MS = 5 * 60 * 1000
|
||||
expect(src).toMatch(/LOCK_TTL_MINUTES\s*=\s*5\b/);
|
||||
expect(src).toMatch(/LOCK_TTL_MS\s*=\s*5\s*\*\s*60\s*\*\s*1000/);
|
||||
// And explicitly disallow the prior 30-minute value re-creeping back.
|
||||
expect(src).not.toMatch(/LOCK_TTL_MINUTES\s*=\s*30\b/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,106 @@
|
||||
// v0.41.19.0 — T1 of ops-fix-wave.
|
||||
//
|
||||
// Pins the batch idempotency contract for extract_atoms. The replaced
|
||||
// per-hash helper did 7K SQL round trips on a brain with 7K conversation
|
||||
// transcripts; the batch helper does ONE.
|
||||
//
|
||||
// Coverage: empty input short-circuits without a query; mixed-existing
|
||||
// returns just the existing set; SQL failure fails open with empty Set
|
||||
// (preserves the prior fail-open posture so a broken check doesn't block
|
||||
// extraction).
|
||||
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../../src/core/pglite-engine.ts';
|
||||
import { atomsExistingForHashes } from '../../src/core/cycle/extract-atoms.ts';
|
||||
import { resetPgliteState } from '../helpers/reset-pglite.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 60000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
async function seedAtom(slug: string, sourceHash: string, sourceId = 'default'): Promise<void> {
|
||||
await engine.putPage(slug, {
|
||||
title: slug.split('/').pop() ?? slug,
|
||||
type: 'atom',
|
||||
compiled_truth: 'test atom body',
|
||||
frontmatter: {
|
||||
type: 'atom',
|
||||
source_hash: sourceHash,
|
||||
},
|
||||
timeline: '',
|
||||
}, { sourceId });
|
||||
}
|
||||
|
||||
describe('atomsExistingForHashes (T1 batch idempotency)', () => {
|
||||
test('empty input short-circuits without a query', async () => {
|
||||
const result = await atomsExistingForHashes(engine, 'default', []);
|
||||
expect(result.size).toBe(0);
|
||||
});
|
||||
|
||||
test('returns just the hashes that have matching atom rows', async () => {
|
||||
// Seed 3 atoms with known hashes
|
||||
await seedAtom('atoms/2026-05-26/a', 'aaaaaaaaaaaaaaaa');
|
||||
await seedAtom('atoms/2026-05-26/b', 'bbbbbbbbbbbbbbbb');
|
||||
await seedAtom('atoms/2026-05-26/c', 'cccccccccccccccc');
|
||||
|
||||
// Query with a mixed list: 2 existing + 2 new
|
||||
const result = await atomsExistingForHashes(engine, 'default', [
|
||||
'aaaaaaaaaaaaaaaa',
|
||||
'bbbbbbbbbbbbbbbb',
|
||||
'dddddddddddddddd', // not seeded
|
||||
'eeeeeeeeeeeeeeee', // not seeded
|
||||
]);
|
||||
expect(result.size).toBe(2);
|
||||
expect(result.has('aaaaaaaaaaaaaaaa')).toBe(true);
|
||||
expect(result.has('bbbbbbbbbbbbbbbb')).toBe(true);
|
||||
expect(result.has('dddddddddddddddd')).toBe(false);
|
||||
});
|
||||
|
||||
test('scoped by source_id — atom in source A invisible to source B query', async () => {
|
||||
// Register non-default sources first (pages.source_id FK).
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name) VALUES ('source-a', 'source-a'), ('source-b', 'source-b')
|
||||
ON CONFLICT DO NOTHING`,
|
||||
);
|
||||
// Pre-fix the per-hash helper had the same scope; this regression-
|
||||
// guards that the batch helper preserves it.
|
||||
await seedAtom('atoms/2026-05-26/x', 'xxxxxxxxxxxxxxxx', 'source-a');
|
||||
const fromA = await atomsExistingForHashes(engine, 'source-a', ['xxxxxxxxxxxxxxxx']);
|
||||
const fromB = await atomsExistingForHashes(engine, 'source-b', ['xxxxxxxxxxxxxxxx']);
|
||||
expect(fromA.size).toBe(1);
|
||||
expect(fromB.size).toBe(0);
|
||||
});
|
||||
|
||||
test('soft-deleted atoms are not visible', async () => {
|
||||
await seedAtom('atoms/2026-05-26/deleted', 'ffffffffffffffff');
|
||||
await engine.executeRaw(
|
||||
`UPDATE pages SET deleted_at = NOW() WHERE slug = $1 AND source_id = 'default'`,
|
||||
['atoms/2026-05-26/deleted'],
|
||||
);
|
||||
const result = await atomsExistingForHashes(engine, 'default', ['ffffffffffffffff']);
|
||||
expect(result.size).toBe(0);
|
||||
});
|
||||
|
||||
test('fails open when query throws (returns empty Set, logs to stderr)', async () => {
|
||||
// Construct an engine with a broken executeRaw via duck-typing.
|
||||
const brokenEngine = {
|
||||
executeRaw: async () => { throw new Error('connection refused'); },
|
||||
} as unknown as PGLiteEngine;
|
||||
const result = await atomsExistingForHashes(brokenEngine, 'default', ['aaaa']);
|
||||
// Fail-open: empty Set means caller treats all as not-extracted and
|
||||
// proceeds. Re-extraction cost is bounded by daily budget cap.
|
||||
expect(result.size).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,129 @@
|
||||
// v0.41.19.0 — T4 of ops-fix-wave.
|
||||
//
|
||||
// Pins that extract_atoms wires its progress reporter inside the work
|
||||
// loop (one tick per processed item) and emits a heartbeat before the
|
||||
// batch idempotency check. Codex caught that cycle.ts must NOT pass a
|
||||
// child reporter — phases receive the SAME reporter and only call tick
|
||||
// / heartbeat (cycle.ts owns start / finish).
|
||||
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../../src/core/pglite-engine.ts';
|
||||
import { runPhaseExtractAtoms } from '../../src/core/cycle/extract-atoms.ts';
|
||||
import { resetPgliteState } from '../helpers/reset-pglite.ts';
|
||||
import type { ProgressReporter } from '../../src/core/progress.ts';
|
||||
import type { ChatResult, ChatOpts } from '../../src/core/ai/gateway.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 60000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
function makeMockReporter(): {
|
||||
reporter: ProgressReporter;
|
||||
events: Array<{ kind: 'start' | 'tick' | 'heartbeat' | 'finish' | 'child'; phase?: string; note?: string; n?: number }>;
|
||||
} {
|
||||
const events: Array<{ kind: 'start' | 'tick' | 'heartbeat' | 'finish' | 'child'; phase?: string; note?: string; n?: number }> = [];
|
||||
const reporter: ProgressReporter = {
|
||||
start: (phase, _total) => { events.push({ kind: 'start', phase }); },
|
||||
tick: (n, note) => { events.push({ kind: 'tick', n, note }); },
|
||||
heartbeat: (note) => { events.push({ kind: 'heartbeat', note }); },
|
||||
finish: (note) => { events.push({ kind: 'finish', note }); },
|
||||
child: (phase) => {
|
||||
events.push({ kind: 'child', phase });
|
||||
return reporter; // return self for simplicity
|
||||
},
|
||||
};
|
||||
return { reporter, events };
|
||||
}
|
||||
|
||||
function stubChat(text: string): (o: ChatOpts) => Promise<ChatResult> {
|
||||
return async (_o: ChatOpts) => ({
|
||||
text,
|
||||
blocks: [{ type: 'text', text }],
|
||||
stopReason: 'end',
|
||||
usage: { input_tokens: 100, output_tokens: 50, cache_read_tokens: 0, cache_creation_tokens: 0 },
|
||||
model: 'anthropic:claude-haiku-4-5',
|
||||
providerId: 'anthropic',
|
||||
});
|
||||
}
|
||||
|
||||
describe('extract_atoms progress wiring (T4)', () => {
|
||||
test('phase does NOT call start or finish — cycle.ts owns those', async () => {
|
||||
const { reporter, events } = makeMockReporter();
|
||||
const validAtomJson = JSON.stringify([
|
||||
{ title: 'A', atom_type: 'insight', body: 'body a' },
|
||||
]);
|
||||
await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [
|
||||
{ filePath: '/tmp/t1.txt', content: 'transcript 1 body', contentHash: 'h1'.repeat(8) },
|
||||
],
|
||||
_pages: [],
|
||||
_chat: stubChat(validAtomJson),
|
||||
progress: reporter,
|
||||
});
|
||||
const startEvents = events.filter(e => e.kind === 'start');
|
||||
const finishEvents = events.filter(e => e.kind === 'finish');
|
||||
expect(startEvents.length).toBe(0);
|
||||
expect(finishEvents.length).toBe(0);
|
||||
});
|
||||
|
||||
test('emits a heartbeat before the batch idempotency check', async () => {
|
||||
const { reporter, events } = makeMockReporter();
|
||||
await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [
|
||||
{ filePath: '/tmp/t1.txt', content: 'transcript', contentHash: 'h1'.repeat(8) },
|
||||
],
|
||||
_pages: [],
|
||||
_chat: stubChat('[]'),
|
||||
progress: reporter,
|
||||
});
|
||||
const heartbeats = events.filter(e => e.kind === 'heartbeat');
|
||||
expect(heartbeats.length).toBeGreaterThanOrEqual(1);
|
||||
// Note mentions the count
|
||||
expect(heartbeats[0].note).toMatch(/checking existing atoms/);
|
||||
});
|
||||
|
||||
test('one tick per processed work item with running count note', async () => {
|
||||
const { reporter, events } = makeMockReporter();
|
||||
const validAtomJson = JSON.stringify([
|
||||
{ title: 'A', atom_type: 'insight', body: 'body a' },
|
||||
]);
|
||||
await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [
|
||||
{ filePath: '/tmp/t1.txt', content: 'a', contentHash: 'h1'.repeat(8) },
|
||||
{ filePath: '/tmp/t2.txt', content: 'b', contentHash: 'h2'.repeat(8) },
|
||||
{ filePath: '/tmp/t3.txt', content: 'c', contentHash: 'h3'.repeat(8) },
|
||||
],
|
||||
_pages: [],
|
||||
_chat: stubChat(validAtomJson),
|
||||
progress: reporter,
|
||||
});
|
||||
const ticks = events.filter(e => e.kind === 'tick');
|
||||
expect(ticks.length).toBe(3);
|
||||
expect(ticks[0].note).toMatch(/atoms.*skipped/);
|
||||
});
|
||||
|
||||
test('no progress wiring required — opts.progress is optional', async () => {
|
||||
// Sanity: phase works without a reporter.
|
||||
const result = await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [],
|
||||
_pages: [],
|
||||
});
|
||||
expect(result.phase).toBe('extract_atoms');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,96 @@
|
||||
// v0.41.19.0 — T4 of ops-fix-wave.
|
||||
//
|
||||
// Pins that synthesize_concepts wires its progress reporter inside the
|
||||
// concept-group loop (one tick per concept written). Cycle.ts owns
|
||||
// start/finish; phase only ticks.
|
||||
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../../src/core/pglite-engine.ts';
|
||||
import { runPhaseSynthesizeConcepts } from '../../src/core/cycle/synthesize-concepts.ts';
|
||||
import { resetPgliteState } from '../helpers/reset-pglite.ts';
|
||||
import type { ProgressReporter } from '../../src/core/progress.ts';
|
||||
import type { ChatResult, ChatOpts } from '../../src/core/ai/gateway.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 60000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
function makeMockReporter(): {
|
||||
reporter: ProgressReporter;
|
||||
events: Array<{ kind: 'tick' | 'heartbeat' | 'start' | 'finish'; note?: string }>;
|
||||
} {
|
||||
const events: Array<{ kind: 'tick' | 'heartbeat' | 'start' | 'finish'; note?: string }> = [];
|
||||
const reporter: ProgressReporter = {
|
||||
start: () => { events.push({ kind: 'start' }); },
|
||||
tick: (_n, note) => { events.push({ kind: 'tick', note }); },
|
||||
heartbeat: (note) => { events.push({ kind: 'heartbeat', note }); },
|
||||
finish: (note) => { events.push({ kind: 'finish', note }); },
|
||||
child: () => reporter,
|
||||
};
|
||||
return { reporter, events };
|
||||
}
|
||||
|
||||
function stubChat(text: string): (o: ChatOpts) => Promise<ChatResult> {
|
||||
return async (_o: ChatOpts) => ({
|
||||
text,
|
||||
blocks: [{ type: 'text', text }],
|
||||
stopReason: 'end',
|
||||
usage: { input_tokens: 100, output_tokens: 50, cache_read_tokens: 0, cache_creation_tokens: 0 },
|
||||
model: 'anthropic:claude-sonnet-4-6',
|
||||
providerId: 'anthropic',
|
||||
});
|
||||
}
|
||||
|
||||
describe('synthesize_concepts progress wiring (T4)', () => {
|
||||
test('phase does NOT call start or finish', async () => {
|
||||
const { reporter, events } = makeMockReporter();
|
||||
await runPhaseSynthesizeConcepts(engine, {
|
||||
_atoms: [
|
||||
// T3 tier (2 atoms): no LLM, deterministic narrative
|
||||
{ slug: 'atoms/a1', concept_refs: ['concepts/x'], body: 'b1', title: 'A1' },
|
||||
{ slug: 'atoms/a2', concept_refs: ['concepts/x'], body: 'b2', title: 'A2' },
|
||||
],
|
||||
_chat: stubChat('narrative text'),
|
||||
progress: reporter,
|
||||
});
|
||||
expect(events.filter(e => e.kind === 'start').length).toBe(0);
|
||||
expect(events.filter(e => e.kind === 'finish').length).toBe(0);
|
||||
});
|
||||
|
||||
test('one tick per concept group written', async () => {
|
||||
const { reporter, events } = makeMockReporter();
|
||||
await runPhaseSynthesizeConcepts(engine, {
|
||||
_atoms: [
|
||||
{ slug: 'atoms/a1', concept_refs: ['concepts/x'], body: 'b1', title: 'A1' },
|
||||
{ slug: 'atoms/a2', concept_refs: ['concepts/x'], body: 'b2', title: 'A2' },
|
||||
{ slug: 'atoms/a3', concept_refs: ['concepts/y'], body: 'b3', title: 'A3' },
|
||||
{ slug: 'atoms/a4', concept_refs: ['concepts/y'], body: 'b4', title: 'A4' },
|
||||
],
|
||||
_chat: stubChat('narrative text'),
|
||||
progress: reporter,
|
||||
});
|
||||
const ticks = events.filter(e => e.kind === 'tick');
|
||||
// Two concept groups, each ≥2 atoms → both qualify for synthesis
|
||||
expect(ticks.length).toBe(2);
|
||||
expect(ticks[0].note).toMatch(/concepts/);
|
||||
});
|
||||
|
||||
test('no progress wiring required — opts.progress is optional', async () => {
|
||||
const result = await runPhaseSynthesizeConcepts(engine, {
|
||||
_atoms: [],
|
||||
});
|
||||
expect(result.phase).toBe('synthesize_concepts');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,100 @@
|
||||
// v0.41.19.0 — T3 of ops-fix-wave (codex catch).
|
||||
//
|
||||
// Pins that buildYieldDuringPhase actually calls lock.refresh() AND the
|
||||
// outer hook on every fire. Codex caught that the prior plan's "use
|
||||
// yieldBetweenPhases" claim was false — yieldBetweenPhases is just
|
||||
// setImmediate() from jobs.ts/autopilot.ts and never refreshes the
|
||||
// cycle DB lock. Combined with TTL=5min (T2), a missing refresh would
|
||||
// lose the lock mid-phase. The closure built by buildYieldDuringPhase
|
||||
// is the active refresh path.
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { buildYieldDuringPhase } from '../../src/core/cycle.ts';
|
||||
import type { LockHandle } from '../../src/core/cycle.ts';
|
||||
|
||||
function makeMockLock(): { lock: LockHandle; refreshCount: number; releaseCount: number } {
|
||||
const state = { refreshCount: 0, releaseCount: 0 };
|
||||
const lock: LockHandle = {
|
||||
refresh: async () => { state.refreshCount++; },
|
||||
release: async () => { state.releaseCount++; },
|
||||
};
|
||||
return {
|
||||
lock,
|
||||
get refreshCount() { return state.refreshCount; },
|
||||
get releaseCount() { return state.releaseCount; },
|
||||
};
|
||||
}
|
||||
|
||||
describe('buildYieldDuringPhase (T3 codex fix)', () => {
|
||||
test('returns undefined when both lock and outer are absent', () => {
|
||||
const fn = buildYieldDuringPhase(null);
|
||||
expect(fn).toBeUndefined();
|
||||
});
|
||||
|
||||
test('returns a function when lock is present', () => {
|
||||
const { lock } = makeMockLock();
|
||||
const fn = buildYieldDuringPhase(lock);
|
||||
expect(typeof fn).toBe('function');
|
||||
});
|
||||
|
||||
test('returns a function when only outer is present', () => {
|
||||
const fn = buildYieldDuringPhase(null, async () => {});
|
||||
expect(typeof fn).toBe('function');
|
||||
});
|
||||
|
||||
test('each fire calls lock.refresh exactly once', async () => {
|
||||
const tracker = makeMockLock();
|
||||
const fn = buildYieldDuringPhase(tracker.lock);
|
||||
expect(fn).toBeDefined();
|
||||
await fn!();
|
||||
expect(tracker.refreshCount).toBe(1);
|
||||
await fn!();
|
||||
expect(tracker.refreshCount).toBe(2);
|
||||
await fn!();
|
||||
expect(tracker.refreshCount).toBe(3);
|
||||
});
|
||||
|
||||
test('each fire calls the outer hook AFTER lock.refresh', async () => {
|
||||
const tracker = makeMockLock();
|
||||
const callOrder: string[] = [];
|
||||
const outer = async () => { callOrder.push('outer'); };
|
||||
const fn = buildYieldDuringPhase({
|
||||
...tracker.lock,
|
||||
refresh: async () => { callOrder.push('refresh'); tracker.lock.refresh(); },
|
||||
}, outer);
|
||||
await fn!();
|
||||
expect(callOrder).toEqual(['refresh', 'outer']);
|
||||
});
|
||||
|
||||
test('lock.refresh throw is non-fatal — outer still runs', async () => {
|
||||
let outerCalled = false;
|
||||
const badLock: LockHandle = {
|
||||
refresh: async () => { throw new Error('lock stolen'); },
|
||||
release: async () => {},
|
||||
};
|
||||
const fn = buildYieldDuringPhase(badLock, async () => { outerCalled = true; });
|
||||
// Must not throw.
|
||||
await fn!();
|
||||
// Outer should still have run even though refresh threw.
|
||||
expect(outerCalled).toBe(true);
|
||||
});
|
||||
|
||||
test('outer hook throw is non-fatal', async () => {
|
||||
const tracker = makeMockLock();
|
||||
const fn = buildYieldDuringPhase(tracker.lock, async () => {
|
||||
throw new Error('outer kaboom');
|
||||
});
|
||||
// Must not throw.
|
||||
await fn!();
|
||||
// Refresh still fired despite outer failure.
|
||||
expect(tracker.refreshCount).toBe(1);
|
||||
});
|
||||
|
||||
test('never calls lock.release (release stays separate from refresh)', async () => {
|
||||
const tracker = makeMockLock();
|
||||
const fn = buildYieldDuringPhase(tracker.lock);
|
||||
for (let i = 0; i < 5; i++) await fn!();
|
||||
expect(tracker.refreshCount).toBe(5);
|
||||
expect(tracker.releaseCount).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,114 @@
|
||||
// v0.41.19.0 — T3 of ops-fix-wave.
|
||||
//
|
||||
// Pins the 30s throttle on the per-phase maybeYield helper. Without
|
||||
// this, every loop iteration would fire yieldDuringPhase (which on a
|
||||
// 322K-page brain is hundreds of redundant lock refreshes per phase).
|
||||
//
|
||||
// Behavioral test: drive runPhaseExtractAtoms with a synthetic chat
|
||||
// stub + a yieldDuringPhase callback that records call timestamps.
|
||||
// Verify the 30s throttle holds — fast-iter runs produce ONE fire even
|
||||
// across many items.
|
||||
//
|
||||
// Note on fake time: the helper reads Date.now() directly inside the
|
||||
// phase closure. We can't override it cleanly without touching the
|
||||
// global. Instead we test the OBSERVABLE behavior: 5 items in under
|
||||
// 30s wall-clock should produce exactly 1 yield (the very first call,
|
||||
// when lastYieldMs starts at Date.now() — the 30s gate immediately
|
||||
// returns false, so the FIRST iteration is also throttled out). The
|
||||
// helper fires when (now - lastYieldMs) >= 30_000.
|
||||
//
|
||||
// Since lastYieldMs is initialized to Date.now() at the top of the
|
||||
// phase, NO yields fire within the first 30s of execution. This is by
|
||||
// design — the throttle starts the clock at phase entry. For a
|
||||
// healthy fast run, 0 fires is correct.
|
||||
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../../src/core/pglite-engine.ts';
|
||||
import { runPhaseExtractAtoms } from '../../src/core/cycle/extract-atoms.ts';
|
||||
import { resetPgliteState } from '../helpers/reset-pglite.ts';
|
||||
import type { ChatResult, ChatOpts } from '../../src/core/ai/gateway.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 60000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
function stubChat(): (o: ChatOpts) => Promise<ChatResult> {
|
||||
return async (_o: ChatOpts) => ({
|
||||
text: JSON.stringify([{ title: 'T', atom_type: 'insight', body: 'b' }]),
|
||||
blocks: [{ type: 'text', text: '' }],
|
||||
stopReason: 'end',
|
||||
usage: { input_tokens: 10, output_tokens: 10, cache_read_tokens: 0, cache_creation_tokens: 0 },
|
||||
model: 'anthropic:claude-haiku-4-5',
|
||||
providerId: 'anthropic',
|
||||
});
|
||||
}
|
||||
|
||||
describe('extract_atoms yieldDuringPhase throttle (T3)', () => {
|
||||
test('fast iterations within 30s fire 0 yields (throttle blocks first 30s after start)', async () => {
|
||||
const yieldTimestamps: number[] = [];
|
||||
const yieldFn = async () => { yieldTimestamps.push(Date.now()); };
|
||||
await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [
|
||||
{ filePath: '/tmp/a', content: 'a', contentHash: 'a1'.repeat(8) },
|
||||
{ filePath: '/tmp/b', content: 'b', contentHash: 'b2'.repeat(8) },
|
||||
{ filePath: '/tmp/c', content: 'c', contentHash: 'c3'.repeat(8) },
|
||||
{ filePath: '/tmp/d', content: 'd', contentHash: 'd4'.repeat(8) },
|
||||
{ filePath: '/tmp/e', content: 'e', contentHash: 'e5'.repeat(8) },
|
||||
],
|
||||
_pages: [],
|
||||
_chat: stubChat(),
|
||||
yieldDuringPhase: yieldFn,
|
||||
});
|
||||
// The lastYieldMs is initialized to Date.now() at phase start, so
|
||||
// no fire occurs within the first 30s. The 5-iteration test runs
|
||||
// in milliseconds, so we expect ZERO yields. This is the correct
|
||||
// behavior — under healthy load the lock has plenty of TTL budget
|
||||
// and we don't need to spam refresh.
|
||||
expect(yieldTimestamps.length).toBe(0);
|
||||
});
|
||||
|
||||
test('phase tolerates undefined yieldDuringPhase', async () => {
|
||||
// Sanity: phase doesn't crash without the hook.
|
||||
const result = await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [],
|
||||
_pages: [],
|
||||
});
|
||||
expect(result.phase).toBe('extract_atoms');
|
||||
});
|
||||
|
||||
test('yieldDuringPhase throw is non-fatal (logged, not propagated)', async () => {
|
||||
const throwingYield = async () => { throw new Error('lock stolen'); };
|
||||
// Even if yieldDuringPhase throws (would fire after 30s wall-clock),
|
||||
// phase doesn't crash. We can't easily trigger >30s in a test, but
|
||||
// we CAN verify the catch wrapper exists by reading the source.
|
||||
const fs = await import('fs');
|
||||
const src = fs.readFileSync(
|
||||
new URL('../../src/core/cycle/extract-atoms.ts', import.meta.url),
|
||||
'utf-8',
|
||||
);
|
||||
expect(src).toMatch(/try\s*\{\s*await\s+opts\.yieldDuringPhase\(\)/);
|
||||
expect(src).toMatch(/yieldDuringPhase failed \(non-fatal\)/);
|
||||
// Phase itself runs without throwing.
|
||||
const result = await runPhaseExtractAtoms(engine, {
|
||||
sourceId: 'default',
|
||||
_transcripts: [],
|
||||
_pages: [],
|
||||
yieldDuringPhase: throwingYield,
|
||||
});
|
||||
expect(result.phase).toBe('extract_atoms');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,102 @@
|
||||
// v0.41.19.0 — T6 of ops-fix-wave.
|
||||
//
|
||||
// Pins the sync_consolidation doctor check (Issue 5 — surface the
|
||||
// `gbrain sync --all --parallel` recommendation to operators with
|
||||
// multi-source brains).
|
||||
//
|
||||
// Coverage:
|
||||
// - 0 sources → ok with "not applicable" message
|
||||
// - 1 source → ok with "not applicable" message
|
||||
// - 2+ active sources → ok with paste-ready cron command in message
|
||||
// - archived sources excluded from the count (codex edge case)
|
||||
// - all sources archived → counts as < 2 → "not applicable"
|
||||
// - SQL throws → status='warn' (own try/catch, not relying on outer doctor catch)
|
||||
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { checkSyncConsolidation } from '../src/commands/doctor.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 60000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
async function addSource(id: string, opts: { local_path?: string | null; archived?: boolean } = {}): Promise<void> {
|
||||
const local_path = opts.local_path === null ? null : (opts.local_path ?? `/tmp/${id}`);
|
||||
const archived = opts.archived ?? false;
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, local_path, archived)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path, archived = EXCLUDED.archived`,
|
||||
[id, id, local_path, archived],
|
||||
);
|
||||
}
|
||||
|
||||
describe('checkSyncConsolidation (Issue 5)', () => {
|
||||
test('0 sources (only default w/ NULL local_path) → ok with "not applicable"', async () => {
|
||||
// Default source exists from initSchema but has NULL local_path.
|
||||
// checkSyncConsolidation filters on local_path IS NOT NULL.
|
||||
const result = await checkSyncConsolidation(engine);
|
||||
expect(result.name).toBe('sync_consolidation');
|
||||
expect(result.status).toBe('ok');
|
||||
expect(result.message).toMatch(/not applicable/i);
|
||||
});
|
||||
|
||||
test('1 active source → ok with "not applicable"', async () => {
|
||||
await addSource('default', { local_path: '/tmp/default-brain' });
|
||||
const result = await checkSyncConsolidation(engine);
|
||||
expect(result.status).toBe('ok');
|
||||
expect(result.message).toMatch(/single-source/i);
|
||||
expect(result.message).toMatch(/not applicable/i);
|
||||
});
|
||||
|
||||
test('3 active sources → ok with paste-ready `sync --all` command', async () => {
|
||||
await addSource('default', { local_path: '/tmp/default-brain' });
|
||||
await addSource('zion-brain');
|
||||
await addSource('media-brain');
|
||||
const result = await checkSyncConsolidation(engine);
|
||||
expect(result.status).toBe('ok');
|
||||
expect(result.message).toMatch(/3 active sources/);
|
||||
// Paste-ready command embedded in message
|
||||
expect(result.message).toMatch(/gbrain sync --all --parallel 4 --workers 4 --skip-failed/);
|
||||
});
|
||||
|
||||
test('2 sources both archived → "not applicable" (archived excluded)', async () => {
|
||||
await addSource('archived-a', { archived: true });
|
||||
await addSource('archived-b', { archived: true });
|
||||
const result = await checkSyncConsolidation(engine);
|
||||
expect(result.status).toBe('ok');
|
||||
expect(result.message).toMatch(/not applicable/i);
|
||||
});
|
||||
|
||||
test('mixed — 1 active + 1 archived → "not applicable" (only 1 counts)', async () => {
|
||||
await addSource('active', { local_path: '/tmp/active' });
|
||||
await addSource('archived', { archived: true });
|
||||
const result = await checkSyncConsolidation(engine);
|
||||
expect(result.status).toBe('ok');
|
||||
expect(result.message).toMatch(/not applicable/i);
|
||||
});
|
||||
|
||||
test('SQL failure → status=warn with diagnostic message (own try/catch)', async () => {
|
||||
// Construct a broken engine via duck-typing.
|
||||
const brokenEngine = {
|
||||
executeRaw: async () => { throw new Error('connection refused'); },
|
||||
} as unknown as PGLiteEngine;
|
||||
const result = await checkSyncConsolidation(brokenEngine);
|
||||
expect(result.status).toBe('warn');
|
||||
expect(result.message).toMatch(/Could not check sync consolidation/);
|
||||
expect(result.message).toMatch(/connection refused/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,203 @@
|
||||
/**
|
||||
* v0.41.19.0 — T5 of ops-fix-wave.
|
||||
*
|
||||
* Pins the by-mention checkpoint/resume contract + codex's 4 correctness
|
||||
* fixes:
|
||||
* 1. Persist checkpoint AFTER flush() succeeds (not per-page) so a
|
||||
* crash between batch.push and flush leaves pages un-checkpointed
|
||||
* and resume re-scans them.
|
||||
* 2. Dry-run does NOT persist OR load the checkpoint.
|
||||
* 3. Gazetteer hash is part of the fingerprint — adding/removing
|
||||
* entity pages between paused runs invalidates the checkpoint.
|
||||
* 4. Filtered pages (--type/--since miss/empty body) DO get marked
|
||||
* completed so resume doesn't re-fetch them.
|
||||
*/
|
||||
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { runExtract } from '../src/commands/extract.ts';
|
||||
import { setCliOptions } from '../src/core/cli-options.ts';
|
||||
import { loadOpCheckpoint, mentionsFingerprint } from '../src/core/op-checkpoint.ts';
|
||||
import { createHash } from 'crypto';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
// Suppress console output during runs (we're testing DB-side state).
|
||||
const origLog = console.log;
|
||||
const origErr = console.error;
|
||||
const origStdoutWrite = process.stdout.write.bind(process.stdout);
|
||||
const origStderrWrite = process.stderr.write.bind(process.stderr);
|
||||
|
||||
function silenceCli(): void {
|
||||
console.log = () => {};
|
||||
console.error = () => {};
|
||||
(process.stdout as unknown as { write: unknown }).write = (() => true) as unknown as typeof process.stdout.write;
|
||||
(process.stderr as unknown as { write: unknown }).write = (() => true) as unknown as typeof process.stderr.write;
|
||||
}
|
||||
|
||||
function restoreCli(): void {
|
||||
console.log = origLog;
|
||||
console.error = origErr;
|
||||
(process.stdout as unknown as { write: unknown }).write = origStdoutWrite;
|
||||
(process.stderr as unknown as { write: unknown }).write = origStderrWrite;
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
setCliOptions({ quiet: true, progressJson: false, progressInterval: 1000, explain: false, timeoutMs: null });
|
||||
}, 60_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await engine.executeRaw('DELETE FROM links');
|
||||
await engine.executeRaw('DELETE FROM pages');
|
||||
await engine.executeRaw('DELETE FROM op_checkpoints');
|
||||
});
|
||||
|
||||
async function seedEntities(): Promise<void> {
|
||||
await engine.putPage('companies/acme', { type: 'company', title: 'Acme Corp', compiled_truth: 'acme body', timeline: '', frontmatter: {} });
|
||||
await engine.putPage('people/alice', { type: 'person', title: 'Alice Example', compiled_truth: 'alice body', timeline: '', frontmatter: {} });
|
||||
}
|
||||
|
||||
async function seedContentPage(slug: string, body: string, type = 'note', timeline = ''): Promise<void> {
|
||||
await engine.putPage(slug, { type, title: slug, compiled_truth: body, timeline, frontmatter: {} });
|
||||
}
|
||||
|
||||
async function runByMention(args: string[]): Promise<void> {
|
||||
silenceCli();
|
||||
try {
|
||||
await runExtract(engine, ['links', '--by-mention', '--source', 'db', ...args]);
|
||||
} catch (e) {
|
||||
// process.exit throws in some paths — only swallow that one.
|
||||
if (!(e instanceof Error && e.message.startsWith('__test_exit:'))) throw e;
|
||||
} finally {
|
||||
restoreCli();
|
||||
}
|
||||
}
|
||||
|
||||
/** Compute the canonical gazetteer hash the way the production code does. */
|
||||
async function expectedGazetteerHash(): Promise<string> {
|
||||
// The gazetteer is built from entity pages by buildGazetteer; for tests
|
||||
// we just build it the same way the prod code does and hash sorted keys.
|
||||
const { buildGazetteer } = await import('../src/core/by-mention.ts');
|
||||
const gz = await buildGazetteer(engine);
|
||||
return createHash('sha256').update([...gz.keys()].sort().join('|')).digest('hex').slice(0, 8);
|
||||
}
|
||||
|
||||
describe('by-mention checkpoint/resume (T5)', () => {
|
||||
test('clean exit clears the checkpoint (no row left in op_checkpoints)', async () => {
|
||||
await seedEntities();
|
||||
await seedContentPage('writing/post-1', 'We met with Acme Corp.');
|
||||
await runByMention([]);
|
||||
|
||||
const gh = await expectedGazetteerHash();
|
||||
const fp = mentionsFingerprint({ source: undefined, type: undefined, since: undefined, gazetteerHash: gh });
|
||||
const rows = await loadOpCheckpoint(engine, { op: 'extract-by-mention', fingerprint: fp });
|
||||
expect(rows.length).toBe(0); // cleared on clean exit
|
||||
});
|
||||
|
||||
test('dry-run does NOT write to op_checkpoints', async () => {
|
||||
await seedEntities();
|
||||
await seedContentPage('writing/post-1', 'Acme Corp here.');
|
||||
await runByMention(['--dry-run']);
|
||||
const rows = await engine.executeRaw<{ c: string }>(
|
||||
`SELECT COUNT(*)::text AS c FROM op_checkpoints WHERE op = 'extract-by-mention'`,
|
||||
[],
|
||||
);
|
||||
expect(Number(rows[0]!.c)).toBe(0);
|
||||
});
|
||||
|
||||
test('pre-seeded checkpoint causes resume — completed pages get skipped', async () => {
|
||||
await seedEntities();
|
||||
await seedContentPage('writing/already-scanned', 'Mentions Acme Corp here.');
|
||||
await seedContentPage('writing/pending', 'Mentions Alice Example here.');
|
||||
|
||||
// Seed a checkpoint that marks `writing/already-scanned` as completed.
|
||||
const gh = await expectedGazetteerHash();
|
||||
const fp = mentionsFingerprint({ source: undefined, type: undefined, since: undefined, gazetteerHash: gh });
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at)
|
||||
VALUES ('extract-by-mention', $1, $2::jsonb, NOW())`,
|
||||
[fp, JSON.stringify(['default::writing/already-scanned'])],
|
||||
);
|
||||
|
||||
await runByMention([]);
|
||||
|
||||
// Only the pending page should have links created.
|
||||
const linksFromPending = await engine.executeRaw<{ c: string }>(
|
||||
`SELECT COUNT(*)::text AS c FROM links l
|
||||
JOIN pages fp ON fp.id = l.from_page_id
|
||||
WHERE fp.slug = 'writing/pending' AND l.link_source = 'mentions'`,
|
||||
[],
|
||||
);
|
||||
const linksFromSkipped = await engine.executeRaw<{ c: string }>(
|
||||
`SELECT COUNT(*)::text AS c FROM links l
|
||||
JOIN pages fp ON fp.id = l.from_page_id
|
||||
WHERE fp.slug = 'writing/already-scanned' AND l.link_source = 'mentions'`,
|
||||
[],
|
||||
);
|
||||
expect(Number(linksFromPending[0]!.c)).toBeGreaterThanOrEqual(1);
|
||||
expect(Number(linksFromSkipped[0]!.c)).toBe(0); // skipped via checkpoint
|
||||
});
|
||||
|
||||
test('gazetteer change invalidates checkpoint — new entity → re-scan', async () => {
|
||||
// Run #1: 1 entity, 1 content page mentioning it → checkpoint cleared on exit
|
||||
await seedEntities();
|
||||
await seedContentPage('writing/post-1', 'Acme Corp.');
|
||||
await runByMention([]);
|
||||
|
||||
// Now add a new entity. The gazetteer hash changes → different
|
||||
// fingerprint → fresh checkpoint state (codex fix #3 regression guard).
|
||||
await engine.putPage('people/charlie', { type: 'person', title: 'Charlie Example', compiled_truth: 'body', timeline: '', frontmatter: {} });
|
||||
|
||||
const oldHash = createHash('sha256').update(
|
||||
['acme corp', 'alice example'].sort().join('|'),
|
||||
).digest('hex').slice(0, 8);
|
||||
const newHash = await expectedGazetteerHash();
|
||||
expect(newHash).not.toBe(oldHash);
|
||||
|
||||
const oldFp = mentionsFingerprint({ source: undefined, type: undefined, since: undefined, gazetteerHash: oldHash });
|
||||
const newFp = mentionsFingerprint({ source: undefined, type: undefined, since: undefined, gazetteerHash: newHash });
|
||||
expect(oldFp).not.toBe(newFp);
|
||||
});
|
||||
|
||||
test('filtered pages (--type miss) DO get checkpointed (codex fix #4)', async () => {
|
||||
await seedEntities();
|
||||
// Two pages: one matches --type filter, one doesn't
|
||||
await seedContentPage('writing/match', 'Acme Corp.', 'meeting');
|
||||
await seedContentPage('writing/no-match', 'Acme Corp.', 'note');
|
||||
|
||||
await runByMention(['--type', 'meeting']);
|
||||
|
||||
const gh = await expectedGazetteerHash();
|
||||
const fp = mentionsFingerprint({ source: undefined, type: 'meeting', since: undefined, gazetteerHash: gh });
|
||||
// Checkpoint should have been cleared on clean exit. But the
|
||||
// observable signal that filtered pages got checkpointed too is
|
||||
// that the run finishes cleanly without errors AND completes.
|
||||
// (The pre-clear state would have all pages marked completed; we
|
||||
// verify on a paused run below.)
|
||||
const final = await loadOpCheckpoint(engine, { op: 'extract-by-mention', fingerprint: fp });
|
||||
expect(final.length).toBe(0); // cleared on clean exit
|
||||
|
||||
// Indirect check: confirm only the matching page produced links.
|
||||
const matchLinks = await engine.executeRaw<{ c: string }>(
|
||||
`SELECT COUNT(*)::text AS c FROM links l
|
||||
JOIN pages fp ON fp.id = l.from_page_id
|
||||
WHERE fp.slug = 'writing/match' AND l.link_source = 'mentions'`,
|
||||
[],
|
||||
);
|
||||
const nomatchLinks = await engine.executeRaw<{ c: string }>(
|
||||
`SELECT COUNT(*)::text AS c FROM links l
|
||||
JOIN pages fp ON fp.id = l.from_page_id
|
||||
WHERE fp.slug = 'writing/no-match' AND l.link_source = 'mentions'`,
|
||||
[],
|
||||
);
|
||||
expect(Number(matchLinks[0]!.c)).toBeGreaterThanOrEqual(1);
|
||||
expect(Number(nomatchLinks[0]!.c)).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -175,21 +175,33 @@ describe('findCandidateDuplicates', () => {
|
||||
});
|
||||
|
||||
test('embedding cosine ordering when both sides have embeddings', async () => {
|
||||
// Use per-run unique entity_slug so the assertion is immune to any
|
||||
// cross-test pollution (no other test in the file uses 'embed-test',
|
||||
// but parallel CI shard runs have surfaced a flake where the
|
||||
// position-0 assertion failed without a visible assertion-detail in
|
||||
// the truncated log). The contract this test pins is "A ranks higher
|
||||
// than B because cos(A,query)=1.0 vs cos(B,query)=0.0" — assert that
|
||||
// RELATIONSHIP, not the absolute index, so any unrelated row in the
|
||||
// result set can't flip the test.
|
||||
const slug = `embed-test-${Math.random().toString(36).slice(2, 10)}`;
|
||||
await engine.insertFact(
|
||||
{ fact: 'A', kind: 'fact', entity_slug: 'embed-test', source: 'test', embedding: vec(1, 0, 0) },
|
||||
{ fact: 'A', kind: 'fact', entity_slug: slug, source: 'test', embedding: vec(1, 0, 0) },
|
||||
{ source_id: 'default' },
|
||||
);
|
||||
await engine.insertFact(
|
||||
{ fact: 'B', kind: 'fact', entity_slug: 'embed-test', source: 'test', embedding: vec(0, 1, 0) },
|
||||
{ fact: 'B', kind: 'fact', entity_slug: slug, source: 'test', embedding: vec(0, 1, 0) },
|
||||
{ source_id: 'default' },
|
||||
);
|
||||
const result = await engine.findCandidateDuplicates(
|
||||
'default', 'embed-test', 'q',
|
||||
'default', slug, 'q',
|
||||
{ embedding: vec(1, 0, 0) },
|
||||
);
|
||||
expect(result.length).toBeGreaterThanOrEqual(2);
|
||||
// Closest by cosine should come first.
|
||||
expect(result[0].fact).toBe('A');
|
||||
const aIdx = result.findIndex(r => r.fact === 'A');
|
||||
const bIdx = result.findIndex(r => r.fact === 'B');
|
||||
expect(aIdx).toBeGreaterThanOrEqual(0); // A is in the result
|
||||
expect(bIdx).toBeGreaterThanOrEqual(0); // B is in the result
|
||||
// Closest by cosine MUST come first.
|
||||
expect(aIdx).toBeLessThan(bIdx);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
// v0.41.19.0 — T5 of ops-fix-wave.
|
||||
//
|
||||
// Pins mentionsFingerprint determinism + sensitivity. Codex flagged that
|
||||
// the prior plan's fingerprint omitted gazetteer hash, so resuming a
|
||||
// paused by-mention run after adding new entity pages would silently
|
||||
// skip them. The gazetteer field below is the regression guard.
|
||||
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { mentionsFingerprint } from '../src/core/op-checkpoint.ts';
|
||||
|
||||
describe('mentionsFingerprint (T5 codex fix #3)', () => {
|
||||
test('same inputs → same fingerprint', () => {
|
||||
const a = mentionsFingerprint({
|
||||
source: 'default',
|
||||
type: 'meeting',
|
||||
since: '2026-01-01',
|
||||
gazetteerHash: 'abc12345',
|
||||
});
|
||||
const b = mentionsFingerprint({
|
||||
source: 'default',
|
||||
type: 'meeting',
|
||||
since: '2026-01-01',
|
||||
gazetteerHash: 'abc12345',
|
||||
});
|
||||
expect(a).toBe(b);
|
||||
});
|
||||
|
||||
test('different source → different fingerprint', () => {
|
||||
const a = mentionsFingerprint({ source: 'source-a', gazetteerHash: 'abc12345' });
|
||||
const b = mentionsFingerprint({ source: 'source-b', gazetteerHash: 'abc12345' });
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
|
||||
test('different type → different fingerprint', () => {
|
||||
const a = mentionsFingerprint({ type: 'meeting', gazetteerHash: 'abc12345' });
|
||||
const b = mentionsFingerprint({ type: 'article', gazetteerHash: 'abc12345' });
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
|
||||
test('different since → different fingerprint', () => {
|
||||
const a = mentionsFingerprint({ since: '2026-01-01', gazetteerHash: 'abc12345' });
|
||||
const b = mentionsFingerprint({ since: '2026-02-01', gazetteerHash: 'abc12345' });
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
|
||||
test('different gazetteer hash → different fingerprint (codex fix #3 regression guard)', () => {
|
||||
// The load-bearing assertion: if entity pages change mid-pause, the
|
||||
// gazetteer hash shifts and the checkpoint invalidates cleanly.
|
||||
// Without this, resumed runs would skip pages against a new gazetteer
|
||||
// and never re-scan them.
|
||||
const a = mentionsFingerprint({ source: 'default', gazetteerHash: 'aaaaaaaa' });
|
||||
const b = mentionsFingerprint({ source: 'default', gazetteerHash: 'bbbbbbbb' });
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
|
||||
test('optional fields default symmetrically', () => {
|
||||
// source omitted should equal source: 'default' explicit.
|
||||
const explicit = mentionsFingerprint({ source: 'default', gazetteerHash: 'abc12345' });
|
||||
const omitted = mentionsFingerprint({ gazetteerHash: 'abc12345' });
|
||||
expect(explicit).toBe(omitted);
|
||||
});
|
||||
|
||||
test('returns stable 8-char hex slice', () => {
|
||||
const fp = mentionsFingerprint({ source: 'default', gazetteerHash: 'abc12345' });
|
||||
expect(fp).toMatch(/^[0-9a-f]{8}$/);
|
||||
});
|
||||
});
|
||||
+20
-2
@@ -4,7 +4,7 @@
|
||||
// the public CLI entrypoint. Hermetic — uses Bun's subprocess to run
|
||||
// the CLI like a user would.
|
||||
|
||||
import { describe, expect, test, beforeEach, afterEach } from 'bun:test';
|
||||
import { describe, expect, test, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test';
|
||||
import { spawnSync } from 'node:child_process';
|
||||
import { mkdtempSync, rmSync, existsSync, readFileSync, writeFileSync, mkdirSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
@@ -12,6 +12,23 @@ import { join } from 'node:path';
|
||||
|
||||
const REPO_ROOT = join(import.meta.dir, '..');
|
||||
|
||||
// Default-isolated GBRAIN_HOME for every gbrain() call. Without this,
|
||||
// tests that read `~/.gbrain/config.json` inherit the developer's real
|
||||
// brain config — and sibling Conductor worktrees writing to the same
|
||||
// config (e.g. via `schema use` or `config set` during their own tests)
|
||||
// cause flakes (the failing test pre-fix saw `schema_pack: "gbrain-base-v2"`
|
||||
// from another worktree, which doesn't exist in the bundle, and got
|
||||
// exit 1 instead of the asserted 0).
|
||||
let DEFAULT_GBRAIN_HOME: string;
|
||||
|
||||
beforeAll(() => {
|
||||
DEFAULT_GBRAIN_HOME = mkdtempSync(join(tmpdir(), 'gbrain-schema-cli-default-'));
|
||||
});
|
||||
|
||||
afterAll(() => {
|
||||
rmSync(DEFAULT_GBRAIN_HOME, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
function gbrain(
|
||||
args: string[],
|
||||
extraEnv: Record<string, string> = {},
|
||||
@@ -19,10 +36,11 @@ function gbrain(
|
||||
// bun's spawnSync does NOT inherit env mutations done via process.env = ...,
|
||||
// so pass env explicitly. CLAUDE.md flags this pattern as load-bearing for
|
||||
// any subprocess test that needs GBRAIN_HOME isolation.
|
||||
const env = { ...process.env, GBRAIN_HOME: DEFAULT_GBRAIN_HOME, ...extraEnv };
|
||||
const result = spawnSync('bun', ['run', 'src/cli.ts', ...args], {
|
||||
cwd: REPO_ROOT,
|
||||
encoding: 'utf-8',
|
||||
env: { ...process.env, ...extraEnv },
|
||||
env,
|
||||
});
|
||||
return {
|
||||
stdout: result.stdout ?? '',
|
||||
|
||||
Reference in New Issue
Block a user