feat(pi-hermes-memory): 导入上游 v0.9.10 源码快照

从官方 Git 仓库导入 chandra447/pi-hermes-memory 在 tag v0.9.10
(df624056a2e8c68bc89ba0a81e1c120856ec9665,2026-10-02)的源码快照。
本次仅导入源码,尚未接入组合包:package.json、加载顺序、依赖与权限策略
均未改动。

- 原样导入 src/、tests/、docs/、scripts/、README、CHANGELOG、PLAN、
  AGENTS 与 LICENSE;该快照共 156 个上游文件。
- 剔除嵌套 .git、上游 .github 工作流、.ralph/、.claude/、node_modules、
  覆盖率与构建产物。
- 新增中文 UPSTREAM.md,记录来源,以及启用前必须完成的待决事项:
  在 pi-compaction 之前注册加载顺序、better-sqlite3 版本与根 allowScripts
  对齐、strip-ansi 解析确认。
- 说明选择 tag 而非 main 的原因:导入时上游 main 已领先至
  24fcd8ad26fc27badb8d7dbaf8034645de07b893(2026-10-11),选 tag 是为
  对应一个已发布、可复现的版本。

尚未接线。启用前还需反转 tests/extension-load.test.ts 中"不得加载
Hermes"的断言,并同步 pi-tool-search 分组、权限策略与文档。

验证:快照 156 个上游文件全部导入,目录内仅多出 UPSTREAM.md 一份;
已确认与上游 main 内容存在差异,证明取的是 tag 而非 main。
This commit is contained in:
云服务部-叶林立
2026-10-13 23:19:35 -07:00
parent 879fc5d37f
commit 312bb2ba5a
157 changed files with 52287 additions and 0 deletions
@@ -0,0 +1,725 @@
/**
* Auto-consolidation — when memory hits capacity, trigger automatic
* consolidation instead of returning an error.
*
* Default transport: in-process direct completion (same mechanism as
* background review — see review-memory-ops.ts), used only when a caller
* supplies model/modelRegistry access (the manual `/memory-consolidate`
* command has it; the automatic over-capacity consolidator registered on
* MemoryStore does not, since MemoryStore itself has no extension-runtime
* access, so that path stays subprocess-only). Falls back to a `pi -p`
* subprocess when direct mode is unavailable, declines, or fails.
*
* The subprocess child process modifies files on disk, so the parent MUST
* reload from disk after a subprocess-based consolidation completes.
*/
import { resolveProjectName, resolveProjectStore, type ProjectNameRef, type ProjectStoreRef } from "../project-context.js";
import * as fs from "node:fs/promises";
import * as path from "node:path";
import { createHash } from "node:crypto";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import { DatabaseManager } from "../store/db.js";
import { getMemoryUsageSignals, type MemoryUsageSignal } from "../store/sqlite-memory-store.js";
import {
CONSOLIDATION_PROMPT,
CONSOLIDATION_CHUNK_CHARS_MIN,
DEFAULT_CONSOLIDATION_CHUNK_CHARS,
MAX_CONSOLIDATION_ROUNDS,
DEFAULT_CONSOLIDATION_TIMEOUT_MS,
DIRECT_CONSOLIDATION_SYSTEM_PROMPT,
ENTRY_DELIMITER,
} from "../constants.js";
import type { ConsolidationResult, MemoryConfig } from "../types.js";
import { AGENT_ROOT } from "../paths.js";
import { execChildPrompt } from "./pi-child-process.js";
import { runDirectMemoryCompletion, usesDirectTransport, type DirectReviewContext } from "./review-memory-ops.js";
import { AtomicLockCoordinator } from "../store/atomic-lock-coordinator.js";
type MemoryTarget = "memory" | "user" | "failure";
type ToolMemoryTarget = MemoryTarget | "project";
type ConsolidationLlmConfig = Pick<
MemoryConfig,
"llmModelOverride" | "llmThinkingOverride" | "reviewTransport"
| "consolidationChunking" | "consolidationChunkChars" | "consolidationUsageSignals"
>;
// staleMs is deliberately decoupled from the consolidation timeout. The holder
// beats every CONSOLIDATION_LOCK_HEARTBEAT_MS while its child runs, so a
// legitimately slow consolidation (up to 2x timeoutMs once retryWithoutOverrides
// fires) never loses its lease, while a holder that stops making progress is
// reclaimable after seconds instead of after its worst-case runtime (#144).
const CONSOLIDATION_LOCK_STALE_MS = 45_000;
const CONSOLIDATION_LOCK_HEARTBEAT_MS = 10_000;
// Contention is usually transient. Poll for the lock the way
// acquireMarkdownMutationLock does instead of hard-failing the memory write
// that triggered auto-consolidation on the very first collision.
const CONSOLIDATION_LOCK_WAIT_MS = 5_000;
const CONSOLIDATION_LOCK_POLL_MS = 50;
const CONSOLIDATION_LOCK_ENV = "PI_HERMES_CONSOLIDATION_LOCK_DIR";
const CONSOLIDATION_LOCK_WAIT_ENV = "PI_HERMES_CONSOLIDATION_LOCK_WAIT_MS";
interface ConsolidationLock {
release: () => Promise<void>;
}
interface ConsolidationLockAttempt {
lock: ConsolidationLock | null;
/** True when the lock was held by someone else on the first attempt. */
contended: boolean;
waitedMs: number;
}
function consolidationLockRoot(): string {
return process.env[CONSOLIDATION_LOCK_ENV]?.trim()
|| path.join(AGENT_ROOT, "pi-hermes-memory", ".consolidation-locks");
}
function sanitizeLockPart(value: string): string {
return value.replace(/[^a-z0-9._-]+/gi, "_").slice(0, 80) || "unknown";
}
function consolidationLockKey(target: MemoryTarget, toolTarget: ToolMemoryTarget, storageIdentity: string): string {
const storageHash = createHash("sha256").update(storageIdentity).digest("hex");
return `${sanitizeLockPart(toolTarget)}:${sanitizeLockPart(target)}:${storageHash}`;
}
function consolidationLockWaitMs(): number {
const configured = Number(process.env[CONSOLIDATION_LOCK_WAIT_ENV]);
return Number.isFinite(configured) && configured >= 0 ? configured : CONSOLIDATION_LOCK_WAIT_MS;
}
async function acquireConsolidationLock(
store: MemoryStore,
target: MemoryTarget,
toolTarget: ToolMemoryTarget,
): Promise<ConsolidationLockAttempt> {
const storageIdentity = await store.getStorageIdentity(target);
const root = consolidationLockRoot();
await fs.mkdir(root, { recursive: true });
const coordinator = AtomicLockCoordinator.shared(path.join(root, "locks.sqlite"));
const key = consolidationLockKey(target, toolTarget, storageIdentity);
const lockOptions = { staleMs: CONSOLIDATION_LOCK_STALE_MS };
const startedAt = Date.now();
let lease = coordinator.tryAcquire(key, lockOptions);
const contended = !lease;
if (contended) {
const deadline = startedAt + consolidationLockWaitMs();
while (!lease && Date.now() < deadline) {
// Same shape as acquireMarkdownMutationLock; Promise.withResolvers would
// need an ES2024 lib this project does not target.
await new Promise((resolve) => setTimeout(resolve, CONSOLIDATION_LOCK_POLL_MS));
lease = coordinator.tryAcquire(key, lockOptions);
}
}
const waitedMs = Date.now() - startedAt;
if (!lease) return { lock: null, contended, waitedMs };
const held = lease;
const heartbeat = setInterval(() => {
try {
held.renew();
} catch {
// A missed beat only moves the lease closer to staleMs; the next beat
// recovers, and a permanently broken lock DB should not crash the run.
}
}, CONSOLIDATION_LOCK_HEARTBEAT_MS);
heartbeat.unref?.();
return {
lock: {
release: async () => {
clearInterval(heartbeat);
held.release();
},
},
contended,
waitedMs,
};
}
function entriesForTarget(store: MemoryStore, target: MemoryTarget): string[] {
if (target === "user") return store.getUserEntries();
if (target === "failure") return store.getAllFailureEntries();
return store.getMemoryEntries();
}
function labelForTarget(target: MemoryTarget, toolTarget: ToolMemoryTarget): string {
if (toolTarget === "project") return "Project Memory";
if (target === "user") return "User Profile";
if (target === "failure") return "Failure Memory";
return "Memory";
}
function describeConsolidationFailure(
result: { code: number; stderr?: string; killed?: boolean },
timeoutMs: number,
): string {
const stderr = result.stderr?.trim();
const terminated = result.killed || result.code === 124 || result.code === 143;
if (terminated) {
return `Consolidation subprocess was terminated (likely timeout or cancellation). Timeout: ${timeoutMs}ms. Raise consolidationTimeoutMs if consolidation legitimately needs longer.`;
}
return `Consolidation process exited with code ${result.code}: ${stderr?.slice(0, 200) || "unknown error"}`;
}
function buildConsolidationPrompt(
target: MemoryTarget,
toolTarget: ToolMemoryTarget,
entries: string[],
scoped = false,
usageSignals: Map<string, MemoryUsageSignal> | null = null,
): string {
const lines = [
CONSOLIDATION_PROMPT,
"",
`--- Current ${labelForTarget(target, toolTarget)} Entries ---`,
entries.join(ENTRY_DELIMITER) || "(empty)",
];
const signalsSection = usageSignalsSectionText(usageSignals, entries);
if (signalsSection) {
lines.push("", signalsSection);
}
lines.push(
"",
`Use memory_add, memory_replace, or memory_remove to consolidate. Target: '${toolTarget}'`,
);
if (scoped) {
// Chunked rounds present a slice of the store, but the child's memory tools
// can see everything. Without an explicit scope the model may modify entries
// it was never shown — observed in real runs (a round wiped 11 out-of-scope
// entries). Real runs 2026-09-10.
lines.push(
"This pass covers ONLY the entries listed above — they are one slice of a larger store being consolidated in rounds.",
"Do NOT add, modify, or remove any entry that is not listed above.",
);
}
return lines.join("\n");
}
/** Longest entry excerpt shown in a usage-signals line. */
const USAGE_SIGNAL_EXCERPT_CHARS = 80;
/** Caps per half (tracked / never-recalled) so a chunk of many small entries —
* or a mature store where everything has been recalled — cannot balloon a
* round's prompt past its bounded budget. Both halves fold their remainder
* into one summary line. */
const USAGE_SIGNAL_TRACKED_MAX_LINES = 20;
const USAGE_SIGNAL_NEVER_RECALLED_MAX_LINES = 20;
function usageSignalExcerpt(entry: string): string {
const flat = entry.replace(/\s+/g, " ").trim();
return flat.length > USAGE_SIGNAL_EXCERPT_CHARS
? `${flat.slice(0, USAGE_SIGNAL_EXCERPT_CHARS - 3)}...`
: flat;
}
/**
* Per-entry recall counts from the memory_search read path — the promotion-gate
* data telling the model which entries are load-bearing and which have never
* earned their place. Null (and therefore absent from prompts) until at least
* one presented entry has actually been recalled, so stores without tracking
* history keep byte-identical prompts.
*/
function usageSignalsSectionText(
usageSignals: Map<string, MemoryUsageSignal> | null,
entries: string[],
): string | null {
if (!usageSignals || usageSignals.size === 0 || entries.length === 0) return null;
const tracked: Array<{ entry: string; signal: MemoryUsageSignal }> = [];
const neverRecalledLines: string[] = [];
for (const entry of entries) {
const signal = usageSignals.get(entry.trim());
const excerpt = usageSignalExcerpt(entry);
if (signal) {
tracked.push({ entry, signal });
} else {
neverRecalledLines.push(`- never recalled: "${excerpt}"`);
}
}
if (tracked.length === 0) return null;
// Both halves are capped with a summary line for the remainder: a chunk of
// many small entries — or a mature store where every entry has been recalled
// — must not balloon a chunked round's prompt past its bounded budget.
// Tracked entries are ranked by recall count so the most load-bearing ones
// survive the cut.
tracked.sort((a, b) => b.signal.hits - a.signal.hits);
const trackedLines = tracked
.slice(0, USAGE_SIGNAL_TRACKED_MAX_LINES)
.map(({ signal, entry }) => `- recalled ${signal.hits}x, last ${signal.lastHit}: "${usageSignalExcerpt(entry)}"`);
const lines = [
"--- Usage Signals (memory_search recall tracking) ---",
...trackedLines,
];
if (tracked.length > USAGE_SIGNAL_TRACKED_MAX_LINES) {
lines.push(`- (+${tracked.length - USAGE_SIGNAL_TRACKED_MAX_LINES} more recalled entries omitted — showing the top ${USAGE_SIGNAL_TRACKED_MAX_LINES} by recall count)`);
}
lines.push(...neverRecalledLines.slice(0, USAGE_SIGNAL_NEVER_RECALLED_MAX_LINES));
if (neverRecalledLines.length > USAGE_SIGNAL_NEVER_RECALLED_MAX_LINES) {
lines.push(`- (+${neverRecalledLines.length - USAGE_SIGNAL_NEVER_RECALLED_MAX_LINES} more listed entries have no recorded recalls)`);
}
lines.push(
"Treat these as tie-breakers, not removal orders: well-recalled entries are load-bearing (prefer keeping or merging them); never-recalled entries are weaker keep candidates.",
);
return lines.join("\n");
}
/**
* Scoped recall stats for the store being consolidated, or null when tracking
* is off, no database is available, or the lookup fails (fail-open).
*/
function loadUsageSignals(
dbManager: DatabaseManager | null,
llmConfig: ConsolidationLlmConfig,
target: MemoryTarget,
toolTarget: ToolMemoryTarget,
projectName: string | null | undefined,
): Map<string, MemoryUsageSignal> | null {
if (!dbManager || llmConfig.consolidationUsageSignals === false) return null;
try {
return getMemoryUsageSignals(dbManager, {
target,
project: toolTarget === "project" ? (projectName ?? null) : null,
});
} catch {
return null;
}
}
function chunkCharsFor(config: ConsolidationLlmConfig): number {
const value = config.consolidationChunkChars;
return typeof value === "number" && Number.isFinite(value) && value >= CONSOLIDATION_CHUNK_CHARS_MIN
? value
: DEFAULT_CONSOLIDATION_CHUNK_CHARS;
}
/** A round shorter than this cannot plausibly boot a child and merge anything. */
const MIN_ROUND_MS = 10_000;
/**
* Head entries worth up to chunkChars of prompt text. Whole entries only; an
* entry larger than chunkChars travels alone so the loop always makes progress.
*/
export function takeChunk(entries: string[], chunkChars: number): string[] {
const batch: string[] = [];
let length = 0;
for (const entry of entries) {
const entryLength = entry.length + ENTRY_DELIMITER.length;
if (batch.length > 0 && length + entryLength > chunkChars) break;
batch.push(entry);
length += entryLength;
}
return batch;
}
export async function triggerConsolidation(
pi: ExtensionAPI,
store: MemoryStore,
target: MemoryTarget,
signal?: AbortSignal,
timeoutMs: number = DEFAULT_CONSOLIDATION_TIMEOUT_MS,
toolTarget: ToolMemoryTarget = target,
llmConfig: ConsolidationLlmConfig = {},
directCtx: DirectReviewContext | null = null,
dbManager: DatabaseManager | null = null,
projectName?: string | null,
deps: { runDirectMemoryCompletion?: typeof runDirectMemoryCompletion } = {},
): Promise<ConsolidationResult> {
const entries = entriesForTarget(store, target);
const currentContent = entries.join(ENTRY_DELIMITER);
const runDirect = deps.runDirectMemoryCompletion ?? runDirectMemoryCompletion;
const usageSignals = loadUsageSignals(dbManager, llmConfig, target, toolTarget, projectName);
if (directCtx && usesDirectTransport(llmConfig)) {
try {
const directEntriesHeader = `--- Current ${labelForTarget(target, toolTarget)} Entries (target: '${toolTarget}') ---`;
const directSignalsSection = usageSignalsSectionText(usageSignals, entries);
const directResult = await runDirect(
directCtx,
store,
toolTarget === "project" ? store : null,
{
systemPrompt: DIRECT_CONSOLIDATION_SYSTEM_PROMPT,
userPrompt: [
directEntriesHeader,
currentContent || "(empty)",
...(directSignalsSection ? ["", directSignalsSection] : []),
"",
`Only emit operations with "target": "${toolTarget}".`,
].join("\n"),
config: llmConfig,
timeoutMs,
signal,
requireAtomicShrink: true,
expectedTarget: toolTarget,
},
dbManager,
projectName,
);
// Consolidation only did its job if it actually freed space — unlike
// review/flush/correction, an empty or fully-skipped result here is a
// failure worth falling back to subprocess for, not a normal outcome.
if (directResult.ok && directResult.appliedCount > 0) {
return { consolidated: true };
}
// An empty completion is terminal (#235): the direct model answered
// with nothing in either channel, so the subprocess child would run
// the same model against the same server-side thinking default and
// fail the same way (#197). The success criterion is unchanged — it
// must actually shrink — but this case reports instead of acquiring
// the consolidation lock and spawning the child.
if (directResult.ok && directResult.fallbackReason === "empty_response") {
const modelRef = directCtx.model?.provider
? `${directCtx.model.provider}/${directCtx.model.id}`
: "model";
return {
consolidated: false,
error: `${modelRef} returned an empty completion; no consolidation attempted`,
};
}
} catch {
// Fall through to subprocess below.
}
}
const chunkChars = chunkCharsFor(llmConfig);
const chunkingEnabled = llmConfig.consolidationChunking === true;
const hasUsage = typeof (store as any).capacityUsage === "function";
const usageOf = (list: string[]): number =>
hasUsage ? (store as any).capacityUsage(target) : list.join(ENTRY_DELIMITER).length;
const goal = typeof store.capacityGoal === "function" ? store.capacityGoal(target) : chunkChars;
let lock: ConsolidationLock | null = null;
try {
const attempt = await acquireConsolidationLock(store, target, toolTarget);
lock = attempt.lock;
if (!lock) {
// Not a failure: the work is already running in another session. Say so
// plainly so the memory-write path can ask for a retry instead of
// reporting a broken consolidation mid-task (#144).
return {
consolidated: false,
deferred: true,
error: `Consolidation already in progress for target '${toolTarget}' in another session — nothing was consolidated here; retry shortly.`,
};
}
let promptEntries = entries;
if (attempt.contended) {
// We queued behind another session's consolidation and it has now
// finished. If it already freed space, running a second LLM pass here
// costs a child turn and over-compresses memory for nothing — hand the
// caller a reload-and-retry instead.
try {
await store.loadFromDisk();
const refreshed = entriesForTarget(store, target);
if (refreshed.join(ENTRY_DELIMITER).length < currentContent.length) {
return { consolidated: true };
}
promptEntries = refreshed;
} catch {
// Reload failed — consolidate the entries we already read instead.
}
}
if (!chunkingEnabled) {
// Legacy single-shot path — the flag default. Byte-identical to
// pre-chunking releases for stores of any size (usage signals are the
// one additive prompt section, and only once tracking data exists).
// When an oversized store times out, the error below names the remedy
// keys so the failure teaches the fix.
const result = await execChildPrompt(pi, buildConsolidationPrompt(target, toolTarget, promptEntries, false, usageSignals), llmConfig, {
signal,
timeoutMs,
retryWithoutOverrides: true,
}) as { code: number; stdout?: string; stderr?: string; killed?: boolean };
if (result.code === 0) {
return { consolidated: true };
}
let error = describeConsolidationFailure(result, timeoutMs);
const terminated = result.killed || result.code === 124 || result.code === 143;
if (terminated && promptEntries.join(ENTRY_DELIMITER).length > chunkChars) {
error += ` This store exceeds consolidationChunkChars (${chunkChars}) — enabling consolidationChunking splits consolidation into bounded rounds.`;
}
return { consolidated: false, error };
}
if (promptEntries.length === 0 || usageOf(promptEntries) <= goal) {
// Within the target's capacity goal (encoded units, same as the cap):
// nothing needs to shrink toward the goal — a clean no-op, not a
// failure. Covers healthy stores of any size, the failure tier, and
// manual triggers on stores that do not need consolidation.
return { consolidated: true, rounds: 0 };
}
const deadline = Date.now() + timeoutMs;
// Chunked path — the store exceeds one child run's prompt budget, and a
// single whole-store LLM merge is what produced the observed "subprocess
// terminated (likely timeout)" failures at cap scale. Rounds share the
// overall budget (deadline): each consolidates a slice, then reloads from
// disk (the child modified files) and re-evaluates. A store that fits one
// prompt runs a single UNSCOPED decisive round — the legacy single-shot
// call, same prompt, same timeout, one invocation. Resume needs no
// cursor: partial progress is already on disk.
let completedRounds = 0;
let progressRounds = 0;
const notes: string[] = [];
let offset = 0;
while (completedRounds < MAX_CONSOLIDATION_ROUNDS) {
if (promptEntries.length === 0) break; // everything merged away
const promptTotal = promptEntries.join(ENTRY_DELIMITER).length;
if (promptTotal <= chunkChars && usageOf(promptEntries) <= goal) break; // capacity goal met
if (signal?.aborted) {
notes.push("aborted between consolidation rounds");
break;
}
const remaining = deadline - Date.now();
if (remaining < MIN_ROUND_MS) {
notes.push(`consolidation time budget (${timeoutMs}ms) exhausted; retrigger consolidation to continue from current state`);
break;
}
// Removals in earlier rounds shift indices left; the walk offset may
// point past the (now shorter) store — wrap it instead of slicing an
// empty batch.
if (offset >= promptEntries.length) offset = 0;
const fitsOneChunk = promptTotal <= chunkChars;
const usageBefore = usageOf(promptEntries);
const batch = fitsOneChunk
? promptEntries // decisive round: whole remaining store, full context
: takeChunk(promptEntries.slice(offset), chunkChars);
const batchSet = new Set(batch);
const beforeRound = promptEntries;
const result = await execChildPrompt(pi, buildConsolidationPrompt(target, toolTarget, batch, !fitsOneChunk, usageSignals), llmConfig, {
signal,
timeoutMs: Math.min(timeoutMs, remaining),
retryWithoutOverrides: true,
}) as { code: number; stdout?: string; stderr?: string; killed?: boolean };
// Reload FIRST — even a failed round may have shrunk the store before
// dying, and the recount below decides how the failure is reported.
try {
await store.loadFromDisk();
promptEntries = entriesForTarget(store, target);
} catch {
notes.push("could not reload memory after a consolidation round");
break;
}
const currentRound = completedRounds + 1;
// Out-of-scope changes: entries that vanished this round without being
// part of the presented slice. The child can see the whole store through
// its tools, and a concurrent session may delete entries in this window
// too — the parent cannot tell a rogue deletion from a legitimate
// cross-slice dedup or a concurrent one, so out-of-scope disappearances
// are REPORTED, never resurrected (resurrection would fight legitimate
// dedup and ping-pong across triggers; the store's recovery snapshots
// remain the repair path for real damage).
const outOfScope = beforeRound.filter(
(entry) => !batchSet.has(entry) && !promptEntries.includes(entry),
);
if (outOfScope.length > 0) {
notes.push(`round ${currentRound} coincided with ${outOfScope.length} out-of-scope entr${outOfScope.length === 1 ? "y" : "ies"} disappearing (by the child or a concurrent writer) — inspect memory if that was not intended`);
}
const usageAfter = usageOf(promptEntries);
const shrank = usageAfter < usageBefore;
if (result.code !== 0) {
if (shrank) {
// The round shrank the store and THEN died (killed mid-merge). The
// shrink is real and on disk — report it as partial progress instead
// of a total failure, and let the next trigger resume.
progressRounds++;
notes.push(describeConsolidationFailure(result, Math.min(timeoutMs, remaining))
+ ` The failing round still shrank the store by ${usageBefore - usageAfter} chars; retrigger consolidation to continue.`);
} else {
notes.push(describeConsolidationFailure(result, Math.min(timeoutMs, remaining))
+ (completedRounds > 0
? ` ${completedRounds} earlier round${completedRounds === 1 ? "" : "s"} shrank the store; retrigger consolidation to continue.`
: ""));
}
break;
}
completedRounds++;
if (!shrank) {
// This round shrank nothing. Walk to the next slice rather than
// repeating it; when nothing in the walk yields, the round-cap exit
// below reports the store as still over its goal.
if (fitsOneChunk) {
notes.push(`consolidation could not shrink the remaining ${promptTotal} chars (capacity goal ${goal}); entries may be distinct facts worth keeping — consider manual pruning`);
break;
}
offset = (offset + batch.length) % Math.max(promptEntries.length, 1);
continue;
}
progressRounds++;
// Progress: keep walking forward past the slice this round consumed
// (removals shift indices left, so this is approximate) — resetting to
// the top would let a store whose head always yields a little progress
// starve the tail forever.
offset = (offset + batch.length) % Math.max(promptEntries.length, 1);
if (usageAfter <= goal) break; // capacity goal met
if (fitsOneChunk) break; // decisive round done; the next trigger starts a fresh pass
}
if (progressRounds > 0) {
const roundNotes = [...notes];
const usageEnd = usageOf(promptEntries);
if (usageEnd > goal) {
roundNotes.push(`store still ${usageEnd - goal} chars over its ${goal}-char capacity goal; entries may be distinct facts worth keeping — consider manual pruning or raising the limit`);
}
return {
consolidated: true,
partial: roundNotes.length > 0,
rounds: completedRounds,
...(roundNotes.length ? { error: roundNotes.join("; ") } : {}),
};
}
notes.push(`consolidation could not shrink the store toward its ${goal}-char capacity goal (${usageOf(promptEntries)} chars); entries may be distinct facts worth keeping — consider manual pruning or raising the limit`);
return {
consolidated: false,
error: notes.join("; "),
};
} catch (err) {
const message = String(err);
if (message.includes("extension ctx is stale")) {
// Session replaced/reloaded while consolidation was running. The new
// session re-initializes the store and will consolidate on its own next
// write, so this is a skip, not a failure — report it as deferred so the
// caller asks for a retry instead of surfacing a stale-ctx error.
return {
consolidated: false,
deferred: true,
error: "session replaced or reloaded during consolidation — will consolidate on next write",
};
}
return {
consolidated: false,
error: `Consolidation failed: ${message.slice(0, 200)}`,
};
} finally {
if (lock) {
try { await lock.release(); } catch { /* best-effort cleanup */ }
}
}
}
/**
* Register the /memory-consolidate command for manual consolidation.
*/
export function registerConsolidateCommand(
pi: ExtensionAPI,
store: MemoryStore,
timeoutMs: number = DEFAULT_CONSOLIDATION_TIMEOUT_MS,
projectStore: ProjectStoreRef = null,
projectName: ProjectNameRef = null,
llmConfig: ConsolidationLlmConfig = {},
dbManager: DatabaseManager | null = null,
deps: { runDirectMemoryCompletion?: typeof runDirectMemoryCompletion } = {},
): void {
pi.registerCommand("memory-consolidate", {
description: "Manually trigger memory consolidation to free up space",
handler: async (_args, ctx) => {
const results: string[] = [];
const activeProjectStore = resolveProjectStore(projectStore);
const activeProjectName = resolveProjectName(projectName);
const targets: Array<{
label: string;
store: MemoryStore;
target: MemoryTarget;
toolTarget: ToolMemoryTarget;
}> = [
{ label: "memory", store, target: "memory", toolTarget: "memory" },
{ label: "user", store, target: "user", toolTarget: "user" },
{ label: "failure", store, target: "failure", toolTarget: "failure" },
];
if (activeProjectStore) {
targets.push({
label: activeProjectName ? `project:${activeProjectName}` : "project",
store: activeProjectStore,
target: "memory",
toolTarget: "project",
});
}
try {
ctx.ui.notify(
`🔄 Starting memory consolidation for ${targets.length} target${targets.length === 1 ? "" : "s"}...`,
"info",
);
} catch {
// Best-effort only. If the command context is already stale, continue
// with the consolidation work rather than failing before it starts.
}
for (const item of targets) {
const entries = entriesForTarget(item.store, item.target);
if (entries.length === 0) {
results.push(`${item.label}: (empty, nothing to consolidate)`);
continue;
}
try {
ctx.ui.notify(
`⏳ Consolidating ${item.label}...`,
"info",
);
} catch {
// Best-effort progress feedback only.
}
const result = await triggerConsolidation(
pi,
item.store,
item.target,
ctx.signal,
timeoutMs,
item.toolTarget,
llmConfig,
ctx,
dbManager,
activeProjectName,
deps,
);
if (result.consolidated) {
await item.store.loadFromDisk();
const roundsNote = result.rounds ? ` (${result.rounds} round${result.rounds === 1 ? "" : "s"})` : "";
const partialNote = result.partial ? ` ⚠️ partial: ${result.error ?? "incomplete"}` : "";
results.push(`${item.label}: ✅ consolidated${roundsNote}${partialNote}`);
} else {
results.push(`${item.label}: ❌ ${result.error}`);
}
}
const summary = `\n 🔄 Memory Consolidation\n ${"─".repeat(30)}\n${results.map((r) => ` ${r}`).join("\n")}`;
try {
ctx.ui.notify(summary, "info");
} catch {
// Child consolidation can indirectly trigger a runtime reload/session
// replacement. If that happens, the original command ctx is stale by
// the time we reach the final summary, so the command should exit
// quietly instead of surfacing a stale-ctx error.
}
},
});
}
@@ -0,0 +1,358 @@
/**
* Background review — learning loop that auto-saves memory every N turns.
* Ported from hermes-agent/run_agent.py (_spawn_background_review, _memory_nudge_interval).
* See PLAN.md → "Hermes Source File Reference Map" for source lines.
*
* Default transport: in-process complete() side-channel (preserves parent LLM cache).
* Fallback: pi.exec("pi", ["-p", ...]) subprocess when direct path is unavailable.
*/
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import {
buildMemoryTargetRoutingGuidance,
COMBINED_REVIEW_PROMPT,
DIRECT_REVIEW_SYSTEM_PROMPT,
} from "../constants.js";
import { MemoryStore } from "../store/memory-store.js";
import { DatabaseManager } from "../store/db.js";
import type { MemoryConfig } from "../types.js";
import type { EnsureMemoryReady } from "../memory-initialization.js";
import { applyRecentMessageLimit, collectMessageParts } from "./message-parts.js";
import { execChildPrompt, resolveChildPiModel } from "./pi-child-process.js";
import { REVIEW_COMPLETION_TIMEOUT_MS, runDirectMemoryCompletion, usesDirectTransport, type DirectReviewResult } from "./review-memory-ops.js";
import { resolveProjectName, resolveProjectStore, type ProjectNameRef, type ProjectStoreRef } from "../project-context.js";
export interface BackgroundReviewOptions {
ensureMemoryReady?: EnsureMemoryReady;
dbManager?: DatabaseManager | null;
projectName?: ProjectNameRef;
deps?: BackgroundReviewDeps;
}
export interface BackgroundReviewDeps {
runDirectReview?: typeof runDirectMemoryCompletion;
execChildPrompt?: typeof execChildPrompt;
/** Test-only hook: called once runReview() has fully settled (after the
* fire-and-forget review work completes and activeReview clears),
* since production callers never await runReview() directly. */
onReviewSettled?: () => void;
/** Test-only override for session_shutdown's wait bound. */
shutdownGraceMs?: number;
}
export interface ReviewPromptInput {
parts: string[];
currentMemory: string;
currentUser: string;
currentProject: string | null;
}
export function buildSubprocessReviewPrompt(input: ReviewPromptInput): string {
const reviewPrompt = [
COMBINED_REVIEW_PROMPT,
"",
buildMemoryTargetRoutingGuidance(input.currentProject !== null),
"",
"--- Current Memory ---",
input.currentMemory || "(empty)",
"",
"--- Current User Profile ---",
input.currentUser || "(empty)",
];
if (input.currentProject !== null) {
reviewPrompt.push(
"",
"--- Current Project Memory ---",
input.currentProject || "(empty)",
);
}
reviewPrompt.push(
"",
"--- Conversation to Review ---",
input.parts.join("\n\n"),
);
return reviewPrompt.join("\n");
}
export function buildDirectReviewUserPrompt(input: ReviewPromptInput): string {
const sections = [
"--- Current Memory ---",
input.currentMemory || "(empty)",
"",
"--- Current User Profile ---",
input.currentUser || "(empty)",
];
if (input.currentProject !== null) {
sections.push(
"",
"--- Current Project Memory ---",
input.currentProject || "(empty)",
);
}
sections.push(
"",
"--- Conversation to Review ---",
input.parts.join("\n\n"),
);
return sections.join("\n");
}
function shouldNotifyDirect(result: DirectReviewResult): boolean {
return result.ok && result.appliedCount > 0;
}
function shouldNotifySubprocess(stdout: string | undefined): boolean {
const output = stdout?.trim();
return !!output && !output.toLowerCase().includes("nothing to save");
}
function diagnosticDetail(value: unknown): string {
const detail = value instanceof Error ? value.message : String(value ?? "").trim();
return detail.replace(/\s+/g, " ").slice(0, 300);
}
/** Shutdown wait only. Review still uses its own 120s timeout; 1s covers abort
* listener + cancel-file write without hanging Pi's awaited session_shutdown. */
const SESSION_REVIEW_SHUTDOWN_GRACE_MS = 1000;
function awaitUpTo(promise: Promise<void>, ms: number): Promise<void> {
return new Promise((resolve) => {
const timer = setTimeout(resolve, ms);
promise.then(
() => {
clearTimeout(timer);
resolve();
},
() => {
clearTimeout(timer);
resolve();
},
);
});
}
async function runSubprocessReview(
pi: ExtensionAPI,
prompt: string,
config: MemoryConfig,
execChild: typeof execChildPrompt,
ctx: Pick<ExtensionContext, "cwd" | "model">,
signal: AbortSignal,
): Promise<{ code: number; stdout?: string; stderr?: string }> {
return execChild(pi, prompt, config, {
cwd: ctx.cwd,
model: resolveChildPiModel(ctx.model),
// Session-scoped signal only. The turn signal belongs to the interactive
// agent run; forwarding it cancels unrelated review on a later user abort.
signal,
timeoutMs: REVIEW_COMPLETION_TIMEOUT_MS,
});
}
export function setupBackgroundReview(
pi: ExtensionAPI,
store: MemoryStore,
projectStore: ProjectStoreRef,
config: MemoryConfig,
options: BackgroundReviewOptions = {},
): void {
const dbManager = options.dbManager ?? null;
const projectName = options.projectName ?? null;
const runDirectReview = options.deps?.runDirectReview ?? runDirectMemoryCompletion;
const execChild = options.deps?.execChildPrompt ?? execChildPrompt;
const onReviewSettled = options.deps?.onReviewSettled;
const shutdownGraceMs = options.deps?.shutdownGraceMs ?? SESSION_REVIEW_SHUTDOWN_GRACE_MS;
let turnsSinceReview = 0;
let toolCallsSinceReview = 0;
let userTurnCount = 0;
let activeReview: Promise<void> | undefined;
const sessionAbort = new AbortController();
let shutdownPromise: Promise<void> | undefined;
const sessionCancelled = () => sessionAbort.signal.aborted;
const shutdownReview = (): Promise<void> => {
shutdownPromise ??= (async () => {
sessionAbort.abort();
const inFlight = activeReview;
if (inFlight) await awaitUpTo(inFlight, shutdownGraceMs);
})();
return shutdownPromise;
};
pi.on("session_shutdown", () => shutdownReview());
pi.on("message_end", async (event, _ctx) => {
if (event.message.role === "user") {
userTurnCount++;
}
});
pi.on("turn_end", async (event, ctx) => {
turnsSinceReview++;
if (!config.reviewEnabled) return;
if (sessionCancelled()) return;
if (activeReview) return;
try {
const msg = event.message;
if (msg?.role === "assistant") {
const content = msg?.content;
if (Array.isArray(content)) {
for (const block of content) {
if (block && typeof block === "object" && block.type === "toolCall") {
toolCallsSinceReview++;
}
}
}
}
} catch {
// If we can't count tool calls, fall back to turn-based only
}
const turnThresholdMet = turnsSinceReview >= config.nudgeInterval;
const toolCallThresholdMet = toolCallsSinceReview >= config.nudgeToolCalls;
if (!turnThresholdMet && !toolCallThresholdMet) return;
if (userTurnCount < 3) return;
if (sessionCancelled()) return;
turnsSinceReview = 0;
toolCallsSinceReview = 0;
const notifyIfSaved = (saved: boolean) => {
if (sessionCancelled()) return;
if (saved) {
ctx.ui.notify("💾 Memory auto-reviewed and updated", "info");
}
};
const notifyTransportFailure = (directFailure: string, subprocessDetail: unknown) => {
if (sessionCancelled()) return;
ctx.ui.notify(
`Memory auto-review failed in both transports. Direct: ${diagnosticDetail(directFailure)}. `
+ `Subprocess: ${diagnosticDetail(subprocessDetail)}. Check the active model/provider or set llmModelOverride.`,
"warning",
);
};
const runReview = async (): Promise<void> => {
if (sessionCancelled()) return;
let allParts: string[] = [];
try {
const entries = ctx.sessionManager.getBranch();
allParts = collectMessageParts(entries);
} catch {
return;
}
if (allParts.length < 4) return;
await options.ensureMemoryReady?.(ctx);
if (sessionCancelled()) return;
const parts = applyRecentMessageLimit(allParts, config.reviewRecentMessages);
const activeProjectStore = resolveProjectStore(projectStore);
const activeProjectName = resolveProjectName(projectName);
const promptInput: ReviewPromptInput = {
parts,
currentMemory: store.getMemoryEntries().join("\n§\n"),
currentUser: store.getUserEntries().join("\n§\n"),
currentProject: activeProjectStore ? activeProjectStore.getMemoryEntries().join("\n§\n") : null,
};
const subprocessPrompt = buildSubprocessReviewPrompt(promptInput);
const directPrompt = buildDirectReviewUserPrompt(promptInput);
let directFailure: string | undefined;
if (usesDirectTransport(config)) {
try {
const directResult = await runDirectReview(
ctx,
store,
activeProjectStore,
{
userPrompt: directPrompt,
systemPrompt: [
DIRECT_REVIEW_SYSTEM_PROMPT,
"",
buildMemoryTargetRoutingGuidance(activeProjectStore !== null),
].join("\n"),
config,
timeoutMs: REVIEW_COMPLETION_TIMEOUT_MS,
signal: sessionAbort.signal,
},
dbManager,
activeProjectName,
);
if (sessionCancelled()) return;
if (directResult.ok) {
notifyIfSaved(shouldNotifyDirect(directResult));
return;
}
if (directResult.fallbackReason === "empty" || directResult.fallbackReason === "aborted") {
return;
}
directFailure = [
directResult.fallbackReason ?? "failed",
directResult.error,
].filter(Boolean).join(": ");
} catch (error) {
if (sessionCancelled()) return;
directFailure = diagnosticDetail(error);
}
}
if (sessionCancelled()) return;
let subprocessResult: { code: number; stdout?: string; stderr?: string };
try {
subprocessResult = await runSubprocessReview(
pi,
subprocessPrompt,
config,
execChild,
ctx,
sessionAbort.signal,
);
} catch (error) {
if (directFailure) {
notifyTransportFailure(directFailure, error);
}
return;
}
if (sessionCancelled()) return;
if (subprocessResult.code === 0) {
notifyIfSaved(shouldNotifySubprocess(subprocessResult.stdout));
} else if (directFailure) {
const subprocessDetail = subprocessResult.stderr?.trim() || subprocessResult.stdout?.trim()
|| `exit code ${subprocessResult.code}`;
notifyTransportFailure(directFailure, subprocessDetail);
}
};
// Occupy the in-flight slot before runReview's synchronous preamble so a
// nested turn_end cannot start a second review.
activeReview = Promise.resolve()
.then(() => runReview())
.catch(() => {
// Best-effort only; transport failures are diagnosed after both paths settle.
})
.finally(() => {
activeReview = undefined;
onReviewSettled?.();
});
});
}
@@ -0,0 +1,90 @@
import { spawn } from "node:child_process";
import { existsSync } from "node:fs";
const [timeoutValue, cancellationPath, command, ...args] = process.argv.slice(2);
const timeoutMs = Number(timeoutValue);
if (!cancellationPath || !command || !Number.isFinite(timeoutMs) || timeoutMs <= 0) {
process.stderr.write("pi-hermes-memory watchdog: invalid invocation\n");
process.exit(2);
}
const child = spawn(command, args, {
detached: process.platform !== "win32",
stdio: ["ignore", "pipe", "pipe"],
});
child.stdout?.pipe(process.stdout);
child.stderr?.pipe(process.stderr);
let timedOut = false;
let cancelled = false;
let terminating = false;
let forceTimer;
function signalTree(signal) {
if (!child.pid) return;
if (process.platform === "win32") {
const killer = spawn("taskkill", ["/pid", String(child.pid), "/T", "/F"], {
stdio: "ignore",
windowsHide: true,
});
killer.unref();
return;
}
try {
process.kill(-child.pid, signal);
} catch {
try { child.kill(signal); } catch {}
}
}
function terminateTree() {
if (terminating) return;
terminating = true;
signalTree("SIGTERM");
forceTimer = setTimeout(() => signalTree("SIGKILL"), 500);
forceTimer.unref();
}
const timeout = setTimeout(() => {
timedOut = true;
process.stderr.write(`[pi-hermes-memory] child timed out after ${timeoutMs}ms; terminating process tree\n`);
terminateTree();
}, timeoutMs);
timeout.unref();
const cancellationPoll = cancellationPath === "-" ? undefined : setInterval(() => {
if (!existsSync(cancellationPath)) return;
cancelled = true;
process.stderr.write("[pi-hermes-memory] child cancellation requested; terminating process tree\n");
terminateTree();
}, 25);
cancellationPoll?.unref();
for (const signal of ["SIGTERM", "SIGINT"]) {
process.on(signal, terminateTree);
}
child.once("error", (error) => {
clearTimeout(timeout);
if (cancellationPoll) clearInterval(cancellationPoll);
if (forceTimer) clearTimeout(forceTimer);
process.stderr.write(`pi-hermes-memory watchdog: ${error.message}\n`);
process.exitCode = timedOut ? 124 : cancelled ? 143 : 127;
});
child.once("close", (code, signal) => {
clearTimeout(timeout);
if (cancellationPoll) clearInterval(cancellationPoll);
if (forceTimer) clearTimeout(forceTimer);
if (timedOut) {
process.exitCode = 124;
} else if (cancelled) {
process.exitCode = 143;
} else if (typeof code === "number") {
process.exitCode = code;
} else {
process.exitCode = signal === "SIGTERM" ? 143 : 1;
}
});
@@ -0,0 +1,305 @@
/**
* Correction detection — detects user corrections in real-time and triggers
* an immediate memory save instead of waiting for the next nudge interval.
*
* Uses a two-pass filter:
* - Strong patterns: always trigger (high confidence)
* - Weak patterns: only trigger if followed by a directive clause
* - Negative patterns: suppress even if a positive pattern matched
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import { DatabaseManager } from "../store/db.js";
import {
buildMemoryTargetRoutingGuidance,
CORRECTION_SAVE_PROMPT,
CORRECTION_STRONG_PATTERNS,
CORRECTION_WEAK_PATTERNS,
CORRECTION_NEGATIVE_PATTERNS,
CORRECTION_DIRECTIVE_WORDS,
DIRECT_CORRECTION_SYSTEM_PROMPT,
ENTRY_DELIMITER,
} from "../constants.js";
import type { MemoryConfig } from "../types.js";
import type { EnsureMemoryReady } from "../memory-initialization.js";
import { getMessageText } from "../types.js";
import { execChildPrompt, resolveChildPiModel } from "./pi-child-process.js";
import { resolveProjectName, resolveProjectStore, type ProjectNameRef, type ProjectStoreRef } from "../project-context.js";
import { runDirectMemoryCompletion, usesDirectTransport } from "./review-memory-ops.js";
/**
* Extract the directive part from a correction message.
* E.g., "no, use pnpm instead" -> "use pnpm instead"
*/
function extractCorrectionDirective(text: string): string {
// Remove common correction starters
const cleaned = text
.replace(/^(no|wrong|actually|stop|don'?t|that'?s not|I said|I told you)[,\.\s!]+/i, '')
.replace(/^(please\s+)?/i, '')
.trim();
return cleaned || text;
}
function compileCorrectionPatterns(
configured: string[] | undefined,
defaults: RegExp[],
): RegExp[] {
if (configured === undefined) return defaults;
const patterns: RegExp[] = [];
for (const source of configured) {
try {
patterns.push(new RegExp(source, "i"));
} catch {
// Ignore invalid configured regex entries; valid entries still apply.
}
}
return patterns;
}
function escapeRegexLiteral(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}
function hasDirectiveWord(remainder: string, words: string[]): boolean {
if (words.length === 0) return false;
const source = words.map(escapeRegexLiteral).join("|");
return new RegExp(`\\b(${source})\\b`, "i").test(remainder);
}
/**
* Check if a user message is a correction using the two-pass filter.
* Returns true if the message should trigger an immediate save.
*/
type CorrectionPatternConfig = Pick<MemoryConfig,
"correctionStrongPatterns" |
"correctionWeakPatterns" |
"correctionNegativePatterns" |
"correctionDirectiveWords"
>;
export function isCorrection(text: string, config?: CorrectionPatternConfig): boolean {
const negativePatterns = compileCorrectionPatterns(
config?.correctionNegativePatterns,
CORRECTION_NEGATIVE_PATTERNS,
);
const strongPatterns = compileCorrectionPatterns(
config?.correctionStrongPatterns,
CORRECTION_STRONG_PATTERNS,
);
const weakPatterns = compileCorrectionPatterns(
config?.correctionWeakPatterns,
CORRECTION_WEAK_PATTERNS,
);
const directiveWords = config?.correctionDirectiveWords ?? CORRECTION_DIRECTIVE_WORDS;
// Check negative patterns first — suppress even if positive matches
for (const pattern of negativePatterns) {
if (pattern.test(text)) return false;
}
// Check strong patterns — always trigger
for (const pattern of strongPatterns) {
if (pattern.test(text)) return true;
}
// Check weak patterns — only trigger if followed by a directive clause
for (const pattern of weakPatterns) {
if (pattern.test(text)) {
// Look for a directive after the weak pattern match
// Directive = a verb or "the/that/this" in the remainder of the text
const match = pattern.exec(text);
if (match && match.index === 0) {
const remainder = text.slice(match[0].length).trim();
// Simple heuristic: remainder contains something directive-ish
if (hasDirectiveWord(remainder, directiveWords)) {
return true;
}
}
}
}
return false;
}
export function setupCorrectionDetector(
pi: ExtensionAPI,
store: MemoryStore,
projectStore: ProjectStoreRef,
config: MemoryConfig,
dbManager: DatabaseManager | null = null,
projectName: ProjectNameRef = null,
deps: { runDirectMemoryCompletion?: typeof runDirectMemoryCompletion; ensureMemoryReady?: EnsureMemoryReady } = {},
): void {
if (!config.correctionDetection) return;
let pendingCorrection: string | undefined;
let turnsSinceLastCorrection = 3; // Start at threshold so first correction can fire immediately
let correctionInProgress = false;
const runDirect = deps.runDirectMemoryCompletion ?? runDirectMemoryCompletion;
// Flag on message_end (user role)
pi.on("message_end", async (event, _ctx) => {
if (event.message.role !== "user") return;
const text = getMessageText(event.message);
if (!text) return;
if (isCorrection(text, config)) {
pendingCorrection = text;
}
});
// Trigger on turn_end (we need full context: user correction + what agent said)
pi.on("turn_end", async (event, ctx) => {
if (!pendingCorrection) {
turnsSinceLastCorrection++;
return;
}
if (correctionInProgress) return;
const correctionText = pendingCorrection;
pendingCorrection = undefined;
// Rate limit: max 1 correction save per 3 turns
if (turnsSinceLastCorrection < 3) return;
correctionInProgress = true;
try {
try {
await deps.ensureMemoryReady?.(ctx, ctx.signal);
} catch {
// Keep the correction eligible for the next turn after a transient load
// failure, without consuming the successful-capture rate limit.
pendingCorrection ??= correctionText;
return;
}
turnsSinceLastCorrection = 0;
// Build conversation snapshot
const entries = ctx.sessionManager.getBranch();
const parts: string[] = [];
for (const entry of entries) {
if (entry.type !== "message") continue;
const msg = entry.message;
const text = getMessageText(msg);
if (!text) continue;
const prefix = msg.role === "user" ? "[USER]" : "[ASSISTANT]";
parts.push(`${prefix}: ${text}`);
}
// Only include last few exchanges (correction context is recent)
const recentParts = parts.slice(-6);
const activeProjectStore = resolveProjectStore(projectStore);
const activeProjectName = resolveProjectName(projectName);
const currentMemory = store.getMemoryEntries().join(ENTRY_DELIMITER);
const currentUser = store.getUserEntries().join(ENTRY_DELIMITER);
const currentProject = activeProjectStore ? activeProjectStore.getMemoryEntries().join(ENTRY_DELIMITER) : null;
const promptBody = [
"--- Current Memory ---",
currentMemory || "(empty)",
"",
"--- Current User Profile ---",
currentUser || "(empty)",
];
if (currentProject !== null) {
promptBody.push(
"",
"--- Current Project Memory ---",
currentProject || "(empty)",
);
}
promptBody.push(
"",
"--- User Correction to Save ---",
correctionText,
"",
"--- Recent Conversation ---",
recentParts.join("\n\n"),
);
let savedViaLlm = false;
const runSubprocessCorrection = async (): Promise<void> => {
const subprocessPrompt = [
CORRECTION_SAVE_PROMPT,
"",
buildMemoryTargetRoutingGuidance(activeProjectStore !== null),
"",
...promptBody,
].join("\n");
const result = await execChildPrompt(pi, subprocessPrompt, config, {
cwd: ctx.cwd,
model: resolveChildPiModel(ctx.model),
signal: ctx.signal,
timeoutMs: 30000,
});
if (result.code === 0 && result.stdout) {
const output = result.stdout.trim();
savedViaLlm = !!output && !output.toLowerCase().includes("nothing to save");
}
};
let handledDirect = false;
if (usesDirectTransport(config)) {
try {
const directResult = await runDirect(
ctx,
store,
activeProjectStore,
{
systemPrompt: [
DIRECT_CORRECTION_SYSTEM_PROMPT,
"",
buildMemoryTargetRoutingGuidance(activeProjectStore !== null),
].join("\n"),
userPrompt: promptBody.join("\n"),
config,
timeoutMs: 30000,
signal: ctx.signal,
},
dbManager,
activeProjectName,
);
if (directResult.ok) {
savedViaLlm = directResult.appliedCount > 0;
handledDirect = true;
}
} catch {
// Fall through to subprocess below.
}
}
if (!handledDirect) {
await runSubprocessCorrection();
}
if (savedViaLlm) {
ctx.ui.notify("🔧 Correction detected — memory updated", "info");
}
// Also save as a failure memory for learning
try {
if (correctionText) {
const directive = extractCorrectionDirective(correctionText);
const failureReason = "User corrected the agent";
const scopedProjectName = activeProjectStore ? activeProjectName : null;
await store.addFailure(directive, {
category: "correction",
failureReason,
project: scopedProjectName ?? undefined,
});
}
} catch {
// Best-effort — don't block the session
}
} catch {
// Best-effort — don't block the session
} finally {
correctionInProgress = false;
}
});
}
@@ -0,0 +1,91 @@
/**
* Index sessions command — /memory-index-sessions imports past sessions into SQLite.
*/
import path from 'node:path';
import fs from 'node:fs';
import type { ExtensionAPI, ExtensionCommandContext } from "@earendil-works/pi-coding-agent";
import { DatabaseManager } from '../store/db.js';
import { indexAllSessions, getSessionStats, retentionCutoffMs } from '../store/session-indexer.js';
import type { MemoryConfig } from '../types.js';
import { AGENT_ROOT } from '../paths.js';
const SESSIONS_DIR = process.env.PI_CODING_AGENT_SESSION_DIR || path.join(AGENT_ROOT, 'sessions');
export function registerIndexSessionsCommand(pi: ExtensionAPI, config: MemoryConfig): void {
pi.registerCommand("memory-index-sessions", {
description: "Import past Pi sessions into the search database",
handler: async (_args, ctx: ExtensionCommandContext) => {
// Show initial progress
ctx.ui.notify('🔍 Scanning session directories...', 'info');
try {
// Count sessions first for progress display
let totalFiles = 0;
let projectDirs: string[] = [];
if (fs.existsSync(SESSIONS_DIR)) {
projectDirs = fs.readdirSync(SESSIONS_DIR)
.filter(d => fs.statSync(path.join(SESSIONS_DIR, d)).isDirectory());
for (const dir of projectDirs) {
const files = fs.readdirSync(path.join(SESSIONS_DIR, dir))
.filter(f => f.endsWith('.jsonl'));
totalFiles += files.length;
}
}
ctx.ui.notify(`📁 Found ${totalFiles} session files across ${projectDirs.length} projects\n⏳ Indexing...`, 'info');
const memoryDir = path.join(AGENT_ROOT, 'pi-hermes-memory');
const dbManager = new DatabaseManager(memoryDir);
try {
// Retention is honored here too: a manual reindex must not re-add the
// expired sessions the auto pruning just deleted.
const result = indexAllSessions(dbManager, SESSIONS_DIR, undefined, retentionCutoffMs(config.sessionRetentionDays));
const stats = getSessionStats(dbManager);
let output = `\n✅ Session indexing complete!\n\n`;
output += `📊 Results:\n`;
output += `├─ Sessions processed: ${result.sessionsProcessed}\n`;
output += `├─ Sessions indexed: ${result.sessionsIndexed}\n`;
output += `├─ Sessions skipped (already indexed): ${result.sessionsSkipped}\n`;
if (result.expiredSkipped) {
output += `├─ Sessions skipped (outside retention): ${result.expiredSkipped}\n`;
}
output += `└─ Messages indexed: ${result.messagesIndexed}\n`;
if (stats.projects.length > 0) {
output += `\n📁 Projects indexed:\n`;
for (const p of stats.projects) {
output += `├─ ${p.project}: ${p.sessions} sessions, ${p.messages} messages\n`;
}
}
// Show totals
output += `\n📈 Database totals:\n`;
output += `├─ ${stats.totalSessions} sessions\n`;
output += `├─ ${stats.totalMessages} messages\n`;
output += `└─ ${stats.projects.length} projects\n`;
if (result.errors.length > 0) {
output += `\n⚠️ Errors (${result.errors.length}):\n`;
for (const err of result.errors.slice(0, 3)) {
output += `├─ ${err}\n`;
}
if (result.errors.length > 3) {
output += `└─ ... and ${result.errors.length - 3} more\n`;
}
}
output += `\n💡 Use the session_search tool to search across indexed sessions.`;
ctx.ui.notify(output, 'info');
} finally {
dbManager.close();
}
} catch (err) {
ctx.ui.notify(`❌ Session indexing failed: ${err instanceof Error ? err.message : String(err)}`, 'error');
}
},
});
}
+152
View File
@@ -0,0 +1,152 @@
/**
* Insights command — /memory-insights shows what's stored in persistent memory.
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import { DatabaseManager } from "../store/db.js";
import { getMemoryUsageSignals, type MemoryUsageSignal } from "../store/sqlite-memory-store.js";
import { resolveProjectName, resolveProjectStore, type ProjectNameRef, type ProjectStoreRef } from "../project-context.js";
/** Entries shown in the usage section's top-recalled list. */
const USAGE_TOP_ENTRIES = 5;
/** Character cap for entry excerpts in the usage section. */
const USAGE_EXCERPT_CHARS = 70;
interface UsageScopeStats {
total: number;
tracked: number;
top: Array<{ content: string; signal: MemoryUsageSignal }>;
}
function usageExcerpt(content: string): string {
const flat = content.replace(/\s+/g, " ").trim();
return flat.length > USAGE_EXCERPT_CHARS
? `${flat.slice(0, USAGE_EXCERPT_CHARS - 3)}...`
: flat;
}
/**
* Recall stats for one store scope. Fail-open: a null dbManager or any lookup
* error renders the insights command without the usage section.
*/
function usageScopeStats(dbManager: DatabaseManager | null, target: "memory" | "user" | "failure", project: string | null): UsageScopeStats | null {
if (!dbManager) return null;
try {
const signals = getMemoryUsageSignals(dbManager, { target, project });
const total = (dbManager.getDb().prepare(
"SELECT COUNT(*) as count FROM memories WHERE target = ? AND project IS ?",
).get(target, project) as { count: number | bigint }).count;
const top = [...signals.entries()]
.sort((a, b) => b[1].hits - a[1].hits)
.slice(0, USAGE_TOP_ENTRIES)
.map(([content, signal]) => ({ content, signal }));
const numTotal = typeof total === "bigint" ? Number(total) : total;
return { total: numTotal, tracked: signals.size, top };
} catch {
return null;
}
}
function renderUsageScope(label: string, stats: UsageScopeStats | null): string[] {
if (!stats || stats.total === 0) return [];
const lines = [` 🔍 USAGE — ${label}: ${stats.tracked} of ${stats.total} entries recalled via memory_search`];
if (stats.top.length === 0) {
lines.push(" (no recorded recalls yet — usage accrues as memory_search runs)");
} else {
stats.top.forEach((entry, i) => {
lines.push(` ${i + 1}. ${entry.signal.hits}x (last ${entry.signal.lastHit}) — ${usageExcerpt(entry.content)}`);
});
}
return lines;
}
export function registerInsightsCommand(
pi: ExtensionAPI,
store: MemoryStore,
projectStore: ProjectStoreRef,
projectName: ProjectNameRef,
dbManager: DatabaseManager | null = null,
): void {
pi.registerCommand("memory-insights", {
description: "Show what's stored in persistent memory",
handler: async (_args, ctx) => {
const memoryEntries = store.getMemoryEntries();
const userEntries = store.getUserEntries();
const activeProjectStore = resolveProjectStore(projectStore);
const activeProjectName = resolveProjectName(projectName);
const projectEntries = activeProjectStore ? activeProjectStore.getMemoryEntries() : null;
const lines: string[] = [];
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 🧠 Memory Insights ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
// Memory section
lines.push(" 📋 MEMORY (your personal notes)");
lines.push(" " + "─".repeat(44));
if (memoryEntries.length === 0) {
lines.push(" (empty)");
} else {
for (let i = 0; i < memoryEntries.length; i++) {
const preview =
memoryEntries[i].length > 100
? memoryEntries[i].slice(0, 100) + "..."
: memoryEntries[i];
lines.push(` ${i + 1}. ${preview}`);
}
}
lines.push("");
// User section
lines.push(" 👤 USER PROFILE");
lines.push(" " + "─".repeat(44));
if (userEntries.length === 0) {
lines.push(" (empty)");
} else {
for (let i = 0; i < userEntries.length; i++) {
const preview =
userEntries[i].length > 100
? userEntries[i].slice(0, 100) + "..."
: userEntries[i];
lines.push(` ${i + 1}. ${preview}`);
}
}
lines.push("");
if (projectEntries !== null) {
lines.push(` 📁 PROJECT MEMORY: ${activeProjectName ?? ""}`);
lines.push(" " + "─".repeat(44));
if (projectEntries.length === 0) {
lines.push(" (empty)");
} else {
for (let i = 0; i < projectEntries.length; i++) {
const preview =
projectEntries[i].length > 100
? projectEntries[i].slice(0, 100) + "..."
: projectEntries[i];
lines.push(` ${i + 1}. ${preview}`);
}
}
lines.push("");
}
// Usage section (promotion-gate data: which entries actually get recalled)
const usageLines = [
...renderUsageScope("global memory", usageScopeStats(dbManager, "memory", null)),
...renderUsageScope(
`project:${activeProjectName ?? ""}`,
activeProjectStore && activeProjectName ? usageScopeStats(dbManager, "memory", activeProjectName) : null,
),
];
if (usageLines.length > 0) {
lines.push(...usageLines);
lines.push("");
}
ctx.ui.notify(lines.join("\n"), "info");
},
});
}
@@ -0,0 +1,37 @@
/**
* Interview command — /memory-interview guides new users through a brief
* onboarding interview to pre-fill their USER.md profile.
*
* This eliminates the "empty memory cold start" problem where users get
* zero value until multiple sessions accumulate facts organically.
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import { INTERVIEW_PROMPT } from "../constants.js";
export function registerInterviewCommand(
pi: ExtensionAPI,
store: MemoryStore,
): void {
pi.registerCommand("memory-interview", {
description: "Answer a few questions to pre-fill your user profile so the agent remembers you across sessions",
handler: async (_args, ctx) => {
const userEntries = store.getUserEntries();
if (userEntries.length > 0) {
// User already has profile entries — acknowledge and offer choices
ctx.ui.notify(
`\n 🧠 You already have ${userEntries.length} profile entr${userEntries.length === 1 ? "y" : "ies"}:\n` +
userEntries.map((e) => ` • ${e.slice(0, 80)}${e.length > 80 ? "..." : ""}`).join("\n") +
"\n\n Starting the interview will add to or update these.\n",
"info",
);
}
// Send the interview prompt as a user message to trigger the agent turn
await ctx.waitForIdle();
pi.sendUserMessage(INTERVIEW_PROMPT);
},
});
}
@@ -0,0 +1,186 @@
/**
* Learn memory tool command — /learn-memory-tool teaches users about the memory system.
*/
import type { ExtensionAPI, ExtensionCommandContext } from "@earendil-works/pi-coding-agent";
export function registerLearnMemoryCommand(pi: ExtensionAPI): void {
pi.registerCommand("learn-memory-tool", {
description: "Learn how to use the pi-hermes-memory extension effectively",
handler: async (_args, ctx: ExtensionCommandContext) => {
// Show main menu first
const section = await ctx.ui.select("Pi Hermes Memory Guide", [
"📦 What Gets Saved",
"🔧 Tools Available",
"📋 Commands",
"✅ Best Practices",
"🔄 How Memory Flows",
"🏗️ Architecture",
"❓ Troubleshooting",
], {});
if (!section) return;
const lines: string[] = [];
if (section.startsWith("📦")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 📦 What Gets Saved ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" Type │ File │ Limit");
lines.push(" ────────────────┼───────────────┼────────────");
lines.push(" 🧠 Memory │ MEMORY.md │ 5,000 chars");
lines.push(" 👤 User Profile │ USER.md │ 5,000 chars");
lines.push(" ⚠️ Failures │ failures.md │ 10,000 chars");
lines.push(" 📚 Skills │ Pi-native skill dirs │ Unlimited");
lines.push(" 💾 Extended │ sessions.db │ Unlimited");
lines.push("");
lines.push(" Memory: Facts — env details, project conventions, tool quirks");
lines.push(" User: Who you are — name, preferences, communication style");
lines.push(" Failures: What didn't work — corrections, failures, insights");
lines.push(" Skills: Procedures — how to debug, deploy, test");
lines.push(" Extended: SQLite search mirror for Markdown memory + backfill");
lines.push("");
lines.push(" Memory Categories:");
lines.push(" ─────────────────");
lines.push(" [failure] What was tried but didn't work");
lines.push(" [correction] User corrected the agent");
lines.push(" [insight] Learning from experience");
lines.push(" [preference] User preference");
lines.push(" [convention] Project convention");
lines.push(" [tool-quirk] Tool-specific knowledge");
}
if (section.startsWith("🔧")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 🔧 Tools Available ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" memory (add/replace/remove)");
lines.push(" Save, update, or delete memories");
lines.push(" Targets: memory, user, failure, project");
lines.push("");
lines.push(" skill_manage (create/view/patch/update/delete)");
lines.push(" Save reusable procedures");
lines.push("");
lines.push(" session_search");
lines.push(" Search past conversations across all sessions");
lines.push("");
lines.push(" memory_search");
lines.push(" Search the SQLite-backed memory mirror/store");
lines.push(" Filters: project, target, category");
lines.push(" Categories: failure, correction, insight, preference, convention, tool-quirk");
}
if (section.startsWith("📋")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 📋 Commands ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" /memory-insights Show everything stored in memory");
lines.push(" /memory-skills List all saved skills");
lines.push(" /memory-consolidate Manually trigger memory cleanup");
lines.push(" /memory-interview Answer questions to pre-fill profile");
lines.push(" /memory-switch-project List all project memories");
lines.push(" /memory-index-sessions Import past sessions for search");
lines.push(" /memory-sync-markdown Backfill Markdown memories into SQLite");
lines.push(" /memory-preview-context Show memory policy or legacy prompt blocks");
}
if (section.startsWith("✅")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ ✅ Best Practices ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" ✅ DO save:");
lines.push(" • User preferences (\"prefers pnpm\", \"uses vim\")");
lines.push(" • Environment facts (\"macOS M1\", \"Node 20\")");
lines.push(" • Corrections (\"don't use npm — use pnpm\")");
lines.push(" • Project conventions (\"monorepo with turborepo\")");
lines.push(" • Failures (\"tried localStorage — XSS vulnerability\")");
lines.push("");
lines.push(" ❌ DON'T save:");
lines.push(" • Task progress (\"finished implementing auth\")");
lines.push(" • Session outcomes (\"PR #42 was merged\")");
lines.push(" • Temporary state (\"currently debugging X\")");
}
if (section.startsWith("🔄")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 🔄 How Memory Flows ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" 1. Session starts → Compact memory policy is injected");
lines.push(" 2. During conversation → Agent searches memory when useful");
lines.push(" 3. Agent saves → Markdown memory + best-effort SQLite sync");
lines.push(" 4. Every 10 turns → Background review saves items");
lines.push(" 5. On correction → Immediate save as [correction] category");
lines.push(" 6. On failure → Saves what failed + why");
lines.push(" 7. When full → Auto-consolidation merges");
lines.push(" 8. Session ends → Final flush");
lines.push("");
lines.push(" Legacy mode: set memoryMode=\"legacy-inject\" to restore full");
lines.push(" MEMORY.md, USER.md, project memory, and failure prompt blocks.");
}
if (section.startsWith("🏗️")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 🏗️ Two-Tier Architecture ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" Default Prompt Context");
lines.push(" ┌─────────────────────────────────────┐");
lines.push(" │ <memory-policy> only │");
lines.push(" │ Explains when to use memory_search │");
lines.push(" │ Memory is context, not instruction │");
lines.push(" │ Repo/tool evidence wins │");
lines.push(" └─────────────────────────────────────┘");
lines.push("");
lines.push(" Searchable on Demand");
lines.push(" ┌─────────────────────────────────────┐");
lines.push(" │ MEMORY.md / USER.md / failures.md │");
lines.push(" │ projects-memory/<project>/MEMORY.md │");
lines.push(" │ session_search(\"auth flow\") │");
lines.push(" │ memory_search(\"testing patterns\") │");
lines.push(" │ /memory-sync-markdown (backfill old md)│");
lines.push(" │ memory_search(\"auth\", cat:\"failure\")│");
lines.push(" └─────────────────────────────────────┘");
lines.push("");
lines.push(" Legacy mode can still inject full memory blocks for users");
lines.push(" who explicitly opt into memoryMode=\"legacy-inject\".");
}
if (section.startsWith("❓")) {
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ ❓ Troubleshooting ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" \"Memory is full\"");
lines.push(" → /memory-consolidate to merge entries");
lines.push(" → If it still fails, the save does NOT silently become SQLite-only");
lines.push("");
lines.push(" \"Can't find something\"");
lines.push(" → memory_search to search the SQLite mirror/store");
lines.push(" → /memory-sync-markdown to import older Markdown entries");
lines.push("");
lines.push(" \"Agent forgot something\"");
lines.push(" → Check /memory-insights, tell agent \"remember X\"");
lines.push("");
lines.push(" \"Want to edit manually\"");
lines.push(" → Files at ~/.pi/agent/memory/ (plain markdown)");
}
if (lines.length > 0) {
ctx.ui.notify(lines.join("\n"), "info");
}
},
});
}
@@ -0,0 +1,27 @@
import { getMessageText } from "../types.js";
export function applyRecentMessageLimit(parts: string[], recentMessages = 0): string[] {
if (Number.isFinite(recentMessages) && recentMessages > 0) {
return parts.slice(-recentMessages);
}
return parts;
}
export function collectMessageParts(entries: unknown[], recentMessages = 0): string[] {
const parts: string[] = [];
for (const entry of entries) {
if (typeof entry !== "object" || entry === null) continue;
if ((entry as { type?: unknown }).type !== "message") continue;
const msg = (entry as { message?: unknown }).message;
const text = getMessageText(msg);
if (!text) continue;
const role = (msg as { role?: unknown } | null)?.role;
const prefix = role === "user" ? "[USER]" : "[ASSISTANT]";
parts.push(`${prefix}: ${text}`);
}
return applyRecentMessageLimit(parts, recentMessages);
}
@@ -0,0 +1,485 @@
import { existsSync, readFileSync, readdirSync } from "node:fs";
import * as fs from "node:fs/promises";
import * as os from "node:os";
import { dirname, join, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import type { MemoryConfig, ThinkingLevel } from "../types.js";
import { AGENT_ROOT } from "../paths.js";
type ChildLlmConfig = Pick<MemoryConfig, "llmModelOverride" | "llmThinkingOverride" | "childExtensionPaths">;
interface PiExecResult {
code: number;
stdout?: string;
stderr?: string;
killed?: boolean;
}
export interface ChildPiModel {
provider: string;
id: string;
}
export function resolveChildPiModel(
model: { provider?: string; id?: string } | undefined,
): ChildPiModel | undefined {
return model?.provider && model.id
? { provider: model.provider, id: model.id }
: undefined;
}
interface ExecChildPromptOptions {
signal?: AbortSignal;
cwd?: string;
model?: ChildPiModel;
timeoutMs: number;
retryWithoutOverrides?: boolean;
}
interface ExecChildPromptDependencies {
removeTemporaryDirectory: (dir: string) => Promise<void>;
}
const DEFAULT_EXEC_CHILD_PROMPT_DEPENDENCIES: ExecChildPromptDependencies = {
removeTemporaryDirectory: async (dir) => {
await fs.rm(dir, { recursive: true, force: true });
},
};
const WATCHDOG_EXIT_GRACE_MS = 5000;
const CHILD_PROCESS_WATCHDOG_PATH = fileURLToPath(
new URL("./child-process-watchdog.mjs", import.meta.url),
);
export interface ChildPiInvocation {
command: string;
args: string[];
}
interface ResolveChildPiInvocationOptions {
platform?: NodeJS.Platform;
execPath?: string;
argv?: string[];
piCliPath?: string | null;
}
const OVERRIDE_FAILURE_SUBJECT = /\b(model|provider|thinking)\b/i;
const OVERRIDE_FAILURE_REASON = /\b(not found|unknown|invalid|unsupported|unavailable|unrecognized|no match|no matches|cannot resolve|failed to resolve)\b/i;
// Resolve the path to pi-hermes-memory's own extension entry point.
// Used to pass -e <path> to child subprocesses so they only load this
// extension instead of all plugins from settings.json.
const OWN_EXTENSION_PATH: string = (() => {
try {
return resolve(dirname(fileURLToPath(import.meta.url)), "../index.ts");
} catch {
return "";
}
})();
function normalizedModelOverride(config: ChildLlmConfig): string | undefined {
const trimmed = config.llmModelOverride?.trim();
return trimmed ? trimmed : undefined;
}
function effectiveThinkingOverride(config: ChildLlmConfig): ThinkingLevel | undefined {
return config.llmThinkingOverride ?? (normalizedModelOverride(config) ? "off" : undefined);
}
export function hasChildLlmOverrides(config: ChildLlmConfig): boolean {
return normalizedModelOverride(config) !== undefined || effectiveThinkingOverride(config) !== undefined;
}
/** @deprecated No longer called after PR #78 — kept for API backward compat. */
export function inheritedExtensionArgs(argv: string[] = process.argv.slice(2)): string[] {
const args: string[] = [];
for (let i = 0; i < argv.length; i++) {
const current = argv[i];
if (current === "-e" || current === "--extension") {
const next = argv[i + 1];
if (typeof next === "string" && next.length > 0) {
args.push(current, next);
i++;
}
continue;
}
if (current.startsWith("--extension=")) {
args.push(current);
}
}
return args;
}
// Provider-auth-adapter packages (e.g. Anthropic/xAI/Codex OAuth) inject
// subscription billing headers via pi.registerProvider(). --no-extensions
// strips these from child `pi -p` subprocesses, which silently rebills
// subscription usage as pay-as-you-go "extra usage" instead (see issue #94).
//
// pi has no runtime API to enumerate loaded extensions or map a registered
// provider back to its extension file, so we can't ask pi "what adapter is
// active" directly. Instead we mirror pi's OWN static package-discovery
// convention (package.json -> "pi": { "extensions": [...] }, the same field
// pi-hermes-memory's own package.json declares) and match sibling package
// names against a naming convention, so a future xai-oauth-adapter or
// pi-codex-oauth-adapter is picked up automatically without a code change
// here — no code execution, just JSON reads of sibling package.json files.
const AUTH_ADAPTER_NAME_PATTERNS: readonly RegExp[] = [
/(^|[-/])oauth-adapter$/,
/(^|[-/])auth-adapter$/,
];
// These provider adapters predate the *-auth-adapter convention. Keep this
// list exact so an arbitrary suffixless *-auth package cannot access child
// maintenance prompts.
const AUTH_ADAPTER_PACKAGE_NAMES: ReadonlySet<string> = new Set([
"pi-claude-auth",
"@gotgenes/pi-anthropic-auth",
]);
function isAuthAdapterPackageName(name: string): boolean {
return (
AUTH_ADAPTER_PACKAGE_NAMES.has(name)
|| AUTH_ADAPTER_NAME_PATTERNS.some((pattern) => pattern.test(name))
);
}
// Read a sibling package's "pi": { "extensions": [...] } manifest field —
// the same field pi's own loader reads — and resolve declared paths
// relative to that package's directory. Mirrors loader.js#resolveExtensionEntries.
function readPackageExtensionEntries(packageDir: string): string[] {
const packageJsonPath = join(packageDir, "package.json");
if (!existsSync(packageJsonPath)) return [];
let manifest: unknown;
try {
manifest = JSON.parse(readFileSync(packageJsonPath, "utf-8"));
} catch {
return [];
}
const declaredExtensions = (manifest as { pi?: { extensions?: unknown } } | null)?.pi?.extensions;
if (!Array.isArray(declaredExtensions)) return [];
const entries: string[] = [];
for (const relativePath of declaredExtensions) {
if (typeof relativePath !== "string") continue;
const resolved = resolve(packageDir, relativePath);
if (existsSync(resolved)) entries.push(resolved);
}
return entries;
}
function scanRootForAuthAdapters(root: string): string[] {
let entries: string[];
try {
entries = readdirSync(root);
} catch {
return [];
}
const detected: string[] = [];
for (const entry of entries) {
if (entry.startsWith("@")) {
// Scoped org, e.g. @xai/pi-oauth-adapter — one extra level, no deeper.
const scopeDir = join(root, entry);
let scopedPackages: string[];
try {
scopedPackages = readdirSync(scopeDir);
} catch {
continue;
}
for (const scopedName of scopedPackages) {
if (!isAuthAdapterPackageName(`${entry}/${scopedName}`)) continue;
detected.push(...readPackageExtensionEntries(join(scopeDir, scopedName)));
}
continue;
}
if (!isAuthAdapterPackageName(entry)) continue;
detected.push(...readPackageExtensionEntries(join(root, entry)));
}
return detected;
}
// `roots` is overridable so tests can point this at fixture directories
// instead of the real sibling-packages trees. Two roots are scanned by
// default: packages installed alongside pi-hermes-memory's own package
// (the common npm-managed-extensions layout), and AGENT_ROOT's npm
// directory (covers pi-hermes-memory being loaded from elsewhere, e.g. a
// local dev checkout via -e, while the adapter is still npm-managed).
export function detectAuthAdapterExtensionPaths(roots?: string[]): string[] {
const searchRoots = roots ?? [
OWN_EXTENSION_PATH ? resolve(dirname(dirname(OWN_EXTENSION_PATH)), "..") : "",
join(AGENT_ROOT, "npm", "node_modules"),
].filter((root) => root.length > 0);
const seenRoots: string[] = [];
const detected: string[] = [];
for (const root of searchRoots) {
if (seenRoots.includes(root)) continue;
seenRoots.push(root);
detected.push(...scanRootForAuthAdapters(root));
}
return detected;
}
function childExtensionSources(config: ChildLlmConfig): string[] {
const seen = new Set<string>();
const sources: string[] = [];
const append = (candidate: string | undefined): void => {
const trimmed = candidate?.trim();
if (!trimmed || seen.has(trimmed)) return;
seen.add(trimmed);
sources.push(trimmed);
};
// These paths are discovered by Hermes rather than explicitly trusted in
// configuration, so keep the local existence check before forwarding them.
if (OWN_EXTENSION_PATH && existsSync(OWN_EXTENSION_PATH)) append(OWN_EXTENSION_PATH);
for (const source of config.childExtensionPaths ?? []) append(source);
for (const adapterPath of detectAuthAdapterExtensionPaths()) {
const normalized = resolve(adapterPath);
if (existsSync(normalized)) append(normalized);
}
return sources;
}
function appendOwnExtensionArgs(args: string[], config: ChildLlmConfig): void {
// Skip all packages from settings.json (--no-extensions) — the subprocess
// loads only Hermes and explicitly trusted provider/auth sources. Leave
// configured sources untouched so Pi's -e resolver owns path expansion and
// package-source handling exactly as it does for normal CLI invocations.
args.push("--no-extensions");
for (const extensionSource of childExtensionSources(config)) {
args.push("-e", extensionSource);
}
}
export function buildChildPiPromptArgs(
prompt: string,
config: ChildLlmConfig,
_argv: string[] = process.argv.slice(2),
activeModel?: ChildPiModel,
): string[] {
const args = ["-p", "--no-session"];
const model = normalizedModelOverride(config)
?? (activeModel?.provider && activeModel.id ? `${activeModel.provider}/${activeModel.id}` : undefined);
const thinking = effectiveThinkingOverride(config);
if (model) args.push("--model", model);
if (thinking) args.push("--thinking", thinking);
appendOwnExtensionArgs(args, config);
args.push(prompt);
return args;
}
function basePromptArgs(prompt: string, config: ChildLlmConfig, activeModel?: ChildPiModel): string[] {
// Always use --no-extensions + own path so the retry also avoids loading
// all settings.json packages — matching the primary code path.
const args = ["-p", "--no-session"];
if (activeModel?.provider && activeModel.id) {
args.push("--model", `${activeModel.provider}/${activeModel.id}`);
}
appendOwnExtensionArgs(args, config);
args.push(prompt);
return args;
}
function isCliJsPath(value: string | undefined): value is string {
if (!value) return false;
return value.replace(/\\/g, "/").toLowerCase().endsWith("/cli.js");
}
function resolvedInstalledPiCliPath(): string | undefined {
try {
const packageEntry = import.meta.resolve("@earendil-works/pi-coding-agent");
const entryPath = fileURLToPath(packageEntry);
const cliPath = join(dirname(entryPath), "cli.js");
return existsSync(cliPath) ? cliPath : undefined;
} catch {
return undefined;
}
}
function resolvedPiCliPath(options: ResolveChildPiInvocationOptions): string | undefined {
if (options.piCliPath !== undefined) {
return options.piCliPath ?? undefined;
}
const argv = options.argv ?? process.argv;
const currentCli = argv[1];
if (isCliJsPath(currentCli) && existsSync(currentCli)) {
return currentCli;
}
return resolvedInstalledPiCliPath();
}
function resolvedWindowsPiInvocation(
args: string[],
execPath: string,
): ChildPiInvocation | undefined {
const pathEntries = (process.env.PATH ?? process.env.Path ?? "")
.split(";")
.map((entry) => entry.trim().replace(/^"|"$/g, ""))
.filter(Boolean);
for (const directory of pathEntries) {
for (const executableName of ["pi.exe", "pi.com"]) {
const executablePath = join(directory, executableName);
if (existsSync(executablePath)) {
return { command: executablePath, args };
}
}
if (!existsSync(join(directory, "pi.cmd")) && !existsSync(join(directory, "pi.bat"))) {
continue;
}
for (const cliPath of [
join(directory, "node_modules", "@earendil-works", "pi-coding-agent", "dist", "cli.js"),
join(directory, "node_modules", "@earendil-works", "pi-coding-agent", "cli.js"),
]) {
if (existsSync(cliPath)) {
return { command: execPath, args: [cliPath, ...args] };
}
}
}
return undefined;
}
export function resolveChildPiInvocation(
args: string[],
options: ResolveChildPiInvocationOptions = {},
): ChildPiInvocation {
const platform = options.platform ?? process.platform;
if (platform !== "win32") {
return { command: "pi", args };
}
const piCliPath = resolvedPiCliPath(options);
if (piCliPath) {
return {
command: options.execPath ?? process.execPath,
args: [piCliPath, ...args],
};
}
const fallback = resolvedWindowsPiInvocation(args, options.execPath ?? process.execPath);
if (fallback) return fallback;
throw new Error("Unable to resolve a directly executable Pi CLI on Windows");
}
export function resolveWatchedChildPiInvocation(
invocation: ChildPiInvocation,
timeoutMs: number,
cancellationPath = "-",
): ChildPiInvocation {
return {
command: process.execPath,
args: [
CHILD_PROCESS_WATCHDOG_PATH,
String(timeoutMs),
cancellationPath,
invocation.command,
...invocation.args,
],
};
}
function shouldRetryWithoutOverridesFromText(text: string | undefined): boolean {
if (!text) return false;
return OVERRIDE_FAILURE_SUBJECT.test(text) && OVERRIDE_FAILURE_REASON.test(text);
}
function shouldRetryWithoutOverrides(result: PiExecResult): boolean {
return shouldRetryWithoutOverridesFromText(result.stderr) || shouldRetryWithoutOverridesFromText(result.stdout);
}
function shouldRetryWithoutOverridesForError(error: unknown): boolean {
return shouldRetryWithoutOverridesFromText(String(error));
}
async function writePromptToTemporaryFile(prompt: string): Promise<{ dir: string; filePath: string }> {
const dir = await fs.mkdtemp(join(os.tmpdir(), "pi-hermes-prompt-"));
const filePath = join(dir, "prompt.md");
try {
await fs.writeFile(filePath, prompt, { encoding: "utf-8", mode: 0o600 });
return { dir, filePath };
} catch (error) {
try { await fs.rm(dir, { recursive: true, force: true }); } catch {}
throw error;
}
}
export async function execChildPrompt(
pi: Pick<ExtensionAPI, "exec">,
prompt: string,
config: ChildLlmConfig,
options: ExecChildPromptOptions,
dependencies: ExecChildPromptDependencies = DEFAULT_EXEC_CHILD_PROMPT_DEPENDENCIES,
): Promise<PiExecResult> {
const execOptions = {
cwd: options.cwd,
timeout: options.timeoutMs + WATCHDOG_EXIT_GRACE_MS,
};
const temporaryPrompt = await writePromptToTemporaryFile(prompt);
const promptReference = `@${temporaryPrompt.filePath}`;
const cancellationPath = join(temporaryPrompt.dir, "cancel");
let cancellationRequest: Promise<void> | undefined;
const requestCancellation = () => {
cancellationRequest ??= fs.writeFile(cancellationPath, "", { mode: 0o600 }).catch(() => {});
};
options.signal?.addEventListener("abort", requestCancellation, { once: true });
if (options.signal?.aborted) {
requestCancellation();
await cancellationRequest;
}
try {
try {
const invocation = resolveWatchedChildPiInvocation(
resolveChildPiInvocation(buildChildPiPromptArgs(promptReference, config, process.argv.slice(2), options.model)),
options.timeoutMs,
cancellationPath,
);
const result = await pi.exec(invocation.command, invocation.args, execOptions) as PiExecResult;
if (
result.code === 0 ||
!options.retryWithoutOverrides ||
!hasChildLlmOverrides(config) ||
!shouldRetryWithoutOverrides(result)
) {
return result;
}
} catch (error) {
if (
!options.retryWithoutOverrides ||
!hasChildLlmOverrides(config) ||
!shouldRetryWithoutOverridesForError(error)
) {
throw error;
}
}
const retryInvocation = resolveWatchedChildPiInvocation(
resolveChildPiInvocation(basePromptArgs(promptReference, config, options.model)),
options.timeoutMs,
cancellationPath,
);
return await pi.exec(retryInvocation.command, retryInvocation.args, execOptions) as PiExecResult;
} finally {
options.signal?.removeEventListener("abort", requestCancellation);
try {
await dependencies.removeTemporaryDirectory(temporaryPrompt.dir);
} catch {
try { await fs.unlink(temporaryPrompt.filePath); } catch {}
}
}
}
@@ -0,0 +1,105 @@
/**
* Preview context command — /memory-preview-context shows the policy-only prompt
* or legacy memory blocks appended to the system prompt.
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import type { StandingInstructions } from "../store/standing-instructions.js";
import { resolveMemoryPolicyPrompt } from "../prompt-context.js";
import type { MemoryConfig } from "../types.js";
import { resolveProjectName, resolveProjectStore, type ProjectNameRef, type ProjectStoreRef } from "../project-context.js";
function appendStandingBlock(lines: string[], standing: StandingInstructions | null): number {
const rendered = standing?.render();
if (!rendered?.block) return 0;
lines.push(" ── STANDING INSTRUCTIONS (always injected) ─────────────────");
lines.push(rendered.block);
if (rendered.omittedCount > 0) {
lines.push(` ⚠️ ${rendered.omittedCount} pinned instruction(s) exceed the budget and are NOT injected.`);
}
lines.push("");
return 1;
}
export function registerPreviewContextCommand(
pi: ExtensionAPI,
store: MemoryStore,
projectStore: ProjectStoreRef,
projectName: ProjectNameRef,
config: Pick<MemoryConfig, "memoryMode" | "memoryPolicyStyle" | "memoryPolicyCustomText"> = { memoryMode: "policy-only" },
standing: StandingInstructions | null = null,
): void {
pi.registerCommand("memory-preview-context", {
description: "Preview the memory policy or legacy memory context blocks",
handler: async (_args, ctx) => {
if (config.memoryMode === "policy-only") {
const policyPrompt = resolveMemoryPolicyPrompt(config);
const lines: string[] = [];
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ Injected Context Preview ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" Mode: policy-only");
lines.push(` Policy style: ${config.memoryPolicyStyle ?? "full"}`);
lines.push(" This is the memory policy appended to the system prompt.");
lines.push(" Full Markdown memories are NOT injected in this mode.");
lines.push("");
let blockCount = 0;
if (policyPrompt) {
blockCount++;
lines.push(policyPrompt);
lines.push("");
} else {
lines.push(" No memory policy context is injected for this policy style.");
lines.push("");
}
blockCount += appendStandingBlock(lines, standing);
lines.push(` Blocks shown: ${blockCount}`);
ctx.ui.notify(lines.join("\n"), "info");
return;
}
const activeProjectStore = resolveProjectStore(projectStore);
const activeProjectName = resolveProjectName(projectName);
const memoryBlock = store.formatForSystemPrompt();
const projectBlock = activeProjectStore ? activeProjectStore.formatProjectBlock(activeProjectName ?? "") : "";
const lines: string[] = [];
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 👀 Injected Context Preview ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" This is the memory context appended to the system prompt.");
lines.push(" (Core hidden system instructions are NOT shown.)");
lines.push("");
let blockCount = 0;
if (memoryBlock) {
blockCount++;
lines.push(" ── MEMORY + USER + RECENT FAILURES ─────────────────────────");
lines.push(memoryBlock);
lines.push("");
}
if (projectBlock) {
blockCount++;
lines.push(` ── PROJECT MEMORY (${activeProjectName ?? ""}) ─────────────────────────`);
lines.push(projectBlock);
lines.push("");
}
blockCount += appendStandingBlock(lines, standing);
if (blockCount === 0) {
lines.push(" No memory context blocks are currently injected.");
lines.push(" Add memory entries, then run this command again.");
lines.push("");
}
lines.push(` Blocks shown: ${blockCount}`);
ctx.ui.notify(lines.join("\n"), "info");
},
});
}
@@ -0,0 +1,965 @@
/**
* Parse and apply structured memory operations from direct background review.
*/
import type { Api, Model } from "@earendil-works/pi-ai";
import { completeSimple, type Message, type SimpleStreamOptions } from "@earendil-works/pi-ai/compat";
import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import type { DatabaseManager } from "../store/db.js";
import type { MemoryCategory, MemoryConfig, MemoryResult, ThinkingLevel } from "../types.js";
export interface ReviewMemoryOperation {
action: "add" | "replace" | "remove";
target: "memory" | "user" | "project" | "failure";
content?: string;
old_text?: string;
category?: MemoryCategory;
failure_reason?: string;
}
export interface ApplyReviewOperationsResult {
appliedCount: number;
skippedCount: number;
error?: string;
aborted?: boolean;
}
export interface DirectReviewResult {
ok: boolean;
appliedCount: number;
fallbackReason?: "no_model" | "no_auth" | "aborted" | "parse_error" | "provider_error" | "empty" | "empty_response";
error?: string;
}
export interface RunDirectMemoryCompletionOptions {
userPrompt: string;
systemPrompt: string;
config: Pick<MemoryConfig, "llmModelOverride" | "llmFallbackModels" | "llmThinkingOverride">;
timeoutMs?: number;
signal?: AbortSignal;
requireAtomicShrink?: boolean;
expectedTarget?: ReviewMemoryOperation["target"];
}
/** Shared transport gate: review/flush/consolidation/correction all default to
* the in-process direct completion path and fall back to a `pi -p` subprocess
* only on failure, unless the user forces `reviewTransport: "subprocess"`. */
export function usesDirectTransport(config: Pick<MemoryConfig, "reviewTransport">): boolean {
return (config.reviewTransport ?? "direct") === "direct";
}
/** One shared budget for a single review/flush/correction/consolidation
* completion, on both the direct transport and the `pi -p` subprocess
* fallback. Deliberately not configurable (#197): the bug was the double
* spend, not the number — after the empty-response short-circuit the
* painful case is one 120s call. Consolidation keeps its separate
* `consolidationTimeoutMs` because it is user-visible and must actually
* shrink. */
export const REVIEW_COMPLETION_TIMEOUT_MS = 120_000;
type ReviewLlmConfig = Pick<MemoryConfig, "llmModelOverride" | "llmFallbackModels" | "llmThinkingOverride">;
function findExactModelReferenceMatch(modelReference: string, availableModels: Model<Api>[]): Model<Api> | undefined {
const trimmedReference = modelReference.trim();
if (!trimmedReference) return undefined;
const normalizedReference = trimmedReference.toLowerCase();
const canonicalMatches = availableModels.filter(
(model) => `${model.provider}/${model.id}`.toLowerCase() === normalizedReference,
);
if (canonicalMatches.length === 1) return canonicalMatches[0];
if (canonicalMatches.length > 1) return undefined;
const slashIndex = trimmedReference.indexOf("/");
if (slashIndex !== -1) {
const provider = trimmedReference.substring(0, slashIndex).trim();
const modelId = trimmedReference.substring(slashIndex + 1).trim();
if (provider && modelId) {
const providerMatches = availableModels.filter(
(model) => model.provider.toLowerCase() === provider.toLowerCase()
&& model.id.toLowerCase() === modelId.toLowerCase(),
);
if (providerMatches.length === 1) return providerMatches[0];
}
}
const idMatches = availableModels.filter((model) => model.id.toLowerCase() === normalizedReference);
return idMatches.length === 1 ? idMatches[0] : undefined;
}
function normalizedModelOverride(config: ReviewLlmConfig): string | undefined {
const trimmed = config.llmModelOverride?.trim();
return trimmed ? trimmed : undefined;
}
function collectFallbackOverrides(config: ReviewLlmConfig): string[] {
const out: string[] = [];
for (const raw of config.llmFallbackModels ?? []) {
const t = raw.trim();
if (t) out.push(t);
}
return out;
}
function allModelOverrides(config: ReviewLlmConfig): string[] {
const primary = normalizedModelOverride(config);
const fallbacks = collectFallbackOverrides(config);
const chain = primary ? [primary, ...fallbacks] : fallbacks;
return [...new Set(chain)];
}
function effectiveThinkingOverride(config: ReviewLlmConfig): ThinkingLevel | undefined {
return config.llmThinkingOverride ?? (normalizedModelOverride(config) ? "off" : undefined);
}
type ReviewModelRegistry = ExtensionContext["modelRegistry"];
/** Derived from the installed SDK so headers/baseUrl track ProviderHeaders instead of a local mirror. */
export type ResolvedRequestAuth = Awaited<ReturnType<ReviewModelRegistry["getApiKeyAndHeaders"]>>;
type DirectReviewAuth = Omit<Extract<ResolvedRequestAuth, { ok: true }>, "ok">;
type DirectRequestHeaders = DirectReviewAuth["headers"];
/** Context the direct transport needs: request routing plus the session the
* gateway attributes the background request to (#250). */
export type DirectReviewContext = Pick<ExtensionContext, "model" | "modelRegistry" | "sessionManager">;
// Restated from pi-coding-agent's internal getSessionHeaders, read at
// dist/core/provider-attribution.js in v0.87.1. The package exports no seam for
// it, and pi-ai only grew its own injection in v0.86.0 while this extension
// declares a 0.80.6 floor. Re-read that file when bumping the floor past
// 0.86.0, and check that pi did not add a provider or change the client value.
//
// The header is set directly rather than through StreamOptions.sessionId
// because pi-ai's openai-responses default of sendSessionIdHeader is true, so
// passing sessionId would also start sending an OpenAI session_id header to
// every OpenAI Responses user.
const OPENCODE_SESSION_HEADER = "x-opencode-session";
const OPENCODE_CLIENT_HEADER = "x-opencode-client";
const OPENCODE_HOST = "opencode.ai";
function matchesHost(baseUrl: string | undefined, expectedHost: string): boolean {
try {
return new URL(baseUrl ?? "").hostname === expectedHost;
} catch {
return false;
}
}
/** A missing session id degrades to the pre-#250 behaviour: the attempt still
* goes out, and the subprocess fallback owns recovery. */
function readSessionId(sessionManager: DirectReviewContext["sessionManager"]): string | undefined {
try {
return sessionManager.getSessionId() || undefined;
} catch {
return undefined;
}
}
/** Returns a copy whenever a header is added: the registry hands out a shared
* object, and the credential-rotation comparison below must not read hermes'
* own header as a credential change.
*
* An operator-configured session header short-circuits both additions, so that
* request carries no client header. pi keeps its client header in that case.
* The divergence is deliberate, because honouring the operator's session id
* completely is the more predictable contract. */
function directRequestHeaders(
model: Model<Api>,
headers: DirectRequestHeaders,
sessionId: string | undefined,
): DirectRequestHeaders {
if (!sessionId) return headers;
const isOpenCode = model.provider === "opencode"
|| model.provider === "opencode-go"
|| matchesHost(model.baseUrl, OPENCODE_HOST);
if (!isOpenCode) return headers;
const alreadySet = Object.keys(headers ?? {}).some(
(key) => key.toLowerCase() === OPENCODE_SESSION_HEADER,
);
if (alreadySet) return headers;
return {
...headers,
[OPENCODE_SESSION_HEADER]: sessionId,
[OPENCODE_CLIENT_HEADER]: "pi",
};
}
export function buildDirectReviewCompletionOptions(
model: Model<Api>,
auth: DirectReviewAuth,
thinking: ThinkingLevel | undefined,
signal: AbortSignal,
sessionId: string | undefined,
): SimpleStreamOptions {
const options: SimpleStreamOptions = {
apiKey: auth.apiKey,
headers: directRequestHeaders(model, auth.headers, sessionId),
env: auth.env,
signal,
};
if (model.reasoning && thinking && thinking !== "off") {
options.reasoning = thinking;
}
return options;
}
export function resolveReviewModel(
ctxModel: Model<Api> | undefined,
modelRegistry: ReviewModelRegistry,
config: ReviewLlmConfig,
): Model<Api> | undefined {
const override = normalizedModelOverride(config);
if (override) {
const matched = findExactModelReferenceMatch(override, modelRegistry.getAll());
if (matched) return matched;
}
return ctxModel;
}
export function resolveReviewModels(
ctxModel: Model<Api> | undefined,
modelRegistry: ReviewModelRegistry,
config: ReviewLlmConfig,
): Model<Api>[] {
const chain = allModelOverrides(config);
if (chain.length === 0) return ctxModel ? [ctxModel] : [];
const all = modelRegistry.getAll();
const resolved: Model<Api>[] = [];
for (const ref of chain) {
const m = findExactModelReferenceMatch(ref, all);
if (m) resolved.push(m);
}
// If none of the chain resolved, fall back to active model so we still try something
if (resolved.length === 0 && ctxModel) resolved.push(ctxModel);
return resolved;
}
export function getReviewModelChain(config: ReviewLlmConfig): string[] {
return allModelOverrides(config);
}
/**
* Provider responses that mean "this key is no longer good", as opposed to a
* transport hiccup or a model error worth falling back to a subprocess for.
*/
const AUTH_REJECTION_PATTERN = new RegExp([
String.raw`\b(401|403)\b`,
"unauthorized",
"forbidden",
String.raw`invalid[\s_-]*api[\s_-]*key`,
String.raw`authentication[\s_-]*(failed|error)`,
String.raw`(invalid|expired|revoked)[\s_-]*(access[\s_-]*)?(token|key|credential)`,
String.raw`(token|key|credential)[\s_-]*(is[\s_-]*|has[\s_-]*been[\s_-]*)?(invalid|expired|revoked)`,
].join("|"), "i");
export function isAuthRejection(message: string): boolean {
return AUTH_REJECTION_PATTERN.test(message);
}
const CREDENTIAL_HEADER_NAMES = new Set(["authorization", "x-api-key", "cf-aig-authorization"]);
function isNonEmptyString(value: unknown): value is string {
return typeof value === "string" && value.trim().length > 0;
}
// Bedrock (AWS profile/SSO) signs inside its SDK, so empty auth is normal there.
// Not for OAuth: empty OAuth auth means the token refresh failed.
function hasRequestAuth(
modelRegistry: ReviewModelRegistry,
model: Model<Api>,
auth: DirectReviewAuth,
): boolean {
if (isNonEmptyString(auth.apiKey)) return true;
const hasCredentialHeader = Object.entries(auth.headers ?? {}).some(
([key, value]) => CREDENTIAL_HEADER_NAMES.has(key.toLowerCase()) && isNonEmptyString(value),
);
if (hasCredentialHeader) return true;
return modelRegistry.hasConfiguredAuth(model) && !modelRegistry.isUsingOAuth(model);
}
function sameStringRecord(
left: Record<string, string> | undefined,
right: Record<string, string> | undefined,
): boolean {
const leftEntries = Object.entries(left ?? {});
const rightKeys = Object.keys(right ?? {});
return leftEntries.length === rightKeys.length
&& leftEntries.every(([key, value]) => right?.[key] === value);
}
function headerPairs(headers: DirectReviewAuth["headers"]): Array<[string, string | null]> {
return Object.entries(headers ?? {}).map(([key, value]) => [key.toLowerCase(), value]);
}
function sameHeaders(
left: DirectReviewAuth["headers"],
right: DirectReviewAuth["headers"],
): boolean {
const remaining = headerPairs(right);
const leftPairs = headerPairs(left);
if (leftPairs.length !== remaining.length) return false;
for (const [key, value] of leftPairs) {
const index = remaining.findIndex((pair) => pair[0] === key && pair[1] === value);
if (index === -1) return false;
remaining.splice(index, 1);
}
return true;
}
function sameRequestAuth(left: DirectReviewAuth, right: DirectReviewAuth): boolean {
return left.apiKey === right.apiKey
&& sameHeaders(left.headers, right.headers)
&& sameStringRecord(left.env, right.env);
}
/**
* Resolve request auth through the public ModelRegistry API. Resolve it again
* after an auth rejection so Pi can supply refreshed credentials when its
* registry supports that, without reaching into version-sensitive internals.
*/
export async function resolveRequestAuth(
modelRegistry: ReviewModelRegistry,
model: Model<Api>,
): Promise<ResolvedRequestAuth> {
return modelRegistry.getApiKeyAndHeaders(model);
}
/** A JSON object extracted from a model response, validated to carry an
* `operations` array. Candidate payloads without one are declined at every
* extraction step, so a log line, counterexample, or stray snippet cannot
* claim the parse and force the subprocess fallback (#235). */
type JsonObjectPayload = { operations?: unknown };
function isJsonObjectPayload(value: unknown): value is JsonObjectPayload {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
/** Balanced {...} regions at any nesting depth, string- and escape-aware.
* Recording every matched pair — not just top-level ones — means an
* unbalanced "{" in surrounding prose cannot hide a later object: the
* object still forms its own balanced region nested inside the phantom
* one (#235). */
function balancedObjectSpans(text: string): Array<[number, number]> {
const spans: Array<[number, number]> = [];
const open: number[] = [];
let inString = false;
let escaped = false;
for (let i = 0; i < text.length; i++) {
const ch = text[i];
if (escaped) {
escaped = false;
continue;
}
if (inString) {
if (ch === "\\") escaped = true;
else if (ch === '"') inString = false;
continue;
}
if (ch === '"') {
inString = true;
} else if (ch === "{") {
open.push(i);
} else if (ch === "}") {
const start = open.pop();
if (start !== undefined) spans.push([start, i]);
}
}
return spans;
}
/** CoT routinely restates the schema before answering, so the first-to-last
* slice above is invalid JSON across that span. Scan balanced objects from
* the end and return the last one that parses with an `operations` array —
* the answer trailing the preamble (#197). */
function lastParseableOperationsObject(text: string): JsonObjectPayload | null {
const spans = balancedObjectSpans(text);
for (let s = spans.length - 1; s >= 0; s--) {
const span = spans[s];
if (!span) continue;
try {
const parsed: unknown = JSON.parse(text.slice(span[0], span[1] + 1));
if (isJsonObjectPayload(parsed) && Array.isArray(parsed.operations)) {
return parsed;
}
} catch {
// keep scanning
}
}
return null;
}
function extractJsonPayload(text: string): JsonObjectPayload | null {
const trimmed = text.trim();
if (!trimmed) return null;
const asObject = (value: unknown): JsonObjectPayload | null =>
isJsonObjectPayload(value) && Array.isArray(value.operations) ? value : null;
// A declined candidate (valid JSON, no `operations` array) must fall through
// to the later paths rather than exit — exiting here is how a stray snippet
// claimed the parse and forced the subprocess while a valid trailing
// answer was present (#235).
try {
const parsed = asObject(JSON.parse(trimmed));
if (parsed) return parsed;
} catch {
// continue
}
const fenced = trimmed.match(/```(?:json)?\s*([\s\S]*?)```/i);
if (fenced?.[1]) {
try {
const parsed = asObject(JSON.parse(fenced[1].trim()));
if (parsed) return parsed;
} catch {
// continue
}
}
const start = trimmed.indexOf("{");
const end = trimmed.lastIndexOf("}");
if (start >= 0 && end > start) {
try {
const parsed = asObject(JSON.parse(trimmed.slice(start, end + 1)));
if (parsed) return parsed;
} catch {
return lastParseableOperationsObject(trimmed);
}
}
return lastParseableOperationsObject(trimmed);
}
/** A winning thinking-channel candidate plus whether it sits at the tail of
* the channel — the trust boundary for thinking-sourced operations. */
interface ThinkingOperationsCandidate {
payload: JsonObjectPayload;
trailing: boolean;
}
/** Thinking-sourced text takes its own extraction (#235): candidates are
* validated on their `operations` array and scanned from the end, because a
* chain of thought routinely restates the schema — or drafts an operation it
* then rejects — before the final answer. The shared cascade would let an
* earlier fenced draft win before this scan runs. The winner is the last
* ops-bearing object; `trailing` reports whether anything follows it, which
* decides how much to trust it. */
function extractThinkingOperations(text: string): ThinkingOperationsCandidate | null {
const spans = balancedObjectSpans(text);
for (let s = spans.length - 1; s >= 0; s--) {
const span = spans[s];
if (!span) continue;
try {
const parsed: unknown = JSON.parse(text.slice(span[0], span[1] + 1));
if (isJsonObjectPayload(parsed) && Array.isArray(parsed.operations)) {
return { payload: parsed, trailing: text.slice(span[1] + 1).trim() === "" };
}
} catch {
// keep scanning
}
}
return null;
}
function isMemoryCategory(value: unknown): value is MemoryCategory {
return value === "failure"
|| value === "correction"
|| value === "insight"
|| value === "preference"
|| value === "convention"
|| value === "tool-quirk";
}
function isReviewTarget(value: unknown): value is ReviewMemoryOperation["target"] {
return value === "memory" || value === "user" || value === "project" || value === "failure";
}
function isReviewAction(value: unknown): value is ReviewMemoryOperation["action"] {
return value === "add" || value === "replace" || value === "remove";
}
export function parseReviewOperations(text: string): ReviewMemoryOperation[] | null {
if (/nothing to save/i.test(text) && !text.includes("{")) {
return [];
}
const payload = extractJsonPayload(text);
if (!payload) {
return null;
}
return mapOperationsArray(payload.operations);
}
/** Map a candidate `operations` array to validated review operations; null
* means the payload is not an operations payload at all. */
function mapOperationsArray(operations: unknown): ReviewMemoryOperation[] | null {
if (!Array.isArray(operations)) return null;
const parsed: ReviewMemoryOperation[] = [];
for (const item of operations) {
if (!item || typeof item !== "object") continue;
const op = item as Record<string, unknown>;
if (!isReviewAction(op.action) || !isReviewTarget(op.target)) continue;
const operation: ReviewMemoryOperation = {
action: op.action,
target: op.target,
};
if (typeof op.content === "string") operation.content = op.content;
if (typeof op.old_text === "string") operation.old_text = op.old_text;
if (isMemoryCategory(op.category)) operation.category = op.category;
if (typeof op.failure_reason === "string") operation.failure_reason = op.failure_reason;
parsed.push(operation);
}
return parsed;
}
export async function applyReviewOperations(
store: MemoryStore,
projectStore: MemoryStore | null,
operations: ReviewMemoryOperation[],
_dbManager: DatabaseManager | null = null,
projectName?: string | null,
options: {
requireAtomicShrink?: boolean;
expectedTarget?: ReviewMemoryOperation["target"];
signal?: AbortSignal;
} = {},
): Promise<ApplyReviewOperationsResult> {
if (options.requireAtomicShrink) {
if (operations.length === 0) {
return {
appliedCount: 0,
skippedCount: 0,
error: "Atomic plan requires at least one operation.",
};
}
const target = operations[0]?.target;
if (!target || operations.some((operation) => operation.target !== target)) {
return {
appliedCount: 0,
skippedCount: operations.length,
error: "Atomic plan must use exactly one target.",
};
}
if (options.expectedTarget && target !== options.expectedTarget) {
return {
appliedCount: 0,
skippedCount: operations.length,
error: `Atomic plan targeted '${target}', expected '${options.expectedTarget}'.`,
};
}
if (target === "project" && !projectStore) {
return {
appliedCount: 0,
skippedCount: operations.length,
error: "Project memory is unavailable.",
};
}
if (options.signal?.aborted) {
return { appliedCount: 0, skippedCount: operations.length, aborted: true };
}
const activeStore = target === "project" ? projectStore! : store;
const memoryTarget = target === "project" ? "memory" : target;
const mutationOperations = operations.map((operation) => ({
action: operation.action,
content: operation.content,
oldText: operation.old_text,
category: target === "failure" ? operation.category ?? "failure" : operation.category,
failureReason: operation.failure_reason,
project: target === "failure" ? projectName ?? undefined : undefined,
}));
const result = await activeStore.applyMutationPlan(memoryTarget, mutationOperations, {
requireShrink: true,
signal: options.signal,
});
if (options.signal?.aborted && !result.success) {
return { appliedCount: 0, skippedCount: operations.length, aborted: true };
}
return result.success
? { appliedCount: operations.length, skippedCount: 0 }
: {
appliedCount: 0,
skippedCount: operations.length,
error: result.error ?? "Atomic memory plan failed.",
};
}
let appliedCount = 0;
let skippedCount = 0;
for (let i = 0; i < operations.length; i++) {
if (options.signal?.aborted) {
skippedCount += operations.length - i;
return { appliedCount, skippedCount, aborted: appliedCount === 0 };
}
const op = operations[i];
if (op.target === "project" && !projectStore) {
skippedCount++;
continue;
}
const rawTarget = op.target;
const memoryTarget = rawTarget === "project" ? "memory" : rawTarget === "failure" ? "failure" : rawTarget;
const activeStore = rawTarget === "project" ? projectStore! : store;
let result: MemoryResult;
switch (op.action) {
case "add": {
if (!op.content?.trim()) {
skippedCount++;
continue;
}
if (rawTarget === "failure") {
const category = op.category ?? "failure";
result = await activeStore.addFailure(op.content, {
category,
failureReason: op.failure_reason,
project: projectName ?? undefined,
signal: options.signal,
});
if (result.success) {
appliedCount++;
} else {
skippedCount++;
}
} else {
result = await activeStore.add(memoryTarget, op.content, options.signal);
if (result.success) {
appliedCount++;
} else {
skippedCount++;
}
}
break;
}
case "replace": {
if (!op.old_text || !op.content?.trim()) {
skippedCount++;
continue;
}
result = await activeStore.replace(memoryTarget, op.old_text, op.content, options.signal);
if (result.success) {
appliedCount++;
} else {
skippedCount++;
}
break;
}
case "remove": {
if (!op.old_text) {
skippedCount++;
continue;
}
result = await activeStore.remove(memoryTarget, op.old_text, options.signal);
if (result.success) {
appliedCount++;
} else {
skippedCount++;
}
break;
}
default:
skippedCount++;
continue;
}
if (options.signal?.aborted) {
skippedCount += operations.length - i - 1;
return { appliedCount, skippedCount, aborted: appliedCount === 0 };
}
}
return { appliedCount, skippedCount };
}
/** Channel-aware text extraction: text blocks when present, otherwise the
* thinking channel — thinking-default servers park the whole answer in the
* reasoning channel (e.g. vLLM with DEFAULT_THINKING=max, regardless of the
* client-side level), so it is recovered there rather than treated as a
* parse error (#197). The source decides which extraction path parses it:
* the text channel keeps the shared cascade, thinking takes its own
* end-scan (#235). */
function responseChannelText(content: unknown): { text: string; source: "text" | "thinking" } {
if (!Array.isArray(content)) return { text: "", source: "text" };
const text = content
.filter((block): block is { type: "text"; text: string } => (
!!block && typeof block === "object" && (block as { type?: string }).type === "text"
))
.map((block) => block.text)
.join("\n");
if (text.trim()) return { text, source: "text" };
// Anthropic redacted_thinking arrives as {type:"thinking", redacted:true}
// after normalization — there is no recoverable payload in it, and letting
// it through would turn a redacted-only completion into a parse_error and
// burn the subprocess fallback (#197).
const thinking = content
.filter((block): block is { type: "thinking"; thinking: string } => (
!!block && typeof block === "object" && (block as { type?: string }).type === "thinking"
&& (block as { redacted?: boolean }).redacted !== true
))
.map((block) => block.thinking)
.join("\n");
return { text: thinking, source: "thinking" };
}
const defaultProviderNoticeState = { logged: false };
/** Fired once per process when a provider returns the whole answer in its
* thinking channel with an empty answer channel. That signature means the
* provider's server-side thinking default is swallowing the output (the #197
* report), so thinking-channel recovery is a workaround for a
* misconfiguration, not a feature to rely on. */
function notifyThinkingChannelProvider(
model: { provider?: unknown; id?: unknown },
deps: { onProviderNotice?: (message: string) => void; providerNoticeState?: { logged: boolean } },
): void {
const state = deps.providerNoticeState ?? defaultProviderNoticeState;
if (state.logged) return;
state.logged = true;
const modelRef = typeof model?.provider === "string" && typeof model?.id === "string"
? `${model.provider}/${model.id}`
: "the active model";
const sink = deps.onProviderNotice ?? ((message: string) => console.warn(message));
sink(
`[pi-hermes-memory] Provider misconfiguration: ${modelRef} returned the whole answer in its thinking channel and left the answer channel empty. `
+ "This usually means server-side thinking is forced on for this provider (for example a vLLM server started with DEFAULT_THINKING=max). "
+ "Review recovery from the thinking channel is best-effort and can miss saves. "
+ "Ask your Pi to fix this as well.",
);
}
export async function runDirectMemoryCompletion(
ctx: DirectReviewContext,
store: MemoryStore,
projectStore: MemoryStore | null,
options: RunDirectMemoryCompletionOptions,
dbManager: DatabaseManager | null = null,
projectName?: string | null,
deps: {
completeSimple?: typeof completeSimple;
/** Test seam: replaces the console.warn sink for the provider notice. */
onProviderNotice?: (message: string) => void;
/** Test seam: dedupe state for the provider notice; omit for the process default. */
providerNoticeState?: { logged: boolean };
} = {},
): Promise<DirectReviewResult> {
const complete = deps.completeSimple ?? completeSimple;
const aborted = (): DirectReviewResult => ({ ok: false, appliedCount: 0, fallbackReason: "aborted" });
if (options.signal?.aborted) return aborted();
const models = resolveReviewModels(ctx.model, ctx.modelRegistry, options.config);
if (models.length === 0) {
return { ok: false, appliedCount: 0, fallbackReason: "no_model" };
}
// One id per completion, reused across the chain: the whole chain is a single
// job for a single conversation.
const sessionId = readSessionId(ctx.sessionManager);
// Try each model in chain: primary + llmFallbackModels. Only retry on
// provider/auth/transport failures; parse errors are prompt-specific so we
// still try the next model as it may have better instruction following.
let lastResult: DirectReviewResult | undefined;
for (let mi = 0; mi < models.length; mi++) {
const model = models[mi]!;
// A caller abort (user cancel, shutdown bound) ends the whole chain: the
// next iteration's listener would never fire for an already-aborted
// signal, so continuing would burn a full timeout per remaining model.
if (options.signal?.aborted) return aborted();
const auth = await resolveRequestAuth(ctx.modelRegistry, model);
if (options.signal?.aborted) return aborted();
if (!auth.ok || !hasRequestAuth(ctx.modelRegistry, model, auth)) {
lastResult = {
ok: false,
appliedCount: 0,
fallbackReason: "no_auth",
error: auth.ok ? `No request authentication for ${model.provider}` : auth.error,
};
if (mi < models.length - 1) continue;
return lastResult;
}
let requestAuth: DirectReviewAuth = { apiKey: auth.apiKey, headers: auth.headers, env: auth.env };
const controller = new AbortController();
const timeoutMs = options.timeoutMs ?? REVIEW_COMPLETION_TIMEOUT_MS;
const timeout = setTimeout(() => controller.abort(), timeoutMs);
const onExternalAbort = () => controller.abort();
if (options.signal) {
options.signal.addEventListener("abort", onExternalAbort, { once: true });
if (options.signal.aborted) controller.abort();
}
const thinking = effectiveThinkingOverride(options.config);
const userMessage: Message = {
role: "user",
content: [{ type: "text", text: options.userPrompt }],
timestamp: Date.now(),
};
const request = { systemPrompt: options.systemPrompt, messages: [userMessage] };
const completeOnce = async () => {
const response = await complete(
model,
request,
buildDirectReviewCompletionOptions(model, requestAuth, thinking, controller.signal, sessionId),
);
if (response.stopReason === "error" && isAuthRejection(response.errorMessage ?? "")) {
throw new Error(response.errorMessage ?? "error");
}
return response;
};
try {
let response;
try {
response = await completeOnce();
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
if (controller.signal.aborted || !isAuthRejection(message)) throw err;
// Thrown failures and error assistant responses share this path. API keys
// and OAuth headers can both rotate, so re-resolve through Pi and retry
// once only when the effective request auth actually changed; otherwise
// this is a real auth problem and the subprocess fallback should handle
// it (#139).
const rotated = await resolveRequestAuth(ctx.modelRegistry, model);
if (!rotated.ok || !hasRequestAuth(ctx.modelRegistry, model, rotated) || sameRequestAuth(rotated, requestAuth)) throw err;
requestAuth = { apiKey: rotated.apiKey, headers: rotated.headers, env: rotated.env };
response = await completeOnce();
}
if (options.signal?.aborted) {
clearTimeout(timeout);
return aborted();
}
if (response.stopReason === "aborted" || controller.signal.aborted) {
lastResult = { ok: false, appliedCount: 0, fallbackReason: "aborted" };
if (mi < models.length - 1) { clearTimeout(timeout); continue; }
return lastResult;
}
const { text, source } = responseChannelText(response.content);
// The thinking-only signature means the provider's server-side thinking
// default is swallowing the answer channel (#197). Recovery works, but
// it is best-effort; surface the misconfiguration once per process so
// the user can fix the provider instead of relying on the recovery.
if (source === "thinking" && text.trim() !== "") {
notifyThinkingChannelProvider(model, deps);
}
// Thinking-sourced text takes its own extraction (#235): candidates are
// validated on their `operations` array and scanned from the end, and
// the winner's trailing-ness sets how much to trust it. The text channel
// keeps the shared cascade untouched so today's parse successes do not
// move.
let operations: ReviewMemoryOperation[] | null;
let unparsableThinking = false;
if (source === "thinking") {
const extraction = extractThinkingOperations(text);
if (extraction) {
operations = mapOperationsArray(extraction.payload.operations) ?? [];
// A candidate that is not trailing is draft-grade: apply only its
// adds. Trusting a draft's replace/remove — or reaching back to an
// earlier candidate when the trailing one is empty — would execute
// operations the model may have rejected; a missed save is
// recoverable, a wrong deletion is not.
if (!extraction.trailing) {
operations = operations.filter((operation) => operation.action === "add");
}
} else {
operations = null;
unparsableThinking = true;
}
} else {
operations = parseReviewOperations(text);
}
// A clean stop with no usable output — nothing in either channel, or
// thinking-sourced output that parses to nothing — settles
// empty_response: the subprocess would run the same model against the
// same server-side thinking default and fail the same way (#197). The
// chain still walks llmFallbackModels first, like parse_error does: a
// silent primary model should not end the review while a configured
// fallback may answer (#235). Never the subprocess on this path.
// Truncated responses (stopReason "length") still fall through to
// parse_error so the chain can retry them.
if ((!text.trim() || unparsableThinking) && response.stopReason === "stop") {
lastResult = { ok: true, appliedCount: 0, fallbackReason: "empty_response" };
if (mi < models.length - 1) { clearTimeout(timeout); continue; }
return lastResult;
}
if (operations === null) {
lastResult = { ok: false, appliedCount: 0, fallbackReason: "parse_error" };
if (mi < models.length - 1) { clearTimeout(timeout); continue; }
return lastResult;
}
if (operations.length === 0) {
clearTimeout(timeout);
return { ok: true, appliedCount: 0, fallbackReason: "empty" };
}
if (controller.signal.aborted || options.signal?.aborted) {
return aborted();
}
const applied = await applyReviewOperations(
store,
projectStore,
operations,
dbManager,
projectName,
{
requireAtomicShrink: options.requireAtomicShrink,
expectedTarget: options.expectedTarget,
signal: options.signal,
},
);
if (applied.aborted && applied.appliedCount === 0) {
return aborted();
}
if (applied.error) {
lastResult = {
ok: false,
appliedCount: 0,
fallbackReason: "provider_error",
error: applied.error,
};
if (mi < models.length - 1) { clearTimeout(timeout); continue; }
return lastResult;
}
clearTimeout(timeout);
return { ok: true, appliedCount: applied.appliedCount };
} catch (err) {
if (options.signal?.aborted) {
clearTimeout(timeout);
return aborted();
}
if (controller.signal.aborted) {
lastResult = { ok: false, appliedCount: 0, fallbackReason: "aborted" };
} else {
lastResult = {
ok: false,
appliedCount: 0,
fallbackReason: "provider_error",
error: err instanceof Error ? err.message : String(err),
};
}
if (mi < models.length - 1) { clearTimeout(timeout); continue; }
return lastResult!;
} finally {
clearTimeout(timeout);
options.signal?.removeEventListener("abort", onExternalAbort);
}
}
return lastResult ?? { ok: false, appliedCount: 0, fallbackReason: "provider_error", error: "All fallback models failed" };
}
@@ -0,0 +1,160 @@
import { measureLifecycleSync } from '../lifecycle-timing.js';
import type { DatabaseManager } from '../store/db.js';
import {
indexChangedSessions,
needsBackfill,
touchBackfillTimestamp,
type BulkIndexResult,
} from '../store/session-indexer.js';
export const SESSION_BACKFILL_SHUTDOWN_TIMEOUT_MS = 5000;
export const SESSION_BACKFILL_MAX_FILES = 50;
type NotifyLevel = 'info' | 'warning' | 'error';
type NotifyFn = (message: string, level: NotifyLevel) => void;
type SetTimeoutFn = (callback: () => void, ms: number) => unknown;
export interface SessionBackfillState {
inProgress: boolean;
promise: Promise<void> | null;
}
export const sessionBackfillState: SessionBackfillState = {
inProgress: false,
promise: null,
};
export interface ScheduleSessionBackfillOptions {
notify?: NotifyFn;
state?: SessionBackfillState;
setTimeoutFn?: SetTimeoutFn;
needsBackfillFn?: typeof needsBackfill;
indexSessionsFn?: typeof indexChangedSessions;
maxFilesToIndex?: number;
touchBackfillTimestampFn?: typeof touchBackfillTimestamp;
/**
* Optional retention cutoff (ms epoch). Files older than this are excluded
* from the backfill file set so that sessions pruned by retention are not
* re-indexed and do not repeatedly schedule a startup backfill.
*/
retentionCutoffMs?: number;
}
function formatBackfillResult(result: BulkIndexResult): string {
const errorSuffix = result.errors.length > 0 ? ` (${result.errors.length} file error${result.errors.length === 1 ? '' : 's'})` : '';
const limitSuffix = result.reachedLimit ? ' (startup limit reached)' : '';
return `🧠 Session backfill complete: ${result.sessionsIndexed} indexed, ${result.sessionsSkipped} skipped, ${result.messagesIndexed} messages${errorSuffix}${limitSuffix}.`;
}
function notifyBestEffort(notify: NotifyFn | undefined, message: string, level: NotifyLevel): void {
try {
notify?.(message, level);
} catch {
// Notification failures must never affect backfill.
}
}
/**
* Schedule a best-effort, bounded incremental backfill of unindexed Pi sessions.
*
* The JSONL parsing work is deferred with setTimeout(0) so session_start can
* resolve first. The scheduled pass only parses files without matching stored
* metadata and caps the number of files parsed per startup.
*
* @returns true when a backfill task was scheduled; false when it was skipped.
*/
export function scheduleSessionBackfill(
dbManager: DatabaseManager,
sessionsDir: string,
options: ScheduleSessionBackfillOptions = {},
): boolean {
const state = options.state ?? sessionBackfillState;
const setTimeoutFn = options.setTimeoutFn ?? setTimeout;
const needsBackfillFn = options.needsBackfillFn ?? needsBackfill;
const indexSessionsFn = options.indexSessionsFn ?? indexChangedSessions;
const maxFilesToIndex = options.maxFilesToIndex ?? SESSION_BACKFILL_MAX_FILES;
const touchBackfillTimestampFn = options.touchBackfillTimestampFn ?? touchBackfillTimestamp;
const retentionCutoffMs = options.retentionCutoffMs ?? 0;
if (state.inProgress) {
return false;
}
try {
if (!measureLifecycleSync(
'session-backfill.check',
() => needsBackfillFn(dbManager, sessionsDir, undefined, retentionCutoffMs),
)) {
return false;
}
} catch (err) {
notifyBestEffort(
options.notify,
`⚠️ Session backfill check failed: ${err instanceof Error ? err.message : String(err)}`,
'warning',
);
return false;
}
state.inProgress = true;
state.promise = new Promise<void>((resolve) => {
setTimeoutFn(() => {
measureLifecycleSync('session-backfill.callback', () => {
try {
const result = indexSessionsFn(dbManager, sessionsDir, { maxFilesToIndex, retentionCutoffMs });
if (!result.reachedLimit) touchBackfillTimestampFn(dbManager);
notifyBestEffort(options.notify, formatBackfillResult(result), result.errors.length > 0 || result.reachedLimit ? 'warning' : 'info');
} catch (err) {
notifyBestEffort(
options.notify,
`⚠️ Session backfill failed: ${err instanceof Error ? err.message : String(err)}`,
'warning',
);
} finally {
state.inProgress = false;
state.promise = null;
resolve();
}
});
}, 0);
});
return true;
}
/**
* First-use readiness has no shutdown deadline: the scheduled pass must settle
* before a search can read its index. Each extension owns its own state.
*/
export async function joinSessionBackfill(state: SessionBackfillState = sessionBackfillState): Promise<void> {
await state.promise;
}
/**
* Wait briefly for an in-progress backfill before shutdown closes SQLite.
*
* @returns true if no backfill was running or it completed before the timeout;
* false if the timeout elapsed first.
*/
export async function waitForSessionBackfill(
timeoutMs = SESSION_BACKFILL_SHUTDOWN_TIMEOUT_MS,
state: SessionBackfillState = sessionBackfillState,
): Promise<boolean> {
const promise = state.promise;
if (!state.inProgress || !promise) {
return true;
}
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
promise.then(() => true),
new Promise<boolean>((resolve) => {
timeout = setTimeout(() => resolve(false), timeoutMs);
}),
]);
} finally {
if (timeout) clearTimeout(timeout);
}
}
@@ -0,0 +1,290 @@
/**
* Session flush — gives the agent one turn to save memories before context is lost.
* Ported from hermes-agent/run_agent.py (flush_memories).
* See PLAN.md → "Hermes Source File Reference Map" for source lines.
*
* Default transport: in-process direct completion (same mechanism as
* background review — see review-memory-ops.ts). Falls back to a `pi -p`
* subprocess only if direct mode fails or reviewTransport forces subprocess.
*
* Compact flush spends at most `flushCompactTimeoutMs` across both transports
* (one shared window). Direct runs first; the `pi -p` fallback gets only the
* remainder, never a second copy of the budget. Shutdown stays a hardcoded
* 10s cap and is silent. `pi.exec` resolves (never rejects) on timeout/kill,
* so a non-zero child exit notifies the same way an exhaust does.
*
* Access pattern (remaining × session-signal × notify):
* | Event | leftover | session signal | notify |
* | Direct ok | n/a (return) | live | none |
* | Direct `no_model` / `parse_error` with time | subprocess leftover, not budget | live | none unless child fails/exits |
* | Direct internal timeout at ceiling | leftover < floor → skip child | live | warning, once |
* | Esc during compact (`event.signal`) | skip even if leftover ≫ 0 | aborted | silent |
* | `reviewTransport: "subprocess"` | one child, leftover = budget | as above | warn on nonzero exit / throw |
* | Shutdown | budget 10000, no session signal | n/a | never |
*/
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
import { MemoryStore } from "../store/memory-store.js";
import { DatabaseManager } from "../store/db.js";
import {
buildMemoryTargetRoutingGuidance,
DEFAULT_FLUSH_COMPACT_TIMEOUT_MS,
DEFAULT_FLUSH_SHUTDOWN_TIMEOUT_MS,
DIRECT_FLUSH_SYSTEM_PROMPT,
ENTRY_DELIMITER,
FLUSH_PROMPT,
} from "../constants.js";
import type { MemoryConfig } from "../types.js";
import type { EnsureMemoryReady } from "../memory-initialization.js";
import { measureLifecycle } from "../lifecycle-timing.js";
import { collectMessageParts } from "./message-parts.js";
import { execChildPrompt, resolveChildPiModel } from "./pi-child-process.js";
import { runDirectMemoryCompletion, usesDirectTransport } from "./review-memory-ops.js";
import { resolveProjectName, resolveProjectStore, type ProjectNameRef, type ProjectStoreRef } from "../project-context.js";
/** Do not spawn pi -p when leftover time is below this, floored by the budget itself. */
const FLUSH_SUBPROCESS_MIN_REMAINING_MS = 5_000;
export type FlushKind = "compact" | "shutdown";
export type FlushHandoff =
| { skip: "session_aborted" }
| { skip: "budget_exhausted" }
| { timeoutMs: number };
type FlushContext = Pick<ExtensionContext, "sessionManager" | "model" | "modelRegistry" | "cwd"> & {
ui?: Pick<ExtensionContext["ui"], "notify">;
};
/** Remaining subprocess budget. null = skip (spent, below spawn floor, or non-positive). */
export function remainingFlushTimeoutMs(budgetMs: number, elapsedMs: number): number | null {
if (budgetMs <= 0) return null;
const elapsed = Math.max(0, elapsedMs);
const remaining = budgetMs - elapsed;
const floor = Math.min(FLUSH_SUBPROCESS_MIN_REMAINING_MS, budgetMs);
if (remaining < floor) return null;
return remaining;
}
export function resolveFlushHandoff(
sessionAborted: boolean,
budgetMs: number,
elapsedMs: number,
): FlushHandoff {
if (sessionAborted) return { skip: "session_aborted" };
const leftover = remainingFlushTimeoutMs(budgetMs, elapsedMs);
if (leftover === null) return { skip: "budget_exhausted" };
return { timeoutMs: leftover };
}
function linkBudget(parent: AbortSignal | undefined, timeoutMs: number): { signal: AbortSignal; dispose: () => void } {
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), timeoutMs);
const onParent = () => controller.abort();
parent?.addEventListener("abort", onParent, { once: true });
if (parent?.aborted) controller.abort();
return {
signal: controller.signal,
dispose: () => {
clearTimeout(timer);
parent?.removeEventListener("abort", onParent);
},
};
}
function notifyCompactFailure(
ctx: FlushContext,
kind: FlushKind,
detail: string,
): void {
if (kind !== "compact") return;
try {
ctx.ui?.notify(
`Memory flush before compact did not save (${detail}). Compaction continues. Raise flushCompactTimeoutMs for slow/local models.`,
"warning",
);
} catch {
// compact ctx can go stale; never throw into session_before_compact
}
}
function buildDirectFlushUserPrompt(
store: MemoryStore,
projectStore: MemoryStore | null,
parts: string[],
): string {
const sections = [
"--- Current Memory ---",
store.getMemoryEntries().join(ENTRY_DELIMITER) || "(empty)",
"",
"--- Current User Profile ---",
store.getUserEntries().join(ENTRY_DELIMITER) || "(empty)",
];
if (projectStore) {
sections.push(
"",
"--- Current Project Memory ---",
projectStore.getMemoryEntries().join(ENTRY_DELIMITER) || "(empty)",
);
}
sections.push(
"",
"--- Conversation ---",
parts.join("\n\n"),
);
return sections.join("\n");
}
export function setupSessionFlush(
pi: ExtensionAPI,
store: MemoryStore,
projectStore: ProjectStoreRef,
config: MemoryConfig,
dbManager: DatabaseManager | null = null,
projectName: ProjectNameRef = null,
deps: {
runDirectMemoryCompletion?: typeof runDirectMemoryCompletion;
now?: () => number;
ensureMemoryReady?: EnsureMemoryReady;
} = {},
): void {
let userTurnCount = 0;
const now = deps.now ?? Date.now;
const runDirect = deps.runDirectMemoryCompletion ?? runDirectMemoryCompletion;
pi.on("message_end", async (event, _ctx) => {
if (event.message.role === "user") userTurnCount++;
});
/** Shared flush logic — builds conversation snapshot and saves memories */
async function flush(
ctx: FlushContext,
signal: AbortSignal | undefined,
timeoutMs: number,
kind: FlushKind,
): Promise<void> {
try {
if (userTurnCount < config.flushMinTurns) return;
let entries;
try {
entries = ctx.sessionManager.getBranch();
} catch {
return; // Context already stale
}
const parts = collectMessageParts(entries, config.flushRecentMessages);
try {
await deps.ensureMemoryReady?.(ctx);
} catch {
return; // Do not flush against an unloaded or partially migrated store.
}
const activeProjectStore = resolveProjectStore(projectStore);
const activeProjectName = resolveProjectName(projectName);
if (signal?.aborted) return;
if (timeoutMs <= 0) return; // explicit disable or degenerate config: silent
const started = now();
const budget = linkBudget(signal, timeoutMs);
try {
if (usesDirectTransport(config)) {
try {
const directResult = await runDirect(
ctx,
store,
activeProjectStore,
{
systemPrompt: [
DIRECT_FLUSH_SYSTEM_PROMPT,
"",
buildMemoryTargetRoutingGuidance(activeProjectStore !== null),
].join("\n"),
userPrompt: buildDirectFlushUserPrompt(store, activeProjectStore, parts),
config,
timeoutMs,
signal: budget.signal,
},
dbManager,
activeProjectName,
);
if (directResult.ok) return;
} catch {
// Fall through with leftover, not a copied ceiling.
}
}
const handoff = resolveFlushHandoff(Boolean(signal?.aborted), timeoutMs, now() - started);
if ("skip" in handoff) {
if (handoff.skip === "budget_exhausted") {
const elapsed = Math.max(0, now() - started);
notifyCompactFailure(
ctx,
kind,
elapsed >= timeoutMs
? `timed out after ${timeoutMs}ms`
: `only ${Math.max(0, timeoutMs - elapsed)}ms left of the ${timeoutMs}ms budget, too little to spawn the fallback`,
);
}
return;
}
const flushMessage = [
FLUSH_PROMPT,
"",
buildMemoryTargetRoutingGuidance(activeProjectStore !== null),
"",
"--- Conversation ---",
parts.join("\n\n"),
].join("\n");
try {
const childResult = await execChildPrompt(pi, flushMessage, config, {
cwd: ctx.cwd,
model: resolveChildPiModel(ctx.model),
signal,
timeoutMs: handoff.timeoutMs,
});
// pi.exec resolves {code, killed} on timeout/kill instead of rejecting.
// A signal kill can arrive as code 0 with killed true, so both shapes
// are misses worth surfacing; a non-zero code is the more specific
// report when the child did exit on its own.
if (!signal?.aborted) {
if (typeof childResult?.code === "number" && childResult.code !== 0) {
notifyCompactFailure(ctx, kind, `child exited with code ${childResult.code}`);
} else if (childResult?.killed === true) {
notifyCompactFailure(ctx, kind, "child was killed before it saved");
}
}
} catch (err) {
if (!signal?.aborted) {
const detail = err instanceof Error ? err.message : String(err);
notifyCompactFailure(ctx, kind, `child error: ${detail}`);
}
}
} finally {
budget.dispose();
}
} catch {
// Best-effort flush — never throw into compaction or shutdown.
}
}
// Flush before compaction (can afford to wait)
pi.on("session_before_compact", async (event, ctx) => {
if (!config.flushOnCompact) return;
await flush(ctx, event.signal, config.flushCompactTimeoutMs ?? DEFAULT_FLUSH_COMPACT_TIMEOUT_MS, "compact");
});
// Flush before session shutdown. Pi awaits async session_shutdown handlers
// before invalidating the session, so await the bounded flush here.
pi.on("session_shutdown", async (event, ctx) => {
if (!config.flushOnShutdown || event.reason === "reload") return;
await measureLifecycle("shutdown.flush", () =>
flush(ctx, undefined, DEFAULT_FLUSH_SHUTDOWN_TIMEOUT_MS, "shutdown"),
);
});
}
@@ -0,0 +1,94 @@
import { measureLifecycleSync } from '../lifecycle-timing.js';
import type { DatabaseManager } from '../store/db.js';
import { indexLiveSession } from '../store/session-indexer.js';
export const SESSION_LIVE_INDEX_DELAY_MS = 50;
export const SESSION_LIVE_INDEX_SHUTDOWN_TIMEOUT_MS = 5000;
type SetTimeoutFn = (callback: () => void, ms: number) => unknown;
type SessionManagerSnapshot = Parameters<typeof indexLiveSession>[1];
export interface SessionLiveIndexState {
inProgress: boolean;
promise: Promise<void> | null;
}
export const sessionLiveIndexState: SessionLiveIndexState = {
inProgress: false,
promise: null,
};
export interface ScheduleLiveSessionIndexOptions {
state?: SessionLiveIndexState;
setTimeoutFn?: SetTimeoutFn;
indexLiveSessionFn?: typeof indexLiveSession;
delayMs?: number;
onError?: (error: unknown) => void;
}
/**
* Schedule non-blocking indexing of the current live session.
*
* Pi emits message_end before it appends the finalized message to the JSONL
* session file/session manager. Deferring briefly lets Pi persist the entry
* first, then we index any message ids not already present in SQLite. Multiple
* message_end events in the same window coalesce into one all-missing sync.
*/
export function scheduleLiveSessionIndex(
dbManager: DatabaseManager,
sessionManager: SessionManagerSnapshot,
options: ScheduleLiveSessionIndexOptions = {},
): boolean {
const state = options.state ?? sessionLiveIndexState;
if (state.inProgress) {
return false;
}
const setTimeoutFn = options.setTimeoutFn ?? setTimeout;
const indexLiveSessionFn = options.indexLiveSessionFn ?? indexLiveSession;
const delayMs = options.delayMs ?? SESSION_LIVE_INDEX_DELAY_MS;
state.inProgress = true;
state.promise = new Promise<void>((resolve) => {
setTimeoutFn(() => {
measureLifecycleSync('live-index.callback', () => {
try {
dbManager.withCorruptionRecovery(() => {
indexLiveSessionFn(dbManager, sessionManager);
});
} catch (err) {
try { options.onError?.(err); } catch { /* best effort */ }
} finally {
state.inProgress = false;
state.promise = null;
resolve();
}
});
}, delayMs);
});
return true;
}
export async function waitForLiveSessionIndex(
timeoutMs = SESSION_LIVE_INDEX_SHUTDOWN_TIMEOUT_MS,
state: SessionLiveIndexState = sessionLiveIndexState,
): Promise<boolean> {
const promise = state.promise;
if (!state.inProgress || !promise) {
return true;
}
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
promise.then(() => true),
new Promise<boolean>((resolve) => {
timeout = setTimeout(() => resolve(false), timeoutMs);
}),
]);
} finally {
if (timeout) clearTimeout(timeout);
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,111 @@
/**
* /memory-pin — the only writer of STANDING.md besides the user's own editor.
*
* Deliberately not a tool: background review, consolidation and the correction
* detector must have no path into the always-injected block (#121). Keeping it
* a slash command makes model-authored standing instructions structurally
* impossible rather than merely forbidden by prompt.
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { STANDING_MAX_CHARS, STANDING_MAX_ENTRIES } from "../constants.js";
import type { StandingInstructions } from "../store/standing-instructions.js";
const SUBCOMMANDS = ["list", "remove", "clear"] as const;
function formatList(store: StandingInstructions): string[] {
const instructions = store.list();
const lines: string[] = [];
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 📌 Standing Instructions ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
if (instructions.length === 0) {
lines.push(" (none pinned)");
lines.push("");
lines.push(" Pin a rule that must hold in every session:");
lines.push(" /memory-pin never run find / or other root-wide searches");
lines.push("");
return lines;
}
const { injectedCount, omittedCount } = store.render();
const used = instructions.join("\n").length;
for (const [index, instruction] of instructions.entries()) {
lines.push(` ${index + 1}. ${instruction}`);
}
lines.push("");
lines.push(` ${instructions.length}/${STANDING_MAX_ENTRIES} entries · ${used}/${STANDING_MAX_CHARS} chars`);
lines.push(` Injected into every session: ${injectedCount}`);
if (omittedCount > 0) {
lines.push(` ⚠️ ${omittedCount} over budget and NOT injected — remove or shorten entries.`);
}
lines.push(` File: ${store.getFilePath()}`);
lines.push("");
lines.push(" /memory-pin remove <n> · /memory-pin clear");
lines.push("");
return lines;
}
export function registerStandingPinCommand(pi: ExtensionAPI, store: StandingInstructions): void {
pi.registerCommand("memory-pin", {
description: "Pin a standing instruction that is injected into every session",
getArgumentCompletions: (prefix) => {
const trimmed = prefix.trimStart();
if (trimmed.includes(" ")) return null;
return SUBCOMMANDS
.filter((name) => name.startsWith(trimmed))
.map((name) => ({ value: name, label: name }));
},
handler: async (args, ctx) => {
if (!store.isLoaded()) await store.load();
const input = (args ?? "").trim();
const [head, ...rest] = input.split(/\s+/);
const subcommand = head?.toLowerCase();
if (input === "" || subcommand === "list") {
ctx.ui.notify(formatList(store).join("\n"), "info");
return;
}
if (subcommand === "clear") {
const result = await store.clear();
ctx.ui.notify(result.success ? `📌 ${result.message}` : `❌ ${result.error}`, result.success ? "info" : "warning");
return;
}
if (subcommand === "remove") {
const position = Number(rest[0]);
const result = await store.remove(position);
if (!result.success) {
ctx.ui.notify(`❌ ${result.error}`, "warning");
return;
}
ctx.ui.notify([`📌 ${result.message}`, "", ...formatList(store)].join("\n"), "info");
return;
}
const result = await store.add(input);
if (!result.success) {
ctx.ui.notify(`❌ ${result.error}`, "warning");
return;
}
// Echo exactly what will be injected, so the user can see the stored text
// rather than trusting that their phrasing survived normalization.
ctx.ui.notify(
[
`📌 ${result.message}`,
"",
" This is now injected into every session, in all memory modes.",
" It takes effect from your next message.",
"",
].join("\n"),
"info",
);
},
});
}
@@ -0,0 +1,75 @@
/**
* Switch project command — /memory-switch-project lets users manually
* set the active project for project-scoped memory.
*
* Normally, the project is auto-detected from cwd at extension load.
* This command is useful when the user wants to view or manage memory
* for a project they're not currently in.
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import type { MemoryConfig } from "../types.js";
import * as fs from "node:fs/promises";
import * as path from "node:path";
import { resolveProjectsRoot } from "../paths.js";
export function registerSwitchProjectCommand(pi: ExtensionAPI, config?: MemoryConfig): void {
const projectsMemoryDir = config?.projectsMemoryDir ?? "projects-memory";
pi.registerCommand("memory-switch-project", {
description: "Switch the active project for project-scoped memory",
async handler(_args, ctx) {
const projectsDir = resolveProjectsRoot(projectsMemoryDir);
// Discover all project directories (subdirectories of projects-memory/ that have MEMORY.md)
let projects: string[] = [];
try {
const entries = await fs.readdir(projectsDir, { withFileTypes: true });
for (const entry of entries) {
if (!entry.isDirectory()) continue;
try {
await fs.access(path.join(projectsDir, entry.name, "MEMORY.md"));
projects.push(entry.name);
} catch { /* no MEMORY.md — skip */ }
}
} catch {
// Directory doesn't exist — no projects
}
if (projects.length === 0) {
ctx.ui.notify(
"\n 📁 No project memories found.\n\n Project memory is automatically created when you use memory_add with\n target 'project' while working in a project directory.\n",
"info",
);
return;
}
const lines: string[] = [];
lines.push("");
lines.push(" ╔══════════════════════════════════════════════╗");
lines.push(" ║ 📁 Project Memory — Switch ║");
lines.push(" ╚══════════════════════════════════════════════╝");
lines.push("");
lines.push(" Available project memories:");
lines.push("");
for (const proj of projects.sort()) {
// Read entry count
let entryCount = 0;
try {
const raw = await fs.readFile(path.join(projectsDir, proj, "MEMORY.md"), "utf-8");
entryCount = raw.split("\n§\n").filter(Boolean).length;
} catch { /* ignore */ }
lines.push(` 📁 ${proj} (${entryCount} ${entryCount === 1 ? "entry" : "entries"})`);
}
lines.push("");
lines.push(" Use memory_add with target 'project' to manage");
lines.push(" project-scoped memory. Project is auto-detected from");
lines.push(` your current directory: ${process.cwd()}`);
ctx.ui.notify(lines.join("\n"), "info");
},
});
}
@@ -0,0 +1,299 @@
/**
* Markdown memory sync command — /memory-sync-markdown reconciles the SQLite
* search mirror with authoritative Markdown memory files.
*/
import fs from 'node:fs';
import path from 'node:path';
import type { ExtensionAPI, ExtensionCommandContext } from "@earendil-works/pi-coding-agent";
import { DatabaseManager } from '../store/db.js';
import {
reconcileMarkdownFailureScopes,
reconcileMarkdownMemoryScope,
} from '../store/sqlite-memory-store.js';
import { ENTRY_DELIMITER, MEMORY_FILE, USER_FILE } from '../constants.js';
import { AGENT_ROOT } from '../paths.js';
import {
migrateExtensionRoot,
type ExtensionRootMigrationOptions,
} from '../extension-root-migration.js';
import { withMarkdownMutationLock } from '../store/markdown-mutation-lock.js';
export interface BackfillCounters {
filesScanned: number;
entriesScanned: number;
imported: number;
skipped: number;
removed: number;
warnings: string[];
}
export interface MigrationSyncOptions extends ExtensionRootMigrationOptions, SyncMarkdownOptions {
onMigrationSucceeded?: () => void;
}
export interface SyncMarkdownOptions {
force?: boolean;
/**
* Startup scope: reconcile only these project scopes (plus the global
* files), so a pi startup never touches every projects-memory folder.
* `undefined`/`null` keeps the full sweep used by /memory-sync-markdown.
*/
onlyProjects?: string[] | null;
}
function readEntries(filePath: string): string[] {
if (!fs.existsSync(filePath)) return [];
const raw = fs.readFileSync(filePath, 'utf-8').trim();
if (!raw) return [];
return raw.split(ENTRY_DELIMITER).map((entry) => entry.trim()).filter(Boolean);
}
function scanProjectDirs(agentRoot: string, globalDir: string, projectsMemoryDir = "projects-memory"): Array<{ name: string; memoryFile: string }> {
const projectsRoot = path.resolve(agentRoot, projectsMemoryDir);
const projects = new Map<string, string>();
if (fs.existsSync(projectsRoot)) {
for (const name of fs.readdirSync(projectsRoot)) {
if (!isSafeProjectName(name, projectsRoot)) continue;
const memoryFile = resolveAuthoritativeMemoryFile(projectsRoot, name);
if (memoryFile) {
projects.set(name, memoryFile);
}
}
}
const resolvedAgentRoot = path.resolve(agentRoot);
const resolvedGlobalDir = path.resolve(globalDir);
const globalDirName = path.dirname(resolvedGlobalDir) === resolvedAgentRoot
? path.basename(resolvedGlobalDir)
: null;
if (fs.existsSync(agentRoot)) {
for (const name of fs.readdirSync(agentRoot)) {
if ((globalDirName && name === globalDirName) || name === projectsMemoryDir || name === 'skills' || name.startsWith('.')) continue;
if (projects.has(name)) continue;
if (!isSafeProjectName(name, resolvedAgentRoot)) continue;
const memoryFile = resolveAuthoritativeMemoryFile(resolvedAgentRoot, name);
if (memoryFile) {
projects.set(name, memoryFile);
}
}
}
return [...projects.entries()]
.map(([name, memoryFile]) => ({ name, memoryFile }))
.filter(({ memoryFile }) => fs.existsSync(memoryFile));
}
function realpathIfPresent(filePath: string): string {
try {
return fs.realpathSync(filePath);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return path.resolve(filePath);
throw error;
}
}
function resolveAuthoritativeMemoryFile(root: string, projectName: string): string | null {
const canonicalRoot = realpathIfPresent(root);
if (!isSafeProjectName(projectName, path.resolve(root))) return null;
const projectDir = path.join(root, projectName);
let projectStat: fs.Stats;
try {
projectStat = fs.lstatSync(projectDir);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return path.join(canonicalRoot, projectName, MEMORY_FILE);
}
throw error;
}
if (projectStat.isSymbolicLink() || !projectStat.isDirectory()) return null;
const canonicalProjectDir = fs.realpathSync(projectDir);
if (path.dirname(canonicalProjectDir) !== canonicalRoot) return null;
const memoryFile = path.join(projectDir, MEMORY_FILE);
let memoryStat: fs.Stats;
try {
memoryStat = fs.lstatSync(memoryFile);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return path.join(canonicalProjectDir, MEMORY_FILE);
}
throw error;
}
if (memoryStat.isSymbolicLink() || !memoryStat.isFile()) return null;
const canonicalMemoryFile = fs.realpathSync(memoryFile);
if (path.dirname(canonicalMemoryFile) !== canonicalProjectDir || path.basename(canonicalMemoryFile) !== MEMORY_FILE) {
return null;
}
return canonicalMemoryFile;
}
function isSafeProjectName(name: string, projectsRoot: string): boolean {
if (!name || name === '.' || name === '..' || name.includes('/') || name.includes('\\') || path.isAbsolute(name)) {
return false;
}
const projectDir = path.resolve(projectsRoot, name);
return path.dirname(projectDir) === projectsRoot && path.basename(projectDir) === name;
}
export async function syncMarkdownMemoriesToSqlite(
dbManager: DatabaseManager,
globalDir: string,
projectsMemoryDir?: string,
agentRoot = AGENT_ROOT,
options?: SyncMarkdownOptions,
): Promise<BackfillCounters & { projectCount: number }> {
const counters: BackfillCounters = {
filesScanned: 0,
entriesScanned: 0,
imported: 0,
skipped: 0,
removed: 0,
warnings: [],
};
const globalMemoryFile = path.join(globalDir, MEMORY_FILE);
const globalUserFile = path.join(globalDir, USER_FILE);
const globalFailureFile = path.join(globalDir, 'failures.md');
const reconcileFile = async (
filePath: string | null,
target: 'memory' | 'user' | 'failure',
project: string | null = null,
) => {
const reconcile = () => {
if (filePath && fs.existsSync(filePath)) counters.filesScanned++;
const entries = filePath ? readEntries(filePath) : [];
counters.entriesScanned += entries.length;
try {
const result = target === 'failure'
? reconcileMarkdownFailureScopes(dbManager, entries, options)
: reconcileMarkdownMemoryScope(dbManager, entries, target, project, options);
counters.imported += result.inserted;
counters.skipped += result.existing;
counters.removed += result.removed;
if (result.degraded) {
counters.warnings.push(
`${path.basename(project ?? 'global')}/${target}: FTS5 search index error (${result.degradedReason}); run /memory-sync-markdown to rebuild`,
);
}
} catch (err) {
counters.warnings.push(
`${path.basename(project ?? 'global')}/${target}: ${err instanceof Error ? err.message : String(err)}`,
);
}
};
if (filePath) await withMarkdownMutationLock(filePath, reconcile);
else reconcile();
};
await reconcileFile(globalMemoryFile, 'memory');
await reconcileFile(globalUserFile, 'user');
await reconcileFile(globalFailureFile, 'failure');
// Startup passes onlyProjects (global files + current project) so boot cost
// never scales with the projects-memory folder count. The full sweep below
// stays for /memory-sync-markdown, which is the repair path for the scopes
// a scoped startup skips (including orphan pruning outside the scope).
const scopedProjects = options?.onlyProjects ?? null;
if (scopedProjects) {
const projectsRoot = path.resolve(agentRoot, projectsMemoryDir ?? 'projects-memory');
for (const projectName of new Set(scopedProjects)) {
const memoryFile = resolveAuthoritativeMemoryFile(projectsRoot, projectName);
await reconcileFile(memoryFile, 'memory', projectName);
}
return { ...counters, projectCount: new Set(scopedProjects).size };
}
const projects = scanProjectDirs(agentRoot, globalDir, projectsMemoryDir);
const projectFiles = new Map(projects.map((project) => [project.name, project.memoryFile]));
const mirroredProjects = dbManager.getDb().prepare(`
SELECT DISTINCT project
FROM memories
WHERE project IS NOT NULL AND target = 'memory'
`).all() as Array<{ project: string }>;
const projectNames = new Set([
...projectFiles.keys(),
...mirroredProjects.map(({ project }) => project),
]);
const projectsRoot = path.resolve(agentRoot, projectsMemoryDir ?? 'projects-memory');
for (const projectName of projectNames) {
const memoryFile = projectFiles.get(projectName)
?? resolveAuthoritativeMemoryFile(projectsRoot, projectName);
await reconcileFile(memoryFile, 'memory', projectName);
}
return { ...counters, projectCount: projectNames.size };
}
export async function migrateThenSyncMarkdownMemories(
dbManager: DatabaseManager,
legacyGlobalDir: string | null,
globalDir: string,
projectsMemoryDir?: string,
agentRoot = AGENT_ROOT,
migrationOptions: MigrationSyncOptions = {},
): Promise<BackfillCounters & { projectCount: number }> {
if (legacyGlobalDir) {
const migration = await migrateExtensionRoot(legacyGlobalDir, globalDir, migrationOptions);
const sessionsFailure = migration.criticalFailures.find((failure) => failure.name === 'sessions.db');
if (sessionsFailure) {
throw new Error(`sessions.db migration failed: ${sessionsFailure.message}`);
}
migrationOptions.onMigrationSucceeded?.();
}
return await syncMarkdownMemoriesToSqlite(dbManager, globalDir, projectsMemoryDir, agentRoot, {
force: migrationOptions.force,
onlyProjects: migrationOptions.onlyProjects,
});
}
export function registerSyncMarkdownMemoriesCommand(
pi: ExtensionAPI,
dbManager: DatabaseManager,
globalDir: string,
projectsMemoryDir?: string,
agentRoot = AGENT_ROOT,
): void {
pi.registerCommand('memory-sync-markdown', {
description: 'Reconcile the SQLite search mirror with Markdown memories',
handler: async (_args, ctx: ExtensionCommandContext) => {
ctx.ui.notify('🔄 Reconciling the SQLite search mirror with Markdown memories...', 'info');
try {
const counters = await syncMarkdownMemoriesToSqlite(dbManager, globalDir, projectsMemoryDir, agentRoot, { force: true });
let output = `\n✅ Markdown → SQLite sync complete!\n\n`;
output += `📊 Results:\n`;
output += `├─ Files scanned: ${counters.filesScanned}\n`;
output += `├─ Entries scanned: ${counters.entriesScanned}\n`;
output += `├─ Imported into SQLite: ${counters.imported}\n`;
output += `├─ Skipped as duplicates: ${counters.skipped}\n`;
output += `└─ Removed orphaned rows: ${counters.removed}\n`;
if (counters.projectCount > 0) {
output += `\n📁 Project memories scanned: ${counters.projectCount}\n`;
}
if (counters.warnings.length > 0) {
output += `\n⚠️ Warnings (${counters.warnings.length}):\n`;
for (const warning of counters.warnings.slice(0, 5)) {
output += `├─ ${warning}\n`;
}
if (counters.warnings.length > 5) {
output += `└─ ... and ${counters.warnings.length - 5} more\n`;
}
}
output += `\n💡 Re-running this command is safe — existing SQLite rows are de-duplicated.`;
ctx.ui.notify(output, 'info');
} catch (err) {
ctx.ui.notify(`❌ Markdown sync failed: ${err instanceof Error ? err.message : String(err)}`, 'error');
}
},
});
}