mirror of
https://github.com/Kilo-Org/kilocode.git
synced 2026-08-28 19:11:03 +08:00
feat(memory): add @kilocode/kilo-memory effect runtime layer (#11845)
* feat(memory): add @kilocode/kilo-memory effect runtime layer * refactor(memory): drop dead defaultLayer alias * chore: restore @kilocode/kilo-memory changeset from core * fix(memory): isolate event sink failures and release per-session turn locks - events: swallow+log host sink failures so best-effort event wiring never fails a memory op that already persisted (restores guard the port dropped) - service/turn: drop the memoized turn lock once a session settles so the lock map no longer grows unbounded in a long-lived shared backend * fix(memory): scope model-reported duplicate skips to claimed file/section verifySkips() verified model-claimed duplicates against all stored items regardless of scope. Add optional file/section to the skip schema, instruct the model to report them for duplicate skips, and thread them into the duplicate check so a cross-scope match downgrades to 'unsupported' instead of confirming. Matches the scoping duplicateOps() already applies to adds. * fix(memory): release per-root abort controllers and reject unscoped duplicate skips - timers: ref-count the per-root AbortController and drop it when the last in-flight capture settles (released via an ensuring finalizer in capture), so the signals map no longer grows for every root a shared backend touches - outcome/prompt: a 'duplicate' skip that names a file but no section can't be verified without risking a cross-section false confirm, so treat it as unverified; instruct the model to report both file and section * fix(memory): only confirm duplicate skips that name the exact file+section Extend the guard so any 'duplicate' skip missing either scope field — not just section — is treated as unverified. A fully-unscoped claim could otherwise still confirm against unrelated memory via fuzzy text matching. Confirm only when the model pins the exact file+section; everything else downgrades to advisory. * fix(memory): ref-count session turn locks instead of eager delete Eager delete could hand a queued close() a fresh semaphore while a peer still held the old one, reintroducing overlapping turn-close work. Ref-count holders (queued acquirers included) and drop the entry only when the last holder leaves; close() always releases its own holder and a deferred flush takes its own turnLock/dropLock pair.
This commit is contained in:
Johnny Eric Amancio
committed by
GitHub
GitHub
parent
98f873a42a
commit
b48570e93b
@@ -2,4 +2,6 @@
|
||||
"@kilocode/kilo-memory": minor
|
||||
---
|
||||
|
||||
Add the project memory storage, indexing, recall, and capture safety foundation.
|
||||
Add the `@kilocode/kilo-memory` foundation package: project memory storage, indexing, recall,
|
||||
consolidation, and the Effect runtime layer (service, capture orchestration, and runtime ports).
|
||||
Host wiring (CLI tools, prompts, HTTP API) lands in follow-up CLI/extension PRs.
|
||||
|
||||
@@ -269,6 +269,7 @@
|
||||
"name": "@kilocode/kilo-memory",
|
||||
"version": "7.3.45",
|
||||
"dependencies": {
|
||||
"effect": "catalog:",
|
||||
"zod": "catalog:",
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
@@ -16,6 +16,18 @@
|
||||
"./capture": "./src/capture/capture.ts",
|
||||
"./commands": "./src/commands.ts",
|
||||
"./digest": "./src/capture/digest.ts",
|
||||
"./effect": "./src/effect/index.ts",
|
||||
"./effect/capture": "./src/effect/capture.ts",
|
||||
"./effect/config": "./src/effect/config.ts",
|
||||
"./effect/errors": "./src/effect/errors.ts",
|
||||
"./effect/events": "./src/effect/events.ts",
|
||||
"./effect/instance": "./src/effect/instance.ts",
|
||||
"./effect/log": "./src/effect/log.ts",
|
||||
"./effect/paths": "./src/effect/paths.ts",
|
||||
"./effect/ports": "./src/effect/ports.ts",
|
||||
"./effect/service": "./src/effect/service.ts",
|
||||
"./effect/timers": "./src/effect/timers.ts",
|
||||
"./effect/turn": "./src/effect/turn.ts",
|
||||
"./memory": "./src/memory.ts",
|
||||
"./ops": "./src/capture/ops.ts",
|
||||
"./paths": "./src/storage/paths.ts",
|
||||
@@ -36,6 +48,7 @@
|
||||
"test:ci": "mkdir -p .artifacts/unit && bun test --timeout 30000 --reporter=junit --reporter-outfile=.artifacts/unit/junit.xml"
|
||||
},
|
||||
"dependencies": {
|
||||
"effect": "catalog:",
|
||||
"zod": "catalog:"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
@@ -249,7 +249,11 @@ export namespace MemoryOperations {
|
||||
function planAdd(plan: Plan, item: Prepared, now: number) {
|
||||
const found = duplicate({ item, inventory: plan.inventory })
|
||||
const next = found ? rekey({ item, key: found.key }) : item
|
||||
const result = MemoryMarkdown.upsert({ text: plan.docs.get(next.file) ?? "", section: next.section, line: next.line })
|
||||
const result = MemoryMarkdown.upsert({
|
||||
text: plan.docs.get(next.file) ?? "",
|
||||
section: next.section,
|
||||
line: next.line,
|
||||
})
|
||||
if (result.changed) {
|
||||
plan.docs.set(next.file, result.text)
|
||||
plan.touched.add(next.file)
|
||||
|
||||
@@ -4,7 +4,12 @@ import { MemoryShared } from "../recall/shared"
|
||||
import type { MemoryFiles } from "../storage/store"
|
||||
import type { CaptureSkip } from "./parse"
|
||||
|
||||
export type CaptureSourceItem = { id: string; text: string }
|
||||
export type CaptureSourceItem = {
|
||||
id: string
|
||||
text: string
|
||||
file?: MemoryOperations.Add["file"]
|
||||
section?: string
|
||||
}
|
||||
|
||||
export type CaptureDetail = {
|
||||
type: "saved" | "skipped"
|
||||
@@ -96,13 +101,21 @@ function tokens(input: string) {
|
||||
return MemoryShared.terms(input)
|
||||
}
|
||||
|
||||
function duplicate(text: string | undefined, items: CaptureSourceItem[]) {
|
||||
function duplicate(input: {
|
||||
text: string | undefined
|
||||
items: CaptureSourceItem[]
|
||||
file?: MemoryOperations.Add["file"]
|
||||
section?: string
|
||||
}) {
|
||||
const text = input.text
|
||||
if (!text) return
|
||||
const query = tokens(text)
|
||||
if (query.length === 0) return
|
||||
// Majority overlap required: a few shared generic terms must not confirm a duplicate.
|
||||
const needed = Math.max(Math.min(3, query.length), Math.ceil(query.length / 2))
|
||||
const hits = items
|
||||
const hits = input.items
|
||||
.filter((item) => !input.file || !item.file || item.file === input.file)
|
||||
.filter((item) => !input.section || !item.section || item.section === input.section)
|
||||
.map((item) => {
|
||||
const hay = tokens(item.text)
|
||||
const found = query.filter((term) => hay.includes(term)).length
|
||||
@@ -113,26 +126,36 @@ function duplicate(text: string | undefined, items: CaptureSourceItem[]) {
|
||||
return hits.at(0)?.item.id
|
||||
}
|
||||
|
||||
/** Model-claimed duplicates are verified against stored entries; unconfirmed claims are rescued as ops instead of lost. */
|
||||
/** Model-claimed duplicates are verified against stored entries; unconfirmed claims are downgraded to
|
||||
* "unsupported" so they read as advisory rather than confirmed against a real entry. */
|
||||
export function verifySkips(input: { skipped: CaptureSkip[]; items: CaptureSourceItem[] }) {
|
||||
const skipped: CaptureSkip[] = []
|
||||
const rescued: MemoryOperations.Op[] = []
|
||||
for (const item of input.skipped) {
|
||||
if (item.reason !== "duplicate" || !item.text) {
|
||||
skipped.push(item)
|
||||
continue
|
||||
}
|
||||
const source = duplicate(item.text, input.items)
|
||||
// A model-claimed duplicate is only confirmable when it names the exact scope (file + section).
|
||||
// Any missing scope field would let fuzzy text matching confirm against unrelated memory, so
|
||||
// downgrade partially-scoped or unscoped claims to advisory instead.
|
||||
const scoped = item.file !== undefined && item.section !== undefined
|
||||
const source = scoped
|
||||
? duplicate({ text: item.text, items: input.items, file: item.file, section: item.section })
|
||||
: undefined
|
||||
if (source) {
|
||||
skipped.push({ ...item, duplicateOf: item.duplicateOf ?? source })
|
||||
continue
|
||||
}
|
||||
skipped.push({ reason: "unsupported", text: item.text })
|
||||
}
|
||||
return { skipped, rescued }
|
||||
return { skipped }
|
||||
}
|
||||
|
||||
export function duplicateOps(input: { ops: MemoryOperations.Op[]; skipped: CaptureSkip[]; items: CaptureSourceItem[] }) {
|
||||
export function duplicateOps(input: {
|
||||
ops: MemoryOperations.Op[]
|
||||
skipped: CaptureSkip[]
|
||||
items: CaptureSourceItem[]
|
||||
}) {
|
||||
const skipped = [...input.skipped]
|
||||
const ops = input.ops.filter((item) => {
|
||||
if (item.action !== "add") return true
|
||||
@@ -141,7 +164,12 @@ export function duplicateOps(input: { ops: MemoryOperations.Op[]; skipped: Captu
|
||||
skipped.push(rejected)
|
||||
return false
|
||||
}
|
||||
const source = duplicate(`${item.key} ${item.text}`, input.items)
|
||||
const source = duplicate({
|
||||
text: `${item.key} ${item.text}`,
|
||||
items: input.items,
|
||||
file: item.file,
|
||||
section: item.section,
|
||||
})
|
||||
if (!source) return true
|
||||
skipped.push({ reason: "duplicate", text: item.text, duplicateOf: source })
|
||||
return false
|
||||
|
||||
@@ -26,8 +26,7 @@ const key = z.string().trim().min(1).max(80)
|
||||
const value = z.string().trim().min(1).max(2_000)
|
||||
const addSchema = (
|
||||
op: "upsert_project_fact" | "upsert_project_decision" | "upsert_project_constraint" | "append_correction",
|
||||
) =>
|
||||
z.object({ op: z.literal(op), key, value }).strict()
|
||||
) => z.object({ op: z.literal(op), key, value }).strict()
|
||||
|
||||
export const typedSchema = z
|
||||
.object({
|
||||
@@ -64,6 +63,10 @@ export const typedSchema = z
|
||||
reason: skip,
|
||||
text: z.string().max(500).optional(),
|
||||
duplicateOf: z.string().max(240).optional(),
|
||||
// Optional scope of the entry this skip claims to duplicate, so duplicate verification
|
||||
// matches within the same file/section instead of across all stored memory.
|
||||
file: z.enum(["project.md", "environment.md", "corrections.md"]).optional(),
|
||||
section: z.string().max(80).optional(),
|
||||
})
|
||||
.strict(),
|
||||
)
|
||||
|
||||
@@ -1,7 +1,16 @@
|
||||
export const MEMORY_USAGE =
|
||||
"/memory [project] enable|status|show|inspect|auto status|auto on|auto off|remember <text>|correct <text>|forget <query>|purge confirm|rebuild|disable"
|
||||
|
||||
export const MEMORY_OPERATIONS = ["enable", "disable", "rebuild", "remember", "correct", "forget", "purge", "auto"] as const
|
||||
export const MEMORY_OPERATIONS = [
|
||||
"enable",
|
||||
"disable",
|
||||
"rebuild",
|
||||
"remember",
|
||||
"correct",
|
||||
"forget",
|
||||
"purge",
|
||||
"auto",
|
||||
] as const
|
||||
export const MEMORY_PROMPT_OPERATIONS = ["remember", "forget"] as const
|
||||
|
||||
export type MemoryOperation = (typeof MEMORY_OPERATIONS)[number]
|
||||
|
||||
@@ -0,0 +1,509 @@
|
||||
import { Cause, Effect } from "effect"
|
||||
import {
|
||||
auditOps,
|
||||
cap,
|
||||
capturePlan,
|
||||
digestPrompt,
|
||||
digestSchema,
|
||||
duplicateOps,
|
||||
errorReason,
|
||||
evidence,
|
||||
fallbackDigest,
|
||||
guardReason,
|
||||
hasDurableDiff,
|
||||
mergeOps,
|
||||
notice,
|
||||
parseDigest,
|
||||
parseJson,
|
||||
parseOps,
|
||||
skipped,
|
||||
summarize,
|
||||
summarizeDiffs,
|
||||
typedPrompt,
|
||||
typedSchema,
|
||||
usage,
|
||||
verifySkips,
|
||||
type CaptureReason,
|
||||
type CaptureSkip,
|
||||
type CaptureSourceItem,
|
||||
} from "../capture/capture"
|
||||
import { MemoryDigest } from "../capture/digest"
|
||||
import type { MemoryOperations } from "../capture/ops"
|
||||
import { MemoryRedact } from "../capture/redact"
|
||||
import { MemorySchema } from "../schema"
|
||||
import { MemoryShared } from "../recall/shared"
|
||||
import { MemoryEvents } from "./events"
|
||||
import { MemoryLog } from "./log"
|
||||
import type { MemoryPorts } from "./ports"
|
||||
import { MemoryService } from "./service"
|
||||
import { MemoryTimers } from "./timers"
|
||||
|
||||
const MESSAGE_WINDOW = 24
|
||||
|
||||
/** Heuristic: an assistant answer that mostly restates injected instructions/source files is not
|
||||
* durable project memory and should not be consolidated. */
|
||||
function provenance(input: { assistant: string }) {
|
||||
const assistant = input.assistant.trim()
|
||||
const markers = [/\bsystem\s*\/\s*developer\b/gi, /\bagents\.md\b/gi, /\bclaude\.md\b/gi].reduce(
|
||||
(sum, item) => sum + (assistant.match(item)?.length ?? 0),
|
||||
0,
|
||||
)
|
||||
const list = assistant.split("\n").filter((line) => /^\s*[-*]\s+\S/.test(line)).length
|
||||
return markers >= 4 || (markers >= 3 && list >= 2)
|
||||
}
|
||||
|
||||
function typedExisting(memory: MemoryService.Interface, root: string) {
|
||||
return memory.sources({ root }).pipe(
|
||||
Effect.map((sources) => {
|
||||
const blocks = MemorySchema.Sources.map((file) => {
|
||||
const body = sources[file].trim()
|
||||
if (!body) return ""
|
||||
return [`### source ${file}`, body].join("\n")
|
||||
})
|
||||
return blocks.filter(Boolean).join("\n")
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
function itemSource(file: MemorySchema.Source, text: string): CaptureSourceItem[] {
|
||||
return MemoryShared.source({ file, text })
|
||||
}
|
||||
|
||||
function typedItems(memory: MemoryService.Interface, root: string) {
|
||||
return memory
|
||||
.sources({ root })
|
||||
.pipe(Effect.map((sources) => MemorySchema.Sources.flatMap((file) => itemSource(file, sources[file]))))
|
||||
}
|
||||
|
||||
export namespace MemoryCapture {
|
||||
export const turn = Effect.fn("MemoryCapture.turn")(function* (input: {
|
||||
root: string
|
||||
sessionID: string
|
||||
session: MemoryPorts.SessionPort
|
||||
model: MemoryPorts.ModelPort
|
||||
reason?: CaptureReason
|
||||
bypassInterval?: boolean
|
||||
memoryModel?: string
|
||||
}) {
|
||||
const root = input.root
|
||||
// Acquire first (sync, cannot fail) so the matching `release` in the finalizer below always pairs
|
||||
// with this acquire regardless of where the turn exits.
|
||||
const signal = MemoryTimers.signal(root)
|
||||
const memory = yield* MemoryService.Service
|
||||
yield* memory.prepare({ root })
|
||||
const state = yield* memory.state({ root })
|
||||
const reported = new Set<string>()
|
||||
const fail = (reason: string) =>
|
||||
Effect.promise(async () => {
|
||||
const safe = MemoryRedact.text(reason)
|
||||
if (reported.has(safe)) return
|
||||
reported.add(safe)
|
||||
await MemoryEvents.publish({
|
||||
event: "error",
|
||||
payload: MemoryEvents.status({
|
||||
root,
|
||||
state,
|
||||
phase: "error",
|
||||
reason: safe,
|
||||
sessionID: input.sessionID,
|
||||
}),
|
||||
})
|
||||
})
|
||||
const skip = (reason: string, opts?: { idleFlush?: boolean }) =>
|
||||
Effect.gen(function* () {
|
||||
if (state.enabled) yield* memory.decide({ root, decision: skipped({ sessionID: input.sessionID, reason }) })
|
||||
yield* Effect.promise(() =>
|
||||
MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root,
|
||||
state,
|
||||
phase: "skipped",
|
||||
reason,
|
||||
sessionID: input.sessionID,
|
||||
}),
|
||||
}),
|
||||
)
|
||||
return { root, skipped: true as const, reason, idleFlush: opts?.idleFlush === true }
|
||||
})
|
||||
if (!state.enabled || !state.capture.turnClose) return yield* skip("disabled")
|
||||
const now = Date.now()
|
||||
const view = yield* input.session.readTurn({ sessionID: input.sessionID, window: MESSAGE_WINDOW })
|
||||
if (!view) return yield* skip("no_turn")
|
||||
if (input.bypassInterval && state.stats.lastConsolidatedMessageID === view.lastAssistantID)
|
||||
return yield* skip("no_new_content")
|
||||
const user = view.user
|
||||
const assistant = view.assistant
|
||||
const recent = view.recent
|
||||
const summary = summarize({ user, assistant, max: state.limits.maxSessionLineChars })
|
||||
const diffs = view.diffs
|
||||
const changed = summarizeDiffs(diffs)
|
||||
const durable = hasDurableDiff(diffs)
|
||||
const completed = !input.reason || input.reason === "completed"
|
||||
// Echo = short lookup answered from memory with no file changes. Long recall-assisted answers
|
||||
// (research, investigations) carry new content and must still be digested.
|
||||
const echo = !durable && assistant.length < 1200 && view.recalledMemory
|
||||
const sourced = provenance({ assistant })
|
||||
const session = completed && !echo && Boolean(summary)
|
||||
const prior = session
|
||||
? yield* memory.session({ root, sessionID: input.sessionID, max: state.limits.maxSessionLineChars })
|
||||
: undefined
|
||||
const priorTime = prior?.time ? Date.parse(prior.time) : 0
|
||||
const plan = capturePlan({
|
||||
reason: input.reason,
|
||||
summary,
|
||||
echo,
|
||||
durable,
|
||||
priorTime,
|
||||
now,
|
||||
minIntervalMs: state.capture.minIntervalMs,
|
||||
lastConsolidatedAt: state.stats.lastConsolidatedAt,
|
||||
bypassInterval: input.bypassInterval,
|
||||
autoConsolidate: state.autoConsolidate,
|
||||
})
|
||||
const digestDue = plan.digestDue
|
||||
const typedCall = plan.typedCall
|
||||
|
||||
if (plan.skipReason) return yield* skip(plan.skipReason, plan.idleFlush ? { idleFlush: true } : undefined)
|
||||
yield* Effect.promise(() =>
|
||||
MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({ root, state, phase: "checking", sessionID: input.sessionID }),
|
||||
}),
|
||||
)
|
||||
|
||||
const model =
|
||||
digestDue || typedCall
|
||||
? yield* Effect.gen(function* () {
|
||||
const resolution = yield* input.model.resolve({
|
||||
configured: input.memoryModel,
|
||||
session: view.sessionModel,
|
||||
})
|
||||
if (resolution.fallback) {
|
||||
yield* memory.append({
|
||||
root,
|
||||
text: `memory_model_config reason=${MemoryShared.brief(
|
||||
MemoryRedact.text(resolution.fallback.reason),
|
||||
160,
|
||||
)} fallback=1`,
|
||||
})
|
||||
}
|
||||
return resolution.handle
|
||||
})
|
||||
: undefined
|
||||
const fallback = MemoryRedact.text(
|
||||
fallbackDigest({ prior: prior?.summary, summary, max: state.limits.maxSessionLineChars }),
|
||||
)
|
||||
const safe = MemoryDigest.empty(fallback) ? "" : fallback
|
||||
const digestEffect = digestDue
|
||||
? Effect.gen(function* () {
|
||||
const body = cap(
|
||||
evidence([
|
||||
{ title: "latest_user", body: user },
|
||||
{ title: "latest_assistant", body: assistant || "(no assistant text)" },
|
||||
{ title: "diff_summary", body: changed || "(none)" },
|
||||
{ title: "previous_digest", body: prior?.summary },
|
||||
{ title: "max_characters", body: String(state.limits.maxSessionLineChars) },
|
||||
]),
|
||||
state.limits.maxConsolidationInputBytes,
|
||||
)
|
||||
const result = yield* Effect.tryPromise({
|
||||
try: () =>
|
||||
input.model.run({
|
||||
handle: model!,
|
||||
system: digestPrompt,
|
||||
prompt: body,
|
||||
timeoutMs: state.capture.timeoutMs,
|
||||
signal,
|
||||
}),
|
||||
catch: (error) => error,
|
||||
}).pipe(
|
||||
Effect.map((result) => ({ ok: true as const, result })),
|
||||
Effect.catch((err: unknown) =>
|
||||
Effect.gen(function* () {
|
||||
if (signal.aborted) return { ok: false as const, reason: "cancelled" }
|
||||
const raw = errorReason(err)
|
||||
const reason = MemoryRedact.text(guardReason(raw) ?? raw)
|
||||
yield* fail(reason)
|
||||
yield* memory.append({ root, text: `digest error=${MemoryShared.brief(reason, 160)} fallback=1` })
|
||||
return { ok: false as const, reason }
|
||||
}),
|
||||
),
|
||||
)
|
||||
if (!result.ok) {
|
||||
return {
|
||||
topic: "",
|
||||
summary: safe,
|
||||
tokens: 0,
|
||||
reason: result.reason,
|
||||
}
|
||||
}
|
||||
const parsed = yield* Effect.try({
|
||||
try: () => parseJson(digestSchema, result.result.text),
|
||||
catch: (error) => error,
|
||||
}).pipe(
|
||||
Effect.catch((err: unknown) =>
|
||||
Effect.gen(function* () {
|
||||
const reason = MemoryRedact.text(errorReason(err))
|
||||
yield* fail("digest parse_error")
|
||||
yield* memory.append({
|
||||
root,
|
||||
text: `digest parse_error=${MemoryShared.brief(reason, 160)} fallback=1`,
|
||||
})
|
||||
return undefined
|
||||
}),
|
||||
),
|
||||
)
|
||||
if (!parsed) {
|
||||
return { topic: "", summary: safe, tokens: usage(result.result.usage), reason: "parse_error" }
|
||||
}
|
||||
const parsedDigest = parseDigest(parsed, fallback, state.limits.maxSessionLineChars)
|
||||
return {
|
||||
topic: MemoryRedact.text(parsedDigest.topic),
|
||||
summary: MemoryRedact.text(parsedDigest.summary),
|
||||
tokens: usage(result.result.usage),
|
||||
reason: undefined as string | undefined,
|
||||
}
|
||||
})
|
||||
: Effect.succeed({
|
||||
topic: "",
|
||||
summary: "",
|
||||
tokens: 0,
|
||||
reason: undefined as string | undefined,
|
||||
})
|
||||
const typedEffect = typedCall
|
||||
? Effect.gen(function* () {
|
||||
if (sourced) {
|
||||
return {
|
||||
ops: [] as MemoryOperations.Op[],
|
||||
tokens: 0,
|
||||
fallback: false,
|
||||
reason: undefined as string | undefined,
|
||||
skipped: [
|
||||
{
|
||||
reason: "out_of_scope" as const,
|
||||
text: "Instruction/source provenance answers are not durable project memory.",
|
||||
},
|
||||
] satisfies CaptureSkip[],
|
||||
fallbackOperationCount: 0,
|
||||
}
|
||||
}
|
||||
const existing = yield* typedExisting(memory, root)
|
||||
const items = yield* typedItems(memory, root)
|
||||
const sessions = yield* memory.recent({
|
||||
root,
|
||||
limit: state.limits.maxSessionFiles,
|
||||
max: state.limits.maxSessionLineChars,
|
||||
})
|
||||
const body = cap(
|
||||
evidence([
|
||||
{ title: "close_reason", body: input.reason ?? "completed" },
|
||||
{ title: "latest_user", body: user },
|
||||
{ title: "latest_assistant", body: assistant || "(no assistant text)" },
|
||||
{ title: "diff_summary", body: changed || "(none)" },
|
||||
{ title: "existing_memory", body: existing },
|
||||
{ title: "recent_session_context", body: recent },
|
||||
{
|
||||
title: "recent_memory_digests",
|
||||
body: sessions
|
||||
.map((item) => `${item.file} session=${item.id} ${item.time} :: ${item.summary}`)
|
||||
.join("\n"),
|
||||
},
|
||||
]),
|
||||
state.limits.maxConsolidationInputBytes,
|
||||
)
|
||||
const result = yield* Effect.tryPromise({
|
||||
try: () =>
|
||||
input.model.run({
|
||||
handle: model!,
|
||||
system: typedPrompt,
|
||||
prompt: body,
|
||||
timeoutMs: state.capture.timeoutMs,
|
||||
signal,
|
||||
}),
|
||||
catch: (error) => error,
|
||||
}).pipe(
|
||||
Effect.map((result) => ({ ok: true as const, result })),
|
||||
Effect.catch((err: unknown) =>
|
||||
Effect.gen(function* () {
|
||||
if (signal.aborted) return { ok: false as const, reason: "cancelled" }
|
||||
const raw = errorReason(err)
|
||||
const reason = MemoryRedact.text(guardReason(raw) ?? raw)
|
||||
yield* fail(reason)
|
||||
yield* memory.append({ root, text: `consolidate error=${MemoryShared.brief(reason, 160)}` })
|
||||
return { ok: false as const, reason }
|
||||
}),
|
||||
),
|
||||
)
|
||||
if (!result.ok) {
|
||||
return {
|
||||
ops: [] as MemoryOperations.Op[],
|
||||
tokens: 0,
|
||||
fallback: true,
|
||||
reason: result.reason,
|
||||
skipped: [] as CaptureSkip[],
|
||||
fallbackOperationCount: 0,
|
||||
}
|
||||
}
|
||||
const parsed = yield* Effect.try({
|
||||
try: () => parseJson(typedSchema, result.result.text),
|
||||
catch: (error) => error,
|
||||
}).pipe(
|
||||
Effect.catch((err: unknown) =>
|
||||
Effect.gen(function* () {
|
||||
const reason = MemoryRedact.text(errorReason(err))
|
||||
yield* fail("consolidate parse_error")
|
||||
yield* memory.append({ root, text: `consolidate parse_error=${MemoryShared.brief(reason, 160)}` })
|
||||
return undefined
|
||||
}),
|
||||
),
|
||||
)
|
||||
if (!parsed) {
|
||||
return {
|
||||
ops: [] as MemoryOperations.Op[],
|
||||
tokens: usage(result.result.usage),
|
||||
fallback: true,
|
||||
reason: "parse_error",
|
||||
skipped: [] as CaptureSkip[],
|
||||
fallbackOperationCount: 0,
|
||||
}
|
||||
}
|
||||
const verified = verifySkips({ skipped: parsed.skipped, items })
|
||||
const deduped = duplicateOps({ ops: parseOps(parsed), skipped: verified.skipped, items })
|
||||
return {
|
||||
ops: deduped.ops,
|
||||
tokens: usage(result.result.usage),
|
||||
fallback: false,
|
||||
reason: undefined as string | undefined,
|
||||
skipped: deduped.skipped,
|
||||
fallbackOperationCount: 0,
|
||||
}
|
||||
})
|
||||
: Effect.succeed({
|
||||
ops: [] as MemoryOperations.Op[],
|
||||
tokens: 0,
|
||||
fallback: false,
|
||||
reason: undefined as string | undefined,
|
||||
skipped: [] as CaptureSkip[],
|
||||
fallbackOperationCount: 0,
|
||||
})
|
||||
// Digest and typed consolidation are independent model calls; run them concurrently.
|
||||
const [digest, generated] = yield* Effect.all([digestEffect, typedEffect], { concurrency: 2 })
|
||||
if (signal.aborted) return yield* skip("cancelled")
|
||||
if (digest.summary) {
|
||||
yield* memory.recordSession({
|
||||
root,
|
||||
sessionID: input.sessionID,
|
||||
topic: digest.topic,
|
||||
summary: digest.summary,
|
||||
time: now,
|
||||
tokens: digest.tokens,
|
||||
})
|
||||
}
|
||||
if (digestDue) {
|
||||
yield* memory.decide({
|
||||
root,
|
||||
decision: {
|
||||
kind: "digest",
|
||||
trigger: "turn-close",
|
||||
sessionID: input.sessionID,
|
||||
result: digest.reason ? "fallback" : digest.summary ? "saved" : "skipped",
|
||||
llm: true,
|
||||
parsed: Boolean(digest.summary && !digest.reason),
|
||||
fallback: Boolean(digest.reason),
|
||||
reason: digest.reason,
|
||||
tokens: digest.tokens,
|
||||
operationCount: digest.summary ? 1 : 0,
|
||||
skippedCount: digest.summary ? 0 : 1,
|
||||
summary: digest.reason
|
||||
? `session digest used fallback after ${digest.reason}`
|
||||
: digest.summary
|
||||
? "session digest saved"
|
||||
: "session digest skipped",
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
const ops = mergeOps(generated.ops)
|
||||
.filter((item) => item.action !== "remove")
|
||||
.slice(0, state.capture.maxOpsPerRun)
|
||||
const project =
|
||||
ops.length > 0 ? yield* memory.apply({ root, ops, trigger: "turn-close", tokens: generated.tokens }) : undefined
|
||||
const count = project?.operationCount ?? 0
|
||||
if (typedCall) {
|
||||
yield* memory.decide({
|
||||
root,
|
||||
decision: {
|
||||
kind: "typed",
|
||||
trigger: "turn-close",
|
||||
sessionID: input.sessionID,
|
||||
result: generated.fallback ? "fallback" : count > 0 ? "saved" : "skipped",
|
||||
llm: true,
|
||||
parsed: !generated.fallback,
|
||||
fallback: generated.fallback,
|
||||
reason: generated.reason,
|
||||
tokens: generated.tokens,
|
||||
operationCount: count,
|
||||
skippedCount: generated.skipped.length,
|
||||
fallbackOperationCount: generated.fallbackOperationCount,
|
||||
skipped: generated.skipped,
|
||||
operations: auditOps(ops),
|
||||
files: [...new Set(ops.flatMap((item) => (item.action === "add" && item.file ? [item.file] : [])))],
|
||||
summary: generated.fallback
|
||||
? `typed consolidation skipped after ${generated.reason ?? "model failure"}`
|
||||
: count > 0
|
||||
? `typed consolidation saved ${count} ops`
|
||||
: `typed consolidation skipped ${generated.skipped.length} candidates`,
|
||||
},
|
||||
})
|
||||
}
|
||||
const tokens = digest.tokens + generated.tokens
|
||||
if (!digest.summary && !typedCall && count === 0) return yield* skip("no_ops")
|
||||
if ((digestDue || typedCall || count > 0) && (!typedCall || !generated.fallback)) {
|
||||
yield* memory.commit({
|
||||
root,
|
||||
now,
|
||||
messageID: view.lastAssistantID,
|
||||
tokens,
|
||||
count,
|
||||
digest: Boolean(digest.summary),
|
||||
skipped: generated.skipped,
|
||||
})
|
||||
}
|
||||
const updated = yield* memory.state({ root })
|
||||
const index = project?.index ?? (yield* memory.index({ root }))
|
||||
const detail = typedCall
|
||||
? notice({
|
||||
count,
|
||||
ops,
|
||||
skipped: generated.skipped,
|
||||
tokens: generated.tokens,
|
||||
})
|
||||
: undefined
|
||||
yield* Effect.promise(() =>
|
||||
MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root,
|
||||
state: updated,
|
||||
index,
|
||||
phase: "idle",
|
||||
sessionID: input.sessionID,
|
||||
consolidation: { trigger: "turn-close", operationCount: count, cost: 0, tokens },
|
||||
...(detail ? { detail } : {}),
|
||||
}),
|
||||
}),
|
||||
)
|
||||
return { root, skipped: false as const, operationCount: count, tokens }
|
||||
},
|
||||
// Release the per-root abort controller acquired at the top once the turn settles (any exit path).
|
||||
(effect, input) => effect.pipe(Effect.ensuring(Effect.sync(() => MemoryTimers.release(input.root)))))
|
||||
|
||||
export function report(cause: Cause.Cause<unknown>) {
|
||||
// Brief message only: API errors carry response headers/bodies that would flood the host log.
|
||||
const err = Cause.squash(cause)
|
||||
MemoryLog.warn("memory capture failed", {
|
||||
err: (err instanceof Error ? err.message : String(err)).slice(0, 200),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
export namespace MemoryConfig {
|
||||
export type Model = { providerID: string; modelID: string }
|
||||
|
||||
/** Parse a `providerID/modelID` memory-model override. Returns undefined when blank or malformed
|
||||
* so callers can fall back to the session model. */
|
||||
export function parse(value: string | undefined): Model | undefined {
|
||||
if (!value) return undefined
|
||||
const [providerID, ...rest] = value.split("/")
|
||||
const modelID = rest.join("/")
|
||||
if (!providerID || !modelID) return undefined
|
||||
return { providerID, modelID }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
import { Schema } from "effect"
|
||||
import { MemoryRedact } from "../capture/redact"
|
||||
|
||||
// Typed API error shapes so the SDK / OpenAPI reflect the real contract.
|
||||
export class MemoryApiClientError extends Schema.ErrorClass<MemoryApiClientError>("MemoryApiClientError")(
|
||||
{
|
||||
name: Schema.Literal("MemoryApiClientError"),
|
||||
data: Schema.Struct({ code: Schema.String, message: Schema.String }),
|
||||
},
|
||||
{ httpApiStatus: 400 },
|
||||
) {}
|
||||
|
||||
export class MemoryApiServerError extends Schema.ErrorClass<MemoryApiServerError>("MemoryApiServerError")(
|
||||
{
|
||||
name: Schema.Literal("MemoryApiServerError"),
|
||||
data: Schema.Struct({ code: Schema.String, message: Schema.String }),
|
||||
},
|
||||
{ httpApiStatus: 503 },
|
||||
) {}
|
||||
|
||||
export class MemoryDisabledError extends Schema.TaggedErrorClass<MemoryDisabledError>()("MemoryDisabledError", {
|
||||
reason: Schema.String,
|
||||
cause: Schema.optional(Schema.Unknown),
|
||||
}) {
|
||||
override get message() {
|
||||
return this.reason
|
||||
}
|
||||
}
|
||||
|
||||
export class MemoryInvalidInputError extends Schema.TaggedErrorClass<MemoryInvalidInputError>()(
|
||||
"MemoryInvalidInputError",
|
||||
{
|
||||
reason: Schema.String,
|
||||
cause: Schema.optional(Schema.Unknown),
|
||||
},
|
||||
) {
|
||||
override get message() {
|
||||
return this.reason
|
||||
}
|
||||
}
|
||||
|
||||
export class MemoryStorageError extends Schema.TaggedErrorClass<MemoryStorageError>()("MemoryStorageError", {
|
||||
reason: Schema.String,
|
||||
cause: Schema.optional(Schema.Unknown),
|
||||
}) {
|
||||
override get message() {
|
||||
return this.reason
|
||||
}
|
||||
}
|
||||
|
||||
export class MemoryRootError extends Schema.TaggedErrorClass<MemoryRootError>()("MemoryRootError", {
|
||||
reason: Schema.String,
|
||||
cause: Schema.optional(Schema.Unknown),
|
||||
}) {
|
||||
override get message() {
|
||||
return this.reason
|
||||
}
|
||||
}
|
||||
|
||||
export class MemoryCorruptStateError extends Schema.TaggedErrorClass<MemoryCorruptStateError>()(
|
||||
"MemoryCorruptStateError",
|
||||
{
|
||||
reason: Schema.String,
|
||||
cause: Schema.optional(Schema.Unknown),
|
||||
},
|
||||
) {
|
||||
override get message() {
|
||||
return this.reason
|
||||
}
|
||||
}
|
||||
|
||||
export class MemoryUnknownError extends Schema.TaggedErrorClass<MemoryUnknownError>()("MemoryUnknownError", {
|
||||
reason: Schema.String,
|
||||
cause: Schema.optional(Schema.Unknown),
|
||||
}) {
|
||||
override get message() {
|
||||
return this.reason
|
||||
}
|
||||
}
|
||||
|
||||
export type MemoryError =
|
||||
| MemoryDisabledError
|
||||
| MemoryInvalidInputError
|
||||
| MemoryStorageError
|
||||
| MemoryRootError
|
||||
| MemoryCorruptStateError
|
||||
| MemoryUnknownError
|
||||
|
||||
function reason(err: unknown) {
|
||||
const raw = err instanceof Error ? err.message : String(err)
|
||||
return MemoryRedact.text(raw.replaceAll(/\s+/g, " ").slice(0, 240)) || "unknown memory error"
|
||||
}
|
||||
|
||||
function tag(err: unknown): MemoryError | undefined {
|
||||
if (!err || typeof err !== "object" || !("_tag" in err)) return
|
||||
const value = String(err._tag)
|
||||
if (!value.startsWith("Memory")) return
|
||||
return err as MemoryError
|
||||
}
|
||||
|
||||
export namespace MemoryError {
|
||||
export function from(err: unknown): MemoryError {
|
||||
const known = tag(err)
|
||||
if (known) return known
|
||||
const text = reason(err)
|
||||
const lower = text.toLowerCase()
|
||||
if (lower.includes("memory is disabled")) return new MemoryDisabledError({ reason: text, cause: err })
|
||||
if (
|
||||
/\b(symlink|memory path|memory root|parent is not a directory|path is not a file|path is not a directory)\b/.test(
|
||||
lower,
|
||||
)
|
||||
) {
|
||||
return new MemoryRootError({ reason: text, cause: err })
|
||||
}
|
||||
if (/\b(state\.json|corrupt|recover|parse error|unexpected token)\b/.test(lower)) {
|
||||
return new MemoryCorruptStateError({ reason: text, cause: err })
|
||||
}
|
||||
if (/\b(lock|eacces|eperm|enoent|eio|emfile|enospc)\b/.test(lower)) {
|
||||
return new MemoryStorageError({ reason: text, cause: err })
|
||||
}
|
||||
if (/\b(invalid|schema|zod|section|key|text|source|secret-like|malformed|reject)\b/.test(lower)) {
|
||||
return new MemoryInvalidInputError({ reason: text, cause: err })
|
||||
}
|
||||
return new MemoryUnknownError({ reason: text, cause: err })
|
||||
}
|
||||
|
||||
export function message(err: unknown) {
|
||||
return from(err).message
|
||||
}
|
||||
|
||||
// Map typed taxonomy to HTTP error contract; redaction already applied via .message.
|
||||
export function toHttp(err: MemoryError): MemoryApiClientError | MemoryApiServerError {
|
||||
const msg = err.message
|
||||
switch (err._tag) {
|
||||
case "MemoryDisabledError":
|
||||
return new MemoryApiClientError({
|
||||
name: "MemoryApiClientError",
|
||||
data: { code: "memory_disabled", message: msg },
|
||||
})
|
||||
case "MemoryInvalidInputError":
|
||||
return new MemoryApiClientError({
|
||||
name: "MemoryApiClientError",
|
||||
data: { code: "memory_invalid_input", message: msg },
|
||||
})
|
||||
case "MemoryRootError":
|
||||
return new MemoryApiClientError({
|
||||
name: "MemoryApiClientError",
|
||||
data: { code: "memory_root_error", message: msg },
|
||||
})
|
||||
case "MemoryStorageError":
|
||||
return new MemoryApiServerError({
|
||||
name: "MemoryApiServerError",
|
||||
data: { code: "memory_storage_error", message: msg },
|
||||
})
|
||||
case "MemoryCorruptStateError":
|
||||
return new MemoryApiServerError({
|
||||
name: "MemoryApiServerError",
|
||||
data: { code: "memory_corrupt_state", message: msg },
|
||||
})
|
||||
default:
|
||||
return new MemoryApiServerError({ name: "MemoryApiServerError", data: { code: "memory_error", message: msg } })
|
||||
}
|
||||
}
|
||||
|
||||
export function toToolOutput(err: unknown, action: string) {
|
||||
return `Kilo memory ${action} failed: ${message(err)}`
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
import { Schema } from "effect"
|
||||
import type { MemorySchema } from "../schema"
|
||||
import { MemoryLog } from "./log"
|
||||
|
||||
export namespace MemoryEvents {
|
||||
const Metric = Schema.Struct({
|
||||
bytes: Schema.Number,
|
||||
estimatedTokens: Schema.Number,
|
||||
truncated: Schema.Boolean,
|
||||
updatedAt: Schema.optional(Schema.Number),
|
||||
})
|
||||
|
||||
const Phase = Schema.Literals(["idle", "checking", "injecting", "updating", "skipped", "error"])
|
||||
const Trigger = Schema.Literals(["explicit", "turn-close", "rebuild"])
|
||||
|
||||
const Consolidation = Schema.Struct({
|
||||
trigger: Trigger,
|
||||
operationCount: Schema.Number,
|
||||
cost: Schema.Number,
|
||||
tokens: Schema.Number,
|
||||
})
|
||||
|
||||
const Detail = Schema.Struct({
|
||||
type: Schema.Literals(["saved", "skipped", "recalled"]),
|
||||
message: Schema.String,
|
||||
reason: Schema.optional(Schema.String),
|
||||
duplicateOf: Schema.optional(Schema.String),
|
||||
tokens: Schema.optional(Schema.Number),
|
||||
operationCount: Schema.optional(Schema.Number),
|
||||
skippedCount: Schema.optional(Schema.Number),
|
||||
sources: Schema.optional(Schema.Array(Schema.String)),
|
||||
files: Schema.optional(Schema.Array(Schema.String)),
|
||||
})
|
||||
|
||||
export const Payload = Schema.Struct({
|
||||
directory: Schema.String,
|
||||
sessionID: Schema.optional(Schema.String),
|
||||
enabled: Schema.Boolean,
|
||||
state: Phase,
|
||||
reason: Schema.optional(Schema.String),
|
||||
project: Metric,
|
||||
consolidation: Schema.optional(Consolidation),
|
||||
detail: Schema.optional(Detail),
|
||||
})
|
||||
|
||||
export type Phase = Schema.Schema.Type<typeof Phase>
|
||||
export type Trigger = Schema.Schema.Type<typeof Trigger>
|
||||
export type Status = Schema.Schema.Type<typeof Payload>
|
||||
export type Index = { bytes: number; tokens: number; truncated: boolean }
|
||||
|
||||
export type Inspect = {
|
||||
root: string
|
||||
state: MemorySchema.State
|
||||
sources: {
|
||||
project: string
|
||||
environment: string
|
||||
corrections: string
|
||||
}
|
||||
index: string
|
||||
changes: string
|
||||
}
|
||||
|
||||
function metric(index?: Index, updatedAt?: number | null) {
|
||||
return {
|
||||
bytes: index?.bytes ?? 0,
|
||||
estimatedTokens: index?.tokens ?? 0,
|
||||
truncated: index?.truncated ?? false,
|
||||
...(updatedAt ? { updatedAt } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
function latest(...items: (number | null)[]) {
|
||||
const values = items.filter((item): item is number => typeof item === "number" && Number.isFinite(item))
|
||||
return values.length ? Math.max(...values) : undefined
|
||||
}
|
||||
|
||||
export function status(input: {
|
||||
root: string
|
||||
state: MemorySchema.State
|
||||
index?: Index
|
||||
phase?: Phase
|
||||
reason?: string
|
||||
sessionID?: string
|
||||
consolidation?: Status["consolidation"]
|
||||
detail?: Status["detail"]
|
||||
}): Status {
|
||||
const updated = latest(input.state.stats.lastInjectedAt, input.state.stats.lastConsolidatedAt)
|
||||
const current = metric(input.index, updated)
|
||||
return {
|
||||
directory: input.root,
|
||||
...(input.sessionID ? { sessionID: input.sessionID } : {}),
|
||||
enabled: input.state.enabled,
|
||||
state: input.phase ?? "idle",
|
||||
...(input.reason ? { reason: input.reason } : {}),
|
||||
project: current,
|
||||
...(input.consolidation ? { consolidation: input.consolidation } : {}),
|
||||
...(input.detail ? { detail: input.detail } : {}),
|
||||
}
|
||||
}
|
||||
|
||||
export type Event = "status" | "updated" | "error"
|
||||
export type Sink = (input: { event?: Event; payload: Status }) => Promise<void> | void
|
||||
|
||||
// Best-effort: opencode wires this to its Bus at bootstrap; defaults to a no-op so the package
|
||||
// never reaches into the host event system on its own.
|
||||
let sink: Sink = () => {}
|
||||
|
||||
export function setSink(next: Sink) {
|
||||
sink = next
|
||||
}
|
||||
|
||||
export async function publish(input: { event?: Event; payload: Status }) {
|
||||
// Event wiring is best-effort: a failing host sink must not fail a memory op that already
|
||||
// persisted, so swallow and log instead of propagating to callers.
|
||||
try {
|
||||
await sink(input)
|
||||
} catch (err) {
|
||||
MemoryLog.warn("memory event publish failed", {
|
||||
err: (err instanceof Error ? err.message : String(err)).slice(0, 200),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,340 @@
|
||||
import { Memory } from "../memory"
|
||||
import type { MemoryOperations } from "../capture/ops"
|
||||
import { MemorySchema } from "../schema"
|
||||
import { MemoryFiles } from "../storage/store"
|
||||
import { MemoryToken } from "../recall/token"
|
||||
import { MemoryEvents } from "./events"
|
||||
import { MemoryPaths } from "./paths"
|
||||
import { MemoryTimers } from "./timers"
|
||||
import { MemoryDisabledError } from "./errors"
|
||||
|
||||
/** Context-bound Kilo adapter over the root-bound package primitives. Prefer ctx inputs at runtime edges. */
|
||||
export namespace KiloMemory {
|
||||
export type Input =
|
||||
| {
|
||||
root: string
|
||||
sessionID?: string
|
||||
record?: boolean
|
||||
}
|
||||
| {
|
||||
ctx: MemoryPaths.Ctx
|
||||
sessionID?: string
|
||||
record?: boolean
|
||||
}
|
||||
|
||||
export type Block = Memory.Block
|
||||
|
||||
function root(input: Input) {
|
||||
return "root" in input ? input.root : MemoryPaths.root(input)
|
||||
}
|
||||
|
||||
async function noop(dir: string): Promise<MemoryOperations.Result> {
|
||||
const text = await MemoryFiles.readIndex(dir)
|
||||
const index = {
|
||||
text,
|
||||
bytes: Buffer.byteLength(text),
|
||||
tokens: MemoryToken.estimate(text),
|
||||
truncated: false,
|
||||
}
|
||||
return { operationCount: 0, added: 0, removed: 0, skipped: [], index }
|
||||
}
|
||||
|
||||
async function requireEnabled(dir: string) {
|
||||
const state = await MemoryFiles.readState(dir)
|
||||
if (state.enabled) return state
|
||||
throw new MemoryDisabledError({ reason: "project memory is disabled" })
|
||||
}
|
||||
|
||||
export async function prepare(input: Input) {
|
||||
return root(input)
|
||||
}
|
||||
|
||||
export async function status(input: Input) {
|
||||
return Memory.status({ root: await prepare(input) })
|
||||
}
|
||||
|
||||
export async function enable(input: Input) {
|
||||
const dir = await prepare(input)
|
||||
const id = "ctx" in input ? MemoryPaths.identity({ ctx: input.ctx }) : undefined
|
||||
const result = await Memory.enable({ root: dir, id })
|
||||
await MemoryEvents.publish({
|
||||
event: "updated",
|
||||
payload: MemoryEvents.status({
|
||||
root: dir,
|
||||
state: result.state,
|
||||
index: result.index,
|
||||
phase: "idle",
|
||||
consolidation: { trigger: "rebuild", operationCount: 0, cost: 0, tokens: result.index.tokens },
|
||||
}),
|
||||
})
|
||||
return result
|
||||
}
|
||||
|
||||
export async function disable(input: Input) {
|
||||
const dir = await prepare(input)
|
||||
MemoryTimers.clear(dir)
|
||||
const result = await Memory.disable({ root: dir })
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({ root: result.root, state: result.state, phase: "idle" }),
|
||||
})
|
||||
return result
|
||||
}
|
||||
|
||||
export async function show(input: Input) {
|
||||
return Memory.show({ root: await prepare(input) })
|
||||
}
|
||||
|
||||
export async function rebuild(input: Input) {
|
||||
const dir = await prepare(input)
|
||||
const state = await MemoryFiles.readState(dir)
|
||||
if (!state.enabled) {
|
||||
const index = (await noop(dir)).index
|
||||
return { root: dir, state, index }
|
||||
}
|
||||
const result = await Memory.rebuild({ root: dir })
|
||||
await MemoryEvents.publish({
|
||||
event: "updated",
|
||||
payload: MemoryEvents.status({
|
||||
root: result.root,
|
||||
state: result.state,
|
||||
index: result.index,
|
||||
phase: "idle",
|
||||
consolidation: { trigger: "rebuild", operationCount: 0, cost: 0, tokens: result.index.tokens },
|
||||
}),
|
||||
})
|
||||
return result
|
||||
}
|
||||
|
||||
export async function configure(
|
||||
input: Input & {
|
||||
settings: Partial<Pick<MemorySchema.State, "autoConsolidate">>
|
||||
},
|
||||
) {
|
||||
const result = await Memory.configure({ root: await prepare(input), settings: input.settings })
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({ root: result.root, state: result.state, phase: "idle" }),
|
||||
})
|
||||
return result
|
||||
}
|
||||
|
||||
export async function context(input: Input) {
|
||||
const result = await Memory.context({
|
||||
root: await prepare(input),
|
||||
sessionID: input.sessionID,
|
||||
record: input.record,
|
||||
})
|
||||
if (result.recorded) {
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root: result.root,
|
||||
state: result.state,
|
||||
index: result.index,
|
||||
phase: "injecting",
|
||||
sessionID: input.sessionID,
|
||||
}),
|
||||
})
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
export async function toolEnabled(input: Input) {
|
||||
return Memory.toolEnabled({ root: "ctx" in input ? await prepare(input) : root(input) })
|
||||
}
|
||||
|
||||
async function publish(input: {
|
||||
output: Memory.Apply
|
||||
sessionID?: string
|
||||
trigger?: Memory.Trigger
|
||||
cost?: number
|
||||
tokens?: number
|
||||
}) {
|
||||
await MemoryEvents.publish({
|
||||
event: "updated",
|
||||
payload: MemoryEvents.status({
|
||||
root: input.output.root,
|
||||
state: input.output.state,
|
||||
index: input.output.result.index,
|
||||
phase: "updating",
|
||||
sessionID: input.sessionID,
|
||||
consolidation: {
|
||||
trigger: input.trigger ?? "explicit",
|
||||
operationCount: input.output.result.operationCount,
|
||||
cost: input.cost ?? 0,
|
||||
tokens: input.tokens ?? 0,
|
||||
},
|
||||
...(input.output.detail ? { detail: input.output.detail } : {}),
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
export async function apply(
|
||||
input: Input & {
|
||||
ops: MemoryOperations.Op[]
|
||||
trigger?: Memory.Trigger
|
||||
cost?: number
|
||||
tokens?: number
|
||||
},
|
||||
) {
|
||||
const dir = await prepare(input)
|
||||
await requireEnabled(dir)
|
||||
const output = await Memory.apply({
|
||||
root: dir,
|
||||
ops: input.ops,
|
||||
trigger: input.trigger,
|
||||
sessionID: input.sessionID,
|
||||
tokens: input.tokens,
|
||||
})
|
||||
await publish({
|
||||
output,
|
||||
sessionID: input.sessionID,
|
||||
trigger: input.trigger,
|
||||
cost: input.cost,
|
||||
tokens: input.tokens,
|
||||
})
|
||||
return output.result
|
||||
}
|
||||
|
||||
export async function forget(input: Input & { query: string }) {
|
||||
const dir = await prepare(input)
|
||||
await requireEnabled(dir)
|
||||
const output = await Memory.forget({ root: dir, query: input.query, sessionID: input.sessionID })
|
||||
await publish({ output, sessionID: input.sessionID })
|
||||
return output.result
|
||||
}
|
||||
|
||||
export async function remember(
|
||||
input: Input & {
|
||||
text: string
|
||||
key?: string
|
||||
file?: MemorySchema.Source
|
||||
section?: string
|
||||
},
|
||||
) {
|
||||
const dir = await prepare(input)
|
||||
await requireEnabled(dir)
|
||||
const output = await Memory.remember({
|
||||
root: dir,
|
||||
text: input.text,
|
||||
key: input.key,
|
||||
file: input.file,
|
||||
section: input.section,
|
||||
sessionID: input.sessionID,
|
||||
})
|
||||
await publish({ output, sessionID: input.sessionID })
|
||||
return output.result
|
||||
}
|
||||
|
||||
export async function correct(input: Input & { text: string; key?: string }) {
|
||||
const dir = await prepare(input)
|
||||
await requireEnabled(dir)
|
||||
const output = await Memory.correct({
|
||||
root: dir,
|
||||
text: input.text,
|
||||
key: input.key,
|
||||
sessionID: input.sessionID,
|
||||
})
|
||||
await publish({ output, sessionID: input.sessionID })
|
||||
return output.result
|
||||
}
|
||||
|
||||
export async function purge(input: Input) {
|
||||
const dir = root(input)
|
||||
MemoryTimers.clear(dir)
|
||||
const result = await Memory.purge({ root: dir })
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root: result.root,
|
||||
state: result.state,
|
||||
phase: "idle",
|
||||
reason: result.purged ? "purged" : "missing",
|
||||
}),
|
||||
})
|
||||
return { root: result.root, purged: result.purged }
|
||||
}
|
||||
|
||||
export async function recall(input: Input & { query: string; sessionID?: string }) {
|
||||
const output = await Memory.recall({ root: await prepare(input), query: input.query, sessionID: input.sessionID })
|
||||
if (!output.state.enabled) return
|
||||
if (!output.result) {
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root: output.root,
|
||||
state: output.state,
|
||||
phase: "skipped",
|
||||
sessionID: input.sessionID,
|
||||
detail: {
|
||||
type: "skipped",
|
||||
message: "Memory skipped · no recall matches",
|
||||
reason: "no_matches",
|
||||
skippedCount: 1,
|
||||
},
|
||||
}),
|
||||
})
|
||||
return
|
||||
}
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root: output.root,
|
||||
state: output.state,
|
||||
phase: "injecting",
|
||||
sessionID: input.sessionID,
|
||||
detail: {
|
||||
type: "recalled",
|
||||
message: `Memory recalled · ${output.result.hits.length} ${output.result.hits.length === 1 ? "item" : "items"}`,
|
||||
tokens: output.result.tokens,
|
||||
operationCount: output.result.hits.length,
|
||||
sources: output.files,
|
||||
files: output.files,
|
||||
},
|
||||
}),
|
||||
})
|
||||
return { root: output.root, ...output.result }
|
||||
}
|
||||
|
||||
export async function recordSession(
|
||||
input: Input & { sessionID: string; topic?: string; summary: string; time?: number; tokens?: number },
|
||||
) {
|
||||
const output = await Memory.recordSession({
|
||||
root: await prepare(input),
|
||||
sessionID: input.sessionID,
|
||||
topic: input.topic,
|
||||
summary: input.summary,
|
||||
time: input.time,
|
||||
tokens: input.tokens,
|
||||
})
|
||||
if (output.skipped) {
|
||||
await MemoryEvents.publish({
|
||||
event: "status",
|
||||
payload: MemoryEvents.status({
|
||||
root: output.root,
|
||||
state: output.state,
|
||||
phase: "skipped",
|
||||
reason: output.reason,
|
||||
sessionID: input.sessionID,
|
||||
}),
|
||||
})
|
||||
return { skipped: true, reason: output.reason }
|
||||
}
|
||||
await MemoryEvents.publish({
|
||||
event: "updated",
|
||||
payload: MemoryEvents.status({
|
||||
root: output.root,
|
||||
state: output.state,
|
||||
index: output.index,
|
||||
phase: "updating",
|
||||
sessionID: input.sessionID,
|
||||
consolidation: { trigger: "turn-close", operationCount: 0, cost: 0, tokens: input.tokens ?? 0 },
|
||||
}),
|
||||
})
|
||||
return { skipped: false, index: output.index }
|
||||
}
|
||||
}
|
||||
|
||||
export { MemoryEvents } from "./events"
|
||||
export { MemoryPaths } from "./paths"
|
||||
@@ -0,0 +1,16 @@
|
||||
/** Injectable instance-context binder. The Effect service bridges async package calls through
|
||||
* this binder so host-provided context (e.g. opencode's per-instance ALS) survives the await.
|
||||
* Defaults to identity so the package stays runnable without a host. */
|
||||
export namespace MemoryInstance {
|
||||
export type Binder = <A>(fn: () => Promise<A>) => () => Promise<A>
|
||||
|
||||
let binder: Binder = (fn) => fn
|
||||
|
||||
export function setBinder(next: Binder) {
|
||||
binder = next
|
||||
}
|
||||
|
||||
export function bind<A>(fn: () => Promise<A>): () => Promise<A> {
|
||||
return binder(fn)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
/** Injectable diagnostic logger. Opencode wires this to its structured logger at bootstrap;
|
||||
* the package defaults to a no-op so it never reaches into the host runtime on its own. */
|
||||
export namespace MemoryLog {
|
||||
export type Fn = (message: string, meta?: Record<string, unknown>) => void
|
||||
|
||||
let warnFn: Fn = () => {}
|
||||
|
||||
export function setWarn(fn: Fn) {
|
||||
warnFn = fn
|
||||
}
|
||||
|
||||
export function warn(message: string, meta?: Record<string, unknown>) {
|
||||
warnFn(message, meta)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
import { homedir } from "os"
|
||||
import path from "path"
|
||||
import { MemoryPaths as Core } from "../storage/paths"
|
||||
|
||||
/** Context-bound paths over the pure core. The host (home/config dirs) is injected at bootstrap so
|
||||
* the package does not hard-code the opencode global directory; defaults to `~/.kilo`. */
|
||||
export namespace MemoryPaths {
|
||||
export type Ctx = Core.Ctx
|
||||
export type Files = Core.Files
|
||||
export type Identity = Core.Identity
|
||||
export type Host = Core.Host
|
||||
|
||||
// A provider (not a snapshot) so hosts that resolve home/config dynamically — e.g. from env at
|
||||
// call time — are reflected on every `root` call.
|
||||
let host: () => Host = () => ({ home: homedir(), config: path.join(homedir(), ".kilo") })
|
||||
|
||||
export function configure(next: () => Host) {
|
||||
host = next
|
||||
}
|
||||
|
||||
export function identity(input: { ctx: Ctx }): Identity {
|
||||
return Core.identity(input)
|
||||
}
|
||||
|
||||
export function root(input: { ctx: Ctx }) {
|
||||
const { home, config } = host()
|
||||
return Core.root({ ctx: input.ctx, home, config })
|
||||
}
|
||||
|
||||
export const files = Core.files
|
||||
export const source = Core.source
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
import type { Effect } from "effect"
|
||||
import type { CaptureDiff } from "../capture/diff"
|
||||
import type { MemoryError } from "./errors"
|
||||
|
||||
/** Runtime ports the capture pipeline depends on. The host (opencode) implements these against its
|
||||
* session store and LLM provider; the package orchestration stays free of `ai`/provider types by
|
||||
* treating the resolved model as an opaque handle and consuming pre-extracted turn primitives. */
|
||||
export namespace MemoryPorts {
|
||||
export type ModelRef = { providerID: string; modelID: string }
|
||||
|
||||
/** Pre-extracted view of the latest turn. All transcript/message-shape handling happens host-side
|
||||
* so the orchestrator never touches the host's message model. */
|
||||
export type TurnView = {
|
||||
user: string
|
||||
assistant: string
|
||||
recent: string
|
||||
lastAssistantID: string
|
||||
sessionModel: ModelRef
|
||||
/** True when the turn was answered from targeted recall (digesting it would echo memory back). */
|
||||
recalledMemory: boolean
|
||||
diffs: CaptureDiff[]
|
||||
}
|
||||
|
||||
export interface SessionPort {
|
||||
readonly readTurn: (input: {
|
||||
sessionID: string
|
||||
window: number
|
||||
}) => Effect.Effect<TurnView | undefined, MemoryError>
|
||||
readonly get: (input: { sessionID: string }) => Effect.Effect<{ parentID?: string } | undefined, MemoryError>
|
||||
}
|
||||
|
||||
/** Opaque resolved-model handle. Carries provider/language/options on the host side; the package
|
||||
* only passes it back to `run`. */
|
||||
export type ModelHandle = unknown
|
||||
|
||||
export type ModelResolution = { handle: ModelHandle; fallback?: { reason: string } }
|
||||
|
||||
export interface ModelPort {
|
||||
readonly resolve: (input: { configured?: string; session: ModelRef }) => Effect.Effect<ModelResolution, MemoryError>
|
||||
readonly run: (input: {
|
||||
handle: ModelHandle
|
||||
system: string
|
||||
prompt: string
|
||||
timeoutMs: number
|
||||
signal?: AbortSignal
|
||||
}) => Promise<{ text: string; usage: unknown }>
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
import { Context, Effect, Layer, Semaphore } from "effect"
|
||||
import { skipLine, type CaptureSkip } from "../capture/capture"
|
||||
import type { Memory } from "../memory"
|
||||
import type { MemoryOperations } from "../capture/ops"
|
||||
import { MemoryRecall } from "../recall/recall"
|
||||
import { MemorySchema } from "../schema"
|
||||
import { MemoryFiles } from "../storage/store"
|
||||
import { MemoryToken } from "../recall/token"
|
||||
import { KiloMemory } from "./index"
|
||||
import { MemoryInstance } from "./instance"
|
||||
import { MemoryError, type MemoryError as Failure } from "./errors"
|
||||
|
||||
type SessionID = string
|
||||
|
||||
const IDLE_SETTLE_MS = 30_000
|
||||
|
||||
type ConfigureInput = KiloMemory.Input & {
|
||||
settings: Partial<Pick<MemorySchema.State, "autoConsolidate">>
|
||||
}
|
||||
|
||||
type ApplyInput = KiloMemory.Input & {
|
||||
ops: MemoryOperations.Op[]
|
||||
trigger?: Memory.Trigger
|
||||
cost?: number
|
||||
tokens?: number
|
||||
}
|
||||
|
||||
type RememberInput = KiloMemory.Input & {
|
||||
text: string
|
||||
key?: string
|
||||
file?: MemorySchema.Source
|
||||
section?: string
|
||||
}
|
||||
|
||||
type CorrectInput = KiloMemory.Input & {
|
||||
text: string
|
||||
key?: string
|
||||
}
|
||||
|
||||
type ForgetInput = KiloMemory.Input & {
|
||||
query: string
|
||||
}
|
||||
|
||||
type RecallInput = KiloMemory.Input & {
|
||||
query: string
|
||||
sessionID?: string
|
||||
}
|
||||
|
||||
type SearchInput = Parameters<typeof MemoryRecall.search>[0]
|
||||
|
||||
type RecordInput = KiloMemory.Input & {
|
||||
sessionID: string
|
||||
topic?: string
|
||||
summary: string
|
||||
time?: number
|
||||
tokens?: number
|
||||
}
|
||||
|
||||
type DecideInput = {
|
||||
root: string
|
||||
decision: MemoryFiles.Decision
|
||||
}
|
||||
|
||||
type ReadSourceInput = {
|
||||
root: string
|
||||
file: MemorySchema.Source
|
||||
}
|
||||
|
||||
type RootInput = {
|
||||
root: string
|
||||
}
|
||||
|
||||
type SessionInput = RootInput & {
|
||||
sessionID: string
|
||||
max: number
|
||||
}
|
||||
|
||||
type RecentInput = RootInput & {
|
||||
limit: number
|
||||
max: number
|
||||
}
|
||||
|
||||
type AppendInput = RootInput & {
|
||||
text: string
|
||||
}
|
||||
|
||||
type Sources = Record<MemorySchema.Source, string>
|
||||
|
||||
type Index = {
|
||||
bytes: number
|
||||
tokens: number
|
||||
truncated: false
|
||||
}
|
||||
|
||||
type CommitInput = RootInput & {
|
||||
now: number
|
||||
messageID: string
|
||||
tokens: number
|
||||
count: number
|
||||
digest: boolean
|
||||
skipped: CaptureSkip[]
|
||||
cost?: number
|
||||
}
|
||||
|
||||
function bridge<A>(fn: () => Promise<A>) {
|
||||
return Effect.tryPromise({
|
||||
try: MemoryInstance.bind(fn),
|
||||
catch: MemoryError.from,
|
||||
})
|
||||
}
|
||||
|
||||
export namespace MemoryService {
|
||||
export type Timing = { settleMs: number }
|
||||
|
||||
export interface Interface {
|
||||
readonly prepare: (input: KiloMemory.Input) => Effect.Effect<string, Failure>
|
||||
readonly status: (input: KiloMemory.Input) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.status>>, Failure>
|
||||
readonly show: (input: KiloMemory.Input) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.show>>, Failure>
|
||||
readonly enable: (input: KiloMemory.Input) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.enable>>, Failure>
|
||||
readonly disable: (
|
||||
input: KiloMemory.Input,
|
||||
) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.disable>>, Failure>
|
||||
readonly rebuild: (
|
||||
input: KiloMemory.Input,
|
||||
) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.rebuild>>, Failure>
|
||||
readonly configure: (
|
||||
input: ConfigureInput,
|
||||
) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.configure>>, Failure>
|
||||
readonly apply: (input: ApplyInput) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.apply>>, Failure>
|
||||
readonly remember: (input: RememberInput) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.remember>>, Failure>
|
||||
readonly correct: (input: CorrectInput) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.correct>>, Failure>
|
||||
readonly forget: (input: ForgetInput) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.forget>>, Failure>
|
||||
readonly purge: (input: KiloMemory.Input) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.purge>>, Failure>
|
||||
readonly recall: (input: RecallInput) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.recall>>, Failure>
|
||||
readonly search: (input: SearchInput) => Effect.Effect<Awaited<ReturnType<typeof MemoryRecall.search>>, Failure>
|
||||
readonly recordSession: (
|
||||
input: RecordInput,
|
||||
) => Effect.Effect<Awaited<ReturnType<typeof KiloMemory.recordSession>>, Failure>
|
||||
readonly state: (input: RootInput) => Effect.Effect<MemorySchema.State, Failure>
|
||||
readonly session: (
|
||||
input: SessionInput,
|
||||
) => Effect.Effect<Awaited<ReturnType<typeof MemoryFiles.readSession>>, Failure>
|
||||
readonly sources: (input: RootInput) => Effect.Effect<Sources, Failure>
|
||||
readonly recent: (
|
||||
input: RecentInput,
|
||||
) => Effect.Effect<Awaited<ReturnType<typeof MemoryFiles.recentSessions>>, Failure>
|
||||
readonly append: (input: AppendInput) => Effect.Effect<void, Failure>
|
||||
readonly index: (input: RootInput) => Effect.Effect<Index, Failure>
|
||||
readonly commit: (input: CommitInput) => Effect.Effect<void, Failure>
|
||||
readonly decide: (input: DecideInput) => Effect.Effect<void, Failure>
|
||||
readonly readSource: (input: ReadSourceInput) => Effect.Effect<string, Failure>
|
||||
readonly turnLock: (sessionID: SessionID) => Semaphore.Semaphore
|
||||
readonly dropLock: (sessionID: SessionID) => void
|
||||
readonly idleSettle: () => number
|
||||
readonly setIdleSettle: (ms: number) => Timing
|
||||
}
|
||||
|
||||
export class Service extends Context.Service<Service, Interface>()("@kilocode/MemoryService") {}
|
||||
|
||||
export function make() {
|
||||
const locks = new Map<SessionID, { sema: Semaphore.Semaphore; holders: number }>()
|
||||
let settle = IDLE_SETTLE_MS
|
||||
return Service.of({
|
||||
prepare: (input) => bridge(() => KiloMemory.prepare(input)),
|
||||
status: (input) => bridge(() => KiloMemory.status(input)),
|
||||
show: (input) => bridge(() => KiloMemory.show(input)),
|
||||
enable: (input) => bridge(() => KiloMemory.enable(input)),
|
||||
disable: (input) => bridge(() => KiloMemory.disable(input)),
|
||||
rebuild: (input) => bridge(() => KiloMemory.rebuild(input)),
|
||||
configure: (input) => bridge(() => KiloMemory.configure(input)),
|
||||
apply: (input) => bridge(() => KiloMemory.apply(input)),
|
||||
remember: (input) => bridge(() => KiloMemory.remember(input)),
|
||||
correct: (input) => bridge(() => KiloMemory.correct(input)),
|
||||
forget: (input) => bridge(() => KiloMemory.forget(input)),
|
||||
purge: (input) => bridge(() => KiloMemory.purge(input)),
|
||||
recall: (input) => bridge(() => KiloMemory.recall(input)),
|
||||
search: (input) => bridge(() => MemoryRecall.search(input)),
|
||||
recordSession: (input) => bridge(() => KiloMemory.recordSession(input)),
|
||||
state: (input) => bridge(() => MemoryFiles.readState(input.root)),
|
||||
session: (input) =>
|
||||
bridge(() => MemoryFiles.readSession(input.root, { sessionID: input.sessionID, max: input.max })),
|
||||
sources: (input) =>
|
||||
bridge(async () => {
|
||||
const entries = await Promise.all(
|
||||
MemorySchema.Sources.map(async (file) => [file, await MemoryFiles.readSource(input.root, file)] as const),
|
||||
)
|
||||
return Object.fromEntries(entries) as Sources
|
||||
}),
|
||||
recent: (input) => bridge(() => MemoryFiles.recentSessions(input.root, input.limit, input.max)),
|
||||
append: (input) => bridge(() => MemoryFiles.append(input.root, input.text)),
|
||||
index: (input) =>
|
||||
bridge(async () => {
|
||||
const text = await MemoryFiles.readIndex(input.root)
|
||||
return { bytes: Buffer.byteLength(text), tokens: MemoryToken.estimate(text), truncated: false }
|
||||
}),
|
||||
commit: (input) =>
|
||||
bridge(() =>
|
||||
MemoryFiles.queue(input.root, async () => {
|
||||
const state = await MemoryFiles.readState(input.root)
|
||||
await MemoryFiles.writeState(input.root, {
|
||||
...state,
|
||||
stats: {
|
||||
...state.stats,
|
||||
lastConsolidatedAt: input.now,
|
||||
lastConsolidatedMessageID: input.messageID,
|
||||
lastConsolidationCost: input.cost ?? state.stats.lastConsolidationCost,
|
||||
lastConsolidationTokens: input.tokens,
|
||||
lastOperationCount: input.count,
|
||||
},
|
||||
})
|
||||
const skip = skipLine(input.skipped)
|
||||
await MemoryFiles.append(
|
||||
input.root,
|
||||
[
|
||||
`consolidate trigger=turn-close digest=${input.digest ? 1 : 0} ops=${input.count} tokens=${input.tokens}`,
|
||||
skip,
|
||||
]
|
||||
.filter(Boolean)
|
||||
.join(" "),
|
||||
)
|
||||
}),
|
||||
),
|
||||
decide: (input) => bridge(() => MemoryFiles.decide(input.root, input.decision)),
|
||||
readSource: (input) => bridge(() => MemoryFiles.readSource(input.root, input.file)),
|
||||
// Ref-counted so every acquirer — in-flight or queued behind `withPermits` — shares one
|
||||
// semaphore. Each call must be balanced by exactly one `dropLock`.
|
||||
turnLock: (sessionID) => {
|
||||
const prior = locks.get(sessionID)
|
||||
if (prior) {
|
||||
prior.holders += 1
|
||||
return prior.sema
|
||||
}
|
||||
const sema = Semaphore.makeUnsafe(1)
|
||||
locks.set(sessionID, { sema, holders: 1 })
|
||||
return sema
|
||||
},
|
||||
// Release one holder. The entry is dropped only when the last holder leaves, so a queued
|
||||
// close() can never be handed a different semaphore than the peer it is waiting on — while the
|
||||
// map still stops growing unbounded in a long-lived shared backend.
|
||||
dropLock: (sessionID) => {
|
||||
const item = locks.get(sessionID)
|
||||
if (!item) return
|
||||
item.holders -= 1
|
||||
if (item.holders <= 0) locks.delete(sessionID)
|
||||
},
|
||||
idleSettle: () => settle,
|
||||
setIdleSettle: (ms) => {
|
||||
const prev = { settleMs: settle }
|
||||
settle = Math.max(1, ms)
|
||||
return prev
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
export const layer = Layer.sync(Service)(make)
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
type SessionID = string
|
||||
|
||||
export namespace MemoryTimers {
|
||||
const pending = new Map<SessionID, { root: string; timer: ReturnType<typeof setTimeout> }>()
|
||||
const signals = new Map<string, { ctl: AbortController; active: number }>()
|
||||
|
||||
export function cancel(sessionID: SessionID) {
|
||||
const item = pending.get(sessionID)
|
||||
if (!item) return
|
||||
clearTimeout(item.timer)
|
||||
pending.delete(sessionID)
|
||||
}
|
||||
|
||||
export function clear(root: string) {
|
||||
for (const [sessionID, item] of pending) {
|
||||
if (item.root !== root) continue
|
||||
clearTimeout(item.timer)
|
||||
pending.delete(sessionID)
|
||||
}
|
||||
signals.get(root)?.ctl.abort()
|
||||
signals.delete(root)
|
||||
}
|
||||
|
||||
// One AbortController per root, shared across concurrent captures and ref-counted so it is dropped
|
||||
// once the last in-flight capture for the root settles (see `release`). Without this the map grows
|
||||
// for every distinct root a long-lived shared backend ever touches. disable/purge still force-abort
|
||||
// via `clear`; `release` tolerates an already-cleared entry.
|
||||
export function signal(root: string) {
|
||||
const prior = signals.get(root)
|
||||
if (prior) {
|
||||
prior.active += 1
|
||||
return prior.ctl.signal
|
||||
}
|
||||
const ctl = new AbortController()
|
||||
signals.set(root, { ctl, active: 1 })
|
||||
return ctl.signal
|
||||
}
|
||||
|
||||
export function release(root: string) {
|
||||
const item = signals.get(root)
|
||||
if (!item) return
|
||||
item.active -= 1
|
||||
if (item.active <= 0) signals.delete(root)
|
||||
}
|
||||
|
||||
export function done(sessionID: SessionID) {
|
||||
pending.delete(sessionID)
|
||||
}
|
||||
|
||||
export function set(sessionID: SessionID, root: string, timer: ReturnType<typeof setTimeout>) {
|
||||
cancel(sessionID)
|
||||
timer.unref?.()
|
||||
pending.set(sessionID, { root, timer })
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
import { Cause, Effect } from "effect"
|
||||
import { MemoryCapture } from "./capture"
|
||||
import { MemoryInstance } from "./instance"
|
||||
import { MemoryLog } from "./log"
|
||||
import type { MemoryPorts } from "./ports"
|
||||
import { MemoryService } from "./service"
|
||||
import { MemoryTimers } from "./timers"
|
||||
|
||||
function brief(cause: Cause.Cause<unknown>) {
|
||||
const err = Cause.squash(cause)
|
||||
return (err instanceof Error ? err.message : String(err)).slice(0, 200)
|
||||
}
|
||||
|
||||
function message(err: unknown) {
|
||||
return (err instanceof Error ? err.message : String(err)).slice(0, 200)
|
||||
}
|
||||
|
||||
export namespace MemoryTurn {
|
||||
export type Reason = "completed" | "error" | "interrupted"
|
||||
type Input = {
|
||||
root: string
|
||||
sessionID: string
|
||||
reason: Reason
|
||||
session: MemoryPorts.SessionPort
|
||||
model: MemoryPorts.ModelPort
|
||||
memoryModel?: string
|
||||
}
|
||||
|
||||
function schedule(input: Input, memory: MemoryService.Interface, root: string) {
|
||||
MemoryTimers.cancel(input.sessionID)
|
||||
const run = MemoryInstance.bind(async () => {
|
||||
MemoryTimers.done(input.sessionID)
|
||||
void Effect.runPromise(
|
||||
memory.turnLock(input.sessionID).withPermits(1)(
|
||||
MemoryCapture.turn({
|
||||
root: input.root,
|
||||
sessionID: input.sessionID,
|
||||
session: input.session,
|
||||
model: input.model,
|
||||
memoryModel: input.memoryModel,
|
||||
reason: "completed",
|
||||
bypassInterval: true,
|
||||
}).pipe(
|
||||
// Timer callbacks run outside the caller's Effect environment, so carry the resolved service from close.
|
||||
Effect.provideService(MemoryService.Service, memory),
|
||||
Effect.catchCause((cause) => Effect.sync(() => MemoryCapture.report(cause))),
|
||||
),
|
||||
),
|
||||
)
|
||||
.catch((err) => MemoryLog.warn("memory idle flush failed", { err: message(err) }))
|
||||
.finally(() => memory.dropLock(input.sessionID))
|
||||
})
|
||||
MemoryTimers.set(input.sessionID, root, setTimeout(run, memory.idleSettle()))
|
||||
}
|
||||
|
||||
export function open(input: { sessionID: string }) {
|
||||
MemoryTimers.cancel(input.sessionID)
|
||||
}
|
||||
|
||||
export const close = Effect.fn("MemoryTurn.close")(function* (input: Input) {
|
||||
const memory = yield* MemoryService.Service
|
||||
yield* memory
|
||||
.turnLock(input.sessionID)
|
||||
.withPermits(1)(
|
||||
Effect.gen(function* () {
|
||||
const info = yield* input.session.get({ sessionID: input.sessionID }).pipe(
|
||||
Effect.catchCause((cause) =>
|
||||
Effect.sync(() => {
|
||||
MemoryLog.warn("memory session lookup failed", { err: brief(cause) })
|
||||
return undefined
|
||||
}),
|
||||
),
|
||||
)
|
||||
if (!info) return
|
||||
if (info.parentID) return
|
||||
const result = yield* MemoryCapture.turn({
|
||||
root: input.root,
|
||||
sessionID: input.sessionID,
|
||||
session: input.session,
|
||||
model: input.model,
|
||||
reason: input.reason,
|
||||
memoryModel: input.memoryModel,
|
||||
}).pipe(
|
||||
Effect.catchCause((cause) =>
|
||||
Effect.sync(() => {
|
||||
MemoryCapture.report(cause)
|
||||
return undefined
|
||||
}),
|
||||
),
|
||||
)
|
||||
if (result?.skipped && result.idleFlush) schedule(input, memory, result.root)
|
||||
}),
|
||||
)
|
||||
.pipe(
|
||||
Effect.catchCause((cause) =>
|
||||
Effect.sync(() => MemoryLog.warn("memory turn-close hook failed", { err: brief(cause) })),
|
||||
),
|
||||
)
|
||||
// Always release this holder. A deferred flush takes its own turnLock/dropLock pair when its
|
||||
// timer fires, so exclusivity across overlapping closes is preserved by the ref count.
|
||||
yield* Effect.sync(() => memory.dropLock(input.sessionID))
|
||||
})
|
||||
}
|
||||
@@ -44,11 +44,7 @@ export namespace Memory {
|
||||
return slug || MemorySlug.hash(text, "memory")
|
||||
}
|
||||
|
||||
async function injected(input: {
|
||||
root: string
|
||||
index: MemoryIndexer.Result
|
||||
sessionID?: string
|
||||
}) {
|
||||
async function injected(input: { root: string; index: MemoryIndexer.Result; sessionID?: string }) {
|
||||
return MemoryFiles.queue(input.root, async () => {
|
||||
const state = await MemoryFiles.readState(input.root)
|
||||
const next = {
|
||||
|
||||
@@ -62,7 +62,9 @@ Output schema:
|
||||
{
|
||||
"reason": "duplicate" | "transient" | "unsupported" | "secret" | "too_specific" | "in_progress" | "policy_belongs_in_docs" | "out_of_scope" | "self_referential" | "quota_guard" | "rate_limit_guard",
|
||||
"text": "short description",
|
||||
"duplicateOf": "source.md:key when the duplicate source is known"
|
||||
"duplicateOf": "source.md:key when the duplicate source is known",
|
||||
"file": "project.md" | "environment.md" | "corrections.md",
|
||||
"section": "the section that holds the existing duplicate, when known"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -77,6 +79,7 @@ Rules:
|
||||
- Use skipped reason "out_of_scope" for personal user-level content that is not project knowledge.
|
||||
- Use skipped reason "self_referential" for statements about memory itself or about facts already being captured.
|
||||
- Use skipped reason "quota_guard" or "rate_limit_guard" if the evidence says memory generation should avoid spending limited quota.
|
||||
- For a "duplicate" skip, set both "file" and "section" to where the existing entry lives, so the duplicate is verified within that exact scope and not against unrelated memory. A "duplicate" claim missing either field cannot be confirmed.
|
||||
- For upsert_environment_fact, use section "Commands" for runnable commands, "Paths" for important directories/files, and "Tooling" for package managers, runtimes, build systems, or test frameworks.
|
||||
- Omit section for other operations.
|
||||
- Do not include markdown.
|
||||
|
||||
@@ -120,9 +120,7 @@ export namespace MemoryRecall {
|
||||
input.state.limits.maxSessionFiles,
|
||||
input.state.limits.maxSessionLineChars,
|
||||
)
|
||||
return items
|
||||
.filter((item) => item.id !== input.currentSessionID && !MemoryDigest.empty(item))
|
||||
.map(digest)
|
||||
return items.filter((item) => item.id !== input.currentSessionID && !MemoryDigest.empty(item)).map(digest)
|
||||
}
|
||||
|
||||
function score(input: { hit: Hit; keys: string[] }) {
|
||||
|
||||
@@ -18,6 +18,9 @@ export namespace MemoryShared {
|
||||
|
||||
export type SourceItem = {
|
||||
id: string
|
||||
file: MemorySchema.Source
|
||||
section: string
|
||||
key: string
|
||||
text: string
|
||||
}
|
||||
|
||||
@@ -37,13 +40,13 @@ export namespace MemoryShared {
|
||||
}
|
||||
|
||||
export function source(input: { file: MemorySchema.Source; text: string }): SourceItem[] {
|
||||
const result: SourceItem[] = []
|
||||
for (const raw of input.text.split("\n")) {
|
||||
const item = entry(raw.trim().replace(/^- /, ""))
|
||||
if (!item) continue
|
||||
result.push({ id: `${input.file}:${item.key}`, text: `${item.key} ${item.text}` })
|
||||
}
|
||||
return result
|
||||
return MemoryMarkdown.parse(input.text).map((item) => ({
|
||||
id: `${input.file}:${item.section}:${item.key}`,
|
||||
file: input.file,
|
||||
section: item.section,
|
||||
key: item.key,
|
||||
text: `${item.key} ${item.text}`,
|
||||
}))
|
||||
}
|
||||
|
||||
export function typed(input: {
|
||||
|
||||
@@ -9,16 +9,11 @@ export namespace MemoryTopics {
|
||||
}
|
||||
|
||||
const limit = {
|
||||
topics: 3,
|
||||
terms: 6,
|
||||
expanded: 24,
|
||||
}
|
||||
const matcher = /[\p{L}\p{N}][\p{L}\p{N}_.-]{1,}/gu
|
||||
|
||||
function uniq(input: MemorySchema.Topic[]): MemorySchema.Topic[] {
|
||||
return [...new Set(input)].slice(0, limit.topics)
|
||||
}
|
||||
|
||||
function section(input: string | undefined) {
|
||||
return input?.trim().toLowerCase() ?? ""
|
||||
}
|
||||
|
||||
@@ -119,14 +119,12 @@ export namespace MemoryAudit {
|
||||
}
|
||||
|
||||
export async function readChanges(root: string) {
|
||||
const lines = (await readDecisions(root))
|
||||
.split("\n")
|
||||
.flatMap((line) => {
|
||||
const data = record(line)
|
||||
if (!data) return []
|
||||
const time = data.time ?? ""
|
||||
return [`${time} ${data.summary}`.trim()]
|
||||
})
|
||||
const lines = (await readDecisions(root)).split("\n").flatMap((line) => {
|
||||
const data = record(line)
|
||||
if (!data) return []
|
||||
const time = data.time ?? ""
|
||||
return [`${time} ${data.summary}`.trim()]
|
||||
})
|
||||
return lines.join("\n")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -164,10 +164,15 @@ export namespace MemoryFs {
|
||||
await rm(file, { recursive: true, force: true })
|
||||
throw error
|
||||
})
|
||||
const timer = setInterval(() => {
|
||||
const now = new Date()
|
||||
void utimes(file, now, now).catch((error: unknown) => warn("failed to refresh memory lock", { error, root }))
|
||||
}, Math.floor(STALE / 3))
|
||||
const timer = setInterval(
|
||||
() => {
|
||||
const now = new Date()
|
||||
void utimes(file, now, now).catch((error: unknown) =>
|
||||
warn("failed to refresh memory lock", { error, root }),
|
||||
)
|
||||
},
|
||||
Math.floor(STALE / 3),
|
||||
)
|
||||
timer.unref()
|
||||
return async () => {
|
||||
clearInterval(timer)
|
||||
|
||||
@@ -69,9 +69,16 @@ export namespace MemoryPaths {
|
||||
const match = text?.match(/^gitdir:\s*(.+)$/m)
|
||||
if (!match?.[1]) return dir
|
||||
const git = path.resolve(dir, match[1])
|
||||
if (!belongs(dot, git)) return dir
|
||||
return checkout(common(git)) ?? dir
|
||||
}
|
||||
|
||||
function belongs(dot: string, git: string) {
|
||||
const back = read(path.join(git, "gitdir"))
|
||||
if (!back) return false
|
||||
return canon(path.resolve(git, back)) === canon(dot)
|
||||
}
|
||||
|
||||
function canon(dir: string) {
|
||||
const resolved = path.resolve(dir)
|
||||
try {
|
||||
|
||||
@@ -132,15 +132,18 @@ export namespace MemorySessions {
|
||||
.filter((item) => item.endsWith(".md"))
|
||||
.sort()
|
||||
.reverse()
|
||||
.reduce(async (prior, file) => {
|
||||
const current = await prior
|
||||
if (current) return current
|
||||
const content = await MemoryFs.read(path.join(listed.paths.sessions, file))
|
||||
if (!content) return
|
||||
const item = parse(file, content, input.max)
|
||||
if (item?.id !== input.sessionID) return
|
||||
return item
|
||||
}, Promise.resolve(undefined as Digest | undefined))
|
||||
.reduce(
|
||||
async (prior, file) => {
|
||||
const current = await prior
|
||||
if (current) return current
|
||||
const content = await MemoryFs.read(path.join(listed.paths.sessions, file))
|
||||
if (!content) return
|
||||
const item = parse(file, content, input.max)
|
||||
if (item?.id !== input.sessionID) return
|
||||
return item
|
||||
},
|
||||
Promise.resolve(undefined as Digest | undefined),
|
||||
)
|
||||
}
|
||||
|
||||
export async function pruneSessions(root: string, max: number) {
|
||||
|
||||
@@ -116,7 +116,10 @@ export namespace MemoryState {
|
||||
if (MemoryFs.miss(error)) return [] as string[]
|
||||
throw error
|
||||
})
|
||||
const digests = files.filter((file) => file.endsWith(".md")).sort().reverse()
|
||||
const digests = files
|
||||
.filter((file) => file.endsWith(".md"))
|
||||
.sort()
|
||||
.reverse()
|
||||
const sources = [
|
||||
paths.project,
|
||||
paths.environment,
|
||||
|
||||
@@ -266,11 +266,21 @@ describe("memory capture parsing", () => {
|
||||
})
|
||||
|
||||
test("verifies duplicate skips and operation duplicates", () => {
|
||||
const items = [{ id: "project.md:repo_tests", text: "repo_tests Run memory tests from packages/opencode." }]
|
||||
const items = [
|
||||
{
|
||||
id: "project.md:Facts:repo_tests",
|
||||
file: "project.md" as const,
|
||||
section: "Facts",
|
||||
key: "repo_tests",
|
||||
text: "repo_tests Run memory tests from packages/opencode.",
|
||||
},
|
||||
]
|
||||
const verified = verifySkips({
|
||||
items,
|
||||
skipped: [
|
||||
{ reason: "duplicate", text: "Run memory tests from packages/opencode." },
|
||||
// Fully scoped to the stored entry → confirmed.
|
||||
{ reason: "duplicate", text: "Run memory tests from packages/opencode.", file: "project.md", section: "Facts" },
|
||||
// Unscoped → unverified regardless of any text overlap.
|
||||
{ reason: "duplicate", text: "New durable workflow preference." },
|
||||
],
|
||||
})
|
||||
@@ -279,29 +289,133 @@ describe("memory capture parsing", () => {
|
||||
skipped: verified.skipped,
|
||||
ops: [
|
||||
{ action: "add", file: "project.md", section: "Facts", key: "repo_tests", text: "Run memory tests." },
|
||||
{ action: "add", file: "project.md", section: "Facts", key: "new_preference", text: "New durable workflow preference." },
|
||||
{
|
||||
action: "add",
|
||||
file: "project.md",
|
||||
section: "Facts",
|
||||
key: "new_preference",
|
||||
text: "New durable workflow preference.",
|
||||
},
|
||||
],
|
||||
})
|
||||
|
||||
expect(verified.skipped[0]?.duplicateOf).toBe("project.md:repo_tests")
|
||||
expect(verified.rescued).toEqual([])
|
||||
expect(verified.skipped[0]?.duplicateOf).toBe("project.md:Facts:repo_tests")
|
||||
expect(verified.skipped).toContainEqual({ reason: "unsupported", text: "New durable workflow preference." })
|
||||
expect(deduped.ops).toEqual([
|
||||
{ action: "add", file: "project.md", section: "Facts", key: "new_preference", text: "New durable workflow preference." },
|
||||
{
|
||||
action: "add",
|
||||
file: "project.md",
|
||||
section: "Facts",
|
||||
key: "new_preference",
|
||||
text: "New durable workflow preference.",
|
||||
},
|
||||
])
|
||||
expect(deduped.skipped.some((item) => item.duplicateOf === "project.md:repo_tests")).toBe(true)
|
||||
expect(deduped.skipped.some((item) => item.duplicateOf === "project.md:Facts:repo_tests")).toBe(true)
|
||||
})
|
||||
|
||||
test("does not pre-skip similar operations from different memory scopes", () => {
|
||||
const filtered = duplicateOps({
|
||||
items: [
|
||||
{
|
||||
id: "corrections.md:Corrections:repo_tests",
|
||||
file: "corrections.md",
|
||||
section: "Corrections",
|
||||
key: "repo_tests",
|
||||
text: "repo_tests Run memory tests from packages/opencode.",
|
||||
},
|
||||
],
|
||||
skipped: [],
|
||||
ops: [
|
||||
{
|
||||
action: "add",
|
||||
file: "project.md",
|
||||
section: "Facts",
|
||||
key: "repo_tests",
|
||||
text: "Run memory tests from packages/opencode.",
|
||||
},
|
||||
],
|
||||
})
|
||||
|
||||
expect(filtered.ops).toHaveLength(1)
|
||||
expect(filtered.skipped).toEqual([])
|
||||
})
|
||||
|
||||
test("scopes model-reported duplicate skips to the claimed file/section", () => {
|
||||
const items = [
|
||||
{
|
||||
id: "corrections.md:Corrections:repo_tests",
|
||||
file: "corrections.md" as const,
|
||||
section: "Corrections",
|
||||
key: "repo_tests",
|
||||
text: "repo_tests Run memory tests from packages/opencode.",
|
||||
},
|
||||
]
|
||||
const verified = verifySkips({
|
||||
items,
|
||||
skipped: [
|
||||
// Claims a duplicate in project.md/Facts, but the only match lives in corrections.md →
|
||||
// unconfirmed, downgraded to advisory instead of confirmed cross-scope.
|
||||
{
|
||||
reason: "duplicate",
|
||||
text: "Run memory tests from packages/opencode.",
|
||||
file: "project.md",
|
||||
section: "Facts",
|
||||
},
|
||||
// Same text, correctly scoped to where the entry actually lives → confirmed.
|
||||
{
|
||||
reason: "duplicate",
|
||||
text: "Run memory tests from packages/opencode.",
|
||||
file: "corrections.md",
|
||||
section: "Corrections",
|
||||
},
|
||||
],
|
||||
})
|
||||
|
||||
expect(verified.skipped[0]).toMatchObject({ reason: "unsupported" })
|
||||
expect(verified.skipped[1]).toMatchObject({
|
||||
reason: "duplicate",
|
||||
duplicateOf: "corrections.md:Corrections:repo_tests",
|
||||
})
|
||||
})
|
||||
|
||||
test("does not confirm a duplicate skip scoped to a file without a section", () => {
|
||||
const items = [
|
||||
{
|
||||
id: "project.md:Decisions:repo_tests",
|
||||
file: "project.md" as const,
|
||||
section: "Decisions",
|
||||
key: "repo_tests",
|
||||
text: "repo_tests Run memory tests from packages/opencode.",
|
||||
},
|
||||
]
|
||||
const verified = verifySkips({
|
||||
items,
|
||||
skipped: [
|
||||
// Claims project.md but not the section; the only match lives in Decisions. Confirming would
|
||||
// risk a cross-section false positive, so it must downgrade to advisory.
|
||||
{ reason: "duplicate", text: "Run memory tests from packages/opencode.", file: "project.md" },
|
||||
],
|
||||
})
|
||||
|
||||
expect(verified.skipped[0]).toEqual({
|
||||
reason: "unsupported",
|
||||
text: "Run memory tests from packages/opencode.",
|
||||
})
|
||||
})
|
||||
|
||||
test("builds capture notices and guard summaries", () => {
|
||||
const ops = [{ action: "add", file: "environment.md", section: "Commands", key: "tests", text: "Run bun test." }] as const
|
||||
const ops = [
|
||||
{ action: "add", file: "environment.md", section: "Commands", key: "tests", text: "Run bun test." },
|
||||
] as const
|
||||
|
||||
expect(notice({ count: 1, ops: [...ops], skipped: [], tokens: 12 })).toMatchObject({
|
||||
type: "saved",
|
||||
message: "Memory saved · environment.md:tests",
|
||||
files: ["environment.md"],
|
||||
})
|
||||
expect(notice({ count: 0, ops: [], skipped: [{ reason: "duplicate", duplicateOf: "project.md:tests" }], tokens: 3 }))
|
||||
.toMatchObject({ type: "skipped", skippedCount: 1 })
|
||||
expect(
|
||||
notice({ count: 0, ops: [], skipped: [{ reason: "duplicate", duplicateOf: "project.md:tests" }], tokens: 3 }),
|
||||
).toMatchObject({ type: "skipped", skippedCount: 1 })
|
||||
expect(skipLine([{ reason: "duplicate", duplicateOf: "project.md:tests" }])).toBe(
|
||||
"reason=duplicate duplicateOf=project.md:tests",
|
||||
)
|
||||
@@ -346,7 +460,11 @@ describe("memory capture parsing", () => {
|
||||
const cases = [
|
||||
["postgres://alice:hunter2@db.local/app", "postgres://[redacted]@db.local/app", "hunter2"],
|
||||
["postgresql://alice:p%40ss@db.local/app", "postgresql://[redacted]@db.local/app", "p%40ss"],
|
||||
["mongodb+srv://user:secret@cluster.mongodb.net/app", "mongodb+srv://[redacted]@cluster.mongodb.net/app", "secret"],
|
||||
[
|
||||
"mongodb+srv://user:secret@cluster.mongodb.net/app",
|
||||
"mongodb+srv://[redacted]@cluster.mongodb.net/app",
|
||||
"secret",
|
||||
],
|
||||
["redis://:cache-secret@localhost:6379/0", "redis://[redacted]@localhost:6379/0", "cache-secret"],
|
||||
["https://user:pass@example.com/path", "https://[redacted]@example.com/path", "pass"],
|
||||
] as const
|
||||
|
||||
@@ -177,6 +177,22 @@ describe("memory core package", () => {
|
||||
})
|
||||
})
|
||||
|
||||
test("does not trust workspace-controlled gitdir pointers for project identity", async () => {
|
||||
await use(async (t) => {
|
||||
const victim = path.join(t.dir, "victim")
|
||||
const other = path.join(t.dir, "other")
|
||||
await mkdir(path.join(other, ".git"), { recursive: true })
|
||||
await mkdir(victim)
|
||||
await writeFile(path.join(victim, ".git"), "gitdir: ../other/.git\n")
|
||||
|
||||
const id = MemoryPaths.identity({ ctx: { directory: victim, worktree: victim } })
|
||||
|
||||
expect(path.basename(id.canonical)).toBe("victim")
|
||||
expect(path.basename(id.canonical)).not.toBe("other")
|
||||
expect(id.folder.startsWith("victim-")).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
test("serializes concurrent operations for one root", async () => {
|
||||
await use(async (t) => {
|
||||
await Memory.enable({ root: t.root })
|
||||
@@ -206,7 +222,10 @@ describe("memory core package", () => {
|
||||
],
|
||||
})
|
||||
await expect(
|
||||
Memory.apply({ root: t.root, ops: [{ action: "add", key: "bad", text: "api_key=sk-abcdefghijklmnopqrstuvwxyz" }] }),
|
||||
Memory.apply({
|
||||
root: t.root,
|
||||
ops: [{ action: "add", key: "bad", text: "api_key=sk-abcdefghijklmnopqrstuvwxyz" }],
|
||||
}),
|
||||
).rejects.toThrow("secret-like content")
|
||||
const shown = await Memory.show({ root: t.root })
|
||||
|
||||
@@ -457,12 +476,10 @@ describe("memory core package", () => {
|
||||
|
||||
const inventory = await MemoryFiles.deriveInventory(t.root)
|
||||
const shown = await Memory.show({ root: t.root })
|
||||
const first = inventory.items[
|
||||
MemoryFiles.inventoryKey({ file: "project.md", section: "Facts", key: "first_hint" })
|
||||
]!
|
||||
const second = inventory.items[
|
||||
MemoryFiles.inventoryKey({ file: "project.md", section: "Facts", key: "second_hint" })
|
||||
]!
|
||||
const first =
|
||||
inventory.items[MemoryFiles.inventoryKey({ file: "project.md", section: "Facts", key: "first_hint" })]!
|
||||
const second =
|
||||
inventory.items[MemoryFiles.inventoryKey({ file: "project.md", section: "Facts", key: "second_hint" })]!
|
||||
|
||||
expect(first.createdAt).toBe(first.updatedAt)
|
||||
expect(second.createdAt).toBe(second.updatedAt)
|
||||
@@ -948,7 +965,9 @@ describe("memory core package", () => {
|
||||
expect(shown.index.match(/type=latest_session_digest/g)?.length).toBe(1)
|
||||
expect(shown.index).toContain("type=latest_session_digest")
|
||||
expect(shown.index).toContain("session=ses_empty_newest")
|
||||
expect(shown.index.indexOf("session=ses_empty_newest")).toBeLessThan(shown.index.indexOf("session=ses_substantive"))
|
||||
expect(shown.index.indexOf("session=ses_empty_newest")).toBeLessThan(
|
||||
shown.index.indexOf("session=ses_substantive"),
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
import { describe, expect, test } from "bun:test"
|
||||
import { mkdtemp, rm } from "fs/promises"
|
||||
import os from "os"
|
||||
import path from "path"
|
||||
import { Effect } from "effect"
|
||||
import { digestPrompt, typedPrompt } from "../src/capture/capture"
|
||||
import { MemoryCapture } from "../src/effect/capture"
|
||||
import { KiloMemory } from "../src/effect/index"
|
||||
import type { MemoryPorts } from "../src/effect/ports"
|
||||
import { MemoryService } from "../src/effect/service"
|
||||
import { MemoryTimers } from "../src/effect/timers"
|
||||
|
||||
async function tmp() {
|
||||
const dir = await mkdtemp(path.join(os.tmpdir(), "kilo-memory-effect-"))
|
||||
return {
|
||||
root: path.join(dir, "memory"),
|
||||
async done() {
|
||||
await rm(dir, { recursive: true, force: true })
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const USAGE = { inputTokens: { total: 12 }, outputTokens: { total: 8 } }
|
||||
|
||||
function view(over: Partial<MemoryPorts.TurnView> = {}): MemoryPorts.TurnView {
|
||||
return {
|
||||
user: "what commands are needed for this repo setup?",
|
||||
assistant: "Use bun install, then bun test ./test from packages/opencode.",
|
||||
recent: "User: setup?\n\nAssistant: bun install then bun test.",
|
||||
lastAssistantID: "msg_assistant",
|
||||
sessionModel: { providerID: "test", modelID: "fake-memory-model" },
|
||||
recalledMemory: false,
|
||||
diffs: [],
|
||||
...over,
|
||||
}
|
||||
}
|
||||
|
||||
/** Session port that always surfaces the given turn (or none). */
|
||||
function session(turn: MemoryPorts.TurnView | undefined): MemoryPorts.SessionPort {
|
||||
return {
|
||||
readTurn: () => Effect.succeed(turn),
|
||||
get: () => Effect.succeed({ parentID: undefined }),
|
||||
}
|
||||
}
|
||||
|
||||
/** Model port that answers digest/typed calls from canned JSON, keyed by system prompt so it is
|
||||
* order-independent (digest and typed run concurrently). */
|
||||
function model(input: { digest: string; typed: string; fallback?: string; onRun?: () => void }): MemoryPorts.ModelPort {
|
||||
return {
|
||||
resolve: () => Effect.succeed({ handle: {}, ...(input.fallback ? { fallback: { reason: input.fallback } } : {}) }),
|
||||
run: async ({ system }) => {
|
||||
input.onRun?.()
|
||||
const text = system === digestPrompt ? input.digest : system === typedPrompt ? input.typed : "{}"
|
||||
return { text, usage: USAGE }
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function run(input: {
|
||||
root: string
|
||||
session: MemoryPorts.SessionPort
|
||||
model: MemoryPorts.ModelPort
|
||||
memoryModel?: string
|
||||
}) {
|
||||
return Effect.runPromise(
|
||||
MemoryCapture.turn({
|
||||
root: input.root,
|
||||
sessionID: "ses_effect",
|
||||
session: input.session,
|
||||
model: input.model,
|
||||
memoryModel: input.memoryModel,
|
||||
reason: "completed",
|
||||
}).pipe(Effect.provideService(MemoryService.Service, MemoryService.make())),
|
||||
)
|
||||
}
|
||||
|
||||
describe("MemoryCapture (fake ports)", () => {
|
||||
test("turn-close typed LLM saves environment memory and audit records", async () => {
|
||||
const t = await tmp()
|
||||
try {
|
||||
await KiloMemory.enable({ root: t.root })
|
||||
await KiloMemory.configure({ root: t.root, settings: { autoConsolidate: true } })
|
||||
|
||||
const result = await run({
|
||||
root: t.root,
|
||||
session: session(view()),
|
||||
model: model({
|
||||
digest: '{"topic":"repo setup","summary":"Explored repo setup commands. Next step: verify memory tests."}',
|
||||
typed:
|
||||
'{"operations":[{"op":"upsert_environment_fact","section":"Commands","key":"cli_memory_tests","value":"Run bun test ./test from packages/opencode."}],"skipped":[]}',
|
||||
}),
|
||||
})
|
||||
|
||||
expect(result).toMatchObject({ skipped: false, operationCount: 1 })
|
||||
if (!("tokens" in result)) throw new Error("expected capture to save memory")
|
||||
expect(result.tokens).toBeGreaterThan(0)
|
||||
|
||||
const shown = await KiloMemory.show({ root: t.root })
|
||||
expect(shown.sources.environment).toContain("cli_memory_tests")
|
||||
expect(shown.decisions).toContain('"kind":"digest"')
|
||||
expect(shown.decisions).toContain('"kind":"typed"')
|
||||
expect(shown.decisions).toContain('"result":"saved"')
|
||||
} finally {
|
||||
await t.done()
|
||||
}
|
||||
})
|
||||
|
||||
test("auto-consolidate off skips digest and typed model writes", async () => {
|
||||
const t = await tmp()
|
||||
try {
|
||||
await KiloMemory.enable({ root: t.root })
|
||||
await KiloMemory.configure({ root: t.root, settings: { autoConsolidate: false } })
|
||||
|
||||
let runs = 0
|
||||
const result = await run({
|
||||
root: t.root,
|
||||
session: session(view()),
|
||||
model: model({
|
||||
digest: '{"topic":"x","summary":"should not be saved"}',
|
||||
typed: '{"operations":[{"op":"upsert_environment_fact","key":"nope","value":"x"}],"skipped":[]}',
|
||||
onRun: () => runs++,
|
||||
}),
|
||||
})
|
||||
|
||||
expect(result).toMatchObject({ skipped: true })
|
||||
expect(runs).toBe(0)
|
||||
const shown = await KiloMemory.show({ root: t.root })
|
||||
expect(shown.sources.environment).not.toContain("nope")
|
||||
} finally {
|
||||
await t.done()
|
||||
}
|
||||
})
|
||||
|
||||
test("records audit when configured memory model is unavailable", async () => {
|
||||
const t = await tmp()
|
||||
try {
|
||||
await KiloMemory.enable({ root: t.root })
|
||||
await KiloMemory.configure({ root: t.root, settings: { autoConsolidate: true } })
|
||||
|
||||
await run({
|
||||
root: t.root,
|
||||
session: session(view()),
|
||||
memoryModel: "test/missing-memory-model",
|
||||
model: model({
|
||||
digest: '{"topic":"repo","summary":"Explored repo setup. Next: verify."}',
|
||||
typed: '{"operations":[],"skipped":[]}',
|
||||
fallback: "model unavailable",
|
||||
}),
|
||||
})
|
||||
|
||||
const shown = await KiloMemory.show({ root: t.root })
|
||||
expect(shown.changes).toContain("memory_model_config reason=model unavailable fallback=1")
|
||||
} finally {
|
||||
await t.done()
|
||||
}
|
||||
})
|
||||
|
||||
test("no turn to capture is skipped", async () => {
|
||||
const t = await tmp()
|
||||
try {
|
||||
await KiloMemory.enable({ root: t.root })
|
||||
const result = await run({
|
||||
root: t.root,
|
||||
session: session(undefined),
|
||||
model: model({ digest: "{}", typed: "{}" }),
|
||||
})
|
||||
expect(result).toMatchObject({ skipped: true, reason: "no_turn" })
|
||||
} finally {
|
||||
await t.done()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe("MemoryService turn-lock ref-counting", () => {
|
||||
test("keeps one semaphore per session until the last holder drops", () => {
|
||||
const svc = MemoryService.make()
|
||||
const a = svc.turnLock("ses_lock")
|
||||
const b = svc.turnLock("ses_lock")
|
||||
expect(b).toBe(a) // a queued close() shares the same semaphore as the holder it waits on
|
||||
svc.dropLock("ses_lock") // first holder settles; second is still queued/holding
|
||||
const c = svc.turnLock("ses_lock")
|
||||
expect(c).toBe(a) // a later close() must not get a fresh semaphore while a holder remains
|
||||
svc.dropLock("ses_lock")
|
||||
svc.dropLock("ses_lock") // last holder leaves → entry dropped
|
||||
const fresh = svc.turnLock("ses_lock")
|
||||
expect(fresh).not.toBe(a) // only now does a new turn get a new semaphore
|
||||
svc.dropLock("ses_lock")
|
||||
})
|
||||
})
|
||||
|
||||
describe("MemoryTimers signal ref-counting", () => {
|
||||
test("shares one controller per root and drops it once the last capture releases", () => {
|
||||
const root = "/kilo-memory/ref-count-root"
|
||||
const first = MemoryTimers.signal(root)
|
||||
const second = MemoryTimers.signal(root)
|
||||
expect(second).toBe(first) // concurrent captures share the controller
|
||||
MemoryTimers.release(root)
|
||||
expect(MemoryTimers.signal(root)).toBe(first) // still alive while one capture remains
|
||||
MemoryTimers.release(root)
|
||||
MemoryTimers.release(root) // last in-flight capture settles → controller dropped
|
||||
const fresh = MemoryTimers.signal(root)
|
||||
expect(fresh).not.toBe(first) // next capture gets a new controller, proving cleanup
|
||||
MemoryTimers.release(root)
|
||||
})
|
||||
})
|
||||
@@ -28,15 +28,27 @@ describe("memory markdown serialization", () => {
|
||||
test("upsert replaces a same-key line in the section and creates absent sections", () => {
|
||||
const base = `## Facts\n${MemoryMarkdown.line("runtime", "Bun 1.0")}\n`
|
||||
|
||||
const replaced = MemoryMarkdown.upsert({ text: base, section: "Facts", line: MemoryMarkdown.line("runtime", "Bun 1.3") })
|
||||
const replaced = MemoryMarkdown.upsert({
|
||||
text: base,
|
||||
section: "Facts",
|
||||
line: MemoryMarkdown.line("runtime", "Bun 1.3"),
|
||||
})
|
||||
expect(replaced.changed).toBe(true)
|
||||
expect(MemoryMarkdown.parse(replaced.text)).toEqual([{ section: "Facts", key: "runtime", text: "Bun 1.3" }])
|
||||
|
||||
const added = MemoryMarkdown.upsert({ text: base, section: "Decisions", line: MemoryMarkdown.line("db", "Postgres") })
|
||||
const added = MemoryMarkdown.upsert({
|
||||
text: base,
|
||||
section: "Decisions",
|
||||
line: MemoryMarkdown.line("db", "Postgres"),
|
||||
})
|
||||
expect(added.changed).toBe(true)
|
||||
expect(MemoryMarkdown.parse(added.text)).toContainEqual({ section: "Decisions", key: "db", text: "Postgres" })
|
||||
|
||||
const noop = MemoryMarkdown.upsert({ text: base, section: "Facts", line: MemoryMarkdown.line("runtime", "Bun 1.0") })
|
||||
const noop = MemoryMarkdown.upsert({
|
||||
text: base,
|
||||
section: "Facts",
|
||||
line: MemoryMarkdown.line("runtime", "Bun 1.0"),
|
||||
})
|
||||
expect(noop.changed).toBe(false)
|
||||
})
|
||||
|
||||
@@ -81,7 +93,10 @@ describe("memory markdown serialization", () => {
|
||||
MemoryMarkdown.line("a", "three"),
|
||||
].join("\n")
|
||||
|
||||
const result = MemoryMarkdown.remove({ text: doc, match: (entry) => entry.section === "Facts" && entry.key === "a" })
|
||||
const result = MemoryMarkdown.remove({
|
||||
text: doc,
|
||||
match: (entry) => entry.section === "Facts" && entry.key === "a",
|
||||
})
|
||||
expect(result.count).toBe(1)
|
||||
expect(MemoryMarkdown.parse(result.text)).toEqual([
|
||||
{ section: "Facts", key: "b", text: "two" },
|
||||
|
||||
Reference in New Issue
Block a user