|
|
@@ -1,29 +1,27 @@
|
|
|
/**
|
|
|
- * Durable, lifecycle-bound feedback for finalized assistant messages.
|
|
|
+ * Canonical Session-log feedback for finalized assistant messages.
|
|
|
* @module @deepseek-ai/dsh-message-feedback
|
|
|
*/
|
|
|
|
|
|
import { Buffer } from 'node:buffer'
|
|
|
import { randomUUID } from 'node:crypto'
|
|
|
+import { isDeepStrictEqual } from 'node:util'
|
|
|
import { Context, Service } from '@deepseek-ai/cordis'
|
|
|
import s from '@deepseek-ai/schemastery'
|
|
|
+import { z } from 'zod'
|
|
|
+import { SessionSeq } from '@deepseek-ai/dsh-session/types'
|
|
|
import { deriveEventMessage, isAppendSurfaceEvent } from '@deepseek-ai/dsh-session/surface'
|
|
|
-import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session/types'
|
|
|
+import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
|
|
|
import type {} from '@deepseek-ai/dsh-session'
|
|
|
import type {} from '@deepseek-ai/dsh-session-persistence'
|
|
|
-import type { KvTable } from '@deepseek-ai/dsh-storage-domain'
|
|
|
import { TypertRemoteService, Remote } from '@deepseek-ai/dsh-typert-protocol'
|
|
|
-import { messageFeedbackDomainSpec } from './spec.ts'
|
|
|
-import type { MessageFeedbackRow, MessageFeedbackSessionIdentity } from './spec.ts'
|
|
|
import type {
|
|
|
MessageFeedbackDeleteRequest,
|
|
|
MessageFeedbackDeleteResult,
|
|
|
- MessageFeedbackDeleteValue,
|
|
|
MessageFeedbackFailure,
|
|
|
MessageFeedbackItem,
|
|
|
MessageFeedbackListRequest,
|
|
|
MessageFeedbackListResult,
|
|
|
- MessageFeedbackListValue,
|
|
|
MessageFeedbackNoteBlank,
|
|
|
MessageFeedbackNoteTooLarge,
|
|
|
MessageFeedbackPutRequest,
|
|
|
@@ -36,15 +34,6 @@ import type {
|
|
|
} from './types.ts'
|
|
|
|
|
|
export type * from './types.ts'
|
|
|
-export {
|
|
|
- messageFeedbackDomainSpec,
|
|
|
- messageFeedbackItemSchema,
|
|
|
- messageFeedbackRatingSchema,
|
|
|
- messageFeedbackRowSchema,
|
|
|
- messageFeedbackSessionIdentitySchema,
|
|
|
- messageFeedbackVersionSchema,
|
|
|
-} from './spec.ts'
|
|
|
-export type { MessageFeedbackRow, MessageFeedbackSessionIdentity } from './spec.ts'
|
|
|
|
|
|
/** Required deployment policy for optional notes. */
|
|
|
export interface Config {
|
|
|
@@ -58,104 +47,63 @@ declare module '@deepseek-ai/cordis' {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-/** Immutable empty list reused only as an input to caller-owned copying. */
|
|
|
-const EMPTY_ITEMS: readonly MessageFeedbackItem[] = Object.freeze([])
|
|
|
-
|
|
|
-/** Validate the one deployment-varying limit at the configuration boundary. */
|
|
|
-function resolveMaxNoteBytes(value: number): number {
|
|
|
- if (!Number.isSafeInteger(value) || value < 1) {
|
|
|
- throw new TypeError(
|
|
|
- `message-feedback: maxNoteBytes must be a positive safe integer, got ${String(value)}`,
|
|
|
- )
|
|
|
- }
|
|
|
- return value
|
|
|
-}
|
|
|
+const timestamp = z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER)
|
|
|
+const itemSchema = z.object({
|
|
|
+ messageId: z.string().min(1),
|
|
|
+ rating: z.enum(['positive', 'negative']),
|
|
|
+ note: z.string().refine(note => note.trim().length > 0).optional(),
|
|
|
+ version: z.uuid(),
|
|
|
+ createdAt: timestamp,
|
|
|
+ updatedAt: timestamp,
|
|
|
+}).refine(item => item.updatedAt >= item.createdAt)
|
|
|
+const putSchema = z.object({ sessionId: z.string().min(1), item: itemSchema })
|
|
|
+const deleteSchema = z.object({ sessionId: z.string().min(1), messageId: z.string().min(1) })
|
|
|
+
|
|
|
+type FeedbackEvent = SessionEvent<'feedback/message-put' | 'feedback/message-delete'>
|
|
|
+type Mutation = Pick<SessionEvent<'feedback/message-put'>, 'type' | 'data'>
|
|
|
+ | Pick<SessionEvent<'feedback/message-delete'>, 'type' | 'data'>
|
|
|
+type Append = (event?: Mutation) => Promise<void>
|
|
|
+type MissingSession = MessageFeedbackRejected<MessageFeedbackSessionNotFound>
|
|
|
+type ResolvedNote = MessageFeedbackSuccess<string | undefined>
|
|
|
+ | MessageFeedbackRejected<MessageFeedbackNoteBlank | MessageFeedbackNoteTooLarge>
|
|
|
|
|
|
-/** Copy and freeze one item before it crosses the service boundary. */
|
|
|
+/** Return a caller-owned immutable value, detached from the log. */
|
|
|
function snapshotItem(item: MessageFeedbackItem): MessageFeedbackItem {
|
|
|
- return Object.freeze({
|
|
|
- messageId: item.messageId,
|
|
|
- rating: item.rating,
|
|
|
- ...(item.note === undefined ? {} : { note: item.note }),
|
|
|
- version: item.version,
|
|
|
- createdAt: item.createdAt,
|
|
|
- updatedAt: item.updatedAt,
|
|
|
- })
|
|
|
-}
|
|
|
-
|
|
|
-/** Copy and freeze a list response. */
|
|
|
-function snapshotList(items: readonly MessageFeedbackItem[]): MessageFeedbackListValue {
|
|
|
- return Object.freeze({ items: Object.freeze(items.map(snapshotItem)) })
|
|
|
+ return Object.freeze({ ...item })
|
|
|
}
|
|
|
|
|
|
-/** Build a frozen success branch. */
|
|
|
function success<T>(value: T): MessageFeedbackSuccess<T> {
|
|
|
return Object.freeze({ ok: true, value })
|
|
|
}
|
|
|
|
|
|
-/** Build a frozen business-failure branch. */
|
|
|
function rejected<E extends MessageFeedbackFailure>(error: E): MessageFeedbackRejected<E> {
|
|
|
return Object.freeze({ ok: false, error: Object.freeze(error) })
|
|
|
}
|
|
|
|
|
|
-/** Project the Session fields that distinguish one persisted log lifecycle. */
|
|
|
-function identityOf(header: SessionHeader): MessageFeedbackSessionIdentity {
|
|
|
- return Object.freeze({
|
|
|
- createdAt: header.createdAt,
|
|
|
- ...(header.cwd === undefined ? {} : { cwd: header.cwd }),
|
|
|
- })
|
|
|
-}
|
|
|
-
|
|
|
-/** Whether a stored row belongs to the inspected Session lifecycle. */
|
|
|
-function sameIdentity(row: MessageFeedbackRow, header: SessionHeader): boolean {
|
|
|
- return row.session.createdAt === header.createdAt && row.session.cwd === header.cwd
|
|
|
-}
|
|
|
-
|
|
|
-/** Whether two observations name the same persisted Session lifecycle. */
|
|
|
-function sameHeaderIdentity(left: SessionHeader, right: SessionHeader): boolean {
|
|
|
- return left.id === right.id && left.createdAt === right.createdAt && left.cwd === right.cwd
|
|
|
-}
|
|
|
-
|
|
|
-/** Freeze the replacement row so storage-domain never exposes mutable aliases. */
|
|
|
-function rowSnapshot(
|
|
|
- session: MessageFeedbackSessionIdentity,
|
|
|
- items: readonly MessageFeedbackItem[],
|
|
|
-): MessageFeedbackRow {
|
|
|
- const copiedItems = items.map(snapshotItem)
|
|
|
- Object.freeze(copiedItems)
|
|
|
- return Object.freeze({
|
|
|
- session,
|
|
|
- items: copiedItems,
|
|
|
- })
|
|
|
-}
|
|
|
-
|
|
|
-/** Generate an opaque equality token for one material mutation. */
|
|
|
-function nextVersion(): MessageFeedbackVersion {
|
|
|
- return randomUUID() as MessageFeedbackVersion
|
|
|
-}
|
|
|
-
|
|
|
-/** Observed session view: header identity plus the logged events. */
|
|
|
-interface SessionObservation {
|
|
|
- readonly meta: SessionHeader
|
|
|
- readonly events: readonly SessionEvent[]
|
|
|
+/** Validate persisted payloads before deriving current, Session-owned feedback. */
|
|
|
+function currentItems(sessionId: SessionId, events: readonly SessionEvent[]): MessageFeedbackItem[] {
|
|
|
+ const items = new Map<MessageFeedbackItem['messageId'], MessageFeedbackItem>()
|
|
|
+ for (const event of events) {
|
|
|
+ switch (event.type) {
|
|
|
+ case 'feedback/message-put':
|
|
|
+ putSchema.parse(event.data)
|
|
|
+ if (event.data.sessionId === sessionId) items.set(event.data.item.messageId, event.data.item)
|
|
|
+ break
|
|
|
+ case 'feedback/message-delete':
|
|
|
+ deleteSchema.parse(event.data)
|
|
|
+ if (event.data.sessionId === sessionId) items.delete(event.data.messageId)
|
|
|
+ break
|
|
|
+ default:
|
|
|
+ // Other plugins' events do not change message feedback.
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return [...items.values()]
|
|
|
}
|
|
|
|
|
|
-/** Session observation result that keeps absence inside the business union. */
|
|
|
-type KnownSession =
|
|
|
- | MessageFeedbackSuccess<SessionObservation>
|
|
|
- | MessageFeedbackRejected<MessageFeedbackSessionNotFound>
|
|
|
-
|
|
|
-/** Validated note or one explicit request failure. */
|
|
|
-type ResolvedNote =
|
|
|
- | MessageFeedbackSuccess<string | undefined>
|
|
|
- | MessageFeedbackRejected<MessageFeedbackNoteBlank | MessageFeedbackNoteTooLarge>
|
|
|
-
|
|
|
-/**
|
|
|
- * Storage-domain sidecar service. It inspects persisted Session history and
|
|
|
- * never creates or resumes an Agent or Session.
|
|
|
- */
|
|
|
+/** Session-log service; cold operations never construct a Session or Agent. */
|
|
|
export class MessageFeedbackService extends TypertRemoteService {
|
|
|
- static inject = ['storageDomain', 'sessionPersistence', 'sessions']
|
|
|
+ static inject = ['sessionPersistence', 'sessions']
|
|
|
|
|
|
/** Loader validation for the required note-size policy. */
|
|
|
static Config: s<Config> = s.object({
|
|
|
@@ -163,205 +111,151 @@ export class MessageFeedbackService extends TypertRemoteService {
|
|
|
})
|
|
|
|
|
|
private readonly maxNoteBytes: number
|
|
|
- private table?: KvTable<SessionId, MessageFeedbackRow>
|
|
|
private readonly operationTails = new Map<SessionId, Promise<void>>()
|
|
|
private mutationAdmissionOpen = true
|
|
|
|
|
|
/**
|
|
|
- * @param ctx - Host context carrying persistence and the storage-domain form.
|
|
|
+ * @param ctx - Host context carrying Session persistence and live owners.
|
|
|
* @param config - Required note-size policy.
|
|
|
*/
|
|
|
constructor(ctx: Context, config: Config) {
|
|
|
super(ctx, 'messageFeedback')
|
|
|
- this.maxNoteBytes = resolveMaxNoteBytes(config.maxNoteBytes)
|
|
|
+ if (!Number.isSafeInteger(config.maxNoteBytes) || config.maxNoteBytes < 1) {
|
|
|
+ throw new TypeError('message-feedback: maxNoteBytes must be a positive safe integer')
|
|
|
+ }
|
|
|
+ this.maxNoteBytes = config.maxNoteBytes
|
|
|
}
|
|
|
|
|
|
- /** Open and own the one message-feedback sidecar domain. */
|
|
|
- protected async [Service.init](): Promise<void> {
|
|
|
- const domain = await this.ctx.storageDomain.open(messageFeedbackDomainSpec)
|
|
|
+ protected [Service.init](): void {
|
|
|
this.ctx.effect(() => async () => {
|
|
|
this.mutationAdmissionOpen = false
|
|
|
await Promise.all(this.operationTails.values())
|
|
|
- await domain.close()
|
|
|
- }, 'message-feedback.domainClose')
|
|
|
- this.table = domain.table('sessions')
|
|
|
+ }, 'message-feedback.drain')
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Read feedback belonging to the current persisted Session lifecycle.
|
|
|
- * A stale row from a reused Session id is invisible.
|
|
|
- * @param request - Session identity to inspect and list.
|
|
|
- * @returns current immutable items or `session-not-found`.
|
|
|
+ * Read current feedback from the canonical log.
|
|
|
+ * @param request - Session to inspect.
|
|
|
+ * @returns immutable items or a definite persistence miss.
|
|
|
*/
|
|
|
@Remote('list')
|
|
|
- async list(request: MessageFeedbackListRequest): Promise<MessageFeedbackListResult> {
|
|
|
- const known = await this.inspectSession(request.sessionId)
|
|
|
- if (!known.ok) return known
|
|
|
- const row = this.requireTable().get(request.sessionId)
|
|
|
- const items = row !== undefined && sameIdentity(row, known.value.meta) ? row.items : EMPTY_ITEMS
|
|
|
- return success(snapshotList(items))
|
|
|
+ list(request: MessageFeedbackListRequest): Promise<MessageFeedbackListResult> {
|
|
|
+ return this.enqueue(request.sessionId, () => this.withSession(request.sessionId, false, events =>
|
|
|
+ success(Object.freeze({ items: Object.freeze(currentItems(request.sessionId, events).map(snapshotItem)) }))))
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Create or replace feedback for one derived append-origin assistant
|
|
|
- * message. Every request must match the addressed item's current version;
|
|
|
- * a matching no-op returns the stored item without changing its revision.
|
|
|
- * @param request - target, desired value, and observed item version.
|
|
|
- * @returns the committed item or an explicit business failure.
|
|
|
+ * Create or replace feedback after checking its current version.
|
|
|
+ * Matching no-ops retain the version and append no event.
|
|
|
+ * @param request - Target, desired value, and observed item version.
|
|
|
+ * @returns the durable item or an explicit business failure.
|
|
|
*/
|
|
|
@Remote('put')
|
|
|
put(request: MessageFeedbackPutRequest): Promise<MessageFeedbackPutResult> {
|
|
|
const note = this.resolveNote(request.note)
|
|
|
if (!note.ok) return Promise.resolve(note)
|
|
|
- return this.enqueue(request.sessionId, async () => {
|
|
|
- const known = await this.inspectSession(request.sessionId)
|
|
|
- if (!known.ok) return known
|
|
|
- if (!this.hasFeedbackTarget(known.value, request.messageId)) {
|
|
|
- return rejected({
|
|
|
- code: 'target-not-found',
|
|
|
- sessionId: request.sessionId,
|
|
|
- messageId: request.messageId,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- const durable = await this.ensureTargetDurable(known.value)
|
|
|
- if (!sameHeaderIdentity(durable.meta, known.value.meta)
|
|
|
- || !this.hasFeedbackTarget(durable, request.messageId)) {
|
|
|
- return rejected({
|
|
|
- code: 'target-not-found',
|
|
|
- sessionId: request.sessionId,
|
|
|
- messageId: request.messageId,
|
|
|
- })
|
|
|
+ return this.enqueue(request.sessionId, () => this.withSession(request.sessionId, true, async (events, append) => {
|
|
|
+ const items = currentItems(request.sessionId, events)
|
|
|
+ if (!events.some(event => event.type === 'assistant/message'
|
|
|
+ && isAppendSurfaceEvent(event)
|
|
|
+ && deriveEventMessage(event)?.id === request.messageId)) {
|
|
|
+ return rejected({ code: 'target-not-found', sessionId: request.sessionId, messageId: request.messageId })
|
|
|
}
|
|
|
-
|
|
|
- const table = this.requireTable()
|
|
|
- const stored = table.get(request.sessionId)
|
|
|
- const current = stored !== undefined && sameIdentity(stored, durable.meta) ? stored : undefined
|
|
|
- const items = current?.items ?? EMPTY_ITEMS
|
|
|
- const index = items.findIndex(item => item.messageId === request.messageId)
|
|
|
- const existing = items[index]
|
|
|
+ const existing = items.find(item => item.messageId === request.messageId)
|
|
|
if (request.ifVersion !== (existing?.version ?? null)) {
|
|
|
return rejected(this.versionConflict(existing ?? null))
|
|
|
}
|
|
|
- if (existing !== undefined
|
|
|
- && existing.rating === request.rating
|
|
|
- && existing.note === note.value) {
|
|
|
+ if (existing !== undefined && existing.rating === request.rating && existing.note === note.value) {
|
|
|
+ await append()
|
|
|
return success(snapshotItem(existing))
|
|
|
}
|
|
|
-
|
|
|
const now = Date.now()
|
|
|
- const item = snapshotItem({
|
|
|
+ const item: MessageFeedbackItem = {
|
|
|
messageId: request.messageId,
|
|
|
rating: request.rating,
|
|
|
...(note.value === undefined ? {} : { note: note.value }),
|
|
|
- version: nextVersion(),
|
|
|
+ version: randomUUID() as MessageFeedbackVersion,
|
|
|
createdAt: existing?.createdAt ?? now,
|
|
|
updatedAt: existing === undefined ? now : Math.max(now, existing.updatedAt),
|
|
|
- })
|
|
|
- const nextItems = [...items]
|
|
|
- if (index === -1) nextItems.push(item)
|
|
|
- else nextItems[index] = item
|
|
|
- await table.put(
|
|
|
- request.sessionId,
|
|
|
- rowSnapshot(identityOf(durable.meta), nextItems),
|
|
|
- )
|
|
|
+ }
|
|
|
+ await append({ type: 'feedback/message-put', data: { sessionId: request.sessionId, item } })
|
|
|
return success(snapshotItem(item))
|
|
|
- })
|
|
|
+ }))
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Delete one feedback item. Absence is successful regardless of the
|
|
|
- * supplied version; an existing item requires an exact version match.
|
|
|
+ * Delete one item after checking its version; absence succeeds without an event.
|
|
|
* @param request - Session, message, and observed item version.
|
|
|
- * @returns the stable absent postcondition, or an explicit failure.
|
|
|
+ * @returns the stable absent postcondition or an explicit failure.
|
|
|
*/
|
|
|
@Remote('delete')
|
|
|
delete(request: MessageFeedbackDeleteRequest): Promise<MessageFeedbackDeleteResult> {
|
|
|
- return this.enqueue(request.sessionId, async () => {
|
|
|
- const known = await this.inspectSession(request.sessionId)
|
|
|
- if (!known.ok) return known
|
|
|
-
|
|
|
- const table = this.requireTable()
|
|
|
- const stored = table.get(request.sessionId)
|
|
|
- const current = stored !== undefined && sameIdentity(stored, known.value.meta) ? stored : undefined
|
|
|
- const items = current?.items ?? EMPTY_ITEMS
|
|
|
- const existing = items.find(item => item.messageId === request.messageId)
|
|
|
- if (existing === undefined) {
|
|
|
- return success<MessageFeedbackDeleteValue>(Object.freeze({ absent: true }))
|
|
|
- }
|
|
|
- if (request.ifVersion !== existing.version) {
|
|
|
- return rejected(this.versionConflict(existing))
|
|
|
+ return this.enqueue(request.sessionId, () => this.withSession(request.sessionId, true, async (events, append) => {
|
|
|
+ const existing = currentItems(request.sessionId, events).find(item => item.messageId === request.messageId)
|
|
|
+ if (existing !== undefined) {
|
|
|
+ if (request.ifVersion !== existing.version) return rejected(this.versionConflict(existing))
|
|
|
+ await append({ type: 'feedback/message-delete', data: { sessionId: request.sessionId, messageId: request.messageId } })
|
|
|
+ } else {
|
|
|
+ await append()
|
|
|
}
|
|
|
-
|
|
|
- await table.put(
|
|
|
- request.sessionId,
|
|
|
- rowSnapshot(identityOf(known.value.meta), items.filter(item => item !== existing)),
|
|
|
- )
|
|
|
- return success<MessageFeedbackDeleteValue>(Object.freeze({ absent: true }))
|
|
|
- })
|
|
|
+ return success(Object.freeze({ absent: true as const }))
|
|
|
+ }))
|
|
|
}
|
|
|
|
|
|
- /**
|
|
|
- * Resolve a live owner directly; otherwise use `stat` as the existence
|
|
|
- * authority before reading the log. Read failures for a Session that `stat`
|
|
|
- * confirmed remain infrastructure failures rather than being guessed into
|
|
|
- * the business `session-not-found` branch.
|
|
|
- */
|
|
|
- private async inspectSession(sessionId: SessionId): Promise<KnownSession> {
|
|
|
- if (this.ctx.sessions.get(sessionId) === undefined) {
|
|
|
- if (await this.ctx.sessionPersistence.stat(sessionId) === undefined
|
|
|
- && this.ctx.sessions.get(sessionId) === undefined) {
|
|
|
- return rejected({ code: 'session-not-found', sessionId })
|
|
|
- }
|
|
|
+ /** Hold cold write ownership across read/compare/append; use live owners directly. */
|
|
|
+ private async withSession<T>(
|
|
|
+ sessionId: SessionId,
|
|
|
+ write: boolean,
|
|
|
+ operation: (events: readonly SessionEvent[], append: Append) => T | Promise<T>,
|
|
|
+ ): Promise<T | MissingSession> {
|
|
|
+ if (this.ctx.sessions.get(sessionId) === undefined
|
|
|
+ && await this.ctx.sessionPersistence.stat(sessionId) === undefined
|
|
|
+ && this.ctx.sessions.get(sessionId) === undefined) {
|
|
|
+ return rejected({ code: 'session-not-found', sessionId })
|
|
|
}
|
|
|
- return success(await this.observeSession(sessionId))
|
|
|
- }
|
|
|
-
|
|
|
- /** Observe a live owner's in-memory log when one exists, else the durable log. */
|
|
|
- private async observeSession(sessionId: SessionId): Promise<SessionObservation> {
|
|
|
const live = this.ctx.sessions.get(sessionId)
|
|
|
- if (live !== undefined) return { meta: live.header, events: live.snapshotEvents() }
|
|
|
- return await this.readDurable(sessionId)
|
|
|
- }
|
|
|
-
|
|
|
- /** Read the complete durable log prefix through a fresh read handle. */
|
|
|
- private async readDurable(sessionId: SessionId): Promise<SessionObservation> {
|
|
|
- const handle = await this.ctx.sessionPersistence.open(sessionId, 'read')
|
|
|
+ if (live !== undefined) {
|
|
|
+ return operation(live.snapshotEvents(), async (event) => {
|
|
|
+ if (event !== undefined) {
|
|
|
+ live.append(event.type, event.data)
|
|
|
+ }
|
|
|
+ const last = live.snapshotEvents().at(-1)
|
|
|
+ if (!(await this.ctx.sessions.flush(live))) {
|
|
|
+ throw new Error(
|
|
|
+ `message-feedback: no durability listener participated for live session '${sessionId}'`,
|
|
|
+ )
|
|
|
+ }
|
|
|
+ // Listener participation alone does not prove this Session has a persistence writer.
|
|
|
+ const handle = await this.ctx.sessionPersistence.open(sessionId, 'read')
|
|
|
+ try {
|
|
|
+ const stored = await handle.read(last?.seq ?? 0, 1)
|
|
|
+ if (!isDeepStrictEqual(
|
|
|
+ [handle.header.id, handle.header.createdAt, handle.header.cwd],
|
|
|
+ [live.header.id, live.header.createdAt, live.header.cwd],
|
|
|
+ )
|
|
|
+ || (last !== undefined && !isDeepStrictEqual(stored[0], last))) {
|
|
|
+ throw new Error(`message-feedback: feedback prefix is not durable for live session '${sessionId}'`)
|
|
|
+ }
|
|
|
+ } finally {
|
|
|
+ await handle.close()
|
|
|
+ }
|
|
|
+ })
|
|
|
+ }
|
|
|
+ const handle = await this.ctx.sessionPersistence.open(sessionId, write ? 'write' : 'read')
|
|
|
try {
|
|
|
- return { meta: handle.header, events: await handle.read() }
|
|
|
+ const events = await handle.read()
|
|
|
+ return await operation(events, async (event) => {
|
|
|
+ if (event !== undefined) {
|
|
|
+ const entry: FeedbackEvent = { ...event, seq: SessionSeq(events.length), time: Date.now() }
|
|
|
+ await handle.append([entry])
|
|
|
+ }
|
|
|
+ await handle.flush()
|
|
|
+ })
|
|
|
} finally {
|
|
|
await handle.close()
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /** Require the exact finalized append-origin assistant message projection. */
|
|
|
- private hasFeedbackTarget(observation: SessionObservation, messageId: MessageFeedbackItem['messageId']): boolean {
|
|
|
- return observation.events.some((event) => {
|
|
|
- if (event.type !== 'assistant/message' || !isAppendSurfaceEvent(event)) return false
|
|
|
- const message = deriveEventMessage(event)
|
|
|
- return message?.role === 'assistant' && message.id === messageId
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * Put the target log prefix behind a durability barrier before its sidecar.
|
|
|
- * A live owner flushes through the SessionStore's canonical checkpoint; the
|
|
|
- * physical durable prefix is then re-read, which observes at least the
|
|
|
- * flushed prefix by the `SessionPersistence` freshness guarantee.
|
|
|
- */
|
|
|
- private async ensureTargetDurable(observation: SessionObservation): Promise<SessionObservation> {
|
|
|
- const live = this.ctx.sessions.get(observation.meta.id)
|
|
|
- if (live !== undefined && sameHeaderIdentity(live.header, observation.meta)) {
|
|
|
- if (!(await this.ctx.sessions.flush(live))) {
|
|
|
- throw new Error(
|
|
|
- `message-feedback: no durability listener participated for live session '${observation.meta.id}'`,
|
|
|
- )
|
|
|
- }
|
|
|
- }
|
|
|
- return await this.readDurable(observation.meta.id)
|
|
|
- }
|
|
|
-
|
|
|
- /** Validate optional-note semantics and the configured complete UTF-8 byte bound. */
|
|
|
private resolveNote(note: string | undefined): ResolvedNote {
|
|
|
if (note === undefined) return success(undefined)
|
|
|
if (note.trim().length === 0) return rejected({ code: 'note-blank' })
|
|
|
@@ -372,19 +266,13 @@ export class MessageFeedbackService extends TypertRemoteService {
|
|
|
return success(note)
|
|
|
}
|
|
|
|
|
|
- /** Return the authoritative item needed to reconcile one failed comparison. */
|
|
|
private versionConflict(current: MessageFeedbackItem | null): MessageFeedbackVersionConflict {
|
|
|
- return {
|
|
|
- code: 'version-conflict',
|
|
|
- current: current === null ? null : snapshotItem(current),
|
|
|
- }
|
|
|
+ return { code: 'version-conflict', current: current === null ? null : snapshotItem(current) }
|
|
|
}
|
|
|
|
|
|
- /** Queue a complete read/compare/write mutation behind this Session's prior mutation. */
|
|
|
+ /** Serialize complete operations and drain their handles before disposal. */
|
|
|
private enqueue<T>(sessionId: SessionId, operation: () => Promise<T>): Promise<T> {
|
|
|
- if (!this.mutationAdmissionOpen) {
|
|
|
- return Promise.reject(new Error('message-feedback: service is disposing'))
|
|
|
- }
|
|
|
+ if (!this.mutationAdmissionOpen) return Promise.reject(new Error('message-feedback: service is disposing'))
|
|
|
const previous = this.operationTails.get(sessionId) ?? Promise.resolve()
|
|
|
const result = previous.then(operation)
|
|
|
const tail = result.then(() => undefined, () => undefined)
|
|
|
@@ -393,14 +281,6 @@ export class MessageFeedbackService extends TypertRemoteService {
|
|
|
if (this.operationTails.get(sessionId) === tail) this.operationTails.delete(sessionId)
|
|
|
})
|
|
|
}
|
|
|
-
|
|
|
- /** Resolve the initialized durable table or fail a broken service lifecycle. */
|
|
|
- private requireTable(): KvTable<SessionId, MessageFeedbackRow> {
|
|
|
- if (this.table === undefined) {
|
|
|
- throw new Error('message-feedback: durable domain is not initialized')
|
|
|
- }
|
|
|
- return this.table
|
|
|
- }
|
|
|
}
|
|
|
|
|
|
export default MessageFeedbackService
|