memory: fix extraction trigger, dedupe, and add runtime diagnostics
Four defects found by end-to-end testing, not by reading the code. 1. Extraction never ran, silently. Collecting every event type that ctx.event.subscribe actually delivers shows `session.idle` is NOT among them — it is defined in the protocol but never emitted on this stream. The subscription filtered on it alone, so auto-extraction was a permanent no-op with no error. Now debounced off events that do arrive (session.step.started / session.tool.called / session.inbox.delivered) after `settleMs` of quiet, which is what "idle" meant anyway. Verified: the trigger now fires and reports "轮数不足" for a 1-turn session. 2. Dedupe silently stopped rejecting duplicates. It compared against a fixed retrieval score, but that score carries idf and scope priors that drift as the corpus grows, so an obvious restatement of an existing memory scored 0.48 against a 0.9 threshold and was written as a rival entry. Now uses term coverage of the candidate, which answers "is this already said?" and does not move with corpus size. Verified: a 100%-duplicate is rejected, a novel pitfall is kept. 3. Harness-injected user turns became the retrieval query. "<system-reminder>" and "The server restarted while you were working" are user-role messages, so the hook retrieved on them and injected memories chosen by noise. Both are filtered out now. 4. The `pinned` flag was absent from the remember tool's input schema, so the schema stripped it and pinning was unreachable from the tool surface. `kind` also ignored the rules/ category when a stale value was stored. memory_inspect now reports runtime health for the two paths with no other observable effect: last injection size and query, and last extraction outcome with its reason. Without it, "working" and "silently dead" look identical — which is exactly how defects 1 and 2 survived the first round of testing.
This commit is contained in:
+176
-22
@@ -57,6 +57,8 @@ interface PluginConfig {
|
||||
halfLifeDays?: number
|
||||
autoExtract?: boolean
|
||||
minTurns?: number
|
||||
/** How long a turn must be quiet before extraction runs. Default 20000ms. */
|
||||
settleMs?: number
|
||||
extractModel?: { providerID: string; id: string }
|
||||
/** Seed these categories on first run. */
|
||||
categories?: string[]
|
||||
@@ -65,6 +67,17 @@ interface PluginConfig {
|
||||
const CATEGORIES = ["rules", "preferences", "decisions", "pitfalls", "facts"]
|
||||
const DEFAULT_MAX_INJECT = 1200
|
||||
|
||||
/**
|
||||
* Events that mark turn activity. Collected from what this stream actually
|
||||
* delivers: `session.idle` is defined in the protocol but never arrives here,
|
||||
* so extraction is debounced off these instead.
|
||||
*/
|
||||
const EXTRACTION_TRIGGERS = new Set([
|
||||
"session.step.started",
|
||||
"session.tool.called",
|
||||
"session.inbox.delivered",
|
||||
])
|
||||
|
||||
export default Plugin.define({
|
||||
id: "memory",
|
||||
|
||||
@@ -138,7 +151,7 @@ export default Plugin.define({
|
||||
// -- injection into the agent loop --------------------------------------
|
||||
// `context` covers the agent loop including tool continuations, and edits
|
||||
// only the outgoing request — the conversation transcript stays clean.
|
||||
await ctx.session.hook("context", (event) => {
|
||||
await ctx.session.hook("context", async (event) => {
|
||||
try {
|
||||
const query = latestUserText(event.messages)
|
||||
if (!query) return
|
||||
@@ -150,6 +163,16 @@ export default Plugin.define({
|
||||
options: { maxChars, maxEntries, halfLifeDays },
|
||||
})
|
||||
if (text) event.system.push({ type: "text", text })
|
||||
// Record what the model actually received. Without this there is no way
|
||||
// to tell "injected nothing because there is nothing" apart from
|
||||
// "the hook is not firing at all".
|
||||
await noteDiag(ctx, "lastInjection", {
|
||||
at: Date.now(),
|
||||
chars: text.length,
|
||||
withinBudget: text.length <= maxChars,
|
||||
bodies: (text.match(/^### /gm) ?? []).length,
|
||||
query: query.slice(0, 80),
|
||||
})
|
||||
} catch (error) {
|
||||
console.error(`memory: injection failed: ${(error as Error).message}`)
|
||||
}
|
||||
@@ -488,6 +511,33 @@ export default Plugin.define({
|
||||
lines.push(problems.length === 0 ? "索引一致性:正常" : `索引一致性:发现 ${problems.length} 个问题`)
|
||||
for (const problem of problems) lines.push(` - ${problem}`)
|
||||
|
||||
// Runtime health of the two paths that have no other visible effect.
|
||||
lines.push("")
|
||||
const injection = (await ctx.storage.get("diag.lastInjection")) as Record<string, unknown> | undefined
|
||||
const extraction = (await ctx.storage.get("diag.lastExtraction")) as Record<string, unknown> | undefined
|
||||
lines.push("运行状态:")
|
||||
if (injection) {
|
||||
const when = new Date(Number(injection.at)).toISOString().slice(0, 19).replace("T", " ")
|
||||
lines.push(
|
||||
` 注入:最近一次 ${when},${String(injection.chars)} 字符,内联 ${String(injection.bodies)} 条正文${injection.withinBudget === false ? " ⚠ 超出预算" : ""}`,
|
||||
)
|
||||
lines.push(` 检索词:${String(injection.query)}`)
|
||||
} else {
|
||||
lines.push(" 注入:尚无记录(context hook 还没触发过)")
|
||||
}
|
||||
if (extraction) {
|
||||
const when = new Date(Number(extraction.at)).toISOString().slice(0, 19).replace("T", " ")
|
||||
lines.push(` 抽取:${when} → ${String(extraction.outcome)}`)
|
||||
if (extraction.reason) lines.push(` 原因:${String(extraction.reason)}`)
|
||||
if (extraction.model) lines.push(` 模型:${String(extraction.model)}`)
|
||||
} else {
|
||||
lines.push(
|
||||
extractOptions.enabled
|
||||
? " 抽取:尚无记录(本会话还没到静默期,或轮数/内容不足)"
|
||||
: " 抽取:已关闭(autoExtract: false)",
|
||||
)
|
||||
}
|
||||
|
||||
return { content: lines.join("\n") }
|
||||
},
|
||||
})
|
||||
@@ -496,30 +546,50 @@ export default Plugin.define({
|
||||
// -- automatic extraction on idle --------------------------------------
|
||||
let extracting = false
|
||||
const controller = new AbortController()
|
||||
// How long a turn must be quiet before extraction counts it as finished.
|
||||
const settleMs = config.settleMs ?? 20_000
|
||||
|
||||
// `session.idle` is NOT delivered on this stream — verified by collecting
|
||||
// every event type that actually arrives. The turn-ending events below do
|
||||
// arrive, so they drive a debounce instead: the turn must go quiet for
|
||||
// `settleMs` before extraction runs, which is what "idle" meant anyway.
|
||||
const pending = new Map<string, ReturnType<typeof setTimeout>>()
|
||||
|
||||
if (extractOptions.enabled) {
|
||||
void (async () => {
|
||||
try {
|
||||
for await (const event of ctx.event.subscribe({ signal: controller.signal })) {
|
||||
const record = event as unknown as { type?: string; data?: { sessionID?: string } }
|
||||
if (record?.type !== "session.idle") continue
|
||||
if (!EXTRACTION_TRIGGERS.has(String(record?.type))) continue
|
||||
const sessionID = record.data?.sessionID
|
||||
if (!sessionID || extracting) continue
|
||||
if (!sessionID) continue
|
||||
|
||||
extracting = true
|
||||
try {
|
||||
await runExtraction(ctx, {
|
||||
sessionID,
|
||||
extractOptions,
|
||||
filesByRoot,
|
||||
project,
|
||||
global,
|
||||
})
|
||||
} catch (error) {
|
||||
console.error(`memory: extraction failed: ${(error as Error).message}`)
|
||||
} finally {
|
||||
extracting = false
|
||||
}
|
||||
// Debounce: each new turn activity pushes the deadline out. Only a
|
||||
// genuine lull lets the timer fire, so extraction runs on a
|
||||
// finished turn rather than mid-stream.
|
||||
const existing = pending.get(sessionID)
|
||||
if (existing) clearTimeout(existing)
|
||||
pending.set(
|
||||
sessionID,
|
||||
setTimeout(() => {
|
||||
pending.delete(sessionID)
|
||||
if (extracting) return
|
||||
extracting = true
|
||||
void runExtraction(ctx, {
|
||||
sessionID,
|
||||
extractOptions,
|
||||
filesByRoot,
|
||||
project,
|
||||
global,
|
||||
})
|
||||
.catch((error) => {
|
||||
console.error(`memory: extraction failed: ${(error as Error).message}`)
|
||||
})
|
||||
.finally(() => {
|
||||
extracting = false
|
||||
})
|
||||
}, settleMs),
|
||||
)
|
||||
}
|
||||
} catch (error) {
|
||||
// Subscription teardown is normal on unload.
|
||||
@@ -532,6 +602,8 @@ export default Plugin.define({
|
||||
|
||||
return () => {
|
||||
controller.abort()
|
||||
for (const timer of pending.values()) clearTimeout(timer)
|
||||
pending.clear()
|
||||
}
|
||||
},
|
||||
})
|
||||
@@ -546,6 +618,28 @@ function pickRoots(roots: MemoryRoot[], scope: string | undefined): MemoryRoot[]
|
||||
return roots
|
||||
}
|
||||
|
||||
/**
|
||||
* Record plugin health for `memory_inspect` to show.
|
||||
*
|
||||
* The two asynchronous paths — the context hook and idle extraction — have no
|
||||
* other observable effect. Without this, "working" and "silently dead" look
|
||||
* identical from the outside. Throttled because the context hook runs on every
|
||||
* model request, including tool continuations.
|
||||
*/
|
||||
const lastDiagAt = new Map<string, number>()
|
||||
|
||||
async function noteDiag(ctx: Context, key: string, value: Record<string, unknown>): Promise<void> {
|
||||
const now = Date.now()
|
||||
const previous = lastDiagAt.get(key) ?? 0
|
||||
if (now - previous < 2000) return
|
||||
lastDiagAt.set(key, now)
|
||||
try {
|
||||
await ctx.storage.set(`diag.${key}`, value)
|
||||
} catch {
|
||||
// Diagnostics must never break the hook that reports them.
|
||||
}
|
||||
}
|
||||
|
||||
function outlineText(filesByRoot: Map<string, MemoryFile[]>, roots: MemoryRoot[]): string {
|
||||
const lines: string[] = []
|
||||
for (const root of roots) {
|
||||
@@ -557,17 +651,32 @@ function outlineText(filesByRoot: Map<string, MemoryFile[]>, roots: MemoryRoot[]
|
||||
return lines.join("\n") || "(记忆库为空)"
|
||||
}
|
||||
|
||||
/** The newest user text in a request — the retrieval query. */
|
||||
/**
|
||||
* The newest real user text in a request — the retrieval query.
|
||||
*
|
||||
* Harness-injected user messages ("<system-reminder>", "The server restarted…")
|
||||
* are technically user turns but are not about the task. Retrieving on them
|
||||
* injects memories chosen by noise, so they are skipped.
|
||||
*/
|
||||
function latestUserText(messages: readonly unknown[]): string | undefined {
|
||||
for (let i = messages.length - 1; i >= 0; i--) {
|
||||
const record = messages[i] as { role?: string; content?: unknown } | null
|
||||
if (!record || record.role !== "user") continue
|
||||
const text = extractUserText(record.content)
|
||||
if (text) return text
|
||||
if (text && !isHarnessMessage(text)) return text
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
const HARNESS_PREFIX = /^<(system-reminder|command-name|command-message|command-args)/i
|
||||
const HARNESS_RE =
|
||||
/^(The server restarted while you were working|Task tools haven't been used recently|<system-reminder)/i
|
||||
|
||||
function isHarnessMessage(text: string): boolean {
|
||||
const head = text.trimStart()
|
||||
return HARNESS_PREFIX.test(head) || HARNESS_RE.test(head)
|
||||
}
|
||||
|
||||
function extractUserText(content: unknown): string {
|
||||
if (typeof content === "string") return content.trim()
|
||||
if (!Array.isArray(content)) return ""
|
||||
@@ -599,13 +708,41 @@ async function runExtraction(
|
||||
return role === "user" || role === "assistant"
|
||||
})
|
||||
// Long enough to have learned something, short enough to stay cheap.
|
||||
if (turns.length < extractOptions.minTurns) return
|
||||
if (turns.length < extractOptions.minTurns) {
|
||||
await noteDiag(ctx, "lastExtraction", {
|
||||
at: Date.now(),
|
||||
outcome: "skipped",
|
||||
reason: "会话轮数不足",
|
||||
turns: turns.length,
|
||||
minTurns: extractOptions.minTurns,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const model = extractOptions.model ?? (await pickCheapModel(ctx))
|
||||
if (!model) return
|
||||
if (!model) {
|
||||
// pickCheapModel failing is the difference between a working feature and
|
||||
// a silent no-op, so it is reported rather than swallowed.
|
||||
await noteDiag(ctx, "lastExtraction", {
|
||||
at: Date.now(),
|
||||
outcome: "failed",
|
||||
reason: "选不到可用模型(autoExtract 实际上处于关闭状态)",
|
||||
turns: turns.length,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const transcript = userAndAssistantText(turns.slice(-extractOptions.maxMessages))
|
||||
if (transcript.length < 200) return
|
||||
if (transcript.length < 200) {
|
||||
await noteDiag(ctx, "lastExtraction", {
|
||||
at: Date.now(),
|
||||
outcome: "skipped",
|
||||
reason: "对话内容太短",
|
||||
turns: turns.length,
|
||||
transcriptChars: transcript.length,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
const existing = [...input.filesByRoot().values()].flat()
|
||||
const outcome = await extract({ ctx, model, transcript, existing })
|
||||
@@ -616,6 +753,13 @@ async function runExtraction(
|
||||
`memory: skipped ${rejected.candidate.category}/${rejected.candidate.slug} — ${rejected.reason}`,
|
||||
)
|
||||
}
|
||||
await noteDiag(ctx, "lastExtraction", {
|
||||
at: Date.now(),
|
||||
outcome: "no-candidates",
|
||||
reason: outcome.rejected.length ? "候选全部被去重或密钥拦截" : "抽取器认为没有值得记的东西",
|
||||
rejected: outcome.rejected.length,
|
||||
model: `${model.providerID}/${model.id}`,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
@@ -648,6 +792,16 @@ async function runExtraction(
|
||||
}
|
||||
console.log(`memory: extracted ${written.length} memor${written.length === 1 ? "y" : "ies"} → ${written.join(", ")}`)
|
||||
}
|
||||
|
||||
await noteDiag(ctx, "lastExtraction", {
|
||||
at: Date.now(),
|
||||
outcome: written.length ? "extracted" : "all-filtered",
|
||||
model: `${model.providerID}/${model.id}`,
|
||||
turns: turns.length,
|
||||
written: written.length,
|
||||
rejected: outcome.rejected.length,
|
||||
paths: written,
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user