refactor(events): remove the agent/stream-chunk mirror of assistant/chunk

The loop recorded every model token delta as a durable `assistant/chunk`
session event AND emitted an identical live `agent/stream-chunk` Cordis event
one line later. Same StreamChunk, same turn/step; the emit added only the live
Agent handle, which the sole consumer discarded. This is the boundary-mirror
duplication the event-domain work removed for turn/step boundaries, applied to
the token stream — a follow-up the boundary RFC explicitly deferred.

The premise is settled: chunk persistence is authoritative (the proposal to
stop persisting chunks was rejected — replay/snapshots depend on it), so
`assistant/chunk` on `session/event` is the load-bearing token stream and
`agent/stream-chunk` is pure redundancy.

- Remove the `agent/stream-chunk` declaration + emit; drop the now-unused
  StreamChunk import from dsh-agent's types.
- Migrate `dsh-ui-stdio` (the only live consumer; ACP already reads
  assistant/chunk off session/event) to render assistant/chunk in its existing
  session/event listener. Consolidating to one listener also makes the
  inReasoning dim-SGR flag deterministic across chunk/boundary events (they no
  longer race across two listeners).
- Repoint the agent-loop tests (cancel/loop) and ui-stdio tests to the
  session/event assistant/chunk feed.
- New RFC (implemented/simplification/2026-07-02-remove-stream-chunk-mirror);
  amend the boundary RFC's retained-list entry to cross-link; update
  architecture, cookbook, event-domain-semantics, the ACP proposal, and the
  regenerated cordis catalog.

Snapshot goldens unchanged (ACP never used the mirror), confirming no
editor-facing transcript change.
This commit is contained in:
Tianyi Cui
2026-07-02 23:42:16 +08:00
parent 2a66b7c4e0
commit b84d4828a8
15 changed files with 116 additions and 89 deletions

View File

@@ -176,7 +176,7 @@ describe('Agent.cancel()', () => {
// the step (the turn-scoped marker, not the step AbortController, is what
// catches this) — no model step runs.
let streamed = false
ctx.on('agent/stream-chunk', () => { streamed = true })
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'turn/start') agent.cancel('from turn-start')
})
@@ -205,7 +205,7 @@ describe('Agent.cancel()', () => {
// cancel check (the one that must closeStep() to balance the already-open
// step) — distinct from a turn-start cancel, caught before the step opens.
let streamed = false
ctx.on('agent/stream-chunk', () => { streamed = true })
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'step/start') agent.cancel('from step-start')
})
@@ -245,7 +245,7 @@ describe('Agent.cancel()', () => {
let disposalDone: Promise<void> | undefined
let streamed = false
ctx.on('agent/stream-chunk', () => { streamed = true })
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'step/start') disposalDone = handle.dispose()
})
@@ -308,7 +308,7 @@ describe('Agent.cancel()', () => {
// runTurn. The second check (after the running flip) must drop the turn —
// runTurn would otherwise throw on the now-empty queue.
let streamed = false
ctx.on('agent/stream-chunk', () => { streamed = true })
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('agent/status', (subject, status) => {
if (subject === agent && status === 'running') agent.cancel('from running listener')
})

View File

@@ -137,21 +137,17 @@ describe('agent loop', () => {
expect(request!.tools?.map(t => t.name)).toEqual(['noop'])
})
it('records raw chunks for replay and emits agent/stream-chunk', async () => {
it('records raw chunks for replay as assistant/chunk session events', async () => {
const adapter = new MockAdapter([textResponse('abc')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
const streamed: StreamChunk[] = []
ctx.on('agent/stream-chunk', (_agent, _turn, _step, chunk) => void streamed.push(chunk))
send(agent, 'hi')
await waitForIdle(ctx, agent)
const chunkEvents = agent.session.events.filter(e => e.type === 'assistant/chunk')
// textResponse('abc') = block-start + 3 deltas + block-end + usage + finish = 7
expect(chunkEvents).toHaveLength(7)
expect(streamed).toHaveLength(7)
// replay: chunk events alone re-assemble to the recorded assistant message
const deltaText = chunkEvents
.flatMap(e => e.type === 'assistant/chunk' ? [e.data.chunk] : [])
@@ -685,10 +681,10 @@ describe('agent loop', () => {
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/start') turns.push(event.data.turn) })
// queue two messages while idle — first starts turn 1 immediately;
// queue the second during turn 1 via a stream-chunk hook
// queue the second during turn 1 when the first assistant chunk streams
let queued = false
ctx.on('agent/stream-chunk', () => {
if (!queued) {
ctx.on('session/event', (_s, event) => {
if (event.type === 'assistant/chunk' && !queued) {
queued = true
send(agent, 'second message')
}