diff --git a/mobile/src/session/MobileNativeChatQueuedMessages.test.tsx b/mobile/src/session/MobileNativeChatQueuedMessages.test.tsx
index b01a0fa66c18..b2a0a22bf0ea 100644
--- a/mobile/src/session/MobileNativeChatQueuedMessages.test.tsx
+++ b/mobile/src/session/MobileNativeChatQueuedMessages.test.tsx
@@ -166,6 +166,26 @@ describe('MobileNativeChatQueuedMessages', () => {
expect(flatStyle(caption.props.style).color).not.toBe('#ef4444')
})
+ it("lets a waiting /clear's caption wrap whole, so its next step is never cut off", async () => {
+ const waits = 'Waiting for background tasks to finish. Stop them to clear now.'
+ const mounted = await mount({
+ cards: [
+ card({
+ messageId: 'clear',
+ text: '/clear',
+ command: true,
+ runsOnItsOwn: true,
+ caption: waits
+ })
+ ]
+ })
+ const caption = mounted.root.find(
+ (node) => String(node.type) === 'Text' && node.props.children === waits
+ )
+ expect(caption.props.numberOfLines).toBeUndefined()
+ expect(flatStyle(caption.props.style).color).not.toBe('#ef4444')
+ })
+
it("lets a returned card's reason wrap whole, in the destructive color, beside an alert", async () => {
const reason =
'The provider did not accept this message: Claude does not support the image type .bmp in a steering message.'
diff --git a/mobile/src/session/MobileNativeChatQueuedMessages.tsx b/mobile/src/session/MobileNativeChatQueuedMessages.tsx
index 80476207c04d..3c05059b9038 100644
--- a/mobile/src/session/MobileNativeChatQueuedMessages.tsx
+++ b/mobile/src/session/MobileNativeChatQueuedMessages.tsx
@@ -130,17 +130,18 @@ export function MobileNativeChatQueuedMessages({
) : null}
{card.caption ? (
- // A returned card's reason only reads whole, often at its end; a hold is one line.
+ // A returned card's reason and a waiting /clear's next step only read whole, often
+ // at their end; a hold is one line.
{card.caption}
) : null}
{/* A command never steers: its Send shows only while the agent is idle. */}
- {card.waitsForAgent ? null : (
+ {card.waitsForAgent || card.runsOnItsOwn ? null : (
{
questionAnswers: false,
queuedMessages: false,
queuedCommands: false,
+ queuedClear: false,
quietRepeatedStop: false,
statusFeed: false
}
@@ -32,6 +34,9 @@ describe('structuredAgentSessionHostSupport', () => {
expect(
structuredAgentSessionHostSupport([AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY])
).toEqual({ ...none, queuedCommands: true })
+ expect(
+ structuredAgentSessionHostSupport([AGENT_SESSION_QUEUED_CLEAR_RUNTIME_CAPABILITY])
+ ).toEqual({ ...none, queuedClear: true })
expect(
structuredAgentSessionHostSupport([AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY])
).toEqual({ ...none, quietRepeatedStop: true })
diff --git a/mobile/src/session/mobile-structured-agent-session-host-support.ts b/mobile/src/session/mobile-structured-agent-session-host-support.ts
index 84c329e63738..f3500c6c4b7c 100644
--- a/mobile/src/session/mobile-structured-agent-session-host-support.ts
+++ b/mobile/src/session/mobile-structured-agent-session-host-support.ts
@@ -1,6 +1,7 @@
import {
AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY,
AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY,
+ AGENT_SESSION_QUEUED_CLEAR_RUNTIME_CAPABILITY,
AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY,
AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY,
AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY,
@@ -15,6 +16,8 @@ export type StructuredAgentSessionHostSupport = {
queuedMessages: boolean
/** A /compact sent while the agent works waits as a card; an older host refuses it. */
queuedCommands: boolean
+ /** The same for a /clear, which the host runs itself when its card's turn comes. */
+ queuedClear: boolean
/** The host publishes every session's status on one stream. A host from before phones could
* read it refuses the call, which reads the same as its absence. */
statusFeed: boolean
@@ -30,6 +33,7 @@ export function structuredAgentSessionHostSupport(
questionAnswers: capabilities.includes(AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY),
queuedMessages: capabilities.includes(AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY),
queuedCommands: capabilities.includes(AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY),
+ queuedClear: capabilities.includes(AGENT_SESSION_QUEUED_CLEAR_RUNTIME_CAPABILITY),
statusFeed: capabilities.includes(AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY),
quietRepeatedStop: capabilities.includes(AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY)
}
diff --git a/mobile/src/session/mobile-structured-composer-command.test.ts b/mobile/src/session/mobile-structured-composer-command.test.ts
index ec36a48dfe2c..b18d02d770a4 100644
--- a/mobile/src/session/mobile-structured-composer-command.test.ts
+++ b/mobile/src/session/mobile-structured-composer-command.test.ts
@@ -139,10 +139,32 @@ describe('mobile structured conversation commands', () => {
const fields = requestFields(sendRequest.mock.calls[0])
expect(fields).toMatchObject({ command: 'compact', delivery: 'queue-if-active' })
expect(input.onError).not.toHaveBeenCalled()
- // A /clear never waits: the busy check still answers it.
+ // A host that holds only /compact: the busy check still answers a /clear.
expect(await dispatchMobileStructuredCommand({ ...input, text: '/clear' })).toBe('rejected')
expect(sendRequest).toHaveBeenCalledOnce()
})
+ it('a /clear the host holds in line asks to wait the same way, and its card is the answer', async () => {
+ const { input, sendRequest } = setup()
+ sendRequest.mockResolvedValueOnce({
+ ok: true,
+ result: {
+ ok: true,
+ value: {
+ command: 'clear',
+ state: 'completed',
+ queued: { messageId: 'op', position: 1, state: 'waiting' }
+ }
+ }
+ })
+ input.busy = () => 'working'
+ input.waitsInLine = () => true
+ expect(await dispatchMobileStructuredCommand({ ...input, text: '/clear' })).toBe('accepted')
+ expect(requestFields(sendRequest.mock.calls[0])).toMatchObject({
+ command: 'clear',
+ delivery: 'queue-if-active'
+ })
+ expect(input.onError).not.toHaveBeenCalled()
+ })
it('a /clear while the agent works says so in plain words', async () => {
const { input, sendRequest } = setup()
input.busy = () => 'working'
diff --git a/mobile/src/session/mobile-structured-queued-message-cards.test.ts b/mobile/src/session/mobile-structured-queued-message-cards.test.ts
index ce3eda3c57b2..f69e591a0e23 100644
--- a/mobile/src/session/mobile-structured-queued-message-cards.test.ts
+++ b/mobile/src/session/mobile-structured-queued-message-cards.test.ts
@@ -102,6 +102,54 @@ describe('mobileQueuedMessageCards', () => {
).toMatchObject({ command: true, waitsForAgent: true })
})
+ const clearDraft = (overrides: Partial = {}) =>
+ draft({
+ messageId: 'c',
+ body: {
+ kind: 'message',
+ role: 'user',
+ blocks: [{ type: 'text', text: '/clear' }],
+ command: { name: 'clear' }
+ },
+ ...overrides
+ })
+
+ it('a /clear the host names as next offers no Send: the queue is about to run it', () => {
+ const idle = { pendingPrompt: false }
+ expect(
+ mobileQueuedMessageCards([clearDraft()], [], { ...idle, nextQueuedMessageId: 'c' })[0]
+ ).toMatchObject({ runsOnItsOwn: true })
+ // Held with nothing published (a reopen holds it until the next turn): Send stays, as on /compact.
+ expect(mobileQueuedMessageCards([clearDraft()], [], idle)[0]?.runsOnItsOwn).toBeUndefined()
+ })
+
+ it("a /clear the host holds on background tasks says so; the strip's Stop is named when offered", () => {
+ const wait = { messageId: 'c', reason: 'background-tasks' as const }
+ const facts = { pendingPrompt: false, nextQueuedMessageWait: wait }
+ const [first, second] = mobileQueuedMessageCards(
+ [clearDraft(), draft({ messageId: 'm', position: 2 })],
+ [],
+ facts
+ )
+ expect(first).toMatchObject({
+ caption: 'Waiting for background tasks to finish',
+ runsOnItsOwn: true
+ })
+ expect(second?.caption).toBeNull()
+ expect(second?.runsOnItsOwn).toBeUndefined()
+ expect(
+ mobileQueuedMessageCards([clearDraft()], [], { ...facts, backgroundTasksStoppable: true })[0]
+ ?.caption
+ ).toBe('Waiting for background tasks to finish. Stop them to clear now.')
+ // A handoff wait offers no Send either, with nothing more to say.
+ expect(
+ mobileQueuedMessageCards([clearDraft()], [], {
+ pendingPrompt: false,
+ nextQueuedMessageWait: { messageId: 'c', reason: 'handoff' }
+ })[0]
+ ).toMatchObject({ caption: null, runsOnItsOwn: true })
+ })
+
it("a send-failed command card's caption names Send only when Send is there", () => {
const failed = draft({
messageId: 'c',
diff --git a/mobile/src/session/mobile-structured-queued-message-cards.ts b/mobile/src/session/mobile-structured-queued-message-cards.ts
index 42883b4ed4df..a484065d98fc 100644
--- a/mobile/src/session/mobile-structured-queued-message-cards.ts
+++ b/mobile/src/session/mobile-structured-queued-message-cards.ts
@@ -17,7 +17,8 @@ import { structuredAgentSessionAttemptFailureParts } from '../../../src/shared/s
import {
QUEUED_MESSAGE_PAUSED_SEND_FAILED,
type AgentSessionQueuedMessage,
- type AgentSessionQueuePause
+ type AgentSessionQueuePause,
+ type AgentSessionQueueWait
} from '../../../src/shared/agent-session-wire'
import { readAgentMessageSource } from '../../../src/shared/agent-session-message-source'
import { agentMessageAttribution } from './mobile-agent-message-attribution'
@@ -38,6 +39,9 @@ export type MobileQueuedMessageCard = {
command?: true
/** A command card while the agent works: it offers no send until the agent is idle. */
waitsForAgent?: true
+ /** A /clear the host names as next, or holds until a wait ends: it runs without a press, so it
+ * offers no Send. */
+ runsOnItsOwn?: true
}
function queuedMessageBodyText(body: AgentSessionQueuedMessage['body']): string {
@@ -115,6 +119,11 @@ export function mobileQueuedMessageCards(
pendingPrompt: boolean
queuePaused?: boolean
agentWorking?: boolean
+ /** The host's next card, and the /clear it holds while a wait ends; read, never re-picked. */
+ nextQueuedMessageId?: string | null
+ nextQueuedMessageWait?: AgentSessionQueueWait | null
+ /** The strip offers a Stop for the background tasks that /clear waits on. */
+ backgroundTasksStoppable?: boolean
agentName?: string
statedFailures?: readonly AgentSessionFailureFact[]
}
@@ -129,13 +138,15 @@ export function mobileQueuedMessageCards(
: []
)
)
+ const shown = queuedMessages.filter(
+ (draft) => draft.state === 'returned' || !handedOff.has(draft.messageId)
+ )
+ const wait = facts.nextQueuedMessageWait ?? null
let behindReturned = false
const cards: MobileQueuedMessageCard[] = []
- for (const draft of queuedMessages) {
- if (draft.state !== 'returned' && handedOff.has(draft.messageId)) {
- continue
- }
+ for (const draft of shown) {
const paused = draft.paused === true
+ const held = wait?.messageId === draft.messageId
const caption =
draft.state === 'returned'
? returnedCaption(draft, facts.agentName, facts.statedFailures)
@@ -151,7 +162,14 @@ export function mobileQueuedMessageCards(
null
: facts.pendingPrompt
? 'Waiting for your answer'
- : null
+ : held && wait.reason === 'background-tasks'
+ ? facts.backgroundTasksStoppable
+ ? 'Waiting for background tasks to finish. Stop them to clear now.'
+ : 'Waiting for background tasks to finish'
+ : null
+ const runsOnItsOwn =
+ draft.body.command?.name === 'clear' &&
+ (held || draft.messageId === facts.nextQueuedMessageId)
cards.push({
messageId: draft.messageId,
text: queuedMessageBodyText(draft.body),
@@ -165,7 +183,8 @@ export function mobileQueuedMessageCards(
...(draft.body.command !== undefined
? {
command: true as const,
- ...(facts.agentWorking ? { waitsForAgent: true as const } : {})
+ ...(facts.agentWorking ? { waitsForAgent: true as const } : {}),
+ ...(runsOnItsOwn ? { runsOnItsOwn: true as const } : {})
}
: {})
})
diff --git a/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx
index aef38e82e6e9..af0378215025 100644
--- a/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx
+++ b/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx
@@ -6,7 +6,10 @@ import { createElement } from 'react'
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { AgentChildWorkView } from '../../../src/shared/agent-status-child-work-view'
-import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire'
+import type {
+ AgentSessionQueuedMessage,
+ AgentSessionSubscribeEvent
+} from '../../../src/shared/agent-session-wire'
import type { RpcClient } from '../transport/rpc-client'
import {
backgroundTaskRowMeta,
@@ -41,8 +44,11 @@ function childView(id: string): AgentChildWorkView {
}
}
-function withChildren(hostNow?: number): AgentSessionSubscribeEvent {
- const snapshot = snapshotEvent()
+function withChildren(
+ hostNow?: number,
+ queuedMessages?: AgentSessionQueuedMessage[]
+): AgentSessionSubscribeEvent {
+ const snapshot = snapshotEvent(queuedMessages ? { queuedMessages } : undefined)
return snapshot.type === 'snapshot'
? {
...snapshot,
@@ -169,6 +175,40 @@ describe('mobile structured session background tasks', () => {
clock.mockRestore()
})
+ it('a /clear the host holds on the tasks once the turn ended says so, with no Send', async () => {
+ const card = (messageId: string, text: string, position: number) => ({
+ messageId,
+ position,
+ state: 'waiting' as const,
+ body: {
+ kind: 'message' as const,
+ role: 'user' as const,
+ blocks: [{ type: 'text' as const, text }],
+ ...(text === '/clear' ? { command: { name: 'clear' as const } } : {})
+ }
+ })
+ const opening = withChildren(undefined, [
+ card('clear-1', '/clear', 1),
+ card('after', 'After', 2)
+ ])
+ await mount(
+ opening.type === 'snapshot'
+ ? {
+ ...opening,
+ nextQueuedMessageWait: { messageId: 'clear-1', reason: 'background-tasks' }
+ }
+ : opening
+ )
+ expect(hook?.isWorking).toBe(false)
+ // The strip offers each child its own Stop, so the caption names it.
+ expect(hook?.queued.cards[0]).toMatchObject({
+ messageId: 'clear-1',
+ caption: 'Waiting for background tasks to finish. Stop them to clear now.',
+ runsOnItsOwn: true
+ })
+ expect(hook?.queued.cards[1]?.caption).toBeNull()
+ })
+
it("stops one child through the host's background-task cancel", async () => {
await mount(withChildren())
await act(async () => {
diff --git a/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx
index 7a14895fd66e..e61f6381abce 100644
--- a/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx
+++ b/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx
@@ -123,6 +123,7 @@ function Harness({
questionAnswers: questionAnswersSupported,
queuedMessages: false,
queuedCommands: false,
+ queuedClear: false,
statusFeed: false,
quietRepeatedStop: false
},
diff --git a/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts b/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts
index 7a0c57b9b352..b5980b61d300 100644
--- a/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts
+++ b/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts
@@ -17,6 +17,7 @@ export const CAPABLE: StructuredAgentSessionHostSupport = {
questionAnswers: false,
queuedMessages: true,
queuedCommands: false,
+ queuedClear: false,
statusFeed: false,
quietRepeatedStop: false
}
diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts
index fb3c69c51aa6..02b69225316e 100644
--- a/mobile/src/session/use-mobile-structured-agent-session.ts
+++ b/mobile/src/session/use-mobile-structured-agent-session.ts
@@ -39,6 +39,7 @@ import {
import type { MobileNativeChatSendErrorReporter } from './use-mobile-native-chat-send-error'
import { useMobileStructuredQueuedMessageControls } from './use-mobile-structured-queued-message-controls'
import { useMobileStructuredBackgroundTasks } from './use-mobile-structured-background-tasks'
+import { structuredSessionBackgroundTasksOfferStop } from '../../../src/shared/structured-session-background-tasks-view'
export function useMobileStructuredAgentSession(args: {
client: RpcClient | null
@@ -71,8 +72,9 @@ export function useMobileStructuredAgentSession(args: {
} = args
// Only a host that queues sends gets the delivery field; any host's published cards show.
const queueCapable = hostSupport?.queuedMessages === true
- // A /compact waits in line only where its card renders.
+ // A command waits in line only where its card renders.
const commandsWait = queueCapable && hostSupport?.queuedCommands === true
+ const clearWaits = queueCapable && hostSupport?.queuedClear === true
const promptCancelSupported = hostSupport?.promptCancel ?? null
const hostAnswersRepeatedStops = hostSupport?.quietRepeatedStop ?? null
const sessionKey = encodeNativeChatTranscriptIdentity([sourceIdentity, agent, sessionId])
@@ -119,6 +121,7 @@ export function useMobileStructuredAgentSession(args: {
enabled,
queueCapable,
commandsWait,
+ clearWaits,
stateRef,
commandPending: commandPendingRef,
controller: sendController,
@@ -221,6 +224,9 @@ export function useMobileStructuredAgentSession(args: {
submissions: state.submissions,
pendingPrompt: approvalPrompt !== null || questionPrompt !== null,
agentWorking: isStructuredAgentSessionMainAgentWorking(turnId, state.submissions, state.fence),
+ nextQueuedMessageId: state.nextQueuedMessageId ?? null,
+ nextQueuedMessageWait: state.nextQueuedMessageWait ?? null,
+ backgroundTasksStoppable: structuredSessionBackgroundTasksOfferStop(backgroundTasks.view),
mutate,
appendComposerText,
onSendError,
diff --git a/mobile/src/session/use-mobile-structured-queued-message-controls.ts b/mobile/src/session/use-mobile-structured-queued-message-controls.ts
index 1a73756841f1..b9fc95770e84 100644
--- a/mobile/src/session/use-mobile-structured-queued-message-controls.ts
+++ b/mobile/src/session/use-mobile-structured-queued-message-controls.ts
@@ -15,6 +15,7 @@ import { agentSessionVisibleFailureFacts } from '../../../src/shared/agent-sessi
import type {
AgentSessionQueuedMessageDeleteResult,
AgentSessionQueuedMessagesResumeResult,
+ AgentSessionQueueWait,
AgentSessionSendResult
} from '../../../src/shared/agent-session-wire'
import {
@@ -58,6 +59,11 @@ export function useMobileStructuredQueuedMessageControls(args: {
pendingPrompt: boolean
/** The chat shows the agent working: a command card offers no send then. */
agentWorking?: boolean
+ /** The host's next card, and the /clear it holds while a wait ends. */
+ nextQueuedMessageId?: string | null
+ nextQueuedMessageWait?: AgentSessionQueueWait | null
+ /** The strip offers a Stop for the background tasks that /clear waits on. */
+ backgroundTasksStoppable?: boolean
mutate: MobileStructuredAgentMutate
/** The active pane's live composer, Edit's copy target; absent = Edit refuses. False when
* nothing was copied. */
@@ -78,11 +84,17 @@ export function useMobileStructuredQueuedMessageControls(args: {
submissions
} = args
const agentWorking = args.agentWorking === true
+ const backgroundTasksStoppable = args.backgroundTasksStoppable === true
+ const nextQueuedMessageId = args.nextQueuedMessageId ?? null
+ const nextQueuedMessageWait = args.nextQueuedMessageWait ?? null
const cards = useMemo(
() =>
mobileQueuedMessageCards(queuedMessages, submissions, {
pendingPrompt,
agentWorking,
+ nextQueuedMessageId,
+ nextQueuedMessageWait,
+ backgroundTasksStoppable,
agentName: args.agentName,
statedFailures: queuedMessages?.some((draft) => draft.state === 'returned')
? agentSessionVisibleFailureFacts(args.journalItems ?? [])
@@ -91,6 +103,9 @@ export function useMobileStructuredQueuedMessageControls(args: {
}),
[
agentWorking,
+ nextQueuedMessageId,
+ nextQueuedMessageWait,
+ backgroundTasksStoppable,
args.agentName,
args.journalItems,
pendingPrompt,
diff --git a/mobile/src/session/use-mobile-structured-send-with-outcome.ts b/mobile/src/session/use-mobile-structured-send-with-outcome.ts
index c6c8e9604e07..c3072c305344 100644
--- a/mobile/src/session/use-mobile-structured-send-with-outcome.ts
+++ b/mobile/src/session/use-mobile-structured-send-with-outcome.ts
@@ -44,6 +44,8 @@ export function useMobileStructuredSendWithOutcome(args: {
queueCapable: boolean
/** The host holds a /compact sent while the agent works as a card. */
commandsWait: boolean
+ /** The same for a /clear. */
+ clearWaits: boolean
stateRef: { readonly current: StructuredAgentSessionState }
commandPending: { current: boolean }
controller: Pick<
@@ -62,6 +64,7 @@ export function useMobileStructuredSendWithOutcome(args: {
client,
commandPending,
commandsWait,
+ clearWaits,
controller,
enabled,
onSendError,
@@ -112,8 +115,7 @@ export function useMobileStructuredSendWithOutcome(args: {
: null,
// A card waiting on a prompt nothing here can answer would hold it forever.
waitsInLine: (command) =>
- command === 'compact' &&
- commandsWait &&
+ (command === 'clear' ? clearWaits : commandsWait) &&
!pendingPromptsAllUnanswerableHere(stateRef.current.items),
onError: onSendError,
timeoutMs
@@ -143,6 +145,7 @@ export function useMobileStructuredSendWithOutcome(args: {
client,
commandPending,
commandsWait,
+ clearWaits,
controller,
enabled,
onSendError,
diff --git a/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx b/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx
index 79f761f5511e..604dbfb0ff22 100644
--- a/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx
+++ b/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx
@@ -266,6 +266,7 @@ describe("useMobileStructuredAgentSession and the host's Stopping", () => {
questionAnswers: false,
queuedMessages: false,
queuedCommands: false,
+ queuedClear: false,
quietRepeatedStop: false,
statusFeed
},
diff --git a/src/main/claude/__fixtures__/claude-lifecycle-capture-stop-task.jsonl b/src/main/claude/__fixtures__/claude-lifecycle-capture-stop-task.jsonl
new file mode 100644
index 000000000000..694f968cda66
--- /dev/null
+++ b/src/main/claude/__fixtures__/claude-lifecycle-capture-stop-task.jsonl
@@ -0,0 +1,7 @@
+{"at": 0, "kind": "meta", "scenario": "stop-task", "providerSessionId": "00000000-0000-4000-8000-00000000c1a0", "note": "Recorded against Claude CLI 2.1.280 through Orca's adapter: the window around a background shell's Stop at idle, from Orca's stop_task to the CLI idling; the task's launch was not recorded. Scrubbed: session id, frame uuids, tool_use id, output path; get_settings exchange and Orca's later close dropped."}
+{"at": 0, "kind": "control", "request": {"subtype": "stop_task", "task_id": "bhgj726rb"}}
+{"at": 3, "kind": "frame", "frame": {"type": "system", "subtype": "background_tasks_changed", "tasks": [], "uuid": "00000000-0000-4000-8000-000000005701", "session_id": "00000000-0000-4000-8000-00000000c1a0"}}
+{"at": 3, "kind": "frame", "frame": {"type": "system", "subtype": "task_updated", "task_id": "bhgj726rb", "patch": {"status": "killed", "end_time": 1791539307432}, "uuid": "00000000-0000-4000-8000-000000005702", "session_id": "00000000-0000-4000-8000-00000000c1a0"}}
+{"at": 3, "kind": "frame", "frame": {"type": "system", "subtype": "task_notification", "task_id": "bhgj726rb", "tool_use_id": "toolu_bg_sleep", "status": "stopped", "output_file": "/tmp/claude/scrubbed/tasks/bhgj726rb.output", "summary": "Sleep for 10 minutes in background", "uuid": "00000000-0000-4000-8000-000000005703", "session_id": "00000000-0000-4000-8000-00000000c1a0"}}
+{"at": 3, "kind": "control-answer", "response": {}}
+{"at": 53, "kind": "frame", "frame": {"type": "system", "subtype": "session_state_changed", "state": "idle", "uuid": "00000000-0000-4000-8000-000000005704", "session_id": "00000000-0000-4000-8000-00000000c1a0"}}
diff --git a/src/main/claude/claude-background-stop-records.test.ts b/src/main/claude/claude-background-stop-records.test.ts
index 0092d9e2f29d..64fab21c2bf0 100644
--- a/src/main/claude/claude-background-stop-records.test.ts
+++ b/src/main/claude/claude-background-stop-records.test.ts
@@ -64,6 +64,40 @@ describe('a background Stop acts on the host child records', () => {
// The CLI acknowledges a task it no longer runs with no frame: the acknowledgement ends it.
expect(records()).toMatchObject([{ membership: 'settled', outcome: 'cancelled' }])
expect(blocked()).toBeNull()
+ // Nor does it owe an ending of its own, so nothing waits for one.
+ expect(run.adapter.stoppedTaskEndingOwed('session-1')).toBe(false)
+ })
+
+ it('owes the ending of a task it still reports running until that ending lands', async () => {
+ const host = hostWithParent()
+ const run = await producer(host)
+ run.replay(RESUMED_BY_MESSAGE.filter((captured) => captured.at < 5_077))
+ const records = () => host.getStructuredChildWorkViews(parent)
+ expect(run.adapter.stoppedTaskEndingOwed('session-1')).toBe(false)
+
+ await run.adapter.stopBackgroundTasks({
+ sessionId: 'session-1',
+ fence: 7,
+ taskIds: agentChildWorkStopTargets(records(), 'agent-1')
+ })
+ // Settled on the acknowledgement, while the CLI still lists the task as running.
+ expect(records()).toMatchObject([{ membership: 'settled', outcome: 'cancelled' }])
+ expect(run.adapter.stoppedTaskEndingOwed('session-1')).toBe(true)
+
+ run.replay([
+ {
+ at: 7_000,
+ frame: {
+ type: 'system',
+ subtype: 'task_notification',
+ task_id: 'agent-1',
+ status: 'stopped',
+ summary: 'Stopped'
+ }
+ }
+ ])
+ expect(run.adapter.stoppedTaskEndingOwed('session-1')).toBe(false)
+ expect(records()).toMatchObject([{ membership: 'settled', outcome: 'cancelled' }])
})
it('stops every task the strip offers a stop when none is named', async () => {
diff --git a/src/main/claude/claude-child-work-decoder.test.ts b/src/main/claude/claude-child-work-decoder.test.ts
index ae4ca6a9c38c..8c0b76750090 100644
--- a/src/main/claude/claude-child-work-decoder.test.ts
+++ b/src/main/claude/claude-child-work-decoder.test.ts
@@ -259,4 +259,30 @@ describe('Claude children waiting on a permission request', () => {
expect(edges.at(-1)).toBe('session-ended')
expect(edges).not.toContain('agent-bg waiting')
})
+
+ it('owes a stopped task its own ending while the CLI still lists it, until that ending lands', () => {
+ const running = ['agent-bg']
+ const stopped = decoderWith(backgroundAgent)
+ expect(stopped.stopEndingOwed(running)).toBe(false)
+ stopped.stopAcknowledged('agent-bg')
+ expect(stopped.stopEndingOwed(running)).toBe(true)
+ // Acknowledged with no frame: the CLI no longer lists it, so no ending is coming.
+ expect(stopped.stopEndingOwed([])).toBe(false)
+ stopped.observe(system('task_notification', { task_id: 'agent-bg', status: 'stopped' }))
+ expect(stopped.stopEndingOwed(running)).toBe(false)
+
+ const restarted = decoderWith(backgroundAgent)
+ restarted.stopAcknowledged('agent-bg')
+ restarted.observe({ ...backgroundAgent, tool_use_id: 'toolu_bg_2' })
+ expect(restarted.stopEndingOwed(running)).toBe(false)
+
+ const closed = decoderWith(backgroundAgent)
+ closed.stopAcknowledged('agent-bg')
+ closed.clear()
+ expect(closed.stopEndingOwed(running)).toBe(false)
+ // Orca's own close ends what is live, and owes nothing: the session is gone.
+ const proven = decoderWith(backgroundAgent)
+ proven.stopLive()
+ expect(proven.stopEndingOwed(running)).toBe(false)
+ })
})
diff --git a/src/main/claude/claude-child-work-decoder.ts b/src/main/claude/claude-child-work-decoder.ts
index ce975e8d39fa..06ee6980048e 100644
--- a/src/main/claude/claude-child-work-decoder.ts
+++ b/src/main/claude/claude-child-work-decoder.ts
@@ -87,6 +87,8 @@ export class ClaudeChildWorkDecoder {
private readonly live = new Map()
/** Ended task ids, with the spawn call each ended under. */
private readonly ended = new Map()
+ /** Tasks ended on a Stop's acknowledgement whose own ending has not landed yet. */
+ private readonly stopEndingsOwed = new Set()
/** The children a pending permission request blocks, as the caller last derived them. */
private waiting: ReadonlySet = new Set()
private pending: PendingEdge[] = []
@@ -126,9 +128,17 @@ export class ClaudeChildWorkDecoder {
stopAcknowledged(id: string): void {
if (this.live.has(id)) {
this.end(id, 'stopped', { basis: 'stop-acknowledged' })
+ this.stopEndingsOwed.add(id)
}
}
+ /** A task ended on a Stop's acknowledgement that the CLI still lists running (`running`): its
+ * own ending, and the agent's turn answering it, are still to come. One the CLI no longer lists
+ * was acknowledged with no frame, and owes none. */
+ stopEndingOwed(running: readonly string[]): boolean {
+ return running.some((id) => this.stopEndingsOwed.has(id))
+ }
+
/** Which children a pending request blocks, re-derived by the caller before every drain. A
* live child that starts or stops waiting is a live edge of its own. */
observeWaiting(waiting: ReadonlySet): void {
@@ -154,6 +164,7 @@ export class ClaudeChildWorkDecoder {
clear(): void {
this.live.clear()
this.ended.clear()
+ this.stopEndingsOwed.clear()
this.waiting = new Set()
this.pending.push((observedAt) => ({ type: 'session-ended', observedAt }))
}
@@ -255,6 +266,7 @@ export class ClaudeChildWorkDecoder {
return
}
this.ended.delete(id)
+ this.stopEndingsOwed.delete(id)
this.live.set(id, task)
this.report(id, task, facts, restart)
}
@@ -289,6 +301,9 @@ export class ClaudeChildWorkDecoder {
this.ended.delete(oldest)
}
}
+ if (reported.basis === undefined) {
+ this.stopEndingsOwed.delete(id)
+ }
const outcome = claudeChildWorkOutcome(status)
this.pending.push((observedAt) => ({
type: 'ended',
diff --git a/src/main/claude/claude-stop-task-capture.test.ts b/src/main/claude/claude-stop-task-capture.test.ts
new file mode 100644
index 000000000000..30848cfb9750
--- /dev/null
+++ b/src/main/claude/claude-stop-task-capture.test.ts
@@ -0,0 +1,165 @@
+// A real Claude CLI 2.1.280 Stop of a background shell at idle
+// (`__fixtures__/claude-lifecycle-capture-stop-task.jsonl`), replayed through the real adapter and a
+// real hook server: the CLI writes the task's own ending before it acknowledges the Stop, so a
+// /clear waiting on background tasks never waits on an owed ending and may run once the CLI idles.
+
+import { readFileSync } from 'node:fs'
+import { join } from 'node:path'
+import { describe, expect, it, vi } from 'vitest'
+import { agentChildWorkStopTargets } from '../../shared/agent-child-work-stop-targets'
+import { readAgentJournalTurn } from '../../shared/agent-session-turn-record'
+import { queuedClearWait } from '../native-chat/agent-session-wire/structured-conversation-clear'
+import {
+ result,
+ spawn,
+ system,
+ toolResult,
+ type CapturedFrame
+} from './claude-captured-frame-builders.test-fixture'
+import { hostWithParent, parent, producer } from './claude-child-work-producer-harness.test-fixture'
+import { PROVIDER_SESSION_ID } from './claude-structured-session-test-support'
+
+vi.mock('../telemetry/client', () => ({ track: vi.fn() }))
+vi.mock('../telemetry/cohort-classifier', () => ({ getCohortAtEmit: vi.fn(() => ({})) }))
+
+type StopCapture = {
+ providerSessionId: string
+ request: Record
+ /** What the CLI wrote between the request and its answer, then after the answer. */
+ beforeAnswer: CapturedFrame[]
+ afterAnswer: CapturedFrame[]
+}
+
+function isRecord(value: unknown): value is Record {
+ return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+function loadStopCapture(): StopCapture {
+ const path = join(__dirname, '__fixtures__', 'claude-lifecycle-capture-stop-task.jsonl')
+ const capture: StopCapture = {
+ providerSessionId: '',
+ request: {},
+ beforeAnswer: [],
+ afterAnswer: []
+ }
+ let answered = false
+ for (const line of readFileSync(path, 'utf8').trim().split('\n')) {
+ const event: unknown = JSON.parse(line)
+ if (!isRecord(event) || typeof event.at !== 'number') {
+ throw new Error('capture line is not a recorded event')
+ }
+ if (event.kind === 'meta' && typeof event.providerSessionId === 'string') {
+ capture.providerSessionId = event.providerSessionId
+ } else if (event.kind === 'control' && isRecord(event.request)) {
+ capture.request = event.request
+ } else if (event.kind === 'control-answer') {
+ answered = true
+ } else if (event.kind === 'frame' && isRecord(event.frame)) {
+ const frame: unknown = JSON.parse(
+ JSON.stringify(event.frame).replaceAll(capture.providerSessionId, PROVIDER_SESSION_ID)
+ )
+ if (!isRecord(frame)) {
+ throw new Error('mapped frame is not a record')
+ }
+ ;(answered ? capture.afterAnswer : capture.beforeAnswer).push({ at: event.at, frame })
+ } else {
+ throw new Error(`capture line has unknown kind: ${String(event.kind)}`)
+ }
+ }
+ return capture
+}
+
+const TASK = 'bhgj726rb'
+const DESCRIPTION = 'Sleep for 10 minutes in background'
+
+/** The launch the capture starts after, in the captured 2.1.280 background-shell shape. */
+const LAUNCH: CapturedFrame[] = [
+ {
+ at: -30_000,
+ frame: spawn(
+ 'toolu_bg_sleep',
+ 'Bash',
+ { command: 'sleep 600', description: DESCRIPTION, run_in_background: true },
+ null
+ )
+ },
+ {
+ at: -29_990,
+ frame: system('background_tasks_changed', {
+ tasks: [{ task_id: TASK, task_type: 'local_bash', description: DESCRIPTION }]
+ })
+ },
+ {
+ at: -29_990,
+ frame: system('task_started', {
+ task_id: TASK,
+ tool_use_id: 'toolu_bg_sleep',
+ description: DESCRIPTION,
+ is_backgrounded: true,
+ task_type: 'local_bash'
+ })
+ },
+ { at: -29_980, frame: toolResult('toolu_bg_sleep', 'Command running in background', null) },
+ { at: -28_000, frame: result('success') },
+ { at: -28_000, frame: system('session_state_changed', { state: 'idle' }) }
+]
+
+describe('a real Claude Stop of a background shell, with a /clear waiting on it', () => {
+ it('never owes the stopped task’s ending, and frees the /clear once the CLI idles', async () => {
+ const capture = loadStopCapture()
+ expect(capture.request).toEqual({ subtype: 'stop_task', task_id: TASK })
+ const host = hostWithParent()
+ const run = await producer(host)
+ run.replay(LAUNCH)
+ const records = () => host.getStructuredChildWorkViews(parent)
+ const turnStates = () =>
+ [...run.journalItems.values()].flatMap(({ body }) => {
+ const turn = readAgentJournalTurn(body)
+ return turn ? [turn.state] : []
+ })
+ const clearWaits = () =>
+ queuedClearWait(
+ null,
+ records(),
+ run.adapter.backgroundTaskStops('session-1'),
+ run.adapter.stoppedTaskEndingOwed('session-1')
+ )
+ expect(records()).toMatchObject([{ membership: 'live', stoppable: true, providerId: TASK }])
+ expect(clearWaits()).toBe('background-tasks')
+
+ // The CLI writes the roster drop and the task's own ending, then answers the Stop.
+ const owedSamples: boolean[] = []
+ run.claude.routes.stop_task = () => {
+ for (const captured of capture.beforeAnswer) {
+ run.replay([captured])
+ owedSamples.push(run.adapter.stoppedTaskEndingOwed('session-1'))
+ }
+ }
+ const outcome = await run.adapter.stopBackgroundTasks({
+ sessionId: 'session-1',
+ fence: 7,
+ taskIds: agentChildWorkStopTargets(records(), TASK)
+ })
+ expect(outcome).toEqual({ cancelled: true })
+ expect(capture.beforeAnswer.map(({ frame }) => frame.subtype)).toEqual([
+ 'background_tasks_changed',
+ 'task_updated',
+ 'task_notification'
+ ])
+ owedSamples.push(run.adapter.stoppedTaskEndingOwed('session-1'))
+ expect(owedSamples).toEqual([false, false, false, false])
+ // The task reads stopped by its own ending, so the /clear no longer waits on background tasks.
+ expect(records()).toMatchObject([
+ { membership: 'settled', outcome: 'cancelled', providerId: TASK }
+ ])
+ expect(clearWaits()).toBeNull()
+ // The ending opens the turn answering it, which the queue waits out before the /clear runs.
+ expect(turnStates()).toEqual(['completed', 'running'])
+
+ run.replay(capture.afterAnswer)
+ expect(run.adapter.stoppedTaskEndingOwed('session-1')).toBe(false)
+ expect(clearWaits()).toBeNull()
+ // The CLI's idle completes it, so the clear's close cuts no turn: no interrupted-response line.
+ expect(turnStates()).toEqual(['completed', 'completed'])
+ })
+})
diff --git a/src/main/claude/claude-structured-event-delivery.ts b/src/main/claude/claude-structured-event-delivery.ts
index 9e543a37a658..6e2203fa847a 100644
--- a/src/main/claude/claude-structured-event-delivery.ts
+++ b/src/main/claude/claude-structured-event-delivery.ts
@@ -26,7 +26,7 @@ export function emitClaudeStructuredSessionEvent({
deps,
publishChildWork
}: ClaudeEventDelivery): void {
- // Host evidence follows the journal; the tracker's roster remains available to contract tests.
+ // Host evidence follows the journal; the tracker also tells whether a Stop's ending is owed.
if (event.type === 'ended') {
session?.childWork.clear()
session?.backgroundTasks.clear()
diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts
index 6f8bc901989e..19d5f5611b0a 100644
--- a/src/main/claude/claude-structured-session-adapter.ts
+++ b/src/main/claude/claude-structured-session-adapter.ts
@@ -200,9 +200,11 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
timeoutMs: this.deps.requestTimeoutMs,
publishChildWork: (session) => this.publishChildWork(input.sessionId, session)
})
- /** The tracker's own roster, for the tests that compare it with the host's child records. No
- * production code reads it: what runs, what a Stop reaches and what blocks a command are all
- * read from the host's child records. */
+ stoppedTaskEndingOwed = (sessionId: string): boolean => {
+ const session = this.sessions.get(sessionId)
+ return session?.childWork.stopEndingOwed(session.backgroundTasks.stoppableTaskIds) ?? false
+ }
+ /** For tests; production reads the tracker only to tell whether a Stop's own ending is owed. */
backgroundTaskState = (sessionId: string): AgentSessionBackgroundTaskState | null | undefined => {
const session = this.sessions.get(sessionId)
return session ? backgroundTaskState(session) : undefined
diff --git a/src/main/native-chat/agent-session-journal/journal-context-clear.ts b/src/main/native-chat/agent-session-journal/journal-context-clear.ts
index 8630fe4d704f..ba090054c1f6 100644
--- a/src/main/native-chat/agent-session-journal/journal-context-clear.ts
+++ b/src/main/native-chat/agent-session-journal/journal-context-clear.ts
@@ -5,6 +5,7 @@ import type { JournalOperationReceipt, JournalRowWriter } from './journal-row-wr
import type { JournalQueuedMessages } from './journal-queued-messages'
import type { JournalTombstoneRow } from './journal-row-schema'
import { buildJournalQueueClearRow } from './journal-stop-and-resume-rows'
+import { QueuedMessageNotConsumableError } from './queued-message-consume-error'
export function appendJournalContextClear(input: {
state: () => JournalReducerState
@@ -13,18 +14,28 @@ export function appendJournalContextClear(input: {
boundary: AgentSessionProviderContextBoundary
receipt: JournalOperationReceipt
settledByOp: string
+ /** A /clear card the queue runs: spent here, and only the cards queued before it were written
+ * for the context it discards. Those behind it were sent for the fresh one and stay as they are. */
+ ranFrom?: string
}) {
- const { boundary, state } = input
+ const { boundary, state, ranFrom } = input
let commandIds: string[] = []
return input.writer
.enqueueRows(
() => {
- const cards = input.cards.list()
+ const all = input.cards.list()
+ const at =
+ ranFrom === undefined ? all.length : all.findIndex((c) => c.messageId === ranFrom)
+ if (at < 0) {
+ throw new QueuedMessageNotConsumableError(ranFrom ?? '', 'waiting')
+ }
+ const cards = all.slice(0, at)
commandIds = cards
.filter(
(card) => card.body.command && (card.state === 'waiting' || card.state === 'returned')
)
.map((card) => card.messageId)
+ .concat(ranFrom === undefined ? [] : [ranFrom])
const messageIds = cards
.filter((card) => card.state === 'waiting' && !card.body.command)
.map((card) => card.messageId)
@@ -52,10 +63,14 @@ export function appendJournalContextClear(input: {
},
{
write: (db) => {
- input.cards.withdrawInTransaction(db, {
+ const withdrawn = input.cards.withdrawInTransaction(db, {
messageIds: commandIds,
settledByOp: input.settledByOp
})
+ // The card and its clear commit together or not at all, so it can never run twice.
+ if (ranFrom !== undefined && !withdrawn.includes(ranFrom)) {
+ throw new QueuedMessageNotConsumableError(ranFrom, 'waiting')
+ }
input.receipt.write(db)
},
committed: input.receipt.committed
diff --git a/src/main/native-chat/agent-session-journal/journal-context-controller.ts b/src/main/native-chat/agent-session-journal/journal-context-controller.ts
index 7403760e8e91..d37ab0e7530d 100644
--- a/src/main/native-chat/agent-session-journal/journal-context-controller.ts
+++ b/src/main/native-chat/agent-session-journal/journal-context-controller.ts
@@ -26,9 +26,16 @@ export class JournalContextController {
clear(
boundary: AgentSessionProviderContextBoundary,
receipt: JournalOperationReceipt,
- settledByOp: string
+ settledByOp: string,
+ ranFrom?: string
) {
- return appendJournalContextClear({ ...this.deps, boundary, receipt, settledByOp })
+ return appendJournalContextClear({
+ ...this.deps,
+ boundary,
+ receipt,
+ settledByOp,
+ ...(ranFrom === undefined ? {} : { ranFrom })
+ })
}
rewind(
diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts
index 3337727689cd..9a1658064150 100644
--- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts
+++ b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts
@@ -243,20 +243,20 @@ export class JournalQueuedMessages {
withdrawInTransaction(
db: Database.Database,
input: { messageIds: readonly string[]; settledByOp: string }
- ): void {
+ ): string[] {
if (this.deps.database().db !== db) {
throw new AgentSessionJournalError('journal_closed', 'withdraw crossed database handles')
}
- this.changeRevision += withdrawQueuedMessages(db, {
- ...input,
- sessionId: this.deps.sessionId,
- now: this.deps.now()
- }).length
+ const { sessionId, now } = this.deps
+ const withdrawn = withdrawQueuedMessages(db, { ...input, sessionId, now: now() })
+ this.changeRevision += withdrawn.length
+ return withdrawn.map((row) => row.messageId)
}
/** One standalone draft-table transaction on the journal's queue; one that
- * changed rows bumps the revision and notifies after COMMIT, `adopted` first. */
- private transact(
+ * changed rows bumps the revision and notifies after COMMIT, `adopted` first. Open to the
+ * draft-table writers outside this file (`queued-message-return.ts`), which take no other path. */
+ transact(
run: (db: Database.Database) => JournalWriteResult,
changed: (result: T) => boolean,
adopted?: () => void
diff --git a/src/main/native-chat/agent-session-journal/queued-message-return.ts b/src/main/native-chat/agent-session-journal/queued-message-return.ts
new file mode 100644
index 000000000000..fc2d30beb3c2
--- /dev/null
+++ b/src/main/native-chat/agent-session-journal/queued-message-return.ts
@@ -0,0 +1,43 @@
+// A card the host runs itself (a queued /clear) that could not run: returned with why, never handed
+// off. Like a refused hand-off's card it blocks the cards behind it until its own Send or Delete.
+
+import type Database from '../../sqlite/sync-database'
+import type { UnreadAgentSessionFailureFact } from '../../../shared/agent-session-failure'
+import type { JournalQueuedMessages } from './journal-queued-messages'
+
+/** waiting → returned with no hand-off; a returned card asked again keeps its place with the newer
+ * reason. False when the card is no longer unsettled. */
+export function returnUnsentQueuedCard(
+ queued: Pick,
+ input: {
+ messageId: string
+ reason: string | null
+ rejection: UnreadAgentSessionFailureFact
+ now: number
+ }
+): Promise {
+ return queued.transact(
+ (db) => returnUnsentQueuedMessage(db, { ...input, sessionId: queued.sessionId }),
+ (changed) => changed
+ )
+}
+
+function returnUnsentQueuedMessage(
+ db: Database.Database,
+ input: {
+ sessionId: string
+ messageId: string
+ reason: string | null
+ rejection: UnreadAgentSessionFailureFact
+ now: number
+ }
+): boolean {
+ const changed = db
+ .prepare(
+ `UPDATE queued_messages
+ SET state = 'returned', hold_reason = NULL, returned_reason = ?, returned_rejection = ?, settled_at = ?
+ WHERE session_id = ? AND message_id = ? AND state IN ('waiting', 'returned')`
+ )
+ .run(input.reason, JSON.stringify(input.rejection), input.now, input.sessionId, input.messageId)
+ return Number(changed.changes ?? 0) > 0
+}
diff --git a/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts b/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts
index 07e197ceb6a8..18ed810a83a0 100644
--- a/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts
+++ b/src/main/native-chat/agent-session-wire/agent-session-subscriber-frame-fields.ts
@@ -69,7 +69,8 @@ export function buildSubscriberFrame(
? {
queuedMessages: queued.queuedMessages,
queuePause: queued.queuePause,
- nextQueuedMessageId: queued.nextQueuedMessageId
+ nextQueuedMessageId: queued.nextQueuedMessageId,
+ nextQueuedMessageWait: queued.nextQueuedMessageWait
}
: {}),
...(backgroundTasks !== undefined ? { backgroundTasks } : {})
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts
index ee8e13847502..02752ff2c792 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts
@@ -121,6 +121,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi
sessionId
) => this.liveOwnerOrNull(sessionId)?.backgroundTaskStops?.(sessionId)
+ stoppedTaskEndingOwed = (sessionId: string): boolean =>
+ this.liveOwnerOrNull(sessionId)?.stoppedTaskEndingOwed?.(sessionId) ?? false
+
holdsDispatch = (sessionId: string): boolean =>
this.liveOwnerOrNull(sessionId)?.holdsDispatch?.(sessionId) ?? false
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts
index d753547d57ee..110e360e50d7 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts
@@ -375,6 +375,9 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop &
/** The stops this provider honours for a live session's background work; undefined when the
* adapter holds no live session for it. */
backgroundTaskStops?(sessionId: string): AgentSessionBackgroundTaskStops | undefined
+ /** The provider acknowledged a background Stop ahead of the task's own ending, which it still
+ * owes: the host's record already reads the task settled. False with no live session. */
+ stoppedTaskEndingOwed?(sessionId: string): boolean
/** The provider reported taking a send it has neither answered nor ended, as a queued follow-up
* or a silent retry does. Derived from the live child; false with none. */
holdsDispatch?(sessionId: string): boolean
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts
index cf44fc3d66b5..a514690e7f89 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-background-task-channel.ts
@@ -54,7 +54,12 @@ export class StructuredAgentSessionBackgroundTaskChannel {
const backgroundTasks = this.read(request.sessionId)
const queue = tryReadQueuePublication(
journal,
- structuredQueueSendGate(this.deps.store, request.sessionId)
+ structuredQueueSendGate(
+ this.deps.store,
+ request.sessionId,
+ this.readChildWork,
+ this.deps.adapter
+ )
)
const hostNow = this.deps.now?.() ?? Date.now()
return {
@@ -68,7 +73,8 @@ export class StructuredAgentSessionBackgroundTaskChannel {
? {
queuedMessages: queue.queuedMessages,
queuePause: queue.queuePause,
- nextQueuedMessageId: queue.nextQueuedMessageId
+ nextQueuedMessageId: queue.nextQueuedMessageId,
+ nextQueuedMessageWait: queue.nextQueuedMessageWait
}
: {}),
backgroundTasks
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts
index c369ce1a5602..dbfd58223b86 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts
@@ -64,7 +64,12 @@ export class StructuredAgentSessionClientDelivery {
readQueuePublication: (sessionId) =>
tryReadQueuePublication(
sessions.get(sessionId)?.journal,
- structuredQueueSendGate(this.deps().store, sessionId)
+ structuredQueueSendGate(
+ this.deps().store,
+ sessionId,
+ (id) => this.readChildWork(id),
+ this.deps().adapter
+ )
),
readBackgroundTasks,
onJournalPublished: (sessionId, journal) => this.publishJournal(sessionId, journal)
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts
index 251ccc88e32b..634d9e5d7cf3 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts
@@ -78,7 +78,7 @@ export class StructuredAgentSessionHost {
() => this.deps,
(sessionId) => this.queued.onJournalActivity(sessionId),
(sessionId) => this.restartResume.onAgentStarted(sessionId),
- (sessionId) => this.backgroundTasks.publish(sessionId),
+ (sessionId) => this.queued.onChildWorkChanged(sessionId, this.backgroundTasks),
(sessionId) => this.backgroundTasks.read(sessionId)
)
private readonly subscribers = this.clientDelivery.subscribers
@@ -97,6 +97,7 @@ export class StructuredAgentSessionHost {
// Every collaborator reads this copy, so a logger that throws cannot fail what it reports.
this.deps = deps = sessionLogger.withNeverThrowingLogger(deps)
this.clientDelivery.watchAtRestCommands(deps.adapter)
+ this.queued.wakeOnHandoffEnded(deps.store)
this.backgroundTasks = new StructuredAgentSessionBackgroundTaskChannel(
deps,
this.sessions,
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts
index 0feaf809f2a5..8183593e71b0 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts
@@ -567,10 +567,16 @@ describe('where the host would refuse the send', () => {
it('a card queued after clear is sendable with the completed clear record still present', async () => {
const { card, journal, record, fence } = await cardAddedAfterClear()
- const gate = structuredQueueSendGate(rig.store, HOST_TEST_SESSION)
+ const gate = structuredQueueSendGate(rig.store, HOST_TEST_SESSION, () => undefined, {})
expect(readQueuePublication(journal, gate).nextQueuedMessageId).toBe(card)
const { conversationCommand: _cleared, ...unblocked } = record
- const next = readQueuePublication(journal, () => ({ record: unblocked, fence }))
+ const next = readQueuePublication(journal, () => ({
+ record: unblocked,
+ fence,
+ childWork: () => undefined,
+ backgroundTaskStops: () => undefined,
+ stoppedTaskEndingOwed: () => false
+ }))
expect(next.nextQueuedMessageId).toBe(card)
})
@@ -586,7 +592,13 @@ describe('where the host would refuse the send', () => {
retained: []
}
const next = (gateRecord: typeof record) =>
- readQueuePublication(journal, () => ({ record: gateRecord, fence })).nextQueuedMessageId
+ readQueuePublication(journal, () => ({
+ record: gateRecord,
+ fence,
+ childWork: () => undefined,
+ backgroundTaskStops: () => undefined,
+ stoppedTaskEndingOwed: () => false
+ })).nextQueuedMessageId
expect(next({ ...unblocked, rewind })).toBeNull()
expect(next({ ...unblocked, rewind: { ...rewind, phase: 'provider-succeeded' } })).toBeNull()
expect(next({ ...unblocked, rewind: { ...rewind, phase: 'completed' } })).not.toBeNull()
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts
new file mode 100644
index 000000000000..df02cf87b8a9
--- /dev/null
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts
@@ -0,0 +1,546 @@
+// A /clear asked to wait (`delivery`) while the agent works: held as a card like a queued send,
+// answered at once, and run by the host itself when its turn comes, in the same conversation. The
+// cards sent after it run in the fresh context, in order. Without the opt-in, today's refusal.
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view'
+import type { AgentSessionBackgroundTaskStops } from '../../../shared/agent-child-work-stop-targets'
+import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger'
+import {
+ createQueuedMessageTestRig,
+ eventually,
+ QUEUED_RIG_CALLER as CALLER,
+ type QueuedMessageTestRig
+} from './structured-agent-session-queued-message-rig.test-fixture'
+import {
+ HOST_TEST_SESSION as SESSION,
+ hostTestOperationId
+} from './structured-agent-session-host-test-data'
+import { QUEUED_CLEAR_CALLER_KEY } from './structured-conversation-clear'
+import { openRigTurnFor } from './structured-agent-session-queued-rig-turn.test-fixture'
+
+let rig: QueuedMessageTestRig
+
+beforeEach(async () => {
+ rig = await createQueuedMessageTestRig()
+})
+
+afterEach(() => rig.dispose())
+
+function clear(delivery?: 'queue-if-active', clientOperationId = hostTestOperationId()) {
+ const fields = { command: 'clear' as const, ...(delivery ? { delivery } : {}) }
+ return rig.host.conversationCommand(CALLER, {
+ envelope: rig.envelope(fields, 'agentSession.conversationCommand', clientOperationId),
+ ...fields,
+ userSend: true
+ })
+}
+
+async function queuedClear(): Promise {
+ const id = hostTestOperationId()
+ expect(await clear('queue-if-active', id)).toMatchObject({
+ ok: true,
+ value: { command: 'clear', state: 'completed', queued: { messageId: id, state: 'waiting' } }
+ })
+ return id
+}
+
+async function queuedSend(text: string): Promise {
+ const sent = await rig.send(text, 'queue-if-active').result
+ if (!sent.ok || !('queued' in sent.value)) {
+ throw new Error('expected a queued receipt')
+ }
+ return sent.value.queued.messageId
+}
+
+function journal() {
+ return rig.host.collaboratorsForTests().sessions.get(SESSION)!.journal
+}
+
+async function dividers(): Promise {
+ return (await rig.host.journalSnapshot(SESSION)).items.filter(
+ (item) => item.body.kind === 'status' && item.body.presentation === 'context-cleared'
+ ).length
+}
+
+const BACKGROUND_TASK: AgentChildWorkView = {
+ id: 'child-dev',
+ providerId: 'task-dev',
+ kind: 'command',
+ description: 'npm run dev',
+ state: 'working',
+ membership: 'live',
+ firstObservedAt: 1,
+ observedAt: 1,
+ stoppable: true,
+ invocation: { invocationId: 'spawn-dev', generation: 1 }
+}
+
+const settleMs = () => new Promise((resolve) => setTimeout(resolve, 150))
+
+describe('a /clear that waits in line', () => {
+ it('behind an unanswered message: a card at once, nothing stopped or cleared yet', async () => {
+ await rig.workingSend()
+ const clearId = await queuedClear()
+ expect(await rig.drafts()).toEqual([{ messageId: clearId, state: 'waiting' }])
+ expect(journal().queuedMessages.get(clearId)?.body).toMatchObject({
+ blocks: [{ type: 'text', text: '/clear' }],
+ command: { name: 'clear' }
+ })
+ await settleMs()
+ expect(rig.closeSession).not.toHaveBeenCalled()
+ expect(rig.store.getRecord(SESSION)?.providerContextBoundary).toBeUndefined()
+ expect(await dividers()).toBe(0)
+ })
+
+ it('runs in the same chat once the turn ends: one divider, spent in that write, never sent', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () => expect(await dividers()).toBe(1))
+ expect(rig.closeSession).toHaveBeenCalled()
+ expect(rig.store.getRecord(SESSION)).toMatchObject({
+ providerHandleChain: [],
+ conversationCommand: {
+ command: 'clear',
+ operationId: clearId,
+ callerKey: QUEUED_CLEAR_CALLER_KEY,
+ phase: 'committed'
+ }
+ })
+ expect(journal().queuedMessages.get(clearId)).toMatchObject({
+ state: 'withdrawn',
+ settledByOp: agentSessionOperationKey(QUEUED_CLEAR_CALLER_KEY, clearId)
+ })
+ expect(await rig.drafts()).toEqual([])
+ expect(await rig.handoff(clearId)).toBeUndefined()
+ expect(rig.dispatch).toHaveBeenCalledOnce()
+ })
+
+ it('cards sent after it run in the fresh context, in order, unpaused', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ const first = await queuedSend('first after the clear')
+ const second = await queuedSend('second after the clear')
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () => expect(await rig.handoff(first)).toBeDefined())
+ expect(await dividers()).toBe(1)
+ expect(await rig.queuePause()).toBeNull()
+ expect(journal().queuedMessages.pauses()).toEqual([])
+ // The first is the agent's work now, so the second waits for it.
+ await settleMs()
+ expect(await rig.handoff(second)).toBeUndefined()
+ expect(await rig.drafts()).toEqual([{ messageId: second, state: 'waiting' }])
+ const sent = (await rig.host.journalSnapshot(SESSION)).submissions
+ expect(sent.flatMap((entry) => (entry.queuedMessageId ? [entry.queuedMessageId] : []))).toEqual(
+ [first]
+ )
+ expect(sent.some((entry) => entry.queuedMessageId === clearId)).toBe(false)
+ })
+
+ it('a resent id answers from its card, and after it ran as withdrawn; one clear', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ expect(await clear('queue-if-active', clearId)).toMatchObject({
+ ok: true,
+ value: { queued: { messageId: clearId, state: 'waiting' } }
+ })
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () => expect(await dividers()).toBe(1))
+ expect(await clear('queue-if-active', clearId)).toMatchObject({
+ ok: true,
+ value: { command: 'clear', queued: { messageId: clearId, state: 'withdrawn' } }
+ })
+ await settleMs()
+ expect(await dividers()).toBe(1)
+ })
+
+ it('a restart after it ran never runs it again', async () => {
+ const working = await rig.workingSend()
+ await queuedClear()
+ const after = await queuedSend('after the clear')
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () => expect(await rig.handoff(after)).toBeDefined())
+ await rig.quitRestartHostProcess()
+ await rig.host.journalSnapshot(SESSION)
+ await settleMs()
+ expect(await dividers()).toBe(1)
+ })
+
+ it('never steers: its Send mid-turn is refused and it keeps waiting', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ expect(await rig.sendNow(clearId)).toMatchObject({
+ ok: false,
+ refusal: { message: "A command can't be sent while the agent is working." }
+ })
+ expect(await dividers()).toBe(0)
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () => expect(await dividers()).toBe(1))
+ })
+
+ it('Delete takes it back: it never runs, and the cards behind it send as usual', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ const after = await queuedSend('after')
+ expect(await rig.deleteQueued(clearId)).toMatchObject({ ok: true, value: { deleted: true } })
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () => expect(await rig.handoff(after)).toBeDefined())
+ expect(await dividers()).toBe(0)
+ })
+
+ it('at rest, clears at once as before', async () => {
+ const answer = await clear('queue-if-active')
+ expect(answer).toMatchObject({ ok: true, value: { command: 'clear', state: 'completed' } })
+ expect(answer.ok && answer.value.queued).toBeFalsy()
+ expect(await dividers()).toBe(1)
+ expect(await rig.drafts()).toEqual([])
+ })
+
+ it('without the opt-in (an older client), is refused while a message is unanswered, as today', async () => {
+ await rig.workingSend()
+ expect(await clear()).toMatchObject({
+ ok: false,
+ refusal: { details: { reason: 'messagesUnsettled' } }
+ })
+ expect(await rig.drafts()).toEqual([])
+ })
+})
+
+/** The provider's background Stop: Claude's strip offers one; Codex's and ACP's offer none. */
+const STOPS_ALL: AgentSessionBackgroundTaskStops = { supportsTaskStop: true, supportsStopAll: true }
+const NO_STOP: AgentSessionBackgroundTaskStops = { supportsTaskStop: false, supportsStopAll: false }
+
+function withStops(stops: AgentSessionBackgroundTaskStops): void {
+ Object.assign(rig.host.deps.adapter, { backgroundTaskStops: () => stops })
+}
+
+/** A turn that ended leaving background tasks running, the agent idle. */
+async function withBackgroundTasks(
+ stops = STOPS_ALL
+): Promise<{ set: (next: AgentChildWorkView[]) => void }> {
+ let tasks: AgentChildWorkView[] = []
+ withStops(stops)
+ Object.assign(rig.host.deps, {
+ statusSink: { publish: () => {}, forget: () => {}, readChildWork: () => tasks }
+ })
+ const working = await rig.workingSend()
+ tasks = [BACKGROUND_TASK]
+ await rig.settleAccepted(working, 'a')
+ return { set: (next) => (tasks = next) }
+}
+
+const SETTLED_TASK: AgentChildWorkView = {
+ ...BACKGROUND_TASK,
+ state: 'done',
+ membership: 'settled',
+ outcome: 'succeeded'
+}
+
+/** A Stop the provider acknowledged ahead of the task's own ending, which it then delivers. */
+function stoppedTaskEnding() {
+ let owed = false
+ Object.assign(rig.host.deps.adapter, { stoppedTaskEndingOwed: () => owed })
+ return {
+ acknowledge(tasks: { set: (next: AgentChildWorkView[]) => void }) {
+ owed = true
+ tasks.set([{ ...BACKGROUND_TASK, state: 'done', membership: 'settled', outcome: 'cancelled' }])
+ },
+ endingLands() {
+ owed = false
+ }
+ }
+}
+
+describe('a /clear card the queue cannot run yet', () => {
+ it('typed while idle with background tasks running, waits as a card too, then runs', async () => {
+ const tasks = await withBackgroundTasks()
+ const clearId = await queuedClear()
+ expect(await dividers()).toBe(0)
+ const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' })
+ expect(page.ok && page.page.nextQueuedMessageWait).toEqual({
+ messageId: clearId,
+ reason: 'background-tasks'
+ })
+ // A message sent now joins the queue behind it, for the cleared chat.
+ const after = await queuedSend('after the clear')
+ tasks.set([SETTLED_TASK])
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await eventually(async () => expect(await rig.handoff(after)).toBeDefined())
+ expect(await dividers()).toBe(1)
+ })
+
+ it('child-work updates while it waits run no drain step', async () => {
+ let tasks: AgentChildWorkView[] = []
+ withStops(STOPS_ALL)
+ Object.assign(rig.host.deps, {
+ statusSink: { publish: () => {}, forget: () => {}, readChildWork: () => tasks }
+ })
+ const working = await rig.workingSend()
+ await queuedClear()
+ tasks = [BACKGROUND_TASK]
+ await rig.settleAccepted(working, 'a')
+ await settleMs()
+ const snapshot = vi.spyOn(journal(), 'snapshot')
+ for (let tick = 0; tick < 5; tick += 1) {
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ }
+ await settleMs()
+ expect(snapshot).not.toHaveBeenCalled()
+ expect(await dividers()).toBe(0)
+ })
+
+ it('held by a reopen, publishes no next card and no wait, and its Send runs it', async () => {
+ await rig.workingSend()
+ const clearId = await queuedClear()
+ await rig.quitRestartHostProcess()
+ const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' })
+ expect(page.ok && page.page.nextQueuedMessageId).toBeNull()
+ expect(page.ok && page.page.nextQueuedMessageWait).toBeNull()
+ expect(await rig.drafts()).toEqual([{ messageId: clearId, state: 'waiting' }])
+ expect(await rig.sendNow(clearId)).toMatchObject({
+ ok: true,
+ value: { queued: { messageId: clearId, state: 'withdrawn' } }
+ })
+ expect(await dividers()).toBe(1)
+ })
+
+ it('waits out background tasks, then runs when they end; the cards behind it wait too', async () => {
+ let tasks: AgentChildWorkView[] = []
+ withStops(STOPS_ALL)
+ Object.assign(rig.host.deps, {
+ statusSink: { publish: () => {}, forget: () => {}, readChildWork: () => tasks }
+ })
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ const after = await queuedSend('after the clear')
+ tasks = [BACKGROUND_TASK]
+ await rig.settleAccepted(working, 'a')
+ await settleMs()
+ expect(await rig.drafts()).toEqual([
+ { messageId: clearId, state: 'waiting' },
+ { messageId: after, state: 'waiting' }
+ ])
+ expect(await dividers()).toBe(0)
+ // Not named as the queue's next send while it waits, so no client reads the chat as busy; the
+ // wait is published instead, so clients read it rather than guess.
+ const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' })
+ expect(page.ok && page.page.nextQueuedMessageId).toBeNull()
+ expect(page.ok && page.page.nextQueuedMessageWait).toEqual({
+ messageId: clearId,
+ reason: 'background-tasks'
+ })
+ // Its Send says why it waits, and changes nothing on the card.
+ expect(await rig.sendNow(clearId)).toMatchObject({
+ ok: false,
+ refusal: { details: { reason: 'backgroundTasksRunning' } }
+ })
+ tasks = [{ ...BACKGROUND_TASK, state: 'done', membership: 'settled', outcome: 'succeeded' }]
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await eventually(async () => expect(await rig.handoff(after)).toBeDefined())
+ expect(await dividers()).toBe(1)
+ })
+
+ it("after the strip's Stop, waits for the stopped task's own ending, not the acknowledgement", async () => {
+ const tasks = await withBackgroundTasks()
+ const stopped = stoppedTaskEnding()
+ const clearId = await queuedClear()
+ // Claude answers the Stop before the task's own "stopped" lands; the record already reads settled.
+ stopped.acknowledge(tasks)
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await settleMs()
+ expect(await dividers()).toBe(0)
+ expect(rig.closeSession).not.toHaveBeenCalled()
+ const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' })
+ expect(page.ok && page.page.nextQueuedMessageWait).toEqual({
+ messageId: clearId,
+ reason: 'background-tasks'
+ })
+ stopped.endingLands()
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await eventually(async () => expect(await dividers()).toBe(1))
+ })
+
+ it('after the stopped task ends, waits out the turn the agent answers it in', async () => {
+ const tasks = await withBackgroundTasks()
+ const stopped = stoppedTaskEnding()
+ await queuedClear()
+ stopped.acknowledge(tasks)
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await settleMs()
+ // The task's own ending opens the agent's turn answering it.
+ stopped.endingLands()
+ await openRigTurnFor(rig, 'task-ended')
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await settleMs()
+ expect(await dividers()).toBe(0)
+ expect(rig.closeSession).not.toHaveBeenCalled()
+ await openRigTurnFor(rig, 'task-ended', 'completed')
+ await eventually(async () => expect(await dividers()).toBe(1))
+ const turns = (await rig.host.journalSnapshot(SESSION)).items.flatMap((item) =>
+ item.body.kind === 'turn' && item.body.turnId === 'turn-of-task-ended' ? [item.body.state] : []
+ )
+ expect(turns).toEqual(['completed'])
+ })
+
+ it('a failed commit returns it with why once; nothing behind it runs; Send retries', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ const after = await queuedSend('after the clear')
+ vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+ vi.spyOn(rig.store.conversationReceipts, 'queuedClear').mockReturnValueOnce({
+ write: () => {
+ throw new Error('disk full')
+ },
+ committed: () => {}
+ })
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () =>
+ expect(await rig.drafts()).toEqual([
+ { messageId: clearId, state: 'returned' },
+ { messageId: after, state: 'waiting' }
+ ])
+ )
+ // Rolled back whole: no divider, no fresh context, and the card was not spent.
+ expect(await dividers()).toBe(0)
+ expect(rig.store.getRecord(SESSION)?.providerContextBoundary).toBeUndefined()
+ expect(journal().queuedMessages.get(clearId)).toMatchObject({
+ state: 'returned',
+ returnedRejection: { kind: 'commandRefused' }
+ })
+ await settleMs()
+ expect(await rig.handoff(after)).toBeUndefined()
+
+ expect(await rig.sendNow(clearId)).toMatchObject({
+ ok: true,
+ value: { queued: { messageId: clearId, state: 'withdrawn' } }
+ })
+ expect(await dividers()).toBe(1)
+ await eventually(async () => expect(await rig.handoff(after)).toBeDefined())
+ })
+
+ it('a returned /clear deleted: the cards behind it run in the uncleared chat', async () => {
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ const after = await queuedSend('after the clear')
+ vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+ vi.spyOn(rig.store.conversationReceipts, 'queuedClear').mockReturnValueOnce({
+ write: () => {
+ throw new Error('disk full')
+ },
+ committed: () => {}
+ })
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () =>
+ expect(journal().queuedMessages.get(clearId)?.state).toBe('returned')
+ )
+ expect(await rig.deleteQueued(clearId)).toMatchObject({ ok: true, value: { deleted: true } })
+ await eventually(async () => expect(await rig.handoff(after)).toBeDefined())
+ expect(await dividers()).toBe(0)
+ })
+
+ it('its Send over a paused queue holds the cards queued before it; Resume sends them', async () => {
+ const working = await rig.workingSend()
+ const before = await queuedSend('before the clear')
+ const clearId = await queuedClear()
+ await rig.stop()
+ const after = await queuedSend('after the clear')
+ await rig.settleAccepted(working, 'a')
+ expect(await rig.sendNow(clearId)).toMatchObject({ ok: true })
+ expect(await dividers()).toBe(1)
+ expect(journal().queuedMessages.pauses()).toContainEqual(
+ expect.objectContaining({ reason: 'cleared', messageIds: [before] })
+ )
+ // The queue never reorders: the card after it waits behind the one it holds.
+ await settleMs()
+ expect(await rig.handoff(before)).toBeUndefined()
+ expect(await rig.handoff(after)).toBeUndefined()
+ expect(await rig.resume()).toMatchObject({ ok: true, value: { resumed: true } })
+ await eventually(async () => expect(await rig.handoff(before)).toBeDefined())
+ })
+})
+
+describe('a /clear blocked by background tasks nothing here can stop', () => {
+ it('typed while idle, is refused at once as before, and the next message goes straight out', async () => {
+ await withBackgroundTasks(NO_STOP)
+ expect(await clear('queue-if-active')).toMatchObject({
+ ok: false,
+ refusal: {
+ details: { reason: 'backgroundTasksRunning' },
+ message: 'Wait for background tasks to finish before using this command.'
+ }
+ })
+ expect(await rig.drafts()).toEqual([])
+ const next = await rig.send('stop the dev server', 'queue-if-active').result
+ expect(next.ok && 'queued' in next.value).toBe(false)
+ expect(await dividers()).toBe(0)
+ })
+
+ it('a live task the strip has no Stop for refuses it too, though the provider can stop others', async () => {
+ await withBackgroundTasks(STOPS_ALL)
+ Object.assign(rig.host.deps, {
+ statusSink: {
+ publish: () => {},
+ forget: () => {},
+ readChildWork: () => [{ ...BACKGROUND_TASK, stoppable: false }]
+ }
+ })
+ Object.assign(rig.host.deps.adapter, {
+ backgroundTaskStops: () => ({ supportsTaskStop: true, supportsStopAll: false })
+ })
+ expect(await clear('queue-if-active')).toMatchObject({
+ ok: false,
+ refusal: { details: { reason: 'backgroundTasksRunning' } }
+ })
+ expect(await rig.drafts()).toEqual([])
+ })
+
+ it('sent mid-turn, is returned once with why when it reaches the front; nothing waits on it', async () => {
+ let tasks: AgentChildWorkView[] = []
+ withStops(NO_STOP)
+ Object.assign(rig.host.deps, {
+ statusSink: { publish: () => {}, forget: () => {}, readChildWork: () => tasks }
+ })
+ const working = await rig.workingSend()
+ const clearId = await queuedClear()
+ tasks = [BACKGROUND_TASK]
+ await rig.settleAccepted(working, 'a')
+ await eventually(async () =>
+ expect(await rig.drafts()).toEqual([{ messageId: clearId, state: 'returned' }])
+ )
+ expect(journal().queuedMessages.get(clearId)).toMatchObject({
+ returnedReason:
+ 'Background tasks are still running. Wait for the background tasks to finish. Run /clear again.',
+ returnedRejection: {
+ kind: 'commandRefused',
+ refusal: { details: { reason: 'backgroundTasksRunning' } }
+ }
+ })
+ const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' })
+ expect(page.ok && page.page.nextQueuedMessageWait).toBeNull()
+ // Returned once: child-work ticks don't return it again or run it.
+ const reason = journal().queuedMessages.get(clearId)?.returnedReason
+ rig.host.publishChildWorkEvidence(SESSION, [])
+ await settleMs()
+ expect(journal().queuedMessages.get(clearId)?.returnedReason).toBe(reason)
+ // A message typed now isn't held behind it.
+ const next = rig.send('stop the dev server', 'queue-if-active')
+ const sent = await next.result
+ expect(sent.ok && 'queued' in sent.value).toBe(false)
+ await eventually(async () => {
+ const submissions = (await rig.host.journalSnapshot(SESSION)).submissions
+ expect(
+ submissions.find((entry) => entry.clientMessageId === next.id)?.handedOverAt
+ ).toBeDefined()
+ })
+ expect(await dividers()).toBe(0)
+ // Its Send runs it once the work has ended.
+ tasks = [SETTLED_TASK]
+ await rig.settleAccepted(next.id, 'b')
+ expect(await rig.sendNow(clearId)).toMatchObject({
+ ok: true,
+ value: { queued: { messageId: clearId, state: 'withdrawn' } }
+ })
+ expect(await dividers()).toBe(1)
+ })
+})
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts
index ee7a26af161e..9500a8346463 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts
@@ -539,19 +539,11 @@ describe('a /compact that waits in line', () => {
})
})
-it('only a /compact may ask to wait; a /clear sent asking to is refused, never queued', async () => {
+it('either command may ask to wait, only with the one delivery the host knows', () => {
const base = { envelope: rig.envelope({}, 'agentSession.conversationCommand', 'op-schema') }
const parse = (fields: Record) =>
ConversationCommandParams.safeParse({ ...base, ...fields }).success
expect(parse({ command: 'compact', delivery: 'queue-if-active' })).toBe(true)
+ expect(parse({ command: 'clear', delivery: 'queue-if-active' })).toBe(true)
expect(parse({ command: 'compact', delivery: 'now' })).toBe(false)
- await rig.workingSend()
- const fields = { command: 'clear' as const, delivery: 'queue-if-active' as const }
- expect(
- await rig.host.conversationCommand(CALLER, {
- envelope: rig.envelope(fields, 'agentSession.conversationCommand', hostTestOperationId()),
- ...fields
- })
- ).toMatchObject({ ok: false })
- expect(await rig.drafts()).toEqual([])
})
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-drain.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-drain.ts
new file mode 100644
index 000000000000..554f9405f66f
--- /dev/null
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-drain.ts
@@ -0,0 +1,173 @@
+// The serialized drain of mid-turn queueing: one step at a time per conversation, each converting
+// at most one draft into an ordinary submission when the queue's gate lets it (or running a /clear
+// card itself). It holds no loop state: every step re-reads the gate from the journal.
+
+import { randomUUID } from 'node:crypto'
+import { QUEUED_MESSAGE_PAUSED_SEND_FAILED } from '../../../shared/agent-session-wire'
+import type { AgentSessionRecord } from '../../../shared/agent-session-record'
+import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view'
+import type { AgentSessionBackgroundTaskStops } from '../../../shared/agent-child-work-stop-targets'
+import { createStructuredAgentSessionOperationId } from '../../../shared/structured-agent-session-mutation'
+import { QueuedMessageNotConsumableError } from '../agent-session-journal/journal-queued-messages'
+import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table'
+import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
+import type { StructuredAgentSessionLogger } from './structured-agent-session-logger'
+import { structuredAgentSessionHostInstance } from './structured-agent-session-queued-pause'
+import {
+ nextStructuredQueuedMessage,
+ structuredQueueHeadMayRun
+} from './structured-agent-session-queued-messages'
+import { isQueuedClearCard } from './structured-conversation-clear'
+
+export type QueuedMessageDrainDeps = {
+ sessions: ReadonlyMap
+ getRecord: (sessionId: string) => AgentSessionRecord | null
+ serialize: (sessionId: string, task: () => Promise) => Promise
+ conversationFence: (sessionId: string) => number
+ /** The consumed submission is ordinary #22821 work from here on. */
+ wakeDelivery: (sessionId: string) => void
+ logger: StructuredAgentSessionLogger
+ readChildWork: (sessionId: string) => readonly AgentChildWorkView[] | undefined
+ backgroundTaskStops: (sessionId: string) => AgentSessionBackgroundTaskStops | undefined
+ stoppedTaskEndingOwed: (sessionId: string) => boolean
+ /** Runs a /clear card itself, inside the step's serialize (`runQueuedConversationClear`). */
+ runClear: (sessionId: string, card: QueuedMessageRow) => Promise
+}
+
+/**
+ * The serialized drain. Woken by every journal commit (turn, submission, prompt,
+ * command and Stop settlements are all commits), by draft mutations, and by the
+ * conversation opening; each step re-derives everything and consumes at most one
+ * draft — the consumed submission then owes work, which gates the next. A /clear
+ * card is run, never consumed: its commit wakes the step that sends the card behind it.
+ */
+export class StructuredAgentSessionQueuedMessageDrain {
+ private readonly scheduled = new Set()
+ private disposed = false
+
+ constructor(private readonly deps: QueuedMessageDrainDeps) {}
+
+ /** Quit, with delivery: a hand-off made now could only be settled by the next process, so a
+ * quit leaves the cards exactly as a crash does. Read by the step at its start, and again
+ * right before it appends, since quit can land while it awaits. */
+ dispose(): void {
+ this.disposed = true
+ }
+
+ schedule(sessionId: string): void {
+ const journal = this.disposed ? undefined : this.deps.sessions.get(sessionId)?.journal
+ if (!journal) {
+ return
+ }
+ // Cheap pre-check so token streams do not pay a serialized step per delta, nor child-work
+ // updates one per tick while a /clear waits on them. Skipping is safe: whatever ends the work or
+ // the wait is itself a commit or a wake that schedules again, and the step re-reads every gate.
+ try {
+ if (
+ !journal.queuedMessages.settlementOwed() &&
+ !structuredQueueHeadMayRun({
+ journal,
+ record: this.deps.getRecord(sessionId),
+ fence: this.deps.conversationFence(sessionId),
+ childWork: () => this.deps.readChildWork(sessionId),
+ backgroundTaskStops: () => this.deps.backgroundTaskStops(sessionId),
+ stoppedTaskEndingOwed: () => this.deps.stoppedTaskEndingOwed(sessionId)
+ })
+ ) {
+ return
+ }
+ } catch {
+ // The handle is opening or closing; the next commit re-schedules.
+ return
+ }
+ if (this.scheduled.has(sessionId)) {
+ return
+ }
+ this.scheduled.add(sessionId)
+ void this.deps
+ .serialize(sessionId, () => {
+ this.scheduled.delete(sessionId)
+ return this.step(sessionId)
+ })
+ .catch((error: unknown) => {
+ this.scheduled.delete(sessionId)
+ this.deps.logger.warn('draining queued messages failed', {
+ scope: 'queued-drain',
+ sessionId,
+ error
+ })
+ })
+ }
+
+ private async step(sessionId: string): Promise {
+ const session = this.deps.sessions.get(sessionId)
+ if (this.disposed || !session) {
+ return
+ }
+ const journal = session.journal
+ if (journal.queuedMessages.settlementOwed() || journal.queuedMessages.deliveredByEchoOwed()) {
+ // A live per-row hook was skipped; heal now, before a draft sends, rather than at reopen.
+ await journal.queuedMessages.settleOwed().catch((error: unknown) => {
+ this.deps.logger.warn('settling owed queued-message bookkeeping failed', {
+ scope: 'queued-settle-owed',
+ sessionId,
+ error
+ })
+ })
+ }
+ const fence = this.deps.conversationFence(sessionId)
+ // Whatever clears a hold publishes or commits, which re-derives this step.
+ const record = this.deps.getRecord(sessionId)
+ const next = nextStructuredQueuedMessage({
+ journal,
+ record,
+ fence,
+ childWork: () => this.deps.readChildWork(sessionId),
+ backgroundTaskStops: () => this.deps.backgroundTaskStops(sessionId),
+ stoppedTaskEndingOwed: () => this.deps.stoppedTaskEndingOwed(sessionId)
+ })
+ if (this.disposed || !next) {
+ return
+ }
+ if (isQueuedClearCard(next)) {
+ await this.deps.runClear(sessionId, next)
+ return
+ }
+ // Always a fresh id: the submission names its draft by `queuedMessageId`, never by id equality.
+ const submissionId = createStructuredAgentSessionOperationId(randomUUID)
+ try {
+ await journal.appendSubmission(
+ {
+ clientMessageId: submissionId,
+ // The queue's own automatic send, never kept as a card by a restart or a close.
+ origin: 'host',
+ payloadFingerprint: next.fingerprint,
+ body: next.body,
+ fence,
+ handoverRecorded: true
+ },
+ {
+ messageId: next.messageId,
+ expect: 'waiting',
+ settledByOp: null,
+ hostInstance: structuredAgentSessionHostInstance(),
+ yieldsToPause: true
+ }
+ )
+ } catch (error) {
+ if (error instanceof QueuedMessageNotConsumableError) {
+ // Lost a race with a Send-now, a Delete or a Stop; their transition stands.
+ return
+ }
+ // Pre-consume failure: the draft stays waiting, held with the marker on
+ // the card (a stored fact, so it survives eviction and restart). The
+ // hold's own commit notification publishes it. An explicit Send retries;
+ // no automatic retry loop.
+ await journal.queuedMessages
+ .hold({ messageIds: [next.messageId], reason: QUEUED_MESSAGE_PAUSED_SEND_FAILED })
+ .catch(() => {})
+ throw error
+ }
+ this.deps.wakeDelivery(sessionId)
+ }
+}
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts
index 249396d38b5d..da9af8538d83 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts
@@ -1,19 +1,17 @@
// Mid-turn queueing: the accept decision that turns a send into a host-held
-// draft, the serialized drain that converts one draft into an ordinary
-// submission when the session stops owing work, and the published draft list.
+// draft, and the one gate the serialized drain (`structured-agent-session-queued-drain.ts`)
+// and the published draft list read.
//
// Drafts are never owed work: they feed no reducer, no working status, no
// teardown and no idle sweep. The drain re-reads every gate inside its own
// serialized step, so there is no loop state to disagree with the journal.
-import { randomUUID } from 'node:crypto'
import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
-import {
- QUEUED_MESSAGE_PAUSED_SEND_FAILED,
- type AgentSessionSendResult,
- type AgentSessionWireRefusal
+import type {
+ AgentSessionQueueWait,
+ AgentSessionSendResult,
+ AgentSessionWireRefusal
} from '../../../shared/agent-session-wire'
-import { createStructuredAgentSessionOperationId } from '../../../shared/structured-agent-session-mutation'
import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation'
import { queuedSendAnswer } from './structured-agent-session-queued-send-answer'
import { structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation'
@@ -22,15 +20,16 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import type { AgentSessionTurnContext } from './structured-agent-session-turns'
-import { QueuedMessageNotConsumableError } from '../agent-session-journal/journal-queued-messages'
import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table'
-import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
import {
structuredAgentSessionHostInstance,
structuredQueuePauses
} from './structured-agent-session-queued-pause'
import { nextSendableQueuedCard } from '../agent-session-journal/queued-message-pause'
-import type { StructuredAgentSessionLogger } from './structured-agent-session-logger'
+import { isQueuedClearCard, queuedClearWait } from './structured-conversation-clear'
+import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view'
+import type { AgentSessionBackgroundTaskStops } from '../../../shared/agent-child-work-stop-targets'
+import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
import { agentSessionAttachmentExpiredRefusal } from './structured-agent-session-turns'
import { isAgentSessionAttachmentExpiredError } from '../agent-session-attachments/agent-session-attachment-claims'
@@ -117,14 +116,24 @@ export function structuredQueueHold(input: {
return null
}
-/** The card the drain sends next, or null while anything holds the queue: the drain's own pick
- * through the one gate, so a client told this reads what the drain acts on. Live facts only; the
- * backlog is never a gate, so a lone draft drains. */
-export function nextStructuredQueuedMessage(input: {
+/** What the gate reads beyond the journal. `childWork`, `backgroundTaskStops` and
+ * `stoppedTaskEndingOwed` are read only when a /clear card is next. */
+export type StructuredQueueGateInput = {
journal: AgentSessionJournal
record: AgentSessionRecord | null
fence: number
-}): QueuedMessageRow | null {
+ childWork: () => readonly AgentChildWorkView[] | undefined
+ backgroundTaskStops: () => AgentSessionBackgroundTaskStops | undefined
+ stoppedTaskEndingOwed: () => boolean
+}
+
+/** The card the queue is on once no hold stops it, and what that card still waits for: a /clear
+ * card runs only once background tasks and a handoff end (`queuedClearWait`). The drain, its
+ * schedule, publication and admission all read this one rule. Live facts only; the backlog is
+ * never a gate, so a lone draft drains. */
+export function structuredQueueHead(
+ input: StructuredQueueGateInput
+): { card: QueuedMessageRow; waitsFor: AgentSessionQueueWait['reason'] | null } | null {
const next = oldestActionableQueuedMessage(input.journal)
const { journal, fence } = input
// The gate's cheap `working` first: publication asks on every streamed frame, and the gate's
@@ -135,7 +144,51 @@ export function nextStructuredQueuedMessage(input: {
) {
return null
}
- return structuredQueueHold(input) === null ? next : null
+ if (structuredQueueHold(input) !== null) {
+ return null
+ }
+ return { card: next, waitsFor: queuedCardWait(next, input) }
+}
+
+function queuedCardWait(
+ card: Pick,
+ input: Pick<
+ StructuredQueueGateInput,
+ 'record' | 'childWork' | 'backgroundTaskStops' | 'stoppedTaskEndingOwed'
+ >
+): AgentSessionQueueWait['reason'] | null {
+ return isQueuedClearCard(card)
+ ? queuedClearWait(
+ input.record,
+ input.childWork(),
+ input.backgroundTaskStops(),
+ input.stoppedTaskEndingOwed()
+ )
+ : null
+}
+
+/** The gate's cheap part, for the drain's schedule: an actionable card, the agent idle, and no wait
+ * on the card. The prompt walk is left to the step, which reads the whole gate. */
+export function structuredQueueHeadMayRun(input: StructuredQueueGateInput): boolean {
+ const next = oldestActionableQueuedMessage(input.journal)
+ return (
+ next !== null &&
+ !isStructuredAgentSessionMainAgentWorking(
+ input.journal.activeTurnId(),
+ input.journal.submissions(),
+ input.fence
+ ) &&
+ queuedCardWait(next, input) === null
+ )
+}
+
+/** The card the drain sends next, or null while anything holds the queue or the card waits: the
+ * drain's own pick, so a client told this reads what the drain acts on. */
+export function nextStructuredQueuedMessage(
+ input: StructuredQueueGateInput
+): QueuedMessageRow | null {
+ const head = structuredQueueHead(input)
+ return head && head.waitsFor === null ? head.card : null
}
/**
@@ -144,23 +197,23 @@ export function nextStructuredQueuedMessage(input: {
* ADMISSION rule only, never a drain gate). A lone returned card, or a paused
* queue, does not trap a new send: the user acting now wins, and that send's
* turn starting is what lifts the pause — Orca's own queue policy, a stated
- * deviation from held-head backlog counting.
+ * deviation from held-head backlog counting. A /clear that would only wait for
+ * a handoff or for background tasks the strip can stop waits as a card too, as one sent mid-turn
+ * does.
*/
-export function shouldQueueStructuredAgentSessionSend(input: {
- journal: AgentSessionJournal
- record: AgentSessionRecord | null
- fence: number
-}): boolean {
+export function shouldQueueStructuredAgentSessionSend(
+ input: StructuredQueueGateInput & { body: AgentJournalMessageItem }
+): boolean {
const hold = structuredQueueHold(input)
if (hold === 'blocked') {
// The immediate path's own refusal (`structuredAgentSessionSendBlock`)
// answers; queueing behind a fence would strand the draft.
return false
}
- if (hold !== null) {
+ if (hold !== null || oldestActionableQueuedMessage(input.journal) !== null) {
return true
}
- return oldestActionableQueuedMessage(input.journal) !== null
+ return queuedCardWait(input, input) !== null
}
/**
@@ -171,7 +224,11 @@ export function shouldQueueStructuredAgentSessionSend(input: {
*/
export async function maybeQueueStructuredAgentSessionSend(
context: {
- deps: { store: { getRecord: (sessionId: string) => AgentSessionRecord | null } }
+ deps: {
+ store: { getRecord: (sessionId: string) => AgentSessionRecord | null }
+ adapter: Pick
+ }
+ readChildWork: (sessionId: string) => readonly AgentChildWorkView[] | undefined
},
ctx: Pick,
params: {
@@ -206,7 +263,12 @@ export async function maybeQueueStructuredAgentSessionSend(
!shouldQueueStructuredAgentSessionSend({
journal: ctx.journal,
record: context.deps.store.getRecord(ctx.sessionId),
- fence: ctx.fence
+ fence: ctx.fence,
+ childWork: () => context.readChildWork(ctx.sessionId),
+ backgroundTaskStops: () => context.deps.adapter.backgroundTaskStops?.(ctx.sessionId),
+ stoppedTaskEndingOwed: () =>
+ context.deps.adapter.stoppedTaskEndingOwed?.(ctx.sessionId) === true,
+ body: params.body
})
) {
return null
@@ -248,137 +310,3 @@ export async function maybeQueueStructuredAgentSessionSend(
}
}
}
-
-export type QueuedMessageDrainDeps = {
- sessions: ReadonlyMap
- getRecord: (sessionId: string) => AgentSessionRecord | null
- serialize: (sessionId: string, task: () => Promise) => Promise
- conversationFence: (sessionId: string) => number
- /** The consumed submission is ordinary #22821 work from here on. */
- wakeDelivery: (sessionId: string) => void
- logger: StructuredAgentSessionLogger
-}
-
-/**
- * The serialized drain. Woken by every journal commit (turn, submission, prompt,
- * command and Stop settlements are all commits), by draft mutations, and by the
- * conversation opening; each step re-derives everything and consumes at most one
- * draft — the consumed submission then owes work, which gates the next.
- */
-export class StructuredAgentSessionQueuedMessageDrain {
- private readonly scheduled = new Set()
- private disposed = false
-
- constructor(private readonly deps: QueuedMessageDrainDeps) {}
-
- /** Quit, with delivery: a hand-off made now could only be settled by the next process, so a
- * quit leaves the cards exactly as a crash does. Read by the step at its start, and again
- * right before it appends, since quit can land while it awaits. */
- dispose(): void {
- this.disposed = true
- }
-
- schedule(sessionId: string): void {
- const journal = this.disposed ? undefined : this.deps.sessions.get(sessionId)?.journal
- if (!journal) {
- return
- }
- // Cheap pre-check so token streams do not pay a serialized step per delta.
- // Skipping while working is safe: whatever ends the work is itself a commit
- // that schedules again, and the step re-reads every gate from the fold.
- try {
- if (
- !journal.queuedMessages.settlementOwed() &&
- (oldestActionableQueuedMessage(journal) === null ||
- isStructuredAgentSessionMainAgentWorking(
- journal.activeTurnId(),
- journal.submissions(),
- this.deps.conversationFence(sessionId)
- ))
- ) {
- return
- }
- } catch {
- // The handle is opening or closing; the next commit re-schedules.
- return
- }
- if (this.scheduled.has(sessionId)) {
- return
- }
- this.scheduled.add(sessionId)
- void this.deps
- .serialize(sessionId, () => {
- this.scheduled.delete(sessionId)
- return this.step(sessionId)
- })
- .catch((error: unknown) => {
- this.scheduled.delete(sessionId)
- this.deps.logger.warn('draining queued messages failed', {
- scope: 'queued-drain',
- sessionId,
- error
- })
- })
- }
-
- private async step(sessionId: string): Promise {
- const session = this.deps.sessions.get(sessionId)
- if (this.disposed || !session) {
- return
- }
- const journal = session.journal
- if (journal.queuedMessages.settlementOwed() || journal.queuedMessages.deliveredByEchoOwed()) {
- // A live per-row hook was skipped; heal now, before a draft sends, rather than at reopen.
- await journal.queuedMessages.settleOwed().catch((error: unknown) => {
- this.deps.logger.warn('settling owed queued-message bookkeeping failed', {
- scope: 'queued-settle-owed',
- sessionId,
- error
- })
- })
- }
- const fence = this.deps.conversationFence(sessionId)
- // Whatever clears a hold publishes or commits, which re-derives this step.
- const record = this.deps.getRecord(sessionId)
- const next = nextStructuredQueuedMessage({ journal, record, fence })
- if (this.disposed || !next) {
- return
- }
- // Always a fresh id: the submission names its draft by `queuedMessageId`, never by id equality.
- const submissionId = createStructuredAgentSessionOperationId(randomUUID)
- try {
- await journal.appendSubmission(
- {
- clientMessageId: submissionId,
- // The queue's own automatic send, never kept as a card by a restart or a close.
- origin: 'host',
- payloadFingerprint: next.fingerprint,
- body: next.body,
- fence,
- handoverRecorded: true
- },
- {
- messageId: next.messageId,
- expect: 'waiting',
- settledByOp: null,
- hostInstance: structuredAgentSessionHostInstance(),
- yieldsToPause: true
- }
- )
- } catch (error) {
- if (error instanceof QueuedMessageNotConsumableError) {
- // Lost a race with a Send-now, a Delete or a Stop; their transition stands.
- return
- }
- // Pre-consume failure: the draft stays waiting, held with the marker on
- // the card (a stored fact, so it survives eviction and restart). The
- // hold's own commit notification publishes it. An explicit Send retries;
- // no automatic retry loop.
- await journal.queuedMessages
- .hold({ messageIds: [next.messageId], reason: QUEUED_MESSAGE_PAUSED_SEND_FAILED })
- .catch(() => {})
- throw error
- }
- this.deps.wakeDelivery(sessionId)
- }
-}
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts
index 28be32c15759..9eef8e59ebc0 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts
@@ -37,6 +37,7 @@ import {
structuredAgentSessionSendBlock
} from './structured-agent-session-send-preparation'
import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
+import { isQueuedClearCard, runQueuedConversationClear } from './structured-conversation-clear'
function invalid(message: string): {
ok: false
@@ -70,7 +71,6 @@ export async function withdrawQueuedMessagesForOperation(
})
}
-
/** Draft actions run like any mutation: admitted on the session's lane, the
* conversation opened for the write. */
function mutateQueued(
@@ -107,7 +107,8 @@ async function refusingNewerOrcaJournal(
* wait — through the same send block and pending-prompt gates as any send;
* supersession, Stop and prepared commands are never overridden. The card goes
* out under this operation's id, never its own, and the submission names it by
- * `queuedMessageId`; one id still means one delivery.
+ * `queuedMessageId`; one id still means one delivery. A /clear card is run by
+ * the host instead, spent in its clear's own transaction, so it runs once.
*/
export function sendQueuedStructuredAgentMessage(
context: StructuredAgentSessionMutationContext,
@@ -116,6 +117,8 @@ export function sendQueuedStructuredAgentMessage(
): Promise> {
const { messageId } = params
const operationId = params.envelope.clientOperationId
+ const settledByOp = (ctx: AgentSessionTurnContext) =>
+ agentSessionOperationKey(ctx.resolvedBy, operationId)
const plan: MutationPlan = {
method: 'agentSession.queuedMessageSend',
fields: { messageId },
@@ -156,6 +159,15 @@ export function sendQueuedStructuredAgentMessage(
if (hold === 'working' && row.body.command) {
return invalid("A command can't be sent while the agent is working.")
}
+ if (isQueuedClearCard(row)) {
+ // Run by the host, as the drain runs it: never a submission. A failure lands on the card
+ // itself, so this answers with the card rather than saying it twice; a wait (background
+ // tasks, a handoff) changes nothing on the card, so this Send says it.
+ const ran = await runQueuedConversationClear(context, ctx, row, settledByOp(ctx))
+ return ran.kind === 'waiting'
+ ? { ok: false, refusal: ran.refusal }
+ : { ok: true, value: queuedClearSendAnswer(ctx, operationId, row) }
+ }
const submissionId = operationId
try {
await ctx.journal.appendSubmission(
@@ -171,7 +183,7 @@ export function sendQueuedStructuredAgentMessage(
{
messageId,
expect: row.state,
- settledByOp: agentSessionOperationKey(ctx.resolvedBy, operationId),
+ settledByOp: settledByOp(ctx),
hostInstance: structuredAgentSessionHostInstance()
}
)
@@ -189,9 +201,12 @@ export function sendQueuedStructuredAgentMessage(
return { ok: true, value: { clientMessageId: submissionId, submission } }
},
replay: (ctx) => {
- const opKey = agentSessionOperationKey(ctx.resolvedBy, operationId)
+ const card = ctx.journal.queuedMessages.get(messageId)
+ if (card && isQueuedClearCard(card)) {
+ return queuedClearSendAnswer(ctx, operationId, card)
+ }
const row = ctx.journal.queuedMessages
- .receipts(opKey)
+ .receipts(settledByOp(ctx))
.find((receipt) => receipt.messageId === messageId)
if (!row || row.state !== 'dispatched') {
return null
@@ -203,6 +218,20 @@ export function sendQueuedStructuredAgentMessage(
return mutateQueued(context, caller, params.envelope, plan)
}
+/** A /clear card's Send answers with the card as it now stands: withdrawn once its clear ran,
+ * returned with why when it could not. */
+function queuedClearSendAnswer(
+ ctx: AgentSessionTurnContext,
+ operationId: string,
+ card: QueuedMessageRow
+): AgentSessionSendResult {
+ const { state } = ctx.journal.queuedMessages.get(card.messageId) ?? card
+ return {
+ clientMessageId: operationId,
+ queued: { messageId: card.messageId, position: card.position, state }
+ }
+}
+
/** Delete = discard, with no body in the answer: the card leaving the published
* list IS the outcome, so a lost answer needs no re-ask. An Edit is the client
* copying the text it already renders, then this Delete. */
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts
index d48d9bf27c39..03d897f44cd6 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts
@@ -7,7 +7,8 @@ import {
QUEUED_MESSAGE_PAUSED_SEND_FAILED,
type AgentSessionQueuedMessage,
type AgentSessionQueuedMessagePausedReason,
- type AgentSessionQueuePause
+ type AgentSessionQueuePause,
+ type AgentSessionQueueWait
} from '../../../shared/agent-session-wire'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import {
@@ -15,10 +16,14 @@ import {
resumableQueuePause
} from '../agent-session-journal/queued-message-pause'
import { structuredQueuePauses } from './structured-agent-session-queued-pause'
-import { nextStructuredQueuedMessage } from './structured-agent-session-queued-messages'
+import {
+ structuredQueueHead,
+ type StructuredQueueGateInput
+} from './structured-agent-session-queued-messages'
import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child'
-import type { AgentSessionRecord } from '../../../shared/agent-session-record'
+import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
export type QueuePublication = {
queuedMessages: AgentSessionQueuedMessage[]
@@ -26,18 +31,26 @@ export type QueuePublication = {
/** The card the drain sends next as soon as nothing runs (`nextStructuredQueuedMessage`), so a
* client reads the run as going across a turn's end and that send, which commit apart. */
nextQueuedMessageId: string | null
+ /** The gate's next card while it waits instead (`structuredQueueHead`): the chat isn't busy, and
+ * the card runs without a press. */
+ nextQueuedMessageWait: AgentSessionQueueWait | null
}
/** What the drain's gate reads beyond the journal; resolved per read. */
-export type QueueSendGate = () => { record: AgentSessionRecord | null; fence: number }
+export type QueueSendGate = () => Omit
export function structuredQueueSendGate(
store: Pick,
- sessionId: string
+ sessionId: string,
+ readChildWork: (sessionId: string) => readonly AgentChildWorkView[] | undefined,
+ adapter: Pick
): QueueSendGate {
return () => ({
record: store.getRecord(sessionId),
- fence: structuredAgentSessionConversationFence(store, sessionId)
+ fence: structuredAgentSessionConversationFence(store, sessionId),
+ childWork: () => readChildWork(sessionId),
+ backgroundTaskStops: () => adapter.backgroundTaskStops?.(sessionId),
+ stoppedTaskEndingOwed: () => adapter.stoppedTaskEndingOwed?.(sessionId) === true
})
}
@@ -135,17 +148,26 @@ export function readQueuePublication(
resumable && !queuePauseLiftOnItsWay(resumable, journal.submissions()) ? resumable : null
// The test narrows the type.
const queuePause = pause?.reason === 'stopped' ? { reason: pause.reason } : null
- const nextQueuedMessageId = nextStructuredQueuedMessage({ journal, ...gate() })?.messageId ?? null
+ const head = structuredQueueHead({ journal, ...gate() })
+ const nextQueuedMessageId = head && head.waitsFor === null ? head.card.messageId : null
const previous = publications.get(journal)
+ const wait = head?.waitsFor ? { messageId: head.card.messageId, reason: head.waitsFor } : null
+ // Reference-stable while unchanged, as the list is.
+ const nextQueuedMessageWait =
+ previous?.nextQueuedMessageWait?.messageId === wait?.messageId &&
+ previous?.nextQueuedMessageWait?.reason === wait?.reason
+ ? (previous?.nextQueuedMessageWait ?? null)
+ : wait
if (
previous &&
previous.queuedMessages === queuedMessages &&
sameQueuePause(previous.queuePause, queuePause) &&
- previous.nextQueuedMessageId === nextQueuedMessageId
+ previous.nextQueuedMessageId === nextQueuedMessageId &&
+ previous.nextQueuedMessageWait === nextQueuedMessageWait
) {
return previous
}
- const publication = { queuedMessages, queuePause, nextQueuedMessageId }
+ const publication = { queuedMessages, queuePause, nextQueuedMessageId, nextQueuedMessageWait }
publications.set(journal, publication)
return publication
}
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-rig-turn.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-rig-turn.test-fixture.ts
index e5eda972aee3..63be850f36b1 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-rig-turn.test-fixture.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-rig-turn.test-fixture.ts
@@ -12,11 +12,11 @@ import {
} from './structured-agent-session-host-test-data'
import type { QueuedMessageTestRig } from './structured-agent-session-queued-message-rig.test-fixture'
-/** Opens the turn `id` started; called again with `interrupted` to end it. */
+/** Opens the turn `id` started; called again with `interrupted` or `completed` to end it. */
export async function openRigTurnFor(
rig: QueuedMessageTestRig,
id: string,
- state: 'running' | 'interrupted' = 'running'
+ state: 'running' | 'interrupted' | 'completed' = 'running'
): Promise {
const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal
if (!journal) {
diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts
index a5ddfeb64748..10d77a998034 100644
--- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts
+++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts
@@ -1,6 +1,6 @@
// Host wiring for mid-turn queueing: builds the serialized drain from the
-// host's mutation context and exposes the draft actions (Send, Delete, Resume), so the host
-// class stays a description of its surface.
+// host's mutation context, with the /clear runner its /clear cards use, and exposes the draft
+// actions (Send, Delete, Resume), so the host class stays a description of its surface.
import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations'
import type {
@@ -8,13 +8,19 @@ import type {
StructuredAgentSessionHostSession
} from './structured-agent-session-host-types'
import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child'
-import { StructuredAgentSessionQueuedMessageDrain } from './structured-agent-session-queued-messages'
+import { StructuredAgentSessionQueuedMessageDrain } from './structured-agent-session-queued-drain'
import {
deleteQueuedStructuredAgentMessage,
resumeStructuredAgentQueue,
sendQueuedStructuredAgentMessage
} from './structured-agent-session-queued-mutations'
import { deferredStructuredAgentSessionLogger } from './structured-agent-session-logger'
+import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger'
+import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import {
+ QUEUED_CLEAR_CALLER_KEY,
+ runQueuedConversationClear
+} from './structured-conversation-clear'
/** `sessions` are the live conversations (their `touch` is the idle sweep's activity renewal,
* which the drain's schedule rides); everything else comes from the host's mutation context,
@@ -33,10 +39,36 @@ export function wireStructuredAgentSessionQueuedMessages(
structuredAgentSessionConversationFence(context().deps.store, sessionId),
wakeDelivery: (sessionId) => context().wakeDelivery(sessionId),
// Read lazily, like the rest of this wiring: the host's deps are not assigned yet.
- logger: deferredStructuredAgentSessionLogger(() => context().deps.logger)
+ logger: deferredStructuredAgentSessionLogger(() => context().deps.logger),
+ readChildWork: (sessionId) => context().readChildWork(sessionId),
+ backgroundTaskStops: (sessionId) => context().deps.adapter.backgroundTaskStops?.(sessionId),
+ stoppedTaskEndingOwed: (sessionId) =>
+ context().deps.adapter.stoppedTaskEndingOwed?.(sessionId) === true,
+ runClear: (sessionId, card) => {
+ const { store, adapter } = context().deps
+ return runQueuedConversationClear(
+ context(),
+ {
+ sessionId,
+ journal: sessions.get(sessionId)!.journal,
+ fence: structuredAgentSessionConversationFence(store, sessionId),
+ adapter
+ },
+ card,
+ agentSessionOperationKey(QUEUED_CLEAR_CALLER_KEY, card.messageId)
+ )
+ }
})
return {
drain,
+ /** Child work changed: the strip republishes, and a /clear card waiting on it may run. */
+ onChildWorkChanged: (sessionId: string, strip: { publish: (sessionId: string) => void }) => {
+ strip.publish(sessionId)
+ drain.schedule(sessionId)
+ },
+ /** A /clear card waits on a handoff, which ends in the record store, not the journal. */
+ wakeOnHandoffEnded: (store: Pick) =>
+ store.onHandoffEnded((sessionId) => drain.schedule(sessionId)),
/** Every journal publish: turn, submission, prompt, command and Stop
* settlements are all commits, and each re-derives the drain's gates. */
onJournalActivity: (sessionId: string) => {
diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-clear.ts b/src/main/native-chat/agent-session-wire/structured-conversation-clear.ts
new file mode 100644
index 000000000000..106e5384c696
--- /dev/null
+++ b/src/main/native-chat/agent-session-wire/structured-conversation-clear.ts
@@ -0,0 +1,194 @@
+// `/clear` under the session's serialize: the agent stopped, then a fresh provider context committed
+// in the same conversation. The command RPC runs it at once, and the queue runs a /clear card when
+// its turn comes; neither ever hands it to the agent.
+
+import { agentSessionFailureFact } from '../../../shared/agent-session-failure'
+import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words'
+import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
+import type { AgentSessionRecord } from '../../../shared/agent-session-record'
+import type { AgentSessionQueueWait } from '../../../shared/agent-session-wire'
+import {
+ agentChildWorkStripOffersStop,
+ type AgentSessionBackgroundTaskStops
+} from '../../../shared/agent-child-work-stop-targets'
+import { agentChildWorkLiveness } from '../../../shared/agent-status-child-work-liveness'
+import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view'
+import {
+ agentSessionRefusalReference,
+ type AgentSessionWireRefusal
+} from '../../../shared/agent-session-wire-refusals'
+import {
+ providerContextBoundaryForClear,
+ type AgentSessionConversationClear
+} from '../../runtime/agent-session-conversation-command-record'
+import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
+import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table'
+import { returnUnsentQueuedCard } from '../agent-session-journal/queued-message-return'
+import type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context'
+import type { TurnOutcome } from './structured-agent-session-turns'
+import {
+ conversationCommandBlocked,
+ type ConversationCommandAdmissionContext
+} from './structured-conversation-command-admission'
+
+export const STRUCTURED_AGENT_SESSION_CLEAR_COMMAND = 'clear'
+
+/** Who a clear the queue ran is recorded as: no client pressed it then, and which one queued it
+ * is not kept, so no caller's later /clear replays it. */
+export const QUEUED_CLEAR_CALLER_KEY = 'orca:queued-clear'
+
+/** What the user sent for `/clear`, as its card shows it. */
+export function structuredAgentSessionClearBody(): AgentJournalMessageItem {
+ return {
+ kind: 'message',
+ role: 'user',
+ blocks: [{ type: 'text', text: '/clear' }],
+ command: { name: STRUCTURED_AGENT_SESSION_CLEAR_COMMAND }
+ }
+}
+
+export function isQueuedClearCard(card: Pick): boolean {
+ return card.body.command?.name === STRUCTURED_AGENT_SESSION_CLEAR_COMMAND
+}
+
+export type ConversationClearContext = ConversationCommandAdmissionContext & {
+ journal: AgentSessionJournal
+}
+
+/**
+ * Refused while anything it would cut off is in flight; else the agent is stopped, admission is
+ * read again at the stop's fence, and `commit` writes the clear. A refusal or a throw before the
+ * commit changed nothing.
+ */
+export async function clearConversationUnderSerialize(
+ context: StructuredAgentSessionMutationContext,
+ ctx: ConversationClearContext,
+ marker: { operationId: string; callerKey: string },
+ commit: (clear: AgentSessionConversationClear) => Promise
+): Promise> {
+ const { sessionId } = ctx
+ const store = context.deps.store
+ const blocked = conversationCommandBlocked(
+ ctx,
+ store.getRecord(sessionId)!,
+ context.readChildWork(sessionId),
+ context.sessions.get(sessionId)?.child ? undefined : 'at-rest'
+ )
+ if (blocked) {
+ return { ok: false, refusal: blocked }
+ }
+ await context.stopAgent(sessionId, { cause: 'context-clear' })
+ const stopped = store.getRecord(sessionId)!
+ const fence = stopped.lease.runtimeFence
+ const rechecked = conversationCommandBlocked(
+ { ...ctx, fence },
+ stopped,
+ context.readChildWork(sessionId),
+ 'at-rest'
+ )
+ if (rechecked) {
+ return { ok: false, refusal: rechecked }
+ }
+ const command: AgentSessionConversationClear['command'] = {
+ command: 'clear',
+ runtimeFence: fence,
+ ...marker,
+ phase: 'committed',
+ state: 'completed'
+ }
+ await commit({ sessionId, fence, command, now: context.now() })
+ return { ok: true, value: command }
+}
+
+/**
+ * What a /clear card at the head of the queue waits for before it may run: a handoff, which ends on
+ * its own, or background tasks the strip offers a Stop for, so the person can end the wait. A task
+ * the person stopped still counts until its own ending lands: the provider acknowledges the Stop
+ * first, and that ending opens the agent's turn answering it, which the clear then waits behind.
+ * Work nothing here can stop is no wait: the clear is refused, as at rest. The drain runs it once
+ * the wait ends (it wakes on child work and on a handoff ending). The queue's one gate reads it for
+ * the drain, publication and admission.
+ */
+export function queuedClearWait(
+ record: Pick | null,
+ childWork: readonly AgentChildWorkView[] | undefined,
+ stops: AgentSessionBackgroundTaskStops | undefined,
+ stoppedTaskEndingOwed: boolean
+): AgentSessionQueueWait['reason'] | null {
+ if (record?.lease.handoffStage || record?.lease.handoffOperationId) {
+ return 'handoff'
+ }
+ return stoppedTaskEndingOwed ||
+ (agentChildWorkLiveness(childWork) !== null &&
+ agentChildWorkStripOffersStop(childWork ?? [], stops))
+ ? 'background-tasks'
+ : null
+}
+
+export type QueuedClearOutcome =
+ | { kind: 'cleared' }
+ /** Still waiting, on what `queuedClearWait` names; nothing was written. */
+ | { kind: 'waiting'; refusal: AgentSessionWireRefusal }
+ | { kind: 'returned' }
+
+/**
+ * A /clear card's turn, from the drain or the card's own Send, inside the session's serialize. The
+ * card is spent in the clear's own transaction, so it runs once. Refused for a wait it stays
+ * waiting; refused otherwise, or failed before its commit, the card is returned with why (the card
+ * is where that is said, once), and the cards behind it wait: they were written for the cleared chat.
+ */
+export async function runQueuedConversationClear(
+ context: StructuredAgentSessionMutationContext,
+ ctx: ConversationClearContext,
+ card: Pick,
+ /** The card's receipt: the drain's own key, or the Send that ran it. */
+ settledByOp: string
+): Promise {
+ let fact = agentSessionFailureFact('commandRefused')
+ try {
+ const cleared = await clearConversationUnderSerialize(
+ context,
+ ctx,
+ { operationId: card.messageId, callerKey: QUEUED_CLEAR_CALLER_KEY },
+ (clear) =>
+ ctx.journal.context.clear(
+ providerContextBoundaryForClear(clear),
+ context.deps.store.conversationReceipts.queuedClear(() => clear),
+ settledByOp,
+ card.messageId
+ )
+ )
+ if (cleared.ok) {
+ return { kind: 'cleared' }
+ }
+ // Refused while the gate's wait holds (a Send, or work that began after the drain's read): the
+ // card keeps waiting, and a Send hears why.
+ if (
+ queuedClearWait(
+ context.deps.store.getRecord(ctx.sessionId),
+ context.readChildWork(ctx.sessionId),
+ ctx.adapter.backgroundTaskStops?.(ctx.sessionId),
+ ctx.adapter.stoppedTaskEndingOwed?.(ctx.sessionId) === true
+ ) !== null
+ ) {
+ return { kind: 'waiting', refusal: cleared.refusal }
+ }
+ fact = agentSessionFailureFact('commandRefused', {
+ refusal: agentSessionRefusalReference(cleared.refusal)
+ })
+ } catch (error) {
+ context.deps.logger.warn('running a queued /clear failed', {
+ scope: 'queued-clear',
+ sessionId: ctx.sessionId,
+ error
+ })
+ }
+ const words = agentSessionFailureWords(fact, { command: 'clear', surface: 'rejection' })
+ await returnUnsentQueuedCard(ctx.journal.queuedMessages, {
+ messageId: card.messageId,
+ reason: words.reason,
+ rejection: words.rejection,
+ now: context.now()
+ })
+ return { kind: 'returned' }
+}
diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts
index 7d81e4a77180..afde9545cd11 100644
--- a/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts
+++ b/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts
@@ -1,7 +1,4 @@
-import {
- agentChildWorkViewOffersStop,
- type AgentSessionBackgroundTaskStops
-} from '../../../shared/agent-child-work-stop-targets'
+import { agentChildWorkStripOffersStop } from '../../../shared/agent-child-work-stop-targets'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
import { agentChildWorkLiveness } from '../../../shared/agent-status-child-work-liveness'
@@ -25,7 +22,7 @@ export type ConversationCommandAdmissionContext = {
snapshot(): Pick, 'items'>
submissions: AgentSessionJournal['submissions']
}
- adapter: Pick
+ adapter: Pick
}
function blocked(
@@ -83,7 +80,10 @@ export function conversationCommandBlocked(
if (agentChildWorkLiveness(childWork) !== null) {
return blocked(
'backgroundTasksRunning',
- stripOffersStop(childWork ?? [], ctx.adapter.backgroundTaskStops?.(ctx.sessionId))
+ agentChildWorkStripOffersStop(
+ childWork ?? [],
+ ctx.adapter.backgroundTaskStops?.(ctx.sessionId)
+ )
? 'Stop background tasks before using this command.'
: 'Wait for background tasks to finish before using this command.'
)
@@ -104,17 +104,3 @@ export function conversationCommandBlocked(
}
return null
}
-
-/** The strip's own stop controls: a per-row stop where the provider can target one, else its
- * single untargeted stop. Asking for a stop it does not render names a control nobody can use. */
-function stripOffersStop(
- childWork: readonly AgentChildWorkView[],
- stops: AgentSessionBackgroundTaskStops | undefined
-): boolean {
- if (!stops) {
- return false
- }
- return stops.supportsTaskStop
- ? childWork.some(agentChildWorkViewOffersStop)
- : stops.supportsStopAll
-}
diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts
index ab6df634d13f..06dbaff3d86f 100644
--- a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts
+++ b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts
@@ -4,14 +4,20 @@ import type {
} from '../../../shared/agent-session-conversation-command'
import type {
AgentSessionMutationEnvelope,
- AgentSessionMutationResult
+ AgentSessionMutationResult,
+ AgentSessionQueuedSendReceipt
} from '../../../shared/agent-session-wire'
import { admitAndRunAgentSessionMutation } from './structured-agent-session-mutation-admission'
import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger'
import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations'
import { sendPreparation } from './structured-agent-session-send-preparation'
import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types'
-import { conversationCommandBlocked } from './structured-conversation-command-admission'
+import {
+ clearConversationUnderSerialize,
+ structuredAgentSessionClearBody
+} from './structured-conversation-clear'
+import { maybeQueueStructuredAgentSessionSend } from './structured-agent-session-queued-messages'
+import { queuedSendAnswer } from './structured-agent-session-queued-send-answer'
import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure'
import {
agentSessionFailureWords,
@@ -37,20 +43,29 @@ export function conversationCommandFailure(
export type ConversationCommandParams = {
envelope: AgentSessionMutationEnvelope
command: AgentSessionConversationCommand
- /** /compact only: wait as a card while the agent works, as a queued send does. */
+ /** Wait as a card while the agent works, as a queued send does. */
delivery?: 'queue-if-active'
/** Host-local, set by the client-facing command RPC as for an ordinary send. */
userSend?: true
}
-/** Stop the provider and record a fresh-context boundary in the same conversation. */
+/** A /clear waiting as a card answers at once; the card is the one surface from here. */
+function queuedClearAnswer(
+ queued: AgentSessionQueuedSendReceipt
+): AgentSessionConversationCommandResult {
+ return { command: 'clear', state: 'completed', queued }
+}
+
+/** Stop the provider and record a fresh-context boundary in the same conversation; asked with
+ * `delivery` while the agent works, wait as a /clear card the queue runs when its turn comes. */
export function runStructuredConversationCommand(
context: StructuredAgentSessionMutationContext,
caller: StructuredAgentSessionCaller,
params: ConversationCommandParams
): Promise> {
- const { envelope, command } = params
+ const { envelope, command, delivery } = params
const { sessionId, clientOperationId } = envelope
const store = context.deps.store
+ const operation = { callerKey: caller.callerKey, operationId: clientOperationId }
const matching = () => {
const record = store.getRecord(sessionId)?.conversationCommand
return record?.operationId === clientOperationId && record.callerKey === caller.callerKey
@@ -73,71 +88,77 @@ export function runStructuredConversationCommand(
now: context.now,
plan: {
method: 'agentSession.conversationCommand',
- fields: { command },
+ fields: { command, ...(delivery ? { delivery } : {}) },
// Written to the conversation, not the agent, so whoever owns the agent does not matter.
conversationWrite: true,
recoverUnknownFromDurableState: true,
settlesWithWrite: true,
- successReceipt: () =>
- store.conversationReceipts.clear(
- () => {
- if (!clear) {
- throw new Error('agent_session_clear_not_completed')
- }
- return clear
- },
- { callerKey: caller.callerKey, operationId: clientOperationId }
- ),
- replay: (_ctx, outcome) => {
+ // The clear's own record write, or for a card, the ledger's plain acceptance.
+ successReceipt: () => {
+ const cleared = store.conversationReceipts.clear(() => {
+ if (!clear) {
+ throw new Error('agent_session_clear_not_completed')
+ }
+ return clear
+ }, operation)
+ const queued = store.operationOutcomeReceipt({
+ ...operation,
+ outcome: { status: 'succeeded', sessionId }
+ })
+ return {
+ write: (db) => (clear ? cleared : queued).write(db),
+ committed: () => (clear ? cleared : queued).committed()
+ }
+ },
+ replay: (ctx, outcome) => {
if (outcome.status === 'succeeded' && outcome.conversationCommand) {
return outcome.conversationCommand
}
+ // A card answers from itself; one pruned after it ran or was deleted, as withdrawn.
+ if (delivery) {
+ const card = queuedSendAnswer(ctx.journal, clientOperationId)
+ if (card && 'queued' in card) {
+ return queuedClearAnswer(card.queued)
+ }
+ if (outcome.status === 'succeeded') {
+ return queuedClearAnswer({
+ messageId: clientOperationId,
+ position: 0,
+ state: 'withdrawn'
+ })
+ }
+ }
const prior = matching()
return prior?.phase === 'committed' ? prior : null
},
// The commit is the only write, so a clear with no committed answer changed nothing.
rerunWhenReplayMissing: () => true,
run: async (ctx) => {
- const record = store.getRecord(sessionId)!
- const blocked = conversationCommandBlocked(
- ctx,
- record,
- context.readChildWork(sessionId),
- context.sessions.get(sessionId)?.child ? undefined : 'at-rest'
- )
- if (blocked) {
- return { ok: false, refusal: blocked }
- }
- await context.stopAgent(sessionId, { cause: 'context-clear' })
- const stopped = store.getRecord(sessionId)!
- const fence = stopped.lease.runtimeFence
- const rechecked = conversationCommandBlocked(
- { ...ctx, fence },
- stopped,
- context.readChildWork(sessionId),
- 'at-rest'
- )
- if (rechecked) {
- return { ok: false, refusal: rechecked }
- }
- const completed: AgentSessionConversationClear['command'] = {
- command: 'clear',
- runtimeFence: fence,
- operationId: clientOperationId,
- callerKey: caller.callerKey,
- phase: 'committed',
- state: 'completed'
+ // The queue's own accept rule decides first, as for /compact: whatever a queued send
+ // waits behind, the /clear waits behind too, and the queue runs it when its turn comes.
+ const card = await maybeQueueStructuredAgentSessionSend(context, ctx, {
+ envelope,
+ body: structuredAgentSessionClearBody(),
+ ...(params.userSend ? { userSend: params.userSend } : {}),
+ ...(delivery ? { delivery } : {})
+ })
+ if (card && !card.ok) {
+ return card
}
- clear = { sessionId, fence, command: completed, now: context.now() }
- if (!ctx.operationReceipt) {
- throw new Error('agent_session_clear_receipt_missing')
+ if (card && 'queued' in card.value) {
+ return { ok: true, value: queuedClearAnswer(card.value.queued) }
}
- await ctx.journal.context.clear(
- providerContextBoundaryForClear(clear),
- ctx.operationReceipt,
- agentSessionOperationKey(caller.callerKey, clientOperationId)
- )
- return { ok: true, value: completed }
+ return clearConversationUnderSerialize(context, ctx, operation, (completed) => {
+ clear = completed
+ if (!ctx.operationReceipt) {
+ throw new Error('agent_session_clear_receipt_missing')
+ }
+ return ctx.journal.context.clear(
+ providerContextBoundaryForClear(completed),
+ ctx.operationReceipt,
+ agentSessionOperationKey(caller.callerKey, clientOperationId)
+ )
+ })
}
}
})
diff --git a/src/main/runtime/agent-session-conversation-receipts.ts b/src/main/runtime/agent-session-conversation-receipts.ts
index 6935f8bb0836..007b6f47881d 100644
--- a/src/main/runtime/agent-session-conversation-receipts.ts
+++ b/src/main/runtime/agent-session-conversation-receipts.ts
@@ -26,6 +26,10 @@ export function createAgentSessionConversationReceipts(
}
})
}),
+ /** A clear the queue ran from a card: no operation of its own to settle, as the card's own
+ * withdrawal in the same transaction is what answers for it. */
+ queuedClear: (clear: () => AgentSessionConversationClear) =>
+ transactions.receipt((draft) => commitConversationClearRecord(draft, clear())),
rewind: (
...args: Parameters extends [unknown, ...infer A]
? A
diff --git a/src/main/runtime/agent-session-record-lease-events.test.ts b/src/main/runtime/agent-session-record-lease-events.test.ts
new file mode 100644
index 000000000000..a44582015fc6
--- /dev/null
+++ b/src/main/runtime/agent-session-record-lease-events.test.ts
@@ -0,0 +1,40 @@
+import { describe, expect, it } from 'vitest'
+import type { AgentSessionRecord } from '../../shared/agent-session-record'
+import { AgentSessionLeaseEventListeners } from './agent-session-record-lease-events'
+
+type Lease = AgentSessionRecord['lease']
+
+function record(lease: Partial): AgentSessionRecord {
+ // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: collect reads only `lease`; the rest of the record is irrelevant here.
+ return { lease: { handoffStage: null, handoffOperationId: null, ...lease } } as AgentSessionRecord
+}
+
+describe('lease events a transaction raises', () => {
+ it('tells handoff listeners only of a handoff that ended, after the commit', () => {
+ const listeners = new AgentSessionLeaseEventListeners()
+ expect(listeners.empty).toBe(true)
+ const ended: string[] = []
+ const stop = listeners.onHandoffEnded((sessionId) => ended.push(sessionId))
+ expect(listeners.empty).toBe(false)
+ const before = new Map([
+ ['ending', record({ handoffStage: 'recovering' }).lease],
+ ['still', record({ handoffOperationId: 'op-1' }).lease],
+ ['idle', record({}).lease]
+ ])
+ const events = listeners.collect(
+ before,
+ new Map([
+ ['ending', record({})],
+ ['still', record({ handoffOperationId: 'op-1' })],
+ ['idle', record({})],
+ ['starting', record({ handoffStage: 'recovering' })]
+ ])
+ )
+ expect(ended).toEqual([])
+ listeners.notify(events)
+ expect(ended).toEqual(['ending'])
+ stop()
+ listeners.notify(events)
+ expect(ended).toEqual(['ending'])
+ })
+})
diff --git a/src/main/runtime/agent-session-record-lease-events.ts b/src/main/runtime/agent-session-record-lease-events.ts
new file mode 100644
index 000000000000..2c07b9e420ec
--- /dev/null
+++ b/src/main/runtime/agent-session-record-lease-events.ts
@@ -0,0 +1,53 @@
+// What a record-store transaction did to leases that listeners outside the store wait on.
+
+import type { AgentSessionRecord } from '../../shared/agent-session-record'
+
+type Lease = AgentSessionRecord['lease']
+
+/** Listeners keyed by event, each told the session the event happened to. */
+export class AgentSessionLeaseEventListeners {
+ readonly deathEvidence = new Set<(sessionId: string) => void>()
+ private readonly handoffEnded = new Set<(sessionId: string) => void>()
+
+ onHandoffEnded = (listener: (sessionId: string) => void): (() => void) => {
+ this.handoffEnded.add(listener)
+ return () => this.handoffEnded.delete(listener)
+ }
+
+ get empty(): boolean {
+ return this.deathEvidence.size === 0 && this.handoffEnded.size === 0
+ }
+
+ /** Read inside the transaction: each session it wrote a new proof of death for, and each whose
+ * handoff it ended. */
+ collect(before: ReadonlyMap, records: ReadonlyMap) {
+ const events: { deathEvidence: string[]; handoffEnded: string[] } = {
+ deathEvidence: [],
+ handoffEnded: []
+ }
+ for (const [sessionId, { lease }] of records) {
+ const prior = before.get(sessionId)
+ if (lease.deathEvidence && lease.deathEvidence !== prior?.deathEvidence) {
+ events.deathEvidence.push(sessionId)
+ }
+ if (handoffInFlight(prior) && !handoffInFlight(lease)) {
+ events.handoffEnded.push(sessionId)
+ }
+ }
+ return events
+ }
+
+ /** Once the transaction committed, so a listener may read the store. */
+ notify(events: ReturnType): void {
+ for (const sessionId of events.deathEvidence) {
+ this.deathEvidence.forEach((listener) => listener(sessionId))
+ }
+ for (const sessionId of events.handoffEnded) {
+ this.handoffEnded.forEach((listener) => listener(sessionId))
+ }
+ }
+}
+
+function handoffInFlight(lease: Lease | undefined): boolean {
+ return Boolean(lease?.handoffStage || lease?.handoffOperationId)
+}
diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts
index 1b4dd7bd9d60..f2d7a26930ec 100644
--- a/src/main/runtime/agent-session-record-store.ts
+++ b/src/main/runtime/agent-session-record-store.ts
@@ -57,6 +57,7 @@ import {
type AgentSessionRestartProbeArgs
} from './agent-session-restart-reconciliation'
import { replaceAgentSessionRecordOptions } from './agent-session-record-options'
+import { AgentSessionLeaseEventListeners } from './agent-session-record-lease-events'
import {
commitAgentSessionReservation,
type AgentSessionReserveRequest,
@@ -85,7 +86,7 @@ export const AGENT_SESSION_LEASE_TTL_MS = 30_000,
AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000
export class AgentSessionRecordStore {
- private readonly deathEvidenceListeners = new Set<(sessionId: string) => void>()
+ private readonly leaseListeners = new AgentSessionLeaseEventListeners()
private readonly firstRecordListeners = new Set<() => void>()
readonly conversationReceipts: ReturnType
@@ -342,9 +343,11 @@ export class AgentSessionRecordStore {
/** Told, once committed, of each session a transaction wrote a proof of death for — whichever
* transition wrote it, since every one lands here. Must not throw. */
onDeathEvidence(listener: (sessionId: string) => void): () => void {
- this.deathEvidenceListeners.add(listener)
- return () => this.deathEvidenceListeners.delete(listener)
+ this.leaseListeners.deathEvidence.add(listener)
+ return () => this.leaseListeners.deathEvidence.delete(listener)
}
+ /** A session's lease handoff ended: what waited on it may run. */
+ onHandoffEnded = this.leaseListeners.onHandoffEnded
/** Told, once committed, when the store records its first chat. Must not throw. */
onFirstRecord(listener: () => void): () => void {
@@ -358,24 +361,20 @@ export class AgentSessionRecordStore {
apply: (draft: AgentSessionStoreState) => T,
options?: { inMemoryWhenReadOnly?: boolean }
): Promise => {
- let proven: string[] = []
+ let events: ReturnType | null = null
let heldBefore = true
const result = await this.transactions.transact((draft) => {
heldBefore = this.holdsRecords()
- if (this.deathEvidenceListeners.size === 0) {
+ if (this.leaseListeners.empty) {
return apply(draft)
}
- const before = new Map(
- [...draft.records].map(([id, record]) => [id, record.lease.deathEvidence])
- )
+ const before = new Map([...draft.records].map(([id, record]) => [id, record.lease]))
const applied = apply(draft)
- proven = [...draft.records]
- .filter(([id, { lease }]) => lease.deathEvidence && lease.deathEvidence !== before.get(id))
- .map(([id]) => id)
+ events = this.leaseListeners.collect(before, draft.records)
return applied
}, options)
- for (const sessionId of proven) {
- this.deathEvidenceListeners.forEach((listener) => listener(sessionId))
+ if (events) {
+ this.leaseListeners.notify(events)
}
if (!heldBefore && this.holdsRecords()) {
this.firstRecordListeners.forEach((listener) => listener())
diff --git a/src/main/runtime/structured-agent-session-failure-logging.test.ts b/src/main/runtime/structured-agent-session-failure-logging.test.ts
index e343a45ab211..14d9bbe4c3bc 100644
--- a/src/main/runtime/structured-agent-session-failure-logging.test.ts
+++ b/src/main/runtime/structured-agent-session-failure-logging.test.ts
@@ -20,7 +20,7 @@ import {
import {
StructuredAgentSessionQueuedMessageDrain,
type QueuedMessageDrainDeps
-} from '../native-chat/agent-session-wire/structured-agent-session-queued-messages'
+} from '../native-chat/agent-session-wire/structured-agent-session-queued-drain'
import {
createStructuredAgentSessionLogger,
neverThrowingStructuredAgentSessionLogger
diff --git a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx
index 968421b334bd..97ff06d90b05 100644
--- a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx
+++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx
@@ -90,6 +90,16 @@ export function queuedMessageCardCaption(
'components.native-chat.queuedMessages.behindReturnedHold',
'Waiting — a message ahead needs attention'
)
+ case 'background-tasks':
+ return translate(
+ 'components.native-chat.queuedMessages.waitingForBackgroundTasks',
+ 'Waiting for background tasks to finish'
+ )
+ case 'background-tasks-stoppable':
+ return translate(
+ 'components.native-chat.queuedMessages.waitingForBackgroundTasksStop',
+ 'Waiting for background tasks to finish. Stop them to clear now.'
+ )
case 'awaiting-answer':
return translate(
'components.native-chat.queuedMessages.awaitingAnswerHold',
@@ -206,8 +216,9 @@ export function NativeChatQueuedMessageCard({
{/* Nothing acts on a send still on its way: the host holds no card for it yet. */}
{card.hold === 'sending' ? null : (
<>
- {/* A command never steers: its Send shows only while the agent is idle. */}
- {card.waitsForAgent ? null : (
+ {/* A command never steers: its Send shows only while the agent is idle, and never on
+ a /clear the host runs by itself. */}
+ {card.waitsForAgent || card.runsOnItsOwn ? null : (