mirror of
https://github.com/wu736139669/hapi.git
synced 2026-08-05 06:24:37 +00:00
feat(opencode): open a fresh session on clear (#1300)
* test(opencode): specify fresh-session clear * feat(opencode): open a fresh session on clear * fix(opencode): release clear latch on cancel * fix(opencode): retry transient clear handoffs * fix(opencode): confirm clear archive delivery * fix(web): preserve superseded session access * fix(clear): invalidate transferred schedules * fix(runner): restore live spawn dedupe * fix(clear): preserve latched scheduled prompts * fix(runner): quarantine unverified children * fix(clear): retain handoff retry ownership * fix(runner): release recovered spawn dedupe * fix(clear): retain archive retry ownership * fix(clear): settle rejected immediate prompts * fix(clear): block reopening replaced sources * fix(clear): settle prompts when clear is cancelled * fix(clear): make fresh-session handoff durable * fix(clear): finalize only after native cleanup * fix(clear): abort failed native handoffs * fix(clear): gate recovery on cleanup proof * fix(clear): retry metadata persistence failures * fix(clear): preserve handoff ownership through teardown * fix(clear): abort incomplete cleanup reservations * fix(clear): require explicit exit before abort * fix(clear): verify owner exit before recovery * fix(clear): guard recovery handoff races * fix(clear): serialize cleanup callbacks * fix(clear): make callback retries idempotent * fix(clear): bind callbacks to reservations * fix(clear): recover pending spawns * fix(clear): deduplicate held prompts * fix(clear): validate redirect ownership * fix(clear): replay prompts in FIFO order * fix(clear): gate replacement delivery
This commit is contained in:
Junmo Kim
@@ -16,6 +16,37 @@ function createPublisher(events: SyncEvent[]): EventPublisher {
|
||||
}
|
||||
|
||||
describe('alive incremental events', () => {
|
||||
it('replays durable immediate prompts on every attach until consumed', () => {
|
||||
const store = new Store(':memory:')
|
||||
const emitted: Array<{ body?: { t?: string; message?: { localId?: string | null } } }> = []
|
||||
const io = {
|
||||
of: () => ({
|
||||
to: () => ({ emit: (_event: string, payload: unknown) => emitted.push(payload as typeof emitted[number]) })
|
||||
})
|
||||
}
|
||||
const engine = new SyncEngine(store, io as never, new RpcRegistry(), { broadcast() {} } as never)
|
||||
try {
|
||||
const session = engine.getOrCreateSession('attach-replay', { path: '/tmp/project', host: 'localhost' }, null, 'default')
|
||||
store.messages.addMessage(session.id, { text: 'queued before attach' }, 'queued-before-attach')
|
||||
const invoked = store.messages.addMessage(session.id, { text: 'already consumed' }, 'already-consumed')
|
||||
store.messages.markMessagesInvoked(session.id, ['already-consumed'], invoked.createdAt + 1)
|
||||
store.messages.addMessage(session.id, { text: 'future scheduled' }, 'future-scheduled', Date.now() + 60_000)
|
||||
expect(emitted).toEqual([])
|
||||
|
||||
engine.handleSessionAlive({ sid: session.id, time: Date.now() })
|
||||
engine.handleSessionAlive({ sid: session.id, time: Date.now() + 1 })
|
||||
expect(emitted.map((update) => update.body?.message?.localId)).toEqual([
|
||||
'queued-before-attach', 'queued-before-attach'
|
||||
])
|
||||
|
||||
store.messages.markMessagesInvoked(session.id, ['queued-before-attach'], Date.now())
|
||||
engine.handleSessionAlive({ sid: session.id, time: Date.now() + 2 })
|
||||
expect(emitted.map((update) => update.body?.message?.localId)).toEqual([
|
||||
'queued-before-attach', 'queued-before-attach'
|
||||
])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('includes active=true in session alive updates', () => {
|
||||
const store = new Store(':memory:')
|
||||
const events: SyncEvent[] = []
|
||||
|
||||
@@ -575,7 +575,7 @@ export class MessageService {
|
||||
sentFrom?: 'telegram-bot' | 'webapp'
|
||||
scheduledAt?: number | null
|
||||
}
|
||||
): Promise<void> {
|
||||
): Promise<string> {
|
||||
// Defence-in-depth invariant for non-REST callers (Telegram bot, MCP,
|
||||
// internal callers). Attachment paths live under the CLI session's
|
||||
// upload directory which `cleanupUploadDir` purges on session end; a
|
||||
@@ -602,13 +602,15 @@ export class MessageService {
|
||||
}
|
||||
}
|
||||
|
||||
const msg = this.store.messages.addMessage(
|
||||
const inserted = this.store.addMessageForCurrentSession(
|
||||
sessionId,
|
||||
content,
|
||||
payload.localId ?? undefined,
|
||||
payload.scheduledAt ?? null
|
||||
)
|
||||
this.onSessionActivity?.(sessionId, msg.createdAt)
|
||||
const actualSessionId = inserted.sessionId
|
||||
const msg = inserted.message
|
||||
this.onSessionActivity?.(actualSessionId, msg.createdAt)
|
||||
|
||||
// Only emit to CLI if the message is not scheduled for the future.
|
||||
// Mature or non-scheduled messages go through immediately; future scheduled
|
||||
@@ -617,14 +619,14 @@ export class MessageService {
|
||||
// the pre-insert `now` capture could misclassify a borderline scheduledAt
|
||||
// as future when it has already become past by the time we check.
|
||||
const isFutureScheduled = msg.scheduledAt !== null && msg.scheduledAt > Date.now()
|
||||
if (!isFutureScheduled) {
|
||||
if (!isFutureScheduled && !this.store.isOpenCodeClearDeliveryGated(actualSessionId)) {
|
||||
const update = {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
createdAt: msg.createdAt,
|
||||
body: {
|
||||
t: 'new-message' as const,
|
||||
sid: sessionId,
|
||||
sid: actualSessionId,
|
||||
message: {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
@@ -634,13 +636,13 @@ export class MessageService {
|
||||
}
|
||||
}
|
||||
}
|
||||
this.io.of('/cli').to(`session:${sessionId}`).emit('update', update)
|
||||
this.io.of('/cli').to(`session:${actualSessionId}`).emit('update', update)
|
||||
}
|
||||
|
||||
// Always emit message-received to Web SSE so the floating bar renders.
|
||||
this.publisher.emit({
|
||||
type: 'message-received',
|
||||
sessionId,
|
||||
sessionId: actualSessionId,
|
||||
message: {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
@@ -651,6 +653,7 @@ export class MessageService {
|
||||
scheduledAt: msg.scheduledAt
|
||||
}
|
||||
})
|
||||
return actualSessionId
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -684,6 +687,59 @@ export class MessageService {
|
||||
return { localIds, invokedAt }
|
||||
}
|
||||
|
||||
/** Replay durable immediate prompts whenever their CLI session attaches. */
|
||||
replayImmediateQueuedMessages(sessionId: string): number {
|
||||
if (this.store.isOpenCodeClearDeliveryGated(sessionId)) return 0
|
||||
const queued = this.store.messages.getImmediateQueuedLocalMessages(sessionId)
|
||||
for (const msg of queued) {
|
||||
const update = {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
createdAt: msg.createdAt,
|
||||
body: {
|
||||
t: 'new-message' as const,
|
||||
sid: sessionId,
|
||||
message: {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
createdAt: msg.createdAt,
|
||||
localId: msg.localId,
|
||||
content: msg.content
|
||||
}
|
||||
}
|
||||
}
|
||||
this.io.of('/cli').to(`session:${sessionId}`).emit('update', update)
|
||||
}
|
||||
return queued.length
|
||||
}
|
||||
|
||||
/** Release a completed clear handoff in finalized seq order. */
|
||||
releaseDeliverableQueuedMessages(sessionId: string, now: number = Date.now()): number {
|
||||
if (this.store.isOpenCodeClearDeliveryGated(sessionId)) return 0
|
||||
const queued = this.store.messages.getUninvokedLocalMessages(sessionId)
|
||||
.filter((msg) => msg.scheduledAt === null || msg.scheduledAt <= now)
|
||||
for (const msg of queued) {
|
||||
const update = {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
createdAt: msg.createdAt,
|
||||
body: {
|
||||
t: 'new-message' as const,
|
||||
sid: sessionId,
|
||||
message: {
|
||||
id: msg.id,
|
||||
seq: msg.seq,
|
||||
createdAt: msg.createdAt,
|
||||
localId: msg.localId,
|
||||
content: msg.content
|
||||
}
|
||||
}
|
||||
}
|
||||
this.io.of('/cli').to(`session:${sessionId}`).emit('update', update)
|
||||
}
|
||||
return queued.length
|
||||
}
|
||||
|
||||
/** Called by the hub 5-second tick (syncEngine.expireInactive).
|
||||
*
|
||||
* Finds all scheduled messages whose scheduled_at <= now and emits them to
|
||||
@@ -702,8 +758,14 @@ export class MessageService {
|
||||
releaseMatureScheduledMessages(now: number, skipSessionIds?: ReadonlySet<string>): void {
|
||||
const mature = this.store.messages.getMatureScheduledMessages(now)
|
||||
const maturedSessionIds = new Set<string>()
|
||||
const deliveryGateBySession = new Map<string, boolean>()
|
||||
for (const msg of mature) {
|
||||
if (skipSessionIds?.has(msg.sessionId)) {
|
||||
let deliveryGated = deliveryGateBySession.get(msg.sessionId)
|
||||
if (deliveryGated === undefined) {
|
||||
deliveryGated = this.store.isOpenCodeClearDeliveryGated(msg.sessionId)
|
||||
deliveryGateBySession.set(msg.sessionId, deliveryGated)
|
||||
}
|
||||
if (skipSessionIds?.has(msg.sessionId) || deliveryGated) {
|
||||
continue
|
||||
}
|
||||
const localId = msg.localId
|
||||
|
||||
@@ -0,0 +1,827 @@
|
||||
import { describe, expect, it, mock } from 'bun:test'
|
||||
import { RpcRegistry } from '../socket/rpcRegistry'
|
||||
import { Store } from '../store'
|
||||
import { SyncEngine, type SyncEvent } from './syncEngine'
|
||||
|
||||
function createEngine(onCliEmit?: (payload: unknown) => void) {
|
||||
const store = new Store(':memory:')
|
||||
const engine = new SyncEngine(store, {
|
||||
of: () => ({ to: () => ({ emit: (_event: string, payload: unknown) => onCliEmit?.(payload) }) })
|
||||
} as never, new RpcRegistry(), { broadcast() {} } as never)
|
||||
engine.getOrCreateMachine(
|
||||
'machine-1',
|
||||
{ host: 'host', platform: 'linux', happyCliVersion: 'test' },
|
||||
null,
|
||||
'default'
|
||||
)
|
||||
return { store, engine }
|
||||
}
|
||||
|
||||
function createClearSource(engine: SyncEngine, metadata: Record<string, unknown> = {}) {
|
||||
return engine.getOrCreateSession('clear-source', {
|
||||
path: '/tmp/project',
|
||||
host: 'host',
|
||||
machineId: 'machine-1',
|
||||
flavor: 'opencode',
|
||||
lifecycleState: 'archived',
|
||||
archiveReason: 'Cleared by /clear',
|
||||
preferredPermissionMode: 'yolo',
|
||||
opencodeSessionId: 'native-source-must-not-resume',
|
||||
...metadata
|
||||
}, null, 'default', 'opencode/model', 'effort-x', 'high')
|
||||
}
|
||||
|
||||
|
||||
function currentReplacementId(engine: SyncEngine, sessionId: string): string {
|
||||
const id = engine.getSessionByNamespace(sessionId, 'default')?.metadata?.opencodeClearOperation?.replacementSessionId
|
||||
if (!id) throw new Error('clear reservation missing')
|
||||
return id
|
||||
}
|
||||
|
||||
function setSpawn(engine: SyncEngine, spawnSession: ReturnType<typeof mock>) {
|
||||
;(engine as unknown as { rpcGateway: { spawnSession: typeof spawnSession } }).rpcGateway.spawnSession = spawnSession
|
||||
}
|
||||
|
||||
describe('SyncEngine.clearOpenCodeSession', () => {
|
||||
it.each(['resume', 'reopen'] as const)('allows %s after a failed native cleanup aborts clear', async (action) => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession(`abort-${action}`, {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'success' })
|
||||
const abortedMetadata = engine.getSessionByNamespace(source.id, 'default')!.metadata!
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'error' })
|
||||
const ended = store.sessions.getSessionByNamespace(source.id, 'default')!
|
||||
store.sessions.updateSessionMetadata(source.id, abortedMetadata, ended.metadataVersion, 'default')
|
||||
;(engine as unknown as { sessionCache: { refreshSession(id: string): unknown } }).sessionCache.refreshSession(source.id)
|
||||
setSpawn(engine, mock(async () => ({ type: 'success' as const, sessionId: source.id })))
|
||||
const result = action === 'resume'
|
||||
? await engine.resumeSession(source.id, 'default')
|
||||
: await engine.reopenSession(source.id, 'default')
|
||||
expect(result).not.toMatchObject({ type: 'error', code: 'resume_unavailable' })
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
it('durably reserves a replacement while the source is active and reuses it after archival', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('active-clear-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
expect(reserved.type).toBe('success')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
expect(typeof reserved.sessionId).toBe('string')
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation).toMatchObject({
|
||||
replacementSessionId: reserved.sessionId, state: 'reserved'
|
||||
})
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'success' })
|
||||
const metadataBeforeEnd = engine.getSessionByNamespace(source.id, 'default')!.metadata!
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'cleared' })
|
||||
const storedAfterEnd = store.sessions.getSessionByNamespace(source.id, 'default')!
|
||||
store.sessions.updateSessionMetadata(source.id, { ...metadataBeforeEnd, lifecycleState: 'archived', archiveReason: 'Cleared by /clear' }, storedAfterEnd.metadataVersion, 'default')
|
||||
;(engine as unknown as { sessionCache: { refreshSession(id: string): unknown } }).sessionCache.refreshSession(source.id)
|
||||
const spawnSession = mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string }))
|
||||
setSpawn(engine, spawnSession)
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'default')).resolves.toEqual({ type: 'success', sessionId: reserved.sessionId })
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('atomically redirects messages arriving after reservation to the replacement', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('active-clear-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
await engine.sendMessage(source.id, { text: 'late immediate', localId: 'late-immediate' })
|
||||
await engine.sendMessage(source.id, { text: 'late scheduled', localId: 'late-scheduled', scheduledAt: Date.now() + 60_000 })
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
expect(store.messages.getAllMessages(reserved.sessionId).map((m) => m.localId)).toEqual(['late-immediate', 'late-scheduled'])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('preserves FIFO from a source prompt before reservation to a redirected target prompt', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('reservation-boundary-fifo', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
store.messages.addMessage(source.id, { text: 'A before reservation' }, 'fifo-a')
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
await engine.sendMessage(source.id, { text: 'B after reservation', localId: 'fifo-b' })
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', reserved.sessionId)).toMatchObject({ type: 'success' })
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'cleared' })
|
||||
setSpawn(engine, mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string })))
|
||||
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
|
||||
expect(store.messages.getAllMessages(reserved.sessionId).map((message) => message.localId)).toEqual(['fifo-a', 'fifo-b'])
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('gates replacement delivery during spawn and releases finalized FIFO after linking', async () => {
|
||||
const emitted: Array<{ body?: { message?: { localId?: string | null } } }> = []
|
||||
const { store, engine } = createEngine((payload) => emitted.push(payload as typeof emitted[number]))
|
||||
try {
|
||||
const source = engine.getOrCreateSession('spawn-delivery-gate', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
emitted.length = 0
|
||||
store.messages.addMessage(source.id, { text: 'A before reservation' }, 'gated-a')
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
await engine.sendMessage(source.id, { text: 'B after reservation', localId: 'gated-b' })
|
||||
store.messages.addMessage(reserved.sessionId, { text: 'mature but gated' }, 'gated-scheduled', Date.now() - 1)
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', reserved.sessionId)).toMatchObject({ type: 'success' })
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'cleared' })
|
||||
let releaseSpawn!: () => void
|
||||
let spawnStarted = false
|
||||
const spawnWait = new Promise<void>((resolve) => { releaseSpawn = resolve })
|
||||
setSpawn(engine, mock(async (...args: unknown[]) => {
|
||||
spawnStarted = true
|
||||
await spawnWait
|
||||
return { type: 'success' as const, sessionId: args[12] as string }
|
||||
}))
|
||||
|
||||
const reconcile = (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
while (!spawnStarted) await Promise.resolve()
|
||||
expect(store.isOpenCodeClearDeliveryGated(reserved.sessionId)).toBe(true)
|
||||
engine.handleSessionAlive({ sid: reserved.sessionId, time: Date.now() })
|
||||
engine.handleSessionAlive({ sid: reserved.sessionId, time: Date.now() + 1 })
|
||||
;(engine as unknown as { messageService: { releaseMatureScheduledMessages(now: number): void } })
|
||||
.messageService.releaseMatureScheduledMessages(Date.now())
|
||||
expect(emitted).toEqual([])
|
||||
|
||||
releaseSpawn()
|
||||
await reconcile
|
||||
expect(store.isOpenCodeClearDeliveryGated(reserved.sessionId)).toBe(false)
|
||||
expect(emitted.map((update) => update.body?.message?.localId)).toEqual([
|
||||
'gated-a', 'gated-b', 'gated-scheduled'
|
||||
])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it.each([
|
||||
['supersededBySessionId', 'foreign'],
|
||||
['opencodeClearOperation', 'foreign'],
|
||||
['supersededBySessionId', 'missing'],
|
||||
['opencodeClearOperation', 'missing']
|
||||
] as const)('fails closed for a forged %s redirect to a %s target', async (field, targetKind) => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession(`forged-${field}-${targetKind}`, {
|
||||
path: '/tmp/project', host: 'host', flavor: 'opencode'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const targetId = `target-${field}-${targetKind}`
|
||||
if (targetKind === 'foreign') {
|
||||
engine.getOrCreateSession(`foreign-${field}`, { path: '/tmp/foreign', host: 'host' }, null, 'other', undefined, undefined, undefined, targetId)
|
||||
}
|
||||
const stored = store.sessions.getSessionByNamespace(source.id, 'default')!
|
||||
const redirect = field === 'supersededBySessionId'
|
||||
? { supersededBySessionId: targetId }
|
||||
: { opencodeClearOperation: { replacementSessionId: targetId, state: 'reserved', updatedAt: Date.now() } }
|
||||
store.sessions.updateSessionMetadata(source.id, {
|
||||
...(stored.metadata as Record<string, unknown>), ...redirect
|
||||
}, stored.metadataVersion, 'default')
|
||||
const events: SyncEvent[] = []
|
||||
engine.subscribe((event) => events.push(event))
|
||||
|
||||
await expect(engine.sendMessage(source.id, { text: 'must not cross namespace', localId: 'forged-local' })).rejects.toThrow(
|
||||
'redirect target is unavailable'
|
||||
)
|
||||
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
if (targetKind === 'foreign') expect(store.messages.getAllMessages(targetId)).toEqual([])
|
||||
expect(events).not.toContainEqual(expect.objectContaining({ type: 'message-received' }))
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it.each(['supersededBySessionId', 'opencodeClearOperation'] as const)(
|
||||
'allows a same-namespace %s redirect',
|
||||
async (field) => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession(`same-namespace-${field}`, { path: '/tmp/project', host: 'host' }, null, 'default')
|
||||
const target = engine.getOrCreateSession(`same-target-${field}`, { path: '/tmp/project', host: 'host' }, null, 'default')
|
||||
const stored = store.sessions.getSessionByNamespace(source.id, 'default')!
|
||||
const redirect = field === 'supersededBySessionId'
|
||||
? { supersededBySessionId: target.id }
|
||||
: { opencodeClearOperation: { replacementSessionId: target.id, state: 'reserved', updatedAt: Date.now() } }
|
||||
store.sessions.updateSessionMetadata(source.id, {
|
||||
...(stored.metadata as Record<string, unknown>), ...redirect
|
||||
}, stored.metadataVersion, 'default')
|
||||
await engine.sendMessage(source.id, { text: 'same namespace', localId: `same-${field}` })
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
expect(store.messages.getAllMessages(target.id)).toEqual([
|
||||
expect.objectContaining({ localId: `same-${field}`, invokedAt: null })
|
||||
])
|
||||
} finally { engine.stop() }
|
||||
}
|
||||
)
|
||||
|
||||
it('recovers cleanup-confirmed clear when the CLI dies before writing archive metadata', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('crashed-clear-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'success' })
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'error' })
|
||||
const spawnSession = mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string }))
|
||||
setSpawn(engine, spawnSession)
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata).toMatchObject({
|
||||
lifecycleState: 'archived', archiveReason: 'Cleared by /clear', supersededBySessionId: reserved.sessionId
|
||||
})
|
||||
expect(spawnSession).toHaveBeenCalledTimes(1)
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('recovers a persisted pending spawn with the exact replacement identity after restart', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const replacementSessionId = 'pending-before-spawn'
|
||||
const source = createClearSource(engine, {
|
||||
opencodeClearOperation: {
|
||||
replacementSessionId,
|
||||
state: 'pending',
|
||||
updatedAt: Date.now()
|
||||
}
|
||||
})
|
||||
const spawnSession = mock(async (...args: unknown[]) => ({
|
||||
type: 'success' as const,
|
||||
sessionId: args[12] as string
|
||||
}))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
|
||||
expect(spawnSession).toHaveBeenCalledTimes(1)
|
||||
expect(spawnSession.mock.calls[0]?.[12]).toBe(replacementSessionId)
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata).toMatchObject({
|
||||
supersededBySessionId: replacementSessionId,
|
||||
opencodeClearOperation: { replacementSessionId, state: 'completed' }
|
||||
})
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('safely aborts an inactive unconfirmed reservation and restores held messages', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('unconfirmed-clear-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
await engine.sendMessage(source.id, { text: 'held during lost response', localId: 'lost-response-held' })
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'error' })
|
||||
const spawnSession = mock(async () => ({ type: 'success' as const, sessionId: 'must-not-spawn' }))
|
||||
setSpawn(engine, spawnSession)
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
expect(spawnSession).not.toHaveBeenCalled()
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('aborted')
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([
|
||||
expect.objectContaining({ localId: 'lost-response-held', invokedAt: null })
|
||||
])
|
||||
expect((engine as unknown as { isOpenCodeClearSource(session: unknown): boolean }).isOpenCodeClearSource(
|
||||
engine.getSessionByNamespace(source.id, 'default')!
|
||||
)).toBe(false)
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('does not treat heartbeat expiry as process-death proof for a live reservation', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('heartbeat-expired-reservation', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() - 120_000 })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
await engine.sendMessage(source.id, { text: 'still owned', localId: 'still-owned' })
|
||||
const cached = engine.getSessionByNamespace(source.id, 'default') as unknown as { activeAt: number }
|
||||
cached.activeAt = Date.now() - 120_000
|
||||
const spawnSession = mock(async () => ({ type: 'success' as const, sessionId: 'must-not-spawn' }))
|
||||
setSpawn(engine, spawnSession)
|
||||
;(engine as unknown as { expireInactive(): void }).expireInactive()
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.active).toBe(false)
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('reserved')
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
expect(spawnSession).not.toHaveBeenCalled()
|
||||
const gateway = (engine as unknown as { rpcGateway: { stopRunnerSession: ReturnType<typeof mock> } }).rpcGateway
|
||||
gateway.stopRunnerSession = mock(async () => 'still_alive' as const)
|
||||
await expect(engine.resumeSession(source.id, 'default')).resolves.toMatchObject({ type: 'error', code: 'resume_unavailable' })
|
||||
gateway.stopRunnerSession = mock(async () => 'already_gone' as const)
|
||||
expect(await (engine as unknown as { recoverInactiveReservedClear(session: unknown, namespace: string): Promise<boolean> })
|
||||
.recoverInactiveReservedClear(engine.getSessionByNamespace(source.id, 'default')!, 'default')).toBe(true)
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('aborted')
|
||||
expect((engine as unknown as { isOpenCodeClearSource(session: unknown): boolean }).isOpenCodeClearSource(
|
||||
engine.getSessionByNamespace(source.id, 'default')!
|
||||
)).toBe(false)
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([
|
||||
expect.objectContaining({ localId: 'still-owned', invokedAt: null })
|
||||
])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it.each(['confirm', 'reactivate'] as const)('does not abort when %s wins during StopSession await', async (winner) => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession(`stop-race-${winner}`, {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
const cached = engine.getSessionByNamespace(source.id, 'default') as unknown as { activeAt: number }
|
||||
cached.activeAt = Date.now() - 120_000
|
||||
;(engine as unknown as { expireInactive(): void }).expireInactive()
|
||||
let release!: () => void
|
||||
const stop = new Promise<void>((resolve) => { release = resolve })
|
||||
;(engine as unknown as { rpcGateway: { stopRunnerSession: unknown } }).rpcGateway.stopRunnerSession = mock(async () => {
|
||||
await stop
|
||||
return 'already_gone' as const
|
||||
})
|
||||
const recovery = (engine as unknown as { recoverInactiveReservedClear(session: unknown, namespace: string): Promise<boolean> })
|
||||
.recoverInactiveReservedClear(engine.getSessionByNamespace(source.id, 'default')!, 'default')
|
||||
await Promise.resolve()
|
||||
if (winner === 'confirm') {
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'success' })
|
||||
} else {
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
}
|
||||
release()
|
||||
expect(await recovery).toBe(false)
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state)
|
||||
.toBe(winner === 'confirm' ? 'cleanup-confirmed' : 'reserved')
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('rejects a delayed cleanup confirmation after explicit exit owns the abort', () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('confirm-after-exit', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
const original = store.abortOpenCodeClearOperation.bind(store)
|
||||
store.abortOpenCodeClearOperation = (() => ({ result: 'error' as const })) as typeof original
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'error' })
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('abort-needed')
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'error' })
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('abort-needed')
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('rejects a delayed cleanup-failure abort after cleanup confirmation', () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('abort-after-confirm', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'success' })
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'error' })
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('cleanup-confirmed')
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('does not confirm a stale reservation identity', () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('stale-confirm-identity', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
const stored = store.sessions.getSessionByNamespace(source.id, 'default')!
|
||||
store.sessions.updateSessionMetadata(source.id, {
|
||||
...(stored.metadata as Record<string, unknown>),
|
||||
opencodeClearOperation: { replacementSessionId: 'new-owner', state: 'reserved', updatedAt: Date.now() }
|
||||
}, stored.metadataVersion, 'default')
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'error' })
|
||||
const persisted = store.sessions.getSessionByNamespace(source.id, 'default')?.metadata as Record<string, unknown>
|
||||
expect(persisted.opencodeClearOperation).toMatchObject({
|
||||
replacementSessionId: 'new-owner', state: 'reserved'
|
||||
})
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('treats a lost cleanup-confirm success response as an idempotent retry', () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('lost-confirm-response', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
const original = store.transitionOpenCodeClearOperation.bind(store)
|
||||
let loseResponse = true
|
||||
store.transitionOpenCodeClearOperation = ((...args: Parameters<typeof original>) => {
|
||||
const result = original(...args)
|
||||
if (loseResponse && result.result === 'success') {
|
||||
loseResponse = false
|
||||
return { result: 'version-mismatch' as const }
|
||||
}
|
||||
return result
|
||||
}) as typeof store.transitionOpenCodeClearOperation
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toEqual({
|
||||
type: 'success', sessionId: reserved.sessionId
|
||||
})
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', currentReplacementId(engine, source.id))).toEqual({
|
||||
type: 'success', sessionId: reserved.sessionId
|
||||
})
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('treats a lost abort success response as an idempotent retry', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('lost-abort-response', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
await engine.sendMessage(source.id, { text: 'restore once', localId: 'restore-once' })
|
||||
const original = store.abortOpenCodeClearOperation.bind(store)
|
||||
let loseResponse = true
|
||||
store.abortOpenCodeClearOperation = ((...args: Parameters<typeof original>) => {
|
||||
const result = original(...args)
|
||||
if (loseResponse && result.result === 'success') {
|
||||
loseResponse = false
|
||||
return { result: 'version-mismatch' as const }
|
||||
}
|
||||
return result
|
||||
}) as typeof store.abortOpenCodeClearOperation
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', currentReplacementId(engine, source.id))).toEqual({ type: 'success', sessionId: source.id })
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', currentReplacementId(engine, source.id))).toEqual({ type: 'success', sessionId: source.id })
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([
|
||||
expect.objectContaining({ localId: 'restore-once', invokedAt: null })
|
||||
])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it.each(['confirm', 'abort'] as const)('does not let delayed reservation A %s mutate reservation B', (callback) => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession(`stale-a-${callback}`, {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const first = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (first.type !== 'success') throw new Error('first reservation failed')
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', first.sessionId)).toMatchObject({ type: 'success' })
|
||||
const second = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (second.type !== 'success') throw new Error('second reservation failed')
|
||||
const result = callback === 'confirm'
|
||||
? engine.confirmOpenCodeClearCleanup(source.id, 'default', first.sessionId)
|
||||
: engine.abortOpenCodeClearSession(source.id, 'default', first.sessionId)
|
||||
expect(result).toMatchObject({ type: 'error' })
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation).toMatchObject({
|
||||
replacementSessionId: second.sessionId,
|
||||
state: 'reserved'
|
||||
})
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('aborts a reservation after native cleanup failure and restores held rows to the source', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('abort-clear-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
await engine.sendMessage(source.id, { text: 'held', localId: 'held' })
|
||||
expect(store.messages.getAllMessages(reserved.sessionId)).toHaveLength(1)
|
||||
expect(store.isOpenCodeClearDeliveryGated(reserved.sessionId)).toBe(true)
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', currentReplacementId(engine, source.id))).toEqual({ type: 'success', sessionId: source.id })
|
||||
expect(store.isOpenCodeClearDeliveryGated(reserved.sessionId)).toBe(false)
|
||||
expect(store.messages.getAllMessages(source.id).map((m) => m.localId)).toEqual(['held'])
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('aborted')
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'error' })
|
||||
expect((engine as unknown as { isOpenCodeClearSource(session: unknown): boolean }).isOpenCodeClearSource(
|
||||
engine.getSessionByNamespace(source.id, 'default')!
|
||||
)).toBe(false)
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('durably retries an explicit-exit abort after a metadata write failure', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('abort-retry-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner',
|
||||
lifecycleState: 'archived', archiveReason: 'Archived before clear abort'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
expect(engine.reserveOpenCodeClearSession(source.id, 'default')).toMatchObject({ type: 'success' })
|
||||
await engine.sendMessage(source.id, { text: 'restore atomically', localId: 'atomic-held' })
|
||||
const original = store.abortOpenCodeClearOperation.bind(store)
|
||||
let fail = true
|
||||
store.abortOpenCodeClearOperation = ((...args: Parameters<typeof original>) => {
|
||||
if (fail) return { result: 'not-found' as const }
|
||||
return original(...args)
|
||||
}) as typeof store.abortOpenCodeClearOperation
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'error' })
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('abort-needed')
|
||||
fail = false
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.state).toBe('aborted')
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([
|
||||
expect.objectContaining({ localId: 'atomic-held', invokedAt: null })
|
||||
])
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('re-reserves an aborted operation with a fresh durable identity', () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('retry-clear-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const first = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (first.type !== 'success') throw new Error('reservation failed')
|
||||
expect(engine.abortOpenCodeClearSession(source.id, 'default', currentReplacementId(engine, source.id))).toMatchObject({ type: 'success' })
|
||||
const second = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
expect(second).toMatchObject({ type: 'success', sessionId: expect.any(String) })
|
||||
if (second.type !== 'success') throw new Error('re-reservation failed')
|
||||
expect(second.sessionId).not.toBe(first.sessionId)
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation).toMatchObject({
|
||||
replacementSessionId: second.sessionId, state: 'reserved'
|
||||
})
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
it.each(['resume', 'reopen'] as const)('blocks %s of an archived clear source before spawning', async (action) => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine)
|
||||
const spawnSession = mock(async () => ({ type: 'success' as const, sessionId: 'must-not-spawn' }))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
const result = action === 'resume'
|
||||
? await engine.resumeSession(source.id, 'default')
|
||||
: await engine.reopenSession(source.id, 'default')
|
||||
|
||||
expect(result).toMatchObject({ type: 'error', code: 'resume_unavailable' })
|
||||
expect(spawnSession).not.toHaveBeenCalled()
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata).toMatchObject({
|
||||
lifecycleState: 'archived',
|
||||
archiveReason: 'Cleared by /clear'
|
||||
})
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('persists a preallocated replacement before spawning, preserving launch settings but never native source identity', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine)
|
||||
let operationAtSpawn: { replacementSessionId: string } | undefined
|
||||
const spawnSession = mock(async (...args: unknown[]) => {
|
||||
operationAtSpawn = engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation
|
||||
return {
|
||||
type: 'success' as const,
|
||||
sessionId: args[12] as string
|
||||
}
|
||||
})
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'default')).resolves.toMatchObject({
|
||||
type: 'success',
|
||||
sessionId: expect.any(String)
|
||||
})
|
||||
const replacementSessionId = spawnSession.mock.calls[0]?.[12] as string
|
||||
expect(replacementSessionId).toEqual(expect.any(String))
|
||||
expect(operationAtSpawn?.replacementSessionId).toBe(replacementSessionId)
|
||||
expect(replacementSessionId).not.toBe(source.id)
|
||||
expect(spawnSession).toHaveBeenCalledWith(
|
||||
'machine-1',
|
||||
'/tmp/project',
|
||||
'opencode',
|
||||
'opencode/model',
|
||||
'high',
|
||||
false,
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
'effort-x',
|
||||
'yolo',
|
||||
undefined,
|
||||
replacementSessionId,
|
||||
undefined
|
||||
)
|
||||
expect(engine.getSessionByNamespace(replacementSessionId, 'default')?.metadata).toMatchObject({
|
||||
flavor: 'opencode',
|
||||
path: '/tmp/project'
|
||||
})
|
||||
expect(engine.getSessionByNamespace(replacementSessionId, 'default')?.metadata?.opencodeSessionId).toBeUndefined()
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata).toMatchObject({
|
||||
supersededBySessionId: replacementSessionId
|
||||
})
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('reserves an independent replacement row for each cleared source', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const first = createClearSource(engine)
|
||||
const second = engine.getOrCreateSession('another-clear-source', {
|
||||
path: '/tmp/another-project', host: 'host', machineId: 'machine-1', flavor: 'opencode',
|
||||
lifecycleState: 'archived', archiveReason: 'Cleared by /clear'
|
||||
}, null, 'default')
|
||||
const spawnSession = mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string }))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
const firstResult = await engine.clearOpenCodeSession(first.id, 'default')
|
||||
const secondResult = await engine.clearOpenCodeSession(second.id, 'default')
|
||||
expect(firstResult).toMatchObject({ type: 'success' })
|
||||
expect(secondResult).toMatchObject({ type: 'success' })
|
||||
if (firstResult.type !== 'success' || secondResult.type !== 'success') throw new Error('expected successful clears')
|
||||
expect(firstResult.sessionId).not.toBe(secondResult.sessionId)
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('retries a failed spawn against the same durable replacement id', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine)
|
||||
const firstSpawn = mock(async () => ({ type: 'error' as const, message: 'runner unavailable' }))
|
||||
setSpawn(engine, firstSpawn)
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'default')).resolves.toMatchObject({
|
||||
type: 'error', code: 'spawn_failed'
|
||||
})
|
||||
|
||||
const pendingId = engine.getSessionByNamespace(source.id, 'default')?.metadata?.opencodeClearOperation?.replacementSessionId
|
||||
expect(pendingId).toEqual(expect.any(String))
|
||||
if (!pendingId) throw new Error('expected durable replacement id')
|
||||
const secondSpawn = mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string }))
|
||||
setSpawn(engine, secondSpawn)
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'default')).resolves.toEqual({
|
||||
type: 'success', sessionId: pendingId
|
||||
})
|
||||
expect(secondSpawn.mock.calls[0]?.[12]).toBe(pendingId)
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('returns the durable replacement to a reconnecting clear source without spawning again', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine, { supersededBySessionId: 'already-fresh' })
|
||||
const spawnSession = mock(async () => ({ type: 'success' as const, sessionId: 'must-not-spawn' }))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'default')).resolves.toEqual({
|
||||
type: 'success', sessionId: 'already-fresh'
|
||||
})
|
||||
expect(spawnSession).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('refuses source metadata that points to a machine outside its namespace', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('cross-namespace-clear', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode',
|
||||
lifecycleState: 'archived', archiveReason: 'Cleared by /clear'
|
||||
}, null, 'other')
|
||||
const spawnSession = mock(async () => ({ type: 'success' as const, sessionId: 'must-not-spawn' }))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'other')).resolves.toMatchObject({
|
||||
type: 'error', code: 'clear_unavailable'
|
||||
})
|
||||
expect(spawnSession).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('moves pending scheduled prompts to the replacement before it links the archived source', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine)
|
||||
const events: Array<{ type: string, sessionId?: string }> = []
|
||||
engine.subscribe((event) => events.push(event))
|
||||
const scheduled = store.messages.addMessage(source.id, { text: 'send later' }, 'scheduled-clear', Date.now() + 60_000)
|
||||
const spawnSession = mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string }))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
const result = await engine.clearOpenCodeSession(source.id, 'default')
|
||||
expect(result).toMatchObject({ type: 'success' })
|
||||
if (result.type !== 'success') throw new Error('expected successful clear')
|
||||
expect(store.messages.getAllMessages(source.id)).not.toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: scheduled.id })
|
||||
]))
|
||||
expect(store.messages.getAllMessages(result.sessionId)).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: scheduled.id, localId: 'scheduled-clear', invokedAt: null })
|
||||
]))
|
||||
expect(events).toContainEqual(expect.objectContaining({ type: 'messages-invalidated', sessionId: source.id }))
|
||||
expect(events).toContainEqual(expect.objectContaining({ type: 'messages-invalidated', sessionId: result.sessionId }))
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('moves every held prompt to the replacement without falsely consuming it', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine)
|
||||
store.messages.addMessage(source.id, { text: 'rejected immediate' }, 'immediate-after-clear')
|
||||
store.messages.addMessage(source.id, { text: 'scheduled transfer' }, 'scheduled-after-clear', Date.now() + 60_000)
|
||||
const events: Array<{ type: string, sessionId?: string, localIds?: string[] }> = []
|
||||
engine.subscribe((event) => events.push(event))
|
||||
setSpawn(engine, mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string })))
|
||||
|
||||
const result = await engine.clearOpenCodeSession(source.id, 'default')
|
||||
if (result.type !== 'success') throw new Error('expected successful clear')
|
||||
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
expect(store.messages.getAllMessages(result.sessionId)).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ localId: 'immediate-after-clear', invokedAt: null }),
|
||||
expect.objectContaining({ localId: 'scheduled-after-clear', invokedAt: null })
|
||||
]))
|
||||
expect(events).not.toContainEqual(expect.objectContaining({ type: 'messages-consumed' }))
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps the replacement copy when source and target share a queued localId', async () => {
|
||||
const { store, engine } = createEngine()
|
||||
try {
|
||||
const source = engine.getOrCreateSession('duplicate-held-source', {
|
||||
path: '/tmp/project', host: 'host', machineId: 'machine-1', flavor: 'opencode', startedBy: 'runner'
|
||||
}, null, 'default')
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const reserved = engine.reserveOpenCodeClearSession(source.id, 'default')
|
||||
if (reserved.type !== 'success') throw new Error('reservation failed')
|
||||
store.messages.addMessage(reserved.sessionId, { text: 'authoritative retry' }, 'duplicate-local-id')
|
||||
store.messages.addMessage(source.id, { text: 'stale source copy' }, 'duplicate-local-id')
|
||||
store.messages.addMessage(source.id, { text: 'unique immediate' }, 'unique-immediate')
|
||||
store.messages.addMessage(source.id, { text: 'unique scheduled' }, 'unique-scheduled', Date.now() + 60_000)
|
||||
expect(engine.confirmOpenCodeClearCleanup(source.id, 'default', reserved.sessionId)).toMatchObject({ type: 'success' })
|
||||
engine.handleSessionEnd({ sid: source.id, time: Date.now(), reason: 'cleared' })
|
||||
setSpawn(engine, mock(async (...args: unknown[]) => ({ type: 'success' as const, sessionId: args[12] as string })))
|
||||
|
||||
await (engine as unknown as { reconcileOpenCodeClears(): Promise<void> }).reconcileOpenCodeClears()
|
||||
|
||||
expect(store.messages.getAllMessages(source.id)).toEqual([])
|
||||
expect(store.messages.getAllMessages(reserved.sessionId).map((message) => ({
|
||||
localId: message.localId,
|
||||
text: (message.content as { text: string }).text,
|
||||
invokedAt: message.invokedAt
|
||||
}))).toEqual([
|
||||
{ localId: 'duplicate-local-id', text: 'authoritative retry', invokedAt: null },
|
||||
{ localId: 'unique-immediate', text: 'unique immediate', invokedAt: null },
|
||||
{ localId: 'unique-scheduled', text: 'unique scheduled', invokedAt: null }
|
||||
])
|
||||
expect(engine.getSessionByNamespace(source.id, 'default')?.metadata?.supersededBySessionId).toBe(reserved.sessionId)
|
||||
} finally { engine.stop() }
|
||||
})
|
||||
|
||||
it('refuses before spawning while the source is still active', async () => {
|
||||
const { engine } = createEngine()
|
||||
try {
|
||||
const source = createClearSource(engine)
|
||||
engine.handleSessionAlive({ sid: source.id, time: Date.now() })
|
||||
const spawnSession = mock(async () => ({ type: 'success' as const, sessionId: 'must-not-spawn' }))
|
||||
setSpawn(engine, spawnSession)
|
||||
|
||||
await expect(engine.clearOpenCodeSession(source.id, 'default')).resolves.toMatchObject({
|
||||
type: 'error', code: 'clear_unavailable'
|
||||
})
|
||||
expect(spawnSession).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
engine.stop()
|
||||
}
|
||||
})
|
||||
})
|
||||
+508
-7
@@ -7,7 +7,7 @@
|
||||
* - No E2E encryption; data is stored as JSON in SQLite
|
||||
*/
|
||||
|
||||
import { isKnownFlavor, type LocalResumeTarget, type ResumableSession } from '@hapi/protocol'
|
||||
import { isKnownFlavor, type LocalResumeTarget, type ResumableSession, type SessionEndReason } from '@hapi/protocol'
|
||||
import type { CursorChatStoreStatus, CursorMigrateOutcome, CursorMigrateToAcpRequest, MessagesResponse, QueuedStateResponse, SlashCommandsResponse } from '@hapi/protocol/apiTypes'
|
||||
import type { AgentFlavor, CodexCollaborationMode, DecryptedMessage, PermissionMode, Session, SyncEvent } from '@hapi/protocol/types'
|
||||
import { unwrapRoleWrappedRecordEnvelope } from '@hapi/protocol/messages'
|
||||
@@ -91,6 +91,14 @@ export type LocalHandoffResult =
|
||||
| { type: 'success' }
|
||||
| { type: 'error'; message: string; code: 'session_not_found' | 'access_denied' | 'already_local' | 'handoff_failed' }
|
||||
|
||||
export type ClearOpencodeSessionResult =
|
||||
| { type: 'success'; sessionId: string }
|
||||
| {
|
||||
type: 'error'
|
||||
message: string
|
||||
code: 'session_not_found' | 'access_denied' | 'clear_unavailable' | 'spawn_failed' | 'replacement_link_failed'
|
||||
}
|
||||
|
||||
export type CursorChatStoreStatusResult =
|
||||
| { type: 'success'; status: CursorChatStoreStatus }
|
||||
| { type: 'error'; message: string; code: 'session_not_found' | 'access_denied' | 'resume_unavailable' | 'no_machine_online' | 'probe_failed' }
|
||||
@@ -161,6 +169,8 @@ export class SyncEngine {
|
||||
private readonly piUnexpectedTempOriginalIds = new Map<string, string>()
|
||||
/** Serialize scratchlist uploads per session so disk-byte caps cannot race. */
|
||||
private readonly scratchlistUploadTails = new Map<string, Promise<unknown>>()
|
||||
/** Coalesce duplicate clear requests so retries cannot spawn two fresh sessions. */
|
||||
private readonly opencodeClearTails = new Map<string, Promise<ClearOpencodeSessionResult>>()
|
||||
/** Serialize fork/rewind per session so concurrent native rollbacks cannot stack. */
|
||||
private readonly historyActionsInFlight = new Set<string>()
|
||||
|
||||
@@ -409,6 +419,7 @@ export class SyncEngine {
|
||||
collaborationMode?: CodexCollaborationMode
|
||||
}): void {
|
||||
this.sessionCache.handleSessionAlive(payload)
|
||||
this.messageService.replayImmediateQueuedMessages(payload.sid)
|
||||
this.triggerDedupIfNeeded(payload.sid)
|
||||
}
|
||||
|
||||
@@ -430,8 +441,14 @@ export class SyncEngine {
|
||||
this.sessionCache.clearQueuedThinkingGrace(sessionId)
|
||||
}
|
||||
|
||||
handleSessionEnd(payload: { sid: string; time: number; reason?: 'completed' | 'terminated' | 'error' }): void {
|
||||
handleSessionEnd(payload: { sid: string; time: number; reason?: SessionEndReason }): void {
|
||||
const before = this.sessionCache.getSession(payload.sid)
|
||||
if (before?.metadata?.opencodeClearOperation?.state === 'reserved' && payload.reason !== 'cleared') {
|
||||
const operation = before.metadata.opencodeClearOperation
|
||||
if (this.transitionClearOperation(payload.sid, before.namespace, operation, 'abort-needed')) {
|
||||
this.abortOpenCodeClearSession(payload.sid, before.namespace, operation.replacementSessionId, 'abort-needed')
|
||||
}
|
||||
}
|
||||
const ownsPiAttempt = before?.metadata?.piResumeAttempt !== undefined
|
||||
const isPiAttemptChild = this.sessionCache.getSessions().some(
|
||||
(session) => session.metadata?.piResumeAttempt?.childSessionId === payload.sid
|
||||
@@ -764,6 +781,31 @@ async uploadScratchlistAttachment(
|
||||
// Piggybacked on the inactivity tick; not a logical part of expireInactive
|
||||
// but shares its 5s cadence (avoids a second timer).
|
||||
this.messageService.releaseMatureScheduledMessages(Date.now(), this.historyActionsInFlight)
|
||||
void this.reconcileOpenCodeClears()
|
||||
}
|
||||
|
||||
private async reconcileOpenCodeClears(): Promise<void> {
|
||||
for (let session of this.sessionCache.getSessions()) {
|
||||
const operation = session.metadata?.opencodeClearOperation
|
||||
if (session.active || !operation) continue
|
||||
if (operation.state === 'reserved') continue
|
||||
if (operation.state === 'abort-needed') {
|
||||
this.abortOpenCodeClearSession(session.id, session.namespace, operation.replacementSessionId, 'abort-needed')
|
||||
continue
|
||||
}
|
||||
if (!['cleanup-confirmed', 'finalizing', 'pending', 'failed'].includes(operation.state)) continue
|
||||
if (session.metadata?.lifecycleState !== 'archived' || session.metadata.archiveReason !== 'Cleared by /clear') {
|
||||
const result = this.store.sessions.updateSessionMetadata(session.id, {
|
||||
...session.metadata,
|
||||
lifecycleState: 'archived',
|
||||
lifecycleStateSince: Date.now(),
|
||||
archiveReason: 'Cleared by /clear'
|
||||
}, session.metadataVersion, session.namespace, { touchUpdatedAt: false })
|
||||
if (result.result !== 'success') continue
|
||||
session = this.sessionCache.refreshSession(session.id) ?? session
|
||||
}
|
||||
await this.clearOpenCodeSession(session.id, session.namespace).catch(() => {})
|
||||
}
|
||||
}
|
||||
|
||||
private reloadAll(): void {
|
||||
@@ -817,9 +859,9 @@ async uploadScratchlistAttachment(
|
||||
if (this.historyActionsInFlight.has(sessionId)) {
|
||||
throw new Error('Conversation history action already in progress')
|
||||
}
|
||||
await this.messageService.sendMessage(sessionId, payload)
|
||||
this.sessionCache.markMessageQueued(sessionId)
|
||||
this.sessionCache.recordSessionActivity(sessionId, Date.now())
|
||||
const actualSessionId = await this.messageService.sendMessage(sessionId, payload)
|
||||
this.sessionCache.markMessageQueued(actualSessionId)
|
||||
this.sessionCache.recordSessionActivity(actualSessionId, Date.now())
|
||||
}
|
||||
|
||||
async cancelQueuedMessage(
|
||||
@@ -1627,6 +1669,421 @@ async uploadScratchlistAttachment(
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn a fresh OpenCode HAPI session from a source that its own CLI has
|
||||
* already archived with the `cleared` lifecycle. Deliberately accepts only
|
||||
* that post-cleanup state: a target must never become active while the
|
||||
* source still owns an in-flight OpenCode turn or native compaction.
|
||||
*/
|
||||
async clearOpenCodeSession(sessionId: string, namespace: string): Promise<ClearOpencodeSessionResult> {
|
||||
const clearTailKey = `${namespace}:${sessionId}`
|
||||
const existing = this.opencodeClearTails.get(clearTailKey)
|
||||
if (existing) {
|
||||
return await existing
|
||||
}
|
||||
|
||||
const task = this.clearOpenCodeSessionOnce(sessionId, namespace)
|
||||
this.opencodeClearTails.set(clearTailKey, task)
|
||||
try {
|
||||
return await task
|
||||
} finally {
|
||||
if (this.opencodeClearTails.get(clearTailKey) === task) {
|
||||
this.opencodeClearTails.delete(clearTailKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
reserveOpenCodeClearSession(sessionId: string, namespace: string): ClearOpencodeSessionResult {
|
||||
const access = this.sessionCache.resolveSessionAccess(sessionId, namespace)
|
||||
if (!access.ok) {
|
||||
return { type: 'error', message: access.reason === 'access-denied' ? 'Session access denied' : 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' }
|
||||
}
|
||||
const source = access.session
|
||||
const metadata = source.metadata
|
||||
if (!source.active || metadata?.flavor !== 'opencode' || metadata.startedBy !== 'runner' || !metadata.machineId || !metadata.path) {
|
||||
return { type: 'error', message: 'Session is not an active runner-backed OpenCode session', code: 'clear_unavailable' }
|
||||
}
|
||||
const existing = metadata.opencodeClearOperation
|
||||
const operation = !existing || existing.state === 'aborted'
|
||||
? { replacementSessionId: randomUUID(), state: 'reserved' as const, updatedAt: Date.now() }
|
||||
: existing
|
||||
if (operation !== existing && !this.persistClearOperation(sessionId, namespace, operation)) {
|
||||
return { type: 'error', message: 'Could not persist the OpenCode clear reservation', code: 'replacement_link_failed' }
|
||||
}
|
||||
const replacementMetadata = { ...metadata }
|
||||
delete replacementMetadata.opencodeSessionId
|
||||
delete replacementMetadata.supersededBySessionId
|
||||
delete replacementMetadata.opencodeClearOperation
|
||||
delete replacementMetadata.lifecycleState
|
||||
delete replacementMetadata.lifecycleStateSince
|
||||
delete replacementMetadata.archivedBy
|
||||
delete replacementMetadata.archiveReason
|
||||
replacementMetadata.startedFromRunner = true
|
||||
replacementMetadata.startedBy = 'runner'
|
||||
this.getOrCreateSession(`opencode-clear-replacement:${operation.replacementSessionId}`, replacementMetadata, null, namespace,
|
||||
source.model ?? undefined, source.effort ?? undefined, source.modelReasoningEffort ?? undefined, operation.replacementSessionId)
|
||||
return { type: 'success', sessionId: operation.replacementSessionId }
|
||||
}
|
||||
|
||||
abortOpenCodeClearSession(
|
||||
sessionId: string,
|
||||
namespace: string,
|
||||
replacementSessionId: string,
|
||||
expectedState: 'reserved' | 'abort-needed' = 'reserved',
|
||||
requireInactive: boolean = false
|
||||
): ClearOpencodeSessionResult {
|
||||
const access = this.sessionCache.resolveSessionAccess(sessionId, namespace)
|
||||
if (!access.ok) return { type: 'error', message: 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' }
|
||||
const operation = access.session.metadata?.opencodeClearOperation
|
||||
if (!operation) return { type: 'error', message: 'Clear reservation not found', code: 'clear_unavailable' }
|
||||
if (operation.state === 'aborted') {
|
||||
return replacementSessionId === operation.replacementSessionId
|
||||
? { type: 'success', sessionId }
|
||||
: { type: 'error', message: 'Clear reservation not found', code: 'clear_unavailable' }
|
||||
}
|
||||
const required = { replacementSessionId, state: expectedState, requireInactive }
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
const latest = this.sessionCache.getSessionByNamespace(sessionId, namespace) ?? this.sessionCache.refreshSession(sessionId)
|
||||
if (!latest?.metadata) break
|
||||
const current = latest.metadata.opencodeClearOperation
|
||||
if (!current) break
|
||||
if (current.replacementSessionId === required.replacementSessionId && current.state === 'aborted') {
|
||||
return { type: 'success', sessionId }
|
||||
}
|
||||
if ((required.requireInactive && latest.active)
|
||||
|| current.replacementSessionId !== required.replacementSessionId
|
||||
|| current.state !== required.state) break
|
||||
const result = this.store.abortOpenCodeClearOperation(sessionId, current.replacementSessionId, {
|
||||
...latest.metadata,
|
||||
opencodeClearOperation: { ...current, state: 'aborted', updatedAt: Date.now(), error: undefined }
|
||||
}, latest.metadataVersion, namespace, required)
|
||||
if (result.result === 'success') {
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
return { type: 'success', sessionId }
|
||||
}
|
||||
if (result.result !== 'version-mismatch') break
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
}
|
||||
return { type: 'error', message: 'Could not abort clear reservation', code: 'replacement_link_failed' }
|
||||
}
|
||||
|
||||
confirmOpenCodeClearCleanup(sessionId: string, namespace: string, replacementSessionId: string): ClearOpencodeSessionResult {
|
||||
const access = this.sessionCache.resolveSessionAccess(sessionId, namespace)
|
||||
if (!access.ok) return { type: 'error', message: 'Session not found', code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found' }
|
||||
const operation = access.session.metadata?.opencodeClearOperation
|
||||
if (!operation) return { type: 'error', message: 'Clear reservation not found', code: 'clear_unavailable' }
|
||||
if (operation.state === 'cleanup-confirmed' && operation.replacementSessionId === replacementSessionId) {
|
||||
return { type: 'success', sessionId: operation.replacementSessionId }
|
||||
}
|
||||
if (operation.state !== 'reserved') return { type: 'error', message: 'Clear reservation not found', code: 'clear_unavailable' }
|
||||
if (operation.replacementSessionId !== replacementSessionId) return { type: 'error', message: 'Clear reservation not found', code: 'clear_unavailable' }
|
||||
if (!this.transitionClearOperation(sessionId, namespace, operation, 'cleanup-confirmed')) {
|
||||
return { type: 'error', message: 'Could not confirm native cleanup', code: 'replacement_link_failed' }
|
||||
}
|
||||
return { type: 'success', sessionId: operation.replacementSessionId }
|
||||
}
|
||||
|
||||
private async clearOpenCodeSessionOnce(sessionId: string, namespace: string): Promise<ClearOpencodeSessionResult> {
|
||||
const access = this.sessionCache.resolveSessionAccess(sessionId, namespace)
|
||||
if (!access.ok) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: access.reason === 'access-denied' ? 'Session access denied' : 'Session not found',
|
||||
code: access.reason === 'access-denied' ? 'access_denied' : 'session_not_found'
|
||||
}
|
||||
}
|
||||
|
||||
const source = access.session
|
||||
const metadata = source.metadata
|
||||
if (source.active
|
||||
|| metadata?.flavor !== 'opencode'
|
||||
|| metadata.lifecycleState !== 'archived'
|
||||
|| metadata.archiveReason !== 'Cleared by /clear') {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'Session must be an archived OpenCode clear source',
|
||||
code: 'clear_unavailable'
|
||||
}
|
||||
}
|
||||
|
||||
// A completed first request is the durable idempotency record used by
|
||||
// reconnecting/retrying CLI processes after their source socket closed.
|
||||
if (metadata.supersededBySessionId) {
|
||||
return { type: 'success', sessionId: metadata.supersededBySessionId }
|
||||
}
|
||||
|
||||
if (!metadata.machineId || !metadata.path) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'OpenCode clear source is missing machine or directory metadata',
|
||||
code: 'clear_unavailable'
|
||||
}
|
||||
}
|
||||
// The source metadata is client-controlled, so validate the recorded
|
||||
// machine through the namespace-scoped cache before any persistent or
|
||||
// runner-facing action.
|
||||
if (!this.getMachineByNamespace(metadata.machineId, namespace)) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'OpenCode clear source machine is unavailable in this namespace',
|
||||
code: 'clear_unavailable'
|
||||
}
|
||||
}
|
||||
|
||||
// Persist the replacement identity *before* asking a runner to create
|
||||
// a process. A retry after a lost RPC response therefore uses this same
|
||||
// HAPI id rather than accidentally spawning a second fresh session.
|
||||
let operation = metadata.opencodeClearOperation
|
||||
if (!operation) {
|
||||
operation = {
|
||||
replacementSessionId: randomUUID(),
|
||||
state: 'pending' as const,
|
||||
updatedAt: Date.now()
|
||||
}
|
||||
if (!this.persistClearOperation(sessionId, namespace, operation)) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'Could not persist the OpenCode clear replacement operation',
|
||||
code: 'replacement_link_failed'
|
||||
}
|
||||
}
|
||||
} else if (operation.state === 'failed') {
|
||||
operation = { ...operation, state: 'pending', updatedAt: Date.now(), error: undefined }
|
||||
if (!this.persistClearOperation(sessionId, namespace, operation)) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'Could not resume the OpenCode clear replacement operation',
|
||||
code: 'replacement_link_failed'
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (operation.state === 'reserved') {
|
||||
return { type: 'error', message: 'Native OpenCode cleanup is not confirmed', code: 'clear_unavailable' }
|
||||
}
|
||||
if (operation.state === 'cleanup-confirmed') {
|
||||
operation = { ...operation, state: 'finalizing', updatedAt: Date.now() }
|
||||
if (!this.persistClearOperation(sessionId, namespace, operation)) {
|
||||
return { type: 'error', message: 'Could not finalize the OpenCode clear reservation', code: 'replacement_link_failed' }
|
||||
}
|
||||
}
|
||||
|
||||
const replacementMetadata = { ...metadata }
|
||||
delete replacementMetadata.opencodeSessionId
|
||||
delete replacementMetadata.supersededBySessionId
|
||||
delete replacementMetadata.opencodeClearOperation
|
||||
delete replacementMetadata.lifecycleState
|
||||
delete replacementMetadata.lifecycleStateSince
|
||||
delete replacementMetadata.archivedBy
|
||||
delete replacementMetadata.archiveReason
|
||||
replacementMetadata.startedFromRunner = true
|
||||
replacementMetadata.startedBy = 'runner'
|
||||
|
||||
// bootstrapExistingSession requires an existing row. The stable id lets
|
||||
// a runner coalesce retries only while its spawned child remains alive;
|
||||
// replacement.active is the durable cross-runner reconciliation signal.
|
||||
const replacement = this.getOrCreateSession(
|
||||
`opencode-clear-replacement:${operation.replacementSessionId}`,
|
||||
replacementMetadata,
|
||||
null,
|
||||
namespace,
|
||||
source.model ?? undefined,
|
||||
source.effort ?? undefined,
|
||||
source.modelReasoningEffort ?? undefined,
|
||||
operation.replacementSessionId
|
||||
)
|
||||
|
||||
// A previous request can have spawned the target but lost the source
|
||||
// link acknowledgement. Do not ask the runner again in that case.
|
||||
if (replacement.active) {
|
||||
return this.finishOpenCodeClear(sessionId, namespace, operation.replacementSessionId, operation)
|
||||
}
|
||||
|
||||
// Do not supply a native OpenCode resume id. existingSessionId is only
|
||||
// the preallocated HAPI row; OpenCode starts a brand-new native thread.
|
||||
const spawned = await this.spawnSession(
|
||||
metadata.machineId,
|
||||
metadata.path,
|
||||
'opencode',
|
||||
source.model ?? undefined,
|
||||
source.modelReasoningEffort ?? undefined,
|
||||
false,
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
source.effort ?? undefined,
|
||||
source.permissionMode ?? metadata.preferredPermissionMode,
|
||||
source.serviceTier ?? undefined,
|
||||
operation.replacementSessionId,
|
||||
source.collaborationMode
|
||||
)
|
||||
if (spawned.type === 'error') {
|
||||
this.persistClearOperationState(sessionId, namespace, operation, spawned.message)
|
||||
return { type: 'error', message: spawned.message, code: 'spawn_failed' }
|
||||
}
|
||||
if (spawned.sessionId !== operation.replacementSessionId) {
|
||||
const message = 'Runner returned an unexpected OpenCode clear replacement id'
|
||||
this.persistClearOperationState(sessionId, namespace, operation, message)
|
||||
return { type: 'error', message, code: 'spawn_failed' }
|
||||
}
|
||||
|
||||
return this.finishOpenCodeClear(sessionId, namespace, operation.replacementSessionId, operation)
|
||||
}
|
||||
|
||||
private finishOpenCodeClear(
|
||||
sessionId: string,
|
||||
namespace: string,
|
||||
replacementSessionId: string,
|
||||
operation: NonNullable<Session['metadata']>['opencodeClearOperation']
|
||||
): ClearOpencodeSessionResult {
|
||||
if (!operation) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'OpenCode clear operation was not persisted',
|
||||
code: 'replacement_link_failed'
|
||||
}
|
||||
}
|
||||
try {
|
||||
const moved = this.store.messages.moveUninvokedMessages(sessionId, replacementSessionId)
|
||||
if (moved > 0) {
|
||||
this.eventPublisher.emit({ type: 'messages-invalidated', sessionId })
|
||||
this.eventPublisher.emit({ type: 'messages-invalidated', sessionId: replacementSessionId })
|
||||
}
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : 'Could not move scheduled prompts to the fresh OpenCode session'
|
||||
this.persistClearOperationState(sessionId, namespace, operation, message)
|
||||
return { type: 'error', message, code: 'replacement_link_failed' }
|
||||
}
|
||||
if (!this.persistClearReplacement(sessionId, namespace, replacementSessionId, operation)) {
|
||||
const message = 'Fresh OpenCode session started but the archived source could not be linked'
|
||||
this.persistClearOperationState(sessionId, namespace, operation, message)
|
||||
return {
|
||||
type: 'error',
|
||||
message,
|
||||
code: 'replacement_link_failed'
|
||||
}
|
||||
}
|
||||
this.messageService.releaseDeliverableQueuedMessages(replacementSessionId)
|
||||
return { type: 'success', sessionId: replacementSessionId }
|
||||
}
|
||||
|
||||
private transitionClearOperation(
|
||||
sessionId: string,
|
||||
namespace: string,
|
||||
expected: NonNullable<NonNullable<Session['metadata']>['opencodeClearOperation']>,
|
||||
state: 'abort-needed' | 'cleanup-confirmed'
|
||||
): boolean {
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
const latest = this.sessionCache.getSessionByNamespace(sessionId, namespace)
|
||||
?? this.sessionCache.refreshSession(sessionId)
|
||||
if (!latest?.metadata) return false
|
||||
const current = latest.metadata.opencodeClearOperation
|
||||
if (current?.replacementSessionId === expected.replacementSessionId && current.state === state) return true
|
||||
if (current?.replacementSessionId !== expected.replacementSessionId || current.state !== expected.state) return false
|
||||
const result = this.store.transitionOpenCodeClearOperation(sessionId, {
|
||||
...latest.metadata,
|
||||
opencodeClearOperation: { ...expected, state, updatedAt: Date.now(), error: undefined }
|
||||
}, latest.metadataVersion, namespace, {
|
||||
replacementSessionId: expected.replacementSessionId,
|
||||
state: expected.state
|
||||
})
|
||||
if (result.result === 'success') {
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
return true
|
||||
}
|
||||
if (result.result !== 'version-mismatch') return false
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
private persistClearOperation(
|
||||
sessionId: string,
|
||||
namespace: string,
|
||||
operation: NonNullable<Session['metadata']>['opencodeClearOperation']
|
||||
): boolean {
|
||||
if (!operation) return false
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
const latest = this.sessionCache.getSessionByNamespace(sessionId, namespace)
|
||||
?? this.sessionCache.refreshSession(sessionId)
|
||||
if (!latest?.metadata) return false
|
||||
if (latest.metadata.supersededBySessionId) {
|
||||
return latest.metadata.supersededBySessionId === operation.replacementSessionId
|
||||
}
|
||||
const existing = latest.metadata.opencodeClearOperation
|
||||
if (existing && existing.replacementSessionId !== operation.replacementSessionId && existing.state !== 'aborted') return false
|
||||
const result = this.store.sessions.updateSessionMetadata(
|
||||
sessionId,
|
||||
{ ...latest.metadata, opencodeClearOperation: operation },
|
||||
latest.metadataVersion,
|
||||
namespace,
|
||||
{ touchUpdatedAt: false }
|
||||
)
|
||||
if (result.result === 'success') {
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
return true
|
||||
}
|
||||
if (result.result !== 'version-mismatch') return false
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
private persistClearOperationState(
|
||||
sessionId: string,
|
||||
namespace: string,
|
||||
operation: NonNullable<Session['metadata']>['opencodeClearOperation'],
|
||||
error: string
|
||||
): void {
|
||||
if (!operation) return
|
||||
this.persistClearOperation(sessionId, namespace, {
|
||||
...operation,
|
||||
state: 'failed',
|
||||
updatedAt: Date.now(),
|
||||
error: error.slice(0, 500)
|
||||
})
|
||||
}
|
||||
|
||||
private persistClearReplacement(
|
||||
sessionId: string,
|
||||
namespace: string,
|
||||
replacementSessionId: string,
|
||||
operation: NonNullable<Session['metadata']>['opencodeClearOperation']
|
||||
): boolean {
|
||||
if (!operation) return false
|
||||
for (let attempt = 0; attempt < 3; attempt += 1) {
|
||||
const latest = this.sessionCache.getSessionByNamespace(sessionId, namespace)
|
||||
?? this.sessionCache.refreshSession(sessionId)
|
||||
if (!latest?.metadata) return false
|
||||
if (latest.metadata.supersededBySessionId) {
|
||||
return latest.metadata.supersededBySessionId === replacementSessionId
|
||||
}
|
||||
const result = this.store.sessions.updateSessionMetadata(
|
||||
sessionId,
|
||||
{
|
||||
...latest.metadata,
|
||||
supersededBySessionId: replacementSessionId,
|
||||
opencodeClearOperation: {
|
||||
...operation,
|
||||
state: 'completed',
|
||||
updatedAt: Date.now(),
|
||||
error: undefined
|
||||
}
|
||||
},
|
||||
latest.metadataVersion,
|
||||
namespace,
|
||||
{ touchUpdatedAt: false }
|
||||
)
|
||||
if (result.result === 'success') {
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
return true
|
||||
}
|
||||
if (result.result !== 'version-mismatch') return false
|
||||
this.sessionCache.refreshSession(sessionId)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
private resolveFlavor(session: Session): AgentFlavor {
|
||||
const flavor = session.metadata?.flavor
|
||||
return isKnownFlavor(flavor) ? flavor : 'claude'
|
||||
@@ -1999,6 +2456,29 @@ async uploadScratchlistAttachment(
|
||||
return this.store.messages.getFirstMessages(sessionId, 1).length === 0
|
||||
}
|
||||
|
||||
private isOpenCodeClearSource(session: Session): boolean {
|
||||
const metadata = session.metadata
|
||||
return metadata?.flavor === 'opencode'
|
||||
&& (metadata.archiveReason === 'Cleared by /clear'
|
||||
|| (metadata.opencodeClearOperation !== undefined && metadata.opencodeClearOperation.state !== 'aborted')
|
||||
|| metadata.supersededBySessionId !== undefined)
|
||||
}
|
||||
|
||||
private async recoverInactiveReservedClear(session: Session, namespace: string): Promise<boolean> {
|
||||
const operation = session.metadata?.opencodeClearOperation
|
||||
const machineId = session.metadata?.machineId
|
||||
if (session.active || operation?.state !== 'reserved' || !machineId) return false
|
||||
try {
|
||||
const status = await this.rpcGateway.stopRunnerSession(machineId, session.id)
|
||||
if (status === 'still_alive') return false
|
||||
return this.abortOpenCodeClearSession(
|
||||
session.id, namespace, operation.replacementSessionId, 'reserved', true
|
||||
).type === 'success'
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async resumeSession(sessionId: string, namespace: string, opts?: { permissionMode?: PermissionMode }): Promise<ResumeSessionResult> {
|
||||
const access = this.sessionCache.resolveSessionAccess(sessionId, namespace)
|
||||
if (!access.ok) {
|
||||
@@ -2009,7 +2489,17 @@ async uploadScratchlistAttachment(
|
||||
}
|
||||
}
|
||||
|
||||
const initialSession = access.session
|
||||
let initialSession = access.session
|
||||
if (await this.recoverInactiveReservedClear(initialSession, namespace)) {
|
||||
initialSession = this.sessionCache.getSessionByNamespace(sessionId, namespace) ?? initialSession
|
||||
}
|
||||
if (this.isOpenCodeClearSource(initialSession)) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'This OpenCode session was replaced by /clear',
|
||||
code: 'resume_unavailable'
|
||||
}
|
||||
}
|
||||
if (initialSession.active) {
|
||||
return { type: 'success', sessionId: access.sessionId }
|
||||
}
|
||||
@@ -2275,9 +2765,20 @@ async uploadScratchlistAttachment(
|
||||
}
|
||||
}
|
||||
|
||||
const session = access.session
|
||||
let session = access.session
|
||||
if (await this.recoverInactiveReservedClear(session, namespace)) {
|
||||
session = this.sessionCache.getSessionByNamespace(sessionId, namespace) ?? session
|
||||
}
|
||||
const metadata = session.metadata
|
||||
|
||||
if (this.isOpenCodeClearSource(session)) {
|
||||
return {
|
||||
type: 'error',
|
||||
message: 'This OpenCode session was replaced by /clear',
|
||||
code: 'resume_unavailable'
|
||||
}
|
||||
}
|
||||
|
||||
if (metadata?.flavor === 'pi' && this.isPiResumeBlocked(access.sessionId)) {
|
||||
if (session.active) {
|
||||
return { type: 'error', message: 'Pi resume is already in progress', code: 'resume_failed' }
|
||||
|
||||
Reference in New Issue
Block a user