Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
112 changes: 112 additions & 0 deletions src/main/native-chat/agent-session-context-snapshots.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
import * as fileOperations from 'node:fs/promises'
import { mkdtemp, readFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, expect, it, vi } from 'vitest'
import * as contextSnapshots from './agent-session-context-snapshots'

vi.mock('node:fs/promises', { spy: true })
const actualFiles = await vi.importActual<typeof fileOperations>('node:fs/promises')
const SESSION = 'pr-31782-publication'
let root: string
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'pr-31782-publication-'))
})
afterEach(async () => {
vi.restoreAllMocks()
await actualFiles.rm(contextSnapshots.agentSessionContextSnapshotRoot(root), {
recursive: true,
force: true
})
await actualFiles.rm(root, { recursive: true, force: true })
})

it('publishes captures in ownership order despite slow staging, dispatched rename, cleanup and a newer failure', async () => {
const write = actualFiles.writeFile
const rename = actualFiles.rename
const remove = actualFiles.rm
const snapshot = (text: string) =>
contextSnapshots.writeAgentSessionContextSnapshot(root, SESSION, text)
const stage = Promise.withResolvers<void>()
const staged = Promise.withResolvers<void>()
const writing = vi.spyOn(fileOperations, 'writeFile').mockImplementationOnce(async (...args) => {
staged.resolve()
await stage.promise
return write(...args)
})
const old = snapshot('old capture').then(
() => 'success',
() => 'replaced'
)
await staged.promise
const path = await snapshot('new capture')
stage.resolve()
expect(await old).toBe('replaced')
expect(await readFile(path, 'utf8')).toBe('new capture')
const dispatched = Promise.withResolvers<void>()
const finishRename = Promise.withResolvers<void>()
const cleaning = Promise.withResolvers<void>()
const finishCleanup = Promise.withResolvers<void>()
const successorStaged = Promise.withResolvers<void>()
let oldPart = ''
let oldRenameSettled = false
const renaming = vi
.spyOn(fileOperations, 'rename')
.mockImplementationOnce(async (from, to) => {
oldPart = String(from)
dispatched.resolve()
await finishRename.promise
await rename(from, to)
oldRenameSettled = true
})
.mockImplementation(async (from, to) => {
expect(oldRenameSettled).toBe(true)
return rename(from, to)
})
renaming.mockClear()
vi.spyOn(fileOperations, 'rm').mockImplementation(async (target, options) => {
if (String(target) === oldPart) {
cleaning.resolve()
await finishCleanup.promise
}
return remove(target, options)
})
writing.mockImplementation(async (...args) => {
await write(...args)
if (args[1] === 'queued successor') {
successorStaged.resolve()
}
})
const prior = snapshot('dispatched predecessor').then(
() => 'success',
() => 'replaced'
)
await dispatched.promise
const next = snapshot('queued successor')
await successorStaged.promise
expect(renaming).toHaveBeenCalledOnce()
finishRename.resolve()
await cleaning.promise
expect(await next).toBe(path)
expect(await readFile(path, 'utf8')).toBe('queued successor')
finishCleanup.resolve()
expect(await prior).toBe('replaced')
const failedStage = Promise.withResolvers<void>()
const oldStaged = Promise.withResolvers<void>()
writing
.mockImplementationOnce(async (...args) => {
oldStaged.resolve()
await failedStage.promise
return write(...args)
})
.mockRejectedValueOnce(new Error('newer stage failed'))
const revoked = snapshot('must not revive').then(
() => 'success',
() => 'replaced'
)
await oldStaged.promise
await expect(snapshot('failed newer capture')).rejects.toThrow('newer stage failed')
failedStage.resolve()
expect(await revoked).toBe('replaced')
expect(await readFile(path, 'utf8')).toBe('queued successor')
})
57 changes: 57 additions & 0 deletions src/main/native-chat/agent-session-context-snapshots.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
import { createHash, randomUUID } from 'node:crypto'
import { rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join, resolve } from 'node:path'
import { runKeyedSerializedOperation } from '../cli/keyed-promise-queue'
import { renameFileWithWindowsRetryAsync } from '../../shared/windows-retry-file-operations'
import {
ensureOwnedTempStagingRoot,
getOwnedTempStagingRoot
} from '../window/owned-temp-staging-root'

const claims = new Map<string, object>()
const publications = new Map<string, Promise<void>>()

export function agentSessionContextSnapshotRoot(stateDirectory: string): string {
const profile = createHash('sha256').update(resolve(stateDirectory)).digest('hex').slice(0, 16)
return getOwnedTempStagingRoot(tmpdir(), `orca-handoff-${profile}`)
}

/** One snapshot per chat, replaced on each hand-off; the OS owns temp retention. */
export async function writeAgentSessionContextSnapshot(
stateDirectory: string,
sessionId: string,
text: string
): Promise<string> {
const root = agentSessionContextSnapshotRoot(stateDirectory)
const name = createHash('sha256').update(sessionId).digest('hex')
const path = join(root, `${name}.txt`)
const temporary = `${path}.${randomUUID()}.part`
const claim = {}
claims.set(path, claim)
const isCurrent = () => claims.get(path) === claim
let published = false
try {
try {
if (!(await ensureOwnedTempStagingRoot(root))) {
throw new Error('Chat context snapshot folder is not private')
}
await writeFile(temporary, text, { flag: 'wx', mode: 0o600 })
// Staging can finish out of order; only publication holds the shared path queue.
published = await runKeyedSerializedOperation(publications, path, () =>
renameFileWithWindowsRetryAsync(temporary, path, isCurrent)
)
} finally {
await rm(temporary, { force: true }).catch(() => undefined)
}
// Cleanup can yield to a newer capture after this attempt's rename completed.
if (!published || !isCurrent()) {
throw new Error('Chat context snapshot publication was replaced')
}
return path
} finally {
if (isCurrent()) {
claims.delete(path)
}
}
}
79 changes: 79 additions & 0 deletions src/main/native-chat/agent-session-saved-prompt-text.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
import type {
AgentJournalRenderItem,
AgentJournalQuestionItem
} from '../../shared/agent-session-journal-types'
import { agentJournalItemRowOrigin } from '../../shared/agent-session-journal-position'
import { agentJournalLinkageFields } from '../../shared/agent-session-journal-producer'
import {
agentSessionPromptQuestions,
legacyAgentSessionQuestionAnswers
} from '../../shared/agent-session-question-answer'
import { nativeChatReceiptAnswers } from '../../shared/native-chat-resolution-receipt'
import type { NativeChatMessage } from '../../shared/native-chat-types'
import { projectStructuredItemToNativeChat } from '../../shared/structured-agent-session-projection'

function questionInterpretation(body: AgentJournalQuestionItem): string[] {
if (body.resolution.state !== 'resolved') {
return []
}
const selected = body.resolution.selectedOptionId
const answers =
body.resolution.answers ?? (selected ? legacyAgentSessionQuestionAnswers(body, selected) : null)
return agentSessionPromptQuestions(body).flatMap((question) => {
const heading = `Question ${JSON.stringify(question.id)}${question.header ? ` (${question.header})` : ''}`
const answer = answers?.find((entry) => entry.questionId === question.id)
if (!answer || (answer.optionIds.length === 0 && answer.other === undefined)) {
return [heading, 'Submitted answer: unavailable']
}
return [
heading,
...answer.optionIds.map((id) => {
const option = question.options.find((choice) => choice.id === id)
return option ? `Selected choice: ${option.label}` : 'Selected choice: unavailable'
}),
...(answer.other !== undefined
? [
body.resolution.answers !== undefined
? `Submitted free text: resolution.answers[${body.resolution.answers.indexOf(answer)}].other`
: `Legacy free text: ${JSON.stringify(answer.other)}`
]
: [])
]
})
}

/** Saved prompts are conversation content even while pending; display projection omits them. */
export function projectSavedChatContextItems(
items: readonly AgentJournalRenderItem[]
): NativeChatMessage[] {
return items.flatMap((item): NativeChatMessage[] => {
const body = item.body
if (body.kind !== 'question' && body.kind !== 'approval') {
const message = projectStructuredItemToNativeChat(item)
return message ? [message] : []
}
const interpretation =
body.kind === 'question'
? questionInterpretation(body)
: body.resolution.state === 'resolved'
? [`Selected choice: ${nativeChatReceiptAnswers(body)[0]?.answer ?? 'unavailable'}`]
: []
return [
{
...agentJournalItemRowOrigin(item),
...agentJournalLinkageFields(item),
role: 'system',
blocks: [
{
type: 'text',
text: [
`Prompt state: ${body.resolution.state}`,
`Saved prompt: ${JSON.stringify(body)}`,
...interpretation
].join('\n')
}
]
}
]
})
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import { projectSavedChatContextItems } from '../agent-session-saved-prompt-text'
import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types'
import type { AgentSessionContinuationContext } from '../../../shared/agent-session-wire'
import { agentSessionCurrentContextRows } from '../../../shared/agent-session-context-clear'
import { trimAgentSessionContext } from '../../../shared/agent-session-context-budget'
import { LOCAL_EXECUTION_HOST_ID } from '../../../shared/execution-host'
import { projectStructuredAgentSessionMessages } from '../../../shared/structured-agent-session-message-projection'
import { formatWorkerTranscriptMessage } from '../../../shared/worker-transcript-text'
import { toLinuxPath } from '../../../shared/wsl-paths'
import { writeAgentSessionContextSnapshot } from '../agent-session-context-snapshots'
import type {
StructuredAgentSessionHostDeps,
StructuredAgentSessionHostSession
} from './structured-agent-session-host-types'

export function continuationContextReader(
deps: StructuredAgentSessionHostDeps,
conversation: (id: string) => Promise<StructuredAgentSessionHostSession>
) {
return async (input: {
sessionId: string
workspaceId: string
expectedCursor: AgentJournalCursor
}): Promise<AgentSessionContinuationContext> => {
const sourceRecord = () => {
const record = deps.store.getRecord(input.sessionId)
return record?.location.workspaceId === input.workspaceId &&
record.location.executionHostId === LOCAL_EXECUTION_HOST_ID
? record
: null
}
if (!sourceRecord()) {
return { status: 'superseded' }
}
const session = await conversation(input.sessionId)
const record = sourceRecord()
const snapshot = session.journal.snapshot()
if (
!record ||
snapshot.cursor.epoch !== input.expectedCursor.epoch ||
snapshot.cursor.sequence < input.expectedCursor.sequence
) {
return { status: 'superseded' }
}
const context = agentSessionCurrentContextRows(snapshot.items, snapshot.submissions)
const text = projectStructuredAgentSessionMessages(
context.items,
[],
context.submissions,
projectSavedChatContextItems
)
.map(formatWorkerTranscriptMessage)
.join('\n\n')
const capturedText = trimAgentSessionContext(text, 'saved chat messages')
if (!text.trim()) {
return { status: 'current', capturedText: '' }
}
// Both outputs describe this stopping point, even if the conversation changes during the write.
const path = await writeAgentSessionContextSnapshot(
deps.journalDatabase.stateDirectory,
input.sessionId,
text
)
return {
status: 'current',
capturedText,
transcriptPath: record.location.wslDistro ? toLinuxPath(path) : path
}
}
}
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { continuationContextReader as contextReader } from './structured-agent-session-continuation-context'
import type { AgentSessionRewindParams } from '../../../shared/agent-session-rewind'
import { rewindStructuredAgentSession } from './structured-agent-session-rewind'
import { StructuredConversationCommandController } from './structured-conversation-command-controller'
Expand Down Expand Up @@ -166,6 +167,7 @@ export class StructuredAgentSessionHost {
closeStatus: (sessionId, options) => this.clientDelivery.closeSession(sessionId, options),
childWorkOpen: this.clientDelivery.childWorkOpen
})
this.continuationContext = contextReader(deps, this.lifetime.conversation)
this.runtimeState.startLeaseRenewal()
this.lifetime.idleSweep.start()
}
Expand Down Expand Up @@ -331,6 +333,8 @@ export class StructuredAgentSessionHost {
history: StructuredAgentSessionBackgroundTaskChannel['history'] = (request, scope) =>
this.backgroundTasks.history(request, scope)

readonly continuationContext: ReturnType<typeof contextReader>

/** The fully reduced timeline, for readers that cannot tolerate a page's ambiguity — rows are
* revised or tombstoned in place, so an item's ABSENCE from a bounded page proves nothing. */
journalSnapshot = async (sessionId: string) =>
Expand Down
Loading
Loading