fix(feedback): address backend review gaps

packages/feedback/message-feedback/tests/helpers.ts

@@ -116,7 +116,7 @@ class TestPersistence extends SessionPersistence {
   inspectFailure: Error | undefined
   inspectCalls = 0
   readFromCalls = 0
-  onReadFrom: (() => void) | undefined
+  onReadFrom: (() => void | Promise<void>) | undefined
   onListSnapshots: (() => void | Promise<void>) | undefined
 
   locate(_meta: SessionHeader): SessionLocation | undefined { return undefined }
@@ -140,16 +140,16 @@ class TestPersistence extends SessionPersistence {
       : Promise.resolve(stored)
   }
 
-  readFrom(
+  async readFrom(
     id: SessionId,
     fromSeq: number,
   ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
     this.readFromCalls += 1
-    this.onReadFrom?.()
+    await this.onReadFrom?.()
     const stored = this.durable.get(id)
     return stored === undefined
       ? Promise.reject(new Error(`test persistence: session '${id}' not found`))
-      : Promise.resolve({ meta: stored.meta, events: stored.events.filter(event => event.seq >= fromSeq) })
+      : { meta: stored.meta, events: stored.events.filter(event => event.seq >= fromSeq) }
   }
 
   list(): Promise<SessionHeader[]> {
@@ -177,6 +177,7 @@ export interface TestHarness {
   readonly ctx: Context
   readonly persistence: TestPersistence
   readonly root: string
+  disposeFeedback(): Promise<void>
   dispose(): Promise<void>
 }
 
@@ -184,22 +185,26 @@ export interface TestHarness {
 export async function setupHarness(maxNoteBytes = 64): Promise<TestHarness> {
   const root = await mkdtemp(join(tmpdir(), 'dsh-message-feedback-test-'))
   const ctx = new Context()
+  let disposeFeedback: (() => Promise<void>) | undefined
   try {
     await ctx.plugin(SessionStore)
     await ctx.plugin(TestPersistence)
     await ctx.plugin(Storage)
     await ctx.plugin(StorageJson, { root })
     await ctx.plugin(StorageDomain, { backend: 'json' })
-    await ctx.plugin(MessageFeedbackService, { maxNoteBytes })
+    const feedbackFiber = await ctx.plugin(MessageFeedbackService, { maxNoteBytes })
+    disposeFeedback = feedbackFiber.dispose
   } catch (error) {
     await ctx.fiber.dispose()
     await rm(root, { recursive: true, force: true })
     throw error
   }
+  if (disposeFeedback === undefined) throw new Error('message feedback test plugin did not load')
   return {
     ctx,
     persistence: ctx.sessionPersistence as unknown as TestPersistence,
     root,
+    disposeFeedback,
     async dispose() {
       await ctx.fiber.dispose()
       await rm(root, { recursive: true, force: true })

packages/feedback/message-feedback/tests/invariant.spec.ts

@@ -0,0 +1,23 @@
+import { describe, expect, it } from 'vitest'
+import InvariantService from '@deepseek-ai/dsh-invariants'
+import * as MessageFeedbackInvariant from '../src/invariant.ts'
+import { setupHarness } from './helpers.ts'
+
+describe('message-feedback invariant companion', () => {
+  it('removes its registry contribution when its fiber is disposed (HMR safety)', async () => {
+    const harness = await setupHarness()
+    try {
+      await harness.ctx.plugin(InvariantService)
+      const fiber = await harness.ctx.plugin(MessageFeedbackInvariant)
+
+      expect(() => {
+        harness.ctx.invariants.register('@deepseek-ai/dsh-message-feedback', () => {})
+      }).toThrow(/already registered/u)
+
+      await fiber.dispose()
+      await expect(harness.ctx.plugin(MessageFeedbackInvariant).await()).resolves.toBeDefined()
+    } finally {
+      await harness.dispose()
+    }
+  })
+})

packages/feedback/message-feedback/tests/message-feedback.spec.ts

@@ -459,6 +459,57 @@ describe('MessageFeedbackService item concurrency', () => {
     }))
     expect(newItem.version).not.toBe(oldItem.version)
   })
+
+  it('drains admitted mutations before domain close and rejects later admission', async () => {
+    const current = await harness()
+    const { ctx, persistence } = current
+    const fixture = messageFixture('dispose-quiescence')
+    persistence.persist(fixture.session)
+    const service = ctx.messageFeedback
+    const lifecycle = service as unknown as { readonly mutationAdmissionOpen: boolean }
+    const started = Promise.withResolvers<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    let physicalReads = 0
+    let committed = 0
+    persistence.onReadFrom = async () => {
+      physicalReads += 1
+      if (physicalReads !== 1) return
+      started.resolve(undefined)
+      await release.promise
+    }
+    ctx.on('domain/changed', (change) => {
+      if (change.domain === 'message_feedback') committed += 1
+    })
+
+    const first = service.put({
+      sessionId: fixture.session.id,
+      messageId: fixture.assistantMessageIds[0],
+      rating: 'positive',
+      ifVersion: null,
+    })
+    await started.promise
+    const second = service.put({
+      sessionId: fixture.session.id,
+      messageId: fixture.assistantMessageIds[1],
+      rating: 'negative',
+      ifVersion: null,
+    })
+    const disposal = current.disposeFeedback()
+    await vi.waitFor(() => { expect(lifecycle.mutationAdmissionOpen).toBe(false) })
+
+    await expect(service.delete({
+      sessionId: fixture.session.id,
+      messageId: fixture.assistantMessageIds[0],
+      ifVersion: staleVersion(),
+    })).rejects.toThrow('message-feedback: service is disposing')
+    release.resolve(undefined)
+
+    expectItem(await first)
+    expectItem(await second)
+    await disposal
+    expect(physicalReads).toBe(2)
+    expect(committed).toBe(2)
+  })
 })
 
 describe('MessageFeedbackService durability ordering', () => {