diff --git a/opensquilla-webui/src/components/chat/AssistantMessage.activity-fold.test.ts b/opensquilla-webui/src/components/chat/AssistantMessage.activity-fold.test.ts index 2381d2972..a5445c478 100644 --- a/opensquilla-webui/src/components/chat/AssistantMessage.activity-fold.test.ts +++ b/opensquilla-webui/src/components/chat/AssistantMessage.activity-fold.test.ts @@ -178,6 +178,7 @@ function planPart(): Extract { function clarifyPart( presentation?: string, + resolution: Extract['resolution'] = 'replied', ): Extract { return { type: 'interrupt', @@ -198,7 +199,7 @@ function clarifyPart( runId: 'plan-run-1', step: 'confirm_scope', }, - resolution: 'replied', + resolution, busy: false, error: '', } @@ -980,6 +981,21 @@ describe('AssistantMessage activity disclosure', () => { expect(el.querySelector('.clarify-outcome--plan')).toBeNull() }) + it('does not render an unavailable questionnaire as an action or success receipt', async () => { + const unavailable = clarifyPart('plan_questionnaire_v1', 'unavailable') + const el = mountMessage(baseMessage({ + text: '', + timelineItems: [approvalTimelineItem(unavailable)], + parts: [unavailable, planPart()], + statusHistory: [], + })) + await nextTick() + + expect(el.querySelector('.plan-card')).not.toBeNull() + expect(el.querySelector('.clarify-card')).toBeNull() + expect(el.querySelector('.clarify-outcome')).toBeNull() + }) + it('keeps intermediate candidate narration inside activity and the final answer outside once', async () => { const el = mountMessage(baseMessage({ text: 'Final verified answer.', diff --git a/opensquilla-webui/src/components/chat/AssistantMessage.vue b/opensquilla-webui/src/components/chat/AssistantMessage.vue index 637181347..2b5c813f5 100644 --- a/opensquilla-webui/src/components/chat/AssistantMessage.vue +++ b/opensquilla-webui/src/components/chat/AssistantMessage.vue @@ -649,7 +649,8 @@ const planParts = computed( const hasPlan = computed(() => planParts.value.length > 0) const standaloneInterruptParts = computed(() => interruptParts.value.filter(part => ( - !timelineResolvedInterruptKeys.value.has(part.key) + !(part.interruptKind === 'clarify' && part.resolution === 'unavailable') + && !timelineResolvedInterruptKeys.value.has(part.key) && !( hasPlan.value && part.interruptKind === 'clarify' @@ -916,11 +917,21 @@ function withoutFailedActivity( }) } +function withoutUnavailableClarifies( + items: ChatStreamTimelineItem[], +): ChatStreamTimelineItem[] { + return items.filter(item => !( + item.type === 'interrupt' + && item.part.interruptKind === 'clarify' + && item.part.resolution === 'unavailable' + )) +} + const visibleActivityItems = computed(() => - withoutFailedActivity(activityProjection.value.activityItems), + withoutUnavailableClarifies(withoutFailedActivity(activityProjection.value.activityItems)), ) const visibleLegacyTimelineItems = computed(() => - withoutFailedActivity(props.message.timelineItems ?? []), + withoutUnavailableClarifies(withoutFailedActivity(props.message.timelineItems ?? [])), ) const visibleActivityCallKeys = computed(() => new Set( visibleActivityItems.value.flatMap(item => diff --git a/opensquilla-webui/src/components/chat/PlanRunRibbon.test.ts b/opensquilla-webui/src/components/chat/PlanRunRibbon.test.ts index 8b831ef66..890dca3ef 100644 --- a/opensquilla-webui/src/components/chat/PlanRunRibbon.test.ts +++ b/opensquilla-webui/src/components/chat/PlanRunRibbon.test.ts @@ -787,6 +787,8 @@ describe('PlanRunRibbon', () => { expect(chatViewSource).toContain('@stop="onComposerStop"') expect(chatViewSource).toContain('@focus-return="focusComposerAfterPlanRun"') expect(chatViewSource).toContain(':stop-targets-plan-run="composerStopsPlanRun"') + expect(chatViewSource).toContain('planRunSettlementPending.value') + expect(chatViewSource).toContain('planActionPending !== null || planRunSettlementPending') expect(chatViewSource).toContain("activePlanRun.value?.status === 'queued'") expect(chatViewSource).toContain("activePlanRun.value?.status === 'running'") }) diff --git a/opensquilla-webui/src/components/chat/parts/InterruptPart.vue b/opensquilla-webui/src/components/chat/parts/InterruptPart.vue index 163f2d70b..095a671bf 100644 --- a/opensquilla-webui/src/components/chat/parts/InterruptPart.vue +++ b/opensquilla-webui/src/components/chat/parts/InterruptPart.vue @@ -12,7 +12,11 @@ @extend="emit('extend', part.approval.approvalId)" /> { function deferred() { let resolve!: (value: T) => void - const promise = new Promise(done => { resolve = done }) - return { promise, resolve } + let reject!: (reason?: unknown) => void + const promise = new Promise((done, fail) => { + resolve = done + reject = fail + }) + return { promise, resolve, reject } } async function harness(statusResult: unknown = { found: true, pending: true, resolved: false }) { @@ -26,6 +31,18 @@ async function harness(statusResult: unknown = { found: true, pending: true, res const rpcCall = vi.fn(async (_method?: string, _params?: Record) => statusResult as T) const appendInterruptFrame = vi.fn() const interruptState = ref>(new Map()) + const currentEpoch = ref(3) + const streamGeneration = ref('generation-1') + const observeStreamGeneration = vi.fn((source: unknown) => { + const value = source && typeof source === 'object' + ? source as Record + : {} + const generation = String(value.stream_generation ?? value.streamGeneration ?? '').trim() + if (!generation || generation === streamGeneration.value) return false + streamGeneration.value = generation + return true + }) + const sessionKey = ref('agent:main:web') const scope = effectScope() const approvalCenter: any = { snapshot: vi.fn(async () => { @@ -94,7 +111,10 @@ async function harness(statusResult: unknown = { found: true, pending: true, res return () => handlers.delete(event) }), }), - sessionKey: ref('agent:main:web'), + sessionKey, + currentEpoch, + streamGeneration, + observeStreamGeneration, runStatus: ref({ status: 'idle', label: '', task: null }), stream: { isStreaming: ref(false), @@ -108,7 +128,19 @@ async function harness(statusResult: unknown = { found: true, pending: true, res const unsubscribe = approvals.subscribe() await vi.waitFor(() => expect(fetch).toHaveBeenCalled()) vi.mocked(fetch).mockClear() - return { approvals, handlers, rpcCall, appendInterruptFrame, interruptState, unsubscribe, scope } + return { + approvals, + handlers, + rpcCall, + appendInterruptFrame, + interruptState, + currentEpoch, + streamGeneration, + observeStreamGeneration, + sessionKey, + unsubscribe, + scope, + } } function installSnapshot(pending: unknown[] = []) { @@ -465,6 +497,151 @@ describe('clarify tool-result recovery', () => { } }) + it('ignores a paused clarify replay from an older session epoch', async () => { + installSnapshot() + const runtime = await harness() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + epoch: runtime.currentEpoch.value - 1, + stream_generation: runtime.streamGeneration.value, + tool_use_id: 'stale-epoch-request', + name: 'request_user_input', + result: { ...planClarifyResult, request_id: 'stale-epoch-request' }, + }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.has('stale-epoch-request')).toBe(false) + expect(runtime.appendInterruptFrame).not.toHaveBeenCalled() + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('ignores a paused clarify replay from a retired transport generation', async () => { + installSnapshot() + const runtime = await harness() + try { + runtime.streamGeneration.value = 'generation-2' + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + epoch: runtime.currentEpoch.value, + stream_generation: 'generation-1', + tool_use_id: 'stale-generation-request', + name: 'request_user_input', + result: { ...planClarifyResult, request_id: 'stale-generation-request' }, + }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.has('stale-generation-request')).toBe(false) + expect(runtime.appendInterruptFrame).not.toHaveBeenCalled() + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('observes a new generation before appending and rejects all retired snapshot state', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + runtime.observeStreamGeneration.mockImplementationOnce((source: unknown) => { + // Model resetLiveTurnState clearing the fold log. If observation runs + // after the approvals append, this erases the only actionable frame. + runtime.appendInterruptFrame.mockClear() + const value = source as Record + runtime.streamGeneration.value = String(value.stream_generation || '') + return true + }) + handler?.({ + session_key: 'agent:main:web', + epoch: runtime.currentEpoch.value, + stream_generation: 'generation-2', + stream_seq: 1, + tool_use_id: 'first-new-generation-request', + name: 'request_user_input', + result: { ...planClarifyResult, request_id: 'first-new-generation-request' }, + }) + + expect(runtime.streamGeneration.value).toBe('generation-2') + expect(runtime.approvals.pendingClarify.value?.requestId) + .toBe('first-new-generation-request') + expect(runtime.observeStreamGeneration).toHaveBeenCalledTimes(1) + expect(runtime.appendInterruptFrame).toHaveBeenCalledTimes(1) + expect(runtime.observeStreamGeneration.mock.invocationCallOrder[0]) + .toBeLessThan(runtime.appendInterruptFrame.mock.invocationCallOrder[0]!) + + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [], + streamGeneration: 'generation-1', + goalSnapshotStreamSeq: 200, + }) + expect(runtime.approvals.pendingClarify.value?.requestId) + .toBe('first-new-generation-request') + const appendCount = runtime.appendInterruptFrame.mock.calls.length + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [{ + ...planClarifyResult, + request_id: 'retired-snapshot-request', + }], + streamGeneration: 'generation-1', + goalSnapshotStreamSeq: 201, + }) + expect(runtime.interruptState.value.has('retired-snapshot-request')).toBe(false) + expect(runtime.approvals.pendingClarify.value?.requestId) + .toBe('first-new-generation-request') + expect(runtime.appendInterruptFrame).toHaveBeenCalledTimes(appendCount) + + // The retired namespace cannot roll the live ingress back either. + handler?.({ + session_key: 'agent:main:web', + epoch: runtime.currentEpoch.value, + stream_generation: 'generation-1', + stream_seq: 200, + tool_use_id: 'late-retired-generation-request', + name: 'request_user_input', + result: { ...planClarifyResult, request_id: 'late-retired-generation-request' }, + }) + + expect(runtime.interruptState.value.has('late-retired-generation-request')).toBe(false) + expect(runtime.appendInterruptFrame).toHaveBeenCalledTimes(appendCount) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('retires a shared generation change when navigation happens before another tool result', async () => { + installSnapshot() + const runtime = await harness() + try { + // A non-tool wildcard event advanced the transport cursor; navigation is + // the first approvals lifecycle hook that observes that transition. + runtime.streamGeneration.value = 'generation-2' + runtime.sessionKey.value = 'agent:other:web' + await nextTick() + + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:other:web', + epoch: runtime.currentEpoch.value, + stream_generation: 'generation-1', + stream_seq: 200, + tool_use_id: 'post-navigation-retired-request', + name: 'request_user_input', + result: { ...planClarifyResult, request_id: 'post-navigation-retired-request' }, + }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.has('post-navigation-retired-request')).toBe(false) + expect(runtime.appendInterruptFrame).not.toHaveBeenCalled() + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + it('hydrates a pending deferred request after reconnect', async () => { installSnapshot() const runtime = await harness() @@ -485,7 +662,7 @@ describe('clarify tool-result recovery', () => { } }) - it('releases a failed submit when reconnect says that request is no longer pending', async () => { + it('marks a request unavailable when an authoritative empty snapshot omits it', async () => { installSnapshot() const runtime = await harness() try { @@ -495,20 +672,12 @@ describe('clarify tool-result recovery', () => { name: 'request_user_input', result: planClarifyResult, }) - runtime.rpcCall.mockRejectedValueOnce(new Error('connection lost after send')) - await runtime.approvals.submitClarify({ scope: 'focused' }) - - expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') - expect(runtime.approvals.clarifyError.value).toContain('connection lost after send') runtime.approvals.applyUserInputBootstrap({ pendingUserInputs: [] }) expect(runtime.approvals.pendingClarify.value).toBeNull() - expect(runtime.approvals.clarifySubmitted.value).toBe(false) - expect(runtime.approvals.clarifyBusy.value).toBe(false) - expect(runtime.approvals.clarifyError.value).toBe('') expect(runtime.interruptState.value.get('input-request-1')).toEqual({ - resolution: 'replied', + resolution: 'unavailable', busy: false, error: '', }) @@ -518,230 +687,1075 @@ describe('clarify tool-result recovery', () => { } }) - it('does not settle a failed submit from a partial snapshot without pending-input state', async () => { + it('does not treat the broker snapshot as authoritative for a legacy Meta clarify', async () => { installSnapshot() const runtime = await harness() try { runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', - tool_use_id: 'request-input-1', + stream_generation: runtime.streamGeneration.value, + stream_seq: 8, + tool_use_id: 'legacy-meta-clarify', name: 'request_user_input', - result: planClarifyResult, + result: { + ...planClarifyResult, + request_id: undefined, + run_id: 'meta-run-1', + }, }) - runtime.rpcCall.mockRejectedValueOnce(new Error('gateway unavailable')) - await runtime.approvals.submitClarify({ scope: 'focused' }) - runtime.approvals.applyUserInputBootstrap({}) + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [], + streamGeneration: runtime.streamGeneration.value, + goalSnapshotStreamSeq: 8, + }) - expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') - expect(runtime.approvals.clarifyError.value).toContain('gateway unavailable') - expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBeNull() + expect(runtime.approvals.pendingClarify.value?.requestId).toBeUndefined() + expect(runtime.approvals.pendingClarify.value?.runId).toBe('meta-run-1') + expect(runtime.interruptState.value.get('meta-run-1|confirm_scope')?.resolution) + .toBeNull() } finally { runtime.unsubscribe() runtime.scope.stop() } }) - it('settles the matching card when the same tool id returns answered', async () => { + it('does not let an older session-read hydration erase a newer live request', async () => { installSnapshot() const runtime = await harness() try { - const handler = runtime.handlers.get('session.event.tool_result') - handler?.({ - session_key: 'agent:main:web', - tool_use_id: 'request-input-1', - name: 'request_user_input', - result: clarifyResult, - }) - handler?.({ + runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', + stream_seq: 8, tool_use_id: 'request-input-1', name: 'request_user_input', - result: { - kind: 'user_input', - status: 'answered', - paused: false, - request_id: 'input-request-1', - answers: { scope: 'focused' }, - }, + result: planClarifyResult, }) - expect(runtime.interruptState.value.get('input-request-1')).toEqual({ - resolution: 'replied', - busy: false, - error: '', - }) + const olderHydration = Object.freeze({ + pendingUserInputs: Object.freeze([]), + goalSnapshotStreamSeq: 7, + deferredFields: Object.freeze([]), + }) satisfies Pick< + SessionReadMetadata, + 'pendingUserInputs' | 'goalSnapshotStreamSeq' | 'deferredFields' + > + runtime.approvals.applyUserInputBootstrap(olderHydration) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBeNull() + + runtime.approvals.applyUserInputBootstrap(Object.freeze({ + pendingUserInputs: Object.freeze([]), + goalSnapshotStreamSeq: 8, + deferredFields: Object.freeze([]), + })) expect(runtime.approvals.pendingClarify.value).toBeNull() - expect(runtime.approvals.clarifySubmitted.value).toBe(false) - expect(runtime.approvals.clarifyBusy.value).toBe(false) - expect(runtime.approvals.clarifyError.value).toBe('') + expect(runtime.interruptState.value.get('input-request-1')?.resolution) + .toBe('unavailable') } finally { runtime.unsubscribe() runtime.scope.stop() } }) - it('keeps a Plan questionnaire actionable when submission fails', async () => { + it('does not compare request cursors across different transport generations', async () => { installSnapshot() const runtime = await harness() try { + runtime.streamGeneration.value = 'generation-2' runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', + stream_generation: 'generation-2', + stream_seq: 2, tool_use_id: 'request-input-1', name: 'request_user_input', result: planClarifyResult, }) - runtime.rpcCall.mockRejectedValueOnce(new Error('gateway unavailable')) - await runtime.approvals.submitClarify({ scope: 'focused' }) + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [], + stream_generation: 'generation-1', + goalSnapshotStreamSeq: 200, + }) - expect(runtime.approvals.pendingClarify.value).toEqual(expect.objectContaining({ - requestId: 'input-request-1', - presentation: 'plan_questionnaire_v1', - })) - expect(runtime.approvals.clarifySubmitted.value).toBe(false) - expect(runtime.approvals.clarifyBusy.value).toBe(false) - expect(runtime.approvals.clarifyError.value).toContain('gateway unavailable') - expect(runtime.interruptState.value.get('input-request-1')).toEqual(expect.objectContaining({ - resolution: null, - busy: false, - })) + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBeNull() + + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [], + stream_generation: 'generation-2', + goalSnapshotStreamSeq: 2, + }) + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.get('input-request-1')?.resolution) + .toBe('unavailable') } finally { runtime.unsubscribe() runtime.scope.stop() } }) - it('does not let a delayed submit response dismiss a newer questionnaire', async () => { + it('lets a new transport generation reconcile requests left by the old one', async () => { installSnapshot() const runtime = await harness() - const submitted = deferred() try { - const handler = runtime.handlers.get('session.event.tool_result') - handler?.({ + runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', + stream_generation: 'generation-1', + stream_seq: 100, tool_use_id: 'request-input-1', name: 'request_user_input', result: planClarifyResult, }) - runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) - const firstSubmit = runtime.approvals.submitClarify({ scope: 'focused' }) - await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledWith( - 'chat.clarify_submit', - expect.objectContaining({ request_id: 'input-request-1' }), - )) + runtime.streamGeneration.value = 'generation-2' - handler?.({ - session_key: 'agent:main:web', - tool_use_id: 'request-input-2', - name: 'request_user_input', - result: { - ...planClarifyResult, - request_id: 'input-request-2', - run_id: 'plan-run-2', - }, + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [], + stream_generation: 'generation-2', + goalSnapshotStreamSeq: 0, }) - submitted.resolve({ resolved: true, request_id: 'input-request-1' }) - await firstSubmit - expect(runtime.approvals.pendingClarify.value).toEqual(expect.objectContaining({ - requestId: 'input-request-2', - runId: 'plan-run-2', - })) - expect(runtime.approvals.clarifySubmitted.value).toBe(false) - expect(runtime.approvals.clarifyBusy.value).toBe(false) - expect(runtime.approvals.clarifyError.value).toBe('') + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.get('input-request-1')?.resolution) + .toBe('unavailable') } finally { runtime.unsubscribe() runtime.scope.stop() } }) - it('does not clear a newer request for a non-matching terminal outcome', async () => { + it('ignores the placeholder pending list while that hydration field is deferred', async () => { installSnapshot() const runtime = await harness() try { - const handler = runtime.handlers.get('session.event.tool_result') - handler?.({ - session_key: 'agent:main:web', - tool_use_id: 'request-input-2', - name: 'request_user_input', - result: { - ...planClarifyResult, - request_id: 'input-request-2', - run_id: 'plan-run-2', - }, - }) - handler?.({ + runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', + stream_seq: 8, tool_use_id: 'request-input-1', name: 'request_user_input', - result: { - kind: 'user_input', - status: 'answered', - paused: false, - request_id: 'input-request-1', - answers: { scope: 'focused' }, - }, + result: planClarifyResult, }) - expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-2') - expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe('replied') + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [], + goalSnapshotStreamSeq: null, + deferred_fields: ['pendingUserInputs', 'goalSnapshotStreamSeq'], + }) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBeNull() } finally { runtime.unsubscribe() runtime.scope.stop() } }) - it('does not resurrect a settled request from a late paused-event replay', async () => { + it('marks a failed submit unavailable when reconnect says it is no longer pending', async () => { installSnapshot() const runtime = await harness() try { - const handler = runtime.handlers.get('session.event.tool_result') - handler?.({ - session_key: 'agent:main:web', - tool_use_id: 'request-input-1', - name: 'request_user_input', - result: { - kind: 'user_input', - status: 'answered', - paused: false, - request_id: 'input-request-1', - answers: { scope: 'focused' }, - }, - }) - const appendCount = runtime.appendInterruptFrame.mock.calls.length - - handler?.({ + runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', tool_use_id: 'request-input-1', name: 'request_user_input', result: planClarifyResult, }) + runtime.rpcCall.mockRejectedValueOnce(new Error('connection lost after send')) + await runtime.approvals.submitClarify({ scope: 'focused' }) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.approvals.clarifyError.value).toContain('connection lost after send') + + runtime.approvals.applyUserInputBootstrap({ pendingUserInputs: [] }) expect(runtime.approvals.pendingClarify.value).toBeNull() - expect(runtime.appendInterruptFrame).toHaveBeenCalledTimes(appendCount) - expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe('replied') + expect(runtime.approvals.clarifySubmitted.value).toBe(false) + expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.approvals.clarifyError.value).toBe('') + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) } finally { runtime.unsubscribe() runtime.scope.stop() } }) - it('retains the legacy cross-turn clarify receipt after a successful send acknowledgement', async () => { + it.each([ + 'cancelled', + 'timeout', + 'failed', + 'abandoned', + 'interrupted', + ] as const)('unlocks a pending questionnaire when its task becomes %s', async (status) => { installSnapshot() const runtime = await harness() try { runtime.handlers.get('session.event.tool_result')?.({ session_key: 'agent:main:web', - tool_use_id: 'legacy-clarify', + tool_use_id: 'request-input-1', name: 'request_user_input', - result: { + result: planClarifyResult, + }) + + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'another-plan-run', + status, + )).toBe(false) + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + status, + )).toBe(true) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + status, + )).toBe(false) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('settles every pending questionnaire owned by the terminal task', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + for (const requestId of ['input-request-1', 'input-request-2']) { + handler?.({ + session_key: 'agent:main:web', + tool_use_id: requestId, + name: 'request_user_input', + result: { + ...planClarifyResult, + request_id: requestId, + step: `confirm_${requestId}`, + }, + }) + } + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-2') + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'cancelled', + )).toBe(true) + + for (const requestId of ['input-request-1', 'input-request-2']) { + expect(runtime.interruptState.value.get(requestId)).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) + } + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'cancelled', + )).toBe(false) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('rejects a previously unseen questionnaire delivered after its task terminated', async () => { + installSnapshot() + const runtime = await harness() + try { + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'cancelled', + )).toBe(false) + + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.appendInterruptFrame).not.toHaveBeenCalled() + expect(runtime.interruptState.value.has('input-request-1')).toBe(false) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('rejects a late positive pending-input hydrate after task termination', async () => { + installSnapshot() + const runtime = await harness() + try { + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'failed', + )).toBe(false) + + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [planClarifyResult], + goalSnapshotStreamSeq: 7, + }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.appendInterruptFrame).not.toHaveBeenCalled() + expect(runtime.interruptState.value.has('input-request-1')).toBe(false) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('settles only questionnaires owned by the terminal task', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-2', + name: 'request_user_input', + result: { + ...planClarifyResult, + request_id: 'input-request-2', + run_id: 'plan-run-2', + }, + }) + + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'timeout', + )).toBe(true) + expect(runtime.interruptState.value.get('input-request-1')?.resolution) + .toBe('unavailable') + expect(runtime.interruptState.value.get('input-request-2')?.resolution).toBeNull() + expect(runtime.approvals.pendingClarify.value).toEqual(expect.objectContaining({ + requestId: 'input-request-2', + runId: 'plan-run-2', + })) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('reconciles every known questionnaire against an authoritative snapshot', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + const secondRequest = { + ...planClarifyResult, + request_id: 'input-request-2', + run_id: 'plan-run-2', + } + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-2', + name: 'request_user_input', + result: secondRequest, + }) + + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [secondRequest], + }) + + expect(runtime.interruptState.value.get('input-request-1')?.resolution) + .toBe('unavailable') + expect(runtime.interruptState.value.get('input-request-2')?.resolution).toBeNull() + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-2') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('does not report replied before the clarify RPC acknowledgement', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledWith( + 'chat.clarify_submit', + expect.objectContaining({ request_id: 'input-request-1' }), + )) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.approvals.clarifySubmitted.value).toBe(false) + expect(runtime.approvals.clarifyBusy.value).toBe(true) + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: null, + busy: true, + error: '', + }) + + submitted.resolve({ resolved: true, request_id: 'input-request-1' }) + await submission + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe('replied') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('does not infer a reply from an empty snapshot before a late RPC rejection', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledWith( + 'chat.clarify_submit', + expect.objectContaining({ request_id: 'input-request-1' }), + )) + runtime.approvals.applyUserInputBootstrap({ pendingUserInputs: [] }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) + + submitted.reject(new Error('late transport failure')) + await submission + + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) + expect(runtime.approvals.clarifyError.value).toBe('') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('lets a late successful RPC acknowledgement upgrade an unavailable snapshot', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + runtime.approvals.applyUserInputBootstrap({ pendingUserInputs: [] }) + expect(runtime.interruptState.value.get('input-request-1')?.resolution) + .toBe('unavailable') + + submitted.resolve({ resolved: true, request_id: 'input-request-1' }) + await submission + + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'replied', + busy: false, + error: '', + }) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('keeps an in-flight request busy across a duplicate paused replay', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + await runtime.approvals.submitClarify({ scope: 'complete' }) + + expect(runtime.rpcCall).toHaveBeenCalledTimes(1) + expect(runtime.approvals.clarifySubmitted.value).toBe(false) + expect(runtime.approvals.clarifyBusy.value).toBe(true) + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: null, + busy: true, + error: '', + }) + + submitted.resolve({ resolved: true, request_id: 'input-request-1' }) + await submission + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('restores the selected request busy state across a multi-request bootstrap', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + const secondRequest = { + ...planClarifyResult, + request_id: 'input-request-2', + step: 'confirm_delivery', + } + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + stream_seq: 4, + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + handler?.({ + session_key: 'agent:main:web', + stream_seq: 5, + tool_use_id: 'request-input-2', + name: 'request_user_input', + result: secondRequest, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + + runtime.approvals.applyUserInputBootstrap({ + pendingUserInputs: [planClarifyResult, secondRequest], + goalSnapshotStreamSeq: 5, + }) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-2') + expect(runtime.approvals.clarifyBusy.value).toBe(true) + expect(runtime.interruptState.value.get('input-request-2')).toEqual({ + resolution: null, + busy: true, + error: '', + }) + + submitted.resolve({ resolved: true, request_id: 'input-request-2' }) + await submission + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('allows a different request to submit while the docked request is busy', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + const firstRequest = { ...runtime.approvals.pendingClarify.value! } + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-2', + name: 'request_user_input', + result: { + ...planClarifyResult, + request_id: 'input-request-2', + step: 'confirm_delivery', + }, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + const dockedSubmission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + + await runtime.approvals.submitClarify({ scope: 'complete' }, firstRequest) + + expect(runtime.rpcCall).toHaveBeenCalledTimes(2) + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe('replied') + expect(runtime.interruptState.value.get('input-request-2')?.busy).toBe(true) + + submitted.resolve({ resolved: true, request_id: 'input-request-2' }) + await dockedSubmission + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('deduplicates an in-flight recovered inline request by its interrupt state', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + const recoveredRequest = { + intro: 'Recover this request.', + fields: [{ + name: 'scope', + type: 'enum', + required: true, + prompt: 'Which scope?', + defaultValue: '', + choices: ['focused', 'complete'], + }], + requestId: 'recovered-request-1', + runId: 'recovered-task-1', + step: 'confirm_scope', + } + try { + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + const firstSubmit = runtime.approvals.submitClarify( + { scope: 'focused' }, + recoveredRequest, + ) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + + await runtime.approvals.submitClarify( + { scope: 'complete' }, + recoveredRequest, + ) + + expect(runtime.rpcCall).toHaveBeenCalledTimes(1) + expect(runtime.interruptState.value.get('recovered-request-1')).toEqual({ + resolution: null, + busy: true, + error: '', + }) + + submitted.resolve({ resolved: true, request_id: 'recovered-request-1' }) + await firstSubmit + expect(runtime.interruptState.value.get('recovered-request-1')?.resolution).toBe('replied') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('does not reopen a terminal request when an in-flight RPC rejects late', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'timeout', + )).toBe(true) + submitted.reject(new Error('late transport failure')) + await submission + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.approvals.clarifyError.value).toBe('') + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('does not settle a failed submit from a partial snapshot without pending-input state', async () => { + installSnapshot() + const runtime = await harness() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockRejectedValueOnce(new Error('gateway unavailable')) + await runtime.approvals.submitClarify({ scope: 'focused' }) + + runtime.approvals.applyUserInputBootstrap({}) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-1') + expect(runtime.approvals.clarifyError.value).toContain('gateway unavailable') + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBeNull() + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('settles the matching card when the same tool id returns answered', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: clarifyResult, + }) + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: { + kind: 'user_input', + status: 'answered', + paused: false, + request_id: 'input-request-1', + answers: { scope: 'focused' }, + }, + }) + + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'replied', + busy: false, + error: '', + }) + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.approvals.clarifySubmitted.value).toBe(false) + expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.approvals.clarifyError.value).toBe('') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it.each(['cancelled', 'expired'] as const)( + 'marks the matching card unavailable when the user-input outcome is %s', + async (status) => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: { + kind: 'user_input', + status, + paused: false, + request_id: 'input-request-1', + }, + }) + + expect(runtime.interruptState.value.get('input-request-1')).toEqual({ + resolution: 'unavailable', + busy: false, + error: '', + }) + expect(runtime.approvals.pendingClarify.value).toBeNull() + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }, + ) + + it('keeps an acknowledged reply dominant over a later unavailable outcome replay', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + for (const status of ['answered', 'expired'] as const) { + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: { + kind: 'user_input', + status, + paused: false, + request_id: 'input-request-1', + ...(status === 'answered' ? { answers: { scope: 'focused' } } : {}), + }, + }) + } + + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe('replied') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('keeps a Plan questionnaire actionable when submission fails', async () => { + installSnapshot() + const runtime = await harness() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockRejectedValueOnce(new Error('gateway unavailable')) + + await runtime.approvals.submitClarify({ scope: 'focused' }) + + expect(runtime.approvals.pendingClarify.value).toEqual(expect.objectContaining({ + requestId: 'input-request-1', + presentation: 'plan_questionnaire_v1', + })) + expect(runtime.approvals.clarifySubmitted.value).toBe(false) + expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.approvals.clarifyError.value).toContain('gateway unavailable') + expect(runtime.interruptState.value.get('input-request-1')).toEqual(expect.objectContaining({ + resolution: null, + busy: false, + })) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('does not let a delayed submit response dismiss a newer questionnaire', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + const firstSubmit = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledWith( + 'chat.clarify_submit', + expect.objectContaining({ request_id: 'input-request-1' }), + )) + + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-2', + name: 'request_user_input', + result: { + ...planClarifyResult, + request_id: 'input-request-2', + run_id: 'plan-run-2', + }, + }) + submitted.resolve({ resolved: true, request_id: 'input-request-1' }) + await firstSubmit + + expect(runtime.approvals.pendingClarify.value).toEqual(expect.objectContaining({ + requestId: 'input-request-2', + runId: 'plan-run-2', + })) + expect(runtime.approvals.clarifySubmitted.value).toBe(false) + expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.approvals.clarifyError.value).toBe('') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('does not clear a newer request for a non-matching terminal outcome', async () => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-2', + name: 'request_user_input', + result: { + ...planClarifyResult, + request_id: 'input-request-2', + run_id: 'plan-run-2', + }, + }) + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: { + kind: 'user_input', + status: 'answered', + paused: false, + request_id: 'input-request-1', + answers: { scope: 'focused' }, + }, + }) + + expect(runtime.approvals.pendingClarify.value?.requestId).toBe('input-request-2') + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe('replied') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it.each([ + ['answered', 'replied'], + ['cancelled', 'unavailable'], + ['expired', 'unavailable'], + ] as const)( + 'does not resurrect a %s request from a late paused-event replay', + async (status, resolution) => { + installSnapshot() + const runtime = await harness() + try { + const handler = runtime.handlers.get('session.event.tool_result') + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: { + kind: 'user_input', + status, + paused: false, + request_id: 'input-request-1', + ...(status === 'answered' ? { answers: { scope: 'focused' } } : {}), + }, + }) + const appendCount = runtime.appendInterruptFrame.mock.calls.length + + handler?.({ + session_key: 'agent:main:web', + tool_use_id: 'request-input-1', + name: 'request_user_input', + result: planClarifyResult, + }) + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.appendInterruptFrame).toHaveBeenCalledTimes(appendCount) + expect(runtime.interruptState.value.get('input-request-1')?.resolution).toBe(resolution) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }, + ) + + it('does not retain a legacy receipt when its accepted task already terminated', async () => { + installSnapshot() + const runtime = await harness() + const submitted = deferred() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'legacy-clarify', + name: 'request_user_input', + result: { + ...clarifyResult, + request_id: undefined, + run_id: 'meta-run-1', + }, + }) + runtime.rpcCall.mockImplementationOnce(async () => await submitted.promise as T) + + const submission = runtime.approvals.submitClarify({ scope: 'focused' }) + await vi.waitFor(() => expect(runtime.rpcCall).toHaveBeenCalledTimes(1)) + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'accepted-continuation-task', + 'succeeded', + )).toBe(false) + + submitted.resolve({ task_id: 'accepted-continuation-task' }) + await submission + + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.get('meta-run-1|confirm_scope')?.resolution) + .toBe('replied') + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('retains a legacy clarify receipt until its accepted task terminates', async () => { + installSnapshot() + const runtime = await harness() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'legacy-clarify', + name: 'request_user_input', + result: { ...clarifyResult, request_id: undefined, }, }) + runtime.rpcCall.mockResolvedValueOnce({ task_id: 'accepted-continuation-task' }) await runtime.approvals.submitClarify({ scope: 'focused' }) expect(runtime.rpcCall).toHaveBeenLastCalledWith('chat.clarify_submit', { @@ -752,6 +1766,52 @@ describe('clarify tool-result recovery', () => { expect(runtime.approvals.pendingClarify.value?.requestId).toBeUndefined() expect(runtime.approvals.clarifySubmitted.value).toBe(true) expect(runtime.approvals.clarifyBusy.value).toBe(false) + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'plan-run-1', + 'timeout', + )).toBe(false) + expect(runtime.approvals.pendingClarify.value).not.toBeNull() + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'accepted-continuation-task', + 'succeeded', + )).toBe(true) + expect(runtime.approvals.pendingClarify.value).toBeNull() + expect(runtime.interruptState.value.get('plan-run-1|confirm_scope')?.resolution) + .toBe('replied') + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'accepted-continuation-task', + 'succeeded', + )).toBe(false) + } finally { + runtime.unsubscribe() + runtime.scope.stop() + } + }) + + it('uses the direct-mode turn id as the retained legacy owner', async () => { + installSnapshot() + const runtime = await harness() + try { + runtime.handlers.get('session.event.tool_result')?.({ + session_key: 'agent:main:web', + tool_use_id: 'legacy-clarify', + name: 'request_user_input', + result: { + ...clarifyResult, + request_id: undefined, + run_id: 'meta-run-direct', + }, + }) + runtime.rpcCall.mockResolvedValueOnce({ turn_id: 'direct-continuation-turn' }) + + await runtime.approvals.submitClarify({ scope: 'focused' }) + + expect(runtime.approvals.pendingClarify.value).not.toBeNull() + expect(runtime.approvals.settlePendingClarifyForTerminalTask( + 'direct-continuation-turn', + 'succeeded', + )).toBe(true) + expect(runtime.approvals.pendingClarify.value).toBeNull() } finally { runtime.unsubscribe() runtime.scope.stop() diff --git a/opensquilla-webui/src/composables/chat/useChatApprovals.source.test.ts b/opensquilla-webui/src/composables/chat/useChatApprovals.source.test.ts index e023fbd63..402a66c63 100644 --- a/opensquilla-webui/src/composables/chat/useChatApprovals.source.test.ts +++ b/opensquilla-webui/src/composables/chat/useChatApprovals.source.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest' import source from './useChatApprovals.ts?raw' +import chatViewSource from '@/views/ChatView.vue?raw' describe('useChatApprovals clarify submit source contract', () => { it('can submit a recovered inline clarify request without pendingClarify', () => { @@ -10,10 +11,34 @@ describe('useChatApprovals clarify submit source contract', () => { expect(source).toContain('if (request.runId) params.run_id = request.runId') }) - it('optimistically acknowledges the click before the backend finishes', () => { - expect(source).toContain('clarifySubmitted.value = true') - expect(source).toContain("setInterruptState(key, { resolution: 'replied', busy: true, error: '' })") + it('keeps the request busy until the backend acknowledges it', () => { + const awaitAck = source.indexOf('await conversation.submitClarify(params)') + const pendingState = source.indexOf( + "setInterruptState(key, { resolution: null, busy: true, error: '' })", + ) + const repliedState = source.indexOf( + "setInterruptState(key, { resolution: 'replied', busy: false })", + ) + + expect(pendingState).toBeGreaterThan(-1) + expect(pendingState).toBeLessThan(awaitAck) + expect(repliedState).toBeGreaterThan(awaitAck) expect(source).toContain('clarifySubmitted.value = false') expect(source).toContain('setInterruptState(key, { resolution: null, busy: false, error: message })') }) + + it('routes live, reconnect, and history terminal owners through one settlement path', () => { + expect(chatViewSource).toContain( + 'const terminalTask = terminalTaskFromRunState(snapshot)', + ) + expect(chatViewSource).toContain( + 'if (terminalStatus) settleTaskTerminalPresentation(taskId, terminalStatus)', + ) + expect(chatViewSource).toContain( + 'settleTaskTerminalPresentation(terminalTask.taskId, terminalTask.status)', + ) + expect(chatViewSource).toContain( + 'settleTaskTerminalPresentation(taskId, status)', + ) + }) }) diff --git a/opensquilla-webui/src/composables/chat/useChatApprovals.ts b/opensquilla-webui/src/composables/chat/useChatApprovals.ts index a7aa41300..ca463de29 100644 --- a/opensquilla-webui/src/composables/chat/useChatApprovals.ts +++ b/opensquilla-webui/src/composables/chat/useChatApprovals.ts @@ -7,7 +7,12 @@ import type { InterruptViewState, } from '@/types/parts' import { clarifyRequestFromValue, userInputOutcomeFromValue } from '@/utils/chat/clarify' -import { isCurrentSessionPayload } from '@/utils/chat/streamEvents' +import { + conversationCursorSignal, + isCurrentSessionPayload, + isStaleEpoch, + type TaskTerminalStatus, +} from '@/utils/chat/streamEvents' import type { ApprovalCenter, ApprovalAvailability, @@ -81,6 +86,12 @@ export interface ChatClarifyRequest { step: string } +interface ActiveClarifyRequest { + request: ChatClarifyRequest + observedStreamSeq?: number + observedStreamGeneration?: string +} + interface ApprovalResolveResponse { approved?: boolean resolved?: boolean @@ -115,6 +126,12 @@ export interface UseChatApprovalsOptions { sessionConversation: SessionConversation approvalCenter: ApprovalCenter sessionKey: Ref + /** Current session epoch, used to reject replay from a retired reset. */ + currentEpoch?: Readonly> + /** Current transport cursor namespace, used to order reconnect hydration. */ + streamGeneration?: Readonly> + /** Atomically adopt a new transport namespace before appending its frame. */ + observeStreamGeneration?: (source: unknown) => boolean runStatus: Ref /** The live-turn stream surface that hosts interrupt frames. */ stream: ApprovalsStreamSurface @@ -208,11 +225,96 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { const clarifySubmitted = ref(false) const clarifyBusy = ref(false) const clarifyError = ref('') - // A request-scoped submit can cross the Gateway boundary even when its RPC - // acknowledgement is lost. Keep that uncertainty until either the submit - // response/tool outcome settles it or an authoritative reconnect snapshot - // confirms that the request is no longer pending. - const clarifySubmitAttempts = new Set() + // The dock presents one questionnaire at a time, but a task may have more + // than one paused tool call represented by inline frames. Track every known + // unresolved request so task settlement and reconnect reconciliation cannot + // leave an older frame actionable. + const activeClarifyRequests = new Map() + // A legacy Meta clarify resumes through a new chat.send turn. Its run_id is + // a Meta-run identifier, not a TaskRuntime task id, so retain the exact task + // returned by the successful RPC instead of comparing those ID domains. + const retainedClarifyTaskOwners = new Map() + // A very short continuation can terminate before its chat.send acceptance + // response reaches this client. Retain a bounded terminal ledger so the late + // ACK cannot install a stale retained receipt. + const terminalClarifyTaskIds = new Set() + let acceptedClarifyStreamGeneration = String( + options.streamGeneration?.value || '', + ).trim() + const retiredClarifyStreamGenerations = new Set() + + function rememberTerminalClarifyTask(taskId: string) { + terminalClarifyTaskIds.delete(taskId) + terminalClarifyTaskIds.add(taskId) + if (terminalClarifyTaskIds.size <= 128) return + const oldest = terminalClarifyTaskIds.values().next().value + if (oldest) terminalClarifyTaskIds.delete(oldest) + } + + function rejectTerminalClarifyRequest( + key: string, + request: ChatClarifyRequest, + ): boolean { + if (!request.requestId || !terminalClarifyTaskIds.has(request.runId)) return false + activeClarifyRequests.delete(key) + if ( + interruptState.value.has(key) + && interruptState.value.get(key)?.resolution !== 'replied' + ) { + setInterruptState(key, { resolution: 'unavailable', busy: false, error: '' }) + } + clearPendingClarify(key) + return true + } + + function retireClarifyStreamGeneration(generation: string) { + if (!generation) return + retiredClarifyStreamGenerations.delete(generation) + retiredClarifyStreamGenerations.add(generation) + if (retiredClarifyStreamGenerations.size <= 16) return + const oldest = retiredClarifyStreamGenerations.values().next().value + if (oldest) retiredClarifyStreamGenerations.delete(oldest) + } + + function syncClarifyStreamGenerationFromShared() { + const shared = String(options.streamGeneration?.value || '').trim() + if ( + !shared + || shared === acceptedClarifyStreamGeneration + || retiredClarifyStreamGenerations.has(shared) + ) return + retireClarifyStreamGeneration(acceptedClarifyStreamGeneration) + acceptedClarifyStreamGeneration = shared + } + + function acceptsClarifyStreamGeneration(source: unknown): boolean { + const incoming = String( + conversationCursorSignal(source).streamGeneration || '', + ).trim() + if (!incoming) return true + + // Exact RPC listeners run before the wildcard lane advances the shared + // cursor. Treat the first unseen generation as the new namespace here so + // its first questionnaire is not dropped, while remembering the prior + // namespace so a late replay can never roll this ingress backwards. + syncClarifyStreamGenerationFromShared() + if (incoming === acceptedClarifyStreamGeneration) return true + if (retiredClarifyStreamGenerations.has(incoming)) return false + retireClarifyStreamGeneration(acceptedClarifyStreamGeneration) + acceptedClarifyStreamGeneration = incoming + // RpcClient dispatches exact listeners before the wildcard conversation + // lane. Advance/reset the shared cursor now so that later wildcard handling + // cannot clear the interrupt frame we are about to append. + const observedSource = source && typeof source === 'object' + ? { + ...(source as Record), + stream_generation: incoming, + streamGeneration: incoming, + } + : { streamGeneration: incoming } + options.observeStreamGeneration?.(observedSource) + return true + } // Resolution view-state for inline interrupt parts is the shared `interruptState` // ref (keyed by approval id, or the clarify composite key). The fold reads it to @@ -250,6 +352,81 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { clarifyError.value = '' } + function streamSeqFrom(value: Record): number | undefined { + const raw = value.stream_seq ?? value.streamSeq + if (raw === null || raw === undefined || raw === '' || typeof raw === 'boolean') { + return undefined + } + const sequence = Number(raw) + return Number.isSafeInteger(sequence) && sequence >= 0 ? sequence : undefined + } + + function streamGenerationFrom(value: Record): string | undefined { + const explicit = String(value.stream_generation ?? value.streamGeneration ?? '').trim() + if (explicit) return explicit + return acceptedClarifyStreamGeneration + || String(options.streamGeneration?.value || '').trim() + || undefined + } + + function rememberActiveClarify( + key: string, + request: ChatClarifyRequest, + observedStreamSeq?: number, + observedStreamGeneration?: string, + ) { + const prior = activeClarifyRequests.get(key) + const priorSequence = prior?.observedStreamSeq + const priorGeneration = prior?.observedStreamGeneration + const currentGeneration = acceptedClarifyStreamGeneration + || String(options.streamGeneration?.value || '').trim() + const priorIsCurrent = Boolean( + currentGeneration && priorGeneration === currentGeneration, + ) + const incomingIsCurrent = Boolean( + currentGeneration && observedStreamGeneration === currentGeneration, + ) + const sameGeneration = !priorGeneration + || !observedStreamGeneration + || priorGeneration === observedStreamGeneration + const keepPrior = Boolean(prior) && ( + (priorIsCurrent && observedStreamGeneration && !incomingIsCurrent) + || ( + sameGeneration + && priorSequence !== undefined + && observedStreamSeq !== undefined + && priorSequence > observedStreamSeq + ) + ) + const nextGeneration = keepPrior + ? priorGeneration + : observedStreamGeneration ?? priorGeneration + const nextSequence = keepPrior + ? priorSequence + : sameGeneration + ? priorSequence === undefined + ? observedStreamSeq + : observedStreamSeq === undefined + ? priorSequence + : Math.max(priorSequence, observedStreamSeq) + : observedStreamSeq + const active: ActiveClarifyRequest = { + request: keepPrior && prior ? prior.request : request, + ...(nextSequence !== undefined ? { observedStreamSeq: nextSequence } : {}), + ...(nextGeneration ? { observedStreamGeneration: nextGeneration } : {}), + } + activeClarifyRequests.set(key, active) + return active + } + + function presentPendingClarify(request: ChatClarifyRequest, key: string) { + pendingClarify.value = request + const state = interruptState.value.get(key) + clarifySubmitted.value = state?.resolution === 'replied' + clarifyBusy.value = state?.busy === true + clarifyError.value = state?.error || '' + } + function pendingClarifyMatches(key: string): boolean { return pendingClarify.value != null && clarifyFrameKey(pendingClarify.value) === key } @@ -566,11 +743,21 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { function handleToolResult(payload: ToolResultPayload) { if (!payload || typeof payload !== 'object') return if (!isCurrentSessionPayload(payload, sessionKey.value)) return + if ( + options.currentEpoch + && isStaleEpoch(payload, options.currentEpoch.value) + ) return + if (!acceptsClarifyStreamGeneration(payload)) return const outcome = userInputOutcomeFromValue(payload.result) if (outcome) { - clarifySubmitAttempts.delete(outcome.requestId) + activeClarifyRequests.delete(outcome.requestId) + const priorResolution = interruptState.value.get(outcome.requestId)?.resolution setInterruptState(outcome.requestId, { - resolution: 'replied', + // A positive acknowledgement dominates later expiry/cancellation + // replays, while a late authoritative answer may upgrade unavailable. + resolution: outcome.status === 'answered' || priorResolution === 'replied' + ? 'replied' + : 'unavailable', busy: false, error: '', }) @@ -580,12 +767,23 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { const request = parseClarifyRequest(payload) if (!request) return const key = clarifyFrameKey(request) + // Terminal delivery can overtake a previously unseen paused result. The + // task ledger is authoritative even when no clarify frame existed yet. + if (rejectTerminalClarifyRequest(key, request)) return // Tool-result replay and reconnect delivery can surface the paused half - // after its terminal outcome. Never resurrect an already-settled request or - // let a duplicate paused event undo optimistic submit feedback. - if (interruptState.value.get(key)?.resolution === 'replied') return - pendingClarify.value = request - resetClarifyPresentation() + // after its terminal outcome. Never resurrect an already-settled request. + if (interruptState.value.get(key)?.resolution) return + const payloadRecord = payload as Record + const active = rememberActiveClarify( + key, + request, + streamSeqFrom(payloadRecord), + streamGenerationFrom(payloadRecord), + ) + // A duplicate paused result can race an in-flight submit. Re-select the + // request from its own state so switching between multiple forms cannot + // erase a request-scoped busy/error fence. + presentPendingClarify(active.request, key) // Mirror the clarify into the turn log so it folds into an inline interrupt // part. The clarify keeps no approval id, so the runId|step composite keys it. const clarifyData: InterruptClarifyData = { @@ -705,49 +903,72 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { requestOverride?: ChatClarifyRequest, ) { const request = requestOverride || pendingClarify.value - if (clarifyBusy.value || !request) return + if (!request) return const key = clarifyFrameKey(request) - if (interruptState.value.get(key)?.resolution === 'replied') return + const currentState = interruptState.value.get(key) + if (currentState?.resolution || currentState?.busy) return if (!requestOverride && clarifySubmitted.value) return + const submittedSessionKey = sessionKey.value const controlsPendingPresentation = pendingClarifyMatches(key) if (controlsPendingPresentation) { clarifyBusy.value = true - clarifySubmitted.value = true + clarifySubmitted.value = false clarifyError.value = '' } - if (request.requestId) clarifySubmitAttempts.add(key) - setInterruptState(key, { resolution: 'replied', busy: true, error: '' }) + rememberActiveClarify(key, request, undefined, streamGenerationFrom({})) + setInterruptState(key, { resolution: null, busy: true, error: '' }) const params: Record = { sessionKey: sessionKey.value, fields } if (request.requestId) params.request_id = request.requestId if (request.runId) params.run_id = request.runId try { - await conversation.submitClarify(params) - clarifySubmitAttempts.delete(key) + const response = await conversation.submitClarify(params) + if (submittedSessionKey !== sessionKey.value) return + activeClarifyRequests.delete(key) + let legacyOwnerAlreadyTerminal = false + if (!request.requestId) { + const acceptedTaskId = String( + response.task_id + ?? response.taskId + ?? response.turn_id + ?? response.turnId + ?? '', + ).trim() + legacyOwnerAlreadyTerminal = Boolean( + acceptedTaskId && terminalClarifyTaskIds.has(acceptedTaskId), + ) + if (acceptedTaskId && !legacyOwnerAlreadyTerminal) { + retainedClarifyTaskOwners.set(key, acceptedTaskId) + } + } setInterruptState(key, { resolution: 'replied', busy: false }) + if (pendingClarifyMatches(key)) clarifySubmitted.value = true // request_id submissions resolve the exact paused tool call in the same // turn. A successful RPC is therefore authoritative and can release the // dock/composer immediately. Legacy clarifications create a new chat turn // and intentionally retain their existing submitted receipt. - if (request.requestId) clearPendingClarify(key) + if (request.requestId || legacyOwnerAlreadyTerminal) clearPendingClarify(key) } catch (err) { + if (submittedSessionKey !== sessionKey.value) return const message = 'Send failed — ' + (err instanceof Error ? err.message : String(err)) const stillPending = pendingClarifyMatches(key) // A terminal tool result or authoritative empty snapshot can win the // race with a rejected/lost RPC acknowledgement. Never reopen that // already-settled request from the late rejection. const terminalConfirmed = !stillPending - && interruptState.value.get(key)?.resolution === 'replied' + && interruptState.value.get(key)?.resolution != null if (stillPending) { clarifySubmitted.value = false clarifyError.value = message } if (terminalConfirmed) { - clarifySubmitAttempts.delete(key) + activeClarifyRequests.delete(key) } else { setInterruptState(key, { resolution: null, busy: false, error: message }) } } finally { - if (pendingClarifyMatches(key)) clarifyBusy.value = false + if (submittedSessionKey === sessionKey.value && pendingClarifyMatches(key)) { + clarifyBusy.value = false + } } } @@ -756,40 +977,127 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { resetClarifyPresentation() } + /** Settle only the structured input owned by the authoritative terminal task. */ + function settlePendingClarifyForTerminalTask( + taskId: string, + _taskStatus: TaskTerminalStatus, + ) { + if (!taskId) return false + rememberTerminalClarifyTask(taskId) + let settled = false + for (const [key, active] of activeClarifyRequests) { + // Broker-owned structured requests stamp run_id with their TaskRuntime + // owner. Legacy Meta run ids use a different identity domain and are + // correlated only after their continuation RPC returns a task id below. + if (!active.request.requestId) continue + if (active.request.runId !== taskId) continue + activeClarifyRequests.delete(key) + if (interruptState.value.get(key)?.resolution !== 'replied') { + setInterruptState(key, { resolution: 'unavailable', busy: false, error: '' }) + } + clearPendingClarify(key) + settled = true + } + for (const [key, ownerTaskId] of retainedClarifyTaskOwners) { + if (ownerTaskId !== taskId) continue + retainedClarifyTaskOwners.delete(key) + if (interruptState.value.get(key)?.resolution !== 'replied') { + setInterruptState(key, { resolution: 'unavailable', busy: false, error: '' }) + } + activeClarifyRequests.delete(key) + clearPendingClarify(key) + settled = true + } + return settled + } + function applyUserInputBootstrap(snapshot: { - pendingUserInputs?: unknown[] - pending_user_inputs?: unknown[] + pendingUserInputs?: readonly unknown[] + pending_user_inputs?: readonly unknown[] + goalSnapshotStreamSeq?: number | null + goal_snapshot_stream_seq?: number | null + streamGeneration?: string + stream_generation?: string + deferredFields?: readonly string[] + deferred_fields?: readonly string[] }) { const hasAuthoritativePendingList = Object.prototype.hasOwnProperty.call( snapshot, 'pendingUserInputs', ) || Object.prototype.hasOwnProperty.call(snapshot, 'pending_user_inputs') if (!hasAuthoritativePendingList) return + const deferred = snapshot.deferredFields ?? snapshot.deferred_fields + if (Array.isArray(deferred) && deferred.some(field => ( + field === 'pendingUserInputs' || field === 'pending_user_inputs' + ))) return + // Reject a retired namespace before either negative reconciliation or + // positive additions can mutate the current dock/inline state. + if (!acceptsClarifyStreamGeneration(snapshot)) return const pending = snapshot.pendingUserInputs || snapshot.pending_user_inputs || [] const requests = pending .map(value => clarifyRequestFromValue(value)) .filter((request): request is ChatClarifyRequest => request != null) const pendingKeys = new Set(requests.map(request => clarifyFrameKey(request))) - const current = pendingClarify.value - if (current?.requestId) { - const currentKey = clarifyFrameKey(current) - if (clarifySubmitAttempts.has(currentKey) && !pendingKeys.has(currentKey)) { - clarifySubmitAttempts.delete(currentKey) - setInterruptState(currentKey, { resolution: 'replied', busy: false, error: '' }) - clearPendingClarify(currentKey) - } + const snapshotStreamSeq = streamSeqFrom({ + streamSeq: snapshot.goalSnapshotStreamSeq ?? snapshot.goal_snapshot_stream_seq, + }) + const snapshotStreamGeneration = streamGenerationFrom(snapshot) + for (const [key, active] of activeClarifyRequests) { + if (pendingKeys.has(key)) continue + // pendingUserInputs is authoritative only for broker-owned structured + // requests. Legacy Meta clarifies are resumed by a follow-up chat turn + // and never appear in this snapshot. + if (!active.request.requestId) continue + // Hydration captures this cursor before reading pending inputs. A live + // request observed after that cursor is newer than an absent entry in + // this snapshot and must survive until an equal/newer snapshot arrives. + const activeGeneration = active.observedStreamGeneration + const currentGeneration = acceptedClarifyStreamGeneration + || String(options.streamGeneration?.value || '').trim() + if (activeGeneration && snapshotStreamGeneration && activeGeneration !== snapshotStreamGeneration) { + // Only the subscription's current generation may supersede state from + // another cursor namespace. An old hydrate that arrives after a new + // live event cannot compare its numeric sequence to that event. + if ( + !currentGeneration + || activeGeneration === currentGeneration + || snapshotStreamGeneration !== currentGeneration + ) continue + } else if ( + snapshotStreamSeq !== undefined + && active.observedStreamSeq !== undefined + && active.observedStreamSeq > snapshotStreamSeq + ) continue + activeClarifyRequests.delete(key) + const priorResolution = interruptState.value.get(key)?.resolution + setInterruptState(key, { + resolution: priorResolution === 'replied' ? 'replied' : 'unavailable', + busy: false, + error: '', + }) + clearPendingClarify(key) } for (const request of requests) { const key = clarifyFrameKey(request) - if (interruptState.value.get(key)?.resolution === 'replied') continue - const sameRequest = pendingClarifyMatches(key) - pendingClarify.value = request + // A hydrate captured before terminal settlement may complete after it. + // Never let that late positive snapshot resurrect the questionnaire. + if (rejectTerminalClarifyRequest(key, request)) continue + if (interruptState.value.get(key)?.resolution) { + activeClarifyRequests.delete(key) + continue + } + const active = rememberActiveClarify( + key, + request, + snapshotStreamSeq, + snapshotStreamGeneration, + ) + if (!interruptState.value.has(key)) setInterruptState(key, {}) // Do not make an in-flight submission actionable again just because a // racing snapshot still contains its pre-submit pending record. - if (!sameRequest || !clarifyBusy.value) resetClarifyPresentation() - if (!interruptState.value.has(key)) setInterruptState(key, {}) + presentPendingClarify(active.request, key) if (!stream.isStreaming.value) stream.ensureInterruptBubble() stream.appendInterruptFrame({ interruptKind: 'clarify', @@ -811,7 +1119,12 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { interruptState.value = new Map() interruptNamespaces.clear() interruptApprovals.clear() - clarifySubmitAttempts.clear() + activeClarifyRequests.clear() + retainedClarifyTaskOwners.clear() + terminalClarifyTaskIds.clear() + // Stream generations belong to the Gateway transport, not one session; + // keep retired namespaces fenced across navigation. + syncClarifyStreamGenerationFromShared() legacyPushBackfills.clear() dismissClarify() if (key) hydrateApprovals() @@ -835,6 +1148,7 @@ export function useChatApprovals(options: UseChatApprovalsOptions) { extendInterrupt, submitClarify, dismissClarify, + settlePendingClarifyForTerminalTask, applyUserInputBootstrap, subscribe, cleanup, diff --git a/opensquilla-webui/src/composables/chat/useChatHistory.test.ts b/opensquilla-webui/src/composables/chat/useChatHistory.test.ts index 32d4a8f0e..c371ab618 100644 --- a/opensquilla-webui/src/composables/chat/useChatHistory.test.ts +++ b/opensquilla-webui/src/composables/chat/useChatHistory.test.ts @@ -2830,6 +2830,30 @@ describe('useChatHistory optimistic local rows', () => { })) }) + it('notifies terminal direct turns that have no TaskRuntime id', async () => { + const onTerminalTask = vi.fn() + const { api } = makeHistory(true, { + onTerminalTask, + response: { + messages: [], + turnOutcomes: [{ + turnId: 'direct-terminal-turn', + status: 'succeeded', + finishedAt: 2_000, + }], + hasMore: false, + canonicalComplete: true, + }, + }) + + await api.loadHistory() + + expect(onTerminalTask).toHaveBeenCalledWith(expect.objectContaining({ + turnId: 'direct-terminal-turn', + status: 'succeeded', + })) + }) + it('restores usage barrier activity and its retryable error from terminal history', async () => { const { api, messages } = makeHistory(true, { response: { diff --git a/opensquilla-webui/src/composables/chat/useChatHistory.ts b/opensquilla-webui/src/composables/chat/useChatHistory.ts index d0f05a15b..cf32d0bc1 100644 --- a/opensquilla-webui/src/composables/chat/useChatHistory.ts +++ b/opensquilla-webui/src/composables/chat/useChatHistory.ts @@ -764,7 +764,7 @@ export function useChatHistory(options: UseChatHistoryOptions) { for (const raw of data.turnOutcomes) { const outcome = normalizeTurnOutcome(turnOutcomeRecord(raw)) if ( - outcome?.taskId + (outcome?.taskId || outcome?.turnId) && ['succeeded', 'failed', 'cancelled', 'timeout', 'abandoned', 'interrupted'] .includes(outcome.status.toLowerCase()) ) { diff --git a/opensquilla-webui/src/composables/chat/useChatPlans.test.ts b/opensquilla-webui/src/composables/chat/useChatPlans.test.ts index 7ef88fd56..d33730fcc 100644 --- a/opensquilla-webui/src/composables/chat/useChatPlans.test.ts +++ b/opensquilla-webui/src/composables/chat/useChatPlans.test.ts @@ -647,6 +647,238 @@ describe('useChatPlans', () => { expect(api.activePlanRun.value?.status).toBe('cancelled') }) + it.each([ + ['cancelled', 'cancelled_by_user'], + ['timeout', 'timeout'], + ['failed', 'failed'], + ['abandoned', 'abandoned'], + ['interrupted', 'interrupted'], + ] as const)( + 'releases the visible run owner when its task becomes %s', + (taskStatus, pauseReason) => { + const { api } = harness() + api.applyBootstrap({ + key: SESSION_ONE, + currentPlan: revision(), + activePlanRun: run('running', { activeTaskId: 'task-terminal-1' }) as never, + }) + + expect(api.settleActiveRunForTerminalTask('task-terminal-1', taskStatus)).toBe(true) + expect(api.activePlanRun.value).toMatchObject({ + status: 'paused', + pauseReason, + currentStepId: 'inspect', + steps: [{ status: 'in_progress' }], + }) + expect(api.activePlanRun.value?.activeTaskId).toBeUndefined() + }, + ) + + it('ignores another task terminal and keeps repeated settlement idempotent', () => { + const { api } = harness() + api.applyBootstrap({ + key: SESSION_ONE, + currentPlan: revision(), + activePlanRun: run('running', { activeTaskId: 'task-owner' }) as never, + }) + + expect(api.settleActiveRunForTerminalTask('task-other', 'timeout')).toBe(false) + expect(api.activePlanRun.value?.status).toBe('running') + expect(api.settleActiveRunForTerminalTask('task-owner', 'failed')).toBe(true) + const settled = api.activePlanRun.value + expect(api.settleActiveRunForTerminalTask('task-owner', 'cancelled')).toBe(false) + expect(api.activePlanRun.value).toBe(settled) + }) + + it('rejects the settled task owner, then adopts authoritative pause and resume', () => { + const { api, handlers } = harness() + api.applyBootstrap({ + key: SESSION_ONE, + currentPlan: revision(), + activePlanRun: run('running', { + activeTaskId: 'task-terminal', + stateRevision: 4, + }) as never, + }) + api.subscribe() + expect(api.settleActiveRunForTerminalTask('task-terminal', 'timeout')).toBe(true) + expect(api.planRunSettlementPending.value).toBe(true) + + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('running', { + activeTaskId: 'task-terminal', + stateRevision: 99, + updatedAt: 999, + }), + }) + expect(api.activePlanRun.value).toMatchObject({ + status: 'paused', + pauseReason: 'timeout', + stateRevision: 4, + }) + + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('paused', { + activeTaskId: undefined, + pauseReason: 'manual_turn_timed_out', + stateRevision: 5, + updatedAt: 500, + }), + }) + expect(api.activePlanRun.value).toMatchObject({ + status: 'paused', + pauseReason: 'manual_turn_timed_out', + stateRevision: 5, + }) + expect(api.planRunSettlementPending.value).toBe(false) + + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('queued', { + activeTaskId: 'task-resumed', + pauseReason: undefined, + stateRevision: 6, + updatedAt: 600, + }), + }) + expect(api.activePlanRun.value).toMatchObject({ + status: 'queued', + activeTaskId: 'task-resumed', + stateRevision: 6, + }) + }) + + it('keeps a lagging queued projection provisional until backend pause and resume', () => { + const { api, handlers } = harness() + api.applyBootstrap({ + key: SESSION_ONE, + currentPlan: revision(), + activePlanRun: run('queued', { + activeTaskId: 'task-terminal', + stateRevision: 4, + }) as never, + }) + api.subscribe() + + expect(api.settleActiveRunForTerminalTask('task-terminal', 'failed')).toBe(true) + expect(api.planRunSettlementPending.value).toBe(true) + expect(api.activePlanRun.value).toMatchObject({ + status: 'paused', + pauseReason: 'failed', + stateRevision: 4, + }) + + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('running', { + activeTaskId: 'task-terminal', + stateRevision: 99, + updatedAt: 999, + }), + }) + expect(api.activePlanRun.value?.stateRevision).toBe(4) + + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('paused', { + activeTaskId: undefined, + pauseReason: 'manual_turn_failed', + stateRevision: 5, + updatedAt: 500, + }), + }) + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('queued', { + activeTaskId: 'task-resumed', + pauseReason: undefined, + stateRevision: 6, + updatedAt: 600, + }), + }) + + expect(api.activePlanRun.value).toMatchObject({ + status: 'queued', + activeTaskId: 'task-resumed', + stateRevision: 6, + }) + expect(api.planRunSettlementPending.value).toBe(false) + }) + + it('lets an authoritative cancellation replace a provisional queued settlement', () => { + const { api, handlers } = harness() + api.applyBootstrap({ + key: SESSION_ONE, + currentPlan: revision(), + activePlanRun: run('queued', { + activeTaskId: 'task-terminal', + stateRevision: 4, + }) as never, + }) + api.subscribe() + + expect(api.settleActiveRunForTerminalTask('task-terminal', 'cancelled')).toBe(true) + expect(api.planRunSettlementPending.value).toBe(true) + expect(api.activePlanRun.value?.status).toBe('paused') + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('cancelled', { + activeTaskId: undefined, + terminalReason: 'implementation_turn_ended_before_start', + stateRevision: 5, + updatedAt: 500, + }), + }) + + expect(api.activePlanRun.value).toMatchObject({ + status: 'cancelled', + terminalReason: 'implementation_turn_ended_before_start', + stateRevision: 5, + }) + expect(api.planRunSettlementPending.value).toBe(false) + }) + + it('blocks Plan mutations until authoritative run settlement arrives', async () => { + const { api, handlers, rpc } = harness() + const target = { planId: 'plan-1', revisionId: 'revision-2' } + api.applyBootstrap({ + key: SESSION_ONE, + currentPlan: revision(), + activePlanRun: run('running', { + activeTaskId: 'task-terminal', + stateRevision: 4, + }) as never, + }) + api.subscribe() + + expect(api.settleActiveRunForTerminalTask('task-terminal', 'failed')).toBe(true) + await api.implement(target, false) + await api.cancelRun() + expect(rpc.call).not.toHaveBeenCalled() + + handlers.get('session.event.plan_run')?.({ + session_key: SESSION_ONE, + plan_run: run('paused', { + activeTaskId: undefined, + pauseReason: 'manual_turn_failed', + stateRevision: 5, + updatedAt: 500, + }), + }) + rpc.call.mockResolvedValueOnce({ planRun: run('queued', { stateRevision: 6 }) }) + await api.implement(target, false) + + expect(rpc.call).toHaveBeenCalledWith( + 'plans.implement', + expect.objectContaining({ + sessionKey: SESSION_ONE, + planRevisionId: 'revision-2', + }), + ) + }) + it('keeps a newer epoch cancellation locked when the old cancellation returns late', async () => { const { api, currentEpoch, rpc } = harness() api.applyBootstrap({ diff --git a/opensquilla-webui/src/composables/chat/useChatPlans.ts b/opensquilla-webui/src/composables/chat/useChatPlans.ts index f443ebce5..28c654d39 100644 --- a/opensquilla-webui/src/composables/chat/useChatPlans.ts +++ b/opensquilla-webui/src/composables/chat/useChatPlans.ts @@ -10,6 +10,7 @@ import type { } from '@/types/plans' import type { PlanCenter } from '@/modules/planCenter' import { createClientRequestId } from '@/utils/chat/messageIdentity' +import type { TaskSettlementStatus } from '@/utils/chat/streamEvents' import { normalizeCollaborationSnapshot, normalizePlanRevisionSnapshot, @@ -174,12 +175,14 @@ export function useChatPlans(options: UseChatPlansOptions) { const pendingAction = ref(null) const modeAppliesNextTurn = ref(false) const replanTarget = ref(null) + const planRunSettlementPending = ref(false) const currentPlanRevisionId = computed(() => currentPlan.value?.revisionId || '') const replanActive = computed(() => replanTarget.value !== null) let acceptedEpoch = 0 let modeMutationOwner: symbol | null = null let actionMutationOwner: symbol | null = null + let settledTaskFence: { runId: string; taskId: string } | null = null function clearPlanState() { // Reset/session changes invalidate in-flight UI mutations. Their delayed @@ -189,6 +192,8 @@ export function useChatPlans(options: UseChatPlansOptions) { collaboration.value = { mode: 'default', revision: 0 } currentPlan.value = null activePlanRun.value = null + settledTaskFence = null + planRunSettlementPending.value = false modeBusy.value = false pendingAction.value = null modeAppliesNextTurn.value = false @@ -265,6 +270,8 @@ export function useChatPlans(options: UseChatPlansOptions) { && activePlanRun.value.planRevisionId !== plan.revisionId ) { activePlanRun.value = null + settledTaskFence = null + planRunSettlementPending.value = false } return true } @@ -275,9 +282,17 @@ export function useChatPlans(options: UseChatPlansOptions) { !run || !currentPlan.value || run.planRevisionId !== currentPlan.value.revisionId + || ( + settledTaskFence?.runId === run.runId + && settledTaskFence.taskId === run.activeTaskId + ) || !shouldAdoptPlanRun(run, activePlanRun.value) ) return false activePlanRun.value = run + if (settledTaskFence) { + settledTaskFence = null + planRunSettlementPending.value = false + } return true } @@ -303,6 +318,8 @@ export function useChatPlans(options: UseChatPlansOptions) { } else if (!staleEnvelope) { currentPlan.value = null activePlanRun.value = null + settledTaskFence = null + planRunSettlementPending.value = false } } const rawRun = source.activePlanRun @@ -317,6 +334,8 @@ export function useChatPlans(options: UseChatPlansOptions) { } } else if (!staleEnvelope) { activePlanRun.value = null + settledTaskFence = null + planRunSettlementPending.value = false } } } @@ -423,7 +442,12 @@ export function useChatPlans(options: UseChatPlansOptions) { } async function revise(request: PlanRevisionRequest): Promise { - if (!options.sessionKey.value || modeBusy.value || pendingAction.value) return false + if ( + !options.sessionKey.value + || modeBusy.value + || pendingAction.value + || planRunSettlementPending.value + ) return false const prompt = request.prompt.trim() if (!prompt) return false const key = options.sessionKey.value @@ -460,7 +484,12 @@ export function useChatPlans(options: UseChatPlansOptions) { } async function implement(target: PlanCardActionTarget, inNewSession: boolean) { - if (!options.sessionKey.value || modeBusy.value || pendingAction.value) return + if ( + !options.sessionKey.value + || modeBusy.value + || pendingAction.value + || planRunSettlementPending.value + ) return const sourceKey = options.sessionKey.value const sourceEpoch = acceptedEpoch const targetKey = inNewSession @@ -502,7 +531,12 @@ export function useChatPlans(options: UseChatPlansOptions) { async function cancelRun() { const run = activePlanRun.value - if (!run || modeBusy.value || pendingAction.value) return + if ( + !run + || modeBusy.value + || pendingAction.value + || planRunSettlementPending.value + ) return const key = options.sessionKey.value const epoch = acceptedEpoch const owner = Symbol('plan-action-mutation') @@ -529,6 +563,33 @@ export function useChatPlans(options: UseChatPlansOptions) { } } + /** + * A task terminal arrives before TaskRuntime settles its attached PlanRun. + * Release the exact task owner immediately for presentation, but do not + * invent a terminal run: running implementations are persisted as resumable + * paused runs. The owner fence rejects delayed pre-terminal run events until + * the authoritative owner-free paused/cancelled snapshot replaces this + * transient projection. + */ + function settleActiveRunForTerminalTask( + taskId: string, + taskStatus: TaskSettlementStatus, + ) { + const run = activePlanRun.value + if (!run || !taskId || run.activeTaskId !== taskId) return false + if (!['queued', 'running', 'paused', 'blocked'].includes(run.status)) return false + const settlementReason = taskStatus === 'cancelled' ? 'cancelled_by_user' : taskStatus + settledTaskFence = { runId: run.runId, taskId } + planRunSettlementPending.value = true + activePlanRun.value = { + ...run, + status: 'paused', + activeTaskId: undefined, + pauseReason: settlementReason, + } + return true + } + reset() return { @@ -537,6 +598,7 @@ export function useChatPlans(options: UseChatPlansOptions) { currentPlan, currentPlanRevisionId, activePlanRun, + planRunSettlementPending, modeBusy, modeAppliesNextTurn, pendingAction, @@ -552,5 +614,6 @@ export function useChatPlans(options: UseChatPlansOptions) { revise, implement, cancelRun, + settleActiveRunForTerminalTask, } } diff --git a/opensquilla-webui/src/composables/chat/useChatRenderedMessages.test.ts b/opensquilla-webui/src/composables/chat/useChatRenderedMessages.test.ts index 36cc6c855..2f9451be6 100644 --- a/opensquilla-webui/src/composables/chat/useChatRenderedMessages.test.ts +++ b/opensquilla-webui/src/composables/chat/useChatRenderedMessages.test.ts @@ -2803,6 +2803,192 @@ describe('useChatRenderedMessages clarify history recovery', () => { expect(clarify?.clarify?.presentation).toBe('plan_questionnaire_v1') }) + it.each(['cancelled', 'expired'] as const)( + 'restores a %s request as unavailable from its preserved request payload', + (status) => { + const api = renderedMessagesFor([ + { + role: 'assistant', + text: '', + ts: 0, + messageId: `m-terminal-${status}-request-user-input`, + tool_calls: [ + { + type: 'tool_result', + tool_use_id: `request-input-${status}`, + name: 'request_user_input', + user_input_request: { + status: 'input_required', + kind: 'user_input', + paused: true, + request_id: `request-${status}-1`, + run_id: 'plan-run-2', + step: 'choose_target', + clarify_schema: { + mode: 'form', + presentation: 'plan_questionnaire_v1', + fields: [{ + name: 'target', + type: 'enum', + required: true, + choices: ['current', 'new'], + }], + }, + }, + result: JSON.stringify({ + status, + kind: 'user_input', + paused: false, + request_id: `request-${status}-1`, + }), + }, + ], + }, + ]) + + const [message] = api.renderedMessages.value + const clarify = message.parts?.find((part): part is ChatPart & { + type: 'interrupt' + interruptKind: 'clarify' + } => part.type === 'interrupt' && part.interruptKind === 'clarify') + + expect(clarify?.resolution).toBe('unavailable') + expect(clarify?.clarify?.requestId).toBe(`request-${status}-1`) + }, + ) + + it('keeps an answered historical request replied after a later expiry replay', () => { + const request = { + status: 'input_required', + kind: 'user_input', + paused: true, + request_id: 'request-terminal-replay', + run_id: 'plan-run-2', + step: 'choose_target', + clarify_schema: { + fields: [{ name: 'target', type: 'string' }], + }, + } + const outcome = (status: 'answered' | 'expired') => ({ + status, + kind: 'user_input', + paused: false, + request_id: 'request-terminal-replay', + }) + const api = renderedMessagesFor([{ + role: 'assistant', + text: '', + ts: 0, + messageId: 'm-terminal-replay', + tool_calls: [ + { + type: 'tool_result', + tool_use_id: 'request-input-terminal-replay', + name: 'request_user_input', + user_input_request: request, + result: outcome('answered'), + }, + { + type: 'tool_result', + tool_use_id: 'request-input-terminal-replay', + name: 'request_user_input', + result: outcome('expired'), + }, + ], + }]) + + const clarify = api.renderedMessages.value[0].parts?.find((part): part is ChatPart & { + type: 'interrupt' + interruptKind: 'clarify' + } => part.type === 'interrupt' && part.interruptKind === 'clarify') + expect(clarify?.resolution).toBe('replied') + }) + + it('expires only the unresolved historical clarify owned by an abnormal terminal task', () => { + const request = (requestId: string, runId: string) => ({ + status: 'input_required', + kind: 'user_input', + paused: true, + request_id: requestId, + run_id: runId, + step: 'choose_target', + clarify_schema: { + fields: [{ name: 'target', type: 'string' }], + }, + }) + const api = renderedMessagesFor([{ + role: 'assistant', + text: '', + ts: 0, + messageId: 'm-terminal-history', + turnId: 'task-terminal-history', + restoredFromHistory: true, + turnOutcome: { + turnId: 'task-terminal-history', + taskId: 'task-terminal-history', + status: 'timeout', + }, + tool_calls: [ + { + type: 'tool_result', + tool_use_id: 'request-terminal-owner', + result: request('request-terminal-owner', 'task-terminal-history'), + }, + { + type: 'tool_result', + tool_use_id: 'request-other-owner', + result: request('request-other-owner', 'task-other'), + }, + ], + }]) + + const clarifies = api.renderedMessages.value[0].parts + ?.filter((part): part is ChatPart & { + type: 'interrupt' + interruptKind: 'clarify' + } => part.type === 'interrupt' && part.interruptKind === 'clarify') + expect(clarifies?.find(part => part.clarify?.requestId === 'request-terminal-owner') + ?.resolution).toBe('unavailable') + expect(clarifies?.find(part => part.clarify?.requestId === 'request-other-owner') + ?.resolution).toBeNull() + }) + + it('expires an abnormal direct-turn clarify when history has no task id', () => { + const api = renderedMessagesFor([{ + role: 'assistant', + text: '', + ts: 0, + messageId: 'm-direct-terminal-history', + turnId: 'direct-terminal-turn', + restoredFromHistory: true, + turnOutcome: { + turnId: 'direct-terminal-turn', + status: 'timeout', + }, + tool_calls: [{ + type: 'tool_result', + tool_use_id: 'request-direct-terminal', + result: { + status: 'input_required', + kind: 'user_input', + paused: true, + request_id: 'request-direct-terminal', + run_id: 'direct-terminal-turn', + step: 'choose_target', + clarify_schema: { + fields: [{ name: 'target', type: 'string' }], + }, + }, + }], + }]) + + const clarify = api.renderedMessages.value[0].parts?.find((part): part is ChatPart & { + type: 'interrupt' + interruptKind: 'clarify' + } => part.type === 'interrupt' && part.interruptKind === 'clarify') + expect(clarify?.resolution).toBe('unavailable') + }) + it('keeps consecutive requests distinct by requestId', () => { const request = (requestId: string) => ({ status: 'input_required', diff --git a/opensquilla-webui/src/composables/chat/useChatRenderedMessages.ts b/opensquilla-webui/src/composables/chat/useChatRenderedMessages.ts index 980b1cf3f..3d5d73f3b 100644 --- a/opensquilla-webui/src/composables/chat/useChatRenderedMessages.ts +++ b/opensquilla-webui/src/composables/chat/useChatRenderedMessages.ts @@ -37,9 +37,10 @@ import { } from '@/utils/chat/routerTiers' import { normalizeRouterTierSnapshot } from '@/utils/chat/routerTierSnapshot' import { clarifyRequestFromValue, userInputOutcomeFromValue } from '@/utils/chat/clarify' +import { turnOutcomePresentation } from '@/utils/chat/turnOutcome' import type { RouterVisualMode } from '@/utils/chat/routerVisualMode' import type { ModelRoutingMode } from '@/types/modelRouting' -import type { InterruptViewState } from '@/types/parts' +import type { InterruptClarifyData, InterruptViewState } from '@/types/parts' import { toParts, toolState, type ToPartsInterrupt } from '@/utils/chat/toParts' import { toSources } from '@/utils/chat/toSources' import { createdSessionFromToolCall } from '@/utils/chat/createdSessions' @@ -129,7 +130,10 @@ function clarifyInterruptFromValue(value: unknown): ToPartsInterrupt | null { } } -function historicalClarifyInterrupts(segments: RawToolCallPayload[] | undefined): ToPartsInterrupt[] { +function historicalClarifyInterrupts( + segments: RawToolCallPayload[] | undefined, + terminalTaskId = '', +): ToPartsInterrupt[] { if (!Array.isArray(segments) || !segments.length) return [] const inputByToolId = new Map() const out: ToPartsInterrupt[] = [] @@ -143,10 +147,14 @@ function historicalClarifyInterrupts(segments: RawToolCallPayload[] | undefined) return } const existing = out[existingIndex] + const incomingResolution = interrupt.resolution === 'unavailable' + && existing.resolution === 'replied' + ? 'replied' + : interrupt.resolution out[existingIndex] = { ...existing, data: { ...existing.data, ...interrupt.data }, - ...(interrupt.resolution ? { resolution: interrupt.resolution } : {}), + ...(incomingResolution ? { resolution: incomingResolution } : {}), } as ToPartsInterrupt } @@ -168,16 +176,31 @@ function historicalClarifyInterrupts(segments: RawToolCallPayload[] | undefined) if (fromMatchingInput) { upsert({ ...fromMatchingInput, - ...(outcome ? { resolution: 'replied' } : {}), + ...(outcome + ? { resolution: outcome.status === 'answered' ? 'replied' : 'unavailable' } + : {}), }) } else if (outcome) { const existingIndex = indexByApprovalId.get(outcome.requestId) if (existingIndex != null) { - out[existingIndex] = { ...out[existingIndex], resolution: 'replied' } + const priorResolution = out[existingIndex].resolution + out[existingIndex] = { + ...out[existingIndex], + resolution: outcome.status === 'answered' || priorResolution === 'replied' + ? 'replied' + : 'unavailable', + } } } } - return out + if (!terminalTaskId) return out + return out.map(interrupt => ( + interrupt.kind === 'clarify' + && (interrupt.data as InterruptClarifyData).runId === terminalTaskId + && !interrupt.resolution + ? { ...interrupt, resolution: 'unavailable' } + : interrupt + )) } function terminatesPriorAssistant(message: ChatMessage, priorAssistant?: ChatMessage): boolean { @@ -512,6 +535,11 @@ export function useChatRenderedMessages(options: UseChatRenderedMessagesOptions) })) const isPlanMessage = msg.role === 'assistant' && planRevisions.length > 0 const normalizedToolCalls = normalizeToolCalls(msg.tool_calls) + const terminalClarifyTaskId = msg.turnOutcome + && (msg.turnOutcome.taskId || msg.turnOutcome.turnId) + && turnOutcomePresentation(msg.turnOutcome) !== 'completed' + ? msg.turnOutcome.taskId || msg.turnOutcome.turnId + : '' const assistantRawText = msg.role === 'assistant' ? options.stripGeneratedArtifactMarkers(msg.text) : msg.text @@ -590,7 +618,7 @@ export function useChatRenderedMessages(options: UseChatRenderedMessagesOptions) options.renderMarkdown, toolCallGroups, ownerKey, - historicalClarifyInterrupts(msg.tool_calls), + historicalClarifyInterrupts(msg.tool_calls, terminalClarifyTaskId), options.interruptState?.value, ) : [] diff --git a/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.test.ts b/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.test.ts index 363659158..33d195c1f 100644 --- a/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.test.ts +++ b/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.test.ts @@ -37,6 +37,7 @@ function createHarness(options: { getCompactionPlacement?: (compactionId: string) => 'activity' | 'standalone' | undefined observeStreamGeneration?: (payload: unknown) => boolean supportsTurnCommitted?: boolean + onTaskTerminal?: (taskId: string, status: string) => void } = {}) { const messages = ref(options.messages ?? []) const sessionKey = ref('agent:main:test') @@ -90,6 +91,7 @@ function createHarness(options: { const loadCurrentSessionUsage = vi.fn(options.loadCurrentSessionUsage ?? (() => {})) const refreshRunModePreference = vi.fn(options.refreshRunModePreference ?? (() => {})) const restoreSteerIntoComposer = vi.fn(options.restoreSteerIntoComposer ?? (() => {})) + const onTaskTerminal = vi.fn(options.onTaskTerminal ?? (() => {})) const scope = effectScope() const rawApi = scope.run(() => useChatRpcEventHandlers({ sessionKey, @@ -137,6 +139,7 @@ function createHarness(options: { handleSessionConnectionState, loadCurrentSessionUsage, refreshRunModePreference, + onTaskTerminal, }))! const api = { ...rawApi, @@ -172,6 +175,7 @@ function createHarness(options: { loadCurrentSessionUsage, refreshRunModePreference, restoreSteerIntoComposer, + onTaskTerminal, stop: () => scope.stop(), } } @@ -1813,6 +1817,142 @@ describe('useChatRpcEventHandlers task group lifecycle', () => { } }) + it.each([ + ['task.succeeded', 'succeeded', {}], + ['task.cancelled', 'cancelled', {}], + ['task.timeout', 'timeout', {}], + ['task.failed', 'failed', {}], + ['task.abandoned', 'abandoned', {}], + ['session.event.error', 'failed', {}], + ['session.event.done', 'cancelled', { reason: 'aborted' }], + ['session.event.error', 'cancelled', { status: 'killed' }], + ['session.event.error', 'timeout', { status: 'timed_out' }], + ])( + 'passes the authoritative %s settlement to terminal presentation owners', + (event, expectedStatus, extra) => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.bindActiveStreamTask('task-terminal-1') + api.handlers.onWireEventFixture(event, { + session_key: 'agent:main:test', + task_id: 'task-terminal-1', + stream_seq: 1, + generation_epoch: 0, + ...extra, + }) + + expect(onTaskTerminal).toHaveBeenCalledWith('task-terminal-1', expectedStatus) + } finally { + stop() + } + }, + ) + + it('uses a direct turn id to settle presentation without TaskRuntime', () => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.handlers.onWireEventFixture('session.event.done', { + session_key: 'agent:main:test', + turn_id: 'direct-turn-1', + stream_seq: 1, + generation_epoch: 0, + }) + + expect(onTaskTerminal).toHaveBeenCalledWith('direct-turn-1', 'succeeded') + } finally { + stop() + } + }) + + it('passes interrupted terminal state from the authoritative session projection', () => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.handlers.onSessionsChanged({ + session_key: 'agent:main:test', + reason: 'task_terminal', + run_status: 'interrupted', + last_task: { task_id: 'task-interrupted-1', status: 'interrupted' }, + }) + + expect(onTaskTerminal).toHaveBeenCalledWith('task-interrupted-1', 'interrupted') + } finally { + stop() + } + }) + + it('passes succeeded terminal state from the authoritative session projection', () => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.handlers.onSessionsChanged({ + session_key: 'agent:main:test', + reason: 'task_terminal', + run_status: 'idle', + last_task: { task_id: 'task-succeeded-1', status: 'succeeded' }, + }) + + expect(onTaskTerminal).toHaveBeenCalledWith('task-succeeded-1', 'succeeded') + } finally { + stop() + } + }) + + it('passes a terminal to its domain owner before rejecting a different render owner', () => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.bindActiveStreamTask('task-rendered') + api.handlers.onWireEventFixture('task.timeout', { + session_key: 'agent:main:test', + task_id: 'task-plan-owner', + stream_seq: 1, + generation_epoch: 0, + }) + + expect(onTaskTerminal).toHaveBeenCalledWith('task-plan-owner', 'timeout') + } finally { + stop() + } + }) + + it('passes a terminal to its domain owner before buffering pending task acceptance', () => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.bindActiveStreamTask(PENDING_STREAM_TASK_ID) + api.handlers.onWireEventFixture('task.timeout', { + session_key: 'agent:main:test', + task_id: 'task-pending-acceptance', + stream_seq: 1, + generation_epoch: 0, + }) + + expect(onTaskTerminal).toHaveBeenCalledWith('task-pending-acceptance', 'timeout') + } finally { + stop() + } + }) + + it('rejects foreign-session and stale-epoch terminal settlements', () => { + const { api, onTaskTerminal, stop } = createHarness() + try { + api.handlers.onWireEventFixture('task.failed', { + session_key: 'agent:other:test', + task_id: 'task-foreign', + stream_seq: 1, + generation_epoch: 0, + }) + api.handlers.onWireEventFixture('task.timeout', { + session_key: 'agent:main:test', + task_id: 'task-stale', + epoch: -1, + stream_seq: 2, + generation_epoch: 0, + }) + + expect(onTaskTerminal).not.toHaveBeenCalled() + } finally { + stop() + } + }) + it('releases pending work when the last background-only task group finishes', () => { const { api, diff --git a/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.ts b/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.ts index efbe2b4f8..6ef75664b 100644 --- a/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.ts +++ b/opensquilla-webui/src/composables/chat/useChatRpcEventHandlers.ts @@ -56,6 +56,8 @@ import { taskGroupId as eventTaskGroupId, taskTerminalAsSessionEvent as normalizeTaskTerminalEvent, taskTerminalStatus as eventTaskTerminalStatus, + taskTerminalStatusFromValue as taskSettlementStatus, + type TaskTerminalStatus, } from '@/utils/chat/streamEvents' import { localizedChatErrorMessage } from '@/utils/chat/errors' import { normalizeTurnOutcome } from '@/utils/chat/turnOutcome' @@ -195,6 +197,7 @@ export interface UseChatRpcEventHandlersOptions { handleSessionConnectionState?: (state: string) => SessionBootstrapRun | undefined loadCurrentSessionUsage: () => void refreshRunModePreference?: () => void | Promise + onTaskTerminal?: (taskId: string, status: TaskTerminalStatus) => void } type ChatDoneUsageFields = { @@ -1557,6 +1560,31 @@ export function useChatRpcEventHandlers(options: UseChatRpcEventHandlersOptions) : null } + function eventTaskSettlementStatus( + eventKind: ConversationSemanticEventKind, + payload: SessionEventPayload, + ): TaskTerminalStatus | '' { + const compactStatus = eventTaskTerminalStatus(eventKind) + if (compactStatus) return compactStatus + + const lastTask = (payload.last_task || payload.lastTask) as { status?: unknown } | undefined + const rawPayloadStatus = String( + payload.run_status + || payload.runStatus + || payload.status + || '', + ) + const payloadStatus = taskSettlementStatus(rawPayloadStatus) + || taskSettlementStatus(options.normalizeRunStatus(rawPayloadStatus)) + || taskSettlementStatus(lastTask?.status) + if (eventKind === 'turn-completed' && payload.reason === 'aborted') { + return payloadStatus || 'cancelled' + } + if (eventKind === 'turn-completed') return payloadStatus || 'succeeded' + if (eventKind === 'turn-failed') return payloadStatus || 'failed' + return '' + } + function isStoppedCancelledTerminalEvent(terminalStatus: string, payload: SessionEventPayload): boolean { const taskId = payloadTaskId(payload) return Boolean( @@ -2040,6 +2068,12 @@ export function useChatRpcEventHandlers(options: UseChatRpcEventHandlersOptions) const changedTask = (payload.changed_task || payload.changedTask) as ChatRunStatusSource['active_task'] const changedTaskStatus = String(changedTask?.status || '').toLowerCase() if (changedTaskStatus === 'queued') options.taskOwnership?.noteQueued(changedTask || '') + const payloadTerminalTask = terminalSessionChangeTask(payload) + const payloadTerminalTaskId = chatTaskId(payloadTerminalTask) + const payloadTerminalStatus = taskSettlementStatus(payloadTerminalTask?.status) + if (payloadTerminalTaskId && payloadTerminalStatus) { + options.onTaskTerminal?.(payloadTerminalTaskId, payloadTerminalStatus) + } // changed_task describes which lifecycle row changed; it is deliberately // non-authoritative when Gateway snapshot generation failed. Only the // direct task.running event or an active_task/run_status projection may @@ -2055,8 +2089,6 @@ export function useChatRpcEventHandlers(options: UseChatRpcEventHandlersOptions) sessionChangeIsTerminal(payload) && bufferPendingTerminalEvent({ kind: 'session-change', payload }) ) return - const payloadTerminalTask = terminalSessionChangeTask(payload) - const payloadTerminalTaskId = chatTaskId(payloadTerminalTask) const activeProjection = (payload.active_task || payload.activeTask) as ChatRunStatusSource['active_task'] const carriesSettledContinuation = Boolean( sessionChangeIsTerminal(payload) @@ -2241,6 +2273,7 @@ export function useChatRpcEventHandlers(options: UseChatRpcEventHandlersOptions) ) ) { if (!acceptStreamSeq(payloadObj)) return + options.onTaskTerminal?.(succeededTaskId, 'succeeded') if ( awaitingCommitTaskIds.value.has(succeededTaskId) && rememberTrackedTask(taskSucceededSyncedIds, succeededTaskId) @@ -2267,6 +2300,15 @@ export function useChatRpcEventHandlers(options: UseChatRpcEventHandlersOptions) // name. Without this, a successor whose done frame was buffered behind A // remains marked running after replay and blocks every future drain. const terminalTaskId = terminalEvent ? payloadTaskId(payloadObj) : '' + const terminalOwnerId = terminalTaskId || ( + terminalEvent + ? String(payloadObj.turn_id ?? payloadObj.turnId ?? '').trim() + : '' + ) + const settlementStatus = eventTaskSettlementStatus(eventKind, payloadObj) + if (terminalOwnerId && settlementStatus) { + options.onTaskTerminal?.(terminalOwnerId, settlementStatus) + } if ( terminalStatus && terminalStatus !== 'succeeded' diff --git a/opensquilla-webui/src/composables/chat/useChatSessionSubscription.test.ts b/opensquilla-webui/src/composables/chat/useChatSessionSubscription.test.ts index 997ca5c80..f83c80e0e 100644 --- a/opensquilla-webui/src/composables/chat/useChatSessionSubscription.test.ts +++ b/opensquilla-webui/src/composables/chat/useChatSessionSubscription.test.ts @@ -77,6 +77,7 @@ function live(overrides: Partial = {}): SessionReadLive { sessionKey: KEY, activity: 'idle', activeTaskId: null, + streamGeneration: 'generation-1', initialMetadata: metadata(), snapshot: null, reloadRequired: null, @@ -403,7 +404,7 @@ describe('useChatSessionSubscription domain lease', () => { }) expect(onLiveSnapshot).toHaveBeenCalledWith(snapshot) expect(onRunModeLock).toHaveBeenCalledWith(initialMetadata.runModeLock) - expect(onSnapshot).toHaveBeenCalledWith(initialMetadata) + expect(onSnapshot).toHaveBeenCalledWith(initialMetadata, 'generation-1') expect(subject.startStreaming).toHaveBeenCalledWith(90_000) expect(subject.activeStreamTaskId.value).toBe('task-live') expect(reconcileStreamTaskClock).toHaveBeenCalledWith({ @@ -555,10 +556,43 @@ describe('useChatSessionSubscription domain lease', () => { 7, hydrated, )) - expect(onSnapshot).toHaveBeenCalledWith(hydrated) + expect(onSnapshot).toHaveBeenCalledWith(hydrated, 'generation-1') expect(taskOwnership.hydrationResolved.value).toBe(true) }) + it('syncs a restarted lease generation before no-event hydration reconciliation', async () => { + const complete = deferred() + const onSnapshot = vi.fn() + const subject = harness(leaseFixture({ + live: live({ + streamGeneration: 'generation-2', + reloadRequired: 'generationChanged', + initialMetadata: metadata({ + hydrationComplete: false, + deferredFields: ['pendingUserInputs', 'goalSnapshotStreamSeq'], + }), + }), + metadata: complete.promise, + }).lease, { + lastStreamSeq: ref(100), + onSnapshot, + }) + subject.api.observeStreamGeneration({ streamGeneration: 'generation-1' }) + + await expect(subject.api.subscribeSession()).resolves.toMatchObject({ + authoritative: true, + }) + expect(subject.api.streamGeneration.value).toBe('generation-2') + expect(subject.lastStreamSeq.value).toBe(0) + + const restartedMetadata = metadata({ goalSnapshotStreamSeq: 0 }) + complete.resolve(restartedMetadata) + await vi.waitFor(() => expect(onSnapshot).toHaveBeenCalledWith( + restartedMetadata, + 'generation-2', + )) + }) + it('honors background activity even before task-group metadata is complete', async () => { const subject = harness(leaseFixture({ live: live({ diff --git a/opensquilla-webui/src/composables/chat/useChatSessionSubscription.ts b/opensquilla-webui/src/composables/chat/useChatSessionSubscription.ts index 994d47df8..14780861b 100644 --- a/opensquilla-webui/src/composables/chat/useChatSessionSubscription.ts +++ b/opensquilla-webui/src/composables/chat/useChatSessionSubscription.ts @@ -62,7 +62,10 @@ export interface UseChatSessionSubscriptionOptions { ) => void onSessionMetadataError?: (key: string, generation: number) => void onSessionMissing?: (key: string) => void - onSnapshot?: (snapshot: SessionReadMetadata) => void + onSnapshot?: ( + snapshot: SessionReadMetadata, + streamGeneration: string | null, + ) => void } export interface SessionMetadataRetryOptions { @@ -202,6 +205,7 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp metadataGeneration: number | undefined, metadata: SessionReadMetadata, activity: SessionReadActivity = 'unknown', + snapshotStreamGeneration: string | null = null, ): SessionSubscriptionOutcome { if (metadataGeneration !== undefined) { options.onSessionMetadata?.(key, metadataGeneration, metadata) @@ -218,7 +222,7 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp ? { ...metadata, runStatus: 'idle', activeTask: null } : metadata const effectiveSource = metadataRunStatusSource(effectiveMetadata) - options.onSnapshot?.(effectiveMetadata) + options.onSnapshot?.(effectiveMetadata, snapshotStreamGeneration) options.taskOwnership?.applySnapshot(effectiveSource, true) // Do not clear an acceptance-result-unknown Stop from an idle snapshot. // The subscription can race ahead of the original ingress commit, so only @@ -324,6 +328,7 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp metadataHydration: number, metadataGeneration: number | undefined, activity: SessionReadActivity, + snapshotStreamGeneration: string | null, signal: AbortSignal, ): void { void lease.metadata.then((metadata) => { @@ -334,7 +339,13 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp if (!metadata.hydrationComplete) { throw new Error('Session state hydration remained incomplete') } - applyHydratedSubscriptionState(key, metadataGeneration, metadata, activity) + applyHydratedSubscriptionState( + key, + metadataGeneration, + metadata, + activity, + snapshotStreamGeneration, + ) }).catch((cause) => { if ( !isCurrentSubscription(lease, key, sequence, signal) @@ -368,6 +379,15 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp if (!isCurrentSubscription(lease, key, sequence, signal)) { return { ...UNAVAILABLE_SUBSCRIPTION, cancelled: true } } + if (live.streamGeneration) { + observeStreamGeneration({ + sessionKey: key, + streamGeneration: live.streamGeneration, + ...(live.reloadRequired === 'generationChanged' + ? { replayGapReason: 'stream_generation_changed' } + : {}), + }) + } let snapshotTaskLive = false const snapshot = live.snapshot if (snapshot?.sessionKey === key) { @@ -379,7 +399,7 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp snapshotTaskLive = Boolean(snapshotTaskId) && !settledSnapshot } if (live.reloadRequired) { - if (live.reloadRequired === 'generationChanged') { + if (live.reloadRequired === 'generationChanged' && !live.streamGeneration) { syncCursor(conversationRuntime.reset(cursor())) options.resetStreamLiveTurnState() } @@ -391,6 +411,7 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp metadataGeneration, live.initialMetadata, live.activity, + live.streamGeneration, ) } if (options.ownershipHydrationRequired?.() !== false) { @@ -406,6 +427,7 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp metadataHydration, metadataGeneration, live.activity, + live.streamGeneration, signal, ) // Fast ACK is authoritative for delivery registration. Deferred storage @@ -482,16 +504,34 @@ export function useChatSessionSubscription(options: UseChatSessionSubscriptionOp ) try { - const hydration = await waitForMetadataRetry( - lease.retryMetadata(), - controller.signal, - timeoutMs, - ) + const [hydration, live] = await Promise.all([ + waitForMetadataRetry( + lease.retryMetadata(), + controller.signal, + timeoutMs, + ), + lease.live, + ]) if (!isCurrent()) return false + if (live.streamGeneration) { + observeStreamGeneration({ + sessionKey: key, + streamGeneration: live.streamGeneration, + ...(live.reloadRequired === 'generationChanged' + ? { replayGapReason: 'stream_generation_changed' } + : {}), + }) + } if (!hydration.hydrationComplete) { throw new Error('Session state hydration remained incomplete') } - applyHydratedSubscriptionState(key, metadataGeneration, hydration) + applyHydratedSubscriptionState( + key, + metadataGeneration, + hydration, + 'unknown', + live.streamGeneration, + ) return true } catch (cause) { if (isCurrent() && metadataGeneration !== undefined) { diff --git a/opensquilla-webui/src/modules/sessionReadLifecycle.test.ts b/opensquilla-webui/src/modules/sessionReadLifecycle.test.ts index d09e31b8b..44ae6864e 100644 --- a/opensquilla-webui/src/modules/sessionReadLifecycle.test.ts +++ b/opensquilla-webui/src/modules/sessionReadLifecycle.test.ts @@ -168,6 +168,7 @@ describe('SessionReadLifecycle', () => { } await expect(lease.live).resolves.toMatchObject({ sessionKey: 'alpha', + streamGeneration: 'stream-1', reloadRequired: null, }) expect(adapter.openRecords).toEqual([{ diff --git a/opensquilla-webui/src/modules/sessionReadLifecycle.ts b/opensquilla-webui/src/modules/sessionReadLifecycle.ts index 8952efef5..036ca903f 100644 --- a/opensquilla-webui/src/modules/sessionReadLifecycle.ts +++ b/opensquilla-webui/src/modules/sessionReadLifecycle.ts @@ -66,6 +66,8 @@ export interface SessionReadLive { readonly sessionKey: string readonly activity: SessionReadActivity readonly activeTaskId: string | null + /** Exact stream namespace shared by this lease's metadata reads. */ + readonly streamGeneration: string | null /** Fast-ACK metadata may be incomplete. Await `lease.metadata` for hydration. */ readonly initialMetadata: SessionReadMetadata readonly snapshot: SessionReadSnapshot | null @@ -218,7 +220,10 @@ export interface SessionReadPortOpenRequest { readonly signal: AbortSignal } -export interface SessionReadPortLive extends Omit { +export interface SessionReadPortLive extends Omit< + SessionReadLive, + 'reloadRequired' | 'streamGeneration' +> { readonly cursor: ConversationCursorSignal readonly snapshotCursor: ConversationCursorSignal | null } @@ -460,6 +465,7 @@ export function createSessionReadLifecycle( sessionKey: value.sessionKey, activity: value.activity, activeTaskId: value.activeTaskId, + streamGeneration: state.cursor.streamGeneration, initialMetadata: value.initialMetadata, snapshot: value.snapshot, reloadRequired: replay.requiresHistory diff --git a/opensquilla-webui/src/utils/chat/streamEvents.ts b/opensquilla-webui/src/utils/chat/streamEvents.ts index fc78fef83..fbb1d771a 100644 --- a/opensquilla-webui/src/utils/chat/streamEvents.ts +++ b/opensquilla-webui/src/utils/chat/streamEvents.ts @@ -58,6 +58,40 @@ export function conversationCursorSignal(source: unknown): ConversationCursorSig export type NormalizeRunStatus = (status: string) => string +export type TaskSettlementStatus = + | 'failed' + | 'cancelled' + | 'timeout' + | 'abandoned' + | 'interrupted' + +export type TaskTerminalStatus = 'succeeded' | TaskSettlementStatus + +const TASK_TERMINAL_STATUS_VALUES = new Set([ + 'succeeded', + 'failed', + 'cancelled', + 'timeout', + 'abandoned', + 'interrupted', +]) + +export function taskTerminalStatusFromValue(value: unknown): TaskTerminalStatus | '' { + const normalized = String(value || '').trim().toLowerCase() + const compatible = normalized === 'success' || normalized === 'complete' + ? 'succeeded' + : normalized === 'error' + ? 'failed' + : normalized === 'killed' + ? 'cancelled' + : normalized === 'timed_out' + ? 'timeout' + : normalized + return TASK_TERMINAL_STATUS_VALUES.has(compatible as TaskTerminalStatus) + ? compatible as TaskTerminalStatus + : '' +} + export const PENDING_STREAM_TASK_ID = '__opensquilla_pending_stream_task__' export const STOPPED_STREAM_TASK_ID = '__opensquilla_stopped_stream_task__' // Tombstone left after a terminal event closes the live turn. Unlike an empty @@ -154,8 +188,10 @@ export function sessionChangeIsTerminal( return ['failed', 'timeout', 'cancelled', 'interrupted'].includes(runStatus) } -export function taskTerminalStatus(event: ConversationSemanticEventKind): string { - const statusByKind: Partial> = { +export function taskTerminalStatus( + event: ConversationSemanticEventKind, +): TaskTerminalStatus | '' { + const statusByKind: Partial> = { 'task-succeeded': 'succeeded', 'task-failed': 'failed', 'task-timed-out': 'timeout', diff --git a/opensquilla-webui/src/views/ChatView.session-missing-wiring.test.ts b/opensquilla-webui/src/views/ChatView.session-missing-wiring.test.ts index 26df44e13..8a832eabd 100644 --- a/opensquilla-webui/src/views/ChatView.session-missing-wiring.test.ts +++ b/opensquilla-webui/src/views/ChatView.session-missing-wiring.test.ts @@ -18,4 +18,26 @@ describe('ChatView missing-session wiring', () => { expect(subscriptionWiring).not.toContain('SESSION_NOT_FOUND') expect(subscriptionWiring).not.toContain('NOT_FOUND') }) + + it('forwards complete session-read metadata to pending-input reconciliation', () => { + const assignmentStart = chatViewSource.indexOf( + 'applyPendingUserInputSnapshot = (snapshot, snapshotStreamGeneration)', + ) + expect(assignmentStart).toBeGreaterThan(-1) + const assignmentEnd = chatViewSource.indexOf('\n})', assignmentStart) + const assignment = chatViewSource.slice(assignmentStart, assignmentEnd) + expect(assignment).toContain('...snapshot') + expect(assignment).toContain('streamGeneration: snapshotStreamGeneration') + expect(assignment).not.toContain('streamGeneration.value') + + const subscriptionStart = chatViewSource.indexOf( + 'const chatSessionSubscription = useChatSessionSubscription({', + ) + const subscriptionEnd = chatViewSource.indexOf('\n})', subscriptionStart) + const subscriptionWiring = chatViewSource.slice(subscriptionStart, subscriptionEnd) + expect(subscriptionWiring).toContain('onSnapshot: (snapshot, snapshotStreamGeneration)') + expect(subscriptionWiring).toContain( + 'applyPendingUserInputSnapshot(snapshot, snapshotStreamGeneration)', + ) + }) }) diff --git a/opensquilla-webui/src/views/ChatView.vue b/opensquilla-webui/src/views/ChatView.vue index 7db5e3966..ef6bd2632 100644 --- a/opensquilla-webui/src/views/ChatView.vue +++ b/opensquilla-webui/src/views/ChatView.vue @@ -551,7 +551,7 @@ @@ -1094,6 +1094,8 @@ import { FINISHED_STREAM_TASK_ID, PENDING_STREAM_TASK_ID, STOPPED_STREAM_TASK_ID, + taskTerminalStatusFromValue, + type TaskTerminalStatus, } from '@/utils/chat/streamEvents' import { copyTextWithFallback, copyImageToClipboard, downloadBlob, shareCopyImageSupported } from '@/utils/browser' import { useCopyFeedback } from '@/composables/chat/useCopyFeedback' @@ -2287,6 +2289,7 @@ const { currentPlan, currentPlanRevisionId, activePlanRun, + planRunSettlementPending, modeBusy: planModeBusy, modeAppliesNextTurn: planModeAppliesNextTurn, pendingAction: planActionPending, @@ -2358,8 +2361,10 @@ const chatHistory = useChatHistory({ stripTimePrefix, scrollToBottom, onTerminalTask: outcome => { - const taskId = outcome.taskId || '' + const taskId = outcome.taskId || outcome.turnId || '' if (!taskId) return + const terminalStatus = taskTerminalStatusFromValue(outcome.status) + if (terminalStatus) settleTaskTerminalPresentation(taskId, terminalStatus) taskOwnership.noteTerminal(taskId) const ownsLiveStream = activeStreamTaskId.value === taskId const ownsRunStatus = chatTaskId(runStatus.value.task) === taskId @@ -2578,7 +2583,21 @@ async function handleRegenerateMessage( settle?.(accepted) } -let applyPendingUserInputSnapshot: (snapshot: SessionReadMetadata) => void = () => {} +function terminalTaskFromRunState(source: SessionReadMetadata) { + const task = source.lastTask || source.activeTask + const taskId = chatTaskId(task) + const status = taskTerminalStatusFromValue(task?.status) + return taskId && status ? { taskId, status } : null +} + +let settleTaskTerminalPresentation: ( + taskId: string, + status: TaskTerminalStatus, +) => void = () => {} +let applyPendingUserInputSnapshot: ( + snapshot: SessionReadMetadata, + streamGeneration: string | null, +) => void = () => {} let applyGoalSnapshot: (snapshot: SessionReadMetadata) => void = () => {} const chatSessionSubscription = useChatSessionSubscription({ sessionReadLeaseReader: sessionReadLifecycle, @@ -2638,11 +2657,15 @@ const chatSessionSubscription = useChatSessionSubscription({ activeProjectWorkspace.failSessionResolution(key, generation) }, onSessionMissing: markSessionMissing, - onSnapshot: snapshot => { + onSnapshot: (snapshot, snapshotStreamGeneration) => { + const terminalTask = terminalTaskFromRunState(snapshot) chatSessionRouting.applyBootstrap(snapshot) chatPlans.applyBootstrap(snapshot) applyGoalSnapshot(snapshot) - applyPendingUserInputSnapshot(snapshot) + applyPendingUserInputSnapshot(snapshot, snapshotStreamGeneration) + if (terminalTask) { + settleTaskTerminalPresentation(terminalTask.taskId, terminalTask.status) + } }, }) const { @@ -3747,6 +3770,9 @@ const chatApprovals = useChatApprovals({ sessionConversation, approvalCenter, sessionKey, + currentEpoch, + streamGeneration, + observeStreamGeneration, runStatus, stream: { isStreaming, appendInterruptFrame, ensureInterruptBubble }, interruptState, @@ -3764,11 +3790,19 @@ const { extendInterrupt, submitClarify, dismissClarify, + settlePendingClarifyForTerminalTask, applyUserInputBootstrap, } = chatApprovals -applyPendingUserInputSnapshot = snapshot => applyUserInputBootstrap({ - pendingUserInputs: [...snapshot.pendingUserInputs], +applyPendingUserInputSnapshot = (snapshot, snapshotStreamGeneration) => applyUserInputBootstrap({ + ...snapshot, + ...(snapshotStreamGeneration ? { streamGeneration: snapshotStreamGeneration } : {}), }) +settleTaskTerminalPresentation = (taskId, status) => { + if (status !== 'succeeded') { + chatPlans.settleActiveRunForTerminalTask(taskId, status) + } + settlePendingClarifyForTerminalTask(taskId, status) +} const dockedPlanQuestionnaire = computed(() => ( pendingClarify.value?.presentation === 'plan_questionnaire_v1' @@ -3895,6 +3929,9 @@ const rpcEventHandlers = useChatRpcEventHandlers({ handleSessionConnectionState(state, !isDraftRoute()), loadCurrentSessionUsage, refreshRunModePreference: refreshPostBootstrapMetadata, + onTaskTerminal: (taskId, status) => { + settleTaskTerminalPresentation(taskId, status) + }, }) bindActiveStreamTask = rpcEventHandlers.bindActiveStreamTask restoreLiveTurnSnapshot = rpcEventHandlers.restoreLiveTurnSnapshot @@ -4477,6 +4514,7 @@ const planCardPendingAction = computed(() => { const planActionsDisabled = computed(() => isStreaming.value || planModeBusy.value + || planRunSettlementPending.value || Boolean(liveSendBlockedReason.value) || planActionPending.value !== null || activePlanRun.value?.status === 'queued'