fix(tui,tool-tasks): finish removing dispose-ness from delivery
The TUI's referenced-prompt snapshot now rides the prompt's own admission transaction instead of a pre-admission inject: while idle, a one-shot prepended agent/prompt-submit wrapper appends the snapshot to the allow decision's additionalContexts, so a blocking hook discards the prompt and its attached context together instead of stranding the snapshot in history for the next unrelated prompt. A prompt discarded before admission releases the wrapper; steering keeps the inject path since it bypasses admission and drains at the same boundary. The session-reference snapshot adapter pinned the old context-before-prompt order; the branch-wide order (prompt first, its contexts after) is now asserted and the fixture re-recorded. tool-tasks drops the last consumer of the removed thrown-disposed contract: completion notices now inject unconditionally, which is well-defined during owner teardown — the loop treats disposal like any cancel, so the notice appends as durable idle context (persisted for resume while the session is attached, dropped with the detached log after). README pair and the owner-disposal tests state the new delivery contract.
This commit is contained in:
@@ -64,6 +64,7 @@ import {
|
||||
type SessionEvent,
|
||||
type SessionHeader,
|
||||
type TodoItem,
|
||||
type UserMessageData,
|
||||
} from '@deepseek-ai/dsh-session'
|
||||
import { foldGoal, type GoalPhase } from '@deepseek-ai/dsh-goal'
|
||||
import {
|
||||
@@ -2777,15 +2778,65 @@ export function createTuiChat(
|
||||
).finally(() => { commandControllers.delete(controller) })
|
||||
}
|
||||
|
||||
const dispatchMessage = (content: ContentBlock[]): void => {
|
||||
const dispatchMessage = (content: ContentBlock[], attachedContext?: UserMessageData): void => {
|
||||
if (disposed) {
|
||||
appendNotice(`Agent "${agent.id}" is disposed.`, 'error')
|
||||
} else if (agent.status === 'running') {
|
||||
return
|
||||
}
|
||||
if (agent.status === 'running') {
|
||||
// Steering is never subject to prompt admission; an attached snapshot
|
||||
// drains beside it at the same step boundary through the outbox.
|
||||
if (attachedContext !== undefined) {
|
||||
agent.inject({ content: attachedContext.content, source: attachedContext.source })
|
||||
}
|
||||
pendingSteering.add(agent.steer({ content, source: { kind: 'user' } }))
|
||||
refreshStatus()
|
||||
} else {
|
||||
agent.followup({ content, source: { kind: 'user' } })
|
||||
return
|
||||
}
|
||||
if (attachedContext === undefined) {
|
||||
agent.followup({ content, source: { kind: 'user' } })
|
||||
return
|
||||
}
|
||||
// Idle: the snapshot rides the prompt's own admission transaction
|
||||
// (PromptDecision.additionalContexts), so a blocking hook discards the
|
||||
// prompt and its attached context together instead of stranding the
|
||||
// snapshot in history for the next unrelated prompt.
|
||||
let cleanedUp = false
|
||||
// Assigned after followup(); cleanup() can run earlier from the catch.
|
||||
let detachDiscard: (() => void) | undefined = undefined
|
||||
const cleanup = (): void => {
|
||||
// Both triggers detach themselves, so a second call needs a future
|
||||
// third trigger; kept so adding one cannot double-release.
|
||||
/* v8 ignore next -- unreachable idempotence guard, see above */
|
||||
if (cleanedUp) return
|
||||
cleanedUp = true
|
||||
detachSubmit()
|
||||
detachDiscard?.()
|
||||
}
|
||||
// Prepended so this wrapper is outermost: it observes the admission
|
||||
// whether a downstream hook allows or blocks, and detaches either way.
|
||||
const detachSubmit = ctx.on('agent/prompt-submit', async (subject, submitted, _source, _signal, next) => {
|
||||
if (subject !== agent || submitted !== content) return next()
|
||||
cleanup()
|
||||
const decision = await next()
|
||||
if (decision.kind !== 'allow') return decision
|
||||
return { ...decision, additionalContexts: [...decision.additionalContexts ?? [], attachedContext] }
|
||||
}, { prepend: true })
|
||||
let id: AgentMessageId
|
||||
// followup() accepts any typed input and contains listener failures;
|
||||
// this guards a future synchronous throw so the wrapper cannot leak.
|
||||
/* v8 ignore start -- future-proofing guard, see above */
|
||||
try {
|
||||
id = agent.followup({ content, source: { kind: 'user' } })
|
||||
} catch (error: unknown) {
|
||||
cleanup()
|
||||
throw error
|
||||
}
|
||||
/* v8 ignore stop */
|
||||
// A prompt discarded before admission (broad cancel) releases the wrapper.
|
||||
detachDiscard = ctx.on('agent/inbox/discard', (subject, messages) => {
|
||||
if (subject === agent && messages.some(message => message.id === id)) cleanup()
|
||||
})
|
||||
}
|
||||
|
||||
/** Deliver a user turn to the agent: steer while running, send while idle, or report a disposed agent. */
|
||||
@@ -3071,10 +3122,9 @@ export function createTuiChat(
|
||||
if (disposed) return
|
||||
editor.addToHistory(text)
|
||||
if (editor.getText() === value) editor.setText('')
|
||||
if (prepared.additionalContext !== undefined) {
|
||||
agent.inject({ content: prepared.additionalContext.content, source: prepared.additionalContext.source })
|
||||
}
|
||||
dispatchMessage(prepared.content)
|
||||
// The snapshot travels with the prompt so a blocking admission hook
|
||||
// discards them together — see dispatchMessage's attached-context path.
|
||||
dispatchMessage(prepared.content, prepared.additionalContext)
|
||||
}, (error: unknown) => {
|
||||
if (!disposed && !controller.signal.aborted) {
|
||||
restoreSubmittedInput()
|
||||
|
||||
@@ -24,10 +24,13 @@ class SnapshotAdapter extends LlmAdapter {
|
||||
|
||||
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
this.requests.push(options)
|
||||
const [context, prompt] = options.messages.slice(-2)
|
||||
// The snapshot rides the prompt's admission: the loop appends the
|
||||
// prompt first, then its additional contexts (the branch-wide ordering
|
||||
// for plugin-sourced context).
|
||||
const [prompt, context] = options.messages.slice(-2)
|
||||
if (context?.role !== 'user' || prompt?.role !== 'user'
|
||||
|| prompt.content[0]?.type !== 'text' || prompt.content[0].text !== 'Use @Source session') {
|
||||
throw new Error('session reference context did not precede the direct user message')
|
||||
throw new Error('session reference context did not follow the direct user message')
|
||||
}
|
||||
yield { type: 'block-start', index: 0, blockType: 'text' }
|
||||
yield { type: 'text-delta', index: 0, text: 'Combined reference request accepted.' }
|
||||
|
||||
@@ -11,18 +11,18 @@ buffer
|
||||
2| " mock • target-session"
|
||||
style 1-23 dim
|
||||
3| <blank>
|
||||
4| " Referenced sessions · Source session (source-session) "
|
||||
style 1-53 dim
|
||||
5| <blank>
|
||||
6| "▌ "
|
||||
4| "▌ "
|
||||
style 0-0 fg=bright-blue
|
||||
7| "▌ You "
|
||||
5| "▌ You "
|
||||
style 0-0 fg=bright-blue
|
||||
style 2-4 fg=bright-blue bold
|
||||
8| "▌ Use @Source session "
|
||||
6| "▌ Use @Source session "
|
||||
style 0-0 fg=bright-blue
|
||||
9| "▌ "
|
||||
7| "▌ "
|
||||
style 0-0 fg=bright-blue
|
||||
8| <blank>
|
||||
9| " Referenced sessions · Source session (source-session) "
|
||||
style 1-53 dim
|
||||
10| <blank>
|
||||
11| " Assistant "
|
||||
style 1-9 fg=bright-magenta bold
|
||||
|
||||
@@ -1893,18 +1893,112 @@ describe('pi-tui chat lifecycle and transcript', () => {
|
||||
result.terminal.send('\r')
|
||||
await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(1) })
|
||||
expect(result.agent.sent).toEqual([[{ type: 'text', text: '@Source chat' }]])
|
||||
expect(result.agent.injected).toHaveLength(1)
|
||||
|
||||
const mention = formatSessionReferenceMention({ sessionId: sourceId, label: 'Source chat' })
|
||||
expect(result.agent.injectedOptions[0]?.source)
|
||||
// Idle: the snapshot rides the prompt's admission (additionalContexts on
|
||||
// the allow decision), not a separate pre-admission inject.
|
||||
expect(result.agent.injected).toHaveLength(0)
|
||||
const decision = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', result.agent.sent[0]!, { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(decision.kind).toBe('allow')
|
||||
expect(decision.kind === 'allow' && decision.additionalContexts?.[0]?.source)
|
||||
.toMatchObject({ kind: 'session-reference', references: [{ sessionId: 'source-session' }] })
|
||||
|
||||
// The one-shot wrapper detached itself at admission: replaying the
|
||||
// waterfall attaches nothing a second time.
|
||||
const replay = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', result.agent.sent[0]!, { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(replay.kind === 'allow' && replay.additionalContexts).toBeUndefined()
|
||||
|
||||
const mention = formatSessionReferenceMention({ sessionId: sourceId, label: 'Source chat' })
|
||||
result.agent.status = 'running'
|
||||
result.terminal.send(`steer ${mention}`)
|
||||
result.terminal.send('\r')
|
||||
await vi.waitFor(() => { expect(result.agent.steered).toHaveLength(1) })
|
||||
expect(result.agent.steered).toEqual([[{ type: 'text', text: 'steer @Source chat' }]])
|
||||
expect(result.agent.injected).toHaveLength(2)
|
||||
// Steering bypasses admission, so its snapshot still arrives via inject.
|
||||
expect(result.agent.injected).toHaveLength(1)
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
it('discards the reference snapshot with its blocked or cancelled prompt', async () => {
|
||||
const result = await setup({
|
||||
async configureContext(ctx) {
|
||||
ctx.provide('tools', { get: () => undefined } as never)
|
||||
await ctx.plugin(TestSessionQueryService)
|
||||
await ctx.plugin(SessionReferenceService)
|
||||
const source = ctx.sessions.create(SessionId('blocked-source'), { meta: { cwd: process.cwd(), createdAt: 1 } })
|
||||
appendUser(source, 'source background')
|
||||
},
|
||||
})
|
||||
// A downstream admission hook blocks the prompt: the attached snapshot
|
||||
// must be discarded with it, not stranded for the next prompt.
|
||||
let blockPrompts = true
|
||||
result.ctx.on('agent/prompt-submit', async (_agent, _content, _source, _signal, next) =>
|
||||
blockPrompts ? { kind: 'block' as const, reason: 'policy' } : next())
|
||||
|
||||
result.terminal.send('@blocked-source')
|
||||
await vi.waitFor(() => { expect(result.terminal.output).toContain('Session · blocked-source') })
|
||||
result.terminal.send('\t')
|
||||
await tick()
|
||||
result.terminal.send('\r')
|
||||
await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(1) })
|
||||
|
||||
const blocked = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', result.agent.sent[0]!, { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(blocked.kind).toBe('block')
|
||||
// Nothing entered history and nothing waits for a later prompt: a fresh
|
||||
// unrelated admission sees no leftover contexts.
|
||||
expect(result.agent.injected).toHaveLength(0)
|
||||
blockPrompts = false
|
||||
const unrelated = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', [{ type: 'text', text: 'unrelated' }], { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(unrelated.kind === 'allow' && unrelated.additionalContexts).toBeUndefined()
|
||||
|
||||
// Second referenced prompt, this time dropped by a broad cancel before
|
||||
// any admission runs: the discard listener releases the wrapper.
|
||||
result.terminal.send('@blocked-source')
|
||||
await vi.waitFor(() => { expect(result.terminal.output).toContain('Session · blocked-source') })
|
||||
result.terminal.send('\t')
|
||||
await tick()
|
||||
result.terminal.send('\r')
|
||||
await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(2) })
|
||||
// A different prompt passing the still-armed wrapper delegates untouched.
|
||||
const passthrough = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', [{ type: 'text', text: 'different prompt' }], { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(passthrough.kind === 'allow' && passthrough.additionalContexts).toBeUndefined()
|
||||
// A foreign agent's discard leaves the wrapper armed.
|
||||
const foreign = { ...result.agent, id: SessionId('foreign') } as unknown as Agent
|
||||
result.ctx.emit('agent/inbox/discard', foreign, [{
|
||||
id: AgentMessageId('stub'),
|
||||
content: result.agent.sent.at(-1)!,
|
||||
source: { kind: 'user' },
|
||||
}])
|
||||
result.ctx.emit('agent/inbox/discard', result.agent, [{
|
||||
id: AgentMessageId('stub'),
|
||||
content: result.agent.sent.at(-1)!,
|
||||
source: { kind: 'user' },
|
||||
}])
|
||||
await tick()
|
||||
// Idempotent: a repeat discard after cleanup is a no-op.
|
||||
result.ctx.emit('agent/inbox/discard', result.agent, [{
|
||||
id: AgentMessageId('stub'),
|
||||
content: result.agent.sent.at(-1)!,
|
||||
source: { kind: 'user' },
|
||||
}])
|
||||
const afterDiscard = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', result.agent.sent.at(-1)!, { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(afterDiscard.kind === 'allow' && afterDiscard.additionalContexts).toBeUndefined()
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
@@ -2054,7 +2148,11 @@ describe('pi-tui chat lifecycle and transcript', () => {
|
||||
expect(result.agent.sent).toEqual([[
|
||||
{ type: 'text', text: '@evil\\x1b\\x07\\x9b\\x0as' },
|
||||
]])
|
||||
expect(result.agent.injectedOptions[0]?.source)
|
||||
const decision = await agentEvents(result.ctx, result.agent).waterfall(
|
||||
'agent/prompt-submit', result.agent.sent[0]!, { kind: 'user' },
|
||||
new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }),
|
||||
)
|
||||
expect(decision.kind === 'allow' && decision.additionalContexts?.[0]?.source)
|
||||
.toMatchObject({ references: [{ sessionId: unsafeId }] })
|
||||
await dispose(result)
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user