diff --git a/apps/sim/lib/copilot/request/handlers/text-intent.test.ts b/apps/sim/lib/copilot/request/handlers/text-intent.test.ts new file mode 100644 index 0000000000..d4e4abc430 --- /dev/null +++ b/apps/sim/lib/copilot/request/handlers/text-intent.test.ts @@ -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('Drafting outline\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('aOnebTwoc'), + 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') + }) +}) diff --git a/apps/sim/lib/copilot/request/handlers/text.ts b/apps/sim/lib/copilot/request/handlers/text.ts index 8f110a82b2..9aaaae2094 100644 --- a/apps/sim/lib/copilot/request/handlers/text.ts +++ b/apps/sim/lib/copilot/request/handlers/text.ts @@ -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 = '' +const INTENT_CLOSE = '' +/** 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 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 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, diff --git a/apps/sim/lib/copilot/request/types.ts b/apps/sim/lib/copilot/request/types.ts index eaaf50d95c..3983b6f82c 100644 --- a/apps/sim/lib/copilot/request/types.ts +++ b/apps/sim/lib/copilot/request/types.ts @@ -150,6 +150,8 @@ export interface StreamingContext { * block. Per-lane keying keeps each subagent's reasoning intact. */ subagentThinkingBlocks: Map + /** Per-lane carry for an tag split across streamed chunks. */ + subagentIntentCarry?: Record isInThinkingBlock: boolean subAgentContent: Record subAgentToolCalls: Record