diff --git a/.changeset/knowledge-reap-guard-deindex.md b/.changeset/knowledge-reap-guard-deindex.md new file mode 100644 index 0000000000..b13689a9da --- /dev/null +++ b/.changeset/knowledge-reap-guard-deindex.md @@ -0,0 +1,29 @@ +--- +'@objectstack/service-knowledge': patch +--- + +知识索引在保留期回收(reap)删除行之前先去索引,不再留下孤儿条目(#4672)。 + +`object` 知识源是**逐记录投影**,而 `LifecycleService` 的保留期回收按谓词删行 —— +ADR-0057 §3.3 又禁止把它扇出成 N 条逐记录事件(清理会回灌正在清空的表)。结果是行没了、 +文档还在:孤儿条目仍会占掉 `topK` 名额(权限过滤发生在适配器返回之后),`isSystem` +调用方还会直接读到它们。 + +现在 `KnowledgeServicePlugin` 会为每个被 `object` 源投影的对象注册一个 ADR-0057 +**reap guard**:sweep 在删除前把候选行交给 guard,guard 按 id 删除对应文档,只确认删除 +成功的行。 + +- **失败方向**:`adapter.delete` 失败 ⇒ 该行本轮**保留**,下次 sweep 重试。允许「行比索引 + 条目活得久」,绝不允许反过来。 +- **组合语义**:多个 guard 按交集组合(#5535),因此它既不会顶掉 `service-storage` 的字节 + 回收 guard,也不会被其顶掉。 +- **分批**沿用 sweep 自身的约束(每批 500 行、每轮 20 批)。 +- **退出开关**:源上的 `refresh.onRecordChange: false` 或插件的 `enableEventSync: false` + 会同时关掉两个方向的内联同步(事件订阅与 reap 去索引)。 + +零新增契约面:未新增任何 spec 键、`IKnowledgeAdapter` 成员或 `packages/objectql` 改动 —— +经由既有的 `ctx.getService('lifecycle')` duck-type + `registerReapGuard` seam 接入,与 +`service-storage` 回收 `sys_file` 字节的方式一致。 + +应用层谓词写(调用方自己的 `multi: true` 写)**仍不覆盖**,#4639 的 warn 保留并已改写为 +准确描述这条分界。 diff --git a/content/docs/protocol/knowledge.mdx b/content/docs/protocol/knowledge.mdx index a915ec1958..6c32b95c28 100644 --- a/content/docs/protocol/knowledge.mdx +++ b/content/docs/protocol/knowledge.mdx @@ -316,11 +316,40 @@ from the event's `after`, and a delete's id from its required `recordId`. than per-record events. A knowledge index is a per-record projection and `matched: 40` names no record, so there is no upsert or delete to derive — the service logs a warning naming the object and count instead of failing - silently. Reconciliation against the source object is the durable fix and is - tracked in [#4672](https://github.com/objectstack-ai/objectstack/issues/4672): - events keep the index *fresh*, reconciliation keeps it *correct*. + silently. Repair such an index with an explicit `reindexSource`. +### Retention deletes are covered by a reap guard + +The platform's own predicate delete — `LifecycleService` reaping rows past +their retention window — is the one bulk path that *is* handled, and it is +handled at the source rather than after the fact +([#4672](https://github.com/objectstack-ai/objectstack/issues/4672)). + +For every object an `object` source projects, the plugin registers an ADR-0057 +**reap guard** with the lifecycle service. The sweep hands the guard the +candidate rows before deleting them; the guard deletes each row's document and +returns the ids it confirms. Rows it does not confirm stay put. + +- **Failure direction.** If `adapter.delete` fails, the row is *kept* and + retried on the next sweep. A row may outlive its index entry; an index entry + must never outlive its row, because an orphan still consumes a `topK` slot + ahead of the permission filter and is read straight through by an `isSystem` + caller. +- **Composition.** Guards from different packages compose by intersection, so + this one cannot displace another registrar's cleanup (or be displaced by it). +- **Batching** is the sweep's: 500 rows per batch, 20 batches per sweep, the + backlog draining across sweeps. +- **Opt-out.** `refresh.onRecordChange: false` on a source — or + `enableEventSync: false` on the plugin — turns off inline index sync in both + directions, event subscription and reap-time de-index alike. An index an + external indexer owns is never a reason to hold up another package's + retention. + +Application-level predicate writes remain uncovered: de-indexing them needs an +adapter-side enumeration capability that no adapter declares today, and that +surface stays unbuilt until a real `object` source calls for it. + `file` / `http` sources rely on explicit `reindexSource` calls (typically triggered by a cron job, a Console button, or a webhook). diff --git a/packages/services/service-knowledge/package.json b/packages/services/service-knowledge/package.json index 9c5047a7fb..972391f131 100644 --- a/packages/services/service-knowledge/package.json +++ b/packages/services/service-knowledge/package.json @@ -22,6 +22,7 @@ "@objectstack/spec": "workspace:*" }, "devDependencies": { + "@objectstack/objectql": "workspace:*", "@types/node": "^26.1.2", "typescript": "^6.0.3", "vitest": "^4.1.10" diff --git a/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts b/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts index 1a728d4ac6..757c0d6e6b 100644 --- a/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts +++ b/packages/services/service-knowledge/src/__tests__/event-sync-data-events.test.ts @@ -141,7 +141,7 @@ describe('#4626 — KnowledgeServicePlugin event sync on data.record.*', () => { // record, so there is nothing to upsert or delete — the adapters take // neither a count nor a predicate. The index is now stale in a way this // subscription cannot repair, and a silent no-op here would read exactly - // like "nothing happened". Reconciliation is tracked in #4672. + // like "nothing happened". expect(upsert).not.toHaveBeenCalled(); expect(del).not.toHaveBeenCalled(); expect(harness.ctx.logger.warn).toHaveBeenCalledWith( @@ -149,4 +149,22 @@ describe('#4626 — KnowledgeServicePlugin event sync on data.record.*', () => { expect.objectContaining({ object: 'task', matched: 12 }), ); }); + + it('[#4672] the warn draws the honest line: reap covered, application predicate writes not', async () => { + // #4672 shipped the reap guard, which covers the platform's OWN predicate + // delete (the retention sweep). It did NOT cover a caller's `multi: true` + // write, and the decision was to say so rather than imply a reconciliation + // pass that does not exist. The warn is the signal that stays honest, so + // its two halves are pinned: a message that stopped naming either one + // would be over- or under-claiming coverage. + const harness = makeCtx(); + const { deliver } = await harness.boot(new KnowledgeServicePlugin()); + + await deliver(BULK_DELETED); + + const [message] = harness.ctx.logger.warn.mock.calls.at(-1) as [string]; + expect(message).toContain('#4639'); + expect(message).toContain('lifecycle reap guard (#4672)'); + expect(message).toContain('application-level predicate writes are not'); + }); }); diff --git a/packages/services/service-knowledge/src/__tests__/reap-guard.test.ts b/packages/services/service-knowledge/src/__tests__/reap-guard.test.ts new file mode 100644 index 0000000000..2f6f977b10 --- /dev/null +++ b/packages/services/service-knowledge/src/__tests__/reap-guard.test.ts @@ -0,0 +1,400 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #4672 — the knowledge index is de-indexed BEFORE the retention sweep deletes + * the row it projects. + * + * A knowledge index built from an `object` source is a per-record projection + * keyed by `sourceRecordId`. `LifecycleService`'s reap deletes rows by + * predicate and ADR-0057 §3.3 forbids fanning that out per record, so before + * this the row vanished and its document stayed: an orphan that still occupies + * a `topK` slot and is read straight through by an `isSystem` caller. + * + * These cases drive the REAL `LifecycleService` rather than a stand-in for it. + * That matters most for the composition case: the property under test (#5535 — + * guards compose by intersection, so the knowledge guard cannot displace + * `service-storage`'s byte-reclaim guard) is a property of the CONSUMER. A fake + * registry would only prove that the fake intersects, which is the shape of a + * test that passes because nothing is produced. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { + LifecycleService, + assertEngineDeleteDispatch, + type LifecycleObjectLike, +} from '@objectstack/objectql'; +import type { KnowledgeSource } from '@objectstack/spec/ai'; +import type { IKnowledgeAdapter } from '@objectstack/spec/contracts'; +import { KnowledgeServicePlugin } from '../knowledge-service-plugin.js'; +import type { KnowledgeService } from '../knowledge-service.js'; + +const FIXED_NOW = 1_700_000_000_000; +const OLD = '2020-01-01T00:00:00Z'; // comfortably past every cutoff below + +/** `task` is knowledge-backed; `audit_log` is the untouched control object. */ +const OBJECTS: LifecycleObjectLike[] = [ + { name: 'task', lifecycle: { class: 'transient', ttl: { field: 'deleted_at', expireAfter: '30d' } } }, + { name: 'audit_log', lifecycle: { class: 'telemetry', retention: { maxAge: '30d' } } }, +]; + +function noteSource(overrides: Partial = {}): KnowledgeSource { + return { + id: 'task_notes', + label: 'Task notes', + adapter: 'fake', + source: { kind: 'object', object: 'task', contentFields: ['notes'] }, + ...overrides, + } as KnowledgeSource; +} + +/** + * Fake ObjectQL engine backing the sweep. + * + * `delete` opens with the producer's own dispatch predicate, so this double + * cannot accept a call the real engine refuses (#4550/#4434) — including the + * `id: { $in: [...] }` shape a hand-mirrored guard always lets through. + */ +function reapEngine(seed: Record>>) { + const store = new Map(Object.entries(seed).map(([k, v]) => [k, [...v]])); + // `id` widened to what the producer's dispatch actually resolves to — the + // double takes the predicate's word for the type, it does not narrow it. + const deleted: Array<{ object: string; id: string | number | bigint }> = []; + + const engine: any = { + registry: { getAllObjects: () => OBJECTS }, + async find(object: string) { + return [...(store.get(object) ?? [])]; + }, + async delete(object: string, options: any) { + const dispatch = assertEngineDeleteDispatch(options); + if (dispatch.kind === 'by-id') { + const rows = store.get(object) ?? []; + store.set(object, rows.filter((r) => String(r.id) !== String(dispatch.id))); + deleted.push({ object, id: dispatch.id }); + return true; + } + // The batched reap never issues one, but the double must not be looser + // than the engine about the shape either. + const before = (store.get(object) ?? []).length; + store.set(object, []); + return before; + }, + getDriverForObject: () => undefined, + datasource(name: string) { + throw new Error(`[ObjectQL] Datasource '${name}' not found`); + }, + }; + + return { + engine, + deleted, + idsLeft: (object: string) => (store.get(object) ?? []).map((r) => String(r.id)), + }; +} + +/** Adapter that records every de-index, and can be told to fail for some ids. */ +function fakeAdapter(failFor: string[] = []) { + const deletedDocs: string[] = []; + const reasons: Array = []; + const adapter: IKnowledgeAdapter = { + id: 'fake', + upsert: async () => undefined, + search: async () => [], + delete: async (documentIds, ctx) => { + for (const documentId of documentIds) { + if (failFor.some((bad) => documentId.endsWith(bad))) { + throw new Error(`backend refused ${documentId}`); + } + deletedDocs.push(documentId); + reasons.push(ctx.reason); + } + }, + }; + return { adapter, deletedDocs, reasons }; +} + +/** + * Boot the plugin against a real `LifecycleService`, exactly as a host does: + * the plugin resolves `lifecycle` by duck-type through `ctx.getService`. + */ +async function boot( + options: ConstructorParameters[0], + opts: { + engine?: any; + lifecycle?: LifecycleService | null; + adapter?: IKnowledgeAdapter; + } = {}, +) { + let readyHook: (() => Promise) | undefined; + let service: KnowledgeService | undefined; + const asked: string[] = []; + + const ctx: any = { + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + hook: (name: string, fn: () => Promise) => { + if (name === 'kernel:ready') readyHook = fn; + }, + registerService: (_name: string, svc: KnowledgeService) => { + service = svc; + }, + getService: (name: string) => { + asked.push(name); + if (name === 'lifecycle' && opts.lifecycle) return opts.lifecycle; + throw new Error(`no service ${name}`); + }, + }; + + const plugin = new KnowledgeServicePlugin(options); + await plugin.init(ctx); + if (opts.adapter) service!.registerAdapter('fake', opts.adapter); + await plugin.start(ctx); + await readyHook?.(); + return { ctx, service: service!, asked }; +} + +function lifecycleOver(engine: any) { + return new LifecycleService({ + getEngine: () => engine, + logger: { info: () => {}, warn: () => {}, debug: () => {} }, + now: () => FIXED_NOW, + initialDelayMs: 1, + sweepIntervalMs: 10, + }); +} + +describe('#4672 — knowledge reap guard', () => { + it('de-indexes each candidate row by id, then confirms it for deletion', async () => { + const { engine, deleted, idsLeft } = reapEngine({ + task: [ + { id: 'k1', deleted_at: OLD }, + { id: 'k2', deleted_at: OLD }, + ], + audit_log: [], + }); + const lifecycle = lifecycleOver(engine); + const { adapter, deletedDocs, reasons } = fakeAdapter(); + await boot({ sources: [noteSource()] }, { lifecycle, adapter }); + + const report = await lifecycle.sweep(); + + // Document id is the same projection key the event-sync delete uses: + // `${sourceId}:${recordId}`. + expect(deletedDocs).toEqual(['task_notes:k1', 'task_notes:k2']); + // Tagged honestly on the EXISTING free-form `AdapterContext.reason`. + expect(reasons).toEqual(['lifecycle-reap', 'lifecycle-reap']); + // …and only then are the rows deleted, by id. + expect(deleted).toEqual([ + { object: 'task', id: 'k1' }, + { object: 'task', id: 'k2' }, + ]); + expect(idsLeft('task')).toEqual([]); + expect(report.errors).toEqual([]); + }); + + it('VETOES a row whose de-index failed — the row outlives its index entry, never the reverse', async () => { + const { engine, deleted, idsLeft } = reapEngine({ + task: [ + { id: 'k1', deleted_at: OLD }, + { id: 'k2', deleted_at: OLD }, + { id: 'k3', deleted_at: OLD }, + ], + audit_log: [], + }); + const lifecycle = lifecycleOver(engine); + const { adapter, deletedDocs } = fakeAdapter(['k2']); + const { ctx } = await boot({ sources: [noteSource()] }, { lifecycle, adapter }); + + await lifecycle.sweep(); + + expect(deletedDocs).toEqual(['task_notes:k1', 'task_notes:k3']); + // k2's row survives this sweep: deleting it would orphan the document the + // adapter still holds. The next sweep retries it. + expect(deleted.map((d) => d.id)).toEqual(['k1', 'k3']); + expect(idsLeft('task')).toEqual(['k2']); + expect(ctx.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("reap guard: de-index failed for source 'task_notes'"), + ); + }); + + it('a failing adapter never leaves the row deleted — the veto holds across repeated sweeps', async () => { + const { engine, deleted, idsLeft } = reapEngine({ + task: [{ id: 'k1', deleted_at: OLD }], + audit_log: [], + }); + const lifecycle = lifecycleOver(engine); + const { adapter } = fakeAdapter(['k1']); + await boot({ sources: [noteSource()] }, { lifecycle, adapter }); + + await lifecycle.sweep(); + await lifecycle.sweep(); + + expect(deleted).toEqual([]); + expect(idsLeft('task')).toEqual(['k1']); + }); + + it('objects with no knowledge source reap exactly as before — no guard, no de-index', async () => { + const { engine, deleted, idsLeft } = reapEngine({ + task: [], + audit_log: [{ id: 'a1', created_at: OLD }, { id: 'a2', created_at: OLD }], + }); + const lifecycle = lifecycleOver(engine); + const { adapter, deletedDocs } = fakeAdapter(); + await boot({ sources: [noteSource()] }, { lifecycle, adapter }); + + await lifecycle.sweep(); + + expect(deleted).toEqual([ + { object: 'audit_log', id: 'a1' }, + { object: 'audit_log', id: 'a2' }, + ]); + expect(idsLeft('audit_log')).toEqual([]); + // The guard is registered per object; `audit_log` never reaches it. + expect(deletedDocs).toEqual([]); + }); +}); + +/** + * [#5535] The consumption-side proof. `registerReapGuard` used to be a + * single-slot `set()`, and this plugin is exactly the "second registrar" + * ADR-0057 §3.3 invites (de-index a derived index by id) — the one that would + * have silently displaced `sys_file`'s byte reclaim. It composes instead. + */ +describe('#4672 — reap guard composition with another registrar', () => { + it('runs alongside a pre-registered guard: both do their cleanup, neither is displaced', async () => { + const { engine, deleted, idsLeft } = reapEngine({ + task: [ + { id: 'k1', deleted_at: OLD }, + { id: 'k2', deleted_at: OLD }, + { id: 'k3', deleted_at: OLD }, + ], + audit_log: [], + }); + const lifecycle = lifecycleOver(engine); + + // Stands in for `service-storage`'s byte-reclaim guard: reclaims the + // external resource for the rows it confirms, and vetoes k3. + const reclaimed: string[] = []; + const byteGuard = vi.fn(async (_object: string, rows: Array>) => { + const confirmed = rows.map((r) => String(r.id)).filter((id) => id !== 'k3'); + reclaimed.push(...confirmed); + return confirmed; + }); + lifecycle.registerReapGuard('task', byteGuard); + + const { adapter, deletedDocs } = fakeAdapter(); + await boot({ sources: [noteSource()] }, { lifecycle, adapter }); + + await lifecycle.sweep(); + + // Both registrars were consulted — the knowledge plugin registering second + // did not unhook the first. + expect(byteGuard).toHaveBeenCalledTimes(1); + expect(byteGuard.mock.calls[0][1].map((r: any) => r.id)).toEqual(['k1', 'k2', 'k3']); + expect(reclaimed).toEqual(['k1', 'k2']); + // …and the knowledge guard was asked only about the rows still standing, so + // it never de-indexes a row another guard is keeping (a de-index is a + // receipt for cleanup already done, not an opinion). + expect(deletedDocs).toEqual(['task_notes:k1', 'task_notes:k2']); + // Intersection: k3 vetoed by one guard is kept whatever the other said. + expect(deleted.map((d) => d.id)).toEqual(['k1', 'k2']); + expect(idsLeft('task')).toEqual(['k3']); + }); + + it("the knowledge guard's own veto keeps a row the other registrar confirmed", async () => { + const { engine, deleted, idsLeft } = reapEngine({ + task: [ + { id: 'k1', deleted_at: OLD }, + { id: 'k2', deleted_at: OLD }, + ], + audit_log: [], + }); + const lifecycle = lifecycleOver(engine); + const otherGuard = vi.fn(async (_o: string, rows: Array>) => + rows.map((r) => String(r.id)), + ); + lifecycle.registerReapGuard('task', otherGuard); + + const { adapter } = fakeAdapter(['k2']); + await boot({ sources: [noteSource()] }, { lifecycle, adapter }); + + await lifecycle.sweep(); + + expect(otherGuard).toHaveBeenCalledTimes(1); + expect(deleted.map((d) => d.id)).toEqual(['k1']); + expect(idsLeft('task')).toEqual(['k2']); + }); +}); + +describe('#4672 — registration seam', () => { + it('goes through the duck-typed lifecycle service, once per guarded object', async () => { + const registered: Array<{ object: string; guard: unknown }> = []; + const fakeLifecycle: any = { + registerReapGuard: (object: string, guard: unknown) => registered.push({ object, guard }), + }; + await boot( + { + sources: [ + noteSource(), + // A second source on the SAME object must not register a second guard… + noteSource({ id: 'task_titles' }), + // …and a source on another object gets its own. + noteSource({ + id: 'doc_body', + source: { kind: 'object', object: 'doc', contentFields: ['body'] }, + } as Partial), + ], + }, + { lifecycle: fakeLifecycle }, + ); + + expect(registered.map((r) => r.object)).toEqual(['task', 'doc']); + // One guard instance for every object: it resolves its targets from the + // object it is called with, so re-registration is a documented no-op. + expect(registered[0].guard).toBe(registered[1].guard); + }); + + it('skips silently when no lifecycle service is registered (bare kernel)', async () => { + const { ctx } = await boot({ sources: [noteSource()] }, { lifecycle: null }); + + // Nothing thrown, nothing concluded about the registry — no sweeper means + // no reap, so there are no orphans to prevent. + expect(ctx.logger.warn).not.toHaveBeenCalledWith(expect.stringContaining('lifecycle')); + expect(ctx.logger.info).not.toHaveBeenCalledWith(expect.stringContaining('reap guards registered')); + }); + + it('never asks for the lifecycle service when no object source is declared', async () => { + const { asked } = await boot( + { + sources: [ + { id: 'handbooks', label: 'Handbooks', adapter: 'fake', source: { kind: 'file', prefix: 'kb/' } } as KnowledgeSource, + ], + }, + { lifecycle: lifecycleOver(reapEngine({ task: [], audit_log: [] }).engine) }, + ); + + expect(asked).not.toContain('lifecycle'); + }); + + it('honours the declared opt-outs rather than inventing a second switch', async () => { + const registrations: string[] = []; + const fakeLifecycle: any = { + registerReapGuard: (object: string) => registrations.push(object), + }; + + // `refresh.onRecordChange: false` — an external indexer owns this index, so + // the platform neither syncs it on record change nor holds up its reaps. + await boot( + { sources: [noteSource({ refresh: { onRecordChange: false } })] }, + { lifecycle: fakeLifecycle }, + ); + expect(registrations).toEqual([]); + + // `enableEventSync: false` is the same statement one level up. + await boot( + { sources: [noteSource()], enableEventSync: false }, + { lifecycle: fakeLifecycle }, + ); + expect(registrations).toEqual([]); + }); +}); diff --git a/packages/services/service-knowledge/src/knowledge-reap-guard.ts b/packages/services/service-knowledge/src/knowledge-reap-guard.ts new file mode 100644 index 0000000000..e948501667 --- /dev/null +++ b/packages/services/service-knowledge/src/knowledge-reap-guard.ts @@ -0,0 +1,127 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { IKnowledgeService } from '@objectstack/spec/contracts'; +import type { KnowledgeSource } from '@objectstack/spec/ai'; +import { documentIdFor, objectSourcesFor } from './knowledge-service.js'; +import type { KnowledgeLogger } from './knowledge-service.js'; + +/** + * Structural shape of a `LifecycleReapGuard` (ADR-0057 amendment). + * + * Declared here rather than imported from `@objectstack/objectql`: the guard + * seam is reached by duck-typing (`ctx.getService('lifecycle')`), exactly as + * `service-storage` reaches it, so this package keeps NO dependency — not even + * a type-only one — on the query engine. The contract is the ADR's, and it is + * the same three facts on both sides: given the object and the candidate rows, + * return the ids you CONFIRM for deletion, having already done your external + * cleanup for them. Ids you do not return are kept this sweep. + */ +type ReapGuard = ( + object: string, + rows: Array>, +) => Promise>; + +/** The slice of `IKnowledgeService` the guard consumes — existing members only. */ +type KnowledgeReapGuardService = Pick; + +/** + * Reap guard that DE-INDEXES a row before the platform lifecycle sweep deletes + * it (#4672). + * + * ## Why this exists + * + * A knowledge index built from an `object` source is a per-record projection + * keyed by `sourceRecordId`. The sweep's retention reap (`LifecycleService`) + * deletes rows by predicate, and ADR-0057 §3.3 forbids fanning that out as N + * per-record events (cleanup must not re-feed the tables it is draining). So + * the row disappears and the document keyed to it does not: an **orphan**. + * Orphans are not merely wasted storage — they occupy `topK` slots ahead of the + * permission filter, diluting real results, and an `isSystem` caller reads them + * straight through. + * + * The guard closes that at the source. It is the ADR's own answer to this shape + * ("a guard is a domain callback, not a second sweeper"), and it is structurally + * identical to `service-storage`'s byte-reclaim guard: reclaim the external + * resource the row points at, then confirm the row. + * + * ## Direction of failure + * + * `adapter.delete` failing means the document is still indexed. Confirming the + * row anyway is precisely the orphan this guard exists to prevent, so a failure + * **vetoes** the row: it is kept, this sweep, and retried on the next one. The + * fail-safe direction is "row outlives its index entry", never the reverse. + * + * A row whose ids partially de-indexed (two sources on one object, one adapter + * down) is vetoed too, and the successful `delete` is re-issued next sweep — + * delete-by-id is idempotent, so the retry costs a call and changes nothing. + * + * Guards compose by intersection (#5535), so vetoing here never overrides + * another registrar's verdict, and confirming here never overrides theirs. + */ +export function createKnowledgeReapGuard( + service: KnowledgeReapGuardService, + logger?: KnowledgeLogger, +): ReapGuard { + return async (object, rows) => { + const confirmed: Array = []; + // Resolved per call, never captured at registration: sources may be + // registered or unregistered while the process runs, and a stale snapshot + // would de-index against a source that no longer exists (or miss one that + // now does) for the life of the kernel. + const targets = objectSourcesFor(service.listSources(), object); + + for (const row of rows) { + const id = row?.id as string | number | undefined | null; + // No usable id → nothing to de-index BY, so nothing can be confirmed. + // (`LifecycleService` drops these before calling a guard; stated here so + // the guard is correct on its own terms, not by a caller's courtesy.) + if (id === undefined || id === null || id === '') continue; + const recordId = String(id); + + let deIndexed = true; + for (const source of targets) { + try { + const adapter = service.getAdapter(source.adapter); + await adapter.delete([documentIdFor(source.id, recordId)], { + source, + // Declared free-form diagnostics tag on the EXISTING + // `AdapterContext.reason` union — an adapter that logs by reason + // sees the reap for what it is, and no spec change is needed. + reason: 'lifecycle-reap', + }); + } catch (err) { + deIndexed = false; + logger?.warn?.( + `[knowledge] reap guard: de-index failed for source '${source.id}' ` + + `(object='${object}', record='${recordId}'): ${(err as Error)?.message ?? err}. ` + + 'Row kept this sweep and retried on the next one — deleting it now would ' + + 'orphan the index entry.', + ); + } + } + if (deIndexed) confirmed.push(id); + } + + return confirmed; + }; +} + +/** + * Objects this service must guard: every distinct object named by an `object` + * source whose row-change sync is on. + * + * Same predicate as the guard's own target resolution, so an object is guarded + * exactly when the guard would have work to do for it. + */ +export function guardedObjectsFor(sources: KnowledgeSource[]): string[] { + const objects = new Set(); + for (const source of sources) { + if (source.source.kind !== 'object') continue; + const object = source.source.object; + // Asked through the SAME predicate the guard uses, never a second copy of + // it: "is this object guarded?" and "what would the guard de-index for it?" + // must never be able to answer differently. + if (objectSourcesFor(sources, object).length > 0) objects.add(object); + } + return [...objects]; +} diff --git a/packages/services/service-knowledge/src/knowledge-service-plugin.ts b/packages/services/service-knowledge/src/knowledge-service-plugin.ts index 79b66fd2c4..1de849e158 100644 --- a/packages/services/service-knowledge/src/knowledge-service-plugin.ts +++ b/packages/services/service-knowledge/src/knowledge-service-plugin.ts @@ -9,6 +9,7 @@ import type { KnowledgeSource } from '@objectstack/spec/ai'; import { KNOWLEDGE_SERVICE } from '@objectstack/spec/contracts'; import { KnowledgeService } from './knowledge-service.js'; import type { KnowledgeLogger } from './knowledge-service.js'; +import { createKnowledgeReapGuard, guardedObjectsFor } from './knowledge-reap-guard.js'; /** * Configuration options for the `KnowledgeServicePlugin`. @@ -69,6 +70,7 @@ export class KnowledgeServicePlugin implements Plugin { private service: KnowledgeService | null = null; private subscriptionId: string | undefined; + private logger: KnowledgeLogger | undefined; constructor(private readonly options: KnowledgeServicePluginOptions = {}) {} @@ -97,6 +99,7 @@ export class KnowledgeServicePlugin implements Plugin { }, }; + this.logger = logger; this.service = new KnowledgeService({ dataEngine: engine, logger, @@ -121,6 +124,10 @@ export class KnowledgeServicePlugin implements Plugin { if (!service) return; ctx.hook('kernel:ready', async () => { + // Reap-path de-indexing first: it is independent of the realtime service, + // and the branch below returns early when that one is absent. + this.installReapGuards(ctx, service); + let realtime: IRealtimeService | null = null; try { realtime = ctx.getService('realtime'); @@ -172,17 +179,27 @@ export class KnowledgeServicePlugin implements Plugin { // Say so rather than falling through to the `return` below: a silent // no-op here reads identically to "nothing happened", which is how the // gap stayed invisible before #4639 gave bulk writes an event at all. - // The durable fix is a reconciliation pass over the object source - // (events keep the index FRESH; reconciliation keeps it CORRECT) — - // tracked separately in #4672. + // + // [#4672] The warn STAYS, and it is deliberately still a warn about a + // gap — but a narrower gap than it named before. The platform's own + // predicate delete (the retention sweep's reap, and the source this + // issue called out as the main one) no longer reaches this handler as a + // stale index at all: `installReapGuards` de-indexes those rows before + // they are deleted. What is left is APPLICATION-level predicate writes + // — a caller's own `multi: true` update/delete — which no guard sits in + // front of. That half is honestly uncovered rather than quietly + // half-claimed, and stays so until a real object source justifies the + // adapter-enumeration surface it would need (#4606's enforced-first + // rule). Reporting it accurately is the whole job of this branch. if (type === 'data.records.updated' || type === 'data.records.deleted') { const matched = payload.matched; ctx.logger.warn?.( `KnowledgeServicePlugin: '${object}' had a predicate write (${type}) affecting ` + `${typeof matched === 'number' ? matched : 'an unreported number of'} record(s). ` + 'A bulk event carries a count, not records, so the knowledge index for this object ' + - 'may now be stale and cannot be repaired from the event stream (#4639; ' + - 'reconciliation tracked in #4672).', + 'may now be stale and cannot be repaired from the event stream (#4639). ' + + 'Retention-sweep deletes are covered separately by the lifecycle reap guard (#4672); ' + + 'application-level predicate writes are not, and need an explicit reindexSource.', { object, type, matched }, ); return; @@ -207,6 +224,75 @@ export class KnowledgeServicePlugin implements Plugin { }); } + /** + * Register the lifecycle reap guard (#4672, ADR-0057 amendment) for every + * object an `object` source projects. + * + * ## Why the guard rather than an event + * + * The retention sweep deletes rows by predicate, and ADR-0057 §3.3 forbids + * fanning that out per record (the cleanup would re-feed the tables it is + * draining). Without a guard the rows go and their documents stay: orphans, + * keyed by a `sourceRecordId` that resolves to nothing. The ADR's own answer + * to this shape is the guard — "a domain callback, not a second sweeper" — + * and it arrives batched and interruptible for free (500 rows a batch, 20 + * batches a sweep), which is the cost bound this work would otherwise owe. + * + * ## Seams, all pre-existing + * + * Reached exactly as `service-storage` reaches it for `sys_file` byte + * reclaim: duck-typed `ctx.getService('lifecycle')` + + * `registerReapGuard(object, guard)`. No spec key, no `IKnowledgeAdapter` + * member, no `packages/objectql` change (#4606's zero-addition boundary). + * Guards compose by intersection (#5535), so registering here cannot displace + * storage's byte-reclaim guard, nor it ours. + * + * Silent when there is no lifecycle service (a bare kernel): the sweep is + * what deletes rows, so no sweeper means no orphans to prevent. Nothing is + * concluded about the registry either — this runs at `kernel:ready`, and the + * absence is neither cached nor asserted. + * + * One guard instance serves every object (it resolves its targets from the + * `object` it is called with), and re-registering the identical function is a + * documented no-op, so re-run wiring cannot double a de-index. + * + * ## Boundary, stated rather than implied + * + * The OBJECT SET is read once, here — every object declared by boot time, + * which is every object the sources a host composes can name. A source + * registered later through `registerSource` for an object that had none at + * boot is therefore unguarded, while one for an already-guarded object is + * picked up (the guard re-resolves its targets on each call). Making the set + * itself dynamic would mean a new notification surface on the service, which + * is precisely the addition #4606 rules out until a real `object` source + * exists to justify it. + */ + private installReapGuards(ctx: PluginContext, service: KnowledgeService): void { + const objects = guardedObjectsFor(service.listSources()); + if (objects.length === 0) return; + + type LifecycleLike = { + registerReapGuard?: (object: string, guard: ReturnType) => void; + }; + let lifecycle: LifecycleLike | undefined; + try { + lifecycle = ctx.getService('lifecycle'); + } catch { + return; // no lifecycle service — nothing reaps, nothing to guard. + } + if (!lifecycle || typeof lifecycle.registerReapGuard !== 'function') return; + + const guard = createKnowledgeReapGuard(service, this.logger); + // Called AS A METHOD, never through a detached reference: the real + // `LifecycleService.registerReapGuard` reads `this.reapGuards`, so + // `const register = lifecycle.registerReapGuard` throws on the first call. + for (const object of objects) lifecycle.registerReapGuard(object, guard); + ctx.logger.info?.( + `KnowledgeServicePlugin: reap guards registered with the lifecycle service for [${objects.join(', ')}] — ` + + 'rows are de-indexed before the retention sweep deletes them.', + ); + } + async stop(ctx: PluginContext): Promise { if (!this.subscriptionId) return; try { diff --git a/packages/services/service-knowledge/src/knowledge-service.ts b/packages/services/service-knowledge/src/knowledge-service.ts index 6609081175..060aab2b44 100644 --- a/packages/services/service-knowledge/src/knowledge-service.ts +++ b/packages/services/service-knowledge/src/knowledge-service.ts @@ -283,17 +283,7 @@ export class KnowledgeService implements IKnowledgeService { } private sourcesForObject(object: string): KnowledgeSource[] { - const out: KnowledgeSource[] = []; - for (const source of this.sources.values()) { - if ( - source.source.kind === 'object' && - (source.source as ObjectKnowledgeSource).object === object && - (source.refresh?.onRecordChange ?? true) !== false - ) { - out.push(source); - } - } - return out; + return objectSourcesFor([...this.sources.values()], object); } /** @@ -374,6 +364,31 @@ export class KnowledgeService implements IKnowledgeService { // ── Helpers ───────────────────────────────────────────────────────── +/** + * The `object` sources whose per-record projection of `object` is kept in step + * with row changes — i.e. those the service may upsert into, delete from, or + * (since #4672) de-index on behalf of, in reaction to a row's lifecycle. + * + * One definition, deliberately: the event-sync path + * (`handleRecordUpsert` / `handleRecordDelete`) and the lifecycle reap guard + * (`knowledge-reap-guard.ts`) are the two halves of one responsibility, and a + * second copy of this predicate is exactly how they would drift into disagreeing + * about which sources a row's disappearance concerns. + * + * `refresh.onRecordChange: false` opts a source out of both halves: it declares + * that an external indexer owns the index, and the reap guard vetoes rows whose + * de-index fails — so honouring it here is also what keeps an opted-out source + * from ever holding up another package's retention. + */ +export function objectSourcesFor(sources: KnowledgeSource[], object: string): KnowledgeSource[] { + return sources.filter( + (source) => + source.source.kind === 'object' && + (source.source as ObjectKnowledgeSource).object === object && + (source.refresh?.onRecordChange ?? true) !== false, + ); +} + /** Deterministic document id derived from the source + record id. */ export function documentIdFor(sourceId: string, recordId: string): string { return `${sourceId}:${recordId}`; diff --git a/packages/services/service-knowledge/vitest.config.ts b/packages/services/service-knowledge/vitest.config.ts index 1a8094c681..3a1e587003 100644 --- a/packages/services/service-knowledge/vitest.config.ts +++ b/packages/services/service-knowledge/vitest.config.ts @@ -9,16 +9,20 @@ export default defineConfig({ environment: 'node', }, resolve: { - alias: { - '@objectstack/core': path.resolve(__dirname, '../../core/src/index.ts'), - '@objectstack/spec/ai': path.resolve(__dirname, '../../spec/src/ai/index.ts'), - '@objectstack/spec/contracts': path.resolve(__dirname, '../../spec/src/contracts/index.ts'), - '@objectstack/spec/data': path.resolve(__dirname, '../../spec/src/data/index.ts'), - '@objectstack/spec/kernel': path.resolve(__dirname, '../../spec/src/kernel/index.ts'), - '@objectstack/spec/system': path.resolve(__dirname, '../../spec/src/system/index.ts'), - // [ADR-0105 D1] Reached transitively via `@objectstack/types` (tenancy posture). - '@objectstack/spec/security': path.resolve(__dirname, '../../spec/src/security/index.ts'), - '@objectstack/spec': path.resolve(__dirname, '../../spec/src/index.ts'), - }, + // Array form with anchored patterns, deliberately. The object form matches + // by PREFIX, so the bare `@objectstack/spec` entry swallowed every subpath + // that was not spelled out above it — `@objectstack/spec/ui` resolved to + // `spec/src/index.ts/ui` and failed with `ENOTDIR`. That made the list of + // namespaces something every new import had to extend by hand, and the + // failure landed on whoever added the import, naming a path nobody wrote. + // One rule for all namespaces cannot go stale that way. + alias: [ + { find: /^@objectstack\/core$/, replacement: path.resolve(__dirname, '../../core/src/index.ts') }, + { + find: /^@objectstack\/spec\/([a-z-]+)$/, + replacement: `${path.resolve(__dirname, '../../spec/src')}/$1/index.ts`, + }, + { find: /^@objectstack\/spec$/, replacement: path.resolve(__dirname, '../../spec/src/index.ts') }, + ], }, }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 76e03abda7..d6a67f5014 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2211,6 +2211,9 @@ importers: specifier: workspace:* version: link:../../spec devDependencies: + '@objectstack/objectql': + specifier: workspace:* + version: link:../../objectql '@types/node': specifier: ^26.1.2 version: 26.1.2