Catch subagent <intent> tags in the server relay

The relay's subagent text handler now runs the split-safe intent extraction
as chunks stream: the latest complete tag is stamped onto the lane's persisted
subagent block (subagentIntent) and stripped from the stored prose, so live,
persisted, and replayed views all agree. Per-lane carry handles tags split
across chunks; a never-closing tag flushes back as plain text.
This commit is contained in:
Siddharth Ganesan
2026-08-14 12:45:24 -07:00
parent 9aaa42810b
commit 95153c0c41
3 changed files with 134 additions and 2 deletions
@@ -0,0 +1,57 @@
/**
* @vitest-environment node
*/
import { describe, expect, it } from 'vitest'
import { handleTextEvent } from '@/lib/copilot/request/handlers/text'
import type { StreamingContext } from '@/lib/copilot/request/types'
function laneTextEvent(text: string) {
return {
type: 'text',
payload: { channel: 'assistant', text },
scope: { lane: 'subagent', parentToolCallId: 'tc-1', agentId: 'file', spanId: 'S1' },
} as never
}
function makeContext(): StreamingContext {
return {
contentBlocks: [{ type: 'subagent', content: 'file', parentToolCallId: 'tc-1', timestamp: 1 }],
subAgentContent: {},
subagentThinkingBlocks: new Map(),
isInThinkingBlock: false,
} as unknown as StreamingContext
}
describe('subagent intent extraction (server relay)', () => {
it('strips a split tag and stamps the lane block intent', async () => {
const ctx = makeContext()
const handler = handleTextEvent('subagent')
await handler(laneTextEvent('<int'), ctx, {} as never, {} as never)
await handler(
laneTextEvent('ent>Drafting outline</intent>\nStarting now.'),
ctx,
{} as never,
{} as never
)
const start = ctx.contentBlocks.find((b) => b.type === 'subagent')
expect(start?.subagentIntent).toBe('Drafting outline')
const text = ctx.contentBlocks.find((b) => b.type === 'subagent_text')
expect(text?.content).toBe('Starting now.')
expect(ctx.subAgentContent['tc-1']).toBe('Starting now.')
})
it('takes the latest tag and keeps surrounding prose', async () => {
const ctx = makeContext()
const handler = handleTextEvent('subagent')
await handler(
laneTextEvent('a<intent>One</intent>b<intent>Two</intent>c'),
ctx,
{} as never,
{} as never
)
const start = ctx.contentBlocks.find((b) => b.type === 'subagent')
expect(start?.subagentIntent).toBe('Two')
expect(ctx.subAgentContent['tc-1']).toBe('abc')
})
})
+75 -2
View File
@@ -1,4 +1,5 @@
import { MothershipStreamV1TextChannel } from '@/lib/copilot/generated/mothership-stream-v1'
import type { StreamingContext } from '@/lib/copilot/request/types'
import type { StreamHandler, ToolScope } from './types'
import {
addContentBlock,
@@ -8,6 +9,72 @@ import {
getScopedSpanIdentity,
} from './types'
const INTENT_OPEN = '<intent>'
const INTENT_CLOSE = '</intent>'
/** A tag that never closes within this many chars flushes back as plain text. */
const INTENT_CARRY_MAX = 240
function partialSuffixLen(buf: string, token: string): number {
const max = Math.min(buf.length, token.length - 1)
for (let len = max; len > 0; len--) {
if (token.startsWith(buf.slice(buf.length - len))) return len
}
return 0
}
/**
* Streams one subagent lane's text through the <intent> protocol: complete
* tags are removed from the persisted prose and the latest one is returned;
* a tag split across chunks is carried (per lane) until its close arrives.
*/
function filterLaneIntent(
context: StreamingContext,
laneId: string,
incoming: string
): { text: string; intent?: string } {
const carries = (context.subagentIntentCarry ??= {})
let buf = (carries[laneId] ?? '') + incoming
carries[laneId] = ''
let out = ''
let intent: string | undefined
while (buf) {
const openIdx = buf.indexOf(INTENT_OPEN)
if (openIdx === -1) {
const keep = partialSuffixLen(buf, INTENT_OPEN)
out += keep ? buf.slice(0, buf.length - keep) : buf
if (keep) carries[laneId] = buf.slice(buf.length - keep)
break
}
out += buf.slice(0, openIdx)
const rest = buf.slice(openIdx)
const closeIdx = rest.indexOf(INTENT_CLOSE, INTENT_OPEN.length)
if (closeIdx === -1) {
if (rest.length > INTENT_CARRY_MAX) {
out += rest
} else {
carries[laneId] = rest
}
break
}
const inner = rest.slice(INTENT_OPEN.length, closeIdx).trim()
if (inner) intent = inner
buf = rest.slice(closeIdx + INTENT_CLOSE.length)
if (buf.startsWith('\n')) buf = buf.slice(1)
}
return intent !== undefined ? { text: out, intent } : { text: out }
}
/** Stamps the latest intent onto the lane's open `subagent` start block. */
function stampLaneIntent(context: StreamingContext, laneId: string, intent: string): void {
for (let i = context.contentBlocks.length - 1; i >= 0; i--) {
const b = context.contentBlocks[i]
if (b.type === 'subagent' && b.parentToolCallId === laneId) {
b.subagentIntent = intent
return
}
}
}
export function handleTextEvent(scope: ToolScope): StreamHandler {
return (event, context) => {
if (event.type !== 'text') {
@@ -48,11 +115,17 @@ export function handleTextEvent(scope: ToolScope): StreamHandler {
if (context.isInThinkingBlock) {
flushThinkingBlock(context)
}
// Catch the lane's <intent> protocol server-side: the latest tag becomes
// the persisted subagent block's status and the stored prose is stripped,
// so every surface (live, persisted, replay) agrees on both.
const { text: cleanChunk, intent } = filterLaneIntent(context, parentToolCallId, chunk)
if (intent) stampLaneIntent(context, parentToolCallId, intent)
if (!cleanChunk) return
context.subAgentContent[parentToolCallId] =
(context.subAgentContent[parentToolCallId] || '') + chunk
(context.subAgentContent[parentToolCallId] || '') + cleanChunk
addContentBlock(context, {
type: 'subagent_text',
content: chunk,
content: cleanChunk,
parentToolCallId,
...(event.scope?.agentId ? { subagent: event.scope.agentId } : {}),
...spanIdentity,
+2
View File
@@ -150,6 +150,8 @@ export interface StreamingContext {
* block. Per-lane keying keeps each subagent's reasoning intact.
*/
subagentThinkingBlocks: Map<string, ContentBlock>
/** Per-lane carry for an <intent> tag split across streamed chunks. */
subagentIntentCarry?: Record<string, string>
isInThinkingBlock: boolean
subAgentContent: Record<string, string>
subAgentToolCalls: Record<string, ToolCallState[]>