mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-08 19:19:42 +00:00
feat(fixtures): SSE patch + pagination scenario fixtures (K7)
Two new golden-fixture suites, expectations machine-generated from the web
implementation (same source-of-truth principle as the chat suite):
- shared/fixtures/sse/ (12 cases): session-updated versioned-patch
application via applySessionDetailPatch — strict version gates for
metadata/agentState/todos/teamState, out-of-order arrival, teamState:null
clear, max-monotonic updatedAt, flat last-write-wins fields, sub-minute
activeAt keep-alive drop, scratchlistUpdatedAt trigger, and the pinned
actual behavior that activeTurnStartedAt is NOT applied by the patch path.
Inputs are stored schema-normalized and validated against SessionSchema /
strict SessionPatchSchema at generation time; per-patch applied/unchanged
verdicts are part of the contract.
- shared/fixtures/pagination/ (11 cases): op scripts driving the real
message-window store with a scripted ApiClient — latest page + SSE ingest,
before-cursor older pages, epoch-mismatch reset (window discard + recorded
internal latest request), reset:true replace preserving optimistic rows,
localId echo reconciliation, messages-consumed invokedAt stamping (no
cursor advance), message-cancelled removal, cancel-too-late invoked-row
ingest, hidden-row cursor advance, 400-row trim preserving queued rows,
and queued-state gap recovery. Documents pin the exact requests the store
issued, older-load outcomes, reconcile candidates, and a minimal window
projection (ids/order, queued/optimistic flags, hasMore, epoch, viewMode,
compound cursors).
Generator: web/scripts/fixtures/{sse,pagination}/ extend the K5 framework
(canonical serialization reused; suites pruned of stale files). Web
self-conformance tests replay every fixture against the real implementation
in bun run test:web. README documents schemas, projections, replay
contracts, and native consumption for both suites.
No time injection needed: the store's only Date.now() reads gate
notification throttling, which never reaches persisted or projected state.
This commit is contained in:
weishu
@@ -3,8 +3,12 @@ import { join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { fixtureCases } from './cases'
|
||||
import { FIXTURE_VERSION, toFixtureInput, type FixtureCase, type FixtureDocument, type FixtureInput } from './fixtureTypes'
|
||||
import { buildPaginationFixtureDocument } from './pagination/build'
|
||||
import { paginationFixtureCases } from './pagination/cases'
|
||||
import { runFixturePipeline } from './pipeline'
|
||||
import { toCanonicalJson } from './serialize'
|
||||
import { buildSseFixtureDocument } from './sse/build'
|
||||
import { sseFixtureCases } from './sse/cases'
|
||||
|
||||
const FIXTURES_DIR = fileURLToPath(new URL('../../../shared/fixtures', import.meta.url))
|
||||
|
||||
@@ -23,31 +27,60 @@ export function buildFixtureDocument(fixtureCase: FixtureCase): FixtureDocument
|
||||
}
|
||||
}
|
||||
|
||||
export function generateAllFixtures(): void {
|
||||
const chatDir = join(FIXTURES_DIR, 'chat')
|
||||
mkdirSync(chatDir, { recursive: true })
|
||||
type SuiteWriter = {
|
||||
dir: string
|
||||
names: Set<string>
|
||||
}
|
||||
|
||||
const names = new Set<string>()
|
||||
function openSuite(suite: string): SuiteWriter {
|
||||
const dir = join(FIXTURES_DIR, suite)
|
||||
mkdirSync(dir, { recursive: true })
|
||||
return { dir, names: new Set() }
|
||||
}
|
||||
|
||||
function writeSuiteDocument(writer: SuiteWriter, name: string, document: unknown): void {
|
||||
if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(name)) {
|
||||
throw new Error(`Fixture name must be kebab-case: ${name}`)
|
||||
}
|
||||
if (writer.names.has(name)) {
|
||||
throw new Error(`Duplicate fixture name: ${name}`)
|
||||
}
|
||||
writer.names.add(name)
|
||||
writeFileSync(join(writer.dir, `${name}.json`), toCanonicalJson(document))
|
||||
}
|
||||
|
||||
// Remove stale fixtures from renamed/deleted cases so natives never keep
|
||||
// passing against a file the web pipeline no longer generates.
|
||||
function pruneSuite(writer: SuiteWriter): void {
|
||||
for (const entry of readdirSync(writer.dir)) {
|
||||
if (entry.endsWith('.json') && !writer.names.has(entry.slice(0, -'.json'.length))) {
|
||||
unlinkSync(join(writer.dir, entry))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function generateAllFixtures(): Promise<void> {
|
||||
const chat = openSuite('chat')
|
||||
for (const fixtureCase of fixtureCases) {
|
||||
if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(fixtureCase.name)) {
|
||||
throw new Error(`Fixture name must be kebab-case: ${fixtureCase.name}`)
|
||||
}
|
||||
if (names.has(fixtureCase.name)) {
|
||||
throw new Error(`Duplicate fixture name: ${fixtureCase.name}`)
|
||||
}
|
||||
names.add(fixtureCase.name)
|
||||
const document = buildFixtureDocument(fixtureCase)
|
||||
writeFileSync(join(chatDir, `${fixtureCase.name}.json`), toCanonicalJson(document))
|
||||
writeSuiteDocument(chat, fixtureCase.name, buildFixtureDocument(fixtureCase))
|
||||
}
|
||||
pruneSuite(chat)
|
||||
|
||||
// Remove stale fixtures from renamed/deleted cases so natives never keep
|
||||
// passing against a file the web pipeline no longer generates.
|
||||
for (const entry of readdirSync(chatDir)) {
|
||||
if (entry.endsWith('.json') && !names.has(entry.slice(0, -'.json'.length))) {
|
||||
unlinkSync(join(chatDir, entry))
|
||||
}
|
||||
const sse = openSuite('sse')
|
||||
for (const fixtureCase of sseFixtureCases) {
|
||||
writeSuiteDocument(sse, fixtureCase.name, buildSseFixtureDocument(fixtureCase))
|
||||
}
|
||||
pruneSuite(sse)
|
||||
|
||||
const pagination = openSuite('pagination')
|
||||
for (const fixtureCase of paginationFixtureCases) {
|
||||
writeSuiteDocument(pagination, fixtureCase.name, await buildPaginationFixtureDocument(fixtureCase))
|
||||
}
|
||||
pruneSuite(pagination)
|
||||
|
||||
writeFileSync(join(FIXTURES_DIR, 'VERSION'), `${FIXTURE_VERSION}\n`)
|
||||
console.log(`Wrote ${names.size} chat fixtures (fixtureVersion ${FIXTURE_VERSION}) to ${chatDir}`)
|
||||
console.log(
|
||||
`Wrote ${chat.names.size} chat + ${sse.names.size} sse + ${pagination.names.size} pagination fixtures `
|
||||
+ `(fixtureVersion ${FIXTURE_VERSION}) to ${FIXTURES_DIR}`
|
||||
)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
import { DecryptedMessageSchema } from '@hapi/protocol/schemas'
|
||||
import type { DecryptedMessage, MessagesResponse } from '@/types/api'
|
||||
import { FIXTURE_VERSION } from '../fixtureTypes'
|
||||
import { toCanonicalJson } from '../serialize'
|
||||
import {
|
||||
mergeOpObservations,
|
||||
paginationFixtureSessionId,
|
||||
runPaginationScript
|
||||
} from './runner'
|
||||
import type { PaginationFixtureCase, PaginationFixtureDocument, PaginationOp } from './types'
|
||||
|
||||
const MESSAGE_STATUSES = new Set(['queued', 'sending', 'sent', 'failed'])
|
||||
const PAGE_DIRECTIONS = new Set(['latest', 'before', 'after'])
|
||||
const PAGE_KEYS = [
|
||||
'direction', 'limit', 'epoch', 'reset',
|
||||
'nextBeforeAt', 'nextBeforeSeq', 'nextAfterAt', 'nextAfterSeq',
|
||||
'snapshotHeadAt', 'snapshotHeadSeq', 'hasMore'
|
||||
] as const
|
||||
|
||||
function fail(label: string, reason: string): never {
|
||||
throw new Error(`${label}: ${reason}`)
|
||||
}
|
||||
|
||||
/** Validate a wire DecryptedMessage. The protocol schema gates the required
|
||||
* envelope; on top of it only the client-side `status` extension is allowed
|
||||
* so authored inputs cannot smuggle non-wire keys into fixtures. */
|
||||
function validateMessage(message: DecryptedMessage, label: string): void {
|
||||
const result = DecryptedMessageSchema.safeParse(message)
|
||||
if (!result.success) {
|
||||
fail(label, `invalid DecryptedMessage: ${result.error.message}`)
|
||||
}
|
||||
const allowed = new Set([...Object.keys(DecryptedMessageSchema.shape), 'status'])
|
||||
for (const key of Object.keys(message)) {
|
||||
if (!allowed.has(key)) {
|
||||
fail(label, `unexpected DecryptedMessage key '${key}'`)
|
||||
}
|
||||
}
|
||||
if (message.status !== undefined && !MESSAGE_STATUSES.has(message.status)) {
|
||||
fail(label, `invalid status '${String(message.status)}'`)
|
||||
}
|
||||
}
|
||||
|
||||
function isNullableNumber(value: unknown): boolean {
|
||||
return value === null || typeof value === 'number'
|
||||
}
|
||||
|
||||
/** Structural gate for a scripted MessagesResponse (`shared/src/apiTypes.ts`
|
||||
* has no zod schema for it — mirror the wire contract by hand). */
|
||||
function validateResponse(response: MessagesResponse, label: string): void {
|
||||
if (!Array.isArray(response.messages)) {
|
||||
fail(label, 'messages must be an array')
|
||||
}
|
||||
response.messages.forEach((message, index) => validateMessage(message, `${label}.messages[${index}]`))
|
||||
const page = response.page as Record<string, unknown>
|
||||
if (!page || typeof page !== 'object') {
|
||||
fail(label, 'page must be an object')
|
||||
}
|
||||
for (const key of Object.keys(page)) {
|
||||
if (!PAGE_KEYS.includes(key as typeof PAGE_KEYS[number])) {
|
||||
fail(label, `unexpected page key '${key}'`)
|
||||
}
|
||||
}
|
||||
if (!PAGE_DIRECTIONS.has(page.direction as string)) fail(label, 'invalid page.direction')
|
||||
if (typeof page.limit !== 'number' || !Number.isInteger(page.limit)) fail(label, 'page.limit must be an integer')
|
||||
if (typeof page.epoch !== 'number' || !Number.isInteger(page.epoch) || page.epoch < 0) {
|
||||
fail(label, 'page.epoch must be a non-negative integer')
|
||||
}
|
||||
if (typeof page.reset !== 'boolean') fail(label, 'page.reset must be a boolean')
|
||||
if (typeof page.hasMore !== 'boolean') fail(label, 'page.hasMore must be a boolean')
|
||||
for (const key of ['nextBeforeAt', 'nextBeforeSeq', 'nextAfterAt', 'nextAfterSeq', 'snapshotHeadAt', 'snapshotHeadSeq'] as const) {
|
||||
if (!isNullableNumber(page[key])) fail(label, `page.${key} must be a number or null`)
|
||||
}
|
||||
for (const [atKey, seqKey] of [
|
||||
['nextBeforeAt', 'nextBeforeSeq'],
|
||||
['nextAfterAt', 'nextAfterSeq'],
|
||||
['snapshotHeadAt', 'snapshotHeadSeq']
|
||||
] as const) {
|
||||
if ((page[atKey] === null) !== (page[seqKey] === null)) {
|
||||
fail(label, `${atKey}/${seqKey} must be paired (both null or both numbers)`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function validateOp(op: PaginationOp, label: string): void {
|
||||
switch (op.op) {
|
||||
case 'sync-tail':
|
||||
case 'fetch-older':
|
||||
op.responses.forEach((response, index) => validateResponse(response, `${label}.responses[${index}]`))
|
||||
return
|
||||
case 'sse-messages':
|
||||
op.messages.forEach((message, index) => validateMessage(message, `${label}.messages[${index}]`))
|
||||
return
|
||||
case 'append-optimistic':
|
||||
validateMessage(op.message, `${label}.message`)
|
||||
return
|
||||
case 'cancel-invoked':
|
||||
validateMessage(op.message, `${label}.message`)
|
||||
return
|
||||
case 'update-status':
|
||||
case 'messages-consumed':
|
||||
case 'message-cancelled':
|
||||
case 'set-view-mode':
|
||||
case 'queued-state':
|
||||
return
|
||||
default: {
|
||||
const exhaustive: never = op
|
||||
fail(label, `unknown op ${JSON.stringify(exhaustive)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function buildPaginationFixtureDocument(
|
||||
fixtureCase: PaginationFixtureCase
|
||||
): Promise<PaginationFixtureDocument> {
|
||||
// Round-trip the ops through canonical JSON before running, so the
|
||||
// expectations are computed from exactly the bytes the file will carry.
|
||||
const ops = JSON.parse(toCanonicalJson(fixtureCase.ops)) as PaginationOp[]
|
||||
ops.forEach((op, index) => validateOp(op, `${fixtureCase.name}: ops[${index}]`))
|
||||
const { observations, expectedState } = await runPaginationScript(
|
||||
paginationFixtureSessionId(fixtureCase.name),
|
||||
ops
|
||||
)
|
||||
return {
|
||||
fixtureVersion: FIXTURE_VERSION,
|
||||
name: fixtureCase.name,
|
||||
description: fixtureCase.description,
|
||||
ops: mergeOpObservations(ops, observations),
|
||||
expectedState
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,460 @@
|
||||
import type { DecryptedMessage, MessageStatus, MessagesResponse } from '@/types/api'
|
||||
import type { PaginationFixtureCase } from './types'
|
||||
|
||||
/**
|
||||
* Hand-authored window-store scenarios. Positions use small millisecond
|
||||
* values (the protocol only compares them); `at = invokedAt ?? createdAt`
|
||||
* and every cursor is an (at, seq) pair. Cases avoid position ties so the
|
||||
* merge order never depends on the id tie-breaker (web uses localeCompare —
|
||||
* keep ids ASCII and positions distinct).
|
||||
*/
|
||||
|
||||
function agentMessage(init: { id: string; seq: number; at: number }): DecryptedMessage {
|
||||
return {
|
||||
id: init.id,
|
||||
seq: init.seq,
|
||||
localId: null,
|
||||
content: {
|
||||
role: 'agent',
|
||||
content: { type: 'codex', data: { type: 'message', message: init.id } }
|
||||
},
|
||||
createdAt: init.at,
|
||||
invokedAt: init.at
|
||||
}
|
||||
}
|
||||
|
||||
/** A row the chat pipeline hides (normalizes to null): meta system output. */
|
||||
function hiddenAgentMessage(init: { id: string; seq: number; at: number }): DecryptedMessage {
|
||||
return {
|
||||
id: init.id,
|
||||
seq: init.seq,
|
||||
localId: null,
|
||||
content: {
|
||||
role: 'agent',
|
||||
content: { type: 'output', data: { type: 'system', isMeta: true } }
|
||||
},
|
||||
createdAt: init.at,
|
||||
invokedAt: init.at
|
||||
}
|
||||
}
|
||||
|
||||
function userMessage(init: {
|
||||
id: string
|
||||
seq?: number | null
|
||||
localId?: string | null
|
||||
createdAt: number
|
||||
invokedAt?: number | null
|
||||
status?: MessageStatus
|
||||
text?: string
|
||||
}): DecryptedMessage {
|
||||
return {
|
||||
id: init.id,
|
||||
seq: init.seq ?? null,
|
||||
localId: init.localId ?? null,
|
||||
content: {
|
||||
role: 'user',
|
||||
content: { type: 'text', text: init.text ?? init.id }
|
||||
},
|
||||
createdAt: init.createdAt,
|
||||
...(init.invokedAt !== undefined ? { invokedAt: init.invokedAt } : {}),
|
||||
...(init.status !== undefined ? { status: init.status } : {})
|
||||
}
|
||||
}
|
||||
|
||||
type Position = { at: number; seq: number }
|
||||
|
||||
function pageResponse(
|
||||
messages: DecryptedMessage[],
|
||||
page: {
|
||||
direction: 'latest' | 'before' | 'after'
|
||||
epoch: number
|
||||
hasMore: boolean
|
||||
reset?: boolean
|
||||
nextBefore?: Position | null
|
||||
nextAfter?: Position | null
|
||||
snapshotHead?: Position | null
|
||||
}
|
||||
): MessagesResponse {
|
||||
return {
|
||||
messages,
|
||||
page: {
|
||||
direction: page.direction,
|
||||
limit: 200,
|
||||
epoch: page.epoch,
|
||||
reset: page.reset ?? false,
|
||||
nextBeforeAt: page.nextBefore?.at ?? null,
|
||||
nextBeforeSeq: page.nextBefore?.seq ?? null,
|
||||
nextAfterAt: page.nextAfter?.at ?? null,
|
||||
nextAfterSeq: page.nextAfter?.seq ?? null,
|
||||
snapshotHeadAt: page.snapshotHead?.at ?? null,
|
||||
snapshotHeadSeq: page.snapshotHead?.seq ?? null,
|
||||
hasMore: page.hasMore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function paddedAgentRun(fromSeq: number, toSeq: number): DecryptedMessage[] {
|
||||
const rows: DecryptedMessage[] = []
|
||||
for (let seq = fromSeq; seq <= toSeq; seq += 1) {
|
||||
rows.push(agentMessage({ id: `a-${String(seq).padStart(3, '0')}`, seq, at: seq * 1000 }))
|
||||
}
|
||||
return rows
|
||||
}
|
||||
|
||||
export const paginationFixtureCases: PaginationFixtureCase[] = [
|
||||
{
|
||||
name: 'latest-page-then-sse-ingest',
|
||||
description: 'Cold start: a latest page seeds the window, epoch, older cursor (page.nextBefore) and newest cursor (page.snapshotHead); a live SSE message appends and advances the newest cursor to its position.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([
|
||||
agentMessage({ id: 'a-1', seq: 1, at: 1_000 }),
|
||||
agentMessage({ id: 'a-2', seq: 2, at: 2_000 })
|
||||
], {
|
||||
direction: 'latest',
|
||||
epoch: 0,
|
||||
hasMore: true,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 2_000, seq: 2 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'sse-messages',
|
||||
messages: [agentMessage({ id: 'a-3', seq: 3, at: 3_000 })]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'fetch-older-before-cursor',
|
||||
description: 'Older pagination sends the compound (beforeAt, beforeSeq) pair from the current older cursor, prepends the rows, and adopts the response page.nextBefore as the new older cursor and page.hasMore as the exhaustion flag.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([agentMessage({ id: 'a-10', seq: 10, at: 10_000 })], {
|
||||
direction: 'latest',
|
||||
epoch: 4,
|
||||
hasMore: true,
|
||||
nextBefore: { at: 10_000, seq: 10 },
|
||||
snapshotHead: { at: 10_000, seq: 10 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'fetch-older',
|
||||
responses: [
|
||||
pageResponse([agentMessage({ id: 'a-9', seq: 9, at: 9_000 })], {
|
||||
direction: 'before',
|
||||
epoch: 4,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 9_000, seq: 9 }
|
||||
})
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'older-page-epoch-mismatch-resets',
|
||||
description: 'An older page answering with a different epoch invalidates every cursor: the load stops with epoch-reset, the window is discarded, and a fresh latest request (issued by the store within the same operation) replaces it wholesale.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([agentMessage({ id: 'a-10', seq: 10, at: 10_000 })], {
|
||||
direction: 'latest',
|
||||
epoch: 1,
|
||||
hasMore: true,
|
||||
nextBefore: { at: 10_000, seq: 10 },
|
||||
snapshotHead: { at: 10_000, seq: 10 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'fetch-older',
|
||||
responses: [
|
||||
pageResponse([], {
|
||||
direction: 'before',
|
||||
epoch: 2,
|
||||
hasMore: false,
|
||||
nextBefore: null
|
||||
}),
|
||||
pageResponse([agentMessage({ id: 'f-20', seq: 20, at: 20_000 })], {
|
||||
direction: 'latest',
|
||||
epoch: 2,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 20_000, seq: 20 },
|
||||
snapshotHead: { at: 20_000, seq: 20 }
|
||||
})
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'reset-latest-preserves-optimistic',
|
||||
description: 'A tail sync that gets page.reset:true replaces every server row from before the request but keeps optimistic rows (id === localId): the not-yet-echoed send survives the epoch bump and the window adopts the new epoch and cursors.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([agentMessage({ id: 'old-1', seq: 1, at: 1_000 })], {
|
||||
direction: 'latest',
|
||||
epoch: 1,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 1_000, seq: 1 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'append-optimistic',
|
||||
message: userMessage({
|
||||
id: 'local-1',
|
||||
localId: 'local-1',
|
||||
createdAt: 1_500,
|
||||
invokedAt: null,
|
||||
status: 'sending',
|
||||
text: 'optimistic send'
|
||||
})
|
||||
},
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([agentMessage({ id: 'fresh-2', seq: 2, at: 2_000 })], {
|
||||
direction: 'latest',
|
||||
epoch: 2,
|
||||
hasMore: false,
|
||||
reset: true,
|
||||
nextBefore: { at: 2_000, seq: 2 },
|
||||
snapshotHead: { at: 2_000, seq: 2 }
|
||||
})
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'optimistic-echo-reconciles-by-localid',
|
||||
description: 'The message-received echo of the caller\'s own send (server id, real seq, same localId) replaces the optimistic row while preserving the client-side status; only one row remains and it is no longer optimistic.',
|
||||
ops: [
|
||||
{
|
||||
op: 'append-optimistic',
|
||||
message: userMessage({
|
||||
id: 'local-1',
|
||||
localId: 'local-1',
|
||||
createdAt: 1_000,
|
||||
invokedAt: null,
|
||||
status: 'sending',
|
||||
text: 'hello agent'
|
||||
})
|
||||
},
|
||||
{ op: 'update-status', localId: 'local-1', status: 'queued' },
|
||||
{
|
||||
op: 'sse-messages',
|
||||
messages: [
|
||||
userMessage({
|
||||
id: 'srv-1',
|
||||
seq: 1,
|
||||
localId: 'local-1',
|
||||
createdAt: 1_000,
|
||||
invokedAt: null,
|
||||
text: 'hello agent'
|
||||
})
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'messages-consumed-stamps-invoked-at',
|
||||
description: 'messages-consumed stamps invokedAt and flips status to sent: the row leaves the queued bar and moves to its invocation position (after the agent row), while the newest cursor does NOT advance from this out-of-band update.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([
|
||||
userMessage({
|
||||
id: 'q-1',
|
||||
seq: 1,
|
||||
localId: 'local-1',
|
||||
createdAt: 1_000,
|
||||
invokedAt: null,
|
||||
text: 'queued question'
|
||||
}),
|
||||
agentMessage({ id: 'a-2', seq: 2, at: 2_000 })
|
||||
], {
|
||||
direction: 'latest',
|
||||
epoch: 0,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 2_000, seq: 2 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{ op: 'messages-consumed', localIds: ['local-1'], invokedAt: 3_000 }
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'message-cancelled-removes-queued-row',
|
||||
description: 'message-cancelled removes the queued row by localId; a repeat delivery is an idempotent no-op on the already-empty window.',
|
||||
ops: [
|
||||
{
|
||||
op: 'append-optimistic',
|
||||
message: userMessage({
|
||||
id: 'local-1',
|
||||
localId: 'local-1',
|
||||
createdAt: 1_000,
|
||||
invokedAt: null,
|
||||
status: 'queued',
|
||||
text: 'cancel me'
|
||||
})
|
||||
},
|
||||
{ op: 'message-cancelled', localId: 'local-1' },
|
||||
{ op: 'message-cancelled', localId: 'local-1' }
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'cancel-too-late-ingests-invoked-row',
|
||||
description: 'DELETE answered status:invoked (the agent consumed the message before the cancel landed): the queued snapshot must NOT be resurrected — the returned authoritative row is ingested with its server invokedAt and client status sent, landing in the thread at its invocation position.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([
|
||||
userMessage({
|
||||
id: 'srv-1',
|
||||
seq: 1,
|
||||
localId: 'local-1',
|
||||
createdAt: 1_000,
|
||||
invokedAt: null,
|
||||
text: 'race with the agent'
|
||||
})
|
||||
], {
|
||||
direction: 'latest',
|
||||
epoch: 0,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 1_000, seq: 1 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'cancel-invoked',
|
||||
localId: 'local-1',
|
||||
message: userMessage({
|
||||
id: 'srv-1',
|
||||
seq: 1,
|
||||
localId: 'local-1',
|
||||
createdAt: 1_000,
|
||||
invokedAt: 5_000,
|
||||
text: 'race with the agent'
|
||||
})
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'sse-hidden-rows-advance-cursor-only',
|
||||
description: 'SSE rows the chat pipeline hides (e.g. meta system output) are not retained in the window, but the newest cursor still advances past their position so a later tail sync does not refetch them.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([agentMessage({ id: 'a-1', seq: 1, at: 1_000 })], {
|
||||
direction: 'latest',
|
||||
epoch: 0,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 1_000, seq: 1 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'sse-messages',
|
||||
messages: [
|
||||
agentMessage({ id: 'a-2', seq: 2, at: 2_000 }),
|
||||
hiddenAgentMessage({ id: 'hidden-3', seq: 3, at: 3_000 })
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'trim-preserves-queued-and-recomputes-cursor',
|
||||
description: 'Tail-mode overflow past the 400-row visible window trims the oldest regular rows: queued rows are never trimmed (the regular budget shrinks by their count), hasMore flips true, and the older cursor is recomputed from the oldest kept row.',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([
|
||||
...paddedAgentRun(1, 199),
|
||||
userMessage({
|
||||
id: 'q-200',
|
||||
seq: 200,
|
||||
localId: 'local-q',
|
||||
createdAt: 350_000,
|
||||
invokedAt: null,
|
||||
text: 'still queued'
|
||||
})
|
||||
], {
|
||||
direction: 'latest',
|
||||
epoch: 0,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 350_000, seq: 200 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'sse-messages',
|
||||
messages: paddedAgentRun(201, 402)
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'queued-state-reconciliation-drops-deleted',
|
||||
description: 'Queued-state recovery after a resume gap: candidates are the user rows with invokedAt strictly null; the server verdict stamps invoked ones like messages-consumed, keeps the still-queued one, and drops candidates in neither list (deleted server-side).',
|
||||
ops: [
|
||||
{
|
||||
op: 'sync-tail',
|
||||
responses: [
|
||||
pageResponse([
|
||||
userMessage({
|
||||
id: 'u-1',
|
||||
seq: 1,
|
||||
localId: 'local-stale',
|
||||
createdAt: 1_000,
|
||||
invokedAt: null,
|
||||
text: 'deleted server-side'
|
||||
}),
|
||||
userMessage({
|
||||
id: 'u-2',
|
||||
seq: 2,
|
||||
localId: 'local-kept',
|
||||
createdAt: 2_000,
|
||||
invokedAt: null,
|
||||
text: 'still queued'
|
||||
}),
|
||||
userMessage({
|
||||
id: 'u-3',
|
||||
seq: 3,
|
||||
localId: 'local-consumed',
|
||||
createdAt: 3_000,
|
||||
invokedAt: null,
|
||||
text: 'consumed during the gap'
|
||||
})
|
||||
], {
|
||||
direction: 'latest',
|
||||
epoch: 0,
|
||||
hasMore: false,
|
||||
nextBefore: { at: 1_000, seq: 1 },
|
||||
snapshotHead: { at: 3_000, seq: 3 }
|
||||
})
|
||||
]
|
||||
},
|
||||
{
|
||||
op: 'queued-state',
|
||||
queuedLocalIds: ['local-kept'],
|
||||
invoked: [{ localId: 'local-consumed', invokedAt: 4_000 }]
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
@@ -0,0 +1,252 @@
|
||||
import type { ApiClient } from '@/api/client'
|
||||
import {
|
||||
clearMessageWindow,
|
||||
fetchOlderMessages,
|
||||
getMessageWindowState,
|
||||
getQueuedReconcileCandidateLocalIds,
|
||||
ingestIncomingMessages,
|
||||
appendOptimisticMessage,
|
||||
markMessagesConsumed,
|
||||
reconcileQueuedLocalIds,
|
||||
removeOptimisticMessage,
|
||||
setMessageViewMode,
|
||||
syncTailMessages,
|
||||
updateMessageStatus,
|
||||
type MessageWindowState,
|
||||
type OlderLoadOutcome
|
||||
} from '@/lib/message-window-store'
|
||||
import { isQueuedForInvocation } from '@/lib/messages'
|
||||
import type { DecryptedMessage, MessagesResponse } from '@/types/api'
|
||||
import { toCanonicalJson } from '../serialize'
|
||||
import type {
|
||||
PageRequest,
|
||||
PaginationOp,
|
||||
PaginationOpObservation,
|
||||
ProjectedOlderOutcome,
|
||||
ProjectedWindowMessage,
|
||||
ProjectedWindowState
|
||||
} from './types'
|
||||
|
||||
/**
|
||||
* The op-script harness: drives the REAL web message-window store with a
|
||||
* scripted ApiClient and records what the store did (requests issued, older
|
||||
* -load outcomes, reconcile candidates) plus the final normative projection.
|
||||
* Shared by the fixture generator and the self-conformance vitest so the
|
||||
* stored expectations can never diverge from a fresh replay.
|
||||
*
|
||||
* Ops run strictly sequentially (each awaited to completion). Determinism:
|
||||
* the store's only wall-clock reads (`Date.now`) gate notification
|
||||
* throttling, which never touches the state this harness observes.
|
||||
*/
|
||||
|
||||
export function paginationFixtureSessionId(name: string): string {
|
||||
return `fixture-pagination-${name}`
|
||||
}
|
||||
|
||||
type ScriptedApi = {
|
||||
api: ApiClient
|
||||
requests: PageRequest[]
|
||||
queue: MessagesResponse[]
|
||||
}
|
||||
|
||||
function createScriptedApi(responses: MessagesResponse[]): ScriptedApi {
|
||||
const scripted: ScriptedApi = {
|
||||
requests: [],
|
||||
queue: [...responses],
|
||||
api: undefined as unknown as ApiClient
|
||||
}
|
||||
const getMessages: ApiClient['getMessages'] = async (_sessionId, options) => {
|
||||
scripted.requests.push(JSON.parse(toCanonicalJson(options ?? {})) as PageRequest)
|
||||
const next = scripted.queue.shift()
|
||||
if (!next) {
|
||||
throw new Error('scripted ApiClient exhausted: unexpected getMessages request')
|
||||
}
|
||||
return next
|
||||
}
|
||||
scripted.api = { getMessages } as ApiClient
|
||||
return scripted
|
||||
}
|
||||
|
||||
function projectOutcome(outcome: OlderLoadOutcome): ProjectedOlderOutcome {
|
||||
if (outcome.kind === 'applied') {
|
||||
return {
|
||||
kind: 'applied',
|
||||
hasMore: outcome.hasMore,
|
||||
addedRenderableCount: outcome.addedRenderableCount
|
||||
}
|
||||
}
|
||||
if (outcome.kind === 'stopped') {
|
||||
return { kind: 'stopped', reason: outcome.reason }
|
||||
}
|
||||
throw new Error(`older-page load failed: ${outcome.error.message}`)
|
||||
}
|
||||
|
||||
function projectMessage(message: DecryptedMessage): ProjectedWindowMessage {
|
||||
return {
|
||||
id: message.id,
|
||||
localId: message.localId ?? null,
|
||||
seq: message.seq ?? null,
|
||||
createdAt: message.createdAt,
|
||||
...(message.invokedAt !== undefined ? { invokedAt: message.invokedAt } : {}),
|
||||
...(message.scheduledAt !== undefined ? { scheduledAt: message.scheduledAt } : {}),
|
||||
...(message.status !== undefined ? { status: message.status } : {}),
|
||||
queued: isQueuedForInvocation(message),
|
||||
optimistic: Boolean(message.localId && message.id === message.localId)
|
||||
}
|
||||
}
|
||||
|
||||
/** The store's compound paging cursors live on the internal state object
|
||||
* behind `MessageWindowState`. The `in` guards fail loudly if the web store
|
||||
* renames them, so the fixtures cannot silently pin nulls. */
|
||||
type InternalCursorFields = {
|
||||
oldestPositionAt: number | null
|
||||
oldestPositionSeq: number | null
|
||||
newestPositionAt: number | null
|
||||
newestPositionSeq: number | null
|
||||
}
|
||||
|
||||
function projectCursor(at: unknown, seq: unknown): { at: number; seq: number } | null {
|
||||
return typeof at === 'number' && typeof seq === 'number' ? { at, seq } : null
|
||||
}
|
||||
|
||||
export function projectWindowState(sessionId: string): ProjectedWindowState {
|
||||
const state = getMessageWindowState(sessionId) as MessageWindowState & Partial<InternalCursorFields>
|
||||
if (!('oldestPositionAt' in state) || !('newestPositionAt' in state)) {
|
||||
throw new Error('message-window-store internals renamed: update the pagination fixture runner')
|
||||
}
|
||||
return {
|
||||
messages: state.messages.map(projectMessage),
|
||||
hasMore: state.hasMore,
|
||||
epoch: state.epoch,
|
||||
viewMode: state.viewMode,
|
||||
olderCursor: projectCursor(state.oldestPositionAt, state.oldestPositionSeq),
|
||||
newestCursor: projectCursor(state.newestPositionAt, state.newestPositionSeq)
|
||||
}
|
||||
}
|
||||
|
||||
function assertSettled(sessionId: string, opIndex: number, scripted: ScriptedApi): void {
|
||||
if (scripted.queue.length > 0) {
|
||||
throw new Error(`ops[${opIndex}]: ${scripted.queue.length} scripted response(s) left unconsumed`)
|
||||
}
|
||||
const state = getMessageWindowState(sessionId)
|
||||
if (state.warning !== null) {
|
||||
throw new Error(`ops[${opIndex}]: store reported a warning: ${state.warning}`)
|
||||
}
|
||||
if (state.isSyncingTail || state.isLoadingMore) {
|
||||
throw new Error(`ops[${opIndex}]: store still busy after the op settled`)
|
||||
}
|
||||
}
|
||||
|
||||
async function executeOp(
|
||||
sessionId: string,
|
||||
op: PaginationOp,
|
||||
opIndex: number
|
||||
): Promise<PaginationOpObservation> {
|
||||
switch (op.op) {
|
||||
case 'sync-tail': {
|
||||
const scripted = createScriptedApi(op.responses)
|
||||
await syncTailMessages(scripted.api, sessionId)
|
||||
assertSettled(sessionId, opIndex, scripted)
|
||||
return { requests: scripted.requests }
|
||||
}
|
||||
case 'fetch-older': {
|
||||
const scripted = createScriptedApi(op.responses)
|
||||
const outcome = await fetchOlderMessages(scripted.api, sessionId)
|
||||
assertSettled(sessionId, opIndex, scripted)
|
||||
return { requests: scripted.requests, outcome: projectOutcome(outcome) }
|
||||
}
|
||||
case 'sse-messages':
|
||||
ingestIncomingMessages(sessionId, op.messages)
|
||||
return {}
|
||||
case 'append-optimistic':
|
||||
appendOptimisticMessage(sessionId, op.message)
|
||||
return {}
|
||||
case 'update-status':
|
||||
updateMessageStatus(sessionId, op.localId, op.status)
|
||||
return {}
|
||||
case 'messages-consumed':
|
||||
markMessagesConsumed(sessionId, op.localIds, op.invokedAt)
|
||||
return {}
|
||||
case 'message-cancelled':
|
||||
removeOptimisticMessage(sessionId, op.localId)
|
||||
return {}
|
||||
case 'cancel-invoked':
|
||||
// Mirrors useCancelQueuedMessage: optimistic removal on mutate,
|
||||
// then the invoked-race response re-ingests the server row as
|
||||
// status 'sent' so the chip lands in the thread, not the bar.
|
||||
removeOptimisticMessage(sessionId, op.localId)
|
||||
appendOptimisticMessage(sessionId, { ...op.message, status: 'sent' })
|
||||
return {}
|
||||
case 'set-view-mode':
|
||||
setMessageViewMode(sessionId, op.mode)
|
||||
return {}
|
||||
case 'queued-state': {
|
||||
// Mirrors reconcileQueuedStateAfterConnect (post tail-sync half):
|
||||
// collect candidates, apply invoked verdicts grouped by
|
||||
// timestamp, then drop candidates in neither list.
|
||||
const candidates = getQueuedReconcileCandidateLocalIds(sessionId)
|
||||
const invokedByTimestamp = new Map<number, string[]>()
|
||||
for (const entry of op.invoked) {
|
||||
const localIds = invokedByTimestamp.get(entry.invokedAt) ?? []
|
||||
localIds.push(entry.localId)
|
||||
invokedByTimestamp.set(entry.invokedAt, localIds)
|
||||
}
|
||||
for (const [invokedAt, localIds] of invokedByTimestamp) {
|
||||
markMessagesConsumed(sessionId, localIds, invokedAt)
|
||||
}
|
||||
reconcileQueuedLocalIds(sessionId, candidates, op.queuedLocalIds)
|
||||
return { candidates }
|
||||
}
|
||||
default: {
|
||||
const exhaustive: never = op
|
||||
throw new Error(`unknown op: ${JSON.stringify(exhaustive)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function runPaginationScript(
|
||||
sessionId: string,
|
||||
ops: PaginationOp[]
|
||||
): Promise<{ observations: PaginationOpObservation[]; expectedState: ProjectedWindowState }> {
|
||||
clearMessageWindow(sessionId)
|
||||
const observations: PaginationOpObservation[] = []
|
||||
for (const [opIndex, op] of ops.entries()) {
|
||||
observations.push(await executeOp(sessionId, op, opIndex))
|
||||
}
|
||||
return { observations, expectedState: projectWindowState(sessionId) }
|
||||
}
|
||||
|
||||
/** Fill the machine-generated `expected*` fields into the authored ops. */
|
||||
export function mergeOpObservations(
|
||||
ops: PaginationOp[],
|
||||
observations: PaginationOpObservation[]
|
||||
): PaginationOp[] {
|
||||
return ops.map((op, index) => {
|
||||
const observation = observations[index] ?? {}
|
||||
return {
|
||||
...op,
|
||||
...(observation.requests !== undefined ? { expectedRequests: observation.requests } : {}),
|
||||
...(observation.outcome !== undefined ? { expectedOutcome: observation.outcome } : {}),
|
||||
...(observation.candidates !== undefined ? { expectedCandidates: observation.candidates } : {})
|
||||
} as PaginationOp
|
||||
})
|
||||
}
|
||||
|
||||
/** Read the stored `expected*` fields back out, shaped like fresh
|
||||
* observations, for exact-match comparison in the conformance tests. */
|
||||
export function extractOpExpectations(ops: PaginationOp[]): PaginationOpObservation[] {
|
||||
return ops.map((op) => {
|
||||
const record = op as Record<string, unknown>
|
||||
return {
|
||||
...(record.expectedRequests !== undefined
|
||||
? { requests: record.expectedRequests as PageRequest[] }
|
||||
: {}),
|
||||
...(record.expectedOutcome !== undefined
|
||||
? { outcome: record.expectedOutcome as ProjectedOlderOutcome }
|
||||
: {}),
|
||||
...(record.expectedCandidates !== undefined
|
||||
? { candidates: record.expectedCandidates as string[] }
|
||||
: {})
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
import type { DecryptedMessage, MessageStatus, MessagesResponse } from '@/types/api'
|
||||
|
||||
/** Exact argument object the store passed to `getMessages` (canonicalized:
|
||||
* sorted keys, `undefined` entries stripped, `null` entries preserved). */
|
||||
export type PageRequest = Record<string, number | null>
|
||||
|
||||
/** Projection of `fetchOlderMessages`' outcome. `historyVersion` (a web
|
||||
* render counter) is dropped; `failed` outcomes are never scripted. */
|
||||
export type ProjectedOlderOutcome =
|
||||
| { kind: 'applied'; hasMore: boolean; addedRenderableCount: number }
|
||||
| { kind: 'stopped'; reason: 'unavailable' | 'busy' | 'invalidated' | 'epoch-reset' | 'exhausted' }
|
||||
|
||||
/**
|
||||
* One scripted operation against the message-window store. Fields named
|
||||
* `expected*` are machine-generated observations of the real web store —
|
||||
* absent from hand-authored cases, present in the emitted document.
|
||||
*/
|
||||
export type PaginationOp =
|
||||
| {
|
||||
/** Run a full tail sync (`syncTailMessages`). `responses` feed the
|
||||
* scripted ApiClient in order; every response must be consumed. */
|
||||
op: 'sync-tail'
|
||||
responses: MessagesResponse[]
|
||||
expectedRequests?: PageRequest[]
|
||||
}
|
||||
| {
|
||||
/** Load one older page (`fetchOlderMessages`). On an epoch mismatch
|
||||
* the store runs an internal tail sync, whose responses (and
|
||||
* recorded requests) belong to this same op. */
|
||||
op: 'fetch-older'
|
||||
responses: MessagesResponse[]
|
||||
expectedRequests?: PageRequest[]
|
||||
expectedOutcome?: ProjectedOlderOutcome
|
||||
}
|
||||
| {
|
||||
/** SSE `message-received` delivery (`ingestIncomingMessages`). */
|
||||
op: 'sse-messages'
|
||||
messages: DecryptedMessage[]
|
||||
}
|
||||
| {
|
||||
/** Local optimistic send append (`appendOptimisticMessage`). The
|
||||
* message is optimistic iff `id === localId`. */
|
||||
op: 'append-optimistic'
|
||||
message: DecryptedMessage
|
||||
}
|
||||
| {
|
||||
/** Client-side send-state transition (`updateMessageStatus`). */
|
||||
op: 'update-status'
|
||||
localId: string
|
||||
status: MessageStatus
|
||||
}
|
||||
| {
|
||||
/** SSE `messages-consumed` event (`markMessagesConsumed`). */
|
||||
op: 'messages-consumed'
|
||||
localIds: string[]
|
||||
invokedAt: number
|
||||
}
|
||||
| {
|
||||
/** SSE `message-cancelled` event, or the optimistic removal before a
|
||||
* DELETE (`removeOptimisticMessage`; matches localId OR id). */
|
||||
op: 'message-cancelled'
|
||||
localId: string
|
||||
}
|
||||
| {
|
||||
/** Client handling of a DELETE answered `{"status":"invoked"}`:
|
||||
* remove the queued row by localId, then re-ingest the returned
|
||||
* authoritative message with client status `sent` (the harness adds
|
||||
* the status, mirroring the web mutation). */
|
||||
op: 'cancel-invoked'
|
||||
localId: string
|
||||
message: DecryptedMessage
|
||||
}
|
||||
| {
|
||||
/** Switch view mode (`setMessageViewMode`). */
|
||||
op: 'set-view-mode'
|
||||
mode: 'tail' | 'history'
|
||||
}
|
||||
| {
|
||||
/** Queued-state recovery round trip after a `resume: 'gap'`:
|
||||
* candidates are collected from the window
|
||||
* (`getQueuedReconcileCandidateLocalIds` — pinned in
|
||||
* `expectedCandidates`), the scripted server verdict is applied via
|
||||
* `markMessagesConsumed` per invoked entry, then
|
||||
* `reconcileQueuedLocalIds` drops candidates in neither list. */
|
||||
op: 'queued-state'
|
||||
queuedLocalIds: string[]
|
||||
invoked: Array<{ localId: string; invokedAt: number }>
|
||||
expectedCandidates?: string[]
|
||||
}
|
||||
|
||||
/** Observation of one executed op, recorded from the real store. */
|
||||
export type PaginationOpObservation = {
|
||||
requests?: PageRequest[]
|
||||
outcome?: ProjectedOlderOutcome
|
||||
candidates?: string[]
|
||||
}
|
||||
|
||||
/** Normative projection of one window row. `invokedAt` / `scheduledAt`
|
||||
* mirror the wire tri-state (absent / null / number); `status` is the
|
||||
* client-side optimistic send state when present. `queued` and `optimistic`
|
||||
* are the derived predicates natives must reproduce. */
|
||||
export type ProjectedWindowMessage = {
|
||||
id: string
|
||||
localId: string | null
|
||||
seq: number | null
|
||||
createdAt: number
|
||||
invokedAt?: number | null
|
||||
scheduledAt?: number | null
|
||||
status?: MessageStatus
|
||||
queued: boolean
|
||||
optimistic: boolean
|
||||
}
|
||||
|
||||
/** Normative projection of the window state after all ops. The cursors are
|
||||
* the store's compound paging positions: `olderCursor` feeds the next
|
||||
* before-request, `newestCursor` the next after-request. */
|
||||
export type ProjectedWindowState = {
|
||||
messages: ProjectedWindowMessage[]
|
||||
hasMore: boolean
|
||||
epoch: number | null
|
||||
viewMode: 'tail' | 'history'
|
||||
olderCursor: { at: number; seq: number } | null
|
||||
newestCursor: { at: number; seq: number } | null
|
||||
}
|
||||
|
||||
/**
|
||||
* A hand-authored pagination case: op scripts with wire-shaped inputs only.
|
||||
* The emitted document carries the same ops with `expected*` observations
|
||||
* filled in, plus the final `expectedState` — all machine-generated by
|
||||
* driving the real web store.
|
||||
*/
|
||||
export type PaginationFixtureCase = {
|
||||
/** Kebab-case name; becomes shared/fixtures/pagination/<name>.json */
|
||||
name: string
|
||||
description: string
|
||||
ops: PaginationOp[]
|
||||
}
|
||||
|
||||
export type PaginationFixtureDocument = {
|
||||
fixtureVersion: number
|
||||
name: string
|
||||
description: string
|
||||
ops: PaginationOp[]
|
||||
expectedState: ProjectedWindowState
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
import { applySessionDetailPatch } from '@/lib/sessionPatch'
|
||||
import type { Session, SessionPatch } from '@/types/api'
|
||||
import type { SessionPatchResult } from './types'
|
||||
|
||||
/**
|
||||
* The exact fold a client must implement for `session-updated` patches on a
|
||||
* detail session cache: apply patches in arrival order; a `null` return means
|
||||
* "nothing changed, keep the previous object". Shared by the fixture
|
||||
* generator and the self-conformance vitest so the two can never diverge.
|
||||
*/
|
||||
export function runSessionPatchScript(
|
||||
initialSession: Session,
|
||||
patches: SessionPatch[]
|
||||
): { expectedPatchResults: SessionPatchResult[]; expectedSession: Session } {
|
||||
let session = initialSession
|
||||
const expectedPatchResults: SessionPatchResult[] = []
|
||||
for (const patch of patches) {
|
||||
const next = applySessionDetailPatch(session, patch)
|
||||
expectedPatchResults.push(next === null ? 'unchanged' : 'applied')
|
||||
session = next ?? session
|
||||
}
|
||||
return { expectedPatchResults, expectedSession: session }
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
import { SessionPatchSchema, SessionSchema } from '@hapi/protocol/schemas'
|
||||
import type { Session, SessionPatch } from '@/types/api'
|
||||
import { FIXTURE_VERSION } from '../fixtureTypes'
|
||||
import { toCanonicalJson } from '../serialize'
|
||||
import { runSessionPatchScript } from './apply'
|
||||
import type { SseFixtureCase, SseFixtureDocument } from './types'
|
||||
|
||||
/** Parse `value` with `schema` and require the stored (parsed) form to be a
|
||||
* fixed point of the schema: parsing it again must be byte-identical. This
|
||||
* guarantees natives can decode the document verbatim without re-running
|
||||
* zod defaults/transforms, and that authored inputs cannot drift from the
|
||||
* wire schema. */
|
||||
function parseToNormalForm<T>(
|
||||
schema: { parse: (input: unknown) => T },
|
||||
value: unknown,
|
||||
label: string
|
||||
): T {
|
||||
const parsed = schema.parse(JSON.parse(toCanonicalJson(value)))
|
||||
const reparsed = schema.parse(JSON.parse(toCanonicalJson(parsed)))
|
||||
if (toCanonicalJson(reparsed) !== toCanonicalJson(parsed)) {
|
||||
throw new Error(`${label} is not schema-normalized: author it in parsed form`)
|
||||
}
|
||||
return JSON.parse(toCanonicalJson(parsed)) as T
|
||||
}
|
||||
|
||||
export function buildSseFixtureDocument(fixtureCase: SseFixtureCase): SseFixtureDocument {
|
||||
const initialSession = parseToNormalForm<Session>(
|
||||
SessionSchema,
|
||||
fixtureCase.initialSession,
|
||||
`${fixtureCase.name}: initialSession`
|
||||
)
|
||||
const patches = fixtureCase.patches.map((patch, index) => {
|
||||
const parsed = parseToNormalForm<SessionPatch>(
|
||||
SessionPatchSchema,
|
||||
patch,
|
||||
`${fixtureCase.name}: patches[${index}]`
|
||||
)
|
||||
if (Object.keys(parsed).length === 0) {
|
||||
throw new Error(`${fixtureCase.name}: patches[${index}] is empty`)
|
||||
}
|
||||
return parsed
|
||||
})
|
||||
const { expectedPatchResults, expectedSession } = runSessionPatchScript(initialSession, patches)
|
||||
// Patch application must never produce a schema-invalid session.
|
||||
SessionSchema.parse(JSON.parse(toCanonicalJson(expectedSession)))
|
||||
return {
|
||||
fixtureVersion: FIXTURE_VERSION,
|
||||
name: fixtureCase.name,
|
||||
description: fixtureCase.description,
|
||||
initialSession,
|
||||
patches,
|
||||
expectedPatchResults,
|
||||
expectedSession: JSON.parse(toCanonicalJson(expectedSession)) as Session
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,291 @@
|
||||
import type { Session } from '@/types/api'
|
||||
import { T0 } from '../cases/support'
|
||||
import type { SseFixtureCase } from './types'
|
||||
|
||||
/**
|
||||
* Hand-authored versioned-patch scenarios. Every case pins how
|
||||
* `applySessionDetailPatch` treats a `session-updated` SessionPatch against a
|
||||
* cached detail Session: flat fields are last-write-wins, `updatedAt` is
|
||||
* max-monotonic, and the four versioned wrappers (metadata / agentState /
|
||||
* todos / teamState) apply only when strictly newer than the cached
|
||||
* watermark. Timestamps derive from the fixed T0 so output is deterministic.
|
||||
*/
|
||||
|
||||
/** Baseline cached session; cases override the fields under test. All
|
||||
* schema-defaulted fields are explicit so the stored document equals its own
|
||||
* `SessionSchema.parse` (asserted by the generator). */
|
||||
function baseSession(overrides: Partial<Session> = {}): Session {
|
||||
return {
|
||||
id: 'sess-sse-fixture',
|
||||
namespace: 'default',
|
||||
seq: 12,
|
||||
createdAt: T0 - 3_600_000,
|
||||
updatedAt: T0,
|
||||
active: true,
|
||||
activeAt: T0,
|
||||
metadata: {
|
||||
path: '/home/dev/project',
|
||||
host: 'devbox',
|
||||
name: 'project',
|
||||
flavor: 'claude'
|
||||
},
|
||||
metadataVersion: 3,
|
||||
agentState: null,
|
||||
agentStateVersion: 5,
|
||||
thinking: false,
|
||||
thinkingAt: 0,
|
||||
activeTurnStartedAt: null,
|
||||
model: 'claude-sonnet-4-6',
|
||||
modelReasoningEffort: null,
|
||||
effort: null,
|
||||
serviceTier: null,
|
||||
permissionMode: 'default',
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
export const sseFixtureCases: SseFixtureCase[] = [
|
||||
{
|
||||
name: 'metadata-newer-version-applied',
|
||||
description: 'A metadata wrapper with version strictly greater than the cached metadataVersion replaces the value and stores the version; the accompanying newer updatedAt advances.',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{
|
||||
updatedAt: T0 + 10_000,
|
||||
metadata: {
|
||||
version: 4,
|
||||
value: {
|
||||
path: '/home/dev/project',
|
||||
host: 'devbox',
|
||||
name: 'project',
|
||||
flavor: 'claude',
|
||||
summary: { text: 'Add retry to upload path', updatedAt: T0 + 9_000 }
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'metadata-equal-version-dropped',
|
||||
description: 'A metadata wrapper whose version equals the cached metadataVersion is dropped even though its value differs — the gate is strictly-greater on version, never a value comparison (dual SSE connections can replay the same version).',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{
|
||||
metadata: {
|
||||
version: 3,
|
||||
value: {
|
||||
path: '/home/dev/project',
|
||||
host: 'devbox',
|
||||
name: 'renamed-should-not-apply',
|
||||
flavor: 'claude'
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'agent-state-stale-version-dropped',
|
||||
description: 'An agentState wrapper older than the cached agentStateVersion is rejected: applying it would resurrect a permission request that was already resolved into completedRequests.',
|
||||
initialSession: baseSession({
|
||||
agentState: {
|
||||
requests: null,
|
||||
completedRequests: {
|
||||
'req-1': {
|
||||
tool: 'Bash',
|
||||
arguments: { command: 'rm -rf build' },
|
||||
createdAt: T0 - 30_000,
|
||||
completedAt: T0 - 20_000,
|
||||
status: 'approved',
|
||||
decision: 'approved'
|
||||
}
|
||||
}
|
||||
},
|
||||
agentStateVersion: 5
|
||||
}),
|
||||
patches: [
|
||||
{
|
||||
agentState: {
|
||||
version: 4,
|
||||
value: {
|
||||
requests: {
|
||||
'req-1': {
|
||||
tool: 'Bash',
|
||||
arguments: { command: 'rm -rf build' },
|
||||
createdAt: T0 - 30_000
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'agent-state-out-of-order-versions',
|
||||
description: 'Out-of-order arrival across the two SSE connections: version 7 lands first and applies; the older version 6 arrives second and is dropped. Final state is the v7 value.',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{
|
||||
agentState: {
|
||||
version: 7,
|
||||
value: {
|
||||
requests: {
|
||||
'req-2': {
|
||||
tool: 'Write',
|
||||
arguments: { file_path: '/home/dev/project/README.md' },
|
||||
createdAt: T0 + 2_000
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
{
|
||||
agentState: {
|
||||
version: 6,
|
||||
value: {
|
||||
requests: {
|
||||
'req-1': {
|
||||
tool: 'Bash',
|
||||
arguments: { command: 'ls' },
|
||||
createdAt: T0 + 1_000
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'todos-version-watermark-from-absent',
|
||||
description: 'todos wrappers gate on todosUpdatedAt with an absent watermark treated as 0: the first patch applies, then an older-versioned todos replay is dropped instead of resurrecting the pre-update list.',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{
|
||||
todos: {
|
||||
version: T0 + 300_000,
|
||||
value: [
|
||||
{ content: 'Wire the SSE reconnect backoff', status: 'completed', priority: 'high', id: 'todo-1' },
|
||||
{ content: 'Port the patch gate to native', status: 'in_progress', priority: 'medium', id: 'todo-2', activeForm: 'Porting the patch gate to native' }
|
||||
]
|
||||
}
|
||||
},
|
||||
{
|
||||
todos: {
|
||||
version: T0 + 200_000,
|
||||
value: [
|
||||
{ content: 'Wire the SSE reconnect backoff', status: 'in_progress', priority: 'high', id: 'todo-1', activeForm: 'Wiring the SSE reconnect backoff' }
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'team-state-null-clear',
|
||||
description: 'teamState with value null means "team deleted": a strictly newer version clears the field entirely (absent in the expected session) and stores the watermark so a lagged pre-delete patch cannot resurrect the team.',
|
||||
initialSession: baseSession({
|
||||
teamState: {
|
||||
teamName: 'hapi-dev',
|
||||
members: [
|
||||
{ name: 'lead', agentType: 'claude', status: 'active' },
|
||||
{ name: 'reviewer', agentType: 'claude', status: 'idle' }
|
||||
],
|
||||
updatedAt: T0 - 100_000
|
||||
},
|
||||
teamStateUpdatedAt: T0 - 100_000
|
||||
}),
|
||||
patches: [
|
||||
{
|
||||
teamState: {
|
||||
version: T0 + 200_000,
|
||||
value: null
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'updated-at-max-monotonic',
|
||||
description: 'updatedAt is max-monotonic: a patch carrying an older updatedAt applies its other fields without rewinding the clock; a newer updatedAt advances it; an older updatedAt alone is a no-op (unchanged).',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{ updatedAt: T0 - 5_000, thinking: true },
|
||||
{ updatedAt: T0 + 5_000 },
|
||||
{ updatedAt: T0 - 2_000 }
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'flat-fields-last-write-wins',
|
||||
description: 'Flat fields (active, thinking, model, modelReasoningEffort, effort, permissionMode) are last-write-wins assignments with no version gate; serviceTier additionally honors an explicit null (key present) as a clear.',
|
||||
initialSession: baseSession({ serviceTier: 'standard' }),
|
||||
patches: [
|
||||
{
|
||||
active: false,
|
||||
thinking: true,
|
||||
model: 'claude-opus-4-6',
|
||||
modelReasoningEffort: 'high',
|
||||
effort: 'high',
|
||||
permissionMode: 'acceptEdits'
|
||||
},
|
||||
{ serviceTier: null }
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'active-turn-started-at-not-applied',
|
||||
description: 'Pins actual web behavior: applySessionDetailPatch has no branch for activeTurnStartedAt, so a patch carrying only it is a no-op, and in a mixed patch the other fields apply while activeTurnStartedAt keeps its cached value. (Clients relying on this field must take it from full-session payloads.)',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{ activeTurnStartedAt: T0 + 5_000 },
|
||||
{ activeTurnStartedAt: T0 + 5_000, thinking: true }
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'keepalive-subminute-active-at-dropped',
|
||||
description: 'Keep-alive noise gate: a patch whose only effective change is an activeAt delta below 60s is render-irrelevant and returns unchanged (the cached activeAt does NOT move); a delta of at least 60s applies.',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{ active: true, thinking: false, activeAt: T0 + 10_000 },
|
||||
{ active: true, thinking: false, activeAt: T0 + 60_000 }
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'mixed-flat-and-versioned-patch',
|
||||
description: 'One patch carrying flat fields, a newer metadata wrapper, and a stale agentState wrapper: the flat fields and metadata apply, the stale agentState is rejected, and all of it happens in a single applied call.',
|
||||
initialSession: baseSession({
|
||||
agentState: { controlledByUser: false },
|
||||
agentStateVersion: 5
|
||||
}),
|
||||
patches: [
|
||||
{
|
||||
thinking: true,
|
||||
updatedAt: T0 + 30_000,
|
||||
metadata: {
|
||||
version: 5,
|
||||
value: {
|
||||
path: '/home/dev/project',
|
||||
host: 'devbox',
|
||||
name: 'project-renamed',
|
||||
flavor: 'claude'
|
||||
}
|
||||
},
|
||||
agentState: {
|
||||
version: 2,
|
||||
value: {
|
||||
requests: {
|
||||
'req-stale': {
|
||||
tool: 'Bash',
|
||||
arguments: { command: 'echo stale' },
|
||||
createdAt: T0 - 60_000
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
name: 'scratchlist-updated-at-trigger-only',
|
||||
description: 'scratchlistUpdatedAt is a refetch trigger, not session state: the patch validates and is render-relevant, but applySessionDetailPatch stores nothing and reports unchanged. Clients react by refetching the scratchlist endpoint, never by mutating the session.',
|
||||
initialSession: baseSession(),
|
||||
patches: [
|
||||
{ scratchlistUpdatedAt: T0 + 900_000 }
|
||||
]
|
||||
}
|
||||
]
|
||||
@@ -0,0 +1,35 @@
|
||||
import type { Session, SessionPatch } from '@/types/api'
|
||||
|
||||
/** Per-patch verdict of `applySessionDetailPatch`: `applied` when the call
|
||||
* returned a new session object, `unchanged` when it returned null (keep the
|
||||
* previous object — version-gated, render-irrelevant, or no-op patch). */
|
||||
export type SessionPatchResult = 'applied' | 'unchanged'
|
||||
|
||||
/**
|
||||
* A hand-authored SSE session-patch case. `initialSession` and every patch
|
||||
* are wire-shaped inputs; the emitted document stores them schema-normalized
|
||||
* (`SessionSchema.parse` / `SessionPatchSchema.parse`) and everything under
|
||||
* `expected*` is machine-generated by running the real web patch fold
|
||||
* (`applySessionDetailPatch`) — never hand-written.
|
||||
*/
|
||||
export type SseFixtureCase = {
|
||||
/** Kebab-case name; becomes shared/fixtures/sse/<name>.json */
|
||||
name: string
|
||||
description: string
|
||||
/** Full Session as cached before the first patch arrives. */
|
||||
initialSession: Session
|
||||
/** SessionPatch payloads of `session-updated` events, in arrival order. */
|
||||
patches: SessionPatch[]
|
||||
}
|
||||
|
||||
export type SseFixtureDocument = {
|
||||
fixtureVersion: number
|
||||
name: string
|
||||
description: string
|
||||
initialSession: Session
|
||||
patches: SessionPatch[]
|
||||
/** Aligned with `patches`: verdict of each applySessionDetailPatch call. */
|
||||
expectedPatchResults: SessionPatchResult[]
|
||||
/** Session state after folding all patches. */
|
||||
expectedSession: Session
|
||||
}
|
||||
@@ -3,4 +3,4 @@
|
||||
// `cd web && bun scripts/generate-fixtures.ts`. Output is byte-deterministic.
|
||||
import { generateAllFixtures } from './fixtures/generate'
|
||||
|
||||
generateAllFixtures()
|
||||
await generateAllFixtures()
|
||||
|
||||
Reference in New Issue
Block a user