fix(native-chat): settle cancelled structured prompts

This commit is contained in:
Merge Sim
2026-09-12 04:32:37 -07:00
parent 9497661e81
commit a2ee2015d3
63 changed files with 2197 additions and 390 deletions
@@ -62,8 +62,8 @@ const HOST_COMPONENT_NAMES = new Set([
'View'
])
const HEAD_MAIN_HOOK_SHA256 = 'c3e33699e3e3fa7e24408f3d4946fcc451e9b9419442d985c4ccde01782e5114'
const HEAD_HOOK_BINDING_SHA256 = '7f907e028893721d662eeee0aa9002ad1e00359948f39fb148d274596cd9b3c0'
const HEAD_MAIN_HOOK_SHA256 = '3e97fba5684419f936b7e49986d6c0a4accdd5b342d2dbde25e1f38bbada5dab'
const HEAD_HOOK_BINDING_SHA256 = 'a0f140869570f13c811e0e241581691d9ef03b23477bbedfd8d1ad72f0a21a3a'
const HEAD_CALLBACK_IDENTITY_SHA256 =
'2a9e4825df007f6ef53b81aa5004991d6318eee7507b44d625c07e630be432eb'
const HEAD_CALLBACK_BODY_SHA256 = 'af7f3c62954250d4be7ee432ecd10dc2689792aad8230fed2d1d68bbc892d776'
@@ -472,7 +472,7 @@ describe('mobile session route extraction parity', () => {
const contentBindings = CONTENT_COMPONENT_NAMES.flatMap(
(name) => readHookFacts(name, definitions).bindings
)
expect(main.hooks).toHaveLength(267)
expect(main.hooks).toHaveLength(268)
expect(hash(main.hooks)).toBe(HEAD_MAIN_HOOK_SHA256)
expect(hash(main.bindings)).toBe(HEAD_HOOK_BINDING_SHA256)
expect(main.callbacks).toHaveLength(77)
@@ -0,0 +1,130 @@
import type {
AgentJournalRenderItem,
AgentJournalResolution
} from '../../../src/shared/agent-session-journal-types'
import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire'
type SnapshotEvent = Extract<AgentSessionSubscribeEvent, { type: 'snapshot' }>
export function snapshotEvent(fence = 3): SnapshotEvent {
return {
type: 'snapshot',
sessionId: 'session-1',
fence,
page: {
sessionId: 'session-1',
epoch: 'epoch-1',
fence,
direction: 'tail',
items: [],
removedItemIds: [],
submissions: [],
window: {
oldest: null,
newest: null,
nextCursor: { epoch: 'epoch-1', sequence: 0 }
},
liveCursor: { epoch: 'epoch-1', sequence: 0 },
hasOlder: false,
hasNewer: false
}
}
}
export function snapshotWithMessage(): SnapshotEvent {
const event = snapshotEvent()
return {
...event,
page: {
...event.page,
items: [
{
itemId: 'msg-1',
revision: 1,
sequence: 1,
observedAt: 10,
body: {
kind: 'message',
role: 'user',
blocks: [{ type: 'text', text: 'sent before the blip' }]
}
}
],
window: {
oldest: { epoch: 'epoch-1', sequence: 1 },
newest: { epoch: 'epoch-1', sequence: 1 },
nextCursor: { epoch: 'epoch-1', sequence: 2 }
},
liveCursor: { epoch: 'epoch-1', sequence: 1 }
}
}
}
function pendingResolution(): AgentJournalResolution {
return {
state: 'pending',
selectedOptionId: null,
resolvedBy: null,
resolvedAt: null
}
}
export function approvalItem(): AgentJournalRenderItem {
return {
itemId: 'approval-1',
revision: 2,
sequence: 1,
observedAt: 10,
body: {
kind: 'approval',
title: 'Allow Bash?',
detail: 'rm -rf build',
options: [
{ id: 'allow-once', label: 'Allow once' },
{ id: 'deny', label: 'Deny' }
],
resolution: pendingResolution()
}
}
}
export function approvalItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem {
return { ...approvalItem(), itemId, revision }
}
export function questionItem(): AgentJournalRenderItem {
return {
itemId: 'question-1',
revision: 7,
sequence: 2,
observedAt: 12,
body: {
kind: 'question',
question: 'Pick destination',
freeTextQuestionId: 'free-q',
options: [
{ id: 'choice-a', label: 'Choice A' },
{ id: 'choice-b', label: 'Choice B' }
],
resolution: pendingResolution()
}
}
}
export function questionItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem {
return { ...questionItem(), itemId, revision }
}
export function runningStatusItem(): AgentJournalRenderItem {
return {
itemId: 'status-1',
revision: 1,
sequence: 3,
observedAt: 14,
body: {
kind: 'status',
text: 'Working',
turnLifecycle: { turnId: 'turn-1', state: 'running' }
}
}
}
@@ -35,6 +35,7 @@ export function useMobileNativeChatController(args: {
nativeChatInputLeaseReady: boolean
/** Live socket state; the lease collapses on disconnect but one render later. */
connState: ConnectionState
promptCancelSupported?: boolean
onSendError: (message: string) => void
/** Retires a held failure banner. Any accepted chat write clears it — a delivered
* answer or permission reply must not sit under a stale "not sent". */
@@ -51,6 +52,7 @@ export function useMobileNativeChatController(args: {
nativeChatTranscriptIsLocalReadable,
nativeChatInputLeaseReady,
connState,
promptCancelSupported,
onSendError,
onSendResolved
} = args
@@ -89,7 +91,9 @@ export function useMobileNativeChatController(args: {
sourceIdentity,
enabled: showNativeChat,
connState,
onSendError
promptCancelSupported,
onSendError,
onCancelResolved: onSendResolved
})
const {
composerText: chatComposerText,
@@ -16,7 +16,9 @@ export function useMobileNativeChatSessionLane({
sourceIdentity,
enabled,
connState,
onSendError
promptCancelSupported,
onSendError,
onCancelResolved
}: {
client: RpcClient | null
structured: boolean
@@ -29,7 +31,9 @@ export function useMobileNativeChatSessionLane({
sourceIdentity: Parameters<typeof useMobileNativeChatSession>[0]['sourceIdentity']
enabled: boolean
connState: ConnectionState
promptCancelSupported?: boolean
onSendError: (message: string) => void
onCancelResolved: () => void
}): {
structuredSession: ReturnType<typeof useMobileStructuredAgentSession>
session: ReturnType<typeof useMobileNativeChatSession>
@@ -50,7 +54,9 @@ export function useMobileNativeChatSessionLane({
// reacquire the provider without clearing the cached transcript.
connected: connState === 'connected',
agent: structured ? agent : null,
onSendError
promptCancelSupported,
onSendError,
onCancelResolved
})
return {
structuredSession,
@@ -1,5 +1,7 @@
import { useEffect, useCallback } from 'react'
import { useFocusEffect } from 'expo-router'
import { AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY } from '../../../src/shared/protocol-version'
import { useHostProtocolGates } from '../components/HostProtocolGate'
import { useMobileDictation } from '../hooks/use-mobile-dictation'
import { triggerError } from '../platform/haptics'
import {
@@ -22,6 +24,7 @@ export function useMobileSessionNativeChatDictation(
scope: MobileSessionFeedbackCapabilitiesModel,
sendLiveTerminalInput: (handle: string, bytes: string) => Promise<boolean>
) {
const { hostCapabilities, statusPending } = useHostProtocolGates()
const {
hostId,
worktreeId,
@@ -72,6 +75,9 @@ export function useMobileSessionNativeChatDictation(
nativeChatTranscriptIsLocalReadable,
nativeChatInputLeaseReady,
connState,
promptCancelSupported: statusPending
? undefined
: hostCapabilities.includes(AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY),
onSendError: nativeChatSendError.show,
onSendResolved: nativeChatSendError.clear
})
@@ -0,0 +1,90 @@
import { useCallback } from 'react'
import type { AgentSessionCancelResult } from '../../../src/shared/agent-session-wire'
import { activeStructuredAgentSessionTurnId } from '../../../src/shared/structured-agent-session-live-turn'
import { AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE } from '../../../src/shared/protocol-version'
import type { RpcClient } from '../transport/rpc-client'
import {
requestStructuredAgentSessionMutation,
retainStructuredSessionOperationId
} from './mobile-structured-agent-session-rpc'
import {
pendingStructuredApproval,
pendingStructuredQuestion
} from './mobile-structured-agent-prompts'
import type { useMobileStructuredAgentState } from './use-mobile-structured-agent-state'
export function useMobileStructuredAgentSessionCancel(args: {
client: RpcClient | null
sessionId: string | null
sessionKey: string
enabled: boolean
promptCancelSupported?: boolean
stateRef: ReturnType<typeof useMobileStructuredAgentState>['stateRef']
operationIdsRef: { current: Map<string, string> }
onSendError: (message: string) => void
onCancelResolved?: () => void
}): () => void {
const {
client,
sessionId,
sessionKey,
enabled,
promptCancelSupported,
stateRef,
operationIdsRef,
onSendError,
onCancelResolved
} = args
return useCallback(() => {
const current = stateRef.current
const turnId = activeStructuredAgentSessionTurnId(current.items)
const prompt =
current.items.find(pendingStructuredApproval) ?? current.items.find(pendingStructuredQuestion)
if (!client || !sessionId || !enabled || current.fence === null || (!turnId && !prompt)) {
onSendError('Stop not sent')
return
}
if (prompt && promptCancelSupported !== true) {
onSendError(
promptCancelSupported === false
? AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE
: 'Checking desktop capabilities — try again in a moment'
)
return
}
const fields = {
...(turnId ? { turnId } : {}),
...(promptCancelSupported && prompt
? { prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } }
: {})
}
const key = `${sessionKey}:agentSession.cancel:${JSON.stringify(fields)}`
const clientOperationId = retainStructuredSessionOperationId(
operationIdsRef.current,
key,
operationIdsRef.current.get(key)
)
void requestStructuredAgentSessionMutation<AgentSessionCancelResult>({
client,
method: 'agentSession.cancel',
fingerprintMethod: 'agentSession.cancel',
sessionId,
expectedRuntimeFence: current.fence,
fields,
clientOperationId
}).then((result) => {
if (result.status !== 'unknown') {
operationIdsRef.current.delete(key)
}
if (result.status === 'unknown') {
onSendError('Stop unconfirmed — check chat before retrying')
} else if (result.status === 'refused') {
onSendError(result.message)
} else if (result.status === 'failed') {
onSendError(result.message === 'Request not sent' ? 'Stop not sent' : result.message)
} else {
onCancelResolved?.()
}
})
}, [client, enabled, onCancelResolved, onSendError, promptCancelSupported, sessionId, sessionKey])
}
@@ -1,143 +1,24 @@
import { createElement } from 'react'
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type {
AgentJournalRenderItem,
AgentJournalResolution
} from '../../../src/shared/agent-session-journal-types'
import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire'
import type { RpcClient } from '../transport/rpc-client'
import { markRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity'
import { formatQuestionFreeTextAnswer } from './mobile-native-chat-question'
import { useMobileStructuredAgentSession } from './use-mobile-structured-agent-session'
import {
approvalItem,
approvalItemWithIdentity,
questionItem,
questionItemWithIdentity,
runningStatusItem,
snapshotEvent,
snapshotWithMessage
} from './mobile-structured-agent-session-test-fixtures'
function ok(result: unknown) {
return { ok: true, result, _meta: { runtimeId: 'runtime-1' } }
}
function snapshotEvent(fence = 3): AgentSessionSubscribeEvent {
return {
type: 'snapshot',
sessionId: 'session-1',
fence,
page: {
sessionId: 'session-1',
epoch: 'epoch-1',
fence,
direction: 'tail',
items: [],
removedItemIds: [],
submissions: [],
window: {
oldest: null,
newest: null,
nextCursor: { epoch: 'epoch-1', sequence: 0 }
},
liveCursor: { epoch: 'epoch-1', sequence: 0 },
hasOlder: false,
hasNewer: false
}
}
}
function snapshotWithMessage(): AgentSessionSubscribeEvent {
const event = snapshotEvent()
return {
...event,
page: {
...event.page,
items: [
{
itemId: 'msg-1',
revision: 1,
sequence: 1,
observedAt: 10,
body: {
kind: 'message',
role: 'user',
blocks: [{ type: 'text', text: 'sent before the blip' }]
}
}
],
window: {
oldest: { epoch: 'epoch-1', sequence: 1 },
newest: { epoch: 'epoch-1', sequence: 1 },
nextCursor: { epoch: 'epoch-1', sequence: 2 }
},
liveCursor: { epoch: 'epoch-1', sequence: 1 }
}
} as AgentSessionSubscribeEvent
}
function pendingResolution(): AgentJournalResolution {
return {
state: 'pending',
selectedOptionId: null,
resolvedBy: null,
resolvedAt: null
}
}
function approvalItem(): AgentJournalRenderItem {
return {
itemId: 'approval-1',
revision: 2,
sequence: 1,
observedAt: 10,
body: {
kind: 'approval',
title: 'Allow Bash?',
detail: 'rm -rf build',
options: [
{ id: 'allow-once', label: 'Allow once' },
{ id: 'deny', label: 'Deny' }
],
resolution: pendingResolution()
}
}
}
function approvalItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem {
return { ...approvalItem(), itemId, revision }
}
function questionItem(): AgentJournalRenderItem {
return {
itemId: 'question-1',
revision: 7,
sequence: 2,
observedAt: 12,
body: {
kind: 'question',
question: 'Pick destination',
freeTextQuestionId: 'free-q',
options: [
{ id: 'choice-a', label: 'Choice A' },
{ id: 'choice-b', label: 'Choice B' }
],
resolution: pendingResolution()
}
}
}
function questionItemWithIdentity(itemId: string, revision: number): AgentJournalRenderItem {
return { ...questionItem(), itemId, revision }
}
function runningStatusItem(): AgentJournalRenderItem {
return {
itemId: 'status-1',
revision: 1,
sequence: 3,
observedAt: 14,
body: {
kind: 'status',
text: 'Working',
turnLifecycle: { turnId: 'turn-1', state: 'running' }
}
}
}
async function defaultSendRequest(method: string, params?: Record<string, unknown>) {
if (method === 'agentSession.send') {
return ok({
@@ -217,6 +98,7 @@ describe('useMobileStructuredAgentSession', () => {
let hook: ReturnType<typeof useMobileStructuredAgentSession> | null = null
let listener: ((value: unknown) => void) | null = null
const onSendError = vi.fn()
const onCancelResolved = vi.fn()
const unsubscribe = vi.fn()
const sendRequest = vi.fn(defaultSendRequest)
const subscribe = vi.fn((_method: string, _params: unknown, onData: (value: unknown) => void) => {
@@ -232,11 +114,13 @@ describe('useMobileStructuredAgentSession', () => {
sessionId = 'session-1',
agent = 'codex',
connected = true,
promptCancelSupported = false,
sourceIdentity = 'host-a\0workspace-a'
}: {
sessionId?: string | null
agent?: string | null
connected?: boolean
promptCancelSupported?: boolean
sourceIdentity?: string
}): null {
hook = useMobileStructuredAgentSession({
@@ -246,7 +130,9 @@ describe('useMobileStructuredAgentSession', () => {
enabled: true,
connected,
agent,
onSendError
promptCancelSupported,
onSendError,
onCancelResolved
} as never)
return null
}
@@ -760,6 +646,92 @@ describe('useMobileStructuredAgentSession', () => {
expect(onSendError).toHaveBeenCalledWith('Stop unconfirmed — check chat before retrying')
})
it('targets the displayed prompt revision when the host supports prompt cancellation', async () => {
act(() => {
renderer = create(createElement(Harness, { promptCancelSupported: true }))
})
await vi.waitFor(() => expect(listener).toEqual(expect.any(Function)))
act(() =>
listener?.({
...snapshotEvent(3),
page: {
...snapshotEvent(3).page,
items: [approvalItem()]
}
})
)
act(() => hook!.cancel())
await vi.waitFor(() =>
expect(sendRequest).toHaveBeenCalledWith(
'agentSession.cancel',
expect.objectContaining({
prompt: { itemId: 'approval-1', expectedRevision: 2 }
}),
expect.any(Object)
)
)
const cancelParams = sendRequest.mock.calls.find(
([method]) => method === 'agentSession.cancel'
)?.[1]
expect(cancelParams).not.toHaveProperty('turnId')
})
it('retires a held error after the host accepts structured Stop', async () => {
act(() => {
renderer = create(createElement(Harness))
})
await vi.waitFor(() => expect(listener).toEqual(expect.any(Function)))
act(() =>
listener?.({
...snapshotEvent(3),
page: { ...snapshotEvent(3).page, items: [runningStatusItem()] }
})
)
sendRequest.mockImplementation(async (method, params) =>
method === 'agentSession.cancel'
? ok({
ok: true,
replayed: false,
fence: 3,
cursor: { epoch: 'epoch-1', sequence: 4 },
value: { turnId: 'turn-1', cancelled: true }
})
: defaultSendRequest(method, params)
)
await act(async () => {
hook!.cancel()
await Promise.resolve()
})
expect(onCancelResolved).toHaveBeenCalledOnce()
})
it('refuses prompt cancellation for a host without prompt cancellation', async () => {
act(() => {
renderer = create(createElement(Harness))
})
await vi.waitFor(() => expect(listener).toEqual(expect.any(Function)))
act(() =>
listener?.({
...snapshotEvent(3),
page: {
...snapshotEvent(3).page,
items: [runningStatusItem(), approvalItem()]
}
})
)
act(() => hook!.cancel())
expect(sendRequest.mock.calls.some(([method]) => method === 'agentSession.cancel')).toBe(false)
expect(onSendError).toHaveBeenCalledWith(
'Cancelling a pending prompt requires a newer Orca server. Update the server and try again.'
)
})
it('releases a landed hold when the structured tab unmounts', async () => {
act(() => {
renderer = create(createElement(Harness))
@@ -1,9 +1,6 @@
import { useCallback, useEffect, useMemo, useRef } from 'react'
import { dispatchMobileStructuredCommand } from './mobile-structured-composer-command'
import type {
AgentSessionCancelResult,
AgentSessionSendResult
} from '../../../src/shared/agent-session-wire'
import type { AgentSessionSendResult } from '../../../src/shared/agent-session-wire'
import {
structuredAgentSessionSendBody,
type StructuredAgentSessionAttachment
@@ -38,6 +35,7 @@ import { useMobileStructuredAgentState } from './use-mobile-structured-agent-sta
import { useMobileStructuredPromptResponses } from './use-mobile-structured-prompt-responses'
import { useMobileStructuredAgentOptions } from './use-mobile-structured-agent-options'
import { useMobileStructuredAgentTurnTiming } from './use-mobile-structured-agent-turn-timing'
import { useMobileStructuredAgentSessionCancel } from './use-mobile-structured-agent-session-cancel'
type StructuredMobileAttachment = StructuredAgentSessionAttachment & { id?: string }
@@ -70,9 +68,21 @@ export function useMobileStructuredAgentSession(args: {
/** Live transport only; gates the connection-scoped hold, nothing else. */
connected: boolean
agent: string | null
promptCancelSupported?: boolean
onSendError: (message: string) => void
onCancelResolved?: () => void
}): StructuredMobileSession {
const { agent, client, connected, sessionId, sourceIdentity = '', enabled, onSendError } = args
const {
agent,
client,
connected,
sessionId,
sourceIdentity = '',
enabled,
promptCancelSupported,
onSendError,
onCancelResolved
} = args
const sessionKey = encodeNativeChatTranscriptIdentity([sourceIdentity, agent, sessionId])
const operationIdsRef = useRef(new Map<string, string>())
const commandPendingRef = useRef(false)
@@ -237,37 +247,17 @@ export function useMobileStructuredAgentSession(args: {
onSendError
})
const cancel = useCallback(() => {
const current = stateRef.current
const turnId = activeStructuredAgentSessionTurnId(current.items)
if (!client || !sessionId || !enabled || current.fence === null || !turnId) {
onSendError('Stop not sent')
return
}
const fields = { turnId }
const key = `${sessionKey}:agentSession.cancel:${JSON.stringify(fields)}`
const clientOperationId = retainOperationId(key, operationIdsRef.current.get(key))
void requestStructuredAgentSessionMutation<AgentSessionCancelResult>({
client,
method: 'agentSession.cancel',
fingerprintMethod: 'agentSession.cancel',
sessionId,
expectedRuntimeFence: current.fence,
fields,
clientOperationId
}).then((result) => {
if (result.status !== 'unknown') {
operationIdsRef.current.delete(key)
}
if (result.status === 'unknown') {
onSendError('Stop unconfirmed — check chat before retrying')
} else if (result.status === 'refused') {
onSendError(result.message)
} else if (result.status === 'failed') {
onSendError(result.message === 'Request not sent' ? 'Stop not sent' : result.message)
}
})
}, [client, enabled, onSendError, sessionId, sessionKey])
const cancel = useMobileStructuredAgentSessionCancel({
client,
sessionId,
sessionKey,
enabled,
promptCancelSupported,
stateRef,
operationIdsRef,
onSendError,
onCancelResolved
})
const messages = useMemo(
() => projectStructuredAgentSessionMessages(state.items, [], state.submissions),
+29 -57
View File
@@ -1,9 +1,10 @@
import { useEffect, useState } from 'react'
import type { RpcClient } from './rpc-client'
import type { ConnectionState, RpcSuccess } from './types'
import type { ConnectionState } from './types'
import { evaluateCompat, type CompatVerdict } from './protocol-compat'
import type { DesktopStatus } from '../worktree/host-worktree-rpc-types'
import { normalizeHostAppVersion, recordHostAppVersion } from './host-app-version-store'
import { startRuntimeStatusProbe } from './runtime-capability-probe'
export type HostStatusGates = {
hostCapabilities: string[]
@@ -39,68 +40,39 @@ export function useHostStatusGates(args: {
setUnverified(true)
return
}
let cancelled = false
const requestClient = client
const settle = (gates: Omit<HostStatusGates, 'statusPending'>) => {
setLoaded({ hostId, client: requestClient, ...gates })
setUnverified(false)
}
void (async () => {
try {
const response = await requestClient.sendRequest('status.get')
if (cancelled) {
return
}
if (!response.ok) {
settle({
hostCapabilities: [],
floatingWorkspaceEnabled: false,
desktopAppVersion: null,
compatVerdict: { kind: 'ok' }
})
return
}
const status = (response as RpcSuccess).result as DesktopStatus & {
capabilities?: string[]
}
const verdict = evaluateCompat({
desktopProtocolVersion: status.protocolVersion,
desktopMinCompatibleMobileVersion: status.minCompatibleMobileVersion
})
const desktopAppVersion = normalizeHostAppVersion(status.appVersion)
if (hostId && desktopAppVersion) {
void recordHostAppVersion(hostId, desktopAppVersion)
}
settle({
hostCapabilities: status.capabilities ?? [],
floatingWorkspaceEnabled: status.floatingWorkspaceEnabled === true,
desktopAppVersion,
compatVerdict: verdict
})
if (verdict.kind === 'blocked') {
// Why: support breadcrumb to confirm a block fired vs a render bug; no PII, just version ints.
console.warn('[protocol-compat] blocked', {
reason: verdict.reason,
desktopVersion: verdict.desktopVersion,
requiredMobileVersion: verdict.requiredMobileVersion,
requiredDesktopVersion: verdict.requiredDesktopVersion
})
}
} catch {
// Why: a transient status failure must not trap navigation; conservative feature gates remain disabled.
if (!cancelled) {
settle({
hostCapabilities: [],
floatingWorkspaceEnabled: false,
desktopAppVersion: null,
compatVerdict: { kind: 'ok' }
})
}
return startRuntimeStatusProbe(requestClient, (result) => {
const status = result as DesktopStatus & {
capabilities?: string[]
}
})()
return () => {
cancelled = true
}
const verdict = evaluateCompat({
desktopProtocolVersion: status.protocolVersion,
desktopMinCompatibleMobileVersion: status.minCompatibleMobileVersion
})
const desktopAppVersion = normalizeHostAppVersion(status.appVersion)
if (hostId && desktopAppVersion) {
void recordHostAppVersion(hostId, desktopAppVersion)
}
settle({
hostCapabilities: status.capabilities ?? [],
floatingWorkspaceEnabled: status.floatingWorkspaceEnabled === true,
desktopAppVersion,
compatVerdict: verdict
})
if (verdict.kind === 'blocked') {
// Why: support breadcrumb to confirm a block fired vs a render bug; no PII, just version ints.
console.warn('[protocol-compat] blocked', {
reason: verdict.reason,
desktopVersion: verdict.desktopVersion,
requiredMobileVersion: verdict.requiredMobileVersion,
requiredDesktopVersion: verdict.requiredDesktopVersion
})
}
})
}, [client, connState, hostId])
// Why: effects run after render, so key loaded gates by host and client to fail closed during route reuse.
@@ -1,5 +1,5 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { startRuntimeCapabilityProbe } from './runtime-capability-probe'
import { startRuntimeCapabilityProbe, startRuntimeStatusProbe } from './runtime-capability-probe'
import { LogicalClientCutoverError } from './stable-logical-rpc-client'
import type { RpcClient } from './rpc-client'
import type { RpcResponse } from './types'
@@ -48,6 +48,26 @@ describe('startRuntimeCapabilityProbe', () => {
cancel()
})
it('shares retry semantics with full host-status consumers', async () => {
const status = {
ok: true as const,
id: '1',
result: { appVersion: '1.2.3', capabilities: ['a.v1'] },
_meta: { runtimeId: 'r1' }
}
const { client, calls } = makeClient([new Error('timeout'), status])
const seen: unknown[] = []
const cancel = startRuntimeStatusProbe(client, (result) => seen.push(result))
await flushMicrotasks()
expect(seen).toEqual([])
await vi.advanceTimersByTimeAsync(1_000)
expect(seen).toEqual([status.result])
expect(calls()).toBe(2)
cancel()
})
it('treats malformed capabilities as unsupported', async () => {
const response: RpcResponse = {
ok: true,
@@ -12,6 +12,23 @@ const FAILURE_RETRY_MAX_DELAY_MS = 15_000
export function startRuntimeCapabilityProbe(
client: Pick<RpcClient, 'sendRequest'>,
onCapabilities: (capabilities: readonly string[]) => void
): () => void {
return startRuntimeStatusProbe(client, (result) => {
const rawCapabilities =
result && typeof result === 'object'
? (result as { capabilities?: unknown }).capabilities
: null
const capabilities =
Array.isArray(rawCapabilities) && rawCapabilities.every((value) => typeof value === 'string')
? rawCapabilities
: []
onCapabilities(capabilities)
})
}
export function startRuntimeStatusProbe(
client: Pick<RpcClient, 'sendRequest'>,
onStatus: (status: unknown) => void
): () => void {
let cancelled = false
let retryTimer: ReturnType<typeof setTimeout> | null = null
@@ -27,17 +44,7 @@ export function startRuntimeCapabilityProbe(
scheduleRetry(false)
return
}
const result = (response as RpcSuccess).result
const rawCapabilities =
result && typeof result === 'object'
? (result as { capabilities?: unknown }).capabilities
: null
const capabilities =
Array.isArray(rawCapabilities) &&
rawCapabilities.every((value) => typeof value === 'string')
? rawCapabilities
: []
onCapabilities(capabilities)
onStatus((response as RpcSuccess).result)
},
(error: unknown) => {
if (cancelled) {
@@ -0,0 +1,46 @@
import type { ClaudePendingPrompt } from './claude-structured-prompt-types'
export class ClaudeHostPromptCancellation {
private readonly active = new Set<ClaudePendingPrompt>()
private readonly suppressedProviderEvents = new Set<ClaudePendingPrompt>()
begin(prompt: ClaudePendingPrompt | null, turnId: string): ClaudePendingPrompt | null {
if (!prompt || prompt.turnId !== turnId) {
return null
}
this.active.add(prompt)
return prompt
}
finish(
prompt: ClaudePendingPrompt,
confirmed: boolean,
isPending: () => boolean,
settle: () => void
): boolean {
if (confirmed && isPending()) {
settle()
prompt.settle(null)
}
this.active.delete(prompt)
const suppressed = this.suppressedProviderEvents.delete(prompt)
return suppressed && !confirmed
}
consume(prompt: ClaudePendingPrompt): boolean {
const consumed = this.active.delete(prompt)
if (consumed) {
this.suppressedProviderEvents.add(prompt)
}
return consumed
}
forget(prompt: ClaudePendingPrompt): void {
this.active.delete(prompt)
}
clear(): void {
this.active.clear()
this.suppressedProviderEvents.clear()
}
}
@@ -87,6 +87,26 @@ describe('cancelClaudeTurn', () => {
})
describe('answerClaudePrompt', () => {
it('groups journal rows only under the live turn that owns the prompt', () => {
const prompts = new ClaudePromptRegistry()
const prompt = prompts.register({
requestId: 'question-1',
turnId: 'turn-1',
toolName: 'AskUserQuestion',
toolUseId: 'tool-1',
input: { questions: [{ question: 'One?' }, { question: 'Two?' }] },
suggestions: [],
settle: vi.fn()
})!
prompts.bindJournalItemId('journal-q1', prompt.promptKey, 'One?')
prompts.bindJournalItemId('journal-q2', prompt.promptKey, 'Two?')
expect(prompts.cancellation('journal-q1')).toEqual({
turnId: 'turn-1',
itemIds: ['journal-q1', 'journal-q2']
})
})
it('settles the pending prompt callback and forgets it', async () => {
const prompts = new ClaudePromptRegistry()
const settle = vi.fn()
@@ -30,6 +30,7 @@ function callbacksFor() {
const { canUseTool, onUserDialog } = buildClaudePermissionCallbacks({
sessionId: 'session-1',
prompts,
activeTurnId: () => 'turn-1',
emit
})
return { prompts, emit, canUseTool, onUserDialog }
@@ -95,6 +96,24 @@ describe('Claude permission callbacks', () => {
expect(control.prompts.find('perm-3')).toBeNull()
})
it('does not tombstone a prompt whose abort belongs to confirmed host cancellation', async () => {
const control = callbacksFor()
const controller = new AbortController()
const answered = control.canUseTool(
'Bash',
{ command: 'ls' },
permissionOptions('perm-host-cancel', 'tool-host-cancel', controller.signal)
)
control.prompts.bindJournalItemId('journal-host-cancel', 'perm-host-cancel')
expect(control.prompts.beginHostCancellation('journal-host-cancel', 'turn-1')).not.toBeNull()
controller.abort()
await expect(answered).resolves.toBeNull()
expect(control.emit).toHaveBeenCalledTimes(1)
expect(control.emit).toHaveBeenCalledWith(expect.objectContaining({ type: 'prompt' }))
})
it('cancels a request whose abort raced ahead of delivery without emitting a prompt', async () => {
const control = callbacksFor()
const controller = new AbortController()
@@ -24,6 +24,7 @@ export type ClaudeBlockingControlSubtype = keyof typeof CLAUDE_BLOCKING_CONTROL_
export type ClaudePermissionCallbackDeps = {
sessionId: string
prompts: ClaudePromptRegistry
activeTurnId?: () => string | null
emit: (event: ClaudeStructuredSessionEvent) => void
}
@@ -53,6 +54,7 @@ export function buildClaudePermissionCallbacks(deps: ClaudePermissionCallbackDep
new Promise<PermissionResult | null>((resolve) => {
const prompt = deps.prompts.register({
requestId: options.requestId,
turnId: deps.activeTurnId?.() ?? null,
toolName,
toolUseId: options.toolUseID,
input,
@@ -64,12 +66,15 @@ export function buildClaudePermissionCallbacks(deps: ClaudePermissionCallbackDep
return
}
const cancel = (): void => {
const hostCancellation = deps.prompts.consumeHostCancellation(prompt)
if (deps.prompts.forgetIfPending(prompt)) {
deps.emit({
type: 'prompt-cancelled',
sessionId: deps.sessionId,
promptKey: prompt.promptKey
})
if (!hostCancellation) {
deps.emit({
type: 'prompt-cancelled',
sessionId: deps.sessionId,
promptKey: prompt.promptKey
})
}
// Null is the SDK's "no response written" sentinel: a cancelled request must not
// be answered, only forgotten.
resolve(null)
@@ -1,32 +1,17 @@
import { decodeAgentSessionQuestionAnswers } from '../../shared/agent-session-question-answer'
import { ClaudeHostPromptCancellation } from './claude-host-prompt-cancellation'
import type {
ClaudePendingPrompt,
ClaudePromptRegistration
} from './claude-structured-prompt-types'
export const CLAUDE_APPROVAL_DECISIONS = ['allow', 'allowForSession', 'deny', 'cancel'] as const
export type ClaudeApprovalDecision = (typeof CLAUDE_APPROVAL_DECISIONS)[number]
/** Settles the SDK's `canUseTool` promise; `null` is the SDK's "no response written" sentinel. */
export type ClaudePromptSettle = (response: Record<string, unknown> | null) => void
export type ClaudePendingPrompt = {
requestId: string
promptKey: string
toolUseId: string
toolName: string
kind: 'approval' | 'question'
input: Record<string, unknown>
suggestions: unknown[]
questionIds: readonly string[]
answers: Map<string, string | readonly string[]>
settle: ClaudePromptSettle
}
export type ClaudePromptRegistration = {
requestId: string
toolName: string
toolUseId: string
input: Record<string, unknown>
suggestions: unknown[]
settle: ClaudePromptSettle
}
export type {
ClaudePendingPrompt,
ClaudePromptRegistration,
ClaudePromptSettle
} from './claude-structured-prompt-types'
type PromptBinding = {
address: string
@@ -108,6 +93,7 @@ export function decodeClaudeQuestionOptionId(
export class ClaudePromptRegistry {
private readonly prompts = new Map<string, ClaudePendingPrompt>()
private readonly journalBindings = new Map<string, PromptBinding>()
private readonly hostCancellation = new ClaudeHostPromptCancellation()
register(registration: ClaudePromptRegistration): ClaudePendingPrompt | null {
const toolUseId = readString(registration.toolUseId)
@@ -120,6 +106,7 @@ export class ClaudePromptRegistry {
const prompt: ClaudePendingPrompt = {
requestId: registration.requestId,
promptKey: registration.requestId,
turnId: registration.turnId ?? null,
toolUseId,
toolName,
kind: questions.length > 0 ? 'question' : 'approval',
@@ -133,7 +120,6 @@ export class ClaudePromptRegistry {
return prompt
}
/** True only if the prompt was still pending; lets an abort and an answer race settle once. */
forgetIfPending(prompt: ClaudePendingPrompt): boolean {
if (!this.prompts.has(prompt.promptKey)) {
return false
@@ -142,6 +128,23 @@ export class ClaudePromptRegistry {
return true
}
beginHostCancellation(itemId: string, turnId: string): ClaudePendingPrompt | null {
return this.hostCancellation.begin(this.find(itemId)?.prompt ?? null, turnId)
}
finishHostCancellation(prompt: ClaudePendingPrompt, confirmed: boolean): boolean {
return this.hostCancellation.finish(
prompt,
confirmed,
() => this.prompts.get(prompt.promptKey) === prompt,
() => this.forget(prompt)
)
}
consumeHostCancellation(prompt: ClaudePendingPrompt): boolean {
return this.hostCancellation.consume(prompt)
}
bindJournalItemId(journalItemId: string, promptKey: string, questionIdForItem?: string): void {
this.journalBindings.set(journalItemId, {
address: promptKey,
@@ -157,6 +160,20 @@ export class ClaudePromptRegistry {
: null
}
cancellation(itemId: string): { turnId: string; itemIds: readonly string[] } | null {
const binding = this.journalBindings.get(itemId)
const prompt = binding ? this.prompts.get(binding.address) : undefined
if (!binding || !prompt?.turnId) {
return null
}
return {
turnId: prompt.turnId,
itemIds: [...this.journalBindings]
.filter(([, candidate]) => candidate.address === binding.address)
.map(([journalItemId]) => journalItemId)
}
}
cancel(requestId: string): ClaudePendingPrompt | null {
const prompt = this.prompts.get(requestId) ?? null
if (prompt) {
@@ -167,6 +184,7 @@ export class ClaudePromptRegistry {
forget(prompt: ClaudePendingPrompt): void {
this.prompts.delete(prompt.promptKey)
this.hostCancellation.forget(prompt)
for (const [itemId, binding] of this.journalBindings) {
if (binding.address === prompt.promptKey) {
this.journalBindings.delete(itemId)
@@ -178,6 +196,7 @@ export class ClaudePromptRegistry {
const pending = [...this.prompts.values()]
this.prompts.clear()
this.journalBindings.clear()
this.hostCancellation.clear()
return pending
}
}
@@ -0,0 +1,25 @@
export type ClaudePromptSettle = (response: Record<string, unknown> | null) => void
export type ClaudePendingPrompt = {
requestId: string
promptKey: string
turnId?: string | null
toolUseId: string
toolName: string
kind: 'approval' | 'question'
input: Record<string, unknown>
suggestions: unknown[]
questionIds: readonly string[]
answers: Map<string, string | readonly string[]>
settle: ClaudePromptSettle
}
export type ClaudePromptRegistration = {
requestId: string
turnId?: string | null
toolName: string
toolUseId: string
input: Record<string, unknown>
suggestions: unknown[]
settle: ClaudePromptSettle
}
@@ -137,6 +137,7 @@ export async function acquireClaudeSession({
const { canUseTool, onUserDialog } = buildClaudePermissionCallbacks({
sessionId,
prompts,
activeTurnId: () => liveSession?.activeTurnId ?? null,
emit: (event) =>
callbacks.deliver(attempt, sessionId, () => callbacks.emit(liveSession, input.events, event))
})
@@ -7,11 +7,116 @@ import { claudeUnwrittenUserMessageError } from './claude-agent-sdk-user-message
import {
acquired,
fakeClaude,
invokeCanUseTool,
PROVIDER_SESSION_ID,
USER_MESSAGE
} from './claude-structured-session-test-support'
describe('ClaudeStructuredSessionAdapter turns and controls', () => {
it('binds a prompt cancellation to the provider turn that created it', async () => {
const claude = fakeClaude({ replayUuid: 'turn-1' })
const adapter = await acquired(claude)
await adapter.dispatch({
sessionId: 'session-1',
clientMessageId: 'client-1',
body: USER_MESSAGE,
fence: 7
})
invokeCanUseTool(claude.connections[0], 'Bash', 'permission-1', 'tool-1')
adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1')
expect(
adapter.promptCancellation?.({
sessionId: 'session-1',
itemId: 'journal-approval',
fence: 7
})
).toEqual({ turnId: 'turn-1', itemIds: ['journal-approval'] })
await expect(
adapter.cancelTurn({
sessionId: 'session-1',
turnId: 'turn-1',
fence: 7,
promptItemId: 'unrelated-prompt'
})
).resolves.toEqual({ cancelled: false })
expect(claude.connections[0].calls.some((call) => call.subtype === 'interrupt')).toBe(false)
})
it('keeps the cancelled receipt when the matching SDK prompt aborts during interrupt', async () => {
const controller = new AbortController()
const events: Parameters<typeof acquired>[2] = []
const claude = fakeClaude({
replayUuid: 'turn-1',
routes: { interrupt: () => controller.abort() }
})
const adapter = await acquired(claude, {}, events)
await adapter.dispatch({
sessionId: 'session-1',
clientMessageId: 'client-1',
body: USER_MESSAGE,
fence: 7
})
const pending = invokeCanUseTool(claude.connections[0], 'Bash', 'permission-1', 'tool-1', {
signal: controller.signal
})
adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1')
await expect(
adapter.cancelTurn({
sessionId: 'session-1',
turnId: 'turn-1',
fence: 7,
promptItemId: 'journal-approval'
})
).resolves.toEqual({ cancelled: true })
await expect(pending.promise).resolves.toBeNull()
expect(events.some((event) => event.type === 'prompt-cancelled')).toBe(false)
expect(
adapter.promptCancellation?.({
sessionId: 'session-1',
itemId: 'journal-approval',
fence: 7
})
).toBeNull()
})
it('publishes an SDK prompt abort when provider interruption is refused', async () => {
const controller = new AbortController()
const events: Parameters<typeof acquired>[2] = []
const claude = fakeClaude({
replayUuid: 'turn-1',
routes: {
interrupt: () => {
controller.abort()
throw new ClaudeControlRequestError('interrupt', 'not running')
}
}
})
const adapter = await acquired(claude, {}, events)
await adapter.dispatch({
sessionId: 'session-1',
clientMessageId: 'client-1',
body: USER_MESSAGE,
fence: 7
})
const pending = invokeCanUseTool(claude.connections[0], 'Bash', 'permission-1', 'tool-1', {
signal: controller.signal
})
adapter.bindPromptItemId('session-1', 'journal-approval', 'permission-1')
await expect(
adapter.cancelTurn({
sessionId: 'session-1',
turnId: 'turn-1',
fence: 7,
promptItemId: 'journal-approval'
})
).resolves.toEqual({ cancelled: false })
await expect(pending.promise).resolves.toBeNull()
expect(events.at(-1)).toMatchObject({ type: 'prompt-cancelled', promptKey: 'permission-1' })
})
it("admits a dispatch on the write and names it from Claude's replay", async () => {
const claude = fakeClaude({ replayUuid: 'user-provider-uuid' })
const settled = vi.fn()
@@ -4,11 +4,7 @@ import type {
StructuredAgentSessionAcquireInput,
StructuredAgentSessionAdapter
} from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import {
answerClaudePrompt,
cancelClaudeTurn,
stopClaudeBackgroundTasks
} from './claude-structured-control-actions'
import { answerClaudePrompt, stopClaudeBackgroundTasks } from './claude-structured-control-actions'
import { dispatchClaudeTurn } from './claude-structured-dispatch'
import { StructuredSessionCompaction } from '../native-chat/agent-session-wire/structured-session-compaction'
import { releaseClaudeAcquisition } from './claude-structured-acquisition-release'
@@ -32,6 +28,7 @@ import {
} from './claude-structured-session-close'
import { readClaudeTranscriptLeafWithReproof } from './claude-transcript-branch-proof'
import type { AgentSessionBackgroundTaskState } from '../../shared/agent-session-wire'
import { cancelClaudeStructuredTurn } from './claude-structured-session-cancel'
export type { ClaudeStructuredLaunch } from './claude-structured-launch-resolution'
export type {
@@ -223,25 +220,35 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
compact: NonNullable<StructuredAgentSessionAdapter['compact']> = (input) =>
compactClaudeSession(this.session(input.sessionId), this.compactions, input)
cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) => {
cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = async (input) => {
const session = this.session(input.sessionId)
const acquisitionGeneration = session.acquisitionGeneration
return cancelClaudeTurn(session, this.deps.requestTimeoutMs, () => {
// Keep every ownership check adjacent to the provider interrupt. The
// session map check fences a replaced child; the turn check fences a
// delayed cancel after a newer turn was admitted on the same child.
return (
this.sessions.get(input.sessionId) === session &&
session.fence === input.fence &&
session.acquisitionGeneration === acquisitionGeneration &&
(this.compactions.ownsTurn(input.sessionId, input.turnId) ||
(session.activeTurnId === undefined
? session.dispatchSequence === 0
: session.activeTurnId === input.turnId &&
session.activeTurnSequence === session.dispatchSequence))
)
return cancelClaudeStructuredTurn({
session,
request: input,
requestTimeoutMs: this.deps.requestTimeoutMs,
stillOwnsTurn: () => {
// Keep every ownership check adjacent to the provider interrupt. The
// session map check fences a replaced child; the turn check fences a
// delayed cancel after a newer turn was admitted on the same child.
return (
this.sessions.get(input.sessionId) === session &&
session.fence === input.fence &&
session.acquisitionGeneration === acquisitionGeneration &&
(this.compactions.ownsTurn(input.sessionId, input.turnId) ||
(session.activeTurnId === undefined
? session.dispatchSequence === 0
: session.activeTurnId === input.turnId &&
session.activeTurnSequence === session.dispatchSequence))
)
},
emitPromptCancelled: (promptKey) => {
this.emit(session, { type: 'prompt-cancelled', sessionId: input.sessionId, promptKey })
}
})
}
promptCancellation: NonNullable<StructuredAgentSessionAdapter['promptCancellation']> = (input) =>
this.session(input.sessionId).prompts.cancellation(input.itemId)
stopBackgroundTasks: StructuredAgentSessionAdapter['stopBackgroundTasks'] = (input) => {
const session = this.session(input.sessionId)
const acquisitionGeneration = session.acquisitionGeneration
@@ -0,0 +1,37 @@
import type { StructuredAgentSessionAdapter } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
import { cancelClaudeTurn } from './claude-structured-control-actions'
import type { ClaudeSession } from './claude-structured-session-state'
type CancelTurnInput = Parameters<StructuredAgentSessionAdapter['cancelTurn']>[0]
export async function cancelClaudeStructuredTurn(input: {
session: ClaudeSession
request: CancelTurnInput
requestTimeoutMs?: number
stillOwnsTurn: () => boolean
emitPromptCancelled: (promptKey: string) => void
}): Promise<{ cancelled: boolean }> {
const { session, request } = input
const prompt = request.promptItemId
? session.prompts.beginHostCancellation(request.promptItemId, request.turnId)
: null
if (request.promptItemId && !prompt) {
return { cancelled: false }
}
let cancelled = false
try {
const result = await cancelClaudeTurn(session, input.requestTimeoutMs, () => {
return (
input.stillOwnsTurn() &&
(!request.promptItemId ||
session.prompts.cancellation(request.promptItemId)?.turnId === request.turnId)
)
})
cancelled = result.cancelled
return result
} finally {
if (prompt && session.prompts.finishHostCancellation(prompt, cancelled)) {
input.emitPromptCancelled(prompt.promptKey)
}
}
}
@@ -116,7 +116,8 @@ describe('persistent command retention', () => {
turnLifecycle: null,
sink,
streams: items.streams,
activeItems: items.activeItems
activeItems: items.activeItems,
pendingPrompts: new Map()
})
).toEqual({ accepted: true })
}
@@ -211,7 +212,8 @@ describe('persistent command retention', () => {
turnLifecycle: null,
sink,
streams: items.streams,
activeItems: items.activeItems
activeItems: items.activeItems,
pendingPrompts: new Map()
})
).toEqual({ accepted: true })
expect(items.activeItems.size).toBe(1)
@@ -8,7 +8,12 @@ export type CodexJournalTranslatorDeps = {
/** Keys restored lifecycle rows to the live identity; without it history restore skips them. */
sessionId?: string
now?: () => number
bindPromptItemId?: (journalItemId: string, threadId: string, promptKey: string) => void
bindPromptItemId?: (
journalItemId: string,
threadId: string,
promptKey: string,
turnId: string | null
) => void
primaryThreadId?: () => string | null
subagentExecutions?: CodexSubagentExecutions
coalesceMs?: number
@@ -18,13 +18,15 @@ import {
publishCodexLifecycle
} from './codex-structured-journal-sink'
import type { CodexPendingJournalPrompt } from './codex-structured-journal-settlement'
import { readCodexTurnId } from './codex-structured-thread-facts'
export class CodexJournalPrompts {
readonly pending = new Map<string, CodexPendingJournalPrompt>()
constructor(
private readonly deps: Pick<CodexJournalTranslatorDeps, 'sink' | 'bindPromptItemId'>,
private readonly detailFor: (threadId: string, itemId: string) => string | null
private readonly detailFor: (threadId: string, itemId: string) => string | null,
private readonly activeTurn: (threadId: string) => string | null
) {}
handle(event: {
@@ -47,12 +49,22 @@ export class CodexJournalPrompts {
}
for (const question of promptItems) {
const itemId = agentJournalItemKey(question.identity)
this.pending.set(itemId, { identity: question.identity, body: question.body })
this.pending.set(itemId, {
threadId: event.threadId,
turnId: readCodexTurnId(event.params) ?? this.activeTurn(event.threadId),
identity: question.identity,
body: question.body
})
const trimAdmission = this.trim()
if (!trimAdmission.accepted) {
return trimAdmission
}
this.deps.bindPromptItemId?.(itemId, event.threadId, event.promptKey)
this.deps.bindPromptItemId?.(
itemId,
event.threadId,
event.promptKey,
this.pending.get(itemId)?.turnId ?? null
)
}
return CODEX_JOURNAL_ADMITTED
}
@@ -70,12 +82,22 @@ export class CodexJournalPrompts {
return admission
}
const itemId = agentJournalItemKey(identity)
this.pending.set(itemId, { identity, body })
this.pending.set(itemId, {
threadId: event.threadId,
turnId: readCodexTurnId(event.params) ?? this.activeTurn(event.threadId),
identity,
body
})
const trimAdmission = this.trim()
if (!trimAdmission.accepted) {
return trimAdmission
}
this.deps.bindPromptItemId?.(itemId, event.threadId, event.promptKey)
this.deps.bindPromptItemId?.(
itemId,
event.threadId,
event.promptKey,
this.pending.get(itemId)?.turnId ?? null
)
return CODEX_JOURNAL_ADMITTED
}
@@ -89,7 +111,7 @@ export class CodexJournalPrompts {
private admit(
event: { method: string; threadId: string; promptKey: string },
items: readonly CodexPendingJournalPrompt[]
items: readonly Pick<CodexPendingJournalPrompt, 'identity' | 'body'>[]
): CodexJournalTranslationAdmission {
return admitCodexLifecycleItems(
this.deps.sink,
@@ -26,6 +26,10 @@ import {
codexTurnLifecycleBody,
codexTurnLifecycleIdentity
} from './codex-structured-journal-translation-turns'
import { collectCodexTurnPromptCancellations } from './codex-structured-prompt-turn-settlement'
import type { CodexPendingJournalPrompt } from './codex-structured-prompt-turn-settlement'
export type { CodexPendingJournalPrompt } from './codex-structured-prompt-turn-settlement'
export type CodexActiveJournalItem = {
threadId: string
@@ -34,11 +38,6 @@ export type CodexActiveJournalItem = {
item: CodexThreadItem
}
export type CodexPendingJournalPrompt = {
identity: AgentJournalItemIdentity
body: AgentJournalItemBody
}
const ADMITTED: StructuredAgentSessionSinkAdmission = { accepted: true }
export function settleCodexJournalSession(input: {
@@ -50,7 +49,6 @@ export function settleCodexJournalSession(input: {
currentTurnIds: ReadonlyMap<string, ReadonlySet<string>>
primaryThreadId: string | null
ordinals: CodexTurnOrdinals
/** Terminal lifecycle for a turn the provider left running when it ended. */
settledTurnLifecycle: (threadId: string, turnId: string) => AgentJournalTurnLifecycle
}): StructuredAgentSessionSinkAdmission {
const mutations: JournalLifecycleMutationInput[] = []
@@ -112,11 +110,11 @@ export function settleCodexJournalTurn(input: {
sessionId: string
threadId: string
turnId: string
/** Null off the primary thread: only the primary turn owns a lifecycle row. */
turnLifecycle: AgentJournalTurnLifecycle | null
sink: StructuredAgentSessionEventSink
streams: CodexStructuredItemStreams
activeItems: Map<string, CodexActiveJournalItem>
pendingPrompts: Map<string, CodexPendingJournalPrompt>
}): StructuredAgentSessionSinkAdmission {
const mutations: JournalLifecycleMutationInput[] = []
const activeItemsToForget: { key: string; threadId: string; itemId: string }[] = []
@@ -137,7 +135,8 @@ export function settleCodexJournalTurn(input: {
}
activeItemsToForget.push({ key, threadId: active.threadId, itemId: active.item.id })
}
// Revised, never tombstoned: the terminal row keeps the turn's duration durable.
const promptCancellations = collectCodexTurnPromptCancellations(input)
mutations.push(...promptCancellations.mutations)
if (input.turnLifecycle) {
mutations.push({
kind: 'item',
@@ -160,6 +159,9 @@ export function settleCodexJournalTurn(input: {
input.streams.forget(active.threadId, active.itemId)
input.activeItems.delete(active.key)
}
for (const itemId of promptCancellations.itemIds) {
input.pendingPrompts.delete(itemId)
}
return ADMITTED
}
@@ -57,7 +57,7 @@ export function publishCodexLifecycle(
export function admitCodexLifecycleItems(
sink: StructuredAgentSessionEventSink,
settlementId: string,
items: readonly CodexPendingJournalPrompt[]
items: readonly Pick<CodexPendingJournalPrompt, 'identity' | 'body'>[]
): CodexJournalTranslationAdmission {
if (items.length === 0) {
return { accepted: false, reason: 'untranslated' }
@@ -566,6 +566,36 @@ describe('codex journal translation', () => {
])
})
it('cancels only prompts owned by the completed turn', () => {
const { translator, tap } = translatorWith()
const ask = (turnId: string, promptKey: string): void => {
translator.handle({
type: 'prompt',
sessionId: SESSION_ID,
threadId: THREAD_ID,
method: CODEX_COMMAND_APPROVAL_METHOD,
params: { turnId, availableDecisions: ['accept', 'decline'] },
codexItemId: `command-${turnId}`,
promptKey
})
}
translator.handle(TURN_STARTED)
ask(TURN_ID, 'approval-turn-1')
translator.handle(notification('turn/started', { turn: { id: 'turn-2' } }))
ask('turn-2', 'approval-turn-2')
translator.handle(notification('turn/completed', { turn: { id: TURN_ID } }))
const promptRows = tap.rows.filter((row) => row.body.kind === 'approval')
const rowsFor = (key: string) => promptRows.filter((row) => row.key.includes(key))
const completed = rowsFor('approval-turn-1')
const active = rowsFor('approval-turn-2')
expect(completed).toHaveLength(2)
expect(completed.at(-1)?.body).toMatchObject({ resolution: { state: 'cancelled' } })
expect(active).toHaveLength(1)
expect(active[0]?.body).toMatchObject({ resolution: { state: 'pending' } })
})
it('journals an approval naming the command the item already announced, and binds it', () => {
const { translator, tap } = translatorWith()
@@ -5,6 +5,7 @@ import {
type CodexJournalTranslationAdmission
} from './codex-structured-journal-contracts'
import type { CodexJournalItems } from './codex-structured-journal-items'
import type { CodexJournalPrompts } from './codex-structured-journal-prompts'
import { settleCodexJournalTurn } from './codex-structured-journal-settlement'
import type { CodexJournalActiveTurns } from './codex-structured-journal-translation-turn-state'
import {
@@ -33,6 +34,7 @@ export class CodexJournalTurnBoundaries {
primaryThreadId: () => string | null
activeTurns: CodexJournalActiveTurns
items: Pick<CodexJournalItems, 'streams' | 'activeItems' | 'ordinals'>
prompts: Pick<CodexJournalPrompts, 'pending'>
flushSuppression: () => CodexJournalTranslationAdmission
resetActivity: (threadId: string) => void
now?: () => number
@@ -93,7 +95,8 @@ export class CodexJournalTurnBoundaries {
)
: null,
streams: this.deps.items.streams,
activeItems: this.deps.items.activeItems
activeItems: this.deps.items.activeItems,
pendingPrompts: this.deps.prompts.pending
})
if (admission.accepted) {
this.deps.items.ordinals.forgetTurn(event.threadId, turnId)
@@ -59,8 +59,10 @@ export function createCodexJournalTranslator(
(threadId, turnId) => genericFrames.suppress(threadId, turnId)
)
const settleOversizedNotification = createCodexOversizedNotificationSettler(deps, items)
const prompts = new CodexJournalPrompts(deps, (threadId, itemId) =>
items.detailFor(threadId, itemId)
const prompts = new CodexJournalPrompts(
deps,
(threadId, itemId) => items.detailFor(threadId, itemId),
(threadId) => activeTurns.current(threadId)
)
const subagents = new CodexSubagentRoster({
sink: deps.sink,
@@ -82,6 +84,7 @@ export function createCodexJournalTranslator(
primaryThreadId: () => deps.primaryThreadId?.() ?? null,
activeTurns,
items,
prompts,
flushSuppression: () => genericFrames.flush(),
resetActivity,
...(deps.now ? { now: deps.now } : {})
@@ -107,6 +107,39 @@ describe('CodexPromptRegistry', () => {
expect(registry.find('codex-item-1')).toBeNull()
})
it('returns every journal row for one prompt only when it belongs to the cancelled turn', () => {
const registry = new CodexPromptRegistry()
registry.register(userInputRequest(['q1', 'q2']))
registry.bindJournalItemId('journal-q1', 'thread-1', 'codex-item-1')
registry.bindJournalItemId('journal-q2', 'thread-1', 'codex-item-1')
expect(registry.cancellation('journal-q1')).toEqual({
turnId: 'turn-1',
itemIds: ['journal-q1', 'journal-q2']
})
expect(registry.cancellation('unbound')).toBeNull()
})
it('reads nested turn identity and adopts the translator fallback when absent', () => {
const nested = new CodexPromptRegistry()
nested.register({
id: 6,
method: 'item/commandExecution/requestApproval',
params: { itemId: 'item-nested', threadId: 'thread-1', turn: { id: 'turn-nested' } }
})
nested.bindJournalItemId('journal-nested', 'thread-1', 'item-nested')
expect(nested.cancellation('journal-nested')?.turnId).toBe('turn-nested')
const fallback = new CodexPromptRegistry()
fallback.register({
id: 7,
method: 'item/commandExecution/requestApproval',
params: { itemId: 'item-fallback', threadId: 'thread-1' }
})
fallback.bindJournalItemId('journal-fallback', 'thread-1', 'item-fallback', 'turn-active')
expect(fallback.cancellation('journal-fallback')?.turnId).toBe('turn-active')
})
it('keeps identical item ids on different threads independently answerable', () => {
const registry = new CodexPromptRegistry()
const register = (id: number, threadId: string) =>
@@ -1,4 +1,5 @@
import type { CodexAppServerConnection } from './codex-app-server-connection'
import { readCodexTurnId } from './codex-structured-thread-facts'
import {
CODEX_PROMPT_MAX_ANSWER_BYTES,
MAX_CODEX_PROMPT_JOURNAL_BINDINGS,
@@ -186,7 +187,7 @@ export class CodexPromptRegistry {
requestId: request.id,
method: request.method,
threadId,
turnId: readString(request.params, 'turnId'),
turnId: readCodexTurnId(request.params),
codexItemId,
promptKey: readString(request.params, 'approvalId') ?? codexItemId,
questionIds,
@@ -222,7 +223,12 @@ export class CodexPromptRegistry {
}
/** Called by the translation module once the prompt has a journal id. */
bindJournalItemId(journalItemId: string, threadId: string, promptKey: string): void {
bindJournalItemId(
journalItemId: string,
threadId: string,
promptKey: string,
turnId?: string | null
): void {
const existing = this.journalItemIds.get(journalItemId)
if (existing) {
this.boundPrompts.delete(journalItemId)
@@ -233,6 +239,9 @@ export class CodexPromptRegistry {
if (!prompt) {
return
}
if (prompt.turnId === null && turnId && prompt.threadId === threadId) {
prompt.turnId = turnId
}
this.journalItemIds.set(journalItemId, address)
this.boundPrompts.set(journalItemId, prompt)
this.trim()
@@ -251,6 +260,19 @@ export class CodexPromptRegistry {
return matches.length === 1 ? matches[0]! : null
}
cancellation(journalItemId: string): { turnId: string; itemIds: readonly string[] } | null {
const prompt = this.find(journalItemId)
if (!prompt?.turnId || this.boundPrompts.get(journalItemId) !== prompt) {
return null
}
return {
turnId: prompt.turnId,
itemIds: [...this.boundPrompts]
.filter(([, candidate]) => candidate === prompt)
.map(([itemId]) => itemId)
}
}
forget(prompt: CodexPendingPrompt): void {
const address = this.address(prompt.threadId, prompt.promptKey)
if (this.byAddress.get(address) === prompt) {
@@ -0,0 +1,33 @@
import type {
AgentJournalItemBody,
AgentJournalItemIdentity
} from '../../shared/agent-session-journal-types'
import { cancelledJournalPromptBody } from '../native-chat/agent-session-journal/journal-prompt-body-bounds'
import type { JournalLifecycleMutationInput } from '../native-chat/agent-session-journal/journal-row-builders'
export type CodexPendingJournalPrompt = {
threadId: string
turnId: string | null
identity: AgentJournalItemIdentity
body: AgentJournalItemBody
}
export function collectCodexTurnPromptCancellations(input: {
threadId: string
turnId: string
pendingPrompts: ReadonlyMap<string, CodexPendingJournalPrompt>
}): { mutations: JournalLifecycleMutationInput[]; itemIds: string[] } {
const mutations: JournalLifecycleMutationInput[] = []
const itemIds: string[] = []
for (const [itemId, prompt] of input.pendingPrompts) {
if (prompt.threadId !== input.threadId || prompt.turnId !== input.turnId) {
continue
}
const body = cancelledJournalPromptBody(prompt.body)
if (body) {
mutations.push({ kind: 'item', identity: prompt.identity, body })
}
itemIds.push(itemId)
}
return { mutations, itemIds }
}
@@ -84,8 +84,8 @@ export async function acquireCodexStructuredSession(input: {
...(deps.now ? { now: deps.now } : {}),
primaryThreadId: () => primaryThreadId,
subagentExecutions,
bindPromptItemId: (journalItemId, threadId, promptKey) =>
acquisition.prompts.bindJournalItemId(journalItemId, threadId, promptKey)
bindPromptItemId: (journalItemId, threadId, promptKey, turnId) =>
acquisition.prompts.bindJournalItemId(journalItemId, threadId, promptKey, turnId)
})
: null
const open = deps.openConnection ?? openCodexAppServerConnection
@@ -204,12 +204,24 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
sessionId: string
turnId: string
fence: number
promptItemId?: string
}): Promise<{ cancelled: boolean }> {
const session = this.session(input.sessionId)
const turnId = this.compactions.providerTurnId(input.sessionId, input.turnId)
return turnId ? this.turnCancellation.cancel(session, turnId) : { cancelled: false }
const prompt = input.promptItemId ? session.prompts.find(input.promptItemId) : null
if (!turnId || (input.promptItemId && prompt?.turnId !== turnId)) {
return { cancelled: false }
}
const result = await this.turnCancellation.cancel(session, turnId)
if (result.cancelled && prompt) {
session.prompts.forget(prompt)
}
return result
}
promptCancellation: NonNullable<StructuredAgentSessionAdapter['promptCancellation']> = (input) =>
this.session(input.sessionId).prompts.cancellation(input.itemId)
rewindSupport: NonNullable<StructuredAgentSessionAdapter['rewindSupport']> = (sessionId) =>
this.sessions.get(sessionId)?.historyMode === 'legacy'
? { supported: false, reason: 'history-not-paginated' }
@@ -1,8 +1,10 @@
import { describe, expect, it, vi } from 'vitest'
import type {
AgentJournalItemBody,
AgentJournalMessageItem,
AgentSessionJournalIdentity
} from '../../shared/agent-session-journal-types'
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
import {
CodexAppServerRequestError,
type CodexAppServerConnection,
@@ -84,7 +86,8 @@ async function acquired(
CodexStructuredSessionAdapterDeps,
'captureTurnProcesses' | 'terminateTurnProcesses' | 'now'
>
> = {}
> = {},
eventSink?: StructuredAgentSessionEventSink
): Promise<CodexStructuredSessionAdapter> {
const adapter = new CodexStructuredSessionAdapter({
resolveLaunch: async () => ({
@@ -101,7 +104,12 @@ async function acquired(
terminateTurnProcesses: async () => true,
...processControl
})
await adapter.acquire({ identity: identity(), fence: 7, spawnToken: 'spawn-9' })
await adapter.acquire({
identity: identity(),
fence: 7,
spawnToken: 'spawn-9',
...(eventSink ? { events: eventSink } : {})
})
return adapter
}
@@ -113,6 +121,40 @@ function completeTurn(codex: ReturnType<typeof fakeCodex>, turnId = 'turn-1'): v
}
describe('CodexStructuredSessionAdapter.cancelTurn', () => {
it('binds prompt cancellation to the live logical request and owning turn', async () => {
const codex = fakeCodex()
const adapter = await acquired(codex)
codex.connections[0].handlers.onServerRequest?.({
id: 1,
method: 'item/tool/requestUserInput',
params: {
itemId: 'questions-1',
threadId: THREAD_ID,
turnId: 'turn-1',
questions: [{ id: 'q1' }, { id: 'q2' }]
}
})
adapter.bindPromptItemId('session-1', 'journal-q1', 'questions-1')
adapter.bindPromptItemId('session-1', 'journal-q2', 'questions-1')
expect(
adapter.promptCancellation?.({
sessionId: 'session-1',
itemId: 'journal-q1',
fence: 7
})
).toEqual({ turnId: 'turn-1', itemIds: ['journal-q1', 'journal-q2'] })
await expect(
adapter.cancelTurn({
sessionId: 'session-1',
turnId: 'turn-1',
fence: 7,
promptItemId: 'unrelated-prompt'
})
).resolves.toEqual({ cancelled: false })
expect(codex.connections[0].calls.some((call) => call.method === 'turn/interrupt')).toBe(false)
})
it('confirms an interrupt Codex acknowledged', async () => {
const codex = fakeCodex()
const adapter = await acquired(codex)
@@ -291,18 +333,50 @@ describe('CodexStructuredSessionAdapter.cancelTurn', () => {
it('does not strand a deferred completion when the interrupt receipt fails', async () => {
const events: CodexStructuredSessionEvent[] = []
const bodies: AgentJournalItemBody[] = []
const sink: StructuredAgentSessionEventSink = {
appendItem: (_identity, body) => bodies.push(body),
appendTombstone: vi.fn(),
publish: vi.fn()
}
const codex = fakeCodex()
codex.routes['turn/interrupt'] = () => {
completeTurn(codex)
throw new Error('interrupt receipt lost')
}
const adapter = await acquired(codex, events, {
terminateTurnProcesses: async () => true
const adapter = await acquired(
codex,
events,
{
terminateTurnProcesses: async () => true
},
sink
)
codex.connections[0].handlers.onNotification?.('turn/started', {
threadId: THREAD_ID,
turn: { id: 'turn-1' }
})
codex.connections[0].handlers.onServerRequest?.({
id: 1,
method: 'item/commandExecution/requestApproval',
params: {
itemId: 'command-1',
approvalId: 'approval-1',
threadId: THREAD_ID,
turnId: 'turn-1',
availableDecisions: ['accept', 'decline']
}
})
await expect(
adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 })
).rejects.toThrow('interrupt receipt lost')
expect(events).toContainEqual(expect.objectContaining({ method: 'turn/completed' }))
expect(bodies).toContainEqual(
expect.objectContaining({
kind: 'approval',
resolution: expect.objectContaining({ state: 'cancelled' })
})
)
})
})
@@ -0,0 +1,71 @@
import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
import type { JournalReducerState } from './journal-reducer'
import { journalLifecycleBatchRowBuilder } from './journal-row-builders'
import type { JournalRowWriter } from './journal-row-writer'
export type JournalPromptCancellationInput = {
prompts: readonly { itemId: string; expectedRevision: number }[]
settlementId: string
resolvedBy: string
resolvedAt: number
fence: number
}
/** Atomically revises only prompt revisions that still match the cancellation snapshot. */
export function cancelJournalPromptsAtRevisions(
rowWriter: JournalRowWriter,
state: () => JournalReducerState,
input: JournalPromptCancellationInput
): Promise<number> {
const identities = input.prompts.map((prompt) => parseAgentJournalItemKey(prompt.itemId))
if (
input.prompts.length === 0 ||
new Set(input.prompts.map((prompt) => prompt.itemId)).size !== input.prompts.length ||
identities.some((identity) => !identity)
) {
return Promise.resolve(0)
}
return rowWriter
.enqueueIf((sequence, timestamp) => {
const currentState = state()
const mutations = input.prompts.flatMap((prompt, index) => {
const resolved = currentState.aliases.get(prompt.itemId) ?? prompt.itemId
const current = currentState.items.get(resolved)
const identity = identities[index]
if (
!identity ||
!current ||
current.revision !== prompt.expectedRevision ||
(current.body.kind !== 'approval' && current.body.kind !== 'question') ||
current.body.resolution.state !== 'pending'
) {
return []
}
return [
{
kind: 'item' as const,
identity,
body: {
...current.body,
resolution: {
state: 'cancelled' as const,
selectedOptionId: null,
resolvedBy: input.resolvedBy,
resolvedAt: input.resolvedAt
}
}
}
]
})
return mutations.length === 0
? null
: journalLifecycleBatchRowBuilder(state, input.settlementId, mutations, {
fence: input.fence
})(sequence, timestamp)
})
.then((row) =>
row?.kind === 'lifecycle-batch'
? row.mutations.filter((mutation) => mutation.kind === 'item').length
: 0
)
}
@@ -18,9 +18,21 @@ export class JournalRowWriter {
constructor(private readonly deps: JournalRowWriterDeps) {}
enqueue(build: (seq: number, ts: number) => JournalRow): Promise<JournalRow> {
return this.enqueueIf(build).then((row) => {
if (!row) {
throw new Error('journal_row_builder_skipped_unconditionally')
}
return row
})
}
enqueueIf(build: (seq: number, ts: number) => JournalRow | null): Promise<JournalRow | null> {
return this.deps.serialize(async () => {
assertJournalWritable(this.deps.readOnly(), this.deps.sessionId)
const row = build(this.deps.nextSequence(), this.deps.now())
if (!row) {
return null
}
assertJournalFence(row.fence, this.deps.highestFence())
const { db } = this.deps.database()
db.exec('BEGIN IMMEDIATE')
@@ -6,6 +6,7 @@ import type {
AgentJournalCursor,
AgentJournalItemBody,
AgentJournalItemIdentity,
AgentJournalRenderItem,
AgentJournalSnapshot,
AgentJournalSubmission,
AgentSessionJournalIdentity
@@ -49,6 +50,10 @@ import { createJournalStoreCollaborators } from './journal-store-collaborators'
import { ensureJournalDir, journalStoreLoadedFields } from './journal-store-open'
import type { JournalItemAppender } from './journal-item-appender'
import type { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender'
import {
cancelJournalPromptsAtRevisions,
type JournalPromptCancellationInput
} from './journal-prompt-cancellation'
export { AgentSessionJournalError } from './journal-write-guards'
@@ -162,6 +167,30 @@ export class AgentSessionJournal {
snapshot = (): AgentJournalSnapshot => renderJournalState(this.state)
/** Orders a CAS preflight behind provider writes already admitted to the journal queue. */
readItem(itemId: string): Promise<AgentJournalRenderItem | null> {
return this.queue.serialize(async () => {
const resolved = this.state.aliases.get(itemId) ?? itemId
return this.state.items.get(resolved) ?? null
})
}
/** Orders a bounded prompt-group snapshot behind admitted provider writes. */
readItems(itemIds: readonly string[]): Promise<AgentJournalRenderItem[] | null> {
return this.queue.serialize(async () => {
const items: AgentJournalRenderItem[] = []
for (const itemId of itemIds) {
const resolved = this.state.aliases.get(itemId) ?? itemId
const item = this.state.items.get(resolved)
if (!item) {
return null
}
items.push(item)
}
return items
})
}
/** Visits reduced items without allocating and sorting a full snapshot. */
visitItems = (visit: (itemId: string, sequence: number) => void): void => {
for (const item of this.state.items.values()) {
@@ -213,6 +242,10 @@ export class AgentSessionJournal {
return this.itemAppender.append(identity, body, options)
}
cancelPromptsAtRevisions(input: JournalPromptCancellationInput): Promise<number> {
return cancelJournalPromptsAtRevisions(this.rowWriter, () => this.state, input)
}
appendTombstone(
identity: AgentJournalItemIdentity,
options: JournalTombstoneInput
@@ -32,6 +32,33 @@ describe('StructuredAgentSessionAdapterRouter.releaseAcquisition', () => {
})
})
describe('StructuredAgentSessionAdapterRouter prompt cancellation', () => {
it('reads the logical prompt group from the session owner', async () => {
const claude = adapterOf(vi.fn(async () => true))
const promptCancellation = vi.fn(() => ({
turnId: 'turn-1',
itemIds: ['prompt-1', 'prompt-2']
}))
claude.promptCancellation = promptCancellation
const router = new StructuredAgentSessionAdapterRouter(
{ claude, codex: adapterOf(vi.fn(async () => false)) },
async () => {}
)
await router.acquire({
identity: { sessionId: 'session-1', agent: 'claude' } as never,
fence: 1,
spawnToken: 'spawn-1'
})
const input = { sessionId: 'session-1', turnId: 'turn-1', itemId: 'prompt-1', fence: 1 }
expect(router.promptCancellation(input)).toEqual({
turnId: 'turn-1',
itemIds: ['prompt-1', 'prompt-2']
})
expect(promptCancellation).toHaveBeenCalledExactlyOnceWith(input)
})
})
describe('StructuredAgentSessionAdapterRouter.closeSession', () => {
it('retains the owner after an unproven close so a later retry reaches the same adapter', async () => {
const claude = adapterOf(vi.fn(async () => true))
@@ -71,6 +71,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi
cancelTurn: StructuredAgentSessionAdapter['cancelTurn'] = (input) =>
this.owner(input.sessionId).cancelTurn(input)
promptCancellation: NonNullable<StructuredAgentSessionAdapter['promptCancellation']> = (input) =>
this.owner(input.sessionId).promptCancellation?.(input) ?? null
stopBackgroundTasks: NonNullable<StructuredAgentSessionAdapter['stopBackgroundTasks']> = (
input
) => {
@@ -188,10 +188,16 @@ export type StructuredAgentSessionAdapter = {
}): Promise<{ error?: string }>
/** Cancels one turn, not the session: a session-wide interrupt would also kill
* a turn the client never asked to stop. */
promptCancellation?(input: {
sessionId: string
itemId: string
fence: number
}): { turnId: string; itemIds: readonly string[] } | null
cancelTurn(input: {
sessionId: string
turnId: string
fence: number
promptItemId?: string
}): Promise<{ cancelled: boolean }>
stopBackgroundTasks?(input: {
sessionId: string
@@ -106,9 +106,10 @@ export function cancelStructuredAgentSessionTurn(
caller: StructuredAgentSessionCaller,
params: {
envelope: AgentSessionMutationEnvelope
turnId: string
turnId?: string
scope?: 'background-tasks'
taskId?: string
prompt?: { itemId: string; expectedRevision: number }
}
): Promise<AgentSessionMutationResult<AgentSessionCancelResult>> {
const command = context.deps.store.getRecord(params.envelope.sessionId)?.conversationCommand
@@ -7,6 +7,7 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire'
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { journalDirectoryFor } from '../agent-session-journal/journal-paths'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open'
import type {
AgentSessionDispatchOutcome,
@@ -58,6 +59,7 @@ let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
let releaseAcquisition: Mock<NonNullable<StructuredAgentSessionAdapter['releaseAcquisition']>>
let dispatch: Mock<StructuredAgentSessionAdapter['dispatch']>
let cancelTurn: Mock<StructuredAgentSessionAdapter['cancelTurn']>
let promptCancellation: Mock<NonNullable<StructuredAgentSessionAdapter['promptCancellation']>>
let answerPrompt: Mock<StructuredAgentSessionAdapter['answerPrompt']>
let setOption: Mock<StructuredAgentSessionAdapter['setOption']>
let ordinal = 0
@@ -75,6 +77,7 @@ function adapter(): StructuredAgentSessionAdapter {
acquire,
releaseAcquisition,
dispatch,
promptCancellation,
cancelTurn,
answerPrompt,
setOption
@@ -87,12 +90,15 @@ async function attach(): Promise<AgentSessionRecord | null> {
return store.getRecord(SESSION)
}
/** Puts a pending approval in the journal BEFORE attach, which is the only way
* 1d can stage one: the adapter that would emit it is phase 2's. */
async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revision: number }> {
const identity = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 99 }
async function promptJournal(): Promise<AgentSessionJournal> {
const live = (
host as unknown as { sessions: Map<string, { journal: AgentSessionJournal }> }
).sessions.get(SESSION)?.journal
if (live) {
return live
}
const journalDir = journalDirectoryFor(root, { workspaceId: 'workspace-1', sessionId: SESSION })
const journal = await journals.open({
return journals.open({
identity: {
sessionId: SESSION,
workspaceId: 'workspace-1',
@@ -102,6 +108,14 @@ async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revis
},
journalDir
})
}
async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revision: number }> {
const identity = {
provider: 'orca' as const,
clientMessageId: `codex-prompt:${THREAD}:approval-1`
}
const journal = await promptJournal()
const appended = await journal.appendItem(
identity,
{
@@ -111,11 +125,33 @@ async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revis
options: [{ id: optionId, label: 'Allow' }],
resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
},
{ fence: 1 }
{ fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1 }
)
return { itemId: appended.itemId, revision: appended.revision }
}
async function seedQuestionGroup(): Promise<{ itemId: string; revision: number }[]> {
const journal = await promptJournal()
const prompts: { itemId: string; revision: number }[] = []
for (const questionId of ['q1', 'q2']) {
const appended = await journal.appendItem(
{
provider: 'orca',
clientMessageId: `codex-prompt:${THREAD}:questions-1:${questionId}`
},
{
kind: 'question',
question: `Question ${questionId}`,
options: [{ id: `${questionId}:yes`, label: 'Yes' }],
resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
},
{ fence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1 }
)
prompts.push({ itemId: appended.itemId, revision: appended.revision })
}
return prompts
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-wire-host-'))
resetHostTestOperationIds()
@@ -138,6 +174,7 @@ beforeEach(async () => {
releaseAcquisition = vi.fn(async () => true)
dispatch = vi.fn(async () => accepted())
cancelTurn = vi.fn(async () => ({ cancelled: true }))
promptCancellation = vi.fn(({ itemId }) => ({ turnId: 'turn-1', itemIds: [itemId] }))
answerPrompt = vi.fn(async () => undefined)
setOption = vi.fn(async () => undefined)
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
@@ -178,6 +215,7 @@ export function hostTestState() {
acquire,
releaseAcquisition,
dispatch,
promptCancellation,
cancelTurn,
answerPrompt,
setOption
@@ -193,5 +231,6 @@ export {
ensureParams,
envelope,
journals,
seedApproval
seedApproval,
seedQuestionGroup
}
@@ -17,7 +17,8 @@ import {
envelope,
hostTestState,
replaceHostTestState,
seedApproval
seedApproval,
seedQuestionGroup
} from './structured-agent-session-host-test-harness'
import {
HOST_TEST_NOW as NOW,
@@ -33,6 +34,7 @@ let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
let releaseAcquisition: Mock<NonNullable<StructuredAgentSessionAdapter['releaseAcquisition']>>
let dispatch: Mock<StructuredAgentSessionAdapter['dispatch']>
let cancelTurn: Mock<StructuredAgentSessionAdapter['cancelTurn']>
let promptCancellation: Mock<NonNullable<StructuredAgentSessionAdapter['promptCancellation']>>
let answerPrompt: Mock<StructuredAgentSessionAdapter['answerPrompt']>
let setOption: Mock<StructuredAgentSessionAdapter['setOption']>
@@ -44,6 +46,7 @@ beforeEach(() => {
acquire,
releaseAcquisition,
dispatch,
promptCancellation,
cancelTurn,
answerPrompt,
setOption
@@ -230,16 +233,237 @@ describe('cancel', () => {
expect(await host.cancel(CALLER, params)).toMatchObject({
ok: true,
replayed: true,
value: { cancelled: false }
value: { cancelled: true }
})
expect(cancelTurn).toHaveBeenCalledTimes(1)
})
it('recovers a confirmed prompt interruption when later status persistence fails', async () => {
await attach()
const prompt = await seedApproval()
const journal = (
host as unknown as { sessions: Map<string, { journal: AgentSessionJournal }> }
).sessions.get(SESSION)!.journal
const appendItem = journal.appendItem.bind(journal)
const statusFailure = vi
.spyOn(journal, 'appendItem')
.mockImplementation((identity, body, options) =>
body.kind === 'status'
? Promise.reject(new Error('status persistence failed'))
: appendItem(identity, body, options)
)
const fields = {
prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision }
}
const params = {
envelope: envelope('agentSession.cancel', fields),
...fields
}
await expect(host.cancel(CALLER, params)).rejects.toThrow('status persistence failed')
statusFailure.mockRestore()
await expect(host.cancel(CALLER, params)).resolves.toMatchObject({
ok: true,
replayed: true,
value: { cancelled: true }
})
expect(cancelTurn).toHaveBeenCalledTimes(1)
})
it('terminalizes the displayed prompt only after the matching turn interruption succeeds', async () => {
await attach()
const prompt = await seedApproval()
const fields = {
turnId: 'turn-1',
prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision }
}
expect(
await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', fields),
...fields
})
).toMatchObject({ ok: true, value: { cancelled: true } })
const page = host.history({ sessionId: SESSION, direction: 'tail' })
const items = page.ok ? page.page.items : []
expect(items.find((item) => item.itemId === prompt.itemId)).toMatchObject({
revision: prompt.revision + 1,
body: {
kind: 'approval',
resolution: {
state: 'cancelled',
selectedOptionId: null,
resolvedBy: CALLER.callerKey,
resolvedAt: NOW
}
}
})
expect(
items.filter(
(item) =>
(item.body.kind === 'approval' || item.body.kind === 'question') &&
item.body.resolution.state === 'pending'
)
).toEqual([])
})
it('terminalizes every pending row for the cancelled logical question request', async () => {
await attach()
const prompts = await seedQuestionGroup()
promptCancellation.mockReturnValue({
turnId: 'provider-child-turn',
itemIds: prompts.map((prompt) => prompt.itemId)
})
const displayed = prompts[0]!
const fields = {
prompt: { itemId: displayed.itemId, expectedRevision: displayed.revision }
}
expect(
await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', fields),
...fields
})
).toMatchObject({ ok: true, value: { cancelled: true } })
expect(cancelTurn).toHaveBeenCalledWith(
expect.objectContaining({
turnId: 'provider-child-turn',
promptItemId: displayed.itemId
})
)
const page = host.history({ sessionId: SESSION, direction: 'tail' })
const group = page.ok
? page.page.items.filter((item) => prompts.some((prompt) => prompt.itemId === item.itemId))
: []
expect(group).toHaveLength(2)
expect(group.every((item) => item.revision === 2)).toBe(true)
expect(
group.every(
(item) =>
(item.body.kind === 'question' || item.body.kind === 'approval') &&
item.body.resolution.state === 'cancelled'
)
).toBe(true)
})
it('does not interrupt or settle when the displayed prompt revision is stale', async () => {
await attach()
const prompt = await seedApproval()
const fields = {
turnId: 'turn-1',
prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision + 1 }
}
expect(
await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', fields),
...fields
})
).toMatchObject({
ok: false,
refusal: { code: 'agent_session_item_revision_stale', currentRevision: prompt.revision }
})
expect(cancelTurn).not.toHaveBeenCalled()
const page = host.history({ sessionId: SESSION, direction: 'tail' })
expect(page.ok && page.page.items.find((item) => item.itemId === prompt.itemId)).toMatchObject({
revision: prompt.revision,
body: { resolution: { state: 'pending' } }
})
})
it('does not let a restored prompt interrupt an unrelated live turn', async () => {
const prompt = await seedApproval()
promptCancellation.mockReturnValue(null)
await attach()
const fields = {
turnId: 'turn-1',
prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision }
}
expect(
await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', fields),
...fields
})
).toMatchObject({ ok: false, refusal: { code: 'agent_session_item_revision_stale' } })
expect(cancelTurn).not.toHaveBeenCalled()
})
it('leaves the displayed prompt pending when the provider refuses cancellation', async () => {
await attach()
const prompt = await seedApproval()
cancelTurn.mockResolvedValueOnce({ cancelled: false })
const fields = {
turnId: 'turn-1',
prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision }
}
expect(
await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', fields),
...fields
})
).toMatchObject({ ok: true, value: { cancelled: false } })
const page = host.history({ sessionId: SESSION, direction: 'tail' })
expect(page.ok && page.page.items.find((item) => item.itemId === prompt.itemId)).toMatchObject({
revision: prompt.revision,
body: { resolution: { state: 'pending' } }
})
})
it('preserves a provider-settled prompt that races the confirmed interruption', async () => {
await attach()
const prompt = await seedApproval()
const journal = (
host as unknown as { sessions: Map<string, { journal: AgentSessionJournal }> }
).sessions.get(SESSION)!.journal
cancelTurn.mockImplementationOnce(async () => {
await journal.appendItem(
{ provider: 'orca', clientMessageId: `codex-prompt:${THREAD}:approval-1` },
{
kind: 'approval',
title: 'Run the command?',
detail: null,
options: [{ id: 'allow', label: 'Allow' }],
resolution: {
state: 'resolved',
selectedOptionId: 'allow',
resolvedBy: 'provider-race',
resolvedAt: NOW - 1
}
},
{ fence: 1 }
)
return { cancelled: true }
})
const fields = {
turnId: 'turn-1',
prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision }
}
expect(
await host.cancel(CALLER, {
envelope: envelope('agentSession.cancel', fields),
...fields
})
).toMatchObject({ ok: true, value: { cancelled: true } })
const page = host.history({ sessionId: SESSION, direction: 'tail' })
expect(page.ok && page.page.items.find((item) => item.itemId === prompt.itemId)).toMatchObject({
revision: prompt.revision + 1,
body: {
resolution: { state: 'resolved', selectedOptionId: 'allow', resolvedBy: 'provider-race' }
}
})
})
})
describe('respondToPrompt', () => {
it('commits the answer before the provider callback', async () => {
const prompt = await seedApproval()
await attach()
const prompt = await seedApproval()
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' }
const result = await host.respondToPrompt(CALLER, {
envelope: envelope('agentSession.respondTo:approval', fields),
@@ -254,8 +478,8 @@ describe('respondToPrompt', () => {
})
it('refuses a second answer to one prompt and says which answer won', async () => {
const prompt = await seedApproval()
await attach()
const prompt = await seedApproval()
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' }
await host.respondToPrompt(CALLER, {
envelope: envelope('agentSession.respondTo:approval', fields),
@@ -281,8 +505,8 @@ describe('respondToPrompt', () => {
})
it('refuses an option the prompt does not offer', async () => {
const prompt = await seedApproval()
await attach()
const prompt = await seedApproval()
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'deny' }
expect(
await host.respondToPrompt(CALLER, {
@@ -295,8 +519,8 @@ describe('respondToPrompt', () => {
})
it("does not turn a recorded refusal into another client's successful answer", async () => {
const prompt = await seedApproval()
await attach()
const prompt = await seedApproval()
const rejectedFields = {
itemId: prompt.itemId,
expectedRevision: prompt.revision,
@@ -326,8 +550,8 @@ describe('respondToPrompt', () => {
})
it('keeps the answer and reports it undelivered when the provider callback throws', async () => {
const prompt = await seedApproval()
await attach()
const prompt = await seedApproval()
answerPrompt.mockRejectedValueOnce(new Error('pipe closed'))
const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' }
const result = await host.respondToPrompt(CALLER, {
@@ -76,27 +76,53 @@ export function sendPlan(params: {
export function cancelPlan(params: {
envelope: AgentSessionMutationEnvelope
turnId: string
turnId?: string
scope?: 'background-tasks'
taskId?: string
prompt?: { itemId: string; expectedRevision: number }
}): MutationPlan<AgentSessionCancelResult> {
return {
method: 'agentSession.cancel',
fields: {
turnId: params.turnId,
...(params.turnId ? { turnId: params.turnId } : {}),
...(params.scope ? { scope: params.scope } : {}),
...(params.taskId ? { taskId: params.taskId } : {})
...(params.taskId ? { taskId: params.taskId } : {}),
...(params.prompt ? { prompt: params.prompt } : {})
},
run: (ctx) =>
performCancel(ctx, {
clientOperationId: params.envelope.clientOperationId,
turnId: params.turnId,
...(params.scope ? { scope: params.scope } : {}),
...(params.taskId ? { taskId: params.taskId } : {})
...(params.taskId ? { taskId: params.taskId } : {}),
...(params.prompt ? { prompt: params.prompt } : {})
}),
// Interrupting twice would kill a turn the client never asked to stop, so a
// replay reports the turn as already handled instead.
replay: () => ({ turnId: params.turnId, cancelled: false })
settledOutcome: (value) => ({
status: 'succeeded',
sessionId: params.envelope.sessionId,
cancelled: value.cancelled,
cancelledTurnId: value.turnId
}),
recoverUnknownFromDurableState: params.prompt !== undefined,
// Interrupting twice would kill a turn the client never asked to stop.
replay: (ctx, outcome) => {
if (outcome.status === 'unknown' && params.prompt) {
const item = ctx.journal
.snapshot()
.items.find((candidate) => candidate.itemId === params.prompt?.itemId)
const cancelled =
(item?.body.kind === 'approval' || item?.body.kind === 'question') &&
item.body.resolution.state === 'cancelled'
return cancelled ? { turnId: params.turnId ?? params.prompt.itemId, cancelled: true } : null
}
return {
turnId:
outcome.status === 'succeeded' && outcome.cancelledTurnId
? outcome.cancelledTurnId
: (params.turnId ?? params.prompt?.itemId ?? 'unknown'),
cancelled: outcome.status === 'succeeded' && outcome.cancelled === true
}
}
}
}
@@ -0,0 +1,111 @@
import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
import type {
AgentJournalRenderItem,
AgentJournalResolution
} from '../../../shared/agent-session-journal-types'
import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
type PromptTarget = { itemId: string; expectedRevision: number }
type PromptCancellationRefusal = Extract<TurnOutcome<never>, { ok: false }>
function invalid(message: string): PromptCancellationRefusal {
return { ok: false, refusal: { code: 'agent_session_operation_invalid', message } }
}
export async function preparePromptCancellation(
ctx: AgentSessionTurnContext,
target: PromptTarget
): Promise<
| PromptCancellationRefusal
| {
ok: true
turnId: string
prompts: PromptTarget[]
}
> {
const displayed = await ctx.journal.readItem(target.itemId)
const displayedRefusal = promptCancellationRefusal(target, displayed)
if (displayedRefusal) {
return displayedRefusal
}
const cancellation = ctx.adapter.promptCancellation?.({
sessionId: ctx.sessionId,
itemId: target.itemId,
fence: ctx.fence
})
if (
!cancellation ||
!cancellation.itemIds.includes(target.itemId) ||
new Set(cancellation.itemIds).size !== cancellation.itemIds.length
) {
return invalid(`Item ${target.itemId} is not pending on a cancellable turn.`)
}
const items = await ctx.journal.readItems(cancellation.itemIds)
if (!items) {
return invalid(`The prompt group for ${target.itemId} is incomplete.`)
}
const prompts: PromptTarget[] = []
for (const item of items) {
if (
!parseAgentJournalItemKey(item.itemId) ||
(item.body.kind !== 'approval' && item.body.kind !== 'question')
) {
return invalid(`Item ${item.itemId} is not a pending prompt.`)
}
if (item.body.resolution.state === 'pending') {
prompts.push({ itemId: item.itemId, expectedRevision: item.revision })
}
}
const currentDisplayed = prompts.find((prompt) => prompt.itemId === target.itemId)
if (!currentDisplayed || currentDisplayed.expectedRevision !== target.expectedRevision) {
return (
promptCancellationRefusal(
target,
items.find((item) => item.itemId === target.itemId) ?? null
) ?? invalid(`Item ${target.itemId} is not a pending prompt.`)
)
}
return { ok: true, turnId: cancellation.turnId, prompts }
}
function promptCancellationRefusal(
target: PromptTarget,
item: AgentJournalRenderItem | null
): PromptCancellationRefusal | null {
if (!item || (item.body.kind !== 'approval' && item.body.kind !== 'question')) {
return invalid(`Item ${target.itemId} is not a pending prompt.`)
}
if (!parseAgentJournalItemKey(target.itemId)) {
return invalid(`Item id ${target.itemId} is not a well-formed item key.`)
}
if (item.revision !== target.expectedRevision) {
return refusal(
'agent_session_item_revision_stale',
`Item ${target.itemId} has moved on.`,
item.revision,
item.body.resolution
)
}
if (item.body.resolution.state !== 'pending') {
return refusal(
'agent_session_already_resolved',
`Item ${target.itemId} was already ${item.body.resolution.state}.`,
item.revision,
item.body.resolution
)
}
return null
}
function refusal(
code: AgentSessionWireRefusal['code'],
message: string,
currentRevision: number,
resolution: AgentJournalResolution
): PromptCancellationRefusal {
return {
ok: false,
refusal: { code, message, currentRevision, resolution }
}
}
@@ -5,6 +5,7 @@ import type { AgentSessionJournal } from '../agent-session-journal/journal-store
import {
runningTurnLifecycleRevisions,
settleStaleRunningTurnsOnAcquire,
stalePromptCancellationRevisions,
turnVerdictFromDeathEvidence
} from './structured-agent-session-stale-turn-verdict'
@@ -43,6 +44,35 @@ function legacyLifecycleItem(turnId: string, startedAt: number): AgentJournalRen
}
}
function approvalItem(state: 'pending' | 'resolved'): AgentJournalRenderItem {
return {
itemId: agentJournalItemKey({
provider: 'codex',
threadId: THREAD,
turnId: 'turn-2',
ordinal: 3
}),
revision: 1,
sequence: 3,
observedAt: 3,
body: {
kind: 'approval',
title: 'Run command?',
detail: null,
options: [],
resolution:
state === 'pending'
? { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
: {
state: 'resolved',
selectedOptionId: 'allow',
resolvedBy: 'client-1',
resolvedAt: 2
}
}
}
}
describe('turn verdict from death evidence', () => {
it('earns an end time only from an observed exit', () => {
expect(
@@ -105,6 +135,25 @@ describe('running turn lifecycle revisions', () => {
})
})
describe('stale prompt revisions', () => {
it('cancels only pending prompts that cannot belong to the replacement child', () => {
expect(
stalePromptCancellationRevisions([approvalItem('pending'), approvalItem('resolved')])
).toEqual([
expect.objectContaining({
body: expect.objectContaining({
resolution: {
state: 'cancelled',
selectedOptionId: null,
resolvedBy: null,
resolvedAt: null
}
})
})
])
})
})
describe('stale running turns on a cold acquire', () => {
function journalWith(items: AgentJournalRenderItem[]) {
const appendLifecycleBatch = vi.fn(async () => ({ epoch: 'epoch-1', sequence: 9 }))
@@ -145,6 +194,32 @@ describe('stale running turns on a cold acquire', () => {
})
})
it('terminalizes a pending prompt restored from the prior provider child', async () => {
const prompt = approvalItem('pending')
const { journal, appendLifecycleBatch } = journalWith([prompt])
await expect(
settleStaleRunningTurnsOnAcquire({
journal,
sessionId: 'session-1',
fence: 14,
acquisitionGeneration: 'generation-2'
})
).resolves.toBe(1)
expect(appendLifecycleBatch).toHaveBeenCalledWith(
expect.objectContaining({
mutations: [
expect.objectContaining({
body: expect.objectContaining({
resolution: expect.objectContaining({ state: 'cancelled' })
})
})
]
})
)
})
it('writes nothing when no turn is running and keys on the journal position without a generation', async () => {
const idle = journalWith([
lifecycleItem('turn-1', 'completed', 1, { startedAt: 10, completedAt: 20 })
@@ -17,6 +17,7 @@ import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-re
import { partitionJournalLifecycleMutations } from '../agent-session-journal/journal-lifecycle-batch-partition'
import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import { cancelledJournalPromptBody } from '../agent-session-journal/journal-prompt-body-bounds'
export type StructuredAgentSessionTurnVerdict =
| { state: 'interrupted'; completedAt: number }
@@ -58,6 +59,27 @@ export function runningTurnLifecycleRevisions(
return revisions
}
/** Pending prompts cannot belong to the replacement provider child. */
export function stalePromptCancellationRevisions(
items: readonly AgentJournalRenderItem[]
): JournalLifecycleMutationInput[] {
const revisions: JournalLifecycleMutationInput[] = []
for (const item of items) {
if (
(item.body.kind !== 'approval' && item.body.kind !== 'question') ||
item.body.resolution.state !== 'pending'
) {
continue
}
const identity = parseAgentJournalItemKey(item.itemId)
const body = cancelledJournalPromptBody(item.body)
if (identity && body) {
revisions.push({ kind: 'item', identity, body })
}
}
return revisions
}
function settledLifecycle(
lifecycle: AgentJournalTurnLifecycle,
verdict: StructuredAgentSessionTurnVerdict
@@ -75,8 +97,7 @@ function settledLifecycle(
return settled
}
/** A running row found when a NEW child is acquired belongs to a generation whose exit nobody
* observed. Must run before that child's buffered events land, or a live turn would be judged. */
/** Rows owned by the prior child must settle before the replacement's buffered events land. */
export async function settleStaleRunningTurnsOnAcquire(input: {
journal: AgentSessionJournal
sessionId: string
@@ -84,10 +105,11 @@ export async function settleStaleRunningTurnsOnAcquire(input: {
acquisitionGeneration: string | null
}): Promise<number> {
const { journal } = input
const revisions = runningTurnLifecycleRevisions(
journal.snapshot().items,
UNVERIFIABLE_TURN_VERDICT
)
const items = journal.snapshot().items
const revisions = [
...runningTurnLifecycleRevisions(items, UNVERIFIABLE_TURN_VERDICT),
...stalePromptCancellationRevisions(items)
]
const generation = input.acquisitionGeneration ?? `seq-${journal.cursor().sequence}`
const settlementId = `stale-turn:${input.sessionId}:${input.fence}:${generation}`
for (const chunk of partitionJournalLifecycleMutations(settlementId, revisions)) {
@@ -25,6 +25,7 @@ import type {
AgentSessionDispatchOutcome,
StructuredAgentSessionAdapter
} from './structured-agent-session-adapter'
import { preparePromptCancellation } from './structured-agent-session-prompt-cancellation'
export { performSetOption } from './structured-agent-session-turns-options'
export { performPrompt } from './structured-agent-session-turns-prompt'
@@ -207,11 +208,21 @@ export async function performCancel(
ctx: AgentSessionTurnContext,
input: {
clientOperationId: string
turnId: string
turnId?: string
scope?: 'background-tasks'
taskId?: string
prompt?: { itemId: string; expectedRevision: number }
}
): Promise<TurnOutcome<AgentSessionCancelResult>> {
const promptCancellation =
input.prompt && !input.scope ? await preparePromptCancellation(ctx, input.prompt) : null
if (promptCancellation && !promptCancellation.ok) {
return promptCancellation
}
const turnId = promptCancellation?.ok ? promptCancellation.turnId : input.turnId
if (!turnId) {
return invalid('Cancellation requires a turn or pending prompt.')
}
let cancelled = false
let note = 'Cancellation requested.'
try {
@@ -226,8 +237,9 @@ export async function performCancel(
: (
await ctx.adapter.cancelTurn({
sessionId: ctx.sessionId,
turnId: input.turnId,
fence: ctx.fence
turnId,
fence: ctx.fence,
...(input.prompt ? { promptItemId: input.prompt.itemId } : {})
})
).cancelled
if (!cancelled) {
@@ -239,9 +251,21 @@ export async function performCancel(
}`
}
if (input.scope) {
return { ok: true, value: { turnId: input.turnId, cancelled } }
return { ok: true, value: { turnId, cancelled } }
}
if (cancelled && promptCancellation?.ok) {
const settled = await ctx.journal.cancelPromptsAtRevisions({
prompts: promptCancellation.prompts,
settlementId: `prompt-cancel:${input.clientOperationId}`,
resolvedBy: ctx.resolvedBy,
resolvedAt: ctx.now(),
fence: ctx.fence
})
if (settled > 0) {
ctx.publish()
}
}
// Keyed by the operation id so a replayed cancel upserts one item, not two.
await appendStatus(ctx, input.clientOperationId, note)
return { ok: true, value: { turnId: input.turnId, cancelled } }
return { ok: true, value: { turnId, cancelled } }
}
@@ -4,6 +4,7 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { setStructuredAgentSessionHost } from '../../../native-chat/agent-session-wire/structured-agent-session-registry'
import {
AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY,
AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY,
RUNTIME_CAPABILITIES,
RUNTIME_PROTOCOL_VERSION,
@@ -148,6 +149,7 @@ describe('capability gating', () => {
it('advertises the capability without bumping the protocol version', () => {
expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY)
expect(RUNTIME_CAPABILITIES).toContain(AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY)
expect(RUNTIME_CAPABILITIES).toContain(AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY)
expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_HOLD_RUNTIME_CAPABILITY)
expect(RUNTIME_CAPABILITIES).toContain(STRUCTURED_AGENT_SESSION_REVEAL_RUNTIME_CAPABILITY)
// Additive methods do not break an old client; bumping would strand every
@@ -606,6 +608,16 @@ describe('method routing', () => {
expect(hostCalls.cancel).toHaveBeenCalledWith(expect.anything(), params)
})
it('routes a prompt compare-and-set target without a client turn id', async () => {
const params = {
envelope: envelope(),
prompt: { itemId: 'item-1', expectedRevision: 2 }
}
expect(await call('agentSession.cancel', params, STRUCTURED_CLIENT)).toMatchObject({ ok: true })
expect(hostCalls.cancel).toHaveBeenCalledWith(expect.anything(), params)
})
it('routes the structured handoff mutation through the host', async () => {
const response = await call('agentSession.requestHandoff', {
envelope: envelope(),
@@ -651,6 +663,21 @@ describe('parameter validation', () => {
expect(hostCalls.cancel).not.toHaveBeenCalled()
})
it('rejects prompt targets on background-task cancellation', async () => {
await rejects('agentSession.cancel', {
envelope: envelope(),
turnId: 'background-tasks',
scope: 'background-tasks',
prompt: { itemId: 'item-1', expectedRevision: 1 }
})
expect(hostCalls.cancel).not.toHaveBeenCalled()
})
it('requires either a turn id or a prompt target for cancellation', async () => {
await rejects('agentSession.cancel', { envelope: envelope() })
expect(hostCalls.cancel).not.toHaveBeenCalled()
})
it('refuses to let a client author anything but a user turn', async () => {
await rejects(
'agentSession.send',
@@ -40,7 +40,7 @@ export function NativeChatApprovalCard({
<button
type="button"
onClick={onCancel}
aria-label={translate('components.native-chat.approval.cancel', 'Cancel')}
aria-label={translate('components.native-chat.question.cancel', 'Cancel')}
className="flex size-6 shrink-0 items-center justify-center rounded-md text-muted-foreground transition-colors hover:bg-accent hover:text-accent-foreground focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring"
>
<X className="size-4" />
@@ -55,7 +55,8 @@ export function createStructuredSessionMocks() {
const moduleFactories = {
structuredAgentSessionClient: () => ({
callStructuredAgentSession: mocks.call
callStructuredAgentSession: mocks.call,
structuredAgentSessionSupportsPromptCancel: () => Promise.resolve(true)
}),
useStructuredAgentSession: async () => {
const { useStructuredAgentSessionOutbox } =
@@ -184,7 +184,7 @@ describe('NativeChatStructuredSession', () => {
expect(screen.queryByTestId('structured-composer')).toBeNull()
act(() => mocks.questionCardProps?.onCancel())
expect(mocks.cancel).toHaveBeenCalledWith('turn-question')
expect(mocks.cancel).toHaveBeenCalledWith('turn-question', legacySingleQuestionPromptItems[0])
expect(mocks.messageListProps?.showLiveTurnActivity).toBe(false)
mocks.promptItems = []
@@ -226,7 +226,7 @@ describe('NativeChatStructuredSession', () => {
mocks.promptItems = approvalItems
mocks.monitoringBackgroundTasks = true
render(
const view = () => (
<NativeChatStructuredSession
isVisible
isFocusedGroup
@@ -236,6 +236,7 @@ describe('NativeChatStructuredSession', () => {
agent="claude"
/>
)
const { rerender } = render(view())
expect(mocks.messageListProps).toMatchObject({
isWorking: true,
@@ -250,7 +251,13 @@ describe('NativeChatStructuredSession', () => {
expect(mocks.messageListProps?.showLiveTurnActivity).toBe(false)
act(() => mocks.approvalCardProps?.onCancel?.())
expect(mocks.cancel).toHaveBeenCalledWith('turn-approval')
expect(mocks.cancel).toHaveBeenCalledWith('turn-approval', approvalItems[0])
mocks.cancel.mockClear()
mocks.turnId = null
rerender(view())
act(() => mocks.approvalCardProps?.onCancel?.())
expect(mocks.cancel).toHaveBeenCalledWith(null, approvalItems[0])
})
// Every background-task test mounts the same local Claude session; only the ids
@@ -246,11 +246,7 @@ export function NativeChatStructuredSession(
}))
}}
onChoose={(optionId) => void controller.respond(prompt, optionId)}
onCancel={() => {
if (controller.turnId) {
void controller.cancel(controller.turnId)
}
}}
onCancel={() => void controller.cancel(controller.turnId, prompt)}
/>
) : null}
{prompt && questionBody ? (
@@ -301,9 +297,7 @@ export function NativeChatStructuredSession(
}
}}
onCancel={() => {
if (controller.turnId) {
void controller.cancel(controller.turnId)
}
void controller.cancel(controller.turnId, prompt)
}}
/>
) : null}
@@ -18,7 +18,8 @@ export type StructuredAgentSessionMutate = <T>(
method: string,
fingerprintMethod: string,
fields: Record<string, unknown>,
operationIdOverride?: string | null
operationIdOverride?: string | null,
expectedEnvironmentPairingRevision?: number
) => Promise<T | null>
export function useStructuredAgentSessionMutate(args: {
@@ -27,17 +28,30 @@ export function useStructuredAgentSessionMutate(args: {
/** Read at settle time, not at call time: the fence can move while a request
* is in flight, and a result from the previous fence is not this session's. */
stateRef: { current: { fence: number | null } }
}): { mutate: StructuredAgentSessionMutate; writeError: string | null } {
}): {
mutate: StructuredAgentSessionMutate
reportWriteError: (error: unknown, expectedFence: number | null) => void
writeError: string | null
} {
const { sessionId, stateRef, target } = args
const [writeError, setWriteError] = useState<string | null>(null)
const operationIds = useRef(new Map<string, string>())
const reportWriteError = useCallback(
(error: unknown, expectedFence: number | null): void => {
if (stateRef.current.fence === expectedFence) {
setWriteError(error instanceof Error ? error.message : 'Request was not sent')
}
},
[stateRef]
)
const mutate = useCallback(
async <T>(
method: string,
fingerprintMethod: string,
fields: Record<string, unknown>,
operationIdOverride?: string | null
operationIdOverride?: string | null,
expectedEnvironmentPairingRevision?: number
): Promise<T | null> => {
if (stateRef.current.fence === null) {
return null
@@ -49,7 +63,7 @@ export function useStructuredAgentSessionMutate(args: {
operationIds.current.set(key, clientOperationId)
let result: AgentSessionMutationResult<T>
try {
result = await callStructuredAgentSession<AgentSessionMutationResult<T>>(target, method, {
const params = {
envelope: {
sessionId,
clientOperationId,
@@ -61,11 +75,22 @@ export function useStructuredAgentSessionMutate(args: {
})
},
...fields
})
} catch (error) {
if (stateRef.current.fence === targetFence) {
setWriteError(error instanceof Error ? error.message : 'Request was not sent')
}
result =
expectedEnvironmentPairingRevision === undefined
? await callStructuredAgentSession<AgentSessionMutationResult<T>>(
target,
method,
params
)
: await callStructuredAgentSession<AgentSessionMutationResult<T>>(
target,
method,
params,
expectedEnvironmentPairingRevision
)
} catch (error) {
reportWriteError(error, targetFence)
return null
}
if (!result.ok) {
@@ -89,8 +114,8 @@ export function useStructuredAgentSessionMutate(args: {
setWriteError(null)
return result.value
},
[sessionId, stateRef, target]
[reportWriteError, sessionId, stateRef, target]
)
return { mutate, writeError }
return { mutate, reportWriteError, writeError }
}
@@ -0,0 +1,193 @@
// @vitest-environment happy-dom
import { act, renderHook } from '@testing-library/react'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types'
import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client'
const mocks = vi.hoisted(() => ({
call: vi.fn(),
supportsPromptCancel: vi.fn(),
operationId: vi.fn(() => 'cancel-operation')
}))
let items: AgentJournalRenderItem[] = []
vi.mock('@/runtime/structured-agent-session-client', () => ({
callStructuredAgentSession: mocks.call,
structuredAgentSessionSupportsPromptCancel: mocks.supportsPromptCancel
}))
vi.mock('@/runtime/runtime-environment-revision', () => ({
captureRuntimeEnvironmentRequestRevision: () => 31
}))
vi.mock('./native-chat-session-option-settings-write', () => ({
enqueueSessionOptionSettingsWrite: vi.fn()
}))
vi.mock('./use-structured-agent-session-read', () => ({
useStructuredAgentSessionRead: () => ({
state: {
fence: 3,
commands: undefined,
items,
submissions: [],
status: 'ready',
error: null,
hasOlder: false,
handoff: null
},
loadingOlder: false,
loadOlder: vi.fn()
})
}))
vi.mock('./use-structured-agent-session-outbox', () => ({
structuredSessionOperationId: mocks.operationId,
useStructuredAgentSessionOutbox: () => ({
outbox: [],
blockedClientMessageId: null,
error: null,
send: vi.fn(),
retry: vi.fn()
})
}))
import { useStructuredAgentSession } from './use-structured-agent-session'
const LOCAL_TARGET = { kind: 'local' } as const
function pendingPrompt(
kind: 'approval' | 'question',
itemId: string,
revision: number
): AgentJournalRenderItem {
const resolution = {
state: 'pending' as const,
selectedOptionId: null,
resolvedBy: null,
resolvedAt: null
}
return {
itemId,
revision,
sequence: 2,
observedAt: 2,
body:
kind === 'approval'
? {
kind,
title: 'Allow?',
detail: null,
options: [{ id: 'allow', label: 'Allow' }],
resolution
}
: {
kind,
question: 'Which approach?',
options: [{ id: 'focused', label: 'Focused' }],
resolution
}
}
}
function renderSession(target: RuntimeClientTarget = LOCAL_TARGET) {
return renderHook(() =>
useStructuredAgentSession({
sessionId: 'session-1',
agent: 'codex',
target,
isVisible: true
})
)
}
describe('useStructuredAgentSession prompt cancellation', () => {
beforeEach(() => {
vi.clearAllMocks()
items = []
mocks.call.mockResolvedValue(null)
mocks.supportsPromptCancel.mockResolvedValue(true)
})
it('cancels the exact displayed prompt revision on a capable host', async () => {
items = [pendingPrompt('question', 'question-1', 4)]
mocks.call.mockImplementation((_target, method) =>
method === 'agentSession.cancel'
? Promise.resolve({ ok: true, value: { turnId: 'turn-1', cancelled: true } })
: Promise.resolve(null)
)
const { result } = renderSession()
await act(async () => void (await result.current.cancel('turn-1', result.current.prompts[0])))
expect(mocks.call).toHaveBeenCalledWith(
LOCAL_TARGET,
'agentSession.cancel',
expect.objectContaining({
turnId: 'turn-1',
prompt: { itemId: 'question-1', expectedRevision: 4 }
})
)
})
it('lets a durable prompt identify its turn after foreground lifecycle settlement', async () => {
items = [pendingPrompt('approval', 'approval-restored', 3)]
mocks.call.mockImplementation((_target, method) =>
method === 'agentSession.cancel'
? Promise.resolve({ ok: true, value: { turnId: 'provider-child-turn', cancelled: true } })
: Promise.resolve(null)
)
const { result } = renderSession()
await act(async () => void (await result.current.cancel(null, result.current.prompts[0])))
const call = mocks.call.mock.calls.find((candidate) => candidate[1] === 'agentSession.cancel')
expect(call?.[2]).not.toHaveProperty('turnId')
expect(call?.[2]).toMatchObject({
prompt: { itemId: 'approval-restored', expectedRevision: 3 }
})
})
it('refuses prompt cancellation on an older remote host', async () => {
items = [pendingPrompt('approval', 'approval-1', 2)]
mocks.supportsPromptCancel.mockResolvedValue(false)
const target = { kind: 'environment', environmentId: 'legacy-host' } as const
const { result } = renderSession(target)
await act(async () => void (await result.current.cancel('turn-1', result.current.prompts[0])))
expect(mocks.call.mock.calls.some((call) => call[1] === 'agentSession.cancel')).toBe(false)
expect(mocks.supportsPromptCancel).toHaveBeenCalledWith(target, 31)
expect(result.current.error).toBe(
'Cancelling a pending prompt requires a newer Orca server. Update the server and try again.'
)
})
it('keeps generic turn cancellation available on an older remote host', async () => {
mocks.call.mockImplementation((_target, method) =>
method === 'agentSession.cancel'
? Promise.resolve({ ok: true, value: { turnId: 'turn-1', cancelled: true } })
: Promise.resolve(null)
)
const target = { kind: 'environment', environmentId: 'legacy-host' } as const
const { result } = renderSession(target)
await act(async () => void (await result.current.cancel('turn-1')))
const call = mocks.call.mock.calls.find((candidate) => candidate[1] === 'agentSession.cancel')
expect(call?.[2]).toMatchObject({ turnId: 'turn-1' })
expect(call?.[2]).not.toHaveProperty('prompt')
expect(call?.[3]).toBe(31)
expect(mocks.supportsPromptCancel).not.toHaveBeenCalled()
})
it('does not cancel when remote prompt-cancel support cannot be verified', async () => {
items = [pendingPrompt('approval', 'approval-1', 2)]
mocks.supportsPromptCancel.mockRejectedValue(new Error('Host unreachable'))
const { result } = renderSession({ kind: 'environment', environmentId: 'offline-host' })
await act(async () => {
await expect(result.current.cancel('turn-1', result.current.prompts[0])).resolves.toBeNull()
})
expect(mocks.call.mock.calls.some((call) => call[1] === 'agentSession.cancel')).toBe(false)
expect(result.current.error).toBe('Host unreachable')
})
})
@@ -5,6 +5,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'
const mocks = vi.hoisted(() => ({
call: vi.fn(),
supportsPromptCancel: vi.fn(),
operationId: vi.fn(),
enqueueSettingsWrite: vi.fn()
}))
@@ -14,7 +15,12 @@ let items: AgentJournalRenderItem[] = []
let submissions: AgentJournalSubmission[] = []
vi.mock('@/runtime/structured-agent-session-client', () => ({
callStructuredAgentSession: mocks.call
callStructuredAgentSession: mocks.call,
structuredAgentSessionSupportsPromptCancel: mocks.supportsPromptCancel
}))
vi.mock('@/runtime/runtime-environment-revision', () => ({
captureRuntimeEnvironmentRequestRevision: () => 31
}))
vi.mock('./native-chat-session-option-settings-write', () => ({
@@ -108,6 +114,7 @@ describe('useStructuredAgentSession working state', () => {
items = []
submissions = []
mocks.call.mockResolvedValue(null)
mocks.supportsPromptCancel.mockResolvedValue(true)
})
it('reports work from an unanswered dispatch, and keeps the turn id provider-minted', () => {
@@ -27,7 +27,11 @@ import {
hasUnansweredStructuredAgentSessionDispatch
} from '../../../../shared/structured-agent-session-projection'
import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client'
import { callStructuredAgentSession } from '@/runtime/structured-agent-session-client'
import { captureRuntimeEnvironmentRequestRevision } from '@/runtime/runtime-environment-revision'
import {
callStructuredAgentSession,
structuredAgentSessionSupportsPromptCancel
} from '@/runtime/structured-agent-session-client'
import { useStructuredAgentSessionHold } from './use-structured-agent-session-hold'
import { useStructuredAgentSessionRead } from './use-structured-agent-session-read'
import {
@@ -39,6 +43,7 @@ import { useStructuredAgentSessionMessages } from './use-structured-agent-sessio
import { selectStructuredAgentTurnActivity } from '../../../../shared/native-chat-turn-activity'
import { enqueueSessionOptionSettingsWrite } from './native-chat-session-option-settings-write'
import { useStructuredAgentTurnTiming } from './use-structured-agent-turn-timing'
import { AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE } from '../../../../shared/protocol-version'
export type { StructuredPromptItem } from './structured-agent-session-message-projection'
@@ -54,7 +59,11 @@ export function useStructuredAgentSession(args: {
useStructuredAgentSessionHold({ sessionId, target, surface: 'desktop-chat', enabled: isVisible })
const { state, loadingOlder, loadOlder } = useStructuredAgentSessionRead(args)
const stateRef = useRef(state)
const { mutate, writeError } = useStructuredAgentSessionMutate({ sessionId, target, stateRef })
const { mutate, reportWriteError, writeError } = useStructuredAgentSessionMutate({
sessionId,
target,
stateRef
})
const [conversationSupport, setConversationSupport] = useState<{
sessionId: string
commands: readonly AgentSessionConversationCommand[]
@@ -220,7 +229,44 @@ export function useStructuredAgentSession(args: {
turnActivity,
backgroundTasks,
turnId,
cancel: (turnId: string) => mutate('agentSession.cancel', 'agentSession.cancel', { turnId }),
cancel: async (turnId: string | null, prompt?: StructuredPromptItem) => {
const targetFence = stateRef.current.fence
const requestRevision =
target.kind === 'environment'
? captureRuntimeEnvironmentRequestRevision(target.environmentId)
: undefined
let promptSupported = false
if (prompt !== undefined) {
try {
promptSupported = await structuredAgentSessionSupportsPromptCancel(
target,
requestRevision
)
} catch (error) {
reportWriteError(error, targetFence)
return null
}
if (!promptSupported) {
reportWriteError(
new Error(AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE),
targetFence
)
return null
}
}
return mutate(
'agentSession.cancel',
'agentSession.cancel',
{
...(turnId ? { turnId } : {}),
...(promptSupported && prompt
? { prompt: { itemId: prompt.itemId, expectedRevision: prompt.revision } }
: {})
},
undefined,
requestRevision
)
},
stopBackgroundTask: (taskId?: string) =>
mutate('agentSession.cancel', 'agentSession.cancel', {
turnId: 'background-tasks',
@@ -1,7 +1,10 @@
// @vitest-environment happy-dom
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { AGENT_SESSION_REWIND_RUNTIME_CAPABILITY } from '../../../shared/protocol-version'
import {
AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY,
AGENT_SESSION_REWIND_RUNTIME_CAPABILITY
} from '../../../shared/protocol-version'
const mocks = vi.hoisted(() => ({
subscribe: vi.fn(),
@@ -10,7 +13,9 @@ const mocks = vi.hoisted(() => ({
}))
vi.mock('./runtime-environment-revision', () => ({
getRuntimeEnvironmentRevision: () => 7
getRuntimeEnvironmentRevision: () => 7,
captureRuntimeEnvironmentRequestRevision: (_environmentId: string, expected?: number) =>
expected ?? 7
}))
vi.mock('./runtime-rpc-client', () => ({
@@ -20,9 +25,59 @@ vi.mock('./runtime-rpc-client', () => ({
import {
callStructuredAgentSession,
structuredAgentSessionSupportsPromptCancel,
subscribeStructuredAgentSession
} from './structured-agent-session-client'
describe('structuredAgentSessionSupportsPromptCancel', () => {
beforeEach(() => vi.resetAllMocks())
it('uses the bundled host locally and capability-gates remote hosts', async () => {
mocks.call.mockResolvedValue({ capabilities: [] })
await expect(structuredAgentSessionSupportsPromptCancel({ kind: 'local' })).resolves.toBe(true)
await expect(
structuredAgentSessionSupportsPromptCancel({
kind: 'environment',
environmentId: 'env-1'
})
).resolves.toBe(false)
expect(mocks.call).toHaveBeenCalledExactlyOnceWith(
{ kind: 'environment', environmentId: 'env-1' },
'status.get',
undefined,
{ timeoutMs: 15_000, expectedEnvironmentPairingRevision: 7 }
)
})
it('recognizes prompt cancellation on the captured pairing revision', async () => {
mocks.call.mockResolvedValue({
capabilities: [AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY]
})
await expect(
structuredAgentSessionSupportsPromptCancel({ kind: 'environment', environmentId: 'env-1' }, 7)
).resolves.toBe(true)
expect(mocks.call).toHaveBeenCalledWith(
expect.anything(),
'status.get',
undefined,
expect.objectContaining({ expectedEnvironmentPairingRevision: 7 })
)
})
it('fails closed when remote capability status is unavailable', async () => {
mocks.call.mockRejectedValue(new Error('Host unreachable'))
await expect(
structuredAgentSessionSupportsPromptCancel({
kind: 'environment',
environmentId: 'env-1'
})
).rejects.toThrow('Host unreachable')
})
})
describe('callStructuredAgentSession rewind capability', () => {
const target = { kind: 'environment', environmentId: 'env-1' } as const
const params = { itemId: 'item-1', expectedEpoch: 'epoch-1' }
@@ -3,8 +3,15 @@ import type {
AgentSessionStatusEvent,
AgentSessionSubscribeEvent
} from '../../../shared/agent-session-wire'
import { getRuntimeEnvironmentRevision } from './runtime-environment-revision'
import { AGENT_SESSION_REWIND_RUNTIME_CAPABILITY } from '../../../shared/protocol-version'
import type { RuntimeStatus } from '../../../shared/runtime-types'
import {
captureRuntimeEnvironmentRequestRevision,
getRuntimeEnvironmentRevision
} from './runtime-environment-revision'
import {
AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY,
AGENT_SESSION_REWIND_RUNTIME_CAPABILITY
} from '../../../shared/protocol-version'
import {
callRuntimeRpc,
runtimeEnvironmentSupportsCapability,
@@ -14,7 +21,8 @@ import {
export async function callStructuredAgentSession<TResult>(
target: RuntimeClientTarget,
method: string,
params?: unknown
params?: unknown,
expectedEnvironmentPairingRevision?: number
): Promise<TResult> {
if (
method === 'agentSession.rewind' &&
@@ -27,8 +35,31 @@ export async function callStructuredAgentSession<TResult>(
throw new Error('Rewinding requires a newer Orca server. Update the server and try again.')
}
return method === 'agentSession.conversationCommand'
? callRuntimeRpc<TResult>(target, method, params, { timeoutMs: 195_000 })
: callRuntimeRpc<TResult>(target, method, params)
? callRuntimeRpc<TResult>(target, method, params, {
timeoutMs: 195_000,
expectedEnvironmentPairingRevision
})
: expectedEnvironmentPairingRevision === undefined
? callRuntimeRpc<TResult>(target, method, params)
: callRuntimeRpc<TResult>(target, method, params, { expectedEnvironmentPairingRevision })
}
export async function structuredAgentSessionSupportsPromptCancel(
target: RuntimeClientTarget,
expectedEnvironmentPairingRevision?: number
): Promise<boolean> {
if (target.kind === 'local') {
return true
}
const requestRevision = captureRuntimeEnvironmentRequestRevision(
target.environmentId,
expectedEnvironmentPairingRevision
)
const status = await callRuntimeRpc<RuntimeStatus>(target, 'status.get', undefined, {
timeoutMs: 15_000,
expectedEnvironmentPairingRevision: requestRevision
})
return status.capabilities?.includes(AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY) === true
}
async function subscribeStructuredAgentSessionMethod<TEvent>(
@@ -31,6 +31,9 @@ export type AgentSessionOperationOutcome =
| {
status: 'succeeded'
sessionId: string
/** Result of a turn-cancel mutation, absent on older rows and other operations. */
cancelled?: boolean
cancelledTurnId?: string
conversationCommand?: AgentSessionConversationCommandResult
rewind?: AgentSessionRewindResult
}
@@ -192,6 +195,8 @@ export function isAgentSessionOperationRow(value: unknown): value is AgentSessio
((outcome.status === 'pending' && true) ||
(outcome.status === 'succeeded' &&
typeof outcome.sessionId === 'string' &&
(outcome.cancelled === undefined || typeof outcome.cancelled === 'boolean') &&
(outcome.cancelledTurnId === undefined || typeof outcome.cancelledTurnId === 'string') &&
(outcome.rewind === undefined || isAgentSessionRewindResult(outcome.rewind)) &&
(outcome.conversationCommand === undefined ||
isAgentSessionConversationCommandResult(outcome.conversationCommand))) ||
+7
View File
@@ -167,6 +167,12 @@ export const STRUCTURED_AGENT_SESSION_RESUME_HISTORY_RUNTIME_CAPABILITY =
export const AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY = 'agent-session.status-feed.v1' as const
// The RPC is registered unconditionally; per-session rewind support is a separate check.
export const AGENT_SESSION_REWIND_RUNTIME_CAPABILITY = 'agent-session.rewind.v1' as const
// Why: agentSession.cancel uses a strict schema, so clients must not send the optional prompt CAS
// target until the host can terminalize that exact durable prompt after a confirmed interruption.
export const AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY =
'agent-session.prompt-cancel.v1' as const
export const AGENT_SESSION_PROMPT_CANCEL_UPDATE_REQUIRED_MESSAGE =
'Cancelling a pending prompt requires a newer Orca server. Update the server and try again.'
// Readers must understand a monitoring roster with no available stop control.
// Why: a `turn` journal item replaced the status row that used to carry a turn's lifecycle. A
// client that predates it would render the unknown kind as text, so the host publishes the legacy
@@ -292,6 +298,7 @@ export const RUNTIME_CAPABILITIES = [
STRUCTURED_AGENT_SESSION_RESUME_HISTORY_RUNTIME_CAPABILITY,
AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY,
AGENT_SESSION_REWIND_RUNTIME_CAPABILITY,
AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY,
AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY,
AGENT_SESSION_TURN_ITEM_CAPABILITY,
AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY,
@@ -162,14 +162,27 @@ export const SendParams = z
export const CancelParams = z
.object({
envelope: MutationEnvelope,
turnId: Identifier('Invalid turn id'),
turnId: Identifier('Invalid turn id').optional(),
scope: z.literal('background-tasks').optional(),
taskId: Identifier('Invalid task id').optional()
taskId: Identifier('Invalid task id').optional(),
prompt: z
.object({
itemId: Identifier('Invalid item id'),
expectedRevision: z.number().int().positive()
})
.strict()
.optional()
})
.strict()
.refine((value) => value.turnId !== undefined || value.prompt !== undefined, {
message: 'A turn id is required unless a prompt identifies its owning turn'
})
.refine((value) => value.taskId === undefined || value.scope === 'background-tasks', {
message: 'A task id requires background-task scope'
})
.refine((value) => value.prompt === undefined || value.scope === undefined, {
message: 'A prompt target cannot use background-task scope'
})
export const RespondParams = z
.object({