Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(feedback): harden backend consistency

Browse Source
This commit is contained in:
ZiyaZhang
2026-08-10 20:51:11 -07:00
parent 3cffc77719
commit c3d0fe1bf9
15 changed files with 131 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/feedback/message-feedback/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/feedback/message-feedback/README.md
README.md: 0fda4fb535e5186c252377383520671cd574b364
README.zh.md: 151db5c10048e6ff32d3f7a4ddbcba986d87597a
README.md: 948cdcd2ef0fa2d1bcea68fdc40f16be08e963a8
README.zh.md: 14e62eea0a182bc60be2c370653d1cb031d6b8a8

8
packages/feedback/message-feedback/README.md
View File

@@ -31,7 +31,7 @@ Each stored row carries the inspected Session header identity `{createdAt, cwd}`
`SessionPersistence.inspect()` supplies a cold-safe observation without publishing or resuming an Agent and without committing cold repair. For a Session without a live owner, `listSnapshots()` first decides definite absence; an `inspect()` failure for a catalogued Session remains an infrastructure failure rather than being guessed into `session-not-found`. `put` accepts only a non-empty, append-origin `assistant/message` with the requested `MessageId`; replacement-origin messages, empty usage-only assistant records, and non-assistant records return `target-not-found`.
After initial validation, `put` establishes a durability barrier before writing the sidecar. A matching live Session commits through the canonical `ctx.sessions.flush` checkpoint; a catalogued cold Session is physically re-read from sequence zero through `SessionPersistence.readFrom`. The resulting observation's header identity and target are validated again. A missing flush participant, changed identity, vanished target, or cold physical-read failure prevents the sidecar commit, so durable feedback never precedes the durable target message.
After initial validation, `put` establishes a durability barrier before writing the sidecar. A matching live Session commits through the canonical `ctx.sessions.flush` checkpoint, then both live and cold paths are physically read from sequence zero through `SessionPersistence.readFrom`. The resulting observation's header identity and target are validated again. A missing flush participant, changed identity, vanished target, or physical-read failure prevents the sidecar commit, so durable feedback never precedes the durable target message.
Message feedback is not Session-log content or a Session projection. It emits no `feedback/record` event, does not enter model history, and does not trigger `FEEDBACK_ONLY` telemetry release.
@@ -45,13 +45,13 @@ The same three `MessageFeedbackService` methods are published by `GatewayService
| `put` | `MessageFeedbackPutRequest { sessionId, messageId, rating, note?, ifVersion }` | committed `MessageFeedbackItem` | `session-not-found`, `target-not-found`, `version-conflict`, `note-blank`, `note-too-large` |
| `delete` | `MessageFeedbackDeleteRequest { sessionId, messageId, ifVersion }` | `MessageFeedbackDeleteValue { absent: true }` | `session-not-found`, `version-conflict` |
`MessageFeedbackVersionConflict` returns the caller's `expected` token and the current `actual` token, each nullable where absence is meaningful. `MessageFeedbackNoteTooLarge` returns both `maxBytes` and `actualBytes`. The Client Remote aggregate does not mount the generated client contribution yet; Host callers can use the service/Remote contract without that client assembly.
`MessageFeedbackVersionConflict` returns the authoritative `current` item, or `null` when no item exists. This lets a caller reconcile the current rating, note, and version without a second `list` request. `MessageFeedbackNoteTooLarge` returns both `maxBytes` and `actualBytes`. The Client Remote aggregate does not mount the generated client contribution yet; Host callers can use the service/Remote contract without that client assembly.
## Compare-and-set and idempotency
`ifVersion: null` requests creation only; a material update requires the exact current item version. The check is per message rather than per Session, so changing one item does not conflict with another. Every material create or update assigns a fresh opaque UUID token.
`ifVersion: null` requests creation only; every request for an existing item requires its exact current version, including a no-op whose desired value already matches. The check is per message rather than per Session, so changing one item does not conflict with another. Every material create or update assigns a fresh opaque UUID token, preventing stale writes from crossing an ABA value cycle.
An exact desired-value retry is recognized before `ifVersion` comparison. It returns the already stored item with unchanged version and timestamps, so a caller may safely retry after losing a success response even with the now-stale token or original `null`. `delete` ignores `ifVersion` when the item is already absent and always returns the stable `{ absent: true }` postcondition after success.
A matching-version no-op returns the already stored item with unchanged version and timestamps. After a lost success response, a retry with the old token receives `version-conflict.current`; the caller can compare that authoritative item with its desired value without an extra read. `delete` ignores `ifVersion` when the item is already absent and always returns the stable `{ absent: true }` postcondition after success.
A per-Session promise queue encloses inspection, durability validation, sidecar read, comparison, and whole-row write. These semantics serialize concurrent mutations through one service instance; storage-domain itself has no cross-process conditional write.

8
packages/feedback/message-feedback/README.zh.md
View File

@@ -31,7 +31,7 @@
`SessionPersistence.inspect()` 提供 cold-safe 观测，不发布或恢复 Agent，也不提交 cold repair。对于没有 live owner 的 Session，系统先用 `listSnapshots()` 判定明确不存在；已进入目录的 Session 若 `inspect()` 失败，仍属于基础设施故障，不会被猜测成 `session-not-found`。`put` 只接受具有指定 `MessageId` 的非空、append-origin `assistant/message`；replacement-origin 消息、仅承载 usage 的空 assistant 记录与非 assistant 记录都返回 `target-not-found`。
初步校验后，`put` 在写入伴随记录前建立 durability barrier。身份匹配的 live Session 通过权威 `ctx.sessions.flush` checkpoint 提交；已进入目录的 cold Session 则通过 `SessionPersistence.readFrom` 从序列零做物理复读。随后再次校验所得观测的 header 身份与目标。缺少 flush 参与方、身份变化、目标消失或 cold 物理读取失败都会阻止伴随记录提交，因此持久反馈绝不会先于其持久目标消息。
初步校验后，`put` 在写入伴随记录前建立 durability barrier。身份匹配的 live Session 先通过权威 `ctx.sessions.flush` checkpoint 提交，随后 live 与 cold 路径都会通过 `SessionPersistence.readFrom` 从序列零做物理复读。之后再次校验所得观测的 header 身份与目标。缺少 flush 参与方、身份变化、目标消失或物理读取失败都会阻止伴随记录提交，因此持久反馈绝不会先于其持久目标消息。
message feedback 不是 Session 日志内容或 Session 投影。它不发出 `feedback/record` 事件，不进入模型历史，也不触发 `FEEDBACK_ONLY` 遥测释放。
@@ -45,13 +45,13 @@ message feedback 不是 Session 日志内容或 Session 投影。它不发出 `f
| `put` | `MessageFeedbackPutRequest { sessionId, messageId, rating, note?, ifVersion }` | 已提交的 `MessageFeedbackItem` | `session-not-found`、`target-not-found`、`version-conflict`、`note-blank`、`note-too-large` |
| `delete` | `MessageFeedbackDeleteRequest { sessionId, messageId, ifVersion }` | `MessageFeedbackDeleteValue { absent: true }` | `session-not-found`、`version-conflict` |
`MessageFeedbackVersionConflict` 返回调用方的 `expected` token 与当前 `actual` token；在表示不存在时，两者可以为 null。`MessageFeedbackNoteTooLarge` 同时返回 `maxBytes` 与 `actualBytes`。客户端 Remote 聚合尚未挂载生成的客户端 contribution；Host 调用方无需该客户端组装即可使用 service/Remote 契约。
`MessageFeedbackVersionConflict` 返回权威 `current` 条目；条目不存在时为 `null`。调用方无需额外执行 `list`，即可协调当前 rating、note 与 version。`MessageFeedbackNoteTooLarge` 同时返回 `maxBytes` 与 `actualBytes`。客户端 Remote 聚合尚未挂载生成的客户端 contribution；Host 调用方无需该客户端组装即可使用 service/Remote 契约。
## Compare-and-set 与幂等性
`ifVersion: null` 表示仅当条目不存在时才创建；实质更新要求与当前条目 version 完全一致。检查按消息而非按 Session 进行，因此修改一个条目不会与另一个条目冲突。每次实质创建或更新都会分配新的 opaque UUID token。
`ifVersion: null` 表示仅当条目不存在时才创建；已有条目的每次请求都必须与其当前 version 完全一致，即使目标值已经相同、不会产生实质更新。检查按消息而非按 Session 进行，因此修改一个条目不会与另一个条目冲突。每次实质创建或更新都会分配新的 opaque UUID token，防止陈旧写入穿过 ABA 值循环。
系统会在比较 `ifVersion` 之前识别与目标值完全相同的重试。它返回已存条目，version 与时间戳均不变，因此调用方在成功响应丢失后，即使用当前已陈旧的 token 或原始 `null` 也可安全重试。条目已不存在时，`delete` 忽略 `ifVersion`；成功后始终返回稳定的 `{ absent: true }` 后置条件。
携带匹配 version 的无变化请求会返回已存条目，version 与时间戳均不变。成功响应丢失后，使用旧 token 重试会得到 `version-conflict.current`；调用方无需额外读取，即可把权威当前值与目标值比较。条目已不存在时，`delete` 忽略 `ifVersion`；成功后始终返回稳定的 `{ absent: true }` 后置条件。
按 Session 划分的 promise 队列覆盖检查、持久性校验、伴随记录读取、比较与整行写入。这些语义会串行化经由同一服务实例的并发变更；storage-domain 自身没有跨进程条件写。

2
packages/feedback/message-feedback/package.json
View File

@@ -1,7 +1,7 @@
{
"name": "@deepseek-ai/dsh-message-feedback",
"description": "Lifecycle-bound per-message rating and note sidecar for the DeepSeek Harness",
"version": "0.0.1",
"version": "0.0.1-rc.1",
"publishConfig": {
"access": "restricted"
},

29
packages/feedback/message-feedback/src/index.ts
View File

@@ -192,8 +192,8 @@ export class MessageFeedbackService extends GatewayService {
/**
* Create or replace feedback for one derived append-origin assistant
* message. An exact desired-value retry returns the stored item before its
* stale or `null` version is considered a conflict.
* 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.
*/
@@ -228,14 +228,14 @@ export class MessageFeedbackService extends GatewayService {
const items = current?.items ?? EMPTY_ITEMS
const index = items.findIndex(item => item.messageId === request.messageId)
const existing = items[index]
if (request.ifVersion !== (existing?.version ?? null)) {
return rejected(this.versionConflict(existing ?? null))
}
if (existing !== undefined
&& existing.rating === request.rating
&& existing.note === note.value) {
return success(snapshotItem(existing))
}
if (request.ifVersion !== (existing?.version ?? null)) {
return rejected(this.versionConflict(request, existing?.version ?? null))
}
const now = Date.now()
const item = snapshotItem({
@@ -278,7 +278,7 @@ export class MessageFeedbackService extends GatewayService {
return success<MessageFeedbackDeleteValue>(Object.freeze({ absent: true }))
}
if (request.ifVersion !== existing.version) {
return rejected(this.versionConflict(request, existing.version))
return rejected(this.versionConflict(existing))
}
await table.put(
@@ -298,7 +298,8 @@ export class MessageFeedbackService extends GatewayService {
private async inspectSession(sessionId: SessionId): Promise<KnownSession> {
if (this.ctx.sessions.get(sessionId) === undefined) {
const snapshots = await this.ctx.sessionPersistence.listSnapshots()
if (!snapshots.some(snapshot => snapshot.header.id === sessionId)) {
if (!snapshots.some(snapshot => snapshot.header.id === sessionId)
&& this.ctx.sessions.get(sessionId) === undefined) {
return rejected({ code: 'session-not-found', sessionId })
}
}
@@ -327,7 +328,7 @@ export class MessageFeedbackService extends GatewayService {
`message-feedback: no durability listener participated for live session '${inspection.meta.id}'`,
)
}
return inspection
return await this.ctx.sessionPersistence.readFrom(inspection.meta.id, 0)
}
return await this.ctx.sessionPersistence.readFrom(inspection.meta.id, 0)
}
@@ -343,17 +344,11 @@ export class MessageFeedbackService extends GatewayService {
return success(note)
}
/** Build a conflict branch without exposing an orderable version. */
private versionConflict(
request: Pick<MessageFeedbackPutRequest, 'sessionId' | 'messageId' | 'ifVersion'>,
actual: MessageFeedbackVersion | null,
): MessageFeedbackVersionConflict {
/** Return the authoritative item needed to reconcile one failed comparison. */
private versionConflict(current: MessageFeedbackItem | null): MessageFeedbackVersionConflict {
return {
code: 'version-conflict',
sessionId: request.sessionId,
messageId: request.messageId,
expected: request.ifVersion,
actual,
current: current === null ? null : snapshotItem(current),
}
}

8
packages/feedback/message-feedback/src/types.ts
View File

@@ -89,12 +89,8 @@ export interface MessageFeedbackTargetNotFound {
/** A material mutation did not match the addressed item's current version. */
export interface MessageFeedbackVersionConflict {
readonly code: 'version-conflict'
readonly sessionId: SessionId
readonly messageId: MessageId
/** Version supplied by the caller (`null` means create-only). */
readonly expected: MessageFeedbackVersion | null
/** Current version, or `null` when the item does not exist. */
readonly actual: MessageFeedbackVersion | null
/** Authoritative current item, or `null` when it does not exist. */
readonly current: MessageFeedbackItem | null
}
/** A supplied note contains no non-whitespace character. */

8
packages/feedback/message-feedback/tests/helpers.ts
View File

@@ -117,6 +117,7 @@ class TestPersistence extends SessionPersistence {
inspectCalls = 0
readFromCalls = 0
onReadFrom: (() => void) | undefined
onListSnapshots: (() => void | Promise<void>) | undefined
locate(_meta: SessionHeader): SessionLocation | undefined { return undefined }
create(_meta: SessionHeader): Promise<void> { return Promise.resolve() }
@@ -155,11 +156,12 @@ class TestPersistence extends SessionPersistence {
return Promise.resolve([...this.durable.values()].map(value => value.meta))
}
listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
return Promise.resolve([...this.durable.values()].map((value, index) => ({
async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
await this.onListSnapshots?.()
return [...this.durable.values()].map((value, index) => ({
header: value.meta,
revision: SessionPersistenceRevision(`test:${index}:${value.events.length}`),
})))
}))
}
persist(session: Session): void {

108
packages/feedback/message-feedback/tests/message-feedback.spec.ts
View File

@@ -68,6 +68,25 @@ describe('MessageFeedbackService public contract', () => {
await expect(ctx.messageFeedback.list({ sessionId: fixture.session.id })).rejects.toBe(corruption)
})
it('rechecks live ownership before returning a cold catalog miss', async () => {
const { ctx, persistence } = await harness()
const sessionId = SessionId('catalog-live-race')
const listed = Promise.withResolvers<undefined>()
const release = Promise.withResolvers<undefined>()
persistence.onListSnapshots = async () => {
listed.resolve(undefined)
await release.promise
}
const pending = ctx.messageFeedback.list({ sessionId })
await listed.promise
ctx.sessions.create(sessionId, { meta: { createdAt: 1_700_000_000_001 } })
release.resolve(undefined)
await expect(pending).resolves.toEqual({ ok: true, value: { items: [] } })
expect(persistence.inspectCalls).toBe(1)
})
it('returns session-not-found from mutations and conflicts on an observed version for an absent item', async () => {
const { ctx, persistence } = await harness()
const missing = SessionId('missing-mutations')
@@ -100,13 +119,7 @@ describe('MessageFeedbackService public contract', () => {
ifVersion: expected,
})).resolves.toEqual({
ok: false,
error: {
code: 'version-conflict',
sessionId: fixture.session.id,
messageId: fixture.assistantMessageIds[0],
expected,
actual: null,
},
error: { code: 'version-conflict', current: null },
})
})
@@ -154,7 +167,7 @@ describe('MessageFeedbackService public contract', () => {
sessionId: fixture.session.id,
messageId,
rating: 'negative',
ifVersion: null,
ifVersion: updated.version,
}))
expect(retry).toEqual(updated)
@@ -320,13 +333,7 @@ describe('MessageFeedbackService item concurrency', () => {
ifVersion: first.version,
})).resolves.toEqual({
ok: false,
error: {
code: 'version-conflict',
sessionId: fixture.session.id,
messageId: firstId,
expected: first.version,
actual: updated.version,
},
error: { code: 'version-conflict', current: updated },
})
const listed = await ctx.messageFeedback.list({ sessionId: fixture.session.id })
@@ -335,6 +342,41 @@ describe('MessageFeedbackService item concurrency', () => {
expect(listed.value.items[1]?.version).toBe(second.version)
})
it('rejects a stale put even when the current value has returned to the same state', async () => {
const { ctx, persistence } = await harness()
const fixture = messageFixture('put-aba')
persistence.persist(fixture.session)
const messageId = fixture.assistantMessageIds[0]
const first = expectItem(await ctx.messageFeedback.put({
sessionId: fixture.session.id,
messageId,
rating: 'positive',
ifVersion: null,
}))
const second = expectItem(await ctx.messageFeedback.put({
sessionId: fixture.session.id,
messageId,
rating: 'negative',
ifVersion: first.version,
}))
const current = expectItem(await ctx.messageFeedback.put({
sessionId: fixture.session.id,
messageId,
rating: 'positive',
ifVersion: second.version,
}))
await expect(ctx.messageFeedback.put({
sessionId: fixture.session.id,
messageId,
rating: 'positive',
ifVersion: first.version,
})).resolves.toEqual({
ok: false,
error: { code: 'version-conflict', current },
})
})
it('makes delete retries stable and prevents delete/recreate ABA', async () => {
const { ctx, persistence } = await harness()
const fixture = messageFixture('delete-aba')
@@ -351,9 +393,9 @@ describe('MessageFeedbackService item concurrency', () => {
sessionId: fixture.session.id,
messageId,
ifVersion: staleVersion(),
})).resolves.toMatchObject({
})).resolves.toEqual({
ok: false,
error: { code: 'version-conflict', actual: created.version },
error: { code: 'version-conflict', current: created },
})
const request = {
sessionId: fixture.session.id,
@@ -376,9 +418,9 @@ describe('MessageFeedbackService item concurrency', () => {
ifVersion: null,
}))
expect(recreated.version).not.toBe(created.version)
await expect(ctx.messageFeedback.delete(request)).resolves.toMatchObject({
await expect(ctx.messageFeedback.delete(request)).resolves.toEqual({
ok: false,
error: { code: 'version-conflict', actual: recreated.version },
error: { code: 'version-conflict', current: recreated },
})
})
@@ -449,7 +491,7 @@ describe('MessageFeedbackService durability ordering', () => {
})
})
it('commits a live target checkpoint before the sidecar write without a cold-log reread', async () => {
it('commits and physically verifies a live target checkpoint before the sidecar write', async () => {
const { ctx, persistence } = await harness()
const session = ctx.sessions.create(SessionId('live-checkpoint'), {
meta: { createdAt: 30, cwd: '/live' },
@@ -463,7 +505,7 @@ describe('MessageFeedbackService durability ordering', () => {
ctx.on('domain/changed', (change) => {
if (change.domain === 'message_feedback') order.push('sidecar:durable')
})
persistence.onReadFrom = () => { order.push('unexpected:cold-read') }
persistence.onReadFrom = () => { order.push('session:verified') }
expectItem(await ctx.messageFeedback.put({
sessionId: session.id,
@@ -471,14 +513,14 @@ describe('MessageFeedbackService durability ordering', () => {
rating: 'positive',
ifVersion: null,
}))
expect(order).toEqual(['session:durable', 'sidecar:durable'])
expect(persistence.readFromCalls).toBe(0)
expect(order).toEqual(['session:durable', 'session:verified', 'sidecar:durable'])
expect(persistence.readFromCalls).toBe(1)
expect(persistence.durable.get(session.id)?.events).toContainEqual(
expect.objectContaining({ type: 'assistant/message' }),
)
})
it('fails closed when a live checkpoint fails or has no participant', async () => {
it('fails closed when a live checkpoint fails, has no participant, or is not physically durable', async () => {
const failed = await harness()
const failedSession = failed.ctx.sessions.create(SessionId('live-flush-failure'))
const failedFixture = appendMessageFixture(failedSession)
@@ -508,6 +550,22 @@ describe('MessageFeedbackService durability ordering', () => {
ok: true,
value: { items: [] },
})
const noDurability = await harness()
const unpersistedSession = noDurability.ctx.sessions.create(SessionId('live-unpersisted'))
const unpersistedFixture = appendMessageFixture(unpersistedSession)
noDurability.ctx.on('session/flush', () => {})
await expect(noDurability.ctx.messageFeedback.put({
sessionId: unpersistedSession.id,
messageId: unpersistedFixture.assistantMessageIds[0],
rating: 'positive',
ifVersion: null,
})).rejects.toThrow(/not found/u)
expect(noDurability.persistence.durable.has(unpersistedSession.id)).toBe(false)
await expect(noDurability.ctx.messageFeedback.list({ sessionId: unpersistedSession.id })).resolves.toEqual({
ok: true,
value: { items: [] },
})
})
it('finishes the captured live checkpoint when the Session detaches mid-flush', async () => {
@@ -537,7 +595,7 @@ describe('MessageFeedbackService durability ordering', () => {
expect(ctx.sessions.get(session.id)).toBeUndefined()
release.resolve(undefined)
expectItem(await pending)
expect(persistence.readFromCalls).toBe(0)
expect(persistence.readFromCalls).toBe(1)
await expect(ctx.messageFeedback.list({ sessionId: session.id })).resolves.toMatchObject({
ok: true,
value: { items: [{ messageId: fixture.assistantMessageIds[0] }] },