wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-10-28 23:01:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(codex): fail closed on ambiguous Web Rewind boundaries (#1707)

Browse Source 
* fix(codex): fail closed on ambiguous Web Rewind boundaries

* fix(codex): reject malformed native rewind history

* fix(codex): reject duplicate rewind identifiers

* feat(codex): offer Fork fallback for ambiguous rewind

* fix(codex): gate rewind Fork fallback on exact native boundary

* fix(codex): reject unresolved rewind boundaries safely

* chore: trigger PR checks

* fix(codex): gate safe rewind fallback on fork support

* fix(codex): require complete user ids for rewind fallback
This commit is contained in:
Ananovo
2026-09-06 14:18:40 +08:00
committed by GitHub
parent f4553dd1ec
commit aa0c1dc808
12 changed files with 627 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/codex/conversationHistory.test.ts
+341 -1
View File
@@ -4,7 +4,7 @@ import { CodexConversationHistory } from './conversationHistory'
function createClient(overrides?: {
fork?: (params: Record<string, unknown>) => Promise<{ thread: { id: string } }>
rollback?: (params: { threadId: string; numTurns: number }) => Promise<unknown>
read?: () => Promise<{ thread: { id: string; turns: Array<Record<string, unknown>> } }>
read?: () => Promise<{ thread: { id: string; turns?: Array<Record<string, unknown>> } }>
}) {
return {
supportsMethod: async () => true,
@@ -90,6 +90,346 @@ describe('CodexConversationHistory', () => {
expect(rollback).toHaveBeenCalledTimes(1)
})
it('returns a safe-Fork code when the selected message starts a steered turn', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{
id: 'turn-b',
items: [
{ type: 'userMessage', clientId: 'local-b' },
{ type: 'userMessage', clientId: 'local-steer' }
]
}
]
}
})
}) as never)
history.setThreadId('thread-1')
await history.probeCapabilities()
const result = await history.rewind('local-b')
expect(result).toMatchObject({
success: false,
outcome: 'rejected',
code: 'ambiguous_native_boundary_fork_safe'
})
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('does not offer the Fork fallback when historical fork is unsupported', async () => {
const rollback = vi.fn(async () => ({}))
const supportsMethod = vi.fn(async (method: string) => method === 'thread/rollback')
const history = new CodexConversationHistory(() => ({
...createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{
id: 'turn-b',
items: [
{ type: 'userMessage', clientId: 'local-b' },
{ type: 'userMessage', clientId: 'local-steer' }
]
}
]
}
})
}),
supportsMethod
}) as never)
history.setThreadId('thread-1')
await history.probeCapabilities()
const result = await history.rewind('local-b')
expect(result).toMatchObject({
success: false,
outcome: 'rejected',
code: 'ambiguous_native_boundary'
})
expect(history.getCapabilityStates()).toMatchObject({
forkAtMessage: 'unsupported',
rewindToMessage: 'supported'
})
expect(rollback).not.toHaveBeenCalled()
})
it('does not offer the Fork fallback when an earlier user message has no client id', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{
id: 'turn-b',
items: [
{ type: 'userMessage' },
{ type: 'userMessage', clientId: 'local-b' }
]
}
]
}
})
}) as never)
history.setThreadId('thread-1')
await history.probeCapabilities()
const result = await history.rewind('local-b')
expect(result).toMatchObject({
success: false,
outcome: 'rejected',
code: 'ambiguous_native_boundary'
})
expect(rollback).not.toHaveBeenCalled()
})
it('does not offer the Fork fallback for the second message in a steered turn', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{
id: 'turn-b',
items: [
{ type: 'userMessage', clientId: 'local-b' },
{ type: 'userMessage', clientId: 'local-steer' }
]
}
]
}
})
}) as never)
history.setThreadId('thread-1')
history.restoreTurns({ localSteer: 'turn-b' })
const result = await history.rewind('local-steer')
expect(result).toMatchObject({
success: false,
outcome: 'rejected',
code: 'ambiguous_native_boundary'
})
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when compaction is present', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{
id: 'turn-b',
items: [
{ type: 'userMessage', clientId: 'local-b' },
{ type: 'contextCompaction' }
]
}
]
}
})
}) as never)
history.setThreadId('thread-1')
const result = await history.rewind('local-b')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when native items are incomplete', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{ id: 'turn-b' }
]
}
})
}) as never)
history.setThreadId('thread-1')
history.restoreTurns({ localB: 'turn-b' })
const result = await history.rewind('localB')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when a user message has no client id', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{ id: 'turn-b', items: [{ type: 'userMessage' }] }
]
}
})
}) as never)
history.setThreadId('thread-1')
history.restoreTurns({ localB: 'turn-b' })
const result = await history.rewind('localB')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when a native turn has no id', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{ items: [{ type: 'assistantMessage' }] },
{ id: 'turn-c', items: [{ type: 'userMessage', clientId: 'local-c' }] }
]
}
})
}) as never)
history.setThreadId('thread-1')
const result = await history.rewind('local-a')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('returns a deterministic rejection when the selected native turn cannot be resolved', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({ thread: { id: 'thread-1' } })
}) as never)
history.setThreadId('thread-1')
history.restoreTurns({ localB: 'turn-b' })
const result = await history.rewind('localB')
expect(result).toMatchObject({
success: false,
code: 'ambiguous_native_boundary',
outcome: 'rejected'
})
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when a native item is malformed', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{ id: 'turn-b', items: [{}] }
]
}
})
}) as never)
history.setThreadId('thread-1')
const result = await history.rewind('local-a')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when native turn ids are duplicated', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-b' }] }
]
}
})
}) as never)
history.setThreadId('thread-1')
const result = await history.rewind('local-a')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('rejects rewind before native mutation when client ids are duplicated', async () => {
const rollback = vi.fn(async () => ({}))
const history = new CodexConversationHistory(() => createClient({
rollback,
read: async () => ({
thread: {
id: 'thread-1',
turns: [
{ id: 'turn-a', items: [{ type: 'userMessage', clientId: 'local-a' }] },
{ id: 'turn-b', items: [{ type: 'userMessage', clientId: 'local-a' }] }
]
}
})
}) as never)
history.setThreadId('thread-1')
const result = await history.rewind('local-a')
expect(result).toMatchObject({ success: false, outcome: 'rejected' })
if (result.success) throw new Error('Expected rewind to be rejected')
expect(result.error).toContain('ambiguous')
expect(rollback).not.toHaveBeenCalled()
})
it('marks rewind unsupported on method-not-found without affecting fork', async () => {
const rollback = vi.fn(async () => {
throw new Error('thread/rollback is unsupported')
cli/src/codex/conversationHistory.ts
+157 -11
View File
@@ -29,11 +29,17 @@ function isMethodNotFound(error: unknown): boolean {
}
type TurnInfo = {
id: string
id: string | null
status?: string
clientIds: string[]
userMessageCount: number
hasCompleteItems: boolean
hasContextCompaction: boolean
}
const AMBIGUOUS_REWIND_ERROR =
'Rewind is unavailable for this Codex history because native turn boundaries are ambiguous (steering, compaction, or incomplete history)'
export class CodexConversationHistory {
private states: ConversationHistoryCapabilityStates = { ...CODEX_CONVERSATION_HISTORY_INITIAL }
private threadId: string | null = null
@@ -127,9 +133,13 @@ export class CodexConversationHistory {
// Prefer stable inclusive lastTurnId of the previous turn. The first
// turn has no predecessor, so fall back to experimental beforeTurnId
// (exclusive) for that single boundary.
const previousTurnId = selectedIndex > 0 ? turns[selectedIndex - 1]?.id : null
if (selectedIndex > 0 && !previousTurnId) {
throw new Error(AMBIGUOUS_REWIND_ERROR)
}
const boundary = selectedIndex === 0
? { beforeTurnId: selectedTurnId }
: { lastTurnId: turns[selectedIndex - 1]!.id }
: { lastTurnId: previousTurnId! }
try {
const response = await client.forkThread({
threadId,
@@ -179,12 +189,57 @@ export class CodexConversationHistory {
}
const turns = await this.listTurns()
const turnId = await this.resolveTurnId(messageLocalId, turns)
const index = turns.findIndex((turn) => turn.id === turnId)
if (index < 0) throw new Error('Selected turn not found')
const turnId = await this.resolveTurnId(messageLocalId, turns).catch(() => null)
const index = turnId ? turns.findIndex((turn) => turn.id === turnId) : -1
if (index < 0) {
return {
success: false,
error: AMBIGUOUS_REWIND_ERROR,
code: 'ambiguous_native_boundary',
outcome: 'rejected'
}
}
if (turns[index]?.status === 'inProgress' || turns[index]?.status === 'in_progress') {
throw new Error('Cannot rewind an in-progress turn')
}
// thread/rollback takes a raw native-turn count, while the Web transcript
// exposes user-message boundaries. Only use that count when every native
// turn is known to contain exactly one identified user message. Steering,
// compaction, and incomplete item data can otherwise leave the model
// context out of sync with the transcript after a successful rollback.
const seenTurnIds = new Set<string>()
const seenClientIds = new Set<string>()
const hasAmbiguousBoundary = turns.some((turn) => {
const clientId = turn.clientIds[0]
if (
!turn.id ||
!turn.hasCompleteItems ||
turn.userMessageCount !== 1 ||
turn.clientIds.length !== 1 ||
!clientId ||
turn.hasContextCompaction ||
seenTurnIds.has(turn.id) ||
seenClientIds.has(clientId)
) {
return true
}
seenTurnIds.add(turn.id)
seenClientIds.add(clientId)
return false
})
if (hasAmbiguousBoundary) {
return {
success: false,
error: AMBIGUOUS_REWIND_ERROR,
code: this.states.forkAtMessage === 'supported'
&& this.isForkFallbackSafe(messageLocalId, turns, index)
? 'ambiguous_native_boundary_fork_safe'
: 'ambiguous_native_boundary',
outcome: 'rejected'
}
}
const numTurns = turns.length - index
if (numTurns <= 0) throw new Error('Invalid rewind count')
@@ -216,7 +271,7 @@ export class CodexConversationHistory {
const list = turns ?? await this.listTurns()
for (const turn of list) {
if (turn.clientIds.includes(localId)) {
if (turn.id && turn.clientIds.includes(localId)) {
this.turnByLocalId.set(localId, turn.id)
return turn.id
}
@@ -224,6 +279,61 @@ export class CodexConversationHistory {
throw new Error(`No native history point for message ${localId}`)
}
private isForkFallbackSafe(messageLocalId: string, turns: TurnInfo[], index: number): boolean {
const selected = turns[index]
if (
!selected
|| !selected.id
|| !selected.hasCompleteItems
|| selected.hasContextCompaction
|| selected.userMessageCount !== selected.clientIds.length
|| selected.clientIds[0] !== messageLocalId
) {
return false
}
if (turns.flatMap((turn) => turn.clientIds).filter((clientId) => clientId === messageLocalId).length !== 1) {
return false
}
if (turns.filter((turn) => turn.id === selected.id).length !== 1) {
return false
}
const previousId = index > 0 ? turns[index - 1]?.id : null
if (index > 0 && !previousId) {
return false
}
if (previousId && turns.filter((turn) => turn.id === previousId).length !== 1) {
return false
}
// The child transcript excludes the selected HAPI message. Native Fork
// excludes the whole selected native turn, so the selected message must
// be the first user item in that turn. Also require the retained prefix
// to be unambiguous; otherwise the child could inherit a different
// projection even though this particular boundary is exact.
const retainedTurnIds = new Set<string>()
const retainedClientIds = new Set<string>()
for (const turn of turns.slice(0, index)) {
const clientId = turn.clientIds[0]
if (
!turn.id
|| !turn.hasCompleteItems
|| turn.hasContextCompaction
|| turn.userMessageCount !== 1
|| turn.clientIds.length !== 1
|| !clientId
|| retainedTurnIds.has(turn.id)
|| retainedClientIds.has(clientId)
) {
return false
}
retainedTurnIds.add(turn.id)
retainedClientIds.add(clientId)
}
return true
}
private async listTurns(): Promise<TurnInfo[]> {
const client = this.getClient()
const threadId = this.threadId
@@ -233,16 +343,46 @@ export class CodexConversationHistory {
const response = await client.readThread({ threadId, includeTurns: true })
const thread = asRecord(response.thread)
const turns = Array.isArray(thread?.turns) ? thread.turns : []
return turns.flatMap((entry) => {
return turns.flatMap((entry): TurnInfo[] => {
const record = asRecord(entry)
const id = asString(record?.id)
if (!id) return []
if (!record || !id) {
return [{
id: null,
status: asString(record?.status) ?? undefined,
clientIds: [],
userMessageCount: 0,
hasCompleteItems: false,
hasContextCompaction: false
}]
}
const clientIds: string[] = []
const items = Array.isArray(record?.items) ? record.items : []
const items = Array.isArray(record?.items) ? record.items : null
if (!items) {
return [{
id,
status: asString(record?.status) ?? undefined,
clientIds,
userMessageCount: 0,
hasCompleteItems: false,
hasContextCompaction: false
}]
}
let userMessageCount = 0
let hasCompleteItems = true
let hasContextCompaction = false
for (const item of items) {
const itemRecord = asRecord(item)
const type = asString(itemRecord?.type) ?? asString(itemRecord?.itemType)
if (!itemRecord || !type) {
hasCompleteItems = false
continue
}
if (type === 'contextCompaction' || type === 'context_compaction') {
hasContextCompaction = true
}
if (type === 'userMessage' || type === 'user_message') {
userMessageCount += 1
const clientId = asString(itemRecord?.clientId) ?? asString(itemRecord?.client_id)
if (clientId) clientIds.push(clientId)
}
@@ -250,7 +390,10 @@ export class CodexConversationHistory {
return [{
id,
status: asString(record?.status) ?? undefined,
clientIds
clientIds,
userMessageCount,
hasCompleteItems,
hasContextCompaction
}]
})
} catch (error) {
@@ -258,7 +401,10 @@ export class CodexConversationHistory {
// Fall back to in-memory mapping only
return Array.from(this.turnByLocalId.entries()).map(([localId, id]) => ({
id,
clientIds: [localId]
clientIds: [localId],
userMessageCount: 1,
hasCompleteItems: false,
hasContextCompaction: false
}))
}
}